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

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

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

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

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

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

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"`
 }

internal/write/hub/hub.go

@@ -15,19 +15,23 @@ 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
 //   - role: current node role (Leader/Follower)
 //   - 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.

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