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()) +}