remake: move proposal to its own module

3 files changed, 271 insertions, 252 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();
diff --git a/src/remake/mod.rs b/src/remake/mod.rs
index 4a27ff4..5042ce9 100644
--- a/src/remake/mod.rs
+++ b/src/remake/mod.rs
@@ -5,7 +5,9 @@
 //!   where writes made after the base go): `compact`, `replace` and `import`.
 //! * `snapshot`: deterministic build of the next epoch's initial document.
 //! * `records`: the replicated remake records under `_continuum/remakes/<id>`.
-//! * `coordinator`: proposal, vote count and decision on the node that was asked.
+//! * `proposal`: writing a proposal (its checks, superseding expired remakes), and when
+//!   a node other than the coordinator may abort an expired one.
+//! * `coordinator`: vote count and decision on the node that was asked.
 //! * `vote`: who votes, for which remake, and when a remake has its majority.
 //! * `content`: the new document of a `replace`/`import` remake, fetched by every voter.
 //! * `switch`: switching this node to a committed epoch (worker).
@@ -26,6 +28,7 @@ mod coordinator;
 mod delta;
 mod kind;
 mod local;
+mod proposal;
 mod quarantine;
 mod records;
 pub(crate) mod snapshot;
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)
+}