diff --git a/Cargo.toml b/Cargo.toml index c5c378f8..a499cda6 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -176,17 +176,24 @@ harness = false name = "cache_ingest" harness = false +[[bench]] +name = "cache_metadata_payload" +harness = false + [[bench]] name = "end_to_end_proxy" harness = false [profile.dev] +debug = 0 # Use the profiling/bench profiles when symbols are needed +incremental = false # Avoid retaining a second full set of object files split-debuginfo = "unpacked" # Faster linking โ€” don't bundle debuginfo into binary [profile.dev.package."*"] opt-level = 1 # Compile dependencies with optimizations in dev mode # Huge runtime speedup for rustls/ring/foyer/moka # Minimal compile-time cost (deps cached after first build) +debug = 0 # Keep dependency artifacts compact in the dev profile [profile.release] debug = false # No debug symbols for smaller binaries diff --git a/benches/cache_metadata_payload.rs b/benches/cache_metadata_payload.rs new file mode 100644 index 00000000..22826ac5 --- /dev/null +++ b/benches/cache_metadata_payload.rs @@ -0,0 +1,120 @@ +//! Mixed hybrid-cache workload for metadata and retained-payload updates. +//! +//! Run with: `cargo bench --bench cache_metadata_payload` + +use divan::Bencher; +use nntp_proxy::cache::{HybridCacheConfig, UnifiedCache}; +use nntp_proxy::protocol::StatusCode; +use nntp_proxy::types::{BackendId, MessageId}; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::Duration; +use tempfile::{TempDir, tempdir}; + +const ARTICLE_BODY: &str = "x"; + +fn main() { + divan::main(); +} + +fn benchmark_cache() -> (tokio::runtime::Runtime, TempDir, UnifiedCache) { + let runtime = tokio::runtime::Runtime::new().expect("benchmark runtime"); + let directory = tempdir().expect("benchmark cache directory"); + let config = HybridCacheConfig { + memory_capacity: 4 * 1024 * 1024, + disk_capacity: 64 * 1024 * 1024, + disk_path: directory.path().to_path_buf(), + ttl: Duration::from_secs(300), + compression: nntp_proxy::config::CompressionCodec::None, + shards: 16, + }; + let cache = runtime + .block_on(UnifiedCache::hybrid(config)) + .expect("hybrid cache"); + (runtime, directory, cache) +} + +fn message_id(sequence: u64) -> MessageId<'static> { + MessageId::new(format!("")).expect("benchmark message ID") +} + +fn article_response(sequence: u64) -> Vec { + format!( + "220 42 \r\nSubject: Benchmark\r\n\r\n{ARTICLE_BODY}\r\n.\r\n" + ) + .into_bytes() +} + +#[divan::bench(sample_count = 20, sample_size = 1)] +fn metadata_only_updates(bencher: Bencher) { + let (runtime, _directory, cache) = benchmark_cache(); + let sequence = AtomicU64::new(0); + + bencher + .with_inputs(|| message_id(sequence.fetch_add(1, Ordering::Relaxed))) + .bench_values(|id| { + runtime.block_on(cache.record_backend_has_status( + id, + StatusCode::new(223), + BackendId::from_index(0), + 0.into(), + )); + }); + + runtime + .block_on(cache.close()) + .expect("close benchmark cache"); +} + +#[divan::bench(sample_count = 20, sample_size = 1)] +fn retained_payload_updates(bencher: Bencher) { + let (runtime, _directory, cache) = benchmark_cache(); + let sequence = AtomicU64::new(0); + + bencher + .with_inputs(|| { + let sequence = sequence.fetch_add(1, Ordering::Relaxed); + (message_id(sequence), article_response(sequence)) + }) + .bench_values(|(id, response)| { + runtime.block_on(cache.upsert_ingest(id, response, BackendId::from_index(0), 0.into())); + }); + + runtime + .block_on(cache.close()) + .expect("close benchmark cache"); +} + +#[divan::bench(sample_count = 20, sample_size = 1)] +fn mixed_metadata_and_payload_updates(bencher: Bencher) { + let (runtime, _directory, cache) = benchmark_cache(); + let sequence = AtomicU64::new(0); + + bencher + .with_inputs(|| { + let sequence = sequence.fetch_add(2, Ordering::Relaxed); + ( + message_id(sequence), + message_id(sequence + 1), + article_response(sequence + 1), + ) + }) + .bench_values(|(metadata_id, payload_id, response)| { + runtime.block_on(async { + cache + .record_backend_has_status( + metadata_id, + StatusCode::new(223), + BackendId::from_index(0), + 0.into(), + ) + .await; + cache + .upsert_ingest(payload_id, response, BackendId::from_index(0), 0.into()) + .await; + }); + }); + + runtime + .block_on(cache.close()) + .expect("close benchmark cache"); +} diff --git a/docs/reference/rfc3977-response-codes.md b/docs/reference/rfc3977-response-codes.md index b068c520..46a4dd29 100644 --- a/docs/reference/rfc3977-response-codes.md +++ b/docs/reference/rfc3977-response-codes.md @@ -114,7 +114,8 @@ The proxy treats these request/status combinations as multiline: - `HEAD` with **221** - `BODY` with **222** - `OVER`/`XOVER` with **224** -- `HDR`/`XHDR` with **225** +- `HDR` with **225** +- `XHDR` with **221** or **225** - `NEWNEWS` with **230** - `NEWGROUPS` with **231** diff --git a/src/cache/availability_identity.rs b/src/cache/availability_identity.rs index 0074895f..62020802 100644 --- a/src/cache/availability_identity.rs +++ b/src/cache/availability_identity.rs @@ -60,6 +60,11 @@ impl AvailabilitySlot { pub(crate) const fn bit(self) -> usize { 1usize << self.0 } + + #[must_use] + pub(crate) const fn index(self) -> usize { + self.0 + } } /// Set of configured availability slots used for exhaustion decisions. diff --git a/src/cache/hybrid.rs b/src/cache/hybrid.rs index 48309d7e..e445e392 100644 --- a/src/cache/hybrid.rs +++ b/src/cache/hybrid.rs @@ -52,6 +52,7 @@ use foyer::{ HybridCachePolicy, LruConfig, PsyncIoEngineConfig, RecoverMode, Source, Spawner, }; use std::hash::{Hash, Hasher}; +use std::mem::size_of; use std::path::Path; use std::sync::atomic::{AtomicU64, Ordering}; use std::time::Duration; @@ -64,6 +65,13 @@ use super::{AvailabilityLayout, AvailabilitySlot}; const HYBRID_CACHE_NAME: &str = "nntp-article-cache-v4"; +fn hybrid_entry_weight(key: &String, value: &DiskCachedArticle) -> usize { + size_of::() + .saturating_add(key.capacity()) + .saturating_add(size_of::()) + .saturating_add(value.encoded_len()) +} + /// Check available disk space at the given path using df command fn check_available_space(_path: &Path) -> Option { // Try to use statfs on Linux/Unix @@ -268,7 +276,7 @@ impl HybridArticleCache { .with_eviction_config(LruConfig { high_priority_pool_ratio: 0.1, }) - .with_weighter(|_key: &String, value: &DiskCachedArticle| value.payload_len().get()) + .with_weighter(|key: &String, value| hybrid_entry_weight(key, value)) .storage() .with_io_engine_config(PsyncIoEngineConfig::new()) .with_engine_config( @@ -344,6 +352,7 @@ impl HybridArticleCache { if entry.availability_epoch() != self.availability_epoch { entry.clear_availability(); } + entry.expire_stale_availability(self.ttl_millis); (!entry.is_expired(self.ttl_millis)).then_some(entry) } @@ -367,6 +376,7 @@ impl HybridArticleCache { if cloned.availability_epoch() != self.availability_epoch { cloned.clear_availability(); } + cloned.expire_stale_availability(self.ttl_millis); // Check tier-aware TTL expiration if cloned.is_expired(self.ttl_millis) { @@ -443,13 +453,22 @@ impl HybridArticleCache { let entry_len = entry.payload_len(); let mut existing_availability = None; + let mut existing_negative_timestamps = None; // Check for existing entry - don't overwrite larger semantic payloads with smaller ones. if let Some(existing) = self.get_fresh_entry_for_mutation(&key).await { existing_availability = Some(existing.availability()); + existing_negative_timestamps = Some(existing.negative_timestamps()); if existing.availability().is_missing_slot(slot) { return; } + let mut merged = existing; + if merged.merge_compatible_sections(&entry) { + merged.set_availability_epoch(self.availability_epoch); + self.cache.insert(key, merged); + return; + } + let existing = merged; let existing_len = existing.payload_len(); let existing_complete = existing.is_complete_article(); let new_complete = entry.is_complete_article(); @@ -477,6 +496,9 @@ impl HybridArticleCache { if let Some(availability) = existing_availability { entry.availability = availability; } + if let Some(timestamps) = existing_negative_timestamps { + entry.set_negative_timestamps(timestamps); + } entry.set_availability_epoch(self.availability_epoch); self.cache.insert(key.clone(), entry); debug!(msg_id = %key, stored_bytes = entry_len.get(), tier = tier.get(), "Hybrid cache upsert"); @@ -653,7 +675,7 @@ impl HybridArticleCache { .with_eviction_config(LruConfig { high_priority_pool_ratio: 0.1, }) - .with_weighter(|_key: &String, value: &DiskCachedArticle| value.payload_len().get()) + .with_weighter(|key: &String, value| hybrid_entry_weight(key, value)) .storage() .with_io_engine_config(Box::new(NoopIoEngineConfig) as Box); @@ -689,6 +711,36 @@ mod tests { //! use super::*; + #[test] + fn hybrid_weight_includes_key_metadata_and_payload_framing() { + let missing = DiskCachedArticle::missing(super::ttl::CacheTier::new(0)); + let article = DiskCachedArticle::from_ingest_response_with_tier( + b"220 1 \r\nSubject: Test\r\n\r\nBody\r\n.\r\n" + .as_slice() + .into(), + super::ttl::CacheTier::new(0), + ) + .expect("valid cache entry"); + + let short_key = String::from("a"); + let long_key = String::from("a-long-message-id"); + let missing_weight = hybrid_entry_weight(&short_key, &missing); + + assert_eq!( + missing_weight, + size_of::() + + short_key.capacity() + + size_of::() + + missing.encoded_len() + ); + assert!(hybrid_entry_weight(&long_key, &missing) > missing_weight); + assert!(hybrid_entry_weight(&short_key, &article) > missing_weight); + + let mut overallocated_key = String::from("a"); + overallocated_key.reserve(128); + assert!(hybrid_entry_weight(&overallocated_key, &missing) > missing_weight); + } + #[test] fn hybrid_cache_name_cold_invalidates_old_disk_formats() { assert_eq!(HYBRID_CACHE_NAME, "nntp-article-cache-v4"); diff --git a/src/cache/hybrid_codec.rs b/src/cache/hybrid_codec.rs index c223f753..89ebcb1a 100644 --- a/src/cache/hybrid_codec.rs +++ b/src/cache/hybrid_codec.rs @@ -7,7 +7,10 @@ //! # Wire Format //! //! ```text -//! [magic:u32][status:u16][availability-epoch:u64][missing:u64][timestamp:u64][tier:u8][payload-kind:u8]... +//! V8: [magic:u32][status:u16][availability-epoch:u64][missing:u64] +//! [negative-timestamps:u64 * MAX_BACKENDS][timestamp:u64][tier:u8][payload-kind:u8]... +//! V6/V7 omit the per-backend timestamp array and use the entry timestamp for +//! negative-availability expiration. //! ``` use crate::protocol::StatusCode; @@ -24,6 +27,7 @@ use super::ttl; const DISK_ENTRY_MAGIC_V6: u32 = 0x4e50_4336; // "NPC6" const DISK_ENTRY_MAGIC_V7: u32 = 0x4e50_4337; // "NPC7" +const DISK_ENTRY_MAGIC_V8: u32 = 0x4e50_4338; // "NPC8" const PAYLOAD_MISSING: u8 = 0; const PAYLOAD_AVAILABILITY_ONLY: u8 = 1; const PAYLOAD_ARTICLE: u8 = 2; @@ -31,6 +35,13 @@ const PAYLOAD_HEAD: u8 = 3; const PAYLOAD_BODY: u8 = 4; const PAYLOAD_STAT: u8 = 5; const NO_ARTICLE_NUMBER: u64 = u64::MAX; +const DISK_ENTRY_FIXED_SIZE: usize = size_of::() + + size_of::() + + size_of::() + + size_of::() * super::MAX_BACKENDS + + size_of::() + + size_of::() + + size_of::(); #[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)] struct CachedSectionLen(u32); @@ -124,7 +135,10 @@ impl TryFrom for CacheableStatusCode { /// Implements foyer's `Code` trait manually for efficient serialization: /// - Pre-allocates buffer on decode (no vec resizing) /// - Simple binary format: -/// [magic:u32][status:u16][missing:u64][timestamp:u64][tier:u8][typed-payload] +/// V8 stores one negative-availability timestamp per backend immediately +/// after the missing-bitset, before the entry timestamp and tier. Older V6/V7 +/// entries omit that array and are decoded with their entry timestamp copied +/// into every slot. #[derive(Clone, Debug)] pub struct DiskCachedArticle { /// Validated NNTP status code; only cacheable outcomes are representable. @@ -133,6 +147,9 @@ pub struct DiskCachedArticle { availability_epoch: u64, /// Backend availability tracking (authoritative missing bitset) pub(super) availability: ArticleAvailability, + /// Per-slot timestamps keep one backend's negative fact from being renewed by another. + /// In V8 these are persisted in backend-slot order after `availability`. + negative_timestamps: [ttl::CacheTimestampMillis; super::MAX_BACKENDS], /// Unix timestamp when availability info was last updated (milliseconds since epoch) /// Used to expire stale availability-only entries (missing articles, STAT responses) /// and for tier-aware TTL calculation @@ -146,7 +163,7 @@ pub struct DiskCachedArticle { impl Code for DiskCachedArticle { fn encode(&self, writer: &mut impl Write) -> foyer::Result<()> { writer - .write_all(&DISK_ENTRY_MAGIC_V7.to_le_bytes()) + .write_all(&DISK_ENTRY_MAGIC_V8.to_le_bytes()) .map_err(foyer::Error::io_error)?; writer .write_all(&self.status_code.as_u16().to_le_bytes()) @@ -157,6 +174,11 @@ impl Code for DiskCachedArticle { writer .write_all(&availability_bits_to_wire(self.availability.missing_bits())?.to_le_bytes()) .map_err(foyer::Error::io_error)?; + for timestamp in &self.negative_timestamps { + writer + .write_all(×tamp.get().to_le_bytes()) + .map_err(foyer::Error::io_error)?; + } writer .write_all(&self.timestamp.get().to_le_bytes()) .map_err(foyer::Error::io_error)?; @@ -173,7 +195,10 @@ impl Code for DiskCachedArticle { .read_exact(&mut magic) .map_err(foyer::Error::io_error)?; let magic = u32::from_le_bytes(magic); - if magic != DISK_ENTRY_MAGIC_V6 && magic != DISK_ENTRY_MAGIC_V7 { + if magic != DISK_ENTRY_MAGIC_V6 + && magic != DISK_ENTRY_MAGIC_V7 + && magic != DISK_ENTRY_MAGIC_V8 + { return Err(foyer::Error::io_error(std::io::Error::new( std::io::ErrorKind::InvalidData, "old hybrid cache entry format", @@ -195,7 +220,7 @@ impl Code for DiskCachedArticle { )) })?; - let availability_epoch = if magic == DISK_ENTRY_MAGIC_V7 { + let availability_epoch = if magic == DISK_ENTRY_MAGIC_V7 || magic == DISK_ENTRY_MAGIC_V8 { let mut epoch_bytes = [0u8; 8]; reader .read_exact(&mut epoch_bytes) @@ -212,12 +237,30 @@ impl Code for DiskCachedArticle { .read_exact(&mut missing_bytes) .map_err(foyer::Error::io_error)?; + let missing_bits = availability_bits_from_wire(u64::from_le_bytes(missing_bytes))?; + let mut negative_timestamps = if magic == DISK_ENTRY_MAGIC_V8 { + let mut timestamps = [ttl::CacheTimestampMillis::new(0); super::MAX_BACKENDS]; + for timestamp in &mut timestamps { + let mut bytes = [0u8; 8]; + reader + .read_exact(&mut bytes) + .map_err(foyer::Error::io_error)?; + *timestamp = ttl::CacheTimestampMillis::new(u64::from_le_bytes(bytes)); + } + timestamps + } else { + [ttl::CacheTimestampMillis::new(0); super::MAX_BACKENDS] + }; + // Read timestamp let mut timestamp_bytes = [0u8; 8]; reader .read_exact(&mut timestamp_bytes) .map_err(foyer::Error::io_error)?; let timestamp = ttl::CacheTimestampMillis::new(u64::from_le_bytes(timestamp_bytes)); + if magic != DISK_ENTRY_MAGIC_V8 { + negative_timestamps = [timestamp; super::MAX_BACKENDS]; + } // Read tier let mut tier_byte = [0u8; 1]; @@ -231,9 +274,8 @@ impl Code for DiskCachedArticle { Ok(Self { status_code, availability_epoch, - availability: ArticleAvailability::from_missing_bits(availability_bits_from_wire( - u64::from_le_bytes(missing_bytes), - )?), + availability: ArticleAvailability::from_missing_bits(missing_bits), + negative_timestamps, timestamp, tier, payload, @@ -241,7 +283,7 @@ impl Code for DiskCachedArticle { } fn estimated_size(&self) -> usize { - 4 + 2 + size_of::() + size_of::() + 8 + 1 + encoded_payload_size(&self.payload) + self.encoded_len() } } @@ -435,6 +477,7 @@ impl DiskCachedArticle { status_code, availability_epoch: 0, availability: ArticleAvailability::new(), + negative_timestamps: [ttl::CacheTimestampMillis::new(0); super::MAX_BACKENDS], timestamp: ttl::CacheTimestampMillis::now(), tier, payload, @@ -471,6 +514,7 @@ impl DiskCachedArticle { status_code, availability_epoch: 0, availability: ArticleAvailability::new(), + negative_timestamps: [ttl::CacheTimestampMillis::new(0); super::MAX_BACKENDS], timestamp: ttl::CacheTimestampMillis::now(), tier, payload: CachedPayload::AvailabilityOnly, @@ -483,6 +527,7 @@ impl DiskCachedArticle { status_code: CacheableStatusCode::Missing, availability_epoch: 0, availability: ArticleAvailability::new(), + negative_timestamps: [ttl::CacheTimestampMillis::new(0); super::MAX_BACKENDS], timestamp: ttl::CacheTimestampMillis::now(), tier, payload: CachedPayload::Missing, @@ -504,6 +549,12 @@ impl DiskCachedArticle { self.payload.len() } + /// Size of the serialized value, including metadata and payload framing. + #[must_use] + pub(crate) fn encoded_len(&self) -> usize { + DISK_ENTRY_FIXED_SIZE + encoded_payload_size(&self.payload) + } + #[must_use] pub(crate) fn into_cached_article(self) -> super::article::CachedArticle { super::article::CachedArticle::from_parts( @@ -532,6 +583,7 @@ impl DiskCachedArticle { pub(crate) fn record_availability_missing(&mut self, slot: AvailabilitySlot) { self.availability.record_missing_slot(slot); + self.negative_timestamps[slot.index()] = ttl::CacheTimestampMillis::now(); } pub(crate) fn set_availability_epoch(&mut self, epoch: u64) { @@ -540,15 +592,93 @@ impl DiskCachedArticle { pub(crate) fn clear_availability(&mut self) { self.availability = ArticleAvailability::new(); + self.negative_timestamps = [ttl::CacheTimestampMillis::new(0); super::MAX_BACKENDS]; + } + + pub(crate) fn merge_compatible_sections(&mut self, other: &Self) -> bool { + let merged = match (&self.payload, &other.payload) { + ( + CachedPayload::Head { + article_number: left_number, + headers, + }, + CachedPayload::Body { + article_number: right_number, + body, + }, + ) if compatible_article_numbers(*left_number, *right_number) => { + Some(CachedPayload::Article { + article_number: (*left_number).or(*right_number), + headers: headers.clone(), + body: body.clone(), + }) + } + ( + CachedPayload::Body { + article_number: left_number, + body, + }, + CachedPayload::Head { + article_number: right_number, + headers, + }, + ) if compatible_article_numbers(*left_number, *right_number) => { + Some(CachedPayload::Article { + article_number: (*left_number).or(*right_number), + headers: headers.clone(), + body: body.clone(), + }) + } + _ => None, + }; + + let Some(payload) = merged else { + return false; + }; + self.status_code = CacheableStatusCode::Article; + self.tier = self.tier.max(other.tier); + self.payload = payload; + self.timestamp = ttl::CacheTimestampMillis::now(); + true + } + + pub(crate) fn set_negative_timestamps( + &mut self, + timestamps: [ttl::CacheTimestampMillis; super::MAX_BACKENDS], + ) { + self.negative_timestamps = timestamps; + } + + #[must_use] + pub(crate) const fn negative_timestamps( + &self, + ) -> [ttl::CacheTimestampMillis; super::MAX_BACKENDS] { + self.negative_timestamps } #[cfg(test)] pub(crate) fn record_backend_missing(&mut self, backend_id: BackendId) { - self.availability.record_missing_slot( + self.record_availability_missing( AvailabilitySlot::new(backend_id.as_index()).expect("backend count fits bitmap"), ); } + pub(crate) fn expire_stale_availability(&mut self, base_ttl: ttl::CacheTtlMillis) { + let mut missing_bits = self.availability.missing_bits(); + for index in 0..super::MAX_BACKENDS { + let Some(slot) = AvailabilitySlot::new(index) else { + continue; + }; + if self.availability.is_missing_slot(slot) + && ttl::is_expired(self.negative_timestamps[index], base_ttl, self.tier) + { + missing_bits &= !slot.bit(); + self.negative_timestamps[index] = ttl::CacheTimestampMillis::new(0); + } + } + self.availability = ArticleAvailability::from_missing_bits(missing_bits); + } + #[must_use] pub(crate) const fn availability_epoch(&self) -> u64 { self.availability_epoch @@ -560,7 +690,10 @@ impl DiskCachedArticle { status_code: CacheableStatusCode, tier: ttl::CacheTier, ) { - if !self.is_complete_article() { + if matches!( + self.payload, + CachedPayload::Missing | CachedPayload::AvailabilityOnly + ) { self.status_code = status_code; self.payload = CachedPayload::AvailabilityOnly; self.tier = tier; @@ -607,6 +740,13 @@ impl DiskCachedArticle { } } +fn compatible_article_numbers( + left: Option, + right: Option, +) -> bool { + left.is_none() || right.is_none() || left == right +} + #[cfg(test)] mod tests { use super::*; @@ -617,6 +757,7 @@ mod tests { fn assert_entry_eq(original: &DiskCachedArticle, decoded: &DiskCachedArticle) { assert_eq!(original.status_code, decoded.status_code); assert_eq!(original.availability, decoded.availability); + assert_eq!(original.negative_timestamps, decoded.negative_timestamps); assert_eq!(original.timestamp, decoded.timestamp); assert_eq!(original.tier, decoded.tier); assert_eq!(original.payload, decoded.payload); @@ -1081,10 +1222,42 @@ mod tests { assert!(!decoded.should_try_backend(BackendId::from_index(2))); } + #[test] + fn test_code_encode_decode_preserves_negative_timestamps() { + let mut entry = disk_cached_article_from_ingest_bytes(b"220 ok\r\n").unwrap(); + entry.record_backend_missing(BackendId::from_index(1)); + let mut timestamps = [ttl::CacheTimestampMillis::new(0); crate::cache::MAX_BACKENDS]; + timestamps[1] = ttl::CacheTimestampMillis::new(123_456); + entry.set_negative_timestamps(timestamps); + + let mut encoded = Vec::new(); + entry.encode(&mut encoded).unwrap(); + let decoded = DiskCachedArticle::decode(&mut encoded.as_slice()).unwrap(); + + assert_eq!(decoded.negative_timestamps(), timestamps); + assert!(!decoded.should_try_backend(BackendId::from_index(1))); + } + + #[test] + fn negative_availability_expires_per_backend_after_positive_update() { + let mut entry = disk_cached_article_from_ingest_bytes(b"220 ok\r\n").unwrap(); + entry.record_backend_missing(BackendId::from_index(0)); + entry.record_backend_missing(BackendId::from_index(1)); + entry.negative_timestamps[0] = ttl::CacheTimestampMillis::new(0); + entry.negative_timestamps[1] = + ttl::CacheTimestampMillis::new(ttl::now_millis().saturating_add(60_000)); + + entry.record_backend_has_status(CacheableStatusCode::Stat, ttl::CacheTier::new(0)); + entry.expire_stale_availability(ttl::CacheTtlMillis::new(1)); + + assert!(entry.should_try_backend(BackendId::from_index(0))); + assert!(!entry.should_try_backend(BackendId::from_index(1))); + } + #[test] fn test_code_estimated_size() { let entry = disk_cached_article_from_ingest_bytes(b"220 article\r\n").unwrap(); - let expected = 4 + 2 + size_of::() + size_of::() + 8 + 1 + 1; + let expected = entry.encoded_len(); assert_eq!(entry.estimated_size(), expected); } @@ -1116,6 +1289,78 @@ mod tests { assert!(!entry.is_complete_article()); } + #[test] + fn successful_status_update_preserves_cached_head() { + let mut entry = + disk_cached_article_from_ingest_bytes(b"221 0 \r\nH: V\r\n.\r\n").unwrap(); + + entry.record_backend_has_status(CacheableStatusCode::Stat, ttl::CacheTier::new(0)); + + assert!( + entry + .cached_response_for(RequestKind::Head, "") + .is_some() + ); + } + + #[test] + fn compatible_head_and_body_sections_merge() { + let mut head = + disk_cached_article_from_ingest_bytes(b"221 0 \r\nH: V\r\n.\r\n").unwrap(); + let body = disk_cached_article_from_ingest_bytes(b"222 0 \r\nB\r\n.\r\n").unwrap(); + + assert!(head.merge_compatible_sections(&body)); + assert!( + head.cached_response_for(RequestKind::Head, "") + .is_some() + ); + assert!( + head.cached_response_for(RequestKind::Body, "") + .is_some() + ); + assert!( + head.cached_response_for(RequestKind::Article, "") + .is_some() + ); + } + + #[test] + fn merging_sections_keeps_the_longer_tier_ttl() { + let mut head = DiskCachedArticle::from_contiguous_ingest_with_tier( + b"221 0 \r\nH: V\r\n.\r\n", + ttl::CacheTier::new(1), + ) + .unwrap(); + let body = DiskCachedArticle::from_contiguous_ingest_with_tier( + b"222 0 \r\nB\r\n.\r\n", + ttl::CacheTier::new(3), + ) + .unwrap(); + + assert!(head.merge_compatible_sections(&body)); + assert_eq!(head.tier(), ttl::CacheTier::new(3)); + } + + #[test] + fn incompatible_article_numbers_do_not_merge_sections() { + let mut head = DiskCachedArticle::from_contiguous_ingest_with_tier( + b"221 10 \r\nH: V\r\n.\r\n", + ttl::CacheTier::new(0), + ) + .unwrap(); + let body = DiskCachedArticle::from_contiguous_ingest_with_tier( + b"222 11 \r\nB\r\n.\r\n", + ttl::CacheTier::new(0), + ) + .unwrap(); + + assert!(!head.merge_compatible_sections(&body)); + assert!( + head.cached_response_for(RequestKind::Article, "") + .is_none() + ); + } + #[test] fn test_is_complete_article_false_for_stat() { let entry = disk_cached_article_from_ingest_bytes(b"223 0 \r\n").unwrap(); diff --git a/src/command/handler.rs b/src/command/handler.rs index b83bee6b..50c05bc6 100644 --- a/src/command/handler.rs +++ b/src/command/handler.rs @@ -246,6 +246,10 @@ const STATEFUL_REJECT: RejectResponse = RejectResponse::new( StatusCode::new(codes::FEATURE_NOT_SUPPORTED), "503 Feature not supported in stateless proxy mode\r\n", ); +const TRANSPORT_REJECT: RejectResponse = RejectResponse::new( + StatusCode::new(codes::FEATURE_NOT_SUPPORTED), + "503 Transport-changing command not supported\r\n", +); const fn wire_status(wire: &str) -> u16 { let bytes = wire.as_bytes(); @@ -368,6 +372,7 @@ fn rejection_for(request: &RequestContext) -> RejectResponse { match request.kind() { RequestKind::Post => POST_REJECT, RequestKind::Ihave => TRANSIT_REJECT, + RequestKind::Compress | RequestKind::StartTls => TRANSPORT_REJECT, _ => match request.route_class() { RequestRouteClass::Stateful => STATEFUL_REJECT, _ => TRANSIT_REJECT, @@ -564,6 +569,33 @@ mod tests { ); } + #[test] + fn compress_is_rejected_without_forwarding_in_any_routing_mode() { + let request = RequestContext::parse(b"COMPRESS DEFLATE\r\n").expect("valid command"); + assert_eq!(request.kind(), RequestKind::Compress); + assert_eq!(request.route_class(), RequestRouteClass::Reject); + + for routing_mode in [ + crate::config::RoutingMode::PerCommand, + crate::config::RoutingMode::Hybrid, + crate::config::RoutingMode::Stateful, + ] { + let plan = CommandHandler::classify_request( + &request, + AuthenticationAccess::Authenticated, + routing_mode, + ); + let CommandPlan::Reject(response) = plan else { + panic!("COMPRESS must be rejected in {routing_mode:?}: {plan:?}"); + }; + assert_eq!(response.status().as_u16(), 503); + assert_eq!( + response.to_string(), + "503 Transport-changing command not supported\r\n" + ); + } + } + /// Bug 2 regression test: RFC 4643 ยง2.3.1 โ€” AUTHINFO is case-insensitive. /// /// Before fix: the username/password extractor only stripped exact "AUTHINFO USER" or diff --git a/src/pool/provider.rs b/src/pool/provider.rs index 352325b8..d4f99525 100644 --- a/src/pool/provider.rs +++ b/src/pool/provider.rs @@ -919,7 +919,9 @@ fn resize_then_drop(pool: &Pool, conn: PooledConnection, new_max: usize) { fn shutdown_and_drop(conn: PooledConnection) { let _ = socket2::SockRef::from(conn.underlying_tcp_stream()).shutdown(std::net::Shutdown::Both); - drop(conn); + // `Object::drop` returns the object to deadpool. Take it first so a retired + // socket cannot re-enter the idle pool after its file descriptor is closed. + drop(deadpool::managed::Object::take(conn)); } impl ConnectionProvider for DeadpoolConnectionProvider { @@ -1330,6 +1332,7 @@ mod tests { provider.remove_without_cooldown(conn); assert_eq!(provider.pool.status().max_size, max_size); + assert_eq!(provider.pool.status().size, 0); assert_eq!(provider.active_cooldowns.load(Ordering::Acquire), 0); } diff --git a/src/protocol/request.rs b/src/protocol/request.rs index ae2629f1..e35cb9df 100644 --- a/src/protocol/request.rs +++ b/src/protocol/request.rs @@ -39,6 +39,7 @@ pub enum RequestKind { TakeThis, AuthInfo, StartTls, + Compress, Unknown, } @@ -263,7 +264,7 @@ impl<'a> RequestLine<'a> { #[must_use] pub fn parse(line: &'a [u8]) -> Self { let bytes = trim_line_end(line); - let split = memchr::memchr(b' ', bytes).unwrap_or(bytes.len()); + let split = memchr::memchr2(b' ', b'\t', bytes).unwrap_or(bytes.len()); let verb = &bytes[..split]; let args = if split < bytes.len() { &bytes[split + 1..] @@ -811,7 +812,8 @@ pub(crate) fn request_kind_has_response_body(kind: RequestKind, status: StatusCo | (RequestKind::Capabilities, 101) | (RequestKind::List, 215) | (RequestKind::Over | RequestKind::Xover, 224) - | (RequestKind::Hdr | RequestKind::Xhdr, 225) + | (RequestKind::Hdr, 225) + | (RequestKind::Xhdr, 221 | 225) | (RequestKind::NewNews, 230) | (RequestKind::NewGroups, 231) ) || matches!(kind, RequestKind::Unknown) && status_implies_response_body(code) @@ -833,7 +835,8 @@ const fn route_class(kind: RequestKind, has_message_id: bool) -> RequestRouteCla | RequestKind::Ihave | RequestKind::Check | RequestKind::TakeThis - | RequestKind::StartTls => RequestRouteClass::Reject, + | RequestKind::StartTls + | RequestKind::Compress => RequestRouteClass::Reject, RequestKind::Article | RequestKind::Body | RequestKind::Head | RequestKind::Stat if has_message_id => { @@ -916,6 +919,7 @@ const fn classify_verb(verb: &[u8]) -> RequestKind { }, 8 => { b"AUTHINFO" => RequestKind::AuthInfo, + b"COMPRESS" => RequestKind::Compress, b"STARTTLS" => RequestKind::StartTls, b"TAKETHIS" => RequestKind::TakeThis, }, @@ -1121,6 +1125,16 @@ mod tests { assert_eq!(spaced.route_class(), RequestRouteClass::ArticleByMessageId); } + #[test] + fn borrowed_request_line_accepts_tab_command_separator() { + let parsed = RequestLine::parse(b"ARTICLE\t\r\n"); + + assert_eq!(parsed.kind(), RequestKind::Article); + assert_eq!(parsed.args(), b""); + assert_eq!(parsed.message_id(), Some("")); + assert_eq!(parsed.route_class(), RequestRouteClass::ArticleByMessageId); + } + #[test] fn request_context_owns_borrowed_request_line() { let parsed = RequestLine::parse(b"BODY \r\n"); @@ -1401,6 +1415,7 @@ mod tests { ("TAKETHIS \r\n", RequestKind::TakeThis), ("AUTHINFO USER test\r\n", RequestKind::AuthInfo), ("STARTTLS\r\n", RequestKind::StartTls), + ("COMPRESS DEFLATE\r\n", RequestKind::Compress), ]; for (line, expected) in cases { @@ -1427,6 +1442,7 @@ mod tests { ("CHECK \r\n", RequestRouteClass::Reject), ("TAKETHIS \r\n", RequestRouteClass::Reject), ("STARTTLS\r\n", RequestRouteClass::Reject), + ("COMPRESS DEFLATE\r\n", RequestRouteClass::Reject), ("XFOO arg\r\n", RequestRouteClass::Stateful), ]; @@ -1450,5 +1466,8 @@ mod tests { assert!(unknown.has_response_body(StatusCode::new(282))); assert!(unknown.has_response_body(StatusCode::new(288))); assert!(!unknown.has_response_body(StatusCode::new(281))); + let xhdr = request_context(b"XHDR Subject 1-10\r\n"); + assert!(xhdr.has_response_body(StatusCode::new(221))); + assert!(xhdr.has_response_body(StatusCode::new(225))); } } diff --git a/src/session/backend.rs b/src/session/backend.rs index da0206ce..a0587283 100644 --- a/src/session/backend.rs +++ b/src/session/backend.rs @@ -49,10 +49,6 @@ impl BackendResponseComplete { Self(()) } - pub(crate) const fn stateful_session() -> Self { - Self(()) - } - #[cfg(test)] pub(crate) const fn for_test() -> Self { Self(()) diff --git a/src/session/handlers/hybrid.rs b/src/session/handlers/hybrid.rs index d85f0a7b..c39817b1 100644 --- a/src/session/handlers/hybrid.rs +++ b/src/session/handlers/hybrid.rs @@ -61,8 +61,18 @@ impl StatefulBackendLease { self.backend_id } - fn complete_success(self, completion: crate::session::backend::BackendResponseComplete) { - let _ = self.connection.complete_success(completion); + fn finalize( + self, + disposition: crate::session::handlers::stateful::StatefulConnectionDisposition, + ) { + match disposition { + crate::session::handlers::stateful::StatefulConnectionDisposition::RetireClient => { + self.connection.fail_client(); + } + crate::session::handlers::stateful::StatefulConnectionDisposition::RetireBackend => { + self.connection.fail_backend(); + } + } } } @@ -173,11 +183,17 @@ impl ClientSession { ); // Forward the triggering request (response handled by proxy loop) - initial_request + if let Err(error) = initial_request .request() .write_wire_to(backend.connection_mut().stream_mut()) .await - .context("Failed to send initial request to backend")?; + .context("Failed to send initial request to backend") + { + backend.finalize( + crate::session::handlers::stateful::StatefulConnectionDisposition::RetireBackend, + ); + return Err(crate::session::SessionError::from(error)); + } // Build initial state with carried-over byte counts let initial_bytes = @@ -189,17 +205,20 @@ impl ClientSession { ); state.mark_backend_request_sent(initial_request.request().kind()); - let prepared = PreparedStatefulLoop::new(backend, state); - match self.mode_state.switch_to_stateful() { crate::session::ModeTransition::Switched => {} transition => { + backend.finalize( + crate::session::handlers::stateful::StatefulConnectionDisposition::RetireBackend, + ); return Err(crate::session::SessionError::Backend(anyhow::anyhow!( "stateful handoff entered from invalid mode: {transition:?}" ))); } } + let prepared = PreparedStatefulLoop::new(backend, state); + let (mut backend, state) = prepared.into_parts(); let backend_id = backend.backend_id(); let (backend_read, backend_write) = tokio::io::split(backend.connection_mut().stream_mut()); @@ -218,15 +237,21 @@ impl ClientSession { // pending_guard automatically calls complete_command via Drop - // H1: Only return connection to pool on success - if result.is_ok() { - backend.complete_success( - crate::session::backend::BackendResponseComplete::stateful_session(), - ); - } // else: lease drop removes the connection with replacement cooldown - // Metrics guard automatically ends session via Drop - result.map_err(crate::session::SessionError::from) + match result { + Ok(outcome) => { + let disposition = outcome.disposition(); + let metrics = outcome.into_metrics(); + backend.finalize(disposition); + Ok(metrics) + } + Err(error) => { + let disposition = error.disposition(); + let source = error.into_source(); + backend.finalize(disposition); + Err(crate::session::SessionError::from(source)) + } + } } /// Acquire a dedicated backend connection for stateful mode diff --git a/src/session/handlers/stateful.rs b/src/session/handlers/stateful.rs index a322d916..4dd87e07 100644 --- a/src/session/handlers/stateful.rs +++ b/src/session/handlers/stateful.rs @@ -14,6 +14,106 @@ use tracing::{debug, error, warn}; use crate::constants::buffer::READER_CAPACITY; +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(in crate::session) enum StatefulConnectionDisposition { + RetireClient, + RetireBackend, +} + +#[must_use] +pub(in crate::session) struct StatefulLoopResult { + metrics: TransferMetrics, + disposition: StatefulConnectionDisposition, +} + +#[derive(Debug)] +pub(in crate::session) struct StatefulLoopError { + source: anyhow::Error, + disposition: StatefulConnectionDisposition, +} + +impl StatefulLoopError { + fn client(source: anyhow::Error) -> Self { + Self { + source, + disposition: StatefulConnectionDisposition::RetireClient, + } + } + + fn backend(source: anyhow::Error) -> Self { + Self { + source, + disposition: StatefulConnectionDisposition::RetireBackend, + } + } + + pub(in crate::session) const fn disposition(&self) -> StatefulConnectionDisposition { + self.disposition + } + + pub(in crate::session) fn into_source(self) -> anyhow::Error { + self.source + } +} + +impl std::fmt::Display for StatefulLoopError { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + self.source.fmt(formatter) + } +} + +impl std::error::Error for StatefulLoopError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + self.source.source() + } +} + +impl StatefulLoopResult { + fn new(metrics: TransferMetrics, disposition: StatefulConnectionDisposition) -> Self { + Self { + metrics, + disposition, + } + } + + pub(in crate::session) const fn disposition(&self) -> StatefulConnectionDisposition { + self.disposition + } + + pub(in crate::session) const fn into_metrics(self) -> TransferMetrics { + self.metrics + } +} + +impl std::ops::Deref for StatefulLoopResult { + type Target = TransferMetrics; + + fn deref(&self) -> &Self::Target { + &self.metrics + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum StatefulSessionExit { + ClientDisconnected, + ClientReadError, + BackendDisconnected, + BackendReadError, +} + +impl StatefulSessionExit { + fn disposition(self) -> StatefulConnectionDisposition { + match self { + Self::ClientDisconnected | Self::ClientReadError => { + StatefulConnectionDisposition::RetireClient + } + Self::BackendDisconnected | Self::BackendReadError => { + StatefulConnectionDisposition::RetireBackend + } + } + } +} + enum StatefulClientLine { Eof, Oversized, @@ -146,15 +246,21 @@ impl ClientSession { client_write: &mut W, backend_write: &mut BW, state: &mut crate::session::state::SessionLoopState, - ) -> Result<()> + ) -> std::result::Result<(), StatefulLoopError> where W: tokio::io::AsyncWrite + Unpin, BW: tokio::io::AsyncWrite + Unpin, { match classify_authenticated_stateful_action(request, state.auth_access) { AuthenticatedStatefulAction::Forward => { - request.write_wire_to(backend_write).await?; - backend_write.flush().await?; + request + .write_wire_to(backend_write) + .await + .map_err(|error| StatefulLoopError::backend(error.into()))?; + backend_write + .flush() + .await + .map_err(|error| StatefulLoopError::backend(error.into()))?; state.add_client_to_backend(request.request_wire_len().get()); state.mark_backend_request_sent(request.kind()); } @@ -163,8 +269,14 @@ impl ClientSession { if state.has_pending_backend_replies() { state.push_deferred_reply(AUTH_ALREADY_AUTHENTICATED); } else { - client_write.write_all(AUTH_ALREADY_AUTHENTICATED).await?; - client_write.flush().await?; + client_write + .write_all(AUTH_ALREADY_AUTHENTICATED) + .await + .map_err(|error| StatefulLoopError::client(error.into()))?; + client_write + .flush() + .await + .map_err(|error| StatefulLoopError::client(error.into()))?; state.add_backend_to_client(AUTH_ALREADY_AUTHENTICATED.len() as u64); } } @@ -174,8 +286,14 @@ impl ClientSession { if state.has_pending_backend_replies() { state.push_deferred_reply(capabilities); } else { - client_write.write_all(capabilities).await?; - client_write.flush().await?; + client_write + .write_all(capabilities) + .await + .map_err(|error| StatefulLoopError::client(error.into()))?; + client_write + .flush() + .await + .map_err(|error| StatefulLoopError::client(error.into()))?; state.add_backend_to_client(capabilities.len() as u64); } } @@ -183,8 +301,14 @@ impl ClientSession { if state.has_pending_backend_replies() { state.push_deferred_reply(response.as_bytes()); } else { - client_write.write_all(response.as_bytes()).await?; - client_write.flush().await?; + client_write + .write_all(response.as_bytes()) + .await + .map_err(|error| StatefulLoopError::client(error.into()))?; + client_write + .flush() + .await + .map_err(|error| StatefulLoopError::client(error.into()))?; state.add_backend_to_client(response.len() as u64); } } @@ -214,15 +338,21 @@ impl ClientSession { buffer: &crate::pool::PooledBuffer, len: usize, state: &mut crate::session::state::SessionLoopState, - ) -> Result<()> + ) -> std::result::Result<(), StatefulLoopError> where W: tokio::io::AsyncWrite + Unpin, { for write in state.client_writes_for_backend_read(&buffer[..len]) { - client_write.write_all(write.as_ref()).await?; + client_write + .write_all(write.as_ref()) + .await + .map_err(|error| StatefulLoopError::client(error.into()))?; state.add_backend_to_client(write.len() as u64); } - client_write.flush().await?; + client_write + .flush() + .await + .map_err(|error| StatefulLoopError::client(error.into()))?; Ok(()) } @@ -275,14 +405,26 @@ impl ClientSession { ) .await; - // H2: Only return connection to pool on success - if result.is_ok() { - let _conn = conn_guard.complete_success( - crate::session::backend::BackendResponseComplete::stateful_session(), - ); - } // else: guard drops -> removes connection with replacement cooldown - - result.map_err(crate::session::SessionError::from) + match result { + Ok(outcome) => { + let disposition = outcome.disposition(); + let metrics = outcome.into_metrics(); + match disposition { + StatefulConnectionDisposition::RetireClient => conn_guard.fail_client(), + StatefulConnectionDisposition::RetireBackend => conn_guard.fail_backend(), + } + Ok(metrics) + } + Err(error) => { + let disposition = error.disposition(); + let source = error.into_source(); + match disposition { + StatefulConnectionDisposition::RetireClient => conn_guard.fail_client(), + StatefulConnectionDisposition::RetireBackend => conn_guard.fail_backend(), + } + Err(crate::session::SessionError::from(source)) + } + } } /// Core bidirectional proxy loop @@ -296,7 +438,7 @@ impl ClientSession { mut backend_write: BW, mut state: crate::session::state::SessionLoopState, backend_id: crate::types::BackendId, - ) -> Result + ) -> std::result::Result where R: tokio::io::AsyncRead + Unpin, W: tokio::io::AsyncWrite + Unpin, @@ -305,7 +447,7 @@ impl ClientSession { { let mut command_reader = StatefulCommandReader::new(); - loop { + let exit = loop { // Periodic metrics flush if state.check_and_maybe_flush_metrics() { state.flush_byte_deltas(&self.metrics, backend_id, self.username()); @@ -314,10 +456,16 @@ impl ClientSession { let replies = state.take_ready_deferred_replies(); if !replies.is_empty() { for reply in replies { - client_write.write_all(reply).await?; + client_write + .write_all(reply) + .await + .map_err(|error| StatefulLoopError::client(error.into()))?; state.add_backend_to_client(reply.len() as u64); } - client_write.flush().await?; + client_write + .flush() + .await + .map_err(|error| StatefulLoopError::client(error.into()))?; continue; } @@ -333,10 +481,10 @@ impl ClientSession { .await?; continue; } - Ok(None) => break, + Ok(None) => break StatefulSessionExit::BackendDisconnected, Err(e) => { warn!(client = %self.client_addr, error = %e, "Backend read error"); - break; + break StatefulSessionExit::BackendReadError; } } } @@ -345,19 +493,31 @@ impl ClientSession { // Client โ†’ Backend result = command_reader.read_next(&mut client_reader) => { match result { - Ok(StatefulClientLine::Eof) => break, // Client disconnected + Ok(StatefulClientLine::Eof) => { + break StatefulSessionExit::ClientDisconnected; + } Ok(StatefulClientLine::Oversized) => { use crate::protocol::COMMAND_TOO_LONG; - client_write.write_all(COMMAND_TOO_LONG).await?; - client_write.flush().await?; + client_write + .write_all(COMMAND_TOO_LONG) + .await + .map_err(|error| StatefulLoopError::client(error.into()))?; + client_write + .flush() + .await + .map_err(|error| StatefulLoopError::client(error.into()))?; state.add_backend_to_client(COMMAND_TOO_LONG.len() as u64); continue; } Ok(StatefulClientLine::Invalid) => { client_write .write_all(crate::protocol::COMMAND_SYNTAX_ERROR_RESPONSE) - .await?; - client_write.flush().await?; + .await + .map_err(|error| StatefulLoopError::client(error.into()))?; + client_write + .flush() + .await + .map_err(|error| StatefulLoopError::client(error.into()))?; state.add_backend_to_client( crate::protocol::COMMAND_SYNTAX_ERROR_RESPONSE.len() as u64, ); @@ -392,13 +552,15 @@ impl ClientSession { }, self.client_addr, |username| self.set_username(username), - ).await?; + ) + .await + .map_err(StatefulLoopError::client)?; state.apply_auth_result(&auth_result); } } Err(e) => { warn!(client = %self.client_addr, error = %e, "Client read error"); - break; + break StatefulSessionExit::ClientReadError; } } } @@ -417,20 +579,21 @@ impl ClientSession { ) .await?; } - Ok(None) => break, // Backend disconnected + Ok(None) => break StatefulSessionExit::BackendDisconnected, Err(e) => { warn!(client = %self.client_addr, error = %e, "Backend read error"); - break; + break StatefulSessionExit::BackendReadError; } } } } - } + }; // Final metrics - report any remaining byte deltas state.flush_byte_deltas(&self.metrics, backend_id, self.username()); - Ok(state.into_metrics()) + let disposition = exit.disposition(); + Ok(StatefulLoopResult::new(state.into_metrics(), disposition)) } } @@ -465,6 +628,23 @@ mod tests { .build() } + #[test] + fn stateful_session_exit_never_assumes_reader_state_is_reset() { + assert_eq!( + super::StatefulSessionExit::ClientDisconnected.disposition(), + super::StatefulConnectionDisposition::RetireClient + ); + + assert_eq!( + super::StatefulSessionExit::ClientReadError.disposition(), + super::StatefulConnectionDisposition::RetireClient + ); + assert_eq!( + super::StatefulSessionExit::BackendDisconnected.disposition(), + super::StatefulConnectionDisposition::RetireBackend + ); + } + struct GateWriterState { bytes: Vec, blocked: bool, @@ -622,7 +802,7 @@ mod tests { assert_eq!(writer_control.bytes(), expected); drop(client_end); - tokio::time::timeout(std::time::Duration::from_secs(1), proxy) + let _ = tokio::time::timeout(std::time::Duration::from_secs(1), proxy) .await .expect("stateful proxy did not stop") .expect("stateful proxy task panicked") diff --git a/src/session/multiline_framing.rs b/src/session/multiline_framing.rs index c29a83cd..b2fafbe2 100644 --- a/src/session/multiline_framing.rs +++ b/src/session/multiline_framing.rs @@ -3054,6 +3054,26 @@ mod tests { assert!(!order.has_deferred_replies()); } + #[test] + fn xhdr_221_response_completes_when_fragmented_before_deferred_reply() { + let request = crate::protocol::RequestContext::parse(b"XHDR Subject 1-10\r\n") + .expect("valid request"); + let mut order = BackendResponseOrder::default(); + order.push_request(request.kind()); + order.push_deferred_reply(b"205 Goodbye\r\n"); + + let first = order.client_writes_for_backend_read(b"221 1 Subject\r\nvalue\r\n"); + assert_eq!(first.len(), 1); + assert_eq!(&first[0][..], b"221 1 Subject\r\nvalue\r\n"); + + let second = order.client_writes_for_backend_read(b".\r\n"); + assert_eq!(second.len(), 2); + assert_eq!(&second[0][..], b".\r\n"); + assert_eq!(&second[1][..], b"205 Goodbye\r\n"); + assert!(!order.has_pending_backend_replies()); + assert!(!order.has_deferred_replies()); + } + #[test] fn backend_response_order_keeps_common_writes_inline_and_borrowed() { let request = crate::protocol::RequestContext::parse(b"DATE\r\n").expect("valid request"); diff --git a/src/session/response_transfer.rs b/src/session/response_transfer.rs index c11b7e72..ac68508c 100644 --- a/src/session/response_transfer.rs +++ b/src/session/response_transfer.rs @@ -14,9 +14,8 @@ pub(crate) enum ResponseTransferError { /// Client write failed after backend response ownership was already established. /// - /// This is treated as terminal for backend-connection reuse to prevent - /// protocol desync when a partially-written client response races with - /// subsequent backend reuse. + /// A complete response with no queued backend bytes leaves the backend + /// reusable even when the client disappears during the final write. ClientWrite(std::io::Error), /// Backend closed connection before sending a complete multiline response. @@ -54,9 +53,8 @@ impl ResponseTransferError { ResponseConnectionReuse::QueuedBytes { .. } => BackendConnectionOutcome::BackendDirty, ResponseConnectionReuse::Reusable => match self { Self::ClientDisconnect(_) => BackendConnectionOutcome::BackendHealthy, - Self::ClientWrite(_) | Self::BackendEof { .. } | Self::Io(_) => { - BackendConnectionOutcome::BackendFailed - } + Self::ClientWrite(_) => BackendConnectionOutcome::BackendHealthy, + Self::BackendEof { .. } | Self::Io(_) => BackendConnectionOutcome::BackendFailed, }, } } @@ -155,7 +153,7 @@ mod tests { ); assert_eq!( client_write.pool_fate(ResponseConnectionReuse::Reusable), - BackendConnectionOutcome::BackendFailed + BackendConnectionOutcome::BackendHealthy ); assert_eq!( eof.pool_fate(ResponseConnectionReuse::Reusable),