From 9c2933de0b45f02011ecf4c144f7fb12169cd5cf Mon Sep 17 00:00:00 2001 From: jeremyhi Date: Fri, 18 Sep 2026 07:28:40 +0000 Subject: [PATCH] feat(log-store): add the object store WAL batch, catalog and I/O (#9216) * feat(log-store): add the object store WAL batch, catalog and I/O Add the three modules between the object format and the store of the object store WAL: - batch: the open batch that accumulates admitted entries into the next object and assigns entry ids at admission. Ids are object-sequence-major, `(object_seq << 20) | position`, with positions starting at one per region per object, so a batch that is rolled back and admitted again under the same sequence hands out the same ids. The time the first entry was admitted is kept for the store's age-based sealing; an empty admission does not start it. - catalog: the in-memory index over object footers, by sequence and per region. Insertion is atomic and rejects an empty footer, duplicate region segments, invalid entry ranges, an already indexed sequence and entry ranges that are not strictly increasing across objects. The next object sequence continues after the largest indexed one and is raised above the largest entry id of any region, and is rejected once it no longer fits an entry id. - io: object store access under `/objects/`: a conditional create whose retry with identical content is a no-op and whose conflicting content is rejected, whole and range reads, and a listing of well-formed keys in sequence order. A prefix with a `.` or `..` component is rejected. When the store reports that the object exists but the read that compares its content fails, the read failure is returned with its retry hint instead of the create failure. The modules have no callers until the store lands, so they are declared with `#[allow(dead_code)]`. Only `entry_id` is exported. Signed-off-by: jeremyhi * test(log-store): drop comments that restate the batch test assertions Signed-off-by: jeremyhi * refactor(log-store): use pub(crate) in the object store WAL batch, catalog and I/O Signed-off-by: jeremyhi * fix(log-store): keep entry_id crate-private and narrow the create collision check Re-export `entry_id` as `pub(crate)`: nothing outside `log-store` uses it yet, and exporting it would freeze the raw `(object_seq, position)` encoding before the store owns id allocation. Treat only `ConditionNotMatch` as the sign that a conditional create collided with an existing object. A store may report `AlreadyExists` for an unrelated path, for example when a parent of the object is a file, and that failure must come back as the write failure rather than as the error of the read that would compare content. Signed-off-by: jeremyhi * refactor(log-store): rename the catalog's out_of_order helper to out_of_order_reason Signed-off-by: jeremyhi --------- Signed-off-by: jeremyhi --- src/log-store/Cargo.toml | 1 + src/log-store/src/error.rs | 79 ++- src/log-store/src/object_store_wal.rs | 24 +- src/log-store/src/object_store_wal/batch.rs | 359 +++++++++++ src/log-store/src/object_store_wal/catalog.rs | 468 +++++++++++++++ src/log-store/src/object_store_wal/io.rs | 558 ++++++++++++++++++ 6 files changed, 1480 insertions(+), 9 deletions(-) create mode 100644 src/log-store/src/object_store_wal/batch.rs create mode 100644 src/log-store/src/object_store_wal/catalog.rs create mode 100644 src/log-store/src/object_store_wal/io.rs diff --git a/src/log-store/Cargo.toml b/src/log-store/Cargo.toml index ee2c6a12dc0..15560b06b2c 100644 --- a/src/log-store/Cargo.toml +++ b/src/log-store/Cargo.toml @@ -50,6 +50,7 @@ common-meta = { workspace = true, features = ["testing"] } common-test-util.workspace = true common-wal = { workspace = true, features = ["testing"] } itertools.workspace = true +object-store = { workspace = true, features = ["testing"] } rand.workspace = true rskafka = { workspace = true, features = ["unstable-fuzzing"] } uuid.workspace = true diff --git a/src/log-store/src/error.rs b/src/log-store/src/error.rs index dec7b42c3a5..54fcc84b690 100644 --- a/src/log-store/src/error.rs +++ b/src/log-store/src/error.rs @@ -315,6 +315,61 @@ pub enum Error { #[snafu(implicit)] location: Location, }, + + #[snafu(display("Invalid WAL object store, {}", reason))] + InvalidWalObjectStore { + reason: String, + #[snafu(implicit)] + location: Location, + }, + + #[snafu(display( + "Invalid WAL entry range, region: {}, start: {}, end: {}", + region_id, + start_entry_id, + end_entry_id + ))] + InvalidWalEntryRange { + region_id: RegionId, + start_entry_id: u64, + end_entry_id: u64, + #[snafu(implicit)] + location: Location, + }, + + #[snafu(display("WAL object sequence is exhausted, last sequence: {}", last_object_seq))] + WalObjectSequenceExhausted { + last_object_seq: u64, + #[snafu(implicit)] + location: Location, + }, + + #[snafu(display( + "WAL entry positions of region {} in one object are exhausted", + region_id + ))] + WalEntryPositionExhausted { + region_id: RegionId, + #[snafu(implicit)] + location: Location, + }, + + #[snafu(display("WAL object already exists with different content, path: {}", path))] + WalObjectConflict { + path: String, + #[snafu(implicit)] + location: Location, + }, + + #[snafu(display("Failed to {} WAL object, path: {}", operation, path))] + WalObjectStore { + operation: &'static str, + path: String, + #[snafu(source)] + error: object_store::Error, + #[snafu(implicit)] + location: Location, + }, } pub type Result = std::result::Result; @@ -345,7 +400,9 @@ impl ErrorExt for Error { | IllegalNamespace { .. } | MissingKey { .. } | MissingValue { .. } - | OverrideCompactedEntry { .. } => StatusCode::InvalidArguments, + | OverrideCompactedEntry { .. } + | InvalidWalObjectStore { .. } + | InvalidWalEntryRange { .. } => StatusCode::InvalidArguments, StartWalTask { .. } | StopWalTask { .. } | IllegalState { .. } @@ -360,12 +417,17 @@ impl ErrorExt for Error { | WaitDumpIndex { .. } | MetaLengthExceededLimit { .. } => StatusCode::Internal, - CorruptedWalObject { .. } => StatusCode::Unexpected, + CorruptedWalObject { .. } + | WalObjectConflict { .. } + | WalObjectSequenceExhausted { .. } + | WalEntryPositionExhausted { .. } => StatusCode::Unexpected, // Object store related errors - CreateWriter { .. } | WriteIndex { .. } | ReadIndex { .. } | Io { .. } => { - StatusCode::StorageUnavailable - } + CreateWriter { .. } + | WriteIndex { .. } + | ReadIndex { .. } + | WalObjectStore { .. } + | Io { .. } => StatusCode::StorageUnavailable, // Raft engine FetchEntry { .. } | RaftEngine { .. } | AddEntryLogBatch { .. } => { StatusCode::StorageUnavailable @@ -391,9 +453,10 @@ impl ErrorExt for Error { use Error::*; match self { - CreateWriter { error, .. } | WriteIndex { error, .. } | ReadIndex { error, .. } => { - retry_hint_from_opendal_error(error) - } + CreateWriter { error, .. } + | WriteIndex { error, .. } + | ReadIndex { error, .. } + | WalObjectStore { error, .. } => retry_hint_from_opendal_error(error), Io { error, .. } => retry_hint_from_io_error(error), FetchEntry { .. } | RaftEngine { .. } | AddEntryLogBatch { .. } => RetryHint::Retryable, ProduceRecord { error, .. } => match error { diff --git a/src/log-store/src/object_store_wal.rs b/src/log-store/src/object_store_wal.rs index f4cd7e081c7..98895869b81 100644 --- a/src/log-store/src/object_store_wal.rs +++ b/src/log-store/src/object_store_wal.rs @@ -29,7 +29,29 @@ //! 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. +//! +//! Object sequences increase monotonically within one prefix and may leave +//! gaps, so recovery continues after the largest sequence it indexed. An object +//! is created conditionally: rewriting a sequence with the content it already +//! holds is a no-op at the object store, while different content under a taken +//! sequence is a conflict. Recovery lists the objects, reads and verifies only +//! the header, trailer and footer of each, and indexes the footers in sequence +//! order to rebuild the object catalog, which rejects a sequence it already +//! holds. Segments are read and checksummed only when a read decodes them. +//! +//! Entry ids are object-sequence-major, see [`entry_id`]: the high bits of an +//! id name the object that holds the entry, the low bits its position among +//! the entries of its region in that object. -// The format has no callers until the store that writes and reads objects lands. +// These modules have no callers until the store that writes and reads objects lands. +#[allow(dead_code)] +mod batch; +#[allow(dead_code)] +mod catalog; #[allow(dead_code)] mod format; +#[allow(dead_code)] +mod io; + +#[allow(unused_imports)] +pub(crate) use batch::entry_id; diff --git a/src/log-store/src/object_store_wal/batch.rs b/src/log-store/src/object_store_wal/batch.rs new file mode 100644 index 00000000000..b07ac8e8cb0 --- /dev/null +++ b/src/log-store/src/object_store_wal/batch.rs @@ -0,0 +1,359 @@ +// Copyright 2023 Greptime Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Accumulation of admitted entries into the batch that becomes the next +//! object, and the entry id scheme. + +use std::collections::HashMap; +use std::time::Instant; + +use snafu::ensure; +use store_api::logstore::EntryId; +use store_api::logstore::entry::Entry; +use store_api::storage::RegionId; + +use crate::error::{Result, WalEntryPositionExhaustedSnafu}; + +/// Bits of an entry id that hold the position of the entry among the entries +/// of its region inside the object; the remaining high bits hold the object +/// sequence. +pub(crate) const POSITION_BITS: u32 = 20; +/// Positions are `1..POSITION_LIMIT`, so an object holds at most 2^20 - 1 +/// entries of one region and a batch seals before a region reaches the limit. +/// This is a theoretical bound: an object is sealed by size long before. +pub(crate) const POSITION_LIMIT: u64 = 1 << POSITION_BITS; +/// Object sequences are `0..OBJECT_SEQ_LIMIT`, the 44 bits an entry id leaves +/// above the position. +pub(crate) const OBJECT_SEQ_LIMIT: u64 = 1 << (u64::BITS - POSITION_BITS); + +/// Returns the id of the entry at `position` among the entries of its region +/// in the object `object_seq`. +/// +/// Ids are object-sequence-major: `entry_id >> 20` names the object that holds +/// the entry, and the low bits are its position in that object, which starts +/// at one so that id zero, the watermark of a region without entries, is never +/// assigned. A region's ids increase with the object sequence and have gaps +/// wherever other regions or other positions took the sequence. +pub(crate) fn entry_id(object_seq: u64, position: u64) -> EntryId { + debug_assert!(object_seq < OBJECT_SEQ_LIMIT && (1..POSITION_LIMIT).contains(&position)); + (object_seq << POSITION_BITS) | position +} + +/// Returns the smallest object sequence whose entry ids are all greater than +/// `entry_id`. Zero names no entry, so it needs no floor. +/// +/// An id assigned under the earlier contiguous scheme carries no object +/// information, but the floor keeps every new id above it just the same. +pub(crate) fn sequence_floor(entry_id: EntryId) -> u64 { + if entry_id == 0 { + 0 + } else { + (entry_id >> POSITION_BITS) + 1 + } +} + +/// Entries admitted since the last seal, together with the position of the +/// last entry admitted per region. +/// +/// Every entry is assigned its id at admission from the sequence the batch +/// takes when it is sealed, so a batch that is rolled back and admitted again +/// under the same sequence hands out the same ids. +#[derive(Debug)] +pub(crate) struct OpenBatch { + max_bytes: usize, + entries: Vec, + estimated_bytes: usize, + /// When the first entry of the batch was admitted. + first_admitted_at: Option, + positions: HashMap, +} + +impl OpenBatch { + pub(crate) fn new(max_bytes: usize) -> Self { + Self { + max_bytes, + entries: Vec::new(), + estimated_bytes: 0, + first_admitted_at: None, + positions: HashMap::new(), + } + } + + /// Returns true when admitting `entries` would take a region past the + /// position range, so that `entries` need a batch of their own. + pub(crate) fn would_exhaust_positions(&self, entries: &[Entry]) -> bool { + self.check_positions(entries).is_err() + } + + /// Admits `entries` into the batch that becomes the object `object_seq`, + /// assigning each the next position of its region, and returns the last id + /// assigned to every region in `entries`. Nothing is admitted when a + /// region would run past the position range. + pub(crate) fn admit( + &mut self, + object_seq: u64, + mut entries: Vec, + ) -> Result> { + self.check_positions(&entries)?; + let mut last_entry_ids = HashMap::new(); + for entry in &mut entries { + let region_id = entry.region_id(); + let position = self.positions.entry(region_id).or_insert(0); + *position += 1; + let entry_id = entry_id(object_seq, *position); + entry.set_entry_id(entry_id); + last_entry_ids.insert(region_id, entry_id); + } + self.estimated_bytes += entries.iter().map(Entry::estimated_size).sum::(); + if !entries.is_empty() { + self.first_admitted_at.get_or_insert_with(Instant::now); + } + self.entries.extend(entries); + Ok(last_entry_ids) + } + + /// Checks that every region in `entries` stays inside the position range + /// once they are admitted. + fn check_positions(&self, entries: &[Entry]) -> Result<()> { + let mut positions = HashMap::new(); + for entry in entries { + let region_id = entry.region_id(); + let position = positions + .entry(region_id) + .or_insert_with(|| self.positions.get(®ion_id).copied().unwrap_or(0)); + *position += 1; + ensure!( + *position < POSITION_LIMIT, + WalEntryPositionExhaustedSnafu { region_id } + ); + } + Ok(()) + } + + pub(crate) fn is_empty(&self) -> bool { + self.entries.is_empty() + } + + /// Returns the estimated size of the admitted entries. + pub(crate) fn estimated_bytes(&self) -> usize { + self.estimated_bytes + } + + /// Returns when the first admitted entry was admitted, if any. + pub(crate) fn first_admitted_at(&self) -> Option { + self.first_admitted_at + } + + /// Returns true once the admitted entries reach the size limit. + pub(crate) fn should_seal(&self) -> bool { + !self.is_empty() && self.estimated_bytes >= self.max_bytes + } + + /// Takes the admitted entries out of the batch together with the time the + /// first of them was admitted. The next admission starts at position one + /// again, under the next sequence. + pub(crate) fn seal(&mut self) -> (Vec, Instant) { + self.estimated_bytes = 0; + self.positions.clear(); + let first_admitted_at = self.first_admitted_at.take().unwrap_or_else(Instant::now); + (std::mem::take(&mut self.entries), first_admitted_at) + } + + /// Drops the admitted entries, so the next admission under the same + /// sequence hands out the same ids again. + pub(crate) fn reset(&mut self) { + let _ = self.seal(); + } +} + +#[cfg(test)] +mod tests { + use store_api::logstore::entry::NaiveEntry; + use store_api::logstore::provider::Provider; + + use super::*; + use crate::error::Error; + + fn entry(region_id: RegionId, payload_len: usize) -> Entry { + Entry::Naive(NaiveEntry { + provider: Provider::object_store_provider(region_id, "wal".to_string()), + region_id, + entry_id: 0, + data: vec![0; payload_len], + }) + } + + fn entry_ids(entries: &[Entry]) -> Vec<(RegionId, EntryId)> { + entries + .iter() + .map(|entry| (entry.region_id(), entry.entry_id())) + .collect() + } + + #[test] + fn test_entry_id_scheme() { + assert_eq!(1, entry_id(0, 1)); + assert_eq!(0x10_0001, entry_id(1, 1)); + // 7 << 52 | 3 + assert_eq!(31_525_197_391_593_475, entry_id(0x7_0000_0000, 3)); + assert_eq!(u64::MAX, entry_id(OBJECT_SEQ_LIMIT - 1, POSITION_LIMIT - 1)); + assert_eq!(1 << 44, OBJECT_SEQ_LIMIT); + + assert_eq!(0, sequence_floor(0)); + assert_eq!(1, sequence_floor(1)); + assert_eq!(1, sequence_floor(POSITION_LIMIT - 1)); + assert_eq!(2, sequence_floor(entry_id(1, 1))); + assert_eq!(8, sequence_floor(entry_id(7, POSITION_LIMIT - 1))); + // A contiguous id well past the first object is placed by its high bits. + assert_eq!(5, sequence_floor(5_000_000)); + assert_eq!(OBJECT_SEQ_LIMIT, sequence_floor(u64::MAX)); + } + + #[test] + fn test_batch_assigns_positions_per_region_under_the_object_sequence() { + let region_a = RegionId::new(1, 1); + let region_b = RegionId::new(1, 2); + let mut batch = OpenBatch::new(usize::MAX); + + let first = batch + .admit(5, vec![entry(region_a, 1), entry(region_b, 1)]) + .unwrap(); + assert_eq!( + HashMap::from([(region_a, entry_id(5, 1)), (region_b, entry_id(5, 1))]), + first + ); + let second = batch + .admit( + 5, + vec![entry(region_b, 1), entry(region_a, 1), entry(region_b, 1)], + ) + .unwrap(); + assert_eq!( + HashMap::from([(region_a, entry_id(5, 2)), (region_b, entry_id(5, 3))]), + second + ); + + assert_eq!( + vec![ + (region_a, entry_id(5, 1)), + (region_b, entry_id(5, 1)), + (region_b, entry_id(5, 2)), + (region_a, entry_id(5, 2)), + (region_b, entry_id(5, 3)), + ], + entry_ids(&batch.seal().0) + ); + assert!(batch.is_empty()); + assert_eq!( + HashMap::from([(region_a, entry_id(6, 1))]), + batch.admit(6, vec![entry(region_a, 1)]).unwrap() + ); + assert_eq!(POSITION_LIMIT, entry_id(6, 1) - entry_id(5, 1)); + } + + #[test] + fn test_batch_seals_at_size_limit() { + let region_id = RegionId::new(1, 1); + let first = entry(region_id, 8); + let second = entry(region_id, 8); + let max_bytes = first.estimated_size() + second.estimated_size(); + let mut batch = OpenBatch::new(max_bytes); + + assert!(!batch.should_seal()); + batch.admit(0, vec![first]).unwrap(); + assert!(!batch.should_seal()); + batch.admit(0, vec![second]).unwrap(); + assert!(batch.should_seal()); + + assert_eq!(2, batch.seal().0.len()); + assert!(!batch.should_seal()); + } + + #[test] + fn test_batch_admission_clock_starts_with_the_first_entry() { + let region_id = RegionId::new(1, 1); + let mut batch = OpenBatch::new(usize::MAX); + + assert!(batch.admit(0, Vec::new()).unwrap().is_empty()); + assert!(batch.is_empty()); + assert_eq!(None, batch.first_admitted_at()); + + let before = Instant::now(); + batch.admit(0, vec![entry(region_id, 1)]).unwrap(); + let first_admitted_at = batch.first_admitted_at().unwrap(); + assert!(first_admitted_at >= before); + batch.admit(0, vec![entry(region_id, 1)]).unwrap(); + assert_eq!(Some(first_admitted_at), batch.first_admitted_at()); + assert_eq!(first_admitted_at, batch.seal().1); + assert_eq!(None, batch.first_admitted_at()); + } + + #[test] + fn test_batch_reset_hands_out_the_same_ids_again() { + let region_id = RegionId::new(1, 1); + let mut batch = OpenBatch::new(usize::MAX); + + assert_eq!( + HashMap::from([(region_id, entry_id(3, 1))]), + batch.admit(3, vec![entry(region_id, 1)]).unwrap() + ); + batch.reset(); + assert!(batch.is_empty()); + assert_eq!( + HashMap::from([(region_id, entry_id(3, 1))]), + batch.admit(3, vec![entry(region_id, 1)]).unwrap() + ); + } + + #[test] + fn test_batch_refuses_a_region_past_the_position_range() { + let region_a = RegionId::new(1, 1); + let region_b = RegionId::new(1, 2); + let mut batch = OpenBatch::new(usize::MAX); + let entries = + |region_id, count: u64| (0..count).map(|_| entry(region_id, 0)).collect::>(); + + // An append that alone runs past the range fits no object. + let error = batch + .admit(0, entries(region_a, POSITION_LIMIT)) + .unwrap_err(); + assert!( + matches!(error, Error::WalEntryPositionExhausted { region_id, .. } if region_id == region_a), + "unexpected error: {error:?}" + ); + assert!(batch.is_empty()); + + // The range holds one entry fewer; the next entry of that region needs + // a new batch, while another region still fits. + let last = batch + .admit(0, entries(region_a, POSITION_LIMIT - 1)) + .unwrap(); + assert_eq!( + HashMap::from([(region_a, entry_id(0, POSITION_LIMIT - 1))]), + last + ); + assert!(batch.would_exhaust_positions(&entries(region_a, 1))); + assert!(!batch.would_exhaust_positions(&entries(region_b, 1))); + let error = batch.admit(0, entries(region_a, 1)).unwrap_err(); + assert!( + matches!(error, Error::WalEntryPositionExhausted { .. }), + "unexpected error: {error:?}" + ); + assert_eq!(POSITION_LIMIT as usize - 1, batch.seal().0.len()); + assert_eq!( + HashMap::from([(region_a, entry_id(1, 1))]), + batch.admit(1, entries(region_a, 1)).unwrap() + ); + } +} diff --git a/src/log-store/src/object_store_wal/catalog.rs b/src/log-store/src/object_store_wal/catalog.rs new file mode 100644 index 00000000000..5e9650d14a5 --- /dev/null +++ b/src/log-store/src/object_store_wal/catalog.rs @@ -0,0 +1,468 @@ +// Copyright 2023 Greptime Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! In-memory index over the footers of the objects of one WAL prefix. + +use std::collections::BTreeMap; +use std::ops::Bound::{Excluded, Unbounded}; + +use snafu::{OptionExt, ensure}; +use store_api::storage::RegionId; + +use crate::error::{ + CorruptedWalObjectSnafu, InvalidWalEntryRangeSnafu, Result, WalObjectSequenceExhaustedSnafu, +}; +use crate::object_store_wal::batch::{OBJECT_SEQ_LIMIT, sequence_floor}; +use crate::object_store_wal::format::FooterEntry; + +/// Indexes objects by sequence and, per region, the objects that hold entries +/// of that region. +#[derive(Debug, Default)] +pub(crate) struct ObjectCatalog { + objects: BTreeMap>, + regions: BTreeMap>, +} + +impl ObjectCatalog { + /// Indexes the footer of the object `object_seq`. Objects may be inserted + /// in any order, which lets recovery index them as it discovers them. + /// Inserting a sequence that is already indexed is rejected, whether or not + /// the footer matches the indexed one. + pub(crate) fn insert_object( + &mut self, + object_seq: u64, + mut footer: Vec, + ) -> Result<()> { + ensure!( + !footer.is_empty(), + CorruptedWalObjectSnafu { + reason: format!("object {object_seq} has an empty footer"), + } + ); + + footer.sort_unstable_by_key(|entry| entry.region_id); + for entries in footer.windows(2) { + ensure!( + entries[0].region_id != entries[1].region_id, + CorruptedWalObjectSnafu { + reason: format!( + "object {object_seq} has duplicate footer entries for region {}", + entries[0].region_id + ), + } + ); + } + for entry in &footer { + ensure!( + entry.entry_count > 0 && entry.min_entry_id <= entry.max_entry_id, + CorruptedWalObjectSnafu { + reason: format!( + "object {object_seq} has invalid entry range {}..={} with {} entries for region {}", + entry.min_entry_id, entry.max_entry_id, entry.entry_count, entry.region_id + ), + } + ); + } + + // Object sequences are unique: recovery indexes every listed key once, and + // accepting a repeated insertion would hide a caller that lost track of it. + // Retrying an identical write stays an object store concern. + ensure!( + !self.objects.contains_key(&object_seq), + CorruptedWalObjectSnafu { + reason: format!("object {object_seq} is already indexed"), + } + ); + + // Validate every region before mutating either index so insertion is atomic. + for entry in &footer { + let Some(region_objects) = self.regions.get(&entry.region_id) else { + continue; + }; + if let Some((&previous_seq, previous)) = region_objects.range(..object_seq).next_back() + { + ensure!( + previous.max_entry_id < entry.min_entry_id, + CorruptedWalObjectSnafu { + reason: out_of_order_reason( + entry.region_id, + previous_seq, + previous.max_entry_id, + object_seq, + entry.min_entry_id + ), + } + ); + } + if let Some((&next_seq, next)) = region_objects + .range((Excluded(object_seq), Unbounded)) + .next() + { + ensure!( + entry.max_entry_id < next.min_entry_id, + CorruptedWalObjectSnafu { + reason: out_of_order_reason( + entry.region_id, + object_seq, + entry.max_entry_id, + next_seq, + next.min_entry_id + ), + } + ); + } + } + + for entry in &footer { + self.regions + .entry(entry.region_id) + .or_default() + .insert(object_seq, entry.clone()); + } + self.objects.insert(object_seq, footer); + Ok(()) + } + + /// 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( + &self, + region_id: RegionId, + start_entry_id: u64, + end_entry_id: u64, + ) -> Result> { + ensure!( + start_entry_id <= end_entry_id, + InvalidWalEntryRangeSnafu { + region_id, + start_entry_id, + end_entry_id, + } + ); + + let Some(objects) = self.regions.get(®ion_id) else { + return Ok(Vec::new()); + }; + Ok(objects + .iter() + .filter(|(_, entry)| { + entry.max_entry_id >= start_entry_id && entry.min_entry_id <= end_entry_id + }) + .map(|(&object_seq, entry)| (object_seq, entry)) + .collect()) + } + + /// Returns the largest entry id indexed for `region_id`. + 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) + } + + /// Returns the sequence to assign to the next object written after recovery. + /// + /// An empty catalog starts at zero, so the first object of a prefix always + /// takes sequence zero. Otherwise the sequence continues after the largest + /// indexed one, which recovery discovers regardless of insertion order, and + /// is raised further when the largest entry id of a region lies at or above + /// the ids that sequence would assign: ids assigned under the earlier + /// contiguous scheme carry no object information, and every new id of a + /// region must be greater than every id it already has. A sequence at or + /// above [`OBJECT_SEQ_LIMIT`] does not fit an entry id and is rejected. + pub(crate) fn next_object_seq(&self) -> Result { + let after_last = match self.objects.last_key_value() { + None => 0, + Some((&last_object_seq, _)) => last_object_seq + .checked_add(1) + .context(WalObjectSequenceExhaustedSnafu { last_object_seq })?, + }; + let floor = self + .regions + .keys() + .filter_map(|region_id| self.region_max_entry_id(*region_id)) + .map(sequence_floor) + .max() + .unwrap_or(0); + let next_object_seq = after_last.max(floor); + ensure!( + next_object_seq < OBJECT_SEQ_LIMIT, + WalObjectSequenceExhaustedSnafu { + last_object_seq: next_object_seq - 1, + } + ); + Ok(next_object_seq) + } + + /// Iterates over the indexed objects ordered by object sequence. + pub(crate) fn objects_in_order(&self) -> impl Iterator + '_ { + self.objects + .iter() + .map(|(&object_seq, footer)| (object_seq, footer.as_slice())) + } +} + +fn out_of_order_reason( + region_id: RegionId, + lower_object_seq: u64, + lower_max_entry_id: u64, + upper_object_seq: u64, + upper_min_entry_id: u64, +) -> String { + format!( + "entry ranges of region {region_id} are not strictly increasing, object {lower_object_seq} ends at {lower_max_entry_id}, object {upper_object_seq} starts at {upper_min_entry_id}" + ) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::error::Error; + use crate::object_store_wal::batch::entry_id; + + #[test] + fn test_catalog_indexes_objects_and_queries_ranges() { + let region_one = RegionId::new(1, 1); + let region_two = RegionId::new(2, 1); + let mut catalog = ObjectCatalog::default(); + + // Recovery may discover objects out of order. + catalog + .insert_object( + 2, + vec![ + footer_entry(region_one, 4, 6), + footer_entry(region_two, 8, 9), + ], + ) + .unwrap(); + catalog + .insert_object(1, vec![footer_entry(region_one, 1, 3)]) + .unwrap(); + catalog + .insert_object(4, vec![footer_entry(region_one, 10, 12)]) + .unwrap(); + + assert_eq!(Some(12), catalog.region_max_entry_id(region_one)); + assert_eq!(Some(9), catalog.region_max_entry_id(region_two)); + assert_eq!(None, catalog.region_max_entry_id(RegionId::new(3, 1))); + + let objects = catalog.objects_for_entry_range(region_one, 3, 10).unwrap(); + assert_eq!(vec![1, 2, 4], object_seqs(&objects)); + assert_eq!( + vec![1, 2, 4], + catalog + .objects_in_order() + .map(|(object_seq, _)| object_seq) + .collect::>() + ); + } + + #[test] + fn test_catalog_rejects_duplicate_object_sequences() { + let region_id = RegionId::new(1, 1); + let mut catalog = ObjectCatalog::default(); + let first = footer_entry(region_id, 1, 2); + let second = footer_entry(RegionId::new(2, 1), 4, 5); + + catalog + .insert_object(1, vec![first.clone(), second.clone()]) + .unwrap(); + + // An identical footer is rejected just like a conflicting one. + assert_corrupted( + catalog.insert_object(1, vec![second, first.clone()]), + "object 1 is already indexed", + ); + + let mut conflicting = first; + conflicting.segment_offset += 1; + assert_corrupted( + catalog.insert_object(1, vec![conflicting]), + "object 1 is already indexed", + ); + assert_eq!(1, catalog.objects_in_order().count()); + } + + #[test] + fn test_catalog_resumes_object_sequence_after_recovery() { + let region_id = RegionId::new(1, 1); + let mut catalog = ObjectCatalog::default(); + assert_eq!(0, catalog.next_object_seq().unwrap()); + + // Recovery may discover objects out of order. + catalog + .insert_object(4, vec![footer_entry(region_id, 10, 12)]) + .unwrap(); + catalog + .insert_object(1, vec![footer_entry(region_id, 1, 3)]) + .unwrap(); + + assert_eq!(5, catalog.next_object_seq().unwrap()); + } + + #[test] + fn test_catalog_raises_object_sequence_above_existing_entry_ids() { + let region_one = RegionId::new(1, 1); + let region_two = RegionId::new(1, 2); + let mut catalog = ObjectCatalog::default(); + + // Ids that fit below the ids of the next sequence leave it alone. + catalog + .insert_object(0, vec![footer_entry(region_one, 1, 3)]) + .unwrap(); + catalog + .insert_object( + 1, + vec![footer_entry(region_two, entry_id(1, 1), entry_id(1, 2))], + ) + .unwrap(); + assert_eq!(2, catalog.next_object_seq().unwrap()); + + // A contiguous id past them names a later object: the sequence + // resumes above it, whichever region holds it. + catalog + .insert_object(2, vec![footer_entry(region_one, 5_000_000, 5_000_000)]) + .unwrap(); + assert_eq!(5, catalog.next_object_seq().unwrap()); + catalog + .insert_object( + 3, + vec![footer_entry(region_two, entry_id(7, 4), entry_id(7, 4))], + ) + .unwrap(); + assert_eq!(8, catalog.next_object_seq().unwrap()); + } + + #[test] + fn test_catalog_rejects_exhausted_object_sequence() { + let region_id = RegionId::new(1, 1); + let assert_exhausted = |catalog: &ObjectCatalog| { + let error = catalog.next_object_seq().unwrap_err(); + assert!( + error.to_string().contains("object sequence is exhausted"), + "unexpected error: {error}" + ); + }; + + // The last sequence that fits an entry id is indexed. + let mut catalog = ObjectCatalog::default(); + catalog + .insert_object(OBJECT_SEQ_LIMIT - 2, vec![footer_entry(region_id, 1, 2)]) + .unwrap(); + assert_eq!(OBJECT_SEQ_LIMIT - 1, catalog.next_object_seq().unwrap()); + catalog + .insert_object(OBJECT_SEQ_LIMIT - 1, vec![footer_entry(region_id, 3, 4)]) + .unwrap(); + assert_exhausted(&catalog); + + // A sequence that does not fit was written by an earlier scheme. + let mut catalog = ObjectCatalog::default(); + catalog + .insert_object(u64::MAX, vec![footer_entry(region_id, 1, 2)]) + .unwrap(); + assert_exhausted(&catalog); + + // An entry id that leaves no sequence above it. + let mut catalog = ObjectCatalog::default(); + catalog + .insert_object(0, vec![footer_entry(region_id, u64::MAX, u64::MAX)]) + .unwrap(); + assert_exhausted(&catalog); + } + + #[test] + fn test_catalog_rejects_overlapping_or_reversed_region_ranges() { + let region_id = RegionId::new(1, 1); + let mut catalog = ObjectCatalog::default(); + catalog + .insert_object(2, vec![footer_entry(region_id, 10, 20)]) + .unwrap(); + + assert_corrupted( + catalog.insert_object(3, vec![footer_entry(region_id, 20, 30)]), + "are not strictly increasing", + ); + assert_corrupted( + catalog.insert_object(3, vec![footer_entry(region_id, 5, 9)]), + "are not strictly increasing", + ); + assert_corrupted( + catalog.insert_object(1, vec![footer_entry(region_id, 15, 19)]), + "are not strictly increasing", + ); + assert_eq!(1, catalog.objects_in_order().count()); + } + + #[test] + fn test_catalog_rejects_duplicate_region_and_invalid_ranges() { + let region_id = RegionId::new(1, 1); + let mut catalog = ObjectCatalog::default(); + assert_corrupted( + catalog.insert_object( + 1, + vec![footer_entry(region_id, 1, 1), footer_entry(region_id, 2, 2)], + ), + "duplicate footer entries", + ); + assert_corrupted( + catalog.insert_object(1, vec![]), + "object 1 has an empty footer", + ); + + let mut invalid = footer_entry(region_id, 2, 1); + invalid.entry_count = 0; + assert_corrupted( + catalog.insert_object(1, vec![invalid]), + "has invalid entry range 2..=1", + ); + + let error = catalog + .objects_for_entry_range(region_id, 2, 1) + .unwrap_err(); + assert!( + matches!(error, Error::InvalidWalEntryRange { start_entry_id, end_entry_id, .. } if start_entry_id == 2 && end_entry_id == 1), + "unexpected error: {error:?}" + ); + } + + fn footer_entry(region_id: RegionId, min_entry_id: u64, max_entry_id: u64) -> FooterEntry { + FooterEntry { + region_id, + min_entry_id, + max_entry_id, + entry_count: max_entry_id + .checked_sub(min_entry_id) + .and_then(|count| count.checked_add(1)) + .unwrap_or(0) as u32, + segment_offset: min_entry_id.wrapping_mul(100), + segment_len: 100, + segment_crc32: min_entry_id as u32, + } + } + + fn object_seqs(objects: &[(u64, &FooterEntry)]) -> Vec { + objects.iter().map(|(object_seq, _)| *object_seq).collect() + } + + fn assert_corrupted(result: Result<()>, expected_reason: &str) { + match result { + Err(Error::CorruptedWalObject { reason, .. }) => assert!( + reason.contains(expected_reason), + "expected reason to contain {expected_reason:?}, actual {reason:?}" + ), + other => panic!("expected a corrupted object error, actual {other:?}"), + } + } +} diff --git a/src/log-store/src/object_store_wal/io.rs b/src/log-store/src/object_store_wal/io.rs new file mode 100644 index 00000000000..0af6ac70278 --- /dev/null +++ b/src/log-store/src/object_store_wal/io.rs @@ -0,0 +1,558 @@ +// Copyright 2023 Greptime Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Object store access for WAL objects: a deterministic key layout and +//! conditional creates that make a retry of the same object a no-op. + +use bytes::Bytes; +use object_store::{ErrorKind, ObjectStore}; +use snafu::{OptionExt, ResultExt, ensure}; + +use crate::error::{ + CorruptedWalObjectSnafu, InvalidWalObjectStoreSnafu, Result, WalObjectConflictSnafu, + WalObjectStoreSnafu, +}; + +/// Width of the zero-padded object sequence in an object key, wide enough for +/// [`u64::MAX`] so that lexicographic and numeric order agree. +const OBJECT_SEQ_WIDTH: usize = 20; +const OBJECT_SUFFIX: &str = ".wal"; + +/// Outcome of a conditional create. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum PutResult { + /// This call created the object. + Created, + /// The object already held the same content, so the call was a retry. + AlreadyPresent, +} + +/// An object discovered by a LIST. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct ListedObject { + pub(crate) object_seq: u64, + pub(crate) path: String, + /// Length of the object in bytes, as reported by the listing. + pub(crate) size: u64, +} + +/// Reads and writes the WAL objects under one prefix. +pub(crate) struct ObjectStoreIo { + store: ObjectStore, + object_prefix: String, +} + +impl ObjectStoreIo { + /// Binds `store` to the objects under `prefix`. The store must support + /// conditional creates, which is what keeps a retried write from + /// overwriting a durable object. + pub(crate) fn new(store: ObjectStore, prefix: impl AsRef) -> Result { + let object_prefix = normalize_prefix(prefix.as_ref())?; + ensure!( + store.info().capability().write_with_if_not_exists, + InvalidWalObjectStoreSnafu { + reason: "object store does not support conditional create", + } + ); + Ok(Self { + store, + object_prefix, + }) + } + + /// Creates the object `object_seq` unless it already exists. A retry with + /// the same content is accepted, different content is rejected. + /// + /// When the store reports that the object exists but the read that + /// compares its content fails, the read failure is returned, so that its + /// retry hint says whether comparing again may settle the outcome. Any + /// other create failure is returned as the write failure it is. + pub(crate) async fn put_if_absent(&self, object_seq: u64, content: Bytes) -> Result { + let path = self.object_path(object_seq); + let write_result = self + .store + .write_with(&path, content.clone()) + .if_not_exists(true) + .await; + + match write_result { + Ok(_) => Ok(PutResult::Created), + Err(error) => match self.store.read(&path).await { + Ok(existing) if existing.to_bytes() == content => Ok(PutResult::AlreadyPresent), + Ok(_) => WalObjectConflictSnafu { path }.fail(), + Err(read_error) if reports_existing_object(&error) => { + Err(read_error).context(WalObjectStoreSnafu { + operation: "read", + path, + }) + } + Err(_) => Err(error).context(WalObjectStoreSnafu { + operation: "write", + path, + }), + }, + } + } + + /// Reads the object `object_seq`. + pub(crate) async fn get(&self, object_seq: u64) -> Result { + let path = self.object_path(object_seq); + self.store + .read(&path) + .await + .map(|content| content.to_bytes()) + .context(WalObjectStoreSnafu { + operation: "read", + path, + }) + } + + /// Reads `len` bytes of the object `object_seq` starting at `offset`. The + /// range must lie inside the object; one that reaches past its end fails. + pub(crate) async fn get_range(&self, object_seq: u64, offset: u64, len: u64) -> Result { + let path = self.object_path(object_seq); + let end = offset + .checked_add(len) + .with_context(|| CorruptedWalObjectSnafu { + reason: format!("byte range {offset}..{len} overflows the object"), + })?; + self.store + .read_with(&path) + .range(offset..end) + .await + .map(|content| content.to_bytes()) + .context(WalObjectStoreSnafu { + operation: "read", + 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> { + let entries = self + .store + .list(&self.object_prefix) + .await + .with_context(|_| WalObjectStoreSnafu { + operation: "list", + path: self.object_prefix.clone(), + })?; + let mut objects = entries + .into_iter() + .filter_map(|entry| { + self.parse_object_seq(entry.path()) + .map(|object_seq| ListedObject { + object_seq, + path: entry.path().to_string(), + size: entry.metadata().content_length(), + }) + }) + .collect::>(); + objects.sort_unstable_by_key(|object| object.object_seq); + Ok(objects) + } + + pub(crate) fn object_path(&self, object_seq: u64) -> String { + format!( + "{}{object_seq:0OBJECT_SEQ_WIDTH$}{OBJECT_SUFFIX}", + self.object_prefix + ) + } + + fn parse_object_seq(&self, path: &str) -> Option { + let value = path + .strip_prefix(&self.object_prefix)? + .strip_suffix(OBJECT_SUFFIX)?; + if value.len() != OBJECT_SEQ_WIDTH || !value.bytes().all(|byte| byte.is_ascii_digit()) { + return None; + } + value.parse().ok() + } +} + +/// Returns true when a conditional create failed because the object exists. +/// Only the precondition failure says so: a store may report `AlreadyExists` +/// for an unrelated path, such as a parent that is a file. +fn reports_existing_object(error: &object_store::Error) -> bool { + error.kind() == ErrorKind::ConditionNotMatch +} + +fn normalize_prefix(prefix: &str) -> Result { + let prefix = prefix.trim(); + ensure!( + !prefix.is_empty() && !prefix.starts_with('/'), + InvalidWalObjectStoreSnafu { + reason: format!("object prefix {prefix:?} is empty or absolute"), + } + ); + let prefix = prefix.strip_suffix('/').unwrap_or(prefix); + ensure!( + !prefix + .split('/') + .any(|segment| segment.is_empty() || segment == "." || segment == ".."), + InvalidWalObjectStoreSnafu { + reason: format!("object prefix {prefix:?} has an empty or relative component"), + } + ); + Ok(format!("{prefix}/objects/")) +} + +#[cfg(test)] +mod tests { + use std::sync::Arc; + + use common_error::ext::{ErrorExt, RetryHint}; + use common_test_util::temp_dir::create_temp_dir; + use object_store::layers::mock::{self, MockLayerBuilder, oio}; + use object_store::secure_fs::SecureFsRoot; + use object_store::services::Memory; + + use super::*; + use crate::error::Error; + + fn memory_store() -> ObjectStore { + ObjectStore::new(Memory::default()).unwrap() + } + + /// A reader whose every read fails with a temporary error. + struct FailingReader; + + impl oio::Read for FailingReader { + async fn open( + &self, + _range: mock::BytesRange, + ) -> mock::Result<(mock::RpRead, Box)> { + Err(injected_failure("read").set_temporary()) + } + + async fn read( + &self, + _range: mock::BytesRange, + ) -> mock::Result<(mock::RpRead, mock::Buffer)> { + Err(injected_failure("read").set_temporary()) + } + } + + /// A writer that fails to close, so nothing is written. + struct FailingWriter; + + impl oio::Write for FailingWriter { + async fn write(&mut self, _buffer: mock::Buffer) -> mock::Result<()> { + Ok(()) + } + + async fn close(&mut self) -> mock::Result { + Err(injected_failure("write")) + } + + async fn abort(&mut self) -> mock::Result<()> { + Ok(()) + } + } + + fn injected_failure(operation: &str) -> mock::Error { + mock::Error::new( + mock::ErrorKind::Unexpected, + format!("injected {operation} failure"), + ) + } + + fn with_failing_reads(store: ObjectStore) -> ObjectStore { + store.layer( + MockLayerBuilder::default() + .reader_factory(Arc::new(|_, _, _| Box::new(FailingReader))) + .build() + .unwrap(), + ) + } + + fn with_failing_writes(store: ObjectStore) -> ObjectStore { + store.layer( + MockLayerBuilder::default() + .writer_factory(Arc::new(|_, _, _| Box::new(FailingWriter))) + .build() + .unwrap(), + ) + } + + fn memory_io() -> ObjectStoreIo { + ObjectStoreIo::new(memory_store(), "datanodes/1/epochs/2/").unwrap() + } + + #[test] + fn test_io_builds_and_parses_deterministic_object_paths() { + let io = ObjectStoreIo::new(memory_store(), " datanodes/1/epochs/2/ ").unwrap(); + assert_eq!( + "datanodes/1/epochs/2/objects/00000000000000000000.wal", + io.object_path(0) + ); + assert_eq!( + "datanodes/1/epochs/2/objects/00000000000000000042.wal", + io.object_path(42) + ); + assert_eq!( + "datanodes/1/epochs/2/objects/18446744073709551615.wal", + io.object_path(u64::MAX) + ); + assert_eq!(Some(42), io.parse_object_seq(&io.object_path(42))); + assert_eq!( + None, + io.parse_object_seq("datanodes/1/epochs/2/objects/42.wal") + ); + } + + #[test] + fn test_io_rejects_invalid_prefixes() { + for prefix in [ + "", + " ", + "/absolute", + ".", + "./node", + "node/./epoch", + "..", + "node/../epoch", + "node//epoch", + "node//", + "node///", + ] { + match ObjectStoreIo::new(memory_store(), prefix) { + Err(Error::InvalidWalObjectStore { .. }) => {} + Err(error) => panic!("unexpected error for prefix {prefix:?}: {error:?}"), + Ok(_) => panic!("expected prefix {prefix:?} to be rejected"), + } + } + } + + #[tokio::test] + async fn test_io_puts_gets_and_lists_objects_by_sequence() { + let io = memory_io(); + + for object_seq in [10, 2, 1] { + assert_eq!( + PutResult::Created, + io.put_if_absent(object_seq, Bytes::from(object_seq.to_string())) + .await + .unwrap() + ); + } + + assert_eq!(Bytes::from_static(b"2"), io.get(2).await.unwrap()); + let objects = io.list().await.unwrap(); + assert_eq!(vec![1, 2, 10], object_seqs(objects.clone())); + assert_eq!(io.object_path(1), objects[0].path); + assert_eq!( + vec![1, 1, 2], + objects.iter().map(|object| object.size).collect::>() + ); + } + + #[tokio::test] + async fn test_io_get_range_reads_a_slice_inside_the_object() { + let io = memory_io(); + io.put_if_absent(3, Bytes::from_static(b"0123456789")) + .await + .unwrap(); + + assert_eq!( + Bytes::from_static(b"234"), + io.get_range(3, 2, 3).await.unwrap() + ); + assert_eq!( + Bytes::from_static(b"89"), + io.get_range(3, 8, 2).await.unwrap() + ); + for (object_seq, offset, len) in [(3, 8, 5), (3, 10, 1), (7, 0, 1)] { + let error = io.get_range(object_seq, offset, len).await.unwrap_err(); + assert!( + matches!(error, Error::WalObjectStore { operation: "read", ref path, .. } if path == &io.object_path(object_seq)), + "unexpected error for range {offset}..{len} of object {object_seq}: {error:?}" + ); + } + let error = io.get_range(3, u64::MAX, 1).await.unwrap_err(); + assert!( + matches!(error, Error::CorruptedWalObject { .. }), + "unexpected error: {error:?}" + ); + } + + #[tokio::test] + async fn test_io_same_content_is_an_idempotent_retry() { + let io = memory_io(); + let content = Bytes::from_static(b"immutable"); + + assert_eq!( + PutResult::Created, + io.put_if_absent(7, content.clone()).await.unwrap() + ); + assert_eq!( + PutResult::AlreadyPresent, + io.put_if_absent(7, content.clone()).await.unwrap() + ); + assert_eq!(content, io.get(7).await.unwrap()); + } + + #[tokio::test] + async fn test_io_rejects_different_content_without_overwriting() { + let io = memory_io(); + let original = Bytes::from_static(b"original"); + io.put_if_absent(7, original.clone()).await.unwrap(); + + let error = io + .put_if_absent(7, Bytes::from_static(b"replacement")) + .await + .unwrap_err(); + + assert!( + matches!(error, Error::WalObjectConflict { ref path, .. } if path == &io.object_path(7)), + "unexpected error: {error:?}" + ); + assert_eq!(original, io.get(7).await.unwrap()); + } + + #[tokio::test] + async fn test_io_retry_of_an_existing_object_reports_the_failed_read() { + let store = memory_store(); + let prefix = "datanodes/1/epochs/2/"; + let content = Bytes::from_static(b"immutable"); + ObjectStoreIo::new(store.clone(), prefix) + .unwrap() + .put_if_absent(7, content.clone()) + .await + .unwrap(); + + // The object exists, so the create is refused, and the read that would + // settle the retry fails: that failure and its retry hint come back. + let io = ObjectStoreIo::new(with_failing_reads(store.clone()), prefix).unwrap(); + let error = io.put_if_absent(7, content.clone()).await.unwrap_err(); + assert!( + matches!(error, Error::WalObjectStore { operation: "read", ref path, .. } if path == &io.object_path(7)), + "unexpected error: {error:?}" + ); + assert_eq!(RetryHint::Retryable, error.retry_hint()); + + // A create that fails for another reason is reported as the write + // failure it is, although the read of the missing object fails too. + let io = ObjectStoreIo::new(with_failing_writes(store), prefix).unwrap(); + let error = io.put_if_absent(8, content).await.unwrap_err(); + assert!( + matches!(error, Error::WalObjectStore { operation: "write", ref path, .. } if path == &io.object_path(8)), + "unexpected error: {error:?}" + ); + assert_eq!(RetryHint::NonRetryable, error.retry_hint()); + } + + #[tokio::test] + async fn test_io_create_under_a_file_parent_is_reported_as_the_write_failure() { + let temp_dir = create_temp_dir("object_store_wal_io_file_parent"); + std::fs::write(temp_dir.path().join("datanodes"), []).unwrap(); + let store = SecureFsRoot::open(temp_dir.path()) + .unwrap() + .build_operator(); + let io = ObjectStoreIo::new(store, "datanodes/1/epochs/2/").unwrap(); + + // The store reports `AlreadyExists` for the parent, not for the + // object, so the create failure itself comes back. + let error = io + .put_if_absent(7, Bytes::from_static(b"wal")) + .await + .unwrap_err(); + assert!( + matches!(error, Error::WalObjectStore { operation: "write", ref path, .. } if path == &io.object_path(7)), + "unexpected error: {error:?}" + ); + } + + #[tokio::test] + async fn test_io_get_of_a_missing_object_fails() { + let io = memory_io(); + + let error = io.get(7).await.unwrap_err(); + assert!( + matches!(error, Error::WalObjectStore { operation: "read", ref path, .. } if path == &io.object_path(7)), + "unexpected error: {error:?}" + ); + } + + #[tokio::test] + async fn test_io_list_ignores_objects_outside_the_wal_layout() { + let io = memory_io(); + io.store.write("unrelated", "outside").await.unwrap(); + io.store + .write( + "datanodes/1/epochs/2/objects/not-a-sequence.wal", + "malformed", + ) + .await + .unwrap(); + io.put_if_absent(3, Bytes::from_static(b"wal")) + .await + .unwrap(); + + assert_eq!(vec![3], object_seqs(io.list().await.unwrap())); + } + + #[tokio::test] + async fn test_io_list_is_isolated_by_prefix() { + let store = memory_store(); + let first = ObjectStoreIo::new(store.clone(), "datanodes/1/epochs/1").unwrap(); + let second = ObjectStoreIo::new(store, "datanodes/1/epochs/2").unwrap(); + first + .put_if_absent(1, Bytes::from_static(b"first")) + .await + .unwrap(); + second + .put_if_absent(2, Bytes::from_static(b"second")) + .await + .unwrap(); + + assert_eq!(vec![1], object_seqs(first.list().await.unwrap())); + assert_eq!(vec![2], object_seqs(second.list().await.unwrap())); + } + + #[tokio::test] + async fn test_io_equal_sequences_in_different_prefixes_do_not_conflict() { + let store = memory_store(); + let first = ObjectStoreIo::new(store.clone(), "datanodes/1/epochs/1").unwrap(); + let second = ObjectStoreIo::new(store, "datanodes/1/epochs/2").unwrap(); + + assert_eq!( + PutResult::Created, + first + .put_if_absent(7, Bytes::from_static(b"first")) + .await + .unwrap() + ); + assert_eq!( + PutResult::Created, + second + .put_if_absent(7, Bytes::from_static(b"different")) + .await + .unwrap() + ); + assert_eq!(Bytes::from_static(b"first"), first.get(7).await.unwrap()); + assert_eq!( + Bytes::from_static(b"different"), + second.get(7).await.unwrap() + ); + } + + fn object_seqs(objects: Vec) -> Vec { + objects + .into_iter() + .map(|object| object.object_seq) + .collect() + } +}