liminal-server 0.24.5

//! DUR-028 R4: a store written before this brief is brought under the
//! namespace once, at boot, before the listener binds.
//!
//! Before DUR-028 R2 every conversation on a shard wrote its checkpoint
//! locator, body, body parts and prune receipts under bare keys, so a shard
//! held one locator and it named one conversation. This migration runs
//! once: for each id the registry restores, one [`OperationBatch`] on the
//! conversation's route, one durable commit. An id whose namespaced locator
//! is present is skipped. Otherwise, when the legacy locator decodes for
//! this id, its body head, its manifest's parts and the legacy receipts
//! naming this id are written under the namespace, the namespaced locator
//! is written, and the legacy locator is deleted by the hash of its bytes,
//! all in that batch. A legacy locator that names another conversation
//! copies nothing: the conversation restores as one without a checkpoint,
//! and when its log cannot rebuild it the restore refuses it by name as
//! unloadable. After every id, each of the four legacy windows is cut on
//! each shard an id's route reached, then the marker is written on the
//! registry route. A boot that finds the marker makes that one read and
//! nothing else.
//!
//! The migration is resumable: a crash after one conversation's batch
//! leaves it migrated (its namespaced locator is present, so the next boot
//! skips it) and the rest untouched; the marker is written last.
//!
//! A store whose registry restores no conversation can hold no legacy
//! checkpoint, so its boot reads nothing and writes nothing here: a fresh
//! store's start costs the same as before this brief. The marker is read
//! only when there is a conversation it could be about, and a store this
//! code wrote already holds it: the registry's first registration batch
//! carries [`fresh_marker`], so a populated boot makes that one read and
//! never the two reads per conversation and the batch a migration costs.
//! Only a store written before R2 and not yet booted by this code pays
//! the migration, once.

use std::collections::BTreeMap;
use std::sync::Arc;

use haematite::tree::Hash;

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

use super::checkpoint_body_parts::{PART_KEY_PREFIX, legacy_part_keys};
use super::operation_batch::OperationBatch;
use super::prune::{PRUNE_RECEIPT_PREFIX, PruneReceipt};
use super::registry::REGISTRY_STREAM_KEY;
use super::restore_checkpoint_locator::{
    CheckpointBodyKey, CheckpointLocator, checkpoint_conversation_route,
};
use super::route_keys::namespaced;

/// The key on the registry route that says the migration ran to its end,
/// or that the store never needed it.
pub(super) const MARKER_KEY: &[u8] = b"registry/route-keys-migrated/v1";

/// The marker a store written under the namespace from its first row
/// carries: it rides the registry's first registration batch (the
/// registry's row at sequence 0), so the only store a boot migrates is one
/// written before DUR-028 R2 and never booted since. Its value says so;
/// the key's presence is all a boot reads.
pub(super) fn fresh_marker() -> RoutedBatchItem {
    RoutedBatchItem {
        key: MARKER_KEY.to_vec(),
        expected: None,
        value: b"written under the namespace from the first registration".to_vec(),
        ttl: None,
    }
}

/// A store read or write the migration could not make. Nothing is claimed
/// migrated past it: the next boot resumes from what is durable.
#[derive(Debug, thiserror::Error)]
#[error("route key migration: {0}")]
pub(super) struct RouteKeyMigrationError(String);

/// What one run of the migration did, by id.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub(super) struct RouteKeyMigration {
    /// Ids whose legacy checkpoint is now under their namespace.
    pub(super) migrated: Vec<u64>,
    /// Ids already under their namespace when the run found them.
    pub(super) skipped: u64,
    /// Ids with no legacy locator of their own: they restore from their log.
    pub(super) without_checkpoint: Vec<u64>,
    /// Ids whose legacy checkpoint could not be copied, with why.
    pub(super) refused: Vec<(u64, String)>,
    /// The store's counters over the run, when the store counts them.
    pub(super) counts: Option<StoreCounts>,
}

impl RouteKeyMigration {
    /// The one line the migration logs.
    pub(super) fn line(&self) -> String {
        let counts = self.counts.as_ref().map_or_else(String::new, |counts| {
            format!(
                "; store counts: get_routed={} scan_page={} apply_routed_batch={} truncate={}",
                counts.get_routed_calls,
                counts.scan_page_calls,
                counts.apply_routed_batch_calls,
                counts.truncate_calls
            )
        });
        format!(
            "route key migration: {} migrated, {} skipped, {} without a checkpoint, {} refused{counts}",
            self.migrated.len(),
            self.skipped,
            self.without_checkpoint.len(),
            self.refused.len()
        )
    }
}

/// What one id's batch did.
enum Outcome {
    Migrated,
    Skipped,
    WithoutCheckpoint,
    Refused(String),
}

/// Where a route's keys live: the shard it hashes to when the store says,
/// else the route itself, so a window is cut once per place.
#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)]
enum Reached {
    Shard(usize),
    Route(String),
}

/// Runs the migration over `ids` unless the marker is present; answers
/// `None` when it is (one read, nothing else), and `None` without a read
/// when `ids` is empty: nothing registered means nothing to migrate.
///
/// # Errors
/// A store read or write that failed, named; what committed stays
/// committed and the next boot resumes.
pub(super) fn migrate_route_keys(
    store: &Arc<dyn DurableStore>,
    ids: &[u64],
) -> Result<Option<RouteKeyMigration>, RouteKeyMigrationError> {
    if ids.is_empty() || read(store, REGISTRY_STREAM_KEY, MARKER_KEY)?.is_some() {
        return Ok(None);
    }
    let before = store.store_counts().map(|handle| handle.snapshot());
    let spine = store.reclaim_spine();
    let mut report = RouteKeyMigration::default();
    let mut reached: BTreeMap<Reached, String> = BTreeMap::new();
    for &id in ids {
        let route = checkpoint_conversation_route(id);
        match migrate_one(store, id, &route)? {
            Outcome::Migrated => report.migrated.push(id),
            Outcome::Skipped => report.skipped += 1,
            Outcome::WithoutCheckpoint => report.without_checkpoint.push(id),
            Outcome::Refused(reason) => report.refused.push((id, reason)),
        }
        let place = spine.as_ref().map_or_else(
            || Reached::Route(route.clone()),
            |spine| Reached::Shard(spine.database().shard_for(route.as_bytes())),
        );
        reached.entry(place).or_insert(route);
    }
    for route in reached.values() {
        for (from, to) in legacy_windows() {
            block_on(store.remove_routed_range(route, &from, &to))
                .map_err(|error| hung("cutting a legacy window", route, &error))?
                .map_err(|error| failed("cutting a legacy window", route, &error))?;
        }
    }
    report.counts = match (before, store.store_counts()) {
        (Some(before), Some(handle)) => Some(handle.snapshot().since(&before)),
        _ => None,
    };
    let line = report.line();
    let mut marker = OperationBatch::new(REGISTRY_STREAM_KEY);
    marker
        .stage(RoutedBatchItem {
            key: MARKER_KEY.to_vec(),
            expected: None,
            value: line.clone().into_bytes(),
            ttl: None,
        })
        .map_err(|error| failed("staging the marker", REGISTRY_STREAM_KEY, &error))?;
    block_on(marker.commit(store.as_ref(), || {}))
        .map_err(|error| hung("writing the marker", REGISTRY_STREAM_KEY, &error))?
        .map_err(|error| failed("writing the marker", REGISTRY_STREAM_KEY, &error))?;
    tracing::info!(
        migrated = report.migrated.len(),
        skipped = report.skipped,
        without_checkpoint = report.without_checkpoint.len(),
        refused = report.refused.len(),
        "{line}"
    );
    Ok(Some(report))
}

/// One id: skipped, copied under its namespace in one batch, left to its
/// log, or refused with why.
fn migrate_one(
    store: &Arc<dyn DurableStore>,
    id: u64,
    route: &str,
) -> Result<Outcome, RouteKeyMigrationError> {
    if read(store, route, &CheckpointLocator::routed_key(id))?.is_some() {
        return Ok(Outcome::Skipped);
    }
    let Some(legacy) = read(store, route, CheckpointLocator::legacy_routed_key())? else {
        return Ok(Outcome::WithoutCheckpoint);
    };
    let named = match CheckpointLocator::decode_named(&legacy) {
        Ok(named) => named,
        Err(error) => {
            return Ok(Outcome::Refused(format!(
                "legacy locator does not read: {error}"
            )));
        }
    };
    if named != id {
        return Ok(Outcome::WithoutCheckpoint);
    }
    let locator = match CheckpointLocator::decode_canonical(&legacy, id) {
        Ok(locator) => locator,
        Err(error) => {
            return Ok(Outcome::Refused(format!(
                "legacy locator does not read: {error}"
            )));
        }
    };
    let body_key = locator.body_key();
    let Some(head) = read(store, route, &body_key.legacy_routed_key())? else {
        return Ok(Outcome::Refused(
            "legacy locator names a body the route does not hold".to_owned(),
        ));
    };
    let mut batch = OperationBatch::new(route);
    let mut put = |key: Vec<u8>, value: Vec<u8>| {
        batch
            .stage(RoutedBatchItem {
                key,
                expected: None,
                value,
                ttl: None,
            })
            .map_err(|error| failed("staging a namespaced key", route, &error))
    };
    for legacy_part in legacy_part_keys(&head) {
        let Some(part) = read(store, route, &legacy_part)? else {
            return Ok(Outcome::Refused(
                "legacy manifest names a part the route does not hold".to_owned(),
            ));
        };
        put(namespaced(id, &legacy_part), part)?;
    }
    put(body_key.routed_key(id), head)?;
    let (from, to) = receipt_window();
    let receipts = block_on(store.scan_routed(route, &from, &to))
        .map_err(|error| hung("reading the legacy prune ledger", route, &error))?
        .map_err(|error| failed("reading the legacy prune ledger", route, &error))?;
    for (key, value) in receipts {
        let names_id = serde_json::from_slice::<PruneReceipt>(&value)
            .is_ok_and(|receipt| receipt.conversation_id == id);
        if names_id {
            put(namespaced(id, &key), value)?;
        }
    }
    put(CheckpointLocator::routed_key(id), legacy.clone())?;
    batch
        .stage_delete(
            CheckpointLocator::legacy_routed_key().to_vec(),
            Hash::of(&legacy),
        )
        .map_err(|error| failed("staging the legacy locator's removal", route, &error))?;
    block_on(batch.commit(store.as_ref(), || {}))
        .map_err(|error| hung("committing the namespaced checkpoint", route, &error))?
        .map_err(|error| failed("committing the namespaced checkpoint", route, &error))?;
    Ok(Outcome::Migrated)
}

/// The four families a store written before DUR-028 holds bare on a route:
/// the locator, bodies, body parts and prune receipts, each as the half-open
/// window of its prefix.
pub(super) fn legacy_windows() -> [(Vec<u8>, Vec<u8>); 4] {
    [
        past(CheckpointLocator::legacy_routed_key()),
        past(CheckpointBodyKey::legacy_key_prefix()),
        past(PART_KEY_PREFIX),
        receipt_window(),
    ]
}

/// The window of every key under `prefix`: the prefix, and the prefix with
/// its last byte raised by one.
pub(super) fn past(prefix: &[u8]) -> (Vec<u8>, Vec<u8>) {
    let from = prefix.to_vec();
    let mut to = from.clone();
    if let Some(last) = to.last_mut() {
        *last = last.saturating_add(1);
    }
    (from, to)
}

/// The window of every bare prune receipt (`prune:` to `prune;`).
fn receipt_window() -> (Vec<u8>, Vec<u8>) {
    past(PRUNE_RECEIPT_PREFIX.as_bytes())
}

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

fn hung(step: &str, route: &str, error: &dyn std::fmt::Display) -> RouteKeyMigrationError {
    RouteKeyMigrationError(format!("{step} on {route} hung: {error}"))
}

fn failed(step: &str, route: &str, error: &dyn std::fmt::Display) -> RouteKeyMigrationError {
    RouteKeyMigrationError(format!("{step} on {route} failed: {error}"))
}