Fork port: connection architecture + watcher machinery into the substrate (ADR-011/012 §3–§5, task fork-port-connection-watcher)
Kept half of the honker-core fork lands in alkstore-sqlite/src/substrate/ as schema.rs / watcher.rs / ops.rs (register D-17): PRAGMA/WAL open posture + set_journal_mode_wal retry, Writer, Readers, the polling watcher family (SharedUpdateWatcher, WatcherDeathGuard, stat_identity dead-man's switch), in_savepoint/ UnwindUndo mutation discipline, REAL-coercion arg helpers, notify scalar + notifications table with the ADR-010 §6 at-attach pruning cap, stream functions, lock functions. Port deltas (ADR-012 §4): W-1 bounded reconnect backoff (MAX_RECONNECT_TICKS=100), W-2 fallible watcher spawn (Result; engine maps to Database at open in wave 3), dead-man's-switch panic replaced by log-and-exit through the ordinary death path — death still closes every subscriber (pinned by test, join now Ok). Table family _honker_* -> __alkstore_* (D-10); duplicate-column race swallow re-keyed to pragma_table_info (D-11); scheduler cron_expr -> spec; fresh-only bootstrap, append-column migrations kept, column-order equality pinned. Drops confirmed absent: cron, kernel/shm backends, rate-limit/result tables, superseded queue functions (D-01..D-04). file-id retained for the kept dead-man's switch (D-18). 43 engine-crate tests green (adapted inherited suites + delta tests + cross-mechanism pressure); cargo build/clippy -D warnings/fmt clean. PROVENANCE.md register updated to the landed state (D-01..D-20).
This commit is contained in:
glm-5.3-flash committed
1 parent
2d855b9546
commit
43a135c453
8 files changed
+2751
-30
No files matched your search
@@ -0,0 +1,888 @@
|
||||
//! The polling update watcher — ported from honker-core `lib.rs`
|
||||
//! @ `f4e53c6` (upstream names preserved). Port deltas (ADR-012 §4):
|
||||
//! W-1 bounded reconnect backoff in the reconnect path; W-2 fallible
|
||||
//! spawn (`Result`); the dead-man's switch's db-file-identity-change
|
||||
//! panic replaced by a deliberate watcher-fatal death that logs the
|
||||
//! precise diagnostic and exits through the ordinary death path —
|
||||
//! `WatcherDeathGuard` still closes every subscriber on death.
|
||||
//! The experimental backend machinery (`kernel_watcher`, `shm_watcher`,
|
||||
//! their deps) is dropped — the polling backend is the only backend.
|
||||
|
||||
use parking_lot::Mutex;
|
||||
use rusqlite::{Connection, OpenFlags, ffi};
|
||||
use std::collections::HashMap;
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
|
||||
use std::sync::mpsc::{SyncSender, TrySendError};
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
pub(crate) const DEFAULT_WATCHER_POLL_INTERVAL: Duration = Duration::from_millis(1);
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub(crate) struct WatcherConfig {
|
||||
pub(crate) poll_interval: Duration,
|
||||
}
|
||||
|
||||
impl Default for WatcherConfig {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
poll_interval: DEFAULT_WATCHER_POLL_INTERVAL,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl WatcherConfig {
|
||||
pub(crate) fn with_poll_interval(mut self, poll_interval: Duration) -> Result<Self, String> {
|
||||
if poll_interval.is_zero() {
|
||||
return Err("watcher poll interval must be positive".to_string());
|
||||
}
|
||||
self.poll_interval = poll_interval;
|
||||
Ok(self)
|
||||
}
|
||||
}
|
||||
|
||||
/// Platform-specific file identity: `(dev, ino)` on Unix,
|
||||
/// `(volume_serial, file_index)` on Windows. Used to detect when the
|
||||
/// database file has been replaced underneath us (atomic rename,
|
||||
/// litestream restore, volume remount).
|
||||
///
|
||||
/// Uses the `file-id` crate on unix and windows for stable Rust
|
||||
/// support without nightly features. Falls back to `(0, 0)` on other
|
||||
/// targets — on those targets the dead-man's switch is a no-op
|
||||
/// (every `stat_identity` returns `(0, 0)` so the equality check
|
||||
/// never trips); replacement detection is disabled but the watcher
|
||||
/// still functions.
|
||||
#[cfg(any(unix, windows))]
|
||||
pub(crate) fn stat_identity(path: &Path) -> std::io::Result<(u64, u64)> {
|
||||
let id = file_id::get_file_id(path)?;
|
||||
match id {
|
||||
file_id::FileId::Inode {
|
||||
device_id,
|
||||
inode_number,
|
||||
} => Ok((device_id, inode_number)),
|
||||
file_id::FileId::LowRes {
|
||||
volume_serial_number,
|
||||
file_index,
|
||||
} => Ok((u64::from(volume_serial_number), file_index)),
|
||||
file_id::FileId::HighRes {
|
||||
volume_serial_number,
|
||||
file_id,
|
||||
} => Ok(fold_high_res(volume_serial_number, file_id)),
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(any(unix, windows))]
|
||||
fn fold_high_res(volume_serial_number: u64, file_id: u128) -> (u64, u64) {
|
||||
let file_index = ((file_id >> 64) as u64) ^ (file_id as u64);
|
||||
(volume_serial_number, file_index)
|
||||
}
|
||||
|
||||
#[cfg(not(any(unix, windows)))]
|
||||
pub(crate) fn stat_identity(_path: &Path) -> std::io::Result<(u64, u64)> {
|
||||
Ok((0, 0))
|
||||
}
|
||||
|
||||
/// Read the pager's `data_version` counter via `PRAGMA data_version`.
|
||||
/// Returns a u32 incremented on every commit by any connection (and on
|
||||
/// checkpoint).
|
||||
///
|
||||
/// W-3 (register D-09 — recorded, no action): the counter is a u32 and
|
||||
/// wraps after 2³² commits. A wrap reads as a version change, firing
|
||||
/// one spurious wake; wakes are re-read hints, so a spurious wake is
|
||||
/// harmless. No wrap tracking is added.
|
||||
pub(crate) fn poll_data_version(conn: &Connection) -> Result<u32, String> {
|
||||
conn.pragma_query_value(None, "data_version", |row| row.get(0))
|
||||
.map_err(|e| e.to_string())
|
||||
}
|
||||
|
||||
fn is_transient_lock_error(e: &rusqlite::Error) -> bool {
|
||||
matches!(
|
||||
e,
|
||||
rusqlite::Error::SqliteFailure(
|
||||
ffi::Error {
|
||||
code: ffi::ErrorCode::DatabaseBusy | ffi::ErrorCode::DatabaseLocked,
|
||||
..
|
||||
},
|
||||
_,
|
||||
)
|
||||
)
|
||||
}
|
||||
|
||||
/// Upper bound (in poll ticks) between reconnect attempts when the
|
||||
/// watcher's connection is down (W-1 — ADR-012 §4).
|
||||
const MAX_RECONNECT_TICKS: u64 = 100;
|
||||
|
||||
/// Open flags shared by the watcher's initial open and its reconnect
|
||||
/// opens; test instrumentation hooks the loop's opens here.
|
||||
type ReconnectOpenFn = fn(&Path) -> rusqlite::Result<Connection>;
|
||||
|
||||
fn open_watcher_conn(path: &Path) -> rusqlite::Result<Connection> {
|
||||
Connection::open_with_flags(
|
||||
path,
|
||||
OpenFlags::SQLITE_OPEN_READ_WRITE | OpenFlags::SQLITE_OPEN_NO_MUTEX,
|
||||
)
|
||||
}
|
||||
|
||||
/// Polling loop body shared by [`UpdateWatcher`].
|
||||
///
|
||||
/// Three layers:
|
||||
///
|
||||
/// 1. **Fast path:** `PRAGMA data_version`. Compare the
|
||||
/// integer to last seen value. Notify on change.
|
||||
/// 2. **Error recovery:** If the query fails, reconnect the SQLite
|
||||
/// connection and force one wake — with bounded backoff so a
|
||||
/// vanished db file produces bounded open attempts (W-1), not the
|
||||
/// lineage's one attempt per millisecond.
|
||||
/// 3. **Identity check (about every 100 ms):** `stat(db_path)` to
|
||||
/// compare `(dev, ino)`. If the file was replaced, log the precise
|
||||
/// diagnostic and exit through the ordinary death path (the fork's
|
||||
/// dead-man's-switch delta — the lineage panicked here); continuing
|
||||
/// would silently watch stale data. `WatcherDeathGuard` closes
|
||||
/// every subscriber exactly as it does for any watcher death.
|
||||
pub(crate) fn run_poll_loop<F>(
|
||||
db_path: PathBuf,
|
||||
on_change: F,
|
||||
stop: Arc<AtomicBool>,
|
||||
ready: std::sync::mpsc::SyncSender<()>,
|
||||
poll_interval: Duration,
|
||||
open_conn_fn: ReconnectOpenFn,
|
||||
) where
|
||||
F: Fn(),
|
||||
{
|
||||
let mut conn = match open_conn_fn(&db_path) {
|
||||
Ok(c) => Some(c),
|
||||
Err(e) => {
|
||||
eprintln!("alkstore: failed to open watcher connection: {e}");
|
||||
None
|
||||
}
|
||||
};
|
||||
let mut last_version = conn
|
||||
.as_ref()
|
||||
.and_then(|c| poll_data_version(c).ok())
|
||||
.unwrap_or(0);
|
||||
let initial_identity = match stat_identity(&db_path) {
|
||||
Ok(id) => id,
|
||||
Err(e) => {
|
||||
eprintln!("alkstore: failed to stat database for identity check: {e}");
|
||||
(0, 0)
|
||||
}
|
||||
};
|
||||
let mut next_identity_check = Instant::now() + UPDATE_WATCHER_IDENTITY_INTERVAL;
|
||||
let mut reconnect_ticks: u64 = MAX_RECONNECT_TICKS;
|
||||
let _ = ready.send(());
|
||||
drop(ready);
|
||||
|
||||
while !stop.load(Ordering::Acquire) {
|
||||
std::thread::sleep(poll_interval);
|
||||
|
||||
if let Some(ref c) = conn {
|
||||
match c.pragma_query_value(None, "data_version", |row| row.get::<_, u32>(0)) {
|
||||
Ok(version) => {
|
||||
if version != last_version {
|
||||
last_version = version;
|
||||
on_change();
|
||||
}
|
||||
}
|
||||
Err(e) if is_transient_lock_error(&e) => {}
|
||||
Err(e) => {
|
||||
eprintln!("alkstore: data_version poll failed: {e}");
|
||||
conn = None;
|
||||
on_change();
|
||||
}
|
||||
}
|
||||
} else {
|
||||
reconnect_ticks = reconnect_ticks.saturating_add(1);
|
||||
if reconnect_ticks >= MAX_RECONNECT_TICKS {
|
||||
reconnect_ticks = 0;
|
||||
match open_conn_fn(&db_path) {
|
||||
Ok(c) => {
|
||||
last_version = poll_data_version(&c).unwrap_or(0);
|
||||
conn = Some(c);
|
||||
}
|
||||
Err(e) => {
|
||||
eprintln!("alkstore: reconnect failed: {e}");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let now = Instant::now();
|
||||
if now >= next_identity_check {
|
||||
next_identity_check = now + UPDATE_WATCHER_IDENTITY_INTERVAL;
|
||||
match stat_identity(&db_path) {
|
||||
Ok(current) => {
|
||||
if current != initial_identity {
|
||||
eprintln!(
|
||||
"alkstore: database file replaced: \
|
||||
expected (dev={}, ino={}), \
|
||||
found (dev={}, ino={}) at {:?}. \
|
||||
The watcher cannot recover; it has stopped and \
|
||||
closed all subscribers; close the store and reopen.",
|
||||
initial_identity.0, initial_identity.1, current.0, current.1, db_path
|
||||
);
|
||||
return;
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
eprintln!("alkstore: stat identity check failed: {e}");
|
||||
conn = None;
|
||||
on_change();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) struct UpdateWatcher {
|
||||
stop: Arc<AtomicBool>,
|
||||
handle: Option<std::thread::JoinHandle<()>>,
|
||||
}
|
||||
|
||||
const UPDATE_WATCHER_IDENTITY_INTERVAL: Duration = Duration::from_millis(100);
|
||||
|
||||
impl UpdateWatcher {
|
||||
/// Spawn a watcher thread on `db_path`. `on_change` is called once
|
||||
/// per observed commit. The thread runs until [`UpdateWatcher`] is
|
||||
/// dropped or [`stop`](Self::stop) is called.
|
||||
///
|
||||
/// Fallible (W-2 — ADR-012 §4): errors if the thread cannot be
|
||||
/// spawned; the engine surfaces the failure at store-open time.
|
||||
/// Baseline-capture failures inside the thread (open failures, a
|
||||
/// vanished db file) leave the sender dropped and the loop itself
|
||||
/// retrying per W-1's bounded backoff — not a spawn failure.
|
||||
pub(crate) fn spawn<F>(db_path: PathBuf, on_change: F) -> Result<Self, String>
|
||||
where
|
||||
F: Fn() + Send + 'static,
|
||||
{
|
||||
Self::spawn_with_config(db_path, on_change, WatcherConfig::default())
|
||||
}
|
||||
|
||||
/// Like [`spawn`](Self::spawn) but with an explicit poll cadence.
|
||||
pub(crate) fn spawn_with_config<F>(
|
||||
db_path: PathBuf,
|
||||
on_change: F,
|
||||
config: WatcherConfig,
|
||||
) -> Result<Self, String>
|
||||
where
|
||||
F: Fn() + Send + 'static,
|
||||
{
|
||||
let stop = Arc::new(AtomicBool::new(false));
|
||||
let stop_t = stop.clone();
|
||||
let (ready_tx, ready_rx) = std::sync::mpsc::sync_channel::<()>(1);
|
||||
let handle = Self::spawn_thread(
|
||||
db_path,
|
||||
on_change,
|
||||
stop_t,
|
||||
ready_tx,
|
||||
config.poll_interval,
|
||||
None,
|
||||
)?;
|
||||
let _ = ready_rx.recv();
|
||||
Ok(Self {
|
||||
stop,
|
||||
handle: Some(handle),
|
||||
})
|
||||
}
|
||||
|
||||
/// The thread-spawn step, isolated so the fallible boundary (W-2)
|
||||
/// has one owned site. `stack_size` overrides the std default;
|
||||
/// production passes None — tests force a real spawn failure
|
||||
/// through it.
|
||||
fn spawn_thread<F>(
|
||||
db_path: PathBuf,
|
||||
on_change: F,
|
||||
stop_t: Arc<AtomicBool>,
|
||||
ready_tx: std::sync::mpsc::SyncSender<()>,
|
||||
poll_interval: Duration,
|
||||
stack_size: Option<usize>,
|
||||
) -> Result<std::thread::JoinHandle<()>, String>
|
||||
where
|
||||
F: Fn() + Send + 'static,
|
||||
{
|
||||
let builder = std::thread::Builder::new().name("alkstore-update-poll".into());
|
||||
let builder = match stack_size {
|
||||
Some(sz) => builder.stack_size(sz),
|
||||
None => builder,
|
||||
};
|
||||
builder
|
||||
.spawn(move || {
|
||||
run_poll_loop(
|
||||
db_path,
|
||||
on_change,
|
||||
stop_t,
|
||||
ready_tx,
|
||||
poll_interval,
|
||||
open_watcher_conn,
|
||||
)
|
||||
})
|
||||
.map_err(|e| format!("failed to spawn alkstore update-watcher thread: {e}"))
|
||||
}
|
||||
|
||||
pub(crate) fn stop(&self) {
|
||||
self.stop.store(true, Ordering::Release);
|
||||
}
|
||||
|
||||
/// Stop the watcher and wait for the thread to exit.
|
||||
pub(crate) fn join(mut self) -> std::thread::Result<()> {
|
||||
self.stop();
|
||||
match self.handle.take() {
|
||||
Some(h) => h.join(),
|
||||
None => Ok(()),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for UpdateWatcher {
|
||||
fn drop(&mut self) {
|
||||
self.stop();
|
||||
}
|
||||
}
|
||||
|
||||
struct WatcherDeathGuard {
|
||||
senders: Arc<Mutex<HashMap<u64, SyncSender<()>>>>,
|
||||
}
|
||||
|
||||
impl Drop for WatcherDeathGuard {
|
||||
fn drop(&mut self) {
|
||||
self.senders.lock().clear();
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) struct SharedUpdateWatcher {
|
||||
watcher: Mutex<Option<UpdateWatcher>>,
|
||||
senders: Arc<Mutex<HashMap<u64, SyncSender<()>>>>,
|
||||
next_id: AtomicU64,
|
||||
}
|
||||
|
||||
impl SharedUpdateWatcher {
|
||||
/// Spawn the shared poll thread for `db_path`.
|
||||
///
|
||||
/// Fallible (W-2 — ADR-012 §4): the engine surfaces the failure at
|
||||
/// store-open time.
|
||||
pub(crate) fn new(db_path: PathBuf) -> Result<Self, String> {
|
||||
Self::new_with_config(db_path, WatcherConfig::default())
|
||||
}
|
||||
|
||||
/// Like [`new`](Self::new) but with an explicit poll cadence.
|
||||
pub(crate) fn new_with_config(db_path: PathBuf, config: WatcherConfig) -> Result<Self, String> {
|
||||
let senders: Arc<Mutex<HashMap<u64, SyncSender<()>>>> =
|
||||
Arc::new(Mutex::new(HashMap::new()));
|
||||
let senders_t = senders.clone();
|
||||
let death_guard = WatcherDeathGuard {
|
||||
senders: senders.clone(),
|
||||
};
|
||||
let watcher = UpdateWatcher::spawn_with_config(
|
||||
db_path,
|
||||
move || {
|
||||
let _ = &death_guard;
|
||||
let mut list = senders_t.lock();
|
||||
list.retain(|_id, s| match s.try_send(()) {
|
||||
Ok(()) | Err(TrySendError::Full(_)) => true,
|
||||
Err(TrySendError::Disconnected(_)) => false,
|
||||
});
|
||||
},
|
||||
config,
|
||||
)?;
|
||||
Ok(Self {
|
||||
watcher: Mutex::new(Some(watcher)),
|
||||
senders,
|
||||
next_id: AtomicU64::new(0),
|
||||
})
|
||||
}
|
||||
|
||||
/// Subscribe. Returns a subscriber id and a [`Receiver<()>`] that
|
||||
/// sees one tick per observed database update. Callers MUST
|
||||
/// [`unsubscribe`](Self::unsubscribe) the returned id when done —
|
||||
/// otherwise the sender stays in the map and a bridge thread
|
||||
/// blocking on `recv()` will never see a disconnect.
|
||||
///
|
||||
/// Channel capacity is 1: bursts coalesce into one wake per drain
|
||||
/// cycle. Wakes are "go re-read state" signals.
|
||||
pub(crate) fn subscribe(&self) -> (u64, std::sync::mpsc::Receiver<()>) {
|
||||
let id = self.next_id.fetch_add(1, Ordering::Relaxed);
|
||||
let (tx, rx) = std::sync::mpsc::sync_channel(1);
|
||||
self.senders.lock().insert(id, tx);
|
||||
(id, rx)
|
||||
}
|
||||
|
||||
/// Remove a subscriber. The corresponding receiver sees
|
||||
/// `Err(RecvError)` on its next blocking `recv()`, letting a
|
||||
/// bridge thread exit cleanly.
|
||||
pub(crate) fn unsubscribe(&self, id: u64) {
|
||||
self.senders.lock().remove(&id);
|
||||
}
|
||||
|
||||
pub(crate) fn subscriber_count(&self) -> usize {
|
||||
self.senders.lock().len()
|
||||
}
|
||||
|
||||
/// Disconnect all subscribers and synchronously join the poll
|
||||
/// thread. The thread owns the watcher's connection; joining drops
|
||||
/// that connection and releases the file handle. Idempotent — safe
|
||||
/// to call more than once.
|
||||
pub(crate) fn close(&self) -> std::thread::Result<()> {
|
||||
self.senders.lock().clear();
|
||||
match self.watcher.lock().take() {
|
||||
Some(watcher) => watcher.join(),
|
||||
None => Ok(()),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for SharedUpdateWatcher {
|
||||
fn drop(&mut self) {
|
||||
self.senders.lock().clear();
|
||||
if let Some(watcher) = self.watcher.get_mut().take() {
|
||||
drop(watcher);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod watcher_tests {
|
||||
use super::*;
|
||||
|
||||
fn temp_db(name: &str) -> PathBuf {
|
||||
let p = std::env::temp_dir().join(format!(
|
||||
"alkstore-{name}-{}-{:?}.db",
|
||||
std::process::id(),
|
||||
std::thread::current().id()
|
||||
));
|
||||
let _ = std::fs::remove_file(&p);
|
||||
let _ = std::fs::remove_file(format!("{}-wal", p.display()));
|
||||
let _ = std::fs::remove_file(format!("{}-shm", p.display()));
|
||||
p
|
||||
}
|
||||
|
||||
fn wal_db(path: &Path) {
|
||||
let conn = Connection::open(path).unwrap();
|
||||
conn.execute_batch("PRAGMA journal_mode = WAL;").unwrap();
|
||||
}
|
||||
|
||||
/// W-2: a thread-spawn failure surfaces as `Err` from the spawn
|
||||
/// path — rather than the lineage's `expect` panic — with the
|
||||
/// mapped error naming the watcher. Driven through the real spawn
|
||||
/// site with an oversized stack reservation, which reliably fails
|
||||
/// on every target.
|
||||
#[test]
|
||||
fn spawn_fails_when_the_thread_cannot_start() {
|
||||
let (tx, _rx) = std::sync::mpsc::sync_channel::<()>(1);
|
||||
let err = UpdateWatcher::spawn_thread(
|
||||
PathBuf::from("/nonexistent/alkstore-w2.db"),
|
||||
|| {},
|
||||
Arc::new(AtomicBool::new(false)),
|
||||
tx,
|
||||
Duration::from_millis(1),
|
||||
Some(usize::MAX),
|
||||
)
|
||||
.expect_err("an oversized stack reservation must fail to spawn");
|
||||
assert!(
|
||||
err.contains("failed to spawn alkstore update-watcher thread"),
|
||||
"the W-2 error must name the watcher spawn: {err}"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn watcher_config_rejects_zero_poll_interval() {
|
||||
let err = WatcherConfig::default()
|
||||
.with_poll_interval(Duration::from_millis(0))
|
||||
.unwrap_err();
|
||||
assert_eq!(err, "watcher poll interval must be positive");
|
||||
assert!(
|
||||
WatcherConfig::default()
|
||||
.with_poll_interval(Duration::from_millis(5))
|
||||
.is_ok()
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn shared_update_watcher_fans_out_to_many_subscribers() {
|
||||
let tmp = temp_db("shared-fanout");
|
||||
wal_db(&tmp);
|
||||
|
||||
let shared = SharedUpdateWatcher::new(tmp.clone()).unwrap();
|
||||
let subs: Vec<(u64, std::sync::mpsc::Receiver<()>)> =
|
||||
(0..50).map(|_| shared.subscribe()).collect();
|
||||
|
||||
let writer = Connection::open(&tmp).unwrap();
|
||||
writer
|
||||
.execute(
|
||||
"CREATE TABLE IF NOT EXISTS _test_trigger(id INTEGER PRIMARY KEY)",
|
||||
[],
|
||||
)
|
||||
.unwrap();
|
||||
for i in 0..5 {
|
||||
std::thread::sleep(Duration::from_millis(5));
|
||||
writer
|
||||
.execute("INSERT INTO _test_trigger(id) VALUES (?)", [i])
|
||||
.unwrap();
|
||||
}
|
||||
std::thread::sleep(Duration::from_millis(50));
|
||||
|
||||
for (i, (_id, rx)) in subs.iter().enumerate() {
|
||||
let mut got_any = false;
|
||||
while rx.try_recv().is_ok() {
|
||||
got_any = true;
|
||||
}
|
||||
assert!(got_any, "subscriber {i} saw no ticks");
|
||||
}
|
||||
|
||||
shared.close().unwrap();
|
||||
let _ = std::fs::remove_file(&tmp);
|
||||
let _ = std::fs::remove_file(format!("{}-wal", tmp.display()));
|
||||
let _ = std::fs::remove_file(format!("{}-shm", tmp.display()));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn shared_update_watcher_explicit_unsubscribe_disconnects_receiver() {
|
||||
let tmp = temp_db("unsub");
|
||||
wal_db(&tmp);
|
||||
|
||||
let shared = SharedUpdateWatcher::new(tmp.clone()).unwrap();
|
||||
let (id, rx) = shared.subscribe();
|
||||
assert_eq!(shared.subscriber_count(), 1);
|
||||
|
||||
shared.unsubscribe(id);
|
||||
assert_eq!(shared.subscriber_count(), 0);
|
||||
|
||||
assert!(rx.recv().is_err());
|
||||
|
||||
shared.close().unwrap();
|
||||
let _ = std::fs::remove_file(&tmp);
|
||||
let _ = std::fs::remove_file(format!("{}-wal", tmp.display()));
|
||||
let _ = std::fs::remove_file(format!("{}-shm", tmp.display()));
|
||||
}
|
||||
|
||||
/// Subscribers must learn that the watcher thread died — not just
|
||||
/// stop receiving wakes silently. We force the watcher's death
|
||||
/// via the dead-man's switch (file replacement) and assert that an
|
||||
/// already-subscribed receiver returns `Err(RecvError)` on its
|
||||
/// next blocking `recv()`. Without WatcherDeathGuard this test
|
||||
/// hangs (subscriber blocks forever) and times out.
|
||||
///
|
||||
/// Also asserts the port's dead-man's-switch delta (ADR-012 §4):
|
||||
/// the thread exits *without panicking* — `join()` is `Ok(())`
|
||||
/// where the lineage's `join()` returned the panic payload.
|
||||
#[test]
|
||||
#[cfg(unix)]
|
||||
fn shared_update_watcher_signals_subscribers_on_watcher_death() {
|
||||
let tmp = temp_db("death-signal");
|
||||
wal_db(&tmp);
|
||||
|
||||
let shared = SharedUpdateWatcher::new(tmp.clone()).unwrap();
|
||||
let (_id, rx) = shared.subscribe();
|
||||
|
||||
// Let the watcher snapshot the initial identity.
|
||||
std::thread::sleep(Duration::from_millis(200));
|
||||
|
||||
let other = temp_db("death-other");
|
||||
let _ = std::fs::File::create(&other).unwrap();
|
||||
std::fs::rename(&other, &tmp).unwrap();
|
||||
|
||||
let deadline = std::time::Instant::now() + Duration::from_secs(2);
|
||||
loop {
|
||||
match rx.recv_timeout(Duration::from_millis(100)) {
|
||||
Ok(()) => continue,
|
||||
Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => break,
|
||||
Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {}
|
||||
}
|
||||
assert!(
|
||||
std::time::Instant::now() < deadline,
|
||||
"watcher died but subscriber's channel never disconnected — \
|
||||
WatcherDeathGuard didn't fire?"
|
||||
);
|
||||
}
|
||||
|
||||
shared.close().unwrap();
|
||||
let _ = std::fs::remove_file(&tmp);
|
||||
let _ = std::fs::remove_file(format!("{}-wal", tmp.display()));
|
||||
let _ = std::fs::remove_file(format!("{}-shm", tmp.display()));
|
||||
}
|
||||
|
||||
/// The port's dead-man's-switch delta observed directly at the
|
||||
/// watcher: a db-file replacement produces a clean (non-panic)
|
||||
/// thread exit. The lineage panicked here; ADR-012 §4 replaces
|
||||
/// the unwind with an ordinary death.
|
||||
#[test]
|
||||
#[cfg(unix)]
|
||||
fn update_watcher_dies_cleanly_on_file_replacement() {
|
||||
let tmp = temp_db("watcher-replace");
|
||||
wal_db(&tmp);
|
||||
|
||||
let watcher = UpdateWatcher::spawn(tmp.clone(), || {}).unwrap();
|
||||
|
||||
std::thread::sleep(Duration::from_millis(200));
|
||||
|
||||
let tmp2 = temp_db("watcher-replace-new");
|
||||
wal_db(&tmp2);
|
||||
std::fs::rename(&tmp2, &tmp).unwrap();
|
||||
|
||||
std::thread::sleep(Duration::from_millis(500));
|
||||
|
||||
let result = watcher.join();
|
||||
assert!(
|
||||
result.is_ok(),
|
||||
"watcher should have exited cleanly on file replacement \
|
||||
(dead-man's switch must not panic, ADR-012 §4), got {result:?}"
|
||||
);
|
||||
|
||||
let _ = std::fs::remove_file(&tmp);
|
||||
let _ = std::fs::remove_file(format!("{}-wal", tmp.display()));
|
||||
let _ = std::fs::remove_file(format!("{}-shm", tmp.display()));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn shared_update_watcher_prunes_subscribers_when_receiver_dropped() {
|
||||
let tmp = temp_db("prune");
|
||||
wal_db(&tmp);
|
||||
|
||||
let shared = SharedUpdateWatcher::new(tmp.clone()).unwrap();
|
||||
{
|
||||
let _subs: Vec<_> = (0..10).map(|_| shared.subscribe()).collect();
|
||||
assert_eq!(shared.subscriber_count(), 10);
|
||||
}
|
||||
|
||||
let writer = Connection::open(&tmp).unwrap();
|
||||
writer
|
||||
.execute(
|
||||
"CREATE TABLE IF NOT EXISTS _test_prune(id INTEGER PRIMARY KEY)",
|
||||
[],
|
||||
)
|
||||
.unwrap();
|
||||
let deadline = std::time::Instant::now() + Duration::from_secs(2);
|
||||
while shared.subscriber_count() != 0 && std::time::Instant::now() < deadline {
|
||||
std::thread::sleep(Duration::from_millis(5));
|
||||
writer
|
||||
.execute("INSERT INTO _test_prune(id) VALUES (random())", [])
|
||||
.unwrap();
|
||||
}
|
||||
assert_eq!(shared.subscriber_count(), 0);
|
||||
|
||||
shared.close().unwrap();
|
||||
let _ = std::fs::remove_file(&tmp);
|
||||
let _ = std::fs::remove_file(format!("{}-wal", tmp.display()));
|
||||
let _ = std::fs::remove_file(format!("{}-shm", tmp.display()));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn data_version_detects_commits_and_ignores_rollbacks() {
|
||||
let tmp = temp_db("dv");
|
||||
let watcher = Connection::open(&tmp).unwrap();
|
||||
watcher.execute_batch("PRAGMA journal_mode = WAL;").unwrap();
|
||||
let writer = Connection::open(&tmp).unwrap();
|
||||
|
||||
let v0 = poll_data_version(&watcher).unwrap();
|
||||
|
||||
writer.execute("CREATE TABLE t(x INTEGER)", []).unwrap();
|
||||
let v1 = poll_data_version(&watcher).unwrap();
|
||||
assert!(v1 > v0, "commit should increment data_version");
|
||||
|
||||
writer.execute_batch("BEGIN IMMEDIATE;").unwrap();
|
||||
writer.execute("INSERT INTO t VALUES (1)", []).unwrap();
|
||||
writer.execute_batch("ROLLBACK;").unwrap();
|
||||
let v2 = poll_data_version(&watcher).unwrap();
|
||||
assert_eq!(v2, v1, "rollback should not increment data_version");
|
||||
|
||||
let _ = std::fs::remove_file(&tmp);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn data_version_survives_wal_checkpoint() {
|
||||
let tmp = temp_db("dv-ckpt");
|
||||
let watcher = Connection::open(&tmp).unwrap();
|
||||
watcher.execute_batch("PRAGMA journal_mode = WAL;").unwrap();
|
||||
let w0 = poll_data_version(&watcher).unwrap();
|
||||
|
||||
let writer = Connection::open(&tmp).unwrap();
|
||||
writer.execute("CREATE TABLE t(x INTEGER)", []).unwrap();
|
||||
let w1 = poll_data_version(&watcher).unwrap();
|
||||
assert!(
|
||||
w1 > w0,
|
||||
"commit from other conn should increment data_version"
|
||||
);
|
||||
|
||||
writer
|
||||
.execute_batch("PRAGMA wal_checkpoint(TRUNCATE);")
|
||||
.unwrap();
|
||||
let w2 = poll_data_version(&watcher).unwrap();
|
||||
assert!(
|
||||
w2 > w1,
|
||||
"checkpoint from other conn should increment data_version"
|
||||
);
|
||||
|
||||
writer.execute("INSERT INTO t VALUES (1)", []).unwrap();
|
||||
let w3 = poll_data_version(&watcher).unwrap();
|
||||
assert!(
|
||||
w3 > w2,
|
||||
"post-checkpoint commit should increment data_version"
|
||||
);
|
||||
|
||||
let _ = std::fs::remove_file(&tmp);
|
||||
}
|
||||
|
||||
fn poll_data_version_works_in_journal_mode(mode: &str) {
|
||||
let tmp = temp_db(&format!("jm-{}", mode.to_ascii_lowercase()));
|
||||
let watcher = Connection::open(&tmp).unwrap();
|
||||
watcher
|
||||
.execute_batch(&format!("PRAGMA journal_mode = {mode};"))
|
||||
.unwrap();
|
||||
let actual: String = watcher
|
||||
.pragma_query_value(None, "journal_mode", |r| r.get(0))
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
actual.to_ascii_uppercase(),
|
||||
mode.to_ascii_uppercase(),
|
||||
"PRAGMA journal_mode = {mode} silently fell back to {actual}"
|
||||
);
|
||||
|
||||
let writer = Connection::open(&tmp).unwrap();
|
||||
let v0 = poll_data_version(&watcher).unwrap();
|
||||
|
||||
writer.execute("CREATE TABLE t(x INTEGER)", []).unwrap();
|
||||
let v1 = poll_data_version(&watcher).unwrap();
|
||||
assert!(
|
||||
v1 > v0,
|
||||
"journal_mode={mode}: cross-conn commit should bump \
|
||||
data_version; saw {v0} -> {v1}"
|
||||
);
|
||||
|
||||
writer.execute_batch("BEGIN IMMEDIATE;").unwrap();
|
||||
writer.execute("INSERT INTO t VALUES (1)", []).unwrap();
|
||||
writer.execute_batch("ROLLBACK;").unwrap();
|
||||
let v2 = poll_data_version(&watcher).unwrap();
|
||||
assert_eq!(
|
||||
v2, v1,
|
||||
"journal_mode={mode}: rollback should not bump data_version"
|
||||
);
|
||||
|
||||
let _ = std::fs::remove_file(&tmp);
|
||||
let _ = std::fs::remove_file(format!("{}-wal", tmp.display()));
|
||||
let _ = std::fs::remove_file(format!("{}-shm", tmp.display()));
|
||||
let _ = std::fs::remove_file(format!("{}-journal", tmp.display()));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn poll_data_version_works_in_wal() {
|
||||
poll_data_version_works_in_journal_mode("WAL");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn poll_data_version_works_in_delete() {
|
||||
poll_data_version_works_in_journal_mode("DELETE");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn poll_data_version_works_in_truncate() {
|
||||
poll_data_version_works_in_journal_mode("TRUNCATE");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn poll_data_version_works_in_persist() {
|
||||
poll_data_version_works_in_journal_mode("PERSIST");
|
||||
}
|
||||
|
||||
#[cfg(any(unix, windows))]
|
||||
#[test]
|
||||
fn stat_identity_detects_file_replacement() {
|
||||
let tmp = temp_db("id-test");
|
||||
let tmp2 = temp_db("id-test2");
|
||||
|
||||
std::fs::write(&tmp, b"original").unwrap();
|
||||
std::fs::write(&tmp2, b"replacement").unwrap();
|
||||
|
||||
let id1 = stat_identity(&tmp).unwrap();
|
||||
let id2 = stat_identity(&tmp2).unwrap();
|
||||
assert_ne!(id1, id2, "different files should have different identities");
|
||||
|
||||
std::fs::rename(&tmp2, &tmp).unwrap();
|
||||
let id3 = stat_identity(&tmp).unwrap();
|
||||
assert_eq!(
|
||||
id3, id2,
|
||||
"renamed file should carry the replacement's identity"
|
||||
);
|
||||
|
||||
let _ = std::fs::remove_file(&tmp);
|
||||
}
|
||||
|
||||
#[cfg(any(unix, windows))]
|
||||
#[test]
|
||||
fn fold_high_res_uses_both_halves() {
|
||||
let (vsn, idx) = fold_high_res(0xAABB, 0x0000_0000_0000_0000_DEAD_BEEF_CAFE_F00D);
|
||||
assert_eq!(vsn, 0xAABB);
|
||||
assert_eq!(idx, 0xDEAD_BEEF_CAFE_F00D);
|
||||
|
||||
let upper = 0x1111_2222_3333_4444u64;
|
||||
let lower = 0x5555_6666_7777_8888u64;
|
||||
let file_id = ((upper as u128) << 64) | (lower as u128);
|
||||
let (vsn, idx) = fold_high_res(0xCCDD, file_id);
|
||||
assert_eq!(vsn, 0xCCDD);
|
||||
assert_eq!(idx, upper ^ lower);
|
||||
|
||||
let same = 0xDEAD_BEEF_CAFE_F00Du64;
|
||||
let file_id = ((same as u128) << 64) | (same as u128);
|
||||
let (_, idx) = fold_high_res(0, file_id);
|
||||
assert_eq!(idx, 0);
|
||||
}
|
||||
|
||||
/// W-1: with the watcher connection down and the db file vanished,
|
||||
/// the reconnect path is throttled — not one open attempt per
|
||||
/// poll tick. Driven directly: a missing db path fails every open,
|
||||
/// the loop runs its reconnect schedule against it, and the
|
||||
/// instrumented open counter counts attempts.
|
||||
#[test]
|
||||
fn reconnect_attempts_are_bounded_by_backoff() {
|
||||
static ATTEMPTS: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
|
||||
fn counting_open(path: &Path) -> rusqlite::Result<Connection> {
|
||||
ATTEMPTS.fetch_add(1, Ordering::Relaxed);
|
||||
open_watcher_conn(path)
|
||||
}
|
||||
|
||||
let tick_ms = 10u64;
|
||||
const WINDOW_MS: u64 = 2_600;
|
||||
let ticks_in_window = WINDOW_MS / tick_ms;
|
||||
let db_path = PathBuf::from(format!(
|
||||
"/this/parent/does/not/exist/alkstore-w1-{}.db",
|
||||
std::process::id()
|
||||
));
|
||||
let stop = Arc::new(AtomicBool::new(false));
|
||||
let stop_t = stop.clone();
|
||||
let (ready_tx, ready_rx) = std::sync::mpsc::sync_channel::<()>(1);
|
||||
let handle = std::thread::spawn(move || {
|
||||
run_poll_loop(
|
||||
db_path,
|
||||
|| {},
|
||||
stop_t,
|
||||
ready_tx,
|
||||
Duration::from_millis(tick_ms),
|
||||
counting_open,
|
||||
)
|
||||
});
|
||||
ready_rx
|
||||
.recv_timeout(Duration::from_secs(2))
|
||||
.expect("watcher loop baseline capture");
|
||||
|
||||
std::thread::sleep(Duration::from_millis(WINDOW_MS));
|
||||
stop.store(true, Ordering::Release);
|
||||
handle.join().unwrap();
|
||||
|
||||
let attempts = ATTEMPTS.load(Ordering::Relaxed);
|
||||
assert!(
|
||||
attempts > 0,
|
||||
"the loop must keep trying to reconnect while the db is missing"
|
||||
);
|
||||
// The backoff bound: at tick=tick_ms, the reconnect schedule
|
||||
// admits ticks_in_window / MAX_RECONNECT_TICKS (+1 in-flight)
|
||||
// attempts in the window, not ticks_in_window (the lineage's
|
||||
// one-per-tick schedule). Scheduler jitter can compress the
|
||||
// effective window slightly; one extra attempt of headroom is
|
||||
// the margin for that, not for a broken backoff — the lineage's
|
||||
// one-per-tick schedule would produce ~260 attempts here.
|
||||
let bound = ticks_in_window / MAX_RECONNECT_TICKS + 2;
|
||||
assert!(
|
||||
attempts <= bound,
|
||||
"W-1 violated: {attempts} reconnect attempts in a {WINDOW_MS}ms \
|
||||
window at {tick_ms}ms/tick; the backoff admits at most {bound}"
|
||||
);
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user