aion_server/worker/queue_service/

census.rs

//! Live poller census for one pool address, and the classification it feeds.
//!
//! The census is taken from the connected-worker registry at the same moment
//! selection misses, so the taxonomy verdict describes the fleet the selection
//! actually saw — never a second, later, disagreeing observation.

use std::time::Duration;

use super::declarations::QueueDeclaration;
use super::taxonomy::QueueServiceReason;

/// What the connected-worker registry holds for one
/// `(namespace, task_queue, activity_type[, node])` address.
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
pub struct PoolCensus {
    /// Workers connected for `(namespace, task_queue)`, whatever they serve.
    pub workers_in_pool: usize,
    /// Of those, workers advertising this activity type.
    pub workers_serving_activity: usize,
    /// Of those, workers that also satisfy the dispatch's node pin. Equal to
    /// [`Self::workers_serving_activity`] for an unpinned dispatch.
    ///
    /// Deliberately NOT filtered by dispatch eligibility, so
    /// [`classify`] can tell an empty pool from an excluded one (#197 R3).
    pub compatible_workers: usize,
    /// Of [`Self::compatible_workers`], those a dispatch could actually be
    /// handed to right now — the rest are excluded by the liveness verdict.
    pub eligible_compatible_workers: usize,
    /// Of the excluded compatible workers, those excluded by
    /// `DispatchExclusion::ReachabilityLost` rather than by an opening
    /// probation.
    ///
    /// 🔴 The whole point of the distinction. An opening probation clears
    /// itself in seconds, so parking silently through one is correct and
    /// announcing it would be noise on every healthy connect. A lost
    /// reachability does NOT clear on its own, so a dispatch parked on one
    /// waits indefinitely — and must say so, or it is a silent stall.
    pub compatible_workers_reachability_lost: usize,
    /// How long ago a compatible worker was last in service, or `None` when
    /// none has ever been seen for this address in this server's life.
    ///
    /// Zero while a compatible worker is connected right now.
    pub last_compatible_poller_age: Option<Duration>,
}

impl PoolCensus {
    /// Whether the address currently has a compatible worker REGISTERED,
    /// whatever its dispatch eligibility.
    ///
    /// A field-level fact, not a decision. It stays deliberately blind to
    /// eligibility because [`classify`] needs the blind count to tell an empty
    /// pool from an excluded one (#197 R3). Callers asking "will a dispatch
    /// aimed here actually proceed?" want [`Self::will_be_served`].
    #[must_use]
    pub const fn is_served(&self) -> bool {
        self.compatible_workers > 0
    }

    /// Whether a dispatch aimed here can be expected to proceed without an
    /// operator doing something first.
    ///
    /// 🔴 The distinction [`Self::is_served`] cannot draw. A pool whose workers
    /// are all serving an opening probation is not dispatchable this instant
    /// but will be within seconds, so it answers `true` — waiting is the
    /// correct outcome and the caller should not be warned. A pool whose
    /// workers have all LOST reachability answers `false`: nothing about it
    /// improves on its own, so a caller told "served" would be told a run is
    /// about to proceed when it is about to park indefinitely.
    #[must_use]
    pub const fn will_be_served(&self) -> bool {
        self.eligible_compatible_workers > 0
            || (self.compatible_workers > 0 && self.compatible_workers_reachability_lost == 0)
    }
}

/// Classify one selection miss into the taxonomy.
///
/// Returns `None` when the caller must simply retry selection rather than
/// report a state that is no longer true. That covers two cases: the address is
/// genuinely served because a worker arrived between the two lock
/// acquisitions, and the address's compatible workers are all serving an
/// opening probation, which clears on its own within seconds.
///
/// 🔴 It does NOT cover a pool whose compatible workers have all LOST
/// reachability. That does not clear on its own, so returning `None` there
/// parks the dispatch indefinitely with nothing published about why — the
/// silent stall this arm exists to prevent.
///
/// Ordering is deliberate: a structurally undeclared queue is reported as
/// [`QueueServiceReason::NoQueueDeclaration`] even when unrelated workers sit
/// in the pool, because that is the deeper fact and the only one that refuses
/// unconditionally. [`QueueDeclaration::Unknown`] never manufactures a
/// structural refusal — the same fail-open discipline worker registration
/// applies when no catalog is reachable.
///
/// [`QueueServiceReason::Saturated`] is never produced here: it is not a
/// selection miss. It is emitted by the delivery path
/// ([`super::delivery::deliver_within_schedule_to_start`]) when a compatible
/// worker is live and refuses intake past the schedule-to-start clock.
#[must_use]
pub fn classify(declaration: QueueDeclaration, census: &PoolCensus) -> Option<QueueServiceReason> {
    if census.eligible_compatible_workers > 0 {
        return None;
    }
    if declaration == QueueDeclaration::NotDeclared {
        return Some(QueueServiceReason::NoQueueDeclaration);
    }
    if census.compatible_workers > 0 {
        // Every compatible worker is excluded by the liveness verdict. Which
        // exclusion decides whether this is worth saying: an opening probation
        // is the ordinary cost of connecting and clears itself, and reporting
        // it would fire on every healthy worker's first seconds.
        //
        // `PollersIncompatible` would be a LIE here and not merely a vague
        // one — its sentence is "workers are connected but none serves this
        // activity", and every one of these serves it. The remedy it implies
        // (deploy a worker that covers the type) is wrong; the real remedy is
        // to read the liveness verdict for workers already present.
        if census.compatible_workers_reachability_lost > 0 {
            return Some(QueueServiceReason::PollersUnreachable);
        }
        return None;
    }
    if census.workers_in_pool > 0 {
        return Some(QueueServiceReason::PollersIncompatible);
    }
    Some(QueueServiceReason::NoLivePollers)
}

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

    #[test]
    fn a_served_address_is_not_classified() {
        let census = PoolCensus {
            workers_in_pool: 1,
            workers_serving_activity: 1,
            compatible_workers: 1,
            eligible_compatible_workers: 1,
            compatible_workers_reachability_lost: 0,
            last_compatible_poller_age: Some(Duration::ZERO),
        };
        assert_eq!(classify(QueueDeclaration::Declared, &census), None);
        assert_eq!(classify(QueueDeclaration::NotDeclared, &census), None);
    }

    #[test]
    fn an_undeclared_queue_is_structural() {
        assert_eq!(
            classify(QueueDeclaration::NotDeclared, &PoolCensus::default()),
            Some(QueueServiceReason::NoQueueDeclaration)
        );
    }

    #[test]
    fn an_empty_pool_on_a_declared_queue_has_no_live_pollers() {
        assert_eq!(
            classify(QueueDeclaration::Declared, &PoolCensus::default()),
            Some(QueueServiceReason::NoLivePollers)
        );
    }

    #[test]
    fn an_unknowable_declaration_never_produces_a_structural_refusal() {
        assert_eq!(
            classify(QueueDeclaration::Unknown, &PoolCensus::default()),
            Some(QueueServiceReason::NoLivePollers)
        );
    }

    /// A pool of `compatible` workers of whom `eligible` can be dispatched to,
    /// with `lost` of the remainder excluded for lost reachability rather than
    /// for an opening probation.
    fn pool(compatible: usize, eligible: usize, lost: usize) -> PoolCensus {
        PoolCensus {
            workers_in_pool: compatible,
            workers_serving_activity: compatible,
            compatible_workers: compatible,
            eligible_compatible_workers: eligible,
            compatible_workers_reachability_lost: lost,
            last_compatible_poller_age: Some(Duration::ZERO),
        }
    }

    /// A pool whose workers are all still serving their opening probation is
    /// NOT reported: the exclusion clears itself within seconds, and reporting
    /// it would fire on every healthy worker's first moments.
    #[test]
    fn a_pool_excluded_only_by_an_opening_probation_is_not_reported() {
        assert_eq!(classify(QueueDeclaration::Declared, &pool(2, 0, 0)), None);
    }

    /// The same pool, excluded for lost reachability, IS reported — because
    /// that exclusion does not clear on its own, so a dispatch parked on it
    /// waits indefinitely with nothing said.
    #[test]
    fn a_pool_that_lost_reachability_is_reported() {
        assert_eq!(
            classify(QueueDeclaration::Declared, &pool(2, 0, 2)),
            Some(QueueServiceReason::PollersUnreachable)
        );
    }

    /// 🔴 THE DISCRIMINATOR, and the reason the two tests above are not enough
    /// on their own.
    ///
    /// Each of them asserts one census against one expected answer, and BOTH
    /// would still pass if `classify` had been written to ignore the cause and
    /// return a constant — the first passes on a constant `None`, the second
    /// on a constant `Some(PollersUnreachable)`. Neither witnesses that the
    /// CAUSE is what decides.
    ///
    /// This one does: two censuses identical in every field except the
    /// exclusion cause must produce different answers. That is the property the
    /// whole change exists for, and it cannot be satisfied by a constant.
    #[test]
    fn the_exclusion_cause_is_what_decides() {
        let probation = pool(2, 0, 0);
        let unreachable = pool(2, 0, 2);
        assert_eq!(
            probation.workers_in_pool, unreachable.workers_in_pool,
            "precondition: the two pools differ ONLY in the exclusion cause"
        );
        assert_eq!(probation.compatible_workers, unreachable.compatible_workers);
        assert_eq!(
            probation.eligible_compatible_workers,
            unreachable.eligible_compatible_workers
        );

        assert_ne!(
            classify(QueueDeclaration::Declared, &probation),
            classify(QueueDeclaration::Declared, &unreachable),
            "a pool serving its probation and a pool that lost reachability must not get the \
             same answer: one clears itself in seconds and the other never does, and the \
             operator's remedies are different"
        );
    }

    /// 🔴 THE CONTROL for the reporting test, and it is not optional.
    ///
    /// Without it, a `classify` hard-coded to report `PollersUnreachable`
    /// whenever nothing is dispatchable would satisfy every test above AND
    /// misdiagnose a genuinely empty pool — telling an operator that starting
    /// the worker they need would not help, which is the opposite of the truth.
    #[test]
    fn a_genuinely_empty_pool_is_still_reported_as_having_no_pollers() {
        assert_eq!(
            classify(QueueDeclaration::Declared, &PoolCensus::default()),
            Some(QueueServiceReason::NoLivePollers)
        );
    }

    /// The second control: workers present that do NOT serve the activity are
    /// still `PollersIncompatible`. The new arm sits ahead of that one, so a
    /// mistake in its guard would swallow this case and send an operator to
    /// read a liveness verdict for a worker whose only problem is that it does
    /// not cover the activity at all.
    #[test]
    fn workers_that_do_not_serve_the_activity_are_still_incompatible_not_unreachable() {
        let census = PoolCensus {
            workers_in_pool: 3,
            workers_serving_activity: 0,
            compatible_workers: 0,
            eligible_compatible_workers: 0,
            compatible_workers_reachability_lost: 0,
            last_compatible_poller_age: None,
        };
        assert_eq!(
            classify(QueueDeclaration::Declared, &census),
            Some(QueueServiceReason::PollersIncompatible)
        );
    }

    /// One eligible worker is enough: the address is served and the caller
    /// simply retries, whatever the others are excluded for.
    #[test]
    fn one_eligible_worker_means_the_address_is_served() {
        assert_eq!(classify(QueueDeclaration::Declared, &pool(3, 1, 2)), None);
    }

    #[test]
    fn pollers_that_do_not_cover_the_activity_are_incompatible() {
        let census = PoolCensus {
            workers_in_pool: 2,
            workers_serving_activity: 0,
            compatible_workers: 0,
            eligible_compatible_workers: 0,
            compatible_workers_reachability_lost: 0,
            last_compatible_poller_age: None,
        };
        assert_eq!(
            classify(QueueDeclaration::Declared, &census),
            Some(QueueServiceReason::PollersIncompatible)
        );
    }

    #[test]
    fn pollers_off_the_pinned_node_are_incompatible() {
        let census = PoolCensus {
            workers_in_pool: 1,
            workers_serving_activity: 1,
            compatible_workers: 0,
            eligible_compatible_workers: 0,
            compatible_workers_reachability_lost: 0,
            last_compatible_poller_age: Some(Duration::from_secs(9)),
        };
        assert_eq!(
            classify(QueueDeclaration::Declared, &census),
            Some(QueueServiceReason::PollersIncompatible)
        );
    }
}