haematite_cli/serve/storage/
dead.rs

1//! A marker for each discarded generation whose history rows are not all
2//! released (docs/design/HAEM-DISCARDED-ROWS-RELEASE-2026-10-06.md). A discard
3//! writes its rows and its receipt before it removes the branch, and nothing
4//! released a name with no record, so every discard kept its branch's last
5//! versions for good. The marker lets release find them in one range read,
6//! never by reading the history.
7
8use std::num::NonZeroUsize;
9
10use haematite::branch::Timestamp;
11use haematite::branch::history::{Key, Row};
12use haematite::db::ReadOnlyDatabase;
13use haematite::tree::Hash;
14use serde::{Deserialize, Serialize};
15
16use super::{BranchRefRecord, Failure, SERVICE, Storage, branch_name};
17use crate::serve::protocol::Version;
18
19const DEAD_PREFIX: &[u8] = b"dead-generation\0";
20
21/// One discarded generation: the name, its last version, and when it was
22/// discarded, which release counts as the version that superseded `last`.
23#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
24#[serde(deny_unknown_fields)]
25pub struct Dead {
26    pub branch: String,
27    pub last: Version,
28    pub discarded_at: Timestamp,
29}
30
31/// The marker key: prefix, the name's hash, then the generation big-endian,
32/// so one name's generations sort oldest first.
33pub(super) fn dead_key(branch: &str, generation: u64) -> Vec<u8> {
34    let mut key = name_prefix(branch);
35    key.extend_from_slice(&generation.to_be_bytes());
36    key
37}
38
39fn name_prefix(branch: &str) -> Vec<u8> {
40    let mut key = Vec::with_capacity(DEAD_PREFIX.len() + 32 + 8);
41    key.extend_from_slice(DEAD_PREFIX);
42    key.extend_from_slice(Hash::of(branch.as_bytes()).as_bytes());
43    key
44}
45
46/// The first key past every key that starts with `prefix`.
47fn past(prefix: &[u8]) -> Vec<u8> {
48    let mut end = prefix.to_vec();
49    if let Some(last) = end.last_mut() {
50        *last = last.saturating_add(1);
51    }
52    end
53}
54
55/// The markers from `start` up to `end`, at most `limit` from each shard,
56/// oldest key first.
57fn page(
58    reader: &ReadOnlyDatabase,
59    service: &BranchRefRecord,
60    (start, end): (&[u8], &[u8]),
61    limit: NonZeroUsize,
62) -> Result<Vec<(Vec<u8>, Dead)>, Failure> {
63    let mut found = Vec::new();
64    for shard in &service.shards {
65        let read = reader
66            .range_page_at(shard.shard_id, shard.head, start, end, limit)
67            .map_err(Failure::observer)?;
68        for (key, bytes) in read.entries {
69            let dead = serde_json::from_slice(&bytes)
70                .map_err(|error| Failure::new("dead_generation_corrupt", error.to_string()))?;
71            found.push((key, dead));
72        }
73    }
74    found.sort_unstable_by(|(left, _), (right, _)| left.cmp(right));
75    Ok(found)
76}
77
78/// `branch`'s discarded generations not yet emptied, oldest first, at most
79/// `limit` (DR-6): one range read of its markers on each shard.
80///
81/// # Errors
82/// `dead_generation_corrupt` for a marker that does not decode; storage failures.
83pub(super) fn of_name(
84    reader: &ReadOnlyDatabase,
85    service: &BranchRefRecord,
86    branch: &str,
87    limit: NonZeroUsize,
88) -> Result<Vec<Dead>, Failure> {
89    let start = name_prefix(branch);
90    let end = past(&start);
91    let mut found = page(reader, service, (&start, &end), limit)?;
92    found.truncate(limit.get());
93    Ok(found.into_iter().map(|(_key, dead)| dead).collect())
94}
95
96/// `branch`'s marker for `generation`, if it is still there: a range read of
97/// at most one entry, its key, on each shard.
98///
99/// # Errors
100/// `dead_generation_corrupt`; storage failures.
101pub(super) fn marker(
102    reader: &ReadOnlyDatabase,
103    service: &BranchRefRecord,
104    branch: &str,
105    generation: u64,
106) -> Result<Option<Dead>, Failure> {
107    #[cfg(test)]
108    super::history::MARKER_READS.with(|reads| reads.set(reads.get() + 1));
109    let key = dead_key(branch, generation);
110    let end = past(&key);
111    let found = page(reader, service, (&key, &end), NonZeroUsize::MIN)?;
112    Ok(found
113        .into_iter()
114        .find(|(found, _dead)| *found == key)
115        .map(|(_key, dead)| dead))
116}
117
118/// The version after the newest row read, as a row at `at` written
119/// `written`: the discard, the version after a generation's last, or a live
120/// branch's next version (DR-3). It tells release when the newest row was
121/// superseded. It names no roots and is never written.
122pub(super) fn discard_row(branch: &str, at: Version, written: Timestamp) -> Row {
123    Row {
124        key: Key {
125            branch: Hash::of(branch.as_bytes()),
126            generation: at.generation,
127            sequence: at.sequence,
128        },
129        timestamp: written,
130        shards: Vec::new(),
131    }
132}
133
134/// The markers a release taking `taken` (oldest first) ends: each generation
135/// among them whose marker's last version is taken, so its last row goes.
136/// One point read for each generation taken, never the history.
137///
138/// # Errors
139/// `dead_generation_corrupt`; storage failures.
140pub(super) fn ended(
141    reader: &ReadOnlyDatabase,
142    service: &BranchRefRecord,
143    branch: &str,
144    taken: &[Version],
145) -> Result<Vec<Vec<u8>>, Failure> {
146    let mut generations: Vec<u64> = taken.iter().map(|at| at.generation).collect();
147    generations.dedup();
148    let mut ended = Vec::new();
149    for generation in generations {
150        if let Some(dead) = marker(reader, service, branch, generation)?
151            && taken.contains(&dead.last)
152        {
153            ended.push(dead_key(branch, generation));
154        }
155    }
156    Ok(ended)
157}
158
159/// One page of the markers, for the release driver to walk them all in
160/// turn (outside review 15): a page that always started at the first marker
161/// would ask the same held generations every time and never reach the rest.
162#[derive(Debug, Clone, PartialEq, Eq)]
163pub struct DeadPage {
164    /// The names on the page with no live record, each once.
165    pub names: Vec<String>,
166    /// The last marker key on the page: where the next page starts after.
167    pub last: Option<Vec<u8>>,
168    /// The page reached the last marker; the next starts at the first.
169    pub end: bool,
170}
171
172impl Storage {
173    /// At most `limit` markers after the key `after` (from the first when
174    /// `None`), oldest key first: one range read of at most `limit` + 1 on
175    /// each shard, never the history.
176    ///
177    /// # Errors
178    /// `reference_missing`; `dead_generation_corrupt`; storage failures.
179    pub fn dead_page(
180        &self,
181        after: Option<&[u8]>,
182        limit: NonZeroUsize,
183    ) -> Result<DeadPage, Failure> {
184        let service = self
185            .refs
186            .get(SERVICE)
187            .ok_or_else(|| Failure::new("reference_missing", "missing service namespace"))?;
188        // The least key past `after`.
189        let start = after.map_or_else(
190            || DEAD_PREFIX.to_vec(),
191            |key| {
192                let mut start = key.to_vec();
193                start.push(0);
194                start
195            },
196        );
197        let read = NonZeroUsize::new(limit.get().saturating_add(1)).unwrap_or(NonZeroUsize::MAX);
198        let mut found = page(&self.reader, service, (&start, &past(DEAD_PREFIX)), read)?;
199        let end = found.len() <= limit.get();
200        found.truncate(limit.get());
201        let last = found.last().map(|(key, _dead)| key.clone());
202        let mut names: Vec<String> = found
203            .into_iter()
204            .map(|(_key, dead)| dead.branch)
205            .filter(|name| self.refs.get(&branch_name(name)).is_none())
206            .collect();
207        names.dedup();
208        Ok(DeadPage { names, last, end })
209    }
210}
haematite_cli/serve/storage/dead.rs

haematite_cli/serve/storage/
dead.rs
