summaryrefslogtreecommitdiffstats
path: root/src/remake/local.rs
diff options
context:
space:
mode:
authortv <tv@krebsco.de>2026-10-25 23:00:56 +0100
committertv <tv@krebsco.de>2026-10-25 23:00:56 +0100
commitc52d4b5ae43c202569c8801c312c2b56954ff8a0 (patch)
treedac7221b966d9f4fdb8e1b91b1e20bce607de6fc /src/remake/local.rs
parent2f7265795b7f6ddf460bf93c5f880fb24bde9d3a (diff)
add clock module
The Clock module is the system clock that every unix timestamp new goes through. It comes with an offset that only tests can set.
Diffstat (limited to 'src/remake/local.rs')
-rw-r--r--src/remake/local.rs30
1 files changed, 18 insertions, 12 deletions
diff --git a/src/remake/local.rs b/src/remake/local.rs
index 65a2735..517367b 100644
--- a/src/remake/local.rs
+++ b/src/remake/local.rs
@@ -13,7 +13,7 @@ use automerge::{ActorId, AutoCommit};
use serde::Serialize;
use tokio::time::Instant;
-use super::records::{Retention, unix_millis};
+use super::records::Retention;
use super::snapshot::Snapshot;
use crate::document::replica::Epoch;
@@ -25,10 +25,11 @@ pub(crate) struct RemakeState {
inner: Mutex<Inner>,
}
-impl Default for RemakeState {
- fn default() -> Self {
+impl RemakeState {
+ /// `started_at`: now, by the node's clock.
+ pub(crate) fn new(started_at: u64) -> Self {
Self {
- started_at: unix_millis(),
+ started_at,
inner: Mutex::default(),
}
}
@@ -215,25 +216,27 @@ impl RemakeState {
}
/// Keeps the previous epoch's document, then drops what has outlived its retention.
+ /// `now`: Unix milliseconds, by the node's clock.
pub(crate) fn archive(
&self,
epoch: Epoch,
doc: AutoCommit,
retention: Retention,
current_generation: u64,
+ now: u64,
) {
let mut inner = self.lock();
inner.archive.push(Archived {
epoch,
doc,
- archived_at: unix_millis(),
+ archived_at: now,
retention,
});
- prune(&mut inner.archive, current_generation);
+ prune(&mut inner.archive, current_generation, now);
}
- pub(crate) fn prune_archive(&self, current_generation: u64) {
- prune(&mut self.lock().archive, current_generation);
+ pub(crate) fn prune_archive(&self, current_generation: u64, now: u64) {
+ prune(&mut self.lock().archive, current_generation, now);
}
pub(crate) fn catch_up(&self) -> Option<CatchUp> {
@@ -369,8 +372,7 @@ impl RemakeState {
}
/// An archive is kept until every limit of its retention has passed.
-fn prune(archive: &mut Vec<Archived>, current_generation: u64) {
- let now = unix_millis();
+fn prune(archive: &mut Vec<Archived>, current_generation: u64, now: u64) {
archive.retain(|archived| {
archived.retention.covers(
current_generation.saturating_sub(archived.epoch.generation),
@@ -385,7 +387,8 @@ mod tests {
#[test]
fn archive_retention() {
- let remake = RemakeState::default();
+ let remake = RemakeState::new(0);
+ let now = 1_000_000;
let epoch = |generation| Epoch {
generation,
id: [generation as u8; 32],
@@ -400,9 +403,10 @@ mod tests {
seconds: None,
},
1,
+ now,
);
assert_eq!(remake.archived_epochs().len(), 1);
- remake.prune_archive(2);
+ remake.prune_archive(2, now);
assert!(remake.archived_epochs().is_empty());
// Kept by time while the epoch count has passed.
@@ -414,6 +418,7 @@ mod tests {
seconds: Some(3600),
},
5,
+ now,
);
assert_eq!(remake.archived_epochs().len(), 1);
@@ -426,6 +431,7 @@ mod tests {
seconds: None,
},
3,
+ now,
);
assert_eq!(remake.archived_epochs().len(), 1);
}