From caf09e234a4bd7a330c0595bfea932d5ebfa865f Mon Sep 17 00:00:00 2001 From: tv Date: Sun, 25 Oct 2026 23:35:18 +0100 Subject: split state module --- src/state/session.rs | 155 +++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 155 insertions(+) create mode 100644 src/state/session.rs (limited to 'src/state/session.rs') diff --git a/src/state/session.rs b/src/state/session.rs new file mode 100644 index 0000000..94f9d71 --- /dev/null +++ b/src/state/session.rs @@ -0,0 +1,155 @@ +//! One Automerge sync with a peer, from a reset to the next. + +use std::time::Duration; + +use automerge::sync; +use tokio::time::Instant; + +use crate::backoff::Backoff; +use crate::swarm::throughput::SyncBytes; + +/// A sync message on its way to our 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, + /// 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, +} + +/// First wait before generating again after our Mycelium 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 { + pub(crate) state: sync::State, + + /// A message being handed to our Mycelium. + pub(crate) sending: Option, + + /// After our Mycelium 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, + /// The wait after the next refusal. + pub(crate) send_backoff: Backoff, + + /// The answer we expect to our last sync message(s), if any. + pub(crate) reply: Option, + /// 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, +} + +impl SyncSession { + /// `reply_backoff`: waits for sync answers (`Timing::sync_reply_backoff`). + pub(crate) fn new(reply_backoff: Backoff) -> Self { + Self { + state: sync::State::new(), + sending: None, + retry_at: None, + send_backoff: Backoff::new(SEND_RETRY, SEND_RETRY_MAX), + reply: None, + reply_backoff, + } + } + + /// 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 { + self.state.their_heads.is_none() && we_ready && peer_ready == Some(false) + } + + /// Our Mycelium took the message. + pub(crate) fn send_succeeded(&mut self) -> bool { + self.retry_at = None; + self.send_backoff.reset(); + self.sending + .take() + .is_some_and(|sending| sending.regenerate) + } + + /// Our Mycelium 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. + pub(crate) fn send_failed(&mut self, now: Instant) -> Option { + let sending = self.sending.take()?; + for hash in &sending.sent { + self.state.sent_hashes.remove(hash); + } + 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 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, + now: Instant, + bytes: SyncBytes, + ) { + match &mut self.reply { + // Still waiting for an earlier one: keep its deadline, remember these too. + Some(wait) => wait.sent.extend(sent), + None => { + self.reply = Some(SyncReplyWait { + deadline: now + self.reply_backoff.current(), + started: (now, bytes), + sent, + }); + } + } + } + + /// 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(super) fn take_overdue_reply(&mut self, now: Instant) -> Option { + 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.reply_backoff.grow(); + Some(wait.sent.len()) + } +} -- cgit v1.3.1