liminal-server 0.24.0

//! Completion, observer reconciliation and replay of the production participant handler, moved out of `handler.rs` unchanged
//! so that file stays within the 500-line bound.

#[cfg(test)]
use super::outbox_log_error;
use super::{
    Arc, CapacityCounter, ConversationAuthority, ConversationId, LogAppender, Mutex,
    OperationBatch, OperationFacts, OperationLog, OperatorCredentialReissueOutcome,
    OperatorCredentialReissueRequest, OutboxLog, ParticipantConnectionContext,
    ParticipantConnectionConversations, ParticipantSemanticError, ProductionParticipantHandler,
    ReceiptCapacityLimits, RefCell, StateError, block_on, bridge_error,
    checkpoint_conversation_route, log_error, restore_error, state_error,
};

impl ProductionParticipantHandler {
    /// Completes the sources this operation appended.
    ///
    /// Every source writes its own Unit 2 extension row in place right after
    /// its base row, and a failed write already refused the operation, so a
    /// commit that reaches here owes nothing and needs only the commit tail.
    /// A nonzero owed count is an invariant break: the commit refuses by name,
    /// never by rebuilding the conversation from its log, and the next
    /// request's load writes the owed row.
    pub(super) fn complete_appended_source(
        &self,
        conversation_id: ConversationId,
        authority: &mut ConversationAuthority,
        appender: &LogAppender<'_>,
    ) -> Result<(), ParticipantSemanticError> {
        let owed = appender.outstanding_extension_rows();
        if owed != 0 {
            return Err(state_error(&StateError::ExtensionRowsOwed {
                conversation_id,
                owed,
            }));
        }
        self.complete_live_commit(conversation_id, authority, appender)
    }

    /// Completes one in-process commit without re-deriving the prefix.
    ///
    /// Board #60 §3c. This is the tail of [`Self::replay_and_repair`] applied
    /// to the live owner — the same calls, in the same order, minus the three
    /// passes whose only job was to rebuild a prefix this owner never lost:
    ///
    /// - `repair_pending_specific_fates` is KEPT. It is not a load-time repair
    ///   in disguise: a live Died/Detached commit can leave a pending specific
    ///   fate whose finalizer must be appended before the response publishes,
    ///   and today's post-append replay is what performs it.
    /// - `reconcile_observer_progress` is KEPT, over the owner's OWN witness
    ///   vector, which is now retained rather than drained. The planner reads
    ///   the whole source history (it validates the durable observer prefix
    ///   against it), and the retained vector is that history — the same
    ///   vector, source for source, that a from-zero replay would rebuild.
    /// - `prune_expired_provenance` and the capacity fold are KEPT: both are
    ///   per-touch clock work over the current owner, not prefix derivation.
    ///
    /// Deliberately NOT carried over: `validate_operation_schema`, the base-log
    /// replay, the extension merge, `validate_replayed_seal` and
    /// `reconcile_load_end_marker_anchors`. The first three re-derive a prefix
    /// the live owner already holds; the last two are load-end repairs of crash
    /// residue, and a source committed in-process under this lock has no crash
    /// residue to repair.
    pub(super) fn complete_live_commit(
        &self,
        conversation_id: ConversationId,
        authority: &mut ConversationAuthority,
        appender: &LogAppender<'_>,
    ) -> Result<(), ParticipantSemanticError> {
        authority
            .repair_pending_specific_fates(appender)
            .map_err(|error| state_error(&error))?;
        self.reconcile_live_observer(conversation_id, authority)?;
        let now = self
            .now_ms()
            .map_err(|error| ParticipantSemanticError::Internal {
                message: format!("participant clock read failed: {error}"),
            })?;
        let now = u128::from(now);
        authority.prune_expired_provenance(now);
        let contribution = authority
            .capacity_contribution(now)
            .map_err(|error| state_error(&error))?;
        self.capacity
            .fold_conversation(conversation_id, contribution)
            .map_err(|error| state_error(&error))?;
        Ok(())
    }

    /// Reconciles a live owner's observer progress with the durable observer
    /// store and prunes the witnesses that are now durable.
    pub(in super::super) fn reconcile_live_observer(
        &self,
        conversation_id: ConversationId,
        authority: &mut ConversationAuthority,
    ) -> Result<(), ParticipantSemanticError> {
        let observer_progress = authority.observer_progress;
        let pruned_through = authority.observer_progress_pruned_through();
        let witnesses = authority.observer_progress_witnesses();
        if !authority.tokens.is_empty() || authority.is_closed() {
            self.reconcile_observer_progress(
                conversation_id,
                witnesses,
                pruned_through,
                observer_progress,
            )?;
            authority
                .prune_durable_observer_witnesses()
                .map_err(|error| state_error(&error))?;
        } else if !witnesses.is_empty() {
            return Err(ParticipantSemanticError::Internal {
                message: format!(
                    "unenrolled conversation {conversation_id} projected observer progress"
                ),
            });
        }
        Ok(())
    }

    /// Completes the pending binding fates a load found, through one routed
    /// batch on the conversation's route committed once, the way every
    /// operation commits. The flat legacy-rows arm wrote them flat beside
    /// the routed rows, where the store refused them by sequence and the
    /// load failed; a test that arms legacy rows still writes them flat.
    pub(super) fn repair_loaded_fates(
        &self,
        conversation_id: ConversationId,
        replayed: &mut ConversationAuthority,
        log: &OperationLog,
        outbox_log: &OutboxLog,
    ) -> Result<(), ParticipantSemanticError> {
        let timestamp = self.operation_timestamp()?;
        let batch = RefCell::new(Some(OperationBatch::new(&checkpoint_conversation_route(
            conversation_id,
        ))));
        #[cfg(test)]
        let flat_legacy = super::super::log::legacy_rows::armed();
        #[cfg(not(test))]
        let flat_legacy = false;
        let appender = LogAppender::new(
            log,
            &self.registry,
            outbox_log,
            Some(super::super::prune::effective_keep(self.config.pruning)),
            (!flat_legacy).then_some(&batch),
            timestamp,
            &self.publisher,
        );
        replayed
            .repair_pending_specific_fates(&appender)
            .map_err(|error| state_error(&error))?;
        if appender.has_staged().map_err(|error| log_error(&error))? {
            appender
                .commit_staged()
                .map_err(|error| log_error(&error))?;
        }
        Ok(())
    }

    /// Cold-replays one conversation's durable log and repairs its observer
    /// registration.
    ///
    /// An enrolled conversation whose durable observer `Track` row was lost
    /// to a crash between the enrollment append and the tracking append is
    /// re-registered idempotently here, so observer recovery is derivable
    /// from the conversation log itself on any first touch.
    pub(in super::super) fn replay_and_repair(
        &self,
        conversation_id: ConversationId,
        log: &OperationLog,
    ) -> Result<ConversationAuthority, ParticipantSemanticError> {
        let outbox_log = OutboxLog::new(Arc::clone(&self.store), conversation_id);
        // The load the boot runs: the validated checkpoint and its routed
        // tail, or a full rebuild through the routed route. The legacy replay
        // walked the flat stream alone and saw none of the rows a live commit
        // stages routed, so an owner discarded after a failed operation came
        // back from its durable log empty, and a retry on a bound member was
        // answered ParticipantUnknown.
        let mut replayed = block_on(super::super::restore_checkpoint_rebuild::restore_at_boot(
            self.store.as_ref(),
            log,
            &outbox_log,
            &self.config,
            self.outbox_limits,
        ))
        .map_err(|error| bridge_error(&error))?
        .map_err(|error| restore_error(&error))?;
        self.repair_loaded_fates(conversation_id, &mut replayed, log, &outbox_log)?;
        let pruned_through = replayed.observer_progress_pruned_through();
        let observer_witnesses = replayed.observer_progress_witnesses();
        // F8B R-SEAL (§6.6). This is the THIRD site that reads `tokens` empty
        // as "never enrolled" — the two the ruling names are the replay
        // invariant twins. A Closed conversation is the state that proxy never
        // anticipated: no tokens, but a log full of records, terminals and
        // drain rows whose observer-progress witnesses are durable truth. It
        // must reconcile exactly as an enrolled conversation does, or the
        // handler's observer state is left behind its own durable log — the
        // failure §6.2's post-drain reconciliation exists to prevent. Bare
        // tokens-empty-with-witnesses REMAINS a refusal for the never-enrolled
        // shape.
        if !replayed.tokens.is_empty() || replayed.is_closed() {
            self.reconcile_observer_progress(
                conversation_id,
                observer_witnesses,
                pruned_through,
                replayed.observer_progress,
            )?;
            replayed
                .prune_durable_observer_witnesses()
                .map_err(|error| state_error(&error))?;
        } else if !observer_witnesses.is_empty() {
            return Err(ParticipantSemanticError::Internal {
                message: format!(
                    "unenrolled conversation {conversation_id} projected observer progress"
                ),
            });
        }
        // Request-time expiry over the replayed state (replay rebuilds every
        // retired rotation's fingerprint; the ones past their deadlines are
        // dropped under this touch's clock read), then fold the
        // conversation's complete server-scope contribution into the ledger
        // — a replace, so a discarded-and-replayed owner never double
        // counts and the ledger self-heals from durable truth.
        let now = self
            .now_ms()
            .map_err(|error| ParticipantSemanticError::Internal {
                message: format!("participant clock read failed: {error}"),
            })?;
        let now = u128::from(now);
        replayed.prune_expired_provenance(now);
        let contribution = replayed
            .capacity_contribution(now)
            .map_err(|error| state_error(&error))?;
        self.capacity
            .fold_conversation(conversation_id, contribution)
            .map_err(|error| state_error(&error))?;
        Ok(replayed)
    }

    /// Runs the frozen pre-W3 aggregate reference without installing any
    /// owner, observer, capacity, or publication state.
    #[cfg(test)]
    pub(in super::super) fn replay_aggregate_reference(
        &self,
        conversation_id: ConversationId,
        log: &OperationLog,
    ) -> Result<ConversationAuthority, ParticipantSemanticError> {
        let outbox_log = OutboxLog::new(Arc::clone(&self.store), conversation_id);
        let extension_rows = block_on(outbox_log.read_all())
            .map_err(|error| bridge_error(&error))?
            .map_err(|error| outbox_log_error(&error))?;
        block_on(ConversationAuthority::replay_aggregate_reference(
            conversation_id,
            log,
            &outbox_log,
            extension_rows,
            &self.config,
            self.outbox_limits,
        ))
        .map_err(|error| bridge_error(&error))?
        .map_err(|error| state_error(&error))
    }

    /// Removes a conversation's registry cell after a durably empty touch.
    ///
    /// Only the exact cell this operation used is removed (a racing request
    /// may have installed a fresh cell already); a concurrent holder of the
    /// evicted cell stays correct because every durable append is optimistic
    /// on its exact sequence and cold replay is the source of truth.
    pub(in super::super) fn evict_uncommitted(
        &self,
        conversation_id: ConversationId,
        cell: &Arc<Mutex<Option<ConversationAuthority>>>,
    ) -> Result<(), ParticipantSemanticError> {
        let mut conversations =
            self.conversations
                .lock()
                .map_err(|_| ParticipantSemanticError::Internal {
                    message: "participant conversation registry lock is poisoned".to_owned(),
                })?;
        if let Some(existing) = conversations.get(&conversation_id) {
            if Arc::ptr_eq(existing, cell) {
                conversations.remove(&conversation_id);
            }
        }
        drop(conversations);
        Ok(())
    }

    /// Number of live conversation registry cells (test observability).
    #[cfg(test)]
    pub(in super::super) fn registry_len(&self) -> usize {
        self.conversations
            .lock()
            .map_or(usize::MAX, |conversations| conversations.len())
    }

    /// Drops only volatile participant/observer owners for cold-first-touch tests.
    #[cfg(test)]
    pub(in super::super) fn discard_owners_for_test(&self) -> Result<(), ParticipantSemanticError> {
        self.conversations
            .lock()
            .map_err(|_| ParticipantSemanticError::Internal {
                message: "participant conversation registry lock is poisoned".to_owned(),
            })?
            .clear();
        self.observer
            .lock()
            .map_err(|_| ParticipantSemanticError::Internal {
                message: "observer recovery aggregate lock is poisoned".to_owned(),
            })?
            .take();
        Ok(())
    }

    /// Runs one `OperatorCredentialReissue` at the serialized
    /// participant-state point (R18 amendment A7, §0.18).
    ///
    /// It reaches the conversation through the SAME owner lock, cold-replay,
    /// and post-commit reconciliation seam as enrollment and credential
    /// attach — never a second path — so every guard reads the same exact
    /// state those operations read, and an unknown conversation is evicted
    /// with no residue exactly as a refused probe already is.
    ///
    /// # Errors
    ///
    /// Returns [`ParticipantSemanticError`] when the operation could not be
    /// DECIDED (a latched service fatal, an unloadable conversation, a durable
    /// failure). Every decided refusal is an `Ok`.
    pub(crate) fn operator_credential_reissue(
        &self,
        request: OperatorCredentialReissueRequest,
    ) -> Result<OperatorCredentialReissueOutcome, ParticipantSemanticError> {
        self.ensure_service_live()?;
        let now_ms = self
            .now_ms()
            .map_err(|error| ParticipantSemanticError::Internal {
                message: format!("participant clock read failed: {error}"),
            })?;
        self.with_conversation_fate_source(
            request.conversation_id,
            None,
            |authority, appender, _| {
                authority.apply_operator_credential_reissue(request, now_ms, appender)
            },
        )
    }

    pub(in super::super) fn operation_facts(
        &self,
        context: ParticipantConnectionContext,
        conversation_id: ConversationId,
        conversations: &ParticipantConnectionConversations,
    ) -> Result<OperationFacts, ParticipantSemanticError> {
        let now_ms = self
            .now_ms()
            .map_err(|error| ParticipantSemanticError::Internal {
                message: format!("participant clock read failed: {error}"),
            })?;
        // The connection map only grows through capacity commits, so its
        // occupancy always fits the validated nonzero signed limit; a counter
        // rejection here is genuine internal drift and fails closed.
        let connection_capacity = CapacityCounter::try_new(
            self.config.max_semantic_conversations_per_connection,
            conversations.occupied(),
        )
        .map_err(|error| ParticipantSemanticError::Internal {
            message: format!(
                "connection-conversation occupancy disagrees with its signed limit: {error:?}"
            ),
        })?;
        Ok(OperationFacts {
            receiving_incarnation: context.connection_incarnation(),
            now_ms,
            identity_slots: self.config.identity_slots,
            attach_receipt_ttl_ms: self.config.attach_receipt_ttl_ms,
            receipt_provenance_ttl_ms: self.config.receipt_provenance_ttl_ms,
            receipt_limits: ReceiptCapacityLimits::from_config(&self.config),
            connection_tracking: conversations.tracking(conversation_id),
            connection_capacity,
        })
    }
}