diff --git a/src/log-store/src/metrics.rs b/src/log-store/src/metrics.rs
index 824058d70af..badb5a80b70 100644
--- a/src/log-store/src/metrics.rs
+++ b/src/log-store/src/metrics.rs
@@ -97,4 +97,18 @@ lazy_static! {
         "object store logstore skipped corrupted segments total",
     )
     .unwrap();
+    /// Counter of objects the object store logstore deleted because no
+    /// region needs them any more.
+    pub static ref METRIC_OBJECT_STORE_WAL_DELETED_OBJECTS_TOTAL: IntCounter = register_int_counter!(
+        "greptime_logstore_object_store_wal_deleted_objects_total",
+        "object store logstore deleted objects total",
+    )
+    .unwrap();
+    /// Counter of object deletes of the object store logstore that failed and
+    /// are attempted again at the next collection.
+    pub static ref METRIC_OBJECT_STORE_WAL_FAILED_DELETES_TOTAL: IntCounter = register_int_counter!(
+        "greptime_logstore_object_store_wal_failed_deletes_total",
+        "object store logstore failed object deletes total",
+    )
+    .unwrap();
 }
diff --git a/src/log-store/src/object_store_wal.rs b/src/log-store/src/object_store_wal.rs
index d4bfc096393..6a25afb3e77 100644
--- a/src/log-store/src/object_store_wal.rs
+++ b/src/log-store/src/object_store_wal.rs
@@ -24,9 +24,9 @@
 //!
 //! The header carries the magic `GTWALOBJ`, the format version, the object
 //! sequence, the epoch of the instance that wrote the object, the link to the
-//! object it extends, and a CRC32 of the header. Each segment holds the
-//! entries of exactly one region, ordered by entry id, and segments are ordered
-//! by region id. The footer indexes every segment with its region id, entry id
+//! object it extends, the reclaim boundary, and a CRC32 of the header. Each
+//! segment holds the entries of exactly one region, ordered by entry id, and
+//! segments are ordered by region id. The footer indexes every segment with its region id, entry id
 //! range, byte range and CRC32. The fixed-size trailer points at the footer and
 //! carries the CRC32 of the footer and of the whole object, so a reader locates
 //! the footer by reading the fixed-length trailer at the end of the object.
@@ -40,12 +40,18 @@
 //! object it extends by sequence and epoch, and recovery replays only the chain
 //! that ends at the complete object with the largest epoch and sequence. An
 //! object is complete when every link on its chain names a present object of
-//! the recorded epoch, back to an object that starts the chain. Objects off the
-//! chain are orphans, which are never replayed but keep their sequences. Each
-//! open writes an object without segments above every present object before
-//! it accepts writes, and its epoch is one above the sequence of that object,
-//! so no two instances share an epoch and a late object of an earlier
-//! instance never ends the chain.
+//! the recorded epoch, back to an object that starts the chain or to a link
+//! below the reclaim boundary of the object. Objects off the chain are
+//! orphans, which are never replayed but keep their sequences. Each open
+//! writes an object without segments above every present object before it
+//! accepts writes, and its epoch is one above the sequence of that object, so
+//! no two instances share an epoch and a late object of an earlier instance
+//! never ends the chain.
+//!
+//! The reclaim boundary of an object states that every object with a lower
+//! sequence was fully flushed when the object was sealed, so the chain need
+//! not reach below it. Every present object below the boundary of the last
+//! indexed object is deleted, on the chain or not.
 //!
 //! Recovery lists the objects, reads and verifies only the header, trailer and
 //! footer of each, and indexes the footers of the chain in sequence order to
diff --git a/src/log-store/src/object_store_wal/catalog.rs b/src/log-store/src/object_store_wal/catalog.rs
index 4866da359cf..70f7aa5b329 100644
--- a/src/log-store/src/object_store_wal/catalog.rs
+++ b/src/log-store/src/object_store_wal/catalog.rs
@@ -14,10 +14,11 @@
 
 //! In-memory index over the footers of the objects of one WAL prefix.
 
-use std::collections::BTreeMap;
+use std::collections::{BTreeMap, HashMap};
 use std::ops::Bound::{Excluded, Unbounded};
 
 use snafu::{OptionExt, ensure};
+use store_api::logstore::EntryId;
 use store_api::storage::RegionId;
 
 use crate::error::{
@@ -32,6 +33,9 @@ use crate::object_store_wal::format::FooterEntry;
 pub(crate) struct ObjectCatalog {
     objects: BTreeMap<u64, Vec<FooterEntry>>,
     regions: BTreeMap<RegionId, BTreeMap<u64, FooterEntry>>,
+    /// Largest entry id ever indexed per region, kept after the object that
+    /// held it was removed.
+    max_entry_ids: HashMap<RegionId, EntryId>,
 }
 
 impl ObjectCatalog {
@@ -123,11 +127,71 @@ impl ObjectCatalog {
                 .entry(entry.region_id)
                 .or_default()
                 .insert(object_seq, entry.clone());
+            self.max_entry_ids
+                .entry(entry.region_id)
+                .and_modify(|max| *max = (*max).max(entry.max_entry_id))
+                .or_insert(entry.max_entry_id);
         }
         self.objects.insert(object_seq, footer);
         Ok(())
     }
 
+    /// Removes the object `object_seq` from the index once it was deleted and
+    /// returns the footer it held, or `None` if it was not indexed. The
+    /// largest entry id of every region it held is kept.
+    pub(crate) fn remove_object(&mut self, object_seq: u64) -> Option<Vec<FooterEntry>> {
+        let footer = self.objects.remove(&object_seq)?;
+        for entry in &footer {
+            if let Some(objects) = self.regions.get_mut(&entry.region_id) {
+                objects.remove(&object_seq);
+                if objects.is_empty() {
+                    self.regions.remove(&entry.region_id);
+                }
+            }
+        }
+        Some(footer)
+    }
+
+    /// Returns true while the object `object_seq` is indexed.
+    pub(crate) fn contains_object(&self, object_seq: u64) -> bool {
+        self.objects.contains_key(&object_seq)
+    }
+
+    /// Returns the reclaim boundary the indexed objects and the obsolete
+    /// watermarks allow, given that every object below `from` holds no live
+    /// segment: the first indexed object at or above `from` that holds a
+    /// live segment, or the sequence after the last indexed object when none
+    /// does, and never less than `from`.
+    ///
+    /// A segment is live when its maximum entry id is above the watermark of
+    /// its region, or its region has no watermark; a region whose watermark
+    /// is [`EntryId::MAX`] holds no live segment. Watermarks never move down
+    /// and a region first appears in an object above the boundary, so the
+    /// scan starts at the previous boundary and never reads an object twice
+    /// once the boundary has passed it.
+    pub(crate) fn reclaim_boundary(
+        &self,
+        from: u64,
+        obsolete_entry_ids: &HashMap<RegionId, EntryId>,
+    ) -> u64 {
+        let live = self.objects.range(from..).find(|(_, footer)| {
+            footer.iter().any(|entry| {
+                obsolete_entry_ids
+                    .get(&entry.region_id)
+                    .is_none_or(|obsolete| entry.max_entry_id > *obsolete)
+            })
+        });
+        match live {
+            Some((&object_seq, _)) => object_seq,
+            None => self
+                .objects
+                .last_key_value()
+                .map_or(from, |(&object_seq, _)| {
+                    object_seq.saturating_add(1).max(from)
+                }),
+        }
+    }
+
     /// Returns the objects that hold entries of `region_id` overlapping
     /// `start_entry_id..=end_entry_id`, ordered by object sequence.
     pub(crate) fn objects_for_entry_range(
@@ -157,12 +221,10 @@ impl ObjectCatalog {
             .collect())
     }
 
-    /// Returns the largest entry id indexed for `region_id`.
+    /// Returns the largest entry id ever indexed for `region_id`, which is
+    /// kept after the object holding it was removed.
     pub(crate) fn region_max_entry_id(&self, region_id: RegionId) -> Option<u64> {
-        self.regions
-            .get(&region_id)?
-            .last_key_value()
-            .map(|(_, entry)| entry.max_entry_id)
+        self.max_entry_ids.get(&region_id).copied()
     }
 
     /// Returns the sequence to assign to the next object written after recovery.
@@ -183,9 +245,9 @@ impl ObjectCatalog {
                 .context(WalObjectSequenceExhaustedSnafu { last_object_seq })?,
         };
         let floor = self
-            .regions
-            .keys()
-            .filter_map(|region_id| self.region_max_entry_id(*region_id))
+            .max_entry_ids
+            .values()
+            .copied()
             .map(sequence_floor)
             .max()
             .unwrap_or(0);
@@ -431,6 +493,98 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_catalog_reclaim_boundary_follows_the_watermarks() {
+        let region_one = RegionId::new(1, 1);
+        let region_two = RegionId::new(2, 1);
+        let mut catalog = ObjectCatalog::default();
+        catalog.insert_object(0, Vec::new()).unwrap();
+        catalog
+            .insert_object(
+                1,
+                vec![
+                    footer_entry(region_one, entry_id(1, 1), entry_id(1, 2)),
+                    footer_entry(region_two, entry_id(1, 1), entry_id(1, 1)),
+                ],
+            )
+            .unwrap();
+        catalog
+            .insert_object(
+                2,
+                vec![footer_entry(region_one, entry_id(2, 1), entry_id(2, 1))],
+            )
+            .unwrap();
+        catalog
+            .insert_object(
+                3,
+                vec![footer_entry(region_two, entry_id(3, 1), entry_id(3, 1))],
+            )
+            .unwrap();
+        let boundary = |catalog: &ObjectCatalog, obsolete: &[(RegionId, EntryId)]| {
+            catalog.reclaim_boundary(0, &obsolete.iter().copied().collect())
+        };
+
+        // A region without a watermark holds the boundary at its first object.
+        assert_eq!(1, boundary(&catalog, &[]));
+        assert_eq!(1, boundary(&catalog, &[(region_one, entry_id(2, 1))]));
+        // A watermark inside an object keeps that object.
+        assert_eq!(
+            1,
+            boundary(
+                &catalog,
+                &[(region_one, entry_id(1, 1)), (region_two, EntryId::MAX)]
+            )
+        );
+        assert_eq!(
+            2,
+            boundary(
+                &catalog,
+                &[(region_one, entry_id(1, 2)), (region_two, entry_id(1, 1))]
+            )
+        );
+        assert_eq!(
+            3,
+            boundary(
+                &catalog,
+                &[(region_one, entry_id(2, 1)), (region_two, entry_id(1, 1))]
+            )
+        );
+        // A region of `obsolete_all` contributes nothing, and when no region
+        // does, the boundary is past the last object.
+        assert_eq!(
+            4,
+            boundary(
+                &catalog,
+                &[(region_one, entry_id(2, 1)), (region_two, EntryId::MAX)]
+            )
+        );
+
+        // Objects below the previous boundary are not scanned again.
+        assert_eq!(3, catalog.reclaim_boundary(3, &HashMap::new()));
+        assert_eq!(5, catalog.reclaim_boundary(5, &HashMap::new()));
+
+        // Removing objects keeps the largest entry id of every region, and a
+        // region without indexed objects contributes nothing.
+        for object_seq in [0, 1, 2] {
+            assert!(catalog.remove_object(object_seq).is_some());
+        }
+        assert!(catalog.remove_object(2).is_none());
+        assert!(!catalog.contains_object(2));
+        assert!(catalog.contains_object(3));
+        assert_eq!(
+            Some(entry_id(2, 1)),
+            catalog.region_max_entry_id(region_one)
+        );
+        assert!(
+            catalog
+                .objects_for_entry_range(region_one, 0, EntryId::MAX)
+                .unwrap()
+                .is_empty()
+        );
+        assert_eq!(3, boundary(&catalog, &[]));
+        assert_eq!(4, catalog.next_object_seq().unwrap());
+    }
+
     fn footer_entry(region_id: RegionId, min_entry_id: u64, max_entry_id: u64) -> FooterEntry {
         FooterEntry {
             region_id,
diff --git a/src/log-store/src/object_store_wal/format.rs b/src/log-store/src/object_store_wal/format.rs
index 6dd46c87242..bd7e9ae6410 100644
--- a/src/log-store/src/object_store_wal/format.rs
+++ b/src/log-store/src/object_store_wal/format.rs
@@ -32,8 +32,9 @@ const FORMAT_VERSION: u16 = 1;
 const NO_PREDECESSOR: u64 = u64::MAX;
 
 /// Length of the object header: magic, version, object sequence, epoch,
-/// predecessor sequence, predecessor epoch and the CRC32 of the header.
-pub(crate) const HEADER_LEN: usize = 8 + 2 + 8 + 8 + 8 + 8 + 4;
+/// predecessor sequence, predecessor epoch, reclaim boundary and the CRC32 of
+/// the header.
+pub(crate) const HEADER_LEN: usize = 8 + 2 + 8 + 8 + 8 + 8 + 8 + 4;
 /// Length of the fixed trailer: footer offset, footer length, footer CRC32,
 /// object CRC32 and magic.
 pub(crate) const TRAILER_LEN: usize = 8 + 8 + 4 + 4 + 8;
@@ -55,6 +56,10 @@ pub(crate) struct Header {
     pub(crate) epoch: u64,
     /// The object this object extends, `None` for the first object of a chain.
     pub(crate) prev: Option<ChainLink>,
+    /// Every object below this sequence was fully flushed when this object
+    /// was sealed, so the chain recovery replays need not reach below it. It
+    /// is at or below the sequence of the object.
+    pub(crate) reclaim_boundary: u64,
 }
 
 /// Names the object another object extends: its sequence and the epoch
@@ -224,6 +229,7 @@ pub(crate) fn decode_header(bytes: &[u8]) -> Result<Header> {
     let epoch = reader.u64("header")?;
     let prev_object_seq = reader.u64("header")?;
     let prev_epoch = reader.u64("header")?;
+    let reclaim_boundary = reader.u64("header")?;
     let header_crc32 = reader.u32("header")?;
     let checksum = crc32fast::hash(&bytes[..HEADER_LEN - 4]);
     ensure!(
@@ -232,6 +238,14 @@ pub(crate) fn decode_header(bytes: &[u8]) -> Result<Header> {
             reason: checksum_mismatch("header", header_crc32, checksum),
         }
     );
+    ensure!(
+        reclaim_boundary <= object_seq,
+        CorruptedWalObjectSnafu {
+            reason: format!(
+                "reclaim boundary {reclaim_boundary} is above the object sequence {object_seq}"
+            ),
+        }
+    );
     let prev = (prev_object_seq != NO_PREDECESSOR).then_some(ChainLink {
         object_seq: prev_object_seq,
         epoch: prev_epoch,
@@ -240,6 +254,7 @@ pub(crate) fn decode_header(bytes: &[u8]) -> Result<Header> {
         object_seq,
         epoch,
         prev,
+        reclaim_boundary,
     })
 }
 
@@ -582,6 +597,7 @@ fn encode_header(header: &Header, output: &mut BytesMut) {
     });
     output.put_u64(prev.object_seq);
     output.put_u64(prev.epoch);
+    output.put_u64(header.reclaim_boundary);
     output.put_u32(crc32fast::hash(&output[start..]));
 }
 
@@ -730,6 +746,7 @@ mod tests {
                 object_seq: 40,
                 epoch: 4,
             }),
+            reclaim_boundary: 37,
         }
     }
 
@@ -838,7 +855,7 @@ mod tests {
 
         // Every field after the version is covered by the header checksum,
         // which recovery verifies without reading the whole object.
-        for offset in [10, 18, 26, 34, HEADER_LEN - 1] {
+        for offset in [10, 18, 26, 34, 42, HEADER_LEN - 1] {
             let mut bad_field = encoded.bytes.to_vec();
             bad_field[offset] ^= 1;
             assert_corrupted(decode_header(&bad_field), "header checksum mismatch");
@@ -855,6 +872,19 @@ mod tests {
         assert_eq!(first, decode_object(&encoded.bytes).unwrap().header);
     }
 
+    #[test]
+    fn test_format_rejects_a_reclaim_boundary_above_the_object() {
+        let above = Header {
+            reclaim_boundary: header().object_seq + 1,
+            ..header()
+        };
+        let encoded = encode_object(above, &records()).unwrap();
+        assert_corrupted(
+            decode_header(&encoded.bytes),
+            "reclaim boundary 43 is above the object sequence 42",
+        );
+    }
+
     #[test]
     fn test_format_rejects_trailer_magic_and_invalid_footer_range() {
         let encoded = encode_object(header(), &records()).unwrap();
@@ -971,15 +1001,16 @@ mod tests {
     /// the encoder and the decoder change together.
     const FIXTURE_V1_HEX: &str = concat!(
         // Header: magic, version 1, object sequence 7, epoch 3, predecessor
-        // sequence 6, predecessor epoch 2, header CRC32.
+        // sequence 6, predecessor epoch 2, reclaim boundary 5, header CRC32.
         "475457414c4f424a",
         "0001",
         "0000000000000007",
         "0000000000000003",
         "0000000000000006",
         "0000000000000002",
-        "1ca121b1",
-        // Segment of region 1 at offset 46: region id, entry count, then
+        "0000000000000005",
+        "118e773f",
+        // Segment of region 1 at offset 54: region id, entry count, then
         // (entry id, payload length, payload) per entry.
         "0000000100000001",
         "00000003",
@@ -992,7 +1023,7 @@ mod tests {
         "0000000000000003",
         "00000003",
         "636363",
-        // Segment of region 2 at offset 100.
+        // Segment of region 2 at offset 108.
         "0000000200000001",
         "00000002",
         "000000000000000a",
@@ -1001,28 +1032,28 @@ mod tests {
         "000000000000000b",
         "00000002",
         "7979",
-        // Footer at offset 139: entry count, then per segment region id, min and
+        // Footer at offset 147: entry count, then per segment region id, min and
         // max entry id, entry count, segment offset, length and CRC32.
         "00000002",
         "0000000100000001",
         "0000000000000001",
         "0000000000000003",
         "00000003",
-        "000000000000002e",
+        "0000000000000036",
         "0000000000000036",
         "25ac0486",
         "0000000200000001",
         "000000000000000a",
         "000000000000000b",
         "00000002",
-        "0000000000000064",
+        "000000000000006c",
         "0000000000000027",
         "65dc08ec",
         // Trailer: footer offset, footer length, footer CRC32, object CRC32, magic.
-        "000000000000008b",
+        "0000000000000093",
         "0000000000000064",
-        "6b97e3e4",
-        "631bb7d1",
+        "d80c9f07",
+        "65c91839",
         "475457414c54524c",
     );
 
@@ -1060,6 +1091,7 @@ mod tests {
                 object_seq: 6,
                 epoch: 2,
             }),
+            reclaim_boundary: 5,
         };
         let footer = vec![
             FooterEntry {
@@ -1067,7 +1099,7 @@ mod tests {
                 min_entry_id: 1,
                 max_entry_id: 3,
                 entry_count: 3,
-                segment_offset: 46,
+                segment_offset: 54,
                 segment_len: 54,
                 segment_crc32: 0x25ac0486,
             },
@@ -1076,12 +1108,12 @@ mod tests {
                 min_entry_id: 10,
                 max_entry_id: 11,
                 entry_count: 2,
-                segment_offset: 100,
+                segment_offset: 108,
                 segment_len: 39,
                 segment_crc32: 0x65dc08ec,
             },
         ];
-        assert_eq!(271, fixture.len());
+        assert_eq!(279, fixture.len());
 
         let decoded = decode_object(&fixture).unwrap();
         assert_eq!(header, decoded.header);
@@ -1089,10 +1121,10 @@ mod tests {
         assert_eq!(fixture_records(), decoded.records);
         assert_eq!(
             FixedTrailer {
-                footer_offset: 139,
+                footer_offset: 147,
                 footer_len: 100,
-                footer_crc32: 0x6b97e3e4,
-                object_crc32: 0x631bb7d1,
+                footer_crc32: 0xd80c9f07,
+                object_crc32: 0x65c91839,
             },
             decode_trailer(&fixture[fixture.len() - TRAILER_LEN..]).unwrap()
         );
diff --git a/src/log-store/src/object_store_wal/io.rs b/src/log-store/src/object_store_wal/io.rs
index 0af6ac70278..4fc7e971535 100644
--- a/src/log-store/src/object_store_wal/io.rs
+++ b/src/log-store/src/object_store_wal/io.rs
@@ -138,6 +138,16 @@ impl ObjectStoreIo {
             })
     }
 
+    /// Deletes the object `object_seq`. Deleting an object that does not
+    /// exist succeeds.
+    pub(crate) async fn delete(&self, object_seq: u64) -> Result<()> {
+        let path = self.object_path(object_seq);
+        self.store.delete(&path).await.context(WalObjectStoreSnafu {
+            operation: "delete",
+            path,
+        })
+    }
+
     /// Lists the objects under the prefix, ordered by object sequence. Keys
     /// that do not follow the object layout are ignored.
     pub(crate) async fn list(&self) -> Result<Vec<ListedObject>> {
@@ -357,6 +367,11 @@ mod tests {
             vec![1, 1, 2],
             objects.iter().map(|object| object.size).collect::<Vec<_>>()
         );
+
+        // A delete of an object that is already gone succeeds.
+        io.delete(2).await.unwrap();
+        io.delete(2).await.unwrap();
+        assert_eq!(vec![1, 10], object_seqs(io.list().await.unwrap()));
     }
 
     #[tokio::test]
diff --git a/src/log-store/src/object_store_wal/store.rs b/src/log-store/src/object_store_wal/store.rs
index 8750a61cc4b..dd3c3958336 100644
--- a/src/log-store/src/object_store_wal/store.rs
+++ b/src/log-store/src/object_store_wal/store.rs
@@ -14,6 +14,7 @@
 
 //! Object store WAL construction, recovery, writes and region reads.
 
+use std::cmp::Reverse;
 use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet, VecDeque};
 use std::fmt;
 use std::ops::Range;
@@ -23,7 +24,7 @@ use std::time::{Duration, Instant};
 
 use async_stream::try_stream;
 use bytes::Bytes;
-use common_telemetry::{info, warn};
+use common_telemetry::{debug, info, warn};
 use common_wal::config::object_store::{AckMode, CorruptedSegmentAction, ObjectStoreWalConfig};
 use futures::future::BoxFuture;
 use futures::stream::FuturesUnordered;
@@ -34,9 +35,7 @@ use store_api::logstore::entry::{Entry, NaiveEntry};
 use store_api::logstore::provider::{ObjectStoreProvider, Provider};
 use store_api::logstore::{AppendBatchResponse, EntryId, LogStore, SendableEntryStream, WalIndex};
 use store_api::storage::RegionId;
-#[cfg(any(test, feature = "testing"))]
-use tokio::sync::watch;
-use tokio::sync::{mpsc, oneshot};
+use tokio::sync::{mpsc, oneshot, watch};
 use tokio::time::MissedTickBehavior;
 
 use crate::error::{
@@ -46,7 +45,10 @@ use crate::error::{
     StaleWalObjectSnafu, UnconfirmedWalEpochStartSnafu, WalObjectSequenceExhaustedSnafu,
     WalObjectSequenceUnsettledSnafu,
 };
-use crate::metrics::METRIC_OBJECT_STORE_WAL_SKIPPED_SEGMENTS_TOTAL;
+use crate::metrics::{
+    METRIC_OBJECT_STORE_WAL_DELETED_OBJECTS_TOTAL, METRIC_OBJECT_STORE_WAL_FAILED_DELETES_TOTAL,
+    METRIC_OBJECT_STORE_WAL_SKIPPED_SEGMENTS_TOTAL,
+};
 use crate::object_store_wal::batch::{OBJECT_SEQ_LIMIT, OpenBatch, entry_id, sequence_floor};
 use crate::object_store_wal::catalog::ObjectCatalog;
 use crate::object_store_wal::format::{
@@ -61,6 +63,7 @@ const COMMAND_BUFFER: usize = 1024;
 const APPEND_BUFFER: usize = 16;
 const MIN_FLUSH_INTERVAL: Duration = Duration::from_millis(10);
 const MAX_IN_FLIGHT_CREATES: usize = 4;
+const MAX_IN_FLIGHT_DELETES: usize = 4;
 /// Delay before a create that failed transiently is attempted again in the
 /// `enqueued` acknowledgement mode, where no caller is left to retry it.
 const CREATE_RETRY_DELAY: Duration = Duration::from_millis(100);
@@ -89,7 +92,8 @@ const RECOVERY_TAIL_WINDOW: usize = 64 * 1024;
 /// `enqueued` mode it returns on admission and the object is created in the
 /// background. While [`MAX_SEALED_BATCHES`] batches
 /// wait to become durable no append is admitted. A segment that a read cannot
-/// decode is handled as `on_corrupted_segment` says.
+/// decode is handled as `on_corrupted_segment` says. Objects below the reclaim
+/// boundary of the last indexed object are deleted in the background.
 pub struct ObjectStoreLogStore {
     prefix: String,
     ack_mode: AckMode,
@@ -99,6 +103,8 @@ pub struct ObjectStoreLogStore {
     obsolete_entry_ids: ObsoleteEntryIds,
     /// Segments that reads skipped, per region and by object sequence.
     wal_holes: WalHoles,
+    /// Objects with a delete in flight, which a read consults.
+    deleting: Arc<DeletingObjects>,
     terminal_error: TerminalError,
     stopped: Arc<AtomicBool>,
     command_tx: mpsc::Sender<Command>,
@@ -135,6 +141,72 @@ pub struct WalHole {
 
 type TerminalError = Arc<Mutex<Option<Arc<Error>>>>;
 
+/// The delete attempts in flight, one per object.
+///
+/// The actor registers an attempt before its delete starts, unindexes the
+/// object if the delete succeeded, and only then settles the attempt and takes
+/// it out, so an object without an attempt was either never collected or is
+/// already unindexed. [`is_collected`] depends on that order.
+#[derive(Debug, Default)]
+struct DeletingObjects {
+    /// The settlement flag of the attempt in flight per object sequence.
+    attempts: Mutex<BTreeMap<u64, watch::Sender<bool>>>,
+}
+
+impl DeletingObjects {
+    fn lock(&self) -> std::sync::MutexGuard<'_, BTreeMap<u64, watch::Sender<bool>>> {
+        self.attempts.lock().unwrap_or_else(PoisonError::into_inner)
+    }
+
+    fn start(&self, object_seq: u64) {
+        self.lock().insert(object_seq, watch::channel(false).0);
+    }
+
+    fn contains(&self, object_seq: u64) -> bool {
+        self.lock().contains_key(&object_seq)
+    }
+
+    /// Settles the attempt of `object_seq` and takes it out, releasing the
+    /// reads that wait for it.
+    fn finish(&self, object_seq: u64) {
+        if let Some(settled) = self.lock().remove(&object_seq) {
+            settled.send_replace(true);
+        }
+    }
+
+    /// Settles every attempt the actor will never complete because it exited:
+    /// their objects stay indexed, so waiting reads report their error.
+    fn abandon_all(&self) {
+        for (_, settled) in std::mem::take(&mut *self.lock()) {
+            settled.send_replace(true);
+        }
+    }
+
+    /// Returns the settlement of the attempt in flight for `object_seq`, or
+    /// `None` when no delete of it is in flight.
+    fn attempt(&self, object_seq: u64) -> Option<watch::Receiver<bool>> {
+        self.lock().get(&object_seq).map(watch::Sender::subscribe)
+    }
+
+    /// Waits until no delete is in flight.
+    #[cfg(any(test, feature = "testing"))]
+    async fn wait_until_empty(&self) {
+        loop {
+            let in_flight = self
+                .lock()
+                .values()
+                .map(watch::Sender::subscribe)
+                .collect::<Vec<_>>();
+            if in_flight.is_empty() {
+                return;
+            }
+            for mut settled in in_flight {
+                let _ = settled.wait_for(|settled| *settled).await;
+            }
+        }
+    }
+}
+
 impl fmt::Debug for ObjectStoreLogStore {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_struct("ObjectStoreLogStore")
@@ -193,6 +265,8 @@ impl ObjectStoreLogStore {
             durable_entry_ids,
             tip,
             max_epoch,
+            reclaim_boundary,
+            unindexed,
         } = recover(io.as_ref()).await?;
         // Every epoch is one above the sequence of the start object its
         // instance created, and every object is at or above its start object,
@@ -205,7 +279,8 @@ impl ObjectStoreLogStore {
                 ),
             }
         );
-        let start = start_epoch(io.as_ref(), next_object_seq, tip).await?;
+        let (start, moved_past) =
+            start_epoch(io.as_ref(), next_object_seq, tip, reclaim_boundary).await?;
         let epoch = start.epoch;
         // The epoch identifies this instance in every object it writes.
         info!(
@@ -217,8 +292,12 @@ impl ObjectStoreLogStore {
             .with_context(|_| InvalidWalObjectSnafu {
                 path: io.object_path(start.object_seq),
             })?;
+        // No watermark is recorded yet.
+        let writer_reclaim_boundary = catalog.reclaim_boundary(reclaim_boundary, &HashMap::new());
         let catalog = Arc::new(RwLock::new(catalog));
         let obsolete_entry_ids = ObsoleteEntryIds::default();
+        let wal_holes = WalHoles::default();
+        let deleting = Arc::new(DeletingObjects::default());
         let terminal_error = TerminalError::default();
         let stopped = Arc::new(AtomicBool::new(false));
         let (command_tx, command_rx) = mpsc::channel(COMMAND_BUFFER);
@@ -240,6 +319,7 @@ impl ObjectStoreLogStore {
             io: io.clone(),
             catalog: catalog.clone(),
             obsolete_entry_ids: obsolete_entry_ids.clone(),
+            wal_holes: wal_holes.clone(),
             terminal_error: terminal_error.clone(),
             stopped: stopped.clone(),
             command_rx,
@@ -262,6 +342,13 @@ impl ObjectStoreLogStore {
                 .filter(|next_object_seq| *next_object_seq < OBJECT_SEQ_LIMIT),
             epoch,
             last_indexed: start,
+            reclaim_boundary: writer_reclaim_boundary,
+            durable_reclaim_boundary: reclaim_boundary,
+            unindexed: unindexed.into_iter().chain(moved_past).collect(),
+            creating: HashSet::new(),
+            deleting: deleting.clone(),
+            deletes: FuturesUnordered::new(),
+            failed_deletes: HashSet::new(),
             flush_interval: config.flush_interval,
             #[cfg(any(test, feature = "testing"))]
             admitted_appends: admitted_appends_tx,
@@ -284,7 +371,8 @@ impl ObjectStoreLogStore {
             io,
             catalog,
             obsolete_entry_ids,
-            wal_holes: WalHoles::default(),
+            wal_holes,
+            deleting,
             terminal_error,
             stopped,
             command_tx,
@@ -509,6 +597,14 @@ impl ObjectStoreLogStore {
         }
     }
 
+    /// Waits until no object delete is in flight. The deletes a collection
+    /// starts are in flight by the time the `obsolete`, `obsolete_all`,
+    /// [`seal_open_batch`](Self::seal_open_batch) or `durable` append that
+    /// started it returns.
+    pub async fn wait_for_garbage_collection(&self) {
+        self.deleting.wait_until_empty().await;
+    }
+
     /// Sets the stopped flag without sending the stop command, which is the
     /// state a store is in between the two steps of [`stop`](LogStore::stop).
     pub fn begin_stop(&self) {
@@ -595,7 +691,8 @@ impl LogStore for ObjectStoreLogStore {
 
     /// Reads the provider's region from `entry_id`, hiding obsolete entries.
     /// The catalog locates each segment, so a caller-supplied index is unnecessary.
-    /// A segment that does not decode fails the read or is skipped and recorded
+    /// An object collected after the read listed it is skipped. Otherwise a
+    /// segment that does not decode fails the read or is skipped and recorded
     /// as a hole, as `on_corrupted_segment` says; a failed fetch always fails
     /// the read.
     async fn read(
@@ -631,6 +728,8 @@ impl LogStore for ObjectStoreLogStore {
         };
 
         let io = self.io.clone();
+        let catalog = self.catalog.clone();
+        let deleting = self.deleting.clone();
         let provider = provider.clone();
         let on_corrupted_segment = self.on_corrupted_segment;
         let wal_holes = self.wal_holes.clone();
@@ -638,10 +737,18 @@ impl LogStore for ObjectStoreLogStore {
             for (object_seq, footer_entry) in objects {
                 let records = match fetch_segment(io.as_ref(), object_seq, &footer_entry).await {
                     Ok(records) => records,
+                    Err(error) if is_collected(&catalog, &deleting, object_seq).await => {
+                        debug!(
+                            "Skipped WAL object {} (sequence {}), collected after the read listed it: {error}",
+                            io.object_path(object_seq),
+                            object_seq
+                        );
+                        continue;
+                    }
                     Err(error @ Error::InvalidWalObject { .. })
                         if on_corrupted_segment == CorruptedSegmentAction::Skip =>
                     {
-                        record_skipped_segment(&wal_holes, io.as_ref(), object_seq, &footer_entry, &error);
+                        record_skipped_segment(&catalog, &wal_holes, io.as_ref(), object_seq, &footer_entry, &error);
                         continue;
                     }
                     Err(error) => Err(error)?,
@@ -742,11 +849,22 @@ impl LogStore for ObjectStoreLogStore {
         }
     }
 
-    /// Hides every entry of the region in memory. The caller re-establishes
-    /// the watermark after a restart.
+    /// Hides every entry of the region in memory and lets collection delete
+    /// its objects. The caller re-establishes the watermark after a restart.
     async fn obsolete_all(&self, provider: &Provider, region_id: RegionId) -> Result<()> {
         self.check_region(provider, region_id)?;
         record_obsolete(&self.obsolete_entry_ids, region_id, EntryId::MAX);
+        let (response_tx, response_rx) = oneshot::channel();
+        if self
+            .command_tx
+            .send(Command::Collect {
+                response: response_tx,
+            })
+            .await
+            .is_ok()
+        {
+            let _ = response_rx.await;
+        }
         Ok(())
     }
 
@@ -813,6 +931,8 @@ enum Command {
         watermark: EntryId,
         response: oneshot::Sender<Result<()>>,
     },
+    /// Answered once the collection the recorded watermarks allow started.
+    Collect { response: oneshot::Sender<()> },
     Stop {
         response: oneshot::Sender<Result<()>>,
     },
@@ -850,6 +970,7 @@ enum CreateState {
 /// created, indexed and acknowledged.
 struct SealedBatch {
     object_seq: u64,
+    reclaim_boundary: u64,
     bytes: Bytes,
     footer: Vec<FooterEntry>,
     first_admitted_at: Instant,
@@ -910,10 +1031,22 @@ type CreateOutcome = (u64, Result<PutResult>);
 /// [`MAX_SEALED_BATCHES`] batches wait or an append is held back at a backlog
 /// threshold of the `enqueued` mode, so commands such as `stop` are handled
 /// while admission is held back.
+///
+/// The actor also collects garbage. Every batch carries the reclaim boundary
+/// current when it was sealed, see [`ObjectCatalog::reclaim_boundary`], which
+/// is recomputed after every `obsolete` and every indexed object and never
+/// moves down. The boundary of the last indexed object is durable: every
+/// present object below it, indexed or not, is deleted in the background, up
+/// to [`MAX_IN_FLIGHT_DELETES`] at a time, after an `obsolete` or an indexing
+/// that moved it. An indexed object is unindexed once its delete succeeded; a
+/// failed delete is attempted again at the next collection. Nothing is
+/// collected once stop began or the store is poisoned, and deletes in flight
+/// complete before `stop` returns.
 struct Actor {
     io: Arc<dyn WalObjectIo>,
     catalog: Arc<RwLock<ObjectCatalog>>,
     obsolete_entry_ids: ObsoleteEntryIds,
+    wal_holes: WalHoles,
     terminal_error: TerminalError,
     stopped: Arc<AtomicBool>,
     command_rx: mpsc::Receiver<Command>,
@@ -950,6 +1083,19 @@ struct Actor {
     /// The last indexed object, which is the start object until a batch is
     /// indexed. A batch sealed while no earlier batch waits extends it.
     last_indexed: ChainLink,
+    /// The reclaim boundary the next sealed batch carries.
+    reclaim_boundary: u64,
+    /// The reclaim boundary of the last indexed object.
+    durable_reclaim_boundary: u64,
+    /// Sequences of objects that may be present but are not indexed: the
+    /// objects recovery listed off the chain and the batches that failed.
+    unindexed: BTreeSet<u64>,
+    /// Sequences whose create is in flight.
+    creating: HashSet<u64>,
+    deleting: Arc<DeletingObjects>,
+    deletes: FuturesUnordered<BoxFuture<'static, (u64, Result<()>)>>,
+    /// Deletes that failed since the last collection started.
+    failed_deletes: HashSet<u64>,
     flush_interval: Duration,
     #[cfg(any(test, feature = "testing"))]
     admitted_appends: watch::Sender<usize>,
@@ -984,6 +1130,9 @@ impl Actor {
                 Some((object_seq, result)) = self.creates.next(), if !self.creates.is_empty() => {
                     self.on_create_completed(object_seq, result);
                 }
+                Some((object_seq, result)) = self.deletes.next(), if !self.deletes.is_empty() => {
+                    self.on_delete_completed(object_seq, result);
+                }
                 Some((entries, response)) = self.append_rx.recv(), if self.stalled.is_none() && self.sealed.len() + 2 <= MAX_SEALED_BATCHES => {
                     self.handle_append(entries, response);
                 }
@@ -994,6 +1143,11 @@ impl Actor {
                     Some(Command::Obsolete { region_id, entry_id, watermark, response }) => {
                         self.handle_obsolete(region_id, entry_id, watermark, response);
                     }
+                    Some(Command::Collect { response }) => {
+                        self.update_reclaim_boundary();
+                        self.collect();
+                        let _ = response.send(());
+                    }
                     Some(Command::Stop { response }) => {
                         self.handle_stop(response);
                     }
@@ -1178,6 +1332,7 @@ impl Actor {
                         epoch: self.epoch,
                     }),
             ),
+            reclaim_boundary: self.reclaim_boundary,
         };
         let encoded = match encode_batch(header, entries) {
             Ok(encoded) => encoded,
@@ -1202,6 +1357,7 @@ impl Actor {
         }
         self.sealed.push_back(SealedBatch {
             object_seq,
+            reclaim_boundary: self.reclaim_boundary,
             bytes: encoded.bytes,
             footer: encoded.footer,
             first_admitted_at,
@@ -1241,6 +1397,7 @@ impl Actor {
             batch.attempts += 1;
             let io = self.io.clone();
             let object_seq = batch.object_seq;
+            self.creating.insert(object_seq);
             let bytes = batch.bytes.clone();
             let epoch = self.epoch;
             let ack_mode = self.ack_mode;
@@ -1290,14 +1447,17 @@ impl Actor {
     }
 
     fn on_create_completed(&mut self, object_seq: u64, result: Result<PutResult>) {
+        self.creating.remove(&object_seq);
         // A batch that already failed, or that the store gave up on when it
-        // poisoned itself: the object may exist, but it is off the chain.
+        // poisoned itself: the object may exist, but it is off the chain, and
+        // collection may delete it now.
         let Some(index) = self
             .sealed
             .iter()
             .position(|batch| batch.object_seq == object_seq)
         else {
             self.start_creates();
+            self.start_deletes();
             return;
         };
         match result {
@@ -1368,6 +1528,13 @@ impl Actor {
             object_seq: batch.object_seq,
             epoch: self.epoch,
         };
+        // The collection starts before the waiters learn that the object is
+        // indexed.
+        self.update_reclaim_boundary();
+        if batch.reclaim_boundary > self.durable_reclaim_boundary {
+            self.durable_reclaim_boundary = batch.reclaim_boundary;
+            self.collect();
+        }
         for waiter in batch.waiters {
             let _ = waiter.response.send(Ok(AppendBatchResponse {
                 last_entry_ids: waiter.last_entry_ids,
@@ -1434,6 +1601,7 @@ impl Actor {
             }
         };
         for batch in self.sealed.drain(..) {
+            self.unindexed.insert(batch.object_seq);
             batch.fail(failure);
         }
         self.reset_open_batch();
@@ -1546,9 +1714,105 @@ impl Actor {
         let result = self.raise_sequence_floor(region_id, entry_id).map(|_| {
             record_obsolete(&self.obsolete_entry_ids, region_id, watermark);
         });
+        if result.is_ok() {
+            self.update_reclaim_boundary();
+            self.collect();
+        }
         let _ = response.send(result);
     }
 
+    /// Raises the reclaim boundary to what the catalog and the obsolete
+    /// watermarks allow.
+    fn update_reclaim_boundary(&mut self) {
+        let boundary = {
+            let obsolete_entry_ids = self
+                .obsolete_entry_ids
+                .lock()
+                .unwrap_or_else(PoisonError::into_inner);
+            self.catalog
+                .read()
+                .unwrap_or_else(PoisonError::into_inner)
+                .reclaim_boundary(self.reclaim_boundary, &obsolete_entry_ids)
+        };
+        self.reclaim_boundary = boundary;
+    }
+
+    /// Starts a collection, which also attempts the deletes that failed since
+    /// the last one again.
+    fn collect(&mut self) {
+        self.failed_deletes.clear();
+        self.start_deletes();
+    }
+
+    /// Starts deleting the objects below the durable reclaim boundary while
+    /// fewer than [`MAX_IN_FLIGHT_DELETES`] deletes are in flight, skipping
+    /// objects whose create is in flight or whose delete failed since the
+    /// collection started.
+    fn start_deletes(&mut self) {
+        if self.is_stopped() || terminal(&self.terminal_error).is_some() {
+            return;
+        }
+        let free = MAX_IN_FLIGHT_DELETES.saturating_sub(self.deletes.len());
+        let boundary = self.durable_reclaim_boundary;
+        let candidates = {
+            let catalog = self.catalog.read().unwrap_or_else(PoisonError::into_inner);
+            catalog
+                .objects_in_order()
+                .map(|(object_seq, _)| object_seq)
+                .take_while(|object_seq| *object_seq < boundary)
+                .chain(self.unindexed.range(..boundary).copied())
+                .filter(|object_seq| {
+                    !self.deleting.contains(*object_seq)
+                        && !self.failed_deletes.contains(object_seq)
+                        && !self.creating.contains(object_seq)
+                })
+                .take(free)
+                .collect::<Vec<_>>()
+        };
+        for object_seq in candidates {
+            self.deleting.start(object_seq);
+            let io = self.io.clone();
+            self.deletes.push(Box::pin(async move {
+                (object_seq, io.delete(object_seq).await)
+            }));
+        }
+    }
+
+    /// Unindexes the object `object_seq` and drops its holes once its delete
+    /// succeeded, then settles its attempt, so a read that the attempt
+    /// releases sees the outcome in the catalog.
+    fn on_delete_completed(&mut self, object_seq: u64, result: Result<()>) {
+        match result {
+            Ok(()) => {
+                self.unindexed.remove(&object_seq);
+                let mut catalog = self.catalog.write().unwrap_or_else(PoisonError::into_inner);
+                let footer = catalog.remove_object(object_seq);
+                let mut wal_holes = self
+                    .wal_holes
+                    .lock()
+                    .unwrap_or_else(PoisonError::into_inner);
+                for entry in footer.iter().flatten() {
+                    if let Some(holes) = wal_holes.get_mut(&entry.region_id) {
+                        holes.remove(&object_seq);
+                    }
+                }
+                METRIC_OBJECT_STORE_WAL_DELETED_OBJECTS_TOTAL.inc();
+            }
+            Err(error) => {
+                warn!(
+                    error;
+                    "Failed to delete WAL object {} (sequence {}), it is deleted again at the next collection",
+                    self.io.object_path(object_seq),
+                    object_seq
+                );
+                METRIC_OBJECT_STORE_WAL_FAILED_DELETES_TOTAL.inc();
+                self.failed_deletes.insert(object_seq);
+            }
+        }
+        self.start_deletes();
+        self.deleting.finish(object_seq);
+    }
+
     /// Moves the sequence of the next object above the object that holds
     /// `entry_id` unless it is there already. The move fails while the open
     /// batch has handed out ids under the next sequence. A floor that does
@@ -1633,9 +1897,14 @@ impl Actor {
     }
 
     /// Answers the callers of `stop` once every sealed batch is settled and
-    /// every create has completed. Returns true when the actor is done.
+    /// every create and delete has completed. Returns true when the actor is
+    /// done.
     fn finish_stop(&mut self) -> bool {
-        if self.stop.is_empty() || !self.sealed.is_empty() || !self.creates.is_empty() {
+        if self.stop.is_empty()
+            || !self.sealed.is_empty()
+            || !self.creates.is_empty()
+            || !self.deletes.is_empty()
+        {
             return false;
         }
         for waiter in self.durable_waiters.drain(..) {
@@ -1686,6 +1955,14 @@ impl Actor {
     }
 }
 
+impl Drop for Actor {
+    /// Settles the deletes the actor still had in flight, so no read waits
+    /// for an attempt that will never complete.
+    fn drop(&mut self) {
+        self.deleting.abandon_all();
+    }
+}
+
 fn terminal(terminal_error: &TerminalError) -> Option<Arc<Error>> {
     terminal_error
         .lock()
@@ -1715,6 +1992,25 @@ fn shared(error: &Arc<Error>) -> Error {
     ObjectStoreWalSnafu.into_error(error.clone())
 }
 
+/// Returns whether collection deleted the object `object_seq` after a read
+/// listed it. The set of deleting objects is observed before the catalog, the
+/// reverse of the order the actor updates them, so an object that no longer
+/// looks deleting but still looks indexed was never collected. An undecided
+/// delete is waited for.
+async fn is_collected(
+    catalog: &RwLock<ObjectCatalog>,
+    deleting: &DeletingObjects,
+    object_seq: u64,
+) -> bool {
+    if let Some(mut settled) = deleting.attempt(object_seq) {
+        let _ = settled.wait_for(|settled| *settled).await;
+    }
+    !catalog
+        .read()
+        .unwrap_or_else(PoisonError::into_inner)
+        .contains_object(object_seq)
+}
+
 /// Fetches and decodes the segment `entry` describes in the object
 /// `object_seq`. A segment that does not decode is fetched once more, since
 /// the download rather than the object may be damaged; when it still does not
@@ -1735,14 +2031,27 @@ async fn fetch_segment(
 }
 
 /// Records the segment `entry` describes in the object `object_seq` as a hole
-/// of its region, once per object, counts the skip and warns about it.
+/// of its region, once per object, counts the skip and warns about it. An
+/// object collected meanwhile is skipped without a hole: the catalog is held
+/// while the hole is recorded, as a collection holds it while it drops the
+/// holes of the object it removes.
 fn record_skipped_segment(
+    catalog: &RwLock<ObjectCatalog>,
     wal_holes: &WalHoles,
     io: &dyn WalObjectIo,
     object_seq: u64,
     entry: &FooterEntry,
     error: &Error,
 ) {
+    let catalog = catalog.read().unwrap_or_else(PoisonError::into_inner);
+    if !catalog.contains_object(object_seq) {
+        debug!(
+            "Skipped WAL object {} (sequence {}), collected after the read listed it: {error}",
+            io.object_path(object_seq),
+            object_seq
+        );
+        return;
+    }
     let hole = WalHole {
         path: io.object_path(object_seq),
         object_seq,
@@ -1791,6 +2100,19 @@ struct Recovered {
     tip: Option<ChainLink>,
     /// Largest epoch any present object carries, zero on an empty prefix.
     max_epoch: u64,
+    /// The reclaim boundary of the tip, zero on an empty prefix.
+    reclaim_boundary: u64,
+    /// Present objects that are not indexed, in sequence order.
+    unindexed: Vec<u64>,
+}
+
+/// The objects recovery fetched, ordered by sequence, and the listed objects
+/// that were gone when their header or footer was fetched, with the error the
+/// fetch met. A collection of an earlier instance may complete after the
+/// listing.
+struct FetchedObjects {
+    objects: Vec<FetchedObject>,
+    vanished: Vec<(u64, Error)>,
 }
 
 /// An object as recovery fetched it: its key, header and footer.
@@ -1812,18 +2134,22 @@ async fn recover(io: &dyn WalObjectIo) -> Result<Recovered> {
     finish_recovery(fetch_footers(io, objects, RECOVERY_CONCURRENCY).await?)
 }
 
-/// Indexes the chain of `objects`, which are ordered by sequence. Objects off
-/// the chain are orphans: they are not indexed, but no later object takes
-/// their sequences.
-fn finish_recovery(objects: Vec<FetchedObject>) -> Result<Recovered> {
+/// Indexes the chain of the fetched objects. Objects off the chain, orphans
+/// and objects below the reclaim boundary of the tip, are not indexed, but no
+/// later object takes their sequences, nor the sequences of vanished objects.
+/// A vanished object at or above the boundary of the tip fails recovery with
+/// the error its fetch met: collection never deletes such an object.
+fn finish_recovery(fetched: FetchedObjects) -> Result<Recovered> {
+    let FetchedObjects { objects, vanished } = fetched;
     let headers = objects
         .iter()
         .map(|fetched| (fetched.object.object_seq, &fetched.header))
         .collect::<BTreeMap<_, _>>();
     let chain = select_chain(&headers);
     // Every open writes an object that starts a chain or extends a complete
-    // one, and nothing removes objects, so present objects without any
-    // complete chain mean objects of the chain are gone.
+    // one, and collection removes only objects below the reclaim boundary of
+    // a later object, so present objects without any complete chain mean
+    // objects of the chain are gone.
     ensure!(
         headers.is_empty() || !chain.is_empty(),
         CorruptedWalObjectSnafu {
@@ -1842,20 +2168,35 @@ fn finish_recovery(objects: Vec<FetchedObject>) -> Result<Recovered> {
         .map(|header| header.epoch)
         .max()
         .unwrap_or(0);
-    let after_listed = match headers.last_key_value() {
+    let last_listed = headers
+        .keys()
+        .copied()
+        .chain(vanished.iter().map(|(object_seq, _)| *object_seq))
+        .max();
+    let after_listed = match last_listed {
         None => 0,
-        Some((&last_object_seq, _)) => last_object_seq
+        Some(last_object_seq) => last_object_seq
             .checked_add(1)
             .context(WalObjectSequenceExhaustedSnafu { last_object_seq })?,
     };
+    let reclaim_boundary = tip.map_or(0, |tip| headers[&tip.object_seq].reclaim_boundary);
+    if let Some((_, error)) = vanished
+        .into_iter()
+        .find(|(object_seq, _)| *object_seq >= reclaim_boundary)
+    {
+        return Err(error);
+    }
     let chain = chain.into_iter().collect::<HashSet<_>>();
 
     let mut catalog = ObjectCatalog::default();
+    let mut unindexed = Vec::new();
     for FetchedObject { object, footer, .. } in objects {
         if chain.contains(&object.object_seq) {
             catalog
                 .insert_object(object.object_seq, footer)
                 .with_context(|_| InvalidWalObjectSnafu { path: object.path })?;
+        } else {
+            unindexed.push(object.object_seq);
         }
     }
     let next_object_seq = catalog.next_object_seq()?.max(after_listed);
@@ -1872,15 +2213,20 @@ fn finish_recovery(objects: Vec<FetchedObject>) -> Result<Recovered> {
         durable_entry_ids,
         tip,
         max_epoch,
+        reclaim_boundary,
+        unindexed,
     })
 }
 
 /// Returns, in sequence order, the objects on the chain that ends at the tip:
 /// the complete object with the largest epoch, then the largest sequence.
 ///
-/// An object is complete when every link on its chain holds: the chain starts
-/// at an object without a predecessor, and every other link names a present
-/// object that carries the epoch the link records. Only one instance writes
+/// An object is complete when every link on its chain holds, judged with the
+/// reclaim boundary of that object: the chain starts at an object without a
+/// predecessor or reaches a predecessor below the boundary, whether or not it
+/// is present, and every other link names a present object that carries the
+/// epoch the link records. The chain of the tip ends there, so objects below
+/// its boundary are not replayed. Only one instance writes
 /// under an epoch, so the epoch tells an object of the linking instance from
 /// another object under the same sequence. A create that
 /// was reported as failed may still leave its object, but every object
@@ -1888,43 +2234,43 @@ fn finish_recovery(objects: Vec<FetchedObject>) -> Result<Recovered> {
 /// an epoch above every object present when it opened, so neither such an
 /// object nor a late object of an earlier instance ends the chosen chain.
 fn select_chain(headers: &BTreeMap<u64, &Header>) -> Vec<u64> {
-    // A predecessor precedes its successor, so one pass in sequence order
-    // settles every object.
-    let mut complete = HashMap::with_capacity(headers.len());
-    for (&object_seq, header) in headers {
-        let holds = match header.prev {
-            None => true,
-            Some(link) if link.object_seq >= object_seq => false,
-            Some(link) => match headers.get(&link.object_seq) {
-                Some(prev) => prev.epoch == link.epoch && complete[&link.object_seq],
-                None => false,
-            },
-        };
-        complete.insert(object_seq, holds);
-    }
-    let Some(tip) = headers
-        .iter()
-        .filter(|(object_seq, _)| complete[*object_seq])
-        .max_by_key(|(object_seq, header)| (header.epoch, **object_seq))
-        .map(|(object_seq, _)| *object_seq)
-    else {
-        return Vec::new();
-    };
+    let mut candidates = headers.iter().collect::<Vec<_>>();
+    candidates.sort_unstable_by_key(|(object_seq, header)| Reverse((header.epoch, **object_seq)));
+    candidates
+        .into_iter()
+        .find_map(|(object_seq, _)| chain_ending_at(headers, *object_seq))
+        .unwrap_or_default()
+}
+
+/// Returns, in sequence order, the chain that ends at `tip` if `tip` is
+/// complete, see [`select_chain`].
+fn chain_ending_at(headers: &BTreeMap<u64, &Header>, tip: u64) -> Option<Vec<u64>> {
+    let reclaim_boundary = headers[&tip].reclaim_boundary;
     let mut chain = vec![tip];
-    while let Some(link) = headers[chain.last().expect("the chain holds the tip")].prev
-        && headers.contains_key(&link.object_seq)
-    {
+    let mut header = headers[&tip];
+    while let Some(link) = header.prev {
+        if link.object_seq >= header.object_seq {
+            return None;
+        }
+        if link.object_seq < reclaim_boundary {
+            break;
+        }
+        header = headers
+            .get(&link.object_seq)
+            .filter(|prev| prev.epoch == link.epoch)?;
         chain.push(link.object_seq);
     }
     chain.reverse();
-    chain
+    Some(chain)
 }
 
 /// Writes the empty object that starts the epoch of this instance at
-/// `object_seq`, linked to the recovered `tip`, and returns it as the tip
-/// later objects extend. The epoch is one above the sequence the create
-/// claims, so no two instances share one however their opens interleave, and
-/// it is above the epoch of every object recovery listed.
+/// `object_seq`, linked to the recovered `tip` and carrying the recovered
+/// `reclaim_boundary`, and returns it as the tip later objects extend,
+/// together with the sequences of earlier-epoch objects it moved past. The
+/// epoch is one above the sequence the create claims, so no two instances
+/// share one however their opens interleave, and it is above the epoch of
+/// every object recovery listed.
 ///
 /// An object an earlier instance left at that sequence after recovery listed
 /// the prefix has a lower epoch and never ends a chain, so the start object
@@ -1938,7 +2284,9 @@ async fn start_epoch(
     io: &dyn WalObjectIo,
     mut object_seq: u64,
     tip: Option<ChainLink>,
-) -> Result<ChainLink> {
+    reclaim_boundary: u64,
+) -> Result<(ChainLink, Vec<u64>)> {
+    let mut moved_past = Vec::new();
     loop {
         ensure!(
             object_seq < OBJECT_SEQ_LIMIT,
@@ -1951,12 +2299,13 @@ async fn start_epoch(
             object_seq,
             epoch,
             prev: tip,
+            reclaim_boundary,
         };
         match io
             .put_if_absent(object_seq, encode_object(header, &[])?.bytes)
             .await
         {
-            Ok(PutResult::Created) => return Ok(ChainLink { object_seq, epoch }),
+            Ok(PutResult::Created) => return Ok((ChainLink { object_seq, epoch }, moved_past)),
             Ok(PutResult::AlreadyPresent) => {
                 return UnconfirmedWalEpochStartSnafu {
                     path: io.object_path(object_seq),
@@ -1968,6 +2317,7 @@ async fn start_epoch(
                 if epoch_of(io, object_seq).await? >= epoch {
                     return Err(error);
                 }
+                moved_past.push(object_seq);
                 object_seq += 1;
             }
             Err(error) => return Err(error),
@@ -2007,27 +2357,43 @@ async fn stale_conflict(
 
 /// Fetches and verifies the headers and footers of `objects`, up to
 /// `concurrency` objects at a time, and returns them ordered by object sequence
-/// whatever the order the fetches complete in. The first failure abandons the
+/// whatever the order the fetches complete in. An object the object store no
+/// longer has is returned as vanished; any other failure abandons the
 /// remaining fetches.
 async fn fetch_footers(
     io: &dyn WalObjectIo,
     objects: Vec<ListedObject>,
     concurrency: usize,
-) -> Result<Vec<FetchedObject>> {
-    let mut fetched = futures::stream::iter(objects)
+) -> Result<FetchedObjects> {
+    let outcomes = futures::stream::iter(objects)
         .map(|object| async move {
-            let (header, footer) = fetch_footer(io, &object).await?;
-            Ok(FetchedObject {
-                object,
-                header,
-                footer,
-            })
+            match fetch_footer(io, &object).await {
+                Ok((header, footer)) => Ok(Ok(FetchedObject {
+                    object,
+                    header,
+                    footer,
+                })),
+                Err(error) if is_not_found(&error) => Ok(Err((object.object_seq, error))),
+                Err(error) => Err(error),
+            }
         })
         .buffer_unordered(concurrency)
         .try_collect::<Vec<_>>()
         .await?;
-    fetched.sort_unstable_by_key(|fetched| fetched.object.object_seq);
-    Ok(fetched)
+    let mut objects = Vec::with_capacity(outcomes.len());
+    let mut vanished = Vec::new();
+    for outcome in outcomes {
+        match outcome {
+            Ok(object) => objects.push(object),
+            Err(gone) => vanished.push(gone),
+        }
+    }
+    objects.sort_unstable_by_key(|fetched| fetched.object.object_seq);
+    Ok(FetchedObjects { objects, vanished })
+}
+
+fn is_not_found(error: &Error) -> bool {
+    matches!(error, Error::WalObjectStore { error, .. } if error.kind() == object_store::ErrorKind::NotFound)
 }
 
 /// Reads the header, trailer and footer of `object` and verifies them: the
@@ -2148,6 +2514,8 @@ pub(crate) trait WalObjectIo: Send + Sync {
 
     async fn get_range(&self, object_seq: u64, offset: u64, len: u64) -> Result<Bytes>;
 
+    async fn delete(&self, object_seq: u64) -> Result<()>;
+
     async fn list(&self) -> Result<Vec<ListedObject>>;
 
     fn object_path(&self, object_seq: u64) -> String;
@@ -2167,6 +2535,10 @@ impl WalObjectIo for ObjectStoreIo {
         ObjectStoreIo::get_range(self, object_seq, offset, len).await
     }
 
+    async fn delete(&self, object_seq: u64) -> Result<()> {
+        ObjectStoreIo::delete(self, object_seq).await
+    }
+
     async fn list(&self) -> Result<Vec<ListedObject>> {
         ObjectStoreIo::list(self).await
     }
@@ -2182,6 +2554,7 @@ mod tests {
 
     use common_base::readable_size::ReadableSize;
     use common_error::ext::{ErrorExt, RetryHint};
+    use futures::FutureExt;
     use object_store::services::Memory;
     use store_api::logstore::entry::{MultiplePartEntry, MultiplePartHeader};
     use tokio::time::timeout;
@@ -2197,6 +2570,9 @@ mod tests {
     /// counter, so that tests running in one process do not interleave
     /// their increments.
     static SKIPPED_SEGMENTS: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());
+    /// Held by every test that fails a delete, so that a test reading the
+    /// process-wide failed deletes counter sees only its own failures.
+    static FAILED_DELETES: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());
 
     fn memory_store() -> ObjectStore {
         ObjectStore::new(Memory::default()).unwrap()
@@ -2318,6 +2694,14 @@ mod tests {
             .collect()
     }
 
+    /// Lists the objects once no delete is in flight.
+    async fn collected_object_seqs(store: &ObjectStoreLogStore) -> Vec<u64> {
+        timeout(WAIT, store.wait_for_garbage_collection())
+            .await
+            .unwrap();
+        object_seqs(store.io.as_ref()).await
+    }
+
     fn unwrap_shared(error: &Error) -> &Error {
         match error {
             Error::ObjectStoreWal { source, .. } => source,
@@ -2332,22 +2716,22 @@ mod tests {
         );
     }
 
-    /// Waits for the next create that `ParkedIo` parked.
-    async fn next_create(
+    /// Waits for the next operation that `ParkedIo` parked.
+    async fn next_parked_operation(
         parked: &mut mpsc::UnboundedReceiver<(u64, oneshot::Sender<bool>)>,
     ) -> (u64, oneshot::Sender<bool>) {
         timeout(WAIT, parked.recv()).await.unwrap().unwrap()
     }
 
-    /// Waits for `count` parked creates and returns their releases by
+    /// Waits for `count` parked operations and returns their releases by
     /// sequence.
-    async fn parked_creates(
+    async fn parked_operations(
         parked: &mut mpsc::UnboundedReceiver<(u64, oneshot::Sender<bool>)>,
         count: usize,
     ) -> HashMap<u64, oneshot::Sender<bool>> {
         let mut releases = HashMap::new();
         for _ in 0..count {
-            let (object_seq, release) = next_create(parked).await;
+            let (object_seq, release) = next_parked_operation(parked).await;
             releases.insert(object_seq, release);
         }
         releases
@@ -2496,7 +2880,10 @@ mod tests {
                 footer: decoded.footer,
             });
         }
-        finish_recovery(objects)
+        finish_recovery(FetchedObjects {
+            objects,
+            vanished: Vec::new(),
+        })
     }
 
     fn assert_same_recovery(expected: &Recovered, actual: &Recovered) {
@@ -2508,6 +2895,8 @@ mod tests {
         assert_eq!(expected.durable_entry_ids, actual.durable_entry_ids);
         assert_eq!(expected.tip, actual.tip);
         assert_eq!(expected.max_epoch, actual.max_epoch);
+        assert_eq!(expected.reclaim_boundary, actual.reclaim_boundary);
+        assert_eq!(expected.unindexed, actual.unindexed);
     }
 
     fn catalog_contents(catalog: &ObjectCatalog) -> Vec<(u64, Vec<FooterEntry>)> {
@@ -2559,6 +2948,7 @@ mod tests {
                 object_seq: prev.object_seq,
                 epoch: prev.epoch,
             }),
+            reclaim_boundary: 0,
         }
     }
 
@@ -2581,6 +2971,7 @@ mod tests {
             object_seq,
             epoch,
             prev: None,
+            reclaim_boundary: 0,
         };
         put_object_with_header(&io, header, &[]).await;
     }
@@ -3017,6 +3408,7 @@ mod tests {
             catalog: Arc::default(),
             obsolete_entry_ids: ObsoleteEntryIds::default(),
             wal_holes: WalHoles::default(),
+            deleting: Arc::default(),
             terminal_error: Arc::default(),
             stopped: Arc::new(AtomicBool::new(false)),
             command_tx,
@@ -3105,6 +3497,7 @@ mod tests {
                 fetch_footers(io.as_ref(), objects, RECOVERY_CONCURRENCY)
                     .await
                     .unwrap()
+                    .objects
                     .into_iter()
                     .map(|fetched| fetched.object.object_seq)
                     .collect::<Vec<_>>()
@@ -3663,7 +4056,7 @@ mod tests {
             HashMap::from([(region_one, id(2, 1)), (region_two, id(2, 1))]),
             third.await.unwrap().unwrap().last_entry_ids
         );
-        assert_eq!(vec![0, 1, 2], object_seqs(store.io.as_ref()).await);
+        assert_eq!(vec![1, 2], collected_object_seqs(&store).await);
         assert_eq!(
             expected_entries(
                 region_one,
@@ -3704,7 +4097,7 @@ mod tests {
             HashMap::from([(region_id, id(1, POSITION_LIMIT - 1))]),
             response.last_entry_ids
         );
-        assert_eq!(vec![0, 1], object_seqs(store.io.as_ref()).await);
+        assert_eq!(vec![1], collected_object_seqs(&store).await);
 
         // An append that alone runs past the range fits no object: it seals
         // the open batch like any append the batch cannot take, then it is
@@ -3730,7 +4123,7 @@ mod tests {
             HashMap::from([(region_id, id(3, 1))]),
             response.last_entry_ids
         );
-        assert_eq!(vec![0, 1, 2, 3], object_seqs(store.io.as_ref()).await);
+        assert_eq!(vec![1, 2, 3], collected_object_seqs(&store).await);
         assert_eq!(
             expected_entries(region_id, &[(id(2, 1), "next"), (id(3, 1), "after")]),
             read_entries(&store, region_id, id(2, 1)).await
@@ -3753,16 +4146,16 @@ mod tests {
                 .await
                 .unwrap()
                 .unwrap();
-            let expected_seqs = (0..=index as u64 + 1).collect::<Vec<_>>();
-            assert_eq!(expected_seqs, object_seqs(store.io.as_ref()).await);
+            // Object 1 carries the boundary past the start object.
+            let expected_seqs = (1..=index as u64 + 1).collect::<Vec<_>>();
+            assert_eq!(expected_seqs, collected_object_seqs(&store).await);
         }
 
         // Ticks with an empty open batch do not create objects.
         tokio::time::sleep(interval * 5).await;
-        let seqs = object_seqs(store.io.as_ref()).await;
-        assert_eq!(vec![0, 1, 2, 3], seqs);
-        // Object 0 is the start object of the store.
-        for object_seq in seqs.into_iter().skip(1) {
+        let seqs = collected_object_seqs(&store).await;
+        assert_eq!(vec![1, 2, 3], seqs);
+        for object_seq in seqs {
             let bytes = store.io.get(object_seq).await.unwrap();
             let decoded = decode_object(&bytes).unwrap();
             assert_eq!(1, decoded.records.len(), "object {object_seq} is empty");
@@ -3829,7 +4222,7 @@ mod tests {
         let appends = spawn_appends(&store, region_id, MAX_IN_FLIGHT_CREATES + 2).await;
 
         // At most the limit of creates run at a time; the rest wait for a slot.
-        let mut releases = parked_creates(&mut parked, MAX_IN_FLIGHT_CREATES).await;
+        let mut releases = parked_operations(&mut parked, MAX_IN_FLIGHT_CREATES).await;
         assert_eq!(
             (1..=MAX_IN_FLIGHT_CREATES as u64).collect::<BTreeSet<_>>(),
             releases.keys().copied().collect::<BTreeSet<_>>()
@@ -3842,7 +4235,7 @@ mod tests {
         // each, but nothing is acknowledged ahead of object 1.
         for (released, expected_next) in [(3, 5), (2, 6)] {
             releases.remove(&released).unwrap().send(true).unwrap();
-            let (object_seq, release) = next_create(&mut parked).await;
+            let (object_seq, release) = next_parked_operation(&mut parked).await;
             assert_eq!(expected_next, object_seq);
             releases.insert(object_seq, release);
         }
@@ -3894,7 +4287,7 @@ mod tests {
             MAX_IN_FLIGHT_CREATES,
             io.max_in_flight.load(Ordering::SeqCst)
         );
-        assert_eq!(vec![0, 1, 2, 3, 4, 5, 6], object_seqs(io.as_ref()).await);
+        assert_eq!(vec![1, 2, 3, 4, 5, 6], collected_object_seqs(&store).await);
         // Every batch extends the batch sealed before it, whichever was
         // durable first.
         for object_seq in 1..=6 {
@@ -3922,11 +4315,11 @@ mod tests {
 
     #[tokio::test]
     async fn test_store_transient_failure_fails_every_batch_that_is_not_indexed() {
-        let (store, io, mut parked) = open_parking_creates(memory_store(), &eager()).await;
+        let (store, _, mut parked) = open_parking_creates(memory_store(), &eager()).await;
         let region_id = region(1);
         // Every slot is taken; the last batch waits for one.
         let appends = spawn_appends(&store, region_id, MAX_IN_FLIGHT_CREATES + 1).await;
-        let mut releases = parked_creates(&mut parked, MAX_IN_FLIGHT_CREATES).await;
+        let mut releases = parked_operations(&mut parked, MAX_IN_FLIGHT_CREATES).await;
         round_trip_actor(&store).await;
         assert!(parked.try_recv().is_err());
 
@@ -3948,14 +4341,14 @@ mod tests {
         // one retry starts until they complete; what they store is off the
         // chain.
         let retries = spawn_appends(&store, region_id, 2).await;
-        let (object_seq, first) = next_create(&mut parked).await;
+        let (object_seq, first) = next_parked_operation(&mut parked).await;
         assert_eq!(6, object_seq);
         round_trip_actor(&store).await;
         assert!(parked.try_recv().is_err());
         for release in releases.into_values() {
             release.send(true).unwrap();
         }
-        let (object_seq, second) = next_create(&mut parked).await;
+        let (object_seq, second) = next_parked_operation(&mut parked).await;
         assert_eq!(7, object_seq);
         first.send(true).unwrap();
         second.send(true).unwrap();
@@ -3966,7 +4359,7 @@ mod tests {
                 response.last_entry_ids
             );
         }
-        assert_eq!(vec![0, 2, 3, 4, 6, 7], object_seqs(io.as_ref()).await);
+        assert_eq!(vec![2, 3, 4, 6, 7], collected_object_seqs(&store).await);
         assert_eq!(
             expected_entries(region_id, &[(id(6, 1), "a1"), (id(7, 1), "a2")]),
             read_entries(&store, region_id, 1).await
@@ -3979,7 +4372,7 @@ mod tests {
         let (store, io, mut parked) = open_parking_creates(object_store.clone(), &eager()).await;
         let region_id = region(1);
         let appends = spawn_appends(&store, region_id, 2).await;
-        let mut releases = parked_creates(&mut parked, 2).await;
+        let mut releases = parked_operations(&mut parked, 2).await;
 
         // Object 2 is created while object 1 failed: both batches fail, and
         // the store keeps serving.
@@ -4003,7 +4396,7 @@ mod tests {
 
         // The retry extends the start object, not the failed batches.
         let retry = spawn_appends(&store, region_id, 1).await;
-        let (object_seq, release) = next_create(&mut parked).await;
+        let (object_seq, release) = next_parked_operation(&mut parked).await;
         assert_eq!(3, object_seq);
         release.send(true).unwrap();
         let response = timeout(WAIT, retry.into_iter().next().unwrap())
@@ -4045,7 +4438,7 @@ mod tests {
         // Another writer of the same epoch took sequence 1.
         put_foreign(&object_store, 1, 1).await;
         let appends = spawn_appends(&store, region_id, 2).await;
-        let mut releases = parked_creates(&mut parked, 2).await;
+        let mut releases = parked_operations(&mut parked, 2).await;
 
         // Object 2 is created; object 1 conflicts with the foreign object.
         releases.remove(&2).unwrap().send(true).unwrap();
@@ -4077,12 +4470,16 @@ mod tests {
     #[tokio::test]
     async fn test_store_conflict_with_an_earlier_epoch_fails_the_batch_without_poisoning() {
         let object_store = memory_store();
-        let (store, io, mut parked) = open_parking_creates(object_store.clone(), &eager()).await;
+        let (store, _, mut parked) = open_parking_creates(object_store.clone(), &eager()).await;
         let region_id = region(1);
         // A late object of an earlier epoch lands under sequence 1.
         put_foreign(&object_store, 1, 0).await;
         let append = spawn_appends(&store, region_id, 1).await;
-        next_create(&mut parked).await.1.send(true).unwrap();
+        next_parked_operation(&mut parked)
+            .await
+            .1
+            .send(true)
+            .unwrap();
         let error = timeout(WAIT, append.into_iter().next().unwrap())
             .await
             .unwrap()
@@ -4103,7 +4500,7 @@ mod tests {
 
         // The store keeps serving: the retry takes the next sequence.
         let retry = spawn_appends(&store, region_id, 1).await;
-        let (object_seq, release) = next_create(&mut parked).await;
+        let (object_seq, release) = next_parked_operation(&mut parked).await;
         assert_eq!(2, object_seq);
         release.send(true).unwrap();
         let response = timeout(WAIT, retry.into_iter().next().unwrap())
@@ -4115,7 +4512,7 @@ mod tests {
             HashMap::from([(region_id, id(2, 1))]),
             response.last_entry_ids
         );
-        assert_eq!(vec![0, 1, 2], object_seqs(io.as_ref()).await);
+        assert_eq!(vec![1, 2], collected_object_seqs(&store).await);
         store.stop().await.unwrap();
 
         // The stale object is off the chain after a restart.
@@ -4197,7 +4594,8 @@ mod tests {
             HashMap::from([(region_one, id(2, 2))]),
             response.last_entry_ids
         );
-        assert_eq!(vec![0, 1, 2], object_seqs(store.io.as_ref()).await);
+        // Object 2 carries the boundary past the watermark of region one.
+        assert_eq!(vec![2], collected_object_seqs(&store).await);
         store
             .obsolete(&provider(region_two), region_two, id(3, 7))
             .await
@@ -4214,7 +4612,7 @@ mod tests {
             HashMap::from([(region_two, id(4, 1))]),
             response.last_entry_ids
         );
-        assert_eq!(vec![0, 1, 2, 4], object_seqs(store.io.as_ref()).await);
+        assert_eq!(vec![2, 4], collected_object_seqs(&store).await);
         // The watermark hides nothing of this prefix; the region has no
         // entry at or below it.
         assert_eq!(
@@ -4260,11 +4658,12 @@ mod tests {
                 response.last_entry_ids
             );
         }
-        assert_eq!(vec![0, 1, 2, 3], object_seqs(io.as_ref()).await);
+        // Object 3 carries the boundary at object 2, so the failed object 1
+        // is collected.
+        assert_eq!(vec![2, 3], collected_object_seqs(&store).await);
         store.stop().await.unwrap();
 
-        // Object 2 extends the start object, so recovery leaves object 1 off
-        // the chain: only the retries replay.
+        // Only the retries replay.
         let store = open_over(io, &manual()).await;
         assert_eq!(
             expected_entries(region_one, &[(id(2, 1), "a1")]),
@@ -4279,11 +4678,11 @@ mod tests {
 
     #[tokio::test]
     async fn test_store_floor_applies_while_a_create_is_in_flight() {
-        let (store, io, mut parked) = open_parking_creates(memory_store(), &eager()).await;
+        let (store, _, mut parked) = open_parking_creates(memory_store(), &eager()).await;
         let region_one = region(1);
         let region_two = region(2);
         let pending = spawn_append_batch(&store, vec![entry(&store, region_one, "a1")]);
-        let (_, release) = next_create(&mut parked).await;
+        let (_, release) = next_parked_operation(&mut parked).await;
 
         // Object 1 is in flight and the open batch is empty: the floor moves
         // the next sequence, which a failure of object 1 does not move back.
@@ -4295,13 +4694,17 @@ mod tests {
         timeout(WAIT, pending).await.unwrap().unwrap().unwrap_err();
 
         let write = spawn_append_batch(&store, vec![entry(&store, region_two, "b1")]);
-        next_create(&mut parked).await.1.send(true).unwrap();
+        next_parked_operation(&mut parked)
+            .await
+            .1
+            .send(true)
+            .unwrap();
         let response = timeout(WAIT, write).await.unwrap().unwrap().unwrap();
         assert_eq!(
             HashMap::from([(region_two, id(6, 1))]),
             response.last_entry_ids
         );
-        assert_eq!(vec![0, 6], object_seqs(io.as_ref()).await);
+        assert_eq!(vec![6], collected_object_seqs(&store).await);
     }
 
     #[tokio::test]
@@ -4349,7 +4752,7 @@ mod tests {
         assert_eq!(vec![0], object_seqs(store.io.as_ref()).await);
         store.release_creates();
         timeout(WAIT, held).await.unwrap().unwrap().unwrap();
-        assert_eq!(vec![0, 1], object_seqs(store.io.as_ref()).await);
+        assert_eq!(vec![1], collected_object_seqs(&store).await);
 
         store.fail_creates();
         let error = append(&store, region_id, "a2").await.unwrap_err();
@@ -4357,7 +4760,7 @@ mod tests {
             matches!(unwrap_shared(&error), Error::WalObjectStore { .. }),
             "unexpected error: {error:?}"
         );
-        assert_eq!(vec![0, 1], object_seqs(store.io.as_ref()).await);
+        assert_eq!(vec![1], object_seqs(store.io.as_ref()).await);
 
         store.begin_stop();
         assert_stopped(&append(&store, region_id, "a3").await.unwrap_err());
@@ -4386,7 +4789,7 @@ mod tests {
             HashMap::from([(region_id, id(2, 1))]),
             response.last_entry_ids
         );
-        assert_eq!(vec![0, 1, 2], object_seqs(store.io.as_ref()).await);
+        assert_eq!(vec![1, 2], collected_object_seqs(&store).await);
         store.stop().await.unwrap();
     }
 
@@ -4455,7 +4858,7 @@ mod tests {
     #[tokio::test]
     async fn test_store_rollback_and_poison_drop_the_open_batch() {
         let object_store = memory_store();
-        let (store, io, mut parked) = open_parking_creates(object_store.clone(), &manual()).await;
+        let (store, _, mut parked) = open_parking_creates(object_store.clone(), &manual()).await;
         let region_id = region(1);
         let spawn_seal = || {
             let store = store.clone();
@@ -4467,7 +4870,7 @@ mod tests {
         let sealed = spawn_append_batch(&store, vec![entry(&store, region_id, "a1")]);
         store.wait_for_admitted_appends(1).await.unwrap();
         let seal = spawn_seal();
-        let (_, release) = next_create(&mut parked).await;
+        let (_, release) = next_parked_operation(&mut parked).await;
         let open = spawn_append_batch(&store, vec![entry(&store, region_id, "a2")]);
         store.wait_for_admitted_appends(2).await.unwrap();
         release.send(false).unwrap();
@@ -4485,7 +4888,11 @@ mod tests {
         let retry = spawn_append_batch(&store, vec![entry(&store, region_id, "b1")]);
         store.wait_for_admitted_appends(3).await.unwrap();
         let seal = spawn_seal();
-        next_create(&mut parked).await.1.send(true).unwrap();
+        next_parked_operation(&mut parked)
+            .await
+            .1
+            .send(true)
+            .unwrap();
         timeout(WAIT, seal).await.unwrap().unwrap().unwrap();
         let response = timeout(WAIT, retry).await.unwrap().unwrap().unwrap();
         assert_eq!(
@@ -4501,7 +4908,7 @@ mod tests {
         let sealed = spawn_append_batch(&store, vec![entry(&store, region_id, "c1")]);
         store.wait_for_admitted_appends(4).await.unwrap();
         let seal = spawn_seal();
-        let (_, release) = next_create(&mut parked).await;
+        let (_, release) = next_parked_operation(&mut parked).await;
         let open = spawn_append_batch(&store, vec![entry(&store, region_id, "c2")]);
         store.wait_for_admitted_appends(5).await.unwrap();
         put_foreign(&object_store, 3, 1).await;
@@ -4514,7 +4921,7 @@ mod tests {
             );
         }
         timeout(WAIT, seal).await.unwrap().unwrap().unwrap_err();
-        assert_eq!(vec![0, 2, 3], object_seqs(io.as_ref()).await);
+        assert_eq!(vec![2, 3], collected_object_seqs(&store).await);
         store.stop().await.unwrap();
     }
 
@@ -4562,7 +4969,7 @@ mod tests {
             let entries = vec![entry(&store, region_id, &format!("w{index}"))];
             appends.push(spawn_append_batch(&store, entries));
         }
-        let mut releases = parked_creates(&mut parked, MAX_IN_FLIGHT_CREATES).await;
+        let mut releases = parked_operations(&mut parked, MAX_IN_FLIGHT_CREATES).await;
         round_trip_actor(&store).await;
         assert_eq!(MAX_SEALED_BATCHES - 1, admitted());
 
@@ -4581,7 +4988,11 @@ mod tests {
         for release in releases.into_values() {
             release.send(true).unwrap();
         }
-        next_create(&mut parked).await.1.send(true).unwrap();
+        next_parked_operation(&mut parked)
+            .await
+            .1
+            .send(true)
+            .unwrap();
         timeout(WAIT, stop).await.unwrap().unwrap().unwrap();
         let mut acknowledged = 0;
         for append in appends {
@@ -4611,7 +5022,7 @@ mod tests {
     async fn test_store_stop_during_failed_flush_reports_stopped() {
         let (store, io, mut parked) = open_parking_creates(memory_store(), &eager()).await;
         let pending = spawn_append_batch(&store, vec![entry(&store, region(1), "a1")]);
-        let (_, release) = next_create(&mut parked).await;
+        let (_, release) = next_parked_operation(&mut parked).await;
         let stop = begin_spawned_stop(&store).await;
         assert!(!stop.is_finished());
 
@@ -4629,7 +5040,7 @@ mod tests {
         let region_id = region(1);
         // Four creates are in flight, the fifth batch waits for a slot.
         let appends = spawn_appends(&store, region_id, MAX_IN_FLIGHT_CREATES + 1).await;
-        let mut releases = parked_creates(&mut parked, MAX_IN_FLIGHT_CREATES).await;
+        let mut releases = parked_operations(&mut parked, MAX_IN_FLIGHT_CREATES).await;
         assert!(parked.try_recv().is_err());
 
         let stop = begin_spawned_stop(&store).await;
@@ -4682,7 +5093,7 @@ mod tests {
         let (store, _, mut parked) = open_parking_creates(object_store.clone(), &eager()).await;
         let region_id = region(1);
         let pending = spawn_append_batch(&store, vec![entry(&store, region_id, "a1")]);
-        let (_, release) = next_create(&mut parked).await;
+        let (_, release) = next_parked_operation(&mut parked).await;
         let stop = begin_spawned_stop(&store).await;
         // Another writer of the same epoch takes the sequence before the
         // create runs.
@@ -4719,6 +5130,7 @@ mod tests {
             object_seq,
             epoch,
             prev: prev.map(|(object_seq, epoch)| ChainLink { object_seq, epoch }),
+            reclaim_boundary: 0,
         }
     }
 
@@ -4802,6 +5214,29 @@ mod tests {
         );
         // A predecessor at or above the object itself never holds.
         assert!(chain_of(&[chain_header(5, 1, Some((5, 1)))]).is_empty());
+
+        // A link below the reclaim boundary of the object it is judged for
+        // holds whether or not the predecessor is present; above it a
+        // missing predecessor still breaks the link.
+        let bounded = |object_seq, prev, reclaim_boundary| Header {
+            reclaim_boundary,
+            ..chain_header(object_seq, 1, prev)
+        };
+        assert_eq!(
+            vec![3, 4],
+            chain_of(&[bounded(3, Some((2, 1)), 3), bounded(4, Some((3, 1)), 3)])
+        );
+        assert!(chain_of(&[bounded(4, Some((3, 1)), 3)]).is_empty());
+        // Object 3 is incomplete with its own boundary, but object 4 only
+        // needs it to be present.
+        assert_eq!(
+            vec![3, 4],
+            chain_of(&[bounded(3, Some((2, 1)), 0), bounded(4, Some((3, 1)), 3)])
+        );
+        assert_eq!(
+            vec![4],
+            chain_of(&[bounded(3, Some((2, 1)), 0), bounded(4, Some((3, 1)), 4)])
+        );
     }
 
     #[tokio::test]
@@ -4948,10 +5383,15 @@ mod tests {
             });
             let result = ObjectStoreLogStore::open(io, &eager(), PREFIX.to_string()).await;
             if moves {
-                result.unwrap().stop().await.unwrap();
+                let store = result.unwrap();
                 let start = decode_header(&fixtures.get(2).await.unwrap()).unwrap();
                 assert_eq!(3, start.epoch);
                 assert_eq!(Some(0), start.prev.map(|prev| prev.object_seq));
+                // Object 3 carries the boundary past the late object, which
+                // recovery never listed.
+                append(&store, region(1), "a").await.unwrap();
+                assert_eq!(vec![3], collected_object_seqs(&store).await);
+                store.stop().await.unwrap();
             } else {
                 let error = result.unwrap_err();
                 assert!(
@@ -5024,9 +5464,15 @@ mod tests {
 
         // Open A resumes at the sequence it recovered: the epoch follows from
         // the sequence it claims, so it differs from the epoch of open B.
-        let a = start_epoch(&io, recovered.next_object_seq, recovered.tip)
-            .await
-            .unwrap();
+        let a = start_epoch(
+            &io,
+            recovered.next_object_seq,
+            recovered.tip,
+            recovered.reclaim_boundary,
+        )
+        .await
+        .unwrap()
+        .0;
         assert_eq!(
             ChainLink {
                 object_seq: 1,
@@ -5113,7 +5559,7 @@ mod tests {
         );
 
         store.seal_open_batch().await.unwrap();
-        assert_eq!(vec![0, 1], object_seqs(store.io.as_ref()).await);
+        assert_eq!(vec![1], collected_object_seqs(&store).await);
         assert_eq!(id(1, 2), latest(&store, region_id));
         assert_eq!(
             id(1, 2),
@@ -5127,8 +5573,7 @@ mod tests {
 
     #[tokio::test]
     async fn test_store_enqueued_durable_id_advances_after_create_and_indexing() {
-        let (store, io, mut parked) =
-            open_parking_creates(memory_store(), &enqueued(eager())).await;
+        let (store, _, mut parked) = open_parking_creates(memory_store(), &enqueued(eager())).await;
         let region_one = region(1);
         let region_two = region(2);
 
@@ -5140,7 +5585,7 @@ mod tests {
             HashMap::from([(region_one, id(1, 1))]),
             response.last_entry_ids
         );
-        let (object_seq, release) = next_create(&mut parked).await;
+        let (object_seq, release) = next_parked_operation(&mut parked).await;
         assert_eq!(1, object_seq);
         assert_eq!(0, store.durable_entry_id(&provider(region_one)).unwrap());
         let wait = |entry_id| {
@@ -5169,7 +5614,7 @@ mod tests {
             store.durable_entry_id(&provider(region_one)).unwrap()
         );
         assert_eq!(0, store.durable_entry_id(&provider(region_two)).unwrap());
-        assert_eq!(vec![0, 1], object_seqs(io.as_ref()).await);
+        assert_eq!(vec![1], collected_object_seqs(&store).await);
     }
 
     #[tokio::test]
@@ -5181,7 +5626,7 @@ mod tests {
         // Object 1 holds region A's id(1, 1) and object 2 region B's id(2, 1).
         for (region_id, data) in [(region_a, "a1"), (region_b, "b1")] {
             let response = append(&store, region_id, data).await.unwrap();
-            let (_, release) = next_create(&mut parked).await;
+            let (_, release) = next_parked_operation(&mut parked).await;
             release.send(true).unwrap();
             let entry_id = response.last_entry_ids[&region_id];
             timeout(WAIT, store.wait_durable(&provider(region_id), entry_id))
@@ -5194,7 +5639,7 @@ mod tests {
             HashMap::from([(region_a, id(3, 1))]),
             response.last_entry_ids
         );
-        let (object_seq, release) = next_create(&mut parked).await;
+        let (object_seq, release) = next_parked_operation(&mut parked).await;
         assert_eq!(3, object_seq);
 
         // Every entry of region A up to id(2, 1) is durable, so the wait does
@@ -5284,7 +5729,7 @@ mod tests {
 
         let stalled = spawn_append_batch(&store, vec![entry(&store, region_id, "a2")]);
         // The stall seals the open batch so that an upload is in flight.
-        let (object_seq, release) = next_create(&mut parked).await;
+        let (object_seq, release) = next_parked_operation(&mut parked).await;
         assert_eq!(1, object_seq);
         for _ in 0..16 {
             tokio::task::yield_now().await;
@@ -5300,7 +5745,7 @@ mod tests {
             max_unpersisted_bytes: ReadableSize(1),
             ..enqueued(manual())
         };
-        let (store, io, mut parked, stalled, release) = stall_second_append(config).await;
+        let (store, _, mut parked, stalled, release) = stall_second_append(config).await;
         let region_id = region(1);
         // Two more appends queue behind the stalled one.
         let third = spawn_append_batch(&store, vec![entry(&store, region_id, "a3")]);
@@ -5320,10 +5765,10 @@ mod tests {
             HashMap::from([(region_id, id(2, 1))]),
             response.last_entry_ids
         );
-        assert_eq!(vec![0, 1], object_seqs(io.as_ref()).await);
+        assert_eq!(vec![1], collected_object_seqs(&store).await);
         assert_eq!(id(1, 1), latest(&store, region_id));
         for (append, object_seq) in [(third, 2), (fourth, 3)] {
-            let (parked_seq, release) = next_create(&mut parked).await;
+            let (parked_seq, release) = next_parked_operation(&mut parked).await;
             assert_eq!(object_seq, parked_seq);
             release.send(true).unwrap();
             let response = timeout(WAIT, append).await.unwrap().unwrap().unwrap();
@@ -5332,7 +5777,7 @@ mod tests {
                 response.last_entry_ids
             );
         }
-        assert_eq!(vec![0, 1, 2, 3], object_seqs(io.as_ref()).await);
+        assert_eq!(vec![1, 2, 3], collected_object_seqs(&store).await);
         assert_eq!(
             expected_entries(
                 region_id,
@@ -5348,7 +5793,7 @@ mod tests {
             max_unpersisted_age: Duration::from_millis(50),
             ..enqueued(manual())
         };
-        let (store, io, _parked, stalled, release) = stall_second_append(config).await;
+        let (store, _, _parked, stalled, release) = stall_second_append(config).await;
         let region_id = region(1);
 
         release.send(true).unwrap();
@@ -5357,7 +5802,7 @@ mod tests {
             HashMap::from([(region_id, id(2, 1))]),
             response.last_entry_ids
         );
-        assert_eq!(vec![0, 1], object_seqs(io.as_ref()).await);
+        assert_eq!(vec![1], collected_object_seqs(&store).await);
         assert_eq!(
             expected_entries(region_id, &[(id(1, 1), "a1")]),
             read_entries(&store, region_id, 1).await
@@ -5401,7 +5846,7 @@ mod tests {
             .await
             .unwrap()
             .unwrap();
-        assert_eq!(vec![0, 1], object_seqs(io.as_ref()).await);
+        assert_eq!(vec![1], collected_object_seqs(&store).await);
         assert_eq!(
             expected_entries(region_id, &[(id(1, 1), "a1")]),
             read_entries(&store, region_id, 1).await
@@ -5415,7 +5860,7 @@ mod tests {
             .await
             .unwrap()
             .unwrap();
-        assert_eq!(vec![0, 1, 2], object_seqs(io.as_ref()).await);
+        assert_eq!(vec![1, 2], collected_object_seqs(&store).await);
     }
 
     #[tokio::test]
@@ -5466,7 +5911,7 @@ mod tests {
             let store = store.clone();
             tokio::spawn(async move { store.seal_open_batch().await })
         };
-        let (object_seq, release) = next_create(&mut parked).await;
+        let (object_seq, release) = next_parked_operation(&mut parked).await;
         assert_eq!(1, object_seq);
 
         // Stop began, but the actor has not received the stop command when
@@ -5644,6 +6089,521 @@ mod tests {
         assert_stopped(&timeout(WAIT, wait).await.unwrap().unwrap().unwrap_err());
     }
 
+    /// The reclaim boundary in the header of the object `object_seq`.
+    async fn reclaim_boundary_of(io: &dyn WalObjectIo, object_seq: u64) -> u64 {
+        decode_header(&io.get(object_seq).await.unwrap())
+            .unwrap()
+            .reclaim_boundary
+    }
+
+    #[tokio::test]
+    async fn test_store_stamps_the_reclaim_boundary_and_collects_below_it() {
+        let store = open(memory_store(), &eager()).await;
+        let io = store.io.clone();
+        let (one, two, three) = (region(1), region(2), region(3));
+        let deleted = METRIC_OBJECT_STORE_WAL_DELETED_OBJECTS_TOTAL.get();
+
+        // Object 1 carries the boundary past the indexed start object, which
+        // is collected; region one then holds the boundary at object 1.
+        append(&store, one, "a1").await.unwrap();
+        append(&store, two, "b1").await.unwrap();
+        assert_eq!(vec![1, 2], collected_object_seqs(&store).await);
+        assert_eq!(1, reclaim_boundary_of(io.as_ref(), 1).await);
+        assert_eq!(1, reclaim_boundary_of(io.as_ref(), 2).await);
+
+        // A watermark moves the boundary; the next indexed object makes it
+        // durable.
+        store.obsolete(&provider(one), one, id(1, 1)).await.unwrap();
+        assert_eq!(vec![1, 2], collected_object_seqs(&store).await);
+        append(&store, one, "a2").await.unwrap();
+        assert_eq!(2, reclaim_boundary_of(io.as_ref(), 3).await);
+        assert_eq!(vec![2, 3], collected_object_seqs(&store).await);
+
+        // Region one dropped and region two flushed through object 2: object
+        // 4 carries the sequence after the last indexed object.
+        store.obsolete(&provider(two), two, id(2, 1)).await.unwrap();
+        store.obsolete_all(&provider(one), one).await.unwrap();
+        append(&store, two, "b2").await.unwrap();
+        assert_eq!(4, reclaim_boundary_of(io.as_ref(), 4).await);
+        assert_eq!(vec![4], collected_object_seqs(&store).await);
+
+        // A region that appears later does not move the boundary back.
+        append(&store, three, "c1").await.unwrap();
+        assert_eq!(4, reclaim_boundary_of(io.as_ref(), 5).await);
+        assert_eq!(vec![4, 5], collected_object_seqs(&store).await);
+        // Other stores of the process count their deletes as well.
+        assert!(METRIC_OBJECT_STORE_WAL_DELETED_OBJECTS_TOTAL.get() >= deleted + 4);
+        assert_eq!(
+            expected_entries(two, &[(id(4, 1), "b2")]),
+            read_entries(&store, two, 0).await
+        );
+        store.stop().await.unwrap();
+    }
+
+    #[tokio::test]
+    async fn test_store_collects_failed_batches_below_a_durable_boundary() {
+        let (io, _) = RecordingIo::over(memory_store());
+        let store = open_over(io.clone(), &eager()).await;
+        let region_id = region(1);
+        append(&store, region_id, "a1").await.unwrap();
+        append(&store, region_id, "a2").await.unwrap();
+        store
+            .obsolete(&provider(region_id), region_id, id(2, 1))
+            .await
+            .unwrap();
+
+        // Object 3 carries boundary 3 but its create is reported as failed,
+        // and the create of object 4 fails before it writes: neither boundary
+        // permits a deletion.
+        io.fail_after_next_put.store(true, Ordering::Relaxed);
+        append(&store, region_id, "a3").await.unwrap_err();
+        assert_eq!(3, reclaim_boundary_of(io.as_ref(), 3).await);
+        io.fail_next_put_persistently();
+        append(&store, region_id, "a4").await.unwrap_err();
+        assert_eq!(vec![1, 2, 3], collected_object_seqs(&store).await);
+
+        // Object 5 makes boundary 3 durable. The failed batches go once a
+        // durable boundary passes them, whether their object exists or not.
+        append(&store, region_id, "a5").await.unwrap();
+        assert_eq!(vec![3, 5], collected_object_seqs(&store).await);
+        store
+            .obsolete(&provider(region_id), region_id, id(5, 1))
+            .await
+            .unwrap();
+        append(&store, region_id, "a6").await.unwrap();
+        assert_eq!(vec![6], collected_object_seqs(&store).await);
+        assert_eq!(
+            expected_entries(region_id, &[(id(6, 1), "a6")]),
+            read_entries(&store, region_id, 0).await
+        );
+        store.stop().await.unwrap();
+    }
+
+    #[tokio::test]
+    async fn test_store_attempts_a_failed_delete_again_at_the_next_collection() {
+        let _serialized = FAILED_DELETES.lock().await;
+        let (io, mut parked) = ParkedIo::parking_creates(memory_store());
+        let store = open_over(io.clone(), &eager()).await;
+        io.deletes_parked.store(true, Ordering::SeqCst);
+        let region_id = region(1);
+        let failed = METRIC_OBJECT_STORE_WAL_FAILED_DELETES_TOTAL.get();
+        append(&store, region_id, "a1").await.unwrap();
+        append(&store, region_id, "a2").await.unwrap();
+        let (object_seq, release) = next_parked_operation(&mut parked).await;
+        assert_eq!(0, object_seq);
+        release.send(false).unwrap();
+
+        // The object stays indexed, and an indexing that leaves the durable
+        // boundary where it is starts no collection.
+        timeout(WAIT, store.wait_for_garbage_collection())
+            .await
+            .unwrap();
+        assert_eq!(
+            failed + 1,
+            METRIC_OBJECT_STORE_WAL_FAILED_DELETES_TOTAL.get()
+        );
+        assert!(store.catalog.read().unwrap().contains_object(0));
+        append(&store, region_id, "a3").await.unwrap();
+        assert!(parked.try_recv().is_err());
+        assert_eq!(vec![0, 1, 2, 3], object_seqs(io.as_ref()).await);
+
+        // The next `obsolete` starts a collection, which deletes it again.
+        store
+            .obsolete(&provider(region_id), region_id, id(1, 1))
+            .await
+            .unwrap();
+        let (object_seq, release) = next_parked_operation(&mut parked).await;
+        assert_eq!(0, object_seq);
+        release.send(true).unwrap();
+        assert_eq!(vec![1, 2, 3], collected_object_seqs(&store).await);
+        assert!(!store.catalog.read().unwrap().contains_object(0));
+        store.stop().await.unwrap();
+    }
+
+    #[tokio::test]
+    async fn test_store_collection_ends_with_stop_and_poison() {
+        for poison in [false, true] {
+            let object_store = memory_store();
+            let (io, mut parked) = ParkedIo::parking_creates(object_store.clone());
+            let store = open_over(io.clone(), &eager()).await;
+            io.deletes_parked.store(true, Ordering::SeqCst);
+            let region_id = region(1);
+            for index in 1..=5 {
+                append(&store, region_id, &format!("a{index}"))
+                    .await
+                    .unwrap();
+            }
+            store
+                .obsolete(&provider(region_id), region_id, id(5, 1))
+                .await
+                .unwrap();
+            append(&store, region_id, "a6").await.unwrap();
+            // Objects 0 to 5 are below the durable boundary; four are deleted
+            // at a time.
+            let releases = parked_operations(&mut parked, MAX_IN_FLIGHT_DELETES).await;
+            assert!(parked.try_recv().is_err());
+
+            let mut stop = None;
+            if poison {
+                put_foreign(&object_store, 7, 100).await;
+                append(&store, region_id, "a7").await.unwrap_err();
+            } else {
+                // The first poll queues the stop command; the actor answers
+                // the command of `obsolete_all` queued after it only once it
+                // handled the stop.
+                let mut stopping = Box::pin(store.stop());
+                assert!((&mut stopping).now_or_never().is_none());
+                store
+                    .obsolete_all(&provider(region_id), region_id)
+                    .await
+                    .unwrap();
+                assert!((&mut stopping).now_or_never().is_none());
+                stop = Some(stopping);
+            }
+            for release in releases.into_values() {
+                release.send(true).unwrap();
+            }
+            match stop {
+                Some(stop) => timeout(WAIT, stop).await.unwrap().unwrap(),
+                None => timeout(WAIT, store.wait_for_garbage_collection())
+                    .await
+                    .unwrap(),
+            }
+
+            // The deletes in flight completed and no other started.
+            assert!(parked.try_recv().is_err());
+            assert_eq!(
+                MAX_IN_FLIGHT_DELETES,
+                io.max_in_flight.load(Ordering::SeqCst)
+            );
+            let mut expected = vec![4, 5, 6];
+            if poison {
+                expected.push(7);
+            }
+            assert_eq!(expected, object_seqs(io.as_ref()).await);
+        }
+    }
+
+    #[tokio::test]
+    async fn test_store_drops_the_holes_of_a_collected_object() {
+        let _serialized = SKIPPED_SEGMENTS.lock().await;
+        let object_store = memory_store();
+        write_corrupted_segment(&object_store).await;
+        let store = open(object_store, &eager()).await;
+        read_entries(&store, region(2), 0).await;
+        assert_eq!(1, store.wal_holes(&provider(region(2))).unwrap().len());
+
+        store
+            .obsolete(&provider(region(1)), region(1), 1)
+            .await
+            .unwrap();
+        store
+            .obsolete(&provider(region(2)), region(2), 10)
+            .await
+            .unwrap();
+        append(&store, region(1), "a").await.unwrap();
+        assert_eq!(vec![3], collected_object_seqs(&store).await);
+        assert!(store.wal_holes(&provider(region(2))).unwrap().is_empty());
+        store.stop().await.unwrap();
+    }
+
+    #[tokio::test]
+    async fn test_store_records_no_hole_for_a_collected_object() {
+        let _serialized = SKIPPED_SEGMENTS.lock().await;
+        let object_store = memory_store();
+        let (_, footer) = write_corrupted_segment(&object_store).await;
+        let store = open(object_store, &eager()).await;
+        let skipped = METRIC_OBJECT_STORE_WAL_SKIPPED_SEGMENTS_TOTAL.get();
+
+        // A read met the damaged segment of object 0, which a collection
+        // removed before the read recorded the hole.
+        store.catalog.write().unwrap().remove_object(0);
+        let error = CorruptedWalObjectSnafu { reason: "damaged" }.build();
+        record_skipped_segment(
+            &store.catalog,
+            &store.wal_holes,
+            store.io.as_ref(),
+            0,
+            &footer[1],
+            &error,
+        );
+        assert!(store.wal_holes(&provider(region(2))).unwrap().is_empty());
+        assert_eq!(
+            skipped,
+            METRIC_OBJECT_STORE_WAL_SKIPPED_SEGMENTS_TOTAL.get()
+        );
+        store.stop().await.unwrap();
+    }
+
+    #[tokio::test]
+    async fn test_store_collects_a_failed_batch_whose_create_completes_late() {
+        let (store, io, mut parked) = open_parking_creates(memory_store(), &eager()).await;
+        let region_id = region(1);
+        // Object 1 fails while the create of object 2 is in flight.
+        let appends = spawn_appends(&store, region_id, 2).await;
+        let mut releases = parked_operations(&mut parked, 2).await;
+        releases.remove(&1).unwrap().send(false).unwrap();
+        for append in appends {
+            timeout(WAIT, append).await.unwrap().unwrap().unwrap_err();
+        }
+
+        // Objects 3 and 4 make boundary 3 durable; object 2 is skipped while
+        // its create is in flight.
+        for object_seq in [3, 4] {
+            let append = spawn_append_batch(&store, vec![entry(&store, region_id, "a")]);
+            let (parked_seq, release) = next_parked_operation(&mut parked).await;
+            assert_eq!(object_seq, parked_seq);
+            release.send(true).unwrap();
+            timeout(WAIT, append).await.unwrap().unwrap().unwrap();
+        }
+        assert_eq!(vec![3, 4], collected_object_seqs(&store).await);
+
+        // The create completes and its object is deleted without another
+        // collection.
+        io.deletes_parked.store(true, Ordering::SeqCst);
+        releases.remove(&2).unwrap().send(true).unwrap();
+        let (object_seq, release) = next_parked_operation(&mut parked).await;
+        assert_eq!(2, object_seq);
+        release.send(true).unwrap();
+        assert_eq!(vec![3, 4], collected_object_seqs(&store).await);
+        store.stop().await.unwrap();
+    }
+
+    async fn recover_listed(io: &ObjectStoreIo, listed: Vec<ListedObject>) -> Result<Recovered> {
+        finish_recovery(fetch_footers(io, listed, RECOVERY_CONCURRENCY).await?)
+    }
+
+    #[tokio::test]
+    async fn test_store_recovery_tolerates_objects_collected_after_the_listing() {
+        let io = ObjectStoreIo::new(memory_store(), PREFIX).unwrap();
+        let bounded = |object_seq, prev, reclaim_boundary| Header {
+            reclaim_boundary,
+            ..chain_header(object_seq, 1, prev)
+        };
+        put_object_with_header(&io, chain_header(0, 1, None), &[]).await;
+        put_object_with_header(&io, chain_header(1, 1, Some((0, 1))), &[]).await;
+        put_object_with_header(&io, bounded(2, Some((1, 1)), 2), &[]).await;
+        put_object_with_header(&io, bounded(3, Some((2, 1)), 2), &[]).await;
+        let listed = io.list().await.unwrap();
+
+        // A delete of an earlier instance removes object 1, below the
+        // boundary of the tip, after the listing.
+        io.delete(1).await.unwrap();
+        let recovered = recover_listed(&io, listed.clone()).await.unwrap();
+        assert_eq!(Some(3), recovered.tip.map(|tip| tip.object_seq));
+        assert_eq!(2, recovered.reclaim_boundary);
+        assert_eq!(vec![0], recovered.unindexed);
+        assert_eq!(4, recovered.next_object_seq);
+
+        // Object 2 is needed above the boundary: its disappearance fails
+        // recovery.
+        io.delete(2).await.unwrap();
+        let error = recover_listed(&io, listed).await.unwrap_err();
+        assert!(is_not_found(&error), "unexpected error: {error:?}");
+    }
+
+    #[tokio::test]
+    async fn test_store_read_skips_an_object_collected_after_it_was_listed() {
+        let _serialized = SKIPPED_SEGMENTS.lock().await;
+        let store = open(memory_store(), &eager()).await;
+        let region_id = region(1);
+        let skipped = METRIC_OBJECT_STORE_WAL_SKIPPED_SEGMENTS_TOTAL.get();
+        append(&store, region_id, "a1").await.unwrap();
+        append(&store, region_id, "a2").await.unwrap();
+
+        // The read lists objects 1 and 2 before a collection deletes object 1.
+        let stream = store.read(&provider(region_id), 0, None).await.unwrap();
+        store
+            .obsolete(&provider(region_id), region_id, id(1, 1))
+            .await
+            .unwrap();
+        append(&store, region_id, "a3").await.unwrap();
+        assert_eq!(vec![2, 3], collected_object_seqs(&store).await);
+
+        let entries = stream
+            .try_collect::<Vec<_>>()
+            .await
+            .unwrap()
+            .into_iter()
+            .flatten()
+            .collect::<Vec<_>>();
+        assert_eq!(expected_entries(region_id, &[(id(2, 1), "a2")]), entries);
+        assert!(store.wal_holes(&provider(region_id)).unwrap().is_empty());
+        assert_eq!(
+            skipped,
+            METRIC_OBJECT_STORE_WAL_SKIPPED_SEGMENTS_TOTAL.get()
+        );
+        store.stop().await.unwrap();
+    }
+
+    #[tokio::test]
+    async fn test_store_read_waits_for_an_undecided_delete() {
+        let _serialized = SKIPPED_SEGMENTS.lock().await;
+        let _failing = FAILED_DELETES.lock().await;
+        for deleted in [true, false] {
+            let object_store = memory_store();
+            let (io, mut parked) = ParkedIo::parking_creates(object_store.clone());
+            let store = open_over(io.clone(), &eager()).await;
+            io.deletes_parked.store(true, Ordering::SeqCst);
+            let region_id = region(1);
+            let skipped = METRIC_OBJECT_STORE_WAL_SKIPPED_SEGMENTS_TOTAL.get();
+            append(&store, region_id, "a1").await.unwrap();
+            next_parked_operation(&mut parked)
+                .await
+                .1
+                .send(true)
+                .unwrap();
+            append(&store, region_id, "a2").await.unwrap();
+
+            // The read lists objects 1 and 2. The delete of object 1 removes
+            // it, but its outcome is not reported yet.
+            let stream = store.read(&provider(region_id), 0, None).await.unwrap();
+            store
+                .obsolete(&provider(region_id), region_id, id(1, 1))
+                .await
+                .unwrap();
+            append(&store, region_id, "a3").await.unwrap();
+            let (object_seq, release) = next_parked_operation(&mut parked).await;
+            assert_eq!(1, object_seq);
+            object_store
+                .delete(&object_path(&object_store, 1))
+                .await
+                .unwrap();
+            let read = tokio::spawn(stream.try_collect::<Vec<_>>());
+            let waiting = || {
+                store
+                    .deleting
+                    .lock()
+                    .get(&1)
+                    .map_or(0, |settled| settled.receiver_count())
+            };
+            timeout(WAIT, async {
+                while waiting() == 0 {
+                    tokio::task::yield_now().await;
+                }
+            })
+            .await
+            .unwrap();
+            assert!(!read.is_finished());
+
+            // A delete that succeeded skips the object; one that failed
+            // leaves it indexed, and the read fails on the fetch.
+            release.send(deleted).unwrap();
+            let result = timeout(WAIT, read).await.unwrap().unwrap();
+            if deleted {
+                let entries = result.unwrap().into_iter().flatten().collect::<Vec<_>>();
+                assert_eq!(expected_entries(region_id, &[(id(2, 1), "a2")]), entries);
+            } else {
+                let error = result.unwrap_err();
+                assert!(is_not_found(&error), "unexpected error: {error:?}");
+            }
+            assert!(store.wal_holes(&provider(region_id)).unwrap().is_empty());
+            assert_eq!(
+                skipped,
+                METRIC_OBJECT_STORE_WAL_SKIPPED_SEGMENTS_TOTAL.get()
+            );
+            store.stop().await.unwrap();
+        }
+    }
+
+    #[tokio::test]
+    async fn test_store_recovers_from_the_reclaim_boundary_of_the_tip() {
+        let object_store = memory_store();
+        let io = ObjectStoreIo::new(object_store.clone(), PREFIX).unwrap();
+        let region_id = region(1);
+        let store = open(object_store.clone(), &eager()).await;
+        append(&store, region_id, "a1").await.unwrap();
+        store
+            .obsolete(&provider(region_id), region_id, id(1, 1))
+            .await
+            .unwrap();
+        append(&store, region_id, "a2").await.unwrap();
+        assert_eq!(vec![2], collected_object_seqs(&store).await);
+        store.stop().await.unwrap();
+
+        // Object 2 links to object 1, below its boundary. The start object
+        // carries the recovered boundary.
+        let store = open(object_store.clone(), &eager()).await;
+        assert_eq!(2, reclaim_boundary_of(&io, 3).await);
+        assert_eq!(
+            expected_entries(region_id, &[(id(2, 1), "a2")]),
+            read_entries(&store, region_id, 0).await
+        );
+        store
+            .obsolete(&provider(region_id), region_id, id(2, 1))
+            .await
+            .unwrap();
+        append(&store, region_id, "a3").await.unwrap();
+        assert_eq!(vec![4], collected_object_seqs(&store).await);
+        store.stop().await.unwrap();
+
+        let store = open(object_store, &eager()).await;
+        assert_eq!(4, reclaim_boundary_of(&io, 5).await);
+        assert_eq!(
+            expected_entries(region_id, &[(id(4, 1), "a3")]),
+            read_entries(&store, region_id, 0).await
+        );
+        assert_eq!(vec![4, 5], object_seqs(&io).await);
+        store.stop().await.unwrap();
+    }
+
+    #[tokio::test]
+    async fn test_store_collects_a_late_object_of_an_earlier_epoch_below_the_boundary() {
+        let object_store = memory_store();
+        let io = ObjectStoreIo::new(object_store.clone(), PREFIX).unwrap();
+        let region_id = region(1);
+        let record = |entry_id, data: &'static str| Record {
+            region_id,
+            entry_id,
+            payload: Bytes::from_static(data.as_bytes()),
+        };
+        let bounded = |object_seq, epoch, prev, reclaim_boundary| Header {
+            reclaim_boundary,
+            ..chain_header(object_seq, epoch, prev)
+        };
+        put_object_with_header(&io, chain_header(0, 1, None), &[]).await;
+        put_object_with_header(
+            &io,
+            chain_header(1, 1, Some((0, 1))),
+            &[record(id(1, 1), "a")],
+        )
+        .await;
+        // Epoch 4 started at object 3, and object 4 carries boundary 4.
+        // Object 2 of epoch 1 landed late; its link holds, but its epoch is
+        // lower.
+        put_object_with_header(
+            &io,
+            bounded(2, 1, Some((1, 1)), 2),
+            &[record(id(2, 1), "late")],
+        )
+        .await;
+        put_object_with_header(&io, chain_header(3, 4, Some((1, 1))), &[]).await;
+        put_object_with_header(
+            &io,
+            bounded(4, 4, Some((3, 4)), 4),
+            &[record(id(4, 1), "b")],
+        )
+        .await;
+        let recovered = recover(&io).await.unwrap();
+        assert_eq!(Some(4), recovered.tip.map(|tip| tip.object_seq));
+        assert_eq!(4, recovered.reclaim_boundary);
+        assert_eq!(vec![0, 1, 2, 3], recovered.unindexed);
+
+        // The start object makes boundary 4 durable, and the first
+        // collection deletes everything below it.
+        let store = open(object_store, &eager()).await;
+        assert_eq!(
+            expected_entries(region_id, &[(id(4, 1), "b")]),
+            read_entries(&store, region_id, 0).await
+        );
+        store
+            .obsolete(&provider(region_id), region_id, id(4, 1))
+            .await
+            .unwrap();
+        assert_eq!(vec![4, 5], collected_object_seqs(&store).await);
+        store.stop().await.unwrap();
+    }
+
     /// Object access whose next create stores the object but reports a
     /// transient failure, as a create whose response is lost.
     struct LostResponseIo {
@@ -5676,6 +6636,10 @@ mod tests {
             self.inner.get_range(object_seq, offset, len).await
         }
 
+        async fn delete(&self, object_seq: u64) -> Result<()> {
+            self.inner.delete(object_seq).await
+        }
+
         async fn list(&self) -> Result<Vec<ListedObject>> {
             self.inner.list().await
         }
@@ -5710,6 +6674,10 @@ mod tests {
             self.inner.get_range(object_seq, offset, len).await
         }
 
+        async fn delete(&self, object_seq: u64) -> Result<()> {
+            self.inner.delete(object_seq).await
+        }
+
         async fn list(&self) -> Result<Vec<ListedObject>> {
             self.inner.list().await
         }
@@ -5719,16 +6687,18 @@ mod tests {
         }
     }
 
-    /// Object access whose whole-object reads, or conditional creates, park
-    /// until the test releases them, counting how many are in flight. A
-    /// release of false fails the operation with a transient error before it
-    /// reaches the object store.
+    /// Object access whose whole-object reads, or conditional creates, and
+    /// on request deletes, park until the test releases them, counting how
+    /// many are in flight. A release of false fails the operation with a
+    /// transient error before it reaches the object store.
     struct ParkedIo {
         inner: ObjectStoreIo,
         parked: mpsc::UnboundedSender<(u64, oneshot::Sender<bool>)>,
         park_creates: bool,
         /// Whether creates park yet; set once the store opened.
         creates_parked: AtomicBool,
+        /// Whether deletes park.
+        deletes_parked: AtomicBool,
         in_flight: AtomicUsize,
         max_in_flight: AtomicUsize,
     }
@@ -5766,6 +6736,7 @@ mod tests {
                     parked,
                     park_creates,
                     creates_parked: AtomicBool::new(false),
+                    deletes_parked: AtomicBool::new(false),
                     in_flight: AtomicUsize::new(0),
                     max_in_flight: AtomicUsize::new(0),
                 }),
@@ -5809,6 +6780,13 @@ mod tests {
             self.inner.get_range(object_seq, offset, len).await
         }
 
+        async fn delete(&self, object_seq: u64) -> Result<()> {
+            if self.deletes_parked.load(Ordering::SeqCst) && !self.park(object_seq).await {
+                return injected_failure("delete", self.object_path(object_seq));
+            }
+            self.inner.delete(object_seq).await
+        }
+
         async fn list(&self) -> Result<Vec<ListedObject>> {
             self.inner.list().await
         }
@@ -5896,6 +6874,10 @@ mod tests {
             Ok(bytes)
         }
 
+        async fn delete(&self, object_seq: u64) -> Result<()> {
+            self.inner.delete(object_seq).await
+        }
+
         async fn list(&self) -> Result<Vec<ListedObject>> {
             self.inner.list().await
         }
diff --git a/src/mito2/src/engine/object_store_wal_recovery_test.rs b/src/mito2/src/engine/object_store_wal_recovery_test.rs
index 917ad614a07..126e38e4b26 100644
--- a/src/mito2/src/engine/object_store_wal_recovery_test.rs
+++ b/src/mito2/src/engine/object_store_wal_recovery_test.rs
@@ -373,6 +373,98 @@ async fn test_reopen_after_partial_flush_replays_only_unflushed_regions(#[case]
     assert_eq!(5, engine.get_region_statistic(REGION_B).unwrap().num_rows);
 }
 
+/// Waits until no WAL object delete is in flight.
+async fn wait_for_collection(store: &ObjectStoreLogStore) {
+    tokio::time::timeout(WAIT, store.wait_for_garbage_collection())
+        .await
+        .expect("collection must complete");
+}
+
+#[rstest]
+#[case(AckMode::Durable)]
+#[case(AckMode::Enqueued)]
+#[tokio::test]
+async fn test_flush_collects_objects_below_the_watermark(#[case] ack_mode: AckMode) {
+    let mut env = TestEnv::with_prefix("object-store-wal-collection").await;
+    let object_store = memory_store();
+    let store = open_store(&object_store, ack_mode).await;
+    let engine = new_engine(&mut env, store.clone()).await;
+    let (table_dir_a, schema_a) = create_region(&engine, REGION_A, &[]).await;
+    let (table_dir_b, schema_b) = create_region(&engine, REGION_B, &[]).await;
+
+    // Object 1 holds entries of both regions, object 2 of region A and
+    // object 3 of region B. Object 1 carries the boundary past the start
+    // object, which is collected.
+    let mut writer = SealedWriter::new(&store);
+    writer
+        .put_and_seal(
+            &engine,
+            vec![
+                (REGION_A, rows(&schema_a, 0, 2)),
+                (REGION_B, rows(&schema_b, 0, 3)),
+            ],
+        )
+        .await;
+    writer
+        .put_and_seal(&engine, vec![(REGION_A, rows(&schema_a, 2, 4))])
+        .await;
+    writer
+        .put_and_seal(&engine, vec![(REGION_B, rows(&schema_b, 3, 5))])
+        .await;
+    wait_for_collection(&store).await;
+    assert_eq!(vec![1, 2, 3], wal_object_seqs(&object_store).await);
+
+    // Region B has no watermark and still needs object 1, so flushing
+    // region A collects nothing.
+    flush_region(&engine, REGION_A, None).await;
+    writer
+        .put_and_seal(&engine, vec![(REGION_A, rows(&schema_a, 4, 6))])
+        .await;
+    wait_for_collection(&store).await;
+    assert_eq!(vec![1, 2, 3, 4], wal_object_seqs(&object_store).await);
+
+    // Once region B is flushed, the first object a region needs is object 4,
+    // the unflushed entry of region A; object 5 carries that boundary.
+    flush_region(&engine, REGION_B, None).await;
+    writer
+        .put_and_seal(&engine, vec![(REGION_B, rows(&schema_b, 5, 7))])
+        .await;
+    wait_for_collection(&store).await;
+    assert_eq!(vec![4, 5], wal_object_seqs(&object_store).await);
+    let rows_a = scan_rows(&engine, REGION_A).await;
+    let rows_b = scan_rows(&engine, REGION_B).await;
+    engine.stop().await.unwrap();
+    drop(engine);
+    drop(writer);
+    drop(store);
+
+    // Recovery stops at the boundary and each region replays its entry above
+    // its flushed id.
+    let store = open_store(&object_store, ack_mode).await;
+    let engine = new_engine(&mut env, store.clone()).await;
+    open_region(&engine, REGION_A, &table_dir_a, &[])
+        .await
+        .unwrap();
+    open_region(&engine, REGION_B, &table_dir_b, &[])
+        .await
+        .unwrap();
+    for (region_id, flushed_entry_id, last_entry_id) in [
+        (REGION_A, entry_id(2, 1), entry_id(4, 1)),
+        (REGION_B, entry_id(3, 1), entry_id(5, 1)),
+    ] {
+        let state = region_recovery_state(&engine, region_id).await;
+        assert_eq!(flushed_entry_id, state.flushed_entry_id);
+        assert_eq!(last_entry_id, state.last_entry_id);
+        assert_eq!(2, state.memtable_rows);
+    }
+    assert_eq!(rows_a, scan_rows(&engine, REGION_A).await);
+    assert_eq!(rows_b, scan_rows(&engine, REGION_B).await);
+    assert_eq!(6, engine.get_region_statistic(REGION_A).unwrap().num_rows);
+    assert_eq!(7, engine.get_region_statistic(REGION_B).unwrap().num_rows);
+    wait_for_collection(&store).await;
+    assert_eq!(vec![4, 5, 6], wal_object_seqs(&object_store).await);
+}
+
 #[rstest]
 #[case(AckMode::Durable)]
 #[case(AckMode::Enqueued)]
@@ -600,12 +692,14 @@ async fn test_reopen_on_empty_prefix_without_durable_entries() {
     assert_eq!(rows_b, scan_rows(&engine, REGION_B).await);
 
     // The region is usable and its first entry lands in the first object
-    // after the start object.
+    // after the start object, which carries the boundary past the start
+    // object.
     let mut writer = SealedWriter::new(&store);
     writer
         .put_and_seal(&engine, vec![(REGION_A, rows(&schema_a, 0, 2))])
         .await;
-    assert_eq!(vec![0, 1], wal_object_seqs(&object_store).await);
+    wait_for_collection(&store).await;
+    assert_eq!(vec![1], wal_object_seqs(&object_store).await);
     assert_eq!(entry_id(1, 1), latest(&store, REGION_A));
     assert_eq!(
         entry_id(1, 1),
@@ -766,7 +860,10 @@ async fn test_reopen_after_a_failed_create_shows_only_the_acknowledged_write() {
     writer
         .put_and_seal(&engine, vec![(REGION_A, key_rows(2))])
         .await;
-    assert_eq!(vec![0, 1, 2], wal_object_seqs(&object_store).await);
+    // Object 2 carries the boundary past the start object, which is
+    // collected.
+    wait_for_collection(&store).await;
+    assert_eq!(vec![1, 2], wal_object_seqs(&object_store).await);
     assert_eq!(entry_id(2, 1), latest(&store, REGION_A));
     let region_a = region(&engine, REGION_A);
     assert_eq!(1, region_a.version_control.committed_sequence());
@@ -979,7 +1076,10 @@ async fn test_enqueued_crash_before_the_object_exists_replays_durable_entries()
         entry_id(2, 1),
         region_recovery_state(&engine, REGION_A).await.last_entry_id
     );
-    assert_eq!(vec![0, 1, 2], wal_object_seqs(&object_store).await);
+    // Opening the region recorded its watermark, so object 2 carries the
+    // boundary past both start objects.
+    wait_for_collection(&store).await;
+    assert_eq!(vec![2], wal_object_seqs(&object_store).await);
     let rows_before = scan_rows(&engine, REGION_A).await;
     assert_eq!(2, engine.get_region_statistic(REGION_A).unwrap().num_rows);
     drop(writer);
diff --git a/tests-integration/tests/object_store_wal.rs b/tests-integration/tests/object_store_wal.rs
index 1ce37992b5f..0aef1e145b0 100644
--- a/tests-integration/tests/object_store_wal.rs
+++ b/tests-integration/tests/object_store_wal.rs
@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use std::path::Path;
+use std::time::Duration;
 
 use common_procedure::options::ProcedureConfig;
 use common_query::Output;
@@ -67,6 +68,18 @@ async fn count_wal_objects(standalone: &GreptimeDbStandalone) -> usize {
         .count()
 }
 
+/// Waits until collection leaves `expected` WAL objects under [`WAL_ROOT`].
+async fn wait_for_wal_objects(standalone: &GreptimeDbStandalone, expected: usize) {
+    let collected = async {
+        while count_wal_objects(standalone).await != expected {
+            tokio::time::sleep(Duration::from_millis(20)).await;
+        }
+    };
+    tokio::time::timeout(Duration::from_secs(30), collected)
+        .await
+        .unwrap_or_else(|_| panic!("expected {expected} WAL objects after collection"));
+}
+
 /// Stops the procedure manager, the region server and the WAL of the
 /// instance.
 async fn shutdown(standalone: &mut GreptimeDbStandalone) {
@@ -103,7 +116,8 @@ async fn restart(
 }
 
 /// Writes batches that land in separate WAL objects, restarts before and
-/// after a flush, and checks that the rows survive both restarts.
+/// after a flush, and checks that the rows survive both restarts and that the
+/// flush lets collection delete every earlier object.
 async fn run_restarts_on_object_store_wal(store_type: StorageType) {
     common_telemetry::init_default_ut_logging();
     let wal_config = ObjectStoreWalConfig {
@@ -129,29 +143,14 @@ async fn run_restarts_on_object_store_wal(store_type: StorageType) {
     // Every insert returns once its entries are durable, so the batches land
     // in separate WAL objects after the object that started the epoch.
     for batch in 0..BATCHES {
-        let values = (0..ROWS_PER_BATCH)
-            .map(|row| {
-                let i = batch * ROWS_PER_BATCH + row;
-                format!(
-                    "('host_{i}', {i}.0, {})",
-                    1_686_567_600_000 + i as i64 * 1000
-                )
-            })
-            .collect::<Vec<_>>()
-            .join(", ");
-        execute_sql(
-            standalone.fe_instance(),
-            &format!("INSERT INTO cpu VALUES {values}"),
-        )
-        .await;
+        insert_batch(&standalone, batch).await;
     }
     let expected = query_rows(&standalone).await;
     assert_eq!(BATCHES * ROWS_PER_BATCH + 4, expected.lines().count());
-    let objects = count_wal_objects(&standalone).await;
-    assert!(
-        objects > BATCHES,
-        "expected more than {BATCHES} WAL objects"
-    );
+    // The region has no watermark, so its first object holds the boundary
+    // and only the object that started the epoch is collected.
+    wait_for_wal_objects(&standalone, BATCHES).await;
+    let objects = BATCHES;
     // No Raft Engine log store is created as a fallback.
     assert!(
         !Path::new(&standalone.opts.storage.data_home)
@@ -165,13 +164,37 @@ async fn run_restarts_on_object_store_wal(store_type: StorageType) {
     assert_eq!(expected, query_rows(&standalone).await);
     assert!(count_wal_objects(&standalone).await > objects);
 
-    // The flushed rows come back from the SST files.
+    // After the flush the next object carries a reclaim boundary past every
+    // earlier object, so collection leaves that object alone. The flushed
+    // rows come back from the SST files and the last batch from the WAL.
     execute_sql(standalone.fe_instance(), "ADMIN FLUSH_TABLE('cpu')").await;
+    insert_batch(&standalone, BATCHES).await;
+    wait_for_wal_objects(&standalone, 1).await;
+    let expected = query_rows(&standalone).await;
     let mut standalone = restart(&builder, standalone).await;
     assert_eq!(expected, query_rows(&standalone).await);
     shutdown(&mut standalone).await;
 }
 
+/// Inserts the rows of `batch`, which land in one WAL object.
+async fn insert_batch(standalone: &GreptimeDbStandalone, batch: usize) {
+    let values = (0..ROWS_PER_BATCH)
+        .map(|row| {
+            let i = batch * ROWS_PER_BATCH + row;
+            format!(
+                "('host_{i}', {i}.0, {})",
+                1_686_567_600_000 + i as i64 * 1000
+            )
+        })
+        .collect::<Vec<_>>()
+        .join(", ");
+    execute_sql(
+        standalone.fe_instance(),
+        &format!("INSERT INTO cpu VALUES {values}"),
+    )
+    .await;
+}
+
 #[tokio::test(flavor = "multi_thread")]
 async fn test_standalone_object_store_wal_survives_restarts_on_file() {
     run_restarts_on_object_store_wal(StorageType::File).await;