feat(worker): floor pause/delay/kill-switch helpers
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,42 @@
|
||||
import os
|
||||
|
||||
from lyra_worker.config import get_config
|
||||
|
||||
_TRUE = {"1", "true", "yes", "on"}
|
||||
|
||||
|
||||
def press_paused(config: dict[str, str]) -> bool:
|
||||
"""True when the whole press is paused — the worker stops claiming new jobs."""
|
||||
return str(config.get("floor.paused", "")).strip().lower() in _TRUE
|
||||
|
||||
|
||||
def job_delay_seconds(config: dict[str, str]) -> float:
|
||||
"""Seconds to wait between finishing one download and claiming the next. Invalid or
|
||||
absent -> 0.0 (back-to-back). Negatives are clamped to 0."""
|
||||
try:
|
||||
return max(0.0, float(int(config.get("floor.jobDelaySeconds", "") or 0)))
|
||||
except (TypeError, ValueError):
|
||||
return 0.0
|
||||
|
||||
|
||||
def kill_requested(config: dict[str, str]) -> bool:
|
||||
"""True when a force-stop of the in-flight download has been requested."""
|
||||
return str(config.get("floor.killRequested", "")).strip().lower() in _TRUE
|
||||
|
||||
|
||||
def consume_kill_switch(conn, exit_fn=os._exit) -> bool:
|
||||
"""If a force-stop is pending, clear the flag and terminate the process (default
|
||||
os._exit) so the in-flight download is aborted. The container restart policy brings the
|
||||
worker back; startup clears staging and requeues the aborted job. Returns whether it
|
||||
fired. exit_fn is injectable for tests."""
|
||||
config = get_config(conn)
|
||||
if not kill_requested(config):
|
||||
return False
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(
|
||||
'UPDATE "Config" SET value = \'false\', "updatedAt" = now() WHERE key = %s',
|
||||
("floor.killRequested",),
|
||||
)
|
||||
conn.commit()
|
||||
exit_fn(0)
|
||||
return True
|
||||
@@ -0,0 +1,50 @@
|
||||
from lyra_worker.floor import (
|
||||
press_paused,
|
||||
job_delay_seconds,
|
||||
kill_requested,
|
||||
consume_kill_switch,
|
||||
)
|
||||
|
||||
|
||||
def test_press_paused_reads_truthy_values():
|
||||
assert press_paused({"floor.paused": "true"}) is True
|
||||
assert press_paused({"floor.paused": "ON"}) is True
|
||||
assert press_paused({"floor.paused": "false"}) is False
|
||||
assert press_paused({}) is False
|
||||
|
||||
|
||||
def test_job_delay_seconds_parses_and_defaults():
|
||||
assert job_delay_seconds({"floor.jobDelaySeconds": "30"}) == 30.0
|
||||
assert job_delay_seconds({}) == 0.0
|
||||
assert job_delay_seconds({"floor.jobDelaySeconds": "-5"}) == 0.0
|
||||
assert job_delay_seconds({"floor.jobDelaySeconds": "junk"}) == 0.0
|
||||
|
||||
|
||||
def test_kill_requested_reads_flag():
|
||||
assert kill_requested({"floor.killRequested": "true"}) is True
|
||||
assert kill_requested({}) is False
|
||||
|
||||
|
||||
def test_consume_kill_switch_clears_flag_and_exits(conn):
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(
|
||||
'INSERT INTO "Config"(key, value, secret, "updatedAt") '
|
||||
"VALUES ('floor.killRequested', 'true', false, now())"
|
||||
)
|
||||
conn.commit()
|
||||
|
||||
calls = []
|
||||
fired = consume_kill_switch(conn, exit_fn=lambda code: calls.append(code))
|
||||
|
||||
assert fired is True
|
||||
assert calls == [0]
|
||||
with conn.cursor() as cur:
|
||||
cur.execute("SELECT value FROM \"Config\" WHERE key = 'floor.killRequested'")
|
||||
assert cur.fetchone()[0] == "false"
|
||||
|
||||
|
||||
def test_consume_kill_switch_noop_when_unset(conn):
|
||||
calls = []
|
||||
fired = consume_kill_switch(conn, exit_fn=lambda code: calls.append(code))
|
||||
assert fired is False
|
||||
assert calls == []
|
||||
Reference in New Issue
Block a user