v1.0.1 (closes #2)

CHANGELOG.md

@@ -2,6 +2,20 @@
 
 All notable changes to this project will be documented in this file. This project adheres to [Semantic Versioning](http://semver.org/).
 
+## [1.0.1] - 2026-07-30
+
+### Added
+- `server.host_down_min_clients`: number of distinct client addresses a
+  failure streak must span before client-attributable failures can mark a destination
+  down.
+- `GET /destinations` now also reports `failure_sources` and `host_level_failure`.
+
+### Changed
+
+### Fixed
+- The circuit breaker no longer marks a destination down because a backend rejects
+  the connections of a single banned client (#2).
+
 ## [1.0.0] - 2026-06-08
 
 ### Added

Cargo.toml

@@ -1,8 +1,7 @@
 [package]
 name = "proxy"
-version = "1.0.0"
+version = "1.0.1"
 edition = "2024"
-rust-version = "1.85"
 license = "AGPL-3.0-only OR LicenseRef-SEL"
 description = "Multi-protocol e-mail proxy"
 authors = ["Stalwart Labs LLC <hello@stalw.art>"]
@@ -17,22 +16,22 @@ name = "proxy"
 path = "src/main.rs"
 
 [dependencies]
-tokio = { version = "1.52", features = ["net", "io-util", "rt-multi-thread", "macros", "signal", "sync", "time"] }
+tokio = { version = "1.53", features = ["net", "io-util", "rt-multi-thread", "macros", "signal", "sync", "time"] }
 rustls = { version = "0.23", default-features = false, features = ["std", "aws_lc_rs", "tls12"] }
 tokio-rustls = { version = "0.26", default-features = false, features = ["aws_lc_rs", "tls12"] }
 rustls-pemfile = "2.2"
-rustls-pki-types = "1.14"
+rustls-pki-types = "1.15"
 aws-lc-rs = "1.17"
 proxy-header = { version = "0.1.2", features = ["tokio"] }
-smtp-proto = "0.2.1"
+smtp-proto = "0.2.2"
 mail-parser = "0.11"
-base64 = "0.22"
+base64 = "0.23"
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 toml = "1"
 globset = "0.4"
-regex = "1.12"
-redis = { version = "1.2", features = ["tokio-comp", "aio", "connection-manager"] }
+regex = "1.13"
+redis = { version = "1.5", features = ["tokio-comp", "aio", "connection-manager"] }
 sqlx = { version = "0.9", default-features = false, features = ["runtime-tokio", "tls-rustls-aws-lc-rs", "any", "postgres", "mysql", "sqlite"] }
 tracing = "0.1"
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }

src/admin.rs

@@ -237,6 +237,8 @@ async fn handle(
                     serde_json::json!({
                         "destination": id,
                         "consecutive_failures": s.consecutive_failures,
+                        "failure_sources": s.failure_sources,
+                        "host_level_failure": s.host_level_failure,
                         "down": s.down,
                     })
                 })

src/config/mod.rs

@@ -140,6 +140,8 @@ pub struct ServerConfig {
     pub backend_connect_retries: u32,
     #[serde(default = "default_host_down_threshold")]
     pub host_down_threshold: u32,
+    #[serde(default = "default_host_down_min_clients")]
+    pub host_down_min_clients: u32,
     #[serde(default = "default_host_down_cooldown", with = "duration")]
     pub host_down_cooldown: Duration,
     #[serde(default = "default_proxy_ttl")]
@@ -159,6 +161,7 @@ impl Default for ServerConfig {
             backend_timeout: DEFAULT_BACKEND_TIMEOUT,
             backend_connect_retries: DEFAULT_BACKEND_CONNECT_RETRIES,
             host_down_threshold: DEFAULT_HOST_DOWN_THRESHOLD,
+            host_down_min_clients: DEFAULT_HOST_DOWN_MIN_CLIENTS,
             host_down_cooldown: DEFAULT_HOST_DOWN_COOLDOWN,
             proxy_ttl: DEFAULT_PROXY_TTL,
             log_level: DEFAULT_LOG_LEVEL.to_string(),
@@ -181,6 +184,7 @@ const DEFAULT_BRIDGE_IDLE: Duration = Duration::from_secs(1800);
 const DEFAULT_BACKEND_TIMEOUT: Duration = Duration::from_secs(10);
 const DEFAULT_BACKEND_CONNECT_RETRIES: u32 = 1;
 const DEFAULT_HOST_DOWN_THRESHOLD: u32 = 5;
+const DEFAULT_HOST_DOWN_MIN_CLIENTS: u32 = 2;
 const DEFAULT_HOST_DOWN_COOLDOWN: Duration = Duration::from_secs(30);
 const DEFAULT_PROXY_TTL: u32 = 7;
 const DEFAULT_LOG_LEVEL: &str = "info";
@@ -200,6 +204,9 @@ fn default_backend_connect_retries() -> u32 {
 fn default_host_down_threshold() -> u32 {
     DEFAULT_HOST_DOWN_THRESHOLD
 }
+fn default_host_down_min_clients() -> u32 {
+    DEFAULT_HOST_DOWN_MIN_CLIENTS
+}
 fn default_host_down_cooldown() -> Duration {
     DEFAULT_HOST_DOWN_COOLDOWN
 }

src/error.rs

@@ -25,6 +25,9 @@ pub enum ProxyError {
     #[error("backend connection failed: {0}")]
     BackendConnect(Cow<'static, str>),
 
+    #[error("backend session failed: {0}")]
+    BackendSession(Cow<'static, str>),
+
     #[error("protocol error: {0}")]
     Protocol(Cow<'static, str>),
 
@@ -52,8 +55,15 @@ impl ProxyError {
         ProxyError::BackendConnect(msg.into())
     }
 
+    pub fn backend_session(msg: impl Into<Cow<'static, str>>) -> Self {
+        ProxyError::BackendSession(msg.into())
+    }
+
     pub fn is_retryable(&self) -> bool {
-        matches!(self, ProxyError::BackendConnect(_) | ProxyError::Io(_))
+        matches!(
+            self,
+            ProxyError::BackendConnect(_) | ProxyError::BackendSession(_) | ProxyError::Io(_)
+        )
     }
 
     pub fn protocol(msg: impl Into<Cow<'static, str>>) -> Self {

src/http/proxy.rs

@@ -502,7 +502,7 @@ async fn serve(
         let _ = write_client(stream, &error_response(502, "Bad Gateway")).await;
         return Ok(Disposition::Close);
     }
-    let mut backend = match crate::proto::common::establish(ctx, &dest_id, || {
+    let mut backend = match crate::proto::common::establish(ctx, &dest_id, peer, || {
         crate::outbound::dial(dest, ep, fwd, &tls_cfg, peer, local)
     })
     .await

src/outbound.rs

@@ -134,7 +134,7 @@ pub async fn tls_connect(
     let tls = connector
         .connect(name, stream)
         .await
-        .map_err(|e| ProxyError::backend_connect(format!("backend TLS handshake failed: {e}")))?;
+        .map_err(|e| ProxyError::backend_session(format!("backend TLS handshake failed: {e}")))?;
     Ok(Box::new(tls))
 }
 
@@ -160,9 +160,9 @@ pub async fn read_line(
         let n = stream
             .read(&mut buf)
             .await
-            .map_err(|e| ProxyError::backend_connect(format!("reading backend: {e}")))?;
+            .map_err(|e| ProxyError::backend_session(format!("reading backend: {e}")))?;
         if n == 0 {
-            return Err(ProxyError::backend_connect(
+            return Err(ProxyError::backend_session(
                 "backend closed before sending a complete line",
             ));
         }

src/proto/common.rs

@@ -6,7 +6,7 @@
 
 use std::collections::HashMap;
 use std::future::Future;
-use std::net::SocketAddr;
+use std::net::{IpAddr, SocketAddr};
 use std::sync::Arc;
 use std::time::{Duration, Instant};
 
@@ -141,9 +141,35 @@ pub struct DestHealth {
 #[derive(Default)]
 struct HealthState {
     consecutive_failures: u32,
+    failure_sources: AHashSet<IpAddr>,
+    host_level_failure: bool,
     down_until: Option<Instant>,
 }
 
+impl HealthState {
+    fn clear_evidence(&mut self) {
+        self.failure_sources.clear();
+        self.host_level_failure = false;
+    }
+}
+
+const MAX_TRACKED_FAILURE_SOURCES: usize = 64;
+
+#[derive(Clone, Copy, PartialEq, Eq, Debug)]
+pub enum Attribution {
+    Host,
+    Client(IpAddr),
+}
+
+impl Attribution {
+    fn of(err: &ProxyError, peer: IpAddr) -> Self {
+        match err {
+            ProxyError::BackendConnect(_) => Attribution::Host,
+            _ => Attribution::Client(peer),
+        }
+    }
+}
+
 impl DestHealth {
     fn gate(&self) -> Result<()> {
         let mut st = self.inner.lock();
@@ -154,26 +180,39 @@ impl DestHealth {
                 ));
             }
             st.down_until = None;
+            st.clear_evidence();
         }
         Ok(())
     }
 
-    fn record(&self, ok: bool, threshold: u32, cooldown: Duration) {
+    fn record_success(&self) {
         let mut st = self.inner.lock();
-        if ok {
-            st.consecutive_failures = 0;
-            st.down_until = None;
-        } else {
-            st.consecutive_failures = st.consecutive_failures.saturating_add(1);
-            if threshold > 0 && st.consecutive_failures >= threshold {
-                st.down_until = Some(Instant::now() + cooldown);
+        st.consecutive_failures = 0;
+        st.clear_evidence();
+        st.down_until = None;
+    }
+
+    fn record_failure(&self, at: Attribution, limits: &HealthLimits) {
+        let mut st = self.inner.lock();
+        st.consecutive_failures = st.consecutive_failures.saturating_add(1);
+        match at {
+            Attribution::Host => st.host_level_failure = true,
+            Attribution::Client(ip) => {
+                if st.failure_sources.len() < MAX_TRACKED_FAILURE_SOURCES {
+                    st.failure_sources.insert(ip);
+                }
             }
         }
+        let broad = st.host_level_failure || st.failure_sources.len() as u32 >= limits.min_clients;
+        if limits.threshold > 0 && st.consecutive_failures >= limits.threshold && broad {
+            st.down_until = Some(Instant::now() + limits.cooldown);
+        }
     }
 
     pub fn reset(&self) {
         let mut st = self.inner.lock();
         st.consecutive_failures = 0;
+        st.clear_evidence();
         st.down_until = None;
     }
 
@@ -182,13 +221,33 @@ impl DestHealth {
         let down = matches!(st.down_until, Some(until) if Instant::now() < until);
         HealthSnapshot {
             consecutive_failures: st.consecutive_failures,
+            failure_sources: st.failure_sources.len() as u32,
+            host_level_failure: st.host_level_failure,
             down,
         }
     }
 }
 
+pub struct HealthLimits {
+    threshold: u32,
+    min_clients: u32,
+    cooldown: Duration,
+}
+
+impl HealthLimits {
+    fn for_server(server: &crate::config::ServerConfig) -> Self {
+        HealthLimits {
+            threshold: server.host_down_threshold,
+            min_clients: server.host_down_min_clients,
+            cooldown: server.host_down_cooldown,
+        }
+    }
+}
+
 pub struct HealthSnapshot {
     pub consecutive_failures: u32,
+    pub failure_sources: u32,
+    pub host_level_failure: bool,
     pub down: bool,
 }
 
@@ -465,7 +524,12 @@ pub fn assert_egress_encrypted(
 
 const BACKEND_RETRY_DELAY: Duration = Duration::from_millis(200);
 
-pub async fn establish<F, Fut, T>(ctx: &Ctx, dest_id: &str, mut open: F) -> Result<T>
+pub async fn establish<F, Fut, T>(
+    ctx: &Ctx,
+    dest_id: &str,
+    peer: SocketAddr,
+    mut open: F,
+) -> Result<T>
 where
     F: FnMut() -> Fut,
     Fut: Future<Output = Result<T>>,
@@ -475,6 +539,7 @@ where
     }
 
     let server = &ctx.config.server;
+    let limits = HealthLimits::for_server(server);
     let mut attempt: u32 = 0;
     loop {
         let outcome = match tokio::time::timeout(server.backend_timeout, open()).await {
@@ -487,7 +552,7 @@ where
         match outcome {
             Ok(value) => {
                 if let Some(health) = ctx.dest_health.get(dest_id) {
-                    health.record(true, server.host_down_threshold, server.host_down_cooldown);
+                    health.record_success();
                 }
                 return Ok(value);
             }
@@ -504,7 +569,15 @@ where
                     continue;
                 }
                 if let Some(health) = ctx.dest_health.get(dest_id) {
-                    health.record(false, server.host_down_threshold, server.host_down_cooldown);
+                    let at = Attribution::of(&err, peer.ip());
+                    if matches!(at, Attribution::Client(_)) {
+                        tracing::debug!(
+                            destination = %dest_id,
+                            error = %err,
+                            "backend failure attributed to the client, not counted as host evidence on its own"
+                        );
+                    }
+                    health.record_failure(at, &limits);
                 }
                 return Err(err);
             }
@@ -659,6 +732,7 @@ pub async fn bridge_authenticated<F, Fut>(
     ctx: &Ctx,
     dest_id: &str,
     mut stream: BoxedStream,
+    peer: SocketAddr,
     identifier: Option<&str>,
     client_residual: &[u8],
     started: Instant,
@@ -685,7 +759,7 @@ where
         return Ok(());
     }
 
-    let (mut backend, mut backend_residual) = match establish(ctx, dest_id, open).await {
+    let (mut backend, mut backend_residual) = match establish(ctx, dest_id, peer, open).await {
         Ok(b) => b,
         Err(e) => {
             tracing::warn!(error = %e, protocol = label, destination = %dest_id, "backend unavailable");
@@ -854,29 +928,189 @@ mod tests {
         ));
     }
 
+    fn limits(threshold: u32, min_clients: u32, cooldown: Duration) -> HealthLimits {
+        HealthLimits {
+            threshold,
+            min_clients,
+            cooldown,
+        }
+    }
+
+    fn client(addr: &str) -> Attribution {
+        Attribution::Client(addr.parse().unwrap())
+    }
+
     #[test]
     fn dest_health_opens_after_threshold_and_recovers() {
         let h = DestHealth::default();
-        let cooldown = Duration::from_secs(60);
+        let lim = limits(3, 2, Duration::from_secs(60));
 
-        h.record(false, 3, cooldown);
-        h.record(false, 3, cooldown);
+        h.record_failure(Attribution::Host, &lim);
+        h.record_failure(Attribution::Host, &lim);
         assert!(h.gate().is_ok());
 
-        h.record(false, 3, cooldown);
+        h.record_failure(Attribution::Host, &lim);
         assert!(h.gate().is_err());
 
-        h.record(true, 3, cooldown);
+        h.record_success();
         assert!(h.gate().is_ok());
     }
 
     #[test]
     fn dest_health_cooldown_elapses_into_a_probe() {
         let h = DestHealth::default();
-        h.record(false, 1, Duration::from_secs(0));
+        h.record_failure(Attribution::Host, &limits(1, 2, Duration::from_secs(0)));
         assert!(h.gate().is_ok());
     }
 
+    #[test]
+    fn attribution_follows_the_failure_stage() {
+        let ip: IpAddr = "203.0.113.9".parse().unwrap();
+        assert_eq!(
+            Attribution::of(&ProxyError::backend_connect("refused"), ip),
+            Attribution::Host
+        );
+        assert_eq!(
+            Attribution::of(&ProxyError::backend("closed after PROXY header"), ip),
+            Attribution::Client(ip)
+        );
+        assert_eq!(
+            Attribution::of(
+                &ProxyError::backend_session("backend closed before sending a complete line"),
+                ip
+            ),
+            Attribution::Client(ip)
+        );
+        assert_eq!(
+            Attribution::of(
+                &ProxyError::backend_session("backend TLS handshake failed"),
+                ip
+            ),
+            Attribution::Client(ip)
+        );
+        assert_eq!(
+            Attribution::of(
+                &ProxyError::Io(std::io::Error::from(std::io::ErrorKind::ConnectionReset)),
+                ip
+            ),
+            Attribution::Client(ip)
+        );
+    }
+
+    #[test]
+    fn banned_client_alone_never_opens_the_breaker() {
+        let h = DestHealth::default();
+        let lim = limits(3, 2, Duration::from_secs(60));
+        for _ in 0..50 {
+            h.record_failure(client("198.51.100.7"), &lim);
+        }
+        assert!(h.gate().is_ok());
+        let s = h.snapshot();
+        assert_eq!(s.consecutive_failures, 50);
+        assert_eq!(s.failure_sources, 1);
+        assert!(!s.down);
+    }
+
+    #[test]
+    fn failures_spanning_distinct_clients_open_the_breaker() {
+        let h = DestHealth::default();
+        let lim = limits(3, 2, Duration::from_secs(60));
+        h.record_failure(client("198.51.100.7"), &lim);
+        h.record_failure(client("198.51.100.7"), &lim);
+        h.record_failure(client("198.51.100.7"), &lim);
+        assert!(h.gate().is_ok());
+
+        h.record_failure(client("203.0.113.4"), &lim);
+        assert!(h.gate().is_err());
+        assert_eq!(h.snapshot().failure_sources, 2);
+    }
+
+    #[test]
+    fn host_level_failure_opens_the_breaker_without_breadth() {
+        let h = DestHealth::default();
+        let lim = limits(3, 2, Duration::from_secs(60));
+        h.record_failure(client("198.51.100.7"), &lim);
+        h.record_failure(client("198.51.100.7"), &lim);
+        h.record_failure(Attribution::Host, &lim);
+        assert!(h.gate().is_err());
+        let s = h.snapshot();
+        assert!(s.host_level_failure);
+        assert!(s.down);
+    }
+
+    #[test]
+    fn cooldown_expiry_clears_stale_evidence() {
+        let h = DestHealth::default();
+        let lim = limits(3, 2, Duration::from_secs(0));
+        for _ in 0..3 {
+            h.record_failure(Attribution::Host, &lim);
+        }
+        assert!(h.gate().is_ok());
+        let s = h.snapshot();
+        assert!(!s.host_level_failure);
+        assert_eq!(s.failure_sources, 0);
+
+        h.record_failure(
+            client("198.51.100.7"),
+            &limits(3, 2, Duration::from_secs(60)),
+        );
+        assert!(h.gate().is_ok());
+    }
+
+    #[test]
+    fn cooldown_expiry_still_re_opens_on_a_failed_host_probe() {
+        let h = DestHealth::default();
+        let lim = limits(3, 2, Duration::from_secs(0));
+        for _ in 0..3 {
+            h.record_failure(Attribution::Host, &lim);
+        }
+        assert!(h.gate().is_ok());
+
+        h.record_failure(Attribution::Host, &limits(3, 2, Duration::from_secs(60)));
+        assert!(h.gate().is_err());
+    }
+
+    #[test]
+    fn success_clears_the_failure_streak() {
+        let h = DestHealth::default();
+        let lim = limits(3, 2, Duration::from_secs(60));
+        h.record_failure(client("198.51.100.7"), &lim);
+        h.record_failure(client("203.0.113.4"), &lim);
+        h.record_success();
+        let s = h.snapshot();
+        assert_eq!(s.consecutive_failures, 0);
+        assert_eq!(s.failure_sources, 0);
+
+        h.record_failure(client("198.51.100.7"), &lim);
+        h.record_failure(client("198.51.100.7"), &lim);
+        h.record_failure(client("198.51.100.7"), &lim);
+        assert!(h.gate().is_ok());
+    }
+
+    #[test]
+    fn min_clients_zero_counts_failures_alone() {
+        let h = DestHealth::default();
+        let lim = limits(3, 0, Duration::from_secs(60));
+        for _ in 0..3 {
+            h.record_failure(client("198.51.100.7"), &lim);
+        }
+        assert!(h.gate().is_err());
+    }
+
+    #[test]
+    fn tracked_failure_sources_are_bounded() {
+        let h = DestHealth::default();
+        let lim = limits(0, 2, Duration::from_secs(60));
+        for i in 0..(MAX_TRACKED_FAILURE_SOURCES + 100) {
+            let ip = IpAddr::from([10, (i / 256) as u8, (i % 256) as u8, 1]);
+            h.record_failure(Attribution::Client(ip), &lim);
+        }
+        assert_eq!(
+            h.snapshot().failure_sources as usize,
+            MAX_TRACKED_FAILURE_SOURCES
+        );
+    }
+
     #[tokio::test]
     async fn conn_registry_kick_signals_live_sessions() {
         let reg = ConnRegistry::default();
@@ -912,7 +1146,7 @@ mod tests {
     fn dest_health_threshold_zero_never_trips() {
         let h = DestHealth::default();
         for _ in 0..100 {
-            h.record(false, 0, Duration::from_secs(60));
+            h.record_failure(Attribution::Host, &limits(0, 2, Duration::from_secs(60)));
         }
         assert!(h.gate().is_ok());
     }

src/proto/imap.rs

@@ -89,6 +89,7 @@ pub async fn handle(
         ctx,
         &dest_id,
         stream,
+        peer,
         identifier.as_deref(),
         &residual,
         started,

src/proto/managesieve.rs

@@ -76,6 +76,7 @@ pub async fn handle(
         ctx,
         &dest_id,
         stream,
+        peer,
         identifier.as_deref(),
         &residual,
         started,

src/proto/pop3.rs

@@ -75,6 +75,7 @@ pub async fn handle(
         ctx,
         &dest_id,
         stream,
+        peer,
         identifier.as_deref(),
         &residual,
         started,

src/proto/smtp.rs

@@ -82,6 +82,7 @@ pub async fn handle_submission(
         ctx,
         &dest_id,
         stream,
+        peer,
         identifier.as_deref(),
         &residual,
         started,
@@ -572,7 +573,7 @@ pub async fn handle_passthrough(
     let (dest, ep, tls_cfg) = ctx.endpoint(&dest_id, proto)?;
     let fwd = dest.forwarding_for(proto);
     forward::guard_loop(&ctx.self_binds, dest, ep, fwd, ctx.config.server.proxy_ttl)?;
-    let backend = common::establish(ctx, &dest_id, || {
+    let backend = common::establish(ctx, &dest_id, peer, || {
         outbound::dial(dest, ep, fwd, &tls_cfg, peer, local)
     })
     .await?;

src/tests/admin.rs

@@ -473,6 +473,10 @@ async fn admin_mapping_crud_and_reads() {
         body.contains("legacy") && body.contains("secondary"),
         "body: {body}"
     );
+    assert!(
+        body.contains("failure_sources") && body.contains("host_level_failure"),
+        "body: {body}"
+    );
 
     let (status, _) = admin_request(
         addr,

src/tests/containers/legacy.rs

@@ -7,10 +7,12 @@
 use std::time::Duration;
 
 use testcontainers::core::{IntoContainerPort, WaitFor};
-use testcontainers::runners::{AsyncBuilder, AsyncRunner};
-use testcontainers::{ContainerAsync, GenericBuildableImage, GenericImage, ImageExt};
+use testcontainers::runners::AsyncRunner;
+use testcontainers::{ContainerAsync, GenericImage, ImageExt};
 use tokio::net::TcpStream;
 
+use super::{ensure_image, owner_labels};
+
 pub const MASTER_USER: &str = "admin";
 pub const MASTER_PASSWORD: &str = "MasterPass#2026";
 
@@ -218,15 +220,17 @@ pub struct Legacy {
 
 impl Legacy {
     pub async fn start() -> Legacy {
-        let image: GenericImage = GenericBuildableImage::new("proxy-legacymail", "test")
-            .with_dockerfile_string(DOCKERFILE.to_string())
-            .with_data(DOVECOT_CONF.as_bytes().to_vec(), "dovecot.conf")
-            .with_data(USERS.as_bytes().to_vec(), "users")
-            .with_data(MASTER_USERS.as_bytes().to_vec(), "master-users")
-            .with_data(ENTRYPOINT.as_bytes().to_vec(), "entrypoint.sh")
-            .build_image()
-            .await
-            .expect("build legacy image");
+        let image = ensure_image(
+            "proxy-legacymail",
+            DOCKERFILE,
+            &[
+                ("dovecot.conf", DOVECOT_CONF.as_bytes()),
+                ("users", USERS.as_bytes()),
+                ("master-users", MASTER_USERS.as_bytes()),
+                ("entrypoint.sh", ENTRYPOINT.as_bytes()),
+            ],
+        )
+        .await;
 
         let container = image
             .with_exposed_port(143u16.tcp())
@@ -239,6 +243,7 @@ impl Legacy {
             .with_exposed_port(25u16.tcp())
             .with_wait_for(WaitFor::message_on_stdout("LEGACY-MAIL READY"))
             .with_startup_timeout(Duration::from_secs(240))
+            .with_labels(owner_labels())
             .start()
             .await
             .expect("start legacy container");

src/tests/containers/mod.rs

@@ -5,12 +5,16 @@
  */
 
 use std::collections::HashMap;
+use std::hash::{BuildHasher, Hash, Hasher};
 use std::net::{SocketAddr, TcpListener as StdTcpListener};
-use std::sync::Arc;
-use std::time::Duration;
+use std::process::{Command, Stdio};
+use std::sync::{Arc, Once, OnceLock};
+use std::time::{Duration, SystemTime, UNIX_EPOCH};
 
+use testcontainers::runners::AsyncBuilder;
+use testcontainers::{GenericBuildableImage, GenericImage};
 use tokio::net::TcpStream;
-use tokio::sync::watch;
+use tokio::sync::{Mutex, watch};
 use tokio::task::JoinHandle;
 
 use crate::net::acceptor::Listener;
@@ -34,11 +38,100 @@ pub mod toolbox_matrix;
 
 pub const HOST_GATEWAY: &str = "host.docker.internal";
 
+pub const OWNER_LABEL: &str = "art.stalw.proxy.test";
+
 pub fn pick_free_port() -> u16 {
     let l = StdTcpListener::bind("127.0.0.1:0").unwrap();
     l.local_addr().unwrap().port()
 }
 
+fn run_id() -> &'static str {
+    static RUN_ID: OnceLock<String> = OnceLock::new();
+    RUN_ID.get_or_init(|| {
+        let nanos = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .unwrap_or_default()
+            .as_nanos();
+        format!("{}-{nanos}", std::process::id())
+    })
+}
+
+pub fn owner_labels() -> [(&'static str, &'static str); 1] {
+    sweep_abandoned_containers();
+    [(OWNER_LABEL, run_id())]
+}
+
+fn docker_ids(filter: &str) -> Vec<String> {
+    Command::new("docker")
+        .args(["ps", "-aq", "--filter", filter])
+        .output()
+        .map(|out| {
+            String::from_utf8_lossy(&out.stdout)
+                .split_whitespace()
+                .map(String::from)
+                .collect()
+        })
+        .unwrap_or_default()
+}
+
+fn sweep_abandoned_containers() {
+    static SWEPT: Once = Once::new();
+    SWEPT.call_once(|| {
+        let mine = docker_ids(&format!("label={OWNER_LABEL}={}", run_id()));
+        let stale: Vec<String> = docker_ids(&format!("label={OWNER_LABEL}"))
+            .into_iter()
+            .filter(|id| !mine.contains(id))
+            .collect();
+        if stale.is_empty() {
+            return;
+        }
+        eprintln!("removing {} abandoned test container(s)", stale.len());
+        let _ = Command::new("docker")
+            .args(["rm", "-f", "-v"])
+            .args(&stale)
+            .output();
+    });
+}
+
+fn content_tag(dockerfile: &str, data: &[(&str, &[u8])]) -> String {
+    let mut hasher = ahash::RandomState::with_seeds(0x5354, 0x414c, 0x5741, 0x5254).build_hasher();
+    dockerfile.hash(&mut hasher);
+    for (target, bytes) in data {
+        target.hash(&mut hasher);
+        bytes.hash(&mut hasher);
+    }
+    format!("{:016x}", hasher.finish())
+}
+
+async fn image_present(reference: &str) -> bool {
+    tokio::process::Command::new("docker")
+        .args(["image", "inspect", reference])
+        .stdout(Stdio::null())
+        .stderr(Stdio::null())
+        .status()
+        .await
+        .map(|status| status.success())
+        .unwrap_or(false)
+}
+
+pub async fn ensure_image(name: &str, dockerfile: &str, data: &[(&str, &[u8])]) -> GenericImage {
+    static BUILD_LOCK: Mutex<()> = Mutex::const_new(());
+    let tag = content_tag(dockerfile, data);
+    let _guard = BUILD_LOCK.lock().await;
+    if image_present(&format!("{name}:{tag}")).await {
+        return GenericImage::new(name.to_owned(), tag);
+    }
+    let mut builder =
+        GenericBuildableImage::new(name, &tag).with_dockerfile_string(dockerfile.to_string());
+    for (target, bytes) in data {
+        builder = builder.with_data(bytes.to_vec(), *target);
+    }
+    builder
+        .build_image()
+        .await
+        .unwrap_or_else(|e| panic!("build image {name}:{tag}: {e}"))
+}
+
 pub struct Proxy {
     shutdown: watch::Sender<bool>,
     handles: Vec<JoinHandle<()>>,

src/tests/containers/stalwart.rs

@@ -12,7 +12,7 @@ use testcontainers::runners::AsyncRunner;
 use testcontainers::{ContainerAsync, GenericImage, ImageExt};
 
 use super::http::{self, basic_auth};
-use super::pick_free_port;
+use super::{owner_labels, pick_free_port};
 
 pub const NEW_IMAGE: &str = "stalwartlabs/stalwart";
 pub const NEW_TAG: &str = "latest";
@@ -114,6 +114,7 @@ impl Stalwart {
             .with_mapped_port(ports.sieve, 4190u16.tcp())
             .with_mapped_port(ports.smtp, 25u16.tcp())
             .with_startup_timeout(Duration::from_secs(180))
+            .with_labels(owner_labels())
             .start()
             .await
             .expect("start new stalwart");
@@ -196,6 +197,7 @@ impl Stalwart {
             .with_mapped_port(ports.sieve, 4190u16.tcp())
             .with_mapped_port(ports.smtp, 25u16.tcp())
             .with_startup_timeout(Duration::from_secs(180))
+            .with_labels(owner_labels())
             .start()
             .await
             .expect("start old stalwart");

src/tests/containers/toolbox.rs

@@ -7,11 +7,11 @@
 use std::time::Duration;
 
 use testcontainers::core::{ExecCommand, Host, WaitFor};
-use testcontainers::runners::{AsyncBuilder, AsyncRunner};
-use testcontainers::{ContainerAsync, GenericBuildableImage, GenericImage, ImageExt};
+use testcontainers::runners::AsyncRunner;
+use testcontainers::{ContainerAsync, GenericImage, ImageExt};
 use tokio::io::AsyncReadExt;
 
-use super::HOST_GATEWAY;
+use super::{HOST_GATEWAY, ensure_image, owner_labels};
 
 const DOCKERFILE: &str = r#"FROM debian:trixie-slim
 ENV DEBIAN_FRONTEND=noninteractive
@@ -54,16 +54,13 @@ impl Exec {
 
 impl Toolbox {
     pub async fn start() -> Toolbox {
-        let image: GenericImage = GenericBuildableImage::new("proxy-toolbox", "test")
-            .with_dockerfile_string(DOCKERFILE.to_string())
-            .build_image()
-            .await
-            .expect("build toolbox image");
+        let image = ensure_image("proxy-toolbox", DOCKERFILE, &[]).await;
 
         let container = image
             .with_wait_for(WaitFor::seconds(1))
             .with_startup_timeout(Duration::from_secs(300))
             .with_host(HOST_GATEWAY, Host::HostGateway)
+            .with_labels(owner_labels())
             .start()
             .await
             .expect("start toolbox container");

src/tests/dest_health.rs

@@ -0,0 +1,274 @@
+/*
+ * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <hello@stalw.art>
+ *
+ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
+ */
+
+use std::net::SocketAddr;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
+
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::net::{TcpListener, TcpStream};
+use tokio::task::JoinHandle;
+
+use super::harness::*;
+use crate::proto::common::Ctx;
+
+fn config(port: u16, mapping_path: &str, threshold: u32, min_clients: u32) -> String {
+    format!(
+        r#"
+[server]
+host_down_threshold = {threshold}
+host_down_min_clients = {min_clients}
+host_down_cooldown = "60s"
+backend_connect_retries = 0
+
+[mapping]
+source = "file"
+normalize = "lowercase"
+[mapping.file]
+path = "{mapping_path}"
+
+[capabilities.imap]
+allow_plain_auth_without_tls = true
+
+[routing]
+default_destination = "legacy"
+
+[destination.legacy]
+host = "127.0.0.1"
+proxy_protocol = false
+allow_plaintext_auth = true
+[destination.legacy.protocol.imap]
+port = {port}
+tls = "plain"
+
+[listener.imap]
+protocol = "imap"
+bind = ["127.0.0.1:0"]
+tls = "plain"
+"#
+    )
+}
+
+struct Backend {
+    port: u16,
+    serving: Arc<AtomicBool>,
+    accepted: Arc<AtomicU32>,
+    handle: JoinHandle<()>,
+}
+
+async fn banning_backend() -> Backend {
+    let listener = TcpListener::bind(unused_addr()).await.unwrap();
+    let port = listener.local_addr().unwrap().port();
+    let serving = Arc::new(AtomicBool::new(false));
+    let accepted = Arc::new(AtomicU32::new(0));
+    let serving_task = serving.clone();
+    let accepted_task = accepted.clone();
+    let handle = tokio::spawn(async move {
+        loop {
+            let Ok((socket, _)) = listener.accept().await else {
+                return;
+            };
+            let _ = socket.set_nodelay(true);
+            accepted_task.fetch_add(1, Ordering::Relaxed);
+            if serving_task.load(Ordering::Relaxed) {
+                tokio::spawn(serve_imap_auth(socket));
+            } else {
+                drop(socket);
+            }
+        }
+    });
+    Backend {
+        port,
+        serving,
+        accepted,
+        handle,
+    }
+}
+
+async fn serve_imap_auth(mut socket: TcpStream) {
+    let _ = socket
+        .write_all(b"* OK [CAPABILITY IMAP4rev2] backend ready\r\n")
+        .await;
+
+    let mut buf = Vec::new();
+    let mut scratch = [0u8; 1024];
+    let tag = loop {
+        let Ok(n) = socket.read(&mut scratch).await else {
+            return;
+        };
+        if n == 0 {
+            return;
+        }
+        buf.extend_from_slice(&scratch[..n]);
+        if let Some(end) = find(&buf, b"\r\n") {
+            let line = String::from_utf8_lossy(&buf[..end]).into_owned();
+            break line.split(' ').next().unwrap_or("a1").to_string();
+        }
+    };
+    let _ = socket
+        .write_all(format!("{tag} OK [CAPABILITY IMAP4rev2] authenticated\r\n").as_bytes())
+        .await;
+
+    loop {
+        if find(&buf, b"NOOP").is_some() {
+            break;
+        }
+        let Ok(n) = socket.read(&mut scratch).await else {
+            return;
+        };
+        if n == 0 {
+            return;
+        }
+        buf.extend_from_slice(&scratch[..n]);
+    }
+    let _ = socket.write_all(b"a2 OK NOOP done\r\n").await;
+}
+
+async fn imap_login(ctx: &Arc<Ctx>, peer: &str, user: &str) -> Vec<u8> {
+    let listener = ctx.config.listener["imap"].clone();
+    let peer: SocketAddr = peer.parse().unwrap();
+    let (mut client, server) = tcp_pair().await;
+    let ctx_task = ctx.clone();
+    let handler = tokio::spawn(async move {
+        let _ = crate::proto::imap::handle(
+            &ctx_task,
+            &listener,
+            Box::new(server),
+            peer,
+            "127.0.0.1:143".parse().unwrap(),
+        )
+        .await;
+    });
+
+    let _greeting = read_line(&mut client).await;
+    send(
+        &mut client,
+        format!("a1 LOGIN {user} secret\r\na2 NOOP\r\n").as_bytes(),
+    )
+    .await;
+    let resp = read_to_eof(&mut client).await;
+    handler.abort();
+    resp
+}
+
+#[tokio::test]
+async fn one_banned_client_does_not_mark_the_destination_down() {
+    let backend = banning_backend().await;
+    let mapping = mapping_file("");
+    let toml = config(backend.port, &mapping.path().display().to_string(), 3, 2);
+    let ctx = make_ctx(&toml).await;
+
+    for i in 0..5 {
+        let resp = imap_login(&ctx, "198.51.100.7:40000", "banned@example.com").await;
+        assert!(
+            find(&resp, b"a1 NO [UNAVAILABLE]").is_some(),
+            "attempt {i} should report the backend unavailable, got {:?}",
+            String::from_utf8_lossy(&resp)
+        );
+    }
+
+    let health = ctx.dest_health.get("legacy").unwrap().snapshot();
+    assert_eq!(health.consecutive_failures, 5);
+    assert_eq!(health.failure_sources, 1);
+    assert!(!health.host_level_failure);
+    assert!(
+        !health.down,
+        "a single client must not open the circuit breaker"
+    );
+
+    backend.serving.store(true, Ordering::Relaxed);
+    let resp = imap_login(&ctx, "203.0.113.4:40000", "alice@example.com").await;
+    assert!(
+        find(&resp, b"a1 OK").is_some(),
+        "an unrelated account must still reach the backend, got {:?}",
+        String::from_utf8_lossy(&resp)
+    );
+    assert!(!ctx.dest_health.get("legacy").unwrap().snapshot().down);
+    assert!(backend.accepted.load(Ordering::Relaxed) >= 6);
+
+    backend.handle.abort();
+}
+
+#[tokio::test]
+async fn rejections_spanning_distinct_clients_mark_the_destination_down() {
+    let backend = banning_backend().await;
+    let mapping = mapping_file("");
+    let toml = config(backend.port, &mapping.path().display().to_string(), 3, 2);
+    let ctx = make_ctx(&toml).await;
+
+    for peer in [
+        "198.51.100.7:40000",
+        "198.51.100.7:40001",
+        "203.0.113.4:40000",
+    ] {
+        let resp = imap_login(&ctx, peer, "user@example.com").await;
+        assert!(find(&resp, b"a1 NO [UNAVAILABLE]").is_some());
+    }
+
+    let health = ctx.dest_health.get("legacy").unwrap().snapshot();
+    assert_eq!(health.failure_sources, 2);
+    assert!(
+        health.down,
+        "failures from distinct clients are evidence the host is unhealthy"
+    );
+
+    let accepted_before = backend.accepted.load(Ordering::Relaxed);
+    backend.serving.store(true, Ordering::Relaxed);
+    let resp = imap_login(&ctx, "192.0.2.9:40000", "alice@example.com").await;
+    assert!(find(&resp, b"a1 NO [UNAVAILABLE]").is_some());
+    assert_eq!(
+        backend.accepted.load(Ordering::Relaxed),
+        accepted_before,
+        "an open breaker must short-circuit before dialing the backend"
+    );
+
+    backend.handle.abort();
+}
+
+#[tokio::test]
+async fn min_clients_zero_restores_counting_by_failures_alone() {
+    let backend = banning_backend().await;
+    let mapping = mapping_file("");
+    let toml = config(backend.port, &mapping.path().display().to_string(), 3, 0);
+    let ctx = make_ctx(&toml).await;
+
+    for _ in 0..3 {
+        let resp = imap_login(&ctx, "198.51.100.7:40000", "banned@example.com").await;
+        assert!(find(&resp, b"a1 NO [UNAVAILABLE]").is_some());
+    }
+
+    let health = ctx.dest_health.get("legacy").unwrap().snapshot();
+    assert_eq!(health.failure_sources, 1);
+    assert!(
+        health.down,
+        "without the breadth requirement a single client still opens the breaker"
+    );
+
+    backend.handle.abort();
+}
+
+#[tokio::test]
+async fn unreachable_host_marks_the_destination_down_from_one_client() {
+    let closed_port = {
+        let l = TcpListener::bind(unused_addr()).await.unwrap();
+        l.local_addr().unwrap().port()
+    };
+    let mapping = mapping_file("");
+    let toml = config(closed_port, &mapping.path().display().to_string(), 3, 2);
+    let ctx = make_ctx(&toml).await;
+
+    for _ in 0..3 {
+        let resp = imap_login(&ctx, "198.51.100.7:40000", "user@example.com").await;
+        assert!(find(&resp, b"a1 NO [UNAVAILABLE]").is_some());
+    }
+
+    let health = ctx.dest_health.get("legacy").unwrap().snapshot();
+    assert!(health.host_level_failure);
+    assert!(
+        health.down,
+        "connect-stage failures are host evidence regardless of how many clients saw them"
+    );
+}

src/tests/mod.rs

@@ -7,6 +7,7 @@
 mod admin;
 mod config_validation;
 mod containers;
+mod dest_health;
 mod forwarding;
 mod harness;
 mod http_proxy;