summaryrefslogtreecommitdiffstats
path: root/src/remake/records.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/remake/records.rs')
-rw-r--r--src/remake/records.rs265
1 files changed, 246 insertions, 19 deletions
diff --git a/src/remake/records.rs b/src/remake/records.rs
index 600d4dc..cd3e5be 100644
--- a/src/remake/records.rs
+++ b/src/remake/records.rs
@@ -8,6 +8,16 @@
//! * `abortRequested`: written by any node on an operator's request. It is an input to
//! the coordinator's decision, not a decision itself.
//!
+//! Once a remake has expired, its coordinator no longer commits it, and any node may
+//! abort it on an operator's request (`api`). Should clocks disagree by more than the
+//! grace margin, a commit and such an abort could still be written concurrently. Such
+//! a record is *contested*: it reads as committed, and nodes don't switch to it on
+//! their own.
+//!
+//! Votes are kept apart, in the votes register `_continuum/votes/<pk>`: one entry per
+//! member for the whole epoch, written by that member only (`vote`). Reading a record
+//! fills in the votes that name it.
+//!
//! A proposal whose coordinator never decides (crash, partition) is reported as
//! `expired` once its settle period plus a grace margin has passed. Expiry is derived,
//! never written, so a late but valid commit can't collide with someone's abort.
@@ -15,13 +25,15 @@
//! Records edited into an unreadable shape through the store API are skipped with a
//! warning.
+use std::collections::BTreeMap;
+
use automerge::{AutoCommit, ChangeHash};
use serde::{Deserialize, Serialize};
use serde_json::Value as JsonValue;
use tracing::warn;
use super::kind::DEFAULT_KIND;
-use super::snapshot::{SnapshotRequest, META_ROOT, REMAKES};
+use super::snapshot::{META_ROOT, REMAKES, SnapshotRequest, VOTES};
use crate::document::replica::Epoch;
use crate::document::store::{self, StoreError};
@@ -127,19 +139,44 @@ pub(crate) struct Proposal {
/// interval if absent.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub(crate) switch_delay_ms: Option<u64>,
- /// Peers that must stay active during the settle period.
- pub(crate) peers_considered: Vec<String>,
+ /// Members whose votes count (sorted), fixed at proposal time (`vote`).
+ pub(crate) electorate: Vec<String>,
+ /// Votes needed to commit: a majority of the electorate.
+ pub(crate) quorum: usize,
/// The remake kind (`compact` for records written before kinds existed).
#[serde(default = "default_kind")]
pub(crate) kind: String,
- /// The kind's build input.
- #[serde(default, skip_serializing_if = "JsonValue::is_null")]
- pub(crate) content: JsonValue,
+ /// The kind's build input, if it has one: held by the coordinator and fetched by
+ /// every voter (`content`), not stored in the record.
+ #[serde(default, skip_serializing_if = "Option::is_none")]
+ pub(crate) content: Option<ContentRef>,
/// The kind's translation input.
#[serde(default, skip_serializing_if = "JsonValue::is_null")]
pub(crate) translation: JsonValue,
}
+/// Content of a remake, by reference.
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub(crate) struct ContentRef {
+ /// BLAKE3 of the bytes, hex.
+ pub(crate) blake3: String,
+ pub(crate) size: u64,
+}
+
+impl ContentRef {
+ pub(crate) fn of(bytes: &[u8]) -> Self {
+ Self {
+ blake3: blake3::hash(bytes).to_hex().to_string(),
+ size: bytes.len() as u64,
+ }
+ }
+
+ pub(crate) fn hash(&self) -> Option<[u8; 32]> {
+ decode_hex32(&self.blake3)
+ }
+}
+
fn default_kind() -> String {
DEFAULT_KIND.to_owned()
}
@@ -173,6 +210,18 @@ pub(crate) struct AbortRequest {
pub(crate) at: u64,
}
+/// A member's vote: its entry in the votes register.
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub(crate) struct Vote {
+ pub(crate) remake: String,
+ pub(crate) at: u64,
+}
+
+/// The votes register: every concurrent value of each member's entry, by member public
+/// key. Two values mean the member voted again without seeing its first vote (`vote`).
+pub(crate) type Register = BTreeMap<String, Vec<Vote>>;
+
#[derive(Clone, Debug, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct Record {
@@ -181,6 +230,18 @@ pub(crate) struct Record {
pub(crate) decision: Option<Decision>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub(crate) abort_requested: Option<AbortRequest>,
+ /// Decisions written concurrently disagree; `decision` is then the commit. Not
+ /// stored: found when the record is read.
+ #[serde(
+ default,
+ skip_deserializing,
+ skip_serializing_if = "std::ops::Not::not"
+ )]
+ pub(crate) contested: bool,
+ /// The register's votes naming this remake, by member public key. Not stored in
+ /// the record: filled in when it is read (`load`).
+ #[serde(default, skip_deserializing)]
+ pub(crate) votes: BTreeMap<String, Vote>,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
@@ -214,8 +275,8 @@ impl Record {
}
}
- /// Everything needed to (re)build this remake's snapshot; None if the proposal
- /// is unreadable.
+ /// Everything needed to (re)build this remake's snapshot but the content (which the
+ /// caller adds, `content::ensure`); None if the proposal is unreadable.
pub(crate) fn snapshot_request(&self, id: &str) -> Option<SnapshotRequest> {
Some(SnapshotRequest {
remake_id: id.to_owned(),
@@ -224,7 +285,7 @@ impl Record {
policy: serde_json::to_value(&self.proposal.policy).ok()?,
proposed_at: self.proposal.proposed_at,
kind: self.proposal.kind.clone(),
- content: self.proposal.content.clone(),
+ content: None,
translation: self.proposal.translation.clone(),
})
}
@@ -254,8 +315,37 @@ pub(crate) fn record_path(id: &str, part: &str) -> Vec<String> {
path
}
-/// The record with this ID, if it exists and is readable.
+/// The record with this ID, if it exists and is readable, with its votes.
pub(crate) fn load(doc: &AutoCommit, id: &str) -> Option<Record> {
+ let record = load_record(doc, id)?;
+ Some(with_votes(id, record, &load_register(doc)))
+}
+
+/// All readable records, with their votes.
+pub(crate) fn load_all(doc: &AutoCommit) -> Vec<(String, Record)> {
+ let register = load_register(doc);
+ store::keys_at(doc, &records_dir())
+ .unwrap_or_default()
+ .into_iter()
+ .filter_map(|id| {
+ let record = load_record(doc, &id)?;
+ Some((id.clone(), with_votes(&id, record, &register)))
+ })
+ .collect()
+}
+
+fn with_votes(id: &str, mut record: Record, register: &Register) -> Record {
+ record.votes = register
+ .iter()
+ .filter_map(|(voter, votes)| {
+ let vote = votes.iter().find(|vote| vote.remake == id)?;
+ Some((voter.clone(), vote.clone()))
+ })
+ .collect();
+ record
+}
+
+fn load_record(doc: &AutoCommit, id: &str) -> Option<Record> {
let value = match store::read_path(doc, &record_path(id, "")) {
Ok(Some(value)) => value,
Ok(None) => return None,
@@ -264,21 +354,73 @@ pub(crate) fn load(doc: &AutoCommit, id: &str) -> Option<Record> {
return None;
}
};
- match serde_json::from_value(value) {
- Ok(record) => Some(record),
+ let mut record: Record = match serde_json::from_value(value) {
+ Ok(record) => record,
Err(error) => {
warn!(remake = %id, error = %error, "Skipping unreadable remake record");
- None
+ return None;
}
+ };
+ if let Some(commit) = contested_decision(doc, id) {
+ record.decision = Some(commit);
+ record.contested = true;
}
+ Some(record)
}
-/// All readable records.
-pub(crate) fn load_all(doc: &AutoCommit) -> Vec<(String, Record)> {
- store::keys_at(doc, &records_dir())
+/// A commit among concurrent decisions of remake `id` that disagree, if they do.
+fn contested_decision(doc: &AutoCommit, id: &str) -> Option<Decision> {
+ let decisions: Vec<Decision> = store::read_candidates(doc, &record_path(id, "decision"))
+ .ok()?
+ .into_iter()
+ .filter_map(|value| serde_json::from_value(value).ok())
+ .collect();
+ let committed: Vec<&EpochRecord> = decisions
+ .iter()
+ .filter_map(|decision| match decision {
+ Decision::Committed { to_epoch, .. } => Some(to_epoch),
+ Decision::Aborted { .. } => None,
+ })
+ .collect();
+ let agree = committed.is_empty()
+ || (committed.len() == decisions.len() && committed.iter().all(|e| *e == committed[0]));
+ if agree {
+ return None;
+ }
+ decisions
+ .into_iter()
+ .find(|decision| matches!(decision, Decision::Committed { .. }))
+}
+
+fn register_path(voter: &str) -> Vec<String> {
+ vec![META_ROOT.to_owned(), VOTES.to_owned(), voter.to_owned()]
+}
+
+/// The votes register; unreadable entries are skipped with a warning.
+pub(crate) fn load_register(doc: &AutoCommit) -> Register {
+ store::keys_at(doc, &[META_ROOT.to_owned(), VOTES.to_owned()])
.unwrap_or_default()
.into_iter()
- .filter_map(|id| load(doc, &id).map(|record| (id, record)))
+ .filter_map(|voter| {
+ let values = match store::read_candidates(doc, &register_path(&voter)) {
+ Ok(values) => values,
+ Err(error) => {
+ warn!(voter = %voter, error = %error, "Failed to read vote");
+ return None;
+ }
+ };
+ let votes: Vec<Vote> = values
+ .into_iter()
+ .filter_map(|value| match serde_json::from_value(value) {
+ Ok(vote) => Some(vote),
+ Err(error) => {
+ warn!(voter = %voter, error = %error, "Skipping unreadable vote");
+ None
+ }
+ })
+ .collect();
+ (!votes.is_empty()).then_some((voter, votes))
+ })
.collect()
}
@@ -316,6 +458,12 @@ pub(crate) fn write_abort_request(
write_part(doc, id, "abortRequested", request)
}
+/// Writes `voter`'s entry in the votes register (replacing its earlier one).
+pub(crate) fn write_vote(doc: &mut AutoCommit, voter: &str, vote: &Vote) -> Result<(), StoreError> {
+ let json = serde_json::to_value(vote).map_err(anyhow::Error::from)?;
+ store::write_path(doc, &register_path(voter), &json)
+}
+
pub(crate) fn decode_hex32(hex: &str) -> Option<[u8; 32]> {
if hex.len() != 64 || !hex.is_ascii() {
return None;
@@ -360,9 +508,10 @@ mod tests {
proposed_at: settle_until.saturating_sub(1_000),
settle_until,
switch_delay_ms: None,
- peers_considered: vec!["node-b".into()],
+ electorate: vec!["node-a".into(), "node-b".into()],
+ quorum: 2,
kind: DEFAULT_KIND.to_owned(),
- content: JsonValue::Null,
+ content: None,
translation: JsonValue::Null,
}
}
@@ -390,6 +539,12 @@ mod tests {
decided_at: 11_000,
};
write_decision(&mut doc, "c-1", &decision).unwrap();
+ let vote = |remake: &str| Vote {
+ remake: remake.into(),
+ at: 7,
+ };
+ write_vote(&mut doc, "node-b", &vote("c-1")).unwrap();
+ write_vote(&mut doc, "node-c", &vote("other")).unwrap();
doc.commit();
let all = load_all(&doc);
@@ -399,15 +554,48 @@ mod tests {
assert_eq!(record.proposal.policy, p.policy);
assert_eq!(record.proposal.base_change_hashes().unwrap().len(), 1);
assert!(record.abort_requested.is_some());
+ assert_eq!(record.votes.keys().collect::<Vec<_>>(), ["node-b"]);
+ assert_eq!(record.proposal.quorum, 2);
assert_eq!(record.phase(0, 0), Phase::Committed);
}
#[test]
+ fn concurrent_votes_are_kept_apart() {
+ let mut a = AutoCommit::new();
+ let vote = |remake: &str, at| Vote {
+ remake: remake.into(),
+ at,
+ };
+ write_vote(&mut a, "node-a", &vote("x", 1)).unwrap();
+ a.commit();
+ let mut b = a.fork();
+ // node-b votes in both, never seeing its other vote (a lost vote).
+ write_vote(&mut a, "node-b", &vote("x", 2)).unwrap();
+ write_vote(&mut b, "node-b", &vote("y", 3)).unwrap();
+ a.commit();
+ b.commit();
+ a.merge(&mut b).unwrap();
+
+ let register = load_register(&a);
+ assert_eq!(register["node-a"], [vote("x", 1)]);
+ let mut twice = register["node-b"].clone();
+ twice.sort_by_key(|vote| vote.at);
+ assert_eq!(twice, [vote("x", 2), vote("y", 3)]);
+
+ // Seen, a vote is replaced.
+ write_vote(&mut a, "node-b", &vote("z", 4)).unwrap();
+ a.commit();
+ assert_eq!(load_register(&a)["node-b"], [vote("z", 4)]);
+ }
+
+ #[test]
fn phases() {
let mut record = Record {
proposal: proposal(10_000),
decision: None,
abort_requested: None,
+ contested: false,
+ votes: BTreeMap::new(),
};
assert_eq!(record.phase(10_000, 500), Phase::Proposed);
assert_eq!(record.phase(10_500, 500), Phase::Proposed);
@@ -420,6 +608,45 @@ mod tests {
}
#[test]
+ fn disagreeing_decisions_are_contested() {
+ let mut base = AutoCommit::new();
+ write_proposal(&mut base, "c-1", &proposal(10_000)).unwrap();
+ base.commit();
+ let (mut a, mut b, mut c) = (base.fork(), base.fork(), base.fork());
+ let committed = Decision::Committed {
+ to_epoch: EpochRecord {
+ generation: 1,
+ id: "cd".repeat(32),
+ },
+ snapshot_bytes: 1,
+ decided_at: 2,
+ };
+ write_decision(&mut a, "c-1", &committed).unwrap();
+ write_decision(
+ &mut b,
+ "c-1",
+ &Decision::Aborted {
+ reason: "test".into(),
+ decided_at: 3,
+ },
+ )
+ .unwrap();
+ // The same decision, concurrently: they agree.
+ write_decision(&mut c, "c-1", &committed).unwrap();
+ a.commit();
+ b.commit();
+ c.commit();
+ let mut agreeing = a.fork();
+ agreeing.merge(&mut c).unwrap();
+ assert!(!load(&agreeing, "c-1").unwrap().contested);
+
+ a.merge(&mut b).unwrap();
+ let record = load(&a, "c-1").unwrap();
+ assert!(record.contested);
+ assert_eq!(record.phase(0, 0), Phase::Committed);
+ }
+
+ #[test]
fn unreadable_records_are_skipped() {
let mut doc = AutoCommit::new();
store::write_path(
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-24 20:19:14 +0000