occstep 0.1.0

An event sourcing library based on a distributed log and optimistic concurrency control
Documentation 
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

// mostly generated by ChatGPT, we should double-check it

use crate::Nonce;
use std::sync::atomic::{AtomicU64, AtomicBool, Ordering};
use std::time::{SystemTime, UNIX_EPOCH};

const EPOCH: u64 = 1700000000000; // Custom epoch (e.g., 2023-11-14T00:00:00Z)

// Bit allocations (assuming 64-bit Snowflake ID)
const TIMESTAMP_BITS: u8 = 41;
const MACHINE_ID_BITS: u8 = 10;
const SEQUENCE_BITS: u8 = 12;

const MAX_MACHINE_ID: u16 = (1 << MACHINE_ID_BITS) - 1;
const MAX_SEQUENCE: u64 = (1 << SEQUENCE_BITS) - 1;

#[derive(Debug)]
pub struct SnowflakeGenerator {
    machine_id: u64,
    last_timestamp: AtomicU64,
    sequence: AtomicU64,
    spinlock: AtomicBool, // Spinlock to protect sequence
}

impl SnowflakeGenerator {
    pub fn new(machine_id: u16) -> Self {
        assert!(machine_id <= MAX_MACHINE_ID, "Machine ID out of range");

        Self {
            machine_id: machine_id as u64,
            last_timestamp: AtomicU64::new(0),
            sequence: AtomicU64::new(0),
            spinlock: AtomicBool::new(false), // Initially unlocked
        }
    }

    pub fn generate_id(&self) -> u64 {
        // Spinlock to safely access sequence
        #[allow(deprecated)] // TODO don't use deprecated method
        while self.spinlock.compare_and_swap(false, true, Ordering::Acquire) {
            // Wait (spin) until lock is released
        }

        let mut timestamp = current_millis();
        let last_timestamp = self.last_timestamp.load(Ordering::Relaxed);

        let mut sequence = self.sequence.load(Ordering::Relaxed);

        if timestamp == last_timestamp {
            // Same millisecond, increment sequence
            sequence = (sequence + 1) & MAX_SEQUENCE;

            // If sequence overflows, wait until next millisecond
            if sequence == 0 {
                while timestamp <= last_timestamp {
                    timestamp = current_millis();
                }
            }
        } else {
            // New millisecond, reset sequence
            sequence = 0;
        }

        // Update the sequence in a relaxed manner (since it's already synchronized with the lock)
        self.sequence.store(sequence, Ordering::Release);
        self.last_timestamp.store(timestamp, Ordering::Relaxed);

        // Release the spinlock
        self.spinlock.store(false, Ordering::Release);

        if timestamp > (1 << TIMESTAMP_BITS) - 1 {
            eprintln!("[WARN] {}: timestamp overflow", std::module_path!());
        }

        // Construct the Snowflake ID
        ((timestamp - EPOCH) << (MACHINE_ID_BITS + SEQUENCE_BITS))
            | (self.machine_id << SEQUENCE_BITS)
            | sequence
    }
}

fn current_millis() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .expect("Time went backwards")
        .as_millis() as u64
}

pub fn snowflake_generator(machine_id: u16) -> impl Fn() -> Nonce + Send + Sync + 'static {
    let generator = SnowflakeGenerator::new(machine_id);
    move || Nonce(generator.generate_id())
}