From 0255fe31c018c6ca2ce9405e0e9890389d9fc07c Mon Sep 17 00:00:00 2001
From: Austin Barrington <austin@hyperbyte.dev>
Date: Tue, 20 Oct 2026 09:19:14 +0100
Subject: [PATCH 1/2] Keep every acknowledged write through a rolling restart

Replicated (sync_quorum + hinted handoff): over one rolling restart of
three nodes the replicas ended 643, 1,825 and 1,051 of 3,204 acked
writes short, and stayed so. Now each holds all of them.

- Replication goes to every member not leaving the cluster. A peer that
  was Syncing, Joining or Draining (every restarting node) was neither
  sent writes nor hinted them.
- The sync trigger attached the DisabledShardMap, so a replicated node
  took the sharded reconnect path: no WAL catch-up, and a shard map its
  peers answer with 404. needs_sync stayed set and peers never
  readmitted the node.
- Hydrating membership from the Raft state machine replaced a node's
  own Active with the Syncing it was added with; the persisted state is
  now merged, keeping self-reported sync state and operator states.
- The WAL truncate barrier and the sync_quorum ack count used only
  Active peers: a syncing peer's WAL was truncated away, and with both
  peers syncing a write was acked with this node's copy alone.
- Replication sends still in flight at shutdown are waited for (sent or
  hinted); they were dropped with the process.
- The drain no longer waits on WAL acks nothing records (90 s, every
  restart of a node that had written).

Sharded: a write whose region primary is unreachable is held in the
forward queue and acknowledged, not failed with 500 for the whole
restart. Queued forwards are dropped after an hour, not 100 attempts.

Proxy: a backend whose forward fails (connection refused/reset, or the
draining 503) leaves rotation at once, so the request is held for a
healthy one; until its next probe it was picked again and every retry
failed within milliseconds.
---
 hyperbytedb-proxy/src/proxy.rs                |  31 +++-
 .../tests/hold_on_failed_backend.rs           |  88 ++++++++++
 .../src/adapters/cluster/peer_client.rs       |  89 +++++++++-
 .../src/adapters/cluster/raft/log_store.rs    |  80 ++++++++-
 .../src/adapters/http/peer_handlers.rs        |   9 +-
 hyperbytedb/src/application/cluster/drain.rs  |  13 +-
 hyperbytedb/src/application/flush_service.rs  |   5 +-
 .../src/application/peer_ingestion_service.rs | 156 ++++++++++++++----
 .../src/application/replication_dispatch.rs   |   2 +-
 hyperbytedb/src/domain/cluster/membership.rs  |  27 ++-
 hyperbytedb/src/ports/replication.rs          |   4 +-
 hyperbytedb/src/server.rs                     |  12 ++
 .../tests/sharding_cluster_integration.rs     |  68 +++++++-
 hyperbytedb/tests/sync_quorum_integration.rs  |  48 ++++++
 14 files changed, 565 insertions(+), 67 deletions(-)
 create mode 100644 hyperbytedb-proxy/tests/hold_on_failed_backend.rs

diff --git a/hyperbytedb-proxy/src/proxy.rs b/hyperbytedb-proxy/src/proxy.rs
index a60ea50..4604c29 100644
--- a/hyperbytedb-proxy/src/proxy.rs
+++ b/hyperbytedb-proxy/src/proxy.rs
@@ -24,7 +24,7 @@ use futures::StreamExt;
 use futures::ready;
 use http_body_util::BodyExt;
 
-use crate::backend::InflightGuard;
+use crate::backend::{Health, InflightGuard};
 use crate::pool::BackendPool;
 
 /// Headers that hop-by-hop semantics (RFC 7230 §6.1) say we must not forward.
@@ -148,7 +148,22 @@ pub async fn handle(State(state): State<ProxyState>, req: Request) -> Response {
                     .record(started.elapsed().as_secs_f64());
                 return attach_inflight_guard(resp, guard);
             }
-            ForwardOutcome::Retryable { status, msg } => {
+            ForwardOutcome::Retryable { status, msg, mark } => {
+                // Take it out of rotation now. Its health probe only notices
+                // on its next run, about a second later, and until then the
+                // pick below handed the same dead backend straight back
+                // (always, in a one-backend pool): every retry failed within
+                // milliseconds and the client got 503 instead of being held
+                // for `hold_timeout`. The probe restores it once it answers.
+                if let Some(health) = mark
+                    && current.set_health(health).is_some()
+                {
+                    tracing::info!(
+                        backend = %current.addr,
+                        to = health.as_str(),
+                        "backend taken out of rotation after a failed forward"
+                    );
+                }
                 tracing::info!(
                     attempt,
                     backend = %current.addr,
@@ -213,6 +228,9 @@ enum ForwardOutcome {
     Retryable {
         status: Option<StatusCode>,
         msg: Option<String>,
+        /// What the failure says about the backend's health, applied at once
+        /// rather than at its next probe.
+        mark: Option<Health>,
     },
     /// Don't retry: the failure is the client's fault (4xx) and shouldn't be
     /// blamed on backend health.
@@ -243,9 +261,12 @@ async fn forward_once(
     let resp = match rb.send().await {
         Ok(r) => r,
         Err(e) => {
-            // Transport error → backend candidate failed; retry elsewhere.
+            // Transport error → backend candidate failed; retry elsewhere. A
+            // refused or reset connection means the backend is gone (a pod
+            // stopping); a timeout may be a slow request on a live one.
             return ForwardOutcome::Retryable {
                 status: None,
+                mark: (!e.is_timeout()).then_some(Health::Down),
                 msg: Some(e.to_string()),
             };
         }
@@ -260,6 +281,7 @@ async fn forward_once(
     if matches!(upstream_status.as_u16(), 502 | 504,) {
         return ForwardOutcome::Retryable {
             status: Some(upstream_status),
+            mark: None,
             msg: Some("backend returned bad-gateway/timeout".into()),
         };
     }
@@ -273,6 +295,7 @@ async fn forward_once(
             Err(e) => {
                 return ForwardOutcome::Retryable {
                     status: Some(upstream_status),
+                    mark: None,
                     msg: Some(format!("response body read: {e}")),
                 };
             }
@@ -280,6 +303,7 @@ async fn forward_once(
         if looks_like_drain(&body_bytes) {
             return ForwardOutcome::Retryable {
                 status: Some(upstream_status),
+                mark: Some(Health::Draining),
                 msg: Some("backend reports draining/syncing".into()),
             };
         }
@@ -298,6 +322,7 @@ async fn forward_once(
             Err(e) => {
                 return ForwardOutcome::Retryable {
                     status: Some(upstream_status),
+                    mark: None,
                     msg: Some(format!("response body read: {e}")),
                 };
             }
diff --git a/hyperbytedb-proxy/tests/hold_on_failed_backend.rs b/hyperbytedb-proxy/tests/hold_on_failed_backend.rs
new file mode 100644
index 0000000..b1530be
--- /dev/null
+++ b/hyperbytedb-proxy/tests/hold_on_failed_backend.rs
@@ -0,0 +1,88 @@
+//! A request whose backend has just gone down is held until a backend is
+//! back, not failed with 503.
+
+use std::sync::Arc;
+use std::time::Duration;
+
+use axum::Router;
+use axum::http::StatusCode;
+use axum::routing::{any, get};
+use hyperbytedb_proxy::backend::{Backend, Health};
+use hyperbytedb_proxy::config::ProxyConfig;
+use hyperbytedb_proxy::pool::BackendPool;
+use hyperbytedb_proxy::proxy::{ProxyState, handle};
+
+fn config(port: u16) -> ProxyConfig {
+    ProxyConfig {
+        listen_addr: "127.0.0.1:0".into(),
+        admin_listen_addr: "127.0.0.1:0".into(),
+        backend_service: "127.0.0.1".into(),
+        backend_port: port,
+        discovery_interval: Duration::from_secs(3600),
+        health_interval: Duration::from_millis(100),
+        health_path: "/health".into(),
+        health_timeout: Duration::from_millis(200),
+        request_timeout: Duration::from_secs(5),
+        hold_timeout: Duration::from_secs(5),
+        max_retries: 3,
+        shutdown_grace: Duration::from_secs(1),
+        self_ip: None,
+        http2_prior_knowledge: false,
+    }
+}
+
+/// The backend went down a moment ago and its last probe still says Active
+/// (probes run about every second): a pool of one, as for a single-node pool
+/// in a rolling restart. The request used to fail at once with 503, every
+/// retry handed the same dead backend; now the failed forward takes it out
+/// of rotation and the request waits for it to come back.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn a_request_to_a_backend_that_just_went_down_is_held() {
+    // A port with nothing listening yet.
+    let port = {
+        let l = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
+        l.local_addr().unwrap().port()
+    };
+    let pool = BackendPool::new(config(port)).unwrap();
+    let backend = Arc::new(Backend::new("127.0.0.1".parse().unwrap(), port));
+    backend.set_health(Health::Active);
+    pool.set_backends_for_test(vec![Arc::clone(&backend)]).await;
+
+    let proxy = Router::new()
+        .route("/write", any(handle))
+        .with_state(ProxyState::new(Arc::clone(&pool)).unwrap());
+    let proxy_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let proxy_url = format!("http://{}", proxy_listener.local_addr().unwrap());
+    tokio::spawn(async move { axum::serve(proxy_listener, proxy).await.unwrap() });
+
+    let request = tokio::spawn(async move {
+        reqwest::Client::new()
+            .post(format!("{proxy_url}/write?db=d"))
+            .body("m v=1")
+            .send()
+            .await
+            .unwrap()
+            .status()
+    });
+
+    // The backend comes back 500 ms later, and its probes start.
+    tokio::time::sleep(Duration::from_millis(500)).await;
+    assert!(
+        !request.is_finished(),
+        "held while the backend is down, not failed"
+    );
+    let app = Router::new()
+        .route("/write", any(|| async { StatusCode::NO_CONTENT }))
+        .route("/health", get(|| async { StatusCode::OK }));
+    let listener = tokio::net::TcpListener::bind(("127.0.0.1", port))
+        .await
+        .unwrap();
+    tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
+    tokio::spawn(Arc::clone(&pool).run_health());
+
+    let status = tokio::time::timeout(Duration::from_secs(5), request)
+        .await
+        .expect("answered once the backend is back")
+        .unwrap();
+    assert_eq!(status, reqwest::StatusCode::NO_CONTENT);
+}
diff --git a/hyperbytedb/src/adapters/cluster/peer_client.rs b/hyperbytedb/src/adapters/cluster/peer_client.rs
index c754866..b253bcb 100644
--- a/hyperbytedb/src/adapters/cluster/peer_client.rs
+++ b/hyperbytedb/src/adapters/cluster/peer_client.rs
@@ -1,6 +1,6 @@
 use bytes::Bytes;
 use metrics::{counter, histogram};
-use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
 use std::sync::{Arc, Mutex};
 use std::time::Duration;
 use tokio::sync::{Semaphore, oneshot};
@@ -40,6 +40,42 @@ pub struct PeerClient {
     outbound_started: AtomicBool,
     batch_semaphore: Arc<Semaphore>,
     max_coalesce_body_bytes: usize,
+    /// Writes queued or being sent to peers. Shutdown waits for them: a
+    /// `sync_quorum` write returns once enough peers acked, and its sends to
+    /// the others ran on in spawned tasks that the exit simply dropped, so
+    /// those peers never got the write, nor a hint for it.
+    inflight: Arc<InFlight>,
+}
+
+/// A count of replication work not yet done, and a wake-up when it is all done.
+#[derive(Default)]
+struct InFlight {
+    count: AtomicUsize,
+    idle: tokio::sync::Notify,
+}
+
+/// Holds `n` units of [`InFlight`] until dropped.
+struct InFlightGuard {
+    inflight: Arc<InFlight>,
+    n: usize,
+}
+
+impl InFlight {
+    fn track(self: &Arc<Self>, n: usize) -> InFlightGuard {
+        self.count.fetch_add(n, Ordering::SeqCst);
+        InFlightGuard {
+            inflight: Arc::clone(self),
+            n,
+        }
+    }
+}
+
+impl Drop for InFlightGuard {
+    fn drop(&mut self) {
+        if self.n > 0 && self.inflight.count.fetch_sub(self.n, Ordering::SeqCst) == self.n {
+            self.inflight.idle.notify_waiters();
+        }
+    }
 }
 
 /// The HTTP client for peer traffic.
@@ -98,6 +134,7 @@ impl PeerClient {
             outbound_started: AtomicBool::new(false),
             batch_semaphore: Arc::new(Semaphore::new(inflight)),
             max_coalesce_body_bytes: max_coalesce_body_bytes.max(1024),
+            inflight: Arc::default(),
         }
     }
 
@@ -177,6 +214,8 @@ impl PeerClient {
         mut rx: tokio::sync::mpsc::Receiver<OutboundReplicationBatch>,
     ) {
         while let Some(first) = rx.recv().await {
+            // Each job was counted when queued; this round's guard releases
+            // them once sent (or hinted).
             // Take everything already queued, up to one request's worth.
             let mut bytes = first.body.len();
             let mut jobs = vec![first];
@@ -189,6 +228,10 @@ impl PeerClient {
                     Err(_) => break,
                 }
             }
+            let queued = InFlightGuard {
+                inflight: Arc::clone(&self.inflight),
+                n: jobs.len(),
+            };
             let jobs = coalesce_jobs(jobs, self.max_coalesce_body_bytes);
 
             let permit = match self.batch_semaphore.clone().acquire_owned().await {
@@ -199,6 +242,7 @@ impl PeerClient {
             tokio::spawn(async move {
                 this.replicate_round(jobs).await;
                 drop(permit);
+                drop(queued);
             });
         }
     }
@@ -401,6 +445,24 @@ impl PeerClient {
         futures::future::join_all(sends).await;
     }
 
+    /// Wait, up to `timeout`, for every queued or in-flight send to a peer to
+    /// finish: delivered, or hinted. Returns how many were still pending.
+    /// Shutdown calls it so writes this node acknowledged reach every peer
+    /// it owes them to, or that peer's hints.
+    pub async fn wait_for_inflight(&self, timeout: Duration) -> usize {
+        let deadline = tokio::time::Instant::now() + timeout;
+        loop {
+            let idle = self.inflight.idle.notified();
+            let pending = self.inflight.count.load(Ordering::SeqCst);
+            if pending == 0 {
+                return 0;
+            }
+            if tokio::time::timeout_at(deadline, idle).await.is_err() {
+                return self.inflight.count.load(Ordering::SeqCst);
+            }
+        }
+    }
+
     pub fn node_addr(&self) -> &str {
         &self.node_addr
     }
@@ -421,9 +483,17 @@ impl PeerClient {
         batch: OutboundReplicationBatch,
     ) -> Result<(), HyperbytedbError> {
         self.start_outbound_processor();
+        // Counted before it is queued: the processor may take it at once.
+        // Released by the round that sends it ([`Self::outbound_loop`]).
+        std::mem::forget(self.inflight.track(1));
         match self.outbound_tx.try_send(batch) {
             Ok(()) => Ok(()),
             Err(tokio::sync::mpsc::error::TrySendError::Full(batch)) => {
+                // Not queued after all: held instead until it is hinted.
+                let queued = InFlightGuard {
+                    inflight: Arc::clone(&self.inflight),
+                    n: 1,
+                };
                 counter!("hyperbytedb_replication_queue_full_total").increment(1);
                 if self.hinted_handoff.is_none() {
                     return Err(HyperbytedbError::Internal(
@@ -440,10 +510,15 @@ impl PeerClient {
                         );
                         counter!("hyperbytedb_replication_queue_drops_total").increment(1);
                     }
+                    drop(queued);
                 });
                 Ok(())
             }
             Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => {
+                drop(InFlightGuard {
+                    inflight: Arc::clone(&self.inflight),
+                    n: 1,
+                });
                 tracing::error!("replication outbound queue closed; dropping batch");
                 counter!("hyperbytedb_replication_queue_drops_total").increment(1);
                 Err(HyperbytedbError::Internal(
@@ -569,9 +644,13 @@ impl PeerClient {
                 /* sync = */ true,
                 batch.region_id,
             );
+            // Held until this peer acks or its hint is written, which may be
+            // after this write has returned to the client.
+            let sending = self.inflight.track(1);
             tokio::spawn(async move {
                 let res = ctx.run().await;
                 let _ = tx.send(res);
+                drop(sending);
             });
         }
 
@@ -915,9 +994,13 @@ impl ReplicationPort for PeerClient {
         PeerClient::replicate_mutation_sync(&self, req).await
     }
 
-    async fn active_peer_count(&self, self_node_id: u64) -> usize {
+    /// The peers a `sync_quorum` write counts its acks against: the ones it
+    /// replicates to, not only the `Active` ones. Counting only `Active` peers
+    /// shrank the quorum while peers restarted: with both peers `Syncing`,
+    /// "majority" of zero peers acknowledged a write held by this node alone.
+    async fn replication_peer_count(&self, self_node_id: u64) -> usize {
         let m = self.membership.read().await;
-        m.active_peers(self_node_id).len()
+        m.replication_peers(self_node_id).len()
     }
 }
 
diff --git a/hyperbytedb/src/adapters/cluster/raft/log_store.rs b/hyperbytedb/src/adapters/cluster/raft/log_store.rs
index 72dbd78..83db739 100644
--- a/hyperbytedb/src/adapters/cluster/raft/log_store.rs
+++ b/hyperbytedb/src/adapters/cluster/raft/log_store.rs
@@ -89,6 +89,38 @@ pub(crate) fn merge_snapshot_membership(
     }
 }
 
+/// Merge the state machine's persisted membership into the live one at
+/// startup. Persisted operator states win (a node restarted mid-decommission
+/// stays `Decommissioning`), but the Active/Syncing/Joining axis is
+/// self-reported: a persisted value there never overrides what is live.
+///
+/// This replaced the whole live membership, this node's own record included.
+/// The state machine still held the state a node had when it was added
+/// (learners start `Syncing`), so a node that had just finished its startup
+/// sync went back to `Syncing`, its `/health` said so, and its peers -- which
+/// readmit a node only on a ready self-report -- kept it out: no writes, no
+/// replication, until it restarted again.
+pub(crate) fn merge_persisted_membership(
+    shared: &mut crate::domain::cluster::membership::ClusterMembership,
+    persisted: crate::domain::cluster::membership::ClusterMembership,
+) {
+    use crate::domain::cluster::membership::NodeState::{Active, Joining, Syncing};
+    shared.version = shared.version.max(persisted.version);
+    for (id, info) in persisted.nodes {
+        match shared.nodes.get_mut(&id) {
+            Some(existing) => {
+                existing.addr = info.addr;
+                let self_reported = matches!(existing.state, Active | Syncing | Joining)
+                    && matches!(info.state, Active | Syncing | Joining);
+                if !self_reported {
+                    existing.state = info.state;
+                }
+            }
+            None => shared.add_node(info),
+        }
+    }
+}
+
 fn raft_cf<'a>(
     db: &'a DB,
     name: &'static str,
@@ -248,11 +280,11 @@ impl RaftStore {
 
         let mut shared = self.shared_membership.write().await;
 
-        // Start from the persisted app-level state to retain operator-set
-        // states (Draining/Leaving/Disconnected), then overlay openraft
-        // topology on top so any voter we forgot about gets re-added.
+        // Merge the persisted app-level state to retain operator-set states
+        // (Draining/Leaving/Disconnected), then overlay openraft topology on
+        // top so any voter we forgot about gets re-added.
         if !self.sm.cluster_membership.nodes.is_empty() {
-            *shared = self.sm.cluster_membership.clone();
+            merge_persisted_membership(&mut shared, self.sm.cluster_membership.clone());
         }
 
         let now = chrono::Utc::now().timestamp();
@@ -1092,7 +1124,7 @@ mod schema_drop_tests {
 mod tests {
     use std::collections::{BTreeMap, BTreeSet, HashMap};
 
-    use super::merge_snapshot_membership;
+    use super::{merge_persisted_membership, merge_snapshot_membership};
     use openraft::{BasicNode, LeaderId, LogId, Membership, StoredMembership};
     use rocksdb::{DB, Options};
     use tempfile::TempDir;
@@ -1227,6 +1259,44 @@ mod tests {
         assert_eq!(g.nodes.get(&1).unwrap().state, NodeState::Active);
     }
 
+    #[test]
+    fn merge_persisted_membership_keeps_live_sync_state_and_operator_states() {
+        let node = |id: u64, state: NodeState| NodeInfo {
+            node_id: id,
+            addr: format!("10.0.0.{id}:8086"),
+            state,
+            joined_at: 1,
+            last_heartbeat: 1,
+            needs_sync: false,
+            consecutive_misses: 0,
+        };
+        // Live: this node (2) finished its startup sync; node 4 restarted
+        // while being decommissioned.
+        let mut shared = ClusterMembership::new();
+        shared.add_node(node(2, NodeState::Active));
+        shared.add_node(node(4, NodeState::Active));
+        // Persisted: 2 as added (a learner, `Syncing`), 4 `Decommissioning`,
+        // and a node 5 this process has not seen.
+        let mut persisted = ClusterMembership::new();
+        persisted.add_node(node(2, NodeState::Syncing));
+        persisted.add_node(node(4, NodeState::Decommissioning));
+        persisted.add_node(node(5, NodeState::Draining));
+
+        merge_persisted_membership(&mut shared, persisted);
+
+        assert_eq!(
+            shared.nodes[&2].state,
+            NodeState::Active,
+            "self-reported sync state is live"
+        );
+        assert_eq!(
+            shared.nodes[&4].state,
+            NodeState::Decommissioning,
+            "operator state persists"
+        );
+        assert_eq!(shared.nodes[&5].state, NodeState::Draining);
+    }
+
     #[test]
     fn merge_snapshot_membership_preserves_self_truth() {
         let mut shared = ClusterMembership::new();
diff --git a/hyperbytedb/src/adapters/http/peer_handlers.rs b/hyperbytedb/src/adapters/http/peer_handlers.rs
index 44ad4c1..69c3c91 100644
--- a/hyperbytedb/src/adapters/http/peer_handlers.rs
+++ b/hyperbytedb/src/adapters/http/peer_handlers.rs
@@ -895,6 +895,7 @@ pub async fn handle_sync_trigger(State(state): State<Arc<AppState>>) -> impl Int
     let mv_service = state.mv_service.clone();
     let max_points_per_request = state.max_points_per_request;
     let shard_map = state.shard_map.clone();
+    let sharded = state.shard_routing.is_some();
     let cluster_auth = state.cluster_auth.clone();
     let runtime_settings = state.runtime_settings.clone();
 
@@ -926,7 +927,13 @@ pub async fn handle_sync_trigger(State(state): State<Arc<AppState>>) -> impl Int
             &cluster_auth,
         );
         sync_client = sync_client.with_runtime_settings(runtime_settings);
-        if let Some(ref map) = shard_map {
+        // Only a sharded node syncs a shard map. `state.shard_map` is a
+        // `DisabledShardMap` otherwise, never `None`: attaching it made a
+        // replicated node take the sharded reconnect path, which skips WAL
+        // catch-up and fetches a shard map its peers answer with 404. The sync
+        // then always failed, `needs_sync` stayed set, peers never readmitted
+        // the node, and writes stopped reaching it.
+        if sharded && let Some(ref map) = shard_map {
             let sm: Arc<dyn crate::ports::sharding::ShardMapPort> = map.clone();
             sync_client = sync_client.with_shard_map(sm);
         }
diff --git a/hyperbytedb/src/application/cluster/drain.rs b/hyperbytedb/src/application/cluster/drain.rs
index 9e97b76..fdb600a 100644
--- a/hyperbytedb/src/application/cluster/drain.rs
+++ b/hyperbytedb/src/application/cluster/drain.rs
@@ -256,8 +256,15 @@ impl DrainService {
         Ok(())
     }
 
+    /// Wait for every active peer to ack this node's schema mutations.
+    ///
+    /// Writes are not waited for here. Peers never record WAL acks (nothing
+    /// sets them), and a peer's WAL sequence is not comparable to this node's
+    /// anyway, so the wait on them always ran its full 90 s whenever this node
+    /// had written anything -- every restart of a replicated node. Writes are
+    /// settled at shutdown instead: each is acked by its peers, or hinted for
+    /// them (`PeerClient::wait_for_inflight`).
     async fn wait_for_replication_acks(&self) -> Result<(), HyperbytedbError> {
-        let local_wal_seq = self.wal.last_sequence().await?;
         let local_mutation_seq = self.replication_log.last_mutation_seq();
         let max_wait = Duration::from_secs(90);
         let start = std::time::Instant::now();
@@ -280,9 +287,7 @@ impl DrainService {
 
             let mut all_acked = true;
             for peer_id in &peers {
-                let wal_ack = self.replication_log.get_wal_ack(*peer_id)?;
-                let mutation_ack = self.replication_log.get_mutation_ack(*peer_id)?;
-                if wal_ack < local_wal_seq || mutation_ack < local_mutation_seq {
+                if self.replication_log.get_mutation_ack(*peer_id)? < local_mutation_seq {
                     all_acked = false;
                 }
             }
diff --git a/hyperbytedb/src/application/flush_service.rs b/hyperbytedb/src/application/flush_service.rs
index 22c2c09..fba1214 100644
--- a/hyperbytedb/src/application/flush_service.rs
+++ b/hyperbytedb/src/application/flush_service.rs
@@ -269,9 +269,12 @@ impl FlushServiceImpl {
                 Vec::new()
             }
         } else if let Some(ref m) = self.membership {
+            // Every peer still owed this node's writes, not only `Active`
+            // ones: a restarting (`Syncing`) peer catches up from this WAL,
+            // and truncating past its ack left it short for good.
             let membership = m.read().await;
             membership
-                .active_peers(self.node_id)
+                .replication_peers(self.node_id)
                 .into_iter()
                 .map(|n| n.node_id)
                 .collect()
diff --git a/hyperbytedb/src/application/peer_ingestion_service.rs b/hyperbytedb/src/application/peer_ingestion_service.rs
index ed27ee8..7a50c24 100644
--- a/hyperbytedb/src/application/peer_ingestion_service.rs
+++ b/hyperbytedb/src/application/peer_ingestion_service.rs
@@ -89,8 +89,11 @@ const FORWARD_DRAIN_CONCURRENCY: usize = 8;
 /// of seconds, and most writes fell back to synchronous forwarding.
 const FORWARD_DRAIN_MAX_ENTRIES: usize = 16;
 const FORWARD_DRAIN_MAX_BYTES: u64 = 1024 * 1024;
-/// Delivery attempts before a queued forward is dropped.
-const FORWARD_MAX_ATTEMPTS: u32 = 100;
+/// How long a queued forward is retried before it is dropped. By age, not
+/// attempts: 100 attempts at the 1 s cap was ~95 s, shorter than a restart
+/// under load, so forwards held for a restarting primary were dropped. Same
+/// as hinted handoff's default `max_hint_age_secs`.
+const FORWARD_MAX_AGE_MS: u64 = 3_600_000;
 
 /// Retry delay. Capped well under the lag limit: a retry keeps its original
 /// timestamp, so a long backoff alone would hold the queue's age over the
@@ -314,6 +317,15 @@ impl PeerIngestionService {
             &mut forward_results,
         )
         .await;
+        self.hold_failed_forwards(
+            ctx,
+            db,
+            rp,
+            precision,
+            &buckets.forward,
+            &mut forward_results,
+        )
+        .await;
         let forward_results: Vec<Result<(), HyperbytedbError>> =
             forward_results.into_iter().map(|(_, r)| r).collect();
         let mut forward_failures = 0usize;
@@ -426,36 +438,68 @@ impl PeerIngestionService {
                 .increment(1);
             return false;
         }
-        // Frames concatenate, so each region's points encode on their own.
-        let mut body = Vec::new();
-        for points in forward.values() {
-            match crate::application::line_protocol::encode_points_binary(points) {
-                Ok(frame) => body.extend_from_slice(&frame),
-                Err(e) => {
-                    tracing::warn!(error = %e, "could not encode forwards for the queue");
-                    return false;
-                }
-            }
-        }
-        let entry = ForwardEntry {
-            db: db.to_string(),
-            rp: rp.to_string(),
-            precision: precision.map(str::to_string),
-            enqueued_ms: now,
-            not_before_ms: 0,
-            attempts: 0,
-            body,
+        enqueue_forwards(queue.as_ref(), db, rp, precision, forward, now, 0).await
+    }
+
+    /// Hold the regions whose synchronous forward failed for a reason that
+    /// passes -- the primary unreachable (restarting), or at capacity after
+    /// its retries -- in the forward queue, and say which were held. The
+    /// drainer delivers them to whichever node is then the primary.
+    ///
+    /// A sharded write used to fail with 500 "write partially applied" for as
+    /// long as any primary it touched was restarting (about 60% of writes
+    /// over 6 minutes of a rolling restart), and clients resent the regions
+    /// that had landed. Only the byte cap applies here: these are not the
+    /// writes `async_forward` would queue, but ones that would fail.
+    async fn hold_failed_forwards(
+        &self,
+        ctx: &ShardRoutingContext,
+        db: &str,
+        rp: &str,
+        precision: Option<&str>,
+        forward: &std::collections::HashMap<u64, Vec<crate::domain::point::Point>>,
+        results: &mut [(u64, Result<(), HyperbytedbError>)],
+    ) {
+        let Some(queue) = self.forward_queue.as_ref() else {
+            return;
         };
-        match queue.enqueue(entry).await {
-            Ok(_) => {
-                counter!("hyperbytedb_forward_queue_enqueued_total").increment(1);
-                true
-            }
-            Err(e) => {
-                tracing::warn!(error = %e, "forward queue enqueue failed; forwarding synchronously");
-                counter!("hyperbytedb_forward_queue_sync_fallback_total", "reason" => "error")
-                    .increment(1);
-                false
+        let held: std::collections::HashMap<u64, Vec<crate::domain::point::Point>> = results
+            .iter()
+            .filter(|(_, r)| {
+                matches!(
+                    r,
+                    Err(HyperbytedbError::PeerUnreachable(_)
+                        | HyperbytedbError::ClusterUnavailable(_))
+                )
+            })
+            .filter_map(|(id, _)| forward.get(id).map(|p| (*id, p.clone())))
+            .collect();
+        if held.is_empty() {
+            return;
+        }
+        if queue.bytes() > ctx.config.async_forward_max_bytes {
+            counter!("hyperbytedb_forward_queue_hold_refused_total", "reason" => "bytes")
+                .increment(1);
+            return;
+        }
+        let now = now_ms();
+        // Not before the first backoff: the primary just failed.
+        if enqueue_forwards(
+            queue.as_ref(),
+            db,
+            rp,
+            precision,
+            &held,
+            now,
+            now + forward_backoff_ms(0),
+        )
+        .await
+        {
+            counter!("hyperbytedb_shard_forward_held_regions_total").increment(held.len() as u64);
+            for (id, r) in results.iter_mut() {
+                if held.contains_key(id) {
+                    *r = Ok(());
+                }
             }
         }
     }
@@ -1082,20 +1126,64 @@ fn partial_write_error(
 /// The entry that retries `failed` (points of `head`'s entries that did not
 /// land), or `None` once they have used up their attempts or cannot be
 /// encoded; either way they are dropped and counted.
+/// Queue `forward`'s points as one entry. False when they could not be.
+async fn enqueue_forwards(
+    queue: &dyn ForwardQueuePort,
+    db: &str,
+    rp: &str,
+    precision: Option<&str>,
+    forward: &std::collections::HashMap<u64, Vec<crate::domain::point::Point>>,
+    now: u64,
+    not_before_ms: u64,
+) -> bool {
+    // Frames concatenate, so each region's points encode on their own.
+    let mut body = Vec::new();
+    for points in forward.values() {
+        match crate::application::line_protocol::encode_points_binary(points) {
+            Ok(frame) => body.extend_from_slice(&frame),
+            Err(e) => {
+                tracing::warn!(error = %e, "could not encode forwards for the queue");
+                return false;
+            }
+        }
+    }
+    let entry = ForwardEntry {
+        db: db.to_string(),
+        rp: rp.to_string(),
+        precision: precision.map(str::to_string),
+        enqueued_ms: now,
+        not_before_ms,
+        attempts: 0,
+        body,
+    };
+    match queue.enqueue(entry).await {
+        Ok(_) => {
+            counter!("hyperbytedb_forward_queue_enqueued_total").increment(1);
+            true
+        }
+        Err(e) => {
+            tracing::warn!(error = %e, "forward queue enqueue failed");
+            counter!("hyperbytedb_forward_queue_sync_fallback_total", "reason" => "error")
+                .increment(1);
+            false
+        }
+    }
+}
+
 fn retry_forwards(
     head: &ForwardEntry,
     failed: &[crate::domain::point::Point],
     now: u64,
 ) -> Option<ForwardEntry> {
     let attempts = head.attempts + 1;
-    if attempts >= FORWARD_MAX_ATTEMPTS {
+    if now.saturating_sub(head.enqueued_ms) > FORWARD_MAX_AGE_MS {
         tracing::error!(
             points = failed.len(),
             attempts,
             db = %head.db,
-            "dropping queued forward after repeated delivery failures"
+            "dropping queued forward undelivered for an hour"
         );
-        counter!("hyperbytedb_forward_queue_dropped_total", "reason" => "attempts").increment(1);
+        counter!("hyperbytedb_forward_queue_dropped_total", "reason" => "age").increment(1);
         metrics::counter!("hyperbytedb_forward_queue_dropped_points_total")
             .increment(failed.len() as u64);
         return None;
diff --git a/hyperbytedb/src/application/replication_dispatch.rs b/hyperbytedb/src/application/replication_dispatch.rs
index 3b8b641..426eae9 100644
--- a/hyperbytedb/src/application/replication_dispatch.rs
+++ b/hyperbytedb/src/application/replication_dispatch.rs
@@ -18,7 +18,7 @@ pub async fn dispatch_outbound_replication(
             let peer_count = if let Some(ref targets) = batch.target_node_ids {
                 targets.len()
             } else {
-                replication.active_peer_count(node_id).await
+                replication.replication_peer_count(node_id).await
             };
             let min_acks: SyncQuorumMinAcks = replication_config.sync_quorum.min_acks;
             let required = min_acks.resolve(peer_count);
diff --git a/hyperbytedb/src/domain/cluster/membership.rs b/hyperbytedb/src/domain/cluster/membership.rs
index 039a5e0..e8fa384 100644
--- a/hyperbytedb/src/domain/cluster/membership.rs
+++ b/hyperbytedb/src/domain/cluster/membership.rs
@@ -195,15 +195,21 @@ impl ClusterMembership {
             .collect()
     }
 
-    /// Peers that should receive outbound write replication. Includes
-    /// `Disconnected` nodes so writes are attempted (and queued in hinted
-    /// handoff on failure) during rolling restarts instead of being skipped.
+    /// Peers that should receive outbound write replication: every member
+    /// that is not leaving the cluster. A write is attempted on each, and
+    /// queued in hinted handoff when it cannot be delivered.
+    ///
+    /// A restarting node passes through `Draining`, then `Syncing` (or
+    /// `Joining`); writes made while it was in either used to be neither sent
+    /// nor hinted, so they never reached it, and its copy fell behind for
+    /// good. Over one rolling restart of three nodes that left the replicas
+    /// 643, 1,825 and 1,051 of 3,204 acknowledged writes short.
     pub fn replication_peers(&self, exclude_id: u64) -> Vec<&NodeInfo> {
         self.nodes
             .values()
             .filter(|n| {
                 n.node_id != exclude_id
-                    && matches!(n.state, NodeState::Active | NodeState::Disconnected)
+                    && !matches!(n.state, NodeState::Decommissioning | NodeState::Leaving)
             })
             .collect()
     }
@@ -321,13 +327,22 @@ mod tests {
         assert_eq!(peers.len(), 1);
         assert_eq!(peers[0].node_id, 2);
 
+        // Every member not leaving: a draining (restarting) node too.
         let repl = m.replication_peers(1);
-        assert_eq!(repl.len(), 2);
+        assert_eq!(repl.len(), 3);
         assert!(repl.iter().any(|n| n.node_id == 2));
         assert!(repl.iter().any(|n| n.node_id == 3));
+        assert!(repl.iter().any(|n| n.node_id == 4));
+        m.add_node(make_node(5, NodeState::Syncing));
+        m.add_node(make_node(6, NodeState::Joining));
+        m.add_node(make_node(7, NodeState::Decommissioning));
+        m.add_node(make_node(8, NodeState::Leaving));
+        let ids: std::collections::BTreeSet<u64> =
+            m.replication_peers(1).iter().map(|n| n.node_id).collect();
+        assert_eq!(ids, [2, 3, 4, 5, 6].into(), "all but departing members");
 
         let all = m.all_peers(1);
-        assert_eq!(all.len(), 3);
+        assert_eq!(all.len(), 7);
     }
 
     #[test]
diff --git a/hyperbytedb/src/ports/replication.rs b/hyperbytedb/src/ports/replication.rs
index 2582b56..896d77c 100644
--- a/hyperbytedb/src/ports/replication.rs
+++ b/hyperbytedb/src/ports/replication.rs
@@ -47,5 +47,7 @@ pub trait ReplicationPort: Send + Sync {
         req: MutationRequest,
     ) -> Result<(), HyperbytedbError>;
 
-    async fn active_peer_count(&self, self_node_id: u64) -> usize;
+    /// Peers a write replicates to (members not leaving), which a
+    /// `sync_quorum` write counts its required acks against.
+    async fn replication_peer_count(&self, self_node_id: u64) -> usize;
 }
diff --git a/hyperbytedb/src/server.rs b/hyperbytedb/src/server.rs
index 9fe609c..b2eb24c 100644
--- a/hyperbytedb/src/server.rs
+++ b/hyperbytedb/src/server.rs
@@ -571,6 +571,18 @@ async fn serve_inner(
         }
     }
 
+    // Writes acknowledged before the API stopped may still be on their way
+    // to peers (a `sync_quorum` write returns once enough peers acked);
+    // let them land, or become hints, before the process exits.
+    if let Some(ref pc) = peer_client {
+        let pending = pc.wait_for_inflight(Duration::from_secs(10)).await;
+        if pending > 0 {
+            tracing::warn!(pending, "replication sends still pending at shutdown");
+        } else {
+            tracing::info!("replication sends settled");
+        }
+    }
+
     // ── Phase 3: Stop remaining background services ─────────────────────
     tracing::info!("stopping remaining background services");
 
diff --git a/hyperbytedb/tests/sharding_cluster_integration.rs b/hyperbytedb/tests/sharding_cluster_integration.rs
index 256552e..3a658d3 100644
--- a/hyperbytedb/tests/sharding_cluster_integration.rs
+++ b/hyperbytedb/tests/sharding_cluster_integration.rs
@@ -2445,7 +2445,8 @@ async fn a_queued_forward_is_acknowledged_and_reaches_its_primary() {
 
 /// With the primary down the write is still acknowledged and stays queued;
 /// once the queue is older than the lag limit, new writes forward
-/// synchronously again and report the failure.
+/// synchronously again -- and, the primary being down, are held in the
+/// queue too rather than failed (they used to report the failure, 500).
 #[tokio::test]
 #[serial(chdb)]
 async fn a_lagging_queue_sends_new_writes_back_to_synchronous_forwarding() {
@@ -2461,14 +2462,12 @@ async fn a_lagging_queue_sends_new_writes_back_to_synchronous_forwarding() {
 
     tokio::time::sleep(std::time::Duration::from_millis(400)).await;
     let second = write_line(&client, &n1.url, "sharddb", "mem,host=b used=2 1000000000").await;
-    assert!(
-        !second.status().is_success(),
-        "past the lag limit a write forwards synchronously and sees the failure"
-    );
-    assert!(
-        !queue.is_empty(),
-        "the first write is still owed, not dropped"
+    assert_eq!(
+        second.status(),
+        204,
+        "past the lag limit a write forwards synchronously, fails, and is held"
     );
+    assert_eq!(queue.len(), 2, "both writes are owed, none dropped");
 }
 
 /// A queued forward is routed against the shard map at delivery: when its
@@ -2501,6 +2500,59 @@ async fn a_queued_forward_follows_its_region_to_a_new_primary() {
     assert!(queue.is_empty());
 }
 
+/// A write whose region's primary is down (restarting) is held in the
+/// coordinator's forward queue and acknowledged, not failed with 500 "write
+/// partially applied" -- which during a rolling restart was about 60% of
+/// writes for 6 minutes. Synchronous forwarding (the default) is on here.
+#[tokio::test]
+#[serial(chdb)]
+async fn a_write_for_a_down_primary_is_held_not_failed() {
+    let dir = tempfile::tempdir().unwrap();
+    let opts = ShardedClusterOptions {
+        forward_queue: true,
+        ..Default::default()
+    };
+    assert!(
+        !opts.sharding.async_forward,
+        "the default: forwards are synchronous"
+    );
+    let [n1, mut n2] = start_sharded_pair_cluster(dir.path(), opts).await;
+    let client = reqwest::Client::new();
+    for node in [&n1, &n2] {
+        create_db(&client, &node.url, "hdb").await;
+    }
+    // Only node 2 holds the measurement, so node 1 forwards every point.
+    let op = ShardMapOp::BootstrapMeasurement {
+        key: MeasurementKey::new("hdb", "autogen", "mem"),
+        region: ShardRegion {
+            region_id: 2,
+            start: 0,
+            end: u64::MAX,
+            epoch: ShardEpoch::default(),
+            peers: vec![2],
+            primary: 2,
+            last_split_at: 0,
+            transfer_verified: None,
+            transfer_first_seen: None,
+            state: Default::default(),
+            staged_from: None,
+            staging_peers: Vec::new(),
+        },
+    };
+    for node in [&n1, &n2] {
+        node.shard_map.apply_op(op.clone()).await.unwrap();
+        let snap = node.shard_map.snapshot().await.unwrap();
+        node.location_cache.refresh_from_map(&snap);
+    }
+    stop_node(&mut n2);
+    let resp = write_line(&client, &n1.url, "hdb", "mem,host=a used=1 1000000000").await;
+    let status = resp.status();
+    let body = resp.text().await.unwrap_or_default();
+    assert_eq!(status, 204, "{body}");
+    let queue = n1.forward_queue.as_ref().unwrap();
+    assert_eq!(queue.len(), 1, "held for the primary until it is back");
+}
+
 /// Two nodes; only node 2 holds `qdb.mem`, so node 1 knows no field types
 /// and only the primary can find a conflict.
 async fn coordinator_without_types(dir: &std::path::Path, queue: bool) -> [ShardedTestNode; 2] {
diff --git a/hyperbytedb/tests/sync_quorum_integration.rs b/hyperbytedb/tests/sync_quorum_integration.rs
index 084b7f8..f5a45f8 100644
--- a/hyperbytedb/tests/sync_quorum_integration.rs
+++ b/hyperbytedb/tests/sync_quorum_integration.rs
@@ -51,6 +51,7 @@ struct TestNode {
     #[allow(dead_code)]
     node_id: u64,
     wal: Arc<RocksDbWal>,
+    membership: hyperbytedb::domain::cluster::membership::SharedMembership,
     _handle: tokio::task::JoinHandle<()>,
 }
 
@@ -118,6 +119,7 @@ async fn start_node_on(
         });
     }
     let shared_membership = new_shared(membership);
+    let membership_handle = shared_membership.clone();
 
     let peer_client = Arc::new(PeerClient::new(
         node_id,
@@ -227,6 +229,7 @@ async fn start_node_on(
         addr,
         node_id,
         wal,
+        membership: membership_handle,
         _handle: handle,
     }
 }
@@ -425,6 +428,51 @@ async fn write_one(client: &reqwest::Client, url: &str, db: &str, line: &str) ->
         .unwrap()
 }
 
+/// Writes reach a peer that this node has as `Syncing` (or `Draining`): a
+/// restarting peer passes through both. They were neither sent nor hinted,
+/// so the peer never got them, and over one rolling restart of three nodes
+/// the replicas fell 643, 1,825 and 1,051 of 3,204 acknowledged writes short.
+#[tokio::test]
+#[serial(chdb)]
+async fn writes_reach_a_peer_that_is_syncing_or_draining() {
+    let dir = tempfile::tempdir().unwrap();
+    let nodes =
+        start_three_node_cluster(dir.path(), [sync_cfg(1, 5000), async_cfg(), async_cfg()]).await;
+    let client = reqwest::Client::new();
+    create_db(&client, &nodes[0].url, "rdb").await;
+    for node in &nodes[1..] {
+        wait_for_database(&client, &node.url, "rdb").await;
+    }
+    {
+        let mut m = nodes[0].membership.write().await;
+        m.set_state(2, NodeState::Syncing);
+        m.set_state(3, NodeState::Draining);
+    }
+    let resp = write_one(
+        &client,
+        &nodes[0].url,
+        "rdb",
+        "cpu,host=a value=1.0 1000000000",
+    )
+    .await;
+    assert_eq!(resp.status(), StatusCode::NO_CONTENT);
+    for (i, node) in nodes[1..].iter().enumerate() {
+        let mut got = 0;
+        for _ in 0..100 {
+            got = node.wal.last_sequence().await.unwrap();
+            if got >= 1 {
+                break;
+            }
+            tokio::time::sleep(Duration::from_millis(20)).await;
+        }
+        assert!(
+            got >= 1,
+            "peer {} must have the write (WAL seq {got})",
+            i + 2
+        );
+    }
+}
+
 #[tokio::test]
 #[serial(chdb)]
 async fn sync_quorum_blocks_until_peer_acks_and_converges() {

From 1a598889b1eb14a4ce52b6e474205149ddaae9a1 Mon Sep 17 00:00:00 2001
From: Austin Barrington <austin@hyperbyte.dev>
Date: Tue, 20 Oct 2026 09:42:15 +0100
Subject: [PATCH 2/2] Hint failing sends at once when shutdown waits on them

Shutdown waited 10 s for replication sends, but a send to a down peer
backs off 1 s, 2 s, 4 s and 8 s before it writes the hint. The process
could exit first, with the acknowledged write neither delivered nor
hinted.

wait_for_inflight now signals the sends: a failed send hints at once
instead of backing off. The wait is 15 s, one request (10 s timeout)
plus the hint.
---
 .../src/adapters/cluster/peer_client.rs       | 125 ++++++++++++++----
 hyperbytedb/src/server.rs                     |   6 +-
 2 files changed, 105 insertions(+), 26 deletions(-)

diff --git a/hyperbytedb/src/adapters/cluster/peer_client.rs b/hyperbytedb/src/adapters/cluster/peer_client.rs
index b253bcb..69bdc4c 100644
--- a/hyperbytedb/src/adapters/cluster/peer_client.rs
+++ b/hyperbytedb/src/adapters/cluster/peer_client.rs
@@ -52,6 +52,10 @@ pub struct PeerClient {
 struct InFlight {
     count: AtomicUsize,
     idle: tokio::sync::Notify,
+    /// Set once shutdown waits on the sends: a send that fails then hints
+    /// at once instead of backing off (1 s, 2 s, 4 s, 8 s) past the wait.
+    stopping: AtomicBool,
+    stop: tokio::sync::Notify,
 }
 
 /// Holds `n` units of [`InFlight`] until dropped.
@@ -61,6 +65,20 @@ struct InFlightGuard {
 }
 
 impl InFlight {
+    /// Sleep `delay` before the next retry. False, at once, when shutdown
+    /// has begun or begins meanwhile: hint now rather than retry.
+    async fn backoff(&self, delay: Duration) -> bool {
+        // Created before the check, so a stop between the two still wakes it.
+        let stop = self.stop.notified();
+        if self.stopping.load(Ordering::SeqCst) {
+            return false;
+        }
+        tokio::select! {
+            () = tokio::time::sleep(delay) => true,
+            () = stop => false,
+        }
+    }
+
     fn track(self: &Arc<Self>, n: usize) -> InFlightGuard {
         self.count.fetch_add(n, Ordering::SeqCst);
         InFlightGuard {
@@ -379,10 +397,9 @@ impl PeerClient {
                         .increment(1);
                 }
             }
-            if attempts >= self.max_retries {
+            if attempts >= self.max_retries || !self.inflight.backoff(delay).await {
                 break;
             }
-            tokio::time::sleep(delay).await;
             delay = (delay * 2).min(Duration::from_secs(30));
         }
 
@@ -448,8 +465,12 @@ impl PeerClient {
     /// Wait, up to `timeout`, for every queued or in-flight send to a peer to
     /// finish: delivered, or hinted. Returns how many were still pending.
     /// Shutdown calls it so writes this node acknowledged reach every peer
-    /// it owes them to, or that peer's hints.
+    /// it owes them to, or that peer's hints. From here on a failed send
+    /// hints at once instead of backing off, so `timeout` need only cover
+    /// one request.
     pub async fn wait_for_inflight(&self, timeout: Duration) -> usize {
+        self.inflight.stopping.store(true, Ordering::SeqCst);
+        self.inflight.stop.notify_waiters();
         let deadline = tokio::time::Instant::now() + timeout;
         loop {
             let idle = self.inflight.idle.notified();
@@ -745,6 +766,7 @@ impl PeerClient {
             sync,
             region_id,
             max_attempts: self.max_retries,
+            inflight: Arc::clone(&self.inflight),
         }
     }
 
@@ -1115,6 +1137,7 @@ struct PeerReplicateCtx {
     /// on-disk hint record predates the region being carried.
     region_id: Option<u64>,
     max_attempts: u32,
+    inflight: Arc<InFlight>,
 }
 
 impl PeerReplicateCtx {
@@ -1210,31 +1233,31 @@ impl PeerReplicateCtx {
                 }
             }
 
-            if attempts >= self.max_attempts {
-                if let Some(ref hh) = self.hinted_handoff {
-                    if let Err(e) = hh.enqueue_hint(self.peer_id, &self.hint_payload) {
-                        tracing::error!(
-                            peer_id = self.peer_id,
-                            error = %e,
-                            "failed to enqueue hinted handoff"
-                        );
-                    } else {
-                        tracing::warn!(
-                            peer_id = self.peer_id,
-                            "replication failed, write queued for hinted handoff"
-                        );
-                    }
-                } else {
+            if attempts < self.max_attempts && self.inflight.backoff(delay).await {
+                delay = (delay * 2).min(Duration::from_secs(30));
+                continue;
+            }
+            if let Some(ref hh) = self.hinted_handoff {
+                if let Err(e) = hh.enqueue_hint(self.peer_id, &self.hint_payload) {
                     tracing::error!(
-                        peer = %self.url,
-                        "giving up replication after {} attempts (no hinted handoff)",
-                        self.max_attempts
+                        peer_id = self.peer_id,
+                        error = %e,
+                        "failed to enqueue hinted handoff"
+                    );
+                } else {
+                    tracing::warn!(
+                        peer_id = self.peer_id,
+                        "replication failed, write queued for hinted handoff"
                     );
                 }
-                return Err(last_error.unwrap_or_else(|| "unknown error".to_string()));
+            } else {
+                tracing::error!(
+                    peer = %self.url,
+                    "giving up replication after {} attempts (no hinted handoff)",
+                    attempts
+                );
             }
-            tokio::time::sleep(delay).await;
-            delay = (delay * 2).min(Duration::from_secs(30));
+            return Err(last_error.unwrap_or_else(|| "unknown error".to_string()));
         }
     }
 }
@@ -1539,6 +1562,60 @@ mod replicate_batch_tests {
         assert_eq!(hh.pending_count(2).unwrap(), 0, "not a replica of region 2");
         assert_eq!(hh.pending_count(4).unwrap(), 0, "not a replica of region 2");
     }
+
+    /// Shutdown waits on sends to a down peer. Backing off 1 s, 2 s, 4 s,
+    /// 8 s before the hint outlasted that wait, and the process exited with
+    /// the acknowledged write neither delivered nor hinted. Node 2 refuses
+    /// connections (the batch path); node 3 predates replicate-batch and
+    /// answers 503 (the per-part path).
+    #[tokio::test]
+    async fn shutdown_hints_a_failing_send_instead_of_backing_off() {
+        let legacy_failing = serve(
+            axum::Router::new()
+                .route(
+                    "/internal/replicate-batch",
+                    axum::routing::post(|| async { axum::http::StatusCode::NOT_FOUND }),
+                )
+                .route(
+                    "/internal/replicate",
+                    axum::routing::post(|| async { axum::http::StatusCode::SERVICE_UNAVAILABLE }),
+                ),
+        )
+        .await;
+        let mut m = ClusterMembership::new();
+        m.add_node(node(2, "127.0.0.1:1"));
+        m.add_node(node(3, &legacy_failing));
+        let dir = tempfile::tempdir().unwrap();
+        let log = Arc::new(ReplicationLog::open(dir.path()).unwrap());
+        let hh = Arc::new(HintedHandoff::new(log.db().clone(), 100).unwrap());
+        let pc = Arc::new(
+            PeerClient::new(
+                1,
+                "127.0.0.1:1".into(),
+                crate::domain::cluster::membership::new_shared(m),
+                log,
+                5,
+                64,
+                2,
+                1 << 20,
+                Default::default(),
+            )
+            .with_hinted_handoff(hh.clone()),
+        );
+        PeerClient::replicate_write(&pc, job(1, vec![2, 3])).unwrap();
+        // Both first attempts fail; both sends are now backing off.
+        tokio::time::sleep(Duration::from_millis(300)).await;
+
+        let started = std::time::Instant::now();
+        assert_eq!(pc.wait_for_inflight(Duration::from_secs(5)).await, 0);
+        assert!(
+            started.elapsed() < Duration::from_secs(2),
+            "{:?}",
+            started.elapsed()
+        );
+        assert_eq!(hh.pending_count(2).unwrap(), 1);
+        assert_eq!(hh.pending_count(3).unwrap(), 1);
+    }
 }
 
 #[cfg(test)]
diff --git a/hyperbytedb/src/server.rs b/hyperbytedb/src/server.rs
index b2eb24c..9c936c7 100644
--- a/hyperbytedb/src/server.rs
+++ b/hyperbytedb/src/server.rs
@@ -573,9 +573,11 @@ async fn serve_inner(
 
     // Writes acknowledged before the API stopped may still be on their way
     // to peers (a `sync_quorum` write returns once enough peers acked);
-    // let them land, or become hints, before the process exits.
+    // let them land, or become hints, before the process exits. Retries
+    // stop backing off now, so this covers one request (10 s timeout) and
+    // the hint.
     if let Some(ref pc) = peer_client {
-        let pending = pc.wait_for_inflight(Duration::from_secs(10)).await;
+        let pending = pc.wait_for_inflight(Duration::from_secs(15)).await;
         if pending > 0 {
             tracing::warn!(pending, "replication sends still pending at shutdown");
         } else {