Merge pull request #119 from hyp3rd/feat/dist-mem-cache

feat(mgmt): add SSE endpoint for live cluster topology events

Author: hyp3rd

CHANGELOG.md

@@ -8,6 +8,36 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ### Added
 
+- **`GET /cluster/events` — SSE stream of topology updates.** New
+  Server-Sent Events endpoint on the management HTTP port that
+  pushes `members` (full membership snapshot) and `heartbeat`
+  (counters snapshot) frames to subscribers, replacing the
+  monitor's 2-second poll cadence with a live stream. Connect-time
+  frames carry the current snapshot so a fresh subscriber doesn't
+  wait for the next mutation. The cache wires an in-process
+  broadcaster ([`internal/eventbus`](internal/eventbus/bus.go))
+  that drops events for slow consumers (per-subscriber bounded
+  buffer) so the SWIM heartbeat loop never backpressures on a
+  stuck operator's browser. Membership state changes propagate via
+  the new [`Membership.OnStateChange`](internal/cluster/membership.go)
+  observer hook; heartbeat snapshots tick at 1 Hz aligned with
+  the SWIM interval. Read-scope auth (matches existing
+  `/cluster/*` routes); honors the management server's lifecycle
+  context for graceful drain on `Stop()`.
+
+### Changed
+
+- **Management server `defaultWriteTimeout` 5 s → 0 (no cap).**
+  fasthttp resets WriteTimeout per response; the 5 s default
+  force-closed the new SSE stream at exactly that mark, the
+  consumer saw "other side closed", and the monitor fell back
+  to polling. Lifting the cap is safe for the mgmt port because
+  it's internal-only and JSON handlers complete in milliseconds;
+  idle keep-alive connections are still bounded by the 60 s
+  IdleTimeout. Operators who need a write cap can opt in via
+  [`WithMgmtWriteTimeout`](management_http.go).
+  Regression-pinned by `TestManagementHTTP_ClusterEvents`, which
+  reads frames for 6 s past the historic deadline.
 - **Per-route scope enforcement on the management HTTP port.**
   `WithMgmtControlAuth` is a new option that wraps the cluster-
   mutating control endpoints (`POST /evict`, `POST /clear`,

cspell.config.yaml

@@ -40,6 +40,7 @@ words:
   - assertable
   - autosync
   - backpressure
+  - backpressures
   - baselining
   - benchmarkdist
   - benchmem
@@ -49,6 +50,7 @@ words:
   - bodyclose
   - bufbuild
   - buildx
+  - bursty
   - cacheerrors
   - cachev
   - calledback
@@ -80,6 +82,7 @@ words:
   - Equalf
   - errcheck
   - errp
+  - eventbus
   - ewrap
   - excalidraw
   - excludeonly
@@ -230,6 +233,8 @@ words:
   - unmarshals
   - unpadded
   - unsharded
+  - unsub
+  - unsubbed
   - upserted
   - upserts
   - varnamelen

hypercache_dist.go

@@ -3,9 +3,28 @@ package hypercache
 import (
 	"context"
 
+	"github.com/hyp3rd/hypercache/internal/eventbus"
 	"github.com/hyp3rd/hypercache/pkg/backend"
 )
 
+// EventBus returns the in-process broadcaster the distributed
+// backend uses to publish topology changes (`members`,
+// `heartbeat`). Returns nil when the backend is not DistMemory —
+// the management HTTP server's SSE handler treats nil as
+// "streaming unsupported" and falls back to a 503.
+//
+// The bus is owned by DistMemory and lives until the backend's
+// lifecycle context is cancelled. SSE handlers Subscribe via the
+// request context so they're reaped when either the client
+// disconnects or the cache shuts down.
+func (hyperCache *HyperCache[T]) EventBus() *eventbus.Bus {
+	if dm, ok := any(hyperCache.backend).(*backend.DistMemory); ok {
+		return dm.EventBus()
+	}
+
+	return nil
+}
+
 // DistMetrics returns distributed backend metrics if the underlying backend is DistMemory.
 // Returns nil if unsupported.
 func (hyperCache *HyperCache[T]) DistMetrics() any { // generic any to avoid exporting type into core interface

internal/cluster/membership.go

@@ -7,17 +7,54 @@ import (
 	"time"
 )
 
+// StateChangeObserver is invoked AFTER a membership mutation
+// (Upsert / Mark / Remove) commits, with the membership lock
+// already released. Observers MUST NOT call back into the
+// Membership they were registered on (deadlock); they SHOULD do
+// minimal work (publish to a channel, increment a counter) and
+// return promptly, since multiple observers run sequentially on
+// the goroutine that performed the mutation.
+//
+// Phase C SSE: the cache binary registers one observer that
+// publishes a `members` event onto the in-process event bus, so
+// SSE subscribers see state transitions without re-deriving them
+// by polling.
+type StateChangeObserver func(id NodeID, state NodeState, version uint64)
+
 // Membership tracks current cluster nodes (static MVP, future: gossip/swim).
 type Membership struct {
-	mu    sync.RWMutex
-	nodes map[NodeID]*Node
-	ring  *Ring
-	ver   MembershipVersion
+	mu        sync.RWMutex
+	nodes     map[NodeID]*Node
+	ring      *Ring
+	ver       MembershipVersion
+	observers []StateChangeObserver
 }
 
 // NewMembership creates a new membership container bound to a ring.
 func NewMembership(ring *Ring) *Membership { return &Membership{nodes: map[NodeID]*Node{}, ring: ring} }
 
+// OnStateChange registers a callback invoked after every membership
+// mutation (Upsert, Mark, Remove). Registration is append-only and
+// not safe for concurrent use with mutations — call this once at
+// construction before the cache starts driving heartbeats / gossip.
+//
+// The callback runs OUTSIDE the membership lock so a slow observer
+// does not block the SWIM heartbeat loop. Observers fire in
+// registration order; one panicking observer would skip every
+// observer registered after it, so observer authors must recover
+// in their own code (the package itself does not wrap with
+// recover() because that would mask programming bugs).
+func (m *Membership) OnStateChange(fn StateChangeObserver) {
+	if fn == nil {
+		return
+	}
+
+	m.mu.Lock()
+
+	m.observers = append(m.observers, fn)
+	m.mu.Unlock()
+}
+
 // Upsert adds or updates a node and rebuilds ring.
 func (m *Membership) Upsert(n *Node) {
 	m.mu.Lock()
@@ -32,10 +69,14 @@ func (m *Membership) Upsert(n *Node) {
 		}
 	}
 
-	m.ver.Next()
+	version := m.ver.Next()
+	observers := m.observers
+	id := n.ID
+	state := n.State
 	m.mu.Unlock()
 
 	m.ring.Build(nodes)
+	notify(observers, id, state, version)
 }
 
 // List returns current nodes snapshot.
@@ -80,10 +121,15 @@ func (m *Membership) Remove(id NodeID) bool {
 		}
 	}
 
-	m.ver.Next()
+	version := m.ver.Next()
+	observers := m.observers
 	m.mu.Unlock()
 
 	m.ring.Build(nodes)
+	// State NodeDead represents "removed from membership" for
+	// observer purposes; the node is gone from the map and will
+	// not appear in any subsequent List() call.
+	notify(observers, id, NodeDead, version)
 
 	return true
 }
@@ -93,18 +139,34 @@ func (m *Membership) Mark(id NodeID, state NodeState) bool {
 	m.mu.Lock()
 
 	n, ok := m.nodes[id]
-	if ok {
-		n.State = state
-		n.Incarnation++
-
-		n.LastSeen = time.Now()
+	if !ok {
+		m.mu.Unlock()
 
-		m.ver.Next()
+		return false
 	}
 
+	n.State = state
+	n.Incarnation++
+
+	n.LastSeen = time.Now()
+
+	version := m.ver.Next()
+	observers := m.observers
+
 	m.mu.Unlock()
 
-	return ok
+	notify(observers, id, state, version)
+
+	return true
+}
+
+// notify invokes each observer in registration order with the
+// resolved state and version. Pulled out so the call sites read
+// "mutate, unlock, notify" uniformly without inline loops.
+func notify(observers []StateChangeObserver, id NodeID, state NodeState, version uint64) {
+	for _, fn := range observers {
+		fn(id, state, version)
+	}
 }
 
 // Version returns current membership version.