aion_server/worker/queue_service/
policy.rs

1//! Per-dispatch service policy and the two service clocks.
2//!
3//! R1 rules two policies and two clocks, and this module is where a dispatch's
4//! pair is resolved from what the operator actually wrote.
5//!
6//! **strict** is the default, everywhere, always: a dispatch to a queue that is
7//! structurally unserved refuses instead of starting a lie. **`durable_pending`**
8//! is a written opt-in per queue — the run is marked unserved immediately and
9//! visibly, then parks durably until a worker arrives.
10//!
11//! The two clocks are never conflated:
12//!
13//! * the **service-availability deadline** bounds waiting for a compatible
14//!   worker to EXIST (`NO_LIVE_POLLERS` / `POLLERS_INCOMPATIBLE`);
15//! * the **schedule-to-start timeout** bounds the hand-off to a worker that
16//!   already exists (`SATURATED`).
17//!
18//! Both are `Option`: unset means unbounded, which is exactly today's
19//! behaviour and the only honest default — a server does not get to invent the
20//! deadline an operator never declared. What DID change is that an unbounded
21//! wait is now loud, typed, and queryable instead of silent.
22
23use std::time::Duration;
24
25use serde::Deserialize;
26
27/// What a dispatch does when its queue is not being served.
28#[derive(Clone, Copy, Debug, Default, Deserialize, Eq, PartialEq)]
29#[serde(rename_all = "snake_case")]
30pub enum QueueServicePolicy {
31    /// Refuse rather than admit work onto a queue that cannot serve it.
32    /// Structural unservability refuses immediately; a fleet condition refuses
33    /// when the service-availability deadline expires. The default.
34    #[default]
35    Strict,
36    /// Accept, mark the run unserved immediately and visibly, and park durably
37    /// until a compatible worker arrives. Written opt-in, per queue. Structural
38    /// unservability still refuses — a queue no deployment declares can never
39    /// be served, so pending on it would be the lie the policy exists to avoid.
40    DurablePending,
41}
42
43impl QueueServicePolicy {
44    /// The canonical spelling used in config and in logs.
45    #[must_use]
46    pub const fn as_str(self) -> &'static str {
47        match self {
48            Self::Strict => "strict",
49            Self::DurablePending => "durable_pending",
50        }
51    }
52}
53
54impl std::fmt::Display for QueueServicePolicy {
55    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
56        formatter.write_str(self.as_str())
57    }
58}
59
60/// One written per-queue policy opt-in.
61#[derive(Clone, Debug, Deserialize, Eq, PartialEq)]
62#[serde(deny_unknown_fields)]
63pub struct QueueServiceOverride {
64    /// Namespace the override applies to, or `None` for every namespace.
65    #[serde(default)]
66    pub namespace: Option<String>,
67    /// Task queue the override applies to.
68    pub task_queue: String,
69    /// Policy this queue is served under.
70    pub policy: QueueServicePolicy,
71}
72
73/// The operator's queue-service settings.
74#[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq)]
75#[serde(default, deny_unknown_fields)]
76pub struct QueueServiceConfig {
77    /// Policy for every queue with no written override. `strict`.
78    pub default_policy: QueueServicePolicy,
79    /// Written per-queue opt-ins. The first entry matching a dispatch wins, so
80    /// a namespace-scoped entry placed before a namespace-less one is honoured.
81    pub overrides: Vec<QueueServiceOverride>,
82    /// How long a dispatch waits for a compatible worker to exist before a
83    /// typed `WorkerUnavailable` refusal. Unset = unbounded.
84    #[serde(default, with = "optional_duration_millis")]
85    pub service_availability_deadline: Option<Duration>,
86    /// How long a dispatch tries to hand its task to a live compatible worker
87    /// before a typed `SATURATED` refusal. Unset = one attempt, exactly as
88    /// before.
89    #[serde(default, with = "optional_duration_millis")]
90    pub schedule_to_start_timeout: Option<Duration>,
91}
92
93impl QueueServiceConfig {
94    /// The policy governing one dispatch address.
95    #[must_use]
96    pub fn policy_for(&self, namespace: &str, task_queue: &str) -> QueueServicePolicy {
97        self.overrides
98            .iter()
99            .find(|entry| {
100                entry.task_queue == task_queue
101                    && entry
102                        .namespace
103                        .as_ref()
104                        .is_none_or(|scoped| scoped == namespace)
105            })
106            .map_or(self.default_policy, |entry| entry.policy)
107    }
108
109    /// Whether the service-availability deadline governs this dispatch.
110    ///
111    /// `durable_pending` is precisely the declaration that this dispatch waits
112    /// for its worker, so the availability deadline — the strict policy's
113    /// instrument for refusing rather than lying — does not cut it short. The
114    /// schedule-to-start clock still applies under both policies: it bounds a
115    /// hand-off to a worker that already exists, which is a different question.
116    #[must_use]
117    pub const fn availability_deadline_for(&self, policy: QueueServicePolicy) -> Option<Duration> {
118        match policy {
119            QueueServicePolicy::Strict => self.service_availability_deadline,
120            QueueServicePolicy::DurablePending => None,
121        }
122    }
123}
124
125/// Milliseconds-or-absent duration fields, matching the `duration_millis`
126/// convention the rest of the server config uses.
127mod optional_duration_millis {
128    use std::time::Duration;
129
130    use serde::{Deserialize, Deserializer};
131
132    pub(super) fn deserialize<'de, D>(deserializer: D) -> Result<Option<Duration>, D::Error>
133    where
134        D: Deserializer<'de>,
135    {
136        let millis = Option::<u64>::deserialize(deserializer)?;
137        Ok(millis.map(Duration::from_millis))
138    }
139}
140
141#[cfg(test)]
142mod tests {
143    use super::*;
144
145    #[test]
146    fn strict_is_the_default_policy_and_no_clock_is_invented() {
147        let config = QueueServiceConfig::default();
148        assert_eq!(config.default_policy, QueueServicePolicy::Strict);
149        assert_eq!(
150            config.policy_for("default", "general"),
151            QueueServicePolicy::Strict
152        );
153        assert_eq!(config.service_availability_deadline, None);
154        assert_eq!(config.schedule_to_start_timeout, None);
155    }
156
157    #[test]
158    fn a_written_override_opts_one_queue_into_durable_pending() {
159        let config = QueueServiceConfig {
160            overrides: vec![QueueServiceOverride {
161                namespace: None,
162                task_queue: "general".to_owned(),
163                policy: QueueServicePolicy::DurablePending,
164            }],
165            ..QueueServiceConfig::default()
166        };
167        assert_eq!(
168            config.policy_for("default", "general"),
169            QueueServicePolicy::DurablePending
170        );
171        // Every other queue keeps the strict default.
172        assert_eq!(
173            config.policy_for("default", "billing"),
174            QueueServicePolicy::Strict
175        );
176    }
177
178    #[test]
179    fn an_override_can_be_scoped_to_one_namespace() {
180        let config = QueueServiceConfig {
181            overrides: vec![QueueServiceOverride {
182                namespace: Some("lab".to_owned()),
183                task_queue: "general".to_owned(),
184                policy: QueueServicePolicy::DurablePending,
185            }],
186            ..QueueServiceConfig::default()
187        };
188        assert_eq!(
189            config.policy_for("lab", "general"),
190            QueueServicePolicy::DurablePending
191        );
192        assert_eq!(
193            config.policy_for("prod", "general"),
194            QueueServicePolicy::Strict
195        );
196    }
197
198    #[test]
199    fn the_availability_deadline_bounds_strict_only() {
200        let config = QueueServiceConfig {
201            service_availability_deadline: Some(Duration::from_secs(30)),
202            ..QueueServiceConfig::default()
203        };
204        assert_eq!(
205            config.availability_deadline_for(QueueServicePolicy::Strict),
206            Some(Duration::from_secs(30))
207        );
208        assert_eq!(
209            config.availability_deadline_for(QueueServicePolicy::DurablePending),
210            None
211        );
212    }
213
214    #[test]
215    fn queue_service_settings_deserialize_from_the_written_form() -> Result<(), toml::de::Error> {
216        let config: QueueServiceConfig = toml::from_str(
217            r#"
218            default_policy = "strict"
219            service_availability_deadline = 45000
220            schedule_to_start_timeout = 5000
221
222            [[overrides]]
223            task_queue = "general"
224            policy = "durable_pending"
225            "#,
226        )?;
227        assert_eq!(config.default_policy, QueueServicePolicy::Strict);
228        assert_eq!(
229            config.service_availability_deadline,
230            Some(Duration::from_secs(45))
231        );
232        assert_eq!(
233            config.schedule_to_start_timeout,
234            Some(Duration::from_secs(5))
235        );
236        assert_eq!(
237            config.policy_for("default", "general"),
238            QueueServicePolicy::DurablePending
239        );
240        Ok(())
241    }
242}
aion_server/worker/queue_service/policy.rs

aion_server/worker/queue_service/
policy.rs
