record the cause that started a sync session

3 files changed, 74 insertions, 12 deletions

diff --git a/src/state.rs b/src/state.rs
index eaa6562..5e9b03b 100644
--- a/src/state.rs
+++ b/src/state.rs
@@ -139,6 +139,9 @@ pub(crate) struct SyncSession {
     /// Who sends the first message of this session.
     pub(crate) opening: Opening,
 
+    /// What started this session.
+    pub(crate) cause: SessionCause,
+
     /// Consecutive exchanges that changed nothing (livelock detection).
     pub(crate) stalled_exchanges: u32,
 }
@@ -160,6 +163,49 @@ pub(crate) enum Opening {
     Theirs { until: Instant },
 }
 
+/// What started a sync session. Kept for the log: when the peer doesn't open within the
+/// grace period, the cause usually tells why.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(crate) enum SessionCause {
+    /// First session with a peer we just learned about.
+    NewPeer,
+    /// The peer sent HELLO: it (re)started its handshake with us.
+    PeerHello,
+    /// Its HELLO_REPLY completed our handshake.
+    HandshakeComplete,
+    /// The peer advertised another epoch than before.
+    PeerEpochChanged,
+    /// We switched epochs.
+    OwnEpochSwitch,
+    /// The peer went dormant.
+    PeerDormant,
+    /// The peer failed the swarm key check.
+    KeyRejected,
+}
+
+impl SessionCause {
+    /// The likely reason the peer didn't open a session started this way.
+    pub(crate) fn missing_opening_hint(self) -> &'static str {
+        match self {
+            Self::NewPeer | Self::PeerHello | Self::HandshakeComplete => {
+                "the peer may not count us as a member yet (our HELLO_REPLY hasn't reached it), \
+                 or its opening was lost"
+            }
+            Self::PeerEpochChanged => {
+                "the peer switched epochs but doesn't know we're on its new one yet, or its \
+                 opening was lost"
+            }
+            Self::OwnEpochSwitch => {
+                "we switched epochs; the peer opens only once it learns that, and it may not \
+                 have yet"
+            }
+            Self::PeerDormant | Self::KeyRejected => {
+                "only our side reset, so the peer had no reason to open"
+            }
+        }
+    }
+}
+
 /// 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.
@@ -181,8 +227,9 @@ const SYNC_REPLY_TIMEOUT_MAX: Duration = Duration::from_secs(300);
 const SYNC_OPENING_GRACE: Duration = Duration::from_secs(3);
 
 impl SyncSession {
-    pub(crate) fn new(leads: bool) -> Self {
+    pub(crate) fn new(leads: bool, cause: SessionCause) -> Self {
         Self {
+            cause,
             state: sync::State::new(),
             outbound: None,
             reply: None,
@@ -212,6 +259,12 @@ impl SyncSession {
         }
     }
 
+    /// The peer should have opened by now but hasn't said anything: we open ourselves.
+    pub(crate) fn opening_overdue(&self, now: Instant) -> bool {
+        self.state.their_heads.is_none()
+            && matches!(self.opening, Opening::Theirs { until } if until <= now)
+    }
+
     /// A sync message went out; `sent` are the changes it carried for the first time.
     pub(crate) fn expect_reply(&mut self, sent: Vec<automerge::ChangeHash>, now: Instant) {
         match &mut self.reply {
@@ -258,7 +311,7 @@ impl PeerState {
             hello_state: HelloState::default(),
             reply_owed: false,
             pending_inbound_sync: Vec::new(),
-            sync: SyncSession::new(leads),
+            sync: SyncSession::new(leads, SessionCause::NewPeer),
             liveness: Liveness::Active,
             added_at: Instant::now(),
             last_seen: None,
@@ -271,8 +324,8 @@ impl PeerState {
 
     /// Start Automerge sync with this peer from scratch. Any envelope generated against
     /// the old state is discarded; a stale RetryDelivery event is then a no-op.
-    pub(crate) fn reset_sync(&mut self) {
-        self.sync = SyncSession::new(self.leads);
+    pub(crate) fn reset_sync(&mut self, cause: SessionCause) {
+        self.sync = SyncSession::new(self.leads, cause);
     }
 
     /// An overdue sync answer from an established, active peer (see
@@ -310,7 +363,7 @@ impl PeerState {
         }
 
         self.liveness = Liveness::Dormant { since: now };
-        self.reset_sync();
+        self.reset_sync(SessionCause::PeerDormant);
 
         if !matches!(self.hello_state, HelloState::Sending { .. }) {
             self.hello_state = HelloState::AwaitingReply {
@@ -977,7 +1030,7 @@ impl Host for AppState {
                     let now = Instant::now();
                     for peer in state.peer_states.blocking_read().values() {
                         let mut peer = peer.blocking_lock();
-                        peer.reset_sync();
+                        peer.reset_sync(SessionCause::OwnEpochSwitch);
                         // A fresh start, not an idle sync: no heal re-handshake for it.
                         peer.last_sync_rx = Some(now);
                     }
diff --git a/src/swarm/peer.rs b/src/swarm/peer.rs
index 836a967..1e8b91d 100644
--- a/src/swarm/peer.rs
+++ b/src/swarm/peer.rs
@@ -13,7 +13,7 @@ use tracing::{debug, info, instrument, trace, warn};
 
 use crate::document::replica::{Epoch, EpochRelation};
 use crate::document::store::{self, StoreError};
-use crate::state::{AppState, HelloState, Liveness, PeerState, Standing};
+use crate::state::{AppState, HelloState, Liveness, PeerState, SessionCause, Standing};
 use crate::swarm::sync::{SyncEvent, handle_sync};
 use crate::transport::{self, ParsedMessage};
 
@@ -503,7 +503,7 @@ async fn establish_peer(
     // Duplicate replies (caused by our own retries) don't reset again.
     if peer_state.hello_state != HelloState::Established {
         peer_state.hello_state = HelloState::Established;
-        peer_state.reset_sync();
+        peer_state.reset_sync(SessionCause::HandshakeComplete);
         info!(peer = %peer, "Handshake complete; sync state reset");
     }
 
@@ -539,7 +539,7 @@ async fn observe_peer_epoch(state: &AppState, peer: &str, remote: Epoch) -> bool
             // The peer switched documents; our sync state describes its old one. Count
             // the restart as sync activity, so the keepalive heal doesn't re-handshake
             // (resetting both sides again) while this fresh sync is under way.
-            peer_state.reset_sync();
+            peer_state.reset_sync(SessionCause::PeerEpochChanged);
             peer_state.last_sync_rx = Some(Instant::now());
         }
         previous
@@ -623,7 +623,7 @@ async fn admit(state: &AppState, peer: &str, message: &ParsedMessage) -> bool {
                 }
                 peer_state.standing = Standing::Stranger;
                 peer_state.hello_state = HelloState::NotSent;
-                peer_state.reset_sync();
+                peer_state.reset_sync(SessionCause::KeyRejected);
             }
             // The one place an operator can see why a node can't join: rate limited,
             // since anyone can send us handshakes.
@@ -690,7 +690,7 @@ pub(crate) async fn handle_message(
             {
                 let peer_state_mutex = state.get_or_create_peer_state(&peer).await;
                 let mut peer_state = peer_state_mutex.lock().await;
-                peer_state.reset_sync();
+                peer_state.reset_sync(SessionCause::PeerHello);
                 peer_state.reply_owed = true;
             }
 
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 1335361..dc2e665 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -63,7 +63,8 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool
         };
 
         if !retry_delivery {
-            if let Some(wait) = peer_state.sync.opening_wait(Instant::now()) {
+            let now = Instant::now();
+            if let Some(wait) = peer_state.sync.opening_wait(now) {
                 trace!(
                     wait_ms = wait.as_millis(),
                     "Leaving the opening sync message to the peer"
@@ -76,6 +77,14 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool
                 });
                 return Ok(());
             }
+            if peer_state.sync.opening_overdue(now) {
+                debug!(
+                    cause = ?peer_state.sync.cause,
+                    reply_owed = peer_state.reply_owed,
+                    likely = peer_state.sync.cause.missing_opening_hint(),
+                    "The peer didn't open the sync within the grace period; opening ourselves"
+                );
+            }
         }
 
         swarm_id.clone()