Make stateful backend forwarding cancellation-safe

Cargo.toml

@@ -113,6 +113,25 @@ tokio-console = ["dep:console-subscriber", "tokio/tracing"]  # Runtime/task inst
 [lints.rust]
 unexpected_cfgs = { level = "warn", check-cfg = ['cfg(tokio_unstable)'] }
 
+[lints.clippy]
+branches_sharing_code = "deny"
+cast_sign_loss = "deny"
+collection_is_never_read = "deny"
+derive_partial_eq_without_eq = "deny"
+format_collect = "deny"
+future_not_send = "deny"
+literal_string_with_formatting_args = "deny"
+map_unwrap_or = "deny"
+match_wildcard_for_single_variants = "deny"
+needless_collect = "deny"
+or_fun_call = "deny"
+redundant_clone = "deny"
+unchecked_time_subtraction = "deny"
+unnecessary_struct_initialization = "deny"
+unnecessary_wraps = "deny"
+useless_let_if_seq = "deny"
+unused_async = "deny"
+
 [[bench]]
 name = "command_parsing"
 harness = false
@@ -157,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

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!("<bench-{sequence}@example.com>")).expect("benchmark message ID")
+}
+
+fn article_response(sequence: u64) -> Vec<u8> {
+    format!(
+        "220 42 <bench-{sequence}@example.com>\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");
+}

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**
 

src/cache/availability_identity.rs

@@ -33,8 +33,7 @@ impl AvailabilityIdentity {
             namespace: server
                 .availability_namespace
                 .as_ref()
-                .map(ToString::to_string)
-                .unwrap_or_else(|| server.host.to_string()),
+                .map_or_else(|| server.host.to_string(), ToString::to_string),
             account: server
                 .username
                 .clone()
@@ -61,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.

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::<String>()
+        .saturating_add(key.capacity())
+        .saturating_add(size_of::<DiskCachedArticle>())
+        .saturating_add(value.encoded_len())
+}
+
 /// Check available disk space at the given path using df command
 fn check_available_space(_path: &Path) -> Option<u64> {
     // 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<dyn foyer::IoEngineConfig>);
 
@@ -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 <weight@example.com>\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::<String>()
+                + short_key.capacity()
+                + size_of::<DiskCachedArticle>()
+                + 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");