path: root/src/swarm/relay.rs

//! Passing changes on (`swarm::topology`): along writers' trees (`swarm::tree`), and by
//! announcement and pull where they don't reach.
//!
//! A node pushes its own changes to its neighbours, but nothing it receives goes any
//! further by itself but along a writer's tree. So after applying changes from one
//! neighbour, a node *announces*
//! its new heads to its other neighbours with a STATUS (heads, about 100 bytes, not the
//! changes; no keepalive, and answered only by a neighbour that lacks some of them). A
//! neighbour that sees heads it lacks *pulls* (those it still lacks after a round trip
//! to the announcer: while writes go on, the newest are usually on their way to it along
//! a tree, `pull_after_grace`): it enters the
//! announced heads as the peer's heads in its sync state, Automerge asks for what's
//! missing (`need`), and the announcer answers with exactly those changes. Regular
//! keepalives work the same way, so any gap is noticed within a keepalive interval.
//!
//! Announcements are coalesced: the first applied change schedules one announcement
//! round `ANNOUNCE_DELAY` later, and every change applied until then rides along. A
//! round costs one STATUS per neighbour (plus the pulls and their replies), whatever the
//! write rate; the delay is the price per hop. A neighbour that announced the same heads
//! to us first is skipped, so between two neighbours with the same news usually only one
//! STATUS goes. Without coalescing, every applied sync
//! message would announce to every neighbour, and announcements would outnumber the
//! data they announce (about 15 to 1 at 250 nodes).
//!
//! Changes that say whose they are (the writer's own, and what passes on along its tree)
//! go at once to the receiver's children in the writer's tree (`swarm::tree`). Then the
//! tree carries them, and the round after `ANNOUNCE_DELAY` goes only to the neighbours
//! the tree leaves out (`Plan::uncovered`: no member they reach comes before them). The
//! others are told in a slow round (`SLOW_ANNOUNCE_DELAY`, longer while nobody lacks
//! anything when told: `SlowRounds`): in case their parent missed the change (it went
//! away, or the nodes' views of the tree differed), coalescing all that came meanwhile. Changes that don't say whose (pulled, a delta, from a node that
//! has no part in the writer's tree) are announced to all, as before.
//!
//! In a full mesh too: the member list makes every member the writer's neighbour, but
//! that says nothing about whether the writer reaches them. One behind NAT may reach a
//! single one of them, one on another transport (on Tor alone) none, and no node can
//! tell which from where it is. A neighbour that has the changes already does nothing
//! with an announcement but answer it.

use std::collections::{HashMap, HashSet};
use std::sync::atomic::Ordering;
use std::time::Duration;

use tokio::time::Instant;
use tracing::debug;

use crate::locks::MutexExt;
use crate::state::{AppState, Standing};
use crate::swarm::peer::announce_heads;
use crate::swarm::topology;
use crate::transport::{self, HeadsSummary};

/// A pull that brought nothing for this long no longer blocks pulling from another peer
/// (sooner once its answer is overdue: `pull_unanswered`).
const PULL_TIMEOUT: Duration = Duration::from_secs(15);

/// The longest a STATUS's heads we lack wait before we pull them (`pull_after_grace`: a
/// round trip to its sender, at most this): over a slow transport (Tor), a change that
/// was missed waits no longer.
const PULL_GRACE_MAX: Duration = Duration::from_secs(1);

/// How long an announcement waits for further changes to ride along, and a random part
/// on top (`ANNOUNCE_JITTER`), so that of two neighbours with the same news one usually
/// announces first and the other, seeing the first's heads are its own, skips it
/// (`unreached`). Added once per hop to propagation; well below a round trip over the
/// transport.
const ANNOUNCE_DELAY: Duration = Duration::from_millis(200);
const ANNOUNCE_JITTER: Duration = Duration::from_millis(200);

/// How long the slow round after changes passed on along writers' trees waits: what a
/// tree misses arrives this much later, and while changes keep coming, every neighbour
/// is told at most this often. Doubles with every slow round that nobody lacked
/// anything of (`SlowRounds`), up to `SLOW_ANNOUNCE_MAX`.
const SLOW_ANNOUNCE_DELAY: Duration = Duration::from_secs(2);
const SLOW_ANNOUNCE_MAX: Duration = Duration::from_secs(16);

/// A STATUS_REPLY this soon after our slow round's announcement to the peer answers it.
const SLOW_ANSWER_WITHIN: Duration = Duration::from_secs(5);

/// How long the slow round waits. Since a neighbour answers an announcement only if it
/// lacks something (`swarm::peer::inbound`), an answer to the slow round means a tree
/// missed: then it is back to `SLOW_ANNOUNCE_DELAY`. While nobody answers, the trees
/// work, and the round waits twice as long each time, up to `SLOW_ANNOUNCE_MAX`: it is
/// a safety net, and most of its announcements go to neighbours that have everything.
/// (A lost message along a tree is noticed sooner by the next one, which its receiver
/// can't apply and asks after; the slow round is for the last of a run of writes, a
/// parent gone away, views of the tree that differ.)
pub(crate) struct SlowRounds {
    delay: Duration,
    /// An answer came since the last slow round was scheduled.
    answered: bool,
}

impl Default for SlowRounds {
    fn default() -> Self {
        // The first round is the short one.
        Self {
            delay: SLOW_ANNOUNCE_DELAY,
            answered: true,
        }
    }
}

impl SlowRounds {
    /// The wait for the slow round being scheduled now: as long as the last if it was
    /// answered (or the first), else twice that (up to the most).
    fn next(&mut self) -> Duration {
        self.delay = if std::mem::take(&mut self.answered) {
            SLOW_ANNOUNCE_DELAY
        } else {
            (self.delay * 2).min(SLOW_ANNOUNCE_MAX)
        };
        self.delay
    }
}

/// A STATUS_REPLY from `peer`, which we told of our heads in a slow round
/// `announced_ago`: it lacked something, so a tree missed; the next slow round waits
/// `SLOW_ANNOUNCE_DELAY` again.
pub(crate) fn slow_round_answered(state: &AppState, announced_ago: Duration) {
    if announced_ago < SLOW_ANSWER_WITHIN {
        let mut rounds = state.slow_rounds.lock_or_recover();
        if !rounds.answered {
            debug!("A slow round's announcement was answered; next one in 2 s");
        }
        rounds.answered = true;
    }
}

/// An announcement round scheduled: for changes from these peers, and the neighbours
/// that were sent them along a writer's tree meanwhile (`swarm::tree`).
#[derive(Default)]
pub(crate) struct Pending {
    senders: HashSet<String>,
    sent: HashSet<String>,
    /// Some change had no tree to go along: every neighbour is told, not only `targets`.
    all: bool,
    /// The neighbours writers' trees leave out (`Plan::uncovered`).
    targets: HashSet<String>,
}

/// After changes from `from` were applied and our heads changed: passed on at once along
/// the trees of their `writers` (by their authors' pair keys, with their changes);
/// then announced, in the
/// next round to those the trees leave out, and to the others in the slow round. To all
/// in the next round if any change has no tree to go along: none known whose
/// (`unauthored`), a writer not in the member list, or we have no part in its tree.
pub(crate) async fn announce(
    state: &AppState,
    from: &str,
    writers: &[(u64, Vec<crate::document::ChangeHash>)],
    unauthored: bool,
) {
    if !state.initial_sync_done.load(Ordering::Acquire) {
        return;
    }
    let mut plans = Vec::new();
    let mut treeless = unauthored || writers.is_empty();
    for (writer, changes) in writers {
        match state
            .dissemination
            .forward(state, *writer, from, changes)
            .await
        {
            Some(plan) => plans.push(plan),
            None => treeless = true,
        }
    }
    let tree = !plans.is_empty();
    let schedule = {
        let mut pending = state.announce_pending.lock_or_recover();
        let schedule = pending.is_none();
        let pending = pending.get_or_insert_with(Pending::default);
        pending.senders.insert(from.to_owned());
        for plan in &plans {
            pending.sent.extend(plan.children.iter().cloned());
            pending.targets.extend(plan.uncovered.iter().cloned());
        }
        pending.all |= treeless;
        schedule
    };
    if schedule {
        round(state, &state.announce_pending, ANNOUNCE_DELAY, false);
    }
    if !tree {
        return;
    }
    let schedule = {
        let mut slow = state.announce_slow.lock_or_recover();
        let schedule = slow.is_none();
        let slow = slow.get_or_insert_with(Pending::default);
        slow.senders.insert(from.to_owned());
        slow.all = true;
        // Our children were sent the changes, but those that reach no member but us
        // hear of a lost message from nobody else (`Plan::alone`).
        for plan in &plans {
            slow.sent.extend(
                plan.children
                    .iter()
                    .filter(|child| !plan.alone.contains(child))
                    .cloned(),
            );
        }
        schedule
    };
    if schedule {
        let delay = state.slow_rounds.lock_or_recover().next();
        round(state, &state.announce_slow, delay, true);
    }
}

/// An announcement round of what `pending` collects, `delay` (and a jitter) from now:
/// the `slow` one, or the quick one.
fn round(
    state: &AppState,
    pending: &std::sync::Arc<std::sync::Mutex<Option<Pending>>>,
    delay: Duration,
    slow: bool,
) {
    let state = state.clone();
    let pending = std::sync::Arc::clone(pending);
    tokio::spawn(async move {
        let jitter = crate::transport::paths::random_nonce()
            % u64::try_from(ANNOUNCE_JITTER.as_millis())
                .unwrap_or(1)
                .max(1);
        tokio::time::sleep(delay + Duration::from_millis(jitter)).await;
        let pending = pending.lock_or_recover().take().unwrap_or_default();
        announce_now(&state, &pending, slow).await;
    });
}

/// One announcement round. The peer whose changes we announce has them already, so a
/// round with a single sender skips it; with several, each lacks the others' changes,
/// so every neighbour is told.
async fn announce_now(state: &AppState, pending: &Pending, slow: bool) {
    let mut neighbours = unreached(state, &pending.senders, &pending.sent).await;
    if !pending.all {
        neighbours.retain(|peer| pending.targets.contains(peer));
    }
    if neighbours.is_empty() {
        return;
    }
    state.stats.announce_round(pending.all, slow);
    for peer in neighbours {
        if slow && let Some(peer_state) = state.get_peer_state(&peer).await {
            peer_state.lock().await.slow_announced_at = Some(Instant::now());
        }
        if let Err(error) = announce_heads(state, &peer, false).await {
            debug!(peer = %peer, error = %format!("{error:#}"), "Announcing our heads failed");
        }
    }
}

/// The neighbours to tell of the changes `senders` sent us. The peer whose changes we
/// announce has them already, so with a single sender it is skipped; with several, each
/// lacks the others' changes. So is a neighbour whose latest heads (its STATUS, an
/// announcement of its own, its STATUS_REPLY, or its latest sync message) are ours, or
/// have all of ours in their history: it has all we have; and one `sent` the changes
/// along a writer's tree, which has them already. Only what the neighbour said counts,
/// not what we sent it: a message along a tree may be lost, which is what the slow
/// round is for.
async fn unreached(
    state: &AppState,
    senders: &HashSet<String>,
    sent: &HashSet<String>,
) -> Vec<String> {
    let skip = match senders.len() {
        1 => senders.iter().next().cloned(),
        _ => None,
    };
    let (epoch, heads) = {
        let stamp = state.stamp.borrow();
        (stamp.epoch, stamp.heads.to_vec())
    };
    let ours = HeadsSummary::of(heads.clone());
    // What each neighbour said it has.
    let mut said = Vec::new();
    for (peer, peer_state) in state.peers_snapshot().await {
        if skip.as_deref() == Some(peer.as_str())
            || sent.contains(&peer)
            || !topology::is_neighbour(state, &peer)
        {
            continue;
        }
        let last = peer_state
            .lock()
            .await
            .last_heads
            .clone()
            .filter(|(theirs, _)| *theirs == epoch)
            .map(|(_, heads)| heads);
        said.push((peer, last));
    }
    let replica = state.doc.read().await;
    if replica.epoch != epoch {
        return said.into_iter().map(|(peer, _)| peer).collect();
    }
    said.into_iter()
        .filter(|(_, last)| {
            !last.as_ref().is_some_and(|theirs| {
                *theirs == ours
                    || crate::document::changes::covers(&replica.doc, theirs.listed(), &heads)
            })
        })
        .map(|(peer, _)| peer)
        .collect()
}

/// Pulls in progress: one neighbour at a time, so each change arrives about once. The
/// neighbours that announced heads we lack meanwhile wait their turn, with what they
/// announced, instead of until their next keepalive.
#[derive(Default)]
pub(crate) struct Pulling {
    /// The neighbour we are pulling from, and since when.
    current: Option<(String, Instant)>,
    /// The latest keepalive of each neighbour waiting to be pulled from.
    waiting: HashMap<String, transport::Status>,
}

/// Whether we lack any of `heads` (with a digest, any of the sample it lists).
pub(crate) async fn lacks(state: &AppState, heads: &HeadsSummary) -> bool {
    let replica = state.doc.read().await;
    !crate::document::changes::missing(&replica.doc, heads.listed()).is_empty()
}

/// A STATUS from `peer` (a keepalive's, or an announcement): if it has heads we lack,
/// pull them, but only those we still lack after a round trip to it (`PULL_GRACE_MAX` at
/// most; at once if we know none). While writes go on, the newest of them are usually
/// on their way to us along their writers' trees, as fast as the STATUS: they come
/// meanwhile, and pulling them would only fetch them twice. What is still missing then
/// was missed. One wait per peer: a STATUS that comes meanwhile is the one judged at the
/// end (`PeerState::pull_pending`).
pub(crate) async fn pull_after_grace(state: &AppState, peer: &str, remote: &transport::Status) {
    if HeadsSummary::of(state.stamp.borrow().heads.to_vec()) != remote.heads {
        let missing = {
            let replica = state.doc.read().await;
            crate::document::changes::missing(&replica.doc, remote.heads.listed())
        };
        if !missing.is_empty() {
            state.dissemination.announced(peer, &missing);
            let grace = state
                .lock_rtt()
                .effective(peer, Instant::now(), state.timing.rtt_fresh_for())
                .map(|rtt| rtt.min(PULL_GRACE_MAX));
            if let Some(grace) = grace {
                tokio::time::sleep(grace).await;
            }
        }
    }
    let latest = match state.get_peer_state(peer).await {
        Some(peer_state) => peer_state.lock().await.pull_pending.take(),
        None => None,
    };
    pull_if_behind(state, peer, latest.as_ref().unwrap_or(remote)).await;
}

/// If `peer` has heads we lack (as its STATUS `remote` says), pull them (one peer at a
/// time; a busy slot queues the peer). Returns whether a pull started.
pub(crate) async fn pull_if_behind(
    state: &AppState,
    peer: &str,
    remote: &transport::Status,
) -> bool {
    if remote.epoch != state.local_epoch() || !topology::is_neighbour(state, peer) {
        return false;
    }
    // The common case, the same heads, needs only the published heads.
    let ours = HeadsSummary::of(state.stamp.borrow().heads.to_vec());
    if ours == remote.heads {
        return false;
    }
    // Those of the listed heads we lack. With a digest only a sample is listed; if we
    // have all of it, a later keepalive lists others.
    let missing = {
        let replica = state.doc.read().await;
        crate::document::changes::missing(&replica.doc, remote.heads.listed())
    };
    if missing.is_empty() {
        return false;
    }
    // All heads, when we know them; otherwise those we lack.
    let theirs = match remote.heads.heads() {
        Some(heads) => heads.to_vec(),
        None => missing,
    };

    let Some(peer_state) = state.get_peer_state(peer).await else {
        return false;
    };
    if !matches!(peer_state.lock().await.standing, Standing::Member { .. }) {
        return false;
    }

    let now = Instant::now();
    {
        let mut pulling = state.pulling.lock_or_recover();
        let busy = matches!(
            &pulling.current,
            Some((other, since)) if other != peer && now < *since + PULL_TIMEOUT
        );
        if busy {
            if pulling
                .waiting
                .insert(peer.to_owned(), remote.clone())
                .is_none()
            {
                state.stats.pull_queued();
            }
            return false;
        }
        pulling.current = Some((peer.to_owned(), now));
        pulling.waiting.remove(peer);
    }

    // Automerge asks for whatever these heads need that we don't have.
    peer_state.lock().await.sync.pull_from(theirs);
    state.stats.pull_started();
    debug!(peer = %peer, "Neighbour has changes we lack; pulling");
    state.request_sync(peer);
    true
}

/// A sync message from `peer` was applied: a pull from it is answered, and the next
/// waiting neighbour's turn comes.
pub(crate) fn pulled(state: &AppState, peer: &str) {
    {
        let mut pulling = state.pulling.lock_or_recover();
        if pulling.current.as_ref().is_none_or(|(p, _)| p != peer) {
            return;
        }
        pulling.current = None;
        if pulling.waiting.is_empty() {
            return;
        }
    }
    let state = state.clone();
    tokio::spawn(async move { pull_next(&state).await });
}

/// Our sync message to `peer` went unanswered for the reply timeout: a pull from it no
/// longer blocks the others, and the next waiting neighbour's turn comes. Otherwise
/// they would wait for `PULL_TIMEOUT`, and then for their next keepalive, as nothing
/// else brings them back.
pub(crate) fn pull_unanswered(state: &AppState, peer: &str) {
    {
        let mut pulling = state.pulling.lock_or_recover();
        if pulling.current.as_ref().is_none_or(|(p, _)| p != peer) {
            return;
        }
        debug!(peer = %peer, "Pull unanswered; pulling from the next neighbour");
        pulling.current = None;
        if pulling.waiting.is_empty() {
            return;
        }
    }
    let state = state.clone();
    tokio::spawn(async move { pull_next(&state).await });
}

/// Pulls from the first waiting neighbour that still has heads we lack.
async fn pull_next(state: &AppState) {
    loop {
        let (peer, remote) = {
            let mut pulling = state.pulling.lock_or_recover();
            if pulling.current.is_some() {
                return;
            }
            let Some(peer) = pulling.waiting.keys().next().cloned() else {
                return;
            };
            let remote = pulling.waiting.remove(&peer).expect("just listed");
            (peer, remote)
        };
        if pull_if_behind(state, &peer, &remote).await {
            return;
        }
    }
}

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

    /// The slow round waits twice as long after each one nobody answered, up to the
    /// most, and the short time again after an answer.
    #[test]
    fn slow_rounds_back_off_while_unanswered() {
        let mut rounds = SlowRounds::default();
        let waits: Vec<u64> = (0..6).map(|_| rounds.next().as_secs()).collect();
        assert_eq!(waits, [2, 4, 8, 16, 16, 16]);
        rounds.answered = true;
        assert_eq!(rounds.next(), SLOW_ANNOUNCE_DELAY);
        assert_eq!(rounds.next(), SLOW_ANNOUNCE_DELAY * 2);
    }
}