mongreldb_core/
retention.rs

1//! Global snapshot-retention registry for a multi-table `Database`.
2//!
3//! Readers register the epoch they pin via [`SnapshotRegistry::register`] and
4//! the returned [`SnapshotGuard`] deregisters on drop. Garbage collection of
5//! superseded runs, dropped tables, and recycled WAL segments is gated on
6//! [`SnapshotRegistry::min_active`]: nothing whose retire epoch is still
7//! observable by an open reader may be physically deleted.
8
9use crate::epoch::Epoch;
10use parking_lot::Mutex;
11use std::collections::BTreeMap;
12use std::collections::HashSet;
13use std::sync::atomic::{AtomicU64, Ordering};
14use std::sync::Arc;
15use std::time::Instant;
16
17/// Set of transaction ids that are currently spilling into `_txn/<txn_id>/`
18/// (spec §8.5, review fix #14). A large transaction registers its id before
19/// writing its pending run and holds the [`SpillGuard`] through publish; GC
20/// consults [`ActiveSpills::is_active`] and never deletes a live txn's pending
21/// dir (deleting it would lose the spill run / fail the commit).
22#[derive(Default)]
23pub struct ActiveSpills {
24    inner: Mutex<HashSet<u64>>,
25}
26
27impl ActiveSpills {
28    pub fn new() -> Self {
29        Self {
30            inner: Mutex::new(HashSet::new()),
31        }
32    }
33
34    /// Register `txn_id` as actively spilling. The id stays protected from GC
35    /// until the returned guard is dropped.
36    pub fn register(self: &Arc<Self>, txn_id: u64) -> SpillGuard {
37        self.inner.lock().insert(txn_id);
38        SpillGuard {
39            registry: Arc::clone(self),
40            txn_id,
41        }
42    }
43
44    /// Whether `txn_id`'s pending `_txn/` dir is currently in use.
45    pub fn is_active(&self, txn_id: u64) -> bool {
46        self.inner.lock().contains(&txn_id)
47    }
48
49    /// Whether no transaction is currently spilling. Used to gate WAL-segment GC.
50    pub fn is_idle(&self) -> bool {
51        self.inner.lock().is_empty()
52    }
53
54    fn release(&self, txn_id: u64) {
55        self.inner.lock().remove(&txn_id);
56    }
57}
58
59/// RAII handle that deregisters its txn id from [`ActiveSpills`] on drop.
60pub struct SpillGuard {
61    registry: Arc<ActiveSpills>,
62    txn_id: u64,
63}
64
65impl Drop for SpillGuard {
66    fn drop(&mut self) {
67        self.registry.release(self.txn_id);
68    }
69}
70
71/// Refcounted multiset of pinned reader epochs. Tracks the lowest live snapshot
72/// so the reaper can decide what is safe to reclaim.
73#[derive(Default)]
74pub struct SnapshotRegistry {
75    /// `epoch -> count` of currently-pinned reader snapshots.
76    live: Mutex<BTreeMap<u64, u64>>,
77    /// Number of prior commit epochs compaction must keep queryable. Zero
78    /// preserves the current-state-only behavior.
79    history_epochs: AtomicU64,
80    /// Earliest epoch known to have been protected since history was enabled.
81    history_start: AtomicU64,
82}
83
84impl SnapshotRegistry {
85    pub fn new() -> Self {
86        Self {
87            live: Mutex::new(BTreeMap::new()),
88            history_epochs: AtomicU64::new(0),
89            history_start: AtomicU64::new(0),
90        }
91    }
92
93    pub fn configure_history(&self, epochs: u64, start_epoch: Epoch) {
94        self.history_start.store(start_epoch.0, Ordering::Release);
95        self.history_epochs.store(epochs, Ordering::Release);
96    }
97
98    pub fn history_config(&self) -> (u64, Epoch) {
99        (
100            self.history_epochs.load(Ordering::Acquire),
101            Epoch(self.history_start.load(Ordering::Acquire)),
102        )
103    }
104
105    /// Earliest epoch guaranteed available by the rolling history window.
106    /// Returns `None` when historical retention is disabled.
107    pub fn history_floor(&self, visible: Epoch) -> Option<Epoch> {
108        let (epochs, start) = self.history_config();
109        (epochs > 0).then(|| Epoch(start.0.max(visible.0.saturating_sub(epochs))))
110    }
111
112    /// Register a pinned reader at `epoch`. The snapshot stays retained until
113    /// the returned guard is dropped.
114    pub fn register(&self, epoch: Epoch) -> SnapshotGuard<'_> {
115        let mut live = self.live.lock();
116        *live.entry(epoch.0).or_insert(0) += 1;
117        SnapshotGuard {
118            registry: self,
119            epoch,
120        }
121    }
122
123    /// The lowest currently-live pinned epoch. If no reader is active, returns
124    /// `visible` (nothing older than the reader watermark is retained, so GC is
125    /// free to reclaim anything strictly below it).
126    pub fn min_active(&self, visible: Epoch) -> Epoch {
127        match self.live.lock().keys().next().copied() {
128            Some(min) => Epoch(min),
129            None => visible,
130        }
131    }
132
133    /// The lowest currently-pinned epoch, or `None` when no reader is active.
134    /// Unlike [`Self::min_active`] this distinguishes "no readers" from "a reader
135    /// pinned at `visible`", which compaction needs: with no readers it may drop
136    /// superseded versions/tombstones freely, but a pin (even at `visible`) must
137    /// preserve the version that reader can still see.
138    pub fn min_pinned(&self) -> Option<Epoch> {
139        self.live.lock().keys().next().copied().map(Epoch)
140    }
141
142    fn release(&self, epoch: Epoch) {
143        let mut live = self.live.lock();
144        if let Some(count) = live.get_mut(&epoch.0) {
145            *count -= 1;
146            if *count == 0 {
147                live.remove(&epoch.0);
148            }
149        }
150    }
151}
152
153/// RAII handle that deregisters its epoch from the registry on drop.
154pub struct SnapshotGuard<'r> {
155    registry: &'r SnapshotRegistry,
156    epoch: Epoch,
157}
158
159impl Drop for SnapshotGuard<'_> {
160    fn drop(&mut self) {
161        self.registry.release(self.epoch);
162    }
163}
164
165/// An owned, shareable guard (across threads / `Arc`) for snapshots that must
166/// outlive a borrow of the registry, e.g. inside an `Arc<Database>`.
167pub struct OwnedSnapshotGuard {
168    registry: Arc<SnapshotRegistry>,
169    epoch: Epoch,
170}
171
172impl OwnedSnapshotGuard {
173    /// The epoch this guard pins.
174    pub fn epoch(&self) -> Epoch {
175        self.epoch
176    }
177}
178
179impl Drop for OwnedSnapshotGuard {
180    fn drop(&mut self) {
181        self.registry.release(self.epoch);
182    }
183}
184
185impl SnapshotRegistry {
186    /// Register a pinned reader and return an owned (clonable-handle) guard
187    /// that does not borrow the registry.
188    pub fn register_owned(self: &Arc<Self>, epoch: Epoch) -> OwnedSnapshotGuard {
189        {
190            let mut live = self.live.lock();
191            *live.entry(epoch.0).or_insert(0) += 1;
192        }
193        OwnedSnapshotGuard {
194            registry: Arc::clone(self),
195            epoch,
196        }
197    }
198}
199
200/// The version-retention pin sources of spec §10.3 (S1C-004). A version may be
201/// reclaimed only when it is older than the oldest pin of **every** source.
202#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
203pub enum PinSource {
204    /// Oldest active transaction/read snapshot (MVCC readers). Projected from
205    /// [`SnapshotRegistry`] and the table-local pin set in diagnostics.
206    TransactionSnapshot,
207    /// Configured rolling history-retention window. Projected from
208    /// [`SnapshotRegistry::history_floor`] in diagnostics.
209    HistoryRetention,
210    /// Oldest backup / point-in-time-recovery pin.
211    BackupPitr,
212    /// Oldest replication-follower requirement.
213    Replication,
214    /// Oldest cursor / immutable read-generation pin.
215    ReadGeneration,
216    /// Oldest online index build still reading historical versions.
217    OnlineIndexBuild,
218}
219
220impl PinSource {
221    /// Every source, in stable declaration order (diagnostics iteration).
222    pub const ALL: [PinSource; 6] = [
223        PinSource::TransactionSnapshot,
224        PinSource::HistoryRetention,
225        PinSource::BackupPitr,
226        PinSource::Replication,
227        PinSource::ReadGeneration,
228        PinSource::OnlineIndexBuild,
229    ];
230
231    /// Stable lowercase label for logs and diagnostics output.
232    pub fn label(self) -> &'static str {
233        match self {
234            PinSource::TransactionSnapshot => "transaction_snapshot",
235            PinSource::HistoryRetention => "history_retention",
236            PinSource::BackupPitr => "backup_pitr",
237            PinSource::Replication => "replication",
238            PinSource::ReadGeneration => "read_generation",
239            PinSource::OnlineIndexBuild => "online_index_build",
240        }
241    }
242}
243
244/// Diagnostics view of one active pin source (S1C-004): the oldest epoch it
245/// holds, when the oldest of its pins was taken, and how many pins are live.
246/// `held_since` is `None` for projected sources ([`PinSource::TransactionSnapshot`]
247/// and [`PinSource::HistoryRetention`]) whose epochs come from the
248/// [`SnapshotRegistry`] rather than from registered [`PinGuard`]s.
249#[derive(Debug, Clone)]
250pub struct PinInfo {
251    pub source: PinSource,
252    pub oldest_epoch: Epoch,
253    pub held_since: Option<Instant>,
254    pub pin_count: usize,
255}
256
257/// Every currently-active pin source, one entry per source (S1C-004
258/// diagnostics). Empty when nothing pins version reclamation.
259#[derive(Debug, Clone, Default)]
260pub struct PinsReport {
261    pub pins: Vec<PinInfo>,
262}
263
264impl PinsReport {
265    /// The oldest epoch held by any source — the version-reclamation floor.
266    pub fn oldest_epoch(&self) -> Option<Epoch> {
267        self.pins.iter().map(|pin| pin.oldest_epoch).min()
268    }
269
270    /// The entry for `source`, if that source currently holds a pin.
271    pub fn get(&self, source: PinSource) -> Option<&PinInfo> {
272        self.pins.iter().find(|pin| pin.source == source)
273    }
274
275    pub fn is_empty(&self) -> bool {
276        self.pins.is_empty()
277    }
278
279    pub fn len(&self) -> usize {
280        self.pins.len()
281    }
282
283    /// Merge a projected epoch for `source` (a floor derived from another
284    /// retention mechanism rather than a registered guard). The reported
285    /// oldest epoch only moves down; `held_since`/`pin_count` keep describing
286    /// the registered guards, if any.
287    pub fn record_projection(&mut self, source: PinSource, epoch: Epoch) {
288        match self.pins.iter_mut().find(|pin| pin.source == source) {
289            Some(info) => info.oldest_epoch = info.oldest_epoch.min(epoch),
290            None => self.pins.push(PinInfo {
291                source,
292                oldest_epoch: epoch,
293                held_since: None,
294                pin_count: 0,
295            }),
296        }
297    }
298}
299
300struct PinEntry {
301    source: PinSource,
302    epoch: Epoch,
303    held_since: Instant,
304}
305
306/// Unified registry of version-retention pins (S1C-004).
307///
308/// Every subsystem that needs historical versions to survive reclamation —
309/// backup/PITR, replication, cursors/read generations, online index builds —
310/// registers a guard here via [`PinRegistry::pin`]. GC consults
311/// [`PinRegistry::oldest_pinned`] in addition to the [`SnapshotRegistry`]
312/// (transaction snapshots) and the configured history window, and
313/// [`PinRegistry::report`] exposes every active source for diagnostics.
314/// Guards are cheap (one map insertion) and deregister on drop.
315#[derive(Default)]
316pub struct PinRegistry {
317    /// `pin id -> entry`; ids are monotonic so equal epochs stay distinguishable.
318    pins: Mutex<BTreeMap<u64, PinEntry>>,
319    next_id: AtomicU64,
320}
321
322impl PinRegistry {
323    pub fn new() -> Self {
324        Self {
325            pins: Mutex::new(BTreeMap::new()),
326            next_id: AtomicU64::new(1),
327        }
328    }
329
330    /// Register a pin of `source` at `epoch`. Versions at or below `epoch`
331    /// stay retained until the returned guard (and every clone of it) drops.
332    pub fn pin(self: &Arc<Self>, source: PinSource, epoch: Epoch) -> PinGuard {
333        let id = self.next_id.fetch_add(1, Ordering::Relaxed);
334        self.pins.lock().insert(
335            id,
336            PinEntry {
337                source,
338                epoch,
339                held_since: Instant::now(),
340            },
341        );
342        PinGuard {
343            registry: Arc::clone(self),
344            id,
345            source,
346            epoch,
347        }
348    }
349
350    /// The oldest epoch held by any registered pin, or `None` when no pin is
351    /// active (GC is then gated only by snapshots/history).
352    pub fn oldest_pinned(&self) -> Option<Epoch> {
353        self.pins.lock().values().map(|entry| entry.epoch).min()
354    }
355
356    /// The oldest epoch held by pins of `source`.
357    pub fn oldest_for(&self, source: PinSource) -> Option<Epoch> {
358        self.pins
359            .lock()
360            .values()
361            .filter(|entry| entry.source == source)
362            .map(|entry| entry.epoch)
363            .min()
364    }
365
366    /// Diagnostics: one entry per source with at least one live pin.
367    pub fn report(&self) -> PinsReport {
368        let pins = self.pins.lock();
369        let mut by_source: BTreeMap<PinSource, PinInfo> = BTreeMap::new();
370        for entry in pins.values() {
371            by_source
372                .entry(entry.source)
373                .and_modify(|info| {
374                    info.oldest_epoch = info.oldest_epoch.min(entry.epoch);
375                    info.held_since = match info.held_since {
376                        Some(since) => Some(since.min(entry.held_since)),
377                        None => Some(entry.held_since),
378                    };
379                    info.pin_count += 1;
380                })
381                .or_insert(PinInfo {
382                    source: entry.source,
383                    oldest_epoch: entry.epoch,
384                    held_since: Some(entry.held_since),
385                    pin_count: 1,
386                });
387        }
388        PinsReport {
389            pins: by_source.into_values().collect(),
390        }
391    }
392
393    fn release(&self, id: u64) {
394        self.pins.lock().remove(&id);
395    }
396}
397
398/// RAII handle that deregisters its pin from the [`PinRegistry`] on drop.
399/// Not [`Clone`]: share it behind an `Arc` when several owners must keep the
400/// same pin alive (the pin releases when the last `Arc` drops).
401pub struct PinGuard {
402    registry: Arc<PinRegistry>,
403    id: u64,
404    source: PinSource,
405    epoch: Epoch,
406}
407
408impl PinGuard {
409    pub fn source(&self) -> PinSource {
410        self.source
411    }
412
413    pub fn epoch(&self) -> Epoch {
414        self.epoch
415    }
416}
417
418impl Drop for PinGuard {
419    fn drop(&mut self) {
420        self.registry.release(self.id);
421    }
422}
423
424#[cfg(test)]
425mod tests {
426    use super::*;
427
428    #[test]
429    fn retention_tracks_min_active_snapshot() {
430        let r = SnapshotRegistry::new();
431        assert_eq!(r.min_active(Epoch(10)), Epoch(10));
432        let g1 = r.register(Epoch(5));
433        let g2 = r.register(Epoch(8));
434        assert_eq!(r.min_active(Epoch(10)), Epoch(5));
435        drop(g1);
436        assert_eq!(r.min_active(Epoch(10)), Epoch(8));
437        drop(g2);
438        assert_eq!(r.min_active(Epoch(10)), Epoch(10));
439    }
440
441    #[test]
442    fn retention_refcounts_duplicate_epochs() {
443        let r = SnapshotRegistry::new();
444        let a = r.register(Epoch(3));
445        let b = r.register(Epoch(3));
446        assert_eq!(r.min_active(Epoch(9)), Epoch(3));
447        drop(a);
448        assert_eq!(r.min_active(Epoch(9)), Epoch(3));
449        drop(b);
450        assert_eq!(r.min_active(Epoch(9)), Epoch(9));
451    }
452
453    #[test]
454    fn pin_registry_tracks_oldest_epoch_per_source() {
455        let registry = Arc::new(PinRegistry::new());
456        assert_eq!(registry.oldest_pinned(), None);
457        let backup = registry.pin(PinSource::BackupPitr, Epoch(7));
458        let replication = registry.pin(PinSource::Replication, Epoch(4));
459        assert_eq!(registry.oldest_pinned(), Some(Epoch(4)));
460        assert_eq!(registry.oldest_for(PinSource::BackupPitr), Some(Epoch(7)));
461        assert_eq!(registry.oldest_for(PinSource::ReadGeneration), None);
462        drop(replication);
463        assert_eq!(registry.oldest_pinned(), Some(Epoch(7)));
464        drop(backup);
465        assert_eq!(registry.oldest_pinned(), None);
466    }
467
468    #[test]
469    fn pin_registry_report_lists_every_active_source_once() {
470        let registry = Arc::new(PinRegistry::new());
471        let mut guards = Vec::new();
472        for (offset, source) in PinSource::ALL.into_iter().enumerate() {
473            guards.push(registry.pin(source, Epoch(offset as u64 + 2)));
474        }
475        // A second, newer pin of one source must not duplicate its entry.
476        guards.push(registry.pin(PinSource::BackupPitr, Epoch(50)));
477
478        let report = registry.report();
479        assert_eq!(report.len(), PinSource::ALL.len());
480        for (offset, source) in PinSource::ALL.into_iter().enumerate() {
481            let info = report.get(source).expect("source listed");
482            assert_eq!(info.oldest_epoch, Epoch(offset as u64 + 2));
483            assert!(
484                info.held_since.is_some(),
485                "registered pins carry a timestamp"
486            );
487        }
488        assert_eq!(report.get(PinSource::BackupPitr).unwrap().pin_count, 2);
489        assert_eq!(report.oldest_epoch(), Some(Epoch(2)));
490
491        drop(guards);
492        assert!(registry.report().is_empty());
493    }
494
495    #[test]
496    fn pins_report_projection_only_lowers_the_floor() {
497        let registry = Arc::new(PinRegistry::new());
498        let guard = registry.pin(PinSource::ReadGeneration, Epoch(9));
499        let mut report = registry.report();
500        // A projection older than the registered pin lowers the source floor;
501        // a newer one is ignored. Projections carry no timestamp or count.
502        report.record_projection(PinSource::ReadGeneration, Epoch(4));
503        report.record_projection(PinSource::ReadGeneration, Epoch(12));
504        report.record_projection(PinSource::TransactionSnapshot, Epoch(6));
505        let info = report.get(PinSource::ReadGeneration).unwrap();
506        assert_eq!(info.oldest_epoch, Epoch(4));
507        assert_eq!(info.pin_count, 1, "projection is not a registered guard");
508        assert!(info.held_since.is_some());
509        let projected = report.get(PinSource::TransactionSnapshot).unwrap();
510        assert_eq!(projected.oldest_epoch, Epoch(6));
511        assert_eq!(projected.pin_count, 0);
512        assert!(projected.held_since.is_none());
513        assert_eq!(report.oldest_epoch(), Some(Epoch(4)));
514        drop(guard);
515    }
516}
mongreldb_core/retention.rs

mongreldb_core/
retention.rs
