compact remake fixes base at the commit

Instead of fixing base heads at when proposed, compact fixes them when
committing.  This shrinks the window where writes can be lost if their
author restarts to the switch delay; during replace and import remakes
writes can still be lost during the voting period.

5 files changed, 138 insertions, 40 deletions

diff --git a/src/remake/api.rs b/src/remake/api.rs
index 826c81e..3530423 100644
--- a/src/remake/api.rs
+++ b/src/remake/api.rs
@@ -17,7 +17,8 @@
 //!   can't commit it any more anyway (`coordinator`). `DELETE /admin/remakes` does the
 //!   same for the undecided remake of the local epoch, without having to know its ID.
 //!
-//! Request body for POST (all optional):
+//! Request body for POST (all optional; `baseHeads` not for `compact`, whose base is
+//! fixed when it commits):
 //! `{ "kind": "compact", "spec": null, "baseHeads": ["<hex>", …],
 //!    "replayWindow": {"epochs": 1, "seconds": 604800},
 //!    "archiveRetention": {"epochs": 2, "seconds": 2592000},
@@ -395,6 +396,15 @@ async fn start_kind<H: Host>(
         }
     };
 
+    if kind.base_at_commit() && request.base_heads.is_some() {
+        return error_response(
+            StatusCode::BAD_REQUEST,
+            format!(
+                "a {kind_name} remake fixes its base when it commits; baseHeads is not accepted"
+            ),
+            None,
+        );
+    }
     let base = match &request.base_heads {
         None => None,
         Some(heads) => {
@@ -454,7 +464,8 @@ async fn start_kind<H: Host>(
     let mut proposal = Proposal {
         coordinator: host.node_id(),
         from_epoch: host.current_epoch().into(),
-        base_heads: Vec::new(),
+        // Filled in when the proposal is written, unless the kind fixes it at the commit.
+        base_heads: None,
         policy: Policy {
             replay_window: request
                 .replay_window
@@ -598,6 +609,7 @@ async fn propose<H: Host>(
     let electorate_host = host.clone();
     let me = host.node_id();
     let started = host.remake().started_at() + 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);
@@ -656,22 +668,22 @@ async fn propose<H: Host>(
             }
 
             // The base is the requested heads, or everything before the proposal (the
-            // proposal itself is never in it).
-            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()));
+            // 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
                     }
-                    base
-                }
-                None => doc.get_heads(),
-            };
-            let mut heads: Vec<String> = base.iter().map(|h| h.to_string()).collect();
-            heads.sort();
-            proposal.base_heads = heads;
+                    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);
diff --git a/src/remake/coordinator.rs b/src/remake/coordinator.rs
index 9bcf8ba..3a7b071 100644
--- a/src/remake/coordinator.rs
+++ b/src/remake/coordinator.rs
@@ -141,7 +141,23 @@ fn obstacle(id: &str, record: &Record, all: &[(String, Record)]) -> Option<Strin
 }
 
 async fn commit<H: Host>(host: &H, id: &str, record: &Record) -> Result<(), StoreError> {
-    let Some(mut request) = record.snapshot_request(id) else {
+    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,
@@ -149,15 +165,8 @@ async fn commit<H: Host>(host: &H, id: &str, record: &Record) -> Result<(), Stor
         )
         .await;
     };
-    if let Some(reference) = &record.proposal.content {
-        match content::ensure(host, reference).await {
-            Ok(bytes) => request.content = Some(bytes),
-            Err(reason) => return decide_abort(host, id, reason).await,
-        }
-    }
-
-    // Clone under the read lock, build without any lock.
-    let copy = host.read_doc(|doc| doc.clone()).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) => {
@@ -170,6 +179,7 @@ async fn commit<H: Host>(host: &H, id: &str, record: &Record) -> Result<(), Stor
 
     let decision = Decision::Committed {
         to_epoch: to_epoch.into(),
+        base_heads,
         snapshot_bytes,
         decided_at: records::unix_millis(),
     };
diff --git a/src/remake/kind.rs b/src/remake/kind.rs
index 8131f7f..b36db4c 100644
--- a/src/remake/kind.rs
+++ b/src/remake/kind.rs
@@ -71,6 +71,14 @@ pub(crate) trait RemakeKind: Send + Sync + 'static {
 
     /// Every node: where a user path written after the base goes.
     fn translate(&self, path: &[String], translation: &JsonValue) -> Translation;
+
+    /// Whether the coordinator fixes the base when it commits instead of when it
+    /// proposes (no `baseHeads` then). Right for a kind whose content is the old
+    /// document at the base: the later the base, the fewer writes need carrying over,
+    /// and a write made after it is lost if its author restarts before switching.
+    fn base_at_commit(&self) -> bool {
+        false
+    }
 }
 
 /// The kind with this name.
@@ -97,6 +105,10 @@ impl RemakeKind for Compact {
         "compact"
     }
 
+    fn base_at_commit(&self) -> bool {
+        true
+    }
+
     fn propose(&self, spec: &JsonValue) -> Result<Prepared, String> {
         match spec {
             JsonValue::Null => Ok(Prepared::default()),
diff --git a/src/remake/records.rs b/src/remake/records.rs
index cd3e5be..0e40960 100644
--- a/src/remake/records.rs
+++ b/src/remake/records.rs
@@ -130,7 +130,9 @@ pub(crate) struct Proposal {
     pub(crate) coordinator: String,
     pub(crate) from_epoch: EpochRecord,
     /// Hex change hashes, sorted. The proposal record itself is not part of the base.
-    pub(crate) base_heads: Vec<String>,
+    /// None for a kind whose base is fixed when it commits (`Decision::Committed`).
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub(crate) base_heads: Option<Vec<String>>,
     pub(crate) policy: Policy,
     pub(crate) settle_ms: u64,
     pub(crate) proposed_at: u64,
@@ -181,13 +183,19 @@ fn default_kind() -> String {
     DEFAULT_KIND.to_owned()
 }
 
-impl Proposal {
-    pub(crate) fn base_change_hashes(&self) -> Option<Vec<ChangeHash>> {
-        self.base_heads
-            .iter()
-            .map(|hex| decode_hex32(hex).map(ChangeHash))
-            .collect()
-    }
+/// Hex change hashes as hashes; None if one is unreadable.
+pub(crate) fn decode_heads(heads: &[String]) -> Option<Vec<ChangeHash>> {
+    heads
+        .iter()
+        .map(|hex| decode_hex32(hex).map(ChangeHash))
+        .collect()
+}
+
+/// Hashes as sorted hex, as records keep them.
+pub(crate) fn encode_heads(heads: &[ChangeHash]) -> Vec<String> {
+    let mut hex: Vec<String> = heads.iter().map(ChangeHash::to_string).collect();
+    hex.sort();
+    hex
 }
 
 #[derive(Clone, Debug, Serialize, Deserialize)]
@@ -196,6 +204,9 @@ pub(crate) enum Decision {
     #[serde(rename_all = "camelCase")]
     Committed {
         to_epoch: EpochRecord,
+        /// The base, when the proposal leaves it to the commit (sorted hex).
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        base_heads: Option<Vec<String>>,
         snapshot_bytes: u64,
         decided_at: u64,
     },
@@ -275,13 +286,39 @@ impl Record {
         }
     }
 
+    /// The base: the proposal's, or for a kind that fixes it at the commit, the
+    /// decision's. None until committed then, or if unreadable.
+    pub(crate) fn base(&self) -> Option<Vec<ChangeHash>> {
+        let heads = match (&self.proposal.base_heads, &self.decision) {
+            (Some(heads), _) => heads,
+            (
+                None,
+                Some(Decision::Committed {
+                    base_heads: Some(heads),
+                    ..
+                }),
+            ) => heads,
+            _ => return None,
+        };
+        decode_heads(heads)
+    }
+
     /// Everything needed to (re)build this remake's snapshot but the content (which the
-    /// caller adds, `content::ensure`); None if the proposal is unreadable.
+    /// caller adds, `content::ensure`); None if the proposal or base is unreadable.
     pub(crate) fn snapshot_request(&self, id: &str) -> Option<SnapshotRequest> {
+        self.snapshot_request_at(id, self.base()?)
+    }
+
+    /// The same, at `base` (the coordinator fixing a base at the commit).
+    pub(crate) fn snapshot_request_at(
+        &self,
+        id: &str,
+        base: Vec<ChangeHash>,
+    ) -> Option<SnapshotRequest> {
         Some(SnapshotRequest {
             remake_id: id.to_owned(),
             previous: self.proposal.from_epoch.to_epoch()?,
-            base_heads: self.proposal.base_change_hashes()?,
+            base_heads: base,
             policy: serde_json::to_value(&self.proposal.policy).ok()?,
             proposed_at: self.proposal.proposed_at,
             kind: self.proposal.kind.clone(),
@@ -499,7 +536,7 @@ mod tests {
         Proposal {
             coordinator: "node-a".into(),
             from_epoch: Epoch::GENESIS.into(),
-            base_heads: vec!["ab".repeat(32)],
+            base_heads: Some(vec!["ab".repeat(32)]),
             policy: Policy {
                 replay_window: Policy::default_replay_window(),
                 archive_retention: Policy::default_archive_retention(),
@@ -535,6 +572,7 @@ mod tests {
                 generation: 1,
                 id: "cd".repeat(32),
             },
+            base_heads: None,
             snapshot_bytes: 123,
             decided_at: 11_000,
         };
@@ -552,7 +590,7 @@ mod tests {
         let (id, record) = &all[0];
         assert_eq!(id, "c-1");
         assert_eq!(record.proposal.policy, p.policy);
-        assert_eq!(record.proposal.base_change_hashes().unwrap().len(), 1);
+        assert_eq!(record.base().unwrap().len(), 1);
         assert!(record.abort_requested.is_some());
         assert_eq!(record.votes.keys().collect::<Vec<_>>(), ["node-b"]);
         assert_eq!(record.proposal.quorum, 2);
@@ -608,6 +646,31 @@ mod tests {
     }
 
     #[test]
+    fn a_base_fixed_at_the_commit() {
+        let mut record = Record {
+            proposal: Proposal {
+                base_heads: None,
+                ..proposal(10_000)
+            },
+            decision: None,
+            abort_requested: None,
+            contested: false,
+            votes: BTreeMap::new(),
+        };
+        assert_eq!(record.base(), None);
+        record.decision = Some(Decision::Committed {
+            to_epoch: EpochRecord {
+                generation: 1,
+                id: "cd".repeat(32),
+            },
+            base_heads: Some(vec!["ef".repeat(32)]),
+            snapshot_bytes: 1,
+            decided_at: 2,
+        });
+        assert_eq!(record.base(), Some(vec![ChangeHash([0xef; 32])]));
+    }
+
+    #[test]
     fn disagreeing_decisions_are_contested() {
         let mut base = AutoCommit::new();
         write_proposal(&mut base, "c-1", &proposal(10_000)).unwrap();
@@ -618,6 +681,7 @@ mod tests {
                 generation: 1,
                 id: "cd".repeat(32),
             },
+            base_heads: None,
             snapshot_bytes: 1,
             decided_at: 2,
         };
diff --git a/src/remake/vote.rs b/src/remake/vote.rs
index 16de87a..b4936dc 100644
--- a/src/remake/vote.rs
+++ b/src/remake/vote.rs
@@ -304,7 +304,7 @@ mod tests {
             proposal: Proposal {
                 coordinator: electorate[0].to_owned(),
                 from_epoch: Epoch::GENESIS.into(),
-                base_heads: Vec::new(),
+                base_heads: Some(Vec::new()),
                 policy: Policy {
                     replay_window: Policy::default_replay_window(),
                     archive_retention: Policy::default_archive_retention(),