path: root/src/state/mod.rs

//! Shared node state: the document, lifecycle, per-peer state and timing.
//!
//! * `peer`: what a node knows about each peer (`PeerState`): standing, liveness,
//!   handshake, incidents.
//! * `timing`: the intervals the protocol runs on (`Timing`).
//! * `read`: serving reads, live or from the pre-rejoin copy.
//! * `host`: the node as the remake module sees it (`remake::Host`).
//!
//! **Lock order.** Where several of these locks are held at once, they are taken in this
//! order: `doc` (the replica), `peer_states` (the map), a peer's `PeerState`, then
//! `lifecycle`. Sync generation and apply take doc → peer; an epoch switch doc → map →
//! each peer; establishing a peer peer → lifecycle. Waiting for a peer while holding the
//! map makes everything that needs the map wait for that peer, so apart from the switch
//! (which holds the map only for reading), code that walks the peers copies the map
//! first (`peers_snapshot`), and the one place that holds the map while looking at a
//! peer only tries its lock (`swarm::peer::forget_unless_alive`). The std mutexes
//! (statistics, round-trip times and the like) are held only briefly, never across an
//! await.

mod host;
mod peer;
mod read;
pub(crate) mod refused;
mod timing;

pub(crate) use peer::{HelloState, Incidents, Keepalive, Liveness, PeerState, Standing};
pub(crate) use read::{ReadSource, StaleView};
pub(crate) use timing::{SYNC_REPLY_TIMEOUT, Timing};

use std::collections::HashMap;
use std::sync::Arc;
use std::sync::atomic::AtomicBool;
use std::time::Duration;

use anyhow::{Result, bail};
use tokio::sync::{Mutex, RwLock, broadcast, mpsc, watch};
use tokio::time::Instant;
use tracing::{debug, error, warn};

use crate::document::lock::DocumentLock;
use crate::document::replica::{Epoch, Stamp};
use crate::document::signatures::{Sign, SignEvery};
use crate::document::store::{ChangeSet, StoreError};
use crate::document::{Document, Transaction};
use crate::locks::{MutexExt, RwLockExt};
use crate::remake::{Host, RemakeState};
use crate::swarm::sync::SyncEvent;
use crate::swarm::throughput::ThroughputTable;
use crate::transport;
use crate::transport::firewall::Firewall;

#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum NodeLifecycle {
    // TODO maybe rename discovered_swarm_id to swarm_id if that doesn't impair readability.
    Joining { discovered_swarm_id: Option<String> },
    Ready { swarm_id: String },
}

#[derive(Clone)]
pub(crate) struct AppState {
    /// The document and its epoch. Derefs to `Document`.
    pub(crate) doc: Arc<DocumentLock>,

    /// `doc`'s epoch label and heads, readable without its lock (`Replica::stamps`):
    /// for cheap reads (handshakes, keepalives, /peers). Only ever changed while holding
    /// `doc`'s write lock, together with `doc.epoch`.
    pub(crate) stamp: watch::Receiver<Stamp>,

    pub(crate) lifecycle: Arc<RwLock<NodeLifecycle>>,

    pub(crate) peer_states: Arc<RwLock<HashMap<String, Arc<Mutex<PeerState>>>>>,
    pub(crate) transport: Arc<dyn transport::Transport>,
    /// How members' neighbours are chosen (`swarm::topology::Topology`), and how a change
    /// reaches every member (`swarm::dissemination::Dissemination`): as the swarm
    /// records them (`swarm::topology::swarm_ways`), swapped when that changes;
    /// `--topology` and `--dissemination` until then (`configured_ways`).
    pub(crate) ways: Arc<std::sync::RwLock<crate::swarm::topology::Ways>>,
    /// `--topology` and `--dissemination`: what a genesis node records, and what counts
    /// where the swarm records nothing this node knows.
    pub(crate) configured_ways: (
        crate::swarm::topology::Kind,
        crate::swarm::dissemination::Kind,
    ),
    pub(crate) firewall: Arc<Firewall>,
    /// Messages and bytes exchanged, per peer (`transport::traffic`).
    pub(crate) traffic: Arc<transport::traffic::Traffic>,
    /// Node-wide figures for /admin/stats (`swarm::stats`).
    pub(crate) stats: Arc<crate::swarm::stats::NodeStats>,

    pub(crate) sync_events: mpsc::UnboundedSender<SyncEvent>,

    /// The document's version: bumped after every change (`notify_store_changed`). The
    /// change publisher (`swarm::sync`) waits on it; a measurement compares it.
    pub(crate) change_tx: watch::Sender<u64>,

    pub(crate) initial_sync_done: Arc<AtomicBool>,
    /// True once the node shuts down (a signal, a failed join, a transport gone): the
    /// HTTP API drains and its long-lived responses end, and keepalives say the node
    /// isn't ready and is leaving (`peer::farewell`).
    pub(crate) closing: Arc<watch::Sender<bool>>,

    pub(crate) change_events: broadcast::Sender<Arc<ChangeSet>>,

    pub(crate) timing: Timing,

    /// Per-node remake state (prebuilt snapshots, switch status, archive).
    pub(crate) remake: Arc<RemakeState>,

    /// Pre-shared key every member must prove in the handshake; None admits any node
    /// that reaches us.
    pub(crate) swarm_key: Option<Arc<crate::swarm::auth::SwarmKey>>,

    /// Random per run of this node; our HELLOs carry it and replies must echo it.
    pub(crate) run_id: transport::RunId,
    /// Handshakes this node started in this run, with any peer (`HelloRequest::handshake`).
    pub(crate) handshakes: Arc<std::sync::atomic::AtomicU32>,

    /// Target number of neighbours (`--neighbours`); up to this many + 1 members the
    /// swarm is a full mesh.
    pub(crate) neighbours_k: usize,
    /// Our alias (`--alias`), published in the member list (`swarm::members`).
    pub(crate) alias: Option<String>,
    /// Where people reach our dashboard (`--dashboard-url`), published in the member list
    /// (`swarm::members`).
    pub(crate) dashboard_url: Option<String>,
    /// Members alive within this window vote on remakes (`--electorate-window`).
    pub(crate) electorate_window: Duration,
    /// This node's clock is off when it differs from most of its peers' by more than
    /// this (`--clock-skew-warn`, `swarm::skew`); 0: never.
    pub(crate) clock_skew_warn: Duration,
    /// How often `GET /stream` says it's alive while nothing changes
    /// (`--stream-heartbeat`); 0: never.
    pub(crate) stream_heartbeat: Duration,
    /// What a replay does with writes a newer one superseded (`--superseded-writes`).
    pub(crate) superseded_writes: crate::remake::Superseded,
    /// How far this node's clock is ahead of its peers' (their median, `swarm::skew`), as
    /// last measured; 0 before. The propagation figures take it off.
    pub(crate) clock_ahead_ms: Arc<std::sync::atomic::AtomicI64>,
    /// What the clock checks remember between them (`swarm::skew`).
    pub(crate) clock_skew: Arc<std::sync::Mutex<crate::swarm::skew::Watch>>,
    /// Chunked transfers served and fetched (`swarm::blob`).
    pub(crate) blobs: Arc<crate::swarm::blob::Blobs>,
    /// HTTP API users who may write protocol data through the store API
    /// (`--protocol-writers`).
    pub(crate) protocol_writers: Arc<std::collections::HashSet<String>>,
    /// The swarm's first users, recorded by a genesis node (`NodeConfig::first_users`).
    pub(crate) first_users: Arc<crate::access::registry::FirstUsers>,
    /// Signs the tokens this node issues and the changes it commits
    /// (`NodeConfig::node_key`).
    pub(crate) node_key: Option<Arc<crate::identity::NodeKey>>,
    /// Which of its changes this node signs besides those a write's `Sign` says
    /// (`--sign-every`).
    pub(crate) sign_every: SignEvery,
    /// Received changes waiting for their dependencies, and those refused, while the
    /// policy has them judged (`access::screen`). Locked after the document.
    pub(crate) screening: Arc<std::sync::Mutex<crate::access::screen::Screening>>,
    /// The challenges logins by key used, until they expire (`access::challenge`).
    pub(crate) challenges: Arc<std::sync::Mutex<crate::access::challenge::Used>>,
    /// Sessions ended on this node in a swarm without a registry, where they can't be
    /// recorded in the document: by session ID, until when each would have lasted
    /// (`api::login::end_session`). This node refuses their tokens; others take them
    /// until they expire, and so does this one after a restart.
    pub(crate) ended_sessions: Arc<std::sync::Mutex<std::collections::BTreeMap<String, u64>>>,
    pub(crate) token_lifetime: Duration,
    pub(crate) session_lifetime: Duration,
    /// How long a challenge to log in by key is valid, and how many logins by key a user
    /// may make within it (`access::challenge`).
    pub(crate) challenge_lifetime: Duration,
    pub(crate) key_logins_per_user: usize,
    /// Writes every node would refuse, by user, and how many a user may have within a
    /// minute (`refused`).
    pub(crate) refused_writes: Arc<std::sync::Mutex<refused::RefusedWrites>>,
    pub(crate) refused_writes_per_minute: usize,
    /// Its epoch's root, as peers name it and as it knows its own (`swarm::roots`), and
    /// the one it was given (`--genesis-root`).
    pub(crate) roots: Arc<std::sync::Mutex<crate::swarm::roots::Roots>>,
    pub(crate) genesis_root: Option<crate::document::ChangeHash>,
    /// Whose signature on a strict remake's manifest this node takes (`access::strict`).
    pub(crate) strict_keys: Arc<Vec<crate::access::sshsig::PublicKey>>,
    /// Origins whose pages may call this node's open endpoints (`api::cors`).
    pub(crate) cors_origins: Arc<Vec<String>>,
    /// This node's own files, and how much it read and wrote them.
    pub(crate) files: Arc<crate::files::Files>,
    /// This node's document on disk (`--document-file`), if configured.
    pub(crate) document_file: Option<Arc<crate::document::file::DocumentFile>>,
    /// What this node remembers across restarts (`--state-file`), if configured.
    pub(crate) state_file: Option<Arc<crate::swarm::state_file::StateFile>>,
    /// The effective configuration with sources, imported values and secrets left out
    /// (`config::Loaded::report`), for `GET /admin/config`.
    pub(crate) config: Arc<crate::config::Reports>,
    /// Our own connectivity record and stability tier (`swarm::stability`).
    pub(crate) stability: Arc<std::sync::Mutex<crate::swarm::stability::Stability>>,
    /// Round-trip times to peers (`swarm::rtt`).
    pub(crate) rtt: Arc<std::sync::Mutex<crate::swarm::rtt::RttTable>>,
    /// Sync throughput with peers (`swarm::throughput`).
    pub(crate) throughput: Arc<std::sync::Mutex<ThroughputTable>>,
    /// Pulls of changes from neighbours (`swarm::relay`).
    pub(crate) pulling: Arc<std::sync::Mutex<crate::swarm::relay::Pulling>>,
    /// An announcement round is scheduled (`swarm::relay`).
    pub(crate) announce_pending: Arc<std::sync::Mutex<Option<crate::swarm::relay::Pending>>>,
    /// The slow round after changes passed on along writers' trees is scheduled
    /// (`swarm::relay`).
    pub(crate) announce_slow: Arc<std::sync::Mutex<Option<crate::swarm::relay::Pending>>>,
    /// How long the next slow round waits.
    pub(crate) slow_rounds: Arc<std::sync::Mutex<crate::swarm::relay::SlowRounds>>,
    /// Distances in writers' trees (`swarm::tree`).
    pub(crate) trees: Arc<std::sync::Mutex<crate::swarm::tree::Trees>>,
    /// Whom we reach, as told our neighbours (`swarm::reach`).
    pub(crate) reach: Arc<std::sync::Mutex<crate::swarm::reach::Reach>>,
    /// Links that differ from our record, and since when (`swarm::working_links`).
    pub(crate) working_links: Arc<std::sync::Mutex<crate::swarm::working_links::Pending>>,
    /// Our neighbours, once known (`swarm::topology`).
    pub(crate) neighbours: Arc<std::sync::RwLock<Option<crate::swarm::topology::NeighbourView>>>,
    /// Something the neighbours are computed from may have changed: the member list or a
    /// peer's state (`swarm::topology::run_worker`, `swarm::topology::change_peer`).
    pub(crate) topology_changed: Arc<tokio::sync::Notify>,
    /// Something a joining node waits for may have happened: the swarm ID learned, the
    /// initial sync done, a rejoin ended, a seed found (`swarm::bootstrap`).
    pub(crate) join_progress: Arc<tokio::sync::Notify>,
    /// The forget records may have changed (`swarm::changes`): the peer worker acts on
    /// them at once instead of at its next periodic check.
    pub(crate) forget_records_changed: Arc<AtomicBool>,
    /// When the peer worker is next due (`swarm::peer::Due`).
    pub(crate) peers_due: Arc<crate::swarm::peer::Due>,

    /// While rejoining a newer epoch, the previous document, served read-only (and
    /// marked stale) instead of answering 503.
    pub(crate) stale: Arc<std::sync::RwLock<Option<StaleView>>>,

    /// While the initial sync isn't done: the one peer we catch up from.
    pub(crate) catch_up_source: Arc<std::sync::Mutex<Option<CatchUpSource>>>,
    /// Since when a closer peer has had a head start to become the source.
    pub(crate) catch_up_preference: Arc<std::sync::Mutex<Option<Instant>>>,
    /// A catch-up source given up for not answering our sync messages, and until when it
    /// isn't chosen again (`swarm::source`).
    pub(crate) catch_up_avoided: Arc<std::sync::Mutex<Option<(String, Instant)>>>,
    /// The node's wall clock (`clock`).
    pub(crate) clock: Arc<crate::clock::Clock>,
}

/// The peer a catching-up node syncs with first (`swarm::source`).
#[derive(Clone, Debug)]
pub(crate) struct CatchUpSource {
    pub(crate) peer: String,
    /// When it was chosen or last sent us anything.
    pub(crate) last_heard: Instant,
    /// When it was chosen or last sent us a sync message: progress, not just liveness.
    pub(crate) last_sync: Instant,
    /// Our sync messages to it in a row whose answer didn't come in time.
    pub(crate) reply_timeouts: u32,
    /// When we last warned that catching up from it stalls.
    pub(crate) last_warned: Option<Instant>,
    /// Our last probe STATUS to it.
    pub(crate) last_probe: Option<Instant>,
    /// Its keepalives in a row saying it wouldn't sync with us.
    pub(crate) refusals: u32,
}

/// The change just made in `replica` (the one head after `before`), judged by the
/// swarm's rules as every node will judge it on receipt (`access::rules`), under a policy
/// judging every change: a write the API let through and the rules refuse would be
/// refused everywhere, with everything built on it.
fn judged_here(
    replica: &crate::document::replica::Replica,
    before: &[crate::document::ChangeHash],
) -> Result<(), String> {
    // By the policy it was written under, as receivers judge it: one removing or
    // changing the policy is judged too.
    let enforced = crate::document::at::At::new(&replica.doc, before)
        .is_some_and(|at| crate::access::policy::of(&at).enforced());
    if !enforced {
        return Ok(());
    }
    let after = replica.doc.get_heads();
    let [hash] = after[..] else {
        return Ok(());
    };
    if after == before {
        return Ok(());
    }
    match crate::access::rules::judge_held(&replica.doc, replica.epoch, &hash) {
        Some(judged) => judged,
        None => Ok(()),
    }
}

/// The document back to `before`, the change just made dropped: it never left this node
/// (the write lock is held since), so its actor goes on with the same sequence.
fn roll_back(
    replica: &mut crate::document::replica::Replica,
    before: &[crate::document::ChangeHash],
) -> Result<(), StoreError> {
    let actor = replica.doc.get_actor().clone();
    replica.doc = replica.doc.fork_at(before)?.with_actor(actor);
    Ok(())
}

impl AppState {
    /// Asks the sync worker to look at `peer` (`SyncEvent::PeerNeedsSync`): whether to
    /// send it a sync message now. Nothing if the worker is gone (the node stops).
    pub(crate) fn request_sync(&self, peer: &str) {
        let _ = self
            .sync_events
            .send(SyncEvent::PeerNeedsSync(peer.to_owned()));
    }

    /// `request_sync`, after `delay`.
    pub(crate) fn request_sync_after(&self, peer: &str, delay: std::time::Duration) {
        let events = self.sync_events.clone();
        let peer = peer.to_owned();
        tokio::spawn(async move {
            tokio::time::sleep(delay).await;
            let _ = events.send(SyncEvent::PeerNeedsSync(peer));
        });
    }

    /// Whether the node shuts down (`closing`).
    pub(crate) fn is_closing(&self) -> bool {
        *self.closing.borrow()
    }

    pub(crate) fn local_id(&self) -> &str {
        self.transport.local_id()
    }

    /// Peers that can serve our epoch's data: active, ready members on our epoch
    /// (`PeerSummary::serves_epoch`).
    pub(crate) async fn active_peers_on_local_epoch(&self) -> Vec<String> {
        let epoch = self.local_epoch();
        self.peer_summaries()
            .await
            .into_iter()
            .filter(|peer| peer.serves_epoch(epoch))
            .map(|peer| peer.id)
            .collect()
    }

    /// How members' neighbours are chosen now (`ways`).
    pub(crate) fn topology(&self) -> Arc<dyn crate::swarm::topology::Topology> {
        Arc::clone(&self.ways.read_or_recover().topology)
    }

    /// How a change reaches every member now (`ways`).
    pub(crate) fn dissemination(&self) -> Arc<dyn crate::swarm::dissemination::Dissemination> {
        Arc::clone(&self.ways.read_or_recover().dissemination)
    }

    pub(crate) fn lock_rtt(&self) -> std::sync::MutexGuard<'_, crate::swarm::rtt::RttTable> {
        self.rtt.lock_or_recover()
    }

    /// The local epoch. Code that labels or applies sync messages must use the epoch
    /// of the locked `Replica` instead, so label and document can't disagree.
    pub(crate) fn local_epoch(&self) -> Epoch {
        self.stamp.borrow().epoch
    }

    /// The local epoch and `Replica::epoch_seq`, read together, without waiting for the
    /// document's lock.
    pub(crate) fn epoch_stamp(&self) -> (Epoch, u32) {
        let stamp = self.stamp.borrow();
        (stamp.epoch, stamp.epoch_seq)
    }

    pub(crate) async fn is_ready(&self) -> bool {
        matches!(*self.lifecycle.read().await, NodeLifecycle::Ready { .. })
    }

    pub(crate) async fn swarm_id(&self) -> Option<String> {
        match &*self.lifecycle.read().await {
            NodeLifecycle::Ready { swarm_id } => Some(swarm_id.clone()),
            NodeLifecycle::Joining {
                discovered_swarm_id,
            } => discovered_swarm_id.clone(),
        }
    }

    // Record the expected swarm without making the node Ready. Bootstrap owns readiness.
    pub(crate) async fn set_expected_swarm_id(&self, swarm_id: String) -> Result<()> {
        let mut lifecycle = self.lifecycle.write().await;
        match &*lifecycle {
            NodeLifecycle::Joining {
                discovered_swarm_id: Some(existing),
            } if existing != &swarm_id => {
                bail!(
                    "Conflicting expected swarm IDs: expected {}, received {}",
                    existing,
                    swarm_id
                );
            }
            NodeLifecycle::Joining {
                discovered_swarm_id: None,
            } => {
                let old_lifecycle = lifecycle.clone();
                *lifecycle = NodeLifecycle::Joining {
                    discovered_swarm_id: Some(swarm_id),
                };
                debug!("Node transitioned from {old_lifecycle:?} {lifecycle:?}");
                self.join_progress.notify_one();
                Ok(())
            }
            NodeLifecycle::Joining {
                discovered_swarm_id: Some(_),
            } => {
                warn!("Node already Joining with the same identity");
                Ok(())
            }
            NodeLifecycle::Ready { swarm_id: existing } if existing == &swarm_id => {
                warn!("Node already Ready with the same identity");
                Ok(())
            }
            NodeLifecycle::Ready { swarm_id: existing } => {
                bail!(
                    "Cannot change established swarm ID: existing {}, received {}",
                    existing,
                    swarm_id
                );
            }
        }
    }

    // Only bootstrap should call this, after validating document metadata.
    pub(crate) async fn mark_ready(&self, swarm_id: String) -> Result<()> {
        let mut lifecycle = self.lifecycle.write().await;
        match &*lifecycle {
            NodeLifecycle::Joining {
                discovered_swarm_id: Some(expected),
            } if expected != &swarm_id => {
                bail!(
                    "Cannot mark node ready: expected swarm {}, received {}",
                    expected,
                    swarm_id
                );
            }
            NodeLifecycle::Joining { .. } => {
                let old_lifecycle = lifecycle.clone();
                *lifecycle = NodeLifecycle::Ready { swarm_id };
                debug!("Node transitioned from {old_lifecycle:?} {lifecycle:?}");
                // Dormancy is judged once ready, and the periodic checks run at once.
                self.peers_due.now();
            }
            NodeLifecycle::Ready { swarm_id: existing } if existing == &swarm_id => {
                debug!("Node already ready with the same identity");
            }
            NodeLifecycle::Ready { swarm_id: existing } => {
                bail!(
                    "Cannot change established swarm ID: existing {}, received {}",
                    existing,
                    swarm_id
                );
            }
        }
        Ok(())
    }

    /// Every known peer with its state, copied out of the peer map. Lock a peer's state
    /// only from such a copy, never while holding the map's lock: the map would then
    /// wait for the peer, and with it everything that only needs the map (a lookup, a
    /// new peer).
    pub(crate) async fn peers_snapshot(&self) -> Vec<(String, Arc<Mutex<PeerState>>)> {
        self.peer_states
            .read()
            .await
            .iter()
            .map(|(peer, state)| (peer.clone(), Arc::clone(state)))
            .collect()
    }

    /// Looks up a peer without creating it. Background work (sync, retries,
    /// keepalives) must use this so a forgotten peer isn't resurrected.
    pub(crate) async fn get_peer_state(&self, pk: &str) -> Option<Arc<Mutex<PeerState>>> {
        self.peer_states.read().await.get(pk).cloned()
    }

    pub(crate) async fn get_or_create_peer_state(&self, pk: &str) -> Arc<Mutex<PeerState>> {
        let mut peers = self.peer_states.write().await;

        Arc::clone(
            peers
                .entry(pk.to_owned())
                .or_insert_with(|| self.new_peer_state(pk)),
        )
    }

    /// State for a peer we haven't known so far (not inserted).
    pub(crate) fn new_peer_state(&self, pk: &str) -> Arc<Mutex<PeerState>> {
        let leads = crate::swarm::pair::leads(self.local_id(), pk);
        self.peers_due.now();
        Arc::new(Mutex::new(PeerState::new(leads, &self.timing)))
    }

    pub(crate) fn notify_store_changed(&self) {
        self.change_tx.send_modify(|version| {
            *version = version.wrapping_add(1);
        });
    }

    /// Reads the document on the blocking pool; `finish` runs after the lock is released.
    pub(crate) async fn render<T, R>(
        &self,
        read: impl FnOnce(&Document) -> T + Send + 'static,
        finish: impl FnOnce(T) -> R + Send + 'static,
    ) -> R
    where
        R: Send + 'static,
    {
        let doc = Arc::clone(&self.doc);
        tokio::task::spawn_blocking(move || {
            let data = {
                let guard = doc.blocking_read();
                read(&guard.doc)
            };
            finish(data)
        })
        .await
        .expect("document read task panicked")
    }

    /// Mutates the document on the blocking pool as one change, then
    /// publishes it. Write and notification run in the same blocking task, so a
    /// cancelled request can't leave a committed write unannounced. While a rejoin runs,
    /// the write goes into the document before it (`StaleView`), replayed when the
    /// rejoin ends; on a node that keeps none, it is refused (`StoreError::Rejoining`,
    /// `f` not called). Protocol data: the change is signed (`document::signatures`).
    pub(crate) async fn mutate_doc<T, E>(
        &self,
        f: impl FnOnce(&mut Transaction<'_>) -> Result<T, E> + Send + 'static,
    ) -> Result<T, E>
    where
        T: Send + 'static,
        E: From<StoreError> + Send + 'static,
    {
        self.mutate_doc_for(None, Sign::Protocol, f)
            .await
            .map(|(out, _)| out)
    }

    /// [`mutate_doc`](Self::mutate_doc) of this node's registration in the member list
    /// (`swarm::members::ensure_registered`): the one write not preceded by it.
    pub(crate) async fn mutate_doc_registering<T, E>(
        &self,
        f: impl FnOnce(&mut Transaction<'_>) -> Result<T, E> + Send + 'static,
    ) -> Result<T, E>
    where
        T: Send + 'static,
        E: From<StoreError> + Send + 'static,
    {
        self.mutate_doc_as(None, Sign::Protocol, false, f)
            .await
            .map(|(out, _)| out)
    }

    /// [`mutate_doc`](Self::mutate_doc) for `user`, who wrote through the API: the
    /// change names them (`replica::Writer`), and is signed as `sign` says, with this
    /// node's key. With the heads of the document written to after the
    /// write (sorted): what a read must have to include it.
    pub(crate) async fn mutate_doc_for<T, E>(
        &self,
        user: Option<String>,
        sign: Sign,
        f: impl FnOnce(&mut Transaction<'_>) -> Result<T, E> + Send + 'static,
    ) -> Result<(T, Vec<crate::document::ChangeHash>), E>
    where
        T: Send + 'static,
        E: From<StoreError> + Send + 'static,
    {
        self.mutate_doc_as(user, sign, true, f).await
    }

    /// [`mutate_doc_for`](Self::mutate_doc_for); with `registering_first`, after this
    /// node's registration if the policy needs it first.
    async fn mutate_doc_as<T, E>(
        &self,
        user: Option<String>,
        sign: Sign,
        registering_first: bool,
        f: impl FnOnce(&mut Transaction<'_>) -> Result<T, E> + Send + 'static,
    ) -> Result<(T, Vec<crate::document::ChangeHash>), E>
    where
        T: Send + 'static,
        E: From<StoreError> + Send + 'static,
    {
        let state = self.clone();

        tokio::task::spawn_blocking(move || {
            let result = {
                let mut replica = state.doc.blocking_write();
                // A write through the API under a policy this node doesn't know: what
                // the others would refuse it can't tell (`access::policy`).
                if user.is_some()
                    && let crate::access::policy::Policy::Unknown(why) =
                        crate::access::policy::of(&replica.doc)
                {
                    return Err(E::from(StoreError::UnknownPolicy(why)));
                }
                // Checked here, under the lock: the caller's readiness check (if any)
                // was before a rejoin could swap in a fresh document.
                if !replica.writable {
                    // Rejoining: into the document before the rejoin, which the rejoin
                    // replays when it ends (`StaleView`); refused by a node that keeps
                    // none (it wasn't ready).
                    let Some((_, stale)) = state.stale_document() else {
                        return Err(E::from(StoreError::Rejoining));
                    };
                    let mut doc = stale.write_or_recover();
                    let writer = crate::document::replica::Writer {
                        node: state.local_id(),
                        user: user.as_deref(),
                    };
                    let out = crate::document::replica::transact(
                        &mut doc,
                        state.clock.now_millis(),
                        &writer,
                        f,
                    )?;
                    let mut heads = doc.get_heads();
                    heads.sort_unstable();
                    return Ok((out, heads));
                }
                // Under a policy judging every change, a change by a node that isn't a
                // member is refused everywhere, and so is every change built on it: a
                // write before this node registered (`ensure_registered`), through the API
                // or its own, comes after its registration.
                if registering_first
                    && crate::access::policy::of(&replica.doc).enforced()
                    && !crate::swarm::members::registered(&replica.doc, state.local_id())
                {
                    let path = crate::swarm::members::member_path(state.local_id());
                    let entry = crate::swarm::members::entry(&state);
                    let writer = crate::document::replica::Writer {
                        node: state.local_id(),
                        user: None,
                    };
                    let before = replica.doc.get_heads();
                    replica
                        .transact(state.clock.now_millis(), &writer, |doc| {
                            crate::document::store::write_path(doc, &path, &entry)
                        })
                        .map_err(E::from)?;
                    state.sign_made(&mut replica, &before, &Sign::Protocol);
                    tracing::info!("Registered in the swarm's member list, before writing");
                }
                // One API write = one change; on error, its partial writes (e.g. the old
                // value deleted, the new one not written) are dropped.
                let writer = crate::document::replica::Writer {
                    node: state.local_id(),
                    user: user.as_deref(),
                };
                // A user whose writes every node refused, again and again: answered untried
                // for a while, as each refusal costs a fork of the document (`refused`).
                if let Some(user) = &user {
                    use crate::locks::MutexExt;
                    if !state.refused_writes.lock_or_recover().allows(
                        user,
                        state.refused_writes_per_minute,
                        std::time::Instant::now(),
                    ) {
                        return Err(E::from(StoreError::TooManyRefused));
                    }
                }
                let before = replica.doc.get_heads();
                let result = replica.transact(state.clock.now_millis(), &writer, f);
                let result = match result {
                    // Judged as every node will judge it, before it leaves; refused: as
                    // if never written.
                    Ok(out) => match judged_here(&replica, &before) {
                        Ok(()) => {
                            state.sign_made(&mut replica, &before, &sign);
                            Ok(out)
                        }
                        Err(why) => {
                            roll_back(&mut replica, &before)?;
                            if let Some(user) = &user {
                                use crate::locks::MutexExt;
                                state
                                    .refused_writes
                                    .lock_or_recover()
                                    .note(user, std::time::Instant::now());
                            }
                            tracing::debug!(reason = %why, "Refused a write every node would refuse");
                            Err(E::from(StoreError::Refused(why)))
                        }
                    },
                    Err(error) => Err(error),
                };
                replica.refresh_heads();
                result.map(|out| (out, replica.heads().to_vec()))
            }; // lock released before notifying

            if result.is_ok() {
                state.publish_change();
            }
            result
        })
        .await
        .expect("document write task panicked")
    }

    /// Signs the change just made in `replica`, if any (the one head after `before`, as
    /// it builds on every head before), as `sign` says, or as `--sign-every` has it; or
    /// keeps it as left unsigned, for the sync message that takes it (`sign_sent`).
    fn sign_made(
        &self,
        replica: &mut crate::document::replica::Replica,
        before: &[crate::document::ChangeHash],
        sign: &Sign,
    ) {
        let Some(key) = &self.node_key else {
            return;
        };
        let after = replica.doc.get_heads();
        if after != before
            && let [hash] = after[..]
        {
            if self.sign_every == SignEvery::Change || sign.wanted(&replica.doc) {
                replica.signatures.sign(&replica.doc, hash, key);
            } else {
                replica.signatures.left_unsigned(hash);
            }
        }
    }

    /// Before a sync message is made from `doc`: this node's latest change signed, if
    /// no signature vouches for it yet and it signs what it sends (`--sign-every sent`,
    /// or `none` under a policy requiring signatures), so it leaves with its signature.
    /// Looked at first under the read lock: mostly there is nothing to sign. Blocking.
    pub(crate) fn sign_sent(&self, doc: &crate::document::lock::DocumentLock) {
        let Some(key) = &self.node_key else {
            return;
        };
        if self.sign_every == SignEvery::Change {
            return;
        }
        let due = {
            let replica = doc.blocking_read();
            replica.signatures.pending()
                && (self.sign_every == SignEvery::Sent
                    || crate::access::policy::of(&replica.doc).required())
        };
        if due {
            let mut replica = doc.blocking_write();
            let replica = &mut *replica;
            replica.signatures.sign_latest(&replica.doc, key);
        }
    }

    pub(crate) fn publish_change(&self) {
        self.notify_store_changed();
        if let Err(error) = self.sync_events.send(SyncEvent::DocumentChanged) {
            // The write is committed; failing the request would misreport it.
            error!(error = %error, "Sync worker is gone; change will not be replicated");
        }
    }
}