nautilus_event_store/
retention.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! Non-destructive retention planning for sealed event-store run files.
17
18use std::{
19    collections::BTreeSet,
20    path::{Path, PathBuf},
21};
22
23use nautilus_system::event_store::RetentionMode;
24
25use crate::{EventStore, EventStoreError, RedbBackend, RunManifest, RunStatus, SnapshotAnchor};
26
27/// A retention decision for one sealed run file.
28#[derive(Clone, Debug, PartialEq, Eq)]
29pub struct RetentionRun {
30    /// The manifest stored in the run file.
31    pub manifest: RunManifest,
32    /// The on-disk run file path.
33    pub path: PathBuf,
34    /// The latest snapshot anchor state observed for the run.
35    pub snapshot_anchor: SnapshotAnchorStatus,
36}
37
38impl RetentionRun {
39    /// Creates a retention planning record for a sealed run file.
40    #[must_use]
41    pub fn new(
42        manifest: RunManifest,
43        path: impl Into<PathBuf>,
44        snapshot_anchor: SnapshotAnchorStatus,
45    ) -> Self {
46        Self {
47            manifest,
48            path: path.into(),
49            snapshot_anchor,
50        }
51    }
52
53    /// Returns the run id recorded in the manifest.
54    #[must_use]
55    pub fn run_id(&self) -> &str {
56        self.manifest.run_id.as_str()
57    }
58
59    /// Returns whether this run can serve as a conservative restore point.
60    #[must_use]
61    pub fn is_known_good_restore_point(&self) -> bool {
62        !matches!(
63            self.manifest.status,
64            RunStatus::Running | RunStatus::Quarantined
65        ) && matches!(&self.snapshot_anchor, SnapshotAnchorStatus::Valid(_))
66    }
67}
68
69/// Snapshot-anchor state used by the retention planner.
70#[derive(Clone, Debug, PartialEq, Eq)]
71pub enum SnapshotAnchorStatus {
72    /// The run has no recorded snapshot anchor.
73    Missing,
74    /// The run has a snapshot anchor that matches the sealed manifest.
75    Valid(SnapshotAnchor),
76    /// The run has an anchor, but retention must not rely on it.
77    Invalid(String),
78}
79
80/// A non-destructive retention plan.
81#[derive(Clone, Debug, Default, PartialEq, Eq)]
82pub struct RetentionPlan {
83    /// Sealed runs visible to the planner, sorted by manifest start time.
84    pub sealed_runs: Vec<RetentionRun>,
85    /// Whole run files the selected policy may reclaim later.
86    pub reclaim_candidates: Vec<RetentionRun>,
87}
88
89/// Lists sealed redb run files and computes the non-destructive retention plan.
90///
91/// # Errors
92///
93/// Returns [`EventStoreError`] when the run directory cannot be listed, a manifest cannot
94/// be decoded, or a sealed run file cannot be opened to inspect its snapshot anchor.
95pub fn plan_redb_retention(
96    base_dir: &Path,
97    instance_id: &str,
98    mode: RetentionMode,
99) -> Result<RetentionPlan, EventStoreError> {
100    Ok(plan_retention(
101        list_redb_sealed_runs(base_dir, instance_id)?,
102        mode,
103    ))
104}
105
106/// Lists sealed redb run files with their latest snapshot-anchor status.
107///
108/// `Running` manifests are excluded so a later destructive phase cannot reclaim a live or
109/// crash-recovery-domain run by accident.
110///
111/// # Errors
112///
113/// Returns [`EventStoreError`] when manifest listing or sealed-run anchor inspection fails.
114pub fn list_redb_sealed_runs(
115    base_dir: &Path,
116    instance_id: &str,
117) -> Result<Vec<RetentionRun>, EventStoreError> {
118    let manifests = RedbBackend::list_runs(base_dir, instance_id)?;
119    let mut runs = Vec::new();
120
121    for manifest in manifests {
122        if !manifest.is_sealed() {
123            continue;
124        }
125
126        let path = base_dir
127            .join(instance_id)
128            .join(format!("{}.redb", manifest.run_id));
129        let reader = RedbBackend::open_sealed(base_dir, instance_id, manifest.run_id.as_str())?;
130        let durable_high_watermark = reader.high_watermark()?;
131        let snapshot_anchor = match reader.latest_snapshot_anchor() {
132            Ok(anchor) => snapshot_anchor_status(durable_high_watermark, anchor),
133            Err(EventStoreError::Corrupted(msg)) => SnapshotAnchorStatus::Invalid(msg),
134            Err(e) => return Err(e),
135        };
136
137        runs.push(RetentionRun::new(manifest, path, snapshot_anchor));
138    }
139
140    Ok(runs)
141}
142
143/// Computes reclaim candidates from sealed runs without deleting anything.
144#[must_use]
145pub fn plan_retention(mut sealed_runs: Vec<RetentionRun>, mode: RetentionMode) -> RetentionPlan {
146    sealed_runs.retain(|run| run.manifest.is_sealed());
147    // Match `list_runs` ordering: break start-time ties on the run id so the keep
148    // window and restore-point selection stay deterministic.
149    sealed_runs.sort_by(|a, b| {
150        a.manifest
151            .start_ts_init
152            .cmp(&b.manifest.start_ts_init)
153            .then_with(|| a.manifest.run_id.cmp(&b.manifest.run_id))
154    });
155
156    let reclaim_candidates = match mode {
157        RetentionMode::Full => Vec::new(),
158        RetentionMode::Bounded { keep_last } => bounded_reclaim_candidates(&sealed_runs, keep_last),
159        RetentionMode::SnapshotAnchored => snapshot_anchored_reclaim_candidates(&sealed_runs),
160    };
161
162    RetentionPlan {
163        sealed_runs,
164        reclaim_candidates,
165    }
166}
167
168fn bounded_reclaim_candidates(sealed_runs: &[RetentionRun], keep_last: usize) -> Vec<RetentionRun> {
169    let Some(latest_restore_point) = latest_known_good_restore_point(sealed_runs) else {
170        return Vec::new();
171    };
172
173    let keep_last = keep_last.min(sealed_runs.len());
174    let mut retained = BTreeSet::new();
175    retained.insert(latest_restore_point);
176
177    if keep_last > 0 {
178        for index in sealed_runs.len() - keep_last..sealed_runs.len() {
179            retained.insert(index);
180        }
181    }
182
183    sealed_runs
184        .iter()
185        .enumerate()
186        .filter(|(index, _)| !retained.contains(index))
187        .map(|(_, run)| run.clone())
188        .collect()
189}
190
191fn snapshot_anchored_reclaim_candidates(sealed_runs: &[RetentionRun]) -> Vec<RetentionRun> {
192    let Some(latest_restore_point) = latest_known_good_restore_point(sealed_runs) else {
193        return Vec::new();
194    };
195
196    sealed_runs[..latest_restore_point].to_vec()
197}
198
199fn latest_known_good_restore_point(sealed_runs: &[RetentionRun]) -> Option<usize> {
200    sealed_runs
201        .iter()
202        .rposition(RetentionRun::is_known_good_restore_point)
203}
204
205fn snapshot_anchor_status(
206    durable_high_watermark: u64,
207    anchor: Option<SnapshotAnchor>,
208) -> SnapshotAnchorStatus {
209    let Some(anchor) = anchor else {
210        return SnapshotAnchorStatus::Missing;
211    };
212
213    // Validate against the durable watermark, not the manifest's: a tail-trimmed run
214    // keeps its manifest value, and trusting it could anoint a restore point the
215    // restore path itself rejects while everything else is reclaimed.
216    if anchor.high_watermark <= durable_high_watermark {
217        return SnapshotAnchorStatus::Valid(anchor);
218    }
219
220    SnapshotAnchorStatus::Invalid(format!(
221        "snapshot anchor high_watermark {} exceeds durable high_watermark {durable_high_watermark}",
222        anchor.high_watermark,
223    ))
224}
225
226#[cfg(test)]
227mod tests {
228    use rstest::rstest;
229
230    use super::*;
231
232    #[rstest]
233    fn snapshot_anchor_status_rejects_anchor_past_durable_watermark() {
234        let status = snapshot_anchor_status(
235            1,
236            Some(SnapshotAnchor::new(2, "cache://snapshots/2", "blake3:abc")),
237        );
238
239        match status {
240            SnapshotAnchorStatus::Invalid(msg) => {
241                assert!(
242                    msg.contains("exceeds durable high_watermark"),
243                    "msg was: {msg}",
244                );
245            }
246            other => panic!("expected Invalid, was {other:?}"),
247        }
248    }
249}
nautilus_event_store/retention.rs

nautilus_event_store/
retention.rs
