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,
    /// Of [`Self::compatible_workers`], those already holding every dispatch
    /// they advertised they would run at once.
    ///
    /// 🔴 A THIRD condition, and the reason [`classify`] needs a third answer.
    /// An at-capacity pool is neither empty nor unreachable: its workers are
    /// connected, serving this activity, and working. The remedy is the
    /// OPPOSITE of the unreachable one — starting another worker helps here and
    /// does not help there — so reporting either existing reason for it would
    /// send an operator the wrong way.
    pub compatible_workers_at_capacity: usize,
    /// Of [`Self::compatible_workers`], those that have registered but have not
    /// yet announced how much work they can take.
    ///
    /// 🔴 DISJOINT from [`Self::compatible_workers_at_capacity`], and counted
    /// apart from it on purpose. Selection excludes both, by the same test —
    /// unknown capacity is treated as full — so a census that folded them
    /// together would be arithmetically fine and operationally a lie: it would
    /// report a pool of idle, mute workers as a pool of busy ones, and attach
    /// the advice for a busy pool ("start another worker") to a condition
    /// another worker of the same build cannot fix.
    ///
    /// Only the liminal transport can produce it, and normally only for the one
    /// round trip between a worker's registration and its capacity
    /// announcement.
    pub compatible_workers_capacity_unannounced: 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.
    ///
    /// A pool whose workers are all AT CAPACITY answers `true` for the same
    /// reason as the probation case: it is not dispatchable this instant and
    /// needs nobody to do anything — a running activity finishes, a slot frees,
    /// and the parked dispatch is selected. Waiting is the correct outcome.
    ///
    /// A pool whose workers have NOT ANNOUNCED their capacity answers `false`,
    /// and is the reason this is no longer two clauses. Nothing about that pool
    /// improves on its own: the announcement either already arrived — in which
    /// case these workers are not in this bucket — or it never will, because the
    /// worker is a build that does not send one or its publish was lost. Saying
    /// "will be served" of a fleet that is registered, idle, mute and
    /// permanently unselectable is the most expensive kind of wrong answer,
    /// because the whole point of this predicate is to tell a caller whether
    /// waiting is enough.
    #[must_use]
    pub const fn will_be_served(&self) -> bool {
        if self.eligible_compatible_workers > 0 {
            return true;
        }
        self.compatible_workers > 0
            && self.compatible_workers_reachability_lost == 0
            && self.compatible_workers_capacity_unannounced == 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);
        }
        // Checked BEFORE the at-capacity arm, and the precedence is the point.
        // Selection excludes an unannounced worker by the same test it excludes
        // a full one, so both conditions can be true of one pool at once — and
        // only one reason can be published. The unannounced one wins because it
        // is the one an operator may have to act on: a busy pool clears itself
        // as work completes, whereas a worker that has not announced its
        // capacity past one round trip never will without being read or
        // replaced. Reporting the self-healing condition and burying the
        // durable one would be the wrong way round.
        //
        // `> 0` rather than `== compatible_workers` for the same reason: one
        // mute worker in an otherwise busy pool is still a fact worth surfacing,
        // and its sentence stays true of the pool ("not one of them has
        // announced" is scoped by the sentence to the workers it describes).
        if census.compatible_workers_capacity_unannounced > 0 {
            return Some(QueueServiceReason::PollersCapacityUnannounced);
        }
        // Every compatible worker is connected, serving this activity, and
        // FULL. That is a third condition, not a quieter version of either one
        // above: the pool is healthy and the work is queued behind real work.
        //
        // Reported only when the whole compatible population is at capacity. A
        // mixed pool — some full, some serving an opening probation — is
        // dominated by the probation, which clears itself within seconds, and
        // announcing anything for it would fire on every healthy worker start.
        if census.compatible_workers_at_capacity == census.compatible_workers {
            return Some(QueueServiceReason::PollersAtCapacity);
        }
        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),
            compatible_workers_at_capacity: 0,
            compatible_workers_capacity_unannounced: 0,
        };
        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 BUSY pool is neither empty nor unreachable, and the ladder must say
    /// so in its own words.
    ///
    /// This is the arm the outage produced and nothing reported: every worker
    /// connected, serving the activity, and full. Before the arm existed the
    /// ladder had two ways to describe it and both were false — it fell through
    /// to `NoLivePollers` ("no worker is connected") once the sweep had
    /// deregistered the busy worker, and would otherwise have returned `None`
    /// and parked in silence.
    #[test]
    fn a_pool_whose_every_worker_is_full_reports_at_capacity() {
        let census = PoolCensus {
            workers_in_pool: 2,
            workers_serving_activity: 2,
            compatible_workers: 2,
            eligible_compatible_workers: 0,
            compatible_workers_reachability_lost: 0,
            compatible_workers_at_capacity: 2,
            compatible_workers_capacity_unannounced: 0,
            last_compatible_poller_age: Some(Duration::ZERO),
        };
        assert_eq!(
            classify(QueueDeclaration::Declared, &census),
            Some(QueueServiceReason::PollersAtCapacity)
        );
        // The discriminating half: the SAME pool with nobody connected is the
        // reason this arm exists to stop being confused with.
        assert_eq!(
            classify(QueueDeclaration::Declared, &PoolCensus::default()),
            Some(QueueServiceReason::NoLivePollers)
        );
        // And a busy pool is one a dispatch should WAIT on: nothing about it
        // needs an operator, so the caller must not be told it will not be
        // served.
        assert!(census.will_be_served());
    }

    /// A pool whose workers have not announced their capacity is reported as
    /// UNANNOUNCED, never as busy.
    ///
    /// Selection excludes both by the same test — unknown capacity is treated as
    /// full — so this is exactly the confusion the separate count exists to
    /// prevent. Reporting these workers as at-capacity would tell an operator
    /// their idle, mute fleet was busy, and hand them the advice for a busy pool
    /// ("start another worker") for a condition another worker of the same build
    /// reproduces exactly.
    #[test]
    fn a_pool_whose_workers_have_not_announced_capacity_is_not_reported_as_busy() {
        let census = PoolCensus {
            workers_in_pool: 2,
            workers_serving_activity: 2,
            compatible_workers: 2,
            eligible_compatible_workers: 0,
            compatible_workers_reachability_lost: 0,
            compatible_workers_at_capacity: 0,
            compatible_workers_capacity_unannounced: 2,
            last_compatible_poller_age: Some(Duration::ZERO),
        };
        assert_eq!(
            classify(QueueDeclaration::Declared, &census),
            Some(QueueServiceReason::PollersCapacityUnannounced)
        );
        assert_ne!(
            classify(QueueDeclaration::Declared, &census),
            Some(QueueServiceReason::PollersAtCapacity),
            "an unannounced pool must never wear the busy pool's name: their remedies differ"
        );
        // And it must NOT be reported as "will be served". Nothing about an
        // unannounced pool improves on its own: either the announcement has
        // already landed — in which case these workers are not in this bucket —
        // or it never will. Answering `true` here told a caller that waiting was
        // enough for a fleet that is registered, idle, and permanently
        // unselectable.
        assert!(
            !census.will_be_served(),
            "an all-unannounced pool is not going to serve anything by itself"
        );
    }

    /// One unannounced worker in an otherwise busy pool still reports
    /// UNANNOUNCED.
    ///
    /// The precedence is the subject. Both conditions are true here and only one
    /// reason can be published; the unannounced one wins because it is the one
    /// that may not clear on its own. A busy pool frees a slot as work
    /// completes; a worker that has not announced past one round trip never will
    /// without being read or replaced.
    #[test]
    fn an_unannounced_worker_outranks_a_busy_one() {
        let census = PoolCensus {
            workers_in_pool: 3,
            workers_serving_activity: 3,
            compatible_workers: 3,
            eligible_compatible_workers: 0,
            compatible_workers_reachability_lost: 0,
            compatible_workers_at_capacity: 2,
            compatible_workers_capacity_unannounced: 1,
            last_compatible_poller_age: Some(Duration::ZERO),
        };
        assert_eq!(
            classify(QueueDeclaration::Declared, &census),
            Some(QueueServiceReason::PollersCapacityUnannounced)
        );
    }

    /// The vacuity control for both tests above: with nobody unannounced, a full
    /// pool still reports AT CAPACITY.
    ///
    /// Without this, a precedence bug that returned `PollersCapacityUnannounced`
    /// unconditionally would satisfy every assertion above.
    #[test]
    fn a_busy_pool_with_nobody_unannounced_still_reports_at_capacity() {
        let census = PoolCensus {
            workers_in_pool: 2,
            workers_serving_activity: 2,
            compatible_workers: 2,
            eligible_compatible_workers: 0,
            compatible_workers_reachability_lost: 0,
            compatible_workers_at_capacity: 2,
            compatible_workers_capacity_unannounced: 0,
            last_compatible_poller_age: Some(Duration::ZERO),
        };
        assert_eq!(
            classify(QueueDeclaration::Declared, &census),
            Some(QueueServiceReason::PollersAtCapacity)
        );
    }

    /// A pool that is full AND has lost reachability reports the incident, not
    /// the capacity: an unreachable worker is not going to free a slot, so
    /// "wait for capacity" would be advice to wait for something that is not
    /// coming.
    #[test]
    fn lost_reachability_outranks_capacity() {
        let census = PoolCensus {
            workers_in_pool: 2,
            workers_serving_activity: 2,
            compatible_workers: 2,
            eligible_compatible_workers: 0,
            compatible_workers_reachability_lost: 1,
            compatible_workers_at_capacity: 2,
            compatible_workers_capacity_unannounced: 0,
            last_compatible_poller_age: Some(Duration::ZERO),
        };
        assert_eq!(
            classify(QueueDeclaration::Declared, &census),
            Some(QueueServiceReason::PollersUnreachable)
        );
    }

    /// A MIXED pool — one worker full, one still serving its opening
    /// probation — is not reported at all. The probation clears itself within
    /// seconds and is the ordinary cost of connecting, so announcing anything
    /// here would fire on every healthy worker start.
    #[test]
    fn a_pool_only_partly_at_capacity_is_not_reported() {
        let census = PoolCensus {
            workers_in_pool: 2,
            workers_serving_activity: 2,
            compatible_workers: 2,
            eligible_compatible_workers: 0,
            compatible_workers_reachability_lost: 0,
            compatible_workers_at_capacity: 1,
            compatible_workers_capacity_unannounced: 0,
            last_compatible_poller_age: Some(Duration::ZERO),
        };
        assert_eq!(classify(QueueDeclaration::Declared, &census), None);
    }

    /// 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),
            compatible_workers_at_capacity: 0,
            compatible_workers_capacity_unannounced: 0,
        }
    }

    /// 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,
            compatible_workers_at_capacity: 0,
            compatible_workers_capacity_unannounced: 0,
        };
        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,
            compatible_workers_at_capacity: 0,
            compatible_workers_capacity_unannounced: 0,
        };
        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)),
            compatible_workers_at_capacity: 0,
            compatible_workers_capacity_unannounced: 0,
        };
        assert_eq!(
            classify(QueueDeclaration::Declared, &census),
            Some(QueueServiceReason::PollersIncompatible)
        );
    }
}