Integration test merge: upstream/main into flow-sec security branch (prefer flow-sec on conflicts)

src/transport/middle_proxy/config_updater.rs

@@ -299,6 +299,11 @@ async fn run_update_cycle(
         cfg.general.hardswap,
         cfg.general.me_pool_drain_ttl_secs,
         cfg.general.me_pool_drain_threshold,
+        cfg.general.me_pool_drain_soft_evict_enabled,
+        cfg.general.me_pool_drain_soft_evict_grace_secs,
+        cfg.general.me_pool_drain_soft_evict_per_writer,
+        cfg.general.me_pool_drain_soft_evict_budget_per_core,
+        cfg.general.me_pool_drain_soft_evict_cooldown_ms,
         cfg.general.effective_me_pool_force_close_secs(),
         cfg.general.me_pool_min_fresh_ratio,
         cfg.general.me_hardswap_warmup_delay_min_ms,
@@ -526,6 +531,11 @@ pub async fn me_config_updater(
                     cfg.general.hardswap,
                     cfg.general.me_pool_drain_ttl_secs,
                     cfg.general.me_pool_drain_threshold,
+                    cfg.general.me_pool_drain_soft_evict_enabled,
+                    cfg.general.me_pool_drain_soft_evict_grace_secs,
+                    cfg.general.me_pool_drain_soft_evict_per_writer,
+                    cfg.general.me_pool_drain_soft_evict_budget_per_core,
+                    cfg.general.me_pool_drain_soft_evict_cooldown_ms,
                     cfg.general.effective_me_pool_force_close_secs(),
                     cfg.general.me_pool_min_fresh_ratio,
                     cfg.general.me_hardswap_warmup_delay_min_ms,

src/transport/middle_proxy/health_adversarial_tests.rs

@@ -83,6 +83,11 @@ async fn make_pool(
         general.hardswap,
         general.me_pool_drain_ttl_secs,
         general.me_pool_drain_threshold,
+        general.me_pool_drain_soft_evict_enabled,
+        general.me_pool_drain_soft_evict_grace_secs,
+        general.me_pool_drain_soft_evict_per_writer,
+        general.me_pool_drain_soft_evict_budget_per_core,
+        general.me_pool_drain_soft_evict_cooldown_ms,
         general.effective_me_pool_force_close_secs(),
         general.me_pool_min_fresh_ratio,
         general.me_hardswap_warmup_delay_min_ms,
@@ -107,6 +112,8 @@ async fn make_pool(
         general.me_warn_rate_limit_ms,
         MeRouteNoWriterMode::default(),
         general.me_route_no_writer_wait_ms,
+        general.me_route_hybrid_max_wait_ms,
+        general.me_route_blocking_send_timeout_ms,
         general.me_route_inline_recovery_attempts,
         general.me_route_inline_recovery_wait_ms,
     );
@@ -220,10 +227,11 @@ async fn set_writer_runtime_state(
 async fn reap_draining_writers_clears_warn_state_when_pool_empty() {
     let (pool, _rng) = make_pool(128, 1, 1).await;
     let mut warn_next_allowed = HashMap::new();
+    let mut soft_evict_next_allowed = HashMap::new();
     warn_next_allowed.insert(11, Instant::now() + Duration::from_secs(5));
     warn_next_allowed.insert(22, Instant::now() + Duration::from_secs(5));
 
-    reap_draining_writers(&pool, &mut warn_next_allowed).await;
+    reap_draining_writers(&pool, &mut warn_next_allowed, &mut soft_evict_next_allowed).await;
 
     assert!(warn_next_allowed.is_empty());
 }
@@ -232,6 +240,8 @@ async fn reap_draining_writers_clears_warn_state_when_pool_empty() {
 async fn reap_draining_writers_respects_threshold_across_multiple_overflow_cycles() {
     let threshold = 3u64;
     let (pool, _rng) = make_pool(threshold, 1, 1).await;
+    pool.me_pool_drain_soft_evict_enabled
+        .store(false, Ordering::Relaxed);
     let now_epoch_secs = MePool::now_epoch_secs();
 
     for writer_id in 1..=60u64 {
@@ -246,8 +256,9 @@ async fn reap_draining_writers_respects_threshold_across_multiple_overflow_cycle
     }
 
     let mut warn_next_allowed = HashMap::new();
+    let mut soft_evict_next_allowed = HashMap::new();
     for _ in 0..64 {
-        reap_draining_writers(&pool, &mut warn_next_allowed).await;
+        reap_draining_writers(&pool, &mut warn_next_allowed, &mut soft_evict_next_allowed).await;
         if writer_count(&pool).await <= threshold as usize {
             break;
         }
@@ -275,11 +286,12 @@ async fn reap_draining_writers_handles_large_empty_writer_population() {
     }
 
     let mut warn_next_allowed = HashMap::new();
+    let mut soft_evict_next_allowed = HashMap::new();
     for _ in 0..24 {
         if writer_count(&pool).await == 0 {
             break;
         }
-        reap_draining_writers(&pool, &mut warn_next_allowed).await;
+        reap_draining_writers(&pool, &mut warn_next_allowed, &mut soft_evict_next_allowed).await;
     }
 
     assert_eq!(writer_count(&pool).await, 0);
@@ -303,11 +315,12 @@ async fn reap_draining_writers_processes_mass_deadline_expiry_without_unbounded_
     }
 
     let mut warn_next_allowed = HashMap::new();
+    let mut soft_evict_next_allowed = HashMap::new();
     for _ in 0..40 {
         if writer_count(&pool).await == 0 {
             break;
         }
-        reap_draining_writers(&pool, &mut warn_next_allowed).await;
+        reap_draining_writers(&pool, &mut warn_next_allowed, &mut soft_evict_next_allowed).await;
     }
 
     assert_eq!(writer_count(&pool).await, 0);
@@ -318,6 +331,7 @@ async fn reap_draining_writers_maintains_warn_state_subset_property_under_bulk_c
     let (pool, _rng) = make_pool(128, 1, 1).await;
     let now_epoch_secs = MePool::now_epoch_secs();
     let mut warn_next_allowed = HashMap::new();
+    let mut soft_evict_next_allowed = HashMap::new();
 
     for wave in 0..40u64 {
         for offset in 0..8u64 {
@@ -331,7 +345,7 @@ async fn reap_draining_writers_maintains_warn_state_subset_property_under_bulk_c
             .await;
         }
 
-        reap_draining_writers(&pool, &mut warn_next_allowed).await;
+        reap_draining_writers(&pool, &mut warn_next_allowed, &mut soft_evict_next_allowed).await;
         assert!(warn_next_allowed.len() <= writer_count(&pool).await);
 
         let ids = sorted_writer_ids(&pool).await;
@@ -339,7 +353,7 @@ async fn reap_draining_writers_maintains_warn_state_subset_property_under_bulk_c
             let _ = pool.remove_writer_and_close_clients(writer_id).await;
         }
 
-        reap_draining_writers(&pool, &mut warn_next_allowed).await;
+        reap_draining_writers(&pool, &mut warn_next_allowed, &mut soft_evict_next_allowed).await;
         assert!(warn_next_allowed.len() <= writer_count(&pool).await);
     }
 }
@@ -361,9 +375,10 @@ async fn reap_draining_writers_budgeted_cleanup_never_increases_pool_size() {
     }
 
     let mut warn_next_allowed = HashMap::new();
+    let mut soft_evict_next_allowed = HashMap::new();
     let mut previous = writer_count(&pool).await;
     for _ in 0..32 {
-        reap_draining_writers(&pool, &mut warn_next_allowed).await;
+        reap_draining_writers(&pool, &mut warn_next_allowed, &mut soft_evict_next_allowed).await;
         let current = writer_count(&pool).await;
         assert!(current <= previous);
         previous = current;

src/transport/middle_proxy/health_integration_tests.rs

@@ -81,6 +81,11 @@ async fn make_pool(
         general.hardswap,
         general.me_pool_drain_ttl_secs,
         general.me_pool_drain_threshold,
+        general.me_pool_drain_soft_evict_enabled,
+        general.me_pool_drain_soft_evict_grace_secs,
+        general.me_pool_drain_soft_evict_per_writer,
+        general.me_pool_drain_soft_evict_budget_per_core,
+        general.me_pool_drain_soft_evict_cooldown_ms,
         general.effective_me_pool_force_close_secs(),
         general.me_pool_min_fresh_ratio,
         general.me_hardswap_warmup_delay_min_ms,
@@ -105,6 +110,8 @@ async fn make_pool(
         general.me_warn_rate_limit_ms,
         MeRouteNoWriterMode::default(),
         general.me_route_no_writer_wait_ms,
+        general.me_route_hybrid_max_wait_ms,
+        general.me_route_blocking_send_timeout_ms,
         general.me_route_inline_recovery_attempts,
         general.me_route_inline_recovery_wait_ms,
     );

src/transport/middle_proxy/health_regression_tests.rs

@@ -4,6 +4,7 @@ use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, AtomicU8, AtomicU32, AtomicU64, Ordering};
 use std::time::{Duration, Instant};
 
+use bytes::Bytes;
 use tokio::sync::mpsc;
 use tokio_util::sync::CancellationToken;
 
@@ -39,7 +40,7 @@ async fn make_pool(me_pool_drain_threshold: u64) -> Arc<MePool> {
         NetworkDecision::default(),
         None,
         Arc::new(SecureRandom::new()),
-        Arc::new(Stats::default()),
+        Arc::new(Stats::new()),
         general.me_keepalive_enabled,
         general.me_keepalive_interval_secs,
         general.me_keepalive_jitter_secs,
@@ -74,6 +75,11 @@ async fn make_pool(me_pool_drain_threshold: u64) -> Arc<MePool> {
         general.hardswap,
         general.me_pool_drain_ttl_secs,
         general.me_pool_drain_threshold,
+        general.me_pool_drain_soft_evict_enabled,
+        general.me_pool_drain_soft_evict_grace_secs,
+        general.me_pool_drain_soft_evict_per_writer,
+        general.me_pool_drain_soft_evict_budget_per_core,
+        general.me_pool_drain_soft_evict_cooldown_ms,
         general.effective_me_pool_force_close_secs(),
         general.me_pool_min_fresh_ratio,
         general.me_hardswap_warmup_delay_min_ms,
@@ -98,6 +104,8 @@ async fn make_pool(me_pool_drain_threshold: u64) -> Arc<MePool> {
         general.me_warn_rate_limit_ms,
         MeRouteNoWriterMode::default(),
         general.me_route_no_writer_wait_ms,
+        general.me_route_hybrid_max_wait_ms,
+        general.me_route_blocking_send_timeout_ms,
         general.me_route_inline_recovery_attempts,
         general.me_route_inline_recovery_wait_ms,
     )
@@ -190,14 +198,15 @@ async fn reap_draining_writers_drops_warn_state_for_removed_writer() {
     let conn_ids =
         insert_draining_writer(&pool, 7, now_epoch_secs.saturating_sub(180), 1, 0).await;
     let mut warn_next_allowed = HashMap::new();
+    let mut soft_evict_next_allowed = HashMap::new();
 
-    reap_draining_writers(&pool, &mut warn_next_allowed).await;
+    reap_draining_writers(&pool, &mut warn_next_allowed, &mut soft_evict_next_allowed).await;
     assert!(warn_next_allowed.contains_key(&7));
 
     let _ = pool.remove_writer_and_close_clients(7).await;
     assert!(pool.registry.get_writer(conn_ids[0]).await.is_none());
 
-    reap_draining_writers(&pool, &mut warn_next_allowed).await;
+    reap_draining_writers(&pool, &mut warn_next_allowed, &mut soft_evict_next_allowed).await;
     assert!(!warn_next_allowed.contains_key(&7));
 }
 
@@ -209,12 +218,96 @@ async fn reap_draining_writers_removes_empty_draining_writers() {
     insert_draining_writer(&pool, 2, now_epoch_secs.saturating_sub(30), 0, 0).await;
     insert_draining_writer(&pool, 3, now_epoch_secs.saturating_sub(20), 1, 0).await;
     let mut warn_next_allowed = HashMap::new();
+    let mut soft_evict_next_allowed = HashMap::new();
 
-    reap_draining_writers(&pool, &mut warn_next_allowed).await;
+    reap_draining_writers(&pool, &mut warn_next_allowed, &mut soft_evict_next_allowed).await;
 
     assert_eq!(current_writer_ids(&pool).await, vec![3]);
 }
 
+#[tokio::test]
+async fn reap_draining_writers_does_not_block_on_stuck_writer_close_signal() {
+    let pool = make_pool(128).await;
+    let now_epoch_secs = MePool::now_epoch_secs();
+
+    let (blocked_tx, blocked_rx) = mpsc::channel::<WriterCommand>(1);
+    assert!(
+        blocked_tx
+            .try_send(WriterCommand::Data(Bytes::from_static(b"stuck")))
+            .is_ok()
+    );
+    let blocked_rx_guard = tokio::spawn(async move {
+        let _hold_rx = blocked_rx;
+        tokio::time::sleep(Duration::from_secs(30)).await;
+    });
+
+    let blocked_writer_id = 90u64;
+    let blocked_writer = MeWriter {
+        id: blocked_writer_id,
+        addr: SocketAddr::new(
+            IpAddr::V4(Ipv4Addr::LOCALHOST),
+            4500 + blocked_writer_id as u16,
+        ),
+        source_ip: IpAddr::V4(Ipv4Addr::LOCALHOST),
+        writer_dc: 2,
+        generation: 1,
+        contour: Arc::new(AtomicU8::new(WriterContour::Draining.as_u8())),
+        created_at: Instant::now() - Duration::from_secs(blocked_writer_id),
+        tx: blocked_tx.clone(),
+        cancel: CancellationToken::new(),
+        degraded: Arc::new(AtomicBool::new(false)),
+        rtt_ema_ms_x10: Arc::new(AtomicU32::new(0)),
+        draining: Arc::new(AtomicBool::new(true)),
+        draining_started_at_epoch_secs: Arc::new(AtomicU64::new(
+            now_epoch_secs.saturating_sub(120),
+        )),
+        drain_deadline_epoch_secs: Arc::new(AtomicU64::new(0)),
+        allow_drain_fallback: Arc::new(AtomicBool::new(false)),
+    };
+    pool.writers.write().await.push(blocked_writer);
+    pool.registry
+        .register_writer(blocked_writer_id, blocked_tx)
+        .await;
+    pool.conn_count.fetch_add(1, Ordering::Relaxed);
+
+    insert_draining_writer(&pool, 91, now_epoch_secs.saturating_sub(110), 0, 0).await;
+
+    let mut warn_next_allowed = HashMap::new();
+    let mut soft_evict_next_allowed = HashMap::new();
+
+    let reap_res = tokio::time::timeout(
+        Duration::from_millis(500),
+        reap_draining_writers(&pool, &mut warn_next_allowed, &mut soft_evict_next_allowed),
+    )
+    .await;
+    blocked_rx_guard.abort();
+
+    assert!(reap_res.is_ok(), "reap should not block on close signal");
+    assert!(current_writer_ids(&pool).await.is_empty());
+    assert_eq!(pool.stats.get_me_writer_close_signal_drop_total(), 2);
+    assert_eq!(pool.stats.get_me_writer_close_signal_channel_full_total(), 1);
+    assert_eq!(pool.stats.get_me_draining_writers_reap_progress_total(), 2);
+    let activity = pool.registry.writer_activity_snapshot().await;
+    assert!(!activity.bound_clients_by_writer.contains_key(&blocked_writer_id));
+    assert!(!activity.bound_clients_by_writer.contains_key(&91));
+    let (probe_conn_id, _rx) = pool.registry.register().await;
+    assert!(
+        !pool.registry
+            .bind_writer(
+                probe_conn_id,
+                blocked_writer_id,
+                ConnMeta {
+                    target_dc: 2,
+                    client_addr: SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 6400),
+                    our_addr: SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 443),
+                    proto_flags: 0,
+                },
+            )
+            .await
+    );
+    let _ = pool.registry.unregister(probe_conn_id).await;
+}
+
 #[tokio::test]
 async fn reap_draining_writers_overflow_closes_oldest_non_empty_writers() {
     let pool = make_pool(2).await;
@@ -224,8 +317,9 @@ async fn reap_draining_writers_overflow_closes_oldest_non_empty_writers() {
     insert_draining_writer(&pool, 33, now_epoch_secs.saturating_sub(20), 1, 0).await;
     insert_draining_writer(&pool, 44, now_epoch_secs.saturating_sub(10), 1, 0).await;
     let mut warn_next_allowed = HashMap::new();
+    let mut soft_evict_next_allowed = HashMap::new();
 
-    reap_draining_writers(&pool, &mut warn_next_allowed).await;
+    reap_draining_writers(&pool, &mut warn_next_allowed, &mut soft_evict_next_allowed).await;
 
     assert_eq!(current_writer_ids(&pool).await, vec![33, 44]);
 }
@@ -243,8 +337,9 @@ async fn reap_draining_writers_deadline_force_close_applies_under_threshold() {
     )
     .await;
     let mut warn_next_allowed = HashMap::new();
+    let mut soft_evict_next_allowed = HashMap::new();
 
-    reap_draining_writers(&pool, &mut warn_next_allowed).await;
+    reap_draining_writers(&pool, &mut warn_next_allowed, &mut soft_evict_next_allowed).await;
 
     assert!(current_writer_ids(&pool).await.is_empty());
 }
@@ -266,8 +361,9 @@ async fn reap_draining_writers_limits_closes_per_health_tick() {
         .await;
     }
     let mut warn_next_allowed = HashMap::new();
+    let mut soft_evict_next_allowed = HashMap::new();
 
-    reap_draining_writers(&pool, &mut warn_next_allowed).await;
+    reap_draining_writers(&pool, &mut warn_next_allowed, &mut soft_evict_next_allowed).await;
 
     assert_eq!(pool.writers.read().await.len(), writer_total - close_budget);
 }
@@ -406,12 +502,13 @@ async fn reap_draining_writers_backlog_drains_across_ticks() {
         .await;
     }
     let mut warn_next_allowed = HashMap::new();
+    let mut soft_evict_next_allowed = HashMap::new();
 
     for _ in 0..8 {
         if pool.writers.read().await.is_empty() {
             break;
         }
-        reap_draining_writers(&pool, &mut warn_next_allowed).await;
+        reap_draining_writers(&pool, &mut warn_next_allowed, &mut soft_evict_next_allowed).await;
     }
 
     assert!(pool.writers.read().await.is_empty());
@@ -435,9 +532,10 @@ async fn reap_draining_writers_threshold_backlog_converges_to_threshold() {
         .await;
     }
     let mut warn_next_allowed = HashMap::new();
+    let mut soft_evict_next_allowed = HashMap::new();
 
     for _ in 0..16 {
-        reap_draining_writers(&pool, &mut warn_next_allowed).await;
+        reap_draining_writers(&pool, &mut warn_next_allowed, &mut soft_evict_next_allowed).await;
         if pool.writers.read().await.len() <= threshold as usize {
             break;
         }
@@ -454,8 +552,9 @@ async fn reap_draining_writers_threshold_zero_preserves_non_expired_non_empty_wr
     insert_draining_writer(&pool, 20, now_epoch_secs.saturating_sub(30), 1, 0).await;
     insert_draining_writer(&pool, 30, now_epoch_secs.saturating_sub(20), 1, 0).await;
     let mut warn_next_allowed = HashMap::new();
+    let mut soft_evict_next_allowed = HashMap::new();
 
-    reap_draining_writers(&pool, &mut warn_next_allowed).await;
+    reap_draining_writers(&pool, &mut warn_next_allowed, &mut soft_evict_next_allowed).await;
 
     assert_eq!(current_writer_ids(&pool).await, vec![10, 20, 30]);
 }
@@ -478,8 +577,9 @@ async fn reap_draining_writers_prioritizes_force_close_before_empty_cleanup() {
     let empty_writer_id = close_budget as u64 + 1;
     insert_draining_writer(&pool, empty_writer_id, now_epoch_secs.saturating_sub(20), 0, 0).await;
     let mut warn_next_allowed = HashMap::new();
+    let mut soft_evict_next_allowed = HashMap::new();
 
-    reap_draining_writers(&pool, &mut warn_next_allowed).await;
+    reap_draining_writers(&pool, &mut warn_next_allowed, &mut soft_evict_next_allowed).await;
 
     assert_eq!(current_writer_ids(&pool).await, vec![empty_writer_id]);
 }
@@ -491,8 +591,9 @@ async fn reap_draining_writers_empty_cleanup_does_not_increment_force_close_metr
     insert_draining_writer(&pool, 1, now_epoch_secs.saturating_sub(60), 0, 0).await;
     insert_draining_writer(&pool, 2, now_epoch_secs.saturating_sub(50), 0, 0).await;
     let mut warn_next_allowed = HashMap::new();
+    let mut soft_evict_next_allowed = HashMap::new();
 
-    reap_draining_writers(&pool, &mut warn_next_allowed).await;
+    reap_draining_writers(&pool, &mut warn_next_allowed, &mut soft_evict_next_allowed).await;
 
     assert!(current_writer_ids(&pool).await.is_empty());
     assert_eq!(pool.stats.get_pool_force_close_total(), 0);
@@ -519,8 +620,9 @@ async fn reap_draining_writers_handles_duplicate_force_close_requests_for_same_w
     )
     .await;
     let mut warn_next_allowed = HashMap::new();
+    let mut soft_evict_next_allowed = HashMap::new();
 
-    reap_draining_writers(&pool, &mut warn_next_allowed).await;
+    reap_draining_writers(&pool, &mut warn_next_allowed, &mut soft_evict_next_allowed).await;
 
     assert!(current_writer_ids(&pool).await.is_empty());
 }
@@ -530,6 +632,7 @@ async fn reap_draining_writers_warn_state_never_exceeds_live_draining_population
     let pool = make_pool(128).await;
     let now_epoch_secs = MePool::now_epoch_secs();
     let mut warn_next_allowed = HashMap::new();
+    let mut soft_evict_next_allowed = HashMap::new();
 
     for wave in 0..12u64 {
         for offset in 0..9u64 {
@@ -542,14 +645,14 @@ async fn reap_draining_writers_warn_state_never_exceeds_live_draining_population
             )
             .await;
         }
-        reap_draining_writers(&pool, &mut warn_next_allowed).await;
+        reap_draining_writers(&pool, &mut warn_next_allowed, &mut soft_evict_next_allowed).await;
         assert!(warn_next_allowed.len() <= pool.writers.read().await.len());
 
         let existing_writer_ids = current_writer_ids(&pool).await;
         for writer_id in existing_writer_ids.into_iter().take(4) {
             let _ = pool.remove_writer_and_close_clients(writer_id).await;
         }
-        reap_draining_writers(&pool, &mut warn_next_allowed).await;
+        reap_draining_writers(&pool, &mut warn_next_allowed, &mut soft_evict_next_allowed).await;
         assert!(warn_next_allowed.len() <= pool.writers.read().await.len());
     }
 }
@@ -559,6 +662,7 @@ async fn reap_draining_writers_mixed_backlog_converges_without_leaking_warn_stat
     let pool = make_pool(6).await;
     let now_epoch_secs = MePool::now_epoch_secs();
     let mut warn_next_allowed = HashMap::new();
+    let mut soft_evict_next_allowed = HashMap::new();
 
     for writer_id in 1..=18u64 {
         let bound_clients = if writer_id % 3 == 0 { 0 } else { 1 };
@@ -578,7 +682,7 @@ async fn reap_draining_writers_mixed_backlog_converges_without_leaking_warn_stat
     }
 
     for _ in 0..16 {
-        reap_draining_writers(&pool, &mut warn_next_allowed).await;
+        reap_draining_writers(&pool, &mut warn_next_allowed, &mut soft_evict_next_allowed).await;
         if pool.writers.read().await.len() <= 6 {
             break;
         }
@@ -588,9 +692,62 @@ async fn reap_draining_writers_mixed_backlog_converges_without_leaking_warn_stat
     assert!(warn_next_allowed.len() <= pool.writers.read().await.len());
 }
 
+#[tokio::test]
+async fn reap_draining_writers_soft_evicts_stuck_writer_with_per_writer_cap() {
+    let pool = make_pool(128).await;
+    pool.me_pool_drain_soft_evict_enabled.store(true, Ordering::Relaxed);
+    pool.me_pool_drain_soft_evict_grace_secs.store(0, Ordering::Relaxed);
+    pool.me_pool_drain_soft_evict_per_writer.store(1, Ordering::Relaxed);
+    pool.me_pool_drain_soft_evict_budget_per_core.store(8, Ordering::Relaxed);
+    pool.me_pool_drain_soft_evict_cooldown_ms
+        .store(1, Ordering::Relaxed);
+
+    let now_epoch_secs = MePool::now_epoch_secs();
+    insert_draining_writer(&pool, 77, now_epoch_secs.saturating_sub(240), 3, 0).await;
+    let mut warn_next_allowed = HashMap::new();
+    let mut soft_evict_next_allowed = HashMap::new();
+
+    reap_draining_writers(&pool, &mut warn_next_allowed, &mut soft_evict_next_allowed).await;
+
+    let activity = pool.registry.writer_activity_snapshot().await;
+    assert_eq!(activity.bound_clients_by_writer.get(&77), Some(&2));
+    assert_eq!(pool.stats.get_pool_drain_soft_evict_total(), 1);
+    assert_eq!(pool.stats.get_pool_drain_soft_evict_writer_total(), 1);
+    assert_eq!(current_writer_ids(&pool).await, vec![77]);
+}
+
+#[tokio::test]
+async fn reap_draining_writers_soft_evict_respects_cooldown_per_writer() {
+    let pool = make_pool(128).await;
+    pool.me_pool_drain_soft_evict_enabled.store(true, Ordering::Relaxed);
+    pool.me_pool_drain_soft_evict_grace_secs.store(0, Ordering::Relaxed);
+    pool.me_pool_drain_soft_evict_per_writer.store(1, Ordering::Relaxed);
+    pool.me_pool_drain_soft_evict_budget_per_core.store(8, Ordering::Relaxed);
+    pool.me_pool_drain_soft_evict_cooldown_ms
+        .store(60_000, Ordering::Relaxed);
+
+    let now_epoch_secs = MePool::now_epoch_secs();
+    insert_draining_writer(&pool, 88, now_epoch_secs.saturating_sub(240), 3, 0).await;
+    let mut warn_next_allowed = HashMap::new();
+    let mut soft_evict_next_allowed = HashMap::new();
+
+    reap_draining_writers(&pool, &mut warn_next_allowed, &mut soft_evict_next_allowed).await;
+    reap_draining_writers(&pool, &mut warn_next_allowed, &mut soft_evict_next_allowed).await;
+
+    let activity = pool.registry.writer_activity_snapshot().await;
+    assert_eq!(activity.bound_clients_by_writer.get(&88), Some(&2));
+    assert_eq!(pool.stats.get_pool_drain_soft_evict_total(), 1);
+    assert_eq!(pool.stats.get_pool_drain_soft_evict_writer_total(), 1);
+}
+
 #[test]
 fn general_config_default_drain_threshold_remains_enabled() {
     assert_eq!(GeneralConfig::default().me_pool_drain_threshold, 128);
+    assert!(GeneralConfig::default().me_pool_drain_soft_evict_enabled);
+    assert_eq!(
+        GeneralConfig::default().me_pool_drain_soft_evict_per_writer,
+        1
+    );
 }
 
 #[tokio::test]

src/transport/middle_proxy/pool_reinit.rs

@@ -70,10 +70,12 @@ impl MePool {
 
         let mut missing_dc = Vec::<i32>::new();
         let mut covered = 0usize;
+        let mut total = 0usize;
         for (dc, endpoints) in desired_by_dc {
             if endpoints.is_empty() {
                 continue;
             }
+            total += 1;
             if endpoints
                 .iter()
                 .any(|addr| active_writer_addrs.contains(&(*dc, *addr)))
@@ -85,7 +87,9 @@ impl MePool {
         }
 
         missing_dc.sort_unstable();
-        let total = desired_by_dc.len().max(1);
+        if total == 0 {
+            return (1.0, missing_dc);
+        }
         let ratio = (covered as f32) / (total as f32);
         (ratio, missing_dc)
     }
@@ -431,29 +435,21 @@ impl MePool {
         }
 
         if hardswap {
-            let mut fresh_missing_dc = Vec::<(i32, usize, usize)>::new();
-            for (dc, endpoints) in &desired_by_dc {
-                if endpoints.is_empty() {
-                    continue;
-                }
-                let required = self.required_writers_for_dc(endpoints.len());
-                let fresh_count = writers
-                    .iter()
-                    .filter(|w| !w.draining.load(Ordering::Relaxed))
-                    .filter(|w| w.generation == generation)
-                    .filter(|w| w.writer_dc == *dc)
-                    .filter(|w| endpoints.contains(&w.addr))
-                    .count();
-                if fresh_count < required {
-                    fresh_missing_dc.push((*dc, fresh_count, required));
-                }
-            }
+            let fresh_writer_addrs: HashSet<(i32, SocketAddr)> = writers
+                .iter()
+                .filter(|w| !w.draining.load(Ordering::Relaxed))
+                .filter(|w| w.generation == generation)
+                .map(|w| (w.writer_dc, w.addr))
+                .collect();
+            let (fresh_coverage_ratio, fresh_missing_dc) =
+                Self::coverage_ratio(&desired_by_dc, &fresh_writer_addrs);
             if !fresh_missing_dc.is_empty() {
                 warn!(
                     previous_generation,
                     generation,
+                    fresh_coverage_ratio = format_args!("{fresh_coverage_ratio:.3}"),
                     missing_dc = ?fresh_missing_dc,
-                    "ME hardswap pending: fresh generation coverage incomplete"
+                    "ME hardswap pending: fresh generation DC coverage incomplete"
                 );
                 return;
             }
@@ -541,3 +537,61 @@ impl MePool {
         self.zero_downtime_reinit_after_map_change(rng).await;
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use std::collections::{HashMap, HashSet};
+    use std::net::{IpAddr, Ipv4Addr, SocketAddr};
+
+    use super::MePool;
+
+    fn addr(octet: u8, port: u16) -> SocketAddr {
+        SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, octet)), port)
+    }
+
+    #[test]
+    fn coverage_ratio_counts_dc_coverage_not_floor() {
+        let dc1 = addr(1, 2001);
+        let dc2 = addr(2, 2002);
+
+        let mut desired_by_dc = HashMap::<i32, HashSet<SocketAddr>>::new();
+        desired_by_dc.insert(1, HashSet::from([dc1]));
+        desired_by_dc.insert(2, HashSet::from([dc2]));
+
+        let active_writer_addrs = HashSet::from([(1, dc1)]);
+        let (ratio, missing_dc) = MePool::coverage_ratio(&desired_by_dc, &active_writer_addrs);
+
+        assert_eq!(ratio, 0.5);
+        assert_eq!(missing_dc, vec![2]);
+    }
+
+    #[test]
+    fn coverage_ratio_ignores_empty_dc_groups() {
+        let dc1 = addr(1, 2001);
+
+        let mut desired_by_dc = HashMap::<i32, HashSet<SocketAddr>>::new();
+        desired_by_dc.insert(1, HashSet::from([dc1]));
+        desired_by_dc.insert(2, HashSet::new());
+
+        let active_writer_addrs = HashSet::from([(1, dc1)]);
+        let (ratio, missing_dc) = MePool::coverage_ratio(&desired_by_dc, &active_writer_addrs);
+
+        assert_eq!(ratio, 1.0);
+        assert!(missing_dc.is_empty());
+    }
+
+    #[test]
+    fn coverage_ratio_reports_missing_dcs_sorted() {
+        let dc1 = addr(1, 2001);
+        let dc2 = addr(2, 2002);
+
+        let mut desired_by_dc = HashMap::<i32, HashSet<SocketAddr>>::new();
+        desired_by_dc.insert(2, HashSet::from([dc2]));
+        desired_by_dc.insert(1, HashSet::from([dc1]));
+
+        let (ratio, missing_dc) = MePool::coverage_ratio(&desired_by_dc, &HashSet::new());
+
+        assert_eq!(ratio, 0.0);
+        assert_eq!(missing_dc, vec![1, 2]);
+    }
+}

src/transport/middle_proxy/pool_status.rs

@@ -40,6 +40,7 @@ pub(crate) struct MeApiDcStatusSnapshot {
     pub floor_max: usize,
     pub floor_capped: bool,
     pub alive_writers: usize,
+    pub coverage_ratio: f64,
     pub coverage_pct: f64,
     pub fresh_alive_writers: usize,
     pub fresh_coverage_pct: f64,
@@ -62,6 +63,7 @@ pub(crate) struct MeApiStatusSnapshot {
     pub available_pct: f64,
     pub required_writers: usize,
     pub alive_writers: usize,
+    pub coverage_ratio: f64,
     pub coverage_pct: f64,
     pub fresh_alive_writers: usize,
     pub fresh_coverage_pct: f64,
@@ -124,6 +126,11 @@ pub(crate) struct MeApiRuntimeSnapshot {
     pub me_reconnect_backoff_cap_ms: u64,
     pub me_reconnect_fast_retry_count: u32,
     pub me_pool_drain_ttl_secs: u64,
+    pub me_pool_drain_soft_evict_enabled: bool,
+    pub me_pool_drain_soft_evict_grace_secs: u64,
+    pub me_pool_drain_soft_evict_per_writer: u8,
+    pub me_pool_drain_soft_evict_budget_per_core: u16,
+    pub me_pool_drain_soft_evict_cooldown_ms: u64,
     pub me_pool_force_close_secs: u64,
     pub me_pool_min_fresh_ratio: f32,
     pub me_bind_stale_mode: &'static str,
@@ -337,6 +344,8 @@ impl MePool {
         let mut available_endpoints = 0usize;
         let mut alive_writers = 0usize;
         let mut fresh_alive_writers = 0usize;
+        let mut coverage_ratio_dcs_total = 0usize;
+        let mut coverage_ratio_dcs_covered = 0usize;
         let floor_mode = self.floor_mode();
         let adaptive_cpu_cores = (self
             .me_adaptive_floor_cpu_cores_effective
@@ -388,6 +397,12 @@ impl MePool {
             available_endpoints += dc_available_endpoints;
             alive_writers += dc_alive_writers;
             fresh_alive_writers += dc_fresh_alive_writers;
+            if endpoint_count > 0 {
+                coverage_ratio_dcs_total += 1;
+                if dc_alive_writers > 0 {
+                    coverage_ratio_dcs_covered += 1;
+                }
+            }
 
             dcs.push(MeApiDcStatusSnapshot {
                 dc,
@@ -410,6 +425,11 @@ impl MePool {
                 floor_max,
                 floor_capped,
                 alive_writers: dc_alive_writers,
+                coverage_ratio: if endpoint_count > 0 && dc_alive_writers > 0 {
+                    100.0
+                } else {
+                    0.0
+                },
                 coverage_pct: ratio_pct(dc_alive_writers, dc_required_writers),
                 fresh_alive_writers: dc_fresh_alive_writers,
                 fresh_coverage_pct: ratio_pct(dc_fresh_alive_writers, dc_required_writers),
@@ -426,6 +446,7 @@ impl MePool {
             available_pct: ratio_pct(available_endpoints, configured_endpoints),
             required_writers,
             alive_writers,
+            coverage_ratio: ratio_pct(coverage_ratio_dcs_covered, coverage_ratio_dcs_total),
             coverage_pct: ratio_pct(alive_writers, required_writers),
             fresh_alive_writers,
             fresh_coverage_pct: ratio_pct(fresh_alive_writers, required_writers),
@@ -562,6 +583,22 @@ impl MePool {
             me_reconnect_backoff_cap_ms: self.me_reconnect_backoff_cap.as_millis() as u64,
             me_reconnect_fast_retry_count: self.me_reconnect_fast_retry_count,
             me_pool_drain_ttl_secs: self.me_pool_drain_ttl_secs.load(Ordering::Relaxed),
+            me_pool_drain_soft_evict_enabled: self
+                .me_pool_drain_soft_evict_enabled
+                .load(Ordering::Relaxed),
+            me_pool_drain_soft_evict_grace_secs: self
+                .me_pool_drain_soft_evict_grace_secs
+                .load(Ordering::Relaxed),
+            me_pool_drain_soft_evict_per_writer: self
+                .me_pool_drain_soft_evict_per_writer
+                .load(Ordering::Relaxed),
+            me_pool_drain_soft_evict_budget_per_core: self
+                .me_pool_drain_soft_evict_budget_per_core
+                .load(Ordering::Relaxed)
+                .min(u16::MAX as u32) as u16,
+            me_pool_drain_soft_evict_cooldown_ms: self
+                .me_pool_drain_soft_evict_cooldown_ms
+                .load(Ordering::Relaxed),
             me_pool_force_close_secs: self.me_pool_force_close_secs.load(Ordering::Relaxed),
             me_pool_min_fresh_ratio: Self::permille_to_ratio(
                 self.me_pool_min_fresh_ratio_permille.load(Ordering::Relaxed),

src/transport/middle_proxy/pool_writer.rs

@@ -8,6 +8,7 @@ use bytes::Bytes;
 use bytes::BytesMut;
 use rand::Rng;
 use tokio::sync::mpsc;
+use tokio::sync::mpsc::error::TrySendError;
 use tokio_util::sync::CancellationToken;
 use tracing::{debug, info, warn};
 
@@ -311,41 +312,28 @@ impl MePool {
                 let mut p = Vec::with_capacity(12);
                 p.extend_from_slice(&RPC_PING_U32.to_le_bytes());
                 p.extend_from_slice(&sent_id.to_le_bytes());
-                {
-                    let mut tracker = ping_tracker_ping.lock().await;
-                    let now_epoch_ms = std::time::SystemTime::now()
-                        .duration_since(std::time::UNIX_EPOCH)
-                        .unwrap_or_default()
-                        .as_millis() as u64;
-                    let mut run_cleanup = false;
-                    if let Some(pool) = pool_ping.upgrade() {
-                        let last_cleanup_ms = pool
+                let now_epoch_ms = std::time::SystemTime::now()
+                    .duration_since(std::time::UNIX_EPOCH)
+                    .unwrap_or_default()
+                    .as_millis() as u64;
+                let mut run_cleanup = false;
+                if let Some(pool) = pool_ping.upgrade() {
+                    let last_cleanup_ms = pool
+                        .ping_tracker_last_cleanup_epoch_ms
+                        .load(Ordering::Relaxed);
+                    if now_epoch_ms.saturating_sub(last_cleanup_ms) >= 30_000
+                        && pool
                             .ping_tracker_last_cleanup_epoch_ms
-                            .load(Ordering::Relaxed);
-                        if now_epoch_ms.saturating_sub(last_cleanup_ms) >= 30_000
-                            && pool
-                                .ping_tracker_last_cleanup_epoch_ms
-                                .compare_exchange(
-                                    last_cleanup_ms,
-                                    now_epoch_ms,
-                                    Ordering::AcqRel,
-                                    Ordering::Relaxed,
-                                )
-                                .is_ok()
-                        {
-                            run_cleanup = true;
-                        }
+                            .compare_exchange(
+                                last_cleanup_ms,
+                                now_epoch_ms,
+                                Ordering::AcqRel,
+                                Ordering::Relaxed,
+                            )
+                            .is_ok()
+                    {
+                        run_cleanup = true;
                     }
-
-                    if run_cleanup {
-                        let before = tracker.len();
-                        tracker.retain(|_, (ts, _)| ts.elapsed() < Duration::from_secs(120));
-                        let expired = before.saturating_sub(tracker.len());
-                        if expired > 0 {
-                            stats_ping.increment_me_keepalive_timeout_by(expired as u64);
-                        }
-                    }
-                    tracker.insert(sent_id, (std::time::Instant::now(), writer_id));
                 }
                 ping_id = ping_id.wrapping_add(1);
                 stats_ping.increment_me_keepalive_sent();
@@ -366,6 +354,16 @@ impl MePool {
                     }
                     break;
                 }
+                let mut tracker = ping_tracker_ping.lock().await;
+                if run_cleanup {
+                    let before = tracker.len();
+                    tracker.retain(|_, (ts, _)| ts.elapsed() < Duration::from_secs(120));
+                    let expired = before.saturating_sub(tracker.len());
+                    if expired > 0 {
+                        stats_ping.increment_me_keepalive_timeout_by(expired as u64);
+                    }
+                }
+                tracker.insert(sent_id, (std::time::Instant::now(), writer_id));
             }
         });
 
@@ -493,11 +491,9 @@ impl MePool {
     }
 
     pub(crate) async fn remove_writer_and_close_clients(self: &Arc<Self>, writer_id: u64) {
-        let conns = self.remove_writer_only(writer_id).await;
-        for bound in conns {
-            let _ = self.registry.route(bound.conn_id, super::MeResponse::Close).await;
-            let _ = self.registry.unregister(bound.conn_id).await;
-        }
+        // Full client cleanup now happens inside `registry.writer_lost` to keep
+        // writer reap/remove paths strictly non-blocking per connection.
+        let _ = self.remove_writer_only(writer_id).await;
     }
 
     pub(crate) async fn remove_writer_if_empty(self: &Arc<Self>, writer_id: u64) -> bool {
@@ -539,6 +535,11 @@ impl MePool {
                 self.conn_count.fetch_sub(1, Ordering::Relaxed);
             }
         }
+        // State invariant:
+        // - writer is removed from `self.writers` (pool visibility),
+        // - writer is removed from registry routing/binding maps via `writer_lost`.
+        // The close command below is only a best-effort accelerator for task shutdown.
+        // Cleanup progress must never depend on command-channel availability.
         let conns = self.registry.writer_lost(writer_id).await;
         {
             let mut tracker = self.ping_tracker.lock().await;
@@ -546,7 +547,25 @@ impl MePool {
         }
         self.rtt_stats.lock().await.remove(&writer_id);
         if let Some(tx) = close_tx {
-            let _ = tx.send(WriterCommand::Close).await;
+            match tx.try_send(WriterCommand::Close) {
+                Ok(()) => {}
+                Err(TrySendError::Full(_)) => {
+                    self.stats.increment_me_writer_close_signal_drop_total();
+                    self.stats
+                        .increment_me_writer_close_signal_channel_full_total();
+                    debug!(
+                        writer_id,
+                        "Skipping close signal for removed writer: command channel is full"
+                    );
+                }
+                Err(TrySendError::Closed(_)) => {
+                    self.stats.increment_me_writer_close_signal_drop_total();
+                    debug!(
+                        writer_id,
+                        "Skipping close signal for removed writer: command channel is closed"
+                    );
+                }
+            }
         }
         if trigger_refill
             && let Some(addr) = removed_addr

src/transport/middle_proxy/reader.rs

@@ -8,6 +8,7 @@ use bytes::{Bytes, BytesMut};
 use tokio::io::AsyncReadExt;
 use tokio::net::TcpStream;
 use tokio::sync::{Mutex, mpsc};
+use tokio::sync::mpsc::error::TrySendError;
 use tokio_util::sync::CancellationToken;
 use tracing::{debug, trace, warn};
 
@@ -173,12 +174,12 @@ pub(crate) async fn reader_loop(
             } else if pt == RPC_CLOSE_EXT_U32 && body.len() >= 8 {
                 let cid = u64::from_le_bytes(body[0..8].try_into().unwrap());
                 debug!(cid, "RPC_CLOSE_EXT from ME");
-                reg.route(cid, MeResponse::Close).await;
+                let _ = reg.route_nowait(cid, MeResponse::Close).await;
                 reg.unregister(cid).await;
             } else if pt == RPC_CLOSE_CONN_U32 && body.len() >= 8 {
                 let cid = u64::from_le_bytes(body[0..8].try_into().unwrap());
                 debug!(cid, "RPC_CLOSE_CONN from ME");
-                reg.route(cid, MeResponse::Close).await;
+                let _ = reg.route_nowait(cid, MeResponse::Close).await;
                 reg.unregister(cid).await;
             } else if pt == RPC_PING_U32 && body.len() >= 8 {
                 let ping_id = i64::from_le_bytes(body[0..8].try_into().unwrap());
@@ -186,13 +187,15 @@ pub(crate) async fn reader_loop(
                 let mut pong = Vec::with_capacity(12);
                 pong.extend_from_slice(&RPC_PONG_U32.to_le_bytes());
                 pong.extend_from_slice(&ping_id.to_le_bytes());
-                if tx
-                    .send(WriterCommand::DataAndFlush(Bytes::from(pong)))
-                    .await
-                    .is_err()
-                {
-                    warn!("PONG send failed");
-                    break;
+                match tx.try_send(WriterCommand::DataAndFlush(Bytes::from(pong))) {
+                    Ok(()) => {}
+                    Err(TrySendError::Full(_)) => {
+                        debug!(ping_id, "PONG dropped: writer command channel is full");
+                    }
+                    Err(TrySendError::Closed(_)) => {
+                        warn!("PONG send failed: writer channel closed");
+                        break;
+                    }
                 }
             } else if pt == RPC_PONG_U32 && body.len() >= 8 {
                 let ping_id = i64::from_le_bytes(body[0..8].try_into().unwrap());
@@ -232,6 +235,13 @@ async fn send_close_conn(tx: &mpsc::Sender<WriterCommand>, conn_id: u64) {
     let mut p = Vec::with_capacity(12);
     p.extend_from_slice(&RPC_CLOSE_CONN_U32.to_le_bytes());
     p.extend_from_slice(&conn_id.to_le_bytes());
-
-    let _ = tx.send(WriterCommand::DataAndFlush(Bytes::from(p))).await;
+    match tx.try_send(WriterCommand::DataAndFlush(Bytes::from(p))) {
+        Ok(()) => {}
+        Err(TrySendError::Full(_)) => {
+            debug!(conn_id, "ME close_conn signal skipped: writer command channel is full");
+        }
+        Err(TrySendError::Closed(_)) => {
+            debug!(conn_id, "ME close_conn signal skipped: writer command channel is closed");
+        }
+    }
 }