feat(log-store): collect object store WAL objects below a persisted reclaim boundary (#9471)

* feat(log-store): collect object store WAL objects below a persisted reclaim boundary

Every object header carries a reclaim boundary: every object below it was
fully flushed when the object was sealed. The writer computes the boundary
from the catalog and the obsolete watermarks after every obsolete and every
indexed object, and stamps it on every sealed object and on the start
object. The boundary of the last indexed object permits deletion: every
present object below it, indexed, listed by recovery off the chain, or
left by a failed batch, is deleted in the background, at most four at a
time.

Recovery judges every candidate with its own boundary: a link to a
predecessor below it holds whether or not the predecessor is present, and
the chain stops there. A read that listed an object before it was
collected skips it instead of failing.

The header grows from 46 to 54 bytes; the format version stays 1.

Signed-off-by: jeremyhi <fengjiachun@gmail.com>

* perf(log-store): scan the reclaim boundary from the previous boundary

The boundary was recomputed by scanning every region's indexed objects
from the first one, so the flushed objects of busy regions were scanned
again on every indexed object while another region held the boundary.
Scan the objects in sequence order from the previous boundary instead
and stop at the first one holding a live segment. Watermarks never move
down and a region first appears above the boundary, so the result is
the same and an object is not scanned again once the boundary passed it.

Signed-off-by: jeremyhi <fengjiachun@gmail.com>

* test(log-store): name the parked operation helpers after both their uses

Signed-off-by: jeremyhi <fengjiachun@gmail.com>

---------

Signed-off-by: jeremyhi <fengjiachun@gmail.com>

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();
 }

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

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,

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()
         );

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]

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);

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;