fix request head timeout (#2611)

actix-http/src/h1/client.rs

@@ -1,4 +1,4 @@
-use std::io;
+use std::{fmt, io};
 
 use actix_codec::{Decoder, Encoder};
 use bitflags::bitflags;
@@ -17,9 +17,9 @@ use crate::{
 
 bitflags! {
     struct Flags: u8 {
-        const HEAD              = 0b0000_0001;
-        const KEEPALIVE_ENABLED = 0b0000_1000;
-        const STREAM            = 0b0001_0000;
+        const HEAD               = 0b0000_0001;
+        const KEEP_ALIVE_ENABLED = 0b0000_1000;
+        const STREAM             = 0b0001_0000;
     }
 }
 
@@ -38,7 +38,7 @@ struct ClientCodecInner {
     decoder: decoder::MessageDecoder<ResponseHead>,
     payload: Option<PayloadDecoder>,
     version: Version,
-    ctype: ConnectionType,
+    conn_type: ConnectionType,
 
     // encoder part
     flags: Flags,
@@ -51,23 +51,32 @@ impl Default for ClientCodec {
     }
 }
 
+impl fmt::Debug for ClientCodec {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("h1::ClientCodec")
+            .field("flags", &self.inner.flags)
+            .finish_non_exhaustive()
+    }
+}
+
 impl ClientCodec {
     /// Create HTTP/1 codec.
     ///
     /// `keepalive_enabled` how response `connection` header get generated.
     pub fn new(config: ServiceConfig) -> Self {
-        let flags = if config.keep_alive_enabled() {
-            Flags::KEEPALIVE_ENABLED
+        let flags = if config.keep_alive().enabled() {
+            Flags::KEEP_ALIVE_ENABLED
         } else {
             Flags::empty()
         };
+
         ClientCodec {
             inner: ClientCodecInner {
                 config,
                 decoder: decoder::MessageDecoder::default(),
                 payload: None,
                 version: Version::HTTP_11,
-                ctype: ConnectionType::Close,
+                conn_type: ConnectionType::Close,
 
                 flags,
                 encoder: encoder::MessageEncoder::default(),
@@ -77,12 +86,12 @@ impl ClientCodec {
 
     /// Check if request is upgrade
     pub fn upgrade(&self) -> bool {
-        self.inner.ctype == ConnectionType::Upgrade
+        self.inner.conn_type == ConnectionType::Upgrade
     }
 
     /// Check if last response is keep-alive
-    pub fn keepalive(&self) -> bool {
-        self.inner.ctype == ConnectionType::KeepAlive
+    pub fn keep_alive(&self) -> bool {
+        self.inner.conn_type == ConnectionType::KeepAlive
     }
 
     /// Check last request's message type
@@ -104,8 +113,8 @@ impl ClientCodec {
 
 impl ClientPayloadCodec {
     /// Check if last response is keep-alive
-    pub fn keepalive(&self) -> bool {
-        self.inner.ctype == ConnectionType::KeepAlive
+    pub fn keep_alive(&self) -> bool {
+        self.inner.conn_type == ConnectionType::KeepAlive
     }
 
     /// Transform payload codec to a message codec
@@ -122,12 +131,12 @@ impl Decoder for ClientCodec {
         debug_assert!(!self.inner.payload.is_some(), "Payload decoder is set");
 
         if let Some((req, payload)) = self.inner.decoder.decode(src)? {
-            if let Some(ctype) = req.conn_type() {
+            if let Some(conn_type) = req.conn_type() {
                 // do not use peer's keep-alive
-                self.inner.ctype = if ctype == ConnectionType::KeepAlive {
-                    self.inner.ctype
+                self.inner.conn_type = if conn_type == ConnectionType::KeepAlive {
+                    self.inner.conn_type
                 } else {
-                    ctype
+                    conn_type
                 };
             }
 
@@ -192,9 +201,9 @@ impl Encoder<Message<(RequestHeadType, BodySize)>> for ClientCodec {
                     .set(Flags::HEAD, head.as_ref().method == Method::HEAD);
 
                 // connection status
-                inner.ctype = match head.as_ref().connection_type() {
+                inner.conn_type = match head.as_ref().connection_type() {
                     ConnectionType::KeepAlive => {
-                        if inner.flags.contains(Flags::KEEPALIVE_ENABLED) {
+                        if inner.flags.contains(Flags::KEEP_ALIVE_ENABLED) {
                             ConnectionType::KeepAlive
                         } else {
                             ConnectionType::Close
@@ -211,7 +220,7 @@ impl Encoder<Message<(RequestHeadType, BodySize)>> for ClientCodec {
                     false,
                     inner.version,
                     length,
-                    inner.ctype,
+                    inner.conn_type,
                     &inner.config,
                 )?;
             }

actix-http/src/h1/codec.rs

@@ -15,9 +15,9 @@ use crate::{
 
 bitflags! {
     struct Flags: u8 {
-        const HEAD              = 0b0000_0001;
-        const KEEPALIVE_ENABLED = 0b0000_0010;
-        const STREAM            = 0b0000_0100;
+        const HEAD               = 0b0000_0001;
+        const KEEP_ALIVE_ENABLED = 0b0000_0010;
+        const STREAM             = 0b0000_0100;
     }
 }
 
@@ -42,7 +42,9 @@ impl Default for Codec {
 
 impl fmt::Debug for Codec {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        write!(f, "h1::Codec({:?})", self.flags)
+        f.debug_struct("h1::Codec")
+            .field("flags", &self.flags)
+            .finish_non_exhaustive()
     }
 }
 
@@ -51,8 +53,8 @@ impl Codec {
     ///
     /// `keepalive_enabled` how response `connection` header get generated.
     pub fn new(config: ServiceConfig) -> Self {
-        let flags = if config.keep_alive_enabled() {
-            Flags::KEEPALIVE_ENABLED
+        let flags = if config.keep_alive().enabled() {
+            Flags::KEEP_ALIVE_ENABLED
         } else {
             Flags::empty()
         };
@@ -76,14 +78,14 @@ impl Codec {
 
     /// Check if last response is keep-alive.
     #[inline]
-    pub fn keepalive(&self) -> bool {
+    pub fn keep_alive(&self) -> bool {
         self.conn_type == ConnectionType::KeepAlive
     }
 
     /// Check if keep-alive enabled on server level.
     #[inline]
-    pub fn keepalive_enabled(&self) -> bool {
-        self.flags.contains(Flags::KEEPALIVE_ENABLED)
+    pub fn keep_alive_enabled(&self) -> bool {
+        self.flags.contains(Flags::KEEP_ALIVE_ENABLED)
     }
 
     /// Check last request's message type.
@@ -124,7 +126,7 @@ impl Decoder for Codec {
             self.version = head.version;
             self.conn_type = head.connection_type();
             if self.conn_type == ConnectionType::KeepAlive
-                && !self.flags.contains(Flags::KEEPALIVE_ENABLED)
+                && !self.flags.contains(Flags::KEEP_ALIVE_ENABLED)
             {
                 self.conn_type = ConnectionType::Close
             }
@@ -179,9 +181,11 @@ impl Encoder<Message<(Response<()>, BodySize)>> for Codec {
                     &self.config,
                 )?;
             }
+
             Message::Chunk(Some(bytes)) => {
                 self.encoder.encode_chunk(bytes.as_ref(), dst)?;
             }
+
             Message::Chunk(None) => {
                 self.encoder.encode_eof(dst)?;
             }

actix-http/src/h1/dispatcher_tests.rs

@@ -17,7 +17,7 @@ use crate::{
     h1::{Codec, ExpectHandler, UpgradeHandler},
     service::HttpFlow,
     test::{TestBuffer, TestSeqBuffer},
-    Error, HttpMessage, KeepAlive, Method, OnConnectData, Request, Response,
+    Error, HttpMessage, KeepAlive, Method, OnConnectData, Request, Response, StatusCode,
 };
 
 fn find_slice(haystack: &[u8], needle: &[u8], from: usize) -> Option<usize> {
@@ -34,7 +34,13 @@ fn stabilize_date_header(payload: &mut [u8]) {
 }
 
 fn ok_service() -> impl Service<Request, Response = Response<impl MessageBody>, Error = Error> {
-    fn_service(|_req: Request| ready(Ok::<_, Error>(Response::ok())))
+    status_service(StatusCode::OK)
+}
+
+fn status_service(
+    status: StatusCode,
+) -> impl Service<Request, Response = Response<impl MessageBody>, Error = Error> {
+    fn_service(move |_req: Request| ready(Ok::<_, Error>(Response::new(status))))
 }
 
 fn echo_path_service(
@@ -65,11 +71,15 @@ fn echo_payload_service() -> impl Service<Request, Response = Response<Bytes>, E
 
 #[actix_rt::test]
 async fn late_request() {
-    let _ = env_logger::try_init();
-
     let mut buf = TestBuffer::empty();
 
-    let cfg = ServiceConfig::new(KeepAlive::Disabled, 100, 0, false, None);
+    let cfg = ServiceConfig::new(
+        KeepAlive::Disabled,
+        Duration::from_millis(100),
+        Duration::ZERO,
+        false,
+        None,
+    );
     let services = HttpFlow::new(ok_service(), ExpectHandler, None);
 
     let h1 = Dispatcher::<_, _, _, _, UpgradeHandler>::new(
@@ -127,10 +137,16 @@ async fn late_request() {
 }
 
 #[actix_rt::test]
-async fn test_basic() {
+async fn oneshot_connection() {
     let buf = TestBuffer::new("GET /abcd HTTP/1.1\r\n\r\n");
 
-    let cfg = ServiceConfig::new(KeepAlive::Disabled, 100, 0, false, None);
+    let cfg = ServiceConfig::new(
+        KeepAlive::Disabled,
+        Duration::from_millis(100),
+        Duration::ZERO,
+        false,
+        None,
+    );
     let services = HttpFlow::new(echo_path_service(), ExpectHandler, None);
 
     let h1 = Dispatcher::<_, _, _, _, UpgradeHandler>::new(
@@ -179,10 +195,16 @@ async fn test_basic() {
 }
 
 #[actix_rt::test]
-async fn test_keep_alive_timeout() {
+async fn keep_alive_timeout() {
     let buf = TestBuffer::new("GET /abcd HTTP/1.1\r\n\r\n");
 
-    let cfg = ServiceConfig::new(KeepAlive::Timeout(1), 100, 0, false, None);
+    let cfg = ServiceConfig::new(
+        KeepAlive::Timeout(Duration::from_millis(200)),
+        Duration::from_millis(100),
+        Duration::ZERO,
+        false,
+        None,
+    );
     let services = HttpFlow::new(echo_path_service(), ExpectHandler, None);
 
     let h1 = Dispatcher::<_, _, _, _, UpgradeHandler>::new(
@@ -229,7 +251,7 @@ async fn test_keep_alive_timeout() {
     .await;
 
     // sleep slightly longer than keep-alive timeout
-    sleep(Duration::from_millis(1100)).await;
+    sleep(Duration::from_millis(250)).await;
 
     lazy(|cx| {
         assert!(
@@ -252,10 +274,16 @@ async fn test_keep_alive_timeout() {
 }
 
 #[actix_rt::test]
-async fn test_keep_alive_follow_up_req() {
+async fn keep_alive_follow_up_req() {
     let mut buf = TestBuffer::new("GET /abcd HTTP/1.1\r\n\r\n");
 
-    let cfg = ServiceConfig::new(KeepAlive::Timeout(2), 100, 0, false, None);
+    let cfg = ServiceConfig::new(
+        KeepAlive::Timeout(Duration::from_millis(500)),
+        Duration::from_millis(100),
+        Duration::ZERO,
+        false,
+        None,
+    );
     let services = HttpFlow::new(echo_path_service(), ExpectHandler, None);
 
     let h1 = Dispatcher::<_, _, _, _, UpgradeHandler>::new(
@@ -302,7 +330,7 @@ async fn test_keep_alive_follow_up_req() {
     .await;
 
     // sleep for less than KA timeout
-    sleep(Duration::from_millis(200)).await;
+    sleep(Duration::from_millis(100)).await;
 
     lazy(|cx| {
         assert!(
@@ -371,7 +399,7 @@ async fn test_keep_alive_follow_up_req() {
 }
 
 #[actix_rt::test]
-async fn test_req_parse_err() {
+async fn req_parse_err() {
     lazy(|cx| {
         let buf = TestBuffer::new("GET /test HTTP/1\r\n\r\n");
 
@@ -413,7 +441,13 @@ async fn pipelining_ok_then_ok() {
                 ",
         );
 
-        let cfg = ServiceConfig::new(KeepAlive::Disabled, 1, 1, false, None);
+        let cfg = ServiceConfig::new(
+            KeepAlive::Disabled,
+            Duration::from_millis(1),
+            Duration::from_millis(1),
+            false,
+            None,
+        );
 
         let services = HttpFlow::new(echo_path_service(), ExpectHandler, None);
 
@@ -477,7 +511,13 @@ async fn pipelining_ok_then_bad() {
                 ",
         );
 
-        let cfg = ServiceConfig::new(KeepAlive::Disabled, 1, 1, false, None);
+        let cfg = ServiceConfig::new(
+            KeepAlive::Disabled,
+            Duration::from_millis(1),
+            Duration::from_millis(1),
+            false,
+            None,
+        );
 
         let services = HttpFlow::new(echo_path_service(), ExpectHandler, None);
 
@@ -531,10 +571,16 @@ async fn pipelining_ok_then_bad() {
 }
 
 #[actix_rt::test]
-async fn test_expect() {
+async fn expect_handling() {
     lazy(|cx| {
         let mut buf = TestSeqBuffer::empty();
-        let cfg = ServiceConfig::new(KeepAlive::Disabled, 0, 0, false, None);
+        let cfg = ServiceConfig::new(
+            KeepAlive::Disabled,
+            Duration::ZERO,
+            Duration::ZERO,
+            false,
+            None,
+        );
 
         let services = HttpFlow::new(echo_payload_service(), ExpectHandler, None);
 
@@ -562,7 +608,6 @@ async fn test_expect() {
 
         // polls: manual
         assert_eq!(h1.poll_count, 1);
-        eprintln!("poll count: {}", h1.poll_count);
 
         if let DispatcherState::Normal { ref inner } = h1.inner {
             let io = inner.io.as_ref().unwrap();
@@ -603,10 +648,16 @@ async fn test_expect() {
 }
 
 #[actix_rt::test]
-async fn test_eager_expect() {
+async fn expect_eager() {
     lazy(|cx| {
         let mut buf = TestSeqBuffer::empty();
-        let cfg = ServiceConfig::new(KeepAlive::Disabled, 0, 0, false, None);
+        let cfg = ServiceConfig::new(
+            KeepAlive::Disabled,
+            Duration::ZERO,
+            Duration::ZERO,
+            false,
+            None,
+        );
 
         let services = HttpFlow::new(echo_path_service(), ExpectHandler, None);
 
@@ -663,7 +714,7 @@ async fn test_eager_expect() {
 }
 
 #[actix_rt::test]
-async fn test_upgrade() {
+async fn upgrade_handling() {
     struct TestUpgrade;
 
     impl<T> Service<(Request, Framed<T, Codec>)> for TestUpgrade {
@@ -683,7 +734,13 @@ async fn test_upgrade() {
 
     lazy(|cx| {
         let mut buf = TestSeqBuffer::empty();
-        let cfg = ServiceConfig::new(KeepAlive::Disabled, 0, 0, false, None);
+        let cfg = ServiceConfig::new(
+            KeepAlive::Disabled,
+            Duration::ZERO,
+            Duration::ZERO,
+            false,
+            None,
+        );
 
         let services = HttpFlow::new(ok_service(), ExpectHandler, Some(TestUpgrade));
 

actix-http/src/h1/encoder.rs

@@ -212,7 +212,7 @@ pub(crate) trait MessageType: Sized {
 
         // optimized date header, set_date writes \r\n
         if !has_date {
-            config.set_date(dst, camel_case);
+            config.write_date_header(dst, camel_case);
         } else {
             // msg eof
             dst.extend_from_slice(b"\r\n");
@@ -318,16 +318,17 @@ impl MessageType for RequestHeadType {
 }
 
 impl<T: MessageType> MessageEncoder<T> {
-    /// Encode message
+    /// Encode chunk.
     pub fn encode_chunk(&mut self, msg: &[u8], buf: &mut BytesMut) -> io::Result<bool> {
         self.te.encode(msg, buf)
     }
 
-    /// Encode eof
+    /// Encode EOF.
     pub fn encode_eof(&mut self, buf: &mut BytesMut) -> io::Result<()> {
         self.te.encode_eof(buf)
     }
 
+    /// Encode message.
     pub fn encode(
         &mut self,
         dst: &mut BytesMut,

actix-http/src/h1/mod.rs

@@ -13,6 +13,7 @@ mod encoder;
 mod expect;
 mod payload;
 mod service;
+mod timer;
 mod upgrade;
 mod utils;
 
@@ -28,9 +29,10 @@ pub use self::utils::SendResponse;
 #[derive(Debug)]
 /// Codec message
 pub enum Message<T> {
-    /// Http message
+    /// HTTP message.
     Item(T),
-    /// Payload chunk
+
+    /// Payload chunk.
     Chunk(Option<Bytes>),
 }
 

actix-http/src/h1/timer.rs

@@ -0,0 +1,80 @@
+use std::{fmt, future::Future, pin::Pin, task::Context};
+
+use actix_rt::time::{Instant, Sleep};
+
+#[derive(Debug)]
+pub(super) enum TimerState {
+    Disabled,
+    Inactive,
+    Active { timer: Pin<Box<Sleep>> },
+}
+
+impl TimerState {
+    pub(super) fn new(enabled: bool) -> Self {
+        if enabled {
+            Self::Inactive
+        } else {
+            Self::Disabled
+        }
+    }
+
+    pub(super) fn is_enabled(&self) -> bool {
+        matches!(self, Self::Active { .. } | Self::Inactive)
+    }
+
+    pub(super) fn set(&mut self, timer: Sleep, line: u32) {
+        if matches!(self, Self::Disabled) {
+            log::trace!("setting disabled timer from line {}", line);
+        }
+
+        *self = Self::Active {
+            timer: Box::pin(timer),
+        };
+    }
+
+    pub(super) fn set_and_init(&mut self, cx: &mut Context<'_>, timer: Sleep, line: u32) {
+        self.set(timer, line);
+        self.init(cx);
+    }
+
+    pub(super) fn clear(&mut self, line: u32) {
+        if matches!(self, Self::Disabled) {
+            log::trace!("trying to clear a disabled timer from line {}", line);
+        }
+
+        if matches!(self, Self::Inactive) {
+            log::trace!("trying to clear an inactive timer from line {}", line);
+        }
+
+        *self = Self::Inactive;
+    }
+
+    pub(super) fn init(&mut self, cx: &mut Context<'_>) {
+        if let TimerState::Active { timer } = self {
+            let _ = timer.as_mut().poll(cx);
+        }
+    }
+}
+
+impl fmt::Display for TimerState {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            TimerState::Disabled => f.write_str("timer is disabled"),
+            TimerState::Inactive => f.write_str("timer is inactive"),
+            TimerState::Active { timer } => {
+                let deadline = timer.deadline();
+                let now = Instant::now();
+
+                if deadline < now {
+                    f.write_str("timer is active and has reached deadline")
+                } else {
+                    write!(
+                        f,
+                        "timer is active and due to expire in {} milliseconds",
+                        ((deadline - now).as_secs_f32() * 1000.0)
+                    )
+                }
+            }
+        }
+    }
+}