fix(odm): fence source work against bucket recreation (#7231)

* fix(odm): fence backfill checkpoints by bucket incarnation

* fix(odm): bind source work to the bucket incarnation

* fix(odm): retain checkpoint fences through owned commit tails

* docs(odm): explain application service and incarnation boundaries

* test(odm): probe lifecycle fence after checkpoint waiter aborts

* fix(odm): defer source identity errors past local reads

* docs(metadata): clarify MinIO target recovery limits

* fix(odm): keep source-free reads independent of capture errors

* fix(odm): retain one source policy snapshot across lookup

* test(odm): name recorded metadata hook snapshots

crates/ecstore/src/bucket/metadata_sys.rs

@@ -50,7 +50,7 @@ use uuid::Uuid;
 
 /// Opaque bucket configuration notifications for application-owned services.
 /// `None` withdraws a configuration; consumers validate nonempty bytes.
-pub type BucketConfigPublishHook = Box<dyn Fn(&str, &str, Option<(&[u8], OffsetDateTime)>) + Send + Sync>;
+pub type BucketConfigPublishHook = Box<dyn Fn(&str, &str, Option<(&[u8], OffsetDateTime, Uuid)>) + Send + Sync>;
 pub static BUCKET_CONFIG_PUBLISH_HOOK: std::sync::OnceLock<BucketConfigPublishHook> = std::sync::OnceLock::new();
 
 const BUCKET_METADATA_REFRESH_INTERVAL: Duration = Duration::from_secs(15 * 60);
@@ -405,7 +405,8 @@ fn sync_on_demand_migration(bucket: &str, bm: &BucketMetadata) {
         hook(
             bucket,
             super::metadata::BUCKET_ON_DEMAND_MIGRATION_CONFIG,
-            bm.on_demand_migration_config(),
+            bm.on_demand_migration_config()
+                .map(|(bytes, stamp)| (bytes, stamp, bm.bucket_incarnation_id)),
         );
     }
 }
@@ -4374,23 +4375,26 @@ mod tests {
 
     const ODM_JSON: &[u8] = br#"{"source":{"provider":"minio","endpoint":"https://legacy.example.com:9000","region":"auto","bucket":"legacy-bucket","credentials":{"access_key":"AK","secret_key":"SK"}}}"#;
 
+    type RecordedOdmConfig = Option<(Vec<u8>, OffsetDateTime, Uuid)>;
+    type RecordedOdmHookCall = (String, RecordedOdmConfig);
+
     /// Every `(bucket, config)` the recording hook has seen. Tests filter by
     /// their own bucket name; the hook is process-wide and set once.
-    static ODM_HOOK_CALLS: std::sync::Mutex<Vec<(String, Option<(Vec<u8>, OffsetDateTime)>)>> = std::sync::Mutex::new(Vec::new());
+    static ODM_HOOK_CALLS: std::sync::Mutex<Vec<RecordedOdmHookCall>> = std::sync::Mutex::new(Vec::new());
 
     fn install_recording_odm_hook() {
         BUCKET_CONFIG_PUBLISH_HOOK.get_or_init(|| {
             Box::new(|bucket, config_file, config| {
                 assert_eq!(config_file, super::super::metadata::BUCKET_ON_DEMAND_MIGRATION_CONFIG);
-                ODM_HOOK_CALLS
-                    .lock()
-                    .unwrap()
-                    .push((bucket.to_string(), config.map(|(bytes, stamp)| (bytes.to_vec(), stamp))));
+                ODM_HOOK_CALLS.lock().unwrap().push((
+                    bucket.to_string(),
+                    config.map(|(bytes, stamp, incarnation)| (bytes.to_vec(), stamp, incarnation)),
+                ));
             })
         });
     }
 
-    fn odm_hook_calls(bucket: &str) -> Vec<Option<(Vec<u8>, OffsetDateTime)>> {
+    fn odm_hook_calls(bucket: &str) -> Vec<RecordedOdmConfig> {
         ODM_HOOK_CALLS
             .lock()
             .unwrap()
@@ -4414,18 +4418,21 @@ mod tests {
             std::fs::create_dir_all(dir.path().join(bucket)).expect("physical bucket should exist");
         }
 
+        let incarnation = Uuid::new_v4();
         let expect_publish = |before: usize, label: &str| {
             let calls = odm_hook_calls(bucket);
             assert_eq!(calls.len(), before + 1, "{label} must publish exactly once");
             assert_eq!(
-                calls.last().unwrap().as_ref().map(|(bytes, _)| bytes.as_slice()),
+                calls.last().unwrap().as_ref().map(|(bytes, _, _)| bytes.as_slice()),
                 Some(ODM_JSON),
                 "{label} must publish the stored bytes"
             );
+            assert_eq!(calls.last().unwrap().as_ref().map(|(_, _, id)| *id), Some(incarnation));
         };
 
         // set (via persist_new_and_set, which installs through `set`).
         let mut bm = BucketMetadata::new(bucket);
+        bm.bucket_incarnation_id = incarnation;
         bm.update_config(crate::bucket::metadata::BUCKET_ON_DEMAND_MIGRATION_CONFIG, ODM_JSON.to_vec())
             .unwrap();
         let writer = BucketMetadataSys::new(ecstore.clone());
@@ -4475,7 +4482,7 @@ mod tests {
         let calls = odm_hook_calls(bucket);
         assert_eq!(calls.len(), before + 1);
         assert_eq!(
-            calls.last().unwrap().as_ref().map(|(bytes, _)| bytes.as_slice()),
+            calls.last().unwrap().as_ref().map(|(bytes, _, _)| bytes.as_slice()),
             Some(b"not-json".as_slice()),
             "the application validates opaque config bytes"
         );

crates/ecstore/src/store/bucket.rs

@@ -329,11 +329,11 @@ impl ECStore {
     /// reuse its result, which is sound because bucket deletion/recreation
     /// requires the lifecycle WRITE lock and therefore cannot have run while
     /// any read guard was continuously held.
-    pub(crate) async fn acquire_bucket_incarnation_fence(
+    pub async fn acquire_bucket_incarnation_fence(
         &self,
         bucket: &str,
         expected: uuid::Uuid,
-    ) -> Result<super::bucket_fence::BucketIncarnationFenceGuard> {
+    ) -> Result<super::BucketIncarnationFenceGuard> {
         let inner = self.acquire_bucket_lifecycle_read_lock(bucket).await?;
         let pieces = super::bucket_fence::FencePieces {
             registry: self.bucket_fence_registry.clone(),

crates/ecstore/src/store/bucket_fence.rs

@@ -150,7 +150,7 @@ impl BucketFenceRegistry {
 /// A held bucket lifecycle read lock plus its registration in the fence
 /// registry. Dropping the guard deregisters it; the memo is cleared when the
 /// last guard for the bucket drops (or a lost lock is observed).
-pub(crate) struct BucketIncarnationFenceGuard {
+pub struct BucketIncarnationFenceGuard {
     inner: Option<NamespaceLockGuard>,
     registry: Arc<BucketFenceRegistry>,
     bucket: String,
@@ -158,6 +158,14 @@ pub(crate) struct BucketIncarnationFenceGuard {
 }
 
 impl BucketIncarnationFenceGuard {
+    /// Propagate lifecycle lock loss into the storage commit checks.
+    /// The caller still owns this guard until the complete write tail drains.
+    pub fn attach_to_object_options(&self, opts: &mut crate::object_api::ObjectOptions) {
+        if let Some(guard) = self.namespace_lock_guard() {
+            opts.add_bucket_lifecycle_lock_guard(guard);
+        }
+    }
+
     pub(crate) fn is_lock_lost(&self) -> bool {
         self.inner.as_ref().is_some_and(NamespaceLockGuard::is_lock_lost)
     }
@@ -346,6 +354,36 @@ mod tests {
         first_pieces.abandon("b", first.token);
     }
 
+    #[tokio::test]
+    async fn checkpoint_options_inherit_bucket_fence_lock_loss() {
+        let lock = NamespaceLock::new("bucket-fence-options".to_string(), Arc::new(LocalClient::new()));
+        let inner = lock
+            .acquire_guard(&lock_request("options"))
+            .await
+            .expect("acquire")
+            .expect("quorum");
+        let pieces = FencePieces {
+            registry: Arc::default(),
+            inner,
+        };
+        let registration = pieces.enter("b");
+        let fence = pieces.into_guard("b", registration.token);
+        let mut opts = crate::object_api::ObjectOptions::default();
+        fence.attach_to_object_options(&mut opts);
+        let inherited = opts
+            .bucket_lifecycle_lock_fence
+            .as_ref()
+            .expect("checkpoint inherits lifecycle guard");
+        assert!(!inherited.is_lock_lost());
+        tokio::time::timeout(
+            Duration::from_secs(2),
+            fence.namespace_lock_guard().expect("held guard").lock_lost_notified(),
+        )
+        .await
+        .expect("distributed guard expires");
+        assert!(inherited.is_lock_lost(), "the actual pre-rename options must observe lifecycle lock loss");
+    }
+
     #[test]
     fn buckets_are_isolated() {
         let reg = BucketFenceRegistry::default();

crates/ecstore/src/store/mod.rs

@@ -417,6 +417,7 @@ const MAX_UPLOADS_LIST: usize = 10000;
 mod bucket;
 mod bucket_fence;
 pub(crate) use bucket::await_bucket_namespace_operation;
+pub use bucket_fence::BucketIncarnationFenceGuard;
 mod heal;
 mod heal_walk;
 pub use heal_walk::HealWalkVersion;