From ec54c4d4431bba4372f7b4455ddcb1ae85e9eed2 Mon Sep 17 00:00:00 2001 From: Chris Olszewski Date: Wed, 13 May 2026 17:06:34 -0400 Subject: [PATCH] fix(sdk): no longer error on cancelling fired timer --- .../integ_tests/workflow_tests/timers.rs | 30 ++++++++++++ crates/sdk/src/workflow_future.rs | 48 ++++++++++++------- 2 files changed, 62 insertions(+), 16 deletions(-) diff --git a/crates/sdk-core/tests/integ_tests/workflow_tests/timers.rs b/crates/sdk-core/tests/integ_tests/workflow_tests/timers.rs index 8b8ec1b2c..dd52e5240 100644 --- a/crates/sdk-core/tests/integ_tests/workflow_tests/timers.rs +++ b/crates/sdk-core/tests/integ_tests/workflow_tests/timers.rs @@ -160,6 +160,36 @@ async fn parallel_timers() { worker.run_until_done().await.unwrap(); } +#[workflow] +#[derive(Default)] +struct CancelAlreadyFiredTimerWf; + +#[workflow_methods] +impl CancelAlreadyFiredTimerWf { + #[run(name = DEFAULT_WORKFLOW_TYPE)] + async fn run(ctx: &mut WorkflowContext) -> WorkflowResult<()> { + let mut t1 = ctx.timer(Duration::from_secs(1)); + let mut t2 = ctx.timer(Duration::from_secs(1)); + temporalio_sdk::workflows::select! { + _ = t1 => {} + _ = t2 => {} + } + t2.cancel(); + Ok(()) + } +} + +#[tokio::test] +async fn cancel_unpolled_timer_after_both_timers_fire_same_activation() { + let mut t = canned_histories::parallel_timer("1", "2"); + t.add_workflow_task_completed(); + t.add_workflow_execution_completed(); + + let mut worker = build_fake_sdk(MockPollCfg::from_hist_builder(t)); + worker.register_workflow::(); + worker.run().await.unwrap(); +} + #[workflow] #[derive(Default)] struct HappyTimerWf; diff --git a/crates/sdk/src/workflow_future.rs b/crates/sdk/src/workflow_future.rs index 9d2f46392..20e15d41f 100644 --- a/crates/sdk/src/workflow_future.rs +++ b/crates/sdk/src/workflow_future.rs @@ -182,23 +182,20 @@ impl WorkflowFuture { } fn unblock(&mut self, event: UnblockEvent) -> Result<(), Error> { - let cmd_id = match event { - UnblockEvent::Timer(seq, _) => CommandID::Timer(seq), - UnblockEvent::Activity(seq, _) => CommandID::Activity(seq), - UnblockEvent::WorkflowStart(seq, _) => CommandID::ChildWorkflowStart(seq), - UnblockEvent::WorkflowComplete(seq, _) => CommandID::ChildWorkflowComplete(seq), - UnblockEvent::SignalExternal(seq, _) => CommandID::SignalExternal(seq), - UnblockEvent::CancelExternal(seq, _) => CommandID::CancelExternal(seq), - UnblockEvent::NexusOperationStart(seq, _) => CommandID::NexusOpStart(seq), - UnblockEvent::NexusOperationComplete(seq, _) => CommandID::NexusOpComplete(seq), + let cmd_id = CommandID::from(&event); + self.maybe_unblock(event) + .then_some(()) + .ok_or_else(|| anyhow!("Command {cmd_id:?} not found to unblock!")) + } + + fn maybe_unblock(&mut self, event: UnblockEvent) -> bool { + let cmd_id = CommandID::from(&event); + let Some(unblocker) = self.command_status.remove(&cmd_id) else { + return false; }; - let unblocker = self.command_status.remove(&cmd_id); let _guard = SdkWakeGuard::new(); - let _ = unblocker - .ok_or_else(|| anyhow!("Command {cmd_id:?} not found to unblock!"))? - .unblocker - .send(event); - Ok(()) + let _ = unblocker.unblocker.send(event); + true } fn fail_wft(&self, run_id: String, fail: Error, cause: Option) { @@ -696,7 +693,11 @@ impl WorkflowFuture { RustWfCmd::Cancel(cancellable_id) => { let cmd_variant = match cancellable_id { CancellableID::Timer(seq) => { - self.unblock(UnblockEvent::Timer(seq, TimerResult::Cancelled))?; + if !self.maybe_unblock(UnblockEvent::Timer(seq, TimerResult::Cancelled)) + { + // If timer is no longer present, do not emit command for cancellation. + continue; + } // Re-poll wf future since a timer is now unblocked res = self.execution.poll_run(cx); workflow_command::Variant::CancelTimer(CancelTimer { seq }) @@ -842,6 +843,21 @@ enum CommandID { NexusOpComplete(u32), } +impl From<&UnblockEvent> for CommandID { + fn from(event: &UnblockEvent) -> Self { + match event { + UnblockEvent::Timer(seq, _) => CommandID::Timer(*seq), + UnblockEvent::Activity(seq, _) => CommandID::Activity(*seq), + UnblockEvent::WorkflowStart(seq, _) => CommandID::ChildWorkflowStart(*seq), + UnblockEvent::WorkflowComplete(seq, _) => CommandID::ChildWorkflowComplete(*seq), + UnblockEvent::SignalExternal(seq, _) => CommandID::SignalExternal(*seq), + UnblockEvent::CancelExternal(seq, _) => CommandID::CancelExternal(*seq), + UnblockEvent::NexusOperationStart(seq, _) => CommandID::NexusOpStart(*seq), + UnblockEvent::NexusOperationComplete(seq, _) => CommandID::NexusOpComplete(*seq), + } + } +} + fn update_response( instance_id: String, resp: update_response::Response,