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.

1 files changed, 76 insertions, 12 deletions

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,
         };