genegraph_storage/
generations.rs

1//! # Transactional generations (genegraph-storage #93, RFC #81 phase P5).
2//!
3//! Artifact generations are **immutable**: every publish (build, append,
4//! recompute) mints a fresh generation `{logical}__g{N}` — its artifacts are
5//! written once and never overwritten — and commits by atomically publishing
6//! the generation's metadata JSON (the single commit pointer, mirroring the
7//! snapshot/consensus-index pattern of segmented logs).
8//!
9//! Invariants:
10//! - A generation without a metadata file **was never committed**: readers
11//!   and discovery ignore it; the sweep API sees it for garbage collection.
12//! - The commit itself is a single atomic filesystem operation
13//!   ([`write_json_atomic`]): readers observe either the previous file or
14//!   the complete new one, never a partial write.
15//! - `__g{digits}` at the end of an instance name is reserved.
16
17use std::path::{Path, PathBuf};
18
19use crate::{StorageError, StorageResult};
20
21/// Separator between a logical dataset name and its generation number.
22pub const GENERATION_SEP: &str = "__g";
23
24/// Logical dataset name: strips a trailing `__g{digits}` generation suffix.
25pub fn logical_name(name: &str) -> &str {
26    if let Some(pos) = name.rfind(GENERATION_SEP) {
27        let suffix = &name[pos + GENERATION_SEP.len()..];
28        if !suffix.is_empty() && suffix.bytes().all(|b| b.is_ascii_digit()) {
29            return &name[..pos];
30        }
31    }
32    name
33}
34
35/// Generation-qualified instance name: `{logical}__g{gen}`.
36pub fn generation_name(logical: &str, generation: u64) -> String {
37    format!("{logical}{GENERATION_SEP}{generation}")
38}
39
40/// Parse the generation number out of a gen-qualified instance name.
41/// Returns `None` for plain logical names or malformed suffixes.
42pub fn parse_generation(name: &str) -> Option<u64> {
43    let pos = name.rfind(GENERATION_SEP)?;
44    let suffix = &name[pos + GENERATION_SEP.len()..];
45    if suffix.is_empty() || !suffix.bytes().all(|b| b.is_ascii_digit()) {
46        return None;
47    }
48    suffix.parse().ok()
49}
50
51/// A committed generation: its number and the metadata file that pins it.
52#[derive(Debug, Clone, PartialEq, Eq)]
53pub struct GenerationInfo {
54    pub generation: u64,
55    pub metadata_path: PathBuf,
56}
57
58/// Committed generations of `logical` under `base`, ascending by generation
59/// number. A generation is committed iff its `{logical}__g{N}_metadata.json`
60/// exists; pre-commit crash residue is invisible here.
61pub async fn list_generations(base: &Path, logical: &str) -> StorageResult<Vec<GenerationInfo>> {
62    let mut out = Vec::new();
63    for (generation, metadata_path) in scan_generation_paths(base, logical).await? {
64        if metadata_path.is_file() {
65            out.push(GenerationInfo {
66                generation,
67                metadata_path,
68            });
69        }
70    }
71    out.sort_by_key(|g| g.generation);
72    Ok(out)
73}
74
75/// Generation numbers with **any** artifact present under `base`, committed
76/// or not (orphans from a pre-commit crash included), ascending.
77pub async fn list_artifact_generations(base: &Path, logical: &str) -> StorageResult<Vec<u64>> {
78    let mut gens = std::collections::BTreeSet::new();
79    let mut rd = tokio::fs::read_dir(base)
80        .await
81        .map_err(|e| StorageError::Io(e.to_string()))?;
82    let artifact_prefix = format!("{logical}{GENERATION_SEP}");
83    while let Some(entry) = rd
84        .next_entry()
85        .await
86        .map_err(|e| StorageError::Io(e.to_string()))?
87    {
88        let name = entry.file_name();
89        let name = name.to_string_lossy();
90        let Some(rest) = name.strip_prefix(&artifact_prefix) else {
91            continue;
92        };
93        let Some(num) = rest.split('_').next() else {
94            continue;
95        };
96        if !num.is_empty() && num.bytes().all(|b| b.is_ascii_digit()) {
97            gens.insert(num.parse::<u64>().unwrap_or(u64::MAX));
98        }
99    }
100    Ok(gens.into_iter().collect())
101}
102
103/// Delete every file of a generation — artifacts and, if present, the
104/// metadata commit pointer. Prefix matching is exact on
105/// `{logical}__g{gen}_`, so sibling datasets (`ds` vs `ds2`) are untouched.
106/// Safe to call on orphaned generations (no metadata) and on missing ones.
107pub async fn delete_generation(base: &Path, logical: &str, generation: u64) -> StorageResult<()> {
108    let prefix = format!("{logical}{GENERATION_SEP}{generation}_");
109    let mut rd = tokio::fs::read_dir(base)
110        .await
111        .map_err(|e| StorageError::Io(e.to_string()))?;
112    while let Some(entry) = rd
113        .next_entry()
114        .await
115        .map_err(|e| StorageError::Io(e.to_string()))?
116    {
117        if !entry.file_name().to_string_lossy().starts_with(&prefix) {
118            continue;
119        }
120        let path = entry.path();
121        if path.is_dir() {
122            tokio::fs::remove_dir_all(&path)
123                .await
124                .map_err(|e| StorageError::Io(e.to_string()))?;
125        } else {
126            tokio::fs::remove_file(&path)
127                .await
128                .map_err(|e| StorageError::Io(e.to_string()))?;
129        }
130    }
131    Ok(())
132}
133
134/// Atomic JSON publish: write to `{path}.tmp`, fsync, rename over `path`.
135///
136/// The rename is the single commit point (POSIX-atomic within a directory):
137/// concurrent readers observe either the previous file or the complete new
138/// one, never a truncated write. This is the ONLY sanctioned way to publish
139/// a metadata file.
140pub fn write_json_atomic(path: &Path, contents: &str) -> StorageResult<()> {
141    use std::io::Write;
142
143    let tmp = path.with_extension("json.tmp");
144    if let Some(parent) = path.parent() {
145        std::fs::create_dir_all(parent).map_err(|e| StorageError::Io(e.to_string()))?;
146    }
147    {
148        let mut f = std::fs::File::create(&tmp).map_err(|e| StorageError::Io(e.to_string()))?;
149        f.write_all(contents.as_bytes())
150            .map_err(|e| StorageError::Io(e.to_string()))?;
151        f.sync_all().map_err(|e| StorageError::Io(e.to_string()))?;
152    }
153    std::fs::rename(&tmp, path).map_err(|e| StorageError::Io(e.to_string()))?;
154    Ok(())
155}
156
157/// `(generation, metadata_path)` pairs implied by `{logical}__g{N}_metadata.json`
158/// names under `base` (the file itself may or may not exist).
159async fn scan_generation_paths(base: &Path, logical: &str) -> StorageResult<Vec<(u64, PathBuf)>> {
160    let mut out = Vec::new();
161    let mut rd = tokio::fs::read_dir(base)
162        .await
163        .map_err(|e| StorageError::Io(e.to_string()))?;
164    let prefix = format!("{logical}{GENERATION_SEP}");
165    while let Some(entry) = rd
166        .next_entry()
167        .await
168        .map_err(|e| StorageError::Io(e.to_string()))?
169    {
170        let name = entry.file_name();
171        let name = name.to_string_lossy().into_owned();
172        let Some(rest) = name.strip_prefix(&prefix) else {
173            continue;
174        };
175        let Some(meta) = rest.strip_suffix("_metadata.json") else {
176            continue;
177        };
178        if !meta.is_empty() && meta.bytes().all(|b| b.is_ascii_digit()) {
179            out.push((meta.parse().unwrap_or(u64::MAX), entry.path()));
180        }
181    }
182    Ok(out)
183}
genegraph_storage/generations.rs

genegraph_storage/
generations.rs
