fix(store): unblock eventlog sync&freeze when writer dies

Lewis: May this revision serve well! <lu5a@proton.me>

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<WriterRequest>,
+}
+
+impl Drop for TerminateOnDrop<'_> {
+    fn drop(&mut self) {
+        self.notify.terminate();
+        self.receiver.drain().for_each(fail_abandoned_request);
+    }
+}
+
 fn writer_loop<S: StorageIO>(
     receiver: &flume::Receiver<WriterRequest>,
     writer: &mut EventLogWriter<S>,
     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 {

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<S: StorageIO + 'static> EventLog<S> {
         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<S: StorageIO + 'static> EventLog<S> {
         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<T>(&self, resp_rx: &flume::Receiver<io::Result<T>>) -> io::Result<T> {
+        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<S: StorageIO + 'static> EventLog<S> {
                 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<SegmentId> = all_segments

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<SimulatedIO>,
+) -> std::sync::Arc<EventLog<std::sync::Arc<SimulatedIO>>> {
+    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<SimulatedIO>,