add send error response state

add changelog

add send error response state

remove comment

actix-files/Cargo.toml

@@ -46,6 +46,6 @@ actix-server = { version = "2.1", optional = true } # ensure matching tokio-urin
 
 [dev-dependencies]
 actix-rt = "2.7"
-actix-test = "0.1.0-beta.13"
+actix-test = "0.1"
 actix-web = "4"
 tempfile = "3.2"

actix-http-test/CHANGES.md

@@ -1,9 +1,24 @@
 # Changes
 
 ## Unreleased - 2022-xx-xx
-- Minimum supported Rust version (MSRV) is now 1.57 due to transitive `time` dependency.
 
 
+## 3.0.0 - 2022-07-24
+- `TestServer::stop` is now async and will wait for the server and system to shutdown. [#2442]
+- Added `TestServer::client_headers` method. [#2097]
+- Update `actix-server` dependency to `2`.
+- Update `actix-tls` dependency to `3`.
+- Update `bytes` to `1.0`. [#1813]
+- Minimum supported Rust version (MSRV) is now 1.57.
+
+[#2442]: https://github.com/actix/actix-web/pull/2442
+[#2097]: https://github.com/actix/actix-web/pull/2097
+[#1813]: https://github.com/actix/actix-web/pull/1813
+
+
+<details>
+<summary>3.0.0 Pre-Releases</summary>
+
 ## 3.0.0-beta.13 - 2022-02-16
 - No significant changes since `3.0.0-beta.12`.
 
@@ -69,6 +84,7 @@
 
 [#1813]: https://github.com/actix/actix-web/pull/1813
 
+</details>
 
 ## 2.1.0 - 2020-11-25
 - Add ability to set address for `TestServer`. [#1645]

actix-http-test/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-http-test"
-version = "3.0.0-beta.13"
+version = "3.0.0"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Various helpers for Actix applications to use during testing"
 keywords = ["http", "web", "framework", "async", "futures"]

actix-http-test/README.md

@@ -3,11 +3,11 @@
 > Various helpers for Actix applications to use during testing.
 
 [![crates.io](https://img.shields.io/crates/v/actix-http-test?label=latest)](https://crates.io/crates/actix-http-test)
-[![Documentation](https://docs.rs/actix-http-test/badge.svg?version=3.0.0-beta.13)](https://docs.rs/actix-http-test/3.0.0-beta.13)
+[![Documentation](https://docs.rs/actix-http-test/badge.svg?version=3.0.0)](https://docs.rs/actix-http-test/3.0.0)
 ![Version](https://img.shields.io/badge/rustc-1.57+-ab6000.svg)
 ![MIT or Apache 2.0 licensed](https://img.shields.io/crates/l/actix-http-test)
 <br>
-[![Dependency Status](https://deps.rs/crate/actix-http-test/3.0.0-beta.13/status.svg)](https://deps.rs/crate/actix-http-test/3.0.0-beta.13)
+[![Dependency Status](https://deps.rs/crate/actix-http-test/3.0.0/status.svg)](https://deps.rs/crate/actix-http-test/3.0.0)
 [![Download](https://img.shields.io/crates/d/actix-http-test.svg)](https://crates.io/crates/actix-http-test)
 [![Chat on Discord](https://img.shields.io/discord/771444961383153695?label=chat&logo=discord)](https://discord.gg/NWpN5mmg3x)
 

actix-http/CHANGES.md

@@ -1,6 +1,10 @@
 # Changes
 
 ## Unreleased - 2022-xx-xx
+### Fixed
+- Dropping the payload early and causing unclean connections no longer causes erroneous 500 responses. [#2745]
+
+[#2745]: https://github.com/actix/actix-web/issues/2745
 
 
 ## 3.2.1 - 2022-07-02

actix-http/Cargo.toml

@@ -97,7 +97,7 @@ flate2 = { version = "1.0.13", optional = true }
 zstd = { version = "0.11", optional = true }
 
 [dev-dependencies]
-actix-http-test = { version = "3.0.0-beta.13", features = ["openssl"] }
+actix-http-test = { version = "3", features = ["openssl"] }
 actix-server = "2"
 actix-tls = { version = "3", features = ["openssl"] }
 actix-web = "4"

actix-http/src/h1/dispatcher.rs

@@ -22,7 +22,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::{
@@ -185,7 +185,9 @@ pin_project! {
         None,
         ExpectCall { #[pin] fut: X::Future },
         ServiceCall { #[pin] fut: S::Future },
+        SendResponse { res: Option<Response<B>> },
         SendPayload { #[pin] body: B },
+        SendErrorResponse { res: Option<Response<BoxBody>> },
         SendErrorPayload { #[pin] body: BoxBody },
     }
 }
@@ -216,9 +218,15 @@ where
             Self::ServiceCall { .. } => {
                 f.debug_struct("State::ServiceCall").finish_non_exhaustive()
             }
+            Self::SendResponse { .. } => f
+                .debug_struct("State::SendResponse")
+                .finish_non_exhaustive(),
             Self::SendPayload { .. } => {
                 f.debug_struct("State::SendPayload").finish_non_exhaustive()
             }
+            Self::SendErrorResponse { .. } => f
+                .debug_struct("State::SendErrorResponse")
+                .finish_non_exhaustive(),
             Self::SendErrorPayload { .. } => f
                 .debug_struct("State::SendErrorPayload")
                 .finish_non_exhaustive(),
@@ -379,11 +387,8 @@ where
         Ok(size)
     }
 
-    fn send_response(
-        mut self: Pin<&mut Self>,
-        res: Response<()>,
-        body: B,
-    ) -> Result<(), DispatchError> {
+    fn send_response(mut self: Pin<&mut Self>, res: Response<B>) -> Result<(), DispatchError> {
+        let (res, body) = res.replace_body(());
         let size = self.as_mut().send_response_inner(res, &body)?;
         let mut this = self.project();
         this.state.set(match size {
@@ -397,11 +402,17 @@ where
         Ok(())
     }
 
+    fn queue_response(self: Pin<&mut Self>, res: Response<B>) {
+        self.project()
+            .state
+            .set(State::SendResponse { res: Some(res) });
+    }
+
     fn send_error_response(
         mut self: Pin<&mut Self>,
-        res: Response<()>,
-        body: BoxBody,
+        res: Response<BoxBody>,
     ) -> Result<(), DispatchError> {
+        let (res, body) = res.replace_body(());
         let size = self.as_mut().send_response_inner(res, &body)?;
         let mut this = self.project();
         this.state.set(match size {
@@ -415,6 +426,12 @@ where
         Ok(())
     }
 
+    fn queue_error_response(self: Pin<&mut Self>, res: Response<BoxBody>) {
+        self.project()
+            .state
+            .set(State::SendErrorResponse { res: Some(res) });
+    }
+
     fn send_continue(self: Pin<&mut Self>) {
         self.project()
             .write_buf
@@ -449,7 +466,8 @@ where
                         // send_response would update InnerDispatcher state to SendPayload or None
                         // (If response body is empty)
                         // continue loop to poll it
-                        self.as_mut().send_error_response(res, BoxBody::new(()))?;
+                        self.as_mut()
+                            .queue_error_response(res.set_body(BoxBody::new(())));
                     }
 
                     // return with upgrade request and poll it exclusively
@@ -470,15 +488,12 @@ where
                     match fut.poll(cx) {
                         // service call resolved. send response.
                         Poll::Ready(Ok(res)) => {
-                            let (res, body) = res.into().replace_body(());
-                            self.as_mut().send_response(res, body)?;
+                            self.as_mut().queue_response(res.into());
                         }
 
                         // send service call error as response
                         Poll::Ready(Err(err)) => {
-                            let res: Response<BoxBody> = err.into();
-                            let (res, body) = res.replace_body(());
-                            self.as_mut().send_error_response(res, body)?;
+                            self.as_mut().queue_error_response(err.into());
                         }
 
                         // service call pending and could be waiting for more chunk messages
@@ -486,14 +501,32 @@ where
                         Poll::Pending => {
                             // no new message is decoded and no new payload is fed
                             // nothing to do except waiting for new incoming data from client
-                            if !self.as_mut().poll_request(cx)? {
-                                return Ok(PollResponse::DoNothing);
-                            }
+
+                            // optimisation disabled so that poll_request is called from only one place
+                            // if !self.as_mut().poll_request(cx)? {
+                            return Ok(PollResponse::DoNothing);
+                            // }
+
                             // else loop
                         }
                     }
                 }
 
+                StateProj::SendResponse { res } => {
+                    let mut res = res.take().expect("response should be take-able");
+
+                    if this.flags.contains(Flags::SHUTDOWN) {
+                        trace!("shutdown flag set; assuming dirty read I/O");
+                        // shutdown flags occur when read I/O is not clean so connections should be
+                        // closed to avoid stuck or erroneous errors on next request
+                        res.head_mut().set_connection_type(ConnectionType::Close);
+                    }
+
+                    self.send_response(res)?;
+
+                    return Ok(PollResponse::DrainWriteBuf);
+                }
+
                 StateProj::SendPayload { mut body } => {
                     // keep populate writer buffer until buffer size limit hit,
                     // get blocked or finished.
@@ -529,6 +562,23 @@ where
                     return Ok(PollResponse::DrainWriteBuf);
                 }
 
+                StateProj::SendErrorResponse { res } => {
+                    // TODO: de-dupe impl with SendResponse
+
+                    let mut res = res.take().expect("response should be take-able");
+
+                    if this.flags.contains(Flags::SHUTDOWN) {
+                        trace!("shutdown flag set; assuming dirty read I/O");
+                        // shutdown flags occur when read I/O is not clean so connections should be
+                        // closed to avoid stuck or erroneous errors on next request
+                        res.head_mut().set_connection_type(ConnectionType::Close);
+                    }
+
+                    self.send_error_response(res)?;
+
+                    return Ok(PollResponse::DrainWriteBuf);
+                }
+
                 StateProj::SendErrorPayload { mut body } => {
                     // TODO: de-dupe impl with SendPayload
 
@@ -583,9 +633,7 @@ where
 
                         // send expect error as response
                         Poll::Ready(Err(err)) => {
-                            let res: Response<BoxBody> = err.into();
-                            let (res, body) = res.replace_body(());
-                            self.as_mut().send_error_response(res, body)?;
+                            self.as_mut().queue_error_response(err.into());
                         }
 
                         // expect must be solved before progress can be made.
@@ -637,9 +685,8 @@ where
                         // on success to notify the dispatcher a new state is set and the outer loop
                         // should be continued
                         Poll::Ready(Err(err)) => {
-                            let res: Response<BoxBody> = err.into();
-                            let (res, body) = res.replace_body(());
-                            return self.send_error_response(res, body);
+                            self.queue_error_response(err.into());
+                            return Ok(());
                         }
 
                         // future is pending; return Ok(()) to notify that a new state is
@@ -655,8 +702,8 @@ where
                         // to notify the dispatcher a new state is set and the outer loop
                         // should be continue.
                         Poll::Ready(Ok(res)) => {
-                            let (res, body) = res.into().replace_body(());
-                            self.as_mut().send_response(res, body)
+                            self.as_mut().queue_response(res.into());
+                            Ok(())
                         }
 
                         // see the comment on ExpectCall state branch's Pending
@@ -664,9 +711,8 @@ where
 
                         // see the comment on ExpectCall state branch's Ready(Err(_))
                         Poll::Ready(Err(err)) => {
-                            let res: Response<BoxBody> = err.into();
-                            let (res, body) = res.replace_body(());
-                            self.as_mut().send_error_response(res, body)
+                            self.as_mut().queue_error_response(err.into());
+                            Ok(())
                         }
                     };
                 }
@@ -688,15 +734,13 @@ where
         cx: &mut Context<'_>,
     ) -> Result<bool, DispatchError> {
         let pipeline_queue_full = self.messages.len() >= MAX_PIPELINED_MESSAGES;
-        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();
-
         let mut updated = false;
 
         // decode from read buf as many full requests as possible
@@ -829,6 +873,72 @@ where
             }
         }
 
+        let can_read = self.can_read(cx);
+        let mut this = self.as_mut().project();
+
+        if !can_read {
+            // request payload is not readable...
+            tracing::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) {
+                    tracing::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) => {
+                                        tracing::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 => {
+                                tracing::debug!(
+                                    "handler did not read whole payload and dispatcher could not \
+                                    drain read buf; close connection"
+                                );
+
+                                this.flags.insert(Flags::SHUTDOWN);
+
+                                return Ok(updated);
+                            }
+                        }
+                    }
+                }
+            } else {
+                // can_not_read and no request payload
+                return Ok(false);
+            }
+        }
+
         Ok(updated)
     }
 
@@ -844,10 +954,10 @@ where
 
                 trace!("timed out on slow request; replying with 408 and closing connection");
 
-                let _ = self.as_mut().send_error_response(
-                    Response::with_body(StatusCode::REQUEST_TIMEOUT, ()),
-                    BoxBody::new(()),
-                );
+                let mut res =
+                    Response::with_body(StatusCode::REQUEST_TIMEOUT, BoxBody::new(()));
+                res.head_mut().set_connection_type(ConnectionType::Close);
+                self.as_mut().send_error_response(res)?;
 
                 self.project().flags.insert(Flags::SHUTDOWN);
             }
@@ -1123,6 +1233,7 @@ where
                         }
                     }
 
+                    // process request(s) and queue response
                     inner.as_mut().poll_request(cx)?;
 
                     if should_disconnect {

actix-http/src/h1/dispatcher_tests.rs

@@ -861,56 +861,26 @@ async fn handler_drop_payload() {
             ",
         ));
 
-        assert!(h1.as_mut().poll(cx).is_pending());
-
-        // polls: manual => manual
-        assert_eq!(h1.poll_count, 2);
-
-        let mut res = BytesMut::from(buf.take_write_buf().as_ref());
-        stabilize_date_header(&mut res);
-        let res = &res[..];
-
-        // expect response immediately even though request side has not finished reading payload
-        let exp = http_msg(
-            r"
-            HTTP/1.1 200 OK
-            content-length: 15
-            date: Thu, 01 Jan 1970 12:34:56 UTC
-
-            payload dropped
-            ",
-        );
-
-        assert_eq!(
-            res,
-            exp,
-            "\nexpected response not in write buffer:\n\
-               response: {:?}\n\
-               expected: {:?}",
-            String::from_utf8_lossy(res),
-            String::from_utf8_lossy(&exp)
-        );
-    })
-    .await;
-
-    lazy(|cx| {
         assert!(h1.as_mut().poll(cx).is_ready());
 
-        // polls: manual => manual => manual
+        // polls: manual => manual => shutdown
         assert_eq!(h1.poll_count, 3);
 
         let mut res = BytesMut::from(buf.take_write_buf().as_ref());
         stabilize_date_header(&mut res);
         let res = &res[..];
 
-        // expect that unrequested error response is sent back since connection could not be cleaned
+        // expect response immediately even though request side has not finished reading payload
+        // since write buffer was "too short" we should expect a closed connection hint
         let exp = http_msg(
+            // connection: close
             r"
-            HTTP/1.1 500 Internal Server Error
-            content-length: 0
+            HTTP/1.1 200 OK
+            content-length: 15
             connection: close
             date: Thu, 01 Jan 1970 12:34:56 UTC
 
+            payload dropped
             ",
         );
 

actix-http/src/h1/payload.rs

@@ -252,19 +252,15 @@ impl Inner {
 
 #[cfg(test)]
 mod tests {
-    use std::panic::{RefUnwindSafe, UnwindSafe};
-
     use actix_utils::future::poll_fn;
     use static_assertions::{assert_impl_all, assert_not_impl_any};
 
     use super::*;
 
     assert_impl_all!(Payload: Unpin);
-    assert_not_impl_any!(Payload: Send, Sync, UnwindSafe, RefUnwindSafe);
+    assert_not_impl_any!(Payload: Send, Sync);
 
     assert_impl_all!(Inner: Unpin, Send, Sync);
-    // assertion not stable wrt rustc versions yet
-    // assert_impl_all!(Inner: UnwindSafe, RefUnwindSafe);
 
     #[actix_rt::test]
     async fn test_unread_data() {

actix-http/src/h2/mod.rs

@@ -103,11 +103,9 @@ where
 
 #[cfg(test)]
 mod tests {
-    use std::panic::{RefUnwindSafe, UnwindSafe};
-
     use static_assertions::assert_impl_all;
 
     use super::*;
 
-    assert_impl_all!(Payload: Unpin, Send, Sync, UnwindSafe, RefUnwindSafe);
+    assert_impl_all!(Payload: Unpin, Send, Sync);
 }

actix-http/src/payload.rs

@@ -97,12 +97,10 @@ where
 
 #[cfg(test)]
 mod tests {
-    use std::panic::{RefUnwindSafe, UnwindSafe};
-
     use static_assertions::{assert_impl_all, assert_not_impl_any};
 
     use super::*;
 
     assert_impl_all!(Payload: Unpin);
-    assert_not_impl_any!(Payload: Send, Sync, UnwindSafe, RefUnwindSafe);
+    assert_not_impl_any!(Payload: Send, Sync);
 }

actix-test/CHANGES.md

@@ -1,6 +1,9 @@
 # Changes
 
 ## Unreleased - 2022-xx-xx
+
+
+## 0.1.0 - 2022-07-24
 - Minimum supported Rust version (MSRV) is now 1.57 due to transitive `time` dependency.
 
 

actix-test/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-test"
-version = "0.1.0-beta.13"
+version = "0.1.0"
 authors = [
     "Nikolay Kim <fafhrd91@gmail.com>",
     "Rob Ede <robjtede@icloud.com>",
@@ -30,7 +30,7 @@ openssl = ["tls-openssl", "actix-http/openssl", "awc/openssl"]
 [dependencies]
 actix-codec = "0.5"
 actix-http = "3"
-actix-http-test = "3.0.0-beta.13"
+actix-http-test = "3"
 actix-rt = "2.1"
 actix-service = "2"
 actix-utils = "3"

actix-web-actors/Cargo.toml

@@ -27,7 +27,7 @@ tokio = { version = "1.13.1", features = ["sync"] }
 
 [dev-dependencies]
 actix-rt = "2.2"
-actix-test = "0.1.0-beta.13"
+actix-test = "0.1"
 awc = { version = "3", default-features = false }
 actix-web = { version = "4", features = ["macros"] }
 

actix-web-codegen/Cargo.toml

@@ -23,7 +23,7 @@ syn = { version = "1", features = ["full", "extra-traits"] }
 [dev-dependencies]
 actix-macros = "0.2.3"
 actix-rt = "2.2"
-actix-test = "0.1.0-beta.13"
+actix-test = "0.1"
 actix-utils = "3.0.0"
 actix-web = "4.0.0"
 

actix-web/Cargo.toml

@@ -102,7 +102,7 @@ url = "2.1"
 
 [dev-dependencies]
 actix-files = "0.6"
-actix-test = { version = "0.1.0-beta.13", features = ["openssl", "rustls"] }
+actix-test = { version = "0.1", features = ["openssl", "rustls"] }
 awc = { version = "3", features = ["openssl"] }
 
 brotli = "3.3.3"

awc/Cargo.toml

@@ -94,9 +94,9 @@ trust-dns-resolver = { version = "0.21", optional = true }
 
 [dev-dependencies]
 actix-http = { version = "3", features = ["openssl"] }
-actix-http-test = { version = "3.0.0-beta.13", features = ["openssl"] }
+actix-http-test = { version = "3", features = ["openssl"] }
 actix-server = "2"
-actix-test = { version = "0.1.0-beta.13", features = ["openssl", "rustls"] }
+actix-test = { version = "0.1", features = ["openssl", "rustls"] }
 actix-tls = { version = "3", features = ["openssl", "rustls"] }
 actix-utils = "3"
 actix-web = { version = "4", features = ["openssl"] }