Repository: PrefectHQ/prefect
File: src/prefect/runner/runner.py (Python)

Code before:
        will_crash = get_current_settings().runner.crash_on_cancellation_failure
        crash_message = (
            "Cancellation observing failed - both websocket and polling mechanisms "
            "are unavailable. Killing flow run process to prevent indefinite execution."
        )
        continue_message = (
            "Cancellation observing failed - both websocket and polling mechanisms "
            "are unavailable. Flow run will continue executing but cannot be "
            "cancelled. Set PREFECT_RUNNER_CRASH_ON_CANCELLATION_FAILURE=true to "
            "crash flow runs and shut down the runner when this occurs."
        )
        for flow_run_id, process_entry in self._flow_run_process_map.items():
            flow_run = process_entry["flow_run"]
            run_logger = self._get_flow_run_logger(flow_run)
            if will_crash:
                run_logger.error(crash_message)
                pid = process_entry.get("pid")
                if pid:
                    try:
                        await self._kill_process(pid)
                    except Exception:
                        run_logger.exception(
                            f"Failed to kill process {pid} for flow run '{flow_run.id}'"
                        )
            else:
                run_logger.warning(continue_message)

    async def _mark_flow_run_as_cancelled(
        self, flow_run: "FlowRun", state_updates: Optional[dict[str, Any]] = None
    ) -> None:
        state_updates = state_updates or {}
        state_updates.setdefault("name", "Cancelled")
        state_updates.setdefault("type", StateType.CANCELLED)
        state = (
            flow_run.state.model_copy(update=state_updates) if flow_run.state else None
        )
        if not state:
            self._logger.warning(
                f"Could not find state for flow run {flow_run.id} and cancellation cannot be guaranteed."
            )
            return

        try:
            await self._client.set_flow_run_state(flow_run.id, state, force=True)
        except ObjectNotFound:
            # Flow run was deleted - log it but don't crash the runner
            run_logger = self._get_flow_run_logger(flow_run)
            run_logger.debug(
                f"Flow run '{flow_run.id}' was deleted before it could be marked as cancelled"
            )


Review comment (bug):
I think, in most cases, just killing the process won't cause the `Runner` to mark the run as crashed, because it's indistinguishable from a cancelled run. I think you still need to explicitly mark each run as crashed here.

Code after:
        will_crash = get_current_settings().runner.crash_on_cancellation_failure
        crash_message = (
            "Cancellation observing failed - both websocket and polling mechanisms "
            "are unavailable. Killing flow run process to prevent indefinite execution."
        )
        continue_message = (
            "Cancellation observing failed - both websocket and polling mechanisms "
            "are unavailable. Flow run will continue executing but cannot be "
            "cancelled. Set PREFECT_RUNNER_CRASH_ON_CANCELLATION_FAILURE=true to "
            "crash flow runs and shut down the runner when this occurs."
        )

        if will_crash:
            self.stopping = True

        # Copy to list to avoid dictionary size changing during iteration
        process_entries = list(self._flow_run_process_map.items())
        for flow_run_id, process_entry in process_entries:
            flow_run = process_entry["flow_run"]
            run_logger = self._get_flow_run_logger(flow_run)
            if will_crash:
                run_logger.error(crash_message)
                pid = process_entry.get("pid")
                if pid:
                    try:
                        await self._kill_process(pid)
                    except Exception:
                        run_logger.exception(
                            f"Failed to kill process {pid} for flow run '{flow_run.id}'"
                        )
            else:
                run_logger.warning(continue_message)

    async def _mark_flow_run_as_cancelled(
        self, flow_run: "FlowRun", state_updates: Optional[dict[str, Any]] = None
    ) -> None:
        state_updates = state_updates or {}
        state_updates.setdefault("name", "Cancelled")
        state_updates.setdefault("type", StateType.CANCELLED)
        state = (
            flow_run.state.model_copy(update=state_updates) if flow_run.state else None
        )
        if not state:
            self._logger.warning(
                f"Could not find state for flow run {flow_run.id} and cancellation cannot be guaranteed."
            )
            return

        try:
            await self._client.set_flow_run_state(flow_run.id, state, force=True)
        except ObjectNotFound: