path: root/src/remake/coordinator.rs

//! The coordinator: the node that received the proposal counts the votes (`vote`) and
//! writes the decision.
//!
//! It commits as soon as a majority of the electorate voted for the remake. It aborts
//! if, before that:
//! * an operator requested an abort,
//! * the local epoch changed,
//! * a competing remake of the same epoch committed,
//! * the voting period (`settleSecs`) passed without a majority, or
//! * a majority is out of reach (too many members voted for a competing remake).
//!
//! The decision is written in the same step that checks the records once more, so a
//! commit never overrides an abort request or a competitor's commit that arrived while
//! the snapshot was being built. Nor is a commit written once the remake has expired:
//! from then on, any node may abort it (`api`), and the two mustn't race.
//!
//! Committing builds the snapshot at the base heads and records the new epoch, so
//! every node can verify its own build against it.
//!
//! The task lives only in memory. A coordinator that restarts takes its undecided
//! remakes up again once ready (`resume`); past the voting period, that means aborting
//! them. If it is gone for good, nobody decides and the proposal expires; its votes
//! stay cast (`vote`).

use std::time::Duration;

use tracing::{info, warn};

use super::Host;
use super::content;
use super::records::{self, Decision, EpochRecord, Phase, Record};
use super::snapshot;
use super::vote;
use crate::document::store::StoreError;

const TICK: Duration = Duration::from_secs(1);

/// Coordinates remake `id` until it is decided; returns at once if this node already
/// does.
pub(crate) async fn run<H: Host>(host: H, id: String) {
    if !host.remake().start_coordinating(&id) {
        return;
    }
    info!(remake = %id, "Coordinating remake");

    loop {
        tokio::time::sleep(TICK).await;

        match step(&host, &id).await {
            Ok(true) => break,
            Ok(false) => {}
            Err(error) => {
                warn!(remake = %id, error = %format!("{error:#}"), "Coordinator step failed; retrying");
            }
        }
    }
    host.remake().end_coordinating(&id);
}

/// Takes up coordinating this node's undecided remakes of `epoch` among `all` that it
/// doesn't coordinate (it restarted since proposing them). Called by the remake worker
/// once the node is ready, so its records are complete.
pub(super) fn resume<H: Host>(host: &H, epoch: &EpochRecord, all: &[(String, Record)]) {
    let me = host.node_id();
    for (id, record) in all {
        if record.decision.is_none()
            && record.proposal.coordinator == me
            && &record.proposal.from_epoch == epoch
            && !host.remake().is_coordinating(id)
        {
            info!(remake = %id, "Resuming coordination of own remake");
            tokio::spawn(run(host.clone(), id.clone()));
        }
    }
}

/// One evaluation. Returns true once the coordinator is done.
async fn step<H: Host>(host: &H, id: &str) -> Result<bool, StoreError> {
    let wanted = id.to_owned();
    let (record, all, register) = host
        .read_doc(move |doc| {
            (
                records::load(doc, &wanted),
                records::load_all(doc),
                records::load_register(doc),
            )
        })
        .await;

    let Some(record) = record else {
        warn!(remake = %id, "Remake record vanished; stopping coordinator");
        return Ok(true);
    };
    if record.decision.is_some() {
        return Ok(true);
    }

    if EpochRecord::from(host.current_epoch()) != record.proposal.from_epoch {
        decide_abort(host, id, "local epoch changed during the vote".to_owned()).await?;
        return Ok(true);
    }
    if let Some(reason) = obstacle(id, &record, &all) {
        decide_abort(host, id, reason).await?;
        return Ok(true);
    }

    let tally = vote::tally(id, &record, &all, &register);
    // The deadline first: a majority counted after it is too late.
    if records::unix_millis() >= record.proposal.settle_until {
        decide_abort(
            host,
            id,
            format!("no majority in the voting period: {tally}"),
        )
        .await?;
        return Ok(true);
    }
    if tally.reached() {
        info!(remake = %id, votes = %tally, "Majority reached");
        commit(host, id, &record).await?;
        return Ok(true);
    }
    if tally.out_of_reach() {
        decide_abort(host, id, format!("a majority is out of reach: {tally}")).await?;
        return Ok(true);
    }
    Ok(false)
}

/// Why remake `id` must be aborted whatever its votes, if it must.
fn obstacle(id: &str, record: &Record, all: &[(String, Record)]) -> Option<String> {
    if let Some(request) = &record.abort_requested {
        return Some(format!("aborted by operator via {}", request.by));
    }
    let (other, _) = all.iter().find(|(other, candidate)| {
        other != id
            && candidate.proposal.from_epoch == record.proposal.from_epoch
            && candidate.committed_epoch().is_some()
    })?;
    Some(format!("competing remake {other} committed"))
}

async fn commit<H: Host>(host: &H, id: &str, record: &Record) -> Result<(), StoreError> {
    let content = match &record.proposal.content {
        Some(reference) => match content::ensure(host, reference).await {
            Ok(bytes) => Some(bytes),
            Err(reason) => return decide_abort(host, id, reason).await,
        },
        None => None,
    };

    // Clone under the read lock, build without any lock.
    let mut copy = host.read_doc(|doc| doc.clone()).await;
    // A base left to the commit is what this node has now. Every node that sees the
    // decision has it too: the decision change depends on it.
    let (base, fixed_now) = match &record.proposal.base_heads {
        None => (Some(copy.get_heads()), true),
        Some(heads) => (records::decode_heads(heads), false),
    };
    let Some(mut request) = base.and_then(|base| record.snapshot_request_at(id, base)) else {
        return decide_abort(
            host,
            id,
            "unreadable epoch or base heads in proposal".into(),
        )
        .await;
    };
    let base_heads = fixed_now.then(|| records::encode_heads(&request.base_heads));
    request.content = content;
    let mut snapshot = match snapshot::build_detached(copy, request).await {
        Ok((snapshot, _copy)) => snapshot,
        Err(error) => {
            return decide_abort(host, id, format!("snapshot build failed: {error}")).await;
        }
    };

    let to_epoch = snapshot.epoch;
    let snapshot_bytes = snapshot.doc.save().len() as u64;

    let decision = Decision::Committed {
        to_epoch: to_epoch.into(),
        base_heads,
        snapshot_bytes,
        decided_at: records::unix_millis(),
    };
    match decide(host, id, decision).await? {
        Some(Decision::Committed { .. }) => {
            // The local switch reuses it instead of building again.
            host.remake().store_prebuilt(id, snapshot);
            info!(remake = %id, to_epoch = %to_epoch, snapshot_bytes, "Remake committed");
        }
        Some(Decision::Aborted { reason, .. }) => {
            warn!(remake = %id, reason = %reason, "Remake aborted while building its snapshot");
        }
        None => {}
    }
    Ok(())
}

async fn decide_abort<H: Host>(host: &H, id: &str, reason: String) -> Result<(), StoreError> {
    let decision = Decision::Aborted {
        reason: reason.clone(),
        decided_at: records::unix_millis(),
    };
    if decide(host, id, decision).await?.is_some() {
        warn!(remake = %id, reason = %reason, "Remake aborted");
    }
    Ok(())
}

/// Writes `decision`, checked again under the write lock against the records as they
/// are now (building the snapshot takes a while): a commit becomes an abort if the
/// operator asked for one, a competing remake committed, the remake expired, or the
/// majority is gone.
/// Returns what was written; nothing if the remake is decided already or gone.
async fn decide<H: Host>(
    host: &H,
    id: &str,
    decision: Decision,
) -> Result<Option<Decision>, StoreError> {
    let id = id.to_owned();
    let grace = host.dormant_after().as_millis() as u64;
    host.commit_doc(move |doc| -> Result<Option<Decision>, StoreError> {
        let all = records::load_all(doc);
        let Some((_, record)) = all.iter().find(|(other, _)| *other == id) else {
            return Ok(None);
        };
        if record.decision.is_some() {
            return Ok(None);
        }
        let decision = match decision {
            Decision::Committed { decided_at, .. } => {
                let register = records::load_register(doc);
                let tally = vote::tally(&id, record, &all, &register);
                let expired = record.phase(records::unix_millis(), grace) == Phase::Expired;
                match obstacle(&id, record, &all) {
                    Some(reason) => Decision::Aborted { reason, decided_at },
                    // Others may abort it by now (`api`): too late to commit.
                    None if expired => Decision::Aborted {
                        reason: format!("expired while building the snapshot: {tally}"),
                        decided_at,
                    },
                    None if !tally.reached() => Decision::Aborted {
                        reason: format!("the majority was lost: {tally}"),
                        decided_at,
                    },
                    None => decision,
                }
            }
            aborted @ Decision::Aborted { .. } => aborted,
        };
        records::write_decision(doc, &id, &decision)?;
        Ok(Some(decision))
    })
    .await
}