zeph_durable/
retention.rs

1// SPDX-FileCopyrightText: 2026 Andrei G <bug-ops>
2// SPDX-License-Identifier: MIT OR Apache-2.0
3
4//! Journal retention, the background prune sweep, and the checkpoint-fold codec.
5//!
6//! Two mechanisms bound journal growth, and neither runs on the step-dispatch hot path (spec NEVER):
7//!
8//! - **Background prune** — [`DurableRetentionService`] is a tokio task that wakes every
9//!   `prune_interval_secs` and calls [`Journal::prune`](crate::Journal::prune), which deletes
10//!   *terminal* executions older than their TTL in `prune_batch_size` batches, yielding between
11//!   batches so a large sweep never holds the write lock.
12//! - **In-execution checkpoint fold** — a long *in-flight* execution that crosses the soft step cap
13//!   (90% of `max_steps_per_execution`) folds its committed-idempotent prefix into a single
14//!   [`Checkpoint`](crate::EntryKind::Checkpoint) entry. The fold packs each folded step's replay
15//!   value into the checkpoint snapshot and deletes the individual rows, so a resume still replays
16//!   those steps from the snapshot rather than re-running them. The hard cap (100%) aborts the
17//!   execution with [`DurableError::StepCapExceeded`].
18
19use std::sync::Arc;
20use std::time::Duration;
21
22use bytes::Bytes;
23use tracing::Instrument as _;
24
25use crate::backend::DurableBackendEnum;
26use crate::config::RetentionPolicy;
27use crate::error::DurableError;
28use crate::journal::Journal as _;
29
30/// Wire-format version for the checkpoint snapshot encoding.
31const CHECKPOINT_FORMAT_V1: u8 = 1;
32
33/// One step's replay value, folded into a checkpoint snapshot.
34///
35/// The fold preserves exactly what a resume needs to replay the step without re-running its
36/// operation: the [`IdempotencyKey`](crate::IdempotencyKey) bytes (so the replay-divergence guard
37/// still matches, INV-3), the payload wire-format version, and the *plaintext* result bytes (the
38/// snapshot as a whole is AEAD-sealed by the backend, so individual step payloads need no further
39/// sealing inside it).
40#[derive(Debug, Clone, PartialEq, Eq)]
41pub(crate) struct FoldedStep {
42    /// Position of the folded step within its execution.
43    pub(crate) step_id: u32,
44    /// The step's 32-byte idempotency key, for the divergence guard on replay.
45    pub(crate) idem_key: [u8; 32],
46    /// Wire-format version of the result payload.
47    pub(crate) payload_version: u8,
48    /// Plaintext result bytes.
49    pub(crate) payload: Bytes,
50}
51
52/// A decoded checkpoint snapshot: the folded prefix of an execution, in step order.
53pub(crate) type CheckpointSnapshot = Vec<FoldedStep>;
54
55/// Per-step fixed framing overhead in the encoded snapshot (step + version + `idem_key` + len).
56const FOLDED_STEP_OVERHEAD: usize = 4 + 1 + 32 + 4;
57
58/// Encoded size one [`FoldedStep`] contributes (framing + payload).
59pub(crate) fn folded_step_encoded_len(payload_len: usize) -> usize {
60    FOLDED_STEP_OVERHEAD.saturating_add(payload_len)
61}
62
63/// Serialize a checkpoint snapshot into a compact, self-describing byte buffer.
64///
65/// Layout: `version(1) || count(u32 le) || [ step(u32 le) version(1) idem_key(32) len(u32 le)
66/// payload(len) ]*`. Fixed-width framing keeps the encoding injective and the per-step size
67/// predictable, so the backend can cut the fold at the payload ceiling without trial encoding.
68pub(crate) fn encode_checkpoint(steps: &[FoldedStep]) -> Vec<u8> {
69    let total: usize = steps
70        .iter()
71        .map(|s| folded_step_encoded_len(s.payload.len()))
72        .sum();
73    let mut out = Vec::with_capacity(5 + total);
74    out.push(CHECKPOINT_FORMAT_V1);
75    out.extend_from_slice(&u32::try_from(steps.len()).unwrap_or(u32::MAX).to_le_bytes());
76    for step in steps {
77        out.extend_from_slice(&step.step_id.to_le_bytes());
78        out.push(step.payload_version);
79        out.extend_from_slice(&step.idem_key);
80        out.extend_from_slice(
81            &u32::try_from(step.payload.len())
82                .unwrap_or(u32::MAX)
83                .to_le_bytes(),
84        );
85        out.extend_from_slice(&step.payload);
86    }
87    out
88}
89
90/// Decode a checkpoint snapshot, failing closed on truncation or an unknown format version.
91///
92/// # Errors
93///
94/// Returns [`DurableError::Decode`] if the buffer is truncated, declares more steps than it
95/// contains, or carries an unrecognized format version.
96pub(crate) fn decode_checkpoint(bytes: &[u8]) -> Result<CheckpointSnapshot, DurableError> {
97    let mut cursor = Reader::new(bytes);
98    let version = cursor.u8()?;
99    if version != CHECKPOINT_FORMAT_V1 {
100        return Err(DurableError::Decode {
101            context: "checkpoint snapshot has an unknown format version",
102        });
103    }
104    let count = cursor.u32()? as usize;
105    let mut steps = Vec::with_capacity(count.min(1024));
106    for _ in 0..count {
107        let step_id = cursor.u32()?;
108        let payload_version = cursor.u8()?;
109        let idem_key = cursor.array32()?;
110        let len = cursor.u32()? as usize;
111        let payload = Bytes::copy_from_slice(cursor.take(len)?);
112        steps.push(FoldedStep {
113            step_id,
114            idem_key,
115            payload_version,
116            payload,
117        });
118    }
119    Ok(steps)
120}
121
122/// A bounds-checked forward reader over the snapshot buffer; every read fails closed on underrun.
123struct Reader<'a> {
124    bytes: &'a [u8],
125    pos: usize,
126}
127
128impl<'a> Reader<'a> {
129    fn new(bytes: &'a [u8]) -> Self {
130        Self { bytes, pos: 0 }
131    }
132
133    fn take(&mut self, len: usize) -> Result<&'a [u8], DurableError> {
134        let end = self.pos.checked_add(len).ok_or(DurableError::Decode {
135            context: "checkpoint snapshot length overflow",
136        })?;
137        let slice = self.bytes.get(self.pos..end).ok_or(DurableError::Decode {
138            context: "checkpoint snapshot is truncated",
139        })?;
140        self.pos = end;
141        Ok(slice)
142    }
143
144    fn u8(&mut self) -> Result<u8, DurableError> {
145        Ok(self.take(1)?[0])
146    }
147
148    fn u32(&mut self) -> Result<u32, DurableError> {
149        let bytes = self.take(4)?;
150        Ok(u32::from_le_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]))
151    }
152
153    fn array32(&mut self) -> Result<[u8; 32], DurableError> {
154        let mut out = [0u8; 32];
155        out.copy_from_slice(self.take(32)?);
156        Ok(out)
157    }
158}
159
160/// Compute the soft and hard step-cap thresholds for a `max_steps_per_execution` budget.
161///
162/// The soft threshold (90%) triggers a checkpoint fold; the hard threshold (the cap itself) aborts.
163/// A `max` of zero disables both (returns `(u32::MAX, u32::MAX)`), so an unconfigured cap never folds
164/// or aborts.
165#[must_use]
166pub(crate) fn step_cap_thresholds(max: u32) -> (u32, u32) {
167    if max == 0 {
168        return (u32::MAX, u32::MAX);
169    }
170    // `max * 9 / 10 <= max`, so the result always fits back into u32; the widening guards the
171    // intermediate product against overflow.
172    let soft = u32::try_from(u64::from(max) * 9 / 10).unwrap_or(max);
173    (soft, max)
174}
175
176/// Background task that prunes terminal executions on a fixed interval.
177///
178/// Spawn [`DurableRetentionService::run`] on a supervised task (alongside the
179/// [`JournalWriter`](crate::JournalWriter)). It owns no write path of its own — it calls
180/// [`Journal::prune`](crate::Journal::prune) on the shared backend, which performs the batched delete
181/// off the hot path.
182#[derive(Debug)]
183pub struct DurableRetentionService {
184    backend: Arc<DurableBackendEnum>,
185    policy: RetentionPolicy,
186    interval: Duration,
187}
188
189impl DurableRetentionService {
190    /// Build the service from the shared backend and the configured retention policy.
191    ///
192    /// # Examples
193    ///
194    /// ```no_run
195    /// # async fn run() -> Result<(), Box<dyn std::error::Error>> {
196    /// use std::sync::Arc;
197    /// use zeph_durable::{DurableBackendEnum, DurableRetentionService, LocalBackend, RetentionPolicy};
198    ///
199    /// let backend = Arc::new(DurableBackendEnum::Local(Arc::new(
200    ///     LocalBackend::open("durable.db", 1_048_576).await?,
201    /// )));
202    /// let service = DurableRetentionService::new(backend, RetentionPolicy::default());
203    /// let task = tokio::spawn(service.run());
204    /// # let _ = task;
205    /// # Ok(()) }
206    /// ```
207    #[must_use]
208    pub fn new(backend: Arc<DurableBackendEnum>, policy: RetentionPolicy) -> Self {
209        let interval = Duration::from_secs(policy.prune_interval_secs.max(1));
210        Self {
211            backend,
212            policy,
213            interval,
214        }
215    }
216
217    /// Run the prune loop until the task is aborted.
218    ///
219    /// Each tick prunes terminal executions older than their TTL; a prune failure is logged and the
220    /// loop continues (a transient database error must not kill retention).
221    #[tracing::instrument(name = "durable.retention.run", skip_all)]
222    pub async fn run(self) {
223        let mut tick = tokio::time::interval(self.interval);
224        tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
225        // The first immediate tick fires at startup; skip pruning on it so a just-launched daemon
226        // does not sweep before the first real interval elapses.
227        tick.tick().await;
228        loop {
229            tick.tick().await;
230            async {
231                match self.backend.prune(&self.policy).await {
232                    Ok(deleted) => {
233                        tracing::debug!(deleted, "durable retention prune sweep completed");
234                    }
235                    Err(error) => {
236                        tracing::warn!(%error, "durable retention prune sweep failed; will retry");
237                    }
238                }
239            }
240            .instrument(tracing::info_span!("durable.retention.run.iter"))
241            .await;
242        }
243    }
244}
245
246/// Run one batched prune pass over the journal, deleting terminal executions past their TTL.
247///
248/// This is the shared body behind [`Journal::prune`](crate::Journal::prune) for the local backend.
249/// It is a free function (rather than a method) so the backend can keep its prune implementation thin
250/// while the batching/yielding policy lives next to the rest of retention. `delete_batch` performs
251/// one bounded `DELETE` transaction and returns the rows it removed; the loop yields between batches
252/// so a large sweep never monopolizes the runtime.
253pub(crate) async fn prune_in_batches<F, Fut>(
254    policy: &RetentionPolicy,
255    now_ms: i64,
256    delete_batch: F,
257) -> Result<u64, DurableError>
258where
259    F: Fn(PruneCutoffs, u64) -> Fut,
260    Fut: Future<Output = Result<u64, DurableError>>,
261{
262    let cutoffs = PruneCutoffs::from_policy(policy, now_ms);
263    let batch = policy.prune_batch_size.max(1);
264    let mut total = 0u64;
265    let span = tracing::info_span!(
266        "durable.journal.prune",
267        deleted_count = tracing::field::Empty
268    );
269    async {
270        loop {
271            let deleted = delete_batch(cutoffs, batch).await?;
272            total = total.saturating_add(deleted);
273            if deleted < batch {
274                break;
275            }
276            // Release the write lock and let other tasks run before the next batch.
277            tokio::task::yield_now().await;
278        }
279        tracing::Span::current().record("deleted_count", total);
280        Ok(total)
281    }
282    .instrument(span)
283    .await
284}
285
286/// The absolute `finalized_at` cutoffs (Unix ms) below which a terminal execution is prunable.
287#[derive(Debug, Clone, Copy)]
288pub(crate) struct PruneCutoffs {
289    /// Completed executions finalized at or before this instant are prunable.
290    pub(crate) completed_before_ms: i64,
291    /// Failed/aborted executions finalized at or before this instant are prunable.
292    pub(crate) failed_before_ms: i64,
293}
294
295impl PruneCutoffs {
296    pub(crate) fn from_policy(policy: &RetentionPolicy, now_ms: i64) -> Self {
297        let completed =
298            i64::try_from(policy.ttl_completed_secs.saturating_mul(1000)).unwrap_or(i64::MAX);
299        let failed = i64::try_from(policy.ttl_failed_secs.saturating_mul(1000)).unwrap_or(i64::MAX);
300        Self {
301            completed_before_ms: now_ms.saturating_sub(completed),
302            failed_before_ms: now_ms.saturating_sub(failed),
303        }
304    }
305}
306
307/// The largest payload the backend will pack into a single checkpoint snapshot.
308///
309/// The fold cuts its prefix at this ceiling so a checkpoint entry obeys the same `max_payload_bytes`
310/// read/write guard as any other payload (INV-11). Steps that do not fit stay un-folded until a later
311/// checkpoint.
312#[must_use]
313pub(crate) fn checkpoint_budget(max_payload_bytes: u64) -> usize {
314    usize::try_from(max_payload_bytes).unwrap_or(usize::MAX)
315}
316
317/// Decide how many leading folded steps fit within the checkpoint payload budget.
318///
319/// Returns the count of steps from the front of `payload_lens` whose cumulative encoded size stays
320/// within `budget` (including the 5-byte snapshot header). A single step larger than the whole budget
321/// yields `0`, leaving it un-folded rather than producing an over-limit checkpoint.
322#[must_use]
323pub(crate) fn fold_prefix_len(payload_lens: &[usize], budget: usize) -> usize {
324    let mut used = 5usize; // version + count header
325    let mut taken = 0usize;
326    for &len in payload_lens {
327        let next = used.saturating_add(folded_step_encoded_len(len));
328        if next > budget {
329            break;
330        }
331        used = next;
332        taken += 1;
333    }
334    taken
335}
336
337#[cfg(test)]
338mod tests {
339    use super::*;
340    use std::assert_matches;
341
342    fn folded(step: u32, payload: &[u8]) -> FoldedStep {
343        FoldedStep {
344            step_id: step,
345            idem_key: [u8::try_from(step % 256).unwrap_or(0); 32],
346            payload_version: 1,
347            payload: Bytes::copy_from_slice(payload),
348        }
349    }
350
351    #[test]
352    fn checkpoint_round_trips() {
353        let steps = vec![
354            folded(0, b"alpha"),
355            folded(1, b""),
356            folded(2, b"gamma-payload"),
357        ];
358        let encoded = encode_checkpoint(&steps);
359        let decoded = decode_checkpoint(&encoded).unwrap();
360        assert_eq!(decoded, steps);
361    }
362
363    #[test]
364    fn decode_rejects_truncation() {
365        let steps = vec![folded(0, b"data")];
366        let mut encoded = encode_checkpoint(&steps);
367        encoded.truncate(encoded.len() - 2);
368        assert_matches!(
369            decode_checkpoint(&encoded),
370            Err(DurableError::Decode { .. })
371        );
372    }
373
374    #[test]
375    fn decode_rejects_unknown_version() {
376        let mut encoded = encode_checkpoint(&[folded(0, b"x")]);
377        encoded[0] = 99;
378        assert_matches!(
379            decode_checkpoint(&encoded),
380            Err(DurableError::Decode { .. })
381        );
382    }
383
384    #[test]
385    fn step_cap_thresholds_are_ninety_percent_and_full() {
386        assert_eq!(step_cap_thresholds(10_000), (9_000, 10_000));
387        assert_eq!(step_cap_thresholds(10), (9, 10));
388        assert_eq!(step_cap_thresholds(0), (u32::MAX, u32::MAX));
389    }
390
391    #[test]
392    fn fold_prefix_respects_budget() {
393        // Each step encodes to 41 + payload; with a 4-byte payload that is 45 bytes + the 5-byte
394        // header. A budget of 5 + 45 + 45 = 95 admits exactly two steps.
395        let lens = vec![4, 4, 4, 4];
396        assert_eq!(fold_prefix_len(&lens, 95), 2);
397        // A step larger than the entire budget is left un-folded.
398        assert_eq!(fold_prefix_len(&[10_000], 50), 0);
399    }
400
401    #[test]
402    fn prune_cutoffs_subtract_ttl_from_now() {
403        let policy = RetentionPolicy {
404            ttl_completed_secs: 10,
405            ttl_failed_secs: 20,
406            ..RetentionPolicy::default()
407        };
408        let cutoffs = PruneCutoffs::from_policy(&policy, 100_000);
409        assert_eq!(cutoffs.completed_before_ms, 90_000);
410        assert_eq!(cutoffs.failed_before_ms, 80_000);
411        assert_eq!(checkpoint_budget(1_048_576), 1_048_576);
412    }
413
414    #[tokio::test]
415    async fn prune_in_batches_loops_until_drained_and_yields() {
416        use std::cell::Cell;
417        // Three full batches (500) then a short one (120) → four calls, totalling 1620.
418        let remaining = Cell::new(1_620u64);
419        let policy = RetentionPolicy::default();
420        let total = prune_in_batches(&policy, 0, |_cutoffs, batch| {
421            let deleted = remaining.get().min(batch);
422            remaining.set(remaining.get() - deleted);
423            async move { Ok(deleted) }
424        })
425        .await
426        .unwrap();
427        assert_eq!(total, 1_620);
428        assert_eq!(remaining.get(), 0);
429    }
430}
zeph_durable/retention.rs

zeph_durable/
retention.rs
