feat(worker): pause gate, inter-job delay, and force-stop watcher in main loop

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Jonathan
2026-07-15 12:06:40 +02:00
parent bff590a00f
commit 8c95ff735b
+30
View File
@@ -11,6 +11,7 @@ from lyra_worker.config import get_config
from lyra_worker.crypto import secret_key_problem
from lyra_worker.db import wait_for_db
from lyra_worker.discovery import DiscoveryConfig, DiscoveryResult, reset_pending, run_discovery
from lyra_worker.floor import press_paused, job_delay_seconds, consume_kill_switch
from lyra_worker.library import clear_staging_root
from lyra_worker.monitor import MonitorConfig, fulfill_owned_releases, reconcile, sweep
from lyra_worker.pipeline import run_pipeline
@@ -226,9 +227,30 @@ def _liveness_heartbeat() -> None:
time.sleep(HEARTBEAT_SECONDS)
KILL_POLL_SECONDS = 1.5
def _kill_switch_watch() -> None:
"""Poll floor.killRequested on a background daemon thread (its own DB connection) so a
force-stop aborts the process even while the main loop is blocked in a download. On any
DB hiccup, reconnect and keep watching."""
conn = wait_for_db()
while True:
try:
consume_kill_switch(conn)
except Exception:
try:
conn.close()
except Exception:
pass
conn = wait_for_db()
time.sleep(KILL_POLL_SECONDS)
def run_forever() -> None:
_require_secret_key()
threading.Thread(target=_liveness_heartbeat, daemon=True).start()
threading.Thread(target=_kill_switch_watch, daemon=True).start()
conn = wait_for_db()
clear_staging_root(STAGING_ROOT) # sweep any staging dirs orphaned by a prior crash
reclaimed = reclaim_stuck_jobs(conn) # requeue jobs a prior crash left mid-pipeline
@@ -252,6 +274,7 @@ def run_forever() -> None:
last_discover_tick = 0.0
last_heartbeat = 0.0
last_prune = 0.0
next_job_allowed = 0.0 # monotonic time before which no new job is claimed (inter-job delay)
try:
while True:
# A transient DB connection loss (e.g. AdminShutdown when Postgres restarts) raises
@@ -283,6 +306,12 @@ def run_forever() -> None:
maybe_run_scan(conn, resolver, probe, browser, config)
# Press paused (downloads only) or the inter-job delay hasn't elapsed: keep the
# loop alive (monitor/scan/discovery above still run) but claim nothing.
if press_paused(config) or now < next_job_allowed:
time.sleep(IDLE_SLEEP)
continue
job_id = claim_next(conn)
if job_id is None:
time.sleep(IDLE_SLEEP)
@@ -296,6 +325,7 @@ def run_forever() -> None:
conn.rollback()
finish_job(conn, job_id, mcfg)
print(f"worker: finished job {job_id}", flush=True)
next_job_allowed = time.monotonic() + job_delay_seconds(config)
except psycopg.OperationalError as e:
print(f"worker: db connection lost ({e}); reconnecting...", flush=True)
try: