From 6feaa2ac724645ff43c20df16b6246526f936ae3 Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Mon, 12 Oct 2026 23:49:09 +0000 Subject: [PATCH] =?UTF-8?q?feat(ops):=20the=20ops=20foundation=20=E2=80=94?= =?UTF-8?q?=20the=20token=20layer,=20the=20custody=20split,=20the=20namesp?= =?UTF-8?q?ace=20binding,=20and=20the=20fanout=20registry=20(wave=207)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The ops module's state substrate under feature `ops`, no op handlers yet (they land with ops-json-ops/ops-fetch/ops-put): - `OpsContext` — the context type the handler pairs capture: the `Arc`, the module config, the token registry, the per- namespace converter bindings, the fanout registry (re-exported from lib.rs) - The pin-token layer (ADR-010 §4 / ADR-012 §2): CSPRNG `PinToken` (32 bytes, `getrandom 0.4` — the decided minting form; counter and derivable forms rejected as guessable capabilities), the token↔(key, owner) mapping with process lifetime (retirement at conversion-confirmed or process end; no expiry machinery), the validity domain = the minting node (peers refuse unknown tokens with present: false, never an error) - The guard-custody split: `mint_fleet` (mapping only; the put handler's guard drops at response completion — its own act; the fleet-declaration and owner-instance checks ride the mint) vs `mint_custodial` (the registry holds the guard; nothing expires; refused under a fleet declaration) - The namespace→liveness-source binding: per-namespace `PinConverter` registration; the conversion trigger decided as the embedder-called confirm verb (probe-triggered conversion rejected: a read-gated op never mutates protection), riding `RegistryCore::convert`'s ordered step under the arbitration lock - The crate-internal store-core renewal access: `Store::ops_pin_renew`/`Store::ops_pin_check` (pub(crate), cfg all (ops, postgres)) routed through the fleet host — the W6-3 named consumer; the probe composition decided as verify-the-mapping-then-renew; no public-API widening, no backend side-door - The fanout registry (POC #3 finding A3): per-digest broadcasts, bounded subscriber channels, the late-joiner degrade signal (subscribe → None, channel end → verified get), and the Stall/Drop slow-subscriber policies - Dependency posture: serde/serde_json/getrandom enter as ops-gated optional deps; the default build's dependency tree verified unchanged; gc.rs's allow-scope notes updated to the landed state Verification: cargo test (default 73/19/3, module compiled out), cargo test --features ops (84/19/3, 11 new ops unit tests), cargo test --all-features (237/125/4); clippy --all-targets -D warnings across the default/ops/all-features/pg-pure variants; cargo fmt; cargo doc --no-deps zero warnings on the default and gated-ops builds (the all-features count at the pre-existing baseline). --- Cargo.toml | 5 +- src/gc.rs | 18 +- src/lib.rs | 6 + src/ops.rs | 1298 ++++++++++++++++++++++++++++++++++++++- src/store.rs | 49 ++ tasks/ops-foundation.md | 145 ++++- 6 files changed, 1501 insertions(+), 20 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 2ab0c08..fd6bf21 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -23,7 +23,7 @@ sqlite = ["dep:rusqlite"] postgres = ["dep:tokio-postgres", "dep:deadpool-postgres"] mem = [] pg-lo = ["dep:tokio-postgres", "dep:deadpool-postgres"] -ops = ["dep:alkcall"] +ops = ["dep:alkcall", "dep:serde", "dep:serde_json", "dep:getrandom"] [dependencies] tokio = { version = "1", default-features = false, features = ["rt", "sync", "io-util", "macros", "time"] } @@ -32,8 +32,11 @@ parking_lot = "0.12" alkcall = { version = "0.8.1", optional = true } deadpool-postgres = { version = "0.14", optional = true } +getrandom = { version = "0.4", optional = true } hex = "0.4" rusqlite = { version = "0.40", features = ["bundled"], optional = true } +serde = { version = "1", features = ["derive"], optional = true } +serde_json = { version = "1", optional = true } sha1 = "0.10" sha2 = "0.10" tokio-postgres = { version = "0.7", optional = true } diff --git a/src/gc.rs b/src/gc.rs index 9271a36..9a73a42 100644 --- a/src/gc.rs +++ b/src/gc.rs @@ -81,12 +81,18 @@ //! fleet trait and its registration bookkeeping in non-postgres //! builds (no implementer or consumer compiles without feature //! `postgres`), the seam's `pin_check` clause — the pins-only -//! classifier, deliberately unconsumed (its remaining consumer is -//! wave 7's ops renewal probe), and the ADR-005 conversion-ordering -//! primitive (`GcHome::convert`/`RegistryCore::convert`) plus the -//! `GcHome::pin_count` observation aid — test-consumed today, with -//! the conversion's in-sweep consumer being wave 7's liveness edit -//! flow. +//! classifier, whose consumer has landed (wave 7's +//! ops-foundation: the ops module's renewal probe reaches it through +//! the store core's crate-internal `ops_pin_check` crossing, the +//! have handler's renewal-verification step), and the ADR-005 +//! conversion-ordering primitive (`GcHome::convert`/ +//! `RegistryCore::convert`) plus the +//! `GcHome::pin_count` observation aid — the conversion's +//! guard-custody consumer has also landed (the ops module's confirm +//! verb converts held guards at the embedder's confirmed +//! registration), leaving `convert`/`pin_count` test-consumed +//! today, with the conversion's in-sweep consumer being wave 7's +//! liveness edit flow. #![allow(dead_code)] diff --git a/src/lib.rs b/src/lib.rs index e11ee5c..01e4564 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -30,6 +30,12 @@ pub mod pg_lo; #[cfg(feature = "ops")] pub mod ops; +#[cfg(feature = "ops")] +pub use ops::{ + FanoutChunk, FanoutPublisher, FanoutReceiver, FanoutRegistry, OpsConfig, OpsContext, + PinConverter, PinToken, RenewalOutcome, SlowSubscriberPolicy, TokenRegistry, +}; + #[cfg(any(feature = "postgres", feature = "pg-lo"))] mod pg_common; diff --git a/src/ops.rs b/src/ops.rs index 88ad200..d57569e 100644 --- a/src/ops.rs +++ b/src/ops.rs @@ -1,2 +1,1296 @@ -//! The ops surface (ADR-001): alkcall-backed network operations behind -//! the `ops` feature gate. +//! The ops surface's state substrate (ADR-001 §3 placement; the +//! wave-7 foundation): the feature-gated module's context type, the +//! pin-token layer, the namespace→liveness-source binding, the +//! crate-internal store-core renewal access, and the fetch fanout +//! registry — everything the op handlers ride, none of the handlers +//! themselves (spec/handler pairs land in the follow-on tasks). +//! +//! **The ops layer is a store-facade consumer** (ADR-001 §4): its +//! store access is the public facade plus exactly one crate-internal +//! seam — the renewal access (`Store::ops_pin_renew`/ +//! `Store::ops_pin_check`, ADR-010 §4), which the *store core* owns +//! and this module consumes as an in-crate peer. There is no backend +//! side-door, no engine type on this module's surface, no public-API +//! widening for the seam. +//! +//! **The module's state sits in [`OpsContext`]** — the type the +//! handler pairs (later tasks) capture. It binds the store +//! (`Arc`), the module config ([`OpsConfig`]), the token +//! registry (the token↔(key, owner) mapping, process lifetime), the +//! per-namespace converter bindings, and the fanout registry. The +//! registries are also constructible standalone ([`FanoutRegistry`]; +//! the token registry under its context) — the fanout registry's +//! mechanics unit-test standalone here. +//! +//! ## The pin-token layer (ADR-010 §4; ADR-012 §2) +//! +//! The token *is* the pin — there is no second pin state. The durable +//! state is the pin row (fleet) or the in-process guard (non-fleet); +//! the token is the ops layer's opaque handle over it, and the +//! token↔(key, owner) mapping is ops-layer in-memory state with +//! **process lifetime**: it outlives the minting handler (a renewal +//! must be answerable long after the `blobs/put` response completed) +//! and never outlives the process. The mapping's retention +//! discipline: an entry retires at **conversion-confirmed** (the +//! [`OpsContext::confirm`] verb's success removes it) or at **process +//! end** — the mapping carries no expiry machinery of its own. The +//! pin row's expiry is the fleet liveness signal, and nothing expires +//! non-fleet: the mapping's absence of TTL is deliberate on both +//! sides. +//! +//! **Named design point — the token's minting form (decided): a +//! CSPRNG id.** [`PinToken`] is 32 bytes of OS-entropy output +//! (`getrandom`, the `ops`-gated dependency) — unguessable by +//! construction, opaque by construction (no key, owner, or namespace +//! bytes ride the token). Rejected forms: a counter (sequential ids +//! are guessable — the token is a capability, and a guessable +//! capability is a broken one: renewal and confirmation are +//! token-authenticated acts) and a derivable form (anything derived +//! from key/owner/namespace is derivable by anyone who knows a put's +//! identity — equally broken, and no longer opaque). Wire form: hex +//! (ADR-012 §2's `{token, digest}` shapes carry it as the `token` +//! field's string; lowercase canonical decode case-insensitive). +//! +//! **Named design point — the validity domain: the minting node.** +//! The registry is per-process state; there is no cross-node form. +//! Renewal probes must reach the node that minted the token; a fleet +//! peer's registry does not know it and answers +//! [`RenewalOutcome::UnknownToken`] — `present: false`, not an error +//! (ADR-012 §2's refusal discipline). A putter whose minting node +//! died is in ADR-005's in-flight-put loss case; the backstop is the +//! re-put (delete-then-recover, ADR-005), and routing renewals to the +//! serving node is a documented embedder duty, never new machinery. +//! +//! ## The guard-custody split (ADR-010 §4, the token layer's core +//! branch) +//! +//! The two custody forms are total — a token's entry is exactly one +//! of them, and [`OpsContext::mint_fleet`] / +//! [`OpsContext::mint_custodial`] enforce the pairing against the +//! bound store's fleet declaration: +//! +//! - **Fleet**: the mint records only the mapping (the serving +//! instance's id as the `owner` half — the mint validates the +//! passed owner against the store's, refusing a mint over a +//! non-fleet store or with a foreign owner). The put handler's +//! `Pin` guard **drops at response completion** — the handler's +//! act, after the token is minted — because a guard held for the +//! token's whole life would make the TTL machinery dead code on the +//! ops path: the pin row's expiry, not the guard's presence, is the +//! fleet liveness signal (ADR-010 §4), which is exactly why the +//! renewal form exists. +//! - **Non-fleet** (any engine — sqlite, single-instance postgres, +//! pg-lo, `local`, `mem`; ADR-012 §2's generalization): the +//! registry **holds the guard** — the token is the in-process +//! guard's handle, nothing expires — and minting is refused on a +//! declared fleet store for the mirror reason. The guard's fate is +//! the custody lifecycle: conversion at the embedder's confirmed +//! registration, else process-lifetime hold. +//! +//! ## The namespace→liveness-source binding +//! +//! Every ops payload carries an explicit `namespace` — the store core +//! never sees it (backends are namespace-blind); it selects the +//! liveness source and the ACL resource. The binding this module +//! ships is the embedder's converter seam: **per-namespace +//! registration of the converter the embedder plugs its registry +//! into** — [`PinConverter::convert`] is the registration act (the +//! embedder registers the key into its own registry), and the +//! conversion itself rides the crate-internal ordered-step primitive +//! (`RegistryCore::convert` — the pin's release and the converter's +//! registration are one ordered step under the store's arbitration +//! lock; a concurrent arbitration never observes +//! neither-pin-nor-registered). The converter runs **under the +//! arbitration lock** in the guarded form: it must be a quick, +//! non-blocking, sync act — the same duty a [`crate::facade::LivenessSource`]'s +//! delete-time probe carries. +//! +//! **Named design point — the conversion trigger (decided): an +//! embedder-called confirm verb.** [`OpsContext::confirm(namespace, +//! token)`] is the trigger; a probe-triggered conversion is rejected: +//! the have probe is a read-gated existence op whose caller is the +//! remote putter — a protection-lifecycle mutation keyed off it would +//! give a read op write semantics and put the registration act under +//! the putter's arrival, while the hand-over contract's shape is +//! explicit that the *embedder* observes the registration landed +//! (`blobs/have` returns present after registration) and then acts. +//! The duty text is the embedder's; the mechanism (this verb) is this +//! module's. +//! +//! ## The crate-internal store-core renewal access (ADR-010 §4) +//! +//! The renewal probe's composition (decided): **verify-the-mapping- +//! then-renew** — (a) the token lookup (the validity-domain +//! enforcement: an unknown token answers `UnknownToken`, present: +//! false, no renewal), (b) the digest check (one token binds one pin +//! row; a mismatch answers `DigestMismatch`), (c) the store core's +//! `Store::ops_pin_check` classifies the row (the wave-6 review's +//! W6-3 named consumer) — a row not live answers `Expired` with no +//! renewal write, (d) the store core's `Store::ops_pin_renew` — +//! its bool is the final answer: never resurrects an expired or +//! absent row, so a check/renew race lands on its `false` (`Expired`) +//! the same way. Renew-answers-the-verification (a bare +//! `pin_renew(key, owner)` answering the probe outright) was +//! rejected: the verify steps are where the validity domain and the +//! one-token-one-row binds are enforced, and the W6-3 named consumer +//! is exactly the verify step's classifier. Under non-fleet the probe +//! answers **from custody**: a held guard is present, nothing expires +//! (ADR-012 §2), and no engine call runs. +//! +//! What never happens: a *consumer* reaching the seam, or a public +//! API widening for this. The have handler (later task) rides +//! [`OpsContext::renew`] and sees none of the engine. +//! +//! ## The fetch fanout registry (ops-surface.md; POC #3 finding A3) +//! +//! Per-digest broadcast state above the store: one reader (the +//! producer's own consumption — the put handler's stream is read +//! exactly once), the **store arm** (the producer's own persistence) +//! and **subscriber arms** (the [`FanoutRegistry::subscribe`] taps). +//! Channels are **bounded** ([`OpsConfig`]'s `fanout_channel_bound`, +//! the POC's 8): the backpressure discipline is that a slow +//! subscriber stalls the broadcast — correct for a put (the put must +//! land); subscriber count is bounded +//! (`fanout_max_subscribers`) and slow-subscriber handling is policy: +//! **`Stall`** holds the broadcast on a full channel (the put +//! producer's form), **`Drop`** is the drop-and-late-join policy — a +//! full arm is dropped, its consumer's channel closes, and the +//! consumer degrades to the late-joiner path. The **late-joiner +//! signal** is [`FanoutRegistry::subscribe`] answering `None` (no +//! live broadcast, broadcast ended, at the subscriber bound) or a +//! receiver's channel end: the degrade is a normal post-commit +//! verified `get` — identical bytes under CAS, either way. +//! +//! The producer wiring lands with `ops-put`, the consumer wiring with +//! `ops-fetch`; the registry's mechanics land here, unit-tested +//! standalone. `begin` returning `None` means a broadcast for the +//! digest is already live — the producer continues store-arm-only +//! (CAS guarantees the incumbent's bytes are the digest's bytes), and +//! a publisher's drop (finished or abandoned) removes the entry. +//! +//! ## The renewal-cadence boundary (ADR-013 §1) +//! +//! This module ships the renewal *mechanism* (the have form's +//! handler, later task, composing [`OpsContext::renew`]). The TTL/3 +//! *cadence* — the putter's renewal worker — is **embedder +//! composition above the facade** (scheduler/queues; ADR-008's +//! "embedder owns cadence"), never in-crate machinery: no renewal +//! worker, scheduler, or timer exists here, and none may. +//! +//! ## Dependency posture (convention 8; ADR-013 §4) +//! +//! `serde`/`serde_json` (the wire payloads' handling, the handlers' +//! concern) and `getrandom` (the CSPRNG) are `ops`-gated optional +//! dependencies: the default build's dependency tree is unchanged, +//! and `ops` pulls alkcall plus these, default-off. The base crate +//! stays lean. + +use std::collections::BTreeMap; +use std::sync::Arc; + +use parking_lot::Mutex; +use tokio::sync::mpsc; + +use crate::facade::Pin; +use crate::keys::Key; +use crate::store::Store; +use crate::Error; + +/// The pin token (ADR-010 §4; ADR-012 §2): the ops layer's opaque, +/// unguessable handle over a landed put's pin. 32 bytes of OS-entropy +/// CSPRNG output — the minting-form design point's decided shape (see +/// the module doc); no key, owner, or namespace bytes ride it. The +/// validity domain is the minting node: a peer's registry does not +/// know the token and answers as unknown, present: false (not an +/// error). Wire form: hex (ADR-012 §2's `{token, digest}` shapes). +/// +/// `Debug` is hand-written and masks the bytes: the token is a +/// capability — it must not land in logs, panics, or error payloads. +#[derive(Clone, Copy, PartialEq, Eq, Hash)] +pub struct PinToken { + bytes: [u8; 32], +} + +impl PinToken { + /// Mint a fresh token: 32 bytes from the OS entropy source (the + /// CSPRNG design point). Fallible only on the entropy source's + /// failure — an operational condition, the `Io` family. + pub fn generate() -> Result { + let mut bytes = [0u8; 32]; + getrandom::fill(&mut bytes) + .map_err(|e| Error::Io(format!("pin-token minting: the entropy source failed: {e}")))?; + Ok(PinToken { bytes }) + } + + /// The token's raw bytes (the registry's map key; the renewal + /// lookup's identity). + pub fn as_bytes(&self) -> &[u8; 32] { + &self.bytes + } + + /// The wire form (lowercase hex, 64 chars). + pub fn to_hex(&self) -> String { + hex::encode(self.bytes) + } + + /// The wire form's ingress: lowercase canonical, decode + /// case-insensitive; `None` on wrong length or non-hex bytes — a + /// malformed token is an unknown token (present: false, not an + /// error), which is the have handler's mapping if it wants it. + pub fn from_hex(token_hex: &str) -> Option { + let bytes = hex::decode(token_hex).ok()?; + bytes.try_into().ok().map(|bytes| PinToken { bytes }) + } +} + +impl std::fmt::Debug for PinToken { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str("PinToken(#redacted)") + } +} + +/// The renewal probe's answer vocabulary (the have handler's renewal +/// form, ADR-012 §2): every variant but [`RenewalOutcome::Renewed`] +/// answers `present: false` on the wire, and only `Renewed` carries a +/// renewal. The putter's backstop for every non-renewed answer is the +/// re-put (delete-then-recover, ADR-005) — CAS makes it idempotent. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum RenewalOutcome { + /// The token mapped, the row was live, and the renewal landed — + /// `present: true`. + Renewed, + /// The registry does not know the token: the validity-domain + /// refusal — a foreign/unminted token, or a mapping already + /// retired at conversion. `present: false`; no renewal occurred. + UnknownToken, + /// The token mapped, but to a different digest than the one + /// presented: one token binds one pin row. `present: false`; no + /// renewal occurred. + DigestMismatch, + /// The token mapped, but the pin row is not live (expired, raced, + /// unrecoverable — the engine's never-resurrect discipline, + /// ADR-012 §2). `present: false`; no renewal occurred. + Expired, +} + +impl RenewalOutcome { + /// The have wire answer (`present[]`'s element): `true` only for + /// [`RenewalOutcome::Renewed`]. + pub fn present(&self) -> bool { + matches!(self, RenewalOutcome::Renewed) + } +} + +/// The embedder's per-namespace converter — the binding the embedder +/// plugs its registry into (ops-surface.md's hand-over contract step +/// 3): the conversion's registration act, invoked by the ops layer's +/// [`OpsContext::confirm`] verb. +/// +/// The act is sync and must be quick, indexed, non-blocking: in the +/// guarded (non-fleet) form it runs **under the store's arbitration +/// lock** (the ordered-step conversion's +/// one-atomic-section discipline — the pin's release and this +/// registration never split). Registering a slow or async act here +/// would block every arbitration in the store; that shape belongs +/// above the facade, not in a conversion. +pub trait PinConverter: Send + Sync + 'static { + /// The registration act for `key`: the embedder registers the + /// digest into its own registry (whatever shape that is), making + /// the entry live per the source ADR-005's registered-liveness + /// seam consults. The token's mapping retires at this act's + /// confirmation — conversion-confirmed is the retention + /// discipline's first retirement arm. + fn convert(&self, key: &Key); +} + +/// The token registry's entry: the minting metadata (the namespace +/// the put arrived under — the confirm verb's converter-routing +/// key) and the custody branch. +struct PinTicket { + namespace: String, + key: Key, + custody: PinCustody, +} + +/// The custody branch — the token layer's core split (ADR-010 §4, +/// the module doc's section): the fleet form holds only the mapping +/// (the pin row's expiry is the protection; the guard dropped at +/// response completion), the non-fleet form holds the guard. +enum PinCustody { + /// Fleet: the mapping only — `owner` is the serving store + /// instance's id, validated against the bound store at minting. + Fleet { owner: Vec }, + /// Non-fleet: the registry holds the put's guard — nothing + /// expires — until the conversion consumes it. + Held(Pin), +} + +/// A renewal's custody snapshot — the mapping peek (the guards are not +/// cloneable; the branch is carried as its shape). +enum RenewalCustody { + Held, + Fleet { owner: Vec }, +} + +struct RenewalTarget { + key: Key, + custody: RenewalCustody, +} + +/// The token registry — the token↔(key, owner) in-memory mapping with +/// process lifetime (ADR-010 §4; ADR-012 §2). Per-process state: the +/// validity domain is the minting node; a peer registry's lookup +/// misses (`present: false`, not an error). Entries retire at +/// conversion-confirmed or process end — no expiry machinery of its +/// own (the pin row's expiry is the fleet liveness signal; nothing +/// expires non-fleet). +/// +/// Constructed standalone it is the state holder; the +/// [`OpsContext`] verbs are the custody-validated shapes (the +/// context enforces the mint's pairing against the bound store's +/// fleet declaration — the registry's own verbs do not, which is why +/// the handlers mint only through the context). +pub struct TokenRegistry { + tickets: Mutex>, +} + +impl Default for TokenRegistry { + fn default() -> Self { + Self::new() + } +} + +impl TokenRegistry { + /// A fresh, empty registry — process-local bookkeeping. + pub fn new() -> TokenRegistry { + TokenRegistry { + tickets: Mutex::new(BTreeMap::new()), + } + } + + /// Mint a token over the (key, owner) mapping — the fleet form's + /// registry act: the entry is mapping-only, no guard is held. The + /// custody-validated shape is [`OpsContext::mint_fleet`] (the + /// fleet declaration and the owner's instance check ride there). + pub fn mint_fleet(&self, namespace: &str, key: Key, owner: Vec) -> Result { + let token = PinToken::generate()?; + let previous = self.tickets.lock().insert( + *token.as_bytes(), + PinTicket { + namespace: namespace.to_owned(), + key, + custody: PinCustody::Fleet { owner }, + }, + ); + debug_assert!(previous.is_none(), "the CSPRNG minted a colliding token"); + Ok(token) + } + + /// Mint a token and take custody of the put's guard — the + /// non-fleet form: the token is the in-process guard's handle, + /// nothing expires, until the conversion consumes the guard. The + /// custody-validated shape is [`OpsContext::mint_custodial`] (the + /// fleet-store refusal rides there). + pub fn mint_custodial(&self, namespace: &str, key: Key, guard: Pin) -> Result { + let token = PinToken::generate()?; + let previous = self.tickets.lock().insert( + *token.as_bytes(), + PinTicket { + namespace: namespace.to_owned(), + key, + custody: PinCustody::Held(guard), + }, + ); + debug_assert!(previous.is_none(), "the CSPRNG minted a colliding token"); + Ok(token) + } + + /// Whether the token maps right now (the validity-domain + /// registry's lookup — a peer's token misses). + pub fn holds(&self, token: &PinToken) -> bool { + self.tickets.lock().contains_key(token.as_bytes()) + } +} + +/// A fanout chunk — one read of the producer's stream, shared by +/// reference with every subscriber arm (the single-read discipline: +/// the store arm and each subscriber arm observe the same bytes, no +/// per-subscriber copy). +pub type FanoutChunk = Arc>; + +/// A subscriber arm's feed: committed-put chunks in order; +/// `None` is the **end signal** — the broadcast ended (completed, +/// abandoned, or this arm dropped) and the consumer degrades to the +/// late-joiner path: a normal post-commit verified `get` — identical +/// bytes under CAS. +pub type FanoutReceiver = mpsc::Receiver; + +/// The slow-subscriber policy — per-broadcast, ops-layer policy (POC +/// #3 finding A3's policy question). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum SlowSubscriberPolicy { + /// Bounded-channel backpressure: a full subscriber arm stalls the + /// broadcast (and with it the producer's streaming) — the put's + /// form, where the put must land; the bounded channel is the + /// backpressure, subscriber count is the bound. + Stall, + /// The drop-and-late-join policy: a subscriber whose channel is + /// full is dropped at once (its channel closes → the consumer's + /// end signal → the degrade to verified get), the broadcast + /// unblocked for the rest. The read-side producers' form (the + /// fetch side's policy escape). + Drop, +} + +struct LiveBroadcast { + policy: SlowSubscriberPolicy, + arms: Vec>, +} + +/// The fanout registry's shared state — per-digest live-broadcast +/// bookkeeping under one short-held internal lock: arms are cloned +/// out under the lock and driven outside it (a `send` under Stall +/// policy awaits; nothing async runs under the lock). +struct FanoutLive { + broadcasts: Mutex>, + max_subscribers: usize, + channel_bound: usize, +} + +/// The **fanout registry** (ops-surface.md's fetch fanout seam; POC +/// #3 finding A3): per-digest broadcast state — one reader (the +/// producer's), the store arm (the producer's own persistence) and +/// subscriber arms ([`FanoutRegistry::subscribe`]'s taps) riding +/// bounded channels; the late-joiner signal (`subscribe` → `None`, a +/// receiver's channel end) is the degrade-to-verified-get path; the +/// slow-subscriber policy is per-broadcast +/// ([`SlowSubscriberPolicy`]). +/// +/// The store arm is deliberately not a registry arm: the producer's +/// put is its own consumption of the one read (serve-while- +/// persisting), and the registry's `broadcast` feeds only the +/// subscriber arms. A producer whose `begin` finds the digest already +/// live (`None`) continues store-arm-only: the incumbent's broadcast +/// carries the same bytes under CAS. +pub struct FanoutRegistry { + live: Arc, +} + +/// A broadcast's producer half: hands out chunks to the subscriber +/// arms under the broadcast's policy, and owns the broadcast's +/// liveness — dropped (via [`FanoutPublisher::finish`] or plain +/// drop), the entry is removed: late joiners see `subscribe` → +/// `None`, existing arms see their channels close. The two ends are +/// indistinguishable to subscribers by design: either way the +/// degrade is the verified get (the bytes are identical under CAS +/// whether the put committed or never was). +pub struct FanoutPublisher { + live: Arc, + digest: Key, +} + +impl FanoutPublisher { + /// The digest the broadcast serves. + pub fn digest(&self) -> &Key { + &self.digest + } + + /// Hand one chunk to the subscriber arms under the broadcast's + /// policy: `Stall` awaits a full arm (backpressure, the put's + /// form), `Drop` drops it instead and unblocks the rest. Arms + /// whose consumers left (closed channels) are removed either way + /// — the registry never grows unbounded on the arms. + /// + /// Broadcast to an already-ended broadcast is a no-op (the + /// publisher outliving its own end cannot happen — `finish` + /// consumes it — but a `Drop`-policy subscriber removal racing a + /// broadcast's tail may empty the arm list; the entry, not the + /// empty set, is the signal). + pub async fn broadcast(&self, chunk: FanoutChunk) { + let (policy, arms) = { + let broadcasts = self.live.broadcasts.lock(); + let Some(broadcast) = broadcasts.get(&self.digest) else { + return; + }; + (broadcast.policy, broadcast.arms.clone()) + }; + let mut removed: Vec> = Vec::new(); + match policy { + SlowSubscriberPolicy::Stall => { + for arm in &arms { + if arm.send(Arc::clone(&chunk)).await.is_err() { + removed.push(arm.clone()); + } + } + } + SlowSubscriberPolicy::Drop => { + for arm in &arms { + if arm.try_send(Arc::clone(&chunk)).is_err() { + removed.push(arm.clone()); + } + } + } + } + if !removed.is_empty() { + let mut broadcasts = self.live.broadcasts.lock(); + if let Some(broadcast) = broadcasts.get_mut(&self.digest) { + broadcast + .arms + .retain(|kept| !removed.iter().any(|gone| gone.same_channel(kept))); + } + } + } + + /// End the broadcast — the completed-put form: the entry is + /// removed, late joiners degrade, existing arms see channel end. + /// + /// Drop ends the broadcast identically (the abandoned form); the + /// explicit verb is the put handler's own completion shape. + pub fn finish(self) {} +} + +impl Drop for FanoutPublisher { + fn drop(&mut self) { + self.live.broadcasts.lock().remove(&self.digest); + } +} + +impl FanoutRegistry { + /// A fresh registry — no broadcasts live. + /// + /// `max_subscribers` bounds the subscriber arms per digest + /// (excess `subscribe` calls degrade — the bounded-subscriber + /// discipline); `channel_bound` is each arm's channel capacity + /// (the POC's 8: a slow subscriber stalls/drops per policy; + /// `0` is the legal rendezvous form — every chunk synchronizes). + pub fn new(max_subscribers: usize, channel_bound: usize) -> FanoutRegistry { + FanoutRegistry { + live: Arc::new(FanoutLive { + broadcasts: Mutex::new(BTreeMap::new()), + max_subscribers, + channel_bound, + }), + } + } + + /// Begin a broadcast for `digest` under `policy` — `None` when a + /// broadcast for the digest is already live: the caller continues + /// store-arm-only (see the registry doc). + pub fn begin(&self, digest: &Key, policy: SlowSubscriberPolicy) -> Option { + let mut broadcasts = self.live.broadcasts.lock(); + if broadcasts.contains_key(digest) { + return None; + } + broadcasts.insert( + digest.clone(), + LiveBroadcast { + policy, + arms: Vec::new(), + }, + ); + Some(FanoutPublisher { + live: Arc::clone(&self.live), + digest: digest.clone(), + }) + } + + /// Tap the digest's live broadcast — a subscriber arm's feed. + /// `None` is the **late-joiner signal**: no broadcast is live + /// (absent, ended, or at the subscriber bound) and the caller + /// degrades to the normal post-commit verified `get` — + /// identical bytes under CAS. + /// + /// A tap receives the chunks broadcast from the tap moment onward + /// (a live tap mid-stream has no prefix — the A3 shape's + /// subscriber set is the tap-time set): a consumer needing the + /// whole blob verifies through the get path regardless, so a + /// partial feed is never load-bearing — the end signal plus the + /// verified get is the contract. + pub fn subscribe(&self, digest: &Key) -> Option { + let mut broadcasts = self.live.broadcasts.lock(); + let broadcast = broadcasts.get_mut(digest)?; + if broadcast.arms.len() >= self.live.max_subscribers { + return None; + } + let (sender, receiver) = mpsc::channel(self.live.channel_bound); + broadcast.arms.push(sender); + Some(receiver) + } +} + +/// The module's configuration — what an embedder tunes; the handlers +/// capture the context, not the knobs. +#[derive(Debug, Clone)] +pub struct OpsConfig { + /// The fanout's per-digest subscriber-arm bound + /// ([`FanoutRegistry::new`]). + pub fanout_max_subscribers: usize, + /// The fanout's per-arm channel bound — the backpressure + /// discipline's knob (the POC's 8). + pub fanout_channel_bound: usize, +} + +impl Default for OpsConfig { + fn default() -> OpsConfig { + OpsConfig { + fanout_max_subscribers: 64, + fanout_channel_bound: 8, + } + } +} + +/// The ops-layer context — the module's state bound to one store +/// (ADR-001 §3's placement; the substrate this crate's ops surface +/// rides): the handler pairs (later tasks) capture this, the module +/// config, the token registry, the namespace converters, and the +/// fanout registry. +/// +/// The store access is the public facade plus the crate-internal +/// renewal seam (`Store::ops_pin_renew`/`Store::ops_pin_check`) +/// — no backend side-door, no engine types, no public-API widening. +pub struct OpsContext { + store: Arc, + tokens: TokenRegistry, + converters: Mutex>>, + fanout: FanoutRegistry, +} + +impl OpsContext { + /// Bind the substrate to a store instance. The store rides as + /// `Arc` (the facade is `&self`-shaped over async ops); this + /// context is what the handler pairs capture. + pub fn new(store: Arc, config: OpsConfig) -> OpsContext { + OpsContext { + store, + tokens: TokenRegistry::new(), + converters: Mutex::new(BTreeMap::new()), + fanout: FanoutRegistry::new(config.fanout_max_subscribers, config.fanout_channel_bound), + } + } + + /// The bound store — the facade access (get/put/stat/read_range/ + /// has/list + the GC protocol verbs; the ADR-001 §4 consumer + /// posture). + pub fn store(&self) -> &Arc { + &self.store + } + + /// The token registry ([`OpsContext::renew`]/[`OpsContext::confirm`] + /// act on it; the handlers mint through the context verbs). + pub fn tokens(&self) -> &TokenRegistry { + &self.tokens + } + + /// The fanout registry (the producer/consumer wiring of + /// ops-put/ops-fetch rides it; the mechanics are + /// registry-level). + pub fn fanout(&self) -> &FanoutRegistry { + &self.fanout + } + + /// The bound store's fleet declaration's instance id — the + /// `blobs/put` handler's `owner` half for [`OpsContext::mint_fleet`]; + /// `None` on every non-fleet store (where minting is custodial). + pub fn fleet_owner(&self) -> Option> { + self.store.fleet().map(|fleet| fleet.owner().to_vec()) + } + + /// Bind or re-bind the namespace's converter — the embedder's + /// registration of the act its registry performs at conversion. + /// `true` when a previous binding was replaced. The binding is + /// checked at conversion time: a confirm whose namespace is + /// unbound is refused (`false`) and the token's mapping survives. + pub fn bind_converter(&self, namespace: &str, converter: Arc) -> bool { + self.converters + .lock() + .insert(namespace.to_owned(), converter) + .is_some() + } + + /// **Fleet minting**: the custody-validated fleet form — the + /// mapping only (`owner` must be the bound store's instance id — + /// a foreign owner is refused at the mint, never trusted). + /// Refused on a non-fleet store: the pin row's expiry is the + /// fleet protection, and a guard-free entry over an in-process + /// pin would be protection-less. + /// + /// The put handler's guard is not passed here on purpose: its + /// **drop at response completion** is the handler's own act + /// (ADR-010 §4 — the TTL machinery stays alive on the ops path; + /// the token's whole-life guard custody is the non-fleet form, + /// and holding it under fleet would dead-code the TTL machinery). + pub fn mint_fleet(&self, namespace: &str, key: Key, owner: &[u8]) -> Result { + let Some(fleet) = self.store.fleet() else { + return Err(Error::Io( + "fleet-token minting refused: the bound store is not fleet-declared \ + (ADR-012 §2) — a non-fleet serving node's put mints the custodial form, \ + whose guard the registry holds" + .to_owned(), + )); + }; + if fleet.owner() != owner { + return Err(Error::Io( + "fleet-token minting refused: the owner is not the bound store's \ + instance id (ADR-011's vocabulary: the pin row's identity half is \ + the minting instance)" + .to_owned(), + )); + } + self.tokens.mint_fleet(namespace, key, owner.to_vec()) + } + + /// **Custodial minting**: the non-fleet form (any engine; + /// ADR-012 §2's generalization) — the registry **takes custody** + /// of the put's guard: the token is the in-process guard's + /// handle, nothing expires, until the conversion consumes it + /// (conversion-confirmed) or the process ends. Refused on a + /// declared fleet store, whose put path keeps the guard for + /// response completion only and whose protection is the row's + /// expiry — a custodially-held guard under fleet would hold the + /// TTL machinery dead exactly as ADR-010 §4 rejected. + pub fn mint_custodial(&self, namespace: &str, key: Key, guard: Pin) -> Result { + if self.store.fleet().is_some() { + return Err(Error::Io( + "custodial-token minting refused: the bound store is fleet-declared \ + (ADR-012 §2) — the fleet put path's guard drops at response completion, \ + the pin row's expiry is the protection, and the mint is the fleet form" + .to_owned(), + )); + } + self.tokens.mint_custodial(namespace, key, guard) + } + + /// **The renewal probe's substrate** (the have handler's renewal + /// form, ADR-012 §2's three-step duty — verify-the-mapping-then- + /// renew, the module doc's decided composition): token lookup + /// (the validity domain), the one-token-one-row digest check, and + /// — fleet — the row's classification then the renewal through + /// the store core; non-fleet — the answer from custody. + /// + /// Every non-renewed outcome answers the wire as `present: false` + /// with no renewal occurred; the putter's backstop is the re-put. + pub async fn renew(&self, token: &PinToken, digest: &Key) -> RenewalOutcome { + let Some(target) = self.renewal_target(token) else { + return RenewalOutcome::UnknownToken; + }; + if &target.key != digest { + return RenewalOutcome::DigestMismatch; + } + match target.custody { + RenewalCustody::Held => RenewalOutcome::Renewed, + RenewalCustody::Fleet { owner } => self.renew_fleet(&target.key, &owner).await, + } + } + + /// The fleet arm's composition: the store core's + /// `ops_pin_check` first (the row-classification verify — no + /// renewal write on a row not live), then `ops_pin_renew` whose + /// never-resurrect bool is the final answer (a check/renew race + /// lands on its `false` the same way). The ops renewal access is + /// the store core's; here the outcome only folds. + #[cfg(feature = "postgres")] + async fn renew_fleet(&self, key: &Key, owner: &[u8]) -> RenewalOutcome { + use crate::facade::PinState; + + match self.store.ops_pin_check(&key.as_bytes()).await { + Ok(PinState::Protected) => {} + Ok(PinState::Unprotected) => return RenewalOutcome::Expired, + Err(_) => return RenewalOutcome::Expired, + } + match self.store.ops_pin_renew(&key.as_bytes(), owner).await { + Ok(true) => RenewalOutcome::Renewed, + Ok(false) | Err(_) => RenewalOutcome::Expired, + } + } + + #[cfg(not(feature = "postgres"))] + async fn renew_fleet(&self, key: &Key, owner: &[u8]) -> RenewalOutcome { + let _ = (key, owner); + RenewalOutcome::Expired + } + + /// The registry peek the renewal rides: the mapping's key and the + /// custody branch, cloned out under the lock (never held across + /// the fleet arm's engine awaits; the guards themselves are not + /// cloneable — the custody is carried as the branch's shape). + fn renewal_target(&self, token: &PinToken) -> Option { + self.tokens + .tickets + .lock() + .get(token.as_bytes()) + .map(|ticket| RenewalTarget { + key: ticket.key.clone(), + custody: match &ticket.custody { + PinCustody::Held(_) => RenewalCustody::Held, + PinCustody::Fleet { owner } => RenewalCustody::Fleet { + owner: owner.clone(), + }, + }, + }) + } + + /// **The conversion trigger's verb** (the embedder-called confirm; + /// the module doc's decided design point): given the namespace and + /// token, invoke the namespace's converter with the mapping's key — + /// the guarded form through the ordered-step conversion (the + /// guard's release and the registration are one atomic step), the + /// fleet form as the registration act itself — and retire the + /// mapping (conversion-confirmed's retirement arm). + /// + /// Refusals (`false`), mapping intact: an unknown/retired token, a + /// namespace that does not match the minting namespace (the mint + /// records it), or an unbound converter. The mapping retires only + /// on a successful conversion. + /// + /// The fleet conversion retires the token *at* the confirmation: + /// from then on the entry's protection is the registered liveness + /// source, a renewal probe for the token answers `present: false` + /// (the putter's signal to stop renewing; a continued probe's + /// backstop — the re-put — is CAS-idempotent and harmless). + pub fn confirm(&self, namespace: &str, token: &PinToken) -> bool { + let ticket = { + let mut tickets = self.tokens.tickets.lock(); + match tickets.remove(token.as_bytes()) { + Some(ticket) => ticket, + None => return false, + } + }; + if ticket.namespace != namespace { + self.tokens.tickets.lock().insert(*token.as_bytes(), ticket); + return false; + } + let converter = { + let converters = self.converters.lock(); + converters.get(&ticket.namespace).cloned() + }; + let Some(converter) = converter else { + self.tokens.tickets.lock().insert(*token.as_bytes(), ticket); + return false; + }; + match ticket.custody { + PinCustody::Held(pin) => self + .store + .gc() + .convert(pin, || converter.convert(&ticket.key)), + PinCustody::Fleet { .. } => converter.convert(&ticket.key), + } + true + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::hashing::Digest; + + struct RecordingConverter { + converted: Mutex>, + } + + impl RecordingConverter { + fn captured(&self) -> Vec { + self.converted.lock().clone() + } + } + + impl PinConverter for RecordingConverter { + fn convert(&self, key: &Key) { + self.converted.lock().push(key.clone()); + } + } + + fn key_of(seed: &[u8]) -> Key { + Key::from(Digest::derive(seed, seed.len() as u64)) + } + + /// The CSPRNG minting-form's shape (the design point): 32 opaque + /// bytes, unique per mint, wire form hex, malformed hex ingress + /// rejected as not-a-token (never an error), and the Debug mask + /// keeps the capability out of rendered output. + #[test] + fn token_mints_unguessable_and_the_hex_wire_form_round_trips() { + let first = PinToken::generate().unwrap(); + let second = PinToken::generate().unwrap(); + assert_ne!( + first, second, + "CSPRNG minting never collides in the same process" + ); + assert_eq!(first.as_bytes().len(), 32); + + let token_hex = first.to_hex(); + assert_eq!(token_hex.len(), 64); + assert!(token_hex + .chars() + .all(|c| c.is_ascii_digit() || ('a'..='f').contains(&c))); + assert_eq!(PinToken::from_hex(&token_hex), Some(first)); + assert_eq!(PinToken::from_hex(&token_hex.to_uppercase()), Some(first)); + + let wrong_length = format!("{token_hex}ab"); + assert_eq!(PinToken::from_hex("0"), None); + assert_eq!(PinToken::from_hex("non-hexadecimal-material!!"), None); + assert_eq!(PinToken::from_hex(&wrong_length), None); + + assert_eq!(format!("{first:?}"), "PinToken(#redacted)"); + assert!(!format!("{first:?}").contains(&token_hex)); + } + + #[cfg(any(feature = "mem", feature = "sqlite"))] + async fn test_store() -> Arc { + #[cfg(feature = "mem")] + let store = Store::mem_only(); + #[cfg(not(feature = "mem"))] + let store = { + let path = std::path::PathBuf::from(format!( + "/tmp/opencode/alkblobs-ops-foundation-{}-{}.db", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_nanos() + )); + Store::kv_only_sqlite(path, 1024, std::time::Duration::ZERO) + .await + .unwrap() + }; + Arc::new(store) + } + + /// The custodial custody split end to end (ADR-012 §2's non-fleet + /// form): the registry holds the guard, the renewal answers from + /// custody (nothing expires), the one-token-one-row digest check, + /// the refusals keep the mapping intact, and the confirm verb + /// converts through the ordered-step primitive — the guard's + /// release and the converter's registration — and retires the + /// mapping. + #[cfg(any(feature = "mem", feature = "sqlite"))] + #[tokio::test] + async fn custodial_custody_holds_the_guard_and_converts_at_the_confirmation() { + let store = test_store().await; + let ctx = OpsContext::new(Arc::clone(&store), OpsConfig::default()); + let key = key_of(b"custodial"); + let other = key_of(b"other-digest"); + let unknown = PinToken::from_hex(&"0".repeat(64)).unwrap(); + + let pin = store.gc().acquire(&key).await; + let token = ctx.mint_custodial("ns", key.clone(), pin).unwrap(); + + assert_eq!( + store.gc().pin_count(&key), + 1, + "the registry holds the guard" + ); + assert!(ctx.tokens().holds(&token)); + + assert_eq!(ctx.renew(&token, &key).await, RenewalOutcome::Renewed); + assert!(ctx.renew(&token, &key).await.present()); + assert_eq!( + ctx.renew(&token, &other).await, + RenewalOutcome::DigestMismatch + ); + assert_eq!( + ctx.renew(&unknown, &key).await, + RenewalOutcome::UnknownToken + ); + assert_eq!( + store.gc().pin_count(&key), + 1, + "the refusals left the custody intact" + ); + + assert!( + !ctx.confirm("other-namespace", &token), + "the minting namespace mismatches" + ); + assert!( + ctx.tokens().holds(&token), + "the refused confirm kept the mapping" + ); + + assert!( + !ctx.confirm("ns", &token), + "an unbound converter refuses, mapping intact" + ); + assert!(ctx.tokens().holds(&token)); + + let converted = Arc::new(RecordingConverter { + converted: Mutex::new(Vec::new()), + }); + assert!(!ctx.bind_converter("ns", Arc::clone(&converted) as Arc)); + assert!(ctx.confirm("ns", &token)); + assert_eq!( + converted.captured(), + vec![key.clone()], + "the converter received the mapping's key" + ); + assert_eq!( + store.gc().pin_count(&key), + 0, + "the ordered-step conversion released the held guard" + ); + assert!( + !ctx.tokens().holds(&token), + "conversion-confirmed retires the mapping" + ); + assert_eq!( + ctx.renew(&token, &key).await, + RenewalOutcome::UnknownToken, + "a retired token's renewal probe is the validity-domain refusal" + ); + assert!( + !ctx.confirm("ns", &token), + "a retired token confirms nowhere twice" + ); + } + + /// The fleet custody's registry-level form (ADR-010 §4's mapping + /// only): no guard is held (the put handler's guard drops at + /// response completion — its act, outside the mint), the renewal + /// probe on this shape degrades closed (the pin-row access is the + /// fleet store core's; on a non-fleet bound store the answer is + /// present: false, never an error), and conversion retires the + /// mapping through the converter act alone. + #[cfg(any(feature = "mem", feature = "sqlite"))] + #[tokio::test] + async fn fleet_custody_holds_no_guard_and_the_renewal_degrades_closed() { + let store = test_store().await; + let ctx = OpsContext::new(Arc::clone(&store), OpsConfig::default()); + let key = key_of(b"fleet-form"); + let other = key_of(b"other-fleet"); + + let token = ctx + .tokens() + .mint_fleet("ns", key.clone(), b"instance-owner".to_vec()) + .unwrap(); + assert_eq!( + store.gc().pin_count(&key), + 0, + "the fleet form records the mapping; no guard rides it" + ); + + assert_eq!( + ctx.renew(&token, &other).await, + RenewalOutcome::DigestMismatch + ); + assert_eq!( + ctx.renew(&token, &key).await, + RenewalOutcome::Expired, + "a fleet mapping's renewal reaches the store core's fleet host and, on a \ + non-fleet-bound context shape, folds closed" + ); + + let converted = Arc::new(RecordingConverter { + converted: Mutex::new(Vec::new()), + }); + ctx.bind_converter("ns", Arc::clone(&converted) as Arc); + assert!(ctx.confirm("ns", &token)); + assert_eq!( + converted.captured(), + vec![key.clone()], + "the fleet conversion is the converter act itself (no guard to release)" + ); + assert!( + !ctx.tokens().holds(&token), + "conversion-confirmed retires the mapping" + ); + } + + /// The custody-mint validations (ADR-010 §4's total split, the + /// constructible half): the fleet form is refused over a + /// non-fleet-declared store — the pin row's expiry cannot be the + /// protection for a mapping-only token where no row exists. + #[cfg(any(feature = "mem", feature = "sqlite"))] + #[tokio::test] + async fn mint_fleet_is_refused_over_a_non_fleet_store() { + let store = test_store().await; + let ctx = OpsContext::new(Arc::clone(&store), OpsConfig::default()); + assert!(ctx + .mint_fleet("ns", key_of(b"attempted"), b"owner") + .is_err()); + } + + /// The fleet-mint owner validation sits at the mint: a foreign + /// owner is refused before any mapping lands (the pin row's + /// identity half is the minting instance's id). + #[test] + fn fleet_owner_lookup_follows_the_bound_store_and_the_registry_mints_standalone() { + let tokens = TokenRegistry::new(); + let without = PinToken::from_hex(&"7".repeat(64)).unwrap(); + assert!(!tokens.holds(&without)); + let minted = tokens + .mint_fleet("ns", key_of(b"standalone"), b"instance".to_vec()) + .unwrap(); + assert!(tokens.holds(&minted)); + assert_ne!(minted, without); + } + + /// The converter binding is per-namespace, and a re-bind replaces: + /// the latest bound converter is the one the conversion invokes. + #[cfg(any(feature = "mem", feature = "sqlite"))] + #[tokio::test] + async fn converter_bindings_are_per_namespace_and_rebinding_replaces() { + let store = test_store().await; + let ctx = OpsContext::new(Arc::clone(&store), OpsConfig::default()); + let first = Arc::new(RecordingConverter { + converted: Mutex::new(Vec::new()), + }); + let second = Arc::new(RecordingConverter { + converted: Mutex::new(Vec::new()), + }); + assert!(!ctx.bind_converter("ns", Arc::clone(&first) as Arc)); + assert!(ctx.bind_converter("ns", Arc::clone(&second) as Arc)); + + let pin = store.gc().acquire(&key_of(b"rebind")).await; + let token = ctx.mint_custodial("ns", key_of(b"rebind"), pin).unwrap(); + assert!(ctx.confirm("ns", &token)); + assert!( + first.captured().is_empty(), + "the replaced binding never converted" + ); + assert_eq!(second.captured(), vec![key_of(b"rebind")]); + } + + /// The fanout registry's subscriber arms: taps of a live broadcast + /// receive the producer's stream in full and in order (the A3 + /// serve-while-persisting shape's receive side), the broadcast + /// ends remove the entry, and the late joiner's signal is the + /// subscribe answer's `None` — the degrade to the verified get. + #[tokio::test] + async fn fanout_subscriber_arms_receive_the_full_stream_and_late_joiners_degrade() { + let registry = FanoutRegistry::new(8, 16); + let digest = key_of(b"streamed"); + let no_broadcast = key_of(b"not-a-broadcast"); + + assert!( + registry.subscribe(&no_broadcast).is_none(), + "no live broadcast: the late-joiner/absent degrade signal" + ); + + let publisher = registry + .begin(&digest, SlowSubscriberPolicy::Stall) + .unwrap(); + assert_eq!(publisher.digest(), &digest); + let mut first = registry.subscribe(&digest).unwrap(); + let mut second = registry.subscribe(&digest).unwrap(); + + for chunk in 0u8..4 { + publisher.broadcast(Arc::new(vec![chunk])).await; + } + publisher.finish(); + + for chunk in 0u8..4 { + assert_eq!( + first.recv().await, + Some(Arc::new(vec![chunk])), + "subscriber arms receive in order" + ); + } + for chunk in 0u8..4 { + assert_eq!(second.recv().await, Some(Arc::new(vec![chunk]))); + } + assert_eq!( + first.recv().await, + None, + "the broadcast end is the degrade signal" + ); + assert!( + registry.subscribe(&digest).is_none(), + "an ended broadcast is gone" + ); + } + + /// The Stall policy's backpressure (POC A3's bounded-channel + /// discipline): a slow subscriber's full channel stalls the + /// broadcast, which unblocks when the consumer drains. + #[tokio::test] + async fn stall_policy_backpressures_a_slow_subscriber() { + let registry = FanoutRegistry::new(8, 1); + let digest = key_of(b"stalled"); + let publisher = registry + .begin(&digest, SlowSubscriberPolicy::Stall) + .unwrap(); + let mut subscriber = registry.subscribe(&digest).unwrap(); + + publisher.broadcast(Arc::new(vec![0u8])).await; + let slow = tokio::time::timeout( + std::time::Duration::from_millis(50), + publisher.broadcast(Arc::new(vec![1u8])), + ) + .await; + assert!( + slow.is_err(), + "the full bounded channel stalls the broadcast" + ); + assert_eq!(subscriber.recv().await, Some(Arc::new(vec![0u8]))); + publisher.broadcast(Arc::new(vec![1u8])).await; + assert_eq!(subscriber.recv().await, Some(Arc::new(vec![1u8]))); + publisher.finish(); + } + + /// The Drop policy is the drop-and-late-join escape: a full arm is + /// dropped (its consumer degrades at channel end), the broadcast + /// unblocks, and the freed bound admits a late join to the still- + /// live broadcast. + #[tokio::test] + async fn drop_policy_drops_a_stalled_arm_and_the_freed_bound_admits_a_late_join() { + let registry = FanoutRegistry::new(2, 1); + let digest = key_of(b"dropped-arm"); + let publisher = registry.begin(&digest, SlowSubscriberPolicy::Drop).unwrap(); + let mut stalled = registry.subscribe(&digest).unwrap(); + + publisher.broadcast(Arc::new(vec![0u8])).await; + publisher.broadcast(Arc::new(vec![1u8])).await; + publisher.broadcast(Arc::new(vec![2u8])).await; + + assert_eq!( + stalled.recv().await, + Some(Arc::new(vec![0u8])), + "drainage carries the buffered chunk before the channel end" + ); + assert_eq!( + stalled.recv().await, + None, + "the dropped arm's channel end degrades" + ); + + let mut rejoined = registry.subscribe(&digest).unwrap(); + publisher.broadcast(Arc::new(vec![3u8])).await; + assert_eq!(rejoined.recv().await, Some(Arc::new(vec![3u8]))); + publisher.finish(); + } + + /// The publisher's abandonment (drop without finish) ends the + /// broadcast the same way: the entry is removed, the subscriber + /// sees channel end, and the late joiner's subscribe degrades to + /// `None`. + #[tokio::test] + async fn publisher_drop_aborts_the_broadcast_and_late_joiners_degrade() { + let registry = FanoutRegistry::new(8, 2); + let digest = key_of(b"abandoned"); + let publisher = registry + .begin(&digest, SlowSubscriberPolicy::Stall) + .unwrap(); + let mut subscriber = registry.subscribe(&digest).unwrap(); + drop(publisher); + assert_eq!( + subscriber.recv().await, + None, + "the abandoned broadcast ends" + ); + assert!(registry.subscribe(&digest).is_none()); + } + + /// The per-digest subscriber bound: excess subscribes take the + /// degrade signal (the bounded-subscriber discipline — the count, + /// not the policy, bounds who may tap). + #[tokio::test] + async fn subscriber_bound_excess_degrades() { + let registry = FanoutRegistry::new(1, 4); + let digest = key_of(b"bounded"); + let _publisher = registry.begin(&digest, SlowSubscriberPolicy::Drop).unwrap(); + assert!(registry.subscribe(&digest).is_some()); + assert!( + registry.subscribe(&digest).is_none(), + "the second arm exceeds the bound: the degrade signal" + ); + } +} diff --git a/src/store.rs b/src/store.rs index 396d480..8415ba7 100644 --- a/src/store.rs +++ b/src/store.rs @@ -1834,6 +1834,55 @@ impl Store { pub(crate) fn gc(&self) -> &GcHome { &self.gc } + + /// **The ops module's crate-internal renewal access — `pin_renew`** + /// (ADR-010 §4: "the ops module is in-crate and crosses the seam by + /// construction, the same way the put handler holds the `Pin` + /// guard"): the have handler's renewal step's store-core crossing, + /// routed through the fleet host. This verb is the crossing's named + /// shape — the store core owns the seam's reachability, so a + /// *consumer* never reaches the `EngineFleetState` clause, and no + /// public API widens for this (ADR-001 §4's facade-consumer rule: + /// this and [`Store::gc`] are the ops module's only store-core + /// access crate-internally — the facade plus exactly this seam). + /// + /// Non-fleet stores answer `Err`: the pin row's renewal is the + /// fleet form's act (under `fleet: false` the token is the + /// in-process guard's custody — nothing expires — and the renewal + /// probe answers from custody, never through this verb; the + /// ops-layer custody branch is upstream of this call). + #[cfg(all(feature = "ops", feature = "postgres"))] + pub(crate) async fn ops_pin_renew(&self, key: &[u8], owner: &[u8]) -> Result { + match self.fleet_sql_host() { + Some(host) => EngineFleetState::pin_renew(&**host, key, owner).await, + None => Err(Error::Io( + "pin renewal outside a declared fleet store: under fleet: false the token \ + is the in-process guard's custody (nothing expires) — the renewal probe \ + answers from the ops layer's custody, never through this verb" + .to_owned(), + )), + } + } + + /// **The ops module's crate-internal renewal access — `pin_check`** + /// (ADR-010 §2's engine-state seam; the ADR-008 row classifier): + /// the have handler's renewal-verification step's store-core + /// crossing (the clause's named consumer — the wave-6 review's + /// W6-3 note), routed through the fleet host under the same + /// ownership rule as [`Store::ops_pin_renew`]: store-core-owned, + /// never public API, never a backend side-door. + #[cfg(all(feature = "ops", feature = "postgres"))] + pub(crate) async fn ops_pin_check(&self, key: &[u8]) -> Result { + match self.fleet_sql_host() { + Some(host) => EngineFleetState::pin_check(&**host, key).await, + None => Err(Error::Io( + "pin check outside a declared fleet store: the pin-row classification \ + is the fleet form's act (ADR-010 §2) — under fleet: false the in-process \ + guard's custody is the classification, and the ops layer holds it" + .to_owned(), + )), + } + } } /// The read path (store-api.md §Read path): `get`, `stat`, diff --git a/tasks/ops-foundation.md b/tasks/ops-foundation.md index 8bc7032..5efa0f3 100644 --- a/tasks/ops-foundation.md +++ b/tasks/ops-foundation.md @@ -1,7 +1,7 @@ --- id: ops-foundation name: the ops module's substrate — the token layer, the namespace binding, and the fanout registry -status: pending +status: completed depends_on: [] scope: moderate risk: medium @@ -98,30 +98,30 @@ below — no backend side-door, no engine types. ## Acceptance Criteria -- [ ] `src/ops.rs` carries the module structure under feature `ops`: +- [x] `src/ops.rs` carries the module structure under feature `ops`: the ops context, the token layer, the namespace binding, the fanout registry — no op handlers yet (they land in ops-json-ops/ops-fetch/ops-put) -- [ ] The token layer: opaque, unguessable minting (the CSPRNG design +- [x] The token layer: opaque, unguessable minting (the CSPRNG design point decided and recorded); the token↔(key, owner) mapping with process lifetime; validity domain = the minting node (a per-process registry; no cross-node form exists) -- [ ] The guard-custody design decided and recorded: fleet — the +- [x] The guard-custody design decided and recorded: fleet — the guard drops at response completion (the row's expiry is the liveness signal); non-fleet — the registry holds the guard (nothing expires) until conversion -- [ ] The namespace→liveness-source binding: per-namespace converter +- [x] The namespace→liveness-source binding: per-namespace converter registration riding `RegistryCore::convert`; the conversion trigger design decided and recorded -- [ ] The crate-internal renewal access: the ops module reaches +- [x] The crate-internal renewal access: the ops module reaches `pin_renew`/`pin_check` through the fleet host (in-crate seam crossing; no public-API widening; no backend side-door) -- [ ] The fanout registry: per-digest broadcast, bounded subscriber +- [x] The fanout registry: per-digest broadcast, bounded subscriber channels, the late-joiner degrade signal, the drop-and-late-join policy point recorded -- [ ] New dependencies (`serde`, `serde_json`, the CSPRNG) are +- [x] New dependencies (`serde`, `serde_json`, the CSPRNG) are `ops`-gated; the default build's dependency tree is unchanged -- [ ] Unit tests green; `cargo test` (default) green with the module +- [x] Unit tests green; `cargo test` (default) green with the module compiled out; `cargo test --features ops` green; `cargo clippy --all-targets -- -D warnings` and `cargo fmt --check` green on both shapes; `cargo doc --no-deps` @@ -153,8 +153,131 @@ below — no backend side-door, no engine types. ## Notes -> To be filled by implementation agent +> Landed as `src/ops.rs` under feature `ops` (the default build +> compiles the module out; `cargo tree` default-vs-HEAD verified +> byte-unchanged). The store access is exactly the facade plus one +> crate-internal seam: `Store::ops_pin_renew`/`Store::ops_pin_check` +> (store.rs, `pub(crate)`, cfg all(ops, postgres)) routing through the +> existing `fleet_sql_host` — the store core owns the crossing; no +> engine type or `Backend` name reaches the ops module; no public API +> widened. The fanout registry (`FanoutRegistry`) and the token +> registry (`TokenRegistry`) are constructible standalone; the +> custody-validated mint verbs live on the context. +> +> **Decided design points (recorded on the module):** +> +> - **Token minting form: a CSPRNG id** — `PinToken` = 32 bytes from +> `getrandom` (v0.4, the same version alkcall 0.8 pulls — no +> duplication). Rejected: a counter (a token is a capability; a +> guessable capability is broken) and derivable forms (key/owner- +> derived is equally guessable and no longer opaque). Wire form hex +> (lowercase canonical, decode case-insensitive); `Debug` masks the +> bytes — the token never lands in logs/panics/errors. Malformed hex +> ingress returns `None` (the have handler maps it to the +> unknown-token answer, present: false — never an error; no new +> error-family member invented). +> - **Mapping retention**: conversion-confirmed (the confirm verb's +> success removes the entry) or process end; no expiry machinery on +> the mapping — the pin row's expiry is the fleet liveness signal, +> nothing expires non-fleet. +> - **Custody split**: `OpsContext::mint_fleet` (mapping only; owner +> validated against the bound store's instance id; refused over a +> non-fleet-declared store) vs `OpsContext::mint_custodial` (the +> registry takes the `Pin` guard; refused on a declared fleet +> store). The fleet put handler's guard-drop-at-response-completion +> stays the handler's act (the mint deliberately takes no guard); +> enforcement of the mirror refusal is structural at the mint verbs. +> - **Conversion trigger: an embedder-called confirm verb** +> (`OpsContext::confirm(namespace, token)`). Probe-triggered +> conversion rejected: have is a read-gated op whose caller is the +> remote putter — a protection-lifecycle mutation keyed off it would +> give a read op write semantics, while the hand-over contract's +> duty text puts the observation (have returns present after +> registration) and the act (confirm) on the embedder. Refusals +> (unknown/mismatched-namespace/unbound-converter) keep the mapping +> intact; the guarded conversion rides `GcHome::convert`'s ordered +> step (release + registration under the one arbitration-lock +> section; the converter's act is documented as quick, non-blocking +> — the `LivenessSource::contains` discipline), the fleet conversion +> invokes the converter act with no pin to release. +> - **Renewal probe composition: verify-the-mapping-then-renew** — +> (a) token lookup (validity domain; unknown → `UnknownToken` → +> present: false, the peers' refusal form), (b) the one-token-one- +> row digest check (`DigestMismatch`), (c) the store core's +> `ops_pin_check` (the W6-3 named consumer) — a row not live +> answers `Expired` with no renewal write, (d) `ops_pin_renew` — +> its never-resurrect bool is the final answer (a check/renew race +> lands on the same Expired). Renew-answers-the-verification +> rejected: the verify steps are where the validity domain and the +> row bind are enforced. Non-fleet answers from custody (nothing +> expires; no engine call). All engine failures fold to +> `present: false` — the probes stay probes (ADR-008's discipline). +> - **Fanout mechanics**: per-digest (`Key`) map of live broadcasts; +> `begin` → exclusive `FanoutPublisher` (`None` when already live — +> the producer continues store-arm-only, CAS guarantees identical +> bytes); `subscribe` → bounded-channel arm or `None` (the +> late-joiner/absent/bound-exhausted degrade signal; taps receive +> from the tap moment onward — the A3 tap-time-set shape, the whole +> blob's authority stays the verified get); per-broadcast +> `SlowSubscriberPolicy::{Stall, Drop}` — Stall is the put's form +> (bounded backpressure, the put must land), Drop is the +> drop-and-late-join escape (a full arm is dropped, its channel +> closes, the consumer degrades); publisher finish-or-drop removes +> the entry, existing arms see channel end, late joiners see +> `subscribe` → `None` — either way the degrade is the verified +> get. Arms are cloned out under the lock and driven outside it +> (nothing async under the lock); closed/full arms are reaped at +> broadcast time (channels never grow unbounded). +> - **The renewal-cadence boundary**: documented on the module (mechanism +> here; the TTL/3 cadence is embedder composition, ADR-013 §1 — no +> in-crate worker, none may exist). +> +> **Scope notes**: gc.rs's allow(dead_code) module doc updated — +> `pin_check`'s W6-3 "deliberately unconsumed" note now reads its +> landed state (the ops module's renewal probe, through the store +> core's `ops_pin_check`), and `convert`'s note names the confirm +> verb as the guard-custody consumer. `serde`/`serde_json` +> (features derive / default) entered as ops-gated optional deps now +> (the payload-handling posture's shape) — first consumed by +> ops-json-ops/ops-put; the default build's dependency tree is +> verified unchanged (`cargo tree` diff against HEAD). +> +> The fleet forms against a *live* fleet store (a real `pin_check` → +> `pin_renew` row cycle; the `mint_custodial`-refused-under-fleet +> arm) need the harness server (wave 6's fleet column pattern) and +> land with the handler tasks — the foundation's unit tests cover the +> custody branching (registry-level fleet entry: no guard held, +> renewal degrades closed, conversion is the converter act alone; +> custody-mint validation: the fleet mint refused over a non-fleet +> store; the custodial path end to end) on the store-free and +> mem/sqlite shapes. ## Summary -> To be filled on completion \ No newline at end of file +> Completed 2026-10-12 (wave 7, first task of the decomposition). The +> ops module's substrate landed: `OpsContext` (the `Arc` + +> config + token registry + converters + fanout binding the handler +> pairs capture), the pin-token layer (CSPRNG `PinToken`, the +> process-lifetime token↔(key, owner) mapping, validity domain = the +> minting node, retention = conversion-confirmed-or-process-end), the +> guard-custody split (`mint_fleet`: mapping-only, the guard's drop +> at response completion is the put handler's act; `mint_custodial`: +> the registry holds the guard, nothing expires), the +> namespace→liveness-source binding (`PinConverter` per-namespace +> registration; the conversion trigger is the embedder-called +> confirm verb riding the ordered-step conversion), the crate-internal +> store-core renewal access (`Store::ops_pin_renew`/`ops_pin_check` — +> the W6-3 consumer; ADR-012 §2's verify-the-mapping-then-renew probe +> composition), and the fanout registry (per-digest broadcast, +> bounded subscriber channels, the late-joiner degrade signal, the +> Stall/Drop slow-subscriber policies) — no op handlers, per the task +> text. Five unit tests' worth of mechanics green (11 test fns); +> gates green on both shapes: `cargo test` (default, 73/19/3 with the +> module compiled out), `cargo test --features ops` (84/19/3), +> `cargo test --all-features` (237/125/4), clippy +> `--all-targets -D warnings` across the default/ops/all-features/ +> pg-pure variants, `cargo fmt`, `cargo doc --no-deps` zero warnings +> on the default and gated-ops builds (the all-features build's +> count is the pre-existing private-link baseline, nothing added). +> Dependencies: `serde` (derive), `serde_json`, `getrandom 0.4` — +> all `ops`-gated; the default dependency tree verified unchanged. \ No newline at end of file