offline_protocol/protocol/

data_sync.rs

//! Replicating documents between two peers over the ordinary message ladder.
//!
//! # Why this is small
//!
//! Almost everything replication needs is already here. Deltas are
//! idempotent and commutative, so at-least-once, unordered, partition
//! tolerant delivery is not a compromise this layer works around: it is
//! exactly enough, by construction. That means no ordering, no sequence
//! numbers, no session streams, and no second delivery path. A sync frame is
//! a message, and every reliability property the message ladder has spent
//! ten releases acquiring becomes a sync property for free.
//!
//! # The space is the session
//!
//! A space replicates with the peer whose address names it. Inbound frames
//! take the space from the authenticated wire sender, outbound frames from
//! the recipient, and neither ever reads a space name off the wire. So a
//! frame from one peer cannot touch another peer's documents, and there is
//! no authorization table to get wrong. The two replicas name the same space
//! differently — each by the *other's* address — which is why the name is
//! absent from the frames.
//!
//! # Anti-entropy, not a protocol
//!
//! On session confirm, on reconnect, and at startup, each side offers its
//! version of every document it holds for that peer. The other side answers
//! with whatever is missing and its own versions, and the exchange stops.
//! Nothing persists about how far a peer has got: state that has to be
//! reconciled after a crash is state that can be wrong, and the version
//! exchange already answers the only question that matters.
//!
//! # Every leg ends
//!
//! An inbound frame produces at most one kind of outbound answer, and every
//! chain of answers is finite: an offer provokes one reply, a delta that
//! cannot apply provokes one request, and the snapshot that request is
//! answered with provokes nothing at all. The one chain longer than a single
//! hop is a reply naming a document this device has never seen: it is asked
//! for, and the answer is the document. That ends too, because the question
//! names only documents the peer has just offered, so the peer creates
//! nothing from it and has nothing further to ask. That is the property to
//! preserve when adding a frame kind. Two replicas that answer each other's
//! answers converge perfectly well and then talk until the battery dies, and
//! the failure has no symptom on either device except traffic.
//!
//! A removal carried by an offer does not lengthen any of this. It is a
//! field on a frame that was already answered exactly once, and what it
//! provokes is a deletion rather than a frame: the answer to an offer
//! carrying removals is the same single reply an offer without them draws.
//! The one thing it must not do is revive what it just removed, which is why
//! removals are applied before anything is created or asked for.
//!
//! One outbound frame is not an answer and does not count against this: a
//! blob arriving for a document with local edits still pending flushes them
//! first, which pushes them. That is work this device already owed the peer
//! and the frame merely reached it, so it cannot recur: what it sends is
//! what was pending before the frame arrived, and flushing is what stops it
//! being pending. When that flush fails and the import then applies, the
//! edit may have been folded into the imported change and suppressed with
//! it, so a version offer is sent instead of the delta. That one is not an
//! answer either, and it cannot recur for the same reason: it costs a
//! storage failure that recovered inside a single frame.
//!
//! # Remote bytes are not our bytes
//!
//! Everywhere else in the data layer, the argument for handing bytes to the
//! document engine is that they came back out of a sealed record whose AEAD
//! tag already vouched for them. That argument does not survive contact with
//! a network. MLS says who sent a blob; it says nothing about its shape, and
//! the engine has an open defect where a malformed one aborts the process
//! rather than returning an error. So blobs arriving here are judged before
//! the engine sees them, and the one that gets through is remembered on disk
//! until it has survived, so a blob that kills the process kills it once
//! rather than every time the sender retries.

use std::collections::{BTreeMap, BTreeSet};
use std::time::{Duration, Instant};

use base64::{engine::general_purpose::STANDARD as BASE64, Engine as _};
use offline_protocol_data::{CatchUp, RemoteImport, VersionToken};
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use tracing::{debug, warn};

use crate::error::{Error, Result};
use crate::events::Event;
use crate::media_envelope::DataPurpose;
use crate::protocol::prefixes::internal_prefixes;
use crate::protocol::types::{storage_keys, MediaSendOptions, DATA_SYNC_V1};
use crate::protocol::OfflineProtocol;
use offline_protocol_core::ContentType;

/// Largest document blob carried inside one sync frame, before base64.
///
/// Sized to the mesh rather than to the record store. An unnegotiated
/// Bluetooth link fragments a message into at most 512 pieces of 139 usable
/// bytes, so roughly 69 KiB is the hard ceiling for a single message on the
/// worst transport we ship; 32 KiB leaves room for base64 expansion, the
/// frame's own JSON, the sealed envelope and the message header, and still
/// arrives in one piece. Anything larger is not refused forever, it is
/// answered by a snapshot offer that the media path will carry.
pub(crate) const MAX_SYNC_BLOB_BYTES: usize = 32 * 1024;

/// Documents listed in one version frame.
///
/// A name is capped at 128 bytes and an encoded version is short, so this
/// keeps a version frame comfortably inside [`MAX_SYNC_BLOB_BYTES`] without
/// having to measure it. A space with more documents than this sends
/// several frames, which converges identically: the exchange is per
/// document, not per frame.
///
/// The declared interest rides every frame and is not weighed here, because
/// it is bounded where it is set: [`MAX_INTEREST_PATTERNS`] names of at most
/// [`MAX_NAME_LEN`] bytes is about 4 KiB, which this budget leaves free. A
/// wider bound on either would have to be weighed rather than assumed.
///
/// [`MAX_INTEREST_PATTERNS`]: super::data::MAX_INTEREST_PATTERNS
/// [`MAX_NAME_LEN`]: offline_protocol_data::MAX_NAME_LEN
const MAX_DOCS_PER_VERSION_FRAME: usize = 128;

/// How long an offer to one peer suppresses the next one.
///
/// Sized to discovery flapping rather than to editing: a change made locally
/// is pushed the moment it is durable and does not wait for this, so the only
/// thing the window delays is the reconciliation sweep, and the next trigger
/// repeats it anyway.
pub(crate) const DATA_SYNC_OFFER_INTERVAL: Duration = Duration::from_secs(30);

/// How long local commits to a space must stay quiet before the committing
/// device sends one follow-up version offer for what it committed: the
/// settle offer.
///
/// A delta that parks behind a lost predecessor is noticed when the *next*
/// delta arrives, so a loss in the middle of a burst of edits heals at once.
/// The last delta of a burst has no successor. What recovers it otherwise is
/// the delivery ladder: every sync frame, 1:1 or group (fanned out per
/// member), is acknowledged, and an unacknowledged one is re-sent after the
/// ACK timeout (10 s by default) and its backoff, until the retry budget is
/// spent. The settle offer asks after [`DATA_SYNC_SETTLE_DELAY`] instead, and
/// still asks once that budget is gone. Congestion is what loses deltas and it
/// strikes during bursts, so the tail is not a corner case.
///
/// Debounced rather than sent per commit: a commit inside the window pushes
/// the deadline out, so a burst costs one offer at the end of it, not one per
/// flush. Three seconds is longer than the gap between flushes of someone
/// still typing on the recommended throttle, so it fires at a pause rather
/// than mid-sentence, and short enough that a lost tail is back before the
/// reader has looked away. It also leaves the burst ahead of it time to drain
/// a slow link, since the offer queues behind it; one that overtakes costs a
/// duplicate delta, which the merge absorbs.
pub(crate) const DATA_SYNC_SETTLE_DELAY: Duration = Duration::from_secs(3);

/// Documents one space may hold on a peer's say-so.
///
/// Every name in an offer that this device does not recognise becomes a
/// stored document, and nothing else bounds how many names one exchange can
/// carry. The ceiling is far above what two people sharing documents reach
/// and far below what a peer streaming fresh names could otherwise spend, so
/// it is a bound on abuse rather than a product limit. It applies only to
/// documents a peer names: an application creating its own is not restricted
/// by it, and those still replicate outward.
pub(crate) const MAX_DOCS_PER_SPACE: usize = 1024;

/// Bytes a removal floor may occupy, on the wire and in the record.
///
/// A floor is one version vector, and the floor held for a name is the
/// union of every floor ever merged into it, which only grows: a member of
/// a space can grow it by naming a fresh replica in every offer, and nothing
/// but this stops the record and every comparison against it growing with
/// it. Sized well past a real document: the version of a document edited
/// across thousands of sessions is still under this, and a document whose
/// version could not fit here could not be offered in a frame either.
pub(crate) const MAX_REMOVAL_FLOOR_BYTES: usize = 64 * 1024;

/// Blob digests remembered per space after a crash.
///
/// Small on purpose. The list only has to outlive the sender's retries of
/// one blob, and every entry in it is a document change we have decided
/// never to apply, so a long list is a liability rather than a safety net.
pub(crate) const MAX_QUARANTINED_BLOBS: usize = 32;

/// Largest document encoding this device will carry over the media path.
///
/// The record ceiling, because that is the real limit: a document larger
/// than one sealed protocol-state record cannot be persisted by the receiver
/// even if every byte arrives, so carrying it would spend a long transfer to
/// fail at the end. Well above the 1 MiB compacted cap a document is held
/// to, which leaves room for the history a raw snapshot carries.
pub(crate) const MAX_MEDIA_SNAPSHOT_BYTES: usize = super::types::MAX_PROTOCOL_STATE_RECORD_BYTES;

/// Attachment fetches one device may have outstanding at once.
///
/// Bounded because each entry is created by a local call and cleared by a
/// remote answer that may never come. Passing the bound evicts the oldest:
/// a fetch that has been waiting longer than sixty-three others is not the
/// one still worth remembering.
pub(crate) const MAX_PENDING_ATTACHMENT_FETCHES: usize = 64;

/// How long a fetch stays outstanding with no sign of an answer.
///
/// Measured from the last sign of life rather than from the question, and
/// the difference is the whole of it: chunks of the answer refresh the
/// entry, so a blob slower to arrive than this does not have its own record
/// expire underneath it. That record is what admits the bytes at the end, so
/// an expiry mid-carriage discards a blob that fully arrived, and the retry
/// it invites takes just as long and dies exactly the same way.
///
/// Generous even so, because a holder may take a while to answer at all and
/// the media path may be Bluetooth. The cost of being slow here is a
/// reference that reports unavailable while somebody is still deciding, and
/// the app can ask again; the cost of no timeout at all is a map that only
/// grows.
pub(crate) const ATTACHMENT_FETCH_TIMEOUT: Duration = Duration::from_secs(15 * 60);

/// Blob requests one peer may have acted on inside a window.
///
/// A budget rather than only a per-hash window, because the two bound
/// different things and only one of them is a hash we have seen before. The
/// per-hash window stops a peer re-asking for the same blob; nothing in it
/// stops a peer asking for a different one every time, and each of those
/// costs a durable map entry and an application callback. A hash is 32 bytes
/// the sender chooses, so without a budget the key space is the whole output
/// of SHA-256 and the cost of exhausting this device's memory is one frame
/// per entry.
pub(crate) const MAX_BLOB_REQUESTS_PER_WINDOW: usize = 32;

/// Blob-request windows remembered across all peers.
///
/// The hard bound under the per-peer budget above, and the one that holds
/// when a peer's budget resets. Oldest out first: a window that has been
/// held longer than 255 others has almost certainly expired anyway.
pub(crate) const MAX_BLOB_REQUEST_WINDOWS: usize = 256;

/// One sync frame's body, inside the decrypted MLS plaintext.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "k")]
enum SyncBody {
    /// "Here is what I have; send me what I am missing."
    ///
    /// `reply` marks the answering leg, which is what stops the exchange:
    /// an offer provokes an answer, and an answer provokes nothing.
    #[serde(rename = "vv")]
    Versions {
        #[serde(default)]
        reply: bool,
        /// Whether this frame carries less than everything the sender holds:
        /// one frame of an offer split across several, or a single document
        /// named because something about it needs answering.
        ///
        /// It exists because a receiver reads a document's *absence* from an
        /// offer as "they have never seen this", and answers with the whole
        /// document. That inference needs the complete list. Drawn from one
        /// frame of a split offer it fires on every document the frame did
        /// not happen to carry, so a peer holding more documents than one
        /// frame fits would be sent the entire space, in full, on every
        /// exchange, while perfectly in sync.
        #[serde(default)]
        partial: bool,
        /// Document name to base64 of its version token.
        docs: BTreeMap<String, String>,
        /// Document name to base64 of the version it was removed at.
        ///
        /// A removal floor rather than a bare "this is gone": it names the
        /// content the removal covered, so a replica holding an edit beyond
        /// it keeps its copy and hands the document back, while one holding
        /// nothing newer deletes it. That is what makes a removal and a
        /// concurrent edit resolve the same way on every replica without
        /// either side knowing which happened first.
        ///
        /// Absent means "no removals to report", which is what every frame
        /// from a build that predates them says, and what a space nobody
        /// has removed anything from says today. Entries are carried for
        /// names that came back as well as names that are still gone: a
        /// third replica that never heard about the removal is holding
        /// exactly the content the floor names.
        #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
        gone: BTreeMap<String, String>,
        /// Which documents the sender wants back, as names or `name*`
        /// prefixes.
        ///
        /// `Option` rather than a plain list because absent and empty mean
        /// opposite things. Absent is "everything", which is what a sender
        /// that has never narrowed says and what every build that predates
        /// the field says; empty is "nothing", which an application can ask
        /// for and must be able to express. Collapsing the two would make
        /// the narrowest possible request read as the widest.
        ///
        /// It scopes what the *receiver* sends back, and nothing else. It is
        /// a request, not a permission: the sender refuses what it does not
        /// want on arrival regardless, which is what makes it safe toward a
        /// peer that ignores the field.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        want: Option<Vec<String>>,
    },
    /// A run of changes, base64.
    #[serde(rename = "delta")]
    Delta { doc: String, blob: String },
    /// A whole document, base64. The answer to a gap no run of changes can
    /// express.
    #[serde(rename = "snap")]
    Snapshot { doc: String, blob: String },
    /// "No run of changes can close my gap in this document; send the whole
    /// thing."
    ///
    /// The frame that makes a refusal recoverable. Without it a receiver
    /// that declines a delta for reaching below its trim point has no way to
    /// say what would work: answering with a version offer instead asks the
    /// sender to compute changes since our version, which is the same
    /// refused delta again, for as long as both sides keep at it.
    #[serde(rename = "need_snap")]
    NeedSnapshot { doc: String },
    /// "I hold a reference to these bytes and not the bytes; do you have
    /// them?"
    ///
    /// Carries no document name. A blob is addressed by what it is rather
    /// than by where a reference to it happens to sit, so the same bytes
    /// referenced from three documents are one blob and one request.
    #[serde(rename = "need_blob")]
    NeedBlob {
        hash: String,
        /// The member this question is for, inside a group.
        ///
        /// Absent on a 1:1 session, where the recipient is the only peer
        /// there is. In a group it is load-bearing rather than a
        /// convenience: a directed frame is promoted to a roster-wide
        /// delivery whenever the sender's ratchet budget runs out, and a
        /// receiver cannot tell a promoted frame from one addressed to it.
        /// Without this field every member's application would be asked for
        /// bytes the asker can only accept from one of them.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        to: Option<String>,
    },
    /// "I do not have those bytes, and asking again will not change that."
    ///
    /// The frame that lets a fetch end. An attachment reference outlives the
    /// bytes it names, because the reference replicates and the bytes do
    /// not, so a peer holding a reference and no blob is ordinary rather
    /// than broken. Without an answer for it the asking side cannot tell
    /// that case from a peer that is merely slow, and shows a spinner
    /// forever.
    #[serde(rename = "blob_gone")]
    BlobGone { hash: String },
    /// One piece of a blob, travelling as a frame rather than as a
    /// transfer.
    ///
    /// The group answer to `need_blob`. A 1:1 fetch rides the media path,
    /// which needs a confirmed pairwise session; two members of a group need
    /// not have one with each other, and requiring it would make an
    /// attachment depend on a handshake that may never happen. Frames need
    /// only the group key, which membership already is.
    ///
    /// `i` is the index and `n` the count, so a receiver knows when it is
    /// done without the sender saying so twice. Carrying the count on every
    /// chunk rather than on the first is what makes the pieces independent:
    /// they may arrive in any order, and a duplicate replaces rather than
    /// appends.
    ///
    /// Provokes nothing, which is the property that lets `need_blob` stay a
    /// single-hop question. That is what bounds a blob to
    /// [`MAX_GROUP_BLOB_CHUNKS`] pieces: a larger one would need the
    /// requester to ask again for each window, and a chain of questions is
    /// exactly what this layer does not have.
    #[serde(rename = "chunk")]
    Chunk {
        hash: String,
        i: u32,
        n: u32,
        blob: String,
    },
}

/// One name in a version offer: its version when we hold it, and its removal
/// floor when it has one. A name can carry both, because a document that came
/// back after a removal is offered *and* still reports the floor, for the
/// sake of a third replica that never heard about the removal.
type OfferEntry = (String, Option<String>, Option<String>);

/// Split an offer into frames, each within [`MAX_DOCS_PER_VERSION_FRAME`].
///
/// A name carrying both a version and a floor counts twice: the budget
/// stands in for bytes, and such a name costs the frame two encoded
/// versions and its own name twice. Counting it once would let a space
/// whose every document came back after a removal build a frame twice the
/// size the budget was chosen to keep inside the transport.
///
/// An empty offer is still one frame. It is what tells a peer holding
/// documents to send them to a replica that has never seen any.
fn offer_batches(entries: &[OfferEntry]) -> Vec<&[OfferEntry]> {
    let mut batches = Vec::new();
    let mut start = 0usize;
    let mut weight = 0usize;
    for (index, (_, version, floor)) in entries.iter().enumerate() {
        let cost = usize::from(version.is_some()) + usize::from(floor.is_some());
        if index > start && weight + cost > MAX_DOCS_PER_VERSION_FRAME {
            batches.push(&entries[start..index]);
            start = index;
            weight = 0;
        }
        weight += cost;
    }
    batches.push(&entries[start..]);
    batches
}

/// One fetch this device is waiting on.
///
/// The record both bounds the wait and admits the answer, which is why its
/// clock is the last sign of life rather than the moment of the question: a
/// record that expires mid-carriage discards a blob that fully arrived, and
/// the retry it invites dies at the same point.
#[derive(Debug)]
pub(crate) struct PendingFetch {
    /// The last sign that an answer is coming.
    pub(crate) last_seen: Instant,
    /// The member being asked, when the space is a group.
    ///
    /// `None` on a 1:1 space, where the peer *is* the space and there is
    /// nothing to choose. In a group there is: the reference names bytes and
    /// not who holds them, so the application says which member to ask.
    pub(crate) holder: Option<String>,
    /// Chunks accepted so far, when the bytes travel as frames.
    ///
    /// Only a group fetch assembles here. A 1:1 fetch rides the media path,
    /// which does its own reassembly and hands over one finished blob.
    pub(crate) assembly: Option<BlobAssembly>,
}

impl PendingFetch {
    /// A fetch just put to a 1:1 peer.
    fn to_peer() -> Self {
        Self {
            last_seen: Instant::now(),
            holder: None,
            assembly: None,
        }
    }

    /// A fetch just put to one member of a group.
    fn to_member(member: &str) -> Self {
        Self {
            last_seen: Instant::now(),
            holder: Some(member.to_string()),
            assembly: None,
        }
    }

    /// Whether `sender` is who this fetch is waiting on.
    fn is_answered_by(&self, space: &str, sender: &str) -> bool {
        match &self.holder {
            Some(member) => member == sender,
            None => space == sender,
        }
    }
}

/// A blob being reassembled from frames.
///
/// What it can hold is bounded by its own two rules rather than by a
/// running total: at most [`MAX_GROUP_BLOB_CHUNKS`] chunks, since an index
/// outside the declared count is refused and the count itself is capped,
/// each at most [`MAX_SYNC_BLOB_BYTES`]. One assembly is therefore 1 MiB at
/// the very most, and the fetch bound puts a ceiling on how many exist.
#[derive(Debug, Default)]
pub(crate) struct BlobAssembly {
    /// How many chunks the first arrival said there would be.
    total: u32,
    /// Chunks by index, so a duplicate replaces rather than appends and the
    /// order they arrive in does not matter.
    chunks: BTreeMap<u32, Vec<u8>>,
}

/// What one version offer says.
///
/// The three fields travel together in both directions and mean the same
/// thing in both: what the sender holds, what it removed, and what it wants
/// back. Grouping them is not tidiness: passed separately they are three
/// arguments a caller can transpose, and two of them have the same type.
#[derive(Debug, Default)]
struct OfferContents {
    /// Document name to base64 of the version held.
    docs: BTreeMap<String, String>,
    /// Document name to base64 of the version it was removed at.
    gone: BTreeMap<String, String>,
    /// Names and `name*` prefixes the sender wants back, or `None` for
    /// everything.
    want: Option<Vec<String>>,
}

/// How many interest patterns one frame is read from.
///
/// A peer chooses these, and every one of them is walked per document when
/// the answer is built, so the bound is the receiver's rather than the
/// sender's: a sender's own list is bounded where it is set.
const MAX_INTEREST_PATTERNS_PER_FRAME: usize = 32;

/// The two bounds are the same number for a reason, and the reason is not
/// coincidence: a pattern an application is allowed to declare has to be one
/// a conforming peer actually reads. Raised past this on the sending side
/// alone, the tail of every declared list would be dropped by every peer
/// with no symptom on either device, and the frame-budget note on
/// [`MAX_DOCS_PER_VERSION_FRAME`] would be reasoning about a list longer than
/// it weighs.
const _: () = assert!(
    super::data::MAX_INTEREST_PATTERNS <= MAX_INTEREST_PATTERNS_PER_FRAME,
    "a pattern an application may declare must be one every peer reads"
);

/// How many frames one blob may be carried in inside a group.
///
/// The bound that keeps `need_blob` a single-hop question: the whole answer
/// leaves at once, so nothing has to ask for the next window and no chain of
/// questions exists to terminate. At [`MAX_SYNC_BLOB_BYTES`] a piece, this
/// puts a group attachment at 1 MiB.
///
/// That is far below what the 1:1 path carries, and deliberately so. Moving
/// a hundred megabytes this way would mean flow control, a holder keeping
/// the bytes between windows, and a re-ask clock, which is the machinery the
/// media path already has and that a frame path would have to grow. A group
/// attachment is a receipt, a signature, a small photo; anything larger
/// travels 1:1, and the sender is told which it is at the call.
pub(crate) const MAX_GROUP_BLOB_CHUNKS: u32 = 32;

/// Where a sync frame goes and what it is sealed under.
///
/// The one thing F4 added to this module. 1:1 replication could treat the
/// space, the peer, and the reply address as a single value because they
/// genuinely were one; a group space is one scope with many members, so the
/// space no longer names the recipient and the two have to be carried
/// separately.
#[derive(Debug, Clone)]
pub(crate) enum SyncChannel {
    /// A 1:1 space, sealed to the peer the space is named after.
    Peer,
    /// A group space, sealed once for the whole roster.
    ///
    /// Used only for changes every member needs: a local commit. Group MLS
    /// produces one ciphertext for everyone, which is what makes group
    /// replication cheap, and it is also why this must never be used for an
    /// answer to one member's question — the other members would each
    /// receive, decrypt, and import a blob they already had.
    GroupBroadcast,
    /// A group space, sealed for the roster but delivered to one member.
    ///
    /// The reply leg. Anti-entropy between two members of a group is still
    /// a conversation between two devices: an offer is answered by the
    /// member that received it and by nobody else. Every other member is
    /// spared the traffic, and the terminating 1.5-round-trip exchange the
    /// 1:1 path relies on keeps working unchanged.
    ///
    /// Sealed under the group key rather than the pair's 1:1 session on
    /// purpose: two members of a group need not have a 1:1 session at all,
    /// and requiring one would make replication depend on a handshake that
    /// may never happen.
    GroupDirected(String),
}

/// Which frame a blob arrived in.
///
/// Carried into the import so a rung of the catch-up ladder knows whether
/// there is a rung above it. A delta that cannot apply has one; a snapshot
/// is already the top, and asking again returns the same bytes.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum BlobKind {
    Delta,
    Snapshot,
}

/// Per-space replication bookkeeping.
///
/// Everything here is about surviving a crash. Nothing in it is needed for
/// convergence, which is deliberate: the version exchange is the recovery
/// mechanism, so this record can be lost without costing a change.
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
struct SyncRecord {
    /// The blob currently being handed to the document engine.
    ///
    /// Written before the engine is called and cleared after it returns. A
    /// value still here at open means the last process to touch this space
    /// did not survive that call.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    in_flight: Option<String>,
    /// Blobs that were in flight when a previous run died, newest last.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    quarantined: Vec<String>,
}

/// The digest a blob is remembered by.
pub(crate) fn blob_digest(blob: &[u8]) -> String {
    let mut hasher = Sha256::new();
    hasher.update(blob);
    // Half a SHA-256 is 128 bits, which is far past what a collision would
    // need to be worth engineering: the prize is having one specific blob
    // refused on one device.
    hex_encode(&hasher.finalize()[..16])
}

/// The rate-limit key for offers toward one member of one group.
///
/// A group space and a 1:1 space share one window map, so the two key
/// spaces must not overlap: a bare peer address is the 1:1 key, and this
/// one is deliberately built with a character no validated peer id, group
/// id or address can contain, so no composed key can ever collide with a
/// peer's own.
fn group_offer_key(member: &str, group_id: &str) -> String {
    format!("{member}{GROUP_OFFER_KEY_SEP}{group_id}")
}

/// The separator [`group_offer_key`] composes with, named so the split that
/// takes a key back apart cannot drift from the format that built it.
const GROUP_OFFER_KEY_SEP: char = '\x01';

fn hex_encode(bytes: &[u8]) -> String {
    bytes.iter().map(|b| format!("{b:02x}")).collect()
}

impl OfflineProtocol {
    // ---- outbound ---------------------------------------------------

    /// Offer our versions of every document shared with `peer`.
    ///
    /// The entry point for all three triggers (session confirm, reconnect,
    /// startup). Cheap and safe to call spuriously: a peer that is already
    /// current answers with nothing.
    pub(crate) fn kick_data_sync(&mut self, peer: &str, cause: &str) {
        if !self.data_sync_active(peer) {
            return;
        }
        if !self.data_sync_offer_due(peer) {
            return;
        }
        self.offer_versions(peer, cause);
    }

    /// Offer our versions of every document in `group_id` to one member.
    ///
    /// The group counterpart of [`Self::kick_data_sync`], and addressed for
    /// the same reason its answers are: a member coming back into range is
    /// a conversation with that member. Broadcasting the offer to the whole
    /// roster would have every other member answer a question nobody asked
    /// them.
    ///
    /// Rate-limited per (member, group) rather than per member, so a peer
    /// shared across several groups reconciles all of them rather than
    /// whichever one its rediscovery happened to reach first.
    #[cfg_attr(not(feature = "data"), allow(dead_code))]
    pub(crate) fn kick_group_data_sync(&mut self, group_id: &str, member: &str, cause: &str) {
        let members = match self.group_roster(group_id) {
            Ok(members) => members,
            Err(_) => return,
        };
        if !self.group_data_sync_active(&members) {
            // Rediscovery is the one trigger a group space has that does
            // not require somebody to be editing, so it is also the only
            // moment a device that never writes anything will notice the
            // gate is shut. Probing only from the local-commit path would
            // leave a member that merely reads waiting for someone else to
            // do it.
            self.probe_group_data_capabilities(group_id, &members);
            return;
        }
        let key = group_offer_key(member, group_id);
        if !self.data_sync_offer_due(&key) {
            return;
        }
        self.offer_versions_over(
            group_id,
            &SyncChannel::GroupDirected(member.to_string()),
            cause,
        );
    }

    /// Ask the members holding a group's replication gate closed what they
    /// support.
    ///
    /// For a group the usual reason the gate is shut is not that somebody
    /// opted out but that we have never heard from them: members added by a
    /// third party exchange no key packages with us, and an inviter running
    /// an SDK from before attestation forwards none. Sending them ours
    /// makes their automatic reply teach us what they support.
    ///
    /// The probe itself is capability-agnostic and guarded by
    /// `key_package_sent_to`, so a group that stays closed does not re-probe
    /// on every commit or every rediscovery. A member already probed is
    /// pushed to again only by `rearm_key_package_for_peer`, on its doubling
    /// backoff, so a closed group costs at most one push per member per
    /// window.
    #[cfg_attr(not(feature = "data"), allow(dead_code))]
    fn probe_group_data_capabilities(&mut self, space: &str, members: &[String]) {
        let unknown = self.group_data_unknown_members(members);
        if unknown.is_empty() {
            return;
        }
        debug!(
            space,
            unknown = unknown.len(),
            "Group replication is held closed by members of unknown capability; probing"
        );
        self.backfill_group_rich_capabilities(&unknown);
    }

    /// Drop every offer window belonging to `peer`: the 1:1 one, and the
    /// group one for each group shared with them.
    ///
    /// The group windows are keyed by (member, group), so removing the bare
    /// peer key leaves them all behind, and each of them suppresses the
    /// first group offer to that peer for a further window once the
    /// capability is relearned. That is the same silent non-sync
    /// [`Self::forget_data_sync_peer`] exists to prevent, one space at a
    /// time.
    ///
    /// Windows left by a group this device *left* are deliberately not
    /// swept here: they name a scope that no longer replicates at all, and
    /// on a re-join the worst they can cost is one suppressed offer that
    /// the next discovery re-sends.
    #[cfg(feature = "data")]
    pub(crate) fn forget_data_sync_offer_windows(&mut self, peer: &str) {
        self.last_data_sync_offer.retain(|key, _| {
            key != peer
                && !key
                    .split_once(GROUP_OFFER_KEY_SEP)
                    .is_some_and(|(member, _)| member == peer)
        });
    }

    /// Whether the reconciliation sweep keyed by `key` is outside its
    /// suppression window, stamping it when it is.
    ///
    /// Stamped here rather than after the send for the reason
    /// [`Self::offer_versions_over`] gives: a read that fails will still
    /// fail a moment later, and retrying it at discovery speed is the
    /// traffic the window exists to prevent.
    fn data_sync_offer_due(&mut self, key: &str) -> bool {
        if let Some(last) = self.last_data_sync_offer.get(key) {
            if Instant::now().duration_since(*last) < DATA_SYNC_OFFER_INTERVAL {
                return false;
            }
        }
        self.last_data_sync_offer
            .insert(key.to_string(), Instant::now());
        true
    }

    /// Offer versions of every group space this device shares with `peer`.
    ///
    /// Wired to the same rediscovery seam as [`Self::kick_data_sync`], and
    /// needed because a group space has no 1:1 trigger to ride: two members
    /// may never establish a session with each other, so without this the
    /// only thing that ever reconciles them is a fresh local commit.
    ///
    /// Which groups this device is in comes from MLS, not from the roster
    /// cache alone. The cache is filled on demand and is empty after a
    /// restart, so reading only it would make this sweep silently do nothing
    /// on a cold process, which is the launch where it matters most: two
    /// members that drifted apart and are not editing have no other trigger
    /// left.
    #[cfg_attr(not(feature = "data"), allow(dead_code))]
    pub(crate) fn kick_shared_group_data_sync(&mut self, peer: &str, cause: &str) {
        if !self.config.data.enabled {
            return;
        }
        self.enumerate_group_spaces_once();
        let shared: Vec<String> = self
            .group_mesh
            .members
            .iter()
            .filter(|(_, members)| members.iter().any(|m| m == peer))
            .map(|(group_id, _)| group_id.clone())
            .collect();
        for group_id in shared {
            self.kick_group_data_sync(&group_id, peer, cause);
        }
    }

    /// Fill the roster cache from MLS once per session, so
    /// [`Self::kick_shared_group_data_sync`] can see a group that nothing has
    /// touched yet.
    ///
    /// Costs one key listing, plus a group-state read for each group not
    /// already cached — the same read the next send to that group would have
    /// paid. Marked done even when a group read fails: the failure is per
    /// group and retrying the whole enumeration on every discovery is the
    /// traffic the offer window exists to prevent. An MLS that is not ready
    /// yet is not a failure of this kind and leaves the flag alone, so the
    /// next discovery tries again.
    fn enumerate_group_spaces_once(&mut self) {
        if self.group_spaces_enumerated {
            return;
        }
        let Ok(groups) = self.list_groups() else {
            return;
        };
        self.group_spaces_enumerated = true;
        for group_id in groups {
            if self.group_mesh.members.contains_key(&group_id) {
                continue;
            }
            if let Err(err) = self.refresh_group_members(&group_id) {
                debug!(group_id = %group_id, error = %err, "Group roster unreadable while enumerating group spaces");
            }
        }
    }

    /// Offer our versions because a change happened that no frame carried,
    /// without waiting for the rate limit.
    ///
    /// `origin` names the peer a change arrived from, so one applied from
    /// the very peer this space replicates with is not announced back.
    ///
    /// The limiter damps discovery flapping, where the offer it suppresses
    /// is a sweep the next trigger repeats anyway. Nothing repeats this one.
    /// A change too large to inline left no trace on the wire at all, so
    /// until somebody asks, both sides hold documents they believe agree;
    /// on a link that never drops there is no next trigger to ask.
    pub(crate) fn nudge_data_sync(&mut self, space: &str, origin: Option<&str>, cause: &str) {
        match self.space_channel(space, origin) {
            Some(SyncChannel::Peer) => self.offer_versions(space, cause),
            // The roster-wide leg, and the one place a group offer is not
            // addressed to one member. What went unsent was a change every
            // member needs, and there is no single member to ask: each of
            // them answers with what they are missing, and a member that is
            // already current answers with nothing.
            Some(channel @ SyncChannel::GroupBroadcast) => {
                self.offer_versions_over(space, &channel, cause)
            }
            Some(SyncChannel::GroupDirected(_)) | None => {}
        }
    }

    /// How a space replicates, or `None` when it does not.
    ///
    /// The single place the three kinds of space are told apart. A space is
    /// a group space when MLS says this device holds a group by that name,
    /// a 1:1 space when the name is a peer that advertised replication, and
    /// local-only otherwise — and "local-only" is the honest answer for a
    /// space whose peer never advertised, whose group is gone, or whose
    /// members are not all group-capable.
    ///
    /// `origin`, when set, names the peer a change was just applied from,
    /// and suppresses announcing that change back to them.
    fn space_channel(&mut self, space: &str, origin: Option<&str>) -> Option<SyncChannel> {
        if let Some(members) = self.group_space_roster(space) {
            // A change that arrived over the group reached every member at
            // once, because that is what one group ciphertext does. Pushing
            // it on would have every member push every change to every
            // other member: N times the frames for nothing, and worse as
            // the group grows. Anti-entropy still closes real gaps.
            if origin.is_some() {
                return None;
            }
            if self.group_data_sync_active(&members) {
                return Some(SyncChannel::GroupBroadcast);
            }
            self.probe_group_data_capabilities(space, &members);
            return None;
        }
        if origin == Some(space) || !self.data_sync_active(space) {
            return None;
        }
        Some(SyncChannel::Peer)
    }

    /// Send our version of every document in a space to the peer that names
    /// it, and start the window that suppresses the next sweep.
    fn offer_versions(&mut self, peer: &str, cause: &str) {
        // Stamped even on the unlimited path so an offer sent right now
        // still suppresses the next sweep: the sweep would re-send what
        // this frame already carried.
        self.last_data_sync_offer
            .insert(peer.to_string(), Instant::now());
        self.offer_versions_over(peer, &SyncChannel::Peer, cause);
    }

    /// [`Self::offer_versions`] over an explicit channel, which is what a
    /// group space needs: the offer is addressed to one member, not
    /// broadcast to the roster.
    fn offer_versions_over(&mut self, space: &str, channel: &SyncChannel, cause: &str) {
        let peer = space;
        let docs = match self.data_sync_versions(peer) {
            Ok(docs) => docs,
            Err(err) => {
                // A whole space that cannot enumerate its documents stops
                // replicating with this peer until the storage recovers, and
                // nothing else reports it. The rate limit above is stamped
                // before this read precisely so warning here cannot become
                // discovery-speed noise.
                warn!(peer = %peer, cause, error = %err, "Could not read document versions");
                return;
            }
        };
        let gone = self.tombstones_for(peer, channel);
        let want = self.interest_for(peer, channel);
        debug!(space = %peer, cause, docs = docs.len(), gone = gone.len(), "Offering document versions");
        self.send_version_frames(
            peer,
            channel,
            OfferContents { docs, gone, want },
            false,
            false,
        );
    }

    /// The removals to carry toward one channel.
    ///
    /// Empty toward a peer that has not advertised them, which is not a
    /// downgrade but the whole compatibility story: such a peer ignores the
    /// field, keeps its copy and re-offers it forever, and this device
    /// refuses each offer against its own floor. Leaving the field out
    /// keeps that exchange as small as it can be.
    ///
    /// A group is different, and is not gated here. One ciphertext reaches
    /// the whole roster, so there is no per-member choice to make, and a
    /// member that does not know the field ignores it exactly as a 1:1 peer
    /// would. The roster-wide gate that decides whether the frame may be
    /// sent at all is the one on entry 2, which is enforced where the frame
    /// is sealed.
    fn tombstones_for(&mut self, space: &str, channel: &SyncChannel) -> BTreeMap<String, String> {
        if matches!(channel, SyncChannel::Peer) && !self.data_tombstones_active(space) {
            return BTreeMap::new();
        }
        match self.data_sync_tombstones(space) {
            Ok(gone) => gone,
            Err(err) => {
                warn!(space, error = %err, "Could not read this space's removals");
                BTreeMap::new()
            }
        }
    }

    /// The interest to declare toward one channel.
    ///
    /// `None` means everything and is what an unnarrowed space sends, so a
    /// space nobody has scoped produces the frame it always did. Gated on
    /// the peer's entry 5 for the same reason removals are gated on entry
    /// 4: a peer that cannot read it would be sent a field it drops, and
    /// this device refuses what it does not want on arrival either way.
    fn interest_for(&mut self, space: &str, channel: &SyncChannel) -> Option<Vec<String>> {
        if matches!(channel, SyncChannel::Peer) && !self.data_interest_active(space) {
            return None;
        }
        self.data_interest(space).cloned()
    }

    /// Send one version frame per batch of documents.
    ///
    /// `force_partial` marks the frames as carrying less than everything we
    /// hold even when they fit in one, which is what a question about
    /// particular documents needs: without it the peer reads the names it
    /// does not see as documents we have never seen, and answers a question
    /// about one document with the whole space.
    fn send_version_frames(
        &mut self,
        space: &str,
        channel: &SyncChannel,
        offer: OfferContents,
        reply: bool,
        force_partial: bool,
    ) {
        let OfferContents { docs, gone, want } = offer;
        // A space with no documents still says so on the offering leg: that
        // is what tells a peer holding documents to send them to a replica
        // that has never seen any.
        if docs.is_empty() && gone.is_empty() && reply {
            return;
        }
        // Removals share the per-frame budget with versions, because they
        // cost the receiver the same thing: one name it has to store
        // something about. Counting only the versions would let a space
        // whose documents were all removed build one frame of a thousand
        // entries, which is the case most likely to produce a lot of them
        // at once.
        let mut entries: Vec<OfferEntry> = Vec::new();
        let mut removals = gone;
        for (doc, version) in docs {
            let floor = removals.remove(&doc);
            entries.push((doc, Some(version), floor));
        }
        for (doc, floor) in removals {
            entries.push((doc, None, Some(floor)));
        }
        entries.sort_by(|left, right| left.0.cmp(&right.0));

        let batches = offer_batches(&entries);
        // Every frame of a split offer is partial, including the last: the
        // inference that needs the complete list cannot be drawn from any
        // one of them, and there is nowhere to accumulate them that a
        // restart would not have to reconcile.
        let partial = force_partial || batches.len() > 1;
        let bodies: Vec<SyncBody> = batches
            .into_iter()
            .map(|batch| {
                let mut batch_docs = BTreeMap::new();
                let mut batch_gone = BTreeMap::new();
                for (doc, version, floor) in batch {
                    if let Some(version) = version {
                        batch_docs.insert(doc.clone(), version.clone());
                    }
                    if let Some(floor) = floor {
                        batch_gone.insert(doc.clone(), floor.clone());
                    }
                }
                SyncBody::Versions {
                    reply,
                    partial,
                    docs: batch_docs,
                    gone: batch_gone,
                    want: want.clone(),
                }
            })
            .collect();
        for body in bodies {
            self.send_sync_frame(space, channel, &body);
        }
    }

    /// Ask the peer for what we are missing in one document, without
    /// starting an exchange.
    ///
    /// Both flags are load-bearing. `reply` collects the catch-up while
    /// suppressing the counter-offer, so a request cannot start a chain of
    /// its own; `partial` keeps the peer from reading one name as the
    /// complete list and answering with every other document in the space.
    fn request_doc_catch_up(&mut self, space: &str, channel: &SyncChannel, doc: &str) {
        let version = match self.data_doc_version(space, doc) {
            Ok(version) => version,
            Err(err) => {
                warn!(space, doc, error = %err, "Could not read our version to ask for the gap");
                return;
            }
        };
        // No interest declared on a question about one document: the peer
        // answers with that document and draws no inference from the rest of
        // a list this frame does not carry.
        self.send_version_frames(
            space,
            channel,
            OfferContents {
                docs: BTreeMap::from([(doc.to_string(), version)]),
                ..OfferContents::default()
            },
            true,
            true,
        );
    }

    /// Seal one frame and put it on the ladder.
    ///
    /// The channel decides what "seal" means: a 1:1 space seals to its
    /// peer, a group space seals once under the group key. Everything
    /// above this function is written in terms of a space and a channel and
    /// does not know which of the two it is driving, which is what keeps
    /// one implementation of the exchange, the catch-up ladder and the
    /// import containment serving both.
    fn send_sync_frame(&mut self, space: &str, channel: &SyncChannel, body: &SyncBody) {
        let plaintext = match serde_json::to_string(body) {
            Ok(json) => format!(
                "{}{{\"v\":{},{}",
                internal_prefixes::DATA_V1,
                // The frame schema is the 1:1 version in both cases. The
                // group entry advertises that a peer *intercepts* these
                // frames inside a group ciphertext; it does not describe a
                // second frame shape, and minting one would mean two
                // parsers for identical bodies.
                DATA_SYNC_V1,
                // The body serializes as an object; splice the version in as
                // its first field rather than nesting, so a future version
                // can be read off a frame whose body shape it cannot parse.
                &json[1..]
            ),
            Err(err) => {
                warn!(space, error = %err, "Failed to encode sync frame");
                return;
            }
        };

        match channel {
            SyncChannel::Peer => self.send_sync_frame_to_peer(space, plaintext),
            SyncChannel::GroupBroadcast => {
                if let Err(err) = self.send_group_internal_frame(space, None, &plaintext) {
                    debug!(space, error = %err, "Group replication frame not sent");
                }
            }
            SyncChannel::GroupDirected(member) => {
                let member = member.clone();
                if let Err(err) = self.send_group_internal_frame(space, Some(&member), &plaintext) {
                    debug!(space, error = %err, "Directed replication frame not sent");
                }
            }
        }
    }

    /// The 1:1 leg: seal to the peer the space is named after.
    ///
    /// Deliberately the strict encryptor: a sync frame never initiates
    /// session establishment. Documents converge when the peers next talk,
    /// and provoking a handshake for an offer nobody asked for would make
    /// every reconnect noisier than the messaging it rides on.
    fn send_sync_frame_to_peer(&mut self, peer: &str, plaintext: String) {
        let encrypted = match self.encrypt_content_for_recipient_strict(peer, &plaintext) {
            Ok(content) => content,
            Err(err) => {
                // No session yet: nothing is lost. The version exchange runs
                // again on the next confirm, which is exactly the event that
                // would make this succeed.
                debug!(peer = %peer, error = %err, "Sync frame skipped: no session");
                return;
            }
        };

        let message = match self.create_message(peer, encrypted, None, None) {
            Ok(message) => message,
            Err(err) => {
                warn!(peer = %peer, error = %err, "Failed to build sync frame");
                return;
            }
        };

        if self.deduplicator.is_duplicate(&message.id) {
            return;
        }
        self.deduplicator.mark_seen_local(message.id.clone());

        // Tier 2: keep the plaintext so a resend after a re-key seals against
        // the peer's current epoch instead of replaying ciphertext they can
        // no longer open. The analogous session-confirm sender does not do
        // this and spends its whole retry budget on dead bytes; there is no
        // reason to inherit that. Staged after the duplicate check so the
        // entry belongs to a frame that is actually on the ladder.
        let recipient = message.recipient.as_str().to_string();
        self.stage_outbox_reseal(
            &message.id,
            &recipient,
            crate::protocol::types::OutboxReseal {
                content: plaintext,
                priority: message.priority,
                reply_to_msg: None,
                forwarded_from: None,
                content_type: message.content_type,
                media_metadata: None,
                rich: None,
                seal_mark: None,
            },
        );

        let previous_transport = self.transport_manager.current_transport();
        match self.transport_manager.send(&message) {
            Ok(()) => {
                let current = self.transport_manager.current_transport();
                let _ = self.handle_send_success(&message, current);
            }
            Err(err) => {
                let current = self.transport_manager.current_transport();
                let _ = self.handle_send_failure(&message, current.or(previous_transport));
                debug!(peer = %peer, error = %err, "Sync frame deferred to the outbox");
            }
        }
    }

    /// Push a freshly committed change to the peer this space replicates
    /// with.
    ///
    /// `origin` names the peer a change arrived from, when it did. Echoing a
    /// change back to whoever sent it is harmless — the merge absorbs it —
    /// but on a mesh it is a frame nobody needed, and a commit produced by
    /// applying a remote delta exports that delta again.
    pub(crate) fn push_data_delta(
        &mut self,
        space: &str,
        doc: &str,
        blob: &[u8],
        origin: Option<&str>,
    ) {
        let Some(channel) = self.space_channel(space, origin) else {
            return;
        };
        // Only a change made here: one applied from a peer is theirs to
        // settle, and settling it from here would be the echo `origin`
        // exists to stop. Armed before the size check so a change that went
        // out as a nudge is settled too; that nudge is one frame and can be
        // lost like any other.
        if origin.is_none() {
            self.arm_data_sync_settle(space, doc);
        }
        if blob.len() > MAX_SYNC_BLOB_BYTES {
            // Too big to inline, so the catch-up ladder has to fetch it: it
            // can answer with a compacted snapshot instead of raw history.
            // Offering now rather than waiting is what makes that happen. On
            // a link that never drops, "the peer's next version offer" is
            // not a time, and until it arrives both replicas believe they
            // agree.
            debug!(
                space,
                doc,
                bytes = blob.len(),
                "Change too large to push inline; offering versions instead"
            );
            self.nudge_data_sync(space, origin, "oversized_delta");
            return;
        }
        self.send_sync_frame(
            space,
            &channel,
            &SyncBody::Delta {
                doc: doc.to_string(),
                blob: BASE64.encode(blob),
            },
        );
    }

    /// Start, or push out, the settle offer for `doc` in `space`.
    ///
    /// Keyed by space, not by document, so a burst across several documents
    /// is still one offer naming all of them. See [`DATA_SYNC_SETTLE_DELAY`].
    fn arm_data_sync_settle(&mut self, space: &str, doc: &str) {
        let due = Instant::now() + DATA_SYNC_SETTLE_DELAY;
        let (at, docs) = self
            .data_sync_settle
            .entry(space.to_string())
            .or_insert_with(|| (due, BTreeSet::new()));
        *at = due;
        docs.insert(doc.to_string());
    }

    /// Send every settle offer whose space has been quiet for
    /// [`DATA_SYNC_SETTLE_DELAY`]. Driven by [`OfflineProtocol::process`],
    /// the binding's periodic tick, like every other timer in the core.
    ///
    /// The offer names only the documents committed since the last one and
    /// is marked partial, so the receiver draws no "never seen it" inference
    /// from the rest of the space. It is an ordinary offer and not a reply,
    /// because the side that is behind is the receiver, and only its
    /// counter-offer carries the version a catch-up is computed from. That
    /// counter-offer is a reply and provokes catch-up only, so the exchange
    /// is the same terminating 1.5 round trips every other offer starts, and
    /// nothing in it re-arms this: only a local commit does, and an import
    /// is never one.
    ///
    /// The offer is small; what it provokes is not. A receiver answers an
    /// offer that is not a reply with its complete version list for the
    /// space, not just the documents named, so one settle costs a whole-space
    /// counter-offer from the 1:1 peer, or from every member of a group, each
    /// an addressed frame spending a ratchet generation. That is why it is
    /// debounced per space rather than sent per commit. Scoping the
    /// counter-offer to a partial offer's names would cut it, and is a
    /// protocol change with its own termination argument.
    ///
    /// Neither consults nor stamps the 30-second offer window. Consulting it
    /// would swallow exactly the offers this exists for: the sweep that set
    /// the window most likely fired when the peer came into range, seconds
    /// before the burst. Stamping it would let an offer about a few documents
    /// suppress a sweep of the whole space. It needs no window of its own
    /// either: an entry is created only by a local commit and removed when
    /// it fires, so this sends at most one offer per space per quiet period,
    /// never more often than once per [`DATA_SYNC_SETTLE_DELAY`], and never
    /// without a commit behind it.
    ///
    /// In a group the offer goes to the whole roster rather than addressed to
    /// each member. The delta it follows was roster-wide, so any member may
    /// have missed it and there is no one member to ask; one roster-wide
    /// encryption costs the sender one ratchet generation every member
    /// observes, and resets the addressed-frame budget rather than spending
    /// it, where one addressed copy per member would spend a generation per
    /// member that the others never see. The channel is re-derived here, so
    /// a group whose replication gate has closed since the commit is sent
    /// nothing, exactly as the commit's own push would have been.
    pub(crate) fn send_due_data_settle_offers(&mut self, now: Instant) {
        let due: Vec<String> = self
            .data_sync_settle
            .iter()
            .filter(|(_, (at, _))| *at <= now)
            .map(|(space, _)| space.clone())
            .collect();
        for space in due {
            let Some((_, docs)) = self.data_sync_settle.remove(&space) else {
                continue;
            };
            let Some(channel) = self.space_channel(&space, None) else {
                continue;
            };
            let mut versions = BTreeMap::new();
            for doc in docs {
                // A document removed since its commit has a floor, which the
                // next sweep carries. Reading its version would be refused
                // (a removed name does not open), so skip it here rather
                // than log that refusal as a failure.
                if !self.data_holds_doc(&space, &doc) {
                    continue;
                }
                match self.data_doc_version(&space, &doc) {
                    Ok(version) => {
                        versions.insert(doc, version);
                    }
                    Err(err) => {
                        warn!(space, doc, error = %err, "Could not read a version to settle");
                    }
                }
            }
            if versions.is_empty() {
                continue;
            }
            debug!(
                space,
                cause = "settle",
                docs = versions.len(),
                "Local commits went quiet; offering their versions"
            );
            self.send_version_frames(
                &space,
                &channel,
                OfferContents {
                    docs: versions,
                    ..OfferContents::default()
                },
                false,
                true,
            );
        }
    }

    // ---- inbound ----------------------------------------------------

    /// Handle a `__DATA_V1__` frame that arrived sealed from `sender`.
    ///
    /// Every outcome is terminal. A frame is never deferred for a
    /// data-layer reason: deferral means "this same ciphertext will succeed
    /// once the session is ready" and nothing else, so using it for a
    /// corrupt blob or a disabled layer would spend the sender's whole retry
    /// budget on a frame that cannot ever be accepted.
    pub(crate) fn handle_data_sync_frame(&mut self, sender: &str, body: &str) {
        // The space is the sender. Never a field on the frame: a peer that
        // could name the space could write into a document it replicates
        // with somebody else.
        self.handle_data_sync_frame_over(sender, &SyncChannel::Peer, sender, body);
    }

    /// A `__DATA_V1__` frame that arrived inside a *group* ciphertext.
    ///
    /// The space is the group, and the same structural argument holds as
    /// for 1:1: the group is not named by the frame, it is the group whose
    /// key opened the ciphertext. A member of one group cannot reach
    /// another group's documents without being a member of that group too,
    /// and the decrypt already proved membership and authenticated the
    /// sender.
    ///
    /// Answers go back to `sender` alone rather than to the roster. Every
    /// other member either has what was asked for or will ask for it
    /// themselves.
    #[cfg_attr(not(feature = "data"), allow(dead_code))]
    pub(crate) fn handle_group_data_sync_frame(
        &mut self,
        group_id: &str,
        sender: &str,
        body: &str,
    ) {
        self.handle_data_sync_frame_over(
            group_id,
            &SyncChannel::GroupDirected(sender.to_string()),
            sender,
            body,
        );
    }

    fn handle_data_sync_frame_over(
        &mut self,
        space: &str,
        channel: &SyncChannel,
        sender: &str,
        body: &str,
    ) {
        if !self.config.data.enabled {
            // Parsing is never capability-gated, but applying is: the layer
            // is off, so the frame is dropped. Nothing is lost permanently —
            // whatever it carried is still on the sender, and the version
            // exchange after the layer comes back on asks for it again.
            debug!(peer = %sender, "Sync frame dropped: the data layer is off");
            return;
        }

        // The version is read before the body so a future frame shape is
        // declined on its version rather than failing to parse. Reversing
        // these two would make every future format look like corruption.
        let value: serde_json::Value = match serde_json::from_str(body) {
            Ok(value) => value,
            Err(err) => {
                warn!(peer = %sender, error = %err, "Malformed sync frame");
                return;
            }
        };
        match value.get("v").and_then(serde_json::Value::as_u64) {
            Some(v) if v == DATA_SYNC_V1 as u64 => {}
            other => {
                debug!(peer = %sender, version = ?other, "Sync frame from an unsupported version");
                return;
            }
        }
        let frame: SyncBody = match serde_json::from_value(value) {
            Ok(frame) => frame,
            Err(err) => {
                warn!(peer = %sender, error = %err, "Unreadable sync frame body");
                return;
            }
        };

        let space = space.to_string();
        let outcome = match frame {
            SyncBody::Versions {
                reply,
                partial,
                docs,
                gone,
                want,
            } => self.answer_version_offer(
                &space,
                channel,
                reply,
                partial,
                OfferContents { docs, gone, want },
            ),
            SyncBody::Delta { doc, blob } => {
                self.accept_remote_blob(&space, channel, &doc, &blob, BlobKind::Delta)
            }
            SyncBody::Snapshot { doc, blob } => {
                self.accept_remote_blob(&space, channel, &doc, &blob, BlobKind::Snapshot)
            }
            SyncBody::NeedSnapshot { doc } => self.answer_snapshot_request(&space, channel, &doc),
            SyncBody::NeedBlob { hash, to } => {
                self.answer_blob_request(&space, channel, &hash, to.as_deref())
            }
            SyncBody::BlobGone { hash } => self.report_blob_gone(&space, sender, &hash),
            // The group road only. A 1:1 fetch is answered over the media
            // path, which has its own reassembly, its own resource caps and
            // its own failure report; admitting a chunk here as well would
            // give one outcome two roads to arrive by, differing in every
            // bound they answer to and alike in nothing but the hash check
            // at the end.
            SyncBody::Chunk { .. } if matches!(channel, SyncChannel::Peer) => {
                warn!(
                    peer = %sender,
                    "A blob chunk arrived on a 1:1 session, where bytes ride the media path"
                );
                Ok(())
            }
            SyncBody::Chunk { hash, i, n, blob } => {
                self.accept_blob_chunk(&space, sender, &hash, i, n, &blob)
            }
        };
        if let Err(err) = outcome {
            warn!(peer = %sender, error = %err, "Sync frame could not be handled");
        }
    }

    /// Answer a peer's version offer with what they are missing.
    fn answer_version_offer(
        &mut self,
        space: &str,
        channel: &SyncChannel,
        reply: bool,
        partial: bool,
        offer: OfferContents,
    ) -> Result<()> {
        let OfferContents {
            docs: theirs,
            gone,
            want,
        } = offer;
        // What the asker wants back. Over-long lists are truncated rather
        // than refused: the patterns are a request, and answering the first
        // few is closer to what was asked for than answering none. The bound
        // is what stops a peer making every document in the space cost a
        // walk of a list it chose the length of.
        let wanted = want.map(|mut patterns| {
            if patterns.len() > MAX_INTEREST_PATTERNS_PER_FRAME {
                warn!(
                    space,
                    patterns = patterns.len(),
                    "A version offer declares more interest patterns than are read"
                );
                patterns.truncate(MAX_INTEREST_PATTERNS_PER_FRAME);
            }
            patterns
        });
        let asked_for = |doc: &str| match &wanted {
            None => true,
            Some(patterns) => patterns
                .iter()
                .any(|pattern| super::data::interest_matches(pattern, doc)),
        };
        // Removals first, before anything is created or asked for. A name
        // the peer has removed must not be created from this same frame and
        // then deleted again, and a version below a floor must not be
        // answered with a question about content the floor already covers.
        for (doc, encoded) in &gone {
            if offline_protocol_data::validate_name(doc).is_err() {
                warn!(space, doc, "A removal names an invalid document");
                continue;
            }
            // Bound the decode by the floor budget, the way a blob is bound
            // by the frame budget: base64 grows by a third, so the encoded
            // length refuses an oversized floor before allocating for it.
            if encoded.len() > MAX_REMOVAL_FLOOR_BYTES * 4 / 3 + 4 {
                warn!(space, doc, "Oversized removal floor in a sync frame");
                continue;
            }
            // A removal costs a stored record for a name this device may
            // never have held, so it answers to the same ceiling every
            // other name a peer introduces does. `continue` rather than
            // `break`: a name already known here always admits, and a
            // removal for a held document sorted after a refused fresh name
            // is exactly the removal a full space still has to apply.
            if !self.data_space_admits_doc(space, doc, MAX_DOCS_PER_SPACE) {
                warn!(
                    space,
                    doc,
                    cap = MAX_DOCS_PER_SPACE,
                    "Space is at its document ceiling; ignoring a removal for a new name"
                );
                continue;
            }
            let floor = match BASE64.decode(encoded) {
                Ok(bytes) => VersionToken::from_bytes(bytes),
                Err(err) => {
                    warn!(space, doc, error = %err, "Undecodable removal floor in a sync frame");
                    continue;
                }
            };
            match self.data_apply_tombstone(space, doc, &floor) {
                Ok(outcome) => {
                    debug!(space, doc, ?outcome, "Applied a removal from a peer");
                }
                Err(err) => {
                    warn!(space, doc, error = %err, "Could not apply a removal from a peer");
                }
            }
        }

        let mut ours = self.data_sync_versions(space).unwrap_or_default();

        // Documents they have and we do not. A created document is empty,
        // and the only thing that ever fills it is this device asking for
        // its contents, so what is created here has to be asked about
        // before this frame is done with: the peer has already told us
        // everything it intends to and will not volunteer it again.
        let mut created = BTreeMap::new();
        for (doc, theirs_encoded) in &theirs {
            if ours.contains_key(doc) || offline_protocol_data::validate_name(doc).is_err() {
                continue;
            }
            // A version a removal already covers is not a document this
            // device has never seen: it is the one it removed. Creating it
            // would undo the removal, and asking for it would ask for
            // exactly the content the removal deleted.
            //
            // Only ever asked of a name this device does not hold, which is
            // what this loop is. Asked of one it does hold, the answer would
            // be wrong in the one case it matters: every version covers the
            // empty one, so a replica that re-created the name and has
            // nothing in it yet would read as holding removed content, and
            // the document it is waiting for would never be sent.
            if self.version_is_removed_content(space, doc, theirs_encoded) {
                debug!(space, doc, "Ignoring an offer of content a removal covers");
                continue;
            }
            // Our own interest, not theirs: what they hold is their
            // business, and what this device stores is ours. This is the
            // half that holds against a peer which never reads the field.
            if !self.data_wants(space, doc) {
                debug!(
                    space,
                    doc, "Not creating a document outside this space's interest"
                );
                continue;
            }
            // The same door a blob for an unknown document goes through,
            // and the same ceiling. Counting the versions read above
            // instead would undercount by every document whose version
            // could not be read, which is how a space gets past its cap.
            if !self.data_space_admits_doc(space, doc, MAX_DOCS_PER_SPACE) {
                warn!(
                    space,
                    cap = MAX_DOCS_PER_SPACE,
                    "Space is at its document ceiling; ignoring the rest of the offer"
                );
                break;
            }
            if self.data_create_doc(space, doc).is_err() {
                continue;
            }
            // Folded into `ours` rather than re-read from the whole space
            // below: one version read per frame instead of two, and the
            // answer then describes exactly the documents this leg judged.
            match self.data_doc_version(space, doc) {
                Ok(version) => {
                    ours.insert(doc.clone(), version.clone());
                    created.insert(doc.clone(), version);
                }
                Err(err) => {
                    warn!(space, doc, error = %err, "Created a document from an offer but cannot read its version");
                }
            }
        }

        for (doc, ours_encoded) in &ours {
            // Their interest, not ours: this loop decides what to *send*.
            // Scoping by what they asked for is safe where scoping by the
            // names this frame happened to carry is not, because a document
            // they want and have never named is still in our list and still
            // matches, so the absence inference below still reaches it.
            if !asked_for(doc) {
                continue;
            }
            let Some(theirs_encoded) = theirs.get(doc) else {
                // They did not name this document. On a complete offer that
                // means they have never seen it and the whole document is
                // the answer. On a partial one it means nothing: the name
                // may be in a frame that has not arrived, or in no frame at
                // all because the offer was about a single document.
                if !partial {
                    self.offer_catch_up(space, channel, doc, None);
                }
                continue;
            };
            if theirs_encoded == ours_encoded {
                continue;
            }
            match BASE64.decode(theirs_encoded) {
                Ok(token) => {
                    self.offer_catch_up(space, channel, doc, Some(VersionToken::from_bytes(token)))
                }
                Err(err) => {
                    warn!(space, doc, error = %err, "Undecodable version token in a sync frame");
                }
            }
        }

        // One answer, and the exchange is over. Answering an answer is how a
        // pair of replicas talk to each other forever.
        //
        // Either way the documents created above are asked about, which is
        // the invariant to keep: the counter-offer names them when there is
        // one, and when there is not they are asked for on their own. An
        // answering leg that skipped the question would leave a document
        // created from it empty for as long as the link stays up, because
        // the peer has said everything it had to say and nothing else here
        // ever asks.
        if !reply {
            let our_gone = self.tombstones_for(space, channel);
            let our_want = self.interest_for(space, channel);
            // The counter-offer carries the complete list, interest or no
            // interest. Scoping it would save a name per unwanted document
            // and buy a familiar hazard: a reply sweep narrowed on the
            // answering side is how a document the peer has never seen gets
            // dropped with no symptom, and the fact that this particular
            // narrowing is provably safe is not worth a rule the next reader
            // has to re-derive. What the interest saves is the *document*,
            // which the catch-up loop above already declined to send.
            //
            // Removals are not scoped either, for a reason that is not about
            // caution: a peer that narrowed its interest keeps what it
            // already held, so a removal for a document it no longer wants
            // is exactly the one it still needs.
            self.send_version_frames(
                space,
                channel,
                OfferContents {
                    docs: ours,
                    gone: our_gone,
                    want: our_want,
                },
                true,
                false,
            );
        } else {
            self.send_version_frames(
                space,
                channel,
                OfferContents {
                    docs: created,
                    ..OfferContents::default()
                },
                true,
                true,
            );
        }
        Ok(())
    }

    /// Whether an offered version names content a removal here already
    /// covered.
    fn version_is_removed_content(&mut self, space: &str, doc: &str, encoded: &str) -> bool {
        match BASE64.decode(encoded) {
            Ok(bytes) => {
                self.data_version_below_floor(space, doc, &VersionToken::from_bytes(bytes))
            }
            // Undecodable is not "removed"; the arms below report it where
            // the decode is actually needed.
            Err(_) => false,
        }
    }

    /// Send `doc` to the peer in whatever form can carry it.
    ///
    /// The ladder, cheapest first: the changes since their version, a whole
    /// document, and then nothing. "Nothing" is a real rung and it is
    /// reported rather than retried: a document too large for every inline
    /// form needs the media path, which the attachment stage brings, and a
    /// silent stall would be indistinguishable from a peer with no changes.
    fn offer_catch_up(
        &mut self,
        space: &str,
        channel: &SyncChannel,
        doc: &str,
        theirs: Option<VersionToken>,
    ) {
        if let Some(token) = theirs {
            match self.data_catch_up(space, doc, &token) {
                Ok(CatchUp::UpToDate) => return,
                Ok(CatchUp::Updates(bytes)) if bytes.len() <= MAX_SYNC_BLOB_BYTES => {
                    self.send_sync_frame(
                        space,
                        channel,
                        &SyncBody::Delta {
                            doc: doc.to_string(),
                            blob: BASE64.encode(bytes),
                        },
                    );
                    return;
                }
                // Either too large to inline or a gap no run of changes can
                // express. Both are answered by the snapshot below.
                Ok(_) => {}
                Err(err) => {
                    warn!(space, doc, error = %err, "Could not compute catch-up");
                    return;
                }
            }
        }

        self.send_snapshot(space, channel, doc);
    }

    /// Send a whole document, the top rung and the answer to every refusal
    /// below it.
    ///
    /// Terminal by construction: a snapshot provokes no answer of its own,
    /// which is what lets the refusals underneath it ask for one freely.
    fn send_snapshot(&mut self, space: &str, channel: &SyncChannel, doc: &str) {
        match self.data_export_snapshot(space, doc) {
            Ok(bytes) if bytes.len() <= MAX_SYNC_BLOB_BYTES => {
                self.send_sync_frame(
                    space,
                    channel,
                    &SyncBody::Snapshot {
                        doc: doc.to_string(),
                        blob: BASE64.encode(bytes),
                    },
                );
            }
            // Too large for any frame. The rung above every frame is the
            // media path, which is where a document this size goes.
            Ok(bytes) => self.carry_snapshot_over_media(space, channel, doc, bytes),
            Err(err) => {
                warn!(space, doc, error = %err, "Could not export document for catch-up");
            }
        }
    }

    /// Send a document the media path carries, because no frame can.
    ///
    /// The top rung of the catch-up ladder and the last one. A document that
    /// cannot go this way either is reported rather than retried: nothing
    /// above this exists to try, and the two replicas stay apart until
    /// somebody makes the document smaller.
    ///
    /// Only ever 1:1, and it stayed that way when attachment bytes stopped
    /// being. The media path is a transfer to a confirmed session and two
    /// members of a group need not have one with each other, so a group
    /// space that reaches this rung reports instead. Carrying it as frames
    /// the way a blob now travels would need a rule this layer does not
    /// have: a blob is asked for, so a chunk of one is admitted against an
    /// outstanding request, while a snapshot is unsolicited by design and
    /// frames carrying one would be admitted on the sender's say-so alone.
    fn carry_snapshot_over_media(
        &mut self,
        space: &str,
        channel: &SyncChannel,
        doc: &str,
        bytes: Vec<u8>,
    ) {
        let size = bytes.len();
        let report = |protocol: &mut Self, reason: &str| {
            warn!(
                space,
                doc,
                bytes = size,
                reason,
                "Document cannot be replicated"
            );
            if let Ok(state) = crate::protocol::lock_shared_state(&protocol.shared_state) {
                state.emit_event(Event::DataDocUnsyncable {
                    space_id: space.to_string(),
                    doc_id: doc.to_string(),
                    bytes: size as u64,
                    reason: reason.to_string(),
                });
            }
        };

        if !matches!(channel, SyncChannel::Peer) {
            report(self, "group_space_has_no_media_path");
            return;
        }
        if size > MAX_MEDIA_SNAPSHOT_BYTES {
            report(self, "over_record_ceiling");
            return;
        }
        if !self.data_media_active(space) {
            report(self, "peer_cannot_carry_snapshots");
            return;
        }

        // Not a dead end and not reported as one: the document is already on
        // its way. This rung is re-entered on its own while a carriage runs,
        // because the peer stays stale for the whole of it and says so on
        // every exchange.
        let purpose = DataPurpose::Snapshot {
            doc: doc.to_string(),
        };
        if self.data_media_transfer_in_flight(space, &purpose) {
            debug!(
                space,
                doc, "This document is already crossing to the peer; not starting a second copy"
            );
            return;
        }

        debug!(
            space,
            doc,
            bytes = size,
            "Carrying a snapshot over the media path"
        );
        if let Err(err) = self.send_media_inner(
            space.to_string(),
            bytes,
            doc.to_string(),
            ContentType::File,
            MediaSendOptions::default(),
            Some(purpose),
        ) {
            // Transient by nature: no confirmed session yet, the per-peer
            // transfer slots are full, or the protocol is not running.
            // Reported at debug and left for anti-entropy, which asks again
            // on the next exchange. Raising it as unsyncable would tell an
            // application a document is permanently stuck when it is merely
            // busy.
            debug!(space, doc, error = %err, "Snapshot carriage deferred");
        }
    }

    /// Answer a peer that cannot apply what we sent it.
    ///
    /// Only a document already held is served. A request naming an unknown
    /// one would otherwise create it, which is a way to spend this device's
    /// storage that does not even require a blob.
    fn answer_snapshot_request(
        &mut self,
        space: &str,
        channel: &SyncChannel,
        doc: &str,
    ) -> Result<()> {
        if offline_protocol_data::validate_name(doc).is_err() {
            warn!(space, doc, "Snapshot request names an invalid document");
            return Ok(());
        }
        if !self.data_holds_doc(space, doc) {
            debug!(space, doc, "Snapshot request for a document we do not hold");
            return Ok(());
        }
        debug!(space, doc, "Serving a snapshot the peer asked for");
        self.send_snapshot(space, channel, doc);
        Ok(())
    }

    /// Apply a blob from a peer, with the containment the engine needs.
    fn accept_remote_blob(
        &mut self,
        space: &str,
        channel: &SyncChannel,
        doc: &str,
        encoded: &str,
        kind: BlobKind,
    ) -> Result<()> {
        if offline_protocol_data::validate_name(doc).is_err() {
            warn!(space, doc, "Sync frame names an invalid document");
            return Ok(());
        }
        // A blob for a document this device does not hold creates one, so
        // this path is bounded by the same ceiling the offer loop is.
        if !self.data_space_admits_doc(space, doc, MAX_DOCS_PER_SPACE) {
            warn!(
                space,
                doc,
                cap = MAX_DOCS_PER_SPACE,
                "Space is at its document ceiling; refusing a new one"
            );
            return Ok(());
        }
        // Bound the decode by the frame budget: base64 grows by a third, so
        // checking the encoded length first refuses an oversized blob before
        // allocating for it.
        if encoded.len() > MAX_SYNC_BLOB_BYTES * 4 / 3 + 4 {
            warn!(space, doc, "Sync frame carries an oversized blob");
            return Ok(());
        }
        let blob = match BASE64.decode(encoded) {
            Ok(blob) => blob,
            Err(err) => {
                warn!(space, doc, error = %err, "Undecodable blob in a sync frame");
                return Ok(());
            }
        };
        if blob.len() > MAX_SYNC_BLOB_BYTES {
            warn!(
                space,
                doc,
                bytes = blob.len(),
                "Sync blob over the frame budget"
            );
            return Ok(());
        }

        self.apply_contained_blob(space, channel, doc, blob, kind)
    }

    /// Apply remote document bytes with the containment the engine needs.
    ///
    /// Split from [`Self::accept_remote_blob`] so bytes that arrived over
    /// the media path go through exactly this, rather than through a second
    /// copy of it. The road the bytes travelled says nothing about what they
    /// are: both roads carry an import that can still end the process, from
    /// a peer who is authenticated and may still be wrong.
    ///
    /// The frame budget is deliberately not checked here. It bounds what one
    /// frame may carry, which is a fact about frames; the media path has its
    /// own, larger bound because it has no such limit.
    fn apply_contained_blob(
        &mut self,
        space: &str,
        channel: &SyncChannel,
        doc: &str,
        blob: Vec<u8>,
        kind: BlobKind,
    ) -> Result<()> {
        if offline_protocol_data::validate_name(doc).is_err() {
            warn!(space, doc, "A remote blob names an invalid document");
            return Ok(());
        }
        if !self.data_wants(space, doc) {
            debug!(
                space,
                doc, "Refusing a blob for a document outside this space's interest"
            );
            return Ok(());
        }
        if !self.data_space_admits_doc(space, doc, MAX_DOCS_PER_SPACE) {
            warn!(
                space,
                doc,
                cap = MAX_DOCS_PER_SPACE,
                "Space is at its document ceiling; refusing a new one"
            );
            return Ok(());
        }

        let digest = blob_digest(&blob);
        let mut record = self.load_sync_record(space);
        if record.quarantined.contains(&digest) {
            // This blob was in flight when a previous run died. It may be a
            // coincidence and it may be the reason; either way applying it
            // again to find out is a bad trade, because the cost of being
            // wrong is the process and the cost of being right is one
            // document change that the sender still holds.
            warn!(space, doc, "Refusing a quarantined blob");
            return Ok(());
        }

        // A blob for a name with a removal floor is judged against the floor
        // before anything acts on it. A blob carrying nothing beyond the
        // floor is the content that removal deleted, arriving from a replica
        // that has not heard about it yet: importing it would recreate the
        // document. Terminal when it refuses, like every other refusal here:
        // the sender is told by the `gone` entry in our next offer, not by
        // an answer to this.
        //
        // The judgement is an engine decode, not a header read. The engine
        // decodes every change the blob carries into a scratch document to
        // find the version they end at, with the same decoder an import
        // runs, and ADR 0019 contains that decoder for a reason: a crafted
        // blob can end the process inside it, and under `panic = "abort"`
        // nothing catches that. So it runs inside its own in-flight marker,
        // exactly as the import below does. A blob that ends the process
        // here leaves its digest behind, and the sender's retry is refused
        // instead of ending the next process too. Outside the marker this
        // would be the one engine decode of a peer's bytes with no
        // quarantine, on a path a peer steers a blob onto by sending the
        // floor first.
        //
        // Paid only by a name that has a floor: two marker writes on a path
        // a space nobody has removed anything from never takes.
        if self.data_doc_has_floor(space, doc) {
            record.in_flight = Some(digest.clone());
            self.persist_sync_record(space, &record);
            let below_floor = self.data_blob_below_floor(space, doc, &blob);
            record.in_flight = None;
            self.persist_sync_record(space, &record);
            if below_floor {
                debug!(
                    space,
                    doc, "Refusing a blob that carries only content a removal covers"
                );
                return Ok(());
            }
        }

        // Past every refusal, the blob carries something the removal never
        // saw, which is an edit made concurrently with it or after it. An
        // edit beats a removal, so the name comes back here, after the last
        // check that can refuse the blob and before the first step that
        // needs the name open. Brought back any earlier, a refused blob
        // would leave the name live with no document under it. The floor
        // stays, so the rest of the removed content is still refused from a
        // replica that has not heard about the removal.
        if let Err(err) = self.data_revive_removed_doc(space, doc) {
            warn!(space, doc, error = %err, "Could not bring back a removed document for a change that outlives its removal");
            return Ok(());
        }

        // Local edits waiting in memory are committed and pushed *before*
        // the import, so they leave on their own delta.
        //
        // A commit exports everything since the last one, regardless of who
        // authored it. So a local edit still pending when a remote blob
        // applies is folded into the same delta as the imported change, and
        // that delta is then suppressed as an echo toward the one peer it
        // was owed to. The edit is durable and announced to nobody: the next
        // commit exports only what came after it, and on a link that never
        // drops there is no reconnect to run an exchange that would notice.
        // Draining first costs one extra delta record and keeps the echo
        // suppression judging a delta that really is nothing but the echo.
        //
        // Reported and stepped over rather than propagated. This is our own
        // housekeeping, and a document held over its cap fails here on every
        // flush while still needing to accept the remote deletions that
        // bring it back under: refusing the import for it would close the
        // only door out.
        //
        // Ahead of the marker below rather than inside its window, so the
        // window stays the length of one import. A crash during this flush
        // would otherwise quarantine a blob the engine never saw.
        //
        // The two errors say opposite things about what is still pending, so
        // they are told apart. `DocTooLarge` is raised after the delta record
        // was written, pushed and announced, and only the size verdict that
        // follows it failed: nothing is left in the pending set and there is
        // nothing to compensate for. Any other error may have left the commit
        // un-exported, because the delta-write failure rewinds it back into
        // the pending set. The import's own flush would then fold that edit
        // into the imported change and suppress the pair toward the one peer
        // it was owed to, which is exactly the loss this pre-flush exists to
        // prevent.
        let mut preflush_left_edits_pending = false;
        match self.data_flush(space, doc) {
            Ok(()) => {}
            Err(Error::DocTooLarge { .. }) => {
                debug!(
                    space,
                    doc,
                    "Local edits were flushed and pushed before applying a remote change; the \
                     document is over its cap and refuses further growth"
                );
            }
            Err(err) => {
                warn!(
                    space,
                    doc,
                    error = %err,
                    "Could not flush local edits before applying a remote change; they may \
                     still be pending"
                );
                preflush_left_edits_pending = true;
            }
        }

        // Written *before* the engine is called. That ordering is the whole
        // mechanism: a blob that ends the process leaves its digest behind,
        // so the sender's next retry is refused instead of ending the next
        // process too.
        record.in_flight = Some(digest.clone());
        self.persist_sync_record(space, &record);

        let outcome = self.data_apply_remote(space, doc, &blob);

        record.in_flight = None;
        self.persist_sync_record(space, &record);

        // Read before the match below consumes the outcome.
        let applied = matches!(outcome, Ok(RemoteImport::Applied));

        match outcome {
            Ok(RemoteImport::Applied) => {
                debug!(space, doc, "Applied a remote change");
            }
            Ok(RemoteImport::AlreadyHave) => {}
            Ok(RemoteImport::Parked) => match kind {
                // The change is held behind a predecessor that has not
                // arrived. Asking for the gap now is what makes the parked
                // change durable: it lives only in the engine's memory, so a
                // restart before the predecessor lands would lose it, and
                // the version exchange is what brings both back. Scoped to
                // this document, because this document is what is missing
                // something.
                BlobKind::Delta => {
                    debug!(space, doc, "Remote change parked; asking for the gap");
                    self.request_doc_catch_up(space, channel, doc);
                }
                // A snapshot is the top of the ladder. Asking again returns
                // the same bytes, so this is reported and left, rather than
                // retried until one side gives up or the battery does.
                BlobKind::Snapshot => {
                    warn!(
                        space,
                        doc,
                        "A snapshot from the peer did not apply; these replicas cannot be \
                         reconciled inline"
                    );
                }
            },
            Ok(RemoteImport::RefusedTrimmedHistory) => match kind {
                // Nothing the peer can compute from our version closes this
                // gap: they would recompute the same refused delta, and
                // answering with a version offer is how that becomes an
                // exchange neither side can end. Name what would work.
                BlobKind::Delta => {
                    debug!(space, doc, "Remote change needs a snapshot; asking for one");
                    self.send_sync_frame(
                        space,
                        channel,
                        &SyncBody::NeedSnapshot {
                            doc: doc.to_string(),
                        },
                    );
                }
                // The end of the line, and the one outcome this layer
                // cannot recover from. The peer forked below a point this
                // replica compacted away, so the ancestors their changes
                // need were deleted here rather than left out of the
                // message: no frame carries them back, including the whole
                // document, which is what just arrived. Reported plainly
                // rather than retried, because a retry returns these same
                // bytes and refusing them is the only thing keeping the
                // process alive.
                BlobKind::Snapshot => {
                    warn!(
                        space,
                        doc,
                        "A snapshot from the peer cannot merge into this replica; the two \
                         have diverged past what replication can reconcile"
                    );
                }
            },
            Err(err) => match kind {
                BlobKind::Delta => {
                    warn!(space, doc, error = %err, "Remote change refused");
                }
                // The top of the ladder, so there is nothing above it to
                // ask for. Unreconcilable divergence is refused before it
                // gets here rather than failing here, which leaves this for
                // bytes that are simply bad.
                BlobKind::Snapshot => {
                    warn!(space, doc, error = %err, "A snapshot from the peer was unreadable");
                }
            },
        }

        // A pre-flush that failed followed by an import that applied is the
        // one combination where a local edit can have been folded into the
        // imported change and then suppressed toward the peer it was owed
        // to. Nothing on the wire carries it and no trigger is left to
        // notice, so the gap is announced and the peer asks for what it is
        // missing. The alternative, sending the fold on with the suppression
        // lifted, would hand the peer its own change back.
        //
        // Gated on `Applied` for the reason that also makes it terminate.
        // The fold needs the import's own flush to have succeeded, and a
        // storage failure that is still failing fails that one too, which
        // makes the import answer `Err`. So every nudge costs a fresh
        // transient failure that recovered inside one frame, and the offers
        // cannot sustain each other. `Applied` and not "anything but `Err`"
        // for the same reason: the outcomes that did not apply left the edit
        // pending, where the next pre-flush still owes it to the peer.
        //
        // This is why a document over its cap must answer `Applied` rather
        // than `DocTooLarge`, and does: its flush wrote the fold and then
        // refused further growth, so the edit is stranded exactly as here,
        // on the documents whose only pending edit is the deletion that
        // brings them back under.
        //
        // One residual is left uncompensated on purpose. A `persist_space`
        // failure after a flush that succeeded also strands a fold, and is
        // indistinguishable here from a flush that failed. Offering on every
        // `Err` would cover it and break the argument above: a store that
        // keeps failing would then emit an offer per frame, and those do
        // sustain each other. The narrower gap is the better trade.
        //
        // `origin` is `None`, not the space: what may be stranded here is
        // this device's own edit, so naming the space would suppress the
        // announcement toward the only peer that could act on it.
        if preflush_left_edits_pending && applied {
            self.nudge_data_sync(space, None, "preflush_failure");
        }

        Ok(())
    }

    // ---- attachments -------------------------------------------------

    /// Key of a pending fetch: the space asked in, then the blob wanted.
    ///
    /// Composed rather than keyed by hash alone because the same bytes may
    /// be referenced from two spaces, and an answer from one peer must not
    /// close a question put to another.
    fn attachment_fetch_key(space: &str, hash: &str) -> String {
        format!("{space}{GROUP_OFFER_KEY_SEP}{hash}")
    }

    /// The SHA-256 of `bytes`, in the spelling an attachment reference uses.
    ///
    /// Exposed because an application writing a reference needs the address
    /// of its own bytes, and computing it anywhere else risks a second
    /// spelling of the same hash.
    pub fn data_attachment_hash(bytes: &[u8]) -> String {
        hex_encode(&Sha256::digest(bytes))
    }

    /// Ask the peer a 1:1 space is named after for the bytes behind a
    /// reference.
    ///
    /// Pull rather than push, and the reason is bandwidth somebody else
    /// pays for: a space may hold references to more bytes than a phone
    /// wants over Bluetooth, so the decision to spend that belongs to the
    /// application, taken per blob, at the moment somebody opens one.
    ///
    /// The answer arrives later as [`Event::DataAttachmentReceived`] or
    /// [`Event::DataAttachmentUnavailable`], never inline.
    ///
    /// Group spaces are refused in this version. A blob travels the media
    /// path, which is a 1:1 transfer to a confirmed session, and two members
    /// of a group need not have one with each other.
    pub fn data_fetch_attachment(&mut self, space: &str, hash: &str) -> Result<()> {
        if !self.config.data.enabled {
            return Err(Error::DataDisabled);
        }
        Self::validate_attachment_hash(hash)?;
        if self.group_space_roster(space).is_some() {
            return Err(Error::InvalidArgument(format!(
                "space {space} is a group: a reference names bytes and not who holds them, so \
                 ask a member with data_fetch_attachment_from"
            )));
        }
        if !self.data_media_active(space) {
            return Err(Error::InvalidArgument(format!(
                "peer {space} did not advertise attachment carriage"
            )));
        }

        let key = Self::attachment_fetch_key(space, hash);
        self.expire_attachment_fetches();
        // The requester's own bound, which the spec makes a MUST and which
        // the holder's window cannot substitute for: an application retrying
        // a spinner on a timer would otherwise seal and send one frame per
        // retry, and the holder answering none of them is the case that
        // provokes the retries.
        if let Some(pending) = self.pending_attachment_fetches.get(&key) {
            if Instant::now().duration_since(pending.last_seen) < DATA_SYNC_OFFER_INTERVAL {
                debug!(
                    space,
                    "Blob fetch repeated inside its window; already asked"
                );
                return Ok(());
            }
        }
        if self.pending_attachment_fetches.len() >= MAX_PENDING_ATTACHMENT_FETCHES
            && !self.pending_attachment_fetches.contains_key(&key)
        {
            // Oldest by last sign of life rather than by age, which is what
            // keeps a fetch whose bytes are actively arriving out of this:
            // every chunk of an answer refreshes its entry, so a live
            // carriage sits among the newest and is never the one displaced.
            if let Some(oldest) = self
                .pending_attachment_fetches
                .iter()
                .min_by_key(|(_, pending)| pending.last_seen)
                .map(|(key, _)| key.clone())
            {
                let evicted = self.pending_attachment_fetches.remove(&oldest);
                // Reported, exactly as an expiry is. Eviction reaches the
                // same end state by a different road: the fetch is over and
                // no bytes will ever be admitted for it, so an application
                // told nothing is left showing the spinner that the whole
                // refusal mechanism exists to end.
                Self::report_fetch_ended(
                    &self.shared_state,
                    &oldest,
                    evicted
                        .as_ref()
                        .and_then(|pending| pending.holder.as_deref()),
                    "evicted",
                );
            }
        }
        self.pending_attachment_fetches
            .insert(key, PendingFetch::to_peer());

        self.send_sync_frame(
            space,
            &SyncChannel::Peer,
            &SyncBody::NeedBlob {
                hash: hash.to_string(),
                to: None,
            },
        );
        Ok(())
    }

    /// Send one chunk frame as a holder would, without the hash check
    /// `data_provide_attachment` makes on the way out.
    ///
    /// Test-only, and it exists because the receive-side hash check has no
    /// other way to be reached: the send surface refuses to carry bytes that
    /// do not hash to the address they are sent under, so a lying holder
    /// cannot be staged through it.
    #[cfg(test)]
    pub(crate) fn send_chunk_for_test(
        &mut self,
        space: &str,
        channel: &SyncChannel,
        hash: &str,
        index: u32,
        total: u32,
        blob: &str,
    ) {
        self.send_sync_frame(
            space,
            channel,
            &SyncBody::Chunk {
                hash: hash.to_string(),
                i: index,
                n: total,
                blob: blob.to_string(),
            },
        );
    }

    /// Take one piece of a blob a member is sending.
    ///
    /// Admitted only against a fetch this device started, and only from the
    /// member it was put to. Both halves matter: the first is what stops a
    /// member spending this device's memory on bytes nobody asked for, and
    /// the second is what stops a different member answering a question that
    /// was not put to them. Neither is a substitute for the hash check at
    /// the end, which is the only thing that says what the bytes *are*.
    ///
    /// Provokes no answer in any outcome, which is what keeps `need_blob` a
    /// single-hop question.
    fn accept_blob_chunk(
        &mut self,
        space: &str,
        sender: &str,
        hash: &str,
        index: u32,
        total: u32,
        encoded: &str,
    ) -> Result<()> {
        if Self::validate_attachment_hash(hash).is_err() {
            warn!(space, "A blob chunk names something that is not a hash");
            return Ok(());
        }
        // The receiver's own bound, not a restatement of the sender's. A
        // holder that ignores the cap would otherwise spend this device's
        // memory up to whatever it felt like declaring.
        if total == 0 || total > MAX_GROUP_BLOB_CHUNKS || index >= total {
            warn!(
                space,
                index, total, "A blob chunk declares a shape this layer does not carry"
            );
            return Ok(());
        }
        let key = Self::attachment_fetch_key(space, hash);
        let Some(pending) = self.pending_attachment_fetches.get_mut(&key) else {
            debug!(space, "Discarding a blob chunk nothing asked for");
            return Ok(());
        };
        if !pending.is_answered_by(space, sender) {
            warn!(
                space,
                sender, "Discarding a blob chunk from a member this fetch was not put to"
            );
            return Ok(());
        }
        let bytes = match BASE64.decode(encoded) {
            Ok(bytes) => bytes,
            Err(err) => {
                warn!(space, error = %err, "Undecodable blob chunk");
                return Ok(());
            }
        };
        if bytes.len() > MAX_SYNC_BLOB_BYTES {
            warn!(
                space,
                bytes = bytes.len(),
                "A blob chunk is over the frame budget"
            );
            return Ok(());
        }

        // Arriving bytes are the sign of life the timeout measures from, and
        // the refresh happens before anything below can refuse this chunk:
        // a carriage in progress must not expire because one piece of it was
        // malformed.
        pending.last_seen = Instant::now();
        let assembly = pending.assembly.get_or_insert_with(BlobAssembly::default);
        if assembly.total == 0 {
            assembly.total = total;
        } else if assembly.total != total {
            // One of the two is lying about the shape. The chunk is refused
            // rather than the fetch, because the hash at the end is what
            // decides the question either way, and ending the fetch here
            // would let one bad frame cancel a carriage that is otherwise
            // arriving.
            warn!(
                space,
                declared = assembly.total,
                arriving = total,
                "A blob chunk contradicts the count its siblings declared"
            );
            return Ok(());
        }
        assembly.chunks.insert(index, bytes);
        if (assembly.chunks.len() as u32) < assembly.total {
            return Ok(());
        }

        // Complete. Whether it is the *right* blob is a separate question,
        // and the only one authentication cannot answer.
        let assembled: Vec<u8> = assembly
            .chunks
            .values()
            .flat_map(|chunk| chunk.iter().copied())
            .collect();
        let asked = self
            .pending_attachment_fetches
            .remove(&key)
            .and_then(|pending| pending.holder);

        let actual = Self::data_attachment_hash(&assembled);
        if actual != hash {
            warn!(
                space,
                sender, "Assembled blob does not hash to what was asked for"
            );
            Self::report_fetch_ended(&self.shared_state, &key, asked.as_deref(), "hash_mismatch");
            return Ok(());
        }
        if let Ok(state) = crate::protocol::lock_shared_state(&self.shared_state) {
            state.emit_event(Event::DataAttachmentReceived {
                space_id: space.to_string(),
                peer_id: sender.to_string(),
                hash: hash.to_string(),
                data: BASE64.encode(&assembled),
            });
        }
        Ok(())
    }

    /// Ask one member of a group for the bytes behind a reference.
    ///
    /// A group needs the member named, where a 1:1 space does not: the
    /// reference replicates to everybody and says nothing about who holds
    /// the bytes. The SDK does not walk the roster on the application's
    /// behalf, because each miss costs the whole silence timeout and a
    /// reference nobody holds would spend one per member. The application
    /// knows who wrote the reference; that is the member to ask.
    ///
    /// The answer arrives as frames under the group key rather than as a
    /// transfer, so it needs no pairwise session with that member, and it is
    /// bounded at 32 pieces of 32 KiB: 1 MiB in all. Anything larger travels
    /// over a 1:1 session, and the holder is told so at the call.
    pub fn data_fetch_attachment_from(
        &mut self,
        space: &str,
        peer: &str,
        hash: &str,
    ) -> Result<()> {
        if !self.config.data.enabled {
            return Err(Error::DataDisabled);
        }
        Self::validate_attachment_hash(hash)?;
        let Some(members) = self.group_space_roster(space) else {
            return Err(Error::InvalidArgument(format!(
                "space {space} is not a group; use data_fetch_attachment"
            )));
        };
        if !members.iter().any(|member| member == peer) {
            return Err(Error::InvalidArgument(format!(
                "{peer} is not a member of {space}"
            )));
        }
        // This device is in the roster it just read, so naming itself is a
        // question it would put to nobody and then wait out the silence
        // timeout for. An application walking a roster to choose a member
        // reaches this, and the honest answer is that it has the bytes or
        // it does not.
        if peer == self.local_id {
            return Err(Error::InvalidArgument(
                "a group fetch names another member to ask, and a device is not a peer of itself"
                    .to_string(),
            ));
        }
        // A question, and the only directed frame this layer puts to a
        // member who asked for nothing. Every other one answers something
        // that member sent, which is itself proof they intercept these
        // frames; this one carries no such proof, so it needs the positive
        // knowledge the roster-wide gate requires of every member before a
        // frame is broadcast to them. Without it a member on a build that
        // does not intercept `__DATA_V1__` is handed the question as chat
        // text, which is the single failure that gate exists to refuse.
        //
        // An inviter attests this for members a device has never dealt with
        // directly, which is what keeps the ordinary case working: a group
        // is not built out of existing 1:1 contacts.
        if !self.data_group_frames_known_received(peer) {
            return Err(Error::InvalidArgument(format!(
                "nothing is known about whether member {peer} receives replication \
                 frames inside a group, and a question they cannot intercept would \
                 be shown to them as text"
            )));
        }
        // Refused only on knowledge, never on its absence. Members do not
        // exchange key packages with each other, so "we have not heard" is
        // the ordinary case and must not read as "they cannot": the
        // inviter's attestation fills it in where there is one, and where
        // there is not, asking and hearing nothing is the honest outcome
        // and the silence timeout reports it.
        if self.data_group_blob_known_absent(peer) {
            return Err(Error::InvalidArgument(format!(
                "member {peer} does not carry attachment bytes inside a group"
            )));
        }

        let key = Self::attachment_fetch_key(space, hash);
        self.expire_attachment_fetches();
        match self.pending_attachment_fetches.get(&key) {
            // The same question asked twice, which is what the window is
            // for: a spinner on a timer would otherwise seal and send one
            // frame per retry.
            Some(pending)
                if pending.holder.as_deref() == Some(peer)
                    && Instant::now().duration_since(pending.last_seen)
                        < DATA_SYNC_OFFER_INTERVAL =>
            {
                debug!(
                    space,
                    peer, "Blob fetch repeated inside its window; already asked"
                );
                return Ok(());
            }
            // A different member is a different question, and the one an
            // application falls back to when the first goes quiet. The
            // record is keyed by the space and the blob, neither of which
            // changes, so without this a fallback inside the window
            // returned `Ok` and asked nobody, and after it silently threw
            // away whatever the first member had already sent.
            //
            // Displacing the first is reported like any other eviction,
            // naming the member it was put to, so an application tracking
            // a question per member is not left holding one that never
            // ends.
            Some(pending) if pending.holder.as_deref() != Some(peer) => {
                let displaced = self.pending_attachment_fetches.remove(&key);
                Self::report_fetch_ended(
                    &self.shared_state,
                    &key,
                    displaced
                        .as_ref()
                        .and_then(|pending| pending.holder.as_deref()),
                    "evicted",
                );
            }
            _ => {}
        }
        if self.pending_attachment_fetches.len() >= MAX_PENDING_ATTACHMENT_FETCHES
            && !self.pending_attachment_fetches.contains_key(&key)
        {
            if let Some(oldest) = self
                .pending_attachment_fetches
                .iter()
                .min_by_key(|(_, pending)| pending.last_seen)
                .map(|(key, _)| key.clone())
            {
                let evicted = self.pending_attachment_fetches.remove(&oldest);
                Self::report_fetch_ended(
                    &self.shared_state,
                    &oldest,
                    evicted
                        .as_ref()
                        .and_then(|pending| pending.holder.as_deref()),
                    "evicted",
                );
            }
        }
        self.pending_attachment_fetches
            .insert(key, PendingFetch::to_member(peer));

        self.send_sync_frame(
            space,
            &SyncChannel::GroupDirected(peer.to_string()),
            &SyncBody::NeedBlob {
                hash: hash.to_string(),
                to: Some(peer.to_string()),
            },
        );
        Ok(())
    }

    /// Answer a peer's request with the bytes.
    ///
    /// The bytes come from the application because this SDK never held them:
    /// a blob does not enter a document and does not enter protocol state,
    /// so the only copy is wherever the app put it.
    ///
    /// Refuses bytes that do not hash to `hash`. That check is here rather
    /// than only on the receiving side so a mistake is reported to the app
    /// that made it, while it still has the file in hand, instead of
    /// travelling the whole media path to be refused by somebody else.
    pub fn data_provide_attachment(
        &mut self,
        space: &str,
        peer: &str,
        hash: &str,
        bytes: Vec<u8>,
    ) -> Result<()> {
        if !self.config.data.enabled {
            return Err(Error::DataDisabled);
        }
        Self::validate_attachment_hash(hash)?;
        if bytes.is_empty() {
            return Err(Error::InvalidArgument(
                "an attachment must have bytes".to_string(),
            ));
        }
        let group = self.group_space_roster(space);
        if group.is_none() && space != peer {
            return Err(Error::InvalidArgument(format!(
                "space {space} is not the 1:1 space with {peer}"
            )));
        }
        if let Some(members) = &group {
            if !members.iter().any(|member| member == peer) {
                return Err(Error::InvalidArgument(format!(
                    "{peer} is not a member of {space}"
                )));
            }
        }
        let actual = Self::data_attachment_hash(&bytes);
        if actual != hash {
            return Err(Error::InvalidArgument(format!(
                "these bytes hash to {actual}, not to {hash}"
            )));
        }

        // A group answer leaves as frames rather than as a transfer, and
        // leaves all at once: that is what keeps the request a single-hop
        // question, and it is why the size is bounded here rather than by
        // the transfer layer's own limit.
        if group.is_some() {
            let chunks = bytes.len().div_ceil(MAX_SYNC_BLOB_BYTES);
            if chunks > MAX_GROUP_BLOB_CHUNKS as usize {
                return Err(Error::InvalidArgument(format!(
                    "an attachment carried inside a group is at most {} bytes, and these are {}: \
                     send it over a 1:1 session instead",
                    MAX_GROUP_BLOB_CHUNKS as usize * MAX_SYNC_BLOB_BYTES,
                    bytes.len()
                )));
            }
            let channel = SyncChannel::GroupDirected(peer.to_string());
            for (index, chunk) in bytes.chunks(MAX_SYNC_BLOB_BYTES).enumerate() {
                self.send_sync_frame(
                    space,
                    &channel,
                    &SyncBody::Chunk {
                        hash: hash.to_string(),
                        i: index as u32,
                        n: chunks as u32,
                        blob: BASE64.encode(chunk),
                    },
                );
            }
            return Ok(());
        }

        // Answering twice is not an error to report back: the app did as it
        // was asked, and the bytes are on their way. Idempotent rather than
        // refused, because a request repeated after its window reopens is
        // ordinary and an app has no way to see that its first answer is
        // still in flight.
        let purpose = DataPurpose::Attachment {
            hash: hash.to_string(),
        };
        if self.data_media_transfer_in_flight(peer, &purpose) {
            debug!(
                peer,
                "These bytes are already crossing to this peer; not starting a second copy"
            );
            return Ok(());
        }

        self.send_media_inner(
            peer.to_string(),
            bytes,
            // The media layer wants a file name and an attachment has none
            // that is load-bearing: the display name lives on the reference
            // in the document, where every replica can already read it.
            hash.to_string(),
            ContentType::File,
            MediaSendOptions::default(),
            Some(purpose),
        )?;
        Ok(())
    }

    /// Tell a peer their fetch will not be answered.
    ///
    /// A first-class answer rather than silence. The asking side cannot
    /// otherwise tell a peer that no longer holds the bytes from one that is
    /// merely slow, and shows a person a spinner that never resolves.
    pub fn data_decline_attachment(&mut self, space: &str, peer: &str, hash: &str) -> Result<()> {
        if !self.config.data.enabled {
            return Err(Error::DataDisabled);
        }
        Self::validate_attachment_hash(hash)?;
        let group = self.group_space_roster(space);
        if group.is_none() && space != peer {
            return Err(Error::InvalidArgument(format!(
                "space {space} is not the 1:1 space with {peer}"
            )));
        }
        let channel = match &group {
            Some(members) => {
                if !members.iter().any(|member| member == peer) {
                    return Err(Error::InvalidArgument(format!(
                        "{peer} is not a member of {space}"
                    )));
                }
                SyncChannel::GroupDirected(peer.to_string())
            }
            None => {
                // Gated like every other frame in this file, and it is the
                // one that was not. A refusal toward a peer that never
                // advertised the capability is unreadable to them, so the
                // caller believes they answered and the asking side keeps
                // waiting; toward a blocked peer it is a liveness signal
                // blocking exists to withhold.
                //
                // A group needs no equivalent: the refusal rides the group
                // key, which every member can already open.
                if !self.data_media_active(peer) {
                    return Err(Error::InvalidArgument(format!(
                        "peer {peer} did not advertise attachment carriage"
                    )));
                }
                SyncChannel::Peer
            }
        };
        self.send_sync_frame(
            space,
            &channel,
            &SyncBody::BlobGone {
                hash: hash.to_string(),
            },
        );
        Ok(())
    }

    /// Whether a peer's request for one blob should reach the application.
    ///
    /// Two bounds, because a peer controls both how often it asks and what it
    /// asks for. The per-hash window stops a repeat; the per-peer budget
    /// stops a flood of distinct hashes, which the window alone cannot see as
    /// a flood at all because every hash in it is new.
    ///
    /// Kept out of the version-offer window map deliberately. Every other key
    /// in that map is a peer or a peer-and-group, both bounded by a roster.
    /// This one would be bounded by nothing, and that map has no cap and no
    /// sweep.
    fn blob_request_due(&mut self, peer: &str, hash: &str) -> bool {
        let now = Instant::now();
        self.blob_request_windows
            .retain(|_, seen| now.duration_since(*seen) < DATA_SYNC_OFFER_INTERVAL);

        let key = format!("{peer}{GROUP_OFFER_KEY_SEP}{hash}");
        if self.blob_request_windows.contains_key(&key) {
            debug!(peer, "Blob request repeated inside its window; ignoring");
            return false;
        }

        let live_for_peer = self
            .blob_request_windows
            .keys()
            .filter(|key| {
                key.split_once(GROUP_OFFER_KEY_SEP)
                    .is_some_and(|(asker, _)| asker == peer)
            })
            .count();
        if live_for_peer >= MAX_BLOB_REQUESTS_PER_WINDOW {
            warn!(
                peer,
                budget = MAX_BLOB_REQUESTS_PER_WINDOW,
                "Peer is over its blob-request budget; ignoring"
            );
            return false;
        }

        // The global bound, which holds when a peer's budget resets and when
        // many peers each stay under theirs.
        if self.blob_request_windows.len() >= MAX_BLOB_REQUEST_WINDOWS {
            if let Some(oldest) = self
                .blob_request_windows
                .iter()
                .min_by_key(|(_, seen)| **seen)
                .map(|(key, _)| key.clone())
            {
                self.blob_request_windows.remove(&oldest);
            }
        }
        self.blob_request_windows.insert(key, now);
        true
    }

    fn validate_attachment_hash(hash: &str) -> Result<()> {
        offline_protocol_data::validate_attachment_hash(hash)
            .map_err(|err| Error::InvalidArgument(err.to_string()))
    }

    /// Whether this device is waiting for the blob behind `hash` from `peer`.
    ///
    /// Read before a transfer is admitted rather than after it completes.
    /// Refusing at the end still refuses, but only once the bytes have been
    /// buffered, reassembled and checksummed, which is the storage and the
    /// battery the check exists to protect.
    pub(crate) fn awaiting_attachment(&self, peer: &str, hash: &str) -> bool {
        self.pending_attachment_fetches
            .contains_key(&Self::attachment_fetch_key(peer, hash))
    }

    /// Forget everything remembered about attachments with one peer.
    ///
    /// Composed through [`Self::attachment_fetch_key`] rather than by
    /// re-spelling the separator, which is what the separator constant exists
    /// for: a key built one way and matched another is a leak that no test
    /// notices until the two spellings diverge.
    pub(crate) fn forget_attachment_state(&mut self, peer: &str) {
        let prefix = Self::attachment_fetch_key(peer, "");
        // Reported, not merely dropped. This is a road that ends a fetch
        // without bytes like any other, and it is reached while an
        // application is waiting: a peer re-keying without the capability,
        // an unblock, a peer forgotten to keep a set bounded. Removing the
        // record silently also forecloses the expiry that would have
        // reported it later, so the spinner this event exists to end never
        // ends at all.
        let ended: Vec<(String, Option<String>)> = self
            .pending_attachment_fetches
            .iter()
            // Two roads reach the same peer. On a 1:1 space the peer *is*
            // the space, so the key carries it; in a group the space is the
            // group and the member asked is inside the record. A question
            // put to a peer this device has stopped replicating with cannot
            // be answered by either road, so both go, and a group fetch
            // left behind would hold its slot until it timed out.
            .filter(|(key, pending)| {
                key.starts_with(&prefix) || pending.holder.as_deref() == Some(peer)
            })
            .map(|(key, pending)| (key.clone(), pending.holder.clone()))
            .collect();
        for (key, holder) in ended {
            self.pending_attachment_fetches.remove(&key);
            Self::report_fetch_ended(&self.shared_state, &key, holder.as_deref(), "peer_gone");
        }
        self.blob_request_windows
            .retain(|key, _| !key.starts_with(&prefix));
    }

    /// [`Self::forget_attachment_state`] for every peer at once, for the
    /// road that forgets every peer at once.
    ///
    /// Not a loop over the roster, because the caller has already emptied
    /// it: the capability sets are cleared wholesale when the bound on
    /// remembered peers is reached, so by then there is no peer left to name
    /// and a fetch keyed by one would be swept by nothing.
    pub(crate) fn end_every_pending_fetch(&mut self) {
        for (key, pending) in std::mem::take(&mut self.pending_attachment_fetches) {
            Self::report_fetch_ended(
                &self.shared_state,
                &key,
                pending.holder.as_deref(),
                "peer_gone",
            );
        }
    }

    /// Report a fetch this device has given up on, keyed the way
    /// [`Self::attachment_fetch_key`] composes them.
    ///
    /// Shared by every road that ends a fetch without bytes, because they all
    /// owe the application the same thing and a second spelling of this event
    /// is a road that quietly owes it nothing. `reason` is a fixed token this
    /// crate chooses, never a peer's words: see the producer rule in
    /// `telemetry::scrub_event`.
    ///
    /// `holder` is the member the question was put to, which every caller
    /// reads off the record it is ending. It is carried rather than derived
    /// because it cannot be: the key composes the *space*, and in a group
    /// the space is not the peer.
    fn report_fetch_ended(
        shared_state: &std::sync::Arc<std::sync::Mutex<crate::protocol::SharedState>>,
        key: &str,
        holder: Option<&str>,
        reason: &str,
    ) {
        let Some((space, hash)) = key.split_once(GROUP_OFFER_KEY_SEP) else {
            return;
        };
        debug!(space, reason, "A blob fetch ended with no bytes");
        if let Ok(state) = crate::protocol::lock_shared_state(shared_state) {
            state.emit_event(Event::DataAttachmentUnavailable {
                space_id: space.to_string(),
                // The member the question was put to. A 1:1 space IS the
                // peer, which is why the key alone composes both there; a
                // group space is one scope with many members, so the record
                // carries the one asked and this reads it. Falling back to
                // the space in a group would hand an application a group id
                // where this event's own documentation promises the peer
                // that was asked, and every road but the refusal would
                // report it.
                peer_id: holder.unwrap_or(space).to_string(),
                hash: hash.to_string(),
                reason: reason.to_string(),
            });
        }
    }

    /// Note that the answer to a fetch is on its way.
    ///
    /// A transfer that is visibly progressing is the answer this device is
    /// waiting for, so the question must not expire while it arrives. Without
    /// this the record times out mid-carriage and
    /// [`Self::accept_data_media_payload`] then drops a blob that fully
    /// arrived as one nobody asked for, which no retry can fix: the retry
    /// carries the same bytes over the same radio and expires at the same
    /// point.
    ///
    /// Only an attachment has a record to refresh. A snapshot is unsolicited
    /// by design and keeps none.
    pub(crate) fn refresh_attachment_fetch(&mut self, peer: &str, purpose: &DataPurpose) {
        let DataPurpose::Attachment { hash } = purpose else {
            return;
        };
        let key = Self::attachment_fetch_key(peer, hash);
        if let Some(seen_at) = self.pending_attachment_fetches.get_mut(&key) {
            seen_at.last_seen = Instant::now();
        }
    }

    /// Whether a transfer for exactly this purpose is already on its way to
    /// this peer.
    ///
    /// Both roads onto the media path are re-entered by things that recur on
    /// their own: a peer stays stale for the whole multi-minute carriage of a
    /// document, so every anti-entropy exchange in that window reaches the
    /// top rung again, and an application answering a repeated request has no
    /// way to know its first answer is still in flight.
    ///
    /// Without this check those repeats become second copies of one transfer.
    /// A peer may hold only [`MAX_CONCURRENT_MEDIA_TRANSFERS_PER_PEER`] at
    /// once, so the duplicates halve the throughput of the transfer that
    /// matters and then fill the slots, and the next thing to fail is the
    /// application's own media send, with a limit error whose cause it cannot
    /// see: the transfers holding those slots are invisible to it by design.
    ///
    /// [`MAX_CONCURRENT_MEDIA_TRANSFERS_PER_PEER`]: crate::constants::MAX_CONCURRENT_MEDIA_TRANSFERS_PER_PEER
    fn data_media_transfer_in_flight(&self, recipient: &str, purpose: &DataPurpose) -> bool {
        self.outbound_media_transfers.values().any(|transfer| {
            transfer.recipient == recipient && transfer.data_purpose.as_ref() == Some(purpose)
        })
    }

    /// Give up on fetches nobody answered, and say so.
    ///
    /// The reporting half is the point. A holder is only asked to answer
    /// `need_blob`, never required to, so a fetch can go unanswered forever
    /// with nothing wrong anywhere. Expiring the entry silently would leave
    /// the asking application exactly where the whole refusal mechanism
    /// exists to stop it being: showing somebody a spinner that never
    /// resolves.
    pub(crate) fn expire_attachment_fetches(&mut self) {
        let now = Instant::now();
        let expired: Vec<(String, Option<String>)> = self
            .pending_attachment_fetches
            .iter()
            .filter(|(_, pending)| {
                now.duration_since(pending.last_seen) >= ATTACHMENT_FETCH_TIMEOUT
            })
            .map(|(key, pending)| (key.clone(), pending.holder.clone()))
            .collect();
        for (key, holder) in expired {
            self.pending_attachment_fetches.remove(&key);
            Self::report_fetch_ended(&self.shared_state, &key, holder.as_deref(), "timeout");
        }
    }

    /// A peer wants the bytes behind a reference.
    ///
    /// Handed straight to the application, because the application is the
    /// only party that has them. Rate limited per blob on the window that
    /// suppresses version offers: a peer retrying a fetch must not turn into
    /// a stream of events, and the answer to the second ask within the
    /// window is the same as the answer to the first.
    fn answer_blob_request(
        &mut self,
        space: &str,
        channel: &SyncChannel,
        hash: &str,
        to: Option<&str>,
    ) -> Result<()> {
        if Self::validate_attachment_hash(hash).is_err() {
            warn!(space, "Blob request names something that is not a hash");
            return Ok(());
        }
        // The member that asked, which in a group is not the space. A
        // roster-wide request is refused rather than answered: there is
        // nobody to send the bytes to, and answering it would mean choosing
        // a member on the asker's behalf.
        let asker = match channel {
            SyncChannel::Peer => space.to_string(),
            SyncChannel::GroupDirected(member) => member.clone(),
            SyncChannel::GroupBroadcast => {
                warn!(
                    space,
                    "A blob request arrived addressed to the whole roster; ignoring"
                );
                return Ok(());
            }
        };
        // Inside a group the question names the member it is for, and one
        // that names somebody else is not ours to answer. The sender's
        // directed frame is promoted to the whole roster when its ratchet
        // budget runs out, so it arrives at every member either way; without
        // this check every member's application would be asked for bytes
        // that only one of them can have accepted.
        if !matches!(channel, SyncChannel::Peer) {
            match to {
                Some(member) if member == self.local_id => {}
                Some(_) => {
                    debug!(
                        space,
                        "A blob request names another member; not ours to answer"
                    );
                    return Ok(());
                }
                None => {
                    warn!(
                        space,
                        "A blob request inside a group names no member; ignoring"
                    );
                    return Ok(());
                }
            }
        }
        // A request this device could not answer is not put to the
        // application. Both answers to `DataAttachmentRequested` refuse
        // toward a peer that never advertised carriage: `provide` cannot
        // reach the media path and `decline` cannot reach the wire. Emitting
        // it anyway hands an app a question with no legal reply, which reads
        // to whoever handles it as an SDK that rejects its own events.
        //
        // Ahead of the rate limiter rather than behind it, so a peer that
        // cannot be answered spends no window slot: the budget exists to
        // bound work this device would otherwise do, and there is none here.
        let answerable = match channel {
            SyncChannel::Peer => self.data_media_active(space),
            // In a group the question is whether *this* device can answer,
            // and it can: the bytes leave as frames under the group key,
            // which every member can already open. What the asker speaks is
            // their problem, and they only get an answer because they asked.
            _ => self.config.data.enabled,
        };
        if !answerable {
            warn!(
                space,
                "A blob request came from a peer that never advertised carriage; ignoring"
            );
            return Ok(());
        }
        // Budgeted against the member that asked rather than the space, or
        // one member could spend the whole group's window.
        if !self.blob_request_due(&asker, hash) {
            return Ok(());
        }
        if let Ok(state) = crate::protocol::lock_shared_state(&self.shared_state) {
            state.emit_event(Event::DataAttachmentRequested {
                space_id: space.to_string(),
                peer_id: asker,
                hash: hash.to_string(),
            });
        }
        Ok(())
    }

    /// A peer says the bytes are gone. End the fetch.
    ///
    /// Checked the same way [`Self::answer_blob_request`] is, and for the
    /// same reason rather than a hypothetical one: a fetch is 1:1 state, and
    /// a frame arriving over a group would compose a key from a group id.
    /// Unreachable today because a group id cannot collide with an address,
    /// which is exactly the kind of accident a later refactor removes.
    fn report_blob_gone(&mut self, space: &str, sender: &str, hash: &str) -> Result<()> {
        if Self::validate_attachment_hash(hash).is_err() {
            warn!(space, "Blob refusal names something that is not a hash");
            return Ok(());
        }
        let key = Self::attachment_fetch_key(space, hash);
        // Only a fetch this device actually made is reported, and only by
        // the peer it was put to. Otherwise a peer could emit refusals for
        // blobs nobody asked about, and in a group any member could end a
        // question that was put to somebody else.
        let answers_us = self
            .pending_attachment_fetches
            .get(&key)
            .is_some_and(|pending| pending.is_answered_by(space, sender));
        if !answers_us {
            debug!(
                space,
                sender, "A blob refusal answers no question we put to this peer"
            );
            return Ok(());
        }
        self.pending_attachment_fetches.remove(&key);
        if let Ok(state) = crate::protocol::lock_shared_state(&self.shared_state) {
            state.emit_event(Event::DataAttachmentUnavailable {
                space_id: space.to_string(),
                peer_id: sender.to_string(),
                hash: hash.to_string(),
                reason: "declined".to_string(),
            });
        }
        Ok(())
    }

    // ---- what arrives over the media path ------------------------------

    /// A data-purposed media transfer completed. Route it by its purpose.
    ///
    /// The space is the authenticated wire sender, never anything the
    /// transfer said about itself, which is the same rule every sync frame
    /// follows and for the same reason: a space a peer can name is a space a
    /// peer can reach.
    ///
    /// Gated on the kill switch, which is belt to the storage seam's braces
    /// rather than the thing that stops an import: `require_data_storage`
    /// already fails closed with `DataDisabled`, so a device with the layer
    /// off writes nothing regardless. This makes the refusal local and free,
    /// and keeps the rule uniform across both roads into this layer. The
    /// admission check in `receive` is the one that earns its keep, because
    /// it is the one that refuses before a whole blob is buffered.
    pub(crate) fn accept_data_media_payload(
        &mut self,
        sender: &str,
        purpose: &DataPurpose,
        bytes: Vec<u8>,
    ) {
        if !self.config.data.enabled {
            debug!(
                peer = %sender,
                "Document-layer transfer dropped: the data layer is off"
            );
            return;
        }
        match purpose {
            DataPurpose::Attachment { hash } => {
                let key = Self::attachment_fetch_key(sender, hash);
                if self.pending_attachment_fetches.remove(&key).is_none() {
                    warn!(
                        peer = %sender,
                        "Attachment bytes arrived for a fetch this device never made; dropping"
                    );
                    return;
                }
                // The address is checked against the bytes, not against what
                // the sender says about them. This is what makes fetching
                // from an authenticated peer safe without trusting that
                // peer: the worst a wrong answer achieves is no answer.
                let actual = Self::data_attachment_hash(&bytes);
                if &actual != hash {
                    warn!(
                        peer = %sender,
                        "Attachment bytes do not hash to what was asked for; dropping"
                    );
                    if let Ok(state) = crate::protocol::lock_shared_state(&self.shared_state) {
                        state.emit_event(Event::DataAttachmentUnavailable {
                            space_id: sender.to_string(),
                            peer_id: sender.to_string(),
                            hash: hash.clone(),
                            reason: "hash_mismatch".to_string(),
                        });
                    }
                    return;
                }
                if let Ok(state) = crate::protocol::lock_shared_state(&self.shared_state) {
                    state.emit_event(Event::DataAttachmentReceived {
                        space_id: sender.to_string(),
                        peer_id: sender.to_string(),
                        hash: hash.clone(),
                        data: BASE64.encode(&bytes),
                    });
                }
            }
            DataPurpose::Snapshot { doc } => {
                if bytes.len() > MAX_MEDIA_SNAPSHOT_BYTES {
                    warn!(
                        peer = %sender,
                        bytes = bytes.len(),
                        "A carried snapshot is larger than a record can hold; refusing"
                    );
                    return;
                }
                // Straight into the containment every remote blob goes
                // through. Arriving by a longer road changes nothing about
                // what these bytes are: an import that can still end the
                // process, from a peer who is exactly who they say they are.
                let doc = doc.clone();
                if let Err(err) = self.apply_contained_blob(
                    sender,
                    &SyncChannel::Peer,
                    &doc,
                    bytes,
                    BlobKind::Snapshot,
                ) {
                    warn!(peer = %sender, doc, error = %err, "A carried snapshot was refused");
                }
            }
        }
    }

    /// A data-purposed media transfer failed. Say so where it can be acted on.
    pub(crate) fn report_data_media_failure(
        &mut self,
        sender: &str,
        purpose: &DataPurpose,
        reason: &str,
    ) {
        match purpose {
            DataPurpose::Attachment { hash } => {
                let key = Self::attachment_fetch_key(sender, hash);
                // Only a fetch still outstanding is reported, the same rule
                // the completion and refusal roads already follow. A fetch
                // that ended while its transfer was still moving (declined
                // mid-carriage, evicted, expired) has had its one answer,
                // and a second one for the same hash reads to an
                // application as a second failure of a thing it is no
                // longer waiting for.
                if self.pending_attachment_fetches.remove(&key).is_none() {
                    debug!(
                        peer = %sender,
                        reason,
                        "A transfer failed for a fetch that has already ended"
                    );
                    return;
                }
                if let Ok(state) = crate::protocol::lock_shared_state(&self.shared_state) {
                    state.emit_event(Event::DataAttachmentUnavailable {
                        space_id: sender.to_string(),
                        peer_id: sender.to_string(),
                        hash: hash.clone(),
                        reason: reason.to_string(),
                    });
                }
            }
            // Nothing to report to an application: a snapshot that did not
            // arrive is a gap anti-entropy will find again, and the next
            // exchange asks for it afresh.
            DataPurpose::Snapshot { doc } => {
                debug!(peer = %sender, doc, reason, "A carried snapshot did not arrive");
            }
        }
    }

    // ---- the crash record -------------------------------------------

    /// Read a space's replication bookkeeping, promoting anything left in
    /// flight by a previous run into the quarantine.
    fn load_sync_record(&mut self, space: &str) -> SyncRecord {
        let Some(storage) = self.data_storage_for_sync() else {
            return SyncRecord::default();
        };
        let mut record =
            match self.read_state_record(storage.as_ref(), storage_keys::DATA_SYNC, space) {
                Ok(Some(bytes)) => serde_json::from_slice::<SyncRecord>(&bytes).unwrap_or_default(),
                _ => SyncRecord::default(),
            };
        if let Some(digest) = record.in_flight.take() {
            warn!(
                space,
                "A previous run did not survive applying a remote change; quarantining it"
            );
            if !record.quarantined.contains(&digest) {
                record.quarantined.push(digest);
                while record.quarantined.len() > MAX_QUARANTINED_BLOBS {
                    record.quarantined.remove(0);
                }
            }
            self.persist_sync_record(space, &record);
        }
        record
    }

    fn persist_sync_record(&mut self, space: &str, record: &SyncRecord) {
        let Some(storage) = self.data_storage_for_sync() else {
            return;
        };
        let bytes = match serde_json::to_vec(record) {
            Ok(bytes) => bytes,
            Err(err) => {
                warn!(space, error = %err, "Could not encode replication bookkeeping");
                return;
            }
        };
        if let Err(err) =
            self.write_state_record(storage.as_ref(), storage_keys::DATA_SYNC, space, &bytes)
        {
            // Not fatal, and deliberately not a refusal to apply: failing to
            // write the marker costs the crash protection for this one blob,
            // while refusing would stop replication on a storage hiccup.
            warn!(space, error = %err, "Could not persist replication bookkeeping");
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn a_name_in_both_maps_costs_two_entries_of_the_frame_budget() {
        let entry = |n: usize, version: bool, floor: bool| -> OfferEntry {
            (
                format!("doc{n}"),
                version.then(|| "AQID".to_string()),
                floor.then(|| "BAUG".to_string()),
            )
        };

        let full: Vec<OfferEntry> = (0..MAX_DOCS_PER_VERSION_FRAME)
            .map(|n| entry(n, true, false))
            .collect();
        assert_eq!(offer_batches(&full).len(), 1);

        let one_over: Vec<OfferEntry> = (0..=MAX_DOCS_PER_VERSION_FRAME)
            .map(|n| entry(n, true, false))
            .collect();
        assert_eq!(offer_batches(&one_over).len(), 2);

        // Every name came back after a removal, so each costs the frame two
        // versions: half the budget in names is the whole budget in bytes.
        let dual: Vec<OfferEntry> = (0..=MAX_DOCS_PER_VERSION_FRAME / 2)
            .map(|n| entry(n, true, true))
            .collect();
        let batches = offer_batches(&dual);
        assert_eq!(
            batches.len(),
            2,
            "a frame of names in both maps was built to twice the byte budget"
        );
        assert_eq!(
            batches.iter().map(|batch| batch.len()).sum::<usize>(),
            dual.len(),
            "an entry was dropped or duplicated across the split"
        );

        assert_eq!(
            offer_batches(&[]).len(),
            1,
            "an empty offer is still one frame: it is what tells a peer to send everything"
        );
    }

    #[test]
    fn a_frame_carries_its_version_where_a_future_build_can_find_it() {
        // The version has to be readable off a frame whose body this build
        // cannot parse, or every future format looks like corruption to
        // every older install.
        let body = SyncBody::Delta {
            doc: "notes".to_string(),
            blob: "AAAA".to_string(),
        };
        let json = serde_json::to_string(&body).unwrap();
        let framed = format!("{{\"v\":{},{}", DATA_SYNC_V1, &json[1..]);

        let value: serde_json::Value = serde_json::from_str(&framed).unwrap();
        assert_eq!(value.get("v").and_then(|v| v.as_u64()), Some(1));
        assert_eq!(value.get("k").and_then(|v| v.as_str()), Some("delta"));

        let round_trip: SyncBody = serde_json::from_value(value).unwrap();
        assert!(matches!(round_trip, SyncBody::Delta { doc, .. } if doc == "notes"));
    }

    #[test]
    fn a_version_frame_round_trips() {
        let body = SyncBody::Versions {
            reply: true,
            partial: true,
            docs: BTreeMap::from([("notes".to_string(), "dgA=".to_string())]),
            gone: BTreeMap::from([("old".to_string(), "ZwA=".to_string())]),
            want: Some(vec!["notes".to_string(), "draft*".to_string()]),
        };
        let json = serde_json::to_string(&body).unwrap();
        let parsed: SyncBody = serde_json::from_str(&json).unwrap();
        match parsed {
            SyncBody::Versions {
                reply,
                partial,
                docs,
                gone,
                want,
            } => {
                assert!(reply);
                assert!(partial);
                assert_eq!(docs.get("notes").map(String::as_str), Some("dgA="));
                assert_eq!(gone.get("old").map(String::as_str), Some("ZwA="));
                assert_eq!(
                    want.as_deref(),
                    Some(["notes".to_string(), "draft*".to_string()].as_slice())
                );
            }
            other => panic!("expected a version frame, got {other:?}"),
        }
    }

    #[test]
    fn an_offer_with_no_removals_carries_no_removal_field() {
        // Byte-identical to what a build that predates removals sends, which
        // is what keeps every frozen vector above valid and every space
        // nobody has removed anything from as cheap as it was.
        let json = serde_json::to_string(&SyncBody::Versions {
            reply: false,
            partial: false,
            docs: BTreeMap::new(),
            gone: BTreeMap::new(),
            want: None,
        })
        .unwrap();
        assert_eq!(
            json,
            r#"{"k":"vv","reply":false,"partial":false,"docs":{}}"#
        );
    }

    #[test]
    fn an_offer_that_says_nothing_about_completeness_is_read_as_complete() {
        // The two flags default to the shape an ordinary whole-space offer
        // has, so a frame that omits them behaves like one. Getting this
        // backwards would silence the inference that carries a document a
        // peer has never seen.
        let parsed: SyncBody = serde_json::from_str(r#"{"k":"vv","docs":{}}"#).unwrap();
        match parsed {
            SyncBody::Versions {
                reply,
                partial,
                docs,
                gone,
                want,
            } => {
                assert!(!reply);
                assert!(!partial);
                assert!(docs.is_empty());
                assert!(
                    want.is_none(),
                    "an offer from a build that predates interest read as asking for a subset"
                );
                assert!(
                    gone.is_empty(),
                    "an offer from a build that predates removals read as removing something"
                );
            }
            other => panic!("expected a version frame, got {other:?}"),
        }
    }

    #[test]
    fn a_snapshot_request_round_trips_and_carries_no_bytes() {
        // It names a document and nothing else: the request exists because
        // the asker cannot express its own gap, so anything it added would
        // be the thing that was already refused.
        let body = SyncBody::NeedSnapshot {
            doc: "notes".to_string(),
        };
        let json = serde_json::to_string(&body).unwrap();
        assert_eq!(json, r#"{"k":"need_snap","doc":"notes"}"#);

        let parsed: SyncBody = serde_json::from_str(&json).unwrap();
        assert!(matches!(parsed, SyncBody::NeedSnapshot { doc } if doc == "notes"));
    }

    #[test]
    fn digests_separate_blobs_and_repeat_for_the_same_one() {
        assert_eq!(blob_digest(b"one"), blob_digest(b"one"));
        assert_ne!(blob_digest(b"one"), blob_digest(b"two"));
        assert_eq!(blob_digest(b"one").len(), 32);
    }
}

/// The frozen wire vectors for `__DATA_V1__`.
///
/// The conformance surface a second implementation is written against, and
/// the reason this layer is a protocol rather than an SDK feature. The file
/// was computed independently of the code it pins, so a mismatch is evidence
/// about the code rather than a disagreement between two copies of the same
/// mistake.
///
/// When one of these fails the wire format has changed. Bump the frame
/// version and negotiate a new `data_versions` entry. Do NOT edit the
/// expected values to make the test pass: every shipped install still speaks
/// the old ones, and a silent change here is a silent break there.
#[cfg(test)]
mod golden_vectors {
    use super::*;

    const VECTORS: &str = include_str!("../../tests/data/data-sync-v1.vectors.json");

    fn vectors() -> serde_json::Value {
        serde_json::from_str(VECTORS).expect("the vector file is JSON")
    }

    /// Frame the way [`OfflineProtocol::send_sync_frame`] does, which is the
    /// only place the version is spliced in. Reproduced rather than called
    /// because calling it needs a whole protocol, and what is pinned here is
    /// the encoding, not the send path.
    fn framed(body: &SyncBody) -> String {
        let json = serde_json::to_string(body).expect("encode");
        format!(
            "{}{{\"v\":{},{}",
            internal_prefixes::DATA_V1,
            DATA_SYNC_V1,
            &json[1..]
        )
    }

    fn wire(vectors: &serde_json::Value, name: &str) -> String {
        vectors["frames"]
            .as_array()
            .expect("frames")
            .iter()
            .find(|frame| frame["name"] == name)
            .unwrap_or_else(|| panic!("no vector named {name}"))["wire"]
            .as_str()
            .expect("wire is a string")
            .to_string()
    }

    #[test]
    fn the_vector_file_is_the_size_it_was() {
        // A file silently shortened by a bad merge would make every
        // assertion below pass by not running.
        //
        // Every array, not only the frames. The rest are read by `for`
        // loops and by index, and a loop over an emptied array is the same
        // green test as a loop over a full one. Counting them here is what
        // makes "the vectors pass" mean the vectors ran.
        let vectors = vectors();
        let len = |name: &str| -> usize {
            vectors[name]
                .as_array()
                .unwrap_or_else(|| panic!("{name} must be an array"))
                .len()
        };
        assert_eq!(len("frames"), 12);
        assert_eq!(len("parse_defaults"), 3);
        assert_eq!(len("attachment_hashes"), 3);
        assert_eq!(len("attachment_references"), 2);
        assert_eq!(len("data_purposes"), 2);
        assert_eq!(len("blob_digests"), 1);
        assert_eq!(vectors["version"].as_u64(), Some(DATA_SYNC_V1 as u64));
        assert_eq!(vectors["prefix"].as_str(), Some(internal_prefixes::DATA_V1));
    }

    #[test]
    fn every_frame_kind_encodes_to_its_vector() {
        let vectors = vectors();

        let mut docs = BTreeMap::new();
        docs.insert("notes".to_string(), "AQID".to_string());
        assert_eq!(
            framed(&SyncBody::Versions {
                reply: false,
                partial: false,
                docs,
                gone: BTreeMap::new(),
                want: None,
            }),
            wire(&vectors, "version offer, complete, asking")
        );

        // Inserted out of order on purpose: the map is ordered, so the wire
        // does not depend on what order a sender happened to hold them in.
        let mut docs = BTreeMap::new();
        docs.insert("b".to_string(), "Zg==".to_string());
        docs.insert("a".to_string(), String::new());
        assert_eq!(
            framed(&SyncBody::Versions {
                reply: true,
                partial: true,
                docs,
                gone: BTreeMap::new(),
                want: None,
            }),
            wire(&vectors, "version offer, answering leg")
        );

        assert_eq!(
            framed(&SyncBody::Versions {
                reply: false,
                partial: false,
                docs: BTreeMap::new(),
                gone: BTreeMap::new(),
                want: None,
            }),
            wire(&vectors, "version offer with no documents")
        );

        let mut docs = BTreeMap::new();
        docs.insert("kept".to_string(), "AQID".to_string());
        let mut gone = BTreeMap::new();
        gone.insert("removed".to_string(), "BAUG".to_string());
        assert_eq!(
            framed(&SyncBody::Versions {
                reply: false,
                partial: false,
                docs,
                gone,
                want: None,
            }),
            wire(&vectors, "version offer carrying a removal")
        );

        let mut docs = BTreeMap::new();
        docs.insert("notes".to_string(), "AQID".to_string());
        assert_eq!(
            framed(&SyncBody::Versions {
                reply: false,
                partial: false,
                docs,
                gone: BTreeMap::new(),
                want: Some(vec!["notes".to_string(), "draft*".to_string()]),
            }),
            wire(&vectors, "version offer declaring an interest")
        );

        let blob = BASE64.encode([0u8, 1, 2, 3, 4, 5, 6, 7]);
        assert_eq!(
            framed(&SyncBody::Delta {
                doc: "notes".to_string(),
                blob: blob.clone(),
            }),
            wire(&vectors, "delta")
        );
        assert_eq!(
            framed(&SyncBody::Snapshot {
                doc: "notes".to_string(),
                blob,
            }),
            wire(&vectors, "snapshot")
        );
        assert_eq!(
            framed(&SyncBody::NeedSnapshot {
                doc: "notes".to_string(),
            }),
            wire(&vectors, "snapshot request")
        );

        let hash = vectors["attachment_hashes"][0]["sha256"]
            .as_str()
            .expect("hash")
            .to_string();
        assert_eq!(
            framed(&SyncBody::NeedBlob {
                hash: hash.clone(),
                to: None,
            }),
            wire(&vectors, "blob request")
        );
        assert_eq!(
            framed(&SyncBody::BlobGone { hash: hash.clone() }),
            wire(&vectors, "blob refusal")
        );

        assert_eq!(
            framed(&SyncBody::NeedBlob {
                hash: hash.clone(),
                to: Some("off1qy60a3pu0l9tnthnk08c4w59teq7u6w28gjwg8zk".to_string()),
            }),
            wire(&vectors, "blob request addressed to one member")
        );

        assert_eq!(
            framed(&SyncBody::Chunk {
                hash,
                i: 0,
                n: 2,
                blob: BASE64.encode([0u8, 1, 2, 3]),
            }),
            wire(&vectors, "blob chunk")
        );
    }

    #[test]
    fn every_vector_parses_back_into_the_kind_it_names() {
        // The other direction, and the one a second implementation's sender
        // is judged by. Encoding-only vectors would let a decoder drift.
        let vectors = vectors();
        for frame in vectors["frames"].as_array().expect("frames") {
            let wire = frame["wire"].as_str().expect("wire");
            let body = wire
                .strip_prefix(internal_prefixes::DATA_V1)
                .expect("every vector carries the marker");
            let value: serde_json::Value = serde_json::from_str(body).expect("the body is JSON");
            assert_eq!(
                value["v"].as_u64(),
                Some(DATA_SYNC_V1 as u64),
                "{}: version",
                frame["name"]
            );
            assert_eq!(
                value["k"].as_str(),
                frame["kind"].as_str(),
                "{}: kind",
                frame["name"]
            );
            let parsed: SyncBody =
                serde_json::from_value(value).expect("the body parses into a frame");
            let kind = match parsed {
                SyncBody::Versions { .. } => "vv",
                SyncBody::Delta { .. } => "delta",
                SyncBody::Snapshot { .. } => "snap",
                SyncBody::NeedSnapshot { .. } => "need_snap",
                SyncBody::NeedBlob { .. } => "blob",
                SyncBody::BlobGone { .. } => "gone",
                SyncBody::Chunk { .. } => "chunk",
            };
            let expected = match frame["kind"].as_str().expect("kind") {
                "need_blob" => "blob",
                "blob_gone" => "gone",
                other => other,
            };
            assert_eq!(kind, expected, "{}: round trip", frame["name"]);
        }
    }

    #[test]
    fn absent_flags_parse_as_false() {
        let vectors = vectors();
        for case in vectors["parse_defaults"]
            .as_array()
            .expect("parse_defaults")
        {
            let value: serde_json::Value =
                serde_json::from_str(case["body"].as_str().expect("body")).expect("JSON");
            let SyncBody::Versions { reply, partial, .. } =
                serde_json::from_value(value).expect("parses")
            else {
                panic!("{}: expected a version offer", case["name"]);
            };
            assert_eq!(reply, case["expect"]["reply"].as_bool().expect("reply"));
            assert_eq!(
                partial,
                case["expect"]["partial"].as_bool().expect("partial")
            );
        }
    }

    #[test]
    fn an_absent_interest_parses_as_asking_for_everything() {
        // Absent and empty are opposite answers, and the frozen cases pin
        // the one a build that predates the field sends.
        let vectors = vectors();
        for case in vectors["parse_defaults"]
            .as_array()
            .expect("parse_defaults")
        {
            let value: serde_json::Value =
                serde_json::from_str(case["body"].as_str().expect("body")).expect("JSON");
            let SyncBody::Versions { want, .. } = serde_json::from_value(value).expect("parses")
            else {
                panic!("{}: expected a version offer", case["name"]);
            };
            assert!(
                case["expect"]["want"].is_null(),
                "{}: this case is about an absent interest",
                case["name"]
            );
            assert!(want.is_none(), "{}", case["name"]);
        }
    }

    #[test]
    fn an_empty_interest_is_not_an_absent_one() {
        // The distinction the `Option` exists for: collapsing them would
        // make "send me nothing" read as "send me everything".
        let asks_for_nothing = serde_json::to_string(&SyncBody::Versions {
            reply: false,
            partial: false,
            docs: BTreeMap::new(),
            gone: BTreeMap::new(),
            want: Some(Vec::new()),
        })
        .unwrap();
        assert!(asks_for_nothing.contains(r#""want":[]"#));

        let parsed: SyncBody = serde_json::from_str(&asks_for_nothing).unwrap();
        let SyncBody::Versions { want, .. } = parsed else {
            panic!("expected a version offer");
        };
        assert_eq!(want, Some(Vec::new()));
    }

    #[test]
    fn an_absent_removal_map_parses_as_removing_nothing() {
        let vectors = vectors();
        for case in vectors["parse_defaults"]
            .as_array()
            .expect("parse_defaults")
        {
            let value: serde_json::Value =
                serde_json::from_str(case["body"].as_str().expect("body")).expect("JSON");
            let SyncBody::Versions { gone, .. } = serde_json::from_value(value).expect("parses")
            else {
                panic!("{}: expected a version offer", case["name"]);
            };
            let expected = case["expect"]["gone"].as_object().expect("gone");
            assert_eq!(gone.len(), expected.len(), "{}", case["name"]);
        }
    }

    #[test]
    fn attachment_hashes_match_their_vectors() {
        let vectors = vectors();
        for case in vectors["attachment_hashes"]
            .as_array()
            .expect("attachment_hashes")
        {
            let bytes = match (case.get("input_utf8"), case.get("input_hex")) {
                (Some(text), _) => text.as_str().expect("utf8").as_bytes().to_vec(),
                (_, Some(hex)) => {
                    let hex = hex.as_str().expect("hex");
                    (0..hex.len())
                        .step_by(2)
                        .map(|i| u8::from_str_radix(&hex[i..i + 2], 16).expect("hex byte"))
                        .collect()
                }
                _ => panic!("a hash vector needs an input"),
            };
            assert_eq!(
                bytes.len() as u64,
                case["size"].as_u64().expect("size"),
                "the vector's own size must describe its own input"
            );
            assert_eq!(
                OfflineProtocol::data_attachment_hash(&bytes),
                case["sha256"].as_str().expect("sha256").to_string()
            );
        }
    }

    #[test]
    fn attachment_references_encode_to_their_vectors() {
        use offline_protocol_data::DataValue;
        let vectors = vectors();
        let hash = vectors["attachment_hashes"][0]["sha256"]
            .as_str()
            .expect("hash")
            .to_string();
        let size = vectors["attachment_hashes"][0]["size"]
            .as_u64()
            .expect("size");

        let full = DataValue::Attachment {
            hash: hash.clone(),
            size,
            name: Some("plan.pdf".to_string()),
            mime: Some("application/pdf".to_string()),
        };
        assert_eq!(
            serde_json::to_string(&full).expect("encode"),
            vectors["attachment_references"][0]["json"]
                .as_str()
                .expect("json")
        );

        // Absent, not null. A reader that has to tell one from the other has
        // been given two spellings of one value.
        let bare = DataValue::attachment(hash, size);
        assert_eq!(
            serde_json::to_string(&bare).expect("encode"),
            vectors["attachment_references"][1]["json"]
                .as_str()
                .expect("json")
        );
        assert!(!vectors["attachment_references"][1]["json"]
            .as_str()
            .expect("json")
            .contains("null"));
    }

    #[test]
    fn data_purposes_encode_to_their_vectors() {
        let vectors = vectors();
        let hash = vectors["attachment_hashes"][0]["sha256"]
            .as_str()
            .expect("hash")
            .to_string();
        assert_eq!(
            serde_json::to_string(&DataPurpose::Attachment { hash }).expect("encode"),
            vectors["data_purposes"][0]["json"].as_str().expect("json")
        );
        assert_eq!(
            serde_json::to_string(&DataPurpose::Snapshot {
                doc: "notes".to_string(),
            })
            .expect("encode"),
            vectors["data_purposes"][1]["json"].as_str().expect("json")
        );
    }

    #[test]
    fn blob_digests_match_their_vectors() {
        let vectors = vectors();
        for case in vectors["blob_digests"].as_array().expect("blob_digests") {
            let hex = case["input_hex"].as_str().expect("hex");
            let bytes: Vec<u8> = (0..hex.len())
                .step_by(2)
                .map(|i| u8::from_str_radix(&hex[i..i + 2], 16).expect("hex byte"))
                .collect();
            assert_eq!(
                blob_digest(&bytes),
                case["digest"].as_str().expect("digest")
            );
        }
    }
}

impl SyncBody {
    /// The custody class of this frame, or `None` for a kind custody never
    /// carries (`docs/spec/custody.md`, "The classes").
    ///
    /// Exhaustive on purpose: a new kind has to decide here whether a copy of
    /// it is idempotent in cost, rather than inheriting an answer.
    pub(crate) fn custody_class(&self) -> Option<&'static str> {
        match self {
            SyncBody::Versions { .. }
            | SyncBody::Delta { .. }
            | SyncBody::Snapshot { .. }
            | SyncBody::BlobGone { .. } => Some(crate::protocol::custody::CUSTODY_CLASS_DATA),
            // Not idempotent in cost: each acted-on copy of a request spends a
            // whole media transfer or a snapshot export, and a carried chunk
            // repeats an answer the requester already had.
            SyncBody::NeedSnapshot { .. } | SyncBody::NeedBlob { .. } | SyncBody::Chunk { .. } => {
                None
            }
        }
    }
}

/// The custody class of a `__DATA_V1__` plaintext the depositor retains for
/// re-sealing, or `None` when it is not a sync frame this build reads, or is a
/// kind custody refuses by name.
///
/// Read the way an arriving frame is read: version before body, so a frame of
/// a future version is simply not deposited rather than misclassified.
pub(crate) fn custody_class_of_plaintext(plaintext: &str) -> Option<&'static str> {
    let body = plaintext.strip_prefix(internal_prefixes::DATA_V1)?;
    let value: serde_json::Value = serde_json::from_str(body).ok()?;
    if value.get("v").and_then(serde_json::Value::as_u64) != Some(u64::from(DATA_SYNC_V1)) {
        return None;
    }
    serde_json::from_value::<SyncBody>(value)
        .ok()?
        .custody_class()
}

#[cfg(test)]
mod custody_class_tests {
    use super::*;

    fn framed(body: &str) -> String {
        format!(
            "{}{{\"v\":{},{}",
            internal_prefixes::DATA_V1,
            DATA_SYNC_V1,
            &body[1..]
        )
    }

    #[test]
    fn class_a_kinds_are_deposited_and_the_rest_are_not() {
        for (kind, expected) in [
            (
                "{\"k\":\"delta\",\"doc\":\"d\",\"blob\":\"AA==\"}",
                Some("data"),
            ),
            (
                "{\"k\":\"snap\",\"doc\":\"d\",\"blob\":\"AA==\"}",
                Some("data"),
            ),
            ("{\"k\":\"vv\",\"docs\":{}}", Some("data")),
            ("{\"k\":\"blob_gone\",\"hash\":\"h\"}", Some("data")),
            ("{\"k\":\"need_snap\",\"doc\":\"d\"}", None),
            ("{\"k\":\"need_blob\",\"hash\":\"h\"}", None),
            (
                "{\"k\":\"chunk\",\"hash\":\"h\",\"i\":0,\"n\":1,\"blob\":\"AA==\"}",
                None,
            ),
        ] {
            assert_eq!(
                custody_class_of_plaintext(&framed(kind)),
                expected,
                "{kind}"
            );
        }
    }

    #[test]
    fn anything_but_a_readable_sync_frame_is_not_deposited() {
        assert_eq!(custody_class_of_plaintext("hello"), None);
        assert_eq!(
            custody_class_of_plaintext(&format!("{}not json", internal_prefixes::DATA_V1)),
            None
        );
        // A future version is not deposited rather than misclassified.
        let future = format!(
            "{}{{\"v\":{},\"k\":\"delta\",\"doc\":\"d\",\"blob\":\"AA==\"}}",
            internal_prefixes::DATA_V1,
            DATA_SYNC_V1 + 1
        );
        assert_eq!(custody_class_of_plaintext(&future), None);
    }
}