haematite_cli/serve/

storage.rs

//! Durable collections and documents over the engine's branch machinery.

use std::collections::{BTreeMap, BTreeSet};
use std::path::Path;
use std::sync::Arc;
use std::time::Duration;

use haematite::branch::history::SERVICE_BRANCH as SERVICE;
use haematite::branch::{
    KeptLimit, Kind as BranchKind, RefRecord as BranchRefRecord, RefStore as BranchRefStore,
    Registry as BranchRegistry, ShardRef as BranchShardRef, create_branch_with_kind,
    current_timestamp, fork_from_with_kind, open as open_branch,
};
use haematite::db::vacuum::CANONICAL_REFS_DIR;
use haematite::db::{
    Database, DatabaseConfig, DatabaseError, ReadOnlyDatabase, branches::BranchDatabase,
};
use haematite::store::NodeCacheBudget;
use haematite::{Retention, RuntimeOptions};
use serde::{Deserialize, Serialize};
use serde_json::Value;

use super::protocol::{Selection, Version};

mod admission;
pub use admission::Admission;

pub mod attribution;
pub mod embed_inherit;
pub mod embed_job;
pub mod embed_pending;
pub mod embed_prune;
pub mod embed_rewalk;
pub mod embed_search;
pub mod embedded;

#[cfg(test)]
#[path = "storage/envelope_tests.rs"]
mod envelope_tests;
#[cfg(test)]
#[path = "storage/retention_tests.rs"]
mod retention_tests;
#[cfg(test)]
#[path = "storage/span_measure_tests.rs"]
mod span_measure_tests;
#[cfg(test)]
#[path = "storage/write_cost_tests.rs"]
mod write_cost_tests;

#[cfg(test)]
#[path = "storage/batch_test_support.rs"]
mod batch_test_support;
#[cfg(test)]
#[path = "storage/documents_delete_many_tests.rs"]
mod documents_delete_many_tests;
#[cfg(test)]
#[path = "storage/edges_batch_cost_tests.rs"]
mod edges_batch_cost_tests;
#[cfg(test)]
#[path = "storage/vectors_many_tests.rs"]
mod vectors_many_tests;

mod branches;
mod catalogue;
mod discards;
mod document_delete_many;
pub use discards::DiscardPasses;
pub(super) mod edges;
pub(super) mod field_index;
pub(super) mod graph;
mod history;
pub use history::{KEPT_VERSIONS_DEFAULT, KEPT_VERSIONS_MOST};
mod counts;
mod dead;
mod merge_holders;
mod merge_journal;
mod merges;
mod mutations;
mod retention;
pub use retention::AppendPruneLimits;
pub(in crate::serve) use retention::append_prune_address;
pub mod reads;
pub(super) mod relations;
mod release;
pub(super) mod review;
mod vector_import;
pub(super) mod vectors;
pub use branches::ForkProvenance;
#[cfg(test)]
pub use catalogue::after_held;
pub use catalogue::{Catalogue, Held};
pub use merge_holders::{MergePin, MergePlan, MergeState};
pub use merge_journal::{
    CandidateAction, CandidateAdvance, MergeChoice, MergeChoiceRow, MergeChoices, MergeChoose,
    MergeCoverage, MergeOwner, MissingChoice, RecordedChoice,
};
pub use release::{
    Answer as ReleaseAnswer, HeldBy, IntentListing, Keep, Rules as ReleaseRules,
    Source as ReleaseSource, UndatedGeneration,
};
#[cfg(test)]
pub use release::{Crash, crash_at};

const BRANCH_PREFIX: &str = haematite::branch::history::DATA_BRANCH_PREFIX;

mod failure;
pub use failure::Failure;
#[cfg(test)]
#[path = "storage/failure_tests.rs"]
mod failure_tests;

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Receipt {
    pub request_id: String,
    pub fingerprint: String,
    pub branch: String,
    pub version: Version,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub retention: Option<super::protocol::RetentionResult>,
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Collection {
    pub name: String,
    pub schema: Option<Value>,
    pub changed_in: String,
    pub changed_at: Version,
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct DocumentRevision {
    pub collection_branch: String,
    pub collection_version: Version,
    pub written_in: String,
    pub written_at: Version,
}

/// A captured read selection contains every physical shard root.
#[derive(Debug, Clone)]
pub struct Captured {
    pub branch: String,
    pub version: Version,
    pub shards: Vec<BranchShardRef>,
}

/// Mutation admission is serialized by the caller. Read-only work uses a
/// captured selection and independent reader after releasing that admission.
pub struct Storage {
    refs: BranchRefStore,
    visible: Catalogue,
    registry: BranchRegistry,
    reader: Arc<ReadOnlyDatabase>,
    writer: BranchDatabase,
    pending_discards: BTreeSet<String>,
    confirmed_candidates: BTreeMap<String, Version>,
    /// STOR-003 `MD5a`: the kept window's limit and the rows known in it.
    kept_rows: history::KeptRows,
    /// What came of the last discard's ask for a reclaim pass.
    discard_passes: DiscardPasses,
}

pub(super) const fn runtime_options(
    long_wait_interval: Duration,
) -> Result<RuntimeOptions, DatabaseError> {
    if long_wait_interval.is_zero() {
        return Err(DatabaseError::InvalidLongWaitInterval {
            given: long_wait_interval,
        });
    }
    Ok(RuntimeOptions {
        long_wait_interval,
        reclaim_start: haematite::ReclaimStart::AtOpen,
    })
}

impl Storage {
    /// Create only in a previously absent directory. Never adopt application data.
    ///
    /// # Errors
    ///
    /// Refuses unsupported writer policies before branch references are created.
    /// Propagates any error from `try_exists`, `open` or `create_branch_with_kind`.
    pub fn create(
        config: DatabaseConfig,
        branch_cache: NodeCacheBudget,
        long_wait_interval: Duration,
    ) -> Result<Self, Failure> {
        Self::create_kept(
            config,
            branch_cache,
            history::default_kept()?,
            long_wait_interval,
        )
    }

    /// [`Self::create`], each branch record keeping `kept` previous versions
    /// (STOR-003 `MD5a`).
    ///
    /// # Errors
    ///
    /// As [`Self::create`].
    pub fn create_kept(
        config: DatabaseConfig,
        branch_cache: NodeCacheBudget,
        kept: KeptLimit,
        long_wait_interval: Duration,
    ) -> Result<Self, Failure> {
        let options = runtime_options(long_wait_interval).map_err(Failure::from)?;
        if config.data_dir.try_exists().map_err(Failure::storage)? {
            return Err(Failure::new(
                "directory_exists",
                "create requires a new directory",
            ));
        }
        let database = Database::create(config, options).map_err(Failure::from)?;
        database.require_path_writer().map_err(Failure::from)?;
        let mut writer = BranchDatabase::new(database, branch_cache).map_err(Failure::database)?;
        let mut refs = BranchRefStore::open_writing(
            writer.data_dir().join(CANONICAL_REFS_DIR),
            writer.writer_lock(),
        )
        .map_err(Failure::branch_ref)?;
        refs.keep_versions(kept);
        let registry = BranchRegistry::new();
        create_branch_with_kind(
            SERVICE,
            writer.base_roots().map_err(Failure::database)?,
            BranchKind::Namespace,
            &mut refs,
            &registry,
            current_timestamp(),
        )
        .map_err(Failure::branch_commit)?;
        counts::initialize_new_store(&mut writer, &mut refs, &registry)?;
        Self::finish_open(writer, refs, registry, kept)
    }

    /// Reopen a served directory under its existing exclusive writer lock.
    ///
    /// # Errors
    ///
    /// Refuses unsupported writer policies before engine startup.
    /// Propagates any error from `open`.
    pub fn open(
        path: &Path,
        branch_cache: NodeCacheBudget,
        long_wait_interval: Duration,
    ) -> Result<Self, Failure> {
        Self::open_kept(
            path,
            branch_cache,
            history::default_kept()?,
            long_wait_interval,
        )
    }

    /// [`Self::open`], each branch record keeping `kept` previous versions
    /// (STOR-003 `MD5a`).
    ///
    /// # Errors
    ///
    /// As [`Self::open`].
    pub fn open_kept(
        path: &Path,
        branch_cache: NodeCacheBudget,
        kept: KeptLimit,
        long_wait_interval: Duration,
    ) -> Result<Self, Failure> {
        let options = runtime_options(long_wait_interval).map_err(Failure::from)?;
        if !path.join(CANONICAL_REFS_DIR).is_dir() {
            return Err(Failure::new(
                "not_served",
                "directory has no served branch references",
            ));
        }
        // The service's own record, read from disk with no ref store opened
        // and nothing attached: only a store the service made is moved below.
        match BranchRefStore::read_record(path.join(CANONICAL_REFS_DIR), SERVICE) {
            Ok(Some(_record)) => {}
            Ok(None) => {
                return Err(Failure::new(
                    "not_served",
                    "directory is not a haem/1 service database",
                ));
            }
            Err(error) => {
                return Err(Failure::new(
                    "service_record_unreadable",
                    format!("the service's branch record cannot be read: {error}"),
                ));
            }
        }
        let database = Database::open_path_writer(path, options).map_err(Failure::from)?;
        let database = if database.reclaim_mark_bytes().is_some() {
            database
        } else {
            // Moved once, from KeepAll to reclaiming under the default bound
            // (STOR-003 H1a): the store's config records it, so every later
            // open finds Reclaim and moves nothing.
            drop(database);
            let moved = Database::open_with_retention(
                path,
                runtime_options(long_wait_interval).map_err(Failure::from)?,
                Retention::reclaim(0),
            )
            .map_err(Failure::from)?;
            moved.require_path_writer().map_err(Failure::from)?;
            moved
        };
        let writer = BranchDatabase::new(database, branch_cache).map_err(Failure::database)?;
        let mut refs =
            BranchRefStore::open_writing(path.join(CANONICAL_REFS_DIR), writer.writer_lock())
                .map_err(Failure::branch_ref)?;
        refs.keep_versions(kept);
        if refs.get(SERVICE).is_none() {
            return Err(Failure::new(
                "not_served",
                "directory is not a haem/1 service database",
            ));
        }
        Self::finish_open(writer, refs, BranchRegistry::new(), kept)
    }

    fn finish_open(
        writer: BranchDatabase,
        mut refs: BranchRefStore,
        registry: BranchRegistry,
        kept: KeptLimit,
    ) -> Result<Self, Failure> {
        if refs.get(SERVICE).is_none() {
            return Err(Failure::new("not_served", "missing service namespace"));
        }
        let reader =
            Arc::new(ReadOnlyDatabase::open(writer.data_dir()).map_err(Failure::observer)?);
        history::require_file_rows(&reader, &refs)?;
        discards::require_file_rows(&reader, &refs)?;
        let main = branch_name("main");
        if refs.get(&main).is_none() {
            let parent = open_branch(SERVICE, &refs, &registry).map_err(Failure::branch_commit)?;
            fork_from_with_kind(
                &parent,
                &main,
                BranchKind::Work,
                &mut refs,
                &registry,
                current_timestamp(),
            )
            .map_err(Failure::branch_commit)?;
        }
        // Recovered bytes are not, by themselves, an acknowledged fence.
        // The fence changes no record, and this open owns the store, so the
        // listing taken after it is the listing it fenced.
        refs.confirm_directory().map_err(Failure::branch_ref)?;
        let visible: BTreeMap<_, _> = refs
            .list()
            .map(|record| {
                let kept = refs.kept(&record.name).to_vec();
                (record.name.clone(), (record.clone(), kept))
            })
            .collect();
        // The holder of `writer`, the database this storage keeps: after a
        // move at open it is the reopened one's, never the first's (H1b).
        let holder = writer.read_holder();
        let mut storage = Self {
            refs,
            visible: Catalogue::new(visible, Arc::clone(&reader), holder),
            registry,
            reader,
            writer,
            pending_discards: BTreeSet::new(),
            confirmed_candidates: BTreeMap::new(),
            kept_rows: history::KeptRows::new(kept),
            discard_passes: DiscardPasses::default(),
        };
        storage.reconcile_discards()?;
        Ok(storage)
    }

    /// # Errors
    ///
    /// Returns `Failure` if `capture` fails.
    pub fn capture(&self, selection: &Selection) -> Result<Captured, Failure> {
        self.visible.capture(selection)
    }

    pub(in crate::serve) fn activate_reclaim(&self) -> Result<(), Failure> {
        self.writer.activate_reclaim().map_err(Failure::from)
    }

    /// Run each shard's reclaim to rest (STOR-003 H1c).
    ///
    /// # Errors
    /// Propagates a failure while draining shard reclaim work.
    #[cfg(test)]
    pub fn quiesce_reclaim(&self) -> Result<(), Failure> {
        self.writer.quiesce_reclaim().map_err(Failure::from)
    }

    /// Each shard's reclaim, as the database answers it (SL2: a test's
    /// diagnosis).
    #[cfg(test)]
    pub fn reclaim_status(&self) -> haematite::db::ReclaimStatus {
        self.writer.reclaim_status()
    }

    /// How many read holds live on this storage's database (STOR-003 H1c).
    #[cfg(test)]
    pub fn live_read_holds(&self) -> usize {
        self.writer.read_holder().status().live
    }

    pub fn catalogue(&self) -> Catalogue {
        self.visible.clone()
    }

    /// What came of the last discard's ask for a reclaim pass, shared with
    /// whoever answers `store.stats`.
    #[must_use]
    pub fn discard_passes(&self) -> DiscardPasses {
        self.discard_passes.clone()
    }

    /// A separate immutable reader can perform captured reads concurrently.
    ///
    /// # Errors
    ///
    /// Returns `Failure` if `open` or `data_dir` fails.
    pub fn reader(&self) -> Result<ReadOnlyDatabase, Failure> {
        ReadOnlyDatabase::open(self.writer.data_dir()).map_err(Failure::observer)
    }

    fn retain_current(&mut self, branch: &str, record: &BranchRefRecord) -> Result<(), Failure> {
        self.retain_history_row(branch, record)
    }

    fn publish_record(&self, name: &str) -> Result<(), Failure> {
        let record = self
            .refs
            .get(name)
            .ok_or_else(|| Failure::new("reference_missing", "committed reference disappeared"))?;
        self.visible
            .publish(record.clone(), self.refs.kept(name).to_vec())
    }
}

const fn version(record: &BranchRefRecord) -> Version {
    Version {
        generation: record.created,
        sequence: record.seq,
    }
}

fn branch_name(name: &str) -> String {
    format!("{BRANCH_PREFIX}{name}")
}

fn identifier(value: &str) -> Result<(), Failure> {
    if value.is_empty() || value.len() > 512 || value.contains('\0') {
        return Err(Failure::new(
            "invalid_identifier",
            "names must contain 1 to 512 UTF-8 bytes and no NUL",
        ));
    }
    Ok(())
}

fn collection_key(name: &str) -> Vec<u8> {
    format!("c\0{name}").into_bytes()
}
pub(in crate::serve) fn document_key(collection: &str, id: &str) -> Vec<u8> {
    format!("d\0{collection}\0{id}").into_bytes()
}

pub(in crate::serve) fn revision_key(collection: &str, id: &str) -> Vec<u8> {
    format!("m\0{collection}\0{id}").into_bytes()
}
fn receipt_key(branch: &str, generation: u64, id: &str) -> Vec<u8> {
    format!("r\0{branch}\0{generation}\0{id}").into_bytes()
}

#[cfg(test)]
mod long_wait_tests {
    use std::error::Error;
    use std::time::Duration;

    use haematite::DatabaseError;
    use haematite::db::DatabaseConfig;
    use haematite::store::NodeCacheBudget;

    use super::{Storage, runtime_options};

    #[test]
    fn zero_is_the_typed_engine_refusal() {
        assert!(matches!(
            runtime_options(Duration::ZERO),
            Err(DatabaseError::InvalidLongWaitInterval { given }) if given == Duration::ZERO
        ));
    }

    #[test]
    fn zero_refuses_create_and_open_before_directory_checks() -> Result<(), Box<dyn Error>> {
        let directory = tempfile::tempdir()?;
        let path = directory.path().join("absent");
        let config = DatabaseConfig {
            data_dir: path.clone(),
            shard_count: 1,
            distributed: None,
            executor_threads: Some(1),
            node_cache_budget: Some(NodeCacheBudget::Unlimited),
            retention: haematite::Retention::reclaim(0),
        };
        for refused in [
            Storage::create(config, NodeCacheBudget::Unlimited, Duration::ZERO),
            Storage::open(&path, NodeCacheBudget::Unlimited, Duration::ZERO),
        ] {
            let failure = refused.err().ok_or("zero interval was accepted")?;
            assert_eq!(failure.code, "storage_unavailable");
            assert_eq!(
                failure.message,
                DatabaseError::InvalidLongWaitInterval {
                    given: Duration::ZERO
                }
                .to_string()
            );
            assert!(!path.try_exists()?);
        }
        Ok(())
    }
}