apalis_workflow/dag/
mod.rs

1use std::{
2    collections::{HashMap, VecDeque},
3    fmt::{self, Debug},
4    marker::PhantomData,
5    sync::Mutex,
6};
7
8use apalis_core::{
9    backend::{BackendExt, codec::Codec},
10    error::BoxDynError,
11    task::Task,
12    task_fn::{TaskFn, task_fn},
13};
14use petgraph::{
15    Direction,
16    algo::toposort,
17    dot::Config,
18    graph::{DiGraph, EdgeIndex, NodeIndex},
19};
20/// DAG executor implementations
21pub mod executor;
22/// DAG service implementations
23pub mod service;
24
25/// DAG error definitions
26pub mod error;
27/// DAG node implementations
28pub mod node;
29
30/// DAG context implementations
31pub mod context;
32
33/// DAG response implementations
34pub mod response;
35
36/// DAG Decoding and encoding utilities
37pub mod decode;
38
39use serde::{Deserialize, Serialize};
40use tower::{Service, util::BoxCloneSyncService};
41
42use crate::{
43    DagService,
44    dag::{decode::DagCodec, error::DagFlowError, executor::DagExecutor, node::NodeService},
45};
46
47pub use context::DagFlowContext;
48pub use service::RootDagService;
49
50/// Directed Acyclic Graph (DAG) workflow builder
51#[derive(Debug)]
52pub struct DagFlow<B>
53where
54    B: BackendExt,
55{
56    name: String,
57    graph: Mutex<DiGraph<DagService<B::Compact, B::Context, B::IdType>, ()>>,
58    node_mapping: Mutex<HashMap<String, NodeIndex>>,
59}
60
61impl<B> fmt::Display for DagFlow<B>
62where
63    B: BackendExt,
64{
65    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
66        writeln!(f, "DAG name: {}", self.name)?;
67        writeln!(f, "Dot format:")?;
68        f.write_str(&self.to_dot())
69    }
70}
71
72impl<B> DagFlow<B>
73where
74    B: BackendExt,
75{
76    /// Create a new DAG workflow builder
77    #[must_use]
78    pub fn new(name: &str) -> Self {
79        Self {
80            name: name.to_owned(),
81            graph: Mutex::new(DiGraph::new()),
82            node_mapping: Mutex::new(HashMap::new()),
83        }
84    }
85
86    /// Add a node to the DAG
87    #[must_use]
88    pub fn add_node<S, Input, CodecError>(
89        &self,
90        name: &str,
91        service: S,
92    ) -> NodeBuilder<'_, Input, S::Response, B>
93    where
94        S: Service<Task<Input, B::Context, B::IdType>> + Send + 'static + Sync + Clone,
95        S::Future: Send + 'static,
96        B::Codec: Codec<Input, Compact = B::Compact, Error = CodecError>
97            + Codec<S::Response, Compact = B::Compact, Error = CodecError>
98            + 'static,
99        CodecError: Into<BoxDynError> + Send + 'static,
100        S::Error: Into<BoxDynError>,
101        B: Send + Sync + 'static,
102        Input: DagCodec<B, Error = CodecError> + Send + Sync + 'static,
103    {
104        let svc: NodeService<S, B, Input> = NodeService::new(service);
105        let node = self
106            .graph
107            .lock()
108            .expect("Failed to lock graph mutex")
109            .add_node(BoxCloneSyncService::new(svc));
110        self.node_mapping
111            .lock()
112            .expect("Failed to lock node_mapping mutex")
113            .insert(name.to_owned(), node);
114        NodeBuilder {
115            id: node,
116            dag: self,
117            _phantom: PhantomData,
118        }
119    }
120
121    /// Add a task function node to the DAG
122    pub fn node<F, Input, O, FnArgs, Err, CodecError>(&self, node: F) -> NodeBuilder<'_, Input, O, B>
123    where
124        TaskFn<F, Input, B::Context, FnArgs>: Service<Task<Input, B::Context, B::IdType>, Response = O, Error = Err> + Clone,
125        F: Send + 'static + Sync,
126        Input: Send + 'static + Sync,
127        FnArgs: Send + 'static + Sync,
128        B::Context: Send + Sync + 'static,
129        <TaskFn<F, Input, B::Context, FnArgs> as Service<Task<Input, B::Context, B::IdType>>>::Future:
130            Send + 'static,
131        B::Codec: Codec<Input, Compact = B::Compact, Error = CodecError> + 'static,
132        B::Codec: Codec<O, Compact = B::Compact, Error = CodecError> + 'static,
133        CodecError: Into<BoxDynError> + Send + 'static,
134        Err: Into<BoxDynError>,
135        B: Send + Sync + 'static,
136        Input: DagCodec<B, Error = CodecError> + Send + Sync + 'static,
137
138    {
139        self.add_node(std::any::type_name::<F>(), task_fn(node))
140    }
141
142    /// Add a routing node to the DAG
143    pub fn route<F, Input, O, FnArgs, Err, CodecError>(
144        &self,
145        router: F,
146    ) -> NodeBuilder<'_, Input, O, B>
147    where
148        TaskFn<F, Input, B::Context, FnArgs>: Service<Task<Input, B::Context, B::IdType>, Response = O, Error = Err> + Clone,
149        F: Send + 'static + Sync,
150        Input: Send + 'static + Sync,
151        FnArgs: Send + 'static + Sync,
152        <TaskFn<F, Input, B::Context, FnArgs> as Service<Task<Input, B::Context, B::IdType>>>::Future:
153            Send + 'static,
154        O: Into<NodeIndex>,
155        B::Context: Send + Sync + 'static,
156        B::Codec: Codec<Input, Compact = B::Compact, Error = CodecError> + 'static,
157        B::Codec: Codec<O, Compact = B::Compact, Error = CodecError> + 'static,
158        CodecError: Into<BoxDynError> + Send + 'static,
159        Err: Into<BoxDynError>,
160        B: Send + Sync + 'static,
161        Input: DagCodec<B, Error = CodecError> + Send + Sync + 'static,
162
163    {
164        self.add_node::<TaskFn<F, Input, B::Context, FnArgs>, Input, CodecError>(
165            std::any::type_name::<F>(),
166            task_fn(router),
167        )
168    }
169
170    /// Validate the DAG for cycles
171    pub fn validate(&self) -> Result<(), DagFlowError> {
172        // Validate DAG (check for cycles)
173        toposort(
174            &*self.graph.lock().expect("Failed to lock graph mutex"),
175            None,
176        )
177        .map_err(DagFlowError::CyclicDAG)?;
178        Ok(())
179    }
180
181    /// Export the DAG to DOT format
182    pub fn to_dot(&self) -> String {
183        let names = self
184            .node_mapping
185            .lock()
186            .expect("could not lock nodes")
187            .iter()
188            .map(|(name, &idx)| (idx, name.clone()))
189            .collect::<HashMap<_, _>>();
190        let get_node_attributes = |_, (index, _)| {
191            format!(
192                "label=\"{}\"",
193                names.get(&index).cloned().unwrap_or_default()
194            )
195        };
196        let graph = self.graph.lock().expect("could not lock graph");
197        let dot = petgraph::dot::Dot::with_attr_getters(
198            &*graph,
199            &[Config::NodeNoLabel, Config::EdgeNoLabel],
200            &|_, _| String::new(),
201            &get_node_attributes,
202        );
203        format!("{dot:?}")
204    }
205
206    /// Build the DAG executor
207    pub(crate) fn build(self) -> Result<DagExecutor<B>, DagFlowError> {
208        // Validate DAG (check for cycles)
209        let sorted = toposort(
210            &*self.graph.lock().expect("Failed to lock graph mutex"),
211            None,
212        )
213        .map_err(DagFlowError::CyclicDAG)?;
214
215        fn find_edge_nodes<N, E>(graph: &DiGraph<N, E>, direction: Direction) -> Vec<NodeIndex> {
216            graph
217                .node_indices()
218                .filter(|&n| graph.neighbors_directed(n, direction).count() == 0)
219                .collect()
220        }
221
222        let graph = self
223            .graph
224            .into_inner()
225            .expect("Failed to unlock graph mutex");
226
227        Ok(DagExecutor {
228            start_nodes: find_edge_nodes(&graph, Direction::Incoming),
229            end_nodes: find_edge_nodes(&graph, Direction::Outgoing),
230            graph,
231            node_mapping: self
232                .node_mapping
233                .into_inner()
234                .expect("Failed to unlock node_mapping mutex"),
235            topological_order: sorted,
236            not_ready: VecDeque::new(),
237        })
238    }
239}
240
241/// Builder for a node in the DAG
242pub struct NodeBuilder<'a, Input, Output, B>
243where
244    B: BackendExt,
245{
246    pub(crate) id: NodeIndex,
247    pub(crate) dag: &'a DagFlow<B>,
248    _phantom: PhantomData<(Input, Output)>,
249}
250
251impl<'a, Input, Output, B> std::fmt::Debug for NodeBuilder<'a, Input, Output, B>
252where
253    B: BackendExt,
254{
255    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
256        f.debug_struct("NodeBuilder")
257            .field("id", &self.id)
258            .finish_non_exhaustive()
259    }
260}
261
262impl<'a, Input, Output, B> Clone for NodeBuilder<'a, Input, Output, B>
263where
264    B: BackendExt,
265{
266    fn clone(&self) -> Self {
267        Self {
268            id: self.id,
269            dag: self.dag,
270            _phantom: PhantomData,
271        }
272    }
273}
274
275impl<Input, Output, B> NodeBuilder<'_, Input, Output, B>
276where
277    B: BackendExt,
278{
279    /// Specify dependencies for this node
280    #[allow(clippy::needless_pass_by_value)]
281    pub fn depends_on<D>(self, deps: D) -> NodeHandle<Input, Output>
282    where
283        D: DepsCheck<Input>,
284    {
285        let mut edges = Vec::new();
286        for dep in deps.to_node_indices() {
287            edges.push(self.dag.graph.lock().unwrap().add_edge(dep, self.id, ()));
288        }
289        NodeHandle {
290            id: self.id,
291            edges,
292            _phantom: PhantomData,
293        }
294    }
295}
296
297/// Handle for a node in the DAG
298#[derive(Clone, Debug)]
299pub struct NodeHandle<Input, Output> {
300    pub(crate) id: NodeIndex,
301    pub(crate) edges: Vec<EdgeIndex>,
302    pub(crate) _phantom: PhantomData<(Input, Output)>,
303}
304
305impl<Input, Output> NodeHandle<Input, Output> {
306    /// Get the node ID
307    #[must_use]
308    pub fn id(&self) -> NodeIndex {
309        self.id
310    }
311
312    /// Get the edge IDs
313    #[must_use]
314    pub fn edges(&self) -> &[EdgeIndex] {
315        &self.edges
316    }
317}
318
319/// Trait for converting dependencies into node IDs
320pub trait DepsCheck<Input> {
321    /// Convert dependencies to node indices
322    fn to_node_indices(&self) -> Vec<NodeIndex>;
323}
324
325impl DepsCheck<()> for () {
326    fn to_node_indices(&self) -> Vec<NodeIndex> {
327        Vec::new()
328    }
329}
330
331impl<'a, Input, Output, B> DepsCheck<Output> for &NodeBuilder<'a, Input, Output, B>
332where
333    B: BackendExt,
334{
335    fn to_node_indices(&self) -> Vec<NodeIndex> {
336        vec![self.id]
337    }
338}
339
340impl<Input, Output> DepsCheck<Output> for &NodeHandle<Input, Output> {
341    fn to_node_indices(&self) -> Vec<NodeIndex> {
342        vec![self.id]
343    }
344}
345
346impl<Input, Output> DepsCheck<Output> for (&NodeHandle<Input, Output>,) {
347    fn to_node_indices(&self) -> Vec<NodeIndex> {
348        vec![self.0.id]
349    }
350}
351
352impl<'a, Input, Output, B> DepsCheck<Output> for (&NodeBuilder<'a, Input, Output, B>,)
353where
354    B: BackendExt,
355{
356    fn to_node_indices(&self) -> Vec<NodeIndex> {
357        vec![self.0.id]
358    }
359}
360
361impl<Output, T: DepsCheck<Output>> DepsCheck<Vec<Output>> for Vec<T> {
362    fn to_node_indices(&self) -> Vec<NodeIndex> {
363        self.iter()
364            .flat_map(|item| item.to_node_indices())
365            .collect()
366    }
367}
368
369macro_rules! impl_deps_check {
370    ($( $len:literal => ( $( $in:ident $out:ident $idx:tt ),+ ) ),+ $(,)?) => {
371        $(
372            impl<'a, $( $in, )+ $( $out, )+ B> DepsCheck<( $( $out, )+ )>
373                for ( $( &NodeBuilder<'a, $in, $out, B>, )+ ) where B: BackendExt
374            {
375                fn to_node_indices(&self) -> Vec<NodeIndex> {
376                    vec![ $( self.$idx.id ),+ ]
377                }
378            }
379
380            impl<$( $in, )+ $( $out, )+> DepsCheck<( $( $out, )+ )>
381                for ( $( &NodeHandle<$in, $out>, )+ )
382            {
383                fn to_node_indices(&self) -> Vec<NodeIndex> {
384                    vec![ $( self.$idx.id ),+ ]
385                }
386            }
387        )+
388    };
389}
390
391impl_deps_check! {
392    1 => (Input1 Output1 0),
393    2 => (Input1 Output1 0, Input2 Output2 1),
394    3 => (Input1 Output1 0, Input2 Output2 1, Input3 Output3 2),
395    4 => (Input1 Output1 0, Input2 Output2 1, Input3 Output3 2, Input4 Output4 3),
396    5 => (Input1 Output1 0, Input2 Output2 1, Input3 Output3 2, Input4 Output4 3, Input5 Output5 4),
397    6 => (Input1 Output1 0, Input2 Output2 1, Input3 Output3 2, Input4 Output4 3, Input5 Output5 4, Input6 Output6 5),
398    7 => (Input1 Output1 0, Input2 Output2 1, Input3 Output3 2, Input4 Output4 3, Input5 Output5 4, Input6 Output6 5, Input7 Output7 6),
399    8 => (Input1 Output1 0, Input2 Output2 1, Input3 Output3 2, Input4 Output4 3, Input5 Output5 4, Input6 Output6 5, Input7 Output7 6, Input8 Output8 7),
400}
401
402/// State of the node in DAG execution
403#[derive(Debug, Clone, Deserialize, Serialize)]
404pub enum DagState {
405    /// Unknown state
406    Unknown,
407    /// State for a single node
408    SingleNode,
409    /// Fan-in state to gather inputs
410    FanIn,
411    /// Fan-out state to distribute inputs
412    FanOut,
413}
414
415#[cfg(test)]
416mod tests {
417    use std::{collections::BTreeMap, num::ParseIntError};
418
419    use apalis_core::{
420        error::BoxDynError,
421        task_fn::task_fn,
422        worker::{
423            builder::WorkerBuilder, context::WorkerContext, event::Event,
424            ext::event_listener::EventListenerExt,
425        },
426    };
427    use apalis_file_storage::JsonStorage;
428    use futures::StreamExt;
429    use petgraph::graph::NodeIndex;
430    use serde::{Deserialize, Serialize};
431    use serde_json::Value;
432
433    use crate::{WorkflowSink, dag::response::DagExecutionResponse};
434
435    use super::*;
436
437    #[tokio::test]
438    async fn test_basic_workflow() {
439        let dag = DagFlow::new("sequential-workflow");
440        let start = dag.add_node("start", task_fn(|task: u32| async move { task as usize }));
441        let middle = dag
442            .add_node(
443                "middle",
444                task_fn(|task: usize| async move { task.to_string() }),
445            )
446            .depends_on(&start);
447
448        let _end = dag
449            .add_node(
450                "end",
451                task_fn(|task: String, worker: WorkerContext| async move {
452                    worker.stop().unwrap();
453                    task.parse::<usize>()
454                }),
455            )
456            .depends_on(&middle);
457
458        println!("DAG in DOT format:\n{}", dag.to_dot());
459
460        let mut backend = JsonStorage::new_temp().unwrap();
461
462        backend.push_start(42).await.unwrap();
463
464        let worker = WorkerBuilder::new("rango-tango")
465            .backend(backend)
466            .on_event(|ctx, ev| {
467                println!("On Event = {ev:?}");
468                if matches!(ev, Event::Error(_)) {
469                    ctx.stop().unwrap();
470                }
471            })
472            .build(dag);
473        worker.run().await.unwrap();
474    }
475
476    #[tokio::test]
477    async fn test_fan_out_workflow() {
478        let dag = DagFlow::new("fan-out-workflow");
479        let source = dag.add_node("source", task_fn(|task: u32| async move { task as usize }));
480        let plus_one = dag
481            .add_node("plus_one", task_fn(|task: usize| async move { task + 1 }))
482            .depends_on(&source);
483
484        let multiply = dag
485            .add_node("multiply", task_fn(|task: usize| async move { task * 2 }))
486            .depends_on(&source);
487        let squared = dag
488            .add_node("squared", task_fn(|task: usize| async move { task * task }))
489            .depends_on(&source);
490
491        let _collector = dag
492            .add_node(
493                "collector",
494                task_fn(|task: (usize, usize, usize), w: WorkerContext| async move {
495                    w.stop().unwrap();
496                    task.0 + task.1 + task.2
497                }),
498            )
499            .depends_on((&plus_one, &multiply, &squared));
500
501        println!("DAG in DOT format:\n{}", dag.to_dot());
502
503        let mut backend: JsonStorage<Value> = JsonStorage::new_temp().unwrap();
504
505        backend.push_start(42).await.unwrap();
506
507        let worker = WorkerBuilder::new("rango-tango")
508            .backend(backend)
509            .on_event(|ctx, ev| {
510                println!("On Event = {ev:?}");
511                if matches!(ev, Event::Error(_)) {
512                    ctx.stop().unwrap();
513                }
514            })
515            .build(dag);
516        worker.run().await.unwrap();
517    }
518
519    #[tokio::test]
520    async fn test_fan_in_workflow() {
521        let dag = DagFlow::new("fan-in-workflow");
522        let get_name = dag.add_node(
523            "get_name",
524            task_fn(|task: usize| async move { task as usize }),
525        );
526        let get_age = dag.add_node(
527            "get_age",
528            task_fn(|task: u32| async move { task.to_string() }),
529        );
530        let get_address = dag.add_node(
531            "get_address",
532            task_fn(|task: i32| async move { task as usize }),
533        );
534        let main_collector = dag
535            .add_node(
536                "main_collector",
537                task_fn(|task: (String, usize, usize)| async move {
538                    task.2 + task.1 + task.0.parse::<usize>().unwrap()
539                }),
540            )
541            .depends_on((&get_age, &get_name, &get_address));
542
543        let side_collector = dag
544            .add_node(
545                "side_collector",
546                task_fn(
547                    |task: (usize, usize)| async move { [task.0, task.1].iter().sum::<usize>() },
548                ),
549            )
550            .depends_on((&get_name, &get_address));
551
552        let _final_node = dag
553            .add_node(
554                "final_node",
555                task_fn(|task: (usize, usize), w: WorkerContext| async move {
556                    w.stop().unwrap();
557                    task.0 + task.1
558                }),
559            )
560            .depends_on((&main_collector, &side_collector));
561
562        println!("DAG in DOT format:\n{}", dag.to_dot());
563
564        let mut backend: JsonStorage<Value> = JsonStorage::new_temp().unwrap();
565
566        backend
567            .start_fan_out((42usize, 43u32, 44i32))
568            .await
569            .unwrap();
570
571        let worker = WorkerBuilder::new("rango-tango")
572            .backend(backend)
573            .on_event(|ctx, ev| {
574                println!("On Event = {ev:?}");
575                if matches!(ev, Event::Error(_)) {
576                    ctx.stop().unwrap();
577                }
578            })
579            .build(dag);
580        worker.run().await.unwrap();
581    }
582
583    #[tokio::test]
584    async fn fan_in_completes_once_with_testworker() {
585        use apalis_core::task_fn::task_fn;
586        use apalis_core::worker::test_worker::TestWorker;
587        use apalis_file_storage::JsonStorage;
588        use serde_json::Value;
589
590        let dag = DagFlow::new("fan-in-testworker");
591
592        let a = dag.add_node("a", task_fn(|t: u32| async move { t }));
593        let b = dag.add_node("b", task_fn(|t: u32| async move { t }));
594
595        let _fan_in = dag
596            .add_node("fan_in", task_fn(|t: (u32, u32)| async move { t.0 + t.1 }))
597            .depends_on((&a, &b));
598
599        let mut backend: JsonStorage<Value> = JsonStorage::new_temp().unwrap();
600        backend.start_fan_out((1u32, 2u32)).await.unwrap();
601
602        let worker = TestWorker::new(backend, dag);
603        let stm = worker.into_stream();
604
605        let (res_map, final_res) = stm.take(5).collect::<Vec<_>>().await.into_iter().fold(
606            (BTreeMap::new(), None),
607            |(mut res_map, final_res), item| {
608                let (_, resp) = match item {
609                    Ok(v) => v,
610                    Err(e) => panic!("worker error: {e:?}"),
611                };
612                let resp = resp.expect("task error");
613
614                let kind = match &resp {
615                    DagExecutionResponse::EntryFanOut { .. } => "EntryFanOut",
616                    DagExecutionResponse::FanOut { .. } => "FanOut",
617                    DagExecutionResponse::EnqueuedNext { .. } => "EnqueuedNext",
618                    DagExecutionResponse::WaitingForDependencies { .. } => "WaitingForDependencies",
619                    DagExecutionResponse::Complete { .. } => "Complete",
620                };
621                *res_map.entry(kind).or_insert(0) += 1;
622
623                if let DagExecutionResponse::Complete { result } = resp {
624                    return (res_map, Some(result));
625                }
626                (res_map, final_res)
627            },
628        );
629
630        assert_eq!(final_res, Some(Value::from(3)));
631        assert_eq!(res_map["EntryFanOut"], 1); // Entry
632        assert_eq!(res_map["EnqueuedNext"], 2); // Node Results
633        assert_eq!(res_map["WaitingForDependencies"], 1); // Non Handler node
634        assert_eq!(res_map["Complete"], 1); // Handler Node
635    }
636
637    #[tokio::test]
638    async fn test_routed_workflow() {
639        let dag = DagFlow::new("routed-workflow");
640
641        let entry1 = dag.add_node("entry1", task_fn(|task: u32| async move { task as usize }));
642        let entry2 = dag.add_node("entry2", task_fn(|task: u32| async move { task as usize }));
643        let entry3 = dag.add_node("entry3", task_fn(|task: u32| async move { task as usize }));
644
645        #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
646        enum EntryRoute {
647            Entry1(NodeIndex),
648            Entry2(NodeIndex),
649            Entry3(NodeIndex),
650        }
651
652        impl Into<NodeIndex> for EntryRoute {
653            fn into(self) -> NodeIndex {
654                match self {
655                    EntryRoute::Entry1(idx) => idx,
656                    EntryRoute::Entry2(idx) => idx,
657                    EntryRoute::Entry3(idx) => idx,
658                }
659            }
660        }
661
662        impl DepsCheck<usize> for EntryRoute {
663            fn to_node_indices(&self) -> Vec<NodeIndex> {
664                vec![(*self).into()]
665            }
666        }
667
668        async fn collect(task: (usize, usize, usize)) -> usize {
669            task.0 + task.1 + task.2
670        }
671        let collector = dag.node(collect).depends_on((&entry1, &entry2, &entry3));
672
673        async fn vec_collect(task: Vec<usize>, _wrk: WorkerContext) -> usize {
674            task.iter().sum::<usize>()
675        }
676
677        let vec_collector = dag
678            .node(vec_collect)
679            .depends_on(vec![&entry1, &entry2, &entry3]);
680
681        async fn exit(task: (usize, usize)) -> Result<u32, ParseIntError> {
682            (task.0.to_string() + &task.1.to_string()).parse()
683        }
684
685        let on_collect = dag.node(exit).depends_on((&collector, &vec_collector));
686
687        async fn check_approval(
688            task: u32,
689            worker: WorkerContext,
690        ) -> Result<EntryRoute, BoxDynError> {
691            println!("Approval check for task: {}", task);
692            worker.stop().unwrap();
693            match task % 3 {
694                0 => Ok(EntryRoute::Entry1(NodeIndex::new(0))),
695                1 => Ok(EntryRoute::Entry2(NodeIndex::new(1))),
696                2 => Ok(EntryRoute::Entry3(NodeIndex::new(2))),
697                _ => Err(BoxDynError::from("Invalid task")),
698            }
699        }
700
701        dag.route(check_approval).depends_on(&on_collect);
702
703        println!("DAG in DOT format:\n{}", dag.to_dot());
704
705        let mut backend = JsonStorage::new_temp().unwrap();
706
707        backend.start_fan_out(vec![17, 18, 19]).await.unwrap();
708
709        let worker = WorkerBuilder::new("rango-tango")
710            .backend(backend)
711            .on_event(|ctx, ev| {
712                println!("On Event = {ev:?}");
713                if matches!(ev, Event::Error(_)) {
714                    ctx.stop().unwrap();
715                }
716            })
717            .build(dag);
718        worker.run().await.unwrap();
719    }
720}
apalis_workflow/dag/mod.rs

apalis_workflow/dag/
mod.rs
