feat(ecstore): observe bounded config under retained write owner

crates/ecstore/src/api/mod.rs

@@ -372,6 +372,25 @@ pub mod config {
             with_server_config_write_lock,
         };
 
+        /// Observe persisted config bytes and their reader-associated metadata within a byte limit.
+        pub async fn read_config_observation(
+            api: std::sync::Arc<crate::store::ECStore>,
+            file: &str,
+            max_bytes: usize,
+        ) -> crate::error::Result<(Vec<u8>, crate::object_api::ObjectInfo)> {
+            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<crate::store::ECStore>,
+            file: &str,
+            max_bytes: usize,
+            guard: &crate::object_api::WriteCommitGuard,
+        ) -> crate::error::Result<(Vec<u8>, 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<crate::store::ECStore>,

crates/ecstore/src/config/com.rs

@@ -449,6 +449,64 @@ where
     Ok(data)
 }
 
+pub(crate) async fn read_config_observation<S>(api: Arc<S>, file: &str, max_bytes: usize) -> Result<(Vec<u8>, ObjectInfo)>
+where
+    S: EcstoreObjectIO,
+{
+    let opts = ObjectOptions {
+        suppress_read_repair: true,
+        no_lock: false,
+        skip_verify_bitrot: false,
+        ..Default::default()
+    };
+    crate::object_api::without_get_object_body_cache_hook(read_config_limited_preserve_empty_with_metadata_opts(
+        api, file, &opts, max_bytes,
+    ))
+    .await
+}
+
+pub(crate) async fn read_config_observation_no_lock<S>(
+    api: Arc<S>,
+    file: &str,
+    max_bytes: usize,
+    guard: &crate::object_api::WriteCommitGuard,
+) -> Result<(Vec<u8>, 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<S>(api: Arc<S>, file: &str, max_bytes: usize) -> Result<Vec<u8>>
 where
     S: EcstoreObjectIO,
@@ -5425,6 +5483,329 @@ mod tests {
         assert_eq!(store.write_calls.load(Ordering::SeqCst), 0);
     }
 
+    struct ObservationStream {
+        inner: Cursor<Vec<u8>>,
+        consumed: Arc<AtomicUsize>,
+        fail: bool,
+        eof_gate: Option<(Option<tokio::sync::oneshot::Sender<()>>, tokio::sync::oneshot::Receiver<()>)>,
+    }
+
+    impl AsyncRead for ObservationStream {
+        fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<std::io::Result<()>> {
+            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);
+            result
+        }
+    }
+
+    struct ObservationStore {
+        body: Vec<u8>,
+        declared_size: i64,
+        error: Option<Error>,
+        stream_error: bool,
+        gets: AtomicUsize,
+        eof_gate: std::sync::Mutex<Option<(Option<tokio::sync::oneshot::Sender<()>>, tokio::sync::oneshot::Receiver<()>)>>,
+        consumed: Arc<AtomicUsize>,
+        options: std::sync::Mutex<Option<(bool, bool, bool)>>,
+        writes: AtomicUsize,
+    }
+
+    impl Debug for ObservationStore {
+        fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+            f.debug_struct("ObservationStore").finish()
+        }
+    }
+
+    impl ObservationStore {
+        fn new(body: Vec<u8>, declared_size: i64) -> Self {
+            Self {
+                body,
+                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),
+            }
+        }
+    }
+
+    #[async_trait::async_trait]
+    impl crate::storage_api_contracts::object::ObjectIO for ObservationStore {
+        type Error = Error;
+        type RangeSpec = HTTPRangeSpec;
+        type HeaderMap = HeaderMap;
+        type ObjectOptions = ObjectOptions;
+        type ObjectInfo = ObjectInfo;
+        type GetObjectReader = GetObjectReader;
+        type PutObjectReader = PutObjReader;
+
+        async fn get_object_reader(
+            &self,
+            bucket: &str,
+            object: &str,
+            range: Option<HTTPRangeSpec>,
+            _headers: HeaderMap,
+            opts: &ObjectOptions,
+        ) -> Result<GetObjectReader> {
+            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));
+            if let Some(error) = &self.error {
+                return Err(error.clone());
+            }
+            Ok(GetObjectReader {
+                stream: Box::new(ObservationStream {
+                    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,
+                    etag: Some("observation-etag".to_string()),
+                    ..Default::default()
+                },
+                buffered_body: None,
+                body_source: Default::default(),
+            })
+        }
+
+        async fn put_object(
+            &self,
+            _bucket: &str,
+            _object: &str,
+            _data: &mut PutObjReader,
+            _opts: &ObjectOptions,
+        ) -> Result<ObjectInfo> {
+            self.writes.fetch_add(1, Ordering::SeqCst);
+            Err(Error::other("observation must not write"))
+        }
+    }
+
+    #[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()] {
+            let store = Arc::new(ObservationStore::new(body.clone(), i64::try_from(body.len()).expect("fixture size")));
+            let result = super::read_config_observation(store.clone(), "config/observation.bin", 32).await;
+            assert!(result.is_ok(), "observation must preserve empty and nonempty objects: {result:?}");
+            let (observed, info) = result.expect("successful observation");
+            assert_eq!(observed, body);
+            assert_eq!(info.etag.as_deref(), Some("observation-etag"));
+            assert_eq!(
+                *store.options.lock().expect("options"),
+                Some((true, false, false)),
+                "I05_REPAIR_SUPPRESSED"
+            );
+            assert_eq!(store.writes.load(Ordering::SeqCst), 0);
+        }
+        for (source, expected) in [
+            (Error::FileNotFound, Error::ConfigNotFound),
+            (
+                Error::ObjectNotFound(RUSTFS_META_BUCKET.to_string(), "config/observation.bin".to_string()),
+                Error::ConfigNotFound,
+            ),
+            (Error::ErasureReadQuorum, Error::ErasureReadQuorum),
+        ] {
+            let mut fixture = ObservationStore::new(Vec::new(), 0);
+            fixture.error = Some(source);
+            let store = Arc::new(fixture);
+            let result = super::read_config_observation(store.clone(), "config/observation.bin", 32).await;
+            assert_eq!(result.expect_err("read error must propagate"), 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 result = super::read_config_observation(store.clone(), "config/observation.bin", 32).await;
+        let error = result.expect_err("stream error must propagate");
+        assert!(error.to_string().contains("observation stream failed"));
+        assert_eq!(store.writes.load(Ordering::SeqCst), 0);
+    }
+
+    #[tokio::test]
+    async fn observation_enforces_byte_limit() {
+        let store = Arc::new(ObservationStore::new(vec![7; 64], 0));
+        let result = super::read_config_observation(store.clone(), "config/observation.bin", 8).await;
+        assert_eq!(result.is_err(), true, "I05_BOUND_ENFORCED");
+        assert_eq!(
+            result.expect_err("bounded observation error"),
+            Error::CorruptedFormat,
+            "I05_BOUND_ENFORCED"
+        );
+        assert_eq!(store.consumed.load(Ordering::SeqCst), 9, "I05_BOUND_ENFORCED: drain exactly limit + 1");
+        let store = Arc::new(ObservationStore::new(vec![7; 8], 8));
+        let result = super::read_config_observation(store.clone(), "config/observation.bin", 8).await;
+        assert!(result.is_ok(), "exact byte limit must succeed: {result:?}");
+        assert_eq!(result.expect("exact limit").0, vec![7; 8]);
+        assert_eq!(store.consumed.load(Ordering::SeqCst), 8);
+    }
+
     #[tokio::test]
     async fn read_config_preserve_empty_distinguishes_empty_object() {
         let store = Arc::new(RecoveryMockStore::new(RecoveryReadState::Blob(Vec::new()), None));

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
     }
 }

crates/ecstore/src/set_disk/mod.rs

@@ -1095,6 +1095,311 @@ mod prepared_get_object_metadata_tests {
     use http::HeaderMap;
     use tokio::io::AsyncReadExt;
 
+    const OBSERVATION_HOOK_BODY_LEN: usize = 1024 * 1024 + 1;
+
+    struct ObservationBodyHook {
+        object: &'static str,
+        calls: Arc<AtomicU64>,
+        body: bytes::Bytes,
+    }
+
+    #[async_trait::async_trait]
+    impl crate::object_api::GetObjectBodyCacheHook for ObservationBodyHook {
+        async fn lookup(&self, bucket: &str, object: &str, info: &ObjectInfo) -> Option<bytes::Bytes> {
+            if bucket == crate::disk::RUSTFS_META_BUCKET && object == self.object {
+                assert_eq!(info.size, i64::try_from(self.body.len()).expect("hook fixture size"));
+                self.calls.fetch_add(1, Ordering::SeqCst);
+                Some(self.body.clone())
+            } else {
+                None
+            }
+        }
+    }
+
+    struct ObservationHookGuard;
+
+    impl Drop for ObservationHookGuard {
+        fn drop(&mut self) {
+            crate::object_api::unregister_get_object_body_cache_hook();
+        }
+    }
+
+    async fn ordinary_observation_get(set: &SetDisks, object: &str) -> Vec<u8> {
+        let mut reader = set
+            .get_object_reader(crate::disk::RUSTFS_META_BUCKET, object, None, HeaderMap::new(), &ObjectOptions::default())
+            .await
+            .expect("ordinary GET positive control");
+        let mut bytes = Vec::new();
+        reader.stream.read_to_end(&mut bytes).await.expect("ordinary GET body");
+        bytes
+    }
+
+    fn observation_register_hook(object: &'static str, body: Vec<u8>) -> (ObservationHookGuard, Arc<AtomicU64>) {
+        let calls = Arc::new(AtomicU64::new(0));
+        crate::object_api::register_get_object_body_cache_hook(Arc::new(ObservationBodyHook {
+            object,
+            calls: calls.clone(),
+            body: bytes::Bytes::from(body),
+        }));
+        (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() {
+        let object = "config/observation-real.bin";
+        let payload = b"persisted config";
+        let (_dirs, set) = make_local_set_disks(4, 2).await;
+        let mut reader = PutObjReader::from_vec(payload.to_vec());
+        let written = set
+            .put_object(crate::disk::RUSTFS_META_BUCKET, object, &mut reader, &ObjectOptions::default())
+            .await
+            .expect("persist real config fixture");
+        let result = crate::config::com::read_config_observation(set.clone(), object, payload.len()).await;
+        assert_eq!(result.is_ok(), true, "I05_OBSERVATION_SUCCESS: {result:?}");
+        let (bytes, info) = result.expect("real observation success");
+        assert_eq!(bytes, payload);
+        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);
+        let over = crate::config::com::read_config_observation(set.clone(), object, payload.len() - 1).await;
+        assert_eq!(over.is_err(), true, "I05_BOUND_ENFORCED");
+        assert_eq!(over.expect_err("over limit must fail"), Error::CorruptedFormat, "I05_BOUND_ENFORCED");
+        let absent = crate::config::com::read_config_observation(set.clone(), "config/observation-absent.bin", 32).await;
+        assert_eq!(absent.expect_err("absent config"), Error::ConfigNotFound);
+        let empty_object = "config/observation-empty.bin";
+        let mut empty = PutObjReader::from_vec(Vec::new());
+        set.put_object(crate::disk::RUSTFS_META_BUCKET, empty_object, &mut empty, &ObjectOptions::default())
+            .await
+            .expect("persist empty config");
+        let empty = crate::config::com::read_config_observation(set, empty_object, 0).await;
+        assert!(empty.is_ok(), "existing empty config must succeed: {empty:?}");
+        let (bytes, info) = empty.expect("empty observation");
+        assert!(bytes.is_empty());
+        assert_eq!(info.size, 0);
+    }
+
+    #[tokio::test]
+    #[serial_test::serial(body_cache_hook)]
+    async fn observation_bypasses_registered_body_hook() {
+        let object = "config/observation-hook.bin";
+        let (_dirs, set) = make_local_set_disks(4, 2).await;
+        let mut reader = PutObjReader::from_vec(vec![b'p'; OBSERVATION_HOOK_BODY_LEN]);
+        set.put_object(crate::disk::RUSTFS_META_BUCKET, object, &mut reader, &ObjectOptions::default())
+            .await
+            .expect("persist real config fixture");
+        let (_hook, calls) = observation_register_hook(object, vec![b'x'; OBSERVATION_HOOK_BODY_LEN]);
+        assert_eq!(ordinary_observation_get(&set, object).await, vec![b'x'; OBSERVATION_HOOK_BODY_LEN]);
+        assert_eq!(calls.swap(0, Ordering::SeqCst), 1);
+        let result = crate::config::com::read_config_observation(set, object, OBSERVATION_HOOK_BODY_LEN).await;
+        assert_eq!(result.is_ok(), true, "I05_HOOK_BYPASS: {result:?}");
+        assert_eq!(
+            result.expect("observed persisted body").0,
+            vec![b'p'; OBSERVATION_HOOK_BODY_LEN],
+            "I05_HOOK_BYPASS"
+        );
+        assert_eq!(calls.load(Ordering::SeqCst), 0, "I05_HOOK_BYPASS");
+    }
+
+    #[tokio::test]
+    #[serial_test::serial(body_cache_hook)]
+    async fn observation_hook_scope_restores_after_error() {
+        let object = "config/observation-hook-error.bin";
+        let (_dirs, set) = make_local_set_disks(4, 2).await;
+        let mut reader = PutObjReader::from_vec(vec![b'p'; OBSERVATION_HOOK_BODY_LEN]);
+        set.put_object(crate::disk::RUSTFS_META_BUCKET, object, &mut reader, &ObjectOptions::default())
+            .await
+            .expect("persist real config fixture");
+        let (_hook, calls) = observation_register_hook(object, vec![b'x'; OBSERVATION_HOOK_BODY_LEN]);
+        let result = crate::config::com::read_config_observation(set.clone(), object, OBSERVATION_HOOK_BODY_LEN - 1).await;
+        assert_eq!(result.expect_err("bounded observation error"), Error::CorruptedFormat);
+        assert_eq!(calls.load(Ordering::SeqCst), 0);
+        assert_eq!(ordinary_observation_get(&set, object).await, vec![b'x'; OBSERVATION_HOOK_BODY_LEN]);
+        assert_eq!(calls.load(Ordering::SeqCst), 1);
+    }
+
+    struct PendingObservationStore {
+        set: Arc<SetDisks>,
+        ready: std::sync::Mutex<Option<tokio::sync::oneshot::Sender<()>>>,
+    }
+
+    impl std::fmt::Debug for PendingObservationStore {
+        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+            f.debug_struct("PendingObservationStore").finish()
+        }
+    }
+
+    #[async_trait::async_trait]
+    impl crate::storage_api_contracts::object::ObjectIO for PendingObservationStore {
+        type Error = Error;
+        type RangeSpec = HTTPRangeSpec;
+        type HeaderMap = HeaderMap;
+        type ObjectOptions = ObjectOptions;
+        type ObjectInfo = ObjectInfo;
+        type GetObjectReader = GetObjectReader;
+        type PutObjectReader = PutObjReader;
+
+        async fn get_object_reader(
+            &self,
+            bucket: &str,
+            object: &str,
+            range: Option<HTTPRangeSpec>,
+            headers: HeaderMap,
+            opts: &ObjectOptions,
+        ) -> Result<GetObjectReader> {
+            let mut reader = self.set.get_object_reader(bucket, object, range, headers, opts).await?;
+            let mut body = Vec::new();
+            reader.stream.read_to_end(&mut body).await.expect("pending producer real GET");
+            assert_eq!(
+                body,
+                vec![b'p'; OBSERVATION_HOOK_BODY_LEN],
+                "I05_HOOK_BYPASS: producer must enter its own scope"
+            );
+            self.ready
+                .lock()
+                .expect("pending signal")
+                .take()
+                .expect("one GET")
+                .send(())
+                .expect("producer entry signal");
+            std::future::pending::<Result<GetObjectReader>>().await
+        }
+
+        async fn put_object(
+            &self,
+            _bucket: &str,
+            _object: &str,
+            _data: &mut PutObjReader,
+            _opts: &ObjectOptions,
+        ) -> Result<ObjectInfo> {
+            assert!(false, "observation must not PUT");
+            Err(Error::other("unexpected observation PUT"))
+        }
+    }
+
+    #[tokio::test]
+    #[serial_test::serial(body_cache_hook)]
+    async fn observation_hook_scope_restores_after_cancel() {
+        let object = "config/observation-hook-cancel.bin";
+        let (_dirs, set) = make_local_set_disks(4, 2).await;
+        let mut reader = PutObjReader::from_vec(vec![b'p'; OBSERVATION_HOOK_BODY_LEN]);
+        set.put_object(crate::disk::RUSTFS_META_BUCKET, object, &mut reader, &ObjectOptions::default())
+            .await
+            .expect("persist real config fixture");
+        let (_hook, calls) = observation_register_hook(object, vec![b'x'; OBSERVATION_HOOK_BODY_LEN]);
+        assert_eq!(ordinary_observation_get(&set, object).await, vec![b'x'; OBSERVATION_HOOK_BODY_LEN]);
+        assert_eq!(calls.swap(0, Ordering::SeqCst), 1);
+        let (ready_tx, ready_rx) = tokio::sync::oneshot::channel();
+        let adapter = Arc::new(PendingObservationStore {
+            set: set.clone(),
+            ready: std::sync::Mutex::new(Some(ready_tx)),
+        });
+        let mut scope = Box::pin(crate::config::com::read_config_observation(adapter, object, OBSERVATION_HOOK_BODY_LEN));
+        tokio::select! {
+            result = ready_rx => { result.expect("producer reached pending GET"); }
+            result = &mut scope => { assert!(false, "producer must remain pending until cancelled: {result:?}"); }
+        }
+        assert_eq!(calls.load(Ordering::SeqCst), 0);
+        drop(scope);
+        assert_eq!(ordinary_observation_get(&set, object).await, vec![b'x'; OBSERVATION_HOOK_BODY_LEN]);
+        assert_eq!(calls.load(Ordering::SeqCst), 1);
+    }
+
     const READ_VERSION_BARRIER_GUARD: std::time::Duration = std::time::Duration::from_secs(10);
 
     fn object_with_initial_data_shards(bucket: &str, prefix: &str) -> String {