From f12a839d6e6bdd29a86ac52d5de836519a41aaf0 Mon Sep 17 00:00:00 2001 From: tv Date: Mon, 26 Oct 2026 05:38:41 +0100 Subject: wrap unix timestamps in newtype --- src/remake/records.rs | 54 +++++++++++++++++++++++++++------------------------ 1 file changed, 29 insertions(+), 25 deletions(-) (limited to 'src/remake/records.rs') diff --git a/src/remake/records.rs b/src/remake/records.rs index 9e2b14e..84286f3 100644 --- a/src/remake/records.rs +++ b/src/remake/records.rs @@ -35,6 +35,7 @@ use tracing::warn; use super::kind::DEFAULT_KIND; use super::snapshot::SnapshotRequest; +use crate::clock::UnixMillis; use crate::document::meta::{self, REMAKES, VOTES}; use crate::document::replica::Epoch; use crate::document::store::{self, StoreError}; @@ -130,8 +131,8 @@ pub(crate) struct Proposal { pub(crate) base_heads: Option>, pub(crate) policy: Policy, pub(crate) settle_ms: u64, - pub(crate) proposed_at: u64, - pub(crate) settle_until: u64, + pub(crate) proposed_at: UnixMillis, + pub(crate) settle_until: UnixMillis, /// How long nodes wait after seeing the commit before switching (ms); one keepalive /// interval if absent. #[serde(default, skip_serializing_if = "Option::is_none")] @@ -203,17 +204,20 @@ pub(crate) enum Decision { #[serde(default, skip_serializing_if = "Option::is_none")] base_heads: Option>, snapshot_bytes: u64, - decided_at: u64, + decided_at: UnixMillis, }, #[serde(rename_all = "camelCase")] - Aborted { reason: String, decided_at: u64 }, + Aborted { + reason: String, + decided_at: UnixMillis, + }, } #[derive(Clone, Debug, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub(crate) struct AbortRequest { pub(crate) by: String, - pub(crate) at: u64, + pub(crate) at: UnixMillis, } /// A member's vote: its entry in the votes register. @@ -221,7 +225,7 @@ pub(crate) struct AbortRequest { #[serde(rename_all = "camelCase")] pub(crate) struct Vote { pub(crate) remake: String, - pub(crate) at: u64, + pub(crate) at: UnixMillis, } /// The votes register: every concurrent value of each member's entry, by member public @@ -322,11 +326,11 @@ impl Record { }) } - pub(crate) fn phase(&self, now_ms: u64, grace_ms: u64) -> Phase { + pub(crate) fn phase(&self, now: UnixMillis, grace_ms: u64) -> Phase { match &self.decision { Some(Decision::Committed { .. }) => Phase::Committed, Some(Decision::Aborted { .. }) => Phase::Aborted, - None if now_ms > self.proposal.settle_until.saturating_add(grace_ms) => Phase::Expired, + None if now > self.proposal.settle_until.plus_ms(grace_ms) => Phase::Expired, None => Phase::Proposed, } } @@ -537,8 +541,8 @@ mod tests { archive_retention: Policy::default_archive_retention(), }, settle_ms: 1_000, - proposed_at: settle_until.saturating_sub(1_000), - settle_until, + proposed_at: UnixMillis(settle_until.saturating_sub(1_000)), + settle_until: UnixMillis(settle_until), switch_delay_ms: None, electorate: vec!["node-a".into(), "node-b".into()], quorum: 2, @@ -558,7 +562,7 @@ mod tests { "c-1", &AbortRequest { by: "x".into(), - at: 5, + at: UnixMillis(5), }, ) .unwrap(); @@ -569,12 +573,12 @@ mod tests { }, base_heads: None, snapshot_bytes: 123, - decided_at: 11_000, + decided_at: UnixMillis(11_000), }; write_decision(&mut doc, "c-1", &decision).unwrap(); let vote = |remake: &str| Vote { remake: remake.into(), - at: 7, + at: UnixMillis(7), }; write_vote(&mut doc, "node-b", &vote("c-1")).unwrap(); write_vote(&mut doc, "node-c", &vote("other")).unwrap(); @@ -589,15 +593,15 @@ mod tests { assert!(record.abort_requested.is_some()); assert_eq!(record.votes.keys().collect::>(), ["node-b"]); assert_eq!(record.proposal.quorum, 2); - assert_eq!(record.phase(0, 0), Phase::Committed); + assert_eq!(record.phase(UnixMillis(0), 0), Phase::Committed); } #[test] fn concurrent_votes_are_kept_apart() { let mut a = AutoCommit::new(); - let vote = |remake: &str, at| Vote { + let vote = |remake: &str, at: u64| Vote { remake: remake.into(), - at, + at: UnixMillis(at), }; write_vote(&mut a, "node-a", &vote("x", 1)).unwrap(); a.commit(); @@ -630,14 +634,14 @@ mod tests { contested: false, votes: BTreeMap::new(), }; - assert_eq!(record.phase(10_000, 500), Phase::Proposed); - assert_eq!(record.phase(10_500, 500), Phase::Proposed); - assert_eq!(record.phase(10_501, 500), Phase::Expired); + assert_eq!(record.phase(UnixMillis(10_000), 500), Phase::Proposed); + assert_eq!(record.phase(UnixMillis(10_500), 500), Phase::Proposed); + assert_eq!(record.phase(UnixMillis(10_501), 500), Phase::Expired); record.decision = Some(Decision::Aborted { reason: "test".into(), - decided_at: 1, + decided_at: UnixMillis(1), }); - assert_eq!(record.phase(0, 0), Phase::Aborted); + assert_eq!(record.phase(UnixMillis(0), 0), Phase::Aborted); } #[test] @@ -660,7 +664,7 @@ mod tests { }, base_heads: Some(vec!["ef".repeat(32)]), snapshot_bytes: 1, - decided_at: 2, + decided_at: UnixMillis(2), }); assert_eq!(record.base(), Some(vec![ChangeHash([0xef; 32])])); } @@ -678,7 +682,7 @@ mod tests { }, base_heads: None, snapshot_bytes: 1, - decided_at: 2, + decided_at: UnixMillis(2), }; write_decision(&mut a, "c-1", &committed).unwrap(); write_decision( @@ -686,7 +690,7 @@ mod tests { "c-1", &Decision::Aborted { reason: "test".into(), - decided_at: 3, + decided_at: UnixMillis(3), }, ) .unwrap(); @@ -702,7 +706,7 @@ mod tests { a.merge(&mut b).unwrap(); let record = load(&a, "c-1").unwrap(); assert!(record.contested); - assert_eq!(record.phase(0, 0), Phase::Committed); + assert_eq!(record.phase(UnixMillis(0), 0), Phase::Committed); } #[test] -- cgit v1.3.1