fix stuck connection when handler doesn't read payload (#2624)

actix-http/src/h1/dispatcher.rs

@@ -21,7 +21,7 @@ use crate::{
     config::ServiceConfig,
     error::{DispatchError, ParseError, PayloadError},
     service::HttpFlow,
-    Error, Extensions, OnConnectData, Request, Response, StatusCode,
+    ConnectionType, Error, Extensions, OnConnectData, Request, Response, StatusCode,
 };
 
 use super::{
@@ -151,7 +151,8 @@ pin_project! {
         error: Option<DispatchError>,
 
         #[pin]
-        state: State<S, B, X>,
+        pub(super) state: State<S, B, X>,
+        // when Some(_) dispatcher is in state of receiving request payload
         payload: Option<PayloadSender>,
         messages: VecDeque<DispatcherMessage>,
 
@@ -174,7 +175,7 @@ enum DispatcherMessage {
 
 pin_project! {
     #[project = StateProj]
-    enum State<S, B, X>
+    pub(super) enum State<S, B, X>
     where
         S: Service<Request>,
         X: Service<Request, Response = Request>,
@@ -194,7 +195,7 @@ where
     X: Service<Request, Response = Request>,
     B: MessageBody,
 {
-    fn is_none(&self) -> bool {
+    pub(super) fn is_none(&self) -> bool {
         matches!(self, State::None)
     }
 }
@@ -686,12 +687,74 @@ where
         let can_not_read = !self.can_read(cx);
 
         // limit amount of non-processed requests
-        if pipeline_queue_full || can_not_read {
+        if pipeline_queue_full {
             return Ok(false);
         }
 
         let mut this = self.as_mut().project();
 
+        if can_not_read {
+            log::debug!("cannot read request payload");
+
+            if let Some(sender) = &this.payload {
+                // ...maybe handler does not want to read any more payload...
+                if let PayloadStatus::Dropped = sender.need_read(cx) {
+                    log::debug!("handler dropped payload early; attempt to clean connection");
+                    // ...in which case poll request payload a few times
+                    loop {
+                        match this.codec.decode(this.read_buf)? {
+                            Some(msg) => {
+                                match msg {
+                                    // payload decoded did not yield EOF yet
+                                    Message::Chunk(Some(_)) => {
+                                        // if non-clean connection, next loop iter will detect empty
+                                        // read buffer and close connection
+                                    }
+
+                                    // connection is in clean state for next request
+                                    Message::Chunk(None) => {
+                                        log::debug!("connection successfully cleaned");
+
+                                        // reset dispatcher state
+                                        let _ = this.payload.take();
+                                        this.state.set(State::None);
+
+                                        // break out of payload decode loop
+                                        break;
+                                    }
+
+                                    // Either whole payload is read and loop is broken or more data
+                                    // was expected in which case connection is closed. In both
+                                    // situations dispatcher cannot get here.
+                                    Message::Item(_) => {
+                                        unreachable!("dispatcher is in payload receive state")
+                                    }
+                                }
+                            }
+
+                            // not enough info to decide if connection is going to be clean or not
+                            None => {
+                                log::error!(
+                                    "handler did not read whole payload and dispatcher could not \
+                                    drain read buf; return 500 and close connection"
+                                );
+
+                                this.flags.insert(Flags::SHUTDOWN);
+                                let mut res = Response::internal_server_error().drop_body();
+                                res.head_mut().set_connection_type(ConnectionType::Close);
+                                this.messages.push_back(DispatcherMessage::Error(res));
+                                *this.error = Some(DispatchError::HandlerDroppedPayload);
+                                return Ok(true);
+                            }
+                        }
+                    }
+                }
+            } else {
+                // can_not_read and no request payload
+                return Ok(false);
+            }
+        }
+
         let mut updated = false;
 
         loop {