remake: move proposal to its own module

1 files changed, 262 insertions, 0 deletions

diff --git a/src/remake/proposal.rs b/src/remake/proposal.rs
new file mode 100644
index 0000000..ede25d2
--- /dev/null
+++ b/src/remake/proposal.rs
@@ -0,0 +1,262 @@
+//! 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 automerge::ChangeHash;
+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::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(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(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) = base
+                            .iter()
+                            .find(|head| doc.get_change_by_hash(head).is_none())
+                        {
+                            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)
+}