diff --git a/src/github_agent_bridge/executor.py b/src/github_agent_bridge/executor.py index ab02e84..05b9141 100644 --- a/src/github_agent_bridge/executor.py +++ b/src/github_agent_bridge/executor.py @@ -584,14 +584,27 @@ def _run_worker( self._loop(worker_id, retire_event) except BaseException as exc: self._set_worker_state(worker_id, "error", error=True) - self._record_worker_heartbeat(worker_id) - with self._worker_failure_lock: - self._worker_failures.append((worker_id, exc)) - self._request_shutdown() + self._record_worker_failure(worker_id, exc) finally: if heartbeat_stop_event is not None: heartbeat_stop_event.set() + def _run_heartbeat( + self, + worker_id: str, + heartbeat_stop_event: threading.Event, + ) -> None: + try: + self._heartbeat_loop(worker_id, heartbeat_stop_event) + except BaseException as exc: + self._set_worker_state(worker_id, "error", error=True) + self._record_worker_failure(f"{worker_id}/heartbeat", exc) + + def _record_worker_failure(self, worker_id: str, exc: BaseException) -> None: + with self._worker_failure_lock: + self._worker_failures.append((worker_id, exc)) + self._request_shutdown() + def _spawn_worker(self) -> _WorkerSlot: with self._worker_slots_lock: worker_number = self._next_worker_number @@ -601,7 +614,7 @@ def _spawn_worker(self) -> _WorkerSlot: heartbeat_stop_event = threading.Event() claim_lock = threading.Lock() heartbeat_thread = threading.Thread( - target=self._heartbeat_loop, + target=self._run_heartbeat, args=(worker_id, heartbeat_stop_event), daemon=True, name=f"gab-heartbeat-{worker_number}", diff --git a/tests/test_executor.py b/tests/test_executor.py index 90f37e4..6ac9423 100644 --- a/tests/test_executor.py +++ b/tests/test_executor.py @@ -1152,6 +1152,29 @@ def fail_heartbeat(*args, **kwargs): pool._heartbeat_loop("worker-test") +def test_run_stops_when_heartbeat_hits_fatal_storage_error(tmp_path, monkeypatch): + queue = JobQueue(tmp_path / "bridge.sqlite3") + pool = ExecutorPool( + queue, + Policy(), + RecordingDispatcher(), + config=ExecutorConfig(run_once=True), + ) + + def fail_heartbeat(*args, **kwargs): + raise sqlite3.OperationalError("disk I/O error") + + monkeypatch.setattr(queue, "record_worker_heartbeat", fail_heartbeat) + + with pytest.raises( + RuntimeError, + match=r"executor worker terminated unexpectedly: .*/heartbeat: OperationalError: disk I/O error", + ): + pool.run() + + assert pool.stop_event.is_set() is True + + def test_acknowledgement_loop_does_not_hide_non_contention_database_errors(tmp_path, monkeypatch): queue = JobQueue(tmp_path / "bridge.sqlite3") pool = ExecutorPool(queue, Policy(), RecordingDispatcher())