fix(mito2): pick TWCS L1 merges by closed units so compaction terminates (#9741)

* refactor(mito2): pick TWCS L1 merges by closed units

TWCS could keep rewriting L1 files without new writes. Overlap-only picks
could split their inputs into one more file than they read, and L1 file
reduction then rewrote up to 16 files to remove that single file. The
outputs, split by key, created new overlaps and the cycle repeated.

Replace the three global pick phases with one pick per window, scheduled by
class across windows. Every pick now decreases a termination measure when the
window receives no writes.

- Active windows only take merges that remove files. L1 files are grouped by
  conflicts and merged in key order, so size-split tails are merged together.
  A reduction rewrites at most 4 output files' worth of bytes per removed file.
- Inactive windows absorb L0 files, then take closed merges: no other file
  intrudes on the inputs' bounding box, so outputs cannot add overlaps. Each
  merge resolves conflicts or removes files, decreasing (conflicts, files).
  Files that do not take part in an overlap are not rewritten.
- Overlaps that only share a boundary key no longer count as conflicts, so a
  size-split sorted run does not look overlapping.
- Closed merges may take 4x the input limit to resolve a whole window once.
- Cache primary key ranges per pick instead of resolving them per comparison.
- Log the pick branch of each compaction output.

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* docs(mito2): state the compactor assumptions behind TWCS termination

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* feat(mito2): resolve active window overlaps on manual compaction

A window that stops receiving writes keeps being the active window, whose
picks only remove files, so its overlaps were never resolved. Manual
compactions now pick every window as inactive, so a closed merge can resolve
them. Automatic compactions are unchanged.

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* fix(mito2): defer overlap resolution in the window before the active one

Late data keeps landing in the window right before the active window. A
closed merge that resolves its overlaps early rewrites the whole window,
then the late data brings the overlaps back, and the balanced run check
keeps them there. Overlap-only merges now skip that window until it is
older. File-reducing merges and manual compactions are unchanged.

On the 12h/1GB out-of-order bench this reaches overlap depth 1.0 instead
of 2.08 at the same write amplification (3.01 vs 3.02).

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* refactor(mito2): only check earlier-starting files when closing units

Closing units only checks files that start before a unit ends, and the
overlap checks compare key bounds without cloning them.

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* fix(mito2): keep settling windows from rewriting L1 files

The window before the active one still receives late data. Skipping only
its overlap-only merges was not enough: file-reducing merges of
conflicting files and mixed L0/L1 merges still rewrote its L1 files,
and a mixed merge could rewrite a nearly full output for each small late
file. A settling window now only merges L0 files among themselves and
conflict-free files; the window resolves its overlaps once it is older.

On the out-of-order bench, write amplification drops from 3.88 to 3.30
at 2GB/6h, 3.05 to 2.94 at 1GB/6h and 3.01 to 2.96 at 1GB/12h, with
overlap depth 1.0 in all three.

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* fix(mito2): resolve late overlaps once a window is quiet

Late data that lands in a window older than the settling one is merged
into a small file covering the whole key range. Resolving it rewrites
the window's dominant sorted run, which the balanced run check refuses,
so the overlap stayed. Removing the check instead rewrites the window
for every late flush.

A window is quiet once the region has written at least as many rows as
the window holds since its newest file. Overlap merges in a quiet window
may rewrite a dominant run, so each quiet period pays for at most one
such rewrite. Sequences are per row and kept by compaction, so this
needs no state.

On the bench with late data up to 2 hours, mean overlap depth drops
from 1.67 to 1.17 at 2GB/6h and from 2.33 to 1.83 at 1GB/6h, at write
amplification 3.70 to 4.25 and 3.26 to 3.74; all inactive windows of
2GB/6h realtime reach depth 1 (write amplification 2.80 to 3.30). The
remaining depth is in the final active window.

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* test(mito2): check closed merges of quiet and settling windows terminate

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* fix(mito2): split TWCS closures over the input limit

A file spanning the keys of a large sorted run joins every run file into one
closure. Once a closure held more files than one compaction accepts, closed
merges skipped it, so the window never merged those files again and late
files kept piling up.

Split such closures into single files, like active-window picks do, and
replace the closed check with a conflict bound: outputs only conflict with
files intruding on the inputs' box, so a merge is taken if `outputs · I`
does not exceed the conflicts its inputs had with other files. Intervals of
whole closures have no such conflicts and still need to be closed, so other
windows pick as before, and `(conflicts, files)` still strictly decreases.

Small run files and late files over a large run can be merged again. A late
file over a run that cannot shrink stays: resolving it would rewrite the
whole run.

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* fix(mito2): address TWCS picker review findings

- Manual compactions may rewrite a dominant sorted run to resolve overlaps.
  The active window is never quiet, so manual compactions could not resolve
  its late-data overlaps before.
- Single-key intervals predict one output: the writer never splits a series,
  so large files of one series were never merged.
- The active window falls back to single files when no group of conflicting
  files qualifies, so a large file no longer keeps its small peers unmerged.
- Compute the settling window distance with checked_sub, which overflowed for
  windows at the timestamp limits.

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* fix(mito2): count inner keys as TWCS conflicts and check balance per group

- A single key strictly inside another file's key range now conflicts, so
  a late file of one series inside a wide file is resolved like other
  overlaps. Files sharing only a boundary key still do not conflict.
  Intrusion covers the new pairs, which a randomized test checks.
- Check the dominant-run guard per group of conflicting files. Runs of
  separate groups never overlap, so pairing one group's run with another
  group's peer made an interval of two dominant runs look balanced.
- A window with unknown row counts, written before they were recorded, is
  not quiet: its threshold would drop to a single row.

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* perf(mito2): index TWCS files by key to avoid quadratic grouping

Grouping a window's files compared every file with all files sharing its
time range, and closing groups and checking intrusions scanned the whole
window. A sorted run split by key shares one time range, so a window of N
such files took O(N²) checks without any conflict. The 1024-file fallback
only covers the sorted-run path, not these groups.

Index the files by lower primary key in a segment tree that keeps each
subtree's largest upper key and its time bounds. Grouping, closing, and the
conflict bound only check the files whose key and time ranges meet, so a
file of a key-sliced run only meets its neighbors. Conflicts among split
files are precomputed per file and counted per interval with a binary
search instead of pairwise.

On a run of 20,000 key-sliced files with 5 late files spanning all keys,
a closed merge pick takes ~62 ms instead of ~3.9 s and an active pick
~19 ms instead of ~3.5 s.

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* test(object-store): retry transient S3 transport failures

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* fix(mito2): bound TWCS closed-merge conflict graph work

Budget the candidate files visited while building conflict units and split-unit adjacency. When the budget is exceeded, treat every file as a split unit and pick a bounded file reduction instead of materializing a quadratic graph.

The fallback candidate is admitted only through adds_no_conflicts, so outputs do not add conflict pairs and the (P, N) termination invariant still holds. Dense picks are logged with their own branch.

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* fix(mito2): charge TWCS index nodes to the conflict-graph budget

The conflict-graph work budget only counted files that reached a callback, so a traversal with poor pruning could still visit a quadratic number of internal nodes without triggering the dense fallback. Charge every ExtentIndex node, including pruned nodes, and stop the traversal as soon as the shared budget is exhausted.

Add a regression test where the root straddles the query time but neither leaf matches, so a callback-only budget would never fire.

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* fix(mito2): retry smaller TWCS dense compaction candidates

If a dense fallback candidate fails admission, try one single-output subinterval within it before giving up. Reuse the reduction guards and check both candidates against the full window's external conflicts, preserving bounded work and termination.\n\nCover the chain regression through both fallback unit constructions and reject one-output retries that introduce conflicts across time gaps.

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* fix(mito2): compact shared primary-key boundaries

Count shared primary-key endpoints as conflicts and apply the same overlap rule to bounding-box intrusions. Existing compaction writers keep series intact, so fresh output ranges remain disjoint.

Align simulated output splits with the writer and cover boundary-only cleanup in inactive and manual windows, plus stable real output files after a subsequent TWCS request.

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* fix(mito2): compact small L1 files in inactive windows

Allow bounded small-L1 reductions in settling and older inactive windows when normal picks cannot proceed. Preserve full-window conflict checks, large-file protection, and LastNonNull closure semantics.\n\nCover selection limits, termination, and real SST compaction with tombstones and reopen checks.

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* docs(mito2): clarify TWCS unit grouping guarantees

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

---------

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

src/mito2/src/compaction/last_non_null.rs

@@ -293,7 +293,7 @@ mod tests {
             .unwrap()
             .set_compacting(busy);
         let request = compact_request::Options::Regular(Default::default());
-        let picker = new_picker(&request, &region.region_options, limit, None);
+        let picker = new_picker(&request, &region.region_options, limit, None, false);
 
         let picked = picker.pick(&region).await.unwrap().unwrap();
         assert_eq!(expected_windows.len(), picked.outputs.len());
@@ -312,7 +312,7 @@ mod tests {
 
         // The LastRow picker must retain its original seed cap and priority.
         region.region_options.merge_mode = Some(MergeMode::LastRow);
-        let picker = new_picker(&request, &region.region_options, limit, None);
+        let picker = new_picker(&request, &region.region_options, limit, None, false);
         let picked = picker.pick(&region).await.unwrap().unwrap();
         assert_eq!(limit.unwrap_or(3).min(3), picked.outputs.len());
         for (output, window) in picked.outputs.iter().rev().zip(windows.iter().rev()) {
@@ -487,6 +487,7 @@ mod tests {
             &region.region_options,
             Some(1),
             TimestampRange::new(Timestamp::new_second(0), Timestamp::new_second(3600)),
+            false,
         );
 
         let picked = picker.pick(&region).await.unwrap();
@@ -543,6 +544,7 @@ mod tests {
             &region.region_options,
             Some(1),
             TimestampRange::new(Timestamp::new_second(10), Timestamp::new_second(20)),
+            false,
         );
 
         let picked = picker.pick(&region).await.unwrap().unwrap();
@@ -590,6 +592,7 @@ mod tests {
             &region.region_options,
             Some(1),
             None,
+            false,
         );
         let picked = picker.pick(&region).await.unwrap().unwrap();
         assert!(picked.outputs.is_empty());

src/mito2/src/compaction/picker.rs

@@ -128,6 +128,7 @@ pub fn new_picker(
     region_options: &RegionOptions,
     max_background_tasks: Option<usize>,
     time_range: Option<TimestampRange>,
+    manual: bool,
 ) -> Arc<dyn Picker> {
     if let compact_request::Options::StrictWindow(window) = compact_request_options {
         let window = if window.window_seconds == 0 {
@@ -148,8 +149,10 @@ pub fn new_picker(
                 time_window_seconds: twcs_opts.time_window_seconds(),
                 max_output_file_size: twcs_opts.max_output_file_size.map(|r| r.as_bytes()),
                 append_mode: region_options.append_mode,
+                merge_mode: region_options.merge_mode(),
                 max_background_tasks,
                 time_range,
+                all_windows_inactive: manual,
             },
         };
         if region_options.merge_mode() == MergeMode::LastNonNull {

src/mito2/src/compaction/scheduler.rs

@@ -571,7 +571,7 @@ impl CompactionScheduler {
             max_parallelism,
         );
         status.merge_waiter(waiter);
-        self.dispatch_compaction_planning(plan_id, request, compact_options, time_range);
+        self.dispatch_compaction_planning(plan_id, request, compact_options, time_range, trigger);
         self.listener.on_compaction_scheduled(region_id);
         Ok(true)
     }
@@ -617,7 +617,13 @@ impl CompactionScheduler {
         // was fetched above.
         let plan_id = Self::next_plan_id(&mut self.next_plan_id);
         status.start_picking_with_trigger(plan_id, CompactionTrigger::Manual);
-        self.dispatch_compaction_planning(plan_id, request, options, time_range);
+        self.dispatch_compaction_planning(
+            plan_id,
+            request,
+            options,
+            time_range,
+            CompactionTrigger::Manual,
+        );
         debug!(
             "Successfully scheduled manual compaction planning for region id: {}",
             region_id
@@ -703,6 +709,7 @@ impl CompactionScheduler {
             request,
             compact_request::Options::Regular(Default::default()),
             None,
+            CompactionTrigger::Automatic,
         );
         debug!(
             "Successfully scheduled next compaction planning for region id: {}",

src/mito2/src/compaction/scheduler/planning.rs

@@ -32,7 +32,9 @@ use crate::access_layer::AccessLayerRef;
 use crate::cache::CacheManagerRef;
 use crate::compaction::compactor::{CompactionRegion, CompactionVersion};
 use crate::compaction::picker::{PickerOutput, new_picker};
-use crate::compaction::scheduler::state::{CompactingFiles, CompactionExecution, CompactionPhase};
+use crate::compaction::scheduler::state::{
+    CompactingFiles, CompactionExecution, CompactionPhase, CompactionTrigger,
+};
 use crate::compaction::scheduler::{CompactionScheduler, CompactionTransition};
 use crate::compaction::unit::CompactionUnit;
 use crate::compaction::{CompactionOutput, find_dynamic_options, format_panic_reason};
@@ -115,6 +117,7 @@ impl CompactionScheduler {
         request: CompactionRequest,
         options: compact_request::Options,
         time_range: Option<TimestampRange>,
+        trigger: CompactionTrigger,
     ) {
         let plugins = self.plugins.clone();
         let max_background_compactions = self.engine_config.max_background_compactions;
@@ -127,6 +130,7 @@ impl CompactionScheduler {
                 plugins,
                 max_background_compactions,
                 time_range,
+                trigger,
             );
             Self::notify_planning_result(region_id, plan_id, request_sender, planning).await;
         });
@@ -180,6 +184,7 @@ impl CompactionScheduler {
         plugins: Plugins,
         max_background_compactions: usize,
         time_range: Option<TimestampRange>,
+        trigger: CompactionTrigger,
     ) -> CompactionPlanningResult {
         let region_id = request.region_id();
         let (dynamic_compaction_opts, ttl) = find_dynamic_options(
@@ -208,6 +213,7 @@ impl CompactionScheduler {
             &region_options,
             Some(max_picker_outputs),
             time_range,
+            trigger == CompactionTrigger::Manual,
         );
         let region_id = request.region_id();
         let CompactionRequest {

src/mito2/src/engine/compaction_test.rs

@@ -212,6 +212,7 @@ async fn test_strict_window_output_file_size(
     assert!(version.ssts.levels()[0].files.is_empty());
     let primary_key_mapper = version.ssts.primary_key_mapper();
     let files = version.ssts.levels()[1].files().collect::<Vec<_>>();
+    let max_output_size = files.iter().map(|file| file.size()).max().unwrap();
     assert_eq!(
         6000,
         files
@@ -229,6 +230,14 @@ async fn test_strict_window_output_file_size(
             .collect::<Vec<_>>();
         if expect_split {
             assert!(window_files.len() > 1, "each window must split by size");
+            assert_eq!(
+                window_files.len() as u64,
+                window_files
+                    .iter()
+                    .map(|file| file.size())
+                    .sum::<u64>()
+                    .div_ceil(max_output_size)
+            );
         } else {
             assert_eq!(1, window_files.len(), "zero must disable size splitting");
         }
@@ -253,6 +262,35 @@ async fn test_strict_window_output_file_size(
         sort_batches_and_print(&before, &["tag_0", "ts"]),
         sort_batches_and_print(&after, &["tag_0", "ts"])
     );
+
+    // Fresh size-split outputs have disjoint series and must not look like
+    // independently written files sharing a boundary key to the TWCS picker.
+    // Raise the tiny split threshold so an overlap merge would be eligible by
+    // size; otherwise the output-count guard alone would reject these files.
+    engine
+        .handle_request(
+            region_id,
+            RegionRequest::Alter(RegionAlterRequest {
+                kind: SetRegionOptions {
+                    options: vec![SetRegionOption::Twsc(
+                        "compaction.twcs.max_output_file_size".into(),
+                        format!("{max_output_size}B"),
+                    )],
+                },
+            }),
+        )
+        .await
+        .unwrap();
+    let output_ids: HashSet<_> = files.iter().map(|file| file.file_id()).collect();
+    compact(&engine, region_id).await;
+    let version = engine.get_region(region_id).unwrap().version();
+    assert_eq!(
+        output_ids,
+        version.ssts.levels()[1]
+            .files()
+            .map(|file| file.file_id())
+            .collect()
+    );
 }
 
 /// Flush may collapse versions within one file, but compaction must not promote
@@ -544,6 +582,169 @@ async fn assert_partial_compaction_preserves_delete_order(flat_format: bool) {
     );
 }
 
+/// Exercise the small-L1 fallback with real SST sizes, preserved tombstones,
+/// and both layouts. No fabricated file metadata or hand-picked merge inputs.
+#[rstest::rstest]
+#[tokio::test]
+async fn test_inactive_small_l1_compaction_preserves_large_peer(
+    #[values(false, true)] flat_format: bool,
+    #[values(3600, 7200)] active_ts: usize,
+) {
+    let mut env = TestEnv::new().await;
+    let region_id = RegionId::new(1, 1);
+    let (engine, columns) = env_for_manual_compaction(&mut env, region_id, flat_format).await;
+    let mut rng = <rand::rngs::StdRng as rand::SeedableRng>::seed_from_u64(41);
+    let mut old_rows = build_rows_for_key("m", 10, 16, 0);
+    for _ in 0..30_000 {
+        old_rows.extend(build_rows_for_key(
+            &format!("a{:032x}", rand::Rng::random::<u128>(&mut rng)),
+            1,
+            2,
+            0,
+        ));
+    }
+    for _ in 0..2 {
+        put_rows(
+            &engine,
+            region_id,
+            Rows {
+                schema: columns.clone(),
+                rows: old_rows.clone(),
+            },
+        )
+        .await;
+        flush(&engine, region_id).await;
+    }
+    compact(&engine, region_id).await;
+    let region = engine.get_region(region_id).unwrap();
+    let table_dir = region.table_dir().to_string();
+    let large = region.version().ssts.levels()[1]
+        .files()
+        .next()
+        .unwrap()
+        .clone();
+    const OUTPUT_SIZE: u64 = 64 * 1024;
+    assert!(large.size() > 8 * OUTPUT_SIZE);
+    engine
+        .handle_request(
+            region_id,
+            RegionRequest::Alter(RegionAlterRequest {
+                kind: SetRegionOptions {
+                    options: vec![
+                        SetRegionOption::Twsc(
+                            "compaction.twcs.max_output_file_size".into(),
+                            "64KiB".into(),
+                        ),
+                        SetRegionOption::Twsc(
+                            "compaction.twcs.inactive_window.l1_merge_trigger".into(),
+                            "2".into(),
+                        ),
+                    ],
+                },
+            }),
+        )
+        .await
+        .unwrap();
+
+    // Each pair of L0 files becomes one small L1. The first contains a Delete
+    // masking the large SST; later pairs update other rows in its key/time box.
+    for ts in 10..14 {
+        for _ in 0..2 {
+            let rows = Rows {
+                schema: columns.clone(),
+                rows: build_rows_for_key("m", ts, ts + 2, ts),
+            };
+            if ts == 10 {
+                engine
+                    .handle_request(
+                        region_id,
+                        RegionRequest::Delete(RegionDeleteRequest {
+                            rows,
+                            hint: None,
+                            partition_expr_version: None,
+                        }),
+                    )
+                    .await
+                    .unwrap();
+            } else {
+                put_rows(&engine, region_id, rows).await;
+            }
+            flush(&engine, region_id).await;
+        }
+        compact(&engine, region_id).await;
+    }
+    put_and_flush(&engine, region_id, &columns, active_ts..active_ts + 1).await;
+    let version = region.version();
+    let small = version.ssts.levels()[1]
+        .files()
+        .filter(|file| file.file_id() != large.file_id())
+        .collect::<Vec<_>>();
+    assert_eq!(4, small.len());
+    assert!(small.iter().all(|file| file.size() <= OUTPUT_SIZE / 4));
+    let small_ids = small
+        .iter()
+        .map(|file| file.file_id())
+        .collect::<HashSet<_>>();
+    let pick = preview_regular_compaction(&engine, region_id).await;
+    assert_eq!(1, pick.outputs.len());
+    assert_eq!(
+        small_ids,
+        pick.outputs[0]
+            .inputs
+            .iter()
+            .map(|file| file.file_id())
+            .collect()
+    );
+    assert!(!pick.outputs[0].filter_deleted);
+    let before = RecordBatches::try_collect(
+        engine
+            .scan_to_stream(region_id, ScanRequest::default())
+            .await
+            .unwrap(),
+    )
+    .await
+    .unwrap()
+    .pretty_print()
+    .unwrap();
+
+    // Even manual cleanup cannot rewrite the dominant peer within the closed
+    // merge's output bound, so it executes the same small-only fallback.
+    compact(&engine, region_id).await;
+    let version = region.version();
+    let l1 = version.ssts.levels()[1].files().collect::<Vec<_>>();
+    assert_eq!(2, l1.len());
+    assert!(l1.iter().any(|file| file.file_id() == large.file_id()));
+    let output = l1
+        .iter()
+        .find(|file| file.file_id() != large.file_id())
+        .unwrap();
+    assert!(!small_ids.contains(&output.file_id()));
+    assert_eq!(1, count_sst_deletes(&engine, region_id, output).await);
+    let after = RecordBatches::try_collect(
+        engine
+            .scan_to_stream(region_id, ScanRequest::default())
+            .await
+            .unwrap(),
+    )
+    .await
+    .unwrap()
+    .pretty_print()
+    .unwrap();
+    assert_eq!(before, after);
+    crate::test_util::reopen_region(&engine, region_id, table_dir, true, HashMap::new()).await;
+    let reopened = RecordBatches::try_collect(
+        engine
+            .scan_to_stream(region_id, ScanRequest::default())
+            .await
+            .unwrap(),
+    )
+    .await
+    .unwrap()
+    .pretty_print()
+    .unwrap();
+    assert_eq!(before, reopened);
+}
+
 /// Region fixture containing only identifiers, so it cannot pin an old SST snapshot.
 struct CompactionTestRegion {
     region_id: RegionId,
@@ -1806,6 +2007,7 @@ async fn preview_regular_compaction(
         &version.options,
         None,
         None,
+        false,
     );
     let compaction_region = CompactionRegion {
         region_id,
@@ -2777,9 +2979,9 @@ async fn test_compaction_region_with_format(flat_format: bool) {
     //          -[15.........29]- (delete)
     //           [15.....24]
     // Count-first compaction consumes the first 4 SSTs as soon as the trigger is reached.
-    // The compacted output and final flush leave 2 SSTs.
+    // The manual compaction then merges the compacted output with the final flush.
     assert_eq!(
-        2,
+        1,
         scanner.num_files(),
         "unexpected files: {:?}",
         scanner.file_ids()

src/mito2/src/gc/worker_test.rs

@@ -399,7 +399,13 @@ async fn test_gc_worker_basic_compact() {
 
     let region = engine.get_region(region_id).unwrap();
     let manifest = region.manifest_ctx.manifest().await;
-    let removed_file_num = manifest.removed_files.removed_files[0].files.len();
+    // A manual compaction may take several rounds, each recording its removals.
+    let removed_file_num: usize = manifest
+        .removed_files
+        .removed_files
+        .iter()
+        .map(|removed| removed.files.len())
+        .sum();
     assert!(removed_file_num > 0);
 
     let version = manifest.manifest_version;
@@ -488,9 +494,11 @@ async fn test_gc_worker_compact_with_ref() {
     let file_ref_manifest = FileRefsManifest {
         file_refs: HashMap::from([(
             region_id,
-            manifest.removed_files.removed_files[0]
-                .files
+            manifest
+                .removed_files
+                .removed_files
                 .iter()
+                .flat_map(|removed| removed.files.iter())
                 .map(|removed_file| match removed_file {
                     RemovedFile::File(file_id, v) => FileRef::new(region_id, *file_id, *v),
                     RemovedFile::Index(file_id, v) => FileRef::new(region_id, *file_id, Some(*v)),

src/object-store/tests/object_store_test.rs

@@ -304,6 +304,9 @@ async fn test_s3_backend() -> Result<()> {
         }
 
         let store = ObjectStore::new(builder).unwrap();
+        // Match production S3 clients: retry transient transport failures without
+        // retrying conditional-write conflicts or weakening their assertions.
+        let store = object_store::util::with_retry_layers(store);
         let store = object_store::util::with_instrument_layers(store, false);
 
         let guard = TempFolder::new(&store, "/");