mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-12 08:06:54 +00:00
fix(site-replication): route state RMW through one locked transaction (#5882)
* 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)
This commit is contained in:
唐小鸭
@@ -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,
|
||||
|
||||
@@ -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<S>(api: Arc<S>, 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<S>(api: Arc<S>, file: &str) -> Result<()>
|
||||
where
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -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;
|
||||
|
||||
|
||||
@@ -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<String, Value>, initialize_sync_state: bool) -> Map<String, Value> {
|
||||
@@ -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<()> {
|
||||
|
||||
@@ -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<tokio::sync::Mutex<()>> =
|
||||
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<T, F, Fut>(operation: F) -> S3Result<T>
|
||||
where
|
||||
T: Send + 'static,
|
||||
F: FnOnce() -> Fut + Send + 'static,
|
||||
Fut: std::future::Future<Output = S3Result<T>> + 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<T, F, Fut>(store: Arc<ECStore>, operation: F) -> S3Result<T>
|
||||
where
|
||||
T: Send + 'static,
|
||||
F: FnOnce() -> Fut + Send + 'static,
|
||||
Fut: std::future::Future<Output = S3Result<T>> + 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<T, F, Fut>(store: Arc<ECStore>, operation: F) -> S3Result<T>
|
||||
where
|
||||
T: Send + 'static,
|
||||
F: FnOnce() -> Fut + Send + 'static,
|
||||
Fut: std::future::Future<Output = S3Result<T>> + 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}")))?
|
||||
}
|
||||
@@ -1034,6 +1034,10 @@ pub(crate) async fn save_config_no_lock(api: Arc<ECStore>, 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<ECStore>, file: &str) -> Result<()> {
|
||||
ecstore_config::com::delete_config_no_lock(api, file).await
|
||||
}
|
||||
|
||||
pub(crate) async fn with_config_object_write_lock<F, Fut, T>(api: Arc<ECStore>, object: String, operation: F) -> Result<T>
|
||||
where
|
||||
F: FnOnce() -> Fut + Send + 'static,
|
||||
|
||||
Reference in New Issue
Block a user