haematite-cli 0.35.0

//! STOR-050: lines (docs/design/storage/LINES.md).
//!
//! R1: a stage's lineage is one row under the reserved prefix `s\0`, keyed by
//! the version it is written at, in the same commit as the stage's change,
//! its receipt and its author row, so the commit's one device flush makes all
//! of them durable together or none of them (STOR-034). A merge carries none
//! of its source's stage rows: its own commit holds one row naming the
//! source and its version and listing, in order, every stage row the source
//! committed since its fork anchor, so the lineage outlives the source.
//!
//! R2: a line is a branch whose fork provenance says so
//! (serve/storage/branches.rs). A release takes nothing of a live line
//! (serve/storage/release.rs), so it keeps every version it commits. Its
//! discard gives every version up: its history rows are released in the
//! discard, as a discarded generation's rows are under the ruling of 08:05
//! (no keep setting asked), and the reclaim pass that release asks for prunes
//! the nodes only the line reached. A version a live fork still holds is
//! released when that fork is discarded.

use std::num::NonZeroUsize;

use haematite::branch::current_timestamp;
use haematite::db::ReadOnlyDatabase;
use serde::{Deserialize, Serialize};

use super::branches::{self, ForkProvenance};
use super::mutations::PreparedWrite;
use super::release::{Answer, Rules};
use super::{Captured, Failure, SERVICE, Storage, branch_name, dead, history, identifier, reads};
use crate::serve::protocol::{Change, Stage, Version};

/// The reserved prefix of stage rows, beside the receipts (`r\0`) and the
/// author rows (`a\0`).
pub const TAG: &[u8] = b"s\0";

const FORMAT: u32 = 1;

/// The most history rows one release of a discarded line names: a line of
/// more versions is released in more than one, each its own three durable
/// operations (serve/storage/release.rs).
const RELEASE_ROWS: usize = 4_096;

/// A branch and one of its versions.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct At {
    pub branch: String,
    pub version: Version,
}

/// One stage row a merge's source committed: where, and the stage.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Listed {
    pub branch: String,
    pub version: Version,
    pub stage: Stage,
}

/// What a merged commit's stage row says: the merge's source at the version
/// merged, and every stage row it committed since its fork anchor, in order.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Merged {
    pub from: At,
    pub stages: Vec<Listed>,
}

/// The row as stored: exactly one of `stage` and `merged`.
#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct Row {
    format: u32,
    version: Version,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    stage: Option<Stage>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    merged: Option<Merged>,
}

/// What a version's stage row says, as history answers it.
#[derive(Debug, Default, PartialEq, Eq)]
pub struct Lineage {
    pub stage: Option<Stage>,
    pub merged: Option<Merged>,
}

/// The key of the stage row written at `version`: the tag, then the
/// generation and sequence big-endian, so rows sort in version order.
#[must_use]
pub fn key(version: Version) -> Vec<u8> {
    let mut key = Vec::with_capacity(TAG.len() + 16);
    key.extend_from_slice(TAG);
    key.extend_from_slice(&version.generation.to_be_bytes());
    key.extend_from_slice(&version.sequence.to_be_bytes());
    key
}

fn key_version(key: &[u8]) -> Result<Version, Failure> {
    let corrupt = || Failure::new("stage_row_corrupt", "stage row key is not a version");
    let rest = key.strip_prefix(TAG).ok_or_else(corrupt)?;
    let (generation, sequence) = rest.split_at_checked(8).ok_or_else(corrupt)?;
    Ok(Version {
        generation: u64::from_be_bytes(generation.try_into().map_err(|_| corrupt())?),
        sequence: u64::from_be_bytes(sequence.try_into().map_err(|_| corrupt())?),
    })
}

fn named(value: &str, path: String) -> Result<(), Failure> {
    identifier(value).map_err(|mut failure| {
        failure.message = format!("{path}: {}", failure.message);
        failure.path = Some(path);
        failure
    })
}

/// A stage beside `change`, checked before anything is read or written:
/// beside branch control it is refused `invalid_operation`, and a member
/// that is no name is refused `invalid_identifier` at its path.
///
/// # Errors
/// As above.
pub(super) fn validate(stage: &Stage, change: &Change) -> Result<(), Failure> {
    if matches!(
        change,
        Change::Fork(_) | Change::Discard(_) | Change::Merge(_)
    ) {
        return Err(Failure {
            code: "invalid_operation".to_owned(),
            message: "a stage is written by a data change: branch.fork, branch.discard and \
                      branch.merge take none"
                .to_owned(),
            path: Some("/stage".to_owned()),
            cause: None,
        });
    }
    named(&stage.line, "/stage/line".to_owned())?;
    named(&stage.run, "/stage/run".to_owned())?;
    named(&stage.stage, "/stage/stage".to_owned())?;
    named(&stage.subject, "/stage/subject".to_owned())?;
    for (index, input) in stage.inputs.iter().enumerate() {
        named(&input.branch, format!("/stage/inputs/{index}/branch"))?;
    }
    Ok(())
}

fn encode(row: &Row) -> Result<Vec<u8>, Failure> {
    serde_json::to_vec(row).map_err(Failure::storage)
}

/// The stage row of a change written at `next`.
///
/// # Errors
/// Serialization failures.
pub(super) fn stage_write(next: Version, stage: &Stage) -> Result<PreparedWrite, Failure> {
    let row = Row {
        format: FORMAT,
        version: next,
        stage: Some(stage.clone()),
        merged: None,
    };
    Ok((key(next), Some(encode(&row)?)))
}

fn decode(key: &[u8], bytes: &[u8]) -> Result<Row, Failure> {
    let row: Row = serde_json::from_slice(bytes)
        .map_err(|error| Failure::new("stage_row_corrupt", error.to_string()))?;
    if row.format != FORMAT {
        return Err(Failure::new(
            "stage_row_format_unsupported",
            format!("stage row format {} is not {FORMAT}", row.format),
        ));
    }
    if row.version != key_version(key)? || row.stage.is_some() == row.merged.is_some() {
        return Err(Failure::new(
            "stage_row_corrupt",
            "a stage row names its own version and holds a stage or a merge, not both",
        ));
    }
    Ok(row)
}

/// The stage rows a merge's source committed since its fork anchor, as the
/// merge's change pages found them, on any shard and in any order.
#[derive(Debug, Default)]
pub(super) struct Found(Vec<(Version, Vec<Listed>)>);

impl Found {
    /// One stage row of `source`, found among its changes.
    ///
    /// # Errors
    /// A row that does not decode, by name.
    pub(super) fn add(&mut self, source: &str, key: &[u8], bytes: &[u8]) -> Result<(), Failure> {
        let row = decode(key, bytes)?;
        let listed = match (row.stage, row.merged) {
            (Some(stage), _) => vec![Listed {
                branch: source.to_owned(),
                version: row.version,
                stage,
            }],
            // A merge the source took in lists its own source's stages: they
            // are listed here in its place, so nothing is nested.
            (None, Some(merged)) => merged.stages,
            (None, None) => Vec::new(),
        };
        self.0.push((row.version, listed));
        Ok(())
    }

    const fn is_empty(&self) -> bool {
        self.0.is_empty()
    }

    /// The merged commit's one stage row, written at `next`: `from` at
    /// `version`, its stages in the order it committed them.
    ///
    /// # Errors
    /// Serialization failures.
    fn write(
        mut self,
        next: Version,
        from: &str,
        version: Version,
    ) -> Result<PreparedWrite, Failure> {
        self.0
            .sort_unstable_by_key(|(at, _)| (at.generation, at.sequence));
        let row = Row {
            format: FORMAT,
            version: next,
            stage: None,
            merged: Some(Merged {
                from: At {
                    branch: from.to_owned(),
                    version,
                },
                stages: self.0.into_iter().flat_map(|(_, listed)| listed).collect(),
            }),
        };
        Ok((key(next), Some(encode(&row)?)))
    }
}

/// The stage row a merged commit at `next` carries, when it carries one: when
/// the source committed a stage row since its fork, or is a line.
///
/// # Errors
/// Reading the source's provenance; serialization failures.
pub(super) fn merged_write(
    reader: &ReadOnlyDatabase,
    source: &Captured,
    found: Found,
    next: Version,
) -> Result<Option<PreparedWrite>, Failure> {
    if found.is_empty() && !is_line(reader, source)? {
        return Ok(None);
    }
    found.write(next, &source.branch, source.version).map(Some)
}

/// What the stage row written at `at`'s own version says: `Lineage::default()`
/// for a version no stage and no merge of stages wrote. One point read.
///
/// # Errors
/// A row that does not decode, by name; storage failures.
pub fn read(reader: &ReadOnlyDatabase, at: &Captured) -> Result<Lineage, Failure> {
    let key = key(at.version);
    let Some(bytes) = reads::get_bytes(reader, at, &key)? else {
        return Ok(Lineage::default());
    };
    let row = decode(&key, &bytes)?;
    Ok(Lineage {
        stage: row.stage,
        merged: row.merged,
    })
}

/// Whether the branch at `at` is a line: its fork provenance says so.
///
/// # Errors
/// As [`branches::provenance`].
pub(super) fn is_line(reader: &ReadOnlyDatabase, at: &Captured) -> Result<bool, Failure> {
    Ok(branches::origin(reader, at)?.is_some_and(|(_origin, line)| line))
}

impl Storage {
    /// After the discard of `branch`'s generation `generation` is durable:
    /// first the generation it was forked from, when that name is discarded
    /// and was a line, since this branch was what still held it; then this
    /// generation itself, when it is a line. Each is read from its discard
    /// marker and the history row of its last version, so a replay of the
    /// discard finds what is left and finishes it, and finds nothing once
    /// it is done. An ordinary discard whose origin is live reads three
    /// entries and writes nothing.
    ///
    /// # Errors
    /// `line_release_unfinished`, naming what stopped it: the discard itself
    /// stands, and a replay of it releases the rest.
    pub(super) fn release_after_discard(
        &mut self,
        branch: &str,
        generation: u64,
    ) -> Result<(), Failure> {
        self.release_lines(branch, generation).map_err(|failure| {
            Failure::new(
                "line_release_unfinished",
                format!(
                    "the discard of {branch} is durable; releasing its line versions stopped: \
                     {failure}"
                ),
            )
        })
    }

    fn release_lines(&mut self, branch: &str, generation: u64) -> Result<(), Failure> {
        let Some((origin, line)) = self.discarded_origin(branch, generation)? else {
            return Ok(());
        };
        let parent = origin.version.generation;
        if self.refs.get(&branch_name(&origin.branch)).is_none()
            && let Some((_, true)) = self.discarded_origin(&origin.branch, parent)?
        {
            self.release_line(&origin.branch, parent)?;
        }
        if line {
            self.release_line(branch, generation)?;
        }
        Ok(())
    }

    /// The fork provenance of `branch`'s discarded generation `generation`,
    /// read at its last version's history row; `None` when its marker is
    /// gone (every row released) or it was never forked.
    fn discarded_origin(
        &self,
        branch: &str,
        generation: u64,
    ) -> Result<Option<(ForkProvenance, bool)>, Failure> {
        let service = self
            .refs
            .get(SERVICE)
            .ok_or_else(|| Failure::new("reference_missing", "missing service namespace"))?;
        let Some(marker) = dead::marker(&self.reader, service, branch, generation)? else {
            return Ok(None);
        };
        let Some(last) = history::capture_kept(&self.reader, service, branch, marker.last)? else {
            return Ok(None);
        };
        branches::origin(&self.reader, &last)
    }

    /// Every history row of `branch`'s discarded line generation `generation`
    /// that nothing else holds, released with no keep setting asked: a
    /// release at a time until none is left or what is left is held.
    fn release_line(&mut self, branch: &str, generation: u64) -> Result<(), Failure> {
        // A name made again since is released by its own rules, never here.
        if self.refs.get(&branch_name(branch)).is_some() {
            return Ok(());
        }
        let max_rows = NonZeroUsize::new(RELEASE_ROWS).unwrap_or(NonZeroUsize::MIN);
        loop {
            let service = self
                .refs
                .get(SERVICE)
                .ok_or_else(|| Failure::new("reference_missing", "missing service namespace"))?;
            if dead::marker(&self.reader, service, branch, generation)?.is_none() {
                return Ok(());
            }
            let rules = Rules {
                keep: None,
                max_rows,
                now: current_timestamp(),
            };
            let answer: Answer = self.release_generation(branch, generation, &rules)?;
            if answer.released.is_none() || !answer.more {
                return Ok(());
            }
        }
    }
}

/// A test-only crash between a change's staged writes and its commit, as a
/// kill would leave it: nothing is installed.
#[cfg(test)]
pub mod before_commit {
    use std::cell::Cell;

    use super::Failure;

    thread_local! {
        static ARMED: Cell<bool> = const { Cell::new(false) };
    }

    pub fn arm() {
        ARMED.with(|armed| armed.set(true));
    }

    pub fn reached() -> Result<(), Failure> {
        if ARMED.with(|armed| armed.replace(false)) {
            return Err(Failure::new(
                "injected_crash",
                "stopped after the writes were staged and before the commit",
            ));
        }
        Ok(())
    }
}