//! 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::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);

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()
    }
}

/// `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,
    /// 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),

    /// 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,
}

/// 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,
            before_receive: None,
            before_generate: (Vec::new(), false),
            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());
        doc.receive_sync_message_log_patches(&mut self.state, message.0, log)
    }

    /// 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 the
    /// peer's heads as far as we know them: where the delta starts.
    pub(crate) fn withdraw_for_delta(&mut self) -> Vec<crate::document::ChangeHash> {
        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.state.their_heads.clone().unwrap_or_default()
    }

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

    /// 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(),
        }
    }

    /// 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())
            });
            let Some(before) = before.filter(|before| older && level(&before.their_heads)) else {
                return false;
            };
            self.state.their_heads = before.their_heads;
            self.state.their_have = before.their_have;
            self.state.their_need = before.their_need;
            self.state.shared_heads = heads.to_vec();
        }
        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`.
    pub(crate) fn resend_if_needed(&mut self, doc: &Document, peer: &str) {
        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"
            );
        }
    }

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

    /// 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.
    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()?;
        for hash in &wait.sent {
            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 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 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();
        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");
    }

    /// 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");
        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");
        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");
        let again = send(&mut a, &ours);
        let carried = sync::Message::decode(&again.0).unwrap().changes.len();
        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());
        let message = a.generate(&ours, now, bytes).unwrap();
        let carried = sync::Message::decode(&message.0).unwrap().changes.len();
        assert_eq!(carried, 1, "only the news");
    }

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