alkblobs 0.2.2

//! The fetch fanout (the registry, the publisher, the subscriber
//! arms, the pumps) and the `blobs/fetch` op (`ops/mod.rs`'s layout
//! note): the registry mechanics and the fanout's consumer side ride
//! one module; the producer side is the put (`put.rs`).

use std::collections::BTreeMap;
use std::sync::Arc;

use alkcall::channels::operations::{
    ChannelPlan, Establishment, EstablishmentError, OpenEstablisher, OpenHandler,
};
use alkcall::core::types::BiStream;
use alkcall::registry::spec::{
    AccessControl, ChannelOpenSpec, OperationSpec, OperationType, Visibility,
};
use parking_lot::Mutex;
use serde_json::{json, Value};
use tokio::io::AsyncWriteExt;
use tokio::sync::mpsc;

use super::tokens::OpsContext;
use super::wire::{
    declared_error, digest_details_schema, digest_ingress, digest_wire_schema, namespace_schema,
    BLOBS_CHANNEL_ALPN, BLOB_MISSING, FETCH_OP_ID,
};
use crate::facade::Blob;
use crate::keys::Key;
use crate::store::Store;
use crate::Error;

/// 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<Vec<u8>>;

/// 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<FanoutChunk>;

/// 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. The read-peg
    /// residual — a tapper that stops reading pegs the broadcast while
    /// its tap lives — is accepted-uncovered (review-001 S-7; the
    /// bounds named on the registry doc).
    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<mpsc::Sender<FanoutChunk>>,
}

/// 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<BTreeMap<Key, LiveBroadcast>>,
    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`]).
///
/// **Accepted-uncovered availability residuals (review-001 S-7)** —
/// the fanout's two griefing residuals are real, bounded, and
/// recorded here as decisions, not oversights:
///
/// - **Broadcast squatting under the unverified offered claim**: the
///   broadcast begins before any byte is read (the put pipe's
///   producer half — `put.rs`), keyed under the *offered* key
///   while the derivation under the key's own algorithm is still
///   unverified against the stream, so a hostile putter can squat a
///   key's broadcast and feed garbage to the concurrent fetchers'
///   taps. **Integrity
///   is intact** — the consumer-side ADR-006 hash-check is the
///   barrier, garbage never verifies, and a failed check degrades to
///   the verified re-fetch; the residual is the wasted re-fetches
///   (an availability cost, accepted).
/// - **The `Stall` policy's read-peg**: a read-granted fetcher can
///   tap a live put under `Stall` and simply stop reading — the
///   bounded arm channel backpressures the broadcast, and with it the
///   producer's streaming; no deadline exists anywhere in the put
///   path. The peg is real while it lasts and bounded: alkcall's
///   channel lifecycle (a disconnect/teardown closes the tap and the
///   broadcast's send unblocks — the dead arm is removed), the put's
///   own completion (the entry drops, arms see channel end), and the
///   subscriber bound capping who can tap at all.
///
/// 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<FanoutLive>,
}

/// 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<FanoutLive>,
    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<mpsc::Sender<FanoutChunk>> = 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<FanoutPublisher> {
        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<FanoutReceiver> {
        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 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.
pub struct FetchRange {
    /// The window's start byte.
    pub offset: u64,
    /// The window's byte count (over-reads clamp at the entry's end).
    pub len: u64,
}

/// The ranged form's bounds, from the context's
/// [`OpsConfig`](super::tokens::OpsConfig): the range-count bound
/// lands in the wire schema as `maxItems` (an over-count open is the
/// registry's `INVALID_INPUT`); the total-bytes bound is checked at
/// the establisher's ranges ingress — JSON Schema cannot sum, so the
/// over-bytes request is the establishment `HandlerError` refusal.
#[derive(Debug, Clone, Copy)]
pub struct RangedBounds {
    /// The range-count bound (enforced at the schema gate as `maxItems`).
    pub max_ranges: usize,
    /// The total-bytes bound (this ingress' own duty: the summed lens
    /// over every range in one request).
    pub max_total_bytes: 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 entry's key plus the requested ranges —
    /// **not** the slices. The pump streams per range via
    /// `read_range`; establishment already read per range only to
    /// compute the ack digests (bytes transient, one range resident at
    /// a time, only the digests retained).
    ///
    /// **The named tradeoff (the double-read)**: the ranged form reads
    /// the entry twice — the establishment digest pass and the pump's
    /// stream pass. Accepted: memory-bounded beats read-count on a
    /// network-facing path (the pre-materializing shape was a single
    /// request's ~TB-size residency); the alternative — digests
    /// computed at stream-end — changes the ack contract, which lands
    /// before the pump runs.
    Ranges { key: Key, ranges: Vec<FetchRange> },
    /// 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): the establishment phase reads per
///   range for the ack's `slice_digests` (bytes transient — the pump
///   re-reads), and the pump **streams per range** via `read_range`
///   (the double-read tradeoff, named on the plan doc). The form is
///   bounded: the ranges schema's `maxItems` (the context's
///   `fetch_max_ranges`) and the total-bytes ingress bound (the
///   context's `fetch_max_total_bytes`). 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 ranged_bounds = RangedBounds {
        max_ranges: ctx.config().fetch_max_ranges,
        max_total_bytes: ctx.config().fetch_max_total_bytes,
    };
    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,
                    "maxItems": ranged_bounds.max_ranges
                }
            },
            "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 the 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 pump_store = Arc::clone(ctx.store());
    let establisher: OpenEstablisher = {
        Arc::new(move |input, _auth| {
            let ctx = Arc::clone(&ctx);
            Box::pin(async move { fetch_establish(&ctx, input, &ranged_bounds).await })
        })
    };
    (spec, establisher, fetch_pump(pump_store))
}

/// 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,
    bounded: &RangedBounds,
) -> 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, bounded) {
        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.key_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
/// slice **digests** ride the ack (one per range, in request order);
/// the bytes are transient, one range resident at a time (nothing
/// materializes into the plan — the pump streams per range, the
/// double-read tradeoff named on the [`FetchPlan::Ranges`] doc). 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 slice_digest_hexes = Vec::with_capacity(ranges.len());
    for range in &ranges {
        match ctx
            .store()
            .read_range(&key.key_bytes(), range.offset, range.len)
            .await
        {
            Ok((_slice, digest)) => 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));
            }
        }
    }
    let plan: ChannelPlan = Arc::new(FetchPlan::Ranges {
        key: key.clone(),
        ranges,
    });
    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 and the
/// total-bytes bound. The schema gate precedes establishment (the
/// item shape, the count bound as `maxItems`); the summed lens are
/// **this** ingress' own duty — JSON Schema cannot sum — so an
/// over-bytes request is reachable from a fully validated wire call.
/// Its rejection answers the establishment `HandlerError` refusal
/// with a stable domain text: the bound's shape named, the wire
/// numbers carried, no raw payload `Value` reflected.
pub fn fetch_ranges_ingress(
    input: &Value,
    bounds: &RangedBounds,
) -> Result<Option<Vec<FetchRange>>, String> {
    let Some(values) = input.get("ranges") else {
        return Ok(None);
    };
    let Some(items) = values.as_array() else {
        return Err("the ranges field failed its ingress: expected an array of \
                    {offset, len} items"
            .to_owned());
    };
    let mut ranges = Vec::with_capacity(items.len());
    let mut total_bytes = 0u64;
    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)) => {
                total_bytes = {
                    match total_bytes.checked_add(len) {
                        Some(summed) => summed,
                        None => {
                            return Err(format!(
                                "the ranges request exceeds the total-bytes bound: the \
                                 requested lens sum past the limit of {} bytes",
                                bounds.max_total_bytes
                            ))
                        }
                    }
                };
                ranges.push(FetchRange { offset, len });
            }
            _ => {
                return Err(
                    "a range item failed its ingress: offset and len are required, \
                            non-negative integers"
                        .to_owned(),
                )
            }
        }
    }
    if total_bytes > bounds.max_total_bytes {
        return Err(format!(
            "the ranges request exceeds the total-bytes bound: {} bytes requested, the \
             limit is {} bytes",
            total_bytes, bounds.max_total_bytes
        ));
    }
    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 the 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).
/// The refusal is the stable domain text (review-001 S-4: the store
/// core's `Display` detail — the same leak class the
/// [`super::wire::store_internal_error`] posture covers — is dropped
/// at the synthesis point, never on the wire).
fn fetch_establish_failure(stage: &str, _error: Error) -> EstablishmentError {
    EstablishmentError::HandlerError {
        message: format!("the store core failed at {stage}"),
    }
}

/// 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). The
/// pump holds the store handle itself (the ranged form's per-range
/// streaming reads through it; the plan carries the key and the
/// ranges, not the bytes).
fn fetch_pump(store: Arc<Store>) -> OpenHandler {
    Arc::new(move |_input, plan, channel_conn, _auth| {
        let store = Arc::clone(&store);
        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 { key, ranges } => {
                    fetch_pump_ranges(bidi, &store, &key, &ranges).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: per-range `read_range`, streamed in request order
/// — the plan carries the key and the ranges (never the slices), one
/// range resident at a time, the memory-bound the streaming reshape
/// exists for (the double-read tradeoff rides the
/// [`FetchPlan::Ranges`] doc). **The mid-stream failure posture**: a
/// delete racing the pump — or any store failure — ends the pump
/// mid-stream; the consumer's slice-digest check fails and the
/// re-fetch is the degrade (the same posture the whole form's pump
/// has; committed entries are immutable, so the honest-case bytes are
/// identical).
async fn fetch_pump_ranges(mut stream: BiStream, store: &Store, key: &Key, ranges: &[FetchRange]) {
    for range in ranges {
        match store
            .read_range(&key.key_bytes(), range.offset, range.len)
            .await
        {
            Ok((slice, _digest)) => {
                if stream.write_all(&slice).await.is_err() {
                    return;
                }
            }
            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::*;
    use crate::ops::tests::key_of;

    /// 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"
        );
    }
}