solti_model/domain/query/
task.rs

1//! # Task query
2//!
3//! [`TaskPage`] returns a snapshot page and optional continuation.
4//! [`TaskFilter`] selects task resources.
5//! [`TaskQuery`] adds pagination.
6
7use serde::{Deserialize, Serialize};
8
9use crate::{LabelSelector, Labels, ModelError, ModelResult, Slot, Task, TaskId, TaskPhase};
10
11/// Default page size when the caller does not specify one.
12pub const DEFAULT_LIMIT: usize = 100;
13
14/// Hard cap on page size.
15///
16/// [`TaskQuery::with_limit`] clamps larger values.
17pub const MAX_LIMIT: usize = 1000;
18
19/// Filters shared by task list and watch operations.
20///
21/// An empty phase filter matches every phase.
22/// Multiple [`with_phase`](Self::with_phase) calls accumulate with OR semantics.
23/// Slot, label and phase filters are ANDed.
24///
25/// ## Example
26///
27/// ```
28/// use solti_model::{Slot, TaskFilter, TaskPhase};
29///
30/// let filter = TaskFilter::new()
31///     .with_slot(Slot::new("build").unwrap())
32///     .with_active();
33///
34/// assert_eq!(filter.slot().unwrap().as_str(), "build");
35/// assert!(filter.matches_phase(&TaskPhase::Pending));
36/// assert!(!filter.matches_phase(&TaskPhase::Failed));
37/// ```
38#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
39#[serde(
40    rename_all = "camelCase",
41    deny_unknown_fields,
42    try_from = "raw::TaskFilterRaw"
43)]
44pub struct TaskFilter {
45    phases: Vec<TaskPhase>,
46    slot: Option<Slot>,
47    label_selector: LabelSelector,
48}
49
50/// Position of the next page in one Task collection snapshot.
51///
52/// The cursor is a domain value, not a wire token.
53/// Transport layers encode it into their own opaque continuation representation.
54#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
55#[serde(
56    rename_all = "camelCase",
57    deny_unknown_fields,
58    try_from = "raw::TaskContinuationRaw"
59)]
60pub struct TaskContinuation {
61    resource_version: String,
62    filter: TaskFilter,
63    after: TaskId,
64}
65
66/// Query parameters for filtered, snapshot-consistent Task listing.
67///
68/// Filtering is carried by [`TaskFilter`].
69/// Pagination applies only to list operations.
70#[derive(Debug, Clone, PartialEq, Eq)]
71pub struct TaskQuery {
72    filter: TaskFilter,
73    limit: usize,
74    continuation: Option<TaskContinuation>,
75}
76
77impl Default for TaskQuery {
78    #[inline]
79    fn default() -> Self {
80        Self::new()
81    }
82}
83
84/// One page from a Task collection snapshot.
85#[derive(Debug, Clone, PartialEq, Eq)]
86pub struct TaskPage<T> {
87    /// Items on this page.
88    pub items: Vec<T>,
89    /// Opaque collection snapshot version.
90    pub resource_version: String,
91    /// Cursor for the next page.
92    ///
93    /// `None` means this is the last page.
94    pub continuation: Option<TaskContinuation>,
95    /// Number of matching items after this page in the same snapshot.
96    pub remaining_item_count: usize,
97}
98
99impl TaskContinuation {
100    /// Creates a domain continuation.
101    ///
102    /// The resource version remains opaque here.
103    /// The state store validates that it belongs to a retained snapshot when the query is executed.
104    ///
105    /// # Errors
106    ///
107    /// Returns [`ModelError::Invalid`] when `resource_version` is empty or whitespace.
108    pub fn new(
109        resource_version: impl Into<String>,
110        filter: TaskFilter,
111        after: TaskId,
112    ) -> ModelResult<Self> {
113        let resource_version = resource_version.into();
114        if resource_version.trim().is_empty() {
115            return Err(ModelError::Invalid(
116                "continuation resourceVersion must not be empty".into(),
117            ));
118        }
119        Ok(Self {
120            resource_version,
121            filter,
122            after,
123        })
124    }
125
126    /// Collection snapshot version carried by this cursor.
127    pub fn resource_version(&self) -> &str {
128        &self.resource_version
129    }
130
131    /// Filters fixed by the first page.
132    pub fn filter(&self) -> &TaskFilter {
133        &self.filter
134    }
135
136    /// Last Task name returned before this cursor.
137    pub fn after(&self) -> &TaskId {
138        &self.after
139    }
140}
141
142mod raw {
143    use super::*;
144
145    #[derive(Deserialize)]
146    #[serde(rename_all = "camelCase", deny_unknown_fields)]
147    pub(super) struct TaskFilterRaw {
148        #[serde(default)]
149        phases: Vec<TaskPhase>,
150        #[serde(default)]
151        slot: Option<Slot>,
152        #[serde(default)]
153        label_selector: LabelSelector,
154    }
155
156    impl TryFrom<TaskFilterRaw> for TaskFilter {
157        type Error = ModelError;
158
159        fn try_from(raw: TaskFilterRaw) -> Result<Self, Self::Error> {
160            raw.label_selector.validate()?;
161            let mut filter = Self {
162                phases: Vec::new(),
163                slot: raw.slot,
164                label_selector: raw.label_selector,
165            };
166            for phase in raw.phases {
167                filter = filter.with_phase(phase);
168            }
169            Ok(filter)
170        }
171    }
172
173    #[derive(Deserialize)]
174    #[serde(rename_all = "camelCase", deny_unknown_fields)]
175    pub(super) struct TaskContinuationRaw {
176        resource_version: String,
177        filter: TaskFilter,
178        after: TaskId,
179    }
180
181    impl TryFrom<TaskContinuationRaw> for TaskContinuation {
182        type Error = ModelError;
183
184        fn try_from(raw: TaskContinuationRaw) -> Result<Self, Self::Error> {
185            TaskContinuation::new(raw.resource_version, raw.filter, raw.after)
186        }
187    }
188}
189
190impl TaskFilter {
191    /// Creates an empty filter.
192    #[inline]
193    pub fn new() -> Self {
194        Self::default()
195    }
196
197    /// Filter by slot name.
198    #[inline]
199    pub fn with_slot(mut self, slot: Slot) -> Self {
200        self.slot = Some(slot);
201        self
202    }
203
204    /// Adds a phase filter.
205    ///
206    /// Multiple calls accumulate with OR semantics.
207    #[inline]
208    pub fn with_phase(mut self, phase: TaskPhase) -> Self {
209        if !self.phases.contains(&phase) {
210            self.phases.push(phase);
211        }
212        self
213    }
214
215    /// Adds phase filters from an iterator.
216    ///
217    /// Values are deduplicated and retain OR semantics.
218    pub fn with_phases(mut self, phases: impl IntoIterator<Item = TaskPhase>) -> Self {
219        for phase in phases {
220            self = self.with_phase(phase);
221        }
222        self
223    }
224
225    /// Filter by labels.
226    ///
227    /// Every selector requirement is ANDed. An empty selector matches all tasks.
228    ///
229    /// # Errors
230    ///
231    /// Returns [`ModelError::Invalid`] when the selector is invalid.
232    #[inline]
233    pub fn with_label_selector(mut self, selector: LabelSelector) -> ModelResult<Self> {
234        selector.validate()?;
235        self.label_selector = selector;
236        Ok(self)
237    }
238
239    /// Filter by all active phases: `Pending` and `Running`.
240    #[inline]
241    pub fn with_active(self) -> Self {
242        self.with_phase(TaskPhase::Pending)
243            .with_phase(TaskPhase::Running)
244    }
245
246    /// Filter by all terminal phases.
247    #[inline]
248    pub fn with_terminal(self) -> Self {
249        self.with_phase(TaskPhase::Succeeded)
250            .with_phase(TaskPhase::Exhausted)
251            .with_phase(TaskPhase::Canceled)
252            .with_phase(TaskPhase::Timeout)
253            .with_phase(TaskPhase::Failed)
254    }
255
256    /// Returns whether a task passes every filter.
257    #[inline]
258    pub fn matches(&self, task: &Task) -> bool {
259        self.slot.as_ref().is_none_or(|slot| slot == task.slot())
260            && self.matches_phase(task.phase())
261            && self.matches_labels(task.labels())
262    }
263
264    /// Returns whether a phase passes the phase filter.
265    ///
266    /// An empty filter matches all phases.
267    #[inline]
268    pub fn matches_phase(&self, phase: &TaskPhase) -> bool {
269        self.phases.is_empty() || self.phases.contains(phase)
270    }
271
272    /// Returns whether labels pass the selector.
273    #[inline]
274    pub fn matches_labels(&self, labels: &Labels) -> bool {
275        self.label_selector.matches(labels)
276    }
277
278    /// Slot filter (if any).
279    #[inline]
280    pub fn slot(&self) -> Option<&Slot> {
281        self.slot.as_ref()
282    }
283
284    /// Phase filters.
285    #[inline]
286    pub fn phases(&self) -> &[TaskPhase] {
287        &self.phases
288    }
289
290    /// Label selector.
291    #[inline]
292    pub fn label_selector(&self) -> &LabelSelector {
293        &self.label_selector
294    }
295}
296
297impl TaskQuery {
298    /// Creates an unfiltered query with default pagination.
299    ///
300    /// ## Example
301    ///
302    /// ```
303    /// use solti_model::{DEFAULT_LIMIT, TaskPhase, TaskQuery};
304    ///
305    /// let query = TaskQuery::new();
306    /// assert_eq!(query.limit(), DEFAULT_LIMIT);
307    /// assert!(query.matches_phase(&TaskPhase::Failed));
308    /// ```
309    #[inline]
310    pub fn new() -> Self {
311        Self::from_filter(TaskFilter::new())
312    }
313
314    /// Creates a query from filters.
315    #[inline]
316    pub fn from_filter(filter: TaskFilter) -> Self {
317        Self {
318            filter,
319            limit: DEFAULT_LIMIT,
320            continuation: None,
321        }
322    }
323
324    /// Filter by slot name.
325    #[inline]
326    pub fn with_slot(mut self, slot: Slot) -> Self {
327        self.filter = self.filter.with_slot(slot);
328        self
329    }
330
331    /// Adds a phase filter.
332    ///
333    /// Multiple calls accumulate with OR semantics.
334    #[inline]
335    pub fn with_phase(mut self, phase: TaskPhase) -> Self {
336        self.filter = self.filter.with_phase(phase);
337        self
338    }
339
340    /// Adds phase filters from an iterator.
341    ///
342    /// Values are deduplicated and retain OR semantics.
343    #[inline]
344    pub fn with_phases(mut self, phases: impl IntoIterator<Item = TaskPhase>) -> Self {
345        self.filter = self.filter.with_phases(phases);
346        self
347    }
348
349    /// Filter by labels.
350    ///
351    /// Every selector requirement is ANDed. An empty selector matches all tasks.
352    ///
353    /// # Errors
354    ///
355    /// Returns [`ModelError::Invalid`] when the selector is invalid.
356    #[inline]
357    pub fn with_label_selector(mut self, selector: LabelSelector) -> ModelResult<Self> {
358        self.filter = self.filter.with_label_selector(selector)?;
359        Ok(self)
360    }
361
362    /// Filter by all active phases: `Pending` and `Running`.
363    #[inline]
364    pub fn with_active(mut self) -> Self {
365        self.filter = self.filter.with_active();
366        self
367    }
368
369    /// Filter by all terminal phases.
370    #[inline]
371    pub fn with_terminal(mut self) -> Self {
372        self.filter = self.filter.with_terminal();
373        self
374    }
375
376    /// Sets the page size.
377    ///
378    /// Zero selects [`DEFAULT_LIMIT`]. Values above [`MAX_LIMIT`] are capped.
379    #[inline]
380    pub fn with_limit(mut self, limit: usize) -> Self {
381        self.limit = if limit == 0 {
382            DEFAULT_LIMIT
383        } else {
384            limit.min(MAX_LIMIT)
385        };
386        self
387    }
388
389    /// Continue a previously returned collection snapshot.
390    #[inline]
391    pub fn with_continuation(mut self, continuation: TaskContinuation) -> Self {
392        self.continuation = Some(continuation);
393        self
394    }
395
396    /// Returns whether a task passes every filter.
397    #[inline]
398    pub fn matches(&self, task: &Task) -> bool {
399        self.filter.matches(task)
400    }
401
402    /// Returns whether a phase passes the phase filter.
403    #[inline]
404    pub fn matches_phase(&self, phase: &TaskPhase) -> bool {
405        self.filter.matches_phase(phase)
406    }
407
408    /// Returns whether labels pass the selector.
409    #[inline]
410    pub fn matches_labels(&self, labels: &Labels) -> bool {
411        self.filter.matches_labels(labels)
412    }
413
414    /// Page size limit.
415    #[inline]
416    pub fn limit(&self) -> usize {
417        self.limit
418    }
419
420    /// Continuation cursor, when this is not the first page.
421    #[inline]
422    pub fn continuation(&self) -> Option<&TaskContinuation> {
423        self.continuation.as_ref()
424    }
425
426    /// Filters applied before pagination.
427    #[inline]
428    pub fn filter(&self) -> &TaskFilter {
429        &self.filter
430    }
431
432    /// Slot filter (if any).
433    #[inline]
434    pub fn slot(&self) -> Option<&Slot> {
435        self.filter.slot()
436    }
437
438    /// Phase filters.
439    #[inline]
440    pub fn phases(&self) -> &[TaskPhase] {
441        self.filter.phases()
442    }
443
444    /// Label selector.
445    #[inline]
446    pub fn label_selector(&self) -> &LabelSelector {
447        self.filter.label_selector()
448    }
449}
450
451/// One change emitted by a task watch.
452#[derive(Debug, Clone, PartialEq, Eq)]
453pub enum TaskWatchEvent {
454    /// A task entered the watched collection.
455    Added(Task),
456    /// A task already in the watched collection changed.
457    Modified(Task),
458    /// A task left the watched collection or was deleted.
459    Deleted(Task),
460}
461
462impl TaskWatchEvent {
463    /// Resource carried by this event.
464    #[inline]
465    pub fn object(&self) -> &Task {
466        match self {
467            Self::Added(task) | Self::Modified(task) | Self::Deleted(task) => task,
468        }
469    }
470
471    /// Opaque store version of this event.
472    #[inline]
473    pub fn resource_version(&self) -> &str {
474        self.object().metadata().resource_version()
475    }
476
477    /// Returns the event resource.
478    #[inline]
479    pub fn into_object(self) -> Task {
480        match self {
481            Self::Added(task) | Self::Modified(task) | Self::Deleted(task) => task,
482        }
483    }
484}
485
486#[cfg(test)]
487mod tests {
488    use super::*;
489    use crate::{EmbeddedSpec, TaskSpec, TaskWorkload};
490
491    fn labels(pairs: &[(&str, &str)]) -> Labels {
492        let mut labels = Labels::new();
493        for (key, value) in pairs {
494            labels.insert(*key, *value);
495        }
496        labels
497    }
498
499    #[test]
500    fn filters_deduplicate_phases_and_match_with_or_semantics() {
501        let query = TaskQuery::new()
502            .with_phase(TaskPhase::Pending)
503            .with_phase(TaskPhase::Running)
504            .with_phase(TaskPhase::Pending);
505
506        assert_eq!(query.phases(), &[TaskPhase::Pending, TaskPhase::Running]);
507        assert!(query.matches_phase(&TaskPhase::Pending));
508        assert!(query.matches_phase(&TaskPhase::Running));
509        assert!(!query.matches_phase(&TaskPhase::Failed));
510
511        let query = TaskQuery::new();
512        assert!(query.matches_phase(&TaskPhase::Failed));
513        assert!(query.matches_labels(&labels(&[("environment", "production")])));
514    }
515
516    #[test]
517    fn label_selector_is_applied() {
518        let query = TaskQuery::new()
519            .with_label_selector(
520                "environment=production,!tainted"
521                    .parse::<LabelSelector>()
522                    .unwrap(),
523            )
524            .unwrap();
525
526        assert!(query.matches_labels(&labels(&[("environment", "production")])));
527        assert!(!query.matches_labels(&labels(&[("environment", "development")])));
528        assert!(!query.matches_labels(&labels(&[
529            ("environment", "production"),
530            ("tainted", "true"),
531        ])));
532    }
533
534    #[test]
535    fn query_keeps_filter_separate_from_pagination() {
536        let filter = TaskFilter::new()
537            .with_slot(Slot::new("build").unwrap())
538            .with_phase(TaskPhase::Running);
539        let continuation =
540            TaskContinuation::new("store:7", filter.clone(), TaskId::new("build-50").unwrap())
541                .unwrap();
542        let query = TaskQuery::from_filter(filter.clone())
543            .with_limit(25)
544            .with_continuation(continuation.clone());
545
546        assert_eq!(query.filter(), &filter);
547        assert_eq!(query.limit(), 25);
548        assert_eq!(query.continuation(), Some(&continuation));
549        assert_eq!(continuation.resource_version(), "store:7");
550        assert_eq!(continuation.filter(), &filter);
551        assert_eq!(continuation.after().as_str(), "build-50");
552    }
553
554    #[test]
555    fn zero_limit_uses_default_and_continuation_requires_resource_version() {
556        assert_eq!(TaskQuery::new().with_limit(0).limit(), DEFAULT_LIMIT);
557        assert!(matches!(
558            TaskContinuation::new("  ", TaskFilter::new(), TaskId::new("build-50").unwrap(),),
559            Err(ModelError::Invalid(_))
560        ));
561    }
562
563    #[test]
564    fn continuation_has_a_strict_serde_roundtrip() {
565        let filter = TaskFilter::new()
566            .with_slot(Slot::new("build").unwrap())
567            .with_phase(TaskPhase::Running)
568            .with_label_selector("environment=production".parse().unwrap())
569            .unwrap();
570        let continuation =
571            TaskContinuation::new("store:7", filter, TaskId::new("build-50").unwrap()).unwrap();
572
573        let json = serde_json::to_string(&continuation).unwrap();
574        let decoded: TaskContinuation = serde_json::from_str(&json).unwrap();
575
576        assert_eq!(decoded, continuation);
577        assert!(
578            serde_json::from_str::<TaskContinuation>(
579                r#"{"resourceVersion":"","filter":{},"after":"build-50"}"#,
580            )
581            .is_err()
582        );
583        assert!(
584            serde_json::from_str::<TaskFilter>(
585                r#"{"labelSelector":{"matchExpressions":[{"key":"tier","operator":"In","values":[]}]}}"#,
586            )
587            .is_err()
588        );
589        assert!(serde_json::from_str::<TaskFilter>(r#"{"unknown":true}"#).is_err());
590    }
591
592    #[test]
593    fn watch_event_exposes_object_resource_version() {
594        let spec = TaskSpec::builder(
595            "build",
596            TaskWorkload::Embedded(EmbeddedSpec::new("v1").unwrap()),
597            1_000_u64,
598        )
599        .build()
600        .unwrap();
601        let mut task = Task::new("build-1", spec).unwrap();
602        task.set_resource_version("store:7").unwrap();
603        let event = TaskWatchEvent::Modified(task.clone());
604
605        assert_eq!(event.object(), &task);
606        assert_eq!(event.resource_version(), "store:7");
607        assert_eq!(event.into_object(), task);
608    }
609}
solti_model/domain/query/task.rs

solti_model/domain/query/
task.rs
