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