From 9c6b0e182ba223b9ec4fd0de07c90eb9696fbd7e Mon Sep 17 00:00:00 2001 From: Hauser Date: Thu, 8 Oct 2026 10:04:34 +0800 Subject: [PATCH] fix: stabilize scanner leases and connect diagnostics (#8394) * fix(scanner): scope observational publication leases Separate ordinary namespace commit churn from data-movement fences and retry transient lease acquisition conflicts. Bind observational leases to the observed usage object while preserving strict authoritative publication checks. Co-Authored-By: heihutu Co-Authored-By: zhi22915 * test(ecstore): expect namespace admission conflict Assert the retryable namespace-specific result when a commit finishes while strict lease admission awaits its snapshot. Co-Authored-By: heihutu Co-Authored-By: zhi22915 * fix(connect): add headroom to real performance probes Give real filesystem and object integration probes enough bounded time on slow CI runners, and keep cleanup independent from measurement latency. Use a genuinely unwritable Linux fixture when tests run as root. Co-Authored-By: heihutu Co-Authored-By: zhi22915 --------- Co-authored-by: zhi22915 --- .../src/cluster/rpc/peer_rest_client.rs | 44 +++- crates/ecstore/src/runtime/instance.rs | 48 +++-- .../ecstore/src/services/notification_sys.rs | 21 +- crates/ecstore/src/store/mod.rs | 190 ++++++++++++++++-- crates/ecstore/src/store/peer.rs | 71 ++++++- .../src/generated/proto_gen/node_service.rs | 6 + crates/protos/src/lib.rs | 38 +++- crates/protos/src/node.proto | 4 + crates/scanner/src/scanner.rs | 59 +++++- crates/scanner/src/scanner/activity.rs | 2 + crates/scanner/src/scanner/tests.rs | 101 +++++++++- crates/scanner/src/storage_api.rs | 9 + rustfs/src/connect/diagnostics/perf_object.rs | 2 +- rustfs/src/storage/rpc/node_service.rs | 9 +- rustfs/tests/connect_perf_drive.rs | 64 ++++-- rustfs/tests/connect_perf_object.rs | 6 +- 16 files changed, 591 insertions(+), 83 deletions(-) diff --git a/crates/ecstore/src/cluster/rpc/peer_rest_client.rs b/crates/ecstore/src/cluster/rpc/peer_rest_client.rs index 6e0c548de..eb9a023f8 100644 --- a/crates/ecstore/src/cluster/rpc/peer_rest_client.rs +++ b/crates/ecstore/src/cluster/rpc/peer_rest_client.rs @@ -640,6 +640,7 @@ pub struct PeerLiveEventsBatch { pub struct ScannerPublicationLease { pub token: Uuid, pub movement_generation: u64, + pub observational_only: bool, /// Stable storage owner identity. This is distinct from the activity /// session and is bound into both acquire and release proofs. pub owner_id: String, @@ -658,6 +659,7 @@ fn validate_scanner_publication_lease_response_fields( response: &ScannerPublicationLeaseResponse, expected_session_id: &str, expected_generation: u64, + expected_observational_only: bool, ) -> Result<(Uuid, String)> { if !response.success { return Err(Error::other( @@ -674,6 +676,9 @@ fn validate_scanner_publication_lease_response_fields( if response.session_id != expected_session_id { return Err(Error::other("peer returned a different scanner publication lease session")); } + if response.observational_only != expected_observational_only { + return Err(Error::other("peer returned a different scanner publication lease purpose")); + } let owner_id = Uuid::parse_str(&response.owner_id) .ok() .filter(|owner_id| !owner_id.is_nil()) @@ -2418,6 +2423,7 @@ impl PeerRestClient { &self, expected_session_id: &str, expected_generation: u64, + observational_only: bool, ) -> Result { let request_started = std::time::Instant::now(); self.finalize_result( @@ -2434,6 +2440,7 @@ impl PeerRestClient { ttl_ms: crate::store::SCANNER_PUBLICATION_LEASE_TTL_MS, expected_session_id: expected_session_id.to_string(), token: Bytes::new(), + observational_only, }); let canonical = rustfs_protos::canonical_scanner_publication_lease_request_body(request.get_ref()) .map_err(|_| Error::other("scanner publication lease request is too large to authenticate"))?; @@ -2444,11 +2451,16 @@ impl PeerRestClient { .map_err(|_| Error::other("scanner publication lease response is too large to authenticate"))?; verify_tonic_rpc_response_proof(&response_body, &response.response_proof) .map_err(|_| Error::other("peer returned an invalid scanner publication lease proof"))?; - let (token, owner_id) = - validate_scanner_publication_lease_response_fields(&response, expected_session_id, expected_generation)?; + let (token, owner_id) = validate_scanner_publication_lease_response_fields( + &response, + expected_session_id, + expected_generation, + observational_only, + )?; Ok(ScannerPublicationLease { token, movement_generation: response.movement_generation, + observational_only, owner_id, session_id: response.session_id, expires_at: scanner_publication_lease_deadline( @@ -2483,6 +2495,7 @@ impl PeerRestClient { ttl_ms: crate::store::SCANNER_PUBLICATION_LEASE_TTL_MS, expected_session_id: lease.session_id.clone(), token: lease.token.as_bytes().to_vec().into(), + observational_only: lease.observational_only, }); let canonical = rustfs_protos::canonical_scanner_publication_lease_request_body(request.get_ref()) .map_err(|_| Error::other("scanner publication lease validation request is too large to authenticate"))?; @@ -2494,8 +2507,12 @@ impl PeerRestClient { )?; verify_tonic_rpc_response_proof(&response_body, &response.response_proof) .map_err(|_| Error::other("peer returned an invalid scanner publication lease validation proof"))?; - let (token, owner_id) = - validate_scanner_publication_lease_response_fields(&response, &lease.session_id, lease.movement_generation)?; + let (token, owner_id) = validate_scanner_publication_lease_response_fields( + &response, + &lease.session_id, + lease.movement_generation, + lease.observational_only, + )?; if token != lease.token { return Err(Error::other("peer returned a different scanner publication lease token")); } @@ -3001,15 +3018,16 @@ mod tests { response_proof: Bytes::new(), owner_id: Uuid::new_v4().to_string(), session_id: "session-a".to_string(), + observational_only: false, }; - assert!(validate_scanner_publication_lease_response_fields(&response, "session-a", 7).is_ok()); + assert!(validate_scanner_publication_lease_response_fields(&response, "session-a", 7, false).is_ok()); let stale_generation = ScannerPublicationLeaseResponse { movement_generation: 6, ..response.clone() }; - let error = validate_scanner_publication_lease_response_fields(&stale_generation, "session-a", 7) + let error = validate_scanner_publication_lease_response_fields(&stale_generation, "session-a", 7, false) .expect_err("a response from an older movement generation must be rejected"); assert!(error.to_string().contains("different scanner publication lease generation")); @@ -3017,9 +3035,21 @@ mod tests { session_id: "session-b".to_string(), ..response }; - let error = validate_scanner_publication_lease_response_fields(&stale_session, "session-a", 7) + let error = validate_scanner_publication_lease_response_fields(&stale_session, "session-a", 7, false) .expect_err("a response from an older scanner session must be rejected"); assert!(error.to_string().contains("different scanner publication lease session")); + + let observational = ScannerPublicationLeaseResponse { + observational_only: true, + ..stale_session + }; + assert!( + validate_scanner_publication_lease_response_fields(&observational, "session-b", 7, true).is_ok(), + "the signed purpose echo should validate an observation lease" + ); + let error = validate_scanner_publication_lease_response_fields(&observational, "session-b", 7, false) + .expect_err("a peer must not upgrade an observational lease to authoritative"); + assert!(error.to_string().contains("different scanner publication lease purpose")); } #[test] diff --git a/crates/ecstore/src/runtime/instance.rs b/crates/ecstore/src/runtime/instance.rs index 7e187c3a0..4c6193c43 100644 --- a/crates/ecstore/src/runtime/instance.rs +++ b/crates/ecstore/src/runtime/instance.rs @@ -72,10 +72,19 @@ pub(crate) const SCANNER_PUBLICATION_LEASE_MAX_ENTRIES: usize = 256; /// have started movement after the lease was abandoned. pub(crate) const SCANNER_PUBLICATION_LEASE_TTL: std::time::Duration = std::time::Duration::from_secs(60); +/// Observational leases hold data-movement fencing but may coexist with +/// ordinary namespace commits; authoritative leases require a stable namespace. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) enum ScannerPublicationLeasePurpose { + Authoritative, + Observational, +} + pub(crate) struct ScannerPublicationLeaseEntry { pub(crate) expires_at: Instant, pub(crate) movement_generation: u64, - pub(crate) namespace_generation: u64, + pub(crate) namespace_generation: Option, + pub(crate) purpose: ScannerPublicationLeasePurpose, pub(crate) _operation_guard: OwnedRwLockReadGuard<()>, } @@ -306,7 +315,8 @@ impl InstanceContext { token: Uuid, expires_at: Instant, movement_generation: u64, - namespace_generation: u64, + namespace_generation: Option, + purpose: ScannerPublicationLeasePurpose, operation_guard: OwnedRwLockReadGuard<()>, ) -> bool { let mut leases = self.scanner_publication_leases.lock().await; @@ -319,6 +329,7 @@ impl InstanceContext { expires_at, movement_generation, namespace_generation, + purpose, _operation_guard: operation_guard, }, ); @@ -329,21 +340,24 @@ impl InstanceContext { self.scanner_publication_leases.lock().await.remove(&token).is_some() } - /// Return both generations from the same live lease while the caller holds - /// the movement read guard. Namespace commits do not take that guard, so - /// the caller must compare the saved namespace generation after this await. + /// Return the generations and purpose from the same live lease while the + /// caller holds the movement read guard. Namespace commits do not take that + /// guard, so strict leases compare their saved generation after this await. /// The process-owned table rejects tokens from a prior instance or expiry. - pub(crate) async fn scanner_publication_lease_generations(&self, token: Uuid) -> Option<(u64, u64)> { + pub(crate) async fn scanner_publication_lease_generations( + &self, + token: Uuid, + ) -> Option<(u64, Option, ScannerPublicationLeasePurpose)> { let mut leases = self.scanner_publication_leases.lock().await; let now = Instant::now(); - let (expires_at, movement_generation, namespace_generation) = leases + let (expires_at, movement_generation, namespace_generation, purpose) = leases .get(&token) - .map(|entry| (entry.expires_at, entry.movement_generation, entry.namespace_generation))?; + .map(|entry| (entry.expires_at, entry.movement_generation, entry.namespace_generation, entry.purpose))?; if expires_at <= now { leases.remove(&token); return None; } - Some((movement_generation, namespace_generation)) + Some((movement_generation, namespace_generation, purpose)) } pub(crate) async fn expire_scanner_publication_lease(&self, token: Uuid, expires_at: Instant) { @@ -859,12 +873,22 @@ mod tests { let gate = ctx.data_movement_operation_gate(); let permit = gate.clone().read_owned().await; assert!( - ctx.install_scanner_publication_lease(token, Instant::now() + SCANNER_PUBLICATION_LEASE_TTL, 7, 11, permit) - .await + ctx.install_scanner_publication_lease( + token, + Instant::now() + SCANNER_PUBLICATION_LEASE_TTL, + 7, + Some(11), + ScannerPublicationLeasePurpose::Authoritative, + permit, + ) + .await ); drop(ctx.begin_namespace_commit()); assert_eq!(ctx.namespace_commit_generation(), 2); - assert_eq!(ctx.scanner_publication_lease_generations(token).await, Some((7, 11))); + assert_eq!( + ctx.scanner_publication_lease_generations(token).await, + Some((7, Some(11), ScannerPublicationLeasePurpose::Authoritative)) + ); assert!(gate.clone().try_write_owned().is_err(), "lookup must retain the stored permit"); tokio::time::advance(SCANNER_PUBLICATION_LEASE_TTL).await; assert_eq!(ctx.scanner_publication_lease_generations(token).await, None); diff --git a/crates/ecstore/src/services/notification_sys.rs b/crates/ecstore/src/services/notification_sys.rs index f8d177c80..655caa47e 100644 --- a/crates/ecstore/src/services/notification_sys.rs +++ b/crates/ecstore/src/services/notification_sys.rs @@ -2888,8 +2888,23 @@ impl NotificationSys { /// leases are released before returning so a partial acquisition cannot /// pin movement on one peer. pub async fn acquire_scanner_publication_leases( + &self, + targets: Vec<(String, String, u64)>, + ) -> Result> { + self.acquire_scanner_publication_leases_with_purpose(targets, false).await + } + + pub async fn acquire_observational_scanner_publication_leases( + &self, + targets: Vec<(String, String, u64)>, + ) -> Result> { + self.acquire_scanner_publication_leases_with_purpose(targets, true).await + } + + async fn acquire_scanner_publication_leases_with_purpose( &self, mut targets: Vec<(String, String, u64)>, + observational_only: bool, ) -> Result> { targets.sort_by(|left, right| left.0.cmp(&right.0)); for pair in targets.windows(2) { @@ -2910,7 +2925,10 @@ impl NotificationSys { let _ = self.release_scanner_publication_leases(grants).await; return Err(Error::other(format!("scanner publication lease peer {host} is unavailable"))); }; - match client.acquire_scanner_publication_lease(&session_id, generation).await { + match client + .acquire_scanner_publication_lease(&session_id, generation, observational_only) + .await + { Ok(lease) => grants.push(ScannerPublicationLeaseGrant { host, lease }), Err(err) => { let _ = self.release_scanner_publication_leases(grants).await; @@ -4949,6 +4967,7 @@ mod tests { lease: ScannerPublicationLease { token: Uuid::new_v4(), movement_generation: 3, + observational_only: false, owner_id: Uuid::new_v4().to_string(), session_id: "session-a".to_string(), expires_at: Instant::now() + Duration::from_secs(30), diff --git a/crates/ecstore/src/store/mod.rs b/crates/ecstore/src/store/mod.rs index 7ae4d2571..008d5f01b 100644 --- a/crates/ecstore/src/store/mod.rs +++ b/crates/ecstore/src/store/mod.rs @@ -42,7 +42,7 @@ use crate::error::{ is_err_strict_volume_not_found, is_err_version_not_found, to_object_err, }; use crate::runtime::global::DISK_RESERVE_FRACTION; -use crate::runtime::instance::InstanceContext; +use crate::runtime::instance::{InstanceContext, ScannerPublicationLeasePurpose}; use crate::runtime::sources as runtime_sources; use crate::services::rebalance::{RebalStatus, RebalanceMeta, is_rebalance_conflicting_with_decommission}; use crate::storage_api_contracts::{ @@ -1098,36 +1098,67 @@ impl ECStore { &self, expected_generation: u64, ttl: std::time::Duration, + ) -> Result<(Uuid, u64)> { + self.acquire_scanner_publication_lease_with_purpose(expected_generation, ttl, false) + .await + } + + pub async fn acquire_scanner_publication_lease_with_purpose( + &self, + expected_generation: u64, + ttl: std::time::Duration, + observational_only: bool, ) -> Result<(Uuid, u64)> { if ttl != crate::runtime::instance::SCANNER_PUBLICATION_LEASE_TTL { return Err(Error::other("scanner publication lease TTL is not supported")); } - // Bind the original activity generation across the asynchronous checks; - // a completed namespace commit must never refresh an existing proof. - let namespace_generation = self.scanner_namespace_mutation_generation(); + let purpose = if observational_only { + ScannerPublicationLeasePurpose::Observational + } else { + ScannerPublicationLeasePurpose::Authoritative + }; + // Strict leases bind the namespace generation before their async + // admission checks. An observation only needs the movement fence, but + // still fails closed if generation tracking is exhausted. + let observed_namespace_generation = self.scanner_namespace_mutation_generation(); let operation_gate = self.ctx.data_movement_operation_gate(); let operation_guard = operation_gate.read_owned().await; if self.ctx.data_movement_generation_exhausted() || self.ctx.data_movement_operation_epoch_exhausted() || self.ctx.data_movement_generation() != expected_generation - || namespace_generation == u64::MAX + || observed_namespace_generation == u64::MAX { return Err(Error::other("scanner publication lease generation is stale")); } - if self.scanner_data_movement_snapshot_locked().await.1 || self.ctx.namespace_commits_pending() { + if self.scanner_data_movement_snapshot_locked().await.1 { return Err(Error::other("scanner publication lease is blocked by data movement")); } - if self.scanner_namespace_mutation_generation() != namespace_generation { - return Err(Error::other("scanner publication lease generation is stale")); - } + let namespace_generation = if observational_only { + None + } else { + if self.ctx.namespace_commits_pending() { + return Err(Error::other("scanner publication lease is blocked by pending namespace commit")); + } + if self.scanner_namespace_mutation_generation() != observed_namespace_generation { + return Err(Error::other("scanner publication lease namespace changed during acquisition")); + } + Some(observed_namespace_generation) + }; let token = Uuid::new_v4(); let expires_at = tokio::time::Instant::now() + ttl; if !self .ctx - .install_scanner_publication_lease(token, expires_at, expected_generation, namespace_generation, operation_guard) + .install_scanner_publication_lease( + token, + expires_at, + expected_generation, + namespace_generation, + purpose, + operation_guard, + ) .await { return Err(Error::other("scanner publication lease capacity is exhausted")); @@ -1151,24 +1182,46 @@ impl ECStore { /// admission; a restarted context has no old token and therefore fails /// closed even if its generation counter has returned to zero. pub async fn validate_scanner_publication_lease(&self, token: Uuid, expected_generation: u64) -> Result<()> { - let _operation_guard = self.acquire_scanner_publication_lease_guard(token).await?; + self.validate_scanner_publication_lease_with_purpose(token, expected_generation, false) + .await + } + + pub async fn validate_scanner_publication_lease_with_purpose( + &self, + token: Uuid, + expected_generation: u64, + observational_only: bool, + ) -> Result<()> { + let expected_purpose = if observational_only { + ScannerPublicationLeasePurpose::Observational + } else { + ScannerPublicationLeasePurpose::Authoritative + }; + let _operation_guard = self + .acquire_scanner_publication_lease_guard_with_target(token, None, true, Some(expected_purpose)) + .await?; if self.ctx.data_movement_generation_exhausted() || self.ctx.data_movement_operation_epoch_exhausted() || self.ctx.data_movement_generation() != expected_generation { return Err(Error::other("scanner publication lease generation is stale")); } - if self.scanner_data_movement_snapshot_locked().await.1 || self.ctx.namespace_commits_pending() { + if self.scanner_data_movement_snapshot_locked().await.1 { return Err(Error::other("scanner publication lease is blocked by data movement")); } - let Some((lease_generation, lease_namespace_generation)) = self.ctx.scanner_publication_lease_generations(token).await + let Some((lease_generation, lease_namespace_generation, purpose)) = + self.ctx.scanner_publication_lease_generations(token).await else { return Err(Error::other("scanner publication lease is unknown or expired")); }; + if purpose == ScannerPublicationLeasePurpose::Authoritative && self.ctx.namespace_commits_pending() { + return Err(Error::other("scanner publication lease is blocked by pending namespace commit")); + } let namespace_generation = self.scanner_namespace_mutation_generation(); if lease_generation != self.ctx.data_movement_generation() - || lease_namespace_generation != namespace_generation || namespace_generation == u64::MAX + || (purpose == ScannerPublicationLeasePurpose::Authoritative + && lease_namespace_generation != Some(namespace_generation)) { return Err(Error::other("scanner publication lease generation is stale")); } @@ -1181,22 +1234,65 @@ impl ECStore { /// restart, expiry, generation changes, or blocked movement all reject it /// before the target disk is touched. pub async fn acquire_scanner_publication_lease_guard(&self, token: Uuid) -> Result> { + self.acquire_scanner_publication_lease_guard_with_target( + token, + None, + false, + Some(ScannerPublicationLeasePurpose::Authoritative), + ) + .await + } + + pub async fn acquire_scanner_publication_lease_guard_for_rename( + &self, + token: Uuid, + volume: &str, + path: &str, + ) -> Result> { + let target = format!( + "{}/{}", + crate::disk::BUCKET_META_PREFIX, + rustfs_data_usage::DATA_USAGE_OBSERVED_OBJECT_NAME + ); + let target = (volume == RUSTFS_META_BUCKET && path == target).then_some((volume, path)); + self.acquire_scanner_publication_lease_guard_with_target(token, target, false, None) + .await + } + + async fn acquire_scanner_publication_lease_guard_with_target( + &self, + token: Uuid, + observed_target: Option<(&str, &str)>, + validation_only: bool, + expected_purpose: Option, + ) -> Result> { let operation_gate = self.ctx.data_movement_operation_gate(); let operation_guard = operation_gate.read_owned().await; 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 || self.ctx.namespace_commits_pending() { + if self.scanner_data_movement_snapshot_locked().await.1 { return Err(Error::other("scanner publication lease is blocked by data movement")); } - let Some((lease_generation, lease_namespace_generation)) = self.ctx.scanner_publication_lease_generations(token).await + let Some((lease_generation, lease_namespace_generation, purpose)) = + self.ctx.scanner_publication_lease_generations(token).await else { return Err(Error::other("scanner publication lease is unknown or expired")); }; + if expected_purpose.is_some_and(|expected| expected != purpose) { + return Err(Error::other("scanner publication lease purpose does not match its request")); + } + if purpose == ScannerPublicationLeasePurpose::Observational && observed_target.is_none() && !validation_only { + return Err(Error::other("observational lease is limited to observed usage publication")); + } + if purpose == ScannerPublicationLeasePurpose::Authoritative && self.ctx.namespace_commits_pending() { + return Err(Error::other("scanner publication lease is blocked by pending namespace commit")); + } let namespace_generation = self.scanner_namespace_mutation_generation(); if lease_generation != self.ctx.data_movement_generation() - || lease_namespace_generation != namespace_generation || namespace_generation == u64::MAX + || (purpose == ScannerPublicationLeasePurpose::Authoritative + && lease_namespace_generation != Some(namespace_generation)) { return Err(Error::other("scanner publication lease generation is stale")); } @@ -2571,6 +2667,59 @@ mod tests { assert!(!store.release_scanner_publication_lease(expiring_token).await); } + #[tokio::test] + async fn observational_lease_only_allows_observed_usage_rename_during_namespace_commits() { + let ctx = Arc::new(InstanceContext::new()); + let store = build_store_with_ctx(ctx.clone()); + let namespace_commit = ctx.begin_namespace_commit(); + let strict_error = store + .acquire_scanner_publication_lease(0, crate::runtime::instance::SCANNER_PUBLICATION_LEASE_TTL) + .await + .expect_err("authoritative lease must wait for namespace commits"); + assert!(strict_error.to_string().contains("pending namespace commit")); + + let (token, movement_generation) = store + .acquire_scanner_publication_lease_with_purpose(0, crate::runtime::instance::SCANNER_PUBLICATION_LEASE_TTL, true) + .await + .expect("an observation may lease-fence movement while a namespace commit is active"); + assert_eq!(movement_generation, 0); + store + .acquire_scanner_publication_lease_guard_for_rename(token, RUSTFS_META_BUCKET, "buckets/.usage.observed.json") + .await + .expect("the lease must admit the exact observational object"); + assert!( + store + .acquire_scanner_publication_lease_guard_for_rename(token, RUSTFS_META_BUCKET, "buckets/.usage.v2.json") + .await + .is_err(), + "an observational lease must not admit the authoritative object" + ); + assert!( + store.acquire_scanner_publication_lease_guard(token).await.is_err(), + "an observational lease must not authorize an unscoped mutation such as delete or cleanup" + ); + + drop(namespace_commit); + assert!(store.ctx.namespace_commit_generation() > 0); + store + .validate_scanner_publication_lease_with_purpose(token, movement_generation, true) + .await + .expect("ordinary namespace generation changes must not invalidate an observation"); + let error = store + .validate_scanner_publication_lease_with_purpose(token, movement_generation, false) + .await + .expect_err("the same token must not validate as authoritative"); + assert!(error.to_string().contains("purpose")); + + let operation_gate = store.ctx.data_movement_operation_gate(); + assert!( + operation_gate.clone().try_write_owned().is_err(), + "the observational lease must still block data movement" + ); + assert!(store.release_scanner_publication_lease(token).await); + assert!(operation_gate.try_write_owned().is_ok()); + } + #[tokio::test] async fn scanner_publication_lease_rejects_a_new_movement_generation() { let store = build_store_with_ctx(Arc::new(InstanceContext::new())); @@ -2613,8 +2762,11 @@ mod tests { let error = tokio::time::timeout(Duration::from_secs(1), acquire) .await .expect("snapshot admission must finish") - .expect_err("acquisition must preserve its original namespace generation"); - assert_eq!(error.to_string(), "Io error: scanner publication lease generation is stale"); + .expect_err("a namespace change during admission must not install a lease"); + assert_eq!( + error.to_string(), + "Io error: scanner publication lease namespace changed during acquisition" + ); assert!(ctx.data_movement_operation_gate().try_write_owned().is_ok(), "no lease was installed"); } diff --git a/crates/ecstore/src/store/peer.rs b/crates/ecstore/src/store/peer.rs index 51aea5dce..655492c66 100644 --- a/crates/ecstore/src/store/peer.rs +++ b/crates/ecstore/src/store/peer.rs @@ -232,7 +232,7 @@ impl ECStore { ) -> DiskResult { let external_guard: Option> = if let Some(token) = scanner_token { Some(Arc::new( - self.acquire_scanner_publication_lease_guard(token) + self.acquire_scanner_publication_lease_guard_for_rename(token, destination.0, destination.1) .await .map_err(|err| DiskError::other(err.to_string()))?, )) @@ -1191,6 +1191,69 @@ mod tests { assert!(!ctx.namespace_commits_pending()); } + #[tokio::test] + async fn observational_lease_renames_only_the_observed_usage_object() { + let ctx = Arc::new(InstanceContext::new()); + let root = tempfile::tempdir().expect("target root"); + let disk = target_disk(&ctx, root.path(), Uuid::new_v4()).await; + let store = super::super::tests::build_store_with_ctx(ctx.clone()); + let disk_ref = disk.endpoint().to_string(); + let volume = crate::disk::RUSTFS_META_BUCKET; + let observed_path = format!( + "{}/{}", + crate::disk::BUCKET_META_PREFIX, + rustfs_data_usage::DATA_USAGE_OBSERVED_OBJECT_NAME + ); + let source_path = format!("{}/.staged-observation", crate::disk::BUCKET_META_PREFIX); + let fi = target_file_info(&observed_path, Uuid::new_v4(), b"non-authoritative"); + seed_target(&disk, volume, &source_path, fi.clone()).await; + + let pending = ctx.begin_namespace_commit(); + let (token, generation) = store + .acquire_scanner_publication_lease_with_purpose(0, crate::runtime::instance::SCANNER_PUBLICATION_LEASE_TTL, true) + .await + .expect("observational lease should permit a pending namespace commit"); + let error = store + .rename_local_data(&disk_ref, (volume, &source_path), &fi, (volume, "buckets/.usage.v2.json"), Some(token)) + .await + .expect_err("observational token must not authorize the authoritative usage path"); + assert!(error.to_string().contains("observed usage publication")); + assert!(root.path().join(volume).join(&source_path).join("xl.meta").exists()); + + store + .rename_local_data(&disk_ref, (volume, &source_path), &fi, (volume, &observed_path), Some(token)) + .await + .expect("observational token may rename only the observed usage key"); + assert!( + store + .validate_scanner_publication_lease_with_purpose(token, generation, true) + .await + .is_ok(), + "namespace commits do not invalidate the movement-only observation lease" + ); + let wrong_purpose = store + .validate_scanner_publication_lease_with_purpose(token, generation, false) + .await + .expect_err("the same token must not validate as authoritative"); + assert!(wrong_purpose.to_string().contains("purpose")); + drop(pending); + assert!(store.release_scanner_publication_lease(token).await); + let observed = disk + .read_version( + volume, + volume, + &observed_path, + &fi.version_id.expect("fixture version").to_string(), + &crate::disk::ReadOptions { + read_data: true, + ..Default::default() + }, + ) + .await + .expect("observed metadata should be present"); + assert_eq!(observed.data.as_deref(), Some(&b"non-authoritative"[..])); + } + #[tokio::test] async fn completed_namespace_commit_rejects_stale_scanner_target_admission() { use futures::FutureExt; @@ -1491,7 +1554,11 @@ mod tests { ); assert_eq!( ctx.scanner_publication_lease_generations(old_token).await, - Some((movement_generation, namespace_generation)), + Some(( + movement_generation, + Some(namespace_generation), + crate::runtime::instance::ScannerPublicationLeasePurpose::Authoritative, + )), "rejection must not refresh or discard the old lease" ); let (fresh_token, fresh_generation) = store diff --git a/crates/protos/src/generated/proto_gen/node_service.rs b/crates/protos/src/generated/proto_gen/node_service.rs index f7372c037..309f485d3 100644 --- a/crates/protos/src/generated/proto_gen/node_service.rs +++ b/crates/protos/src/generated/proto_gen/node_service.rs @@ -1396,6 +1396,10 @@ pub struct ScannerPublicationLeaseRequest { /// the proof without changing the v7 activity wire shape. #[prost(bytes = "bytes", tag = "5")] pub token: ::prost::bytes::Bytes, + /// Observations may race ordinary namespace commits but still remain + /// non-authoritative. Peers bind this mode to the observed-usage rename path. + #[prost(bool, tag = "6")] + pub observational_only: bool, } #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] pub struct ScannerPublicationLeaseResponse { @@ -1415,6 +1419,8 @@ pub struct ScannerPublicationLeaseResponse { pub owner_id: ::prost::alloc::string::String, #[prost(string, tag = "8")] pub session_id: ::prost::alloc::string::String, + #[prost(bool, tag = "9")] + pub observational_only: bool, } #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] pub struct ScannerPublicationLeaseReleaseRequest { diff --git a/crates/protos/src/lib.rs b/crates/protos/src/lib.rs index 0c930df9a..a521c3a9f 100644 --- a/crates/protos/src/lib.rs +++ b/crates/protos/src/lib.rs @@ -610,11 +610,17 @@ pub fn canonical_scanner_dirty_usage_snapshot_response_body( pub fn canonical_scanner_publication_lease_request_body( request: &proto_gen::node_service::ScannerPublicationLeaseRequest, ) -> Result, std::num::TryFromIntError> { - const DOMAIN: &[u8] = b"rustfs-scanner-publication-lease-request-v1\0"; + const STRICT_DOMAIN: &[u8] = b"rustfs-scanner-publication-lease-request-v1\0"; + const OBSERVATIONAL_DOMAIN: &[u8] = b"rustfs-scanner-observed-usage-lease-request-v1\0"; + let domain = if request.observational_only { + OBSERVATIONAL_DOMAIN + } else { + STRICT_DOMAIN + }; let challenge = request.challenge.as_ref(); let session_id = request.expected_session_id.as_bytes(); - let mut body = Vec::with_capacity(DOMAIN.len() + challenge.len() + session_id.len() + 40); - body.extend_from_slice(DOMAIN); + let mut body = Vec::with_capacity(domain.len() + challenge.len() + session_id.len() + 40); + body.extend_from_slice(domain); body.extend_from_slice(&u64::try_from(challenge.len())?.to_be_bytes()); body.extend_from_slice(challenge); body.extend_from_slice(&request.expected_movement_generation.to_be_bytes()); @@ -659,16 +665,22 @@ pub fn canonical_scanner_publication_lease_response_body( challenge: &[u8], response: &proto_gen::node_service::ScannerPublicationLeaseResponse, ) -> Result, std::num::TryFromIntError> { - const DOMAIN: &[u8] = b"rustfs-scanner-publication-lease-response-v1\0"; + const STRICT_DOMAIN: &[u8] = b"rustfs-scanner-publication-lease-response-v1\0"; + const OBSERVATIONAL_DOMAIN: &[u8] = b"rustfs-scanner-observed-usage-lease-response-v1\0"; + let domain = if response.observational_only { + OBSERVATIONAL_DOMAIN + } else { + STRICT_DOMAIN + }; let token = response.token.as_ref(); let owner_id = response.owner_id.as_bytes(); let session_id = response.session_id.as_bytes(); let error_info = response.error.as_ref().map(|error| error.error_info.as_bytes()); let error_code = response.error.as_ref().map_or(0, |error| error.code); let mut body = Vec::with_capacity( - DOMAIN.len() + challenge.len() + token.len() + owner_id.len() + session_id.len() + error_info.map_or(0, |v| v.len()) + 72, + domain.len() + challenge.len() + token.len() + owner_id.len() + session_id.len() + error_info.map_or(0, |v| v.len()) + 72, ); - body.extend_from_slice(DOMAIN); + body.extend_from_slice(domain); body.extend_from_slice(&u64::try_from(challenge.len())?.to_be_bytes()); body.extend_from_slice(challenge); body.push(u8::from(response.success)); @@ -2196,6 +2208,7 @@ mod scanner_activity_tests { ttl_ms: 60_000, expected_session_id: "session-a".to_string(), token: Vec::new().into(), + observational_only: false, }; let baseline = canonical_scanner_publication_lease_request_body(&request).unwrap(); for variant in [ @@ -2213,6 +2226,10 @@ mod scanner_activity_tests { }, ScannerPublicationLeaseRequest { token: vec![3; 16].into(), + ..request.clone() + }, + ScannerPublicationLeaseRequest { + observational_only: true, ..request }, ] { @@ -2243,6 +2260,7 @@ mod scanner_activity_tests { response_proof: Vec::new().into(), owner_id: "owner-a".to_string(), session_id: "session-a".to_string(), + observational_only: false, }; let response_changed = ScannerPublicationLeaseResponse { owner_id: "owner-b".to_string(), @@ -2252,6 +2270,14 @@ mod scanner_activity_tests { canonical_scanner_publication_lease_response_body(&[1; 16], &response).unwrap(), canonical_scanner_publication_lease_response_body(&[1; 16], &response_changed).unwrap() ); + let observational_response = ScannerPublicationLeaseResponse { + observational_only: true, + ..response.clone() + }; + assert_ne!( + canonical_scanner_publication_lease_response_body(&[1; 16], &response).unwrap(), + canonical_scanner_publication_lease_response_body(&[1; 16], &observational_response).unwrap() + ); } } diff --git a/crates/protos/src/node.proto b/crates/protos/src/node.proto index 302bbe103..a99f73185 100644 --- a/crates/protos/src/node.proto +++ b/crates/protos/src/node.proto @@ -981,6 +981,9 @@ message ScannerPublicationLeaseRequest { // existing lease. Keeping this on the existing RPC lets old peers reject // the proof without changing the v7 activity wire shape. bytes token = 5; + // Observations may race ordinary namespace commits but still remain + // non-authoritative. Peers bind this mode to the observed-usage rename path. + bool observational_only = 6; } message ScannerPublicationLeaseResponse { @@ -992,6 +995,7 @@ message ScannerPublicationLeaseResponse { bytes response_proof = 6; string owner_id = 7; string session_id = 8; + bool observational_only = 9; } message ScannerPublicationLeaseReleaseRequest { diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index 2377c317c..c41b01176 100644 --- a/crates/scanner/src/scanner.rs +++ b/crates/scanner/src/scanner.rs @@ -1924,7 +1924,7 @@ where // fence remains enforced by the usage store and the cycle is advanced // as partial without acknowledging dirty usage. Ok(result) - if result.has_observational_snapshot() + if scanner_cycle_uses_observational_publication(result.status, result.has_observational_snapshot()) && matches!( result.status, ScannerCycleStatus::Deferred(ScannerCycleDeferReason::ActivityBaselineUnavailable) @@ -1932,9 +1932,19 @@ where { None } - Ok(result) => final_data_usage_publication_defer_reason(storeapi.as_ref(), result.status).await, + Ok(result) => { + final_data_usage_publication_defer_reason( + storeapi.as_ref(), + result.status, + scanner_cycle_uses_observational_publication(result.status, result.has_observational_snapshot()), + ) + .await + } Err(_) => Some(ScannerCycleDeferReason::ActivityBaselineUnavailable), }; + let observational_only = scan_result + .as_ref() + .is_ok_and(|result| scanner_cycle_uses_observational_publication(result.status, result.has_observational_snapshot())); let publication_epoch = scan_result.as_ref().ok().and_then(ScannerCycleResult::publication_epoch); let remote_publication_lease_targets = if publication_defer_reason.is_none() { scan_result @@ -1954,7 +1964,19 @@ where &publication_proof_ctx, cycle_info.current, "lease_acquire", - || notification_system.acquire_scanner_publication_leases(remote_publication_lease_targets.clone()), + || { + let targets = remote_publication_lease_targets.clone(); + let notification_system = Arc::clone(¬ification_system); + async move { + if observational_only { + notification_system + .acquire_observational_scanner_publication_leases(targets) + .await + } else { + notification_system.acquire_scanner_publication_leases(targets).await + } + } + }, |err| scanner_publication_lease_error_is_retryable(&err.to_string()), ) .await; @@ -2041,6 +2063,7 @@ where let storeapi_clone = storeapi.clone(); let ctx_clone = ctx.clone(); let route_probe_store = storeapi.clone(); + let route_probe_observational_only = observational_only; let remote_lease_fence = remote_lease_fence.clone(); let remote_lease_release_safe_for_task = Arc::clone(&remote_lease_release_safe); let remote_lease_tokens = remote_publication_leases @@ -2070,6 +2093,7 @@ where move || { let storeapi = route_probe_store.clone(); let remote_lease_probe = remote_lease_probe.clone(); + let observational_only = route_probe_observational_only; async move { if let Some((notification_system, grants)) = remote_lease_probe.as_ref() && notification_system.validate_scanner_publication_leases(grants).await.is_err() @@ -2091,7 +2115,7 @@ where probe_scanner_activity(storeapi.as_ref(), true).await, )); } - scanner_local_publication_defer_reason(storeapi.as_ref()).await + scanner_local_publication_defer_reason(storeapi.as_ref(), observational_only).await } }, ) @@ -3497,13 +3521,17 @@ impl Drop for ScannerScanModeGuard { } } -async fn final_data_usage_publication_defer_reason(storeapi: &S, status: ScannerCycleStatus) -> Option +async fn final_data_usage_publication_defer_reason( + storeapi: &S, + status: ScannerCycleStatus, + observational_only: bool, +) -> Option where S: ScannerStorage, { match status { ScannerCycleStatus::Complete | ScannerCycleStatus::Superseded => { - if let Some(reason) = scanner_local_publication_defer_reason(storeapi).await { + if let Some(reason) = scanner_local_publication_defer_reason(storeapi, observational_only).await { return Some(reason); } if status == ScannerCycleStatus::Complete { @@ -3527,16 +3555,21 @@ where } } -async fn scanner_local_publication_defer_reason(storeapi: &S) -> Option +async fn scanner_local_publication_defer_reason(storeapi: &S, observational_only: bool) -> Option where S: ScannerStorage, { - if !storeapi.scanner_data_usage_publication_blocked().await { + let blocked = if observational_only { + storeapi.scanner_data_movement_publication_blocked().await + } else { + storeapi.scanner_data_usage_publication_blocked().await + }; + if !blocked { return None; } // Pending namespace commits invalidate this publication attempt, but only // storage movement creates durable, rate-limited catch-up debt. - if storeapi.scanner_data_movement_pause_status().await.paused { + if storeapi.scanner_data_movement_publication_blocked().await { debug!( target: "rustfs::scanner", event = EVENT_SCANNER_PERSIST_STATE, @@ -3561,6 +3594,14 @@ where } } +fn scanner_cycle_uses_observational_publication(status: ScannerCycleStatus, has_observational_candidate: bool) -> bool { + has_observational_candidate + && matches!( + status, + ScannerCycleStatus::Superseded | ScannerCycleStatus::Deferred(ScannerCycleDeferReason::ActivityBaselineUnavailable) + ) +} + fn scanner_post_lease_activity_defer_reason( expected_digest: Option<[u8; 32]>, activity: Result, diff --git a/crates/scanner/src/scanner/activity.rs b/crates/scanner/src/scanner/activity.rs index 06accff5b..2b63d115a 100644 --- a/crates/scanner/src/scanner/activity.rs +++ b/crates/scanner/src/scanner/activity.rs @@ -183,6 +183,8 @@ pub(crate) fn scanner_publication_lease_error_is_retryable(error: &str) -> bool scanner_publication_activity_error_is_retryable(error) || error.ends_with("scanner publication lease capacity is exhausted") || error.ends_with("scanner publication lease response arrived after its safety window") + || error.ends_with("scanner publication lease is blocked by pending namespace commit") + || error.ends_with("scanner publication lease namespace changed during acquisition") } pub(crate) enum ScannerPublicationProofWait { diff --git a/crates/scanner/src/scanner/tests.rs b/crates/scanner/src/scanner/tests.rs index f94f1c960..6cc15580a 100644 --- a/crates/scanner/src/scanner/tests.rs +++ b/crates/scanner/src/scanner/tests.rs @@ -369,10 +369,15 @@ async fn running_main_loop_catches_up_pause_cleared_after_startup_observe() { let pause_status = store.scanner_data_movement_pause_status().await; assert!(pause_status.paused); assert_eq!( - scanner_local_publication_defer_reason(store.as_ref()).await, + scanner_local_publication_defer_reason(store.as_ref(), false).await, Some(ScannerCycleDeferReason::DataMovement), "an actual data-movement pause must retain durable catch-up tracking" ); + assert_eq!( + scanner_local_publication_defer_reason(store.as_ref(), true).await, + Some(ScannerCycleDeferReason::DataMovement), + "an observational lease must retain the data-movement fence" + ); paused_probe.wait().await; drop(paused_probe); @@ -1425,7 +1430,7 @@ async fn run_data_scanner_cycle_publishes_activity_for_owner_lifetime() { #[tokio::test] #[serial] -async fn coordinator_walks_during_pending_put_without_persisting_or_acknowledging_usage() { +async fn coordinator_persists_only_an_observation_during_pending_put() { crate::scanner_io::clear_dirty_usage_buckets_for_tests(); let (_temp_dir, store) = setup_scanner_cycle_store().await; let mut pause_backlog = ScannerPauseBacklogController::claim(store.clone(), scanner_pause_backlog_now()) @@ -1459,10 +1464,15 @@ async fn coordinator_walks_during_pending_put_without_persisting_or_acknowledgin .expect("fixture usage baseline should be readable"); let pending = ecstore_hold_namespace_commit(store.as_ref()); assert_eq!( - scanner_local_publication_defer_reason(store.as_ref()).await, + scanner_local_publication_defer_reason(store.as_ref(), false).await, Some(ScannerCycleDeferReason::ActivityBaselineUnavailable), "an ordinary namespace commit must not be classified as data movement" ); + assert_eq!( + scanner_local_publication_defer_reason(store.as_ref(), true).await, + None, + "an observational snapshot may publish while only an ordinary namespace commit is pending" + ); let pause_backlog_attempt = pause_backlog.begin_attempt(scanner_pause_backlog_now()).await; assert_eq!(pause_backlog_attempt, ScannerPauseBacklogAttemptDecision::Untracked); let ctx = CancellationToken::new(); @@ -1490,18 +1500,23 @@ async fn coordinator_walks_during_pending_put_without_persisting_or_acknowledgin .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::ActivityBaselineUnavailable) + assert!( + matches!( + outcome, + ScannerCycleOutcome::Superseded | ScannerCycleOutcome::Partial | ScannerCycleOutcome::CompletedWithPendingMaintenance + ), + "a pending PUT may store an observation without publishing authority: {outcome:?}" ); finish_scanner_pause_backlog_cycle(&mut pause_backlog, &store, pause_backlog_attempt, outcome).await; let pause_backlog_status = scanner_pause_backlog_status(store.clone()).await; assert_eq!(pause_backlog_status.phase, ScannerPauseBacklogPhase::Idle); assert!(!pause_backlog_status.pending_full_scan); assert_eq!(pause_backlog_status.catch_up_attempts, 0); - 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!( + matches!(revision, DataUsageCacheRevision::Etag(_)), + "the cycle-state floor should advance after its durable observation" + ); assert_eq!( read_config(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str()) .await @@ -1509,6 +1524,11 @@ async fn coordinator_walks_during_pending_put_without_persisting_or_acknowledgin baseline, "the pending candidate must not replace the authoritative baseline" ); + let observed = read_config(store.clone(), DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str()) + .await + .expect("the pending cycle should persist its non-authoritative observation"); + let observed: DataUsageInfo = serde_json::from_slice(&observed).expect("observed usage should decode"); + assert_eq!(observed.usage_snapshot_converged, Some(false)); let committed_body = b"committed-after-walk"; let mut reader = PutObjReader::from_vec(committed_body.to_vec()); @@ -1530,6 +1550,7 @@ async fn coordinator_walks_during_pending_put_without_persisting_or_acknowledgin 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 retry_cycle = cycle_info.next; let outcome = tokio::time::timeout( Duration::from_secs(30), run_data_scanner_cycle_with_budget( @@ -1556,14 +1577,14 @@ async fn coordinator_walks_during_pending_put_without_persisting_or_acknowledgin outcome, ScannerCycleOutcome::Completed | ScannerCycleOutcome::CompletedWithPendingMaintenance )); - assert_eq!(cycle_info.next, 2); + assert_eq!(cycle_info.next, retry_cycle + 1); 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.scanner_cycle, Some(retry_cycle)); assert_eq!(usage.objects_total_count, 1); assert_eq!( usage.objects_total_size, @@ -1579,6 +1600,33 @@ async fn coordinator_walks_during_pending_put_without_persisting_or_acknowledgin crate::scanner_io::clear_dirty_usage_buckets_for_tests(); } +#[tokio::test] +#[serial] +async fn complete_cycle_keeps_strict_publication_fence_with_observational_flag() { + let (_temp_dir, store) = setup_scanner_cycle_store().await; + let pending = ecstore_hold_namespace_commit(store.as_ref()); + + let complete_uses_observation = scanner_cycle_uses_observational_publication(ScannerCycleStatus::Complete, true); + assert!(!complete_uses_observation, "a converged Complete snapshot is authoritative"); + assert_eq!( + scanner_local_publication_defer_reason(store.as_ref(), complete_uses_observation).await, + Some(ScannerCycleDeferReason::ActivityBaselineUnavailable), + "a Complete cycle must remain fenced while an ordinary namespace commit is pending" + ); + + assert!(scanner_cycle_uses_observational_publication(ScannerCycleStatus::Superseded, true)); + assert!(scanner_cycle_uses_observational_publication( + ScannerCycleStatus::Deferred(ScannerCycleDeferReason::ActivityBaselineUnavailable), + true + )); + assert!(!scanner_cycle_uses_observational_publication(ScannerCycleStatus::Incomplete, true)); + assert!(!scanner_cycle_uses_observational_publication( + ScannerCycleStatus::Deferred(ScannerCycleDeferReason::DataMovement), + true + )); + drop(pending); +} + #[tokio::test] #[serial] async fn test_finalize_partial_scan_cycle_advances_and_persists_counter() { @@ -8411,6 +8459,8 @@ fn publication_lease_retry_preserves_only_recoverable_candidates() { for error in [ "scanner publication lease acquisition failed: scanner publication lease capacity is exhausted", "scanner publication lease acquisition failed: scanner publication lease response arrived after its safety window", + "scanner publication lease acquisition failed: scanner publication lease is blocked by pending namespace commit", + "scanner publication lease acquisition failed: scanner publication lease namespace changed during acquisition", "scanner publication lease acquisition failed: peer node3 is temporarily offline", ] { assert!(scanner_publication_lease_error_is_retryable(error), "{error}"); @@ -8420,12 +8470,43 @@ fn publication_lease_retry_preserves_only_recoverable_candidates() { "scanner publication lease acquisition failed: scanner publication lease generation is stale", "scanner publication lease acquisition failed: peer returned a different scanner publication lease session", "scanner publication lease acquisition failed: scanner publication lease is blocked by data movement", + "scanner publication lease acquisition failed: scanner publication lease generation is exhausted", "scanner publication lease acquisition failed: peer returned an invalid scanner publication lease proof", ] { assert!(!scanner_publication_lease_error_is_retryable(error), "{error}"); } } +#[tokio::test(start_paused = true)] +async fn publication_lease_retries_pending_namespace_commit_until_it_clears() { + let ctx = CancellationToken::new(); + let attempts = Arc::new(AtomicUsize::new(0)); + let proof_attempts = attempts.clone(); + let started_at = Instant::now(); + + let result = await_scanner_publication_proof( + &ctx, + 17, + "lease_acquire", + move || { + let attempt = proof_attempts.fetch_add(1, Ordering::SeqCst); + async move { + if attempt == 0 { + Err("scanner publication lease is blocked by pending namespace commit".to_string()) + } else { + Ok(()) + } + } + }, + |error| scanner_publication_lease_error_is_retryable(error), + ) + .await; + + assert!(matches!(result, ScannerPublicationProofWait::Ready(()))); + assert_eq!(attempts.load(Ordering::SeqCst), 2); + assert_eq!(started_at.elapsed(), scanner_publication_proof_retry_delay(1)); +} + #[tokio::test(start_paused = true)] async fn publication_proof_retains_candidate_until_activity_recovers() { let ctx = CancellationToken::new(); diff --git a/crates/scanner/src/storage_api.rs b/crates/scanner/src/storage_api.rs index 295a9ef62..c0ff85a96 100644 --- a/crates/scanner/src/storage_api.rs +++ b/crates/scanner/src/storage_api.rs @@ -379,6 +379,7 @@ pub(crate) trait ScannerStorage: fn scanner_namespace_mutation_generation(&self) -> u64; async fn scanner_data_movement_activity(&self) -> (bool, bool, u64); async fn scanner_data_usage_publication_blocked(&self) -> bool; + async fn scanner_data_movement_publication_blocked(&self) -> bool; async fn scanner_data_movement_pause_status(&self) -> ScannerDataMovementPauseStatus; fn scanner_data_movement_generation(&self) -> u64; fn scanner_data_movement_changed(&self) -> Arc; @@ -418,6 +419,10 @@ impl ScannerStorage for EcstoreStore { EcstoreStore::scanner_data_usage_publication_blocked(self).await } + async fn scanner_data_movement_publication_blocked(&self) -> bool { + EcstoreStore::scanner_data_movement_activity(self).await.1 + } + async fn scanner_data_movement_pause_status(&self) -> ScannerDataMovementPauseStatus { EcstoreStore::scanner_data_movement_pause_status(self).await } @@ -587,6 +592,10 @@ mod tests { false } + async fn scanner_data_movement_publication_blocked(&self) -> bool { + false + } + async fn scanner_data_movement_pause_status(&self) -> ScannerDataMovementPauseStatus { ScannerDataMovementPauseStatus::default() } diff --git a/rustfs/src/connect/diagnostics/perf_object.rs b/rustfs/src/connect/diagnostics/perf_object.rs index 70aab597e..6487c4d21 100644 --- a/rustfs/src/connect/diagnostics/perf_object.rs +++ b/rustfs/src/connect/diagnostics/perf_object.rs @@ -64,7 +64,7 @@ const SIGNATURE_PATH: &str = "envelope.sig"; const RESULT_PATH: &str = "result.json"; const MAX_OBJECT_RESPONSE_BYTES: usize = 16_384; const CLEANUP_RESERVE_MAX: Duration = Duration::from_millis(250); -const CLEANUP_TIMEOUT: Duration = Duration::from_secs(2); +const CLEANUP_TIMEOUT: Duration = Duration::from_secs(5); const OUTPUT_MODE: u32 = 0o600; static OBJECT_COLLECTOR_ACTIVE: AtomicBool = AtomicBool::new(false); diff --git a/rustfs/src/storage/rpc/node_service.rs b/rustfs/src/storage/rpc/node_service.rs index b4c2d9912..54f2315b4 100644 --- a/rustfs/src/storage/rpc/node_service.rs +++ b/rustfs/src/storage/rpc/node_service.rs @@ -2594,6 +2594,7 @@ impl Node for NodeService { }; let challenge = request.get_ref().challenge.clone(); let request = request.into_inner(); + let observational_only = request.observational_only; let store = self .resolve_object_store() .ok_or_else(|| Status::unavailable("storage layer is not initialized"))?; @@ -2603,14 +2604,15 @@ impl Node for NodeService { let owner_id = store.id.to_string(); let result = match validation_token { Some(token) => store - .validate_scanner_publication_lease(token, request.expected_movement_generation) + .validate_scanner_publication_lease_with_purpose(token, request.expected_movement_generation, observational_only) .await .map(|()| (token, request.expected_movement_generation)), None => { store - .acquire_scanner_publication_lease( + .acquire_scanner_publication_lease_with_purpose( request.expected_movement_generation, Duration::from_millis(request.ttl_ms), + observational_only, ) .await } @@ -2625,6 +2627,7 @@ impl Node for NodeService { response_proof: Bytes::new(), owner_id: owner_id.clone(), session_id: session_id.clone(), + observational_only, }, Err(err) => ScannerPublicationLeaseResponse { success: false, @@ -2638,6 +2641,7 @@ impl Node for NodeService { response_proof: Bytes::new(), owner_id: owner_id.clone(), session_id: session_id.clone(), + observational_only, }, }; let response_body = rustfs_protos::canonical_scanner_publication_lease_response_body(&challenge, &response) @@ -8763,6 +8767,7 @@ mod tests { ttl_ms: SCANNER_PUBLICATION_LEASE_TTL_MS, expected_session_id: String::new(), token: Bytes::new(), + observational_only: false, } ); assert_tampered!( diff --git a/rustfs/tests/connect_perf_drive.rs b/rustfs/tests/connect_perf_drive.rs index d39577d1a..cd1206da6 100644 --- a/rustfs/tests/connect_perf_drive.rs +++ b/rustfs/tests/connect_perf_drive.rs @@ -80,7 +80,11 @@ fn request(scratch_root: &Path) -> DrivePerformanceRequest { async fn real_local_write_read_is_measured_and_scratch_is_removed() { let _guard = TEST_LOCK.lock().await; let root = tempfile::tempdir().expect("scratch root"); - let request = request(root.path()); + let mut request = request(root.path()); + // This exercises real sync_all and filesystem reads. Keep a generous + // deadline so a slow CI filesystem does not turn the integration test + // into an assertion about runner load. + request.duration = Duration::from_secs(10); let measurement = measure_drive(&request, &CancellationToken::new()) .await .expect("real drive benchmark"); @@ -108,7 +112,7 @@ async fn real_local_write_read_is_measured_and_scratch_is_removed() { assert_eq!(measurement.target.reason_code, DriveTargetReasonCode::Complete); assert_eq!(measurement.target.parameters.scratch_bytes, 32_768); assert_eq!(measurement.target.parameters.block_bytes, 4_096); - assert_eq!(measurement.target.parameters.duration_millis, 1_000); + assert_eq!(measurement.target.parameters.duration_millis, 10_000); assert_eq!(measurement.target.parameters.concurrency, 1); assert_eq!(measurement.target.units.bytes, "BYTE"); assert_eq!(measurement.target.units.duration, "MILLISECOND"); @@ -293,14 +297,38 @@ async fn invalid_roots_and_existing_scratch_fail_without_clobbering() { #[tokio::test] async fn read_only_root_and_full_drive_errors_are_classified() { let _guard = TEST_LOCK.lock().await; - let root = tempfile::tempdir().expect("scratch root"); - fs::set_permissions(root.path(), fs::Permissions::from_mode(0o500)).expect("make root read only"); - let measurement = measure_drive(&request(root.path()), &CancellationToken::new()) + let _temp_root = tempfile::tempdir().expect("scratch root"); + // Permission bits do not make a directory unwritable to root. Linux's + // sysfs is a real directory whose mount rejects child creation even for + // privileged test runners; other Unix runners use a mode-restricted temp + // directory. + #[cfg(target_os = "linux")] + let read_only_root = Path::new("/sys"); + #[cfg(not(target_os = "linux"))] + let read_only_root = { + fs::set_permissions(_temp_root.path(), fs::Permissions::from_mode(0o500)).expect("make root read only"); + _temp_root.path() + }; + #[cfg(target_os = "linux")] + { + let metadata = fs::symlink_metadata(read_only_root).expect("read-only root metadata"); + assert!( + metadata.is_dir() && !metadata.file_type().is_symlink(), + "Linux read-only root must be a real directory" + ); + } + let measurement = measure_drive(&request(read_only_root), &CancellationToken::new()) .await .expect("typed permission result"); - fs::set_permissions(root.path(), fs::Permissions::from_mode(0o700)).expect("restore permissions"); + #[cfg(not(target_os = "linux"))] + fs::set_permissions(read_only_root, fs::Permissions::from_mode(0o700)).expect("restore permissions"); assert_eq!(measurement.result.outcome(), DriveOutcome::Failed); - assert_eq!(measurement.result.reason_code(), DriveReasonCode::PermissionDenied); + assert_eq!( + measurement.result.reason_code(), + DriveReasonCode::PermissionDenied, + "target result: {:?}", + measurement.target + ); assert_eq!(measurement.target.reason_code, DriveTargetReasonCode::PermissionDenied); let failed = serde_json::to_value(&measurement.result).expect("failed result JSON"); assert_eq!(failed["outcome"], "FAILED"); @@ -475,9 +503,17 @@ fn production_cli_measures_and_signs_exact_binary_provenance() { let identity = rustfs::connect::IdentityStore::new(state.join("identity")) .load_or_create() .expect("enrolled identity"); - let result = drive_command(&state, &scratch, &output, "019e3ae0-0000-7000-8000-000000000014", 32_768, true) - .output() - .expect("run production rustfs binary"); + let result = drive_command( + &state, + &scratch, + &output, + "019e3ae0-0000-7000-8000-000000000014", + 32_768, + Duration::from_secs(10), + true, + ) + .output() + .expect("run production rustfs binary"); assert!(result.status.success(), "stderr: {}", String::from_utf8_lossy(&result.stderr)); assert!(fs::read_dir(&scratch).expect("scratch directory").next().is_none()); @@ -492,7 +528,7 @@ fn production_cli_measures_and_signs_exact_binary_provenance() { assert_eq!(target["reasonCode"], "COMPLETE"); assert_eq!(target["parameters"]["scratchBytes"], 32_768); assert_eq!(target["parameters"]["blockBytes"], 4_096); - assert_eq!(target["parameters"]["durationMillis"], 1_000); + assert_eq!(target["parameters"]["durationMillis"], 10_000); assert_eq!(target["parameters"]["concurrency"], 1); assert_eq!(target["units"]["bytes"], "BYTE"); assert_eq!(target["units"]["duration"], "MILLISECOND"); @@ -555,6 +591,7 @@ fn production_cli_rejects_missing_l1_consent_and_limits_before_artifacts() { &no_consent_output, "019e3ae0-0000-7000-8000-000000000024", 32_768, + Duration::from_secs(1), false, ) .output() @@ -571,6 +608,7 @@ fn production_cli_rejects_missing_l1_consent_and_limits_before_artifacts() { &over_limit_output, "019e3ae0-0000-7000-8000-000000000034", 524_289, + Duration::from_secs(1), true, ); let over_limit = over_limit.output().expect("run over limit"); @@ -592,6 +630,7 @@ fn production_cli_rejects_missing_l1_consent_and_limits_before_artifacts() { &failed_output, "019e3ae0-0000-7000-8000-000000000044", 32_768, + Duration::from_secs(1), true, ) .output() @@ -615,6 +654,7 @@ fn drive_command( output: &Path, artifact_uid: &str, scratch_bytes: u64, + duration: Duration, acknowledge_l1: bool, ) -> Command { let current = now(); @@ -644,7 +684,7 @@ fn drive_command( "--expires-at", &(current + 60).to_string(), "--duration-millis", - "1000", + &duration.as_millis().to_string(), ]); command .arg("--scratch-bytes") diff --git a/rustfs/tests/connect_perf_object.rs b/rustfs/tests/connect_perf_object.rs index d2253e154..62dd99d18 100644 --- a/rustfs/tests/connect_perf_object.rs +++ b/rustfs/tests/connect_perf_object.rs @@ -40,7 +40,7 @@ mod common; static TEST_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); // Real storage setup shares the measurement window. Verify functionality with // an integration budget; deadline behavior is checked against a stalled peer. -const REAL_SERVER_MEASUREMENT_BUDGET: Duration = Duration::from_secs(10); +const REAL_SERVER_MEASUREMENT_BUDGET: Duration = Duration::from_secs(20); fn now() -> i64 { SystemTime::now().duration_since(UNIX_EPOCH).expect("current time").as_secs() as i64 @@ -471,7 +471,9 @@ async fn cleanup_uses_a_budget_independent_of_the_measurement_window() { let mut request = vec![0_u8; 16 * 1024]; let _ = socket.read(&mut request).await.expect("request headers"); if index >= 3 { - tokio::time::sleep(Duration::from_millis(600)).await; + // Exercise cleanup beyond the former two-second budget while + // keeping both deletes inside the new bounded allowance. + tokio::time::sleep(Duration::from_millis(1_200)).await; } socket.write_all(response).await.expect("response"); }