From 603bdea51625439fe2b5ec125392b966d0db4d5e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=94=90=E5=B0=8F=E9=B8=AD?= Date: Tue, 11 Aug 2026 13:41:28 +0800 Subject: [PATCH] fix(site-replication): route state RMW through one locked transaction (#5882) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * test(site-replication): pin retry-event lost-update against locked RMW (red) P1-15 (rustfs/backlog#1675 B2): the site-replication retry-event writers (enqueue/dequeue, which hang off every hook broadcast path) perform a load -> mutate -> persist without taking SITE_REPLICATION_STATE_LOCK, so a single process can lose a concurrent lock-holding writer's update; the service-side reload path is equally unlocked, and no writer holds a distributed lock across the read-modify-write, so multi-node RMW loses updates even where the process lock is held. Red evidence (current main): replaying enqueue's exact three steps around a completed mark_pending_rotation_peer_acked commit wipes the rotation ack — the final state holds the retry event but not the ack. * fix(site-replication): route state RMW through one locked transaction P1-15 PR1 (rustfs/backlog#1675 B2). The site-replication state object (config/site-replication/state.json, which also carries the retry-event queue) was mutated through read-modify-write sequences with inconsistent locking: the retry-event writers on every hook broadcast path and the RPC-driven service reload took no lock at all (single-process lost updates, pinned by the red commit), and no writer held a distributed lock across the whole RMW (cross-node lost updates everywhere). - New admin/site_replication_state module: the state transaction boundary `with_site_replication_state_lock[_on]` — process mutex plus the distributed config-object write lock (the pattern proven by the repair state), with the shared path constant. The process mutex is transitional until PR2 migrates the remaining ~26 call sites. - handlers: typed `update_site_replication_state` (no-lock load / persist-or-clear inside the boundary; normalizes the peer map exactly once, retiring the double-clone/double-normalize persist path, P2-22). Migrated: retry-event enqueue (always-write), dequeue (lock-free probe, transaction on hit), mark_pending_rotation/remove_peer_acked. - service reload: the tolerant byte-level read->normalize->save now runs inside the same boundary via no-lock IO — a cluster-wide reload fan-out can no longer overwrite a concurrent state writer. Normalization semantics untouched (all six service-side tests unchanged and green). - Add/PeerJoin/Edit handlers release the state guard before their peer fan-out: the transport helpers' retry-event bookkeeping now re-enters the state transaction and must not nest inside the guard (the adversarial review caught this as a re-entrancy deadlock; the fix mirrors the Remove/Rotate handlers' existing scope). The Edit non- refresh branch commits before fanning out — the old fanout-first order recorded retry events pointing at a state the local site had not saved. - ecstore: delete_config_no_lock (+ facade/bridge exports) so the clear half of persist-or-clear works under the held object lock. Red -> green: the red commit pinned the deterministic lost-update interleaving (stale retry-event persist wiping a committed rotation ack); the test now drives the real functions concurrently for 8 rounds and asserts every retry event and every ack survives. Full handlers/service site-replication unit suites green (171 + 6); dual-node site-replication e2e (state edit fresh/stale, object replication) green; fmt / clippy / logging guardrails clean. Adversarial review: one blocking finding (the re-entrancy deadlock above) fixed and re-verified by a full second pass over all 30 lock sites and the Add/Join/Edit call graphs. Non-blocking notes recorded for PR2: mark_* now persists on miss (persist-or-clear semantics; a miss-skip return is a cheap follow-up), Add still holds the guard across the peer join probe (pre-existing availability debt), and a timeout-guarded unreachable-peer regression test for the fan-out paths. * fix(site-replication): keep the state mutex behind an owner helper CI's architecture migration guard lists SITE_REPLICATION_STATE_LOCK as an owner-local static, so it may not be `pub(crate)`. Keep it private to the new module and let the not-yet-migrated RMW call sites take it through `site_replication_state_process_guard()` — the sanctioned owner-helper pattern; the helper disappears with the mutex in PR2. * fix(site-replication): keep peer-edit delivery under the state guard Review follow-up (#5882). Releasing the guard before the fan-out (my deadlock fix) traded the ordering the guard used to provide: edit A could commit and stall while edit B committed and reached a peer first, then A arrived last and won. The peer edit handler applies whatever arrives — it has no generation or updated-at fence — and a successful stale delivery is not repaired by the retry queue, so the sites diverge silently. The fan-out is back under the guard. What actually could not run there is the retry-event bookkeeping, which re-enters the state transaction, so the edit branch now delivers with the plain transport and settles the retry queue after the guard is released: successes dequeue, the first failure enqueues and is returned. Ordering and bookkeeping both preserved. The add handler keeps its peer-edit finalize fan-out under the guard for the same reason and releases only before bootstrap/back-fill, which send bucket-ops (not peer edits) through retry-event transports. The concurrency test could not tell the two guards apart — both writers took both locks, so it passed with either removed. Replaced by two tests that isolate one guard each, both verified by mutation: - a process-only legacy writer (the shape the not-yet-migrated call sites still use) racing the transaction: fails when the transaction stops taking the process mutex; - two writers that bypass the process mutex, as separate nodes do, driving the production object-lock path (`with_site_replication_state_object_lock` factored out for exactly this): fails when the distributed lock is removed. Verification: handlers 173 + service 6 unit tests green; site-replication dual-node and three-node edit e2e green; arch/layer/logging guardrails, fmt and clippy clean. * fix(site-replication): fence peer-edit delivery by generation Review follow-up on the two remaining holes in the edit path. Ordering was only process-local. `SITE_REPLICATION_STATE_LOCK` is per node, so holding it across the fan-out orders the edits ONE node accepts and nothing else: two nodes of the same site can both commit and reach a peer in the opposite order, and the peer edit handler applied whatever arrived last. Each edit now takes a generation from `SiteReplicationState::edit_generation`, allocated in the same commit as the edit itself — i.e. under the distributed state-object lock, so two nodes can never share one. The generation rides the peer-edit request as query parameters and the receiver rejects (acks without applying) a delivery at or below the mark it already applied for that origin site, recording the mark in the same commit as the edit it fences. Peers that predate the fence send no parameters and are applied as before. Retry settlement could discard a newer failure. After the guard is released, a success for edit A removed every retry event for (peer, peer-edit): if edit B committed, failed its own delivery and enqueued while A was in flight, A erased it — local state B, peer on A, nothing queued to converge them. Settlement now only removes events whose recorded generation is not newer than the one being settled, and a later failure never lowers the fence. Broadcast paths carry no generation and settle unconditionally as before; their events live under their own paths and cannot collide with a peer-edit delivery. A departed peer's mark is dropped on load: a site that leaves drops below two peers, which clears its state object and restarts its counter at zero, so a leftover mark would reject every edit it sends after it rejoins. Tests: two-node generation uniqueness (drop the object lock and the two nodes collide), the receiver's staleness predicate and its wiring, the settlement interleaving (drop the fence and B's retry is erased), and the rejoin reset. Refs: rustfs/backlog#1675 (P1-15) --- crates/ecstore/src/api/mod.rs | 2 +- crates/ecstore/src/config/com.rs | 38 + rustfs/src/admin/handlers/site_replication.rs | 837 ++++++++++++++++-- rustfs/src/admin/mod.rs | 1 + rustfs/src/admin/service/site_replication.rs | 49 +- rustfs/src/admin/site_replication_state.rs | 110 +++ rustfs/src/storage/storage_api.rs | 4 + 7 files changed, 938 insertions(+), 103 deletions(-) create mode 100644 rustfs/src/admin/site_replication_state.rs diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index f7b930241..d376cb0b7 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -281,7 +281,7 @@ pub mod config { pub mod com { pub use crate::config::com::{ COMMA_SEPARATED_LISTS, CONFIG_PREFIX, ENV_CONFIG_RECOVER_ON_CORRUPTION, STORAGE_CLASS_SUB_SYS, - ServerConfigCorruptError, ServerConfigSaveResult, ServerConfigSnapshot, delete_config, + ServerConfigCorruptError, ServerConfigSaveResult, ServerConfigSnapshot, delete_config, delete_config_no_lock, is_server_config_corrupt_error, lookup_configs, read_config, read_config_no_lock, read_config_with_metadata, read_config_without_migrate, read_config_without_migrate_no_lock, read_existing_server_config_no_lock, read_server_config_snapshot, save_config, save_config_no_lock, save_config_with_opts, save_server_config, diff --git a/crates/ecstore/src/config/com.rs b/crates/ecstore/src/config/com.rs index 5498e570d..259c34f65 100644 --- a/crates/ecstore/src/config/com.rs +++ b/crates/ecstore/src/config/com.rs @@ -584,6 +584,44 @@ where .await } +/// `delete_config` with `no_lock` set — for callers already holding the +/// config object's namespace lock (e.g. inside `with_config_object_write_lock`), +/// where the locked variant would self-deadlock. +pub async fn delete_config_no_lock(api: Arc, file: &str) -> Result<()> +where + S: ObjectOperations< + Error = Error, + ObjectInfo = ObjectInfo, + ObjectOptions = ObjectOptions, + FileInfo = FileInfo, + ObjectToDelete = ObjectToDelete, + DeletedObject = DeletedObject, + >, +{ + match api + .delete_object( + RUSTFS_META_BUCKET, + file, + ObjectOptions { + delete_prefix: true, + delete_prefix_object: true, + no_lock: true, + ..Default::default() + }, + ) + .await + { + Ok(_) => Ok(()), + Err(err) => { + if err == Error::FileNotFound || matches!(err, Error::ObjectNotFound(_, _)) { + Err(Error::ConfigNotFound) + } else { + Err(err) + } + } + } +} + #[instrument(skip(api))] pub async fn delete_config(api: Arc, file: &str) -> Result<()> where diff --git a/rustfs/src/admin/handlers/site_replication.rs b/rustfs/src/admin/handlers/site_replication.rs index 10fd7f2fc..fde6890c5 100644 --- a/rustfs/src/admin/handlers/site_replication.rs +++ b/rustfs/src/admin/handlers/site_replication.rs @@ -46,7 +46,7 @@ use crate::config::get_config_snapshot; use crate::error::ApiError; use crate::server::{ADMIN_PREFIX, RemoteAddr}; use crate::storage::storage_api::{ - lock_bucket_targets_metadata, read_config_no_lock, save_config_no_lock, with_config_object_read_lock, + delete_config_no_lock, lock_bucket_targets_metadata, read_config_no_lock, save_config_no_lock, with_config_object_read_lock, with_config_object_write_lock, }; use base64::Engine; @@ -111,7 +111,11 @@ const LOG_COMPONENT_ADMIN: &str = "admin"; const LOG_SUBSYSTEM_SITE_REPLICATION: &str = "site_replication"; const EVENT_ADMIN_SITE_REPLICATION_STATE: &str = "admin_site_replication_state"; const SERVICE_ACCOUNT_ENVELOPE_VERSION: u64 = 2; -const SITE_REPLICATION_STATE_PATH: &str = "config/site-replication/state.json"; +#[cfg(test)] +use crate::admin::site_replication_state::with_site_replication_state_object_lock; +use crate::admin::site_replication_state::{ + SITE_REPLICATION_STATE_PATH, site_replication_state_process_guard, with_site_replication_state_lock, +}; const SITE_REPLICATION_REPAIR_STATE_PATH: &str = "config/site-replication/repair-state.json"; const SITE_REPLICATION_REPAIR_EXECUTION_LOCK_PATH: &str = "config/site-replication/repair-execution.lock"; const SITE_REPL_ADD_SUCCESS: &str = "Requested sites were configured for replication successfully."; @@ -147,6 +151,11 @@ const SITE_REPLICATION_PEER_EDIT_CAPABILITY_PATH: &str = const SITE_REPLICATION_PEER_TLS_CAPABILITY_PATH: &str = "/rustfs/admin/v3/site-replication/peer/edit-capabilities?capability=peer-tls-settings"; const SITE_REPLICATION_PEER_EDIT_REFRESH_PATH: &str = "/rustfs/admin/v3/site-replication/peer/edit?refresh-targets=true"; +/// Peer-edit fencing token, carried as query parameters so a peer that predates +/// the fence simply ignores them (unknown query keys are dropped) and keeps the +/// previous last-writer-wins behaviour. +const SITE_REPLICATION_EDIT_ORIGIN_QUERY: &str = "editOrigin"; +const SITE_REPLICATION_EDIT_GENERATION_QUERY: &str = "editGeneration"; const SITE_REPLICATION_ENDPOINT_REFRESH_RETRY_PATH: &str = "internal:endpoint-target-refresh"; const SITE_REPLICATION_PEER_REMOVE_PATH: &str = "/rustfs/admin/v3/site-replication/peer/remove"; const SITE_REPLICATION_DEVNULL_PATH: &str = "/rustfs/admin/v3/site-replication/devnull"; @@ -344,7 +353,8 @@ impl TryFrom<&PeerSite> for PeerConnection { static SITE_REPLICATION_PEER_CLIENT: LazyLock>> = LazyLock::new(|| Mutex::new(None)); // Lock order: lifecycle -> bucket operation -> repair admission -> state -> per-bucket metadata. -static SITE_REPLICATION_STATE_LOCK: LazyLock> = LazyLock::new(|| Mutex::new(())); +// The state mutex lives in crate::admin::site_replication_state and is +// taken through site_replication_state_process_guard (P1-15). static SITE_REPLICATION_LIFECYCLE_LOCK: LazyLock> = LazyLock::new(|| Mutex::new(())); static SITE_REPLICATION_BUCKET_OP_LOCK: LazyLock> = LazyLock::new(|| RwLock::new(())); static SITE_REPLICATION_ADD_BOOTSTRAP: LazyLock>> = @@ -457,6 +467,20 @@ struct SiteReplicationState { retry_queue: Vec, #[serde(default)] sync_state_initialized: bool, + /// Fencing token for peer-edit delivery, allocated inside the state + /// transaction (process mutex + distributed state-object lock). Two nodes + /// of THIS site that accept admin edits concurrently therefore get + /// strictly ordered generations even though the process mutex cannot + /// serialize them, and a delivery that stalls can be recognised as stale + /// by the receiving site. + #[serde(default)] + edit_generation: u64, + /// Per-origin high-water mark of the peer edits already applied here, + /// keyed by the origin site's deployment id. A delivery whose generation + /// is not above the mark arrived out of order and must not overwrite the + /// newer edit that already landed. + #[serde(default, skip_serializing_if = "BTreeMap::is_empty")] + applied_edit_generations: BTreeMap, } #[derive(Debug, Clone, Serialize, Deserialize, Default)] @@ -595,6 +619,12 @@ struct SiteReplicationRetryEvent { last_error: String, #[serde(default, with = "time::serde::rfc3339::option", skip_serializing_if = "Option::is_none")] updated_at: Option, + /// Peer-edit generation whose delivery failed, when the failing send + /// carried one. Settling a *later* success for the same (peer, path) must + /// not erase a failure recorded for a NEWER generation — see + /// [`settle_site_replication_retry_events`]. + #[serde(default, skip_serializing_if = "Option::is_none")] + edit_generation: Option, } #[derive(Debug, Clone, Serialize, Deserialize, Default)] @@ -1036,24 +1066,34 @@ fn parse_public_peer_edit(body: &[u8]) -> S3Result<(PeerInfo, PeerTlsFieldPresen Ok((parse_site_replication_json(body)?, parse_site_replication_json(body)?)) } +fn parse_site_replication_state(data: &[u8]) -> S3Result { + let mut state: SiteReplicationState = serde_json::from_slice(data) + .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("invalid site replication state: {e}")))?; + state.peers = normalize_peer_map_by_identity(state.peers); + // A peer-edit high-water mark only fences a CURRENT peer. A site that + // leaves drops below two peers, which clears its own state object and + // restarts its generation counter at zero — a mark left over from the + // previous membership would then reject every edit it sends after it + // rejoins. Dropping departed origins on load also keeps the map bounded. + state + .applied_edit_generations + .retain(|origin, _| state.peers.contains_key(origin)); + if !state.sync_state_initialized { + if state.enabled() { + mark_unknown_peer_sync_enabled(&mut state.peers); + } + state.sync_state_initialized = true; + } + Ok(state) +} + async fn load_site_replication_state() -> S3Result { let Some(store) = current_object_store_handle() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; match read_admin_config(store, SITE_REPLICATION_STATE_PATH).await { - Ok(data) => { - let mut state: SiteReplicationState = serde_json::from_slice(&data) - .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("invalid site replication state: {e}")))?; - state.peers = normalize_peer_map_by_identity(state.peers); - if !state.sync_state_initialized { - if state.enabled() { - mark_unknown_peer_sync_enabled(&mut state.peers); - } - state.sync_state_initialized = true; - } - Ok(state) - } + Ok(data) => parse_site_replication_state(&data), Err(StorageError::ConfigNotFound) => Ok(SiteReplicationState::default()), Err(err) => Err(S3Error::with_message( S3ErrorCode::InternalError, @@ -1062,6 +1102,78 @@ async fn load_site_replication_state() -> S3Result { } } +async fn load_site_replication_state_no_lock(store: Arc) -> S3Result { + match read_config_no_lock(store, SITE_REPLICATION_STATE_PATH).await { + Ok(data) => parse_site_replication_state(&data), + Err(StorageError::ConfigNotFound) => Ok(SiteReplicationState::default()), + Err(err) => Err(S3Error::with_message( + S3ErrorCode::InternalError, + format!("failed to load site replication state: {err}"), + )), + } +} + +/// Persist-or-clear under an already-held state object lock. Normalizes the +/// peer map exactly once (the historical persist path normalized twice with +/// two full clones — P2-22). +async fn persist_site_replication_state_no_lock(store: Arc, mut state: SiteReplicationState) -> S3Result<()> { + state.peers = normalize_peer_map_by_identity(state.peers); + if state.peers.len() <= 1 && state.pending_rotation.is_none() && state.pending_remove.is_none() { + match delete_config_no_lock(store, SITE_REPLICATION_STATE_PATH).await { + Ok(()) | Err(StorageError::ConfigNotFound) => Ok(()), + Err(err) => Err(S3Error::with_message(S3ErrorCode::InternalError, format!("clear state failed: {err}"))), + } + } else { + let data = serde_json::to_vec(&state) + .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("serialize state failed: {e}")))?; + save_config_no_lock(store, SITE_REPLICATION_STATE_PATH, data) + .await + .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("save state failed: {e}"))) + } +} + +/// A second node's view of the same transaction: identical production code +/// path minus the process mutex, which is per process and therefore cannot +/// serialize anything across nodes. Used by the separate-nodes regression +/// test so that removing the distributed lock breaks it. +#[cfg(test)] +async fn update_site_replication_state_as_separate_node(update: F) -> S3Result +where + T: Send + 'static, + F: FnOnce(&mut SiteReplicationState) -> S3Result + Send + 'static, +{ + let store = + current_object_store_handle().ok_or_else(|| S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()))?; + let lock_store = store.clone(); + with_site_replication_state_object_lock(lock_store, move || async move { + let mut state = load_site_replication_state_no_lock(store.clone()).await?; + let result = update(&mut state)?; + persist_site_replication_state_no_lock(store, state).await?; + Ok(result) + }) + .await +} + +/// The site-replication state RMW transaction: load, mutate, persist — all +/// under the process mutex plus the distributed state-object write lock +/// (see crate::admin::site_replication_state). No peer network calls and no +/// other config locks inside `update`. +async fn update_site_replication_state(update: F) -> S3Result +where + T: Send + 'static, + F: FnOnce(&mut SiteReplicationState) -> S3Result + Send + 'static, +{ + with_site_replication_state_lock(move || async move { + let store = current_object_store_handle() + .ok_or_else(|| S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()))?; + let mut state = load_site_replication_state_no_lock(store.clone()).await?; + let result = update(&mut state)?; + persist_site_replication_state_no_lock(store, state).await?; + Ok(result) + }) + .await +} + async fn load_site_replication_repair_state_from_store(store: Arc) -> S3Result { match read_config_no_lock(store, SITE_REPLICATION_REPAIR_STATE_PATH).await { Ok(data) => serde_json::from_slice(&data).map_err(|e| { @@ -2633,7 +2745,7 @@ fn is_missing_service_account_error(err: &rustfs_iam::error::Error) -> bool { /// disables every control-plane push while `replicate info` still reports the site enabled. /// Both used to require deleting and recreating the account by hand. async fn reconcile_site_replicator_service_account() -> S3Result<()> { - let _state_guard = SITE_REPLICATION_STATE_LOCK.lock().await; + let _state_guard = site_replication_state_process_guard().await; let state = load_site_replication_state().await?; if !state.enabled() || state.service_account_access_key != SITE_REPLICATOR_SERVICE_ACCOUNT { return Ok(()); @@ -3805,7 +3917,7 @@ async fn persist_site_replication_repair_task( ) -> S3Result<()> { persist_site_replication_repair_operation(operation).await?; - let _state_guard = SITE_REPLICATION_STATE_LOCK.lock().await; + let _state_guard = site_replication_state_process_guard().await; let mut latest = load_site_replication_state().await?; let family_status = operation .sites @@ -3822,6 +3934,7 @@ async fn persist_site_replication_repair_task( .first() .map(String::as_str) .unwrap_or("remote-operation-failed"), + None, ); } else { dequeue_site_replication_retry_events(&mut latest.retry_queue, peer, path); @@ -3883,7 +3996,7 @@ async fn execute_site_replication_repair_locked( request: SiteReplicationRepairExecutionRequest, ) -> S3Result> { let state = { - let _state_guard = SITE_REPLICATION_STATE_LOCK.lock().await; + let _state_guard = site_replication_state_process_guard().await; let state = load_site_replication_state().await?; if !state.enabled() || state.service_account_access_key.is_empty() { return Err(s3_error!(InvalidRequest, "site replication is not configured")); @@ -4006,7 +4119,7 @@ async fn execute_site_replication_repair_locked( pub async fn site_replication_make_bucket_hook(bucket: &str, lock_enabled: bool) -> S3Result<()> { let _bucket_op_guard = SITE_REPLICATION_BUCKET_OP_LOCK.read().await; let runtime = { - let _state_guard = SITE_REPLICATION_STATE_LOCK.lock().await; + let _state_guard = site_replication_state_process_guard().await; let Some(runtime) = runtime_site_replication_targets().await? else { return Ok(()); }; @@ -5215,6 +5328,7 @@ fn set_pending_endpoint_refresh(state: &mut SiteReplicationState, pending: Pendi failed: false, last_error: "endpoint target refresh pending".to_string(), updated_at: Some(OffsetDateTime::now_utc()), + edit_generation: None, }); state.pending_endpoint_refresh = Some(pending); Ok(()) @@ -5783,17 +5897,99 @@ fn summarize_peer_error_detail(detail: &str) -> String { summary } +/// Allocate the next peer-edit generation. Called inside the state +/// transaction, so the counter is handed out under the distributed +/// state-object lock and two nodes of this site can never take the same one. +fn next_peer_edit_generation(state: &mut SiteReplicationState) -> u64 { + state.edit_generation = state.edit_generation.saturating_add(1); + state.edit_generation +} + +/// Build the peer-edit request path carrying the fencing token. The bare +/// constant stays the retry-queue key: the query only fences the wire +/// delivery, and a per-generation key would make every retry event unique. +/// Without a local deployment id there is nothing to fence against, so the +/// unstamped path is sent and the receiver keeps its pre-fence behaviour. +fn peer_edit_path_with_fence(origin: Option<&str>, generation: u64) -> String { + let Some(origin) = origin.filter(|origin| !origin.is_empty()) else { + return SITE_REPLICATION_PEER_EDIT_PATH.to_string(); + }; + let query = form_urlencoded::Serializer::new(String::new()) + .append_pair(SITE_REPLICATION_EDIT_ORIGIN_QUERY, origin) + .append_pair(SITE_REPLICATION_EDIT_GENERATION_QUERY, &generation.to_string()) + .finish(); + format!("{SITE_REPLICATION_PEER_EDIT_PATH}?{query}") +} + +/// The (origin site, generation) fence an incoming peer edit carries, when the +/// sender stamped one. An unstamped edit (older peer) has no fence and is +/// applied as before. +fn peer_edit_fence(queries: &HashMap) -> Option<(String, u64)> { + let origin = queries + .get(SITE_REPLICATION_EDIT_ORIGIN_QUERY) + .filter(|origin| !origin.is_empty())?; + let generation = queries.get(SITE_REPLICATION_EDIT_GENERATION_QUERY)?.parse::().ok()?; + Some((origin.clone(), generation)) +} + +/// True when a newer edit from the same origin site already landed here. The +/// process mutex on the sending node cannot order deliveries issued by two +/// nodes of that site, so ordering is decided here, on the generation the +/// sender allocated under the distributed lock. +fn peer_edit_delivery_is_stale(state: &SiteReplicationState, origin: &str, generation: u64) -> bool { + state + .applied_edit_generations + .get(origin) + .is_some_and(|applied| *applied >= generation) +} + +fn record_applied_peer_edit_generation(state: &mut SiteReplicationState, origin: &str, generation: u64) { + let applied = state.applied_edit_generations.entry(origin.to_string()).or_default(); + *applied = (*applied).max(generation); +} + fn retry_event_matches(event: &SiteReplicationRetryEvent, peer: &PeerInfo, path: &str) -> bool { (event.peer_deployment_id == peer.deployment_id || event.peer_endpoint == peer.endpoint) && event.path == path } fn dequeue_site_replication_retry_events(queue: &mut Vec, peer: &PeerInfo, path: &str) -> usize { + settle_site_replication_retry_events(queue, peer, path, None) +} + +/// Remove the retry events for (peer, path) that `generation` is entitled to +/// settle. A successful delivery only proves the peer reached the state the +/// delivery carried: while it was in flight another edit can commit, fail its +/// own delivery, and enqueue for the same (peer, path). Erasing that event +/// would leave the peer on the older edit with no retry left, so an event +/// stamped with a NEWER generation survives. `None` settles unconditionally — +/// the broadcast paths that carry no generation, whose retry events live under +/// their own paths and never collide with peer-edit deliveries. +fn settle_site_replication_retry_events( + queue: &mut Vec, + peer: &PeerInfo, + path: &str, + generation: Option, +) -> usize { let before = queue.len(); - queue.retain(|event| !retry_event_matches(event, peer, path)); + queue.retain(|event| { + if !retry_event_matches(event, peer, path) { + return true; + } + match (generation, event.edit_generation) { + (Some(settled), Some(failed)) => failed > settled, + _ => false, + } + }); before.saturating_sub(queue.len()) } -fn upsert_site_replication_retry_event(queue: &mut Vec, peer: &PeerInfo, path: &str, error: &str) { +fn upsert_site_replication_retry_event( + queue: &mut Vec, + peer: &PeerInfo, + path: &str, + error: &str, + generation: Option, +) { let now = OffsetDateTime::now_utc(); let detail = summarize_peer_error_detail(error); if let Some(event) = queue.iter_mut().find(|event| retry_event_matches(event, peer, path)) { @@ -5801,6 +5997,9 @@ fn upsert_site_replication_retry_event(queue: &mut Vec= SITE_REPLICATION_RETRY_FAILED_AFTER; event.last_error = detail; event.updated_at = Some(now); + // Keep the newest generation: an older delivery that fails afterwards + // must not lower the fence and let its own success settle the event. + event.edit_generation = event.edit_generation.max(generation); return; } @@ -5813,6 +6012,7 @@ fn upsert_site_replication_retry_event(queue: &mut Vec SITE_REPLICATION_RETRY_QUEUE_LIMIT { let overflow = queue.len() - SITE_REPLICATION_RETRY_QUEUE_LIMIT; @@ -5839,11 +6039,22 @@ fn retry_stats_for_state(state: &SiteReplicationState) -> Option { } async fn enqueue_site_replication_retry_event(peer: &PeerInfo, path: &str, error: &S3Error) { - let result = async { - let mut state = load_site_replication_state().await?; - upsert_site_replication_retry_event(&mut state.retry_queue, peer, path, &error.to_string()); - persist_site_replication_state(&state).await - } + enqueue_site_replication_retry_event_for_generation(peer, path, error, None).await +} + +async fn enqueue_site_replication_retry_event_for_generation( + peer: &PeerInfo, + path: &str, + error: &S3Error, + generation: Option, +) { + let peer_owned = peer.clone(); + let path_owned = path.to_string(); + let error_text = error.to_string(); + let result = update_site_replication_state(move |state| { + upsert_site_replication_retry_event(&mut state.retry_queue, &peer_owned, &path_owned, &error_text, generation); + Ok(()) + }) .await; if let Err(err) = result { @@ -5879,11 +6090,25 @@ fn retry_event_replayed_by_bootstrap(event: &SiteReplicationRetryEvent) -> bool /// This is a no-op (load + no-op persist skipped) when no matching entry exists, /// avoiding unnecessary I/O on the common path. async fn dequeue_site_replication_retry_event(peer: &PeerInfo, path: &str) { + dequeue_site_replication_retry_event_for_generation(peer, path, None).await +} + +async fn dequeue_site_replication_retry_event_for_generation(peer: &PeerInfo, path: &str, generation: Option) { let result = async { - let mut state = load_site_replication_state().await?; - if dequeue_site_replication_retry_events(&mut state.retry_queue, peer, path) > 0 { - persist_site_replication_state(&state).await?; + // Fast path: this sits on every successful hook broadcast, so probe + // with a plain read first and only enter the locked RMW on a hit + // (the transaction re-checks under the lock). + let mut probe = load_site_replication_state().await?; + if settle_site_replication_retry_events(&mut probe.retry_queue, peer, path, generation) == 0 { + return Ok(()); } + let peer_owned = peer.clone(); + let path_owned = path.to_string(); + update_site_replication_state(move |state| { + settle_site_replication_retry_events(&mut state.retry_queue, &peer_owned, &path_owned, generation); + Ok(()) + }) + .await?; Ok::<_, S3Error>(()) } .await; @@ -5987,7 +6212,7 @@ async fn record_pending_rotation_secret_candidate(rotation_id: &str, secret: Str return Ok(()); } - let _state_guard = SITE_REPLICATION_STATE_LOCK.lock().await; + let _state_guard = site_replication_state_process_guard().await; let mut state = load_site_replication_state().await?; if let Some(pending) = state.pending_rotation.as_mut() && pending.id == rotation_id @@ -6003,7 +6228,7 @@ async fn record_pending_remove_secret_candidate(remove_id: &str, secret: String) return Ok(()); } - let _state_guard = SITE_REPLICATION_STATE_LOCK.lock().await; + let _state_guard = site_replication_state_process_guard().await; let mut state = load_site_replication_state().await?; if let Some(pending) = state.pending_remove.as_mut() && pending.id == remove_id @@ -6015,31 +6240,35 @@ async fn record_pending_remove_secret_candidate(remove_id: &str, secret: String) } async fn mark_pending_rotation_peer_acked(rotation_id: &str, deployment_id: &str) -> S3Result<()> { - let _state_guard = SITE_REPLICATION_STATE_LOCK.lock().await; - let mut state = load_site_replication_state().await?; - if let Some(pending) = state.pending_rotation.as_mut() - && pending.id == rotation_id - { - pending.acked_deployment_ids.insert(deployment_id.to_string()); - save_site_replication_state(&state).await?; - } - Ok(()) + let rotation_id = rotation_id.to_string(); + let deployment_id = deployment_id.to_string(); + update_site_replication_state(move |state| { + if let Some(pending) = state.pending_rotation.as_mut() + && pending.id == rotation_id + { + pending.acked_deployment_ids.insert(deployment_id); + } + Ok(()) + }) + .await } async fn mark_pending_remove_peer_acked(remove_id: &str, deployment_id: &str) -> S3Result<()> { - let _state_guard = SITE_REPLICATION_STATE_LOCK.lock().await; - let mut state = load_site_replication_state().await?; - if let Some(pending) = state.pending_remove.as_mut() - && pending.id == remove_id - { - pending.acked_deployment_ids.insert(deployment_id.to_string()); - save_site_replication_state(&state).await?; - } - Ok(()) + let remove_id = remove_id.to_string(); + let deployment_id = deployment_id.to_string(); + update_site_replication_state(move |state| { + if let Some(pending) = state.pending_remove.as_mut() + && pending.id == remove_id + { + pending.acked_deployment_ids.insert(deployment_id); + } + Ok(()) + }) + .await } async fn finalize_pending_rotation_if_complete(rotation_id: &str, local_peer: &PeerInfo) -> S3Result { - let _state_guard = SITE_REPLICATION_STATE_LOCK.lock().await; + let _state_guard = site_replication_state_process_guard().await; let mut state = load_site_replication_state().await?; let Some(pending) = state.pending_rotation.as_ref() else { return Ok(true); @@ -6057,7 +6286,7 @@ async fn finalize_pending_rotation_if_complete(rotation_id: &str, local_peer: &P } async fn pending_remove_ready_to_finalize(remove_id: &str, local_peer: &PeerInfo) -> S3Result> { - let _state_guard = SITE_REPLICATION_STATE_LOCK.lock().await; + let _state_guard = site_replication_state_process_guard().await; let state = load_site_replication_state().await?; let Some(pending) = state.pending_remove.as_ref() else { return Ok(None); @@ -6073,7 +6302,7 @@ async fn pending_remove_ready_to_finalize(remove_id: &str, local_peer: &PeerInfo } async fn clear_pending_remove(remove_id: &str) -> S3Result<()> { - let _state_guard = SITE_REPLICATION_STATE_LOCK.lock().await; + let _state_guard = site_replication_state_process_guard().await; let mut state = load_site_replication_state().await?; if state .pending_remove @@ -7152,7 +7381,7 @@ async fn refresh_bucket_targets_after_endpoint_edit(pending_id: &str, service_ac let expected_incarnation_id = metadata_sys::capture_bucket_metadata_incarnation(&bucket.name) .await .map_err(ApiError::from)?; - let _state_guard = SITE_REPLICATION_STATE_LOCK.lock().await; + let _state_guard = site_replication_state_process_guard().await; let state = load_site_replication_state().await?; let Some(pending) = pending_endpoint_refresh(&state).filter(|pending| pending.id == pending_id) else { return Err(s3_error!(InvalidRequest, "endpoint target refresh state changed during update")); @@ -7473,7 +7702,7 @@ async fn refresh_site_resync_status(mut status: SRResyncOpStatus, peer: &PeerInf } async fn persist_site_resync_status(peer_id: &str, status: &SRResyncOpStatus) -> S3Result<()> { - let _state_guard = SITE_REPLICATION_STATE_LOCK.lock().await; + let _state_guard = site_replication_state_process_guard().await; let mut state = load_site_replication_state().await?; if state .resync_status @@ -7487,7 +7716,7 @@ async fn persist_site_resync_status(peer_id: &str, status: &SRResyncOpStatus) -> } async fn persist_new_site_resync_status(peer_id: &str, status: &SRResyncOpStatus) -> S3Result<()> { - let _state_guard = SITE_REPLICATION_STATE_LOCK.lock().await; + let _state_guard = site_replication_state_process_guard().await; let mut state = load_site_replication_state().await?; if state.resync_status.get(peer_id).is_some_and(site_resync_is_active) { return Err(s3_error!(InvalidRequest, "site replication resync is already active")); @@ -8103,7 +8332,7 @@ impl Operation for SiteReplicationAddHandler { reject_site_replicator_on_public_admin(&cred)?; let replicate_ilm_expiry = sr_add_replicate_ilm_expiry(&req.uri); let lifecycle_guard = SiteReplicationLifecycleGuard::acquire().await; - let state_guard = SITE_REPLICATION_STATE_LOCK.lock().await; + let state_guard = site_replication_state_process_guard().await; let current_state = load_site_replication_state().await?; if pending_endpoint_refresh(¤t_state).is_some() { return Err(s3_error!(InvalidRequest, "endpoint target refresh is pending")); @@ -8126,7 +8355,7 @@ impl Operation for SiteReplicationAddHandler { let expected_updated_at = current_state.updated_at; drop(state_guard); require_add_peer_tls_capability(&sites, &local_peer).await?; - let _state_guard = SITE_REPLICATION_STATE_LOCK.lock().await; + let _state_guard = site_replication_state_process_guard().await; let latest_state = load_site_replication_state().await?; if latest_state.updated_at != expected_updated_at || pending_endpoint_refresh(&latest_state).is_some() { return Err(s3_error!(InvalidRequest, "site replication state changed during capability probe")); @@ -8207,6 +8436,11 @@ impl Operation for SiteReplicationAddHandler { mark_unknown_peer_sync_enabled(&mut state.peers); persist_site_replication_state(&state).await?; + // The finalize fan-out below delivers peer-edit payloads, so it stays + // under the state guard: ordering against a concurrent edit matters + // and the peer edit handler has no generation fence. It uses the + // plain transport (no retry-event bookkeeping), so nothing re-enters + // the state transaction while the guard is held. for target in state.peers.values() { if target.deployment_id == local_peer.deployment_id || same_identity_endpoint(&target.endpoint, &local_peer.endpoint) { @@ -8236,6 +8470,12 @@ impl Operation for SiteReplicationAddHandler { } } + // Bootstrap and back-fill send bucket-ops, not peer edits, so their + // ordering is not state-sensitive — and their transports do record + // retry events, which re-enter the state transaction. Release the + // guard before them. + drop(_state_guard); + initial_sync_errors.extend(bootstrap_existing_metadata_after_add(&state, &local_peer, &service_account_secret_key).await); // Fix 1: back-fill pre-existing buckets so objects created before `replicate add` @@ -8263,7 +8503,7 @@ impl Operation for SiteReplicationRemoveHandler { let _lifecycle_guard = SiteReplicationLifecycleGuard::acquire().await; let (pending_remove, local_peer) = { let _bucket_op_guard = SITE_REPLICATION_BUCKET_OP_LOCK.write().await; - let _state_guard = SITE_REPLICATION_STATE_LOCK.lock().await; + let _state_guard = site_replication_state_process_guard().await; let current_state = load_site_replication_state().await?; if pending_endpoint_refresh(¤t_state).is_some() { return Err(s3_error!(InvalidRequest, "endpoint target refresh is pending")); @@ -8469,7 +8709,7 @@ impl Operation for SRPeerJoinHandler { async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { let cred = validate_site_replication_admin_request(&req, AdminAction::SiteReplicationAddAction).await?; let bootstrap_token = site_replication_bootstrap_token(&req.uri); - let _state_guard = SITE_REPLICATION_STATE_LOCK.lock().await; + let _state_guard = site_replication_state_process_guard().await; let mut state = load_site_replication_state().await?; let local_peer = current_local_peer(&req, &state); let join_envelope: SRPeerJoinEnvelope = read_site_replication_json(req, &cred.secret_key, true).await?; @@ -8556,6 +8796,10 @@ impl Operation for SRPeerJoinHandler { .filter(|name| !name.is_empty()) .unwrap_or_else(|| local_peer.name.clone()); persist_site_replication_state(&state).await?; + // Committed; release the state lock before the reverse-reachability + // probe and bucket back-fill — their transport helpers' retry-event + // bookkeeping re-enters the state transaction (P1-15). + drop(_state_guard); // Fix 1 (receiving side): ensure the joining peer also sets up replication for any // buckets it already owns so the reverse direction works from the start. Per-bucket // failures are logged (BUG2) so a reverse-direction back-fill gap is observable. @@ -8741,7 +8985,7 @@ impl Operation for SiteReplicationEditHandler { let ilm_expiry_override = sr_edit_ilm_expiry_override(&req.uri); let body = read_site_replication_body(req, &cred.secret_key, true).await?; let (mut incoming, tls_presence) = parse_public_peer_edit(&body)?; - let mut state_guard = Some(SITE_REPLICATION_STATE_LOCK.lock().await); + let mut state_guard = Some(site_replication_state_process_guard().await); let current_state = load_site_replication_state().await?; apply_public_peer_edit_tls_presence(¤t_state, &mut incoming, tls_presence); if !incoming.deployment_id.is_empty() || !incoming.endpoint.is_empty() || !incoming.name.is_empty() { @@ -8788,7 +9032,7 @@ impl Operation for SiteReplicationEditHandler { if tls_transport_probe_required { probe_proposed_peer_tls_transport(&incoming, ¤t_state.service_account_access_key, &secret).await?; } - state_guard = Some(SITE_REPLICATION_STATE_LOCK.lock().await); + state_guard = Some(site_replication_state_process_guard().await); let latest_state = load_site_replication_state().await?; if latest_state.updated_at != expected_updated_at || pending_endpoint_refresh(&latest_state).as_ref().map(|pending| &pending.id) @@ -8860,7 +9104,7 @@ impl Operation for SiteReplicationEditHandler { } } - state_guard = Some(SITE_REPLICATION_STATE_LOCK.lock().await); + state_guard = Some(site_replication_state_process_guard().await); let latest_state = load_site_replication_state().await?; if latest_state.updated_at != expected_updated_at || pending_endpoint_refresh(&latest_state).as_ref().map(|pending| &pending.id) @@ -8913,7 +9157,7 @@ impl Operation for SiteReplicationEditHandler { } } - let _state_guard = SITE_REPLICATION_STATE_LOCK.lock().await; + let _state_guard = site_replication_state_process_guard().await; let mut state = load_site_replication_state().await?; let Some(pending) = pending_endpoint_refresh(&state).filter(|pending| pending.id == pending_id) else { return Err(s3_error!(InvalidRequest, "endpoint target refresh state changed during update")); @@ -8928,7 +9172,7 @@ impl Operation for SiteReplicationEditHandler { site_replicator_service_account_secret(&state.service_account_access_key).await?; drop(_state_guard); refresh_bucket_targets_after_endpoint_edit(&pending_id, &service_account_secret_key).await?; - let _state_guard = SITE_REPLICATION_STATE_LOCK.lock().await; + let _state_guard = site_replication_state_process_guard().await; let mut state = load_site_replication_state().await?; let Some(pending) = pending_endpoint_refresh(&state).filter(|pending| pending.id == pending_id) else { return Err(s3_error!(InvalidRequest, "endpoint target refresh state changed during update")); @@ -8937,21 +9181,67 @@ impl Operation for SiteReplicationEditHandler { clear_pending_endpoint_refresh(&mut state); save_site_replication_state(&state).await?; } else { - for target in remote_targets { + // Commit before the peer fan-out (mirrors the add/join + // handlers): a failed notification is recorded as a retry + // event and converges from the committed local state — + // fanning out first meant the retry event pointed at a state + // the local site had not saved. The generation is allocated in + // the same commit, i.e. under the state-object lock, so it + // orders this edit against one another node accepts + // concurrently — the process guard below cannot. + let edit_generation = next_peer_edit_generation(&mut state); + save_site_replication_state(&state).await?; + let edit_path = peer_edit_path_with_fence(local_deployment_id.as_deref(), edit_generation); + let delivery_fence = local_deployment_id.is_some().then_some(edit_generation); + + // The fan-out stays UNDER the state guard so deliveries issued + // by THIS node keep their commit order; the generation fence + // above is what covers the cross-node case, where the peer + // rejects a delivery an older edit is still trying to make. + // The retry-event bookkeeping is what cannot run under the + // guard — it re-enters the state transaction (P1-15) — so + // deliver with the plain transport here and settle the retry + // queue after the guard is released. + let mut delivered: Vec = Vec::new(); + let mut failure: Option<(PeerInfo, S3Error)> = None; + 'fanout: for target in remote_targets { let transport = PeerTransport::for_runtime_peer(target).await?; for peer in &peers_to_send { - send_peer_admin_request_with_retry_event_transport( - target, - &transport, - SITE_REPLICATION_PEER_EDIT_PATH, + if let Err(err) = send_peer_admin_request_with_client( + &transport.client, + &transport.connection, + &edit_path, ¤t_state.service_account_access_key, &service_account_secret_key, peer, ) - .await?; + .await + { + failure = Some((target.clone(), err)); + break 'fanout; + } } + delivered.push(target.clone()); + } + drop(state_guard.take()); + + // Settle only what this generation is entitled to: a newer + // edit that committed and failed its own delivery while this + // fan-out was in flight left a retry event that must survive. + for target in &delivered { + dequeue_site_replication_retry_event_for_generation(target, SITE_REPLICATION_PEER_EDIT_PATH, delivery_fence) + .await; + } + if let Some((target, err)) = failure { + enqueue_site_replication_retry_event_for_generation( + &target, + SITE_REPLICATION_PEER_EDIT_PATH, + &err, + delivery_fence, + ) + .await; + return Err(err); } - save_site_replication_state(&state).await?; } } @@ -8990,8 +9280,24 @@ impl Operation for SRPeerEditHandler { let queries = query_pairs(&req.uri); let ilm_expiry_override = sr_edit_ilm_expiry_override(&req.uri); let endpoint_refresh_requested = queries.get("refresh-targets").is_some_and(|value| value == "true"); - let state_guard = SITE_REPLICATION_STATE_LOCK.lock().await; + let delivery_fence = peer_edit_fence(&queries); + let state_guard = site_replication_state_process_guard().await; let state = load_site_replication_state().await?; + // Ordering fence: the sending site allocates the generation under its + // state-object lock, so a delivery that lost the race carries a + // generation this site has already passed. Applying it would roll the + // peer back to the older edit. Ack it — the newer edit already landed, + // so the sender has nothing to retry. + if let Some((origin, generation)) = delivery_fence.as_ref() + && peer_edit_delivery_is_stale(&state, origin, *generation) + { + return json_response(&ReplicateEditStatus { + success: true, + status: SITE_REPL_EDIT_SUCCESS.to_string(), + api_version: Some(SITE_REPL_API_VERSION.to_string()), + ..Default::default() + }); + } if endpoint_refresh_requested && (state.pending_rotation.is_some() || state.pending_remove.is_some()) { return json_response(&ReplicateEditStatus { success: false, @@ -9060,6 +9366,12 @@ impl Operation for SRPeerEditHandler { }, )?; } + // Raise the origin's high-water mark in the same commit as the edit it + // fences: a crash between the two would let the superseded delivery + // apply on the next attempt. + if let Some((origin, generation)) = delivery_fence.as_ref() { + record_applied_peer_edit_generation(&mut state, origin, *generation); + } save_site_replication_state(&state).await?; if endpoint_refresh_requested { if state.service_account_access_key.is_empty() { @@ -9074,7 +9386,7 @@ impl Operation for SRPeerEditHandler { let pending_id = refresh_id.unwrap_or_default(); drop(state_guard); refresh_bucket_targets_after_endpoint_edit(&pending_id, &service_account_secret_key).await?; - let _state_guard = SITE_REPLICATION_STATE_LOCK.lock().await; + let _state_guard = site_replication_state_process_guard().await; let mut state = load_site_replication_state().await?; let Some(pending) = pending_endpoint_refresh(&state).filter(|pending| pending.id == pending_id) else { return json_response(&ReplicateEditStatus { @@ -9107,7 +9419,7 @@ impl Operation for SRPeerRemoveHandler { let remove_req: SRRemoveReq = read_site_replication_json(req, "", false).await?; let _lifecycle_guard = SiteReplicationLifecycleGuard::acquire().await; let _bucket_op_guard = SITE_REPLICATION_BUCKET_OP_LOCK.write().await; - let _state_guard = SITE_REPLICATION_STATE_LOCK.lock().await; + let _state_guard = site_replication_state_process_guard().await; let current_state = load_site_replication_state().await?; if pending_endpoint_refresh(¤t_state).is_some() { return Err(s3_error!(InvalidRequest, "endpoint target refresh is pending")); @@ -9151,7 +9463,7 @@ impl Operation for SiteReplicationResyncOpHandler { let requested_peer: PeerInfo = read_site_replication_json(req, "", false).await?; let _lifecycle_guard = SiteReplicationLifecycleGuard::acquire().await; let (peer, existing_status) = { - let _state_guard = SITE_REPLICATION_STATE_LOCK.lock().await; + let _state_guard = site_replication_state_process_guard().await; let state = load_site_replication_state().await?; let local_peer = current_local_runtime_peer(&state); let requested_peer = normalize_peer_info(requested_peer); @@ -9312,7 +9624,7 @@ impl Operation for SRStateEditHandler { let cred = validate_site_replication_admin_request(&req, AdminAction::SiteReplicationOperationAction).await?; reject_site_replicator_on_public_admin(&cred)?; let body: SRStateEditReq = read_site_replication_json(req, "", false).await?; - let _state_guard = SITE_REPLICATION_STATE_LOCK.lock().await; + let _state_guard = site_replication_state_process_guard().await; let state = apply_state_edit_req(load_site_replication_state().await?, body); save_site_replication_state(&state).await?; Ok(empty_response(StatusCode::OK)) @@ -9327,7 +9639,7 @@ impl Operation for SiteReplicationRepairHandler { let cred = validate_site_replication_admin_request(&req, AdminAction::SiteReplicationOperationAction).await?; reject_site_replicator_on_public_admin(&cred)?; let (state, local_peer) = { - let _state_guard = SITE_REPLICATION_STATE_LOCK.lock().await; + let _state_guard = site_replication_state_process_guard().await; let state = load_site_replication_state().await?; if !state.enabled() || state.service_account_access_key.is_empty() { return Err(s3_error!(InvalidRequest, "site replication is not configured")); @@ -9438,7 +9750,7 @@ impl Operation for SRRotateServiceAccountHandler { let cred = validate_site_replication_admin_request(&req, AdminAction::SiteReplicationOperationAction).await?; reject_site_replicator_on_public_admin(&cred)?; let (pending_rotation, local_peer, previous_access_key) = { - let _state_guard = SITE_REPLICATION_STATE_LOCK.lock().await; + let _state_guard = site_replication_state_process_guard().await; let mut state = load_site_replication_state().await?; if !state.enabled() { return Err(s3_error!(InvalidRequest, "site replication is not configured")); @@ -11306,6 +11618,28 @@ mod tests { .contains("validate_site_replication_admin_request(&req, AdminAction::SiteReplicationAddAction).await?;"), "SRPeerEditHandler must not require SiteReplicationAddAction for internal peer edits" ); + // P1-15 review follow-up: the ordering fence is only worth anything if + // the handler both rejects a superseded delivery and raises the mark it + // rejects against — dropping either half silently restores + // last-writer-wins between two nodes of the sending site. + assert!( + handler_block.contains("peer_edit_delivery_is_stale(&state, origin, *generation)"), + "SRPeerEditHandler must reject peer edits a newer generation already superseded" + ); + assert!( + handler_block.contains("record_applied_peer_edit_generation(&mut state, origin, *generation);"), + "SRPeerEditHandler must record the applied generation so later stale deliveries are recognised" + ); + + let sender_block = src + .split("impl Operation for SiteReplicationEditHandler") + .nth(1) + .and_then(|rest| rest.split("pub struct SRPeerEditCapabilitiesHandler").next()) + .expect("SiteReplicationEditHandler block should exist"); + assert!( + sender_block.contains("let edit_generation = next_peer_edit_generation(&mut state);"), + "the edit handler must allocate the generation inside the committed state, not outside the lock" + ); } #[test] @@ -11582,14 +11916,14 @@ mod tests { let lifecycle = SiteReplicationLifecycleGuard::acquire().await; let add_guard = SiteReplicationAddInProgressGuard::start(lifecycle, HashSet::new()).expect("start site replication add guard"); - let add_state = SITE_REPLICATION_STATE_LOCK.lock().await; + let add_state = site_replication_state_process_guard().await; let (started_tx, started_rx) = tokio::sync::oneshot::channel(); let (entered_tx, mut entered_rx) = tokio::sync::oneshot::channel(); let remove = tokio::spawn(async move { let _ = started_tx.send(()); let _lifecycle = SiteReplicationLifecycleGuard::acquire().await; let _bucket_op = SITE_REPLICATION_BUCKET_OP_LOCK.write().await; - let _state = SITE_REPLICATION_STATE_LOCK.lock().await; + let _state = site_replication_state_process_guard().await; let _ = entered_tx.send(()); }); started_rx.await.expect("remove task started"); @@ -12377,9 +12711,9 @@ mod tests { }; let mut queue = Vec::new(); - upsert_site_replication_retry_event(&mut queue, &peer, "/rustfs/admin/v3/site-replication/peer/iam-item", "first"); - upsert_site_replication_retry_event(&mut queue, &peer, "/rustfs/admin/v3/site-replication/peer/iam-item", "second"); - upsert_site_replication_retry_event(&mut queue, &peer, "/rustfs/admin/v3/site-replication/peer/iam-item", "third"); + upsert_site_replication_retry_event(&mut queue, &peer, "/rustfs/admin/v3/site-replication/peer/iam-item", "first", None); + upsert_site_replication_retry_event(&mut queue, &peer, "/rustfs/admin/v3/site-replication/peer/iam-item", "second", None); + upsert_site_replication_retry_event(&mut queue, &peer, "/rustfs/admin/v3/site-replication/peer/iam-item", "third", None); assert_eq!(queue.len(), 1); assert_eq!(queue[0].retry_count, SITE_REPLICATION_RETRY_FAILED_AFTER); @@ -12387,6 +12721,120 @@ mod tests { assert_eq!(queue[0].last_error, "third"); } + /// P1-15 review follow-up: a successful peer-edit delivery only proves the + /// peer reached the state THAT delivery carried. Settling it must not + /// erase a retry event a newer edit left behind, or the local site sits on + /// edit B, the peer on edit A, and nothing is queued to converge them. + #[test] + fn retry_settlement_must_not_erase_a_newer_generation_failure() { + let peer = PeerInfo { + deployment_id: "remote-dep".to_string(), + ..peer("remote", "https://remote.example.com") + }; + let mut queue = Vec::new(); + + // Edit A (generation 5) delivered successfully and is stalled before + // settling. Edit B (generation 6) commits meanwhile, fails delivery to + // the same peer, and enqueues. + upsert_site_replication_retry_event(&mut queue, &peer, SITE_REPLICATION_PEER_EDIT_PATH, "peer offline", Some(6)); + + // A resumes: its own settlement must leave B's retry alone. + assert_eq!( + settle_site_replication_retry_events(&mut queue, &peer, SITE_REPLICATION_PEER_EDIT_PATH, Some(5)), + 0 + ); + assert_eq!(queue.len(), 1, "the newer edit's retry event was erased by an older success"); + assert_eq!(queue[0].edit_generation, Some(6)); + + // An even older delivery failing afterwards must not lower the fence. + upsert_site_replication_retry_event(&mut queue, &peer, SITE_REPLICATION_PEER_EDIT_PATH, "still offline", Some(4)); + assert_eq!(queue[0].edit_generation, Some(6)); + + // B's own delivery succeeding is what clears it. + assert_eq!( + settle_site_replication_retry_events(&mut queue, &peer, SITE_REPLICATION_PEER_EDIT_PATH, Some(6)), + 1 + ); + assert!(queue.is_empty()); + + // Broadcast paths carry no generation and keep settling unconditionally + // — their retry events live under their own path and never collide + // with a peer-edit delivery. + let iam_path = "/rustfs/admin/v3/site-replication/peer/iam-item"; + upsert_site_replication_retry_event(&mut queue, &peer, iam_path, "peer offline", None); + assert_eq!(dequeue_site_replication_retry_events(&mut queue, &peer, iam_path), 1); + } + + /// P1-15 review follow-up: the receiving side of the ordering fence. The + /// sender's process mutex is per node, so two nodes of the same site can + /// fan out in the opposite order to their commits; the receiver decides + /// ordering from the generation the sender allocated under the distributed + /// state lock. + #[test] + fn peer_edit_fence_rejects_a_delivery_the_newer_edit_already_passed() { + let mut state = SiteReplicationState::default(); + let path = peer_edit_path_with_fence(Some("origin-site"), 7); + let queries = query_pairs(&path.parse::().expect("the fenced path must be a valid request uri")); + let (origin, generation) = peer_edit_fence(&queries).expect("the fence must round-trip through the request path"); + assert_eq!((origin.as_str(), generation), ("origin-site", 7)); + + assert!(!peer_edit_delivery_is_stale(&state, &origin, generation)); + record_applied_peer_edit_generation(&mut state, &origin, generation); + + // The delivery that lost the race carries the older generation. + assert!(peer_edit_delivery_is_stale(&state, "origin-site", 6)); + // A replay of the generation already applied is stale too. + assert!(peer_edit_delivery_is_stale(&state, "origin-site", 7)); + // The next edit from that origin still applies... + assert!(!peer_edit_delivery_is_stale(&state, "origin-site", 8)); + // ...and another origin site is ordered independently. + assert!(!peer_edit_delivery_is_stale(&state, "other-site", 1)); + + // A sender with no deployment id has nothing to fence against, and a + // peer that predates the fence sends no query: both keep the previous + // last-writer-wins behaviour rather than being rejected. + assert_eq!(peer_edit_path_with_fence(None, 9), SITE_REPLICATION_PEER_EDIT_PATH); + assert_eq!(peer_edit_path_with_fence(Some(""), 9), SITE_REPLICATION_PEER_EDIT_PATH); + assert!(peer_edit_fence(&HashMap::new()).is_none()); + } + + /// P1-15 review follow-up: a site that leaves the mesh drops below two + /// peers, which clears its state object and restarts its generation + /// counter at zero. A mark left over from its previous membership would + /// make every edit it sends after rejoining look stale — i.e. the fence + /// would silently swallow that site's edits forever. + #[test] + fn peer_edit_marks_do_not_outlive_the_peer_that_earned_them() { + let mut state = SiteReplicationState::default(); + state.peers.insert( + "origin-site".to_string(), + PeerInfo { + deployment_id: "origin-site".to_string(), + ..peer("origin", "https://origin.example:9000") + }, + ); + record_applied_peer_edit_generation(&mut state, "origin-site", 12); + let retained = serde_json::to_vec(&state).expect("serialize state with a live peer"); + assert_eq!( + parse_site_replication_state(&retained) + .expect("reload") + .applied_edit_generations + .get("origin-site"), + Some(&12), + "the mark for a current peer must survive a reload" + ); + + state.peers.remove("origin-site"); + let departed = serde_json::to_vec(&state).expect("serialize state after the peer left"); + let reloaded = parse_site_replication_state(&departed).expect("reload"); + assert!( + reloaded.applied_edit_generations.is_empty(), + "a departed peer's mark must not fence its edits after it rejoins: {:?}", + reloaded.applied_edit_generations + ); + assert!(!peer_edit_delivery_is_stale(&reloaded, "origin-site", 1)); + } + #[test] fn test_retry_stats_for_state_counts_pending_and_failed() { let state = SiteReplicationState { @@ -14752,4 +15200,227 @@ mod tests { newer.generation += 1; assert!(parse_site_resync_page(&query, &newer).is_err()); } + + /// P1-15 review follow-up: isolates the PROCESS guard. One writer uses + /// the legacy shape that the not-yet-migrated call sites still use (take + /// the process mutex, then load / mutate / save through the plain + /// helpers, which take their own per-IO object locks); the other runs the + /// full transaction. They only stay serialized because the transaction + /// also takes the process mutex — drop it there and this test loses one + /// of the two updates. + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + #[serial] + async fn test_transaction_serializes_against_a_process_only_legacy_writer() { + publish_ready_iam_context().await; + + let seed = SiteReplicationState { + pending_rotation: Some(PendingRotation { + id: "rot-legacy".to_string(), + access_key: "svc-account".to_string(), + ..Default::default() + }), + ..Default::default() + }; + save_site_replication_state(&seed).await.expect("seed state"); + + const ROUNDS: usize = 8; + for round in 0..ROUNDS { + let legacy_id = format!("legacy-{round}"); + let legacy = tokio::spawn(async move { + // Exactly what an unmigrated call site does today. + let _guard = site_replication_state_process_guard().await; + let mut state = load_site_replication_state().await.expect("legacy load"); + if let Some(pending) = state.pending_rotation.as_mut() { + pending.secret_candidates.push(legacy_id); + } + save_site_replication_state(&state).await.expect("legacy save"); + }); + let ack_id = format!("ack-{round}"); + let migrated = tokio::spawn(async move { + mark_pending_rotation_peer_acked("rot-legacy", &ack_id) + .await + .expect("transaction writer"); + }); + legacy.await.expect("legacy task"); + migrated.await.expect("transaction task"); + } + + let final_state = load_site_replication_state().await.expect("reload"); + let pending = final_state.pending_rotation.expect("pending rotation survives"); + for round in 0..ROUNDS { + assert!( + pending.secret_candidates.contains(&format!("legacy-{round}")), + "legacy writer update {round} was lost; candidates: {:?}", + pending.secret_candidates + ); + assert!( + pending.acked_deployment_ids.contains(&format!("ack-{round}")), + "transaction writer update {round} was lost; acks: {:?}", + pending.acked_deployment_ids + ); + } + } + + /// P1-15 review follow-up: isolates the DISTRIBUTED guard. Both writers + /// bypass the process mutex, which is what two separate nodes do — the + /// mutex is per process and cannot serialize them. Only the state-object + /// write lock keeps their read-modify-write sequences apart; drop it and + /// this test loses an update. + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + #[serial] + async fn test_state_object_lock_serializes_writers_from_separate_nodes() { + publish_ready_iam_context().await; + + let seed = SiteReplicationState { + pending_rotation: Some(PendingRotation { + id: "rot-nodes".to_string(), + access_key: "svc-account".to_string(), + ..Default::default() + }), + ..Default::default() + }; + save_site_replication_state(&seed).await.expect("seed state"); + + // Each "node" runs the production transaction minus the process + // mutex — the distributed state-object lock is the only thing left + // to keep them apart. + fn node_local_update(candidate: String) -> impl std::future::Future> { + update_site_replication_state_as_separate_node(move |state| { + if let Some(pending) = state.pending_rotation.as_mut() { + pending.secret_candidates.push(candidate); + } + Ok(()) + }) + } + + const ROUNDS: usize = 8; + for round in 0..ROUNDS { + let node_a = tokio::spawn(node_local_update(format!("node-a-{round}"))); + let node_b = tokio::spawn(node_local_update(format!("node-b-{round}"))); + node_a.await.expect("node a task").expect("node a update"); + node_b.await.expect("node b task").expect("node b update"); + } + + let final_state = load_site_replication_state().await.expect("reload"); + let pending = final_state.pending_rotation.expect("pending rotation survives"); + for round in 0..ROUNDS { + for node in ["node-a", "node-b"] { + assert!( + pending.secret_candidates.contains(&format!("{node}-{round}")), + "{node} update {round} was lost across nodes; candidates: {:?}", + pending.secret_candidates + ); + } + } + } + + /// P1-15 review follow-up: the sending side of the peer-edit ordering + /// fence, driven by two separate nodes. Both bypass the process mutex — + /// which is exactly what two nodes of one site do — so the generation is + /// only unique because it is allocated inside the state transaction, under + /// the distributed state-object lock. Two nodes sharing a generation would + /// leave the receiver unable to tell which edit is newer. + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + #[serial] + async fn test_peer_edit_generations_are_unique_across_nodes() { + publish_ready_iam_context().await; + // A configured site: `persist_site_replication_state_no_lock` clears + // the object once a site drops below two peers, and a cleared object + // would reset the counter between allocations. + let seed = SiteReplicationState { + peers: ["site-a", "site-b"] + .into_iter() + .map(|name| (name.to_string(), peer(name, &format!("https://{name}.example:9000")))) + .collect(), + ..Default::default() + }; + save_site_replication_state(&seed).await.expect("seed state"); + + fn node_local_allocate() -> impl std::future::Future> { + update_site_replication_state_as_separate_node(|state| Ok(next_peer_edit_generation(state))) + } + + const ROUNDS: usize = 8; + let mut generations = Vec::new(); + for _ in 0..ROUNDS { + let node_a = tokio::spawn(node_local_allocate()); + let node_b = tokio::spawn(node_local_allocate()); + generations.push(node_a.await.expect("node a task").expect("node a allocation")); + generations.push(node_b.await.expect("node b task").expect("node b allocation")); + } + + let unique: BTreeSet = generations.iter().copied().collect(); + assert_eq!( + unique.len(), + generations.len(), + "two nodes took the same edit generation, so their deliveries cannot be ordered: {generations:?}" + ); + assert_eq!( + load_site_replication_state().await.expect("reload").edit_generation, + generations.len() as u64, + "the persisted counter must account for every allocation" + ); + } + + /// P1-15 (rustfs/backlog#1675 B2): every state RMW — including the + /// retry-event writers that hang off the hook broadcast paths — now runs + /// through `update_site_replication_state`, which holds the process + /// mutex plus the distributed state-object write lock for the whole + /// load -> mutate -> persist. Before the fix the retry writers took no + /// lock at all: this concurrent mix deterministically lost one side + /// (the red-light commit pinned the exact interleaving). + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + #[serial] + async fn test_retry_event_persist_must_not_wipe_concurrent_locked_rmw() { + publish_ready_iam_context().await; + + let seed = SiteReplicationState { + pending_rotation: Some(PendingRotation { + id: "rot-1".to_string(), + access_key: "svc-account".to_string(), + ..Default::default() + }), + ..Default::default() + }; + save_site_replication_state(&seed).await.expect("seed state"); + + const ROUNDS: usize = 8; + for round in 0..ROUNDS { + let peer = PeerInfo { + endpoint: format!("https://peer-{round}.example:9000"), + deployment_id: format!("peer-{round}-deployment"), + ..Default::default() + }; + let enqueue = tokio::spawn(async move { + let error = S3Error::with_message(S3ErrorCode::InternalError, "peer offline".to_string()); + enqueue_site_replication_retry_event(&peer, "bucket-meta", &error).await; + }); + let ack_id = format!("ack-{round}-deployment"); + let ack = tokio::spawn(async move { + mark_pending_rotation_peer_acked("rot-1", &ack_id) + .await + .expect("locked writer must succeed"); + }); + enqueue.await.expect("enqueue task"); + ack.await.expect("ack task"); + } + + let final_state = load_site_replication_state().await.expect("reload final state"); + assert_eq!( + final_state.retry_queue.len(), + ROUNDS, + "every concurrently-enqueued retry event must survive" + ); + let acked = &final_state + .pending_rotation + .as_ref() + .expect("pending rotation must survive") + .acked_deployment_ids; + for round in 0..ROUNDS { + assert!( + acked.contains(&format!("ack-{round}-deployment")), + "rotation ack {round} must survive the concurrent retry-event writers; acked: {acked:?}" + ); + } + } } diff --git a/rustfs/src/admin/mod.rs b/rustfs/src/admin/mod.rs index 1671ffa2b..be2ea82d9 100644 --- a/rustfs/src/admin/mod.rs +++ b/rustfs/src/admin/mod.rs @@ -24,6 +24,7 @@ pub mod router; pub(crate) mod runtime_sources; pub mod service; pub mod site_replication_identity; +pub(crate) mod site_replication_state; pub(crate) mod storage_api; pub mod utils; diff --git a/rustfs/src/admin/service/site_replication.rs b/rustfs/src/admin/service/site_replication.rs index 8af98d6a7..db6fdde22 100644 --- a/rustfs/src/admin/service/site_replication.rs +++ b/rustfs/src/admin/service/site_replication.rs @@ -16,14 +16,14 @@ use crate::admin::runtime_sources::{AppContext, current_app_context, current_obj use crate::admin::site_replication_identity::{ deployment_id_for_endpoint, mark_unknown_peer_sync_enabled, normalize_peer_map_by_identity_with, }; -use crate::admin::storage_api::config::{read_admin_config, save_admin_config}; +use crate::admin::site_replication_state::{SITE_REPLICATION_STATE_PATH, with_site_replication_state_lock_on}; use crate::admin::storage_api::error::Error as StorageError; +use crate::storage::storage_api::{read_config_no_lock, save_config_no_lock}; use rustfs_madmin::PeerInfo; use s3s::{S3Error, S3ErrorCode, S3Result}; use serde_json::{Map, Value}; use tracing::info; -const SITE_REPLICATION_STATE_PATH: &str = "config/site-replication/state.json"; const SYNC_STATE_INITIALIZED_FIELD: &str = "sync_state_initialized"; fn normalize_peers_map(peers: &Map, initialize_sync_state: bool) -> Map { @@ -113,25 +113,36 @@ pub async fn reload_site_replication_runtime_state_for_context(context: Option<& return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; - match read_admin_config(store.clone(), SITE_REPLICATION_STATE_PATH).await { - Ok(data) => { - if let Some(normalized) = - normalize_site_replication_state_json(&data).map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, e))? - { - save_admin_config(store, SITE_REPLICATION_STATE_PATH, normalized) - .await - .map_err(|e| { - S3Error::with_message(S3ErrorCode::InternalError, format!("normalize site replication state failed: {e}")) - })?; + // The whole read -> normalize -> save is one RMW: run it inside the + // shared state transaction boundary (P1-15) so a cluster-wide reload + // fan-out cannot overwrite a concurrent state writer. IO must be the + // no-lock variants — the boundary already holds the object lock. + let lock_store = store.clone(); + with_site_replication_state_lock_on(lock_store, move || async move { + match read_config_no_lock(store.clone(), SITE_REPLICATION_STATE_PATH).await { + Ok(data) => { + if let Some(normalized) = normalize_site_replication_state_json(&data) + .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, e))? + { + save_config_no_lock(store, SITE_REPLICATION_STATE_PATH, normalized) + .await + .map_err(|e| { + S3Error::with_message( + S3ErrorCode::InternalError, + format!("normalize site replication state failed: {e}"), + ) + })?; + } + Ok(()) } - Ok(()) + Err(StorageError::ConfigNotFound) => Ok(()), + Err(err) => Err(S3Error::with_message( + S3ErrorCode::InternalError, + format!("failed to load site replication state: {err}"), + )), } - Err(StorageError::ConfigNotFound) => Ok(()), - Err(err) => Err(S3Error::with_message( - S3ErrorCode::InternalError, - format!("failed to load site replication state: {err}"), - )), - } + }) + .await } pub async fn reload_site_replication_runtime_state() -> S3Result<()> { diff --git a/rustfs/src/admin/site_replication_state.rs b/rustfs/src/admin/site_replication_state.rs new file mode 100644 index 000000000..7270a4c58 --- /dev/null +++ b/rustfs/src/admin/site_replication_state.rs @@ -0,0 +1,110 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Locking primitive for the site-replication state object (P1-15, +//! rustfs/backlog#1675 B2). +//! +//! `config/site-replication/state.json` is mutated by read-modify-write +//! sequences spread over many call sites: admin handlers, the retry-event +//! writers on every hook broadcast path, and the service-side reload driven +//! over node RPC. Historically only some of them held the process-local +//! mutex and none held a distributed lock across the whole RMW, so +//! concurrent writers overwrote each other (single-process for the unlocked +//! writers, cross-node for everyone). +//! +//! `with_site_replication_state_lock` is the single transaction boundary: +//! it holds the process-local mutex AND the distributed config-object write +//! lock (the pattern proven by the repair state, +//! `update_site_replication_repair_state`) for the duration of the caller's +//! closure. All IO inside the closure must use the `*_no_lock` config +//! helpers — the locked variants would self-deadlock on the same object +//! lock. Do not perform peer network calls or take other config locks +//! inside the closure. +//! +//! The process-local mutex is transitional: call sites still outside this +//! primitive serialize against migrated ones through it. Once every RMW +//! call site goes through here (P1-15 PR2) it will be removed, leaving the +//! object lock as the only mechanism. +//! +//! Lock order (unchanged from the historical comment next to the mutex): +//! lifecycle -> bucket operation -> repair admission -> state (process +//! mutex, then state object lock) -> per-bucket metadata. + +use crate::admin::storage_api::runtime::ECStore; +use crate::storage::storage_api::with_config_object_write_lock; +use s3s::{S3Error, S3ErrorCode, S3Result}; +use std::sync::Arc; + +use super::runtime_sources::current_object_store_handle; + +/// Config object holding the whole site-replication state, including the +/// retry-event queue. Shared by the typed handler-side accessors and the +/// byte-level tolerant reload on the service side. +pub(crate) const SITE_REPLICATION_STATE_PATH: &str = "config/site-replication/state.json"; + +/// Transitional process-local mutex — see the module docs. Stays private to +/// this module (owner-local static, enforced by +/// `scripts/check_architecture_migration_rules.sh`); callers go through +/// [`site_replication_state_process_guard`]. +static SITE_REPLICATION_STATE_LOCK: std::sync::LazyLock> = + std::sync::LazyLock::new(|| tokio::sync::Mutex::new(())); + +/// Owner helper for the transitional process mutex: the RMW call sites in +/// `handlers::site_replication` that PR2 has not migrated to +/// [`with_site_replication_state_lock`] yet hold this guard so they stay +/// mutually exclusive with the migrated ones. Removed together with the +/// mutex once every call site runs inside the transaction boundary. +pub(crate) async fn site_replication_state_process_guard() -> tokio::sync::MutexGuard<'static, ()> { + SITE_REPLICATION_STATE_LOCK.lock().await +} + +/// Run `operation` under the site-replication state transaction boundary: +/// process mutex first, then the distributed state-object write lock. +pub(crate) async fn with_site_replication_state_lock(operation: F) -> S3Result +where + T: Send + 'static, + F: FnOnce() -> Fut + Send + 'static, + Fut: std::future::Future> + Send + 'static, +{ + let store = + current_object_store_handle().ok_or_else(|| S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()))?; + with_site_replication_state_lock_on(store, operation).await +} + +/// Context-store variant for callers that resolve their store from an +/// explicit [`AppContext`] (the service-side reload driven over node RPC). +pub(crate) async fn with_site_replication_state_lock_on(store: Arc, operation: F) -> S3Result +where + T: Send + 'static, + F: FnOnce() -> Fut + Send + 'static, + Fut: std::future::Future> + Send + 'static, +{ + let _process_guard = SITE_REPLICATION_STATE_LOCK.lock().await; + with_site_replication_state_object_lock(store, operation).await +} + +/// The distributed half of the boundary on its own: the state-object write +/// lock, without the process mutex. This is the only thing that serializes +/// writers in *different* processes (the mutex cannot), so it is also what +/// the separate-nodes regression test drives. +pub(crate) async fn with_site_replication_state_object_lock(store: Arc, operation: F) -> S3Result +where + T: Send + 'static, + F: FnOnce() -> Fut + Send + 'static, + Fut: std::future::Future> + Send + 'static, +{ + with_config_object_write_lock(store, SITE_REPLICATION_STATE_PATH.to_string(), operation) + .await + .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("lock site replication state failed: {e}")))? +} diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index 21f577ad7..9387929a5 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -1034,6 +1034,10 @@ pub(crate) async fn save_config_no_lock(api: Arc, file: &str, data: Vec ecstore_config::com::save_config_no_lock(api, file, data).await } +pub(crate) async fn delete_config_no_lock(api: Arc, file: &str) -> Result<()> { + ecstore_config::com::delete_config_no_lock(api, file).await +} + pub(crate) async fn with_config_object_write_lock(api: Arc, object: String, operation: F) -> Result where F: FnOnce() -> Fut + Send + 'static,