haematite_cli/serve/storage/

discards.rs

//! Completed receipts are point-read rows; only interrupted removals keep intents.

use haematite::branch::{CommitDurability, CommitRequest, RefRecord, remove as remove_branch};
use haematite::db::ReadOnlyDatabase;
use haematite::tree::Hash;
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};

use super::{
    BranchKind, BranchRefStore, Failure, Receipt, SERVICE, Storage, branch_name,
    create_branch_with_kind, current_timestamp,
    dead::{Dead, dead_key},
    identifier, open_branch, version,
};
use crate::serve::protocol::{Mutation, ReceiptQuery, Version};

const FILE_PREFIX: &str = "@haem/discard/";
const INTENT_PREFIX: &str = "@haem/discard-pending/";

#[derive(Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct Intent {
    branch: String,
    request_id: String,
    version: Version,
}

impl Intent {
    fn name(&self) -> Result<String, Failure> {
        Ok(format!(
            "{INTENT_PREFIX}{}",
            serde_json::to_string(self).map_err(Failure::storage)?
        ))
    }

    fn decode(record: &RefRecord, shard_count: usize) -> Result<Self, Failure> {
        let corrupt = |message| {
            Failure::new(
                "discard_intent_corrupt",
                format!("{}: {message}", record.name),
            )
        };
        let suffix = record
            .name
            .strip_prefix(INTENT_PREFIX)
            .ok_or_else(|| corrupt("intent namespace differs".to_owned()))?;
        let intent: Self =
            serde_json::from_str(suffix).map_err(|error| corrupt(error.to_string()))?;
        identifier(&intent.branch).map_err(|error| corrupt(error.to_string()))?;
        identifier(&intent.request_id).map_err(|error| corrupt(error.to_string()))?;
        if intent.branch == "main"
            || record.kind != BranchKind::Namespace
            || record.seq != 0
            || intent.name()? != record.name
            || record.shards.len() != shard_count
            || record
                .shards
                .iter()
                .enumerate()
                .any(|(id, shard)| id != shard.shard_id)
        {
            return Err(corrupt("intent identity or kind differs".to_owned()));
        }
        Ok(intent)
    }
}

impl Storage {
    /// The discards asked for and not finished, as (branch, version) each,
    /// read from their intent records (STOR-003 H1b2): what a release's floor
    /// relies on besides heads, forks and merges. An intent that does not
    /// decode refuses the listing by its name; none is passed over.
    ///
    /// # Errors
    /// `discard_intent_corrupt`, naming the record.
    pub fn pending_discards(&self) -> Result<Vec<(String, Version)>, Failure> {
        let shard_count = self.reader.shard_count();
        self.refs
            .list_prefix(INTENT_PREFIX)
            .map(|record| {
                Intent::decode(record, shard_count).map(|intent| (intent.branch, intent.version))
            })
            .collect()
    }
}

pub(super) fn key(branch: &str, request: &str) -> Vec<u8> {
    let mut key = Vec::with_capacity(66);
    key.extend_from_slice(b"d\0");
    key.extend_from_slice(Hash::of(branch.as_bytes()).as_bytes());
    key.extend_from_slice(Hash::of(request.as_bytes()).as_bytes());
    key
}

fn service_record<'a>(
    reader: &ReadOnlyDatabase,
    refs: &'a BranchRefStore,
) -> Result<&'a RefRecord, Failure> {
    let record = refs
        .get(SERVICE)
        .ok_or_else(|| Failure::new("reference_missing", "missing service namespace"))?;
    if record.kind != BranchKind::Namespace
        || record.shards.len() != reader.shard_count()
        || record
            .shards
            .iter()
            .enumerate()
            .any(|(id, shard)| id != shard.shard_id)
    {
        return Err(Failure::new(
            "discard_service_invalid",
            "service must name every physical shard",
        ));
    }
    Ok(record)
}

fn read_row(
    reader: &ReadOnlyDatabase,
    refs: &BranchRefStore,
    key: &[u8],
) -> Result<Option<Receipt>, Failure> {
    let service = service_record(reader, refs)?;
    let shard_id = reader.shard_for(key);
    let root = service
        .shards
        .get(shard_id)
        .ok_or_else(|| Failure::new("incomplete_roots", "service row shard is missing"))?
        .head;
    reader
        .get_at(shard_id, root, key)
        .map_err(Failure::observer)?
        .map(|bytes| {
            let corrupt = |message| Failure::new("discard_record_corrupt", message);
            let receipt: Receipt =
                serde_json::from_slice(&bytes).map_err(|error| corrupt(error.to_string()))?;
            identifier(&receipt.branch).map_err(|error| corrupt(error.to_string()))?;
            identifier(&receipt.request_id).map_err(|error| corrupt(error.to_string()))?;
            if receipt.branch == "main"
                || self::key(&receipt.branch, &receipt.request_id) != key
                || receipt.fingerprint.len() != 64
                || !receipt
                    .fingerprint
                    .bytes()
                    .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
            {
                return Err(corrupt("receipt identity differs from its row".to_owned()));
            }
            Ok(receipt)
        })
        .transpose()
}

pub(super) fn require_file_rows(
    reader: &ReadOnlyDatabase,
    refs: &BranchRefStore,
) -> Result<(), Failure> {
    for record in refs.list_prefix(FILE_PREFIX) {
        let suffix = record
            .name
            .strip_prefix(FILE_PREFIX)
            .ok_or_else(|| Failure::new("discard_ref_corrupt", &record.name))?;
        let (branch, request) = suffix
            .split_once('/')
            .ok_or_else(|| Failure::new("discard_ref_corrupt", &record.name))?;
        let mut key = b"d\0".to_vec();
        key.extend_from_slice(&decode_hash(branch).map_err(|error| {
            Failure::new("discard_ref_corrupt", format!("{}: {error}", record.name))
        })?);
        key.extend_from_slice(&decode_hash(request).map_err(|error| {
            Failure::new("discard_ref_corrupt", format!("{}: {error}", record.name))
        })?);
        if read_row(reader, refs, &key)?.is_none() {
            return Err(Failure::new(
                "discard_rows_incomplete",
                format!("discard file {} has no matching row", record.name),
            ));
        }
    }
    Ok(())
}

fn decode_hash(hex: &str) -> Result<[u8; 32], Failure> {
    let invalid = || {
        Failure::new(
            "discard_ref_corrupt",
            "discard reference requires canonical hashes",
        )
    };
    if hex.len() != 64
        || !hex
            .bytes()
            .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
    {
        return Err(invalid());
    }
    let mut hash = [0; 32];
    for (byte, pair) in hash.iter_mut().zip(hex.as_bytes().as_chunks::<2>().0) {
        let digits = std::str::from_utf8(pair).map_err(|_| invalid())?;
        *byte = u8::from_str_radix(digits, 16).map_err(|_| invalid())?;
    }
    Ok(hash)
}

impl Storage {
    pub(super) fn discard_request(
        &mut self,
        request: &Mutation,
        fingerprint: &str,
        target: &str,
        record: &RefRecord,
    ) -> Result<Receipt, Failure> {
        identifier(target)?;
        if target == "main" {
            return Err(Failure::new("main_protected", "main cannot be discarded"));
        }
        if target != request.branch {
            return Err(Failure::new(
                "discard_target_mismatch",
                "discard must name the mutation's branch",
            ));
        }
        // A line gives up every version it kept, version 0 to its head, and
        // its receipt says how many (STOR-050 R2): read before anything is
        // written, so the stored receipt is the answer every time.
        let head = super::Captured {
            branch: target.to_owned(),
            version: version(record),
            shards: record.shards.clone(),
        };
        let versions_released = if super::lineage::is_line(&self.reader, &head)? {
            Some(record.seq.checked_add(1).ok_or_else(|| {
                Failure::new("sequence_exhausted", "a line's versions cannot be counted")
            })?)
        } else {
            None
        };
        // The head and the versions the record keeps leave with it: each has
        // its row first, in one service commit (STOR-003 MD5a, MD5c).
        self.rows_before_discard(target, record)?;
        #[cfg(test)]
        super::history::after_rows::reached()?;
        let intent = Intent {
            branch: target.to_owned(),
            request_id: request.request_id.clone(),
            version: request.expected,
        };
        let name = intent.name()?;
        let result = create_branch_with_kind(
            &name,
            record
                .shards
                .iter()
                .map(|shard| (shard.shard_id, shard.head)),
            BranchKind::Namespace,
            &mut self.refs,
            &self.registry,
            current_timestamp(),
        );
        if self.refs.get(&name).is_some() {
            self.pending_discards.insert(record.name.clone());
            self.visible.remove(&record.name)?;
        }
        result.map_err(|error| {
            Failure::new(
                if self.refs.get(&name).is_some() {
                    "durability_unknown"
                } else {
                    "write_refused"
                },
                error.to_string(),
            )
        })?;
        let receipt = Receipt {
            retention: None,
            request_id: request.request_id.clone(),
            fingerprint: fingerprint.to_owned(),
            branch: target.to_owned(),
            version: request.expected,
            versions_released,
        };
        self.commit_and_finish_discard(&name, &intent, &receipt)?;
        // Durable now: the line's rows, and its origin's when that is a
        // discarded line this branch held, go to reclaim (STOR-050 R2).
        self.release_after_discard(target, record.created)?;
        Ok(receipt)
    }

    fn commit_and_finish_discard(
        &mut self,
        name: &str,
        intent: &Intent,
        receipt: &Receipt,
    ) -> Result<(), Failure> {
        if let Err(refused) = self.commit_discard_row(receipt) {
            if refused.code == "write_refused" {
                self.finish_intent(name, intent, None).map_err(|abort| {
                    Failure::new(
                        "discard_abort_failed",
                        format!("{refused}; failed to abort discard intent: {abort}"),
                    )
                })?;
            }
            return Err(refused);
        }
        self.finish_intent(name, intent, Some(receipt))
    }

    fn commit_discard_row(&mut self, receipt: &Receipt) -> Result<(), Failure> {
        #[cfg(test)]
        let fault = ROW_COMMIT_FAULT.with(std::cell::Cell::take);
        #[cfg(test)]
        if matches!(fault, Some(CommitFault::RefuseBeforeCommit)) {
            return Err(Failure::new(
                "write_refused",
                "injected refusal before discard row commit",
            ));
        }
        let key = key(&receipt.branch, &receipt.request_id);
        let bytes = serde_json::to_vec(receipt).map_err(Failure::storage)?;
        let now = current_timestamp();
        // The generation's marker rides the receipt's own commit: a discard
        // whose receipt is durable removes its branch, and only such a one
        // leaves rows with no record for release to find.
        let dead_key = dead_key(&receipt.branch, receipt.version.generation);
        let dead = serde_json::to_vec(&Dead {
            branch: receipt.branch.clone(),
            last: receipt.version,
            discarded_at: now,
        })
        .map_err(Failure::storage)?;
        let at = version(service_record(&self.reader, &self.refs)?);
        let service =
            open_branch(SERVICE, &self.refs, &self.registry).map_err(Failure::branch_commit)?;
        service
            .put(self.writer.shard_for(&key), key, bytes)
            .map_err(Failure::branch)?;
        service
            .put(self.writer.shard_for(&dead_key), dead_key, dead)
            .map_err(Failure::branch)?;
        self.writer
            .commit(
                &service,
                &self.registry,
                CommitRequest {
                    durability: CommitDurability::Durable {
                        refs: &mut self.refs,
                    },
                    extra_parents: &[],
                    timestamp: now,
                    note: None,
                },
            )
            .map_err(|error| {
                let installed = self.refs.get(SERVICE).is_some_and(|now| version(now) != at);
                Failure::new(
                    if installed {
                        "durability_unknown"
                    } else {
                        "write_refused"
                    },
                    error.to_string(),
                )
            })?;
        #[cfg(test)]
        if matches!(fault, Some(CommitFault::LoseCommittedAnswer)) {
            return Err(Failure::new(
                "durability_unknown",
                "injected lost answer after discard row commit",
            ));
        }
        self.publish_record(SERVICE)
    }

    pub(super) fn discard_receipt(
        &mut self,
        query: &ReceiptQuery,
    ) -> Result<Option<Receipt>, Failure> {
        let receipt = read_row(
            &self.reader,
            &self.refs,
            &key(&query.branch, &query.request_id),
        )?;
        if let Some(receipt) = &receipt {
            if receipt.branch != query.branch || receipt.request_id != query.request_id {
                return Err(Failure::new(
                    "discard_record_corrupt",
                    "receipt differs from requested identity",
                ));
            }
            let at = version(service_record(&self.reader, &self.refs)?);
            self.refs
                .confirm_version(SERVICE, at.generation, at.sequence)
                .map_err(|error| Failure::new("durability_unknown", error.to_string()))?;
            self.publish_record(SERVICE)?;
        }
        // The receipt supplies the original generation after unlink. Before
        // row commit, admission still holds the target at its original head.
        let target = branch_name(&query.branch);
        if self.pending_discards.contains(&target) {
            let expected = receipt
                .as_ref()
                .map(|receipt| receipt.version)
                .or_else(|| self.refs.get(&target).map(version))
                .ok_or_else(|| {
                    Failure::new(
                        "discard_pending",
                        "pending removal has no receipt or target",
                    )
                })?;
            let intent = Intent {
                branch: query.branch.clone(),
                request_id: query.request_id.clone(),
                version: expected,
            };
            let name = intent.name()?;
            if let Some(record) = self.refs.get(&name) {
                let intent = Intent::decode(record, self.reader.shard_count())?;
                self.finish_intent(&name, &intent, receipt.as_ref())?;
            }
        }
        Ok(receipt)
    }

    pub(super) fn reconcile_discards(&mut self) -> Result<(), Failure> {
        // Intent records are removed after both the row and branch unlink are
        // fenced. Clean open has no intents and performs no receipt point read.
        let mut targets = std::collections::BTreeSet::new();
        let mut intents = Vec::new();
        for record in self.refs.list_prefix(INTENT_PREFIX) {
            let intent = Intent::decode(record, self.reader.shard_count())?;
            if !targets.insert(intent.branch.clone()) {
                return Err(Failure::new(
                    "discard_intent_corrupt",
                    "multiple intents name the same branch",
                ));
            }
            intents.push((record.name.clone(), intent));
        }
        for (name, intent) in intents {
            let receipt = read_row(
                &self.reader,
                &self.refs,
                &key(&intent.branch, &intent.request_id),
            )?;
            self.finish_intent(&name, &intent, receipt.as_ref())?;
        }
        Ok(())
    }

    fn finish_intent(
        &mut self,
        name: &str,
        intent: &Intent,
        receipt: Option<&Receipt>,
    ) -> Result<(), Failure> {
        let target = branch_name(&intent.branch);
        self.pending_discards.insert(target.clone());
        let mut removed = false;
        if let Some(receipt) = receipt {
            if receipt.branch != intent.branch
                || receipt.request_id != intent.request_id
                || receipt.version != intent.version
            {
                return Err(Failure::new(
                    "discard_intent_corrupt",
                    "intent differs from its receipt row",
                ));
            }
            if let Some(current) = self.refs.get(&target)
                && current.created == intent.version.generation
            {
                if version(current) != intent.version {
                    return Err(Failure::new(
                        "discard_conflict",
                        "branch advanced after its removal was recorded",
                    ));
                }
                self.visible.remove(&target)?;
                remove_branch(&target, &mut self.refs)
                    .map_err(|error| Failure::new("durability_unknown", error.to_string()))?;
                removed = true;
            }
        }
        // An absent row proves the intent never reached a committed removal.
        // A newer generation is preserved even if an old intent resurfaces.
        remove_branch(name, &mut self.refs)
            .map_err(|error| Failure::new("durability_unknown", error.to_string()))?;
        self.visible.remove(name)?;
        if self.refs.get(&target).is_some() {
            self.publish_record(&target)?;
        } else {
            self.visible.remove(&target)?;
        }
        self.pending_discards.remove(&target);
        // The pass is asked for only now: the intent removed just above holds
        // the target's heads, so a pass marked before it would keep every node
        // the discard gave up. As after a release (release.rs carry_out), the
        // removal is durable by now and nothing is waited on. What came of the
        // ask is said in `store.stats`, never in the receipt: a receipt is the
        // stored row, the same every time it is answered.
        if removed {
            self.ask_pass(intent);
        }
        Ok(())
    }

    /// Ask the engine for a reclaim pass and return at once: removing a ref
    /// writes no node, so no pass falls due by itself (STOR-003 RB2). A
    /// refusal is kept by name for `store.stats`; the discard stands.
    fn ask_pass(&self, intent: &Intent) {
        let asked = self.writer.request_reclaim();
        #[cfg(test)]
        let asked = PASS_REFUSED.with(std::cell::Cell::take).map_or(asked, Err);
        self.discard_passes.record(LastPass {
            branch: intent.branch.clone(),
            request_id: intent.request_id.clone(),
            asked: asked.is_ok(),
            refused: asked.err().map(|error| error.to_string()),
        });
    }
}

/// What came of the last discard's ask for a reclaim pass.
#[derive(Debug, Clone, Serialize)]
struct LastPass {
    branch: String,
    request_id: String,
    asked: bool,
    refused: Option<String>,
}

/// The last discard's ask for a reclaim pass, shared between the storage that
/// asks and the service that answers `store.stats`.
#[derive(Debug, Clone, Default)]
pub struct DiscardPasses(std::sync::Arc<std::sync::Mutex<Option<LastPass>>>);

impl DiscardPasses {
    fn record(&self, pass: LastPass) {
        *self
            .0
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(pass);
    }

    /// The last ask as `store.stats` says it: null before any discard asked,
    /// else its branch, request id, whether it was asked, and the refusal by
    /// name when it was refused.
    #[must_use]
    pub fn status(&self) -> Value {
        self.0
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .as_ref()
            .map_or(Value::Null, |pass| json!(pass))
    }
}

#[cfg(test)]
#[derive(Clone, Copy)]
enum CommitFault {
    RefuseBeforeCommit,
    LoseCommittedAnswer,
}

#[cfg(test)]
thread_local! {
    static ROW_COMMIT_FAULT: std::cell::Cell<Option<CommitFault>> =
        const { std::cell::Cell::new(None) };
    /// The refusal the next ask for a pass is answered with, in place of the
    /// engine's answer.
    static PASS_REFUSED: std::cell::Cell<Option<haematite::db::DatabaseError>> =
        const { std::cell::Cell::new(None) };
}

#[cfg(test)]
#[path = "discards_tests.rs"]
mod tests;