aion_server/worker/queue_service/
census.rs

1//! Live poller census for one pool address, and the classification it feeds.
2//!
3//! The census is taken from the connected-worker registry at the same moment
4//! selection misses, so the taxonomy verdict describes the fleet the selection
5//! actually saw — never a second, later, disagreeing observation.
6
7use std::time::Duration;
8
9use super::declarations::QueueDeclaration;
10use super::taxonomy::QueueServiceReason;
11
12/// What the connected-worker registry holds for one
13/// `(namespace, task_queue, activity_type[, node])` address.
14#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
15pub struct PoolCensus {
16    /// Workers connected for `(namespace, task_queue)`, whatever they serve.
17    pub workers_in_pool: usize,
18    /// Of those, workers advertising this activity type.
19    pub workers_serving_activity: usize,
20    /// Of those, workers that also satisfy the dispatch's node pin. Equal to
21    /// [`Self::workers_serving_activity`] for an unpinned dispatch.
22    pub compatible_workers: usize,
23    /// How long ago a compatible worker was last in service, or `None` when
24    /// none has ever been seen for this address in this server's life.
25    ///
26    /// Zero while a compatible worker is connected right now.
27    pub last_compatible_poller_age: Option<Duration>,
28}
29
30impl PoolCensus {
31    /// Whether the address currently has a worker that could take the dispatch.
32    #[must_use]
33    pub const fn is_served(&self) -> bool {
34        self.compatible_workers > 0
35    }
36}
37
38/// Classify one selection miss into the taxonomy.
39///
40/// Returns `None` when the address is in fact served — the caller raced a
41/// worker arriving and must simply retry selection rather than report a state
42/// that is no longer true.
43///
44/// Ordering is deliberate: a structurally undeclared queue is reported as
45/// [`QueueServiceReason::NoQueueDeclaration`] even when unrelated workers sit
46/// in the pool, because that is the deeper fact and the only one that refuses
47/// unconditionally. [`QueueDeclaration::Unknown`] never manufactures a
48/// structural refusal — the same fail-open discipline worker registration
49/// applies when no catalog is reachable.
50///
51/// [`QueueServiceReason::Saturated`] is never produced here: it is not a
52/// selection miss. It is emitted by the delivery path
53/// ([`super::delivery::deliver_within_schedule_to_start`]) when a compatible
54/// worker is live and refuses intake past the schedule-to-start clock.
55#[must_use]
56pub fn classify(declaration: QueueDeclaration, census: &PoolCensus) -> Option<QueueServiceReason> {
57    if census.is_served() {
58        return None;
59    }
60    if declaration == QueueDeclaration::NotDeclared {
61        return Some(QueueServiceReason::NoQueueDeclaration);
62    }
63    if census.workers_in_pool > 0 {
64        return Some(QueueServiceReason::PollersIncompatible);
65    }
66    Some(QueueServiceReason::NoLivePollers)
67}
68
69#[cfg(test)]
70mod tests {
71    use super::*;
72
73    #[test]
74    fn a_served_address_is_not_classified() {
75        let census = PoolCensus {
76            workers_in_pool: 1,
77            workers_serving_activity: 1,
78            compatible_workers: 1,
79            last_compatible_poller_age: Some(Duration::ZERO),
80        };
81        assert_eq!(classify(QueueDeclaration::Declared, &census), None);
82        assert_eq!(classify(QueueDeclaration::NotDeclared, &census), None);
83    }
84
85    #[test]
86    fn an_undeclared_queue_is_structural() {
87        assert_eq!(
88            classify(QueueDeclaration::NotDeclared, &PoolCensus::default()),
89            Some(QueueServiceReason::NoQueueDeclaration)
90        );
91    }
92
93    #[test]
94    fn an_empty_pool_on_a_declared_queue_has_no_live_pollers() {
95        assert_eq!(
96            classify(QueueDeclaration::Declared, &PoolCensus::default()),
97            Some(QueueServiceReason::NoLivePollers)
98        );
99    }
100
101    #[test]
102    fn an_unknowable_declaration_never_produces_a_structural_refusal() {
103        assert_eq!(
104            classify(QueueDeclaration::Unknown, &PoolCensus::default()),
105            Some(QueueServiceReason::NoLivePollers)
106        );
107    }
108
109    #[test]
110    fn pollers_that_do_not_cover_the_activity_are_incompatible() {
111        let census = PoolCensus {
112            workers_in_pool: 2,
113            workers_serving_activity: 0,
114            compatible_workers: 0,
115            last_compatible_poller_age: None,
116        };
117        assert_eq!(
118            classify(QueueDeclaration::Declared, &census),
119            Some(QueueServiceReason::PollersIncompatible)
120        );
121    }
122
123    #[test]
124    fn pollers_off_the_pinned_node_are_incompatible() {
125        let census = PoolCensus {
126            workers_in_pool: 1,
127            workers_serving_activity: 1,
128            compatible_workers: 0,
129            last_compatible_poller_age: Some(Duration::from_secs(9)),
130        };
131        assert_eq!(
132            classify(QueueDeclaration::Declared, &census),
133            Some(QueueServiceReason::PollersIncompatible)
134        );
135    }
136}
aion_server/worker/queue_service/census.rs

aion_server/worker/queue_service/
census.rs
