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    sealed_runs.sort_by_key(|run| run.manifest.start_ts_init);
148
149    let reclaim_candidates = match mode {
150        RetentionMode::Full => Vec::new(),
151        RetentionMode::Bounded { keep_last } => bounded_reclaim_candidates(&sealed_runs, keep_last),
152        RetentionMode::SnapshotAnchored => snapshot_anchored_reclaim_candidates(&sealed_runs),
153    };
154
155    RetentionPlan {
156        sealed_runs,
157        reclaim_candidates,
158    }
159}
160
161fn bounded_reclaim_candidates(sealed_runs: &[RetentionRun], keep_last: usize) -> Vec<RetentionRun> {
162    let Some(latest_restore_point) = latest_known_good_restore_point(sealed_runs) else {
163        return Vec::new();
164    };
165
166    let keep_last = keep_last.min(sealed_runs.len());
167    let mut retained = BTreeSet::new();
168    retained.insert(latest_restore_point);
169
170    if keep_last > 0 {
171        for index in sealed_runs.len() - keep_last..sealed_runs.len() {
172            retained.insert(index);
173        }
174    }
175
176    sealed_runs
177        .iter()
178        .enumerate()
179        .filter(|(index, _)| !retained.contains(index))
180        .map(|(_, run)| run.clone())
181        .collect()
182}
183
184fn snapshot_anchored_reclaim_candidates(sealed_runs: &[RetentionRun]) -> Vec<RetentionRun> {
185    let Some(latest_restore_point) = latest_known_good_restore_point(sealed_runs) else {
186        return Vec::new();
187    };
188
189    sealed_runs[..latest_restore_point].to_vec()
190}
191
192fn latest_known_good_restore_point(sealed_runs: &[RetentionRun]) -> Option<usize> {
193    sealed_runs
194        .iter()
195        .rposition(RetentionRun::is_known_good_restore_point)
196}
197
198fn snapshot_anchor_status(
199    durable_high_watermark: u64,
200    anchor: Option<SnapshotAnchor>,
201) -> SnapshotAnchorStatus {
202    let Some(anchor) = anchor else {
203        return SnapshotAnchorStatus::Missing;
204    };
205
206    // Validate against the durable watermark, not the manifest's: a tail-trimmed run
207    // keeps its manifest value, and trusting it could anoint a restore point the
208    // restore path itself rejects while everything else is reclaimed.
209    if anchor.high_watermark <= durable_high_watermark {
210        return SnapshotAnchorStatus::Valid(anchor);
211    }
212
213    SnapshotAnchorStatus::Invalid(format!(
214        "snapshot anchor high_watermark {} exceeds durable high_watermark {durable_high_watermark}",
215        anchor.high_watermark,
216    ))
217}
218
219#[cfg(test)]
220mod tests {
221    use rstest::rstest;
222
223    use super::*;
224
225    #[rstest]
226    fn snapshot_anchor_status_rejects_anchor_past_durable_watermark() {
227        let status = snapshot_anchor_status(
228            1,
229            Some(SnapshotAnchor::new(2, "cache://snapshots/2", "blake3:abc")),
230        );
231
232        match status {
233            SnapshotAnchorStatus::Invalid(msg) => {
234                assert!(
235                    msg.contains("exceeds durable high_watermark"),
236                    "msg was: {msg}",
237                );
238            }
239            other => panic!("expected Invalid, was {other:?}"),
240        }
241    }
242}
nautilus_event_store/retention.rs

nautilus_event_store/
retention.rs
