From 29d5854e6a648281b04c36a4bdcc94b3ab47789d Mon Sep 17 00:00:00 2001 From: jeremyhi Date: Mon, 12 Oct 2026 07:17:44 +0000 Subject: [PATCH] test(log-store): drive recovery, read, delete and stall failures through the fault matrix (#9488) * test(log-store): drive recovery, read, delete and stall failures through the fault matrix Add the second half of the object store WAL fault matrix: the failures of the object store access and of the store state that the conditional-create rows did not drive. Each case asserts the failed waiters and their error, whether the store is poisoned, the sequence, both indexed gauges at the failure, the exact movement of every counter and what a read sees after. Recovery: a listing, a trailer fetch and a footer fetch that fail, the catalog rejecting an object after every footer was read, and an object gone by the time it is fetched, below the boundary of the tip (ignored) and at it (construction fails); each leaves the gauges and the recovery metrics untouched. Reads: a segment fetch that fails, without a hole or a read timer sample, and a later read of the region that succeeds. Collection: a delete that fails and is attempted again at the next collection, and a stop that waits for a delete in flight, with the catalog agreeing with the listing in both outcomes. Store state: per-region entry id exhaustion refusing the append without touching the batch it sealed, and the durable stop paths with several creates in flight that succeed, fail transiently or conflict after stop began. Stalled appends released by admission, by a stop, by a terminal error and by the teardown of the actor, the first two timed as a pair. The recovery and read timers observed around parked fetches, twice each, with holds that cannot overlap. The cases use the existing doubles, extended with a failing listing and a failing range read at an offset on RecordingIo and parked range reads on ParkedIo. Case::row is now optional, and the assertion that every row of the failure matrix has a case is unchanged. Signed-off-by: jeremyhi * test(log-store): lose a wide object at its header and footer reads in the fault matrix The vanished-object case now runs three ways: a short object read whole, and a wide object deleted before its header and tail window reads or before its separate footer read, each below and at the boundary of the tip. The entry id exhaustion case reads the other region of the sealed batch back, the teardown case asserts the indexed gauges before and after the actor is gone, and two comments that restated their names are gone. Signed-off-by: jeremyhi * test(log-store): prove the stop-path ordering and share the wide-footer fixture in the fault matrix The case of creates that succeed after stop began keeps object 1 parked while objects 4 to 2 complete, round-trips a command through the actor and asserts that no append is acknowledged and nothing is indexed before object 1 is released. The stop-with-delete case replaces its yields with the same command barrier, queued behind the stop command. The wide-footer fixture moves to the tests module and is shared with the recovery test that fetches a footer longer than the tail window. The gauge helper is named for its side effect and the vanished-object stage enum for what it names. Signed-off-by: jeremyhi * test(log-store): keep the four-create scenario doc on its fault matrix case Signed-off-by: jeremyhi --------- Signed-off-by: jeremyhi --- src/log-store/src/object_store_wal/store.rs | 1155 ++++++++++++++++++- 1 file changed, 1118 insertions(+), 37 deletions(-) diff --git a/src/log-store/src/object_store_wal/store.rs b/src/log-store/src/object_store_wal/store.rs index d3a3acc6968..98b0c8cad6b 100644 --- a/src/log-store/src/object_store_wal/store.rs +++ b/src/log-store/src/object_store_wal/store.rs @@ -3292,15 +3292,7 @@ mod tests { async fn test_store_recovery_fetches_a_footer_longer_than_the_tail_window() { let _serialized = SERIALIZED.lock().await; let object_store = memory_store(); - let regions = (RECOVERY_TAIL_WINDOW / FOOTER_ENTRY_LEN + 100) as u32; - let records = (1..=regions) - .map(|number| Record { - region_id: region(number), - entry_id: 1, - payload: Bytes::from_static(b"wide"), - }) - .collect::>(); - put_records(&object_store, 0, &records).await; + put_wide_object(&object_store).await; put_object(&object_store, 1, region(1), &[id(1, 1)]).await; let (io, reads) = RecordingIo::over(object_store.clone()); @@ -3312,7 +3304,7 @@ mod tests { let recovered = recover(io.as_ref()).await.unwrap(); let expected = recover_by_decoding(io.as_ref()).await.unwrap(); assert_same_recovery(&expected, &recovered); - assert_eq!(regions as usize, recovered.durable_entry_ids.len()); + assert_eq!(WIDE_REGIONS as usize, recovered.durable_entry_ids.len()); // The wide object took the header, the tail window and the footer; // the narrow one was read whole. @@ -3345,7 +3337,7 @@ mod tests { let store = open(object_store, &eager()).await; assert_eq!(id(1, 1), latest(&store, region(1))); - assert_eq!(1, latest(&store, region(regions))); + assert_eq!(1, latest(&store, region(WIDE_REGIONS))); store.stop().await.unwrap(); } @@ -3366,6 +3358,33 @@ mod tests { put_records(object_store, object_seq, &records).await; } + /// Regions enough for a footer longer than the window recovery reads + /// from the end of an object, so that recovery fetches the header, that + /// window and the footer of such an object separately. + const WIDE_REGIONS: u32 = (RECOVERY_TAIL_WINDOW / FOOTER_ENTRY_LEN + 100) as u32; + + /// Writes object 0 with a footer longer than the tail window and returns + /// the offsets of the window and of the footer. + async fn put_wide_object(object_store: &ObjectStore) -> (u64, u64) { + let records = (1..=WIDE_REGIONS) + .map(|number| Record { + region_id: region(number), + entry_id: 1, + payload: Bytes::from_static(b"wide"), + }) + .collect::>(); + put_records(object_store, 0, &records).await; + let bytes = object_store + .read(&object_path(object_store, 0)) + .await + .unwrap() + .to_vec(); + let (trailer, _) = footer_of(&bytes); + let tail_start = (bytes.len() - RECOVERY_TAIL_WINDOW) as u64; + assert!(trailer.footer_offset < tail_start); + (tail_start, trailer.footer_offset) + } + #[tokio::test] async fn test_store_resumes_sequence_and_durable_ids() { let _serialized = SERIALIZED.lock().await; @@ -6870,9 +6889,9 @@ mod tests { } /// 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. + /// on request deletes and range reads, 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)>, @@ -6881,6 +6900,7 @@ mod tests { creates_parked: AtomicBool, /// Whether deletes park. deletes_parked: AtomicBool, + range_reads_parked: AtomicBool, in_flight: AtomicUsize, max_in_flight: AtomicUsize, } @@ -6919,6 +6939,7 @@ mod tests { park_creates, creates_parked: AtomicBool::new(false), deletes_parked: AtomicBool::new(false), + range_reads_parked: AtomicBool::new(false), in_flight: AtomicUsize::new(0), max_in_flight: AtomicUsize::new(0), }), @@ -6959,6 +6980,9 @@ mod tests { } async fn get_range(&self, object_seq: u64, offset: u64, len: u64) -> Result { + if self.range_reads_parked.load(Ordering::SeqCst) && !self.park(object_seq).await { + return injected_failure("read", self.object_path(object_seq)); + } self.inner.get_range(object_seq, offset, len).await } @@ -6984,7 +7008,8 @@ mod tests { /// length) and, on request, reports the next conditional create as failed /// after it wrote the object, fails it with a given error before it /// writes, stores the same content before it writes, or damages the bytes - /// the next range read returns. + /// the next range read returns; it also fails the next listing, or the + /// next range read of an object at an offset, with a transient error. struct RecordingIo { inner: ObjectStoreIo, reads: RangeReads, @@ -6992,6 +7017,8 @@ mod tests { fail_next_put: Mutex>, damage_next_range_read: AtomicBool, duplicate_next_put: AtomicBool, + fail_next_list: AtomicBool, + fail_range_read_at: Mutex>, } impl RecordingIo { @@ -7004,6 +7031,8 @@ mod tests { fail_next_put: Mutex::new(None), damage_next_range_read: AtomicBool::new(false), duplicate_next_put: AtomicBool::new(false), + fail_next_list: AtomicBool::new(false), + fail_range_read_at: Mutex::new(None), }; (Arc::new(io), reads) } @@ -7053,6 +7082,15 @@ mod tests { async fn get_range(&self, object_seq: u64, offset: u64, len: u64) -> Result { self.reads.lock().unwrap().push((object_seq, offset, len)); + if self + .fail_range_read_at + .lock() + .unwrap() + .take_if(|at| *at == (object_seq, offset)) + .is_some() + { + return injected_failure("read", self.object_path(object_seq)); + } let bytes = self.inner.get_range(object_seq, offset, len).await?; if self.damage_next_range_read.swap(false, Ordering::Relaxed) { let mut damaged = bytes.to_vec(); @@ -7069,6 +7107,9 @@ mod tests { } async fn list(&self) -> Result> { + if self.fail_next_list.swap(false, Ordering::Relaxed) { + return injected_failure("list", PREFIX.to_string()); + } self.inner.list().await } @@ -7077,8 +7118,9 @@ mod tests { } } - /// Drives every outcome of a conditional create through the store, one - /// case per fault, and checks that every row of the *Failure matrix* has one. + /// Drives every outcome of a conditional create and every failure of the + /// object store access through the store, one case per fault, and checks + /// that every row of the *Failure matrix* has one. mod fault_matrix { use object_store::layers::mock::{self, MockLayerBuilder, oio}; use prometheus::Histogram; @@ -7126,10 +7168,11 @@ mod tests { Row::StartFails, ]; - /// One failure of the object store access and the row it exercises. + /// One failure of the object store access and the row it exercises, + /// when the failure is an outcome of a conditional create. struct Case { fault: &'static str, - row: Row, + row: Option, run: fn() -> BoxFuture<'static, ()>, } @@ -7362,89 +7405,169 @@ mod tests { let cases = [ Case { fault: "a create that succeeds", - row: Row::Created, + row: Some(Row::Created), run: || Box::pin(create_succeeds()), }, Case { fault: "a create that finds an object with identical content", - row: Row::Created, + row: Some(Row::Created), run: || Box::pin(create_finds_an_identical_object()), }, Case { fault: "a create whose response is lost", - row: Row::Created, + row: Some(Row::Created), run: || Box::pin(create_response_is_lost()), }, Case { fault: "a create that fails transiently with later objects created and in flight, `durable` mode", - row: Row::TransientDurable, + row: Some(Row::TransientDurable), run: || Box::pin(create_fails_transiently_durable()), }, Case { fault: "a create that fails transiently, `enqueued` mode", - row: Row::TransientEnqueued, + row: Some(Row::TransientEnqueued), run: || Box::pin(create_fails_transiently_enqueued()), }, Case { fault: "a create that fails transiently after `stop` began, `enqueued` mode", - row: Row::TransientEnqueuedAfterStop, + row: Some(Row::TransientEnqueuedAfterStop), run: || Box::pin(create_fails_transiently_after_stop_began()), }, Case { fault: "a create that finds an object of an earlier epoch, `durable` mode", - row: Row::EarlierEpochDurable, + row: Some(Row::EarlierEpochDurable), run: || Box::pin(create_finds_an_earlier_epoch()), }, Case { fault: "a create that finds an object of an earlier epoch, `enqueued` mode", - row: Row::Conflict, + row: Some(Row::Conflict), run: || Box::pin(create_finds_an_earlier_epoch_enqueued()), }, Case { fault: "a create that finds an object of the same or a later epoch", - row: Row::Conflict, + row: Some(Row::Conflict), run: || Box::pin(create_finds_another_writer()), }, Case { fault: "a create that finds an object whose header cannot be read", - row: Row::Conflict, + row: Some(Row::Conflict), run: || Box::pin(create_finds_an_unreadable_object()), }, Case { fault: "a create that completes after a conflict poisoned the store", - row: Row::Conflict, + row: Some(Row::Conflict), run: || Box::pin(create_completes_after_the_store_was_poisoned()), }, Case { fault: "a created object the catalog rejects", - row: Row::CatalogError, + row: Some(Row::CatalogError), run: || Box::pin(catalog_rejects_the_created_object()), }, Case { fault: "a create at the last representable object sequence", - row: Row::LastSequence, + row: Some(Row::LastSequence), run: || Box::pin(create_takes_the_last_object_sequence()), }, Case { fault: "a start object that meets an object of an earlier epoch", - row: Row::StartMovesPast, + row: Some(Row::StartMovesPast), run: || Box::pin(start_object_meets_an_earlier_epoch()), }, Case { fault: "a start object that meets an object of the same or a later epoch", - row: Row::StartFails, + row: Some(Row::StartFails), run: || Box::pin(start_object_meets_another_writer()), }, Case { fault: "a start object that meets an object with identical content", - row: Row::StartFails, + row: Some(Row::StartFails), run: || Box::pin(start_object_meets_an_identical_object()), }, Case { fault: "a start object whose create has an unknown outcome", - row: Row::StartFails, + row: Some(Row::StartFails), run: || Box::pin(start_object_create_has_an_unknown_outcome()), }, + Case { + fault: "a listing that fails at recovery", + row: None, + run: || Box::pin(listing_fails_at_recovery()), + }, + Case { + fault: "a trailer fetch that fails at recovery", + row: None, + run: || Box::pin(trailer_fetch_fails_at_recovery()), + }, + Case { + fault: "a footer fetch that fails at recovery", + row: None, + run: || Box::pin(footer_fetch_fails_at_recovery()), + }, + Case { + fault: "a recovery the catalog rejects after every footer was read", + row: None, + run: || Box::pin(catalog_rejects_the_recovery()), + }, + Case { + fault: "an object gone by the time recovery fetches it", + row: None, + run: || Box::pin(object_vanishes_before_its_fetch()), + }, + Case { + fault: "a segment fetch that fails at a read", + row: None, + run: || Box::pin(segment_fetch_fails_at_a_read()), + }, + Case { + fault: "a delete that fails at a collection", + row: None, + run: || Box::pin(delete_fails_at_a_collection()), + }, + Case { + fault: "a stop with a delete in flight", + row: None, + run: || Box::pin(stop_waits_for_a_delete_in_flight()), + }, + Case { + fault: "an append past the entry id range of its region", + row: None, + run: || Box::pin(region_exhausts_its_entry_ids()), + }, + Case { + fault: "creates that succeed after `stop` began, `durable` mode", + row: None, + run: || Box::pin(creates_succeed_after_stop_began()), + }, + Case { + fault: "a create that fails transiently after `stop` began, `durable` mode", + row: None, + run: || Box::pin(create_fails_transiently_after_stop_began_durable()), + }, + Case { + fault: "a create that conflicts after `stop` began", + row: None, + run: || Box::pin(create_conflicts_after_stop_began()), + }, + Case { + fault: "a stalled append released by admission, then one released by a stop", + row: None, + run: || Box::pin(stalled_appends_released_by_admission_and_by_a_stop()), + }, + Case { + fault: "a stalled append released by a terminal error", + row: None, + run: || Box::pin(stalled_append_released_by_a_terminal_error()), + }, + Case { + fault: "a stalled append released by the teardown of the actor", + row: None, + run: || Box::pin(stalled_append_released_by_the_teardown()), + }, + Case { + fault: "an object store that answers slowly", + row: None, + run: || Box::pin(object_store_answers_slowly()), + }, ]; common_telemetry::init_default_ut_logging(); @@ -7454,7 +7577,7 @@ mod tests { } for row in ROWS { assert!( - cases.iter().any(|case| case.row == row), + cases.iter().any(|case| case.row == Some(row)), "no case covers the failure matrix row {row:?}" ); } @@ -8226,5 +8349,963 @@ mod tests { ) .await; } + + async fn assert_appends_stopped( + appends: Vec>>, + ) { + for append in appends { + assert_stopped(&timeout(WAIT, append).await.unwrap().unwrap().unwrap_err()); + } + } + + /// Opens and stops a store on an empty prefix, which sets both + /// process-wide indexed gauges to its one start object and moves the + /// recovery and create counters, so a case that abandons a recovery + /// can tell the gauges standing still from a publication of what its + /// prefix holds. It runs before the case samples `Counters`. + async fn initialize_indexed_gauges() -> (i64, i64) { + let store = open(memory_store(), &eager()).await; + store.stop().await.unwrap(); + let gauges = indexed_gauges(); + assert_eq!((1, object_len(&[]) as i64), gauges); + gauges + } + + /// Opens a store over `io` and asserts that the recovery was + /// abandoned: `open` fails with an error `recognise` accepts and + /// `retry_hint`, no start object is written so the prefix still holds + /// `present`, both gauges stand still and no metric moves. + async fn assert_recovery_abandoned( + io: Arc, + present: &[u64], + recognise: impl Fn(&Error) -> bool, + retry_hint: RetryHint, + ) { + let gauges = initialize_indexed_gauges().await; + let before = Counters::sample(); + let error = ObjectStoreLogStore::open(io.clone(), &eager(), PREFIX.to_string()) + .await + .unwrap_err(); + assert!(recognise(&error), "unexpected error: {error:?}"); + assert_eq!(retry_hint, error.retry_hint()); + assert_eq!(present, object_seqs(io.as_ref()).await.as_slice()); + assert_eq!(gauges, indexed_gauges()); + assert_eq!(Counters::default(), Counters::since(before)); + } + + /// Writes object `object_seq` holding one entry of region 1 with + /// `data`, chained to the object below it. + async fn put_entry(object_store: &ObjectStore, object_seq: u64, data: &str) { + let record = Record { + region_id: region(1), + entry_id: id(object_seq, 1), + payload: Bytes::from(data.as_bytes().to_vec()), + }; + put_records(object_store, object_seq, &[record]).await; + } + + async fn listing_fails_at_recovery() { + let object_store = memory_store(); + put_entry(&object_store, 0, "a0").await; + put_entry(&object_store, 1, "a1").await; + let (io, _) = RecordingIo::over(object_store); + io.fail_next_list.store(true, Ordering::Relaxed); + assert_recovery_abandoned( + io.clone(), + &[0, 1], + |error| { + matches!( + error, + Error::WalObjectStore { + operation: "list", + .. + } + ) + }, + RetryHint::Retryable, + ) + .await; + assert!(!io.fail_next_list.load(Ordering::Relaxed)); + + let before = Counters::sample(); + let store = open_over(io.clone(), &eager()).await; + assert_indexed(io.as_ref(), &[0, 1, 2]).await; + assert_eq!( + expected_entries(region(1), &[(id(0, 1), "a0"), (id(1, 1), "a1")]), + read_entries(&store, region(1), 1).await + ); + assert_eq!( + Counters { + created_objects: 1, + recovered_objects: 2, + recoveries: 1, + reads: 1, + ..Counters::default() + }, + Counters::since(before) + ); + store.stop().await.unwrap(); + } + + async fn trailer_fetch_fails_at_recovery() { + fetch_fails_at_recovery(false).await; + } + + async fn footer_fetch_fails_at_recovery() { + fetch_fails_at_recovery(true).await; + } + + /// The range read of the tail window, or of the footer, of the one + /// object fails; the next open recovers it. + async fn fetch_fails_at_recovery(footer: bool) { + let object_store = memory_store(); + let (tail_start, footer_offset) = put_wide_object(&object_store).await; + let (io, _) = RecordingIo::over(object_store); + let offset = if footer { footer_offset } else { tail_start }; + *io.fail_range_read_at.lock().unwrap() = Some((0, offset)); + let path = io.object_path(0); + assert_recovery_abandoned( + io.clone(), + &[0], + |error| { + matches!(error, Error::WalObjectStore { operation: "read", path: actual, .. } if *actual == path) + }, + RetryHint::Retryable, + ) + .await; + assert!(io.fail_range_read_at.lock().unwrap().is_none()); + + let before = Counters::sample(); + let store = open_over(io.clone(), &eager()).await; + assert_indexed(io.as_ref(), &[0, 1]).await; + assert_eq!( + expected_entries(region(1), &[(1, "wide")]), + read_entries(&store, region(1), 1).await + ); + assert_eq!( + Counters { + created_objects: 1, + recovered_objects: 1, + recoveries: 1, + reads: 1, + ..Counters::default() + }, + Counters::since(before) + ); + store.stop().await.unwrap(); + } + + /// The entry ranges of the region do not increase with the sequence, + /// which the catalog refuses once every footer is read. + async fn catalog_rejects_the_recovery() { + let object_store = memory_store(); + put_object(&object_store, 0, region(1), &[id(3, 1), id(3, 2)]).await; + put_object(&object_store, 1, region(1), &[id(2, 1)]).await; + let io = Arc::new(ObjectStoreIo::new(object_store, PREFIX).unwrap()); + let path = io.object_path(1); + assert_recovery_abandoned( + io.clone(), + &[0, 1], + |error| { + matches!(error, Error::InvalidWalObject { path: actual, source, .. } + if *actual == path && matches!(**source, Error::CorruptedWalObject { .. })) + }, + RetryHint::NonRetryable, + ) + .await; + } + + /// Writes the chain 0 to 3 of epoch 1, one entry of region 1 each, + /// and `WIDE_REGIONS` more regions per object when `wide`; objects 2 + /// and 3 carry reclaim boundary 2, so object 1 is below the boundary + /// of the tip and object 2 is not. + async fn put_bounded_chain(object_store: &ObjectStore, wide: bool) { + let io = ObjectStoreIo::new(object_store.clone(), PREFIX).unwrap(); + for object_seq in 0..=3 { + let header = Header { + reclaim_boundary: if object_seq >= 2 { 2 } else { 0 }, + ..chain_header(object_seq, 1, (object_seq > 0).then(|| (object_seq - 1, 1))) + }; + let mut records = vec![Record { + region_id: region(1), + entry_id: id(object_seq, 1), + payload: Bytes::from(format!("a{object_seq}")), + }]; + if wide { + records.extend((2..=WIDE_REGIONS).map(|number| Record { + region_id: region(number), + entry_id: id(object_seq, 1), + payload: Bytes::from_static(b"wide"), + })); + } + put_object_with_header(&io, header, &records).await; + } + } + + /// The fetch of recovery at which an object is found gone. + #[derive(Debug, Clone, Copy, PartialEq, Eq)] + enum RecoveryFetchStage { + /// The read of a short object as a whole. + WholeObject, + /// The reads of the header and of the tail window of a wide object. + Header, + /// The separate read of the footer of a wide object. + Footer, + } + + /// Waits for `count` parked operations and returns their releases. + async fn parked_releases( + parked: &mut mpsc::UnboundedReceiver<(u64, oneshot::Sender)>, + count: usize, + ) -> Vec> { + let mut releases = Vec::with_capacity(count); + for _ in 0..count { + releases.push(next_parked_operation(parked).await.1); + } + releases + } + + /// Every fetch of the listed objects parks and one object is deleted + /// before one of its reads: below the boundary of the tip the object + /// is ignored, at or above it the construction fails. + async fn object_vanishes_before_its_fetch() { + for (vanished, fails) in [(1, false), (2, true)] { + for stage in [ + RecoveryFetchStage::WholeObject, + RecoveryFetchStage::Header, + RecoveryFetchStage::Footer, + ] { + object_vanishes_at(vanished, fails, stage).await; + } + } + } + + async fn object_vanishes_at(vanished: u64, fails: bool, stage: RecoveryFetchStage) { + let object_store = memory_store(); + put_bounded_chain(&object_store, stage != RecoveryFetchStage::WholeObject).await; + let (io, mut parked) = ParkedIo::over(object_store.clone()); + io.range_reads_parked.store(true, Ordering::SeqCst); + let gauges = initialize_indexed_gauges().await; + let before = Counters::sample(); + let opening = { + let io = io.clone(); + tokio::spawn(async move { + ObjectStoreLogStore::open(io, &eager(), PREFIX.to_string()).await + }) + }; + let path = object_path(&object_store, vanished); + let delete = || object_store.delete(&path); + let release_all = |releases: Vec>| { + for release in releases { + release.send(true).unwrap(); + } + }; + // A short object parks once, a wide one twice, at its header and + // at its tail window, and then once more at its footer. + let whole = stage == RecoveryFetchStage::WholeObject; + let first = parked_releases(&mut parked, if whole { 4 } else { 8 }).await; + if stage != RecoveryFetchStage::Footer { + delete().await.unwrap(); + } + release_all(first); + match stage { + RecoveryFetchStage::WholeObject => {} + RecoveryFetchStage::Header => release_all(parked_releases(&mut parked, 3).await), + RecoveryFetchStage::Footer => { + let footers = parked_releases(&mut parked, 4).await; + delete().await.unwrap(); + release_all(footers); + } + } + let opened = timeout(WAIT, opening).await.unwrap().unwrap(); + io.range_reads_parked.store(false, Ordering::SeqCst); + if fails { + let error = opened.unwrap_err(); + assert!( + is_not_found(&error) + && matches!(&error, Error::WalObjectStore { path, .. } if *path == io.object_path(2)), + "unexpected error: {error:?}" + ); + assert_eq!(vec![0, 1, 3], object_seqs(io.as_ref()).await); + assert_eq!(gauges, indexed_gauges()); + assert_eq!(Counters::default(), Counters::since(before)); + return; + } + let store = opened.unwrap(); + assert_eq!(vec![0, 2, 3, 4], object_seqs(io.as_ref()).await); + assert_indexed(io.as_ref(), &[2, 3, 4]).await; + assert_eq!( + expected_entries(region(1), &[(id(2, 1), "a2"), (id(3, 1), "a3")]), + read_entries(&store, region(1), 1).await + ); + assert_eq!( + Counters { + created_objects: 1, + recovered_objects: 3, + recoveries: 1, + reads: 1, + ..Counters::default() + }, + Counters::since(before) + ); + store.stop().await.unwrap(); + } + + async fn segment_fetch_fails_at_a_read() { + let (io, _) = RecordingIo::over(memory_store()); + let store = open_over(io.clone(), &eager()).await; + let before = Counters::sample(); + let region_id = region(1); + append(&store, region_id, "a1").await.unwrap(); + append(&store, region_id, "a2").await.unwrap(); + assert_eq!(vec![1, 2], collected_object_seqs(&store).await); + let (path, segment) = store + .segment_location(&provider(region_id), 1) + .unwrap() + .unwrap(); + *io.fail_range_read_at.lock().unwrap() = Some((1, segment.start)); + let error = store + .read(&provider(region_id), 1, None) + .await + .unwrap() + .try_collect::>() + .await + .unwrap_err(); + assert!( + matches!(&error, Error::WalObjectStore { operation: "read", path: actual, .. } if *actual == path), + "unexpected error: {error:?}" + ); + assert_eq!(RetryHint::Retryable, error.retry_hint()); + assert!(io.fail_range_read_at.lock().unwrap().is_none()); + assert!(store.wal_holes(&provider(region_id)).unwrap().is_empty()); + assert!(terminal(&store.terminal_error).is_none()); + assert_indexed(io.as_ref(), &[1, 2]).await; + let written = Counters { + created_objects: 2, + seal_to_durable: 2, + sealed_objects: 2, + object_bytes: 2 * one_entry_object(), + object_entries: 2, + acknowledged_appends: 2, + deleted_objects: 1, + ..Counters::default() + }; + assert_eq!(written, Counters::since(before)); + + assert_eq!( + expected_entries(region_id, &[(id(1, 1), "a1"), (id(2, 1), "a2")]), + read_entries(&store, region_id, 1).await + ); + assert_eq!( + Counters { + reads: 1, + ..written + }, + Counters::since(before) + ); + store.stop().await.unwrap(); + } + + /// The delete of the start object fails at the collection that + /// indexing object 1 starts; a watermark that leaves the boundary + /// where it is starts the next collection, which deletes it. + async fn delete_fails_at_a_collection() { + 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 before = Counters::sample(); + let region_id = region(1); + append(&store, region_id, "a1").await.unwrap(); + let (object_seq, release) = next_parked_operation(&mut parked).await; + assert_eq!(0, object_seq); + release.send(false).unwrap(); + timeout(WAIT, store.wait_for_garbage_collection()) + .await + .unwrap(); + assert!(store.catalog.read().unwrap().contains_object(0)); + assert!(terminal(&store.terminal_error).is_none()); + assert_eq!(vec![0, 1], object_seqs(io.as_ref()).await); + assert_indexed(io.as_ref(), &[0, 1]).await; + assert_eq!( + expected_entries(region_id, &[(id(1, 1), "a1")]), + read_entries(&store, region_id, 1).await + ); + let written = Counters { + created_objects: 1, + seal_to_durable: 1, + sealed_objects: 1, + object_bytes: one_entry_object(), + object_entries: 1, + acknowledged_appends: 1, + reads: 1, + failed_deletes: 1, + ..Counters::default() + }; + assert_eq!(written, Counters::since(before)); + + store + .obsolete(&provider(region_id), region_id, 0) + .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], collected_object_seqs(&store).await); + assert!(!store.catalog.read().unwrap().contains_object(0)); + assert_indexed(io.as_ref(), &[1]).await; + assert_eq!( + expected_entries(region_id, &[(id(1, 1), "a1")]), + read_entries(&store, region_id, 1).await + ); + assert_eq!( + Counters { + reads: 2, + deleted_objects: 1, + ..written + }, + Counters::since(before) + ); + store.stop().await.unwrap(); + } + + /// `stop` returns only after the delete in flight completed, and the + /// catalog then holds exactly what the prefix holds, whether the + /// delete succeeded or failed. + async fn stop_waits_for_a_delete_in_flight() { + for deleted in [true, false] { + 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 before = Counters::sample(); + let region_id = region(1); + append(&store, region_id, "a1").await.unwrap(); + let (object_seq, release) = next_parked_operation(&mut parked).await; + assert_eq!(0, object_seq); + // The first poll queues the stop command; the command queued + // behind it is answered once the stop was handled, and the + // delete in flight still holds the answer to the stop back. + let mut stop = Box::pin(store.stop()); + assert!((&mut stop).now_or_never().is_none()); + round_trip_stopping_actor(&store).await; + assert!((&mut stop).now_or_never().is_none()); + release.send(deleted).unwrap(); + timeout(WAIT, stop).await.unwrap().unwrap(); + + let present = if deleted { vec![1] } else { vec![0, 1] }; + assert_eq!(present, object_seqs(io.as_ref()).await); + let indexed = store + .catalog + .read() + .unwrap() + .objects_in_order() + .map(|(object_seq, _)| object_seq) + .collect::>(); + assert_eq!(present, indexed); + assert_indexed(io.as_ref(), &present).await; + assert!(terminal(&store.terminal_error).is_none()); + assert_eq!( + expected_entries(region_id, &[(id(1, 1), "a1")]), + read_entries(&store, region_id, 1).await + ); + assert_eq!( + Counters { + created_objects: 1, + seal_to_durable: 1, + sealed_objects: 1, + object_bytes: one_entry_object(), + object_entries: 1, + acknowledged_appends: 1, + reads: 1, + deleted_objects: deleted as u64, + failed_deletes: (!deleted) as u64, + ..Counters::default() + }, + Counters::since(before) + ); + } + } + + /// An append that alone runs past the position range of its region + /// is refused; the open batch it sealed first, which holds another + /// region too, is acknowledged, and the store keeps serving. + async fn region_exhausts_its_entry_ids() { + let store = open(memory_store(), &manual()).await; + let before = Counters::sample(); + let (region_id, other) = (region(1), region(2)); + let shared = spawn_append_batch( + &store, + vec![entry(&store, other, "b1"), entry(&store, region_id, "a1")], + ); + store.wait_for_admitted_appends(1).await.unwrap(); + let exhausting = (0..POSITION_LIMIT) + .map(|_| entry(&store, region_id, "")) + .collect::>(); + let error = store.append_batch(exhausting).await.unwrap_err(); + assert!( + matches!(error, Error::WalEntryPositionExhausted { region_id: actual, .. } if actual == region_id), + "unexpected error: {error:?}" + ); + let response = timeout(WAIT, shared).await.unwrap().unwrap().unwrap(); + assert_eq!( + HashMap::from([(other, id(1, 1)), (region_id, id(1, 1))]), + response.last_entry_ids + ); + assert!(terminal(&store.terminal_error).is_none()); + + let next = spawn_append_batch(&store, vec![entry(&store, region_id, "a2")]); + store.wait_for_admitted_appends(2).await.unwrap(); + store.seal_open_batch().await.unwrap(); + let response = timeout(WAIT, next).await.unwrap().unwrap().unwrap(); + assert_eq!(Some(&id(2, 1)), response.last_entry_ids.get(®ion_id)); + assert_eq!(vec![1, 2], collected_object_seqs(&store).await); + assert_indexed(store.io.as_ref(), &[1, 2]).await; + assert_eq!( + expected_entries(region_id, &[(id(1, 1), "a1"), (id(2, 1), "a2")]), + read_entries(&store, region_id, 1).await + ); + assert_eq!( + expected_entries(other, &[(id(1, 1), "b1")]), + read_entries(&store, other, 1).await + ); + assert_eq!( + Counters { + created_objects: 2, + seal_to_durable: 2, + sealed_objects: 2, + object_bytes: object_bytes_of(&[(other, "b1"), (region_id, "a1")]) + + one_entry_object(), + object_entries: 3, + acknowledged_appends: 2, + reads: 2, + deleted_objects: 1, + ..Counters::default() + }, + Counters::since(before) + ); + store.stop().await.unwrap(); + } + + /// Round-trips a command through the actor once `stop` began, when + /// the seal command is refused: the collection that `obsolete_all` of + /// a region without entries requests is answered by the actor, and + /// hides nothing of the regions a case reads. + async fn round_trip_stopping_actor(store: &ObjectStoreLogStore) { + let region_id = region(9); + store + .obsolete_all(&provider(region_id), region_id) + .await + .unwrap(); + } + + /// Four creates are in flight and a fifth batch waits for a slot when + /// `stop` begins: the fifth is dropped; objects 4 to 2 complete while + /// object 1 stays parked and nothing is acknowledged or indexed; then + /// object 1 releases the four acknowledgements in order. + async fn creates_succeed_after_stop_began() { + let (store, io, mut parked) = open_parking_creates(memory_store(), &eager()).await; + let before = Counters::sample(); + let region_id = region(1); + let mut appends = spawn_appends(&store, region_id, MAX_IN_FLIGHT_CREATES + 1).await; + let mut releases = parked_operations(&mut parked, MAX_IN_FLIGHT_CREATES).await; + let dropped = appends.pop().unwrap(); + let stop = { + let store = store.clone(); + tokio::spawn(async move { store.stop().await }) + }; + assert_stopped(&timeout(WAIT, dropped).await.unwrap().unwrap().unwrap_err()); + for object_seq in (2..=MAX_IN_FLIGHT_CREATES as u64).rev() { + releases.remove(&object_seq).unwrap().send(true).unwrap(); + } + wait_until(|| { + Counters::since(before).created_objects == MAX_IN_FLIGHT_CREATES as u64 - 1 + }) + .await; + round_trip_stopping_actor(&store).await; + assert!(appends.iter().all(|append| !append.is_finished())); + assert_eq!(0, latest(&store, region_id)); + assert_indexed(io.as_ref(), &[0]).await; + + releases.remove(&1).unwrap().send(true).unwrap(); + for (index, append) in appends.into_iter().enumerate() { + let response = timeout(WAIT, append).await.unwrap().unwrap().unwrap(); + assert_eq!( + Some(&id(index as u64 + 1, 1)), + response.last_entry_ids.get(®ion_id) + ); + } + timeout(WAIT, stop).await.unwrap().unwrap().unwrap(); + + // Nothing is collected once stop began. + assert_eq!(vec![0, 1, 2, 3, 4], object_seqs(io.as_ref()).await); + assert_indexed(io.as_ref(), &[0, 1, 2, 3, 4]).await; + assert!(terminal(&store.terminal_error).is_none()); + assert_eq!( + expected_entries( + region_id, + &[ + (id(1, 1), "a1"), + (id(2, 1), "a2"), + (id(3, 1), "a3"), + (id(4, 1), "a4") + ] + ), + read_entries(&store, region_id, 1).await + ); + assert_eq!( + Counters { + created_objects: 4, + seal_to_durable: 4, + sealed_objects: 5, + object_bytes: 5 * one_entry_object(), + object_entries: 5, + acknowledged_appends: 4, + reads: 1, + ..Counters::default() + }, + Counters::since(before) + ); + } + + /// Object 2 is created and object 3 in flight when object 1 fails + /// after `stop` began: every batch receives the stopped error, `stop` + /// waits for object 3, whose late create is an orphan. + async fn create_fails_transiently_after_stop_began_durable() { + let (store, io, mut parked) = open_parking_creates(memory_store(), &eager()).await; + let before = Counters::sample(); + let region_id = region(1); + let appends = spawn_appends(&store, region_id, 3).await; + let mut releases = parked_operations(&mut parked, 3).await; + let stop = { + let store = store.clone(); + tokio::spawn(async move { store.stop().await }) + }; + wait_until(|| store.stopped.load(Ordering::Acquire)).await; + releases.remove(&2).unwrap().send(true).unwrap(); + wait_until(|| Counters::since(before).created_objects == 1).await; + releases.remove(&1).unwrap().send(false).unwrap(); + assert_appends_stopped(appends).await; + assert!(!stop.is_finished()); + assert_indexed(io.as_ref(), &[0]).await; + assert_eq!( + Counters { + created_objects: 1, + seal_to_durable: 1, + create_failures: 1, + sealed_objects: 3, + object_bytes: 3 * one_entry_object(), + object_entries: 3, + ..Counters::default() + }, + Counters::since(before) + ); + + releases.remove(&3).unwrap().send(true).unwrap(); + timeout(WAIT, stop).await.unwrap().unwrap().unwrap(); + assert_eq!(vec![0, 2, 3], object_seqs(io.as_ref()).await); + assert!(terminal(&store.terminal_error).is_none()); + assert!(read_entries(&store, region_id, 1).await.is_empty()); + assert_indexed(io.as_ref(), &[0]).await; + assert_eq!( + Counters { + created_objects: 2, + seal_to_durable: 2, + create_failures: 1, + sealed_objects: 3, + object_bytes: 3 * one_entry_object(), + object_entries: 3, + reads: 1, + ..Counters::default() + }, + Counters::since(before) + ); + } + + /// Object 2 is created and object 3 in flight when object 1 meets + /// another writer after `stop` began: the store is poisoned, every + /// batch receives the stopped error and `stop` still returns. + async fn create_conflicts_after_stop_began() { + let object_store = memory_store(); + let (store, io, mut parked) = + open_parking_creates(object_store.clone(), &eager()).await; + put_foreign(&object_store, 1, 1).await; + let before = Counters::sample(); + let region_id = region(1); + let appends = spawn_appends(&store, region_id, 3).await; + let mut releases = parked_operations(&mut parked, 3).await; + let stop = { + let store = store.clone(); + tokio::spawn(async move { store.stop().await }) + }; + wait_until(|| store.stopped.load(Ordering::Acquire)).await; + releases.remove(&2).unwrap().send(true).unwrap(); + wait_until(|| Counters::since(before).created_objects == 1).await; + releases.remove(&1).unwrap().send(true).unwrap(); + assert_appends_stopped(appends).await; + assert!(matches!( + terminal(&store.terminal_error).as_deref(), + Some(Error::WalObjectConflict { .. }) + )); + assert_read_is_terminal(&store, region_id).await; + assert!(!stop.is_finished()); + assert_indexed(io.as_ref(), &[0]).await; + + releases.remove(&3).unwrap().send(true).unwrap(); + timeout(WAIT, stop).await.unwrap().unwrap().unwrap(); + assert_eq!(vec![0, 1, 2, 3], object_seqs(io.as_ref()).await); + assert_indexed(io.as_ref(), &[0]).await; + assert_eq!( + Counters { + created_objects: 2, + seal_to_durable: 2, + create_conflicts: 1, + poisoned: 1, + sealed_objects: 3, + object_bytes: 3 * one_entry_object(), + object_entries: 3, + ..Counters::default() + }, + Counters::since(before) + ); + } + + /// The `enqueued` configuration under which the second append stalls + /// behind the create of the first. + fn stalling() -> ObjectStoreWalConfig { + ObjectStoreWalConfig { + max_unpersisted_bytes: ReadableSize(1), + ..enqueued(manual()) + } + } + + /// What `stall_second_append` leaves in the metrics. + fn stalled_once() -> Counters { + Counters { + created_objects: 1, + sealed_objects: 1, + object_bytes: one_entry_object(), + object_entries: 1, + recoveries: 1, + stalled_appends: 1, + ..Counters::default() + } + } + + /// The first stalled append is admitted once the upload completes; + /// the entry it admits puts the backlog at the threshold again, so + /// the next append stalls behind the create it seals, and the stop + /// releases it with the stopped error. Both waits are timed. + async fn stalled_appends_released_by_admission_and_by_a_stop() { + const SHORT: Duration = Duration::from_millis(20); + let region_id = region(1); + let before = Counters::sample(); + let stalled_at = Instant::now(); + let (store, io, mut parked, stalled, release) = stall_second_append(stalling()).await; + assert_eq!(stalled_once(), Counters::since(before)); + + let timer = timer_before(&METRIC_OBJECT_STORE_WAL_STALLED_APPEND_SECONDS); + let holding_since = Instant::now(); + tokio::time::sleep(SHORT).await; + let held = holding_since.elapsed(); + release.send(true).unwrap(); + let response = timeout(WAIT, stalled).await.unwrap().unwrap().unwrap(); + assert_eq!(Some(&id(2, 1)), response.last_entry_ids.get(®ion_id)); + let admitted = Timed::since( + &METRIC_OBJECT_STORE_WAL_STALLED_APPEND_SECONDS, + timer, + held, + stalled_at.elapsed(), + ); + assert_eq!(vec![1], collected_object_seqs(&store).await); + + let stalled_at = Instant::now(); + let refused = spawn_append_batch(&store, vec![entry(&store, region_id, "a3")]); + let (object_seq, release) = next_parked_operation(&mut parked).await; + assert_eq!(2, object_seq); + let timer = timer_before(&METRIC_OBJECT_STORE_WAL_STALLED_APPEND_SECONDS); + let holding_since = Instant::now(); + tokio::time::sleep(admitted.elapsed + SHORT).await; + let held = holding_since.elapsed(); + let stop = { + let store = store.clone(); + tokio::spawn(async move { store.stop().await }) + }; + assert_stopped(&timeout(WAIT, refused).await.unwrap().unwrap().unwrap_err()); + let refused = Timed::since( + &METRIC_OBJECT_STORE_WAL_STALLED_APPEND_SECONDS, + timer, + held, + stalled_at.elapsed(), + ); + Timed::assert_pair("stalled append", admitted, refused); + + release.send(true).unwrap(); + timeout(WAIT, stop).await.unwrap().unwrap().unwrap(); + assert_eq!(vec![1, 2], object_seqs(io.as_ref()).await); + assert_indexed(io.as_ref(), &[1, 2]).await; + assert!(terminal(&store.terminal_error).is_none()); + assert_eq!( + expected_entries(region_id, &[(id(1, 1), "a1"), (id(2, 1), "a2")]), + read_entries(&store, region_id, 1).await + ); + assert_eq!( + Counters { + created_objects: 3, + seal_to_durable: 2, + sealed_objects: 2, + object_bytes: 2 * one_entry_object(), + object_entries: 2, + recoveries: 1, + reads: 1, + deleted_objects: 1, + stalled_appends: 2, + stalled_waits: 2, + ..Counters::default() + }, + Counters::since(before) + ); + } + + /// The create the stalled append waits for meets another writer, + /// which poisons the `enqueued` store: the append is released with + /// the terminal error and its wait is recorded. + async fn stalled_append_released_by_a_terminal_error() { + let region_id = region(1); + let before = Counters::sample(); + let (store, io, _parked, stalled, release) = stall_second_append(stalling()).await; + assert_eq!(stalled_once(), Counters::since(before)); + let foreign = Header { + object_seq: 1, + epoch: 1, + prev: None, + reclaim_boundary: 0, + }; + put_object_with_header(&io.inner, foreign, &[]).await; + release.send(true).unwrap(); + let error = timeout(WAIT, stalled).await.unwrap().unwrap().unwrap_err(); + assert!( + matches!(unwrap_shared(&error), Error::WalObjectConflict { .. }), + "unexpected error: {error:?}" + ); + assert_read_is_terminal(&store, region_id).await; + let error = store.stop().await.unwrap_err(); + assert!( + matches!(unwrap_shared(&error), Error::WalObjectConflict { .. }), + "unexpected error: {error:?}" + ); + assert_eq!(vec![0, 1], object_seqs(io.as_ref()).await); + assert_indexed(io.as_ref(), &[0]).await; + assert_eq!( + Counters { + create_conflicts: 1, + poisoned: 1, + stalled_waits: 1, + ..stalled_once() + }, + Counters::since(before) + ); + } + + /// The store is dropped without `stop` while the append is stalled: + /// the actor's teardown releases it and records its wait, and the + /// parked create never runs. + async fn stalled_append_released_by_the_teardown() { + let before = Counters::sample(); + let (store, io, _parked, stalled, _release) = stall_second_append(stalling()).await; + assert_eq!(stalled_once(), Counters::since(before)); + assert_indexed(io.as_ref(), &[0]).await; + + stalled.abort(); + drop(store); + wait_until(|| LIVE_ACTORS.load(Ordering::SeqCst) == 0).await; + assert_eq!(vec![0], object_seqs(io.as_ref()).await); + assert_indexed(io.as_ref(), &[0]).await; + assert_eq!( + Counters { + stalled_waits: 1, + ..stalled_once() + }, + Counters::since(before) + ); + } + + /// Every fetch of recovery and of a read parks until the case releases + /// it, which is how the recovery and read timers are read for the + /// durations they recorded. Each is observed around two holds, the + /// second longer than anything the first observations took. + async fn object_store_answers_slowly() { + const SHORT: Duration = Duration::from_millis(20); + let object_store = memory_store(); + put_entry(&object_store, 0, "a0").await; + let (io, mut parked) = ParkedIo::over(object_store); + let before = Counters::sample(); + let region_id = region(1); + let mut recoveries = Vec::new(); + let mut reads = Vec::new(); + let mut hold = SHORT; + for objects in 1..=2 { + let timer = timer_before(&METRIC_OBJECT_STORE_WAL_RECOVERY_SECONDS); + let started_at = Instant::now(); + let opening = { + let io = io.clone(); + tokio::spawn(async move { + ObjectStoreLogStore::open(io, &eager(), PREFIX.to_string()).await + }) + }; + // Every object is short enough to be read whole, and every + // such read parks. + let releases = parked_operations(&mut parked, objects).await; + let parked_at = Instant::now(); + tokio::time::sleep(hold).await; + let held = parked_at.elapsed(); + for release in releases.into_values() { + release.send(true).unwrap(); + } + let store = timeout(WAIT, opening).await.unwrap().unwrap().unwrap(); + recoveries.push(Timed::since( + &METRIC_OBJECT_STORE_WAL_RECOVERY_SECONDS, + timer, + held, + started_at.elapsed(), + )); + + io.range_reads_parked.store(true, Ordering::SeqCst); + let timer = timer_before(&METRIC_OBJECT_STORE_WAL_READ_SECONDS); + let started_at = Instant::now(); + let reading = { + let store = store.clone(); + tokio::spawn(async move { read_entries(&store, region_id, 1).await }) + }; + let (object_seq, release) = next_parked_operation(&mut parked).await; + assert_eq!(0, object_seq); + let parked_at = Instant::now(); + tokio::time::sleep(hold).await; + let held = parked_at.elapsed(); + release.send(true).unwrap(); + let entries = timeout(WAIT, reading).await.unwrap().unwrap(); + reads.push(Timed::since( + &METRIC_OBJECT_STORE_WAL_READ_SECONDS, + timer, + held, + started_at.elapsed(), + )); + io.range_reads_parked.store(false, Ordering::SeqCst); + assert_eq!(expected_entries(region_id, &[(id(0, 1), "a0")]), entries); + store.stop().await.unwrap(); + hold = recoveries[0].elapsed.max(reads[0].elapsed) + SHORT; + } + Timed::assert_pair("recovery", recoveries[0], recoveries[1]); + Timed::assert_pair("read", reads[0], reads[1]); + assert_indexed(io.as_ref(), &[0, 1, 2]).await; + assert_eq!( + Counters { + created_objects: 2, + recovered_objects: 3, + recoveries: 2, + reads: 2, + ..Counters::default() + }, + Counters::since(before) + ); + } } }