Feat v0.6.0 cluster registry (#36)

Co-authored-by: Jeffrey Smith <jasafpro@gmail.com>
Co-committed-by: Jeffrey Smith <jasafpro@gmail.com>

server/cluster/registry.go

@@ -0,0 +1,143 @@
+package cluster
+
+import (
+	"context"
+	"encoding/json"
+	"log"
+	"os"
+	"runtime"
+	"sync"
+	"time"
+
+	"switchboard-core/store"
+)
+
+// ConnCounter provides WebSocket connection count (satisfied by events.Hub).
+type ConnCounter interface {
+	ConnCount() int
+}
+
+// RegistryConfig holds tunable cluster registry parameters.
+type RegistryConfig struct {
+	HeartbeatInterval time.Duration
+	StaleThreshold    time.Duration
+}
+
+// Registry manages node self-registration, heartbeat, and stale sweep.
+// Follows the workflow.Scanner pattern: stopCh + wg + ticker.
+type Registry struct {
+	nodeID    string
+	endpoint  string
+	store     store.ClusterStore
+	hub       ConnCounter
+	cfg       RegistryConfig
+	startTime time.Time
+	stopCh    chan struct{}
+	wg        sync.WaitGroup
+}
+
+// NewRegistry creates a cluster registry instance.
+func NewRegistry(nodeID, endpoint string, s store.ClusterStore, hub ConnCounter, cfg RegistryConfig) *Registry {
+	return &Registry{
+		nodeID:    nodeID,
+		endpoint:  endpoint,
+		store:     s,
+		hub:       hub,
+		cfg:       cfg,
+		startTime: time.Now(),
+		stopCh:    make(chan struct{}),
+	}
+}
+
+// NodeID returns this node's identifier.
+func (r *Registry) NodeID() string { return r.nodeID }
+
+// Start registers the node and launches the heartbeat goroutine.
+func (r *Registry) Start() error {
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	if err := r.store.Register(ctx, r.nodeID, r.endpoint); err != nil {
+		return err
+	}
+
+	r.wg.Add(1)
+	go func() {
+		defer r.wg.Done()
+		ticker := time.NewTicker(r.cfg.HeartbeatInterval)
+		defer ticker.Stop()
+
+		for {
+			select {
+			case <-ticker.C:
+				r.tick()
+			case <-r.stopCh:
+				return
+			}
+		}
+	}()
+	log.Printf("[cluster] started (node=%s, heartbeat=%s, stale=%s)",
+		r.nodeID, r.cfg.HeartbeatInterval, r.cfg.StaleThreshold)
+	return nil
+}
+
+// Stop deregisters the node and waits for the heartbeat goroutine to exit.
+func (r *Registry) Stop() {
+	close(r.stopCh)
+	r.wg.Wait()
+
+	// Best-effort deregister so peers don't have to wait for stale sweep.
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
+	if err := r.store.Deregister(ctx, r.nodeID); err != nil {
+		log.Printf("[cluster] deregister failed: %v", err)
+	}
+	log.Printf("[cluster] stopped (node=%s)", r.nodeID)
+}
+
+// tick runs one heartbeat + sweep cycle.
+func (r *Registry) tick() {
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	stats := r.collectStats()
+
+	rows, err := r.store.Heartbeat(ctx, r.nodeID, stats)
+	if err != nil {
+		log.Printf("[cluster] heartbeat error: %v", err)
+		return
+	}
+	if rows == 0 {
+		// Self-eviction: this node was swept by a peer.
+		// Kubernetes will restart the process → re-register.
+		log.Printf("[cluster] FATAL: node %s evicted from registry — shutting down", r.nodeID)
+		os.Exit(1)
+	}
+
+	swept, err := r.store.SweepStale(ctx, r.cfg.StaleThreshold)
+	if err != nil {
+		log.Printf("[cluster] sweep error: %v", err)
+		return
+	}
+	if swept > 0 {
+		log.Printf("[cluster] swept %d stale node(s)", swept)
+	}
+}
+
+// collectStats gathers runtime metrics for the stats JSONB column.
+func (r *Registry) collectStats() json.RawMessage {
+	var m runtime.MemStats
+	runtime.ReadMemStats(&m)
+
+	stats := map[string]any{
+		"goroutines":  runtime.NumGoroutine(),
+		"heap_alloc":  m.HeapAlloc,
+		"heap_sys":    m.HeapSys,
+		"gc_cycles":   m.NumGC,
+		"gc_pause_ns": m.PauseNs[(m.NumGC+255)%256],
+		"uptime_sec":  time.Since(r.startTime).Seconds(),
+		"ws_clients":  r.hub.ConnCount(),
+	}
+
+	data, _ := json.Marshal(stats)
+	return data
+}

server/cluster/registry_test.go

@@ -0,0 +1,112 @@
+package cluster
+
+import (
+	"context"
+	"encoding/json"
+	"testing"
+	"time"
+
+	"switchboard-core/store"
+)
+
+// mockClusterStore implements store.ClusterStore for unit tests.
+type mockClusterStore struct {
+	nodes          []store.ClusterNode
+	heartbeatRows  int64
+	heartbeatErr   error
+	sweepCalled    bool
+	registerCalled bool
+	lastStats      json.RawMessage
+}
+
+func (m *mockClusterStore) Register(_ context.Context, nodeID, endpoint string) error {
+	m.registerCalled = true
+	return nil
+}
+
+func (m *mockClusterStore) Heartbeat(_ context.Context, nodeID string, stats json.RawMessage) (int64, error) {
+	m.lastStats = stats
+	return m.heartbeatRows, m.heartbeatErr
+}
+
+func (m *mockClusterStore) SweepStale(_ context.Context, _ time.Duration) (int64, error) {
+	m.sweepCalled = true
+	return 0, nil
+}
+
+func (m *mockClusterStore) ListNodes(_ context.Context) ([]store.ClusterNode, error) {
+	return m.nodes, nil
+}
+
+func (m *mockClusterStore) Deregister(_ context.Context, _ string) error {
+	return nil
+}
+
+// mockHub implements ConnCounter.
+type mockHub struct{ count int }
+
+func (h *mockHub) ConnCount() int { return h.count }
+
+func TestCollectStats(t *testing.T) {
+	hub := &mockHub{count: 5}
+	reg := NewRegistry("test-node", "http://localhost:8080", nil, hub, RegistryConfig{
+		HeartbeatInterval: 10 * time.Second,
+		StaleThreshold:    30 * time.Second,
+	})
+
+	data := reg.collectStats()
+
+	var stats map[string]any
+	if err := json.Unmarshal(data, &stats); err != nil {
+		t.Fatalf("collectStats returned invalid JSON: %v", err)
+	}
+
+	// Verify expected keys
+	expectedKeys := []string{"goroutines", "heap_alloc", "heap_sys", "gc_cycles", "gc_pause_ns", "uptime_sec", "ws_clients"}
+	for _, key := range expectedKeys {
+		if _, ok := stats[key]; !ok {
+			t.Errorf("missing expected stats key: %s", key)
+		}
+	}
+
+	// ws_clients should reflect hub count
+	if ws, ok := stats["ws_clients"].(float64); !ok || int(ws) != 5 {
+		t.Errorf("ws_clients = %v, want 5", stats["ws_clients"])
+	}
+}
+
+func TestRegistryStartStop(t *testing.T) {
+	ms := &mockClusterStore{heartbeatRows: 1}
+	hub := &mockHub{count: 0}
+
+	reg := NewRegistry("test-node", "", ms, hub, RegistryConfig{
+		HeartbeatInterval: 50 * time.Millisecond,
+		StaleThreshold:    150 * time.Millisecond,
+	})
+
+	if err := reg.Start(); err != nil {
+		t.Fatalf("Start() error: %v", err)
+	}
+	if !ms.registerCalled {
+		t.Error("Register was not called on Start")
+	}
+
+	// Let at least one tick fire
+	time.Sleep(100 * time.Millisecond)
+
+	reg.Stop()
+
+	if !ms.sweepCalled {
+		t.Error("SweepStale was never called during tick")
+	}
+	if ms.lastStats == nil {
+		t.Error("Heartbeat was never called with stats")
+	}
+}
+
+func TestNodeID(t *testing.T) {
+	reg := NewRegistry("my-node-123", "", nil, &mockHub{}, RegistryConfig{})
+	if got := reg.NodeID(); got != "my-node-123" {
+		t.Errorf("NodeID() = %q, want %q", got, "my-node-123")
+	}
+}