didbot_serve/

subscribe.rs

//! `com.atproto.sync.subscribeRepos`: this server's repositories, as atproto
//! wants to hear about them.
//!
//! A WebSocket of framed DAG-CBOR carrying signed commits and the blocks
//! that prove them, which is what a relay crawls.
//!
//! # The frame
//!
//! One WebSocket binary message is two concatenated DAG-CBOR values: a header
//! and a body. The header is `{"op": 1, "t": "#commit"}` for a message, and
//! `{"op": -1}` for an error, whose body is `{"error": .., "message": ..}`.
//! Nothing separates the two values; a reader decodes the header and continues
//! from where that decode stopped.
//!
//! # Sequence numbers and cursors
//!
//! Every message frame carries a sequence number, and a consumer hands back
//! the last one it applied as `?cursor=`. The cursor is a bare integer with
//! nowhere to name the run of the server that minted it, so the number
//! itself survives a restart — see
//! [`didbot_pds::sequence`] for how, and for why a restart skips numbers
//! rather than repeating them.
//!
//! The replay buffer holds at most [`DEFAULT_REPLAY_FRAMES`] frames and
//! [`DEFAULT_REPLAY_BYTES`] of them, evicting the oldest first, unless a
//! deployment sets its own [`ReplayBounds`]. It lives in memory, so a restart
//! empties it even though the numbering survives. A cursor the buffer cannot
//! reach is answered with `#info OutdatedCursor` and
//! then everything still held: the consumer has a hole it must fill by reading the
//! repositories, and being told so is the point. A cursor naming a number
//! this server has not assigned yet is a different thing — it cannot be
//! filled in from anywhere — so it is the `FutureCursor` error the lexicon
//! declares, and the connection closes.
//!
//! A boot numbers frames before anything can connect: the positions that
//! changed while the process was down, and a restore's resets. A producer
//! built with [`Repos::booting`] keeps every one of them until
//! [`Repos::listening`], and its bounds' worth more on top, so a relay that
//! connects after the boot is handed all of them.
//!
//! # One copy of each frame
//!
//! The replay buffer is the only place a frame is held. A consumer keeps its
//! position in the buffer and reads each frame from there when it is ready
//! to send it, the replay and the live stream alike.
//!
//! A frame an open stream has not sent yet is kept past the bounds, up to
//! twice them, so a consumer resuming from the oldest frame held has as much
//! room to fall behind as one following live. Past twice the bounds the
//! oldest frames go anyway, and a consumer that had not sent one is ended
//! with `ConsumerTooSlow`: its next frame is gone. So the firehose's memory
//! is at most twice the bounds, plus the one frame each connection is
//! writing to its socket.
//!
//! # Only what is durable
//!
//! A record write is in the kernel when it returns and on the medium within
//! [`DEFAULT_SYNC_INTERVAL`]. A `#commit` published in between announces a
//! commit a power cut can take back, and a relay that applied it holds a
//! head this server no longer has. So a producer over a write-ahead log holds
//! each commit, beside the position the log had reached when it was handed
//! over, and publishes it once the log has synced through that position. A
//! commit's frame is therefore up to one sync interval later than its write.
//!
//! Every other frame syncs the log and publishes the commits held before it
//! first, so the stream keeps the order the writes made: an erasure's
//! `#account` after the repository's last `#commit`, and a restore's `#sync`
//! after the commit it carries is on the medium.

use std::collections::{HashMap, VecDeque};
use std::sync::{Arc, Condvar, Mutex, Weak};
use std::time::Duration;

use bytes::Bytes;
use didbot_data::{dag_cbor, Cid, Value};
use didbot_pds::journal::Mark;
use didbot_pds::wal::DEFAULT_SYNC_INTERVAL;
use didbot_pds::{RepoAccount, RepoCommit, RepoEventSink, RepoIdentity, RepoSync, Sequence, Wal};
use tokio::sync::watch;
use tracing::{debug, info, warn};

/// How many frames the replay buffer holds, if a deployment does not choose.
///
/// Most frames are a few kilobytes — the commit block, the tree nodes on one
/// path and one record — so at this many the frame bound is the one reached
/// first, and it is a consumer's reconnect window. Each frame held costs a
/// few dozen bytes beyond its encoding, which is what this bound limits when
/// frames are small. See `[capacity] firehose_replay_frames`.
pub const DEFAULT_REPLAY_FRAMES: usize = 4096;

/// How many encoded bytes the replay buffer holds, if a deployment does not
/// choose.
///
/// A commit may carry up to 2,000,000 bytes of blocks, so a frame count
/// alone does not bound memory: 4096 such frames are 8 GB. The buffer holds
/// up to twice this for streams that have not sent their frames, and 128 MiB
/// is a sixteenth of the 2 GiB the default instance has. 64 MiB still holds
/// [`DEFAULT_REPLAY_FRAMES`] frames of up to 16 KiB, or 32 of the largest.
/// See `[capacity] firehose_replay_bytes`.
pub const DEFAULT_REPLAY_BYTES: usize = 64 * 1024 * 1024;

/// How much the replay buffer holds. It evicts its oldest frame when a new
/// one would take it past either bound, unless an open stream has not sent
/// that frame yet. Those it keeps up to twice the bounds. See the module docs.
///
/// The newest frame is always held, even when it alone is larger than
/// `bytes`: consumers read live frames from the buffer, so a frame it did
/// not hold would reach no one.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ReplayBounds {
    /// The most frames. At least 1.
    pub frames: usize,
    /// The most encoded bytes, counting each frame's header and body.
    pub bytes: usize,
}

impl Default for ReplayBounds {
    fn default() -> Self {
        Self {
            frames: DEFAULT_REPLAY_FRAMES,
            bytes: DEFAULT_REPLAY_BYTES,
        }
    }
}

impl ReplayBounds {
    /// `frames` frames, and the default byte bound.
    #[must_use]
    pub fn frames(frames: usize) -> Self {
        Self {
            frames,
            ..Self::default()
        }
    }
}

/// The `op` a message frame's header carries.
const OP_MESSAGE: i64 = 1;

/// The `op` an error frame's header carries.
const OP_ERROR: i64 = -1;

/// Which of the lexicon's message types a frame is.
///
/// Kept beside the encoded bytes so a per-connection filter can drop a frame
/// without decoding it again — see `excludeCommit` and its siblings.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum FrameKind {
    /// `#commit`.
    Commit,
    /// `#identity`.
    Identity,
    /// `#account`.
    Account,
    /// `#sync`.
    Sync,
}

impl FrameKind {
    /// The `t` this kind's header carries.
    #[must_use]
    pub fn tag(&self) -> &'static str {
        match self {
            Self::Commit => "#commit",
            Self::Identity => "#identity",
            Self::Account => "#account",
            Self::Sync => "#sync",
        }
    }
}

/// One numbered frame, already encoded.
///
/// The bytes are built once, when the frame is numbered, and the buffer keeps
/// them: a replay is neither a re-encode nor a copy, and two consumers reading
/// the same frame are guaranteed the same bytes.
///
/// They are [`Bytes`] rather than a `Vec<u8>` because every consumer sending
/// a frame holds it while the socket write is under way. As a `Vec<u8>` each
/// of those is a copy. Cloning `Bytes` bumps a refcount instead, and `axum`'s
/// `Message::Binary` takes one, so the frame reaches the socket without ever
/// being copied.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Frame {
    /// Where this frame sits in the stream.
    pub seq: u64,
    /// Which message type it carries.
    pub kind: FrameKind,
    /// Header and body, concatenated, as they go on the wire.
    pub bytes: Bytes,
}

/// Encodes one message frame: the header, then the body.
///
/// Public because the tests that matter here decode what this produced, and a
/// consumer written against this server is entitled to the same encoder.
#[must_use]
pub fn encode_message(tag: &str, body: &Value) -> Vec<u8> {
    let mut bytes = dag_cbor::encode(&map([
        ("op", Value::Integer(OP_MESSAGE)),
        ("t", Value::String(tag.to_owned())),
    ]));
    bytes.extend(dag_cbor::encode(body));
    bytes
}

/// Encodes one error frame, which closes the connection after it.
#[must_use]
pub fn encode_error(error: &str, message: &str) -> Vec<u8> {
    let mut bytes = dag_cbor::encode(&map([("op", Value::Integer(OP_ERROR))]));
    bytes.extend(dag_cbor::encode(&map([
        ("error", Value::String(error.to_owned())),
        ("message", Value::String(message.to_owned())),
    ])));
    bytes
}

/// The lexicon's `seq`, which is a signed integer where this counter is not.
///
/// A deployment that assigned more than nine quintillion frames has other
/// problems; saturating is still the right answer, because the alternative is
/// a frame that wraps to a number below the one before it and a consumer that
/// concludes the stream has gone backwards.
fn clamp(seq: u64) -> i64 {
    seq.min(i64::MAX as u64) as i64
}

/// A DAG-CBOR map from pairs, so the bodies below read as their lexicons.
fn map<const N: usize>(pairs: [(&str, Value); N]) -> Value {
    Value::Map(
        pairs
            .into_iter()
            .map(|(key, value)| (key.to_owned(), value))
            .collect(),
    )
}

/// A CID-link, or null for the fields the lexicon marks nullable.
///
/// A CID that will not parse is written as null rather than dropped or
/// guessed at. It cannot happen — every one of these came from a repository
/// this process just signed — and if it did, a null in a nullable field is a
/// frame a consumer can still read.
fn link(cid: Option<&str>) -> Value {
    match cid.map(Cid::parse) {
        Some(Ok(cid)) => Value::Link(cid),
        Some(Err(error)) => {
            warn!(%error, "a commit named a CID that will not parse; sending null");
            Value::Null
        }
        None => Value::Null,
    }
}

/// The `#commit` body, as the lexicon defines it.
///
/// `rebase`, `tooBig` and `blobs` are the lexicon's three deprecated fields.
/// They are still required, so they are written at their fixed empty values —
/// false, false and the empty list — rather than omitted.
fn commit_body(seq: u64, event: &RepoCommit, time: &str) -> Value {
    let ops = event
        .ops
        .iter()
        .map(|op| {
            let mut fields = map([
                ("action", Value::String(op.action().to_owned())),
                ("cid", link(op.cid.as_deref())),
                ("path", Value::String(op.path.clone())),
            ]);
            // Optional rather than nullable, and the lexicon says a creation
            // "should not define" it — so it is absent for a create and
            // carries the replaced or removed record's CID otherwise. A
            // consumer inverting a commit reads the old value here and
            // nowhere else.
            if let (Some(prev), Value::Map(fields)) = (&op.prev, &mut fields) {
                fields.insert("prev".to_owned(), link(Some(prev)));
            }
            fields
        })
        .collect();
    let mut body = map([
        ("blobs", Value::List(Vec::new())),
        ("blocks", Value::Bytes(event.blocks.clone())),
        ("commit", link(Some(&event.commit))),
        ("ops", Value::List(ops)),
        ("rebase", Value::Bool(false)),
        ("repo", Value::String(event.did.clone())),
        ("rev", Value::String(event.rev.clone())),
        ("seq", Value::Integer(clamp(seq))),
        (
            "since",
            match &event.since {
                Some(rev) => Value::String(rev.clone()),
                None => Value::Null,
            },
        ),
        ("time", Value::String(time.to_owned())),
        ("tooBig", Value::Bool(false)),
    ]);
    // Optional rather than nullable, so it is absent rather than null when
    // there is no previous commit to name a tree for.
    if let Some(prev) = &event.prev_data {
        if let Value::Map(fields) = &mut body {
            fields.insert("prevData".to_owned(), link(Some(prev)));
        }
    }
    body
}

/// The `#identity` body.
fn identity_body(seq: u64, event: &RepoIdentity, time: &str) -> Value {
    let mut body = map([
        ("did", Value::String(event.did.clone())),
        ("seq", Value::Integer(clamp(seq))),
        ("time", Value::String(time.to_owned())),
    ]);
    // Optional, so absent rather than null for a deployment that issues no
    // handles.
    if let Some(handle) = &event.handle {
        if let Value::Map(fields) = &mut body {
            fields.insert("handle".to_owned(), Value::String(handle.clone()));
        }
    }
    body
}

/// The `#account` body.
fn account_body(seq: u64, event: &RepoAccount, time: &str) -> Value {
    let mut body = map([
        ("active", Value::Bool(event.active)),
        ("did", Value::String(event.did.clone())),
        ("seq", Value::Integer(clamp(seq))),
        ("time", Value::String(time.to_owned())),
    ]);
    // "If active=false, this optional field indicates a reason" — so an
    // active account carries no `status` at all rather than one saying it is
    // fine.
    if let Some(status) = event.status {
        if let Value::Map(fields) = &mut body {
            fields.insert(
                "status".to_owned(),
                Value::String(status.as_str().to_owned()),
            );
        }
    }
    body
}

/// The `#sync` body.
fn sync_body(seq: u64, event: &RepoSync, time: &str) -> Value {
    map([
        ("blocks", Value::Bytes(event.blocks.clone())),
        ("did", Value::String(event.did.clone())),
        ("rev", Value::String(event.rev.clone())),
        ("seq", Value::Integer(clamp(seq))),
        ("time", Value::String(time.to_owned())),
    ])
}

/// The `#info` body.
fn info_body(name: &str, message: &str) -> Value {
    map([
        ("message", Value::String(message.to_owned())),
        ("name", Value::String(name.to_owned())),
    ])
}

/// The one `name` this server ever sends on an `#info` frame.
pub const OUTDATED_CURSOR: &str = "OutdatedCursor";

/// The lexicon's error for a cursor ahead of the stream.
pub const FUTURE_CURSOR: &str = "FutureCursor";

/// Which message types one connection wants.
///
/// The lexicon's `excludeCommit`, `excludeIdentity` and `excludeAccount`, read
/// once at connect time. They are a *delivery* filter and nothing more:
/// sequence numbers are not renumbered per consumer, and an excluded frame
/// still spends its number. That is what lets a consumer that excluded
/// something reconnect with a cursor which skips past the frames it never
/// saw — the numbers it did see are the same numbers everyone else saw.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct Exclusions {
    /// Drop `#commit` frames.
    pub commit: bool,
    /// Drop `#identity` frames.
    pub identity: bool,
    /// Drop `#account` frames.
    pub account: bool,
}

impl Exclusions {
    /// Whether a frame of this kind should reach the consumer.
    ///
    /// No parameter excludes `#sync`: it says a repository was reset, which
    /// a consumer holding that repository cannot do without.
    #[must_use]
    pub fn wants(&self, kind: FrameKind) -> bool {
        match kind {
            FrameKind::Commit => !self.commit,
            FrameKind::Identity => !self.identity,
            FrameKind::Account => !self.account,
            FrameKind::Sync => true,
        }
    }

    /// Whether anything at all is excluded.
    #[must_use]
    pub fn any(&self) -> bool {
        self.commit || self.identity || self.account
    }
}

/// What a connection asked for.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Resume {
    /// From now. No cursor was given.
    Live,
    /// After this sequence number.
    After(u64),
    /// Ahead of anything this server has assigned.
    Future(u64),
}

/// Where one consumer starts reading the replay buffer, decided in
/// [`Repos::start_for`].
struct Start {
    /// The oldest sequence number the buffer held, or 0 when it was empty.
    oldest: u64,
    /// Whether the cursor named a frame the buffer no longer held, which is
    /// what puts an `#info OutdatedCursor` in front of the stream.
    outdated: bool,
    /// The stream sends the frames after this sequence number.
    after: u64,
    /// The stream's key in [`Ring::streams`].
    stream: u64,
}

/// The producer: the sequence and the replay buffer consumers read from.
///
/// One per server. Hand a clone to
/// [`Provisioner::with_repo_sink`](didbot_pds::Provisioner::with_repo_sink)
/// and the same one to [`app_with_repos`](crate::app_with_repos); every
/// clone shares `Inner`.
#[derive(Debug, Clone)]
pub struct Repos {
    inner: Arc<Inner>,
}

#[derive(Debug)]
struct Inner {
    sequence: Arc<Sequence>,
    bounds: ReplayBounds,
    /// The replay buffer, and the lock that makes the stream ordered.
    ///
    /// The sequence number is assigned and the frame buffered while this is
    /// held, so two concurrent writes cannot reach the buffer in the
    /// opposite order to their numbers. Out of order, a consumer would be
    /// handed a frame it had already applied and skip one it had not.
    ring: Mutex<Ring>,
    /// The newest sequence number buffered, which every stream waits on for
    /// its next frame. One receiver per stream.
    published: watch::Sender<u64>,
    /// Commits waiting for the log to sync past them, for a producer over a
    /// log. See the module docs.
    held: Option<Arc<Held>>,
    /// Set once, when every stream ends after what it holds. See
    /// [`Repos::close`].
    closing: watch::Sender<bool>,
}

/// Commits handed over before the log they are written to has synced past
/// them. See the module docs.
#[derive(Debug)]
struct Held {
    log: Arc<Wal>,
    /// Oldest first, each beside the position the log had reached when it
    /// arrived. The position is read under this lock, so the positions never
    /// go down.
    commits: Mutex<VecDeque<(Mark, RepoCommit)>>,
    /// Signalled when a commit arrives.
    arrived: Condvar,
}

impl Held {
    fn over(log: Arc<Wal>) -> Self {
        Self {
            log,
            commits: Mutex::new(VecDeque::new()),
            arrived: Condvar::new(),
        }
    }

    fn lock(&self) -> std::sync::MutexGuard<'_, VecDeque<(Mark, RepoCommit)>> {
        self.commits
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
    }

    /// Holds `event` until the log has synced past where it is now, which is
    /// past every entry the commit wrote.
    fn hold(&self, event: &RepoCommit) {
        let mut commits = self.lock();
        commits.push_back((self.log.mark(), event.clone()));
        self.arrived.notify_one();
    }

    /// The position the oldest held commit waits for, waiting up to
    /// `timeout` for a commit when none is held.
    fn oldest(&self, timeout: Duration) -> Option<Mark> {
        let commits = self.lock();
        let (commits, _) = self
            .arrived
            .wait_timeout_while(commits, timeout, |commits| commits.is_empty())
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        commits.front().map(|(mark, _)| *mark)
    }

    /// The oldest held commit, if the log has synced through `synced` past
    /// it.
    fn take_through(&self, synced: Mark) -> Option<RepoCommit> {
        let mut commits = self.lock();
        if commits.front().is_some_and(|(mark, _)| *mark <= synced) {
            return commits.pop_front().map(|(_, event)| event);
        }
        None
    }
}

/// Publishes held commits as the log syncs past them, on a thread of its
/// own.
///
/// Numbering a frame can reserve sequence numbers, which is an append that
/// syncs the log, so publishing cannot happen inside a sync. The thread holds
/// the producer weakly and ends at the first wake after it is dropped.
fn publish_as_synced(inner: &Arc<Inner>, held: Arc<Held>) -> std::io::Result<()> {
    let producer = Arc::downgrade(inner);
    std::thread::Builder::new()
        .name("didbot-firehose-held".to_owned())
        .spawn(move || loop {
            let oldest = held.oldest(DEFAULT_SYNC_INTERVAL);
            if producer.strong_count() == 0 {
                return;
            }
            let Some(oldest) = oldest else {
                continue;
            };
            let synced = held.log.wait_synced(oldest, DEFAULT_SYNC_INTERVAL);
            let Some(inner) = producer.upgrade() else {
                return;
            };
            if synced {
                Repos { inner }.release_synced();
            }
        })
        .map(drop)
}

/// The replay buffer's frames, and how much it holds.
#[derive(Debug)]
struct Ring {
    frames: VecDeque<Frame>,
    /// What `frames` come to, in encoded bytes.
    bytes: usize,
    /// The most it holds, or `None` while a boot is numbering, when it holds
    /// every frame. See [`Repos::listening`].
    bound: Option<ReplayBounds>,
    /// How far past `bound` it holds frames an open stream has not sent.
    slack: ReplayBounds,
    /// The newest sequence number evicted, or 0 before the first eviction.
    /// A stream whose position is below it has lost a frame.
    evicted: u64,
    /// Each open stream's position, by stream: the frames after it are the
    /// ones that stream has yet to send.
    streams: HashMap<u64, u64>,
    /// The key the next stream is given in `streams`.
    next_stream: u64,
}

impl Ring {
    /// Buffers `frame`, first evicting the oldest frames until it fits.
    ///
    /// A frame an open stream has yet to send is kept while the buffer is
    /// within its bound plus its slack.
    fn push(&mut self, frame: Frame) {
        if let Some(bound) = self.bound {
            let incoming = frame.bytes.len();
            let within = |ring: &Self, frames: usize, bytes: usize| {
                ring.frames.len() < frames && ring.bytes.saturating_add(incoming) <= bytes
            };
            let hard_frames = bound.frames.saturating_add(self.slack.frames);
            let hard_bytes = bound.bytes.saturating_add(self.slack.bytes);
            while let Some(oldest) = self.frames.front() {
                if within(self, bound.frames, bound.bytes) {
                    break;
                }
                let unsent = self.pinned().is_some_and(|after| oldest.seq > after);
                if unsent && within(self, hard_frames, hard_bytes) {
                    break;
                }
                self.bytes -= oldest.bytes.len();
                self.evicted = oldest.seq;
                self.frames.pop_front();
            }
        }
        self.bytes += frame.bytes.len();
        self.frames.push_back(frame);
    }

    /// The lowest position of any open stream that has not already lost a
    /// frame. A stream that has is ended when it next reads, so keeping
    /// frames for it would only hold memory.
    fn pinned(&self) -> Option<u64> {
        self.streams
            .values()
            .copied()
            .filter(|after| *after >= self.evicted)
            .min()
    }
}

impl Inner {
    /// Takes the ring lock, recovering from poisoning.
    ///
    /// Every operation under it is
    /// a push and a pop on a deque, so a panic elsewhere cannot have left it
    /// half-updated, and refusing to serve the stream over an unrelated panic
    /// would be a worse outcome than serving it.
    fn ring(&self) -> std::sync::MutexGuard<'_, Ring> {
        self.ring
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
    }
}

impl Repos {
    /// Builds a producer numbering frames from `sequence` and holding up to
    /// `bounds` of them for replay.
    #[must_use]
    pub fn new(sequence: Arc<Sequence>, bounds: ReplayBounds) -> Self {
        Self::build(sequence, bounds, false, None)
    }

    /// Builds a producer for a boot: it holds every frame numbered until
    /// [`Self::listening`], and `bounds` more after it.
    ///
    /// With `log`, the write-ahead log commits are written to, a commit is
    /// published only once `log` has synced past it. See the module docs.
    ///
    /// # Errors
    ///
    /// The thread that publishes held commits would not start.
    pub fn booting(
        sequence: Arc<Sequence>,
        bounds: ReplayBounds,
        log: Option<Arc<Wal>>,
    ) -> std::io::Result<Self> {
        let held = log.map(|log| Arc::new(Held::over(log)));
        let repos = Self::build(sequence, bounds, true, held.clone());
        if let Some(held) = held {
            publish_as_synced(&repos.inner, held)?;
        }
        Ok(repos)
    }

    fn build(
        sequence: Arc<Sequence>,
        bounds: ReplayBounds,
        booting: bool,
        held: Option<Arc<Held>>,
    ) -> Self {
        let bounds = ReplayBounds {
            frames: bounds.frames.max(1),
            ..bounds
        };
        Self {
            inner: Arc::new(Inner {
                sequence,
                bounds,
                ring: Mutex::new(Ring {
                    frames: VecDeque::new(),
                    bytes: 0,
                    bound: (!booting).then_some(bounds),
                    slack: bounds,
                    evicted: 0,
                    streams: HashMap::new(),
                    next_stream: 0,
                }),
                published: watch::Sender::new(0),
                held,
                closing: watch::Sender::new(false),
            }),
        }
    }

    /// Ends the boot [`Self::booting`] began: the buffer keeps every frame
    /// the boot numbered, and holds [`Self::bounds`] more on top of them,
    /// with the same slack for frames a stream has not sent.
    ///
    /// Called once the listener is bound. Nothing before that could have
    /// read a frame, and a relay that reconnects after a boot is owed every
    /// one the boot numbered. A producer that is not booting is left as it
    /// is.
    pub fn listening(&self) {
        let mut ring = self.ring();
        if ring.bound.is_none() {
            let held = ring.frames.len();
            let held_bytes = ring.bytes;
            let bounds = self.inner.bounds;
            ring.bound = Some(ReplayBounds {
                frames: held.saturating_add(bounds.frames),
                bytes: held_bytes.saturating_add(bounds.bytes),
            });
            info!(
                held,
                held_bytes,
                frames = bounds.frames,
                bytes = bounds.bytes,
                "the replay buffer holds every frame the boot numbered"
            );
        }
    }

    /// The newest sequence number assigned, or the restart floor if none has
    /// been.
    #[must_use]
    pub fn newest(&self) -> u64 {
        self.inner.sequence.latest()
    }

    /// How many streams are open right now.
    ///
    /// The per-connection state this producer holds is one receiver of
    /// `published` and one of `closing`, so this figure is the whole of it: a consumer that hung up
    /// and left its receiver behind would show up here and nowhere else.
    /// Observability rather than a bound — nothing refuses a subscriber for
    /// being the n-th — and the number a test watches to see a departed
    /// consumer actually freed.
    #[must_use]
    pub fn subscribers(&self) -> usize {
        self.inner.published.receiver_count()
    }

    /// How much the replay buffer holds at most, once any boot is over.
    #[must_use]
    pub fn bounds(&self) -> ReplayBounds {
        self.inner.bounds
    }

    fn ring(&self) -> std::sync::MutexGuard<'_, Ring> {
        self.inner.ring()
    }

    /// The oldest sequence number the buffer can still replay, or 0 when it
    /// is empty. As [`Self::since`], for tests: the replay path reads the
    /// same figure under the lock it does everything else under, in
    /// [`Self::start_for`].
    #[cfg(test)]
    fn oldest(&self) -> u64 {
        self.ring()
            .frames
            .front()
            .map(|frame| frame.seq)
            .unwrap_or(0)
    }

    /// Everything held after `seq`, oldest first.
    ///
    /// A stream reads the buffer a frame at a time, in [`ReposStream::next`];
    /// this is how a test looks at the buffer.
    #[cfg(test)]
    fn since(&self, seq: u64) -> Vec<Frame> {
        self.ring()
            .frames
            .iter()
            .filter(|frame| frame.seq > seq)
            .cloned()
            .collect()
    }

    /// Everything held, oldest first. As [`Self::since`], for tests.
    #[cfg(test)]
    fn all(&self) -> Vec<Frame> {
        self.ring().frames.iter().cloned().collect()
    }

    /// Where a consumer resuming from `resume` starts, decided under one
    /// acquisition of the ring lock, which also registers its stream so the
    /// frames after its start are kept for it.
    ///
    /// One acquisition for the whole decision, which is what makes the
    /// `outdated` answer true of the position beside it. Measuring the
    /// buffer and then reading it takes two acquisitions, and a publish
    /// landing between them evicts the very frame the cursor was just tested
    /// against: the consumer is then handed a later start with no `#info
    /// OutdatedCursor` in front of it, and believes it applied everything.
    /// A frame evicted after the decision is the stream's to notice, in
    /// [`ReposStream::next`].
    fn start_for(&self, resume: Resume, newest: u64) -> Start {
        let mut ring = self.ring();
        let oldest = ring.frames.front().map(|frame| frame.seq).unwrap_or(0);
        let (outdated, after) = match resume {
            Resume::Live => (false, ring.frames.back().map_or(0, |frame| frame.seq)),
            // The buffer is empty exactly when this run has published nothing,
            // so there is nothing to be behind. A cursor from before a restart
            // is still out of reach, so it is outdated even though nothing
            // this run sent is missing from it.
            Resume::After(seq) if oldest == 0 => (seq < newest, seq),
            // `seq + 1` is the frame the consumer wants next. It has to still
            // be in the buffer, or something between the two was dropped.
            Resume::After(seq) if seq + 1 >= oldest => (false, seq),
            // Everything held.
            Resume::After(_) => (true, oldest - 1),
            Resume::Future(_) => unreachable!("refused before the buffer is read"),
        };
        let stream = ring.next_stream;
        ring.next_stream += 1;
        ring.streams.insert(stream, after);
        Start {
            oldest,
            outdated,
            after,
            stream,
        }
    }

    /// Numbers, encodes, buffers and publishes one commit.
    ///
    /// Returns the frame so a caller — a test, mostly — can see what went
    /// out, and `None` when the stream could not number it: a frame numbered
    /// over a reservation that did not reach the log wears a number the next
    /// run will hand out again. See [`didbot_pds::Sequence::next`].
    pub fn publish_commit(&self, event: &RepoCommit) -> Option<Frame> {
        let time = now_rfc3339();
        self.publish(FrameKind::Commit, |seq| commit_body(seq, event, &time))
    }

    /// Numbers, encodes, buffers and publishes one identity event.
    pub fn publish_identity(&self, event: &RepoIdentity) -> Option<Frame> {
        let time = now_rfc3339();
        self.publish(FrameKind::Identity, |seq| identity_body(seq, event, &time))
    }

    /// Numbers, encodes, buffers and publishes one account event.
    pub fn publish_account(&self, event: &RepoAccount) -> Option<Frame> {
        let time = now_rfc3339();
        self.publish(FrameKind::Account, |seq| account_body(seq, event, &time))
    }

    /// Numbers, encodes, buffers and publishes one repository reset.
    pub fn publish_sync(&self, event: &RepoSync) -> Option<Frame> {
        let time = now_rfc3339();
        self.publish(FrameKind::Sync, |seq| sync_body(seq, event, &time))
    }

    /// Numbers a frame, puts it in the buffer and hands it to the live
    /// channel, under one lock.
    ///
    /// `None` when the number could not be reserved durably. The event is not
    /// buffered and not sent: a frame numbered over a reservation that did
    /// not land wears a number the next run will hand out again, and the
    /// consumer holding it as a cursor is told nothing. See
    /// [`didbot_pds::Sequence::next`].
    fn publish(&self, kind: FrameKind, body: impl FnOnce(u64) -> Value) -> Option<Frame> {
        let mut ring = self.ring();
        self.publish_locked(&mut ring, kind, body)
    }

    /// [`Self::publish`], under a ring lock the caller holds.
    fn publish_locked(
        &self,
        ring: &mut Ring,
        kind: FrameKind,
        body: impl FnOnce(u64) -> Value,
    ) -> Option<Frame> {
        // Inside the lock, so the buffer's order and the numbering agree.
        let Some(seq) = self.inner.sequence.next() else {
            warn!(
                ?kind,
                "an event was not put on subscribeRepos: it could not be numbered"
            );
            return None;
        };
        let frame = Frame {
            seq,
            kind,
            bytes: Bytes::from(encode_message(kind.tag(), &body(seq))),
        };
        ring.push(frame.clone());

        // Waking the streams never waits on one, so a consumer that cannot
        // keep up costs a writer nothing: the buffer evicts past it, and it
        // is told it fell behind when it next reads.
        self.inner.published.send_replace(seq);
        debug!(
            seq,
            ?kind,
            bytes = frame.bytes.len(),
            held = ring.frames.len(),
            held_bytes = ring.bytes,
            subscribers = self.inner.published.receiver_count(),
            "firehose frame published"
        );
        Some(frame)
    }

    /// Syncs the log and publishes every commit held back for it, oldest
    /// first.
    ///
    /// A clean shutdown calls this once serving has stopped, so the commits
    /// held back go out before [`Self::close`] ends the streams.
    pub fn release(&self) {
        self.sync_held_log();
        self.release_synced();
    }

    /// Publishes every held commit the log has synced past, oldest first.
    fn release_synced(&self) {
        let mut ring = self.ring();
        self.release_locked(&mut ring);
    }

    fn release_locked(&self, ring: &mut Ring) {
        let Some(held) = &self.inner.held else {
            return;
        };
        let synced = held.log.synced();
        while let Some(event) = held.take_through(synced) {
            let time = now_rfc3339();
            self.publish_locked(ring, FrameKind::Commit, |seq| {
                commit_body(seq, &event, &time)
            });
        }
    }

    /// Syncs the log the held commits are in, when there is one.
    fn sync_held_log(&self) {
        if let Some(held) = &self.inner.held {
            if let Err(error) = held.log.sync() {
                warn!(%error, "the log did not sync; the commits held for it stay held");
            }
        }
    }

    /// Publishes a frame that is not a commit, after every commit handed over
    /// before it. See the module docs.
    fn publish_after_held(
        &self,
        kind: FrameKind,
        body: impl FnOnce(u64) -> Value,
    ) -> Option<Frame> {
        self.sync_held_log();
        let mut ring = self.ring();
        self.release_locked(&mut ring);
        self.publish_locked(&mut ring, kind, body)
    }

    /// Ends every consumer's stream once it has sent what is already
    /// published.
    ///
    /// A `subscribeRepos` connection outlives the server that upgraded it
    /// until this process exits. A clean shutdown calls this after
    /// [`Self::release`], so each connection sends what it holds, closes and
    /// gives its slot back. See [`Self::detached`].
    pub fn close(&self) {
        self.inner.closing.send_replace(true);
    }

    /// Waits, for at most `timeout`, until no consumer is attached. Answers
    /// how many still are.
    pub async fn detached(&self, timeout: Duration) -> usize {
        let deadline = tokio::time::Instant::now() + timeout;
        while self.subscribers() > 0 && tokio::time::Instant::now() < deadline {
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
        self.subscribers()
    }

    /// Opens one connection's stream, resuming from `cursor` if it can.
    ///
    /// `Err` is the frame to send before closing, which is how the lexicon's
    /// declared errors go out: an error is a frame on the socket, not an HTTP
    /// status, because by the time a cursor has been read the upgrade has
    /// already happened.
    pub fn subscribe(
        &self,
        cursor: Option<i64>,
        exclude: Exclusions,
    ) -> Result<ReposStream, Vec<u8>> {
        let newest = self.newest();
        let resume = match cursor {
            None => Resume::Live,
            // A negative cursor names nothing. Read as "from the beginning",
            // which is the only reading that is not an error, and the buffer
            // then says how far back the beginning actually goes.
            Some(seq) if seq < 0 => Resume::After(0),
            Some(seq) if (seq as u64) > newest => Resume::Future(seq as u64),
            Some(seq) => Resume::After(seq as u64),
        };

        if let Resume::Future(seq) = resume {
            warn!(
                cursor = seq,
                newest,
                "a consumer asked to resume from a frame this server has not sent; if this data \
                 directory was restored from a snapshot, the restore skipped creating {}",
                didbot_pds::RESTORED_FILE
            );
            return Err(encode_error(
                FUTURE_CURSOR,
                &format!("cursor {seq} is ahead of this stream; the newest frame sent is {newest}"),
            ));
        }

        let Start {
            oldest,
            outdated,
            after,
            stream,
        } = self.start_for(resume, newest);

        if outdated {
            warn!(
                ?cursor,
                oldest, newest, "a consumer asked for a cursor this server no longer holds"
            );
        }
        let info = outdated.then(|| {
            Bytes::from(encode_message(
                "#info",
                &info_body(
                    OUTDATED_CURSOR,
                    &format!(
                        "this cursor is outside the replay buffer, which holds {oldest} to \
                         {newest}. Everything before it must be read from the repositories"
                    ),
                ),
            ))
        });
        info!(
            ?cursor,
            outdated,
            excluding = exclude.any(),
            after,
            "subscribeRepos consumer connected"
        );

        Ok(ReposStream {
            info,
            producer: Arc::downgrade(&self.inner),
            stream,
            after,
            published: self.inner.published.subscribe(),
            exclude,
            lagged: 0,
            closing: self.inner.closing.subscribe(),
        })
    }
}

/// A commit is held when the producer is over a log; every other frame
/// publishes after the commits held before it. See the module docs.
impl RepoEventSink for Repos {
    fn commit(&self, event: &RepoCommit) {
        match &self.inner.held {
            Some(held) => held.hold(event),
            None => {
                self.publish_commit(event);
            }
        }
    }

    fn identity(&self, event: &RepoIdentity) {
        let time = now_rfc3339();
        self.publish_after_held(FrameKind::Identity, |seq| identity_body(seq, event, &time));
    }

    fn account(&self, event: &RepoAccount) -> bool {
        let time = now_rfc3339();
        self.publish_after_held(FrameKind::Account, |seq| account_body(seq, event, &time))
            .is_some()
    }

    fn sync(&self, event: &RepoSync) -> bool {
        let time = now_rfc3339();
        self.publish_after_held(FrameKind::Sync, |seq| sync_body(seq, event, &time))
            .is_some()
    }
}

impl Default for Repos {
    fn default() -> Self {
        Self::new(Arc::new(Sequence::in_memory()), ReplayBounds::default())
    }
}

/// One consumer's stream: the replay, then whatever arrives live, both read
/// from the replay buffer.
///
/// A plain reader rather than a `Stream`, because a WebSocket is written to
/// rather than polled from — see `routes::subscribe_repos`.
#[derive(Debug)]
pub struct ReposStream {
    /// The `#info OutdatedCursor` frame to send before anything else.
    info: Option<Bytes>,
    /// Held weakly, so a stream does not keep a dropped producer alive.
    producer: Weak<Inner>,
    /// This stream's key in [`Ring::streams`].
    stream: u64,
    /// The highest sequence number this stream has sent or passed over. It
    /// sends the buffer's frames after this one. Copied to [`Ring::streams`]
    /// whenever it moves.
    after: u64,
    /// Changes whenever a frame is buffered.
    published: watch::Receiver<u64>,
    /// Which message types this consumer asked not to be sent.
    exclude: Exclusions,
    /// How many sequence numbers the buffer evicted before this consumer
    /// was sent them.
    lagged: u64,
    /// Whether the producer is closing. See [`Repos::close`].
    closing: watch::Receiver<bool>,
}

/// What one look at the buffer found for a stream.
enum Read {
    /// The next frame to send.
    Frame(Bytes),
    /// The buffer evicted frames after the stream's position; this many
    /// sequence numbers' worth.
    Behind(u64),
    /// Nothing new.
    Nothing,
}

impl ReposStream {
    /// How many frames this consumer has been too slow to receive.
    #[must_use]
    pub fn lagged(&self) -> u64 {
        self.lagged
    }

    /// The next frame's bytes, or `None` when the producer is gone, or is
    /// closing and this stream has sent what it was published.
    ///
    /// A consumer the buffer evicted a frame from before it was sent is
    /// dropped rather than silently skipped: the lexicon has
    /// `ConsumerTooSlow` for exactly this, and a stream that resumed after a
    /// gap would leave a relay believing it had seen every commit.
    pub async fn next(&mut self) -> Option<Bytes> {
        if let Some(info) = self.info.take() {
            return Some(info);
        }
        loop {
            // Marked seen before the buffer is read, so a frame buffered
            // after the read wakes the wait below.
            self.published.borrow_and_update();
            let producer = self.producer.upgrade()?;
            match self.read(&producer) {
                Read::Frame(bytes) => return Some(bytes),
                Read::Behind(skipped) => {
                    self.lagged = self.lagged.saturating_add(skipped);
                    warn!(
                        skipped,
                        after = self.after,
                        "a subscribeRepos consumer fell behind"
                    );
                    return None;
                }
                Read::Nothing => {}
            }
            drop(producer);
            // Everything buffered has gone out, so a close ends the stream.
            if *self.closing.borrow() {
                return None;
            }
            tokio::select! {
                changed = self.published.changed() => {
                    if changed.is_err() {
                        return None;
                    }
                }
                _ = self.closing.wait_for(|closing| *closing) => {}
            }
        }
    }

    /// The next frame after this stream's position that it wants, moving the
    /// position past it and past any frame the exclusions drop.
    fn read(&mut self, producer: &Inner) -> Read {
        let mut ring = producer.ring();
        if ring.evicted > self.after {
            return Read::Behind(ring.evicted - self.after);
        }
        let start = ring.frames.partition_point(|frame| frame.seq <= self.after);
        let mut found = Read::Nothing;
        for frame in ring.frames.range(start..) {
            self.after = frame.seq;
            // A dropped frame's sequence number is still spent, so the next
            // frame this consumer sees has a gap in front of it that it
            // asked for.
            if self.exclude.wants(frame.kind) {
                found = Read::Frame(frame.bytes.clone());
                break;
            }
        }
        ring.streams.insert(self.stream, self.after);
        found
    }
}

/// Gives back the frames this stream was keeping in the buffer.
impl Drop for ReposStream {
    fn drop(&mut self) {
        if let Some(producer) = self.producer.upgrade() {
            producer.ring().streams.remove(&self.stream);
        }
    }
}

/// The current time, RFC 3339, seconds precision.
fn now_rfc3339() -> String {
    time::OffsetDateTime::now_utc()
        .replace_nanosecond(0)
        .unwrap_or_else(|_| time::OffsetDateTime::now_utc())
        .format(&time::format_description::well_known::Rfc3339)
        .unwrap_or_default()
}

#[cfg(test)]
mod tests {
    use super::*;
    use didbot_pds::{AccountStatus, RepoAction, RepoOp, RepoSync};

    fn commit(rev: &str, since: Option<&str>) -> RepoCommit {
        RepoCommit {
            did: "did:web:a.agents.localhost%3A3000".to_owned(),
            commit: "bafyreie5737gdxlw5i64vzichcalba3z2v5n6icifvx5xytvske7mr3hpm".to_owned(),
            rev: rev.to_owned(),
            since: since.map(ToOwned::to_owned),
            prev_data: None,
            blocks: vec![0x0a, 0x0b],
            ops: vec![RepoOp {
                action: RepoAction::Create,
                path: "com.example.thing/3lq".to_owned(),
                cid: Some("bafyreie5737gdxlw5i64vzichcalba3z2v5n6icifvx5xytvske7mr3hpm".to_owned()),
                prev: None,
            }],
        }
    }

    /// Decodes a frame back into its header and body, the way a consumer does:
    /// one value, then another starting where the first stopped.
    fn decode(bytes: &[u8]) -> (Value, Value) {
        // The encoder is the only thing that knows where the split is, so the
        // test finds it by trying every offset a header could end at. Cheap,
        // and it proves the two values really are concatenated rather than
        // wrapped in something.
        for split in 1..bytes.len() {
            let (Ok(header), Ok(body)) = (
                dag_cbor::decode(&bytes[..split]),
                dag_cbor::decode(&bytes[split..]),
            ) else {
                continue;
            };
            return (header, body);
        }
        panic!("a frame did not decode as two concatenated DAG-CBOR values");
    }

    /// The `seq` a frame's body carries.
    fn seq_of(bytes: &[u8]) -> u64 {
        let body = decode(bytes).1;
        let Value::Integer(seq) = field(&body, "seq") else {
            panic!("seq is not an integer");
        };
        *seq as u64
    }

    fn field<'a>(value: &'a Value, key: &str) -> &'a Value {
        let Value::Map(fields) = value else {
            panic!("not a map");
        };
        fields.get(key).unwrap_or_else(|| panic!("no {key}"))
    }

    #[test]
    fn a_commit_frame_is_a_header_then_the_body_the_lexicon_names() {
        let repos = Repos::default();
        let frame = repos
            .publish_commit(&commit("3lqaaa", None))
            .expect("the frame was numbered");
        let (header, body) = decode(&frame.bytes);
        assert_eq!(field(&header, "op"), &Value::Integer(1));
        assert_eq!(field(&header, "t"), &Value::String("#commit".to_owned()));
        assert_eq!(field(&body, "seq"), &Value::Integer(1));
        assert_eq!(
            field(&body, "repo"),
            &Value::String(commit("3lqaaa", None).did)
        );
        assert_eq!(field(&body, "rev"), &Value::String("3lqaaa".to_owned()));
        assert_eq!(field(&body, "blocks"), &Value::Bytes(vec![0x0a, 0x0b]));
        assert_eq!(field(&body, "rebase"), &Value::Bool(false));
        assert_eq!(field(&body, "tooBig"), &Value::Bool(false));
        assert_eq!(field(&body, "blobs"), &Value::List(Vec::new()));
    }

    /// `since` is required and nullable, so a repository's first commit sends
    /// it as null rather than leaving it out.
    #[test]
    fn a_first_commit_sends_a_null_since_rather_than_no_since() {
        let repos = Repos::default();
        let first = repos
            .publish_commit(&commit("3lqaaa", None))
            .expect("the frame was numbered");
        assert_eq!(field(&decode(&first.bytes).1, "since"), &Value::Null);
        let second = repos
            .publish_commit(&commit("3lqbbb", Some("3lqaaa")))
            .expect("the frame was numbered");
        assert_eq!(
            field(&decode(&second.bytes).1, "since"),
            &Value::String("3lqaaa".to_owned())
        );
    }

    #[test]
    fn an_op_carries_its_action_path_and_cid_link() {
        let repos = Repos::default();
        let frame = repos
            .publish_commit(&commit("3lqaaa", None))
            .expect("the frame was numbered");
        let body = decode(&frame.bytes).1;
        let Value::List(ops) = field(&body, "ops") else {
            panic!("ops is not a list");
        };
        assert_eq!(ops.len(), 1);
        assert_eq!(
            field(&ops[0], "action"),
            &Value::String("create".to_owned())
        );
        assert_eq!(
            field(&ops[0], "path"),
            &Value::String("com.example.thing/3lq".to_owned())
        );
        assert!(matches!(field(&ops[0], "cid"), Value::Link(_)));
    }

    #[test]
    fn a_deletion_sends_a_null_cid_and_names_what_it_removed() {
        let repos = Repos::default();
        let mut event = commit("3lqaaa", None);
        event.ops[0] = RepoOp {
            action: RepoAction::Delete,
            path: "com.example.thing/3lq".to_owned(),
            cid: None,
            prev: Some("bafyreie5737gdxlw5i64vzichcalba3z2v5n6icifvx5xytvske7mr3hpm".to_owned()),
        };
        let frame = repos
            .publish_commit(&event)
            .expect("the frame was numbered");
        let body = decode(&frame.bytes).1;
        let Value::List(ops) = field(&body, "ops") else {
            panic!("ops is not a list");
        };
        assert_eq!(
            field(&ops[0], "action"),
            &Value::String("delete".to_owned())
        );
        assert_eq!(field(&ops[0], "cid"), &Value::Null);
        assert!(
            matches!(field(&ops[0], "prev"), Value::Link(_)),
            "a delete op must name the record it removed, or nothing can invert it"
        );
    }

    /// A create carries no `prev` at all — not `prev: null`. The lexicon
    /// marks the field optional and says a creation "should not define" it,
    /// and a null would claim the key held something nameless before.
    #[test]
    fn a_create_leaves_prev_off_the_wire_entirely() {
        let repos = Repos::default();
        let frame = repos
            .publish_commit(&commit("3lqaaa", None))
            .expect("the frame was numbered");
        let body = decode(&frame.bytes).1;
        let Value::List(ops) = field(&body, "ops") else {
            panic!("ops is not a list");
        };
        let Value::Map(fields) = &ops[0] else {
            panic!("an op is not a map");
        };
        assert!(!fields.contains_key("prev"), "a create defined `prev`");
    }

    #[test]
    fn sequence_numbers_start_at_one_and_do_not_skip() {
        let repos = Repos::default();
        let seqs: Vec<u64> = (0..5)
            .map(|_| {
                repos
                    .publish_commit(&commit("3lqaaa", None))
                    .expect("the frame was numbered")
                    .seq
            })
            .collect();
        assert_eq!(seqs, vec![1, 2, 3, 4, 5]);
        assert_eq!(repos.newest(), 5);
    }

    /// The replay is a share of the buffer's frames, not a copy of them.
    ///
    /// Pointer identity rather than a memory measurement: what makes a
    /// `?cursor=0` connection affordable is that its queue and the ring point
    /// at the same allocation, and that is a structural fact a test can hold
    /// even though "how many bytes did that cost" is not.
    #[tokio::test]
    async fn a_replayed_frame_points_at_the_buffer_rather_than_a_copy() {
        let repos = Repos::new(Arc::new(Sequence::in_memory()), ReplayBounds::frames(8));
        for _ in 0..3 {
            repos
                .publish_commit(&commit("3lqaaa", None))
                .expect("the frame was numbered");
        }
        let held: Vec<*const u8> = repos
            .ring()
            .frames
            .iter()
            .map(|frame| frame.bytes.as_ptr())
            .collect();

        let mut stream = repos
            .subscribe(Some(0), Exclusions::default())
            .expect("a cursor of 0 replays everything held");
        let mut replayed = Vec::new();
        for _ in 0..held.len() {
            replayed.push(stream.next().await.expect("a replayed frame").as_ptr());
        }
        assert_eq!(
            replayed, held,
            "every replayed frame is the buffer's own allocation"
        );
    }

    /// The same for the live half, which is the other place a frame is
    /// handed to every stream.
    #[tokio::test]
    async fn a_live_frame_reaches_every_subscriber_as_one_allocation() {
        let repos = Repos::new(Arc::new(Sequence::in_memory()), ReplayBounds::frames(8));
        let mut streams: Vec<ReposStream> = (0..3)
            .map(|_| {
                repos
                    .subscribe(None, Exclusions::default())
                    .expect("a live subscription refuses nothing")
            })
            .collect();

        let published = repos
            .publish_commit(&commit("3lqaaa", None))
            .expect("the frame was numbered")
            .bytes
            .as_ptr();
        for stream in &mut streams {
            let got = stream.next().await.expect("the live frame");
            assert_eq!(
                got.as_ptr(),
                published,
                "a live frame is shared with every receiver, not copied per receiver"
            );
        }
    }

    /// The live channel hands frames out in sequence order, however many
    /// writers race to publish: a relay reads `seq` as monotonic within a
    /// connection, so a frame that arrives after a higher one is a gap and
    /// then a stale frame, not a reordering it can undo.
    ///
    /// Arrival order, not merely the set: the replay buffer is ordered by
    /// the same lock, so a sorted comparison would pass with the send outside
    /// it. See [`crate::race`] for what makes the race one this test loses
    /// every time. The channel is deep enough that the reader never lags.
    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
    async fn live_frames_arrive_in_sequence_order_under_concurrent_writes() {
        const WRITERS: usize = 16;
        const EACH: usize = 200;
        let total = WRITERS * EACH;
        let repos = Repos::new(Arc::new(Sequence::in_memory()), ReplayBounds::frames(total));
        let mut stream = repos
            .subscribe(None, Exclusions::default())
            .expect("a live subscription refuses nothing");

        let writer = repos.clone();
        crate::race::under_pressure(WRITERS, EACH, move || {
            writer
                .publish_commit(&commit("3lqaaa", None))
                .expect("the frame was numbered");
        });

        let mut arrived = Vec::with_capacity(total);
        for _ in 0..total {
            let bytes = stream.next().await.expect("every published frame arrives");
            arrived.push(seq_of(&bytes));
        }
        let expected: Vec<u64> = (1..=total as u64).collect();
        assert_eq!(
            arrived, expected,
            "a live subscriber received a frame out of sequence order"
        );
        assert_eq!(stream.lagged(), 0);
    }

    fn sync(did: &str) -> RepoSync {
        RepoSync {
            did: did.to_owned(),
            rev: "3lqaaa".to_owned(),
            blocks: vec![0x0a, 0x0b],
        }
    }

    /// `#sync` carries the five fields the lexicon requires, and no others.
    #[test]
    fn a_sync_frame_carries_the_did_the_rev_and_the_commit_block() {
        let repos = Repos::default();
        let frame = repos
            .publish_sync(&sync("did:web:a"))
            .expect("the frame was numbered");
        let (header, body) = decode(&frame.bytes);
        assert_eq!(field(&header, "t"), &Value::String("#sync".to_owned()));
        let Value::Map(fields) = &body else {
            panic!("not a map");
        };
        assert_eq!(
            fields.keys().map(String::as_str).collect::<Vec<_>>(),
            ["blocks", "did", "rev", "seq", "time"]
        );
        assert_eq!(field(&body, "blocks"), &Value::Bytes(vec![0x0a, 0x0b]));
        assert_eq!(field(&body, "did"), &Value::String("did:web:a".to_owned()));
        assert_eq!(field(&body, "rev"), &Value::String("3lqaaa".to_owned()));
        assert_eq!(field(&body, "seq"), &Value::Integer(1));
    }

    /// A consumer excluding every kind a parameter names still hears that a
    /// repository was reset.
    #[tokio::test]
    async fn no_exclusion_drops_a_sync() {
        let repos = Repos::new(Arc::new(Sequence::in_memory()), ReplayBounds::frames(8));
        repos
            .publish_commit(&commit("3lqaaa", None))
            .expect("the frame was numbered");
        repos
            .publish_sync(&sync("did:web:a"))
            .expect("the frame was numbered");
        let mut stream = repos
            .subscribe(
                Some(0),
                Exclusions {
                    commit: true,
                    identity: true,
                    account: true,
                },
            )
            .expect("a resumable cursor");
        let (header, _) = decode(&stream.next().await.expect("the #sync"));
        assert_eq!(field(&header, "t"), &Value::String("#sync".to_owned()));
    }

    /// Every frame a boot numbers is held for the first consumer, however
    /// many there are, and once the listener binds the buffer holds its
    /// capacity on top of them.
    #[tokio::test]
    async fn a_boot_keeps_every_frame_it_numbered_and_its_capacity_on_top() {
        let repos = Repos::booting(
            Arc::new(Sequence::in_memory()),
            ReplayBounds::frames(3),
            None,
        )
        .expect("a producer");
        for _ in 0..10 {
            repos
                .publish_account(&account("did:web:a", true))
                .expect("the frame was numbered");
        }
        assert_eq!(repos.all().len(), 10, "a boot evicts nothing");

        repos.listening();
        for _ in 0..3 {
            repos
                .publish_commit(&commit("3lqaaa", None))
                .expect("the frame was numbered");
        }
        assert_eq!(repos.all().len(), 13, "the capacity is on top of the boot");
        repos
            .publish_commit(&commit("3lqbbb", None))
            .expect("the frame was numbered");
        assert_eq!(repos.all().len(), 13, "and bounded after it");
        assert_eq!(repos.oldest(), 2);

        let mut stream = repos
            .subscribe(Some(1), Exclusions::default())
            .expect("a resumable cursor");
        let mut seqs = Vec::new();
        for _ in 0..13 {
            seqs.push(seq_of(&stream.next().await.expect("a replayed frame")));
        }
        assert_eq!(seqs, (2..=14).collect::<Vec<u64>>());
    }

    #[test]
    fn the_buffer_drops_its_oldest_rather_than_growing() {
        let repos = Repos::new(Arc::new(Sequence::in_memory()), ReplayBounds::frames(3));
        for _ in 0..6 {
            repos
                .publish_commit(&commit("3lqaaa", None))
                .expect("the frame was numbered");
        }
        assert_eq!(repos.oldest(), 4);
        assert_eq!(repos.all().len(), 3);
    }

    #[test]
    fn resuming_replays_exactly_what_came_after_the_cursor() {
        let repos = Repos::new(Arc::new(Sequence::in_memory()), ReplayBounds::frames(16));
        for _ in 0..5 {
            repos
                .publish_commit(&commit("3lqaaa", None))
                .expect("the frame was numbered");
        }
        let seqs: Vec<u64> = repos.since(2).iter().map(|frame| frame.seq).collect();
        assert_eq!(seqs, vec![3, 4, 5]);
    }

    /// A cursor naming a frame this server has not sent cannot be filled in
    /// from anywhere, so it is the lexicon's error rather than an `#info`.
    #[test]
    fn a_future_cursor_is_the_error_the_lexicon_declares() {
        let repos = Repos::default();
        repos
            .publish_commit(&commit("3lqaaa", None))
            .expect("the frame was numbered");
        let refused = repos
            .subscribe(Some(9), Exclusions::default())
            .expect_err("a future cursor errors");
        let (header, body) = decode(&refused);
        assert_eq!(field(&header, "op"), &Value::Integer(-1));
        assert_eq!(
            field(&body, "error"),
            &Value::String(FUTURE_CURSOR.to_owned())
        );
        assert!(
            repos.subscribe(Some(1), Exclusions::default()).is_ok(),
            "the newest frame is not future"
        );
    }

    fn identity(did: &str) -> RepoIdentity {
        RepoIdentity {
            did: did.to_owned(),
            handle: Some("kestrel.agents.localhost".to_owned()),
        }
    }

    fn account(did: &str, active: bool) -> RepoAccount {
        RepoAccount {
            did: did.to_owned(),
            active,
            status: (!active).then_some(AccountStatus::Deleted),
        }
    }

    #[test]
    fn an_identity_frame_names_the_did_and_the_handle_it_answers_to() {
        let repos = Repos::default();
        let frame = repos
            .publish_identity(&identity("did:web:a"))
            .expect("the frame was numbered");
        let (header, body) = decode(&frame.bytes);
        assert_eq!(field(&header, "t"), &Value::String("#identity".to_owned()));
        assert_eq!(field(&body, "did"), &Value::String("did:web:a".to_owned()));
        assert_eq!(
            field(&body, "handle"),
            &Value::String("kestrel.agents.localhost".to_owned())
        );
        assert_eq!(field(&body, "seq"), &Value::Integer(1));
    }

    /// The lexicon says `status` is a reason an account is *not* active, so an
    /// active one carries no status at all rather than one saying it is fine.
    #[test]
    fn an_active_account_carries_no_status_and_a_deleted_one_says_deleted() {
        let repos = Repos::default();
        let live = decode(
            &repos
                .publish_account(&account("did:web:a", true))
                .expect("the frame was numbered")
                .bytes,
        )
        .1;
        assert_eq!(field(&live, "active"), &Value::Bool(true));
        let Value::Map(fields) = &live else {
            panic!("not a map");
        };
        assert!(!fields.contains_key("status"), "an active account has none");

        let gone = decode(
            &repos
                .publish_account(&account("did:web:a", false))
                .expect("the frame was numbered")
                .bytes,
        )
        .1;
        assert_eq!(field(&gone, "active"), &Value::Bool(false));
        assert_eq!(field(&gone, "status"), &Value::String("deleted".to_owned()));
    }

    /// The three kinds share one sequence space, so a consumer resuming from
    /// a cursor sees all of them, interleaved, in order, once each.
    #[tokio::test]
    async fn all_three_kinds_share_one_sequence_and_replay_together() {
        let repos = Repos::new(Arc::new(Sequence::in_memory()), ReplayBounds::frames(64));
        repos
            .publish_identity(&identity("did:web:a"))
            .expect("the frame was numbered");
        repos
            .publish_account(&account("did:web:a", true))
            .expect("the frame was numbered");
        repos
            .publish_commit(&commit("3lqaaa", None))
            .expect("the frame was numbered");
        repos
            .publish_commit(&commit("3lqbbb", Some("3lqaaa")))
            .expect("the frame was numbered");
        repos
            .publish_account(&account("did:web:a", false))
            .expect("the frame was numbered");

        let mut stream = repos
            .subscribe(Some(1), Exclusions::default())
            .expect("a resumable cursor");
        let mut seen = Vec::new();
        for _ in 0..4 {
            let bytes = stream.next().await.expect("a replayed frame");
            let (header, _) = decode(&bytes);
            let Value::String(tag) = field(&header, "t") else {
                panic!("no tag");
            };
            seen.push((tag.clone(), seq_of(&bytes)));
        }
        assert_eq!(
            seen,
            vec![
                ("#account".to_owned(), 2),
                ("#commit".to_owned(), 3),
                ("#commit".to_owned(), 4),
                ("#account".to_owned(), 5),
            ],
            "the three kinds must share one gapless, ordered sequence"
        );
    }

    /// The load-bearing one for the exclusions: a consumer that excludes a
    /// kind never sees it, still sees the rest, and can reconnect with a
    /// cursor that skipped past frames it never saw.
    #[tokio::test]
    async fn an_exclusion_drops_frames_without_disturbing_the_sequence() {
        let repos = Repos::new(Arc::new(Sequence::in_memory()), ReplayBounds::frames(64));
        repos
            .publish_identity(&identity("did:web:a"))
            .expect("the frame was numbered");
        repos
            .publish_commit(&commit("3lqaaa", None))
            .expect("the frame was numbered");
        repos
            .publish_identity(&identity("did:web:b"))
            .expect("the frame was numbered");
        repos
            .publish_account(&account("did:web:b", true))
            .expect("the frame was numbered");

        let exclude = Exclusions {
            identity: true,
            ..Exclusions::default()
        };
        let mut stream = repos.subscribe(Some(0), exclude).expect("from the start");
        let first = stream.next().await.expect("a frame");
        assert_eq!(seq_of(&first), 2, "the identity at 1 was not delivered");
        let second = stream.next().await.expect("a frame");
        assert_eq!(seq_of(&second), 4, "the identity at 3 was not delivered");

        // Its cursor is now 4, which is past a frame it never saw. Resuming
        // from it is neither an error nor a replay of the excluded frames.
        repos
            .publish_commit(&commit("3lqbbb", Some("3lqaaa")))
            .expect("the frame was numbered");
        let mut again = repos.subscribe(Some(4), exclude).expect("a live cursor");
        assert_eq!(seq_of(&again.next().await.expect("a frame")), 5);
    }

    /// Excluding everything is legal and produces a connection that stays
    /// open and says nothing, rather than one that errors.
    #[tokio::test]
    async fn excluding_every_kind_leaves_a_stream_that_is_merely_quiet() {
        let repos = Repos::new(Arc::new(Sequence::in_memory()), ReplayBounds::frames(64));
        repos
            .publish_commit(&commit("3lqaaa", None))
            .expect("the frame was numbered");
        let mut stream = repos
            .subscribe(
                Some(0),
                Exclusions {
                    commit: true,
                    identity: true,
                    account: true,
                },
            )
            .expect("excluding everything is not an error");
        assert!(
            tokio::time::timeout(std::time::Duration::from_millis(50), stream.next())
                .await
                .is_err(),
            "a frame arrived on a stream that excluded every kind"
        );
    }

    /// A cursor at the eviction boundary, resumed while writes are landing.
    ///
    /// Whether the cursor is still in the buffer and what the buffer holds
    /// are one answer, taken from one snapshot. Read apart, a publish between
    /// them evicts the frame the cursor was just tested against, and the
    /// consumer is handed a later start with no `#info OutdatedCursor` in
    /// front of it — believing it applied everything, which is the one thing
    /// that frame exists to prevent.
    ///
    /// The assertion is one-sided: a replay that neither begins at
    /// `cursor + 1` nor is announced as outdated is only reachable through
    /// that gap, so this fails when the gap is back and at no other time.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn a_cursor_evicted_while_the_buffer_is_read_is_told_it_is_outdated() {
        use std::sync::atomic::{AtomicBool, Ordering};

        const CAPACITY: usize = 4;
        const ROUNDS: usize = 4096;
        let repos = Repos::new(
            Arc::new(Sequence::in_memory()),
            ReplayBounds::frames(CAPACITY),
        );
        // One frame up front, so the first round resumes against a buffer
        // that already holds something.
        repos
            .publish_commit(&commit("3lqaaa", None))
            .expect("the frame was numbered");

        let writing = Arc::new(AtomicBool::new(true));
        let writer = repos.clone();
        let publishing = writing.clone();
        let publisher = std::thread::spawn(move || {
            while publishing.load(Ordering::Relaxed) {
                writer
                    .publish_commit(&commit("3lqaaa", None))
                    .expect("the frame was numbered");
            }
        });

        for _ in 0..ROUNDS {
            // One eviction from being out of reach, which is the only cursor
            // the gap can be observed through.
            let cursor = repos.oldest().saturating_sub(1);
            let mut stream = repos
                .subscribe(Some(cursor as i64), Exclusions::default())
                .expect("a cursor inside the buffer is not ahead of the stream");
            // `None` is a consumer that fell behind the live channel, which
            // is its own answer and not this one.
            let Some(frame) = stream.next().await else {
                continue;
            };
            let (header, body) = decode(&frame);
            if field(&header, "t") == &Value::String("#info".to_owned()) {
                assert_eq!(
                    field(&body, "name"),
                    &Value::String(OUTDATED_CURSOR.to_owned()),
                    "the only `#info` this server sends is an outdated cursor"
                );
                continue;
            }
            assert_eq!(
                seq_of(&frame),
                cursor + 1,
                "a consumer whose cursor was evicted while the buffer was read \
                 was served a later frame with nothing in front of it saying so"
            );
        }

        writing.store(false, Ordering::Relaxed);
        publisher.join().expect("the publisher does not panic");
    }

    #[tokio::test]
    async fn a_cursor_outside_the_buffer_gets_an_info_frame_and_then_what_is_left() {
        let repos = Repos::new(Arc::new(Sequence::in_memory()), ReplayBounds::frames(3));
        for _ in 0..6 {
            repos
                .publish_commit(&commit("3lqaaa", None))
                .expect("the frame was numbered");
        }
        let mut stream = repos
            .subscribe(Some(1), Exclusions::default())
            .expect("1 is not a future cursor");
        let (header, body) = decode(&stream.next().await.expect("an info frame"));
        assert_eq!(field(&header, "t"), &Value::String("#info".to_owned()));
        assert_eq!(
            field(&body, "name"),
            &Value::String(OUTDATED_CURSOR.to_owned())
        );
        // Then everything the buffer still holds, in order and once each.
        let mut seqs = Vec::new();
        for _ in 0..3 {
            seqs.push(seq_of(&stream.next().await.expect("a replayed frame")));
        }
        assert_eq!(seqs, vec![4u64, 5, 6]);
    }

    /// What a restart does to a cursor.
    ///
    /// The numbering survives a restart and the replay buffer does not, so
    /// every cursor minted before one is out of reach. It must be answered
    /// with `#info OutdatedCursor` — a consumer that is silently treated as
    /// caught up would resume against a repository it has an unfillable hole
    /// in and never learn of it. And it must *not* be refused as a
    /// `FutureCursor`: it names a number below the reservation floor this run
    /// resumed at, so it is behind, not ahead.
    #[tokio::test]
    async fn a_cursor_from_before_a_restart_is_told_it_is_outdated() {
        // The previous run handed out seven numbers and reserved a whole
        // block, so this one resumes at the top of that block.
        let floor = didbot_pds::RESERVATION;
        let sequence = Sequence::new(Arc::new(didbot_pds::MemorySequenceLog), floor);
        let repos = Repos::new(Arc::new(sequence), ReplayBounds::frames(64));
        assert_eq!(repos.newest(), floor, "the restart resumed at the floor");
        assert_eq!(repos.all().len(), 0, "and with an empty replay buffer");

        let mut stream = repos
            .subscribe(Some(7), Exclusions::default())
            .expect("a cursor from before a restart is behind, not ahead");
        // Timed out rather than awaited: the failure this guards against is
        // a stream that says nothing at all, and an unbounded await on one
        // would hang instead of failing.
        let frame = tokio::time::timeout(std::time::Duration::from_secs(10), stream.next())
            .await
            .expect("a consumer from before the restart was quietly treated as caught up")
            .expect("the stream is open");
        let (header, body) = decode(&frame);
        assert_eq!(field(&header, "t"), &Value::String("#info".to_owned()));
        assert_eq!(
            field(&body, "name"),
            &Value::String(OUTDATED_CURSOR.to_owned())
        );

        // And nothing this run sends can wear a number the last one did.
        assert!(
            repos
                .publish_commit(&commit("3lqaaa", None))
                .expect("the frame was numbered")
                .seq
                > floor,
            "a frame repeated a number a previous run could have handed out"
        );
    }

    /// The load-bearing one for a consumer that stops reading: it is *ended*,
    /// not silently fast-forwarded. A stream that resumed after a gap would
    /// leave a relay believing it had applied every commit when it had not,
    /// and the gap would never be visible again.
    #[tokio::test]
    async fn a_stream_that_fell_behind_ends_rather_than_skipping_to_the_present() {
        let repos = Repos::new(Arc::new(Sequence::in_memory()), ReplayBounds::frames(4));
        let mut stream = repos
            .subscribe(None, Exclusions::default())
            .expect("a live consumer");
        // Nothing is read while these are published, so the live channel
        // overruns and the frames at the front of it are gone.
        for _ in 0..12 {
            repos
                .publish_commit(&commit("3lqaaa", None))
                .expect("the frame was numbered");
        }
        assert_eq!(
            stream.next().await,
            None,
            "a consumer that fell behind was handed a frame with a hole in front of it"
        );
        assert!(
            stream.lagged() > 0,
            "the stream ended without recording that anything was dropped"
        );
    }

    /// The writer's side of the same fact: publishing does not wait on a
    /// subscriber that is not reading. `Repos` is driven from inside a write,
    /// so a `publish` that blocked here would be a way for any consumer to
    /// halt every write on the server.
    #[tokio::test]
    async fn publishing_does_not_wait_on_a_subscriber_that_never_reads() {
        let repos = Repos::new(Arc::new(Sequence::in_memory()), ReplayBounds::frames(4));
        let _idle = repos
            .subscribe(None, Exclusions::default())
            .expect("a live consumer");
        let published = tokio::task::spawn_blocking({
            let repos = repos.clone();
            move || {
                // Far past the channel's depth, from a thread with no
                // reactor under it: if `publish` waited on the reader, this
                // never returns.
                (0..64)
                    .map(|_| {
                        repos
                            .publish_commit(&commit("3lqaaa", None))
                            .expect("the frame was numbered")
                            .seq
                    })
                    .collect::<Vec<u64>>()
            }
        });
        let published = tokio::time::timeout(std::time::Duration::from_secs(10), published)
            .await
            .expect("publishing blocked on a subscriber that was not reading")
            .expect("the publishing thread did not panic");
        assert_eq!(
            published,
            (1..=64).collect::<Vec<u64>>(),
            "every write was numbered, in order, with the reader stalled"
        );
        assert_eq!(repos.all().len(), 4, "and the buffer stayed at its bound");
    }

    /// The exact edge of the replay window. A cursor naming the frame just
    /// before the oldest one held is still reachable — the next frame the
    /// consumer wants is the oldest one — and one number below that is not.
    #[tokio::test]
    async fn the_oldest_frame_held_is_reachable_and_the_one_before_it_is_not() {
        let repos = Repos::new(Arc::new(Sequence::in_memory()), ReplayBounds::frames(3));
        for _ in 0..6 {
            repos
                .publish_commit(&commit("3lqaaa", None))
                .expect("the frame was numbered");
        }
        let oldest = repos.oldest();
        assert_eq!(oldest, 4, "the buffer holds 4, 5 and 6");

        let mut reachable = repos
            .subscribe(Some(oldest as i64 - 1), Exclusions::default())
            .expect("not a future cursor");
        assert_eq!(
            seq_of(&reachable.next().await.expect("a frame")),
            oldest,
            "a cursor one below the oldest frame held must replay it, not admit a gap"
        );

        let mut lost = repos
            .subscribe(Some(oldest as i64 - 2), Exclusions::default())
            .expect("not a future cursor");
        let (header, body) = decode(&lost.next().await.expect("a frame"));
        assert_eq!(field(&header, "t"), &Value::String("#info".to_owned()));
        assert_eq!(
            field(&body, "name"),
            &Value::String(OUTDATED_CURSOR.to_owned()),
            "a cursor two below the oldest frame held has a hole and must be told so"
        );
    }

    /// The load-bearing one: a consumer that resumes from its last cursor sees
    /// every frame after it exactly once, with nothing repeated and nothing
    /// skipped, including frames published while it was reconnecting.
    #[tokio::test]
    async fn a_resumed_consumer_sees_every_frame_after_its_cursor_exactly_once() {
        let repos = Repos::new(Arc::new(Sequence::in_memory()), ReplayBounds::frames(64));
        for _ in 0..3 {
            repos
                .publish_commit(&commit("3lqaaa", None))
                .expect("the frame was numbered");
        }
        let mut first = repos
            .subscribe(None, Exclusions::default())
            .expect("a live consumer");
        for _ in 0..2 {
            repos
                .publish_commit(&commit("3lqbbb", None))
                .expect("the frame was numbered");
        }
        let mut seen = Vec::new();
        for _ in 0..2 {
            seen.push(seq_of(&first.next().await.expect("a live frame")));
        }
        assert_eq!(seen, vec![4, 5]);

        // It reconnects from the last number it applied, and two more frames
        // landed while it was away.
        drop(first);
        for _ in 0..2 {
            repos
                .publish_commit(&commit("3lqccc", None))
                .expect("the frame was numbered");
        }
        let mut again = repos
            .subscribe(Some(5), Exclusions::default())
            .expect("a resumable cursor");
        for _ in 0..2 {
            seen.push(seq_of(&again.next().await.expect("a replayed frame")));
        }
        assert_eq!(seen, vec![4, 5, 6, 7], "no gaps and no repeats");
    }

    /// A commit at the size the lexicon allows its blocks.
    fn largest_commit() -> RepoCommit {
        RepoCommit {
            blocks: vec![0; 2_000_000],
            ..commit("3lqaaa", None)
        }
    }

    /// The byte bound is what holds when records are large: a frame count
    /// alone lets a burst of the largest commits take gigabytes.
    #[test]
    fn a_burst_of_large_frames_stays_under_the_byte_bound_and_keeps_the_newest() {
        const BYTES: usize = 16 * 1024 * 1024;
        let repos = Repos::new(
            Arc::new(Sequence::in_memory()),
            ReplayBounds {
                frames: 4096,
                bytes: BYTES,
            },
        );
        for _ in 0..64 {
            repos
                .publish_commit(&largest_commit())
                .expect("the frame was numbered");
        }
        let held = repos.all();
        let total: usize = held.iter().map(|frame| frame.bytes.len()).sum();
        assert!(
            total <= BYTES,
            "the buffer holds {total} bytes, past {BYTES}"
        );
        assert_eq!(
            repos.ring().bytes,
            total,
            "the running total is the real one"
        );
        assert_eq!(
            held.len(),
            BYTES / held[0].bytes.len(),
            "as many frames as fit are kept"
        );
        let seqs: Vec<u64> = held.iter().map(|frame| frame.seq).collect();
        let first = 64 - held.len() as u64 + 1;
        assert_eq!(
            seqs,
            (first..=64).collect::<Vec<u64>>(),
            "the oldest went first"
        );
    }

    /// Live consumers read from the buffer, so it keeps the newest frame
    /// even when that frame alone is past the byte bound.
    #[test]
    fn a_frame_larger_than_the_byte_bound_is_still_held_alone() {
        let repos = Repos::new(
            Arc::new(Sequence::in_memory()),
            ReplayBounds {
                frames: 64,
                bytes: 1024,
            },
        );
        repos
            .publish_commit(&commit("3lqaaa", None))
            .expect("the frame was numbered");
        repos
            .publish_commit(&largest_commit())
            .expect("the frame was numbered");
        let seqs: Vec<u64> = repos.all().iter().map(|frame| frame.seq).collect();
        assert_eq!(seqs, vec![2]);
    }

    #[tokio::test]
    async fn a_cursor_the_byte_bound_evicted_is_told_it_is_outdated() {
        let size = Repos::default()
            .publish_commit(&commit("3lqaaa", None))
            .expect("the frame was numbered")
            .bytes
            .len();
        let repos = Repos::new(
            Arc::new(Sequence::in_memory()),
            ReplayBounds {
                frames: 64,
                bytes: 3 * size,
            },
        );
        for _ in 0..6 {
            repos
                .publish_commit(&commit("3lqaaa", None))
                .expect("the frame was numbered");
        }
        assert_eq!(repos.oldest(), 4, "three frames fit in the byte bound");

        let mut stream = repos
            .subscribe(Some(1), Exclusions::default())
            .expect("1 is not a future cursor");
        let (header, body) = decode(&stream.next().await.expect("an info frame"));
        assert_eq!(field(&header, "t"), &Value::String("#info".to_owned()));
        assert_eq!(
            field(&body, "name"),
            &Value::String(OUTDATED_CURSOR.to_owned())
        );
        let mut seqs = Vec::new();
        for _ in 0..3 {
            seqs.push(seq_of(&stream.next().await.expect("a replayed frame")));
        }
        assert_eq!(seqs, vec![4u64, 5, 6]);
    }

    /// A consumer that stops reading holds nothing the buffer has let go
    /// of. The buffer evicts past it, and it is ended rather than handed a
    /// gap.
    #[tokio::test]
    async fn a_stalled_consumer_is_ended_once_the_byte_bound_evicts_past_it() {
        const BYTES: usize = 8 * 1024 * 1024;
        let repos = Repos::new(
            Arc::new(Sequence::in_memory()),
            ReplayBounds {
                frames: 4096,
                bytes: BYTES,
            },
        );
        let mut stalled = repos
            .subscribe(None, Exclusions::default())
            .expect("a live consumer");
        for _ in 0..16 {
            repos
                .publish_commit(&largest_commit())
                .expect("the frame was numbered");
            assert!(
                repos.ring().bytes <= 2 * BYTES,
                "a stream's unsent frames held the buffer past twice its bound"
            );
        }
        assert!(
            repos.ring().bytes <= BYTES,
            "a stream that had already lost a frame kept the buffer past its bound"
        );
        assert_eq!(
            stalled.next().await,
            None,
            "a consumer the buffer evicted past was handed a frame with a hole in front of it"
        );
        assert!(stalled.lagged() > 0);
    }

    /// The review probe for a resume at the eviction edge. The buffer is
    /// full, the cursor is inside it, and one more frame lands before the
    /// stream reads. The stream must still send every frame after its
    /// cursor: its unsent frames are kept for it.
    #[tokio::test]
    async fn a_cursor_at_the_eviction_edge_replays_everything_after_it() {
        let repos = Repos::new(Arc::new(Sequence::in_memory()), ReplayBounds::frames(4));
        for _ in 0..4 {
            repos
                .publish_commit(&commit("3lqaaa", None))
                .expect("the frame was numbered");
        }
        let mut stream = repos
            .subscribe(Some(0), Exclusions::default())
            .expect("a cursor inside the buffer");
        repos
            .publish_commit(&commit("3lqbbb", None))
            .expect("the frame was numbered");

        let mut seqs = Vec::new();
        for _ in 0..5 {
            let frame = stream
                .next()
                .await
                .expect("a stream with a valid cursor was ended");
            seqs.push(seq_of(&frame));
        }
        assert_eq!(seqs, vec![1, 2, 3, 4, 5]);
        assert_eq!(stream.lagged(), 0);

        // Once sent, the extra frame is let go at the next publish.
        repos
            .publish_commit(&commit("3lqccc", None))
            .expect("the frame was numbered");
        assert_eq!(repos.all().len(), 4, "the bound applies again");
    }

    /// An outdated cursor starts at the oldest frame held, which is the
    /// eviction edge. It must get room to catch up, not be cut off by the
    /// next write and loop between `OutdatedCursor` and `ConsumerTooSlow`.
    #[tokio::test]
    async fn an_outdated_stream_starting_at_the_oldest_frame_is_not_cut_off() {
        let repos = Repos::new(Arc::new(Sequence::in_memory()), ReplayBounds::frames(4));
        for _ in 0..6 {
            repos
                .publish_commit(&commit("3lqaaa", None))
                .expect("the frame was numbered");
        }
        let mut stream = repos
            .subscribe(Some(0), Exclusions::default())
            .expect("not a future cursor");
        for _ in 0..3 {
            repos
                .publish_commit(&commit("3lqbbb", None))
                .expect("the frame was numbered");
        }
        let (header, _) = decode(&stream.next().await.expect("an info frame"));
        assert_eq!(field(&header, "t"), &Value::String("#info".to_owned()));
        let mut seqs = Vec::new();
        for _ in 0..7 {
            seqs.push(seq_of(&stream.next().await.expect("a frame")));
        }
        assert_eq!(seqs, (3..=9).collect::<Vec<u64>>());
    }

    /// The edge of falling behind: the buffer evicting the very frame a
    /// stream last sent costs that stream nothing.
    #[tokio::test]
    async fn a_stream_survives_the_eviction_of_a_frame_it_already_sent() {
        let size = Repos::default()
            .publish_commit(&commit("3lqaaa", None))
            .expect("the frame was numbered")
            .bytes
            .len();
        let repos = Repos::new(
            Arc::new(Sequence::in_memory()),
            ReplayBounds {
                frames: 4096,
                bytes: 2 * size,
            },
        );
        let mut stream = repos
            .subscribe(None, Exclusions::default())
            .expect("a live consumer");
        repos
            .publish_commit(&commit("3lqaaa", None))
            .expect("the frame was numbered");
        assert_eq!(seq_of(&stream.next().await.expect("frame 1")), 1);

        for _ in 0..2 {
            repos
                .publish_commit(&commit("3lqaaa", None))
                .expect("the frame was numbered");
        }
        assert_eq!(repos.oldest(), 2, "the byte bound evicted frame 1");
        assert_eq!(repos.ring().evicted, 1);

        assert_eq!(seq_of(&stream.next().await.expect("frame 2")), 2);
        assert_eq!(seq_of(&stream.next().await.expect("frame 3")), 3);
        assert_eq!(stream.lagged(), 0);
    }
}