oximedia_workflow/
fan_pattern.rs

1//! Fan-out / fan-in parallel task execution patterns.
2//!
3//! This module provides three composable primitives:
4//!
5//! - [`FanOut`]: spawns N tasks concurrently from a single input.
6//! - `FanIn`: collects the results of N concurrent tasks into one aggregated
7//!   result.
8//! - [`FanPattern`]: combines `FanOut` + `FanIn` for the complete scatter/gather
9//!   pattern commonly used in media processing pipelines (e.g. encode to multiple
10//!   bitrates simultaneously, then merge the manifests).
11//!
12//! [`FanExecutor`] drives the async execution and result aggregation using
13//! `tokio::task::JoinSet`.
14//!
15//! # Example
16//!
17//! ```rust,no_run
18//! use oximedia_workflow::fan_pattern::{FanTask, FanExecutor, FanPattern, AggregationStrategy};
19//!
20//! # let rt = tokio::runtime::Runtime::new().unwrap();
21//! # rt.block_on(async {
22//! let tasks = vec![
23//!     FanTask::new("encode-1080p", |_| async { Ok::<i64, String>(1080) }),
24//!     FanTask::new("encode-720p",  |_| async { Ok::<i64, String>(720) }),
25//!     FanTask::new("encode-480p",  |_| async { Ok::<i64, String>(480) }),
26//! ];
27//!
28//! let pattern = FanPattern::new("multi-bitrate-encode", tasks)
29//!     .with_strategy(AggregationStrategy::CollectAll);
30//!
31//! let executor = FanExecutor::new();
32//! let result = executor.execute(pattern).await.expect("all tasks succeed");
33//! assert_eq!(result.task_results.len(), 3);
34//! # });
35//! ```
36
37use std::{
38    collections::HashMap,
39    fmt,
40    future::Future,
41    pin::Pin,
42    sync::Arc,
43    time::{Duration, Instant},
44};
45
46use tokio::task::JoinSet;
47
48// ---------------------------------------------------------------------------
49// FanTask
50// ---------------------------------------------------------------------------
51
52/// Type alias for a boxed async factory that produces a task future.
53///
54/// The factory receives the shared fan-out context (a `HashMap<String, String>`)
55/// and returns a pinned, boxed `Future` that resolves to `Result<i64, String>`.
56pub type TaskFactory = Arc<
57    dyn Fn(
58            Arc<HashMap<String, String>>,
59        ) -> Pin<Box<dyn Future<Output = Result<i64, String>> + Send>>
60        + Send
61        + Sync,
62>;
63
64/// A single unit of work in a fan-out/fan-in pattern.
65///
66/// Each `FanTask` wraps a named async closure that receives shared context
67/// from the fan-out root and returns either a numeric output or an error string.
68#[derive(Clone)]
69pub struct FanTask {
70    /// Human-readable name (used in result maps and error messages).
71    pub name: String,
72    /// Async factory invoked by the executor.
73    factory: TaskFactory,
74    /// Optional per-task timeout.  When `None`, the executor default is used.
75    pub timeout: Option<Duration>,
76    /// Task weight used by weighted-sum aggregation (default = 1).
77    pub weight: u32,
78}
79
80impl fmt::Debug for FanTask {
81    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
82        f.debug_struct("FanTask")
83            .field("name", &self.name)
84            .field("timeout", &self.timeout)
85            .field("weight", &self.weight)
86            .finish()
87    }
88}
89
90impl FanTask {
91    /// Create a `FanTask` from an async factory closure.
92    ///
93    /// The closure accepts the shared context and must return a
94    /// `Future<Output = Result<i64, String>>`.
95    pub fn new<N, F, Fut>(name: N, factory: F) -> Self
96    where
97        N: Into<String>,
98        F: Fn(Arc<HashMap<String, String>>) -> Fut + Send + Sync + 'static,
99        Fut: Future<Output = Result<i64, String>> + Send + 'static,
100    {
101        Self {
102            name: name.into(),
103            factory: Arc::new(move |ctx| Box::pin(factory(ctx))),
104            timeout: None,
105            weight: 1,
106        }
107    }
108
109    /// Set a per-task timeout.
110    #[must_use]
111    pub fn with_timeout(mut self, timeout: Duration) -> Self {
112        self.timeout = Some(timeout);
113        self
114    }
115
116    /// Set the aggregation weight for this task (used with
117    /// [`AggregationStrategy::WeightedSum`]).
118    #[must_use]
119    pub fn with_weight(mut self, weight: u32) -> Self {
120        self.weight = weight;
121        self
122    }
123
124    /// Invoke the factory and get the future.
125    fn invoke(
126        &self,
127        ctx: Arc<HashMap<String, String>>,
128    ) -> Pin<Box<dyn Future<Output = Result<i64, String>> + Send>> {
129        (self.factory)(ctx)
130    }
131}
132
133// ---------------------------------------------------------------------------
134// FanOut / FanIn
135// ---------------------------------------------------------------------------
136
137/// Distributes a single input context to N parallel tasks.
138///
139/// `FanOut` owns the initial context key-value pairs that will be passed
140/// (via `Arc`) to every spawned task.
141#[derive(Debug, Default, Clone)]
142pub struct FanOut {
143    /// Shared context forwarded to all tasks.
144    pub context: HashMap<String, String>,
145}
146
147impl FanOut {
148    /// Create an empty fan-out context.
149    #[must_use]
150    pub fn new() -> Self {
151        Self::default()
152    }
153
154    /// Insert a key-value pair into the shared context.
155    pub fn insert(&mut self, key: impl Into<String>, value: impl Into<String>) -> &mut Self {
156        self.context.insert(key.into(), value.into());
157        self
158    }
159
160    /// Builder-style context insertion.
161    #[must_use]
162    pub fn with(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
163        self.context.insert(key.into(), value.into());
164        self
165    }
166}
167
168/// Collects N parallel task outputs and aggregates them into one value.
169#[derive(Debug, Clone, Copy, PartialEq, Eq)]
170pub enum AggregationStrategy {
171    /// Gather all outputs; fail if any task fails.
172    CollectAll,
173    /// Succeed as soon as the first task succeeds; ignore subsequent failures.
174    FirstSuccess,
175    /// Sum all successful outputs; fail only if *all* tasks fail.
176    Sum,
177    /// Compute the maximum among successful outputs; fail only if all fail.
178    Max,
179    /// Compute the minimum among successful outputs; fail only if all fail.
180    Min,
181    /// Compute the weighted sum: `sum(output * weight)` for successful tasks.
182    WeightedSum,
183}
184
185impl Default for AggregationStrategy {
186    fn default() -> Self {
187        Self::CollectAll
188    }
189}
190
191/// The merged result of a completed fan-in phase.
192#[derive(Debug, Clone)]
193pub struct FanInResult {
194    /// Individual per-task results (name → output or error).
195    pub task_results: HashMap<String, Result<i64, String>>,
196    /// Aggregated scalar value computed by the chosen strategy.
197    ///
198    /// `None` when the strategy could not produce a value (e.g., all tasks
199    /// failed and `Sum` was chosen).
200    pub aggregated: Option<i64>,
201    /// Total wall-clock duration of the fan execution.
202    pub elapsed: Duration,
203    /// Number of tasks that succeeded.
204    pub success_count: usize,
205    /// Number of tasks that failed.
206    pub failure_count: usize,
207}
208
209impl FanInResult {
210    /// Returns `true` when all tasks succeeded.
211    #[must_use]
212    pub fn all_succeeded(&self) -> bool {
213        self.failure_count == 0
214    }
215
216    /// Returns `true` when at least one task succeeded.
217    #[must_use]
218    pub fn any_succeeded(&self) -> bool {
219        self.success_count > 0
220    }
221}
222
223// ---------------------------------------------------------------------------
224// FanPattern
225// ---------------------------------------------------------------------------
226
227/// Combined fan-out + fan-in pattern.
228///
229/// Describes *what* to run (tasks + context) and *how* to aggregate the results.
230/// Execution is delegated to [`FanExecutor::execute`].
231#[derive(Debug)]
232pub struct FanPattern {
233    /// Human-readable name for this pattern (used in logs and error messages).
234    pub name: String,
235    /// The tasks to execute in parallel.
236    pub tasks: Vec<FanTask>,
237    /// Shared context pushed to all tasks via fan-out.
238    pub fan_out: FanOut,
239    /// Strategy used to aggregate results in the fan-in phase.
240    pub strategy: AggregationStrategy,
241    /// Default per-task timeout applied when a task has no explicit timeout.
242    pub default_timeout: Option<Duration>,
243    /// When `true`, the executor stops dispatching further tasks as soon as
244    /// one task fails (only meaningful for `CollectAll` strategy).
245    pub fail_fast: bool,
246}
247
248impl FanPattern {
249    /// Create a new pattern with the given name and task list.
250    ///
251    /// Defaults: `CollectAll` strategy, no timeout, `fail_fast = false`.
252    #[must_use]
253    pub fn new(name: impl Into<String>, tasks: Vec<FanTask>) -> Self {
254        Self {
255            name: name.into(),
256            tasks,
257            fan_out: FanOut::new(),
258            strategy: AggregationStrategy::CollectAll,
259            default_timeout: None,
260            fail_fast: false,
261        }
262    }
263
264    /// Set the aggregation strategy.
265    #[must_use]
266    pub fn with_strategy(mut self, strategy: AggregationStrategy) -> Self {
267        self.strategy = strategy;
268        self
269    }
270
271    /// Set a default per-task timeout.
272    #[must_use]
273    pub fn with_default_timeout(mut self, timeout: Duration) -> Self {
274        self.default_timeout = Some(timeout);
275        self
276    }
277
278    /// Enable fail-fast mode.
279    #[must_use]
280    pub fn with_fail_fast(mut self, fail_fast: bool) -> Self {
281        self.fail_fast = fail_fast;
282        self
283    }
284
285    /// Add a key-value pair to the shared context.
286    #[must_use]
287    pub fn with_context(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
288        self.fan_out.insert(key, value);
289        self
290    }
291
292    /// Returns the number of tasks in this pattern.
293    #[must_use]
294    pub fn task_count(&self) -> usize {
295        self.tasks.len()
296    }
297}
298
299// ---------------------------------------------------------------------------
300// FanError
301// ---------------------------------------------------------------------------
302
303/// Errors produced by fan execution.
304#[derive(Debug, thiserror::Error)]
305pub enum FanError {
306    /// All tasks failed; wraps the per-task error map.
307    #[error("All {count} fan tasks failed")]
308    AllFailed {
309        /// Total number of tasks.
310        count: usize,
311        /// Per-task error strings.
312        errors: HashMap<String, String>,
313    },
314    /// A subset of tasks failed under `CollectAll` strategy.
315    #[error("{failure_count} of {total} fan tasks failed")]
316    SomeFailed {
317        /// Number of failures.
318        failure_count: usize,
319        /// Total number of tasks.
320        total: usize,
321        /// Per-task error strings for failed tasks.
322        errors: HashMap<String, String>,
323    },
324    /// A task exceeded its timeout.
325    #[error("Fan task '{name}' timed out after {timeout_ms}ms")]
326    TaskTimeout {
327        /// Name of the timed-out task.
328        name: String,
329        /// Configured timeout in milliseconds.
330        timeout_ms: u64,
331    },
332    /// An underlying tokio join error (task panicked).
333    #[error("Fan task '{name}' panicked: {reason}")]
334    JoinError {
335        /// Task name.
336        name: String,
337        /// Panic message.
338        reason: String,
339    },
340}
341
342// ---------------------------------------------------------------------------
343// FanExecutor
344// ---------------------------------------------------------------------------
345
346/// Drives concurrent fan-out / fan-in execution.
347///
348/// `FanExecutor` is stateless and can be reused across multiple
349/// [`FanPattern`] executions.
350#[derive(Debug, Default, Clone)]
351pub struct FanExecutor {
352    /// Global default timeout applied when neither the task nor the pattern
353    /// specifies one.
354    pub global_timeout: Option<Duration>,
355}
356
357impl FanExecutor {
358    /// Create a new executor with no global timeout.
359    #[must_use]
360    pub fn new() -> Self {
361        Self::default()
362    }
363
364    /// Set a global default timeout.
365    #[must_use]
366    pub fn with_global_timeout(mut self, timeout: Duration) -> Self {
367        self.global_timeout = Some(timeout);
368        self
369    }
370
371    /// Execute a [`FanPattern`] and return the aggregated [`FanInResult`].
372    ///
373    /// # Errors
374    ///
375    /// Returns [`FanError::AllFailed`] when every task fails and the strategy
376    /// requires at least one success.  Returns [`FanError::SomeFailed`] under
377    /// `CollectAll` when any task fails.  Returns [`FanError::TaskTimeout`]
378    /// when a task exceeds its timeout.
379    pub async fn execute(&self, pattern: FanPattern) -> Result<FanInResult, FanError> {
380        let start = Instant::now();
381        let ctx = Arc::new(pattern.fan_out.context.clone());
382        let strategy = pattern.strategy;
383
384        // Spawn all tasks concurrently via JoinSet.
385        let mut join_set: JoinSet<(String, u32, Result<i64, String>)> = JoinSet::new();
386
387        for task in pattern.tasks {
388            let task_name = task.name.clone();
389            let task_weight = task.weight;
390            let task_ctx = ctx.clone();
391            let effective_timeout = task
392                .timeout
393                .or(pattern.default_timeout)
394                .or(self.global_timeout);
395            let fut = task.invoke(task_ctx);
396
397            join_set.spawn(async move {
398                match effective_timeout {
399                    Some(timeout) => match tokio::time::timeout(timeout, fut).await {
400                        Ok(result) => (task_name, task_weight, result),
401                        Err(_elapsed) => (
402                            task_name.clone(),
403                            task_weight,
404                            Err(format!("task '{}' timed out", task_name)),
405                        ),
406                    },
407                    None => {
408                        let result = fut.await;
409                        (task_name, task_weight, result)
410                    }
411                }
412            });
413        }
414
415        // Collect all results.
416        let mut task_results: HashMap<String, Result<i64, String>> = HashMap::new();
417        let mut weights: HashMap<String, u32> = HashMap::new();
418        let mut first_success: Option<(String, i64)> = None;
419        let mut errors: HashMap<String, String> = HashMap::new();
420
421        while let Some(join_result) = join_set.join_next().await {
422            match join_result {
423                Ok((name, weight, outcome)) => {
424                    weights.insert(name.clone(), weight);
425                    match &outcome {
426                        Ok(val) => {
427                            if first_success.is_none() {
428                                first_success = Some((name.clone(), *val));
429                            }
430                        }
431                        Err(e) => {
432                            errors.insert(name.clone(), e.clone());
433                        }
434                    }
435                    task_results.insert(name, outcome);
436                }
437                Err(join_err) => {
438                    // tokio JoinError means the task panicked.
439                    let name = format!("<unknown-task-{}>", errors.len());
440                    let reason = join_err.to_string();
441                    errors.insert(name.clone(), reason.clone());
442                    task_results.insert(name, Err(format!("join error: {reason}")));
443                }
444            }
445        }
446
447        let elapsed = start.elapsed();
448        let total = task_results.len();
449        let failure_count = errors.len();
450        let success_count = total - failure_count;
451
452        // Aggregate according to strategy.
453        let aggregated = Self::aggregate(&task_results, &weights, strategy, &first_success);
454
455        // Check for errors based on strategy.
456        match strategy {
457            AggregationStrategy::CollectAll => {
458                if failure_count > 0 {
459                    return Err(FanError::SomeFailed {
460                        failure_count,
461                        total,
462                        errors,
463                    });
464                }
465            }
466            AggregationStrategy::FirstSuccess => {
467                if first_success.is_none() {
468                    return Err(FanError::AllFailed {
469                        count: total,
470                        errors,
471                    });
472                }
473            }
474            AggregationStrategy::Sum
475            | AggregationStrategy::Max
476            | AggregationStrategy::Min
477            | AggregationStrategy::WeightedSum => {
478                if success_count == 0 {
479                    return Err(FanError::AllFailed {
480                        count: total,
481                        errors,
482                    });
483                }
484            }
485        }
486
487        Ok(FanInResult {
488            task_results,
489            aggregated,
490            elapsed,
491            success_count,
492            failure_count,
493        })
494    }
495
496    fn aggregate(
497        task_results: &HashMap<String, Result<i64, String>>,
498        weights: &HashMap<String, u32>,
499        strategy: AggregationStrategy,
500        first_success: &Option<(String, i64)>,
501    ) -> Option<i64> {
502        let successes: Vec<(i64, u32)> = task_results
503            .iter()
504            .filter_map(|(name, r)| {
505                r.as_ref().ok().map(|&v| {
506                    let w = weights.get(name).copied().unwrap_or(1);
507                    (v, w)
508                })
509            })
510            .collect();
511
512        match strategy {
513            AggregationStrategy::CollectAll => {
514                // Sum all as a convenience scalar when all succeed.
515                if task_results.values().all(|r| r.is_ok()) {
516                    Some(successes.iter().map(|(v, _)| *v).sum())
517                } else {
518                    None
519                }
520            }
521            AggregationStrategy::FirstSuccess => first_success.as_ref().map(|(_, v)| *v),
522            AggregationStrategy::Sum => {
523                if successes.is_empty() {
524                    None
525                } else {
526                    Some(successes.iter().map(|(v, _)| *v).sum())
527                }
528            }
529            AggregationStrategy::Max => successes.iter().map(|(v, _)| *v).max(),
530            AggregationStrategy::Min => successes.iter().map(|(v, _)| *v).min(),
531            AggregationStrategy::WeightedSum => {
532                if successes.is_empty() {
533                    None
534                } else {
535                    Some(
536                        successes
537                            .iter()
538                            .map(|(v, w)| v.saturating_mul(i64::from(*w)))
539                            .fold(0i64, i64::saturating_add),
540                    )
541                }
542            }
543        }
544    }
545}
546
547// ---------------------------------------------------------------------------
548// Tests
549// ---------------------------------------------------------------------------
550
551#[cfg(test)]
552mod tests {
553    use super::*;
554
555    fn make_ok_task(name: &str, value: i64) -> FanTask {
556        FanTask::new(name.to_string(), move |_ctx| async move {
557            Ok::<i64, String>(value)
558        })
559    }
560
561    fn make_fail_task(name: &str, reason: &str) -> FanTask {
562        let reason = reason.to_string();
563        FanTask::new(name.to_string(), move |_ctx| {
564            let r = reason.clone();
565            async move { Err::<i64, String>(r) }
566        })
567    }
568
569    // -----------------------------------------------------------------------
570    // FanOut context
571    // -----------------------------------------------------------------------
572
573    #[test]
574    fn fan_out_insert_and_with() {
575        let mut fo = FanOut::new();
576        fo.insert("k1", "v1");
577        let fo = fo.with("k2".to_string(), "v2".to_string());
578        assert_eq!(fo.context.get("k1").map(String::as_str), Some("v1"));
579        assert_eq!(fo.context.get("k2").map(String::as_str), Some("v2"));
580    }
581
582    // -----------------------------------------------------------------------
583    // FanTask
584    // -----------------------------------------------------------------------
585
586    #[test]
587    fn fan_task_weight_default_is_one() {
588        let t = make_ok_task("t", 42);
589        assert_eq!(t.weight, 1);
590    }
591
592    #[test]
593    fn fan_task_with_weight() {
594        let t = make_ok_task("t", 0).with_weight(5);
595        assert_eq!(t.weight, 5);
596    }
597
598    #[test]
599    fn fan_task_with_timeout() {
600        let t = make_ok_task("t", 0).with_timeout(Duration::from_millis(100));
601        assert!(t.timeout.is_some());
602    }
603
604    // -----------------------------------------------------------------------
605    // FanPattern
606    // -----------------------------------------------------------------------
607
608    #[test]
609    fn fan_pattern_task_count() {
610        let p = FanPattern::new("p", vec![make_ok_task("a", 1), make_ok_task("b", 2)]);
611        assert_eq!(p.task_count(), 2);
612    }
613
614    #[test]
615    fn fan_pattern_defaults() {
616        let p = FanPattern::new("p", vec![]);
617        assert_eq!(p.strategy, AggregationStrategy::CollectAll);
618        assert!(!p.fail_fast);
619        assert!(p.default_timeout.is_none());
620    }
621
622    // -----------------------------------------------------------------------
623    // FanExecutor — happy paths
624    // -----------------------------------------------------------------------
625
626    #[tokio::test]
627    async fn execute_collect_all_success() {
628        let tasks = vec![
629            make_ok_task("a", 10),
630            make_ok_task("b", 20),
631            make_ok_task("c", 30),
632        ];
633        let pattern = FanPattern::new("test", tasks).with_strategy(AggregationStrategy::CollectAll);
634        let result = FanExecutor::new()
635            .execute(pattern)
636            .await
637            .expect("all tasks succeed");
638
639        assert_eq!(result.success_count, 3);
640        assert_eq!(result.failure_count, 0);
641        assert!(result.all_succeeded());
642        // CollectAll aggregated = sum = 60
643        assert_eq!(result.aggregated, Some(60));
644    }
645
646    #[tokio::test]
647    async fn execute_sum_strategy() {
648        let tasks = vec![
649            make_ok_task("a", 5),
650            make_ok_task("b", 15),
651            make_fail_task("c", "fail"),
652        ];
653        let pattern = FanPattern::new("test", tasks).with_strategy(AggregationStrategy::Sum);
654        let result = FanExecutor::new()
655            .execute(pattern)
656            .await
657            .expect("sum succeeds when at least one succeeds");
658
659        assert_eq!(result.success_count, 2);
660        assert_eq!(result.failure_count, 1);
661        assert_eq!(result.aggregated, Some(20));
662    }
663
664    #[tokio::test]
665    async fn execute_max_strategy() {
666        let tasks = vec![
667            make_ok_task("a", 3),
668            make_ok_task("b", 100),
669            make_ok_task("c", 7),
670        ];
671        let pattern = FanPattern::new("test", tasks).with_strategy(AggregationStrategy::Max);
672        let result = FanExecutor::new()
673            .execute(pattern)
674            .await
675            .expect("max succeeds");
676        assert_eq!(result.aggregated, Some(100));
677    }
678
679    #[tokio::test]
680    async fn execute_min_strategy() {
681        let tasks = vec![
682            make_ok_task("a", 50),
683            make_ok_task("b", 3),
684            make_ok_task("c", 25),
685        ];
686        let pattern = FanPattern::new("test", tasks).with_strategy(AggregationStrategy::Min);
687        let result = FanExecutor::new()
688            .execute(pattern)
689            .await
690            .expect("min succeeds");
691        assert_eq!(result.aggregated, Some(3));
692    }
693
694    #[tokio::test]
695    async fn execute_weighted_sum_strategy() {
696        let tasks = vec![
697            make_ok_task("a", 10).with_weight(3), // 30
698            make_ok_task("b", 5).with_weight(2),  // 10
699        ];
700        let pattern =
701            FanPattern::new("test", tasks).with_strategy(AggregationStrategy::WeightedSum);
702        let result = FanExecutor::new()
703            .execute(pattern)
704            .await
705            .expect("weighted sum ok");
706        assert_eq!(result.aggregated, Some(40));
707    }
708
709    #[tokio::test]
710    async fn execute_first_success_strategy() {
711        // Mix of fail + success tasks; first_success should return a value.
712        let tasks = vec![
713            make_fail_task("a", "oops"),
714            make_ok_task("b", 77),
715            make_ok_task("c", 99),
716        ];
717        let pattern =
718            FanPattern::new("test", tasks).with_strategy(AggregationStrategy::FirstSuccess);
719        let result = FanExecutor::new()
720            .execute(pattern)
721            .await
722            .expect("first_success finds at least one success");
723        assert!(result.any_succeeded());
724        assert!(result.aggregated.is_some());
725    }
726
727    #[tokio::test]
728    async fn execute_collect_all_fails_on_any_failure() {
729        let tasks = vec![
730            make_ok_task("a", 1),
731            make_fail_task("b", "boom"),
732            make_ok_task("c", 3),
733        ];
734        let pattern = FanPattern::new("test", tasks).with_strategy(AggregationStrategy::CollectAll);
735        let err = FanExecutor::new()
736            .execute(pattern)
737            .await
738            .expect_err("should fail");
739
740        assert!(matches!(
741            err,
742            FanError::SomeFailed {
743                failure_count: 1,
744                total: 3,
745                ..
746            }
747        ));
748    }
749
750    #[tokio::test]
751    async fn execute_all_fail_returns_all_failed() {
752        let tasks = vec![make_fail_task("a", "err-a"), make_fail_task("b", "err-b")];
753        let pattern = FanPattern::new("test", tasks).with_strategy(AggregationStrategy::Sum);
754        let err = FanExecutor::new()
755            .execute(pattern)
756            .await
757            .expect_err("all fail");
758        assert!(matches!(err, FanError::AllFailed { count: 2, .. }));
759    }
760
761    #[tokio::test]
762    async fn execute_with_context_passed_to_tasks() {
763        let tasks = vec![FanTask::new(
764            "ctx-reader",
765            |ctx: Arc<HashMap<String, String>>| async move {
766                let val: i64 = ctx
767                    .get("multiplier")
768                    .and_then(|v| v.parse().ok())
769                    .unwrap_or(1);
770                Ok::<i64, String>(val * 10)
771            },
772        )];
773        let pattern = FanPattern::new("ctx-test", tasks)
774            .with_context("multiplier", "7")
775            .with_strategy(AggregationStrategy::CollectAll);
776
777        let result = FanExecutor::new().execute(pattern).await.expect("ok");
778        let output = result.task_results["ctx-reader"].as_ref().expect("ok");
779        assert_eq!(*output, 70);
780    }
781
782    #[tokio::test]
783    async fn execute_empty_tasks_collect_all() {
784        let pattern =
785            FanPattern::new("empty", vec![]).with_strategy(AggregationStrategy::CollectAll);
786        let result = FanExecutor::new()
787            .execute(pattern)
788            .await
789            .expect("empty is fine");
790        assert_eq!(result.success_count, 0);
791        assert_eq!(result.failure_count, 0);
792    }
793}
oximedia_workflow/fan_pattern.rs

oximedia_workflow/
fan_pattern.rs
