aft/gc/
mod.rs

1//! Budgeted mark-and-sweep for immutable family blob stores.
2//!
3//! References from the retained manifests, live assembly pins, and active query
4//! read markers are all marked before the budget selects eviction candidates.
5
6use std::collections::{BTreeMap, BTreeSet};
7use std::fmt;
8use std::fs;
9use std::path::{Path, PathBuf};
10
11use rusqlite::{params, Connection};
12
13use crate::blob_store::BlobPlane;
14use crate::pins::{self, PinMetadata, PIN_TTL_MS};
15use crate::root_cache;
16
17/// Payloads newer than this stay available even when a store is over budget.
18pub const BLOB_AGE_FLOOR_MS: u64 = 15 * 60 * 1_000;
19
20#[derive(Debug)]
21pub enum SweepError {
22    Io(std::io::Error),
23    Sqlite(rusqlite::Error),
24    Pin(pins::PinError),
25    Metadata(serde_json::Error),
26}
27
28impl fmt::Display for SweepError {
29    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
30        match self {
31            Self::Io(error) => write!(f, "GC I/O error: {error}"),
32            Self::Sqlite(error) => write!(f, "GC SQLite error: {error}"),
33            Self::Pin(error) => write!(f, "GC pin error: {error}"),
34            Self::Metadata(error) => write!(f, "GC pin metadata error: {error}"),
35        }
36    }
37}
38
39impl std::error::Error for SweepError {}
40
41impl From<std::io::Error> for SweepError {
42    fn from(error: std::io::Error) -> Self {
43        Self::Io(error)
44    }
45}
46impl From<rusqlite::Error> for SweepError {
47    fn from(error: rusqlite::Error) -> Self {
48        Self::Sqlite(error)
49    }
50}
51impl From<pins::PinError> for SweepError {
52    fn from(error: pins::PinError) -> Self {
53        Self::Pin(error)
54    }
55}
56impl From<serde_json::Error> for SweepError {
57    fn from(error: serde_json::Error) -> Self {
58        Self::Metadata(error)
59    }
60}
61
62/// References assembled from the current and previous manifests. `generation_keys`
63/// additionally lets active query markers protect an otherwise unretained generation.
64#[derive(Clone, Debug, Default)]
65pub struct SweepReferences {
66    pub retained_keys: BTreeSet<[u8; 32]>,
67    pub generation_keys: BTreeMap<String, BTreeSet<[u8; 32]>>,
68}
69
70#[derive(Clone, Debug)]
71pub struct SweepRequest<'a> {
72    pub storage: &'a Path,
73    pub family: &'a str,
74    pub view_dir: &'a Path,
75    pub byte_budget: u64,
76    pub now_ms: u64,
77    pub references: SweepReferences,
78}
79
80#[derive(Clone, Debug, Default, PartialEq, Eq)]
81pub struct SweepReport {
82    pub deleted_blobs: usize,
83    pub deleted_bytes: u64,
84    pub retained_bytes: u64,
85    pub protected_pin_keys: usize,
86    pub reclaimed_pins: usize,
87    pub reclaimed_read_markers: usize,
88}
89
90/// Performs one mark-and-sweep pass. It deletes only unreferenced payloads older
91/// than the age floor, stopping as soon as the family fits within its byte budget.
92pub fn sweep(request: SweepRequest<'_>) -> Result<SweepReport, SweepError> {
93    let mut report = SweepReport::default();
94    let mut references = request.references.retained_keys.clone();
95    mark_live_assembly_pins(&request, &mut references, &mut report)?;
96    mark_live_query_pins(&request, &mut references, &mut report);
97
98    for plane in [BlobPlane::Semantic, BlobPlane::Callgraph] {
99        let path = plane_path(request.storage, request.family, plane);
100        if !path.exists() {
101            continue;
102        }
103        sweep_plane(
104            &path,
105            request.now_ms,
106            request.byte_budget,
107            &references,
108            &mut report,
109        )?;
110    }
111    Ok(report)
112}
113
114fn mark_live_assembly_pins(
115    request: &SweepRequest<'_>,
116    references: &mut BTreeSet<[u8; 32]>,
117    report: &mut SweepReport,
118) -> Result<(), SweepError> {
119    let pins_dir = request.view_dir.join("pins");
120    let entries = match fs::read_dir(&pins_dir) {
121        Ok(entries) => entries,
122        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(()),
123        Err(error) => return Err(error.into()),
124    };
125
126    for entry in entries {
127        let entry = entry?;
128        let path = entry.path();
129        if path.extension().and_then(|extension| extension.to_str()) != Some("json") {
130            continue;
131        }
132        let metadata: PinMetadata = match serde_json::from_slice(&fs::read(&path)?) {
133            Ok(metadata) => metadata,
134            Err(_) => continue, // A partial metadata write is never a reason to delete blobs.
135        };
136        if metadata.family != request.family || metadata.view.is_empty() {
137            continue;
138        }
139        let (metadata_path, keys_path) = pins::pin_paths(request.view_dir, &metadata.generation);
140        if metadata_path != path {
141            continue;
142        }
143        let expired = request.now_ms.saturating_sub(metadata.renewed_at) > PIN_TTL_MS;
144        if expired || !pins::owner_is_live(&metadata.owner) {
145            let _ = fs::remove_file(&metadata_path);
146            let _ = fs::remove_file(&keys_path);
147            crate::fs_lock::sync_parent(&metadata_path);
148            report.reclaimed_pins += 1;
149            continue;
150        }
151        let keys = pins::read_keys(&keys_path)?;
152        report.protected_pin_keys += keys.len();
153        references.extend(keys);
154    }
155    Ok(())
156}
157
158fn mark_live_query_pins(
159    request: &SweepRequest<'_>,
160    references: &mut BTreeSet<[u8; 32]>,
161    report: &mut SweepReport,
162) {
163    let readers = request.view_dir.join("readers");
164    let Ok(entries) = fs::read_dir(readers) else {
165        return;
166    };
167    for entry in entries.flatten() {
168        let Ok(file_type) = entry.file_type() else {
169            continue;
170        };
171        if !file_type.is_dir() {
172            continue;
173        }
174        let Some(generation) = entry.file_name().to_str().map(str::to_owned) else {
175            continue;
176        };
177        let marker_sweep = root_cache::sweep_read_markers(request.view_dir, &generation);
178        report.reclaimed_read_markers += marker_sweep.removed_stale;
179        if marker_sweep.protected {
180            if let Some(keys) = request.references.generation_keys.get(&generation) {
181                references.extend(keys.iter().copied());
182            }
183        }
184    }
185}
186
187fn sweep_plane(
188    path: &Path,
189    now_ms: u64,
190    byte_budget: u64,
191    references: &BTreeSet<[u8; 32]>,
192    report: &mut SweepReport,
193) -> Result<(), SweepError> {
194    let connection = Connection::open(path)?;
195    let mut candidates = Vec::new();
196    let mut total_bytes = 0_u64;
197    {
198        let mut statement = connection.prepare(
199            "SELECT full_key, length(payload), created_at_ms
200             FROM blob_payloads ORDER BY created_at_ms ASC, full_key ASC",
201        )?;
202        let rows = statement.query_map([], |row| {
203            Ok((
204                row.get::<_, Vec<u8>>(0)?,
205                row.get::<_, u64>(1)?,
206                row.get::<_, u64>(2)?,
207            ))
208        })?;
209        for row in rows {
210            let (key, bytes, created_at_ms) = row?;
211            total_bytes = total_bytes.saturating_add(bytes);
212            let Ok(key) = <Vec<u8> as TryInto<[u8; 32]>>::try_into(key) else {
213                continue;
214            };
215            candidates.push((key, bytes, created_at_ms));
216        }
217    }
218
219    for (key, bytes, created_at_ms) in candidates {
220        if total_bytes <= byte_budget {
221            break;
222        }
223        // This reference check is the safety boundary: retained manifests and
224        // live pins must win over budget pressure.
225        if references.contains(&key) || now_ms.saturating_sub(created_at_ms) < BLOB_AGE_FLOOR_MS {
226            continue;
227        }
228        let deleted = connection.execute(
229            "DELETE FROM blob_payloads WHERE full_key = ?1",
230            params![key.as_slice()],
231        )?;
232        if deleted == 1 {
233            total_bytes = total_bytes.saturating_sub(bytes);
234            report.deleted_blobs += 1;
235            report.deleted_bytes = report.deleted_bytes.saturating_add(bytes);
236        }
237    }
238    report.retained_bytes = report.retained_bytes.saturating_add(total_bytes);
239    Ok(())
240}
241
242fn plane_path(storage: &Path, family: &str, plane: BlobPlane) -> PathBuf {
243    storage
244        .join("blobs")
245        .join(family)
246        .join(format!("{}.sqlite", plane.as_str()))
247}
aft/gc/mod.rs

aft/gc/
mod.rs
