//! One Automerge sync with a peer, from a reset to the next.
//!
//! Automerge's sync state (`automerge::sync::State`) is private to this module: the
//! rest of the node generates, receives and inspects through `SyncSession`'s methods.
//! The workarounds for its assumption of reliable delivery (resending after a timeout,
//! forgetting what was sent when the peer still needs it) live here with it, so an
//! Automerge upgrade that changes the state touches this file only.

use std::collections::HashSet;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Duration;

use automerge::AutomergeError;
use automerge::sync::{self, SyncDoc};
use tokio::time::Instant;
use tracing::debug;

use super::{Document, ReadDoc};
use crate::backoff::Backoff;
use crate::swarm::throughput::SyncBytes;
use crate::transport::SyncPayload;

/// A sync message from a peer, decoded (`decode`), to apply (`SyncSession::receive`).
pub(crate) struct Received(sync::Message);

/// The changes none of the others depend on, sorted as heads are (`get_heads`).
fn tips<'a>(
    changes: impl Iterator<
        Item = (
            crate::document::ChangeHash,
            &'a [crate::document::ChangeHash],
        ),
    >,
) -> Vec<crate::document::ChangeHash> {
    let changes: Vec<_> = changes.collect();
    let depended: HashSet<crate::document::ChangeHash> = changes
        .iter()
        .flat_map(|(_, deps)| deps.iter().copied())
        .collect();
    let mut tips: Vec<crate::document::ChangeHash> = changes
        .into_iter()
        .map(|(hash, _)| hash)
        .filter(|hash| !depended.contains(hash))
        .collect();
    tips.sort_unstable();
    tips
}

impl Received {
    /// Whether it carries changes (it may carry only heads and what the peer needs).
    pub(crate) fn carries_changes(&self) -> bool {
        !self.0.changes.is_empty()
    }

    /// Changes pushed along a writer's tree (`SyncBody::Changes`), as the Automerge sync
    /// message that would have brought them: the sender's `heads`, nothing needed, and
    /// as what it has, the history of those heads (nothing beyond them: its heads are
    /// all it has). Where the changes wait for predecessors and some of those heads stay
    /// unknown, they give way to the heads we share (`known_last_sync`), as for an
    /// overtaken message.
    ///
    /// No heads stand for the tips of the changes (`SyncSession::push`). None without
    /// changes, or without heads where a change doesn't decode: no heads would say the
    /// sender lost all it had, and we'd send it everything again.
    pub(crate) fn pushed(
        heads: Vec<crate::document::ChangeHash>,
        changes: Vec<Vec<u8>>,
    ) -> Option<Self> {
        if changes.is_empty() {
            return None;
        }
        let heads = if heads.is_empty() {
            let decoded = changes
                .iter()
                .map(|bytes| automerge::Change::try_from(bytes.as_slice()))
                .collect::<Result<Vec<_>, _>>()
                .ok()?;
            tips(decoded.iter().map(|change| (change.hash(), change.deps())))
        } else {
            heads
        };
        Some(Self(sync::Message {
            heads: heads.clone(),
            need: Vec::new(),
            have: vec![sync::Have {
                last_sync: heads,
                bloom: sync::BloomFilter::default(),
            }],
            changes: changes.into(),
            flags: None,
            version: sync::MessageVersion::V2,
        }))
    }

    /// The sender's heads, as it says.
    pub(crate) fn heads(&self) -> &[crate::document::ChangeHash] {
        &self.0.heads
    }

    /// The bytes of the changes it carries.
    pub(crate) fn change_bytes(&self) -> usize {
        self.0.changes.iter().map(<[u8]>::len).sum()
    }
}

/// `payload` decoded. Before taking the document's lock: decoding a large message takes
/// a while.
pub(crate) fn decode(payload: &SyncPayload) -> anyhow::Result<Received> {
    sync::Message::decode(&payload.0)
        .map(Received)
        .map_err(|error| anyhow::anyhow!("Automerge decode failed: {error:?}"))
}

/// A sync message being handed to our transport (for Mycelium: the HTTP request is in
/// progress).
#[derive(Clone)]
pub(crate) struct Sending {
    /// Changes it carries for the first time; taken back if the send fails.
    pub(crate) sent: Vec<crate::document::ChangeHash>,
    /// A sync pass was requested meanwhile. Once this one is handed over, the peer must
    /// be enqueued again so the deferred changes go out.
    pub(crate) regenerate: bool,
    /// Taken as delivered (`assume_delivered`): what we knew of the peer before, put
    /// back if the send fails.
    pub(crate) assumed: Option<PeerView>,
}

/// What the sync state knows of the peer: its heads, the heads we share, what it has
/// and needs. Saved to be put back: by `assume_delivered` if the send fails, and before
/// receiving, for a message older than the last (`answered_as_delivered`).
#[derive(Clone)]
pub(crate) struct PeerView {
    their_heads: Option<Vec<crate::document::ChangeHash>>,
    shared_heads: Vec<crate::document::ChangeHash>,
    their_have: Option<Vec<sync::Have>>,
    their_need: Option<Vec<crate::document::ChangeHash>>,
}

/// Session IDs, unique within the process.
static NEXT_SESSION: AtomicU64 = AtomicU64::new(1);

/// First wait before generating again after our transport refused a message; doubles
/// while it keeps refusing.
const SEND_RETRY: Duration = Duration::from_millis(500);
const SEND_RETRY_MAX: Duration = Duration::from_secs(30);

/// One Automerge sync with a peer, from a reset to the next. Everything a reset must
/// forget lives here, so a reset replaces it whole and can't miss a field.
#[derive(Clone)]
pub(crate) struct SyncSession {
    /// Tells this session from the one a reset puts in its place (`NEXT_SESSION`).
    id: u64,
    state: sync::State,

    /// A message being handed to our transport.
    pub(crate) sending: Option<Sending>,

    /// After our transport refused a message: no new one before this. The refused one is
    /// not kept; the next one is generated afresh and carries everything pending.
    pub(crate) retry_at: Option<Instant>,
    /// The wait after the next refusal.
    pub(crate) send_backoff: Backoff,

    /// We offered the peer a delta instead of a message too large to send
    /// (`withdraw_for_delta`): no more messages until it answers, or the answer is
    /// overdue (then the offer goes out again).
    pub(crate) delta_offered: bool,
    /// The delta last offered: its heads (sorted, as `DeltaSpec`), and the size of the
    /// message it replaced (`offer_again`).
    last_offer: Option<(
        Vec<crate::document::ChangeHash>,
        Vec<crate::document::ChangeHash>,
        usize,
    )>,
    /// What the sync state knew of the peer before the last message from it
    /// (`answered_as_delivered`).
    before_receive: Option<PeerView>,
    /// What the last `generate` changed in the sync state besides the hashes sent: the
    /// heads it last sent, and whether it had answered yet. Restored when the message is
    /// withdrawn for a delta: otherwise Automerge believes the peer heard our heads, and
    /// once the peer has the delta and says its heads are ours, never answers it, so the
    /// peer never learns our heads (a joining node then never finishes its initial
    /// sync).
    before_generate: (Vec<crate::document::ChangeHash>, bool),
    /// Changes counted as sent because another parent brings them (`known_to_peer`),
    /// and when: still in flight perhaps, so they go along with a change of ours that
    /// builds on them (`with_predecessors`).
    known_elsewhere: std::collections::HashMap<crate::document::ChangeHash, Instant>,

    /// The heads the peer said last, in the newest of its messages so far (one that
    /// comes late doesn't count: `older_keeps_newer`).
    said_heads: Option<Vec<crate::document::ChangeHash>>,

    /// The answer we expect to our last sync message(s), if any.
    pub(crate) reply: Option<SyncReplyWait>,
    /// How long to wait for it; doubles while answers stay out.
    pub(crate) reply_backoff: Backoff,
}

/// How long a change counted as sent because another parent brings it may still be in
/// flight to the peer (`SyncSession::with_predecessors`): a slow round trip, and more.
const KNOWN_ELSEWHERE_FOR: Duration = Duration::from_secs(2);
/// How long such a change is remembered as counted so: to tell, should the peer ask for
/// it, how long after (`SyncSession::resend_if_needed`).
const KNOWN_ELSEWHERE_KEPT: Duration = Duration::from_secs(30);

/// Sync messages we sent and expect an answer to. Automerge assumes reliable delivery:
/// a lost message leaves both sides waiting (`in_flight`), and changes it carried are
/// never sent again (`sent_hashes`). We resend after a timeout instead.
#[derive(Clone, Debug)]
pub(crate) struct SyncReplyWait {
    pub(crate) deadline: Instant,
    /// When the first of the unanswered messages was generated, and the sync bytes
    /// exchanged with the peer by then (`swarm::throughput`).
    pub(crate) started: (Instant, SyncBytes),
    /// Changes first sent in the unanswered messages.
    pub(crate) sent: Vec<crate::document::ChangeHash>,
    /// The largest of them, in bytes: if they go unanswered, messages that large may
    /// not make it to the peer whole (`PeerState::whole_failed`).
    pub(crate) largest: usize,
}

impl SyncSession {
    /// `reply_backoff`: waits for sync answers (`Timing::sync_reply_backoff`).
    pub(crate) fn new(reply_backoff: Backoff) -> Self {
        Self {
            id: NEXT_SESSION.fetch_add(1, Ordering::Relaxed),
            state: sync::State::new(),
            sending: None,
            retry_at: None,
            send_backoff: Backoff::new(SEND_RETRY, SEND_RETRY_MAX),
            delta_offered: false,
            last_offer: None,
            before_receive: None,
            before_generate: (Vec::new(), false),
            known_elsewhere: std::collections::HashMap::new(),
            said_heads: None,
            reply: None,
            reply_backoff,
        }
    }

    /// Whether the peer said anything in this session yet (its heads are known).
    pub(crate) fn has_opened(&self) -> bool {
        self.state.their_heads.is_some()
    }

    /// The next sync message for the peer from `doc`, encoded, if there is anything to say. A
    /// message is marked as being handed over (`sending`), and an answer to it is
    /// expected, unless the peer's heads are ours already: it has everything, and
    /// Automerge answers only when it has something to say, so waiting for an answer
    /// would only end in a pointless resend. `bytes`: the sync bytes exchanged with the
    /// peer so far (`swarm::throughput`).
    pub(crate) fn generate(
        &mut self,
        doc: &Document,
        now: Instant,
        bytes: SyncBytes,
    ) -> Option<SyncPayload> {
        let sent_before = self.state.sent_hashes.clone();
        self.before_generate = (
            self.state.last_sent_heads.clone(),
            self.state.have_responded,
        );
        let message = doc.generate_sync_message(&mut self.state);
        let in_step = self.state.their_heads.as_ref() == Some(&doc.get_heads());
        let sent: Vec<_> = self
            .state
            .sent_hashes
            .difference(&sent_before)
            .copied()
            .collect();
        let payload = message.map(|message| SyncPayload(message.encode().into()));
        if in_step {
            self.reply_settled();
        } else if let Some(payload) = &payload {
            self.expect_reply(sent.clone(), now, bytes, payload.0.len());
        }
        if payload.is_some() {
            // Marked together with the generation, so no second message is generated
            // while this one is handed over.
            self.sending = Some(Sending {
                sent,
                regenerate: false,
                assumed: None,
            });
        }
        payload
    }

    /// Applies a sync message from the peer to `doc`. Doesn't settle the awaited
    /// answer: the caller measures the exchange first (`reply`).
    #[cfg(test)]
    pub(crate) fn receive(
        &mut self,
        doc: &mut Document,
        message: Received,
    ) -> Result<(), AutomergeError> {
        self.receive_logged(doc, message, &mut automerge::PatchLog::inactive())
    }

    /// Applies a sync message from the peer to `doc`, what it changes recorded in `log`
    /// (`Replica::logged`). Doesn't settle the awaited answer: the caller measures the
    /// exchange first (`reply`).
    pub(crate) fn receive_logged(
        &mut self,
        doc: &mut Document,
        message: Received,
        log: &mut automerge::PatchLog,
    ) -> Result<(), AutomergeError> {
        self.delta_offered = false;
        self.before_receive = Some(self.peer_view());
        let said = message.0.heads.clone();
        doc.receive_sync_message_log_patches(&mut self.state, message.0, log)?;
        self.older_keeps_newer(doc, &said);
        self.known_last_sync(doc);
        Ok(())
    }

    /// A message older than one the peer sent before (it came late: overtaken, or a
    /// copy) names heads in the history of those it said then: Automerge takes them as
    /// the peer's now, and what came since would seem to lack to it, the changes it
    /// sent us among them (sent back to it, by a push above all: `push` sends what lies
    /// beyond its heads). So the peer's heads, and those we share, are what it said then
    /// (`said_heads`). Only by what it said: heads we took for its (a message of ours
    /// taken as delivered, `assume_delivered`) may be more than it has, and a peer
    /// back from away says older heads than those. Not where it says it has nothing (no
    /// heads: it lost everything).
    fn older_keeps_newer(&mut self, doc: &Document, said: &[crate::document::ChangeHash]) {
        let newer = self.said_heads.clone();
        let older = newer.as_ref().is_some_and(|newer| {
            // Older: all its heads in our history, and in that of the heads said before
            // (a message whose changes wait for predecessors names heads we lack:
            // newer).
            !said.is_empty()
                && said != newer.as_slice()
                && crate::document::changes::holds_all(doc, said)
                && crate::document::changes::holds_all(doc, newer)
                && crate::document::changes::covers(doc, newer, said)
        });
        match newer {
            Some(newer) if older => {
                self.state.their_heads = Some(newer.clone());
                self.state.shared_heads = newer;
            }
            _ => self.said_heads = Some(said.to_vec()),
        }
    }

    /// The peer's `have` names what it last shared with us by heads (`last_sync`), and
    /// one that took a message to us as delivered names the heads that message brought.
    /// Where the next message overtook it, we don't have those heads yet: Automerge
    /// would take that for a document that lost changes and answer with a reset, and
    /// the peer would forget all it ever sent us and send it again, what is still on its
    /// way included (on slow links, every change several times). So heads in it that we
    /// lack give way to the heads we share with the peer (`shared_heads`). The bloom
    /// filter stays: it covers the changes since its own heads, which leaves out at
    /// worst a few that go again.
    fn known_last_sync(&mut self, doc: &Document) {
        let Some(haves) = &mut self.state.their_have else {
            return;
        };
        for have in haves.iter_mut() {
            if have
                .last_sync
                .iter()
                .all(|hash| doc.get_change_by_hash(hash).is_some())
            {
                continue;
            }
            let mut known: Vec<crate::document::ChangeHash> = have
                .last_sync
                .iter()
                .filter(|hash| doc.get_change_by_hash(hash).is_some())
                .chain(self.state.shared_heads.iter())
                .copied()
                .collect();
            known.sort_unstable();
            known.dedup();
            have.last_sync = known;
        }
    }

    /// Instead of `generate`, where the message would be taken as delivered (the peer
    /// has nothing we lack): the changes it would carry, as they are, with our heads
    /// (`SyncBody::Changes`). Those since the peer's heads that don't count as sent
    /// (as Automerge picks them; `known_to_peer`, `with_predecessors` alike), counted as
    /// sent now, the message marked as handed over as `generate` does; the caller takes
    /// it as delivered. None if there is nothing to push, the peer's heads aren't all
    /// known, or the changes come to more than `limit` bytes (then `generate`).
    pub(crate) fn push(
        &mut self,
        doc: &Document,
        limit: usize,
    ) -> Option<(Vec<crate::document::ChangeHash>, Vec<Vec<u8>>)> {
        let theirs = self.state.their_heads.as_ref()?;
        if theirs
            .iter()
            .any(|hash| doc.get_change_by_hash(hash).is_none())
        {
            return None;
        }
        let changes: Vec<automerge::Change> = doc
            .get_changes(theirs)
            .into_iter()
            .filter(|change| !self.state.sent_hashes.contains(&change.hash()))
            .collect();
        let bytes: usize = changes.iter().map(|change| change.raw_bytes().len()).sum();
        if changes.is_empty() || bytes > limit {
            return None;
        }
        let sent: Vec<crate::document::ChangeHash> =
            changes.iter().map(automerge::Change::hash).collect();
        self.before_generate = (
            self.state.last_sent_heads.clone(),
            self.state.have_responded,
        );
        self.state.sent_hashes.extend(sent.iter().copied());
        let heads = doc.get_heads();
        self.state.last_sent_heads = heads.clone();
        self.state.have_responded = true;
        self.sending = Some(Sending {
            sent,
            regenerate: false,
            assumed: None,
        });
        // Our heads are mostly the tips of the changes pushed: then they go as none, and
        // the receiver finds them in the changes (`Received::pushed`).
        let tips = tips(changes.iter().map(|change| (change.hash(), change.deps())));
        Some((
            if tips == heads { Vec::new() } else { heads },
            changes
                .iter()
                .map(|change| change.raw_bytes().to_vec())
                .collect(),
        ))
    }

    /// The message just generated is too large to send in one piece: take back what it
    /// carried, as if it had never been generated, and offer a delta instead. The
    /// message being handed over becomes the offer (it carries nothing); the answer is
    /// still awaited, and if it doesn't come, the offer is made again. Returns where
    /// the delta starts (`delta_from`).
    ///
    /// An answer is awaited even where none was (the peer's heads were ours: `generate`),
    /// so the offer lapses at the reply timeout at the latest: nothing else ends it but a
    /// message from the peer, and one that has everything has nothing to say. Without
    /// that, no message would go to it again.
    pub(crate) fn withdraw_for_delta(
        &mut self,
        doc: &Document,
        now: Instant,
    ) -> Vec<crate::document::ChangeHash> {
        if self.reply.is_none() {
            self.expect_reply(Vec::new(), now, SyncBytes::default(), 0);
        }
        if let Some(sending) = &mut self.sending {
            let sent = std::mem::take(&mut sending.sent);
            for hash in &sent {
                self.state.sent_hashes.remove(hash);
            }
            if let Some(wait) = &mut self.reply {
                wait.sent.retain(|hash| !sent.contains(hash));
            }
        }
        self.state.in_flight = false;
        (self.state.last_sent_heads, self.state.have_responded) = self.before_generate.clone();
        self.delta_offered = true;
        self.delta_from(doc)
    }

    /// Where a delta to the peer starts, sorted: its heads as far as we know them, of
    /// those we hold. Every node that serves the delta must hold all of them
    /// (`document::changes::delta_bytes`), so the same bytes come from each; we do. In
    /// place of heads we lack (changes it has and we don't), the heads we share: the
    /// delta may then carry some it has, but nothing it lacks.
    fn delta_from(&self, doc: &Document) -> Vec<crate::document::ChangeHash> {
        let known = |hash: &&crate::document::ChangeHash| doc.get_change_by_hash(hash).is_some();
        let theirs = self.state.their_heads.as_deref().unwrap_or_default();
        let mut from: Vec<_> = theirs.iter().filter(known).copied().collect();
        if from.len() < theirs.len() {
            from.extend(self.state.shared_heads.iter().filter(known).copied());
        }
        from.sort_unstable();
        from.dedup();
        from
    }

    /// We offered the delta from `from` to `to` in place of a message of `bytes`.
    pub(crate) fn offered(
        &mut self,
        from: &[crate::document::ChangeHash],
        to: &[crate::document::ChangeHash],
        bytes: usize,
    ) {
        self.last_offer = Some((from.to_vec(), to.to_vec(), bytes));
    }

    /// The delta to offer again, if nothing changed since the last offer (`offered`):
    /// where it starts (`delta_from`), our `heads`, and the message too large for
    /// `threshold` still. Then the message would be the same, and so the delta: its
    /// `from`, offered as by `withdraw_for_delta`, without generating the message.
    pub(crate) fn offer_again(
        &mut self,
        doc: &Document,
        heads: &[crate::document::ChangeHash],
        threshold: usize,
        now: Instant,
    ) -> Option<Vec<crate::document::ChangeHash>> {
        let (from, to, bytes) = self.last_offer.as_ref()?;
        let sorted = |heads: &[crate::document::ChangeHash]| {
            let mut heads = heads.to_vec();
            heads.sort_unstable();
            heads.dedup();
            heads
        };
        if *bytes <= threshold || self.delta_from(doc) != *from || sorted(heads) != *to {
            return None;
        }
        let from = from.clone();
        if self.reply.is_none() {
            self.expect_reply(Vec::new(), now, SyncBytes::default(), 0);
        }
        // Handed over like a generated message (`send_failed` then retries).
        self.sending = Some(Sending {
            sent: Vec::new(),
            regenerate: false,
            assumed: None,
        });
        self.delta_offered = true;
        Some(from)
    }

    /// The message just generated, withdrawn for a delta (`withdraw_for_delta`), turns
    /// out to carry nothing the peer lacks (it has our heads): none goes, and nothing is
    /// offered or awaited.
    pub(crate) fn nothing_to_send(&mut self) {
        self.sending = None;
        self.delta_offered = false;
        self.reply_settled();
    }

    /// Whether `doc` holds everything the peer had when it last told us its heads: they
    /// are known, not empty (empty heads are "contained" in every document, so they
    /// prove nothing), and all in `doc`.
    pub(crate) fn have_all_theirs(&self, doc: &Document) -> bool {
        self.state.their_heads.as_ref().is_some_and(|heads| {
            !heads.is_empty()
                && heads
                    .iter()
                    .all(|head| doc.get_change_by_hash(head).is_some())
        })
    }

    /// The message just generated goes to a peer that had nothing we lack
    /// (`have_all_theirs`), and no answer was awaited before it: once it has the message,
    /// it holds exactly our `heads`, and answers nothing (`answered_as_delivered`; the
    /// envelope says so, `SyncEnvelope::delivered`). So it is taken as delivered: the
    /// peer's heads are ours, and no answer is awaited (otherwise about half the sync
    /// messages are those answers: along trees, to pulls, to what is sent again). Should
    /// it be lost, the peer still lacks it: where it asked (a pull), it asks again after
    /// its own reply timeout; otherwise the announcements make it pull (the slow round of
    /// its other neighbours, or ours where it reaches no other member:
    /// `swarm::tree::Plan::alone`), or our next message, which it can't apply, makes it
    /// ask. Should the peer have changes we lack after all, it answers with them, as its
    /// heads differ from ours. Undone if our transport refuses the message
    /// (`send_failed`).
    pub(crate) fn assume_delivered(&mut self, heads: &[crate::document::ChangeHash]) {
        let view = self.peer_view();
        let Some(sending) = &mut self.sending else {
            return;
        };
        sending.assumed = Some(view);
        self.state.their_heads = Some(heads.to_vec());
        self.state.shared_heads = heads.to_vec();
        // And it has all up to our heads, needing nothing: the next message carries
        // what came since, and if what we sent is forgotten (`resend_if_needed`), only
        // that goes again, not all since the peer's last message, which may be long ago
        // (no answers come).
        self.state.their_have = Some(vec![sync::Have {
            last_sync: heads.to_vec(),
            bloom: sync::BloomFilter::default(),
        }]);
        self.state.their_need = Some(Vec::new());
        self.state.in_flight = false;
        self.reply = None;
    }

    /// The changes since `before` in `doc` came from the peer (we just applied its
    /// message, or a delta it offered): it has them, so they never go back to it.
    /// Automerge goes by what it last heard the peer has, and a message that overtook
    /// another makes that older than what the peer sent since: the answer would carry
    /// the peer's own changes back. A delta comes outside the sync protocol, so
    /// Automerge would think the peer lacks all of it, and offer it back as a delta of
    /// our own. Counted as sent (`sent_hashes`), they don't; Automerge forgets them as
    /// sent once the peer's heads cover them. Returns how many changes that is. (Changes kept until their predecessors came,
    /// from another peer, count too: should this one lack them, it asks, and gets them,
    /// `resend_if_needed`.)
    pub(crate) fn came_from_peer(
        &mut self,
        doc: &Document,
        before: &[crate::document::ChangeHash],
    ) -> usize {
        let hashes: Vec<_> = doc
            .get_changes_meta(before)
            .iter()
            .map(|c| c.hash)
            .collect();
        self.these_came_from_peer(&hashes)
    }

    /// `came_from_peer`, for the changes `hashes` (found already).
    pub(crate) fn these_came_from_peer(&mut self, hashes: &[crate::document::ChangeHash]) -> usize {
        self.state.sent_hashes.extend(hashes.iter().copied());
        hashes.len()
    }

    /// Of the changes in `doc` beyond `from`, how many the peer has too, as far as we
    /// know (its heads' history, and what went either way since, `sent_hashes`). A delta
    /// it offers from `from` brings those again: offered before it heard what we have
    /// now (while we downloaded its last one, say, and it waited for our answer).
    pub(crate) fn known_beyond(
        &self,
        doc: &Document,
        from: &[crate::document::ChangeHash],
    ) -> usize {
        let from: Vec<crate::document::ChangeHash> = from
            .iter()
            .copied()
            .filter(|hash| doc.get_change_by_hash(hash).is_some())
            .collect();
        let beyond: std::collections::HashMap<_, _> = doc
            .get_changes_meta(&from)
            .into_iter()
            .map(|change| (change.hash, change.deps))
            .collect();
        // A change's predecessors outside `beyond` are in the history of `from`.
        let mut known = std::collections::HashSet::new();
        let mut pending: Vec<crate::document::ChangeHash> =
            self.state.their_heads.iter().flatten().copied().collect();
        pending.extend(self.state.sent_hashes.iter().copied());
        while let Some(hash) = pending.pop() {
            if let Some(deps) = beyond.get(&hash)
                && known.insert(hash)
            {
                pending.extend(deps.iter().copied());
            }
        }
        known.len()
    }

    /// The peer has `changes`, or will without us (another parent in their writer's
    /// tree brings them, `swarm::tree::forward`): counted as sent, they don't go to it,
    /// unless it asks for them (`resend_if_needed`).
    pub(crate) fn known_to_peer(&mut self, changes: &[crate::document::ChangeHash], now: Instant) {
        self.known_elsewhere
            .retain(|_, at| now.saturating_duration_since(*at) < KNOWN_ELSEWHERE_KEPT);
        for hash in changes {
            if self.state.sent_hashes.insert(*hash) {
                self.known_elsewhere.insert(*hash, now);
            }
        }
    }

    /// We pass `changes` on to the peer: of their predecessors, those counted as sent
    /// lately only because another parent brings them (`known_to_peer`) go along. That
    /// one may not have arrived yet (with writers in turn, each change builds on the
    /// last writer's, which comes along another tree): without them, the peer couldn't
    /// apply ours, and would ask for them, a round trip later.
    pub(crate) fn with_predecessors(
        &mut self,
        doc: &Document,
        changes: &[crate::document::ChangeHash],
        now: Instant,
    ) {
        for hash in changes {
            let Some(change) = doc.get_change_by_hash(hash) else {
                continue;
            };
            for dep in change.deps() {
                let lately = self
                    .known_elsewhere
                    .get(dep)
                    .is_some_and(|at| now.saturating_duration_since(*at) < KNOWN_ELSEWHERE_FOR);
                if lately {
                    self.known_elsewhere.remove(dep);
                    self.state.sent_hashes.remove(dep);
                }
            }
        }
    }

    /// What the sync state knows of the peer (`PeerView`).
    fn peer_view(&self) -> PeerView {
        PeerView {
            their_heads: self.state.their_heads.clone(),
            shared_heads: self.state.shared_heads.clone(),
            their_have: self.state.their_have.clone(),
            their_need: self.state.their_need.clone(),
        }
    }

    /// Whether the peer, as the message just applied says, is behind us: all its heads
    /// in our history, and not all of ours. Asked before `answered_as_delivered`: a
    /// message taken as delivered from a peer behind us is answered with what it lacks
    /// (a node that got its writer's changes before its parent did, by another way).
    pub(crate) fn peer_behind(
        &self,
        doc: &Document,
        heads: &[crate::document::ChangeHash],
    ) -> bool {
        self.state.their_heads.as_ref().is_some_and(|theirs| {
            theirs
                .iter()
                .all(|hash| doc.get_change_by_hash(hash).is_some())
                && heads.iter().any(|hash| !theirs.contains(hash))
        })
    }

    /// We applied a message its sender took as delivered (`assume_delivered`,
    /// `SyncEnvelope::delivered`) to `doc`: if we hold exactly its sender's heads now,
    /// Automerge's answer, our heads, would tell it nothing. So none goes: we count as
    /// having said our `heads`. Returns whether so.
    ///
    /// So also where the message is older than the last one from the peer (they
    /// overtake each other, as no answers are awaited) and we hold exactly the heads
    /// that one said: the older one tells nothing new of the peer, all its heads are in
    /// our history, and we are level with the peer as of the newer. Otherwise the
    /// older one's view of what we have would stand, and the answer to it would carry
    /// back to the peer the changes the newer one brought us.
    pub(crate) fn answered_as_delivered(
        &mut self,
        doc: &Document,
        heads: &[crate::document::ChangeHash],
    ) -> bool {
        let sorted = |heads: &[crate::document::ChangeHash]| {
            let mut heads = heads.to_vec();
            heads.sort_unstable();
            heads
        };
        let ours = sorted(heads);
        let level = |theirs: &Option<Vec<crate::document::ChangeHash>>| {
            theirs.as_deref().map(sorted).as_ref() == Some(&ours)
        };
        let before = self.before_receive.take();
        if !level(&self.state.their_heads) {
            let older = self.state.their_heads.as_ref().is_some_and(|theirs| {
                theirs
                    .iter()
                    .all(|hash| doc.get_change_by_hash(hash).is_some())
            });
            if !before.is_some_and(|before| older && level(&before.their_heads)) {
                return false;
            }
        }
        // Level: the peer has exactly our heads and needs nothing. Said so outright, not
        // as the message's view (or the one before it): a view whose heads we lacked
        // names older ones (`known_last_sync`), by which it would seem to lack what came
        // since.
        self.state.their_heads = Some(heads.to_vec());
        self.state.shared_heads = heads.to_vec();
        self.state.their_have = Some(vec![sync::Have {
            last_sync: heads.to_vec(),
            bloom: sync::BloomFilter::default(),
        }]);
        self.state.their_need = Some(Vec::new());
        self.state.last_sent_heads = heads.to_vec();
        self.state.have_responded = true;
        true
    }

    /// The peer has heads we lack (its keepalive said so): ask for what they need,
    /// even though Automerge thinks a message is still in flight.
    pub(crate) fn pull_from(&mut self, heads: Vec<crate::document::ChangeHash>) {
        self.state.their_heads = Some(heads);
        self.state.in_flight = false;
    }

    /// Automerge never resends a change it believes the peer has received
    /// (`sent_hashes`). If the message carrying it was lost while later ones arrive, the
    /// peer keeps asking for it (`need`, from the heads we advertise) and we keep not
    /// sending it: a livelock the reply timeout can't see (answers keep coming) and the
    /// keepalive heal can't either (sync never goes idle). So when the peer needs a
    /// change we think it has, what we think of it is wrong: forget having sent it, and
    /// what it builds on that we sent too (`forget_sent_for_need`), and the next message
    /// carries them. Not what we sent after it: messages overtake each other (we don't
    /// wait for answers to those taken as delivered), so a peer often needs what is only
    /// late, while the messages after it are still on their way. Call after `receive`.
    ///
    /// Returns how many of the changes it asked for we had counted as its because another
    /// parent brings them (`known_to_peer`): within `KNOWN_ELSEWHERE_FOR` of counting
    /// them so, and later (`swarm::stats`).
    pub(crate) fn resend_if_needed(
        &mut self,
        doc: &Document,
        peer: &str,
        now: Instant,
    ) -> (usize, usize) {
        let mut asked = (0, 0);
        for hash in self.state.their_need.iter().flatten() {
            if let Some(at) = self.known_elsewhere.remove(hash) {
                if now.saturating_duration_since(at) < KNOWN_ELSEWHERE_FOR {
                    asked.0 += 1;
                } else {
                    asked.1 += 1;
                }
            }
        }
        let forgotten = forget_sent_for_need(&mut self.state, doc);
        if forgotten > 0 {
            debug!(
                peer = %peer,
                forgotten,
                "Peer needs changes we had sent; sending them again"
            );
        }
        if answer_for_everything_since_shared(&mut self.state, doc) {
            debug!(
                peer = %peer,
                need = self.state.their_need.as_ref().map_or(0, Vec::len),
                "Peer lacks the changes ours build on; sending everything since what we share"
            );
        }
        asked
    }

    /// For the simulator's dump: the peer's heads as Automerge knows them, and whether
    /// it thinks a message is in flight.
    #[cfg(test)]
    pub(crate) fn sync_view(&self) -> (Option<&[crate::document::ChangeHash]>, bool) {
        (self.state.their_heads.as_deref(), self.state.in_flight)
    }

    /// Whether we leave the first message of this session to the peer.
    ///
    /// After a reset, both sides usually open at once, and the openings cross. Between
    /// peers that share history that costs nothing. But a peer with an empty document
    /// answers our opening with empty heads, which Automerge takes for "lost
    /// everything", so the whole document would go out a second time. So a ready node
    /// never opens towards a peer that isn't ready (one that is still catching up); that
    /// peer opens itself.
    pub(crate) fn leave_opening(&self, we_ready: bool, peer_ready: Option<bool>) -> bool {
        self.state.their_heads.is_none() && we_ready && peer_ready == Some(false)
    }

    /// Which session this is: a message generated from it reports its send's outcome
    /// with this (`send_succeeded`, `send_failed`).
    pub(crate) fn id(&self) -> u64 {
        self.id
    }

    /// Our transport took the message generated from session `session`. Returns whether
    /// to generate again: a pass was requested while it was handed over, or a reset
    /// replaced the session meanwhile (any pass requested then was lost with the old
    /// session, and the new one hasn't necessarily opened yet).
    pub(crate) fn send_succeeded(&mut self, session: u64) -> bool {
        if session != self.id {
            return true;
        }
        self.retry_at = None;
        self.send_backoff.reset();
        self.sending
            .take()
            .is_some_and(|sending| sending.regenerate)
    }

    /// Our transport refused the message: as far as the peer is concerned, it was never
    /// sent. Takes back its changes (they go out with the next message), clears
    /// `in_flight`, and returns how long to wait before generating again. None if a reset
    /// already discarded what was being sent (the message came from another `session`).
    pub(crate) fn send_failed(&mut self, session: u64, now: Instant) -> Option<Duration> {
        if session != self.id {
            return None;
        }
        let sending = self.sending.take()?;
        for hash in &sending.sent {
            self.state.sent_hashes.remove(hash);
        }
        if let Some(assumed) = sending.assumed {
            self.state.their_heads = assumed.their_heads;
            self.state.shared_heads = assumed.shared_heads;
            self.state.their_have = assumed.their_have;
            self.state.their_need = assumed.their_need;
        }
        if let Some(wait) = &mut self.reply {
            wait.sent.retain(|hash| !sending.sent.contains(hash));
        }
        self.state.in_flight = false;
        let delay = self.send_backoff.next();
        self.retry_at = Some(now + delay);
        Some(delay)
    }

    /// A sync message of `len` bytes went out; `sent` are the changes it carried for the
    /// first time. `bytes` are the sync bytes exchanged with the peer before it.
    pub(crate) fn expect_reply(
        &mut self,
        sent: Vec<crate::document::ChangeHash>,
        now: Instant,
        bytes: SyncBytes,
        len: usize,
    ) {
        match &mut self.reply {
            // Still waiting for an earlier one: keep its deadline, remember these too.
            Some(wait) => {
                wait.sent.extend(sent);
                wait.largest = wait.largest.max(len);
            }
            None => {
                self.reply = Some(SyncReplyWait {
                    deadline: now + self.reply_backoff.current(),
                    started: (now, bytes),
                    sent,
                    largest: len,
                });
            }
        }
    }

    /// No answer is due any more: the peer answered, or keepalives show both documents
    /// are the same.
    pub(crate) fn reply_settled(&mut self) {
        self.reply = None;
        self.reply_backoff.reset();
    }

    /// The peer answered our message with a delta offer (`SyncBody::Delta`): the wait
    /// ends, and to Automerge no message of ours is in flight any more, as an answer of
    /// its own would have told it. Else, with our heads unchanged (an offer we don't
    /// fetch, or a download that fails), it would say nothing again, and the peer would
    /// never hear what we have to offer afresh.
    pub(crate) fn answered_by_offer(&mut self) {
        self.reply_settled();
        self.state.in_flight = false;
    }

    /// If the answer is overdue: forgets that the unanswered changes were sent and that
    /// a message is in flight, so the next sync pass sends them again. Returns how many
    /// changes that concerns. Those of a message still being handed over stay sent: they
    /// are on their way (and taken back if its send fails, `send_failed`).
    pub(crate) fn take_overdue_reply(&mut self, now: Instant) -> Option<usize> {
        if self.reply.as_ref().is_none_or(|wait| now < wait.deadline) {
            return None;
        }
        let wait = self.reply.take()?;
        let handing_over: HashSet<_> = self
            .sending
            .iter()
            .flat_map(|sending| sending.sent.iter().copied())
            .collect();
        for hash in wait.sent.iter().filter(|hash| !handing_over.contains(hash)) {
            self.state.sent_hashes.remove(hash);
        }
        self.state.in_flight = false;
        self.delta_offered = false;
        self.reply_backoff.grow();
        Some(wait.sent.len())
    }
}

/// A peer that got a change of ours without the changes it builds on (they were lost,
/// a later one arrived) asks for those it lacks, and while it does, Automerge leaves
/// out what it has (its bloom filter: `have` waits until all it needs is in our heads).
/// Without a `have`, Automerge sends exactly what was asked for, and the peer, finding
/// each change's own predecessors missing in turn, asks again: one round trip per
/// missing change, hundreds after a short outage. So answer as if it had said it has
/// nothing since the heads we share: everything since then goes in one message (a
/// delta when large: `swarm::sync`), and what it had already only twice. But what we
/// sent and it doesn't need may be on its way still: that stays sent, but for what it
/// needs and what that builds on (`forget_sent_for_need`). Returns whether it did. Call
/// after receiving.
fn answer_for_everything_since_shared(state: &mut sync::State, doc: &Document) -> bool {
    let asks_without_saying = state.their_have.as_ref().is_some_and(Vec::is_empty)
        && state
            .their_need
            .as_ref()
            .is_some_and(|need| !need.is_empty());
    if !asks_without_saying {
        return false;
    }
    state.their_have = Some(vec![sync::Have {
        last_sync: state.shared_heads.clone(),
        bloom: sync::BloomFilter::default(),
    }]);
    // What it needs, and what that builds on, may be what was lost.
    forget_sent_for_need(state, doc);
    true
}

/// Forgets having sent what the peer needs (`their_need`) and the changes those build
/// on that we sent too, following their predecessors as long as we sent them: so all of
/// a run of lost messages goes again at once. Changes sent after them stay sent (they
/// may only be late). Returns how many were forgotten.
fn forget_sent_for_need(state: &mut sync::State, doc: &Document) -> usize {
    let Some(need) = &state.their_need else {
        return 0;
    };
    let mut forgotten = 0;
    let mut next: Vec<crate::document::ChangeHash> = need
        .iter()
        .filter(|hash| state.sent_hashes.contains(hash))
        .copied()
        .collect();
    while let Some(hash) = next.pop() {
        if !state.sent_hashes.remove(&hash) {
            continue;
        }
        forgotten += 1;
        if let Some(change) = doc.get_change_by_hash(&hash) {
            next.extend(
                change
                    .deps()
                    .iter()
                    .filter(|dep| state.sent_hashes.contains(dep))
                    .copied(),
            );
        }
    }
    forgotten
}

#[cfg(test)]
mod model;

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

    /// After a loss, the changes the peer lacks go in one message, not one per round
    /// trip (`answer_for_everything_since_shared`).
    #[test]
    fn missing_predecessors_go_at_once() {
        use crate::document::testing::change;
        use automerge::transaction::Transactable;
        let mut a = Document::new();
        let mut b = Document::new();
        let (mut at_a, mut at_b) = (sync::State::new(), sync::State::new());
        change(&mut a, |a| a.put(automerge::ROOT, "early", 1).unwrap());
        // Until in step.
        for _ in 0..10 {
            let to_b = a.generate_sync_message(&mut at_a);
            let to_a = b.generate_sync_message(&mut at_b);
            if to_b.is_none() && to_a.is_none() {
                break;
            }
            if let Some(m) = to_b {
                b.receive_sync_message(&mut at_b, m).unwrap();
            }
            if let Some(m) = to_a {
                a.receive_sync_message(&mut at_a, m).unwrap();
            }
        }
        assert_eq!(a.get_heads(), b.get_heads());
        // Five changes, whose message is lost; then a sixth, which arrives alone.
        for i in 0..5 {
            change(&mut a, |a| {
                a.put(automerge::ROOT, format!("lost{i}"), i).unwrap()
            });
        }
        let _lost = a.generate_sync_message(&mut at_a).unwrap();
        change(&mut a, |a| a.put(automerge::ROOT, "last", 6).unwrap());
        at_a.in_flight = false; // As after our reply timeout.
        let last = a.generate_sync_message(&mut at_a).unwrap();
        assert_eq!(last.changes.len(), 1);
        b.receive_sync_message(&mut at_b, last).unwrap();
        // b asks for the fifth, saying nothing of what it has.
        let ask = b.generate_sync_message(&mut at_b).unwrap();
        assert!(ask.have.is_empty() && ask.need.len() == 1);
        a.receive_sync_message(&mut at_a, ask).unwrap();
        assert!(answer_for_everything_since_shared(&mut at_a, &a));
        // All since the shared heads, in one message (here the whole document: more
        // than a third of it is missing).
        let answer = a.generate_sync_message(&mut at_a).unwrap();
        b.receive_sync_message(&mut at_b, answer).unwrap();
        assert_eq!(b.get_heads(), a.get_heads());
        // A message with what it has is left alone.
        assert!(!answer_for_everything_since_shared(&mut at_b, &b));
    }

    fn session() -> SyncSession {
        SyncSession::new(Backoff::new(
            Duration::from_secs(1),
            Duration::from_secs(30),
        ))
    }

    fn sending() -> Option<Sending> {
        Some(Sending {
            sent: Vec::new(),
            regenerate: false,
            assumed: None,
        })
    }

    #[test]
    fn a_send_from_a_replaced_session_leaves_the_new_one_alone() {
        let old = session();
        let mut new = session();
        new.sending = sending();

        // A reset discarded the old session while its message was handed over: the
        // new one's own message stays, and the peer is synced again.
        assert!(new.send_succeeded(old.id()));
        assert!(new.sending.is_some());
        assert_eq!(new.send_failed(old.id(), Instant::now()), None);
        assert!(new.sending.is_some());
        assert_eq!(new.retry_at, None);

        // Its own outcome still applies.
        assert!(!new.send_succeeded(new.id()));
        assert!(new.sending.is_none());
    }

    /// A delta offered where no answer was awaited (the peer's heads were ours) still
    /// lapses at the reply timeout: else nothing ends the wait for a peer with nothing
    /// to say, and no message goes to it again.
    #[test]
    fn an_offered_delta_always_lapses() {
        let mut a = session();
        a.sending = sending();
        assert!(a.reply.is_none());
        let now = Instant::now();
        a.withdraw_for_delta(&Document::new(), now);
        assert!(a.delta_offered);
        assert!(
            a.take_overdue_reply(now + Duration::from_secs(3600))
                .is_some()
        );
        assert!(!a.delta_offered);
    }

    /// An offer that lapsed goes again as it was while nothing changed; otherwise, or
    /// once the message would fit, it is generated afresh.
    #[test]
    fn the_same_offer_goes_again_ungenerated() {
        use crate::document::testing::change;
        use automerge::{ROOT, transaction::Transactable};
        let mut ours = Document::new();
        change(&mut ours, |doc| doc.put(ROOT, "k", 1).unwrap());
        let heads = ours.get_heads();
        let mut a = session();
        a.state.their_heads = Some(Vec::new());
        let now = Instant::now();
        assert_eq!(
            a.offer_again(&ours, &heads, 1000, now),
            None,
            "nothing offered yet"
        );

        a.offered(&[], &heads, 5000);
        assert_eq!(a.offer_again(&ours, &heads, 1000, now), Some(Vec::new()));
        assert!(a.delta_offered && a.reply.is_some() && a.sending.is_some());
        assert_eq!(
            a.offer_again(&ours, &heads, 5000, now),
            None,
            "it would fit now"
        );

        change(&mut ours, |doc| doc.put(ROOT, "k", 2).unwrap());
        assert_eq!(
            a.offer_again(&ours, &ours.get_heads(), 1000, now),
            None,
            "ours moved"
        );
        a.state.their_heads = Some(heads.clone());
        assert_eq!(
            a.offer_again(&ours, &heads, 1000, now),
            None,
            "theirs moved"
        );
    }

    /// A message too large to send is taken back: the peer, after the offer and an
    /// overdue answer, gets the same changes in the next message.
    #[test]
    fn withdrawing_for_a_delta_keeps_the_changes_for_later() {
        use crate::document::testing::change;
        use automerge::{ROOT, transaction::Transactable};

        // 16 KiB of hex, about 8 KiB of information: Automerge compresses what it sends
        // (a peer with nothing gets the document saved), so this stays above 4 KB.
        let mut rng = crate::testrand::Rng::new(1);
        let big: String = (0..1024)
            .map(|_| format!("{:016x}", rng.next_u64()))
            .collect();
        let mut ours = Document::new();
        change(&mut ours, |ours| ours.put(ROOT, "big", big).unwrap());
        let mut theirs = Document::new();
        let (mut a, mut b) = (session(), session());
        let now = Instant::now();
        let bytes = SyncBytes::default();

        // a opens, b answers with its (empty) heads, then a sends the changes.
        let opening = a.generate(&ours, now, bytes).unwrap();
        a.send_succeeded(a.id());
        b.receive(&mut theirs, decode(&opening).unwrap()).unwrap();
        let answer = b.generate(&theirs, now, bytes).unwrap();
        b.send_succeeded(b.id());
        a.receive(&mut ours, decode(&answer).unwrap()).unwrap();
        let carrying = a.generate(&ours, now, bytes).unwrap();
        let carried = carrying.0.len();
        assert!(carried > 4_000, "the changes took {carried} bytes");

        let from = a.withdraw_for_delta(&ours, now);
        assert!(from.is_empty());
        assert!(a.delta_offered);
        assert!(!a.send_succeeded(a.id()));

        // No answer to the offer: it lapses, and the changes go out again.
        assert!(
            a.take_overdue_reply(now + Duration::from_secs(3600))
                .is_some()
        );
        assert!(!a.delta_offered);
        let again = a.generate(&ours, now, bytes).unwrap();
        let carried_again = again.0.len();
        assert!(carried_again > 4_000, "sent again in {carried_again} bytes");
    }

    /// A reply overdue while its message is still being handed over: what that message
    /// carries stays sent (it is on its way), so the next pass doesn't send it twice.
    #[test]
    fn what_is_being_handed_over_stays_sent() {
        use crate::document::testing::change;
        use automerge::{ROOT, transaction::Transactable};
        let (mut ours, mut theirs) = (Document::new(), Document::new());
        change(&mut ours, |doc| doc.put(ROOT, "early", 1).unwrap());
        let (mut a, _b) = in_step(&mut ours, &mut theirs);
        let (now, bytes) = (Instant::now(), SyncBytes::default());
        change(&mut ours, |doc| doc.put(ROOT, "news", 2).unwrap());
        a.generate(&ours, now, bytes).expect("news");
        assert!(a.reply.is_some(), "an answer awaited");
        let news = ours.get_heads()[0];
        assert!(a.state.sent_hashes.contains(&news));

        // Overdue while still in the transport's hands: still sent.
        assert!(
            a.take_overdue_reply(now + Duration::from_secs(3600))
                .is_some()
        );
        assert!(a.state.sent_hashes.contains(&news), "on its way");
        // The send fails: taken back, to go again.
        a.send_failed(a.id(), now);
        assert!(!a.state.sent_hashes.contains(&news));
    }

    /// Automerge, as we rely on it: a message with no heads says the peer lost all it
    /// had, and what we counted as its goes again (why a push never goes without heads:
    /// `Received::pushed`, and an opening never goes empty: `leave_opening`).
    #[test]
    fn no_heads_say_the_peer_lost_everything() {
        use crate::document::testing::change;
        use automerge::{ROOT, transaction::Transactable};
        let (mut ours, mut theirs) = (Document::new(), Document::new());
        change(&mut ours, |doc| doc.put(ROOT, "early", 1).unwrap());
        let (mut a, _b) = in_step(&mut ours, &mut theirs);
        let (now, bytes) = (Instant::now(), SyncBytes::default());
        // A change sent, not yet known to have arrived.
        change(&mut ours, |doc| doc.put(ROOT, "news", 2).unwrap());
        a.generate(&ours, now, bytes).expect("the news");
        assert!(!a.state.sent_hashes.is_empty());

        // From a session over an empty document: no heads.
        let mut lost = session();
        let empty = lost
            .generate(&Document::new(), now, bytes)
            .expect("an opening");
        let empty = decode(&empty).unwrap();
        assert!(empty.heads().is_empty());
        a.receive(&mut ours, empty).unwrap();
        assert!(a.state.sent_hashes.is_empty(), "all of it goes again");
    }

    /// Automerge, as we rely on it: a change that came without its predecessor waits,
    /// and what the receiver asks for (`need`) is that predecessor alone, not the
    /// change it holds already (a lost push costs one ask: `swarm::relay`).
    #[test]
    fn a_lost_predecessor_is_asked_for_alone() {
        use crate::document::testing::change;
        use automerge::{ROOT, transaction::Transactable};
        let (mut ours, mut theirs) = (Document::new(), Document::new());
        change(&mut ours, |doc| doc.put(ROOT, "early", 1).unwrap());
        let (mut a, mut b) = in_step(&mut ours, &mut theirs);
        let (now, bytes) = (Instant::now(), SyncBytes::default());

        change(&mut ours, |doc| doc.put(ROOT, "lost", 2).unwrap());
        let lost = ours.get_heads()[0];
        a.push(&ours, 1 << 20).expect("pushed, and lost");
        a.assume_delivered(&ours.get_heads());
        a.send_succeeded(a.id());
        change(&mut ours, |doc| doc.put(ROOT, "later", 3).unwrap());
        let (heads, changes) = a.push(&ours, 1 << 20).expect("the later one");
        assert_eq!(
            changes.len(),
            1,
            "only the later one: the lost one counts as sent"
        );

        b.receive(
            &mut theirs,
            Received::pushed(heads, changes).expect("decodes"),
        )
        .unwrap();
        assert!(theirs.get_change_by_hash(&lost).is_none());
        let ask = b.generate(&theirs, now, bytes).expect("an ask");
        assert_eq!(decode(&ask).unwrap().0.need, vec![lost]);
    }

    /// Automerge, as we rely on it: a message taken back for a delta leaves Automerge as
    /// if it hadn't been generated (`withdraw_for_delta` restores what it had sent): the
    /// peer opened, our answer was the delta's message, so our heads never went. Once
    /// the peer has the delta and says its heads are ours, Automerge still answers with
    /// them (it hadn't answered yet), and the peer knows it holds all we have (a joining
    /// node finishes its initial sync on that).
    #[test]
    fn after_a_delta_the_offerer_still_answers() {
        use crate::document::testing::change;
        use automerge::{ROOT, transaction::Transactable};
        let mut rng = crate::testrand::Rng::new(2);
        let big: String = (0..1024)
            .map(|_| format!("{:016x}", rng.next_u64()))
            .collect();
        let mut ours = Document::new();
        change(&mut ours, |ours| ours.put(ROOT, "big", big).unwrap());
        let mut theirs = Document::new();
        let (mut a, mut b) = (session(), session());
        let (now, bytes) = (Instant::now(), SyncBytes::default());

        // b opens (it has nothing); a's first message would carry everything.
        let opening = b.generate(&theirs, now, bytes).expect("b opens");
        b.send_succeeded(b.id());
        a.receive(&mut ours, decode(&opening).unwrap()).unwrap();
        a.generate(&ours, now, bytes).expect("the changes");
        a.withdraw_for_delta(&ours, now);

        // The delta: b loads all of a's changes, as from a blob.
        let before = theirs.get_heads();
        theirs.merge(&mut ours.clone()).unwrap();
        b.came_from_peer(&theirs, &before);
        let has_all = b.generate(&theirs, now, bytes).expect("b says what it has");
        b.send_succeeded(b.id());
        a.receive(&mut ours, decode(&has_all).unwrap()).unwrap();
        let heads = a
            .generate(&ours, now, bytes)
            .expect("a answers, though b has all of it");
        b.receive(&mut theirs, decode(&heads).unwrap()).unwrap();
        assert!(b.have_all_theirs(&theirs));
    }

    /// A message that comes late (a copy of one before, or overtaken) names heads older
    /// than the peer said since: they stay its heads, and what it sent us since doesn't
    /// go back to it (as it did by a push: beyond the older heads, and no longer counted
    /// as sent once the peer's heads showed it).
    #[test]
    fn a_late_message_takes_the_peer_back_nowhere() {
        use crate::document::testing::change;
        use automerge::{ROOT, transaction::Transactable};
        let (mut ours, mut theirs) = (Document::new(), Document::new());
        change(&mut ours, |doc| doc.put(ROOT, "early", 1).unwrap());
        let (mut a, mut b) = in_step(&mut ours, &mut theirs);
        let (now, bytes) = (Instant::now(), SyncBytes::default());

        change(&mut ours, |doc| doc.put(ROOT, "first", 2).unwrap());
        let first = a.generate(&ours, now, bytes).expect("the first");
        a.send_succeeded(a.id());
        change(&mut ours, |doc| doc.put(ROOT, "then", 3).unwrap());
        let then = ours.get_heads();
        let second = a.generate(&ours, now, bytes).expect("the second");
        a.send_succeeded(a.id());

        // Both arrive, the second twice (so it no longer counts as sent: the peer's
        // heads show it); then the first again, late.
        for message in [&first, &second, &second] {
            let before = theirs.get_heads();
            b.receive(&mut theirs, decode(message).unwrap()).unwrap();
            b.came_from_peer(&theirs, &before);
        }
        b.receive(&mut theirs, decode(&first).unwrap()).unwrap();
        assert_eq!(b.state.their_heads.as_deref(), Some(&then[..]));
        assert!(
            b.push(&theirs, 1 << 20).is_none(),
            "nothing of a's goes back to it"
        );
    }

    /// Two sessions exchanging messages until neither has anything to say.
    fn in_step(ours: &mut Document, theirs: &mut Document) -> (SyncSession, SyncSession) {
        let (mut a, mut b) = (session(), session());
        let (now, bytes) = (Instant::now(), SyncBytes::default());
        for _ in 0..10 {
            let to_b = a.generate(ours, now, bytes);
            a.send_succeeded(a.id());
            let to_a = b.generate(theirs, now, bytes);
            b.send_succeeded(b.id());
            if to_b.is_none() && to_a.is_none() {
                break;
            }
            if let Some(m) = to_b {
                b.receive(theirs, decode(&m).unwrap()).unwrap();
            }
            if let Some(m) = to_a {
                a.receive(ours, decode(&m).unwrap()).unwrap();
            }
        }
        assert_eq!(ours.get_heads(), theirs.get_heads());
        (a, b)
    }

    /// A message to a peer that had everything else: taken as delivered, and not
    /// answered; neither side has more to say.
    #[test]
    fn delivered_messages_need_no_answer() {
        use crate::document::testing::change;
        use automerge::{ROOT, transaction::Transactable};
        let (mut ours, mut theirs) = (Document::new(), Document::new());
        change(&mut ours, |doc| doc.put(ROOT, "early", 1).unwrap());
        let (mut a, mut b) = in_step(&mut ours, &mut theirs);
        let (now, bytes) = (Instant::now(), SyncBytes::default());

        change(&mut ours, |doc| doc.put(ROOT, "news", 2).unwrap());
        assert!(a.have_all_theirs(&ours));
        let message = a.generate(&ours, now, bytes).unwrap();
        a.assume_delivered(&ours.get_heads());
        a.send_succeeded(a.id());
        assert!(a.reply.is_none(), "no answer awaited");

        b.receive(&mut theirs, decode(&message).unwrap()).unwrap();
        assert!(b.answered_as_delivered(&theirs, &theirs.get_heads()));
        assert!(
            b.generate(&theirs, now, bytes).is_none(),
            "nothing to answer"
        );
        assert!(
            a.generate(&ours, now, bytes).is_none(),
            "nothing more to send"
        );
        assert_eq!(ours.get_heads(), theirs.get_heads());
    }

    /// The peer wrote meanwhile: its heads differ from ours after our message, so it
    /// answers, with its change.
    #[test]
    fn delivered_messages_are_answered_with_changes_we_lack() {
        use crate::document::testing::change;
        use automerge::{ROOT, transaction::Transactable};
        let (mut ours, mut theirs) = (Document::new(), Document::new());
        change(&mut ours, |doc| doc.put(ROOT, "early", 1).unwrap());
        let (mut a, mut b) = in_step(&mut ours, &mut theirs);
        let (now, bytes) = (Instant::now(), SyncBytes::default());

        change(&mut ours, |doc| doc.put(ROOT, "ours", 2).unwrap());
        change(&mut theirs, |doc| doc.put(ROOT, "theirs", 3).unwrap());
        let message = a.generate(&ours, now, bytes).unwrap();
        a.assume_delivered(&ours.get_heads());
        a.send_succeeded(a.id());

        b.receive(&mut theirs, decode(&message).unwrap()).unwrap();
        assert!(!b.answered_as_delivered(&theirs, &theirs.get_heads()));
        let answer = b.generate(&theirs, now, bytes).expect("b answers");
        b.send_succeeded(b.id());
        a.receive(&mut ours, decode(&answer).unwrap()).unwrap();
        assert_eq!(ours.get_heads(), theirs.get_heads());
    }

    /// A message taken as delivered is lost, and the next one arrives: the peer can't
    /// apply it, answers what it needs, and gets the lost change then.
    #[test]
    fn a_lost_delivered_message_is_asked_for() {
        use crate::document::testing::change;
        use automerge::{ROOT, transaction::Transactable};
        let (mut ours, mut theirs) = (Document::new(), Document::new());
        change(&mut ours, |doc| doc.put(ROOT, "early", 1).unwrap());
        let (mut a, mut b) = in_step(&mut ours, &mut theirs);
        let (now, bytes) = (Instant::now(), SyncBytes::default());

        change(&mut ours, |doc| doc.put(ROOT, "lost", 2).unwrap());
        let _lost = a.generate(&ours, now, bytes).unwrap();
        a.assume_delivered(&ours.get_heads());
        a.send_succeeded(a.id());
        change(&mut ours, |doc| doc.put(ROOT, "next", 3).unwrap());
        assert!(
            a.have_all_theirs(&ours),
            "a takes the lost one as delivered"
        );
        let next = a.generate(&ours, now, bytes).unwrap();
        a.assume_delivered(&ours.get_heads());
        a.send_succeeded(a.id());

        b.receive(&mut theirs, decode(&next).unwrap()).unwrap();
        assert!(!b.answered_as_delivered(&theirs, &theirs.get_heads()));
        let ask = b
            .generate(&theirs, now, bytes)
            .expect("b asks for what it lacks");
        b.send_succeeded(b.id());
        a.receive(&mut ours, decode(&ask).unwrap()).unwrap();
        a.resend_if_needed(&ours, "b", Instant::now());
        let answer = a.generate(&ours, now, bytes).expect("a sends the lost one");
        a.send_succeeded(a.id());
        b.receive(&mut theirs, decode(&answer).unwrap()).unwrap();
        assert_eq!(ours.get_heads(), theirs.get_heads());
    }

    /// Two messages taken as delivered arrive in reverse order: the peer asks for the
    /// first, which arrives meanwhile; only the first is sent again (the second was on
    /// its way), taken as delivered too. Nobody waits for an answer that won't come.
    #[test]
    fn reordered_delivered_messages_leave_nothing_awaited() {
        use crate::document::testing::change;
        use automerge::{ROOT, transaction::Transactable};
        let (mut ours, mut theirs) = (Document::new(), Document::new());
        change(&mut ours, |doc| doc.put(ROOT, "early", 1).unwrap());
        let (mut a, mut b) = in_step(&mut ours, &mut theirs);
        let (now, bytes) = (Instant::now(), SyncBytes::default());
        let send = |a: &mut SyncSession, ours: &Document| {
            assert!(a.reply.is_none() && a.have_all_theirs(ours));
            let message = a.generate(ours, now, bytes).unwrap();
            a.assume_delivered(&ours.get_heads());
            a.send_succeeded(a.id());
            message
        };

        change(&mut ours, |doc| doc.put(ROOT, "first", 2).unwrap());
        let first = send(&mut a, &ours);
        change(&mut ours, |doc| doc.put(ROOT, "second", 3).unwrap());
        let second = send(&mut a, &ours);

        b.receive(&mut theirs, decode(&second).unwrap()).unwrap();
        assert!(!b.answered_as_delivered(&theirs, &theirs.get_heads()));
        let ask = b
            .generate(&theirs, now, bytes)
            .expect("b asks for the first");
        // The second named the heads the first brought as shared, which b lacks: no
        // reset for that, else a forgets all it sent.
        let flags = sync::Message::decode(&ask.0).unwrap().flags;
        assert!(
            !flags.is_some_and(|flags| flags.contains(sync::MessageFlags::SYNC_RESET)),
            "b asked with a reset"
        );
        b.send_succeeded(b.id());
        b.receive(&mut theirs, decode(&first).unwrap()).unwrap();
        assert_eq!(theirs.get_heads(), ours.get_heads(), "b has both now");
        // The first is older than the second, whose heads b holds now: level with a as
        // of that, b says nothing, and doesn't send a the second back.
        assert!(b.answered_as_delivered(&theirs, &theirs.get_heads()));
        assert!(
            b.generate(&theirs, now, bytes).is_none(),
            "nothing sent back"
        );

        a.receive(&mut ours, decode(&ask).unwrap()).unwrap();
        a.resend_if_needed(&ours, "b", Instant::now());
        let sent_before = a.state.sent_hashes.clone();
        let again = send(&mut a, &ours);
        // What the message carried: the changes it added to what a counts as sent.
        let carried = a.state.sent_hashes.difference(&sent_before).count();
        assert_eq!(
            carried, 1,
            "only the first again, not the second, on its way"
        );
        b.receive(&mut theirs, decode(&again).unwrap()).unwrap();
        assert!(b.answered_as_delivered(&theirs, &theirs.get_heads()));
        assert!(
            b.generate(&theirs, now, bytes).is_none(),
            "nothing to answer"
        );
        assert!(a.reply.is_none(), "nothing awaited");
        assert_eq!(ours.get_heads(), theirs.get_heads());
    }

    /// After messages taken as delivered (no answers, so nothing new heard of what the
    /// peer has), forgetting what was sent resends only what came since the last of
    /// them, not all since the peer's last message.
    #[test]
    fn after_delivered_messages_only_news_goes_again() {
        use crate::document::testing::change;
        use automerge::{ROOT, transaction::Transactable};
        let (mut ours, mut theirs) = (Document::new(), Document::new());
        change(&mut ours, |doc| doc.put(ROOT, "early", 1).unwrap());
        let (mut a, mut b) = in_step(&mut ours, &mut theirs);
        let (now, bytes) = (Instant::now(), SyncBytes::default());
        for i in 0..5 {
            change(&mut ours, |doc| doc.put(ROOT, format!("k{i}"), i).unwrap());
            let message = a.generate(&ours, now, bytes).unwrap();
            a.assume_delivered(&ours.get_heads());
            a.send_succeeded(a.id());
            b.receive(&mut theirs, decode(&message).unwrap()).unwrap();
            assert!(b.answered_as_delivered(&theirs, &theirs.get_heads()));
        }
        // As `resend_if_needed` does when the peer needs what we sent.
        a.state.sent_hashes.clear();
        change(&mut ours, |doc| doc.put(ROOT, "news", 6).unwrap());
        a.generate(&ours, now, bytes).unwrap();
        // What the message carried: what a counts as sent now (nothing before).
        assert_eq!(a.state.sent_hashes.len(), 1, "only the news");
    }

    /// Three messages arrive in reverse order: once the first is in, all three apply,
    /// and whatever goes back to the peer carries none of its own changes.
    #[test]
    fn changes_from_the_peer_never_go_back() {
        use crate::document::testing::change;
        use automerge::{ROOT, transaction::Transactable};
        let (mut ours, mut theirs) = (Document::new(), Document::new());
        change(&mut ours, |doc| doc.put(ROOT, "early", 1).unwrap());
        let (mut a, mut b) = in_step(&mut ours, &mut theirs);
        let (now, bytes) = (Instant::now(), SyncBytes::default());
        let mut messages = Vec::new();
        for i in 0..3 {
            change(&mut ours, |doc| doc.put(ROOT, format!("k{i}"), i).unwrap());
            messages.push(a.generate(&ours, now, bytes).unwrap());
            a.assume_delivered(&ours.get_heads());
            a.send_succeeded(a.id());
        }
        for message in messages.iter().rev() {
            let before = theirs.get_heads();
            b.receive(&mut theirs, decode(message).unwrap()).unwrap();
            b.came_from_peer(&theirs, &before);
            b.answered_as_delivered(&theirs, &theirs.get_heads());
        }
        assert_eq!(theirs.get_heads(), ours.get_heads());
        if let Some(answer) = b.generate(&theirs, now, bytes) {
            let carried = decode(&answer).unwrap().change_bytes();
            assert_eq!(carried, 0, "a's own changes went back to it");
        }
    }

    /// A delta starts at the peer's heads where we hold them all; where it has one we
    /// lack, at those we hold and the heads we share, so every server of the delta
    /// holds where it starts.
    #[test]
    fn a_delta_starts_where_we_hold_its_start() {
        use crate::document::testing::change;
        use automerge::{ROOT, transaction::Transactable};
        let mut doc = Document::new();
        change(&mut doc, |doc| doc.put(ROOT, "k", 1).unwrap());
        let shared = doc.get_heads();
        change(&mut doc, |doc| doc.put(ROOT, "k", 2).unwrap());
        let known = doc.get_heads();
        let unknown = crate::document::ChangeHash([7; 32]);
        let mut a = session();
        a.state.shared_heads = shared.clone();

        a.state.their_heads = Some(known.clone());
        assert_eq!(a.delta_from(&doc), known);
        a.state.their_heads = Some(vec![known[0], unknown]);
        let mut expected = vec![known[0], shared[0]];
        expected.sort_unstable();
        assert_eq!(a.delta_from(&doc), expected);
    }

    /// A delta offered from heads we since moved past, by changes from the offerer,
    /// would bring those again; our own, which it lacks, don't count.
    #[test]
    fn offers_from_before_count_what_came_since() {
        use crate::document::changes::{delta_bytes, load_delta};
        use crate::document::testing::change;
        use automerge::{ROOT, transaction::Transactable};
        let (mut ours, mut theirs) = (Document::new(), Document::new());
        change(&mut ours, |doc| doc.put(ROOT, "early", 1).unwrap());
        let (_a, mut b) = in_step(&mut ours, &mut theirs);

        let before = theirs.get_heads();
        change(&mut ours, |doc| doc.put(ROOT, "news", 2).unwrap());
        let delta = delta_bytes(&ours, &before, &ours.get_heads()).unwrap();
        load_delta(&mut theirs, &delta).unwrap();
        b.came_from_peer(&theirs, &before);
        change(&mut theirs, |doc| doc.put(ROOT, "own", 3).unwrap());

        assert_eq!(
            b.known_beyond(&theirs, &before),
            1,
            "the delta's change again"
        );
        assert_eq!(b.known_beyond(&theirs, &theirs.get_heads()), 0);
    }

    /// Changes the peer gets from elsewhere (another parent in their writer's tree)
    /// don't go to it, but do if it asks.
    #[test]
    fn changes_known_to_the_peer_go_only_when_asked() {
        use crate::document::testing::change;
        use automerge::{ROOT, transaction::Transactable};
        let (mut ours, mut theirs) = (Document::new(), Document::new());
        change(&mut ours, |doc| doc.put(ROOT, "early", 1).unwrap());
        let (mut a, mut b) = in_step(&mut ours, &mut theirs);
        let (now, bytes) = (Instant::now(), SyncBytes::default());

        change(&mut ours, |doc| doc.put(ROOT, "elsewhere", 2).unwrap());
        a.known_to_peer(&ours.get_heads(), Instant::now());
        let message = a.generate(&ours, now, bytes).expect("our heads moved");
        a.send_succeeded(a.id());
        let carried = decode(&message).unwrap().change_bytes();
        assert_eq!(carried, 0, "not sent: the peer gets it from elsewhere");

        // It didn't: it asks, and gets it.
        b.receive(&mut theirs, decode(&message).unwrap()).unwrap();
        let asking = b.generate(&theirs, now, bytes).expect("it asks");
        b.send_succeeded(b.id());
        a.receive(&mut ours, decode(&asking).unwrap()).unwrap();
        a.resend_if_needed(&ours, "b", Instant::now());
        let answer = a.generate(&ours, now, bytes).expect("answered");
        b.receive(&mut theirs, decode(&answer).unwrap()).unwrap();
        assert_eq!(theirs.get_heads(), ours.get_heads());
    }

    /// A change counted as known to the peer lately goes along with one of ours that
    /// builds on it: the peer applies both at once, without asking.
    #[test]
    fn predecessors_known_elsewhere_go_along() {
        use crate::document::testing::change;
        use automerge::{ROOT, transaction::Transactable};
        let (mut ours, mut theirs) = (Document::new(), Document::new());
        change(&mut ours, |doc| doc.put(ROOT, "early", 1).unwrap());
        let (mut a, mut b) = in_step(&mut ours, &mut theirs);
        let (now, bytes) = (Instant::now(), SyncBytes::default());

        change(&mut ours, |doc| doc.put(ROOT, "elsewhere", 2).unwrap());
        a.known_to_peer(&ours.get_heads(), now);
        change(&mut ours, |doc| doc.put(ROOT, "along", 3).unwrap());
        a.with_predecessors(&ours, &ours.get_heads(), now);
        let message = a.generate(&ours, now, bytes).expect("our heads moved");
        b.receive(&mut theirs, decode(&message).unwrap()).unwrap();
        assert_eq!(theirs.get_heads(), ours.get_heads(), "both in one message");
    }

    /// Pushed changes (`push`) do what the sync message they stand for would: the peer
    /// applies them, is level with us, and has nothing to answer; nothing goes twice.
    #[test]
    fn pushed_changes_stand_for_a_message() {
        use crate::document::testing::change;
        use automerge::{ROOT, transaction::Transactable};
        let (mut ours, mut theirs) = (Document::new(), Document::new());
        change(&mut ours, |doc| doc.put(ROOT, "early", 1).unwrap());
        let (mut a, mut b) = in_step(&mut ours, &mut theirs);
        let (now, bytes) = (Instant::now(), SyncBytes::default());

        change(&mut ours, |doc| doc.put(ROOT, "news", 2).unwrap());
        let (heads, changes) = a.push(&ours, 1 << 20).expect("news to push");
        assert_eq!(changes.len(), 1);
        assert!(
            heads.is_empty(),
            "our heads are the change's: they go as none"
        );
        a.assume_delivered(&ours.get_heads());
        a.send_succeeded(a.id());
        assert!(a.push(&ours, 1 << 20).is_none(), "sent already");

        let before = theirs.get_heads();
        let message = Received::pushed(heads, changes).expect("decodes");
        assert_eq!(message.heads(), ours.get_heads(), "found in the changes");
        b.receive(&mut theirs, message).unwrap();
        b.came_from_peer(&theirs, &before);
        assert_eq!(theirs.get_heads(), ours.get_heads());
        assert!(b.answered_as_delivered(&theirs, &theirs.get_heads()));
        assert!(
            b.generate(&theirs, now, bytes).is_none(),
            "nothing to answer"
        );
    }

    /// A push says the sender has its heads, no more: what we have beyond them goes
    /// back with the answer, as it would to an Automerge message.
    #[test]
    fn what_the_pusher_lacks_goes_back() {
        use crate::document::testing::change;
        use automerge::{ROOT, transaction::Transactable};
        let (mut ours, mut theirs) = (Document::new(), Document::new());
        change(&mut ours, |doc| doc.put(ROOT, "early", 1).unwrap());
        let (mut a, mut b) = in_step(&mut ours, &mut theirs);
        let (now, bytes) = (Instant::now(), SyncBytes::default());

        change(&mut theirs, |doc| doc.put(ROOT, "theirs", 3).unwrap());
        let only_theirs = theirs.get_heads();
        change(&mut ours, |doc| doc.put(ROOT, "news", 2).unwrap());
        let (heads, changes) = a.push(&ours, 1 << 20).expect("news to push");
        a.assume_delivered(&ours.get_heads());
        a.send_succeeded(a.id());

        let before = theirs.get_heads();
        b.receive(
            &mut theirs,
            Received::pushed(heads, changes).expect("decodes"),
        )
        .unwrap();
        b.came_from_peer(&theirs, &before);
        assert!(!b.answered_as_delivered(&theirs, &theirs.get_heads()));
        let answer = b.generate(&theirs, now, bytes).expect("an answer");
        let answer = decode(&answer).unwrap();
        assert!(answer.carries_changes(), "with what the pusher lacks");
        a.receive(&mut ours, answer).unwrap();
        assert!(ours.get_change_by_hash(&only_theirs[0]).is_some());
        assert_eq!(ours.get_heads(), theirs.get_heads());
    }

    /// A push leaves the receiver where Automerge's own message would have: same
    /// document, same answer, and the sender's next turn the same. If an Automerge
    /// upgrade changes what a sync message's parts mean, this fails, not the swarm.
    #[test]
    fn pushes_do_what_automerges_messages_do() {
        use crate::document::testing::change;
        use automerge::{ROOT, transaction::Transactable};
        let (now, bytes) = (Instant::now(), SyncBytes::default());
        for concurrent in [false, true] {
            let (mut ours, mut theirs) = (Document::new(), Document::new());
            change(&mut ours, |doc| doc.put(ROOT, "early", 1).unwrap());
            let (a, b) = in_step(&mut ours, &mut theirs);
            if concurrent {
                change(&mut theirs, |doc| doc.put(ROOT, "theirs", 3).unwrap());
            }
            change(&mut ours, |doc| doc.put(ROOT, "news", 2).unwrap());

            // Each way: what the receiver holds, whether it answers, with what, and
            // what the sender holds and says next once it has the answer.
            let outcome = |pushing: bool| {
                let (mut a, mut b) = (a.clone(), b.clone());
                let (mut ours, mut theirs) = (ours.clone(), theirs.clone());
                let message = if pushing {
                    let (heads, changes) = a.push(&ours, 1 << 20).expect("news to push");
                    Received::pushed(heads, changes).expect("decodes")
                } else {
                    decode(&a.generate(&ours, now, bytes).expect("news")).unwrap()
                };
                a.assume_delivered(&ours.get_heads());
                a.send_succeeded(a.id());
                let before = theirs.get_heads();
                b.receive(&mut theirs, message).unwrap();
                b.came_from_peer(&theirs, &before);
                let level = b.answered_as_delivered(&theirs, &theirs.get_heads());
                let answer = b.generate(&theirs, now, bytes).map(|m| decode(&m).unwrap());
                let answered = answer
                    .as_ref()
                    .map(|m| (m.heads().to_vec(), m.change_bytes()));
                if let Some(answer) = answer {
                    a.receive(&mut ours, answer).unwrap();
                }
                let next = a
                    .generate(&ours, now, bytes)
                    .map(|m| decode(&m).unwrap().change_bytes());
                (theirs.get_heads(), level, answered, ours.get_heads(), next)
            };
            assert_eq!(outcome(true), outcome(false), "concurrent: {concurrent}");
        }
    }

    /// A push without changes, or without heads and changes that don't decode, is no
    /// message: no heads would say the sender lost all it had.
    #[test]
    fn broken_pushes_are_refused() {
        let hash = crate::document::ChangeHash([1; 32]);
        assert!(Received::pushed(vec![hash], Vec::new()).is_none());
        assert!(Received::pushed(Vec::new(), vec![vec![1, 2, 3]]).is_none());
    }

    /// Tips are the changes none of the others depend on, sorted as heads are.
    #[test]
    fn tips_are_what_nothing_depends_on() {
        let hash = |n: u8| crate::document::ChangeHash([n; 32]);
        let (a, b, c, d) = (hash(1), hash(2), hash(3), hash(4));
        let deps = [vec![], vec![a], vec![a], vec![b, c, hash(9)]];
        let all = [a, b, c, d];
        let tips_of = |which: &[usize]| tips(which.iter().map(|&i| (all[i], deps[i].as_slice())));
        assert_eq!(tips_of(&[0, 1, 2, 3]), vec![d]);
        assert_eq!(tips_of(&[2, 0, 1]), vec![b, c]);
        assert_eq!(tips_of(&[]), Vec::new());
    }

    /// Changes larger than the limit aren't pushed: they go as `generate` sends them.
    #[test]
    fn large_changes_are_not_pushed() {
        use crate::document::testing::change;
        use automerge::{ROOT, transaction::Transactable};
        let (mut ours, mut theirs) = (Document::new(), Document::new());
        change(&mut ours, |doc| doc.put(ROOT, "early", 1).unwrap());
        let (mut a, _b) = in_step(&mut ours, &mut theirs);
        change(&mut ours, |doc| {
            doc.put(ROOT, "news", "x".repeat(100)).unwrap()
        });
        assert!(a.push(&ours, 10).is_none());
        assert!(a.sending.is_none(), "nothing handed over");
        assert!(a.push(&ours, 1 << 20).is_some());
    }

    /// Our transport refused a message taken as delivered: the peer's heads are as
    /// before, so the next message carries the change again.
    #[test]
    fn a_refused_delivered_message_goes_again() {
        use crate::document::testing::change;
        use automerge::{ROOT, transaction::Transactable};
        let (mut ours, mut theirs) = (Document::new(), Document::new());
        change(&mut ours, |doc| doc.put(ROOT, "early", 1).unwrap());
        let (mut a, mut b) = in_step(&mut ours, &mut theirs);
        let (now, bytes) = (Instant::now(), SyncBytes::default());

        change(&mut ours, |doc| doc.put(ROOT, "news", 2).unwrap());
        let _refused = a.generate(&ours, now, bytes).unwrap();
        a.assume_delivered(&ours.get_heads());
        assert!(a.send_failed(a.id(), now).is_some());
        let again = a.generate(&ours, now, bytes).expect("sent again");
        a.send_succeeded(a.id());
        b.receive(&mut theirs, decode(&again).unwrap()).unwrap();
        assert_eq!(ours.get_heads(), theirs.get_heads());
    }

    /// A joining node opens, the answer is too large and becomes a delta, the node gets
    /// the delta and says so: the answer to that carries our heads, so the node knows it
    /// has everything (`have_all_theirs`).
    #[test]
    fn after_a_delta_the_peer_learns_our_heads() {
        use crate::document::testing::change;
        use automerge::{ROOT, transaction::Transactable};

        let mut rng = crate::testrand::Rng::new(2);
        let big: String = (0..1024)
            .map(|_| format!("{:016x}", rng.next_u64()))
            .collect();
        let mut ours = Document::new();
        change(&mut ours, |ours| ours.put(ROOT, "big", big).unwrap());
        let mut theirs = Document::new();
        let (mut a, mut b) = (session(), session());
        let now = Instant::now();
        let bytes = SyncBytes::default();

        // b (joining) opens; a's answer would carry everything: withdrawn for a delta.
        let opening = b.generate(&theirs, now, bytes).unwrap();
        b.send_succeeded(b.id());
        a.receive(&mut ours, decode(&opening).unwrap()).unwrap();
        assert!(a.generate(&ours, now, bytes).is_some());
        a.withdraw_for_delta(&ours, now);
        a.send_succeeded(a.id());

        // b loads the delta and tells a its heads, which are a's now.
        theirs.merge(&mut ours.clone()).unwrap();
        assert!(!b.have_all_theirs(&theirs), "b hasn't heard a's heads yet");
        let after_delta = b.generate(&theirs, now, bytes).unwrap();
        b.send_succeeded(b.id());
        a.receive(&mut ours, decode(&after_delta).unwrap()).unwrap();

        let answer = a
            .generate(&ours, now, bytes)
            .expect("a answers with its heads");
        b.receive(&mut theirs, decode(&answer).unwrap()).unwrap();
        assert!(b.have_all_theirs(&theirs));
    }
}