Core value types: opts structs, Job/JobState, Schedule, StreamEvent, Wake, StopToken, payload encode/decode (ADR-008 §1/§3, ADR-010 §3, ADR-015 §3, ADR-017 §3, ADR-019 §3/§4, ADR-020 §1–§4, ADR-021 §2)

alkstore/Cargo.toml

@@ -6,4 +6,6 @@ license.workspace = true
 repository.workspace = true
 
 [dependencies]
+serde = "1"
+serde_json = "1"
 thiserror = "2"

alkstore/src/job.rs

@@ -0,0 +1,91 @@
+//! Queue value types (ADR-019 §3, ADR-021 §2): `Job` and `JobState`.
+//!
+//! `#[non_exhaustive]` per ADR-017 §3 — consumers read but never
+//! construct; field additions ride ADR-017 class 2 (semver-minor).
+
+use serde::de::DeserializeOwned;
+
+use crate::payload;
+
+/// A job's lifecycle state. Dead-letter is a move to dead storage
+/// (ADR-010 §4); `get_job` sees dead rows with `last_error`/`died_at`.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[non_exhaustive]
+pub enum JobState {
+    /// Enqueued, awaiting its ready time or a claim.
+    Pending,
+    /// Currently claimed and running (or within its claim deadline).
+    Processing,
+    /// Dead-lettered.
+    Dead,
+}
+
+/// A job row, as returned by `get_job` and carried inside a claim's
+/// boxed `JobHandle` (`job(&self) -> &Job`).
+///
+/// `#[non_exhaustive]` (ADR-017 §3). The resolved values — not the raw
+/// enqueue fields — are what a `Job` shows: the ready time after
+/// delay-over-`run_at` resolution, the absolute `expires_at` after
+/// relative-`expires` resolution (ADR-020 §1–§2), and the `QueueOpts`
+/// stamps attached at enqueue (ADR-010 §3a).
+///
+/// Dead rows carry their stamps too (`get_job` returns the stamps with
+/// the row); `last_error`/`died_at` are dead-only (`None` on live
+/// rows).
+#[derive(Debug, Clone, PartialEq, Eq)]
+#[non_exhaustive]
+pub struct Job {
+    /// The row id.
+    pub id: i64,
+    /// The queue name the job was enqueued into.
+    pub queue: String,
+    /// The lifecycle state.
+    pub state: JobState,
+    /// Payload bytes — the serde_json serialization of the trait
+    /// `Value` at enqueue time (ADR-020 §4); decode with
+    /// [`Job::payload_as`].
+    pub payload: Vec<u8>,
+    /// The claim-order key (priority DESC).
+    pub priority: i64,
+    /// The resolved ready time — the claim-order key within equal
+    /// priority (unix seconds).
+    pub run_at: i64,
+    /// Claim count; every claim counts, reclaims included.
+    pub attempts: i64,
+    /// The row's max-attempts stamp (resolved at enqueue).
+    pub max_attempts: i64,
+    /// The claimant (`worker_id` on claims); `None` pre-claim.
+    pub worker_id: Option<String>,
+    /// Unix seconds at the most recent claim (fresh or reclaim);
+    /// `None` before any claim (ADR-021 §2) — the dual-execution
+    /// window's diagnostic field, paired with `claim_expires_at`.
+    pub claimed_at: Option<i64>,
+    /// The claim deadline (unix seconds); `None` pre-claim.
+    pub claim_expires_at: Option<i64>,
+    /// Unix seconds at enqueue.
+    pub created_at: i64,
+    /// The resolved absolute job-level expiry; `None` = never expires.
+    pub expires_at: Option<i64>,
+    /// The visibility-timeout stamp resolved at enqueue.
+    pub visibility_timeout_s: i64,
+    /// The backoff-base stamp resolved at enqueue.
+    pub backoff_base_s: i64,
+    /// The dead-letter-retention stamp resolved at enqueue; `None` =
+    /// dead rows live forever.
+    pub dead_letter_retention_s: Option<i64>,
+    /// Dead-only: the recorded failure reason (`None` on live rows).
+    pub last_error: Option<String>,
+    /// Dead-only: unix seconds at death (`None` on live rows).
+    pub died_at: Option<i64>,
+}
+
+impl Job {
+    /// Decode the stored payload bytes into a concrete target type.
+    ///
+    /// The stored bytes are the serde_json serialization of the
+    /// enqueue-time `Value` (ADR-020 §4); any deserialization failure
+    /// is [`Error::Codec`](crate::Error::Codec).
+    pub fn payload_as<T: DeserializeOwned>(&self) -> crate::Result<T> {
+        payload::payload_as(&self.payload)
+    }
+}

alkstore/src/lib.rs

@@ -6,13 +6,29 @@
 //! this crate and implement its traits.
 
 mod error;
+mod job;
+mod opts;
+mod payload;
+mod schedule;
+mod stop_token;
+mod stream;
 mod validation;
+mod wake;
 
 pub use error::{Error, Result};
+pub use job::{Job, JobState};
+pub use opts::{EnqueueOpts, QueueOpts, ScheduleOpts};
+pub use payload::{encode_payload, payload_as};
+pub use schedule::Schedule;
+pub use stop_token::StopToken;
+pub use stream::StreamEvent;
 pub use validation::{
     NameKind, RESERVED_LISTENER_RECONNECTED, RESERVED_PREFIX, validate_local_name, validate_name,
     validate_shared_name,
 };
+pub use wake::Wake;
 
 #[cfg(test)]
 mod tests;
+#[cfg(test)]
+mod value_type_tests;

alkstore/src/opts.rs

@@ -0,0 +1,95 @@
+//! Opts structs (ADR-008 §1, ADR-010 §3/§3a, ADR-020 §1–§3).
+//!
+//! Pinned v1 field sets, carried whole. Deliberately *not*
+//! `#[non_exhaustive]` (ADR-017 §3): consumers construct these — an
+//! opts *field* addition post-release is a class 3 breaking change,
+//! priced that way deliberately.
+//!
+//! Field semantics (delay-over-`run_at` precedence, relative `expires`)
+//! are engine-side resolution (ADR-020 §1–§2) — core carries only the
+//! fields and their doc text.
+
+/// Options for enqueueing a job (`Queue::enqueue`, `enqueue_tx`,
+/// `outbox_enqueue_tx`).
+///
+/// Every field has a default-resolved meaning, so all are
+/// constructor-settable with `..Default::default()` (`Default` gives
+/// all-`None` and priority 0).
+#[derive(Debug, Clone, PartialEq, Eq, Default)]
+pub struct EnqueueOpts {
+    /// Relative delay in seconds from the enqueue instant.
+    ///
+    /// When set, the row's ready time resolves to `now + delay` —
+    /// **wins over `run_at`** if both are supplied (ADR-020 §1;
+    /// precedence is defined, not rejected). `None` leaves the
+    /// resolution to `run_at`, or to "now" when both are unset.
+    pub delay: Option<i64>,
+    /// Absolute ready time (unix seconds).
+    ///
+    /// The absolute-time counterpart of `delay`: used literally when
+    /// set and `delay` is not (ADR-020 §1). `get_job` shows the
+    /// resolved ready time, never the raw fields.
+    pub run_at: Option<i64>,
+    /// The claim-order key (priority DESC). Default 0.
+    pub priority: i64,
+    /// Per-job override of the queue-level max-attempts stamp.
+    ///
+    /// `None` = the queue-level stamp decides (ADR-010 §3a).
+    pub max_attempts: Option<i64>,
+    /// Relative job-level expiry: seconds from the enqueue instant.
+    ///
+    /// Resolved to an absolute row `expires_at` at enqueue (ADR-020
+    /// §2); `None` = never expires.
+    pub expires: Option<i64>,
+}
+
+/// Queue-level option stamps (ADR-010 §3/§3a).
+///
+/// Queues are names, not registered objects: these stamps resolve and
+/// attach to every job row at enqueue — claims, heartbeats, retries,
+/// and sweeps read the job's own stamps, never a live registry.
+/// Opts changes apply to *future enqueues only*.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct QueueOpts {
+    /// Claim visibility timeout in seconds — the default claim
+    /// deadline stamped onto rows. Default 300.
+    pub visibility_timeout_s: i64,
+    /// Default max attempts before dead-letter; overridden per-job by
+    /// `EnqueueOpts::max_attempts` when set. Default 3.
+    pub max_attempts: i64,
+    /// Base seconds for the retry backoff curve. Default 5.
+    pub backoff_base_s: i64,
+    /// Dead-letter retention in seconds; `None` = dead rows live
+    /// forever (the default). Default `None`.
+    pub dead_letter_retention_s: Option<i64>,
+}
+
+impl Default for QueueOpts {
+    /// The pinned defaults (ADR-010 §3): 300 / 3 / 5 / none.
+    fn default() -> Self {
+        QueueOpts {
+            visibility_timeout_s: 300,
+            max_attempts: 3,
+            backoff_base_s: 5,
+            dead_letter_retention_s: None,
+        }
+    }
+}
+
+/// Options for `schedule()` — types identical to their `EnqueueOpts`
+/// twins (no delay/run-at fields: boundary fires are ready at fire
+/// time, ADR-009 §3).
+///
+/// Applied over the target queue's derived engine defaults at each
+/// boundary fire (ADR-020 §3).
+#[derive(Debug, Clone, PartialEq, Eq, Default)]
+pub struct ScheduleOpts {
+    /// The claim-order key for fired jobs; default 0.
+    pub priority: i64,
+    /// Per-schedule override of the queue-level max-attempts stamp;
+    /// `None` = the default stamp decides.
+    pub max_attempts: Option<i64>,
+    /// Relative job-level expiry in seconds from the fire instant
+    /// (ADR-020 §2); `None` = never expires.
+    pub expires: Option<i64>,
+}

alkstore/src/payload.rs

@@ -0,0 +1,33 @@
+//! Payload encoding posture (ADR-020 §4): payloads cross the trait as
+//! [`serde_json::Value`]; engines serialize with serde_json and store
+//! exactly those bytes — nothing re-encodes on the way out.
+//!
+//! Core provides the encode helper (the `Value` → stored-bytes
+//! serialization the engines call) and the [`payload_as`] decode
+//! convenience used by [`Job`](crate::Job) and
+//! [`StreamEvent`](crate::StreamEvent). `serde`/`serde_json` are core's
+//! only dependencies beyond `thiserror`.
+
+use serde::de::DeserializeOwned;
+
+use crate::Error;
+
+/// Serialize a trait-crossing payload `Value` to the stored byte form.
+///
+/// The engines call this at enqueue/publish time and store exactly the
+/// returned bytes (ADR-020 §4): the stored bytes of a job row or
+/// stream event row are, contract-pinned, the serde_json serialization
+/// of the `Value` the call carried. Cross-engine byte equality falls
+/// out — the same `Value` serializes identically on both engines.
+pub fn encode_payload(value: &serde_json::Value) -> Vec<u8> {
+    serde_json::to_vec(value).unwrap_or_else(|_| Vec::new())
+}
+
+/// Decode stored payload bytes into a concrete target type.
+///
+/// Used by [`Job::payload_as`](crate::Job::payload_as) and
+/// [`StreamEvent::payload_as`](crate::StreamEvent::payload_as). The
+/// error is [`Error::Codec`] on any deserialization failure (pinned).
+pub fn payload_as<T: DeserializeOwned>(payload: &[u8]) -> crate::Result<T> {
+    serde_json::from_slice(payload).map_err(|e| Error::Codec(e.to_string()))
+}

alkstore/src/schedule.rs

@@ -0,0 +1,22 @@
+//! Schedule read-back value (ADR-019 §3).
+//!
+//! `#[non_exhaustive]` per ADR-017 §3 — consumers read but never
+//! construct.
+
+use crate::opts::ScheduleOpts;
+
+/// The `schedule()` read-back value (upsert by name).
+#[derive(Debug, Clone, PartialEq, Eq)]
+#[non_exhaustive]
+pub struct Schedule {
+    /// The schedule name (shared namespace — validated at the entry
+    /// point with the reserved prefix rejected).
+    pub name: String,
+    /// The v1 spec grammar: `@every <n><unit>` only (`s|m|h|d`).
+    pub spec: String,
+    /// The target queue name fired jobs enqueue into.
+    pub queue: String,
+    /// The schedule's opts, applied over the target queue's derived
+    /// engine defaults at each boundary fire (ADR-020 §3).
+    pub opts: ScheduleOpts,
+}

alkstore/src/stop_token.rs

@@ -0,0 +1,56 @@
+//! The stop token (ADR-019 §4): core-owned, contract-visible handle
+//! for `run_schedules(stop)`.
+//!
+//! The token is pinned in core because the parameter type is contract
+//! surface (ADR-017 — a pinned method's signature); the *await
+//! mechanics* are engine-internal (tokio watch per engine — engines
+//! may wrap their own watch channels around this token). Core
+//! implements the flip-everywhere semantics with a shared
+//! `Arc<AtomicBool>`.
+
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
+
+/// A cloneable cancel handle for `run_schedules(stop)`.
+///
+/// `cancel()` on any clone flips every clone; the engine's
+/// `run_schedules` ends its sleep early and returns `Ok(())` — the
+/// clean-stop return (leadership loss still returns
+/// `Err(LeadershipLost)` regardless of the token, ADR-009 §1).
+#[derive(Clone)]
+pub struct StopToken {
+    cancelled: Arc<AtomicBool>,
+}
+
+impl StopToken {
+    /// Create a fresh, uncancelled token.
+    pub fn new() -> Self {
+        StopToken {
+            cancelled: Arc::new(AtomicBool::new(false)),
+        }
+    }
+
+    /// Cancel the token — flips every clone (they share one flag).
+    pub fn cancel(&self) {
+        self.cancelled.store(true, Ordering::SeqCst);
+    }
+
+    /// Whether the token has been cancelled (by any clone).
+    pub fn is_cancelled(&self) -> bool {
+        self.cancelled.load(Ordering::SeqCst)
+    }
+}
+
+impl Default for StopToken {
+    fn default() -> Self {
+        StopToken::new()
+    }
+}
+
+impl std::fmt::Debug for StopToken {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("StopToken")
+            .field("cancelled", &self.is_cancelled())
+            .finish()
+    }
+}

alkstore/src/stream.rs

@@ -0,0 +1,43 @@
+//! Stream event value (ADR-015 §3).
+//!
+//! `#[non_exhaustive]` per ADR-017 §3 — consumers read but never
+//! construct.
+
+use serde::de::DeserializeOwned;
+
+use crate::payload;
+
+/// A stream event — honker's shape with one rename (`topic` →
+/// `stream`, ADR-015 §3).
+///
+/// `#[non_exhaustive]` (ADR-017 §3). Offsets are positions in the
+/// stream's log — immutable, never renumbered; order is global FIFO
+/// under publish sequence. The key is carried metadata only: not a
+/// partitioning instruction, not a dedup key, not an ordering key.
+#[derive(Debug, Clone, PartialEq, Eq)]
+#[non_exhaustive]
+pub struct StreamEvent {
+    /// Position in the stream's log; immutable.
+    pub offset: i64,
+    /// The stream name.
+    pub stream: String,
+    /// The carried key; `None` unless published keyed.
+    pub key: Option<String>,
+    /// Payload bytes — the serde_json serialization of the trait
+    /// `Value` at publish time (ADR-020 §4); decode with
+    /// [`StreamEvent::payload_as`].
+    pub payload: Vec<u8>,
+    /// Unix seconds at publish (the engines' single clock source).
+    pub created_at: i64,
+}
+
+impl StreamEvent {
+    /// Decode the stored payload bytes into a concrete target type.
+    ///
+    /// The stored bytes are the serde_json serialization of the
+    /// publish-time `Value` (ADR-020 §4); any deserialization failure
+    /// is [`Error::Codec`](crate::Error::Codec).
+    pub fn payload_as<T: DeserializeOwned>(&self) -> crate::Result<T> {
+        payload::payload_as(&self.payload)
+    }
+}

alkstore/src/value_type_tests.rs

@@ -0,0 +1,163 @@
+use std::collections::HashMap;
+
+use serde_json::{Value, json};
+
+use crate::{EnqueueOpts, Job, JobState, QueueOpts, ScheduleOpts, StopToken, StreamEvent, Wake};
+
+#[test]
+fn enqueue_opts_default_all_none_priority_zero() {
+    let d = EnqueueOpts::default();
+    assert_eq!(
+        d,
+        EnqueueOpts {
+            delay: None,
+            run_at: None,
+            priority: 0,
+            max_attempts: None,
+            expires: None,
+        }
+    );
+}
+
+#[test]
+fn queue_opts_default_pinned_300_3_5_none() {
+    let d = QueueOpts::default();
+    assert_eq!(
+        d,
+        QueueOpts {
+            visibility_timeout_s: 300,
+            max_attempts: 3,
+            backoff_base_s: 5,
+            dead_letter_retention_s: None,
+        }
+    );
+}
+
+#[test]
+fn schedule_opts_default_matches_enqueue_opts_twin_types() {
+    let d = ScheduleOpts::default();
+    assert_eq!(
+        d,
+        ScheduleOpts {
+            priority: 0,
+            max_attempts: None,
+            expires: None,
+        }
+    );
+    let e = EnqueueOpts::default();
+    assert_eq!(e.priority, d.priority);
+    assert_eq!(e.max_attempts, d.max_attempts);
+    assert_eq!(e.expires, d.expires);
+}
+
+fn sample_job(payload: Vec<u8>) -> Job {
+    Job {
+        id: 42,
+        queue: "orders".to_string(),
+        state: JobState::Dead,
+        payload,
+        priority: 7,
+        run_at: 1000,
+        attempts: 3,
+        max_attempts: 3,
+        worker_id: Some("w1".to_string()),
+        claimed_at: Some(999),
+        claim_expires_at: Some(1299),
+        created_at: 900,
+        expires_at: Some(5000),
+        visibility_timeout_s: 300,
+        backoff_base_s: 5,
+        dead_letter_retention_s: None,
+        last_error: Some("max attempts exceeded".to_string()),
+        died_at: Some(1100),
+    }
+}
+
+fn sample_event(payload: Vec<u8>) -> StreamEvent {
+    StreamEvent {
+        offset: 17,
+        stream: "events".to_string(),
+        key: Some("k".to_string()),
+        payload,
+        created_at: 1234,
+    }
+}
+
+#[test]
+fn payload_as_round_trips_json_value_through_job() {
+    let v = json!({"id": 9, "nested": {"ok": [true, null, 1.5]}, "s": "ünï"});
+    let bytes = crate::encode_payload(&v);
+    assert_eq!(bytes, serde_json::to_vec(&v).unwrap_or_default());
+    let job = sample_job(bytes);
+    let decoded: Value = job
+        .payload_as()
+        .unwrap_or_else(|e| panic!("decode failed: {e}"));
+    assert_eq!(decoded, v);
+}
+
+#[test]
+fn event_payload_as_round_trips() {
+    let v = json!([1, 2, 3, {"x": "y"}]);
+    let event = sample_event(crate::encode_payload(&v));
+    let decoded: Value = event
+        .payload_as()
+        .unwrap_or_else(|e| panic!("decode failed: {e}"));
+    assert_eq!(decoded, v);
+}
+
+#[test]
+fn payload_as_concrete_typed_targets() {
+    let v = json!({"alpha": 1, "beta": 2});
+    let job = sample_job(crate::encode_payload(&v));
+    let decoded: HashMap<String, i64> = job.payload_as().unwrap_or_else(|e| panic!("decode: {e}"));
+    assert_eq!(
+        decoded,
+        HashMap::from([("alpha".into(), 1), ("beta".into(), 2)])
+    );
+}
+
+#[test]
+fn payload_as_undecodable_target_yields_codec() {
+    let v = json!({"name": "x"});
+    let job = sample_job(crate::encode_payload(&v));
+    let r: crate::Result<String> = job.payload_as();
+    match r {
+        Err(crate::Error::Codec(msg)) => assert!(!msg.is_empty()),
+        other => panic!("expected Codec, got {other:?}"),
+    }
+    let r2: crate::Result<bool> =
+        serde_json::from_slice(b"not json").map_err(|e| crate::Error::Codec(e.to_string()));
+    assert!(matches!(r2, Err(crate::Error::Codec(_))));
+}
+
+#[test]
+fn stop_token_cancel_flips_all_clones() {
+    let token = StopToken::new();
+    let clone = token.clone();
+    let deep = clone.clone();
+    assert!(!token.is_cancelled());
+    assert!(!clone.is_cancelled());
+    assert!(!deep.is_cancelled());
+    clone.cancel();
+    assert!(token.is_cancelled());
+    assert!(clone.is_cancelled());
+    assert!(deep.is_cancelled());
+}
+
+#[test]
+fn stop_token_independent_tokens_do_not_flip_each_other() {
+    let a = StopToken::default();
+    let b = StopToken::default();
+    a.cancel();
+    assert!(a.is_cancelled());
+    assert!(!b.is_cancelled());
+}
+
+#[test]
+fn wake_carries_only_the_channel() {
+    let w = Wake {
+        channel: "ch-1".to_string(),
+    };
+    assert_eq!(w.channel, "ch-1");
+    assert_eq!(w.clone(), w);
+}

alkstore/src/wake.rs

@@ -0,0 +1,22 @@
+//! Wake value (ADR-008 §3): the `listen(channel)` channel-carried
+//! hint.
+//!
+//! `#[non_exhaustive]` per ADR-017 §3 — consumers read but never
+//! construct.
+
+/// A wake: the one piece of semantic content the wake contract allows
+/// — the channel name. No payload is surfaced (honker's
+/// `Notification { id, channel, payload }` is not); consumers needing
+/// content use streams (the guarantee split, ADR-006). Wakes are
+/// hints: their multiplicity is not part of the contract (coalescing
+/// on SQLite, per-notify on Postgres).
+///
+/// The Postgres engine's synthetic reconnect-wake arrives as
+/// `Wake { channel: RESERVED_LISTENER_RECONNECTED }` on every
+/// subscriber's receiver after a watcher reconnect.
+#[derive(Debug, Clone, PartialEq, Eq)]
+#[non_exhaustive]
+pub struct Wake {
+    /// The channel the wake was delivered on.
+    pub channel: String,
+}