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..3048158ef 100644 --- a/docs/operations/hub-failure-modes.md +++ b/docs/operations/hub-failure-modes.md @@ -27,12 +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 you should do:** nothing. If reconnects are looping, check -firewall state on the hub and `ctx hub status` output. +**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:** 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, 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:** 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 @@ -61,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 @@ -197,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/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/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/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/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/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..5c9826599 100644 --- a/internal/hub/doc.go +++ b/internal/hub/doc.go @@ -62,8 +62,15 @@ // // [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 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 09b510269..dfda1aef5 100644 --- a/internal/hub/fanout.go +++ b/internal/hub/fanout.go @@ -6,6 +6,11 @@ package hub +import ( + 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 @@ -33,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 @@ -41,20 +53,48 @@ 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) } // 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. +// +// 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: @@ -62,9 +102,10 @@ func (f *fanOut) broadcast(entries []Entry) { // Slow listener: disconnect to prevent loss. delete(f.subs, ch) close(ch) - f.dropped++ + counts = append(counts, f.dropped.Add(1)) } } + return counts } // count returns the number of active listeners. @@ -80,3 +121,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 f.dropped.Load() +} diff --git a/internal/hub/fanout_test.go b/internal/hub/fanout_test.go index cd119bee5..4114562fc 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,135 @@ 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") + } +} + +// 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() + 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..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] { @@ -245,6 +259,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/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 ab5530e1d..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,12 +145,17 @@ type Server struct { // // Fields: // - mu: serializes subscribe/unsubscribe/broadcast -// - subs: active listener channels -// - dropped: count of disconnected slow listeners +// - 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. @@ -273,11 +279,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/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/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()) +} 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.