liminal-server 0.24.9

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

use super::{
    Arc, ConversationAuthority, ConversationCell, ConversationId, DispatchImpactAccumulator,
    DurableAppend, HashMap, LifecycleReceipt, LogAppender, Mutex, MutexGuard, OperationBatch,
    OperationLog, OutboxLog, OwedReceipts, ParticipantSemanticError, ProductionParticipantHandler,
    RefCell, StateError, checkpoint_conversation_route, install_staged, lock_owner,
    state_error_with_issuance,
};
#[cfg(test)]
use super::{BTreeMap, Ordering, log_error};

impl ProductionParticipantHandler {
    /// Every conversation this node has refused as unloadable, against the
    /// load failure's own text.
    ///
    /// In-process view for the tests that assert on containment attribution.
    /// The operator's view of the same record is
    /// `GET /unloadable-conversations`, served from the clone published by
    /// [`Self::unloadable_record`]; this accessor exists so a test can read the
    /// record without standing up an HTTP endpoint.
    #[cfg(test)]
    pub(in super::super) fn unloadable_conversations(&self) -> BTreeMap<ConversationId, String> {
        self.unloadable
            .snapshot()
            .into_iter()
            .map(|entry| (entry.conversation_id, entry.reason))
            .collect()
    }

    #[cfg(test)]
    pub(in super::super) fn registered_conversation_ids(
        &self,
    ) -> Result<Vec<ConversationId>, ParticipantSemanticError> {
        self.registry.restore().map_err(|error| log_error(&error))
    }

    /// Every cell this process holds, with its conversation id, taken under
    /// one short registry lock and locked by the caller one at a time.
    pub(in super::super) fn conversation_cells(
        &self,
    ) -> Result<Vec<(ConversationId, ConversationCell)>, ParticipantSemanticError> {
        let conversations =
            self.conversations
                .lock()
                .map_err(|_| ParticipantSemanticError::Internal {
                    message: "participant conversation registry lock is poisoned".to_owned(),
                })?;
        Ok(conversations
            .iter()
            .map(|(conversation_id, cell)| (*conversation_id, Arc::clone(cell)))
            .collect())
    }

    pub(in super::super) fn existing_cell(
        &self,
        conversation_id: ConversationId,
    ) -> Result<Option<Arc<Mutex<Option<ConversationAuthority>>>>, ParticipantSemanticError> {
        // DUR-020 R2: a pending conversation is restored before any reader
        // sees its cell, so "no cell" keeps meaning "no such conversation".
        self.ensure_restored(conversation_id)?;
        let conversations =
            self.conversations
                .lock()
                .map_err(|_| ParticipantSemanticError::Internal {
                    message: "participant conversation registry lock is poisoned".to_owned(),
                })?;
        Ok(conversations.get(&conversation_id).map(Arc::clone))
    }

    pub(in super::super) fn cell(
        &self,
        conversation_id: ConversationId,
    ) -> Result<Arc<Mutex<Option<ConversationAuthority>>>, ParticipantSemanticError> {
        // DUR-020 R2: every path that reaches a conversation's cell reaches
        // it through here or `existing_cell`, so the first touch, whatever
        // touches, runs the conversation's restore unit first.
        self.ensure_restored(conversation_id)?;
        #[cfg(test)]
        self.owner_lock_takes.fetch_add(1, Ordering::SeqCst);
        let mut conversations =
            self.conversations
                .lock()
                .map_err(|_| ParticipantSemanticError::Internal {
                    message: "participant conversation registry lock is poisoned".to_owned(),
                })?;
        let cell = Arc::clone(
            conversations
                .entry(conversation_id)
                .or_insert_with(|| Arc::new(Mutex::new(None))),
        );
        drop(conversations);
        Ok(cell)
    }

    pub(in super::super) fn with_conversation_impact<T>(
        &self,
        conversation_id: ConversationId,
        impact: &mut DispatchImpactAccumulator,
        operation: impl FnOnce(
            &mut ConversationAuthority,
            &dyn DurableAppend,
            &mut DispatchImpactAccumulator,
        ) -> Result<T, StateError>,
    ) -> Result<T, ParticipantSemanticError> {
        self.with_conversation_reconciliation(
            conversation_id,
            true,
            Some(impact),
            |authority, appender, impact| {
                impact.map_or_else(
                    || Err(StateError::invariant("impact owner is unavailable")),
                    |impact| operation(authority, appender, impact),
                )
            },
        )
    }

    /// Runs a fate-source append and reconciles its exact Unit 2 projection before
    /// retaining the transitioned live owner. Died/Detached joined the exhaustive
    /// v3 replay projection pass with W1b, so this boundary must use the same
    /// post-append repair as semantic sources to keep live and cold owners equal.
    pub(in super::super) fn with_conversation_fate_source<T>(
        &self,
        conversation_id: ConversationId,
        impact: Option<&mut DispatchImpactAccumulator>,
        operation: impl FnOnce(
            &mut ConversationAuthority,
            &dyn DurableAppend,
            Option<&mut DispatchImpactAccumulator>,
        ) -> Result<T, StateError>,
    ) -> Result<T, ParticipantSemanticError> {
        self.with_conversation_reconciliation(conversation_id, true, impact, operation)
    }

    pub(super) fn operation_timestamp(&self) -> Result<u64, ParticipantSemanticError> {
        self.now_ms()
            .map_err(|error| ParticipantSemanticError::Internal {
                message: format!("participant clock read failed: {error}"),
            })?
            .checked_mul(1_000_000)
            .ok_or_else(|| ParticipantSemanticError::Internal {
                message: "participant timestamp exceeds nanoseconds".to_owned(),
            })
    }

    pub(super) fn with_conversation_reconciliation<T>(
        &self,
        conversation_id: ConversationId,
        reconcile_appended_source: bool,
        mut impact: Option<&mut DispatchImpactAccumulator>,
        operation: impl FnOnce(
            &mut ConversationAuthority,
            &dyn DurableAppend,
            Option<&mut DispatchImpactAccumulator>,
        ) -> Result<T, StateError>,
    ) -> Result<T, ParticipantSemanticError> {
        let cell = self.cell(conversation_id)?;
        let mut owner = lock_owner(&cell, conversation_id)?;
        let log = OperationLog::new(Arc::clone(&self.store), conversation_id);
        let outbox_log = OutboxLog::new(Arc::clone(&self.store), conversation_id);
        let mut owed = Vec::new();
        if owner.is_none() {
            // The request-side half of containment. A conversation the node
            // cannot load refuses HERE, by name, instead of surfacing a bare
            // decode failure that names no subject at all.
            // The load reads and writes through its own log handle: a
            // rebuild's checkpoint publication is the load's settled write,
            // not this request's, and must not mark the request's write as
            // issued. Sharing the handle turned a request refused before it
            // wrote anything into "write outcome unknown" whenever its load
            // had published (11 October 2026).
            let load_log = OperationLog::new(Arc::clone(&self.store), conversation_id);
            let replayed = self
                .replay_and_repair(conversation_id, &load_log)
                .map_err(|error| self.record_unloadable(conversation_id, &error))?;
            self.clear_unloadable(conversation_id);
            owed.extend(self.settle_owed_receipts(conversation_id, replayed.next_log_sequence)?);
            *owner = Some(replayed);
        }
        let Some(authority) = owner.as_mut() else {
            return Err(ParticipantSemanticError::Internal {
                message: format!("participant conversation {conversation_id} owner is absent"),
            });
        };
        let timestamp = self.operation_timestamp()?;
        let batch = RefCell::new(Some(OperationBatch::new(&checkpoint_conversation_route(
            conversation_id,
        ))));
        // A test that arms legacy rows writes them where a legacy store in the
        // field holds them: flat, one append per row, never the routed batch.
        #[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,
        );
        let starting_log_sequence = authority.next_log_sequence;
        let operation_result =
            operation(authority, &appender, impact.as_deref_mut()).and_then(|value| {
                appender.finish_operation(
                    authority,
                    starting_log_sequence,
                    reconcile_appended_source,
                    &self.config,
                )?;
                Ok(value)
            });
        let mut receipts = owed;
        let (result, durably_empty) = match operation_result {
            Ok(value)
                if reconcile_appended_source
                    && authority.next_log_sequence > starting_log_sequence =>
            {
                // The v2 source barrier crossed. Complete its exact Unit 2
                // projection under this same conversation lock before the
                // caller can publish the correlated terminal response.
                match self.complete_appended_source(conversation_id, authority, &appender) {
                    Ok(()) => {
                        install_staged(impact.as_deref_mut());
                        let durably_empty = authority.next_log_sequence == 0;
                        (Ok(value), durably_empty)
                    }
                    Err(error) => {
                        // The committed rows are durable; this owner may not
                        // be. The next request loads the conversation, which
                        // completes any row this one left owed.
                        install_staged(impact.as_deref_mut());
                        receipts.extend(std::mem::take(&mut authority.lifecycle_receipts));
                        *owner = None;
                        (
                            Err(ParticipantSemanticError::WriteOutcomeUnknown {
                                message: format!("committed participant follow-up failed: {error}"),
                            }),
                            false,
                        )
                    }
                }
            }
            Ok(value) => {
                install_staged(impact.as_deref_mut());
                let durably_empty = authority.next_log_sequence == 0;
                (Ok(value), durably_empty)
            }
            Err(error) => {
                // Pending rows and receipts have no publication authority.
                // The next request restores the store's actual committed state.
                if let Some(impact) = impact {
                    impact.discard_staged();
                }
                let durably_empty =
                    self.fail_operation(conversation_id, starting_log_sequence, authority, &log)?;
                *owner = None;
                (
                    Err(state_error_with_issuance(&error, log.issuance())),
                    durably_empty,
                )
            }
        };
        if let Some(authority) = owner.as_mut() {
            receipts.extend(std::mem::take(&mut authority.lifecycle_receipts));
        }
        self.emit_lifecycle_receipts(conversation_id, receipts, result.as_ref().err())?;
        let retired = result.is_ok() && self.retire_owner_if_closed(conversation_id, &mut owner);
        drop(owner);
        if durably_empty || retired {
            self.evict_uncommitted(conversation_id, &cell)?;
        }
        result
    }

    /// Settles a failed operation's receipts and answers whether the
    /// conversation is durably empty. A batch that was never issued left
    /// nothing behind, so the conversation is empty exactly when its log is.
    /// A batch that was issued and whose landing nothing proved leaves the
    /// conversation possibly written, and its receipts owed to the next
    /// load, which decides.
    pub(super) fn fail_operation(
        &self,
        conversation_id: ConversationId,
        starting_log_sequence: u64,
        authority: &mut ConversationAuthority,
        log: &OperationLog,
    ) -> Result<bool, ParticipantSemanticError> {
        if !log.issuance().was_submitted() {
            return Ok(authority.next_log_sequence == 0);
        }
        self.hold_owed_receipts(
            conversation_id,
            starting_log_sequence,
            std::mem::take(&mut authority.lifecycle_receipts),
        )?;
        Ok(false)
    }

    /// Holds `receipts` for `conversation_id` until its next load shows
    /// whether the operation that started at `starting_log_sequence` landed.
    pub(super) fn hold_owed_receipts(
        &self,
        conversation_id: ConversationId,
        starting_log_sequence: u64,
        receipts: Vec<LifecycleReceipt>,
    ) -> Result<(), ParticipantSemanticError> {
        if receipts.is_empty() {
            return Ok(());
        }
        let mut owed = self.lock_owed_receipts()?;
        owed.insert(
            conversation_id,
            OwedReceipts {
                starting_log_sequence,
                receipts,
            },
        );
        drop(owed);
        Ok(())
    }

    /// The receipts owed to `conversation_id`, now that a load found its log
    /// head at `loaded_log_head`: all of them when the operation that owed
    /// them landed rows past where it started, none when it did not. Either
    /// way they are no longer held.
    pub(super) fn settle_owed_receipts(
        &self,
        conversation_id: ConversationId,
        loaded_log_head: u64,
    ) -> Result<Vec<LifecycleReceipt>, ParticipantSemanticError> {
        let mut owed = self.lock_owed_receipts()?;
        let held = owed.remove(&conversation_id);
        drop(owed);
        Ok(held
            .filter(|held| loaded_log_head > held.starting_log_sequence)
            .map(|held| held.receipts)
            .unwrap_or_default())
    }

    /// Emits the receipts owed to `conversation_id`, if its load at
    /// `loaded_log_head` shows that the operation that owed them landed: the
    /// one call for a load path that installs an owner outside an operation.
    pub(in super::super) fn emit_owed_receipts(
        &self,
        conversation_id: ConversationId,
        loaded_log_head: u64,
    ) -> Result<(), ParticipantSemanticError> {
        let owed = self.settle_owed_receipts(conversation_id, loaded_log_head)?;
        if owed.is_empty() {
            return Ok(());
        }
        self.emit_lifecycle_receipts(conversation_id, owed, None)
    }

    pub(super) fn lock_owed_receipts(
        &self,
    ) -> Result<MutexGuard<'_, HashMap<ConversationId, OwedReceipts>>, ParticipantSemanticError>
    {
        self.owed_receipts
            .lock()
            .map_err(|_| ParticipantSemanticError::Internal {
                message: "owed lifecycle receipt lock poisoned".to_owned(),
            })
    }

    pub(super) fn emit_lifecycle_receipts(
        &self,
        conversation_id: ConversationId,
        receipts: Vec<LifecycleReceipt>,
        operation_error: Option<&ParticipantSemanticError>,
    ) -> Result<(), ParticipantSemanticError> {
        if let Some(signals) = self.signals.get() {
            if let Err(error) = signals.lifecycle_batch(
                conversation_id,
                receipts
                    .into_iter()
                    .map(|receipt| (receipt.generation, receipt.transition)),
            ) {
                if let Some(operation_error) = operation_error {
                    tracing::error!(%operation_error, "operation and lifecycle batch both failed");
                } else {
                    tracing::error!(conversation_id, %error, "committed lifecycle batch failed");
                    return Ok(());
                }
                return Err(ParticipantSemanticError::WriteOutcomeUnknown {
                    message: format!("committed lifecycle batch failed: {error}"),
                });
            }
        }
        Ok(())
    }
}