celers_core/
result_groups.rs

1//! Grouped result handling for Celery `group` and `chord` primitives.
2//!
3//! A [`ResultGroup`] aggregates the result metadata of a set of child tasks
4//! launched together (a `group`, or the header of a `chord`). It answers the
5//! questions a caller typically asks about a group of results:
6//!
7//! * Are **all** children ready (terminal)?
8//! * Did the group **succeed** (all children succeeded) or **fail** (at least
9//!   one child failed/was revoked/rejected)?
10//! * What are the **ordered collected values**, in the order the children were
11//!   registered?
12//!
13//! Everything here is a pure, synchronous, in-memory data structure built on
14//! top of the existing [`crate::result::TaskResultValue`] and [`crate::TaskId`]
15//! types. It performs no I/O — callers populate the child metas (e.g. from a
16//! [`crate::result::ResultStore`]) and then query the rollups.
17//!
18//! # Example
19//!
20//! ```rust
21//! use celers_core::result::TaskResultValue;
22//! use celers_core::result_groups::{ResultGroup, GroupStatus};
23//! use serde_json::json;
24//! use uuid::Uuid;
25//!
26//! let group_id = Uuid::new_v4();
27//! let a = Uuid::new_v4();
28//! let b = Uuid::new_v4();
29//!
30//! let mut group = ResultGroup::new(group_id, vec![a, b]);
31//! group.set_result(a, TaskResultValue::Success(json!(1)));
32//! assert!(!group.ready());            // b not done yet
33//! assert_eq!(group.status(), GroupStatus::Pending);
34//!
35//! group.set_result(b, TaskResultValue::Success(json!(2)));
36//! assert!(group.ready());
37//! assert!(group.successful());
38//! assert_eq!(group.status(), GroupStatus::Succeeded);
39//!
40//! // Ordered collected values match the registration order [a, b].
41//! let values = group.collected_values().unwrap();
42//! assert_eq!(values, vec![json!(1), json!(2)]);
43//! ```
44
45use crate::result::TaskResultValue;
46use crate::TaskId;
47use serde_json::Value;
48use std::collections::HashMap;
49
50/// Per-child slot inside a [`ResultGroup`].
51///
52/// Records the child's `task_id` and its latest known result value (if any).
53#[derive(Debug, Clone)]
54pub struct GroupChild {
55    /// The child task identifier.
56    pub task_id: TaskId,
57
58    /// The latest known result for this child, or `None` if not yet reported.
59    pub result: Option<TaskResultValue>,
60}
61
62impl GroupChild {
63    /// Create a new, unreported child slot.
64    #[must_use]
65    pub const fn new(task_id: TaskId) -> Self {
66        Self {
67            task_id,
68            result: None,
69        }
70    }
71
72    /// Create a child slot pre-populated with a result.
73    #[must_use]
74    pub const fn with_result(task_id: TaskId, result: TaskResultValue) -> Self {
75        Self {
76            task_id,
77            result: Some(result),
78        }
79    }
80
81    /// Returns `true` if the child has reported a terminal result.
82    #[inline]
83    #[must_use]
84    pub fn is_ready(&self) -> bool {
85        self.result.as_ref().is_some_and(TaskResultValue::is_ready)
86    }
87
88    /// Returns `true` if the child reported a successful result.
89    #[inline]
90    #[must_use]
91    pub fn is_successful(&self) -> bool {
92        self.result
93            .as_ref()
94            .is_some_and(TaskResultValue::is_successful)
95    }
96
97    /// Returns `true` if the child reported a failed/rejected result.
98    #[inline]
99    #[must_use]
100    pub fn is_failed(&self) -> bool {
101        self.result.as_ref().is_some_and(TaskResultValue::is_failed)
102    }
103
104    /// Returns `true` if the child reported a revoked result.
105    #[inline]
106    #[must_use]
107    pub fn is_revoked(&self) -> bool {
108        matches!(self.result, Some(TaskResultValue::Revoked))
109    }
110}
111
112/// Aggregate readiness/outcome status of a [`ResultGroup`].
113#[derive(Debug, Clone, Copy, PartialEq, Eq)]
114pub enum GroupStatus {
115    /// At least one child is still not in a terminal state.
116    Pending,
117
118    /// All children are terminal and every child succeeded.
119    Succeeded,
120
121    /// All children are terminal but at least one did not succeed
122    /// (failed, revoked, or rejected).
123    Failed,
124}
125
126impl GroupStatus {
127    /// Returns `true` if the group has finished (every child is terminal).
128    #[inline]
129    #[must_use]
130    pub const fn is_ready(&self) -> bool {
131        matches!(self, GroupStatus::Succeeded | GroupStatus::Failed)
132    }
133
134    /// Returns `true` if the group finished successfully.
135    #[inline]
136    #[must_use]
137    pub const fn is_successful(&self) -> bool {
138        matches!(self, GroupStatus::Succeeded)
139    }
140
141    /// Returns `true` if the group finished with at least one non-success.
142    #[inline]
143    #[must_use]
144    pub const fn is_failed(&self) -> bool {
145        matches!(self, GroupStatus::Failed)
146    }
147}
148
149/// An aggregated set of child task results for a `group` or `chord` header.
150///
151/// The group preserves child registration order so that
152/// [`ResultGroup::collected_values`] and [`ResultGroup::results`] return
153/// values in a deterministic, caller-meaningful order.
154#[derive(Debug, Clone)]
155pub struct ResultGroup {
156    /// Identifier of the group / chord this aggregate represents.
157    group_id: TaskId,
158
159    /// Child task ids in registration order. This is the canonical ordering.
160    order: Vec<TaskId>,
161
162    /// Latest known result per child (sparse until children report).
163    results: HashMap<TaskId, TaskResultValue>,
164}
165
166impl ResultGroup {
167    /// Create a new result group for `group_id` with the given child task ids,
168    /// in order. Duplicate ids are de-duplicated while preserving the first
169    /// occurrence's position.
170    #[must_use]
171    pub fn new(group_id: TaskId, children: Vec<TaskId>) -> Self {
172        let mut order = Vec::with_capacity(children.len());
173        for child in children {
174            if !order.contains(&child) {
175                order.push(child);
176            }
177        }
178        Self {
179            group_id,
180            order,
181            results: HashMap::new(),
182        }
183    }
184
185    /// Create an empty result group with no children yet registered.
186    #[must_use]
187    pub fn empty(group_id: TaskId) -> Self {
188        Self {
189            group_id,
190            order: Vec::new(),
191            results: HashMap::new(),
192        }
193    }
194
195    /// The group / chord identifier.
196    #[inline]
197    #[must_use]
198    pub fn group_id(&self) -> TaskId {
199        self.group_id
200    }
201
202    /// The child task ids in registration order.
203    #[inline]
204    #[must_use]
205    pub fn child_ids(&self) -> &[TaskId] {
206        &self.order
207    }
208
209    /// Number of registered children.
210    #[inline]
211    #[must_use]
212    pub fn len(&self) -> usize {
213        self.order.len()
214    }
215
216    /// Returns `true` if no children are registered.
217    #[inline]
218    #[must_use]
219    pub fn is_empty(&self) -> bool {
220        self.order.is_empty()
221    }
222
223    /// Register an additional child task id, appended to the end of the order.
224    ///
225    /// No-op if the child is already registered (order is preserved).
226    pub fn add_child(&mut self, task_id: TaskId) {
227        if !self.order.contains(&task_id) {
228            self.order.push(task_id);
229        }
230    }
231
232    /// Register an additional child together with its result.
233    pub fn add_child_with_result(&mut self, task_id: TaskId, result: TaskResultValue) {
234        self.add_child(task_id);
235        self.results.insert(task_id, result);
236    }
237
238    /// Set (or replace) the latest result for a child.
239    ///
240    /// If the child was not previously registered it is appended to the
241    /// ordering so its value still participates in collection.
242    pub fn set_result(&mut self, task_id: TaskId, result: TaskResultValue) {
243        self.add_child(task_id);
244        self.results.insert(task_id, result);
245    }
246
247    /// Borrow the latest result for a specific child, if reported.
248    #[must_use]
249    pub fn result_for(&self, task_id: TaskId) -> Option<&TaskResultValue> {
250        self.results.get(&task_id)
251    }
252
253    /// Build a snapshot of all children (with their results) in order.
254    #[must_use]
255    pub fn children(&self) -> Vec<GroupChild> {
256        self.order
257            .iter()
258            .map(|id| GroupChild {
259                task_id: *id,
260                result: self.results.get(id).cloned(),
261            })
262            .collect()
263    }
264
265    /// Number of children that have reported any result (terminal or not).
266    #[must_use]
267    pub fn reported_count(&self) -> usize {
268        self.results.len()
269    }
270
271    /// Number of children that have reached a terminal/ready state.
272    #[must_use]
273    pub fn completed_count(&self) -> usize {
274        self.order
275            .iter()
276            .filter(|id| self.results.get(id).is_some_and(TaskResultValue::is_ready))
277            .count()
278    }
279
280    /// Number of children that succeeded.
281    #[must_use]
282    pub fn successful_count(&self) -> usize {
283        self.order
284            .iter()
285            .filter(|id| {
286                self.results
287                    .get(id)
288                    .is_some_and(TaskResultValue::is_successful)
289            })
290            .count()
291    }
292
293    /// Number of children that failed or were rejected.
294    #[must_use]
295    pub fn failed_count(&self) -> usize {
296        self.order
297            .iter()
298            .filter(|id| self.results.get(id).is_some_and(TaskResultValue::is_failed))
299            .count()
300    }
301
302    /// Number of children that were revoked.
303    #[must_use]
304    pub fn revoked_count(&self) -> usize {
305        self.order
306            .iter()
307            .filter(|id| matches!(self.results.get(id), Some(TaskResultValue::Revoked)))
308            .count()
309    }
310
311    /// Fraction of children that are complete, in `[0.0, 1.0]`.
312    ///
313    /// An empty group is considered fully complete (`1.0`).
314    #[must_use]
315    pub fn completion_ratio(&self) -> f64 {
316        if self.order.is_empty() {
317            return 1.0;
318        }
319        self.completed_count() as f64 / self.order.len() as f64
320    }
321
322    /// Returns `true` if **all** registered children are terminal (ready).
323    ///
324    /// An empty group is trivially ready.
325    #[must_use]
326    pub fn ready(&self) -> bool {
327        self.order
328            .iter()
329            .all(|id| self.results.get(id).is_some_and(TaskResultValue::is_ready))
330    }
331
332    /// Returns `true` if the group is ready and every child succeeded.
333    #[must_use]
334    pub fn successful(&self) -> bool {
335        self.ready()
336            && self.order.iter().all(|id| {
337                self.results
338                    .get(id)
339                    .is_some_and(TaskResultValue::is_successful)
340            })
341    }
342
343    /// Returns `true` if the group is ready and at least one child did not
344    /// succeed (failed, revoked, or rejected).
345    #[must_use]
346    pub fn failed(&self) -> bool {
347        self.ready() && !self.successful()
348    }
349
350    /// Compute the aggregate [`GroupStatus`] (readiness + success/failure
351    /// rollup).
352    #[must_use]
353    pub fn status(&self) -> GroupStatus {
354        if !self.ready() {
355            GroupStatus::Pending
356        } else if self.successful() {
357            GroupStatus::Succeeded
358        } else {
359            GroupStatus::Failed
360        }
361    }
362
363    /// The ordered list of latest results, one entry per registered child in
364    /// registration order. Children that have not reported are `None`.
365    #[must_use]
366    pub fn results(&self) -> Vec<Option<TaskResultValue>> {
367        self.order
368            .iter()
369            .map(|id| self.results.get(id).cloned())
370            .collect()
371    }
372
373    /// The ordered collected success values.
374    ///
375    /// * Returns `Some(values)` (in registration order) only if the group is
376    ///   ready *and* every child succeeded.
377    /// * Returns `None` if the group is not yet ready or any child did not
378    ///   produce a success value — mirroring the semantics of a `chord`
379    ///   header, which only fires when all members succeed.
380    #[must_use]
381    pub fn collected_values(&self) -> Option<Vec<Value>> {
382        if !self.ready() {
383            return None;
384        }
385        let mut values = Vec::with_capacity(self.order.len());
386        for id in &self.order {
387            match self
388                .results
389                .get(id)
390                .and_then(TaskResultValue::success_value)
391            {
392                Some(value) => values.push(value.clone()),
393                None => return None,
394            }
395        }
396        Some(values)
397    }
398
399    /// Collect the success values of the children that *did* succeed, ignoring
400    /// the rest, in registration order. Unlike [`Self::collected_values`] this
401    /// never returns `None` and is useful for partial / best-effort rollups.
402    #[must_use]
403    pub fn successful_values(&self) -> Vec<Value> {
404        self.order
405            .iter()
406            .filter_map(|id| {
407                self.results
408                    .get(id)
409                    .and_then(TaskResultValue::success_value)
410                    .cloned()
411            })
412            .collect()
413    }
414
415    /// Collect the error messages of all non-successful children, in
416    /// registration order.
417    #[must_use]
418    pub fn error_messages(&self) -> Vec<String> {
419        self.order
420            .iter()
421            .filter_map(|id| {
422                self.results
423                    .get(id)
424                    .and_then(TaskResultValue::error_message)
425                    .map(String::from)
426            })
427            .collect()
428    }
429}
430
431#[cfg(test)]
432mod tests {
433    use super::*;
434    use serde_json::json;
435    use uuid::Uuid;
436
437    fn ids(n: usize) -> Vec<TaskId> {
438        (0..n).map(|_| Uuid::new_v4()).collect()
439    }
440
441    #[test]
442    fn empty_group_is_ready_and_successful() {
443        let group = ResultGroup::empty(Uuid::new_v4());
444        assert!(group.is_empty());
445        assert!(group.ready());
446        assert!(group.successful());
447        assert!(!group.failed());
448        assert_eq!(group.status(), GroupStatus::Succeeded);
449        assert_eq!(group.collected_values(), Some(vec![]));
450        assert_eq!(group.completion_ratio(), 1.0);
451    }
452
453    #[test]
454    fn readiness_requires_all_children() {
455        let v = ids(3);
456        let mut group = ResultGroup::new(Uuid::new_v4(), v.clone());
457        assert!(!group.ready());
458        assert_eq!(group.status(), GroupStatus::Pending);
459
460        group.set_result(v[0], TaskResultValue::Success(json!(10)));
461        group.set_result(v[1], TaskResultValue::Success(json!(20)));
462        assert!(!group.ready());
463        assert_eq!(group.completed_count(), 2);
464        assert_eq!(group.completion_ratio(), 2.0 / 3.0);
465
466        group.set_result(v[2], TaskResultValue::Success(json!(30)));
467        assert!(group.ready());
468        assert_eq!(group.completed_count(), 3);
469    }
470
471    #[test]
472    fn non_terminal_result_does_not_count_as_ready() {
473        let v = ids(2);
474        let mut group = ResultGroup::new(Uuid::new_v4(), v.clone());
475        group.set_result(v[0], TaskResultValue::Success(json!(1)));
476        // Started is non-terminal.
477        group.set_result(v[1], TaskResultValue::Started);
478        assert!(!group.ready());
479        assert_eq!(group.completed_count(), 1);
480        assert_eq!(group.reported_count(), 2);
481    }
482
483    #[test]
484    fn success_rollup_and_ordered_values() {
485        let v = ids(3);
486        let mut group = ResultGroup::new(Uuid::new_v4(), v.clone());
487        // Report out of order; ordering must follow registration order.
488        group.set_result(v[2], TaskResultValue::Success(json!("c")));
489        group.set_result(v[0], TaskResultValue::Success(json!("a")));
490        group.set_result(v[1], TaskResultValue::Success(json!("b")));
491
492        assert!(group.successful());
493        assert_eq!(group.status(), GroupStatus::Succeeded);
494        assert_eq!(group.successful_count(), 3);
495
496        let values = group.collected_values().unwrap();
497        assert_eq!(values, vec![json!("a"), json!("b"), json!("c")]);
498    }
499
500    #[test]
501    fn failure_rollup() {
502        let v = ids(3);
503        let mut group = ResultGroup::new(Uuid::new_v4(), v.clone());
504        group.set_result(v[0], TaskResultValue::Success(json!(1)));
505        group.set_result(
506            v[1],
507            TaskResultValue::Failure {
508                error: "boom".to_string(),
509                traceback: None,
510            },
511        );
512        group.set_result(v[2], TaskResultValue::Success(json!(3)));
513
514        assert!(group.ready());
515        assert!(!group.successful());
516        assert!(group.failed());
517        assert_eq!(group.status(), GroupStatus::Failed);
518        assert_eq!(group.failed_count(), 1);
519        assert_eq!(group.successful_count(), 2);
520
521        // collected_values returns None because not all succeeded.
522        assert!(group.collected_values().is_none());
523
524        // Partial collection still yields the two successes in order.
525        assert_eq!(group.successful_values(), vec![json!(1), json!(3)]);
526        assert_eq!(group.error_messages(), vec!["boom".to_string()]);
527    }
528
529    #[test]
530    fn revoked_child_marks_failure() {
531        let v = ids(2);
532        let mut group = ResultGroup::new(Uuid::new_v4(), v.clone());
533        group.set_result(v[0], TaskResultValue::Success(json!(1)));
534        group.set_result(v[1], TaskResultValue::Revoked);
535
536        assert!(group.ready());
537        assert!(group.failed());
538        assert_eq!(group.status(), GroupStatus::Failed);
539        assert_eq!(group.revoked_count(), 1);
540        assert!(group.collected_values().is_none());
541    }
542
543    #[test]
544    fn rejected_child_marks_failure() {
545        let v = ids(2);
546        let mut group = ResultGroup::new(Uuid::new_v4(), v.clone());
547        group.set_result(v[0], TaskResultValue::Success(json!(1)));
548        group.set_result(
549            v[1],
550            TaskResultValue::Rejected {
551                reason: "bad input".to_string(),
552            },
553        );
554        assert!(group.failed());
555        assert_eq!(group.failed_count(), 1);
556        assert_eq!(group.error_messages(), vec!["bad input".to_string()]);
557    }
558
559    #[test]
560    fn duplicate_children_are_deduped() {
561        let id = Uuid::new_v4();
562        let group = ResultGroup::new(Uuid::new_v4(), vec![id, id, id]);
563        assert_eq!(group.len(), 1);
564        assert_eq!(group.child_ids(), &[id]);
565    }
566
567    #[test]
568    fn set_result_for_unknown_child_registers_it() {
569        let mut group = ResultGroup::empty(Uuid::new_v4());
570        let id = Uuid::new_v4();
571        group.set_result(id, TaskResultValue::Success(json!(42)));
572        assert_eq!(group.len(), 1);
573        assert!(group.ready());
574        assert_eq!(group.collected_values(), Some(vec![json!(42)]));
575    }
576
577    #[test]
578    fn children_snapshot_matches_order() {
579        let v = ids(2);
580        let mut group = ResultGroup::new(Uuid::new_v4(), v.clone());
581        group.set_result(v[0], TaskResultValue::Success(json!(1)));
582        let children = group.children();
583        assert_eq!(children.len(), 2);
584        assert_eq!(children[0].task_id, v[0]);
585        assert!(children[0].is_ready());
586        assert!(children[0].is_successful());
587        assert_eq!(children[1].task_id, v[1]);
588        assert!(!children[1].is_ready());
589        assert!(children[1].result.is_none());
590    }
591
592    #[test]
593    fn group_status_predicates() {
594        assert!(GroupStatus::Succeeded.is_ready());
595        assert!(GroupStatus::Succeeded.is_successful());
596        assert!(!GroupStatus::Succeeded.is_failed());
597        assert!(GroupStatus::Failed.is_ready());
598        assert!(GroupStatus::Failed.is_failed());
599        assert!(!GroupStatus::Pending.is_ready());
600    }
601
602    #[test]
603    fn add_child_with_result_and_result_for() {
604        let mut group = ResultGroup::empty(Uuid::new_v4());
605        let id = Uuid::new_v4();
606        group.add_child_with_result(id, TaskResultValue::Success(json!(7)));
607        assert_eq!(
608            group
609                .result_for(id)
610                .and_then(TaskResultValue::success_value),
611            Some(&json!(7))
612        );
613        assert!(group.result_for(Uuid::new_v4()).is_none());
614    }
615}
celers_core/result_groups.rs

celers_core/
result_groups.rs
