fxcp_core/
tombstone.rs

1// SPDX-License-Identifier: GPL-2.0-or-later
2// Copyright (C) 2025 Joel Wirāmu Pauling <aenertia@aenertia.net>
3//
4// fxcp-core/src/tombstone.rs  --  Persistent tombstone journal for delete propagation
5
6//! Append-only JSONL journal recording file/directory deletions.
7//!
8//! Used by foxingd to persist deletion events across daemon restarts, and by
9//! fxcp `--delete` to avoid full target tree walks. Entries are idempotent  -- 
10//! replaying a tombstone for an already-deleted path is a no-op.
11//!
12//! Format: one JSON object per line (JSONL), crash-resilient for lines < PIPE_BUF.
13//! ```json
14//! {"p":"relative/path.txt","d":false,"t":1741400000,"s":42}
15//! ```
16
17use std::fs::{self, File, OpenOptions};
18use std::io::{BufRead, BufReader, Write};
19use std::path::{Path, PathBuf};
20use std::sync::atomic::{AtomicU64, Ordering};
21use tracing::{debug, warn};
22
23/// A single tombstone entry recording a deletion event.
24#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
25pub struct TombstoneEntry {
26    /// Relative path from source/target root.
27    #[serde(rename = "p")]
28    pub rel_path: PathBuf,
29    /// Whether this was a directory removal (Rmdir) vs file (Unlink).
30    #[serde(rename = "d")]
31    pub is_dir: bool,
32    /// Wall-clock timestamp of the deletion (UTC epoch seconds).
33    #[serde(rename = "t")]
34    pub timestamp: i64,
35    /// BPF sequence number at time of capture (0 if from fxcp scan).
36    #[serde(rename = "s")]
37    pub seq: u64,
38}
39
40/// Persistent append-only tombstone journal backed by a JSONL file.
41#[derive(Debug)]
42pub struct TombstoneJournal {
43    path: PathBuf,
44    entry_count: AtomicU64,
45}
46
47impl TombstoneJournal {
48    /// Open or create a tombstone journal at the given path.
49    pub fn open(path: &Path) -> std::io::Result<Self> {
50        let count = if path.exists() {
51            let file = File::open(path)?;
52            BufReader::new(file).lines().count() as u64
53        } else {
54            0
55        };
56        Ok(Self {
57            path: path.to_path_buf(),
58            entry_count: AtomicU64::new(count),
59        })
60    }
61
62    /// Append a tombstone entry to the journal.
63    ///
64    /// Uses append mode  --  atomic for lines shorter than PIPE_BUF (4096 bytes)
65    /// on POSIX systems when the journal is on a local filesystem.
66    pub fn append(&self, entry: &TombstoneEntry) -> std::io::Result<()> {
67        let mut line = serde_json::to_string(entry)
68            .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;
69        line.push('\n');
70
71        let mut file = OpenOptions::new()
72            .create(true)
73            .append(true)
74            .open(&self.path)?;
75        file.write_all(line.as_bytes())?;
76        self.entry_count.fetch_add(1, Ordering::Relaxed);
77        Ok(())
78    }
79
80    /// Read all valid entries from the journal, skipping malformed lines.
81    pub fn read_all(&self) -> std::io::Result<Vec<TombstoneEntry>> {
82        if !self.path.exists() {
83            return Ok(Vec::new());
84        }
85        let file = File::open(&self.path)?;
86        let reader = BufReader::new(file);
87        let mut entries = Vec::new();
88        for (line_num, line) in reader.lines().enumerate() {
89            let line = match line {
90                Ok(l) => l,
91                Err(e) => {
92                    warn!("tombstone journal line {}: read error: {}", line_num + 1, e);
93                    continue;
94                }
95            };
96            if line.trim().is_empty() {
97                continue;
98            }
99            match serde_json::from_str::<TombstoneEntry>(&line) {
100                Ok(entry) => entries.push(entry),
101                Err(e) => {
102                    debug!("tombstone journal line {}: malformed, skipping: {}", line_num + 1, e);
103                }
104            }
105        }
106        Ok(entries)
107    }
108
109    /// Remove entries older than `max_age_secs` and atomically replace the journal.
110    ///
111    /// Also removes entries whose path has been re-created (nullification):
112    /// if a Create event with a later timestamp exists for the same path,
113    /// the tombstone is stale.
114    pub fn compact(&self, max_age_secs: i64) -> std::io::Result<u64> {
115        let entries = self.read_all()?;
116        let now = chrono::Utc::now().timestamp();
117        let cutoff = now - max_age_secs;
118
119        let retained: Vec<&TombstoneEntry> = entries.iter()
120            .filter(|e| e.timestamp >= cutoff)
121            .collect();
122
123        let removed = entries.len() as u64 - retained.len() as u64;
124
125        // Atomic replace: write to temp file, then rename
126        let tmp_path = self.path.with_extension("jsonl.tmp");
127        {
128            let mut tmp = File::create(&tmp_path)?;
129            for entry in &retained {
130                let mut line = serde_json::to_string(entry)
131                    .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;
132                line.push('\n');
133                tmp.write_all(line.as_bytes())?;
134            }
135            tmp.sync_all()?;
136        }
137        fs::rename(&tmp_path, &self.path)?;
138        self.entry_count.store(retained.len() as u64, Ordering::Relaxed);
139
140        Ok(removed)
141    }
142
143    /// Clear the journal (truncate to empty).
144    pub fn clear(&self) -> std::io::Result<()> {
145        if self.path.exists() {
146            File::create(&self.path)?; // truncate
147            self.entry_count.store(0, Ordering::Relaxed);
148        }
149        Ok(())
150    }
151
152    /// Current number of entries (approximate  --  may lag slightly under concurrent appends).
153    pub fn len(&self) -> u64 {
154        self.entry_count.load(Ordering::Relaxed)
155    }
156
157    /// Whether the journal is empty.
158    pub fn is_empty(&self) -> bool {
159        self.len() == 0
160    }
161
162    /// Path to the journal file.
163    pub fn path(&self) -> &Path {
164        &self.path
165    }
166}
167
168/// Replay tombstone entries against a target directory, removing files/dirs.
169///
170/// Returns the number of successfully removed entries.
171pub fn replay_tombstones(
172    target: &Path,
173    entries: &[TombstoneEntry],
174    excludes: &[glob::Pattern],
175) -> crate::Result<u64> {
176    let mut deleted = 0u64;
177    for entry in entries {
178        if excludes.iter().any(|p| p.matches_path(&entry.rel_path)) {
179            continue;
180        }
181        let target_path = target.join(&entry.rel_path);
182        let result = if entry.is_dir {
183            fs::remove_dir(&target_path)
184        } else {
185            fs::remove_file(&target_path)
186        };
187        match result {
188            Ok(()) => { deleted += 1; }
189            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {} // already gone
190            Err(e) => {
191                debug!("tombstone replay {:?}: {}", entry.rel_path, e);
192            }
193        }
194    }
195    Ok(deleted)
196}
197
198#[cfg(test)]
199mod tests {
200    #![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
201    use super::*;
202
203    fn tmp_journal() -> (tempfile::TempDir, TombstoneJournal) {
204        let dir = tempfile::tempdir().unwrap();
205        let path = dir.path().join(".foxing_tombstones.jsonl");
206        let journal = TombstoneJournal::open(&path).unwrap();
207        (dir, journal)
208    }
209
210    #[test]
211    fn test_append_and_read() {
212        let (_dir, journal) = tmp_journal();
213
214        journal.append(&TombstoneEntry {
215            rel_path: PathBuf::from("foo/bar.txt"),
216            is_dir: false,
217            timestamp: 1000,
218            seq: 1,
219        }).unwrap();
220
221        journal.append(&TombstoneEntry {
222            rel_path: PathBuf::from("baz/"),
223            is_dir: true,
224            timestamp: 1001,
225            seq: 2,
226        }).unwrap();
227
228        assert_eq!(journal.len(), 2);
229
230        let entries = journal.read_all().unwrap();
231        assert_eq!(entries.len(), 2);
232        assert_eq!(entries[0].rel_path, PathBuf::from("foo/bar.txt"));
233        assert!(!entries[0].is_dir);
234        assert_eq!(entries[1].rel_path, PathBuf::from("baz/"));
235        assert!(entries[1].is_dir);
236    }
237
238    #[test]
239    fn test_malformed_lines_skipped() {
240        let dir = tempfile::tempdir().unwrap();
241        let path = dir.path().join(".foxing_tombstones.jsonl");
242
243        // Write a mix of valid and invalid lines
244        let mut f = File::create(&path).unwrap();
245        writeln!(f, r#"{{"p":"good.txt","d":false,"t":1000,"s":1}}"#).unwrap();
246        writeln!(f, "this is not json").unwrap();
247        writeln!(f, r#"{{"p":"also-good.txt","d":false,"t":1001,"s":2}}"#).unwrap();
248        writeln!(f).unwrap(); // empty line
249        drop(f);
250
251        let journal = TombstoneJournal::open(&path).unwrap();
252        let entries = journal.read_all().unwrap();
253        assert_eq!(entries.len(), 2);
254        assert_eq!(entries[0].rel_path, PathBuf::from("good.txt"));
255        assert_eq!(entries[1].rel_path, PathBuf::from("also-good.txt"));
256    }
257
258    #[test]
259    fn test_clear() {
260        let (_dir, journal) = tmp_journal();
261
262        journal.append(&TombstoneEntry {
263            rel_path: PathBuf::from("x.txt"),
264            is_dir: false,
265            timestamp: 1000,
266            seq: 1,
267        }).unwrap();
268        assert_eq!(journal.len(), 1);
269
270        journal.clear().unwrap();
271        assert_eq!(journal.len(), 0);
272        assert!(journal.read_all().unwrap().is_empty());
273    }
274
275    #[test]
276    fn test_compact_removes_old() {
277        let (_dir, journal) = tmp_journal();
278        let now = chrono::Utc::now().timestamp();
279
280        // Old entry (beyond TTL)
281        journal.append(&TombstoneEntry {
282            rel_path: PathBuf::from("old.txt"),
283            is_dir: false,
284            timestamp: now - 100_000,
285            seq: 1,
286        }).unwrap();
287
288        // Recent entry (within TTL)
289        journal.append(&TombstoneEntry {
290            rel_path: PathBuf::from("new.txt"),
291            is_dir: false,
292            timestamp: now - 10,
293            seq: 2,
294        }).unwrap();
295
296        let removed = journal.compact(43200).unwrap(); // 12 hour TTL
297        assert_eq!(removed, 1);
298
299        let entries = journal.read_all().unwrap();
300        assert_eq!(entries.len(), 1);
301        assert_eq!(entries[0].rel_path, PathBuf::from("new.txt"));
302    }
303
304    #[test]
305    fn test_replay_tombstones() {
306        let dir = tempfile::tempdir().unwrap();
307        let target = dir.path();
308
309        // Create files to delete
310        fs::write(target.join("keep.txt"), "keep").unwrap();
311        fs::write(target.join("delete.txt"), "delete").unwrap();
312        fs::create_dir(target.join("rmdir")).unwrap();
313
314        let entries = vec![
315            TombstoneEntry { rel_path: PathBuf::from("delete.txt"), is_dir: false, timestamp: 1000, seq: 1 },
316            TombstoneEntry { rel_path: PathBuf::from("rmdir"), is_dir: true, timestamp: 1001, seq: 2 },
317            TombstoneEntry { rel_path: PathBuf::from("nonexistent.txt"), is_dir: false, timestamp: 1002, seq: 3 },
318        ];
319
320        let deleted = replay_tombstones(target, &entries, &[]).unwrap();
321        assert_eq!(deleted, 2);
322        assert!(target.join("keep.txt").exists());
323        assert!(!target.join("delete.txt").exists());
324        assert!(!target.join("rmdir").exists());
325    }
326
327    #[test]
328    fn test_empty_journal() {
329        let dir = tempfile::tempdir().unwrap();
330        let path = dir.path().join("nonexistent.jsonl");
331        let journal = TombstoneJournal::open(&path).unwrap();
332        assert_eq!(journal.len(), 0);
333        assert!(journal.is_empty());
334        assert!(journal.read_all().unwrap().is_empty());
335    }
336}
fxcp_core/tombstone.rs

fxcp_core/
tombstone.rs
