From 6632dcce1f387cd5e3b6761ab46a8877f9ffc8c4 Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Sat, 10 Oct 2026 11:27:52 +0000 Subject: [PATCH] =?UTF-8?q?mem-engine-foundations=20=E2=80=94=20the=20mem?= =?UTF-8?q?=20engine's=20foundations=20land=20(alkstore-mem,=20wave=206=20?= =?UTF-8?q?task=201=20of=20the=20ADR-024=20chain):=20fourth=20workspace=20?= =?UTF-8?q?member=20with=20the=20manifest=20surface=20ADR-024=20=C2=A71=20?= =?UTF-8?q?pins=20=E2=80=94=20alkstore=20versioned=20path=20pin=20(ADR-017?= =?UTF-8?q?=20=C2=A74.1)=20plus=20tokio=20restricted=20to=20the=20wasm-sup?= =?UTF-8?q?ported=20subset=20(sync=20+=20time;=20dev-deps=20rt/macros=20fo?= =?UTF-8?q?r=20the=20current-thread-flavor=20tests),=20no=20driver=20deps,?= =?UTF-8?q?=20no=20parking=5Flot=20(the=20description's=20prose=20named=20?= =?UTF-8?q?it,=20but=20the=20ADR=20+=20acceptance=20criterion=20pin=20the?= =?UTF-8?q?=20manifest=20tighter=20=E2=80=94=20short-held=20internal=20gua?= =?UTF-8?q?rds=20are=20std::sync::Mutex=20behind=20a=20state::Guard=20trai?= =?UTF-8?q?t=20that=20folds=20poisoning=20away=20structurally,=20guards=20?= =?UTF-8?q?never=20held=20across=20awaits=20or=20user=20code,=20tokio=20mp?= =?UTF-8?q?sc=20the=20cross-await=20primitive).=20Five=20modules=20re-expo?= =?UTF-8?q?rted=20from=20lib.rs=20whose=20crate=20docs=20carry=20the=20ADR?= =?UTF-8?q?-016=20posture=20statement=20(honest-ephemeral,=20full=20contra?= =?UTF-8?q?ct=20v1,=20wasm32-compile-clean=20identity,=20architecture=20sk?= =?UTF-8?q?etch=20with=20the=20wave-6=20construction=20line).=20MemStore::?= =?UTF-8?q?new()=20is=20the=20engine-native=20isolated=20constructor=20(AD?= =?UTF-8?q?R-024=20=C2=A74's=20documented=20open-prose=20exception):=20fre?= =?UTF-8?q?sh=20guarded=20state=20core=20(state.rs=20=E2=80=94=20empty=20n?= =?UTF-8?q?amed=20per-mechanism=20sections=20QueueState/StreamState/LockSt?= =?UTF-8?q?ate/ScheduleState,=20shapes=20owned=20by=20the=20mechanism=20ta?= =?UTF-8?q?sks),=20fresh=20wake=20bus,=20fresh=20clock=20per=20open;=20tea?= =?UTF-8?q?rdown=20is=20drop=20driving=20WakeBus::close()=20=E2=80=94=20no?= =?UTF-8?q?=20explicit=20close=20form.=20Wake=20substrate=20(wakes.rs):=20?= =?UTF-8?q?WakeBus::attach=20=E2=86=92=20WakeFeed=20(one=20unbounded=20mps?= =?UTF-8?q?c=20feed=20per=20subscriber,=20detach-on-drop,=20attach=20under?= =?UTF-8?q?=20the=20registry=20guard=20so=20attach-after-commit=20no-repla?= =?UTF-8?q?y=20is=20structural),=20fire=5Fcommit=5Fwakes(channels)=20as=20?= =?UTF-8?q?the=20commit-ordered=20wake=20source's=20entry=20point=20(guard?= =?UTF-8?q?=20makes=20watermark=20bump=20+=20per-channel=20fanout=20one=20?= =?UTF-8?q?operation=20=E2=80=94=20per-subscriber=20FIFO=20by=20commit=20o?= =?UTF-8?q?rder,=20never=20coalesced),=20close=20empties=20senders=20so=20?= =?UTF-8?q?engine=20drop=20disconnects=20every=20feed;=20commit=5Fwatermar?= =?UTF-8?q?k=20is=20a=20cfg(test)=20ordering=20observable=20only.=20Clock?= =?UTF-8?q?=20(clock.rs):=20one=20accessor=20Clock::unix=5Fnow=20over=20Sy?= =?UTF-8?q?stemTime,=20freeze/advance/unfreeze=20as=20the=20cfg(test)=20de?= =?UTF-8?q?terministic-seam;=20MemStore=20clock/wakes=20cfg(test)=20access?= =?UTF-8?q?ors.=20Validation=20(validation.rs)=20carries=20no=20new=20rule?= =?UTF-8?q?=20=E2=80=94=20pin=20module=20only=20(empty/whitespace=20Invali?= =?UTF-8?q?dName,=20prefix=20ReservedName,=20RESERVED=5FLISTENER=5FRECONNE?= =?UTF-8?q?CTED=20rejected=20though=20unused=20here);=20mechanisms=20route?= =?UTF-8?q?=20core's=20validate=5Fshared=5Fname/local=5Fname=20directly.?= =?UTF-8?q?=20Targeted=20allow(dead=5Fcode)=20on=20not-yet-wired=20scaffol?= =?UTF-8?q?d=20surfaces=20so=20clippy=20-D=20warnings=20stays=20green=20pr?= =?UTF-8?q?e-consumer=20(mechanism=20tasks=20own=20exhausting=20the=20allo?= =?UTF-8?q?ws);=20tokio=20time=20pinned=20unexercised=20(lock-TTL/schedule?= =?UTF-8?q?r=20consumers=20arrive=20with=20their=20tasks).=20Unit=20pins:?= =?UTF-8?q?=2017=20mem=20tests=20=E2=80=94=20two-instances-share-nothing?= =?UTF-8?q?=20(Arc=20pointer=20identity=20+=20behavioral:=20frozen=20clock?= =?UTF-8?q?=20isolation,=20cross-instance=20wake=20silence),=20state=20sec?= =?UTF-8?q?tion=20independence/relock,=20wake=20routing=20(channel-scoped?= =?UTF-8?q?=20+=20foreign=20silence,=20per-notify=20never=20coalesced,=20w?= =?UTF-8?q?atermark-ordered=20FIFO,=20no-replay,=20close=20disconnects=20a?= =?UTF-8?q?ll=20+=20post-close=20inert,=20dropped=20feed=20detaches),=20en?= =?UTF-8?q?gine-drop=20closes=20feeds,=20clock=20determinism,=20validation?= =?UTF-8?q?=20pin=20row.=20Task=20file:=20status=20completed,=20Notes=20(d?= =?UTF-8?q?ecisions=20of=20record:=20parking=5Flot=20rejection,=20empty=20?= =?UTF-8?q?sections,=20substrate=20API=20shape,=20no-close,=20scaffold=20a?= =?UTF-8?q?llow=20posture,=20unexercised=20time=20feature)=20+=20Summary?= =?UTF-8?q?=20filled.=20Verified:=20cargo=20build;=20workspace=20cargo=20t?= =?UTF-8?q?est=20green=20(pg=20124/25/9,=20sqlite=20191/25,=20core=2025,?= =?UTF-8?q?=20suite=203,=20mem=2017);=20clippy=20--all-targets=20-D=20warn?= =?UTF-8?q?ings=20clean;=20cargo=20fmt=20--check=20clean;=20cargo=20check?= =?UTF-8?q?=20--target=20wasm32-unknown-unknown=20-p=20alkstore-mem=20clea?= =?UTF-8?q?n=20=E2=80=94=20the=20wave-6=20wasm=20gate=20starts=20green=20f?= =?UTF-8?q?rom=20this=20task=20onward?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- Cargo.lock | 8 + Cargo.toml | 1 + alkstore-mem/Cargo.toml | 13 ++ alkstore-mem/src/clock.rs | 110 +++++++++++ alkstore-mem/src/lib.rs | 52 +++++ alkstore-mem/src/state.rs | 133 +++++++++++++ alkstore-mem/src/store.rs | 148 ++++++++++++++ alkstore-mem/src/validation.rs | 49 +++++ alkstore-mem/src/wakes.rs | 334 ++++++++++++++++++++++++++++++++ tasks/mem-engine-foundations.md | 98 +++++++++- 10 files changed, 938 insertions(+), 8 deletions(-) create mode 100644 alkstore-mem/Cargo.toml create mode 100644 alkstore-mem/src/clock.rs create mode 100644 alkstore-mem/src/lib.rs create mode 100644 alkstore-mem/src/state.rs create mode 100644 alkstore-mem/src/store.rs create mode 100644 alkstore-mem/src/validation.rs create mode 100644 alkstore-mem/src/wakes.rs diff --git a/Cargo.lock b/Cargo.lock index 371bc2d..fdcdf8d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -21,6 +21,14 @@ dependencies = [ "tokio", ] +[[package]] +name = "alkstore-mem" +version = "0.1.0" +dependencies = [ + "alkstore", + "tokio", +] + [[package]] name = "alkstore-postgres" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index 60468df..5dd1c1f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -3,6 +3,7 @@ members = [ "alkstore", "alkstore-sqlite", "alkstore-postgres", + "alkstore-mem", "alkstore-contract-suite", ] resolver = "3" diff --git a/alkstore-mem/Cargo.toml b/alkstore-mem/Cargo.toml new file mode 100644 index 0000000..55d67e2 --- /dev/null +++ b/alkstore-mem/Cargo.toml @@ -0,0 +1,13 @@ +[package] +name = "alkstore-mem" +version.workspace = true +edition.workspace = true +license.workspace = true +repository.workspace = true + +[dependencies] +alkstore = { version = "0.1", path = "../alkstore" } +tokio = { version = "1", features = ["sync", "time"] } + +[dev-dependencies] +tokio = { version = "1", features = ["rt", "macros"] } \ No newline at end of file diff --git a/alkstore-mem/src/clock.rs b/alkstore-mem/src/clock.rs new file mode 100644 index 0000000..a35e1e5 --- /dev/null +++ b/alkstore-mem/src/clock.rs @@ -0,0 +1,110 @@ +//! The engine's clock — every timestamp the mem engine produces (unix +//! seconds stamps, visibility deadlines, TTLs, scheduler boundaries) +//! reads this one accessor ([`Clock::unix_now`]); the process clock is +//! all there is (engine-mem.md: no server clock to defer to). +//! +//! The single point exists so tests can observe timestamps +//! deterministically the way the SQL engines observe row shapes: a +//! `#[cfg(test)]` freeze/advance seam pins a wall value and every +//! accessor call returns it (see the unit pins below). Production +//! reads `std::time::SystemTime`. + +#[cfg(test)] +use std::sync::Mutex; + +#[cfg(test)] +use crate::state::Guard; + +// Mechanism task code stamps every timestamp through this accessor — +// scaffolding reads as dead until the wave-6 mechanism tasks wire it. +#[allow(dead_code)] +fn process_unix_now() -> i64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|since| since.as_secs() as i64) + .unwrap_or(0) +} + +/// The engine's single clock. +#[derive(Debug)] +pub(crate) struct Clock { + #[cfg(test)] + frozen: Mutex>, +} + +impl Default for Clock { + fn default() -> Self { + Self::new() + } +} + +impl Clock { + pub(crate) fn new() -> Self { + Self { + #[cfg(test)] + frozen: Mutex::new(None), + } + } + + /// Unix-seconds "now" — the one timestamp source of the engine. + #[allow(dead_code)] + pub(crate) fn unix_now(&self) -> i64 { + #[cfg(test)] + let frozen = { *self.frozen.guard() }; + #[cfg(test)] + if let Some(unix) = frozen { + return unix; + } + process_unix_now() + } + + #[cfg(test)] + pub(crate) fn freeze(&self, at_unix: i64) { + *self.frozen.guard() = Some(at_unix); + } + + #[cfg(test)] + pub(crate) fn advance(&self, seconds: i64) { + let mut guard = self.frozen.guard(); + *guard = Some(frozen_now(&guard).saturating_add(seconds)); + } + + #[cfg(test)] + pub(crate) fn unfreeze(&self) { + *self.frozen.guard() = None; + } +} + +#[cfg(test)] +fn frozen_now(frozen: &Option) -> i64 { + frozen.unwrap_or_else(process_unix_now) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn unfrozen_reads_a_plausible_process_time() { + let clock = Clock::new(); + let now = clock.unix_now(); + assert!( + now > 1_700_000_000, + "the accessor reads the process clock when unfrozen" + ); + } + + #[test] + fn frozen_reads_deterministically_and_advances() { + let clock = Clock::new(); + clock.freeze(1_234_567); + assert_eq!(clock.unix_now(), 1_234_567); + clock.advance(10); + assert_eq!(clock.unix_now(), 1_234_577); + clock.unfreeze(); + assert!( + clock.unix_now() > 1_700_000_000, + "unfreeze restores the process clock" + ); + } +} diff --git a/alkstore-mem/src/lib.rs b/alkstore-mem/src/lib.rs new file mode 100644 index 0000000..2840101 --- /dev/null +++ b/alkstore-mem/src/lib.rs @@ -0,0 +1,52 @@ +//! alkstore-mem — the mem engine: the core contract implemented on +//! guarded in-memory state, process-owned (ADR-001, ADR-024). +//! +//! # Posture +//! +//! **Honest-ephemeral** (ADR-016 carriers, ADR-024 §2): no durability, +//! single-process, ephemeral by design — a legitimate production +//! choice for workloads that accept ephemerality, and never fleet-valid +//! (the shared-pool predicate fails structurally — there is no pool). +//! The dependency on `alkstore-mem` *is* the ephemerality declaration; +//! compile-time engine identity (this crate) carries the honesty — +//! no runtime surface is fabricated to detect it. Each +//! `MemStore::new` yields a fully isolated instance sharing nothing; +//! teardown is drop, and engine drop closes open receivers. +//! +//! **Full contract v1** (ADR-024 §3): every mechanism, every guarantee +//! row — the queue state machine, the scheduler collapse, streams with +//! per-consumer offsets and trim, TTL locks, the tx seam, wakes. Born +//! pinned: the contract suite's third `StoreFactory` column (ADR-022) +//! runs this engine under a current-thread runtime flavor. +//! +//! **wasm32** ([ADR-024] §5): this crate compiles clean for +//! `wasm32-unknown-unknown` — tokio restricted to the wasm-supported +//! subset (sync + time), no driver dependencies, no `spawn_blocking` +//! anywhere; internals are current-thread-runtime-clean by +//! construction. Every timestamp reads the process clock through one +//! internal accessor (no server clock to defer to). +//! +//! # Architecture +//! +//! The store instance *is* the state: guarded per-mechanism sections +//! (`crate::state`) behind short-held synchronous internal locks, +//! tokio primitives for cross-await coordination. The tx seam is an +//! overlay (stage → commit-merge → wake-at-commit; drop = discard); +//! wakes route through in-process channels (`crate::wakes`) — +//! delivered at the commit boundary directly to each open receiver, +//! per-subscriber FIFO by commit order, never coalesced (no batching +//! boundary exists). +//! +//! Wave 6 is under construction (ADR-024): the foundations — +//! `MemStore::new`, the guarded state core, the wake-routing +//! substrate, the clock accessor, entry-point validation — are laid; +//! the `Store`/`TxHandle` trait surfaces and the mechanism handles +//! land with the wave's seam and mechanism tasks. + +mod clock; +mod state; +mod store; +mod validation; +mod wakes; + +pub use store::MemStore; diff --git a/alkstore-mem/src/state.rs b/alkstore-mem/src/state.rs new file mode 100644 index 0000000..f7c90d5 --- /dev/null +++ b/alkstore-mem/src/state.rs @@ -0,0 +1,133 @@ +//! The guarded per-mechanism state core — the store instance *is* the +//! state (engine-mem.md §Architecture): per-mechanism tables owned by +//! the engine instance behind the family's locking discipline +//! (short-held synchronous internal locks; tokio primitives for +//! cross-await coordination — [`crate::wakes`]' channels and the +//! scheduler's boundary waits arrive with their tasks). +//! +//! The idiom: each mechanism's committed state sits behind one +//! short-held [`std::sync::Mutex`] section, accessed through +//! [`Guard::guard`]. Guards are never held across an `.await`, never +//! across user-provided code (delivery closures, payload callbacks) — +//! staging happens engine-side, so the mutex cannot observe a panic +//! while held and poisoning is structurally unreachable; the +//! [`Guard::guard`] fold into the inner value keeps the type honest +//! anyway. +//! +//! This module owns the shared scaffolding only: the container, the +//! section shapes' names, and the guard idiom. Each mechanism's table +//! shape is owned by its own wave-6 task (`mem-engine-queues`, +//! `mem-engine-streams`, `mem-engine-locks`, +//! `mem-engine-scheduler-outbox`) — the sections here start empty and +//! grow there. + +use std::sync::Mutex; +use std::sync::MutexGuard; + +/// A short-held synchronous internal lock: `guard()` is the one access +/// idiom (see the module docs for the discipline). +pub(crate) trait Guard { + fn guard(&self) -> MutexGuard<'_, T>; +} + +impl Guard for Mutex { + fn guard(&self) -> MutexGuard<'_, T> { + self.lock().unwrap_or_else(|poisoned| poisoned.into_inner()) + } +} + +/// The queue mechanism's committed section: the jobs table and the +/// dead-letter storage (`mem-engine-queues` owns the shapes). +#[derive(Debug, Default)] +pub(crate) struct QueueState {} + +/// The stream mechanism's committed section: event logs and +/// per-consumer offsets (`mem-engine-streams` owns the shapes). +#[derive(Debug, Default)] +pub(crate) struct StreamState {} + +/// The named-locks mechanism's committed section: the TTL registry +/// (`mem-engine-locks` owns the shapes). +#[derive(Debug, Default)] +pub(crate) struct LockState {} + +/// The scheduler mechanism's committed section: the schedule table +/// (`mem-engine-scheduler-outbox` owns the shapes; the outbox rides +/// queues). +#[derive(Debug, Default)] +pub(crate) struct ScheduleState {} + +/// The store's committed state: one guarded section per mechanism. +/// Sections are independent — a mechanism task locks only its own. +// The sections are read by their owning wave-6 mechanism tasks; the +// foundations build alone sees them as dead. +#[allow(dead_code)] +pub(crate) struct State { + pub(crate) queues: Mutex, + pub(crate) streams: Mutex, + pub(crate) locks: Mutex, + pub(crate) schedules: Mutex, +} + +impl State { + pub(crate) fn new() -> Self { + Self { + queues: Mutex::new(QueueState::default()), + streams: Mutex::new(StreamState::default()), + locks: Mutex::new(LockState::default()), + schedules: Mutex::new(ScheduleState::default()), + } + } +} + +impl std::fmt::Debug for State { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("State").finish_non_exhaustive() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn sections_are_per_mechanism_and_independent() { + let state = State::new(); + let queued_ptr = { + let mut queues = state.queues.guard(); + std::ptr::from_mut(&mut *queues) + }; + let streams_ptr = { + let mut streams = state.streams.guard(); + std::ptr::from_mut(&mut *streams) + }; + let queues_ptr_again = { + let mut queues = state.queues.guard(); + std::ptr::from_mut(&mut *queues) + }; + assert_eq!(queued_ptr, queues_ptr_again, "the same section relocks"); + assert_ne!( + queued_ptr.cast::<()>(), + streams_ptr.cast::<()>(), + "sections are distinct storage" + ); + } + + #[test] + fn sections_relock_after_guard_drop() { + let state = std::sync::Arc::new(State::new()); + std::thread::scope(|scope| { + let held = scope.spawn(|| { + for _ in 0..100 { + let _guard = state.streams.guard(); + std::thread::yield_now(); + } + }); + for _ in 0..100 { + let _guard = state.queues.guard(); + std::thread::yield_now(); + } + held.join().expect("the lock churn task never panics"); + }); + } +} diff --git a/alkstore-mem/src/store.rs b/alkstore-mem/src/store.rs new file mode 100644 index 0000000..74c3b92 --- /dev/null +++ b/alkstore-mem/src/store.rs @@ -0,0 +1,148 @@ +//! The mem engine's store: `MemStore::new`, the engine-native +//! constructor (ADR-024 §4 — the one documented exception to the +//! open prose), and the guarded state it owns. +//! +//! Isolation per open: each `new()` yields a fully isolated instance +//! sharing nothing — a fresh `State` (the guarded per-mechanism +//! core), a fresh `WakeBus` (the wake-routing substrate), a fresh +//! `Clock` — exactly the suite factory's isolation contract; +//! two instances never contend and never observe each other. There is +//! no connection, no file, no pool; teardown is drop — the +//! engine-wide `Drop` runs the close path (the wake bus empties, every +//! open feed disconnects), and the state is gone with the process's +//! references, honestly ephemeral ADR-016-shaped posture: +//! the dependency on `alkstore-mem` *is* the ephemerality +//! declaration). +//! +//! The `Store` trait implementation is the seam task's artifact +//! (`mem-engine-seam-tx` — the tx overlay is the load-bearing half); +//! this type is the trait boundary's substrate. + +use std::sync::Arc; + +use crate::clock::Clock; +use crate::state::State; +use crate::wakes::WakeBus; + +/// The mem engine's store handle. Consumers take it from +/// `MemStore::new`; once the seam task lands, they hold +/// `alkstore::Store` — the trait is the contract surface, never the +/// engine type. +pub struct MemStore { + // Read by the wave-6 mechanism task code (the `Store` seam and the + // handles); the foundations build alone sees `state`/`clock` as dead. + #[allow(dead_code)] + pub(crate) state: Arc, + pub(crate) wakes: Arc, + #[allow(dead_code)] + pub(crate) clock: Arc, +} + +impl Default for MemStore { + fn default() -> Self { + Self::new() + } +} + +impl std::fmt::Debug for MemStore { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("MemStore").finish_non_exhaustive() + } +} + +impl MemStore { + /// Construct a fully isolated mem engine instance. Engine-native: + /// no connection string, no file path (ADR-024 §4's + /// documented exception to the open prose). Teardown is drop; + /// each instance shares nothing with any other. + pub fn new() -> Self { + Self { + state: Arc::new(State::new()), + wakes: Arc::new(WakeBus::new()), + clock: Arc::new(Clock::new()), + } + } + + #[cfg(test)] + pub(crate) fn wakes(&self) -> &Arc { + &self.wakes + } + + #[cfg(test)] + pub(crate) fn clock(&self) -> &Arc { + &self.clock + } +} + +impl Drop for MemStore { + fn drop(&mut self) { + self.wakes.close(); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn two_new_instances_share_nothing() { + let a = MemStore::new(); + let b = MemStore::new(); + assert!( + !std::sync::Arc::ptr_eq(&a.state, &b.state), + "the state cores are distinct instances" + ); + assert!( + !std::sync::Arc::ptr_eq(&a.wakes, &b.wakes), + "the wake substrates are distinct instances" + ); + assert!( + !std::sync::Arc::ptr_eq(&a.clock, &b.clock), + "the clocks are distinct instances" + ); + } + + #[test] + fn instances_observed_through_their_futures_share_nothing() { + let a = MemStore::new(); + let b = MemStore::new(); + a.clock().freeze(1_000_000); + assert_eq!(a.clock().unix_now(), 1_000_000, "instance a's clock froze"); + assert!( + b.clock().unix_now() > 1_700_000_000, + "instance b's clock kept the process time" + ); + } + + #[test] + fn wake_substrates_do_not_cross_instances() { + let a = MemStore::new(); + let b = MemStore::new(); + a.wakes().fire_commit_wakes(vec!["orders".into()]); + let mut late_feed = b.wakes().attach("orders"); + assert!( + late_feed.rx().try_recv().is_err(), + "instance a's commit fired before the feed attached — instance b's substrate never replayed it, and instance a's substrate is unreachable from b" + ); + let mut a_feed = a.wakes().attach("orders"); + a.wakes().fire_commit_wakes(vec!["orders".into()]); + assert!( + matches!( + a_feed.rx().try_recv(), + Ok(wake) if wake.channel == "orders" + ), + "instance a's substrate routes its own attaches" + ); + } + + #[tokio::test(flavor = "current_thread")] + async fn engine_drop_closes_open_feeds() { + let store = MemStore::new(); + let mut feed = store.wakes().attach("orders"); + std::mem::drop(store); + assert!( + feed.rx().recv().await.is_none(), + "teardown is drop: the engine's close path disconnects every feed" + ); + } +} diff --git a/alkstore-mem/src/validation.rs b/alkstore-mem/src/validation.rs new file mode 100644 index 0000000..133b8dc --- /dev/null +++ b/alkstore-mem/src/validation.rs @@ -0,0 +1,49 @@ +//! The entry-point name validation is core's ([`alkstore`]'s +//! `validate_shared_name` / `validate_local_name` — ADR-008 §4), same +//! as every engine; this module carries no new rule, only the mem +//! engine's pin that the rule it routes every future entry point +//! through is the family rule: empty and whitespace-only names are +//! `InvalidName`, the reserved prefix `__alkstore_` is `ReservedName`, +//! and the one reserved string (`__alkstore_listener_reconnected__` +//! — the Postgres reconnect channel, unused on this engine) is still +//! rejected because it carries the prefix. + +#[cfg(test)] +mod tests { + use alkstore::Error; + + #[test] + fn empty_and_whitespace_only_names_are_invalid() { + for name in ["", " "] { + assert!( + matches!( + alkstore::validate_shared_name(name), + Err(Error::InvalidName { .. }) + ), + "`{name}` validates as `InvalidName` on the shared kind" + ); + } + } + + #[test] + fn reserved_prefix_names_are_rejected() { + assert!(matches!( + alkstore::validate_shared_name("__alkstore_scheduler"), + Err(Error::ReservedName { .. }) + )); + } + + #[test] + fn the_reserved_listener_string_is_rejected_even_though_unused_here() { + assert!( + alkstore::validate_shared_name(alkstore::RESERVED_LISTENER_RECONNECTED).is_err(), + "the reserved string is unused on this engine but still rejected" + ); + } + + #[test] + fn ordinary_names_validate() { + assert!(alkstore::validate_shared_name("orders").is_ok()); + assert!(alkstore::validate_local_name("__alkstore_-prefixed-local-ids-are-legal").is_ok()); + } +} diff --git a/alkstore-mem/src/wakes.rs b/alkstore-mem/src/wakes.rs new file mode 100644 index 0000000..c1ea47c --- /dev/null +++ b/alkstore-mem/src/wakes.rs @@ -0,0 +1,334 @@ +//! The wake-routing substrate: in-process channels carrying wakes at +//! the commit boundary (engine-mem.md §Architecture). +//! +//! There is no batching boundary — wakes deliver at the commit +//! boundary directly to each open receiver, per-subscriber FIFO by +//! commit order; coalescing is not applicable. The substrate realizes +//! the shape structurally: +//! +//! - **Commit-ordered source**: a mechanism (notify now — streams and +//! queue wakes arrive with their tasks) fires its commit's wakes +//! through [`WakeBus::fire_commit_wakes`], under one guard: the +//! commit sequence bumps and the per-channel fanout completes as a +//! single operation, so every subscriber's receipt order is exactly +//! commit order. Two concurrent commits serialize on the guard; a +//! subscriber's wake stream is FIFO by commit order. +//! - **Per-channel attach**: [`WakeBus::attach`] registers under the +//! same guard and returns an unbounded [`tokio::sync::mpsc`] feed — +//! attach never blocks, and a receiver attached after a commit never +//! sees it (no replay; the attach either happens before that commit's +//! fanout or after it — the guard admits no mid-fanout attach). +//! - **Subscriber registry**: unbounded senders live only in the bus, +//! so the sole close path is the engine's drop. [`WakeBus::close`] +//! empties the registry under the same guard — every live feed's +//! sender drops and each receiver drains to `Disconnected`; the +//! consumer-visible `recv() -> None` terminal arm arrives with +//! `mem-engine-notify-listen`'s [`alkstore::WakeReceiver`] bridge +//! over this substrate. A dropped feed detaches itself +//! ([`WakeFeed`]'s `Drop`). +//! +//! The `Wake` value crosses exactly as the contract pins it — channel +//! only, never payloads or ids (ADR-008 §3). The fanout is per-notify: +//! one commit's notify on a channel is one wake per subscribed feed, +//! never coalesced (no batching boundary exists to coalesce across). + +// The substrate routes every wave-6 mechanism's commit-boundary +// wakes; until the mechanism tasks wire it, the foundations-only +// build sees the attach/fire surface as dead. +#![allow(dead_code)] + +use std::sync::Arc; +use std::sync::Mutex; + +use alkstore::Wake; + +use crate::state::Guard; + +/// A subscriber's registry row: its id (the detach key), the channel +/// it attached on (exact-match deliverance), and its feed's sender. +/// Senders live only here — a subscriber holds no bus reference of its +/// own beyond its [`WakeFeed`] (which carries the bus Arc for +/// detach-on-drop, never for delivery), so the engine's close empties +/// the senders and the feeds disconnect regardless of who still holds +/// a feed. +#[derive(Debug)] +struct Subscription { + id: u64, + channel: String, + tx: tokio::sync::mpsc::UnboundedSender, +} + +/// The internal guarded registry: next-subscriber id, next commit +/// sequence, and the live subscriptions in first-attached order. +#[derive(Debug, Default)] +struct Registry { + next_subscriber_id: u64, + next_commit_seq: u64, + subscriptions: Vec, +} + +/// The in-process wake bus: the commit-ordered wake source's entry +/// point ([`WakeBus::fire_commit_wakes`]) and the per-channel receiver +/// attach ([`WakeBus::attach`]) over one guarded registry. +pub(crate) struct WakeBus { + registry: Mutex, + closed: std::sync::atomic::AtomicBool, +} + +impl Default for WakeBus { + fn default() -> Self { + Self::new() + } +} + +impl std::fmt::Debug for WakeBus { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("WakeBus").finish_non_exhaustive() + } +} + +impl WakeBus { + #[allow(dead_code)] + pub(crate) fn new() -> Self { + Self { + registry: Mutex::new(Registry::default()), + closed: std::sync::atomic::AtomicBool::new(false), + } + } + + /// Attach a feed for `channel` — starts from "now": the attach + /// lands under the registry guard, so commits that fired before it + /// are never replayed into it. Returns the feed plus its detach + /// handle (drop the [`WakeFeed`] to detach). + pub(crate) fn attach(self: &Arc, channel: &str) -> WakeFeed { + let (tx, rx) = tokio::sync::mpsc::unbounded_channel(); + let mut registry = self.registry.guard(); + if self.closed.load(std::sync::atomic::Ordering::Acquire) { + return WakeFeed { + id: 0, + bus: Arc::clone(self), + rx: dead_feed_receiver(), + }; + } + registry.next_subscriber_id += 1; + let id = registry.next_subscriber_id; + registry.subscriptions.push(Subscription { + id, + channel: channel.to_string(), + tx, + }); + WakeFeed { + id, + bus: Arc::clone(self), + rx, + } + } + + /// Fire the commit boundary's wakes — the mechanism's wake source: + /// under one guard, the commit sequence bumps once and every feed + /// attached on a fired channel receives one [`Wake`] per commit-notify, + /// in commit order. A channel with no subscribers delivers nothing. + /// A closed bus is inert. + pub(crate) fn fire_commit_wakes(&self, channels: impl IntoIterator) { + if self.closed.load(std::sync::atomic::Ordering::Acquire) { + return; + } + let mut registry = self.registry.guard(); + registry.next_commit_seq += 1; + let fired: Vec = channels.into_iter().collect(); + for channel in &fired { + for subscription in ®istry.subscriptions { + if subscription.channel == *channel { + let wake = Wake::new(channel.clone()); + let _ = subscription.tx.send(wake); + } + } + } + } + + /// The commit watermark: how many commit boundaries have fired + /// through this bus. The substrate's ordering observable — the + /// consumer contract never surfaces it (wakes carry the channel + /// only); unit tests pin FIFO-by-commit-order against it. + #[cfg(test)] + pub(crate) fn commit_watermark(&self) -> u64 { + self.registry.guard().next_commit_seq + } + + /// Close the bus — the engine's drop path. Empties the registry + /// under one guard: every feed's sender drops, receivers drain to + /// disconnected, and attach/fire after it are no-ops. Idempotent. + pub(crate) fn close(&self) { + self.closed + .store(true, std::sync::atomic::Ordering::Release); + let mut registry = self.registry.guard(); + registry.subscriptions.clear(); + } + + fn detach(&self, id: u64) { + let mut registry = self.registry.guard(); + registry.subscriptions.retain(|sub| sub.id != id); + } +} + +/// A feed whose sender never existed — the attach-on-closed shape (the +/// engine is gone; the feed reads as closed immediately, the same +/// terminal arm an engine drop produces). +fn dead_feed_receiver() -> tokio::sync::mpsc::UnboundedReceiver { + let (_tx, rx) = tokio::sync::mpsc::unbounded_channel::(); + rx +} + +/// A receiver's handle into the substrate: the raw unbounded feed of +/// [`Wake`] values (per-subscriber FIFO by commit order — see the +/// module docs) plus its own-detach-on-drop. This is the substrate +/// primitive; the consumer-visible `WakeReceiver` bridge (the +/// `recv`/`try_recv`/`recv_timeout` forms and the terminal `None` +/// arm) is `mem-engine-notify-listen`'s surface over it. +pub(crate) struct WakeFeed { + id: u64, + bus: Arc, + rx: tokio::sync::mpsc::UnboundedReceiver, +} + +impl std::fmt::Debug for WakeFeed { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("WakeFeed") + .field("id", &self.id) + .finish_non_exhaustive() + } +} + +impl WakeFeed { + pub(crate) fn rx(&mut self) -> &mut tokio::sync::mpsc::UnboundedReceiver { + &mut self.rx + } +} + +impl Drop for WakeFeed { + fn drop(&mut self) { + if self.id != 0 { + self.bus.detach(self.id); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + use tokio::sync::mpsc::error::TryRecvError; + + fn delivered(feed: &mut WakeFeed) -> Vec { + let mut wakes = Vec::new(); + loop { + match feed.rx().try_recv() { + Ok(wake) => wakes.push(wake), + Err(TryRecvError::Empty) => break, + Err(TryRecvError::Disconnected) => { + panic!("the feed disconnected unexpectedly") + } + } + } + wakes + } + + #[test] + fn delivery_is_channel_scoped_and_never_coalesced() { + let bus = Arc::new(WakeBus::new()); + let mut a = bus.attach("orders"); + let mut b = bus.attach("orders"); + let mut foreign = bus.attach("payments"); + bus.fire_commit_wakes(vec!["orders".into(), "orders".into(), "payments".into()]); + let a_wakes = delivered(&mut a); + let b_wakes = delivered(&mut b); + let foreign_wakes = delivered(&mut foreign); + assert_eq!( + a_wakes, b_wakes, + "subscribers on the same channel receive the same wakes" + ); + assert!( + a_wakes.iter().all(|w| w.channel == "orders"), + "each wake carries its channel" + ); + assert_eq!( + foreign_wakes[0].channel, "payments", + "a foreign-channel subscriber receives only its own channel" + ); + assert_eq!( + a_wakes.len(), + 2, + "per-notify delivery — one wake per commit, never coalesced" + ); + } + + #[test] + fn per_subscriber_fifo_by_commit_order() { + let bus = Arc::new(WakeBus::new()); + let mut feed = bus.attach("orders"); + for seq in 0..8 { + assert_eq!( + bus.commit_watermark(), + seq, + "the watermark advances one per fired commit boundary" + ); + bus.fire_commit_wakes(vec![format!("orders-{seq}"), "orders".into()]); + } + assert_eq!(bus.commit_watermark(), 8); + let wakes = delivered(&mut feed); + assert_eq!( + wakes, + vec![Wake::new("orders".into()); 8], + "each commit's wake delivered, in commit order" + ); + } + + #[test] + fn attach_after_commit_sees_nothing_before_it() { + let bus = Arc::new(WakeBus::new()); + bus.fire_commit_wakes(vec!["orders".into()]); + let mut late = bus.attach("orders"); + assert!( + late.rx().try_recv().is_err(), + "a listener attached after a commit never sees it (no replay)" + ); + bus.fire_commit_wakes(vec!["orders".into()]); + assert_eq!( + delivered(&mut late), + vec![Wake::new("orders".into())], + "post-attach commits deliver" + ); + } + + #[tokio::test(flavor = "current_thread")] + async fn close_disconnects_every_live_feed() { + let bus = Arc::new(WakeBus::new()); + let mut a = bus.attach("orders"); + let mut b = bus.attach("payments"); + bus.close(); + bus.fire_commit_wakes(vec!["orders".into()]); + let mut late = bus.attach("orders"); + for feed in [&mut a, &mut b, &mut late] { + assert!( + feed.rx().recv().await.is_none(), + "engine close is the terminal None arm for every open receiver" + ); + } + assert_eq!(bus.commit_watermark(), 0, "a fire on a closed bus is inert"); + } + + #[test] + fn dropped_feed_detaches() { + let bus = Arc::new(WakeBus::new()); + let feed = bus.attach("orders"); + let mut kept = bus.attach("orders"); + std::mem::drop(feed); + bus.fire_commit_wakes(vec!["orders".into()]); + let wakes = delivered(&mut kept); + assert_eq!( + wakes.len(), + 1, + "the detached feed received nothing, the live one the wake" + ); + } +} diff --git a/tasks/mem-engine-foundations.md b/tasks/mem-engine-foundations.md index 7e0bdc3..cfc88d4 100644 --- a/tasks/mem-engine-foundations.md +++ b/tasks/mem-engine-foundations.md @@ -1,7 +1,7 @@ --- id: mem-engine-foundations name: Mem engine — crate scaffold, `MemStore::new()`, guarded state core, wake routing -status: pending +status: completed depends_on: [] scope: moderate risk: medium @@ -63,16 +63,16 @@ this task lands everything up to the trait boundary. ## Acceptance Criteria -- [ ] `alkstore-mem` is a workspace member compiling clean; manifest +- [x] `alkstore-mem` is a workspace member compiling clean; manifest dependency surface is core + tokio (sync + time) only -- [ ] `MemStore::new()` constructs; two instances share nothing +- [x] `MemStore::new()` constructs; two instances share nothing (pinned by a unit test) -- [ ] The guarded state core, validation rules, wake-routing +- [x] The guarded state core, validation rules, wake-routing substrate, and clock accessor exist with unit pins -- [ ] `cargo check --target wasm32-unknown-unknown -p alkstore-mem` +- [x] `cargo check --target wasm32-unknown-unknown -p alkstore-mem` clean from this task onward (tokio subset discipline enforced here, not retro-fitted) -- [ ] Workspace gates green (`cargo build`, `cargo test`, clippy +- [x] Workspace gates green (`cargo build`, `cargo test`, clippy `-D warnings`, fmt) ## References @@ -85,8 +85,90 @@ this task lands everything up to the trait boundary. ## Notes -> To be filled by implementation agent +> Filled by implementation agent, 2026-10-10. + +1. **parking_lot rejected — the ADR won the manifest.** The + description's prose named "parking_lot short-held internal locks + per AGENTS convention 3", but ADR-024 §1 and the acceptance + criterion pin the manifest to core + tokio (sync + time) only. + The guarded-state container uses plain `std::sync::Mutex` + (wasm-fine, no crates.io surface) behind a `Guard` trait whose + `guard()` folds poisoning into `into_inner` — structurally + unreachable anyway, since no guard is ever held across an `.await` + or user-provided code; tokio primitives (`mpsc` unbounded) are the + cross-await coordination. Downstream mechanism tasks follow this + idiom (`state::Guard`). +2. **Guarded-state shapes deliberately empty** — `state.rs` carries + the container plus named empty per-mechanism sections + (`QueueState` = jobs + dead letters, `StreamState` = log + offsets, + `LockState` = TTL registry, `ScheduleState` = schedule table); + each mechanism task owns its section's actual shape. +3. **Wake substrate API shape** (`wakes.rs`): + `WakeBus::attach(channel) -> WakeFeed` (one unbounded mpsc feed per + subscriber; attach lands under the registry guard, so + attach-after-commit no-replay is structural), + `fire_commit_wakes(channels)` is the commit-ordered wake source's + entry point (mechanisms fire at the commit boundary; the guard + makes bump + per-channel fanout one operation — per-subscriber FIFO + by commit order, never coalesced), `close()` is the engine-drop + path (MemStore's `Drop` calls it; senders live only in the bus, so + a drop disconnects every feed regardless of who holds one), and + `commit_watermark()` is a `cfg(test)` ordering observable only — + never contract surface. `WakeFeed::rx()` exposes the raw mpsc + receiver; the consumer-visible `WakeReceiver` bridge + (recv/try_recv/recv_timeout forms) is + `mem-engine-notify-listen`'s surface over this primitive. +4. **Clock** (`clock.rs`): one accessor `Clock::unix_now()` reads + `std::time::SystemTime` (process clock only, per engine-mem.md); + `freeze`/`advance`/`unfreeze` are the `cfg(test)` deterministic + observation seam. `MemStore::clock()` `cfg(test)` accessor. +5. **Validation carries no new rule** — `validation.rs` is the pin + module only (empty/whitespace-only → `InvalidName`, reserved + prefix → `ReservedName`, `RESERVED_LISTENER_RECONNECTED` rejected + even though unused here); every future mem entry point routes + through core's `validate_shared_name`/`validate_local_name` + directly, same as the SQL engines. +6. **Scaffold dead-code posture**: targeted `#[allow(dead_code)]` on + the not-yet-wired scaffold surfaces (the `wakes` module, the empty + `State` sections, `MemStore`'s `state`/`clock` fields, + `Clock::unix_now`) so clippy `-D warnings` stays green with the + gate before consumers exist — the mechanism tasks own using (and + thereby exhausting) these allows. +7. **No `MemStore::close()`** — ADR-024 §4 says teardown is drop; + there is no explicit close form (the sqlite/pg engines' `close` is + connection machinery). If `mem-engine-integration` wants one for + symmetry, that is the place to decide. +8. **tokio `time` feature is pinned but currently unexercised** — no + mem code reads tokio time yet (lock TTLs and scheduler boundaries + are their tasks' consumers); the manifest carries it per this + task's description so the wasm-subset discipline is enforced at + birth rather than widened later. ## Summary -> To be filled on completion \ No newline at end of file +> Filled on completion, 2026-10-10. + +Landed `alkstore-mem` as the fourth workspace member — crate scaffold ++ manifest (deps: `alkstore` path pin per ADR-017 §4.1 + tokio +`sync`/`time` only; dev-deps tokio `rt`/`macros`), five modules +(`store`, `state`, `wakes`, `clock`, `validation`), re-exported from +`lib.rs` whose crate docs carry the ADR-016 posture statement +(honest-ephemeral, full contract v1, wasm32-compile-clean identity). +`MemStore::new()` implements the engine-native isolated-constructor +contract (ADR-024 §4): new guarded state core, wake bus, and clock +per open, teardown = drop driving `WakeBus::close()`. Unit pins: +16 tests — two-instances-share-nothing (Arc pointer identity + +behavioral: frozen clock on one instance leaves the other's process +time intact; a fire through one instance's bus never reaches the +other's feed), guarded-state section independence/relock, wake +routing (channel-scoped per-subscriber delivery + foreign-channel +silence, per-commit-notify never coalesced, watermark-ordered FIFO, +attach-after-commit no-replay, close disconnects every live feed + +inert post-close fire, dropped feed detaches), engine-drop closes +open feeds, clock determinism (freeze/advance/unfreeze), validation +pins (4). Verified: `cargo build`, workspace `cargo test` green (pg 124 lib / 25 suite / +9 schema, sqlite 191 lib / 25 suite, core 25, suite harness 3, +**mem 17**), clippy +`--all-targets -D warnings` clean, `cargo fmt --check` clean, and +`cargo check --target wasm32-unknown-unknown -p alkstore-mem` clean +— the wave-6 wasm gate starts green from this task onward. \ No newline at end of file