krabka_log/
delivery.rs

1//! Deliver-at-time visibility: when a scheduled batch becomes readable.
2//!
3//! On a topic set to [`DeliveryPolicy::Scheduled`](crate::DeliveryPolicy), a
4//! record batch is not visible the moment it is durable. It becomes visible
5//! once its activation time has passed. The activation time is the batch's
6//! `max_timestamp`, the v2 header field, so a producer schedules delivery
7//! through the timestamp it stamps and no sidecar is needed.
8//!
9//! The schedule lives entirely in the records, which is why nothing here is
10//! persisted. A watermark is derived state: after a restart or a leader
11//! change it is recomputed from the segments, and a recomputation can only
12//! agree with what the records say.
13
14use {crate::DeliveryPolicy, krabka_ids::Offset};
15
16/// Result of
17/// [`Log::advance_delivery_watermark`](crate::Log::advance_delivery_watermark).
18#[derive(Debug, Clone, Copy, PartialEq, Eq)]
19pub struct DeliveryAdvance {
20    /// First offset that is not visible yet. Every offset below it has
21    /// reached its activation time, so a fetch may serve it.
22    ///
23    /// On a topic that does not schedule delivery this is the log end offset:
24    /// everything written is visible.
25    pub watermark: Offset,
26
27    /// Epoch-millisecond instant at which the first waiting batch above
28    /// [`Self::watermark`] becomes visible, or `None` when nothing is waiting.
29    ///
30    /// The value already includes the configured clock-uncertainty bound, so
31    /// a scheduler compares it against its own clock and adds nothing. A
32    /// caller that wakes at this instant and advances the watermark again
33    /// finds that batch visible.
34    pub next_deadline_ms: Option<i64>,
35}
36
37/// The instant a batch that activates at `activation_ms` becomes visible.
38///
39/// The bound is added, never subtracted: if the clock reads `c` while true
40/// time lies in `[c - e, c + e]`, then `c >= activation + e` proves true time
41/// has reached the activation instant, so delivery is never early.
42pub fn visible_at_ms(activation_ms: i64, uncertainty_ms: i64) -> i64 {
43    activation_ms.saturating_add(uncertainty_ms)
44}
45
46/// Apply the shared scheduled-delivery visibility rule to one batch.
47#[must_use]
48pub fn batch_is_deliverable(
49    policy: DeliveryPolicy,
50    uncertainty_ms: i64,
51    activation_ms: i64,
52    now_ms: i64,
53) -> bool {
54    krabka_verified::scheduled_delivery_visible(
55        policy == DeliveryPolicy::Scheduled,
56        uncertainty_ms,
57        activation_ms,
58        now_ms,
59    )
60}
61
62/// Merge inclusive offset ranges that touch or overlap.
63///
64/// The walk produces them in ascending order, one per batch, so a single pass
65/// is enough. Adjacent batches merge because `[0, 4]` and `[5, 9]` describe
66/// one uninterrupted run of waiting records.
67pub fn coalesce_ranges(ranges: Vec<(Offset, Offset)>) -> Vec<(Offset, Offset)> {
68    let mut out: Vec<(Offset, Offset)> = Vec::with_capacity(ranges.len());
69    for (low, high) in ranges {
70        match out.last_mut() {
71            Some(last) => {
72                let (merge, merged_high) =
73                    krabka_verified::coalesce_delivery_range(last.1.0, low.0, high.0);
74                if merge {
75                    last.1 = Offset(merged_high);
76                } else {
77                    out.push((low, high));
78                }
79            }
80            None => out.push((low, high)),
81        }
82    }
83    out
84}
85
86#[cfg(test)]
87mod tests {
88    use {super::*, assert2::check};
89
90    #[test]
91    fn adjacent_and_overlapping_ranges_merge_but_separated_ones_do_not() {
92        let merged = coalesce_ranges(vec![
93            (Offset(0), Offset(4)),
94            // Touches the previous range.
95            (Offset(5), Offset(9)),
96            // Contained in the run so far: the end must not move backwards.
97            (Offset(6), Offset(7)),
98            // One offset of daylight: a separate range.
99            (Offset(11), Offset(12)),
100        ]);
101        check!(merged == vec![(Offset(0), Offset(9)), (Offset(11), Offset(12))]);
102        check!(
103            coalesce_ranges(vec![
104                (Offset(i64::MAX - 1), Offset(i64::MAX)),
105                (Offset(i64::MAX), Offset(i64::MAX)),
106            ]) == vec![(Offset(i64::MAX - 1), Offset(i64::MAX))]
107        );
108    }
109
110    #[test]
111    fn no_ranges_merge_to_nothing() {
112        check!(coalesce_ranges(Vec::new()).is_empty());
113    }
114
115    #[test]
116    fn the_visible_instant_adds_the_bound_and_saturates() {
117        check!(visible_at_ms(1_000, 250) == 1_250);
118        check!(visible_at_ms(i64::MAX, 250) == i64::MAX);
119    }
120
121    #[test]
122    fn batch_visibility_delegates_to_the_proved_deadline_rule() {
123        check!(batch_is_deliverable(
124            crate::DeliveryPolicy::Immediate,
125            250,
126            10_000,
127            0
128        ));
129        check!(!batch_is_deliverable(
130            crate::DeliveryPolicy::Scheduled,
131            250,
132            10_000,
133            10_249
134        ));
135        check!(batch_is_deliverable(
136            crate::DeliveryPolicy::Scheduled,
137            250,
138            10_000,
139            10_250
140        ));
141    }
142}
krabka_log/delivery.rs

krabka_log/
delivery.rs
