Something went wrong. Try again.
A downstream fork of Tranquil PDS for use in margin.cafe
Something went wrong. Try again.
38 kB · 1152 lines
Rust
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153use std::cell::Cell;use std::collections::HashMap;use std::io;use std::sync::Arc;
use bytes::Bytes;use parking_lot::RwLock;use tracing::warn;
use crate::io::{MappedFile, StorageIO};
use super::manager::SegmentManager;use super::segment_file::{ReadEventRecord, SEGMENT_HEADER_SIZE, decode_event_record};use super::segment_index::{DEFAULT_INDEX_INTERVAL, SegmentIndex, rebuild_from_segment};use super::sidecar::{SidecarIndex, build_sidecar_from_segment};use super::types::{ DidHash, EventSequence, EventTypeTag, SegmentId, SegmentOffset, TimestampMicros,};
const FIRST_EVENT_OFFSET: SegmentOffset = SegmentOffset::new(SEGMENT_HEADER_SIZE as u64);
#[derive(Debug, Clone)]pub struct SequenceGap { pub after_segment: SegmentId, pub expected_seq: EventSequence, pub actual_seq: EventSequence,}
#[derive(Debug, Clone)]pub struct SequenceContiguityResult { pub total_segments: u64, pub min_seq: Option<EventSequence>, pub max_seq: Option<EventSequence>, pub gaps: Vec<SequenceGap>,}
impl SequenceContiguityResult { pub fn is_contiguous(&self) -> bool { self.gaps.is_empty() }}
#[derive(Debug, Clone)]pub struct RawEvent { pub seq: EventSequence, pub timestamp: TimestampMicros, pub did_hash: DidHash, pub event_type: EventTypeTag, pub payload: Bytes,}
#[derive(Debug, Clone, Copy)]struct SegmentRange { id: SegmentId, first: EventSequence, last: EventSequence,}
pub struct EventLogReader<S: StorageIO> { manager: Arc<SegmentManager<S>>, indexes: RwLock<HashMap<SegmentId, Arc<SegmentIndex>>>, sidecars: RwLock<HashMap<SegmentId, Arc<SidecarIndex>>>, ranges: RwLock<Vec<SegmentRange>>, mmaps: RwLock<HashMap<SegmentId, Arc<MappedFile>>>, active_segment: RwLock<Option<SegmentId>>, use_mmap: bool, skip_sealed_checksum: bool, max_payload: u32,}
impl<S: StorageIO> EventLogReader<S> { pub fn new( manager: Arc<SegmentManager<S>>, use_mmap: bool, skip_sealed_checksum: bool, max_payload: u32, ) -> Self { Self { manager, indexes: RwLock::new(HashMap::new()), sidecars: RwLock::new(HashMap::new()), ranges: RwLock::new(Vec::new()), mmaps: RwLock::new(HashMap::new()), active_segment: RwLock::new(None), use_mmap, skip_sealed_checksum, max_payload, } }
pub fn max_payload(&self) -> u32 { self.max_payload }
pub fn set_active_segment(&self, id: SegmentId) { *self.active_segment.write() = Some(id); }
pub fn extend_active_range(&self, first_seq: EventSequence, last_seq: EventSequence) { debug_assert!(first_seq <= last_seq); let active_id = match *self.active_segment.read() { Some(id) => id, None => return, };
let mut ranges = self.ranges.write(); match ranges.last_mut() { Some(last_range) if last_range.id == active_id => { last_range.last = last_seq; } _ => { ranges.push(SegmentRange { id: active_id, first: first_seq, last: last_seq, }); } } }
pub fn seed_index(&self, segment_id: SegmentId, index: SegmentIndex) { self.indexes.write().insert(segment_id, Arc::new(index)); }
pub fn load_index(&self, segment_id: SegmentId) -> io::Result<Arc<SegmentIndex>> { if let Some(idx) = self.indexes.read().get(&segment_id) { return Ok(Arc::clone(idx)); }
let index = match SegmentIndex::load( self.manager.io(), &self.manager.index_path(segment_id), ) { Ok(Some(idx)) => idx, Ok(None) => self.rebuild_index(segment_id)?, Err(e) => { warn!(segment = %segment_id, error = %e, "index load failed, rebuilding from segment scan"); self.rebuild_index(segment_id)? } };
let arc = Arc::new(index); self.indexes.write().insert(segment_id, Arc::clone(&arc)); Ok(arc) }
fn rebuild_index(&self, segment_id: SegmentId) -> io::Result<SegmentIndex> { let handle = self.manager.open_for_read(segment_id)?; let (idx, _) = rebuild_from_segment( self.manager.io(), handle.fd(), DEFAULT_INDEX_INTERVAL, self.max_payload, )?; let _ = idx.save(self.manager.io(), &self.manager.index_path(segment_id)); Ok(idx) }
pub fn find_segment_for_seq(&self, target_seq: EventSequence) -> Option<SegmentId> { let ranges = self.ranges.read(); let idx = ranges.partition_point(|r| r.last < target_seq); ranges .get(idx) .and_then(|r| (r.first <= target_seq).then_some(r.id)) }
pub fn refresh_segment_ranges(&self) -> io::Result<()> { let segment_ids = self.manager.list_segments()?; let active = *self.active_segment.read();
let new_ranges: Vec<SegmentRange> = segment_ids .iter() .filter_map(|&id| { let is_active = active.is_some_and(|a| a == id); let idx = match is_active { true => { if let Some(cached) = self.indexes.read().get(&id).cloned() { Ok(cached) } else { self.rebuild_index(id).map(|rebuilt| { let arc = Arc::new(rebuilt); self.indexes.write().insert(id, Arc::clone(&arc)); arc }) } } false => self.load_index(id), }; match idx { Ok(idx) => match (idx.first_seq(), idx.last_seq()) { (Some(first), Some(last)) => Some(SegmentRange { id, first, last }), _ => None, }, Err(e) => { warn!(segment = %id, error = %e, "failed to load index for range cache"); None } } }) .collect(); *self.ranges.write() = new_ranges; Ok(()) }
pub fn check_sequence_contiguity(&self) -> SequenceContiguityResult { let ranges = self.ranges.read(); let mut gaps: Vec<SequenceGap> = Vec::new(); let total_segments = ranges.len() as u64;
ranges.windows(2).for_each(|pair| { let expected_next = pair[0].last.next(); let actual_next = pair[1].first; if actual_next != expected_next { gaps.push(SequenceGap { after_segment: pair[0].id, expected_seq: expected_next, actual_seq: actual_next, }); } });
let max_seq = ranges.last().map(|r| r.last); let min_seq = ranges.first().map(|r| r.first);
SequenceContiguityResult { total_segments, min_seq, max_seq, gaps, } }
fn is_mmap_eligible(&self, segment_id: SegmentId) -> bool { self.use_mmap && self .active_segment .read() .is_none_or(|active| active != segment_id) }
fn get_mmap(&self, segment_id: SegmentId) -> io::Result<Arc<MappedFile>> { if let Some(m) = self.mmaps.read().get(&segment_id) { return Ok(Arc::clone(m)); }
let handle = self.manager.open_for_read(segment_id)?; let mapped = self.manager.io().mmap_file(handle.fd())?; let arc = Arc::new(mapped); self.mmaps.write().insert(segment_id, Arc::clone(&arc)); Ok(arc) }
fn scan_events_from_offset( &self, segment_id: SegmentId, start_offset: SegmentOffset, start_seq: EventSequence, limit: usize, events: &mut Vec<RawEvent>, predicate: impl FnMut(&EventSequence) -> bool, ) -> io::Result<bool> { if self.is_mmap_eligible(segment_id) { self.scan_mmap( segment_id, start_offset, start_seq, limit, events, predicate, ) } else { let handle = self.manager.open_for_read(segment_id)?; let file_size = self.manager.io().file_size(handle.fd())?; self.scan_direct( handle.fd(), file_size, start_offset, start_seq, limit, events, predicate, ) } }
fn scan_mmap( &self, segment_id: SegmentId, start_offset: SegmentOffset, start_seq: EventSequence, limit: usize, events: &mut Vec<RawEvent>, mut predicate: impl FnMut(&EventSequence) -> bool, ) -> io::Result<bool> { let mmap = self.get_mmap(segment_id)?; let mmap_bytes = Bytes::from_owner(OwnedMmap(Arc::clone(&mmap))); let data: &[u8] = (*mmap).as_ref(); let file_size = data.len() as u64; let skip_checksum = self.skip_sealed_checksum && self.is_mmap_eligible(segment_id); let offset = Cell::new(start_offset); let collected = Cell::new(0usize);
let max_payload = self.max_payload; std::iter::from_fn(|| { let cur = offset.get(); (cur.raw() < file_size && collected.get() < limit).then(|| { decode_mmap_event( data, &mmap_bytes, cur, file_size, segment_id, skip_checksum, max_payload, ) }) }) .try_for_each(|result| -> io::Result<()> { match result? { MmapDecodeResult::Valid(event, next_offset) => { offset.set(next_offset); if event.seq > start_seq && predicate(&event.seq) { events.push(event); collected.set(collected.get() + 1); } } MmapDecodeResult::Corrupted | MmapDecodeResult::Truncated | MmapDecodeResult::EndOfSegment => { offset.set(SegmentOffset::new(file_size)); } } Ok(()) })?; Ok(collected.get() >= limit) }
#[allow(clippy::too_many_arguments)] fn scan_direct( &self, fd: crate::io::FileId, file_size: u64, start_offset: SegmentOffset, start_seq: EventSequence, limit: usize, events: &mut Vec<RawEvent>, mut predicate: impl FnMut(&EventSequence) -> bool, ) -> io::Result<bool> { let offset = Cell::new(start_offset); let collected = Cell::new(0usize);
std::iter::from_fn(|| { let cur = offset.get(); (cur.raw() < file_size && collected.get() < limit).then(|| { decode_event_record(self.manager.io(), fd, cur, file_size, self.max_payload) }) }) .try_for_each(|result| -> io::Result<()> { match result? { Some(ReadEventRecord::Valid { event, next_offset }) => { offset.set(next_offset); if event.seq > start_seq && predicate(&event.seq) { events.push(RawEvent { seq: event.seq, timestamp: event.timestamp, did_hash: event.did_hash, event_type: event.event_type, payload: Bytes::from(event.payload), }); collected.set(collected.get() + 1); } } Some(ReadEventRecord::Corrupted { .. } | ReadEventRecord::Truncated { .. }) | None => { offset.set(SegmentOffset::new(file_size)); } } Ok(()) })?; Ok(collected.get() >= limit) }
pub fn read_events_from( &self, start_seq: EventSequence, limit: usize, ) -> io::Result<Vec<RawEvent>> { if limit == 0 { return Ok(Vec::new()); }
let ranges = self.ranges.read().clone(); let start_idx = match start_seq { EventSequence::BEFORE_ALL => Some(0), seq => { let point = ranges.partition_point(|r| r.last <= seq); (point < ranges.len()).then_some(point) } };
let start_idx = match start_idx { Some(idx) => idx, None => return Ok(Vec::new()), };
let mut events = Vec::with_capacity(limit.min(1024)); let done = Cell::new(false);
ranges[start_idx..] .iter() .enumerate() .take_while(|_| !done.get()) .try_for_each(|(i, range)| -> io::Result<()> { let remaining = limit - events.len(); let is_first = i == 0;
let (scan_offset, effective_seq) = match (is_first, start_seq) { (_, EventSequence::BEFORE_ALL) | (false, _) => { (FIRST_EVENT_OFFSET, EventSequence::BEFORE_ALL) } (true, seq) => { let index = self.load_index(range.id)?; (index.lookup(seq).unwrap_or(FIRST_EVENT_OFFSET), seq) } };
done.set(self.scan_events_from_offset( range.id, scan_offset, effective_seq, remaining, &mut events, |_| true, )?); Ok(()) })?;
Ok(events) }
pub fn read_event_at(&self, seq: EventSequence) -> io::Result<Option<RawEvent>> { let segment_id = match self.find_segment_for_seq(seq) { Some(id) => id, None => return Ok(None), };
let index = self.load_index(segment_id)?; let scan_offset = index.lookup(seq).unwrap_or(FIRST_EVENT_OFFSET);
let mut events = Vec::with_capacity(1); self.scan_events_from_offset( segment_id, scan_offset, seq.prev_or_before_all(), 1, &mut events, |s| *s == seq, )?;
Ok(events.into_iter().next()) }
pub fn on_segment_rotated( &self, sealed_id: SegmentId, new_active_id: SegmentId, ) -> io::Result<()> { self.invalidate_index(sealed_id); self.invalidate_mmap(sealed_id); self.invalidate_sidecar(sealed_id); self.set_active_segment(new_active_id); self.refresh_segment_ranges() }
pub fn invalidate_mmap(&self, segment_id: SegmentId) { self.mmaps.write().remove(&segment_id); }
pub fn invalidate_index(&self, segment_id: SegmentId) { self.indexes.write().remove(&segment_id); }
pub fn invalidate_sidecar(&self, segment_id: SegmentId) { self.sidecars.write().remove(&segment_id); }
pub fn load_sidecar(&self, segment_id: SegmentId) -> io::Result<Option<Arc<SidecarIndex>>> { if let Some(sc) = self.sidecars.read().get(&segment_id) { return Ok(Some(Arc::clone(sc))); }
let sidecar = match SidecarIndex::load( self.manager.io(), &self.manager.sidecar_path(segment_id), ) { Ok(Some(sc)) => sc, Ok(None) => return Ok(None), Err(e) => { warn!(segment = %segment_id, error = %e, "sidecar load failed, attempting rebuild"); match self.rebuild_sidecar(segment_id) { Ok(sc) => sc, Err(rebuild_err) => { warn!(segment = %segment_id, error = %rebuild_err, "sidecar rebuild also failed"); return Ok(None); } } } };
let arc = Arc::new(sidecar); self.sidecars.write().insert(segment_id, Arc::clone(&arc)); Ok(Some(arc)) }
fn rebuild_sidecar(&self, segment_id: SegmentId) -> io::Result<SidecarIndex> { let handle = self.manager.open_for_read(segment_id)?; let sidecar = build_sidecar_from_segment(self.manager.io(), handle.fd(), self.max_payload)?; let _ = sidecar.save(self.manager.io(), &self.manager.sidecar_path(segment_id)); Ok(sidecar) }}
struct OwnedMmap(Arc<MappedFile>);
impl AsRef<[u8]> for OwnedMmap { fn as_ref(&self) -> &[u8] { (*self.0).as_ref() }}
enum MmapDecodeResult { Valid(RawEvent, SegmentOffset), Corrupted, Truncated, EndOfSegment,}
fn decode_mmap_event( data: &[u8], mmap_bytes: &Bytes, offset: SegmentOffset, file_size: u64, segment_id: SegmentId, skip_checksum: bool, max_payload: u32,) -> io::Result<MmapDecodeResult> { use super::segment_file::EVENT_HEADER_SIZE;
let raw = offset.raw(); if raw > file_size { warn!( segment = %segment_id, offset = raw, file_size, "decode offset past file size, index likely corrupt" ); return Ok(MmapDecodeResult::Corrupted); } let remaining = file_size - raw; if remaining == 0 { return Ok(MmapDecodeResult::EndOfSegment); }
if remaining < EVENT_HEADER_SIZE as u64 { warn!( segment = %segment_id, offset = raw, remaining, "truncated record in sealed segment: not enough bytes for header" ); return Ok(MmapDecodeResult::Truncated); }
let base = usize::try_from(raw).expect("file offset exceeds platform address space"); let header_slice = &data[base..base + EVENT_HEADER_SIZE];
let seq_raw = u64::from_le_bytes(header_slice[0..8].try_into().unwrap()); if seq_raw == 0 { warn!( segment = %segment_id, offset = raw, "corrupted record in sealed segment: seq == 0" ); return Ok(MmapDecodeResult::Corrupted); } let seq = EventSequence::new(seq_raw);
let timestamp = TimestampMicros::new(u64::from_le_bytes(header_slice[8..16].try_into().unwrap())); let did_hash = DidHash::from_raw(u32::from_le_bytes(header_slice[16..20].try_into().unwrap())); let event_type = match EventTypeTag::from_raw(header_slice[20]) { Some(t) => t, None => { warn!( segment = %segment_id, offset = raw, tag = header_slice[20], "corrupted record in sealed segment: invalid event type" ); return Ok(MmapDecodeResult::Corrupted); } };
let payload_len = u32::from_le_bytes(header_slice[21..25].try_into().unwrap()); if payload_len > max_payload { warn!( segment = %segment_id, offset = raw, payload_len, "corrupted record in sealed segment: payload exceeds maximum" ); return Ok(MmapDecodeResult::Corrupted); }
let record_size = super::segment_file::EVENT_RECORD_OVERHEAD as u64 + u64::from(payload_len); if record_size > remaining { warn!( segment = %segment_id, offset = raw, record_size, remaining, "truncated record in sealed segment: record extends past file end" ); return Ok(MmapDecodeResult::Truncated); }
let payload_start = base + EVENT_HEADER_SIZE; let payload_end = payload_start + usize::try_from(payload_len).expect("payload_len fits usize");
if !skip_checksum { let checksum_start = payload_end; let stored_checksum = u32::from_le_bytes(data[checksum_start..checksum_start + 4].try_into().unwrap());
let mut hasher = xxhash_rust::xxh3::Xxh3::new(); hasher.update(header_slice); hasher.update(&data[payload_start..payload_end]); let computed = hasher.digest() as u32;
if stored_checksum != computed { warn!( segment = %segment_id, offset = raw, seq = %seq, stored = stored_checksum, computed, "corrupted record in sealed segment: checksum mismatch" ); return Ok(MmapDecodeResult::Corrupted); } }
let next_offset = offset.advance(record_size); Ok(MmapDecodeResult::Valid( RawEvent { seq, timestamp, did_hash, event_type, payload: mmap_bytes.slice(payload_start..payload_end), }, next_offset, ))}
#[cfg(test)]mod tests { use super::*; use crate::eventlog::segment_file::EVENT_RECORD_OVERHEAD; use crate::eventlog::types::MAX_EVENT_PAYLOAD; use crate::eventlog::writer::EventLogWriter; use crate::sim::SimulatedIO; use std::path::PathBuf;
fn setup_manager(max_segment_size: u64) -> Arc<SegmentManager<SimulatedIO>> { let sim = SimulatedIO::pristine(42); Arc::new(SegmentManager::new(sim, PathBuf::from("/segments"), max_segment_size).unwrap()) }
fn setup_with_events( event_count: u64, payload_size: usize, max_segment_size: u64, ) -> ( Arc<SegmentManager<SimulatedIO>>, EventLogReader<SimulatedIO>, ) { let mgr = setup_manager(max_segment_size); { let mut writer = EventLogWriter::open(Arc::clone(&mgr), DEFAULT_INDEX_INTERVAL, MAX_EVENT_PAYLOAD) .unwrap(); (1..=event_count).for_each(|i| { writer .append( DidHash::from_did(&format!("did:plc:user{i}")), EventTypeTag::COMMIT, vec![0xAA; payload_size], ) .unwrap(); }); writer.shutdown().unwrap(); } mgr.shutdown();
let reader = EventLogReader::new(Arc::clone(&mgr), false, false, MAX_EVENT_PAYLOAD); reader.refresh_segment_ranges().unwrap(); (mgr, reader) }
fn setup_multi_segment( events_per_segment: u64, num_segments: u64, payload_size: usize, ) -> ( Arc<SegmentManager<SimulatedIO>>, EventLogReader<SimulatedIO>, ) { let record_size = EVENT_RECORD_OVERHEAD + payload_size; let max_segment_size = (SEGMENT_HEADER_SIZE + record_size * events_per_segment as usize) as u64;
let mgr = setup_manager(max_segment_size); { let mut writer = EventLogWriter::open(Arc::clone(&mgr), DEFAULT_INDEX_INTERVAL, MAX_EVENT_PAYLOAD) .unwrap(); let total = events_per_segment * num_segments; (1..=total).for_each(|i| { writer .append( DidHash::from_did(&format!("did:plc:user{i}")), EventTypeTag::COMMIT, vec![i as u8; payload_size], ) .unwrap(); if i % events_per_segment == 0 && i < total { writer.sync().unwrap(); writer.rotate_if_needed().unwrap(); } }); writer.shutdown().unwrap(); } mgr.shutdown();
let reader = EventLogReader::new(Arc::clone(&mgr), false, false, MAX_EVENT_PAYLOAD); reader.refresh_segment_ranges().unwrap(); (mgr, reader) }
#[test] fn read_events_from_single_segment() { let (_, reader) = setup_with_events(10, 50, 64 * 1024);
let events = reader .read_events_from(EventSequence::BEFORE_ALL, 100) .unwrap(); assert_eq!(events.len(), 10); events.iter().enumerate().for_each(|(i, e)| { assert_eq!(e.seq, EventSequence::new(i as u64 + 1)); assert_eq!(e.event_type, EventTypeTag::COMMIT); assert_eq!(e.payload.len(), 50); }); }
#[test] fn read_events_from_cursor() { let (_, reader) = setup_with_events(10, 50, 64 * 1024);
let events = reader.read_events_from(EventSequence::new(5), 100).unwrap(); assert_eq!(events.len(), 5); assert_eq!(events[0].seq, EventSequence::new(6)); assert_eq!(events[4].seq, EventSequence::new(10)); }
#[test] fn read_events_respects_limit() { let (_, reader) = setup_with_events(10, 50, 64 * 1024);
let events = reader .read_events_from(EventSequence::BEFORE_ALL, 3) .unwrap(); assert_eq!(events.len(), 3); assert_eq!(events[0].seq, EventSequence::new(1)); assert_eq!(events[2].seq, EventSequence::new(3)); }
#[test] fn read_events_empty_on_zero_limit() { let (_, reader) = setup_with_events(5, 50, 64 * 1024); let events = reader .read_events_from(EventSequence::BEFORE_ALL, 0) .unwrap(); assert!(events.is_empty()); }
#[test] fn read_event_at_existing() { let (_, reader) = setup_with_events(10, 50, 64 * 1024);
let event = reader.read_event_at(EventSequence::new(5)).unwrap(); assert!(event.is_some()); let event = event.unwrap(); assert_eq!(event.seq, EventSequence::new(5)); assert_eq!(event.payload.len(), 50); }
#[test] fn read_event_at_missing() { let (_, reader) = setup_with_events(5, 50, 64 * 1024);
let event = reader.read_event_at(EventSequence::new(100)).unwrap(); assert!(event.is_none()); }
#[test] fn cross_segment_read() { let (_, reader) = setup_multi_segment(3, 3, 50);
let events = reader .read_events_from(EventSequence::BEFORE_ALL, 100) .unwrap(); assert_eq!(events.len(), 9); events.iter().enumerate().for_each(|(i, e)| { assert_eq!(e.seq, EventSequence::new(i as u64 + 1)); }); }
#[test] fn cross_segment_cursor_resumption() { let (_, reader) = setup_multi_segment(3, 3, 50);
let events = reader.read_events_from(EventSequence::new(4), 100).unwrap(); assert_eq!(events.len(), 5); assert_eq!(events[0].seq, EventSequence::new(5)); assert_eq!(events[4].seq, EventSequence::new(9)); }
#[test] fn cross_segment_limit_respected() { let (_, reader) = setup_multi_segment(3, 3, 50);
let events = reader .read_events_from(EventSequence::BEFORE_ALL, 5) .unwrap(); assert_eq!(events.len(), 5); assert_eq!(events[0].seq, EventSequence::new(1)); assert_eq!(events[4].seq, EventSequence::new(5)); }
#[test] fn cross_segment_limit_at_boundary() { let (_, reader) = setup_multi_segment(3, 3, 50);
let events = reader.read_events_from(EventSequence::new(2), 5).unwrap(); assert_eq!(events.len(), 5); assert_eq!(events[0].seq, EventSequence::new(3)); assert_eq!(events[4].seq, EventSequence::new(7)); }
#[test] fn find_segment_for_seq_locates_correct_segment() { let (_, reader) = setup_multi_segment(3, 2, 50);
assert_eq!( reader.find_segment_for_seq(EventSequence::new(1)), Some(SegmentId::new(1)) ); assert_eq!( reader.find_segment_for_seq(EventSequence::new(3)), Some(SegmentId::new(1)) ); assert_eq!( reader.find_segment_for_seq(EventSequence::new(4)), Some(SegmentId::new(2)) ); assert_eq!( reader.find_segment_for_seq(EventSequence::new(6)), Some(SegmentId::new(2)) ); assert_eq!(reader.find_segment_for_seq(EventSequence::new(100)), None); }
#[test] fn index_caching_returns_same_arc() { let (_, reader) = setup_with_events(5, 50, 64 * 1024);
let idx1 = reader.load_index(SegmentId::new(1)).unwrap(); let idx2 = reader.load_index(SegmentId::new(1)).unwrap(); assert!(Arc::ptr_eq(&idx1, &idx2)); }
#[test] fn refresh_after_segment_deletion() { let (mgr, reader) = setup_multi_segment(3, 3, 50);
assert_eq!( reader.find_segment_for_seq(EventSequence::new(1)), Some(SegmentId::new(1)) );
mgr.delete_segment(SegmentId::new(1)).unwrap(); reader.invalidate_index(SegmentId::new(1)); reader.invalidate_mmap(SegmentId::new(1)); reader.refresh_segment_ranges().unwrap();
assert_eq!(reader.find_segment_for_seq(EventSequence::new(1)), None); assert_eq!( reader.find_segment_for_seq(EventSequence::new(4)), Some(SegmentId::new(2)) ); }
#[test] fn mmap_read_matches_direct_read() { let (mgr, direct_reader) = setup_with_events(10, 50, 64 * 1024);
let mmap_reader = EventLogReader::new(Arc::clone(&mgr), true, false, MAX_EVENT_PAYLOAD); mmap_reader.refresh_segment_ranges().unwrap();
let direct_events = direct_reader .read_events_from(EventSequence::BEFORE_ALL, 100) .unwrap(); let mmap_events = mmap_reader .read_events_from(EventSequence::BEFORE_ALL, 100) .unwrap();
assert_eq!(direct_events.len(), mmap_events.len()); direct_events .iter() .zip(mmap_events.iter()) .for_each(|(d, m)| { assert_eq!(d.seq, m.seq); assert_eq!(d.timestamp, m.timestamp); assert_eq!(d.did_hash, m.did_hash); assert_eq!(d.event_type, m.event_type); assert_eq!(d.payload, m.payload); }); }
#[test] fn read_event_at_first_and_last() { let (_, reader) = setup_with_events(20, 50, 64 * 1024);
let first = reader .read_event_at(EventSequence::new(1)) .unwrap() .unwrap(); assert_eq!(first.seq, EventSequence::new(1));
let last = reader .read_event_at(EventSequence::new(20)) .unwrap() .unwrap(); assert_eq!(last.seq, EventSequence::new(20)); }
#[test] fn empty_reader_returns_empty() { let mgr = setup_manager(64 * 1024); let reader = EventLogReader::new(Arc::clone(&mgr), false, false, MAX_EVENT_PAYLOAD); reader.refresh_segment_ranges().unwrap();
let events = reader .read_events_from(EventSequence::BEFORE_ALL, 100) .unwrap(); assert!(events.is_empty());
let event = reader.read_event_at(EventSequence::new(1)).unwrap(); assert!(event.is_none()); }
#[test] fn cursor_past_end_returns_empty() { let (_, reader) = setup_with_events(5, 50, 64 * 1024); let events = reader.read_events_from(EventSequence::new(5), 100).unwrap(); assert!(events.is_empty()); }
#[test] fn cross_segment_read_event_at() { let (_, reader) = setup_multi_segment(3, 3, 50);
(1..=9).for_each(|i| { let event = reader .read_event_at(EventSequence::new(i)) .unwrap() .unwrap(); assert_eq!(event.seq, EventSequence::new(i)); assert_eq!(event.payload[0], i as u8); }); }
#[test] fn different_event_types_preserved() { let mgr = setup_manager(64 * 1024); { let mut writer = EventLogWriter::open(Arc::clone(&mgr), DEFAULT_INDEX_INTERVAL, MAX_EVENT_PAYLOAD) .unwrap(); let types = [ EventTypeTag::COMMIT, EventTypeTag::IDENTITY, EventTypeTag::ACCOUNT, EventTypeTag::SYNC, ]; types.iter().enumerate().for_each(|(i, &et)| { writer .append( DidHash::from_did(&format!("did:plc:user{i}")), et, vec![0xAA; 32], ) .unwrap(); }); writer.shutdown().unwrap(); } mgr.shutdown();
let reader = EventLogReader::new(Arc::clone(&mgr), false, false, MAX_EVENT_PAYLOAD); reader.refresh_segment_ranges().unwrap();
let events = reader .read_events_from(EventSequence::BEFORE_ALL, 100) .unwrap(); assert_eq!(events[0].event_type, EventTypeTag::COMMIT); assert_eq!(events[1].event_type, EventTypeTag::IDENTITY); assert_eq!(events[2].event_type, EventTypeTag::ACCOUNT); assert_eq!(events[3].event_type, EventTypeTag::SYNC); }
#[test] fn active_segment_excludes_mmap() { let (mgr, _) = setup_with_events(10, 50, 64 * 1024);
let reader = EventLogReader::new(Arc::clone(&mgr), true, false, MAX_EVENT_PAYLOAD); reader.set_active_segment(SegmentId::new(1)); reader.refresh_segment_ranges().unwrap();
assert!(!reader.is_mmap_eligible(SegmentId::new(1))); assert!(reader.is_mmap_eligible(SegmentId::new(2))); }
#[test] fn no_active_segment_mmaps_all() { let reader: EventLogReader<SimulatedIO> = EventLogReader::new(setup_manager(64 * 1024), true, false, MAX_EVENT_PAYLOAD);
assert!(reader.is_mmap_eligible(SegmentId::new(1))); assert!(reader.is_mmap_eligible(SegmentId::new(99))); }
#[test] fn corrupt_index_offset_does_not_panic() { let mgr = setup_manager(64 * 1024); { let mut writer = EventLogWriter::open(Arc::clone(&mgr), DEFAULT_INDEX_INTERVAL, MAX_EVENT_PAYLOAD) .unwrap(); (1..=5).for_each(|i| { writer .append( DidHash::from_did(&format!("did:plc:user{i}")), EventTypeTag::COMMIT, vec![0xAA; 50], ) .unwrap(); }); writer.shutdown().unwrap(); } mgr.shutdown();
let mut bad_index = SegmentIndex::new(); bad_index.record(EventSequence::new(1), SegmentOffset::new(999_999)); bad_index.record(EventSequence::new(5), SegmentOffset::new(999_999)); bad_index .save(mgr.io(), &mgr.index_path(SegmentId::new(1))) .unwrap();
let reader = EventLogReader::new(Arc::clone(&mgr), false, false, MAX_EVENT_PAYLOAD); reader.refresh_segment_ranges().unwrap();
let events = reader .read_events_from(EventSequence::BEFORE_ALL, 100) .unwrap(); assert!(events.is_empty() || events.len() <= 5);
let event = reader.read_event_at(EventSequence::new(3)).unwrap(); assert!(event.is_none() || event.is_some_and(|e| e.seq == EventSequence::new(3))); }
#[test] fn corrupt_index_offset_mmap_does_not_panic() { let mgr = setup_manager(64 * 1024); { let mut writer = EventLogWriter::open(Arc::clone(&mgr), DEFAULT_INDEX_INTERVAL, MAX_EVENT_PAYLOAD) .unwrap(); (1..=5).for_each(|i| { writer .append( DidHash::from_did(&format!("did:plc:user{i}")), EventTypeTag::COMMIT, vec![0xAA; 50], ) .unwrap(); }); writer.shutdown().unwrap(); } mgr.shutdown();
let mut bad_index = SegmentIndex::new(); bad_index.record(EventSequence::new(1), SegmentOffset::new(999_999)); bad_index.record(EventSequence::new(5), SegmentOffset::new(999_999)); bad_index .save(mgr.io(), &mgr.index_path(SegmentId::new(1))) .unwrap();
let reader = EventLogReader::new(Arc::clone(&mgr), true, false, MAX_EVENT_PAYLOAD); reader.refresh_segment_ranges().unwrap();
let events = reader .read_events_from(EventSequence::BEFORE_ALL, 100) .unwrap(); assert!(events.is_empty() || events.len() <= 5); }
#[test] fn on_segment_rotated_updates_state() { let (mgr, _direct_reader) = setup_multi_segment(3, 2, 50);
let reader = EventLogReader::new(Arc::clone(&mgr), true, false, MAX_EVENT_PAYLOAD); reader.refresh_segment_ranges().unwrap();
assert_eq!( reader.find_segment_for_seq(EventSequence::new(1)), Some(SegmentId::new(1)) );
reader .on_segment_rotated(SegmentId::new(1), SegmentId::new(2)) .unwrap();
assert!(!reader.is_mmap_eligible(SegmentId::new(2))); assert!(reader.is_mmap_eligible(SegmentId::new(1)));
let events = reader .read_events_from(EventSequence::BEFORE_ALL, 100) .unwrap(); assert_eq!(events.len(), 6); }}