ops_rs/
loop_op.rs

1use crate::prelude::*;
2use crate::{DryContext, Op, OpMetadata, WetContext};
3use async_trait::async_trait;
4
5/// Loop operation that executes a batch of operations repeatedly until a limit is reached
6pub struct LoopOp<T> {
7    counter_var: String,
8    limit: usize,
9    ops: Vec<Arc<dyn Op<T>>>,
10    loop_id: String,
11    continue_var: String,
12    break_var: String,
13    continue_on_error: bool,
14}
15
16impl<T> LoopOp<T>
17where
18    T: Send + 'static,
19{
20    /// Create a new loop operation
21    ///
22    /// # Arguments
23    /// * `counter_var` - Name of the counter variable to store in context
24    /// * `limit` - Maximum number of iterations to perform
25    /// * `ops` - Vector of operations to execute in each iteration
26    pub fn new(counter_var: String, limit: usize, ops: Vec<Arc<dyn Op<T>>>) -> Self {
27        // Generate a unique loop ID for scoped control flow variables
28        // Keep the preceding double underscores to minimize risk of collisions
29        let loop_id = ::uuid::Uuid::new_v4().to_string();
30        Self {
31            counter_var,
32            limit,
33            ops,
34            continue_var: format!("__continue_loop_{}", loop_id),
35            break_var: format!("__break_loop_{}", loop_id),
36            loop_id,
37            continue_on_error: false,
38        }
39    }
40
41    /// Add an operation to the loop
42    pub fn add_op(mut self, op: Arc<dyn Op<T>>) -> Self {
43        self.ops.push(op);
44        self
45    }
46
47    /// Set whether to continue on error
48    pub fn with_continue_on_error(mut self, continue_on_error: bool) -> Self {
49        self.continue_on_error = continue_on_error;
50        self
51    }
52
53    /// Rollback succeeded ops from the current iteration in reverse order (LIFO)
54    async fn rollback_iteration_ops(
55        &self,
56        succeeded_ops: &[Arc<dyn Op<T>>],
57        dry: &mut DryContext,
58        wet: &mut WetContext,
59    ) {
60        for op in succeeded_ops.iter().rev() {
61            if let Err(rollback_error) = op.rollback(dry, wet).await {
62                error!(
63                    "Failed to rollback op {} in loop iteration: {}",
64                    op.metadata().name,
65                    rollback_error
66                );
67            } else {
68                debug!(
69                    "Successfully rolled back op {} in loop iteration",
70                    op.metadata().name
71                );
72            }
73        }
74    }
75
76    /// Get the current counter value from dry context
77    fn get_counter(&self, dry: &DryContext) -> usize {
78        dry.get::<usize>(&self.counter_var).unwrap_or(0)
79    }
80
81    /// Set the counter value in dry context
82    fn set_counter(&self, dry: &mut DryContext, value: usize) {
83        dry.insert(&self.counter_var, value);
84    }
85}
86
87#[async_trait]
88impl<T> Op<Vec<T>> for LoopOp<T>
89where
90    T: Send + 'static,
91{
92    async fn perform(&self, dry: &mut DryContext, wet: &mut WetContext) -> OpResult<Vec<T>> {
93        let mut results = Vec::new();
94        let mut counter = self.get_counter(dry);
95
96        // Initialize counter in context if it doesn't exist
97        if !dry.contains(&self.counter_var) {
98            self.set_counter(dry, counter);
99        }
100
101        // Set loop context for scoped control flow
102        dry.insert("__current_loop_id", &self.loop_id);
103
104        while counter < self.limit {
105            // Check if we should abort before each iteration
106            if dry.is_aborted() {
107                let reason = dry
108                    .abort_reason()
109                    .cloned()
110                    .unwrap_or_else(|| "Loop operation aborted".to_string());
111                return Err(OpError::Aborted(reason));
112            }
113
114            // Clear scoped control flags for this iteration
115            dry.insert(&self.continue_var, false);
116            dry.insert(&self.break_var, false);
117
118            // Track succeeded ops in this iteration for potential rollback
119            let mut iteration_succeeded_ops = Vec::new();
120
121            // Execute all operations in the batch for this iteration
122            for op in &self.ops {
123                // Check abort before each op
124                if dry.is_aborted() {
125                    // Rollback succeeded ops from current iteration before aborting
126                    self.rollback_iteration_ops(&iteration_succeeded_ops, dry, wet)
127                        .await;
128                    let reason = dry
129                        .abort_reason()
130                        .cloned()
131                        .unwrap_or_else(|| "Loop operation aborted".to_string());
132                    return Err(OpError::Aborted(reason));
133                }
134
135                match op.perform(dry, wet).await {
136                    Ok(result) => {
137                        results.push(result);
138                        iteration_succeeded_ops.push(op.clone());
139
140                        // Check scoped continue flag
141                        if dry.get::<bool>(&self.continue_var).unwrap_or(false) {
142                            dry.insert(&self.continue_var, false); // Clear flag
143                            break; // Break out of ops loop, continue to next iteration
144                        }
145
146                        // Check scoped break flag
147                        if dry.get::<bool>(&self.break_var).unwrap_or(false) {
148                            dry.insert(&self.break_var, false); // Clear flag
149                            return Ok(results); // Break out of entire loop
150                        }
151                    }
152                    Err(OpError::Aborted(reason)) => {
153                        // Rollback succeeded ops from current iteration before aborting
154                        self.rollback_iteration_ops(&iteration_succeeded_ops, dry, wet)
155                            .await;
156                        return Err(OpError::Aborted(reason));
157                    }
158                    Err(error) => {
159                        if self.continue_on_error {
160                            // Log the error and continue with next iteration
161                            warn!("Operation {} failed in loop iteration {}: {}. Continuing with next iteration.", 
162                                  op.metadata().name, counter, error);
163                            // Rollback succeeded ops from current iteration
164                            self.rollback_iteration_ops(&iteration_succeeded_ops, dry, wet)
165                                .await;
166                            break; // Break out of ops loop, continue to next iteration
167                        } else {
168                            // Rollback succeeded ops from current iteration before failing
169                            self.rollback_iteration_ops(&iteration_succeeded_ops, dry, wet)
170                                .await;
171                            return Err(error);
172                        }
173                    }
174                }
175            }
176
177            // Increment counter and update context
178            counter += 1;
179            self.set_counter(dry, counter);
180        }
181
182        Ok(results)
183    }
184
185    fn metadata(&self) -> OpMetadata {
186        let description = if self.continue_on_error {
187            format!(
188                "Loop {} times over {} ops (continue on error)",
189                self.limit,
190                self.ops.len()
191            )
192        } else {
193            format!("Loop {} times over {} ops", self.limit, self.ops.len())
194        };
195        OpMetadata::builder("LoopOp")
196            .description(description)
197            .build()
198    }
199}
200
201#[cfg(test)]
202mod tests {
203    use super::*;
204
205    struct TestOp {
206        value: i32,
207    }
208
209    #[async_trait]
210    impl Op<i32> for TestOp {
211        async fn perform(&self, _dry: &mut DryContext, _wet: &mut WetContext) -> OpResult<i32> {
212            Ok(self.value)
213        }
214
215        fn metadata(&self) -> OpMetadata {
216            OpMetadata::builder("TestOp").build()
217        }
218    }
219
220    struct CounterOp;
221
222    #[async_trait]
223    impl Op<usize> for CounterOp {
224        async fn perform(&self, dry: &mut DryContext, _wet: &mut WetContext) -> OpResult<usize> {
225            let counter: usize = dry.get("loop_counter").unwrap_or(0);
226            Ok(counter)
227        }
228
229        fn metadata(&self) -> OpMetadata {
230            OpMetadata::builder("CounterOp").build()
231        }
232    }
233
234    // TEST0067: Run a LoopOp for 3 iterations with 2 ops each and verify all 6 results in order
235    #[tokio::test]
236    async fn test0067_loop_op_basic() {
237        let mut dry = DryContext::new();
238        let mut wet = WetContext::new();
239
240        let ops: Vec<Arc<dyn Op<i32>>> = vec![
241            Arc::new(TestOp { value: 10 }),
242            Arc::new(TestOp { value: 20 }),
243        ];
244
245        let loop_op = LoopOp::new("loop_counter".to_string(), 3, ops);
246        let results = loop_op.perform(&mut dry, &mut wet).await.unwrap();
247
248        // Should have 6 results (2 ops * 3 iterations)
249        assert_eq!(results.len(), 6);
250        assert_eq!(results, vec![10, 20, 10, 20, 10, 20]);
251    }
252
253    // TEST0068: Run a LoopOp where each op reads the loop counter and verify values are 0, 1, 2
254    #[tokio::test]
255    async fn test0068_loop_op_with_counter_access() {
256        let mut dry = DryContext::new();
257        let mut wet = WetContext::new();
258
259        let ops: Vec<Arc<dyn Op<usize>>> = vec![Arc::new(CounterOp)];
260
261        let loop_op = LoopOp::new("loop_counter".to_string(), 3, ops);
262        let results = loop_op.perform(&mut dry, &mut wet).await.unwrap();
263
264        // Should have counter values: [0, 1, 2]
265        assert_eq!(results, vec![0, 1, 2]);
266    }
267
268    // TEST0069: Start a LoopOp with a pre-initialized counter and verify it only executes the remaining iterations
269    #[tokio::test]
270    async fn test0069_loop_op_existing_counter() {
271        let mut dry = DryContext::new().with_value("my_counter", 2_usize);
272        let mut wet = WetContext::new();
273
274        let ops: Vec<Arc<dyn Op<i32>>> = vec![Arc::new(TestOp { value: 42 })];
275
276        let loop_op = LoopOp::new("my_counter".to_string(), 4, ops);
277        let results = loop_op.perform(&mut dry, &mut wet).await.unwrap();
278
279        // Should execute 2 times (from 2 to 4)
280        assert_eq!(results.len(), 2);
281        assert_eq!(results, vec![42, 42]);
282    }
283
284    // TEST0070: Run a LoopOp with a zero iteration limit and verify no ops are executed
285    #[tokio::test]
286    async fn test0070_loop_op_zero_limit() {
287        let mut dry = DryContext::new();
288        let mut wet = WetContext::new();
289
290        let ops: Vec<Arc<dyn Op<i32>>> = vec![Arc::new(TestOp { value: 99 })];
291
292        let loop_op = LoopOp::new("counter".to_string(), 0, ops);
293        let results = loop_op.perform(&mut dry, &mut wet).await.unwrap();
294
295        // Should not execute any operations
296        assert_eq!(results.len(), 0);
297    }
298
299    // TEST0071: Build a LoopOp with add_op chaining and verify all added ops run across all iterations
300    #[tokio::test]
301    async fn test0071_loop_op_builder_pattern() {
302        let mut dry = DryContext::new();
303        let mut wet = WetContext::new();
304
305        let loop_op = LoopOp::new("builder_counter".to_string(), 2, vec![])
306            .add_op(Arc::new(TestOp { value: 1 }))
307            .add_op(Arc::new(TestOp { value: 2 }));
308
309        let results = loop_op.perform(&mut dry, &mut wet).await.unwrap();
310
311        assert_eq!(results.len(), 4);
312        assert_eq!(results, vec![1, 2, 1, 2]);
313    }
314
315    // TEST0072: Run a LoopOp where the third op fails and verify succeeded ops are rolled back in reverse order
316    #[tokio::test]
317    async fn test0072_loop_op_rollback_on_iteration_failure() {
318        use std::sync::{Arc, Mutex};
319
320        struct RollbackTrackingOp {
321            id: u32,
322            should_fail: bool,
323            performed: Arc<Mutex<Vec<u32>>>, // Track all perform calls
324            rolled_back: Arc<Mutex<Vec<u32>>>, // Track all rollback calls
325        }
326
327        #[async_trait]
328        impl Op<u32> for RollbackTrackingOp {
329            async fn perform(&self, _dry: &mut DryContext, _wet: &mut WetContext) -> OpResult<u32> {
330                self.performed.lock().unwrap().push(self.id);
331                if self.should_fail {
332                    Err(OpError::ExecutionFailed(format!("Op {} failed", self.id)))
333                } else {
334                    Ok(self.id)
335                }
336            }
337
338            async fn rollback(&self, _dry: &mut DryContext, _wet: &mut WetContext) -> OpResult<()> {
339                self.rolled_back.lock().unwrap().push(self.id);
340                Ok(())
341            }
342
343            fn metadata(&self) -> OpMetadata {
344                OpMetadata::builder(&format!("RollbackTrackingOp{}", self.id)).build()
345            }
346        }
347
348        // Create tracking state
349        let performed = Arc::new(Mutex::new(Vec::new()));
350        let rolled_back = Arc::new(Mutex::new(Vec::new()));
351
352        let ops = vec![
353            Arc::new(RollbackTrackingOp {
354                id: 1,
355                should_fail: false,
356                performed: performed.clone(),
357                rolled_back: rolled_back.clone(),
358            }) as Arc<dyn Op<u32>>,
359            Arc::new(RollbackTrackingOp {
360                id: 2,
361                should_fail: false,
362                performed: performed.clone(),
363                rolled_back: rolled_back.clone(),
364            }) as Arc<dyn Op<u32>>,
365            Arc::new(RollbackTrackingOp {
366                id: 3,
367                should_fail: true, // This will fail in the first iteration
368                performed: performed.clone(),
369                rolled_back: rolled_back.clone(),
370            }) as Arc<dyn Op<u32>>,
371        ];
372
373        let loop_op = LoopOp::new("test_counter".to_string(), 2, ops);
374        let mut dry = DryContext::new();
375        let mut wet = WetContext::new();
376
377        // Execute loop - should fail on op3 in first iteration
378        let result = loop_op.perform(&mut dry, &mut wet).await;
379        assert!(result.is_err());
380
381        // Verify execution state - only first iteration should have run
382        let performed_calls = performed.lock().unwrap();
383        assert_eq!(
384            *performed_calls,
385            vec![1, 2, 3],
386            "Should have performed ops 1, 2, 3 in first iteration"
387        );
388
389        // Verify rollback state - only succeeded ops from failed iteration should be rolled back
390        let rolled_back_calls = rolled_back.lock().unwrap();
391        assert_eq!(
392            *rolled_back_calls,
393            vec![2, 1],
394            "Should have rolled back ops 2, 1 in reverse order (op 3 failed so no rollback)"
395        );
396    }
397
398    // TEST0073: Run a LoopOp where the last op fails and verify rollback occurs in LIFO order within the iteration
399    #[tokio::test]
400    async fn test0073_loop_op_rollback_order_within_iteration() {
401        use std::sync::{Arc, Mutex};
402
403        struct OrderTrackingOp {
404            id: u32,
405            rollback_order: Arc<Mutex<Vec<u32>>>,
406        }
407
408        #[async_trait]
409        impl Op<u32> for OrderTrackingOp {
410            async fn perform(&self, _dry: &mut DryContext, _wet: &mut WetContext) -> OpResult<u32> {
411                Ok(self.id)
412            }
413
414            async fn rollback(&self, _dry: &mut DryContext, _wet: &mut WetContext) -> OpResult<()> {
415                self.rollback_order.lock().unwrap().push(self.id);
416                Ok(())
417            }
418
419            fn metadata(&self) -> OpMetadata {
420                OpMetadata::builder(&format!("OrderTrackingOp{}", self.id)).build()
421            }
422        }
423
424        struct FailingOp;
425
426        #[async_trait]
427        impl Op<u32> for FailingOp {
428            async fn perform(&self, _dry: &mut DryContext, _wet: &mut WetContext) -> OpResult<u32> {
429                Err(OpError::ExecutionFailed("Intentional failure".to_string()))
430            }
431
432            fn metadata(&self) -> OpMetadata {
433                OpMetadata::builder("FailingOp").build()
434            }
435        }
436
437        let rollback_order = Arc::new(Mutex::new(Vec::new()));
438
439        let ops = vec![
440            Arc::new(OrderTrackingOp {
441                id: 1,
442                rollback_order: rollback_order.clone(),
443            }) as Arc<dyn Op<u32>>,
444            Arc::new(OrderTrackingOp {
445                id: 2,
446                rollback_order: rollback_order.clone(),
447            }) as Arc<dyn Op<u32>>,
448            Arc::new(OrderTrackingOp {
449                id: 3,
450                rollback_order: rollback_order.clone(),
451            }) as Arc<dyn Op<u32>>,
452            Arc::new(FailingOp) as Arc<dyn Op<u32>>, // Fails, triggering rollback
453        ];
454
455        let loop_op = LoopOp::new("test_counter".to_string(), 1, ops);
456        let mut dry = DryContext::new();
457        let mut wet = WetContext::new();
458
459        // Execute loop - should fail on FailingOp
460        let result = loop_op.perform(&mut dry, &mut wet).await;
461        assert!(result.is_err());
462
463        // Verify rollback order is LIFO (reverse of execution order)
464        let order = rollback_order.lock().unwrap();
465        assert_eq!(
466            *order,
467            vec![3, 2, 1],
468            "Rollback should happen in reverse order within iteration"
469        );
470    }
471
472    // TEST0074: Run a LoopOp that fails on iteration 2 and verify previously completed iterations are not rolled back
473    #[tokio::test]
474    async fn test0074_loop_op_successful_iterations_not_rolled_back() {
475        use std::sync::{Arc, Mutex};
476
477        struct IterationTrackingOp {
478            id: u32,
479            fail_on_iteration: Option<usize>, // Fail on specific iteration (0-based)
480            performed_iterations: Arc<Mutex<Vec<usize>>>,
481            rolled_back_iterations: Arc<Mutex<Vec<usize>>>,
482        }
483
484        #[async_trait]
485        impl Op<u32> for IterationTrackingOp {
486            async fn perform(&self, dry: &mut DryContext, _wet: &mut WetContext) -> OpResult<u32> {
487                let counter: usize = dry.get("test_counter").unwrap_or(0);
488                self.performed_iterations.lock().unwrap().push(counter);
489
490                if let Some(fail_iteration) = self.fail_on_iteration {
491                    if counter == fail_iteration {
492                        return Err(OpError::ExecutionFailed(format!(
493                            "Op {} failed on iteration {}",
494                            self.id, counter
495                        )));
496                    }
497                }
498
499                Ok(self.id)
500            }
501
502            async fn rollback(&self, dry: &mut DryContext, _wet: &mut WetContext) -> OpResult<()> {
503                let counter: usize = dry.get("test_counter").unwrap_or(0);
504                self.rolled_back_iterations.lock().unwrap().push(counter);
505                Ok(())
506            }
507
508            fn metadata(&self) -> OpMetadata {
509                OpMetadata::builder(&format!("IterationTrackingOp{}", self.id)).build()
510            }
511        }
512
513        let performed_iterations = Arc::new(Mutex::new(Vec::new()));
514        let rolled_back_iterations = Arc::new(Mutex::new(Vec::new()));
515
516        let ops = vec![Arc::new(IterationTrackingOp {
517            id: 1,
518            fail_on_iteration: Some(2), // Fail on third iteration (index 2)
519            performed_iterations: performed_iterations.clone(),
520            rolled_back_iterations: rolled_back_iterations.clone(),
521        }) as Arc<dyn Op<u32>>];
522
523        let loop_op = LoopOp::new("test_counter".to_string(), 5, ops);
524        let mut dry = DryContext::new();
525        let mut wet = WetContext::new();
526
527        // Execute loop - should fail on third iteration
528        let result = loop_op.perform(&mut dry, &mut wet).await;
529        assert!(result.is_err());
530
531        // Verify execution: should have run iterations 0, 1, 2
532        let performed = performed_iterations.lock().unwrap();
533        assert_eq!(
534            *performed,
535            vec![0, 1, 2],
536            "Should have performed iterations 0, 1, 2"
537        );
538
539        // Verify rollback: no rollback should happen because the op failed during perform()
540        // (ops that fail during perform() are not added to succeeded_ops, so they don't get rolled back)
541        let rolled_back = rolled_back_iterations.lock().unwrap();
542        assert_eq!(
543            *rolled_back,
544            Vec::<usize>::new(),
545            "No rollback should happen - the failing op wasn't successfully performed"
546        );
547    }
548
549    // TEST0075: Run a LoopOp where op2 fails on iteration 1 and verify only op1 from that iteration is rolled back
550    #[tokio::test]
551    async fn test0075_loop_op_mixed_iteration_with_rollback() {
552        use std::sync::{Arc, Mutex};
553
554        struct MixedIterationOp {
555            id: u32,
556            fail_on_iteration: Option<usize>,
557            performed_iterations: Arc<Mutex<Vec<(u32, usize)>>>, // (op_id, iteration)
558            rolled_back_iterations: Arc<Mutex<Vec<(u32, usize)>>>, // (op_id, iteration)
559        }
560
561        #[async_trait]
562        impl Op<u32> for MixedIterationOp {
563            async fn perform(&self, dry: &mut DryContext, _wet: &mut WetContext) -> OpResult<u32> {
564                let counter: usize = dry.get("test_counter").unwrap_or(0);
565                self.performed_iterations
566                    .lock()
567                    .unwrap()
568                    .push((self.id, counter));
569
570                if let Some(fail_iteration) = self.fail_on_iteration {
571                    if counter == fail_iteration {
572                        return Err(OpError::ExecutionFailed(format!(
573                            "Op {} failed on iteration {}",
574                            self.id, counter
575                        )));
576                    }
577                }
578
579                Ok(self.id)
580            }
581
582            async fn rollback(&self, dry: &mut DryContext, _wet: &mut WetContext) -> OpResult<()> {
583                let counter: usize = dry.get("test_counter").unwrap_or(0);
584                self.rolled_back_iterations
585                    .lock()
586                    .unwrap()
587                    .push((self.id, counter));
588                Ok(())
589            }
590
591            fn metadata(&self) -> OpMetadata {
592                OpMetadata::builder(&format!("MixedIterationOp{}", self.id)).build()
593            }
594        }
595
596        let performed_iterations = Arc::new(Mutex::new(Vec::new()));
597        let rolled_back_iterations = Arc::new(Mutex::new(Vec::new()));
598
599        // Create ops: first succeeds, second fails on iteration 1
600        let ops = vec![
601            Arc::new(MixedIterationOp {
602                id: 1,
603                fail_on_iteration: None, // Never fails
604                performed_iterations: performed_iterations.clone(),
605                rolled_back_iterations: rolled_back_iterations.clone(),
606            }) as Arc<dyn Op<u32>>,
607            Arc::new(MixedIterationOp {
608                id: 2,
609                fail_on_iteration: Some(1), // Fail on second iteration (index 1)
610                performed_iterations: performed_iterations.clone(),
611                rolled_back_iterations: rolled_back_iterations.clone(),
612            }) as Arc<dyn Op<u32>>,
613        ];
614
615        let loop_op = LoopOp::new("test_counter".to_string(), 3, ops);
616        let mut dry = DryContext::new();
617        let mut wet = WetContext::new();
618
619        // Execute loop - should fail on op2 in second iteration
620        let result = loop_op.perform(&mut dry, &mut wet).await;
621        assert!(result.is_err());
622
623        // Verify execution: should have run:
624        // Iteration 0: op1, op2 (both succeed)
625        // Iteration 1: op1 (succeeds), op2 (fails)
626        let performed = performed_iterations.lock().unwrap();
627        assert_eq!(
628            *performed,
629            vec![(1, 0), (2, 0), (1, 1), (2, 1)],
630            "Should have performed all ops"
631        );
632
633        // Verify rollback: only op1 from iteration 1 should be rolled back
634        // (op2 failed so it doesn't get rolled back, and iteration 0 was successful so no rollback)
635        let rolled_back = rolled_back_iterations.lock().unwrap();
636        assert_eq!(
637            *rolled_back,
638            vec![(1, 1)],
639            "Should only rollback op1 from failed iteration 1"
640        );
641    }
642
643    struct BreakOp {
644        should_break: bool,
645        value: i32,
646    }
647
648    #[async_trait]
649    impl Op<i32> for BreakOp {
650        async fn perform(&self, dry: &mut DryContext, _wet: &mut WetContext) -> OpResult<i32> {
651            if self.should_break {
652                // Set the scoped break flag — LoopOp reads __break_loop_{loop_id}
653                // We set it via __current_loop_id which LoopOp stores before the loop
654                let loop_id: String = dry.get("__current_loop_id").unwrap_or_default();
655                dry.insert(format!("__break_loop_{}", loop_id), true);
656            }
657            Ok(self.value)
658        }
659        fn metadata(&self) -> OpMetadata {
660            OpMetadata::builder("BreakOp").build()
661        }
662    }
663
664    // TEST0113: Run a LoopOp where an op sets the break flag and verify the loop terminates early
665    #[tokio::test]
666    async fn test0113_loop_op_break_terminates_loop() {
667        let ops: Vec<Arc<dyn Op<i32>>> = vec![
668            Arc::new(TestOp { value: 10 }),
669            Arc::new(BreakOp {
670                should_break: true,
671                value: 99,
672            }),
673            Arc::new(TestOp { value: 20 }), // should NOT execute after break
674        ];
675
676        let loop_op = LoopOp::new("counter".to_string(), 5, ops);
677        let mut dry = DryContext::new();
678        let mut wet = WetContext::new();
679
680        let results = loop_op.perform(&mut dry, &mut wet).await.unwrap();
681        // Only two results: TestOp(10) and BreakOp(99) from iteration 0, then loop stops
682        assert_eq!(results, vec![10, 99]);
683    }
684
685    // TEST0114: Run LoopOp::with_continue_on_error where an op fails and verify the loop continues
686    #[tokio::test]
687    async fn test0114_loop_op_continue_on_error_skips_failed_iterations() {
688        use std::sync::{Arc as StdArc, Mutex};
689
690        let iterations_seen: StdArc<Mutex<Vec<usize>>> = StdArc::new(Mutex::new(vec![]));
691        let log = iterations_seen.clone();
692
693        struct IterationLogOp {
694            fail_on: Option<usize>,
695            log: StdArc<Mutex<Vec<usize>>>,
696        }
697
698        #[async_trait]
699        impl Op<i32> for IterationLogOp {
700            async fn perform(&self, dry: &mut DryContext, _wet: &mut WetContext) -> OpResult<i32> {
701                let counter: usize = dry.get("it_counter").unwrap_or(0);
702                self.log.lock().unwrap().push(counter);
703                if Some(counter) == self.fail_on {
704                    return Err(OpError::ExecutionFailed(format!("fail on {}", counter)));
705                }
706                Ok(counter as i32)
707            }
708            fn metadata(&self) -> OpMetadata {
709                OpMetadata::builder("IterationLogOp").build()
710            }
711        }
712
713        let loop_op = LoopOp::new(
714            "it_counter".to_string(),
715            4,
716            vec![Arc::new(IterationLogOp {
717                fail_on: Some(1),
718                log: log,
719            }) as Arc<dyn Op<i32>>],
720        )
721        .with_continue_on_error(true);
722
723        let mut dry = DryContext::new();
724        let mut wet = WetContext::new();
725        let results = loop_op.perform(&mut dry, &mut wet).await.unwrap();
726
727        let seen = iterations_seen.lock().unwrap().clone();
728        // All 4 iterations were attempted despite failure on iteration 1
729        assert_eq!(seen, vec![0, 1, 2, 3]);
730        // Iteration 1 produced no result (failed), others did
731        assert_eq!(results, vec![0, 2, 3]);
732    }
733
734    // TEST0115: Run an empty LoopOp with a non-zero limit and verify it produces no results
735    #[tokio::test]
736    async fn test0115_loop_op_with_no_ops_produces_no_results() {
737        let loop_op: LoopOp<i32> = LoopOp::new("counter".to_string(), 5, vec![]);
738        let mut dry = DryContext::new();
739        let mut wet = WetContext::new();
740        let results = loop_op.perform(&mut dry, &mut wet).await.unwrap();
741        assert!(results.is_empty());
742        // Counter advances to limit
743        assert_eq!(dry.get::<usize>("counter").unwrap(), 5);
744    }
745
746    // TEST0076: Run a LoopOp configured to continue on error and verify subsequent iterations still execute
747    #[tokio::test]
748    async fn test0076_loop_op_continue_on_error() {
749        use std::sync::{Arc, Mutex};
750
751        struct ContinueOnErrorOp {
752            id: u32,
753            fail_on_iteration: Option<usize>,
754            performed_iterations: Arc<Mutex<Vec<(u32, usize)>>>, // (op_id, iteration)
755            rolled_back_iterations: Arc<Mutex<Vec<(u32, usize)>>>, // (op_id, iteration)
756        }
757
758        #[async_trait]
759        impl Op<u32> for ContinueOnErrorOp {
760            async fn perform(&self, dry: &mut DryContext, _wet: &mut WetContext) -> OpResult<u32> {
761                let counter: usize = dry.get("test_counter").unwrap_or(0);
762                self.performed_iterations
763                    .lock()
764                    .unwrap()
765                    .push((self.id, counter));
766
767                if let Some(fail_iteration) = self.fail_on_iteration {
768                    if counter == fail_iteration {
769                        return Err(OpError::ExecutionFailed(format!(
770                            "Op {} failed on iteration {}",
771                            self.id, counter
772                        )));
773                    }
774                }
775
776                Ok(self.id)
777            }
778
779            async fn rollback(&self, dry: &mut DryContext, _wet: &mut WetContext) -> OpResult<()> {
780                let counter: usize = dry.get("test_counter").unwrap_or(0);
781                self.rolled_back_iterations
782                    .lock()
783                    .unwrap()
784                    .push((self.id, counter));
785                Ok(())
786            }
787
788            fn metadata(&self) -> OpMetadata {
789                OpMetadata::builder(&format!("ContinueOnErrorOp{}", self.id)).build()
790            }
791        }
792
793        let performed_iterations = Arc::new(Mutex::new(Vec::new()));
794        let rolled_back_iterations = Arc::new(Mutex::new(Vec::new()));
795
796        // Create ops: first succeeds, second fails on iteration 1
797        let ops = vec![
798            Arc::new(ContinueOnErrorOp {
799                id: 1,
800                fail_on_iteration: None, // Never fails
801                performed_iterations: performed_iterations.clone(),
802                rolled_back_iterations: rolled_back_iterations.clone(),
803            }) as Arc<dyn Op<u32>>,
804            Arc::new(ContinueOnErrorOp {
805                id: 2,
806                fail_on_iteration: Some(1), // Fail on second iteration (index 1)
807                performed_iterations: performed_iterations.clone(),
808                rolled_back_iterations: rolled_back_iterations.clone(),
809            }) as Arc<dyn Op<u32>>,
810        ];
811
812        let loop_op = LoopOp::new("test_counter".to_string(), 3, ops).with_continue_on_error(true);
813        let mut dry = DryContext::new();
814        let mut wet = WetContext::new();
815
816        // Execute loop - should continue despite op2 failure in iteration 1
817        let result = loop_op.perform(&mut dry, &mut wet).await;
818        assert!(result.is_ok());
819        let results = result.unwrap();
820
821        // Verify execution: should have run:
822        // Iteration 0: op1, op2 (both succeed) -> results: [1, 2]
823        // Iteration 1: op1 (succeeds), op2 (fails, iteration continues) -> results: [1, 2, 1]
824        // Iteration 2: op1, op2 (both succeed) -> results: [1, 2, 1, 1, 2]
825        let performed = performed_iterations.lock().unwrap();
826        assert_eq!(
827            *performed,
828            vec![(1, 0), (2, 0), (1, 1), (2, 1), (1, 2), (2, 2)],
829            "Should have performed all ops across all iterations"
830        );
831
832        // Should have 5 successful results (op1 and op2 from iteration 0, op1 from iteration 1, op1 and op2 from iteration 2)
833        assert_eq!(
834            results,
835            vec![1, 2, 1, 1, 2],
836            "Should have results from successful operations only"
837        );
838
839        // Verify rollback: only op1 from iteration 1 should be rolled back (when op2 failed)
840        let rolled_back = rolled_back_iterations.lock().unwrap();
841        assert_eq!(
842            *rolled_back,
843            vec![(1, 1)],
844            "Should only rollback op1 from failed iteration 1"
845        );
846    }
847}
ops_rs/loop_op.rs

ops_rs/
loop_op.rs
