Forward termination to steps when a user cancels a resumable run

python_modules/dagster/dagster/_core/instance/methods/run_launcher_methods.py

@@ -204,6 +204,15 @@ def count_resume_run_attempts(self, run_id: str) -> int:
 
     def run_will_resume(self, run_id: str) -> bool:
         """Check if run will resume."""
+        from dagster._core.storage.dagster_run import DagsterRunStatus
+
         if not self.run_monitoring_enabled:
             return False
+        # A run that the user is canceling (or has already canceled) will not be resumed by the
+        # daemon, which only resumes runs that are still in the STARTED status. Treating such a run
+        # as resumable would suppress forwarding the termination signal to in-flight steps, leaving
+        # them running indefinitely.
+        run = self.get_run_by_id(run_id)
+        if run and run.status in (DagsterRunStatus.CANCELING, DagsterRunStatus.CANCELED):
+            return False
         return self.count_resume_run_attempts(run_id) < self.run_monitoring_max_resume_run_attempts

python_modules/dagster/dagster_tests/daemon_tests/test_monitoring_daemon.py

@@ -286,6 +286,23 @@ def test_monitor_canceling(instance: DagsterInstance, logger: Logger):
     assert run.status == DagsterRunStatus.CANCELED
 
 
+def test_run_will_resume_with_canceling_run(instance: DagsterInstance):
+    # A started run that has not exhausted its resume attempts is resumable.
+    run = create_run_for_test(instance, job_name="foo", status=DagsterRunStatus.STARTED)
+    assert instance.run_will_resume(run.run_id)
+
+    # Once the user cancels the run it is no longer resumable, so the executor knows to
+    # forward the termination signal to in-flight steps instead of leaving them running.
+    report_canceling_event(instance, run, timestamp=time.time())
+    assert instance.get_run_by_id(run.run_id).status == DagsterRunStatus.CANCELING  # type: ignore
+    assert not instance.run_will_resume(run.run_id)
+
+
+def test_run_will_resume_with_canceled_run(instance: DagsterInstance):
+    run = create_run_for_test(instance, job_name="foo", status=DagsterRunStatus.CANCELED)
+    assert not instance.run_will_resume(run.run_id)
+
+
 def test_monitor_started(
     instance: DagsterInstance, workspace_context: WorkspaceProcessContext, logger: Logger
 ):