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::Timing;
use crate::backoff::Backoff;
use crate::document::replica::Epoch;
use crate::document::sync::SyncSession;
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 {
    /// Nothing heard from it since this node started: known only from the member list
    /// or another peer. It counts as active (sync, keepalives and dormancy treat it so)
    /// until its first message makes it active, or `Timing::dormant_after` of silence
    /// dormant.
    Unknown,
    Active,
    /// No inbound message for Timing::dormant_after. Kept indefinitely, probed slowly.
    Dormant {
        since: Instant,
    },
}

impl Liveness {
    pub(crate) fn is_dormant(self) -> bool {
        matches!(self, Self::Dormant { .. })
    }
}

#[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 `PeerState::probe_backoff` while
    /// unreachable).
    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);
/// How long after it became known (`added_at`) a peer never heard from is HELLOed that
/// often, unreachable or not, rather than probed (`dormant_probe`, ten keepalives):
/// members that restart together may fail to reach each other at first, as the others
/// aren't up yet, and would otherwise wait for a probe to find each other.
const FIRST_CONTACT: Duration = Duration::from_secs(300);

#[derive(Clone)]
pub(crate) struct PeerState {
    /// Whether we lead the pair (`swarm::pair`): we send the first keepalive. 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,
    /// Wait between HELLOs to an unreachable peer (probes): `dormant_probe`, doubling
    /// while none is answered. Reset by any message from it.
    pub(crate) probe_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>,
    /// Our last keepalive to it (a PING or a STATUS).
    pub(crate) last_keepalive_at: Option<Instant>,
    /// Its last keepalive to us (a PING or a STATUS): after it, the next is ours.
    pub(crate) keepalive_received_at: Option<Instant>,
    /// The last STATUS between us, either way (`status_refresh`).
    pub(crate) last_status_at: Option<Instant>,
    /// What we wait for from it with nothing heard from it since: an answer to a
    /// keepalive, or to a sync message whose answer is overdue (`AwaitingAnswer`).
    /// Cleared by any message from it (`mark_seen`).
    awaiting_answer: Option<AwaitingAnswer>,

    /// 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>,
    /// The newest handshake the peer started with us: its run and number
    /// (`HelloRequest::handshake`). A HELLO repeating it is a retry or a late copy.
    pub(crate) last_hello: Option<(crate::transport::RunId, u32)>,
    /// The number of our current handshake with the peer (`AppState::handshakes`).
    pub(crate) handshake: u32,

    /// 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>,
    /// The heads its most recent STATUS or STATUS_REPLY carried (an announcement's too),
    /// with its epoch: once they are ours, it has every change we would announce to it
    /// (`swarm::relay`).
    pub(crate) last_heads: Option<(transport::Epoch, transport::HeadsSummary)>,
    /// Whom it said it reaches (`swarm::reach`), by pair key, and when.
    pub(crate) reaches: Option<(Instant, Vec<u64>)>,
    /// Whose change the next sync message to it passes on along the writer's tree
    /// (`swarm::tree`); taken when it is generated.
    pub(crate) tree_writer: Option<u64>,

    /// It said farewell (`say_farewell`): what else that run of it still sends while it
    /// stops doesn't make it active again; a message from another run does (`returned`).
    pub(crate) left: bool,

    /// 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). Only a message from it clears it (`mark_seen`): until then it is no
    /// neighbour candidate, and a member (by its HELLO_REPLY, or by the member list) gets
    /// probes (`probe_backoff`) instead.
    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,

    /// The largest sync message to it that didn't make it (its send failed, or it went
    /// unanswered), in bytes, and when: messages about that large go as deltas for a
    /// while (`swarm::blob::Blobs::delta_threshold_for`), which a slow link takes chunk
    /// by chunk.
    whole_failed: Option<(usize, Instant)>,
}

/// An answer we wait for (`PeerState::awaiting_answer`): unanswered for half of
/// `Timing::unreached_after`, the peer is asked again with a PING
/// (`take_ask_again_due`), so one lost message or answer doesn't do it; unanswered for
/// all of it, we don't count it as reached (`swarm::reach`). A sync message whose answer
/// is overdue is asked after at once: by then it waited that half or longer (the waits
/// for sync answers grow while none comes), and a peer with nothing to say answers no
/// sync message, however often it goes.
#[derive(Clone, Copy, Debug)]
struct AwaitingAnswer {
    /// When to ask again; None once asked.
    ask_again_at: Option<Instant>,
    /// When it no longer counts as reached.
    unreached_at: Instant,
}

impl AwaitingAnswer {
    /// Waiting from `now` on, for what was just sent.
    fn sent(timing: &Timing, now: Instant) -> Self {
        Self {
            ask_again_at: Some(now + timing.unreached_after / 2),
            unreached_at: now + timing.unreached_after,
        }
    }

    /// Waiting for an overdue sync answer: asked again now.
    fn overdue(timing: &Timing, now: Instant) -> Self {
        Self {
            ask_again_at: Some(now),
            unreached_at: now + timing.unreached_after / 2,
        }
    }
}

/// How long a sync message that didn't make it keeps others that large from going whole.
const WHOLE_FAILED_KEPT: Duration = Duration::from_secs(3600);

/// 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,
/// every STATUS_REPLY a STATUS.
#[derive(Debug, Default, Clone, Copy, serde::Serialize)]
#[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;
    }
}

/// What a keepalive due is (`PeerState::take_keepalive_due`).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum Keepalive {
    /// A PING: the peer's transport answers it with a PONG.
    Ping,
    /// A STATUS: our state, answered with the peer's.
    Status,
}

/// 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),
            probe_backoff: timing.probe_backoff(),
            sync: SyncSession::new(timing.sync_reply_backoff()),
            liveness: Liveness::Unknown,
            added_at: Instant::now(),
            last_seen: None,
            last_keepalive_at: None,
            keepalive_received_at: None,
            last_status_at: None,
            awaiting_answer: None,
            last_sync_rx: None,
            last_resync_at: None,
            diverged_since: None,
            epoch: None,
            epoch_seq: None,
            run_id: None,
            last_hello: None,
            handshake: 0,
            reports_ready: None,
            syncs_with_us: None,
            reported_heads: None,
            last_heads: None,
            reaches: None,
            tree_writer: None,
            left: false,
            hello_attempts: 0,
            unreachable_since: None,
            incidents: Incidents::default(),
            whole_failed: None,
        }
    }

    /// Start Automerge sync with this peer from scratch. A message still being handed to
    /// the transport 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);
        // Said over the session that ended.
        self.reaches = None;
    }

    /// An overdue sync answer from an established, active peer (see
    /// `SyncSession::take_overdue_reply`).
    pub(crate) fn take_overdue_sync_reply(
        &mut self,
        timing: &Timing,
        now: Instant,
    ) -> Option<usize> {
        if self.liveness.is_dormant() || self.hello_state != HelloState::Established {
            return None;
        }
        let largest = self.sync.reply.as_ref().map(|wait| wait.largest);
        let overdue = self.sync.take_overdue_reply(now);
        if overdue.is_some() {
            self.incidents.reply_timeouts += 1;
            self.awaiting_answer
                .get_or_insert(AwaitingAnswer::overdue(timing, now));
            if let Some(largest) = largest {
                self.whole_failed(largest, now);
            }
        }
        overdue
    }

    /// A sync message of `bytes` to it didn't make it (`whole_failed`).
    pub(crate) fn whole_failed(&mut self, bytes: usize, now: Instant) {
        let kept = self
            .whole_failed
            .filter(|(failed, at)| *failed >= bytes && now < *at + WHOLE_FAILED_KEPT);
        self.whole_failed = kept.or(Some((bytes, now)));
    }

    /// The largest sync message to it that didn't make it lately, in bytes.
    pub(crate) fn failed_whole(&self, now: Instant) -> Option<usize> {
        self.whole_failed
            .filter(|(_, at)| now < *at + WHOLE_FAILED_KEPT)
            .map(|(bytes, _)| bytes)
    }

    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. From a peer that
    /// said farewell (`left`), only that it was heard from.
    pub(crate) fn mark_seen(&mut self, now: Instant) -> bool {
        self.last_seen = Some(now);
        self.awaiting_answer = None;
        if self.left {
            return false;
        }
        self.hello_attempts = 0;
        self.unreachable_since = None;
        self.probe_backoff.reset();

        if self.liveness == Liveness::Unknown {
            self.liveness = Liveness::Active;
        }
        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 we sync with it on `epoch`: an active member on that epoch (what our
    /// STATUS tells it as `syncs_with_you`).
    pub(crate) fn syncs_on(&self, epoch: Epoch) -> bool {
        !self.liveness.is_dormant()
            && matches!(self.standing, Standing::Member { .. })
            && self.epoch == Some(epoch)
    }

    /// Whether what we sent it went unanswered, with nothing else heard from it, even
    /// once asked again (`AwaitingAnswer`): then we don't count it as reached
    /// (`swarm::reach`), long before it would be dormant.
    pub(crate) fn answer_overdue(&self, now: Instant) -> bool {
        self.awaiting_answer
            .is_some_and(|awaiting| now >= awaiting.unreached_at)
    }

    /// Whether to ask an established, active peer again, with a PING, as nothing answered
    /// us for a while (`AwaitingAnswer`); records it.
    pub(crate) fn take_ask_again_due(&mut self, now: Instant) -> bool {
        if self.liveness.is_dormant() || self.hello_state != HelloState::Established {
            return false;
        }
        match &mut self.awaiting_answer {
            Some(awaiting) if awaiting.ask_again_at.is_some_and(|at| now >= at) => {
                awaiting.ask_again_at = None;
                true
            }
            _ => false,
        }
    }

    /// Whether to leave the peer out when choosing neighbours (`swarm::topology`).
    pub(crate) fn is_unreachable(&self) -> bool {
        self.unreachable_since.is_some()
    }

    /// What choosing neighbours reads of this peer (`swarm::topology::refresh`): whether
    /// it is unreachable, whether it was ever heard from, whether it is a member, and
    /// whether it syncs with us. A change to any may change the neighbours
    /// (`swarm::topology::change_peer`).
    pub(crate) fn neighbour_inputs(&self) -> impl PartialEq + use<> {
        (
            self.is_unreachable(),
            self.last_seen.is_some(),
            matches!(self.standing, Standing::Member { .. }),
            self.syncs_with_us,
        )
    }

    /// The wait before the next HELLO: a probe's if unreachable, but while a peer never
    /// heard from is new (`FIRST_CONTACT`).
    pub(crate) fn next_hello_wait(&mut self, now: Instant) -> Duration {
        let new = self.last_seen.is_none()
            && now.saturating_duration_since(self.added_at) < FIRST_CONTACT;
        if self.is_unreachable() && !new {
            self.probe_backoff.next()
        } else {
            self.hello_backoff.next()
        }
    }

    /// No more HELLO retries (the peer isn't our neighbour any more). A HELLO in flight
    /// finishes; a later handshake starts over. An unreachable member keeps its probes:
    /// it is no neighbour because it is unreachable, and only its answer can change that.
    /// A member is one whose HELLO_REPLY said so, or one `listed` in the member list: after
    /// a restart, none has answered yet, and members that restart together may each fail
    /// to reach the others at first; without probes, none would try again.
    pub(crate) fn stop_hello_retries(&mut self, listed: bool) {
        let probed =
            self.is_unreachable() && (listed || matches!(self.standing, Standing::Member { .. }));
        if !probed && 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.is_dormant() || self.quiet_for(now) < timing.dormant_after {
            return false;
        }
        self.incidents.dormancies += 1;
        self.go_dormant(now);
        true
    }

    /// The peer said farewell (it is shutting down): Dormant now, without waiting for
    /// its silence, and not counted as an incident. Returns true on the transition.
    pub(crate) fn say_farewell(&mut self, now: Instant) -> bool {
        self.left = true;
        if self.liveness.is_dormant() {
            return false;
        }
        self.go_dormant(now);
        true
    }

    /// A message from `run` of the peer (a HELLO's or HELLO_REPLY's): a run other than
    /// the one that said farewell is back (`left`).
    pub(crate) fn returned(&mut self, run: crate::transport::RunId) {
        if self.run_id != Some(run) {
            self.left = false;
        }
    }

    fn go_dormant(&mut self, now: Instant) {
        self.liveness = Liveness::Dormant { since: now };
        self.unreachable_since.get_or_insert(now);
        self.reset_sync();

        if self.hello_state != HelloState::Sending {
            self.hello_state = HelloState::AwaitingReply {
                next_attempt: now + self.probe_backoff.next(),
            };
        }
    }

    /// The keepalive an established, quiet peer needs, if any (and records it): a PING,
    /// or a STATUS if none went between us for `status_refresh`.
    ///
    /// A PING and its PONG (or a STATUS and its STATUS_REPLY) tell both sides that the
    /// other is alive, so one per interval does for the pair. The two sides take turns:
    /// after a keepalive from the peer, the next is ours (the one leading the pair,
    /// `leads`, sends the first), so both measure the round trip and each sends every
    /// other interval. On our turn we send once the pair was quiet for an interval; on its
    /// turn only after half an interval more, in case it doesn't (a keepalive or its
    /// answer was lost). Without turns, both sides would get due at about the same time
    /// (each one's quiet time starts with the same exchange) and their keepalives would
    /// often cross.
    ///
    /// When they do cross anyway (both sent before seeing the other's: the peer's came
    /// within half an interval after ours), each side would take the next turn for its
    /// own and they would keep crossing; then the next turn is the leader's.
    pub(crate) fn take_keepalive_due(
        &mut self,
        timing: &Timing,
        now: Instant,
    ) -> Option<Keepalive> {
        if self.keepalive_due_at(timing).is_none_or(|at| now < at) {
            return None;
        }
        self.last_keepalive_at = Some(now);
        self.awaiting_answer
            .get_or_insert(AwaitingAnswer::sent(timing, now));
        let status = self
            .last_status_at
            .is_none_or(|t| now.saturating_duration_since(t) >= timing.status_refresh);
        Some(if status {
            Keepalive::Status
        } else {
            Keepalive::Ping
        })
    }

    /// When the next keepalive is due (`take_keepalive_due`), if any is: none to a
    /// dormant peer, nor before the handshake.
    pub(crate) fn keepalive_due_at(&self, timing: &Timing) -> Option<Instant> {
        if self.liveness.is_dormant() || self.hello_state != HelloState::Established {
            return None;
        }
        let our_turn = match (self.last_keepalive_at, self.keepalive_received_at) {
            (None, None) => self.leads,
            (None, Some(_)) => true,
            (Some(_), None) => false,
            (Some(sent), Some(received)) if received > sent => {
                let crossed = received.saturating_duration_since(sent) < timing.keepalive / 2;
                !crossed || self.leads
            }
            (Some(_), Some(_)) => false,
        };
        let quiet_enough = if our_turn {
            timing.keepalive
        } else {
            timing.keepalive + timing.keepalive / 2
        };
        let quiet_since = self.last_seen.unwrap_or(self.added_at);
        let at = (self.added_at + timing.keepalive).max(quiet_since + quiet_enough);
        Some(match self.last_keepalive_at {
            Some(sent) => at.max(sent + timing.keepalive),
            None => at,
        })
    }

    /// When the peer worker next has something to do for this peer
    /// (`swarm::peer::run_peer_worker`): a keepalive (to a `neighbour`), dormancy (of a
    /// neighbour, once `ready`), a HELLO retry, an overdue sync answer, asking again for
    /// an answer and its being overdue (`take_ask_again_due`, `answer_overdue`). None if nothing is due until something
    /// changes.
    pub(crate) fn next_due(
        &self,
        timing: &Timing,
        neighbour: bool,
        ready: bool,
    ) -> Option<Instant> {
        let dormancy = (ready && neighbour && !self.liveness.is_dormant())
            .then(|| self.last_seen.unwrap_or(self.added_at) + timing.dormant_after);
        let keepalive = if neighbour {
            self.keepalive_due_at(timing)
        } else {
            None
        };
        let hello = match self.hello_state {
            HelloState::AwaitingReply { next_attempt } => Some(next_attempt),
            _ => None,
        };
        let reply = (!self.liveness.is_dormant() && self.hello_state == HelloState::Established)
            .then(|| self.sync.reply.as_ref().map(|wait| wait.deadline))
            .flatten();
        let unreached = self
            .awaiting_answer
            .map(|awaiting| awaiting.ask_again_at.unwrap_or(awaiting.unreached_at))
            // Once past, nothing more is due for it.
            .filter(|at| !self.liveness.is_dormant() && *at > Instant::now());
        [dormancy, keepalive, hello, reply, unreached]
            .into_iter()
            .flatten()
            .min()
    }

    /// Self-healing check on a STATUS or STATUS_REPLY. 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.is_dormant()
            || 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.is_dormant()
            && self
                .last_seen
                .is_some_and(|t| now.saturating_duration_since(t) < timing.dormant_after)
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    /// Unknown until heard from; silent long enough, dormant like an active peer.
    #[test]
    fn unknown_until_heard_from() {
        let timing = Timing::from_keepalive(Duration::from_secs(60), Duration::from_secs(300));
        let mut heard = PeerState::new(true, &timing);
        assert_eq!(heard.liveness, Liveness::Unknown);
        assert!(
            !heard.mark_seen(heard.added_at),
            "no re-handshake: none yet"
        );
        assert_eq!(heard.liveness, Liveness::Active);
        let mut silent = PeerState::new(true, &timing);
        assert!(silent.check_dormancy(&timing, silent.added_at + timing.dormant_after));
        assert!(silent.liveness.is_dormant());
    }

    /// After its farewell, a peer stays dormant whatever its stopping run still sends;
    /// its next run makes it active again.
    #[test]
    fn dormant_after_farewell_until_back() {
        let timing = Timing::from_keepalive(Duration::from_secs(60), Duration::from_secs(300));
        let mut peer = PeerState::new(true, &timing);
        let now = peer.added_at;
        peer.run_id = Some([1; 16]);
        peer.mark_seen(now);
        assert!(peer.say_farewell(now));
        assert!(!peer.mark_seen(now), "a straggler from the same run");
        assert!(peer.liveness.is_dormant() && peer.is_unreachable());
        peer.returned([1; 16]);
        assert!(!peer.mark_seen(now), "the same run again");
        peer.returned([2; 16]);
        assert!(peer.mark_seen(now), "its next run");
        assert!(!peer.liveness.is_dormant() && !peer.is_unreachable());
    }

    /// The two sides of a pair take turns: each sends every other interval.
    #[test]
    fn keepalives_take_turns() {
        let timing = Timing::from_keepalive(Duration::from_secs(60), Duration::from_secs(300));
        let mut a = PeerState::new(true, &timing);
        let mut b = PeerState::new(false, &timing);
        a.hello_state = HelloState::Established;
        b.hello_state = HelloState::Established;
        b.added_at = a.added_at;
        let mut senders = Vec::new();
        let mut now = a.added_at;
        for _ in 0..40 {
            now += Duration::from_secs(5);
            // Whoever is due sends; the other one gets it at once, and both hear from
            // each other (the keepalive, its answer).
            if a.take_keepalive_due(&timing, now).is_some() {
                senders.push("a");
                b.keepalive_received_at = Some(now);
            } else if b.take_keepalive_due(&timing, now).is_some() {
                senders.push("b");
                a.keepalive_received_at = Some(now);
            } else {
                continue;
            }
            a.mark_seen(now);
            b.mark_seen(now);
        }
        // 200 s: one keepalive a minute, a first, then taking turns.
        assert_eq!(senders, ["a", "b", "a"]);
    }

    /// Keepalives that crossed (both sides sent before seeing the other's) leave the next
    /// turn to the leader, rather than both taking it and crossing again.
    #[test]
    fn crossed_keepalives_go_back_to_taking_turns() {
        let timing = Timing::from_keepalive(Duration::from_secs(60), Duration::from_secs(300));
        let mut a = PeerState::new(true, &timing);
        let mut b = PeerState::new(false, &timing);
        let start = a.added_at;
        b.added_at = start;
        for peer in [&mut a, &mut b] {
            peer.hello_state = HelloState::Established;
        }
        // Both sent at 100 s and got the other's a moment later.
        let sent = start + Duration::from_secs(100);
        for peer in [&mut a, &mut b] {
            peer.last_keepalive_at = Some(sent);
            peer.keepalive_received_at = Some(sent + Duration::from_millis(50));
            peer.mark_seen(sent + Duration::from_millis(50));
        }
        let next = sent + timing.keepalive + Duration::from_secs(1);
        assert!(a.take_keepalive_due(&timing, next).is_some()); // The leader's turn.
        assert!(b.take_keepalive_due(&timing, next).is_none());
    }

    /// A keepalive nothing answers: asked again halfway, after `unreached_after` not
    /// reached; any message from the peer, reached again. One lost answer doesn't do it.
    #[test]
    fn unanswered_keepalives_stop_counting_as_reached() {
        let timing = Timing::from_keepalive(Duration::from_secs(60), Duration::from_secs(300));
        let mut peer = PeerState::new(true, &timing);
        peer.hello_state = HelloState::Established;
        let sent = peer.added_at + timing.keepalive;
        assert!(peer.take_keepalive_due(&timing, sent).is_some());
        let half = sent + timing.unreached_after / 2;
        assert!(!peer.take_ask_again_due(half - Duration::from_millis(1)));
        assert!(peer.take_ask_again_due(half));
        assert!(!peer.take_ask_again_due(half), "asked again once");
        assert!(!peer.answer_overdue(half));
        // The second ask was answered: reached all along.
        peer.mark_seen(half + Duration::from_millis(50));
        assert!(!peer.answer_overdue(sent + timing.unreached_after));
        // The next keepalive, and its second ask, go unanswered.
        let sent = half + timing.keepalive * 2;
        assert!(peer.take_keepalive_due(&timing, sent).is_some());
        assert!(peer.take_ask_again_due(sent + timing.unreached_after / 2));
        assert!(peer.answer_overdue(sent + timing.unreached_after));
        peer.mark_seen(sent + timing.unreached_after * 2);
        assert!(!peer.answer_overdue(sent + timing.unreached_after * 2));
    }

    /// A sync message whose answer is overdue is asked after with a PING too: a peer with
    /// nothing to say answers none, however often it goes, and still counts as reached
    /// once it answers the PING. Also when the wait for the sync answer grew longer than
    /// `unreached_after`: the PING gets its own time.
    #[test]
    fn unanswered_sync_messages_are_asked_after() {
        let timing = Timing::from_keepalive(Duration::from_secs(60), Duration::from_secs(300));
        let mut peer = PeerState::new(true, &timing);
        peer.hello_state = HelloState::Established;
        let sent = peer.added_at;
        peer.sync.expect_reply(
            Vec::new(),
            sent,
            crate::swarm::throughput::SyncBytes::default(),
            100,
        );
        let overdue = sent + timing.sync_reply_timeout;
        assert!(peer.take_overdue_sync_reply(&timing, overdue).is_some());
        assert!(peer.take_ask_again_due(overdue));
        peer.mark_seen(overdue + Duration::from_millis(50));
        assert!(!peer.answer_overdue(sent + timing.unreached_after));
        // The next one waits twice as long for its answer, all of `unreached_after`.
        let sent = overdue + Duration::from_secs(1);
        peer.sync.expect_reply(
            Vec::new(),
            sent,
            crate::swarm::throughput::SyncBytes::default(),
            100,
        );
        let overdue = sent + timing.sync_reply_timeout * 2;
        assert!(peer.take_overdue_sync_reply(&timing, overdue).is_some());
        assert!(
            !peer.answer_overdue(overdue),
            "not before the PING had its time"
        );
        assert!(peer.take_ask_again_due(overdue));
        assert!(
            !peer.answer_overdue(overdue + timing.unreached_after / 2 - Duration::from_millis(1))
        );
        assert!(peer.answer_overdue(overdue + timing.unreached_after / 2));
    }

    /// Keepalives are PINGs, every `status_refresh` a STATUS; only when quiet.
    #[test]
    fn keepalives_are_pings_and_now_and_then_a_status() {
        let timing = Timing::from_keepalive(Duration::from_secs(60), Duration::from_secs(300));
        let mut peer = PeerState::new(true, &timing);
        peer.hello_state = HelloState::Established;
        let start = peer.added_at;
        assert_eq!(peer.take_keepalive_due(&timing, start), None); // Too new.
        let mut kinds = Vec::new();
        for interval in 1..=6 {
            let now = start + timing.keepalive * interval;
            let due = peer.take_keepalive_due(&timing, now);
            if due == Some(Keepalive::Status) {
                peer.last_status_at = Some(now);
            }
            kinds.extend(due);
            // Not again within the interval.
            assert_eq!(peer.take_keepalive_due(&timing, now), None);
        }
        use Keepalive::{Ping, Status};
        assert_eq!(kinds, [Status, Ping, Ping, Status, Ping, Ping]);
        // Heard from just now: nothing due.
        let now = start + timing.keepalive * 7;
        peer.mark_seen(now);
        assert_eq!(peer.take_keepalive_due(&timing, now), None);
    }

    /// R41: unreachability lasts until the peer is heard from, not for a set time, and
    /// probes back off meanwhile.
    #[test]
    fn unreachable_until_heard_from() {
        let timing = Timing::from_keepalive(Duration::from_secs(60), Duration::from_secs(300));
        let mut peer = PeerState::new(true, &timing);
        let start = Instant::now();
        for _ in 0..UNREACHABLE_AFTER_HELLOS {
            peer.hello_sent(start);
        }
        assert!(peer.is_unreachable());

        // Known for a while: probed.
        let later = start + FIRST_CONTACT;
        let probes: Vec<Duration> = (0..4).map(|_| peer.next_hello_wait(later)).collect();
        let probe = timing.dormant_probe;
        assert_eq!(probes, [probe, probe * 2, probe * 4, probe * 6]);
        peer.hello_sent(start + probe * 20);
        assert!(peer.is_unreachable());

        peer.mark_seen(start + probe * 21);
        assert!(!peer.is_unreachable());
        peer.hello_sent(start + probe * 22);
        peer.hello_sent(start + probe * 22);
        peer.hello_sent(start + probe * 22);
        assert_eq!(peer.next_hello_wait(start + probe * 22), probe);
    }

    /// A peer never heard from is HELLOed at the HELLO backoff while it is new
    /// (`FIRST_CONTACT`), unreachable or not; probed after.
    #[test]
    fn new_peers_retried_before_probed() {
        let timing = Timing::from_keepalive(Duration::from_secs(60), Duration::from_secs(300));
        let mut peer = PeerState::new(true, &timing);
        let start = peer.added_at;
        for _ in 0..UNREACHABLE_AFTER_HELLOS {
            peer.hello_sent(start);
        }
        assert!(peer.is_unreachable());
        let waits: Vec<Duration> = (0..7).map(|_| peer.next_hello_wait(start)).collect();
        assert!(
            waits.iter().all(|wait| *wait <= HELLO_RETRY_MAX),
            "{waits:?}"
        );
        assert_eq!(
            peer.next_hello_wait(start + FIRST_CONTACT),
            timing.dormant_probe
        );
    }
}