haematite-cli 0.21.9

//! A history clock uses the same retained versions as history listing.

use std::num::NonZeroUsize;

use haematite::Hash;
use haematite::branch::history::{Key, Row};
use haematite::branch::{KeptHead, RefRecord};
use haematite::db::ReadOnlyDatabase;

use super::{Catalogue, SERVICE, branch_name, history, version};
use crate::serve::protocol::Version;
use crate::serve::storage::{Captured, Failure};

impl Catalogue {
    /// Keep the service roots alive while reading spilled version timestamps.
    pub(in crate::serve) fn window_clock(
        &self,
        at: &Captured,
        maximum: NonZeroUsize,
    ) -> Result<Vec<(Version, u64)>, Failure> {
        let visible = self.read()?;
        let (record, kept) = visible
            .get(&branch_name(&at.branch))
            .ok_or_else(|| Failure::new("branch_missing", &at.branch))?;
        if record.created != at.version.generation || record.seq < at.version.sequence {
            return Err(Failure::new(
                "generation_mismatch",
                "window history no longer contains the captured generation",
            ));
        }
        let (service, _) = visible
            .get(SERVICE)
            .ok_or_else(|| Failure::new("reference_missing", "missing service namespace"))?;
        let roots: Vec<_> = service
            .shards
            .iter()
            .map(|shard| (shard.shard_id, shard.head))
            .collect();
        let hold = self
            .holder
            .hold(&roots)
            .map_err(|error| Failure::new("read_hold_refused", error.to_string()))?;
        let record = record.clone();
        let kept = kept.clone();
        let service = service.clone();
        drop(visible);
        let current = Captured {
            branch: at.branch.clone(),
            version: version(&record),
            shards: record.shards.clone(),
        };
        let listed = history::list_kept(&self.reader, &service, &current, &kept, None, maximum)?;
        let mut clock = Vec::new();
        for row in listed
            .into_iter()
            .take_while(|row| row.version.sequence <= at.version.sequence)
        {
            let timestamp = timestamp(&self.reader, &service, &record, &kept, &row)?;
            clock.push((row.version, timestamp));
        }
        drop(hold);
        if clock
            .last()
            .is_none_or(|(version, _)| *version != at.version)
        {
            return Err(Failure::new(
                "query_window_history_limit",
                "window instant resolution exceeds the history scan allowance",
            ));
        }
        Ok(clock)
    }
}

fn timestamp(
    reader: &ReadOnlyDatabase,
    service: &RefRecord,
    current: &RefRecord,
    kept: &[KeptHead],
    at: &Captured,
) -> Result<u64, Failure> {
    if at.version.sequence == current.seq {
        return Ok(current.timestamp);
    }
    if let Some(entry) = kept.iter().find(|entry| entry.seq == at.version.sequence) {
        return Ok(entry.timestamp);
    }
    let key = Key {
        branch: Hash::of(at.branch.as_bytes()),
        generation: at.version.generation,
        sequence: at.version.sequence,
    }
    .encode();
    let shard_id = reader.shard_for(&key);
    let shard = service.shards.get(shard_id).ok_or_else(|| {
        Failure::new(
            "incomplete_roots",
            "window history service shard is missing",
        )
    })?;
    let bytes = reader
        .get_at(shard_id, shard.head, &key)
        .map_err(Failure::observer)?
        .ok_or_else(|| Failure::new("version_missing", "window history row is missing"))?;
    let row = Row::decode(&key, &bytes)
        .map_err(|error| Failure::new("history_row_corrupt", error.to_string()))?;
    Ok(row.timestamp)
}

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