liminal-server 0.24.9

//! A checkpoint body larger than one part is stored in parts (DUR-024).
//!
//! The door room's body reached 101 MB on 11 October 2026: one item over
//! the store's 64 MiB bound, refused however the batches around it were
//! laid out. A body over [`BODY_PART_BYTES`] is cut into numbered parts,
//! each stored under its own content address on the conversation's route,
//! and the body key holds a manifest naming the parts in order. The parts
//! are published like history nodes, in the batches before the locator, so
//! the locator still lands last and names a body every part of which is
//! durable. A body of one part is stored whole under the body key, byte for
//! byte as before.
//!
//! The body key stays the digest of the whole body, so a reader that
//! reassembles the parts checks the same digest the locator names.

use liminal::durability::{DurabilityError, DurableStore};

use super::restore_checkpoint_locator::CheckpointBodyKey;

/// The largest part a body is cut into: half the 16 MiB batch target, so
/// two parts pack in one publication batch and every part is far under any
/// store bound.
pub(super) const BODY_PART_BYTES: usize = 8 * 1024 * 1024;

/// What a manifest begins with. A body is never mistaken for a manifest:
/// the reader takes a value as the whole body when its digest matches the
/// body key, and reads it as a manifest only when it does not.
const MANIFEST_MAGIC: &[u8; 20] = b"liminal-body-parts-1";

/// The routed key prefix every body part is written under.
pub(super) const PART_KEY_PREFIX: &[u8] = b"checkpoint/body/part/v1/";

/// One body, laid out for staging: what goes under the body key, and the
/// parts (key and bytes) a manifest names.
pub(super) struct BodyLayout {
    pub(super) head: Vec<u8>,
    pub(super) parts: Vec<(Vec<u8>, Vec<u8>)>,
}

/// The routed key of the part with these bytes, under the conversation's
/// namespace (DUR-028 R2).
fn part_key(conversation_id: u64, part: &[u8]) -> Vec<u8> {
    part_key_of(conversation_id, blake3::hash(part).as_bytes())
}

/// The routed key of the part with this digest, under the namespace.
fn part_key_of(conversation_id: u64, digest: &[u8]) -> Vec<u8> {
    super::route_keys::namespaced(conversation_id, &legacy_part_key_of(digest))
}

/// The bare key a store written before DUR-028 holds the part with this
/// digest under, read only by the migration (R4).
pub(super) fn legacy_part_key_of(digest: &[u8]) -> Vec<u8> {
    let mut key = PART_KEY_PREFIX.to_vec();
    key.extend_from_slice(digest);
    key
}

/// The part keys a prior body head names that `kept` (the new manifest's
/// part keys) does not, each with the logical hash of the bytes it holds:
/// a part is stored under its own blake3 digest, which is the hash a delete
/// names (DUR-028 R3). A head that is a whole body names no part.
pub(super) fn superseded_parts(
    conversation_id: u64,
    prior_head: &[u8],
    kept: &[Vec<u8>],
) -> Vec<(Vec<u8>, haematite::tree::Hash)> {
    let Some((_, digests)) = manifest_digests(prior_head) else {
        return Vec::new();
    };
    digests
        .chunks(blake3::OUT_LEN)
        .filter_map(|digest| {
            let key = part_key_of(conversation_id, digest);
            let digest: [u8; blake3::OUT_LEN] = digest.try_into().ok()?;
            (!kept.contains(&key)).then_some((key, haematite::tree::Hash::from_bytes(digest)))
        })
        .collect()
}

/// The part keys a legacy (bare) manifest names, in order: what the
/// migration copies under the namespace (R4).
pub(super) fn legacy_part_keys(head: &[u8]) -> Vec<Vec<u8>> {
    manifest_digests(head).map_or_else(Vec::new, |(_, digests)| {
        digests
            .chunks(blake3::OUT_LEN)
            .map(legacy_part_key_of)
            .collect()
    })
}

/// Lays `body` out: whole when it fits one part, otherwise a manifest of
/// its total length and its parts' digests, in order, and the parts.
pub(super) fn lay_out(conversation_id: u64, body: Vec<u8>) -> BodyLayout {
    if body.len() <= BODY_PART_BYTES {
        return BodyLayout {
            head: body,
            parts: Vec::new(),
        };
    }
    let parts: Vec<(Vec<u8>, Vec<u8>)> = body
        .chunks(BODY_PART_BYTES)
        .map(|part| (part_key(conversation_id, part), part.to_vec()))
        .collect();
    let mut head = MANIFEST_MAGIC.to_vec();
    head.extend_from_slice(&u64::try_from(body.len()).unwrap_or(u64::MAX).to_be_bytes());
    for (_, part) in &parts {
        head.extend_from_slice(blake3::hash(part).as_bytes());
    }
    BodyLayout { head, parts }
}

/// The total length and the part keys a manifest names, or `None` when the
/// value is not a manifest.
fn manifest(conversation_id: u64, value: &[u8]) -> Option<(usize, Vec<Vec<u8>>)> {
    let (total, digests) = manifest_digests(value)?;
    let keys = digests
        .chunks(blake3::OUT_LEN)
        .map(|digest| part_key_of(conversation_id, digest))
        .collect();
    Some((total, keys))
}

/// The total length and the digests (concatenated, in order) a manifest
/// names, or `None` when the value is not a manifest.
fn manifest_digests(value: &[u8]) -> Option<(usize, &[u8])> {
    let rest = value.strip_prefix(MANIFEST_MAGIC.as_slice())?;
    let (total, digests) = rest.split_at_checked(std::mem::size_of::<u64>())?;
    let total = usize::try_from(u64::from_be_bytes(total.try_into().ok()?)).ok()?;
    if digests.is_empty() || digests.len() % blake3::OUT_LEN != 0 {
        return None;
    }
    Some((total, digests))
}

/// The body under `body_key` on the conversation's route: the value itself when it is the
/// whole body, the parts reassembled when it is a manifest, and `None` when
/// the body key or any part is absent. A part whose bytes do not match its
/// address, or a reassembly of the wrong length, answers the value as read,
/// so the caller's digest check refuses it by name.
///
/// # Errors
/// The store's own failure.
pub(super) async fn read_body(
    store: &dyn DurableStore,
    conversation_id: u64,
    body_key: CheckpointBodyKey,
) -> Result<Option<Vec<u8>>, DurabilityError> {
    let route = &super::restore_checkpoint_locator::checkpoint_conversation_route(conversation_id);
    let Some(head) = store
        .get_routed(route, &body_key.routed_key(conversation_id))
        .await?
    else {
        return Ok(None);
    };
    if body_key.verify_body(&head).is_ok() {
        return Ok(Some(head));
    }
    let Some((total, keys)) = manifest(conversation_id, &head) else {
        return Ok(Some(head));
    };
    let mut body = Vec::with_capacity(total);
    for key in keys {
        let Some(part) = store.get_routed(route, &key).await? else {
            return Ok(None);
        };
        if part_key(conversation_id, &part) != key {
            return Ok(Some(head));
        }
        body.extend_from_slice(&part);
    }
    if body.len() != total {
        return Ok(Some(head));
    }
    Ok(Some(body))
}

/// [`read_body`] from a test that holds no runtime.
///
/// # Errors
/// The bridge's or the store's failure.
#[cfg(test)]
pub(super) fn read_body_for_tests(
    store: &dyn DurableStore,
    conversation_id: u64,
    body_key: CheckpointBodyKey,
) -> Result<Option<Vec<u8>>, Box<dyn std::error::Error>> {
    Ok(liminal::durability::bridge::block_on(read_body(
        store,
        conversation_id,
        body_key,
    ))??)
}

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