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
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
use parking_lot::Mutex;
use serde::{Deserialize, Serialize};
use std::collections::BTreeSet;
use std::sync::atomic::{AtomicU64, Ordering};
/// A monotonically increasing commit number. Every successful commit bumps the
/// epoch. Readers pin a [`Snapshot`] and only observe data with
/// `committed_epoch <= snapshot.epoch`. This is the value that tags every cache
/// entry and every WAL record, giving correctness-by-construction cache
/// invalidation.
#[derive(
Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize, Default,
)]
pub struct Epoch(pub u64);
impl Epoch {
pub const ZERO: Epoch = Epoch(0);
#[inline]
pub fn next(self) -> Epoch {
Epoch(self.0.wrapping_add(1))
}
}
/// A point-in-time read view.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Snapshot {
pub epoch: Epoch,
}
impl Snapshot {
#[inline]
pub fn at(epoch: Epoch) -> Self {
Self { epoch }
}
/// A cache page tagged with `page_epoch` is visible to this snapshot iff
/// the page was committed at or before the snapshot.
#[inline]
pub fn observes(&self, page_epoch: Epoch) -> bool {
page_epoch <= self.epoch
}
}
/// Atomic source of commit epochs. Shared by the writer and all readers.
#[derive(Debug, Default)]
pub struct EpochClock {
current: AtomicU64,
}
impl EpochClock {
pub fn new(start: u64) -> Self {
Self {
current: AtomicU64::new(start),
}
}
#[inline]
pub fn now(&self) -> Epoch {
Epoch(self.current.load(Ordering::Acquire))
}
#[inline]
pub fn snapshot(&self) -> Snapshot {
Snapshot::at(self.now())
}
/// Advance to the next epoch and return it. Called once per committed txn.
#[inline]
pub fn bump(&self) -> Epoch {
Epoch(self.current.fetch_add(1, Ordering::AcqRel) + 1)
}
/// Move the clock forward to at least `e` (used during recovery). Never
/// moves backward.
pub fn advance_to(&self, e: Epoch) {
loop {
let cur = self.current.load(Ordering::Acquire);
if e.0 <= cur {
return;
}
let _ = self
.current
.compare_exchange(cur, e.0, Ordering::AcqRel, Ordering::Acquire);
}
}
}
/// Dual-counter epoch authority for a multi-table `Database`. `assigned` is the
/// commit-order ticket (advanced the instant a txn is sequenced); `visible` is
/// the in-order reader watermark, advanced only once a committed txn has been
/// fully published. Readers pin `visible`; writers reserve `assigned`. The two
/// counters decouple "what order commits happened" from "what is safe to read".
#[derive(Debug)]
pub struct EpochAuthority {
assigned: AtomicU64,
visible: AtomicU64,
/// Epochs that have finished publishing but cannot yet be absorbed into the
/// `visible` watermark because an earlier assigned epoch is still in flight.
/// Shared across every commit path (cross-table transactions, single-table
/// `Table::commit`, and DDL) so the watermark only ever advances in assigned
/// order regardless of which path or thread completes first.
pending: Mutex<BTreeSet<u64>>,
}
impl EpochAuthority {
pub fn new(start: u64) -> Self {
Self {
assigned: AtomicU64::new(start),
visible: AtomicU64::new(start),
pending: Mutex::new(BTreeSet::new()),
}
}
/// The reader watermark: the highest epoch fully published and visible.
#[inline]
pub fn visible(&self) -> Epoch {
Epoch(self.visible.load(Ordering::Acquire))
}
/// Reserve the next commit-order ticket. Returns the assigned epoch.
#[inline]
pub fn bump_assigned(&self) -> Epoch {
Epoch(self.assigned.fetch_add(1, Ordering::AcqRel) + 1)
}
/// Advance the reader watermark to `e`, monotonically. Stale (lower) values
/// are ignored so out-of-order publish completions never regress visibility.
pub fn publish_visible(&self, e: Epoch) {
let mut cur = self.visible.load(Ordering::Acquire);
while e.0 > cur {
match self
.visible
.compare_exchange_weak(cur, e.0, Ordering::AcqRel, Ordering::Acquire)
{
Ok(_) => break,
Err(actual) => cur = actual,
}
}
}
/// Publish a fully-committed `e` and advance `visible` in assigned order:
/// `e` becomes visible only once it and every prior assigned epoch have
/// also been published (spec §9.3e). Because the pending set lives on the
/// shared authority, interleaved commits from different paths/threads can
/// never make the watermark jump past an epoch whose writes are not yet
/// applied. Each assigned epoch must call this exactly once.
pub fn publish_in_order(&self, e: Epoch) {
let mut pending = self.pending.lock();
pending.insert(e.0);
let mut vis = self.visible.load(Ordering::Acquire);
while pending.remove(&(vis + 1)) {
vis += 1;
}
// `vis` only ever moves forward here; `publish_visible` is monotonic.
drop(pending);
self.publish_visible(Epoch(vis));
}
/// Recovery: set both counters to `e` (e.g. the max committed epoch on open).
pub fn set_recovered(&self, e: Epoch) {
self.assigned.store(e.0, Ordering::Release);
self.visible.store(e.0, Ordering::Release);
}
/// Monotonically raise both counters to at least `e` (used while opening
/// tables that share one authority — each advances the shared clock to its
/// own manifest epoch; the max wins).
pub fn advance_recovered(&self, e: Epoch) {
raise_to(&self.assigned, e.0);
raise_to(&self.visible, e.0);
}
/// The current `assigned` counter (test/diagnostic use).
#[inline]
pub fn assigned(&self) -> Epoch {
Epoch(self.assigned.load(Ordering::Acquire))
}
}
fn raise_to(cell: &AtomicU64, target: u64) {
let mut cur = cell.load(Ordering::Acquire);
while target > cur {
match cell.compare_exchange_weak(cur, target, Ordering::AcqRel, Ordering::Acquire) {
Ok(_) => break,
Err(actual) => cur = actual,
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn snapshot_visibility_is_monotonic() {
let clock = EpochClock::new(10);
assert_eq!(clock.now(), Epoch(10));
let s = clock.snapshot();
assert!(s.observes(Epoch(10)));
assert!(!s.observes(Epoch(11)));
let next = clock.bump();
assert_eq!(next, Epoch(11));
assert!(clock.snapshot().observes(Epoch(11)));
}
#[test]
fn epoch_authority_assigned_and_visible_advance_in_order() {
let a = EpochAuthority::new(0);
assert_eq!(a.visible(), Epoch(0));
let e1 = a.bump_assigned();
let e2 = a.bump_assigned();
assert_eq!((e1, e2), (Epoch(1), Epoch(2)));
a.publish_visible(Epoch(2));
assert_eq!(a.visible(), Epoch(2));
a.publish_visible(Epoch(1));
assert_eq!(a.visible(), Epoch(2));
}
#[test]
fn publish_in_order_gates_until_gap_filled() {
let a = EpochAuthority::new(0);
let e1 = a.bump_assigned();
let e2 = a.bump_assigned();
let e3 = a.bump_assigned();
assert_eq!((e1, e2, e3), (Epoch(1), Epoch(2), Epoch(3)));
// A later epoch finishing first must NOT advance the watermark past the
// still-in-flight earlier epochs.
a.publish_in_order(e3);
assert_eq!(a.visible(), Epoch(0), "e3 cannot be visible before e1/e2");
a.publish_in_order(e2);
assert_eq!(a.visible(), Epoch(0), "e2 still gated on e1");
// Filling the gap drains everything consecutively in one shot.
a.publish_in_order(e1);
assert_eq!(a.visible(), Epoch(3));
}
}