From 20869bdc1366b294967fb4f1d92a9e82b9f1a946 Mon Sep 17 00:00:00 2001 From: Jonathan Date: Wed, 15 Jul 2026 11:57:34 +0200 Subject: [PATCH] feat(worker): floor pause/delay/kill-switch helpers Co-Authored-By: Claude Opus 4.8 (1M context) --- worker/lyra_worker/floor.py | 42 +++++++++++++++++++++++++++++++ worker/tests/test_floor.py | 50 +++++++++++++++++++++++++++++++++++++ 2 files changed, 92 insertions(+) create mode 100644 worker/lyra_worker/floor.py create mode 100644 worker/tests/test_floor.py diff --git a/worker/lyra_worker/floor.py b/worker/lyra_worker/floor.py new file mode 100644 index 0000000..873aa62 --- /dev/null +++ b/worker/lyra_worker/floor.py @@ -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 diff --git a/worker/tests/test_floor.py b/worker/tests/test_floor.py new file mode 100644 index 0000000..88fd40e --- /dev/null +++ b/worker/tests/test_floor.py @@ -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 == []