liminal-server 0.24.9

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

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

use super::checkpoint_batches::{
    CheckpointPublishError, batch_bytes, publication_bound, split_history,
};
use super::log::{OperationLog, OperationLogError, RETIRED_GENERATION};
use super::operation_batch::{CheckpointStageError, OperationBatch};
use super::outbox_log::{OutboxLog, OutboxLogError};
use super::restore_checkpoint::{CheckpointError, CheckpointLiveState, RestoreCheckpoint};
pub(super) use super::restore_checkpoint_boot_error::BootCheckpointError;
use super::restore_checkpoint_history::{CheckpointHistoryRoot, ROOT_ENCODED_BYTES};
use super::restore_checkpoint_history_capture::PreparedCheckpointHistory;
use super::restore_checkpoint_locator::{CheckpointLocator, 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),
    /// DUR-024 R2: one item alone is over the store's bound.
    #[error(transparent)]
    Refused(#[from] CheckpointPublishError),
    /// DUR-024 R1: a history batch the store refused, by its 1-based index.
    #[error("history batch {index} refused: {source}")]
    Batch {
        index: usize,
        #[source]
        source: DurabilityError,
    },
}

/// A rebuilt publication that did not complete, with the fresh checkpoint
/// handed back when the refusal came after it was prepared, so the owner
/// serves from its replay (DUR-024 R3) instead of being lost with the error.
pub(super) struct RebuiltPublicationFailure<L, const N: usize> {
    pub(super) error: CheckpointRebuildError,
    pub(super) checkpoint: Option<Box<RestoreCheckpoint<L, N>>>,
}

impl<L, const N: usize> RebuiltPublicationFailure<L, N> {
    /// A failure after the checkpoint was prepared: the owner is handed
    /// back to serve from its replay.
    fn after(
        error: impl Into<CheckpointRebuildError>,
        checkpoint: RestoreCheckpoint<L, N>,
    ) -> Self {
        Self {
            error: error.into(),
            checkpoint: Some(Box::new(checkpoint)),
        }
    }
}

/// 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. DUR-024 R1: a history over the batch bound is committed in
/// batches first, each under the bound, and the heads, the body and the
/// locator ride in the final commit only.
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>, RebuiltPublicationFailure<L, N>> {
    let RebuiltCheckpointPublication {
        checkpoint,
        history,
        prefix_digest,
    } = publication;
    let refused_conversation_id = reported.required.conversation_id;
    let (positions, root, live) = checkpoint.into_parts();
    let fresh = match reported.required.prepare_fresh(positions, root, live) {
        Ok(fresh) => fresh,
        Err(error) => {
            return Err(RebuiltPublicationFailure {
                error: error.into(),
                checkpoint: None,
            });
        }
    };
    let checkpoint = fresh.checkpoint;
    if positions.conversation_id != refused_conversation_id
        || positions.conversation_id != log.conversation_id()
    {
        return Err(RebuiltPublicationFailure {
            error: CheckpointRebuildError::ConversationMismatch,
            checkpoint: Some(Box::new(checkpoint)),
        });
    }
    if history.root.encode_canonical().as_slice() != checkpoint.history_root().as_slice() {
        return Err(RebuiltPublicationFailure {
            error: CheckpointRebuildError::Fresh(
                super::restore_checkpoint::CheckpointError::Association {
                    field: "rebuilt HISTORY root",
                },
            ),
            checkpoint: Some(Box::new(checkpoint)),
        });
    }
    let route = checkpoint_conversation_route(positions.conversation_id);
    let commit = RebuiltCommit {
        history,
        prefix_digest,
        observed_locator,
        route: &route,
    };
    let (commits, checkpoint) = commit_rebuilt(store, log, outbox_log, checkpoint, commit).await?;
    super::restore_checkpoint_history::retire_v1_history(store, positions.conversation_id).await;
    tracing::info!(
        conversation_id = positions.conversation_id,
        refusal = %fresh.refusal,
        commits,
        "CHECKPOINT_REBUILD: replayed heads and fresh checkpoint are durably published"
    );
    Ok(checkpoint)
}

/// What a rebuilt publication commits beside the checkpoint itself: the
/// prepared history, the digest of the prefix it covers, the locator observed
/// before the rebuild (the conditional expectation) and the conversation's
/// route.
struct RebuiltCommit<'a> {
    history: PreparedCheckpointHistory,
    prefix_digest: [u8; 32],
    observed_locator: Option<&'a [u8]>,
    route: &'a str,
}

/// The commits of a rebuilt publication: history batches under the bound,
/// then the final batch with the heads, the body and the locator and
/// whatever history fitted beside them. Answers how many commits it made,
/// with the checkpoint handed back either way: the owner it carries serves
/// from its replay when the store refuses (R3). The checkpoint is owned
/// here, never borrowed across an await, so the future stays `Send`.
async fn commit_rebuilt<L: CheckpointLiveState + Send, const N: usize>(
    store: &dyn DurableStore,
    log: &OperationLog,
    outbox_log: &OutboxLog,
    checkpoint: RestoreCheckpoint<L, N>,
    commit: RebuiltCommit<'_>,
) -> Result<(usize, RestoreCheckpoint<L, N>), RebuiltPublicationFailure<L, N>> {
    let RebuiltCommit {
        history,
        prefix_digest,
        observed_locator,
        route,
    } = commit;
    let positions = checkpoint.positions();
    let bound = publication_bound(store);
    let (_, mut nodes) = history.into_nodes();
    let mut batch = OperationBatch::new(route);
    if let Err(error) = log
        .stage_bootstrap_head(&mut batch, positions.next_log_sequence)
        .await
    {
        return Err(RebuiltPublicationFailure::after(error, checkpoint));
    }
    if let Err(error) = outbox_log
        .stage_bootstrap_head(&mut batch, positions.extension_sequence)
        .await
    {
        return Err(RebuiltPublicationFailure::after(error, checkpoint));
    }
    let staged =
        batch.stage_rebuilt_checkpoint(store, &checkpoint, prefix_digest, observed_locator);
    if let Err(error) = staged.await {
        return Err(RebuiltPublicationFailure::after(error, checkpoint));
    }
    // A body's parts go with the history, before the locator.
    nodes.extend(
        batch
            .take_body_parts()
            .into_iter()
            .map(|part| (part.key, part.value)),
    );
    let mut batches = match split_history(nodes, bound.bound, bound.item_limit()) {
        Ok(batches) => batches,
        Err(error) => return Err(RebuiltPublicationFailure::after(error, checkpoint)),
    };
    let tail = batch.value_bytes();
    if tail > bound.item_limit() {
        let refusal = CheckpointPublishError::ItemExceedsStoreBound {
            key: "checkpoint body".to_owned(),
            bytes: tail,
            bound: bound.item_limit(),
        };
        return Err(RebuiltPublicationFailure::after(refusal, checkpoint));
    }
    if batches
        .last()
        .is_some_and(|last| batch_bytes(last).saturating_add(tail) <= bound.bound)
    {
        if let Some(last) = batches.pop() {
            for item in last {
                if let Err(error) = batch.stage(item) {
                    return Err(RebuiltPublicationFailure::after(error, checkpoint));
                }
            }
        }
    }
    let commits = batches.len().saturating_add(1);
    for (index, items) in batches.into_iter().enumerate() {
        if let Err(source) = apply_routed_batch_settled(store, route, items).await {
            let error = CheckpointRebuildError::Batch {
                index: index.saturating_add(1),
                source,
            };
            return Err(RebuiltPublicationFailure::after(error, checkpoint));
        }
    }
    if let Err(error) = batch
        .commit(store, || log.issuance().mark_submitted())
        .await
    {
        return Err(RebuiltPublicationFailure::after(error, checkpoint));
    }
    Ok((commits, checkpoint))
}

/// What a load left published for its conversation (DUR-020 R6).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) enum BootCheckpoint {
    /// A validated checkpoint was restored, or the conversation has no rows.
    Restored,
    /// The rebuild published a checkpoint.
    Published,
    /// The rebuilt owner could not be captured, so nothing was published and
    /// the next load rebuilds again unless the restore unit publishes it once
    /// its drain has finished the fate that refused the capture.
    Unpublished,
}

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> {
    Box::pin(restore_at_boot_reporting(
        store, log, outbox_log, config, limits,
    ))
    .await
    .map(|(owner, _)| owner)
}

/// [`restore_at_boot`], also answering what the load left published.
pub(super) async fn restore_at_boot_reporting(
    store: &dyn DurableStore,
    log: &OperationLog,
    outbox_log: &OutboxLog,
    config: &crate::config::types::ParticipantConfig,
    limits: super::outbox::ConversationOutboxLimits,
) -> Result<(super::state::ConversationAuthority, BootCheckpoint), BootCheckpointError> {
    let conversation_id = log.conversation_id();
    // The tombstone first: a retired conversation refuses by name before any
    // checkpoint or row is read, never as a rebuild's failure. The key is
    // read once; every read below is on the generation it named.
    let generation = log.generation().await?;
    if generation == RETIRED_GENERATION {
        return Err(BootCheckpointError::Retired { conversation_id });
    }
    let pin = &OperationLog::at_generation(log.store(), conversation_id, generation);
    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) => {
                    // The root alone, so a restore reads the same at any
                    // history length; each identity is cross-checked by the
                    // warm once the conversation serves.
                    match super::restore_checkpoint_history_capture::verify_history_root(
                        store,
                        prepared.root,
                    )
                    .await
                    {
                        // The body's identities, before the tail replays
                        // onto them: what the index at this root holds.
                        Ok(()) => super::restore_checkpoint_history_capture::capture_history_check(
                            &prepared.owner,
                            prepared.root,
                        )
                        .map(|check| {
                            let mut prepared = prepared;
                            prepared.owner.checkpoint_history_check = Some(check);
                            prepared
                        })
                        .map_err(super::restore_checkpoint_history_capture::HistoryCaptureError::checkpoint),
                        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(pin, outbox_log, config, prepared))
            .await
            .map(|owner| (owner, BootCheckpoint::Restored)),
        // Boxed as the restore arm is: the rebuild's future carries the
        // batches it lays out, and the boot stack is not where it lives.
        Err(required) => {
            Box::pin(rebuild_before_ready(
                store,
                (pin, 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)
}

/// `logs` is the reader pinned to the unit's generation and the operation's
/// own handle, whose issuance the publication marks.
async fn rebuild_before_ready(
    store: &dyn DurableStore,
    logs: (&OperationLog, &OperationLog),
    outbox_log: &OutboxLog,
    config: &crate::config::types::ParticipantConfig,
    limits: super::outbox::ConversationOutboxLimits,
    required: RebuildRequired,
    observed_locator: Option<&[u8]>,
) -> Result<(ConversationAuthority, BootCheckpoint), BootCheckpointError> {
    let (pin, log) = logs;
    let conversation_id = log.conversation_id();
    let (owner, audit, schema, prefix) =
        ConversationAuthority::replay_for_rebuild(conversation_id, pin, 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, BootCheckpoint::Restored));
    }
    if let Err(refusal) = owner
        .checkpoint_lifecycle_capturable()
        .and_then(|()| owner.checkpoint_positions_framable())
    {
        // 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.
        // Why the stored checkpoint was not used is said too: a full replay
        // with no stored-checkpoint reason beside it was silent about the
        // re-replay it caused (11 October 2026).
        tracing::warn!(
            conversation_id,
            refusal = %refusal,
            checkpoint_refusal = %required.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.
        return serve_from_replay(owner, &audit, schema, prefix, conversation_id);
    }
    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 = match publish_rebuilt_checkpoint(
        reported,
        store,
        log,
        outbox_log,
        RebuiltCheckpointPublication {
            checkpoint,
            history,
            prefix_digest: prefix,
        },
        observed_locator,
    )
    .await
    {
        Ok(checkpoint) => checkpoint,
        Err(RebuiltPublicationFailure {
            error,
            checkpoint: Some(checkpoint),
        }) => {
            // DUR-024 R3: a publication the store refused leaves the owner
            // serving from its replay, exactly as an uncapturable owner
            // serves; the next crossing publishes, over whatever batches
            // landed. Never a full replay, never a containment, for this.
            tracing::warn!(
                conversation_id,
                refusal = %error,
                "CHECKPOINT_REBUILD: the rebuilt checkpoint's publication was refused by the store; the owner serves from its replay and the next crossing publishes its checkpoint"
            );
            let (_, _, owner) = (*checkpoint).into_parts();
            return serve_from_replay(owner, &audit, schema, prefix, conversation_id);
        }
        Err(RebuiltPublicationFailure { error, .. }) => return Err(error.into()),
    };
    let metadata = CheckpointMetadata::from_rebuilt(&checkpoint, prefix)?;
    let (_, _, mut owner) = checkpoint.into_parts();
    owner.install_checkpoint_metadata(metadata)?;
    Ok((owner, BootCheckpoint::Published))
}

/// The owner serving from its replay with nothing published: validated
/// metadata over the empty history root, so the crossing that publishes
/// captures the identity history from nothing exactly as the rebuild would
/// have (DUR-020 R6, DUR-024 R3).
fn serve_from_replay(
    mut owner: ConversationAuthority,
    audit: &super::fenced_attach_terminal::ComposedTerminalValidation,
    schema: super::restore_checkpoint::CheckpointSchema,
    prefix: [u8; 32],
    conversation_id: u64,
) -> Result<(ConversationAuthority, BootCheckpoint), BootCheckpointError> {
    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)?;
    Ok((owner, BootCheckpoint::Unpublished))
}