1 files changed, 352 insertions, 0 deletions
diff --git a/src/state/peer.rs b/src/state/peer.rs
new file mode 100644
index 0000000..e96cd9c
--- /dev/null
+++ b/src/state/peer.rs
@@ -0,0 +1,352 @@
+//! What a node knows about each peer: standing, liveness, handshake, incidents.
+
+use std::time::Duration;
+
+use tokio::time::Instant;
+
+use super::{SyncSession, Timing};
+use crate::backoff::Backoff;
+use crate::document::replica::Epoch;
+use crate::transport;
+
+/// How far a peer got into the swarm, as far as we're concerned.
+#[derive(Debug, Clone, Default, PartialEq, Eq)]
+pub(crate) enum Standing {
+    /// Nothing proven yet. With a swarm key, everything but a handshake from it is
+    /// ignored (`swarm::peer::admit`).
+    #[default]
+    Stranger,
+    /// Proved the swarm key in a HELLO; no HELLO_REPLY yet. Only reached with a key.
+    Admitted,
+    /// Its HELLO_REPLY named our swarm (and, with a key, proved it): we sync with it.
+    Member { swarm_id: String },
+}
+
+impl Standing {
+    /// The swarm a member belongs to; None before its HELLO_REPLY.
+    pub(crate) fn swarm_id(&self) -> Option<&str> {
+        match self {
+            Self::Member { swarm_id } => Some(swarm_id),
+            Self::Stranger | Self::Admitted => None,
+        }
+    }
+
+    /// Proved the swarm key (meaningful only with one).
+    pub(crate) fn is_admitted(&self) -> bool {
+        !matches!(self, Self::Stranger)
+    }
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(crate) enum Liveness {
+    Active,
+    /// No inbound message for Timing::dormant_after. Kept indefinitely, probed slowly.
+    Dormant {
+        since: Instant,
+    },
+}
+
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+pub(crate) enum HelloState {
+    #[default]
+    NotSent,
+    /// A send is in flight.
+    Sending,
+    /// HELLO was sent (or the send failed); no HELLO_REPLY yet. Retried at
+    /// `next_attempt` (`PeerState::hello_backoff`, or `dormant_probe` while dormant).
+    AwaitingReply { next_attempt: Instant },
+    /// HELLO_REPLY received. No more retries.
+    Established,
+}
+
+/// First wait for a HELLO_REPLY before sending the HELLO again; doubles while none
+/// comes.
+const HELLO_RETRY: Duration = Duration::from_secs(1);
+const HELLO_RETRY_MAX: Duration = Duration::from_secs(30);
+
+#[derive(Clone)]
+pub(crate) struct PeerState {
+    /// Whether we lead the pair (`swarm::pair`): we send the keepalive PINGs. Fixed for
+    /// the pair, so computed once.
+    pub(crate) leads: bool,
+
+    pub(crate) standing: Standing,
+    pub(crate) hello_state: HelloState,
+    /// Wait between unanswered HELLOs; reset whenever a new handshake starts.
+    pub(crate) hello_backoff: Backoff,
+
+    /// Everything about the current Automerge sync with this peer; a reset replaces it.
+    pub(crate) sync: SyncSession,
+
+    pub(crate) liveness: Liveness,
+    pub(crate) added_at: Instant,
+    pub(crate) last_seen: Option<Instant>,
+    pub(crate) last_ping_at: Option<Instant>,
+
+    /// Last time a sync message from this peer was applied (survives resets: the
+    /// keepalive heal looks at it).
+    pub(crate) last_sync_rx: Option<Instant>,
+
+    /// Last self-healing re-handshake (cooldown).
+    pub(crate) last_resync_at: Option<Instant>,
+
+    /// Since when keepalives show the peer's heads differ from ours with no sync message
+    /// from it in between (None while they match).
+    pub(crate) diverged_since: Option<Instant>,
+
+    /// Epoch the peer advertised in its most recent message (late ones aside: see
+    /// `epoch_seq`).
+    pub(crate) epoch: Option<Epoch>,
+    /// The peer's `epoch_seq` with `epoch`: how many times it changed epoch in its run.
+    /// An epoch with a lower one is older than what we have, delivered late.
+    pub(crate) epoch_seq: Option<u32>,
+    /// The peer's run, from its HELLO or HELLO_REPLY: a new one starts `epoch_seq` over.
+    pub(crate) run_id: Option<crate::transport::RunId>,
+
+    /// Whether the peer was ready according to its most recent sync message or keepalive.
+    pub(crate) reports_ready: Option<bool>,
+
+    /// Whether the peer said, in its most recent keepalive, that it would sync with us.
+    pub(crate) syncs_with_us: Option<bool>,
+
+    /// How many heads its most recent keepalive listed (None if it sent a digest).
+    pub(crate) reported_heads: Option<usize>,
+
+    /// HELLOs sent in the current handshake attempt without hearing anything back.
+    pub(crate) hello_attempts: u32,
+    /// Since when we consider the peer unreachable (three unanswered HELLOs, or
+    /// dormant). Cleared by any message from it; ignored after `dormant_probe`, so the
+    /// peer becomes a neighbour candidate again and gets tried once more.
+    pub(crate) unreachable_since: Option<Instant>,
+
+    /// What went wrong with this peer so far (counted only, for /peers and /admin/stats).
+    pub(crate) incidents: Incidents,
+}
+
+/// Trouble with a peer, counted since it was first listed. Lost keepalives and refused
+/// sends are in the traffic counters (`transport::traffic`): every PONG answers a PING.
+#[derive(Debug, Default, Clone, Copy, serde::Serialize, utoipa::ToSchema)]
+#[serde(rename_all = "camelCase")]
+pub(crate) struct Incidents {
+    /// Our sync messages left unanswered until the reply timeout (then sent again).
+    pub(crate) reply_timeouts: u64,
+    /// Re-handshakes because our documents differed while sync was idle.
+    pub(crate) heals: u64,
+    /// Times the peer went dormant.
+    pub(crate) dormancies: u64,
+    /// Time spent dormant, in milliseconds, up to its last return (the current
+    /// dormancy, if any, isn't included).
+    pub(crate) dormant_ms: u64,
+}
+
+impl Incidents {
+    pub(crate) fn add(&mut self, other: &Incidents) {
+        self.reply_timeouts += other.reply_timeouts;
+        self.heals += other.heals;
+        self.dormancies += other.dormancies;
+        self.dormant_ms += other.dormant_ms;
+    }
+}
+
+/// Unanswered HELLOs after which a peer counts as unreachable (1 + 2 + 4 s apart).
+const UNREACHABLE_AFTER_HELLOS: u32 = 3;
+
+impl PeerState {
+    pub(crate) fn new(leads: bool, timing: &Timing) -> Self {
+        Self {
+            leads,
+            standing: Standing::default(),
+            hello_state: HelloState::default(),
+            hello_backoff: Backoff::new(HELLO_RETRY, HELLO_RETRY_MAX),
+            sync: SyncSession::new(timing.sync_reply_backoff()),
+            liveness: Liveness::Active,
+            added_at: Instant::now(),
+            last_seen: None,
+            last_ping_at: None,
+            last_sync_rx: None,
+            last_resync_at: None,
+            diverged_since: None,
+            epoch: None,
+            epoch_seq: None,
+            run_id: None,
+            reports_ready: None,
+            syncs_with_us: None,
+            reported_heads: None,
+            hello_attempts: 0,
+            unreachable_since: None,
+            incidents: Incidents::default(),
+        }
+    }
+
+    /// Start Automerge sync with this peer from scratch. A message still being handed to
+    /// Mycelium was generated against the old state; its outcome is then ignored.
+    pub(crate) fn reset_sync(&mut self) {
+        // The same waits for answers, from the start.
+        let mut reply_backoff = self.sync.reply_backoff;
+        reply_backoff.reset();
+        self.sync = SyncSession::new(reply_backoff);
+    }
+
+    /// An overdue sync answer from an established, active peer (see
+    /// `SyncSession::take_overdue_reply`).
+    pub(crate) fn take_overdue_sync_reply(&mut self, now: Instant) -> Option<usize> {
+        if self.liveness != Liveness::Active || self.hello_state != HelloState::Established {
+            return None;
+        }
+        let overdue = self.sync.take_overdue_reply(now);
+        if overdue.is_some() {
+            self.incidents.reply_timeouts += 1;
+        }
+        overdue
+    }
+
+    fn quiet_for(&self, now: Instant) -> Duration {
+        now.saturating_duration_since(self.last_seen.unwrap_or(self.added_at))
+    }
+
+    /// Records inbound activity. Returns true if the peer was dormant.
+    pub(crate) fn mark_seen(&mut self, now: Instant) -> bool {
+        self.last_seen = Some(now);
+        self.hello_attempts = 0;
+        self.unreachable_since = None;
+
+        if let Liveness::Dormant { since } = self.liveness {
+            self.incidents.dormant_ms += now.saturating_duration_since(since).as_millis() as u64;
+            self.liveness = Liveness::Active;
+            // Force a fresh handshake; its reply resets sync on both sides.
+            if self.hello_state != HelloState::Sending {
+                self.hello_state = HelloState::NotSent;
+            }
+            return true;
+        }
+        false
+    }
+
+    /// A HELLO went out (or failed to). After the third unanswered one, the peer counts
+    /// as unreachable.
+    pub(crate) fn hello_sent(&mut self, now: Instant) {
+        self.hello_attempts += 1;
+        if self.hello_attempts >= UNREACHABLE_AFTER_HELLOS {
+            self.unreachable_since.get_or_insert(now);
+        }
+    }
+
+    /// Whether to leave the peer out when choosing neighbours (`swarm::topology`).
+    pub(crate) fn is_unreachable(&self, timing: &Timing, now: Instant) -> bool {
+        self.unreachable_since
+            .is_some_and(|since| now.saturating_duration_since(since) < timing.dormant_probe)
+    }
+
+    /// No more HELLO retries (the peer isn't our neighbour any more). A HELLO in flight
+    /// finishes; a later handshake starts over.
+    pub(crate) fn stop_hello_retries(&mut self) {
+        if matches!(self.hello_state, HelloState::AwaitingReply { .. }) {
+            self.hello_state = HelloState::NotSent;
+        }
+    }
+
+    /// Moves a silent peer to Dormant. Returns true on the transition.
+    pub(crate) fn check_dormancy(&mut self, timing: &Timing, now: Instant) -> bool {
+        if self.liveness != Liveness::Active || self.quiet_for(now) < timing.dormant_after {
+            return false;
+        }
+
+        self.liveness = Liveness::Dormant { since: now };
+        self.incidents.dormancies += 1;
+        self.unreachable_since.get_or_insert(now);
+        self.reset_sync();
+
+        if self.hello_state != HelloState::Sending {
+            self.hello_state = HelloState::AwaitingReply {
+                next_attempt: now + timing.dormant_probe,
+            };
+        }
+        true
+    }
+
+    /// True (and records the ping) if an established, quiet peer needs a keepalive.
+    ///
+    /// One PING and its PONG tell both sides that the other is alive, so only one side of
+    /// a pair needs to ping: the one leading the pair (`leads`). Each of its PINGs
+    /// makes the other side's peer look recently heard from, so the other side never
+    /// pings while it does. If the pings stop, the other side pings once it has heard
+    /// nothing for half an interval more. Without that, both sides would get due at about
+    /// the same time (each one's quiet time starts with the same exchange) and their
+    /// PINGs would often cross.
+    pub(crate) fn take_keepalive_due(&mut self, timing: &Timing, now: Instant) -> bool {
+        let quiet_enough = if self.leads {
+            timing.keepalive
+        } else {
+            timing.keepalive + timing.keepalive / 2
+        };
+        let due = self.liveness == Liveness::Active
+            && self.hello_state == HelloState::Established
+            && now.saturating_duration_since(self.added_at) >= timing.keepalive
+            && self.quiet_for(now) >= quiet_enough
+            && self
+                .last_ping_at
+                .is_none_or(|t| now.saturating_duration_since(t) >= timing.keepalive);
+
+        if due {
+            self.last_ping_at = Some(now);
+        }
+        due
+    }
+
+    /// Self-healing check on PING/PONG. Returns true if a re-handshake should start.
+    pub(crate) fn needs_heal(
+        &mut self,
+        local: &transport::HeadsSummary,
+        remote: &transport::HeadsSummary,
+        timing: &Timing,
+        now: Instant,
+    ) -> bool {
+        if local == remote {
+            self.diverged_since = None;
+            return false;
+        }
+
+        // How long the heads have differed without sync making progress. Measured from
+        // when we noticed, not from the last sync message: in a quiet swarm that one is
+        // long past, and the first keepalive after a lost message would heal at once,
+        // before the sender's reply timeout could resend it. Any sync message restarts
+        // the clock.
+        let noticed = match self.diverged_since {
+            Some(since) if self.last_sync_rx.is_none_or(|rx| rx <= since) => since,
+            _ => *self.diverged_since.insert(now),
+        };
+
+        // While an answer to our last sync message is due, a lost message is the reply
+        // timeout's to resend; healing is for when neither side has anything outstanding.
+        if self.liveness != Liveness::Active
+            || self.hello_state != HelloState::Established
+            || self.sync.reply.is_some()
+        {
+            return false;
+        }
+
+        let persisted = now.saturating_duration_since(noticed) >= timing.resync_after;
+        let cooled_down = self
+            .last_resync_at
+            .is_none_or(|t| now.saturating_duration_since(t) >= timing.resync_after);
+
+        if !(persisted && cooled_down) {
+            return false;
+        }
+        self.diverged_since = None;
+        self.incidents.heals += 1;
+
+        self.last_resync_at = Some(now);
+        self.hello_state = HelloState::NotSent; // establish_peer resets sync on the reply
+        true
+    }
+
+    /// Heard from recently. A returning node counts as alive from its first message,
+    /// before the handshake finishes, so a swarm-wide forget can't make it flap.
+    pub(crate) fn is_alive(&self, timing: &Timing, now: Instant) -> bool {
+        self.liveness == Liveness::Active
+            && self
+                .last_seen
+                .is_some_and(|t| now.saturating_duration_since(t) < timing.dormant_after)
+    }
+}