wrap unix timestamps in newtype

17 files changed, 207 insertions, 123 deletions

diff --git a/src/api/peers.rs b/src/api/peers.rs
index c7b3c50..e22f3cd 100644
--- a/src/api/peers.rs
+++ b/src/api/peers.rs
@@ -223,7 +223,8 @@ pub(crate) async fn delete_peer(
         }
     }
 
-    let record = serde_json::json!({ "at": state.clock.now_millis(), "by": state.local_id() });
+    let record =
+        serde_json::json!({ "at": state.clock.now_millis().as_u64(), "by": state.local_id() });
 
     let peer_clone = peer.clone();
     state
diff --git a/src/clock.rs b/src/clock.rs
index dafd44b..7613930 100644
--- a/src/clock.rs
+++ b/src/clock.rs
@@ -7,6 +7,46 @@
 //! isn't affected.
 
 use std::sync::atomic::{AtomicI64, Ordering};
+use std::time::Duration;
+
+use serde::{Deserialize, Serialize};
+
+/// A point in time as Unix milliseconds, as stored in the document and reported: not a
+/// duration (those are `Duration`, or plain milliseconds named `…_ms`). Comparable
+/// across nodes as far as their clocks agree; for elapsed time within a node use
+/// tokio's `Instant`.
+#[derive(
+    Clone, Copy, Debug, Default, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize,
+)]
+#[serde(transparent)]
+pub(crate) struct UnixMillis(pub(crate) u64);
+
+impl UnixMillis {
+    pub(crate) fn as_u64(self) -> u64 {
+        self.0
+    }
+
+    /// `ms` milliseconds later (saturating).
+    pub(crate) fn plus_ms(self, ms: u64) -> Self {
+        Self(self.0.saturating_add(ms))
+    }
+
+    /// `duration` later (saturating, to the millisecond).
+    pub(crate) fn plus(self, duration: Duration) -> Self {
+        self.plus_ms(u64::try_from(duration.as_millis()).unwrap_or(u64::MAX))
+    }
+
+    /// Milliseconds from `earlier` to this; 0 if `earlier` is later.
+    pub(crate) fn ms_since(self, earlier: Self) -> u64 {
+        self.0.saturating_sub(earlier.0)
+    }
+}
+
+impl std::fmt::Display for UnixMillis {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        self.0.fmt(f)
+    }
+}
 
 /// The node's wall clock: the system clock plus an offset (zero outside tests).
 #[derive(Debug, Default)]
@@ -15,10 +55,10 @@ pub(crate) struct Clock {
 }
 
 impl Clock {
-    /// Now, in Unix milliseconds.
-    pub(crate) fn now_millis(&self) -> u64 {
+    /// Now.
+    pub(crate) fn now_millis(&self) -> UnixMillis {
         let offset = self.offset_ms.load(Ordering::Relaxed);
-        system_millis().saturating_add_signed(offset)
+        UnixMillis(system_millis().saturating_add_signed(offset))
     }
 
     /// Puts this clock `offset_ms` ahead of the system clock (behind if negative).
@@ -46,7 +86,7 @@ mod tests {
         let before = clock.now_millis();
         clock.set_offset_ms(3_600_000);
         let ahead = clock.now_millis();
-        assert!(ahead >= before + 3_600_000);
+        assert!(ahead >= before.plus_ms(3_600_000));
         clock.set_offset_ms(-3_600_000);
         assert!(clock.now_millis() < before);
     }
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
         );
 
diff --git a/src/state/host.rs b/src/state/host.rs
index fd40bf3..6506274 100644
--- a/src/state/host.rs
+++ b/src/state/host.rs
@@ -85,8 +85,8 @@ impl Host for AppState {
         self.timing.dormant_after
     }
 
-    fn electorate(&self, doc: &AutoCommit, now_ms: u64) -> Vec<String> {
-        crate::swarm::members::electorate(doc, now_ms, self.electorate_window)
+    fn electorate(&self, doc: &AutoCommit, now: crate::clock::UnixMillis) -> Vec<String> {
+        crate::swarm::members::electorate(doc, now.as_u64(), self.electorate_window)
     }
 
     fn fetch_change(
diff --git a/src/state/mod.rs b/src/state/mod.rs
index f93bbef..438ad52 100644
--- a/src/state/mod.rs
+++ b/src/state/mod.rs
@@ -385,7 +385,7 @@ impl AppState {
                         // how long the change took to reach them (`swarm::stats`).
                         replica.doc.commit_with(
                             automerge::transaction::CommitOptions::default()
-                                .with_time(state.clock.now_millis() as i64),
+                                .with_time(state.clock.now_millis().as_u64() as i64),
                         );
                     }
                     // Discard partial ops (e.g. old value deleted, new one not written).
diff --git a/src/swarm/members.rs b/src/swarm/members.rs
index 19abf2b..cb8a558 100644
--- a/src/swarm/members.rs
+++ b/src/swarm/members.rs
@@ -167,7 +167,7 @@ pub(crate) async fn ensure_registered(state: &AppState) {
         )
     };
     let wanted = state.alias.clone();
-    let now = state.clock.now_millis();
+    let now = state.clock.now_millis().as_u64();
 
     if !registered {
         let mut entry = json!({ "since": now, "alive": now });
diff --git a/src/swarm/state_file.rs b/src/swarm/state_file.rs
index ba9192e..73b80a2 100644
--- a/src/swarm/state_file.rs
+++ b/src/swarm/state_file.rs
@@ -132,7 +132,7 @@ pub(crate) async fn update(state: &AppState, now: bool) {
             &replica.doc,
             state.local_id(),
             state.electorate_window,
-            state.clock.now_millis(),
+            state.clock.now_millis().as_u64(),
         )
     };
     let remembered = Remembered { swarm_id, seeds };
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 07f66a3..805d700 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -336,7 +336,7 @@ pub(crate) async fn handle_sync(
             stats.heads_seen(heads_after.len());
             let changed = heads_after != heads_before;
             if changed && measure_propagation {
-                let now_ms = clock.now_millis() as i64;
+                let now_ms = clock.now_millis().as_u64() as i64;
                 // Time 0: not stamped (snapshots, older nodes).
                 let delays: Vec<i64> = replica
                     .doc
diff --git a/src/units.rs b/src/units.rs
index 5128b6e..cec5ffc 100644
--- a/src/units.rs
+++ b/src/units.rs
@@ -162,28 +162,13 @@ pub(crate) mod duration_option {
     }
 }
 
-/// Serde for an optional `Duration` in a request (JSON body or query string), read as
-/// [`duration_option`] reads it; absent is `None`. Use with `#[serde(default)]`.
-pub(crate) mod optional_duration {
-    use std::time::Duration;
-
-    use serde::{Deserializer, Serializer};
-
-    pub(crate) fn serialize<S: Serializer>(
-        value: &Option<Duration>,
-        s: S,
-    ) -> Result<S::Ok, S::Error> {
-        match value {
-            Some(duration) => super::duration_option::serialize(duration, s),
-            None => s.serialize_none(),
-        }
-    }
-
-    pub(crate) fn deserialize<'de, D: Deserializer<'de>>(
-        d: D,
-    ) -> Result<Option<Duration>, D::Error> {
-        super::duration_option::deserialize(d).map(Some)
-    }
+/// Reads an optional `Duration` in a request (JSON body or query string) as
+/// [`duration_option`] reads it; absent is `None`. Use as
+/// `#[serde(default, deserialize_with = "crate::units::optional_duration")]`.
+pub(crate) fn optional_duration<'de, D: serde::Deserializer<'de>>(
+    d: D,
+) -> Result<Option<Duration>, D::Error> {
+    duration_option::deserialize(d).map(Some)
 }
 
 /// Serde for a size option: a string as [`parse_size`] reads it, or a number of bytes;