From af30b7e679b58c1d6aacc380d234636ea8760112 Mon Sep 17 00:00:00 2001
From: asonix <asonix@asonix.dog>
Date: Thu, 7 Aug 2025 17:56:44 -0500
Subject: [PATCH] actix-http: h1: stop pipelining when not reading full
 requests

The existing pipelining behavior of the h1 dispatcher can cause client timeouts
if the entire request body isn't read. It puts the dispatcher into a state where
it refuses to read more (payload dropped) but there are still bytes in the buffer
from the request body.

This solution adds the SHUTDOWN flag in addition to the FINISHED flag
when completing a response when both the following are true:

1. There are no messages in `this.messages`
2. There is still a payload in `this.payload`

This combination implies two things. First, that we have not parsed a
pipelined request after the request we have just responded to. Second,
that the current request payload has not been fed an EOF. Because there
are no pipelined requests, we know that the current request payload
belongs to the request we have just responded to, and because the
request payload has not been fed an EOF, we know we never finished
reading it.

When this occurs, adding the SHUTDOWN flag to the dispatcher triggers a
`flush` and a `poll_shutdown` on the IO resource on the next poll.
---
 actix-http/src/h1/dispatcher.rs | 54 ++++++++++++++++++++++++++++++---
 1 file changed, 49 insertions(+), 5 deletions(-)

diff --git a/actix-http/src/h1/dispatcher.rs b/actix-http/src/h1/dispatcher.rs
index 00b51360e..adcda10c6 100644
--- a/actix-http/src/h1/dispatcher.rs
+++ b/actix-http/src/h1/dispatcher.rs
@@ -386,7 +386,14 @@ where
         let mut this = self.project();
         this.state.set(match size {
             BodySize::None | BodySize::Sized(0) => {
-                this.flags.insert(Flags::FINISHED);
+                let payload_unfinished = this.payload.is_some();
+
+                if payload_unfinished {
+                    this.flags.insert(Flags::SHUTDOWN | Flags::FINISHED);
+                } else {
+                    this.flags.insert(Flags::FINISHED);
+                }
+
                 State::None
             }
             _ => State::SendPayload { body },
@@ -404,7 +411,14 @@ where
         let mut this = self.project();
         this.state.set(match size {
             BodySize::None | BodySize::Sized(0) => {
-                this.flags.insert(Flags::FINISHED);
+                let payload_unfinished = this.payload.is_some();
+
+                if payload_unfinished {
+                    this.flags.insert(Flags::SHUTDOWN | Flags::FINISHED);
+                } else {
+                    this.flags.insert(Flags::FINISHED);
+                }
+
                 State::None
             }
             _ => State::SendErrorPayload { body },
@@ -503,10 +517,25 @@ where
                             Poll::Ready(None) => {
                                 this.codec.encode(Message::Chunk(None), this.write_buf)?;
 
+                                // if we have not yet pipelined to the next request, then
+                                // this.payload was the payload for the request we just finished
+                                // responding to. We can check to see if we finished reading it
+                                // yet, and if not, shutdown the connection.
+                                let payload_unfinished = this.payload.is_some();
+                                let not_pipelined = this.messages.is_empty();
+
+                                println!("not pipelined: {not_pipelined}");
+                                println!("payload unfinished: {payload_unfinished}");
+
                                 // payload stream finished.
                                 // set state to None and handle next message
                                 this.state.set(State::None);
-                                this.flags.insert(Flags::FINISHED);
+
+                                if not_pipelined && payload_unfinished {
+                                    this.flags.insert(Flags::SHUTDOWN | Flags::FINISHED);
+                                } else {
+                                    this.flags.insert(Flags::FINISHED);
+                                }
 
                                 continue 'res;
                             }
@@ -542,10 +571,25 @@ where
                             Poll::Ready(None) => {
                                 this.codec.encode(Message::Chunk(None), this.write_buf)?;
 
-                                // payload stream finished
+                                // if we have not yet pipelined to the next request, then
+                                // this.payload was the payload for the request we just finished
+                                // responding to. We can check to see if we finished reading it
+                                // yet, and if not, shutdown the connection.
+                                let payload_unfinished = this.payload.is_some();
+                                let not_pipelined = this.messages.is_empty();
+
+                                println!("not pipelined: {not_pipelined}");
+                                println!("payload unfinished: {payload_unfinished}");
+
+                                // payload stream finished.
                                 // set state to None and handle next message
                                 this.state.set(State::None);
-                                this.flags.insert(Flags::FINISHED);
+
+                                if not_pipelined && payload_unfinished {
+                                    this.flags.insert(Flags::SHUTDOWN | Flags::FINISHED);
+                                } else {
+                                    this.flags.insert(Flags::FINISHED);
+                                }
 
                                 continue 'res;
                             }