aion-server 0.31.0

//! The queue-service taxonomy and its typed dispatch failure.
//!
//! Every reason in [`QueueServiceReason`] is emitted by a real path at the
//! bridge seam — six from the selection census ([`super::census::classify`])
//! and [`QueueServiceReason::Saturated`] from the delivery-backpressure path
//! ([`super::delivery`]). A variant that could not fire would be a name that
//! does not mean what it says.
//!
//! R1 defined four of these. [`QueueServiceReason::PollersUnreachable`] is the
//! fifth: a pool whose workers all serve the activity but have all lost
//! dispatch eligibility used to classify as served — because the census counts
//! compatible workers without an eligibility filter — and so parked with
//! nothing published about why.
//!
//! [`QueueServiceReason::PollersAtCapacity`] is the sixth, and the only one
//! that describes a fleet with nothing wrong with it: every compatible worker
//! is connected, serving the activity, and already running as many activities
//! as it advertised it would. It exists because the alternative was to report a
//! busy pool with one of the five sentences above, all of which are false about
//! it — and one of which advises the opposite of what would actually help.
//!
//! [`QueueServiceReason::PollersCapacityUnannounced`] is the seventh, and it
//! exists for the same reason the sixth does: a state that is real, that no
//! existing sentence is true of, and that would otherwise be reported as a
//! busy pool. A liminal worker cannot state its capacity in its registration
//! frame — `liminal::protocol::WorkerRegistration` is a published wire type
//! with no such field — so it registers with capacity UNKNOWN and announces the
//! number a frame later. Selection treats unknown as full, which is the safe
//! direction and the wrong WORD: a pool of workers that have not yet said what
//! they can take is not a pool of workers that are busy, and the two have
//! opposite remedies. Waiting fixes the busy one; only the worker announcing
//! (or being replaced) fixes this one, and starting another worker of the same
//! build fixes nothing at all.

use std::fmt;
use std::time::Duration;

use super::census::PoolCensus;

/// Failure-reason prefix classifying a dispatch failure as non-retryable.
const TERMINAL_PREFIX: &str = "terminal:";
/// Failure-reason prefix classifying a dispatch failure as retryable, matching
/// the vocabulary `aion`'s retry executor reads (`is_retryable_reason`).
const RETRYABLE_PREFIX: &str = "retryable:";
/// Stable tag identifying a queue-service refusal inside a dispatch failure.
const UNAVAILABLE_TAG: &str = "WORKER_UNAVAILABLE";
/// Separator between the machine-readable head and the human tail.
const TAIL_SEPARATOR: char = '—';
/// Field value standing in for an absent optional field.
const ABSENT: &str = "none";

/// Why a `(namespace, task_queue, activity_type)` address is not being served.
///
/// The names are R1's, verbatim, and are the strings that reach logs, dispatch
/// failures, and the queue-service accessor.
#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub enum QueueServiceReason {
    /// No deployed `.v4` contract declares this task queue at all. Structural:
    /// no worker can ever legitimately arrive for it, so it refuses
    /// unconditionally under every policy.
    NoQueueDeclaration,
    /// The queue is declared (or its declaration is unknowable) and no worker
    /// is connected for the pool at all.
    NoLivePollers,
    /// Workers are connected for the pool but none of them covers this
    /// activity type, or none of them sits on the dispatch's pinned node.
    PollersIncompatible,
    /// A compatible worker is connected and its intake is refusing work: the
    /// task could not be handed over before the schedule-to-start clock
    /// expired.
    Saturated,
    /// Workers are connected and DO serve this activity on this node, but the
    /// liveness verdict has withdrawn dispatch eligibility from all of them
    /// after losing reachability.
    ///
    /// 🔴 Distinct from [`Self::PollersIncompatible`] because the remedies are
    /// close to opposite. Incompatible pollers need a worker deployed that
    /// covers the activity; these workers already cover it, and deploying
    /// another changes nothing — the fault is on the wire to the ones already
    /// there. Distinct from an opening probation, which is not reported at all
    /// because it clears itself within seconds.
    PollersUnreachable,
    /// Workers are connected, DO serve this activity on this node, and every
    /// one of them is already running every activity it advertised it would run
    /// at once. The work is queued and starts as capacity frees.
    ///
    /// 🔴 The one reason in this taxonomy that describes a HEALTHY fleet, which
    /// is why it could not be folded into any of the others. Every existing
    /// sentence here would be a lie about a busy pool: nothing is missing, no
    /// wire has failed, no deployment is absent. Its remedy is the exact
    /// opposite of [`Self::PollersUnreachable`]'s — starting another worker
    /// helps here and does not help there — and two opposite pieces of advice
    /// cannot share one name. ([`Self::NoLivePollers`] shares the remedy but not
    /// the diagnosis: there the pool is empty, here it is full and working.)
    PollersAtCapacity,
    /// Workers are connected and DO serve this activity on this node, at least
    /// one of them has not announced how much work it can take, and no worker
    /// whose capacity IS known has a free slot — so the server has nothing it
    /// can honestly select against.
    ///
    /// Reported on `> 0` rather than on the whole population, and it OUTRANKS
    /// [`Self::PollersAtCapacity`] in [`super::census::classify`]. Both
    /// conditions can hold of one pool at once, only one reason can be
    /// published, and this is the one that may need an operator: a busy pool
    /// clears itself as work completes, a mute worker does not.
    ///
    /// 🔴 NOT a quieter [`Self::PollersAtCapacity`], and the distinction is the
    /// whole reason this variant exists. Selection excludes an unannounced
    /// worker by treating unknown capacity as full — the safe direction, since
    /// the alternative is inventing a number for a process that is about to
    /// state its own — but reporting that exclusion as "at capacity" would tell
    /// an operator their fleet was busy when it is idle and mute. The advice
    /// attached to a busy pool ("start another worker, or raise concurrency")
    /// is actively wrong here: another worker of the same build will be just as
    /// silent.
    ///
    /// Ordinarily invisible. It is true for exactly one round trip after a
    /// liminal worker registers, and a dispatch has to miss inside that window
    /// to see it. Persisting past that is the signal: the worker is an older
    /// build that does not announce, or its announcement was lost.
    PollersCapacityUnannounced,
}

impl QueueServiceReason {
    /// Every variant, and the single source of that set.
    ///
    /// 🔴 [`Self::parse`] and the round-trip test both read THIS, so a variant
    /// missing here is a variant that cannot be parsed back from its own
    /// spelling — a wire value the server writes and then fails to read. The
    /// exhaustive `match` in [`Self::as_str`] forces a new variant to be given
    /// a spelling, but nothing forces it into this list; `all_lists_every_variant`
    /// is what does, by matching exhaustively so the compiler refuses a variant
    /// this list has not been told about.
    pub const ALL: [Self; 7] = [
        Self::NoQueueDeclaration,
        Self::NoLivePollers,
        Self::PollersIncompatible,
        Self::PollersUnreachable,
        Self::PollersAtCapacity,
        Self::PollersCapacityUnannounced,
        Self::Saturated,
    ];

    /// The canonical wire/log spelling.
    #[must_use]
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::NoQueueDeclaration => "NO_QUEUE_DECLARATION",
            Self::NoLivePollers => "NO_LIVE_POLLERS",
            Self::PollersIncompatible => "POLLERS_INCOMPATIBLE",
            Self::Saturated => "SATURATED",
            Self::PollersUnreachable => "POLLERS_UNREACHABLE",
            Self::PollersAtCapacity => "POLLERS_AT_CAPACITY",
            Self::PollersCapacityUnannounced => "POLLERS_CAPACITY_UNANNOUNCED",
        }
    }

    /// Whether the unservability is structural — provable from the deployed
    /// contract records alone, unchangeable by waiting or retrying.
    #[must_use]
    pub const fn is_structural(self) -> bool {
        matches!(self, Self::NoQueueDeclaration)
    }

    /// The operator-facing sentence for this reason at `address`: what has to
    /// be fixed, in words.
    ///
    /// This is the single source of that sentence. The dispatch refusal's human
    /// tail ([`WorkerUnavailable::explain`]) and the describe projection's
    /// `detail` both read it here, so a queue that is unserved cannot be
    /// explained one way in a refusal and another way in a read.
    #[must_use]
    pub fn explain(self, address: &ServiceAddress) -> String {
        let ServiceAddress {
            namespace,
            task_queue,
            activity_type,
            node,
        } = address;
        let node = node
            .as_ref()
            .map_or_else(String::new, |node| format!(" pinned to node `{node}`"));
        match self {
            Self::NoQueueDeclaration => format!(
                "no deployed contract declares task queue `{task_queue}`, \
                 so activity `{activity_type}` in namespace `{namespace}` \
                 can never be served{node}"
            ),
            Self::NoLivePollers => format!(
                "no worker is connected for `{namespace}`/`{task_queue}`, \
                 so activity `{activity_type}` is unserved{node}"
            ),
            Self::PollersIncompatible => format!(
                "workers are connected for `{namespace}`/`{task_queue}` but \
                 none serves activity `{activity_type}`{node}"
            ),
            Self::Saturated => format!(
                "a worker serving `{namespace}`/`{task_queue}` activity \
                 `{activity_type}`{node} would not accept the task before the \
                 schedule-to-start timeout"
            ),
            Self::PollersUnreachable => format!(
                "every worker serving `{namespace}`/`{task_queue}` activity \
                 `{activity_type}`{node} has lost dispatch eligibility because \
                 the server cannot reach it. This is not an empty pool and not \
                 an unserved activity — starting another worker will not help. \
                 Read the liveness verdict for these workers; the reachability \
                 lines carry the worker ids and the failure"
            ),
            Self::PollersAtCapacity => format!(
                "every worker serving `{namespace}`/`{task_queue}` activity \
                 `{activity_type}`{node} is running every activity it advertised it \
                 would run at once. Nothing has failed: the workers are connected, \
                 they serve this activity, and this dispatch is queued behind real \
                 work. It starts as soon as one of them frees a slot. If the queue \
                 is persistently behind, start another worker for it or raise the \
                 concurrency of the ones already there — which is the opposite of \
                 what an unreachable pool needs"
            ),
            Self::PollersCapacityUnannounced => format!(
                "at least one worker serving `{namespace}`/`{task_queue}` activity \
                 `{activity_type}`{node} is connected and healthy at the transport but \
                 has not announced how much work it can take, and no worker whose \
                 capacity IS known has a free slot. The server is holding this dispatch \
                 rather than guessing a number on a worker's behalf. This is normally \
                 true for a fraction of a second after a worker connects. If it persists \
                 past one round trip, that worker is an older build that does not \
                 announce its capacity, or its announcement was lost — read that \
                 worker's own log, where the failure is at ERROR level. Starting another \
                 worker of the same build will NOT help, and the silent worker is not a \
                 busy one: it is idle"
            ),
        }
    }

    /// Read a reason back from its canonical spelling.
    #[must_use]
    pub fn parse(text: &str) -> Option<Self> {
        Self::ALL.into_iter().find(|reason| reason.as_str() == text)
    }
}

impl fmt::Display for QueueServiceReason {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter.write_str(self.as_str())
    }
}

/// Which of the two never-conflated service clocks expired.
#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub enum ExpiredClock {
    /// Time spent waiting for ANY compatible worker to exist.
    ServiceAvailability,
    /// Time spent between dispatch and a live compatible worker accepting the
    /// task into its intake.
    ScheduleToStart,
}

impl ExpiredClock {
    /// The canonical wire/log spelling.
    #[must_use]
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::ServiceAvailability => "service_availability",
            Self::ScheduleToStart => "schedule_to_start",
        }
    }

    /// Read a clock back from its canonical spelling.
    #[must_use]
    pub fn parse(text: &str) -> Option<Self> {
        [Self::ServiceAvailability, Self::ScheduleToStart]
            .into_iter()
            .find(|clock| clock.as_str() == text)
    }
}

impl fmt::Display for ExpiredClock {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter.write_str(self.as_str())
    }
}

/// The pool address a dispatch could not be served at.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ServiceAddress {
    /// Correctness/isolation boundary of the dispatch.
    pub namespace: String,
    /// Pool selector within the namespace.
    pub task_queue: String,
    /// Activity type the dispatch needs served.
    pub activity_type: String,
    /// Optional within-pool node pin carried by the dispatch.
    pub node: Option<String>,
}

/// A typed dispatch refusal: the taxonomy reason, which clock expired (if
/// any), how long the dispatch waited, and the live poller census behind the
/// verdict. Never a generic timeout, never a bare string.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct WorkerUnavailable {
    /// Taxonomy state of the queue at refusal time.
    pub reason: QueueServiceReason,
    /// The clock whose expiry produced the refusal, or `None` for a structural
    /// refusal (which needs no clock).
    pub clock: Option<ExpiredClock>,
    /// How long this dispatch waited before refusing.
    pub waited: Duration,
    /// Pool address that could not be served.
    pub address: ServiceAddress,
    /// Live poller census at refusal time.
    pub census: PoolCensus,
}

impl WorkerUnavailable {
    /// Whether the engine's retry executor should treat this failure as
    /// retryable.
    ///
    /// Structural unservability is terminal: no retry can conjure a deployment
    /// that declares the queue. Every other reason is a transient fleet
    /// condition (a worker may connect, backpressure may drain), so it carries
    /// the retryable prefix and the workflow's own declared retry policy — or
    /// its absence — decides what happens next. No retry budget is invented
    /// here.
    #[must_use]
    pub const fn is_retryable(&self) -> bool {
        !self.reason.is_structural()
    }

    /// The failure string handed back at the engine dispatch seam.
    #[must_use]
    pub fn reason_string(&self) -> String {
        self.to_string()
    }

    /// Read the typed head back out of a dispatch failure string.
    ///
    /// Returns `None` for any failure that is not a queue-service refusal, so
    /// surfacing code can ask "is this a `WorkerUnavailable`?" without string
    /// sniffing.
    #[must_use]
    pub fn parse(failure: &str) -> Option<UnavailableSummary> {
        let body = failure
            .strip_prefix(TERMINAL_PREFIX)
            .or_else(|| failure.strip_prefix(RETRYABLE_PREFIX))?;
        let body = body.strip_prefix(UNAVAILABLE_TAG)?;
        let head = body.split(TAIL_SEPARATOR).next().unwrap_or(body);
        let mut reason = None;
        let mut clock = None;
        let mut waited = Duration::ZERO;
        let mut last_compatible_poller_age = None;
        for token in head.split_whitespace() {
            let Some((key, value)) = token.split_once('=') else {
                continue;
            };
            match key {
                "reason" => reason = QueueServiceReason::parse(value),
                "clock" => clock = ExpiredClock::parse(value),
                "waited_ms" => waited = parse_millis(value).unwrap_or_default(),
                "last_compatible_poller_age_ms" => {
                    last_compatible_poller_age = parse_millis(value);
                }
                _ => {}
            }
        }
        Some(UnavailableSummary {
            reason: reason?,
            clock,
            waited,
            last_compatible_poller_age,
        })
    }
}

impl fmt::Display for WorkerUnavailable {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        let prefix = if self.is_retryable() {
            RETRYABLE_PREFIX
        } else {
            TERMINAL_PREFIX
        };
        let clock = self.clock.map_or(ABSENT, ExpiredClock::as_str);
        let age = self
            .census
            .last_compatible_poller_age
            .map_or_else(|| ABSENT.to_owned(), |age| millis(age).to_string());
        write!(
            formatter,
            "{prefix}{UNAVAILABLE_TAG} reason={} clock={clock} waited_ms={} \
             last_compatible_poller_age_ms={age} workers_in_pool={} \
             workers_serving_activity={} compatible_workers={} \
             {TAIL_SEPARATOR} {}",
            self.reason,
            millis(self.waited),
            self.census.workers_in_pool,
            self.census.workers_serving_activity,
            self.census.compatible_workers,
            self.explain(),
        )
    }
}

impl WorkerUnavailable {
    /// The human tail of the failure string: what an operator has to fix.
    fn explain(&self) -> String {
        self.reason.explain(&self.address)
    }
}

/// The typed head of a queue-service refusal, recovered from its string form.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct UnavailableSummary {
    /// Taxonomy state recorded in the refusal.
    pub reason: QueueServiceReason,
    /// Clock that expired, if any.
    pub clock: Option<ExpiredClock>,
    /// How long the refused dispatch waited.
    pub waited: Duration,
    /// Age of the last compatible poller at refusal time, if one was ever seen.
    pub last_compatible_poller_age: Option<Duration>,
}

/// Milliseconds of a duration, saturating rather than wrapping.
pub(super) fn millis(duration: Duration) -> u64 {
    u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)
}

fn parse_millis(value: &str) -> Option<Duration> {
    value.parse::<u64>().ok().map(Duration::from_millis)
}

#[cfg(test)]
mod tests {
    use super::*;

    fn address() -> ServiceAddress {
        ServiceAddress {
            namespace: "default".to_owned(),
            task_queue: "general".to_owned(),
            activity_type: "greet".to_owned(),
            node: None,
        }
    }

    fn census() -> PoolCensus {
        PoolCensus {
            workers_in_pool: 0,
            workers_serving_activity: 0,
            compatible_workers: 0,
            eligible_compatible_workers: 0,
            compatible_workers_reachability_lost: 0,
            last_compatible_poller_age: Some(Duration::from_millis(1_500)),
            compatible_workers_at_capacity: 0,
            compatible_workers_capacity_unannounced: 0,
        }
    }

    #[test]
    fn structural_refusal_is_terminal_and_carries_no_clock() {
        let failure = WorkerUnavailable {
            reason: QueueServiceReason::NoQueueDeclaration,
            clock: None,
            waited: Duration::ZERO,
            address: address(),
            census: PoolCensus::default(),
        };
        assert!(!failure.is_retryable());
        let rendered = failure.reason_string();
        assert!(rendered.starts_with("terminal:"), "{rendered}");
        assert!(
            rendered.contains("reason=NO_QUEUE_DECLARATION"),
            "{rendered}"
        );
        assert!(rendered.contains("clock=none"), "{rendered}");
    }

    #[test]
    fn fleet_refusals_are_retryable_and_name_the_expired_clock() {
        let failure = WorkerUnavailable {
            reason: QueueServiceReason::NoLivePollers,
            clock: Some(ExpiredClock::ServiceAvailability),
            waited: Duration::from_secs(2),
            address: address(),
            census: census(),
        };
        assert!(failure.is_retryable());
        let rendered = failure.reason_string();
        assert!(rendered.starts_with("retryable:"), "{rendered}");
        assert!(
            rendered.contains("clock=service_availability"),
            "{rendered}"
        );
        assert!(rendered.contains("waited_ms=2000"), "{rendered}");
        assert!(
            rendered.contains("last_compatible_poller_age_ms=1500"),
            "{rendered}"
        );
    }

    #[test]
    fn the_two_clocks_render_distinctly() {
        let availability = ExpiredClock::ServiceAvailability.as_str();
        let schedule_to_start = ExpiredClock::ScheduleToStart.as_str();
        assert_ne!(availability, schedule_to_start);
        assert_eq!(
            ExpiredClock::parse(availability),
            Some(ExpiredClock::ServiceAvailability)
        );
        assert_eq!(
            ExpiredClock::parse(schedule_to_start),
            Some(ExpiredClock::ScheduleToStart)
        );
    }

    #[test]
    fn a_refusal_round_trips_back_to_its_typed_head() {
        let failure = WorkerUnavailable {
            reason: QueueServiceReason::Saturated,
            clock: Some(ExpiredClock::ScheduleToStart),
            waited: Duration::from_millis(750),
            address: ServiceAddress {
                node: Some("n1".to_owned()),
                ..address()
            },
            census: census(),
        };
        let parsed = WorkerUnavailable::parse(&failure.reason_string());
        assert_eq!(
            parsed,
            Some(UnavailableSummary {
                reason: QueueServiceReason::Saturated,
                clock: Some(ExpiredClock::ScheduleToStart),
                waited: Duration::from_millis(750),
                last_compatible_poller_age: Some(Duration::from_millis(1_500)),
            })
        );
    }

    #[test]
    fn an_ordinary_failure_is_not_read_as_a_queue_service_refusal() {
        assert_eq!(WorkerUnavailable::parse("lost:worker lost"), None);
        assert_eq!(WorkerUnavailable::parse("terminal:boom"), None);
        assert_eq!(WorkerUnavailable::parse("parked:server-draining"), None);
    }

    #[test]
    fn every_reason_round_trips_through_its_canonical_spelling() {
        for reason in QueueServiceReason::ALL {
            assert_eq!(QueueServiceReason::parse(reason.as_str()), Some(reason));
        }
    }

    /// 🔴 The guard that makes the test above mean what its name says.
    ///
    /// Iterating `ALL` proves every reason IN THE LIST round-trips; it cannot
    /// prove the list is the whole enum. This match is exhaustive, so adding a
    /// variant stops the crate compiling until it is named here — and naming it
    /// here without adding it to `ALL` then fails the assertion. The previous
    /// version of the round-trip test spelled its own list inline and had
    /// neither property: a new variant simply went untested and unparseable.
    #[test]
    fn all_lists_every_variant() {
        fn is_listed(reason: QueueServiceReason) -> bool {
            match reason {
                QueueServiceReason::NoQueueDeclaration
                | QueueServiceReason::NoLivePollers
                | QueueServiceReason::PollersIncompatible
                | QueueServiceReason::PollersUnreachable
                | QueueServiceReason::PollersAtCapacity
                | QueueServiceReason::PollersCapacityUnannounced
                | QueueServiceReason::Saturated => QueueServiceReason::ALL.contains(&reason),
            }
        }
        for reason in QueueServiceReason::ALL {
            assert!(is_listed(reason), "{reason} is not in ALL");
        }
    }
}