move blocking error to web (#2660)

actix-http/src/encoding/decoder.rs

@@ -19,7 +19,7 @@ use zstd::stream::write::Decoder as ZstdDecoder;
 
 use crate::{
     encoding::Writer,
-    error::{BlockingError, PayloadError},
+    error::PayloadError,
     header::{ContentEncoding, HeaderMap, CONTENT_ENCODING},
 };
 
@@ -47,14 +47,17 @@ where
             ContentEncoding::Brotli => Some(ContentDecoder::Brotli(Box::new(
                 brotli::DecompressorWriter::new(Writer::new(), 8_096),
             ))),
+
             #[cfg(feature = "compress-gzip")]
             ContentEncoding::Deflate => Some(ContentDecoder::Deflate(Box::new(
                 ZlibDecoder::new(Writer::new()),
             ))),
+
             #[cfg(feature = "compress-gzip")]
             ContentEncoding::Gzip => Some(ContentDecoder::Gzip(Box::new(GzDecoder::new(
                 Writer::new(),
             )))),
+
             #[cfg(feature = "compress-zstd")]
             ContentEncoding::Zstd => Some(ContentDecoder::Zstd(Box::new(
                 ZstdDecoder::new(Writer::new()).expect(
@@ -98,8 +101,12 @@ where
 
         loop {
             if let Some(ref mut fut) = this.fut {
-                let (chunk, decoder) =
-                    ready!(Pin::new(fut).poll(cx)).map_err(|_| BlockingError)??;
+                let (chunk, decoder) = ready!(Pin::new(fut).poll(cx)).map_err(|_| {
+                    PayloadError::Io(io::Error::new(
+                        io::ErrorKind::Other,
+                        "Blocking task was cancelled unexpectedly",
+                    ))
+                })??;
 
                 *this.decoder = Some(decoder);
                 this.fut.take();
@@ -159,10 +166,13 @@ where
 enum ContentDecoder {
     #[cfg(feature = "compress-gzip")]
     Deflate(Box<ZlibDecoder<Writer>>),
+
     #[cfg(feature = "compress-gzip")]
     Gzip(Box<GzDecoder<Writer>>),
+
     #[cfg(feature = "compress-brotli")]
     Brotli(Box<brotli::DecompressorWriter<Writer>>),
+
     // We need explicit 'static lifetime here because ZstdDecoder need lifetime
     // argument, and we use `spawn_blocking` in `Decoder::poll_next` that require `FnOnce() -> R + Send + 'static`
     #[cfg(feature = "compress-zstd")]

actix-http/src/encoding/encoder.rs

@@ -23,7 +23,6 @@ use zstd::stream::write::Encoder as ZstdEncoder;
 use super::Writer;
 use crate::{
     body::{self, BodySize, MessageBody},
-    error::BlockingError,
     header::{self, ContentEncoding, HeaderValue, CONTENT_ENCODING},
     ResponseHead, StatusCode,
 };
@@ -173,7 +172,12 @@ where
 
             if let Some(ref mut fut) = this.fut {
                 let mut encoder = ready!(Pin::new(fut).poll(cx))
-                    .map_err(|_| EncoderError::Blocking(BlockingError))?
+                    .map_err(|_| {
+                        EncoderError::Io(io::Error::new(
+                            io::ErrorKind::Other,
+                            "Blocking task was cancelled unexpectedly",
+                        ))
+                    })?
                     .map_err(EncoderError::Io)?;
 
                 let chunk = encoder.take();
@@ -400,12 +404,11 @@ fn new_brotli_compressor() -> Box<brotli::CompressorWriter<Writer>> {
 #[derive(Debug, Display)]
 #[non_exhaustive]
 pub enum EncoderError {
+    /// Wrapped body stream error.
     #[display(fmt = "body")]
     Body(Box<dyn StdError>),
 
-    #[display(fmt = "blocking")]
-    Blocking(BlockingError),
-
+    /// Generic I/O error.
     #[display(fmt = "io")]
     Io(io::Error),
 }
@@ -414,7 +417,6 @@ impl StdError for EncoderError {
     fn source(&self) -> Option<&(dyn StdError + 'static)> {
         match self {
             EncoderError::Body(err) => Some(&**err),
-            EncoderError::Blocking(err) => Some(err),
             EncoderError::Io(err) => Some(err),
         }
     }

actix-http/src/error.rs

@@ -51,7 +51,7 @@ impl Error {
         Self::new(Kind::SendResponse)
     }
 
-    #[allow(unused)] // reserved for future use (TODO: remove allow when being used)
+    #[allow(unused)] // available for future use
     pub(crate) fn new_io() -> Self {
         Self::new(Kind::Io)
     }
@@ -252,12 +252,6 @@ impl From<ParseError> for Response<BoxBody> {
     }
 }
 
-/// A set of errors that can occur running blocking tasks in thread pool.
-#[derive(Debug, Display, Error)]
-#[display(fmt = "Blocking thread pool is gone")]
-// TODO: non-exhaustive
-pub struct BlockingError;
-
 /// A set of errors that can occur during payload parsing.
 #[derive(Debug, Display)]
 #[non_exhaustive]
@@ -295,13 +289,13 @@ impl std::error::Error for PayloadError {
     fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
         match self {
             PayloadError::Incomplete(None) => None,
-            PayloadError::Incomplete(Some(err)) => Some(err as &dyn std::error::Error),
+            PayloadError::Incomplete(Some(err)) => Some(err),
             PayloadError::EncodingCorrupted => None,
             PayloadError::Overflow => None,
             PayloadError::UnknownLength => None,
             #[cfg(feature = "http2")]
-            PayloadError::Http2Payload(err) => Some(err as &dyn std::error::Error),
-            PayloadError::Io(err) => Some(err as &dyn std::error::Error),
+            PayloadError::Http2Payload(err) => Some(err),
+            PayloadError::Io(err) => Some(err),
         }
     }
 }
@@ -325,15 +319,6 @@ impl From<io::Error> for PayloadError {
     }
 }
 
-impl From<BlockingError> for PayloadError {
-    fn from(_: BlockingError) -> Self {
-        PayloadError::Io(io::Error::new(
-            io::ErrorKind::Other,
-            "Operation is canceled",
-        ))
-    }
-}
-
 impl From<PayloadError> for Error {
     fn from(err: PayloadError) -> Self {
         Self::new_payload().with_cause(err)