liminal-server 0.24.1

//! DUR-028 R6: the installed store counted, read-only, so the morning knows
//! who the shared key hurt.
//!
//! Before DUR-028 every conversation on a shard wrote its checkpoint locator
//! under one bare key, so the shard held one locator and it named one
//! conversation: every other conversation on that shard restored against a
//! locator that was not its own. This census reads a store (or a copy) and
//! answers, per store: the conversations per shard, the legacy locators and
//! which conversation each decodes for, the conversations whose only
//! locator names another, and of those, how many have an operation log cut
//! by a prune, which a rebuild from nothing cannot replay. It writes
//! nothing: the registry is read without its snapshot migration, and the
//! store's own counters say so when the store counts.

use std::collections::{BTreeMap, BTreeSet};
use std::sync::Arc;

use liminal::durability::{DurableStore, StoreCounts, bridge::block_on};

use super::checkpoint_batches::CensusError;
use super::prune::{PRUNE_RECEIPT_PREFIX, PruneReceipt, PrunedStream};
use super::registry::ConversationRegistry;
use super::restore_checkpoint_locator::{CheckpointLocator, checkpoint_conversation_route};

/// The most ids one line names before it says how many more there are.
const NAMED_IDS: usize = 16;

/// What one store holds, counted without writing (R6).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct StoreCensus {
    pub live: usize,
    pub sweeping: usize,
    /// The store's shards; one when the store has no spine to ask.
    pub shards: usize,
    /// Conversations per shard, by the shard their checkpoint route hashes to.
    pub per_shard: BTreeMap<usize, usize>,
    /// Each legacy (bare) locator found, by shard: the conversation it names,
    /// or why its bytes do not read.
    pub legacy_locators: BTreeMap<usize, Result<u64, String>>,
    /// Conversations without a locator of their own whose shard's legacy
    /// locator names another conversation: (conversation, the one named).
    pub names_another: Vec<(u64, u64)>,
    /// Of `names_another`, the conversations whose operation log carries a
    /// prune receipt: a rebuild from nothing cannot replay a cut log.
    pub pruned_below_rebuild: Vec<u64>,
    /// The store's counters over the census, when the store counts them.
    pub counts: Option<StoreCounts>,
}

impl StoreCensus {
    /// The figures as plain lines, one per kind; no key or row is printed.
    #[must_use]
    pub fn lines(&self) -> Vec<String> {
        let mut lines = vec![format!(
            "store: {} conversations ({} live, {} sweeping) on {} shards",
            self.live + self.sweeping,
            self.live,
            self.sweeping,
            self.shards
        )];
        for (shard, count) in &self.per_shard {
            lines.push(format!("shard {shard}: {count} conversations"));
        }
        for (shard, named) in &self.legacy_locators {
            lines.push(match named {
                Ok(id) => format!("legacy locator on shard {shard}: names conversation {id}"),
                Err(reason) => format!("legacy locator on shard {shard}: does not read: {reason}"),
            });
        }
        lines.push(format!(
            "conversations whose locator names another: {}{}",
            self.names_another.len(),
            named(
                self.names_another
                    .iter()
                    .map(|(id, other)| format!("{id} names {other}"))
            )
        ));
        lines.push(format!(
            "of those, logs pruned below a rebuild: {}{}",
            self.pruned_below_rebuild.len(),
            named(self.pruned_below_rebuild.iter().map(u64::to_string))
        ));
        if let Some(counts) = &self.counts {
            lines.push(format!(
                "store counts over the census: get_routed={} read_from={} read_value={} writes={}",
                counts.get_routed_calls,
                counts.read_from_calls,
                counts.read_value_calls,
                counts.apply_routed_batch_calls + counts.append_calls + counts.append_all_calls
            ));
        }
        lines
    }
}

/// ` (a, b, c)` for the first [`NAMED_IDS`] names, then how many more.
fn named(names: impl ExactSizeIterator<Item = String>) -> String {
    let more = names.len().saturating_sub(NAMED_IDS);
    let shown: Vec<String> = names.take(NAMED_IDS).collect();
    match (shown.is_empty(), more) {
        (true, _) => String::new(),
        (false, 0) => format!(" ({})", shown.join(", ")),
        (false, more) => format!(" ({}, and {more} more)", shown.join(", ")),
    }
}

/// Counts `store` as [`StoreCensus`] says. Reads only.
///
/// # Errors
/// [`CensusError::Store`] naming the read that failed.
pub fn store_census(store: &Arc<dyn DurableStore>) -> Result<StoreCensus, CensusError> {
    let before = store.store_counts().map(|handle| handle.snapshot());
    let sets = ConversationRegistry::new(Arc::clone(store))
        .read_sets()
        .map_err(|error| CensusError::Store(format!("the registry could not be read: {error}")))?;
    let spine = store.reclaim_spine();
    let shards = spine
        .as_ref()
        .map_or(1, |spine| spine.database().shard_count());
    let mut census = StoreCensus {
        live: sets.live.len(),
        sweeping: sets.sweeping.len(),
        shards,
        per_shard: BTreeMap::new(),
        legacy_locators: BTreeMap::new(),
        names_another: Vec::new(),
        pruned_below_rebuild: Vec::new(),
        counts: None,
    };
    let ids: BTreeSet<u64> = sets.live.union(&sets.sweeping).copied().collect();
    for id in ids {
        let route = checkpoint_conversation_route(id);
        let shard = spine
            .as_ref()
            .map_or(0, |spine| spine.database().shard_for(route.as_bytes()));
        *census.per_shard.entry(shard).or_insert(0) += 1;
        let own = read(store, &route, &CheckpointLocator::routed_key(id))?;
        let legacy = read(store, &route, CheckpointLocator::legacy_routed_key())?;
        let Some(legacy) = legacy else {
            continue;
        };
        let names = CheckpointLocator::decode_named(&legacy).map_err(|error| error.to_string());
        if let Ok(other) = names
            && own.is_none()
            && other != id
        {
            census.names_another.push((id, other));
            if pruned(store, &route, id)? {
                census.pruned_below_rebuild.push(id);
            }
        }
        census.legacy_locators.entry(shard).or_insert(names);
    }
    census.counts = match (before, store.store_counts()) {
        (Some(before), Some(handle)) => Some(handle.snapshot().since(&before)),
        _ => None,
    };
    Ok(census)
}

fn read(
    store: &Arc<dyn DurableStore>,
    route: &str,
    key: &[u8],
) -> Result<Option<Vec<u8>>, CensusError> {
    block_on(store.get_routed(route, key))
        .map_err(|error| CensusError::Store(format!("a routed read of {route} hung: {error}")))?
        .map_err(|error| CensusError::Store(format!("a routed read of {route} failed: {error}")))
}

/// Whether `id`'s operation log carries a prune receipt, under its namespace
/// or bare on the route (a store written before DUR-028), naming this
/// conversation and a cut above zero.
fn pruned(store: &Arc<dyn DurableStore>, route: &str, id: u64) -> Result<bool, CensusError> {
    let from = format!("{PRUNE_RECEIPT_PREFIX}operations:").into_bytes();
    let to = format!("{PRUNE_RECEIPT_PREFIX}operations;").into_bytes();
    let windows = [
        (from.clone(), to.clone()),
        (
            super::route_keys::namespaced(id, &from),
            super::route_keys::namespaced(id, &to),
        ),
    ];
    for (from, to) in windows {
        let ledger = |error: &dyn std::fmt::Display| {
            CensusError::Store(format!(
                "the prune ledger of {route} could not be read: {error}"
            ))
        };
        let receipts = block_on(store.scan_routed(route, &from, &to))
            .map_err(|error| ledger(&error))?
            .map_err(|error| ledger(&error))?;
        let cut = receipts.iter().any(|(_, value)| {
            serde_json::from_slice::<PruneReceipt>(value).is_ok_and(|receipt| {
                receipt.conversation_id == id
                    && receipt.stream == PrunedStream::Operations
                    && receipt.cut_before > 0
            })
        });
        if cut {
            return Ok(true);
        }
    }
    Ok(false)
}

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