aft/views/
generation.rs

1//! Generation-owned derived artifacts. Unpublished files are protected by the
2//! assembly pin; published files by the pointer and query read markers.
3use std::{
4    collections::HashMap,
5    fs,
6    path::{Path, PathBuf},
7    sync::{
8        atomic::{AtomicBool, Ordering},
9        Arc, Mutex, OnceLock, Weak,
10    },
11    time::{Duration, Instant},
12};
13
14use rusqlite::Connection;
15
16use super::{Result, ViewError, ViewStore};
17
18impl ViewStore {
19    pub fn derived_path(&self, generation: &str) -> Result<PathBuf> {
20        super::validate_generation(generation)?;
21        Ok(self.view_dir().join(format!("derived-{generation}.sqlite")))
22    }
23
24    pub fn trigram_path(&self, generation: &str) -> Result<PathBuf> {
25        super::validate_generation(generation)?;
26        Ok(self.view_dir().join(format!("trigram-{generation}.bin")))
27    }
28
29    pub fn blob_references_by_generation(
30        &self,
31    ) -> Result<std::collections::BTreeMap<String, std::collections::BTreeSet<[u8; 32]>>> {
32        let mut result = std::collections::BTreeMap::new();
33        for entry in fs::read_dir(self.view_dir())? {
34            let entry = entry?;
35            let name = entry.file_name();
36            let Some(generation) = name
37                .to_str()
38                .and_then(|name| name.strip_prefix("manifest-"))
39                .and_then(|name| name.strip_suffix(".json"))
40            else {
41                continue;
42            };
43            let manifest = self.load_manifest(generation)?;
44            let keys = manifest
45                .plane_keys()
46                .filter_map(|(_, key)| {
47                    if key.len() != 64 {
48                        return None;
49                    }
50                    let bytes = (0..64)
51                        .step_by(2)
52                        .map(|offset| u8::from_str_radix(&key[offset..offset + 2], 16).ok())
53                        .collect::<Option<Vec<_>>>()?;
54                    bytes.try_into().ok()
55                })
56                .collect();
57            result.insert(generation.to_owned(), keys);
58        }
59        Ok(result)
60    }
61
62    /// Remove generation files only after checking both durable publication and
63    /// liveness. A dead assembler can leave a derived file before any manifest.
64    pub fn sweep_generations(&self) -> Result<usize> {
65        let current = self.current_generation()?;
66        let mut generations = std::collections::BTreeSet::new();
67        for entry in fs::read_dir(self.view_dir())? {
68            let entry = entry?;
69            let name = entry.file_name();
70            let Some(name) = name.to_str() else { continue };
71            let generation = name
72                .strip_prefix("derived-")
73                .and_then(|s| s.strip_suffix(".sqlite"))
74                .or_else(|| {
75                    name.strip_prefix("manifest-")
76                        .and_then(|s| s.strip_suffix(".json"))
77                })
78                .or_else(|| {
79                    name.strip_prefix("trigram-")
80                        .and_then(|s| s.strip_suffix(".bin"))
81                })
82                .or_else(|| {
83                    name.strip_prefix(".manifest-")
84                        .and_then(|s| s.split_once(".json.tmp.").map(|(generation, _)| generation))
85                });
86            if let Some(generation) = generation {
87                generations.insert(generation.to_owned());
88            }
89        }
90        let mut removed = 0;
91        for generation in generations {
92            if current.as_deref() == Some(&generation) {
93                continue;
94            }
95            let (metadata_path, keys_path) = crate::pins::pin_paths(self.view_dir(), &generation);
96            if metadata_path.exists() {
97                let Ok(bytes) = fs::read(&metadata_path) else {
98                    continue;
99                };
100                let Ok(metadata) = serde_json::from_slice::<crate::pins::PinMetadata>(&bytes)
101                else {
102                    continue;
103                };
104                if crate::pins::owner_is_live(&metadata.owner)
105                    && crate::pins::now_ms().saturating_sub(metadata.renewed_at)
106                        <= crate::pins::PIN_TTL_MS
107                {
108                    continue;
109                }
110                let _ = fs::remove_file(metadata_path);
111                let _ = fs::remove_file(keys_path);
112            }
113            if crate::root_cache::sweep_read_markers(self.view_dir(), &generation).protected {
114                continue;
115            }
116            // Recheck after pin inspection: a publisher keeps its assembly pin
117            // until its committed pointer is visible.
118            if self.current_generation()?.as_deref() == Some(&generation) {
119                continue;
120            }
121            self.remove_generation_files(&generation);
122            removed += 1;
123        }
124        Ok(removed)
125    }
126
127    pub(super) fn remove_generation_files(&self, generation: &str) {
128        if super::validate_generation(generation).is_err() {
129            return;
130        }
131        let temporary_prefix = format!(".manifest-{generation}.json.tmp.");
132        if let Ok(entries) = fs::read_dir(self.view_dir()) {
133            for entry in entries.flatten() {
134                if entry
135                    .file_name()
136                    .to_str()
137                    .is_some_and(|name| name.starts_with(&temporary_prefix))
138                {
139                    let _ = fs::remove_file(entry.path());
140                }
141            }
142        }
143        for path in [
144            self.derived_path(generation),
145            self.trigram_path(generation),
146            self.manifest_path(generation),
147        ]
148        .into_iter()
149        .flatten()
150        {
151            for suffix in ["", "-wal", "-shm"] {
152                let mut name = path.as_os_str().to_owned();
153                name.push(suffix);
154                let _ = fs::remove_file(PathBuf::from(name));
155            }
156        }
157    }
158}
159
160#[derive(Clone)]
161struct DeferredCheckpointJob {
162    path: PathBuf,
163    cancelled: Arc<AtomicBool>,
164}
165
166static DEFERRED_CHECKPOINTS: OnceLock<Mutex<HashMap<PathBuf, DeferredCheckpointJob>>> =
167    OnceLock::new();
168static CHECKPOINT_LOCKS: OnceLock<Mutex<HashMap<PathBuf, Weak<Mutex<()>>>>> = OnceLock::new();
169const DEFERRED_CHECKPOINT_IDLE_DELAY: Duration = Duration::from_millis(250);
170
171fn checkpoint_lock(path: &Path) -> Arc<Mutex<()>> {
172    let mut locks = CHECKPOINT_LOCKS
173        .get_or_init(|| Mutex::new(HashMap::new()))
174        .lock()
175        .unwrap_or_else(std::sync::PoisonError::into_inner);
176    if let Some(lock) = locks.get(path).and_then(Weak::upgrade) {
177        return lock;
178    }
179    let lock = Arc::new(Mutex::new(()));
180    locks.insert(path.to_path_buf(), Arc::downgrade(&lock));
181    lock
182}
183
184fn checkpoint_derived(path: &Path, connection: Option<&Connection>) -> Result<()> {
185    let lock = checkpoint_lock(path);
186    let _guard = lock
187        .lock()
188        .unwrap_or_else(std::sync::PoisonError::into_inner);
189    let owned;
190    let connection = if let Some(connection) = connection {
191        connection
192    } else {
193        owned = Connection::open(path)?;
194        &owned
195    };
196    connection.busy_timeout(Duration::from_secs(5))?;
197    connection.pragma_update(None, "synchronous", "FULL")?;
198    let (busy, log_frames, checkpointed_frames): (i64, i64, i64) =
199        connection.query_row("PRAGMA wal_checkpoint(TRUNCATE)", [], |row| {
200            Ok((row.get(0)?, row.get(1)?, row.get(2)?))
201        })?;
202    if busy != 0 {
203        return Err(ViewError::InvalidManifest(format!(
204            "derived WAL checkpoint remained busy: path={} log_frames={} checkpointed_frames={}",
205            path.display(),
206            log_frames,
207            checkpointed_frames
208        )));
209    }
210    super::sync_file(path)?;
211    super::sync_parent(path)?;
212    Ok(())
213}
214
215/// Run the generation-sized checkpoint after pointer publication. A later
216/// publication cancels a not-yet-started obsolete job; its clone has already
217/// forced the source checkpoint through [`clone_derived`].
218pub(super) fn schedule_derived_checkpoint(path: PathBuf, connection: Connection) {
219    let key = path.parent().unwrap_or(&path).to_path_buf();
220    let cancelled = Arc::new(AtomicBool::new(false));
221    let job = DeferredCheckpointJob {
222        path: path.clone(),
223        cancelled: Arc::clone(&cancelled),
224    };
225    if let Some(previous) = DEFERRED_CHECKPOINTS
226        .get_or_init(|| Mutex::new(HashMap::new()))
227        .lock()
228        .unwrap_or_else(std::sync::PoisonError::into_inner)
229        .insert(key.clone(), job)
230    {
231        previous.cancelled.store(true, Ordering::Release);
232        log::debug!(
233            "view derived checkpoint superseded path={}",
234            previous.path.display()
235        );
236    }
237    let path_for_error = path.clone();
238    let key_for_error = key.clone();
239    let cancelled_for_error = Arc::clone(&cancelled);
240    let spawn = std::thread::Builder::new()
241        .name("aft-view-checkpoint".to_owned())
242        .spawn(move || {
243            std::thread::sleep(DEFERRED_CHECKPOINT_IDLE_DELAY);
244            let started = Instant::now();
245            if !cancelled.load(Ordering::Acquire) {
246                match checkpoint_derived(&path, Some(&connection)) {
247                    Ok(()) => log::info!(
248                        "view derived checkpoint completed ms={} path={}",
249                        started.elapsed().as_millis(),
250                        path.display()
251                    ),
252                    Err(error) => log::warn!(
253                        "view derived checkpoint deferred to next clone path={} error={}",
254                        path.display(),
255                        error
256                    ),
257                }
258            }
259            let mut jobs = DEFERRED_CHECKPOINTS
260                .get_or_init(|| Mutex::new(HashMap::new()))
261                .lock()
262                .unwrap_or_else(std::sync::PoisonError::into_inner);
263            if jobs
264                .get(&key)
265                .is_some_and(|current| Arc::ptr_eq(&current.cancelled, &cancelled))
266            {
267                jobs.remove(&key);
268            }
269        });
270    if let Err(error) = spawn {
271        let mut jobs = DEFERRED_CHECKPOINTS
272            .get_or_init(|| Mutex::new(HashMap::new()))
273            .lock()
274            .unwrap_or_else(std::sync::PoisonError::into_inner);
275        if jobs
276            .get(&key_for_error)
277            .is_some_and(|current| Arc::ptr_eq(&current.cancelled, &cancelled_for_error))
278        {
279            jobs.remove(&key_for_error);
280        }
281        log::warn!(
282            "view derived checkpoint worker unavailable; next clone will checkpoint path={} error={}",
283            path_for_error.display(),
284            error
285        );
286    }
287}
288
289/// Checkpoint the source before copying its main file. This also recovers a
290/// commit-durable WAL left by a process that exited before detached maintenance.
291pub(super) fn clone_derived(source: &Path, destination: &Path) -> Result<()> {
292    checkpoint_derived(source, None)?;
293    let started = Instant::now();
294    let mechanism = if try_clone(source, destination) {
295        if cfg!(target_os = "macos") {
296            "clonefile"
297        } else {
298            "reflink"
299        }
300    } else {
301        let _ = fs::remove_file(destination);
302        fs::copy(source, destination)?;
303        "copy"
304    };
305    log::info!(
306        "view derived clone mechanism={} ms={} source={} destination={}",
307        mechanism,
308        started.elapsed().as_millis(),
309        source.display(),
310        destination.display()
311    );
312    Ok(())
313}
314
315#[cfg(test)]
316mod tests {
317    use super::*;
318
319    #[test]
320    fn deferred_checkpoint_runs_after_the_publication_path_returns() {
321        let directory = tempfile::tempdir().unwrap();
322        let source = directory.path().join("source.sqlite");
323        let connection = Connection::open(&source).unwrap();
324        connection
325            .pragma_update(None, "journal_mode", "WAL")
326            .unwrap();
327        connection
328            .pragma_update(None, "synchronous", "FULL")
329            .unwrap();
330        connection
331            .pragma_update(None, "wal_autocheckpoint", 0)
332            .unwrap();
333        connection
334            .execute_batch(
335                "CREATE TABLE state (value TEXT NOT NULL);\
336                 INSERT INTO state VALUES ('durable');",
337            )
338            .unwrap();
339        let wal = PathBuf::from(format!("{}-wal", source.display()));
340        assert!(fs::metadata(&wal).unwrap().len() > 0);
341
342        schedule_derived_checkpoint(source.clone(), connection);
343
344        assert!(
345            fs::metadata(&wal).is_ok_and(|metadata| metadata.len() > 0),
346            "checkpoint ran synchronously on the publication path"
347        );
348        let deadline = Instant::now() + Duration::from_secs(5);
349        while fs::metadata(&wal).is_ok_and(|metadata| metadata.len() > 0) {
350            assert!(
351                Instant::now() < deadline,
352                "detached derived checkpoint did not finish"
353            );
354            std::thread::sleep(Duration::from_millis(10));
355        }
356        assert_eq!(
357            Connection::open(&source)
358                .unwrap()
359                .query_row("SELECT value FROM state", [], |row| row.get::<_, String>(0))
360                .unwrap(),
361            "durable"
362        );
363    }
364
365    #[test]
366    fn clone_checkpoints_committed_wal_before_copying_the_main_file() {
367        let directory = tempfile::tempdir().unwrap();
368        let source = directory.path().join("source.sqlite");
369        let destination = directory.path().join("destination.sqlite");
370        let connection = Connection::open(&source).unwrap();
371        connection
372            .pragma_update(None, "journal_mode", "WAL")
373            .unwrap();
374        connection
375            .pragma_update(None, "synchronous", "FULL")
376            .unwrap();
377        connection
378            .pragma_update(None, "wal_autocheckpoint", 0)
379            .unwrap();
380        connection
381            .execute_batch(
382                "CREATE TABLE state (value TEXT NOT NULL);\
383                 INSERT INTO state VALUES ('committed-in-wal');",
384            )
385            .unwrap();
386        let wal = PathBuf::from(format!("{}-wal", source.display()));
387        assert!(fs::metadata(&wal).unwrap().len() > 0);
388
389        clone_derived(&source, &destination).unwrap();
390
391        assert_eq!(
392            Connection::open(&destination)
393                .unwrap()
394                .query_row("SELECT value FROM state", [], |row| row.get::<_, String>(0))
395                .unwrap(),
396            "committed-in-wal"
397        );
398    }
399}
400
401#[cfg(target_os = "macos")]
402fn try_clone(source: &Path, destination: &Path) -> bool {
403    use std::{ffi::CString, os::unix::ffi::OsStrExt};
404    let (Ok(source), Ok(destination)) = (
405        CString::new(source.as_os_str().as_bytes()),
406        CString::new(destination.as_os_str().as_bytes()),
407    ) else {
408        return false;
409    };
410    // Both C strings remain alive for the duration of clonefile.
411    unsafe { libc::clonefile(source.as_ptr(), destination.as_ptr(), 0) == 0 }
412}
413
414#[cfg(target_os = "linux")]
415fn try_clone(source: &Path, destination: &Path) -> bool {
416    use std::os::fd::AsRawFd;
417    let (Ok(source), Ok(destination)) = (
418        fs::File::open(source),
419        fs::OpenOptions::new()
420            .write(true)
421            .create_new(true)
422            .open(destination),
423    ) else {
424        return false;
425    };
426    // FICLONE takes the source descriptor as its third argument.
427    unsafe {
428        libc::ioctl(
429            destination.as_raw_fd(),
430            0x40049409 as libc::c_ulong,
431            source.as_raw_fd(),
432        ) == 0
433    }
434}
435
436#[cfg(not(any(target_os = "linux", target_os = "macos")))]
437fn try_clone(_source: &Path, _destination: &Path) -> bool {
438    false
439}
aft/views/generation.rs

aft/views/
generation.rs
