diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 12aff9a83..dd18dd971 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -381,6 +381,16 @@ pub mod config { crate::config::com::read_config_observation(api, file, max_bytes).await } + /// Observe bounded config bytes while the caller retains the object's write owner. + pub async fn read_config_observation_no_lock( + api: std::sync::Arc, + file: &str, + max_bytes: usize, + guard: &crate::object_api::WriteCommitGuard, + ) -> crate::error::Result<(Vec, crate::object_api::ObjectInfo)> { + crate::config::com::read_config_observation_no_lock(api, file, max_bytes, guard).await + } + /// Read config bytes and metadata without treating an empty payload as absent. pub async fn read_config_preserve_empty_with_metadata_opts( api: std::sync::Arc, diff --git a/crates/ecstore/src/config/com.rs b/crates/ecstore/src/config/com.rs index 578447687..fa0a9eeb4 100644 --- a/crates/ecstore/src/config/com.rs +++ b/crates/ecstore/src/config/com.rs @@ -465,6 +465,48 @@ where .await } +pub(crate) async fn read_config_observation_no_lock( + api: Arc, + file: &str, + max_bytes: usize, + guard: &crate::object_api::WriteCommitGuard, +) -> Result<(Vec, ObjectInfo)> +where + S: EcstoreObjectIO, +{ + if !guard.matches(RUSTFS_META_BUCKET, file) { + return Err(Error::InvalidArgument( + RUSTFS_META_BUCKET.to_owned(), + file.to_owned(), + "locked observation requires a matching write owner".to_owned(), + )); + } + let lock_lost_error = || Error::NamespaceLockQuorumUnavailable { + mode: "locked_config_observation", + bucket: RUSTFS_META_BUCKET.to_owned(), + object: file.to_owned(), + required: 1, + achieved: 0, + }; + if guard.is_lock_lost() { + return Err(lock_lost_error()); + } + let opts = ObjectOptions { + suppress_read_repair: true, + no_lock: true, + skip_verify_bitrot: false, + ..Default::default() + }; + let result = crate::object_api::without_get_object_body_cache_hook(read_config_limited_preserve_empty_with_metadata_opts( + api, file, &opts, max_bytes, + )) + .await; + if guard.is_lock_lost() { + return Err(lock_lost_error()); + } + result +} + pub(crate) async fn read_config_limited_preserve_empty(api: Arc, file: &str, max_bytes: usize) -> Result> where S: EcstoreObjectIO, @@ -5445,6 +5487,7 @@ mod tests { inner: Cursor>, consumed: Arc, fail: bool, + eof_gate: Option<(Option>, tokio::sync::oneshot::Receiver<()>)>, } impl AsyncRead for ObservationStream { @@ -5452,6 +5495,18 @@ mod tests { if self.fail { return Poll::Ready(Err(std::io::Error::other("observation stream failed"))); } + if self.inner.position() == u64::try_from(self.inner.get_ref().len()).expect("fixture length") + && let Some((ready, release)) = self.eof_gate.as_mut() + { + if let Some(ready) = ready.take() { + ready.send(()).expect("EOF observer remains active"); + } + match std::future::Future::poll(Pin::new(release), cx) { + Poll::Pending => return Poll::Pending, + Poll::Ready(result) => result.expect("EOF gate released"), + } + self.eof_gate = None; + } let before = buf.filled().len(); let result = Pin::new(&mut self.inner).poll_read(cx, buf); self.consumed.fetch_add(buf.filled().len() - before, Ordering::SeqCst); @@ -5464,6 +5519,8 @@ mod tests { declared_size: i64, error: Option, stream_error: bool, + gets: AtomicUsize, + eof_gate: std::sync::Mutex>, tokio::sync::oneshot::Receiver<()>)>>, consumed: Arc, options: std::sync::Mutex>, writes: AtomicUsize, @@ -5482,6 +5539,8 @@ mod tests { declared_size, error: None, stream_error: false, + gets: AtomicUsize::new(0), + eof_gate: std::sync::Mutex::new(None), consumed: Arc::new(AtomicUsize::new(0)), options: std::sync::Mutex::new(None), writes: AtomicUsize::new(0), @@ -5509,6 +5568,7 @@ mod tests { ) -> Result { assert_eq!(bucket, RUSTFS_META_BUCKET); assert_eq!(object, "config/observation.bin"); + self.gets.fetch_add(1, Ordering::SeqCst); assert!(range.is_none()); *self.options.lock().expect("observation options") = Some((opts.suppress_read_repair, opts.no_lock, opts.skip_verify_bitrot)); @@ -5520,6 +5580,7 @@ mod tests { inner: Cursor::new(self.body.clone()), consumed: self.consumed.clone(), fail: self.stream_error, + eof_gate: self.eof_gate.lock().expect("EOF gate").take(), }), object_info: ObjectInfo { size: self.declared_size, @@ -5543,6 +5604,150 @@ mod tests { } } + #[tokio::test] + async fn locked_observation_preserves_options_and_typed_errors() { + let object = "config/observation.bin"; + let manager = Arc::new(rustfs_lock::GlobalLockManager::new()); + let wrapper = rustfs_lock::NamespaceLockWrapper::new( + rustfs_lock::NamespaceLock::with_local_manager("locked-observation-options".to_owned(), manager.clone()), + rustfs_lock::ObjectKey::new(RUSTFS_META_BUCKET, object), + "locked-observation-options".to_owned(), + ); + let guard = crate::object_api::WriteCommitGuard::acquire(&wrapper, std::time::Duration::from_secs(5)) + .await + .expect("matching real owner"); + let store = Arc::new(ObservationStore::new(b"raw".to_vec(), 3)); + let result = super::read_config_observation_no_lock(store.clone(), object, 3, &guard).await; + assert_eq!(result.is_ok(), true, "I06L_READ_FLAGS: {result:?}"); + assert_eq!( + *store.options.lock().expect("options") == Some((true, true, false)), + true, + "I06L_READ_FLAGS" + ); + assert_eq!(result.expect("raw observation").0, b"raw"); + assert_eq!(store.writes.load(Ordering::SeqCst), 0); + for (bucket, key) in [(RUSTFS_META_BUCKET, "config/other.bin"), ("wrong-bucket", object)] { + let wrapper = rustfs_lock::NamespaceLockWrapper::new( + rustfs_lock::NamespaceLock::with_local_manager("locked-observation-target".to_owned(), manager.clone()), + rustfs_lock::ObjectKey::new(bucket, key), + "locked-observation-target".to_owned(), + ); + let wrong = crate::object_api::WriteCommitGuard::acquire(&wrapper, std::time::Duration::from_secs(5)) + .await + .expect("wrong-target real owner"); + let store = Arc::new(ObservationStore::new(Vec::new(), 0)); + let result = super::read_config_observation_no_lock(store.clone(), object, 8, &wrong).await; + assert_eq!(matches!(result, Err(Error::InvalidArgument(..))), true, "I06L_TARGET_REJECTED"); + assert_eq!(store.gets.load(Ordering::SeqCst) == 0, true, "I06L_TARGET_REJECTED"); + } + for error in [ + Error::FileNotFound, + Error::ObjectNotFound(RUSTFS_META_BUCKET.to_owned(), object.to_owned()), + Error::ErasureReadQuorum, + ] { + let expected = if error == Error::FileNotFound || matches!(error, Error::ObjectNotFound(..)) { + Error::ConfigNotFound + } else { + error.clone() + }; + let mut fixture = ObservationStore::new(Vec::new(), 0); + fixture.error = Some(error); + let store = Arc::new(fixture); + assert_eq!( + super::read_config_observation_no_lock(store.clone(), object, 8, &guard) + .await + .expect_err("typed error"), + expected + ); + assert_eq!(store.writes.load(Ordering::SeqCst), 0); + } + let mut fixture = ObservationStore::new(Vec::new(), 0); + fixture.stream_error = true; + let store = Arc::new(fixture); + let err = super::read_config_observation_no_lock(store.clone(), object, 8, &guard) + .await + .expect_err("stream failure"); + assert_eq!(err.to_string().contains("observation stream failed"), true, "I06L_TYPED_STREAM_ERROR"); + assert_eq!(store.writes.load(Ordering::SeqCst), 0); + let store = Arc::new(ObservationStore::new(vec![7; 64], 0)); + let result = super::read_config_observation_no_lock(store.clone(), object, 8, &guard).await; + assert_eq!(matches!(result, Err(Error::CorruptedFormat)), true, "I06L_STREAM_BOUND"); + assert_eq!(store.consumed.load(Ordering::SeqCst) == 9, true, "I06L_STREAM_BOUND"); + assert_eq!(store.writes.load(Ordering::SeqCst), 0); + } + + #[tokio::test] + async fn locked_observation_rejects_actual_owner_loss_before_and_after_read() { + let object = "config/observation.bin"; + for after_read in [false, true] { + let manager = Arc::new(rustfs_lock::GlobalLockManager::new()); + let client = Arc::new(RefreshControlledClient::new(manager)); + let wrapper = rustfs_lock::NamespaceLockWrapper::new( + rustfs_lock::NamespaceLock::new("locked-observation-loss".to_owned(), client.clone()), + rustfs_lock::ObjectKey::new(RUSTFS_META_BUCKET, object), + "locked-observation-loss".to_owned(), + ); + let guard = crate::object_api::WriteCommitGuard::acquire(&wrapper, std::time::Duration::from_secs(5)) + .await + .expect("real distributed owner"); + let signal = guard.lock_lost_signal().expect("acquired distributed owner signal"); + let store = Arc::new(ObservationStore::new(b"raw".to_vec(), 3)); + if after_read { + let (ready_tx, ready_rx) = tokio::sync::oneshot::channel(); + let (release_tx, release_rx) = tokio::sync::oneshot::channel(); + *store.eof_gate.lock().expect("EOF gate") = Some((Some(ready_tx), release_rx)); + let mut read = Box::pin(super::read_config_observation_no_lock(store.clone(), object, 3, &guard)); + tokio::time::timeout(std::time::Duration::from_secs(60), async { + tokio::select! { + ready = ready_rx => ready.expect("read reached EOF"), + result = &mut read => assert_eq!(false, true, "I06L_LOSS_AFTER_READ: returned before EOF gate: {result:?}"), + } + }).await.expect("EOF watchdog"); + client.reject_refreshes(); + tokio::time::timeout(std::time::Duration::from_secs(60), signal.notified()) + .await + .expect("actual owner loss watchdog"); + assert_eq!(guard.is_lock_lost(), true, "I06L_LOSS_AFTER_READ"); + release_tx.send(()).expect("release EOF"); + let result = tokio::time::timeout(std::time::Duration::from_secs(60), read) + .await + .expect("read watchdog"); + assert_eq!( + matches!( + result, + Err(Error::NamespaceLockQuorumUnavailable { + mode: "locked_config_observation", + .. + }) + ), + true, + "I06L_LOSS_AFTER_READ" + ); + assert_eq!(store.gets.load(Ordering::SeqCst), 1); + } else { + client.reject_refreshes(); + tokio::time::timeout(std::time::Duration::from_secs(60), signal.notified()) + .await + .expect("actual owner loss watchdog"); + assert_eq!(guard.is_lock_lost(), true, "I06L_LOSS_BEFORE_GET"); + let result = super::read_config_observation_no_lock(store.clone(), object, 3, &guard).await; + assert_eq!( + matches!( + result, + Err(Error::NamespaceLockQuorumUnavailable { + mode: "locked_config_observation", + .. + }) + ), + true, + "I06L_LOSS_BEFORE_GET" + ); + assert_eq!(store.gets.load(Ordering::SeqCst) == 0, true, "I06L_LOSS_BEFORE_GET"); + } + assert_eq!(store.writes.load(Ordering::SeqCst), 0); + } + } + #[tokio::test] async fn observation_preserves_options_and_errors() { for body in [Vec::new(), b"raw config".to_vec()] { diff --git a/crates/ecstore/src/object_api/write_commit_context.rs b/crates/ecstore/src/object_api/write_commit_context.rs index 769255c1d..79edbec09 100644 --- a/crates/ecstore/src/object_api/write_commit_context.rs +++ b/crates/ecstore/src/object_api/write_commit_context.rs @@ -68,7 +68,7 @@ impl WriteCommitGuard { self.guards.iter().any(|guard| guard.is_lock_lost() || guard.is_released()) } - fn matches(&self, bucket: &str, object: &str) -> bool { + pub(crate) fn matches(&self, bucket: &str, object: &str) -> bool { self.bucket.as_ref() == bucket && self.object.as_ref() == object } } diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 0f98bd3e5..5e5095920 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -1144,6 +1144,100 @@ mod prepared_get_object_metadata_tests { (ObservationHookGuard, calls) } + #[tokio::test] + #[serial_test::serial(body_cache_hook)] + async fn locked_observation_uses_real_outer_guard() { + let object = "config/locked-observation-real.bin"; + let payload = vec![b'p'; OBSERVATION_HOOK_BODY_LEN]; + let (_dirs, set) = make_local_set_disks(4, 2).await; + let mut reader = PutObjReader::from_vec(payload.clone()); + let written = set + .put_object(crate::disk::RUSTFS_META_BUCKET, object, &mut reader, &ObjectOptions::default()) + .await + .expect("persist locked observation fixture"); + let (_hook, calls) = observation_register_hook(object, vec![b'x'; payload.len()]); + assert_eq!(ordinary_observation_get(&set, object).await, vec![b'x'; payload.len()]); + assert_eq!(calls.swap(0, Ordering::SeqCst), 1); + let namespace = set + .new_ns_lock(crate::disk::RUSTFS_META_BUCKET, object) + .await + .expect("namespace"); + let guard = crate::object_api::WriteCommitGuard::acquire(&namespace, std::time::Duration::from_secs(5)) + .await + .expect("matching real owner"); + let mut contender = Box::pin(crate::object_api::WriteCommitGuard::acquire( + &namespace, + std::time::Duration::from_secs(30), + )); + assert_eq!(futures::poll!(contender.as_mut()).is_pending(), true, "I06L_OWNER_RETAINED"); + let result = tokio::time::timeout( + std::time::Duration::from_secs(10), + crate::config::com::read_config_observation_no_lock(set.clone(), object, payload.len(), &guard), + ) + .await + .expect("locked observation hang watchdog"); + assert_eq!(result.is_ok(), true, "I06L_NO_NESTED_LOCK: {result:?}"); + let (bytes, info) = result.expect("locked observation"); + assert_eq!(bytes == payload, true, "I06L_HOOK_BYPASS"); + assert_eq!(calls.load(Ordering::SeqCst) == 0, true, "I06L_HOOK_BYPASS"); + assert_eq!(info.size, written.size); + assert_eq!(info.etag, written.etag); + assert_eq!(info.data_dir, written.data_dir); + assert_eq!(info.name, object); + assert_eq!(futures::poll!(contender.as_mut()).is_pending(), true, "I06L_OWNER_RETAINED"); + drop(guard); + let next = tokio::time::timeout(std::time::Duration::from_secs(10), contender) + .await + .expect("owner release watchdog") + .expect("competing owner acquired"); + drop(next); + assert_eq!(ordinary_observation_get(&set, object).await, vec![b'x'; payload.len()]); + assert_eq!(calls.load(Ordering::SeqCst), 1); + } + + #[tokio::test] + #[serial_test::serial(body_cache_hook)] + async fn locked_observation_preserves_empty_missing_and_bound() { + let (_dirs, set) = make_local_set_disks(4, 2).await; + for (object, body) in [ + ("config/locked-empty.bin", Vec::new()), + ("config/locked-bound.bin", b"raw".to_vec()), + ] { + let mut reader = PutObjReader::from_vec(body.clone()); + let written = set + .put_object(crate::disk::RUSTFS_META_BUCKET, object, &mut reader, &ObjectOptions::default()) + .await + .expect("persist boundary fixture"); + let namespace = set + .new_ns_lock(crate::disk::RUSTFS_META_BUCKET, object) + .await + .expect("namespace"); + let guard = crate::object_api::WriteCommitGuard::acquire(&namespace, std::time::Duration::from_secs(5)) + .await + .expect("matching boundary owner"); + let result = crate::config::com::read_config_observation_no_lock(set.clone(), object, body.len(), &guard).await; + assert_eq!(result.is_ok(), true, "I06L_EMPTY_DISTINCT: {result:?}"); + let (bytes, info) = result.expect("boundary observation"); + assert_eq!(bytes, body); + assert_eq!(info.size, written.size); + if !body.is_empty() { + let result = + crate::config::com::read_config_observation_no_lock(set.clone(), object, body.len() - 1, &guard).await; + assert_eq!(matches!(result, Err(Error::CorruptedFormat)), true, "I06L_BOUND_ENFORCED"); + } + } + let object = "config/locked-absent.bin"; + let namespace = set + .new_ns_lock(crate::disk::RUSTFS_META_BUCKET, object) + .await + .expect("namespace"); + let guard = crate::object_api::WriteCommitGuard::acquire(&namespace, std::time::Duration::from_secs(5)) + .await + .expect("matching missing owner"); + let result = crate::config::com::read_config_observation_no_lock(set, object, 0, &guard).await; + assert_eq!(matches!(result, Err(Error::ConfigNotFound)), true, "I06L_EMPTY_DISTINCT"); + } + #[tokio::test] #[serial_test::serial(body_cache_hook)] async fn observation_real_backend_boundaries() {