talos_session/
artifacts.rs

1//! Session-owned transcript and pending-submission artifact lifecycle.
2
3use std::collections::{BTreeMap, BTreeSet, HashSet};
4use std::fs::{self, OpenOptions};
5use std::io::Write;
6use std::path::{Path, PathBuf};
7use std::time::Duration;
8
9use rusqlite::{Connection, ErrorCode, OpenFlags};
10use uuid::Uuid;
11
12use crate::SessionError;
13
14/// Default maximum number of filesystem directory entries inspected in one pass.
15pub const DEFAULT_MAX_ORPHAN_SIDECAR_ENTRIES: usize = 4_096;
16/// Default grace period before a transcript-less sidecar can be reconciled.
17pub const DEFAULT_ORPHAN_SIDECAR_MINIMUM_AGE: Duration = Duration::from_secs(300);
18const ORPHAN_SIDECAR_SCAN_STATE_FILE: &str = ".orphan-sidecar-scan-budget";
19
20/// Successful deletion details for one Session-owned artifact operation.
21#[derive(Debug, Clone, Default, PartialEq, Eq)]
22pub struct SessionArtifactCleanupReport {
23    /// Number of files actually removed. Missing files do not increment this value.
24    pub removed_artifacts: usize,
25    /// Total bytes reported by filesystem metadata for removed files.
26    pub bytes_removed: u64,
27    /// Exact paths removed, in deletion order.
28    pub removed_paths: Vec<PathBuf>,
29}
30
31impl SessionArtifactCleanupReport {
32    pub(crate) fn merge(&mut self, other: Self) {
33        self.removed_artifacts = self
34            .removed_artifacts
35            .saturating_add(other.removed_artifacts);
36        self.bytes_removed = self.bytes_removed.saturating_add(other.bytes_removed);
37        self.removed_paths.extend(other.removed_paths);
38    }
39}
40
41/// Safety policy for bounded orphan-sidecar reconciliation.
42#[derive(Debug, Clone, PartialEq, Eq)]
43pub struct OrphanSidecarReconciliationPolicy {
44    /// Session IDs that are live, deferred, or otherwise protected by a caller.
45    pub protected_session_ids: Vec<Uuid>,
46    /// Maximum filesystem directory entries inspected in this pass.
47    pub max_entries: usize,
48    /// Minimum age required before a transcript-less sidecar can be removed.
49    pub minimum_age: Duration,
50}
51
52impl Default for OrphanSidecarReconciliationPolicy {
53    fn default() -> Self {
54        Self {
55            protected_session_ids: Vec::new(),
56            max_entries: DEFAULT_MAX_ORPHAN_SIDECAR_ENTRIES,
57            minimum_age: DEFAULT_ORPHAN_SIDECAR_MINIMUM_AGE,
58        }
59    }
60}
61
62/// One reconciliation failure retained without aborting the whole bounded pass.
63#[derive(Debug, Clone, PartialEq, Eq)]
64pub struct OrphanSidecarFailure {
65    /// Session ID inferred from the strictly validated filename.
66    pub session_id: Uuid,
67    /// Exact SQLite sidecar path associated with the failed set.
68    pub path: PathBuf,
69    /// Content-free filesystem or SQLite diagnostic.
70    pub error: String,
71}
72
73/// Result of scanning Session roots for transcript-less pending SQLite artifacts.
74#[derive(Debug, Clone, Default, PartialEq, Eq)]
75pub struct OrphanSidecarReconciliationReport {
76    /// Number of filesystem directory entries inspected, including unrelated names.
77    pub scanned_entries: usize,
78    /// Number of complete orphan sets for which at least one artifact was removed.
79    pub removed_sets: usize,
80    /// Number of individual SQLite/WAL/SHM files removed.
81    pub removed_artifacts: usize,
82    /// Total bytes removed from orphan sidecars.
83    pub bytes_removed: u64,
84    /// Number of sets skipped due to safety checks or concurrent disappearance.
85    pub skipped_sets: usize,
86    /// Failures retained for diagnosis; other candidate sets are still processed.
87    pub failures: Vec<OrphanSidecarFailure>,
88    /// Whether the configured scan bound stopped the pass before exhaustion.
89    pub bounded: bool,
90}
91
92#[derive(Debug, Clone, PartialEq, Eq)]
93struct SessionArtifactSet {
94    transcript: PathBuf,
95    sqlite: PathBuf,
96    wal: PathBuf,
97    shm: PathBuf,
98}
99
100impl SessionArtifactSet {
101    fn for_transcript(transcript: &Path) -> Result<Self, SessionError> {
102        let extension = transcript.extension().and_then(|value| value.to_str());
103        if !matches!(extension, Some("tlog" | "jsonl")) {
104            return Err(SessionError::ParseError(format!(
105                "invalid Session transcript extension: {}",
106                transcript.display()
107            )));
108        }
109        let stem = transcript
110            .file_stem()
111            .and_then(|value| value.to_str())
112            .ok_or_else(|| {
113                SessionError::ParseError(format!(
114                    "invalid Session transcript path: {}",
115                    transcript.display()
116                ))
117            })?;
118        Uuid::parse_str(stem).map_err(|_| {
119            SessionError::ParseError(format!(
120                "Session transcript filename is not a UUID: {}",
121                transcript.display()
122            ))
123        })?;
124        let sqlite = transcript.with_file_name(format!("{stem}.pending.sqlite"));
125        Ok(Self {
126            transcript: transcript.to_path_buf(),
127            wal: PathBuf::from(format!("{}-wal", sqlite.display())),
128            shm: PathBuf::from(format!("{}-shm", sqlite.display())),
129            sqlite,
130        })
131    }
132
133    fn for_workspace(workspace: &Path, id: Uuid) -> Self {
134        let sqlite = workspace.join(format!("{id}.pending.sqlite"));
135        Self {
136            transcript: workspace.join(format!("{id}.tlog")),
137            wal: PathBuf::from(format!("{}-wal", sqlite.display())),
138            shm: PathBuf::from(format!("{}-shm", sqlite.display())),
139            sqlite,
140        }
141    }
142
143    fn sidecars(&self) -> [PathBuf; 3] {
144        [self.wal.clone(), self.shm.clone(), self.sqlite.clone()]
145    }
146}
147
148fn existing_paths(paths: &[PathBuf]) -> Vec<PathBuf> {
149    paths
150        .iter()
151        .filter(|path| fs::symlink_metadata(path).is_ok())
152        .cloned()
153        .collect()
154}
155
156fn remove_paths(paths: &[PathBuf]) -> Result<SessionArtifactCleanupReport, SessionError> {
157    let mut report = SessionArtifactCleanupReport::default();
158    for (index, path) in paths.iter().enumerate() {
159        match fs::symlink_metadata(path) {
160            Ok(metadata) => {
161                if let Err(source) = fs::remove_file(path) {
162                    return Err(SessionError::ArtifactCleanup {
163                        path: path.clone(),
164                        source,
165                        removed: report.removed_paths,
166                        remaining: existing_paths(&paths[index..]),
167                    });
168                }
169                report.removed_artifacts = report.removed_artifacts.saturating_add(1);
170                report.bytes_removed = report.bytes_removed.saturating_add(metadata.len());
171                report.removed_paths.push(path.clone());
172            }
173            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
174            Err(source) => {
175                return Err(SessionError::ArtifactCleanup {
176                    path: path.clone(),
177                    source,
178                    removed: report.removed_paths,
179                    remaining: existing_paths(&paths[index..]),
180                });
181            }
182        }
183    }
184    Ok(report)
185}
186
187/// Removes WAL, SHM and pending SQLite in that order while retaining the transcript.
188///
189/// A failure leaves the transcript discoverable so `/delete` or retention can retry.
190pub fn remove_session_sidecars_for_transcript(
191    transcript_path: &Path,
192) -> Result<SessionArtifactCleanupReport, SessionError> {
193    let set = SessionArtifactSet::for_transcript(transcript_path)?;
194    remove_paths(&set.sidecars())
195}
196
197/// Removes only the transcript, the final discoverability commit point.
198pub fn remove_session_transcript(
199    transcript_path: &Path,
200) -> Result<SessionArtifactCleanupReport, SessionError> {
201    let set = SessionArtifactSet::for_transcript(transcript_path)?;
202    remove_paths(&[set.transcript])
203}
204
205/// Removes the complete Session-owned filesystem set with transcript last.
206pub fn remove_session_artifacts_for_transcript(
207    transcript_path: &Path,
208) -> Result<SessionArtifactCleanupReport, SessionError> {
209    let mut report = remove_session_sidecars_for_transcript(transcript_path)?;
210    report.merge(remove_session_transcript(transcript_path)?);
211    Ok(report)
212}
213
214fn parse_sidecar_session_id(name: &str) -> Option<Uuid> {
215    [
216        ".pending.sqlite-wal",
217        ".pending.sqlite-shm",
218        ".pending.sqlite",
219    ]
220    .iter()
221    .find_map(|suffix| name.strip_suffix(suffix))
222    .and_then(|stem| Uuid::parse_str(stem).ok())
223}
224
225fn path_is_symlink(path: &Path) -> bool {
226    fs::symlink_metadata(path).is_ok_and(|metadata| metadata.file_type().is_symlink())
227}
228
229fn sidecars_are_old_enough(set: &SessionArtifactSet, minimum_age: Duration) -> bool {
230    set.sidecars()
231        .iter()
232        .filter_map(|path| fs::symlink_metadata(path).ok())
233        .all(|metadata| {
234            metadata
235                .modified()
236                .ok()
237                .and_then(|modified| modified.elapsed().ok())
238                .is_some_and(|age| age >= minimum_age)
239        })
240}
241
242fn sqlite_sidecar_is_busy(path: &Path) -> Result<bool, SessionError> {
243    if !path.exists() {
244        return Ok(false);
245    }
246    let connection = Connection::open_with_flags(
247        path,
248        OpenFlags::SQLITE_OPEN_READ_WRITE | OpenFlags::SQLITE_OPEN_NO_MUTEX,
249    )
250    .map_err(|error| SessionError::OrphanReconciliation {
251        path: path.to_path_buf(),
252        message: error.to_string(),
253    })?;
254    connection.busy_timeout(Duration::ZERO).map_err(|error| {
255        SessionError::OrphanReconciliation {
256            path: path.to_path_buf(),
257            message: error.to_string(),
258        }
259    })?;
260    match connection.execute_batch("BEGIN EXCLUSIVE; ROLLBACK;") {
261        Ok(()) => Ok(false),
262        Err(rusqlite::Error::SqliteFailure(failure, _))
263            if matches!(
264                failure.code,
265                ErrorCode::DatabaseBusy | ErrorCode::DatabaseLocked
266            ) =>
267        {
268            Ok(true)
269        }
270        Err(error) => Err(SessionError::OrphanReconciliation {
271            path: path.to_path_buf(),
272            message: error.to_string(),
273        }),
274    }
275}
276
277fn orphan_scan_state_path(canonical_root: &Path) -> PathBuf {
278    canonical_root.join(ORPHAN_SIDECAR_SCAN_STATE_FILE)
279}
280
281fn read_orphan_scan_limit(canonical_root: &Path, base_limit: usize) -> Result<usize, SessionError> {
282    let state_path = orphan_scan_state_path(canonical_root);
283    match fs::symlink_metadata(&state_path) {
284        Ok(metadata) if metadata.file_type().is_symlink() || !metadata.file_type().is_file() => {
285            return Err(SessionError::OrphanReconciliation {
286                path: state_path,
287                message: "scan continuation state must be a regular file".to_string(),
288            });
289        }
290        Ok(_) => {}
291        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(base_limit),
292        Err(error) => return Err(error.into()),
293    }
294
295    let persisted = fs::read_to_string(&state_path)?
296        .trim()
297        .parse::<usize>()
298        .ok()
299        .filter(|value| *value > 0);
300    match persisted {
301        Some(limit) => Ok(limit.max(base_limit)),
302        None => {
303            fs::remove_file(&state_path)?;
304            Ok(base_limit)
305        }
306    }
307}
308
309fn persist_next_orphan_scan_limit(
310    canonical_root: &Path,
311    current_limit: usize,
312) -> Result<(), SessionError> {
313    let state_path = orphan_scan_state_path(canonical_root);
314    if fs::symlink_metadata(&state_path)
315        .is_ok_and(|metadata| metadata.file_type().is_symlink() || !metadata.file_type().is_file())
316    {
317        return Err(SessionError::OrphanReconciliation {
318            path: state_path,
319            message: "scan continuation state must be a regular file".to_string(),
320        });
321    }
322    let next_limit = current_limit
323        .saturating_mul(2)
324        .max(current_limit.saturating_add(1));
325    let mut state = OpenOptions::new()
326        .create(true)
327        .truncate(true)
328        .write(true)
329        .open(&state_path)?;
330    writeln!(state, "{next_limit}")?;
331    state.sync_all()?;
332    Ok(())
333}
334
335fn clear_orphan_scan_limit(canonical_root: &Path) -> Result<(), SessionError> {
336    let state_path = orphan_scan_state_path(canonical_root);
337    match fs::remove_file(state_path) {
338        Ok(()) => Ok(()),
339        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
340        Err(error) => Err(error.into()),
341    }
342}
343
344pub(crate) fn reconcile_orphan_sidecars_in_root(
345    sessions_dir: &Path,
346    policy: &OrphanSidecarReconciliationPolicy,
347) -> Result<OrphanSidecarReconciliationReport, SessionError> {
348    let mut report = OrphanSidecarReconciliationReport::default();
349    if !sessions_dir.exists() {
350        return Ok(report);
351    }
352    let canonical_root = sessions_dir.canonicalize()?;
353    let protected: HashSet<Uuid> = policy.protected_session_ids.iter().copied().collect();
354    let base_limit = policy.max_entries.max(1);
355    let limit = read_orphan_scan_limit(&canonical_root, base_limit)?;
356    let mut sets: BTreeMap<(PathBuf, Uuid), SessionArtifactSet> = BTreeMap::new();
357    let mut blocked: BTreeSet<(PathBuf, Uuid)> = BTreeSet::new();
358
359    'workspaces: for workspace_entry in fs::read_dir(&canonical_root)? {
360        let workspace_entry = workspace_entry?;
361        if workspace_entry.file_name().to_str() == Some(ORPHAN_SIDECAR_SCAN_STATE_FILE) {
362            continue;
363        }
364        if report.scanned_entries >= limit {
365            report.bounded = true;
366            break;
367        }
368        report.scanned_entries = report.scanned_entries.saturating_add(1);
369        let metadata = fs::symlink_metadata(workspace_entry.path())?;
370        if !metadata.file_type().is_dir() || metadata.file_type().is_symlink() {
371            continue;
372        }
373        let workspace = workspace_entry.path().canonicalize()?;
374        if !workspace.starts_with(&canonical_root) {
375            continue;
376        }
377        for entry in fs::read_dir(&workspace)? {
378            if report.scanned_entries >= limit {
379                report.bounded = true;
380                break 'workspaces;
381            }
382            report.scanned_entries = report.scanned_entries.saturating_add(1);
383            let entry = entry?;
384            let name = entry.file_name();
385            let Some(name) = name.to_str() else {
386                continue;
387            };
388            let Some(id) = parse_sidecar_session_id(name) else {
389                continue;
390            };
391            let key = (workspace.clone(), id);
392            if fs::symlink_metadata(entry.path()).is_ok_and(|value| value.file_type().is_symlink())
393            {
394                blocked.insert(key.clone());
395            }
396            sets.entry(key)
397                .or_insert_with(|| SessionArtifactSet::for_workspace(&workspace, id));
398        }
399    }
400
401    for ((workspace, id), set) in sets {
402        if blocked.contains(&(workspace.clone(), id)) || protected.contains(&id) {
403            report.skipped_sets = report.skipped_sets.saturating_add(1);
404            continue;
405        }
406        let tlog = workspace.join(format!("{id}.tlog"));
407        let jsonl = workspace.join(format!("{id}.jsonl"));
408        if fs::symlink_metadata(&tlog).is_ok() || fs::symlink_metadata(&jsonl).is_ok() {
409            report.skipped_sets = report.skipped_sets.saturating_add(1);
410            continue;
411        }
412        if set.sidecars().iter().any(|path| path_is_symlink(path))
413            || !sidecars_are_old_enough(&set, policy.minimum_age)
414        {
415            report.skipped_sets = report.skipped_sets.saturating_add(1);
416            continue;
417        }
418        match sqlite_sidecar_is_busy(&set.sqlite) {
419            Ok(true) => {
420                report.skipped_sets = report.skipped_sets.saturating_add(1);
421                continue;
422            }
423            Err(error) => {
424                report.failures.push(OrphanSidecarFailure {
425                    session_id: id,
426                    path: set.sqlite.clone(),
427                    error: error.to_string(),
428                });
429                continue;
430            }
431            Ok(false) => {}
432        }
433        match remove_paths(&set.sidecars()) {
434            Ok(cleanup) if cleanup.removed_artifacts > 0 => {
435                report.removed_sets = report.removed_sets.saturating_add(1);
436                report.removed_artifacts = report
437                    .removed_artifacts
438                    .saturating_add(cleanup.removed_artifacts);
439                report.bytes_removed = report.bytes_removed.saturating_add(cleanup.bytes_removed);
440            }
441            Ok(_) => {
442                report.skipped_sets = report.skipped_sets.saturating_add(1);
443            }
444            Err(error) => report.failures.push(OrphanSidecarFailure {
445                session_id: id,
446                path: set.sqlite,
447                error: error.to_string(),
448            }),
449        }
450    }
451
452    if report.bounded {
453        persist_next_orphan_scan_limit(&canonical_root, limit)?;
454    } else {
455        clear_orphan_scan_limit(&canonical_root)?;
456    }
457
458    Ok(report)
459}
talos_session/artifacts.rs

talos_session/
artifacts.rs
