oximedia_workflow/fanout.rs
1//! Fan-out / fan-in parallel execution pattern for OxiMedia workflows.
2//!
3//! The fan-out / fan-in pattern allows a single parent workflow step to spawn
4//! multiple child tasks in parallel and then merge the results back once a
5//! configurable completion condition is met.
6//!
7//! # Merge strategies
8//!
9//! | Strategy | Description |
10//! |-------------------|-----------------------------------------------------------|
11//! | [`MergeStrategy::WaitAll`] | Block until **every** child task completes. |
12//! | [`MergeStrategy::WaitAny`] | Unblock as soon as **any one** child completes.|
13//! | [`MergeStrategy::WaitN`] | Unblock when at least `n` children complete. |
14//!
15//! # Example
16//!
17//! ```rust
18//! use oximedia_workflow::fanout::{
19//! FanOutExecutor, FanOutGroup, FanOutTask, MergeStrategy,
20//! };
21//!
22//! let mut exec = FanOutExecutor::new();
23//!
24//! let group = FanOutGroup {
25//! parent_step: "encode-variants".to_string(),
26//! children: vec![
27//! FanOutTask { id: "t-1080p".to_string(), parent_id: "encode-variants".to_string(), payload: r#"{"profile":"1080p"}"#.to_string() },
28//! FanOutTask { id: "t-720p".to_string(), parent_id: "encode-variants".to_string(), payload: r#"{"profile":"720p"}"#.to_string() },
29//! FanOutTask { id: "t-480p".to_string(), parent_id: "encode-variants".to_string(), payload: r#"{"profile":"480p"}"#.to_string() },
30//! ],
31//! merge_strategy: MergeStrategy::WaitAll,
32//! };
33//!
34//! exec.submit_group(group);
35//! assert!(!exec.is_merge_condition_met("encode-variants"));
36//!
37//! exec.mark_complete("t-1080p");
38//! exec.mark_complete("t-720p");
39//! exec.mark_complete("t-480p");
40//! assert!(exec.is_merge_condition_met("encode-variants"));
41//! ```
42
43#![allow(dead_code)]
44
45use std::collections::{HashMap, HashSet};
46
47// ---------------------------------------------------------------------------
48// FanOutTask
49// ---------------------------------------------------------------------------
50
51/// A single child task spawned by a fan-out operation.
52#[derive(Debug, Clone, PartialEq, Eq)]
53pub struct FanOutTask {
54 /// Unique identifier for this task within the entire workflow.
55 pub id: String,
56 /// ID of the parent step that spawned this task.
57 pub parent_id: String,
58 /// Opaque payload passed to the task handler (e.g. a JSON string).
59 pub payload: String,
60}
61
62impl FanOutTask {
63 /// Create a new task with the given id, parent id, and payload.
64 #[must_use]
65 pub fn new(
66 id: impl Into<String>,
67 parent_id: impl Into<String>,
68 payload: impl Into<String>,
69 ) -> Self {
70 Self {
71 id: id.into(),
72 parent_id: parent_id.into(),
73 payload: payload.into(),
74 }
75 }
76}
77
78// ---------------------------------------------------------------------------
79// MergeStrategy
80// ---------------------------------------------------------------------------
81
82/// Determines when a fan-out group is considered "done" and execution can
83/// continue to the fan-in step.
84#[derive(Debug, Clone, PartialEq, Eq)]
85pub enum MergeStrategy {
86 /// Wait until **all** child tasks have completed.
87 WaitAll,
88 /// Unblock as soon as **any** child task completes.
89 WaitAny,
90 /// Unblock when at least `n` child tasks have completed.
91 WaitN(usize),
92}
93
94// ---------------------------------------------------------------------------
95// FanOutGroup
96// ---------------------------------------------------------------------------
97
98/// A named group of child tasks associated with a single parent step.
99#[derive(Debug, Clone)]
100pub struct FanOutGroup {
101 /// The parent step that owns this fan-out group. Used as the key in
102 /// [`FanOutExecutor`].
103 pub parent_step: String,
104 /// The child tasks to run in parallel.
105 pub children: Vec<FanOutTask>,
106 /// The strategy used to decide when the group is done.
107 pub merge_strategy: MergeStrategy,
108}
109
110// ---------------------------------------------------------------------------
111// FanOutGroupState (internal)
112// ---------------------------------------------------------------------------
113
114/// Internal runtime state tracked by [`FanOutExecutor`] for one group.
115#[derive(Debug)]
116struct FanOutGroupState {
117 group: FanOutGroup,
118 /// IDs of tasks that have reported completion.
119 completed: HashSet<String>,
120}
121
122impl FanOutGroupState {
123 fn new(group: FanOutGroup) -> Self {
124 Self {
125 group,
126 completed: HashSet::new(),
127 }
128 }
129
130 /// Record that `task_id` has finished. No-op if the id is unknown.
131 fn mark_complete(&mut self, task_id: &str) {
132 // Only mark tasks that actually belong to this group.
133 if self.group.children.iter().any(|t| t.id == task_id) {
134 self.completed.insert(task_id.to_string());
135 }
136 }
137
138 /// Check whether the merge condition is satisfied.
139 fn is_done(&self) -> bool {
140 let n_complete = self.completed.len();
141 let n_total = self.group.children.len();
142
143 match &self.group.merge_strategy {
144 MergeStrategy::WaitAll => n_complete >= n_total,
145 MergeStrategy::WaitAny => n_complete >= 1,
146 MergeStrategy::WaitN(n) => n_complete >= *n,
147 }
148 }
149
150 /// Return the number of completed tasks.
151 fn completed_count(&self) -> usize {
152 self.completed.len()
153 }
154
155 /// Return the total number of child tasks in the group.
156 fn total_count(&self) -> usize {
157 self.group.children.len()
158 }
159}
160
161// ---------------------------------------------------------------------------
162// FanOutExecutor
163// ---------------------------------------------------------------------------
164
165/// Manages multiple [`FanOutGroup`]s and tracks their completion state.
166///
167/// This type is **not** an async runtime; it is a pure bookkeeping layer.
168/// The caller is responsible for actually executing child tasks (e.g. via
169/// a thread pool or Tokio task set) and calling [`mark_complete`] when each
170/// one finishes.
171///
172/// [`mark_complete`]: FanOutExecutor::mark_complete
173#[derive(Debug, Default)]
174pub struct FanOutExecutor {
175 /// Groups indexed by `parent_step`.
176 groups: HashMap<String, FanOutGroupState>,
177 /// Reverse map: `task_id -> parent_step` for O(1) mark-complete routing.
178 task_to_group: HashMap<String, String>,
179}
180
181impl FanOutExecutor {
182 /// Create a new, empty executor.
183 #[must_use]
184 pub fn new() -> Self {
185 Self::default()
186 }
187
188 /// Register a new [`FanOutGroup`].
189 ///
190 /// If a group with the same `parent_step` already exists it is replaced
191 /// and all previous completion records for it are discarded.
192 pub fn submit_group(&mut self, group: FanOutGroup) {
193 // Remove stale task→group mappings for the old group (if any).
194 if let Some(old_state) = self.groups.get(&group.parent_step) {
195 for task in &old_state.group.children {
196 self.task_to_group.remove(&task.id);
197 }
198 }
199
200 // Build task→group mappings.
201 for task in &group.children {
202 self.task_to_group
203 .insert(task.id.clone(), group.parent_step.clone());
204 }
205
206 self.groups
207 .insert(group.parent_step.clone(), FanOutGroupState::new(group));
208 }
209
210 /// Mark a child task as complete.
211 ///
212 /// The executor looks up the group the task belongs to and records the
213 /// completion there. Unknown task IDs are silently ignored.
214 pub fn mark_complete(&mut self, task_id: &str) {
215 if let Some(group_id) = self.task_to_group.get(task_id).cloned() {
216 if let Some(state) = self.groups.get_mut(&group_id) {
217 state.mark_complete(task_id);
218 }
219 }
220 }
221
222 /// Return `true` when the merge condition for the group identified by
223 /// `group_id` (i.e. `parent_step`) is satisfied.
224 ///
225 /// Returns `false` for unknown groups.
226 #[must_use]
227 pub fn is_merge_condition_met(&self, group_id: &str) -> bool {
228 self.groups
229 .get(group_id)
230 .map(|s| s.is_done())
231 .unwrap_or(false)
232 }
233
234 /// Return the number of tasks that have completed in a group.
235 ///
236 /// Returns `0` for unknown groups.
237 #[must_use]
238 pub fn completed_count(&self, group_id: &str) -> usize {
239 self.groups
240 .get(group_id)
241 .map(|s| s.completed_count())
242 .unwrap_or(0)
243 }
244
245 /// Return the total number of child tasks in a group.
246 ///
247 /// Returns `0` for unknown groups.
248 #[must_use]
249 pub fn total_count(&self, group_id: &str) -> usize {
250 self.groups
251 .get(group_id)
252 .map(|s| s.total_count())
253 .unwrap_or(0)
254 }
255
256 /// Return the number of registered groups.
257 #[must_use]
258 pub fn group_count(&self) -> usize {
259 self.groups.len()
260 }
261
262 /// Remove a group and its task mappings, returning the group if present.
263 pub fn remove_group(&mut self, group_id: &str) -> Option<FanOutGroup> {
264 if let Some(state) = self.groups.remove(group_id) {
265 for task in &state.group.children {
266 self.task_to_group.remove(&task.id);
267 }
268 Some(state.group)
269 } else {
270 None
271 }
272 }
273}
274
275// ===========================================================================
276// FanoutStep — simple fan-out convenience wrapper
277// ===========================================================================
278
279/// A simple fan-out step that distributes a single input across multiple named branches.
280///
281/// Each branch receives a copy of the input tagged with the branch name. This is
282/// a higher-level convenience API built on top of [`FanOutExecutor`].
283///
284/// # Example
285///
286/// ```rust
287/// use oximedia_workflow::fanout::FanoutStep;
288///
289/// let step = FanoutStep::new(vec!["1080p".to_string(), "720p".to_string()]);
290/// let results = step.execute("source.mp4");
291/// assert_eq!(results.len(), 2);
292/// assert!(results[0].contains("1080p"));
293/// ```
294#[derive(Debug, Clone)]
295pub struct FanoutStep {
296 /// Named branches to fan out to.
297 pub branches: Vec<String>,
298}
299
300impl FanoutStep {
301 /// Create a new fan-out step with the given branch names.
302 #[must_use]
303 pub fn new(branches: Vec<String>) -> Self {
304 Self { branches }
305 }
306
307 /// Distribute `input` to all branches, returning one result string per branch.
308 ///
309 /// Each result is formatted as `"{branch}:{input}"` so the caller can identify
310 /// which branch produced which output.
311 #[must_use]
312 pub fn execute(&self, input: &str) -> Vec<String> {
313 self.branches
314 .iter()
315 .map(|branch| format!("{branch}:{input}"))
316 .collect()
317 }
318
319 /// Return the number of configured branches.
320 #[must_use]
321 pub fn branch_count(&self) -> usize {
322 self.branches.len()
323 }
324}
325
326// ===========================================================================
327// Tests
328// ===========================================================================
329
330#[cfg(test)]
331mod tests {
332 use super::*;
333
334 fn three_task_group(strategy: MergeStrategy) -> FanOutGroup {
335 FanOutGroup {
336 parent_step: "encode-variants".to_string(),
337 children: vec![
338 FanOutTask::new("t-1080p", "encode-variants", r#"{"profile":"1080p"}"#),
339 FanOutTask::new("t-720p", "encode-variants", r#"{"profile":"720p"}"#),
340 FanOutTask::new("t-480p", "encode-variants", r#"{"profile":"480p"}"#),
341 ],
342 merge_strategy: strategy,
343 }
344 }
345
346 // ------------------------------------------------------------------
347 // WaitAll
348 // ------------------------------------------------------------------
349
350 #[test]
351 fn wait_all_not_done_until_all_complete() {
352 let mut exec = FanOutExecutor::new();
353 exec.submit_group(three_task_group(MergeStrategy::WaitAll));
354
355 assert!(!exec.is_merge_condition_met("encode-variants"));
356 exec.mark_complete("t-1080p");
357 assert!(!exec.is_merge_condition_met("encode-variants"));
358 exec.mark_complete("t-720p");
359 assert!(!exec.is_merge_condition_met("encode-variants"));
360 exec.mark_complete("t-480p");
361 assert!(exec.is_merge_condition_met("encode-variants"));
362 }
363
364 #[test]
365 fn wait_all_idempotent_double_complete() {
366 let mut exec = FanOutExecutor::new();
367 exec.submit_group(three_task_group(MergeStrategy::WaitAll));
368
369 exec.mark_complete("t-1080p");
370 exec.mark_complete("t-1080p"); // duplicate — should not double-count
371 exec.mark_complete("t-720p");
372 exec.mark_complete("t-480p");
373
374 assert!(exec.is_merge_condition_met("encode-variants"));
375 assert_eq!(exec.completed_count("encode-variants"), 3);
376 }
377
378 // ------------------------------------------------------------------
379 // WaitAny
380 // ------------------------------------------------------------------
381
382 #[test]
383 fn wait_any_done_after_first_complete() {
384 let mut exec = FanOutExecutor::new();
385 exec.submit_group(three_task_group(MergeStrategy::WaitAny));
386
387 assert!(!exec.is_merge_condition_met("encode-variants"));
388 exec.mark_complete("t-720p");
389 assert!(exec.is_merge_condition_met("encode-variants"));
390 }
391
392 // ------------------------------------------------------------------
393 // WaitN
394 // ------------------------------------------------------------------
395
396 #[test]
397 fn wait_n_done_after_n_completions() {
398 let mut exec = FanOutExecutor::new();
399 exec.submit_group(three_task_group(MergeStrategy::WaitN(2)));
400
401 exec.mark_complete("t-1080p");
402 assert!(!exec.is_merge_condition_met("encode-variants"));
403 exec.mark_complete("t-480p");
404 assert!(exec.is_merge_condition_met("encode-variants"));
405 }
406
407 #[test]
408 fn wait_n_1_same_as_wait_any() {
409 let mut exec = FanOutExecutor::new();
410 exec.submit_group(three_task_group(MergeStrategy::WaitN(1)));
411
412 assert!(!exec.is_merge_condition_met("encode-variants"));
413 exec.mark_complete("t-1080p");
414 assert!(exec.is_merge_condition_met("encode-variants"));
415 }
416
417 // ------------------------------------------------------------------
418 // Unknown group / task
419 // ------------------------------------------------------------------
420
421 #[test]
422 fn unknown_group_returns_false() {
423 let exec = FanOutExecutor::new();
424 assert!(!exec.is_merge_condition_met("non-existent-group"));
425 }
426
427 #[test]
428 fn unknown_task_id_is_silently_ignored() {
429 let mut exec = FanOutExecutor::new();
430 exec.submit_group(three_task_group(MergeStrategy::WaitAll));
431 exec.mark_complete("ghost-task-xyz"); // should not panic or break state
432 assert!(!exec.is_merge_condition_met("encode-variants"));
433 assert_eq!(exec.completed_count("encode-variants"), 0);
434 }
435
436 // ------------------------------------------------------------------
437 // Multiple groups
438 // ------------------------------------------------------------------
439
440 #[test]
441 fn multiple_groups_independent() {
442 let mut exec = FanOutExecutor::new();
443
444 exec.submit_group(FanOutGroup {
445 parent_step: "group-a".to_string(),
446 children: vec![FanOutTask::new("a1", "group-a", "")],
447 merge_strategy: MergeStrategy::WaitAll,
448 });
449 exec.submit_group(FanOutGroup {
450 parent_step: "group-b".to_string(),
451 children: vec![
452 FanOutTask::new("b1", "group-b", ""),
453 FanOutTask::new("b2", "group-b", ""),
454 ],
455 merge_strategy: MergeStrategy::WaitAll,
456 });
457
458 exec.mark_complete("a1");
459 assert!(exec.is_merge_condition_met("group-a"));
460 assert!(!exec.is_merge_condition_met("group-b"));
461
462 exec.mark_complete("b1");
463 exec.mark_complete("b2");
464 assert!(exec.is_merge_condition_met("group-b"));
465 }
466
467 // ------------------------------------------------------------------
468 // Group replacement
469 // ------------------------------------------------------------------
470
471 #[test]
472 fn submit_group_replaces_existing() {
473 let mut exec = FanOutExecutor::new();
474 exec.submit_group(three_task_group(MergeStrategy::WaitAll));
475 exec.mark_complete("t-1080p");
476 assert_eq!(exec.completed_count("encode-variants"), 1);
477
478 // Replace with a fresh group — old completion records must be gone.
479 exec.submit_group(three_task_group(MergeStrategy::WaitAll));
480 assert_eq!(exec.completed_count("encode-variants"), 0);
481 assert!(!exec.is_merge_condition_met("encode-variants"));
482 }
483
484 // ------------------------------------------------------------------
485 // Remove group
486 // ------------------------------------------------------------------
487
488 #[test]
489 fn remove_group_clears_task_mappings() {
490 let mut exec = FanOutExecutor::new();
491 exec.submit_group(three_task_group(MergeStrategy::WaitAll));
492
493 let removed = exec.remove_group("encode-variants");
494 assert!(removed.is_some());
495 assert_eq!(exec.group_count(), 0);
496
497 // Completing a task whose group was removed must be a no-op.
498 exec.mark_complete("t-1080p");
499 assert!(!exec.is_merge_condition_met("encode-variants"));
500 }
501
502 // ------------------------------------------------------------------
503 // Empty group
504 // ------------------------------------------------------------------
505
506 #[test]
507 fn wait_all_on_empty_group_is_immediately_done() {
508 let mut exec = FanOutExecutor::new();
509 exec.submit_group(FanOutGroup {
510 parent_step: "empty-group".to_string(),
511 children: vec![],
512 merge_strategy: MergeStrategy::WaitAll,
513 });
514 // 0 completed >= 0 total → true
515 assert!(exec.is_merge_condition_met("empty-group"));
516 }
517
518 #[test]
519 fn wait_any_on_empty_group_is_not_done() {
520 let mut exec = FanOutExecutor::new();
521 exec.submit_group(FanOutGroup {
522 parent_step: "empty-any".to_string(),
523 children: vec![],
524 merge_strategy: MergeStrategy::WaitAny,
525 });
526 // 0 completed < 1 → false
527 assert!(!exec.is_merge_condition_met("empty-any"));
528 }
529
530 // ------------------------------------------------------------------
531 // Count helpers
532 // ------------------------------------------------------------------
533
534 #[test]
535 fn total_count_and_completed_count() {
536 let mut exec = FanOutExecutor::new();
537 exec.submit_group(three_task_group(MergeStrategy::WaitAll));
538
539 assert_eq!(exec.total_count("encode-variants"), 3);
540 assert_eq!(exec.completed_count("encode-variants"), 0);
541
542 exec.mark_complete("t-480p");
543 assert_eq!(exec.completed_count("encode-variants"), 1);
544 }
545
546 #[test]
547 fn counts_for_unknown_group_are_zero() {
548 let exec = FanOutExecutor::new();
549 assert_eq!(exec.total_count("ghost"), 0);
550 assert_eq!(exec.completed_count("ghost"), 0);
551 }
552}