fix(hub): surface fanout slow-listener disconnects - #164
Conversation
1a16152 to
a3543b0
Compare
The fanout broadcaster disconnects a listener whose buffer is full rather than dropping its entries, but the only record of it was f.dropped -- a counter incremented at fanout.go and read nowhere else in the codebase. Operators had no way to know listeners were being kicked. Make the counter observable on two surfaces: - broadcast() warns on stderr at the moment of disconnect, through a new HubFanOutSlowListener format in internal/config/warn, so log aggregators can rate the event as it happens. - StatusResponse gains DroppedListeners, populated from a new droppedCount() accessor. ctx hub status prints "Dropped listeners: N" only when N > 0, so a healthy hub's output is unchanged. The counter moves to sync/atomic as the acceptance list asks: broadcast increments with atomic.AddUint64 (its return value is what the warning reports) and droppedCount reads with atomic.LoadUint64, so the Status RPC handler never contends with an in-flight broadcast. Every release target is 64-bit, so the struct-field alignment caveat for the raw atomic.*Uint64 helpers does not apply here. Pin the contract with two tests. TestFanOut_DisconnectsSlowListener asserts the channel closes, the subscriber is removed from f.subs, and the counter increments; verified by mutation -- removing the delete/close block fails it while the original three pass. TestFanOut_DroppedCountRaceWithBroadcast reads the counter from four goroutines while broadcast disconnects listeners, so -race fails if the counter stops being atomic; also verified by mutation. The rendered status line gets its own pin. desc.Text returns "" for an unknown key, so a renamed text key would blank the line silently; TestClusterStatus_DroppedListeners asserts the rendered count and TestClusterStatus_NoDroppedListeners asserts the omission at zero. Verified by mutation: renaming the key fails the first. Leaves fanOutBuffer alone -- tuning it before the counter is observable would be tuning blind. Closes ActiveMemory#94 Signed-off-by: CoderMungan <codermungan@gmail.com>
a3543b0 to
472b684
Compare
|
Hey @CoderMungan ; thanks for the hard work, here are some changes that need to be made. The observability plumbing this PR adds is well built: the warning The PR cannot merge as-is because the documentation it adds asserts a I verified all three failure legs empirically against head with a scratch
So the new failure-modes entry — "The client sees an EOF and reconnects To be fair, the busy-spin, the double-close panic, and the missing reconnect all More importantly, this PR is the one closing #94, whose entire point was that this
Other Findings that Need Fixes
|
Review of ActiveMemory#164 showed the disconnect this PR made observable was itself broken in two ways, so the counter would have ticked once and then the process holding it would have died. listenEntries received with `case entries := <-ch` and ignored the closed state. A closed channel is always receivable, so after broadcast disconnected a listener the handler drained the buffer and then spun on nil forever at full CPU: the RPC never returned, the client never saw the stream end, and it silently missed every entry published afterwards. It now receives with `entries, live` and returns errSlowListener -- a package-level ResourceExhausted sentinel over a new cfgHub.ErrSlowListener -- so the stream ends with a reason that reaches the client. unsubscribe closed unconditionally, and every Listen stream runs it via defer. Once broadcast had already closed that channel, the deferred close panicked with "close of closed channel"; the gRPC server is built with no recovery interceptor, so every real slow-listener disconnect was a pending hub-daemon crash. Membership in f.subs is now the open/closed record and unsubscribe is idempotent. Two smaller review points. logWarn.Warn ran inside the f.mu critical section, where a stalled stderr pipe would have frozen subscribe, unsubscribe, every publisher, and the Status RPC (which takes f.mu via count()); broadcast now splits into a locked deliver plus an unlocked warn loop. And `dropped` becomes an atomic.Uint64, which drops the 32-bit alignment caveat the raw atomic.*Uint64 helpers carried. Three new tests, each verified by mutation. TestFanOut_UnsubscribeAfterDisconnect panics without the membership guard. TestListenEntries_SlowListenerEndsStream drives the real handler with a stalled send and hangs to its deadline if the closed-channel check is reverted; it also pins that the already-buffered entries are delivered before the error. TestIntegration_SlowListenerReachesClient proves the same contract over a real gRPC stream: Client.Listen returns ResourceExhausted, which is what makes `ctx connection listen` exit non-zero instead of reporting success on a stream it no longer receives. TestListenEntries_ContextCancelEndsStream keeps the clean shutdown path pinned. Docs follow the mechanism rather than an invented recovery. The Slow Listener entry no longer claims the client "sees an EOF and reconnects"; reconnect is manual, and the failure-modes doc says so and names the duplicate-append consequence of the hardcoded sinceSequence=0. Also corrected while here, because the new text contradicted them: three claims across hub.md, hub-failure-modes.md and render/doc.go that the hub or store "deduplicates by entry ID" -- Store.Append assigns a new sequence unconditionally and render.appendShared appends without inspecting what is there. render/doc.go's WriteEntries example was stale against its signature, as internal/write/hub/doc.go's ClusterStatus example was. Spec: specs/fix-hub-fanout-drop-observability.md Signed-off-by: CoderMungan <codermungan@gmail.com>
|
Thanks for the review @josealekhine — you were right on all three The mechanism1. No EOF, silent loss. 2. Delayed daemon crash. 3. No client reconnect. Not implemented — taking you up on the Stream-level regression test
Plus Your other three
One thing I fixed that you didn't ask forThree places claimed the hub or store "deduplicates by entry ID" Verification
Still nothing here touching the shell, PowerShell, OpenCode-plugin, or |
2 participants
Closes #94.
Covers items #1 (make the drop-counter observable) and #2 (regression
test) from the issue. Item #3 (configurable
fanOutBuffer) is deliberatelyleft out — as the issue puts it, tuning the constant before the counter is
observable would be tuning blind.
#1 —
f.droppedis no longer deadThe counter is now readable on two surfaces, both options the issue listed:
Warning at the moment of disconnect.
broadcast()callslogWarn.Warnwhen it cuts a slow listener loose, using a new
HubFanOutSlowListenerformat constant in
internal/config/warn(same pattern as the existingHubReplicate*family — no literal in the hub package). The message carriesthe cumulative count so log aggregators can rate it:
Cumulative count on the Status RPC.
StatusResponsegainsDroppedListeners uint64(json:"dropped_listeners"), populated byhubStatusfrom a newdroppedCount()accessor.ctx hub statusprintsonly when the count is non-zero, so a healthy hub's output is byte-for-byte
what it was before and no existing test or doc example changes.
On
sync/atomicPer the third acceptance item,
broadcastincrements withatomic.AddUint64(its return value is what the warning reports) anddroppedCountreads withatomic.LoadUint64, so the Status RPC handlerreads the counter without contending with an in-flight broadcast.
Every release target in
hack/build-all.shis 64-bit (darwin, linux andwindows on amd64/arm64), so the struct-field alignment caveat that applies to
the raw
atomic.*Uint64helpers on 32-bit platforms doesn't bite here.#2 — regression test
TestFanOut_DisconnectsSlowListeneroverflows the buffer against a listenerthat never drains, then asserts all three halves of the contract: the channel
is closed, the subscriber is gone from
f.subs, and the counter incremented.TestFanOut_DroppedCountStartsAtZeropins the healthy path so the countercan't start drifting upward.
TestFanOut_DroppedCountRaceWithBroadcastcovers the concurrency theacceptance item is really about: four goroutines call
droppedCount()— theStatus RPC handler's read path — while
broadcastdisconnects listeners.I verified both by mutation rather than trusting them:
delete(f.subs, ch); close(ch)block failsTestFanOut_DisconnectsSlowListener(count = 1, want 0 after disconnectand
disconnected channel never closed) while the original three testsstill pass — the exact silent regression the issue describes;
f.dropped++/return f.droppedmakesgo test -racereportWARNING: DATA RACEand failTestFanOut_DroppedCountRaceWithBroadcast.The tests redirect
logWarn.SetSink(io.Discard)so the new warning doesn'tpollute test output.
The rendered line is pinned too
desc.Textreturns""for an unknown key, so a renamed text key would blankthe new
ctx hub statusline silently — the same shape of bug this issue isabout.
internal/write/hubgainsTestClusterStatus_DroppedListeners(asserts the rendered count) and
TestClusterStatus_NoDroppedListeners(asserts the omission at zero, and that the existing stats line survives).
Renaming
DescKeyWriteHubDroppedListenersfails the first one.Docs
docs/cli/hub.md— notes the conditionalDropped listeners:line.docs/operations/hub-failure-modes.md— new Slow Listener Disconnectedentry under Network, explaining that the disconnect is the loss-prevention
mechanism (the client reconnects with its last-seen sequence and the hub
replays), what the warning and counter mean, and when a climbing count is
worth acting on.
internal/hub/doc.go— the Concurrency section said "slow subscribers aredropped", which read as entry loss; corrected to describe the disconnect
and point at the counter.
Verification
Ran the CI commands verbatim:
CGO_ENABLED=0 go build ./...CGO_ENABLED=0 go test ./...go test -race ./internal/hub/CGO_ENABLED=0 go vet ./...golangci-lint runhack/lint-docstrings.shNothing here touches the shell, PowerShell, OpenCode-plugin, or VS Code
extension surfaces.