156 lines
5.0 KiB
Python
156 lines
5.0 KiB
Python
"""Hermes worker loop with lease renewal and bounded recovery."""
|
|
from __future__ import annotations
|
|
|
|
import threading
|
|
from dataclasses import dataclass
|
|
from typing import Any, Callable, Mapping
|
|
|
|
from .adapter import HermesAdapter, HermesConflict
|
|
from .contract import HermesJob, HermesJobState
|
|
|
|
|
|
JobHandler = Callable[[HermesJob], Mapping[str, Any]]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class WorkerRunResult:
|
|
claimed: bool
|
|
job_id: str | None = None
|
|
state: str | None = None
|
|
error: str | None = None
|
|
|
|
|
|
class HermesWorker:
|
|
"""Poll one tenant/cell/profile partition and execute jobs safely."""
|
|
|
|
def __init__(
|
|
self,
|
|
adapter: HermesAdapter,
|
|
handler: JobHandler,
|
|
*,
|
|
worker_id: str,
|
|
tenant_id: str,
|
|
cell_id: str,
|
|
profile_id: str,
|
|
poll_seconds: float = 1.0,
|
|
heartbeat_seconds: float = 10.0,
|
|
) -> None:
|
|
if poll_seconds <= 0 or heartbeat_seconds <= 0:
|
|
raise ValueError("poll_seconds and heartbeat_seconds must be positive")
|
|
for name, value in {
|
|
"worker_id": worker_id,
|
|
"tenant_id": tenant_id,
|
|
"cell_id": cell_id,
|
|
"profile_id": profile_id,
|
|
}.items():
|
|
if not str(value or "").strip():
|
|
raise ValueError(f"{name} is required")
|
|
self.adapter = adapter
|
|
self.handler = handler
|
|
self.worker_id = worker_id
|
|
self.tenant_id = tenant_id
|
|
self.cell_id = cell_id
|
|
self.profile_id = profile_id
|
|
self.poll_seconds = float(poll_seconds)
|
|
self.heartbeat_seconds = float(heartbeat_seconds)
|
|
self._stop = threading.Event()
|
|
|
|
def stop(self) -> None:
|
|
self._stop.set()
|
|
|
|
def run_once(self) -> WorkerRunResult:
|
|
job = self.adapter.claim(
|
|
tenant_id=self.tenant_id,
|
|
cell_id=self.cell_id,
|
|
profile_id=self.profile_id,
|
|
worker_id=self.worker_id,
|
|
)
|
|
if job is None:
|
|
return WorkerRunResult(claimed=False)
|
|
|
|
running = self.adapter.start(
|
|
job.job_id,
|
|
tenant_id=self.tenant_id,
|
|
worker_id=self.worker_id,
|
|
expected_revision=job.revision,
|
|
)
|
|
|
|
heartbeat_stop = threading.Event()
|
|
heartbeat_thread = threading.Thread(
|
|
target=self._heartbeat_loop,
|
|
args=(running.job_id, heartbeat_stop),
|
|
name=f"hermes-heartbeat-{self.worker_id}",
|
|
daemon=True,
|
|
)
|
|
heartbeat_thread.start()
|
|
|
|
output: Mapping[str, Any] | None = None
|
|
execution_error: Exception | None = None
|
|
try:
|
|
output = dict(self.handler(running))
|
|
except Exception as exc: # runtime boundary: normalize below
|
|
execution_error = exc
|
|
finally:
|
|
# Stop the lease heartbeat before reading the final revision. This
|
|
# prevents a revision increment racing with complete/fail.
|
|
heartbeat_stop.set()
|
|
heartbeat_thread.join(timeout=max(self.heartbeat_seconds * 2, 1.0))
|
|
|
|
try:
|
|
current = self.adapter.get(running.job_id, tenant_id=self.tenant_id)
|
|
if execution_error is None:
|
|
result = self.adapter.complete(
|
|
running.job_id,
|
|
tenant_id=self.tenant_id,
|
|
worker_id=self.worker_id,
|
|
output=dict(output or {}),
|
|
expected_revision=current.revision,
|
|
)
|
|
return WorkerRunResult(
|
|
claimed=True,
|
|
job_id=result.job_id,
|
|
state=result.state.value,
|
|
)
|
|
|
|
result = self.adapter.fail(
|
|
running.job_id,
|
|
tenant_id=self.tenant_id,
|
|
worker_id=self.worker_id,
|
|
error={
|
|
"code": "worker_error",
|
|
"message": str(execution_error),
|
|
},
|
|
expected_revision=current.revision,
|
|
)
|
|
return WorkerRunResult(
|
|
claimed=True,
|
|
job_id=result.job_id,
|
|
state=result.state.value,
|
|
error=str(execution_error),
|
|
)
|
|
except HermesConflict as exc:
|
|
return WorkerRunResult(
|
|
claimed=True,
|
|
job_id=running.job_id,
|
|
state=HermesJobState.FAILED.value,
|
|
error=f"reconciliation_conflict: {exc}",
|
|
)
|
|
|
|
def run_forever(self) -> None:
|
|
while not self._stop.is_set():
|
|
result = self.run_once()
|
|
if not result.claimed:
|
|
self._stop.wait(self.poll_seconds)
|
|
|
|
def _heartbeat_loop(self, job_id: str, stop_event: threading.Event) -> None:
|
|
while not stop_event.wait(self.heartbeat_seconds):
|
|
try:
|
|
self.adapter.heartbeat(
|
|
job_id,
|
|
tenant_id=self.tenant_id,
|
|
worker_id=self.worker_id,
|
|
)
|
|
except Exception:
|
|
stop_event.set()
|
|
return
|