Return PayloadError::Incomplete on server disconnect (#545)

ntex-io/src/tasks.rs

@@ -537,7 +537,9 @@ impl IoContext {
                                 self.0.tag(),
                                 nbytes
                             );
-                            inner.dispatch_task.wake();
+                            if !inner.dispatch_task.wake_checked() {
+                                log::error!("Dispatcher waker is not registered");
+                            }
                         } else {
                             if nbytes >= hw {
                                 // read task is paused because of read back-pressure
@@ -735,22 +737,6 @@ impl IoContext {
         false
     }
 
-    pub fn is_write_ready(&self) -> bool {
-        if let Some(waker) = self.0 .0.write_task.take() {
-            let ready = self
-                .0
-                .filter()
-                .poll_write_ready(&mut Context::from_waker(&waker));
-            if !matches!(
-                ready,
-                Poll::Ready(WriteStatus::Ready | WriteStatus::Shutdown)
-            ) {
-                return true;
-            }
-        }
-        false
-    }
-
     pub fn with_read_buf<F>(&self, f: F) -> Poll<()>
     where
         F: FnOnce(&mut BytesVec) -> Poll<io::Result<usize>>,
@@ -803,7 +789,9 @@ impl IoContext {
                                 self.0.tag(),
                                 nbytes
                             );
-                            inner.dispatch_task.wake();
+                            if !inner.dispatch_task.wake_checked() {
+                                log::error!("Dispatcher waker is not registered");
+                            }
                         } else {
                             if nbytes >= hw {
                                 // read task is paused because of read back-pressure

ntex-rt/src/lib.rs

@@ -112,6 +112,8 @@ mod tokio {
     ///
     /// This function panics if ntex system is not running.
     #[inline]
+    #[doc(hidden)]
+    #[deprecated]
     pub fn spawn_fn<F, R>(f: F) -> tok_io::task::JoinHandle<R::Output>
     where
         F: FnOnce() -> R + 'static,
@@ -196,6 +198,8 @@ mod compio {
     ///
     /// This function panics if ntex system is not running.
     #[inline]
+    #[doc(hidden)]
+    #[deprecated]
     pub fn spawn_fn<F, R>(f: F) -> JoinHandle<R::Output>
     where
         F: FnOnce() -> R + 'static,
@@ -323,6 +327,8 @@ mod neon {
     ///
     /// This function panics if ntex system is not running.
     #[inline]
+    #[doc(hidden)]
+    #[deprecated]
     pub fn spawn_fn<F, R>(f: F) -> Task<R::Output>
     where
         F: FnOnce() -> R + 'static,
@@ -377,7 +383,7 @@ mod neon {
 
     impl<T> JoinHandle<T> {
         pub fn is_finished(&self) -> bool {
-            false
+            self.fut.is_none()
         }
     }
 

ntex/CHANGES.md

@@ -1,5 +1,11 @@
 # Changes
 
+## [2.12.4] - 2025-03-28
+
+* http: Return PayloadError::Incomplete on server disconnect
+
+* web: Expose WebStack for external wrapper support in downstream crates #542
+
 ## [2.12.3] - 2025-03-22
 
 * web: Export web::app_service::AppService #534

ntex/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex"
-version = "2.12.3"
+version = "2.12.4"
 authors = ["ntex contributors <team@ntex.rs>"]
 description = "Framework for composable network services"
 readme = "README.md"

ntex/src/http/client/h1proto.rs

@@ -1,13 +1,11 @@
-use std::{
+use std::{future::poll_fn, io, io::Write, pin::Pin, task, task::Poll, time::Instant};
-    future::poll_fn, io, io::Write, pin::Pin, task::Context, task::Poll, time::Instant,
-};
 
 use crate::http::body::{BodySize, MessageBody};
 use crate::http::error::PayloadError;
-use crate::http::h1;
 use crate::http::header::{HeaderMap, HeaderValue, HOST};
 use crate::http::message::{RequestHeadType, ResponseHead};
 use crate::http::payload::{Payload, PayloadStream};
+use crate::http::{h1, Version};
 use crate::io::{IoBoxed, RecvError};
 use crate::time::{timeout_checked, Millis};
 use crate::util::{ready, BufMut, Bytes, BytesMut, Stream};
@@ -101,7 +99,13 @@ where
             Ok((head, Payload::None))
         }
         _ => {
-            let pl: PayloadStream = Box::pin(PlStream::new(io, codec, created, pool));
+            let pl: PayloadStream = Box::pin(PlStream::new(
+                io,
+                codec,
+                created,
+                pool,
+                head.version == Version::HTTP_10,
+            ));
             Ok((head, pl.into()))
         }
     }
@@ -137,6 +141,7 @@ pub(super) struct PlStream {
     io: Option<IoBoxed>,
     codec: h1::ClientPayloadCodec,
     created: Instant,
+    http_10: bool,
     pool: Option<Acquired>,
 }
 
@@ -146,12 +151,14 @@ impl PlStream {
         codec: h1::ClientCodec,
         created: Instant,
         pool: Option<Acquired>,
+        http_10: bool,
     ) -> Self {
         PlStream {
             io: Some(io),
             codec: codec.into_payload_codec(),
             created,
             pool,
+            http_10,
         }
     }
 }
@@ -161,12 +168,12 @@ impl Stream for PlStream {
 
     fn poll_next(
         mut self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
+        cx: &mut task::Context<'_>,
     ) -> Poll<Option<Self::Item>> {
         let mut this = self.as_mut();
         loop {
-            return Poll::Ready(Some(
+            let item = ready!(this.io.as_ref().unwrap().poll_recv(&this.codec, cx));
-                match ready!(this.io.as_ref().unwrap().poll_recv(&this.codec, cx)) {
+            return Poll::Ready(Some(match item {
                 Ok(chunk) => {
                     if let Some(chunk) = chunk {
                         Ok(chunk)
@@ -184,18 +191,23 @@ impl Stream for PlStream {
                     Err(io::Error::new(io::ErrorKind::TimedOut, "Keep-alive").into())
                 }
                 Err(RecvError::Stop) => {
-                        Err(io::Error::new(io::ErrorKind::Other, "Dispatcher stopped")
+                    Err(io::Error::new(io::ErrorKind::Other, "Dispatcher stopped").into())
-                            .into())
                 }
                 Err(RecvError::WriteBackpressure) => {
                     ready!(this.io.as_ref().unwrap().poll_flush(cx, false))?;
                     continue;
                 }
                 Err(RecvError::Decoder(err)) => Err(err),
-                    Err(RecvError::PeerGone(Some(err))) => Err(err.into()),
+                Err(RecvError::PeerGone(Some(err))) => {
-                    Err(RecvError::PeerGone(None)) => return Poll::Ready(None),
+                    Err(PayloadError::Incomplete(Some(err)))
-                },
+                }
-            ));
+                Err(RecvError::PeerGone(None)) => {
+                    if this.http_10 {
+                        return Poll::Ready(None);
+                    }
+                    Err(PayloadError::Incomplete(None))
+                }
+            }));
         }
     }
 }

ntex/src/http/client/response.rs

@@ -387,8 +387,8 @@ impl Future for ReadBody {
         let this = self.get_mut();
 
         loop {
-            return match Pin::new(&mut this.stream).poll_next(cx)? {
+            return match Pin::new(&mut this.stream).poll_next(cx) {
-                Poll::Ready(Some(chunk)) => {
+                Poll::Ready(Some(Ok(chunk))) => {
                     if this.limit > 0 && (this.buf.len() + chunk.len()) > this.limit {
                         Poll::Ready(Err(PayloadError::Overflow))
                     } else {
@@ -397,6 +397,7 @@ impl Future for ReadBody {
                     }
                 }
                 Poll::Ready(None) => Poll::Ready(Ok(this.buf.split().freeze())),
+                Poll::Ready(Some(Err(err))) => Poll::Ready(Err(err)),
                 Poll::Pending => {
                     if this.timeout.poll_elapsed(cx).is_ready() {
                         Poll::Ready(Err(PayloadError::Incomplete(Some(