diff options
| -rw-r--r-- | src/remake/api.rs | 46 | ||||
| -rw-r--r-- | src/remake/coordinator.rs | 30 | ||||
| -rw-r--r-- | src/remake/kind.rs | 12 | ||||
| -rw-r--r-- | src/remake/records.rs | 88 | ||||
| -rw-r--r-- | src/remake/vote.rs | 2 |
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(), |