solti_model/resource/
spec.rs

1//! # Task specification.
2//!
3//! [`TaskSpec`] defines the desired state; constructed via [`TaskSpecBuilder`].
4
5use serde::{Deserialize, Serialize};
6
7use crate::{
8    AdmissionPolicy, BackoffPolicy, Labels, RestartPolicy, RunnerSelector, Slot, TaskKind, Timeout,
9    error::{ModelError, ModelResult},
10};
11
12/// Desired state specification.
13///
14/// `TaskSpec` describes *what* should be run and *how* it should be managed by the runtime.
15///
16/// Fields cover:
17/// - logical grouping (`slot`)
18/// - execution backend (`kind`)
19/// - concurrency control (`admission`)
20/// - lifecycle policies (`timeout`, `restart`, `backoff`)
21///
22/// ## Also
23///
24/// - [`TaskSpecBuilder`] validated builder (via [`TaskSpec::builder`]).
25/// - [`TaskKind`] execution backend variants.
26/// - [`RestartPolicy`] / [`BackoffPolicy`] lifecycle policies.
27/// - [`AdmissionPolicy`] duplicate handling.
28/// - [`RunnerSelector`] label-based runner routing.
29#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
30#[serde(rename_all = "camelCase")]
31#[serde(try_from = "raw::TaskSpecRaw")]
32pub struct TaskSpec {
33    slot: Slot,
34    kind: TaskKind,
35
36    timeout: Timeout,
37    restart: RestartPolicy,
38    backoff: BackoffPolicy,
39    admission: AdmissionPolicy,
40    #[serde(default)]
41    max_retries: u32,
42
43    #[serde(default, skip_serializing_if = "Option::is_none")]
44    runner_selector: Option<RunnerSelector>,
45    #[serde(default, skip_serializing_if = "Labels::is_empty")]
46    labels: Labels,
47}
48
49impl TaskSpec {
50    /// Logical slot name for concurrency control.
51    #[inline]
52    pub fn slot(&self) -> &Slot {
53        &self.slot
54    }
55
56    /// Execution backend used to run the task.
57    #[inline]
58    pub fn kind(&self) -> &TaskKind {
59        &self.kind
60    }
61
62    /// Hard timeout in milliseconds.
63    #[inline]
64    pub fn timeout(&self) -> Timeout {
65        self.timeout
66    }
67
68    /// Restart policy applied after completion or failure.
69    #[inline]
70    pub fn restart(&self) -> RestartPolicy {
71        self.restart
72    }
73
74    /// Backoff configuration between restart attempts.
75    #[inline]
76    pub fn backoff(&self) -> &BackoffPolicy {
77        &self.backoff
78    }
79
80    /// Admission policy for handling slot conflicts.
81    #[inline]
82    pub fn admission(&self) -> AdmissionPolicy {
83        self.admission
84    }
85
86    /// Maximum failure-driven retries per run (`0` = unlimited, the default).
87    ///
88    /// Counts only failure retries (the counter resets on success); when the
89    /// budget is exhausted the supervisor stops restarting the task.
90    #[inline]
91    pub fn max_retries(&self) -> u32 {
92        self.max_retries
93    }
94
95    /// Label selector for runner routing (if present).
96    #[inline]
97    pub fn runner_selector(&self) -> Option<&RunnerSelector> {
98        self.runner_selector.as_ref()
99    }
100
101    /// Metadata labels for routing / scheduling / observability.
102    #[inline]
103    pub fn labels(&self) -> &Labels {
104        &self.labels
105    }
106}
107
108impl TaskSpec {
109    /// Create a [`TaskSpecBuilder`] with the three required fields.
110    ///
111    /// ```rust
112    /// use solti_model::{TaskSpec, TaskKind, SubprocessSpec, SubprocessMode, RestartPolicy};
113    ///
114    /// let spec = TaskSpec::builder(
115    ///     "my-slot",
116    ///     TaskKind::Subprocess(SubprocessSpec {
117    ///         mode: SubprocessMode::Command {
118    ///             command: "echo".into(),
119    ///             args: vec!["hello".into()],
120    ///         },
121    ///         env: Default::default(),
122    ///         cwd: None,
123    ///         fail_on_non_zero: Default::default(),
124    ///     }),
125    ///     5_000u64,
126    /// )
127    /// .restart(RestartPolicy::OnFailure)
128    /// .build()
129    /// .expect("valid spec");
130    /// ```
131    pub fn builder(
132        slot: impl Into<Slot>,
133        kind: TaskKind,
134        timeout: impl Into<Timeout>,
135    ) -> TaskSpecBuilder {
136        TaskSpecBuilder::new(slot, kind, timeout)
137    }
138}
139
140impl TaskSpec {
141    /// Attach a runner selector used by the router (consuming builder-style).
142    #[inline]
143    pub fn with_runner_selector(mut self, sel: RunnerSelector) -> Self {
144        self.runner_selector = Some(sel);
145        self
146    }
147
148    /// Override the admission policy (consuming builder-style).
149    ///
150    /// Used by the apply/upgrade path to force [`AdmissionPolicy::Replace`] regardless of the spec's declared admission.
151    #[inline]
152    pub fn with_admission(mut self, admission: AdmissionPolicy) -> Self {
153        self.admission = admission;
154        self
155    }
156}
157
158impl TaskSpec {
159    /// Validate the spec at the **submit boundary**.
160    pub fn validate(&self) -> ModelResult<()> {
161        self.validate_structural()?;
162        if matches!(self.kind, TaskKind::Embedded) {
163            return Err(ModelError::Invalid(
164                "TaskKind::Embedded cannot be submitted via runner; use submit_with_task".into(),
165            ));
166        }
167        Ok(())
168    }
169
170    /// Structural validation of all fields.
171    ///
172    /// Checks:
173    /// - `slot` is not empty
174    /// - `timeout` is greater than zero
175    /// - `kind` specific constraints (e.g. non-empty command)
176    /// - `backoff` parameters are sane
177    /// - `runner_selector` requirements are structurally valid
178    fn validate_structural(&self) -> ModelResult<()> {
179        self.slot.validate_format()?;
180        if self.timeout.as_millis() == 0 {
181            return Err(ModelError::Invalid(
182                "timeout must be greater than zero".into(),
183            ));
184        }
185        self.kind.validate()?;
186        self.backoff.validate()?;
187        if let Some(ref sel) = self.runner_selector {
188            for req in &sel.match_expressions {
189                req.validate()?;
190            }
191        }
192        Ok(())
193    }
194}
195
196/// Builder for [`TaskSpec`] that validates structural invariants on [`build`](TaskSpecBuilder::build).
197///
198/// Required fields (`slot`, `kind`, `timeout`) are set in the constructor.
199/// Optional fields have sensible defaults:
200/// - `backoff`: [`BackoffPolicy::default`] (full jitter, 1 s → 30 s, factor 2)
201/// - `admission`: [`AdmissionPolicy::DropIfRunning`]
202/// - `restart`: [`RestartPolicy::Never`]
203/// - `runner_selector`: `None`
204/// - `labels`: empty
205///
206/// ## Also
207///
208/// - [`TaskSpec::builder`] entry point.
209/// - [`TaskSpec::validate`] submit-boundary validation (rejects `Embedded`).
210pub struct TaskSpecBuilder {
211    runner_selector: Option<RunnerSelector>,
212
213    kind: TaskKind,
214    slot: Slot,
215
216    backoff: BackoffPolicy,
217    restart: RestartPolicy,
218    timeout: Timeout,
219    max_retries: u32,
220
221    admission: AdmissionPolicy,
222    labels: Labels,
223}
224
225impl TaskSpecBuilder {
226    fn new(slot: impl Into<Slot>, kind: TaskKind, timeout: impl Into<Timeout>) -> Self {
227        Self {
228            runner_selector: None,
229
230            kind,
231            slot: slot.into(),
232
233            restart: RestartPolicy::default(),
234            backoff: BackoffPolicy::default(),
235            timeout: timeout.into(),
236
237            admission: AdmissionPolicy::default(),
238            max_retries: 0,
239            labels: Labels::new(),
240        }
241    }
242
243    /// Set restart policy.
244    #[must_use]
245    pub fn restart(mut self, restart: RestartPolicy) -> Self {
246        self.restart = restart;
247        self
248    }
249
250    /// Set the failure-retry budget (`0` = unlimited, the default).
251    #[must_use]
252    pub fn max_retries(mut self, max_retries: u32) -> Self {
253        self.max_retries = max_retries;
254        self
255    }
256
257    /// Set backoff configuration.
258    #[must_use]
259    pub fn backoff(mut self, backoff: BackoffPolicy) -> Self {
260        self.backoff = backoff;
261        self
262    }
263
264    /// Set admission policy.
265    #[must_use]
266    pub fn admission(mut self, admission: AdmissionPolicy) -> Self {
267        self.admission = admission;
268        self
269    }
270
271    /// Set runner selector.
272    #[must_use]
273    pub fn runner_selector(mut self, sel: RunnerSelector) -> Self {
274        self.runner_selector = Some(sel);
275        self
276    }
277
278    /// Set metadata labels.
279    #[must_use]
280    pub fn labels(mut self, labels: Labels) -> Self {
281        self.labels = labels;
282        self
283    }
284
285    /// Build the [`TaskSpec`], validating structural invariants.
286    ///
287    /// This checks everything **except** the [`TaskKind::Embedded`] business rule
288    /// (which is enforced at the submit boundary by [`TaskSpec::validate`]).
289    ///
290    /// # Errors
291    ///
292    /// Returns [`ModelError::Invalid`] if:
293    /// - `slot` is empty
294    /// - `timeout` is zero
295    /// - `kind` fails kind-specific validation
296    /// - `backoff` parameters are invalid
297    /// - `runner_selector` requirements are invalid
298    pub fn build(self) -> ModelResult<TaskSpec> {
299        let spec = TaskSpec {
300            runner_selector: self.runner_selector,
301
302            kind: self.kind,
303            slot: self.slot,
304
305            restart: self.restart,
306            backoff: self.backoff,
307            timeout: self.timeout,
308
309            admission: self.admission,
310            max_retries: self.max_retries,
311            labels: self.labels,
312        };
313        spec.validate_structural()?;
314        Ok(spec)
315    }
316}
317
318mod raw {
319    use super::*;
320
321    #[derive(Deserialize)]
322    #[serde(rename_all = "camelCase")]
323    pub(super) struct TaskSpecRaw {
324        slot: Slot,
325        kind: TaskKind,
326        timeout: Timeout,
327        restart: RestartPolicy,
328        backoff: BackoffPolicy,
329        admission: AdmissionPolicy,
330        #[serde(default)]
331        max_retries: u32,
332
333        #[serde(default)]
334        labels: Labels,
335        #[serde(default)]
336        runner_selector: Option<RunnerSelector>,
337    }
338
339    impl TryFrom<TaskSpecRaw> for TaskSpec {
340        type Error = ModelError;
341
342        fn try_from(r: TaskSpecRaw) -> Result<Self, Self::Error> {
343            let spec = Self {
344                runner_selector: r.runner_selector,
345
346                kind: r.kind,
347                slot: r.slot,
348
349                restart: r.restart,
350                backoff: r.backoff,
351                timeout: r.timeout,
352
353                admission: r.admission,
354                max_retries: r.max_retries,
355                labels: r.labels,
356            };
357            spec.validate_structural()?;
358            Ok(spec)
359        }
360    }
361}
362
363#[cfg(test)]
364mod tests {
365    use super::*;
366    use crate::{Flag, SubprocessMode, SubprocessSpec, TaskEnv};
367
368    fn valid_spec() -> TaskSpec {
369        TaskSpec::builder(
370            "test",
371            TaskKind::Subprocess(SubprocessSpec {
372                mode: SubprocessMode::Command {
373                    command: "echo".into(),
374                    args: vec![],
375                },
376                env: TaskEnv::default(),
377                cwd: None,
378                fail_on_non_zero: Flag::enabled(),
379            }),
380            5_000u64,
381        )
382        .build()
383        .expect("test spec must be valid")
384    }
385
386    #[test]
387    fn valid_spec_passes() {
388        assert!(valid_spec().validate().is_ok());
389    }
390
391    #[test]
392    fn builder_rejects_empty_slot() {
393        let err = TaskSpec::builder("", TaskKind::Embedded, 5_000u64)
394            .build()
395            .unwrap_err();
396        assert!(err.to_string().contains("slot"));
397    }
398
399    #[test]
400    fn builder_rejects_zero_timeout() {
401        let err = TaskSpec::builder("test", TaskKind::Embedded, 0u64)
402            .build()
403            .unwrap_err();
404        assert!(err.to_string().contains("timeout"));
405    }
406
407    #[test]
408    fn builder_allows_embedded_kind() {
409        let spec = TaskSpec::builder("test", TaskKind::Embedded, 5_000u64)
410            .build()
411            .expect("Embedded is structurally valid");
412        assert!(matches!(spec.kind(), TaskKind::Embedded));
413    }
414
415    #[test]
416    fn validate_rejects_embedded_kind() {
417        let spec = TaskSpec::builder("test", TaskKind::Embedded, 5_000u64)
418            .build()
419            .unwrap();
420        let err = spec.validate().unwrap_err();
421        assert!(err.to_string().contains("TaskKind::Embedded"));
422    }
423
424    #[test]
425    fn getters_return_expected_values() {
426        let spec = TaskSpec::builder("my-slot", TaskKind::Embedded, 10_000u64)
427            .restart(RestartPolicy::OnFailure)
428            .admission(AdmissionPolicy::Replace)
429            .build()
430            .unwrap();
431
432        assert_eq!(spec.slot(), "my-slot");
433        assert_eq!(spec.timeout().as_millis(), 10_000);
434        assert_eq!(spec.restart(), RestartPolicy::OnFailure);
435        assert_eq!(spec.admission(), AdmissionPolicy::Replace);
436    }
437
438    #[test]
439    fn with_admission_overrides_policy() {
440        let spec = valid_spec().with_admission(AdmissionPolicy::Replace);
441        assert_eq!(spec.admission(), AdmissionPolicy::Replace);
442    }
443
444    #[test]
445    fn serde_roundtrip() {
446        let spec = valid_spec();
447        let json = serde_json::to_string(&spec).unwrap();
448        let back: TaskSpec = serde_json::from_str(&json).unwrap();
449        assert_eq!(back, spec);
450    }
451
452    #[test]
453    fn serde_rejects_empty_slot() {
454        let spec = valid_spec();
455        let mut json: serde_json::Value = serde_json::to_value(&spec).unwrap();
456        json["slot"] = serde_json::Value::String(String::new());
457
458        let err = serde_json::from_value::<TaskSpec>(json).unwrap_err();
459        assert!(err.to_string().contains("slot"), "error: {err}");
460    }
461
462    #[test]
463    fn serde_rejects_zero_timeout() {
464        let spec = valid_spec();
465        let mut json: serde_json::Value = serde_json::to_value(&spec).unwrap();
466        json["timeout"] = serde_json::json!(0);
467
468        let err = serde_json::from_value::<TaskSpec>(json).unwrap_err();
469        assert!(err.to_string().contains("timeout"), "error: {err}");
470    }
471}
solti_model/resource/spec.rs

solti_model/resource/
spec.rs
