liminal-server 0.24.1

//! DUR-027 R5: the retirement of one closed, unarmed conversation, as one
//! ordered transition.
//!
//! It runs holding the conversation's owner cell, after the commit whose
//! apply left the conversation closed, and after the commit whose progress
//! advance fired the last arm on a closed conversation (an arm leaves the
//! observer aggregate only by firing, and it fires inside that commit). It is
//! attempted on those events alone: nothing here waits, sleeps or polls.
//!
//! The steps, in order:
//!
//! 1. the registry retire row (the commit point: the id is sweeping);
//! 2. the generation key moved by compare-and-swap to [`RETIRED_GENERATION`],
//!    so opening the log refuses from here on (the tombstone);
//! 3. the observer retire row, and `decide_retire` committed on the live
//!    aggregate;
//! 4. the conversation's streams cut whole, each by one commit: the base
//!    stream, every generation stream, the outbox stream, and the checkpoint
//!    windows it owns on its route (DUR-028 R5), never the route's whole key
//!    space;
//! 5. the registry swept row, then the conversation's capacity contribution
//!    and its owner cell go.
//!
//! Every step is skipped when the store shows it done (the registry's sets,
//! the generation key, the aggregate's progress row, a stream with nothing
//! below its head), never by a flag in memory, so a retirement stopped after
//! step 1 is finished by the next boot (R6) through [`finish_retirement`].
//!
//! [`finish_retirement`]: ProductionParticipantHandler::finish_retirement

use std::sync::{Arc, MutexGuard};

use liminal::durability::DurableStore;
use liminal::durability::bridge::block_on;
use liminal_protocol::lifecycle::{ObserverProgressRetireDecision, ObserverProgressRetireError};
use liminal_protocol::wire::ConversationId;

use crate::server::participant::ParticipantSemanticError;

use super::capacity::ConversationContribution;
use super::handler::{ObserverOwner, ProductionParticipantHandler, bridge_error, log_error};
use super::log::{GENERATION_STREAM_PREFIX, OperationLog, RETIRED_GENERATION, STREAM_PREFIX};
use super::observer::{ObserverLog, ObserverRow, snapshot_before};
use super::outbox_log::OUTBOX_STREAM_PREFIX;
use super::registry::Membership;
use super::restore_checkpoint_locator::checkpoint_conversation_route;
use super::route_keys::conversation_windows;
use super::routed_stream::read_head;
use super::state::ConversationAuthority;

/// The step a retirement stopped at.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) enum RetireStep {
    /// The observer aggregate could not be read before step 1.
    Observer,
    /// Step 1, the registry retire row.
    Registry,
    /// Step 2, the generation tombstone.
    Generation,
    /// Step 3, the observer retire row.
    ObserverRow,
    /// Step 4, the stream cuts.
    Streams,
    /// Step 5, the registry swept row.
    Swept,
}

/// What one attempt came to.
#[derive(Debug, PartialEq, Eq)]
pub(super) enum Retirement {
    /// Every step is done, now or before.
    Retired,
    /// An observer arm is installed on the conversation: nothing was written.
    Armed,
    /// A step failed with the store's error; after step 1 the id stays
    /// sweeping for the next boot.
    Stopped {
        /// The step that failed.
        step: RetireStep,
        /// The failure, in the store's words.
        error: String,
    },
}

const fn stopped(step: RetireStep, error: String) -> Retirement {
    Retirement::Stopped { step, error }
}

impl ProductionParticipantHandler {
    /// After a commit or a restore left `authority` as it is, retires it when
    /// it is closed; `true` when the owner and its cell are to go. The caller
    /// holds the owner cell.
    pub(super) fn retire_if_closed(
        &self,
        conversation_id: ConversationId,
        authority: &ConversationAuthority,
    ) -> bool {
        authority.is_closed() && self.retire_closed(conversation_id) == Retirement::Retired
    }

    /// DUR-027 R5: a commit that left the conversation closed, or fired the
    /// last arm on a closed one, retires it under its owner, which goes.
    pub(super) fn retire_owner_if_closed(
        &self,
        conversation_id: ConversationId,
        owner: &mut Option<ConversationAuthority>,
    ) -> bool {
        let retired = owner
            .as_ref()
            .is_some_and(|authority| self.retire_if_closed(conversation_id, authority));
        if retired {
            *owner = None;
        }
        retired
    }

    /// Retires `conversation_id`, closed, whose owner cell the caller holds;
    /// a stop is logged by its step and the store's error, and never fails
    /// the commit that came before it.
    fn retire_closed(&self, conversation_id: ConversationId) -> Retirement {
        let outcome = self.retire_conversation(conversation_id);
        if let Retirement::Stopped { step, error } = &outcome {
            tracing::error!(
                conversation_id,
                ?step,
                %error,
                "conversation retirement stopped; it is resumed by the next boot"
            );
        }
        outcome
    }

    /// One attempt on `conversation_id` as a closing commit makes it.
    #[cfg(test)]
    pub(super) fn retire_for_test(&self, conversation_id: ConversationId) -> Retirement {
        self.retire_closed(conversation_id)
    }

    fn retire_conversation(&self, conversation_id: ConversationId) -> Retirement {
        let mut observer = match self.loaded_observer() {
            Ok(observer) => observer,
            Err(error) => return stopped(RetireStep::Observer, error.to_string()),
        };
        // The arm check and step 3 run under one observer lock, so no arm is
        // installed between them.
        let armed = observer.as_ref().is_some_and(|owner| {
            owner
                .aggregate
                .armed_rows()
                .iter()
                .any(|(armed, _)| *armed == conversation_id)
        });
        if armed {
            return Retirement::Armed;
        }
        if self.registry.membership(conversation_id) == Membership::Live {
            if let Err(error) = self.registry.retire(conversation_id) {
                return stopped(RetireStep::Registry, error.to_string());
            }
        }
        let outcome = self.finish_retirement(conversation_id, &mut observer);
        drop(observer);
        outcome
    }

    /// DUR-027 R6: the registry's live ids for boot, after every retirement a
    /// stop interrupted is finished: steps 2 to 5 for each sweeping id, each
    /// skipped when the store shows it done. That set is bounded by the
    /// retirements a crash interrupted, never by history. A step that fails
    /// leaves the id sweeping for the next boot and is logged by name; a
    /// residue is never a boot fault, so the server still binds.
    ///
    /// # Errors
    /// The registry read, which stays fatal: without it there is no
    /// conversation to attribute a failure to.
    pub(super) fn restore_registry_finishing_sweeps(
        &self,
    ) -> Result<Vec<ConversationId>, ParticipantSemanticError> {
        let sets = self
            .registry
            .restore_sets()
            .map_err(|error| log_error(&error))?;
        for &conversation_id in &sets.sweeping {
            let mut observer = match self.loaded_observer() {
                Ok(observer) => observer,
                Err(error) => {
                    tracing::error!(conversation_id, %error, "interrupted retirement not finished at boot: the observer could not be read; it stays sweeping");
                    continue;
                }
            };
            match self.finish_retirement(conversation_id, &mut observer) {
                Retirement::Retired => {
                    tracing::info!(conversation_id, "interrupted retirement finished at boot");
                }
                Retirement::Armed | Retirement::Stopped { .. } => {
                    tracing::error!(
                        conversation_id,
                        "interrupted retirement not finished at boot; it stays sweeping for the next"
                    );
                }
            }
        }
        Ok(sets.live.into_iter().collect())
    }

    /// Steps 2 to 5 for an id whose retirement committed, each skipped when
    /// the store shows it done; boot runs it for every sweeping id (R6).
    pub(super) fn finish_retirement(
        &self,
        conversation_id: ConversationId,
        observer: &mut Option<ObserverOwner>,
    ) -> Retirement {
        if let Err(error) = self.tombstone(conversation_id) {
            return stopped(RetireStep::Generation, error);
        }
        if let Err(error) = self.retire_observer(conversation_id, observer) {
            return stopped(RetireStep::ObserverRow, error);
        }
        if let Err(error) = cut_streams(self.store.as_ref(), conversation_id) {
            return stopped(RetireStep::Streams, error);
        }
        if self.registry.membership(conversation_id) == Membership::Sweeping {
            if let Err(error) = self.registry.swept(conversation_id) {
                return stopped(RetireStep::Swept, error.to_string());
            }
        }
        let released = self.capacity.fold_conversation(
            conversation_id,
            ConversationContribution {
                identity: 0,
                entries: Vec::new(),
            },
        );
        if let Err(error) = released {
            // The rows are gone; an identity share left counted can only
            // refuse early, never admit past a cap.
            tracing::error!(conversation_id, %error, "retired conversation's capacity share kept");
        }
        Retirement::Retired
    }

    /// The observer owner, restored from its stream when absent, under its
    /// lock.
    pub(super) fn loaded_observer(
        &self,
    ) -> Result<MutexGuard<'_, Option<ObserverOwner>>, ParticipantSemanticError> {
        let observer_log = ObserverLog::new(Arc::clone(&self.store));
        let mut owner = self
            .observer
            .lock()
            .map_err(|_| ParticipantSemanticError::Internal {
                message: "observer recovery aggregate lock is poisoned".to_owned(),
            })?;
        if owner.is_none() {
            let restored = block_on(observer_log.restore())
                .map_err(|error| bridge_error(&error))?
                .map_err(|error| log_error(&error))?;
            *owner = Some(ObserverOwner {
                aggregate: restored.aggregate,
                head: restored.next_sequence,
                arm_targets: std::collections::BTreeMap::new(),
            });
        }
        self.refresh_uncertain_observer(&observer_log, owner.as_mut())?;
        Ok(owner)
    }

    /// Step 2: the generation key moves to the tombstone unless it is there.
    fn tombstone(&self, conversation_id: ConversationId) -> Result<(), String> {
        let log = OperationLog::new(Arc::clone(&self.store), conversation_id);
        let generation = block_on(log.generation())
            .map_err(|error| error.to_string())?
            .map_err(|error| error.to_string())?;
        if generation == RETIRED_GENERATION {
            return Ok(());
        }
        block_on(log.switch_generation(generation, RETIRED_GENERATION))
            .map_err(|error| error.to_string())?
            .map_err(|error| error.to_string())
    }

    /// Step 3: the retire row, then the aggregate's own commit. An aggregate
    /// with no progress row for the id has retired it already.
    fn retire_observer(
        &self,
        conversation_id: ConversationId,
        observer: &mut Option<ObserverOwner>,
    ) -> Result<(), String> {
        let ObserverOwner {
            aggregate,
            head,
            mut arm_targets,
        } = observer
            .take()
            .ok_or("the observer recovery aggregate is absent")?;
        let snapshot = snapshot_before(head, &aggregate);
        let transaction = match aggregate.decide_retire(conversation_id) {
            ObserverProgressRetireDecision::Commit(transaction) => transaction,
            ObserverProgressRetireDecision::Refuse { aggregate, error } => {
                *observer = Some(ObserverOwner {
                    aggregate,
                    head,
                    arm_targets,
                });
                return match error {
                    ObserverProgressRetireError::ConversationUnknown { .. } => Ok(()),
                    ObserverProgressRetireError::Armed { epoch, .. } => {
                        Err(format!("an observer arm at epoch {epoch:?} is installed"))
                    }
                };
            }
        };
        let appended = block_on(ObserverLog::new(Arc::clone(&self.store)).append(
            &ObserverRow::Retire { conversation_id },
            head,
            snapshot,
        ));
        match appended {
            Ok(Ok(next_head)) => {
                arm_targets.remove(&conversation_id);
                *observer = Some(ObserverOwner {
                    aggregate: transaction.commit(),
                    head: next_head,
                    arm_targets,
                });
                Ok(())
            }
            // Nothing installed; the aggregate is restored from the store on
            // its next use, which shows whether the row landed.
            Ok(Err(error)) => Err(error.to_string()),
            Err(error) => Err(error.to_string()),
        }
    }
}

/// Step 4: every stream of the conversation, flat and routed, and every
/// window of its checkpoint route that holds its keys and only its keys
/// (DUR-028 R5), each cut by one commit. A route's whole key space is never
/// cut: the route names a shard, which other conversations share.
fn cut_streams(store: &dyn DurableStore, conversation_id: ConversationId) -> Result<(), String> {
    let route = checkpoint_conversation_route(conversation_id);
    let mut streams = vec![
        format!("{STREAM_PREFIX}{conversation_id}"),
        format!("{OUTBOX_STREAM_PREFIX}{conversation_id}"),
    ];
    // Generations count up from 1 (log_migration), so the first with no head
    // in either form ends them; a target a stopped migration wrote past the
    // key is cut too.
    let mut generation: u64 = 1;
    while let Some(stream) = generation_stream(store, &route, conversation_id, generation)? {
        streams.push(stream);
        generation = generation
            .checked_add(1)
            .ok_or("the generation streams reach the end of their range")?;
    }
    for stream in &streams[..2] {
        cut_flat(store, stream)?;
    }
    let windows =
        conversation_windows(conversation_id, &streams).map_err(|error| error.to_string())?;
    for (from, to) in windows {
        block_on(store.remove_routed_range(&route, &from, &to))
            .map_err(|error| error.to_string())?
            .map_err(|error| error.to_string())?;
    }
    Ok(())
}

/// Generation `generation`'s stream, its flat rows cut, when it has a head
/// in either form; `None` when it has neither.
fn generation_stream(
    store: &dyn DurableStore,
    route: &str,
    conversation_id: ConversationId,
    generation: u64,
) -> Result<Option<String>, String> {
    let stream = format!("{GENERATION_STREAM_PREFIX}{generation}:{conversation_id}");
    let routed = block_on(read_head(store, route, &stream))
        .map_err(|error| error.to_string())?
        .map_err(|error| error.to_string())?
        .is_some();
    let flat = cut_flat(store, &stream)?;
    Ok((flat || routed).then_some(stream))
}

/// Cuts every row of a flat stream below its head in one commit; `false`
/// when the stream has no head at all.
fn cut_flat(store: &dyn DurableStore, stream_key: &str) -> Result<bool, String> {
    let head = block_on(store.read_stream_next_seq(stream_key))
        .map_err(|error| error.to_string())?
        .map_err(|error| error.to_string())?;
    let Some(head) = head else {
        return Ok(false);
    };
    if head > 0 {
        block_on(store.truncate_before(stream_key, head))
            .map_err(|error| error.to_string())?
            .map_err(|error| error.to_string())?;
    }
    Ok(true)
}