From 3be4ec3aa0e076e5386cfd1b93b07d12c40a88cb Mon Sep 17 00:00:00 2001 From: tv Date: Mon, 26 Oct 2026 06:02:14 +0100 Subject: add session IDs This prevents the regenerate flag from being lost across reset_sync during an in-flight send. --- src/state/session.rs | 68 ++++++++++++++++++++++++++++++++++++++++++++++++---- 1 file changed, 64 insertions(+), 4 deletions(-) (limited to 'src/state/session.rs') diff --git a/src/state/session.rs b/src/state/session.rs index 0dcc76d..618d65b 100644 --- a/src/state/session.rs +++ b/src/state/session.rs @@ -6,6 +6,7 @@ //! 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::sync::{self, SyncDoc}; @@ -26,6 +27,9 @@ pub(crate) struct Sending { pub(crate) regenerate: bool, } +/// Session IDs, unique within the process. +static NEXT_SESSION: AtomicU64 = AtomicU64::new(1); + /// First wait before generating again after our Mycelium refused a message; doubles /// while it keeps refusing. const SEND_RETRY: Duration = Duration::from_millis(500); @@ -35,6 +39,8 @@ const SEND_RETRY_MAX: Duration = Duration::from_secs(30); /// 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 Mycelium. @@ -69,6 +75,7 @@ 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, @@ -193,8 +200,20 @@ impl SyncSession { 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 { + /// 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 Mycelium 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 @@ -205,8 +224,11 @@ impl SyncSession { /// 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 { + /// already discarded what was being sent (the message came from another `session`). + pub(crate) fn send_failed(&mut self, session: u64, now: Instant) -> Option { + if session != self.id { + return None; + } let sending = self.sending.take()?; for hash in &sending.sent { self.state.sent_hashes.remove(hash); @@ -264,3 +286,41 @@ impl SyncSession { Some(wait.sent.len()) } } + +#[cfg(test)] +mod tests { + use super::*; + + fn session() -> SyncSession { + SyncSession::new(Backoff::new( + Duration::from_secs(1), + Duration::from_secs(30), + )) + } + + fn sending() -> Option { + Some(Sending { + sent: Vec::new(), + regenerate: false, + }) + } + + #[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()); + } +} -- cgit v1.3.1