remake: move proposal to its own module

1 files changed, 5 insertions, 251 deletions

diff --git a/src/remake/api.rs b/src/remake/api.rs
index 52338dd..f919be7 100644
--- a/src/remake/api.rs
+++ b/src/remake/api.rs
@@ -50,7 +50,6 @@
 //! their paths:
 //! `curl --data-binary @backup.automerge -H 'Content-Type: application/octet-stream' --url-query 'rules=[{"from":"","to":""}]' …/admin/remakes`
 
-use std::collections::HashSet;
 use std::time::Duration;
 
 use automerge::ChangeHash;
@@ -71,15 +70,14 @@ use tokio::sync::broadcast;
 use tracing::{error, info, warn};
 use uuid::Uuid;
 
-use super::coordinator;
 use super::local::LocalStatus;
+use super::proposal::{StartError, Supersede, abort_obstacle, abortable_at, active_peers, propose};
 use super::records::{
     self, AbortRequest, ContentRef, Decision, EpochRecord, Phase, Policy, Proposal, Record,
-    Retention, Vote,
+    Retention,
 };
-use super::{Host, kind, vote};
+use super::{Host, kind};
 use crate::api::error_response;
-use crate::clock::UnixMillis;
 use crate::document::store::StoreError;
 use crate::force::Force;
 
@@ -206,79 +204,6 @@ const FORCE: &[&str] = &["switchDelay", "supersede", "coordinatorActive"];
 /// What `?force=` may override on an abort.
 const ABORT_FORCE: &[&str] = &["coordinatorActive", "coordinatorInactive"];
 
-/// 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.
-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).
-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.
-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)]
-struct ExpiredRemake {
-    id: String,
-    coordinator: String,
-    coordinator_active: bool,
-    /// When it expired.
-    since: UnixMillis,
-    /// When another node may abort it (`abortable_at`).
-    abortable_at: UnixMillis,
-    tally: vote::Tally,
-}
-
-/// Superseding expired remakes, as the proposal's `force` asks.
-struct Supersede {
-    force: Force,
-    active_peers: HashSet<String>,
-}
-
 #[derive(Debug, Serialize)]
 #[serde(rename_all = "camelCase")]
 struct Report {
@@ -291,28 +216,6 @@ struct Report {
     local: Option<LocalStatus>,
 }
 
-#[derive(Debug, thiserror::Error)]
-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),
-}
-
 fn grace<H: Host>(host: &H) -> u64 {
     host.dormant_after().as_millis() as u64
 }
@@ -637,157 +540,6 @@ async fn start_kind<H: Host>(
         .into_response()
 }
 
-/// 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.
-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)
-}
-
 pub(crate) async fn list<H: Host>(State(host): State<H>) -> Response {
     let all = host.read_doc(records::load_all).await;
     let mut reports: Vec<Report> = all
@@ -1015,6 +767,8 @@ pub(crate) async fn epochs<H: Host>(State(host): State<H>) -> Response {
 #[cfg(test)]
 pub(crate) mod testing {
     use super::*;
+    use crate::clock::UnixMillis;
+    use crate::remake::vote;
 
     async fn into_json(response: Response) -> (StatusCode, JsonValue) {
         let status = response.status();