path: root/src/remake/proposal.rs

//! Proposing a remake: the checks (one open remake per epoch, the proposer free to
//! vote, a known base), superseding expired remakes, the electorate and quorum, and the
//! proposal and own vote written in one change; then its coordinator. And when a node
//! other than the coordinator may abort an expired remake.

use std::collections::HashSet;

use tracing::info;

use super::coordinator;
use super::records::{self, Decision, Phase, Proposal, Record, Vote};
use super::{Host, kind, vote};
use crate::clock::UnixMillis;
use crate::document::ChangeHash;
use crate::document::store::StoreError;
use crate::force::Force;

/// When a node other than the coordinator may abort an expired remake: a whole grace
/// period after it expired. Its coordinator stops committing at expiry
/// (`coordinator::decide`), so an abort from elsewhere and a commit can only cross if
/// their clocks are more than a grace period apart.
pub(super) fn abortable_at(record: &Record, grace: u64) -> UnixMillis {
    record
        .proposal
        .settle_until
        .plus_ms(record.grace(grace).saturating_mul(2))
}

/// Why this node may not abort expired `record` now, and the `force` name that would
/// override it (none for one that only time cures).
pub(super) fn abort_obstacle(
    record: &Record,
    me: &str,
    active_peers: &HashSet<String>,
    force: &Force,
    now: UnixMillis,
    grace: u64,
) -> Option<(String, Option<&'static str>)> {
    let at = abortable_at(record, grace);
    if now < at {
        return Some((
            format!(
                "only its coordinator may abort it until a grace period after it expired ({}s from now)",
                at.ms_since(now).div_ceil(1000)
            ),
            None,
        ));
    }
    let coordinator = &record.proposal.coordinator;
    if coordinator != me && active_peers.contains(coordinator) && !force.allows("coordinatorActive")
    {
        return Some((
            format!(
                "its coordinator {coordinator} is an active peer, so it is stuck rather than gone (look at its log); add force=coordinatorActive to abort it anyway"
            ),
            Some("coordinatorActive"),
        ));
    }
    None
}

/// The active peers this node sees.
pub(super) async fn active_peers<H: Host>(host: &H) -> HashSet<String> {
    host.peer_summaries()
        .await
        .into_iter()
        .filter(|peer| peer.active)
        .map(|peer| peer.id)
        .collect()
}

/// An expired remake of the epoch a proposal would supersede.
#[derive(Debug)]
pub(super) struct ExpiredRemake {
    pub(super) id: String,
    pub(super) coordinator: String,
    pub(super) coordinator_active: bool,
    /// When it expired.
    pub(super) since: UnixMillis,
    /// When another node may abort it (`abortable_at`).
    pub(super) abortable_at: UnixMillis,
    pub(super) tally: vote::Tally,
}

/// Superseding expired remakes, as the proposal's `force` asks.
pub(super) struct Supersede {
    pub(super) force: Force,
    pub(super) active_peers: HashSet<String>,
}

#[derive(Debug, thiserror::Error)]
pub(super) enum StartError {
    #[error("remake {0} is already open for this epoch")]
    Open(String),
    /// Expired remakes still hold votes, and aren't superseded: not asked to
    /// (`force=supersede`), or not possible yet (`abort_obstacle`).
    #[error("{message}")]
    Expired {
        message: String,
        remakes: Vec<ExpiredRemake>,
        /// The `force` name that would override it, if any.
        force: Option<&'static str>,
    },
    /// The proposer votes for its own remake, so it must be free to.
    #[error("this node can't vote for a new remake: {0}")]
    VoteHeld(String),
    #[error("base head {0} is not in this node's document")]
    UnknownBase(String),
    #[error(transparent)]
    Store(#[from] StoreError),
}

/// Writes the proposal (unless another one is open for the epoch) and starts its
/// coordinator. Runs as its own task, independent of the HTTP request.
pub(super) async fn propose<H: Host>(
    host: H,
    id: String,
    mut proposal: Proposal,
    base: Option<Vec<ChangeHash>>,
    supersede: Supersede,
    now: UnixMillis,
    grace: u64,
) -> Result<Record, StartError> {
    let record_id = id.clone();
    let electorate_host = host.clone();
    let me = host.node_id();
    let started = host.remake().started_at().plus_ms(grace);
    let base_at_commit = kind::by_name(&proposal.kind).is_some_and(|kind| kind.base_at_commit());
    let record = host
        .commit_doc(move |doc| -> Result<Record, StartError> {
            let mut all = records::load_all(doc);
            if let Some((open, _)) = all.iter().find(|(_, record)| {
                record.proposal.from_epoch == proposal.from_epoch
                    && record.phase(now, grace).is_open()
            }) {
                return Err(StartError::Open(open.clone()));
            }
            let mut register = records::load_register(doc);

            // Expired remakes hold their votes; their coordinators can't commit them
            // any more, so they may be aborted here.
            let expired: Vec<&String> = all
                .iter()
                .filter(|(_, record)| {
                    record.proposal.from_epoch == proposal.from_epoch
                        && record.phase(now, grace) == Phase::Expired
                })
                .map(|(id, _)| id)
                .collect();
            if !expired.is_empty() {
                let remakes = || {
                    all.iter()
                        .filter(|(id, _)| expired.contains(&id))
                        .map(|(id, record)| ExpiredRemake {
                            id: id.clone(),
                            coordinator: record.proposal.coordinator.clone(),
                            coordinator_active: supersede
                                .active_peers
                                .contains(&record.proposal.coordinator),
                            since: record.proposal.settle_until.plus_ms(record.grace(grace)),
                            abortable_at: abortable_at(record, grace),
                            tally: vote::tally(id, record, &all, &register),
                        })
                        .collect()
                };
                if !supersede.force.allows("supersede") {
                    return Err(StartError::Expired {
                        message: "an expired remake of this epoch still holds votes: abort it, \
                                  or add ?force=supersede to abort it and propose in its place"
                            .to_owned(),
                        remakes: remakes(),
                        force: Some("supersede"),
                    });
                }
                let obstacle =
                    all.iter()
                        .filter(|(id, _)| expired.contains(&id))
                        .find_map(|(id, record)| {
                            let (why, force) = abort_obstacle(
                                record,
                                &me,
                                &supersede.active_peers,
                                &supersede.force,
                                now,
                                grace,
                            )?;
                            Some((format!("can't supersede remake {id}: {why}"), force))
                        });
                if let Some((message, force)) = obstacle {
                    return Err(StartError::Expired {
                        message,
                        remakes: remakes(),
                        force,
                    });
                }
                let decision = Decision::Aborted {
                    reason: format!("superseded by remake {record_id}"),
                    decided_at: now,
                };
                for id in expired {
                    records::write_decision(doc, id, &decision)?;
                    info!(remake = %id, by = %record_id, "Superseded expired remake");
                }
                all = records::load_all(doc);
                register = records::load_register(doc);
            }

            match vote::freedom(&me, &proposal.from_epoch, &all, &register, started) {
                vote::Freedom::Any => {}
                vote::Freedom::Only(id) => {
                    return Err(StartError::VoteHeld(format!(
                        "it may have voted for remake {id} before its start"
                    )));
                }
                vote::Freedom::Held(reason) => return Err(StartError::VoteHeld(reason)),
            }

            // The base is the requested heads, or everything before the proposal (the
            // proposal itself is never in it); or, for a compaction, left to the commit.
            if !base_at_commit {
                let base = match base {
                    Some(base) => {
                        if let Some(unknown) = crate::document::changes::missing(doc, &base).first()
                        {
                            return Err(StartError::UnknownBase(unknown.to_string()));
                        }
                        base
                    }
                    None => doc.get_heads(),
                };
                proposal.base_heads = Some(records::encode_heads(&base));
            }

            // We vote for our own remake, so we are always in the electorate.
            let mut electorate = electorate_host.electorate(doc, now);
            if let Err(position) = electorate.binary_search(&me) {
                electorate.insert(position, me.clone());
            }
            proposal.quorum = vote::quorum(electorate.len());
            proposal.electorate = electorate;

            records::write_proposal(doc, &record_id, &proposal)?;
            let own_vote = Vote {
                remake: record_id.clone(),
                at: now,
            };
            records::write_vote(doc, &me, &own_vote)?;
            Ok(Record {
                proposal,
                decision: None,
                abort_requested: None,
                contested: false,
                votes: [(me, own_vote)].into(),
            })
        })
        .await?;

    tokio::spawn(coordinator::run(host, id));
    Ok(record)
}