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

Code before:
            run_logger.debug(
                f"Flow run '{flow_run.id}' was deleted before state could be updated"
            )
        except Exception:
            run_logger.exception(f"Failed to update state of flow run '{flow_run.id}'")
        else:
            if state.is_crashed():
                run_logger.info(
                    f"Reported flow run '{flow_run.id}' as crashed: {message}"
                )
        return state

    async def _handle_cancellation_observer_failure(self) -> None:
        """Handle failure of the cancellation observer.

        This is called when both the websocket and polling mechanisms for detecting
        cancellation events have failed. Without cancellation observing, flow runs
        cannot be cancelled and may run indefinitely.

        Behavior depends on the `PREFECT_RUNNER_CRASH_ON_CANCELLATION_FAILURE` setting:
        - When enabled: kills all in-flight flow run processes, which triggers crash
          handling when the processes terminate
        - When disabled (default): logs an error but allows execution to continue
        """
        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)


Review comment (suggestion):
It's probably a good idea to set `self.stopping = True` so that new flow runs stop being submitted when in polling mode.

Code after:
            run_logger.debug(
                f"Flow run '{flow_run.id}' was deleted before state could be updated"
            )
        except Exception:
            run_logger.exception(f"Failed to update state of flow run '{flow_run.id}'")
        else:
            if state.is_crashed():
                run_logger.info(
                    f"Reported flow run '{flow_run.id}' as crashed: {message}"
                )
        return state

    async def _handle_cancellation_observer_failure(self) -> None:
        """Handle failure of the cancellation observer.

        This is called when both the websocket and polling mechanisms for detecting
        cancellation events have failed. Without cancellation observing, flow runs
        cannot be cancelled and may run indefinitely.

        Behavior depends on the `PREFECT_RUNNER_CRASH_ON_CANCELLATION_FAILURE` setting:
        - When enabled: kills all in-flight flow run processes, which triggers crash
          handling when the processes terminate, and stops the runner
        - When disabled (default): logs an error but allows execution to continue
        """
        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: