From 83efdf5e5e634e7a5b0b33e21e986eefb28d7142 Mon Sep 17 00:00:00 2001 From: Brooklyn Nicholson Date: Thu, 3 Sep 2026 12:03:58 +0700 Subject: [PATCH] [verified] fix(cron): close restart handoff races --- cron/delivery_queue.py | 102 ++++++++++++++++-- cron/executions.py | 23 ++-- gateway/run.py | 22 +++- tests/cron/test_delivery_queue.py | 56 ++++++++-- tests/cron/test_execution_ledger.py | 66 ++++++++++++ tests/cron/test_restart_safe_worker.py | 31 +++++- .../test_cron_delivery_housekeeping.py | 58 +++++++++- tools/process_registry.py | 2 +- 8 files changed, 329 insertions(+), 31 deletions(-) diff --git a/cron/delivery_queue.py b/cron/delivery_queue.py index 99757f7cdf..013bef5048 100644 --- a/cron/delivery_queue.py +++ b/cron/delivery_queue.py @@ -19,6 +19,7 @@ from contextlib import contextmanager from pathlib import Path from typing import Any, Callable, Iterator, Optional +from agent.redact import redact_sensitive_text from hermes_constants import get_hermes_home from hermes_time import now as _hermes_now @@ -28,6 +29,7 @@ _lock = threading.RLock() _ACTIVE_DELIVERIES: set[str] = set() _TERMINAL = ("delivered", "failed", "unknown") MAX_TERMINAL_DELIVERIES = 1000 +DEFAULT_DELIVERY_WAIT_TIMEOUT_SECONDS = 300.0 def _prune_terminal_unlocked(conn: sqlite3.Connection) -> None: @@ -46,6 +48,15 @@ def _prune_terminal_unlocked(conn: sqlite3.Connection) -> None: ) excess = terminal_count - keep if excess > 0: + conn.execute( + """INSERT OR IGNORE INTO delivery_tombstones + (execution_id, terminal_status, finished_at) + SELECT execution_id, status, finished_at FROM deliveries + WHERE status IN ('delivered','failed','unknown') + ORDER BY finished_at, created_at, execution_id + LIMIT ?""", + (excess,), + ) conn.execute( """DELETE FROM deliveries WHERE execution_id IN ( SELECT execution_id FROM deliveries @@ -92,6 +103,14 @@ def _transaction() -> Iterator[sqlite3.Connection]: error TEXT )""" ) + conn.execute( + """CREATE TABLE IF NOT EXISTS delivery_tombstones ( + execution_id TEXT PRIMARY KEY, + terminal_status TEXT NOT NULL CHECK(terminal_status IN + ('delivered','failed','unknown')), + finished_at TEXT + )""" + ) columns = { str(row[1]) for row in conn.execute("PRAGMA table_info(deliveries)") } @@ -138,6 +157,17 @@ def enqueue( ) -> dict: """Persist one idempotent delivery request before the worker waits.""" with _transaction() as conn: + tombstone = conn.execute( + "SELECT terminal_status, finished_at FROM delivery_tombstones " + "WHERE execution_id=?", + (str(execution_id),), + ).fetchone() + if tombstone is not None: + return { + "execution_id": str(execution_id), + "status": tombstone["terminal_status"], + "finished_at": tombstone["finished_at"], + } conn.execute( """INSERT OR IGNORE INTO deliveries (execution_id, job_json, content, for_failure, status, created_at) @@ -161,7 +191,21 @@ def get_status(execution_id: str) -> Optional[dict]: row = conn.execute( "SELECT * FROM deliveries WHERE execution_id=?", (str(execution_id),) ).fetchone() - return dict(row) if row is not None else None + if row is not None: + return dict(row) + tombstone = conn.execute( + "SELECT execution_id, terminal_status, finished_at " + "FROM delivery_tombstones WHERE execution_id=?", + (str(execution_id),), + ).fetchone() + if tombstone is None: + return None + return { + "execution_id": tombstone["execution_id"], + "status": tombstone["terminal_status"], + "finished_at": tombstone["finished_at"], + "error": None, + } def claim_next() -> Optional[dict]: @@ -194,6 +238,11 @@ def claim_next() -> Optional[dict]: def _finish(execution_id: str, *, error: Optional[str]) -> bool: status = "failed" if error else "delivered" + safe_error = ( + redact_sensitive_text(str(error), force=True, redact_url_credentials=True) + if error + else None + ) with _transaction() as conn: cur = conn.execute( """UPDATE deliveries SET status=?, finished_at=?, error=? @@ -202,7 +251,7 @@ def _finish(execution_id: str, *, error: Optional[str]) -> bool: ( status, _hermes_now().isoformat(), - error, + safe_error, execution_id, _PROCESS_ID, os.getpid(), @@ -275,6 +324,40 @@ def drain( return processed +def _terminalize_wait_timeout(execution_id: str) -> str: + """Fence a delivery whose worker can no longer wait for confirmation.""" + now = _hermes_now().isoformat() + pending_error = "timed out waiting for a live gateway; delivery was not attempted" + uncertain_error = ( + "timed out while gateway delivery was in progress; outcome is unknown and " + "was not retried" + ) + with _transaction() as conn: + cur = conn.execute( + """UPDATE deliveries SET status='failed', finished_at=?, error=? + WHERE execution_id=? AND status='pending'""", + (now, pending_error, str(execution_id)), + ) + if cur.rowcount: + _prune_terminal_unlocked(conn) + return pending_error + conn.execute( + """UPDATE deliveries SET status='unknown', finished_at=?, error=? + WHERE execution_id=? AND status='delivering'""", + (now, uncertain_error, str(execution_id)), + ) + row = conn.execute( + "SELECT status, error FROM deliveries WHERE execution_id=?", + (str(execution_id),), + ).fetchone() + _prune_terminal_unlocked(conn) + if row is None: + return "timed out waiting for live gateway delivery" + if row["status"] == "delivered": + return "" + return str(row["error"] or f"delivery {row['status']}") + + def enqueue_and_wait( execution_id: str, job: dict, @@ -284,13 +367,20 @@ def enqueue_and_wait( timeout: Optional[float] = None, ) -> Optional[str]: """Queue delivery and wait for a gateway's terminal at-most-once outcome.""" - enqueue(execution_id, job, content, for_failure=for_failure) - deadline = None if timeout is None else time.monotonic() + timeout - while deadline is None or time.monotonic() < deadline: + queued = enqueue(execution_id, job, content, for_failure=for_failure) + if queued["status"] in _TERMINAL: + return None if queued["status"] == "delivered" else str( + queued.get("error") or f"delivery {queued['status']}" + ) + wait_timeout = ( + DEFAULT_DELIVERY_WAIT_TIMEOUT_SECONDS if timeout is None else max(0.0, timeout) + ) + deadline = time.monotonic() + wait_timeout + while time.monotonic() < deadline: row = get_status(execution_id) if row and row["status"] in _TERMINAL: return None if row["status"] == "delivered" else str( row.get("error") or f"delivery {row['status']}" ) time.sleep(0.25) - return "timed out waiting for live gateway delivery; request remains pending" + return _terminalize_wait_timeout(execution_id) or None diff --git a/cron/executions.py b/cron/executions.py index 9cf204e559..3abd5f54df 100644 --- a/cron/executions.py +++ b/cron/executions.py @@ -148,7 +148,7 @@ def _prune_unlocked(conn: sqlite3.Connection) -> None: """DELETE FROM executions WHERE id IN ( SELECT id FROM executions WHERE status IN ('completed','failed','unknown') - ORDER BY claimed_at DESC, id DESC LIMIT -1 OFFSET ? + ORDER BY finished_at DESC, claimed_at DESC, id DESC LIMIT -1 OFFSET ? )""", (limit,), ) @@ -211,7 +211,7 @@ def adopt_claimed_execution(execution_id: str) -> Optional[Dict[str, Any]]: SET process_id=?, pid=?, process_started_at=?, status='running', started_at=?, handoff_pending=0, handoff_started_at=NULL - WHERE id=? AND status='claimed'""", + WHERE id=? AND status='claimed' AND handoff_pending=1""", (_PROCESS_ID, pid, process_started_at, now, execution_id), ) if cur.rowcount != 1: @@ -231,8 +231,9 @@ def mark_execution_running(execution_id: str) -> Optional[Dict[str, Any]]: """UPDATE executions SET status='running', started_at=?, handoff_pending=0, handoff_started_at=NULL - WHERE id=? AND status='claimed' AND handoff_pending=0""", - (now, execution_id), + WHERE id=? AND status='claimed' AND handoff_pending=0 + AND process_id=? AND pid=?""", + (now, execution_id, _PROCESS_ID, os.getpid()), ) if cur.rowcount != 1: return None @@ -256,8 +257,9 @@ def finish_execution( """UPDATE executions SET status=?, finished_at=?, error=?, handoff_pending=0, handoff_started_at=NULL - WHERE id=? AND status IN ('claimed','running')""", - (status, now, detail, execution_id), + WHERE id=? AND status IN ('claimed','running') + AND process_id=? AND pid=?""", + (status, now, detail, execution_id, _PROCESS_ID, os.getpid()), ) if cur.rowcount != 1: return None @@ -276,7 +278,7 @@ def recover_interrupted_executions() -> int: recovered: List[Dict[str, Any]] = [] with _transaction() as conn: rows = conn.execute( - """SELECT id, process_id, pid, process_started_at, + """SELECT id, status, process_id, pid, process_started_at, handoff_pending, handoff_started_at FROM executions WHERE status IN ('claimed','running')""" @@ -298,11 +300,14 @@ def recover_interrupted_executions() -> int: """UPDATE executions SET status='unknown', finished_at=?, error=?, handoff_pending=0, handoff_started_at=NULL - WHERE id=? AND status IN ('claimed','running')""", + WHERE id=? AND status=? AND process_id=? AND pid=? + AND handoff_pending=? + AND handoff_started_at IS ?""", (now, "Scheduler restarted after this execution's owner exited before a durable " "terminal state; whether side effects ran is unknown.", - row["id"]), + row["id"], row["status"], row["process_id"], row["pid"], + row["handoff_pending"], row["handoff_started_at"]), ) changed += cur.rowcount if cur.rowcount: diff --git a/gateway/run.py b/gateway/run.py index e17ab9eddf..22b9fa3265 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -33280,15 +33280,27 @@ def _drain_restart_safe_cron_deliveries(adapters, loop, runner=None) -> None: """Drain each profile's worker queue through its matching live adapters.""" from cron import scheduler as cron_scheduler - if adapters is not None: - cron_scheduler.drain_delivery_queue(adapters, loop) if runner is None: + if adapters is not None: + cron_scheduler.drain_delivery_queue(adapters, loop) return - for profile_name, profile_home in _handoff_watch_scopes(runner)[1:]: - profile_adapters = getattr(runner, "_profile_adapters", {}).get(profile_name) + for profile_name, profile_home in _handoff_watch_scopes(runner): + scoped_home = profile_home or get_hermes_home() + if profile_name is None: + profile_adapters = adapters + else: + profile_adapters = getattr(runner, "_profile_adapters", {}).get( + profile_name + ) if profile_adapters is None: continue - with _profile_runtime_scope(profile_home): + with _profile_runtime_scope(scoped_home): + if profile_name is not None and not profile_adapters and adapters: + routes = cron_scheduler._primary_profile_routes_for_current_home() + if routes: + profile_adapters = cron_scheduler.SharedRouteAdapters( + adapters, routes + ) cron_scheduler.drain_delivery_queue(profile_adapters, loop) diff --git a/tests/cron/test_delivery_queue.py b/tests/cron/test_delivery_queue.py index 67539a6ba3..b944fda224 100644 --- a/tests/cron/test_delivery_queue.py +++ b/tests/cron/test_delivery_queue.py @@ -35,11 +35,22 @@ def test_terminal_delivery_retention_is_bounded(tmp_path, monkeypatch): assert queue.claim_next()["execution_id"] == execution_id assert queue._finish(execution_id, error=None) - assert queue.get_status("exec-0") is None - assert queue.get_status("exec-1") is None + # Pruning may discard verbose outcome rows, but never the durable + # idempotency tombstone for an execution that could be replayed later. + pruned = queue.get_status("exec-0") + assert pruned is not None + assert pruned["status"] == "delivered" + assert queue.get_status("exec-1")["status"] == "delivered" assert queue.get_status("exec-2")["status"] == "delivered" assert queue.get_status("exec-3")["status"] == "delivered" + # Pruning may discard verbose outcome rows, but never the durable + # idempotency tombstone for an execution that could be replayed later. + queue.enqueue("exec-0", {"id": "job-replayed"}, "duplicate brief") + send = Mock(return_value=None) + assert queue.drain(send) == 0 + send.assert_not_called() + def test_failure_delivery_lane_survives_durable_handoff(tmp_path, monkeypatch): import cron.delivery_queue as queue @@ -92,6 +103,29 @@ def test_legacy_queue_schema_adds_failure_lane_before_enqueue(tmp_path, monkeypa assert queue.get_status("exec-migrated")["for_failure"] == 1 +def test_wait_timeout_marks_inflight_delivery_unknown_without_retry( + tmp_path, monkeypatch +): + import cron.delivery_queue as queue + + monkeypatch.setattr(queue, "DELIVERY_DB", tmp_path / "deliveries.db") + queue.enqueue("exec-inflight", {"id": "job-inflight"}, "result") + assert queue.claim_next() is not None + + error = queue.enqueue_and_wait( + "exec-inflight", {"id": "job-inflight"}, "result", timeout=0 + ) + + assert error is not None + assert "outcome is unknown" in error + status = queue.get_status("exec-inflight") + assert status is not None + assert status["status"] == "unknown" + send = Mock(return_value=None) + assert queue.drain(send) == 0 + send.assert_not_called() + + def test_dead_delivery_owner_becomes_unknown_and_is_not_retried( tmp_path, monkeypatch ): @@ -110,19 +144,23 @@ def test_dead_delivery_owner_becomes_unknown_and_is_not_retried( assert queue.get_status("exec-1")["status"] == "unknown" -def test_delivery_failure_is_terminal_and_not_retried(tmp_path, monkeypatch): +def test_delivery_failure_is_terminal_not_retried_and_redacted( + tmp_path, monkeypatch +): import cron.delivery_queue as queue monkeypatch.setattr(queue, "DELIVERY_DB", tmp_path / "deliveries.db") queue.enqueue("exec-1", {"id": "job-1"}, "brief") - send = Mock(return_value="transport failed") + send = Mock(return_value="request failed: https://example.test/?token=TOKEN123") assert queue.drain(send) == 1 assert queue.drain(send) == 0 assert send.call_count == 1 status = queue.get_status("exec-1") + assert status is not None assert status["status"] == "failed" - assert status["error"] == "transport failed" + assert "TOKEN123" not in status["error"] + assert "token=***" in status["error"] def test_wait_timeout_cancels_unclaimed_delivery(tmp_path, monkeypatch): @@ -134,10 +172,12 @@ def test_wait_timeout_cancels_unclaimed_delivery(tmp_path, monkeypatch): error = queue.enqueue_and_wait("exec-3", job, "result", timeout=0) assert "timed out" in error - assert queue.get_status("exec-3")["status"] == "pending" + status = queue.get_status("exec-3") + assert status is not None + assert status["status"] == "failed" send = Mock(return_value=None) - assert queue.drain(send) == 1 - send.assert_called_once() + assert queue.drain(send) == 0 + send.assert_not_called() def test_same_gateway_recovers_terminalization_failure_without_resending( diff --git a/tests/cron/test_execution_ledger.py b/tests/cron/test_execution_ledger.py index 641be1fdf9..d68259ae4b 100644 --- a/tests/cron/test_execution_ledger.py +++ b/tests/cron/test_execution_ledger.py @@ -87,6 +87,54 @@ def test_stale_external_handoff_is_recovered_unknown(monkeypatch, tmp_path): assert recovered["handoff_pending"] == 0 +def test_recovery_does_not_overwrite_concurrent_worker_adoption(monkeypatch, tmp_path): + executions = _point_ledger(monkeypatch, tmp_path) + record = executions.create_execution("adoption-race", source="builtin") + pending = executions.mark_execution_handoff_pending(record["id"]) + assert pending is not None + monkeypatch.setattr(executions, "_PROCESS_ID", "replacement-scheduler") + monkeypatch.setattr( + executions.time, + "time", + lambda: pending["handoff_started_at"] + + executions.HANDOFF_ADOPTION_GRACE_SECONDS + + 1, + ) + + def adopt_while_liveness_is_checked(_pid, _started_at): + monkeypatch.setattr(executions, "_PROCESS_ID", "external-worker") + monkeypatch.setattr(executions.os, "getpid", lambda: 4242) + monkeypatch.setattr(executions, "_process_start_time", lambda _pid: 9876) + assert executions.adopt_claimed_execution(record["id"]) is not None + return False + + monkeypatch.setattr(executions, "_owner_is_live", adopt_while_liveness_is_checked) + + assert executions.recover_interrupted_executions() == 0 + current = executions.get_execution(record["id"]) + assert current is not None + assert current["status"] == "running" + assert current["process_id"] == "external-worker" + assert current["pid"] == 4242 + + +def test_foreign_process_cannot_start_or_finish_execution(monkeypatch, tmp_path): + executions = _point_ledger(monkeypatch, tmp_path) + record = executions.create_execution("owner-fence", source="builtin") + original_process_id = executions._PROCESS_ID + original_pid = record["pid"] + + monkeypatch.setattr(executions, "_PROCESS_ID", "foreign-process") + monkeypatch.setattr(executions.os, "getpid", lambda: original_pid + 1) + assert executions.mark_execution_running(record["id"]) is None + assert executions.finish_execution(record["id"], success=True) is None + + monkeypatch.setattr(executions, "_PROCESS_ID", original_process_id) + monkeypatch.setattr(executions.os, "getpid", lambda: original_pid) + assert executions.mark_execution_running(record["id"]) is not None + assert executions.finish_execution(record["id"], success=True) is not None + + def test_execution_ledger_follows_the_current_profile_home(monkeypatch, tmp_path): import cron.executions as executions @@ -131,6 +179,24 @@ def test_retention_bounds_terminal_history_but_preserves_inflight(monkeypatch, t assert executions.latest_execution("live")["status"] == "running" +def test_recently_finished_long_running_execution_survives_retention( + monkeypatch, tmp_path +): + executions = _point_ledger(monkeypatch, tmp_path) + monkeypatch.setattr(executions, "MAX_TERMINAL_EXECUTIONS", 1) + long_running = executions.create_execution("long-running", source="builtin") + assert executions.mark_execution_running(long_running["id"]) is not None + newer = executions.create_execution("newer", source="builtin") + assert executions.finish_execution(newer["id"], success=True) is not None + + finished = executions.finish_execution(long_running["id"], success=True) + + assert finished is not None + assert finished["status"] == "completed" + assert executions.get_execution(long_running["id"])["status"] == "completed" + assert executions.get_execution(newer["id"]) is None + + def test_corrupt_store_fails_closed_without_overwrite(monkeypatch, tmp_path): executions = _point_ledger(monkeypatch, tmp_path) executions.EXECUTIONS_FILE.parent.mkdir(parents=True) diff --git a/tests/cron/test_restart_safe_worker.py b/tests/cron/test_restart_safe_worker.py index 8e1a2135d2..ad996fd56d 100644 --- a/tests/cron/test_restart_safe_worker.py +++ b/tests/cron/test_restart_safe_worker.py @@ -28,6 +28,7 @@ def test_execution_owner_moves_to_external_worker_before_running( execution_ledger, monkeypatch ): record = execution_ledger.create_execution("job-1", source="builtin") + assert execution_ledger.mark_execution_handoff_pending(record["id"]) is not None monkeypatch.setattr(execution_ledger.os, "getpid", lambda: 4242) monkeypatch.setattr(execution_ledger, "_process_start_time", lambda pid: 9876) @@ -41,10 +42,22 @@ def test_execution_owner_moves_to_external_worker_before_running( assert execution_ledger.mark_execution_running(record["id"]) is None +def test_external_worker_cannot_adopt_execution_without_handoff_fence( + execution_ledger, monkeypatch +): + record = execution_ledger.create_execution("job-unfenced", source="builtin") + monkeypatch.setattr(execution_ledger.os, "getpid", lambda: 4242) + monkeypatch.setattr(execution_ledger, "_process_start_time", lambda _pid: 9876) + + assert execution_ledger.adopt_claimed_execution(record["id"]) is None + assert execution_ledger.get_execution(record["id"])["status"] == "claimed" + + def test_genuine_external_worker_crash_is_recovered_unknown( execution_ledger, monkeypatch ): record = execution_ledger.create_execution("job-crash", source="builtin") + assert execution_ledger.mark_execution_handoff_pending(record["id"]) is not None script = ( "import os\n" "from pathlib import Path\n" @@ -64,10 +77,10 @@ def test_genuine_external_worker_crash_is_recovered_unknown( assert "whether side effects ran is unknown" in recovered["error"] +@pytest.mark.linux_only def test_restart_safe_gateway_child_fails_closed_without_scope(monkeypatch): import tools.process_registry as process_registry - monkeypatch.setattr(process_registry, "_IS_WINDOWS", False) monkeypatch.setattr(process_registry, "_is_supervised_gateway_process", lambda: True) monkeypatch.setenv("INVOCATION_ID", "managed-service") monkeypatch.setattr(process_registry, "_systemd_run_user_scope_available", lambda: False) @@ -89,6 +102,22 @@ def test_restart_safe_gateway_child_is_unchanged_outside_managed_gateway(monkeyp ) is command +def test_restart_safe_gateway_child_never_probes_systemd_off_linux(monkeypatch): + import tools.process_registry as process_registry + + command = ["python", "worker.py"] + probe = Mock(side_effect=AssertionError("systemd probe ran off Linux")) + monkeypatch.setattr(process_registry, "_IS_LINUX", False) + monkeypatch.setattr(process_registry, "_is_supervised_gateway_process", lambda: True) + monkeypatch.setattr(process_registry, "_systemd_run_user_scope_available", probe) + monkeypatch.setenv("INVOCATION_ID", "managed-service") + + assert process_registry.restart_safe_gateway_child_argv( + command, unit_suffix="cron-job-1" + ) is command + probe.assert_not_called() + + def test_external_worker_adopts_execution_and_runs_payload_once( tmp_path, monkeypatch ): diff --git a/tests/gateway/test_cron_delivery_housekeeping.py b/tests/gateway/test_cron_delivery_housekeeping.py index 9260c031fb..1228b04772 100644 --- a/tests/gateway/test_cron_delivery_housekeeping.py +++ b/tests/gateway/test_cron_delivery_housekeeping.py @@ -55,7 +55,7 @@ def test_gateway_housekeeping_drains_cron_delivery_without_connected_adapters(mo assert calls == [(adapters, loop)] -def test_multiplex_housekeeping_drains_each_profile_with_its_adapters( +def test_multiplex_housekeeping_scopes_primary_and_drains_each_profile( tmp_path, monkeypatch ): root_adapters = {} @@ -65,9 +65,12 @@ def test_multiplex_housekeeping_drains_each_profile_with_its_adapters( adapters=root_adapters, _profile_adapters={"secondary": secondary_adapters}, ) + root_home = tmp_path / "root" secondary_home = tmp_path / "secondary" calls = [] + monkeypatch.setattr(gateway_run, "get_hermes_home", lambda: root_home) + monkeypatch.setattr( gateway_run, "_handoff_watch_scopes", @@ -95,7 +98,60 @@ def test_multiplex_housekeeping_drains_each_profile_with_its_adapters( ) assert calls == [ + ("scope", root_home), ("drain", root_adapters), ("scope", secondary_home), ("drain", secondary_adapters), ] + + +def test_multiplex_housekeeping_uses_primary_routes_for_credentialless_satellite( + tmp_path, monkeypatch +): + root_adapters = {"slack": object()} + secondary_home = tmp_path / "secondary" + runner = SimpleNamespace( + config=SimpleNamespace(multiplex_profiles=True), + adapters=root_adapters, + _profile_adapters={"secondary": {}}, + ) + calls = [] + routed = object() + + monkeypatch.setattr( + gateway_run, + "_handoff_watch_scopes", + lambda _runner: [(None, None), ("secondary", secondary_home)], + ) + + @contextmanager + def fake_scope(_home): + yield + + class FakeSharedRouteAdapters: + def __new__(cls, adapters, routes): + calls.append(("routed", adapters, routes)) + return routed + + monkeypatch.setattr(gateway_run, "_profile_runtime_scope", fake_scope) + monkeypatch.setattr(scheduler, "SharedRouteAdapters", FakeSharedRouteAdapters) + monkeypatch.setattr( + scheduler, + "_primary_profile_routes_for_current_home", + lambda: ["route-to-secondary"], + ) + monkeypatch.setattr( + scheduler, + "drain_delivery_queue", + lambda adapters, _loop: calls.append(("drain", adapters)), + ) + + gateway_run._drain_restart_safe_cron_deliveries( + root_adapters, object(), runner + ) + + assert calls == [ + ("drain", root_adapters), + ("routed", root_adapters, ["route-to-secondary"]), + ("drain", routed), + ] diff --git a/tools/process_registry.py b/tools/process_registry.py index 4ec461df14..8c1d5b66c7 100644 --- a/tools/process_registry.py +++ b/tools/process_registry.py @@ -334,7 +334,7 @@ def restart_safe_gateway_child_argv( closed if it cannot be established. Standalone processes, non-systemd supervisors, and non-Linux hosts retain the direct command. """ - if _IS_WINDOWS: + if not _IS_LINUX: return command if not _is_supervised_gateway_process() or not os.environ.get("INVOCATION_ID"): return command