liminal-server 0.24.1

//! The log appender of the production participant handler, moved out of `handler.rs` unchanged
//! so that file stays within the 500-line bound.

use super::{
    Cell, CheckpointPublication, CheckpointPublisher, ConversationAuthority, ConversationId,
    ConversationRegistry, DispatchImpactAccumulator, DurableAppend, Mutex, MutexGuard,
    OperationBatch, OperationLog, OperationLogError, OutboxLog, OutboxLogError, ParticipantConfig,
    ParticipantSemanticError, PublicationBound, PublicationLayoutError, PublicationRefusal,
    RefCell, RoutedStreamRefusal, StateError, StoredOperation, block_on,
    checkpoint_conversation_route, owes_extension_row, publication_bound,
};

/// Bridges the synchronous state seam onto the async durable log, and keeps
/// the conversation registry complete by construction: the one
/// conversation-creating append (genesis at sequence zero) is preceded by a
/// durable registry row, so startup can enumerate every conversation stream
/// that exists.
pub(in super::super) struct LogAppender<'a> {
    pub(in super::super) log: &'a OperationLog,
    pub(in super::super) registry: &'a ConversationRegistry,
    pub(in super::super) conversation_id: ConversationId,
    /// Board #60 §3c. The Unit 2 extension log a committed source completes
    /// itself into, under the same conversation lock as its base append.
    pub(in super::super) outbox_log: &'a OutboxLog,
    /// Acknowledged rows kept behind a checkpoint this appender stages
    /// (DUR-018, DUR-024 R4: [`super::super::prune::effective_keep`]'s answer, the
    /// configured window or the pinned default); `None` for a drain that
    /// stages no cut of its own.
    pub(in super::super) prune_keep: Option<u64>,
    /// Owed-minus-written Unit 2 extension rows for the operation this
    /// appender is serving. Incremented by every appended row that owes one
    /// ([`owes_extension_row`]), decremented by every row written in place.
    /// Zero at the end of the operation is the OUTCOME GATE that lets the
    /// commit path skip its from-zero replay.
    pub(in super::super) outstanding_extension_rows: Cell<u64>,
    pub(in super::super) batch: Option<&'a RefCell<Option<OperationBatch>>>,
    pub(in super::super) timestamp: u64,
    /// Where a checkpoint too large for one batch is handed off (DUR-024).
    pub(in super::super) publisher: &'a CheckpointPublisher,
}

/// The conversation's owner cell, locked; a poisoned lock refuses by name.
pub(super) fn lock_owner(
    cell: &Mutex<Option<ConversationAuthority>>,
    conversation_id: ConversationId,
) -> Result<MutexGuard<'_, Option<ConversationAuthority>>, ParticipantSemanticError> {
    cell.lock().map_err(|_| ParticipantSemanticError::Internal {
        message: format!("participant conversation {conversation_id} owner lock is poisoned"),
    })
}

/// Installs the effects an operation staged on its impact accumulator, when
/// the caller gave one.
pub(super) fn install_staged(impact: Option<&mut DispatchImpactAccumulator>) {
    if let Some(impact) = impact {
        impact.install_staged();
    }
}

pub(super) fn checkpoint_state_error(
    error: super::super::restore_checkpoint::CheckpointError,
) -> StateError {
    OperationLogError::RoutedStream(RoutedStreamRefusal::Checkpoint(Box::new(error))).into()
}

impl<'a> LogAppender<'a> {
    /// An appender over `log`: routed through `batch` when one is given, flat
    /// (one append per row) when none is.
    pub(in super::super) const fn new(
        log: &'a OperationLog,
        registry: &'a ConversationRegistry,
        outbox_log: &'a OutboxLog,
        prune_keep: Option<u64>,
        batch: Option<&'a RefCell<Option<OperationBatch>>>,
        timestamp: u64,
        publisher: &'a CheckpointPublisher,
    ) -> Self {
        Self {
            log,
            registry,
            conversation_id: log.conversation_id(),
            outbox_log,
            prune_keep,
            outstanding_extension_rows: Cell::new(0),
            batch,
            timestamp,
            publisher,
        }
    }

    pub(super) fn finish_operation(
        &self,
        authority: &mut ConversationAuthority,
        starting_log_sequence: u64,
        reconcile_appended_source: bool,
        config: &ParticipantConfig,
    ) -> Result<(), StateError> {
        if reconcile_appended_source && authority.next_log_sequence > starting_log_sequence {
            authority.repair_pending_specific_fates(self)?;
            let owed = self.outstanding_extension_rows();
            if owed != 0 {
                return Err(StateError::ExtensionRowsOwed {
                    conversation_id: self.conversation_id,
                    owed,
                });
            }
        }
        if self.has_staged()? {
            self.commit_with_checkpoint(authority, starting_log_sequence, config, false)?;
        }
        Ok(())
    }

    /// Commits what is staged together with the checkpoint the crossing from
    /// `starting_log_sequence` owes (`checkpoint_due`: a period boundary or the
    /// closed marker), and installs the checkpoint metadata the commit proved.
    /// `force` stages the checkpoint with no crossing: the boot drain's owed
    /// publication of a rebuild that could not be captured (DUR-020 R6).
    pub(super) fn commit_with_checkpoint(
        &self,
        authority: &mut ConversationAuthority,
        starting_log_sequence: u64,
        config: &ParticipantConfig,
        force: bool,
    ) -> Result<(), StateError> {
        let (metadata, audit) =
            self.staged_checkpoint_metadata(authority, starting_log_sequence, config)?;
        let metadata = self.prepare_crossing_checkpoint(
            authority,
            starting_log_sequence,
            metadata,
            &audit,
            force,
        )?;
        // A checkpoint handed to the publisher stages nothing here; an
        // operation with no rows of its own (the boot drain's owed
        // publication) then has nothing to commit, and an empty commit is
        // an unsettled write that contains the room (11 October, 20:48Z).
        if self.has_staged()? {
            self.commit_staged()?;
        }
        let crossed = force
            || OperationBatch::checkpoint_due(
                starting_log_sequence,
                authority.next_log_sequence,
                authority.closed,
            )
            .map_err(OperationLogError::from)?;
        authority
            .install_checkpoint_metadata(metadata)
            .map_err(checkpoint_state_error)?;
        if crossed {
            self.publisher
                .crossing_committed(&crate::server::connection::CheckpointCrossing {
                    conversation_id: self.conversation_id,
                    covered_through: authority.next_log_sequence,
                    delivery_head: authority
                        .outbox
                        .as_ref()
                        .map_or(0, |outbox| outbox.checkpoint_boundaries().1),
                    recipients: authority.crossing_census(),
                });
        }
        Ok(())
    }

    pub(super) fn prepare_crossing_checkpoint(
        &self,
        authority: &mut ConversationAuthority,
        before: u64,
        mut metadata: super::super::restore_from_checkpoint::CheckpointMetadata,
        audit: &super::super::fenced_attach_terminal::ComposedTerminalValidation,
        force: bool,
    ) -> Result<super::super::restore_from_checkpoint::CheckpointMetadata, StateError> {
        use super::super::restore_checkpoint_history::CheckpointHistoryRoot;
        let mut root =
            CheckpointHistoryRoot::decode_canonical(metadata.history_root(), self.conversation_id)
                .map_err(|source| {
                    checkpoint_state_error(
                        super::super::restore_checkpoint::CheckpointError::LiveState {
                            source: Box::new(source),
                        },
                    )
                })?;
        let due = force
            || OperationBatch::checkpoint_due(
                before,
                authority.next_log_sequence,
                authority.closed,
            )
            .map_err(OperationLogError::from)?;
        // DUR-024 R3: a publication still in flight for this conversation
        // is left to finish; this crossing stages no checkpoint and the next
        // one retries over whatever the publisher committed.
        let due = due && !self.publication_in_flight();
        let mut staged = false;
        // A checkpoint the owner cannot frame is refused here, before
        // anything is staged, and the operation commits without it: the
        // owner serves on and the next crossing retries. A framing failure
        // never fails the operation (11 October, 21:16Z: the door room was
        // contained by its first post's crossing).
        let body_bytes = if due {
            self.capturable_body_bytes(authority)
        } else {
            None
        };
        if let Some(body_bytes) = body_bytes {
            let store = self.log.store();
            let bound = publication_bound(store.as_ref());
            let mut history = block_on(
                super::super::restore_checkpoint_history_capture::capture_identity_history(
                    store.as_ref(),
                    authority,
                    root,
                ),
            )?
            .map_err(|error| checkpoint_state_error(error.checkpoint()))?;
            // The operation's own commit carries the checkpoint only when
            // the history, the body and the operation's rows together fit
            // the bound. Measured on the history alone, the door room's
            // held Open 44 (11 October) put a 30 MB history and a 101 MB
            // body into one 131 MB write; the store refused it, the
            // outcome was unknown, and the room was contained.
            let inline_bytes = history
                .byte_total()
                .saturating_add(body_bytes)
                .saturating_add(self.staged_value_bytes()?);
            if inline_bytes <= bound.bound {
                // Under the bound: one commit with the operation's own rows,
                // byte for byte what every crossing committed before DUR-024.
                root = history.root;
                metadata = metadata
                    .with_history_root(root.encode_canonical())
                    .map_err(checkpoint_state_error)?;
                let pending = self.batch.ok_or(OperationLogError::RoutedStream(
                    RoutedStreamRefusal::BatchRefused,
                ))?;
                let mut held = pending.try_borrow_mut().map_err(|_| {
                    OperationLogError::RoutedStream(RoutedStreamRefusal::BatchRefused)
                })?;
                let pending = held.as_mut().ok_or(OperationLogError::RoutedStream(
                    RoutedStreamRefusal::BatchRefused,
                ))?;
                // The owner adopts the new root below, and with it the
                // digests of what that root holds, so the next crossing puts
                // only what changes before it.
                authority.history_digests = history.take_digests();
                history.stage(pending).map_err(OperationLogError::from)?;
                staged = true;
            } else {
                // Over the bound: the history, the body and the locator go
                // to the publisher in batches, off this write's path. The
                // owner keeps the root whose nodes are all present, so the
                // next crossing's capture stages only what is still absent.
                self.publisher.ledger().log_bound_once(bound);
                self.hand_off_publication(authority, &metadata, history, bound, audit)?;
            }
        }
        authority
            .retain_checkpoint_metadata(root, audit)
            .map_err(checkpoint_state_error)?;
        if staged {
            let extension = authority
                .outbox
                .as_ref()
                .ok_or_else(|| StateError::invariant("checkpoint outbox is absent"))?
                .next_extension_sequence();
            let positions = authority
                .checkpoint_positions(metadata.schema(), authority.next_log_sequence, extension)
                .map_err(checkpoint_state_error)?;
            self.stage_borrowed_checkpoint(
                positions,
                metadata.history_root(),
                authority,
                metadata.prefix_digest(),
            )?;
            self.stage_prune(authority, extension)?;
        }
        Ok(metadata)
    }

    /// The bytes of the owner's checkpoint body, or `None`, said by name,
    /// when the owner cannot be framed now.
    ///
    /// The positions the staged checkpoint will carry are validated here
    /// too, before anything is staged: a position refused after staging
    /// failed the operation and contained the room, where every other
    /// framing refusal lets the owner serve on (11 October 2026: a drained
    /// `Detached` terminal advances observer progress past the outbox's
    /// highest delivery, and once the drained suffix could be captured the
    /// position check was the first refusal it reached).
    pub(super) fn capturable_body_bytes(&self, authority: &ConversationAuthority) -> Option<usize> {
        let framed = authority
            .checkpoint_lifecycle_capturable()
            .and_then(|()| authority.checkpoint_positions_framable())
            .and_then(|()| {
                super::super::restore_checkpoint::CheckpointLiveState::encode_live(authority)
            });
        match framed {
            Ok(body) => Some(body.len()),
            Err(refusal) => {
                tracing::warn!(
                    conversation_id = self.conversation_id,
                    refusal = %refusal,
                    "checkpoint not captured at this crossing; the owner serves on and the next crossing retries"
                );
                None
            }
        }
    }

    /// Whether this conversation's publication is still in flight, said
    /// once per crossing that finds it so.
    pub(super) fn publication_in_flight(&self) -> bool {
        let in_flight = self.publisher.ledger().in_flight(self.conversation_id);
        if in_flight {
            tracing::info!(
                conversation_id = self.conversation_id,
                "checkpoint publication still in flight; this crossing stages no checkpoint and the next one retries"
            );
        }
        in_flight
    }

    /// DUR-024 R1 and R3: a history larger than the batch bound is laid out
    /// as batches and handed to the publisher, with the body and the
    /// locator (captured here, under the owner's lock, at this crossing's
    /// positions) and the prune receipts in the final batch. The owner's
    /// lock is released before anything is submitted, and the publisher
    /// never takes it. A publication refused before submission (one item
    /// over the store's bound, R2) is recorded owed and nothing is sent.
    pub(super) fn hand_off_publication(
        &self,
        authority: &mut ConversationAuthority,
        metadata: &super::super::restore_from_checkpoint::CheckpointMetadata,
        history: super::super::restore_checkpoint_history_capture::PreparedCheckpointHistory,
        bound: PublicationBound,
        audit: &super::super::fenced_attach_terminal::ComposedTerminalValidation,
    ) -> Result<(), StateError> {
        let store = self.log.store();
        let route = checkpoint_conversation_route(self.conversation_id);
        let (new_root, nodes) = history.into_nodes();
        // The body names the root its publication installs, and the owner's
        // live validation frames a body only against the root it retains;
        // so the new root is retained for the framing, and the caller
        // retains the old root again afterwards, the one whose nodes are
        // all present until this publication lands.
        authority
            .retain_checkpoint_metadata(new_root, audit)
            .map_err(checkpoint_state_error)?;
        let extension = authority
            .outbox
            .as_ref()
            .ok_or_else(|| StateError::invariant("checkpoint outbox is absent"))?
            .next_extension_sequence();
        let positions = authority
            .checkpoint_positions(metadata.schema(), authority.next_log_sequence, extension)
            .map_err(checkpoint_state_error)?;
        let mut final_batch = OperationBatch::new(&route);
        block_on(final_batch.stage_borrowed_checkpoint(
            store.as_ref(),
            positions,
            &new_root.encode_canonical(),
            authority,
            metadata.prefix_digest(),
        ))?
        .map_err(OperationLogError::from)?;
        if let Some(plan) = self.plan_prune(authority, extension)? {
            final_batch.stage_prune(plan)?;
        }
        if !self.publisher.ledger().begin(self.conversation_id) {
            return Ok(());
        }
        let publication = match CheckpointPublication::new(
            self.conversation_id,
            &route,
            nodes,
            final_batch,
            bound,
        ) {
            Ok(publication) => publication,
            Err(PublicationLayoutError::Refused(refusal)) => {
                let super::super::checkpoint_batches::CheckpointPublishError::ItemExceedsStoreBound {
                    bytes,
                    ..
                } = &refusal;
                self.publisher.refuse_before_submission(
                    self.conversation_id,
                    PublicationRefusal {
                        batch_index: 0,
                        bytes: *bytes,
                        bound: bound.bound,
                        refusal: refusal.to_string(),
                    },
                );
                return Ok(());
            }
            Err(PublicationLayoutError::Store(error)) => {
                self.publisher.refuse_before_submission(
                    self.conversation_id,
                    PublicationRefusal {
                        batch_index: 0,
                        bytes: 0,
                        bound: bound.bound,
                        refusal: error.to_string(),
                    },
                );
                return Err(OperationLogError::from(error).into());
            }
        };
        // A thread that could not start is recorded owed by `spawn` itself.
        drop(self.publisher.spawn(store, publication));
        Ok(())
    }
}

impl DurableAppend for LogAppender<'_> {
    fn append_extension(
        &self,
        row: &crate::server::participant::production::outbox_log::OutboxRow,
        expected_sequence: u64,
    ) -> Result<(), crate::server::participant::production::state::ExtensionRowCause> {
        if let Some(batch) = self.batch {
            let mut pending = batch.try_borrow_mut().map_err(|_| {
                super::super::state::ExtensionRowCause::Log(OutboxLogError::RoutedStream(
                    RoutedStreamRefusal::BatchRefused,
                ))
            })?;
            let pending = pending.as_mut().ok_or_else(|| {
                super::super::state::ExtensionRowCause::Log(OutboxLogError::RoutedStream(
                    RoutedStreamRefusal::BatchRefused,
                ))
            })?;
            return block_on(self.outbox_log.stage_into(
                pending,
                row,
                expected_sequence,
                self.timestamp,
            ))
            .map_err(super::super::state::ExtensionRowCause::Bridge)?
            .map_err(super::super::state::ExtensionRowCause::Log);
        }
        let Some(log) = self.extension_log() else {
            return Err(
                crate::server::participant::production::state::ExtensionRowCause::Log(
                    crate::server::participant::production::outbox_log::OutboxLogError::Durability(
                        liminal::durability::DurabilityError::BatchUnsupported {
                            store: std::any::type_name::<Self>(),
                        },
                    ),
                ),
            );
        };
        liminal::durability::bridge::block_on(log.append_flat(row, expected_sequence))
            .map_err(crate::server::participant::production::state::ExtensionRowCause::Bridge)?
            .map_err(crate::server::participant::production::state::ExtensionRowCause::Log)
    }
    fn append(
        &self,
        operation: &StoredOperation,
        expected_sequence: u64,
    ) -> Result<(), OperationLogError> {
        if expected_sequence == 0 && matches!(operation, StoredOperation::Genesis { .. }) {
            self.registry
                .register_with_issuance(self.conversation_id, self.log.issuance())?;
        }
        if let Some(batch) = self.batch {
            let mut pending = batch
                .try_borrow_mut()
                .map_err(|_| OperationLogError::RoutedStream(RoutedStreamRefusal::BatchRefused))?;
            let pending = pending.as_mut().ok_or(OperationLogError::RoutedStream(
                RoutedStreamRefusal::BatchRefused,
            ))?;
            block_on(
                self.log
                    .stage_into(pending, operation, expected_sequence, self.timestamp),
            )??;
        } else {
            block_on(self.log.append_flat(operation, expected_sequence))??;
        }
        if owes_extension_row(operation) {
            self.outstanding_extension_rows
                .set(self.outstanding_extension_rows.get().saturating_add(1));
        }
        Ok(())
    }

    fn extension_log(&self) -> Option<&OutboxLog> {
        Some(self.outbox_log)
    }

    fn owed_extension_rows(&self) -> u64 {
        self.outstanding_extension_rows.get()
    }

    fn discharge_owed_extension_row(&self) {
        self.outstanding_extension_rows
            .set(self.outstanding_extension_rows.get().saturating_sub(1));
    }
}