Compare commits

..

15 Commits

Author SHA1 Message Date
overtrue 5bf0fc11d6 chore: sync upstream PR evidence guidance 2026-09-05 19:40:21 +08:00
overtrue a26025d2db fix(scanner): fence cached snapshots by scan execution 2026-09-05 19:40:15 +08:00
Zhengchao An f053862aad docs: request concrete behavior evidence in pull requests (#7196) 2026-09-05 18:56:26 +08:00
overtrue 6e8f409b89 chore: integrate upstream rename build repair 2026-09-05 18:25:09 +08:00
overtrue 8b7c45a710 fix(scanner): invalidate bucket work after namespace completion 2026-09-05 17:46:11 +08:00
overtrue d6d1787aed style(scanner): order merged test imports 2026-09-05 17:42:23 +08:00
overtrue 43df6d1404 chore: sync scanner publication with latest main 2026-09-05 17:42:03 +08:00
overtrue 1541cd6e02 fix(storage): integrate main and preserve recovery progress 2026-09-05 17:40:40 +08:00
Zhengchao An 5439aad3f8 fix(ci): satisfy new clippy lints 2026-09-05 17:35:39 +08:00
houseme 87bf504a54 Merge remote-tracking branch 'origin/main' into houseme/fix/local-rename-merge-regression 2026-09-05 16:31:59 +08:00
houseme 2e86c174c3 fix(ecstore): remove duplicate local rename implementation
Keep the canonical commit module after concurrent storage changes merged.
The control-write and rollback changes are already present there.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>
2026-09-05 16:28:20 +08:00
overtrue a1b1eec21f test(heal): settle PUT rename tails before disk-wipe fixtures 2026-09-05 15:24:39 +08:00
overtrue 742f4b789d fix(app): simplify absent SSE configuration matching 2026-09-05 14:54:03 +08:00
overtrue 1301deb9f5 chore: sync main for ODM regression validation 2026-09-05 14:44:39 +08:00
overtrue 5cb670d360 fix(storage): harden ODM and scanner publication 2026-09-05 14:44:01 +08:00
57 changed files with 2430 additions and 2609 deletions
+5 -5
View File
@@ -10,16 +10,16 @@ Use N/A when there is no related issue.
## Summary of Changes
<!--
Briefly explain what changed and why reviewers should accept it.
Focus on behavior, compatibility, and review-relevant context.
Describe the concrete problem and resulting behavior. For a behavior change, name the input or state that triggers it and the expected outcome. Explain any new dependency or abstraction that the change needs.
-->
## Verification
<!--
List the commands or checks you ran, for example:
- `make pre-commit`
Give 13 concrete pieces of evidence for the changed behavior: the test or command, its observed result, and the regression it catches. For a bug fix, record a failing-before/passing-after check or explain why it was unavailable.
Use N/A only when verification is not applicable.
Identify the tested commit and any local changes. When testing a prebuilt binary or external service, include its source/version and artifact identity; a successful run against a different build is not evidence for this change.
List relevant checks not run and the remaining risk. Use the validation tier in AGENTS.md; do not run broader checks solely to fill this section. For documentation-only changes, list the applicable documentation checks. Use N/A only when verification is not applicable.
-->
## Impact
+8 -3
View File
@@ -204,9 +204,8 @@ pub mod bucket {
get_public_access_block_config, get_quota_config, get_replication_config, get_request_payment_config, get_sse_config,
get_tagging_config, get_versioning_config, get_website_config, init_bucket_metadata_sys, list_bucket_targets,
reload_bucket_metadata, remove_bucket_metadata, set_bucket_metadata, update,
update_bucket_targets_under_transaction_lock, update_config_with, update_if_incarnation, update_if_incarnation_at,
update_quota_if_incarnation, update_quota_if_incarnation_at, update_under_transaction_lock,
update_under_transaction_lock_at,
update_bucket_targets_under_transaction_lock, update_config_with, update_if_incarnation, update_quota_if_incarnation,
update_under_transaction_lock,
};
}
@@ -562,6 +561,12 @@ pub mod set_disk {
pub mod test_util {
pub use crate::bucket::quota::reservation::fail_next_quota_ledger_save_for_test;
pub use crate::set_disk::{MultipartCommitBarrier, MultipartCommitPause, PutObjectCommitBarrier, PutObjectCommitPause};
/// Keep a namespace commit pending until the returned owner is dropped.
#[must_use]
pub fn hold_namespace_commit(store: &crate::store::ECStore) -> impl Send + Sync {
store.ctx.begin_namespace_commit()
}
}
}
+1 -47
View File
@@ -802,22 +802,9 @@ impl BucketMetadata {
}
}
/// Replace one config payload and stamp its `*_config_updated_at` with the
/// local clock. This is the entry for edits that originate here: the
/// local write time is the edit's source time.
pub fn update_config(&mut self, config_file: &str, data: Vec<u8>) -> Result<OffsetDateTime> {
self.update_config_at(config_file, data, OffsetDateTime::now_utc())
}
let updated = OffsetDateTime::now_utc();
/// [`Self::update_config`] with an explicit `updated_at` stamp.
///
/// For a config replicated from another site the edit's source time is
/// the peer's `updated_at`, not the moment it lands here: staleness of
/// the next incoming item is judged against the stored stamp, so stamping
/// the local apply time would reject a newer source edit that was merely
/// delivered late (backlog#2292). Only replication receivers should pass
/// a foreign time; local edits keep [`Self::update_config`].
pub fn update_config_at(&mut self, config_file: &str, data: Vec<u8>, updated: OffsetDateTime) -> Result<OffsetDateTime> {
match config_file {
BUCKET_POLICY_CONFIG => {
self.policy_config_json = data;
@@ -1556,39 +1543,6 @@ mod test {
assert_eq!(metadata.bucket_incarnation_id, incarnation);
}
/// backlog#2292: a replicated config is stamped with the source
/// `updated_at` it was given, not the local clock, while the plain
/// `update_config` entry keeps stamping the local clock.
#[test]
fn update_config_at_stamps_the_given_time_and_update_config_stamps_now() {
let source_time = OffsetDateTime::now_utc() - time::Duration::hours(3);
let mut metadata = BucketMetadata::new("source-stamped");
let stamped = metadata
.update_config_at(BUCKET_POLICY_CONFIG, br#"{"Version":"2012-10-17","Statement":[]}"#.to_vec(), source_time)
.unwrap();
assert_eq!(stamped, source_time);
assert_eq!(metadata.policy_config_updated_at, source_time);
let tagging = b"<Tagging><TagSet><Tag><Key>k</Key><Value>v</Value></Tag></TagSet></Tagging>".to_vec();
let stamped = metadata
.update_config_at(BUCKET_TAGGING_CONFIG, tagging, source_time)
.unwrap();
assert_eq!(stamped, source_time);
assert_eq!(metadata.tagging_config_updated_at, source_time);
let before = OffsetDateTime::now_utc();
let stamped = metadata
.update_config(BUCKET_POLICY_CONFIG, br#"{"Version":"2012-10-17","Statement":[]}"#.to_vec())
.unwrap();
assert!(stamped >= before, "a local edit is stamped with the local clock");
assert_eq!(metadata.policy_config_updated_at, stamped);
assert_eq!(
metadata.tagging_config_updated_at, source_time,
"restamping one config must not move another config's stamp"
);
}
#[test]
fn object_locking_requires_lock_metadata_not_plain_versioning() {
use s3s::dto::ObjectLockEnabled;
+14 -145
View File
@@ -581,32 +581,6 @@ pub async fn update_if_incarnation(
config_file,
data,
Some(expected_incarnation_id),
None,
))
.await
}
/// [`update_if_incarnation`] stamping the config with `updated_at` instead of
/// the local clock.
///
/// For a site-replication receiver the edit's source time is the peer's
/// `updated_at`; persisting it keeps the stored `*_config_updated_at` on the
/// source clock so the next item's staleness is judged source-time against
/// source-time (backlog#2292). See [`BucketMetadata::update_config_at`].
pub async fn update_if_incarnation_at(
bucket: &str,
config_file: &str,
data: Vec<u8>,
expected_incarnation_id: Uuid,
updated_at: OffsetDateTime,
) -> Result<OffsetDateTime> {
Box::pin(update_with_sys_expected(
get_bucket_metadata_sys()?,
bucket,
config_file,
data,
Some(expected_incarnation_id),
Some(updated_at),
))
.await
}
@@ -638,22 +612,18 @@ async fn update_with_sys(
config_file: &str,
data: Vec<u8>,
) -> Result<OffsetDateTime> {
update_with_sys_expected(sys, bucket, config_file, data, None, None).await
update_with_sys_expected(sys, bucket, config_file, data, None).await
}
/// `updated_at` is the stamp persisted on the config; `None` uses the local
/// clock (the edit originates here), `Some` carries a replicated edit's
/// source time (backlog#2292).
async fn update_with_sys_expected(
sys: Arc<RwLock<BucketMetadataSys>>,
bucket: &str,
config_file: &str,
data: Vec<u8>,
expected_incarnation_id: Option<Uuid>,
updated_at: Option<OffsetDateTime>,
) -> Result<OffsetDateTime> {
let guard = acquire_config_write_guard_for_incarnation(sys.clone(), bucket, expected_incarnation_id).await?;
update_under_config_write_guard(sys, &guard, config_file, data, updated_at).await
update_under_config_write_guard(sys, &guard, config_file, data).await
}
/// [`delete`] against an explicitly supplied metadata system. See
@@ -816,21 +786,7 @@ pub async fn update_under_transaction_lock(
data: Vec<u8>,
) -> Result<OffsetDateTime> {
guard.ensure_valid(bucket)?;
update_under_config_write_guard(get_bucket_metadata_sys()?, guard, config_file, data, None).await
}
/// [`update_under_transaction_lock`] stamping the config with `updated_at`
/// (a replicated edit's source time) instead of the local clock; see
/// [`update_if_incarnation_at`] (backlog#2292).
pub async fn update_under_transaction_lock_at(
guard: &BucketMetadataMutationGuard,
bucket: &str,
config_file: &str,
data: Vec<u8>,
updated_at: OffsetDateTime,
) -> Result<OffsetDateTime> {
guard.ensure_valid(bucket)?;
update_under_config_write_guard(get_bucket_metadata_sys()?, guard, config_file, data, Some(updated_at)).await
update_under_config_write_guard(get_bucket_metadata_sys()?, guard, config_file, data).await
}
/// Clear one config file while the caller holds this bucket's transaction lock.
@@ -848,29 +804,6 @@ pub async fn update_quota_if_incarnation(
data: Vec<u8>,
expected_incarnation_id: Uuid,
proof: &crate::services::notification_sys::CrossPoolFenceFleetProofToken,
) -> Result<OffsetDateTime> {
update_quota_if_incarnation_stamped(bucket, data, expected_incarnation_id, proof, None).await
}
/// [`update_quota_if_incarnation`] stamping the quota config with
/// `updated_at` (a replicated edit's source time) instead of the local
/// clock; see [`update_if_incarnation_at`] (backlog#2292).
pub async fn update_quota_if_incarnation_at(
bucket: &str,
data: Vec<u8>,
expected_incarnation_id: Uuid,
proof: &crate::services::notification_sys::CrossPoolFenceFleetProofToken,
updated_at: OffsetDateTime,
) -> Result<OffsetDateTime> {
update_quota_if_incarnation_stamped(bucket, data, expected_incarnation_id, proof, Some(updated_at)).await
}
async fn update_quota_if_incarnation_stamped(
bucket: &str,
data: Vec<u8>,
expected_incarnation_id: Uuid,
proof: &crate::services::notification_sys::CrossPoolFenceFleetProofToken,
updated_at: Option<OffsetDateTime>,
) -> Result<OffsetDateTime> {
let sys = get_bucket_metadata_sys()?;
let guard = Box::pin(acquire_config_write_guard_for_incarnation(
@@ -888,7 +821,7 @@ async fn update_quota_if_incarnation_stamped(
achieved: 0,
});
}
update_under_config_write_guard(sys, &guard, rustfs_config::QUOTA_CONFIG_FILE, data, updated_at).await
update_under_config_write_guard(sys, &guard, rustfs_config::QUOTA_CONFIG_FILE, data).await
}
pub async fn update_bucket_targets_under_transaction_lock(
@@ -904,7 +837,6 @@ async fn update_under_config_write_guard(
guard: &BucketMetadataMutationGuard,
config_file: &str,
data: Vec<u8>,
updated_at: Option<OffsetDateTime>,
) -> Result<OffsetDateTime> {
guard.ensure_valid(&guard.bucket)?;
let metadata_sys = sys.read().await.clone();
@@ -916,7 +848,7 @@ async fn update_under_config_write_guard(
Some(&guard.transaction_guard),
&guard.bucket,
"bucket config transaction",
metadata_sys.update_checked(&guard.bucket, config_file, data, true, guard.incarnation_id, updated_at),
metadata_sys.update_checked(&guard.bucket, config_file, data, true, guard.incarnation_id),
),
)
.await?;
@@ -939,7 +871,7 @@ async fn delete_under_config_write_guard(
Some(&guard.transaction_guard),
&guard.bucket,
"bucket config deletion transaction",
metadata_sys.update_checked(&guard.bucket, config_file, Vec::new(), false, guard.incarnation_id, None),
metadata_sys.update_checked(&guard.bucket, config_file, Vec::new(), false, guard.incarnation_id),
),
)
.await?;
@@ -1838,17 +1770,15 @@ impl BucketMetadataSys {
/// `update` and the config read alone). Keep these boxed.
pub async fn update(&self, bucket: &str, config_file: &str, data: Vec<u8>) -> Result<OffsetDateTime> {
let incarnation_id = Box::pin(self.get_bucket_incarnation_id(bucket)).await?;
Box::pin(self.update_checked(bucket, config_file, data, true, incarnation_id, None)).await
Box::pin(self.update_checked(bucket, config_file, data, true, incarnation_id)).await
}
pub async fn delete(&self, bucket: &str, config_file: &str) -> Result<OffsetDateTime> {
let incarnation_id = self.get_bucket_incarnation_id(bucket).await?;
self.update_checked(bucket, config_file, Vec::new(), false, incarnation_id, None)
self.update_checked(bucket, config_file, Vec::new(), false, incarnation_id)
.await
}
/// `updated_at`: `None` stamps the local clock; `Some` persists a
/// replicated edit's source time (backlog#2292).
async fn update_checked(
&self,
bucket: &str,
@@ -1856,7 +1786,6 @@ impl BucketMetadataSys {
data: Vec<u8>,
parse: bool,
expected_incarnation_id: Uuid,
updated_at: Option<OffsetDateTime>,
) -> Result<OffsetDateTime> {
// Load through this system's own store, the one `save` persists to
// (backlog#1052 S7). Reading from the ambient handle instead made the
@@ -1867,10 +1796,7 @@ impl BucketMetadataSys {
return Err(Error::BucketNotFound(bucket.to_string()));
}
let updated = match updated_at {
Some(updated_at) => bm.update_config_at(config_file, data, updated_at)?,
None => bm.update_config(config_file, data)?,
};
let updated = bm.update_config(config_file, data)?;
Box::pin(self.save(bm)).await?;
@@ -3839,57 +3765,6 @@ mod tests {
);
}
/// backlog#2292: the explicit-stamp write path persists the given source
/// time as the config's `*_config_updated_at` — through the incarnation
/// path and through an already-held transaction guard — and survives a
/// reload from disk, while the plain path keeps stamping the local clock.
#[tokio::test]
async fn explicit_updated_at_is_persisted_as_the_config_stamp() {
let (dirs, ecstore) = isolated_store_over_temp_disks().await;
let bucket = "source-stamped-config";
for dir in &dirs {
std::fs::create_dir_all(dir.path().join(bucket)).expect("bucket volume should be created");
}
let sys = Arc::new(RwLock::new(BucketMetadataSys::new(ecstore)));
let source_time = OffsetDateTime::now_utc() - Duration::from_secs(3 * 3600);
let policy = br#"{"Version":"2012-10-17","Statement":[]}"#.to_vec();
let tagging = b"<Tagging><TagSet><Tag><Key>k</Key><Value>v</Value></Tag></TagSet></Tagging>".to_vec();
// Incarnation path (`update_if_incarnation_at` minus the ambient lookup).
let stamped =
update_with_sys_expected(sys.clone(), bucket, BUCKET_POLICY_CONFIG, policy.clone(), None, Some(source_time))
.await
.expect("source-stamped policy write should persist");
assert_eq!(stamped, source_time);
// Held-guard path (`update_under_transaction_lock_at` minus the ambient lookup).
let guard = acquire_config_write_guard(sys.clone(), bucket).await.expect("write guard");
let stamped = update_under_config_write_guard(sys.clone(), &guard, BUCKET_TAGGING_CONFIG, tagging, Some(source_time))
.await
.expect("source-stamped tagging write should persist");
drop(guard);
assert_eq!(stamped, source_time);
let metadata_sys = sys.read().await.clone();
metadata_sys.metadata_map.write().await.clear();
let reloaded = metadata_sys.get_config_from_disk(bucket).await.expect("reload from disk");
assert_eq!(reloaded.policy_config_updated_at, source_time);
assert_eq!(reloaded.tagging_config_updated_at, source_time);
// The plain path is unchanged: a local edit is stamped with the local clock.
let before = OffsetDateTime::now_utc();
let stamped = update_with_sys(sys.clone(), bucket, BUCKET_POLICY_CONFIG, policy)
.await
.expect("locally stamped policy write should persist");
assert!(stamped >= before, "the plain write path must keep stamping the local clock");
let reloaded = metadata_sys.get_config_from_disk(bucket).await.expect("reload from disk");
assert_eq!(reloaded.policy_config_updated_at, stamped);
assert_eq!(
reloaded.tagging_config_updated_at, source_time,
"an unrelated config keeps its source stamp"
);
}
/// The load and the persisted write share one write guard, so concurrent
/// rewrites of the same config compose instead of clobbering each other.
/// Moving the load outside that guard loses all but the last tag.
@@ -4106,16 +3981,10 @@ mod tests {
let new_incarnation = store.bucket_incarnation_id_from_disk(bucket).await.unwrap();
assert_ne!(old_incarnation, new_incarnation);
let err = update_with_sys_expected(
sys.clone(),
bucket,
BUCKET_TAGGING_CONFIG,
b"<Tagging/>".to_vec(),
Some(old_incarnation),
None,
)
.await
.expect_err("a request authorized for the deleted incarnation must fail closed");
let err =
update_with_sys_expected(sys.clone(), bucket, BUCKET_TAGGING_CONFIG, b"<Tagging/>".to_vec(), Some(old_incarnation))
.await
.expect_err("a request authorized for the deleted incarnation must fail closed");
assert!(matches!(err, Error::BucketNotFound(name) if name == bucket));
let persisted = sys.read().await.get_config_from_disk(bucket).await.unwrap();
@@ -4150,7 +4019,7 @@ mod tests {
}],
})
.unwrap();
update_under_config_write_guard(sys, &guard, BUCKET_TAGGING_CONFIG, tagging, None)
update_under_config_write_guard(sys, &guard, BUCKET_TAGGING_CONFIG, tagging)
.await
.unwrap();
assert!(!delete.is_finished());
@@ -25,8 +25,8 @@
//! [`BACKFILL_SAVE_INTERVAL`], and at every page end, with an `If-Match`
//! compare-and-set so a concurrent cancel or takeover is never overwritten.
//! - The `continuation_token` only advances once every pull queued from the
//! page before it has reported back, so a crash re-lists at most one page
//! (already-present keys are then skipped, never re-pulled).
//! page before it has succeeded. After a failure it stays at that page,
//! so crash recovery cannot skip failed pulls (existing keys are skipped).
//! - The owner holds a lease of [`BACKFILL_LEASE`] renewed by every save. The
//! recovery loop ([`run_backfill_recovery_loop`]) scans the buckets this
//! node has an ODM state for every [`BACKFILL_RECOVERY_INTERVAL`] and takes
@@ -367,9 +367,8 @@ pub struct LocalBackfillObject {
pub source_etag: Option<String>,
}
/// Receiver of one queued pull's report; `None` when the pull was coalesced
/// into one already running.
pub type PullReport = Option<oneshot::Receiver<QueuedPullOutcome>>;
/// Shared report of a new or coalesced pull; absent only when not admitted.
pub type PullReport = Option<super::pull::QueuedPullReport>;
/// Everything the job needs from its bucket, so the loop can run against a
/// mock in unit tests. Production: [`BucketBackfillContext`].
@@ -1191,9 +1190,11 @@ impl Job {
}
async fn main_loop(&mut self) -> Result<(), Stop> {
let mut cursor = self.checkpoint.continuation_token.clone();
let failed_at_resume = self.checkpoint.failed;
loop {
self.check_cancel()?;
let page = self.list_page().await?;
let page = self.list_page(cursor.as_deref()).await?;
for object in &page.objects {
self.check_cancel()?;
self.checkpoint.listed += 1;
@@ -1205,10 +1206,13 @@ impl Job {
self.drain_ready();
self.tick(false).await?;
}
// Only advance the cursor once every pull of this page reported
// back, so a takeover re-lists at most this page.
// A persisted cursor certifies successful work, not just listing
// progress. Keep it at the first failed page for crash recovery.
self.drain_all().await?;
self.checkpoint.continuation_token = page.next_continuation_token.clone();
cursor = page.next_continuation_token;
if self.checkpoint.failed == failed_at_resume {
self.checkpoint.continuation_token = cursor.clone();
}
self.tick(true).await?;
if !page.is_truncated {
return Ok(());
@@ -1223,7 +1227,7 @@ impl Job {
}
}
async fn list_page(&mut self) -> Result<SourcePage, Stop> {
async fn list_page(&mut self, cursor: Option<&str>) -> Result<SourcePage, Stop> {
let mut attempt = 0;
loop {
while !self.context.source_available() {
@@ -1231,7 +1235,7 @@ impl Job {
self.tick(false).await?;
}
let prefix = self.checkpoint.prefix.clone();
let token = self.checkpoint.continuation_token.clone();
let token = cursor.map(str::to_string);
match self
.context
.list_page(prefix.as_deref(), token.as_deref(), BACKFILL_LIST_PAGE_SIZE)
@@ -1305,9 +1309,10 @@ impl Job {
}
loop {
match self.context.enqueue(key) {
(EnqueueOutcome::Enqueued, report) => {
(EnqueueOutcome::Enqueued | EnqueueOutcome::Coalesced, report) => {
self.checkpoint.enqueued += 1;
if let Some(rx) = report {
let rx = report.ok_or(Stop::Unavailable)?;
{
let key = key.to_string();
self.outstanding.push(Box::pin(async move { (key, rx.await) }));
}
@@ -1322,11 +1327,6 @@ impl Job {
);
return Ok(());
}
(EnqueueOutcome::Coalesced, _) => {
// Someone else pulls it; its result is not ours to count.
self.checkpoint.enqueued += 1;
return Ok(());
}
(EnqueueOutcome::QueueFull, _) => {
// Wait, never drop: one completion frees a slot.
if self.outstanding.is_empty() {
@@ -1640,6 +1640,7 @@ mod tests {
queue_capacity: usize,
pending: Mutex<Vec<(String, oneshot::Sender<QueuedPullOutcome>)>>,
fail_keys: HashSet<String>,
coalesced: bool,
auto_complete: AtomicBool,
cancel: CancellationToken,
config_updated_at: Mutex<Option<OffsetDateTime>>,
@@ -1667,6 +1668,7 @@ mod tests {
queue_capacity: usize::MAX,
pending: Mutex::new(Vec::new()),
fail_keys: HashSet::new(),
coalesced: false,
auto_complete: AtomicBool::new(true),
cancel: CancellationToken::new(),
config_updated_at: Mutex::new(Some(ts(1_700_000_000))),
@@ -1746,7 +1748,12 @@ mod tests {
} else {
self.pending.lock().push((key.to_string(), tx));
}
(EnqueueOutcome::Enqueued, Some(rx))
let outcome = if self.coalesced {
EnqueueOutcome::Coalesced
} else {
EnqueueOutcome::Enqueued
};
(outcome, Some(futures::FutureExt::shared(rx)))
}
fn cancel_token(&self) -> CancellationToken {
@@ -1912,7 +1919,7 @@ mod tests {
#[tokio::test]
async fn failed_pulls_are_counted_hashed_and_finish_with_failures() {
let bucket = "backfill-failed";
let mut context = MockContext::new(5, 1000);
let mut context = MockContext::new(5, 2);
Arc::get_mut(&mut context)
.expect("unshared")
.fail_keys
@@ -1927,12 +1934,52 @@ mod tests {
.checkpoint;
assert_eq!(cp.state, BackfillState::CompletedWithFailures);
assert_eq!((cp.pulled, cp.failed), (4, 1));
assert_eq!(cp.continuation_token.as_deref(), Some("2"), "retain the first failed page for recovery");
assert_eq!(cp.failed_keys, vec![key_hash("k/00002")]);
let last = cp.last_error.expect("last error");
assert_eq!(last.class, "local_write");
assert_eq!(last.key_hash.as_deref(), Some(key_hash("k/00002").as_str()));
}
#[tokio::test]
async fn coalesced_pulls_block_the_checkpoint_and_report_failures() {
let bucket = "backfill-coalesced";
let mut context = MockContext::new(1, 1);
{
let ctx = Arc::get_mut(&mut context).expect("unshared");
ctx.coalesced = true;
ctx.auto_complete = AtomicBool::new(false);
ctx.fail_keys.insert("k/00000".to_string());
}
let (_dirs, store, runner) = runner_with("node-a", bucket, Arc::clone(&context)).await;
runner.start(bucket, BackfillRequest::default()).await.expect("start");
tokio::time::timeout(Duration::from_secs(10), async {
while context.pending.lock().is_empty() {
tokio::task::yield_now().await;
}
})
.await
.expect("job enqueued");
assert!(runner.is_running_locally(bucket), "coalescing is not completion");
let cp = read_checkpoint(&store, bucket)
.await
.expect("read")
.expect("checkpoint")
.checkpoint;
assert!(cp.state.is_active());
assert!(cp.continuation_token.is_none());
context.complete_pending();
runner.wait_until_idle(bucket).await;
let cp = read_checkpoint(&store, bucket)
.await
.expect("read")
.expect("checkpoint")
.checkpoint;
assert_eq!(cp.state, BackfillState::CompletedWithFailures);
assert_eq!((cp.enqueued, cp.pulled, cp.failed), (1, 0, 1));
assert_eq!(cp.failed_keys, vec![key_hash("k/00000")]);
}
#[tokio::test]
async fn listing_failure_marks_the_job_failed_with_the_error_class() {
let bucket = "backfill-list-error";
@@ -2145,6 +2192,68 @@ mod tests {
assert_eq!(runner.recover_once().await.taken_over, 0, "a finished job is not recovered");
}
#[tokio::test]
async fn recovery_advances_past_historical_failures_but_pins_new_failures() {
let bucket = "backfill-takeover-failed";
let mut context = MockContext::new(8, 2);
{
let ctx = Arc::get_mut(&mut context).expect("unshared");
ctx.auto_complete = AtomicBool::new(false);
ctx.fail_keys.insert("k/00004".to_string());
}
let (_dirs, store, runner) = runner_with("node-b", bucket, Arc::clone(&context)).await;
let crashed_at = OffsetDateTime::now_utc() - Duration::from_secs(300);
let mut crashed = BackfillCheckpoint::new(&BackfillRequest::default(), ts(1_700_000_000), "node-a", crashed_at);
crashed.continuation_token = Some("2".to_string());
crashed.failed = 1;
crashed.record_failure("local_write", Some("k/00002"), crashed_at);
write_checkpoint(&store, bucket, &crashed, None)
.await
.expect("seed failed page with an expired lease");
assert_eq!(runner.recover_once().await.taken_over, 1);
for (page_start, durable_token, failures) in [(2, "2", 1), (4, "4", 1), (6, "4", 2)] {
tokio::time::timeout(Duration::from_secs(10), async {
loop {
if context.pending.lock().len() == 2 {
break;
}
tokio::task::yield_now().await;
}
})
.await
.expect("resumed page enqueued before its reports complete");
assert_eq!(
context.pending.lock().iter().map(|(key, _)| key.clone()).collect::<Vec<_>>(),
vec![format!("k/{page_start:05}"), format!("k/{:05}", page_start + 1)]
);
let cp = read_checkpoint(&store, bucket)
.await
.expect("read persisted page boundary")
.expect("checkpoint")
.checkpoint;
assert_eq!(cp.job_id, crashed.job_id);
assert_eq!(cp.owner.as_ref().map(|owner| owner.node.as_str()), Some("node-b"));
assert_eq!(cp.continuation_token.as_deref(), Some(durable_token));
assert_eq!(cp.failed, failures);
context.complete_pending();
}
runner.wait_until_idle(bucket).await;
let cp = read_checkpoint(&store, bucket)
.await
.expect("read completed checkpoint")
.expect("checkpoint")
.checkpoint;
assert_eq!(cp.state, BackfillState::CompletedWithFailures);
assert_eq!((cp.pulled, cp.failed), (5, 2));
assert_eq!(cp.continuation_token.as_deref(), Some("4"));
assert_eq!(cp.failed_keys, vec![key_hash("k/00002"), key_hash("k/00004")]);
assert_eq!(
context.list_requests.lock().as_slice(),
&[Some("2".to_string()), Some("4".to_string()), Some("6".to_string())]
);
}
#[tokio::test]
async fn recovery_cancels_a_job_whose_config_changed_and_reclaims_own_node_jobs() {
let bucket = "backfill-recovery-config";
@@ -32,6 +32,9 @@ pub const LIST_THROUGH_TOKEN_VERSION: u32 = 1;
/// listing's own marker, so the decoder needs a positive signal before it
/// treats an opaque token as a merged one.
const LIST_THROUGH_TOKEN_TAG: &str = "odm-list";
// Object keys cannot contain NUL (bucket::utils::is_valid_object_prefix),
// so this framing cannot collide with a local key used as an opaque marker.
const LIST_THROUGH_TOKEN_PREFIX: &str = "\0odm-list:";
/// Pages fetched per side per request: the first page, plus at most one refill
/// when the first one was mostly consumed by the previous page. Two pages of
@@ -86,8 +89,7 @@ pub struct MergePick {
}
/// The continuation-token envelope. Opaque to clients: it is serialized as
/// JSON and then base64-encoded by the same helper that encodes a plain local
/// marker, so the wire shape is `base64(json)`.
/// framed JSON and then base64-encoded by the same helper as a local marker.
///
/// A `null` cursor with `done = false` means "list that side from the start";
/// `done = true` means the side is finished and must not be listed again.
@@ -129,7 +131,7 @@ impl ListThroughToken {
pub fn encode(&self) -> String {
// The envelope is built here from owned strings, so serialization
// cannot fail; the fallback keeps the signature infallible.
serde_json::to_string(self).unwrap_or_default()
format!("{LIST_THROUGH_TOKEN_PREFIX}{}", serde_json::to_string(self).unwrap_or_default())
}
}
@@ -153,21 +155,18 @@ pub enum ListThroughTokenError {
/// Classifies an already base64-decoded continuation token.
///
/// Only a JSON object carrying the envelope marker is read as a merged token;
/// Only a framed JSON object is read as a merged token;
/// anything else is a local marker, so a bucket that turns `list_through` off
/// keeps paginating with the tokens it handed out. A token that *is* an
/// envelope but was tampered with (unknown version, unknown field, truncated
/// JSON) is an error, never a silent fallback.
pub fn decode_continuation_token(decoded: &str) -> Result<ListThroughCursor, ListThroughTokenError> {
if !decoded.starts_with('{') {
return Ok(ListThroughCursor::Local(decoded.to_string()));
}
let Ok(value) = serde_json::from_str::<serde_json::Value>(decoded) else {
// Not JSON at all: an object key may legitimately start with '{'.
let Some(payload) = decoded.strip_prefix(LIST_THROUGH_TOKEN_PREFIX) else {
return Ok(ListThroughCursor::Local(decoded.to_string()));
};
let value = serde_json::from_str::<serde_json::Value>(payload).map_err(|_| ListThroughTokenError::Malformed)?;
if value.get("t").and_then(serde_json::Value::as_str) != Some(LIST_THROUGH_TOKEN_TAG) {
return Ok(ListThroughCursor::Local(decoded.to_string()));
return Err(ListThroughTokenError::Malformed);
}
match value.get("v").and_then(serde_json::Value::as_u64) {
Some(version) if version == u64::from(LIST_THROUGH_TOKEN_VERSION) => {}
@@ -986,14 +985,21 @@ mod tests {
assert_eq!(decode_continuation_token(&extra), Err(ListThroughTokenError::Malformed));
let truncated = &encoded[..encoded.len() - 3];
assert_eq!(decode_continuation_token(truncated), Ok(ListThroughCursor::Local(truncated.to_string())));
assert_eq!(decode_continuation_token(truncated), Err(ListThroughTokenError::Malformed));
let no_version = "{\"t\":\"odm-list\"}";
let no_version = "\0odm-list:{\"t\":\"odm-list\"}";
assert_eq!(decode_continuation_token(no_version), Err(ListThroughTokenError::Malformed));
}
#[test]
fn a_plain_local_marker_stays_local() {
for marker in [
r#"{"t":"odm-list","v":1}"#,
r#"{"t":"odm-list","v":2,"local_done":true}"#,
r#"{"t":"odm-list"}"#,
] {
assert_eq!(decode_continuation_token(marker), Ok(ListThroughCursor::Local(marker.to_string())));
}
assert_eq!(
decode_continuation_token("photos/2024/01.jpg"),
Ok(ListThroughCursor::Local("photos/2024/01.jpg".to_string()))
@@ -46,10 +46,10 @@ use super::stats::{PullFailureReason, PullPath};
use super::sys::{BucketOdmState, OnDemandMigrationSys, PullError, PullOutcome, PullSlot};
use async_trait::async_trait;
use bytes::Bytes;
use futures::{Stream, StreamExt};
use futures::{FutureExt, Stream, StreamExt, future::Shared};
use parking_lot::Mutex;
use rand::RngExt;
use std::collections::{HashMap, HashSet};
use std::collections::HashMap;
use std::fmt;
use std::io;
use std::pin::Pin;
@@ -133,6 +133,8 @@ pub enum QueuedPullOutcome {
Failed(PullError),
}
pub type QueuedPullReport = Shared<oneshot::Receiver<QueuedPullOutcome>>;
/// Result of [`PullQueue::enqueue`].
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
pub enum EnqueueOutcome {
@@ -251,6 +253,7 @@ pub struct WriteBackRequest {
pub preserve_etag: bool,
/// `policy.emit_events`.
pub emit_events: bool,
pub respect_delete_marker: bool,
/// Source tags to copy (`policy.copy_tags`), `None` to skip.
pub tags: Option<HashMap<String, String>>,
}
@@ -266,6 +269,7 @@ impl WriteBackRequest {
pulled_at: OffsetDateTime::now_utc(),
preserve_etag: config.policy.preserve_etag,
emit_events: config.policy.emit_events,
respect_delete_marker: config.policy.respect_local_delete_marker,
tags,
}
}
@@ -830,7 +834,7 @@ pub struct PullQueue {
bucket: String,
tx: mpsc::Sender<PullJob>,
/// Keys queued or running; the job removes its key when it ends.
pending: Mutex<HashSet<String>>,
pending: Mutex<HashMap<String, QueuedPullReport>>,
capacity: usize,
cancel: CancellationToken,
stats: Arc<super::stats::OdmStats>,
@@ -869,7 +873,7 @@ impl PullQueue {
let queue = Arc::new(Self {
bucket: state.bucket().to_string(),
tx,
pending: Mutex::new(HashSet::new()),
pending: Mutex::new(HashMap::new()),
capacity,
cancel: state.cancel_token(),
stats: Arc::clone(state.stats()),
@@ -903,29 +907,24 @@ impl PullQueue {
self.enqueue_with_report(key, reason).0
}
/// [`Self::enqueue`] that also hands back the job's report channel when
/// a new job was queued (`Coalesced` pulls report to their first
/// requester only).
pub fn enqueue_with_report(
&self,
key: &str,
reason: PullReason,
) -> (EnqueueOutcome, Option<oneshot::Receiver<QueuedPullOutcome>>) {
/// [`Self::enqueue`] with a shared report, including for coalesced pulls.
pub fn enqueue_with_report(&self, key: &str, reason: PullReason) -> (EnqueueOutcome, Option<QueuedPullReport>) {
if self.cancel.is_cancelled() {
return (EnqueueOutcome::Unavailable, None);
}
let mut pending = self.pending.lock();
if pending.contains(key) {
return (EnqueueOutcome::Coalesced, None);
if let Some(report) = pending.get(key) {
return (EnqueueOutcome::Coalesced, Some(report.clone()));
}
let (report_tx, report_rx) = oneshot::channel();
let report_rx = report_rx.shared();
match self.tx.try_send(PullJob {
key: key.to_string(),
reason,
report: Some(report_tx),
}) {
Ok(()) => {
pending.insert(key.to_string());
pending.insert(key.to_string(), report_rx.clone());
(EnqueueOutcome::Enqueued, Some(report_rx))
}
Err(TrySendError::Full(_)) => {
@@ -1072,7 +1071,7 @@ impl BucketOdmState {
self: &Arc<Self>,
key: &str,
reason: PullReason,
) -> (EnqueueOutcome, Option<oneshot::Receiver<QueuedPullOutcome>>) {
) -> (EnqueueOutcome, Option<QueuedPullReport>) {
match self.pull_queue() {
Some(queue) => queue.enqueue_with_report(key, reason),
None => (EnqueueOutcome::Unavailable, None),
@@ -1399,13 +1398,21 @@ mod tests {
assert_eq!(queue.capacity(), 1024);
let mut outcomes = HashMap::new();
let mut shared_report = None;
for _ in 0..100 {
*outcomes.entry(queue.enqueue("a", PullReason::RangeGet)).or_insert(0) += 1;
let (outcome, report) = queue.enqueue_with_report("a", PullReason::RangeGet);
*outcomes.entry(outcome).or_insert(0) += 1;
shared_report = report;
}
assert_eq!(outcomes.get(&EnqueueOutcome::Enqueued), Some(&1));
assert_eq!(outcomes.get(&EnqueueOutcome::Coalesced), Some(&99));
assert_eq!(queue.pending_keys(), 1);
assert_eq!(
shared_report.expect("coalesced report").await,
Ok(QueuedPullOutcome::Stored { size: 1000 })
);
wait_until("first pull to finish", || queue.pending_keys() == 0).await;
assert_eq!(source.head_calls.load(Ordering::SeqCst), 1);
assert_eq!(source.get_calls.load(Ordering::SeqCst), 1);
@@ -1438,6 +1445,23 @@ mod tests {
assert_eq!(queue.enqueue("a", PullReason::RangeGet), EnqueueOutcome::Unavailable);
}
#[tokio::test]
async fn coalesced_enqueues_share_failure_reports() {
let sys = OnDemandMigrationSys::new();
let state = enabled_state(&sys, &config()).await;
let source = MockSource::with_object("missing", 1000, BodyKind::Bytes(body_bytes(1000)));
let queue = PullQueue::start(Arc::clone(&state), source, Arc::new(MockWriteBack::default()));
let (first, first_report) = queue.enqueue_with_report("absent", PullReason::RangeGet);
let (second, second_report) = queue.enqueue_with_report("absent", PullReason::Backfill);
assert_eq!(first, EnqueueOutcome::Enqueued);
assert_eq!(second, EnqueueOutcome::Coalesced);
let (first, second) = tokio::join!(first_report.expect("leader report"), second_report.expect("coalesced report"));
assert_eq!(first, second);
assert!(matches!(first, Ok(QueuedPullOutcome::Failed(_))));
sys.remove(BUCKET);
queue.wait_until_stopped().await;
}
#[tokio::test]
async fn queue_full_is_reported_and_cancel_drains_without_leaking_tasks() {
let sys = OnDemandMigrationSys::new();
@@ -1467,7 +1491,8 @@ mod tests {
wait_until("dispatcher to wait for a slot", || state.stats().queue_depth() == 1).await;
assert_eq!(queue.enqueue("c", PullReason::LargeObject), EnqueueOutcome::Enqueued);
assert_eq!(queue.enqueue("d", PullReason::LargeObject), EnqueueOutcome::QueueFull);
assert_eq!(queue.enqueue("c", PullReason::LargeObject), EnqueueOutcome::Coalesced);
let (coalesced, canceled_report) = queue.enqueue_with_report("c", PullReason::LargeObject);
assert_eq!(coalesced, EnqueueOutcome::Coalesced);
assert_eq!(queue.pending_keys(), 3);
assert_eq!(failures(&state).get("queue_full"), Some(&1));
assert!(!queue.is_stopped());
@@ -1477,6 +1502,12 @@ mod tests {
.await
.expect("dispatcher and in-flight job must exit after cancel");
assert!(queue.is_stopped());
assert!(
tokio::time::timeout(Duration::from_secs(5), canceled_report.expect("coalesced cancellation report"))
.await
.expect("cancellation closes the report")
.is_err()
);
assert_eq!(queue.pending_keys(), 0);
assert_eq!(state.inflight_keys(), 0);
assert_eq!(state.stats().inflight_pulls(), 0);
@@ -153,8 +153,8 @@ pub struct SourceClientSpec {
/// Wire requests one logical source call may cost. The pull pipeline and
/// the backfill job own the retry budget (`pull.rs` `PULL_MAX_RETRIES`,
/// `backfill.rs` `LIST_MAX_RETRIES`) and the breaker counts logical calls,
/// so ODM declares [`RemoteS3RetryPolicy::Disabled`] and keeps one counted
/// failure equal to one request against a struggling source.
/// so ODM declares [`RemoteS3RetryPolicy::Disabled`]. An ambiguous HEAD
/// 404 additionally probes the bucket before declaring a key absent.
pub retry: RemoteS3RetryPolicy,
/// Bytes per second the pull pipeline may consume from this source;
/// `None` means unlimited. Enforced by the consumer, not by this client.
@@ -262,7 +262,7 @@ const THROTTLE_CODES: &[&str] = &[
"TooManyRequests",
"RequestThrottled",
];
const NOT_FOUND_CODES: &[&str] = &["NoSuchKey", "NotFound", "NoSuchBucket", "NoSuchVersion"];
const NOT_FOUND_CODES: &[&str] = &["NoSuchKey"];
const ACCESS_DENIED_CODES: &[&str] = &[
"AccessDenied",
"InvalidAccessKeyId",
@@ -285,7 +285,6 @@ fn classify_status(status: u16, code: Option<&str>, message: String) -> SourceEr
}
}
match status {
404 => SourceError::NotFound,
401 | 403 => SourceError::AccessDenied,
429 | 503 => SourceError::Throttled,
500..=599 => SourceError::ServerError(status),
@@ -631,8 +630,7 @@ impl SourceClient {
}
/// `config` must come from [`SourceClientSpec::endpoint_spec`], which is
/// where the retry policy that keeps one logical call equal to one wire
/// request is declared.
/// where the policy disabling SDK-level retries is declared.
fn from_config_builder(config: aws_sdk_s3::config::Builder, endpoint: String, spec: &SourceClientSpec) -> Self {
let client = S3Client::from_conf(config.interceptor(SourceProxyMarkerInterceptor::new()).build());
Self {
@@ -754,15 +752,16 @@ impl SourceClient {
#[async_trait::async_trait]
impl SourceBackend for S3SourceBackend {
async fn head(&self, key: &str) -> Result<SourceHead, SourceError> {
let output = self
.client
.head_object()
.bucket(&self.bucket)
.key(key)
.send()
.await
.map_err(classify_sdk_error)?;
source_head_from_head_output(output)
match self.client.head_object().bucket(&self.bucket).key(key).send().await {
Ok(output) => source_head_from_head_output(output),
Err(err) if err.raw_response().is_some_and(|response| response.status().as_u16() == 404) => {
// HEAD has no error body: a missing bucket must not poison
// the per-key negative cache as though only the key was absent.
self.probe().await?;
Err(SourceError::NotFound)
}
Err(err) => Err(classify_sdk_error(err)),
}
}
/// Streams the object; `range` is passed through as an HTTP `Range`
@@ -809,8 +808,8 @@ impl SourceBackend for S3SourceBackend {
.contents
.unwrap_or_default()
.into_iter()
.filter_map(s3_source_object)
.collect();
.map(s3_source_object)
.collect::<Result<Vec<_>, _>>()?;
let common_prefixes = output
.common_prefixes
.unwrap_or_default()
@@ -849,14 +848,20 @@ impl SourceBackend for S3SourceBackend {
}
}
fn s3_source_object(object: SdkObject) -> Option<SourceObject> {
let key = object.key?;
fn s3_source_object(object: SdkObject) -> Result<SourceObject, SourceError> {
let key = object
.key
.ok_or_else(|| SourceError::Other("source listing object has no key".to_string()))?;
let size = object
.size
.and_then(|size| u64::try_from(size).ok())
.ok_or_else(|| SourceError::Other("source listing object has no valid size".to_string()))?;
let etag = normalize_etag(object.e_tag);
let is_multipart_etag = etag.as_deref().is_some_and(is_multipart_etag);
Some(SourceObject {
Ok(SourceObject {
key,
etag,
size: object.size.and_then(|size| u64::try_from(size).ok()).unwrap_or(0),
size,
last_modified: system_time(object.last_modified),
storage_class: object.storage_class.map(|class| class.as_str().to_string()),
is_multipart_etag,
@@ -1489,7 +1494,10 @@ mod tests {
#[tokio::test]
async fn source_error_classification_covers_every_class() {
let cases: Vec<(Scripted, &str, bool)> = vec![
(status(404, ""), "not_found", false),
(status(404, ""), "other", false),
(status(404, "<Error><Code>NoSuchKey</Code></Error>"), "not_found", false),
(status(404, "<Error><Code>NoSuchBucket</Code></Error>"), "other", false),
(status(404, "<Error><Code>NoSuchVersion</Code></Error>"), "other", false),
(status(403, ACCESS_DENIED_BODY), "access_denied", false),
(status(401, ""), "access_denied", false),
(status(429, ""), "throttled", true),
@@ -1512,14 +1520,35 @@ mod tests {
}
}
// HEAD carries no error body, so the classification must work from the
// status alone as well.
let (client, _) = scripted_client(&spec(None), vec![status(404, "")]).await;
let (client, requests) = scripted_client(&spec(None), vec![status(404, ""), status(200, "")]).await;
assert!(matches!(client.head_object("missing").await, Err(SourceError::NotFound)));
assert_eq!(recorded(&requests).len(), 2, "ambiguous HEAD 404 must check the bucket");
let (client, _) = scripted_client(&spec(None), vec![status(404, ""), status(404, "")]).await;
assert!(matches!(client.head_object("missing").await, Err(SourceError::Other(_))));
let (client, _) = scripted_client(&spec(None), vec![status(404, ""), status(403, "")]).await;
assert!(matches!(client.head_object("missing").await, Err(SourceError::AccessDenied)));
let (client, _) = scripted_client(&spec(None), vec![status(403, "")]).await;
assert!(matches!(client.head_object("secret").await, Err(SourceError::AccessDenied)));
}
#[test]
fn source_listing_rejects_missing_and_negative_sizes() {
for size in [None, Some(-1)] {
let object = SdkObject::builder().key("key").set_size(size).build();
assert!(matches!(s3_source_object(object), Err(SourceError::Other(_))));
}
assert!(matches!(
s3_source_object(SdkObject::builder().size(0).build()),
Err(SourceError::Other(_))
));
assert_eq!(
s3_source_object(SdkObject::builder().key("empty").size(0).build())
.expect("empty object")
.size,
0
);
}
#[tokio::test]
async fn source_client_debug_redacts_credentials() {
let (client, _) = scripted_client(&spec(Some("data/")), Vec::new()).await;
@@ -20,9 +20,9 @@ pub use rustfs_replication::{
pub(crate) use rustfs_replication::{
ReplicationDeleteSource, ReplicationMultipartPartInput, ReplicationResyncTargetObject, delete_marker_purge_mrf_entry,
delete_marker_purge_version_id, delete_replication_creates_marker, delete_replication_missing_source_decision,
delete_replication_object_opts, delete_replication_target_version_id, heal_uses_delete_replication_path,
is_object_lock_denied_delete, is_retryable_delete_replication_head_error, is_version_delete_replication,
replicate_delete_outcome, replication_etags_match, replication_multipart_complete_actual_size,
replication_multipart_part_plan, replication_single_put_size_error, resync_existing_delete_replication_info,
resync_target_for_object, should_retry_delete_marker_purge, single_part_replica_etag_mismatch,
delete_replication_object_opts, heal_uses_delete_replication_path, is_object_lock_denied_delete,
is_retryable_delete_replication_head_error, is_version_delete_replication, replicate_delete_outcome, replication_etags_match,
replication_multipart_complete_actual_size, replication_multipart_part_plan, replication_single_put_size_error,
resync_existing_delete_replication_info, resync_target_for_object, should_retry_delete_marker_purge,
single_part_replica_etag_mismatch, target_delete_version_id,
};
@@ -882,20 +882,6 @@ fn reconstructed_heal_delete_info(
) -> DeletedObjectReplicationInfo {
let mut rstate = oi.replication_state();
rstate.replicate_decision_str = dsc.to_string();
// The caller hands us a blank ObjectInfo (the source marker may already be
// gone), so the state above carries no target-assigned marker version ids.
// Restore them from the journal: `delete_marker_purge_version_id` must hit
// the id the target reported, not fall back to the source marker id, which
// a target that mints its own ids answers with an idempotent 204 that would
// acknowledge the intent while the real marker stays behind (backlog#2290).
// The corrupt flag rides along so a refusal stays a refusal after restart.
for (arn, version_id) in &entry.target_delete_marker_version_ids {
rstate
.target_delete_marker_version_ids
.entry(arn.clone())
.or_insert_with(|| version_id.clone());
}
rstate.target_delete_marker_version_ids_corrupt |= entry.target_delete_marker_version_ids_corrupt;
let delete_marker_mtime = entry
.delete_marker_mtime
@@ -6615,87 +6601,4 @@ mod tests {
replacement_data
);
}
/// backlog#2290: a delete-marker purge intent that survives a restart
/// through the MRF journal addresses the marker version the TARGET
/// assigned, exactly as the live watcher does (see the
/// `requires_delayed_purge` spawn). The journal carries the per-ARN ids
/// (`targetDeleteMarkerVersionIDs`) and replay restores them into the
/// reconstructed replication state; without that the replay would fall
/// back to the source marker id, which a target that mints its own ids
/// answers with an idempotent 204 — the entry would be acknowledged while
/// the real marker stayed behind.
#[test]
fn mrf_delete_marker_purge_replay_preserves_target_assigned_marker_version() {
use super::super::replication_object_decision_boundary::{delete_marker_purge_mrf_entry, delete_marker_purge_version_id};
let arn = "arn:minio:replication::generic-target:photos".to_string();
let source_marker = uuid::Uuid::new_v4();
let remote_marker = "remote-assigned-marker-version".to_string();
let live_oi = ObjectInfo {
bucket: "photos".to_string(),
name: "obj".to_string(),
version_id: Some(source_marker),
delete_marker: true,
..Default::default()
};
let mut live_state = live_oi.replication_state();
live_state.replicate_decision_str = replicate_decision_for_admitted_targets(std::slice::from_ref(&arn)).to_string();
live_state
.target_delete_marker_version_ids
.insert(arn.clone(), remote_marker.clone());
let live = DeletedObjectReplicationInfo {
delete_object: ReplicationDeletedObject {
object_name: "obj".to_string(),
delete_marker: true,
delete_marker_version_id: Some(source_marker),
replication_state: Some(live_state),
..Default::default()
},
bucket: "photos".to_string(),
..Default::default()
};
assert_eq!(
delete_marker_purge_version_id(live.delete_object.replication_state.as_ref(), &arn, source_marker),
Some(Some(remote_marker.clone())),
"the live purge addresses the recorded target version"
);
// Watch window exhausted: persist the intent, restart, replay it.
let entry = delete_marker_purge_mrf_entry(&live, vec![arn.clone()]);
let replay_oi = ObjectInfo {
bucket: entry.bucket.clone(),
name: entry.object.clone(),
version_id: entry.version_id,
delete_marker: entry.delete_marker,
..Default::default()
};
let dsc = replicate_decision_for_admitted_targets(&entry.target_arns);
let replayed = reconstructed_heal_delete_info(&entry, &replay_oi, &dsc);
assert_eq!(
delete_marker_purge_version_id(replayed.delete_object.replication_state.as_ref(), &arn, source_marker),
Some(Some(remote_marker)),
"the MRF replay must address the target-assigned marker version, not source marker {source_marker}"
);
// A refusal (inconsistent recorded ids) must stay a refusal across the
// journal round trip instead of degrading into the source-id fallback.
let mut refused = live;
refused
.delete_object
.replication_state
.as_mut()
.expect("state was set above")
.target_delete_marker_version_ids_corrupt = true;
let entry = delete_marker_purge_mrf_entry(&refused, vec![arn.clone()]);
assert!(entry.target_delete_marker_version_ids_corrupt);
let replayed = reconstructed_heal_delete_info(&entry, &replay_oi, &dsc);
assert_eq!(
delete_marker_purge_version_id(replayed.delete_object.replication_state.as_ref(), &arn, source_marker),
None,
"the MRF replay must keep refusing to guess when the recorded ids were inconsistent"
);
}
}
@@ -32,11 +32,11 @@ use super::replication_msgp_boundary::ReplicationMsgpCodec;
use super::replication_object_config::{ReplicationConfig, get_replication_config, must_replicate};
use super::replication_object_decision_boundary::{
MustReplicateOptions, ReplicationMultipartPartInput, delete_marker_purge_mrf_entry, delete_marker_purge_version_id,
delete_replication_creates_marker, delete_replication_target_version_id, heal_uses_delete_replication_path,
is_object_lock_denied_delete, is_retryable_delete_replication_head_error, is_version_delete_replication,
replicate_delete_outcome, replication_etags_match, replication_multipart_complete_actual_size,
replication_multipart_part_plan, replication_single_put_size_error, resync_existing_delete_replication_info,
should_retry_delete_marker_purge, single_part_replica_etag_mismatch,
delete_replication_creates_marker, heal_uses_delete_replication_path, is_object_lock_denied_delete,
is_retryable_delete_replication_head_error, is_version_delete_replication, replicate_delete_outcome, replication_etags_match,
replication_multipart_complete_actual_size, replication_multipart_part_plan, replication_single_put_size_error,
resync_existing_delete_replication_info, should_retry_delete_marker_purge, single_part_replica_etag_mismatch,
target_delete_version_id,
};
use super::replication_queue_boundary::{DeletedObjectReplicationInfo, ReplicationQueueAdmission};
use super::replication_resync_boundary::ResyncStatusType;
@@ -2051,11 +2051,7 @@ pub(crate) async fn replicate_delete_with_outcome<S: ReplicationStorage>(
let is_version_purge = is_version_delete_replication(&dobj.delete_object);
// The watcher exists to purge a replicated marker once the SOURCE marker
// vanishes. A version purge is that purge already (its failures reach the
// journal as a purge entry), so it must not spawn a second watcher that
// journals a duplicate intent (backlog#2290).
let requires_delayed_purge = should_retry_delete_marker_purge(&dobj.delete_object) && !is_version_purge;
let requires_delayed_purge = should_retry_delete_marker_purge(&dobj.delete_object);
let (replication_status, prev_status) = if !is_version_purge {
(
@@ -2765,6 +2761,12 @@ fn unavailable_delete_target_info(dobj: &DeletedObjectReplicationInfo, arn: &str
}
async fn replicate_delete_to_target(dobj: &DeletedObjectReplicationInfo, tgt_client: Arc<TargetClient>) -> ReplicatedTargetInfo {
let version_id = if let Some(version_id) = &dobj.delete_object.delete_marker_version_id {
version_id.to_owned()
} else {
dobj.delete_object.version_id.unwrap_or_default()
};
let mut rinfo = dobj
.delete_object
.replication_state
@@ -2797,25 +2799,7 @@ async fn replicate_delete_to_target(dobj: &DeletedObjectReplicationInfo, tgt_cli
return rinfo;
}
// Purging a replicated delete marker addresses the version the target
// assigned (recorded when the marker was created there); see
// `delete_replication_target_version_id`. A corrupt record is a failure,
// not a guess: the entry stays visible until the metadata is repaired.
let Some(version_id) = delete_replication_target_version_id(&dobj.delete_object, &tgt_client.arn) else {
warn!(
event = EVENT_DELETE_MARKER_PURGE_FAILED,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
bucket = tgt_client.bucket,
object = dobj.delete_object.object_name,
arn = %tgt_client.arn,
reason = "recorded_target_version_inconsistent",
"Replicated version purge refused: recorded target delete-marker version metadata is inconsistent"
);
rinfo.version_purge_status = VersionPurgeStatusType::Failed;
rinfo.error = Some("recorded target delete-marker version metadata is inconsistent".to_string());
return rinfo;
};
let version_id = target_delete_version_id(version_id, is_version_purge);
if dobj.delete_object.delete_marker && dobj.delete_object.delete_marker_version_id.is_some() {
match head_object_for_worker(
+3
View File
@@ -956,6 +956,9 @@ pub struct ObjectOptions {
pub preserve_etag: Option<String>,
pub metadata_chg: bool,
pub http_preconditions: Option<HTTPPreconditions>,
/// Internal create-only writes may also preserve an acknowledged deletion.
/// Evaluated with `http_preconditions` under the namespace commit lock.
pub preserve_delete_marker: bool,
pub delete_replication: Option<ReplicationState>,
pub delete_replication_config_snapshot: Option<Arc<DeleteReplicationConfigSnapshot>>,
+112
View File
@@ -78,6 +78,21 @@ pub(crate) struct ScannerPublicationLeaseEntry {
pub(crate) _operation_guard: OwnedRwLockReadGuard<()>,
}
pub(crate) struct NamespaceCommitGuard {
ctx: Arc<InstanceContext>,
counted: bool,
}
impl Drop for NamespaceCommitGuard {
fn drop(&mut self) {
if self.counted {
// Publish the new generation before a zero-pending publication probe.
self.ctx.advance_namespace_commit_generation();
self.ctx.namespace_commits.fetch_sub(1, Ordering::AcqRel);
}
}
}
/// Runtime state owned by a single `ECStore` instance.
///
/// This is intentionally minimal in the first migration slice; subsequent
@@ -209,9 +224,13 @@ pub struct InstanceContext {
/// Last storage-owned movement snapshot observed under the operation
/// gate. SetDisks cache writers fail closed until ECStore refreshes it.
scanner_publication_state: AtomicU8,
namespace_commits: AtomicU64,
namespace_commit_generation: AtomicU64,
/// Resolves object-encryption material at the application boundary.
object_encryption_resolver: OnceLock<Arc<dyn ObjectEncryptionResolver>>,
tier_delete_journal_recovery_stores: std::sync::Mutex<HashSet<Uuid>>,
#[cfg(test)]
suppress_tier_delete_journal_recovery: bool,
transition_transaction_recovery_stores: std::sync::Mutex<HashSet<Uuid>>,
tier_delete_journal_recovery_wakeup: tokio::sync::Notify,
}
@@ -256,8 +275,12 @@ impl InstanceContext {
data_movement_generation_exhausted: AtomicBool::new(false),
data_movement_generation_notify: Arc::new(Notify::new()),
scanner_publication_state: AtomicU8::new(SCANNER_PUBLICATION_STATE_UNKNOWN),
namespace_commits: AtomicU64::new(0),
namespace_commit_generation: AtomicU64::new(0),
object_encryption_resolver: OnceLock::new(),
tier_delete_journal_recovery_stores: std::sync::Mutex::new(HashSet::new()),
#[cfg(test)]
suppress_tier_delete_journal_recovery: false,
transition_transaction_recovery_stores: std::sync::Mutex::new(HashSet::new()),
tier_delete_journal_recovery_wakeup: tokio::sync::Notify::new(),
}
@@ -385,6 +408,36 @@ impl InstanceContext {
&& self.scanner_publication_state.load(Ordering::Acquire) == SCANNER_PUBLICATION_STATE_ALLOWED
}
pub(crate) fn begin_namespace_commit(self: &Arc<Self>) -> Arc<NamespaceCommitGuard> {
let counted = self
.namespace_commits
.fetch_update(Ordering::AcqRel, Ordering::Acquire, |count| count.checked_add(1))
.is_ok();
if counted {
self.advance_namespace_commit_generation();
} else {
self.namespace_commit_generation.store(u64::MAX, Ordering::Release);
}
Arc::new(NamespaceCommitGuard {
ctx: Arc::clone(self),
counted,
})
}
fn advance_namespace_commit_generation(&self) {
let _ = self
.namespace_commit_generation
.fetch_update(Ordering::AcqRel, Ordering::Acquire, |generation| Some(generation.saturating_add(1)));
}
pub(crate) fn namespace_commit_generation(&self) -> u64 {
self.namespace_commit_generation.load(Ordering::Acquire)
}
pub(crate) fn namespace_commits_pending(&self) -> bool {
self.namespace_commits.load(Ordering::Acquire) != 0 || self.namespace_commit_generation() == u64::MAX
}
pub(crate) fn set_scanner_publication_state(&self, blocked: bool) {
self.scanner_publication_state.store(
if blocked {
@@ -640,12 +693,21 @@ impl InstanceContext {
}
pub(crate) fn mark_tier_delete_journal_recovery_started(&self, store_id: Uuid) -> bool {
#[cfg(test)]
if self.suppress_tier_delete_journal_recovery {
return false;
}
self.tier_delete_journal_recovery_stores
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.insert(store_id)
}
#[cfg(test)]
pub(crate) fn suppress_tier_delete_journal_recovery_for_test(&mut self) {
self.suppress_tier_delete_journal_recovery = true;
}
pub(crate) fn mark_transition_transaction_recovery_started(&self, store_id: Uuid) -> bool {
self.transition_transaction_recovery_stores
.lock()
@@ -756,6 +818,50 @@ pub fn bootstrap_ctx() -> Arc<InstanceContext> {
mod tests {
use super::*;
#[test]
fn namespace_commit_guards_are_instance_local_and_count_until_last_owner() {
let first = Arc::new(InstanceContext::new());
let other = Arc::new(InstanceContext::new());
first.set_scanner_publication_state(false);
other.set_scanner_publication_state(false);
assert!(first.scanner_publication_state_allowed());
let one = first.begin_namespace_commit();
let shared_owner = Arc::clone(&one);
let two = first.begin_namespace_commit();
assert!(first.namespace_commits_pending());
assert!(first.scanner_publication_state_allowed(), "pending writes must not block scan admission");
assert_eq!(first.namespace_commit_generation(), 2);
assert!(!other.namespace_commits_pending());
assert_eq!(other.namespace_commit_generation(), 0);
assert!(other.scanner_publication_state_allowed());
drop(one);
assert_eq!(first.namespace_commit_generation(), 2);
drop(shared_owner);
assert!(first.namespace_commits_pending());
assert_eq!(first.namespace_commit_generation(), 3);
drop(two);
assert!(!first.namespace_commits_pending());
assert_eq!(first.namespace_commit_generation(), 4);
assert!(first.scanner_publication_state_allowed());
}
#[test]
fn namespace_commit_counter_exhaustion_keeps_publication_blocked() {
for (count, generation) in [(0, u64::MAX - 1), (u64::MAX, 0)] {
let ctx = Arc::new(InstanceContext::new());
ctx.set_scanner_publication_state(false);
ctx.namespace_commits.store(count, Ordering::Release);
ctx.namespace_commit_generation.store(generation, Ordering::Release);
let guard = ctx.begin_namespace_commit();
assert!(ctx.namespace_commits_pending());
assert_eq!(ctx.namespace_commit_generation(), u64::MAX);
drop(guard);
assert!(ctx.namespace_commits_pending());
assert_eq!(ctx.namespace_commit_generation(), u64::MAX);
assert_eq!(ctx.namespace_commits.load(Ordering::Acquire), count);
}
}
// The SetupType inputs must derive the exact (is_erasure,
// is_dist_erasure, is_erasure_sd) triples that the original three
// process-global erasure bools produced via update_erasure_type().
@@ -1073,6 +1179,12 @@ mod tests {
assert!(!ctx_a.mark_tier_delete_journal_recovery_started(store_a));
assert!(ctx_a.mark_tier_delete_journal_recovery_started(store_b));
assert!(ctx_b.mark_tier_delete_journal_recovery_started(store_a));
let mut manual_ctx = InstanceContext::new();
manual_ctx.suppress_tier_delete_journal_recovery_for_test();
assert!(!manual_ctx.mark_tier_delete_journal_recovery_started(store_a));
assert!(!manual_ctx.mark_tier_delete_journal_recovery_started(store_b));
assert!(ctx_b.mark_tier_delete_journal_recovery_started(store_b));
}
#[test]
+245 -73
View File
@@ -3558,6 +3558,11 @@ impl RenameRollbackReceipt {
}
}
struct RenameRollbackOwnership {
receipt: Option<RenameRollbackReceipt>,
namespace_commit_guard: Option<Arc<crate::runtime::instance::NamespaceCommitGuard>>,
}
async fn inspect_incomplete_rename_rollback(
disks: &[Option<DiskStore>],
bucket: &str,
@@ -3604,8 +3609,12 @@ async fn rollback_failed_rename(
dispatch_states: &[RenameDispatchState],
rollback_dirs: &[Option<Uuid>],
dst: (&str, &str),
receipt: Option<RenameRollbackReceipt>,
ownership: RenameRollbackOwnership,
) {
let RenameRollbackOwnership {
receipt,
namespace_commit_guard,
} = ownership;
let owned_disks = disks.to_vec();
let owned_errs = errs.to_vec();
let owned_dispatch_states = dispatch_states.to_vec();
@@ -3651,7 +3660,9 @@ async fn rollback_failed_rename(
let fi = std::mem::take(&mut file_infos[disk_index]);
let bucket = bucket.to_string();
let object = object.to_string();
let disk_namespace_commit_guard = namespace_commit_guard.clone();
let task = tokio::spawn(async move {
let _namespace_commit_guard = disk_namespace_commit_guard;
#[allow(clippy::let_unit_value)]
let _task_guard = SetDisks::rename_fanout_task_guard(&object);
SetDisks::rename_fanout_barrier(&object, disk_index, rename_fanout_barrier_phase::ROLLBACK).await;
@@ -3672,6 +3683,9 @@ async fn rollback_failed_rename(
});
tasks.push(async move { (disk_index, task.await) });
}
#[cfg(test)]
rollback_fault_injection::after_undo_dispatch(object);
let _namespace_commit_guard = namespace_commit_guard;
for (disk_index, result) in join_all(tasks).await {
outcomes[disk_index].outcome = rename_rollback_task_outcome(result);
}
@@ -3778,6 +3792,7 @@ pub(in crate::set_disk) struct RenameDataFenceOptions<'a> {
write_quorum: usize,
scanner_publication_lease_tokens: Option<&'a HashMap<String, Uuid>>,
scanner_publication_commit_scope: Option<crate::object_api::ScannerPublicationCommitScope>,
namespace_commit_guard: Option<Arc<crate::runtime::instance::NamespaceCommitGuard>>,
rollback_receipt: Option<RenameRollbackReceipt>,
}
@@ -3790,6 +3805,7 @@ impl<'a> RenameDataFenceOptions<'a> {
write_quorum,
scanner_publication_lease_tokens,
scanner_publication_commit_scope: None,
namespace_commit_guard: None,
rollback_receipt: None,
}
}
@@ -3806,6 +3822,14 @@ impl<'a> RenameDataFenceOptions<'a> {
self.scanner_publication_commit_scope = scanner_publication_commit_scope;
self
}
pub(in crate::set_disk) fn with_namespace_commit_guard(
mut self,
namespace_commit_guard: Option<Arc<crate::runtime::instance::NamespaceCommitGuard>>,
) -> Self {
self.namespace_commit_guard = namespace_commit_guard;
self
}
}
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
@@ -4164,6 +4188,7 @@ impl SetDisks {
write_quorum,
scanner_publication_lease_tokens,
scanner_publication_commit_scope: _scanner_publication_commit_scope,
namespace_commit_guard,
rollback_receipt,
} = fence_options;
if let Some(file_info) = disks
@@ -4210,7 +4235,9 @@ impl SetDisks {
let dst_object = fanout_dst_object.clone();
let file_info = file_info.clone();
let successful_rename_completion_rank = successful_rename_completion_rank.clone();
let namespace_commit_guard = namespace_commit_guard.clone();
tasks.spawn(async move {
let _namespace_commit_guard = namespace_commit_guard;
let mut dispatch_state = RenameDispatchState::NotDispatched;
let result = std::panic::AssertUnwindSafe(async {
#[allow(clippy::let_unit_value)]
@@ -4372,7 +4399,10 @@ impl SetDisks {
&dispatch_states,
&data_dirs,
(&fanout_dst_bucket, &fanout_dst_object),
rollback_receipt,
RenameRollbackOwnership {
receipt: rollback_receipt,
namespace_commit_guard,
},
)
.await;
if let Some(commit_tx) = commit_tx.take() {
@@ -4528,6 +4558,7 @@ impl SetDisks {
write_quorum,
scanner_publication_lease_tokens,
scanner_publication_commit_scope,
namespace_commit_guard,
rollback_receipt,
} = fence_options;
if let Some(file_info) = disks
@@ -4561,6 +4592,7 @@ impl SetDisks {
let fanout_dst_bucket = dst_bucket.clone();
let fanout_dst_object = dst_object.clone();
let fanout_publication_scope = scanner_publication_commit_scope.clone();
let fanout_namespace_commit_guard = namespace_commit_guard.clone();
// Keep one coordinator task so a cancelled caller cannot drop partially
// completed disk mutations. Per-disk futures stay ordered in `join_all`,
// preserving slot-indexed quorum and convergence accounting without a
@@ -4569,6 +4601,7 @@ impl SetDisks {
// Keep the storage-owned movement permit attached to the actual
// fan-out owner, even if the caller future is cancelled.
let _fanout_publication_scope = fanout_publication_scope;
let _namespace_commit_guard = fanout_namespace_commit_guard;
let successful_rename_completion_rank =
rustfs_io_metrics::put_stage_metrics_enabled().then(|| Arc::new(AtomicUsize::new(0)));
let futures = fanout_disks
@@ -4790,7 +4823,10 @@ impl SetDisks {
&dispatch_states,
&data_dirs,
(&dst_bucket, &dst_object),
rollback_receipt,
RenameRollbackOwnership {
receipt: rollback_receipt,
namespace_commit_guard,
},
)
.await;
return Err(ret_err);
@@ -6503,9 +6539,9 @@ impl SetDisks {
match oi {
Ok(oi) => {
// Ordinary writes may proceed past a top-level delete marker;
// data movement must not replace an acknowledged deletion.
// data movement and guarded internal writes must preserve it.
if oi.delete_marker {
return opts.data_movement.then_some(StorageError::PreconditionFailed);
return (opts.data_movement || opts.preserve_delete_marker).then_some(StorageError::PreconditionFailed);
}
let if_none_match = http_preconditions.if_none_match_value().map(str::to_owned);
let if_match = http_preconditions.if_match_value().map(str::to_owned);
@@ -6754,6 +6790,7 @@ pub(in crate::set_disk) mod rollback_fault_injection {
VolumeNotFoundAfterRename,
PanicAfterRename,
CoordinatorPanic,
RollbackCoordinatorPanic,
}
fn registry() -> &'static Mutex<HashMap<String, (usize, Fault)>> {
@@ -6816,6 +6853,17 @@ pub(in crate::set_disk) mod rollback_fault_injection {
panic!("injected rename coordinator panic");
}
}
pub(super) fn after_undo_dispatch(object: &str) {
let fault = registry()
.lock()
.expect("rollback registry should not poison")
.get(object)
.copied();
if matches!(fault, Some((_, Fault::RollbackCoordinatorPanic))) {
panic!("injected rollback coordinator panic");
}
}
}
/// Test-only per-disk call counters for the metadata fan-out (backlog#1325,
@@ -6977,7 +7025,7 @@ pub(crate) mod rename_fanout_barrier {
use tokio::sync::Notify;
pub use super::rename_fanout_barrier_phase::{
CLEANUP as PHASE_CLEANUP, READ_VERSION as PHASE_READ_VERSION, RENAME as PHASE_RENAME,
CLEANUP as PHASE_CLEANUP, READ_VERSION as PHASE_READ_VERSION, RENAME as PHASE_RENAME, ROLLBACK as PHASE_ROLLBACK,
};
/// One armed barrier: the fan-out task matching `(disk_index, phase)` pauses.
@@ -10814,79 +10862,177 @@ mod tests {
#[tokio::test]
#[serial_test::serial(capacity_dirty_scope)]
async fn rename_rollback_incomplete_receipt_waits_for_undo_barrier() {
for cancel_caller in [false, true] {
let bucket = "rename-rollback-barrier";
let object = if cancel_caller {
"rollback-barrier-cancelled"
} else {
"rollback-barrier-object"
};
let (dirs, disks) = call_counter_local_disks(bucket, 4).await;
prepare_rename_source_dirs(&dirs, &disks, "source").await;
let mut old = metadata_test_fileinfo(object);
old.mod_time = Some(OffsetDateTime::now_utc());
old.data = Some(Bytes::from_static(b"old-inline-body"));
old.set_inline_data();
old.metadata.insert("etag".to_string(), "old-etag".to_string());
for disk in disks.iter().flatten() {
disk.write_metadata(bucket, bucket, object, old.clone())
.await
.expect("old metadata should be staged");
}
let _rename_fault = rename_fault_injection::fail_rename_on(object, &[2, 3]);
let _undo_fault = rollback_fault_injection::arm(object, 0, rollback_fault_injection::Fault::Io);
let barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier_phase::ROLLBACK);
let receipt = RenameRollbackReceipt::default();
let mut rename = Box::pin(SetDisks::rename_data_owned_with_fence(
&disks,
(RUSTFS_META_TMP_BUCKET, "source"),
rename_commit_fileinfos(object, 4, "new-etag"),
(bucket, object),
false,
RenameDataFenceOptions::new(3, None).with_rollback_receipt(receipt.clone()),
));
tokio::time::timeout(BARRIER_PAUSE_GUARD, async {
tokio::select! {
() = barrier.wait_until_paused() => {}
_ = rename.as_mut() => panic!("rename returned before the armed rollback barrier"),
temp_env::async_with_vars([(ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("true"))], async {
for (allow_early_ack, cancel_caller, object) in [
(false, false, "rollback-barrier-object"),
(false, true, "rollback-barrier-cancelled"),
(true, false, "rollback-barrier-early-object"),
(true, true, "rollback-barrier-early-cancelled"),
] {
let ctx = Arc::new(crate::runtime::instance::InstanceContext::new());
let bucket = "rename-rollback-barrier";
let (dirs, disks) = call_counter_local_disks(bucket, 4).await;
prepare_rename_source_dirs(&dirs, &disks, "source").await;
let mut old = metadata_test_fileinfo(object);
old.mod_time = Some(OffsetDateTime::now_utc());
old.data = Some(Bytes::from_static(b"old-inline-body"));
old.set_inline_data();
old.metadata.insert("etag".to_string(), "old-etag".to_string());
for disk in disks.iter().flatten() {
disk.write_metadata(bucket, bucket, object, old.clone())
.await
.expect("old metadata should be staged");
}
})
.await
.expect("undo must reach its disk barrier");
assert!(receipt.0.get().is_none(), "pending undo must not be recorded as success");
if cancel_caller {
drop(rename);
let _rename_fault = rename_fault_injection::fail_rename_on(object, &[2, 3]);
let _undo_fault = rollback_fault_injection::arm(object, 0, rollback_fault_injection::Fault::Io);
let barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier_phase::ROLLBACK);
let receipt = RenameRollbackReceipt::default();
let mut rename = Box::pin(SetDisks::rename_data_owned_with_fence(
&disks,
(RUSTFS_META_TMP_BUCKET, "source"),
rename_commit_fileinfos(object, 4, "new-etag"),
(bucket, object),
allow_early_ack,
RenameDataFenceOptions::new(3, None)
.with_rollback_receipt(receipt.clone())
.with_namespace_commit_guard(Some(ctx.begin_namespace_commit())),
));
tokio::time::timeout(BARRIER_PAUSE_GUARD, async {
tokio::select! {
() = barrier.wait_until_paused() => {}
_ = rename.as_mut() => panic!("rename returned before the armed rollback barrier"),
}
})
.await
.expect("undo must reach its disk barrier");
assert!(receipt.0.get().is_none(), "pending undo must not be recorded as success");
assert!(ctx.namespace_commits_pending());
assert_eq!(ctx.namespace_commit_generation(), 1);
if cancel_caller {
drop(rename);
assert!(ctx.namespace_commits_pending(), "caller cancellation must not retire pending undo work");
assert_eq!(ctx.namespace_commit_generation(), 1);
barrier.release();
tokio::time::timeout(BARRIER_PAUSE_GUARD, async {
while receipt.0.get().is_none() || ctx.namespace_commits_pending() {
tokio::task::yield_now().await;
}
})
.await
.expect("cancelled caller must not cancel rollback accounting");
} else {
barrier.release();
assert!(rename.await.is_err());
}
assert!(
!ctx.namespace_commits_pending(),
"the completed rollback must release its namespace ownership"
);
assert_eq!(ctx.namespace_commit_generation(), 2);
assert!(receipt.is_incomplete(), "drained undo failure must survive in the receipt");
for dir in dirs.iter().skip(1) {
let reopened = reopen_local_disk(dir).await;
let restored = reopened
.read_version(
"",
bucket,
object,
"",
&ReadOptions {
read_data: true,
..Default::default()
},
)
.await
.expect("old version must remain readable after caller cancellation");
assert_eq!(restored.data.as_deref(), Some(b"old-inline-body".as_slice()));
}
}
})
.await;
}
#[tokio::test]
#[serial_test::serial(capacity_dirty_scope)]
async fn rename_rollback_children_keep_namespace_ownership_after_coordinator_panic() {
temp_env::async_with_vars([(ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("true"))], async {
for (allow_early_ack, object) in [
(false, "rollback-coordinator-panic"),
(true, "rollback-coordinator-panic-early"),
] {
let ctx = Arc::new(crate::runtime::instance::InstanceContext::new());
let bucket = "rename-rollback-coordinator-panic";
let (dirs, disks) = call_counter_local_disks(bucket, 4).await;
prepare_rename_source_dirs(&dirs, &disks, "source").await;
let mut old = metadata_test_fileinfo(object);
old.mod_time = Some(OffsetDateTime::now_utc());
old.data = Some(Bytes::from_static(b"old-inline-body"));
old.set_inline_data();
old.metadata.insert("etag".to_string(), "old-etag".to_string());
for disk in disks.iter().flatten() {
disk.write_metadata(bucket, bucket, object, old.clone())
.await
.expect("old metadata should be staged");
}
let _rename_fault = rename_fault_injection::fail_rename_on(object, &[2, 3]);
let _rollback_fault =
rollback_fault_injection::arm(object, 0, rollback_fault_injection::Fault::RollbackCoordinatorPanic);
let barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier_phase::ROLLBACK);
let receipt = RenameRollbackReceipt::default();
let result = tokio::time::timeout(
BARRIER_PAUSE_GUARD,
SetDisks::rename_data_owned_with_fence(
&disks,
(RUSTFS_META_TMP_BUCKET, "source"),
rename_commit_fileinfos(object, 4, "new-etag"),
(bucket, object),
allow_early_ack,
RenameDataFenceOptions::new(3, None)
.with_rollback_receipt(receipt.clone())
.with_namespace_commit_guard(Some(ctx.begin_namespace_commit())),
),
)
.await
.expect("coordinator failure must return without waiting for detached undo tasks");
assert!(result.is_err());
tokio::time::timeout(BARRIER_PAUSE_GUARD, barrier.wait_until_paused())
.await
.expect("detached undo must reach its disk barrier");
assert!(
receipt.is_incomplete(),
"coordinator failure must preserve indeterminate recovery evidence"
);
assert!(ctx.namespace_commits_pending(), "the paused child must retain namespace ownership");
assert_eq!(ctx.namespace_commit_generation(), 1);
barrier.release();
tokio::time::timeout(BARRIER_PAUSE_GUARD, async {
while receipt.0.get().is_none() {
while ctx.namespace_commits_pending() {
tokio::task::yield_now().await;
}
})
.await
.expect("cancelled caller must not cancel rollback accounting");
} else {
barrier.release();
assert!(rename.await.is_err());
.expect("completed undo children must release their namespace ownership");
assert_eq!(ctx.namespace_commit_generation(), 2);
for dir in &dirs {
let reopened = reopen_local_disk(dir).await;
let restored = reopened
.read_version(
"",
bucket,
object,
"",
&ReadOptions {
read_data: true,
..Default::default()
},
)
.await
.expect("old version must remain readable after rollback coordinator failure");
assert_eq!(restored.data.as_deref(), Some(b"old-inline-body".as_slice()));
}
}
assert!(receipt.is_incomplete(), "drained undo failure must survive in the receipt");
for dir in dirs.iter().skip(1) {
let reopened = reopen_local_disk(dir).await;
let restored = reopened
.read_version(
"",
bucket,
object,
"",
&ReadOptions {
read_data: true,
..Default::default()
},
)
.await
.expect("old version must remain readable after caller cancellation");
assert_eq!(restored.data.as_deref(), Some(b"old-inline-body".as_slice()));
}
}
})
.await;
}
#[tokio::test]
@@ -11001,9 +11147,35 @@ mod tests {
let mut file_infos = rename_commit_fileinfos(object, DISKS, "fresh-rollback-etag");
file_infos[3] = FileInfo::default();
SetDisks::rename_data(&disks, RUSTFS_META_TMP_BUCKET, "source", &file_infos, bucket, object, 4)
let ctx = Arc::new(crate::runtime::instance::InstanceContext::new());
ctx.set_scanner_publication_state(false);
let barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_ROLLBACK);
let rename = SetDisks::rename_data_owned_with_fence(
&disks,
(RUSTFS_META_TMP_BUCKET, "source"),
file_infos,
(bucket, object),
false,
RenameDataFenceOptions::new(4, None).with_namespace_commit_guard(Some(ctx.begin_namespace_commit())),
);
let control = async {
barrier.wait_until_paused().await;
assert!(ctx.namespace_commits_pending(), "rollback must retain namespace publication ownership");
assert!(ctx.scanner_publication_state_allowed(), "rollback must not disable namespace walks");
assert_eq!(ctx.namespace_commit_generation(), 1);
barrier.release();
};
let (result, ()) = tokio::time::timeout(BARRIER_PAUSE_GUARD, async { tokio::join!(rename, control) })
.await
.expect_err("three successful disks must fail a strict write quorum of four");
.expect("rename rollback must reach its barrier and finish after release");
assert_eq!(
result.err(),
Some(DiskError::ErasureWriteQuorum),
"three successful disks must fail a strict write quorum of four"
);
assert!(!ctx.namespace_commits_pending());
assert!(ctx.scanner_publication_state_allowed());
assert_eq!(ctx.namespace_commit_generation(), 2);
for (idx, dir) in dirs.iter().enumerate() {
let reopened = reopen_local_disk(dir).await;
+35 -19
View File
@@ -4051,6 +4051,7 @@ mod tests {
let _ = drain_global_dirty_scopes();
let rename_barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME);
let rename_tasks = rename_fanout_barrier::observe_tasks(object);
let complete_store = Arc::clone(&set_disks);
let mut complete = tokio::spawn(async move {
let mut opts = ObjectOptions::default();
@@ -4062,16 +4063,6 @@ mod tests {
tokio::time::timeout(Duration::from_secs(30), rename_barrier.wait_until_paused())
.await
.expect("multipart completion should pause one tail disk during rename");
assert!(
tokio::time::timeout(Duration::from_millis(100), &mut complete).await.is_err(),
"multipart completion must not publish success while a tail rename is still paused"
);
let initial = drain_global_dirty_scopes().into_iter().collect::<HashSet<_>>();
assert!(
initial.is_empty(),
"capacity must not be marked as committed before the full multipart rename finishes"
);
let abort_store = Arc::clone(&set_disks);
let abort = tokio::spawn(async move {
@@ -4080,21 +4071,46 @@ mod tests {
.await
});
signaling.wait_for_attempts(2).await;
assert!(!abort.is_finished(), "the in-flight completion must retain the multipart upload guard");
let retained_staging = futures::future::join_all(
disk_stores
.iter()
.map(|disk| disk.read_all(RUSTFS_META_MULTIPART_BUCKET, &staged_part)),
)
// A paused rename does not establish that the other disks reached quorum.
let retained_staging = tokio::time::timeout(Duration::from_secs(30), async {
loop {
let mut retained = 0;
for result in futures::future::join_all(
disk_stores
.iter()
.map(|disk| disk.read_all(RUSTFS_META_MULTIPART_BUCKET, &staged_part)),
)
.await
{
match result {
Ok(_) => retained += 1,
Err(DiskError::FileNotFound) => {}
Err(error) => panic!("staged rename source lookup failed: {error}"),
}
}
if retained <= 1 && rename_tasks.running() == 1 {
break retained;
}
tokio::time::sleep(Duration::from_millis(10)).await;
}
})
.await
.into_iter()
.filter(|result| result.is_ok())
.count();
.expect("unpaused multipart renames should finish before the tail is released");
assert_eq!(
retained_staging, 1,
"only the paused tail disk should still retain the multipart rename source"
);
assert!(
tokio::time::timeout(Duration::from_millis(100), &mut complete).await.is_err(),
"multipart completion must not publish success while a tail rename is still paused"
);
let initial = drain_global_dirty_scopes().into_iter().collect::<HashSet<_>>();
assert!(
initial.is_empty(),
"capacity must not be marked as committed before the full multipart rename finishes"
);
assert!(!abort.is_finished(), "the in-flight completion must retain the multipart upload guard");
signaling.set_target(rustfs_lock::ObjectKey::new(bucket, object));
let object_attempt = signaling.attempts.load(Ordering::Acquire) + 1;
+4 -1
View File
@@ -4459,7 +4459,10 @@ impl SetDisks {
commit_scanner_publication_lease_tokens.as_ref(),
)
.with_publication_scope(commit_scanner_publication_scope.clone())
.with_rollback_receipt(commit_rollback_receipt.clone()),
.with_rollback_receipt(commit_rollback_receipt.clone())
.with_namespace_commit_guard(
(!is_meta_bucketname(&commit_bucket)).then(|| commit_set.ctx.begin_namespace_commit()),
),
)
.await;
if let Some(scope) = commit_scanner_publication_scope.as_ref() {
+12 -2
View File
@@ -1059,6 +1059,7 @@ mod tests {
use crate::storage_api_contracts::{
bucket::{BucketOperations as _, BucketOptions, DeleteBucketOptions, MakeBucketOptions, SRBucketDeleteOp},
list::ListOperations as _,
namespace::NamespaceLocking as _,
object::{ObjectIO as _, ObjectOperations as _},
};
use crate::store::{ECStore, init_local_disks_with_instance_ctx};
@@ -1486,10 +1487,19 @@ mod tests {
.put_object(bucket, object, &mut reader, &ObjectOptions::default())
.await
.expect("object should be written");
let lock = ecstore.pools[0].disk_set[0]
.new_ns_lock(bucket, object)
.await
.expect("fixture namespace lock should be created");
drop(
lock.get_write_lock(Duration::from_secs(30))
.await
.expect("fixture rename tail should finish before checking its generation"),
);
assert_eq!(
ecstore.scanner_namespace_mutation_generation(),
generation_before_put.saturating_add(1),
"successful object creation should advance scanner namespace activity"
generation_before_put.saturating_add(3),
"successful object creation must observe the logical mutation and both fanout boundaries"
);
ecstore
.get_object_info(bucket, object, &ObjectOptions::default())
+517 -25
View File
@@ -787,6 +787,12 @@ impl ECStore {
pub fn single_pool(&self) -> bool {
self.pools.len() == 1
}
/// The set-local create-only check is atomic only when every object
/// mutation uses that same, enabled namespace lock domain.
pub fn supports_atomic_create_only_write_back(&self) -> bool {
!self.ctx.lock_manager().is_disabled() && self.pools.len() == 1 && self.pools[0].disk_set.len() == 1
}
}
#[cfg(test)]
@@ -2127,7 +2133,7 @@ mod tests {
.iter()
.map(|&drives_per_set| (1, drives_per_set))
.collect::<Vec<_>>();
build_isolated_test_store_with_layout(temp_dir, cmd_line, &pool_layouts, shutdown).await
build_isolated_test_store_with_layout(temp_dir, cmd_line, &pool_layouts, shutdown, None).await
}
async fn build_isolated_test_store_with_layout(
@@ -2135,6 +2141,7 @@ mod tests {
cmd_line: &str,
pool_layouts: &[(usize, usize)],
shutdown: CancellationToken,
instance_ctx: Option<Arc<crate::runtime::instance::InstanceContext>>,
) -> (
Arc<crate::runtime::instance::InstanceContext>,
Arc<crate::store::ECStore>,
@@ -2167,7 +2174,7 @@ mod tests {
let endpoint_pools = EndpointServerPools(pools);
crate::services::notification_sys::install_cross_pool_fence_fleet_proof_for_test();
let instance_ctx = Arc::new(crate::runtime::instance::InstanceContext::new());
let instance_ctx = instance_ctx.unwrap_or_else(|| Arc::new(crate::runtime::instance::InstanceContext::new()));
crate::store::init_local_disks_with_instance_ctx(&instance_ctx, endpoint_pools.clone())
.await
.expect("register local disks into the fresh context");
@@ -2535,6 +2542,348 @@ mod tests {
shutdown.cancel();
}
#[cfg(feature = "test-util")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial_test::serial(storage_class_env)]
async fn early_ack_put_tails_block_scanner_publication_until_all_renames_finish() {
use crate::storage_api_contracts::namespace::NamespaceLocking as _;
let temp_dir = tempfile::tempdir().expect("create scanner PUT tail store dir");
let (ctx, store, shutdown) =
without_storage_class_env(build_isolated_test_store(temp_dir.path(), "scanner-put-tails", &[4])).await;
crate::bucket::metadata_sys::init_bucket_metadata_sys(Arc::clone(&store), Vec::new()).await;
let bucket = format!("scanner-put-tails-{}", Uuid::new_v4());
store
.make_bucket(&bucket, &MakeBucketOptions::default())
.await
.expect("create scanner PUT tail bucket");
let set = &store.pools[0].disk_set[0];
let objects = [("scanner-tail-a", vec![0xA1; 273]), ("scanner-tail-b", vec![0xB2; 379])];
temp_env::async_with_vars([(crate::set_disk::ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("true"))], async {
let (active, blocked, movement_generation) = store.scanner_data_movement_activity().await;
assert!(!active && !blocked);
assert!(ctx.scanner_publication_state_allowed(), "the set admission cache should start allowed");
let (old_lease, _) = store
.acquire_scanner_publication_lease(movement_generation, crate::runtime::instance::SCANNER_PUBLICATION_LEASE_TTL)
.await
.expect("publication lease should be admitted before either PUT starts");
let barriers: Vec<_> = objects
.iter()
.map(|(object, _)| {
crate::set_disk::rename_fanout_barrier::arm(object, 0, crate::set_disk::rename_fanout_barrier::PHASE_RENAME)
})
.collect();
let trackers: Vec<_> = objects
.iter()
.map(|(object, _)| crate::set_disk::rename_fanout_barrier::observe_tasks(object))
.collect();
let puts: Vec<_> = objects
.iter()
.map(|(object, body)| {
let put_store = Arc::clone(&store);
let put_bucket = bucket.clone();
let object = *object;
let body = body.clone();
tokio::spawn(async move {
let mut reader = PutObjReader::from_vec(body);
put_store
.put_object(&put_bucket, object, &mut reader, &ObjectOptions::default())
.await
})
})
.collect();
let committed = tokio::time::timeout(Duration::from_secs(30), async {
for barrier in &barriers {
barrier.wait_until_paused().await;
}
let mut committed = Vec::with_capacity(puts.len());
for put in puts {
committed.push(
put.await
.expect("early-ACK PUT task should join while its tail is paused")
.expect("root PUT should return after quorum without waiting for its tail"),
);
}
committed
})
.await
.expect("both root PUTs must quorum-ACK while their tail disks remain paused");
assert!(trackers.iter().all(|tracker| tracker.running() >= 1));
assert!(ctx.namespace_commits_pending());
assert!(
ctx.scanner_publication_state_allowed(),
"pending PUT tails must not disable scanner namespace walks"
);
let (active, blocked, observed_movement_generation) = store.scanner_data_movement_activity().await;
assert!(!active, "ordinary PUT tails are not decommission or rebalance work");
assert!(!blocked, "ordinary PUT tails must not block the movement-only scan baseline");
assert_eq!(observed_movement_generation, movement_generation);
assert!(store.scanner_data_usage_publication_blocked().await);
assert!(store.scanner_data_usage_publication_admission_guard().await.is_some());
assert!(set.scanner_data_usage_publication_admission_guard().await.is_some());
for error in [
store
.acquire_scanner_publication_lease(
movement_generation,
crate::runtime::instance::SCANNER_PUBLICATION_LEASE_TTL,
)
.await
.expect_err("a new remote publication lease must reject pending PUT tails"),
store
.validate_scanner_publication_lease(old_lease, movement_generation)
.await
.expect_err("an existing remote lease must not bypass pending PUT tails"),
store
.acquire_scanner_publication_lease_guard(old_lease)
.await
.expect_err("target-side publication admission must reject pending PUT tails"),
] {
assert!(
error.to_string().contains("blocked"),
"publication must fail because of active tails: {error}"
);
}
store.release_scanner_publication_lease(old_lease).await;
for (index, barrier) in barriers.iter().enumerate() {
let commit_generation = ctx.namespace_commit_generation();
let namespace_generation = store.scanner_namespace_mutation_generation();
barrier.release();
tokio::time::timeout(Duration::from_secs(30), async {
while trackers[index].running() != 0 || ctx.namespace_commit_generation() <= commit_generation {
tokio::task::yield_now().await;
}
if index + 1 == barriers.len() {
while ctx.namespace_commits_pending() {
tokio::task::yield_now().await;
}
}
})
.await
.expect("released tail must drain and publish its terminal namespace generation");
assert!(store.scanner_namespace_mutation_generation() > namespace_generation);
let pending = index + 1 < barriers.len();
assert_eq!(ctx.namespace_commits_pending(), pending);
assert_eq!(store.scanner_data_usage_publication_blocked().await, pending);
assert!(!store.scanner_data_movement_activity().await.1);
assert!(store.scanner_data_usage_publication_admission_guard().await.is_some());
assert!(set.scanner_data_usage_publication_admission_guard().await.is_some());
}
let (lease, generation) = store
.acquire_scanner_publication_lease(movement_generation, crate::runtime::instance::SCANNER_PUBLICATION_LEASE_TTL)
.await
.expect("remote publication lease should resume after both tails drain");
store
.validate_scanner_publication_lease(lease, generation)
.await
.expect("a resumed remote publication lease should validate");
drop(
store
.acquire_scanner_publication_lease_guard(lease)
.await
.expect("target-side publication admission should resume after both tails drain"),
);
assert!(store.release_scanner_publication_lease(lease).await);
let disks = set.disk_inventory().await;
assert_eq!(disks.len(), 4);
for ((object, body), committed) in objects.iter().zip(&committed) {
let logical_size = i64::try_from(body.len()).expect("fixture payload size should fit i64");
let etag = committed.etag.as_ref().expect("root PUT should return a committed ETag");
for (disk_index, disk) in disks.iter().enumerate() {
let file_info = disk
.as_ref()
.expect("every fixture disk should remain online")
.read_version(
"",
&bucket,
object,
"",
&crate::disk::ReadOptions {
read_data: true,
..Default::default()
},
)
.await
.unwrap_or_else(|err| panic!("disk {disk_index} should publish {object} after its tail finishes: {err}"));
assert_eq!(file_info.size, logical_size);
assert_eq!(file_info.metadata.get(http::header::ETAG.as_str()), Some(etag));
assert!(
file_info.inline_data(),
"small fixture payloads should have an inline shard on every disk"
);
let inline_data = file_info.data.as_ref().expect("every disk should retain its inline shard");
let erasure = crate::erasure::coding::Erasure::try_new_with_options(
file_info.erasure.data_blocks,
file_info.erasure.parity_blocks,
file_info.erasure.block_size,
file_info.uses_legacy_checksum,
)
.expect("persisted erasure geometry should be valid");
let shard_size =
usize::try_from(erasure.shard_file_size(logical_size)).expect("fixture shard size should fit usize");
crate::erasure::coding::bitrot_verify(
Cursor::new(inline_data.clone()),
inline_data.len(),
shard_size,
rustfs_utils::HashAlgorithm::HighwayHash256S,
erasure.shard_size(),
)
.await
.unwrap_or_else(|err| panic!("disk {disk_index} should retain a complete valid shard for {object}: {err}"));
}
let mut reader = store
.get_object_reader(&bucket, object, None, HeaderMap::new(), &ObjectOptions::default())
.await
.expect("fully drained PUT should be readable");
let mut actual = Vec::new();
reader.stream.read_to_end(&mut actual).await.expect("PUT body should drain");
assert_eq!(&actual, body);
}
let generation_before_internal_put = ctx.namespace_commit_generation();
let internal_object = "scanner-tail-regression/internal-metadata";
let internal_body = b"scanner metadata must not invalidate its own publication";
let mut internal_reader = PutObjReader::from_vec(internal_body.to_vec());
store
.put_object(RUSTFS_META_BUCKET, internal_object, &mut internal_reader, &ObjectOptions::default())
.await
.expect("internal metadata PUT should commit without scanner self-invalidation");
let internal_lock = set
.new_ns_lock(RUSTFS_META_BUCKET, internal_object)
.await
.expect("internal metadata tail lock should be available");
drop(
internal_lock
.get_write_lock(Duration::from_secs(30))
.await
.expect("internal metadata tail should drain"),
);
assert_eq!(ctx.namespace_commit_generation(), generation_before_internal_put);
assert!(!ctx.namespace_commits_pending());
assert!(store.scanner_data_usage_publication_admission_guard().await.is_some());
assert!(set.scanner_data_usage_publication_admission_guard().await.is_some());
let mut internal_reader = store
.get_object_reader(RUSTFS_META_BUCKET, internal_object, None, HeaderMap::new(), &ObjectOptions::default())
.await
.expect("internal metadata should remain readable");
let mut actual = Vec::new();
internal_reader
.stream
.read_to_end(&mut actual)
.await
.expect("internal metadata body should drain");
assert_eq!(actual, internal_body);
})
.await;
shutdown.cancel();
}
#[cfg(feature = "test-util")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial_test::serial(storage_class_env)]
async fn cancelled_early_ack_put_keeps_scanner_publication_blocked_until_tail_finishes() {
let temp_dir = tempfile::tempdir().expect("create cancelled scanner PUT tail store dir");
let (ctx, store, shutdown) =
without_storage_class_env(build_isolated_test_store(temp_dir.path(), "scanner-cancelled-put-tail", &[4])).await;
crate::bucket::metadata_sys::init_bucket_metadata_sys(Arc::clone(&store), Vec::new()).await;
let bucket = format!("scanner-cancelled-put-tail-{}", Uuid::new_v4());
let object = "scanner-cancelled-tail";
let body = vec![0xC3; 273];
store
.make_bucket(&bucket, &MakeBucketOptions::default())
.await
.expect("create cancelled scanner PUT tail bucket");
temp_env::async_with_vars([(crate::set_disk::ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("true"))], async {
let tracker = crate::set_disk::rename_fanout_barrier::observe_tasks(object);
let tail =
crate::set_disk::rename_fanout_barrier::arm(object, 0, crate::set_disk::rename_fanout_barrier::PHASE_RENAME);
let quorum = crate::set_disk::PutObjectCommitBarrier::install(
&bucket,
object,
crate::set_disk::PutObjectCommitPause::AfterRenameQuorum,
);
let handoff = crate::set_disk::PutObjectCommitBarrier::install(
&bucket,
object,
crate::set_disk::PutObjectCommitPause::AfterRenameHandoff,
);
let put_store = Arc::clone(&store);
let put_bucket = bucket.clone();
let put_body = body.clone();
let put = tokio::spawn(async move {
let mut reader = PutObjReader::from_vec(put_body);
put_store
.put_object(&put_bucket, object, &mut reader, &ObjectOptions::default())
.await
});
tokio::time::timeout(Duration::from_secs(30), tail.wait_until_paused())
.await
.expect("cancelled PUT should pause one disk before rename");
quorum.wait_until_paused().await;
put.abort();
assert!(
put.await
.expect_err("caller should be cancelled after rename quorum")
.is_cancelled()
);
quorum.release();
handoff.wait_until_paused().await;
assert!(tracker.running() >= 1);
assert!(ctx.namespace_commits_pending());
assert!(!store.scanner_data_movement_activity().await.1);
assert!(store.scanner_data_usage_publication_blocked().await);
assert!(store.scanner_data_usage_publication_admission_guard().await.is_some());
assert!(
store.pools[0].disk_set[0]
.scanner_data_usage_publication_admission_guard()
.await
.is_some()
);
let generation = store.scanner_namespace_mutation_generation();
handoff.release();
tail.release();
tokio::time::timeout(Duration::from_secs(30), async {
while tracker.running() != 0 || ctx.namespace_commits_pending() {
tokio::task::yield_now().await;
}
})
.await
.expect("cancelled request's detached fanout must release scanner admission after finishing");
assert!(store.scanner_namespace_mutation_generation() > generation);
assert!(!store.scanner_data_usage_publication_blocked().await);
assert!(store.scanner_data_usage_publication_admission_guard().await.is_some());
for (disk_index, disk) in store.pools[0].disk_set[0].disk_inventory().await.iter().enumerate() {
let file_info = disk
.as_ref()
.expect("cancelled PUT fixture disk should remain online")
.read_version("", &bucket, object, "", &crate::disk::ReadOptions::default())
.await
.unwrap_or_else(|err| panic!("cancelled PUT must still publish on disk {disk_index}: {err}"));
assert_eq!(file_info.size, i64::try_from(body.len()).expect("fixture body size should fit i64"));
}
let mut reader = store
.get_object_reader(&bucket, object, None, HeaderMap::new(), &ObjectOptions::default())
.await
.expect("a cancelled caller must not discard its quorum-committed object");
let mut actual = Vec::new();
reader
.stream
.read_to_end(&mut actual)
.await
.expect("cancelled PUT body should drain");
assert_eq!(actual, body);
})
.await;
shutdown.cancel();
}
#[cfg(feature = "test-util")]
#[test]
#[serial_test::serial(storage_class_env)]
@@ -2986,8 +3335,9 @@ mod tests {
) -> crate::core::pools::DecommissionTestFaultDecision {
let target_bucket = bucket.to_string();
let target_object = object.to_string();
Arc::new(move |stage, bucket, object, _attempt, succeeded| {
Arc::new(move |stage, bucket, object, attempt, succeeded| {
if !succeeded
|| attempt >= crate::core::pools::DECOMMISSION_VERSION_COPY_ATTEMPTS
|| stage != DECOMMISSION_TEST_FAULT_STAGE_MIGRATE_OBJECT
|| bucket != target_bucket
|| object != target_object
@@ -2997,6 +3347,7 @@ mod tests {
// Entry retries reset the local attempt; real copy errors can skip
// successful attempts. Only injected faults spend this global budget.
// A real failure may consume an attempt, so preserve the final chance.
faults
.fetch_update(Ordering::SeqCst, Ordering::SeqCst, |faults| {
(faults < crate::core::pools::DECOMMISSION_VERSION_COPY_ATTEMPTS.saturating_sub(1))
@@ -5018,6 +5369,7 @@ mod tests {
"decommission-delete-fence",
&[(2, 4), (1, 4)],
CancellationToken::new(),
None,
))
.await;
crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
@@ -5149,7 +5501,15 @@ mod tests {
#[test]
fn decommission_retry_fault_budget_counts_successes_across_attempt_changes() {
for attempts in [[1, 2, 3], [1, 1, 2], [1, 3, 3]] {
let cases: &[&[(usize, bool, bool)]] = &[
&[(1, true, true), (2, true, true), (3, true, false)],
&[(1, true, true), (1, true, true), (2, true, false)],
&[(1, true, true), (3, true, false), (3, true, false)],
&[(1, true, true), (2, false, false), (1, true, true), (2, true, false)],
&[(1, true, true), (2, false, false), (3, true, false)],
&[(3, true, false), (4, true, false)],
];
for case in cases {
let faults = Arc::new(AtomicUsize::new(0));
let hook = decommission_retry_fault_hook("bucket", "object", Arc::clone(&faults));
@@ -5163,14 +5523,16 @@ mod tests {
}
assert_eq!(faults.load(Ordering::SeqCst), 0, "unrelated or failed copies must not consume faults");
for (index, attempt) in attempts.into_iter().enumerate() {
let mut expected_faults = 0;
for &(attempt, succeeded, expected) in *case {
assert_eq!(
hook(DECOMMISSION_TEST_FAULT_STAGE_MIGRATE_OBJECT, "bucket", "object", attempt, true),
index < 2,
"attempts={attempts:?}, index={index}"
hook(DECOMMISSION_TEST_FAULT_STAGE_MIGRATE_OBJECT, "bucket", "object", attempt, succeeded),
expected,
"fault plan {case:?} at attempt {attempt}"
);
expected_faults += usize::from(expected);
assert_eq!(faults.load(Ordering::SeqCst), expected_faults);
}
assert_eq!(faults.load(Ordering::SeqCst), 2, "attempts={attempts:?}");
}
}
@@ -5306,6 +5668,15 @@ mod tests {
changed_result.expect("SourceChanged entry retry must converge");
other_result.expect("other bucket entry must continue through ordinary copy retries");
assert_eq!(
store.pool_meta.read().await.pools[0]
.decommission
.as_ref()
.expect("decommission progress should remain available")
.items_decommission_failed,
0,
"entry completion must not hide an exhausted copy failure"
);
assert!(!rx.is_cancelled(), "entry-level SourceChanged must not cancel the shared worker token");
assert_eq!(mutation_calls.load(Ordering::SeqCst), 2, "entry must be re-listed after SourceChanged");
assert_eq!(ordinary_faults.load(Ordering::SeqCst), 2, "ordinary copy must consume the retry budget");
@@ -5934,6 +6305,7 @@ mod tests {
"reverse-decommission-fixed-target",
&[(1, 4), (1, 4)],
CancellationToken::new(),
None,
))
.await;
crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
@@ -6355,6 +6727,7 @@ mod tests {
"multi-set-decommission-source-cleanup",
&[(2, 4)],
CancellationToken::new(),
None,
))
.await;
crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
@@ -8870,18 +9243,17 @@ mod tests {
const MANIFEST_COUNT: usize = 10;
let temp_dir = tempfile::tempdir().expect("create fast manifest pass recovery store dir");
let (ctx, store, _shutdown) =
without_storage_class_env(build_isolated_test_store(temp_dir.path(), "tier-delete-fast-manifest-pass", &[4])).await;
let mut instance_ctx = crate::runtime::instance::InstanceContext::new();
instance_ctx.suppress_tier_delete_journal_recovery_for_test();
let (ctx, store, shutdown) = without_storage_class_env(build_isolated_test_store_with_layout(
temp_dir.path(),
"tier-delete-fast-manifest-pass",
&[(1, 4)],
CancellationToken::new(),
Some(Arc::new(instance_ctx)),
))
.await;
crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
let bucket = "tier-delete-fast-manifest-pass-bucket";
store
.make_bucket(bucket, &MakeBucketOptions::default())
.await
.expect("fast manifest pass bucket should be created");
let incarnation = store
.bucket_incarnation_id(bucket)
.await
.expect("fast manifest pass bucket incarnation should resolve");
let tier_name = "FAST-MANIFEST-PASS";
let backend = register_mock_tier(&ctx.tier_config_mgr(), tier_name).await;
let backend_identity = TierConfigMgr::acquire_operation_lease(&ctx.tier_config_mgr(), tier_name)
@@ -8889,9 +9261,19 @@ mod tests {
.expect("fast manifest pass tier lease should resolve")
.backend_identity();
for index in 0..MANIFEST_COUNT {
// Pagination must not depend on same-bucket lock wait deadlines.
let bucket = format!("tier-delete-fast-manifest-pass-{index}");
store
.make_bucket(&bucket, &MakeBucketOptions::default())
.await
.expect("fast manifest pass bucket should be created");
let incarnation = store
.bucket_incarnation_id(&bucket)
.await
.expect("fast manifest pass bucket incarnation should resolve");
install_aborting_dispatch_fixture(
store.clone(),
bucket,
&bucket,
incarnation,
&format!("manifest-page-{index:06}/"),
tier_name,
@@ -8922,12 +9304,78 @@ mod tests {
"one production pass must cross the default eight-manifest page limit"
);
assert_eq!(stats.manifests.scanned, MANIFEST_COUNT);
assert_eq!(stats.manifests.deleted, MANIFEST_COUNT);
assert_eq!(stats.manifests.failed, 0);
assert_eq!(stats.manifests.deleted, MANIFEST_COUNT, "full recovery result: {stats:?}");
assert_eq!(stats.manifests.failed, 0, "full recovery result: {stats:?}");
assert_eq!(manifest_marker, None);
assert_eq!(tier_delete_dispatch_manifest_count(store.clone()).await, 0);
assert_eq!(tier_delete_journal_count(store).await, 0);
assert_eq!(backend.remove_count().await, 0, "rollback recovery must not call the remote tier");
shutdown.cancel();
}
#[cfg(feature = "test-util")]
#[tokio::test]
#[serial_test::serial(storage_class_env)]
async fn tier_delete_manual_pass_retains_manifest_owned_by_startup_recovery() {
let temp_dir = tempfile::tempdir().expect("create automatic recovery ownership store dir");
let (ctx, store, shutdown) =
without_storage_class_env(build_isolated_test_store(temp_dir.path(), "tier-delete-auto-owner", &[4])).await;
crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
let bucket = "tier-delete-auto-owner-bucket";
store
.make_bucket(bucket, &MakeBucketOptions::default())
.await
.expect("automatic recovery bucket should be created");
let incarnation = store.bucket_incarnation_id(bucket).await.expect("bucket incarnation");
let tier_name = "AUTO-OWNER";
let backend = register_mock_tier(&ctx.tier_config_mgr(), tier_name).await;
let identity = TierConfigMgr::acquire_operation_lease(&ctx.tier_config_mgr(), tier_name)
.await
.expect("automatic recovery tier lease")
.backend_identity();
// The automatic worker must not observe a partially installed fixture.
let lifecycle_guard = store
.acquire_bucket_lifecycle_write_lock(bucket)
.await
.expect("fixture lifecycle lock");
let (manifest_name, entries) =
install_aborting_dispatch_fixture(store.clone(), bucket, incarnation, "auto-owner/", tier_name, identity, 1).await;
let journal_name = tier_delete_journal_object_name(&entries[0]);
let hook = TierDeleteDispatchRollbackTestHook::install_slow_delete(&journal_name, &journal_name);
drop(lifecycle_guard);
ctx.wake_tier_delete_journal_recovery();
tokio::time::timeout(Duration::from_secs(30), hook.wait_until_delete_paused())
.await
.expect("startup recovery should own the manifest before a manual pass");
assert!(tier_delete_dispatch_manifest_recovery_inflight_for_test(&store, &manifest_name));
let stats = recover_tier_delete_dispatch_manifests(store.clone(), 8, None)
.await
.expect("manual recovery scan");
assert_eq!(stats.scanned, 1, "{stats:?}");
assert_eq!(stats.retained, 1, "{stats:?}");
assert_eq!(stats.deleted, 0, "{stats:?}");
assert_eq!(stats.failed, 0, "{stats:?}");
assert_eq!(tier_delete_dispatch_manifest_count(store.clone()).await, 1);
assert_eq!(tier_delete_journal_count(store.clone()).await, 1);
hook.release_delete();
tokio::time::timeout(Duration::from_secs(30), async {
loop {
let manifest_gone = matches!(com::read_config(store.clone(), &manifest_name).await, Err(Error::ConfigNotFound));
if manifest_gone && !tier_delete_dispatch_manifest_recovery_inflight_for_test(&store, &manifest_name) {
break;
}
tokio::time::sleep(Duration::from_millis(10)).await;
}
})
.await
.expect("automatic recovery should converge without a manual retry");
assert_eq!(tier_delete_dispatch_manifest_count(store.clone()).await, 0);
assert_eq!(tier_delete_journal_count(store).await, 0);
assert_eq!(backend.remove_count().await, 0, "rollback must not delete from the remote tier");
shutdown.cancel();
}
#[cfg(feature = "test-util")]
@@ -10302,8 +10750,17 @@ mod tests {
const JOURNAL_COUNT: usize = 40;
let temp_dir = tempfile::tempdir().expect("create rollback retry store dir");
let (ctx, store, _shutdown) =
without_storage_class_env(build_isolated_test_store(temp_dir.path(), "dispatch-rollback-retry", &[4])).await;
// Manual retries must own progress between fault removal and the next attempt.
let mut instance_ctx = crate::runtime::instance::InstanceContext::new();
instance_ctx.suppress_tier_delete_journal_recovery_for_test();
let (ctx, store, shutdown) = without_storage_class_env(build_isolated_test_store_with_layout(
temp_dir.path(),
"dispatch-rollback-retry",
&[(1, 4)],
CancellationToken::new(),
Some(Arc::new(instance_ctx)),
))
.await;
crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
let bucket = "dispatch-rollback-retry-bucket";
store
@@ -10377,6 +10834,7 @@ mod tests {
assert_eq!(tier_delete_dispatch_manifest_count(store.clone()).await, 0);
assert_eq!(backend.remove_count().await, 0, "rollback retries must never call the remote tier");
shutdown.cancel();
}
#[cfg(feature = "test-util")]
@@ -13204,6 +13662,7 @@ mod tests {
"partial-set-prefix-delete",
&[(2, 4)],
CancellationToken::new(),
None,
))
.await;
crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
@@ -16576,6 +17035,7 @@ mod tests {
"prepared-directory-recovery",
&[(2, 4)],
shutdown,
None,
))
.await;
crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
@@ -17228,6 +17688,38 @@ mod tests {
.expect("test thread should complete");
}
#[cfg(feature = "test-util")]
#[tokio::test]
#[serial_test::serial(storage_class_env)]
async fn odm_write_back_requires_one_set_and_enabled_namespace_locking() {
for (layout, locking, supported) in [
(&[(1, 4)][..], true, true),
(&[(1, 4), (1, 4)][..], true, false),
(&[(2, 4)][..], true, false),
(&[(1, 4)][..], false, false),
] {
temp_env::async_with_vars([("RUSTFS_LOCK_ENABLED", Some(if locking { "true" } else { "false" }))], async {
let dir = tempfile::tempdir().expect("isolated topology");
let shutdown = CancellationToken::new();
let (_ctx, store, _) = without_storage_class_env(build_isolated_test_store_with_layout(
dir.path(),
"odm-topology",
layout,
shutdown.clone(),
None,
))
.await;
assert_eq!(
store.supports_atomic_create_only_write_back(),
supported,
"layout={layout:?}, locking={locking}"
);
shutdown.cancel();
})
.await;
}
}
#[cfg(feature = "test-util")]
#[tokio::test]
#[serial_test::serial(storage_class_env)]
+8 -7
View File
@@ -848,7 +848,7 @@ impl ECStore {
}
pub fn scanner_namespace_mutation_generation(&self) -> u64 {
list_objects::scanner_namespace_mutation_generation()
list_objects::scanner_namespace_mutation_generation().saturating_add(self.ctx.namespace_commit_generation())
}
pub async fn scanner_data_movement_active(&self) -> bool {
@@ -857,7 +857,7 @@ impl ECStore {
}
/// Return the storage-owned movement state and generation as one
/// authenticated activity snapshot. The read lock is acquired before
/// authenticated activity snapshot. The read lock is acquired before
/// the state locks (cancelers, pool metadata, then rebalance metadata),
/// matching the transition writer order and preventing a terminal state
/// from being reported with the preceding generation.
@@ -886,11 +886,12 @@ impl ECStore {
/// Returns whether scanner metadata may still be hidden by a local
/// data-movement state. Terminal failed/canceled decommission entries
/// remain suspended until an operator clears or retries them, so they are
/// a publication barrier even after the worker has stopped.
/// a publication barrier even after the worker has stopped. Active PUT
/// rename fanouts also defer publication, including post-ACK tails.
pub async fn scanner_data_usage_publication_blocked(&self) -> bool {
let operation_gate = self.ctx.data_movement_operation_gate();
let _operation_guard = operation_gate.read_owned().await;
self.scanner_data_usage_publication_snapshot_blocked().await
self.scanner_data_usage_publication_snapshot_blocked().await || self.ctx.namespace_commits_pending()
}
pub async fn scanner_data_movement_pause_status(&self) -> ScannerDataMovementPauseStatus {
@@ -1070,7 +1071,7 @@ impl ECStore {
{
return Err(Error::other("scanner publication lease generation is stale"));
}
if self.scanner_data_movement_snapshot_locked().await.1 {
if self.scanner_data_movement_snapshot_locked().await.1 || self.ctx.namespace_commits_pending() {
return Err(Error::other("scanner publication lease is blocked by data movement"));
}
@@ -1109,7 +1110,7 @@ impl ECStore {
{
return Err(Error::other("scanner publication lease generation is stale"));
}
if self.scanner_data_movement_snapshot_locked().await.1 {
if self.scanner_data_movement_snapshot_locked().await.1 || self.ctx.namespace_commits_pending() {
return Err(Error::other("scanner publication lease is blocked by data movement"));
}
if !self.ctx.scanner_publication_lease_is_active(token).await {
@@ -1129,7 +1130,7 @@ impl ECStore {
if self.ctx.data_movement_generation_exhausted() || self.ctx.data_movement_operation_epoch_exhausted() {
return Err(Error::other("scanner publication lease generation is exhausted"));
}
if self.scanner_data_movement_snapshot_locked().await.1 {
if self.scanner_data_movement_snapshot_locked().await.1 || self.ctx.namespace_commits_pending() {
return Err(Error::other("scanner publication lease is blocked by data movement"));
}
let Some(lease_generation) = self.ctx.scanner_publication_lease_generation(token).await else {
@@ -44,7 +44,9 @@ use walkdir::WalkDir;
mod storage_api;
use storage_api::integration::{BucketOperations, ECStore, MakeBucketOptions, ObjectIO as _, ObjectOperations as _};
use storage_api::integration::{
BucketOperations, ECStore, MakeBucketOptions, NamespaceLocking as _, ObjectIO as _, ObjectOperations as _,
};
/// 256 KiB + change: large enough to be stored as non-inline erasure shards
/// (so each data version materializes as an on-disk `part.*` file we can assert
@@ -106,6 +108,7 @@ async fn put_versioned(ecstore: &Arc<ECStore>, bucket: &str, object: &str, data:
.put_object(bucket, object, &mut reader, &opts)
.await
.expect("versioned put_object failed");
wait_for_put_tail(ecstore, bucket, object).await;
info.version_id
.map(|u| u.to_string())
.expect("versioned put must return a version id")
@@ -117,6 +120,7 @@ async fn put_unversioned(ecstore: &Arc<ECStore>, bucket: &str, object: &str, dat
.put_object(bucket, object, &mut reader, &ObjectOptions::default())
.await
.expect("unversioned put_object failed");
wait_for_put_tail(ecstore, bucket, object).await;
}
/// Create a delete-marker as the latest version (versioned:true, no version_id)
@@ -160,20 +164,16 @@ fn xl_meta_path(obj_dir: &Path) -> PathBuf {
obj_dir.join("xl.meta")
}
async fn wait_for_two_version_copies(disks: &[PathBuf], bucket: &str, object: &str) {
tokio::time::timeout(Duration::from_secs(5), async {
loop {
if disks.iter().all(|disk| {
let object_dir = object_dir(disk, bucket, object);
xl_meta_path(&object_dir).exists() && count_part_files(&object_dir) >= 2
}) {
break;
}
tokio::time::sleep(Duration::from_millis(10)).await;
}
})
.await
.expect("PUT rename tails must converge before wiping the versioned fixture");
async fn wait_for_put_tail(ecstore: &Arc<ECStore>, bucket: &str, object: &str) {
// Shards and xl.meta can exist before the detached PUT owner finishes.
let lock = ecstore
.new_ns_lock(bucket, object)
.await
.expect("fixture namespace lock should be created");
let _settled = lock
.get_write_lock(Duration::from_secs(30))
.await
.expect("PUT rename tail must finish before inspecting or wiping the fixture");
}
fn recreate_heal_opts() -> HealOpts {
@@ -305,7 +305,13 @@ mod serial_tests {
let data_v2 = versioned_test_data(20);
let v1 = put_versioned(&ecstore, bucket, object, &data_v1).await; // OLD, non-latest
let v2 = put_versioned(&ecstore, bucket, object, &data_v2).await; // latest
wait_for_two_version_copies(&disk_paths, bucket, object).await;
assert!(
disk_paths.iter().all(|disk| {
let dir = object_dir(disk, bucket, object);
xl_meta_path(&dir).exists() && count_part_files(&dir) >= 2
}),
"both versions must exist on every disk before wiping the fixture"
);
// ── Pre-wipe: prove the fixture actually has 2 versions on disk[0] ──
let obj_dir0 = object_dir(&disk_paths[0], bucket, object);
+1
View File
@@ -23,6 +23,7 @@ pub(crate) mod integration {
pub(crate) use rustfs_ecstore::api::storage::ECStore;
pub(crate) use rustfs_storage_api::BucketOperations;
pub(crate) use rustfs_storage_api::MakeBucketOptions;
pub(crate) use rustfs_storage_api::NamespaceLocking;
pub(crate) use rustfs_storage_api::ObjectIO;
pub(crate) use rustfs_storage_api::ObjectOperations;
}
+8 -34
View File
@@ -429,27 +429,6 @@ where
}
}
/// The cached mapping record for one user or group, looked up in the same
/// cache partition `policy_db_set` writes it to (group / STS / regular+service
/// user). `None` when no mapping is stored.
pub async fn get_mapped_policy_record(&self, name: &str, user_type: UserType, is_group: bool) -> Option<MappedPolicy> {
let cache = self.cache.snapshot();
if is_group {
cache.group_policies.get(name).cloned()
} else if user_type == UserType::Sts {
cache.sts_policies.get(name).cloned()
} else {
cache.user_policies.get(name).cloned()
}
}
/// The cached group record (members, status, own timestamp) without the
/// mapped-policy overlay `get_group_description` applies. `None` when the
/// group does not exist.
pub async fn get_group_info(&self, name: &str) -> Option<GroupInfo> {
self.cache.snapshot().groups.get(name).cloned()
}
pub async fn get_policy(&self, name: &str) -> Result<Policy> {
if name.is_empty() {
return Err(Error::InvalidArgument);
@@ -1714,10 +1693,6 @@ where
}
}
// The group's own timestamp moves with every membership or status
// change: site replication judges an incoming group item against it
// (backlog#2291), so it must reflect the last change, not creation.
let now = OffsetDateTime::now_utc();
let gi = match cache.groups.get(group) {
Some(res) => {
let mut gi = res.clone();
@@ -1726,7 +1701,6 @@ where
uniq_set.extend(members.iter().cloned());
gi.members = uniq_set.into_iter().collect();
gi.update_at = Some(now);
gi
}
None => GroupInfo::new(members.clone()),
@@ -1735,7 +1709,8 @@ where
self.api.save_group_info(group, gi.clone()).await?;
self.cache.with_write_lock(|cache| {
let now = self.cache.with_write_lock(|cache| {
let now = OffsetDateTime::now_utc();
cache.add_or_update_group(group, &gi, now);
let user_group_memberships = Arc::clone(&cache.state().user_group_memberships);
@@ -1744,6 +1719,7 @@ where
m.insert(group.to_string());
cache.add_or_update_user_group_membership(member, &m, now);
});
now
});
Ok(now)
@@ -1767,14 +1743,12 @@ where
} else {
gi.status = STATUS_DISABLED.to_owned();
}
let now = OffsetDateTime::now_utc();
gi.update_at = Some(now);
self.api.save_group_info(name, gi.clone()).await?;
self.cache.add_or_update_group(name, &gi, now);
self.cache.add_or_update_group(name, &gi, OffsetDateTime::now_utc());
Ok(now)
Ok(OffsetDateTime::now_utc())
}
pub async fn get_group_description(&self, name: &str) -> Result<GroupDesc> {
@@ -1856,14 +1830,13 @@ where
let s: HashSet<&String> = HashSet::from_iter(gi.members.iter());
let d: HashSet<&String> = HashSet::from_iter(members.iter());
gi.members = s.difference(&d).map(|v| v.to_string()).collect::<Vec<String>>();
let now = OffsetDateTime::now_utc();
gi.update_at = Some(now);
if !update_cache_only {
self.api.save_group_info(name, gi.clone()).await?;
}
self.cache.with_write_lock(|cache| {
let now = self.cache.with_write_lock(|cache| {
let now = OffsetDateTime::now_utc();
cache.add_or_update_group(name, &gi, now);
let user_group_memberships = Arc::clone(&cache.state().user_group_memberships);
@@ -1874,6 +1847,7 @@ where
cache.add_or_update_user_group_membership(member, &m, now);
}
});
now
});
Ok(now)
-16
View File
@@ -1055,22 +1055,6 @@ impl<T: Store> IamSys<T> {
self.store.get_group_description(group).await
}
/// The stored group record itself (see `IamCache::get_group_info`).
pub async fn get_group_info(&self, group: &str) -> Option<GroupInfo> {
self.store.get_group_info(group).await
}
/// The stored policy document, `Error::NoSuchPolicy` when absent.
pub async fn get_policy_doc(&self, name: &str) -> Result<PolicyDoc> {
self.store.get_policy_doc(name).await
}
/// The stored mapping record for one user or group (see
/// `IamCache::get_mapped_policy_record`).
pub async fn get_mapped_policy_record(&self, name: &str, user_type: UserType, is_group: bool) -> Option<MappedPolicy> {
self.store.get_mapped_policy_record(name, user_type, is_group).await
}
pub async fn list_groups_load(&self) -> Result<Vec<String>> {
self.store.update_groups().await
}
+3 -163
View File
@@ -76,21 +76,6 @@ impl ReplicationWorkerOperation for DeletedObjectReplicationInfo {
.delete_object
.delete_marker_mtime
.and_then(|t| i64::try_from(t.unix_timestamp_nanos()).ok()),
// Carry the target-assigned marker version ids (and the fail-closed corrupt
// flag) into the journal so a purge intent replayed after a restart addresses
// the same version the live path did (backlog#2290). Only delete-marker state
// ever records these; other deletes serialize an empty map.
target_delete_marker_version_ids: self
.delete_object
.replication_state
.as_ref()
.map(|state| state.target_delete_marker_version_ids.clone())
.unwrap_or_default(),
target_delete_marker_version_ids_corrupt: self
.delete_object
.replication_state
.as_ref()
.is_some_and(|state| state.target_delete_marker_version_ids_corrupt),
target_arns: self.admitted_target_arns(),
force_delete_id: self.delete_object.force_delete_id,
force_delete_generation: self.delete_object.force_delete_generation,
@@ -253,28 +238,6 @@ pub fn delete_marker_purge_version_id(
})
}
/// The version a delete replication addresses on `arn`, or `None` to refuse.
///
/// A version purge whose purged version is a delete marker must address the
/// marker version the TARGET assigned — the recorded mapping, exactly as the
/// delayed-purge watcher does. The source-side `DELETE ?versionId=<marker>`
/// replicates as such a purge, and a generic S3 target answers a DELETE of an
/// unknown versionId with 204 while keeping its marker, so addressing it by
/// the source id reported success and left the marker behind (backlog#2290,
/// R6.1 on the VMs). Nothing recorded falls back to the source-derived id
/// (id-mirroring peers); a corrupt record refuses, as the watcher does.
pub fn delete_replication_target_version_id(dobj: &DeletedObject, arn: &str) -> Option<Option<String>> {
let is_version_purge = is_version_delete_replication(dobj);
if is_version_purge
&& !dobj.delete_marker
&& let Some(marker) = dobj.delete_marker_version_id
{
return delete_marker_purge_version_id(dobj.replication_state.as_ref(), arn, marker);
}
let source_version = dobj.delete_marker_version_id.or(dobj.version_id).unwrap_or_default();
Some(target_delete_version_id(source_version, is_version_purge))
}
/// Shape an exhausted purge intent as a marker-creation delete entry. Replay
/// reconstructs it with `delete_marker: true`, finds the source marker gone,
/// and funnels into the stale-marker branch of `replicate_delete_with_outcome`
@@ -295,9 +258,9 @@ mod tests {
use super::{
DeletedObjectReplicationInfo, delete_marker_purge_mrf_entry, delete_marker_purge_version_id,
delete_replication_creates_marker, delete_replication_target_version_id, is_object_lock_denied_delete,
is_retryable_delete_replication_head_error, is_version_delete_replication, replicate_delete_outcome,
resync_existing_delete_replication_info, should_retry_delete_marker_purge, target_delete_version_id,
delete_replication_creates_marker, is_object_lock_denied_delete, is_retryable_delete_replication_head_error,
is_version_delete_replication, replicate_delete_outcome, resync_existing_delete_replication_info,
should_retry_delete_marker_purge, target_delete_version_id,
};
use crate::storage_api::DeletedObject;
use crate::{
@@ -632,76 +595,6 @@ mod tests {
assert_eq!(entry.retry_count, 0);
assert_eq!(entry.bucket, "bucket-a");
assert_eq!(entry.object, "doc.txt");
assert!(
entry.target_delete_marker_version_ids.is_empty(),
"no recorded target marker ids means the journal carries none"
);
assert!(!entry.target_delete_marker_version_ids_corrupt);
}
/// backlog#2290: a purge intent journaled to MRF must carry the marker
/// version ids the targets assigned, plus the fail-closed corrupt flag,
/// so a replay after restart addresses the same version the live path did.
#[test]
fn delete_marker_purge_mrf_entry_carries_target_assigned_marker_versions() {
let delete_marker_version_id = Uuid::new_v4();
let mut state = ReplicationState::default();
state
.target_delete_marker_version_ids
.insert("arn:a".to_string(), "remote-marker-a".to_string());
state
.target_delete_marker_version_ids
.insert("arn:b".to_string(), "remote-marker-b".to_string());
let mut dobj = DeletedObjectReplicationInfo {
delete_object: DeletedObject {
object_name: "doc.txt".to_string(),
delete_marker: false,
version_id: Some(Uuid::new_v4()),
delete_marker_version_id: Some(delete_marker_version_id),
replication_state: Some(state),
..Default::default()
},
bucket: "bucket-a".to_string(),
..Default::default()
};
let entry = delete_marker_purge_mrf_entry(&dobj, vec!["arn:a".to_string()]);
assert_eq!(
entry.target_delete_marker_version_ids,
HashMap::from([
("arn:a".to_string(), "remote-marker-a".to_string()),
("arn:b".to_string(), "remote-marker-b".to_string()),
]),
"every recorded target marker id survives the journal, regardless of the retried ARN subset"
);
assert!(!entry.target_delete_marker_version_ids_corrupt);
assert_eq!(
delete_marker_purge_version_id(
Some(&ReplicationState {
target_delete_marker_version_ids: entry.target_delete_marker_version_ids,
..Default::default()
}),
"arn:a",
delete_marker_version_id
),
Some(Some("remote-marker-a".to_string()))
);
// The live path refuses to purge on inconsistent metadata and reports the target
// as failed; the journaled intent must keep refusing after a restart.
dobj.delete_object
.replication_state
.as_mut()
.expect("state was set above")
.target_delete_marker_version_ids_corrupt = true;
let entry = delete_marker_purge_mrf_entry(&dobj, vec!["arn:a".to_string()]);
assert!(entry.target_delete_marker_version_ids_corrupt);
// A delete without replication state journals an empty map.
dobj.delete_object.replication_state = None;
let entry = dobj.to_mrf_entry();
assert!(entry.target_delete_marker_version_ids.is_empty());
assert!(!entry.target_delete_marker_version_ids_corrupt);
}
#[test]
@@ -763,57 +656,4 @@ mod tests {
assert!(!is_object_lock_denied_delete(Some("InternalError"), Some("retention lookup failed")));
assert!(!is_object_lock_denied_delete(None, Some("legal hold")));
}
fn purge_of_marker(marker: Uuid, state: Option<ReplicationState>) -> DeletedObject {
DeletedObject {
object_name: "obj".to_string(),
delete_marker: false,
delete_marker_version_id: Some(marker),
version_id: None,
replication_state: state,
..Default::default()
}
}
#[test]
fn delete_replication_target_version_id_addresses_recorded_marker_for_purges() {
let arn = "arn:minio:replication::generic:photos";
let marker = Uuid::new_v4();
let mut state = ReplicationState::default();
state
.target_delete_marker_version_ids
.insert(arn.to_string(), "remote-marker".to_string());
// purge of a replicated marker: the target's own version
assert_eq!(
delete_replication_target_version_id(&purge_of_marker(marker, Some(state.clone())), arn),
Some(Some("remote-marker".to_string()))
);
// nothing recorded for this arn: the source-derived id (id-mirroring peers)
assert_eq!(
delete_replication_target_version_id(&purge_of_marker(marker, None), arn),
Some(Some(marker.to_string()))
);
// corrupt record: refuse instead of guessing
state.target_delete_marker_version_ids_corrupt = true;
assert_eq!(delete_replication_target_version_id(&purge_of_marker(marker, Some(state)), arn), None);
// marker creation keeps the source id (the target mints its own on a
// versionless DELETE; the id only travels in the source header)
let creation = DeletedObject {
object_name: "obj".to_string(),
delete_marker: true,
delete_marker_version_id: Some(marker),
..Default::default()
};
assert_eq!(delete_replication_target_version_id(&creation, arn), Some(Some(marker.to_string())));
// plain version purge: the source version id
let version = Uuid::new_v4();
let purge = DeletedObject {
object_name: "obj".to_string(),
version_id: Some(version),
..Default::default()
};
assert_eq!(delete_replication_target_version_id(&purge, arn), Some(Some(version.to_string())));
}
}
-20
View File
@@ -641,26 +641,6 @@ pub struct MrfReplicateEntry {
#[serde(rename = "deleteMarkerMtime", skip_serializing_if = "Option::is_none", default)]
pub delete_marker_mtime: Option<i64>,
// For delete-marker purge intents: the exact version id each target assigned to the
// replicated marker, keyed by target ARN. A generic S3 target mints its own version ids
// and answers a DELETE of an unknown id with 204, so a replay that fell back to the source
// marker id would be acknowledged while the real marker stayed behind (backlog#2290).
// Old files lack this key; default=empty means "unknown" and replay keeps the source-id
// fallback it always had.
#[serde(rename = "targetDeleteMarkerVersionIDs", skip_serializing_if = "HashMap::is_empty", default)]
pub target_delete_marker_version_ids: HashMap<String, String>,
// Companion to the map above: the source metadata disagreed about the recorded ids when
// the intent was journaled, so the live path refused to guess and reported the target as
// failed. Replay must keep refusing instead of falling back to the source id. Old files
// lack this key; default=false.
#[serde(
rename = "targetDeleteMarkerVersionIDsCorrupt",
skip_serializing_if = "std::ops::Not::not",
default
)]
pub target_delete_marker_version_ids_corrupt: bool,
#[serde(rename = "targetARNs", skip_serializing_if = "Vec::is_empty", default)]
pub target_arns: Vec<String>,
+3 -3
View File
@@ -41,9 +41,9 @@ pub use config::{
};
pub use delete::{
DeletedObjectReplicationInfo, delete_marker_purge_mrf_entry, delete_marker_purge_version_id,
delete_replication_creates_marker, delete_replication_target_version_id, is_object_lock_denied_delete,
is_retryable_delete_replication_head_error, is_version_delete_replication, replicate_delete_outcome,
resync_existing_delete_replication_info, should_retry_delete_marker_purge, target_delete_version_id,
delete_replication_creates_marker, is_object_lock_denied_delete, is_retryable_delete_replication_head_error,
is_version_delete_replication, replicate_delete_outcome, resync_existing_delete_replication_info,
should_retry_delete_marker_purge, target_delete_version_id,
};
pub use filemeta::{
NULL_VERSION_ID, REPLICATE_EXISTING, REPLICATE_EXISTING_DELETE, REPLICATE_HEAL, REPLICATE_HEAL_DELETE, REPLICATE_INCOMING,
+2 -167
View File
@@ -31,13 +31,8 @@ const CAPABILITY_OPERATION_KIND: u64 = 1 << 0;
const CAPABILITY_TARGET_ARNS: u64 = 1 << 1;
const CAPABILITY_FORCE_DELETE: u64 = 1 << 2;
const CAPABILITY_DELETE_MARKER_MTIME: u64 = 1 << 3;
// Per-ARN target-assigned delete-marker version ids on purge intents (backlog#2290).
const CAPABILITY_TARGET_DELETE_MARKER_VERSION_IDS: u64 = 1 << 4;
const MRF_KNOWN_CAPABILITIES: u64 = CAPABILITY_OPERATION_KIND
| CAPABILITY_TARGET_ARNS
| CAPABILITY_FORCE_DELETE
| CAPABILITY_DELETE_MARKER_MTIME
| CAPABILITY_TARGET_DELETE_MARKER_VERSION_IDS;
const MRF_KNOWN_CAPABILITIES: u64 =
CAPABILITY_OPERATION_KIND | CAPABILITY_TARGET_ARNS | CAPABILITY_FORCE_DELETE | CAPABILITY_DELETE_MARKER_MTIME;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum MrfCapability {
@@ -45,7 +40,6 @@ pub enum MrfCapability {
TargetArns,
ForceDelete,
DeleteMarkerMtime,
TargetDeleteMarkerVersionIds,
}
impl MrfCapability {
@@ -55,7 +49,6 @@ impl MrfCapability {
Self::TargetArns => CAPABILITY_TARGET_ARNS,
Self::ForceDelete => CAPABILITY_FORCE_DELETE,
Self::DeleteMarkerMtime => CAPABILITY_DELETE_MARKER_MTIME,
Self::TargetDeleteMarkerVersionIds => CAPABILITY_TARGET_DELETE_MARKER_VERSION_IDS,
}
}
}
@@ -608,17 +601,9 @@ pub fn decode_mrf_file(data: &[u8]) -> Result<Vec<MrfReplicateEntry>> {
#[cfg(test)]
mod tests {
use super::*;
use std::collections::HashMap;
use uuid::Uuid;
// Capability word 31 = OperationKind | TargetArns | ForceDelete | DeleteMarkerMtime |
// TargetDeleteMarkerVersionIds (backlog#2290).
const ENVELOPE_FIXTURE: &[u8] = &[
b'M', b'R', b'F', b'E', 1, 0, 1, 0, 1, 0, 0, 0, 31, 0, 0, 0, 0, 0, 0, 0, 3, 0, 0, 0, 1, 2, 3,
];
// The envelope a binary from before backlog#2290 writes: same header, capability word 15.
const PRE_TARGET_MARKER_IDS_ENVELOPE_FIXTURE: &[u8] = &[
b'M', b'R', b'F', b'E', 1, 0, 1, 0, 1, 0, 0, 0, 15, 0, 0, 0, 0, 0, 0, 0, 3, 0, 0, 0, 1, 2, 3,
];
@@ -641,8 +626,6 @@ mod tests {
delete_marker_version_id: None,
delete_marker: false,
delete_marker_mtime: None,
target_delete_marker_version_ids: HashMap::new(),
target_delete_marker_version_ids_corrupt: false,
target_arns: vec!["arn:target-a".to_string()],
},
MrfReplicateEntry {
@@ -659,8 +642,6 @@ mod tests {
delete_marker_version_id: None,
delete_marker: false,
delete_marker_mtime: None,
target_delete_marker_version_ids: HashMap::new(),
target_delete_marker_version_ids_corrupt: false,
target_arns: vec!["arn:target-a".to_string(), "arn:target-b".to_string()],
},
MrfReplicateEntry {
@@ -677,11 +658,6 @@ mod tests {
delete_marker_version_id: Some(del_vid),
delete_marker: true,
delete_marker_mtime: Some(1_705_312_200_123_456_789),
target_delete_marker_version_ids: HashMap::from([
("arn:target-a".to_string(), "remote-marker-a".to_string()),
("arn:target-b".to_string(), "remote-marker-b".to_string()),
]),
target_delete_marker_version_ids_corrupt: false,
target_arns: vec!["arn:target-a".to_string()],
},
];
@@ -709,54 +685,6 @@ mod tests {
Some(1_705_312_200_123_456_789),
"delete-marker mtime must survive the MRF disk round-trip"
);
assert!(decoded[0].target_delete_marker_version_ids.is_empty());
assert!(decoded[1].target_delete_marker_version_ids.is_empty());
assert_eq!(
decoded[2].target_delete_marker_version_ids,
HashMap::from([
("arn:target-a".to_string(), "remote-marker-a".to_string()),
("arn:target-b".to_string(), "remote-marker-b".to_string()),
]),
"target-assigned marker version ids must survive the MRF disk round-trip (backlog#2290)"
);
assert!(!decoded[2].target_delete_marker_version_ids_corrupt);
}
/// backlog#2290: the corrupt flag rides the same journal round trip, and an
/// entry that carries neither field encodes exactly as it did before the
/// field existed (both keys are skipped when empty/false).
#[test]
fn mrf_file_round_trips_target_marker_ids_corrupt_flag_and_skips_empty_keys() {
let corrupt = MrfReplicateEntry {
bucket: "bucket-a".to_string(),
object: "delete-a".to_string(),
op: MrfOpKind::Delete,
delete_marker: true,
delete_marker_version_id: Some(Uuid::new_v4()),
target_delete_marker_version_ids_corrupt: true,
target_arns: vec!["arn:target-a".to_string()],
..Default::default()
};
let decoded = decode_mrf_file(&encode_mrf_file(std::slice::from_ref(&corrupt)).expect("mrf file should encode"))
.expect("mrf file should decode");
assert_eq!(decoded, vec![corrupt]);
assert!(decoded[0].target_delete_marker_version_ids_corrupt);
let plain = MrfReplicateEntry {
bucket: "bucket-a".to_string(),
object: "delete-a".to_string(),
op: MrfOpKind::Delete,
delete_marker: true,
target_arns: vec!["arn:target-a".to_string()],
..Default::default()
};
let encoded = encode_mrf_file(std::slice::from_ref(&plain)).expect("mrf file should encode");
let payload = String::from_utf8_lossy(&encoded);
assert!(
!payload.contains("targetDeleteMarkerVersionIDs"),
"an entry without recorded ids must not grow the new keys: {payload}"
);
assert_eq!(decode_mrf_file(&encoded).expect("mrf file should decode"), vec![plain]);
}
#[test]
@@ -791,99 +719,6 @@ mod tests {
// Old files lack the deleteMarkerMtime key; it must default to None so replay keeps the
// pre-#867 fallback to the current time.
assert_eq!(decoded[0].delete_marker_mtime, None);
// Old files also lack the target marker id keys; they must default to an empty map
// and a clear corrupt flag so replay keeps the pre-#2290 source-id fallback.
assert!(decoded[0].target_delete_marker_version_ids.is_empty());
assert!(!decoded[0].target_delete_marker_version_ids_corrupt);
}
/// backlog#2290: a delete-marker entry written by a binary that predates the
/// `targetDeleteMarkerVersionIDs` key decodes with an empty map and a clear
/// corrupt flag — the exact shape replay handled before the field existed.
#[test]
fn mrf_pre_target_marker_ids_delete_entry_decodes_with_empty_map() {
let marker_version_id = Uuid::new_v4();
let mut payload = Vec::new();
rmp::encode::write_array_len(&mut payload, 1).expect("array len should encode");
rmp::encode::write_map_len(&mut payload, 9).expect("map len should encode");
rmp::encode::write_str(&mut payload, "bucket").expect("bucket key should encode");
rmp::encode::write_str(&mut payload, "old-bucket").expect("bucket value should encode");
rmp::encode::write_str(&mut payload, "object").expect("object key should encode");
rmp::encode::write_str(&mut payload, "old-key").expect("object value should encode");
rmp::encode::write_str(&mut payload, "retryCount").expect("retry key should encode");
rmp::encode::write_i32(&mut payload, 0).expect("retry value should encode");
rmp::encode::write_str(&mut payload, "size").expect("size key should encode");
rmp::encode::write_i64(&mut payload, 0).expect("size value should encode");
rmp::encode::write_str(&mut payload, "op").expect("op key should encode");
rmp::encode::write_str(&mut payload, "delete").expect("op value should encode");
rmp::encode::write_str(&mut payload, "forceDelete").expect("forceDelete key should encode");
rmp::encode::write_bool(&mut payload, false).expect("forceDelete value should encode");
rmp::encode::write_str(&mut payload, "deleteMarkerVersionID").expect("marker id key should encode");
// Uuid serializes as a 16-byte bin in the MessagePack journal.
rmp::encode::write_bin(&mut payload, marker_version_id.as_bytes()).expect("marker id value should encode");
rmp::encode::write_str(&mut payload, "deleteMarker").expect("deleteMarker key should encode");
rmp::encode::write_bool(&mut payload, true).expect("deleteMarker value should encode");
rmp::encode::write_str(&mut payload, "targetARNs").expect("targetARNs key should encode");
rmp::encode::write_array_len(&mut payload, 1).expect("targetARNs len should encode");
rmp::encode::write_str(&mut payload, "arn:target-a").expect("targetARNs value should encode");
let mut data = Vec::with_capacity(4 + payload.len());
data.extend_from_slice(&MRF_META_FORMAT.to_le_bytes());
data.extend_from_slice(&MRF_META_VERSION.to_le_bytes());
data.extend_from_slice(&payload);
let decoded = decode_mrf_file(&data).expect("pre-#2290 delete-marker entry should decode");
assert_eq!(decoded.len(), 1);
assert_eq!(decoded[0].op, MrfOpKind::Delete);
assert!(decoded[0].delete_marker);
assert_eq!(decoded[0].delete_marker_version_id, Some(marker_version_id));
assert_eq!(decoded[0].target_arns, vec!["arn:target-a".to_string()]);
assert!(decoded[0].target_delete_marker_version_ids.is_empty());
assert!(!decoded[0].target_delete_marker_version_ids_corrupt);
}
/// backlog#2290: the new field is fenced by its own capability bit exactly
/// like the earlier optional fields — a reader without the bit refuses an
/// envelope that advertises it, while the current reader still accepts the
/// pre-#2290 envelope.
#[test]
fn envelope_target_marker_ids_capability_is_fenced_and_backward_compatible() {
assert!(MrfCapabilities::current().contains(MrfCapability::TargetDeleteMarkerVersionIds));
assert_eq!(MrfCapabilities::with(MrfCapability::TargetDeleteMarkerVersionIds).bits(), 1 << 4);
// Old envelope, current reader: accepted, and the negotiated set lacks the new bit.
let legacy = MrfEnvelope::decode(PRE_TARGET_MARKER_IDS_ENVELOPE_FIXTURE, MrfProtocolCapabilities::current())
.expect("pre-#2290 envelope should decode");
assert_eq!(legacy.protocol().capabilities().bits(), 15);
assert!(
!legacy
.protocol()
.capabilities()
.contains(MrfCapability::TargetDeleteMarkerVersionIds)
);
assert_eq!(legacy.payload(), &[1, 2, 3]);
// Current envelope, reader that only knows the pre-#2290 bits: refused.
let pre_2290_reader = MrfProtocolCapabilities::new(1, 1, MrfCapabilities::from_bits(15).expect("known bits"));
assert_eq!(
MrfEnvelope::decode(ENVELOPE_FIXTURE, pre_2290_reader),
Err(MrfEnvelopeError::MissingCapabilities {
required: 31,
available: 15,
})
);
// Negotiation with such a peer drops the bit instead of failing.
let negotiated = MrfProtocolCapabilities::current()
.negotiate(pre_2290_reader)
.expect("negotiation with a pre-#2290 peer should succeed");
assert!(
!negotiated
.capabilities()
.contains(MrfCapability::TargetDeleteMarkerVersionIds)
);
assert!(negotiated.capabilities().contains(MrfCapability::DeleteMarkerMtime));
}
#[test]
+10 -2
View File
@@ -196,7 +196,7 @@ pub(crate) async fn read_config_revision<S: ScannerObjectIO>(store: Arc<S>, path
}
}
#[derive(Clone, Debug)]
#[derive(Clone, Debug, PartialEq, Eq)]
pub(crate) struct DataUsageCacheRevisions {
main: DataUsageCacheRevision,
backup: Option<DataUsageCacheRevision>,
@@ -503,6 +503,10 @@ pub struct DataUsageCacheInfo {
pub lkg_leader_epoch: Option<u64>,
#[serde(default)]
pub lkg_scan_plan_digest: Option<DataUsageScanPlanDigest>,
/// Activity-sensitive identity for same-cycle set snapshot reuse. The
/// structural plan remains reusable across ordinary bucket writes.
#[serde(default)]
pub scan_execution_digest: Option<DataUsageScanPlanDigest>,
}
impl Serialize for DataUsageCacheInfo {
@@ -519,7 +523,8 @@ impl Serialize for DataUsageCacheInfo {
+ usize::from(self.lkg_next_cycle.is_some())
+ usize::from(self.lkg_last_update.is_some())
+ usize::from(self.lkg_leader_epoch.is_some())
+ usize::from(self.lkg_scan_plan_digest.is_some());
+ usize::from(self.lkg_scan_plan_digest.is_some())
+ usize::from(self.scan_execution_digest.is_some());
let mut state = serializer.serialize_map(Some(field_count))?;
state.serialize_entry("name", &self.name)?;
state.serialize_entry("next_cycle", &self.next_cycle)?;
@@ -558,6 +563,9 @@ impl Serialize for DataUsageCacheInfo {
if let Some(scan_plan_digest) = self.lkg_scan_plan_digest {
state.serialize_entry("lkg_scan_plan_digest", &scan_plan_digest)?;
}
if let Some(scan_execution_digest) = self.scan_execution_digest {
state.serialize_entry("scan_execution_digest", &scan_execution_digest)?;
}
state.end()
}
}
@@ -1067,6 +1067,7 @@ fn test_data_usage_cache_info_deserialize_defaults_scan_resume_after() {
assert!(decoded.source.is_none());
assert!(!decoded.snapshot_complete);
assert!(decoded.scan_plan_digest.is_none());
assert!(decoded.scan_execution_digest.is_none());
assert_eq!(decoded.cache_key_format, 0);
}
@@ -1109,6 +1110,7 @@ fn test_data_usage_cache_info_unmarshal_old_msgpack_defaults_scan_resume_after()
assert!(decoded.source.is_none());
assert!(!decoded.snapshot_complete);
assert!(decoded.scan_plan_digest.is_none());
assert!(decoded.scan_execution_digest.is_none());
assert_eq!(decoded.cache_key_format, 0);
}
@@ -1145,6 +1147,7 @@ fn test_new_data_usage_cache_msgpack_round_trips_and_supports_old_reader() {
source: Some(DataUsageCacheSource::new(1, 2)),
snapshot_complete: true,
scan_plan_digest: Some(TEST_PLAN_DIGEST),
scan_execution_digest: Some(DataUsageScanPlanDigest([42; 32])),
cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT,
..Default::default()
},
@@ -1164,6 +1167,7 @@ fn test_new_data_usage_cache_msgpack_round_trips_and_supports_old_reader() {
assert_eq!(current.info.source, Some(DataUsageCacheSource::new(1, 2)));
assert!(current.info.snapshot_complete);
assert_eq!(current.info.scan_plan_digest, Some(TEST_PLAN_DIGEST));
assert_eq!(current.info.scan_execution_digest, Some(DataUsageScanPlanDigest([42; 32])));
assert_eq!(current.info.cache_key_format, DATA_USAGE_CACHE_KEY_FORMAT);
assert_eq!(current.find("bucket").map(|entry| entry.objects), Some(3));
+31 -5
View File
@@ -1616,7 +1616,7 @@ where
// Refresh the storage-owned movement snapshot before reading background
// heal state. A missing heal object yields an in-memory default; do not
// let that default influence a cycle while publication is blocked.
if storeapi.scanner_data_usage_publication_blocked().await {
if storeapi.scanner_data_movement_pause_status().await.paused {
mark_scan_cycle_idle(cycle_info, &mut cycle_metrics_guard).await;
return ScannerCycleOutcome::Deferred(ScannerCycleDeferReason::DataMovement);
}
@@ -1816,6 +1816,19 @@ where
let publication_defer_reason = publication_defer_reason
.or(remote_lease_defer_reason)
.or(remote_lease_fence_defer_reason);
// A PUT tail can finish between the walk and lease acquisition without
// changing the movement epoch accepted by those leases. Re-prove the
// namespace baseline only after every peer has granted publication.
let post_lease_activity_defer_reason = if publication_defer_reason.is_none()
&& remote_publication_leases.is_some()
&& let Ok(result) = &scan_result
&& result.status == ScannerCycleStatus::Complete
{
scanner_post_lease_activity_defer_reason(result.activity_digest(), probe_scanner_activity(storeapi.as_ref(), true).await)
} else {
None
};
let publication_defer_reason = publication_defer_reason.or(post_lease_activity_defer_reason);
// Include reasons discovered while acquiring or validating remote leases.
let publication_deferred = publication_defer_reason.is_some();
let budget_elapsed = cycle_budget.budget_elapsed() && !ctx.is_cancelled();
@@ -3240,6 +3253,21 @@ where
}
}
fn scanner_post_lease_activity_defer_reason(
expected_digest: Option<[u8; 32]>,
activity: Result<ScannerActivitySnapshot, String>,
) -> Option<ScannerCycleDeferReason> {
match activity {
Ok(snapshot)
if scanner_activity_allows_usage_publication(&snapshot)
&& expected_digest == Some(scanner_activity_snapshot_digest(&snapshot)) =>
{
None
}
Ok(_) | Err(_) => Some(ScannerCycleDeferReason::ActivityBaselineUnavailable),
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum ScannerCyclePreCommitOutcome {
RecoverCacheCycle(u64),
@@ -3428,13 +3456,11 @@ use cycle_state::*;
use leadership::*;
use usage_store::*;
#[cfg(test)]
pub(crate) use activity::scanner_activity_snapshot_digest;
pub use activity::scanner_topology_digest;
pub(crate) use activity::{
ScannerActivitySnapshot, ScannerDirtyUsageAcknowledgement, probe_scanner_activity, scanner_activity_allows_usage_publication,
scanner_activity_dirty_usage_state_for_host, scanner_activity_publication_lease_targets, scanner_activity_structural_digest,
scanner_dirty_usage_acknowledgements,
scanner_activity_dirty_usage_state_for_host, scanner_activity_publication_lease_targets, scanner_activity_snapshot_digest,
scanner_activity_structural_digest, scanner_dirty_usage_acknowledgements,
};
pub(crate) use activity::{ScannerCycleOutcome, scanner_cycle_outcome_with_pending_maintenance};
pub use backlog::{
-1
View File
@@ -902,7 +902,6 @@ where
observation
}
#[cfg(test)]
pub(crate) fn scanner_activity_snapshot_digest(snapshot: &ScannerActivitySnapshot) -> [u8; 32] {
let mut hasher = Sha256::new();
hasher.update(u64::try_from(snapshot.len()).unwrap_or(u64::MAX).to_be_bytes());
+172 -1
View File
@@ -15,7 +15,8 @@
use super::heal_info::{classify_background_heal_read_error, decode_background_heal_info};
use super::*;
use crate::EcstoreResult;
use crate::storage_api::scan::BucketOperations as _;
use crate::storage_api::owner::ecstore_hold_namespace_commit;
use crate::storage_api::scan::{BucketOperations as _, ObjectIO as _};
use crate::{
DATA_USAGE_BLOOM_RECOVERY_PATH, DATA_USAGE_CACHE_KEY_FORMAT, DATA_USAGE_CACHE_NAME, DATA_USAGE_ROOT,
DataUsageCachePrepareOutcome, DataUsageCacheSource, DataUsageEntry, DataUsageScanPlanDigest, Endpoint, EndpointServerPools,
@@ -1165,6 +1166,116 @@ async fn run_data_scanner_cycle_publishes_activity_for_owner_lifetime() {
global_metrics().set_cycle(None).await;
}
#[tokio::test]
#[serial]
async fn coordinator_walks_during_pending_put_without_persisting_or_acknowledging_usage() {
crate::scanner_io::clear_dirty_usage_buckets_for_tests();
let (_temp_dir, store) = setup_scanner_cycle_store().await;
let bucket = format!("scanner-coordinator-pending-{}", Uuid::new_v4().simple());
store
.make_bucket(&bucket, &crate::storage_api::scan::MakeBucketOptions::default())
.await
.expect("fixture bucket should be created");
let mut reader = PutObjReader::from_vec(b"first".to_vec());
store.pools[0].disk_set[0]
.put_object(
&bucket,
"object",
&mut reader,
&ObjectOptions {
no_lock: true,
..Default::default()
},
)
.await
.expect("fixture object should finish its rename fanout");
crate::scanner_io::record_dirty_usage_bucket(&bucket);
let dirty_before = crate::scanner_io::dirty_usage_buckets_for_tests();
let baseline = read_config(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str())
.await
.expect("fixture usage baseline should be readable");
let pending = ecstore_hold_namespace_commit(store.as_ref());
let ctx = CancellationToken::new();
let budget = ScannerCycleBudget::new_with_progress_tracking(&ctx, ScannerCycleBudgetConfig::default());
let mut cycle_info = CurrentCycle {
next: 1,
..Default::default()
};
let mut revision = DataUsageCacheRevision::Missing;
let outcome = tokio::time::timeout(
Duration::from_secs(30),
run_data_scanner_cycle_with_budget(&ctx, &store, &mut cycle_info, &mut revision, 1, Arc::clone(&budget)),
)
.await
.expect("the coordinator must finish its namespace walk while a PUT is pending");
assert_eq!(budget.progress().0, 1, "the coordinator must reach actual object traversal");
assert_eq!(outcome, ScannerCycleOutcome::Deferred(ScannerCycleDeferReason::DataMovement));
assert_eq!(cycle_info.next, 1, "a rejected publication must not advance the cycle");
assert_eq!(revision, DataUsageCacheRevision::Missing);
assert_eq!(crate::scanner_io::dirty_usage_buckets_for_tests(), dirty_before);
assert_eq!(
read_config(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str())
.await
.expect("the prior authoritative usage must remain readable"),
baseline,
"the pending candidate must not replace the authoritative baseline"
);
let committed_body = b"committed-after-walk";
let mut reader = PutObjReader::from_vec(committed_body.to_vec());
store.pools[0].disk_set[0]
.put_object(
&bucket,
"object",
&mut reader,
&ObjectOptions {
no_lock: true,
..Default::default()
},
)
.await
.expect("the pending tail must change the physical object before it drains");
assert_eq!(crate::scanner_io::dirty_usage_buckets_for_tests(), dirty_before);
drop(pending);
let retry_budget = ScannerCycleBudget::new_with_progress_tracking(&ctx, ScannerCycleBudgetConfig::default());
let outcome = tokio::time::timeout(
Duration::from_secs(30),
run_data_scanner_cycle_with_budget(&ctx, &store, &mut cycle_info, &mut revision, 1, Arc::clone(&retry_budget)),
)
.await
.expect("the same cycle must converge after the pending PUT drains");
assert_eq!(
retry_budget.progress().0,
1,
"the same-cycle retry must not reuse the pre-tail bucket cache"
);
assert!(matches!(
outcome,
ScannerCycleOutcome::Completed | ScannerCycleOutcome::CompletedWithPendingMaintenance
));
assert_eq!(cycle_info.next, 2);
assert!(!crate::scanner_io::dirty_usage_buckets_for_tests().contains_key(&bucket));
let usage = read_config(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str())
.await
.expect("the converged usage should be persisted");
let usage: DataUsageInfo = serde_json::from_slice(&usage).expect("the persisted usage should decode");
assert_eq!(usage.usage_snapshot_converged, Some(true));
assert_eq!(usage.scanner_cycle, Some(1));
assert_eq!(usage.objects_total_count, 1);
assert_eq!(
usage.objects_total_size,
u64::try_from(committed_body.len()).expect("fixture body length")
);
let bucket_usage = usage
.buckets_usage
.get(&bucket)
.expect("the scanned bucket should be published");
assert_eq!(bucket_usage.objects_count, 1);
assert_eq!(bucket_usage.size, u64::try_from(committed_body.len()).expect("fixture body length"));
global_metrics().set_cycle(None).await;
crate::scanner_io::clear_dirty_usage_buckets_for_tests();
}
#[tokio::test]
#[serial]
async fn test_finalize_partial_scan_cycle_advances_and_persists_counter() {
@@ -8485,6 +8596,66 @@ fn scanner_node_activity(epoch: &str, namespace_generation: u64, maintenance_gen
}
}
#[test]
fn post_lease_activity_proof_rejects_a_put_tail_that_finished_before_lease_acquisition() {
let before = BTreeMap::from([("node-2".to_string(), scanner_node_activity("epoch-a", 7, 3))]);
let expected_digest = Some(scanner_activity_snapshot_digest(&before));
assert_eq!(scanner_post_lease_activity_defer_reason(expected_digest, Ok(before.clone())), None);
let mut after = before.clone();
after
.get_mut("node-2")
.expect("writer should be present")
.namespace_generation += 1;
assert_eq!(
before["node-2"].movement_generation, after["node-2"].movement_generation,
"the existing movement-only lease remains valid after a PUT tail drains"
);
assert!(scanner_activity_allows_usage_publication(&after));
let reason = scanner_post_lease_activity_defer_reason(expected_digest, Ok(after));
assert_eq!(reason, Some(ScannerCycleDeferReason::ActivityBaselineUnavailable));
let result = ScannerCycleResult::new(ScannerCycleStatus::Complete, None).with_remote_dirty_usage_acknowledgements(vec![
ScannerDirtyUsageAcknowledgement {
host: "node-2".to_string(),
instance_id: "epoch-a".to_string(),
generation: 5,
},
]);
let (outcome, _, acknowledgements) = finalize_scanner_cycle_result(
result,
DataUsagePersistOutcome::Deferred(reason.expect("changed namespace should defer publication")),
);
assert_eq!(
outcome,
ScannerCycleOutcome::Deferred(ScannerCycleDeferReason::ActivityBaselineUnavailable)
);
assert!(
acknowledgements.is_empty(),
"a rejected publication must not acknowledge the peer's dirty usage"
);
}
#[test]
fn post_lease_activity_proof_requires_a_complete_matching_baseline() {
let before = BTreeMap::from([("node-2".to_string(), scanner_node_activity("epoch-a", 7, 3))]);
let digest = scanner_activity_snapshot_digest(&before);
let mut blocked = before.clone();
blocked.get_mut("node-2").expect("peer should be present").publication_blocked = true;
let blocked_digest = scanner_activity_snapshot_digest(&blocked);
for (expected, observed) in [
(None, Ok(before)),
(Some(digest), Err("peer is unavailable".to_string())),
(Some(digest), Ok(BTreeMap::new())),
(Some(blocked_digest), Ok(blocked)),
] {
assert_eq!(
scanner_post_lease_activity_defer_reason(expected, observed),
Some(ScannerCycleDeferReason::ActivityBaselineUnavailable)
);
}
}
#[test]
fn scanner_activity_snapshot_digest_fences_storage_topology() {
let first = BTreeMap::from([("node-2".to_string(), scanner_node_activity("epoch-a", 7, 3))]);
+18 -2
View File
@@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use crate::data_usage_define::DATA_USAGE_CACHE_KEY_FORMAT;
use crate::data_usage_define::{DATA_USAGE_CACHE_KEY_FORMAT, DataUsageCacheRevisions};
use crate::scanner_budget::ScannerCycleBudget;
use crate::scanner_folder::{ScannerItem, scan_data_folder};
use crate::sleeper::SCANNER_SLEEPER;
@@ -271,6 +271,8 @@ pub struct ScannerBucketScanPlan {
all_buckets: Arc<Vec<BucketInfo>>,
scope: ScannerBucketScanScope,
digest: DataUsageScanPlanDigest,
// Cache work must invalidate on namespace completion even when its scoped baseline remains reusable.
execution_digest: DataUsageScanPlanDigest,
leader_epoch: u64,
tier_registry_generation: u64,
/// Epoch captured once for the whole scanner cycle. `None` is retained
@@ -456,9 +458,12 @@ async fn scanner_cycle_activity_status<S>(
where
S: ScannerStorage,
{
// Read the pending-commit barrier before sampling its completion generation.
// A tail that drains during this await must invalidate the earlier baseline.
let publication_blocked = store.scanner_data_usage_publication_blocked().await;
match crate::scanner::probe_scanner_activity(store, distributed).await {
Ok(after) => {
let status = if after == *before {
let status = if !publication_blocked && after == *before {
ScannerCycleActivityStatus::Unchanged
} else {
ScannerCycleActivityStatus::Changed
@@ -760,6 +765,7 @@ fn scanner_activity_preflight(
pub(crate) struct ScannerCycleResult {
pub(crate) status: ScannerCycleStatus,
publication_epoch: Option<u64>,
activity_digest: Option<[u8; 32]>,
observational_snapshot_published: bool,
dirty_usage_clear: Option<DirtyUsageBuckets>,
remote_dirty_usage_acknowledgements: Vec<crate::scanner::ScannerDirtyUsageAcknowledgement>,
@@ -774,6 +780,7 @@ impl ScannerCycleResult {
Self {
status,
publication_epoch: None,
activity_digest: None,
observational_snapshot_published: false,
dirty_usage_clear,
remote_dirty_usage_acknowledgements: Vec::new(),
@@ -793,6 +800,15 @@ impl ScannerCycleResult {
self.publication_epoch
}
fn with_activity_digest(mut self, activity_digest: [u8; 32]) -> Self {
self.activity_digest = Some(activity_digest);
self
}
pub(crate) fn activity_digest(&self) -> Option<[u8; 32]> {
self.activity_digest
}
pub(crate) fn with_observational_snapshot_published(mut self, published: bool) -> Self {
self.observational_snapshot_published = published;
self
+30 -12
View File
@@ -604,10 +604,12 @@ pub(super) async fn persist_and_publish_cache_snapshot(
store: Arc<SetDisks>,
updates: &mpsc::Sender<DataUsageCache>,
mut cache_snapshot: DataUsageCache,
initial_revisions: Option<&DataUsageCacheRevisions>,
cache_cycle_floor: &AtomicU64,
expected_publication_epoch: u64,
) -> Option<SystemTime> {
let source = cache_snapshot.info.source?;
let execution_digest = cache_snapshot.info.scan_execution_digest?;
let guard = match acquire_scanner_cache_locks(store.as_ref(), DATA_USAGE_CACHE_NAME, source).await {
Ok(guard) => guard,
Err(err) => {
@@ -672,20 +674,36 @@ pub(super) async fn persist_and_publish_cache_snapshot(
);
return None;
}
if matches!(
current_cache_root_entry_with_generation(
&persisted,
DATA_USAGE_ROOT,
source,
cache_snapshot.info.next_cycle,
cache_snapshot.info.leader_epoch,
scan_plan_digest,
cache_snapshot.info.tier_registry_generation,
),
Ok(Some(_))
) {
if persisted.info.scan_execution_digest == Some(execution_digest)
&& matches!(
current_cache_root_entry_with_generation(
&persisted,
DATA_USAGE_ROOT,
source,
cache_snapshot.info.next_cycle,
cache_snapshot.info.leader_epoch,
scan_plan_digest,
cache_snapshot.info.tier_registry_generation,
),
Ok(Some(_))
)
{
cache_snapshot = persisted;
} else {
// A later execution may have completed while this scan was walking.
// Only replace the cache revision from which this scan started.
if initial_revisions != Some(&revisions) {
warn!(
target: "rustfs::scanner::io",
event = EVENT_SCANNER_CACHE_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_IO,
state = "scan_baseline_revision_changed",
cache_name = DATA_USAGE_CACHE_NAME,
"Scanner skipped set snapshot without an unchanged baseline revision"
);
return None;
}
if guard.is_lock_lost() {
error!(
target: "rustfs::scanner::io",
+24 -15
View File
@@ -118,6 +118,7 @@ impl ScannerIOCache for SetDisks {
all_buckets,
scope,
digest: scan_plan_digest,
execution_digest,
leader_epoch,
tier_registry_generation,
publication_epoch,
@@ -137,20 +138,24 @@ impl ScannerIOCache for SetDisks {
.ok_or_else(|| StorageError::other("scanner cache publication is blocked by data movement"))?,
};
let mut old_cache = DataUsageCache::default();
if let Err(e) = old_cache.load(self.clone(), DATA_USAGE_CACHE_NAME).await {
warn!(
target: "rustfs::scanner::io",
event = EVENT_SCANNER_CACHE_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_IO,
pool = self.pool_index,
set = self.set_index,
cache_name = DATA_USAGE_CACHE_NAME,
state = "old_cache_load_failed",
error = %e,
"Scanner old data usage cache load failed; rebuilding from bucket caches"
);
}
let initial_revisions = match old_cache.load_with_revisions(self.clone(), DATA_USAGE_CACHE_NAME).await {
Ok(revisions) => Some(revisions),
Err(e) => {
warn!(
target: "rustfs::scanner::io",
event = EVENT_SCANNER_CACHE_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_IO,
pool = self.pool_index,
set = self.set_index,
cache_name = DATA_USAGE_CACHE_NAME,
state = "old_cache_load_failed",
error = %e,
"Scanner old data usage cache load failed; rebuilding from bucket caches"
);
None
}
};
let scoped_scan = prepare_scoped_set_scan(
&old_cache,
&buckets,
@@ -195,6 +200,7 @@ impl ScannerIOCache for SetDisks {
};
cache.info.last_update = Some(now);
cache.info.snapshot_complete = true;
cache.info.scan_execution_digest = Some(execution_digest);
cache.info.lkg_snapshot_complete = false;
cache.info.lkg_next_cycle = None;
cache.info.lkg_last_update = None;
@@ -208,6 +214,7 @@ impl ScannerIOCache for SetDisks {
self,
&updates,
cache,
initial_revisions.as_ref(),
cache_cycle_floor.as_ref(),
expected_publication_epoch,
)
@@ -637,7 +644,7 @@ impl ScannerIOCache for SetDisks {
let cache_name = path_join_buf(&[&bucket.name, DATA_USAGE_CACHE_NAME]);
let bucket_scan_plan_digest =
scanner_bucket_cache_digest(scan_plan_digest, dirty_usage_buckets_clone.get(&bucket.name).copied());
scanner_bucket_cache_digest(execution_digest, dirty_usage_buckets_clone.get(&bucket.name).copied());
if let Some(server_epoch) = remote_server_epoch {
let request_sequence = remote_session_sequence;
@@ -1360,6 +1367,7 @@ impl ScannerIOCache for SetDisks {
cache.info.next_cycle = want_cycle;
cache.info.last_update.get_or_insert_with(SystemTime::now);
cache.info.snapshot_complete = true;
cache.info.scan_execution_digest = Some(execution_digest);
cache.info.lkg_snapshot_complete = false;
cache.info.lkg_next_cycle = None;
cache.info.lkg_last_update = None;
@@ -1371,6 +1379,7 @@ impl ScannerIOCache for SetDisks {
self.clone(),
&updates,
cache_snapshot,
initial_revisions.as_ref(),
cache_cycle_floor.as_ref(),
expected_publication_epoch,
)
+9 -1
View File
@@ -180,7 +180,7 @@ where
// canceled decommission remains suspended after its worker exits, so
// starting a scan in that state could build a snapshot that cannot be
// routed to the authoritative metadata object.
if store.scanner_data_usage_publication_blocked().await {
if store.scanner_data_movement_pause_status().await.paused {
debug!(
target: "rustfs::scanner::io",
event = EVENT_SCANNER_SET_STATE,
@@ -260,8 +260,13 @@ where
}
}
bucket_plan_complete &= buckets_by_source.keys().copied().collect::<HashSet<_>>() == *expected_sources;
let activity_digest = crate::scanner::scanner_activity_snapshot_digest(&activity_before);
let scan_plan_digest =
scanner_bucket_plan_digest(&all_buckets, crate::scanner::scanner_activity_structural_digest(&activity_before));
let mut execution_hasher = Sha256::new();
execution_hasher.update(scan_plan_digest.0);
execution_hasher.update(activity_digest);
let execution_digest = DataUsageScanPlanDigest(execution_hasher.finalize().into());
let dirty_usage_snapshot = Arc::new(snapshot_dirty_usage_buckets(&all_buckets, dirty_generation_before_bucket_list));
let scan_scope = resolve_scanner_bucket_scan_scope(
store,
@@ -326,6 +331,7 @@ where
};
return Ok(ScannerCycleResult::new(status, dirty_usage_clear)
.with_publication_epoch(publication_epoch)
.with_activity_digest(activity_digest)
.with_observational_snapshot_published(observational_snapshot_published)
.with_remote_publication_lease_targets(remote_publication_lease_targets)
.with_remote_dirty_usage_acknowledgements(remote_dirty_usage_acknowledgements));
@@ -410,6 +416,7 @@ where
all_buckets: Arc::clone(&all_buckets),
scope: scan_scope.clone(),
digest: scan_plan_digest,
execution_digest,
leader_epoch,
tier_registry_generation,
publication_epoch,
@@ -598,6 +605,7 @@ where
};
Ok(ScannerCycleResult::new(cycle_status, dirty_usage_clear)
.with_publication_epoch(publication_epoch)
.with_activity_digest(activity_digest)
.with_observational_snapshot_published(observational_snapshot_published)
.with_remote_publication_lease_targets(remote_publication_lease_targets)
.with_remote_dirty_usage_acknowledgements(remote_dirty_usage_acknowledgements)
+236 -1
View File
@@ -20,6 +20,7 @@ use crate::scanner_folder::ScannerItem;
use crate::storage_api::EcstoreScannerPeerDirtyUsageSnapshot;
use crate::storage_api::owner::{
EcstorePoolDecommissionInfo, EcstoreRebalStatus, EcstoreRebalanceInfo, EcstoreRebalanceMeta, EcstoreRebalanceStats,
ecstore_hold_namespace_commit,
};
use crate::storage_api::scan::{BucketOperations as _, DeleteBucketOptions, MakeBucketOptions, ObjectIO as _};
use crate::{
@@ -343,6 +344,16 @@ async fn multi_pool_scanner_cycle_publishes_combined_usage() {
.put_object(&bucket, object, &mut reader, &ScannerObjectOptions::default())
.await
.expect("object should be written to its selected pool");
// Quorum ACK can precede tail publication on the disk chosen to scan.
let lock = store.pools[pool_index].disk_set[0]
.new_ns_lock(&bucket, object)
.await
.expect("fixture namespace lock should be created");
let _settled = lock
.get_write_lock(Duration::from_secs(30))
.await
.expect("fixture rename tail should finish before the usage scan");
}
let ctx = CancellationToken::new();
@@ -362,7 +373,7 @@ async fn multi_pool_scanner_cycle_publishes_combined_usage() {
.buckets_usage
.get(&bucket)
.expect("combined bucket usage should be present");
assert_eq!(bucket_usage.objects_count, 2);
assert_eq!(bucket_usage.objects_count, 2, "{usage:?}");
assert_eq!(bucket_usage.size, 11);
assert_eq!(usage.objects_total_count, 2);
assert_eq!(usage.objects_total_size, 11);
@@ -372,6 +383,102 @@ async fn multi_pool_scanner_cycle_publishes_combined_usage() {
);
}
#[tokio::test]
#[serial]
async fn pending_put_commit_keeps_scanner_walk_live_without_authoritative_usage() {
let (_temp_dir, store) = setup_two_pool_scanner_store().await;
let bucket = format!("scanner-pending-put-{}", Uuid::new_v4().simple());
store
.make_bucket(&bucket, &MakeBucketOptions::default())
.await
.expect("bucket should be created across both pools");
for (pool_index, (object, body)) in [("pool-a", b"first".as_slice()), ("pool-b", b"second".as_slice())]
.into_iter()
.enumerate()
{
let mut reader = ScannerPutObjReader::from_vec(body.to_vec());
store.pools[pool_index].disk_set[0]
.put_object(
&bucket,
object,
&mut reader,
&ScannerObjectOptions {
no_lock: true,
..Default::default()
},
)
.await
.expect("fixture objects must finish their rename fanouts before scanning");
}
let mut pending = Some(ecstore_hold_namespace_commit(store.as_ref()));
let mut previous_activity_digest = None;
let mut structural_plan_digest = None;
for (cycle, converged) in [(1, false), (2, true)] {
if converged {
drop(pending.take());
}
assert_eq!(store.scanner_data_usage_publication_blocked().await, !converged);
assert!(!store.scanner_data_movement_pause_status().await.paused);
let activity = crate::scanner::probe_scanner_activity(store.as_ref(), false)
.await
.expect("the fixture activity should be observable");
let activity_digest = crate::scanner::scanner_activity_snapshot_digest(&activity);
if let Some(previous) = previous_activity_digest.replace(activity_digest) {
assert_ne!(previous, activity_digest, "draining a namespace commit must change the publication proof");
}
let ctx = CancellationToken::new();
let budget = ScannerCycleBudget::new_with_progress_tracking(&ctx, ScannerCycleBudgetConfig::default());
let (updates, mut receiver) = mpsc::channel(1);
let result = tokio::time::timeout(
Duration::from_secs(30),
ScannerIOCycle::nsscanner_with_status(
store.as_ref(),
ctx,
Arc::clone(&budget),
updates,
cycle,
1,
HealScanMode::Normal,
),
)
.await
.expect("namespace scanning must finish while a PUT commit is pending")
.expect("namespace scanning must remain available during a pending PUT commit");
assert_eq!(result.activity_digest(), Some(activity_digest));
if !converged {
assert_eq!(budget.progress().0, 2, "the pending commit must not suppress actual object traversal");
}
assert_eq!(
result.status,
if converged {
ScannerCycleStatus::Complete
} else {
ScannerCycleStatus::Superseded
}
);
let usage = receiver
.recv()
.await
.expect("the completed walk should produce a usage candidate");
assert_eq!(usage.usage_snapshot_converged, Some(converged));
assert_eq!(usage.scanner_cycle, Some(cycle));
assert_eq!(usage.objects_total_count, 2);
assert_eq!(usage.objects_total_size, 11);
assert_eq!(usage.usage_snapshot_set_states.len(), 2);
for state in &usage.usage_snapshot_set_states {
let digest = state
.scan_plan_digest
.expect("each set must retain its structural cache identity");
assert_eq!(*structural_plan_digest.get_or_insert(digest), digest);
}
let bucket_usage = usage.buckets_usage.get(&bucket).expect("the walked bucket must be present");
assert_eq!(bucket_usage.objects_count, 2);
assert_eq!(bucket_usage.size, 11);
assert!(receiver.recv().await.is_none(), "each walk must emit exactly one terminal candidate");
}
}
#[tokio::test]
#[serial]
async fn multi_pool_scanner_cycle_zero_fills_bucket_absent_from_first_pool() {
@@ -387,6 +494,16 @@ async fn multi_pool_scanner_cycle_zero_fills_bucket_absent_from_first_pool() {
.put_object(&bucket, "pool-b", &mut reader, &ScannerObjectOptions::default())
.await
.expect("object should be written only to the second pool");
{
let lock = store.pools[1].disk_set[0]
.new_ns_lock(&bucket, "pool-b")
.await
.expect("fixture namespace lock should be created");
let _settled = lock
.get_write_lock(Duration::from_secs(30))
.await
.expect("fixture rename tail should finish before the usage scan");
}
store.pools[0]
.delete_bucket(&bucket, &DeleteBucketOptions::default())
.await
@@ -797,6 +914,124 @@ fn complete_set_usage_cache(buckets: &[(&str, usize)], scan_plan_digest: DataUsa
cache
}
#[tokio::test]
#[serial]
async fn set_snapshot_reuse_requires_execution_identity_and_fences_stale_writers() {
let (_temp_dir, store) = setup_two_pool_scanner_store().await;
let set = Arc::clone(&store.pools[0].disk_set[0]);
let epoch = scanner_publication_epoch(Arc::clone(&set)).await.expect("idle set admission");
let mut legacy = complete_set_usage_cache(&[("photos", 5)], DataUsageScanPlanDigest([1; 32]));
legacy.info.source = Some(DataUsageCacheSource::new(0, 0));
legacy
.save(Arc::clone(&set), DATA_USAGE_CACHE_NAME)
.await
.expect("seed legacy set cache");
let mut persisted = DataUsageCache::default();
let initial = persisted
.load_with_revisions(Arc::clone(&set), DATA_USAGE_CACHE_NAME)
.await
.expect("capture the shared starting revision");
let mut fresh = legacy.clone();
fresh.info.scan_execution_digest = Some(DataUsageScanPlanDigest([2; 32]));
fresh.replace(
"photos",
DATA_USAGE_ROOT,
DataUsageEntry {
size: 20,
objects: 1,
..Default::default()
},
);
let cycle_floor = AtomicU64::new(fresh.info.next_cycle);
let (tx, mut rx) = mpsc::channel(1);
assert!(
persist_and_publish_cache_snapshot(Arc::clone(&set), &tx, fresh.clone(), Some(&initial), &cycle_floor, epoch)
.await
.is_some(),
"a legacy cache without execution identity must be refreshed"
);
let published = rx.try_recv().expect("fresh snapshot should be forwarded");
assert_eq!(published.find("photos").expect("published bucket").size, 20);
assert_eq!(published.info.scan_execution_digest, fresh.info.scan_execution_digest);
let current = persisted
.load_with_revisions(Arc::clone(&set), DATA_USAGE_CACHE_NAME)
.await
.expect("capture the current revision for the unidentified execution");
let mut stale = legacy.clone();
stale.info.scan_execution_digest = Some(DataUsageScanPlanDigest([3; 32]));
for (candidate, revisions) in [(stale, &initial), (legacy, &current)] {
assert!(
persist_and_publish_cache_snapshot(Arc::clone(&set), &tx, candidate, Some(revisions), &cycle_floor, epoch)
.await
.is_none(),
"a stale or unidentified execution must not replace the newer snapshot"
);
assert!(matches!(rx.try_recv(), Err(mpsc::error::TryRecvError::Empty)));
}
fresh.info.scan_execution_digest = Some(DataUsageScanPlanDigest([4; 32]));
assert!(
persist_and_publish_cache_snapshot(Arc::clone(&set), &tx, fresh.clone(), None, &cycle_floor, epoch)
.await
.is_none(),
"an unreadable starting revision must not authorize an overwrite"
);
fresh.info.scan_execution_digest = published.info.scan_execution_digest;
fresh.replace("photos", DATA_USAGE_ROOT, DataUsageEntry::default());
assert!(
persist_and_publish_cache_snapshot(Arc::clone(&set), &tx, fresh, Some(&initial), &cycle_floor, epoch)
.await
.is_some(),
"an overlapping identical execution must reuse the completed snapshot"
);
assert_eq!(
rx.try_recv()
.expect("reused snapshot")
.find("photos")
.expect("reused bucket")
.size,
20
);
persisted
.load(Arc::clone(&set), DATA_USAGE_CACHE_NAME)
.await
.expect("read the final durable set cache");
assert_eq!(persisted.find("photos").expect("durable bucket").size, 20);
assert_eq!(persisted.info.scan_execution_digest, published.info.scan_execution_digest);
let ctx = CancellationToken::new();
let empty_execution = DataUsageScanPlanDigest([5; 32]);
set.nsscanner_cache(
ctx.clone(),
ScannerCycleBudget::new(&ctx, ScannerCycleBudgetConfig::default()),
ScannerBucketScanPlan {
buckets: Vec::new(),
all_buckets: Arc::new(Vec::new()),
scope: ScannerBucketScanScope::default(),
digest: DataUsageScanPlanDigest([6; 32]),
execution_digest: empty_execution,
leader_epoch: 11,
tier_registry_generation: 13,
publication_epoch: Some(epoch),
dirty_usage_buckets: Arc::new(HashMap::new()),
bucket_failures: ScannerBucketFailureState::default(),
pending_maintenance_work: Arc::new(AtomicBool::new(false)),
cache_cycle_floor: Arc::new(AtomicU64::new(8)),
},
tx,
8,
HealScanMode::Normal,
)
.await
.expect("empty set scope should replace its prior nonempty cache");
let empty = rx.try_recv().expect("empty set snapshot should be published");
assert_eq!(empty.info.scan_execution_digest, Some(empty_execution));
assert!(empty.info.snapshot_complete);
let root = empty.checked_flatten(DATA_USAGE_ROOT).expect("complete empty root");
assert_eq!((root.size, root.objects), (0, 0));
}
fn complete_usage_baseline(
source: DataUsageCacheSource,
scan_plan_digest: DataUsageScanPlanDigest,
+3
View File
@@ -127,6 +127,9 @@ pub(crate) use rustfs_lifecycle::{
use rustfs_storage_api as storage_contracts;
pub(crate) mod owner {
#[cfg(test)]
pub(crate) use rustfs_ecstore::api::set_disk::test_util::hold_namespace_commit as ecstore_hold_namespace_commit;
pub(crate) use super::storage_contracts::{
HTTPPreconditions, HTTPRangeSpec, NS_SCANNER_PROTOCOL_VERSION, ObjectIO, ObjectOperations, ObjectToDelete,
};
+47 -1
View File
@@ -23,6 +23,30 @@ therefore has three identities:
If any identity changes before commit, the result is a candidate for retry or
observation, not an authoritative baseline.
Ordinary PUT rename fanouts also track instance-scoped in-flight work. A quorum
ACK does not release it: the actual disk tasks retain ownership until their
rename work ends, including when the request caller is cancelled. Scan admission
remains movement-only so sustained PUTs do not stop namespace walks and
scanner-driven lifecycle discovery. The post-walk local publication check and
remote publication leases reject pending fanouts. Begin/end namespace generations
invalidate scans and cached plans across the fanout; after acquiring remote
leases, the coordinator rechecks the full activity digest before publishing an
authoritative aggregate. This catches a tail that finishes between the scan's
last probe and lease acquisition.
This adds no namespace or movement lock. An already-verified older snapshot may
still precede a newly started write. Sustained or stalled PUT tails can delay
authoritative usage publication, which resumes through the existing retry
schedule rather than a new immediate-wakeup protocol. Intermediate per-set and
prefix cache readers retain their existing approximate-cache semantics. A
prolonged pending tail with no generation changes can also delay cycle advancement
and fresh rescans of already-current caches; this is not a guarantee of lifecycle
progress under indefinitely stalled storage I/O.
This PUT-tail protection requires every writer node to be upgraded. It does not
prove that a failed tail replica has healed, and it does not extend the same
in-flight tracking to multipart or other namespace mutation paths.
## Fences
The protocol uses separate fences because they exclude different stale inputs.
@@ -33,7 +57,7 @@ They must not be collapsed unless the replacement proves the same exclusions.
| Scanner leadership claim | scanner | competing scanner leaders and stale cycle writers |
| Storage publication epoch | ECStore | usage computed across rebalance, decommission, or other data-movement generations |
| Publication lease | scanner peers through ECStore-facing activity probes | remote dirty-usage or maintenance state that has not acknowledged the candidate |
| CAS revision | backing config object store | lost updates to `.usage.v2.json`, `.usage.json`, or cycle-state objects |
| CAS revision | backing config object store | lost updates to usage snapshots, scanner caches, or cycle-state objects |
| Per-set freshness | scanner aggregation | a merged usage snapshot that combines stale and current set results |
| Tier registry generation | scanner tier accounting | bytes classified against a different warm-tier registry |
| Usage floor identity | scanner publication and ECStore quota fallback | empty or legacy values becoming plausible authoritative quota input |
@@ -42,6 +66,28 @@ A reader that cannot prove the required fence for its surface must fail closed
or use the documented observed path below. It must not synthesize an empty usage
snapshot for a missing or corrupt authoritative object.
## Cache Execution Identity
The structural scan-plan digest can remain stable across ordinary bucket writes
so a scoped scan can retain unaffected baseline buckets. It is not sufficient
proof for reusing a completed result within the same cycle. Bucket work uses an
execution digest combining the structural plan and the full activity snapshot,
with the bucket's dirty generation included in its cache identity. Completed set
caches carry the same execution digest separately from their structural plan.
The persisted set-root fast path requires equal execution identities as well as
the existing source, cycle, leader, tier, and cache-structure checks.
A set scan also captures its starting cache revisions. When the persisted
execution differs, replacement requires those revisions to remain unchanged;
otherwise a slow scan could overwrite a newer completed result. The existing
cache lock, conditional save, and movement admission still fence the commit.
The optional `scan_execution_digest` field is appended to the map-encoded cache
metadata. Legacy caches remain readable but cannot satisfy same-cycle set-root
reuse without this identity. Older readers can ignore the added map key, but
older writers do not enforce its fence; readability is not a mixed-version
publication-safety guarantee.
## Persisted Objects
The persisted objects are part of the compatibility contract. Removing one
+14 -4
View File
@@ -79,9 +79,11 @@ Setting `"enabled": false` in the config has the same read-path effect as deleti
The status endpoint reports **the node that answered the request**. Counters, queue depth and breaker state are per-node runtime state, so in a distributed deployment query every node; the saved configuration and `updated_at` are cluster-wide.
### Backfill (ships with ODM-12)
### Backfill
Read-through only migrates what clients touch. The background backfill job walks the source listing and pulls the remainder, with a persisted checkpoint (`.rustfs.sys/buckets/<bucket>/on-demand-migration-backfill.json`), a single-owner lease, resume after restart, and `POST .../{bucket}/backfill?op=start|cancel` plus `GET .../{bucket}/backfill` admin routes. That slice (rustfs/backlog#2159) is not part of the build this page was written against: the shape above is the agreed design, and the exact request/response bodies must be re-checked against `docs/architecture/admin-route-action-snapshot.md` once it lands.
Backfill waits for the result of every pull, including a pull already queued by an online request. A failed or cancelled shared pull is counted as a failure, never as successful migration. The persisted continuation cursor stays at the first failed page; a takeover replays from there and skips objects already present locally. `completed_with_failures` is not a cutover-ready state.
Read-through only migrates what clients touch. The background backfill job walks the source listing and pulls the remainder, with a persisted checkpoint (`.rustfs.sys/buckets/<bucket>/on-demand-migration-backfill.json`), a single-owner lease, resume after restart, and `POST .../{bucket}/backfill?op=start|cancel` plus `GET .../{bucket}/backfill` admin routes. See `docs/architecture/admin-route-action-snapshot.md` for the route contract.
## Configuration reference
@@ -119,7 +121,7 @@ The persisted blob is `on-demand-migration.json` in the bucket's metadata. Unkno
| `policy.source_timeout.idle_ms` | integer | `30000` | `100..=600000`; enforced per body chunk on both the background pump and the inline tee |
| `policy.bandwidth_limit_bytes_per_sec` | integer \| null | `null` | When set, at least `65536` |
Values that are **not** configurable: the breaker opens after 5 consecutive counted failures inside a 30 s window, stays open for 30 s and then admits one probe (`breaker.rs`); the negative cache holds at most 100 000 keys per bucket with LRU eviction (`negative_cache.rs`); a background pull retries a retryable source failure at most 3 times with 1 s / 4 s / 16 s base delays plus up to 25 % jitter (`pull.rs`). The SDK's own retry policy is disabled on the source client, so one logical source call is exactly one wire request and the retry budget above is the only one.
Values that are **not** configurable: the breaker opens after 5 consecutive counted failures inside a 30 s window, stays open for 30 s and then admits one probe (`breaker.rs`); the negative cache holds at most 100 000 keys per bucket with LRU eviction (`negative_cache.rs`); a background pull retries a retryable source failure at most 3 times with 1 s / 4 s / 16 s base delays plus up to 25 % jitter (`pull.rs`). The SDK's own retry policy is disabled on the source client. Each SDK operation makes one wire request; an ambiguous HEAD 404 additionally probes the bucket, within the same configured first-byte budget.
Validation also rejects two shapes outright: a source whose endpoint and bucket name **this** bucket on this deployment (`SelfReference`), and a source that matches one of the bucket's own replication targets (`ReplicationLoop`) — that pairing would amplify a write-back into a loop.
@@ -150,6 +152,14 @@ No write, delete, ACL or versioning permission is required or used. Scope the po
Behaviour a client can observe. The "Test" column names the case that pins it: `*_test.rs` files live under `crates/e2e_test/src/on_demand_migration/`, and the unit tests live next to the code in `rustfs/src/app/object/get.rs`, `head.rs` and `shared.rs`.
ODM merged continuation tokens use a NUL-prefixed JSON envelope inside the existing base64 encoding. NUL is not valid in a local object key, so a legitimate JSON-shaped key can never be mistaken for a merged cursor. Upgrade every node before using list-through, and restart any in-progress ODM listing issued by an older build: its unframed JSON tokens cannot be distinguished from legitimate local keys. Ordinary local listing tokens remain unchanged. Tokens issued by this build can still resume the local side after list-through is disabled.
Source `HEAD` responses with status 404 require a successful bucket probe before being negative-cached. The source credential therefore needs permission for `HeadBucket` (S3 `ListBucket`); a prefix-restricted ListBucket policy can deny that probe, in which case the response is a source failure rather than a cached miss. A missing/inaccessible source bucket, a missing source version, or an ambiguous GET 404 is not proof that the requested key is absent. Conditional GET validators are checked against the actual source GET metadata as well as the advisory HEAD; a missing required validator fails with 424. Source LIST entries without a key or a non-negative size fail the page rather than fabricating an empty object.
Write-back currently requires namespace locking enabled and exactly one pool with one erasure set. Other topologies fail write-back explicitly as `unsupported`: source reads remain available, but backfill cannot complete successfully or certify cutover. This restriction avoids relying on a set-local condition across distinct pool or lock domains; it does not restrict ordinary S3 writes. Full cross-pool migration requires a globally fenced commit protocol.
On the supported topology, write-back uses a create-only check under the local storage commit lock for both single-part PUT and multipart completion. A client write that commits while ODM is reading the source is preserved. With `respect_local_delete_marker=true`, a concurrent versioned deletion is preserved too. An explicit `respect_local_delete_marker=false` still permits revival; an unversioned deletion has no tombstone and therefore cannot be distinguished from a key that has never existed locally.
| Situation | Behaviour | Test |
|---|---|---|
| GET miss, object at or below `inline_max_bytes` | One source GET, teed: the client streams while the same bytes are written locally. Later reads are local and carry no source marker | `get_basic_test.rs::get_miss_pulls_inline_and_serves_locally_afterwards`, `get.rs::odm_get_inline_streams_to_client_and_commits_the_same_bytes` |
@@ -219,7 +229,7 @@ Five provenance keys are written on every pulled object under both internal pref
| Concurrency limit | Local write amplification | `max_concurrent_pulls` permits shared by inline and background pulls |
| Bounded queue | Unbounded memory on a burst | `pull_queue_capacity` waiting jobs; overflow is counted as `queue_full` and never fails a client response |
| Bandwidth limit | Source and network saturation | `bandwidth_limit_bytes_per_sec` (minimum 64 KiB/s) on the source client |
| Retry budget | Transient source blips | Background pulls retry a retryable failure up to 3 times (1 s / 4 s / 16 s plus jitter). Inline pulls never retry: the bytes are already on their way to the client. The SDK retry policy on the source client is disabled (`RemoteS3RetryPolicy::Disabled`), so this is the only retry budget and one logical source call is exactly one wire request — replication targets keep the SDK's three attempts, declared on their own spec |
| Retry budget | Transient source blips | Background pulls retry a retryable failure up to 3 times (1 s / 4 s / 16 s plus jitter). Inline pulls never retry: the bytes are already on their way to the client. The SDK retry policy is disabled (`RemoteS3RetryPolicy::Disabled`); HEAD 404 also requires one bucket probe. Replication targets keep their separately declared three SDK attempts |
| Idle timeout | A source that answers and then goes quiet mid-body | `source_timeout.idle_ms` per body chunk on both paths. The budget measures the source read, upstream of the inline tee, so a slow client is never mistaken for an idle source; when it fires the client stream ends in an error and the write-back is discarded |
| Anti-loop marker | Migration chains between RustFS/MinIO deployments | Every source request carries `x-rustfs-source-proxy-request` and `x-minio-source-proxy-request`; a request carrying it is always answered locally |
| Outbound endpoint policy | SSRF | See [outbound-connection-policy.md](outbound-connection-policy.md) |
+80 -627
View File
@@ -66,8 +66,8 @@ use rustfs_madmin::{
ReplicateRemoveStatus, ResyncBucketStatus, SITE_REPL_API_VERSION, SR_IAM_ITEM_STS_ACC, SR_IAM_ITEM_STS_ACC_LEGACY,
SRBucketInfo, SRBucketMeta, SRBucketStatsSummary, SRGroupInfo, SRGroupStatsSummary, SRIAMItem, SRIAMUser,
SRILMExpiryStatsSummary, SRInfo, SRMetric, SRMetricsSummary, SRPeerError, SRPeerJoinReq, SRPendingOperation, SRPolicyMapping,
SRPolicyStatsSummary, SRRemoveReq, SRResyncOpStatus, SRSTSCredential, SRSiteSummary, SRStateEditReq, SRStateInfo,
SRStatusInfo, SRSvcAccChange, SRSvcAccCreate, SRUserStatsSummary, SiteReplicationInfo, SyncStatus, WorkerStat,
SRPolicyStatsSummary, SRRemoveReq, SRResyncOpStatus, SRSTSCredential, SRSessionPolicy, SRSiteSummary, SRStateEditReq,
SRStateInfo, SRStatusInfo, SRSvcAccChange, SRSvcAccCreate, SRUserStatsSummary, SiteReplicationInfo, SyncStatus, WorkerStat,
};
use rustfs_policy::policy::{
Policy,
@@ -87,7 +87,7 @@ use std::sync::{LazyLock, Mutex as StdMutex};
use std::time::Duration;
use time::OffsetDateTime;
use tokio::sync::Mutex;
use tracing::{debug, info, warn};
use tracing::{info, warn};
use url::Url;
use url::form_urlencoded;
use uuid::Uuid;
@@ -98,6 +98,7 @@ use uuid::Uuid;
// paths keep resolving while this file keeps only the HTTP handlers.
pub(crate) use crate::site_replication::*;
const SERVICE_ACCOUNT_ENVELOPE_VERSION: u64 = 2;
// Serializes peer-join admission (staleness check -> IAM upsert -> state
// commit) across every node of this site; see admit_peer_join. Never an
// actual object — only a namespace-lock key, like the repair execution lock.
@@ -1979,7 +1980,7 @@ async fn bootstrap_existing_metadata_after_add(
return errors;
}
};
let plan = match build_site_replication_bootstrap_plan(&info).await {
let plan = match site_replication_bootstrap_plan(&info) {
Ok(plan) => plan,
Err(err) => {
let mut errors = SiteReplicationErrorSummary::default();
@@ -3497,7 +3498,6 @@ fn remove_sites(mut state: SiteReplicationState, req: SRRemoveReq) -> SiteReplic
state.resync_status.clear();
state.retry_queue.clear();
state.iam_deletion_replays.clear();
state.iam_deletion_marks.clear();
state.pending_endpoint_refresh = None;
state.updated_at = Some(OffsetDateTime::now_utc());
return state;
@@ -3509,7 +3509,6 @@ fn remove_sites(mut state: SiteReplicationState, req: SRRemoveReq) -> SiteReplic
state.resync_status.clear();
state.retry_queue.clear();
state.iam_deletion_replays.clear();
state.iam_deletion_marks.clear();
state.pending_endpoint_refresh = None;
state.updated_at = Some(OffsetDateTime::now_utc());
return state;
@@ -5387,38 +5386,6 @@ fn is_stale_update(local_updated_at: OffsetDateTime, incoming_updated_at: Option
incoming_updated_at.is_some_and(|incoming_updated_at| incoming_updated_at < local_updated_at)
}
/// Verdict for an incoming IAM item judged against the local record it would
/// overwrite or delete (backlog#2291).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum IamItemVerdict {
/// Apply the item: there is no local record, the item carries no source
/// timestamp (older peer), or it is at least as new as the local record.
Apply,
/// The local record was written from a newer source change; acknowledge the
/// item without touching the record. Covers both directions: a delayed
/// grant must not undo a newer revoke, and a delayed revoke must not undo a
/// newer grant.
SkipStale,
}
/// Ordering rule shared by the `policy`, `policy-mapping` and `group-info`
/// item paths (and matching `iam-user` / `service-account`).
///
/// `local_record_updated_at` is `None` when the targeted record does not
/// exist locally: nothing can be stale relative to an absent record, so a
/// create is applied and a delete falls through to the idempotent no-op paths
/// (backlog#2071). A record that exists but predates timestamps passes
/// `Some(UNIX_EPOCH)` and therefore never rejects an item.
fn judge_iam_item_staleness(
local_record_updated_at: Option<OffsetDateTime>,
incoming_updated_at: Option<OffsetDateTime>,
) -> IamItemVerdict {
match local_record_updated_at {
Some(local_updated_at) if is_stale_update(local_updated_at, incoming_updated_at) => IamItemVerdict::SkipStale,
_ => IamItemVerdict::Apply,
}
}
fn bucket_meta_local_updated_at(
bucket_meta: &crate::admin::storage_api::bucket::metadata::BucketMetadata,
config_file: &str,
@@ -5619,17 +5586,6 @@ async fn apply_bucket_meta_item(item: SRBucketMeta) -> S3Result<()> {
_ => unreachable!(),
};
// Persist the SOURCE `updated_at` as the stored `*_config_updated_at`
// stamp (backlog#2292). The staleness gate above compares the next item's
// source time against that stamp, so stamping the local apply time would
// reject a newer source edit that was merely delivered after this write
// (two quick edits under delivery delay, or a peer clock ahead of ours).
// Items without a source time keep the local stamp; lc-config keeps it
// too: its staleness axis is the in-document `expiry_updated_at` the merge
// above records, and the whole-config time is only its deletion / legacy
// lower bound.
let source_updated_at = if item.r#type == "lc-config" { None } else { item.updated_at };
if !skip_config_write {
if let Some(data) = data {
if item.r#type == "quota-config" {
@@ -5648,25 +5604,13 @@ async fn apply_bucket_meta_item(item: SRBucketMeta) -> S3Result<()> {
"durable quota capability is not confirmed across the cluster".to_string(),
)
})?;
match source_updated_at {
Some(source_updated_at) => {
metadata_sys::update_quota_if_incarnation_at(
&item.bucket,
data,
expected_incarnation_id,
&proof,
source_updated_at,
)
.await
}
None => {
metadata_sys::update_quota_if_incarnation(&item.bucket, data, expected_incarnation_id, &proof).await
}
}
.map_err(ApiError::from)?;
metadata_sys::update_quota_if_incarnation(&item.bucket, data, expected_incarnation_id, &proof)
.await
.map_err(ApiError::from)?;
} else {
write_replicated_bucket_config(&item.bucket, config_file, data, expected_incarnation_id, source_updated_at)
.await?;
metadata_sys::update_if_incarnation(&item.bucket, config_file, data, expected_incarnation_id)
.await
.map_err(ApiError::from)?;
}
} else {
if let Some(guard) = lifecycle_guard.as_ref() {
@@ -5674,8 +5618,9 @@ async fn apply_bucket_meta_item(item: SRBucketMeta) -> S3Result<()> {
.await
.map_err(ApiError::from)?;
} else {
write_replicated_bucket_config(&item.bucket, config_file, data, expected_incarnation_id, source_updated_at)
.await?;
metadata_sys::update_if_incarnation(&item.bucket, config_file, data, expected_incarnation_id)
.await
.map_err(ApiError::from)?;
}
}
} else {
@@ -5711,30 +5656,45 @@ async fn apply_bucket_meta_item(item: SRBucketMeta) -> S3Result<()> {
Ok(())
}
/// Write one replicated bucket config, stamped with the item's source
/// `updated_at` when it carries one and with the local clock otherwise
/// (backlog#2292; see [`apply_bucket_meta_item`]).
async fn write_replicated_bucket_config(
bucket: &str,
config_file: &str,
data: Vec<u8>,
expected_incarnation_id: Uuid,
source_updated_at: Option<OffsetDateTime>,
) -> S3Result<()> {
match source_updated_at {
Some(source_updated_at) => {
metadata_sys::update_if_incarnation_at(bucket, config_file, data, expected_incarnation_id, source_updated_at).await
}
None => metadata_sys::update_if_incarnation(bucket, config_file, data, expected_incarnation_id).await,
}
.map_err(ApiError::from)?;
Ok(())
}
fn group_info_requires_upsert(update: &rustfs_madmin::GroupAddRemove) -> bool {
!update.is_remove
}
pub(crate) fn encode_service_account_replication_policy(
claims: &HashMap<String, Value>,
session_policy: Option<&str>,
) -> S3Result<(SRSessionPolicy, Option<rustfs_madmin::SRSvcAccReplicationEnvelope>)> {
if !claims.contains_key(OIDC_VIRTUAL_PARENT_CLAIM) {
return session_policy
.map(SRSessionPolicy::from_json)
.transpose()
.map(|policy| policy.unwrap_or_default())
.map(|policy| (policy, None))
.map_err(|err| s3_error!(InvalidArgument, "marshal policy failed: {:?}", err));
}
let policy = match session_policy {
Some(policy) => serde_json::from_str::<Policy>(policy)
.map_err(|err| s3_error!(InvalidArgument, "invalid service account replication policy: {:?}", err))?,
None => Policy::default(),
};
if policy.statements.is_empty() && (!policy.id.is_empty() || !policy.version.is_empty())
|| policy.version.is_empty() && !policy.statements.is_empty()
{
return Err(s3_error!(InvalidArgument, "service account replication policy is not normalized"));
}
let policy = serde_json::to_string(&policy)
.map_err(|err| s3_error!(InternalError, "marshal service account replication policy failed: {:?}", err))?;
let policy = SRSessionPolicy::from_json(&policy)
.map_err(|err| s3_error!(InternalError, "marshal service account replication policy failed: {:?}", err))?;
Ok((
policy,
Some(rustfs_madmin::SRSvcAccReplicationEnvelope {
version: SERVICE_ACCOUNT_ENVELOPE_VERSION,
}),
))
}
#[derive(Debug)]
struct ReplicatedServiceAccountPolicy {
policy: Option<Policy>,
@@ -5802,87 +5762,27 @@ async fn apply_iam_item(item: SRIAMItem) -> S3Result<()> {
return Err(s3_error!(InvalidRequest, "iam not init"));
};
let incoming_updated_at = item.updated_at;
let deletion_mark_entities = iam_item_deletion_mark_entities(&item);
let verdict = match item.r#type.as_str() {
"policy" => apply_iam_policy_item(&iam_sys, &item.name, item.policy, incoming_updated_at).await?,
"policy-mapping" => apply_iam_policy_mapping_item(&iam_sys, item.policy_mapping, incoming_updated_at).await?,
"group-info" => apply_iam_group_info_item(&iam_sys, item.group_info, incoming_updated_at).await?,
match item.r#type.as_str() {
"policy" => apply_iam_policy_item(&iam_sys, &item.name, item.policy).await,
"policy-mapping" => apply_iam_policy_mapping_item(&iam_sys, item.policy_mapping).await,
"group-info" => apply_iam_group_info_item(&iam_sys, item.group_info).await,
// MinIO madmin-go sends `SRIAMItemSTSAcc = "sts-account"`. The legacy alias
// `sts-credential` (emitted by older RustFS releases) stays accepted permanently
// so mixed-version RustFS sites keep replicating STS credentials during rolling
// upgrades; it is a compatibility layer, not temporary code.
SR_IAM_ITEM_STS_ACC | SR_IAM_ITEM_STS_ACC_LEGACY => {
apply_iam_sts_account_item(&iam_sys, item.sts_credential).await?;
IamItemVerdict::Apply
}
"iam-user" => apply_iam_user_item(&iam_sys, item.iam_user, incoming_updated_at).await?,
"service-account" => {
apply_iam_service_account_item(&iam_sys, item.svc_acc_change, incoming_updated_at).await?;
IamItemVerdict::Apply
}
_ => {
return Err(s3_error!(
NotImplemented,
"site replication IAM item type `{}` is not supported",
item.r#type
));
}
};
// A committed deletion leaves no record for the gate to judge later items
// against, so its source timestamp is kept as a mark (backlog#2291). The
// mark is part of applying the deletion: failing here makes the sender
// retry the (idempotent) deletion rather than leave a revoke that a stale
// grant could still undo.
if verdict == IamItemVerdict::Apply
&& let Some(deleted_at) = incoming_updated_at.filter(|_| !deletion_mark_entities.is_empty())
{
commit_iam_deletion_marks(deletion_mark_entities, deleted_at).await?;
}
Ok(())
}
/// The deletion mark consulted by the staleness gate when the targeted record
/// is absent: the newest recorded deletion of any of `entities`. An
/// unreadable state falls back to today's behaviour (no mark, the item is
/// applied) — the gate must not turn a state-object outage into rejected
/// IAM replication.
async fn local_iam_deletion_mark(entities: &[String]) -> Option<OffsetDateTime> {
match load_site_replication_state().await {
Ok(state) => iam_deletion_mark(&state, entities),
Err(err) => {
debug!(
component = LOG_COMPONENT_ADMIN,
subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
event = EVENT_ADMIN_SITE_REPLICATION_STATE,
result = "iam_deletion_mark_unavailable",
error = ?err,
"site replication state unreadable; applying IAM item without a deletion mark"
);
None
}
SR_IAM_ITEM_STS_ACC | SR_IAM_ITEM_STS_ACC_LEGACY => apply_iam_sts_account_item(&iam_sys, item.sts_credential).await,
"iam-user" => apply_iam_user_item(&iam_sys, item.iam_user, incoming_updated_at).await,
"service-account" => apply_iam_service_account_item(&iam_sys, item.svc_acc_change, incoming_updated_at).await,
_ => Err(s3_error!(
NotImplemented,
"site replication IAM item type `{}` is not supported",
item.r#type
)),
}
}
async fn apply_iam_policy_item(
iam_sys: &IamSys<ObjectStore>,
name: &str,
policy: Option<Value>,
incoming_updated_at: Option<OffsetDateTime>,
) -> S3Result<IamItemVerdict> {
// Judge the item against the local document's own timestamp so a delayed
// older body (or delete) cannot overwrite a newer edit; once the document
// is deleted, its deletion mark stands in for it (backlog#2291).
let local_updated_at = match iam_sys.get_policy_doc(name).await {
Ok(doc) => Some(doc.update_date.unwrap_or(OffsetDateTime::UNIX_EPOCH)),
Err(err) if rustfs_iam::error::is_err_no_such_policy(&err) => {
local_iam_deletion_mark(&[iam_policy_deletion_mark_entity(name)]).await
}
Err(err) => return Err(ApiError::from(err).into()),
};
if judge_iam_item_staleness(local_updated_at, incoming_updated_at) == IamItemVerdict::SkipStale {
return Ok(IamItemVerdict::SkipStale);
}
async fn apply_iam_policy_item(iam_sys: &IamSys<ObjectStore>, name: &str, policy: Option<Value>) -> S3Result<()> {
if let Some(policy) = policy {
let policy: Policy =
serde_json::from_value(policy).map_err(|e| s3_error!(InvalidRequest, "invalid policy body: {}", e))?;
@@ -5897,78 +5797,26 @@ async fn apply_iam_policy_item(
Err(err) => return Err(ApiError::from(err).into()),
}
}
Ok(IamItemVerdict::Apply)
Ok(())
}
async fn apply_iam_policy_mapping_item(
iam_sys: &IamSys<ObjectStore>,
policy_mapping: Option<SRPolicyMapping>,
incoming_updated_at: Option<OffsetDateTime>,
) -> S3Result<IamItemVerdict> {
async fn apply_iam_policy_mapping_item(iam_sys: &IamSys<ObjectStore>, policy_mapping: Option<SRPolicyMapping>) -> S3Result<()> {
let Some(mapping) = policy_mapping else {
return Err(s3_error!(InvalidRequest, "policyMapping is required"));
};
let user_type = user_type_from_sr_wire(mapping.user_type).ok_or_else(|| s3_error!(InvalidRequest, "invalid userType"))?;
// Judge the item against the stored mapping's timestamp so a delayed older
// attach (or an older detach, `policy == ""`) cannot overwrite a newer one
// (backlog#2291). A detach removes the mapping outright, so once it is
// gone the detach's deletion mark stands in for the record.
let local_updated_at = match iam_sys
.get_mapped_policy_record(&mapping.user_or_group, user_type, mapping.is_group)
.await
{
Some(record) => Some(record.update_at),
None => {
local_iam_deletion_mark(&[iam_policy_mapping_deletion_mark_entity(
&mapping.user_or_group,
mapping.user_type,
mapping.is_group,
)])
.await
}
};
if judge_iam_item_staleness(local_updated_at, incoming_updated_at) == IamItemVerdict::SkipStale {
return Ok(IamItemVerdict::SkipStale);
}
iam_sys
.policy_db_set(&mapping.user_or_group, user_type, mapping.is_group, &mapping.policy)
.await
.map_err(ApiError::from)?;
Ok(IamItemVerdict::Apply)
Ok(())
}
async fn apply_iam_group_info_item(
iam_sys: &IamSys<ObjectStore>,
group_info: Option<SRGroupInfo>,
incoming_updated_at: Option<OffsetDateTime>,
) -> S3Result<IamItemVerdict> {
async fn apply_iam_group_info_item(iam_sys: &IamSys<ObjectStore>, group_info: Option<SRGroupInfo>) -> S3Result<()> {
let Some(group_info) = group_info else {
return Err(s3_error!(InvalidRequest, "groupInfo is required"));
};
let update = group_info.update_req;
// The record is the group itself: its own timestamp moves on every
// membership or status change, so a delayed older add cannot re-add a
// member a newer removal took out, and a delayed older removal (or group
// delete) cannot undo a newer add (backlog#2291). Once the group is gone
// the marks of its deletion and of its members' removals stand in for it,
// so a stale add cannot re-create it or re-add a removed member.
let local_updated_at = match iam_sys.get_group_info(&update.group).await {
Some(group) => Some(group.update_at.unwrap_or(OffsetDateTime::UNIX_EPOCH)),
None => {
let entities: Vec<String> = std::iter::once(iam_group_deletion_mark_entity(&update.group))
.chain(
update
.members
.iter()
.map(|member| iam_group_member_deletion_mark_entity(&update.group, member)),
)
.collect();
local_iam_deletion_mark(&entities).await
}
};
if judge_iam_item_staleness(local_updated_at, incoming_updated_at) == IamItemVerdict::SkipStale {
return Ok(IamItemVerdict::SkipStale);
}
if !group_info_requires_upsert(&update) {
// Idempotent removal: a replayed deletion may find the group or a
// member already gone (deleted here earlier, or the user tombstone
@@ -5983,14 +5831,14 @@ async fn apply_iam_group_info_item(
}
}
if members.is_empty() && !update.members.is_empty() {
return Ok(IamItemVerdict::Apply);
return Ok(());
}
match iam_sys.remove_users_from_group(&update.group, members).await {
Ok(_) => {}
Err(err) if rustfs_iam::error::is_err_no_such_group(&err) => {}
Err(err) => return Err(ApiError::from(err).into()),
}
return Ok(IamItemVerdict::Apply);
return Ok(());
}
iam_sys
@@ -6001,7 +5849,7 @@ async fn apply_iam_group_info_item(
.set_group_status(&update.group, matches!(update.status, GroupStatus::Enabled))
.await
.map_err(ApiError::from)?;
Ok(IamItemVerdict::Apply)
Ok(())
}
async fn apply_iam_sts_account_item(iam_sys: &IamSys<ObjectStore>, sts_credential: Option<SRSTSCredential>) -> S3Result<()> {
@@ -6043,18 +5891,14 @@ async fn apply_iam_user_item(
iam_sys: &IamSys<ObjectStore>,
iam_user: Option<SRIAMUser>,
incoming_updated_at: Option<OffsetDateTime>,
) -> S3Result<IamItemVerdict> {
) -> S3Result<()> {
let Some(user) = iam_user else {
return Err(s3_error!(InvalidRequest, "iamUser is required"));
};
// Once the identity is deleted, its deletion mark stands in for the
// record so a stale re-create cannot resurrect it (backlog#2291).
let local_updated_at = match iam_sys.get_user(&user.access_key).await {
Some(local) => Some(local.update_at.unwrap_or(OffsetDateTime::UNIX_EPOCH)),
None => local_iam_deletion_mark(&[iam_user_deletion_mark_entity(&user.access_key)]).await,
};
if judge_iam_item_staleness(local_updated_at, incoming_updated_at) == IamItemVerdict::SkipStale {
return Ok(IamItemVerdict::SkipStale);
if let Some(local) = iam_sys.get_user(&user.access_key).await
&& is_stale_update(local.update_at.unwrap_or(OffsetDateTime::UNIX_EPOCH), incoming_updated_at)
{
return Ok(());
}
if user.is_delete_req {
iam_sys.delete_user(&user.access_key, true).await.map_err(ApiError::from)?;
@@ -6075,7 +5919,7 @@ async fn apply_iam_user_item(
.map_err(ApiError::from)?;
}
}
Ok(IamItemVerdict::Apply)
Ok(())
}
async fn apply_iam_service_account_item(
@@ -6088,14 +5932,10 @@ async fn apply_iam_service_account_item(
};
let envelope = change.oidc_service_account_envelope;
if let Some(create) = change.create {
// Like the user path: with the account already deleted here, the
// recorded deletion mark is the timestamp a stale create/update
// (a snapshot or a delayed delivery) has to beat (backlog#2291).
let local_updated_at = match iam_sys.get_user(&create.access_key).await {
Some(local) => Some(local.update_at.unwrap_or(OffsetDateTime::UNIX_EPOCH)),
None if create.access_key == SITE_REPLICATOR_SERVICE_ACCOUNT => None,
None => local_iam_deletion_mark(&[format!("svc-acc:{}", create.access_key)]).await,
};
let local_updated_at = iam_sys
.get_user(&create.access_key)
.await
.map(|local| local.update_at.unwrap_or(OffsetDateTime::UNIX_EPOCH));
let replicated_policy = if create.access_key == SITE_REPLICATOR_SERVICE_ACCOUNT {
if local_updated_at.is_some_and(|local_updated_at| is_stale_update(local_updated_at, incoming_updated_at)) {
return Ok(());
@@ -6139,8 +5979,6 @@ async fn apply_iam_service_account_item(
.map_err(ApiError::from)?;
}
Err(err) if is_err_no_such_service_account(&err) => {
let access_key = create.access_key.clone();
let status = create.status.clone();
iam_sys
.new_service_account(
&create.parent,
@@ -6158,28 +5996,6 @@ async fn apply_iam_service_account_item(
)
.await
.map_err(ApiError::from)?;
// A snapshot (bootstrap / repair / retry resend) carries the
// account's current status; creation always enables, so a
// disabled account must be switched off in a second step or
// the peer keeps accepting credentials the source rejects.
if !status.is_empty() && status != "on" {
iam_sys
.update_service_account(
&access_key,
UpdateServiceAccountOpts {
session_policy: None,
secret_key: None,
name: None,
description: None,
expiration: None,
status: Some(status),
parent_user: None,
allow_site_replicator_account: false,
},
)
.await
.map_err(ApiError::from)?;
}
}
Err(err) => return Err(ApiError::from(err).into()),
}
@@ -6295,7 +6111,6 @@ fn adopt_add_commit_state(state: &mut SiteReplicationState, next_state: SiteRepl
sync_state_initialized,
edit_generation: _,
applied_edit_generations: _,
iam_deletion_marks: _,
} = next_state;
state.name = name;
state.service_account_access_key = service_account_access_key;
@@ -7877,7 +7692,7 @@ impl Operation for SiteReplicationRepairHandler {
let local_peer = current_local_peer(&req, &state);
let body: SiteReplicationRepairRequest = read_site_replication_json(req, "", false).await?;
let info = build_sr_info(&state, &local_peer).await?;
let plan = build_site_replication_bootstrap_plan(&info).await?;
let plan = site_replication_bootstrap_plan(&info)?;
let signing_key = current_token_signing_key().ok_or_else(|| {
S3Error::with_message(S3ErrorCode::InternalError, "token signing key is not initialized".to_string())
})?;
@@ -8070,7 +7885,6 @@ impl Operation for SRRotateServiceAccountHandler {
mod tests {
use super::*;
use crate::site_replication::identity::deployment_id_for_endpoint;
use rustfs_madmin::SRSessionPolicy;
/// A peer the status probe could not reach must render as offline.
///
@@ -12372,299 +12186,6 @@ mod tests {
assert!(!is_stale_update(local, None));
}
/// Minimal model of one replicated IAM record (a policy document body, a
/// user/group mapping, or a group's member set) as the apply paths treat
/// it: `None` is "absent", `Some((content, stamp))` is the local record
/// with the timestamp of the change that last wrote it. Applying an item
/// goes through `judge_iam_item_staleness` exactly like the three apply
/// functions do; a delete (`incoming == None`) on an absent record is the
/// idempotent no-op of backlog#2071.
fn apply_iam_item_to_model(
record: &mut Option<(&'static str, OffsetDateTime)>,
incoming: Option<&'static str>,
incoming_updated_at: Option<OffsetDateTime>,
) -> IamItemVerdict {
let verdict = judge_iam_item_staleness(record.map(|(_, stamp)| stamp), incoming_updated_at);
if verdict == IamItemVerdict::Apply {
*record = incoming.map(|content| (content, incoming_updated_at.unwrap_or(OffsetDateTime::UNIX_EPOCH)));
}
verdict
}
fn at(seconds: i64) -> OffsetDateTime {
OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(seconds)
}
/// backlog#2291: a revoke (narrowed policy body, detached mapping, member
/// removed from the group) followed by the delayed delivery of the older
/// grant must leave the revoke in place.
#[test]
fn test_iam_item_stale_grant_after_revoke_is_not_applied() {
let mut record = None;
assert_eq!(apply_iam_item_to_model(&mut record, Some("grant"), Some(at(10))), IamItemVerdict::Apply);
assert_eq!(apply_iam_item_to_model(&mut record, Some("revoke"), Some(at(20))), IamItemVerdict::Apply);
// The older grant is redelivered (retry drain, slow peer) after the revoke.
assert_eq!(
apply_iam_item_to_model(&mut record, Some("grant"), Some(at(10))),
IamItemVerdict::SkipStale,
"a grant older than the local revoke must be acknowledged without being applied"
);
assert_eq!(record, Some(("revoke", at(20))), "the revoke must survive the stale grant");
}
/// backlog#2291: the mirror image — a grant followed by the delayed delivery
/// of an older revoke (older body, older detach, older member removal, or
/// an older delete) must leave the grant in place.
#[test]
fn test_iam_item_stale_revoke_after_grant_is_not_applied() {
let mut record = None;
assert_eq!(apply_iam_item_to_model(&mut record, Some("revoke"), Some(at(10))), IamItemVerdict::Apply);
assert_eq!(apply_iam_item_to_model(&mut record, Some("grant"), Some(at(20))), IamItemVerdict::Apply);
assert_eq!(
apply_iam_item_to_model(&mut record, Some("revoke"), Some(at(10))),
IamItemVerdict::SkipStale,
"a revoke older than the local grant must not be applied"
);
assert_eq!(
apply_iam_item_to_model(&mut record, None, Some(at(15))),
IamItemVerdict::SkipStale,
"a delete older than the local record must not remove it"
);
assert_eq!(record, Some(("grant", at(20))));
}
/// backlog#2291: an item at least as new as the local record is applied,
/// including a newer delete; equal timestamps are not stale (same rule as
/// `iam-user`).
#[test]
fn test_iam_item_newer_than_local_record_is_applied() {
let mut record = Some(("grant", at(20)));
assert_eq!(apply_iam_item_to_model(&mut record, Some("revoke"), Some(at(20))), IamItemVerdict::Apply);
assert_eq!(record, Some(("revoke", at(20))));
assert_eq!(apply_iam_item_to_model(&mut record, Some("grant"), Some(at(30))), IamItemVerdict::Apply);
assert_eq!(record, Some(("grant", at(30))));
assert_eq!(apply_iam_item_to_model(&mut record, None, Some(at(40))), IamItemVerdict::Apply);
assert_eq!(record, None, "a newer delete removes the record");
}
/// backlog#2291: peers that predate item timestamps keep today's
/// last-writer-wins behaviour — an item without `updatedAt` is applied even
/// over a newer local record.
#[test]
fn test_iam_item_without_source_timestamp_is_applied() {
let mut record = Some(("grant", at(20)));
assert_eq!(apply_iam_item_to_model(&mut record, Some("revoke"), None), IamItemVerdict::Apply);
assert_eq!(record.map(|(content, _)| content), Some("revoke"));
assert_eq!(judge_iam_item_staleness(Some(at(20)), None), IamItemVerdict::Apply);
assert_eq!(judge_iam_item_staleness(None, None), IamItemVerdict::Apply);
}
/// backlog#2291: nothing is stale relative to an absent record. A create
/// with any timestamp is applied, and a delete falls through to the
/// idempotent no-op paths (backlog#2071) instead of being judged.
#[test]
fn test_iam_item_targeting_absent_record_is_applied() {
assert_eq!(judge_iam_item_staleness(None, Some(at(1))), IamItemVerdict::Apply);
let mut record = None;
assert_eq!(apply_iam_item_to_model(&mut record, None, Some(at(1))), IamItemVerdict::Apply);
assert_eq!(record, None);
assert_eq!(apply_iam_item_to_model(&mut record, Some("grant"), Some(at(1))), IamItemVerdict::Apply);
assert_eq!(record, Some(("grant", at(1))));
// A record that predates timestamps is reported as UNIX_EPOCH by the
// apply paths and therefore never rejects an item.
assert_eq!(
judge_iam_item_staleness(Some(OffsetDateTime::UNIX_EPOCH), Some(at(1))),
IamItemVerdict::Apply
);
}
/// The apply paths once the record is gone: the local timestamp the gate
/// sees is the deletion mark (or `None` when no deletion was recorded),
/// and a committed deletion records its source timestamp as the mark —
/// the same sequence `apply_iam_item` runs.
fn apply_iam_item_to_deleted_record_model(
marks: &mut SiteReplicationState,
entity: &str,
incoming_is_delete: bool,
incoming_updated_at: Option<OffsetDateTime>,
) -> IamItemVerdict {
let entities = vec![entity.to_string()];
let verdict = judge_iam_item_staleness(iam_deletion_mark(marks, &entities), incoming_updated_at);
if verdict == IamItemVerdict::Apply
&& incoming_is_delete
&& let Some(deleted_at) = incoming_updated_at
{
record_iam_deletion_marks(marks, &entities, deleted_at);
}
verdict
}
/// backlog#2291 (real-VM case R6.3a of backlog#2080): a detach deletes the
/// mapping outright, so the older grant that arrives afterwards finds no
/// record — the deletion mark must stand in for it and reject the grant.
/// The same holds for a deleted policy document, user or group.
#[test]
fn test_iam_item_stale_grant_after_record_deletion_is_not_applied() {
let mut marks = SiteReplicationState::default();
let entity = "policy-mapping:alice:0:false";
// The revoke (detach) is applied first: the record is gone, the mark stays.
assert_eq!(
apply_iam_item_to_deleted_record_model(&mut marks, entity, true, Some(at(20))),
IamItemVerdict::Apply
);
assert_eq!(marks.iam_deletion_marks.get(entity), Some(&at(20)));
// The older grant is delivered after the revoke.
assert_eq!(
apply_iam_item_to_deleted_record_model(&mut marks, entity, false, Some(at(10))),
IamItemVerdict::SkipStale,
"a grant older than the recorded deletion must not re-create the record"
);
// A replayed copy of the same revoke stays a no-op and keeps the mark.
assert_eq!(
apply_iam_item_to_deleted_record_model(&mut marks, entity, true, Some(at(20))),
IamItemVerdict::Apply
);
assert_eq!(marks.iam_deletion_marks.get(entity), Some(&at(20)));
// An older replayed revoke is stale against the newer one.
assert_eq!(
apply_iam_item_to_deleted_record_model(&mut marks, entity, true, Some(at(5))),
IamItemVerdict::SkipStale
);
assert_eq!(
marks.iam_deletion_marks.get(entity),
Some(&at(20)),
"an older deletion never lowers the mark"
);
}
/// backlog#2291: a mark only fences items older than the deletion. A grant
/// newer than (or as new as) the recorded deletion re-creates the record,
/// an unmarked entity and an item without a source timestamp keep today's
/// behaviour.
#[test]
fn test_iam_item_newer_than_deletion_mark_is_applied() {
let mut marks = SiteReplicationState::default();
let entity = "policy:readonly";
assert_eq!(
apply_iam_item_to_deleted_record_model(&mut marks, entity, true, Some(at(20))),
IamItemVerdict::Apply
);
assert_eq!(
apply_iam_item_to_deleted_record_model(&mut marks, entity, false, Some(at(20))),
IamItemVerdict::Apply,
"a grant as new as the deletion is not stale"
);
assert_eq!(
apply_iam_item_to_deleted_record_model(&mut marks, entity, false, Some(at(30))),
IamItemVerdict::Apply
);
assert_eq!(
apply_iam_item_to_deleted_record_model(&mut marks, entity, false, None),
IamItemVerdict::Apply,
"an item from a peer without timestamps keeps last-writer-wins"
);
assert_eq!(
apply_iam_item_to_deleted_record_model(&mut marks, "policy:other", false, Some(at(1))),
IamItemVerdict::Apply,
"no mark, no record: nothing to be stale against"
);
assert_eq!(
judge_iam_item_staleness(iam_deletion_mark(&marks, &[]), Some(at(1))),
IamItemVerdict::Apply
);
}
/// backlog#2291: a group's removal marks are per member (plus the group
/// itself for a group delete), so with the group gone a stale add is
/// judged against the newest mark among the group and the members it
/// would add.
#[test]
fn test_iam_group_item_after_deletion_is_judged_against_member_marks() {
let mut marks = SiteReplicationState::default();
let bob = iam_group_member_deletion_mark_entity("devs", "bob");
let group = iam_group_deletion_mark_entity("devs");
record_iam_deletion_marks(&mut marks, std::slice::from_ref(&bob), at(20));
record_iam_deletion_marks(&mut marks, std::slice::from_ref(&group), at(30));
// The gate for an add of `bob` to the (deleted) group.
let add_bob = [group.clone(), bob.clone()];
assert_eq!(iam_deletion_mark(&marks, &add_bob), Some(at(30)));
assert_eq!(
judge_iam_item_staleness(iam_deletion_mark(&marks, &add_bob), Some(at(25))),
IamItemVerdict::SkipStale
);
assert_eq!(
judge_iam_item_staleness(iam_deletion_mark(&marks, &add_bob), Some(at(30))),
IamItemVerdict::Apply
);
// An add of `carol` to a group that was only ever partially emptied
// (no group delete) is judged against carol's own mark only.
marks.iam_deletion_marks.remove(&group);
let add_carol = [group.clone(), iam_group_member_deletion_mark_entity("devs", "carol")];
assert_eq!(iam_deletion_mark(&marks, &add_carol), None);
assert_eq!(
judge_iam_item_staleness(iam_deletion_mark(&marks, &add_carol), Some(at(1))),
IamItemVerdict::Apply
);
let add_bob = [group, bob];
assert_eq!(
judge_iam_item_staleness(iam_deletion_mark(&marks, &add_bob), Some(at(10))),
IamItemVerdict::SkipStale
);
}
/// Cheap wiring guard for backlog#2291: every one of the `policy`,
/// `policy-mapping`, `group-info` and `iam-user` apply paths must route
/// through the shared staleness verdict before it writes or deletes
/// anything, and must fall back to the deletion mark when the record is
/// absent; `apply_iam_item` must record the mark of a committed deletion.
/// The ordering rule itself is covered by the `test_iam_item_*` behaviour
/// tests above; this only pins that no path bypasses it again.
#[test]
fn test_iam_policy_mapping_and_group_items_gate_on_incoming_updated_at() {
let source = include_str!("site_replication.rs");
for (start, end) in [
("async fn apply_iam_policy_item(", "async fn apply_iam_policy_mapping_item("),
("async fn apply_iam_policy_mapping_item(", "async fn apply_iam_group_info_item("),
("async fn apply_iam_group_info_item(", "async fn apply_iam_sts_account_item("),
("async fn apply_iam_user_item(", "async fn apply_iam_service_account_item("),
] {
let body = source
.split(start)
.nth(1)
.and_then(|rest| rest.split(end).next())
.expect(start);
assert!(
body.contains("judge_iam_item_staleness(local_updated_at, incoming_updated_at)"),
"{start} must judge the item against the local record before applying it"
);
assert!(
body.contains("local_iam_deletion_mark("),
"{start} must fall back to the deletion mark when the record is absent"
);
}
let dispatch = source
.split("async fn apply_iam_item(")
.nth(1)
.and_then(|rest| rest.split("async fn local_iam_deletion_mark(").next())
.expect("apply_iam_item");
assert!(
dispatch.contains("commit_iam_deletion_marks(deletion_mark_entities, deleted_at)"),
"apply_iam_item must record the mark of a deletion it committed"
);
}
#[test]
fn test_apply_state_edit_req_only_updates_ilm_expiry_flags() {
let mut state = SiteReplicationState::default();
@@ -14196,72 +13717,4 @@ mod tests {
);
}
}
/// backlog#2292: the receiver persists the SOURCE `updated_at` of an
/// applied bucket config and judges the next item's source time against
/// it. Stamping the local apply time instead rejected a source edit that
/// was newer than the applied one but delivered after the local stamp
/// (two quick source edits under delivery delay; a peer clock ahead of
/// ours) and acknowledged it with 200.
#[test]
fn test_bucket_meta_staleness_is_judged_against_the_applied_source_timestamp() {
let apply_wall_clock = OffsetDateTime::now_utc();
let source_edit_t1 = apply_wall_clock - time::Duration::seconds(30);
let source_edit_t2 = source_edit_t1 + time::Duration::seconds(2);
let source_edit_t0 = source_edit_t1 - time::Duration::seconds(2);
assert!(
source_edit_t2 < apply_wall_clock,
"T2 is newer at the source yet older than the local apply clock"
);
// Edit T1 arrives first and is applied the way apply_bucket_meta_item
// persists a replicated config: stamped with its source time.
let mut meta = crate::admin::storage_api::bucket::metadata::BucketMetadata::new("photos");
meta.update_config_at(
BUCKET_POLICY_CONFIG,
br#"{"Version":"2012-10-17","Statement":[]}"#.to_vec(),
source_edit_t1,
)
.expect("apply edit T1");
let local_updated_at = bucket_meta_local_updated_at(&meta, BUCKET_POLICY_CONFIG);
assert_eq!(
local_updated_at, source_edit_t1,
"the stored stamp is the source time, not the apply clock"
);
// Edit T2 is newer at the source but delivered late: it must apply.
assert!(
!is_stale_update(local_updated_at, Some(source_edit_t2)),
"edit T2 ({source_edit_t2}) is newer than applied edit T1 ({source_edit_t1}) but is rejected against local stamp {local_updated_at}"
);
// Edit T0 predates the applied edit: it stays rejected.
assert!(
is_stale_update(local_updated_at, Some(source_edit_t0)),
"edit T0 ({source_edit_t0}) is older than applied edit T1 ({source_edit_t1}) and must be rejected"
);
// An item without a source time is never judged stale (unchanged).
assert!(!is_stale_update(local_updated_at, None));
}
/// backlog#2292: the replicated-config write in `apply_bucket_meta_item`
/// must go through the source-stamped entries; a plain
/// `update_if_incarnation` there would reintroduce local stamping.
#[test]
fn test_apply_bucket_meta_item_writes_through_the_source_stamped_entries() {
let source = include_str!("site_replication.rs");
let apply = source
.split("async fn apply_bucket_meta_item")
.nth(1)
.and_then(|rest| rest.split("fn group_info_requires_upsert").next())
.expect("apply_bucket_meta_item source");
assert!(
apply.contains("update_quota_if_incarnation_at("),
"durable quota must carry the source stamp"
);
assert!(apply.contains("update_if_incarnation_at("), "bucket configs must carry the source stamp");
assert!(
!apply.contains("metadata_sys::update_if_incarnation(&item.bucket"),
"no replicated config write may bypass the source stamp"
);
}
}
-38
View File
@@ -353,25 +353,6 @@ pub(crate) mod metadata_sys {
super::ecstore_bucket::metadata_sys::update_if_incarnation(bucket, config_file, data, expected_incarnation_id).await
}
/// [`update_if_incarnation`] stamping the config with a replicated edit's
/// source `updated_at` instead of the local clock (backlog#2292).
pub(crate) async fn update_if_incarnation_at(
bucket: &str,
config_file: &str,
data: Vec<u8>,
expected_incarnation_id: uuid::Uuid,
updated_at: OffsetDateTime,
) -> Result<OffsetDateTime> {
super::ecstore_bucket::metadata_sys::update_if_incarnation_at(
bucket,
config_file,
data,
expected_incarnation_id,
updated_at,
)
.await
}
pub(crate) async fn update_quota_if_incarnation(
bucket: &str,
data: Vec<u8>,
@@ -381,25 +362,6 @@ pub(crate) mod metadata_sys {
super::ecstore_bucket::metadata_sys::update_quota_if_incarnation(bucket, data, expected_incarnation_id, proof).await
}
/// [`update_quota_if_incarnation`] stamping the quota with a replicated
/// edit's source `updated_at` instead of the local clock (backlog#2292).
pub(crate) async fn update_quota_if_incarnation_at(
bucket: &str,
data: Vec<u8>,
expected_incarnation_id: uuid::Uuid,
proof: &super::ecstore_notification::CrossPoolFenceFleetProofToken,
updated_at: OffsetDateTime,
) -> Result<OffsetDateTime> {
super::ecstore_bucket::metadata_sys::update_quota_if_incarnation_at(
bucket,
data,
expected_incarnation_id,
proof,
updated_at,
)
.await
}
pub(crate) async fn capture_bucket_metadata_incarnation(bucket: &str) -> Result<uuid::Uuid> {
super::ecstore_bucket::metadata_sys::capture_bucket_metadata_incarnation(bucket).await
}
+3
View File
@@ -517,6 +517,9 @@ mod tests {
#[test]
fn a_plain_local_token_is_passed_through_and_a_tampered_one_is_rejected() {
let json_key = r#"{"t":"odm-list","v":1,"local_done":true}"#;
assert!(decode_list_cursor(Some(json_key)).expect("valid local key").is_none());
assert!(matches!(local_cursor(Some(json_key), None), LocalListCursor::Token(Some(local)) if local == json_key));
assert!(
decode_list_cursor(Some("photos/a.jpg"))
.expect("plain markers decode")
+83 -5
View File
@@ -4615,6 +4615,7 @@ fn odm_inline_client_body(primary: TeePrimary) -> StreamingBlob {
async fn odm_get_passthrough<S: OdmGetSource>(
state: &Arc<BucketOdmState>,
source: &S,
headers: &HeaderMap,
key: &str,
range: Option<&HTTPRangeSpec>,
backfill: Option<PullReason>,
@@ -4623,6 +4624,9 @@ async fn odm_get_passthrough<S: OdmGetSource>(
Ok(get) => get,
Err(err) => return OdmGetReply::Error(odm_get_source_failure(state, &err)),
};
if let Err(err) = odm_check_source_preconditions(headers, &get.head) {
return OdmGetReply::Error(err);
}
let content_length = match odm_content_length(get.head.size) {
Ok(length) => length,
Err(err) => {
@@ -4648,6 +4652,7 @@ async fn odm_get_passthrough<S: OdmGetSource>(
async fn odm_get_inline<S: OdmGetSource>(
state: &Arc<BucketOdmState>,
source: &S,
headers: &HeaderMap,
key: &str,
leader: PullLeader,
request_context: Option<request_context::RequestContext>,
@@ -4676,6 +4681,12 @@ async fn odm_get_inline<S: OdmGetSource>(
body,
content_range,
} = get;
// HEAD and GET can observe different source versions. Validate the
// representation whose body will actually be returned and persisted.
if let Err(err) = odm_check_source_preconditions(headers, &head) {
leader.complete(Err(PullError::canceled("source GET did not satisfy request preconditions")));
return OdmGetReply::Error(err);
}
// The object outgrew the inline budget between HEAD and GET: followers
// stream through on their own and the background pull stores it.
if head.size > policy.inline_max_bytes {
@@ -4758,19 +4769,19 @@ pub(super) async fn odm_get_from_source<S: OdmGetSource>(
let policy = &state.config().policy;
if let Some(range) = range {
let backfill = (policy.range_get == RangeGetPolicy::ServeAndBackfill).then_some(PullReason::RangeGet);
return odm_get_passthrough(state, source, key, Some(range), backfill).await;
return odm_get_passthrough(state, source, headers, key, Some(range), backfill).await;
}
if head.size > policy.inline_max_bytes {
return odm_get_passthrough(state, source, key, None, Some(PullReason::LargeObject)).await;
return odm_get_passthrough(state, source, headers, key, None, Some(PullReason::LargeObject)).await;
}
let slot = match state.acquire_pull_slot(key).await {
Ok(slot) => slot,
// The bucket state was torn down under this request: serve it
// without queueing anything on the old state.
Err(_) => return odm_get_passthrough(state, source, key, None, None).await,
Err(_) => return odm_get_passthrough(state, source, headers, key, None, None).await,
};
match slot {
PullSlot::Leader(leader) => odm_get_inline(state, source, key, leader, request_context).await,
PullSlot::Leader(leader) => odm_get_inline(state, source, headers, key, leader, request_context).await,
PullSlot::Follower(follower) => {
let first_byte = Duration::from_millis(policy.source_timeout.first_byte_ms);
match tokio::time::timeout(first_byte, follower.wait()).await {
@@ -4778,7 +4789,7 @@ pub(super) async fn odm_get_from_source<S: OdmGetSource>(
stats.record_request(OdmOp::Get, OdmOutcome::SourceHit);
OdmGetReply::RetryLocal
}
Ok(Err(_)) | Err(_) => odm_get_passthrough(state, source, key, None, None).await,
Ok(Err(_)) | Err(_) => odm_get_passthrough(state, source, headers, key, None, None).await,
}
}
}
@@ -5296,6 +5307,73 @@ mod on_demand_migration_tests {
assert!(rt.write_back.puts().is_empty());
}
#[tokio::test]
async fn odm_get_rechecks_conditions_against_the_get_representation() {
for inline_max_bytes in [0, 1024] {
for range in [
None,
Some(HTTPRangeSpec {
is_suffix_length: false,
start: 0,
end: 2,
}),
] {
let rt = runtime(
"changed-source",
PolicyConfig {
inline_max_bytes,
..Default::default()
},
)
.await;
let state = rt.state("changed-source");
let before = source_head(b"before");
let after = source_head(b"after!");
let source = ScriptedSource::new(vec![Ok(before.clone())], vec![Ok((after, b"after!".to_vec(), None))]);
let mut headers = HeaderMap::new();
headers.insert(
http::header::IF_MATCH,
HeaderValue::from_str(&format!("\"{}\"", before.etag.expect("etag"))).expect("header"),
);
let error = failed(odm_get_from_source(&state, &source, &headers, KEY, range.as_ref(), None).await);
assert_eq!(error.code(), &S3ErrorCode::PreconditionFailed);
assert_eq!(source.get_calls(), 1);
assert_eq!(state.inflight_keys(), 0);
assert!(rt.write_back.puts().is_empty(), "a failed condition must not start write-back");
}
}
}
#[tokio::test]
async fn odm_get_missing_validators_cannot_bypass_a_condition() {
for inline_max_bytes in [0, 1024] {
let rt = runtime(
"missing-validator",
PolicyConfig {
inline_max_bytes,
..Default::default()
},
)
.await;
let state = rt.state("missing-validator");
let before = source_head(b"before");
let after = SourceHead {
size: 6,
..Default::default()
};
let source = ScriptedSource::new(vec![Ok(before.clone())], vec![Ok((after, b"after!".to_vec(), None))]);
let mut headers = HeaderMap::new();
headers.insert(
http::header::IF_MATCH,
HeaderValue::from_str(&format!("\"{}\"", before.etag.expect("etag"))).expect("header"),
);
let error = failed(odm_get_from_source(&state, &source, &headers, KEY, None, None).await);
assert_eq!(error.status_code(), Some(StatusCode::FAILED_DEPENDENCY));
assert_eq!(error.message(), Some("missing_source_validator"));
assert!(rt.write_back.puts().is_empty());
}
}
#[tokio::test]
async fn odm_get_source_not_found_is_404_and_negative_cached() {
let rt = runtime("n", PolicyConfig::default()).await;
+17 -2
View File
@@ -57,6 +57,9 @@ pub(crate) struct InternalPutContext {
pub(crate) expected_md5_hex: Option<String>,
/// ETag to store instead of the computed one.
pub(crate) preserve_etag: Option<String>,
/// Reject an existing current object under the storage commit lock.
pub(crate) if_absent: bool,
pub(crate) preserve_delete_marker: bool,
pub(crate) content_headers: HashMap<String, String>,
pub(crate) user_metadata: HashMap<String, String>,
pub(crate) tags: Option<String>,
@@ -240,6 +243,8 @@ impl DefaultObjectUsecase {
size,
expected_md5_hex,
preserve_etag,
if_absent,
preserve_delete_marker,
content_headers,
user_metadata,
tags,
@@ -252,7 +257,10 @@ impl DefaultObjectUsecase {
};
let size = i64::try_from(size).map_err(|_| ApiError::invalid_request("internal put size exceeds the supported range"))?;
let headers = internal_put_headers(&content_headers)?;
let mut headers = internal_put_headers(&content_headers)?;
if if_absent {
headers.insert(http::header::IF_NONE_MATCH, HeaderValue::from_static("*"));
}
validate_internal_write_target(&key, &bucket, &headers).await?;
remove_source_replication_bookkeeping(&mut internal_metadata);
@@ -287,6 +295,7 @@ impl DefaultObjectUsecase {
origin: PutObjectOrigin::Internal {
principal_id,
emit_events,
preserve_delete_marker,
},
};
let committed = self
@@ -527,10 +536,14 @@ impl DefaultObjectUsecase {
.map_err(api_error_from_s3)?;
let store = self.object_store().ok_or_else(not_initialized)?;
let headers = HeaderMap::new();
let mut headers = HeaderMap::new();
if ctx.if_absent {
headers.insert(http::header::IF_NONE_MATCH, HeaderValue::from_static("*"));
}
let mut opts =
get_complete_multipart_upload_opts_with_replication_authorization(&headers, false).map_err(ApiError::from)?;
opts.preserve_etag = ctx.preserve_etag.clone();
opts.preserve_delete_marker = ctx.preserve_delete_marker;
let versioned = BucketVersioningSys::prefix_enabled(&bucket, &key).await;
opts.versioned = versioned;
opts.version_suspended = BucketVersioningSys::prefix_suspended(&bucket, &key).await;
@@ -747,6 +760,8 @@ mod tests {
size: Some(body.len() as u64),
expected_md5_hex: Some(md5_hex(body)),
preserve_etag: None,
if_absent: false,
preserve_delete_marker: false,
content_headers: HashMap::from([
("Content-Type".to_string(), "text/plain".to_string()),
("Cache-Control".to_string(), "max-age=60".to_string()),
@@ -66,6 +66,15 @@ impl OnDemandMigrationWriteBack {
.object_store()
.ok_or_else(|| WriteBackError::Local("object store is not initialized".to_string()))
}
fn require_atomic_write_back(&self) -> Result<(), WriteBackError> {
if !self.store()?.supports_atomic_create_only_write_back() {
return Err(WriteBackError::Unsupported(
"write-back requires namespace locking and exactly one pool with one erasure set".to_string(),
));
}
Ok(())
}
}
fn rfc3339(time: OffsetDateTime) -> String {
@@ -161,6 +170,8 @@ pub(super) async fn write_back_context(request: &WriteBackRequest, single_part:
size: Some(head.size),
expected_md5_hex: single_part.then(|| expected_md5_hex(head)).flatten(),
preserve_etag,
if_absent: true,
preserve_delete_marker: request.respect_delete_marker,
content_headers: content_headers(head),
user_metadata: head.user_metadata.clone(),
tags: request.tags.as_ref().and_then(encode_tags),
@@ -207,6 +218,7 @@ impl OdmWriteBack for OnDemandMigrationWriteBack {
}
async fn put_object(&self, request: &WriteBackRequest, body: WriteBackBody) -> Result<WriteBackOutcome, WriteBackError> {
self.require_atomic_write_back()?;
let ctx = write_back_context(request, true).await;
self.usecase()
.internal_put_object(ctx, body)
@@ -216,6 +228,7 @@ impl OdmWriteBack for OnDemandMigrationWriteBack {
}
async fn create_multipart_upload(&self, request: &WriteBackRequest) -> Result<String, WriteBackError> {
self.require_atomic_write_back()?;
let ctx = write_back_context(request, false).await;
self.usecase()
.internal_create_multipart_upload(&ctx)
@@ -249,6 +262,7 @@ impl OdmWriteBack for OnDemandMigrationWriteBack {
upload_id: &str,
parts: Vec<WriteBackPart>,
) -> Result<WriteBackOutcome, WriteBackError> {
self.require_atomic_write_back()?;
let ctx = write_back_context(request, false).await;
let parts = parts
.into_iter()
@@ -334,6 +348,7 @@ mod tests {
pulled_at: OffsetDateTime::from_unix_timestamp(1_756_800_000).expect("valid timestamp"),
preserve_etag: true,
emit_events: true,
respect_delete_marker: true,
tags: Some(HashMap::from([
("team".to_string(), "storage".to_string()),
("env".to_string(), "prod".to_string()),
@@ -486,6 +501,33 @@ mod tests {
assert!(!local.delete_marker);
}
#[tokio::test]
#[serial_test::serial]
async fn write_back_rejects_unsupported_topology_before_any_mutation() {
let (_dir, _paths, store) = crate::app::gating_test_env::isolated_multi_pool_ecstore().await;
crate::app::runtime_sources::install_test_app_context(Arc::clone(&store)).await;
let bucket = "odm-unsupported";
store
.make_bucket(bucket, &MakeBucketOptions::default())
.await
.expect("bucket");
let write_back = OnDemandMigrationWriteBack::new();
let req = request(bucket, "key", source_head(b"source"));
assert!(matches!(
write_back.put_object(&req, body_stream(b"source")).await,
Err(WriteBackError::Unsupported(_))
));
assert!(matches!(
write_back.create_multipart_upload(&req).await,
Err(WriteBackError::Unsupported(_))
));
assert!(matches!(
write_back.complete_multipart_upload(&req, "no-session", Vec::new()).await,
Err(WriteBackError::Unsupported(_))
));
assert_nothing_left(&store, bucket, "key").await;
}
#[tokio::test]
#[serial_test::serial]
async fn write_back_integrity_failure_leaves_nothing_behind() {
@@ -503,6 +545,133 @@ mod tests {
assert_nothing_left(&store, &bucket, "wrong.bin").await;
}
#[tokio::test]
#[serial_test::serial]
async fn write_back_commit_does_not_overwrite_a_concurrent_client_put() {
use crate::app::storage_api::test::set_disk::{PutObjectCommitBarrier, PutObjectCommitPause};
for versioned in [false, true] {
let (store, bucket) = write_back_test_bucket("odm-wb-race", versioned).await;
let source = b"old source bytes";
let client = b"new client bytes";
let req = request(&bucket, "race", source_head(source));
let client_req = request(&bucket, "race", source_head(client));
let mut client_ctx = write_back_context(&client_req, true).await;
client_ctx.if_absent = false;
let client_after = PutObjectCommitBarrier::install(&bucket, "race", PutObjectCommitPause::AfterNamespace);
let client_put = tokio::spawn(async move {
DefaultObjectUsecase::from_global()
.internal_put_object(client_ctx, body_stream(client))
.await
});
client_after.wait_until_paused().await;
let source_before = PutObjectCommitBarrier::install(&bucket, "race", PutObjectCommitPause::BeforeNamespace);
let write_back = OnDemandMigrationWriteBack::new();
let (result, ()) = tokio::join!(write_back.put_object(&req, body_stream(source)), async {
source_before.wait_until_paused().await;
drop(source_before);
drop(client_after);
});
let committed = client_put.await.expect("client task").expect("ordinary client write wins");
assert!(
matches!(result, Err(WriteBackError::Local(ref error)) if error.contains("PreconditionFailed")),
"{result:?}"
);
let stored = stored_object(&store, &bucket, "race").await;
assert_eq!(stored.etag, committed.etag);
assert_eq!(stored.version_id, committed.version_id);
assert_eq!(committed.version_id.is_some(), versioned);
assert_eq!(raw_object_bytes(&store, &bucket, "race").await, client);
}
}
#[tokio::test]
#[serial_test::serial]
async fn write_back_multipart_completion_preserves_a_client_put_after_staging() {
let (store, bucket) = write_back_test_bucket("odm-mpu-race", false).await;
let write_back = OnDemandMigrationWriteBack::new();
let req = request(&bucket, "race", source_head(b"source"));
let upload_id = write_back.create_multipart_upload(&req).await.expect("create");
let part = write_back
.upload_part(&req, &upload_id, 1, 6, body_stream(b"source"))
.await
.expect("stage");
let mut client_ctx = write_back_context(&request(&bucket, "race", source_head(b"client")), true).await;
client_ctx.if_absent = false;
let committed = DefaultObjectUsecase::from_global()
.internal_put_object(client_ctx, body_stream(b"client"))
.await
.expect("client put after staging");
let result = write_back.complete_multipart_upload(&req, &upload_id, vec![part]).await;
assert!(
matches!(result, Err(WriteBackError::Local(ref error)) if error.contains("PreconditionFailed")),
"{result:?}"
);
write_back
.abort_multipart_upload(&bucket, "race", &upload_id)
.await
.expect("abort rejected upload");
let stored = stored_object(&store, &bucket, "race").await;
assert_eq!(stored.etag, committed.etag);
assert_eq!(stored.version_id, committed.version_id);
assert_eq!(raw_object_bytes(&store, &bucket, "race").await, b"client");
}
#[tokio::test]
#[serial_test::serial]
async fn write_back_preserves_delete_markers_unless_policy_allows_revival() {
for multipart in [false, true] {
let (store, bucket) = write_back_test_bucket("odm-wb-tombstone", true).await;
let write_back = OnDemandMigrationWriteBack::new();
let mut req = request(&bucket, "deleted", source_head(b"source"));
let staged = if multipart {
let id = write_back.create_multipart_upload(&req).await.expect("create");
let part = write_back
.upload_part(&req, &id, 1, 6, body_stream(b"source"))
.await
.expect("part");
Some((id, part))
} else {
None
};
store
.delete_object(
&bucket,
"deleted",
ObjectOptions {
versioned: true,
..Default::default()
},
)
.await
.expect("delete marker");
let marker = stored_object(&store, &bucket, "deleted").await;
assert!(marker.delete_marker);
let rejected = if let Some((id, part)) = staged {
let result = write_back.complete_multipart_upload(&req, &id, vec![part]).await;
write_back
.abort_multipart_upload(&bucket, "deleted", &id)
.await
.expect("abort");
result
} else {
write_back.put_object(&req, body_stream(b"source")).await
};
assert!(
matches!(rejected, Err(WriteBackError::Local(ref error)) if error.contains("PreconditionFailed")),
"{rejected:?}"
);
let retained = stored_object(&store, &bucket, "deleted").await;
assert!(retained.delete_marker);
assert_eq!(retained.version_id, marker.version_id);
req.respect_delete_marker = false;
write_back
.put_object(&req, body_stream(b"source"))
.await
.expect("explicit revival policy");
assert!(!stored_object(&store, &bucket, "deleted").await.delete_marker);
}
}
#[tokio::test]
#[serial_test::serial]
async fn write_back_truncated_stream_leaves_nothing_behind() {
+12 -1
View File
@@ -949,7 +949,11 @@ pub(super) enum PutObjectOrigin<'a> {
/// request and no credential: managed-SSE authorization treats the write
/// as internal, and the creation event, when requested, names
/// `principal_id` instead of an access key.
Internal { principal_id: &'static str, emit_events: bool },
Internal {
principal_id: &'static str,
emit_events: bool,
preserve_delete_marker: bool,
},
}
impl PutObjectOrigin<'_> {
@@ -1603,6 +1607,12 @@ impl DefaultObjectUsecase {
if let Some(etag) = preserve_etag {
opts.preserve_etag = Some(etag);
}
if let PutObjectOrigin::Internal {
preserve_delete_marker, ..
} = &origin
{
opts.preserve_delete_marker = *preserve_delete_marker;
}
if let Some(quota_check) = quota_check.as_ref() {
apply_quota_admission(&mut opts, quota_check)?;
}
@@ -1769,6 +1779,7 @@ impl DefaultObjectUsecase {
PutObjectOrigin::Internal {
principal_id,
emit_events,
..
} => {
let principal_id = *principal_id;
let request_context = request_context::RequestContext::fallback();
+69 -2
View File
@@ -1014,11 +1014,44 @@ pub(crate) fn mark_on_demand_migration_list_local_only(headers: &mut HeaderMap)
/// forwarded to the source: a 304/412 answered by the source would be
/// indistinguishable from a source failure.
pub(crate) fn odm_check_source_preconditions(headers: &HeaderMap, head: &SourceHead) -> S3Result<()> {
let if_match = headers
.get(http::header::IF_MATCH)
.and_then(|value| value.to_str().ok())
.map(str::trim);
let if_none_match = headers
.get(http::header::IF_NONE_MATCH)
.and_then(|value| value.to_str().ok())
.map(str::trim);
let needs_etag = if_match.is_some_and(|value| value != "*") || if_none_match.is_some_and(|value| value != "*");
let needs_mtime = (!headers.contains_key(http::header::IF_MATCH) && headers.contains_key(http::header::IF_UNMODIFIED_SINCE))
|| (!headers.contains_key(http::header::IF_NONE_MATCH) && headers.contains_key(http::header::IF_MODIFIED_SINCE));
if (needs_etag && head.etag.is_none()) || (needs_mtime && head.last_modified.is_none()) {
return Err(odm_source_unavailable_error("missing_source_validator"));
}
let info = ObjectInfo {
etag: head.etag.clone(),
mod_time: head.last_modified.map(OffsetDateTime::from),
..Default::default()
};
// A successful source read establishes wildcard existence, but the
// remaining conditions must still run in their ordinary precedence.
if head.etag.is_none() && (if_match == Some("*") || if_none_match == Some("*")) {
let mut remaining = headers.clone();
if if_match == Some("*") {
remaining.remove(http::header::IF_MATCH);
remaining.remove(http::header::IF_UNMODIFIED_SINCE);
}
if if_none_match == Some("*") {
remaining.remove(http::header::IF_NONE_MATCH);
remaining.remove(http::header::IF_MODIFIED_SINCE);
}
check_preconditions(&remaining, &info)?;
return if if_none_match == Some("*") {
Err(S3Error::new(S3ErrorCode::NotModified))
} else {
Ok(())
};
}
check_preconditions(headers, &info)
}
@@ -2075,8 +2108,42 @@ mod on_demand_migration_tests {
.expect_err("modified since an earlier date is 412");
assert_eq!(err.code(), &S3ErrorCode::PreconditionFailed);
// A source without validators cannot fail a precondition.
let bare = SourceHead::default();
assert!(odm_check_source_preconditions(&headers_with(http::header::IF_MATCH, "\"other\""), &bare).is_ok());
let err =
odm_check_source_preconditions(&headers_with(http::header::IF_MATCH, "\"other\""), &bare).expect_err("missing ETag");
assert_eq!(err.status_code(), Some(http::StatusCode::FAILED_DEPENDENCY));
assert!(odm_check_source_preconditions(&headers_with(http::header::IF_MATCH, "*"), &bare).is_ok());
let err =
odm_check_source_preconditions(&headers_with(http::header::IF_NONE_MATCH, "*"), &bare).expect_err("source exists");
assert_eq!(err.code(), &S3ErrorCode::NotModified);
let dated = SourceHead {
last_modified: head.last_modified,
..Default::default()
};
assert!(odm_check_source_preconditions(&headers_with(http::header::IF_MATCH, "*"), &dated).is_ok());
let mut combined = headers_with(http::header::IF_NONE_MATCH, "*");
combined.insert(http::header::IF_MATCH, HeaderValue::from_static("\"other\""));
assert_eq!(
odm_check_source_preconditions(&combined, &dated)
.expect_err("specific ETag unavailable")
.status_code(),
Some(http::StatusCode::FAILED_DEPENDENCY)
);
combined.remove(http::header::IF_MATCH);
combined.insert(
http::header::IF_UNMODIFIED_SINCE,
HeaderValue::from_static("Wed, 21 Oct 2015 07:28:00 GMT"),
);
assert_eq!(
odm_check_source_preconditions(&combined, &dated)
.expect_err("unmodified-since fails before none-match")
.code(),
&S3ErrorCode::PreconditionFailed
);
for header in [http::header::IF_MODIFIED_SINCE, http::header::IF_UNMODIFIED_SINCE] {
let err = odm_check_source_preconditions(&headers_with(header, "Wed, 21 Oct 2015 07:28:00 GMT"), &bare)
.expect_err("missing timestamp");
assert_eq!(err.status_code(), Some(http::StatusCode::FAILED_DEPENDENCY));
}
}
}
-2
View File
@@ -726,8 +726,6 @@ pub(crate) mod bucket {
delete_marker_version_id: None,
delete_marker: false,
delete_marker_mtime: None,
target_delete_marker_version_ids: Default::default(),
target_delete_marker_version_ids_corrupt: false,
target_arns,
force_delete_id: Some(operation_id),
force_delete_generation: Some(i64::try_from(generation.unix_timestamp_nanos()).unwrap_or(i64::MAX)),
+18 -229
View File
@@ -302,164 +302,7 @@ pub(crate) fn site_replication_state_replicates_ilm_expiry(state: &SiteReplicati
state.peers.values().any(|peer| peer.replicate_ilm_expiry)
}
/// Secret-bearing half of the IAM snapshot. `SRInfo` is served to admin
/// callers (`site-replication/info`, status, add preflight) and must stay
/// secret-free, so the bootstrap plan receives credentials through this
/// separate value, built only on the paths that deliver to peers (site add
/// bootstrap, repair, retry snapshot resend). Never persisted, never served.
#[derive(Debug, Clone, Default)]
pub(crate) struct SiteReplicationIamCredentials {
/// Built-in users (access key -> credential); temp and service accounts
/// are excluded, external/IdP users never appear here.
pub(crate) users: BTreeMap<String, SiteReplicationUserCredential>,
/// Every service account except the site replicator's own, already
/// shaped as the `service-account` create item the live hook emits.
pub(crate) service_accounts: Vec<SiteReplicationServiceAccountSnapshot>,
}
#[derive(Debug, Clone)]
pub(crate) struct SiteReplicationUserCredential {
pub(crate) secret_key: String,
pub(crate) status: AccountStatus,
/// The user record's own update time (the axis the receiver's staleness
/// check compares against), unlike `UserInfo::updated_at` which
/// `list_users` overwrites with the policy mapping's time.
pub(crate) updated_at: Option<OffsetDateTime>,
}
#[derive(Debug, Clone)]
pub(crate) struct SiteReplicationServiceAccountSnapshot {
pub(crate) create: SRSvcAccCreate,
pub(crate) envelope: Option<SRSvcAccReplicationEnvelope>,
pub(crate) updated_at: Option<OffsetDateTime>,
}
pub(crate) const SERVICE_ACCOUNT_ENVELOPE_VERSION: u64 = 2;
pub(crate) fn encode_service_account_replication_policy(
claims: &HashMap<String, Value>,
session_policy: Option<&str>,
) -> S3Result<(SRSessionPolicy, Option<SRSvcAccReplicationEnvelope>)> {
if !claims.contains_key(OIDC_VIRTUAL_PARENT_CLAIM) {
return session_policy
.map(SRSessionPolicy::from_json)
.transpose()
.map(|policy| policy.unwrap_or_default())
.map(|policy| (policy, None))
.map_err(|err| s3_error!(InvalidArgument, "marshal policy failed: {:?}", err));
}
let policy = match session_policy {
Some(policy) => serde_json::from_str::<Policy>(policy)
.map_err(|err| s3_error!(InvalidArgument, "invalid service account replication policy: {:?}", err))?,
None => Policy::default(),
};
if policy.statements.is_empty() && (!policy.id.is_empty() || !policy.version.is_empty())
|| policy.version.is_empty() && !policy.statements.is_empty()
{
return Err(s3_error!(InvalidArgument, "service account replication policy is not normalized"));
}
let policy = serde_json::to_string(&policy)
.map_err(|err| s3_error!(InternalError, "marshal service account replication policy failed: {:?}", err))?;
let policy = SRSessionPolicy::from_json(&policy)
.map_err(|err| s3_error!(InternalError, "marshal service account replication policy failed: {:?}", err))?;
Ok((
policy,
Some(SRSvcAccReplicationEnvelope {
version: SERVICE_ACCOUNT_ENVELOPE_VERSION,
}),
))
}
/// Read the credentials the IAM snapshot needs straight from the IAM store:
/// `list_users` deliberately strips secret keys and skips service accounts,
/// which is right for an admin listing and wrong for a peer snapshot (the
/// plan builder used to drop every user for lack of a secret, so a status
/// change or secret rotation committed while a peer was unreachable never
/// reached it — backlog#2289).
pub(crate) async fn build_sr_iam_credentials() -> S3Result<SiteReplicationIamCredentials> {
let mut credentials = SiteReplicationIamCredentials::default();
let Some(iam_sys) = current_iam_handle() else {
return Ok(credentials);
};
let mut users = HashMap::new();
iam_sys.load_users(UserType::Reg, &mut users).await.map_err(ApiError::from)?;
for (access_key, identity) in users {
if identity.credentials.is_temp() || identity.credentials.is_service_account() {
continue;
}
credentials.users.insert(
access_key,
SiteReplicationUserCredential {
secret_key: identity.credentials.secret_key,
status: if identity.credentials.status == "off" {
AccountStatus::Disabled
} else {
AccountStatus::Enabled
},
updated_at: identity.update_at,
},
);
}
let mut service_accounts = HashMap::new();
iam_sys
.load_users(UserType::Svc, &mut service_accounts)
.await
.map_err(ApiError::from)?;
let mut service_accounts: Vec<_> = service_accounts.into_iter().collect();
service_accounts.sort_by(|(a, _), (b, _)| a.cmp(b));
for (access_key, identity) in service_accounts {
// The replicator account is installed by join / rotate, never by a snapshot.
if access_key == SITE_REPLICATOR_SERVICE_ACCOUNT || !identity.credentials.is_service_account() {
continue;
}
let claims = iam_sys.get_claims_for_svc_acc(&access_key).await.map_err(ApiError::from)?;
let (account, session_policy) = iam_sys.get_service_account(&access_key).await.map_err(ApiError::from)?;
let session_policy = session_policy
.map(|policy| serde_json::to_string(&policy))
.transpose()
.map_err(|err| {
S3Error::with_message(
S3ErrorCode::InternalError,
format!("marshal service account session policy failed: {err:?}"),
)
})?;
let (session_policy, envelope) = encode_service_account_replication_policy(&claims, session_policy.as_deref())?;
credentials.service_accounts.push(SiteReplicationServiceAccountSnapshot {
create: SRSvcAccCreate {
parent: identity.credentials.parent_user,
access_key,
secret_key: identity.credentials.secret_key,
groups: identity.credentials.groups.unwrap_or_default(),
claims,
session_policy,
status: identity.credentials.status,
name: account.name.unwrap_or_default(),
description: account.description.unwrap_or_default(),
expiration: account.expiration,
api_version: Some(SITE_REPL_API_VERSION.to_string()),
},
envelope,
updated_at: identity.update_at,
});
}
Ok(credentials)
}
/// The bootstrap plan for peer delivery: `info` (secret-free) plus the IAM
/// credentials read at this moment.
pub(crate) async fn build_site_replication_bootstrap_plan(info: &SRInfo) -> S3Result<SiteReplicationBootstrapPlan> {
let credentials = build_sr_iam_credentials().await?;
site_replication_bootstrap_plan(info, &credentials)
}
pub(crate) fn site_replication_bootstrap_plan(
info: &SRInfo,
credentials: &SiteReplicationIamCredentials,
) -> S3Result<SiteReplicationBootstrapPlan> {
pub(crate) fn site_replication_bootstrap_plan(info: &SRInfo) -> S3Result<SiteReplicationBootstrapPlan> {
let mut plan = SiteReplicationBootstrapPlan::default();
let replicate_ilm_expiry = site_replication_info_replicates_ilm_expiry(info);
@@ -475,57 +318,24 @@ pub(crate) fn site_replication_bootstrap_plan(
}
for (access_key, user) in &info.user_info_map {
// Credentials come from the store snapshot; an inline `secret_key` on
// the SRInfo entry (older callers, tests) is accepted as a fallback.
// Users with neither (external / IdP identities) have nothing a peer
// could install and are skipped.
let credential = credentials.users.get(access_key);
let Some(secret_key) = credential
.map(|credential| credential.secret_key.clone())
.or_else(|| user.secret_key.clone())
.filter(|secret_key| !secret_key.is_empty())
else {
continue;
};
let status = credential
.map(|credential| credential.status.clone())
.unwrap_or_else(|| user.status.clone());
let updated_at = credential.and_then(|credential| credential.updated_at).or(user.updated_at);
plan.iam_items.push(SRIAMItem {
r#type: "iam-user".to_string(),
iam_user: Some(rustfs_madmin::SRIAMUser {
access_key: access_key.clone(),
is_delete_req: false,
user_req: Some(AddOrUpdateUserReq {
secret_key,
policy: user.policy_name.clone(),
status,
if let Some(secret_key) = &user.secret_key {
plan.iam_items.push(SRIAMItem {
r#type: "iam-user".to_string(),
iam_user: Some(rustfs_madmin::SRIAMUser {
access_key: access_key.clone(),
is_delete_req: false,
user_req: Some(AddOrUpdateUserReq {
secret_key: secret_key.clone(),
policy: user.policy_name.clone(),
status: user.status.clone(),
}),
api_version: Some(SITE_REPL_API_VERSION.to_string()),
}),
api_version: Some(SITE_REPL_API_VERSION.to_string()),
}),
updated_at,
api_version: Some(SITE_REPL_API_VERSION.to_string()),
..Default::default()
});
}
// Service accounts follow their parents: the receiver creates a missing
// account under `parent` and updates an existing one (secret, status,
// session policy), so a rotation or disable committed during an outage
// converges through the same snapshot as users do.
for account in &credentials.service_accounts {
plan.iam_items.push(SRIAMItem {
r#type: "service-account".to_string(),
svc_acc_change: Some(SRSvcAccChange {
create: Some(account.create.clone()),
oidc_service_account_envelope: account.envelope.clone(),
updated_at: user.updated_at,
api_version: Some(SITE_REPL_API_VERSION.to_string()),
..Default::default()
}),
updated_at: account.updated_at,
api_version: Some(SITE_REPL_API_VERSION.to_string()),
..Default::default()
});
});
}
}
for (name, desc) in &info.group_desc_map {
@@ -708,12 +518,7 @@ pub(crate) async fn broadcast_site_replication_make_bucket(
} else {
path
};
// Both steps run to completion on their own: the broadcast attempts every
// peer and reports the first failure (backlog#2293), so stopping here on
// that error would skip `configure-replication` for the peers whose
// `make` just succeeded — and nothing records a retry for that gap. The
// failed peer's retry events cover both steps independently.
let make_result = broadcast_site_replication_json_using_runtime(runtime, &path, &serde_json::json!({})).await;
broadcast_site_replication_json_using_runtime(runtime, &path, &serde_json::json!({})).await?;
let configure_path = bootstrap_bucket_op_path(bucket, "configure-replication");
let configure_path = if let Some(token) = bootstrap_token {
@@ -721,8 +526,7 @@ pub(crate) async fn broadcast_site_replication_make_bucket(
} else {
configure_path
};
let configure_result = broadcast_site_replication_json_using_runtime(runtime, &configure_path, &serde_json::json!({})).await;
make_result.and(configure_result)
broadcast_site_replication_json_using_runtime(runtime, &configure_path, &serde_json::json!({})).await
}
const SITE_REPLICATION_DELETE_INTENT_PENDING: &str =
@@ -1028,21 +832,6 @@ pub async fn site_replication_iam_change_hook(item: SRIAMItem) -> S3Result<()> {
let Some(runtime) = runtime_site_replication_targets().await? else {
return Ok(());
};
// A local revoke must out-rank a stale grant a peer delivers later, so its
// mark is committed before the broadcast (backlog#2291). The broadcast
// still goes out when the mark cannot be persisted: the peers' own records
// remain the primary gate, the mark only covers the deleted case.
if let Err(err) = record_iam_deletion_marks_for_item(&item).await {
warn!(
component = LOG_COMPONENT_ADMIN,
subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
event = EVENT_ADMIN_SITE_REPLICATION_STATE,
item_type = %item.r#type,
result = "iam_deletion_mark_not_recorded",
error = ?err,
"failed to record local IAM deletion mark before broadcast"
);
}
let mut first_error: Option<S3Error> = None;
for peer in runtime.state.peers.values() {
if peer.deployment_id == runtime.local_peer.deployment_id
+3 -26
View File
@@ -79,16 +79,13 @@ use http::header::{CONTENT_TYPE, HOST};
use http::{HeaderMap, HeaderValue, Uri};
use hyper::{Method, StatusCode};
use rustfs_config::{DEFAULT_CONSOLE_ADDRESS, DEFAULT_RUSTFS_TLS_PATH, ENV_RUSTFS_CONSOLE_ADDRESS, ENV_RUSTFS_TLS_PATH};
use rustfs_iam::federation::OIDC_VIRTUAL_PARENT_CLAIM;
use rustfs_iam::store::{MappedPolicy, UserType, sr_wire_user_type};
use rustfs_iam::sys::SITE_REPLICATOR_SERVICE_ACCOUNT;
use rustfs_madmin::{
AccountStatus, AddOrUpdateUserReq, GroupAddRemove, GroupStatus, PeerInfo, PeerSite, ReplicateEditStatus,
SITE_REPL_API_VERSION, SRBucketInfo, SRBucketMeta, SRGroupInfo, SRIAMItem, SRIAMPolicy, SRInfo, SRPolicyMapping, SRRemoveReq,
SRResyncOpStatus, SRRetryStats, SRSessionPolicy, SRStateInfo, SRSvcAccChange, SRSvcAccCreate, SRSvcAccDelete,
SRSvcAccReplicationEnvelope, SyncStatus,
AddOrUpdateUserReq, GroupAddRemove, GroupStatus, PeerInfo, PeerSite, ReplicateEditStatus, SITE_REPL_API_VERSION,
SRBucketInfo, SRBucketMeta, SRGroupInfo, SRIAMItem, SRIAMPolicy, SRInfo, SRPolicyMapping, SRRemoveReq, SRResyncOpStatus,
SRRetryStats, SRStateInfo, SyncStatus,
};
use rustfs_policy::policy::Policy;
use rustfs_signer::constants::UNSIGNED_PAYLOAD;
use rustfs_signer::sign_v4;
use rustfs_tls_runtime::{GlobalPublishedOutboundTlsState, TlsGeneration};
@@ -110,26 +107,6 @@ use tracing::{info, warn};
use url::{Url, form_urlencoded};
use uuid::Uuid;
/// Serialize `value` with every JSON object's keys sorted, for hashing and
/// equality checks. `HashMap` fields (service-account claims) iterate in a
/// per-instance random order and `serde_json` is built with `preserve_order`,
/// so two identical plans would otherwise hash differently: the repair
/// preflight token went stale between dry-run and execute, and a retry
/// snapshot resend never looked "stable" (backlog#2289 follow-up).
pub(crate) fn canonical_json_vec<T: Serialize>(value: &T) -> serde_json::Result<Vec<u8>> {
fn sort_keys(value: Value) -> Value {
match value {
Value::Object(map) => {
let sorted: BTreeMap<String, Value> = map.into_iter().map(|(key, value)| (key, sort_keys(value))).collect();
Value::Object(sorted.into_iter().collect())
}
Value::Array(items) => Value::Array(items.into_iter().map(sort_keys).collect()),
other => other,
}
}
serde_json::to_vec(&sort_keys(serde_json::to_value(value)?))
}
pub(crate) const LOG_COMPONENT_ADMIN: &str = "admin";
pub(crate) const LOG_SUBSYSTEM_SITE_REPLICATION: &str = "site_replication";
+3 -3
View File
@@ -234,9 +234,9 @@ impl SiteReplicationRepairTask<'_> {
pub(crate) fn id(&self) -> S3Result<String> {
let payload = match self {
Self::Iam(item) => canonical_json_vec(item),
Self::Iam(item) => serde_json::to_vec(item),
Self::BucketMake(_) | Self::Replication(_) => serde_json::to_vec(&serde_json::json!({})),
Self::BucketMetadata(item) => canonical_json_vec(item),
Self::BucketMetadata(item) => serde_json::to_vec(item),
}
.map_err(|err| S3Error::with_message(S3ErrorCode::InternalError, format!("serialize repair task failed: {err}")))?;
let mut digest = Sha256::new();
@@ -726,7 +726,7 @@ pub(crate) async fn execute_site_replication_repair_locked(
return Err(s3_error!(InvalidRequest, "site replication is not configured"));
}
let info = build_sr_info(&state, &request.local_peer).await?;
let plan = build_site_replication_bootstrap_plan(&info).await?;
let plan = site_replication_bootstrap_plan(&info)?;
let plan_token = site_replication_repair_plan_token(&state, &plan)?;
let preflight_token = site_replication_repair_preflight_token(&state, &plan, request.signing_key.as_bytes())?;
let sites = site_replication_repair_sites(&state, &request.local_peer, &plan, request.signing_key.as_bytes())?;
+7 -107
View File
@@ -397,12 +397,12 @@ pub(crate) fn iam_deletion_replay_matches(record: &SiteReplicationIamDeletionRep
/// newer revision of one another.
pub(crate) fn iam_item_deletion_entity(item: &SRIAMItem) -> Option<String> {
match item.r#type.as_str() {
"policy" if item.policy.is_none() => Some(iam_policy_deletion_mark_entity(&item.name)),
"policy" if item.policy.is_none() => Some(format!("policy:{}", item.name)),
"iam-user" => item
.iam_user
.as_ref()
.filter(|user| user.is_delete_req)
.map(|user| iam_user_deletion_mark_entity(&user.access_key)),
.map(|user| format!("iam-user:{}", user.access_key)),
"group-info" => item
.group_info
.as_ref()
@@ -416,7 +416,7 @@ pub(crate) fn iam_item_deletion_entity(item: &SRIAMItem) -> Option<String> {
.policy_mapping
.as_ref()
.filter(|mapping| mapping.policy.is_empty())
.map(|mapping| iam_policy_mapping_deletion_mark_entity(&mapping.user_or_group, mapping.user_type, mapping.is_group)),
.map(|mapping| format!("policy-mapping:{}:{}:{}", mapping.user_or_group, mapping.user_type, mapping.is_group)),
"service-account" => item
.svc_acc_change
.as_ref()
@@ -426,82 +426,6 @@ pub(crate) fn iam_item_deletion_entity(item: &SRIAMItem) -> Option<String> {
}
}
/// The entities whose deletion a deletion-shaped IAM item commits, keyed the
/// way the receive-side staleness gate looks them up once the local record is
/// gone (backlog#2291); empty for creates and updates. Group member removal
/// yields one entity per removed member so a stale re-add of that member can
/// be judged, and a group delete (no members) yields the group itself.
pub(crate) fn iam_item_deletion_mark_entities(item: &SRIAMItem) -> Vec<String> {
if item.r#type == "group-info" {
let Some(update) = item
.group_info
.as_ref()
.map(|group| &group.update_req)
.filter(|update| update.is_remove)
else {
return Vec::new();
};
if update.members.is_empty() {
return vec![iam_group_deletion_mark_entity(&update.group)];
}
return update
.members
.iter()
.map(|member| iam_group_member_deletion_mark_entity(&update.group, member))
.collect();
}
iam_item_deletion_entity(item).into_iter().collect()
}
pub(crate) fn iam_policy_deletion_mark_entity(name: &str) -> String {
format!("policy:{name}")
}
pub(crate) fn iam_user_deletion_mark_entity(access_key: &str) -> String {
format!("iam-user:{access_key}")
}
/// `user_type` is the SR wire integer, as carried by the item on both sides.
pub(crate) fn iam_policy_mapping_deletion_mark_entity(user_or_group: &str, user_type: i64, is_group: bool) -> String {
format!("policy-mapping:{user_or_group}:{user_type}:{is_group}")
}
pub(crate) fn iam_group_deletion_mark_entity(group: &str) -> String {
format!("group:{group}")
}
pub(crate) fn iam_group_member_deletion_mark_entity(group: &str, member: &str) -> String {
format!("group-member:{group}:{member}")
}
/// Persist the deletion marks of `item` (its source `updated_at` per entity
/// of [`iam_item_deletion_mark_entities`]) through the state transaction.
/// No-op for creates/updates and for items without a source timestamp
/// (older peers): a mark without a source clock could not be ordered against
/// later items. Called before a local deletion is broadcast and after a
/// replicated deletion is applied, so both sides out-rank a stale grant that
/// arrives later.
pub(crate) async fn record_iam_deletion_marks_for_item(item: &SRIAMItem) -> S3Result<()> {
let entities = iam_item_deletion_mark_entities(item);
let Some(deleted_at) = item.updated_at.filter(|_| !entities.is_empty()) else {
return Ok(());
};
commit_iam_deletion_marks(entities, deleted_at).await
}
/// [`record_iam_deletion_marks`] under the state transaction; the write is
/// skipped when no mark moves.
pub(crate) async fn commit_iam_deletion_marks(entities: Vec<String>, deleted_at: OffsetDateTime) -> S3Result<()> {
update_site_replication_state_when_changed(move |state| {
Ok(if record_iam_deletion_marks(state, &entities, deleted_at) {
StateCommit::Changed(())
} else {
StateCommit::Unchanged(())
})
})
.await
}
/// Failure bookkeeping for one IAM item delivery: upsert the collapsed retry
/// event and, when the item is a deletion, record its body for replay. Both
/// live in the same state so the caller commits them in one transaction — a
@@ -867,8 +791,8 @@ impl RetrySnapshot {
pub(crate) fn fingerprint(&self) -> S3Result<Vec<Vec<u8>>> {
let mut payloads = match self {
Self::Iam(items) => items.iter().map(canonical_json_vec).collect::<Result<Vec<_>, _>>(),
Self::BucketMetadata(items) => items.iter().map(canonical_json_vec).collect::<Result<Vec<_>, _>>(),
Self::Iam(items) => items.iter().map(serde_json::to_vec).collect::<Result<Vec<_>, _>>(),
Self::BucketMetadata(items) => items.iter().map(serde_json::to_vec).collect::<Result<Vec<_>, _>>(),
}
.map_err(|err| S3Error::with_message(S3ErrorCode::InternalError, format!("serialize retry snapshot failed: {err}")))?;
payloads.sort_unstable();
@@ -1030,7 +954,6 @@ pub(crate) enum IamSnapshotKey {
User(String),
Group(String),
PolicyMapping { target: String, user_type: i64, is_group: bool },
ServiceAccount(String),
}
pub(crate) fn iam_snapshot_key(item: &SRIAMItem) -> Option<IamSnapshotKey> {
@@ -1049,11 +972,6 @@ pub(crate) fn iam_snapshot_key(item: &SRIAMItem) -> Option<IamSnapshotKey> {
user_type: mapping.user_type,
is_group: mapping.is_group,
}),
"service-account" => item
.svc_acc_change
.as_ref()
.and_then(|change| change.create.as_ref())
.map(|create| IamSnapshotKey::ServiceAccount(create.access_key.clone())),
_ => None,
}
}
@@ -1088,24 +1006,6 @@ pub(crate) fn iam_snapshot_tombstones(item: &SRIAMItem, observed_at: OffsetDateT
mapping.policy.clear();
}
}
"service-account" => {
let Some(access_key) = item
.svc_acc_change
.as_ref()
.and_then(|change| change.create.as_ref())
.map(|create| create.access_key.clone())
else {
return Vec::new();
};
tombstone.svc_acc_change = Some(SRSvcAccChange {
delete: Some(SRSvcAccDelete {
access_key,
api_version: Some(SITE_REPL_API_VERSION.to_string()),
}),
api_version: Some(SITE_REPL_API_VERSION.to_string()),
..Default::default()
});
}
_ => return Vec::new(),
}
vec![tombstone]
@@ -1801,7 +1701,7 @@ pub(crate) async fn drain_site_replication_retry_queue_locked(
// tick and only when a snapshot resend is actually due.
let plan = if needs_plan {
let info = build_sr_info(&runtime.state, &runtime.local_peer).await?;
Some(build_site_replication_bootstrap_plan(&info).await?)
Some(site_replication_bootstrap_plan(&info)?)
} else {
None
};
@@ -1941,7 +1841,7 @@ pub(crate) async fn drain_one_site_replication_retry_event(
}
}
let fresh_info = build_sr_info(&runtime.state, &runtime.local_peer).await?;
let fresh_plan = build_site_replication_bootstrap_plan(&fresh_info).await?;
let fresh_plan = site_replication_bootstrap_plan(&fresh_info)?;
let fresh_snapshot = RetrySnapshot::from_plan(&action, &fresh_plan).expect("snapshot action has a snapshot");
if fresh_snapshot.fingerprint()? == current_fingerprint {
if is_iam {
-80
View File
@@ -64,86 +64,6 @@ pub(crate) struct SiteReplicationState {
/// newer edit that already landed.
#[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
pub(crate) applied_edit_generations: BTreeMap<String, u64>,
/// Source timestamp of the newest IAM deletion committed on this site,
/// keyed by the deleted entity (`iam_item_deletion_mark_entities`). A
/// deletion leaves no local record to judge a later item against, so this
/// is what lets the receive-side staleness gate reject a grant that is
/// older than the revoke it would otherwise undo (backlog#2291). Bounded
/// by [`SITE_REPLICATION_IAM_DELETION_MARK_LIMIT`]; the oldest mark is
/// evicted first.
#[serde(default, with = "rfc3339_map", skip_serializing_if = "BTreeMap::is_empty")]
pub(crate) iam_deletion_marks: BTreeMap<String, OffsetDateTime>,
}
/// Upper bound on [`SiteReplicationState::iam_deletion_marks`].
pub(crate) const SITE_REPLICATION_IAM_DELETION_MARK_LIMIT: usize = 1024;
/// Record that deletions of `entities` with source timestamp `deleted_at`
/// were committed here. Newest wins per entity: an older deletion never
/// lowers a mark. Returns whether the state changed.
pub(crate) fn record_iam_deletion_marks(
state: &mut SiteReplicationState,
entities: &[String],
deleted_at: OffsetDateTime,
) -> bool {
let mut changed = false;
for entity in entities {
if state
.iam_deletion_marks
.get(entity)
.is_some_and(|existing| *existing >= deleted_at)
{
continue;
}
state.iam_deletion_marks.insert(entity.clone(), deleted_at);
changed = true;
}
while state.iam_deletion_marks.len() > SITE_REPLICATION_IAM_DELETION_MARK_LIMIT {
let Some(oldest) = state
.iam_deletion_marks
.iter()
.min_by_key(|(_, deleted_at)| **deleted_at)
.map(|(entity, _)| entity.clone())
else {
break;
};
state.iam_deletion_marks.remove(&oldest);
}
changed
}
/// Newest deletion mark among `entities`, or `None` when no deletion of any
/// of them was recorded here. The receive-side staleness gate feeds this in
/// as the local timestamp when the targeted record is absent.
pub(crate) fn iam_deletion_mark(state: &SiteReplicationState, entities: &[String]) -> Option<OffsetDateTime> {
entities
.iter()
.filter_map(|entity| state.iam_deletion_marks.get(entity).copied())
.max()
}
/// RFC 3339 map values, matching the other timestamps in the state object
/// (`time::serde::rfc3339` only applies to a single field).
mod rfc3339_map {
use serde::{Deserialize, Deserializer, Serialize, Serializer};
use std::collections::BTreeMap;
use time::OffsetDateTime;
#[derive(Serialize, Deserialize)]
#[serde(transparent)]
struct Stamp(#[serde(with = "time::serde::rfc3339")] OffsetDateTime);
pub(super) fn serialize<S: Serializer>(map: &BTreeMap<String, OffsetDateTime>, serializer: S) -> Result<S::Ok, S::Error> {
serializer.collect_map(map.iter().map(|(entity, deleted_at)| (entity, Stamp(*deleted_at))))
}
pub(super) fn deserialize<'de, D: Deserializer<'de>>(deserializer: D) -> Result<BTreeMap<String, OffsetDateTime>, D::Error> {
let map = BTreeMap::<String, Stamp>::deserialize(deserializer)?;
Ok(map
.into_iter()
.map(|(entity, Stamp(deleted_at))| (entity, deleted_at))
.collect())
}
}
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
+5 -465
View File
@@ -554,121 +554,6 @@ fn test_iam_item_deletion_entity_shapes() {
assert!(iam_item_deletion_entity(&policy_set).is_none());
}
/// Deletion marks (backlog#2291) key on the same entities as the replay
/// records, except that a group member removal is marked per member (so a
/// stale re-add of one member can be judged) and a group delete marks the
/// group itself. Creates and updates leave no mark.
#[test]
fn test_iam_item_deletion_mark_entities_shapes() {
assert_eq!(
iam_item_deletion_mark_entities(&user_delete_item("alice")),
vec!["iam-user:alice".to_string()]
);
assert_eq!(
iam_item_deletion_mark_entities(&policy_delete_item("readonly")),
vec!["policy:readonly".to_string()]
);
let mut group_remove = SRIAMItem {
r#type: "group-info".to_string(),
group_info: Some(SRGroupInfo {
update_req: GroupAddRemove {
group: "devs".to_string(),
members: vec!["bob".to_string(), "alice".to_string()],
status: GroupStatus::Enabled,
is_remove: true,
},
api_version: None,
}),
..Default::default()
};
assert_eq!(
iam_item_deletion_mark_entities(&group_remove),
vec!["group-member:devs:bob".to_string(), "group-member:devs:alice".to_string()]
);
group_remove
.group_info
.as_mut()
.expect("group info")
.update_req
.members
.clear();
assert_eq!(
iam_item_deletion_mark_entities(&group_remove),
vec!["group:devs".to_string()],
"a removal without members deletes the group"
);
group_remove.group_info.as_mut().expect("group info").update_req.is_remove = false;
assert!(iam_item_deletion_mark_entities(&group_remove).is_empty());
let mapping_clear = SRIAMItem {
r#type: "policy-mapping".to_string(),
policy_mapping: Some(SRPolicyMapping {
user_or_group: "alice".to_string(),
user_type: 0,
is_group: false,
policy: String::new(),
..Default::default()
}),
..Default::default()
};
assert_eq!(
iam_item_deletion_mark_entities(&mapping_clear),
vec!["policy-mapping:alice:0:false".to_string()]
);
let mut user_create = user_delete_item("alice");
user_create.iam_user.as_mut().expect("iam user").is_delete_req = false;
assert!(iam_item_deletion_mark_entities(&user_create).is_empty());
}
/// Newest wins per entity, the map stays bounded by evicting the oldest
/// mark, and the timestamps survive the state object as RFC 3339.
#[test]
fn test_record_iam_deletion_marks_newest_wins_and_stays_bounded() {
let at = |seconds: i64| OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(seconds);
let mut state = SiteReplicationState::default();
let alice = vec!["iam-user:alice".to_string()];
assert!(record_iam_deletion_marks(&mut state, &alice, at(20)));
assert!(
!record_iam_deletion_marks(&mut state, &alice, at(10)),
"an older deletion does not move the mark"
);
assert!(
!record_iam_deletion_marks(&mut state, &alice, at(20)),
"a replayed deletion is not a change"
);
assert_eq!(iam_deletion_mark(&state, &alice), Some(at(20)));
assert!(record_iam_deletion_marks(&mut state, &alice, at(30)));
assert_eq!(iam_deletion_mark(&state, &alice), Some(at(30)));
assert_eq!(iam_deletion_mark(&state, &["iam-user:bob".to_string()]), None);
assert!(!record_iam_deletion_marks(&mut state, &[], at(40)));
// Fill past the bound with marks older than alice's; the oldest go first.
let members: Vec<String> = (0..SITE_REPLICATION_IAM_DELETION_MARK_LIMIT)
.map(|index| format!("group-member:devs:user-{index:04}"))
.collect();
for (index, member) in members.iter().enumerate() {
record_iam_deletion_marks(&mut state, std::slice::from_ref(member), at(index as i64 - 2000));
}
assert_eq!(state.iam_deletion_marks.len(), SITE_REPLICATION_IAM_DELETION_MARK_LIMIT);
assert_eq!(iam_deletion_mark(&state, &alice), Some(at(30)), "the newest mark survives eviction");
assert_eq!(iam_deletion_mark(&state, &members[..1]), None, "the oldest mark is evicted first");
assert_eq!(iam_deletion_mark(&state, &members[1..2]), Some(at(-1999)));
let json = serde_json::to_value(&state).expect("serialize state");
assert_eq!(json["iam_deletion_marks"]["iam-user:alice"], serde_json::json!("1970-01-01T00:00:30Z"));
let reloaded = parse_site_replication_state(&serde_json::to_vec(&state).expect("serialize state")).expect("parse state");
assert_eq!(reloaded.iam_deletion_marks, state.iam_deletion_marks);
assert!(
parse_site_replication_state(br#"{"name":"a","service_account_access_key":"","service_account_parent":"","peers":{},"updated_at":null,"resync_status":{}}"#)
.expect("state without marks")
.iam_deletion_marks
.is_empty()
);
}
/// A failed deletion delivery persists a replay record next to the collapsed
/// retry entry; a fresh entry is stamped `deletions_recorded` so a later
/// replay can settle it, and a repeated deletion of the same entity keeps the
@@ -1794,8 +1679,7 @@ fn test_site_replication_bootstrap_plan_includes_replayable_snapshot_items() {
},
);
let plan =
site_replication_bootstrap_plan(&info, &SiteReplicationIamCredentials::default()).expect("bootstrap plan should build");
let plan = site_replication_bootstrap_plan(&info).expect("bootstrap plan should build");
assert_eq!(plan.iam_items.iter().map(|item| item.r#type.as_str()).collect::<Vec<_>>(), {
vec!["policy", "iam-user", "group-info", "policy-mapping"]
@@ -1833,8 +1717,7 @@ fn test_site_replication_bootstrap_plan_skips_lifecycle_by_default() {
},
);
let plan =
site_replication_bootstrap_plan(&info, &SiteReplicationIamCredentials::default()).expect("bootstrap plan should build");
let plan = site_replication_bootstrap_plan(&info).expect("bootstrap plan should build");
assert!(!plan.bucket_items.iter().any(|item| item.r#type == "lc-config"));
}
@@ -1865,8 +1748,7 @@ fn test_site_replication_bootstrap_plan_emits_timestamped_lifecycle_delete() {
},
);
let plan =
site_replication_bootstrap_plan(&info, &SiteReplicationIamCredentials::default()).expect("bootstrap plan should build");
let plan = site_replication_bootstrap_plan(&info).expect("bootstrap plan should build");
let item = plan
.bucket_items
@@ -2053,8 +1935,8 @@ fn test_site_replication_repair_preflight_token_is_deterministic_for_equal_state
},
);
let plan_a = site_replication_bootstrap_plan(&info, &SiteReplicationIamCredentials::default()).expect("first plan");
let plan_b = site_replication_bootstrap_plan(&info, &SiteReplicationIamCredentials::default()).expect("second plan");
let plan_a = site_replication_bootstrap_plan(&info).expect("first plan");
let plan_b = site_replication_bootstrap_plan(&info).expect("second plan");
let token_a = site_replication_repair_preflight_token(&state, &plan_a, b"test-signing-key").expect("first token");
let token_b = site_replication_repair_preflight_token(&state, &plan_b, b"test-signing-key").expect("second token");
@@ -3337,345 +3219,3 @@ fn test_reconcile_adds_missing_peer_rules_to_existing_config() {
assert!(rule_ids.contains(&"site-repl-dep-b"));
assert!(rule_ids.contains(&"site-repl-dep-c"));
}
/// backlog#2289: the IAM snapshot (retry resend, repair, site-add bootstrap)
/// used to be built from `list_users`, whose `UserInfo` never carries a
/// secret key, so the plan dropped every user and a status change or secret
/// rotation committed while a peer was unreachable never reached it. The
/// credentials now come from a separate store read; SRInfo stays secret-free.
#[test]
fn test_bootstrap_plan_carries_users_from_the_credential_snapshot() {
let mut info = SRInfo::default();
// Exactly what `list_users` builds: status, policy, updated_at — never secret_key.
info.user_info_map.insert(
"alice".to_string(),
rustfs_madmin::UserInfo {
status: rustfs_madmin::AccountStatus::Disabled,
policy_name: Some("readwrite".to_string()),
updated_at: Some(OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("timestamp")),
..Default::default()
},
);
info.user_info_map.insert(
"external-idp-user".to_string(),
rustfs_madmin::UserInfo {
status: rustfs_madmin::AccountStatus::Enabled,
..Default::default()
},
);
let user_updated_at = OffsetDateTime::from_unix_timestamp(1_700_000_500).expect("timestamp");
let mut credentials = SiteReplicationIamCredentials::default();
credentials.users.insert(
"alice".to_string(),
SiteReplicationUserCredential {
secret_key: "alice-secret".to_string(),
status: rustfs_madmin::AccountStatus::Disabled,
updated_at: Some(user_updated_at),
},
);
let plan = site_replication_bootstrap_plan(&info, &credentials).expect("bootstrap plan should build");
let users: Vec<_> = plan.iam_items.iter().filter(|item| item.r#type == "iam-user").collect();
assert_eq!(users.len(), 1, "only the user with a credential travels: {:?}", plan.iam_items);
let alice = users[0].iam_user.as_ref().expect("iam user body");
assert_eq!(alice.access_key, "alice");
let req = alice.user_req.as_ref().expect("user request");
assert_eq!(req.secret_key, "alice-secret");
assert_eq!(req.status, rustfs_madmin::AccountStatus::Disabled);
assert_eq!(req.policy.as_deref(), Some("readwrite"));
// the user record's own axis, not the policy-mapping time list_users reports
assert_eq!(users[0].updated_at, Some(user_updated_at));
}
fn service_account_snapshot(access_key: &str, parent: &str, status: &str) -> SiteReplicationServiceAccountSnapshot {
SiteReplicationServiceAccountSnapshot {
create: rustfs_madmin::SRSvcAccCreate {
parent: parent.to_string(),
access_key: access_key.to_string(),
secret_key: format!("{access_key}-secret"),
groups: Vec::new(),
claims: HashMap::new(),
session_policy: SRSessionPolicy::default(),
status: status.to_string(),
name: String::new(),
description: String::new(),
expiration: None,
api_version: Some(SITE_REPL_API_VERSION.to_string()),
},
envelope: None,
updated_at: Some(OffsetDateTime::from_unix_timestamp(1_700_000_600).expect("timestamp")),
}
}
/// backlog#2289: service accounts were absent from every snapshot (the
/// listing filters them). They now travel as the create item the live hook
/// emits — after their parents — carrying secret and status.
#[test]
fn test_bootstrap_plan_emits_service_accounts_after_their_parents() {
let mut info = SRInfo::default();
info.user_info_map
.insert("alice".to_string(), rustfs_madmin::UserInfo::default());
let mut credentials = SiteReplicationIamCredentials::default();
credentials.users.insert(
"alice".to_string(),
SiteReplicationUserCredential {
secret_key: "alice-secret".to_string(),
status: rustfs_madmin::AccountStatus::Enabled,
updated_at: None,
},
);
credentials
.service_accounts
.push(service_account_snapshot("alice-svc", "alice", "off"));
let plan = site_replication_bootstrap_plan(&info, &credentials).expect("bootstrap plan should build");
let types: Vec<_> = plan.iam_items.iter().map(|item| item.r#type.as_str()).collect();
assert_eq!(types, vec!["iam-user", "service-account"]);
let change = plan.iam_items[1].svc_acc_change.as_ref().expect("service account change");
let create = change.create.as_ref().expect("create body");
assert_eq!((create.access_key.as_str(), create.parent.as_str()), ("alice-svc", "alice"));
assert_eq!(create.secret_key, "alice-svc-secret");
assert_eq!(create.status, "off", "a disabled account must arrive disabled");
assert!(change.delete.is_none() && change.update.is_none());
}
/// A service account present in the previous snapshot but gone from the
/// fresh one is replayed as an explicit delete, like the other IAM kinds.
#[test]
fn test_retry_snapshot_tombstones_removed_service_accounts() {
let observed_at = OffsetDateTime::from_unix_timestamp(1_700_001_000).expect("timestamp");
let mut info = SRInfo::default();
info.user_info_map
.insert("alice".to_string(), rustfs_madmin::UserInfo::default());
let mut credentials = SiteReplicationIamCredentials::default();
credentials.users.insert(
"alice".to_string(),
SiteReplicationUserCredential {
secret_key: "alice-secret".to_string(),
status: rustfs_madmin::AccountStatus::Enabled,
updated_at: None,
},
);
let mut with_account = credentials.clone();
with_account
.service_accounts
.push(service_account_snapshot("alice-svc", "alice", "on"));
let previous = site_replication_bootstrap_plan(&info, &with_account).expect("previous plan");
let fresh = site_replication_bootstrap_plan(&info, &credentials).expect("fresh plan");
let replay = RetrySnapshot::replay_after_change(
&RetrySnapshot::Iam(previous.iam_items),
&RetrySnapshot::Iam(fresh.iam_items),
observed_at,
);
let RetrySnapshot::Iam(items) = replay else {
panic!("IAM snapshot expected");
};
let tombstone = items
.iter()
.find(|item| item.r#type == "service-account")
.expect("service account tombstone");
let change = tombstone.svc_acc_change.as_ref().expect("change");
assert_eq!(change.delete.as_ref().map(|delete| delete.access_key.as_str()), Some("alice-svc"));
assert!(change.create.is_none());
assert_eq!(tombstone.updated_at, Some(observed_at));
}
/// Spawns a one-shot HTTP peer that answers 200 and flips the returned flag
/// once a request head has arrived.
async fn spawn_reached_probe_peer() -> (String, Arc<AtomicBool>, tokio::task::JoinHandle<()>) {
let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind healthy peer");
let endpoint = format!("http://{}", listener.local_addr().expect("healthy peer address"));
let reached = Arc::new(AtomicBool::new(false));
let reached_by_server = reached.clone();
let server = tokio::spawn(async move {
let Ok((mut stream, _)) = listener.accept().await else {
return;
};
let mut request = Vec::new();
let mut buffer = [0_u8; 1024];
loop {
let Ok(read) = stream.read(&mut buffer).await else {
return;
};
if read == 0 {
return;
}
request.extend_from_slice(&buffer[..read]);
if request.windows(4).any(|window| window == b"\r\n\r\n") {
break;
}
}
reached_by_server.store(true, Ordering::SeqCst);
let _ = stream
.write_all(b"HTTP/1.1 200 OK\r\ncontent-length: 2\r\nconnection: close\r\n\r\nok")
.await;
});
(endpoint, reached, server)
}
/// Three-peer runtime whose local peer is `local`; BTreeMap order visits the
/// failing peer `b` before the healthy peer `c`.
fn broadcast_runtime_with_failing_peer_before_healthy(failing_endpoint: &str, healthy_endpoint: &str) -> SiteReplicationRuntime {
let local_peer = PeerInfo {
deployment_id: "local".to_string(),
..peer("local", "http://127.0.0.1:9")
};
let mut state = SiteReplicationState {
name: "local".to_string(),
service_account_access_key: "site-replicator-0".to_string(),
..Default::default()
};
state.peers.insert("local".to_string(), local_peer.clone());
state.peers.insert(
"b".to_string(),
PeerInfo {
deployment_id: "b".to_string(),
..peer("b", failing_endpoint)
},
);
state.peers.insert(
"c".to_string(),
PeerInfo {
deployment_id: "c".to_string(),
..peer("c", healthy_endpoint)
},
);
SiteReplicationRuntime {
state,
local_peer,
service_account_secret_key: "site-replicator-secret".to_string(),
}
}
const BROADCAST_PROBE_DELETE_BUCKET_PATH: &str =
"/rustfs/admin/v3/site-replication/peer/bucket-ops?bucket=photos&operation=delete-bucket";
/// The generic JSON broadcast (bucket make/delete, bucket-meta hook, bucket
/// ops) attempts every remote peer: a peer whose request fails must not stop
/// delivery to the peers that follow it in deployment-id order, and the
/// failure is still reported to the caller (backlog#2293).
#[tokio::test]
#[serial]
async fn test_broadcast_json_reaches_healthy_peers_after_a_failed_peer() {
// Peer "b": nothing listens on the port, so the connect is refused.
let refused = TcpListener::bind("127.0.0.1:0").await.expect("bind refused-peer probe");
let refused_endpoint = format!("http://{}", refused.local_addr().expect("refused-peer address"));
drop(refused);
let (healthy_endpoint, reached, server) = spawn_reached_probe_peer().await;
let runtime = broadcast_runtime_with_failing_peer_before_healthy(&refused_endpoint, &healthy_endpoint);
let result = temp_env::async_with_vars([(ALLOW_LOOPBACK_REPLICATION_TARGET_ENV, Some("true"))], async {
broadcast_site_replication_json_with_runtime(&runtime, BROADCAST_PROBE_DELETE_BUCKET_PATH, &serde_json::json!({})).await
})
.await;
let err = result.expect_err("peer b refuses connections, the broadcast must report it");
assert!(
reached.load(Ordering::SeqCst),
"peer c never received the broadcast once peer b failed: {err}"
);
server.abort();
}
/// Same guarantee when the failing peer never gets a transport: an endpoint
/// that `PeerTransport::for_runtime_peer` rejects must be skipped past (and
/// reported), not abort the broadcast before the healthy peers (backlog#2293).
#[tokio::test]
#[serial]
async fn test_broadcast_json_reaches_healthy_peers_after_a_peer_without_transport() {
// Peer "b": a scheme the peer connection validator refuses outright.
let forbidden_endpoint = "ftp://peer-b.example.com";
let (healthy_endpoint, reached, server) = spawn_reached_probe_peer().await;
let runtime = broadcast_runtime_with_failing_peer_before_healthy(forbidden_endpoint, &healthy_endpoint);
let result = temp_env::async_with_vars([(ALLOW_LOOPBACK_REPLICATION_TARGET_ENV, Some("true"))], async {
broadcast_site_replication_json_with_runtime(&runtime, BROADCAST_PROBE_DELETE_BUCKET_PATH, &serde_json::json!({})).await
})
.await;
let err = result.expect_err("peer b has no usable transport, the broadcast must report it");
assert!(
err.to_string().contains("invalid persisted site replication peer"),
"the reported error must be peer b's transport failure: {err}"
);
assert!(
reached.load(Ordering::SeqCst),
"peer c never received the broadcast once peer b failed to get a transport: {err}"
);
server.abort();
}
fn service_account_item_with_claims(order: &[&str]) -> SRIAMItem {
let mut claims = HashMap::new();
for key in order {
claims.insert((*key).to_string(), serde_json::json!(format!("value-of-{key}")));
}
SRIAMItem {
r#type: "service-account".to_string(),
svc_acc_change: Some(SRSvcAccChange {
create: Some(rustfs_madmin::SRSvcAccCreate {
parent: "alice".to_string(),
access_key: "alice-svc".to_string(),
secret_key: "alice-svc-secret".to_string(),
groups: Vec::new(),
claims,
session_policy: SRSessionPolicy::default(),
status: "on".to_string(),
name: String::new(),
description: String::new(),
expiration: None,
api_version: Some(SITE_REPL_API_VERSION.to_string()),
}),
api_version: Some(SITE_REPL_API_VERSION.to_string()),
..Default::default()
}),
updated_at: Some(OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("timestamp")),
api_version: Some(SITE_REPL_API_VERSION.to_string()),
..Default::default()
}
}
/// The repair preflight token and the retry-snapshot fingerprint hash the
/// serialized items. Service-account claims live in a `HashMap`, whose
/// iteration order differs between instances, so the hash must not depend on
/// it (the real-VM repair returned 412 "preflight is stale" between dry-run
/// and execute once snapshots carried service accounts).
#[test]
fn test_repair_task_id_and_retry_fingerprint_ignore_claim_map_order() {
let forward = service_account_item_with_claims(&["accessKey", "exp", "parent", "sa-policy", "sub", "tenant"]);
let backward = service_account_item_with_claims(&["tenant", "sub", "sa-policy", "parent", "exp", "accessKey"]);
let canonical = canonical_json_vec(&forward).expect("canonical json");
let text = String::from_utf8(canonical).expect("utf8");
let positions: Vec<usize> = [
"\"accessKey\"",
"\"exp\"",
"\"parent\"",
"\"sa-policy\"",
"\"sub\"",
"\"tenant\"",
]
.iter()
.map(|key| text.find(key).expect("claim key present"))
.collect();
assert!(
positions.windows(2).all(|pair| pair[0] < pair[1]),
"claim keys must serialize sorted: {text}"
);
assert_eq!(
SiteReplicationRepairTask::Iam(&forward).id().expect("id"),
SiteReplicationRepairTask::Iam(&backward).id().expect("id"),
"identical items must yield the same repair task id regardless of claim map order"
);
assert_eq!(
RetrySnapshot::Iam(vec![forward]).fingerprint().expect("fingerprint"),
RetrySnapshot::Iam(vec![backward]).fingerprint().expect("fingerprint"),
"identical snapshots must fingerprint equal regardless of claim map order"
);
}
+6 -24
View File
@@ -876,14 +876,6 @@ pub(crate) async fn broadcast_site_replication_json<T: Serialize>(path: &str, bo
broadcast_site_replication_json_with_runtime(&runtime, path, body).await
}
/// PUT `body` to `path` on every remote peer of the runtime.
///
/// Every peer is attempted: one peer's failure — transport construction
/// included — must not skip the peers that follow it in deployment-id order,
/// or they silently miss the change with no retry record (backlog#2293). A
/// success settles the peer/path's queued retry event, a failure enqueues one
/// under the request `path` (so the drain classifies it as today), and the
/// first error is returned once all peers were attempted.
pub(crate) async fn broadcast_site_replication_json_with_runtime<T: Serialize>(
runtime: &SiteReplicationRuntime,
path: &str,
@@ -891,30 +883,20 @@ pub(crate) async fn broadcast_site_replication_json_with_runtime<T: Serialize>(
) -> S3Result<()> {
let state = &runtime.state;
let local_peer = &runtime.local_peer;
let mut first_error: Option<S3Error> = None;
for peer in state.peers.values() {
if peer.deployment_id == local_peer.deployment_id || same_identity_endpoint(&peer.endpoint, &local_peer.endpoint) {
continue;
}
let sent = match PeerTransport::for_runtime_peer(peer).await {
Ok(transport) => PeerAdminRequest::put(&transport.connection, path, &state.service_account_access_key)
.with_client(&transport.client)
.send_with_retry_event(peer, &runtime.service_account_secret_key, body)
.await
.map(|_| ()),
Err(err) => {
enqueue_site_replication_retry_event(peer, path, &err).await;
Err(err)
}
};
if let Err(err) = sent {
first_error.get_or_insert(err);
}
let transport = PeerTransport::for_runtime_peer(peer).await?;
PeerAdminRequest::put(&transport.connection, path, &state.service_account_access_key)
.with_client(&transport.client)
.send_with_retry_event(peer, &runtime.service_account_secret_key, body)
.await?;
}
first_error.map_or(Ok(()), Err)
Ok(())
}
pub(crate) fn parse_endpoint_refresh_status(peer: &PeerInfo, body: &[u8]) -> S3Result<()> {
+86 -1
View File
@@ -2131,9 +2131,9 @@ impl Node for NodeService {
)
.map_err(|err| Status::failed_precondition(err.to_string()))?;
}
let namespace_generation = store.scanner_namespace_mutation_generation();
let topology_digest = rustfs_scanner::scanner_topology_digest(store.as_ref());
let (data_movement_active, publication_blocked, movement_generation) = store.scanner_data_movement_activity().await;
let namespace_generation = store.scanner_namespace_mutation_generation();
let mut response = match request_protocol {
SCANNER_ACTIVITY_LEGACY_PROTOCOL_VERSION | SCANNER_ACTIVITY_PREVIOUS_PROTOCOL_VERSION => {
previous_scanner_activity_response(namespace_generation, topology_digest, data_movement_active)
@@ -6264,6 +6264,91 @@ mod tests {
assert_eq!(unavailable.code(), tonic::Code::Unavailable);
}
#[tokio::test]
async fn scanner_activity_samples_namespace_generation_after_waiting_for_movement_state() {
use crate::storage::storage_api::{ObjectOptions, PutObjReader, contract::object::ObjectIO as _};
let _ = rustfs_credentials::set_global_rpc_secret("scanner-activity-generation-test-secret".to_string());
let _ = rustfs_credentials::init_global_action_credentials(
Some("TESTROOTACCESSKEY".to_string()),
Some("TESTROOTSECRET123".to_string()),
);
let temp_dir = tempfile::tempdir().expect("scanner activity RPC test directory");
let env = rustfs_test_utils::TestECStoreEnv::builder()
.base_dir(temp_dir.path())
.build()
.await;
ObjectStore::new(Arc::clone(&env.ecstore))
.save_iam_config(serde_json::json!({"version": 1}), format!("{}/format.json", *IAM_CONFIG_PREFIX))
.await
.expect("seed IAM format");
let iam = rustfs_iam::build_iam_sys(Arc::clone(&env.ecstore))
.await
.expect("build isolated IAM");
let context = Arc::new(crate::runtime_sources::AppContext::with_default_interfaces(
Arc::clone(&env.ecstore),
iam,
Arc::new(KmsServiceManager::new()),
));
let service = make_server_for_context(Some(context));
let bucket = "scanner-activity-generation";
env.make_bucket(bucket, false).await;
let generation_before = env.ecstore.scanner_namespace_mutation_generation();
let mut request = Request::new(ScannerActivityRequest {
challenge: vec![7; 16].into(),
protocol_version: rustfs_scanner::SCANNER_ACTIVITY_PROTOCOL_VERSION,
acknowledge_instance_id: String::new(),
acknowledge_dirty_usage_generation: 0,
});
let canonical = rustfs_protos::canonical_scanner_activity_request_body(request.get_ref())
.expect("scanner activity request should encode");
set_tonic_canonical_body_digest(&mut request, &canonical).expect("digest metadata should encode");
mark_v2_authenticated(&mut request);
let pool_meta = env.ecstore.pool_meta.write().await;
drop(
env.ecstore
.decommission_cancelers
.try_write()
.expect("movement snapshot should not hold the cancelers before the RPC"),
);
let mut activity = Box::pin(tokio::task::unconstrained(service.scanner_activity(request)));
assert!(futures::poll!(activity.as_mut()).is_pending());
assert!(
env.ecstore.decommission_cancelers.try_write().is_err(),
"the RPC must hold the cancelers read guard while waiting for pool metadata"
);
// Select the existing set directly: ECStore pool selection reads the lock held by this test.
let mut reader = PutObjReader::from_vec(b"namespace changed during activity probe".to_vec());
tokio::time::timeout(
Duration::from_secs(30),
env.ecstore.pools[0].disk_set[0].put_object(
bucket,
"object",
&mut reader,
&ObjectOptions {
no_lock: true,
..Default::default()
},
),
)
.await
.expect("the namespace mutation must not wait for the RPC's pool lock")
.expect("the namespace mutation must complete while the RPC waits");
let generation_after = env.ecstore.scanner_namespace_mutation_generation();
assert!(generation_after > generation_before);
drop(pool_meta);
let response = tokio::time::timeout(Duration::from_secs(30), activity)
.await
.expect("scanner activity RPC should resume after the pool lock is released")
.expect("authenticated scanner activity RPC should succeed")
.into_inner();
assert_eq!(response.namespace_generation, generation_after);
assert_eq!(response.publication_blocked, Some(false));
}
#[tokio::test]
async fn test_scanner_dirty_usage_snapshot_requires_body_bound_auth_and_signs_a_consistent_view() {
let _ = rustfs_credentials::set_global_rpc_secret("scanner-dirty-usage-snapshot-test-secret".to_string());