summaryrefslogtreecommitdiffstats
path: root/src/remake
diff options
context:
space:
mode:
authortv <tv@krebsco.de>2026-10-26 05:38:41 +0100
committertv <tv@krebsco.de>2026-10-26 05:38:41 +0100
commitf12a839d6e6bdd29a86ac52d5de836519a41aaf0 (patch)
treeb34da34a22292793c991b15a95908f6cd78bb70b /src/remake
parent504435444c65efc59c0f7c6881991296258538a7 (diff)
wrap unix timestamps in newtype
Diffstat (limited to 'src/remake')
-rw-r--r--src/remake/api.rs33
-rw-r--r--src/remake/catchup.rs29
-rw-r--r--src/remake/content.rs7
-rw-r--r--src/remake/local.rs27
-rw-r--r--src/remake/mod.rs6
-rw-r--r--src/remake/quarantine.rs3
-rw-r--r--src/remake/records.rs54
-rw-r--r--src/remake/snapshot.rs15
-rw-r--r--src/remake/vote.rs64
9 files changed, 148 insertions, 90 deletions
diff --git a/src/remake/api.rs b/src/remake/api.rs
index f5063e0..f019696 100644
--- a/src/remake/api.rs
+++ b/src/remake/api.rs
@@ -75,6 +75,7 @@ use super::records::{
};
use super::{Host, kind, vote};
use crate::api::error_response;
+use crate::clock::UnixMillis;
use crate::document::store::StoreError;
use crate::force::Force;
@@ -105,9 +106,9 @@ pub(crate) struct StartQuery {
rules: Option<String>,
/// Comma-separated hex.
base_heads: Option<String>,
- #[serde(default, with = "crate::units::optional_duration")]
+ #[serde(default, deserialize_with = "crate::units::optional_duration")]
settle: Option<Duration>,
- #[serde(default, with = "crate::units::optional_duration")]
+ #[serde(default, deserialize_with = "crate::units::optional_duration")]
switch_delay: Option<Duration>,
}
@@ -168,11 +169,11 @@ struct StartRequest {
replay_window: Option<RetentionRequest>,
archive_retention: Option<RetentionRequest>,
/// How long votes are collected; twice the dormancy threshold if absent.
- #[serde(default, with = "crate::units::optional_duration")]
+ #[serde(default, deserialize_with = "crate::units::optional_duration")]
settle: Option<Duration>,
/// How long nodes wait after seeing the commit before switching; one keepalive
/// interval if absent. Shorter only with `?force=switchDelay`.
- #[serde(default, with = "crate::units::optional_duration")]
+ #[serde(default, deserialize_with = "crate::units::optional_duration")]
switch_delay: Option<Duration>,
}
@@ -181,7 +182,7 @@ struct StartRequest {
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct RetentionRequest {
epochs: Option<u64>,
- #[serde(default, with = "crate::units::optional_duration")]
+ #[serde(default, deserialize_with = "crate::units::optional_duration")]
age: Option<Duration>,
}
@@ -205,11 +206,11 @@ const ABORT_FORCE: &[&str] = &["coordinatorActive"];
/// period after it expired. Its coordinator stops committing at expiry
/// (`coordinator::decide`), so an abort from elsewhere and a commit can only cross if
/// their clocks are more than a grace period apart.
-fn abortable_at(record: &Record, grace: u64) -> u64 {
+fn abortable_at(record: &Record, grace: u64) -> UnixMillis {
record
.proposal
.settle_until
- .saturating_add(grace.saturating_mul(2))
+ .plus_ms(grace.saturating_mul(2))
}
/// Why this node may not abort expired `record` now, and the `force` name that would
@@ -219,7 +220,7 @@ fn abort_obstacle(
me: &str,
active_peers: &HashSet<String>,
force: &Force,
- now: u64,
+ now: UnixMillis,
grace: u64,
) -> Option<(String, Option<&'static str>)> {
let at = abortable_at(record, grace);
@@ -227,7 +228,7 @@ fn abort_obstacle(
return Some((
format!(
"only its coordinator may abort it until a grace period after it expired ({}s from now)",
- (at - now).div_ceil(1000)
+ at.ms_since(now).div_ceil(1000)
),
None,
));
@@ -261,10 +262,10 @@ struct ExpiredRemake {
id: String,
coordinator: String,
coordinator_active: bool,
- /// When it expired (Unix milliseconds).
- since: u64,
+ /// When it expired.
+ since: UnixMillis,
/// When another node may abort it (`abortable_at`).
- abortable_at: u64,
+ abortable_at: UnixMillis,
tally: vote::Tally,
}
@@ -516,7 +517,7 @@ async fn start_kind<H: Host>(
},
settle_ms,
proposed_at: now,
- settle_until: now + settle_ms,
+ settle_until: now.plus(settle),
switch_delay_ms: switch_delay.map(|delay| delay.as_millis() as u64),
// Filled in when the proposal is written, from the document at that moment.
electorate: Vec::new(),
@@ -640,13 +641,13 @@ async fn propose<H: Host>(
mut proposal: Proposal,
base: Option<Vec<ChangeHash>>,
supersede: Supersede,
- now: u64,
+ now: UnixMillis,
grace: u64,
) -> Result<Record, StartError> {
let record_id = id.clone();
let electorate_host = host.clone();
let me = host.node_id();
- let started = host.remake().started_at() + grace;
+ let started = host.remake().started_at().plus_ms(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> {
@@ -679,7 +680,7 @@ async fn propose<H: Host>(
coordinator_active: supersede
.active_peers
.contains(&record.proposal.coordinator),
- since: record.proposal.settle_until.saturating_add(grace),
+ since: record.proposal.settle_until.plus_ms(grace),
abortable_at: abortable_at(record, grace),
tally: vote::tally(id, record, &all, &register),
})
diff --git a/src/remake/catchup.rs b/src/remake/catchup.rs
index 405b9f9..b00a210 100644
--- a/src/remake/catchup.rs
+++ b/src/remake/catchup.rs
@@ -41,6 +41,7 @@ use super::kind::Translator;
use super::local::{CatchUp, LocalStatus, Rejoin};
use super::quarantine;
use super::snapshot::{self, EpochMetadata};
+use crate::clock::UnixMillis;
use crate::document::replica::Epoch;
/// Pause before trying the same target again after a failed rejoin.
@@ -412,7 +413,7 @@ fn prepare(
previous: Epoch,
target: Epoch,
mut own: Vec<ActorId>,
- now: u64,
+ now: UnixMillis,
) -> Prepared {
if copy.get_change_by_hash(&ChangeHash(target.id)).is_none() {
return Prepared::NotYet(old);
@@ -486,7 +487,7 @@ fn prepare(
),
},
(Some(next), Ok(translator)) => {
- let away_ms = now.saturating_sub(next.proposed_at);
+ let away_ms = now.ms_since(next.proposed_at);
if next.policy.replay_window.covers(gap, away_ms) {
Plan::Replay(paths, translator)
} else {
@@ -570,7 +571,7 @@ mod tests {
previous: Epoch::GENESIS,
base_heads: &base,
policy,
- proposed_at: 0,
+ proposed_at: UnixMillis(0),
kind: "compact",
content: None,
translation: &JsonValue::Null,
@@ -591,7 +592,14 @@ mod tests {
#[test]
fn replays_within_the_window() {
let (old, swarm, target) = scenario(&json!({}));
- match prepare(swarm, old, Epoch::GENESIS, target, Vec::new(), 0) {
+ match prepare(
+ swarm,
+ old,
+ Epoch::GENESIS,
+ target,
+ Vec::new(),
+ UnixMillis(0),
+ ) {
Prepared::Ready {
plan: Plan::Replay(paths, _),
own_missing,
@@ -607,7 +615,14 @@ mod tests {
#[test]
fn quarantines_outside_the_window() {
let (old, swarm, target) = scenario(&json!({ "replayWindow": { "epochs": 0 } }));
- match prepare(swarm, old, Epoch::GENESIS, target, Vec::new(), 0) {
+ match prepare(
+ swarm,
+ old,
+ Epoch::GENESIS,
+ target,
+ Vec::new(),
+ UnixMillis(0),
+ ) {
Prepared::Ready {
plan: Plan::Quarantine { paths, .. },
..
@@ -626,7 +641,7 @@ mod tests {
Epoch::GENESIS,
target,
Vec::new(),
- 0
+ UnixMillis(0)
),
Prepared::Ready {
plan: Plan::Nothing,
@@ -646,7 +661,7 @@ mod tests {
Epoch::GENESIS,
target,
Vec::new(),
- 0
+ UnixMillis(0)
),
Prepared::NotYet(_)
));
diff --git a/src/remake/content.rs b/src/remake/content.rs
index e5d78c5..f3e549e 100644
--- a/src/remake/content.rs
+++ b/src/remake/content.rs
@@ -85,7 +85,12 @@ async fn fetch<H: Host>(host: &H, hash: [u8; 32], reference: &ContentRef) -> Opt
}
/// Drops content no open remake in `records` refers to (after its grace period).
-pub(super) fn prune<H: Host>(host: &H, records: &[(String, Record)], now: u64, grace: u64) {
+pub(super) fn prune<H: Host>(
+ host: &H,
+ records: &[(String, Record)],
+ now: crate::clock::UnixMillis,
+ grace: u64,
+) {
let referenced: HashSet<[u8; 32]> = records
.iter()
.filter(|(_, record)| record.phase(now, grace).is_open())
diff --git a/src/remake/local.rs b/src/remake/local.rs
index c6a4c18..a9a88b4 100644
--- a/src/remake/local.rs
+++ b/src/remake/local.rs
@@ -15,20 +15,21 @@ use tokio::time::Instant;
use super::records::Retention;
use super::snapshot::Snapshot;
+use crate::clock::UnixMillis;
use crate::document::replica::Epoch;
use crate::locks::MutexExt;
pub(crate) struct RemakeState {
/// When this node started (Unix milliseconds): votes it cast before may be lost
/// (`vote`).
- started_at: u64,
+ started_at: UnixMillis,
// A std mutex: every access is short and never spans an await.
inner: Mutex<Inner>,
}
impl RemakeState {
/// `started_at`: now, by the node's clock.
- pub(crate) fn new(started_at: u64) -> Self {
+ pub(crate) fn new(started_at: UnixMillis) -> Self {
Self {
started_at,
inner: Mutex::default(),
@@ -94,7 +95,7 @@ pub(crate) struct Quarantined {
/// The epoch the changes were made in, and the one the node rejoined.
pub(crate) epoch: Epoch,
pub(crate) target: Epoch,
- pub(crate) created_at: u64,
+ pub(crate) created_at: UnixMillis,
pub(crate) reason: String,
/// Store paths the changes touched (minimal, prefix-free).
pub(crate) paths: Vec<Vec<String>>,
@@ -118,7 +119,7 @@ pub(crate) struct CatchUpReport {
#[serde(rename_all = "camelCase")]
pub(crate) struct LocalStatus {
#[serde(skip_serializing_if = "Option::is_none")]
- pub(crate) switched_at: Option<u64>,
+ pub(crate) switched_at: Option<UnixMillis>,
#[serde(skip_serializing_if = "Option::is_none")]
pub(crate) replayed_paths: Option<usize>,
#[serde(skip_serializing_if = "Option::is_none")]
@@ -136,7 +137,7 @@ struct Archived {
// Kept for auditing and recovery by hand; not read by the protocol.
#[allow(dead_code)]
doc: AutoCommit,
- archived_at: u64,
+ archived_at: UnixMillis,
retention: Retention,
}
@@ -146,11 +147,11 @@ struct Archived {
pub(crate) struct ArchivedEpoch {
pub(crate) generation: u64,
pub(crate) id: String,
- pub(crate) archived_at: u64,
+ pub(crate) archived_at: UnixMillis,
}
impl RemakeState {
- pub(crate) fn started_at(&self) -> u64 {
+ pub(crate) fn started_at(&self) -> UnixMillis {
self.started_at
}
@@ -222,7 +223,7 @@ impl RemakeState {
doc: AutoCommit,
retention: Retention,
current_generation: u64,
- now: u64,
+ now: UnixMillis,
) {
let mut inner = self.lock();
inner.archive.push(Archived {
@@ -234,7 +235,7 @@ impl RemakeState {
prune(&mut inner.archive, current_generation, now);
}
- pub(crate) fn prune_archive(&self, current_generation: u64, now: u64) {
+ pub(crate) fn prune_archive(&self, current_generation: u64, now: UnixMillis) {
prune(&mut self.lock().archive, current_generation, now);
}
@@ -371,11 +372,11 @@ impl RemakeState {
}
/// An archive is kept until every limit of its retention has passed.
-fn prune(archive: &mut Vec<Archived>, current_generation: u64, now: u64) {
+fn prune(archive: &mut Vec<Archived>, current_generation: u64, now: UnixMillis) {
archive.retain(|archived| {
archived.retention.covers(
current_generation.saturating_sub(archived.epoch.generation),
- now.saturating_sub(archived.archived_at),
+ now.ms_since(archived.archived_at),
)
});
}
@@ -386,8 +387,8 @@ mod tests {
#[test]
fn archive_retention() {
- let remake = RemakeState::new(0);
- let now = 1_000_000;
+ let remake = RemakeState::new(UnixMillis(0));
+ let now = UnixMillis(1_000_000);
let epoch = |generation| Epoch {
generation,
id: [generation as u8; 32],
diff --git a/src/remake/mod.rs b/src/remake/mod.rs
index bceb717..38c2b9c 100644
--- a/src/remake/mod.rs
+++ b/src/remake/mod.rs
@@ -62,8 +62,8 @@ pub(crate) trait Host: Clone + Send + Sync + 'static {
/// judged with it. Shared, so closures run under the document lock can read it.
fn clock(&self) -> Arc<crate::clock::Clock>;
- /// Now, in Unix milliseconds, by [`clock`](Self::clock).
- fn now_millis(&self) -> u64 {
+ /// Now, by [`clock`](Self::clock).
+ fn now_millis(&self) -> crate::clock::UnixMillis {
self.clock().now_millis()
}
@@ -78,7 +78,7 @@ pub(crate) trait Host: Clone + Send + Sync + 'static {
/// The electorate of a remake proposed now (`vote`): the members that said they are
/// alive recently enough, read from `doc`.
- fn electorate(&self, doc: &AutoCommit, now_ms: u64) -> Vec<String>;
+ fn electorate(&self, doc: &AutoCommit, now: crate::clock::UnixMillis) -> Vec<String>;
/// Downloads a change from peers on the local epoch, in chunks from several at once,
/// checked against its hash.
diff --git a/src/remake/quarantine.rs b/src/remake/quarantine.rs
index b5166c9..5300f54 100644
--- a/src/remake/quarantine.rs
+++ b/src/remake/quarantine.rs
@@ -31,6 +31,7 @@ use super::delta;
use super::kind::Translator;
use super::local::Quarantined;
use crate::api::error_response;
+use crate::clock::UnixMillis;
use crate::document::replica::Epoch;
use crate::document::store;
@@ -42,7 +43,7 @@ struct Summary {
epoch: String,
/// The epoch the node rejoined.
target: String,
- created_at: u64,
+ created_at: UnixMillis,
reason: String,
paths: usize,
}
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<Vec<String>>,
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<Vec<String>>,
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::<Vec<_>>(), ["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]
diff --git a/src/remake/snapshot.rs b/src/remake/snapshot.rs
index 301b32f..d093a33 100644
--- a/src/remake/snapshot.rs
+++ b/src/remake/snapshot.rs
@@ -30,6 +30,7 @@ use sha2::{Digest, Sha256};
use super::kind::{DEFAULT_KIND, RemakeKind, Translator, by_name};
use super::records::{EpochRecord, Policy};
+use crate::clock::UnixMillis;
use crate::document::meta::{self, EPOCHS, REMAKES, VOTES};
use crate::document::replica::Epoch;
use crate::document::store::{self, StoreError};
@@ -44,8 +45,8 @@ pub(crate) struct SnapshotParams<'a> {
pub(crate) base_heads: &'a [ChangeHash],
/// Retention policy from the remake request; stored verbatim (canonicalized).
pub(crate) policy: &'a JsonValue,
- /// When the remake was proposed (unix ms); recorded for replay windows.
- pub(crate) proposed_at: u64,
+ /// When the remake was proposed; recorded for replay windows.
+ pub(crate) proposed_at: UnixMillis,
/// Name of the remake kind, which builds the user content.
pub(crate) kind: &'a str,
/// The kind's build input (the proposal's content).
@@ -87,7 +88,7 @@ pub(crate) struct SnapshotRequest {
pub(crate) previous: Epoch,
pub(crate) base_heads: Vec<ChangeHash>,
pub(crate) policy: JsonValue,
- pub(crate) proposed_at: u64,
+ pub(crate) proposed_at: UnixMillis,
pub(crate) kind: String,
pub(crate) content: Option<std::sync::Arc<Vec<u8>>>,
pub(crate) translation: JsonValue,
@@ -136,8 +137,8 @@ pub(crate) struct EpochMetadata {
/// Highest sequence number per actor of `previous` contained in the snapshot.
pub(crate) clock: BTreeMap<String, u64>,
pub(crate) policy: Policy,
- /// When the remake was proposed (unix ms): the point the swarm moved on.
- pub(crate) proposed_at: u64,
+ /// When the remake was proposed: the point the swarm moved on.
+ pub(crate) proposed_at: UnixMillis,
/// The remake kind and its translation input, for writes made before the epoch.
pub(crate) kind: String,
pub(crate) translation: JsonValue,
@@ -163,7 +164,7 @@ struct MetadataRecord {
clock: BTreeMap<String, u64>,
policy: Policy,
#[serde(default)]
- proposed_at: u64,
+ proposed_at: UnixMillis,
#[serde(default = "default_kind")]
kind: String,
#[serde(default)]
@@ -393,7 +394,7 @@ mod tests {
previous: Epoch::GENESIS,
base_heads: heads,
policy,
- proposed_at: 0,
+ proposed_at: UnixMillis(0),
kind: "compact",
content: None,
translation: &NULL,
diff --git a/src/remake/vote.rs b/src/remake/vote.rs
index 40ec6bf..cc805e3 100644
--- a/src/remake/vote.rs
+++ b/src/remake/vote.rs
@@ -29,6 +29,7 @@ use tracing::{info, warn};
use super::Host;
use super::content;
use super::records::{self, Decision, EpochRecord, Phase, Record, Register, Vote};
+use crate::clock::UnixMillis;
use crate::document::store::StoreError;
/// A majority of `electorate` members.
@@ -162,7 +163,7 @@ pub(crate) fn freedom(
epoch: &EpochRecord,
all: &[(String, Record)],
register: &Register,
- started: u64,
+ started: UnixMillis,
) -> Freedom {
match held_by(me, register, all) {
Held::Free => {}
@@ -200,8 +201,8 @@ pub(crate) fn choice(
epoch: &EpochRecord,
all: &[(String, Record)],
register: &Register,
- started: u64,
- now: u64,
+ started: UnixMillis,
+ now: UnixMillis,
grace: u64,
) -> Option<String> {
let only = match freedom(me, epoch, all, register, started) {
@@ -231,7 +232,7 @@ pub(super) async fn cast<H: Host>(host: &H) {
let me = host.node_id();
let epoch = EpochRecord::from(host.current_epoch());
let grace = host.dormant_after().as_millis() as u64;
- let started = host.remake().started_at() + grace;
+ let started = host.remake().started_at().plus_ms(grace);
// Usually there is nothing to vote on: look first, without writing.
let (voter, wanted) = (me.clone(), epoch.clone());
@@ -294,6 +295,7 @@ mod tests {
use serde_json::Value as JsonValue;
fn record(proposed_at: u64, electorate: &[&str]) -> Record {
+ let proposed_at = UnixMillis(proposed_at);
Record {
proposal: Proposal {
coordinator: electorate[0].to_owned(),
@@ -305,7 +307,7 @@ mod tests {
},
settle_ms: 1_000,
proposed_at,
- settle_until: proposed_at + 1_000,
+ settle_until: proposed_at.plus_ms(1_000),
switch_delay_ms: None,
electorate: electorate.iter().map(|m| (*m).to_owned()).collect(),
quorum: quorum(electorate.len()),
@@ -329,7 +331,7 @@ mod tests {
.iter()
.map(|remake| Vote {
remake: (*remake).to_owned(),
- at: 0,
+ at: UnixMillis(0),
})
.collect();
((*voter).to_owned(), votes)
@@ -340,14 +342,14 @@ mod tests {
fn aborted(mut record: Record) -> Record {
record.decision = Some(Decision::Aborted {
reason: "test".into(),
- decided_at: 150,
+ decided_at: UnixMillis(150),
});
record
}
const ALL: [&str; 5] = ["a", "b", "c", "d", "e"];
/// Long before any proposal: no restart concerns.
- const STARTED: u64 = 0;
+ const STARTED: UnixMillis = UnixMillis(0);
fn genesis() -> EpochRecord {
Epoch::GENESIS.into()
@@ -395,7 +397,7 @@ mod tests {
assert_eq!((y.yes, y.elsewhere, y.pending), (1, 0, 4));
// c may vote again.
assert_eq!(
- choice("c", &genesis(), &all, &votes, STARTED, 250, 0).as_deref(),
+ choice("c", &genesis(), &all, &votes, STARTED, UnixMillis(250), 0).as_deref(),
Some("y")
);
}
@@ -408,9 +410,9 @@ mod tests {
("y".to_owned(), record(5_000, &ALL)),
];
let votes = register(&[("a", &["x"]), ("b", &["x"]), ("c", &["x"])]);
- assert_eq!(all[0].1.phase(5_000, 0), Phase::Expired);
+ assert_eq!(all[0].1.phase(UnixMillis(5_000), 0), Phase::Expired);
assert_eq!(
- choice("a", &genesis(), &all, &votes, STARTED, 5_000, 0),
+ choice("a", &genesis(), &all, &votes, STARTED, UnixMillis(5_000), 0),
None
);
let y = tally("y", &all[1].1, &all, &votes);
@@ -452,21 +454,32 @@ mod tests {
];
let none = register(&[]);
assert_eq!(
- choice("a", &genesis(), &all, &none, STARTED, 150, 0).as_deref(),
+ choice("a", &genesis(), &all, &none, STARTED, UnixMillis(150), 0).as_deref(),
Some("x")
);
assert_eq!(
- choice("outsider", &genesis(), &all, &none, STARTED, 150, 0),
+ choice(
+ "outsider",
+ &genesis(),
+ &all,
+ &none,
+ STARTED,
+ UnixMillis(150),
+ 0
+ ),
None
);
let voted = register(&[("a", &["y"])]);
- assert_eq!(choice("a", &genesis(), &all, &voted, STARTED, 150, 0), None);
+ assert_eq!(
+ choice("a", &genesis(), &all, &voted, STARTED, UnixMillis(150), 0),
+ None
+ );
}
#[test]
fn after_a_start_only_the_remake_voted_for_before() {
- let started = 150;
+ let started = UnixMillis(150);
// Proposed before the start: a may have voted for x then.
let all = vec![
("x".to_owned(), record(100, &ALL)),
@@ -478,7 +491,16 @@ mod tests {
Freedom::Only("x".to_owned())
);
assert_eq!(
- choice("a", &genesis(), &all, &none, started, 250, 1_000).as_deref(),
+ choice(
+ "a",
+ &genesis(),
+ &all,
+ &none,
+ started,
+ UnixMillis(250),
+ 1_000
+ )
+ .as_deref(),
Some("x")
);
@@ -493,7 +515,15 @@ mod tests {
Freedom::Held(_)
));
assert_eq!(
- choice("a", &genesis(), &all, &none, started, 250, 1_000),
+ choice(
+ "a",
+ &genesis(),
+ &all,
+ &none,
+ started,
+ UnixMillis(250),
+ 1_000
+ ),
None
);