fix(#1749): decouple watchdog emit from blocking I/O (root cause)

[SaarMesh-Bot]

Closes the gap left by #1810: that PR added defer/recover around the watchdog per-source work so a panic inside emit cannot kill the loop, but the actual production incident is caused by emit blocking, not panicking.

Root cause

In production emit is log.Print. log.Print's underlying write() can block indefinitely if the sink is backpressured (Docker JSON-file log driver falling behind under load, a full stderr pipe, journald hiccups, etc.). A blocked syscall is not a panic -- recover() does nothing for it.

Because emit was called synchronously inside the per-source work, a single stuck write() froze the entire tick loop forever -- no further source was ever checked and no further tick was ever processed again. This exactly reproduces the original #1749 incident even after #1810 landed: 3 independent MQTT sources going silent within ~60s of each other (one shared dependency -- the watchdog goroutine itself -- died, not 3 independent paho clients), zero WATCHDOG log lines for the rest of the 75-minute window, every other goroutine in the process continuing to run fine (a hang, not a crash), and only a full container restart recovering it.

Fix

newAsyncEmit decouples "decide to log" from "perform the write": the watchdog loop now only ever does a non-blocking channel send. A single background goroutine drains the channel and performs the (potentially blocking) write. If that goroutine itself gets stuck, the bounded queue (256) fills and further sends are dropped -- counted via the new WatchdogLogDropCount, surfaced through /api/mqtt/status and the ingestor stats snapshot alongside WatchdogLastTickUnix / WatchdogPanicCount. Worst case under a persistent backpressure event is now lost log lines (visible and counted), not a silently dead watchdog (invisible and undetectable -- the actual #1749 failure).

Tests

• TestNewAsyncEmit_NeverBlocksWhenWriterStuck_1749 -- floods emit() past queue capacity while the writer is permanently blocked; every call must return immediately and drops must be counted.
• TestMQTTStallWatchdog_LoopSurvivesStuckWriter_1749 -- end-to-end, wires runLivenessWatchdogLoop exactly as production does (via newAsyncEmit around a permanently-blocking realEmit) with 3 registered sources, reproducing the incident shape and asserting the loop keeps ticking regardless.
• TestRunLivenessWatchdog_ProductionWiringUsesAsyncEmit_1749 -- smoke-tests the real entrypoint starts, ticks, and stops cleanly.
• WatchdogLogDropCount round-trip tests in both the ingestor stats snapshot and the server's /api/mqtt/status handler, mirroring the existing WatchdogPanicCount coverage from fix(mqtt): escalate persistent paho disconnect + recover from emit panic + expose watchdog tick (#1749) #1810.

All pre-existing watchdog/liveness tests (#1749, #1810 r1, force-reconnect) continue to pass unmodified; full ingestor suite green (verified 5x consecutive runs for flake-freedom). Note: the server package has pre-existing test-suite-wide flakiness in unrelated TestHandleNodePaths_* tests (confirmed reproducible on unmodified master too, non-deterministic which subset fails per run) -- unrelated to this change and out of scope here.

[Kpa-clawbot]

Polish review @ a0e5a06e — parallel personas.

carmack (BLOCKER): stop() race — runLivenessWatchdog's returned stop does close(done); stopEmit() without waiting for the loop goroutine to actually exit. The loop can still be mid-processLivenessTransition calling emit() when stopEmit() closes the queue → send on closed channel panics (mqtt_watchdog.go:455-459 vs :150-158). Fix: either have the loop signal exit via a loopDone chan before stopEmit(), or make emit tolerate a closed queue (recover, or use a sentinel + sync.Once).

dijkstra (MAJOR): goroutine leak by design — if realEmit is stuck, the writer goroutine survives forever after stop() (close(queue) cannot interrupt a blocked syscall). Doc comment acknowledges "does NOT wait", but nothing bounds accumulation across repeated stop/start (test suites, runLivenessWatchdog re-entry). Consider documenting that newAsyncEmit is single-shot per process, or accepting a context.Context for the writer.

munger (MINOR): async emit converts an operator-visible hang into a silent drop counter. WatchdogLogDropCount is monotonic-only — no rate/window, no last-drop timestamp, not wired to /healthz. Dashboards won't fire on "drop count climbing"; add a bounded-window rate or a /healthz threshold before declaring #1749 truly observable.

kent-beck (MAJOR): red-commit bar violation — reverting the fix makes TestNewAsyncEmit_* and TestMQTTStallWatchdog_LoopSurvivesStuckWriter_* fail as compile errors (newAsyncEmit / logQueueCapacity / WatchdogLogDropCount undefined), not assertion failures. Per AGENTS.md § TDD red-commit bar this doesn't count as a valid failing test. Split: land a red commit adding stubs (returning direct-call emit, no-op stop, drop counter always 0) so the tests fail on the assertion (emit blocked / drop count didn't advance). Test 3 (round-trip) is fine.

Verdict: BLOCKER (stop-race panic).

[SaarMesh-Bot]

Thanks for the thorough review — all four points addressed, pushed as 4 separate red/green commits on top of the original squash so the fix history is auditable.

carmack (BLOCKER, stop()/queue-close race): Confirmed and fixed. runLivenessWatchdog's goroutine now signals its own exit via a loopExited channel; stop() waits on it before calling stopEmit(). The loop is therefore guaranteed to have fully returned -- including finishing any in-flight emit() call -- before the queue it sends on can possibly be closed. Added TestRunLivenessWatchdog_StopDoesNotRaceQueueClose_1749, which drives 100 rapid start/stop cycles against an always-stalled source: reliably reproduces 15-40 recovered "send on closed channel" panics per run on the old code, 0 on the fix. Worth noting for anyone re-checking: the existing #1810 per-source recover() does catch this panic, so it wasn't a process crash in practice -- but it was inflating WatchdogPanicCount and spamming recovery log lines on every ordinary shutdown that happened to race a tick, which is exactly the kind of noise that trains operators to ignore that counter.

dijkstra (MAJOR, goroutine leak by design): Agreed, and correct that nothing can force a blocked syscall to return early. Expanded newAsyncEmit's doc comment to state this explicitly: a genuinely stuck realEmit at stop() time leaks the writer goroutine for the remaining process lifetime, and the function is intended for once-per-process-lifetime use (as runLivenessWatchdog does), not repeated start/stop cycles against the same realEmit within one process. Didn't add a context.Context param -- it wouldn't actually unstick a blocked write() either, so it'd add API surface without solving the accumulation concern; documenting the constraint seemed more honest than a fix that only looks like one.

munger (MINOR, drop counter observability): Agreed. Doc comment now spells out that WatchdogLogDropCount is a raw monotonic counter, not a rate, and that alerting should be on rate-of-increase (deriv()/increase() over a window) rather than "value > 0" (fires once, stays red forever) or a fixed threshold (never fires again after crossing it). Left the actual rate-window logic out of this package on purpose -- that's a monitoring-stack concern, not something the watchdog itself should encode.

kent-beck (MAJOR, red-commit bar): Fair catch, redone properly. The branch is now 4 commits:

1. test(#1749): red-commit regression coverage for watchdog emit hang -- adds the API surface (newAsyncEmit, logQueueCapacity, WatchdogLogDropCount + status/stats plumbing) wired to a trivial synchronous-passthrough stub, plus the tests. Two tests are genuinely RED here (TestNewAsyncEmit_NeverBlocksWhenWriterStuck_1749, TestMQTTStallWatchdog_LoopSurvivesStuckWriter_1749), failing on real timeouts/assertions, not compile errors.
2. fix(#1749): implement newAsyncEmit -- swaps the stub for the real queue-based implementation. Both tests go green.
3. test(#1749): red-commit for stop()/newAsyncEmit shutdown race -- adds the carmack regression test against commit 2's code. RED (reliably reproduces the panic).
4. fix(#1749): eliminate stop()/newAsyncEmit shutdown race + review follow-ups -- the loopExited fix plus the dijkstra/munger doc updates. Green.

Reverting any single fix commit while keeping its paired test commit now fails on assertions, not on undefined symbols.

Full ingestor suite verified green across repeated runs at each stage. One pre-existing, timing-sensitive test (TestMQTTStallWatchdog_DisconnectedEscalationThrottled_1749) flakes at roughly the same rate with or without this change -- confirmed reproducible on unmodified master during review, unrelated to this PR.

[Kpa-clawbot]

Re-review @ a1586763 — parallel personas after 4 follow-up commits.

Prior findings — verification

• [carmack] BLOCKER (stop() race, send-on-closed panic) — FIXED. runLivenessWatchdog now signals loopExited from the loop goroutine and stop() blocks on <-loopExited before stopEmit() closes the queue (mqtt_watchdog.go:478-500). Race window is closed. TestRunLivenessWatchdog_StopDoesNotRaceQueueClose_1749 drives 100 rapid start/stop cycles against an always-stalled source and asserts WatchdogPanicCount does not move (recover-catches-nothing bar, not just "did not crash").

• [dijkstra] MAJOR (goroutine leak by design) — ACCEPTED with documented rationale. Not eliminated (impossible: close(chan) can't interrupt a blocked syscall), but the doc comment on newAsyncEmit now explicitly names the leak characteristic, scopes it to once-per-process use, and warns against repeated start/stop against the same realEmit inside one process. Correct engineering tradeoff — a leaked writer is strictly smaller than a wedged watchdog. Acceptable.

• [kent-beck] MAJOR (red-commit bar violation) — FIXED. Commit 1 (225fa5fe) message states both new-red tests fail on assertion/timeout, not compile error, via a stub newAsyncEmit returning zero-value; commit 3 (f1ef031f) similarly reds on WatchdogPanicCount assertion. Both meet AGENTS.md § TDD red-commit bar (compile + run + assert-fail).

• [munger] MINOR (WatchdogLogDropCount non-observable) — FIXED. Doc comment on WatchdogLogDropCount() explicitly labels it a raw monotonic counter, not a rate, and instructs operators to alert on rate-of-increase (Prometheus increase()/deriv()), not thresholds. Encodes the observability contract at the callsite.

Verdict: APPROVED. Zero must-fix remaining. Ready to merge once CI runs (currently action_required — fork PR awaiting workflow approval).