laminar_db/checkpoint_timing/
mod.rs

1//! Bounded process-local evidence for cluster checkpoint barrier pauses.
2
3use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
4use std::sync::Arc;
5use std::time::{Duration, Instant};
6
7use laminar_core::checkpoint::CheckpointAttempt;
8use laminar_core::cluster::control::LocalProcessAuthorityIdentity;
9
10/// Process-lifetime barrier observations retained before incremental collection must run.
11pub const CHECKPOINT_BARRIER_TIMING_CAPACITY: usize = 1_024;
12/// Maximum records copied by one bounded snapshot.
13pub const MAX_CHECKPOINT_BARRIER_TIMING_PAGE_RECORDS: usize = 64;
14
15/// Local role performed for one checkpoint barrier.
16#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize)]
17#[serde(rename_all = "snake_case")]
18pub enum CheckpointBarrierRole {
19    /// This process coordinated the attempt.
20    Leader,
21    /// This process followed another process's announcement.
22    Follower,
23}
24
25/// One exact process-local observation corresponding to one pipeline-stall histogram sample.
26#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize)]
27pub struct CheckpointBarrierTimingRecord {
28    /// Monotonic process-local sequence. Zero is never emitted.
29    pub sequence: u64,
30    /// Exact local process generation that performed the barrier.
31    pub process: LocalProcessAuthorityIdentity,
32    /// Canonical runtime checkpoint attempt.
33    pub attempt: CheckpointAttempt,
34    /// Local role in this attempt.
35    pub role: CheckpointBarrierRole,
36    /// Exact assignment version admitted for this attempt.
37    pub assignment_version: u64,
38    /// Digest of the complete admitted assignment certificate.
39    pub assignment_digest: [u8; 32],
40    /// Complete pipeline-pause observation in nanoseconds.
41    pub pipeline_stall_ns: u64,
42    /// Local barrier work in nanoseconds.
43    pub local_barrier_ns: u64,
44    /// Aligned-resume wait, when a cluster shuffle ran that stage.
45    pub aligned_resume_ns: Option<u64>,
46    /// Whether capture produced a durable-tail handoff before the local stage closed.
47    pub durable_tail_handoff: bool,
48    /// Whether the attempt's absolute deadline was exhausted when the pause closed.
49    pub deadline_exhausted: bool,
50}
51
52#[derive(Debug, Clone, Copy)]
53pub(crate) struct CheckpointBarrierTimingObservation {
54    pub process: LocalProcessAuthorityIdentity,
55    pub attempt: CheckpointAttempt,
56    pub role: CheckpointBarrierRole,
57    pub assignment_version: u64,
58    pub assignment_digest: [u8; 32],
59    pub pipeline_stall_ns: u64,
60    pub local_barrier_ns: u64,
61    pub aligned_resume_ns: Option<u64>,
62    pub durable_tail_handoff: bool,
63    pub deadline_exhausted: bool,
64}
65
66/// Canonical non-timing dimensions captured before one barrier pause.
67#[derive(Debug, Clone, Copy)]
68pub(crate) struct CheckpointBarrierTimingContext {
69    pub process: LocalProcessAuthorityIdentity,
70    pub attempt: CheckpointAttempt,
71    pub role: CheckpointBarrierRole,
72    pub assignment_version: u64,
73    pub assignment_digest: [u8; 32],
74}
75
76impl CheckpointBarrierTimingObservation {
77    fn is_canonical(self) -> bool {
78        self.process.is_canonical()
79            && self.attempt.is_canonical()
80            && self.assignment_version != 0
81            && self.assignment_digest != [0; 32]
82            && self.local_barrier_ns <= self.pipeline_stall_ns
83            && self
84                .aligned_resume_ns
85                .is_none_or(|duration| duration <= self.pipeline_stall_ns)
86            && self
87                .local_barrier_ns
88                .checked_add(self.aligned_resume_ns.unwrap_or(0))
89                .is_some_and(|duration| duration <= self.pipeline_stall_ns)
90            && (self.durable_tail_handoff || self.aligned_resume_ns.is_none())
91    }
92
93    fn with_sequence(self, sequence: u64) -> CheckpointBarrierTimingRecord {
94        CheckpointBarrierTimingRecord {
95            sequence,
96            process: self.process,
97            attempt: self.attempt,
98            role: self.role,
99            assignment_version: self.assignment_version,
100            assignment_digest: self.assignment_digest,
101            pipeline_stall_ns: self.pipeline_stall_ns,
102            local_barrier_ns: self.local_barrier_ns,
103            aligned_resume_ns: self.aligned_resume_ns,
104            durable_tail_handoff: self.durable_tail_handoff,
105            deadline_exhausted: self.deadline_exhausted,
106        }
107    }
108}
109
110struct CheckpointBarrierTimingState {
111    slots: Box<[Option<CheckpointBarrierTimingRecord>]>,
112    first: usize,
113    len: usize,
114    next_sequence: u64,
115    overwritten_record_count: u64,
116    process: Option<LocalProcessAuthorityIdentity>,
117}
118
119/// Fixed-capacity, process-lifetime checkpoint barrier timing ledger.
120///
121/// Writers never wait: contention, invalid metadata, and counter exhaustion increment explicit
122/// loss state. The only allocation occurs during construction and bounded reader snapshots.
123pub(crate) struct CheckpointBarrierTimingLedger {
124    state: parking_lot::Mutex<CheckpointBarrierTimingState>,
125    recording_loss_count: AtomicU64,
126    metadata_exhausted: AtomicBool,
127}
128
129/// Drop-observing timer that feeds the existing histograms and one exact ledger record.
130pub(crate) struct CheckpointBarrierTimingGuard {
131    stall_histogram: prometheus::Histogram,
132    local_histogram: prometheus::Histogram,
133    aligned_histogram: prometheus::Histogram,
134    ledger: Arc<CheckpointBarrierTimingLedger>,
135    context: Option<CheckpointBarrierTimingContext>,
136    stall_started: Instant,
137    local_started: Instant,
138    local_barrier_duration: Option<Duration>,
139    aligned_started: Option<Instant>,
140    aligned_resume_duration: Option<Duration>,
141    durable_tail_handoff: bool,
142    absolute_deadline: tokio::time::Instant,
143    invalid_duration_state: bool,
144}
145
146impl CheckpointBarrierTimingGuard {
147    pub(crate) fn start_with_context(
148        context: impl FnOnce() -> Option<CheckpointBarrierTimingContext>,
149        metrics: &crate::engine_metrics::EngineMetrics,
150        ledger: &Arc<CheckpointBarrierTimingLedger>,
151        absolute_deadline: tokio::time::Instant,
152    ) -> Self {
153        let started = Instant::now();
154        let context = context();
155        Self {
156            stall_histogram: metrics.checkpoint_pipeline_stall_duration.clone(),
157            local_histogram: metrics.checkpoint_barrier_local_duration.clone(),
158            aligned_histogram: metrics.checkpoint_aligned_resume_wait.clone(),
159            ledger: Arc::clone(ledger),
160            context,
161            stall_started: started,
162            local_started: started,
163            local_barrier_duration: None,
164            aligned_started: None,
165            aligned_resume_duration: None,
166            durable_tail_handoff: false,
167            absolute_deadline,
168            invalid_duration_state: false,
169        }
170    }
171
172    pub(crate) fn finish_local_barrier_with_handoff(&mut self) {
173        self.durable_tail_handoff = true;
174        self.finish_local_barrier();
175    }
176
177    pub(crate) fn begin_aligned_resume(&mut self) {
178        if self.aligned_started.is_some() || self.aligned_resume_duration.is_some() {
179            self.invalid_duration_state = true;
180            return;
181        }
182        self.aligned_started = Some(Instant::now());
183    }
184
185    pub(crate) fn finish_aligned_resume(&mut self) {
186        let Some(started) = self.aligned_started.take() else {
187            self.invalid_duration_state = true;
188            return;
189        };
190        self.aligned_resume_duration = Some(started.elapsed());
191    }
192
193    fn finish_local_barrier(&mut self) {
194        if self.local_barrier_duration.is_some() {
195            return;
196        }
197        self.local_barrier_duration = Some(self.local_started.elapsed());
198    }
199}
200
201impl Drop for CheckpointBarrierTimingGuard {
202    fn drop(&mut self) {
203        self.finish_local_barrier();
204        if self.aligned_started.is_some() {
205            self.finish_aligned_resume();
206        }
207        let stall = self.stall_started.elapsed();
208        let local_barrier_duration = self
209            .local_barrier_duration
210            .expect("local barrier duration is closed before observation");
211
212        // Publish every duration only when the complete barrier scope closes. In particular, a
213        // process killed after the durable-tail handoff must not leave a local/aligned sample
214        // without the corresponding stall sample and exact ledger record.
215        self.local_histogram
216            .observe(local_barrier_duration.as_secs_f64());
217        if let Some(aligned_resume_duration) = self.aligned_resume_duration {
218            self.aligned_histogram
219                .observe(aligned_resume_duration.as_secs_f64());
220        }
221        self.stall_histogram.observe(stall.as_secs_f64());
222        let Some(pipeline_stall_ns) = duration_ns(stall) else {
223            self.ledger.note_recording_loss();
224            return;
225        };
226        let Some(local_barrier_ns) = duration_ns(local_barrier_duration) else {
227            self.ledger.note_recording_loss();
228            return;
229        };
230        let aligned_resume_ns = if let Some(duration) = self.aligned_resume_duration {
231            let Some(duration_ns) = duration_ns(duration) else {
232                self.ledger.note_recording_loss();
233                return;
234            };
235            Some(duration_ns)
236        } else {
237            None
238        };
239        if self.invalid_duration_state {
240            self.ledger.note_recording_loss();
241            return;
242        }
243        let Some(context) = self.context else {
244            self.ledger.note_recording_loss();
245            return;
246        };
247        self.ledger.try_record(CheckpointBarrierTimingObservation {
248            process: context.process,
249            attempt: context.attempt,
250            role: context.role,
251            assignment_version: context.assignment_version,
252            assignment_digest: context.assignment_digest,
253            pipeline_stall_ns,
254            local_barrier_ns,
255            aligned_resume_ns,
256            durable_tail_handoff: self.durable_tail_handoff,
257            deadline_exhausted: tokio::time::Instant::now() >= self.absolute_deadline,
258        });
259    }
260}
261
262fn duration_ns(duration: Duration) -> Option<u64> {
263    u64::try_from(duration.as_nanos()).ok()
264}
265
266impl CheckpointBarrierTimingLedger {
267    pub(crate) fn new() -> Self {
268        Self::with_capacity(CHECKPOINT_BARRIER_TIMING_CAPACITY)
269    }
270
271    fn with_capacity(capacity: usize) -> Self {
272        assert!(capacity > 0, "checkpoint timing capacity must be nonzero");
273        Self {
274            state: parking_lot::Mutex::new(CheckpointBarrierTimingState {
275                slots: vec![None; capacity].into_boxed_slice(),
276                first: 0,
277                len: 0,
278                next_sequence: 1,
279                overwritten_record_count: 0,
280                process: None,
281            }),
282            recording_loss_count: AtomicU64::new(0),
283            metadata_exhausted: AtomicBool::new(false),
284        }
285    }
286
287    /// Attempt one O(1), allocation-free record write without waiting for a reader or writer.
288    pub(crate) fn try_record(&self, observation: CheckpointBarrierTimingObservation) -> bool {
289        if !observation.is_canonical() {
290            self.note_recording_loss();
291            return false;
292        }
293        let Some(mut state) = self.state.try_lock() else {
294            self.note_recording_loss();
295            return false;
296        };
297        if state.next_sequence == u64::MAX {
298            self.metadata_exhausted.store(true, Ordering::Release);
299            drop(state);
300            self.note_recording_loss();
301            return false;
302        }
303        if state.len == state.slots.len() && state.overwritten_record_count == u64::MAX {
304            self.metadata_exhausted.store(true, Ordering::Release);
305            drop(state);
306            self.note_recording_loss();
307            return false;
308        }
309        if state
310            .process
311            .is_some_and(|process| process != observation.process)
312        {
313            drop(state);
314            self.note_recording_loss();
315            return false;
316        }
317
318        let sequence = state.next_sequence;
319        state.next_sequence += 1;
320        state.process.get_or_insert(observation.process);
321        let index = if state.len < state.slots.len() {
322            let index = (state.first + state.len) % state.slots.len();
323            state.len += 1;
324            index
325        } else {
326            let index = state.first;
327            state.first = (state.first + 1) % state.slots.len();
328            state.overwritten_record_count += 1;
329            index
330        };
331        state.slots[index] = Some(observation.with_sequence(sequence));
332        true
333    }
334
335    pub(crate) fn note_recording_loss(&self) {
336        if self
337            .recording_loss_count
338            .fetch_update(Ordering::AcqRel, Ordering::Acquire, |current| {
339                current.checked_add(1)
340            })
341            .is_err()
342        {
343            self.metadata_exhausted.store(true, Ordering::Release);
344        }
345    }
346
347    /// Copy one bounded, exclusive-cursor page without performing I/O.
348    pub(crate) fn snapshot_after(
349        &self,
350        after_sequence: u64,
351        limit: usize,
352    ) -> Result<CheckpointBarrierTimingSnapshot, CheckpointBarrierTimingSnapshotError> {
353        if !(1..=MAX_CHECKPOINT_BARRIER_TIMING_PAGE_RECORDS).contains(&limit) {
354            return Err(CheckpointBarrierTimingSnapshotError::InvalidLimit { limit });
355        }
356        let mut records = Vec::with_capacity(limit);
357        let Some(state) = self.state.try_lock() else {
358            return Err(CheckpointBarrierTimingSnapshotError::Busy);
359        };
360        if after_sequence >= state.next_sequence {
361            return Err(CheckpointBarrierTimingSnapshotError::CursorAhead {
362                after_sequence,
363                next_sequence: state.next_sequence,
364            });
365        }
366
367        let oldest_retained_sequence = (state.len != 0).then(|| {
368            state.slots[state.first]
369                .expect("a retained timing slot must be populated")
370                .sequence
371        });
372        if oldest_retained_sequence.is_some_and(|oldest| after_sequence.saturating_add(1) < oldest)
373        {
374            return Err(CheckpointBarrierTimingSnapshotError::CursorOverwritten {
375                after_sequence,
376                oldest_retained_sequence: oldest_retained_sequence.unwrap_or(1),
377            });
378        }
379
380        let logical_record = |offset: usize| {
381            let index = (state.first + offset) % state.slots.len();
382            state.slots[index].expect("a retained timing slot must be populated")
383        };
384        let mut low = 0;
385        let mut high = state.len;
386        while low < high {
387            let middle = low + (high - low) / 2;
388            if logical_record(middle).sequence <= after_sequence {
389                low = middle + 1;
390            } else {
391                high = middle;
392            }
393        }
394        let end = low.saturating_add(limit).min(state.len);
395        for offset in low..end {
396            records.push(logical_record(offset));
397        }
398        let snapshot = CheckpointBarrierTimingSnapshot {
399            process: state.process,
400            capacity: state.slots.len(),
401            oldest_retained_sequence,
402            next_sequence: state.next_sequence,
403            overwritten_record_count: state.overwritten_record_count,
404            recording_loss_count: self.recording_loss_count.load(Ordering::Acquire),
405            metadata_exhausted: self.metadata_exhausted.load(Ordering::Acquire),
406            has_more: end < state.len,
407            records,
408        };
409        drop(state);
410        Ok(snapshot)
411    }
412}
413
414/// One bounded in-memory ledger page.
415#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
416pub struct CheckpointBarrierTimingSnapshot {
417    /// Process identity that owns this sequence domain, or `None` before the first record.
418    pub process: Option<LocalProcessAuthorityIdentity>,
419    /// Fixed process-lifetime record capacity.
420    pub capacity: usize,
421    /// Oldest successful record still retained, or `None` before the first record.
422    pub oldest_retained_sequence: Option<u64>,
423    /// Next successful sequence; the last accepted sequence is one less.
424    pub next_sequence: u64,
425    /// Successful records evicted by fixed-capacity overwrite.
426    pub overwritten_record_count: u64,
427    /// Observations lost to contention, invalid metadata/durations, or exhausted counters.
428    pub recording_loss_count: u64,
429    /// Sequence or loss metadata can no longer advance without ambiguity.
430    pub metadata_exhausted: bool,
431    /// More retained records follow this page.
432    pub has_more: bool,
433    /// Ordered records whose sequence is strictly greater than the requested cursor.
434    pub records: Vec<CheckpointBarrierTimingRecord>,
435}
436
437/// Failure to take one trustworthy bounded ledger page.
438#[derive(Debug, thiserror::Error, PartialEq, Eq)]
439pub enum CheckpointBarrierTimingSnapshotError {
440    /// The caller supplied a zero or oversized page.
441    #[error("checkpoint timing page limit {limit} is outside the supported range")]
442    InvalidLimit {
443        /// Requested record count.
444        limit: usize,
445    },
446    /// The nonblocking reader found another ledger operation in progress.
447    #[error("checkpoint timing ledger is busy")]
448    Busy,
449    /// Fixed retention already evicted records after the supplied cursor.
450    #[error(
451        "checkpoint timing cursor {after_sequence} precedes oldest retained sequence {oldest_retained_sequence}"
452    )]
453    CursorOverwritten {
454        /// Caller-supplied exclusive cursor.
455        after_sequence: u64,
456        /// First record still retained.
457        oldest_retained_sequence: u64,
458    },
459    /// The supplied cursor is outside the current numeric sequence range.
460    #[error(
461        "checkpoint timing cursor {after_sequence} is at or beyond next sequence {next_sequence}"
462    )]
463    CursorAhead {
464        /// Caller-supplied exclusive cursor.
465        after_sequence: u64,
466        /// Next sequence this process may issue.
467        next_sequence: u64,
468    },
469}
470
471/// One process-bound, bounded in-memory ledger page.
472#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
473pub struct CheckpointBarrierTimingPage {
474    /// Live local process identity sampled around the ledger read.
475    pub process: LocalProcessAuthorityIdentity,
476    /// Bounded page from that process's sequence domain.
477    pub snapshot: CheckpointBarrierTimingSnapshot,
478}
479
480/// Failure to read a process-bound checkpoint timing page.
481#[derive(Debug, thiserror::Error, PartialEq, Eq)]
482pub enum CheckpointBarrierTimingReadError {
483    /// A nonzero sequence cursor omitted its process-generation identity.
484    #[error("checkpoint timing continuation cursor requires a process identity")]
485    ProcessIdentityRequired,
486    /// A live local process identity could not be sampled without waiting.
487    #[error("checkpoint timing process identity is unavailable")]
488    ProcessIdentityUnavailable,
489    /// The supplied cursor belongs to a different process generation.
490    #[error("checkpoint timing cursor process identity does not match the live process")]
491    ProcessIdentityMismatch {
492        /// Process identity supplied with the cursor.
493        expected: LocalProcessAuthorityIdentity,
494        /// Live process identity observed before the read.
495        actual: LocalProcessAuthorityIdentity,
496    },
497    /// Process authority changed while the in-memory ledger was read.
498    #[error("checkpoint timing process identity changed during the read")]
499    ProcessIdentityChanged {
500        /// Identity observed before the ledger read.
501        before: LocalProcessAuthorityIdentity,
502        /// Identity observed after the ledger read.
503        after: LocalProcessAuthorityIdentity,
504    },
505    /// The ledger's immutable sequence domain contradicts current process authority.
506    #[error("checkpoint timing ledger belongs to a different process identity")]
507    LedgerProcessMismatch {
508        /// Live process identity sampled around the read.
509        expected: LocalProcessAuthorityIdentity,
510        /// Process identity bound to the ledger's sequence domain.
511        actual: LocalProcessAuthorityIdentity,
512    },
513    /// The bounded ledger snapshot itself failed.
514    #[error(transparent)]
515    Snapshot(#[from] CheckpointBarrierTimingSnapshotError),
516}
517
518#[cfg(test)]
519mod tests;
laminar_db/checkpoint_timing/mod.rs

laminar_db/checkpoint_timing/
mod.rs
