diff --git a/crates/tranquil-store/src/eventlog/commit_loop.rs b/crates/tranquil-store/src/eventlog/commit_loop.rs index 2ab0764..ec89d05 100644 --- a/crates/tranquil-store/src/eventlog/commit_loop.rs +++ b/crates/tranquil-store/src/eventlog/commit_loop.rs @@ -21,6 +21,10 @@ const SYNC_TIMEOUT: Duration = Duration::from_secs(30); const REORDER_TIMEOUT: Duration = Duration::from_millis(100); const GAP_ABANDON_TIMEOUT: Duration = Duration::from_secs(5); +pub(super) fn writer_terminated() -> io::Error { + io::Error::other("eventlog writer thread terminated") +} + pub struct FreezeResponse { pub synced_through: EventSequence, pub segment_id: SegmentId, @@ -42,6 +46,7 @@ pub enum WriterRequest { pub struct WriterNotify { synced_seq: AtomicU64, poisoned: AtomicBool, + terminated: AtomicBool, mutex: Mutex<()>, cond: Condvar, } @@ -128,11 +133,16 @@ impl WriterNotify { Self { synced_seq: AtomicU64::new(initial_synced), poisoned: AtomicBool::new(false), + terminated: AtomicBool::new(false), mutex: Mutex::new(()), cond: Condvar::new(), } } + pub(super) fn is_terminated(&self) -> bool { + self.terminated.load(Ordering::Acquire) + } + pub fn wait_for_sync(&self, target: EventSequence) -> io::Result<()> { let target_raw = target.raw(); @@ -143,6 +153,9 @@ impl WriterNotify { if self.poisoned.load(Ordering::Acquire) { return Err(io::Error::other("eventlog writer poisoned")); } + if self.terminated.load(Ordering::Acquire) { + return Err(io::Error::other("eventlog writer thread terminated")); + } let deadline = Instant::now() + SYNC_TIMEOUT; let mut guard = self.mutex.lock(); @@ -154,6 +167,9 @@ impl WriterNotify { if self.poisoned.load(Ordering::Acquire) { return Err(io::Error::other("eventlog writer poisoned")); } + if self.terminated.load(Ordering::Acquire) { + return Err(io::Error::other("eventlog writer thread terminated")); + } let now = Instant::now(); if now >= deadline { @@ -178,6 +194,12 @@ impl WriterNotify { let _guard = self.mutex.lock(); self.cond.notify_all(); } + + fn terminate(&self) { + self.terminated.store(true, Ordering::Release); + let _guard = self.mutex.lock(); + self.cond.notify_all(); + } } struct ReorderBuffer { @@ -372,12 +394,40 @@ impl<'a> Drop for CloseOnDrop<'a> { } } +fn fail_abandoned_request(request: WriterRequest) { + match request { + WriterRequest::SyncBarrier { response } => { + let _ = response.send(Err(writer_terminated())); + } + WriterRequest::Freeze { response, .. } => { + let _ = response.send(Err(writer_terminated())); + } + WriterRequest::Append(_) | WriterRequest::Shutdown => {} + } +} + +struct TerminateOnDrop<'a> { + notify: &'a WriterNotify, + receiver: &'a flume::Receiver, +} + +impl Drop for TerminateOnDrop<'_> { + fn drop(&mut self) { + self.notify.terminate(); + self.receiver.drain().for_each(fail_abandoned_request); + } +} + fn writer_loop( receiver: &flume::Receiver, writer: &mut EventLogWriter, ctx: &WriterCtx<'_, S>, ) { let _close = CloseOnDrop(ctx.pending_bytes); + let _terminate = TerminateOnDrop { + notify: ctx.notify, + receiver, + }; let mut reorder = ReorderBuffer::new(writer.current_seq().next().raw()); loop { diff --git a/crates/tranquil-store/src/eventlog/mod.rs b/crates/tranquil-store/src/eventlog/mod.rs index fa08a46..b26520b 100644 --- a/crates/tranquil-store/src/eventlog/mod.rs +++ b/crates/tranquil-store/src/eventlog/mod.rs @@ -26,7 +26,10 @@ use crate::blockstore::BlocksSynced; use crate::fsync_order::PostBlockstoreHook; use crate::io::StorageIO; -use commit_loop::{CommitThread, FreezeResponse, PendingBytesBudget, WriterNotify, WriterRequest}; +use commit_loop::{ + CommitThread, FreezeResponse, PendingBytesBudget, WriterNotify, WriterRequest, + writer_terminated, +}; pub use bridge::{DeferredBroadcast, EventLogBridge}; pub use manager::{SEGMENT_FILE_EXTENSION, SegmentManager, parse_segment_id, segment_path}; @@ -51,6 +54,7 @@ pub use writer::{EventLogWriter, SyncResult}; const DEFAULT_BROADCAST_BUFFER: usize = 16384; pub const DEFAULT_PENDING_BYTES_BUDGET: u64 = 1024 * 1024 * 1024; +const WRITER_RESPONSE_POLL: Duration = Duration::from_millis(100); pub struct EventWithMutations { pub event: SequencedEvent, @@ -187,7 +191,7 @@ impl EventLog { self.commit_thread .sender() .send(WriterRequest::Append(event)) - .map_err(|_| io::Error::other("eventlog writer thread terminated")) + .map_err(|_| writer_terminated()) } pub fn append_event( @@ -261,10 +265,24 @@ impl EventLog { self.commit_thread .sender() .send(WriterRequest::SyncBarrier { response: resp_tx }) - .map_err(|_| io::Error::other("eventlog writer thread terminated"))?; - resp_rx - .recv() - .map_err(|_| io::Error::other("eventlog writer thread terminated"))? + .map_err(|_| writer_terminated())?; + self.await_writer_response(&resp_rx) + } + + fn await_writer_response(&self, resp_rx: &flume::Receiver>) -> io::Result { + loop { + match resp_rx.recv_timeout(WRITER_RESPONSE_POLL) { + Ok(result) => return result, + Err(flume::RecvTimeoutError::Disconnected) => { + return Err(writer_terminated()); + } + Err(flume::RecvTimeoutError::Timeout) => { + if self.notify.is_terminated() { + return Err(writer_terminated()); + } + } + } + } } pub fn get_events_since( @@ -432,11 +450,9 @@ impl EventLog { response: resp_tx, resume: resume_rx, }) - .map_err(|_| io::Error::other("eventlog writer thread terminated"))?; + .map_err(|_| writer_terminated())?; - let freeze_resp: FreezeResponse = resp_rx - .recv() - .map_err(|_| io::Error::other("eventlog writer thread terminated"))??; + let freeze_resp: FreezeResponse = self.await_writer_response(&resp_rx)?; let all_segments = self.manager.list_segments()?; let sealed_segments: Vec = all_segments diff --git a/crates/tranquil-store/tests/sim_cross_store.rs b/crates/tranquil-store/tests/sim_cross_store.rs index b6801ba..156d435 100644 --- a/crates/tranquil-store/tests/sim_cross_store.rs +++ b/crates/tranquil-store/tests/sim_cross_store.rs @@ -51,9 +51,8 @@ fn seed_repo(stores: &TestStores, idx: u64) -> Uuid { uid } -fn append_event(stores: &TestStores, idx: u64) { - let did = test_did(idx); - let event = SequencedEvent { +fn make_commit_event(did: &Did, idx: u64) -> SequencedEvent { + SequencedEvent { seq: SequenceNumber::from_raw(0), did: did.clone(), created_at: chrono::Utc::now(), @@ -68,7 +67,12 @@ fn append_event(stores: &TestStores, idx: u64) { active: None, status: None, rev: Some(format!("rev{idx}")), - }; + } +} + +fn append_event(stores: &TestStores, idx: u64) { + let did = test_did(idx); + let event = make_commit_event(&did, idx); stores .eventlog .append_event(&did, RepoEventType::Commit, &event) @@ -991,6 +995,71 @@ fn sim_cross_store_coordinated_commit_under_faults() { }); } +fn open_fault_eventlog( + dir: &std::path::Path, + sim: &std::sync::Arc, +) -> std::sync::Arc>> { + std::sync::Arc::new( + EventLog::open( + EventLogConfig { + segments_dir: dir.join("eventlog/segments"), + ..EventLogConfig::default() + }, + std::sync::Arc::clone(sim), + ) + .unwrap(), + ) +} + +fn run_eventlog_commit_loop_under_faults(seed: u64) { + let dir = tempfile::TempDir::new().unwrap(); + std::fs::create_dir_all(dir.path().join("eventlog/segments")).unwrap(); + let fault = sim_fault_for(seed); + let sim = std::sync::Arc::new(SimulatedIO::new(seed, fault)); + let did = test_did(seed); + let event_count = (seed % 40) + 20; + + let mut acked_sync_seq: u64 = 0; + { + sim.set_pristine_mode(true); + let eventlog = open_fault_eventlog(dir.path(), &sim); + sim.set_pristine_mode(false); + + (0..event_count).for_each(|i| { + if eventlog + .append_event(&did, RepoEventType::Commit, &make_commit_event(&did, i)) + .is_ok() + && let Ok(result) = eventlog.sync() + { + acked_sync_seq = acked_sync_seq.max(result.synced_through.raw()); + } + }); + + sim.crash(); + drop(eventlog); + } + + sim.set_pristine_mode(true); + let reopened = open_fault_eventlog(dir.path(), &sim); + let recovered = reopened.max_seq().raw(); + assert!( + recovered >= acked_sync_seq, + "seed={seed} fault={fault:?}: events acknowledged synced through {acked_sync_seq} must survive crash, recovered max_seq={recovered}" + ); + assert!( + recovered <= event_count, + "seed={seed} fault={fault:?}: recovery invented events beyond the {event_count} appended, recovered max_seq={recovered}" + ); + let _ = reopened.shutdown(); +} + +#[test] +fn sim_eventlog_commit_loop_durability_under_faults() { + sim_seed_range().into_par_iter().for_each(|seed| { + run_eventlog_commit_loop_under_faults(seed); + }); +} + fn open_sim_blockstore( dir: &std::path::Path, sim: &std::sync::Arc,