fix(ilm): chunk large tier-delete dispatches (#7123)

* fix(ilm): chunk large tier-delete dispatches

* fix(ilm): bound tier-delete chunk dispatch stack use

crates/ecstore/src/bucket/lifecycle/durable_namespace.rs

@@ -182,6 +182,16 @@ pub(crate) enum DurableIlmRecordCheckpoint {
         identity_sha256: String,
         state: tier_delete_journal::TierDeleteDispatchManifestState,
     },
+    TierDeleteDispatchParent {
+        content_sha256: String,
+        identity_sha256: String,
+        revision: u64,
+        next_chunk_sequence: u64,
+        completed_journal_count: u64,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        active_chunk_identity_sha256: Option<String>,
+        completed: bool,
+    },
     TransitionTransaction {
         content_sha256: String,
         identity_sha256: String,
@@ -220,6 +230,7 @@ impl DurableIlmRecordCheckpoint {
         match self {
             Self::TierDeleteJournal { content_sha256, .. }
             | Self::TierDeleteDispatchManifest { content_sha256, .. }
+            | Self::TierDeleteDispatchParent { content_sha256, .. }
             | Self::TransitionTransaction { content_sha256, .. }
             | Self::ManualTransitionJob { content_sha256, .. }
             | Self::ManualTransitionScope { content_sha256, .. }
@@ -341,6 +352,52 @@ impl DurableIlmRecordCheckpoint {
                         (Preparing, DispatchAuthorized | Aborting) | (Aborting, Aborted) | (DispatchAuthorized, Completed)
                     )
             }
+            (
+                Self::TierDeleteDispatchParent {
+                    identity_sha256: previous_identity,
+                    revision: previous_revision,
+                    next_chunk_sequence: previous_sequence,
+                    completed_journal_count: previous_completed_journals,
+                    active_chunk_identity_sha256: previous_active,
+                    completed: previous_completed,
+                    ..
+                },
+                Self::TierDeleteDispatchParent {
+                    identity_sha256: next_identity,
+                    revision: next_revision,
+                    next_chunk_sequence: next_sequence,
+                    completed_journal_count: next_completed_journals,
+                    active_chunk_identity_sha256: next_active,
+                    completed: next_completed,
+                    ..
+                },
+            ) => {
+                let Some((sequence_delta, completed_journal_delta)) = tier_delete_dispatch_parent_progress_delta(
+                    *previous_sequence,
+                    *previous_completed_journals,
+                    *next_sequence,
+                    *next_completed_journals,
+                ) else {
+                    return Err(Error::other("durable ILM record generation is not a monotonic successor"));
+                };
+                let same_position_transition = sequence_delta == 0
+                    && completed_journal_delta == 0
+                    && matches!(
+                        (previous_active.as_ref(), next_active.as_ref(), previous_completed, next_completed),
+                        (None, Some(_), false, false) | (Some(_), None, false, false) | (None, None, false, true)
+                    );
+                let progress_transition = sequence_delta > 0
+                    && completed_journal_delta > 0
+                    && !matches!(
+                        (previous_active.as_ref(), next_active.as_ref()),
+                        (Some(previous), Some(next)) if previous == next
+                    );
+                previous_identity == next_identity
+                    && !previous_completed
+                    && next_revision > previous_revision
+                    && (!next_completed || next_active.is_none())
+                    && (same_position_transition || progress_transition)
+            }
             (
                 Self::TransitionTransaction {
                     identity_sha256: previous_identity,
@@ -475,11 +532,58 @@ impl DurableIlmRecordCheckpoint {
                     ..
                 },
             ) => previous_identity == terminal_identity,
+            (
+                Self::TierDeleteDispatchParent {
+                    identity_sha256: previous_identity,
+                    revision: previous_revision,
+                    next_chunk_sequence: previous_sequence,
+                    completed_journal_count: previous_completed_journals,
+                    active_chunk_identity_sha256: previous_active,
+                    completed: false,
+                    ..
+                },
+                Self::TierDeleteDispatchParent {
+                    identity_sha256: terminal_identity,
+                    revision: terminal_revision,
+                    next_chunk_sequence: terminal_sequence,
+                    completed_journal_count: terminal_completed_journals,
+                    active_chunk_identity_sha256: None,
+                    completed: true,
+                    ..
+                },
+            ) => {
+                previous_identity == terminal_identity
+                    && terminal_revision > previous_revision
+                    && tier_delete_dispatch_parent_progress_delta(
+                        *previous_sequence,
+                        *previous_completed_journals,
+                        *terminal_sequence,
+                        *terminal_completed_journals,
+                    )
+                    .is_some_and(|(sequence_delta, completed_journal_delta)| {
+                        if sequence_delta == 0 && completed_journal_delta == 0 {
+                            previous_active.is_none()
+                        } else {
+                            sequence_delta > 0 && completed_journal_delta > 0
+                        }
+                    })
+            }
             _ => false,
         }
     }
 }
 
+fn tier_delete_dispatch_parent_progress_delta(
+    previous_sequence: u64,
+    previous_completed_journals: u64,
+    next_sequence: u64,
+    next_completed_journals: u64,
+) -> Option<(u64, u64)> {
+    let sequence_delta = next_sequence.checked_sub(previous_sequence)?;
+    let completed_journal_delta = next_completed_journals.checked_sub(previous_completed_journals)?;
+    (sequence_delta <= completed_journal_delta).then_some((sequence_delta, completed_journal_delta))
+}
+
 fn transition_state_distance(
     from: transition_transaction::TransitionTransactionState,
     to: transition_transaction::TransitionTransactionState,
@@ -913,17 +1017,42 @@ pub(crate) fn validate_durable_ilm_record(path: &str, data: &[u8]) -> Result<Val
             )
         }
         DurableIlmRecordKind::TierDeleteDispatchManifest => {
-            let (operation_id, identity_sha256, state) =
-                tier_delete_journal::validate_tier_delete_dispatch_manifest_record(path, data)?;
-            (
-                "operation_id",
-                hex_sha256(operation_id.as_bytes(), ToOwned::to_owned),
-                DurableIlmRecordCheckpoint::TierDeleteDispatchManifest {
-                    content_sha256,
+            match tier_delete_journal::validate_tier_delete_dispatch_manifest_record(path, data)? {
+                tier_delete_journal::TierDeleteDispatchDurableRecord::Manifest {
+                    operation_id,
                     identity_sha256,
                     state,
-                },
-            )
+                } => (
+                    "operation_id",
+                    hex_sha256(operation_id.as_bytes(), ToOwned::to_owned),
+                    DurableIlmRecordCheckpoint::TierDeleteDispatchManifest {
+                        content_sha256,
+                        identity_sha256,
+                        state,
+                    },
+                ),
+                tier_delete_journal::TierDeleteDispatchDurableRecord::Parent {
+                    operation_id,
+                    identity_sha256,
+                    revision,
+                    next_chunk_sequence,
+                    completed_journal_count,
+                    active_chunk_identity_sha256,
+                    completed,
+                } => (
+                    "operation_id",
+                    hex_sha256(operation_id.as_bytes(), ToOwned::to_owned),
+                    DurableIlmRecordCheckpoint::TierDeleteDispatchParent {
+                        content_sha256,
+                        identity_sha256,
+                        revision,
+                        next_chunk_sequence,
+                        completed_journal_count,
+                        active_chunk_identity_sha256,
+                        completed,
+                    },
+                ),
+            }
         }
         DurableIlmRecordKind::TransitionTransaction => {
             let transaction = transition_transaction::decode_transition_transaction_record(path, data)
@@ -1143,6 +1272,85 @@ mod tests {
         assert!(aborted.validate_successor(&preparing).is_err());
     }
 
+    #[test]
+    fn tier_delete_dispatch_parent_checkpoint_is_monotonic_across_chunks() {
+        let identity = "a".repeat(64);
+        let checkpoint = |revision, sequence, completed_journals, active: Option<&str>, completed| {
+            DurableIlmRecordCheckpoint::TierDeleteDispatchParent {
+                content_sha256: format!("{revision:064x}"),
+                identity_sha256: identity.clone(),
+                revision,
+                next_chunk_sequence: sequence,
+                completed_journal_count: completed_journals,
+                active_chunk_identity_sha256: active.map(ToOwned::to_owned),
+                completed,
+            }
+        };
+        let idle = checkpoint(0, 0, 0, None, false);
+        let first_child = "b".repeat(64);
+        let second_child = "c".repeat(64);
+        let bound = checkpoint(1, 0, 0, Some(&first_child), false);
+        let advanced = checkpoint(2, 1, 2, None, false);
+        let next_bound = checkpoint(3, 1, 2, Some(&second_child), false);
+        let completed = checkpoint(4, 2, 3, None, true);
+        let terminal_after_more_chunks = checkpoint(6, 4, 7, None, true);
+
+        idle.validate_successor(&bound).expect("an idle parent may bind one child");
+        bound
+            .validate_successor(&advanced)
+            .expect("a completed child may advance the parent sequence");
+        advanced
+            .validate_successor(&next_bound)
+            .expect("the next sequence may bind a new immutable child");
+        next_bound
+            .validate_successor(&completed)
+            .expect("receipt progress may skip directly to a later terminal checkpoint");
+        assert!(
+            bound.is_predecessor_of_terminal(&terminal_after_more_chunks),
+            "terminal cleanup may still recognize a valid multi-chunk predecessor"
+        );
+        assert!(
+            advanced.is_predecessor_of_terminal(&terminal_after_more_chunks),
+            "terminal cleanup may still skip over later valid parent generations"
+        );
+        assert!(
+            idle.validate_successor(&checkpoint(1, 0, 1, Some(&first_child), false))
+                .is_err()
+        );
+        assert!(bound.validate_successor(&checkpoint(2, 1, 0, None, false)).is_err());
+        assert!(
+            bound.validate_successor(&checkpoint(2, 2, 1, None, false)).is_err(),
+            "sequence cannot advance beyond completed journal evidence"
+        );
+        assert!(
+            advanced.validate_successor(&checkpoint(3, 1, 3, None, false)).is_err(),
+            "completed journal count cannot grow without a completed child sequence"
+        );
+        assert!(
+            bound.validate_successor(&checkpoint(2, 0, 0, None, true)).is_err(),
+            "an active child cannot be marked completed without completion evidence"
+        );
+        assert!(
+            bound
+                .validate_successor(&checkpoint(2, 0, 0, Some(&second_child), false))
+                .is_err(),
+            "an active child cannot be replaced at the same parent sequence"
+        );
+        assert!(
+            bound
+                .validate_successor(&checkpoint(2, 1, 1, Some(&first_child), false))
+                .is_err(),
+            "sequence growth cannot retain the same active child identity"
+        );
+        assert!(
+            !bound.is_predecessor_of_terminal(&checkpoint(2, 0, 0, None, true)),
+            "terminal cleanup must not treat an active child as completed without count evidence"
+        );
+        assert!(completed.validate_successor(&checkpoint(5, 3, 4, None, true)).is_err());
+        assert!(completed.validate_successor(&advanced).is_err());
+        assert!(advanced.validate_successor(&idle).is_err());
+    }
+
     #[test]
     fn tier_delete_journal_checkpoint_binds_dispatch_and_full_state_monotonically() {
         use crate::bucket::lifecycle::tier_sweeper::TierDeleteJournalState::{Committed, Dispatched, Prepared};

crates/ecstore/src/store/init.rs

@@ -816,6 +816,7 @@ mod tests {
             },
             tier_delete_journal::{
                 DecommissionCheckpointTargetFailureHook, TIER_DELETE_DISPATCH_MANIFEST_PREFIX, TIER_DELETE_JOURNAL_PREFIX,
+                TierDeleteChunkTestBarrier, TierDeleteChunkTestStage, TierDeleteDispatchBatchLimitGuard,
                 TierDeleteDispatchManifestState, TierDeleteDispatchMemberReadTestHook, TierDeleteDispatchMemberReadTestStage,
                 TierDeleteDispatchRollbackTestHook, complete_tier_delete_dispatch, encode_tier_delete_journal_entry,
                 install_test_tier_delete_dispatch_fixture, persist_tier_delete_journal_entry, prepare_tier_delete_dispatch,
@@ -11624,6 +11625,612 @@ mod tests {
         assert_eq!(backend.remove_versions().await.len(), objects.len());
     }
 
+    #[cfg(feature = "test-util")]
+    #[test]
+    #[serial_test::serial(storage_class_env)]
+    fn tier_delete_prefix_limit_and_multi_chunk_batches_converge() {
+        run_large_stack_async_test(
+            "tier-delete-prefix-limit-and-multi-chunk",
+            tier_delete_prefix_limit_and_multi_chunk_batches_converge_case,
+        );
+    }
+
+    #[cfg(feature = "test-util")]
+    async fn tier_delete_prefix_limit_and_multi_chunk_batches_converge_case() {
+        let _batch_limit = TierDeleteDispatchBatchLimitGuard::install(2);
+        let temp_dir = tempfile::tempdir().expect("create chunked prefix-delete store dir");
+        let (ctx, store, shutdown) =
+            without_storage_class_env(build_isolated_test_store(temp_dir.path(), "chunked-prefix-delete", &[4])).await;
+        crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
+        let tier_name = "CHUNKED-PREFIX-DELETE";
+        let backend = register_mock_tier(&ctx.tier_config_mgr(), tier_name).await;
+        let bucket = "chunked-prefix-delete-bucket";
+        store
+            .make_bucket(bucket, &MakeBucketOptions::default())
+            .await
+            .expect("chunked prefix source bucket should be created");
+
+        for (prefix, count) in [("at-limit", 2), ("limit-plus-one", 3), ("multi-chunk", 5)] {
+            for index in 0..count {
+                let object = format!("{prefix}/object-{index}.bin");
+                let mut reader = PutObjReader::from_vec(vec![b'a' + index as u8; 1024 * 1024]);
+                let source = store
+                    .put_object(bucket, &object, &mut reader, &ObjectOptions::default())
+                    .await
+                    .expect("chunked prefix source should be written");
+                store
+                    .transition_object(
+                        bucket,
+                        &object,
+                        &ObjectOptions {
+                            transition: TransitionOptions {
+                                status: TRANSITION_PENDING.to_string(),
+                                tier: tier_name.to_string(),
+                                etag: source.etag.clone().expect("chunked prefix source should have an etag"),
+                                ..Default::default()
+                            },
+                            mod_time: source.mod_time,
+                            ..Default::default()
+                        },
+                    )
+                    .await
+                    .expect("chunked prefix source should transition");
+            }
+        }
+        backend.set_remove_failure(true);
+
+        store
+            .delete_object_with_tier_delete_journal(
+                bucket,
+                "at-limit/",
+                ObjectOptions {
+                    delete_prefix: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("exactly one batch must retain the v1 one-shot path");
+
+        let limit_plus_one_first = store
+            .delete_object_with_tier_delete_journal(
+                bucket,
+                "limit-plus-one/",
+                ObjectOptions {
+                    delete_prefix: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect_err("limit plus one must start a bounded parent transaction");
+        assert!(
+            limit_plus_one_first.to_string().contains("retry the next durable batch"),
+            "unexpected first limit-plus-one result: {limit_plus_one_first}"
+        );
+        let active_limit_plus_one_records = store
+            .clone()
+            .list_objects_v2(
+                RUSTFS_META_BUCKET,
+                TIER_DELETE_DISPATCH_MANIFEST_PREFIX,
+                None,
+                None,
+                10,
+                false,
+                None,
+                false,
+            )
+            .await
+            .expect("active limit-plus-one records should be listable");
+        let mut active_limit_plus_one_parent_seen = false;
+        for record in active_limit_plus_one_records
+            .objects
+            .iter()
+            .filter(|record| !record.name.contains("/chunks/"))
+        {
+            let data = com::read_config(store.clone(), &record.name)
+                .await
+                .expect("an active dispatch root should be readable");
+            let value: serde_json::Value = serde_json::from_slice(&data).expect("an active dispatch root should contain JSON");
+            if value["prefix"] == "limit-plus-one/" {
+                assert_eq!(value["record_type"], "chunked_parent");
+                active_limit_plus_one_parent_seen = true;
+            }
+        }
+        assert!(
+            active_limit_plus_one_parent_seen,
+            "limit plus one must establish the fail-closed parent at the legacy root"
+        );
+
+        let mut limit_plus_one_completed = false;
+        let mut limit_plus_one_retries = 1;
+        for _ in 0..3 {
+            match store
+                .delete_object_with_tier_delete_journal(
+                    bucket,
+                    "limit-plus-one/",
+                    ObjectOptions {
+                        delete_prefix: true,
+                        ..Default::default()
+                    },
+                )
+                .await
+            {
+                Ok(_) => {
+                    limit_plus_one_completed = true;
+                    break;
+                }
+                Err(err) if err.to_string().contains("retry the next durable batch") => {
+                    limit_plus_one_retries += 1;
+                }
+                Err(err) => panic!("limit-plus-one delete returned an unexpected error: {err}"),
+            }
+        }
+        assert!(limit_plus_one_completed, "limit plus one must converge through two bounded children");
+        assert_eq!(limit_plus_one_retries, 2, "limit plus one must require exactly two child batches");
+
+        let first_batch = store
+            .delete_object_with_tier_delete_journal(
+                bucket,
+                "multi-chunk/",
+                ObjectOptions {
+                    delete_prefix: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect_err("the first bounded child must request a successor batch");
+        assert!(
+            first_batch.to_string().contains("retry the next durable batch"),
+            "unexpected first child result: {first_batch}"
+        );
+
+        let newcomer = "multi-chunk/zzz-newcomer.bin";
+        let mut newcomer_reader = PutObjReader::from_vec(vec![b'n'; 1024 * 1024]);
+        let newcomer_source = store
+            .put_object(bucket, newcomer, &mut newcomer_reader, &ObjectOptions::default())
+            .await
+            .expect("a source created between chunks should be written");
+        store
+            .transition_object(
+                bucket,
+                newcomer,
+                &ObjectOptions {
+                    transition: TransitionOptions {
+                        status: TRANSITION_PENDING.to_string(),
+                        tier: tier_name.to_string(),
+                        etag: newcomer_source
+                            .etag
+                            .clone()
+                            .expect("the between-chunks source should have an etag"),
+                        ..Default::default()
+                    },
+                    mod_time: newcomer_source.mod_time,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("a source created between chunks should transition");
+
+        let mut completed = false;
+        let mut durable_batch_retries = 1;
+        for _ in 0..7 {
+            match store
+                .delete_object_with_tier_delete_journal(
+                    bucket,
+                    "multi-chunk/",
+                    ObjectOptions {
+                        delete_prefix: true,
+                        ..Default::default()
+                    },
+                )
+                .await
+            {
+                Ok(_) => {
+                    completed = true;
+                    break;
+                }
+                Err(err)
+                    if err.to_string().contains("retry the next durable batch")
+                        || err.to_string().contains("retry the next batch") =>
+                {
+                    durable_batch_retries += 1;
+                }
+                Err(err) => panic!("chunked prefix delete returned an unexpected error: {err}"),
+            }
+        }
+        assert!(completed, "six entries must converge through three bounded child batches");
+        assert_eq!(durable_batch_retries, 3, "limit two should require exactly three child batches");
+        assert_eq!(tier_delete_journal_count(store.clone()).await, 11);
+        assert_eq!(
+            backend.object_count().await,
+            11,
+            "remote cleanup must remain durable while the tier is unavailable"
+        );
+        let dispatch_records = store
+            .clone()
+            .list_objects_v2(
+                RUSTFS_META_BUCKET,
+                TIER_DELETE_DISPATCH_MANIFEST_PREFIX,
+                None,
+                None,
+                20,
+                false,
+                None,
+                false,
+            )
+            .await
+            .expect("bounded dispatch records should be listable");
+        let mut child_batches = 0;
+        let mut legacy_at_limit_seen = false;
+        for record in &dispatch_records.objects {
+            let data = com::read_config(store.clone(), &record.name)
+                .await
+                .expect("a retained dispatch record should be readable");
+            let value: serde_json::Value = serde_json::from_slice(&data).expect("a retained dispatch record should contain JSON");
+            if record.name.contains("/chunks/") {
+                let journal_count = value["journal_count"]
+                    .as_u64()
+                    .expect("a retained child should declare its journal count");
+                assert!(journal_count <= 2, "a child batch exceeded the configured resource bound");
+                child_batches += 1;
+            } else if value["prefix"] == "at-limit/" {
+                assert!(
+                    value.get("record_type").is_none(),
+                    "the exact-limit root must remain a legacy v1 manifest"
+                );
+                assert_eq!(value["journal_count"], 2);
+                legacy_at_limit_seen = true;
+            }
+        }
+        assert!(
+            legacy_at_limit_seen,
+            "the exact-limit dispatch must retain its byte-compatible root shape"
+        );
+        assert_eq!(child_batches, 5, "nine chunked sources should persist exactly five bounded children");
+
+        backend.set_remove_failure(false);
+        drive_tier_delete_dispatch_restart_to_convergence(store.clone()).await;
+        assert_eq!(tier_delete_journal_count(store.clone()).await, 0);
+        assert_eq!(tier_delete_dispatch_manifest_count(store.clone()).await, 0);
+        assert_eq!(backend.object_count().await, 0);
+        assert_eq!(
+            backend.remove_versions().await.len(),
+            11,
+            "each remote version must be removed exactly once"
+        );
+        shutdown.cancel();
+    }
+
+    #[cfg(feature = "test-util")]
+    #[test]
+    #[serial_test::serial(storage_class_env)]
+    fn tier_delete_chunk_crash_boundaries_resume_without_skipping_sources() {
+        run_large_stack_async_test(
+            "tier-delete-chunk-crash-boundaries",
+            tier_delete_chunk_crash_boundaries_resume_without_skipping_sources_case,
+        );
+    }
+
+    #[cfg(feature = "test-util")]
+    async fn tier_delete_chunk_crash_boundaries_resume_without_skipping_sources_case() {
+        let _batch_limit = TierDeleteDispatchBatchLimitGuard::install(1);
+        let stages = [
+            TierDeleteChunkTestStage::ParentPersisted,
+            TierDeleteChunkTestStage::ChildManifestPersisted,
+            TierDeleteChunkTestStage::ParentBound,
+            TierDeleteChunkTestStage::DispatchAuthorized,
+            TierDeleteChunkTestStage::LocalReplayCompleted,
+            TierDeleteChunkTestStage::ChildCompleted,
+            TierDeleteChunkTestStage::ParentProgressed,
+            TierDeleteChunkTestStage::FinalLocalDeletionCompleted,
+            TierDeleteChunkTestStage::ParentCompleted,
+        ];
+        for (case, stage) in stages.into_iter().enumerate() {
+            let temp_dir = tempfile::tempdir().expect("create chunk crash-boundary store dir");
+            let initial_name = format!("chunk-crash-boundary-{case}");
+            let (ctx, store, shutdown) =
+                without_storage_class_env(build_isolated_test_store(temp_dir.path(), &initial_name, &[4])).await;
+            crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
+            let tier_name = format!("CHUNK-CRASH-BOUNDARY-{case}");
+            let backend = register_mock_tier(&ctx.tier_config_mgr(), &tier_name).await;
+            backend.set_remove_failure(true);
+            let bucket = format!("chunk-crash-boundary-{case}-bucket");
+            let prefix = "prefix/";
+            store
+                .make_bucket(&bucket, &MakeBucketOptions::default())
+                .await
+                .expect("chunk crash-boundary bucket should be created");
+            for index in 0..2 {
+                let object = format!("{prefix}object-{index}.bin");
+                let mut reader = PutObjReader::from_vec(vec![b'a' + index as u8; 1024 * 1024]);
+                let source = store
+                    .put_object(&bucket, &object, &mut reader, &ObjectOptions::default())
+                    .await
+                    .expect("chunk crash-boundary source should be written");
+                store
+                    .transition_object(
+                        &bucket,
+                        &object,
+                        &ObjectOptions {
+                            transition: TransitionOptions {
+                                status: TRANSITION_PENDING.to_string(),
+                                tier: tier_name.clone(),
+                                etag: source.etag.clone().expect("chunk crash-boundary source should have an etag"),
+                                ..Default::default()
+                            },
+                            mod_time: source.mod_time,
+                            ..Default::default()
+                        },
+                    )
+                    .await
+                    .expect("chunk crash-boundary source should transition");
+            }
+            let local_only = format!("{prefix}local-only.bin");
+            let mut local_reader = PutObjReader::from_vec(vec![b'l'; 1024 * 1024]);
+            store
+                .put_object(&bucket, &local_only, &mut local_reader, &ObjectOptions::default())
+                .await
+                .expect("chunk crash-boundary local-only object should be written");
+            let tier_config = ctx
+                .tier_config_mgr()
+                .read()
+                .await
+                .tiers
+                .get(&tier_name)
+                .expect("chunk crash-boundary tier config should remain available for restart")
+                .clone_with_credentials();
+
+            if matches!(
+                stage,
+                TierDeleteChunkTestStage::FinalLocalDeletionCompleted | TierDeleteChunkTestStage::ParentCompleted
+            ) {
+                for _ in 0..2 {
+                    let retry = store
+                        .delete_object_with_tier_delete_journal(
+                            &bucket,
+                            prefix,
+                            ObjectOptions {
+                                delete_prefix: true,
+                                ..Default::default()
+                            },
+                        )
+                        .await
+                        .expect_err("each bounded child must complete before the final crash boundary");
+                    assert!(
+                        retry.to_string().contains("retry the next durable batch"),
+                        "unexpected bounded-child result before {stage:?}: {retry}"
+                    );
+                }
+            }
+
+            let barrier = TierDeleteChunkTestBarrier::install(stage);
+            let worker_store = store.clone();
+            let worker_bucket = bucket.clone();
+            let worker = tokio::spawn(async move {
+                worker_store
+                    .delete_object_with_tier_delete_journal(
+                        &worker_bucket,
+                        prefix,
+                        ObjectOptions {
+                            delete_prefix: true,
+                            ..Default::default()
+                        },
+                    )
+                    .await
+            });
+            tokio::time::timeout(Duration::from_secs(30), barrier.wait_until_paused())
+                .await
+                .unwrap_or_else(|_| panic!("chunk delete did not reach crash boundary {stage:?}"));
+            worker.abort();
+            let _ = worker.await;
+            drop(barrier);
+            let released_bucket_guard =
+                tokio::time::timeout(Duration::from_secs(5), store.acquire_bucket_lifecycle_write_lock(&bucket))
+                    .await
+                    .unwrap_or_else(|_| panic!("canceling at {stage:?} did not release the bucket lifecycle lock"))
+                    .unwrap_or_else(|err| {
+                        panic!("bucket lifecycle lock reacquire failed after cancellation at {stage:?}: {err}")
+                    });
+            drop(released_bucket_guard);
+            shutdown.cancel();
+            drop(store);
+            drop(ctx);
+
+            let restarted_name = format!("chunk-crash-boundary-{case}-restart");
+            let (restarted_ctx, restarted_store, restarted_shutdown) =
+                without_storage_class_env(build_isolated_test_store(temp_dir.path(), &restarted_name, &[4])).await;
+            {
+                let tier_config_mgr = restarted_ctx.tier_config_mgr();
+                let mut manager = tier_config_mgr.write().await;
+                manager.tiers.insert(tier_name.clone(), tier_config);
+                manager
+                    .install_test_driver(&tier_name, Box::new(backend.clone()))
+                    .expect("the exact chunk crash-boundary tier driver should reinstall after restart");
+            }
+            crate::bucket::metadata_sys::init_bucket_metadata_sys(restarted_store.clone(), Vec::new()).await;
+
+            let mut completed = false;
+            for _ in 0..20 {
+                let recovery = recover_tier_delete_dispatch_manifests(restarted_store.clone(), 100, None)
+                    .await
+                    .expect("chunk crash-boundary manifest recovery should remain readable");
+                assert_eq!(recovery.failed, 0, "recovery must not quarantine a valid chunk boundary");
+                match restarted_store
+                    .delete_object_with_tier_delete_journal(
+                        &bucket,
+                        prefix,
+                        ObjectOptions {
+                            delete_prefix: true,
+                            ..Default::default()
+                        },
+                    )
+                    .await
+                {
+                    Ok(_) => {
+                        completed = true;
+                        break;
+                    }
+                    Err(err)
+                        if err.to_string().contains("retry")
+                            || err.to_string().contains("rollback")
+                            || err.to_string().contains("durable cleanup") => {}
+                    Err(err) => panic!("chunk crash-boundary retry returned an unexpected error at {stage:?}: {err}"),
+                }
+            }
+            assert!(completed, "chunk state must converge after cancellation at {stage:?}");
+            for index in 0..2 {
+                let object = format!("{prefix}object-{index}.bin");
+                assert!(
+                    restarted_store.pools[0]
+                        .get_disks_by_key(&object)
+                        .load_file_info_versions_exact(&bucket, &object)
+                        .await
+                        .expect("chunk crash-boundary source lookup should succeed")
+                        .is_none(),
+                    "source {object} must not be skipped after cancellation at {stage:?}"
+                );
+            }
+            assert!(
+                restarted_store.pools[0]
+                    .get_disks_by_key(&local_only)
+                    .load_file_info_versions_exact(&bucket, &local_only)
+                    .await
+                    .expect("chunk crash-boundary local-only source lookup should succeed")
+                    .is_none(),
+                "the final raw delete must remove the local-only source after cancellation at {stage:?}"
+            );
+            assert_eq!(backend.object_count().await, 2);
+            backend.set_remove_failure(false);
+            drive_tier_delete_dispatch_restart_to_convergence(restarted_store.clone()).await;
+            assert_eq!(tier_delete_journal_count(restarted_store.clone()).await, 0);
+            assert_eq!(tier_delete_dispatch_manifest_count(restarted_store.clone()).await, 0);
+            assert_eq!(backend.object_count().await, 0);
+            assert_eq!(
+                backend.remove_versions().await.len(),
+                2,
+                "each crash case must retain exactly one cleanup owner per remote version"
+            );
+            restarted_shutdown.cancel();
+        }
+    }
+
+    #[cfg(feature = "test-util")]
+    #[test]
+    #[serial_test::serial(storage_class_env)]
+    fn tier_delete_chunk_missing_child_with_journals_fails_closed() {
+        run_large_stack_async_test(
+            "tier-delete-chunk-missing-child",
+            tier_delete_chunk_missing_child_with_journals_fails_closed_case,
+        );
+    }
+
+    #[cfg(feature = "test-util")]
+    async fn tier_delete_chunk_missing_child_with_journals_fails_closed_case() {
+        let _batch_limit = TierDeleteDispatchBatchLimitGuard::install(1);
+        let temp_dir = tempfile::tempdir().expect("create missing-child store dir");
+        let (ctx, store, shutdown) =
+            without_storage_class_env(build_isolated_test_store(temp_dir.path(), "chunk-missing-child", &[4])).await;
+        crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
+        let tier_name = "CHUNK-MISSING-CHILD";
+        let backend = register_mock_tier(&ctx.tier_config_mgr(), tier_name).await;
+        backend.set_remove_failure(true);
+        let bucket = "chunk-missing-child-bucket";
+        let prefix = "prefix/";
+        store
+            .make_bucket(bucket, &MakeBucketOptions::default())
+            .await
+            .expect("missing-child bucket should be created");
+        for index in 0..2 {
+            let object = format!("{prefix}object-{index}.bin");
+            let mut reader = PutObjReader::from_vec(vec![b'm' + index as u8; 1024 * 1024]);
+            let source = store
+                .put_object(bucket, &object, &mut reader, &ObjectOptions::default())
+                .await
+                .expect("missing-child source should be written");
+            store
+                .transition_object(
+                    bucket,
+                    &object,
+                    &ObjectOptions {
+                        transition: TransitionOptions {
+                            status: TRANSITION_PENDING.to_string(),
+                            tier: tier_name.to_string(),
+                            etag: source.etag.clone().expect("missing-child source should have an etag"),
+                            ..Default::default()
+                        },
+                        mod_time: source.mod_time,
+                        ..Default::default()
+                    },
+                )
+                .await
+                .expect("missing-child source should transition");
+        }
+
+        let barrier = TierDeleteChunkTestBarrier::install(TierDeleteChunkTestStage::DispatchAuthorized);
+        let worker_store = store.clone();
+        let worker = tokio::spawn(async move {
+            worker_store
+                .delete_object_with_tier_delete_journal(
+                    bucket,
+                    prefix,
+                    ObjectOptions {
+                        delete_prefix: true,
+                        ..Default::default()
+                    },
+                )
+                .await
+        });
+        tokio::time::timeout(Duration::from_secs(30), barrier.wait_until_paused())
+            .await
+            .expect("chunk delete should persist its authorization before corruption injection");
+        worker.abort();
+        let _ = worker.await;
+        drop(barrier);
+
+        let records = store
+            .clone()
+            .list_objects_v2(
+                RUSTFS_META_BUCKET,
+                TIER_DELETE_DISPATCH_MANIFEST_PREFIX,
+                None,
+                None,
+                10,
+                false,
+                None,
+                false,
+            )
+            .await
+            .expect("parent and child records should be listable");
+        let child = records
+            .objects
+            .iter()
+            .find(|object| object.name.contains("/chunks/"))
+            .expect("the bound child manifest should exist")
+            .name
+            .clone();
+        com::delete_config(store.clone(), &child)
+            .await
+            .expect("the test should remove only the child manifest");
+        assert_eq!(tier_delete_journal_count(store.clone()).await, 1);
+
+        let error = store
+            .delete_object_with_tier_delete_journal(
+                bucket,
+                prefix,
+                ObjectOptions {
+                    delete_prefix: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect_err("a missing child with retained journals must quarantine the parent");
+        assert!(
+            error.to_string().contains("missing child with retained journals"),
+            "unexpected missing-child result: {error}"
+        );
+        assert_eq!(backend.object_count().await, 2, "fail-closed inspection must not remove a remote version");
+        shutdown.cancel();
+    }
+
     #[cfg(feature = "test-util")]
     #[test]
     #[serial_test::serial(storage_class_env)]
@@ -11647,6 +12254,7 @@ mod tests {
 
     #[cfg(feature = "test-util")]
     async fn authorized_prefix_retry_replays_predecessor_before_newcomer_case() {
+        let _batch_limit = TierDeleteDispatchBatchLimitGuard::install(1);
         let temp_dir = tempfile::tempdir().expect("create authorized predecessor replay store dir");
         let (ctx, store, _shutdown) =
             without_storage_class_env(build_isolated_test_store(temp_dir.path(), "authorized-prefix-predecessor-replay", &[4]))
@@ -11659,8 +12267,8 @@ mod tests {
             .expect("authorized replay tier lease should resolve");
         let bucket = "authorized-prefix-predecessor-replay-bucket";
         let prefix = "prefix/";
-        let predecessor = "prefix/predecessor.bin";
-        let newcomer = "prefix/newcomer.bin";
+        let predecessor = "prefix/000-predecessor.bin";
+        let newcomer = "prefix/zzz-newcomer.bin";
         store
             .make_bucket(bucket, &MakeBucketOptions::default())
             .await

crates/ecstore/src/store/object.rs

@@ -18,7 +18,9 @@ use crate::bucket::lifecycle::{
     get_expiry_configs,
     tier_delete_journal::{
         ActiveTierDeleteDispatch, EVENT_LIFECYCLE_TIER_DELETE_JOURNAL, LOG_COMPONENT_ECSTORE, LOG_SUBSYSTEM_LIFECYCLE,
-        complete_tier_delete_dispatch, prepare_tier_delete_dispatch, record_tier_delete_journal_backend_identity,
+        TierDeleteChunkParentInspection, complete_tier_delete_chunk_parent, complete_tier_delete_dispatch,
+        inspect_tier_delete_chunk_parent, prepare_tier_delete_chunk_dispatch, prepare_tier_delete_dispatch,
+        record_tier_delete_journal_backend_identity, resume_tier_delete_dispatch, tier_delete_dispatch_batch_limit,
         tier_delete_journal_object_name, tier_delete_source_matches_dispatch_scope,
     },
     tier_sweeper::{
@@ -41,7 +43,10 @@ use crate::object_api::{
     NamespaceLockFence, ObjectLockConfigSnapshot, ScannerPublicationCommitScopeGuard, ScannerPublicationCommitState,
     TierFreeVersionReceiptSink,
 };
-use crate::services::notification_sys::acquire_tier_delete_journal_fleet_proof;
+use crate::services::notification_sys::{
+    TierDeleteJournalFleetProofToken, acquire_tier_delete_journal_fleet_proof, tier_delete_journal_fleet_proof_matches,
+    tier_delete_journal_topology_generation,
+};
 use crate::services::tier::tier::{TierConfigMgr, TierDestinationId, TierOperationLease, tier_destination_id_from_metadata};
 use crate::set_disk::{
     SetDisks, get_lock_acquire_timeout, get_object_lock_diag_slow_acquire_threshold, get_object_lock_diag_slow_hold_threshold,
@@ -52,6 +57,7 @@ use crate::storage_api_contracts::{
     namespace::NamespaceLocking as _,
     object::{DeleteAccounting, ObjectIO as _, ObjectOperations as _},
 };
+use futures::StreamExt as _;
 use parking_lot::Mutex as ParkingMutex;
 use rustfs_filemeta::ObjectPartInfo;
 use rustfs_io_metrics::{
@@ -73,6 +79,7 @@ const RECURSIVE_DELETE_VERSION_SCAN_PAGE_SIZE: i32 = 1000;
 const RECURSIVE_DELETE_VERSION_SCAN_PAGE_SIZE: i32 = 2;
 const RESTORE_WORKER_LOCK_PREFIX: &str = "ilm/restore-worker-locks";
 const RESTORE_WORKER_LOCK_PROBE_TIMEOUT: Duration = Duration::from_millis(50);
+const TIER_DELETE_DISPATCH_LOCAL_REPLAY_CONCURRENCY: usize = 16;
 
 fn install_tier_free_version_receipt_sink(opts: &mut ObjectOptions) -> Option<TierFreeVersionReceiptSink> {
     if opts.tier_free_version_receipt_sink.is_some() || opts.skip_free_version || opts.delete_prefix {
@@ -147,14 +154,96 @@ async fn prepare_prefix_tier_delete_journal_entries(
     Box::pin(prepare_prefix_tier_delete_journal_entries_inner(api, bucket, prefix, opts)).await
 }
 
+type TierDeleteLeaseReference = (String, Option<TierDestinationId>);
+
+fn tier_delete_walk_cancellation_is_expected(truncated: bool, limit_cancellation: bool, error: &Error) -> bool {
+    truncated && limit_cancellation && matches!(error, Error::OperationCanceled)
+}
+
+fn combine_tier_delete_walk_results(results: impl IntoIterator<Item = Result<()>>) -> Result<()> {
+    let mut cancelled = false;
+    for result in results {
+        match result {
+            Ok(()) => {}
+            Err(Error::OperationCanceled) => cancelled = true,
+            Err(err) => return Err(err),
+        }
+    }
+    if cancelled { Err(Error::OperationCanceled) } else { Ok(()) }
+}
+
+async fn acquire_prefix_tier_delete_reference_leases(
+    api: &Arc<ECStore>,
+    tier_references: &std::collections::HashSet<TierDeleteLeaseReference>,
+) -> Result<Vec<TierOperationLease>> {
+    let mut tier_references = tier_references.iter().cloned().collect::<Vec<_>>();
+    tier_references.sort_unstable();
+    let mut leases = Vec::with_capacity(tier_references.len());
+    for (tier_name, backend_identity) in tier_references {
+        let lease = match backend_identity {
+            Some(backend_identity) => {
+                TierConfigMgr::acquire_operation_lease_for_backend_identity(&api.tier_config_mgr(), &tier_name, backend_identity)
+                    .await
+            }
+            None => TierConfigMgr::acquire_operation_lease(&api.tier_config_mgr(), &tier_name).await,
+        }
+        .map_err(Error::other)?;
+        leases.push(lease);
+    }
+    Ok(leases)
+}
+
+async fn acquire_prefix_tier_delete_leases(api: &Arc<ECStore>, entries: &[Jentry]) -> Result<Vec<TierOperationLease>> {
+    let tier_references = entries
+        .iter()
+        .map(|entry| (entry.tier_name.clone(), entry.backend_identity))
+        .collect::<std::collections::HashSet<_>>();
+    acquire_prefix_tier_delete_reference_leases(api, &tier_references).await
+}
+
 async fn prepare_prefix_tier_delete_journal_entries_inner(
     api: &Arc<ECStore>,
     bucket: &str,
     prefix: &str,
     opts: &ObjectOptions,
 ) -> Result<PreparedPrefixTierDelete> {
+    let (chunk_parent_active, legacy_manifest_active, chunk_parent_topology_generation) = if is_meta_bucketname(bucket) {
+        (false, false, None)
+    } else {
+        let bucket_incarnation = opts.expected_bucket_incarnation_id.ok_or(StorageError::PreconditionFailed)?;
+        let bucket_fence = opts
+            .bucket_lifecycle_lock_fence
+            .as_ref()
+            .ok_or_else(|| Error::other("tier delete dispatch requires a bucket lifecycle write fence"))?;
+        match Box::pin(inspect_tier_delete_chunk_parent(
+            Arc::clone(api),
+            bucket,
+            bucket_incarnation,
+            prefix,
+            bucket_fence,
+        ))
+        .await?
+        {
+            TierDeleteChunkParentInspection::NoParent => (false, false, None),
+            TierDeleteChunkParentInspection::LegacyManifest => (false, true, None),
+            TierDeleteChunkParentInspection::Ready(topology_generation) => (true, false, Some(topology_generation)),
+            TierDeleteChunkParentInspection::Resume(dispatch) => {
+                let leases = acquire_prefix_tier_delete_leases(api, dispatch.entries()?).await?;
+                return Ok(PreparedPrefixTierDelete {
+                    dispatch: Some(*dispatch),
+                    chunk_parent_active: true,
+                    chunk_parent_fleet_proof: None,
+                    _leases: leases,
+                });
+            }
+            TierDeleteChunkParentInspection::RetryRequired => {
+                return Err(Error::other("tier delete chunk parent made durable progress; retry the next batch"));
+            }
+        }
+    };
     let mut tier_references = std::collections::HashSet::<(String, Option<TierDestinationId>)>::new();
     let mut entries_by_name = std::collections::BTreeMap::new();
+    let batch_limit = tier_delete_dispatch_batch_limit();
     let logical_prefix = decode_dir_object(prefix);
     let exact_object = opts.delete_prefix_object.then(|| logical_prefix.clone());
     let physical_sets = api
@@ -164,12 +253,11 @@ async fn prepare_prefix_tier_delete_journal_entries_inner(
         .collect::<Vec<_>>();
     let (tx, mut rx) = tokio::sync::mpsc::channel::<ObjectInfoOrErr>(100);
     let cancellation = tokio_util::sync::CancellationToken::new();
+    let limit_cancellation = Arc::new(AtomicBool::new(false));
     let walk_cancel = cancellation.clone();
     let bucket_owned = bucket.to_string();
     let prefix_owned = prefix.to_string();
     let walk = async move {
-        use futures::StreamExt as _;
-
         let results = futures::stream::iter(physical_sets.into_iter().map(|set| {
             let tx = tx.clone();
             let cancellation = walk_cancel.clone();
@@ -199,14 +287,28 @@ async fn prepare_prefix_tier_delete_journal_entries_inner(
         .collect::<Vec<_>>()
         .await;
         drop(tx);
-        results.into_iter().collect::<Result<Vec<_>>>().map(|_| ())
+        combine_tier_delete_walk_results(results)
     };
+    let collect_limit_cancellation = limit_cancellation.clone();
     let collect = async {
+        let mut truncated = false;
         while let Some(result) = rx.recv().await {
             if let Some(err) = result.err {
+                // Once limit + 1 has been observed this request can authorize
+                // only the exact retained batch; it cannot infer prefix
+                // absence or run the raw delete. Drain only the explicit
+                // cancellation fallout; a real walker error must still fail
+                // the request even when another set reached the limit first.
+                if tier_delete_walk_cancellation_is_expected(truncated, collect_limit_cancellation.load(Ordering::Acquire), &err)
+                {
+                    continue;
+                }
                 cancellation.cancel();
                 return Err(err);
             }
+            if truncated {
+                continue;
+            }
             let Some(source) = result.item else {
                 continue;
             };
@@ -223,62 +325,125 @@ async fn prepare_prefix_tier_delete_journal_entries_inner(
                     "recursive prefix delete cannot discard an existing tier free-version cleanup obligation",
                 ));
             }
-            if source.transitioned_object.status == rustfs_filemeta::TRANSITION_COMPLETE {
+            let tier_reference = if source.transitioned_object.status == rustfs_filemeta::TRANSITION_COMPLETE {
                 let backend_identity = tier_destination_id_from_metadata(&source.user_defined).map_err(Error::other)?;
-                tier_references.insert((source.transitioned_object.tier.clone(), backend_identity));
-            }
+                Some((source.transitioned_object.tier.clone(), backend_identity))
+            } else {
+                None
+            };
             if let Some(entry) = build_tier_delete_journal_entry(bucket, &object, opts, &source)? {
-                entries_by_name
-                    .entry(tier_delete_journal_object_name(&entry))
-                    .or_insert(entry);
+                let name = tier_delete_journal_object_name(&entry);
+                let at_limit = entries_by_name.len() == batch_limit;
+                match entries_by_name.entry(name) {
+                    std::collections::btree_map::Entry::Occupied(_) => {}
+                    std::collections::btree_map::Entry::Vacant(_) if at_limit => {
+                        truncated = true;
+                        collect_limit_cancellation.store(true, Ordering::Release);
+                        cancellation.cancel();
+                    }
+                    std::collections::btree_map::Entry::Vacant(slot) => {
+                        if let Some(tier_reference) = tier_reference {
+                            tier_references.insert(tier_reference);
+                        }
+                        slot.insert(entry);
+                    }
+                }
+            } else if let Some(tier_reference) = tier_reference {
+                tier_references.insert(tier_reference);
             }
         }
-        Ok(())
+        Ok(truncated)
     };
     let (walk_result, collect_result) = tokio::join!(walk, collect);
-    collect_result?;
-    walk_result?;
-    let entries = entries_by_name.into_values().collect::<Vec<_>>();
-
-    let mut tier_references = tier_references.into_iter().collect::<Vec<_>>();
-    tier_references.sort_unstable();
-    let mut leases = Vec::with_capacity(tier_references.len());
-    for (tier_name, backend_identity) in tier_references {
-        let lease = match backend_identity {
-            Some(backend_identity) => {
-                TierConfigMgr::acquire_operation_lease_for_backend_identity(&api.tier_config_mgr(), &tier_name, backend_identity)
-                    .await
-            }
-            None => TierConfigMgr::acquire_operation_lease(&api.tier_config_mgr(), &tier_name).await,
-        }
-        .map_err(Error::other)?;
-        leases.push(lease);
+    let truncated = collect_result?;
+    // A truncated walk normally reports OperationCanceled from the physical
+    // walkers. That cancellation is expected; any other result is a genuine
+    // scan failure and cannot be hidden by the bounded batch.
+    if let Err(err) = walk_result
+        && !tier_delete_walk_cancellation_is_expected(truncated, limit_cancellation.load(Ordering::Acquire), &err)
+    {
+        return Err(err);
     }
+    let entries = entries_by_name.into_values().collect::<Vec<_>>();
+    let mut leased_tier_references = tier_references;
+    let mut leases = acquire_prefix_tier_delete_reference_leases(api, &leased_tier_references).await?;
     if entries.is_empty() {
+        let chunk_parent_fleet_proof = if let Some(expected_topology) = chunk_parent_topology_generation.as_deref() {
+            let fleet_proof = acquire_tier_delete_journal_fleet_proof()
+                .ok_or_else(|| Error::other("tier delete chunk parent fleet capability is unavailable"))?;
+            if tier_delete_journal_topology_generation(&fleet_proof) != expected_topology {
+                return Err(Error::other("tier delete chunk parent topology changed during final source scan"));
+            }
+            Some(fleet_proof)
+        } else {
+            None
+        };
         return Ok(PreparedPrefixTierDelete {
             dispatch: None,
+            chunk_parent_active,
+            chunk_parent_fleet_proof,
             _leases: leases,
         });
     }
 
     let bucket_incarnation = opts.expected_bucket_incarnation_id.ok_or(StorageError::PreconditionFailed)?;
-    let fleet_proof = acquire_tier_delete_journal_fleet_proof()
-        .ok_or_else(|| Error::other("tier delete journal v6 fleet capability is unavailable"))?;
     let bucket_fence = opts
         .bucket_lifecycle_lock_fence
         .as_ref()
         .ok_or_else(|| Error::other("tier delete dispatch requires a bucket lifecycle write fence"))?;
-    let dispatch =
+    let fleet_proof = acquire_tier_delete_journal_fleet_proof()
+        .ok_or_else(|| Error::other("tier delete journal v6 fleet capability is unavailable"))?;
+    if chunk_parent_topology_generation
+        .as_deref()
+        .is_some_and(|expected| tier_delete_journal_topology_generation(&fleet_proof) != expected)
+    {
+        return Err(Error::other("tier delete chunk parent topology changed during source scan"));
+    }
+    let mut dispatch = if !legacy_manifest_active && (chunk_parent_active || truncated) {
+        Box::pin(prepare_tier_delete_chunk_dispatch(
+            Arc::clone(api),
+            bucket,
+            bucket_incarnation,
+            prefix,
+            entries,
+            truncated && !chunk_parent_active,
+            fleet_proof,
+            bucket_fence,
+        ))
+        .await?
+    } else if legacy_manifest_active && truncated {
+        resume_tier_delete_dispatch(Arc::clone(api), bucket, bucket_incarnation, prefix, entries, fleet_proof, bucket_fence)
+            .await?
+    } else {
         prepare_tier_delete_dispatch(Arc::clone(api), bucket, bucket_incarnation, prefix, entries, fleet_proof, bucket_fence)
-            .await?;
+            .await?
+    };
+    // A resumed legacy authorization may own predecessors that are absent
+    // from this bounded scan. Pin every backend generation in the actual
+    // permit before any local mutation, while avoiding duplicate leases for
+    // entries already covered by the scan.
+    let additional_tier_references = dispatch
+        .entries()?
+        .iter()
+        .map(|entry| (entry.tier_name.clone(), entry.backend_identity))
+        .filter(|reference| leased_tier_references.insert(reference.clone()))
+        .collect::<std::collections::HashSet<_>>();
+    leases.extend(acquire_prefix_tier_delete_reference_leases(api, &additional_tier_references).await?);
+    if legacy_manifest_active && truncated {
+        dispatch.require_exact_predecessor_replay();
+    }
     Ok(PreparedPrefixTierDelete {
         dispatch: Some(dispatch),
+        chunk_parent_active: chunk_parent_active || truncated,
+        chunk_parent_fleet_proof: None,
         _leases: leases,
     })
 }
 
 struct PreparedPrefixTierDelete {
     dispatch: Option<crate::bucket::lifecycle::tier_delete_journal::PreparedTierDeleteDispatch>,
+    chunk_parent_active: bool,
+    chunk_parent_fleet_proof: Option<TierDeleteJournalFleetProofToken>,
     _leases: Vec<TierOperationLease>,
 }
 
@@ -365,12 +530,16 @@ async fn delete_prefix_with_tier_delete_journal(
     let Some(api) = tier_journal_api else {
         return store.delete_prefix(bucket, object, opts).await;
     };
-    let PreparedPrefixTierDelete { dispatch, _leases } =
-        prepare_prefix_tier_delete_journal_entries(api, bucket, object, opts).await?;
+    let PreparedPrefixTierDelete {
+        dispatch,
+        chunk_parent_active,
+        chunk_parent_fleet_proof,
+        _leases,
+    } = prepare_prefix_tier_delete_journal_entries(api, bucket, object, opts).await?;
     let Some(dispatch) = dispatch else {
-        // There is no remote-cleanup candidate, so no v6 manifest or fleet
-        // proof is required.  Keep any compatibility-path tier leases alive
-        // until the local delete has committed.
+        // There is no new remote-cleanup candidate. Keep compatibility-path
+        // tier leases and, for a chunked final pass, the matching parent fleet
+        // proof alive until local deletion and parent completion both commit.
         let _tier_leases = _leases;
         let mut operation_opts = opts.clone();
         // `tier_delete_journal_api` means a v6 dispatch authorization must be
@@ -379,7 +548,42 @@ async fn delete_prefix_with_tier_delete_journal(
         // transitioned metadata retains its FreeVersion fallback.
         operation_opts.tier_delete_journal_api = None;
         operation_opts.tier_delete_dispatch_authorization = None;
-        return store.delete_prefix(bucket, object, &operation_opts).await;
+        let parent_fleet_proof = if chunk_parent_active {
+            Some(
+                chunk_parent_fleet_proof
+                    .as_ref()
+                    .filter(|proof| tier_delete_journal_fleet_proof_matches(proof))
+                    .ok_or_else(|| Error::other("tier delete chunk parent fleet proof changed before final deletion"))?,
+            )
+        } else {
+            None
+        };
+        store.delete_prefix(bucket, object, &operation_opts).await?;
+        if let Some(parent_fleet_proof) = parent_fleet_proof {
+            #[cfg(all(test, feature = "test-util"))]
+            crate::bucket::lifecycle::tier_delete_journal::tier_delete_chunk_test_pause(
+                crate::bucket::lifecycle::tier_delete_journal::TierDeleteChunkTestStage::FinalLocalDeletionCompleted,
+            )
+            .await;
+            let bucket_incarnation = opts.expected_bucket_incarnation_id.ok_or(StorageError::PreconditionFailed)?;
+            let bucket_fence = opts
+                .bucket_lifecycle_lock_fence
+                .as_ref()
+                .ok_or_else(|| Error::other("tier delete dispatch requires a bucket lifecycle write fence"))?;
+            if !Box::pin(complete_tier_delete_chunk_parent(
+                Arc::clone(api),
+                bucket,
+                bucket_incarnation,
+                object,
+                bucket_fence,
+                parent_fleet_proof,
+            ))
+            .await?
+            {
+                return Err(Error::other("tier delete chunk parent disappeared after final local deletion"));
+            }
+        }
+        return Ok(());
     };
     let bucket_incarnation = opts.expected_bucket_incarnation_id.ok_or(StorageError::PreconditionFailed)?;
     let bucket_fence = opts
@@ -396,9 +600,19 @@ async fn delete_prefix_with_tier_delete_journal(
     // Keep every backend generation lease until the whole local operation has
     // either committed its journal set or returned an ambiguous mutation.
     let _tier_leases = _leases;
-    if active.predecessor_replay_required() {
+    if active.predecessor_replay_required() || active.is_chunked() {
         replay_authorized_tier_delete_sources(store, bucket, object, &active, &operation_opts).await?;
+        #[cfg(all(test, feature = "test-util"))]
+        if active.is_chunked() {
+            crate::bucket::lifecycle::tier_delete_journal::tier_delete_chunk_test_pause(
+                crate::bucket::lifecycle::tier_delete_journal::TierDeleteChunkTestStage::LocalReplayCompleted,
+            )
+            .await;
+        }
         complete_tier_delete_dispatch(Arc::clone(api), &active, bucket_fence).await?;
+        if active.is_chunked() {
+            return Err(Error::other("tier delete chunk completed; retry the next durable batch"));
+        }
         return Err(Error::other("authorized tier delete predecessor completed; retry the successor dispatch"));
     }
     let result = store.delete_prefix(bucket, object, &operation_opts).await;
@@ -443,7 +657,7 @@ async fn replay_authorized_tier_delete_sources(
     let authorization = active.authorization();
     authorization.mark_mutation_started(bucket, bucket_incarnation, prefix)?;
 
-    let mut source_objects = std::collections::BTreeSet::new();
+    let mut source_objects = std::collections::HashSet::with_capacity(active.entries().len());
     for entry in active.entries() {
         let source = entry
             .source
@@ -453,61 +667,130 @@ async fn replay_authorized_tier_delete_sources(
         if !tier_delete_source_matches_replay_scope(source, bucket, prefix, opts.delete_prefix_object) {
             return Err(Error::other("authorized tier delete predecessor source escaped its prefix scope"));
         }
-        source_objects.insert(source.object.clone());
+        source_objects.insert(source.object.as_str());
     }
 
-    let mut deleted = 0;
-    for object in source_objects {
-        if bucket_fence.is_lock_lost() {
-            return Err(Error::other("tier delete dispatch namespace fence was lost during predecessor replay"));
+    if let Some(scope) = publication_scope {
+        if scope.state() == ScannerPublicationCommitState::Admitted {
+            scope
+                .try_begin()
+                .map_err(|_| Error::other("scanner publication predecessor replay scope cannot start"))?;
         }
-        let encoded_object = encode_dir_object(&object);
-        let guards = if opts.delete_prefix_object {
-            store
-                .acquire_remaining_physical_object_write_locks("tier_delete_dispatch_predecessor_replay", bucket, &encoded_object)
-                .await?
-        } else {
-            store
-                .acquire_all_physical_object_write_locks("tier_delete_dispatch_predecessor_replay", bucket, &encoded_object)
-                .await?
-        };
-        authorization.ensure_current(bucket, bucket_incarnation, prefix)?;
-        if let Some(scope) = publication_scope {
-            if scope.state() == ScannerPublicationCommitState::Admitted {
-                scope
-                    .try_begin()
-                    .map_err(|_| Error::other("scanner publication predecessor replay scope cannot start"))?;
-            }
-            if !scope.can_commit() {
-                let _ = scope.mark_indeterminate();
-                return Err(StorageError::OperationCanceled);
-            }
-        }
-        let mut replay_opts = opts.clone();
-        replay_opts.no_lock = true;
-        replay_opts.delete_prefix = false;
-        replay_opts.delete_prefix_object = false;
-        for guard in &guards {
-            guard.add_namespace_lock_fence(&mut replay_opts);
-        }
-        for pool in &store.pools {
-            for set in &pool.disk_set {
-                authorization.ensure_current(bucket, bucket_incarnation, prefix)?;
-                deleted += set
-                    .replay_authorized_tier_delete_sources(bucket, &object, &authorization, &replay_opts)
-                    .await?;
-            }
-        }
-        if bucket_fence.is_lock_lost() || guards.iter().any(ObjectLockDiagGuard::is_lock_lost) {
-            return Err(Error::other("tier delete dispatch namespace fence was lost during predecessor replay"));
+        if !scope.can_commit() {
+            let _ = scope.mark_indeterminate();
+            return Err(StorageError::OperationCanceled);
         }
     }
-    if let Some(scope) = publication_scope {
-        let _ = scope.mark_committed();
+
+    let stopped = Arc::new(AtomicBool::new(false));
+    // The caller holds bucket lifecycle WRITE. Each bounded future acquires
+    // only one logical object's physical lock set and releases it before
+    // completion; no future nests locks for two object keys.
+    let make_replay = |object: String| {
+        let stopped = stopped.clone();
+        let authorization = authorization.clone();
+        async move {
+            if stopped.load(Ordering::Acquire) {
+                return Ok::<_, Error>(0usize);
+            }
+            let result = async {
+                if bucket_fence.is_lock_lost() {
+                    return Err(Error::other("tier delete dispatch namespace fence was lost during predecessor replay"));
+                }
+                let encoded_object = encode_dir_object(&object);
+                let guards = if opts.delete_prefix_object {
+                    store
+                        .acquire_remaining_physical_object_write_locks(
+                            "tier_delete_dispatch_predecessor_replay",
+                            bucket,
+                            &encoded_object,
+                        )
+                        .await?
+                } else {
+                    store
+                        .acquire_all_physical_object_write_locks(
+                            "tier_delete_dispatch_predecessor_replay",
+                            bucket,
+                            &encoded_object,
+                        )
+                        .await?
+                };
+                authorization.ensure_current(bucket, bucket_incarnation, prefix)?;
+                if publication_scope.is_some_and(|scope| !scope.can_commit()) {
+                    return Err(StorageError::OperationCanceled);
+                }
+                let mut replay_opts = opts.clone();
+                replay_opts.no_lock = true;
+                replay_opts.delete_prefix = false;
+                replay_opts.delete_prefix_object = false;
+                for guard in &guards {
+                    guard.add_namespace_lock_fence(&mut replay_opts);
+                }
+                let mut deleted = 0usize;
+                for pool in &store.pools {
+                    for set in &pool.disk_set {
+                        authorization.ensure_current(bucket, bucket_incarnation, prefix)?;
+                        deleted = deleted
+                            .checked_add(
+                                set.replay_authorized_tier_delete_sources(bucket, &object, &authorization, &replay_opts)
+                                    .await?,
+                            )
+                            .ok_or_else(|| Error::other("tier delete dispatch replay count overflow"))?;
+                    }
+                }
+                if bucket_fence.is_lock_lost() || guards.iter().any(ObjectLockDiagGuard::is_lock_lost) {
+                    return Err(Error::other("tier delete dispatch namespace fence was lost during predecessor replay"));
+                }
+                Ok(deleted)
+            }
+            .await;
+            if result.is_err() {
+                stopped.store(true, Ordering::Release);
+            }
+            result
+        }
+    };
+    let mut objects = source_objects.into_iter();
+    let mut replays = futures::stream::FuturesUnordered::new();
+    for _ in 0..TIER_DELETE_DISPATCH_LOCAL_REPLAY_CONCURRENCY {
+        let Some(object) = objects.next().map(ToOwned::to_owned) else {
+            break;
+        };
+        replays.push(make_replay(object));
+    }
+    let mut deleted = 0usize;
+    let mut first_error = None;
+    while let Some(result) = replays.next().await {
+        match result {
+            Ok(count) => {
+                deleted = deleted
+                    .checked_add(count)
+                    .ok_or_else(|| Error::other("tier delete dispatch replay count overflow"))?;
+            }
+            Err(err) if first_error.is_none() => first_error = Some(err),
+            Err(_) => {}
+        }
+        if first_error.is_none()
+            && !stopped.load(Ordering::Acquire)
+            && let Some(object) = objects.next().map(ToOwned::to_owned)
+        {
+            replays.push(make_replay(object));
+        }
     }
     if deleted > 0 {
         super::list_objects::observe_list_objects_mutation(store, bucket).await;
     }
+    if let Some(err) = first_error {
+        if publication_scope.is_some_and(|scope| !scope.can_commit())
+            && let Some(scope) = publication_scope
+        {
+            let _ = scope.mark_indeterminate();
+        }
+        return Err(err);
+    }
+    if let Some(scope) = publication_scope {
+        let _ = scope.mark_committed();
+    }
     Ok(())
 }
 
@@ -6518,6 +6801,28 @@ mod tests {
         );
     }
 
+    #[test]
+    fn tier_delete_walk_only_accepts_explicit_limit_cancellation() {
+        let cancelled = Error::OperationCanceled;
+        assert!(tier_delete_walk_cancellation_is_expected(true, true, &cancelled));
+        assert!(!tier_delete_walk_cancellation_is_expected(false, true, &cancelled));
+        assert!(!tier_delete_walk_cancellation_is_expected(true, false, &cancelled));
+        assert!(!tier_delete_walk_cancellation_is_expected(true, true, &Error::other("scan failed")));
+    }
+
+    #[test]
+    fn tier_delete_walk_results_prioritize_real_errors_over_cancellation() {
+        let err = combine_tier_delete_walk_results([Err(Error::OperationCanceled), Ok(()), Err(StorageError::FileAccessDenied)])
+            .expect_err("a real walk error must not be hidden by earlier cancellation");
+        assert_eq!(err, StorageError::FileAccessDenied);
+
+        assert_eq!(
+            combine_tier_delete_walk_results([Ok(()), Err(Error::OperationCanceled)])
+                .expect_err("cancellation must remain visible when there is no real error"),
+            Error::OperationCanceled
+        );
+    }
+
     impl Drop for BodyCacheHookGuard {
         fn drop(&mut self) {
             clear_get_object_body_cache_hook();

docs/architecture/compat-cleanup-register.md

@@ -12,6 +12,7 @@
 ## Open Items
 
 - `backlog-1337` legacy restore orphan recovery: releases that predate the restore worker-lock marker can leave a valid operation-id and `ongoing-request="true"` after cancellation or process failure, with no durable liveness proof. New servers allow an exact, non-nil legacy generation to be superseded only when its consistently parsed request date is at least 24 hours old. Remove the clock-based legacy fallback after the minimum supported direct-upgrade release writes the v1 worker-lock marker on every restore and operators have resolved every retained pre-v1 ongoing generation.
+- `backlog-2133-tier-delete-chunk-parent` bounded tier-delete dispatch compatibility: prefixes at or below the legacy manifest limit keep the byte-compatible v1 single-manifest protocol, while larger prefixes place a chunk-parent sentinel at the original deterministic root path and use operation-scoped child manifests. Older binaries reject the sentinel and child paths, preserving the v6 sole-owner downgrade fence instead of starting a competing local delete. Remove the v1 reader and fail-closed mixed-version sentinel only after every supported rollback release validates the parent/child protocol and migration tooling confirms that no retained v1 dispatch manifest remains.
 - `tokio-tar-extension-limits` bounded archive parser hardening: Snowball extraction depends on per-entry and cumulative GNU long-name, GNU long-link, and PAX extension limits; physical-entry, GNU sparse-map, and sparse-continuation limits; cancellation-safe sparse parsing; and fused entry streams after parser errors. The released tokio-tar API does not provide this complete boundary. Keep the reviewed fork pin until astral-sh/tokio-tar#118 is merged and one published tokio-tar release contains every listed capability with the Snowball regression fixtures passing against that release.
 - `backlog-2102` rc.2/rc.3 empty scanner usage floor recovery: old DeleteBucket cleanup could synthesize an empty incomplete v2 usage primary/backup before leadership added an epoch, while newer scanners require a durable authoritative baseline identity. New scanners recognize only that exact serialized empty-fence shape, preserve its epoch through a CAS-protected recovery marker, and rebuild namespace coverage without treating zero usage as authoritative. Remove this recovery path and marker after rc.2 and rc.3 are no longer supported direct-upgrade sources.
 - `backlog-2122` rc.1-rc.3 non-empty scanner usage floor recovery: leadership fencing in those releases can stamp scanner_epoch onto a real bucket-usage snapshot before any scanner cycle completed, leaving a non-empty floor with no scanner_cycle and no authoritative baseline identity. New scanners recognize only this consistent incomplete fenced shape, preserve the epoch through the CAS-protected recovery marker, and rebuild namespace coverage without treating the old usage data as authoritative. Remove this recovery path after rc.1, rc.2, and rc.3 are no longer supported direct-upgrade sources.

docs/architecture/ilm-tiering-persistence-contracts.md

@@ -30,7 +30,7 @@ These are approved-target invariants. A protocol's explicitly labeled current ex
 | Remote PUT is in flight or its response is unknown | Transition transaction | Only cleanup of its own canonical candidate, subject to the transaction recovery predicate | Durable transaction identity plus a known remote-version state; the approved target also requires expiry and durable takeover of the creator fence |
 | Local transition commit is complete | Exact transitioned version in `xl.meta` | No | Current recovery finds the transaction's logical bucket/object/version and checks `TRANSITION_COMPLETE` plus the same remote object, tier, and remote version. It does not compare the recorded data directory, modification time, size, or ETag; the approved target adds that full source comparison |
 | An ordinary delete removes that transitioned version | Hidden `xl.meta` free-version | Yes | Metadata quorum atomically removes the visible version and preserves its exact tier tuple in the free-version |
-| A recursive prefix/delete-all operation cannot preserve per-object markers | v6 journal bound to an immutable dispatch manifest | Yes, but only after manifest completion and all-pool absence proof | `DispatchAuthorized`, local destructive mutation, every journal `Committed`, then manifest `Completed` |
+| A recursive prefix/delete-all operation cannot preserve per-object markers | v6 journal bound to an immutable single dispatch manifest or a chunk-parent-bound child manifest | Yes, but only after child/manifest completion and all-pool absence proof | `DispatchAuthorized`, exact local destructive mutation, every journal `Committed`, then child/manifest `Completed`; a chunk parent advances only after that child completion |
 | Tier configuration mutation, manual job, or decommission receipt | Intent/admission/copy proof only | No | These records gate configuration, scheduling, or migration; they never become remote-object cleanup owners |
 
 An old journal and a free-version can coexist during compatibility recovery. That coexistence is evidence of multiple possible owners, not permission to choose one: the journal path must retain its record until the version-specific recovery rule proves which owner is authoritative.
@@ -50,11 +50,12 @@ All keys below are objects in the internal metadata bucket. The table gives the
 | Manual worker result | `rustfs-manual-transition-worker-result-v1` | `ilm/manual-transition/results/<job shards>/<job-id>/<task-key>.json` | The worker persists it after an actual result; no current GC owner | Immutable job/task key and outcome/reason | Append-only create with `If-None-Match: *` and maximum parity |
 | Legacy tier-delete journal | Versions 1 through 5 | `ilm/tier-delete-journal/<identity-digest>.json` | The deleting path creates it; version-specific journal recovery cleans it | Remote tuple; v2 adds backend identity, v3 exact version, v4 version state, v5 stable source and transaction state | v5 state changes use ETag CAS; v3/v4 recovery rereads and conditionally cleans. Initial legacy-compatible writes can still be unconditional |
 | Sole-owner tier-delete journal | Version 6 | `ilm/tier-delete-journal-v6/<operation-id>/<identity-digest>.json` | The manifest coordinator creates/dispatches it; the journal worker deletes the remote object and cleans the record | Exact remote/source/backend identity plus manifest/operation/topology binding; mutable state `Prepared`/`Dispatched`/`Committed` | Create-only and fenced ETag CAS. Record cleanup writes a terminal receipt first only while a decommission run is active; ordinary recovery without one conditionally deletes the exact ETag directly |
-| Tier-delete dispatch manifest | Version 1 | `ilm/tier-delete-dispatch-manifests/<scope-digest>.json` | The prefix-delete coordinator creates it; manifest recovery is its only rollback/completion owner | Immutable operation, bucket/incarnation/prefix, sorted journal set/count/digest, topology generation; mutable manifest state | Create-only and fenced ETag CAS; lost authorization response requires exact strong readback |
+| Tier-delete dispatch manifest | Version 1 | Single dispatch: `ilm/tier-delete-dispatch-manifests/<scope-digest>.json`; chunk child: `ilm/tier-delete-dispatch-manifests/chunks/<scope-digest>/<operation-id>.json` | The prefix-delete coordinator creates it; manifest recovery is its only rollback/completion owner | Immutable operation, bucket/incarnation/prefix, sorted journal set/count/digest, topology generation; mutable manifest state | Create-only and fenced ETag CAS; lost authorization response requires exact strong readback. A child cannot authorize local mutation without the exact active parent binding |
+| Tier-delete chunk parent | Version 1 with `record_type = "chunked_parent"` | `ilm/tier-delete-dispatch-manifests/<scope-digest>.json` | The over-limit prefix-delete coordinator creates and advances it; parent recovery advances completed children and removes the terminal parent | Immutable operation, bucket/incarnation/prefix/topology; mutable monotonic revision, next child sequence, completed journal count, one optional exact child binding, and `Active`/`Completed` state | Create-only and fenced ETag CAS. The parent binds a `Preparing` child before it can become `DispatchAuthorized`; final `Completed` follows an error-free, non-truncated empty-candidate rescan and local prefix deletion |
 | Decommission durable-namespace receipt | `v2` | `decommission/ilm-receipts/<run-token>/<source-path>/<id-kind>/<id>.json` | The decommission coordinator writes target/source proof and is the only cleanup owner for that run | Source path, namespace and record identity, monotonic checkpoint, optional terminal checkpoint, optional v6 topology generation | Create-only then ETag CAS merge; checksum envelope; maximum parity |
 | Decommission expected-receipt manifest | `v1` | `decommission/ilm-manifests/<run-token>.json` | The source-pool decommission coordinator creates and cleans it | Run token plus exact sorted receipt-path count/digest | Create-only, exact readback, and verification before pool removal |
 
-`durable_namespace.rs` registers exactly the two tier-journal namespaces, dispatch manifest, transaction, and four manual-job namespaces. A path beginning with `ilm/` that is not in that registry is an error during decommission rather than an ignorable object.
+`durable_namespace.rs` registers exactly the two tier-journal namespaces, the dispatch-record namespace shared by single manifests, chunk children, and chunk parents, the transaction namespace, and four manual-job namespaces. A path beginning with `ilm/` that is not in that registry is an error during decommission rather than an ignorable object.
 
 ## Durable fences and write primitives
 
@@ -89,8 +90,8 @@ Lock ordering is part of the recovery contract. Callers acquire only the locks n
 | Path | Current acquisition order | Operations allowed while held | Operations forbidden while held |
 |---|---|---|---|
 | Tier edit/remove/clear | Tier-config namespace WRITE lock; dedicated owned `admin_updates` serialization mutex; short `TierConfigMgr` state locks only while accessing manager/runtime state | The dedicated `admin_updates` guard intentionally spans awaited backend validation/probes, peer Prepare/Commit/Abort RPC, reference scans, config CAS, and candidate publication in the current protocol | Ordinary manager `RwLock` and runtime-state `Mutex` guards must not cross awaited network I/O; that rule does not prohibit the dedicated `admin_updates` guard from spanning those awaits. Remote object DELETE is never part of mutation |
-| v6 manifest prepare | Caller already holds the bucket-lifecycle WRITE fence; caller acquires a bucket-metadata transaction READ guard covering the Object Lock and bucket-incarnation snapshot and keeps it through local mutation; exact tier-generation leases; fleet/topology proof; synthetic manifest-operation WRITE lock | Build and write the immutable journal set and manifest, validate exact set/digest, then authorize local dispatch while both caller-held bucket guards and all leases remain current | Remote tier DELETE; per-object worker cleanup; releasing the metadata guard or a required lease before the authorized local mutation completes |
-| v6 manifest recovery | Fleet/topology proof; bucket-lifecycle WRITE lock; synthetic manifest-operation WRITE lock | Read/write manifest and journal metadata, verify exact set/digest, authorize, converge, or roll back local records | Remote tier DELETE; per-object worker cleanup; rollback after authorization |
+| v6 manifest prepare | Caller already holds the bucket-lifecycle WRITE fence; caller acquires a bucket-metadata transaction READ guard covering the Object Lock and bucket-incarnation snapshot and keeps it through local mutation; exact tier-generation leases; fleet/topology proof; for a single dispatch, synthetic manifest-operation WRITE; for a child, parent-operation WRITE then child-operation WRITE | Build and write one immutable bounded journal set and manifest, validate exact set/digest, then authorize local dispatch while both caller-held bucket guards and all leases remain current. A parent binding is durable before child authorization | Remote tier DELETE; per-object worker cleanup; releasing the metadata guard or a required lease before the authorized local mutation completes; child-to-parent nested lock acquisition |
+| v6 manifest/parent recovery | Fleet/topology proof; bucket-lifecycle WRITE lock; then exactly one synthetic manifest- or parent-operation WRITE lock | Read/write manifest, parent, and journal metadata; verify exact set/digest/binding; converge or roll back child records; advance a parent only after child completion | Remote tier DELETE; per-object worker cleanup; rollback after authorization; taking a child lock while holding a parent lock in background recovery |
 | v5 journal destructive recovery | Synthetic per-journal recovery lock; bucket-lifecycle READ lock; exact tier-generation lease; all physical object READ locks in stable pool/set order | Authoritative source/free-version scan; fenced state CAS; for an eligible terminal state, one bounded remote DELETE; conditional record cleanup | Any delete when a lock or lease is lost; publishing local metadata; selecting an arbitrary backend/version |
 | v6 journal destructive recovery | Synthetic per-journal recovery lock; fleet/topology proof; bucket-lifecycle READ lock; exact tier-generation lease; all physical object READ locks in stable pool/set order | Immutable manifest/topology validation, authoritative source/free-version scan, fenced state CAS, and, for an eligible terminal state, one bounded remote DELETE followed by record cleanup | Any delete when a lock, lease, or fleet proof is lost; publishing local metadata; selecting an arbitrary backend/version |
 | Free-version cleanup | Bucket-lifecycle READ lock; exact tier-generation lease; all physical object WRITE locks in stable pool/set order | Exact all-pool scan; bounded remote DELETE; local marker removal; post-delete rescan | Deleting before the free-version is the sole owner or after any fence changes |
@@ -242,7 +243,9 @@ The lease interval is 60 seconds. CAS behavior is phase-specific: cancellation t
 | v5 | Adds stable source identity and transaction state. Recovery proves source/free-version presence across physical sets before deciding abort, retain, or commit |
 | v6 | Sole-owner record bound to immutable operation/manifest/topology. It is the only new journal format for destructive prefix dispatch |
 
-The v1 manifest binds a bucket incarnation and prefix to an operation UUID, topology generation, and sorted journal names/count/digest. Its legal edges are:
+For a complete source set at or below 200,000 journals, the byte-compatible v1 single manifest remains at the deterministic scope-digest root. For a larger source set, that same root instead contains a strict `chunked_parent` sentinel, and each bounded child uses the unchanged v1 manifest payload at `chunks/<scope-digest>/<operation-id>.json`. A pre-chunking reader rejects the parent schema and the non-root child path, so it cannot start a competing single dispatch while chunking is active.
+
+A v1 single/child manifest binds a bucket incarnation and prefix to an operation UUID, topology generation, and sorted journal names/count/digest. Its legal edges are:
 
 ```text
 Preparing -> DispatchAuthorized -> Completed
@@ -251,6 +254,16 @@ Preparing -> DispatchAuthorized -> Completed
 
 The journal edge is `Prepared -> Dispatched -> Committed`. A manifest coordinator owns the whole `Prepared` set and is the only actor that may roll it back or complete the manifest. A per-journal worker cannot remove one prepared member.
 
+The chunk parent stores only monotonic O(1) progress and binds at most one child:
+
+```text
+Active(no child) -> Active(bound Preparing child)
+Active(bound Completed child) -> Active(no child, next sequence/count)
+Active(no child, final empty rescan and local delete complete) -> Completed
+```
+
+Child creation is ordered `Preparing` child create, parent binding CAS, journal preparation/dispatch, then child `DispatchAuthorized`. One request exactly replays one bounded child under source-object locks, commits every child journal, marks the child `Completed`, advances the parent, and returns retry-required. A successor request rescans from the prefix start; no listing cursor crosses bucket-lock lifetimes. New or changed source identities are therefore admitted only by a fresh child. Final success requires an error-free, non-truncated scan with no v6 candidate, local prefix deletion under the same bucket fence, and the parent `Completed` CAS.
+
 ### Journal recovery decisions
 
 | Record/state and evidence | Unique current owner | Current recovery decision | Remote DELETE admission |
@@ -278,11 +291,24 @@ The journal edge is `Prepared -> Dispatched -> Committed`. A manifest coordinato
 | `Completed` | Journal workers own member cleanup; coordinator owns final manifest cleanup | Wait for all member records to disappear, then conditionally delete manifest | Journal workers are the remote-delete owners |
 | Missing member, set/digest mismatch, wrong incarnation/topology, corrupt state, scan ambiguity, or cancellation | No actor acquires new destructive authority | Retain | Fail closed; an authorized operation never rolls back |
 
-Manifest preparation is bounded by 200,000 journals and a 32 MiB record. Journal recovery scans bounded batches with per-entry timeouts and limited concurrency. Those are work bounds, not retention bounds: v1/v2 quarantine and unresolved v6 operations can remain indefinitely.
+### Chunk-parent recovery decisions
+
+| Parent/child state and evidence | Unique current owner | Current recovery decision | Authority |
+|---|---|---|---|
+| Active parent with no child | A later prefix-delete retry under bucket WRITE | Retain the parent and rescan from the prefix start | No local or remote deletion |
+| Active parent with exact bound `Preparing`/`Aborting`/`Aborted` child | Child manifest coordinator | Retain parent while child recovery rolls back and removes the child | Never authorize or advance that child |
+| Active parent with exact bound `DispatchAuthorized` child | The bound child permit or journal recovery | Resume exact-source replay on request; otherwise retain until all journals become `Committed` and the child becomes `Completed` | Only the exact parent-bound child may authorize local replay |
+| Active parent with exact bound `Completed` child | Parent coordinator | CAS the next sequence/count and clear the binding | Parent progress only; remote DELETE remains owned by committed child journals |
+| Bound child missing and its exact operation journal namespace is non-empty or unreadable | No actor can prove safe abandonment | Retain and fail closed | Never clear the binding |
+| Bound child missing and its exact operation journal namespace is proven empty | Parent coordinator | CAS-clear the stale binding and retry from the prefix start | No deletion; the fresh scan reconstructs any remaining source work |
+| Completed parent with no active child | Parent recovery | Record terminal decommission evidence when applicable, then conditionally delete the exact parent ETag | Metadata cleanup only |
+| Parent identity, child binding, topology, incarnation, sequence/count, CAS generation, or fence mismatches | No actor acquires progress authority | Retain | Fail closed |
+
+Single and child manifest preparation is bounded by 200,000 journals and a 32 MiB record. On the first unique candidate beyond the bound, the physical walks are cancelled and only the retained exact batch can proceed; cancellation fallout is not absence proof. The parent never accumulates child names, and exact local replay uses bounded concurrency. Journal and manifest recovery retain their existing bounded pages, per-entry timeouts, and concurrency. These are work bounds, not retention bounds: v1/v2 quarantine and unresolved v6 operations can remain indefinitely.
 
 ### Approved target and open design
 
-- New destructive prefix paths use only v6 plus a manifest. No new v1-v5 sole-owner records may be created.
+- New destructive prefix paths use only v6 plus either one byte-compatible manifest or one parent-bound sequence of byte-compatible child manifests. No new v1-v5 sole-owner records may be created.
 - Preserve the two-phase authorization barrier: all prepared records, durable barrier, all dispatched records, durable `DispatchAuthorized`, local mutation, journals committed, durable `Completed`, then remote DELETE.
 - Do not downgrade every v6-aware recovery worker while v6 records remain. v5-and-older readers reject and retain v6 records; older nodes may continue producing fallback free-versions until the fleet is homogeneous.
 - **Open:** bounded age/count policy and operator disposition for quarantined v1/v2, incomplete manifests, and repeatedly failing exact deletes. Capacity rejection and recovery throughput must not be “fixed” by weakening ownership proof.
@@ -309,9 +335,9 @@ How historical objects without RustFS transition-version-state can be upgraded s
 
 ### Current contract
 
-Decommission cannot treat durable ILM objects as ordinary configuration blobs. `validate_durable_ilm_record` validates namespace, size, schema/checksum, identity, and a protocol-specific checkpoint, and most protocol branches recompute the canonical path. Its transition-transaction branch currently inherits the weaker final-component parser: mismatched shard directories, extra components, and uppercase hex can pass when the final UUID and record contents agree. Exact transition-path validation is therefore an approved target, not a current decommission guarantee. Checkpoint successors enforce journal/manifest legal states, transition identity and revision progression, monotonic manual-job progress, scope ownership, and immutable task/result payloads.
+Decommission cannot treat durable ILM objects as ordinary configuration blobs. `validate_durable_ilm_record` validates namespace, size, schema/checksum, identity, and a protocol-specific checkpoint, and most protocol branches recompute the canonical path. Its transition-transaction branch currently inherits the weaker final-component parser: mismatched shard directories, extra components, and uppercase hex can pass when the final UUID and record contents agree. Exact transition-path validation is therefore an approved target, not a current decommission guarantee. Checkpoint successors enforce journal/manifest legal states, chunk-parent revision/sequence/count/binding progression, transition identity and revision progression, monotonic manual-job progress, scope ownership, and immutable task/result payloads.
 
-The decommission coordinator copies and validates a durable record on a target, persists a receipt for that exact source path/identity/checkpoint, and records the expected receipt set on the source. While a matching decommission operation is active, protocol writers advance receipts as records change and terminal cleanup records a terminal checkpoint before deleting a covered record. Without an active decommission operation, the receipt helper creates no terminal receipt and ordinary protocol recovery proceeds with that protocol's current delete primitive: v6 journal/manifest cleanup uses the exact ETag, while transition-transaction cleanup remains unconditional as documented above. Completion verifies every expected receipt and target checkpoint before the source pool can be removed.
+The decommission coordinator copies and validates a durable record on a target, persists a receipt for that exact source path/identity/checkpoint, and records the expected receipt set on the source. While a matching decommission operation is active, protocol writers advance receipts as records change and terminal cleanup records a terminal checkpoint before deleting a covered record. Without an active decommission operation, the receipt helper creates no terminal receipt and ordinary protocol recovery proceeds with that protocol's current delete primitive: v6 journal/manifest/parent cleanup uses the exact ETag, while transition-transaction cleanup remains unconditional as documented above. Completion verifies every expected receipt and target checkpoint before the source pool can be removed.
 
 A terminal receipt is proof that an exact target copy reached a terminal checkpoint. It may authorize conditional removal of the matching source record when every active target copy is covered; it never authorizes remote DELETE. A terminal receipt on one target cannot hide a later nonterminal receipt on another target.
 
@@ -371,7 +397,7 @@ Transition transaction v1, manual job/task/result v1, and receipt v2 do not curr
 | Journal v1/v2 | Readers decode but quarantine because remote-version authority is missing; compatibility writers can preserve these forms | Retain indefinitely unless a separately approved, authoritative repair protocol resolves them; never translate empty version ID to known-disabled |
 | Journal v3/v4 | Readers recover supported committed records according to exact or explicit version-state semantics; current compatible writes use v4 for known state | Unknown/inconsistent state is retained. These legacy paths are not evidence that a new sole-owner operation may omit v5/v6 source proof |
 | Journal v5 | Readers use stable source/all-pool proof; decoded v5 can be checkpointed, while new online sole-owner transactions are not emitted as v5 | Retain and recover conservatively during upgrade. Do not manufacture v5 from older records or use it to bypass v6 manifest authorization |
-| Journal v6 and dispatch manifest v1 | v6-aware writers/readers require immutable manifest membership and topology; v5-and-older readers reject and retain v6 | Gate v6 writers on fleet capability. Drain v6 before removing all v6-aware workers; do not downgrade by rewriting a live v6 operation |
+| Journal v6, dispatch manifest v1, and chunk parent v1 | v6-aware writers/readers require immutable manifest membership and topology. Complete sets at or below 200,000 retain the legacy root manifest bytes; larger sets install a strict parent at that root and operation-scoped v1 child payloads. Pre-chunking v6 readers reject the parent schema and child paths, while v5-and-older readers reject and retain v6 journals | Gate writers on the current fleet capability and retain the root parent for the entire active chunk sequence. Drain v6 before removing all v6-aware workers; do not downgrade by rewriting a live v6 operation |
 | Decommission receipt v2 and expected manifest v1 | Current decommission readers validate exact schema/checksum/path/checkpoint and fail completion on unknown input | No ignore path. Mixed-version decommission must not complete unless every participant preserves the registered durable namespace; broader downgrade negotiation is open |
 
 ## Reconcile, observability, and retention