liminal-server 0.20.1

//! Publication of a loudly rebuilt checkpoint before its owner becomes ready.

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

use super::log::{OperationLog, OperationLogError};
use super::operation_batch::{CheckpointStageError, OperationBatch};
use super::outbox_log::{OutboxLog, OutboxLogError};
use super::restore_checkpoint::{CheckpointError, CheckpointLiveState, RestoreCheckpoint};
use super::restore_checkpoint_history::{CheckpointHistoryRoot, ROOT_ENCODED_BYTES};
use super::restore_checkpoint_history_capture::PreparedCheckpointHistory;
use super::restore_checkpoint_locator::CheckpointLocator;
use super::restore_checkpoint_locator::checkpoint_conversation_route;
use super::restore_from_checkpoint::{
    CheckpointDecision, CheckpointMetadata, CheckpointRefusal, PreparedCheckpointRestore,
    RebuildRequired, TailReplayError, load_checkpoint, prepare_checkpoint_restore,
    prepare_replayed_checkpoint,
};
use super::routed_stream::RoutedStreamRefusal;
use super::state::ConversationAuthority;
use super::state::checkpoint_owner::ConversationAuthorityRestoreContext;

pub(super) struct ReportedRebuild {
    required: RebuildRequired,
}

#[derive(Debug, thiserror::Error)]
pub(super) enum CheckpointRebuildError {
    #[error("rebuilt checkpoint conversation differs from its refused locator or operation log")]
    ConversationMismatch,
    #[error(transparent)]
    Operation(#[from] OperationLogError),
    #[error(transparent)]
    Extension(#[from] OutboxLogError),
    #[error(transparent)]
    Checkpoint(#[from] CheckpointStageError),
    #[error(transparent)]
    Store(#[from] DurabilityError),
    #[error(transparent)]
    Fresh(#[from] super::restore_checkpoint::CheckpointError),
}

/// Records the exact refusal before the caller starts authoritative replay.
pub(super) fn report_rebuild(required: RebuildRequired) -> ReportedRebuild {
    tracing::warn!(
        conversation_id = required.conversation_id,
        refusal = %required.refusal,
        "CHECKPOINT_REBUILD: full authoritative replay is required before readiness"
    );
    ReportedRebuild { required }
}

pub(super) struct RebuiltCheckpointPublication<L, const N: usize> {
    checkpoint: RestoreCheckpoint<L, N>,
    history: PreparedCheckpointHistory,
    prefix_digest: [u8; 32],
}

/// Publishes replayed heads and returns the same owner after the conditional operation.
pub(super) async fn publish_rebuilt_checkpoint<L: CheckpointLiveState + Send, const N: usize>(
    reported: ReportedRebuild,
    store: &dyn DurableStore,
    log: &OperationLog,
    outbox_log: &OutboxLog,
    publication: RebuiltCheckpointPublication<L, N>,
    observed_locator: Option<&[u8]>,
) -> Result<RestoreCheckpoint<L, N>, CheckpointRebuildError> {
    let RebuiltCheckpointPublication {
        checkpoint,
        history,
        prefix_digest,
    } = publication;
    let refused_conversation_id = reported.required.conversation_id;
    let (positions, root, live) = checkpoint.into_parts();
    let fresh = reported.required.prepare_fresh(positions, root, live)?;
    let checkpoint = fresh.checkpoint;
    if positions.conversation_id != refused_conversation_id
        || positions.conversation_id != log.conversation_id()
    {
        return Err(CheckpointRebuildError::ConversationMismatch);
    }
    let route = checkpoint_conversation_route(positions.conversation_id);
    if history.root.encode_canonical().as_slice() != checkpoint.history_root().as_slice() {
        return Err(CheckpointRebuildError::Fresh(
            super::restore_checkpoint::CheckpointError::Association {
                field: "rebuilt HISTORY root",
            },
        ));
    }
    let mut batch = OperationBatch::new(&route);
    history.stage(&mut batch)?;
    log.stage_bootstrap_head(&mut batch, positions.next_log_sequence)
        .await?;
    outbox_log
        .stage_bootstrap_head(&mut batch, positions.extension_sequence)
        .await?;
    batch
        .stage_rebuilt_checkpoint(store, &checkpoint, prefix_digest, observed_locator)
        .await?;
    batch
        .commit(store, || log.issuance().mark_submitted())
        .await?;
    tracing::info!(
        conversation_id = positions.conversation_id,
        refusal = %fresh.refusal,
        "CHECKPOINT_REBUILD: replayed heads and fresh checkpoint are durably published"
    );
    Ok(checkpoint)
}

#[derive(Debug, thiserror::Error)]
pub(super) enum BootCheckpointError {
    #[error(transparent)]
    Store(#[from] DurabilityError),
    #[error(transparent)]
    Checkpoint(#[from] super::restore_checkpoint::CheckpointError),
    #[error(transparent)]
    Locator(#[from] super::restore_checkpoint_locator::CheckpointLocatorError),
    #[error(transparent)]
    Replay(#[from] super::outbox_replay::RestoreError),
    #[error(transparent)]
    Operation(#[from] OperationLogError),
    #[error(transparent)]
    Publish(#[from] CheckpointRebuildError),
}

pub(super) async fn restore_at_boot(
    store: &dyn DurableStore,
    log: &OperationLog,
    outbox_log: &OutboxLog,
    config: &crate::config::types::ParticipantConfig,
    limits: super::outbox::ConversationOutboxLimits,
) -> Result<super::state::ConversationAuthority, BootCheckpointError> {
    let conversation_id = log.conversation_id();
    let loaded = load_checkpoint::<ConversationAuthority, ROOT_ENCODED_BYTES>(
        store,
        conversation_id,
        |positions, root| ConversationAuthorityRestoreContext::new(*positions, root, config),
        |locator, root| {
            CheckpointHistoryRoot::decode_canonical(root, locator.conversation_id())
                .map(|_| ())
                .map_err(|source| CheckpointError::LiveState {
                    source: Box::new(source),
                })
        },
    )
    .await?;
    let prepared =
        match loaded.decision {
            CheckpointDecision::Restore(checkpoint) => {
                let preparation =
                    (|| {
                        let raw = loaded.observed_locator.as_deref().ok_or(
                            CheckpointError::Association {
                                field: "validated checkpoint locator",
                            },
                        )?;
                        let locator = CheckpointLocator::decode_canonical(raw, conversation_id)
                            .map_err(|error| CheckpointError::LiveState {
                                source: Box::new(error),
                            })?;
                        prepare_checkpoint_restore(checkpoint, &locator, config)
                    })();
                let preparation = match preparation {
                Ok(prepared) => {
                    match super::restore_checkpoint_history_capture::validate_identity_history(
                        store,
                        &prepared.owner,
                        prepared.root,
                    )
                    .await
                    {
                        Ok(()) => Ok(prepared),
                        Err(
                            super::restore_checkpoint_history_capture::HistoryCaptureError::Index(
                                super::restore_checkpoint_history_index::HistoryIndexError::Store(
                                    error,
                                ),
                            ),
                        ) => {
                            return Err(BootCheckpointError::Store(error));
                        }
                        Err(error) => Err(error.checkpoint()),
                    }
                }
                Err(error) => Err(error),
            };
                preparation.map_err(|error| RebuildRequired {
                    conversation_id,
                    refusal: CheckpointRefusal::Invalid(error),
                })
            }
            CheckpointDecision::Rebuild(required) => Err(required),
        };
    match prepared {
        Ok(prepared) => Box::pin(restore_validated_tail(log, outbox_log, config, prepared)).await,
        Err(required) => {
            rebuild_before_ready(
                store,
                log,
                outbox_log,
                config,
                limits,
                required,
                loaded.observed_locator.as_deref(),
            )
            .await
        }
    }
}

async fn restore_validated_tail(
    log: &OperationLog,
    outbox_log: &OutboxLog,
    config: &crate::config::types::ParticipantConfig,
    prepared: PreparedCheckpointRestore,
) -> Result<ConversationAuthority, BootCheckpointError> {
    let conversation_id = log.conversation_id();
    let root = prepared.root;
    let owner = prepared.owner;
    let generation = log.generation().await?;
    let native = OperationLog::at_generation(log.store(), conversation_id, generation);
    let native_head = native.read_stream_next_seq().await?.unwrap_or(0);
    let route = checkpoint_conversation_route(conversation_id);
    let source = |sequence| {
        let native = &native;
        let route = &route;
        async move {
            if sequence < native_head {
                native.read_at(sequence).await
            } else {
                // A boundary with no routed tail behind it selects no row,
                // the answer `OperationLog::read_committed_at` gives: a
                // migrated or flat-only stream has a native head and no
                // routed head, and its tail is simply empty.
                match log.read_routed_at(route, sequence).await {
                    Err(OperationLogError::RoutedStream(RoutedStreamRefusal::MissingHead {
                        ..
                    })) => Ok(None),
                    read => read,
                }
            }
        }
    };
    let (mut owner, audit, schema, prefix) = owner
        .replay_checkpoint_tail(
            prepared.start,
            (prepared.audit, prepared.prefix_digest),
            source,
            log,
            outbox_log,
            config,
        )
        .await
        .map_err(|error| match error {
            TailReplayError::Replay(error) => BootCheckpointError::Replay(error),
            TailReplayError::Source(error) => BootCheckpointError::Operation(error),
        })?;
    let extension = owner
        .outbox
        .as_ref()
        .ok_or(CheckpointError::Position {
            field: "restored outbox",
        })?
        .checkpoint_boundaries()
        .0;
    let positions = owner.checkpoint_positions(schema, owner.next_log_sequence, extension)?;
    owner.retain_checkpoint_metadata(root, &audit)?;
    owner.advance_checkpoint_metadata(positions, root.encode_canonical(), prefix)?;
    Ok(owner)
}

async fn rebuild_before_ready(
    store: &dyn DurableStore,
    log: &OperationLog,
    outbox_log: &OutboxLog,
    config: &crate::config::types::ParticipantConfig,
    limits: super::outbox::ConversationOutboxLimits,
    required: RebuildRequired,
    observed_locator: Option<&[u8]>,
) -> Result<ConversationAuthority, BootCheckpointError> {
    let conversation_id = log.conversation_id();
    let (owner, audit, schema, prefix) =
        ConversationAuthority::replay_for_rebuild(conversation_id, log, outbox_log, config, limits)
            .await?;
    if owner.next_log_sequence == 0 {
        // A conversation with no rows has nothing to rebuild and nothing to
        // report: this is every first touch of a fresh id on the request path.
        return Ok(owner);
    }
    if let Err(refusal) = owner.checkpoint_lifecycle_capturable() {
        // The rebuilt owner is mid-finalization: a binding fate the drain
        // has yet to finish leaves its slot, frontier or retained origin in
        // a state the checkpoint's own round-trip refuses. A load is not the
        // place to refuse it. The owner serves from its replay, as the
        // legacy replay served it, the repair and the drain finish the fate,
        // and the next crossing or seal publishes the checkpoint; a boot
        // before then rebuilds once more through the routed route. The
        // refusal is said by name so a capture refused for any other reason
        // is never silent.
        tracing::warn!(
            conversation_id,
            refusal = %refusal,
            "CHECKPOINT_REBUILD: the rebuilt owner cannot be captured yet; it serves from its replay and the next crossing publishes its checkpoint"
        );
        // The owner commits only against validated checkpoint metadata, so it
        // keeps its replay's audit, positions and prefix the way a restored
        // tail does, over the empty history root: nothing is published, and
        // the crossing that publishes captures the identity history from
        // nothing exactly as this rebuild would have.
        let mut owner = owner;
        let root = CheckpointHistoryRoot::empty(conversation_id);
        let extension_sequence = owner
            .outbox
            .as_ref()
            .ok_or(super::restore_checkpoint::CheckpointError::Position {
                field: "absent replayed outbox",
            })?
            .next_extension_sequence();
        let positions =
            owner.checkpoint_positions(schema, owner.next_log_sequence, extension_sequence)?;
        owner.retain_checkpoint_metadata(root, &audit)?;
        let metadata = CheckpointMetadata::validated(positions, root.encode_canonical(), prefix)?;
        owner.install_checkpoint_metadata(metadata)?;
        return Ok(owner);
    }
    let reported = report_rebuild(required);
    let history = super::restore_checkpoint_history_capture::capture_identity_history(
        store,
        &owner,
        CheckpointHistoryRoot::empty(conversation_id),
    )
    .await
    .map_err(|error| BootCheckpointError::Checkpoint(error.checkpoint()))?;
    let root = history.root;
    let checkpoint = prepare_replayed_checkpoint(owner, &audit, schema, root, prefix)?;
    let checkpoint = publish_rebuilt_checkpoint(
        reported,
        store,
        log,
        outbox_log,
        RebuiltCheckpointPublication {
            checkpoint,
            history,
            prefix_digest: prefix,
        },
        observed_locator,
    )
    .await?;
    let metadata = CheckpointMetadata::from_rebuilt(&checkpoint, prefix)?;
    let (_, _, mut owner) = checkpoint.into_parts();
    owner.install_checkpoint_metadata(metadata)?;
    Ok(owner)
}