liminal-server 0.20.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. 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.
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};

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

/// 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)
        }
    }
}

impl LiminalConnectionServices {
    /// 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, and its index is recorded by the first keyed publish
    /// instead of on the boot path, before the listener binds.
    ///
    /// # 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::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()))?;
        tracing::info!(
            namespace = self.dedup.namespace(),
            claims = claims.len(),
            elapsed_ms = u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX),
            "in-flight claims resolve finished"
        );
        for (idempotency_key, evidence) in claims {
            let Some(evidence) = evidence else {
                tracing::error!(
                    idempotency_key = idempotency_key.as_str(),
                    "in-flight claim cannot be checked"
                );
                self.held_keys.hold(&idempotency_key, HeldKey::Uncheckable);
                continue;
            };
            let channel = evidence.channel().to_owned();
            self.held_keys.hold(
                &idempotency_key,
                HeldKey::Checkable {
                    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(())
    }
}

#[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:?}"
        );
    }
}