Merge complementary cached article sections

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

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/hybrid.rs

@@ -462,6 +462,13 @@ impl HybridArticleCache {
             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();

src/cache/hybrid_codec.rs

@@ -595,6 +595,53 @@ impl DiskCachedArticle {
         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],
@@ -643,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;
@@ -690,6 +740,13 @@ impl DiskCachedArticle {
     }
 }
 
+fn compatible_article_numbers(
+    left: Option<CachedArticleNumber>,
+    right: Option<CachedArticleNumber>,
+) -> bool {
+    left.is_none() || right.is_none() || left == right
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -1232,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 <id>\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, "<id>")
+                .is_some()
+        );
+    }
+
+    #[test]
+    fn compatible_head_and_body_sections_merge() {
+        let mut head =
+            disk_cached_article_from_ingest_bytes(b"221 0 <id>\r\nH: V\r\n.\r\n").unwrap();
+        let body = disk_cached_article_from_ingest_bytes(b"222 0 <id>\r\nB\r\n.\r\n").unwrap();
+
+        assert!(head.merge_compatible_sections(&body));
+        assert!(
+            head.cached_response_for(RequestKind::Head, "<id>")
+                .is_some()
+        );
+        assert!(
+            head.cached_response_for(RequestKind::Body, "<id>")
+                .is_some()
+        );
+        assert!(
+            head.cached_response_for(RequestKind::Article, "<id>")
+                .is_some()
+        );
+    }
+
+    #[test]
+    fn merging_sections_keeps_the_longer_tier_ttl() {
+        let mut head = DiskCachedArticle::from_contiguous_ingest_with_tier(
+            b"221 0 <id>\r\nH: V\r\n.\r\n",
+            ttl::CacheTier::new(1),
+        )
+        .unwrap();
+        let body = DiskCachedArticle::from_contiguous_ingest_with_tier(
+            b"222 0 <id>\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 <id>\r\nH: V\r\n.\r\n",
+            ttl::CacheTier::new(0),
+        )
+        .unwrap();
+        let body = DiskCachedArticle::from_contiguous_ingest_with_tier(
+            b"222 11 <id>\r\nB\r\n.\r\n",
+            ttl::CacheTier::new(0),
+        )
+        .unwrap();
+
+        assert!(!head.merge_compatible_sections(&body));
+        assert!(
+            head.cached_response_for(RequestKind::Article, "<id>")
+                .is_none()
+        );
+    }
+
     #[test]
     fn test_is_complete_article_false_for_stat() {
         let entry = disk_cached_article_from_ingest_bytes(b"223 0 <t@x>\r\n").unwrap();

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