Pipeline compatible article requests upstream

benches/cache_miss_roundtrip_callgrind.rs

@@ -7,16 +7,11 @@
 //!
 //! Run with: `cargo bench --bench cache_miss_roundtrip_callgrind`
 
-macro_rules! supported {
-    ($($item:item)*) => {
-        $(
-            #[cfg(all(target_os = "linux", any(target_arch = "x86_64", target_arch = "aarch64")))]
-            $item
-        )*
-    };
-}
-
-supported! {
+#[cfg(all(
+    target_os = "linux",
+    any(target_arch = "x86_64", target_arch = "aarch64")
+))]
+mod benchmarks {
     use gungraun::{
         Callgrind, LibraryBenchmarkConfig, library_benchmark, library_benchmark_group, main,
     };
@@ -49,10 +44,10 @@ supported! {
         Config {
             servers: vec![
                 Server::builder("127.0.0.1", Port::try_new(backend_port).unwrap())
-                .name("bench-backend")
-                .max_connections(MaxConnections::try_new(4).unwrap())
-                .build()
-                .unwrap(),
+                    .name("bench-backend")
+                    .max_connections(MaxConnections::try_new(4).unwrap())
+                    .build()
+                    .unwrap(),
             ],
             cache,
             ..Default::default()
@@ -191,8 +186,36 @@ supported! {
         }
 
         async fn article_roundtrip(&mut self) -> usize {
-            self.stream.write_all(b"ARTICLE <bench@example.com>\r\n").await.unwrap();
-            read_exact_response_into(&mut self.stream, &mut self.response_buffer, self.response_len).await
+            self.stream
+                .write_all(b"ARTICLE <bench@example.com>\r\n")
+                .await
+                .unwrap();
+            read_exact_response_into(
+                &mut self.stream,
+                &mut self.response_buffer,
+                self.response_len,
+            )
+            .await
+        }
+
+        async fn article_pair_roundtrip(&mut self) -> usize {
+            self.stream
+                .write_all(b"ARTICLE <bench@example.com>\r\nARTICLE <bench@example.com>\r\n")
+                .await
+                .unwrap();
+            let first = read_exact_response_into(
+                &mut self.stream,
+                &mut self.response_buffer,
+                self.response_len,
+            )
+            .await;
+            let second = read_exact_response_into(
+                &mut self.stream,
+                &mut self.response_buffer,
+                self.response_len,
+            )
+            .await;
+            first + second
         }
     }
 
@@ -207,10 +230,14 @@ supported! {
         }
     }
 
-    async fn read_exact_response_into(stream: &mut TcpStream, buffer: &mut [u8], expected: usize) -> usize {
+    async fn read_exact_response_into(
+        stream: &mut TcpStream,
+        buffer: &mut [u8],
+        expected: usize,
+    ) -> usize {
         let mut total = 0usize;
         while total < expected {
-            let n = stream.read(&mut buffer[total..]).await.unwrap();
+            let n = stream.read(&mut buffer[total..expected]).await.unwrap();
             assert_ne!(n, 0, "proxy closed during benchmark response");
             total += n;
         }
@@ -249,6 +276,12 @@ supported! {
         black_box(harness.rt.block_on(harness.client.article_roundtrip()))
     }
 
+    #[library_benchmark]
+    #[bench::article_64k(args = (ARTICLE_64K), setup = setup_no_configured_cache_roundtrip)]
+    fn run_no_configured_cache_pair_roundtrip(mut harness: BenchHarness) -> usize {
+        black_box(harness.rt.block_on(harness.client.article_pair_roundtrip()))
+    }
+
     #[library_benchmark]
     #[bench::article_64k(args = (ARTICLE_64K), setup = setup_metadata_only_cache_roundtrip)]
     #[bench::article_768k(args = (ARTICLE_768K), setup = setup_metadata_only_cache_roundtrip)]
@@ -267,6 +300,7 @@ supported! {
         name = cache_miss_roundtrip;
         benchmarks =
             run_no_configured_cache_roundtrip,
+            run_no_configured_cache_pair_roundtrip,
             run_metadata_only_cache_roundtrip,
             run_direct_backend_roundtrip
     );
@@ -280,6 +314,18 @@ supported! {
         );
         library_benchmark_groups = cache_miss_roundtrip
     );
+
+    pub(super) fn run() {
+        main();
+    }
+}
+
+#[cfg(all(
+    target_os = "linux",
+    any(target_arch = "x86_64", target_arch = "aarch64")
+))]
+fn main() {
+    benchmarks::run();
 }
 
 #[cfg(not(all(

src/proxy/lifecycle.rs

@@ -402,19 +402,23 @@ impl NntpProxy {
     ///
     /// This creates a session with the router, allowing commands from this client
     /// to be routed to different backends based on load balancing.
-    pub async fn handle_client_per_command_routing(
+    pub fn handle_client_per_command_routing(
         &self,
         client_stream: TcpStream,
         client_addr: ClientAddress,
-    ) -> Result<(), SessionError> {
-        // Check for stale pools before handling (lazy recreation after idle)
-        self.check_and_clear_stale_pools();
-        self.increment_active_clients();
-
-        let result = Box::pin(self.handle_per_command_client(client_stream, client_addr)).await;
+    ) -> futures::future::BoxFuture<'_, Result<(), SessionError>> {
+        Box::pin(async move {
+            // Check for stale pools before handling (lazy recreation after idle)
+            self.check_and_clear_stale_pools();
+            self.increment_active_clients();
+
+            let result = self
+                .handle_per_command_client(client_stream, client_addr)
+                .await;
 
-        self.decrement_active_clients();
-        result
+            self.decrement_active_clients();
+            result
+        })
     }
 
     /// Handle a per-command routing session

src/session/backend.rs

@@ -273,6 +273,24 @@ where
     read_until_backend_reply(conn, request, buffer).await
 }
 
+/// Read a response for a request that was already written as part of an
+/// upstream pipeline window.
+pub(crate) async fn read_presend_request_classified<C>(
+    conn: &mut C,
+    request: &RequestContext,
+    buffer: &mut PooledBuffer,
+) -> Result<BackendReadResult>
+where
+    C: AsyncReadExt + Unpin,
+{
+    let n = buffer.read_from(conn).await?;
+    if n == 0 {
+        anyhow::bail!("Backend connection closed unexpectedly");
+    }
+
+    read_until_backend_reply(conn, request, buffer).await
+}
+
 pub(crate) async fn execute_request_classified_timed<C>(
     conn: &mut C,
     request: &RequestContext,

src/session/handlers/command_execution.rs

@@ -143,6 +143,11 @@ enum BackendReadAttemptError {
     Backend(anyhow::Error),
 }
 
+pub(super) enum PresendResponseError {
+    Read(anyhow::Error),
+    Transfer(ResponseTransferError),
+}
+
 impl From<SessionError> for BackendReadAttemptError {
     fn from(err: SessionError) -> Self {
         Self::Backend(anyhow::Error::new(err))
@@ -821,7 +826,7 @@ impl ClientSession {
         ))
     }
 
-    fn handle_response_transfer_error(
+    pub(super) fn handle_response_transfer_error(
         &self,
         conn: crate::pool::ConnectionGuard,
         backend_id: BackendId,
@@ -940,7 +945,7 @@ impl ClientSession {
         }
     }
 
-    async fn checkout_direct_backend_connection(
+    pub(super) async fn checkout_direct_backend_connection(
         &self,
         provider: &crate::pool::DeadpoolConnectionProvider,
         backend_id: crate::types::BackendId,
@@ -1247,6 +1252,92 @@ impl ClientSession {
         Ok((response, buffer, timings))
     }
 
+    pub(super) async fn read_and_write_presend_response<W>(
+        &self,
+        conn: &mut crate::pool::ConnectionGuard,
+        client_write: &mut W,
+        backend: &ArticleBackend,
+        request: &mut RequestContext,
+        availability: &mut crate::cache::ArticleAvailability,
+        backend_to_client_bytes: &mut BackendToClientBytes,
+    ) -> Result<crate::session::backend::BackendResponseComplete, PresendResponseError>
+    where
+        W: AsyncWrite + Unpin,
+    {
+        let backend_id = backend.backend_id();
+        let mut buffer = self.buffer_pool.acquire();
+        let read =
+            backend::read_presend_request_classified(conn.stream_mut(), request, &mut buffer)
+                .await
+                .map_err(PresendResponseError::Read)?;
+        let Some(status_code) = read.status_code() else {
+            read.log_warnings(&buffer, self.client_addr, backend_id);
+            return Err(PresendResponseError::Read(anyhow::anyhow!(
+                "backend returned an invalid response to a presend request"
+            )));
+        };
+
+        if let Ok(missing) = AuthoritativeArticleMissing::from_status_code(*backend, status_code) {
+            self.record_authoritative_article_missing(&missing, availability);
+            if let Some(article_request) =
+                crate::command::CommandHandler::article_lookup_request(request)
+            {
+                self.cache
+                    .record_availability_missing(
+                        article_request.message_id(),
+                        missing.availability_slot(),
+                    )
+                    .await;
+            }
+            let completion = backend::observe_response(
+                request,
+                &mut buffer,
+                conn.stream_mut(),
+                &self.buffer_pool,
+                backend_id,
+            )
+            .await
+            .map_err(PresendResponseError::Transfer)?;
+            self.send_430_to_client(client_write, backend_to_client_bytes)
+                .await
+                .map_err(|error| {
+                    PresendResponseError::Transfer(classify_response_write_err(error))
+                })?;
+            client_write.flush().await.map_err(|error| {
+                PresendResponseError::Transfer(classify_response_write_err(error))
+            })?;
+            return Ok(completion);
+        }
+
+        let params = ResponseWriteParams {
+            request,
+            article_request: crate::command::CommandHandler::article_lookup_request(request),
+            status_code,
+        };
+        let (bytes_written, completion) = self
+            .write_response_to_client(conn.stream_mut(), client_write, backend, buffer, params)
+            .await
+            .map_err(PresendResponseError::Transfer)?;
+        client_write
+            .flush()
+            .await
+            .map_err(|error| PresendResponseError::Transfer(classify_response_write_err(error)))?;
+        self.record_response_metrics(
+            backend_id,
+            request,
+            status_code,
+            request.request_wire_len().as_u64(),
+            bytes_written,
+        );
+        let response = RequestResponseMetadata::new(
+            status_code,
+            usize::try_from(bytes_written).unwrap_or(usize::MAX).into(),
+        );
+        request.record_backend_response(backend_id, response);
+        *backend_to_client_bytes = backend_to_client_bytes.add(response.wire_len().get());
+        Ok(completion)
+    }
+
     #[inline]
     fn should_use_stat_missing_probe(
         provider: &crate::pool::DeadpoolConnectionProvider,
@@ -1493,7 +1584,7 @@ impl ClientSession {
         &self,
         client_write: &mut W,
         backend_to_client_bytes: &mut BackendToClientBytes,
-    ) -> Result<()>
+    ) -> std::io::Result<()>
     where
         W: AsyncWrite + Unpin,
     {