liminal-server 0.21.1

//! When the services start over a durable store, every in-flight claim in the
//! durable delivery dedup namespace is looked up by its key before any publish
//! is judged: a claim left by a crash or by an unresolved publish is completed
//! when its row is in the stream, released when it is not, and held when
//! nothing can say yet; a claim that carries no evidence, left by a binary
//! before the evidence lane, is released there and then (DUR-022). The claims are read from the namespace's in-flight
//! index (`liminal` `durability/dedup/in_flight.rs`), never by walking the
//! namespace; the first start over a store written before the index existed
//! walks it once to build the index. A store created empty by this process
//! holds no claim, so it is not read at all.
//!
//! The resolve runs behind the listening door, on a thread of its own, the
//! way the participant warm does (12 October 2026: 1,098 ms of the door's
//! 1,172 ms service construction was this resolve finding no claim). Nothing
//! is judged before it: a keyed durable admission that comes first runs it
//! itself, and one that comes while it runs waits for it.
use std::path::Path;

use std::sync::Arc;

use super::unresolved::HeldKey;
use super::{
    DELIVERY_DEDUP_NAMESPACE, LiminalConnectionServices, database_exists, database_untouched,
    durable_data_dir,
};
use std::future::Future;
use std::time::Instant;

use crate::ServerError;
use liminal::durability::{DedupCache, DurableStore, InFlightClaim};

/// The name of the thread that lists the in-flight claims at start.
const CLAIMS_RESOLVE_THREAD: &str = "liminal-claims-resolve";

/// The name of the thread that releases the claims nothing can check.
const CLAIMS_RELEASE_THREAD: &str = "liminal-claims-release";
/// The thread that records the delivery namespace's empty index start at the
/// start that created its store (DUR-023).
const INDEX_START_RECORD_THREAD: &str = "liminal-index-start-record";

/// Why a start-up store future could not be driven to its answer.
#[derive(Debug, thiserror::Error)]
enum StartupThreadError {
    #[error("the start-up thread could not be spawned: {0}")]
    Spawn(std::io::Error),
    #[error("the start-up executor could not be built: {0}")]
    Executor(std::io::Error),
    #[error("the start-up thread panicked")]
    Panicked,
}

/// Drives one start-up store future on a thread of its own under an executor
/// of its own, and answers what it answered.
///
/// The synchronous bridge (`liminal::durability::bridge::block_on`) polls a
/// future with no waker and is sound only for a future that is ready on its
/// first poll, as its module says. The haematite store answers a listing from
/// a shard actor on another thread, so the future that waits for that answer
/// is woken, not re-polled: driven by the bridge, the boot thread parks on the
/// answer and the server never listens (the door, 10 October 2026, boots 1 to
/// 4 at `resolve_claims_at_start`). A current-thread tokio executor on a
/// dedicated thread drives the future to its answer, and the calling thread,
/// the boot thread, which owns no executor, joins it. `make` builds the future
/// on the dedicated thread, so the future never crosses a thread.
fn run_on_startup_thread<F, Make>(name: &str, make: Make) -> Result<F::Output, StartupThreadError>
where
    F: Future,
    F::Output: Send,
    Make: FnOnce() -> F + Send,
{
    std::thread::scope(|scope| {
        let worker = std::thread::Builder::new()
            .name(name.to_owned())
            .spawn_scoped(scope, move || {
                let executor = tokio::runtime::Builder::new_current_thread()
                    .enable_all()
                    .build()
                    .map_err(StartupThreadError::Executor)?;
                Ok(executor.block_on(make()))
            })
            .map_err(StartupThreadError::Spawn)?;
        worker.join().map_err(|_| StartupThreadError::Panicked)?
    })
}

/// Whether the durable store the services start over can already hold
/// in-flight claims.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) enum StoreHistory {
    /// The store was created empty for this process and dies with it: no
    /// earlier server wrote to it, so it holds no claim to look up.
    #[cfg(any(test, feature = "test-support"))]
    Fresh,
    /// The store is persistent and holds nothing yet: this start creates it,
    /// or an earlier start created it and nothing was ever written to it. It
    /// holds no claim, and outlives the process.
    Created,
    /// The store may hold claims an earlier server left: a persistent store
    /// that already exists, or one the caller handed in.
    Recorded,
}

impl StoreHistory {
    /// The history of the store that will be built for `persistence_path`,
    /// answered before it is opened: a persistent store holds nothing when its database
    /// does not exist yet or holds only what its creation wrote, and otherwise
    /// may hold what the servers before this one wrote.
    pub(super) fn of(path: &Path) -> Self {
        let data_dir = durable_data_dir(path);
        if database_exists(&data_dir) && !database_untouched(&data_dir) {
            Self::Recorded
        } else {
            Self::Created
        }
    }

    /// The durable delivery dedup cache over `store`: a store created by this
    /// process holds no claim, so its in-flight index starts empty without
    /// reading it.
    pub(super) fn delivery_dedup(self, store: Arc<dyn DurableStore>) -> DedupCache {
        if self == Self::Recorded {
            DedupCache::new(store, DELIVERY_DEDUP_NAMESPACE)
        } else {
            DedupCache::new_empty(store, DELIVERY_DEDUP_NAMESPACE)
        }
    }
}

/// The name of the thread that resolves the claims behind the listening door.
const CLAIMS_BEHIND_LISTENER_THREAD: &str = "liminal-claims-behind-listener";

impl LiminalConnectionServices {
    /// Resolves the in-flight claims once, unless that has been done: every
    /// keyed durable admission calls this before it is judged. The lock is
    /// held across the resolve, so an admission that comes while it runs
    /// waits for it; a resolve that fails stays pending, and the next
    /// admission tries again and is refused by name if it fails again.
    ///
    /// # Errors
    ///
    /// Returns the resolve's error when the claims cannot be listed.
    pub(crate) fn ensure_claims_resolved(&self) -> Result<(), ServerError> {
        let mut pending = self
            .claims_start
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        // The guard is held across the resolve on purpose: an admission that
        // comes while it runs waits here for its end.
        let resolved = pending.map_or(Ok(()), |history| self.resolve_claims_at_start(history));
        if resolved.is_ok() {
            *pending = None;
        }
        drop(pending);
        resolved
    }

    /// Starts the resolve behind the listening door, on a thread of its own.
    /// A failure is logged; the first keyed durable admission tries again.
    pub(super) fn resolve_claims_behind_listener(
        self: Arc<Self>,
    ) -> Option<std::thread::JoinHandle<()>> {
        let spawned = std::thread::Builder::new()
            .name(CLAIMS_BEHIND_LISTENER_THREAD.to_owned())
            .spawn(move || {
                if let Err(error) = self.ensure_claims_resolved() {
                    tracing::error!(
                        %error,
                        "in-flight claims could not be resolved behind the listening door; \
                         the first keyed durable admission tries again"
                    );
                }
            });
        spawned
            .inspect_err(|error| {
                tracing::error!(
                    %error,
                    "the claims resolve thread could not start; the first keyed durable admission resolves them"
                );
            })
            .ok()
    }

    /// Resolves the durable namespace's in-flight claims, once, at start.
    ///
    /// A store this process created (`Fresh` or `Created`) is not read: it
    /// holds nothing. A `Created` store, which outlives the process, has its
    /// empty index start recorded here, one compare-and-set on the start-up
    /// thread (DUR-023), so the next start over it reads the index instead of
    /// walking a namespace nothing was ever claimed in; a `Fresh` store dies
    /// with the process and records nothing.
    ///
    /// # Errors
    ///
    /// Returns [`ServerError::ListenerAccept`] naming the dedup namespace when
    /// its claims cannot be listed. A claim whose stream cannot be read is not
    /// an error: it stays held for the channel's next admission.
    pub(super) fn resolve_claims_at_start(&self, history: StoreHistory) -> Result<(), ServerError> {
        if history == StoreHistory::Created {
            self.record_empty_index_start();
            return Ok(());
        }
        if history != StoreHistory::Recorded {
            return Ok(());
        }
        let listing_failed = |message: String| ServerError::ListenerAccept {
            message: format!(
                "in-flight claims in dedup namespace '{}' could not be listed: {message}",
                self.dedup.namespace()
            ),
        };
        tracing::info!(
            namespace = self.dedup.namespace(),
            "in-flight claims resolve started"
        );
        let started = Instant::now();
        let claims = run_on_startup_thread(CLAIMS_RESOLVE_THREAD, || self.dedup.in_flight_claims())
            .map_err(|error| listing_failed(error.to_string()))?
            .map_err(|error| listing_failed(error.to_string()))?;
        let listed = claims.len();
        let (checkable, unchecked): (Vec<_>, Vec<_>) = claims
            .into_iter()
            .partition(|claim| claim.evidence.is_some());
        let released_unchecked = self.release_unchecked(unchecked)?;
        tracing::info!(
            namespace = self.dedup.namespace(),
            claims = listed,
            released_unchecked,
            elapsed_ms = u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX),
            "in-flight claims resolve finished"
        );
        for claim in checkable {
            let idempotency_key = claim.idempotency_key;
            let Some(evidence) = claim.evidence else {
                continue;
            };
            let channel = evidence.channel().to_owned();
            self.held_keys.hold(
                &idempotency_key,
                HeldKey {
                    channel: channel.clone(),
                    from: evidence.positions().to_vec(),
                },
            );
            // A channel not registered at start keeps the claim held; its first
            // keyed admission after registration looks the row up.
            let Ok(configured) = self.admit_channel(&channel) else {
                continue;
            };
            if let Err(error) =
                self.settle_held_with(&configured.handle, &channel, &idempotency_key)
            {
                tracing::warn!(
                    channel = channel.as_str(),
                    idempotency_key = idempotency_key.as_str(),
                    %error,
                    "in-flight claim could not be resolved at start; held for the next admission"
                );
            }
        }
        Ok(())
    }

    /// Records the delivery namespace's empty index start at the start that
    /// created its store (DUR-023), on a start-up thread of its own.
    ///
    /// Before this, the start of an index begun empty was recorded only by
    /// its first claim, so a store nothing was ever claimed in (every seeded
    /// fixture, every fresh install until its first keyed publish) reached
    /// its next start with no recorded start and was taken for a store
    /// written before the index existed and walked, shard by shard, to find
    /// nothing: 150 ms of the door's 420 ms resolve over an empty namespace
    /// (DUR-023's purpose). One compare-and-set here, on a store this boot is
    /// already creating, and that start reads the index and never walks.
    ///
    /// A refused record is logged and not fatal: the next start finds no
    /// recorded start and walks once, as before this change. Nothing else is
    /// read or written: the namespace holds no claim to list.
    fn record_empty_index_start(&self) {
        let started = Instant::now();
        let recorded = run_on_startup_thread(INDEX_START_RECORD_THREAD, || {
            self.dedup.record_empty_start()
        });
        let elapsed_ms = u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX);
        match recorded {
            Ok(Ok(())) => tracing::info!(
                namespace = self.dedup.namespace(),
                elapsed_ms,
                "dedup index start recorded at a created store"
            ),
            Ok(Err(error)) => tracing::error!(
                namespace = self.dedup.namespace(),
                elapsed_ms,
                %error,
                "dedup index start could not be recorded at a created store; the next start walks the namespace once"
            ),
            Err(error) => tracing::error!(
                namespace = self.dedup.namespace(),
                elapsed_ms,
                %error,
                "dedup index start could not be recorded at a created store; the next start walks the namespace once"
            ),
        }
    }

    /// Releases every listed claim that carries no evidence and answers how
    /// many were released (DUR-022).
    ///
    /// Such a claim was written by a binary before the evidence lane: nothing
    /// names the channel its row would be in, so no publish can ever settle it
    /// by looking the row up, and the one rule that ends it, its TTL in the
    /// dedup sweep, runs on a timer the server never has. Held, it refuses its
    /// producer's every retry by name, forever. Released, the producer's retry
    /// publishes; the cost is bounded by one row per key, if and only if the
    /// crashed publish had landed its row, and that row sits in a channel the
    /// producer reads back under its own key. Completed it is never: a receipt
    /// vouches for a row, and nothing can say the row exists.
    ///
    /// The releases run on their own thread under their own executor, as the
    /// listing does. A release the store refuses is logged with its key and
    /// not held: the next start lists the claim again and releases it then.
    ///
    /// # Errors
    ///
    /// Returns [`ServerError::ListenerAccept`] when the release thread cannot
    /// run at all; a single refused release is not an error.
    fn release_unchecked(&self, unchecked: Vec<InFlightClaim>) -> Result<usize, ServerError> {
        if unchecked.is_empty() {
            return Ok(0);
        }
        let now_millis = super::dedup_timestamp_millis();
        let dedup = &self.dedup;
        let outcomes = run_on_startup_thread(CLAIMS_RELEASE_THREAD, || async move {
            let mut outcomes = Vec::with_capacity(unchecked.len());
            for claim in unchecked {
                let released = dedup.release_claim(&claim.idempotency_key).await;
                outcomes.push((claim, released));
            }
            outcomes
        })
        .map_err(|error| ServerError::ListenerAccept {
            message: format!(
                "in-flight claims in dedup namespace '{}' could not be released: {error}",
                self.dedup.namespace()
            ),
        })?;
        let mut released = 0_usize;
        for (claim, outcome) in outcomes {
            match outcome {
                Ok(()) => {
                    released = released.saturating_add(1);
                    tracing::warn!(
                        idempotency_key = claim.idempotency_key.as_str(),
                        claimed_at_millis = claim.claimed_at_millis,
                        age_ms = now_millis.saturating_sub(claim.claimed_at_millis),
                        "in-flight claim released: no evidence"
                    );
                }
                Err(error) => tracing::error!(
                    idempotency_key = claim.idempotency_key.as_str(),
                    claimed_at_millis = claim.claimed_at_millis,
                    %error,
                    "in-flight claim with no evidence could not be released; the next start releases it"
                ),
            }
        }
        Ok(released)
    }
}

#[cfg(test)]
mod tests {
    use std::future::Future;
    use std::pin::Pin;
    use std::sync::{Arc, Mutex, PoisonError};
    use std::task::{Context, Poll, Waker};
    use std::time::Duration;

    use liminal::durability::bridge::{BridgeError, block_on};

    use super::run_on_startup_thread;

    /// A future answered from another thread, the way a shard actor answers a
    /// listing: pending until that thread stores the answer and wakes the
    /// waker it was polled with.
    struct AnsweredElsewhere {
        slot: Arc<Mutex<(Option<u32>, Option<Waker>)>>,
    }

    impl AnsweredElsewhere {
        fn new(answer: u32, after: Duration) -> Self {
            let slot: Arc<Mutex<(Option<u32>, Option<Waker>)>> = Arc::new(Mutex::new((None, None)));
            let answering = Arc::clone(&slot);
            std::thread::spawn(move || {
                std::thread::sleep(after);
                let mut slot = answering.lock().unwrap_or_else(PoisonError::into_inner);
                slot.0 = Some(answer);
                if let Some(waker) = slot.1.take() {
                    waker.wake();
                }
            });
            Self { slot }
        }
    }

    impl Future for AnsweredElsewhere {
        type Output = u32;

        fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<u32> {
            let mut slot = self.slot.lock().unwrap_or_else(PoisonError::into_inner);
            if let Some(answer) = slot.0 {
                return Poll::Ready(answer);
            }
            slot.1 = Some(cx.waker().clone());
            Poll::Pending
        }
    }

    /// A listing that panics instead of answering.
    async fn panicking_listing() -> u32 {
        std::panic::resume_unwind(Box::new("the listing panicked"))
    }

    const ANSWER: u32 = 7;
    const ANSWER_AFTER: Duration = Duration::from_millis(50);

    #[test]
    fn a_future_answered_from_another_thread_is_driven_to_its_answer() {
        let answered = run_on_startup_thread("claims-resolve-test", || {
            AnsweredElsewhere::new(ANSWER, ANSWER_AFTER)
        });
        assert!(
            matches!(answered, Ok(ANSWER)),
            "the dedicated executor must wait for the answer: {answered:?}"
        );
    }

    #[test]
    fn the_bridge_gives_up_on_the_same_future() {
        // The bridge is the thing the start-up resolve no longer uses: with no
        // waker and its bounded polls it reports the answer as never coming.
        let given_up = block_on(AnsweredElsewhere::new(ANSWER, ANSWER_AFTER));
        assert!(
            matches!(given_up, Err(BridgeError::DidNotComplete { .. })),
            "{given_up:?}"
        );
    }

    #[test]
    fn a_panic_on_the_start_up_thread_is_an_error_not_a_hang() {
        let outcome = run_on_startup_thread("claims-resolve-panic-test", panicking_listing);
        assert!(
            matches!(outcome, Err(super::StartupThreadError::Panicked)),
            "{outcome:?}"
        );
    }
}