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/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/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/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");