diff --git a/CHANGELOG.md b/CHANGELOG.md index 9b3cac079..39eebcdce 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -38,8 +38,8 @@ relevant information. ### Added * `Connection::raw_unary` makes a unary gRPC call by method path, with the request and the - response as encoded protobuf. It keeps the connection's TLS, API key, headers, interceptors, - metrics and retry policy, so you can call a method the client does not list. + response as encoded protobuf. The call uses the connection's TLS, API key, headers, + interceptors, metrics and retry policy. Use it to call a method that the client doesn't list. ### Breaking Changes * Workflow replay now reports nondeterminism when a scheduled Nexus operation's service or operation diff --git a/crates/client/src/lib.rs b/crates/client/src/lib.rs index 09a2b1cd5..6aaae42cf 100644 --- a/crates/client/src/lib.rs +++ b/crates/client/src/lib.rs @@ -991,7 +991,8 @@ pub struct TemporalServiceClient { cloud_svc_client: Box, test_svc_client: Box, health_svc_client: Box, - /// Calls by method path on the same service. A client built from mock services has none. + /// Makes calls by method path on the same gRPC service. It's `None` for a client built from + /// mock services. raw_unary: Option>, } diff --git a/crates/client/src/raw.rs b/crates/client/src/raw.rs index 23574c469..37cfb2a2e 100644 --- a/crates/client/src/raw.rs +++ b/crates/client/src/raw.rs @@ -1,6 +1,5 @@ -//! Unary gRPC calls by method path, with encoded protobuf in and out. A crate that generates its -//! own messages can call a method this client does not list, and keep the connection's TLS, API -//! key, headers, interceptors, metrics and retries. +//! Unary gRPC calls by method path, with encoded protobuf messages. A crate that generates its own +//! messages can use them to call a method that this client doesn't list. use crate::{Connection, grpc::RawGrpcCaller}; use bytes::{Buf, BufMut, Bytes}; @@ -93,11 +92,11 @@ impl Decoder for BytesCodec { impl Connection { /// Makes a unary gRPC call to the method at `path`, such as /// `/temporal.api.workflowservice.v1.WorkflowService/GetSystemInfo`, with `request` as the - /// encoded request message. Answers with the encoded response message. + /// encoded request message. Returns the encoded response message. /// - /// The call goes through the same stack as the listed methods, with the connection's TLS, - /// API key, headers, interceptors, metrics and retry policy. `path` also names the call for - /// the retry policy. Request extensions and metadata work as on a listed method. Set the + /// The call uses the same stack as the listed methods, with the connection's TLS, API key, + /// headers, interceptors, metrics and retry policy. The retry policy uses `path` as the call + /// name. Request extensions and metadata work as they do on a listed method. Set the /// `temporal-namespace` metadata to route by namespace, as the listed methods do. pub async fn raw_unary( &self, @@ -128,11 +127,11 @@ mod tests { use tonic::{Code, Request, Status}; use url::Url; - /// What the server saw of each call that the system info read did not make. + /// The requests that the server got, other than the `GetSystemInfo` call of the connect. type Seen = Arc>>; - /// A connection whose server answers each call with `answer`, after the connect's own - /// system info read. + /// A connection to a server that answers each call with `answer`, other than the + /// `GetSystemInfo` call of the connect. async fn connection( answer: impl Fn(&GrpcRequest, usize) -> Result, Status> + Send + Sync + 'static, ) -> (Connection, Seen) { diff --git a/crates/protos/protos/local/temporal/sdk/core/activity_task/activity_task.proto b/crates/protos/protos/local/temporal/sdk/core/activity_task/activity_task.proto index d0cd45fab..e6bb485bb 100644 --- a/crates/protos/protos/local/temporal/sdk/core/activity_task/activity_task.proto +++ b/crates/protos/protos/local/temporal/sdk/core/activity_task/activity_task.proto @@ -65,14 +65,14 @@ message Start { bool is_local = 17; // Run ID of this activity execution. Only set for standalone activities. string run_id = 19; - // For a local activity, the workflow task in which Core dispatched this execution. Retries - // that Core runs without a new task keep it. Unset for other activities. + // For a local activity, the workflow task in which Core dispatched this execution. The retries + // that Core runs without a new workflow task have the same value. Unset for other activities. WorkflowTaskAttempt workflow_task = 20; } -// One attempt of a workflow task, with the values from the poll response that started it. +// One attempt of a workflow task, from the poll response and the history of that attempt. message WorkflowTaskAttempt { - // The id of the task's WorkflowTaskStarted event. + // The event id of the task's WorkflowTaskStarted event. int64 started_event_id = 1; uint32 attempt = 2; // When the server started the task. @@ -80,8 +80,8 @@ message WorkflowTaskAttempt { // The started event id of the run's last completed workflow task, from the poll response. // 0 when no task of the run has completed. int64 previous_started_event_id = 4; - // The started event ids of the run's failed and timed out workflow tasks in history, in event - // order. It holds the last 100 only, to keep each start small. + // The started event ids of the run's failed and timed out workflow tasks, in event order. Core + // sets only the last 100, so that each start stays small. repeated int64 failed_tasks = 5; } diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md index 5423ac4e7..9a548b419 100644 --- a/crates/sdk-core/CHANGELOG.md +++ b/crates/sdk-core/CHANGELOG.md @@ -35,8 +35,10 @@ relevant information. ### Added * A Local Activity's `ActivityTask.Start` has `workflow_task` (field 20), with the started event - id, attempt and start time of the Workflow Task in which Core dispatched it. Retries that Core - runs on its own keep the value. + id, attempt and start time of the Workflow Task in which Core dispatched it. `workflow_task` + also has `previous_started_event_id`, the started event id of the run's last completed Workflow + Task, and `failed_tasks`, the started event ids of the run's last 100 failed or timed out + Workflow Tasks. The retries that Core runs without a new Workflow Task have the same value. ### Security * Ephemeral server downloads now use `zip` 9.0 for ZIP extraction, replacing the 8.x dependency @@ -50,8 +52,6 @@ relevant information. nondeterminism while retaining compatibility with histories recorded before these checks. ### Fixed -* With the workflow cache off, the eviction after a terminal completion that the server accepted - has reason `WORKFLOW_EXECUTION_ENDING`, as with a cache, not `CACHE_FULL`. * Dev and test server startup now stops waiting at the configured deadline even if a readiness request stalls. Timeout messages include the last connection failure when available. * Workers with large workflow caches avoid unnecessary CPU usage when no workflow tasks are waiting diff --git a/crates/sdk-core/src/core_tests/local_activity_workflow_task.rs b/crates/sdk-core/src/core_tests/local_activity_workflow_task.rs index da3732cb1..69fc76a77 100644 --- a/crates/sdk-core/src/core_tests/local_activity_workflow_task.rs +++ b/crates/sdk-core/src/core_tests/local_activity_workflow_task.rs @@ -1,11 +1,11 @@ -//! The workflow task that each local activity start names. - use crate::{ - replay::TestHistoryBuilder, + replay::{DEFAULT_ACTIVITY_TYPE, TestHistoryBuilder}, test_help::{ - MockPollCfg, ResponseType, WorkerExt, WorkerTestHelpers, build_mock_pollers, mock_worker, - schedule_local_activity_cmd, start_timer_cmd, + MockPollCfg, ResponseType, WorkerExt, WorkerTestHelpers, build_mock_pollers, + hist_to_poll_resp, mock_worker, schedule_local_activity_cmd, single_hist_mock_sg, + start_timer_cmd, }, + worker::client::mocks::mock_worker_client, }; use std::time::{Duration, SystemTime}; use temporalio_common::protos::{ @@ -17,7 +17,12 @@ use temporalio_common::protos::{ workflow_commands::{ActivityCancellationType, ScheduleLocalActivity, workflow_command}, workflow_completion::WorkflowActivationCompletion, }, - temporal::api::{common::v1::RetryPolicy, enums::v1::EventType, failure::v1::Failure}, + temporal::api::{ + common::v1::RetryPolicy, + enums::v1::{EventType, WorkflowTaskFailedCause}, + failure::v1::Failure, + workflowservice::v1::GetWorkflowExecutionHistoryResponse, + }, }; async fn complete( @@ -34,10 +39,10 @@ async fn complete( .unwrap(); } -fn la_cmd() -> workflow_command::Variant { +fn la_cmd(seq: u32) -> workflow_command::Variant { schedule_local_activity_cmd( - 1, - "1", + seq, + &seq.to_string(), ActivityCancellationType::TryCancel, Duration::from_secs(60), ) @@ -100,7 +105,7 @@ async fn a_local_activity_names_the_live_task_that_dispatched_it() { .await; let fired = worker.poll_workflow_activation().await.unwrap(); assert!(!fired.is_replaying); - complete(&worker, &fired, vec![la_cmd()]).await; + complete(&worker, &fired, vec![la_cmd(1)]).await; let la = worker.poll_activity_task().await.unwrap(); let start = start_of(&la); @@ -123,7 +128,7 @@ async fn a_rerun_after_a_timed_out_task_names_the_later_task() { let mut t = TestHistoryBuilder::default(); t.add_wfe_started_with_wft_timeout(Duration::from_secs(60)); t.add_full_wf_task(); - // The task after it ran the local activity, then the worker died and the task timed out. + // The second task ran the local activity. Then the worker stopped and the task timed out. t.add_workflow_task_scheduled_and_started(); t.add_workflow_task_timed_out(); t.add_workflow_task_scheduled_and_started(); @@ -132,12 +137,11 @@ async fn a_rerun_after_a_timed_out_task_names_the_later_task() { let worker = mock_worker(mock); let first = worker.poll_workflow_activation().await.unwrap(); - complete(&worker, &first, vec![la_cmd()]).await; + complete(&worker, &first, vec![la_cmd(1)]).await; - // History has no marker for it, so it runs again, dispatched by the live task. + // History has no marker for the local activity, so it runs again in the current task. let la = worker.poll_activity_task().await.unwrap(); assert_eq!(dispatching(start_of(&la)), (9, 1)); - // The start also names the last completed task and the task that timed out. let task = start_of(&la).workflow_task.as_ref().unwrap(); assert_eq!(task.previous_started_event_id, 3); assert_eq!(task.failed_tasks, [6]); @@ -193,3 +197,173 @@ async fn a_retry_that_core_runs_keeps_the_dispatching_task() { assert_eq!(dispatching(start_of(&second_attempt)), (3, 1)); finish(worker, second_attempt).await; } + +#[tokio::test] +async fn local_activities_of_one_activation_name_the_same_task() { + let mut t = TestHistoryBuilder::default(); + t.add_by_type(EventType::WorkflowExecutionStarted); + t.add_workflow_task_scheduled_and_started(); + let mut mock = build_mock_pollers(MockPollCfg::from_resps(t, [ResponseType::AllHistory])); + mock.worker_cfg(|wc| wc.max_cached_workflows = 1); + let worker = mock_worker(mock); + + let first = worker.poll_workflow_activation().await.unwrap(); + complete(&worker, &first, vec![la_cmd(1), la_cmd(2)]).await; + + let one = worker.poll_activity_task().await.unwrap(); + let two = worker.poll_activity_task().await.unwrap(); + assert_eq!(dispatching(start_of(&one)), (3, 1)); + assert_eq!(start_of(&one).workflow_task, start_of(&two).workflow_task); + worker + .complete_activity_task(ActivityTaskCompletion { + task_token: one.task_token, + result: Some(ActivityExecutionResult::ok(vec![1].into())), + }) + .await + .unwrap(); + let resolved = worker.poll_workflow_activation().await.unwrap(); + complete(&worker, &resolved, vec![]).await; + finish(worker, two).await; +} + +#[tokio::test] +async fn a_reschedule_after_a_backoff_timer_names_the_later_task() { + let mut t = TestHistoryBuilder::default(); + t.add_by_type(EventType::WorkflowExecutionStarted); + t.add_full_wf_task(); + // The first attempt failed with a backoff past the local retry threshold. + t.add_local_activity_marker( + 1, + "1", + None, + Some(Failure::application_failure("retry me".to_string(), false)), + |m| m.backoff = Some(Duration::from_secs(10).try_into().unwrap()), + ); + let timer_started = t.add_by_type(EventType::TimerStarted); + t.add_timer_fired(timer_started, "1".to_string()); + t.add_workflow_task_scheduled_and_started(); + let mut mock = build_mock_pollers(MockPollCfg::from_resps(t, [ResponseType::AllHistory])); + mock.worker_cfg(|wc| wc.max_cached_workflows = 1); + let worker = mock_worker(mock); + + let first = worker.poll_workflow_activation().await.unwrap(); + complete(&worker, &first, vec![la_cmd(1)]).await; + let backoff = worker.poll_workflow_activation().await.unwrap(); + assert!(backoff.is_replaying); + complete( + &worker, + &backoff, + vec![start_timer_cmd(1, Duration::from_secs(10))], + ) + .await; + let fired = worker.poll_workflow_activation().await.unwrap(); + assert!(!fired.is_replaying); + complete( + &worker, + &fired, + vec![ + ScheduleLocalActivity { + seq: 2, + activity_id: "1".to_string(), + activity_type: DEFAULT_ACTIVITY_TYPE.to_string(), + attempt: 2, + original_schedule_time: Some(SystemTime::now().into()), + start_to_close_timeout: Some(Duration::from_secs(60).try_into().unwrap()), + ..Default::default() + } + .into(), + ], + ) + .await; + + let la = worker.poll_activity_task().await.unwrap(); + assert_eq!(start_of(&la).attempt, 2); + assert_eq!(dispatching(start_of(&la)), (9, 1)); + assert_eq!( + start_of(&la) + .workflow_task + .as_ref() + .unwrap() + .previous_started_event_id, + 3 + ); + finish(worker, la).await; +} + +#[tokio::test] +async fn failed_tasks_from_every_history_page_are_named() { + let wf_id = "fake_wf_id"; + let mut t = TestHistoryBuilder::default(); + t.add_wfe_started_with_wft_timeout(Duration::from_secs(60)); + t.add_full_wf_task(); + let timer_started = t.add_by_type(EventType::TimerStarted); + t.add_timer_fired(timer_started, "1".to_string()); + t.add_workflow_task_scheduled_and_started(); + t.add_workflow_task_failed_with_failure( + WorkflowTaskFailedCause::WorkflowWorkerUnhandledFailure, + Failure::default(), + ); + t.add_full_wf_task(); + let timer_started = t.add_by_type(EventType::TimerStarted); + let second_page_end = timer_started as usize; + t.add_timer_fired(timer_started, "2".to_string()); + t.add_workflow_task_scheduled_and_started(); + t.add_workflow_task_timed_out(); + t.add_workflow_task_scheduled_and_started(); + + // The poll response has the first page and each history fetch adds one page. The failed task + // is on the second page and the timed out task is on the last. + let mut first_page = hist_to_poll_resp(&t, wf_id, ResponseType::AllHistory).resp; + first_page.history.as_mut().unwrap().events.truncate(6); + first_page.next_page_token = vec![1]; + let mut second_page: GetWorkflowExecutionHistoryResponse = + t.get_full_history_info().unwrap().into(); + second_page + .history + .as_mut() + .unwrap() + .events + .truncate(second_page_end); + second_page.next_page_token = vec![2]; + let last_page: GetWorkflowExecutionHistoryResponse = t.get_full_history_info().unwrap().into(); + let mut client = mock_worker_client(); + client + .expect_get_workflow_execution_history() + .returning(move |_, _, _| Ok(second_page.clone())) + .times(1); + client + .expect_get_workflow_execution_history() + .returning(move |_, _, _| Ok(last_page.clone())) + .times(1); + let mut mock = single_hist_mock_sg(wf_id, t, [ResponseType::Raw(first_page)], client, true); + mock.worker_cfg(|wc| wc.max_cached_workflows = 1); + let worker = mock_worker(mock); + + let first = worker.poll_workflow_activation().await.unwrap(); + complete( + &worker, + &first, + vec![start_timer_cmd(1, Duration::from_secs(1))], + ) + .await; + let fired = worker.poll_workflow_activation().await.unwrap(); + assert!(fired.is_replaying); + complete( + &worker, + &fired, + vec![start_timer_cmd(2, Duration::from_secs(1))], + ) + .await; + let fired = worker.poll_workflow_activation().await.unwrap(); + assert!(!fired.is_replaying); + complete(&worker, &fired, vec![la_cmd(1)]).await; + + let la = worker.poll_activity_task().await.unwrap(); + let task = start_of(&la).workflow_task.as_ref().unwrap(); + assert_eq!( + (task.started_event_id, task.previous_started_event_id), + (19, 11) + ); + assert_eq!(task.failed_tasks, [8, 16]); + finish(worker, la).await; +} diff --git a/crates/sdk-core/src/core_tests/workflow_tasks.rs b/crates/sdk-core/src/core_tests/workflow_tasks.rs index f7c68629b..2cc51f438 100644 --- a/crates/sdk-core/src/core_tests/workflow_tasks.rs +++ b/crates/sdk-core/src/core_tests/workflow_tasks.rs @@ -4278,64 +4278,3 @@ async fn cached_run_fetch_failure_doesnt_spam_task_fails() { 2 ); } - -/// The eviction that follows a terminal completion the server accepted, with the cache disabled. -async fn eviction_after_completing_workflow_with_zero_cache( - completion_result: Result<(), tonic::Status>, -) -> EvictionReason { - let wfid = "fake_wf_id"; - let mut t = TestHistoryBuilder::default(); - t.add_by_type(EventType::WorkflowExecutionStarted); - t.add_full_wf_task(); - let mut mock_cfg = MockPollCfg::from_resp_batches(wfid, t, [1], mock_worker_client()); - mock_cfg.make_poll_stream_interminable = true; - let mut completion_result = Some(completion_result); - mock_cfg.completion_mock_fn = Some(Box::new(move |_| { - completion_result - .take() - .unwrap_or(Ok(())) - .map(|()| 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 activation = core.poll_workflow_activation().await.unwrap(); - core.complete_workflow_activation(WorkflowActivationCompletion::from_cmd( - activation.run_id, - CompleteWorkflowExecution { result: None }.into(), - )) - .await - .unwrap(); - let eviction = core.poll_workflow_activation().await.unwrap(); - let reason = assert_matches!( - eviction.jobs.as_slice(), - [WorkflowActivationJob { - variant: Some(workflow_activation_job::Variant::RemoveFromCache(r)), - }] => r.reason() - ); - core.complete_workflow_activation(WorkflowActivationCompletion::empty(eviction.run_id)) - .await - .unwrap(); - core.drain_pollers_and_shutdown().await; - reason -} - -#[tokio::test] -async fn accepted_terminal_completion_with_zero_cache_evicts_as_execution_ending() { - assert_eq!( - eviction_after_completing_workflow_with_zero_cache(Ok(())).await, - EvictionReason::WorkflowExecutionEnding - ); -} - -#[tokio::test] -async fn rejected_terminal_completion_with_zero_cache_evicts_with_rejection_reason() { - assert_eq!( - eviction_after_completing_workflow_with_zero_cache(Err(tonic::Status::not_found( - "Workflow task not found." - ))) - .await, - EvictionReason::TaskNotFound - ); -} diff --git a/crates/sdk-core/src/worker/activities/local_activities.rs b/crates/sdk-core/src/worker/activities/local_activities.rs index 06558a7c4..ff7e61be1 100644 --- a/crates/sdk-core/src/worker/activities/local_activities.rs +++ b/crates/sdk-core/src/worker/activities/local_activities.rs @@ -141,7 +141,7 @@ pub(crate) struct NewLocalAct { pub(crate) workflow_type: String, pub(crate) workflow_exec_info: WorkflowExecution, pub(crate) schedule_time: SystemTime, - /// The workflow task in which the run dispatched this request. Retries keep it. + /// The workflow task in which the run dispatched this request. Retries reuse it. pub(crate) workflow_task: Option, } diff --git a/crates/sdk-core/src/worker/workflow/machines/workflow_machines.rs b/crates/sdk-core/src/worker/workflow/machines/workflow_machines.rs index cd31ec838..5de620de9 100644 --- a/crates/sdk-core/src/worker/workflow/machines/workflow_machines.rs +++ b/crates/sdk-core/src/worker/workflow/machines/workflow_machines.rs @@ -79,7 +79,7 @@ type Result = std::result::Result; slotmap::new_key_type! { struct MachineKey; } -/// The most failed workflow tasks that a local activity start names. +/// The maximum number of failed workflow tasks in a local activity start. const MAX_FAILED_WFTS: usize = 100; /// Handles all the logic for driving a workflow. It orchestrates many state machines that together @@ -90,8 +90,8 @@ pub(crate) struct WorkflowMachines { /// kept because the lang side polls & completes for every workflow task, but we do not need /// to poll the server that often during replay. last_history_from_server: HistoryUpdate, - /// The started event ids of the run's failed and timed out workflow tasks, for the local - /// activities that need to know which tasks never completed. + /// The started event ids of the run's failed and timed out workflow tasks. Each local activity + /// start has them, so that the activity knows which tasks never completed. failed_wft_started_ids: VecDeque, /// Protocol messages that have yet to be processed for the current WFT. protocol_msgs: Vec, @@ -430,8 +430,7 @@ impl WorkflowMachines { self.failed_wft_started_ids.iter().copied().collect() } - /// Only the last ones are kept, which bounds the list on a long run. A local activity's task - /// facts only need the failed tasks since its run's last write to a Stream. + /// Keeps only the last `MAX_FAILED_WFTS` ids, so that the list stays small on a long run. fn note_failed_wfts(&mut self, events: &[HistoryEvent]) { for event in events { let started = match &event.attributes { @@ -443,7 +442,7 @@ impl WorkflowMachines { } _ => continue, }; - // A page of history may come again after a cache miss. + // Core can get the same page of history again after a cache miss. if self .failed_wft_started_ids .back() diff --git a/crates/sdk-core/src/worker/workflow/machines/workflow_machines/local_acts.rs b/crates/sdk-core/src/worker/workflow/machines/workflow_machines/local_acts.rs index d8daba14c..4fc2cd4f6 100644 --- a/crates/sdk-core/src/worker/workflow/machines/workflow_machines/local_acts.rs +++ b/crates/sdk-core/src/worker/workflow/machines/workflow_machines/local_acts.rs @@ -70,7 +70,7 @@ impl LocalActivityData { self.executing.insert(sa.seq); LocalActRequest::New(NewLocalAct { schedule_time: SystemTime::now(), - // The run stamps the task that dispatches the request. + // `ManagedRun` sets this when it dispatches the request. workflow_task: None, schedule_cmd: sa, workflow_type: wf_type.to_string(), diff --git a/crates/sdk-core/src/worker/workflow/managed_run.rs b/crates/sdk-core/src/worker/workflow/managed_run.rs index 77b5a70d8..662ef4ac9 100644 --- a/crates/sdk-core/src/worker/workflow/managed_run.rs +++ b/crates/sdk-core/src/worker/workflow/managed_run.rs @@ -973,24 +973,6 @@ impl ManagedRun { self.task_buffer.has_tasks() } - /// Reports an eviction already requested because the cache is full (or disabled) as - /// [EvictionReason::WorkflowExecutionEnding], once the server has accepted the run's terminal - /// completion. With the cache disabled the cache-full eviction is requested before the - /// completion is reported, so without this lang would never see that the execution ended. - /// - /// The caller cannot tell acceptance from rejection: [WFTReportStatus::Reported] is also the - /// status of a completion the server rejected. This relies on a rejection having already - /// replaced the cache-full reason: the rejection's eviction request is sent before the - /// post-activation message, and [Self::request_eviction] keeps the most recent reason. - pub(super) fn report_cache_full_eviction_as_execution_ending(&mut self) { - if let Some(info) = self.trying_to_evict.as_mut() - && info.reason == EvictionReason::CacheFull - { - info.reason = EvictionReason::WorkflowExecutionEnding; - info.message = "Workflow completed".to_string(); - } - } - pub(super) fn request_eviction(&mut self, info: RequestEvictMsg) -> EvictionRequestResult { // If we were waiting on a page fetch and we're getting evicted because fetching failed, // then make sure we allow the completion to proceed, otherwise we're stuck waiting forever. @@ -1295,8 +1277,8 @@ impl ManagedRun { } } - /// Names the current workflow task on each new local activity, since a local activity cannot - /// learn by itself which task dispatched it. + /// Sets the current workflow task on each new local activity request. The activity can't find + /// out by itself which task dispatched it. fn stamp_dispatching_task(&self, requests: &mut [LocalActRequest]) { let Some(wft) = self.wft.as_ref() else { if requests diff --git a/crates/sdk-core/src/worker/workflow/mod.rs b/crates/sdk-core/src/worker/workflow/mod.rs index c19d693ce..51f00e35d 100644 --- a/crates/sdk-core/src/worker/workflow/mod.rs +++ b/crates/sdk-core/src/worker/workflow/mod.rs @@ -1092,7 +1092,7 @@ pub(crate) enum OutstandingActivation { struct WorkflowTaskInfo { task_token: TaskToken, attempt: u32, - /// From the poll response, like the attempt. + /// From the poll response. started_event_id: i64, started_time: Option, /// The started event id of the run's last completed task, from the poll response. diff --git a/crates/sdk-core/src/worker/workflow/workflow_stream.rs b/crates/sdk-core/src/worker/workflow/workflow_stream.rs index 2cd039745..d00173f64 100644 --- a/crates/sdk-core/src/worker/workflow/workflow_stream.rs +++ b/crates/sdk-core/src/worker/workflow/workflow_stream.rs @@ -441,15 +441,13 @@ impl WFStream { if res.is_none() && let Some(rh) = self.runs.get_mut(run_id) { - let completed_workflow = rh.workflow_is_finished() - && matches!(report.wft_report_status, WFTReportStatus::Reported { .. }); - if completed_workflow { - rh.report_cache_full_eviction_as_execution_ending(); - } // Attempt to produce the next activation if needed res = rh.check_more_activations(); // If there's no more work and we reported workflow completion to server, evict. - if res.is_none() && completed_workflow { + if res.is_none() + && rh.workflow_is_finished() + && matches!(report.wft_report_status, WFTReportStatus::Reported { .. }) + { res = rh .request_eviction(RequestEvictMsg { run_id: run_id.to_string(),