path: root/src/state/peer.rs

//! 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)
    }
}