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..69bdc4c 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,60 @@ 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,
+    /// 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.
+struct InFlightGuard {
+    inflight: Arc<InFlight>,
+    n: usize,
+}
+
+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 {
+            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 +152,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 +232,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 +246,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 +260,7 @@ impl PeerClient {
             tokio::spawn(async move {
                 this.replicate_round(jobs).await;
                 drop(permit);
+                drop(queued);
             });
         }
     }
@@ -335,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));
         }
 
@@ -401,6 +462,28 @@ 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. 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();
+            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 +504,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 +531,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 +665,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);
             });
         }
 
@@ -666,6 +766,7 @@ impl PeerClient {
             sync,
             region_id,
             max_attempts: self.max_retries,
+            inflight: Arc::clone(&self.inflight),
         }
     }
 
@@ -915,9 +1016,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()
     }
 }
 
@@ -1032,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 {
@@ -1127,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()));
         }
     }
 }
@@ -1456,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/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..9c936c7 100644
--- a/hyperbytedb/src/server.rs
+++ b/hyperbytedb/src/server.rs
@@ -571,6 +571,20 @@ 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. 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(15)).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() {