eventsdb_core/
position.rs

1//! Coordinates: where an event sits in its stream, and in the database.
2//!
3//! There are two monotonic `u64`s in this system and they have different
4//! scopes. `seq` counts within one stream; [`Position`] counts across every
5//! stream of one database. Passing one where the other belongs produces a
6//! read of the wrong range — a subscription that silently skips or repeats —
7//! rather than a failure anyone would notice, so `Position` is a newtype and
8//! `seq` stays a bare `u64` it cannot be confused with.
9
10use serde_json::{Map, Value};
11
12use crate::upcast::Current;
13
14/// A global coordinate: the order an event was committed in, across every
15/// stream of one database.
16///
17/// Positions are dense and gap-free *as read*: a reader never observes
18/// `n + 1` while `n` is still uncommitted, so a subscription is a plain
19/// `position > cursor` range read with no grace window.
20///
21/// The mechanism is narrower than "one writer", and worth stating precisely
22/// because the narrower version is also stronger. The position is allocated
23/// **inside the transaction that commits it**, and that transaction holds
24/// SQLite's write lock from `BEGIN`, because every write the backend makes is
25/// `IMMEDIATE`. Allocation order and commit order therefore cannot diverge —
26/// and that argument does not depend on there being one connection. It holds
27/// for two processes on one file, and it is measured: 120 appends through two
28/// separately-opened logs come back as exactly `1..=120`, in order.
29///
30/// What a second connection *does* cost is the wake-up, not the order — see
31/// [`crate::log::EventLog::subscribe`].
32#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Default)]
33pub struct Position(u64);
34
35impl Position {
36    /// Before the first event. Where a consumer with no checkpoint starts.
37    pub const BEGINNING: Position = Position(0);
38
39    pub const fn new(value: u64) -> Self {
40        Position(value)
41    }
42
43    pub const fn get(self) -> u64 {
44        self.0
45    }
46
47    /// The value as SQLite stores it, or `None` when it does not fit.
48    ///
49    /// A position is a rowid, so every position the store ever assigns is in
50    /// range. [`Position::new`] is public, though, and a `u64` above
51    /// `i64::MAX` would bind as a negative number: `position > -1` reads the
52    /// whole log rather than nothing, and a checkpoint written from one would
53    /// silently replay everything through the exactly-once path. Callers that
54    /// bind a position use this and refuse rather than wrap.
55    pub fn as_stored(self) -> Option<i64> {
56        i64::try_from(self.0).ok()
57    }
58}
59
60impl std::fmt::Display for Position {
61    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
62        write!(f, "{}", self.0)
63    }
64}
65
66/// What a write returned: the coordinates the store assigned it.
67///
68/// A caller never reads back to learn where its event landed, and never
69/// supplies these fields — they are the store's to give.
70#[derive(Debug, Clone, Copy, PartialEq, Eq)]
71pub struct Committed {
72    /// Per-stream sequence, from 1.
73    pub seq: u64,
74    /// Wall clock at append time.
75    pub epoch_ms: u64,
76    /// Global coordinate, or `None` from a backend that keeps one stream and
77    /// therefore has no database-wide order to place the event in.
78    pub position: Option<Position>,
79}
80
81/// An event read back from the log, with where it sits.
82///
83/// The `event` has already been through the upcaster chain, so a reader sees
84/// the current shape whatever version the bytes were written under.
85#[derive(Debug, Clone)]
86pub struct Recorded {
87    pub position: Position,
88    pub stream: String,
89    pub event: Current,
90}
91
92impl Recorded {
93    /// The event's `kind`, as it reads after upcasting.
94    pub fn kind(&self) -> &str {
95        self.event.kind()
96    }
97
98    /// The per-stream sequence.
99    pub fn seq(&self) -> u64 {
100        self.event.seq()
101    }
102
103    /// The underlying object.
104    pub fn into_inner(self) -> Map<String, Value> {
105        self.event.into_inner()
106    }
107}
eventsdb_core/position.rs

eventsdb_core/
position.rs
