fix(gateway): log a cron store notice that fails to send
Gate finding C12 r3 #2 (quality): the notice listener dropped the future from run_coroutine_threadsafe, so a send that raised (adapter error, a profile scope that failed to bind) vanished without a trace. Keep the future and log failures from a done-callback, the same way plugin message injection does in gateway/run_plugin_injection.py.
This commit is contained in:
@@ -9,6 +9,7 @@ send path the state.db warning uses.
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import concurrent.futures
|
||||
import contextlib
|
||||
import logging
|
||||
import threading
|
||||
@@ -45,13 +46,23 @@ def install_cron_store_notices(runner, loop: asyncio.AbstractEventLoop) -> None:
|
||||
else:
|
||||
return # its outage was never announced: no lone "recovered"
|
||||
last_sent[store] = now
|
||||
asyncio.run_coroutine_threadsafe(send_cron_store_notice(runner, event, record), loop)
|
||||
future = asyncio.run_coroutine_threadsafe(send_cron_store_notice(runner, event, record), loop)
|
||||
future.add_done_callback(lambda done: _log_notice_failure(done, event, store))
|
||||
|
||||
set_transition_listener(on_transition)
|
||||
for record in degraded_records():
|
||||
on_transition("unwritable", record)
|
||||
|
||||
|
||||
def _log_notice_failure(done, event: str, store: str) -> None:
|
||||
try:
|
||||
done.result()
|
||||
except (asyncio.CancelledError, concurrent.futures.CancelledError):
|
||||
return
|
||||
except Exception: # the ticker thread already moved on: log, never raise
|
||||
logger.warning("Cron store %s notice for %s failed", event, store, exc_info=True)
|
||||
|
||||
|
||||
def _profile_home(profile):
|
||||
from hermes_constants import get_routing_process_hermes_home
|
||||
if profile is None:
|
||||
|
||||
@@ -225,7 +225,7 @@ def test_unwritable_store_degrades_once_throttles_and_catches_up_once(cron_store
|
||||
|
||||
@pytest.mark.platforms("posix") # POSIX mode bits; root ignores them
|
||||
@pytest.mark.skipif(hasattr(os, "geteuid") and os.geteuid() == 0, reason="root bypasses directory modes")
|
||||
def test_unwritable_store_is_shown_in_cron_status_and_announced_once(cron_store, monkeypatch, capsys):
|
||||
def test_unwritable_store_is_shown_in_cron_status_and_announced_once(cron_store, monkeypatch, capsys, caplog):
|
||||
"""`hermes cron status` probes the store itself (a real 0500 dir) and leads with the red
|
||||
headline + fix; the gateway posts ONE home-channel notice on entry and ONE on recovery,
|
||||
each naming the store, the error, since-when and the skipped runs; re-entry within the hour
|
||||
@@ -267,9 +267,12 @@ def test_unwritable_store_is_shown_in_cron_status_and_announced_once(cron_store,
|
||||
save_jobs([_due_job()])
|
||||
cronjobs.warn_store_unwritable(enospc, "x", "scan", [_due_job()]) # flaps back within the hour
|
||||
loop.run_until_complete(asyncio.sleep(0.05))
|
||||
install_cron_store_notices(SimpleNamespace(), loop) # a send that raises is logged, not lost
|
||||
loop.run_until_complete(asyncio.sleep(0.05))
|
||||
finally:
|
||||
loop.close()
|
||||
assert len(sent) == 2
|
||||
assert str(cron_dir) in sent[0] and "ENOSPC: No space left on device" in sent[0] and "since " in sent[0]
|
||||
assert f"fix permissions on {cron_dir}" in sent[0]
|
||||
assert "writable again; 1 skipped run(s), catching up once per job" in sent[1]
|
||||
assert any(r.levelname == "WARNING" and "unwritable notice for" in r.getMessage() for r in caplog.records)
|
||||
|
||||
Reference in New Issue
Block a user