diff --git a/aw-datastore/tests/close_durability.rs b/aw-datastore/tests/close_durability.rs new file mode 100644 index 000000000..af3f0f516 --- /dev/null +++ b/aw-datastore/tests/close_durability.rs @@ -0,0 +1,93 @@ +//! `Datastore::close` racing active writers on the shared worker. +//! +//! aw-tauri shuts down by closing the server's datastore while tray and API +//! clones may still be writing. Every write the worker acknowledged before +//! Close must be durable afterwards, and Close must not be starved by them. +//! Close does not drain in-flight HTTP handlers; this covers worker requests. + +use aw_datastore::Datastore; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{mpsc, Arc, Barrier}; +use std::thread; +use std::time::{Duration, Instant}; + +const WRITERS: usize = 4; +const ROUNDS: usize = 3; + +#[test] +fn close_keeps_every_acknowledged_write_from_racing_writers() { + for round in 0..ROUNDS { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join(format!("close-{round}.db")); + let path = path.to_str().unwrap().to_owned(); + let ds = Datastore::new(path.clone(), false); + ds.set_key_value_and_commit("initial", "ready").unwrap(); + + let barrier = Arc::new(Barrier::new(WRITERS + 1)); + let closing = Arc::new(AtomicBool::new(false)); + let (writer_done_tx, writer_done_rx) = mpsc::channel(); + let writers: Vec<_> = (0..WRITERS) + .map(|id| { + let worker = ds.clone(); + let barrier = barrier.clone(); + let closing = closing.clone(); + let writer_done_tx = writer_done_tx.clone(); + thread::spawn(move || { + let key = format!("writer-{id}"); + // Each writer holds an acknowledged value before Close races it. + worker.set_key_value(&key, "0").unwrap(); + let mut last_ok = 0; + barrier.wait(); + for n in 1..10_000 { + match worker.set_key_value(&key, &n.to_string()) { + Ok(()) => last_ok = n, + Err(_) => { + assert!( + closing.load(Ordering::SeqCst), + "write failed before Close" + ); + break; + } + } + } + writer_done_tx.send(()).unwrap(); + (key, last_ok) + }) + }) + .collect(); + + barrier.wait(); + closing.store(true, Ordering::SeqCst); + let closer = ds.clone(); + let (done_tx, done_rx) = mpsc::channel(); + thread::spawn(move || { + closer.close(); + done_tx.send(()).unwrap(); + }); + done_rx + .recv_timeout(Duration::from_secs(5)) + .expect("active writers starved Close"); + + drop(writer_done_tx); + let deadline = Instant::now() + Duration::from_secs(5); + for _ in 0..WRITERS { + writer_done_rx + .recv_timeout(deadline.saturating_duration_since(Instant::now())) + .expect("writer did not finish after Close"); + } + let acknowledged: Vec<_> = writers.into_iter().map(|w| w.join().unwrap()).collect(); + let reader = Datastore::open_reader(path).unwrap(); + for (key, value) in acknowledged { + assert_eq!( + reader.get_key_value(&key).unwrap(), + value.to_string(), + "round {round}: acknowledged write to {key} lost by Close" + ); + } + reader.close(); + assert!( + ds.get_buckets().is_err(), + "closed worker still served a request" + ); + } +}