fix: attest multi-pool bootstrap per creator and contain lock RPC storms (#7473)

* fix(ecstore): attest fresh multi-pool bootstrap per pool creator

A fresh deployment whose pools have their first endpoint on different nodes could never publish its initial pool.bin: each node held fresh-bootstrap proof only for the pool it formatted, combine_across_pools collapsed the deployment-wide proof to None, the elected writer never wrote a pending identity, and startup died with "no durable bootstrap identity or pool.bin replica is available" once the init retry budget ran out.

Track first-hand bootstrap authority per pool. The first pool's creator mints the pending cluster identity on the pool it created, every other creator copies that nonce-bound identity onto the pool it formatted first-hand (a scoped identity write that only ever touches pools the process holds first-hand proof for), and the elected writer publishes pool.bin once it holds first-hand proof for pool 0 and every pool replica carries the same pending identity. Fresh + None is still never promoted, corrupt or disagreeing replicas still fail closed, an initialized deployment never reopens bootstrap for an expansion pool, and an elected restart without first-hand proof still cannot reuse a complete pending set.

Startup classification no longer latches the pool-metadata write gate for the two transient outcomes a healthy bootstrap passes through (a non-elected node waiting for the elected writer, the elected writer waiting for the other creators); recover_pool_meta_transaction never clears write_blocked, so a non-elected node that started before pool.bin existed stayed write-blocked for the life of the process. Genuine recovery-required states still latch.

Refs rustfs/backlog#2375, rustfs/backlog#2338

* fix(lock): contain remote lock RPC timeout storms

A lock RPC deadline evicted the shared internode HTTP/2 channel and re-dialed it unconditionally, so one slow lock endpoint produced a cluster-wide RST_STREAM / GOAWAY too_many_resets / reconnect loop (rustfs#7363).

The remote lock client now keeps a per-peer channel history: a timeout evicts only when the peer has completed no lock RPC for two deadlines, evictions and transport-failure re-dials are rate limited per peer (RUSTFS_OBJECT_LOCK_RPC_EVICTION_COOLDOWN_MS, default 5 s), and a timed-out request is detached instead of cancelled, bounded per peer by RUSTFS_OBJECT_LOCK_RPC_DETACHED_LIMIT (default 256) and by the internode RPC timeout; a lock granted after its caller gave up is released immediately. Unlocks that fail the quick retries continue on a deferred 1/2/4/8/16 s schedule before the server lease reclaims them. New rustfs_remote_lock_* metrics cover timeouts, evictions, suppressed evictions, detached streams, late completions and late releases per peer; docs/operations/lock-rpc-storm-protection.md documents the policy.

Refs rustfs/backlog#2375, rustfs#7363

* ci: refresh nightly test selection digests

The replication nightly membership guard expected the 68-test digest from #7422 while the current listing has 71 tests (additions only: test_bucket_replication_sse_c_compressed_passthrough from #7366, matrix_mint_own_version_ids_addresses_mutations_through_the_ledger and matrix_removed_replication_config_abandons_pending_purge from #7368), and the cluster fault lane expected 50 tests while #7374 added test_cluster_root_heal_recovers_remote_shards_after_background_target_crash. Both lanes have failed before running a single test since 2026-09-07. Bind the Linux digests to the listings from scheduled run 34187469350 and the Darwin e2e-nightly digest to the matching local listing.

Refs rustfs/backlog#2375

.config/e2e-nightly-selection.txt

@@ -1,2 +1,2 @@
-sha256-darwin=a5665318c9bdc0947514fb7008ba1b83b114b739fac775c3c446f207058b7c7a
-sha256-linux=45d80e1723de5d25bb5b81f3ef5c82f583efc3e4f036a8cd2bb99e4f1eca9e51
+sha256-darwin=364f2329a7b72eb9f1608dbe1a3af37af4095354014f3cbe23ca448492d89961
+sha256-linux=60983f1ebe7068cf660d473c5f76c76a650410ccc99d71934ddca7fd67607987

.config/e2e-repl-nightly-selection.txt

@@ -1 +1 @@
-sha256=0fe8408874ccec3620262a9812d67920ddd72dc9edf0e36e0d0aed3f8bad026e
+sha256=0e338d305260229e17ccfb2adc48a6212dbdfea36a9ebfb5a4e0d38658e6cc45

CHANGELOG.md

@@ -11,6 +11,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - **Presigned URLs honour only signed headers** (GHSA-g8w9-qw9q-fghr): a SigV4 presigned request that carries an `x-amz-*` request header not listed in `X-Amz-SignedHeaders` is now rejected with `403 AccessDenied` ("There were headers present in the request which were not signed"), matching AWS S3. Previously the holder of a presigned `PutObject` URL could add unsigned `x-amz-tagging`, `x-amz-storage-class`, `x-amz-website-redirect-location`, ACL, metadata, Object Lock or SSE headers and have them applied. Presigners that intend a property must set it before signing so the SDK lists the header in `SignedHeaders`; `x-amz-cf-id` (CloudFront) remains tolerated unsigned. Header-signed SigV4 and SigV2 requests are unchanged.
 
 ### Fixed
+- **Fresh multi-pool bootstrap with distinct format creators**: a new deployment whose pools have their first endpoint on different nodes (for example two single-node pools) could never publish its initial `pool.bin`: each node held fresh-bootstrap proof only for the pool it formatted, the deployment-wide proof collapsed to none, and every node died with `pool metadata recovery required: no durable bootstrap identity or pool.bin replica is available` after the startup retry budget. The first pool's creator now mints the pending cluster identity on its own pool, every other creator copies that nonce-bound identity onto the pool it formatted first-hand, and the elected writer publishes `pool.bin` once every pool replica carries the same pending identity. Corrupt or disagreeing replicas, pools that merely have a format, expansion pools joining an initialized deployment, and restarts without first-hand proof still fail closed. Non-elected nodes that start before `pool.bin` exists, and the elected writer while it waits for the other creators, no longer latch their pool-metadata write gate for the life of the process. Refs rustfs/backlog#2338, rustfs/backlog#2375.
+- **Lock RPC timeout storms** (#7363): the remote lock client no longer evicts and re-dials the shared internode HTTP/2 channel on every request deadline. A timeout evicts only when the peer has not completed any lock RPC for two deadlines, evictions and transport-failure re-dials are rate limited per peer (`RUSTFS_OBJECT_LOCK_RPC_EVICTION_COOLDOWN_MS`, default 5 s), and a timed-out request is left running instead of being reset (bounded per peer by `RUSTFS_OBJECT_LOCK_RPC_DETACHED_LIMIT`, default 256), so a slow lock endpoint can no longer drive the `RST_STREAM`/`GOAWAY too_many_resets`/reconnect loop. A lock granted after its caller timed out is released immediately, and unlocks that fail the quick retries continue on a deferred 1/2/4/8/16 s schedule before the server lease reclaims them. New `rustfs_remote_lock_*` metrics cover timeouts, evictions, suppressed evictions, detached streams, late completions and late releases per peer. Operator guide at `docs/operations/lock-rpc-storm-protection.md`.
 - **Multipart admission queue**: an `UploadPart` waiting for a foreground write permit now waits at most 10 s by default (`RUSTFS_PUT_MULTIPART_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS`, previously 30 s), so a queued part returns S3 `SlowDown` before the client's socket write timeout drops the connection. Separately, the API listener no longer forces a 4 MiB `SO_RCVBUF` on every accepted socket (kernel autotuning applies; `RUSTFS_HTTP_SOCKET_RECV_BUFFER_BYTES` restores a fixed size), so a queued part no longer lets up to 8 MiB of unread body accumulate in kernel memory per connection, which is what throttled whole nodes under SDK-default multipart concurrency. Fixes #7385.
 - **Helm Ingress**: `customAnnotations` are now merged with class-specific annotations (nginx/traefik) instead of being ignored when `ingress.className` is set.
 - **Per-pool erasure parity**: Erasure parity (STANDARD and reduced-redundancy) is now resolved independently for every pool instead of reusing the first pool's value. A heterogeneous topology — for example a 4-drive pool plus a 2-drive pool created during expansion — previously inherited the first pool's parity and could resolve to zero data shards in the smaller pool, panicking Reed-Solomon construction on write. Automatic parity now resolves per pool (for example `2+2` in the 4-drive pool and `1+1` in the 2-drive pool). Fixes #4801.

crates/config/src/constants/object.rs

@@ -609,6 +609,36 @@ pub const ENV_OBJECT_LOCK_RPC_TIMEOUT_MS: &str = "RUSTFS_OBJECT_LOCK_RPC_TIMEOUT
 /// Default remote lock RPC transport timeout: 3000 milliseconds.
 pub const DEFAULT_OBJECT_LOCK_RPC_TIMEOUT_MS: u64 = 3000;
 
+/// Environment variable for the minimum interval between evictions of the
+/// cached lock RPC channel to one peer, in milliseconds.
+///
+/// A lock RPC that fails on transport, or that times out while the peer has
+/// not completed any lock RPC for two deadlines, evicts the shared HTTP/2
+/// channel so the next request re-dials. Evictions are rate limited per peer
+/// so one slow lock endpoint cannot drive a reset/GOAWAY/reconnect loop
+/// (issue #7363). `0` disables the cooldown.
+///
+/// Default: 5000 milliseconds.
+pub const ENV_OBJECT_LOCK_RPC_EVICTION_COOLDOWN_MS: &str = "RUSTFS_OBJECT_LOCK_RPC_EVICTION_COOLDOWN_MS";
+
+/// Default minimum interval between lock RPC channel evictions per peer: 5000 milliseconds.
+pub const DEFAULT_OBJECT_LOCK_RPC_EVICTION_COOLDOWN_MS: u64 = 5000;
+
+/// Environment variable for how many timed-out lock RPCs per peer may keep
+/// running in the background instead of being cancelled.
+///
+/// Cancelling a timed-out stream sends `RST_STREAM`; enough of them make the
+/// peer answer `GOAWAY too_many_resets` and drop every stream on the
+/// connection. A detached RPC ends on its own within the internode RPC
+/// timeout, and a lock it acquires after its caller gave up is released
+/// immediately. Beyond this budget timed-out RPCs are cancelled as before.
+///
+/// Default: 256.
+pub const ENV_OBJECT_LOCK_RPC_DETACHED_LIMIT: &str = "RUSTFS_OBJECT_LOCK_RPC_DETACHED_LIMIT";
+
+/// Default per-peer budget of detached (timed-out but still running) lock RPCs: 256.
+pub const DEFAULT_OBJECT_LOCK_RPC_DETACHED_LIMIT: usize = 256;
+
 /// Environment variable to enable object namespace lock diagnostics.
 ///
 /// When enabled, RustFS emits slow lock acquisition and long lock hold

crates/ecstore/src/cluster/rpc/remote_locker.rs

@@ -22,21 +22,152 @@ use rustfs_lock::{
     LockClient, LockError, LockInfo, LockRequest, LockResponse, LockStats, LockStatus, LockType, Result,
     types::{LockId, LockMetadata, LockPriority},
 };
-use rustfs_protos::proto_gen::node_service::{BatchGenerallyLockRequest, GenerallyLockRequest, PingRequest};
+use rustfs_protos::proto_gen::node_service::{
+    BatchGenerallyLockRequest, BatchGenerallyLockResponse, GenerallyLockRequest, GenerallyLockResponse, GenerallyLockResult,
+    PingRequest,
+};
 use rustfs_protos::{
     ConnectionEvictionLogLevel, evict_failed_connection_with_log_level, models::PingBodyBuilder,
     proto_gen::node_service::node_service_client::NodeServiceClient,
 };
-use std::{sync::OnceLock, time::Duration};
-use tokio::time::timeout;
-use tonic::Request;
+use std::collections::HashMap;
+use std::future::Future;
+use std::pin::Pin;
+use std::sync::{Mutex, OnceLock};
+use std::time::Duration;
+use tokio::task::JoinHandle;
+use tokio::time::{Instant, timeout};
 use tonic::service::interceptor::InterceptedService;
+use tonic::{Request, Response};
 use tracing::{debug, info, warn};
 
 fn attach_lock_mutation_body_digest<T: rustfs_protos::CanonicalMutationBody>(request: &mut Request<T>) -> std::io::Result<()> {
     set_tonic_rolling_mutation_body_digest(request)
 }
 
+/// Work to run if an RPC that already timed out for its caller completes later.
+type LateCompletion<T> = Option<Box<dyn FnOnce(T) -> Pin<Box<dyn Future<Output = ()> + Send>> + Send>>;
+
+/// The liveness window is this many RPC deadlines: a peer that completed a
+/// lock RPC within it is slow, not gone, and keeps its channel on a timeout.
+const LOCK_RPC_LIVENESS_WINDOW_DEADLINES: u32 = 2;
+
+/// Recent history of the shared lock channel to one peer (issue #7363).
+///
+/// A single request deadline says nothing about the HTTP/2 connection it ran
+/// on: a peer whose lock service is merely slow keeps answering other streams.
+/// Evicting the cached channel on every timeout turned that slowness into a
+/// `RST_STREAM`/`GOAWAY too_many_resets`/re-dial loop across the cluster, so
+/// eviction now requires the peer to have gone quiet and is rate limited.
+#[derive(Debug, Clone, Copy, Default)]
+struct LockPeerChannelHealth {
+    last_success: Option<Instant>,
+    last_eviction: Option<Instant>,
+    consecutive_timeouts: u32,
+    /// Timed-out RPCs still running in the background for this peer.
+    detached_rpcs: usize,
+}
+
+fn lock_peer_channel_health() -> &'static Mutex<HashMap<String, LockPeerChannelHealth>> {
+    static HEALTH: OnceLock<Mutex<HashMap<String, LockPeerChannelHealth>>> = OnceLock::new();
+    HEALTH.get_or_init(Mutex::default)
+}
+
+fn with_lock_peer_health<R>(addr: &str, update: impl FnOnce(&mut LockPeerChannelHealth) -> R) -> R {
+    let mut peers = lock_peer_channel_health()
+        .lock()
+        .unwrap_or_else(std::sync::PoisonError::into_inner);
+    update(peers.entry(addr.to_string()).or_default())
+}
+
+#[cfg(test)]
+fn lock_peer_health_for_test(addr: &str) -> LockPeerChannelHealth {
+    lock_peer_channel_health()
+        .lock()
+        .unwrap_or_else(std::sync::PoisonError::into_inner)
+        .get(addr)
+        .copied()
+        .unwrap_or_default()
+}
+
+#[cfg(test)]
+fn reset_lock_peer_health_for_test(addr: &str) {
+    lock_peer_channel_health()
+        .lock()
+        .unwrap_or_else(std::sync::PoisonError::into_inner)
+        .remove(addr);
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum EvictionTrigger {
+    /// The caller's deadline expired while the stream was still open.
+    Timeout,
+    /// The transport itself reported the failure (refused, reset, GOAWAY, ...).
+    Transport,
+}
+
+impl EvictionTrigger {
+    fn as_str(self) -> &'static str {
+        match self {
+            Self::Timeout => "timeout",
+            Self::Transport => "transport",
+        }
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum EvictionVerdict {
+    Evict,
+    /// The peer completed a lock RPC within the liveness window: slow, not gone.
+    PeerRecentlyServed,
+    /// The channel was re-dialed within the cooldown; let it prove itself first.
+    CoolingDown,
+}
+
+impl EvictionVerdict {
+    fn as_str(self) -> &'static str {
+        match self {
+            Self::Evict => "evict",
+            Self::PeerRecentlyServed => "peer_recently_served",
+            Self::CoolingDown => "cooling_down",
+        }
+    }
+}
+
+/// Decide whether a failed lock RPC may evict the shared channel to its peer.
+fn eviction_verdict(
+    health: &LockPeerChannelHealth,
+    now: Instant,
+    trigger: EvictionTrigger,
+    liveness_window: Duration,
+    cooldown: Duration,
+) -> EvictionVerdict {
+    if trigger == EvictionTrigger::Timeout
+        && health
+            .last_success
+            .is_some_and(|at| now.saturating_duration_since(at) < liveness_window)
+    {
+        return EvictionVerdict::PeerRecentlyServed;
+    }
+    if health
+        .last_eviction
+        .is_some_and(|at| now.saturating_duration_since(at) < cooldown)
+    {
+        return EvictionVerdict::CoolingDown;
+    }
+    EvictionVerdict::Evict
+}
+
+/// Lock ids whose batch entry the server reports as granted.
+fn acquired_lock_ids(lock_ids: &[LockId], results: &[GenerallyLockResult]) -> Vec<LockId> {
+    results
+        .iter()
+        .zip(lock_ids)
+        .filter(|(result, _)| result.success)
+        .map(|(_, lock_id)| lock_id.clone())
+        .collect()
+}
+
 /// Remote lock client implementation
 #[derive(Debug, Clone)]
 pub struct RemoteClient {
@@ -198,14 +329,202 @@ impl RemoteClient {
         )
     }
 
-    async fn execute_rpc<T, F>(&self, op: &'static str, resource_summary: &str, future: F) -> std::result::Result<T, LockError>
+    fn eviction_cooldown() -> Duration {
+        Duration::from_millis(rustfs_utils::get_env_u64(
+            rustfs_config::ENV_OBJECT_LOCK_RPC_EVICTION_COOLDOWN_MS,
+            rustfs_config::DEFAULT_OBJECT_LOCK_RPC_EVICTION_COOLDOWN_MS,
+        ))
+    }
+
+    fn detached_rpc_limit() -> usize {
+        rustfs_utils::get_env_usize(
+            rustfs_config::ENV_OBJECT_LOCK_RPC_DETACHED_LIMIT,
+            rustfs_config::DEFAULT_OBJECT_LOCK_RPC_DETACHED_LIMIT,
+        )
+    }
+
+    fn liveness_window(deadline: Duration) -> Duration {
+        deadline.saturating_mul(LOCK_RPC_LIVENESS_WINDOW_DEADLINES)
+    }
+
+    fn record_rpc_success(&self) {
+        with_lock_peer_health(&self.addr, |health| {
+            health.last_success = Some(Instant::now());
+            health.consecutive_timeouts = 0;
+        });
+    }
+
+    /// Apply the per-peer eviction policy after a failed RPC.
+    async fn maybe_evict_connection(
+        &self,
+        op: &'static str,
+        reason: &str,
+        resource_summary: &str,
+        trigger: EvictionTrigger,
+        deadline: Duration,
+    ) {
+        let now = Instant::now();
+        let cooldown = Self::eviction_cooldown();
+        let liveness_window = Self::liveness_window(deadline);
+        let (verdict, consecutive_timeouts) = with_lock_peer_health(&self.addr, |health| {
+            if trigger == EvictionTrigger::Timeout {
+                health.consecutive_timeouts = health.consecutive_timeouts.saturating_add(1);
+            }
+            let verdict = eviction_verdict(health, now, trigger, liveness_window, cooldown);
+            if verdict == EvictionVerdict::Evict {
+                health.last_eviction = Some(now);
+            }
+            (verdict, health.consecutive_timeouts)
+        });
+        if verdict == EvictionVerdict::Evict {
+            rustfs_io_metrics::lock_metrics::record_remote_lock_channel_eviction(&self.addr, trigger.as_str());
+            self.evict_connection(op, reason, resource_summary).await;
+            return;
+        }
+        rustfs_io_metrics::lock_metrics::record_remote_lock_channel_eviction_suppressed(&self.addr, verdict.as_str());
+        debug!(
+            addr = %self.addr,
+            op,
+            resource_summary,
+            trigger = trigger.as_str(),
+            verdict = verdict.as_str(),
+            consecutive_timeouts,
+            "Keeping cached remote lock connection after RPC failure"
+        );
+    }
+
+    /// Keep a timed-out RPC running instead of cancelling its stream.
+    ///
+    /// Dropping the future sends `RST_STREAM`; under load those resets pile up
+    /// in the server's pending-accept queue until it answers `GOAWAY
+    /// too_many_resets` and kills every stream on the connection. A detached
+    /// stream ends on its own within the internode RPC timeout, the number per
+    /// peer is bounded, and a lock granted after its caller gave up is released.
+    fn detach_timed_out_rpc<T: Send + 'static>(
+        &self,
+        op: &'static str,
+        resource_summary: &str,
+        handle: JoinHandle<std::result::Result<T, tonic::Status>>,
+        late: LateCompletion<T>,
+    ) {
+        let limit = Self::detached_rpc_limit();
+        let admitted = with_lock_peer_health(&self.addr, |health| {
+            if health.detached_rpcs >= limit {
+                false
+            } else {
+                health.detached_rpcs += 1;
+                true
+            }
+        });
+        if !admitted {
+            handle.abort();
+            rustfs_io_metrics::lock_metrics::record_remote_lock_rpc_detached(op, "aborted");
+            debug!(
+                addr = %self.addr,
+                op,
+                resource_summary,
+                limit,
+                "Cancelled timed-out remote lock RPC because the detached stream budget is exhausted"
+            );
+            return;
+        }
+        rustfs_io_metrics::lock_metrics::record_remote_lock_rpc_detached(op, "detached");
+        let addr = self.addr.clone();
+        tokio::spawn(async move {
+            let outcome = handle.await;
+            with_lock_peer_health(&addr, |health| health.detached_rpcs = health.detached_rpcs.saturating_sub(1));
+            match outcome {
+                Ok(Ok(response)) => {
+                    with_lock_peer_health(&addr, |health| {
+                        health.last_success = Some(Instant::now());
+                        health.consecutive_timeouts = 0;
+                    });
+                    rustfs_io_metrics::lock_metrics::record_remote_lock_rpc_late_completion(op, "success");
+                    if let Some(late) = late {
+                        late(response).await;
+                    }
+                }
+                Ok(Err(status)) => {
+                    rustfs_io_metrics::lock_metrics::record_remote_lock_rpc_late_completion(op, "error");
+                    debug!(
+                        addr = %addr,
+                        op,
+                        tonic_code = ?status.code(),
+                        tonic_message = status.message(),
+                        "Detached remote lock RPC failed after its caller timed out"
+                    );
+                }
+                Err(join_error) => {
+                    rustfs_io_metrics::lock_metrics::record_remote_lock_rpc_late_completion(op, "join_error");
+                    debug!(addr = %addr, op, error = %join_error, "Detached remote lock RPC task ended abnormally");
+                }
+            }
+        });
+    }
+
+    fn late_release_hook(&self, lock_id: LockId) -> LateCompletion<Response<GenerallyLockResponse>> {
+        let client = self.clone();
+        Some(Box::new(move |response: Response<GenerallyLockResponse>| {
+            Box::pin(async move {
+                if response.get_ref().success {
+                    client.release_late_acquisitions(vec![lock_id]).await;
+                }
+            })
+        }))
+    }
+
+    fn late_release_batch_hook(&self, lock_ids: Vec<LockId>) -> LateCompletion<Response<BatchGenerallyLockResponse>> {
+        let client = self.clone();
+        Some(Box::new(move |response: Response<BatchGenerallyLockResponse>| {
+            Box::pin(async move {
+                let acquired = acquired_lock_ids(&lock_ids, &response.get_ref().results);
+                if !acquired.is_empty() {
+                    client.release_late_acquisitions(acquired).await;
+                }
+            })
+        }))
+    }
+
+    /// A lock granted after its caller stopped waiting is an orphan until its
+    /// lease expires; hand it back right away, best effort.
+    async fn release_late_acquisitions(&self, lock_ids: Vec<LockId>) {
+        let outcome = match self.release_locks_batch(&lock_ids).await {
+            Ok(released) if released.iter().all(|released| *released) => "released",
+            Ok(_) => "partial",
+            Err(_) => "failed",
+        };
+        rustfs_io_metrics::lock_metrics::record_remote_lock_late_release(outcome);
+        if outcome == "released" {
+            debug!(addr = %self.addr, count = lock_ids.len(), "Released remote locks granted after their caller timed out");
+        } else {
+            warn!(
+                addr = %self.addr,
+                count = lock_ids.len(),
+                outcome,
+                "Could not release every remote lock granted after its caller timed out; the server lease will expire it"
+            );
+        }
+    }
+
+    async fn execute_rpc<T, Fut>(
+        &self,
+        op: &'static str,
+        resource_summary: &str,
+        deadline: Duration,
+        future: Fut,
+        late: LateCompletion<T>,
+    ) -> std::result::Result<T, LockError>
     where
-        F: std::future::Future<Output = std::result::Result<T, tonic::Status>>,
+        Fut: Future<Output = std::result::Result<T, tonic::Status>> + Send + 'static,
+        T: Send + 'static,
     {
-        let lock_timeout = Self::rpc_timeout();
-        match timeout(lock_timeout, future).await {
-            Ok(Ok(response)) => Ok(response),
-            Ok(Err(err)) => {
+        let mut handle = tokio::spawn(future);
+        match timeout(deadline, &mut handle).await {
+            Ok(Ok(Ok(response))) => {
+                self.record_rpc_success();
+                Ok(response)
+            }
+            Ok(Ok(Err(err))) => {
                 let reason = err.to_string();
                 // Only evict (and re-dial) the cached channel when the failure is a genuine
                 // transport problem. A server-produced application status (auth denied, peer
@@ -217,7 +536,7 @@ impl RemoteClient {
                     debug!(
                         addr = %self.addr,
                         op,
-                        timeout_ms = lock_timeout.as_millis(),
+                        timeout_ms = deadline.as_millis(),
                         resource_summary,
                         tonic_code = ?err.code(),
                         tonic_message = err.message(),
@@ -228,7 +547,7 @@ impl RemoteClient {
                     warn!(
                         addr = %self.addr,
                         op,
-                        timeout_ms = lock_timeout.as_millis(),
+                        timeout_ms = deadline.as_millis(),
                         resource_summary,
                         tonic_code = ?err.code(),
                         tonic_message = err.message(),
@@ -237,17 +556,29 @@ impl RemoteClient {
                     );
                 }
                 if transport_failure {
-                    self.evict_connection(op, &reason, resource_summary).await;
+                    self.maybe_evict_connection(op, &reason, resource_summary, EvictionTrigger::Transport, deadline)
+                        .await;
                 }
                 Err(LockError::internal(format!("{op} RPC failed: {reason}")))
             }
+            Ok(Err(join_error)) => {
+                warn!(
+                    addr = %self.addr,
+                    op,
+                    resource_summary,
+                    error = %join_error,
+                    "Remote lock RPC task ended abnormally"
+                );
+                Err(LockError::internal(format!("{op} RPC task failed: {join_error}")))
+            }
             Err(_) => {
-                let reason = format!("RPC timed out after {:?}", lock_timeout);
+                let reason = format!("RPC timed out after {deadline:?}");
+                rustfs_io_metrics::lock_metrics::record_remote_lock_rpc_timeout(&self.addr, op);
                 if Self::is_scanner_leader_lock(resource_summary) {
                     debug!(
                         addr = %self.addr,
                         op,
-                        timeout_ms = lock_timeout.as_millis(),
+                        timeout_ms = deadline.as_millis(),
                         resource_summary,
                         "Remote lock RPC timed out for scanner leader lock"
                     );
@@ -255,13 +586,15 @@ impl RemoteClient {
                     warn!(
                         addr = %self.addr,
                         op,
-                        timeout_ms = lock_timeout.as_millis(),
+                        timeout_ms = deadline.as_millis(),
                         resource_summary,
                         "Remote lock RPC timed out"
                     );
                 }
-                self.evict_connection(op, &reason, resource_summary).await;
-                Err(LockError::timeout(format!("remote lock RPC {op} on {}", self.addr), lock_timeout))
+                self.maybe_evict_connection(op, &reason, resource_summary, EvictionTrigger::Timeout, deadline)
+                    .await;
+                self.detach_timed_out_rpc(op, resource_summary, handle, late);
+                Err(LockError::timeout(format!("remote lock RPC {op} on {}", self.addr), deadline))
             }
         }
     }
@@ -354,8 +687,18 @@ impl LockClient for RemoteClient {
                 .map_err(|e| LockError::internal(format!("Failed to serialize request: {e}")))?,
         });
         attach_lock_mutation_body_digest(&mut req)?;
+        let late = self.late_release_hook(request.lock_id.clone());
 
-        let resp = match self.execute_rpc("lock", &resource_summary, client.lock(req)).await {
+        let resp = match self
+            .execute_rpc(
+                "lock",
+                &resource_summary,
+                Self::rpc_timeout(),
+                async move { client.lock(req).await },
+                late,
+            )
+            .await
+        {
             Ok(resp) => resp.into_inner(),
             Err(err @ LockError::Timeout { .. }) => return Ok(Self::rpc_timeout_failure_response(request, &err)),
             Err(err) => return Ok(Self::rpc_failure_response(request, &err)),
@@ -393,9 +736,16 @@ impl LockClient for RemoteClient {
                 .collect::<Result<Vec<_>>>()?,
         });
         attach_lock_mutation_body_digest(&mut req)?;
+        let late = self.late_release_batch_hook(requests.iter().map(|request| request.lock_id.clone()).collect());
 
         let resp = match self
-            .execute_rpc("lock_batch", &resource_summary, client.lock_batch(req))
+            .execute_rpc(
+                "lock_batch",
+                &resource_summary,
+                Self::rpc_timeout(),
+                async move { client.lock_batch(req).await },
+                late,
+            )
             .await
         {
             Ok(resp) => resp.into_inner(),
@@ -436,7 +786,13 @@ impl LockClient for RemoteClient {
         let mut req = Request::new(GenerallyLockRequest { args: request_string });
         attach_lock_mutation_body_digest(&mut req)?;
         let resp = self
-            .execute_rpc("release", &resource_summary, client.un_lock(req))
+            .execute_rpc(
+                "release",
+                &resource_summary,
+                Self::rpc_timeout(),
+                async move { client.un_lock(req).await },
+                None,
+            )
             .await?
             .into_inner();
         if let Some(error_info) = resp.error_info {
@@ -464,7 +820,13 @@ impl LockClient for RemoteClient {
         attach_lock_mutation_body_digest(&mut req)?;
 
         let resp = self
-            .execute_rpc("release_batch", &resource_summary, client.un_lock_batch(req))
+            .execute_rpc(
+                "release_batch",
+                &resource_summary,
+                Self::rpc_timeout(),
+                async move { client.un_lock_batch(req).await },
+                None,
+            )
             .await?
             .into_inner();
 
@@ -486,7 +848,13 @@ impl LockClient for RemoteClient {
         });
         attach_lock_mutation_body_digest(&mut req)?;
         let resp = self
-            .execute_rpc("refresh", &resource_summary, client.refresh(req))
+            .execute_rpc(
+                "refresh",
+                &resource_summary,
+                Self::rpc_timeout(),
+                async move { client.refresh(req).await },
+                None,
+            )
             .await?
             .into_inner();
         if let Some(error_info) = resp.error_info {
@@ -506,7 +874,13 @@ impl LockClient for RemoteClient {
         });
         attach_lock_mutation_body_digest(&mut req)?;
         let resp = self
-            .execute_rpc("force_release", &resource_summary, client.force_un_lock(req))
+            .execute_rpc(
+                "force_release",
+                &resource_summary,
+                Self::rpc_timeout(),
+                async move { client.force_un_lock(req).await },
+                None,
+            )
             .await?
             .into_inner();
         if let Some(error_info) = resp.error_info {
@@ -523,16 +897,26 @@ impl LockClient for RemoteClient {
         let status_request = Self::create_unlock_request(lock_id);
         let resource_summary = status_request.resource.to_string();
         let mut client = self.get_client().await?;
+        let args = serde_json::to_string(&status_request)
+            .map_err(|e| LockError::internal(format!("Failed to serialize request: {e}")))?;
 
         // Try to acquire a very short-lived lock to test availability
-        let mut req = Request::new(GenerallyLockRequest {
-            args: serde_json::to_string(&status_request)
-                .map_err(|e| LockError::internal(format!("Failed to serialize request: {e}")))?,
-        });
+        let mut req = Request::new(GenerallyLockRequest { args: args.clone() });
         attach_lock_mutation_body_digest(&mut req)?;
+        // A probe lock granted after the deadline must not linger on the peer.
+        let late = self.late_release_hook(lock_id.clone());
 
         // Try exclusive lock first with very short timeout
-        let resp = match self.execute_rpc("check_status", &resource_summary, client.lock(req)).await {
+        let resp = match self
+            .execute_rpc(
+                "check_status",
+                &resource_summary,
+                Self::rpc_timeout(),
+                async move { client.lock(req).await },
+                late,
+            )
+            .await
+        {
             Ok(response) => response.into_inner(),
             Err(_) => return Ok(Some(Self::unknown_lock_info(lock_id))),
         };
@@ -540,14 +924,19 @@ impl LockClient for RemoteClient {
         if resp.success {
             // If we successfully acquired the lock, the resource was free.
             // Immediately release it on a best-effort basis.
-            let mut release_req = Request::new(GenerallyLockRequest {
-                args: serde_json::to_string(&status_request)
-                    .map_err(|e| LockError::internal(format!("Failed to serialize request: {e}")))?,
-            });
+            let mut release_req = Request::new(GenerallyLockRequest { args });
             attach_lock_mutation_body_digest(&mut release_req)?;
-            let _ = self
-                .execute_rpc("check_status_release", &resource_summary, client.un_lock(release_req))
-                .await;
+            if let Ok(mut client) = self.get_client().await {
+                let _ = self
+                    .execute_rpc(
+                        "check_status_release",
+                        &resource_summary,
+                        Self::rpc_timeout(),
+                        async move { client.un_lock(release_req).await },
+                        None,
+                    )
+                    .await;
+            }
 
             Ok(None)
         } else {
@@ -582,19 +971,8 @@ impl LockClient for RemoteClient {
 
     async fn is_online(&self) -> bool {
         let online_timeout = Self::online_check_timeout();
-        match timeout(online_timeout, async {
-            let mut client = self.get_client().await?;
-            let ping_req = Request::new(Self::build_ping_request());
-            self.execute_rpc("ping", Self::ONLINE_CHECK_RESOURCE, client.ping(ping_req))
-                .await?;
-            Ok::<(), LockError>(())
-        })
-        .await
-        {
-            Ok(Ok(())) => {
-                debug!(addr = %self.addr, timeout_ms = online_timeout.as_millis(), "remote lock client is online");
-                true
-            }
+        let mut client = match timeout(online_timeout, self.get_client()).await {
+            Ok(Ok(client)) => client,
             Ok(Err(err)) => {
                 debug!(
                     addr = %self.addr,
@@ -602,16 +980,39 @@ impl LockClient for RemoteClient {
                     error = %err,
                     "remote lock client online check failed"
                 );
-                false
+                return false;
             }
             Err(_) => {
-                let reason = format!("online check timed out after {:?}", online_timeout);
                 warn!(
                     addr = %self.addr,
                     timeout_ms = online_timeout.as_millis(),
-                    "remote lock client online check timed out"
+                    "remote lock client online check timed out while dialing"
+                );
+                return false;
+            }
+        };
+        let ping_req = Request::new(Self::build_ping_request());
+        match self
+            .execute_rpc(
+                "ping",
+                Self::ONLINE_CHECK_RESOURCE,
+                online_timeout,
+                async move { client.ping(ping_req).await },
+                None,
+            )
+            .await
+        {
+            Ok(_) => {
+                debug!(addr = %self.addr, timeout_ms = online_timeout.as_millis(), "remote lock client is online");
+                true
+            }
+            Err(err) => {
+                debug!(
+                    addr = %self.addr,
+                    timeout_ms = online_timeout.as_millis(),
+                    error = %err,
+                    "remote lock client online check failed"
                 );
-                self.evict_connection("ping", &reason, Self::ONLINE_CHECK_RESOURCE).await;
                 false
             }
         }
@@ -673,6 +1074,232 @@ mod tests {
             .with_priority(LockPriority::Normal)
     }
 
+    #[test]
+    fn eviction_verdict_distinguishes_slow_peers_from_dead_channels() {
+        let now = Instant::now() + Duration::from_secs(3600);
+        let window = Duration::from_secs(6);
+        let cooldown = Duration::from_secs(5);
+
+        let idle = LockPeerChannelHealth::default();
+        assert_eq!(
+            eviction_verdict(&idle, now, EvictionTrigger::Timeout, window, cooldown),
+            EvictionVerdict::Evict
+        );
+
+        let serving = LockPeerChannelHealth {
+            last_success: Some(now - Duration::from_secs(1)),
+            ..Default::default()
+        };
+        assert_eq!(
+            eviction_verdict(&serving, now, EvictionTrigger::Timeout, window, cooldown),
+            EvictionVerdict::PeerRecentlyServed,
+            "a timeout on a peer that just answered is load, not a dead channel"
+        );
+        assert_eq!(
+            eviction_verdict(&serving, now, EvictionTrigger::Transport, window, cooldown),
+            EvictionVerdict::Evict,
+            "a transport failure is reported by the channel itself and still evicts"
+        );
+
+        let quiet = LockPeerChannelHealth {
+            last_success: Some(now - Duration::from_secs(30)),
+            ..Default::default()
+        };
+        assert_eq!(
+            eviction_verdict(&quiet, now, EvictionTrigger::Timeout, window, cooldown),
+            EvictionVerdict::Evict
+        );
+
+        let just_evicted = LockPeerChannelHealth {
+            last_eviction: Some(now - Duration::from_secs(1)),
+            ..Default::default()
+        };
+        assert_eq!(
+            eviction_verdict(&just_evicted, now, EvictionTrigger::Timeout, window, cooldown),
+            EvictionVerdict::CoolingDown
+        );
+        assert_eq!(
+            eviction_verdict(&just_evicted, now, EvictionTrigger::Transport, window, cooldown),
+            EvictionVerdict::CoolingDown
+        );
+
+        let cooled = LockPeerChannelHealth {
+            last_eviction: Some(now - Duration::from_secs(10)),
+            ..Default::default()
+        };
+        assert_eq!(
+            eviction_verdict(&cooled, now, EvictionTrigger::Timeout, window, cooldown),
+            EvictionVerdict::Evict
+        );
+    }
+
+    #[test]
+    fn acquired_lock_ids_picks_only_granted_batch_entries() {
+        let lock_ids = vec![
+            LockId::new_unique(&ObjectKey::new("bucket", "a")),
+            LockId::new_unique(&ObjectKey::new("bucket", "b")),
+            LockId::new_unique(&ObjectKey::new("bucket", "c")),
+        ];
+        let results = vec![
+            GenerallyLockResult {
+                success: true,
+                ..Default::default()
+            },
+            GenerallyLockResult {
+                success: false,
+                ..Default::default()
+            },
+        ];
+        let acquired = acquired_lock_ids(&lock_ids, &results);
+        assert_eq!(
+            acquired,
+            vec![lock_ids[0].clone()],
+            "only granted entries with a matching id are released"
+        );
+        assert!(acquired_lock_ids(&lock_ids, &[]).is_empty());
+    }
+
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn test_remote_client_timeout_keeps_channel_of_recently_serving_peer() {
+        ensure_test_rpc_secret();
+        let Some((addr, accept_task)) = spawn_hanging_listener().await else {
+            return;
+        };
+        reset_lock_peer_health_for_test(&addr);
+        cache_lazy_channel(&addr).await;
+        with_lock_peer_health(&addr, |health| health.last_success = Some(Instant::now()));
+
+        temp_env::async_with_vars([(rustfs_config::ENV_OBJECT_LOCK_RPC_TIMEOUT_MS, Some("50"))], async {
+            let client = RemoteClient::new(addr.clone());
+            let response = client
+                .acquire_lock(&test_lock_request(Duration::from_millis(5)))
+                .await
+                .unwrap();
+            assert!(!response.success, "timed out lock acquisition should fail");
+            assert!(
+                runtime_sources::test_node_channel_is_cached(&addr).await,
+                "a peer that served a lock RPC within the liveness window is slow, not gone"
+            );
+            assert_eq!(lock_peer_health_for_test(&addr).consecutive_timeouts, 1);
+        })
+        .await;
+
+        accept_task.abort();
+        reset_lock_peer_health_for_test(&addr);
+    }
+
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn test_remote_client_repeated_timeouts_evict_at_most_once_per_cooldown() {
+        ensure_test_rpc_secret();
+        let Some((addr, accept_task)) = spawn_hanging_listener().await else {
+            return;
+        };
+        reset_lock_peer_health_for_test(&addr);
+        cache_lazy_channel(&addr).await;
+
+        temp_env::async_with_vars(
+            [
+                (rustfs_config::ENV_OBJECT_LOCK_RPC_TIMEOUT_MS, Some("50")),
+                (rustfs_config::ENV_OBJECT_LOCK_RPC_EVICTION_COOLDOWN_MS, Some("60000")),
+            ],
+            async {
+                let client = RemoteClient::new(addr.clone());
+                let request = test_lock_request(Duration::from_millis(5));
+
+                let _ = client.acquire_lock(&request).await.unwrap();
+                assert!(
+                    !runtime_sources::test_node_channel_is_cached(&addr).await,
+                    "the first timeout on a quiet peer evicts the cached channel"
+                );
+
+                cache_lazy_channel(&addr).await;
+                let _ = client.acquire_lock(&request).await.unwrap();
+                assert!(
+                    runtime_sources::test_node_channel_is_cached(&addr).await,
+                    "a second timeout inside the cooldown must not tear the fresh channel down again"
+                );
+                assert_eq!(lock_peer_health_for_test(&addr).consecutive_timeouts, 2);
+            },
+        )
+        .await;
+
+        accept_task.abort();
+        reset_lock_peer_health_for_test(&addr);
+    }
+
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn test_remote_client_detaches_timed_out_rpc_and_reclaims_its_slot() {
+        ensure_test_rpc_secret();
+        let Some((addr, accept_task)) = spawn_hanging_listener().await else {
+            return;
+        };
+        reset_lock_peer_health_for_test(&addr);
+        cache_lazy_channel(&addr).await;
+
+        temp_env::async_with_vars([(rustfs_config::ENV_OBJECT_LOCK_RPC_TIMEOUT_MS, Some("50"))], async {
+            let client = RemoteClient::new(addr.clone());
+            let _ = client
+                .acquire_lock(&test_lock_request(Duration::from_millis(5)))
+                .await
+                .unwrap();
+            assert_eq!(
+                lock_peer_health_for_test(&addr).detached_rpcs,
+                1,
+                "the timed-out stream keeps running instead of being reset"
+            );
+
+            // The hanging listener drops its socket after two seconds; the detached
+            // task then observes the transport failure and frees its slot.
+            let deadline = Instant::now() + Duration::from_secs(10);
+            while lock_peer_health_for_test(&addr).detached_rpcs != 0 {
+                assert!(Instant::now() < deadline, "detached RPC slot must be reclaimed once the stream ends");
+                tokio::time::sleep(Duration::from_millis(50)).await;
+            }
+        })
+        .await;
+
+        accept_task.abort();
+        reset_lock_peer_health_for_test(&addr);
+    }
+
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn test_remote_client_cancels_timed_out_rpc_when_detached_budget_is_exhausted() {
+        ensure_test_rpc_secret();
+        let Some((addr, accept_task)) = spawn_hanging_listener().await else {
+            return;
+        };
+        reset_lock_peer_health_for_test(&addr);
+        cache_lazy_channel(&addr).await;
+
+        temp_env::async_with_vars(
+            [
+                (rustfs_config::ENV_OBJECT_LOCK_RPC_TIMEOUT_MS, Some("50")),
+                (rustfs_config::ENV_OBJECT_LOCK_RPC_DETACHED_LIMIT, Some("0")),
+            ],
+            async {
+                let client = RemoteClient::new(addr.clone());
+                let response = client
+                    .acquire_lock(&test_lock_request(Duration::from_millis(5)))
+                    .await
+                    .unwrap();
+                assert!(!response.success);
+                assert_eq!(
+                    lock_peer_health_for_test(&addr).detached_rpcs,
+                    0,
+                    "an exhausted detached budget falls back to cancelling the stream"
+                );
+            },
+        )
+        .await;
+
+        accept_task.abort();
+        reset_lock_peer_health_for_test(&addr);
+    }
+
     #[test]
     fn lock_mutation_helper_marks_single_and_batch_requests_for_rolling_auth() {
         let mut single = Request::new(GenerallyLockRequest {
@@ -714,6 +1341,7 @@ mod tests {
         let Some((addr, accept_task)) = spawn_hanging_listener().await else {
             return;
         };
+        reset_lock_peer_health_for_test(&addr);
         cache_lazy_channel(&addr).await;
         assert!(runtime_sources::test_node_channel_is_cached(&addr).await);
 
@@ -759,6 +1387,7 @@ mod tests {
         let Some((addr, accept_task)) = spawn_hanging_listener().await else {
             return;
         };
+        reset_lock_peer_health_for_test(&addr);
         cache_lazy_channel(&addr).await;
         assert!(runtime_sources::test_node_channel_is_cached(&addr).await);
 
@@ -805,6 +1434,7 @@ mod tests {
         let Some((addr, accept_task)) = spawn_hanging_listener().await else {
             return;
         };
+        reset_lock_peer_health_for_test(&addr);
         cache_lazy_channel(&addr).await;
         assert!(runtime_sources::test_node_channel_is_cached(&addr).await);
 
@@ -842,6 +1472,7 @@ mod tests {
         let Some((addr, accept_task)) = spawn_hanging_listener().await else {
             return;
         };
+        reset_lock_peer_health_for_test(&addr);
         cache_lazy_channel(&addr).await;
         assert!(runtime_sources::test_node_channel_is_cached(&addr).await);
 
@@ -884,6 +1515,7 @@ mod tests {
         let Some(addr) = closed_listener_addr().await else {
             return;
         };
+        reset_lock_peer_health_for_test(&addr);
         cache_lazy_channel(&addr).await;
         assert!(runtime_sources::test_node_channel_is_cached(&addr).await);
 

crates/ecstore/src/core/pools.rs

@@ -4462,9 +4462,41 @@ pub(crate) struct PoolMetaWriteState {
     cluster_epoch: Option<u64>,
     pool_meta_absent: bool,
     bootstrap_authority: PoolMetaBootstrapAuthority,
+    /// First-hand bootstrap authority this process holds for each pool
+    /// (index = pool index): `Fresh` only for pools it formatted itself,
+    /// `LegacyAdoption` only for pools whose migration it verified. Empty when
+    /// the caller tracks deployment-wide authority only.
+    pool_bootstrap_authorities: Vec<PoolMetaBootstrapAuthority>,
+    /// Whether this process hosts the first endpoint of the first pool and is
+    /// therefore the only writer allowed to publish the initial `pool.bin`.
+    /// `None` when the caller did not say; unknown writers are treated as
+    /// elected so every fail-closed rule still applies to them.
+    elected_bootstrap_writer: Option<bool>,
     identity_initialized: Option<bool>,
     identity_fresh_bootstrap_nonce: Option<uuid::Uuid>,
     identity_needs_repair: bool,
+    /// At least one pool has no identity replica at all.
+    identity_replicas_missing: bool,
+    /// At least one pool has a replica that is present but not a valid identity.
+    identity_replicas_invalid: bool,
+}
+
+/// Why an all-missing `pool.bin` set may not be initialized right now.
+enum MissingMetadataRejection {
+    /// Another node still has to act (mint, attest, or publish); retrying the
+    /// startup loop is the remedy, so the write gate stays open.
+    BootstrapPending(Error),
+    /// The durable state contradicts a fresh bootstrap; writes stay blocked
+    /// until an operator recovers the metadata.
+    RecoveryRequired(Error),
+}
+
+impl MissingMetadataRejection {
+    fn into_error(self) -> Error {
+        match self {
+            Self::BootstrapPending(err) | Self::RecoveryRequired(err) => err,
+        }
+    }
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
@@ -4496,6 +4528,7 @@ impl PoolMetaWriteState {
         Self::for_startup_with_bootstrap_authority(cluster_id, bootstrap_authority)
     }
 
+    #[cfg(test)]
     pub(crate) fn for_startup_with_bootstrap_authority(
         cluster_id: uuid::Uuid,
         bootstrap_authority: PoolMetaBootstrapAuthority,
@@ -4507,10 +4540,82 @@ impl PoolMetaWriteState {
         }
     }
 
+    /// Startup state for a process that loaded every pool format itself and
+    /// remembers, per pool, whether it created (or adopted) that pool
+    /// first-hand. Deployment-wide authority is the conjunction across pools:
+    /// any pool this process merely read yields `None`, exactly as before.
+    pub(crate) fn for_startup_with_pool_bootstrap_authorities(
+        cluster_id: uuid::Uuid,
+        pool_bootstrap_authorities: Vec<PoolMetaBootstrapAuthority>,
+        elected_bootstrap_writer: bool,
+    ) -> Self {
+        let bootstrap_authority = pool_bootstrap_authorities
+            .iter()
+            .copied()
+            .reduce(PoolMetaBootstrapAuthority::combine_across_pools)
+            .unwrap_or_default();
+        Self {
+            expected_cluster_id: Some(cluster_id),
+            bootstrap_authority,
+            pool_bootstrap_authorities,
+            elected_bootstrap_writer: Some(elected_bootstrap_writer),
+            ..Default::default()
+        }
+    }
+
     pub(crate) fn bootstrap_identity_proven(&self) -> bool {
         self.bootstrap_authority.is_proven()
     }
 
+    fn pool_bootstrap_authority_proven(&self, pool_idx: usize) -> bool {
+        self.pool_bootstrap_authorities
+            .get(pool_idx)
+            .is_some_and(|authority| authority.is_proven())
+    }
+
+    /// Pools this process formatted or adopted first-hand during this startup.
+    pub(crate) fn attested_pool_indices(&self) -> Vec<usize> {
+        self.pool_bootstrap_authorities
+            .iter()
+            .enumerate()
+            .filter(|(_, authority)| authority.is_proven())
+            .map(|(pool_idx, _)| pool_idx)
+            .collect()
+    }
+
+    /// Deployment-level proof assembled from per-pool creators: this process
+    /// created the first pool itself, and every pool replica carries the same
+    /// pending identity. A pending replica is only ever written by the process
+    /// that formatted that pool with first-hand proof (see
+    /// [`PoolMetaIdentityWriteScope::Pools`]), so a complete, agreeing pending
+    /// set proves that every pool joined this bootstrap fresh. A missing,
+    /// corrupt, or disagreeing replica keeps the writer fail-closed, and a
+    /// restart without first-hand proof never reopens bootstrap on its own.
+    fn pending_identity_attested_by_every_pool(&self) -> bool {
+        self.elected_bootstrap_writer == Some(true)
+            && self.pool_bootstrap_authority_proven(0)
+            && self.identity_initialized == Some(false)
+            && !self.identity_needs_repair
+            && self.identity_fresh_bootstrap_nonce.is_some()
+    }
+
+    /// The elected writer minted (or holds) the nonce and the only thing
+    /// standing between it and a complete attestation is a pool whose creator
+    /// has not written its replica yet. Corrupt replicas are never transient.
+    fn awaiting_creator_attestation(&self) -> bool {
+        self.elected_bootstrap_writer == Some(true)
+            && self.pool_bootstrap_authority_proven(0)
+            && self.identity_initialized == Some(false)
+            && self.identity_fresh_bootstrap_nonce.is_some()
+            && self.identity_needs_repair
+            && self.identity_replicas_missing
+            && !self.identity_replicas_invalid
+    }
+
+    fn is_non_elected_bootstrap_observer(&self) -> bool {
+        self.elected_bootstrap_writer == Some(false)
+    }
+
     pub(crate) fn identity_is_pending(&self) -> bool {
         self.identity_initialized == Some(false)
     }
@@ -4630,9 +4735,19 @@ impl PoolMetaWriteState {
         self.identity_needs_repair = selection.needs_repair;
         self.identity_initialized = selection.identity.map(|identity| identity.initialized);
         self.identity_fresh_bootstrap_nonce = selection.identity.and_then(|identity| identity.fresh_bootstrap_nonce);
+        self.identity_replicas_missing = selection
+            .cas_tokens
+            .iter()
+            .any(|token| matches!(token, PoolMetaCasToken::Missing));
+        self.identity_replicas_invalid = selection
+            .valid_replicas
+            .iter()
+            .zip(&selection.cas_tokens)
+            .any(|(valid, token)| !valid && !matches!(token, PoolMetaCasToken::Missing));
         if let Some(identity) = selection.identity {
             if identity.initialized {
                 self.bootstrap_authority = PoolMetaBootstrapAuthority::None;
+                self.pool_bootstrap_authorities.clear();
             }
             if let Some(metadata_epoch) = self.cluster_epoch
                 && metadata_epoch != identity.epoch
@@ -4655,22 +4770,44 @@ impl PoolMetaWriteState {
         if !self.pool_meta_absent {
             return Ok(());
         }
-        self.validate_missing_metadata_can_initialize()
-            .map_err(|err| block_pool_meta_validation(self, err, "metadata_absence"))
+        match self.validate_missing_metadata_can_initialize() {
+            Ok(()) => Ok(()),
+            Err(MissingMetadataRejection::BootstrapPending(err)) => Err(err),
+            Err(MissingMetadataRejection::RecoveryRequired(err)) => {
+                Err(block_pool_meta_validation(self, err, "metadata_absence"))
+            }
+        }
     }
 
-    fn validate_missing_metadata_can_initialize(&self) -> Result<()> {
+    fn validate_missing_metadata_can_initialize(&self) -> std::result::Result<(), MissingMetadataRejection> {
+        use MissingMetadataRejection::{BootstrapPending, RecoveryRequired};
         match self.identity_initialized {
-            Some(false) if self.bootstrap_identity_proven() && self.identity_fresh_bootstrap_nonce.is_some() => Ok(()),
-            Some(false) => Err(Error::other(
-                "pool metadata recovery required: pending cluster identity exists but this startup has no verified fresh-bootstrap proof or legacy-adoption proof",
-            )),
-            Some(true) => Err(Error::other(
+            Some(false)
+                if self.identity_fresh_bootstrap_nonce.is_some()
+                    && (self.bootstrap_identity_proven() || self.pending_identity_attested_by_every_pool()) =>
+            {
+                Ok(())
+            }
+            Some(false) if self.awaiting_creator_attestation() => Err(BootstrapPending(Error::other(
+                "pool metadata bootstrap pending: waiting for every pool creator to attest the pending cluster identity",
+            ))),
+            Some(false) if self.is_non_elected_bootstrap_observer() && self.identity_fresh_bootstrap_nonce.is_some() => {
+                Err(BootstrapPending(Error::other(
+                    "pool metadata bootstrap pending: waiting for the elected writer to publish the initial pool.bin",
+                )))
+            }
+            Some(false) => Err(RecoveryRequired(Error::other(
+                "pool metadata recovery required: pending cluster identity exists but this startup has no verified fresh-bootstrap proof, legacy-adoption proof, or complete per-pool creator attestation",
+            ))),
+            Some(true) => Err(RecoveryRequired(Error::other(
                 "pool metadata recovery required: initialized cluster identity exists but every pool.bin replica is missing",
-            )),
-            None => Err(Error::other(
+            ))),
+            None if self.is_non_elected_bootstrap_observer() => Err(BootstrapPending(Error::other(
+                "pool metadata bootstrap pending: waiting for the elected writer to establish the cluster identity",
+            ))),
+            None => Err(RecoveryRequired(Error::other(
                 "pool metadata recovery required: no durable bootstrap identity or pool.bin replica is available",
-            )),
+            ))),
         }
     }
 
@@ -5381,7 +5518,9 @@ where
     write_state.validate_selection(&selection)?;
     selection.replica_state.ensure_write_safe(operation)?;
     if selection.absent && (write_state.expected_cluster_id.is_some() || write_state.identity_initialized.is_some()) {
-        write_state.validate_missing_metadata_can_initialize()?;
+        write_state
+            .validate_missing_metadata_can_initialize()
+            .map_err(MissingMetadataRejection::into_error)?;
     }
     Ok(selection)
 }
@@ -5483,6 +5622,8 @@ struct PoolMetaIdentitySelection {
     needs_repair: bool,
     repair_write_safe: bool,
     cas_tokens: Vec<PoolMetaCasToken>,
+    /// Per pool: whether the replica decoded as a valid identity.
+    valid_replicas: Vec<bool>,
 }
 
 fn encode_pool_meta_identity(identity: PersistedPoolMetaIdentity) -> Result<Vec<u8>> {
@@ -5531,6 +5672,17 @@ pub(crate) fn pool_meta_identity_initialized_for_test(data: &[u8]) -> Result<boo
     }
 }
 
+#[cfg(test)]
+pub(crate) fn pending_pool_meta_identity_for_test(cluster_id: uuid::Uuid, epoch: u64, nonce: uuid::Uuid) -> Result<Vec<u8>> {
+    encode_pool_meta_identity(PersistedPoolMetaIdentity {
+        version: POOL_META_IDENTITY_VERSION,
+        cluster_id,
+        epoch,
+        initialized: false,
+        fresh_bootstrap_nonce: Some(nonce),
+    })
+}
+
 #[cfg(test)]
 pub(crate) fn initialized_pool_meta_identity_for_test(cluster_id: uuid::Uuid, epoch: u64) -> Result<Vec<u8>> {
     encode_pool_meta_identity(PersistedPoolMetaIdentity {
@@ -5571,6 +5723,10 @@ fn select_pool_meta_identity(
     expected_cluster_id: uuid::Uuid,
 ) -> Result<PoolMetaIdentitySelection> {
     let cas_tokens = reads.iter().map(|read| read.cas.clone()).collect();
+    let valid_replicas = reads
+        .iter()
+        .map(|read| matches!(read.replica, PoolMetaIdentityReplica::Valid(_)))
+        .collect();
     let mut selected: Option<PersistedPoolMetaIdentity> = None;
     let mut needs_repair = false;
     let mut repair_write_safe = true;
@@ -5628,6 +5784,7 @@ fn select_pool_meta_identity(
         needs_repair,
         repair_write_safe,
         cas_tokens,
+        valid_replicas,
     })
 }
 
@@ -5897,10 +6054,43 @@ where
     result
 }
 
+/// Which pool replicas a cluster-identity write may touch.
+#[derive(Debug, Clone, Copy)]
+enum PoolMetaIdentityWriteScope<'a> {
+    /// Every pool. Creating a pending identity here requires deployment-wide
+    /// fresh-bootstrap or legacy-adoption proof.
+    All,
+    /// Only the listed pools, each of which this process formatted or adopted
+    /// first-hand. Multi-pool bootstraps whose pools have distinct format
+    /// creators use this scope: the first pool's creator mints the deployment
+    /// nonce and every other creator copies it to its own pool, so the elected
+    /// writer can verify a complete, agreeing pending set instead of trusting
+    /// an in-process flag it cannot observe on another node.
+    Pools(&'a [usize]),
+}
+
+fn identity_write_satisfied(
+    selection: &PoolMetaIdentitySelection,
+    identity: PersistedPoolMetaIdentity,
+    scope: PoolMetaIdentityWriteScope<'_>,
+    targets: &[usize],
+) -> bool {
+    if selection.identity != Some(identity) {
+        return false;
+    }
+    match scope {
+        PoolMetaIdentityWriteScope::All => !selection.needs_repair,
+        PoolMetaIdentityWriteScope::Pools(_) => targets
+            .iter()
+            .all(|pool_idx| selection.valid_replicas.get(*pool_idx).copied().unwrap_or(false)),
+    }
+}
+
 async fn persist_pool_meta_identity<S>(
     pools: Vec<Arc<S>>,
     write_state: &mut PoolMetaWriteState,
     initialized: bool,
+    scope: PoolMetaIdentityWriteScope<'_>,
     fence: &PoolMetaPersistenceFence<'_>,
     transaction_arm: &mut PoolMetaTransactionArm,
 ) -> Result<()>
@@ -5910,6 +6100,23 @@ where
     let Some(cluster_id) = write_state.expected_cluster_id else {
         return Ok(());
     };
+    let targets: Vec<usize> = match scope {
+        PoolMetaIdentityWriteScope::All => (0..pools.len()).collect(),
+        PoolMetaIdentityWriteScope::Pools(indices) => {
+            if initialized {
+                return Err(Error::other("pool metadata identity commit must address every pool"));
+            }
+            if indices
+                .iter()
+                .any(|pool_idx| *pool_idx >= pools.len() || !write_state.pool_bootstrap_authority_proven(*pool_idx))
+            {
+                return Err(Error::other(
+                    "pool metadata recovery required: a pending cluster identity can only be attested for pools this startup formatted or adopted first-hand",
+                ));
+            }
+            indices.to_vec()
+        }
+    };
     for attempt in 0..POOL_META_CAS_MAX_ATTEMPTS {
         let selection = load_pool_meta_identity_selection_observing(pools.clone(), write_state, cluster_id).await?;
         if !selection.repair_write_safe {
@@ -5918,20 +6125,39 @@ where
                 "pool metadata recovery required: cluster identity has an unreadable replica",
             ));
         }
-        let identity = match selection.identity {
-            Some(identity) if identity.initialized || initialized => PersistedPoolMetaIdentity {
+        let identity = match (selection.identity, scope) {
+            // An initialized deployment (for example a pool expansion) never
+            // reopens bootstrap: first-hand proof for a new pool is not a
+            // reason to publish a pending identity.
+            (Some(identity), PoolMetaIdentityWriteScope::Pools(_)) if identity.initialized => return Ok(()),
+            (Some(identity), _) if identity.initialized || initialized => PersistedPoolMetaIdentity {
                 initialized: true,
                 fresh_bootstrap_nonce: None,
                 ..identity
             },
-            Some(identity) => identity,
-            None if !initialized && !write_state.bootstrap_identity_proven() => {
+            (Some(identity), _) => identity,
+            // Only the first pool's creator mints the deployment nonce; every
+            // other creator waits until it is durable and copies it, so two
+            // concurrent creators can never publish disagreeing replicas.
+            (None, PoolMetaIdentityWriteScope::Pools(indices)) => {
+                if !indices.contains(&0) {
+                    return Ok(());
+                }
+                PersistedPoolMetaIdentity {
+                    version: POOL_META_IDENTITY_VERSION,
+                    cluster_id,
+                    epoch: write_state.cluster_epoch.unwrap_or(POOL_META_INITIAL_EPOCH),
+                    initialized: false,
+                    fresh_bootstrap_nonce: Some(uuid::Uuid::new_v4()),
+                }
+            }
+            (None, PoolMetaIdentityWriteScope::All) if !initialized && !write_state.bootstrap_identity_proven() => {
                 write_state.block_writes();
                 return Err(Error::other(
                     "pool metadata recovery required: cannot create a pending cluster identity without verified fresh-bootstrap proof or legacy-adoption proof",
                 ));
             }
-            None => PersistedPoolMetaIdentity {
+            (None, PoolMetaIdentityWriteScope::All) => PersistedPoolMetaIdentity {
                 version: POOL_META_IDENTITY_VERSION,
                 cluster_id,
                 epoch: write_state.cluster_epoch.unwrap_or(POOL_META_INITIAL_EPOCH),
@@ -5939,12 +6165,15 @@ where
                 fresh_bootstrap_nonce: (!initialized).then(uuid::Uuid::new_v4),
             },
         };
-        if selection.identity == Some(identity) && !selection.needs_repair {
+        if identity_write_satisfied(&selection, identity, scope, &targets) {
             return Ok(());
         }
         let data = encode_pool_meta_identity(identity)?;
         let mut conflict = false;
-        for (pool, token) in pools.iter().cloned().zip(&selection.cas_tokens) {
+        for (pool_idx, (pool, token)) in pools.iter().cloned().zip(&selection.cas_tokens).enumerate() {
+            if !targets.contains(&pool_idx) {
+                continue;
+            }
             match save_pool_meta_object_cas(
                 pool,
                 POOL_META_IDENTITY_NAME,
@@ -5971,7 +6200,7 @@ where
             return Err(Error::PreconditionFailed);
         }
         let confirmed = load_pool_meta_identity_selection_observing(pools.clone(), write_state, cluster_id).await?;
-        if confirmed.identity == Some(identity) && !confirmed.needs_repair {
+        if identity_write_satisfied(&confirmed, identity, scope, &targets) {
             return Ok(());
         }
     }
@@ -6003,6 +6232,34 @@ where
         pools,
         write_state,
         initialized,
+        PoolMetaIdentityWriteScope::All,
+        &PoolMetaPersistenceFence::Distributed(None),
+        &mut transaction_arm,
+    )
+    .await?;
+    transaction_arm.disarm();
+    Ok(())
+}
+
+/// Attest, during startup, the pending cluster identity for the pools this
+/// process formatted or adopted first-hand. The first pool's creator mints the
+/// deployment nonce; every other creator copies it once it is durable. Nothing
+/// is written while the deployment is already initialized or while the nonce
+/// is not yet durable, so callers simply retry through the startup loop.
+pub(crate) async fn persist_pool_meta_identity_for_attested_pools<S>(
+    pools: Vec<Arc<S>>,
+    write_state: &mut PoolMetaWriteState,
+    pool_indices: &[usize],
+) -> Result<()>
+where
+    S: EcstoreObjectIO,
+{
+    let mut transaction_arm = write_state.arm_transaction();
+    persist_pool_meta_identity(
+        pools,
+        write_state,
+        false,
+        PoolMetaIdentityWriteScope::Pools(pool_indices),
         &PoolMetaPersistenceFence::Distributed(None),
         &mut transaction_arm,
     )
@@ -6618,7 +6875,15 @@ where
                 .await?;
         }
     }
-    persist_pool_meta_identity(pools.clone(), write_state, true, fence, &mut transaction_arm).await?;
+    persist_pool_meta_identity(
+        pools.clone(),
+        write_state,
+        true,
+        PoolMetaIdentityWriteScope::All,
+        fence,
+        &mut transaction_arm,
+    )
+    .await?;
     let confirmed = load_pool_meta_for_transaction_recovery(pools, write_state).await?;
     if confirmed.revision != expected_revision
         || confirmed.canonical.as_ref() != Some(&expected_canonical)
@@ -7235,7 +7500,15 @@ impl PoolMeta {
             }
             if !selection.absent && write_state.identity_requires_repair() {
                 let initialized = write_state.identity_initialized != Some(false) || selection.revision.is_generation_protocol();
-                persist_pool_meta_identity(pools.clone(), write_state, initialized, fence, transaction_arm).await?;
+                persist_pool_meta_identity(
+                    pools.clone(),
+                    write_state,
+                    initialized,
+                    PoolMetaIdentityWriteScope::All,
+                    fence,
+                    transaction_arm,
+                )
+                .await?;
             }
         }
         // Startup is the only path allowed to create an all-missing metadata
@@ -7425,7 +7698,7 @@ impl PoolMeta {
             confirmed
         };
         if confirmed.revision == revision && confirmed.canonical.as_ref() == Some(&durable) {
-            persist_pool_meta_identity(pools, write_state, true, fence, transaction_arm).await?;
+            persist_pool_meta_identity(pools, write_state, true, PoolMetaIdentityWriteScope::All, fence, transaction_arm).await?;
             #[cfg(feature = "e2e-test-hooks")]
             startup_cas_test_observe(serde_json::json!({
                 "kind": "confirmed", "object": POOL_META_NAME,

crates/ecstore/src/store/init.rs

@@ -14,7 +14,7 @@
 
 use super::*;
 use crate::core::pools::{
-    PoolMetaBootstrapAuthority, PoolMetaReplicaState, PoolMetaWriteState, local_decommission_queue_prefix,
+    PoolMetaReplicaState, PoolMetaWriteState, local_decommission_queue_prefix, persist_pool_meta_identity_for_attested_pools,
     persist_pool_meta_identity_for_startup, pool_meta_has_active_decommission,
 };
 use crate::runtime::instance::InstanceContext;
@@ -174,9 +174,20 @@ where
     S: EcstoreObjectIO,
 {
     if elected_writer && write_state.bootstrap_identity_proven() {
-        persist_pool_meta_identity_for_startup(pools, write_state, false).await?;
+        return persist_pool_meta_identity_for_startup(pools, write_state, false).await;
     }
-    Ok(())
+    if write_state.bootstrap_identity_proven() {
+        return Ok(());
+    }
+    // Multi-pool bootstrap whose pools were formatted by different nodes: no
+    // single process can prove the whole deployment fresh in memory, so each
+    // creator attests the pools it formatted first-hand with the shared nonce
+    // and the elected writer waits for a complete, agreeing pending set.
+    let attested = write_state.attested_pool_indices();
+    if attested.is_empty() {
+        return Ok(());
+    }
+    persist_pool_meta_identity_for_attested_pools(pools, write_state, &attested).await
 }
 
 async fn save_validated_pool_meta_for_startup<S>(
@@ -407,7 +418,7 @@ impl ECStore {
         preflight_startup_rpc_secret(&endpoint_pools)?;
 
         let mut deployment_id = None;
-        let mut pool_meta_bootstrap_authority = None;
+        let mut pool_meta_bootstrap_authorities = Vec::new();
 
         // let (endpoint_pools, _) = EndpointServerPools::create_server_endpoints(address.as_str(), &layouts)?;
 
@@ -523,12 +534,10 @@ impl ECStore {
                     }
                 }
             }?;
-            pool_meta_bootstrap_authority = Some(pool_meta_bootstrap_authority.map_or(
-                loaded_format.pool_meta_bootstrap_authority,
-                |authority: PoolMetaBootstrapAuthority| {
-                    authority.combine_across_pools(loaded_format.pool_meta_bootstrap_authority)
-                },
-            ));
+            // First-hand authority for this pool only: `Fresh` when this process
+            // formatted it, `LegacyAdoption` when it verified the migration, and
+            // `None` when it merely read a format another node created.
+            pool_meta_bootstrap_authorities.push(loaded_format.pool_meta_bootstrap_authority);
             let fm = loaded_format.format;
 
             // Format loading succeeded, enable health monitoring on all disks
@@ -569,9 +578,13 @@ impl ECStore {
         let peer_sys = S3PeerSys::new_with_instance_ctx(&endpoint_pools, instance_ctx.clone());
         let mut pool_meta = PoolMeta::new(&pools, &PoolMeta::default());
         pool_meta.dont_save = true;
-        let pool_meta_write_state = PoolMetaWriteState::for_startup_with_bootstrap_authority(
+        let elected_bootstrap_writer = pools
+            .first()
+            .is_some_and(|pool| pool_first_endpoint_is_local(&pool.endpoints));
+        let pool_meta_write_state = PoolMetaWriteState::for_startup_with_pool_bootstrap_authorities(
             deployment_id,
-            pool_meta_bootstrap_authority.unwrap_or_default(),
+            pool_meta_bootstrap_authorities,
+            elected_bootstrap_writer,
         );
 
         let decommission_cancelers = RwLock::new(vec![None; pools.len()]);
@@ -961,8 +974,9 @@ mod tests {
         bucket::replication::{ReplicationState, ReplicationStatusType, replication_statuses_map},
         core::pools::{
             DecommissionErasureLayout, DecommissionPoolCapacityInfo, POOL_META_IDENTITY_NAME, POOL_META_NAME, POOL_META_VERSION,
-            PoolDecommissionInfo, PoolMeta, PoolStatus, pool_meta_identity_initialized_for_test,
-            pool_meta_v3_commit_state_for_test, set_decommission_capacity_info_overrides_for_test,
+            PoolDecommissionInfo, PoolMeta, PoolStatus, pending_pool_meta_identity_for_test,
+            pool_meta_identity_initialized_for_test, pool_meta_v3_commit_state_for_test,
+            set_decommission_capacity_info_overrides_for_test,
         },
         disk::endpoint::Endpoint,
         error::{Error, Result, StorageError},
@@ -1465,6 +1479,331 @@ mod tests {
         .await;
     }
 
+    fn startup_object(storage: &StartupPoolMetaStorage, object: &str) -> Option<Vec<u8>> {
+        storage
+            .objects
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .get(object)
+            .map(|(payload, _)| payload.clone())
+    }
+
+    /// Startup errors wrap their cause in context whose `Display` hides the
+    /// source, so assertions walk the chain the same way
+    /// `Error::pool_metadata_failure` does.
+    fn error_chain_text(err: &Error) -> String {
+        let mut parts = vec![err.to_string()];
+        let mut current: Option<&(dyn std::error::Error + 'static)> = Some(err);
+        while let Some(error) = current {
+            current = if let Some(io) = error.downcast_ref::<std::io::Error>() {
+                io.get_ref().map(|inner| inner as &(dyn std::error::Error + 'static))
+            } else {
+                error.source()
+            };
+            if let Some(next) = current {
+                parts.push(next.to_string());
+            }
+        }
+        parts.join(" <- ")
+    }
+
+    fn inject_startup_object(storage: &StartupPoolMetaStorage, object: &str, payload: Vec<u8>) {
+        storage
+            .objects
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .insert(object.to_string(), (payload, format!("injected-{object}")));
+    }
+
+    fn init_test_pool_meta_with_pools(pool_count: usize) -> PoolMeta {
+        PoolMeta {
+            version: POOL_META_VERSION,
+            pools: (0..pool_count)
+                .map(|id| PoolStatus {
+                    id,
+                    cmd_line: format!("pool-{id}"),
+                    last_update: OffsetDateTime::UNIX_EPOCH,
+                    decommission: None,
+                })
+                .collect(),
+            dont_save: false,
+        }
+    }
+
+    /// Two single-node pools whose formats were created by different nodes:
+    /// node0 formatted pool0 and only read pool1's format, node1 the reverse.
+    fn two_pool_creator_states(deployment_id: Uuid) -> (PoolMetaWriteState, PoolMetaWriteState) {
+        let node0 = PoolMetaWriteState::for_startup_with_pool_bootstrap_authorities(
+            deployment_id,
+            vec![PoolMetaBootstrapAuthority::Fresh, PoolMetaBootstrapAuthority::None],
+            true,
+        );
+        let node1 = PoolMetaWriteState::for_startup_with_pool_bootstrap_authorities(
+            deployment_id,
+            vec![PoolMetaBootstrapAuthority::None, PoolMetaBootstrapAuthority::Fresh],
+            false,
+        );
+        (node0, node1)
+    }
+
+    #[tokio::test]
+    async fn test_two_pool_bootstrap_with_distinct_format_creators_converges_through_creator_attestation() {
+        let deployment_id = Uuid::new_v4();
+        let pool0 = Arc::new(StartupPoolMetaStorage::new(Vec::new()));
+        let pool1 = Arc::new(StartupPoolMetaStorage::new(Vec::new()));
+        let pools = vec![pool0.clone(), pool1.clone()];
+        let (mut node0, mut node1) = two_pool_creator_states(deployment_id);
+        assert!(!node0.bootstrap_identity_proven(), "reading pool1's format is not deployment-wide proof");
+        assert!(!node1.bootstrap_identity_proven());
+
+        // node1 (pool1 creator, non-elected) starts first: no durable nonce exists
+        // yet, so it must neither mint one nor latch its write gate while waiting.
+        establish_pool_meta_bootstrap_identity_if_proven(pools.clone(), &mut node1, false)
+            .await
+            .expect("a non-first creator without a durable nonce writes nothing");
+        assert!(startup_object(&pool0, POOL_META_IDENTITY_NAME).is_none());
+        assert!(startup_object(&pool1, POOL_META_IDENTITY_NAME).is_none());
+        let err = load_pool_meta_for_startup(pools.clone(), &mut node1)
+            .await
+            .expect_err("nothing durable authorizes a non-elected node");
+        assert!(err.to_string().contains("bootstrap pending"), "{err}");
+        node1
+            .ensure_write_safe("waiting non-elected creator")
+            .expect("waiting for the elected writer must not latch the write gate");
+
+        // node0 (pool0 creator, elected) mints the nonce on the pool it created;
+        // pool1 is still unattested, so it cannot publish pool.bin and must not latch.
+        establish_pool_meta_bootstrap_identity_if_proven(pools.clone(), &mut node0, true)
+            .await
+            .expect("the first pool's creator mints the pending identity");
+        let minted = startup_object(&pool0, POOL_META_IDENTITY_NAME).expect("pool0 pending identity");
+        assert!(!pool_meta_identity_initialized_for_test(&minted).expect("decode pending identity"));
+        assert!(
+            startup_object(&pool1, POOL_META_IDENTITY_NAME).is_none(),
+            "node0 holds no first-hand proof for pool1 and must not attest it"
+        );
+        let err = load_pool_meta_for_startup(pools.clone(), &mut node0)
+            .await
+            .expect_err("an unattested pool keeps the elected writer from publishing");
+        assert!(err.to_string().contains("waiting for every pool creator"), "{err}");
+        node0
+            .ensure_write_safe("waiting elected writer")
+            .expect("waiting for creators must not latch the write gate");
+        assert!(startup_object(&pool0, POOL_META_NAME).is_none());
+
+        // node1 retries: it copies pool0's pending identity (same nonce) onto the
+        // pool it created, then keeps waiting for the elected writer's pool.bin.
+        establish_pool_meta_bootstrap_identity_if_proven(pools.clone(), &mut node1, false)
+            .await
+            .expect("the pool1 creator attests with the durable nonce");
+        assert_eq!(startup_object(&pool1, POOL_META_IDENTITY_NAME).as_deref(), Some(minted.as_slice()));
+        let err = load_pool_meta_for_startup(pools.clone(), &mut node1)
+            .await
+            .expect_err("a complete pending set never unlocks a non-elected node");
+        assert!(err.to_string().contains("waiting for the elected writer to publish"), "{err}");
+        node1
+            .ensure_write_safe("attested non-elected creator")
+            .expect("waiting for pool.bin must not latch the write gate");
+        assert!(startup_object(&pool0, POOL_META_NAME).is_none());
+
+        // node0 retries: every pool is attested under one nonce, so it publishes
+        // pool.bin and commits the identity on both pools.
+        establish_pool_meta_bootstrap_identity_if_proven(pools.clone(), &mut node0, true)
+            .await
+            .expect("re-establishing an already minted identity is idempotent");
+        let (_, replica_state) = load_pool_meta_for_startup(pools.clone(), &mut node0)
+            .await
+            .expect("complete creator attestation authorizes the initial pool metadata write");
+        persist_pool_meta_for_startup_if_safe(
+            &init_test_pool_meta_with_pools(2),
+            pools.clone(),
+            replica_state,
+            &mut node0,
+            true,
+            true,
+        )
+        .await
+        .expect("the elected writer publishes pool.bin and commits the identity");
+        for pool in [&pool0, &pool1] {
+            assert!(startup_object(pool, POOL_META_NAME).is_some());
+            let identity = startup_object(pool, POOL_META_IDENTITY_NAME).expect("committed identity");
+            assert!(pool_meta_identity_initialized_for_test(&identity).expect("decode committed identity"));
+        }
+
+        // node1 retries once more: pool.bin exists and nothing is rewritten.
+        let before = startup_object(&pool1, POOL_META_IDENTITY_NAME);
+        establish_pool_meta_bootstrap_identity_if_proven(pools.clone(), &mut node1, false)
+            .await
+            .expect("an initialized deployment never reopens bootstrap");
+        assert_eq!(startup_object(&pool1, POOL_META_IDENTITY_NAME), before);
+        load_pool_meta_for_startup(pools, &mut node1)
+            .await
+            .expect("published pool metadata admits the non-elected node");
+        node1
+            .ensure_write_safe("converged non-elected creator")
+            .expect("no latch remains after convergence");
+    }
+
+    #[tokio::test]
+    async fn test_two_pool_bootstrap_rejects_pending_replicas_from_different_bootstraps() {
+        let deployment_id = Uuid::new_v4();
+        let pool0 = Arc::new(StartupPoolMetaStorage::new(Vec::new()));
+        let pool1 = Arc::new(StartupPoolMetaStorage::new(Vec::new()));
+        let pools = vec![pool0.clone(), pool1.clone()];
+        let (mut node0, _) = two_pool_creator_states(deployment_id);
+        establish_pool_meta_bootstrap_identity_if_proven(pools.clone(), &mut node0, true)
+            .await
+            .expect("the first pool's creator mints the pending identity");
+        inject_startup_object(
+            &pool1,
+            POOL_META_IDENTITY_NAME,
+            pending_pool_meta_identity_for_test(deployment_id, 1, Uuid::new_v4()).expect("encode foreign pending identity"),
+        );
+
+        let err = load_pool_meta_for_startup(pools.clone(), &mut node0)
+            .await
+            .expect_err("a pending replica bound to another bootstrap nonce must fail closed");
+        let chain = error_chain_text(&err);
+        assert!(chain.contains("disagree on fresh-bootstrap proof"), "{chain}");
+        node0
+            .ensure_write_safe("split bootstrap")
+            .expect_err("a split bootstrap latches the write gate");
+        assert!(startup_object(&pool0, POOL_META_NAME).is_none());
+    }
+
+    #[tokio::test]
+    async fn test_two_pool_bootstrap_treats_corrupt_creator_replica_as_recovery_not_waiting() {
+        let deployment_id = Uuid::new_v4();
+        let pool0 = Arc::new(StartupPoolMetaStorage::new(Vec::new()));
+        let pool1 = Arc::new(StartupPoolMetaStorage::new(Vec::new()));
+        let pools = vec![pool0.clone(), pool1.clone()];
+        let (mut node0, _) = two_pool_creator_states(deployment_id);
+        establish_pool_meta_bootstrap_identity_if_proven(pools.clone(), &mut node0, true)
+            .await
+            .expect("the first pool's creator mints the pending identity");
+        // Keep the on-disk format/version header so the replica classifies as
+        // corrupt (undecodable payload) rather than as an incompatible format.
+        let mut corrupt = pending_pool_meta_identity_for_test(deployment_id, 1, Uuid::new_v4()).expect("encode identity");
+        corrupt.truncate(4);
+        corrupt.extend_from_slice(b"not a cluster identity");
+        inject_startup_object(&pool1, POOL_META_IDENTITY_NAME, corrupt);
+
+        let err = load_pool_meta_for_startup(pools.clone(), &mut node0)
+            .await
+            .expect_err("a corrupt replica is not a creator that is still catching up");
+        let chain = error_chain_text(&err);
+        assert!(chain.contains("no verified fresh-bootstrap proof"), "{chain}");
+        node0
+            .ensure_write_safe("corrupt attestation")
+            .expect_err("a corrupt attestation latches the write gate");
+        assert!(startup_object(&pool0, POOL_META_NAME).is_none());
+    }
+
+    #[tokio::test]
+    async fn test_elected_restart_without_first_hand_proof_cannot_reuse_a_complete_pending_set() {
+        let deployment_id = Uuid::new_v4();
+        let pool0 = Arc::new(StartupPoolMetaStorage::new(Vec::new()));
+        let pool1 = Arc::new(StartupPoolMetaStorage::new(Vec::new()));
+        let pools = vec![pool0.clone(), pool1.clone()];
+        let (mut node0, mut node1) = two_pool_creator_states(deployment_id);
+        establish_pool_meta_bootstrap_identity_if_proven(pools.clone(), &mut node0, true)
+            .await
+            .expect("mint");
+        establish_pool_meta_bootstrap_identity_if_proven(pools.clone(), &mut node1, false)
+            .await
+            .expect("attest");
+        assert_eq!(
+            startup_object(&pool0, POOL_META_IDENTITY_NAME),
+            startup_object(&pool1, POOL_META_IDENTITY_NAME),
+            "both creators attested the same pending identity"
+        );
+
+        // The elected node restarts before publishing: it now merely reads both
+        // formats, so the complete pending set alone must not reopen bootstrap.
+        let mut restarted = PoolMetaWriteState::for_startup_with_pool_bootstrap_authorities(
+            deployment_id,
+            vec![PoolMetaBootstrapAuthority::None, PoolMetaBootstrapAuthority::None],
+            true,
+        );
+        establish_pool_meta_bootstrap_identity_if_proven(pools.clone(), &mut restarted, true)
+            .await
+            .expect("a restart without first-hand proof writes nothing");
+        let err = load_pool_meta_for_startup(pools.clone(), &mut restarted)
+            .await
+            .expect_err("a pending set alone never authorizes a writer without first-hand proof");
+        assert!(err.to_string().contains("no verified fresh-bootstrap proof"), "{err}");
+        restarted
+            .ensure_write_safe("unproven restart")
+            .expect_err("the rejected restart latches the write gate");
+        assert!(startup_object(&pool0, POOL_META_NAME).is_none());
+    }
+
+    #[tokio::test]
+    async fn test_fresh_pool_joining_an_initialized_deployment_never_reopens_bootstrap() {
+        let deployment_id = Uuid::new_v4();
+        let pool0 = Arc::new(StartupPoolMetaStorage::new(Vec::new()));
+        let mut founder = PoolMetaWriteState::for_startup(deployment_id, true);
+        establish_pool_meta_bootstrap_identity_if_proven(vec![pool0.clone()], &mut founder, true)
+            .await
+            .expect("the founder mints");
+        let (_, replica_state) = load_pool_meta_for_startup(vec![pool0.clone()], &mut founder)
+            .await
+            .expect("the founder may initialize");
+        persist_pool_meta_for_startup_if_safe(
+            &init_test_pool_meta(None),
+            vec![pool0.clone()],
+            replica_state,
+            &mut founder,
+            true,
+            true,
+        )
+        .await
+        .expect("the founder commits");
+        let founded = startup_object(&pool0, POOL_META_IDENTITY_NAME).expect("committed identity");
+        assert!(pool_meta_identity_initialized_for_test(&founded).expect("decode committed identity"));
+
+        // Expansion: pool1 is fresh and was formatted first-hand by the node
+        // hosting its first endpoint, whether or not that node is elected.
+        let pool1 = Arc::new(StartupPoolMetaStorage::new(Vec::new()));
+        let pools = vec![pool0.clone(), pool1.clone()];
+        for elected in [false, true] {
+            let mut joiner = PoolMetaWriteState::for_startup_with_pool_bootstrap_authorities(
+                deployment_id,
+                vec![PoolMetaBootstrapAuthority::None, PoolMetaBootstrapAuthority::Fresh],
+                elected,
+            );
+            establish_pool_meta_bootstrap_identity_if_proven(pools.clone(), &mut joiner, elected)
+                .await
+                .expect("an initialized deployment ignores first-hand proof for a new pool");
+            assert!(
+                startup_object(&pool1, POOL_META_IDENTITY_NAME).is_none(),
+                "no pending identity may be written to an expansion pool"
+            );
+            assert_eq!(startup_object(&pool0, POOL_META_IDENTITY_NAME).as_deref(), Some(founded.as_slice()));
+            let (_, replica_state) = load_pool_meta_for_startup(pools.clone(), &mut joiner)
+                .await
+                .expect("published pool metadata admits the joiner");
+            joiner
+                .ensure_write_safe("expansion joiner")
+                .expect("joining never latches the write gate");
+            if elected {
+                persist_pool_meta_for_startup_if_safe(
+                    &init_test_pool_meta_with_pools(2),
+                    pools.clone(),
+                    replica_state,
+                    &mut joiner,
+                    true,
+                    true,
+                )
+                .await
+                .expect("the topology update repairs the new pool's replicas");
+                let identity = startup_object(&pool1, POOL_META_IDENTITY_NAME).expect("expansion pool identity");
+                assert!(pool_meta_identity_initialized_for_test(&identity).expect("decode repaired identity"));
+                assert!(startup_object(&pool1, POOL_META_NAME).is_some());
+            }
+        }
+    }
+
     #[tokio::test]
     async fn test_store_init_distinguishes_fresh_deployment_from_wiped_lagging_node() {
         let deployment_id = Uuid::new_v4();

crates/io-metrics/src/lock_metrics.rs

@@ -62,6 +62,51 @@ pub fn record_contention_event() {
     counter!("rustfs_lock_contentions").increment(1);
 }
 
+/// Record a remote lock RPC that exceeded its caller's deadline.
+#[inline(always)]
+pub fn record_remote_lock_rpc_timeout(peer: &str, op: &'static str) {
+    use metrics::counter;
+    counter!("rustfs_remote_lock_rpc_timeouts_total", "peer" => peer.to_string(), "op" => op).increment(1);
+}
+
+/// Record the cached lock channel to `peer` being evicted after an RPC failure.
+#[inline(always)]
+pub fn record_remote_lock_channel_eviction(peer: &str, trigger: &'static str) {
+    use metrics::counter;
+    counter!("rustfs_remote_lock_channel_evictions_total", "peer" => peer.to_string(), "trigger" => trigger).increment(1);
+}
+
+/// Record an RPC failure that did not evict the cached lock channel to `peer`
+/// because the peer recently served a request or was re-dialed too recently.
+#[inline(always)]
+pub fn record_remote_lock_channel_eviction_suppressed(peer: &str, verdict: &'static str) {
+    use metrics::counter;
+    counter!("rustfs_remote_lock_channel_evictions_suppressed_total", "peer" => peer.to_string(), "verdict" => verdict)
+        .increment(1);
+}
+
+/// Record a timed-out lock RPC that was left running (`detached`) or cancelled
+/// because the per-peer detached budget was exhausted (`aborted`).
+#[inline(always)]
+pub fn record_remote_lock_rpc_detached(op: &'static str, outcome: &'static str) {
+    use metrics::counter;
+    counter!("rustfs_remote_lock_rpc_detached_total", "op" => op, "outcome" => outcome).increment(1);
+}
+
+/// Record how a detached lock RPC eventually ended.
+#[inline(always)]
+pub fn record_remote_lock_rpc_late_completion(op: &'static str, outcome: &'static str) {
+    use metrics::counter;
+    counter!("rustfs_remote_lock_rpc_late_completions_total", "op" => op, "outcome" => outcome).increment(1);
+}
+
+/// Record the release of a lock that was granted after its caller timed out.
+#[inline(always)]
+pub fn record_remote_lock_late_release(outcome: &'static str) {
+    use metrics::counter;
+    counter!("rustfs_remote_lock_late_releases_total", "outcome" => outcome).increment(1);
+}
+
 /// Record object namespace lock diagnostics being enabled.
 #[inline(always)]
 pub fn record_object_lock_diag_enabled(enabled: bool) {
@@ -183,6 +228,12 @@ mod tests {
             record_lock_hold_time(Duration::from_millis(100));
             record_early_release();
             record_contention_event();
+            record_remote_lock_rpc_timeout("http://peer:9000", "lock");
+            record_remote_lock_channel_eviction("http://peer:9000", "timeout");
+            record_remote_lock_channel_eviction_suppressed("http://peer:9000", "peer_recently_served");
+            record_remote_lock_rpc_detached("lock", "detached");
+            record_remote_lock_rpc_late_completion("lock", "success");
+            record_remote_lock_late_release("released");
         });
 
         let emitted: std::collections::HashSet<String> = snapshotter
@@ -199,6 +250,12 @@ mod tests {
             "rustfs_lock_hold_time_secs",
             "rustfs_lock_early_releases",
             "rustfs_lock_contentions",
+            "rustfs_remote_lock_rpc_timeouts_total",
+            "rustfs_remote_lock_channel_evictions_total",
+            "rustfs_remote_lock_channel_evictions_suppressed_total",
+            "rustfs_remote_lock_rpc_detached_total",
+            "rustfs_remote_lock_rpc_late_completions_total",
+            "rustfs_remote_lock_late_releases_total",
         ] {
             assert!(emitted.contains(expected), "{expected} must be emitted by its record helper");
         }

crates/lock/src/distributed_lock.rs

@@ -38,6 +38,16 @@ use uuid::Uuid;
 
 const UNLOCK_RETRY_ATTEMPTS: usize = 3;
 const UNLOCK_RETRY_BACKOFF: Duration = Duration::from_millis(100);
+/// Slow retry schedule for unlocks that survive the fast retry loop. Lock RPC
+/// timeouts under load are transient (issue #7363); giving up after three
+/// quick attempts left orphaned entries for the server lease to expire.
+const DEFERRED_UNLOCK_BACKOFF: [Duration; 5] = [
+    Duration::from_secs(1),
+    Duration::from_secs(2),
+    Duration::from_secs(4),
+    Duration::from_secs(8),
+    Duration::from_secs(16),
+];
 const LOCK_ACQUIRE_RETRY_INITIAL_BACKOFF: Duration = Duration::from_millis(250);
 const LOCK_ACQUIRE_ATTEMPT_TIMEOUT: Duration = Duration::from_secs(1);
 const LOCK_ACQUIRE_SPARE_HEDGES: usize = 1;
@@ -719,22 +729,7 @@ impl DistributedLock {
         let mut pending = entries;
 
         for attempt in 1..=UNLOCK_RETRY_ATTEMPTS {
-            let release_results = join_all(pending.into_iter().map(|(lock_id, client)| async move {
-                match client.release(&lock_id).await {
-                    Ok(true) => None,
-                    Ok(false) => {
-                        warn!(%lock_id, attempt, context, "distributed unlock did not find lock on client");
-                        Some((lock_id, client))
-                    }
-                    Err(err) => {
-                        warn!(%lock_id, attempt, context, "distributed unlock failed on client: {}", err);
-                        Some((lock_id, client))
-                    }
-                }
-            }))
-            .await;
-
-            pending = release_results.into_iter().flatten().collect();
+            pending = Self::release_pending_once(pending, attempt, context).await;
             if pending.is_empty() {
                 debug!(attempt, context, "distributed unlock completed");
                 return;
@@ -749,7 +744,54 @@ impl DistributedLock {
             remaining = pending.len(),
             attempts = UNLOCK_RETRY_ATTEMPTS,
             context,
-            "distributed unlock left unreleased entries after retry"
+            "distributed unlock left unreleased entries after retry; continuing with deferred retries"
+        );
+        Self::release_entries_deferred(pending, context).await;
+    }
+
+    async fn release_pending_once(
+        pending: Vec<(LockId, Arc<dyn LockClient>)>,
+        attempt: usize,
+        context: &'static str,
+    ) -> Vec<(LockId, Arc<dyn LockClient>)> {
+        let release_results = join_all(pending.into_iter().map(|(lock_id, client)| async move {
+            match client.release(&lock_id).await {
+                Ok(true) => None,
+                Ok(false) => {
+                    warn!(%lock_id, attempt, context, "distributed unlock did not find lock on client");
+                    Some((lock_id, client))
+                }
+                Err(err) => {
+                    warn!(%lock_id, attempt, context, "distributed unlock failed on client: {}", err);
+                    Some((lock_id, client))
+                }
+            }
+        }))
+        .await;
+
+        release_results.into_iter().flatten().collect()
+    }
+
+    /// Bounded slow retries for entries the fast loop could not release. Every
+    /// caller runs on a background task, so waiting here blocks nobody; after
+    /// the schedule is exhausted the server lease reclaims the entry.
+    async fn release_entries_deferred(mut pending: Vec<(LockId, Arc<dyn LockClient>)>, context: &'static str) {
+        let mut attempt = UNLOCK_RETRY_ATTEMPTS;
+        for delay in DEFERRED_UNLOCK_BACKOFF {
+            tokio::time::sleep(delay).await;
+            attempt += 1;
+            pending = Self::release_pending_once(pending, attempt, context).await;
+            if pending.is_empty() {
+                debug!(attempt, context, "deferred distributed unlock converged");
+                return;
+            }
+        }
+
+        warn!(
+            remaining = pending.len(),
+            attempts = attempt,
+            context,
+            "distributed unlock abandoned entries after deferred retry; the server lease will expire them"
         );
     }
 
@@ -795,7 +837,9 @@ impl DistributedLock {
                             continue;
                         };
 
-                        Self::release_entries(vec![(lock_id, client.clone())], context).await;
+                        // Deferred retries may wait tens of seconds; never hold up the
+                        // next late completion behind them.
+                        drop(tokio::spawn(Self::release_entries(vec![(lock_id, client.clone())], context)));
                     }
                     Ok((idx, Ok(resp))) => {
                         tracing::debug!(
@@ -1198,8 +1242,8 @@ fn record_lock_held_release(lock_type: LockType) {
 #[cfg(test)]
 mod tests {
     use super::{
-        DistributedLock, LOCK_ACQUIRE_ATTEMPT_TIMEOUT, LOCK_ACQUIRE_RETRY_INITIAL_BACKOFF, LockAcquireFailureKind,
-        LockLostSignal, is_remote_lock_rpc_failure, should_warn_lock_failure,
+        DEFERRED_UNLOCK_BACKOFF, DistributedLock, LOCK_ACQUIRE_ATTEMPT_TIMEOUT, LOCK_ACQUIRE_RETRY_INITIAL_BACKOFF,
+        LockAcquireFailureKind, LockLostSignal, UNLOCK_RETRY_ATTEMPTS, is_remote_lock_rpc_failure, should_warn_lock_failure,
     };
     use crate::{LockError, LockId, LockInfo, LockRequest, LockResponse, LockStats, LockType, ObjectKey, client::LockClient};
     use rand::{SeedableRng as _, TryRng, rngs::StdRng};
@@ -1692,6 +1736,94 @@ mod tests {
         drop(guard);
     }
 
+    /// Fails `release` a fixed number of times before succeeding, mimicking a
+    /// peer whose lock RPCs time out under load and then recover.
+    #[derive(Debug)]
+    struct FlakyReleaseClient {
+        failures_left: AtomicUsize,
+        release_calls: Arc<AtomicUsize>,
+    }
+
+    #[async_trait::async_trait]
+    impl LockClient for FlakyReleaseClient {
+        async fn acquire_lock(&self, _request: &LockRequest) -> crate::Result<LockResponse> {
+            Ok(LockResponse::failure("unused", Duration::ZERO))
+        }
+
+        async fn release(&self, _lock_id: &LockId) -> crate::Result<bool> {
+            self.release_calls.fetch_add(1, Ordering::SeqCst);
+            if self.failures_left.load(Ordering::SeqCst) > 0 {
+                self.failures_left.fetch_sub(1, Ordering::SeqCst);
+                return Err(LockError::internal("remote lock rpc timed out: release"));
+            }
+            Ok(true)
+        }
+
+        async fn refresh(&self, _lock_id: &LockId) -> crate::Result<bool> {
+            Ok(false)
+        }
+
+        async fn force_release(&self, _lock_id: &LockId) -> crate::Result<bool> {
+            Ok(false)
+        }
+
+        async fn check_status(&self, _lock_id: &LockId) -> crate::Result<Option<LockInfo>> {
+            Ok(None)
+        }
+
+        async fn get_stats(&self) -> crate::Result<LockStats> {
+            Ok(LockStats::default())
+        }
+
+        async fn close(&self) -> crate::Result<()> {
+            Ok(())
+        }
+
+        async fn is_online(&self) -> bool {
+            true
+        }
+
+        async fn is_local(&self) -> bool {
+            false
+        }
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn release_entries_keeps_retrying_transient_failures_after_the_fast_loop() {
+        let release_calls = Arc::new(AtomicUsize::new(0));
+        let client: Arc<dyn LockClient> = Arc::new(FlakyReleaseClient {
+            failures_left: AtomicUsize::new(UNLOCK_RETRY_ATTEMPTS + 2),
+            release_calls: release_calls.clone(),
+        });
+        let lock_id = LockId::new_unique(&ObjectKey::new("bucket", "object"));
+
+        DistributedLock::release_entries(vec![(lock_id, client)], "test_deferred_unlock").await;
+
+        assert_eq!(
+            release_calls.load(Ordering::SeqCst),
+            UNLOCK_RETRY_ATTEMPTS + 3,
+            "two deferred attempts fail, the third releases the entry"
+        );
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn release_entries_gives_up_after_the_deferred_schedule() {
+        let release_calls = Arc::new(AtomicUsize::new(0));
+        let client: Arc<dyn LockClient> = Arc::new(FlakyReleaseClient {
+            failures_left: AtomicUsize::new(usize::MAX),
+            release_calls: release_calls.clone(),
+        });
+        let lock_id = LockId::new_unique(&ObjectKey::new("bucket", "object"));
+
+        DistributedLock::release_entries(vec![(lock_id, client)], "test_deferred_unlock_abandoned").await;
+
+        assert_eq!(
+            release_calls.load(Ordering::SeqCst),
+            UNLOCK_RETRY_ATTEMPTS + DEFERRED_UNLOCK_BACKOFF.len(),
+            "the retry budget is bounded; the server lease reclaims what remains"
+        );
+    }
+
     #[derive(Debug)]
     struct ResponseClient {
         response: LockResponse,

docs/operations/lock-rpc-storm-protection.md

@@ -0,0 +1,43 @@
+# Lock RPC storm protection
+
+**Use this when:** a slow lock endpoint turns into cluster-wide `Remote lock RPC timed out`, `Evicting cached remote lock connection`, and `GOAWAY too_many_resets` log floods, or when you tune how the remote lock client reacts to per-request deadlines (rustfs#7363).
+
+## What the client does on a failed lock RPC
+
+Every remote lock call (`lock`, `lock_batch`, `release`, `refresh`, `force_release`, `check_status`, and the readiness `ping`) runs under the deadline from `RUSTFS_OBJECT_LOCK_RPC_TIMEOUT_MS` (readiness uses `RUSTFS_HEALTH_LOCK_ONLINE_TIMEOUT_MS`). A deadline only says that one stream was slow; it says nothing about the shared HTTP/2 channel it ran on. The client therefore keeps a small per-peer history and decides per failure:
+
+| Failure | Verdict | Effect |
+| --- | --- | --- |
+| Deadline expired, peer completed any lock RPC within two deadlines | `peer_recently_served` | Channel kept. The peer is slow, not gone. |
+| Deadline expired, peer quiet for longer than two deadlines | `evict` | Cached channel evicted once, then the next request re-dials. |
+| Any failure while the last eviction is younger than the cooldown | `cooling_down` | Channel kept so the fresh dial can prove itself; no re-dial burst. |
+| Transport failure (refused, reset, `GOAWAY`) outside the cooldown | `evict` | Cached channel evicted once. |
+
+A timed-out request is no longer cancelled. Cancelling sends `RST_STREAM`, and enough resets against a server that is slow to accept streams make it answer `GOAWAY too_many_resets`, which kills every stream on the connection and restarts the loop. Instead the stream is detached: it keeps running in the background (bounded by the internode RPC timeout), the caller still gets its timeout error, and if the peer grants a lock after the caller gave up the client releases it immediately instead of leaving an orphan for the lease to expire.
+
+Unlocks that fail three quick retries no longer stop there. The background task continues with a deferred schedule (1s, 2s, 4s, 8s, 16s) before it gives up and leaves the entry to the server-side lease.
+
+## Configuration
+
+| Environment variable | Default | Behavior |
+| --- | ---: | --- |
+| `RUSTFS_OBJECT_LOCK_RPC_TIMEOUT_MS` | `3000` | Per-request deadline for remote lock RPCs. |
+| `RUSTFS_OBJECT_LOCK_RPC_EVICTION_COOLDOWN_MS` | `5000` | Minimum interval between channel evictions per peer. `0` restores eviction on every qualifying failure. |
+| `RUSTFS_OBJECT_LOCK_RPC_DETACHED_LIMIT` | `256` | How many timed-out lock RPCs per peer may keep running in the background. Beyond the budget a timed-out stream is cancelled as before. |
+
+## Metrics
+
+| Metric | Labels | Meaning |
+| --- | --- | --- |
+| `rustfs_remote_lock_rpc_timeouts_total` | `peer`, `op` | Remote lock RPCs that exceeded their deadline. |
+| `rustfs_remote_lock_channel_evictions_total` | `peer`, `trigger` | Cached channel evictions; `trigger` is `timeout` or `transport`. |
+| `rustfs_remote_lock_channel_evictions_suppressed_total` | `peer`, `verdict` | Failures that kept the channel; `verdict` is `peer_recently_served` or `cooling_down`. |
+| `rustfs_remote_lock_rpc_detached_total` | `op`, `outcome` | Timed-out RPCs left running (`detached`) or cancelled for budget (`aborted`). |
+| `rustfs_remote_lock_rpc_late_completions_total` | `op`, `outcome` | How detached RPCs ended (`success`, `error`, `join_error`). |
+| `rustfs_remote_lock_late_releases_total` | `outcome` | Releases of locks granted after their caller timed out (`released`, `partial`, `failed`). |
+
+## Reading an incident
+
+A healthy-but-slow endpoint now shows a rising `rustfs_remote_lock_rpc_timeouts_total{peer}` with `evictions_suppressed_total{verdict="peer_recently_served"}` and at most one eviction per cooldown. A dead endpoint shows `evictions_total{trigger="transport"}` once per cooldown while the connection re-dials. Sustained `GOAWAY too_many_resets` in the server log means detached streams are being cancelled, which only happens once `RUSTFS_OBJECT_LOCK_RPC_DETACHED_LIMIT` is exhausted; raise the limit or fix the slow lock service (`http_request_inflight_slow` on `NodeService/Lock` names the endpoint).
+
+The client code lives in `crates/ecstore/src/cluster/rpc/remote_locker.rs`; the deferred unlock schedule lives in `crates/lock/src/distributed_lock.rs`.