liminal-server 0.24.11

//! The checkpoint half of an [`OperationBatch`]: the body, its parts and
//! the locator staged behind the operation's rows, and (DUR-028 R3) the
//! body and parts the new locator supersedes removed in the same commit.

use haematite::tree::Hash;

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

use super::super::restore_checkpoint::CheckpointPositions;
use super::super::restore_checkpoint_locator::{
    CheckpointBodyKey, CheckpointLocator, checkpoint_conversation_route,
};
use super::{CheckpointStageError, OperationBatch};

impl OperationBatch {
    pub(super) async fn prepare_checkpoint(
        &mut self,
        store: &dyn DurableStore,
        positions: CheckpointPositions,
        body: Vec<u8>,
        prefix_digest: [u8; 32],
        rebuilt: bool,
        observed_locator: Option<&[u8]>,
    ) -> Result<(), CheckpointStageError> {
        if self.route_key != checkpoint_conversation_route(positions.conversation_id) {
            return Err(CheckpointStageError::RouteMismatch);
        }
        let body_key = CheckpointBodyKey::for_body(&body);
        let locator = CheckpointLocator::new(
            positions.conversation_id,
            positions.covered_log_sequence,
            positions.extension_sequence,
            prefix_digest,
            body_key,
        )?;
        let locator_key = CheckpointLocator::routed_key(positions.conversation_id);
        let prior = if rebuilt {
            observed_locator.map(<[u8]>::to_vec)
        } else {
            store.get_routed(&self.route_key, &locator_key).await?
        };
        // The locator this one replaces, when its bytes read: a rebuild
        // replaces corrupt bytes it cannot read, and then has no body to
        // name for removal.
        let previous = match prior.as_deref() {
            None => None,
            Some(bytes) if rebuilt => {
                CheckpointLocator::decode_canonical(bytes, positions.conversation_id).ok()
            }
            Some(bytes) => {
                let previous =
                    CheckpointLocator::decode_canonical(bytes, positions.conversation_id)?;
                if previous.covered_sequence() > positions.covered_log_sequence
                    || previous.extension_boundary() > positions.extension_sequence
                {
                    return Err(CheckpointStageError::PositionRegression);
                }
                Some(previous)
            }
        };
        let key = body_key.routed_key(positions.conversation_id);
        let layout = super::super::checkpoint_body_parts::lay_out(positions.conversation_id, body);
        let kept: Vec<Vec<u8>> = layout.parts.iter().map(|(key, _)| key.clone()).collect();
        let existing_body = store.get_routed(&self.route_key, &key).await?;
        if !rebuilt
            && existing_body
                .as_ref()
                .is_some_and(|bytes| bytes != &layout.head)
        {
            return Err(CheckpointStageError::BodyMismatch);
        }
        self.stage(RoutedBatchItem {
            key: key.clone(),
            expected: existing_body.as_deref().map(Hash::of),
            value: layout.head,
            ttl: None,
        })?;
        for (part_key, part) in layout.parts {
            // A content-addressed part already present is these bytes.
            if store
                .get_routed(&self.route_key, &part_key)
                .await?
                .is_none()
            {
                self.body_parts.push(RoutedBatchItem {
                    key: part_key,
                    expected: None,
                    value: part,
                    ttl: None,
                });
            }
        }
        if let Some(previous) = previous {
            self.stage_superseded(
                store,
                positions.conversation_id,
                previous.body_key(),
                &key,
                &kept,
            )
            .await?;
        }
        self.stage(RoutedBatchItem {
            key: locator_key,
            expected: prior.as_deref().map(Hash::of),
            value: locator.encode_canonical().to_vec(),
            ttl: None,
        })?;
        Ok(())
    }

    /// Stages the removal of the body the prior locator names when the new
    /// body is not it, and of every part its manifest names that the new
    /// manifest (`kept`) does not, each by the hash of the bytes it holds,
    /// in this batch (DUR-028 R3). The set is the difference of the two
    /// manifests the commit reads; no window is scanned. A prior body
    /// already absent names nothing to remove.
    async fn stage_superseded(
        &mut self,
        store: &dyn DurableStore,
        conversation_id: u64,
        prior: CheckpointBodyKey,
        body_key: &[u8],
        kept: &[Vec<u8>],
    ) -> Result<(), CheckpointStageError> {
        let prior_key = prior.routed_key(conversation_id);
        if prior_key == body_key {
            return Ok(());
        }
        let Some(head) = store.get_routed(&self.route_key, &prior_key).await? else {
            return Ok(());
        };
        for (part_key, expected) in
            super::super::checkpoint_body_parts::superseded_parts(conversation_id, &head, kept)
        {
            self.stage_delete(part_key, expected)?;
        }
        self.stage_delete(prior_key, Hash::of(&head))?;
        Ok(())
    }
}