From eefee9ec1d9fac055f983c6b4443ed094924a139 Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Wed, 7 Oct 2026 15:50:02 +0000 Subject: [PATCH] =?UTF-8?q?Core=20trait=20surface:=20Store,=20TxHandle,=20?= =?UTF-8?q?mechanism=20handles,=20receivers,=20with=5Ftx=20(ADR-007=20?= =?UTF-8?q?=C2=A7with=5Ftx,=20ADR-008=20=C2=A71=E2=80=93=C2=A73/=C2=A78,?= =?UTF-8?q?=20ADR-009=20=C2=A71/=C2=A76,=20ADR-014,=20ADR-015=20=C2=A72,?= =?UTF-8?q?=20ADR-019=20=C2=A71=E2=80=93=C2=A76,=20ADR-021=20=C2=A71/?= =?UTF-8?q?=C2=A74/=C2=A75)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- Cargo.lock | 1 + alkstore/Cargo.toml | 3 + alkstore/src/error.rs | 15 ++ alkstore/src/event_receiver.rs | 58 +++++ alkstore/src/future.rs | 17 ++ alkstore/src/job_handle.rs | 57 +++++ alkstore/src/lib.rs | 22 ++ alkstore/src/lock.rs | 26 ++ alkstore/src/outbox.rs | 56 +++++ alkstore/src/queue.rs | 74 ++++++ alkstore/src/store.rs | 134 ++++++++++ alkstore/src/stream_handle.rs | 77 ++++++ alkstore/src/tests.rs | 12 + alkstore/src/trait_probe_tests.rs | 397 ++++++++++++++++++++++++++++++ alkstore/src/tx.rs | 144 +++++++++++ alkstore/src/wake_receiver.rs | 38 +++ tasks/core-trait-surface.md | 81 +++++- 17 files changed, 1204 insertions(+), 8 deletions(-) create mode 100644 alkstore/src/event_receiver.rs create mode 100644 alkstore/src/future.rs create mode 100644 alkstore/src/job_handle.rs create mode 100644 alkstore/src/lock.rs create mode 100644 alkstore/src/outbox.rs create mode 100644 alkstore/src/queue.rs create mode 100644 alkstore/src/store.rs create mode 100644 alkstore/src/stream_handle.rs create mode 100644 alkstore/src/trait_probe_tests.rs create mode 100644 alkstore/src/tx.rs create mode 100644 alkstore/src/wake_receiver.rs diff --git a/Cargo.lock b/Cargo.lock index 0e237d1..e86c9ed 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9,6 +9,7 @@ dependencies = [ "serde", "serde_json", "thiserror", + "tokio", ] [[package]] diff --git a/alkstore/Cargo.toml b/alkstore/Cargo.toml index ea2c4e1..dda44fb 100644 --- a/alkstore/Cargo.toml +++ b/alkstore/Cargo.toml @@ -9,3 +9,6 @@ repository.workspace = true serde = "1" serde_json = "1" thiserror = "2" + +[dev-dependencies] +tokio = { version = "1", features = ["macros", "rt", "time"] } diff --git a/alkstore/src/error.rs b/alkstore/src/error.rs index 99edb47..7f36017 100644 --- a/alkstore/src/error.rs +++ b/alkstore/src/error.rs @@ -21,6 +21,21 @@ pub enum Error { /// A name failed entry-point validation. #[error("invalid name `{name}`")] InvalidName { name: String }, + /// A schedule spec that doesn't parse under the v1 grammar + /// (`@every ` with `` a positive integer and `` + /// in `s|m|h|d`). + /// + /// Rejected by `schedule()` before storage, with no engine round + /// trip; the caller fixes the spec string (ADR-009 §2, §6). + #[error("invalid schedule spec `{spec}`")] + InvalidSpec { spec: String }, + /// The `run_schedules` runner's leadership lock was lost mid-run. + /// + /// A distinct, matchable outcome from the clean `Ok(())` a stop + /// token produces: the caller acts on it by respawning the runner + /// (ADR-009 §1, §6). + #[error("scheduler leadership lost")] + LeadershipLost, /// An operation against a closed receiver/source. /// /// The `try_recv`/`recv_timeout` form of wake close (never diff --git a/alkstore/src/event_receiver.rs b/alkstore/src/event_receiver.rs new file mode 100644 index 0000000..309d62c --- /dev/null +++ b/alkstore/src/event_receiver.rs @@ -0,0 +1,58 @@ +//! The durable stream subscription receiver (ADR-008 §8; ADR-021 §5; +//! ADR-019 §6). +//! +//! Explicit saves only — no `save_every`, no save-on-drop (the +//! auto-checkpoint ambiguity is not inherited); checkpointing is the +//! consumer's call. + +use crate::Result; +use crate::future::BoxedFuture; +use crate::stream::StreamEvent; + +/// The durable stream subscription handle `StreamHandle::subscribe` +/// returns. +/// +/// # Close and error arms (ADR-021 §5) +/// +/// `recv()`'s `Err` arm carries `Database` only — payloads cross +/// raw, and `Codec` lives at `payload_as` on the yielded +/// [`StreamEvent`]; `Closed` is the `None` arm by construction. Close +/// is terminal; what causes it is the documented engine asymmetry +/// (SQLite: watcher death; Postgres: engine shutdown — reconnect +/// wakes arrive on the receiver). A closed receiver never reopens; +/// recovery is a fresh `subscribe(consumer)` resuming from the saved +/// offset. No `recv_timeout` exists here — deliberate absence (a +/// stream consumer's steady state is wake-driven re-read; polling +/// needs are served by `try_recv` and the offset-anchored +/// `read_since`). +pub trait EventReceiver: Send { + /// Wait for the next event. `None` = the stream receiver closed + /// (terminal). `Some(Err(...))` carries `Database` only. + fn recv<'a>(&'a mut self) -> BoxedFuture<'a, Option>>; + + /// Poll without waiting. `Ok(None)` = no event pending right now; + /// `Err` carries `Database` or `Closed` (closed source). + fn try_recv(&mut self) -> Result>; + + /// Offset-anchored read from the stream, independent of this + /// receiver's position (saved checkpoints do not gate reads — + /// replay is a read concern). + fn read_since<'a>( + &'a mut self, + offset: i64, + limit: i64, + ) -> BoxedFuture<'a, Result>>; + + /// Checkpoint the last-yielded event's offset through the same + /// monotone op the direct `save_offset` form uses — the forms + /// interleave freely and cannot regress one another (ADR-019 §6). + /// A save before any event has been yielded is a no-op (the + /// stored checkpoint untouched). The `Err` arm carries `Database` + /// only (ADR-021 §5). + fn save_offset(&mut self) -> Result<()>; + + /// The receiver's current offset (the last-yielded event's + /// offset), used when a consumer wants to interleave the direct + /// save form with the receiver's position. + fn offset(&self) -> i64; +} diff --git a/alkstore/src/future.rs b/alkstore/src/future.rs new file mode 100644 index 0000000..f6b5267 --- /dev/null +++ b/alkstore/src/future.rs @@ -0,0 +1,17 @@ +//! The boxed-future alias (ADR-008 §2's "desugared boxed-future form" +//! — the family-standard async trait posture): every async trait +//! method crosses the contract as a named future type, keeping the +//! traits object-safe without the `async_trait` macro's erasure. +//! +//! `Send` bounds the future: handles are constructed and polled across +//! `await` points on multi-threaded tokio runtimes, and +//! `Store::begin_tx` returns `Box` — the futures +//! its methods produce must be equally portable across the runtime's +//! worker threads (SQLite's engine realizes this by round-tripping +//! `spawn_blocking` to the writer thread behind the lease, ADR-003). + +use std::future::Future; +use std::pin::Pin; + +/// The future type every async contract method returns. +pub type BoxedFuture<'a, T> = Pin + Send + 'a>>; diff --git a/alkstore/src/job_handle.rs b/alkstore/src/job_handle.rs new file mode 100644 index 0000000..3adb2aa --- /dev/null +++ b/alkstore/src/job_handle.rs @@ -0,0 +1,57 @@ +//! The boxed job handle a claim returns (ADR-019 §3; ADR-010 §2). +//! +//! One-shot/repeatable split, deliberate — do not "fix" it at +//! implementation: `ack`/`retry`/`fail` consume the handle +//! (`self: Box` — each is a terminal transition of the claim); +//! `heartbeat` takes `&self` and is repeatable (renewal is inherently +//! repeated within one claim). + +use crate::Result; +use crate::future::BoxedFuture; +use crate::job::Job; + +/// The claim handle: the row value plus its ops. +/// +/// # The uniform validity predicate +/// +/// Every op succeeds only while the row is `processing` **and** the +/// caller's claim deadline is unexpired (ADR-010 §2) — deadline lapse +/// refuses all ops (heartbeats, acks included) and a reclaim does +/// too. `false`, not an error: the dual-execution window's worker may +/// still complete its work, but its op will not land — at-least-once, +/// as documented. +pub trait JobHandle: Send { + /// The job row, as a value (the [`Job`] the claim returned). + fn job(&self) -> &Job; + + /// Ack: deletes the row (completed work leaves no row, ADR-010 + /// §1). `true` = the op landed under the validity predicate. + fn ack(self: Box) -> BoxedFuture<'static, Result>; + + /// Retry: return the row to `pending` for redelivery. `None` + /// delay = the queue's equal-jitter exponential curve computed + /// engine-side; `Some(d)` overrides it (ADR-010 §3). Exhausting + /// the attempt budget dead-letters instead. `false` = refused + /// under the validity predicate — the row keeps its pre-call + /// state (at-least-once redelivery governs). + fn retry( + self: Box, + err: Option, + delay: Option, + ) -> BoxedFuture<'static, Result>; + + /// Fail: immediate dead-letter — the "stop retrying this" + /// operator (ADR-010 §3). `None` err records the engine's + /// documented-default string `"failed"` (row content, not a + /// matched error variant). `false` = refused under the validity + /// predicate. + fn fail(self: Box, err: Option) -> BoxedFuture<'static, Result>; + + /// Heartbeat: renewal — an **absolute reset** of the claim + /// deadline from now (`extend` is the new full deadline in + /// seconds, *not* additive to elapsed time; ADR-010 §2). Repeatable + /// (`&self`); late heartbeat refused (`false`) — it can never + /// steal the job back from a reclaimer; a reclaim consumes an + /// attempt. + fn heartbeat<'a>(&'a self, extend: i64) -> BoxedFuture<'a, Result>; +} diff --git a/alkstore/src/lib.rs b/alkstore/src/lib.rs index 4dd3dce..fa2fb5f 100644 --- a/alkstore/src/lib.rs +++ b/alkstore/src/lib.rs @@ -6,29 +6,51 @@ //! this crate and implement its traits. mod error; +mod event_receiver; +mod future; mod job; +mod job_handle; +mod lock; mod opts; +mod outbox; mod payload; +mod queue; mod schedule; mod stop_token; +mod store; mod stream; +mod stream_handle; +mod tx; mod validation; mod wake; +mod wake_receiver; pub use error::{Error, Result}; +pub use event_receiver::EventReceiver; +pub use future::BoxedFuture; pub use job::{Job, JobState}; +pub use job_handle::JobHandle; +pub use lock::Lock; pub use opts::{EnqueueOpts, QueueOpts, ScheduleOpts}; +pub use outbox::{Delivery, Outbox}; pub use payload::{encode_payload, payload_as}; +pub use queue::Queue; pub use schedule::Schedule; pub use stop_token::StopToken; +pub use store::{Store, WithTxClosure}; pub use stream::StreamEvent; +pub use stream_handle::StreamHandle; +pub use tx::TxHandle; pub use validation::{ NameKind, RESERVED_LISTENER_RECONNECTED, RESERVED_PREFIX, validate_local_name, validate_name, validate_shared_name, }; pub use wake::Wake; +pub use wake_receiver::WakeReceiver; #[cfg(test)] mod tests; #[cfg(test)] +mod trait_probe_tests; +#[cfg(test)] mod value_type_tests; diff --git a/alkstore/src/lock.rs b/alkstore/src/lock.rs new file mode 100644 index 0000000..8f454b1 --- /dev/null +++ b/alkstore/src/lock.rs @@ -0,0 +1,26 @@ +//! The named-lock handle (ADR-019 §1; ADR-008 §7's locks guarantee row). +//! +//! Lock ops are not part of the tx seam — there is no `lock_tx`; +//! acquisition is a separate auto-commit op and the TTL discipline +//! governs. + +use crate::Result; +use crate::future::BoxedFuture; + +/// A held named lock. +pub trait Lock: Send { + /// The lock name the handle was constructed with (shared + /// namespace — validated at the constructor). + fn name(&self) -> &str; + + /// Renew: a new **full** TTL window from now. `false` = lost it + /// (expired and re-acquired elsewhere — exclusivity lapses + /// silently on expiry; no revocation event; ADR-008 §7). Holders + /// of long work must renew within TTL. + fn renew<'a>(&'a self, ttl: i64) -> BoxedFuture<'a, Result>; + + /// Release the lock, consuming the handle (RAII shape). `true` = + /// it was still held (released now); `false` = already expired or + /// re-acquired — releasing then is a no-op, not an error. + fn release(self: Box) -> BoxedFuture<'static, Result>; +} diff --git a/alkstore/src/outbox.rs b/alkstore/src/outbox.rs new file mode 100644 index 0000000..8c111fa --- /dev/null +++ b/alkstore/src/outbox.rs @@ -0,0 +1,56 @@ +//! The outbox helper handle (ADR-019 §1; the outbox section of +//! core-contract.md). +//! +//! A helper over queues, not a separate mechanism: the tx-side +//! transactional enqueue is `TxHandle::outbox_enqueue_tx` (ADR-014); +//! this handle carries the auto-commit enqueue and the delivery +//! worker entry point. + +use serde_json::Value; + +use crate::Result; +use crate::future::BoxedFuture; +use crate::job_handle::JobHandle; +use crate::opts::EnqueueOpts; + +/// The delivery callable [`Outbox::run_once`] takes. +/// +/// A consumer-supplied async callable taking [`Box`], +/// returning the boxed future of `Result<()>` (ADR-019 §5). Object +/// safety encoded as a named one-method trait object. Outcome +/// posture: `Ok` ⇒ the engine acks the claim; `Err(e)` ⇒ the engine +/// retries with the queue's curve (`retry(err, None)`). +/// +/// No engine-issued heartbeat inside delivery — the boxed +/// [`JobHandle`]'s own `heartbeat` remains available to the closure; +/// renewal inside the deadline is the consumer's obligation. +pub trait Delivery: Send { + /// Process the claimed job's work. Returns the boxed future of + /// `Result<()>`. + fn deliver<'a>(&'a mut self, job: Box) -> BoxedFuture<'a, Result<()>>; +} + +/// The outbox-scoped handle for one outbox name. +/// +/// The backing queue is derived engine-side under the reserved prefix +/// (`__alkstore_outbox:{name}`) — reachable only through this surface +/// and `outbox_enqueue_tx`; enqueue stamps the outbox's derived +/// `QueueOpts` set (60/5/5, ADR-010 §3). +pub trait Outbox: Send + Sync { + /// The outbox name the handle was constructed with. + fn name(&self) -> &str; + + /// Enqueue into the derived backing queue (auto-commit; the + /// transactional path is `TxHandle::outbox_enqueue_tx`). + fn enqueue<'a>(&'a self, payload: Value, opts: EnqueueOpts) -> BoxedFuture<'a, Result>; + + /// Pull op: claim one job, run `delivery`, ack on `Ok`, + /// `retry(err, None)` (the queue curve) on `Err`. The consumer + /// calls it in its own loop. `true` = claimed and processed; + /// `false` = no work (a value, not an error). + fn run_once<'a>( + &'a mut self, + worker_id: &str, + delivery: &'a mut dyn Delivery, + ) -> BoxedFuture<'a, Result>; +} diff --git a/alkstore/src/queue.rs b/alkstore/src/queue.rs new file mode 100644 index 0000000..4151062 --- /dev/null +++ b/alkstore/src/queue.rs @@ -0,0 +1,74 @@ +//! The queue mechanism handle (ADR-019 §1, §2; ADR-010 §1–§5). +//! +//! Claim/maintenance ops live on the queue-scoped handle — placement +//! mirroring the honker-rs starting artifact. There is deliberately no +//! `claim_tx`: a claim's visibility deadline must not be tied to a +//! business transaction's lifetime (ADR-019 §1). + +use serde_json::Value; + +use crate::Result; +use crate::future::BoxedFuture; +use crate::job::Job; +use crate::job_handle::JobHandle; +use crate::opts::EnqueueOpts; + +/// The queue-scoped handle: enqueue, claims, and maintenance ops for +/// one queue name. +/// +/// Queues are names, not registered objects: the handle carries the +/// name (validated at the constructor — shared-namespace rules) and +/// the `QueueOpts` stamps resolve per enqueue, never against a live +/// registry (ADR-010 §3a). +pub trait Queue: Send + Sync { + /// The queue name the handle was constructed with. + fn name(&self) -> &str; + + /// Enqueue a job (auto-commit; each op atomic alone) with the + /// handle's `QueueOpts` stamping the row (ADR-020 §3: the one + /// handle-carrying shape resolves over this handle's opts). + fn enqueue<'a>(&'a self, payload: Value, opts: EnqueueOpts) -> BoxedFuture<'a, Result>; + + /// Claim one claimable row — exactly-once handout under + /// concurrency; `None` = no work (a value, not an error). + /// Claim ordering: `priority` DESC, then ready-time, then enqueue + /// order (ADR-010 §1). `worker_id` is the claimant identity — + /// consumer-local, non-empty only (`InvalidName`). + fn claim_one<'a>( + &'a self, + worker_id: &str, + ) -> BoxedFuture<'a, Result>>>; + + /// Claim up to `n` rows, same ordering and exclusivity as + /// [`Queue::claim_one`]. + fn claim_batch<'a>( + &'a self, + worker_id: &str, + n: i64, + ) -> BoxedFuture<'a, Result>>>; + + /// Batch ack: applies the uniform validity predicate per id + /// (processing state + unexpired claim deadline); lapsed, + /// non-claimed, acked-elsewhere, or cancelled ids silently not + /// counted. Returns the count that landed; partial success is + /// ordinary (ADR-010 §1). + fn ack_batch<'a>(&'a self, ids: &[i64]) -> BoxedFuture<'a, Result>; + + /// Unconditionally delete a row in either `pending` or + /// `processing`, regardless of which worker holds it. Not an + /// interrupt — the holder's next handle op returns `false`, the + /// same shape as deadline expiry. `false` = no such live row. + fn cancel<'a>(&'a self, job_id: i64) -> BoxedFuture<'a, Result>; + + /// Get a job row — sees dead rows too (with `last_error`/ + /// `died_at`; ADR-010 §4). A pure data read: the returned `Job` + /// carries no ops (no ghost handles; a claim's `Box` + /// is the only ops carrier, ADR-019 §3). + fn get_job<'a>(&'a self, job_id: i64) -> BoxedFuture<'a, Result>>; + + /// Sweep this queue: move every past-`expires_at` row (any state — + /// pending and processing alike, the no-stranded-rows property, + /// ADR-010 §5) to dead, and delete dead rows past their retention + /// stamp. Returns the count moved. No leader lock required. + fn sweep_expired<'a>(&'a self) -> BoxedFuture<'a, Result>; +} diff --git a/alkstore/src/store.rs b/alkstore/src/store.rs new file mode 100644 index 0000000..e36f8cc --- /dev/null +++ b/alkstore/src/store.rs @@ -0,0 +1,134 @@ +//! The `Store` trait — the unified engine-agnostic surface (ADR-008 +//! §1/§6, ADR-019 §1, ADR-009 §1). +//! +//! Constructors and their option structs live in the engine crates +//! (ADR-008 §6): the engine's `open` returns this trait, and a store +//! handed out by an engine crate honors the full v1 surface. Consumer +//! code never branches on engine type. + +use serde_json::Value; + +use crate::Result; +use crate::future::BoxedFuture; +use crate::lock::Lock; +use crate::opts::{QueueOpts, ScheduleOpts}; +use crate::outbox::Outbox; +use crate::queue::Queue; +use crate::schedule::Schedule; +use crate::stop_token::StopToken; +use crate::stream_handle::StreamHandle; +use crate::tx::TxHandle; +use crate::wake_receiver::WakeReceiver; + +/// The pinned closure shape `with_tx` takes (ADR-007): a +/// `FnOnce(&mut dyn TxHandle) -> BoxedFuture>` — dyn-encoded +/// for object safety, `Send` because the transaction's future must be. +pub type WithTxClosure<'a> = + Box BoxedFuture<'_, Result<()>> + Send + 'a>; + +/// The store handle: one object, every mechanism behind it. +/// +/// # Entry-point validation obligation +/// +/// Every name-bearing method validates its names at the entry point — +/// before any engine round trip, on auto-commit paths (and the tx +/// path via [`TxHandle`], whose docs carry the same obligation): +/// non-empty otherwise `InvalidName`; reserved-prefix (`__alkstore_`) +/// rejected with `ReservedName` for the shared-namespace kinds +/// (channels, streams, queues, outboxes, locks, schedule names, and +/// `schedule()`'s queue argument). Consumer-local identifiers +/// (`try_lock`'s `owner`, stream consumers, `worker_id`) take the +/// non-empty rule only. Implemented engines share the helpers from +/// `validation`. +pub trait Store: Send + Sync { + /// Begin a transaction — the caller-held [`TxHandle`] whose `*_tx` + /// methods are commit-atomic with a business write. Dropping the + /// handle without `commit` rolls it back (ADR-021 §4). + fn begin_tx(&self) -> BoxedFuture<'_, Result>>; + + /// Fire-and-forget notify on `channel`. No durability, no replay; + /// a listener attached after a commit never sees it. Payload limit + /// pg-side (8000 bytes, client-side checked — `PayloadTooLarge` + /// before any round trip); no limit on SQLite. + fn notify<'a>(&'a self, channel: &str, payload: Value) -> BoxedFuture<'a, Result<()>>; + + /// Listen on `channel` — starts from "now" (no replay); delivers + /// opaque [`Wake`](crate::Wake) hints (channel only, never + /// payloads). + fn listen<'a>(&'a self, channel: &str) -> BoxedFuture<'a, Result>>; + + /// Take the stream-scoped handle for `name`. + fn stream<'a>(&'a self, name: &str) -> BoxedFuture<'a, Result>>; + + /// Take the queue-scoped handle for `name`, with the `QueueOpts` + /// stamps future enqueues resolve over (ADR-010 §3a). + fn queue<'a>(&'a self, name: &str, opts: QueueOpts) -> BoxedFuture<'a, Result>>; + + /// Take the outbox-scoped handle for `name` (backing queue derived + /// engine-side under the reserved prefix). + fn outbox<'a>(&'a self, name: &str) -> BoxedFuture<'a, Result>>; + + /// Acquire a named TTL lock, or fail: `None` = someone holds it + /// (no-work is a value, not an error). `owner` is consumer-local + /// (non-empty only); `ttl` is seconds. Exclusion lapses silently + /// at TTL expiry — holders renew within TTL (ADR-008 §7). + fn try_lock<'a>( + &'a self, + name: &str, + owner: &str, + ttl: i64, + ) -> BoxedFuture<'a, Result>>>; + + /// Register (upsert by name) a schedule firing into `queue`. + /// Both the schedule name and the queue argument validate as + /// shared-namespace names at the entry point (ADR-021 §3 — a + /// schedule cannot target a reserved name). `spec` grammar v1: + /// `@every ` only (s|m|h|d) — else `InvalidSpec` + /// (ADR-009 §2), rejected before storage. Schedules never fire + /// without a consumer running [`Store::run_schedules`]. + fn schedule<'a>( + &'a self, + name: &str, + spec: &str, + queue: &str, + payload: Value, + opts: ScheduleOpts, + ) -> BoxedFuture<'a, Result>; + + /// Remove a schedule registration. `true` = a row was removed; + /// `false` = no such schedule (a value, not an error). + fn unschedule<'a>(&'a self, name: &str) -> BoxedFuture<'a, Result>; + + /// The opt-in scheduler runner: acquire the leadership lock (the + /// reserved name `__alkstore_scheduler`), loop {renew; fire due + /// boundaries; sleep until the next boundary or `stop`}. A + /// cancelled [`StopToken`] ends the sleep and returns the clean + /// `Ok(())`; leadership loss returns `Err(LeadershipLost)` — the + /// caller's recipe is to respawn (ADR-009 §1). + fn run_schedules<'a>(&'a self, stop: StopToken) -> BoxedFuture<'a, Result<()>>; + + /// Closure-scoped transaction — a wrapper over [`Store::begin_tx`] + /// (provided method; ADR-007, signature pinned 2026-10-07). + /// + /// Deliberately non-generic (a `Store` trait object must stay + /// object-safe — ADR-008 §2): the pinned `FnOnce(&mut dyn TxHandle) + /// -> BoxedFuture>` shape crosses dyn-encoded — as + /// `Box` — with values returning through the + /// closure's captured state; no handle escapes the closure. + /// + /// Disposition: `Ok` ⇒ commit; `Err`, drop, or panic across the + /// closure ⇒ rollback (ADR-021 §4's drop = rollback). `with_tx` + /// is the convenience; `begin_tx` is the seam. + fn with_tx<'a>(&'a self, f: WithTxClosure<'a>) -> BoxedFuture<'a, Result<()>> { + Box::pin(async move { + let mut handle = self.begin_tx().await?; + match Box::pin(f(&mut *handle)).await { + Ok(()) => handle.commit().await, + Err(e) => { + drop(handle); + Err(e) + } + } + }) + } +} diff --git a/alkstore/src/stream_handle.rs b/alkstore/src/stream_handle.rs new file mode 100644 index 0000000..caf00bb --- /dev/null +++ b/alkstore/src/stream_handle.rs @@ -0,0 +1,77 @@ +//! The stream mechanism handle (ADR-019 §1; ADR-015). +//! +//! Publish/read/trim ops live on the stream-scoped handle — placement +//! mirroring the honker-rs starting artifact; the transactional twins +//! ride the `TxHandle` (ADR-021 §1). + +use serde_json::Value; + +use crate::Result; +use crate::event_receiver::EventReceiver; +use crate::future::BoxedFuture; +use crate::stream::StreamEvent; + +/// The stream-scoped handle: publish, cursor reads, offsets, trim, and +/// the durable subscription. +pub trait StreamHandle: Send + Sync { + /// The stream name the handle was constructed with. + fn name(&self) -> &str; + + /// Append an event to the stream's durable log (auto-commit). + /// Returns the assigned offset — engine-assigned, monotone per + /// stream, immutable. + fn publish<'a>(&'a self, payload: Value) -> BoxedFuture<'a, Result>; + + /// Append a keyed event — [`StreamHandle::publish`]'s keyed twin. + /// The key is carried metadata (stored, round-tripped, no + /// engine-enforced behavioral role); the ordering guarantee stays + /// global FIFO by offset per stream (ADR-015 §1). A present key + /// must be non-empty (`InvalidName`). Returns the assigned + /// offset. + fn publish_with_key<'a>( + &'a self, + key: Option, + payload: Value, + ) -> BoxedFuture<'a, Result>; + + /// Cursor read: events with `offset > ` cursor, `offset ASC` + /// (global FIFO — ADR-015 §4), up to `limit`. A read from a + /// trimmed-away region resumes at the trim horizon's first + /// remaining row. + fn read_since<'a>( + &'a self, + offset: i64, + limit: i64, + ) -> BoxedFuture<'a, Result>>; + + /// Cursor read from a consumer's stored checkpoint (absent + /// consumer = 0). + fn read_from_consumer<'a>( + &'a self, + consumer: &str, + limit: i64, + ) -> BoxedFuture<'a, Result>>; + + /// Save a consumer's checkpoint — always explicit (no + /// auto-checkpoint cadence). Monotone: a save below the stored + /// checkpoint is a silent no-op (ADR-019 §6); the `Err` arm + /// carries `Database` only (ADR-021 §5). + fn save_offset<'a>(&'a self, consumer: &str, offset: i64) -> BoxedFuture<'a, Result<()>>; + + /// Inspect a consumer's stored checkpoint; absent consumer = 0 + /// (the pre-save state, ADR-019 §1). + fn get_offset<'a>(&'a self, consumer: &str) -> BoxedFuture<'a, Result>; + + /// Bounded growth, consumer-invoked: delete events with + /// `offset <= horizon` (exact boundary; ADR-015 §5). Surviving + /// events keep their offsets (gaps legal, never renumbered); trim + /// emits no wake. Returns the deleted-row count. + fn trim_to<'a>(&'a self, horizon: i64) -> BoxedFuture<'a, Result>; + + /// Durable consumption: attach a consumer that reads to the + /// current tail, resumes after restart from the stored offset, + /// and saves checkpoints explicitly (no `save_every`, no + /// save-on-drop). `consumer` is a consumer-local identifier + /// (non-empty only). + fn subscribe<'a>(&'a self, consumer: &str) -> BoxedFuture<'a, Result>>; +} diff --git a/alkstore/src/tests.rs b/alkstore/src/tests.rs index c1c3682..ebf7e00 100644 --- a/alkstore/src/tests.rs +++ b/alkstore/src/tests.rs @@ -129,3 +129,15 @@ fn reserved_constants_exact() { "__alkstore_listener_reconnected__" ); } + +#[test] +fn scheduler_error_variants() { + let e = Error::InvalidSpec { + spec: "0 0 * * *".into(), + }; + assert!(e.to_string().contains("0 0 * * *")); + assert_eq!( + Error::LeadershipLost.to_string(), + "scheduler leadership lost" + ); +} diff --git a/alkstore/src/trait_probe_tests.rs b/alkstore/src/trait_probe_tests.rs new file mode 100644 index 0000000..94c3744 --- /dev/null +++ b/alkstore/src/trait_probe_tests.rs @@ -0,0 +1,397 @@ +//! Object-safety compile probes and signature-conformance checks for +//! the trait surface (the task's acceptance criterion): a trivial +//! local impl of every trait must construct as `Box` — +//! proving the whole surface is object-safe and shape-complete. + +use std::time::Duration; + +use serde_json::{Value, json}; + +use crate::event_receiver::EventReceiver; +use crate::job::Job; +use crate::job_handle::JobHandle; +use crate::lock::Lock; +use crate::outbox::{Delivery, Outbox}; +use crate::queue::Queue; +use crate::stream::StreamEvent; +use crate::stream_handle::StreamHandle; +use crate::tx::TxHandle; +use crate::wake_receiver::WakeReceiver; +use crate::{ + BoxedFuture, EnqueueOpts, Error, QueueOpts, Result, Schedule, ScheduleOpts, StopToken, Store, + Wake, +}; + +#[derive(Default)] +struct ProbeTx; + +impl TxHandle for ProbeTx { + fn enqueue_tx<'a>( + &'a mut self, + _queue: &str, + _opts: EnqueueOpts, + _payload: Value, + ) -> BoxedFuture<'a, Result> { + Box::pin(std::future::ready(Ok(1))) + } + fn publish_tx<'a>( + &'a mut self, + _stream: &str, + _payload: Value, + ) -> BoxedFuture<'a, Result> { + Box::pin(std::future::ready(Ok(1))) + } + fn publish_with_key_tx<'a>( + &'a mut self, + _stream: &str, + _key: Option, + _payload: Value, + ) -> BoxedFuture<'a, Result> { + Box::pin(std::future::ready(Ok(1))) + } + fn notify_tx<'a>(&'a mut self, _channel: &str, _payload: Value) -> BoxedFuture<'a, Result<()>> { + Box::pin(std::future::ready(Ok(()))) + } + fn save_offset_tx<'a>( + &'a mut self, + _stream: &str, + _consumer: &str, + _offset: i64, + ) -> BoxedFuture<'a, Result<()>> { + Box::pin(std::future::ready(Ok(()))) + } + fn get_job_tx<'a>( + &'a mut self, + _queue: &str, + _job_id: i64, + ) -> BoxedFuture<'a, Result>> { + Box::pin(std::future::ready(Ok(None))) + } + fn get_offset_tx<'a>( + &'a mut self, + _stream: &str, + _consumer: &str, + ) -> BoxedFuture<'a, Result> { + Box::pin(std::future::ready(Ok(0))) + } + fn read_since_tx<'a>( + &'a mut self, + _stream: &str, + _offset: i64, + _limit: i64, + ) -> BoxedFuture<'a, Result>> { + Box::pin(std::future::ready(Ok(Vec::new()))) + } + fn read_from_consumer_tx<'a>( + &'a mut self, + _stream: &str, + _consumer: &str, + _limit: i64, + ) -> BoxedFuture<'a, Result>> { + Box::pin(std::future::ready(Ok(Vec::new()))) + } + fn outbox_enqueue_tx<'a>( + &'a mut self, + _outbox: &str, + _opts: EnqueueOpts, + _payload: Value, + ) -> BoxedFuture<'a, Result> { + Box::pin(std::future::ready(Ok(1))) + } + fn commit(self: Box) -> BoxedFuture<'static, Result<()>> { + Box::pin(std::future::ready(Ok(()))) + } +} + +struct ProbeJob; + +impl JobHandle for ProbeJob { + fn job(&self) -> &Job { + unreachable!() + } + fn ack(self: Box) -> BoxedFuture<'static, Result> { + Box::pin(std::future::ready(Ok(true))) + } + fn retry( + self: Box, + _err: Option, + _delay: Option, + ) -> BoxedFuture<'static, Result> { + Box::pin(std::future::ready(Ok(true))) + } + fn fail(self: Box, _err: Option) -> BoxedFuture<'static, Result> { + Box::pin(std::future::ready(Ok(true))) + } + fn heartbeat<'a>(&'a self, _extend: i64) -> BoxedFuture<'a, Result> { + Box::pin(std::future::ready(Ok(true))) + } +} + +struct ProbeQueue; + +impl Queue for ProbeQueue { + fn name(&self) -> &str { + "probe" + } + fn enqueue<'a>(&'a self, _payload: Value, _opts: EnqueueOpts) -> BoxedFuture<'a, Result> { + Box::pin(std::future::ready(Ok(1))) + } + fn claim_one<'a>( + &'a self, + _worker_id: &str, + ) -> BoxedFuture<'a, Result>>> { + Box::pin(std::future::ready(Ok(None))) + } + fn claim_batch<'a>( + &'a self, + _worker_id: &str, + _n: i64, + ) -> BoxedFuture<'a, Result>>> { + Box::pin(std::future::ready(Ok(Vec::new()))) + } + fn ack_batch<'a>(&'a self, _ids: &[i64]) -> BoxedFuture<'a, Result> { + Box::pin(std::future::ready(Ok(0))) + } + fn cancel<'a>(&'a self, _job_id: i64) -> BoxedFuture<'a, Result> { + Box::pin(std::future::ready(Ok(false))) + } + fn get_job<'a>(&'a self, _job_id: i64) -> BoxedFuture<'a, Result>> { + Box::pin(std::future::ready(Ok(None))) + } + fn sweep_expired<'a>(&'a self) -> BoxedFuture<'a, Result> { + Box::pin(std::future::ready(Ok(0))) + } +} + +struct ProbeStream; + +impl StreamHandle for ProbeStream { + fn name(&self) -> &str { + "probe" + } + fn publish<'a>(&'a self, _payload: Value) -> BoxedFuture<'a, Result> { + Box::pin(std::future::ready(Ok(1))) + } + fn publish_with_key<'a>( + &'a self, + _key: Option, + _payload: Value, + ) -> BoxedFuture<'a, Result> { + Box::pin(std::future::ready(Ok(1))) + } + fn read_since<'a>( + &'a self, + _offset: i64, + _limit: i64, + ) -> BoxedFuture<'a, Result>> { + Box::pin(std::future::ready(Ok(Vec::new()))) + } + fn read_from_consumer<'a>( + &'a self, + _consumer: &str, + _limit: i64, + ) -> BoxedFuture<'a, Result>> { + Box::pin(std::future::ready(Ok(Vec::new()))) + } + fn save_offset<'a>(&'a self, _consumer: &str, _offset: i64) -> BoxedFuture<'a, Result<()>> { + Box::pin(std::future::ready(Ok(()))) + } + fn get_offset<'a>(&'a self, _consumer: &str) -> BoxedFuture<'a, Result> { + Box::pin(std::future::ready(Ok(0))) + } + fn trim_to<'a>(&'a self, _horizon: i64) -> BoxedFuture<'a, Result> { + Box::pin(std::future::ready(Ok(0))) + } + fn subscribe<'a>(&'a self, _consumer: &str) -> BoxedFuture<'a, Result>> { + Box::pin(std::future::ready(Ok( + Box::new(ProbeEventReceiver) as Box + ))) + } +} + +struct ProbeEventReceiver; + +impl EventReceiver for ProbeEventReceiver { + fn recv<'a>(&'a mut self) -> BoxedFuture<'a, Option>> { + Box::pin(std::future::ready(None)) + } + fn try_recv(&mut self) -> Result> { + Ok(None) + } + fn read_since<'a>( + &'a mut self, + _offset: i64, + _limit: i64, + ) -> BoxedFuture<'a, Result>> { + Box::pin(std::future::ready(Ok(Vec::new()))) + } + fn save_offset(&mut self) -> Result<()> { + Ok(()) + } + fn offset(&self) -> i64 { + 0 + } +} + +struct ProbeOutbox; + +impl Outbox for ProbeOutbox { + fn name(&self) -> &str { + "probe" + } + fn enqueue<'a>(&'a self, _payload: Value, _opts: EnqueueOpts) -> BoxedFuture<'a, Result> { + Box::pin(std::future::ready(Ok(1))) + } + fn run_once<'a>( + &'a mut self, + _worker_id: &str, + _delivery: &'a mut dyn Delivery, + ) -> BoxedFuture<'a, Result> { + Box::pin(std::future::ready(Ok(false))) + } +} + +struct ProbeLock; + +impl Lock for ProbeLock { + fn name(&self) -> &str { + "probe" + } + fn renew<'a>(&'a self, _ttl: i64) -> BoxedFuture<'a, Result> { + Box::pin(std::future::ready(Ok(true))) + } + fn release(self: Box) -> BoxedFuture<'static, Result> { + Box::pin(std::future::ready(Ok(true))) + } +} + +struct ProbeWakeReceiver; + +impl WakeReceiver for ProbeWakeReceiver { + fn recv<'a>(&'a mut self) -> BoxedFuture<'a, Option> { + Box::pin(std::future::ready(None)) + } + fn try_recv(&mut self) -> Result> { + Err(Error::Closed) + } + fn recv_timeout<'a>(&'a mut self, _timeout: Duration) -> BoxedFuture<'a, Result>> { + Box::pin(std::future::ready(Ok(None))) + } +} + +struct ProbeStore; + +impl Store for ProbeStore { + fn begin_tx(&self) -> BoxedFuture<'_, Result>> { + Box::pin(std::future::ready(Ok( + Box::new(ProbeTx) as Box + ))) + } + fn notify<'a>(&'a self, _channel: &str, _payload: Value) -> BoxedFuture<'a, Result<()>> { + Box::pin(std::future::ready(Ok(()))) + } + fn listen<'a>(&'a self, _channel: &str) -> BoxedFuture<'a, Result>> { + Box::pin(std::future::ready(Ok( + Box::new(ProbeWakeReceiver) as Box + ))) + } + fn stream<'a>(&'a self, _name: &str) -> BoxedFuture<'a, Result>> { + Box::pin(std::future::ready(Ok( + Box::new(ProbeStream) as Box + ))) + } + fn queue<'a>( + &'a self, + _name: &str, + _opts: QueueOpts, + ) -> BoxedFuture<'a, Result>> { + Box::pin(std::future::ready(Ok( + Box::new(ProbeQueue) as Box + ))) + } + fn outbox<'a>(&'a self, _name: &str) -> BoxedFuture<'a, Result>> { + Box::pin(std::future::ready(Ok( + Box::new(ProbeOutbox) as Box + ))) + } + fn try_lock<'a>( + &'a self, + _name: &str, + _owner: &str, + _ttl: i64, + ) -> BoxedFuture<'a, Result>>> { + Box::pin(std::future::ready(Ok(None))) + } + fn schedule<'a>( + &'a self, + _name: &str, + _spec: &str, + _queue: &str, + _payload: Value, + _opts: ScheduleOpts, + ) -> BoxedFuture<'a, Result> { + Box::pin(std::future::ready(Ok(Schedule { + name: "probe".into(), + spec: "@every 60s".into(), + queue: "q".into(), + opts: ScheduleOpts::default(), + }))) + } + fn unschedule<'a>(&'a self, _name: &str) -> BoxedFuture<'a, Result> { + Box::pin(std::future::ready(Ok(false))) + } + fn run_schedules<'a>(&'a self, _stop: StopToken) -> BoxedFuture<'a, Result<()>> { + Box::pin(std::future::ready(Ok(()))) + } +} + +#[test] +fn trait_objects_construct() { + let _tx: Box = Box::new(ProbeTx); + let _job: Box = Box::new(ProbeJob); + let _queue: Box = Box::new(ProbeQueue); + let _stream: Box = Box::new(ProbeStream); + let _receiver: Box = Box::new(ProbeEventReceiver); + let _outbox: Box = Box::new(ProbeOutbox); + let _lock: Box = Box::new(ProbeLock); + let _wake_receiver: Box = Box::new(ProbeWakeReceiver); + let _store: Box = Box::new(ProbeStore); +} + +#[tokio::test] +async fn with_tx_commit_and_rollback_dispositions() { + let store: Box = Box::new(ProbeStore); + + let committed = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); + let flag = committed.clone(); + store + .with_tx(Box::new(move |_tx| { + let flag = flag.clone(); + Box::pin(async move { + flag.store(true, std::sync::atomic::Ordering::SeqCst); + Ok(()) + }) as BoxedFuture<'_, Result<()>> + })) + .await + .unwrap_or_else(|e| panic!("with_tx commit failed: {e}")); + assert!(committed.load(std::sync::atomic::Ordering::SeqCst)); + + let rolled_back = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); + let flag = rolled_back.clone(); + let r = store + .with_tx(Box::new(move |tx| { + let flag = flag.clone(); + Box::pin(async move { + flag.store(true, std::sync::atomic::Ordering::SeqCst); + let _ = tx + .enqueue_tx("q", EnqueueOpts::default(), json!({})) + .await?; + Err::<(), _>(Error::InvalidName { + name: "probe".into(), + }) + }) as BoxedFuture<'_, Result<()>> + })) + .await; + assert!(matches!(r, Err(Error::InvalidName { .. }))); + assert!(rolled_back.load(std::sync::atomic::Ordering::SeqCst)); +} diff --git a/alkstore/src/tx.rs b/alkstore/src/tx.rs new file mode 100644 index 0000000..e0a0fa7 --- /dev/null +++ b/alkstore/src/tx.rs @@ -0,0 +1,144 @@ +//! The transactional seam's caller-held handle (ADR-007, ADR-008 §2, +//! ADR-014, ADR-015 §2, ADR-021 §1/§4). +//! +//! The `*_tx` methods make side effects commit-atomic with a business +//! write: every op lands in *the caller's transaction*. The trait is +//! core-owned — engines implement it for their concrete handle; no +//! downcast, no enum, no generic method (object safety, ADR-008 §2). + +use serde_json::Value; + +use crate::future::BoxedFuture; +use crate::opts::EnqueueOpts; +use crate::{Job, Result, StreamEvent}; + +/// The caller-held transaction handle. +/// +/// Engine shape notes, not contract: the Postgres handle holds the +/// pooled client directly (straight `.await`s); the SQLite handle is a +/// writer-slot lease whose ops round-trip `spawn_blocking` (ADR-003, +/// ADR-004). +/// +/// # Entry-point validation obligation +/// +/// Every name-bearing method (`enqueue_tx`, `publish_tx`, +/// `publish_with_key_tx`, `notify_tx`, `save_offset_tx`, +/// `get_job_tx`/`get_offset_tx`/`read_since_tx`/ +/// `read_from_consumer_tx`, `outbox_enqueue_tx`) validates its names +/// before any engine round trip — identically on the auto-commit and +/// tx paths: non-empty (else `InvalidName`), reserved-prefix rejected +/// for shared-namespace kinds (else `ReservedName`). Validation must +/// not be bypassable by the commit-atomic path (ADR-008 §4). +/// +/// # Drop = rollback +/// +/// A handle dropped without `commit` rolls back (ADR-021 §4): the +/// no-ghosts property holds through RAII paths, not only the explicit +/// ones. There is no `rollback` method — dropping *is* the rollback; +/// a caller wanting an explicit rollback drops the handle. +pub trait TxHandle: Send { + /// Enqueue a job into `queue`, commit-atomic with the caller's + /// business write. Returns the job id. + /// + /// No-handle-open shape: the row stamps the engine's derived + /// default `QueueOpts` (300/3/5/none, ADR-020 §3) — a `TxHandle` + /// carries no queue opts. Validates `queue` as a shared-namespace + /// name (`InvalidName`/`ReservedName`) before any round trip. + fn enqueue_tx<'a>( + &'a mut self, + queue: &str, + opts: EnqueueOpts, + payload: Value, + ) -> BoxedFuture<'a, Result>; + + /// Publish an event to `stream`, commit-atomic. Returns the + /// assigned offset. Validates `stream` as a shared-namespace name. + fn publish_tx<'a>(&'a mut self, stream: &str, payload: Value) -> BoxedFuture<'a, Result>; + + /// Publish a keyed event to `stream` — [`TxHandle::publish_tx`]'s + /// keyed twin; commit-atomic keyed publishes are the point + /// (ADR-015 §2). The key is carried metadata; a present key must + /// be non-empty (`InvalidName`, ADR-008 §5 annotated). Returns the + /// assigned offset. + fn publish_with_key_tx<'a>( + &'a mut self, + stream: &str, + key: Option, + payload: Value, + ) -> BoxedFuture<'a, Result>; + + /// Deliver a fire-and-forget notify, commit-atomic — delivered + /// only at commit; rollback drops it (ADR-007). Validates + /// `channel`; may fail with `PayloadTooLarge` pg-side + /// (client-side check before the round trip, the contract-wide + /// matchable engine asymmetry). + fn notify_tx<'a>(&'a mut self, channel: &str, payload: Value) -> BoxedFuture<'a, Result<()>>; + + /// Save a stream consumer's checkpoint, commit-atomic. Saves are + /// monotone — a save below the stored checkpoint is a silent + /// no-op (ADR-019 §6); the `Err` arm carries `Database` only + /// (ADR-021 §5). Validates `stream`; `consumer` is a + /// consumer-local identifier (non-empty only). + fn save_offset_tx<'a>( + &'a mut self, + stream: &str, + consumer: &str, + offset: i64, + ) -> BoxedFuture<'a, Result<()>>; + + /// Read a job row inside the caller's transaction — reads the + /// transaction's own writes (read-your-own-writes, ADR-021 §1). + /// Sees dead rows; a pure read, no ops ride it. Validates `queue`. + fn get_job_tx<'a>( + &'a mut self, + queue: &str, + job_id: i64, + ) -> BoxedFuture<'a, Result>>; + + /// Read a consumer's stored checkpoint inside the caller's + /// transaction; absent consumer = 0 (ADR-019 §1). Pure read. + /// Validates `stream`. + fn get_offset_tx<'a>( + &'a mut self, + stream: &str, + consumer: &str, + ) -> BoxedFuture<'a, Result>; + + /// Read stream events from `offset`, inside the caller's + /// transaction (`offset ASC`, global FIFO — ADR-015 §4). Pure + /// read. Validates `stream`. + fn read_since_tx<'a>( + &'a mut self, + stream: &str, + offset: i64, + limit: i64, + ) -> BoxedFuture<'a, Result>>; + + /// Read stream events from a consumer's stored checkpoint, inside + /// the caller's transaction. Pure read. Validates `stream`. + fn read_from_consumer_tx<'a>( + &'a mut self, + stream: &str, + consumer: &str, + limit: i64, + ) -> BoxedFuture<'a, Result>>; + + /// Enqueue into an outbox's derived backing queue, commit-atomic + /// (ADR-014). Takes the *outbox* name — validated like + /// `store.outbox(name)` — and derives the queue engine-side; the + /// derived name (reserved-prefixed) is unreachable by + /// [`TxHandle::enqueue_tx`] by design, so this is the only + /// transactional path into the backing queue. Stamps the outbox's + /// derived `QueueOpts` set (60/5/5). + fn outbox_enqueue_tx<'a>( + &'a mut self, + outbox: &str, + opts: EnqueueOpts, + payload: Value, + ) -> BoxedFuture<'a, Result>; + + /// Commit the transaction, consuming the handle. The alternative + /// disposition is drop = rollback. One-shot: callers re-`begin_tx` + /// for the next transaction. + fn commit(self: Box) -> BoxedFuture<'static, Result<()>>; +} diff --git a/alkstore/src/wake_receiver.rs b/alkstore/src/wake_receiver.rs new file mode 100644 index 0000000..ac2ddeb --- /dev/null +++ b/alkstore/src/wake_receiver.rs @@ -0,0 +1,38 @@ +//! The wake receiver (ADR-008 §3, §8): the `listen(channel)` return. +//! +//! Wakes are best-effort hints, not per-commit delivery promises — +//! possibly coalesced (SQLite) or per-notify (Postgres), possibly +//! repeated after reconnect, possibly absent entirely — recovery is +//! the consumer's re-read, never a guaranteed deliver. Consumers must +//! be idempotent on wake. + +use std::time::Duration; + +use crate::Result; +use crate::future::BoxedFuture; +use crate::wake::Wake; + +/// The wake receiver: the durable-subscription-free listen side of +/// the notify mechanism. +/// +/// Close semantics (ADR-008 §3): `recv() -> None` means closed +/// (source death — SQLite's watcher death; terminal). The synthetic +/// reconnect-wake (the reserved channel) covers Postgres connection +/// gaps and arrives as a normal [`Wake`]. +pub trait WakeReceiver: Send { + /// Wait for the next wake. `None` = the source closed (terminal; + /// a closed receiver never reopens — recovery is a fresh + /// `listen`). + fn recv<'a>(&'a mut self) -> BoxedFuture<'a, Option>; + + /// Poll without waiting. `Ok(None)` = no wake pending right now + /// (idle); `Err(Closed)` = the source closed. Closed and idle are + /// distinguishable states. + fn try_recv(&mut self) -> Result>; + + /// Bounded wait for the next wake. Timeout expiry without a wake + /// returns `Ok(None)` — idle is the documented `Ok(None)` arm; + /// the `Err` arm carries `Database` (storage failure) or `Closed` + /// (source death), never a timeout signal (ADR-008 §3). + fn recv_timeout<'a>(&'a mut self, timeout: Duration) -> BoxedFuture<'a, Result>>; +} diff --git a/tasks/core-trait-surface.md b/tasks/core-trait-surface.md index 26b2aba..d52cac2 100644 --- a/tasks/core-trait-surface.md +++ b/tasks/core-trait-surface.md @@ -1,7 +1,7 @@ --- id: core-trait-surface name: Core trait surface (Store, TxHandle, mechanism handles, receivers) -status: pending +status: completed depends_on: [core-value-types] scope: moderate risk: medium @@ -87,17 +87,17 @@ drop=rollback, no-claim_tx absence rationale). ## Acceptance Criteria -- [ ] Every trait above with the exact pinned method sets and +- [x] Every trait above with the exact pinned method sets and signatures; compiles as object-safe (`Box` everywhere the contract says) -- [ ] `with_tx` provided method present with the pinned signature -- [ ] A compile-probe test constructs `Box`, +- [x] `with_tx` provided method present with the pinned signature +- [x] A compile-probe test constructs `Box`, `Box`, etc. from a trivial local impl, proving object safety of the whole surface -- [ ] Trait docs state: entry-point validation obligation, drop = +- [x] Trait docs state: entry-point validation obligation, drop = rollback, the uniform handle-op validity predicate reference, heartbeat's absolute-reset semantics, no `claim_tx` rationale -- [ ] `cargo test -p alkstore`, clippy `-D warnings`, fmt clean +- [x] `cargo test -p alkstore`, clippy `-D warnings`, fmt clean ## References @@ -110,8 +110,73 @@ drop=rollback, no-claim_tx absence rationale). ## Notes -> To be filled by implementation agent +- **Async posture**: the "desugared boxed-future form" (ADR-008 §2) is + realized as a named alias, `BoxedFuture<'a, T> = + Pin + Send + 'a>>` in + `src/future.rs` — no `async_trait` macro, no core dependency beyond + the pinned posture (Send bounds the futures: `begin_tx` returns + `Box`, so its futures must be equally + portable). Sync methods (`try_recv`/`save_offset`/`offset` on + receivers, `job(&self)`, `name(&self)`) stay sync as pinned. +- **`with_tx` dyn encoding**: the pinned signature is + non-generic — a `Store` trait object must stay object-safe — so the + generic-`F` form does not compile on the trait; the pinned + `FnOnce(&mut dyn TxHandle) -> BoxedFuture>` shape crosses + dyn-encoded as `Box`, aliased `WithTxClosure<'a>` + (re-exported) to keep the complex type out of signatures. Values + return through captured state; the provided default body commits on + `Ok`, drops (rolls back) on `Err`. +- **Delivery callable encoding** (ADR-019 §5's implementer's choice): + the named one-method trait object — `trait Delivery { fn + deliver(&mut self, Box) -> BoxedFuture>; + }` — with `run_once(worker_id, delivery: &mut dyn Delivery)`. + `&mut dyn FnMut` cannot return boxed futures of borrowed data under + object safety, so the named-trait form is the workable answer; the + engine acks on the closure's `Ok` and retries on its `Err`. +- **Fallibility**: the ADR sketches elide `Result<>` for brevity + (ADR-008 §2's note: every fallible op returns the §5 taxonomy); + entry points validating names must err, so constructors + (`stream`/`queue`/`outbox`/`listen`) are `Result`-returning as is + `notify` (PayloadTooLarge); `begin_tx` returns `Result` (SQLite + writer-slot acquisition can fail); no-claim and no-lock outcomes + stay values (`Option`), handle-op outcomes stay `bool`s per + ADR-008 §5's act-differently rule. +- **Scheduler error variants** (`InvalidSpec { spec }`, + `LeadershipLost` — ADR-009 §6) joined `Error` here per + `core-errors-and-validation`'s note (exactly-six list was that + task's scope; their signatures pin in this task). Unit-tested. +- **`tokio` added as a *dev*-dependency only** (test to drive the + boxed-future probes) — runtime deps remain exactly + serde/serde_json/thiserror (ADR-020 §4, ADR-001: no driver deps in + core). +- `Wake`'s `#[non_exhaustive]` prevented the probe's struct-literal + construction — the probe re-exports or constructs nothing of the + read types beyond the existing value-type tests; trait-object + construction is the probe's whole point. +- Every trait doc carries the pinned semantics text: entry-point + validation obligations (Store; TxHandle's own mirror — name-bearing + `*_tx` methods validate like their auto-commit counterparts), + drop=rollback (TxHandle), no-`claim_tx` rationale (Queue), the + uniform validity predicate + absolute-reset heartbeat (JobHandle), + monotone saves (StreamHandle/EventReceiver), terminal-close arms + and the deliberate no-`recv_timeout` absence (receivers). ## Summary -> To be filled on completion \ No newline at end of file +The core trait surface — contract v1 in code — is implemented and +re-exported from `alkstore/src/lib.rs`: `BoxedFuture` + +`TxHandle` (the eleven pinned methods incl. ADR-021 §1's four tx +reads, `commit(self: Box)`, drop=rollback docs); +`Store` (the pinned method set + the `with_tx` provided method in the +pinned dyn-encoded signature, with commit/rollback dispositions — +unit-tested); mechanism handles `Queue`/`JobHandle`/`StreamHandle`/ +`Outbox` (+ `Delivery`)/`Lock` with the exact ADR-019 §1/§3/§5 method +sets; receivers `WakeReceiver` (timeout expiry = `Ok(None)`; `Err` = +Database/Closed only) and `EventReceiver` (no `recv_timeout`, +deliberate); `Error::InvalidSpec`/`LeadershipLost` added (ADR-009 +§6). Object-safety compile probes construct every trait object from +trivial local impls (`trait_objects_construct`), and the `with_tx` +dispositions run against a probe store (`with_tx_commit_and_ +rollback_dispositions`) — 23 tests green; `cargo test -p alkstore`, +workspace `cargo build`, clippy `-D warnings` (workspace), and +`cargo fmt --check` all clean. \ No newline at end of file