aion-server 0.27.1

//! Whether the caller that started a delivery still wants it, asked again while
//! the delivery waits.
//!
//! # Why this is the CALLER's state and not the transport's
//!
//! A liminal push blocks for the worker's correlated reply, and re-polls a
//! predicate throughout that wait so a delivery nobody is waiting for any more
//! can be abandoned instead of held open. Two of the questions that predicate
//! asks belong to the transport — the deployment is draining, the worker is
//! still registered — but the third does not: the outbox arm also asks whether
//! **the pass that started this delivery still owns the row's claim**.
//!
//! That question cannot be answered from inside the transport, and the reason is
//! one fact about [`DeliveryGate`](super::outbox_dispatcher::DeliveryGate): its
//! `holds` is plain set membership, so **a key that was never begun reads
//! identically to a key that was released**. A delivery dispatched from the
//! worker dispatcher never takes a hold, so a transport that derived the row's
//! dispatch key for itself would find `holds == false` at the delivery's very
//! first poll and abandon before the worker could possibly reply — reported as
//! `delivery wait abandoned`, which reads to an operator as a **worker** fault.
//! Dropping the term instead would silently remove the outbox arm's live claim
//! guard. Neither is acceptable, so the predicate travels as an argument.
//!
//! # Why a named trait and not a closure
//!
//! Each implementation below names **one caller** and states that caller's whole
//! abandonment condition in one place. A bare `impl Fn() -> bool` would let any
//! call site assemble a predicate inline, which is precisely how a dispatcher
//! pass could silently borrow the outbox's claim check, or how "no abandonment
//! condition" could be spelled `|| true` and never be found again. This is the
//! same law [`Undeliverable`](super::task_delivery::Undeliverable) follows: the
//! decision has a name, and one definition per caller.

use std::sync::Arc;

use super::outbox_dispatcher::DeliveryGate;
use super::registry::{ConnectedWorkerRegistry, WorkerId};

/// Asked repeatedly while a delivery waits: does the caller still want it?
///
/// Returning `false` abandons the wait. A late reply that arrives afterwards is
/// discarded, so an implementation must only answer `false` when the caller
/// genuinely no longer owns the delivery — not merely because it is slow.
pub trait DeliveryIntent: Send + Sync {
    /// `true` while the caller still wants this delivery to complete.
    fn still_wanted(&self) -> bool;
}

/// The outbox pass's intent: it wants the delivery while the deployment is not
/// draining, it still holds the row's claim, **and** the chosen worker is still
/// registered.
///
/// The claim is the half no transport can ask about — see the module docs.
///
/// # Why the registration term is here and not left to the transport
///
/// An intent answers the caller's whole question. If a transport also applied
/// terms of its own, `still_wanted()` would be a partial answer whose real
/// meaning depended on which transport asked it, and the two halves could drift
/// apart with nothing to catch it. So every term the outbox arm's original
/// closure asked lives here — including the registration re-check, which that
/// closure did ask (`liminal_transport.rs:1209-1220`) and which would otherwise
/// be silently lost in the move.
pub struct OutboxClaim {
    gate: DeliveryGate,
    dispatch_key: String,
    registry: ConnectedWorkerRegistry,
    worker: WorkerId,
}

impl std::fmt::Debug for OutboxClaim {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("OutboxClaim")
            .field("dispatch_key", &self.dispatch_key)
            .field("worker", &self.worker)
            .finish_non_exhaustive()
    }
}

impl OutboxClaim {
    /// Bind this intent to the gate, the dispatch key of the row being
    /// delivered, and the worker the pass selected.
    #[must_use]
    pub fn new(
        gate: DeliveryGate,
        dispatch_key: String,
        registry: ConnectedWorkerRegistry,
        worker: WorkerId,
    ) -> Self {
        Self {
            gate,
            dispatch_key,
            registry,
            worker,
        }
    }
}

impl DeliveryIntent for OutboxClaim {
    fn still_wanted(&self) -> bool {
        if self.gate.is_draining() || !self.gate.holds(&self.dispatch_key) {
            return false;
        }
        worker_still_registered(&self.registry, self.worker)
    }
}

/// The registration re-check both real intents apply, with one definition so the
/// two cannot answer it differently — including on the error path, where an
/// unreadable registry must not license a continued wait.
fn worker_still_registered(registry: &ConnectedWorkerRegistry, worker: WorkerId) -> bool {
    match registry.worker_by_id(worker) {
        Ok(Some(_)) => true,
        Ok(None) => false,
        Err(error) => {
            // A registry that cannot be read cannot license a continued wait:
            // the honest answer is to abandon and let the caller's retry path
            // run, loudly, rather than hold a delivery open on an unreadable
            // premise. This mirrors the original closure's behaviour, warning
            // line included.
            tracing::warn!(
                %error,
                worker = ?worker,
                "delivery wait could not verify worker registration; abandoning"
            );
            false
        }
    }
}

/// The worker dispatcher's intent: it wants the delivery while the deployment is
/// not draining **and** the chosen worker is still registered.
///
/// 🔴 It deliberately does **not** consult the delivery gate's held keys. This
/// pass never took a hold, and a never-held key is indistinguishable from a
/// released one, so consulting it would abandon every dispatcher-originated
/// delivery at its first poll.
pub struct DispatcherPass {
    gate: DeliveryGate,
    registry: ConnectedWorkerRegistry,
    worker: WorkerId,
}

impl std::fmt::Debug for DispatcherPass {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("DispatcherPass")
            .field("worker", &self.worker)
            .finish_non_exhaustive()
    }
}

impl DispatcherPass {
    /// Bind this intent to the deployment's drain gate and the worker this pass
    /// selected.
    #[must_use]
    pub fn new(gate: DeliveryGate, registry: ConnectedWorkerRegistry, worker: WorkerId) -> Self {
        Self {
            gate,
            registry,
            worker,
        }
    }
}

impl DeliveryIntent for DispatcherPass {
    fn still_wanted(&self) -> bool {
        if self.gate.is_draining() {
            return false;
        }
        match self.registry.worker_by_id(self.worker) {
            Ok(Some(_)) => true,
            Ok(None) => false,
            Err(error) => {
                // A registry that cannot be read cannot license a continued
                // wait: the honest answer is to abandon and let the caller's
                // retry path run, loudly, rather than hold a delivery open on an
                // unreadable premise.
                tracing::warn!(
                    %error,
                    worker = ?self.worker,
                    "delivery wait could not verify worker registration; abandoning"
                );
                false
            }
        }
    }
}

/// A caller with no abandonment condition at all.
///
/// Named, rather than spelled `|| true` at a call site, so that the decision is
/// **visible and greppable**: every place a delivery is allowed to wait
/// unconditionally says so by naming this type.
#[derive(Debug, Clone, Copy, Default)]
pub struct AlwaysWanted;

impl DeliveryIntent for AlwaysWanted {
    fn still_wanted(&self) -> bool {
        true
    }
}

/// Shared handle to a caller's intent, for the delivery paths that must move it
/// onto a blocking thread.
pub type SharedDeliveryIntent = Arc<dyn DeliveryIntent>;