diff --git a/CHANGELOG.md b/CHANGELOG.md index 27fa290f5..1e17bcf62 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -89,6 +89,8 @@ to docs, or any other relevant information. ### Fixed * Panics from update validators now reject the update instead of repeatedly failing workflow tasks. +* Workers with `max_cached_workflows` set to 0 no longer stall when a local activity resolves while + the resolution for an earlier one is still being delivered. ## [0.6.0] - 2026-08-04 ## [0.5.0] diff --git a/crates/sdk-core/src/worker/workflow/workflow_stream.rs b/crates/sdk-core/src/worker/workflow/workflow_stream.rs index ee47df21a..56a9c258c 100644 --- a/crates/sdk-core/src/worker/workflow/workflow_stream.rs +++ b/crates/sdk-core/src/worker/workflow/workflow_stream.rs @@ -343,7 +343,11 @@ impl WFStream { .collect(); // Keeping the run until its LAs resolve lets their incremental activations share the // current WFT. The final completion will queue the zero-cache eviction as usual. - if has_zero_sized_cache && !rh.waiting_on_local_activities() { + // Jobs still queued for this WFT (ex: resolutions for LAs that finished while an earlier + // one was being delivered) count as not-yet-resolved too: they can schedule further LAs, + // and the commands they produce are only flushed by the completion that finally answers + // the WFT. Evicting first would strand those commands in the discarded machines. + if has_zero_sized_cache && !rh.waiting_on_local_activities() && !rh.more_pending_work() { acts.extend(self.request_eviction_of_lru_run().into_run_update_resp()) } acts diff --git a/crates/sdk-core/tests/integ_tests/workflow_tests/local_activities.rs b/crates/sdk-core/tests/integ_tests/workflow_tests/local_activities.rs index f7f29c05f..03e0692ac 100644 --- a/crates/sdk-core/tests/integ_tests/workflow_tests/local_activities.rs +++ b/crates/sdk-core/tests/integ_tests/workflow_tests/local_activities.rs @@ -26,6 +26,7 @@ use temporalio_common::{ coresdk::{ ActivityTaskCompletion, AsJsonPayloadExt, FromJsonPayloadExt, activity_result::ActivityExecutionResult, + activity_task::activity_task as act_task, common::extract_local_activity_marker_data, workflow_activation::{ WorkflowActivation, WorkflowActivationJob, workflow_activation_job, @@ -3015,6 +3016,146 @@ async fn local_activity_resolutions_are_delivered_incrementally() { handle.fetch_history_and_replay(&mut worker).await.unwrap(); } +/// With a zero-sized cache, a run keeps its workflow task until every local activity resolution has +/// been delivered and the task answered. Resolutions queued while an earlier one is outstanding with +/// lang may schedule further activities, and the markers for all of them belong on the completion +/// that finally answers the task. +#[tokio::test] +async fn zero_cache_doesnt_evict_before_wft_is_answered() { + let wfid = "fake_wf_id"; + let mut t = TestHistoryBuilder::default(); + t.add_by_type(EventType::WorkflowExecutionStarted); + t.add_workflow_task_scheduled_and_started(); + + let reported_commands = Arc::new(SegQueue::new()); + let recorder = reported_commands.clone(); + let mut mock_cfg = + MockPollCfg::from_resp_batches(wfid, t, [ResponseType::AllHistory], mock_worker_client()); + mock_cfg.enforce_correct_number_of_polls = false; + mock_cfg.completion_mock_fn = Some(Box::new(move |c| { + recorder.push( + c.commands + .iter() + .map(|cmd| cmd.command_type()) + .collect::>(), + ); + Ok(Default::default()) + })); + let mut mock = build_mock_pollers(mock_cfg); + mock.worker_cfg(|wc| wc.max_cached_workflows = 0); + let core = mock_worker(mock); + + let la_cmd = |seq: u32, id: &str| { + schedule_local_activity_cmd( + seq, + id, + ProtoActivityCancellationType::TryCancel, + Duration::from_secs(60), + ) + }; + + let task = core.poll_workflow_activation().await.unwrap(); + core.complete_workflow_activation(WorkflowActivationCompletion::from_cmds( + task.run_id, + vec![la_cmd(1, "1"), la_cmd(2, "2")], + )) + .await + .unwrap(); + + // Which activity each task belongs to is matched on activity id, since the order the two are + // handed out in is not something this test should depend on. + let queued_las = [ + core.poll_activity_task().await.unwrap(), + core.poll_activity_task().await.unwrap(), + ]; + let token_for = |activity_id: &str| { + queued_las + .iter() + .find(|t| { + matches!(&t.variant, Some(act_task::Variant::Start(start)) + if start.activity_id == activity_id) + }) + .unwrap_or_else(|| panic!("no local activity task for id {activity_id}")) + .task_token + .clone() + }; + + core.complete_activity_task(ActivityTaskCompletion { + task_token: token_for("1"), + result: Some(ActivityExecutionResult::ok(vec![1].into())), + }) + .await + .unwrap(); + let resolve_first = core.poll_workflow_activation().await.unwrap(); + assert_matches!( + resolve_first.jobs.as_slice(), + [WorkflowActivationJob { + variant: Some(workflow_activation_job::Variant::ResolveActivity(resolution)), + }] => assert_eq!(resolution.seq, 1) + ); + + // Finishing the second activity while the first resolution is outstanding with lang is what + // queues it rather than delivering it, leaving no outstanding activities but an undelivered + // job once the completion below lands. + core.complete_activity_task(ActivityTaskCompletion { + task_token: token_for("2"), + result: Some(ActivityExecutionResult::ok(vec![2].into())), + }) + .await + .unwrap(); + core.complete_workflow_activation(WorkflowActivationCompletion::empty(resolve_first.run_id)) + .await + .unwrap(); + + // Scheduling from the queued resolution keeps the task open with the first two markers still + // buffered, which is the point at which the run must survive to report them. + let resolve_second = core.poll_workflow_activation().await.unwrap(); + assert_matches!( + resolve_second.jobs.as_slice(), + [WorkflowActivationJob { + variant: Some(workflow_activation_job::Variant::ResolveActivity(resolution)), + }] => assert_eq!(resolution.seq, 2) + ); + core.complete_workflow_activation(WorkflowActivationCompletion::from_cmd( + resolve_second.run_id, + la_cmd(3, "3"), + )) + .await + .unwrap(); + + let third_la = core.poll_activity_task().await.unwrap(); + core.complete_activity_task(ActivityTaskCompletion { + task_token: third_la.task_token, + result: Some(ActivityExecutionResult::ok(vec![3].into())), + }) + .await + .unwrap(); + let resolve_third = core.poll_workflow_activation().await.unwrap(); + assert_matches!( + resolve_third.jobs.as_slice(), + [WorkflowActivationJob { + variant: Some(workflow_activation_job::Variant::ResolveActivity(resolution)), + }] => assert_eq!(resolution.seq, 3) + ); + core.complete_execution(&resolve_third.run_id).await; + + core.shutdown().await; + + let mut all_reported = vec![]; + while let Some(cmds) = reported_commands.pop() { + all_reported.push(cmds); + } + let markers_reported: usize = all_reported + .iter() + .flatten() + .filter(|c| **c == CommandType::RecordMarker) + .count(); + assert_eq!( + markers_reported, 3, + "all three local activity markers should reach the server, got {all_reported:?}" + ); +} + #[workflow] #[derive(Default)] struct OldBatchedLocalActivityHistoryWf;