fix(sdk): no longer error on cancelling fired timer

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<Self>) -> 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::<CancelAlreadyFiredTimerWf>();
+    worker.run().await.unwrap();
+}
+
 #[workflow]
 #[derive(Default)]
 struct HappyTimerWf;

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<WorkflowTaskFailedCause>) {
@@ -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,