refactor: Add Processed flag on matching await-next with all-processed

Author: tomasol

crates/testing/test-programs/stub/workflow/src/lib.rs

@@ -3,7 +3,7 @@ use crate::obelisk::workflow::workflow_support::{self, ClosingStrategy};
 use crate::testing::stub_activity::activity;
 use crate::testing::stub_activity_obelisk_ext::activity as activity_ext;
 use crate::testing::stub_activity_obelisk_stub::activity as activity_stub;
-use obelisk::types::execution::{ExecutionId, StubError};
+use obelisk::types::execution::{ExecutionError, ExecutionId, StubError};
 use wit_bindgen::generate;
 
 generate!({ generate_all });
@@ -17,7 +17,7 @@ impl Guest for Component {
         activity_stub::foo_stub(&execution_id, Ok(&format!("stubbing {arg}")))
             .expect("stubbed activity must accept returned value once");
         let (actual_execution_id, ret_val) =
-            activity_ext::foo_await_next(&join_set).expect("stubbed activity must resolve");
+            activity_ext::foo_await_next(&join_set).expect("stubbed execution result above");
         assert_eq!(execution_id.id, actual_execution_id.id);
         ret_val
     }
@@ -67,4 +67,12 @@ impl Guest for Component {
     fn stub_subworkflow(execution_id: ExecutionId, retval: String) -> Result<(), StubError> {
         activity_stub::foo_stub(&execution_id, Ok(&format!("stubbing {retval}")))
     }
+
+    fn await_next_produces_all_processed_error() {
+        let join_set = workflow_support::new_join_set_generated(ClosingStrategy::Complete);
+        let ExecutionError::AllProcessed = activity_ext::foo_await_next(&join_set).unwrap_err()
+        else {
+            unreachable!()
+        };
+    }
 }

crates/testing/test-programs/stub/workflow/wit/deps/testing_stub-workflow/stub-workflow.wit

@@ -10,4 +10,6 @@ interface workflow {
 
     submit-race: func() -> string;
     stub-subworkflow: func(execution-id: execution-id, retval: string) -> result<_, stub-error>;
+
+    await-next-produces-all-processed-error: func();
 }

crates/wasm-workers/src/workflow/event_history.rs

@@ -706,6 +706,7 @@ impl EventHistory {
                 && Some(requested_ffqn) == found_requested_ffqn.as_ref() =>
             {
                 trace!(%join_set_id, "matched JoinNextChild with JoinNextTooMany");
+                self.event_history[found_idx].1 = Processed;
                 let all_processed = ExecutionErrorVariant::AllProcessed.as_wast_val();
                 Ok(FindMatchingResponse::Found(ChildReturnValue::WastVal(
                     WastVal::Result(Err(Some(Box::new(all_processed)))),

crates/wasm-workers/src/workflow/workflow_worker.rs

@@ -689,6 +689,7 @@ impl<C: ClockFn + 'static, S: Sleep + 'static> Worker for WorkflowWorker<C, S> {
 #[cfg(test)]
 pub(crate) mod tests {
     use super::*;
+    use crate::activity::activity_worker::tests::compile_activity_stub;
     use crate::{
         activity::activity_worker::tests::{
             FIBO_10_INPUT, FIBO_10_OUTPUT, compile_activity, spawn_activity_fibo, wasm_file_name,
@@ -1772,8 +1773,6 @@ pub(crate) mod tests {
     async fn stub(
         #[values(db_tests::Database::Memory, db_tests::Database::Sqlite)] db: db_tests::Database,
     ) {
-        use crate::activity::activity_worker::tests::compile_activity_stub;
-
         const FFQN_WORKFLOW_STUB: FunctionFqn = FunctionFqn::new_static_tuple(
             test_programs_stub_workflow_builder::exports::testing::stub_workflow::workflow::SUBMIT_STUB_AWAIT,
         );
@@ -1997,4 +1996,83 @@ pub(crate) mod tests {
         drop(exec_task);
         db_pool.close().await.unwrap();
     }
+
+    #[rstest::rstest]
+    #[tokio::test]
+    async fn await_next_produces_all_processed_error(
+        #[values(db_tests::Database::Memory, db_tests::Database::Sqlite)] db: db_tests::Database,
+    ) {
+        const FFQN: FunctionFqn = FunctionFqn::new_static_tuple(
+            test_programs_stub_workflow_builder::exports::testing::stub_workflow::workflow::AWAIT_NEXT_PRODUCES_ALL_PROCESSED_ERROR
+        );
+        test_utils::set_up();
+        let sim_clock = SimClock::epoch();
+        let (_guard, db_pool) = db.set_up().await;
+        let fn_registry = TestingFnRegistry::new_from_components(vec![
+            compile_activity_stub(test_programs_stub_activity_builder::TEST_PROGRAMS_STUB_ACTIVITY),
+            compile_workflow(test_programs_stub_workflow_builder::TEST_PROGRAMS_STUB_WORKFLOW),
+        ]);
+
+        let worker = compile_workflow_worker(
+            test_programs_stub_workflow_builder::TEST_PROGRAMS_STUB_WORKFLOW,
+            db_pool.clone(),
+            sim_clock.clone(),
+            TokioSleep,
+            JoinNextBlockingStrategy::Interrupt,
+            &fn_registry,
+        );
+        let exec_task = ExecTask::new(
+            worker,
+            ExecConfig {
+                batch_size: 1,
+                lock_expiry: Duration::from_secs(1),
+                tick_sleep: TICK_SLEEP,
+                component_id: ComponentId::dummy_workflow(),
+                task_limiter: None,
+                executor_id: ExecutorId::generate(),
+            },
+            sim_clock.clone(),
+            db_pool.clone(),
+            Arc::new([FFQN]),
+        );
+
+        let execution_id = ExecutionId::generate();
+        let db_connection = db_pool.connection();
+        db_connection
+            .create(CreateRequest {
+                created_at: sim_clock.now(),
+                execution_id: execution_id.clone(),
+                ffqn: FFQN,
+                params: Params::empty(),
+                parent: None,
+                metadata: concepts::ExecutionMetadata::empty(),
+                scheduled_at: sim_clock.now(),
+                retry_exp_backoff: Duration::ZERO,
+                max_retries: u32::MAX,
+                component_id: ComponentId::dummy_workflow(),
+                scheduled_by: None,
+            })
+            .await
+            .unwrap();
+
+        {
+            let task_count = exec_task
+                .tick_test(sim_clock.now(), RunId::generate())
+                .await
+                .unwrap()
+                .wait_for_tasks()
+                .await
+                .unwrap();
+            assert_eq!(1, task_count);
+        }
+
+        let res = db_connection.get(&execution_id).await.unwrap();
+        assert_matches!(
+            res.into_finished_result().unwrap().unwrap(),
+            SupportedFunctionReturnValue::None
+        );
+
+        drop(exec_task);
+        db_pool.close().await.unwrap();
+    }
 }