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

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 #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.

---------

Co-authored-by: SaarMesh-Bot <300107934+SaarMesh-Bot@users.noreply.github.com>
Co-authored-by: Claude <noreply@anthropic.com>

cmd/ingestor/mqtt_watchdog.go

@@ -74,6 +74,123 @@ func WatchdogPanicCount() int64 {
 	return watchdogPanicCount.Load()
 }
 
+// logQueueCapacity (#1749 root-cause fix) bounds the async emit queue
+// (see newAsyncEmit). Sized generously above any plausible per-scan
+// emit volume (a handful of sources x a couple of log lines each) so
+// legitimate bursts never spuriously drop, while still being small
+// enough that a permanently stuck writer fills it -- and starts
+// dropping -- within seconds rather than leaking memory unbounded.
+const logQueueCapacity = 256
+
+// watchdogLogDropCount (#1749 root-cause fix) counts emit() calls
+// dropped because the async log queue (see newAsyncEmit) was
+// saturated -- i.e. the background writer goroutine has not drained a
+// message in a while, almost certainly because its underlying write()
+// is itself blocked (Docker JSON-file log driver backpressure, full
+// stderr pipe, etc. -- the exact production shape from #1749: 3 MQTT
+// sources silent simultaneously, zero WATCHDOG log lines for 75+
+// minutes, container otherwise healthy, only a restart recovered).
+// Monotonic; 0 means the writer has never fallen behind.
+var watchdogLogDropCount atomic.Int64
+
+// WatchdogLogDropCount returns the running total of log lines dropped
+// by the async emit queue because the background writer could not
+// keep up (#1749 root-cause fix). Surfaced alongside
+// WatchdogLastTickUnix / WatchdogPanicCount so external monitoring can
+// distinguish "watchdog dead" (stale tick) from "watchdog alive, but
+// its log sink is stuck" (ticking normally, drop count climbing).
+//
+// Observability note: this is a raw monotonic counter, not a rate. A
+// dashboard alerting on "value > 0" will fire once and then stay
+// permanently red after the first drop, ever. A dashboard alerting on
+// "value == some threshold" will never fire again after crossing it.
+// The actionable signal is the counter's RATE OF INCREASE (e.g.
+// deriv() / increase() over a window in Prometheus terms) -- a
+// handful of drops during a brief legitimate burst is not the same
+// incident as hundreds of drops per minute because the writer has
+// been stuck for an hour. This function intentionally exposes only
+// the raw counter; rate-based alerting is a monitoring-stack
+// concern, not something this package should encode.
+func WatchdogLogDropCount() int64 {
+	return watchdogLogDropCount.Load()
+}
+
+// newAsyncEmit (#1749 root-cause fix) decouples "the watchdog decided
+// to log something" from "a log line was actually written".
+//
+// Background: PR #1810 added defer/recover around the per-source
+// watchdog work so a PANIC inside emit cannot kill the loop. That
+// does not cover the actual production failure mode: realEmit is
+// log.Print in production, and log.Print's underlying write() can
+// BLOCK -- rather than panic -- if the sink is backpressured (Docker's
+// 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 ever got
+// checked and no further tick was ever processed again. That is
+// exactly the #1749 incident shape: 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 (no
+// crash, no panic -- a hang), and only a full container restart
+// (which spawns a fresh watchdog goroutine) recovering it.
+//
+// The returned emit function performs a non-blocking channel send
+// and never waits on the real writer. A single background goroutine
+// drains the channel calling realEmit; if THAT goroutine itself gets
+// stuck on a slow/blocked write, the bounded channel simply fills up
+// and further sends fall through to the `default` branch -- counted
+// via watchdogLogDropCount -- instead of blocking. The watchdog tick
+// loop can therefore never be wedged by a stuck log sink, no matter
+// how long the sink stays stuck; worst case under a persistent
+// backpressure event is lost log lines (visible and counted), not a
+// silently dead watchdog (invisible and undetectable -- the actual
+// #1749 failure).
+//
+// stop() closes the queue, asking the writer goroutine to drain and
+// exit; it does NOT wait for the writer to finish (a stuck realEmit
+// mid-drain must not make shutdown hang either). Consequently, if
+// realEmit is genuinely stuck on a blocked syscall when stop() is
+// called, the writer goroutine leaks for the remaining lifetime of
+// the process -- close(queue) cannot interrupt a blocked write(), and
+// nothing here (or anywhere in the Go standard library) can force a
+// blocked syscall to return early. This is intentional: the whole
+// point of this function is that the watchdog must survive a stuck
+// writer, and a single leaked goroutine per stop() call is a vastly
+// smaller cost than a permanently wedged watchdog.
+//
+// This DOES mean newAsyncEmit is meant to be called once per process
+// lifetime, the way runLivenessWatchdog does, not repeatedly against
+// the same realEmit across many stop()/start() cycles within one
+// process (e.g. a test suite that starts and stops the watchdog
+// hundreds of times) -- each cycle against a stuck realEmit leaks
+// another writer goroutine, and nothing here bounds that
+// accumulation. Callers that need repeated start/stop within a single
+// process (tests) should either use a realEmit that is guaranteed to
+// return promptly, or track and bound the number of cycles.
+func newAsyncEmit(realEmit func(...any)) (emit func(...any), stop func()) {
+	queue := make(chan string, logQueueCapacity)
+	go func() {
+		for msg := range queue {
+			realEmit(msg)
+		}
+	}()
+	emit = func(args ...any) {
+		msg := fmt.Sprint(args...)
+		select {
+		case queue <- msg:
+		default:
+			watchdogLogDropCount.Add(1)
+		}
+	}
+	stop = func() {
+		close(queue)
+	}
+	return emit, stop
+}
+
 // LivenessKind enumerates the watchdog verdicts for a source. Edge-triggered
 // transitions use this to decide whether to emit (and what severity).
 type LivenessKind int
@@ -358,10 +475,30 @@ func SnapshotLivenessClocks() map[string]SourceLivenessSnapshot {
 func runLivenessWatchdog(interval, threshold time.Duration) (stop func()) {
 	t := time.NewTicker(interval)
 	done := make(chan struct{})
-	go runLivenessWatchdogLoop(t.C, done, threshold, log.Print)
+	loopExited := make(chan struct{})
+	// #1749 root-cause fix: wrap log.Print via newAsyncEmit rather than
+	// passing it directly. See newAsyncEmit's doc comment -- a
+	// synchronous log.Print can block forever on a stuck writer, and
+	// that used to wedge this entire goroutine (the production
+	// incident this fix addresses).
+	emit, stopEmit := newAsyncEmit(log.Print)
+	go func() {
+		defer close(loopExited)
+		runLivenessWatchdogLoop(t.C, done, threshold, emit)
+	}()
 	return func() {
 		t.Stop()
 		close(done)
+		// Wait for the loop goroutine to actually return before
+		// tearing down the async emit queue. Without this, the loop
+		// can still be mid processLivenessTransition -> emit() when
+		// stopEmit() closes the queue out from under it: emit()'s
+		// `queue <- msg` send races a closed channel and panics
+		// (send on closed channel). done being closed only makes the
+		// loop exit on its NEXT select check -- it does not abort
+		// in-flight per-source work already past that check.
+		<-loopExited
+		stopEmit()
 	}
 }
 

cmd/ingestor/mqtt_watchdog_1749_hang_test.go

@@ -0,0 +1,277 @@
+package main
+
+import (
+	"bytes"
+	"encoding/json"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+// Issue #1749 (root-cause update) — the panic-recovery + escalation
+// fixes from PR #1810 address a PANIC inside emit, but production
+// emit is log.Print, whose underlying write() can BLOCK rather than
+// panic (Docker JSON-file log driver backpressure, full stderr pipe,
+// etc.). A blocking write is not a panic — defer/recover does not
+// catch it — and because emit was called SYNCHRONOUSLY inside the
+// watchdog tick loop, a single stuck write froze checking every
+// source on every subsequent tick. This exactly reproduces the
+// original incident (3 sources silent simultaneously, zero WATCHDOG
+// log lines for 75+ minutes, container otherwise healthy, only a
+// restart recovered) even after #1810 landed.
+//
+// Fix: newAsyncEmit decouples "decide to log" from "perform the
+// write". The watchdog loop only ever does a non-blocking channel
+// send; a dedicated background goroutine performs the actual
+// (potentially blocking) write. If that goroutine itself wedges, the
+// bounded channel fills and further sends are DROPPED (counted via
+// WatchdogLogDropCount) instead of blocking — the watchdog tick loop
+// can never be blocked by a stuck log sink, no matter how long the
+// sink stays stuck.
+
+// TestNewAsyncEmit_NeverBlocksWhenWriterStuck_1749 (RED before the
+// fix — newAsyncEmit did not exist and emit was called directly):
+// floods emit() far past the queue capacity while the background
+// writer goroutine is permanently blocked on its first write. Every
+// call MUST return immediately.
+func TestNewAsyncEmit_NeverBlocksWhenWriterStuck_1749(t *testing.T) {
+	writerEntered := make(chan struct{})
+	release := make(chan struct{})
+	defer close(release)
+
+	var enteredOnce sync.Once
+	realEmit := func(args ...any) {
+		enteredOnce.Do(func() { close(writerEntered) })
+		<-release // simulates a permanently blocked write() (#1749)
+	}
+
+	emit, stop := newAsyncEmit(realEmit)
+	defer stop()
+
+	before := WatchdogLogDropCount()
+
+	// Every emit() call -- including the very first -- MUST return
+	// promptly regardless of whether the underlying writer is stuck.
+	// The whole burst (first call + flood past queue capacity) runs
+	// in one goroutine bounded by a single timeout, so a still-
+	// blocking emit() fails cleanly via t.Fatal instead of hanging
+	// the entire test binary until `go test -timeout` kills it.
+	done := make(chan struct{})
+	go func() {
+		emit("first message — picked up by the writer goroutine and blocks it")
+		for i := 0; i < logQueueCapacity+50; i++ {
+			emit("flood while writer is stuck")
+		}
+		close(done)
+	}()
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("emit() blocked despite a permanently stuck writer — the #1749 hang has regressed")
+	}
+
+	select {
+	case <-writerEntered:
+	case <-time.After(2 * time.Second):
+		t.Fatal("writer goroutine never received the first message")
+	}
+
+	if after := WatchdogLogDropCount(); after <= before {
+		t.Fatalf("expected WatchdogLogDropCount to advance once the async queue saturated; before=%d after=%d", before, after)
+	}
+}
+
+// TestMQTTStallWatchdog_LoopSurvivesStuckWriter_1749 is the
+// end-to-end regression: wires runLivenessWatchdogLoop exactly the
+// way runLivenessWatchdog does in production (via newAsyncEmit around
+// a realEmit that blocks forever on its very first call) and proves
+// the loop keeps ticking and keeps checking all registered sources
+// anyway — reproducing the #1749 incident shape (3 sources, one
+// shared watchdog goroutine) but asserting the FIXED behavior. On the
+// pre-fix code (emit called synchronously) the very first emit() call
+// would block forever and every subsequent tick would never be
+// consumed.
+func TestMQTTStallWatchdog_LoopSurvivesStuckWriter_1749(t *testing.T) {
+	defer snapshotAndResetRegistry(t)()
+
+	release := make(chan struct{})
+	defer close(release)
+	realEmit := func(args ...any) { <-release } // permanently stuck write()
+	emit, stop := newAsyncEmit(realEmit)
+	defer stop()
+
+	threshold := 100 * time.Millisecond
+	tags := []string{"src-a-1749hang", "src-b-1749hang", "src-c-1749hang"}
+	for _, tag := range tags {
+		s := &SourceLivenessState{
+			Tag:           tag,
+			Broker:        "tcp://" + tag + ":1883",
+			IsConnectedFn: func() bool { return true },
+		}
+		atomic.StoreInt64(&s.LastMessageUnix, time.Now().Add(-time.Hour).Unix())
+		atomic.StoreInt64(&s.StartedAt, time.Now().Add(-2*time.Hour).Unix())
+		if err := registerLivenessState(s); err != nil {
+			t.Fatalf("setup: %v", err)
+		}
+	}
+
+	tick, done, exited := setupWatchdogTestLoop(t, threshold, emit)
+	defer close(done)
+
+	base := time.Now()
+	// Drive several ticks. Every tick MUST be consumed promptly — on
+	// the pre-fix code, the very first tick's first emit() call (for
+	// whichever source the map iteration visits first) would block
+	// forever inside processLivenessTransition, and tick 2 would
+	// never be accepted.
+	for i := 0; i < 5; i++ {
+		sendTickOrFail(t, tick, base.Add(time.Duration(i)*time.Second), time.Second,
+			"stuck-writer regression tick")
+	}
+
+	select {
+	case <-exited:
+		t.Fatal("watchdog loop exited unexpectedly while writer was stuck")
+	default:
+	}
+}
+
+// TestRunLivenessWatchdog_ProductionWiringUsesAsyncEmit_1749 smoke-
+// tests the actual production entrypoint (not just the loop body) to
+// confirm it still starts, ticks, and stops cleanly now that its emit
+// is wrapped via newAsyncEmit. Pure wiring regression — no assertions
+// on log content, just that runLivenessWatchdog remains well-behaved.
+func TestRunLivenessWatchdog_ProductionWiringUsesAsyncEmit_1749(t *testing.T) {
+	defer snapshotAndResetRegistry(t)()
+
+	stop := runLivenessWatchdog(5*time.Millisecond, time.Minute)
+	time.Sleep(50 * time.Millisecond)
+	if got := WatchdogLastTickUnix(); got == 0 {
+		t.Fatalf("expected runLivenessWatchdog to have ticked at least once")
+	}
+
+	stopped := make(chan struct{})
+	go func() {
+		stop()
+		close(stopped)
+	}()
+	select {
+	case <-stopped:
+	case <-time.After(2 * time.Second):
+		t.Fatal("runLivenessWatchdog's stop() did not return — newAsyncEmit wiring must not hang shutdown")
+	}
+	// stop() signals the loop goroutine to exit but does not wait for
+	// it (see runLivenessWatchdog) — give it a moment to actually
+	// return before the deferred snapshotAndResetRegistry restores the
+	// REAL shared registry. Without this, a still-running loop
+	// iteration from THIS test's 5ms ticker can race the very next
+	// test's registration and double-process its source (observed:
+	// an unrelated throttle test intermittently saw 2 invocations
+	// instead of 1 when this smoke test ran immediately before it).
+	time.Sleep(50 * time.Millisecond)
+}
+
+// TestIngestorStatsSnapshot_WatchdogLogDropCountRoundTrip_1749
+// mirrors TestIngestorStatsSnapshot_WatchdogFieldsRoundTrip_1810 for
+// the new field: it must serialize through JSON and deserialize via
+// the server's envelope shape alongside the existing watchdog fields.
+func TestIngestorStatsSnapshot_WatchdogLogDropCountRoundTrip_1749(t *testing.T) {
+	snap := IngestorStatsSnapshot{
+		SampledAt:            "2026-07-18T12:30:00Z",
+		WatchdogLastTickUnix: 1752841800,
+		WatchdogPanicCount:   0,
+		WatchdogLogDropCount: 413,
+	}
+	b, err := json.Marshal(snap)
+	if err != nil {
+		t.Fatalf("marshal: %v", err)
+	}
+	if !bytes.Contains(b, []byte(`"watchdogLogDropCount":413`)) {
+		t.Fatalf("watchdogLogDropCount missing from JSON: %s", string(b))
+	}
+	var back IngestorStatsSnapshot
+	if err := json.Unmarshal(b, &back); err != nil {
+		t.Fatalf("unmarshal: %v", err)
+	}
+	if back.WatchdogLogDropCount != 413 {
+		t.Fatalf("round-trip mismatch: got %d, want 413", back.WatchdogLogDropCount)
+	}
+}
+
+// TestRunLivenessWatchdog_StopDoesNotRaceQueueClose_1749 is a targeted
+// regression for a review finding on this PR: runLivenessWatchdog's
+// stop() used to call close(done) followed immediately by stopEmit()
+// (which closes the async queue) WITHOUT waiting for the loop
+// goroutine to actually return. done being closed only makes the loop
+// exit on its NEXT select check -- it does not abort in-flight
+// per-source work already past that check. If the loop was mid
+// processLivenessTransition -> emit() (i.e. actively sending on the
+// queue) at the exact moment stopEmit() closed it, that send raced a
+// closed channel and panicked ("send on closed channel").
+//
+// To have any chance of hitting that window, a source must be
+// actively triggering emit() on essentially every tick (LivenessOK
+// never emits) and stop() must be called as close as possible to a
+// tick firing. This drives many rapid start/stop cycles against a
+// fast ticker with an always-stalled source and asserts no panic
+// escapes -- run with -race for the strongest signal, but the panic
+// itself does not require -race to reproduce.
+func TestRunLivenessWatchdog_StopDoesNotRaceQueueClose_1749(t *testing.T) {
+	defer snapshotAndResetRegistry(t)()
+
+	s := &SourceLivenessState{
+		Tag:           "src-race-1749",
+		Broker:        "tcp://src-race-1749:1883",
+		IsConnectedFn: func() bool { return true },
+	}
+	// Stalled from the start and re-armed every iteration below so
+	// every single tick has real emit-triggering work to do.
+	atomic.StoreInt64(&s.LastMessageUnix, time.Now().Add(-time.Hour).Unix())
+	atomic.StoreInt64(&s.StartedAt, time.Now().Add(-2*time.Hour).Unix())
+	if err := registerLivenessState(s); err != nil {
+		t.Fatalf("setup: %v", err)
+	}
+
+	panicsBefore := WatchdogPanicCount()
+
+	const iterations = 100
+	for i := 0; i < iterations; i++ {
+		// Re-arm: force LivenessStalled again (clear the edge-trigger
+		// cooldown) so this iteration's ticks have emit() work to do,
+		// same as production would after MarkReconnected.
+		atomic.StoreInt64(&s.LastAlertUnix, 0)
+		atomic.StoreInt64(&s.LastMessageUnix, time.Now().Add(-time.Hour).Unix())
+
+		stop := runLivenessWatchdog(time.Microsecond, time.Nanosecond)
+		// stop() is called essentially back-to-back with start,
+		// maximizing the chance the loop goroutine is caught mid
+		// per-source work (including an in-flight emit()) when
+		// shutdown begins -- exactly the window the fix must close.
+		// The 1ms pacing sleep is NOT part of the race window (it
+		// happens after stop() has already returned); it exists only
+		// to spread this test's goroutine churn out over ~200ms of
+		// wall time instead of a sub-millisecond burst, so it does
+		// not perturb the real-time throttle assertions in
+		// neighboring tests via OS scheduler pressure.
+		stop()
+		time.Sleep(3 * time.Millisecond)
+	}
+
+	// The per-source defer/recover from #1810 means a "send on closed
+	// channel" panic here does NOT crash the test binary -- it gets
+	// silently caught, logged, and counted. A bare "did not crash"
+	// assertion would therefore pass on the OLD, racy stop()
+	// implementation too (verified: the race reproduces reliably
+	// under this exact test with 2+ recovered panics per run on the
+	// pre-fix code). The real assertion is that WatchdogPanicCount
+	// must NOT have moved: ordinary, correctly-sequenced shutdown
+	// should never panic in the first place, recovered or not.
+	if got := WatchdogPanicCount(); got != panicsBefore {
+		t.Fatalf("WatchdogPanicCount advanced from %d to %d across %d start/stop cycles -- "+
+			"stop() is racing the async emit queue close against an in-flight emit() "+
+			"(send on closed channel, recovered by the #1810 per-source defer/recover "+
+			"but should never happen on a correctly sequenced shutdown)",
+			panicsBefore, got, iterations)
+	}
+}

cmd/ingestor/stats_file.go

@@ -81,6 +81,18 @@ type IngestorStatsSnapshot struct {
 	// a panic in emit / log sink). Monotonic; 0 means no recovered
 	// panics yet. Additive — omitempty so older server builds ignore.
 	WatchdogPanicCount int64 `json:"watchdogPanicCount,omitempty"`
+	// WatchdogLogDropCount (#1749 root-cause fix) is the running total
+	// of watchdog log lines dropped by the async emit queue because
+	// the background writer goroutine could not keep up — almost
+	// always because its underlying write() is itself stuck (Docker
+	// JSON-file log driver backpressure, full stderr pipe, etc.).
+	// Surfaced alongside WatchdogLastTickUnix / WatchdogPanicCount so
+	// external monitoring can distinguish "watchdog dead" (stale tick)
+	// from "watchdog alive, but its log sink is stuck" (ticking
+	// normally, drop count climbing). Monotonic; 0 means the writer
+	// has never fallen behind. Additive — omitempty so older server
+	// builds ignore it.
+	WatchdogLogDropCount int64 `json:"watchdogLogDropCount,omitempty"`
 }
 
 // SourceLivenessSnapshot is the per-source two-clock view exposed for
@@ -273,6 +285,7 @@ func StartStatsFileWriter(s *Store, interval time.Duration) {
 				SourceStatuses:       SnapshotSourceStatuses(tickAt),
 				WatchdogLastTickUnix: WatchdogLastTickUnix(),
 				WatchdogPanicCount:   WatchdogPanicCount(),
+				WatchdogLogDropCount: WatchdogLogDropCount(),
 			}
 			buf.Reset()
 			if err := enc.Encode(&snap); err != nil {

cmd/server/mqtt_status.go

@@ -123,6 +123,15 @@ type MqttStatusResponse struct {
 	// per-source processing is broken. 0 / omitted: no recovered
 	// panics OR older ingestor build.
 	WatchdogPanicCount int64 `json:"watchdogPanicCount,omitempty"`
+	// WatchdogLogDropCount (#1749 root-cause fix) is the running total
+	// of watchdog log lines dropped because the async emit queue's
+	// background writer could not keep up — almost always because its
+	// underlying write() is itself stuck (Docker JSON-file log driver
+	// backpressure, full stderr pipe, etc.). Distinguishes "watchdog
+	// dead" (WatchdogLastTickUnix stale) from "watchdog alive, but its
+	// log sink is stuck" (ticking normally, this climbing). 0 /
+	// omitted: writer never fell behind OR older ingestor build.
+	WatchdogLogDropCount int64 `json:"watchdogLogDropCount,omitempty"`
 }
 
 // ingestorMqttStatusEnvelope is the partial shape the server decodes from
@@ -132,6 +141,7 @@ type ingestorMqttStatusEnvelope struct {
 	SourceStatuses       []MqttSourceStatus `json:"source_statuses"`
 	WatchdogLastTickUnix int64              `json:"watchdogLastTickUnix"`
 	WatchdogPanicCount   int64              `json:"watchdogPanicCount"`
+	WatchdogLogDropCount int64              `json:"watchdogLogDropCount"`
 }
 
 // handleMqttStatus serves GET /api/mqtt/status. Reads the ingestor stats
@@ -153,6 +163,7 @@ func (s *Server) handleMqttStatus(w http.ResponseWriter, r *http.Request) {
 	resp.SampleAt = env.SampledAt
 	resp.WatchdogLastTickUnix = env.WatchdogLastTickUnix
 	resp.WatchdogPanicCount = env.WatchdogPanicCount
+	resp.WatchdogLogDropCount = env.WatchdogLogDropCount
 	for _, src := range env.SourceStatuses {
 		src.Broker = maskBrokerURL(src.Broker)
 		// Broker libraries occasionally quote the failing URL in the

cmd/server/mqtt_status_1749_test.go

@@ -0,0 +1,60 @@
+package main
+
+import (
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"testing"
+)
+
+// Issue #1749 (root-cause fix): /api/mqtt/status must surface
+// WatchdogLogDropCount alongside WatchdogLastTickUnix /
+// WatchdogPanicCount so external monitoring can distinguish "watchdog
+// dead" (stale tick) from "watchdog alive, but its log sink is stuck"
+// (ticking normally, drop count climbing) — the actual root cause
+// behind the original #1749 production incident, which the #1810
+// panic-recovery fix did not address (a blocked write() hangs, it
+// does not panic).
+func TestMqttStatus_ExposesWatchdogLogDropCount_1749(t *testing.T) {
+	tmp := t.TempDir()
+	statsPath := filepath.Join(tmp, "ingestor-stats.json")
+	t.Setenv("CORESCOPE_INGESTOR_STATS", statsPath)
+
+	stub := map[string]any{
+		"sampledAt":            "2026-07-18T12:30:00Z",
+		"source_statuses":      []map[string]any{},
+		"watchdogLastTickUnix": int64(1752841800),
+		"watchdogPanicCount":   int64(0),
+		"watchdogLogDropCount": int64(413),
+	}
+	data, err := json.Marshal(stub)
+	if err != nil {
+		t.Fatalf("marshal stub: %v", err)
+	}
+	if err := os.WriteFile(statsPath, data, 0o600); err != nil {
+		t.Fatalf("write stub: %v", err)
+	}
+
+	srv := &Server{}
+	req := httptest.NewRequest(http.MethodGet, "/api/mqtt/status", nil)
+	rec := httptest.NewRecorder()
+	srv.handleMqttStatus(rec, req)
+
+	if rec.Code != http.StatusOK {
+		t.Fatalf("status = %d, want 200; body=%s", rec.Code, rec.Body.String())
+	}
+	var resp MqttStatusResponse
+	if err := json.Unmarshal(rec.Body.Bytes(), &resp); err != nil {
+		t.Fatalf("unmarshal: %v; body=%s", err, rec.Body.String())
+	}
+	if resp.WatchdogLogDropCount != 413 {
+		t.Errorf("WatchdogLogDropCount = %d, want 413; body=%s", resp.WatchdogLogDropCount, rec.Body.String())
+	}
+	// Sanity: the two pre-existing watchdog fields must still round-trip
+	// alongside the new one (no accidental field clobbering).
+	if resp.WatchdogLastTickUnix != 1752841800 {
+		t.Errorf("WatchdogLastTickUnix = %d, want 1752841800; body=%s", resp.WatchdogLastTickUnix, rec.Body.String())
+	}
+}