From cb59afe443ded2dfc09e5f6560062576eb8831a8 Mon Sep 17 00:00:00 2001 From: Patrick Buckley Date: Tue, 7 Jul 2026 16:25:47 -0700 Subject: [PATCH] fix(nudge): log refused wakes; correct the already-dispatched hold-clear comment MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The wake gate documented exactly one info line per call past its gates, but a send() refusal (the authoritative under-lock _closed re-check catching a teardown the gate's lockless peek missed) emitted nothing — a dropped wake should stay traceable to its trigger, so the refusal now logs nudge_wake.refused. The already-dispatched branch's comment claimed a held reminder can coexist with the terminal mark via a redelivery whose commit raised — impossible with the current control flow (_redeliver_pending clears the hold before committing). Reworded to what the clear actually is: the last line of defense against any coexisting hold leaking forever once this branch deactivates the row, since inactive rows never re-list. Test comment updated to match. --- tests/test_idle_nudge_watcher.py | 20 ++++++++++++++++++++ tests/test_watch.py | 10 ++++++---- turnstone/core/idle_nudge_watcher.py | 7 +++++++ turnstone/core/watch.py | 13 ++++++++----- 4 files changed, 41 insertions(+), 9 deletions(-) diff --git a/tests/test_idle_nudge_watcher.py b/tests/test_idle_nudge_watcher.py index 8744107a..2c0b4a68 100644 --- a/tests/test_idle_nudge_watcher.py +++ b/tests/test_idle_nudge_watcher.py @@ -296,3 +296,23 @@ class TestWakeWorkstreamIfPending: assert deferred[0].levelno == logging.INFO assert "trigger=" in deferred[0].getMessage() assert not any("nudge_wake.dispatched" in r.getMessage() for r in caplog.records) + + def test_refused_path_logs_refusal(self, fake_mgr_and_ws, caplog): + """``send`` refusing outright — its authoritative under-lock + ``_closed`` re-check caught a teardown the gate's lockless peek + missed — emits ``nudge_wake.refused``: a dropped wake must stay + traceable to its trigger, not vanish silently.""" + _mgr, ws = fake_mgr_and_ws + ws.session._nudge_queue.enqueue("watch_triggered", "output", "any") + with ( + patch("turnstone.core.session_worker.send", return_value=False) as mock_send, + caplog.at_level(logging.INFO, logger="turnstone.core.idle_nudge_watcher"), + ): + assert wake_workstream_if_pending(ws, trigger="watch-fire") is False + assert mock_send.call_count == 1 + refused = [r for r in caplog.records if "nudge_wake.refused" in r.getMessage()] + assert len(refused) == 1 + assert refused[0].levelno == logging.INFO + assert "trigger=" in refused[0].getMessage() + assert not any("nudge_wake.dispatched" in r.getMessage() for r in caplog.records) + assert not any("nudge_wake.deferred_worker_busy" in r.getMessage() for r in caplog.records) diff --git a/tests/test_watch.py b/tests/test_watch.py index d7e7138c..9114b8fa 100644 --- a/tests/test_watch.py +++ b/tests/test_watch.py @@ -775,10 +775,12 @@ class TestWatchRunnerDeliveryRetry: assert "ws-1" not in runner._restoring def test_pending_cleared_on_already_dispatched_retry(self): - # A re-delivery that succeeded but whose row-commit raised leaves - # the id in BOTH _terminal_dispatched and _pending_delivery. The - # next tick's already-dispatched branch must clear the hold too, or - # it leaks once the row deactivates. + # Constructs the id-in-both-sets state DIRECTLY: no current path + # produces it (_redeliver_pending clears the hold before its + # commit), but the already-dispatched branch deactivates the row — + # after which it never re-lists — so it is the last line of + # defense against any such hold leaking forever. Pin that it + # clears the hold alongside the retry-deactivate. storage = MagicMock() storage.update_watch.return_value = True runner = self._make_runner(storage) diff --git a/turnstone/core/idle_nudge_watcher.py b/turnstone/core/idle_nudge_watcher.py index e92ac325..0758a22b 100644 --- a/turnstone/core/idle_nudge_watcher.py +++ b/turnstone/core/idle_nudge_watcher.py @@ -87,6 +87,11 @@ def wake_workstream_if_pending(ws: Workstream, *, trigger: str = "unspecified") queued; the owning worker's exit backstop (or its next drain seam) delivers it. * ``nudge_wake.dispatched`` — a fresh wake daemon was spawned. + * ``nudge_wake.refused`` — ``session_worker.send`` declined the + spawn: its authoritative under-lock ``_closed`` re-check caught a + teardown this gate's lockless peek missed. The entry dies with + the workstream; logged here so a dropped wake is traceable to its + trigger during production troubleshooting. Returns ``True`` iff the wake was handed to ``session_worker.send`` — which may still downgrade it to a no-op @@ -116,6 +121,8 @@ def wake_workstream_if_pending(ws: Workstream, *, trigger: str = "unspecified") log.info("nudge_wake.deferred_worker_busy ws=%s trigger=%s", ws.id[:8], trigger) elif ok: log.info("nudge_wake.dispatched ws=%s trigger=%s", ws.id[:8], trigger) + else: + log.info("nudge_wake.refused ws=%s trigger=%s", ws.id[:8], trigger) return ok diff --git a/turnstone/core/watch.py b/turnstone/core/watch.py index 7c2caae0..e29767aa 100644 --- a/turnstone/core/watch.py +++ b/turnstone/core/watch.py @@ -653,11 +653,14 @@ class WatchRunner: self._storage.update_watch(watch_id, active=False, next_poll="") with self._terminal_dispatched_lock: self._terminal_dispatched.discard(watch_id) - # A held reminder for an already-dispatched watch means a - # re-delivery SUCCEEDED but its row-commit raised (leaving - # the id in both sets). The reminder is delivered, so drop - # the hold here — otherwise, once we deactivate the row it - # never re-lists and the entry would leak forever. + # Belt-and-braces: no current path leaves an id in both + # ``_terminal_dispatched`` AND ``_pending_delivery`` + # (``_redeliver_pending`` clears the hold BEFORE its + # commit), but this branch deactivates the row — after + # which it never re-lists — so any hold that ever DID + # coexist with the terminal mark would leak forever + # without this clear. The mark means the reminder was + # delivered; a coexisting hold is by definition stale. self._clear_pending_delivery(watch_id) except Exception: log.exception("watch_runner.retry_deactivate_failed", extra={"watch_id": watch_id})