fix: prevent reuse of the stream after an error

When a stream timeouts, `tokio_io_timeout::TimeoutStream`
returns an error once, but then allows to keep using
the stream, e.g. calling `poll_read()` again.

This can be dangerous if the error is ignored.
For example in case of IMAP stream,
if IMAP command is sent,
but then reading the response
times out and the error is ignored,
it is possible to send another IMAP command.
In this case leftover response
from a previous command may be read
and interpreted as the response
to the new IMAP command.

ErrorCapturingStream wraps the stream
to prevent its reuse after an error.

CHANGELOG.md

@@ -1,20 +1,5 @@
 # Changelog
 
-## [2.4.0] - 2025-07-21
-
-### Fixes
-
-- Do not ignore errors when draining FETCH responses. This avoids IMAP loop getting stuck in an infinite loop retrying reading from the connection.
-- Update `tokio-io-timeout` to 1.2.1. This release includes a fix to reset timeout after every error, so timeout error is returned at most once a minute if read is attempted after a timeout.
-
-### Miscellaneous Tasks
-
-- Update async-imap to 0.11.0.
-
-### Refactor
-
-- Use `try_next()` when processing FETCH responses.
-
 ## [2.3.0] - 2025-07-19
 
 ### Features / Changes
@@ -6510,4 +6495,3 @@ https://github.com/chatmail/core/pulls?q=is%3Apr+is%3Aclosed
 [2.1.0]: https://github.com/chatmail/core/compare/v2.0.0..v2.1.0
 [2.2.0]: https://github.com/chatmail/core/compare/v2.1.0..v2.2.0
 [2.3.0]: https://github.com/chatmail/core/compare/v2.2.0..v2.3.0
-[2.4.0]: https://github.com/chatmail/core/compare/v2.3.0..v2.4.0

Cargo.lock

@@ -268,9 +268,9 @@ dependencies = [
 
 [[package]]
 name = "async-imap"
-version = "0.11.0"
+version = "0.10.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8e9f9a9c94a403cf46aa2b4cecbceefc6e4284441ebbeca79b80f3bab4394458"
+checksum = "ca726c61b73c471f531b65e83e161776ba62c2b6ba4ec73d51fad357009ed00a"
 dependencies = [
  "async-channel 2.3.1",
  "async-compression",
@@ -1285,7 +1285,7 @@ dependencies = [
 
 [[package]]
 name = "deltachat"
-version = "2.4.0"
+version = "2.3.0"
 dependencies = [
  "anyhow",
  "async-broadcast",
@@ -1395,7 +1395,7 @@ dependencies = [
 
 [[package]]
 name = "deltachat-jsonrpc"
-version = "2.4.0"
+version = "2.3.0"
 dependencies = [
  "anyhow",
  "async-channel 2.3.1",
@@ -1417,7 +1417,7 @@ dependencies = [
 
 [[package]]
 name = "deltachat-repl"
-version = "2.4.0"
+version = "2.3.0"
 dependencies = [
  "anyhow",
  "deltachat",
@@ -1433,7 +1433,7 @@ dependencies = [
 
 [[package]]
 name = "deltachat-rpc-server"
-version = "2.4.0"
+version = "2.3.0"
 dependencies = [
  "anyhow",
  "deltachat",
@@ -1462,7 +1462,7 @@ dependencies = [
 
 [[package]]
 name = "deltachat_ffi"
-version = "2.4.0"
+version = "2.3.0"
 dependencies = [
  "anyhow",
  "deltachat",
@@ -3826,7 +3826,7 @@ version = "0.7.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "af1844ef2428cc3e1cb900be36181049ef3d3193c63e43026cfe202983b27a56"
 dependencies = [
- "proc-macro-crate 3.2.0",
+ "proc-macro-crate 2.0.0",
  "proc-macro2",
  "quote",
  "syn 2.0.104",
@@ -6073,9 +6073,9 @@ dependencies = [
 
 [[package]]
 name = "tokio-io-timeout"
-version = "1.2.1"
+version = "1.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0bd86198d9ee903fedd2f9a2e72014287c0d9167e4ae43b5853007205dda1b76"
+checksum = "30b74022ada614a1b4834de765f9bb43877f910cc8ce4be40e89042c9223a8bf"
 dependencies = [
  "pin-project-lite",
  "tokio",

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "deltachat"
-version = "2.4.0"
+version = "2.3.0"
 edition = "2024"
 license = "MPL-2.0"
 rust-version = "1.85"
@@ -44,7 +44,7 @@ ratelimit = { path = "./deltachat-ratelimit" }
 anyhow = { workspace = true }
 async-broadcast = "0.7.2"
 async-channel = { workspace = true }
-async-imap = { version = "0.11.0", default-features = false, features = ["runtime-tokio", "compress"] }
+async-imap = { version = "0.10.4", default-features = false, features = ["runtime-tokio", "compress"] }
 async-native-tls = { version = "0.5", default-features = false, features = ["runtime-tokio"] }
 async-smtp = { version = "0.10.2", default-features = false, features = ["runtime-tokio"] }
 async_zip = { version = "0.0.17", default-features = false, features = ["deflate", "tokio-fs"] }
@@ -101,7 +101,7 @@ strum_macros = "0.27"
 tagger = "4.3.4"
 textwrap = "0.16.2"
 thiserror = { workspace = true }
-tokio-io-timeout = "1.2.1"
+tokio-io-timeout = "1.2.0"
 tokio-rustls = { version = "0.26.2", default-features = false }
 tokio-stream = { version = "0.1.17", features = ["fs"] }
 tokio-tar = { version = "0.3" } # TODO: integrate tokio into async-tar

deltachat-ffi/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "deltachat_ffi"
-version = "2.4.0"
+version = "2.3.0"
 description = "Deltachat FFI"
 edition = "2018"
 readme = "README.md"

deltachat-jsonrpc/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "deltachat-jsonrpc"
-version = "2.4.0"
+version = "2.3.0"
 description = "DeltaChat JSON-RPC API"
 edition = "2021"
 license = "MPL-2.0"

deltachat-jsonrpc/typescript/package.json

@@ -54,5 +54,5 @@
   },
   "type": "module",
   "types": "dist/deltachat.d.ts",
-  "version": "2.4.0"
+  "version": "2.3.0"
 }

deltachat-repl/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "deltachat-repl"
-version = "2.4.0"
+version = "2.3.0"
 license = "MPL-2.0"
 edition = "2021"
 repository = "https://github.com/chatmail/core"

deltachat-rpc-client/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "deltachat-rpc-client"
-version = "2.4.0"
+version = "2.3.0"
 description = "Python client for Delta Chat core JSON-RPC interface"
 classifiers = [
     "Development Status :: 5 - Production/Stable",

deltachat-rpc-server/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "deltachat-rpc-server"
-version = "2.4.0"
+version = "2.3.0"
 description = "DeltaChat JSON-RPC server"
 edition = "2021"
 readme = "README.md"

deltachat-rpc-server/npm-package/package.json

@@ -15,5 +15,5 @@
   },
   "type": "module",
   "types": "index.d.ts",
-  "version": "2.4.0"
+  "version": "2.3.0"
 }

deny.toml

@@ -33,6 +33,7 @@ skip = [
      { name = "lru", version = "0.12.3" },
      { name = "netlink-packet-route", version = "0.17.1" },
      { name = "nom", version = "7.1.3" },
+     { name = "proc-macro-crate", version = "2.0.0" },
      { name = "rand_chacha", version = "0.3.1" },
      { name = "rand_core", version = "0.6.4" },
      { name = "rand", version = "0.8.5" },
@@ -48,6 +49,7 @@ skip = [
      { name = "syn", version = "1.0.109" },
      { name = "thiserror-impl", version = "1.0.69" },
      { name = "thiserror", version = "1.0.69" },
+     { name = "toml_edit", version = "0.20.7" },
      { name = "wasi", version = "0.11.0+wasi-snapshot-preview1" },
      { name = "windows" },
      { name = "windows_aarch64_gnullvm" },
@@ -65,6 +67,7 @@ skip = [
      { name = "windows_x86_64_gnu" },
      { name = "windows_x86_64_gnullvm" },
      { name = "windows_x86_64_msvc" },
+     { name = "winnow", version = "0.5.40" },
      { name = "zerocopy", version = "0.7.32" },
 ]
 

python/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "deltachat"
-version = "2.4.0"
+version = "2.3.0"
 description = "Python bindings for the Delta Chat Core library using CFFI against the Rust-implemented libdeltachat"
 readme = "README.rst"
 requires-python = ">=3.8"

release-date.in

@@ -1 +1 @@
-2025-07-21
+2025-07-19

src/imap.rs

@@ -1384,15 +1384,14 @@ impl Session {
 
                 // Try to find a requested UID in returned FETCH responses.
                 while fetch_response.is_none() {
-                    let Some(next_fetch_response) = fetch_responses
-                        .try_next()
-                        .await
-                        .context("Failed to process IMAP FETCH result")?
-                    else {
+                    let Some(next_fetch_response) = fetch_responses.next().await else {
                         // No more FETCH responses received from the server.
                         break;
                     };
 
+                    let next_fetch_response =
+                        next_fetch_response.context("Failed to process IMAP FETCH result")?;
+
                     if let Some(next_uid) = next_fetch_response.uid {
                         if next_uid == request_uid {
                             fetch_response = Some(next_fetch_response);
@@ -1492,16 +1491,7 @@ impl Session {
 
             // If we don't process the whole response, IMAP client is left in a broken state where
             // it will try to process the rest of response as the next response.
-            //
-            // Make sure to not ignore the errors, because
-            // if connection times out, it will return
-            // infinite stream of `Some(Err(_))` results.
-            while fetch_responses
-                .try_next()
-                .await
-                .context("Failed to drain FETCH responses")?
-                .is_some()
-            {}
+            while fetch_responses.next().await.is_some() {}
 
             if count != request_uids.len() {
                 warn!(

src/imap/client.rs

@@ -216,8 +216,8 @@ impl Client {
         let mut client = Client::new(session_stream);
         let _greeting = client
             .read_response()
-            .await?
-            .context("Failed to read greeting")?;
+            .await
+            .context("failed to read greeting")??;
         Ok(client)
     }
 
@@ -231,8 +231,8 @@ impl Client {
         let mut client = Client::new(session_stream);
         let _greeting = client
             .read_response()
-            .await?
-            .context("Failed to read greeting")?;
+            .await
+            .context("failed to read greeting")??;
         Ok(client)
     }
 
@@ -253,8 +253,8 @@ impl Client {
         let mut client = async_imap::Client::new(buffered_tcp_stream);
         let _greeting = client
             .read_response()
-            .await?
-            .context("Failed to read greeting")?;
+            .await
+            .context("failed to read greeting")??;
         client
             .run_command_and_check_ok("STARTTLS", None)
             .await
@@ -287,8 +287,8 @@ impl Client {
         let mut client = Client::new(session_stream);
         let _greeting = client
             .read_response()
-            .await?
-            .context("Failed to read greeting")?;
+            .await
+            .context("failed to read greeting")??;
         Ok(client)
     }
 
@@ -304,8 +304,8 @@ impl Client {
         let mut client = Client::new(session_stream);
         let _greeting = client
             .read_response()
-            .await?
-            .context("Failed to read greeting")?;
+            .await
+            .context("failed to read greeting")??;
         Ok(client)
     }
 
@@ -325,8 +325,8 @@ impl Client {
         let mut client = ImapClient::new(buffered_proxy_stream);
         let _greeting = client
             .read_response()
-            .await?
-            .context("Failed to read greeting")?;
+            .await
+            .context("failed to read greeting")??;
         client
             .run_command_and_check_ok("STARTTLS", None)
             .await

src/net.rs

@@ -16,12 +16,14 @@ use crate::sql::Sql;
 use crate::tools::time;
 
 pub(crate) mod dns;
+pub(crate) mod error_capturing_stream;
 pub(crate) mod http;
 pub(crate) mod proxy;
 pub(crate) mod session;
 pub(crate) mod tls;
 
 use dns::lookup_host_with_cache;
+pub(crate) use error_capturing_stream::ErrorCapturingStream;
 pub use http::{Response as HttpResponse, read_url, read_url_blob};
 use tls::wrap_tls;
 
@@ -105,7 +107,7 @@ pub(crate) async fn load_connection_timestamp(
 /// to the network, which is important to reduce the latency of interactive protocols such as IMAP.
 pub(crate) async fn connect_tcp_inner(
     addr: SocketAddr,
-) -> Result<Pin<Box<TimeoutStream<TcpStream>>>> {
+) -> Result<Pin<Box<ErrorCapturingStream<TimeoutStream<TcpStream>>>>> {
     let tcp_stream = timeout(TIMEOUT, TcpStream::connect(addr))
         .await
         .context("connection timeout")?
@@ -118,7 +120,9 @@ pub(crate) async fn connect_tcp_inner(
     timeout_stream.set_write_timeout(Some(TIMEOUT));
     timeout_stream.set_read_timeout(Some(TIMEOUT));
 
-    Ok(Box::pin(timeout_stream))
+    let error_capturing_stream = ErrorCapturingStream::new(timeout_stream);
+
+    Ok(Box::pin(error_capturing_stream))
 }
 
 /// Attempts to establish TLS connection
@@ -235,7 +239,7 @@ pub(crate) async fn connect_tcp(
     host: &str,
     port: u16,
     load_cache: bool,
-) -> Result<Pin<Box<TimeoutStream<TcpStream>>>> {
+) -> Result<Pin<Box<ErrorCapturingStream<TimeoutStream<TcpStream>>>>> {
     let connection_futures = lookup_host_with_cache(context, host, port, "", load_cache)
         .await?
         .into_iter()

src/net/error_capturing_stream.rs

@@ -0,0 +1,136 @@
+use std::io::IoSlice;
+use std::net::SocketAddr;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+use std::time::Duration;
+use tokio::io::{self, AsyncRead, AsyncWrite, ReadBuf};
+
+use pin_project::pin_project;
+
+use crate::net::SessionStream;
+
+/// Stream that remembers the first error
+/// and keeps returning it afterwards.
+///
+/// It is needed to avoid accidentally using
+/// the stream after read timeout.
+#[derive(Debug)]
+#[pin_project]
+pub(crate) struct ErrorCapturingStream<T: AsyncRead + AsyncWrite + std::fmt::Debug> {
+    #[pin]
+    inner: T,
+
+    /// If true, the stream has already returned an error once.
+    ///
+    /// All read and write operations return error in this case.
+    is_broken: bool,
+}
+
+impl<T: AsyncRead + AsyncWrite + std::fmt::Debug> ErrorCapturingStream<T> {
+    pub fn new(inner: T) -> Self {
+        Self {
+            inner,
+            is_broken: false,
+        }
+    }
+
+    /// Gets a reference to the underlying stream.
+    pub fn get_ref(&self) -> &T {
+        &self.inner
+    }
+
+    /// Gets a pinned mutable reference to the underlying stream.
+    pub fn get_pin_mut(self: Pin<&mut Self>) -> Pin<&mut T> {
+        self.project().inner
+    }
+}
+
+impl<T: AsyncRead + AsyncWrite + std::fmt::Debug> AsyncRead for ErrorCapturingStream<T> {
+    fn poll_read(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut ReadBuf,
+    ) -> Poll<io::Result<()>> {
+        let this = self.project();
+        if *this.is_broken {
+            return Poll::Ready(Err(io::Error::other("Broken stream")));
+        }
+        let res = this.inner.poll_read(cx, buf);
+        if let Poll::Ready(Err(_)) = res {
+            *this.is_broken = true;
+        }
+        res
+    }
+}
+
+impl<T: AsyncRead + AsyncWrite + std::fmt::Debug> AsyncWrite for ErrorCapturingStream<T> {
+    fn poll_write(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<io::Result<usize>> {
+        let this = self.project();
+        if *this.is_broken {
+            return Poll::Ready(Err(io::Error::other("Broken stream")));
+        }
+        let res = this.inner.poll_write(cx, buf);
+        if let Poll::Ready(Err(_)) = res {
+            *this.is_broken = true;
+        }
+        res
+    }
+
+    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        let this = self.project();
+        if *this.is_broken {
+            return Poll::Ready(Err(io::Error::other("Broken stream")));
+        }
+        let res = this.inner.poll_flush(cx);
+        if let Poll::Ready(Err(_)) = res {
+            *this.is_broken = true;
+        }
+        res
+    }
+
+    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        let this = self.project();
+        if *this.is_broken {
+            return Poll::Ready(Err(io::Error::other("Broken stream")));
+        }
+        let res = this.inner.poll_shutdown(cx);
+        if let Poll::Ready(Err(_)) = res {
+            *this.is_broken = true;
+        }
+        res
+    }
+
+    fn poll_write_vectored(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        bufs: &[IoSlice<'_>],
+    ) -> Poll<io::Result<usize>> {
+        let this = self.project();
+        if *this.is_broken {
+            return Poll::Ready(Err(io::Error::other("Broken stream")));
+        }
+        let res = this.inner.poll_write_vectored(cx, bufs);
+        if let Poll::Ready(Err(_)) = res {
+            *this.is_broken = true;
+        }
+        res
+    }
+
+    fn is_write_vectored(&self) -> bool {
+        self.inner.is_write_vectored()
+    }
+}
+
+impl<T: SessionStream> SessionStream for ErrorCapturingStream<T> {
+    fn set_read_timeout(&mut self, timeout: Option<Duration>) {
+        self.inner.set_read_timeout(timeout)
+    }
+
+    fn peer_addr(&self) -> anyhow::Result<SocketAddr> {
+        self.inner.peer_addr()
+    }
+}

src/net/proxy.rs

@@ -21,9 +21,9 @@ use url::Url;
 use crate::config::Config;
 use crate::constants::NON_ALPHANUMERIC_WITHOUT_DOT;
 use crate::context::Context;
-use crate::net::connect_tcp;
 use crate::net::session::SessionStream;
 use crate::net::tls::wrap_rustls;
+use crate::net::{ErrorCapturingStream, connect_tcp};
 use crate::sql::Sql;
 
 /// Default SOCKS5 port according to [RFC 1928](https://tools.ietf.org/html/rfc1928).
@@ -118,7 +118,7 @@ impl Socks5Config {
         target_host: &str,
         target_port: u16,
         load_dns_cache: bool,
-    ) -> Result<Socks5Stream<Pin<Box<TimeoutStream<TcpStream>>>>> {
+    ) -> Result<Socks5Stream<Pin<Box<ErrorCapturingStream<TimeoutStream<TcpStream>>>>>> {
         let tcp_stream = connect_tcp(context, &self.host, self.port, load_dns_cache)
             .await
             .context("Failed to connect to SOCKS5 proxy")?;

src/net/session.rs

@@ -7,6 +7,8 @@ use tokio::io::{AsyncBufRead, AsyncRead, AsyncWrite, BufStream, BufWriter};
 use tokio::net::TcpStream;
 use tokio_io_timeout::TimeoutStream;
 
+use crate::net::ErrorCapturingStream;
+
 pub(crate) trait SessionStream:
     AsyncRead + AsyncWrite + Unpin + Send + Sync + std::fmt::Debug
 {
@@ -61,13 +63,13 @@ impl<T: SessionStream> SessionStream for BufWriter<T> {
         self.get_ref().peer_addr()
     }
 }
-impl SessionStream for Pin<Box<TimeoutStream<TcpStream>>> {
+impl SessionStream for Pin<Box<ErrorCapturingStream<TimeoutStream<TcpStream>>>> {
     fn set_read_timeout(&mut self, timeout: Option<Duration>) {
-        self.as_mut().set_read_timeout_pinned(timeout);
+        self.as_mut().get_pin_mut().set_read_timeout_pinned(timeout);
     }
 
     fn peer_addr(&self) -> Result<SocketAddr> {
-        Ok(self.get_ref().peer_addr()?)
+        Ok(self.get_ref().get_ref().peer_addr()?)
     }
 }
 impl<T: SessionStream> SessionStream for Socks5Stream<T> {