fix(hub): surface fanout slow-listener disconnects

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

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

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
 

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:

internal/cli/connection/core/render/doc.go

@@ -18,24 +18,33 @@
 //
 // # 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
 //
 //   - `.context/hub/decisions.md`
 //   - `.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
 //

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
 }

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"

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

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

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
 //

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

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,38 +38,74 @@ 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
 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:
 		default:
 			// 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()
+}