1pub const HIGHWATER_STATE_KEY: &str = "hlc_highwater";
10
11pub const MAX_FUTURE_SKEW_MS: u64 = 30 * 24 * 60 * 60 * 1000;
20
21pub(crate) const COUNTER_MAX: u16 = 9999;
24
25#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
27pub struct Timestamp {
28 pub millis: u64,
29 pub counter: u16,
30 pub device_id: String,
31}
32
33impl Timestamp {
34 pub fn new(millis: u64, counter: u16, device_id: String) -> Self {
35 Self {
36 millis,
37 counter,
38 device_id,
39 }
40 }
41
42 pub fn is_within_future_bound(&self, receiver_wall_ms: u64) -> bool {
49 self.millis <= receiver_wall_ms.saturating_add(MAX_FUTURE_SKEW_MS)
50 }
51
52 pub fn parse(s: &str) -> Option<Self> {
54 let mut parts = s.splitn(3, '-');
55 let millis = parts.next()?.parse::<u64>().ok()?;
56 let counter = parts.next()?.parse::<u16>().ok()?;
57 let device_id = parts.next()?;
58 if device_id.is_empty() {
59 return None;
60 }
61 if counter > COUNTER_MAX {
62 return None;
63 }
64 Some(Self {
65 millis,
66 counter,
67 device_id: device_id.to_string(),
68 })
69 }
70}
71
72impl std::fmt::Display for Timestamp {
73 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
74 write!(
75 f,
76 "{:013}-{:04}-{}",
77 self.millis, self.counter, self.device_id
78 )
79 }
80}
81
82use std::sync::{Arc, Mutex, MutexGuard};
104
105#[derive(Clone, Copy)]
110struct HlcState {
111 millis: u64,
112 counter: u16,
113}
114
115fn increment(state: &mut HlcState) {
116 if state.counter < COUNTER_MAX {
117 state.counter += 1;
118 } else if let Some(next_millis) = state.millis.checked_add(1) {
119 state.millis = next_millis;
120 state.counter = 0;
121 } else {
122 state.counter = COUNTER_MAX;
123 }
124}
125
126impl HlcState {
127 fn seed(&mut self, floor: &Timestamp) {
128 if floor.millis > self.millis
129 || (floor.millis == self.millis && floor.counter > self.counter)
130 {
131 self.millis = floor.millis;
132 self.counter = floor.counter;
133 }
134 }
135
136 fn tick(&mut self, wall: u64) {
137 if wall > self.millis {
138 self.millis = wall;
139 self.counter = 0;
140 } else {
141 increment(self);
142 }
143 }
144
145 fn advance_past(&mut self, wall: u64, remote: &Timestamp) {
146 if wall > self.millis && wall > remote.millis {
147 self.millis = wall;
148 self.counter = 0;
149 } else {
150 self.seed(remote);
151 }
152 }
153
154 fn timestamp(&self, device_id: &str) -> Timestamp {
155 Timestamp::new(self.millis, self.counter, device_id.to_owned())
156 }
157}
158
159pub struct Hlc {
160 device_id: String,
161 state: Mutex<HlcState>,
162 clock: coven_foundation::clock::ClockRef,
163}
164
165impl Hlc {
166 pub fn try_new(
167 device_id: String,
168 clock: coven_foundation::clock::ClockRef,
169 ) -> Result<Self, coven_foundation::store_dir::PathTokenError> {
170 coven_foundation::store_dir::validate_path_token(&device_id)?;
171 Ok(Self {
172 device_id,
173 state: Mutex::new(HlcState {
174 millis: 0,
175 counter: 0,
176 }),
177 clock,
178 })
179 }
180
181 pub fn new(device_id: String, clock: coven_foundation::clock::ClockRef) -> Self {
183 Self::try_new(device_id, clock).expect("device_id must be a safe path token")
184 }
185
186 pub fn device_id(&self) -> &str {
187 &self.device_id
188 }
189
190 fn wall_millis(&self) -> u64 {
191 self.clock
192 .now()
193 .timestamp_millis()
194 .try_into()
195 .expect("clock before UNIX epoch")
196 }
197
198 pub fn seed(&self, high_water: &Timestamp) {
205 self.state.lock().unwrap().seed(high_water);
206 }
207
208 pub fn high_water(&self) -> Timestamp {
213 let state = self.state.lock().unwrap();
214 state.timestamp(&self.device_id)
215 }
216
217 pub fn transaction(&self) -> HlcTransaction<'_> {
225 let state = self.state.lock().unwrap();
226 let staged = *state;
227 HlcTransaction {
228 hlc: self,
229 state,
230 staged,
231 }
232 }
233
234 pub fn wall_now_ms(&self) -> u64 {
241 self.wall_millis()
242 }
243
244 pub fn now(&self) -> Timestamp {
247 let wall = self.wall_millis();
248 let mut state = self.state.lock().unwrap();
249
250 state.tick(wall);
251 state.timestamp(&self.device_id)
252 }
253
254 pub fn advance_past(&self, remote: &Timestamp) {
264 let wall = self.wall_millis();
265 let mut state = self.state.lock().unwrap();
266
267 state.advance_past(wall, remote);
268 }
269}
270
271pub struct HlcTransaction<'clock> {
278 hlc: &'clock Hlc,
279 state: MutexGuard<'clock, HlcState>,
280 staged: HlcState,
281}
282
283impl HlcTransaction<'_> {
284 pub fn now(&mut self) -> Timestamp {
286 self.staged.tick(self.hlc.wall_millis());
287 self.high_water()
288 }
289
290 pub fn advance_past(&mut self, remote: &Timestamp) {
292 self.staged.advance_past(self.hlc.wall_millis(), remote);
293 }
294
295 pub fn high_water(&self) -> Timestamp {
297 self.staged.timestamp(&self.hlc.device_id)
298 }
299
300 pub fn commit(mut self) {
302 *self.state = self.staged;
303 }
304}
305
306#[derive(Clone)]
316pub struct UpdatedAtStamper {
317 hlc: Arc<Hlc>,
318}
319
320impl UpdatedAtStamper {
321 pub fn new(hlc: Arc<Hlc>) -> Self {
322 Self { hlc }
323 }
324
325 pub fn stamp(&self) -> String {
329 self.hlc.now().to_string()
330 }
331}
332
333#[cfg(any(test, feature = "test-utils"))]
340pub fn now_wall_ms(clock: &dyn coven_foundation::clock::Clock) -> u64 {
341 clock
342 .now()
343 .timestamp_millis()
344 .try_into()
345 .expect("clock before UNIX epoch")
346}
347
348#[cfg(test)]
349mod tests {
350 use super::*;
351 use coven_foundation::clock::{ClosureClock, FixedClock, SystemClock};
352 use std::sync::atomic::{AtomicU64, Ordering};
353
354 fn instant(ms: u64) -> chrono::DateTime<chrono::Utc> {
355 chrono::DateTime::from_timestamp_millis(ms.try_into().expect("test millis fit in i64"))
356 .expect("valid test clock instant")
357 }
358
359 fn fixed_clock(ms: u64) -> coven_foundation::clock::ClockRef {
360 Arc::new(FixedClock(instant(ms)))
361 }
362
363 fn advancing_clock(start: u64) -> (Arc<AtomicU64>, coven_foundation::clock::ClockRef) {
364 let time = Arc::new(AtomicU64::new(start));
365 let time_clone = time.clone();
366 (
367 time,
368 Arc::new(ClosureClock(move || {
369 instant(time_clone.load(Ordering::SeqCst))
370 })),
371 )
372 }
373
374 #[test]
375 fn basic_monotonicity() {
376 let hlc = Hlc::new("dev-1".into(), Arc::new(SystemClock));
377 let t1 = hlc.now();
378 let t2 = hlc.now();
379 let t3 = hlc.now();
380
381 assert!(t2 > t1, "t2={t2} should be > t1={t1}");
382 assert!(t3 > t2, "t3={t3} should be > t2={t2}");
383 }
384
385 #[test]
386 fn new_rejects_empty_device_id() {
387 assert!(matches!(
388 Hlc::try_new(String::new(), Arc::new(SystemClock)),
389 Err(coven_foundation::store_dir::PathTokenError::Empty),
390 ));
391 }
392
393 #[test]
394 fn counter_increments_when_clock_stalls() {
395 let hlc = Hlc::new("dev-1".into(), fixed_clock(1000));
396
397 let t1 = hlc.now();
398 assert_eq!(t1.millis, 1000);
399 assert_eq!(t1.counter, 0);
400
401 let t2 = hlc.now();
402 assert_eq!(t2.millis, 1000);
403 assert_eq!(t2.counter, 1);
404
405 let t3 = hlc.now();
406 assert_eq!(t3.millis, 1000);
407 assert_eq!(t3.counter, 2);
408
409 assert!(t3 > t2);
410 assert!(t2 > t1);
411 }
412
413 #[test]
414 fn transaction_rollback_discards_accepted_floor_and_stamps() {
415 let hlc = Hlc::new("dev-local".into(), fixed_clock(1000));
416 let before = hlc.now();
417 let remote = Timestamp::new(5000, COUNTER_MAX, "dev-remote".into());
418 {
419 let mut transaction = hlc.transaction();
420 transaction.advance_past(&remote);
421 let first = transaction.now();
422 let second = transaction.now();
423 assert!(first > remote);
424 assert!(second > first);
425 assert_eq!(transaction.high_water(), second);
426 }
427 assert_eq!(hlc.high_water(), before);
428 assert_eq!(hlc.now(), Timestamp::new(1000, 1, "dev-local".into()));
429 }
430
431 #[test]
432 fn transaction_commit_publishes_stamps_above_the_accepted_floor() {
433 let hlc = Hlc::new("dev-local".into(), fixed_clock(1000));
434 let remote = Timestamp::new(5000, COUNTER_MAX, "dev-remote".into());
435 let mut transaction = hlc.transaction();
436 transaction.advance_past(&remote);
437 let first = transaction.now();
438 let second = transaction.now();
439 assert_eq!(first, Timestamp::new(5001, 0, "dev-local".into()));
440 assert!(second > first);
441 transaction.advance_past(&remote);
442 let persisted = transaction.high_water();
443 assert_eq!(persisted, second);
444 transaction.commit();
445
446 assert_eq!(hlc.high_water(), persisted);
447 assert!(hlc.now() > persisted);
448 }
449
450 #[test]
451 fn transaction_serializes_external_stamping_on_commit_and_rollback() {
452 for commit in [true, false] {
453 let hlc = Arc::new(Hlc::new("dev-local".into(), fixed_clock(1000)));
454 let before = hlc.now();
455 let mut transaction = hlc.transaction();
456 transaction.advance_past(&Timestamp::new(5000, 3, "dev-remote".into()));
457 let staged = transaction.now();
458 let stamper = UpdatedAtStamper::new(hlc.clone());
459 let (ready, started) = std::sync::mpsc::channel();
460 let worker = std::thread::spawn(move || {
461 ready.send(()).expect("notify concurrent stamper started");
462 stamper.stamp()
463 });
464 started
465 .recv_timeout(std::time::Duration::from_secs(30))
466 .expect("concurrent stamper starts");
467 assert!(matches!(
468 hlc.state.try_lock(),
469 Err(std::sync::TryLockError::WouldBlock)
470 ));
471 if commit {
472 transaction.commit();
473 } else {
474 drop(transaction);
475 }
476 let external = Timestamp::parse(&worker.join().expect("concurrent stamp succeeds"))
477 .expect("stamper returns canonical timestamp");
478 if commit {
479 assert_eq!(external.millis, staged.millis);
480 assert_eq!(external.counter, staged.counter + 1);
481 assert!(external > staged);
482 } else {
483 assert_eq!(external.millis, before.millis);
484 assert_eq!(external.counter, before.counter + 1);
485 assert!(external < staged);
486 }
487 }
488 }
489
490 #[test]
491 fn wall_clock_advance_resets_counter() {
492 let (time, clock) = advancing_clock(1000);
493 let hlc = Hlc::new("dev-1".into(), clock);
494
495 let t1 = hlc.now();
496 assert_eq!(t1.millis, 1000);
497 assert_eq!(t1.counter, 0);
498
499 let t2 = hlc.now();
501 assert_eq!(t2.counter, 1);
502
503 time.store(2000, Ordering::SeqCst);
505 let t3 = hlc.now();
506 assert_eq!(t3.millis, 2000);
507 assert_eq!(t3.counter, 0);
508
509 assert!(t3 > t2);
510 }
511
512 #[test]
513 fn advance_past_remote_ahead() {
514 let hlc = Hlc::new("dev-local".into(), fixed_clock(1000));
515
516 let remote = Timestamp::new(5000, 3, "dev-remote".into());
518 hlc.advance_past(&remote);
519
520 let t = hlc.now();
522 assert!(
523 t.to_string() > remote.to_string(),
524 "t={t} must beat {remote}"
525 );
526 assert_eq!(t.millis, 5000);
527 assert_eq!(t.device_id, "dev-local");
528 }
529
530 #[test]
531 fn advance_past_remote_behind() {
532 let hlc = Hlc::new("dev-local".into(), fixed_clock(5000));
533
534 let primed = hlc.now();
536
537 let remote = Timestamp::new(1000, 10, "dev-remote".into());
539 hlc.advance_past(&remote);
540
541 let t = hlc.now();
542 assert!(
543 t > primed,
544 "t={t} must stay above the primed clock {primed}"
545 );
546 assert_eq!(t.millis, 5000);
547 }
548
549 #[test]
556 fn advance_past_far_future_applied_row_is_not_capped() {
557 let hlc = Hlc::new("dev-local".into(), fixed_clock(1000));
558
559 let far_future = 1000 + 48 * 60 * 60 * 1000;
562 let applied = Timestamp::new(far_future, 7, "dev-remote".into());
563 hlc.advance_past(&applied);
564
565 let next = hlc.now();
568 assert!(
569 next.to_string() > applied.to_string(),
570 "next stamp {next} regressed below applied row {applied}: the clock \
571 refused to advance past an authoritative register value",
572 );
573 assert_eq!(next.millis, far_future);
574 }
575
576 #[test]
577 fn string_roundtrip() {
578 let ts = Timestamp::new(1707580800000, 42, "dev-abc123".into());
579 let s = ts.to_string();
580 let parsed = Timestamp::parse(&s).expect("parse should succeed");
581
582 assert_eq!(parsed, ts);
583 assert_eq!(s, "1707580800000-0042-dev-abc123");
584 }
585
586 #[test]
587 fn string_format_is_zero_padded() {
588 let ts = Timestamp::new(1000, 0, "d".into());
589 assert_eq!(ts.to_string(), "0000000001000-0000-d");
590
591 let ts2 = Timestamp::new(9999999999999, 9999, "d".into());
592 assert_eq!(ts2.to_string(), "9999999999999-9999-d");
593 }
594
595 #[test]
596 fn lexicographic_ordering_matches_causal_ordering() {
597 let timestamps = [
598 Timestamp::new(1000, 0, "dev-a".into()),
599 Timestamp::new(1000, 1, "dev-a".into()),
600 Timestamp::new(1000, 1, "dev-b".into()),
601 Timestamp::new(2000, 0, "dev-a".into()),
602 Timestamp::new(2000, 0, "dev-b".into()),
603 ];
604
605 let strings: Vec<String> = timestamps.iter().map(|t| t.to_string()).collect();
606
607 for i in 1..strings.len() {
609 assert!(
610 strings[i] > strings[i - 1],
611 "Expected {:?} > {:?}",
612 strings[i],
613 strings[i - 1]
614 );
615 }
616 }
617
618 #[test]
619 fn device_id_breaks_ties() {
620 let ts_a = Timestamp::new(5000, 3, "aaa".into());
621 let ts_b = Timestamp::new(5000, 3, "bbb".into());
622
623 assert!(ts_b > ts_a);
625
626 assert!(ts_b.to_string() > ts_a.to_string());
628 }
629
630 #[test]
631 fn future_bound_admits_honest_and_rejects_grossly_future() {
632 let wall: u64 = 1_700_000_000_000;
633
634 assert!(Timestamp::new(wall, 0, "d".into()).is_within_future_bound(wall));
636 assert!(Timestamp::new(0, 0, "d".into()).is_within_future_bound(wall));
637
638 let just_inside = wall + MAX_FUTURE_SKEW_MS - 1;
640 assert!(Timestamp::new(just_inside, 0, "d".into()).is_within_future_bound(wall));
641 let at_bound = wall + MAX_FUTURE_SKEW_MS;
643 assert!(Timestamp::new(at_bound, 0, "d".into()).is_within_future_bound(wall));
644
645 let just_beyond = wall + MAX_FUTURE_SKEW_MS + 1;
647 assert!(!Timestamp::new(just_beyond, 0, "d".into()).is_within_future_bound(wall));
648 assert!(!Timestamp::new(u64::MAX, 0, "d".into()).is_within_future_bound(wall));
650
651 assert!(Timestamp::new(u64::MAX, 0, "d".into()).is_within_future_bound(u64::MAX));
654 }
655
656 #[test]
657 fn parse_rejects_invalid_input() {
658 assert!(Timestamp::parse("").is_none());
659 assert!(Timestamp::parse("not-a-timestamp").is_none());
660 assert!(Timestamp::parse("1000-0000").is_none()); assert!(Timestamp::parse("1000-0000-").is_none()); assert!(Timestamp::parse("abc-0000-dev").is_none()); assert!(Timestamp::parse("1000-xyz-dev").is_none()); }
665
666 #[test]
667 fn parse_rejects_counter_above_format_width() {
668 assert!(Timestamp::parse("0000000001000-10000-dev").is_none());
669 assert!(Timestamp::parse("0000000001000-65535-dev").is_none());
670 }
671
672 #[test]
673 fn advance_past_counter_bound_carries_into_millis() {
674 let hlc = Hlc::new("dev-local".into(), fixed_clock(1000));
675 let observed = Timestamp::new(1000, 9999, "dev-remote".into());
676
677 hlc.advance_past(&observed);
678 let next = hlc.now();
679
680 assert!(
681 next.to_string() > observed.to_string(),
682 "next stamp {next} must sort after observed stamp {observed}",
683 );
684 assert_eq!(next.millis, 1001);
685 assert_eq!(next.counter, 0);
686 }
687
688 #[test]
689 fn now_counter_bound_carries_into_millis() {
690 let hlc = Hlc::new("dev-local".into(), fixed_clock(1000));
691 hlc.seed(&Timestamp::new(1000, 9999, "dev-remote".into()));
692
693 let next = hlc.now();
694
695 assert_eq!(next.millis, 1001);
696 assert_eq!(next.counter, 0);
697 assert_eq!(next.to_string(), "0000000001001-0000-dev-local");
698 }
699
700 #[test]
701 fn minted_counters_keep_fixed_width_lexical_order() {
702 let hlc = Hlc::new("dev-local".into(), fixed_clock(1000));
703 hlc.seed(&Timestamp::new(1000, 9998, "dev-remote".into()));
704
705 let stamps = [hlc.now(), hlc.now(), hlc.now()];
706 for stamp in &stamps {
707 let rendered = stamp.to_string();
708 let counter = rendered
709 .split('-')
710 .nth(1)
711 .expect("timestamp has a counter field");
712 assert_eq!(counter.len(), 4);
713 }
714 for pair in stamps.windows(2) {
715 assert!(
716 pair[1] > pair[0],
717 "timestamp ordering must advance from {} to {}",
718 pair[0],
719 pair[1],
720 );
721 assert!(
722 pair[1].to_string() > pair[0].to_string(),
723 "string ordering must advance from {} to {}",
724 pair[0],
725 pair[1],
726 );
727 }
728 }
729
730 #[test]
731 fn parse_handles_device_id_with_dashes() {
732 let ts = Timestamp::new(1000, 0, "550e8400-e29b-41d4-a716-446655440000".into());
735 let s = ts.to_string();
736 let parsed = Timestamp::parse(&s).expect("parse should handle UUID device_id");
737 assert_eq!(parsed.device_id, "550e8400-e29b-41d4-a716-446655440000");
738 assert_eq!(parsed, ts);
739 }
740}