krabka_log/
stamp_source.rs

1//! The injected stamp source. It is the seam through which a partition's log
2//! gets the additional internal `TimestampSource` coordinate that it records
3//! in the `.stampindex` sidecar.
4//!
5//! The trait is deliberately tiny, and it lives in `krabka-log` rather than in
6//! the broker. The log owns both the append seam that folds stamps in and the
7//! `.stampindex` that they land in. The seam therefore stays here, so tests
8//! can check the log's own stamp behavior with a deterministic in-crate
9//! source, and the log needs no dependency edge onto a real clock. The broker
10//! injects the tenant source when internal stamping is enabled. A log with no
11//! injected source stamps nothing and behaves exactly as before.
12
13use std::fmt::Debug;
14
15/// A source of monotonic internal stamps folded into a partition's
16/// `.stampindex`.
17///
18/// The log calls [`StampSource::next_stamp`] for each
19/// non-transactional batch at append time and for each pure-Kafka transaction
20/// when its commit marker is durable. A cross-domain coordinator supplies its
21/// own commit stamp, which the log folds into this source with
22/// [`StampSource::observe`].
23///
24/// Implementations use interior mutability, because the log holds the source
25/// behind a shared `Arc` and stamps through `&mut self` appends. They must be
26/// cheap. They must never read or alter the wire-exact `.log` bytes, the
27/// offset assignment, the LSO, or the high-watermark. The stamp is an
28/// additional server-side coordinate only.
29pub trait StampSource: Debug + Send + Sync {
30    /// Return the next stamp.
31    fn next_stamp(&self) -> u64;
32
33    /// Fold an externally allocated commit stamp into this source so every
34    /// later allocation is greater than it.
35    ///
36    /// Centralized sources can keep the default no-op because they are the
37    /// sole timestamp authority. Distributed clocks override this with their
38    /// Lamport/HLC receive rule.
39    fn observe(&self, _stamp: u64) {}
40}
41
42#[cfg(any(test, feature = "test-helpers"))]
43mod test_sources {
44    use {
45        super::StampSource,
46        std::sync::atomic::{AtomicU64, Ordering},
47    };
48
49    /// A deterministic monotonic stamp source for tests.
50    ///
51    /// It yields `start`,
52    /// `start + step`, `start + 2*step`, and so on, on successive calls. The
53    /// sequence is fully predictable, so tests can assert the exact stamps
54    /// that a sequence of appends writes into the `.stampindex`.
55    #[derive(Debug)]
56    pub struct MonotonicStampSource {
57        next: AtomicU64,
58        step: u64,
59    }
60
61    impl MonotonicStampSource {
62        /// Create a source that yields `start` first and then advances by
63        /// `step` on each later call.
64        #[must_use]
65        pub fn new(start: u64, step: u64) -> Self {
66            Self {
67                next: AtomicU64::new(start),
68                step,
69            }
70        }
71    }
72
73    impl StampSource for MonotonicStampSource {
74        fn next_stamp(&self) -> u64 { self.next.fetch_add(self.step, Ordering::Relaxed) }
75
76        fn observe(&self, stamp: u64) {
77            let next = stamp.saturating_add(self.step);
78            self.next.fetch_max(next, Ordering::Relaxed);
79        }
80    }
81}
82
83#[cfg(any(test, feature = "test-helpers"))]
84pub use test_sources::MonotonicStampSource;
krabka_log/stamp_source.rs

krabka_log/
stamp_source.rs
