nodedb 0.4.0

Local-first, real-time, edge-to-cloud hybrid database for multi-modal workloads
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130

// SPDX-License-Identifier: BUSL-1.1

//! Spatial engine reclaim — unlink per-collection R*-tree checkpoint
//! + docmap files.
//!
//! Checkpoint layout (see `spatial_checkpoint.rs`):
//! `{data_dir}/spatial-ckpt/{db}_{tid}_{enc(coll)}_{enc(field)}.ckpt` + `.docmap`.
//! The filename prefix is built by the SAME encoder the write path uses
//! ([`spatial_checkpoint_prefix`]) so the `starts_with` match can never drift
//! from the on-disk names.

use std::path::Path;

use tracing::debug;

use super::{ReclaimError, ReclaimStats, Result};
use crate::data::executor::spatial_checkpoint::spatial_checkpoint_prefix;

/// Unlink every spatial checkpoint + docmap file for
/// `(database_id, tenant_id, collection)`. Returns stats; idempotent.
pub fn reclaim_spatial_checkpoints(
    data_dir: &Path,
    database_id: u64,
    tenant_id: u64,
    collection: &str,
) -> Result<ReclaimStats> {
    let ckpt_dir = data_dir.join("spatial-ckpt");
    let entries = match std::fs::read_dir(&ckpt_dir) {
        Ok(entries) => entries,
        Err(source) if source.kind() == std::io::ErrorKind::NotFound => {
            return Ok(ReclaimStats::default());
        }
        Err(source) => {
            return Err(ReclaimError::Io {
                operation: "read spatial checkpoint directory",
                path: ckpt_dir,
                source,
            });
        }
    };

    // Build the prefix via the shared encoder so it always matches the
    // filenames produced by `checkpoint_spatial_indexes`.
    let prefix = spatial_checkpoint_prefix(database_id, tenant_id, collection);

    let mut stats = ReclaimStats::default();
    for entry in entries {
        let entry = entry.map_err(|source| ReclaimError::Io {
            operation: "read spatial checkpoint entry",
            path: ckpt_dir.clone(),
            source,
        })?;
        let path = entry.path();
        let Some(name) = path.file_name().and_then(|s| s.to_str()) else {
            continue;
        };
        if !name.starts_with(&prefix) {
            continue;
        }
        // Only sweep `.ckpt`, `.ckpt.tmp`, `.docmap`, `.docmap.tmp`.
        let is_ours = name.ends_with(".ckpt")
            || name.ends_with(".ckpt.tmp")
            || name.ends_with(".docmap")
            || name.ends_with(".docmap.tmp");
        if !is_ours {
            continue;
        }
        let size = entry.metadata().map(|m| m.len()).unwrap_or(0);
        match std::fs::remove_file(&path) {
            Ok(()) => {
                stats.files_unlinked = stats.files_unlinked.saturating_add(1);
                stats.bytes_freed = stats.bytes_freed.saturating_add(size);
                debug!(path = %path.display(), size, "spatial reclaim: unlinked");
            }
            Err(source) if source.kind() == std::io::ErrorKind::NotFound => {}
            Err(source) => {
                return Err(ReclaimError::Io {
                    operation: "unlink spatial checkpoint",
                    path,
                    source,
                });
            }
        }
    }
    Ok(stats)
}

#[cfg(test)]
mod tests {
    use super::*;
    use tempfile::TempDir;

    fn write(path: &Path, bytes: &[u8]) {
        if let Some(parent) = path.parent() {
            std::fs::create_dir_all(parent).unwrap();
        }
        std::fs::write(path, bytes).unwrap();
    }

    #[test]
    fn unlinks_ckpt_and_docmap_for_matching_field_indexes() {
        let tmp = TempDir::new().unwrap();
        let base = tmp.path();
        let ckpt = base.join("spatial-ckpt");
        // New-format names: {db}_{tid}_{coll}_{field}.
        write(&ckpt.join("0_1_places_geom.ckpt"), b"x");
        write(&ckpt.join("0_1_places_geom.docmap"), b"yy");
        write(&ckpt.join("0_1_places_home.ckpt"), b"zzz");
        // Keep: different collection.
        write(&ckpt.join("0_1_stores_geom.ckpt"), b"keep");
        // Keep: different tenant.
        write(&ckpt.join("0_2_places_geom.ckpt"), b"keep2");
        // Keep: different database.
        write(&ckpt.join("1_1_places_geom.ckpt"), b"keep3");

        let stats = reclaim_spatial_checkpoints(base, 0, 1, "places").unwrap();
        assert_eq!(stats.files_unlinked, 3);
        assert_eq!(stats.bytes_freed, 1 + 2 + 3);
        assert!(ckpt.join("0_1_stores_geom.ckpt").exists());
        assert!(ckpt.join("0_2_places_geom.ckpt").exists());
        assert!(ckpt.join("1_1_places_geom.ckpt").exists());
    }

    #[test]
    fn empty_dir_is_noop() {
        let tmp = TempDir::new().unwrap();
        let s = reclaim_spatial_checkpoints(tmp.path(), 0, 1, "x").unwrap();
        assert_eq!(s.files_unlinked, 0);
    }
}