liminal-server 0.25.4

//! The incarnation stream's start: its replay, the recovery of unmatched
//! Opens and the restart repair, guarded so a stop ends it at its next write.
//!
//! SRV-042 R2: a stop that begins while the incarnation start runs ends the
//! start at its next write, so the stop waits for at most the step in hand.
//!
//! The incarnation stream is written to survive a crash at any write: the
//! replay only reads, and each write (an epoch's append, its flush, a
//! roll-forward's anchor swap, a recovery's Complete) is a point a crash may
//! stop at and the next start resume from. The rotation's own anchor swap is
//! not the start's: it is settled after the start publishes (SRV-048). The start therefore reads its store through
//! [`DuringStart`], which refuses to begin a write once the server is
//! stopping; a write already begun completes. Once the start has finished it
//! refuses nothing, so a graceful stop's own writes (the fates its closing
//! connections record) are never refused.

use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};

use liminal::durability::bridge::block_on;
use liminal::durability::{
    DurabilityError, DurableStore, RoutedBatchEntry, RoutedBatchItem, ScanCursor, ScanPage,
    StoreCountsHandle, StoredEntry,
};
use liminal_protocol::outcome::ConnectionIncarnationExhausted;

use super::ConnectionIncarnationAuthority;
use crate::ServerError;
use crate::server::participant::ParticipantSemanticHandler;
use crate::server::participant::incarnation_stream::{
    ConnectionFateRecovery, IncarnationStartup, IncarnationStream, StartedIncarnationStream,
};

/// `ServerError::ParticipantIncarnation::phase` for a start the server's stop
/// ended (SRV-042 R2).
pub(in crate::server::connection) const ABANDONED_PHASE: &str =
    "incarnation start abandoned at shutdown";

fn abandoned() -> ServerError {
    ServerError::ParticipantIncarnation {
        phase: ABANDONED_PHASE,
        message: "the server stopped before the incarnation start finished".to_owned(),
    }
}

impl ConnectionIncarnationAuthority {
    /// [`Self::start_stream_unguarded`] over a store that refuses to begin a
    /// write once `stopping` is set, until the start returns; a start ended
    /// that way, or not begun because the server was already stopping,
    /// answers [`ABANDONED_PHASE`].
    pub(super) fn start_stream(
        store: Arc<dyn DurableStore>,
        maximum_references: usize,
        handler: &dyn ParticipantSemanticHandler,
        stopping: &Arc<AtomicBool>,
    ) -> Result<StartedIncarnationStream, ServerError> {
        if stopping.load(Ordering::SeqCst) {
            return Err(abandoned());
        }
        let during = Arc::new(DuringStart {
            inner: store,
            stopping: Arc::clone(stopping),
            finished: AtomicBool::new(false),
        });
        let guarded: Arc<dyn DurableStore> = during.clone();
        let started = Self::start_stream_unguarded(guarded, maximum_references, handler);
        during.finished.store(true, Ordering::SeqCst);
        started.map_err(|error| {
            if stopping.load(Ordering::SeqCst) {
                abandoned()
            } else {
                error
            }
        })
    }
}

/// The start's store: refuses to begin a write while the server is stopping
/// and the start has not finished.
#[derive(Debug)]
struct DuringStart {
    inner: Arc<dyn DurableStore>,
    stopping: Arc<AtomicBool>,
    finished: AtomicBool,
}

impl DuringStart {
    fn may_write(&self) -> Result<(), DurabilityError> {
        if self.stopping.load(Ordering::SeqCst) && !self.finished.load(Ordering::SeqCst) {
            return Err(DurabilityError::ConfigError(
                "the server is stopping; the incarnation start begins no further write".to_owned(),
            ));
        }
        Ok(())
    }
}

#[async_trait::async_trait]
impl DurableStore for DuringStart {
    fn reclaim_spine(&self) -> Option<Arc<haematite::EventStore>> {
        self.inner.reclaim_spine()
    }

    fn max_request_bytes(&self) -> Option<usize> {
        self.inner.max_request_bytes()
    }

    async fn apply_routed_batch(
        &self,
        route_key: &str,
        items: Vec<RoutedBatchItem>,
    ) -> Result<(), DurabilityError> {
        self.may_write()?;
        self.inner.apply_routed_batch(route_key, items).await
    }

    async fn apply_routed_entries(
        &self,
        route_key: &str,
        entries: Vec<RoutedBatchEntry>,
    ) -> Result<(), DurabilityError> {
        self.may_write()?;
        self.inner.apply_routed_entries(route_key, entries).await
    }

    async fn get_routed(
        &self,
        route_key: &str,
        key: &[u8],
    ) -> Result<Option<Vec<u8>>, DurabilityError> {
        self.inner.get_routed(route_key, key).await
    }

    async fn append(
        &self,
        stream_key: &str,
        payload: Vec<u8>,
        expected_seq: u64,
    ) -> Result<u64, DurabilityError> {
        self.may_write()?;
        self.inner.append(stream_key, payload, expected_seq).await
    }

    async fn read_from(
        &self,
        stream_key: &str,
        offset: u64,
        limit: usize,
    ) -> Result<Vec<StoredEntry>, DurabilityError> {
        self.inner.read_from(stream_key, offset, limit).await
    }

    async fn append_all(
        &self,
        stream_key: &str,
        payloads: Vec<Vec<u8>>,
        expected_seq: u64,
    ) -> Result<u64, DurabilityError> {
        self.may_write()?;
        self.inner
            .append_all(stream_key, payloads, expected_seq)
            .await
    }

    async fn read_at(
        &self,
        stream_key: &str,
        sequence: u64,
    ) -> Result<Option<StoredEntry>, DurabilityError> {
        self.inner.read_at(stream_key, sequence).await
    }

    async fn cas(&self, key: &str, old_value: u64, new_value: u64) -> Result<(), DurabilityError> {
        self.may_write()?;
        self.inner.cas(key, old_value, new_value).await
    }

    async fn read_stream_next_seq(&self, stream_key: &str) -> Result<Option<u64>, DurabilityError> {
        self.inner.read_stream_next_seq(stream_key).await
    }

    async fn read_value(&self, key: &str) -> Result<Option<u64>, DurabilityError> {
        self.inner.read_value(key).await
    }

    async fn scan_page(
        &self,
        prefix: &str,
        cursor: Option<ScanCursor>,
        limit: usize,
    ) -> Result<ScanPage, DurabilityError> {
        self.inner.scan_page(prefix, cursor, limit).await
    }

    async fn flush(&self) -> Result<(), DurabilityError> {
        self.may_write()?;
        self.inner.flush().await
    }

    async fn truncate_before(
        &self,
        stream_key: &str,
        before_seq: u64,
    ) -> Result<u64, DurabilityError> {
        self.may_write()?;
        self.inner.truncate_before(stream_key, before_seq).await
    }

    async fn scan_routed(
        &self,
        route_key: &str,
        from: &[u8],
        to: &[u8],
    ) -> Result<Vec<(Vec<u8>, Vec<u8>)>, DurabilityError> {
        self.inner.scan_routed(route_key, from, to).await
    }

    async fn remove_routed_range(
        &self,
        route_key: &str,
        from: &[u8],
        to: &[u8],
    ) -> Result<u64, DurabilityError> {
        self.may_write()?;
        self.inner.remove_routed_range(route_key, from, to).await
    }

    fn store_counts(&self) -> Option<StoreCountsHandle> {
        self.inner.store_counts()
    }
}

impl ConnectionIncarnationAuthority {
    /// Replays the stream, recovers any unmatched Open, flushes the server
    /// incarnation transition and repairs an unclean predecessor; run by
    /// [`Self::start_stream`] (SRV-042 R2).
    fn start_stream_unguarded(
        store: Arc<dyn DurableStore>,
        maximum_references: usize,
        handler: &dyn ParticipantSemanticHandler,
    ) -> Result<StartedIncarnationStream, ServerError> {
        let starting = std::time::Instant::now();
        let startup = block_on(IncarnationStream::new(store, maximum_references).startup());
        crate::server::boot_steps::finished("incarnation_startup", starting);
        let startup = startup
            .map_err(|error| ServerError::ParticipantIncarnation {
                phase: "server startup bridge",
                message: error.to_string(),
            })?
            .map_err(|error| ServerError::ParticipantIncarnation {
                phase: "server startup persistence",
                message: error.to_string(),
            })?;
        match startup {
            IncarnationStartup::Started(stream) => Self::finish_startup(stream, handler),
            IncarnationStartup::RecoveryRequired(mut recovery) => {
                let recovering = std::time::Instant::now();
                Self::complete_unmatched_opens(&mut recovery, handler)?;
                let resumed = block_on(recovery.finish_startup());
                crate::server::boot_steps::finished("incarnation_recovery", recovering);
                let resumed = resumed
                    .map_err(|error| ServerError::ParticipantIncarnation {
                        phase: "post-recovery startup bridge",
                        message: error.to_string(),
                    })?
                    .map_err(|error| ServerError::ParticipantIncarnation {
                        phase: "post-recovery startup persistence",
                        message: error.to_string(),
                    })?;
                match resumed {
                    IncarnationStartup::Started(stream) => Self::finish_startup(stream, handler),
                    IncarnationStartup::RecoveryRequired(_) => {
                        Err(ServerError::ParticipantIncarnation {
                            phase: "post-recovery startup",
                            message: "completed recovery returned another unmatched Open set"
                                .to_owned(),
                        })
                    }
                    IncarnationStartup::Exhausted(
                        ConnectionIncarnationExhausted::ServerIncarnation,
                    ) => Err(ServerError::ServerIncarnationExhausted),
                    IncarnationStartup::Exhausted(
                        ConnectionIncarnationExhausted::ConnectionOrdinal {
                            attempted_server_incarnation,
                        },
                    ) => Err(ServerError::ParticipantIncarnation {
                        phase: "post-recovery startup protocol",
                        message: format!(
                            "unexpected connection-ordinal exhaustion for server incarnation {attempted_server_incarnation}"
                        ),
                    }),
                }
            }
            IncarnationStartup::Exhausted(ConnectionIncarnationExhausted::ServerIncarnation) => {
                Err(ServerError::ServerIncarnationExhausted)
            }
            IncarnationStartup::Exhausted(ConnectionIncarnationExhausted::ConnectionOrdinal {
                attempted_server_incarnation,
            }) => Err(ServerError::ParticipantIncarnation {
                phase: "server startup protocol",
                message: format!(
                    "unexpected connection-ordinal exhaustion for server incarnation \
                     {attempted_server_incarnation} during startup"
                ),
            }),
        }
    }

    /// Applies and completes every unmatched Open the previous process left,
    /// with a progress line per Open.
    fn complete_unmatched_opens(
        recovery: &mut ConnectionFateRecovery,
        handler: &dyn ParticipantSemanticHandler,
    ) -> Result<(), ServerError> {
        let intents = recovery.intents();
        if intents.is_empty() {
            return Err(ServerError::ParticipantIncarnation {
                phase: "connection-fate recovery",
                message: "recovery owner returned no unmatched Open".to_owned(),
            });
        }
        let total = intents.len();
        let started = std::time::Instant::now();
        tracing::info!(
            step = "connection-fate recovery",
            unmatched_opens = total,
            "connection-fate recovery started"
        );
        for (index, intent) in intents.into_iter().enumerate() {
            let work_item = intent.work_item();
            tracing::info!(
                step = "connection-fate recovery",
                open_sequence = intent.open_sequence,
                conversations = work_item.tracked_conversations.len(),
                done = index,
                total,
                elapsed_ms = u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX),
                "connection-fate recovery applying an unmatched Open"
            );
            handler.handle_connection_fate(work_item).map_err(|error| {
                ServerError::ParticipantIncarnation {
                    phase: "connection-fate handler recovery",
                    message: format!(
                        "Open {} failed before Complete: {error}",
                        intent.open_sequence
                    ),
                }
            })?;
            block_on(recovery.complete(intent.open_sequence))
                .map_err(|error| ServerError::ParticipantIncarnation {
                    phase: "connection-fate Complete bridge",
                    message: error.to_string(),
                })?
                .map_err(|error| ServerError::ParticipantIncarnation {
                    phase: "connection-fate Complete persistence",
                    message: error.to_string(),
                })?;
        }
        tracing::info!(
            step = "connection-fate recovery",
            done = total,
            total,
            elapsed_ms = u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX),
            "connection-fate recovery finished"
        );
        Ok(())
    }

    fn finish_startup(
        stream: StartedIncarnationStream,
        handler: &dyn ParticipantSemanticHandler,
    ) -> Result<StartedIncarnationStream, ServerError> {
        crate::server::boot_steps::timed("restart_repair", || {
            handler.repair_unclean_server_restart(stream.server_incarnation())
        })
        .map_err(|error| ServerError::ParticipantIncarnation {
            phase: "server restart repair",
            message: error.to_string(),
        })?;
        Ok(stream)
    }
}