diff --git a/docs/architecture/ops-surface.md b/docs/architecture/ops-surface.md index b87d3ad..16e580e 100644 --- a/docs/architecture/ops-surface.md +++ b/docs/architecture/ops-surface.md @@ -1,7 +1,7 @@ --- status: draft -last_updated: 2026-10-03 -notes: "ADR-012 §2 — put response {token, digest}; have token-renewal form; fetch verification wording; non-fleet token scope" +last_updated: 2026-10-13 +notes: "ADR-012 §2 — put response {token, digest}; have token-renewal form; fetch verification wording; non-fleet token scope; fetch establishment ack + miss mapping" --- # Ops surface (alkcall-backed network operations) @@ -79,17 +79,37 @@ JSON control ops (`Visibility` per ADR-001 §Decision): Binary channel ops (registered via the `channel_open` marker): -- **`blobs/fetch`** (`Sub` — the server→client streaming op shape) +- **`blobs/fetch`** (`Sub` — the server→client streaming op shape; + open-op id `channels/blobs/sub`, the `alk/blobs` data-plane ALPN) — `{namespace, digest, ranges?}` in; verified bytes out: the consumer hash-checks the received whole against the carried digest (ADR-006's whole-blob verification — there are no per-chunk or per-range checks derivable under the canonical digest; ranged fetch responses carry the same ADR-006 - slice-digest convention `read_range` returns). Broadcast fanout - above the store: - one reader, store arm + subscriber arms (POC #3 finding A3); late - joiners degrade to a normal post-commit `get` — identical bytes under - CAS. Slow-subscriber policy (drop-and-late-join) is ops-layer + slice-digest convention `read_range` returns). The establishment + phase (ADR-049) carries the ack in the open reply's fields: + `{source, len?, slice_digests?}` — `source` names the serving path + (`store`/`fanout`); the whole form acks `len` from the store path + (the hash-check's expected whole) and nothing from a fanout tap (a + live put's tail; the hash-check runs at stream end); the ranged + form acks one slice digest per range, in request order, over + exactly the served bytes (over-reads clamp; the digest covers the + clamp). The fanout consult (POC #3 finding A3's consumer side): a + live broadcast → a subscriber arm taps it; no broadcast, or one + that ended (a late joiner) → the normal post-commit `get` — + identical bytes under CAS; a mid-put tap has no prefix, so the + consumer's hash-check on a partial feed fails and the re-fetch is + the degrade. Fetches never produce read-side broadcasts (recorded + against the A3 text — the put handler's stream reader is the + canonical producer; concurrent fetchers each ride their own facade + read). The miss (absent digest, no broadcast) is the typed + missing-error schema: the declared `BLOB_MISSING` contract, answered + on the wire as the establishment refusal (`channel:open_failed`, + reason `unknown_resource`, ADR-049's "the requested resource does + not exist"), the digest in the refusal's message. The need + announcement is the fetch miss — the have/need diff is caller-side; + no separate need op exists (the wire carries only concrete fetch + requests). Slow-subscriber policy (drop-and-late-join) is ops-layer policy, not a store concern. - **`blobs/put`** (`Sink` — the client→server streaming op shape) — `{namespace, digest?, len?}` offer + diff --git a/src/lib.rs b/src/lib.rs index 0412e54..9af7af6 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -32,10 +32,10 @@ pub mod ops; #[cfg(feature = "ops")] pub use ops::{ - delete_pair, have_pair, stat_pair, FanoutChunk, FanoutPublisher, FanoutReceiver, - FanoutRegistry, OpsConfig, OpsContext, PinConverter, PinToken, RenewalOutcome, - SlowSubscriberPolicy, TokenRegistry, BLOB_MISSING, GC_REFUSED, INVALID_DIGEST, - INVALID_RENEWAL_FORM, + delete_pair, fetch_openables, have_pair, stat_pair, FanoutChunk, FanoutPublisher, + FanoutReceiver, FanoutRegistry, OpsConfig, OpsContext, PinConverter, PinToken, RenewalOutcome, + SlowSubscriberPolicy, TokenRegistry, BLOBS_CHANNEL_ALPN, BLOB_MISSING, FETCH_OP_ID, GC_REFUSED, + INVALID_DIGEST, INVALID_RENEWAL_FORM, }; #[cfg(any(feature = "postgres", feature = "pg-lo"))] diff --git a/src/ops.rs b/src/ops.rs index 75c23d5..27ada66 100644 --- a/src/ops.rs +++ b/src/ops.rs @@ -4,8 +4,8 @@ //! 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 — plus the JSON control -//! ops' exported pairs; the binary (`blobs/fetch`/`blobs/put`) data -//! plane lands in the follow-on tasks. +//! ops' exported pairs and `blobs/fetch` (the binary data plane's +//! `Sub` op); `blobs/put` (the `Sink` op) lands in the follow-on task. //! //! **The ops layer is a store-facade consumer** (ADR-001 §4): its //! store access is the public facade plus exactly one crate-internal @@ -164,11 +164,32 @@ //! 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. +//! `blobs/fetch` (lands with that op — the section below); 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. +//! +//! **Named design point — the fetch is the fanout's consumer side, not +//! a producer (decided, against POC #3 finding A3's text).** The A3 +//! seam is the *put stream's* broadcast: one reader (the put handler's +//! stream), the store arm (persistence) and the subscriber arms (the +//! fetches tapping it), with late joiners degrading to the post-commit +//! verified `get` — A3 never makes the get path a rebroadcast. `blobs/ +//! fetch` keeps that exactly: a fetch taps a live broadcast or serves +//! from the store, and it **does not begin a read-side broadcast of its +//! own** — concurrent fetchers of one stored digest each ride their own +//! facade read, not one shared store read fanned out. The rejected +//! shape adds a second producer class whose lifecycle has no owner (a +//! read-side broadcast's end would be an arbitrary policy bolted onto +//! the store serve, and its arms hold memory exactly where the large +//! tier exists to avoid materializing), for a concurrency case bounded +//! by the put window the put-side broadcast already covers; under CAS +//! every one of the per-fetcher reads returns identical bytes. The put +//! handler's stream reader remains the canonical producer +//! (serve-while-persisting), and it decides the slow-subscriber policy +//! for the broadcast it begins. //! //! ## The renewal-cadence boundary (ADR-013 §1) //! @@ -228,20 +249,92 @@ //! scoping), gated by the internal authority context the embedder //! declares at registration (alkcall ADR-017: internal calls switch //! authority context, never skip ACL). +//! +//! ## `blobs/fetch` (the binary data plane's consumer side: the +//! verified `Sub` op, ops-surface.md §Op family; alkcall ADR-047/049) +//! +//! The server→client streaming op: `{namespace, digest, ranges?}` in, +//! verified bytes out over a `channel_open`-allocated channel. The op +//! rides alkcall's machinery — `channels/blobs/sub` (the standard +//! `channels//sub` shape, `OperationType::Sub`), the +//! `channel_open` marker on the `alk/blobs` data-plane ALPN +//! ([`BLOBS_CHANNEL_ALPN`]; the same ALPN `blobs/put`'s `Sink` op +//! will share), the **read** action on the namespace resource +//! (closed-by-default, exactly like stat/have), and the awaited +//! **establishment phase** ([`fetch_openables`]' establisher) whose +//! success reply fields carry the wire ack. +//! +//! **The establishment ack** (alkcall ADR-049 amendment 3's reply +//! fields): merged into the open reply beside the wrapper-owned +//! `channel_id` — `{source, len?, slice_digests?}`. `source` names the +//! serving path (`"store"` / `"fanout"` — which path served, the +//! consumer's provenance signal). The **whole form** stores ack +//! `{source: "store", len}` (the entry's length: the hash-check's +//! expected whole) and a tapped broadcast acks `{source: "fanout"}` +//! (no length: the bytes are streaming from a live put; the +//! hash-check runs at stream end). The **ranged form** acks +//! `{source: "store", slice_digests[]}` — the ADR-006 slice-digest +//! convention, SHA-256 over exactly the returned bytes, one per +//! range in request order, carried out-of-band; over-reads clamp at +//! the entry's end and the digest covers exactly the clamped bytes. +//! +//! **The serving paths** (the fanout seam's consumer side, in consult +//! order): (1) the whole form consults the store first — a present +//! entry streams through the facade's verified `get` (a store hit +//! precedes a fanout tap: a commit that raced the broadcast's end is +//! the same bytes under CAS); (2) absent, a **live broadcast** for +//! the digest exists → a **subscriber arm taps it** +//! ([`FanoutRegistry::subscribe`]) and the pump forwards the arm's +//! feed — a tap has no prefix (the tap-time set per the A3 shape), so +//! a mid-put tap serves a suffix; the consumer's hash-check fails on +//! a partial feed and the re-fetch is the degrade — post-commit, the +//! same request is the store path, identical bytes; (3) absent and no +//! broadcast (or at the subscriber bound) → **the miss**. The ranged +//! form serves the committed entry only ([`Store::read_range`]); a +//! not-yet-committed digest's ranged fetch is the miss (ranges have +//! no relationship to a broadcast's tap-window feed), never a fanout +//! tap. +//! +//! **The miss** (the need announcement; ops-surface.md's have/need +//! note — the diff is caller-side, no need op exists): a fetch of an +//! absent digest answers as the establishment refusal — alkcall +//! ADR-049's typed vocabulary, reason `unknown_resource` ("the +//! requested resource does not exist"), the digest carried in the +//! refusal's message; the declared [`BLOB_MISSING`] error definition +//! is the op's disclosed missing-error contract (the same code + +//! digest-details schema the JSON ops declare), and the wire shape a +//! consumer branches on is `channel:open_failed` + +//! `details.reason = "unknown_resource"`. Never a phantom channel +//! with an in-band error frame — that is exactly the ADR-049 +//! antipattern. +//! +//! **Verified streaming is consumer-side** (ADR-006 decision 4): the +//! server streams verified-store bytes (the facade's `get` re-derives +//! canonically on the kv tier) and performs no second verification of +//! its own stream; the *consumer* hash-checks the received whole +//! against the digest it carried. Whole-blob verification only — no +//! per-chunk checks are derivable under the canonical digest; the +//! ranged form's integrity is the slice-digest ack. Handlers touch +//! only the store facade — no backend side-door (ADR-001 §4). use std::collections::BTreeMap; use std::sync::Arc; +use alkcall::channels::operations::{ + ChannelPlan, Establishment, EstablishmentError, OpenEstablisher, OpenHandler, +}; +use alkcall::core::types::BiStream; use alkcall::protocol::wire::{CallError, ResponseEnvelope}; use alkcall::registry::registration::{make_handler, Handler}; use alkcall::registry::spec::{ - AccessControl, ErrorDefinition, OperationSpec, OperationType, Visibility, + AccessControl, ChannelOpenSpec, ErrorDefinition, OperationSpec, OperationType, Visibility, }; use parking_lot::Mutex; use serde_json::{json, Value}; +use tokio::io::AsyncWriteExt; use tokio::sync::mpsc; -use crate::facade::Pin; +use crate::facade::{Blob, Pin}; use crate::hashing::Digest; use crate::keys::Key; use crate::store::Store; @@ -945,6 +1038,15 @@ pub const INVALID_RENEWAL_FORM: &str = "INVALID_RENEWAL_FORM"; /// is pinned, liveness-live, or protect-callback-named — nothing /// deleted. pub const GC_REFUSED: &str = "GC_REFUSED"; +/// The binary data plane's ALPN (`blobs/fetch` and, when it lands, +/// `blobs/put`): the `channel_open` marker's `alk/blobs` — the +/// standard `channels/blobs/sub` op name derives it, so no explicit +/// `channel_open_alpn` rides the spec (ADR-047, the standard shapes). +pub const BLOBS_CHANNEL_ALPN: &str = "alk/blobs"; +/// `blobs/fetch`'s open-op id — the standard `channels//sub` +/// shape over [`BLOBS_CHANNEL_ALPN`] (`Sub`-typed per alkcall +/// ADR-021: the server→client streaming op kind). +pub const FETCH_OP_ID: &str = "channels/blobs/sub"; /// The digest wire form's ingress (see the section doc): the key /// layer's hex parse — lowercase 40/64-char digest hex — wrapped into @@ -1369,6 +1471,362 @@ pub fn delete_pair(ctx: Arc) -> (OperationSpec, Handler) { (spec, handler) } +/// The whole form's serving window — one facade `read_at` per pump +/// iteration, the bounded-memory streaming shape (the tier's read +/// discipline: kv slices materialized content, the large tier +/// window-reads the committed file — never a whole-entry load on the +/// large tier). +const FETCH_WINDOW: u64 = 64 * 1024; + +/// The ranged form's wire item: `{offset, len}` — the byte window the +/// consumer asks for. Over-reads clamp at the entry's end (the store's +/// pread discipline); the ack's slice digest covers exactly the served +/// bytes. +struct FetchRange { + offset: u64, + len: u64, +} + +/// The establisher's plan payload (ADR-049 amendment 2's typed-opaque +/// `ChannelPlan` — establisher and pump agree on the type, alkcall +/// never inspects it): the serving path the establishment phase chose +/// and what the pump streams. +enum FetchPlan { + /// The whole form over the store path: the entry's `Blob` read open + /// at establishment (the commit/delete race window between + /// establishment and pump is closed — committed entries are + /// immutable), streamed as windows. + Whole { blob: Blob }, + /// The ranged form: the slices read at establishment (their digests + /// ride the ack, which lands before the pump runs), streamed in + /// request order. + Ranges { slices: Vec> }, + /// The whole form over the fanout path: the subscriber arm tapped + /// at establishment; the pump forwards the arm's feed and the + /// broadcast's end closes the stream. A tap has no prefix (the + /// tap-time set) — a consumer whose hash-check fails re-fetches; + /// post-commit the same request is the store path. + Fanout { receiver: FanoutReceiver }, +} + +/// **`blobs/fetch`** (`channels/blobs/sub` over [`BLOBS_CHANNEL_ALPN`]) +/// — the verified `Sub` op (ops-surface.md §Op family; alkcall +/// ADR-047/049): `{namespace, digest, ranges?}` in; verified bytes out +/// over the `channel_open`-allocated channel. Exported as the open +/// triple the embedder assembles — the `OperationSpec` (the +/// `channel_open` marker + the **read** action on the namespace +/// resource), the establisher (the awaited establishment phase whose +/// reply fields carry the ack), and the pump handler (the data plane) — +/// registered through `ChannelCore::register_openable_with_establisher` +/// (the harness's test tree realizes the assembly pattern). +/// +/// **The serving paths, in consult order** (see the module doc's +/// `blobs/fetch` section for the full contract): +/// +/// - **store** (the whole form): the facade's `get` opened into the +/// plan at establishment; the pump streams the whole in bounded +/// read windows (`FETCH_WINDOW`). The ack carries +/// `{source: "store", len}`. +/// - **fanout** (the whole form over a live broadcast): a tapped +/// subscriber arm rides the plan; the pump forwards the feed and the +/// broadcast's end closes the stream. A fetch never produces a +/// broadcast — the fanout's consumer side only (the module doc's +/// named design point against POC #3 finding A3). The ack carries +/// `{source: "fanout"}` — no length, a live put's tail; the +/// consumer's ADR-006 hash-check runs at stream end and a failed +/// check re-fetches (the late-joiner degrade). +/// - **store** (the ranged form): `read_range` per range at +/// establishment — the ack's `slice_digests` must ride the reply, +/// not the pump — the pump streams the slices in request order. A +/// not-yet-committed digest's ranged fetch is the miss (ranges are +/// the committed entry's shapes; no tap-window feed serves them). +/// +/// **The miss** (absent digest, no live broadcast — the need +/// announcement; the have/need diff is caller-side, no need op exists): +/// the establishment refusal — `channel:open_failed` with reason +/// `unknown_resource` (ADR-049's "the requested resource does not +/// exist"), the digest in the refusal's message; the declared +/// [`BLOB_MISSING`] error definition discloses the missing-error +/// contract. A store-core failure mid-establishment refuses with +/// reason `handler_error` — never serve unverifiable bytes +/// (a verification-failed read refuses too). +/// +/// **Verified streaming is consumer-side** (ADR-006 decision 4): the +/// server streams verified-store bytes; the *consumer* hash-checks the +/// received whole against the digest it carried — the ack's `len` is +/// the expected whole's length where the store path served it. +pub fn fetch_openables(ctx: Arc) -> (OperationSpec, OpenEstablisher, OpenHandler) { + let spec = OperationSpec::new( + FETCH_OP_ID, + OperationType::Sub, + Visibility::External, + json!({ + "type": "object", + "properties": { + "namespace": namespace_schema(), + "digest": digest_wire_schema(), + "ranges": { + "type": "array", + "items": { + "type": "object", + "properties": { + "offset": { "type": "integer", "minimum": 0 }, + "len": { "type": "integer", "minimum": 0 } + }, + "required": ["offset", "len"], + "additionalProperties": false + }, + "minItems": 1 + } + }, + "required": ["namespace", "digest"], + "additionalProperties": false + }), + json!({ + "type": "object", + "properties": { + "channel_id": { "type": "integer", "minimum": 1 }, + "source": { "type": "string", "enum": ["store", "fanout"] }, + "len": { "type": "integer", "minimum": 0 }, + "slice_digests": { + "type": "array", + "items": { "type": "string", "pattern": "^[0-9a-f]{64}$" } + } + }, + "required": ["channel_id"], + "additionalProperties": false + }), + vec![declared_error( + BLOB_MISSING, + "the digest is absent from its namespace's pool (no committed entry and no \ + live broadcast): the establishment refusal answers channel:open_failed with \ + reason unknown_resource, the digest in the refusal's message", + digest_details_schema(), + )], + AccessControl { + resource_type: Some("blob-namespace".to_owned()), + resource_action: Some("read".to_owned()), + ..AccessControl::default() + }, + Some("$.namespace".to_owned()), + ) + .with_channel_open(ChannelOpenSpec::new(BLOBS_CHANNEL_ALPN)) + .with_description( + "the verified fetch: {namespace, digest, ranges?} in; verified bytes out over \ + the alk/blobs channel (consumer-side whole-blob hash-check; the ranged form \ + acks the ADR-006 slice digests)", + ); + let establisher: OpenEstablisher = { + Arc::new(move |input, _auth| { + let ctx = Arc::clone(&ctx); + Box::pin(async move { fetch_establish(&ctx, input).await }) + }) + }; + (spec, establisher, fetch_pump()) +} + +/// The establisher's body — the await-then-answer establishment phase: +/// the ingest (post-schema, the wire gate precedes establishment), +/// the ranged or whole consult, the ack. +async fn fetch_establish( + ctx: &OpsContext, + input: Value, +) -> Result { + let digest_hex = input + .get("digest") + .and_then(Value::as_str) + .unwrap_or_default() + .to_owned(); + let key = match digest_ingress(&digest_hex) { + Ok(key) => key, + Err(_) => { + return Err(EstablishmentError::HandlerError { + message: "the digest ingress rejected after the schema gate (unreachable \ + post-validation)" + .to_owned(), + }) + } + }; + let ranges = match fetch_ranges_ingress(&input) { + Ok(ranges) => ranges, + Err(message) => { + return Err(EstablishmentError::HandlerError { message }); + } + }; + + if let Some(ranges) = ranges { + return fetch_establish_ranges(ctx, &digest_hex, &key, ranges).await; + } + + match ctx.store().get(&key.as_bytes()).await { + Ok(Some((entry_len, blob))) => { + let plan: ChannelPlan = Arc::new(FetchPlan::Whole { blob }); + Ok(Establishment::new(plan) + .with_reply_field("source", json!("store")) + .with_reply_field("len", json!(entry_len))) + } + Ok(None) => { + if let Some(receiver) = ctx.fanout().subscribe(&key) { + let plan: ChannelPlan = Arc::new(FetchPlan::Fanout { receiver }); + Ok(Establishment::new(plan).with_reply_field("source", json!("fanout"))) + } else { + Err(fetch_miss(&digest_hex)) + } + } + Err(error) => Err(fetch_establish_failure("the verified read", error)), + } +} + +/// The ranged form's establishment: `read_range` per range — the +/// slices ride the plan, the slice digests the ack (one per range, in +/// request order). A both-tier miss is the fetch miss; any other store +/// failure refuses establishment. +async fn fetch_establish_ranges( + ctx: &OpsContext, + digest_hex: &str, + key: &Key, + ranges: Vec, +) -> Result { + let mut slices = Vec::with_capacity(ranges.len()); + let mut slice_digest_hexes = Vec::with_capacity(ranges.len()); + for range in &ranges { + match ctx + .store() + .read_range(&key.as_bytes(), range.offset, range.len) + .await + { + Ok((slice, digest)) => { + slices.push(slice); + slice_digest_hexes.push(json!(hex::encode(digest))); + } + Err(Error::Missing) => return Err(fetch_miss(digest_hex)), + Err(error) => { + return Err(fetch_establish_failure("the ranged read", error)); + } + } + } + slices.shrink_to_fit(); + let plan: ChannelPlan = Arc::new(FetchPlan::Ranges { slices }); + Ok(Establishment::new(plan) + .with_reply_field("source", json!("store")) + .with_reply_field("slice_digests", Value::Array(slice_digest_hexes))) +} + +/// The post-schema ranges ingress: `{offset, len}` items (the wire +/// gate's schema precedes; a rejection here is unreachable from a +/// validated wire call). +fn fetch_ranges_ingress(input: &Value) -> Result>, String> { + let Some(values) = input.get("ranges") else { + return Ok(None); + }; + let items = values.as_array().ok_or_else(|| { + format!("the ranges field failed its ingress: expected an array, got {values}") + })?; + let mut ranges = Vec::with_capacity(items.len()); + for value in items { + match ( + value.get("offset").and_then(Value::as_u64), + value.get("len").and_then(Value::as_u64), + ) { + (Some(offset), Some(len)) => ranges.push(FetchRange { offset, len }), + _ => { + return Err(format!("the range item failed its ingress: {value}")); + } + } + } + Ok(Some(ranges)) +} + +/// The fetch miss (the need announcement): ADR-049's typed vocabulary +/// for "the requested resource does not exist", the digest carried. +fn fetch_miss(digest_hex: &str) -> EstablishmentError { + EstablishmentError::UnknownResource { + message: format!( + "the digest {digest_hex} is absent from its namespace's pool (no committed \ + entry and no live broadcast)" + ), + } +} + +/// A store-core failure mid-establishment: refuse, never serve +/// unverifiable bytes (a verification-failed read lands here too). +fn fetch_establish_failure(stage: &str, error: Error) -> EstablishmentError { + EstablishmentError::HandlerError { + message: format!("the store core failed at {stage}: {error}"), + } +} + +/// The pump handler: accept the channel's yield-once stream, downcast +/// the plan, and serve the chosen path to completion — awaited inline +/// by the spawned task (R-02: the returned `JoinHandle` tracks the +/// data-plane lifetime; an early return is teardown-at-birth). +fn fetch_pump() -> OpenHandler { + Arc::new(move |_input, plan, channel_conn, _auth| { + tokio::spawn(async move { + let Some(bidi) = channel_conn.accept_bi().await.ok() else { + return; + }; + let Some(plan) = plan.and_then(|p| p.downcast::().ok()) else { + return; + }; + let plan = match Arc::try_unwrap(plan) { + Ok(plan) => plan, + Err(_) => return, + }; + match plan { + FetchPlan::Whole { blob } => fetch_pump_whole(bidi, blob).await, + FetchPlan::Ranges { slices } => fetch_pump_slices(bidi, slices).await, + FetchPlan::Fanout { receiver } => fetch_pump_fanout(bidi, receiver).await, + } + }) + }) +} + +/// The store path's whole form: the entry streamed as bounded windows +/// (pread-style short reads advance past clamps; a `Blob` stays valid +/// across concurrent pool activity — committed entries are immutable). +/// A store or transport failure ends the pump mid-stream; the +/// consumer's hash-check fails and the re-fetch is the degrade. +async fn fetch_pump_whole(mut stream: BiStream, blob: Blob) { + let len = blob.len(); + let mut pos = 0u64; + while pos < len { + let window = FETCH_WINDOW.min(len - pos); + match blob.read_at(pos, window).await { + Ok(bytes) => { + if stream.write_all(&bytes).await.is_err() { + return; + } + pos += bytes.len() as u64; + } + Err(_) => return, + } + } +} + +/// The ranged form: the establishment-phase slices, streamed in +/// request order — the ack's slice digests are their transport- +/// integrity check (ADR-006 decision 2, out-of-band). +async fn fetch_pump_slices(mut stream: BiStream, slices: Vec>) { + for slice in &slices { + if stream.write_all(slice).await.is_err() { + return; + } + } +} + +/// The fanout path: the tapped arm's feed, forwarded chunk-for-chunk; +/// the broadcast's end (`None`) closes the stream either way the put +/// ended (the two ends are indistinguishable by design — the degrade +/// is the consumer's verified re-fetch, identical bytes under CAS). +async fn fetch_pump_fanout(mut stream: BiStream, mut receiver: FanoutReceiver) { + while let Some(chunk) = receiver.recv().await { + if stream.write_all(&chunk).await.is_err() { + return; + } + } +} + #[cfg(test)] mod tests { use super::*; @@ -1815,6 +2273,7 @@ mod harness { ChannelCore, Establishment, EstablishmentError, OpenEstablisher, OpenHandler, }; use alkcall::channels::policy::NoCap; + use alkcall::channels::reassembly::{MpscRecvStream, MpscSendStream}; use alkcall::client::CallClient; use alkcall::core::auth::{AuthContext, AuthToken, Identity, IdentityProvider}; use alkcall::core::ownership::OwnershipProvider; @@ -2632,6 +3091,364 @@ mod harness { ); } + // ---------------------------------------------------------------- + // blobs/fetch (the binary data plane's consumer side): the verified + // Sub op through the channel halves. + // ---------------------------------------------------------------- + + #[cfg(any(feature = "mem", feature = "sqlite"))] + fn fetch_openable(ctx: Arc) -> ChannelOpenable { + let (spec, establisher, open_handler) = fetch_openables(ctx); + ChannelOpenable { + spec, + establisher, + open_handler, + } + } + + #[cfg(any(feature = "mem", feature = "sqlite"))] + async fn open_fetch( + client: &ChannelClient, + namespace: &str, + digest_hex: &str, + ranges: Option, + ) -> Result< + (u32, Value, MpscSendStream, MpscRecvStream), + alkcall::channels::client::ChannelOpenError, + > { + let mut input = json!({ "namespace": namespace, "digest": digest_hex }); + if let Some(ranges) = ranges { + input["ranges"] = ranges; + } + tokio::time::timeout( + TEST_TIMEOUT, + client.open_channel_with_reply(FETCH_OP_ID, input, BLOBS_CHANNEL_ALPN), + ) + .await + .expect("the open resolves") + } + + /// The whole-blob verified round-trip (the op's core criterion): + /// the seeded whole streams byte-exact over the channel and the + /// consumer-side hash-check passes against the carried digest; the + /// ack carries the store path's `{source, len}` shape. + #[cfg(any(feature = "mem", feature = "sqlite"))] + #[tokio::test] + async fn fetch_serves_the_verified_whole_and_the_consumer_hash_checks() { + let store = test_store().await; + let ctx = Arc::new(OpsContext::new(Arc::clone(&store), OpsConfig::default())); + let harness = OpsHarness::assemble(vec![]); + harness.with_token("token:alice", identity_named("alice")); + harness.grant("alice", "ns-1", &["read"]); + let client = harness + .spawn_channel_halves( + vec![fetch_openable(Arc::clone(&ctx))], + Some(identity_named("alice")), + ) + .await; + + let seed: &[u8] = b"the fetched whole blob"; + let (key, _pin) = store + .put(Some(seed.len() as u64), std::io::Cursor::new(seed)) + .await + .expect("seed put lands"); + let digest_hex = hex::encode(key.digest().bytes()); + + let (channel_id, reply, send, mut recv) = open_fetch(&client, "ns-1", &digest_hex, None) + .await + .expect("the fetch channel establishes"); + assert!(channel_id > 0, "a data channel, not channel 0 itself"); + assert_eq!( + reply, + json!({ "channel_id": channel_id, "source": "store", "len": seed.len() }), + "the store path's ack: source + the expected whole's length" + ); + + let mut received = Vec::new(); + tokio::time::timeout(TEST_TIMEOUT, recv.read_to_end(&mut received)) + .await + .expect("the stream reads") + .expect("the stream ends"); + assert_eq!(received, seed, "byte-exact"); + assert_eq!( + Key::from(Digest::derive(&received, received.len() as u64)), + key, + "the consumer's whole-blob hash-check passes" + ); + drop(send); + } + + /// The ranged form (criterion: the ADR-006 slice-digest convention, + /// out-of-band, over exactly the returned bytes): the ack carries + /// one slice digest per range in request order — read_range's + /// convention verbatim — and the served concatenation is the + /// ranges' slices; an over-read range clamps (its digest covers + /// exactly the clamped bytes). + #[cfg(any(feature = "mem", feature = "sqlite"))] + #[tokio::test] + async fn fetch_serves_ranges_and_the_ack_carries_the_slice_digests() { + let store = test_store().await; + let ctx = Arc::new(OpsContext::new(Arc::clone(&store), OpsConfig::default())); + let harness = OpsHarness::assemble(vec![]); + harness.with_token("token:alice", identity_named("alice")); + harness.grant("alice", "ns-1", &["read"]); + let client = harness + .spawn_channel_halves( + vec![fetch_openable(Arc::clone(&ctx))], + Some(identity_named("alice")), + ) + .await; + + let seed: Vec = (0u8..64).collect(); + let (key, _pin) = store + .put(Some(seed.len() as u64), std::io::Cursor::new(seed)) + .await + .expect("seed put lands"); + let digest_hex = hex::encode(key.digest().bytes()); + + let (slice0, digest0) = store + .read_range(&key.as_bytes(), 2, 5) + .await + .expect("the store's ranged read"); + let (_, digest1) = store + .read_range(&key.as_bytes(), 1000, 9999) + .await + .expect("the store's past-EOF ranged read"); + + let (channel_id, reply, send, mut recv) = open_fetch( + &client, + "ns-1", + &digest_hex, + Some(json!([{ "offset": 2, "len": 5 }, { "offset": 1000, "len": 9999 }])), + ) + .await + .expect("the fetch channel establishes"); + assert_eq!( + reply, + json!({ + "channel_id": channel_id, + "source": "store", + "slice_digests": [hex::encode(digest0), hex::encode(digest1)] + }), + "the ack carries read_range's slice digests, one per range, in request order" + ); + + let mut received = Vec::new(); + tokio::time::timeout(TEST_TIMEOUT, recv.read_to_end(&mut received)) + .await + .expect("the stream reads") + .expect("the stream ends"); + let mut expected = slice0; + expected.extend( + store + .read_range(&key.as_bytes(), 1000, 9999) + .await + .expect("re-read") + .0, + ); + assert_eq!(received, expected, "the concatenation of the served slices"); + drop(send); + } + + /// The fanout seam's consumer side, live-broadcast arm (criterion: + /// a live broadcast tapped — the subscriber receives while the + /// producer runs): the entry never committed, so only the tap could + /// have served; the ack names the path (`source: "fanout"`). + #[cfg(any(feature = "mem", feature = "sqlite"))] + #[tokio::test] + async fn fetch_taps_a_live_broadcast_and_serves_while_the_producer_runs() { + let store = test_store().await; + let ctx = Arc::new(OpsContext::new(Arc::clone(&store), OpsConfig::default())); + let harness = OpsHarness::assemble(vec![]); + harness.with_token("token:alice", identity_named("alice")); + harness.grant("alice", "ns-1", &["read"]); + let client = harness + .spawn_channel_halves( + vec![fetch_openable(Arc::clone(&ctx))], + Some(identity_named("alice")), + ) + .await; + + let seed: Vec = (0u8..16).collect(); + let digest = Digest::derive(&seed, seed.len() as u64); + let key = Key::from(digest.clone()); + let digest_hex = hex::encode(digest.bytes()); + let publisher = ctx + .fanout() + .begin(&key, SlowSubscriberPolicy::Stall) + .expect("the broadcast begins"); + + let (channel_id, reply, send, mut recv) = open_fetch(&client, "ns-1", &digest_hex, None) + .await + .expect("the fetch channel establishes"); + assert_eq!( + reply, + json!({ "channel_id": channel_id, "source": "fanout" }), + "the tapped arm's ack: no length (a live put's tail), the path named" + ); + assert!( + !store.has(&key.as_bytes()).await.unwrap(), + "no commit landed: only the tap could have served" + ); + + publisher.broadcast(Arc::new(seed[..8].to_vec())).await; + let mut first = vec![0u8; 8]; + tokio::time::timeout(TEST_TIMEOUT, recv.read_exact(&mut first)) + .await + .expect("the chunk reads") + .expect("the chunk reads"); + publisher.broadcast(Arc::new(seed[8..].to_vec())).await; + let mut second = vec![0u8; 8]; + tokio::time::timeout(TEST_TIMEOUT, recv.read_exact(&mut second)) + .await + .expect("the chunk reads") + .expect("the chunk reads"); + assert_eq!([first, second].concat(), seed, "the tapped feed, in order"); + + publisher.finish(); + let mut rest = Vec::new(); + tokio::time::timeout(TEST_TIMEOUT, recv.read_to_end(&mut rest)) + .await + .expect("the stream drains") + .expect("the stream ends"); + assert!(rest.is_empty(), "the broadcast end closes the stream"); + drop(send); + } + + /// The late joiner (criterion: a broadcast that ended degrades to + /// the post-commit verified `get` — asserted by which path served): + /// the broadcast begun and ended, the entry committed, the fetch + /// answers from the store path and byte-exact. + #[cfg(any(feature = "mem", feature = "sqlite"))] + #[tokio::test] + async fn fetch_late_joiner_after_the_broadcast_end_serves_the_store_path() { + let store = test_store().await; + let ctx = Arc::new(OpsContext::new(Arc::clone(&store), OpsConfig::default())); + let harness = OpsHarness::assemble(vec![]); + harness.with_token("token:alice", identity_named("alice")); + harness.grant("alice", "ns-1", &["read"]); + let client = harness + .spawn_channel_halves( + vec![fetch_openable(Arc::clone(&ctx))], + Some(identity_named("alice")), + ) + .await; + + let seed: &[u8] = b"the late joiner's bytes"; + let (key, _pin) = store + .put(Some(seed.len() as u64), std::io::Cursor::new(seed)) + .await + .expect("seed put lands"); + let digest_key = key.clone(); + + let publisher = ctx + .fanout() + .begin(&digest_key, SlowSubscriberPolicy::Drop) + .expect("the broadcast begins"); + publisher.broadcast(Arc::new(b"stale feed".to_vec())).await; + publisher.finish(); + + let digest_hex = hex::encode(key.digest().bytes()); + let (channel_id, reply, send, mut recv) = open_fetch(&client, "ns-1", &digest_hex, None) + .await + .expect("the fetch channel establishes"); + assert_eq!( + reply, + json!({ "channel_id": channel_id, "source": "store", "len": seed.len() }), + "the ended broadcast's late joiner is the store path" + ); + + let mut received = Vec::new(); + tokio::time::timeout(TEST_TIMEOUT, recv.read_to_end(&mut received)) + .await + .expect("the stream reads") + .expect("the stream ends"); + assert_eq!(received, seed, "identical bytes under CAS"); + drop(send); + } + + /// The miss (criterion: the typed missing-error schema; the have/ + /// need diff stays caller-side): an absent digest with no live + /// broadcast refuses establishment — `channel:open_failed` with + /// reason `unknown_resource`, the digest in the refusal's message, + /// the declared [`BLOB_MISSING`] contract on the spec. + #[cfg(any(feature = "mem", feature = "sqlite"))] + #[tokio::test] + async fn fetch_miss_refuses_establishment_with_the_typed_missing_shape() { + let store = test_store().await; + let ctx = Arc::new(OpsContext::new(Arc::clone(&store), OpsConfig::default())); + let (spec, _establisher, _pump) = fetch_openables(Arc::clone(&ctx)); + assert!( + spec.error_schemas + .iter() + .any(|error| error.code == BLOB_MISSING), + "the declared missing-error contract is disclosed on the spec" + ); + let harness = OpsHarness::assemble(vec![]); + harness.with_token("token:alice", identity_named("alice")); + harness.grant("alice", "ns-1", &["read"]); + let client = harness + .spawn_channel_halves( + vec![fetch_openable(Arc::clone(&ctx))], + Some(identity_named("alice")), + ) + .await; + + let absent = hex::encode(Digest::derive(b"never landed", 11).bytes()); + let refused = open_fetch(&client, "ns-1", &absent, None).await; + match refused { + Err(error) => { + let call_error = error.call_error().expect("the wire error"); + assert_eq!(call_error.code, "channel:open_failed"); + assert_eq!( + error.establishment_reason(), + Some("unknown_resource"), + "ADR-049's typed vocabulary for the absent resource" + ); + assert!( + call_error.message.contains(&absent), + "the refusal carries the digest: {}", + call_error.message + ); + } + Ok(_) => panic!("the absent digest refused, got a channel"), + } + } + + /// The read gate (criterion: closed-by-default, no read grant on + /// the namespace → denied): the ACL verdict is alkcall's, evaluated + /// at op entry ahead of the establishment phase. + #[cfg(any(feature = "mem", feature = "sqlite"))] + #[tokio::test] + async fn fetch_denies_the_ungranted_namespace_closed_by_default() { + let store = test_store().await; + let ctx = Arc::new(OpsContext::new(Arc::clone(&store), OpsConfig::default())); + let harness = OpsHarness::assemble(vec![]); + harness.with_token("token:stranger", identity_named("stranger")); + let client = harness + .spawn_channel_halves( + vec![fetch_openable(Arc::clone(&ctx))], + Some(identity_named("stranger")), + ) + .await; + + let (digest_hex, _) = seeded_entry(&store, b"denied fetch").await; + let denied = tokio::time::timeout( + TEST_TIMEOUT, + client.call_open_op( + FETCH_OP_ID, + json!({ "namespace": "ns-unlisted", "digest": digest_hex }), + ), + ) + .await + .expect("the denied open resolves"); + assert_eq!( + error_code(&denied), + Some("FORBIDDEN"), + "closed-by-default: a grant per namespace is the deferred cost" + ); + } + // ---------------------------------------------------------------- // The JSON control plane's spec/handler pairs (ops-json-ops): the // stat/have/delete families through the assembled registry. diff --git a/tasks/ops-fetch.md b/tasks/ops-fetch.md index a024fd7..3521a69 100644 --- a/tasks/ops-fetch.md +++ b/tasks/ops-fetch.md @@ -1,7 +1,7 @@ --- id: ops-fetch name: blobs/fetch — the verified Sub op and the fanout seam's consumer side -status: pending +status: completed depends_on: [ops-foundation, ops-call-harness] scope: moderate risk: medium @@ -94,8 +94,101 @@ out** over the binary data channel. ## Notes -> To be filled by implementation agent +Implementation agent, 2026-10-13. The op landed as `fetch_openables(ctx) +-> (OperationSpec, OpenEstablisher, OpenHandler)` in `src/ops.rs` — +the `Sub` open triple the embedder assembles through +`ChannelCore::register_openable_with_establisher` (spec +`channels/blobs/sub`, `OperationType::Sub`, `channel_open` marker on +the `alk/blobs` ALPN `BLOBS_CHANNEL_ALPN`; the standard shape, so the +ALPN derives from the name and no explicit `channel_open_alpn` rides). +Exports: `fetch_openables`, `BLOBS_CHANNEL_ALPN`, `FETCH_OP_ID`. + +Design points decided and recorded (op doc + the module doc's fetch +section + ops-surface.md's fetch bullet): + +- **The ack rides ADR-049 amendment 3's establishment reply fields**: + the open reply merges `{channel_id, source, len?, slice_digests?}`. + `source` names the serving path (`store`/`fanout` — the test + criterion "asserted by which path served" is wire-visible, and a + consumer knows whether the stream is a live put's tail); the whole + form acks `{source: "store", len}` or `{source: "fanout"}` (no + length on a tapped broadcast — the hash-check runs at stream end); + the ranged form acks `{source: "store", slice_digests[]}` — + read_range's convention verbatim, one per range in request order, + over exactly the served bytes (over-reads clamp; the digest covers + the clamp). +- **The establisher's consult order**: store `get` first (a commit + that raced the broadcast's end is the same bytes under CAS; the + Blob opens into the plan at establishment, closing the + establish→pump race window), then the fanout subscribe (a live + broadcast taps), then the miss. The ranged form serves + `read_range` on the committed entry only — a not-yet-committed + digest's ranged fetch is the miss (ranges are the committed entry's + shapes; no tap-window feed serves them). The ranged form's reads + happen at establishment because the slice digests must ride the + ack, which lands before the pump runs. +- **The miss maps onto ADR-049's established vocabulary** — the only + op-plane failure surface an open op has: the establishment phase + refuses with reason `unknown_resource` ("the requested resource + does not exist"), the digest in the refusal's message; the + declared `BLOB_MISSING` ErrorDefinition (same code + digest-details + schema the JSON ops declare) is the op's disclosed missing-error + contract. `channel:open_failed` + `details.reason` is what a + consumer branches on; a phantom channel with an in-band error frame + (the ADR-049 antipattern) is never reachable. The same closure + covers the refusal-to-serve-unverifiable shape: a store-core error + mid-establishment (including a verification-failed read) refuses + with reason `handler_error` rather than streaming unverified bytes. +- **The producer-side design point (decided, against the A3 text)**: + fetches do NOT produce read-side broadcasts — the put handler's + stream reader is the canonical producer, concurrent fetchers of one + stored digest each ride their own facade read. Recorded in the + module doc's fanout section (a named design point paragraph), the + op doc, and ops-surface.md's fetch bullet. Rationale: A3's fanout + is the put stream's seam (its late-joiner note degrades to get, + never get-then-rebroadcast); a read-side broadcast adds a second + producer class with no lifecycle owner (its end would be an + arbitrary policy, its arms hold memory where the large tier exists + to avoid materializing), for a concurrency case bounded by the put + window the put-side broadcast already covers; under CAS every + per-fetcher read returns identical bytes. +- **Slow-subscriber policy stays producer-owned** (per-broadcast, set + by the put at `begin` — the foundation's registry semantics); a + fetch tap rides the broadcast's policy. The tap-time-set discipline + (a mid-put tap has no prefix) is documented on the op + registry: + the consumer's hash-check on a partial feed fails and the re-fetch + is the degrade, identical bytes post-commit. +- **A mid-stream note for the whole form**: between a store-hit + establishment and the pump, the entry cannot vanish (the Blob read + is already open; committed entries are immutable), so the ack's + `len` is trustworthy; a mid-stream store/transport failure ends the + pump (teardown closes the stream), the consumer's hash-check fails, + and the re-fetch is the degrade. + +Verification: `cargo test` (default) green, `cargo test --features ops` +green (the five harness tests: verified whole round-trip + consumer +hash-check; ranged + ack slice digests vs `read_range`; live-broadcast +tap served without a commit; late joiner served by the store path; the +typed miss; the closed-by-default read denial), `--all-features` green, +clippy `-D warnings` on all three shapes, `cargo fmt --check` green, +`cargo doc --no-deps` clean on both shapes (the private +`FETCH_WINDOW` doc link de-linked, the PGLO-1 class). The pure +`--no-default-features --features ops` build fails on main already +(engine-typed match arms upstream of this task, unreachable without an +engine feature) — pre-existing, not this task's regression. ## Summary -> To be filled on completion \ No newline at end of file +`blobs/fetch` landed complete: the verified `Sub` op as the +`fetch_openables` open triple (`channels/blobs/sub` + the +`channel_open` marker on `alk/blobs`; read-gated, closed-by-default), +the establishment ack in ADR-049's reply fields +(`{source, len?, slice_digests?}`), whole-blob consumer-side +verification per ADR-006, the ranged form carrying `read_range`'s +slice-digest convention, the fanout seam's consumer side (live tap / +late-joiner store degrade, the path named in the ack), the miss as the +typed missing-error refusal (BLOB_MISSING disclosed; the wire shape is +ADR-049's `unknown_resource`), and the producer-side design point +decided and recorded against the A3 text (fetches never produce +read-side broadcasts). All acceptance criteria verified; harness tests +green; all gates green on both feature shapes. \ No newline at end of file