haematite-cli 0.21.0

//! A marker for each discarded generation whose history rows are not all
//! released (docs/design/HAEM-DISCARDED-ROWS-RELEASE-2026-10-06.md). A discard
//! writes its rows and its receipt before it removes the branch, and nothing
//! released a name with no record, so every discard kept its branch's last
//! versions for good. The marker lets release find them in one range read,
//! never by reading the history.

use std::num::NonZeroUsize;

use haematite::branch::Timestamp;
use haematite::branch::history::{Key, Row};
use haematite::db::ReadOnlyDatabase;
use haematite::tree::Hash;
use serde::{Deserialize, Serialize};

use super::{BranchRefRecord, Failure, SERVICE, Storage, branch_name};
use crate::serve::protocol::Version;

const DEAD_PREFIX: &[u8] = b"dead-generation\0";

/// One discarded generation: the name, its last version, and when it was
/// discarded, which release counts as the version that superseded `last`.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Dead {
    pub branch: String,
    pub last: Version,
    pub discarded_at: Timestamp,
}

/// The marker key: prefix, the name's hash, then the generation big-endian,
/// so one name's generations sort oldest first.
pub(super) fn dead_key(branch: &str, generation: u64) -> Vec<u8> {
    let mut key = name_prefix(branch);
    key.extend_from_slice(&generation.to_be_bytes());
    key
}

fn name_prefix(branch: &str) -> Vec<u8> {
    let mut key = Vec::with_capacity(DEAD_PREFIX.len() + 32 + 8);
    key.extend_from_slice(DEAD_PREFIX);
    key.extend_from_slice(Hash::of(branch.as_bytes()).as_bytes());
    key
}

/// The first key past every key that starts with `prefix`.
fn past(prefix: &[u8]) -> Vec<u8> {
    let mut end = prefix.to_vec();
    if let Some(last) = end.last_mut() {
        *last = last.saturating_add(1);
    }
    end
}

/// The markers from `start` up to `end`, at most `limit` from each shard,
/// oldest key first.
fn page(
    reader: &ReadOnlyDatabase,
    service: &BranchRefRecord,
    (start, end): (&[u8], &[u8]),
    limit: NonZeroUsize,
) -> Result<Vec<(Vec<u8>, Dead)>, Failure> {
    let mut found = Vec::new();
    for shard in &service.shards {
        let read = reader
            .range_page_at(shard.shard_id, shard.head, start, end, limit)
            .map_err(Failure::observer)?;
        for (key, bytes) in read.entries {
            let dead = serde_json::from_slice(&bytes)
                .map_err(|error| Failure::new("dead_generation_corrupt", error.to_string()))?;
            found.push((key, dead));
        }
    }
    found.sort_unstable_by(|(left, _), (right, _)| left.cmp(right));
    Ok(found)
}

/// `branch`'s discarded generations not yet emptied, oldest first, at most
/// `limit` (DR-6): one range read of its markers on each shard.
///
/// # Errors
/// `dead_generation_corrupt` for a marker that does not decode; storage failures.
pub(super) fn of_name(
    reader: &ReadOnlyDatabase,
    service: &BranchRefRecord,
    branch: &str,
    limit: NonZeroUsize,
) -> Result<Vec<Dead>, Failure> {
    let start = name_prefix(branch);
    let end = past(&start);
    let mut found = page(reader, service, (&start, &end), limit)?;
    found.truncate(limit.get());
    Ok(found.into_iter().map(|(_key, dead)| dead).collect())
}

/// `branch`'s marker for `generation`, if it is still there: a range read of
/// at most one entry, its key, on each shard.
///
/// # Errors
/// `dead_generation_corrupt`; storage failures.
pub(super) fn marker(
    reader: &ReadOnlyDatabase,
    service: &BranchRefRecord,
    branch: &str,
    generation: u64,
) -> Result<Option<Dead>, Failure> {
    #[cfg(test)]
    super::history::MARKER_READS.with(|reads| reads.set(reads.get() + 1));
    let key = dead_key(branch, generation);
    let end = past(&key);
    let found = page(reader, service, (&key, &end), NonZeroUsize::MIN)?;
    Ok(found
        .into_iter()
        .find(|(found, _dead)| *found == key)
        .map(|(_key, dead)| dead))
}

/// The version after the newest row read, as a row at `at` written
/// `written`: the discard, the version after a generation's last, or a live
/// branch's next version (DR-3). It tells release when the newest row was
/// superseded. It names no roots and is never written.
pub(super) fn discard_row(branch: &str, at: Version, written: Timestamp) -> Row {
    Row {
        key: Key {
            branch: Hash::of(branch.as_bytes()),
            generation: at.generation,
            sequence: at.sequence,
        },
        timestamp: written,
        shards: Vec::new(),
    }
}

/// The markers a release taking `taken` (oldest first) ends: each generation
/// among them whose marker's last version is taken, so its last row goes.
/// One point read for each generation taken, never the history.
///
/// # Errors
/// `dead_generation_corrupt`; storage failures.
pub(super) fn ended(
    reader: &ReadOnlyDatabase,
    service: &BranchRefRecord,
    branch: &str,
    taken: &[Version],
) -> Result<Vec<Vec<u8>>, Failure> {
    let mut generations: Vec<u64> = taken.iter().map(|at| at.generation).collect();
    generations.dedup();
    let mut ended = Vec::new();
    for generation in generations {
        if let Some(dead) = marker(reader, service, branch, generation)?
            && taken.contains(&dead.last)
        {
            ended.push(dead_key(branch, generation));
        }
    }
    Ok(ended)
}

/// One page of the markers, for the release driver to walk them all in
/// turn (outside review 15): a page that always started at the first marker
/// would ask the same held generations every time and never reach the rest.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DeadPage {
    /// The names on the page with no live record, each once.
    pub names: Vec<String>,
    /// The last marker key on the page: where the next page starts after.
    pub last: Option<Vec<u8>>,
    /// The page reached the last marker; the next starts at the first.
    pub end: bool,
}

impl Storage {
    /// At most `limit` markers after the key `after` (from the first when
    /// `None`), oldest key first: one range read of at most `limit` + 1 on
    /// each shard, never the history.
    ///
    /// # Errors
    /// `reference_missing`; `dead_generation_corrupt`; storage failures.
    pub fn dead_page(
        &self,
        after: Option<&[u8]>,
        limit: NonZeroUsize,
    ) -> Result<DeadPage, Failure> {
        let service = self
            .refs
            .get(SERVICE)
            .ok_or_else(|| Failure::new("reference_missing", "missing service namespace"))?;
        // The least key past `after`.
        let start = after.map_or_else(
            || DEAD_PREFIX.to_vec(),
            |key| {
                let mut start = key.to_vec();
                start.push(0);
                start
            },
        );
        let read = NonZeroUsize::new(limit.get().saturating_add(1)).unwrap_or(NonZeroUsize::MAX);
        let mut found = page(&self.reader, service, (&start, &past(DEAD_PREFIX)), read)?;
        let end = found.len() <= limit.get();
        found.truncate(limit.get());
        let last = found.last().map(|(key, _dead)| key.clone());
        let mut names: Vec<String> = found
            .into_iter()
            .map(|(_key, dead)| dead.branch)
            .filter(|name| self.refs.get(&branch_name(name)).is_none())
            .collect();
        names.dedup();
        Ok(DeadPage { names, last, end })
    }
}