From e4872ec1a2d0d040d9f05c370f238cd7f411328e Mon Sep 17 00:00:00 2001 From: jeremyhi Date: Thu, 8 Oct 2026 13:16:14 +0000 Subject: [PATCH] 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 * 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 * test(log-store): name the parked operation helpers after both their uses Signed-off-by: jeremyhi --------- Signed-off-by: jeremyhi --- src/log-store/src/metrics.rs | 14 + src/log-store/src/object_store_wal.rs | 24 +- src/log-store/src/object_store_wal/catalog.rs | 172 ++- src/log-store/src/object_store_wal/format.rs | 70 +- src/log-store/src/object_store_wal/io.rs | 15 + src/log-store/src/object_store_wal/store.rs | 1276 +++++++++++++++-- .../engine/object_store_wal_recovery_test.rs | 108 +- tests-integration/tests/object_store_wal.rs | 67 +- 8 files changed, 1536 insertions(+), 210 deletions(-) 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>, regions: BTreeMap>, + /// Largest entry id ever indexed per region, kept after the object that + /// held it was removed. + max_entry_ids: HashMap, } 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> { + 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, + ) -> 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 { - self.regions - .get(®ion_id)? - .last_key_value() - .map(|(_, entry)| entry.max_entry_id) + self.max_entry_ids.get(®ion_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, + /// 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
{ 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
{ 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
{ 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> { @@ -357,6 +367,11 @@ mod tests { vec![1, 1, 2], objects.iter().map(|object| object.size).collect::>() ); + + // 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, terminal_error: TerminalError, stopped: Arc, command_tx: mpsc::Sender, @@ -135,6 +141,72 @@ pub struct WalHole { type TerminalError = Arc>>>; +/// 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>>, +} + +impl DeletingObjects { + fn lock(&self) -> std::sync::MutexGuard<'_, BTreeMap>> { + 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> { + 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::>(); + 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>, }, + /// Answered once the collection the recorded watermarks allow started. + Collect { response: oneshot::Sender<()> }, Stop { response: oneshot::Sender>, }, @@ -850,6 +970,7 @@ enum CreateState { /// created, indexed and acknowledged. struct SealedBatch { object_seq: u64, + reclaim_boundary: u64, bytes: Bytes, footer: Vec, first_admitted_at: Instant, @@ -910,10 +1031,22 @@ type CreateOutcome = (u64, Result); /// [`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, catalog: Arc>, obsolete_entry_ids: ObsoleteEntryIds, + wal_holes: WalHoles, terminal_error: TerminalError, stopped: Arc, command_rx: mpsc::Receiver, @@ -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, + /// Sequences whose create is in flight. + creating: HashSet, + deleting: Arc, + deletes: FuturesUnordered)>>, + /// Deletes that failed since the last collection started. + failed_deletes: HashSet, flush_interval: Duration, #[cfg(any(test, feature = "testing"))] admitted_appends: watch::Sender, @@ -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) { + 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::>() + }; + 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> { terminal_error .lock() @@ -1715,6 +1992,25 @@ fn shared(error: &Arc) -> 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, + 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, 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, /// 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, +} + +/// 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, + 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 { 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) -> Result { +/// 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 { + let FetchedObjects { objects, vanished } = fetched; let headers = objects .iter() .map(|fetched| (fetched.object.object_seq, &fetched.header)) .collect::>(); 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) -> Result { .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::>(); 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) -> Result { 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) -> Result { /// 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) -> Vec { - // 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::>(); + 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, tip: u64) -> Option> { + 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, -) -> Result { + reclaim_boundary: u64, +) -> Result<(ChainLink, Vec)> { + 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, concurrency: usize, -) -> Result> { - let mut fetched = futures::stream::iter(objects) +) -> Result { + 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::>() .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; + async fn delete(&self, object_seq: u64) -> Result<()>; + async fn list(&self) -> Result>; 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> { 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 { + 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)>, ) -> (u64, oneshot::Sender) { 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)>, count: usize, ) -> HashMap> { 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)> { @@ -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::>() @@ -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::>(); - 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::>(); + 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::>(), releases.keys().copied().collect::>() @@ -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[®ion_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) -> Result { + 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::>() + .await + .unwrap() + .into_iter() + .flatten() + .collect::>(); + 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::>()); + 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::>(); + 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> { 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> { 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)>, 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> { 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> { 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::>() - .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::>() + .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;