feat(ops): blobs/fetch — the verified Sub op and the fanout seam's consumer side (wave 7)

- the open triple (fetch_openables): the channels/blobs/sub spec with
  the channel_open marker on the alk/blobs ALPN (read action,
  closed-by-default), the awaited establisher whose reply fields carry
  the ack {source, len?, slice_digests?}, and the pump handler over
  the channel's stream (bounded read windows; the fanout feed
  forwarded chunk-for-chunk)
- verified streaming per ADR-006: store-first consult, a live
  broadcast taps (the tap-time-set discipline documented; the
  re-fetch is the partial-feed degrade), the ranged form carries
  read_range's slice-digest convention over exactly the served bytes
- the miss maps onto ADR-049's vocabulary (channel:open_failed,
  reason unknown_resource, the digest in the message; the declared
  BLOB_MISSING contract disclosed on the spec); unverifiable reads
  refuse with handler_error, never stream
- the producer-side design point decided and recorded against the A3
  text: fetches never produce read-side broadcasts (module doc,
  op doc, ops-surface.md)
- five harness tests through the channel halves: the verified whole
  round-trip + consumer hash-check, ranged + ack digests, the
  live-broadcast tap served without a commit, the late joiner served
  by the store path, the typed miss, the closed-by-default denial

Verification: cargo test (default) + --features ops + --all-features
green; clippy -D warnings on all three shapes; fmt --check clean;
cargo doc --no-deps clean on both shapes.
This commit is contained in:
glm-5.3-flash committed 2026-10-13 00:43:52 +00:00
1 parent 5144b6a733
commit d927da2ee5
4 files changed
+954 -24

No files matched your search

+28 -8
View File
@@ -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 +
+4 -4
View File
@@ -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"))]
+826 -9
View File
@@ -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/<alpn>/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/<alpn>/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<OpsContext>) -> (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<Vec<u8>> },
/// 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<OpsContext>) -> (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<Establishment, EstablishmentError> {
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<FetchRange>,
) -> Result<Establishment, EstablishmentError> {
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<Option<Vec<FetchRange>>, 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::<FetchPlan>().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<Vec<u8>>) {
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<OpsContext>) -> 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<Value>,
) -> 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<u8> = (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<u8> = (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.
+96 -3
View File
@@ -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
`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.