mirror of
https://github.com/rustfs/rustfs.git
synced 2026-07-26 08:18:18 +00:00
fix(replication): compute the PUT replication decision exactly once (#4934)
The PutObject usecase computed `must_replicate_object` twice with the same inputs: once before commit to persist the pending replication metadata, and again after commit to drive `schedule_object_replication`. Besides repeating the versioning/config/target traversal on the hot path, the two computations read the replication configuration independently, so a config hot update landing between them could split the two phases into a pending-without-schedule or schedule-without-pending divergence. Reuse the single immutable `ReplicateDecision` computed before commit for both the pending metadata and the post-commit schedule. The two former computations were already equivalent for a stable config (`must_replicate` reads only `opts.replication_request` from the options, which the pending-suffix insertion does not touch), so this preserves replication semantics while removing the redundant traversal and closing the config-race window. The decision carries only stable target/rule/status identifiers (arn, replicate, synchronous, id) and no secrets. Add a white-box regression test that drives a real PutObject through the usecase and asserts, via a test-only invocation counter on `must_replicate_object`, that a single PUT computes the decision exactly once; reverting to the pre-commit + post-commit recompute makes the counter observe 2 and fails the test. Refs: https://github.com/rustfs/backlog/issues/1320
This commit is contained in:
@@ -1659,3 +1659,53 @@ async fn restore_object_usecase_reports_ongoing_conflict_and_completion() {
|
||||
"GET of a restored object must be served locally, not from the tier"
|
||||
);
|
||||
}
|
||||
|
||||
/// rustfs/backlog#1320: a single PUT must compute the replication decision
|
||||
/// exactly once. Historically the PUT path computed `must_replicate_object`
|
||||
/// twice with the same inputs — once before commit to persist the pending
|
||||
/// replication metadata, and again after commit to drive the schedule. Besides
|
||||
/// repeating the versioning/config/target traversal on the hot path, a
|
||||
/// replication-config hot update between the two computations could split the
|
||||
/// two phases (pending-without-schedule or schedule-without-pending). The fix
|
||||
/// reuses one immutable decision for both phases; this test observes the
|
||||
/// computation counter and fails if a second computation is reintroduced.
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
|
||||
#[serial]
|
||||
#[ignore = "global-state usecase integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
|
||||
async fn put_object_computes_replication_decision_exactly_once() {
|
||||
use super::storage_api::object_usecase::bucket::replication::MUST_REPLICATE_OBJECT_CALLS;
|
||||
use std::sync::atomic::Ordering;
|
||||
|
||||
let (_disk_paths, ecstore) = setup_test_env().await;
|
||||
let fs = FS::new();
|
||||
let usecase = DefaultObjectUsecase::from_global();
|
||||
|
||||
let bucket = format!("test-repl-decision-once-{}", &Uuid::new_v4().simple().to_string()[..8]);
|
||||
let object = "test/object.txt";
|
||||
let payload = b"replication decision single-compute payload";
|
||||
|
||||
create_test_bucket(&ecstore, bucket.as_str()).await;
|
||||
|
||||
// Reset immediately before the single PUT so the counter reflects only this
|
||||
// request. `#[serial]` guarantees no other usecase PUT runs concurrently, so
|
||||
// no unrelated `must_replicate_object` call can perturb the count.
|
||||
MUST_REPLICATE_OBJECT_CALLS.store(0, Ordering::SeqCst);
|
||||
|
||||
let input = PutObjectInput::builder()
|
||||
.bucket(bucket.clone())
|
||||
.key(object.to_string())
|
||||
.body(Some(streaming_blob_from_bytes(payload)))
|
||||
.content_length(Some(payload.len() as i64))
|
||||
.build()
|
||||
.unwrap();
|
||||
Box::pin(usecase.execute_put_object(&fs, build_request(input, Method::PUT)))
|
||||
.await
|
||||
.expect("Failed to upload object through usecase");
|
||||
|
||||
let calls = MUST_REPLICATE_OBJECT_CALLS.load(Ordering::SeqCst);
|
||||
assert_eq!(
|
||||
calls, 1,
|
||||
"a single PUT must compute the replication decision exactly once; reverting to a \
|
||||
pre-commit + post-commit recompute makes this observe 2 (rustfs/backlog#1320)"
|
||||
);
|
||||
}
|
||||
|
||||
@@ -4054,6 +4054,11 @@ impl DefaultObjectUsecase {
|
||||
.map(|ctx| ctx.request_id.clone())
|
||||
.unwrap_or_else(|| request_context::RequestContext::fallback().request_id);
|
||||
|
||||
// Compute the replication decision exactly once per PUT. The same
|
||||
// immutable `dsc` drives both the pending metadata written below and the
|
||||
// post-commit schedule (see the reuse site further down), so a
|
||||
// replication-config hot update can no longer split the two phases
|
||||
// (https://github.com/rustfs/backlog/issues/1320).
|
||||
let dsc =
|
||||
must_replicate_object(&bucket, &key, &mt2, "".to_string(), opts.delete_marker_replication_status(), opts.clone())
|
||||
.await;
|
||||
@@ -4193,9 +4198,15 @@ impl DefaultObjectUsecase {
|
||||
|
||||
let e_tag = obj_info.etag.clone().map(|etag| to_s3s_etag(&etag));
|
||||
|
||||
let dsc = must_replicate_object(&bucket, &key, &mt2, "".to_string(), opts.delete_marker_replication_status(), opts).await;
|
||||
let expiration = resolve_put_object_expiration(&bucket, &obj_info).await;
|
||||
|
||||
// Reuse the single replication decision computed before commit (see `dsc`
|
||||
// above) so the pending metadata persisted with the object and the
|
||||
// post-commit schedule always derive from the same immutable decision.
|
||||
// Recomputing here would repeat the versioning/config/target traversal and,
|
||||
// worse, allow a replication-config hot update between the two phases to
|
||||
// produce a pending-without-schedule or schedule-without-pending divergence
|
||||
// (https://github.com/rustfs/backlog/issues/1320).
|
||||
if dsc.replicate_any() {
|
||||
schedule_object_replication(obj_info.clone(), store, dsc).await;
|
||||
}
|
||||
|
||||
@@ -640,6 +640,16 @@ pub(crate) mod bucket {
|
||||
#[cfg(test)]
|
||||
pub(crate) use replication_contracts::replication_statuses_map;
|
||||
|
||||
/// Test-only counter of `must_replicate_object` invocations.
|
||||
///
|
||||
/// Used by white-box regression tests to assert that a single PUT
|
||||
/// computes the replication decision exactly once (see
|
||||
/// https://github.com/rustfs/backlog/issues/1320). A revert that
|
||||
/// re-introduces a second computation makes the count observe 2 and
|
||||
/// fails the test.
|
||||
#[cfg(test)]
|
||||
pub(crate) static MUST_REPLICATE_OBJECT_CALLS: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
|
||||
|
||||
pub(crate) async fn check_replicate_delete(
|
||||
bucket: &str,
|
||||
dobj: &super::super::storage_contracts::ObjectToDelete,
|
||||
@@ -669,6 +679,8 @@ pub(crate) mod bucket {
|
||||
status: ReplicationStatusType,
|
||||
opts: crate::storage::storage_api::StorageObjectOptions,
|
||||
) -> ReplicateDecision {
|
||||
#[cfg(test)]
|
||||
MUST_REPLICATE_OBJECT_CALLS.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
|
||||
let mopts = ReplicationObjectBridge::must_replicate_options(
|
||||
user_defined,
|
||||
user_tags,
|
||||
|
||||
Reference in New Issue
Block a user