1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109

use std::time::Duration;

use datacake_crdt::HLCTimestamp;
use tokio::sync::oneshot;

use crate::NodeId;

const CLOCK_BACKPRESSURE_LIMIT: u16 = u16::MAX - 10;

#[derive(Clone)]
pub struct Clock {
    node_id: NodeId,
    tx: flume::Sender<Event>,
}

impl Clock {
    pub fn new(node_id: NodeId) -> Self {
        let ts = HLCTimestamp::now(0, node_id);
        let (tx, rx) = flume::bounded(1000);

        tokio::spawn(run_clock(ts, rx));

        Self { node_id, tx }
    }

    pub async fn register_ts(&self, ts: HLCTimestamp) {
        if ts.node() == self.node_id {
            return;
        }

        self.tx
            .send_async(Event::Register(ts))
            .await
            .expect("Clock actor should never die");
    }

    pub async fn get_time(&self) -> HLCTimestamp {
        let (tx, rx) = oneshot::channel();

        self.tx
            .send_async(Event::Get(tx))
            .await
            .expect("Clock actor should never die");

        rx.await.expect("Responder should not be dropped")
    }
}

pub enum Event {
    Get(oneshot::Sender<HLCTimestamp>),
    Register(HLCTimestamp),
}

async fn run_clock(mut clock: HLCTimestamp, reqs: flume::Receiver<Event>) {
    while let Ok(event) = reqs.recv_async().await {
        match event {
            Event::Get(tx) => {
                let ts = clock.send().expect("Clock counter should not overflow");

                if clock.counter() >= CLOCK_BACKPRESSURE_LIMIT {
                    tokio::time::sleep(Duration::from_millis(1)).await;
                }

                let _ = tx.send(ts);
            },
            Event::Register(remote_ts) => {
                let _ = clock.recv(&remote_ts);

                if clock.counter() >= CLOCK_BACKPRESSURE_LIMIT {
                    tokio::time::sleep(Duration::from_millis(1)).await;
                }
            },
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn test_clock() {
        let clock = Clock::new(0);

        let ts1 = clock.get_time().await;
        clock.register_ts(ts1).await;
        let ts2 = clock.get_time().await;
        assert!(ts1 < ts2);

        let ts1 = clock.get_time().await;
        let ts2 = clock.get_time().await;
        let ts3 = clock.get_time().await;
        assert!(ts1 < ts2);
        assert!(ts2 < ts3);

        let drift_ts =
            HLCTimestamp::new(ts3.datacake_timestamp() + Duration::from_secs(50), 0, 1);
        clock.register_ts(drift_ts).await;
        let ts = clock.get_time().await;
        assert!(
            drift_ts < ts,
            "New timestamp should be monotonic relative to drifted ts."
        );

        let old_ts =
            HLCTimestamp::new(ts3.datacake_timestamp() + Duration::from_secs(5), 0, 1);
        clock.register_ts(old_ts).await;
    }
}