From 472b6845f710815ce6fd95d69f1c92aaa607e86b Mon Sep 17 00:00:00 2001 From: CoderMungan Date: Thu, 27 Aug 2026 10:52:25 +0300 Subject: [PATCH 1/2] fix(hub): surface fanout slow-listener disconnects 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 #94 Signed-off-by: CoderMungan --- docs/cli/hub.md | 6 ++ docs/operations/hub-failure-modes.md | 20 +++++ internal/assets/commands/text/write.yaml | 2 + internal/cli/hub/core/status/status.go | 1 + internal/config/embed/text/write_hub.go | 4 + internal/config/warn/warn.go | 9 ++ internal/hub/doc.go | 8 +- internal/hub/fanout.go | 27 +++++- internal/hub/fanout_test.go | 106 +++++++++++++++++++++++ internal/hub/handler.go | 1 + internal/hub/types.go | 6 +- internal/write/hub/hub.go | 12 ++- internal/write/hub/hub_test.go | 53 ++++++++++++ internal/write/hub/testmain_test.go | 21 +++++ 14 files changed, 270 insertions(+), 6 deletions(-) create mode 100644 internal/write/hub/hub_test.go create mode 100644 internal/write/hub/testmain_test.go diff --git a/docs/cli/hub.md b/docs/cli/hub.md index 57ac34509..f8233c637 100644 --- a/docs/cli/hub.md +++ b/docs/cli/hub.md @@ -114,6 +114,12 @@ Safe to rerun: if no daemon is running, returns a Show cluster status: role, peers, sync state, entry count, and uptime. +When the hub has disconnected any slow listeners, the output +gains a `Dropped listeners:` line with the cumulative count. +The line is omitted while that count is zero, so a healthy hub +looks exactly as it did before. See +[Slow Listener Disconnected](../operations/hub-failure-modes.md#slow-listener-disconnected). + **Examples**: ```bash diff --git a/docs/operations/hub-failure-modes.md b/docs/operations/hub-failure-modes.md index eea628bb6..a8c409efe 100644 --- a/docs/operations/hub-failure-modes.md +++ b/docs/operations/hub-failure-modes.md @@ -34,6 +34,26 @@ its last-seen sequence; the hub replays everything newer. **What you should do:** nothing. If reconnects are looping, check firewall state on the hub and `ctx hub status` output. +### Slow Listener Disconnected + +**What happens:** each `ctx connection listen` stream gets a +buffered fan-out channel. A client that stops draining it (paused +process, saturated link, a laptop that went to sleep) fills the +buffer. Rather than block every publisher or silently discard the +client's entries, the hub disconnects that one listener and closes +its channel. The client sees an EOF and reconnects with its +last-seen sequence, so the missed entries are replayed. Nothing is +lost; the reconnect is the recovery. + +Each disconnect writes a warning to the hub's stderr and increments +a cumulative counter reported as `Dropped listeners:` in +`ctx hub status`. + +**What you should do:** an occasional disconnect is normal and +self-healing. A count that climbs steadily means listeners cannot +keep up with the publish rate — check the listening client's health +and the link to it before assuming the hub is at fault. + ### Partition: Majority Side Reachable **What happens:** clients routed to the majority side continue to diff --git a/internal/assets/commands/text/write.yaml b/internal/assets/commands/text/write.yaml index 8548b3ec6..1e1a4c4d6 100644 --- a/internal/assets/commands/text/write.yaml +++ b/internal/assets/commands/text/write.yaml @@ -1125,6 +1125,8 @@ write.connect-hub-stats: short: 'Entries: %d Clients: %d' write.hub-cluster-stats: short: 'Entries: %d Peers: %d' +write.hub-dropped-listeners: + short: 'Dropped listeners: %d (slow subscribers disconnected)' write.agent-section-hub: short: "## ctx Hub" write.connect-hub-sync: diff --git a/internal/cli/hub/core/status/status.go b/internal/cli/hub/core/status/status.go index db2e05e54..8412506f8 100644 --- a/internal/cli/hub/core/status/status.go +++ b/internal/cli/hub/core/status/status.go @@ -61,6 +61,7 @@ func Run(cmd *cobra.Command, _ []string) error { cmd, role, cfg.HubAddr, resp.TotalEntries, len(resp.EntriesByProject), + resp.DroppedListeners, ) return nil } diff --git a/internal/config/embed/text/write_hub.go b/internal/config/embed/text/write_hub.go index 2654d7be3..d20923ebc 100644 --- a/internal/config/embed/text/write_hub.go +++ b/internal/config/embed/text/write_hub.go @@ -26,6 +26,10 @@ const ( // DescKeyWriteHubClusterStats is the text key for hub // cluster statistics. DescKeyWriteHubClusterStats = "write.hub-cluster-stats" + // DescKeyWriteHubDroppedListeners is the text key for the + // cumulative slow-listener disconnect count. Printed only + // when the count is non-zero. + DescKeyWriteHubDroppedListeners = "write.hub-dropped-listeners" // DescKeyWriteHubRevoked is the text key for the hub client // revocation confirmation. DescKeyWriteHubRevoked = "write.hub-revoked" diff --git a/internal/config/warn/warn.go b/internal/config/warn/warn.go index e31af0b96..a574d9c4e 100644 --- a/internal/config/warn/warn.go +++ b/internal/config/warn/warn.go @@ -113,6 +113,15 @@ const ( // not vanish. CloseHubClient = "close hub client: %v" + // HubFanOutSlowListener is the stderr format for a listener + // disconnected because its fan-out buffer was full. Takes the + // cumulative disconnect count. The broadcaster cannot block on + // a slow subscriber and will not drop entries silently, so the + // listener is cut loose instead; without this warning the only + // record of it was a counter nothing read. + HubFanOutSlowListener = "hub fanout: disconnected slow listener " + + "(buffer full); cumulative disconnects: %d" + // HubReplicateAppend is the stderr format for a failed // [Store.Append] inside the follower replication stream. The // loop is best-effort and has no return path, so a dropped diff --git a/internal/hub/doc.go b/internal/hub/doc.go index a3474c8f7..3465b7cf8 100644 --- a/internal/hub/doc.go +++ b/internal/hub/doc.go @@ -62,8 +62,12 @@ // // [Store] guards its indexes and appender with a // single mutex. Listen streams subscribe to a -// fan-out channel; slow subscribers are dropped -// rather than blocking publishers. +// fan-out channel; a subscriber that lets its buffer +// fill is disconnected rather than blocking +// publishers or silently losing entries. Each +// disconnect warns on stderr and bumps a cumulative +// counter reported as DroppedListeners by the Status +// RPC. // // # Encryption // diff --git a/internal/hub/fanout.go b/internal/hub/fanout.go index 09b510269..18c52ec07 100644 --- a/internal/hub/fanout.go +++ b/internal/hub/fanout.go @@ -6,6 +6,13 @@ package hub +import ( + "sync/atomic" + + cfgWarn "github.com/ActiveMemory/ctx/internal/config/warn" + logWarn "github.com/ActiveMemory/ctx/internal/log/warn" +) + // fanOutBuffer is the channel buffer size for each listener. const fanOutBuffer = 64 @@ -47,7 +54,9 @@ func (f *fanOut) unsubscribe(ch chan []Entry) { // broadcast sends entries to all active listeners. // Non-blocking: slow listeners get disconnected to prevent -// unbounded buffering. +// unbounded buffering. Each disconnect emits a warning so the +// event is visible to operators rather than only bumping a +// counter. // // Parameters: // - entries: entries to deliver to all subscribers @@ -62,7 +71,10 @@ func (f *fanOut) broadcast(entries []Entry) { // Slow listener: disconnect to prevent loss. delete(f.subs, ch) close(ch) - f.dropped++ + logWarn.Warn( + cfgWarn.HubFanOutSlowListener, + atomic.AddUint64(&f.dropped, 1), + ) } } } @@ -80,3 +92,14 @@ func (f *fanOut) count() uint32 { } return uint32(n) //nolint:gosec // len is non-negative } + +// droppedCount returns the cumulative number of listeners +// disconnected for being too slow. The read is atomic rather +// than mutex-guarded so the Status RPC handler never contends +// with an in-flight broadcast. +// +// Returns: +// - uint64: cumulative slow-listener disconnects +func (f *fanOut) droppedCount() uint64 { + return atomic.LoadUint64(&f.dropped) +} diff --git a/internal/hub/fanout_test.go b/internal/hub/fanout_test.go index cd119bee5..fbe0c7486 100644 --- a/internal/hub/fanout_test.go +++ b/internal/hub/fanout_test.go @@ -7,8 +7,13 @@ package hub import ( + "fmt" + "io" + "sync" "testing" "time" + + logWarn "github.com/ActiveMemory/ctx/internal/log/warn" ) func TestFanOut_SubscribeAndBroadcast(t *testing.T) { @@ -61,3 +66,104 @@ func TestFanOut_BroadcastToNone(t *testing.T) { // Should not panic. fo.broadcast([]Entry{{ID: "noop"}}) } + +func TestFanOut_DisconnectsSlowListener(t *testing.T) { + // The disconnect warns on stderr; keep test output clean. + restore := logWarn.SetSink(io.Discard) + defer restore() + + fo := newFanOut() + slow := fo.subscribe() + + // Never read from slow. One broadcast past the buffer has + // nowhere to go, so the listener is disconnected. + for i := 0; i < fanOutBuffer+1; i++ { + fo.broadcast([]Entry{{ID: fmt.Sprintf("e%d", i)}}) + } + + if got := fo.count(); got != 0 { + t.Errorf("count = %d, want 0 after disconnect", got) + } + if got := fo.droppedCount(); got != 1 { + t.Errorf("droppedCount = %d, want 1", got) + } + + // Drain the buffered entries, then observe the close. + deadline := time.After(time.Second) + for i := 0; i <= fanOutBuffer; i++ { + select { + case _, ok := <-slow: + if !ok { + return + } + case <-deadline: + t.Fatal("disconnected channel never closed") + } + } + select { + case _, ok := <-slow: + if ok { + t.Fatal("channel still open after disconnect") + } + case <-deadline: + t.Fatal("disconnected channel never closed") + } +} + +func TestFanOut_DroppedCountStartsAtZero(t *testing.T) { + fo := newFanOut() + ch := fo.subscribe() + fo.broadcast([]Entry{{ID: "x"}}) + <-ch + + if got := fo.droppedCount(); got != 0 { + t.Errorf("droppedCount = %d, want 0 for a healthy listener", + got) + } +} + +// TestFanOut_DroppedCountRaceWithBroadcast exercises the read +// path the Status RPC handler uses: droppedCount from another +// goroutine while broadcast is disconnecting listeners. Run +// under -race, it fails if the counter stops being atomic. +func TestFanOut_DroppedCountRaceWithBroadcast(t *testing.T) { + restore := logWarn.SetSink(io.Discard) + defer restore() + + fo := newFanOut() + + var wg sync.WaitGroup + done := make(chan struct{}) + + // Readers stand in for concurrent Status RPC handlers. + for i := 0; i < 4; i++ { + wg.Add(1) + go func() { + defer wg.Done() + for { + select { + case <-done: + return + default: + _ = fo.droppedCount() + } + } + }() + } + + // Each round subscribes a listener that never drains, then + // overflows it so broadcast disconnects it. + for round := 0; round < 20; round++ { + fo.subscribe() + for i := 0; i <= fanOutBuffer; i++ { + fo.broadcast([]Entry{{ID: fmt.Sprintf("r%d-%d", round, i)}}) + } + } + + close(done) + wg.Wait() + + if got := fo.droppedCount(); got != 20 { + t.Errorf("droppedCount = %d, want 20", got) + } +} diff --git a/internal/hub/handler.go b/internal/hub/handler.go index 5857596f0..8371111ea 100644 --- a/internal/hub/handler.go +++ b/internal/hub/handler.go @@ -245,6 +245,7 @@ func (s *Server) hubStatus( return &StatusResponse{ TotalEntries: total, ConnectedClients: s.listeners.count(), + DroppedListeners: s.listeners.droppedCount(), EntriesByType: byType, EntriesByProject: byProject, }, nil diff --git a/internal/hub/types.go b/internal/hub/types.go index ab5530e1d..3500bd4c9 100644 --- a/internal/hub/types.go +++ b/internal/hub/types.go @@ -145,7 +145,9 @@ type Server struct { // Fields: // - mu: serializes subscribe/unsubscribe/broadcast // - subs: active listener channels -// - dropped: count of disconnected slow listeners +// - dropped: count of disconnected slow listeners; accessed +// with sync/atomic so readers on other goroutines (the +// Status RPC handler) never take the broadcast mutex type fanOut struct { mu sync.Mutex subs map[chan []Entry]struct{} @@ -273,11 +275,13 @@ type EntryMsg struct { // Fields: // - TotalEntries: total number of entries // - ConnectedClients: active listener count +// - DroppedListeners: cumulative slow-listener disconnects // - EntriesByType: entry count per type // - EntriesByProject: entry count per origin project type StatusResponse struct { TotalEntries uint64 `json:"total_entries"` ConnectedClients uint32 `json:"connected_clients"` + DroppedListeners uint64 `json:"dropped_listeners"` EntriesByType map[string]uint64 `json:"entries_by_type"` EntriesByProject map[string]uint64 `json:"entries_by_project"` } diff --git a/internal/write/hub/hub.go b/internal/write/hub/hub.go index 136cb8fa2..01a9138c8 100644 --- a/internal/write/hub/hub.go +++ b/internal/write/hub/hub.go @@ -15,7 +15,9 @@ import ( "github.com/ActiveMemory/ctx/internal/config/embed/text" ) -// ClusterStatus prints cluster role and stats. +// ClusterStatus prints cluster role and stats. The dropped-listener +// line is omitted when the count is zero so a healthy hub keeps its +// current output. // // Parameters: // - cmd: Cobra command for output @@ -23,11 +25,13 @@ import ( // - leader: leader address // - entries: total entry count // - peers: number of peers +// - dropped: cumulative slow-listener disconnects func ClusterStatus( cmd *cobra.Command, role, leader string, entries uint64, peers int, + dropped uint64, ) { cmd.Println(fmt.Sprintf( desc.Text(text.DescKeyWriteHubRole), role, @@ -39,6 +43,12 @@ func ClusterStatus( desc.Text(text.DescKeyWriteHubClusterStats), entries, peers, )) + if dropped > 0 { + cmd.Println(fmt.Sprintf( + desc.Text(text.DescKeyWriteHubDroppedListeners), + dropped, + )) + } } // PeerAdded confirms a peer was added. diff --git a/internal/write/hub/hub_test.go b/internal/write/hub/hub_test.go new file mode 100644 index 000000000..e830e9423 --- /dev/null +++ b/internal/write/hub/hub_test.go @@ -0,0 +1,53 @@ +// / ctx: https://ctx.ist +// ,'`./ do you remember? +// `.,'\ +// \ Copyright 2026-present Context contributors. +// SPDX-License-Identifier: Apache-2.0 + +package hub_test + +import ( + "bytes" + "strings" + "testing" + + "github.com/spf13/cobra" + + writeHub "github.com/ActiveMemory/ctx/internal/write/hub" +) + +// clusterStatus renders ClusterStatus into a buffer. +func clusterStatus(dropped uint64) string { + var buf bytes.Buffer + cmd := &cobra.Command{} + cmd.SetOut(&buf) + writeHub.ClusterStatus( + cmd, "leader", "127.0.0.1:9901", 42, 2, dropped, + ) + return buf.String() +} + +// TestClusterStatus_DroppedListeners pins the conditional +// slow-listener line. desc.Text returns "" for an unknown key, so a +// renamed text key would silently blank the line; asserting on the +// rendered count catches that. +func TestClusterStatus_DroppedListeners(t *testing.T) { + out := clusterStatus(3) + + if !strings.Contains(out, "Dropped listeners: 3") { + t.Errorf("want dropped-listener line with count, got:\n%s", out) + } +} + +// TestClusterStatus_NoDroppedListeners pins the omission at zero so +// a healthy hub's output stays what it was. +func TestClusterStatus_NoDroppedListeners(t *testing.T) { + out := clusterStatus(0) + + if strings.Contains(out, "Dropped listeners") { + t.Errorf("want no dropped-listener line at zero, got:\n%s", out) + } + if !strings.Contains(out, "Entries: 42") { + t.Errorf("want the existing stats line intact, got:\n%s", out) + } +} diff --git a/internal/write/hub/testmain_test.go b/internal/write/hub/testmain_test.go new file mode 100644 index 000000000..f355f9e44 --- /dev/null +++ b/internal/write/hub/testmain_test.go @@ -0,0 +1,21 @@ +// / ctx: https://ctx.ist +// ,'`./ do you remember? +// `.,'\ +// \ Copyright 2026-present Context contributors. +// SPDX-License-Identifier: Apache-2.0 + +package hub_test + +import ( + "os" + "testing" + + "github.com/ActiveMemory/ctx/internal/assets/read/lookup" +) + +// TestMain initializes the embedded text-asset lookup so the write +// helpers resolve their DescKey-based strings. +func TestMain(m *testing.M) { + lookup.Init() + os.Exit(m.Run()) +} From cf2a541eb34a8f5958c528c5807c588265c4f0f5 Mon Sep 17 00:00:00 2001 From: CoderMungan Date: Mon, 7 Sep 2026 21:58:10 +0300 Subject: [PATCH 2/2] fix(hub): end the stream when a slow listener is disconnected Review of #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 --- docs/operations/hub-failure-modes.md | 59 +++++--- docs/operations/hub.md | 8 +- internal/cli/connection/core/render/doc.go | 29 ++-- internal/config/hub/hub.go | 6 + internal/hub/doc.go | 13 +- internal/hub/err_check.go | 12 ++ internal/hub/fanout.go | 45 +++++- internal/hub/fanout_test.go | 31 ++++ internal/hub/handler.go | 18 ++- internal/hub/handler_test.go | 167 +++++++++++++++++++++ internal/hub/integration_test.go | 118 +++++++++++++++ internal/hub/types.go | 14 +- internal/write/hub/doc.go | 7 +- specs/fix-hub-fanout-drop-observability.md | 161 ++++++++++++++++++++ 14 files changed, 637 insertions(+), 51 deletions(-) create mode 100644 internal/hub/handler_test.go create mode 100644 specs/fix-hub-fanout-drop-observability.md diff --git a/docs/operations/hub-failure-modes.md b/docs/operations/hub-failure-modes.md index a8c409efe..3048158ef 100644 --- a/docs/operations/hub-failure-modes.md +++ b/docs/operations/hub-failure-modes.md @@ -27,32 +27,56 @@ should do. Complementary to ### Client Loses Connection Mid-Stream -**What happens:** `ctx connection listen` detects the EOF, waits -with exponential backoff, and reconnects. On reconnect it passes -its last-seen sequence; the hub replays everything newer. +**What happens:** the stream ends and `ctx connection listen` +exits. There is no automatic reconnect: the command calls Listen +once and returns when the stream ends. -**What you should do:** nothing. If reconnects are looping, check -firewall state on the hub and `ctx hub status` output. +**What you should do:** re-run `ctx connection listen`. Nothing +is lost on the hub — its log is append-only, and the replay +covers every entry newer than the sequence the client asks for. +If disconnects repeat, check firewall state on the hub and +`ctx hub status` output. + +!!! warning "Reconnect Is Manual Today" + Two consequences until automatic reconnect lands: + + - Keeping a listener up across disconnects is a supervisor's + job (systemd, a shell loop), not the command's. + - The re-run asks for sequence `0`, not the client's + last-seen sequence, so entries already written to + `.context/hub/` are appended a second time. ### Slow Listener Disconnected **What happens:** each `ctx connection listen` stream gets a buffered fan-out channel. A client that stops draining it (paused process, saturated link, a laptop that went to sleep) fills the -buffer. Rather than block every publisher or silently discard the -client's entries, the hub disconnects that one listener and closes -its channel. The client sees an EOF and reconnects with its -last-seen sequence, so the missed entries are replayed. Nothing is -lost; the reconnect is the recovery. +buffer. Rather than block every publisher, the hub disconnects +that one listener: it drops the subscription and closes the +channel. The stream then ends with a `ResourceExhausted` error +(`listener disconnected: stream not drained, fan-out buffer +full`), so `ctx connection listen` exits non-zero with that +message instead of hanging on a stream that will never carry +another entry. + +Only that one client is affected. Other listeners and every +publisher keep going, and nothing is removed from the hub's log — +the entries the disconnected client missed are still there, and a +fresh `ctx connection listen` picks up from the sequence it asks +for. Each disconnect writes a warning to the hub's stderr and increments a cumulative counter reported as `Dropped listeners:` in `ctx hub status`. -**What you should do:** an occasional disconnect is normal and -self-healing. A count that climbs steadily means listeners cannot -keep up with the publish rate — check the listening client's health -and the link to it before assuming the hub is at fault. +**What you should do:** re-run `ctx connection listen` on the +affected client. As with any lost stream, reconnect is manual +today — see +[Client Loses Connection Mid-Stream](#client-loses-connection-mid-stream) +for the caveats. A count that climbs steadily means listeners +cannot keep up with the publish rate: check the listening +client's health and the link to it before assuming the hub is at +fault. ### Partition: Majority Side Reachable @@ -81,8 +105,9 @@ a warning and exits non-zero on the share leg only. `--share` is best-effort; it never blocks local context updates. **What you should do:** run `ctx connection publish` later to -backfill, or rely on another `--share` for the same entry ID. -The hub deduplicates by entry ID. +backfill. Publish the entry once: the hub's log is append-only +and does not deduplicate by entry ID, so re-sharing the same +entry adds a second copy under a new sequence number. ## Storage @@ -217,7 +242,7 @@ clock is the culprit. | "No leader" errors | Cluster quorum; run `ctx hub status` on each peer | | Hub won't start after crash | Last line of `entries.jsonl` | | Entries missing after restore | Check `clients.json` sequence vs local `.sync-state.json` | -| Duplicate entries in shared feed | Client replayed after restore, safe (dedup by ID) | +| Duplicate entries in shared feed | A client re-published; the hub never dedups by ID | | Followers lagging | Disk or network on the follower, not the leader | ## See Also diff --git a/docs/operations/hub.md b/docs/operations/hub.md index f501d39da..ae739770b 100644 --- a/docs/operations/hub.md +++ b/docs/operations/hub.md @@ -146,9 +146,11 @@ ctx hub start --daemon ``` Clients that pushed sequences **above** the restored watermark -will re-publish on the next `listen` reconnect, because the hub -now reports a lower sequence than what clients have on disk. This -is safe; the store deduplicates by entry ID. +will re-publish, because the hub now reports a lower sequence +than what clients have on disk. Nothing is lost, but the store is +append-only and does not deduplicate by entry ID: those entries +come back with new sequence numbers, so the shared feed shows +them twice. Prune the duplicates offline if they matter. ## Log Rotation diff --git a/internal/cli/connection/core/render/doc.go b/internal/cli/connection/core/render/doc.go index 0fd4b863a..f71bbfee4 100644 --- a/internal/cli/connection/core/render/doc.go +++ b/internal/cli/connection/core/render/doc.go @@ -18,15 +18,21 @@ // // # Public Surface // -// - **[WriteEntries](dir, entries)**: appends -// each entry to the matching per-type file +// - **[WriteEntries](entries)**: appends each +// entry to the matching per-type file // (`decisions.md`, `learnings.md`, -// `conventions.md`, `tasks.md`) under `dir`, -// formatting via [HubEntryMarkdown]. Idempotent -// by entry sequence number; re-running with -// the same sequence range produces no -// duplicates because the importer tracks last- -// seen sequence per file. +// `conventions.md`, `tasks.md`) under +// `.context/hub/`, formatting via +// [HubEntryMarkdown]. It appends +// unconditionally: it is not idempotent and does +// not deduplicate, so handing it the same entry +// twice writes it twice. Skipping what has +// already landed is the caller's job — +// `ctx connection sync` does it by passing the +// hub only the sequences above its last-seen +// watermark, while `ctx connection listen` asks +// for sequence 0 on every run and therefore +// re-appends its backlog. // // # File Layout // @@ -34,8 +40,11 @@ // - `.context/hub/learnings.md` // - `.context/hub/conventions.md` // - `.context/hub/tasks.md` -// - `.context/hub/.sync-state.json`: last-seen -// sequence per type so resume is exact. +// - `.context/hub/.sync-state.json`: the single +// last-seen hub sequence, written by +// `ctx connection sync` so its resume is exact. +// `ctx connection listen` neither reads nor +// writes it. // // # Concurrency // diff --git a/internal/config/hub/hub.go b/internal/config/hub/hub.go index a156f746a..072ddca97 100644 --- a/internal/config/hub/hub.go +++ b/internal/config/hub/hub.go @@ -223,6 +223,12 @@ const ( ErrMissingToken = "missing token" // ErrInvalidToken is the gRPC error for invalid auth token. ErrInvalidToken = "invalid token" + // ErrSlowListener is the gRPC error ending a Listen stream + // whose fan-out channel the broadcaster closed because the + // client stopped draining it. The stream is over; the client + // must open a new one from its last-seen sequence. + ErrSlowListener = "listener disconnected: " + + "stream not drained, fan-out buffer full" ) // StructTagJSON is the struct tag key used by types.go for diff --git a/internal/hub/doc.go b/internal/hub/doc.go index 3465b7cf8..5c9826599 100644 --- a/internal/hub/doc.go +++ b/internal/hub/doc.go @@ -63,11 +63,14 @@ // [Store] guards its indexes and appender with a // single mutex. Listen streams subscribe to a // fan-out channel; a subscriber that lets its buffer -// fill is disconnected rather than blocking -// publishers or silently losing entries. Each -// disconnect warns on stderr and bumps a cumulative -// counter reported as DroppedListeners by the Status -// RPC. +// fill is disconnected rather than blocking every +// publisher. The disconnect ends that client's +// stream with a ResourceExhausted error, so it +// learns the stream is over instead of waiting on +// one that will never carry another entry. Each +// disconnect warns on stderr (outside the fan-out +// mutex) and bumps a cumulative counter reported as +// DroppedListeners by the Status RPC. // // # Encryption // diff --git a/internal/hub/err_check.go b/internal/hub/err_check.go index 4a1fd00c0..450771e86 100644 --- a/internal/hub/err_check.go +++ b/internal/hub/err_check.go @@ -9,6 +9,18 @@ package hub import ( "google.golang.org/grpc/codes" "google.golang.org/grpc/status" + + cfgHub "github.com/ActiveMemory/ctx/internal/config/hub" +) + +// errSlowListener terminates a Listen stream whose fan-out +// channel [fanOut.broadcast] closed for being too slow. It is a +// package-level sentinel so [Server.listenEntries] returns the +// same value every time and tests can match it with errors.Is, +// while the ResourceExhausted code travels to the client: the +// stream ends with a reason instead of a silent EOF. +var errSlowListener = status.Error( + codes.ResourceExhausted, cfgHub.ErrSlowListener, ) // authErr reports whether err is an authentication or diff --git a/internal/hub/fanout.go b/internal/hub/fanout.go index 18c52ec07..dfda1aef5 100644 --- a/internal/hub/fanout.go +++ b/internal/hub/fanout.go @@ -7,8 +7,6 @@ package hub import ( - "sync/atomic" - cfgWarn "github.com/ActiveMemory/ctx/internal/config/warn" logWarn "github.com/ActiveMemory/ctx/internal/log/warn" ) @@ -40,7 +38,14 @@ func (f *fanOut) subscribe() chan []Entry { return ch } -// unsubscribe removes and closes a listener channel. +// unsubscribe removes and closes a listener channel. It is +// idempotent: [fanOut.broadcast] may already have disconnected +// and closed ch, and every Listen stream unsubscribes on the way +// out via defer. Membership in f.subs is the open/closed record, +// so a channel already gone from the map is left alone rather +// than closed a second time — which would panic and, with no +// recovery interceptor on the gRPC server, take the hub daemon +// down. // // Parameters: // - ch: channel previously returned by subscribe @@ -48,6 +53,9 @@ func (f *fanOut) unsubscribe(ch chan []Entry) { f.mu.Lock() defer f.mu.Unlock() + if _, live := f.subs[ch]; !live { + return + } delete(f.subs, ch) close(ch) } @@ -58,12 +66,35 @@ func (f *fanOut) unsubscribe(ch chan []Entry) { // event is visible to operators rather than only bumping a // counter. // +// The warnings are written after f.mu is released. Warn writes +// to stderr, and a stalled stderr pipe holding the broadcast +// mutex would freeze subscribe, unsubscribe and the Status RPC +// along with every publisher. +// // Parameters: // - entries: entries to deliver to all subscribers func (f *fanOut) broadcast(entries []Entry) { + for _, n := range f.deliver(entries) { + logWarn.Warn(cfgWarn.HubFanOutSlowListener, n) + } +} + +// deliver is the locked half of [fanOut.broadcast]: it offers +// entries to every subscriber and disconnects the ones that +// cannot take them. +// +// Parameters: +// - entries: entries to deliver to all subscribers +// +// Returns: +// - []uint64: cumulative disconnect count after each +// disconnect this call made, one element per disconnected +// listener; nil (and unallocated) on the healthy path +func (f *fanOut) deliver(entries []Entry) []uint64 { f.mu.Lock() defer f.mu.Unlock() + var counts []uint64 for ch := range f.subs { select { case ch <- entries: @@ -71,12 +102,10 @@ func (f *fanOut) broadcast(entries []Entry) { // Slow listener: disconnect to prevent loss. delete(f.subs, ch) close(ch) - logWarn.Warn( - cfgWarn.HubFanOutSlowListener, - atomic.AddUint64(&f.dropped, 1), - ) + counts = append(counts, f.dropped.Add(1)) } } + return counts } // count returns the number of active listeners. @@ -101,5 +130,5 @@ func (f *fanOut) count() uint32 { // Returns: // - uint64: cumulative slow-listener disconnects func (f *fanOut) droppedCount() uint64 { - return atomic.LoadUint64(&f.dropped) + return f.dropped.Load() } diff --git a/internal/hub/fanout_test.go b/internal/hub/fanout_test.go index fbe0c7486..4114562fc 100644 --- a/internal/hub/fanout_test.go +++ b/internal/hub/fanout_test.go @@ -110,6 +110,37 @@ func TestFanOut_DisconnectsSlowListener(t *testing.T) { } } +// TestFanOut_UnsubscribeAfterDisconnect covers the collision +// between the two paths that close a listener channel: broadcast +// disconnecting a slow listener, and the Listen stream's deferred +// unsubscribe on the way out. The second close must be a no-op — +// closing an already-closed channel panics, and the gRPC server +// has no recovery interceptor to catch it. +func TestFanOut_UnsubscribeAfterDisconnect(t *testing.T) { + restore := logWarn.SetSink(io.Discard) + defer restore() + + fo := newFanOut() + slow := fo.subscribe() + + for i := 0; i <= fanOutBuffer; i++ { + fo.broadcast([]Entry{{ID: fmt.Sprintf("e%d", i)}}) + } + if got := fo.count(); got != 0 { + t.Fatalf("count = %d, want 0 after disconnect", got) + } + + // Would panic without the membership guard. + fo.unsubscribe(slow) + // Idempotent for any number of callers. + fo.unsubscribe(slow) + + if got := fo.droppedCount(); got != 1 { + t.Errorf("droppedCount = %d, want 1 (unsubscribe must "+ + "not count as a drop)", got) + } +} + func TestFanOut_DroppedCountStartsAtZero(t *testing.T) { fo := newFanOut() ch := fo.subscribe() diff --git a/internal/hub/handler.go b/internal/hub/handler.go index 8371111ea..c9fc2ab9b 100644 --- a/internal/hub/handler.go +++ b/internal/hub/handler.go @@ -179,13 +179,24 @@ func (s *Server) syncEntries( // listenEntries handles the Listen RPC (long-lived stream). // +// The fan-out channel doubles as the disconnect signal: when +// [fanOut.broadcast] cuts a slow listener loose it closes the +// channel, so a receive that reports !ok means this stream was +// disconnected. Ending the RPC with [errSlowListener] is what +// makes that visible — a closed channel is always receivable, +// so a receive that ignored ok would spin on nil forever at +// full CPU while the client waited on a stream that would never +// carry another entry. +// // Parameters: // - req: listen request with type filter and sequence // - send: callback to send each entry to the client // - ctx: context for cancellation // // Returns: -// - error: non-nil if send fails +// - error: [errSlowListener] if this listener was +// disconnected for not draining, otherwise non-nil if send +// fails func (s *Server) listenEntries( req *ListenRequest, send func(*EntryMsg) error, @@ -214,7 +225,10 @@ func (s *Server) listenEntries( select { case <-ctx.Done(): return nil - case entries := <-ch: + case entries, live := <-ch: + if !live { + return errSlowListener + } for i := range entries { if len(typeSet) > 0 && !typeSet[entries[i].Type] { diff --git a/internal/hub/handler_test.go b/internal/hub/handler_test.go new file mode 100644 index 000000000..4f1d6aa40 --- /dev/null +++ b/internal/hub/handler_test.go @@ -0,0 +1,167 @@ +// / ctx: https://ctx.ist +// ,'`./ do you remember? +// `.,'\ +// \ Copyright 2026-present Context contributors. +// SPDX-License-Identifier: Apache-2.0 + +package hub + +import ( + "context" + "errors" + "fmt" + "io" + "sync/atomic" + "testing" + "time" + + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + + logWarn "github.com/ActiveMemory/ctx/internal/log/warn" +) + +// listenTestServer returns a Server over an empty store. Serve is +// never called: the Listen handler is driven directly so the test +// owns the send callback and can stall it. +func listenTestServer(t *testing.T) *Server { + t.Helper() + + store, storeErr := NewStore(t.TempDir()) + if storeErr != nil { + t.Fatal(storeErr) + } + adminTok, tokErr := GenerateAdminToken() + if tokErr != nil { + t.Fatal(tokErr) + } + return NewServer(store, adminTok) +} + +// waitForListeners blocks until the server's subscriber count +// reaches want, failing the test if it does not settle in time. +func waitForListeners(t *testing.T, srv *Server, want uint32) { + t.Helper() + + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + if srv.listeners.count() == want { + return + } + time.Sleep(time.Millisecond) + } + t.Fatalf("listener count = %d, want %d", + srv.listeners.count(), want) +} + +// TestListenEntries_SlowListenerEndsStream pins the whole +// disconnect path end to end. A stalled send lets the fan-out +// buffer fill; broadcast then disconnects the listener and closes +// its channel. The handler must notice the close and return +// errSlowListener rather than spin on a channel that is always +// receivable, and its deferred unsubscribe must survive the +// channel broadcast already closed — a second close panics, and +// the gRPC server carries no recovery interceptor, so the panic +// would take the hub daemon down. +func TestListenEntries_SlowListenerEndsStream(t *testing.T) { + restore := logWarn.SetSink(io.Discard) + defer restore() + + srv := listenTestServer(t) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + release := make(chan struct{}) + var delivered atomic.Int64 + done := make(chan error, 1) + + go func() { + done <- srv.listenEntries( + &ListenRequest{}, + func(*EntryMsg) error { + <-release + delivered.Add(1) + return nil + }, + ctx, + ) + }() + + waitForListeners(t, srv, 1) + + // One broadcast past the buffer, plus one for the entry the + // handler takes off the channel before stalling in send. + for i := 0; i <= fanOutBuffer+1; i++ { + srv.listeners.broadcast( + []Entry{{ID: fmt.Sprintf("e%d", i)}}, + ) + } + + waitForListeners(t, srv, 0) + if got := srv.listeners.droppedCount(); got != 1 { + t.Fatalf("droppedCount = %d, want 1", got) + } + + close(release) + + select { + case listenErr := <-done: + if !errors.Is(listenErr, errSlowListener) { + t.Fatalf("listenEntries = %v, want errSlowListener", + listenErr) + } + if code := status.Code(listenErr); code != + codes.ResourceExhausted { + t.Errorf("status code = %v, want %v", + code, codes.ResourceExhausted) + } + case <-time.After(5 * time.Second): + t.Fatal("listenEntries never returned after its " + + "listener was disconnected") + } + + // Whatever the buffer still held at disconnect time is sent + // before the stream ends: the disconnect costs the client the + // stream, not the entries already handed to it. + if got := delivered.Load(); got < fanOutBuffer { + t.Errorf("delivered %d entries before the disconnect, "+ + "want at least %d", got, fanOutBuffer) + } +} + +// TestListenEntries_ContextCancelEndsStream keeps the healthy +// shutdown path honest: a cancelled stream returns nil, and the +// deferred unsubscribe closes a channel that is still live. +func TestListenEntries_ContextCancelEndsStream(t *testing.T) { + srv := listenTestServer(t) + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + + go func() { + done <- srv.listenEntries( + &ListenRequest{}, + func(*EntryMsg) error { return nil }, + ctx, + ) + }() + + waitForListeners(t, srv, 1) + cancel() + + select { + case listenErr := <-done: + if listenErr != nil { + t.Fatalf("listenEntries = %v, want nil on cancel", + listenErr) + } + case <-time.After(5 * time.Second): + t.Fatal("listenEntries never returned after cancel") + } + + waitForListeners(t, srv, 0) + if got := srv.listeners.droppedCount(); got != 0 { + t.Errorf("droppedCount = %d, want 0 for a clean exit", got) + } +} diff --git a/internal/hub/integration_test.go b/internal/hub/integration_test.go index 05ac0f852..2037605ae 100644 --- a/internal/hub/integration_test.go +++ b/internal/hub/integration_test.go @@ -8,11 +8,17 @@ package hub import ( "context" + "io" "net" + "strings" "testing" "time" "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + + logWarn "github.com/ActiveMemory/ctx/internal/log/warn" ) // TestIntegration_PublishAndSync spins up a hub, registers @@ -298,3 +304,115 @@ func TestIntegration_ClientLib(t *testing.T) { // server on a random port, and returns a connected client. // authedCtx and callRegister are also in server_test.go. + +// slowListenerPayload is large enough that a handful of entries +// saturate the gRPC stream window once the client stops reading, +// which is what stalls the server's send and lets the fan-out +// buffer fill. +const slowListenerPayload = 128 << 10 + +// slowListenerBroadcastCap bounds the publish loop in +// [TestIntegration_SlowListenerReachesClient] so a regression +// fails the test instead of broadcasting forever. +const slowListenerBroadcastCap = 512 + +// TestIntegration_SlowListenerReachesClient is the client-visible +// half of the disconnect contract, over a real gRPC stream. A +// listener that stops draining is cut loose server-side; the +// client must learn that from a ResourceExhausted error, not from +// a stream that stays open forever while entries pass it by. This +// is what makes `ctx connection listen` exit non-zero rather than +// report success on a stream it is no longer receiving. +func TestIntegration_SlowListenerReachesClient(t *testing.T) { + restore := logWarn.SetSink(io.Discard) + defer restore() + + _, _, adminTok := startTestServer(t) + + store, storeErr := NewStore(t.TempDir()) + if storeErr != nil { + t.Fatal(storeErr) + } + srv := NewServer(store, adminTok) + lis, lisErr := net.Listen("tcp", "127.0.0.1:0") + if lisErr != nil { + t.Fatal(lisErr) + } + go func() { _ = srv.Serve(lis) }() + t.Cleanup(func() { srv.GracefulStop() }) + + addr := lis.Addr().String() + admin, adminDialErr := NewClient(addr, "") + if adminDialErr != nil { + t.Fatal(adminDialErr) + } + regResp, regErr := admin.Register( + context.Background(), adminTok, "slow-proj", + ) + if closeErr := admin.Close(); closeErr != nil { + t.Log(closeErr) + } + if regErr != nil { + t.Fatal(regErr) + } + + client, dialErr := NewClient(addr, regResp.ClientToken) + if dialErr != nil { + t.Fatal(dialErr) + } + defer func() { _ = client.Close() }() + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + // The handler stalls on release, so the client stops calling + // Recv: the stream window fills, the server's send blocks, + // and the fan-out channel behind it has nowhere to drain. + release := make(chan struct{}) + listenErrCh := make(chan error, 1) + go func() { + listenErrCh <- client.Listen( + ctx, nil, 0, + func(EntryMsg) error { + <-release + return nil + }, + ) + }() + + waitForListeners(t, srv, 1) + + entries := []Entry{{ + ID: "slow", + Type: "learning", + Content: strings.Repeat("x", slowListenerPayload), + }} + var broadcasts int + for srv.listeners.count() != 0 { + if broadcasts == slowListenerBroadcastCap { + t.Fatalf("listener still subscribed after %d "+ + "broadcasts to a client that never reads", + broadcasts) + } + srv.listeners.broadcast(entries) + broadcasts++ + } + + close(release) + + select { + case listenErr := <-listenErrCh: + if listenErr == nil { + t.Fatal("Listen returned nil: the client cannot " + + "tell a disconnect from a clean end of stream") + } + if code := status.Code(listenErr); code != + codes.ResourceExhausted { + t.Errorf("Listen error code = %v (%v), want %v", + code, listenErr, codes.ResourceExhausted) + } + case <-time.After(30 * time.Second): + t.Fatal("Listen never returned after the server " + + "disconnected its listener") + } +} diff --git a/internal/hub/types.go b/internal/hub/types.go index 3500bd4c9..d20340758 100644 --- a/internal/hub/types.go +++ b/internal/hub/types.go @@ -9,6 +9,7 @@ package hub import ( "encoding/json" "sync" + "sync/atomic" "time" cfgHub "github.com/ActiveMemory/ctx/internal/config/hub" @@ -144,14 +145,17 @@ type Server struct { // // Fields: // - mu: serializes subscribe/unsubscribe/broadcast -// - subs: active listener channels -// - dropped: count of disconnected slow listeners; accessed -// with sync/atomic so readers on other goroutines (the -// Status RPC handler) never take the broadcast mutex +// - subs: active listener channels. Membership is also the +// open/closed record for each channel: a channel absent +// from the map has already been closed, so unsubscribe +// knows not to close it twice +// - dropped: count of disconnected slow listeners. Atomic so +// readers on other goroutines (the Status RPC handler) +// never take the broadcast mutex type fanOut struct { mu sync.Mutex subs map[chan []Entry]struct{} - dropped uint64 + dropped atomic.Uint64 } // RegisterRequest is the input for the Register RPC. diff --git a/internal/write/hub/doc.go b/internal/write/hub/doc.go index 1f89494a0..7602ae62c 100644 --- a/internal/write/hub/doc.go +++ b/internal/write/hub/doc.go @@ -12,6 +12,9 @@ // [ClusterStatus] prints the full cluster dashboard: // the current node role (Leader or Follower), the // leader address, total entry count, and peer count. +// It also prints the cumulative slow-listener +// disconnect count, but only when that count is +// non-zero, so a healthy hub keeps its former output. // // # Peer Management // @@ -32,7 +35,9 @@ // // # Usage // -// hub.ClusterStatus(cmd, role, leader, entries, peers) +// hub.ClusterStatus( +// cmd, role, leader, entries, peers, dropped, +// ) // hub.PeerAdded(cmd, peerAddr) // hub.SteppedDown(cmd) package hub diff --git a/specs/fix-hub-fanout-drop-observability.md b/specs/fix-hub-fanout-drop-observability.md new file mode 100644 index 000000000..6dda676a0 --- /dev/null +++ b/specs/fix-hub-fanout-drop-observability.md @@ -0,0 +1,161 @@ +# Fix Hub Fanout Drop Observability + +The fan-out broadcaster disconnected a listener whose buffer +filled, but the only record of it was `f.dropped` — a counter +incremented in `internal/hub/fanout.go` and read nowhere. The +disconnect itself was also broken in two ways that made it a +pending daemon crash. Upstream issue: +[ActiveMemory/ctx#94](https://github.com/ActiveMemory/ctx/issues/94). + +## Problem + +### The counter nothing read — `internal/hub/fanout.go` + +`broadcast` bumped `f.dropped` on the `default` branch of its +non-blocking send and moved on. No RPC exposed it, no warning +fired, no test asserted it. An operator whose listeners were +being cut loose had no surface that said so, and a regression +that deleted the `delete(f.subs, ch); close(ch)` block would +have passed the whole suite. + +### The stream never ended — `internal/hub/handler.go` + +`listenEntries` received with `case entries := <-ch` and never +checked the closed state. Once `broadcast` closed the channel, +a closed channel is always receivable: the handler drained the +buffered slices and then spun on `nil` forever at full CPU. The +RPC never returned, so the client never saw the stream end and +silently missed every entry published afterwards. + +### The double close — `internal/hub/fanout.go` + +Every Listen stream runs `defer s.listeners.unsubscribe(ch)`, +and `unsubscribe` closed unconditionally. When the stream +finally ended (client TCP drop, shutdown), that second close +landed on a channel `broadcast` had already closed: +`panic: close of closed channel`. `grpc.NewServer()` is +constructed with no recovery interceptor +(`internal/hub/server.go`), so grpc-go does not catch it — +every slow-listener disconnect was a pending hub-daemon crash. + +## Solution + +### Make the disconnect a real, terminating event + +1. `internal/hub/fanout.go` — `unsubscribe` is idempotent. + Membership in `f.subs` is the open/closed record for each + channel: a channel already gone from the map has already + been closed and is left alone. Both closers (broadcast's + disconnect, the stream's deferred unsubscribe) can now run + in either order. +2. `internal/config/hub/hub.go` — `ErrSlowListener`, the + wire-visible message, alongside the existing handler error + constants. +3. `internal/hub/err_check.go` — `errSlowListener`, a + package-level `status.Error(codes.ResourceExhausted, ...)` + sentinel, so the handler returns the same value every time, + tests can match it with `errors.Is`, and the code travels to + the client. +4. `internal/hub/handler.go` — `listenEntries` receives with + `entries, live := <-ch` and returns `errSlowListener` on + `!live`. The stream ends with a reason instead of spinning, + and its deferred unsubscribe is now safe. + +### Make the counter observable + +5. `internal/config/warn/warn.go` — `HubFanOutSlowListener` + format (the `HubReplicate*` family's pattern), carrying the + cumulative count so aggregators can rate it. +6. `internal/hub/fanout.go` — `broadcast` splits into a locked + `deliver` that returns the post-increment count per + disconnect, and an unlocked warn loop. `logWarn.Warn` writes + to stderr; doing that under `f.mu` would let a stalled + stderr pipe freeze subscribe, unsubscribe, every publisher, + and the Status RPC (which takes `f.mu` via `count()`). +7. `internal/hub/types.go` — `dropped` is an `atomic.Uint64`. + Writes are already serialized under `f.mu`; atomic makes + the Status RPC's read safe without taking the broadcast + mutex, and the typed field carries no 32-bit alignment + caveat. +8. `StatusResponse.DroppedListeners` (additive, wire- + compatible) is populated from a new `droppedCount()` and + rendered by `internal/write/hub.ClusterStatus` through + `DescKeyWriteHubDroppedListeners`. The line is printed only + when the count is non-zero, so a healthy hub's output is + byte-for-byte what it was. + +### Correct the docs the mechanism falsified + +9. `docs/operations/hub-failure-modes.md` — the Slow Listener + entry states what actually happens: one client's stream ends + with `ResourceExhausted`, other listeners and publishers are + unaffected, nothing leaves the hub's log. Client reconnect + is documented as manual, because it is. +10. `docs/operations/hub.md`, `hub-failure-modes.md`, + `internal/cli/connection/core/render/doc.go` — three claims + that the hub or store "deduplicates by entry ID" were + false; `Store.Append` assigns a new sequence + unconditionally and `render.appendShared` appends without + inspecting existing content. Corrected rather than left to + contradict the new text. + +## Tests + +`logWarn.SetSink(io.Discard)` (existing seam) keeps the new +warning out of test output. + +- `TestFanOut_DisconnectsSlowListener` — the channel closes, + the subscriber leaves `f.subs`, the counter increments. + Verified by mutation: deleting the `delete`/`close` block + fails it while the original three fan-out tests pass. +- `TestFanOut_UnsubscribeAfterDisconnect` — unsubscribe after a + disconnect, twice. Verified by mutation: removing the + membership guard panics with `close of closed channel`. +- `TestFanOut_DroppedCountStartsAtZero` — pins the healthy path + so the counter cannot drift upward. +- `TestFanOut_DroppedCountRaceWithBroadcast` — four goroutines + read `droppedCount()` (the Status RPC's read path) while + `broadcast` disconnects listeners. Verified by mutation: a + plain `f.dropped++` makes `-race` report a data race. +- `TestListenEntries_SlowListenerEndsStream` — drives the real + handler with a stalled `send`, then asserts it returns + `errSlowListener` with code `ResourceExhausted`, that the + already-buffered entries were delivered first, and that its + deferred unsubscribe does not panic. Verified by mutation: + restoring `case entries := <-ch` hangs it to the deadline. +- `TestListenEntries_ContextCancelEndsStream` — the healthy + shutdown still returns nil and still closes a live channel. +- `TestIntegration_SlowListenerReachesClient` — the same + contract over a real gRPC stream: a client whose handler + stalls is disconnected server-side and `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. Verified by the + same mutation, which hangs it. +- `TestClusterStatus_DroppedListeners` / + `TestClusterStatus_NoDroppedListeners` — + `desc.Text` returns `""` for an unknown key, so a renamed + text key would blank the new line silently. The first + asserts the rendered count, the second the omission at zero + and the survival of the existing stats line. + +## Out of Scope + +- **Configurable `fanOutBuffer`** (issue item #3). Tuning the + constant before the counter is observable would be tuning + blind. +- **Automatic client reconnect with backoff.** No backoff + implementation exists anywhere in the repo, and + `internal/cli/connection/core/listen` calls `Listen` once + with `sinceSequence` hardcoded to `0` — the same shape of + latent full-refetch that + `fix-hub-silent-error-suppression.md` parked for the hubsync + hook. Until reconnect lands, the failure-modes doc says + reconnect is manual and names the duplicate-append + consequence, rather than describing a recovery the code does + not perform. +- **A recovery interceptor on `grpc.NewServer()`.** It would + have contained the double-close panic, but containing a + panic is not the same as not having one; the idempotent + `unsubscribe` removes the cause. A general interceptor is a + server-wide decision, not a fan-out fix.