diff --git a/.config/e2e-nightly-selection.txt b/.config/e2e-nightly-selection.txt index 163d3235b..1bb5125c9 100644 --- a/.config/e2e-nightly-selection.txt +++ b/.config/e2e-nightly-selection.txt @@ -1,2 +1,2 @@ -sha256-darwin=a5665318c9bdc0947514fb7008ba1b83b114b739fac775c3c446f207058b7c7a -sha256-linux=45d80e1723de5d25bb5b81f3ef5c82f583efc3e4f036a8cd2bb99e4f1eca9e51 +sha256-darwin=364f2329a7b72eb9f1608dbe1a3af37af4095354014f3cbe23ca448492d89961 +sha256-linux=60983f1ebe7068cf660d473c5f76c76a650410ccc99d71934ddca7fd67607987 diff --git a/.config/e2e-repl-nightly-selection.txt b/.config/e2e-repl-nightly-selection.txt index b879da5f6..6626a2217 100644 --- a/.config/e2e-repl-nightly-selection.txt +++ b/.config/e2e-repl-nightly-selection.txt @@ -1 +1 @@ -sha256=0fe8408874ccec3620262a9812d67920ddd72dc9edf0e36e0d0aed3f8bad026e +sha256=0e338d305260229e17ccfb2adc48a6212dbdfea36a9ebfb5a4e0d38658e6cc45 diff --git a/CHANGELOG.md b/CHANGELOG.md index 2ffd8896f..40ec98e4e 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/crates/config/src/constants/object.rs b/crates/config/src/constants/object.rs index 448e7ae68..b96ae3cde 100644 --- a/crates/config/src/constants/object.rs +++ b/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 diff --git a/crates/ecstore/src/cluster/rpc/remote_locker.rs b/crates/ecstore/src/cluster/rpc/remote_locker.rs index e42d63c0c..121d1ec66 100644 --- a/crates/ecstore/src/cluster/rpc/remote_locker.rs +++ b/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(request: &mut Request) -> 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 = Option Pin + 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, + last_eviction: Option, + consecutive_timeouts: u32, + /// Timed-out RPCs still running in the background for this peer. + detached_rpcs: usize, +} + +fn lock_peer_channel_health() -> &'static Mutex> { + static HEALTH: OnceLock>> = OnceLock::new(); + HEALTH.get_or_init(Mutex::default) +} + +fn with_lock_peer_health(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 { + 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(&self, op: &'static str, resource_summary: &str, future: F) -> std::result::Result + 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( + &self, + op: &'static str, + resource_summary: &str, + handle: JoinHandle>, + late: LateCompletion, + ) { + 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> { + let client = self.clone(); + Some(Box::new(move |response: Response| { + 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) -> LateCompletion> { + let client = self.clone(); + Some(Box::new(move |response: Response| { + 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) { + 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( + &self, + op: &'static str, + resource_summary: &str, + deadline: Duration, + future: Fut, + late: LateCompletion, + ) -> std::result::Result where - F: std::future::Future>, + Fut: Future> + 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::>>()?, }); 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); diff --git a/crates/ecstore/src/core/pools.rs b/crates/ecstore/src/core/pools.rs index f70a65e99..daf895e60 100644 --- a/crates/ecstore/src/core/pools.rs +++ b/crates/ecstore/src/core/pools.rs @@ -4462,9 +4462,41 @@ pub(crate) struct PoolMetaWriteState { cluster_epoch: Option, 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, + /// 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, identity_initialized: Option, identity_fresh_bootstrap_nonce: Option, 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, + 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 { + 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, + /// Per pool: whether the replica decoded as a valid identity. + valid_replicas: Vec, } fn encode_pool_meta_identity(identity: PersistedPoolMetaIdentity) -> Result> { @@ -5531,6 +5672,17 @@ pub(crate) fn pool_meta_identity_initialized_for_test(data: &[u8]) -> Result Result> { + 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> { encode_pool_meta_identity(PersistedPoolMetaIdentity { @@ -5571,6 +5723,10 @@ fn select_pool_meta_identity( expected_cluster_id: uuid::Uuid, ) -> Result { 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 = 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( pools: Vec>, 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 = 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( + pools: Vec>, + 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, diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index 6b55c1d93..94441e98b 100644 --- a/crates/ecstore/src/store/init.rs +++ b/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( @@ -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> { + 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::() { + 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) { + 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(); diff --git a/crates/io-metrics/src/lock_metrics.rs b/crates/io-metrics/src/lock_metrics.rs index ae517ac75..f81016b9f 100644 --- a/crates/io-metrics/src/lock_metrics.rs +++ b/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 = 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"); } diff --git a/crates/lock/src/distributed_lock.rs b/crates/lock/src/distributed_lock.rs index 04d6aa804..1576d1fc3 100644 --- a/crates/lock/src/distributed_lock.rs +++ b/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)>, + attempt: usize, + context: &'static str, + ) -> Vec<(LockId, Arc)> { + 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)>, 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, + } + + #[async_trait::async_trait] + impl LockClient for FlakyReleaseClient { + async fn acquire_lock(&self, _request: &LockRequest) -> crate::Result { + Ok(LockResponse::failure("unused", Duration::ZERO)) + } + + async fn release(&self, _lock_id: &LockId) -> crate::Result { + 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 { + Ok(false) + } + + async fn force_release(&self, _lock_id: &LockId) -> crate::Result { + Ok(false) + } + + async fn check_status(&self, _lock_id: &LockId) -> crate::Result> { + Ok(None) + } + + async fn get_stats(&self) -> crate::Result { + 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 = 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 = 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, diff --git a/docs/operations/lock-rpc-storm-protection.md b/docs/operations/lock-rpc-storm-protection.md new file mode 100644 index 000000000..ae43745b6 --- /dev/null +++ b/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`.