diff --git a/CHANGELOG.md b/CHANGELOG.md index 66c8cbb..767ac2e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,44 @@ All notable changes to Switchboard Core are documented here. +## v0.6.0 — Cluster Registry + HA + +MVP convergence point. PG-backed cluster registry for horizontal scaling — +zero new infrastructure (no etcd/Consul/Redis). + +### Added + +- **Cluster registry**: `node_registry` UNLOGGED table with self-registration + on startup (`INSERT ... ON CONFLICT DO UPDATE`), heartbeat tick (default 10s), + stale sweep (default 30s threshold), and self-eviction (`os.Exit(1)` when + swept by a peer — K8s restarts the process). +- **Cluster admin API**: `GET /api/v1/admin/cluster` returns all registered + nodes with runtime stats (goroutines, heap, GC, uptime, ws_clients) in the + standard `{data: [...]}` envelope. +- **Health endpoint cluster info**: `GET /health` and `GET /api/v1/health` + now include `node_id` and `cluster: {size, peers, heartbeat_age_ms}` when + running on Postgres with the registry active. +- **Cluster dashboard**: `cluster-dashboard` admin surface package renders one + card per node with live runtime stats. Auto-refreshes every 10s. Stats are + JSONB — new keys render automatically without schema migration. +- **Cluster config**: `CLUSTER_NODE_ID` (default hostname-PID), + `CLUSTER_HEARTBEAT_INTERVAL` (default 10s), `CLUSTER_STALE_THRESHOLD` + (default 30s), `CLUSTER_ENDPOINT` (Phase 2 mesh, auto-detect). +- **3-replica E2E test**: `docker-compose-e2e.yml` now runs 3 replicas. + `ci/e2e-cluster-test.sh` verifies registration, stale sweep on stop, and + re-registration on restart. +- **5 new tests**: 3 cluster registry unit tests (stats collection, start/stop + lifecycle, node ID) + 2 handler tests (list nodes, empty response). +- **Curated bundle expanded**: `chat` and `cluster-dashboard` added to the + default bundle (now 10 packages). Production deployments can override with + `BUNDLED_PACKAGES=notes,chat,chat-core,cluster-dashboard` for a lean set. + +### Changed + +- SQLite deployments are unaffected — cluster store is nil, all cluster code + is guarded behind `database.IsPostgres() && stores.Cluster != nil`. + Cluster dashboard gracefully shows "0 nodes registered" on SQLite. + ## v0.5.5 — Upgrade Testing ### Fixed diff --git a/ROADMAP.md b/ROADMAP.md index d76dba9..7a8f411 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -418,24 +418,24 @@ externally usable release. Design docs: `docs/DESIGN-cluster-registry.md` — PG-backed cluster registry and self-assembling mesh. -### v0.6.0 — Cluster Registry + HA (planned) +### v0.6.0 — Cluster Registry + HA PG is the consensus layer. Zero new infrastructure. `UNLOGGED` table + `LISTEN/NOTIFY` replaces etcd/Consul/Redis for homelab-to-small-team scale. | Step | Status | Description | |------|--------|-------------| -| `node_registry` table | | `UNLOGGED TABLE` — node_id, endpoint, seq, registered_at, heartbeat, stats JSONB. Folds into existing schema init. | -| Node registration | | Self-registration on startup: `INSERT ... ON CONFLICT DO UPDATE`. `node_id` = `hostname-PID` or `CLUSTER_NODE_ID` env override. | -| Heartbeat tick | | Every 10s: update own heartbeat + collect runtime stats (goroutines, heap, GC, uptime, ws_clients, extensions). | -| Stale sweep | | Every heartbeat tick: `DELETE WHERE heartbeat < now() - 30s`. All nodes run it — idempotent, no ring topology. | -| Self-eviction | | If heartbeat UPDATE returns 0 rows: node was swept by peer → log error + `os.Exit(1)`. K8s restarts → re-register. | -| LISTEN/NOTIFY routing | | Durable events (messages, state changes) fan-out via `pg_notify`. All replicas receive, push to local WS subscribers, drop if irrelevant. Phase 1: ephemeral events (typing, presence) also via NOTIFY (8KB limit, ~60 bytes each). | -| Cluster API | | `GET /api/admin/cluster` — `SELECT node_id, endpoint, registered_at, heartbeat, stats FROM node_registry ORDER BY seq`. Returns `{data: [...]}` envelope. | -| Admin cluster dashboard | | Extension renders one card per node: node_id, endpoint, uptime, ws_clients, heap, GC pause. JSONB stats — future keys render automatically, no schema migration. | -| Health endpoint | | `GET /health` includes `cluster: {size, peers, heartbeat_age_ms}`. Satisfies MVP health monitoring requirement. | -| Config | | `CLUSTER_NODE_ID`, `CLUSTER_HEARTBEAT_INTERVAL` (default 10s), `CLUSTER_STALE_THRESHOLD` (default 30s), `CLUSTER_ENDPOINT` (Phase 2 mesh, auto-detect). | -| Single-node regression | | One-node behavior identical to pre-cluster: one registry row, NOTIFY delivers back to same instance. No special-casing. | -| Multi-node integration test | | Docker Compose: 3 instances, shared PG. Verify registration, heartbeat, peer discovery, stale sweep on shutdown, WS fan-out. | +| `node_registry` table | ✅ | `UNLOGGED TABLE` — node_id, endpoint, seq, registered_at, heartbeat, stats JSONB. Postgres migration 013. | +| Node registration | ✅ | Self-registration on startup: `INSERT ... ON CONFLICT DO UPDATE`. `node_id` = `hostname-PID` or `CLUSTER_NODE_ID` env override. | +| Heartbeat tick | ✅ | Every 10s: update own heartbeat + collect runtime stats (goroutines, heap, GC, uptime, ws_clients). | +| Stale sweep | ✅ | Every heartbeat tick: `DELETE WHERE heartbeat < now() - 30s`. All nodes run it — idempotent, no ring topology. | +| Self-eviction | ✅ | If heartbeat UPDATE returns 0 rows: node was swept by peer → log error + `os.Exit(1)`. K8s restarts → re-register. | +| LISTEN/NOTIFY routing | ✅ | Durable events (messages, state changes) fan-out via `pg_notify`. All replicas receive, push to local WS subscribers, drop if irrelevant. Phase 1: ephemeral events (typing, presence) also via NOTIFY (8KB limit, ~60 bytes each). | +| Cluster API | ✅ | `GET /api/v1/admin/cluster` — returns `{data: [...]}` envelope with all registered nodes. | +| Admin cluster dashboard | ✅ | `cluster-dashboard` surface package renders one card per node: node_id, endpoint, uptime, ws_clients, heap, GC pause. JSONB stats — future keys render automatically, no schema migration. Auto-refresh every 10s. | +| Health endpoint | ✅ | `GET /health` includes `node_id` and `cluster: {size, peers, heartbeat_age_ms}`. | +| Config | ✅ | `CLUSTER_NODE_ID`, `CLUSTER_HEARTBEAT_INTERVAL` (default 10s), `CLUSTER_STALE_THRESHOLD` (default 30s), `CLUSTER_ENDPOINT` (Phase 2 mesh, auto-detect). | +| Single-node regression | ✅ | One-node behavior identical to pre-cluster: one registry row, NOTIFY delivers back to same instance. No special-casing. SQLite returns nil store — all cluster code guarded. | +| Multi-node integration test | ✅ | Docker Compose: 3 instances, shared PG. `ci/e2e-cluster-test.sh`: registration, stale sweep on stop, re-registration on restart. 3 unit tests + 2 handler tests. | ### v0.6.1 — Backup/Restore + Documentation (planned) diff --git a/VERSION b/VERSION index d1d899f..a918a2a 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.5.5 +0.6.0 diff --git a/ci/e2e-cluster-test.sh b/ci/e2e-cluster-test.sh new file mode 100755 index 0000000..18048ca --- /dev/null +++ b/ci/e2e-cluster-test.sh @@ -0,0 +1,145 @@ +#!/usr/bin/env bash +# e2e-cluster-test.sh — Cluster registry E2E test (v0.6.0) +# +# Requires: docker-compose-e2e.yml running with 3 replicas. +# Usage: +# docker compose -f docker-compose-e2e.yml up --build -d +# ./ci/e2e-cluster-test.sh +# docker compose -f docker-compose-e2e.yml down -v + +set -euo pipefail + +LB="http://localhost:3000" +DIRECT_1="http://localhost:8081" +STALE_WAIT=20 # seconds — must exceed CLUSTER_STALE_THRESHOLD (15s) + +PASS=0 +FAIL=0 + +pass() { PASS=$((PASS + 1)); echo " PASS: $1"; } +fail() { FAIL=$((FAIL + 1)); echo " FAIL: $1"; } + +# ── Wait for all 3 replicas ────────────────── +echo "=== Waiting for replicas ===" +for port in 8081 8082 8083; do + for i in $(seq 1 30); do + if curl -sf "http://localhost:$port/health" > /dev/null 2>&1; then + echo " replica :$port ready" + break + fi + sleep 1 + if [ "$i" -eq 30 ]; then + echo "FATAL: replica :$port did not become healthy" + exit 1 + fi + done +done + +# ── Login as admin ──────────────────────────── +echo "=== Authenticating ===" +TOKEN=$(curl -sf "$LB/api/v1/auth/login" \ + -H "Content-Type: application/json" \ + -d '{"login":"admin","password":"admin"}' | jq -r '.access_token') + +if [ -z "$TOKEN" ] || [ "$TOKEN" = "null" ]; then + echo "FATAL: login failed" + exit 1 +fi +echo " admin token acquired" + +auth() { echo "Authorization: Bearer $TOKEN"; } + +# Wait for heartbeat to propagate (at least 2 tick cycles at 5s each) +echo "=== Waiting for heartbeat convergence ===" +sleep 12 + +# ── Test 1: All 3 nodes registered ─────────── +echo "=== Test 1: Cluster shows 3 nodes ===" +NODES=$(curl -sf "$LB/api/v1/admin/cluster" -H "$(auth)") +COUNT=$(echo "$NODES" | jq '.data | length') +if [ "$COUNT" -eq 3 ]; then + pass "cluster has 3 nodes" +else + fail "cluster has $COUNT nodes (expected 3)" + echo " response: $NODES" +fi + +# Verify all expected node IDs present +for nid in node-1 node-2 node-3; do + if echo "$NODES" | jq -e ".data[] | select(.node_id == \"$nid\")" > /dev/null 2>&1; then + pass "$nid present" + else + fail "$nid missing from cluster" + fi +done + +# ── Test 2: Health endpoint includes cluster ── +echo "=== Test 2: Health includes cluster info ===" +HEALTH=$(curl -sf "$DIRECT_1/health") +CLUSTER_SIZE=$(echo "$HEALTH" | jq '.cluster.size // 0') +if [ "$CLUSTER_SIZE" -eq 3 ]; then + pass "health shows cluster.size=3" +else + fail "health shows cluster.size=$CLUSTER_SIZE (expected 3)" +fi + +NODE_ID=$(echo "$HEALTH" | jq -r '.node_id // ""') +if [ -n "$NODE_ID" ]; then + pass "health includes node_id=$NODE_ID" +else + fail "health missing node_id" +fi + +# ── Test 3: Stats contain expected keys ─────── +echo "=== Test 3: Node stats ===" +STATS=$(echo "$NODES" | jq '.data[0].stats') +for key in goroutines heap_alloc uptime_sec ws_clients; do + if echo "$STATS" | jq -e ".$key" > /dev/null 2>&1; then + pass "stats.$key present" + else + fail "stats.$key missing" + fi +done + +# ── Test 4: Stop one replica → stale sweep ──── +echo "=== Test 4: Stop node-3, wait for sweep ===" +docker compose -f docker-compose-e2e.yml stop switchboard-3 + +echo " waiting ${STALE_WAIT}s for stale sweep..." +sleep "$STALE_WAIT" + +NODES_AFTER=$(curl -sf "$LB/api/v1/admin/cluster" -H "$(auth)") +COUNT_AFTER=$(echo "$NODES_AFTER" | jq '.data | length') +if [ "$COUNT_AFTER" -eq 2 ]; then + pass "cluster swept to 2 nodes after stopping node-3" +else + fail "cluster has $COUNT_AFTER nodes after stop (expected 2)" +fi + +# ── Test 5: Restart replica → re-registers ──── +echo "=== Test 5: Restart node-3 ===" +docker compose -f docker-compose-e2e.yml start switchboard-3 + +# Wait for startup + at least one heartbeat +for i in $(seq 1 30); do + if curl -sf "http://localhost:8083/health" > /dev/null 2>&1; then + break + fi + sleep 1 +done +sleep 8 # allow heartbeat to register + +NODES_RESTART=$(curl -sf "$LB/api/v1/admin/cluster" -H "$(auth)") +COUNT_RESTART=$(echo "$NODES_RESTART" | jq '.data | length') +if [ "$COUNT_RESTART" -eq 3 ]; then + pass "cluster back to 3 nodes after restart" +else + fail "cluster has $COUNT_RESTART nodes after restart (expected 3)" +fi + +# ── Summary ─────────────────────────────────── +echo "" +echo "=== Results: $PASS passed, $FAIL failed ===" +if [ "$FAIL" -gt 0 ]; then + exit 1 +fi diff --git a/ci/e2e-nginx.conf b/ci/e2e-nginx.conf index 6df504c..7231890 100644 --- a/ci/e2e-nginx.conf +++ b/ci/e2e-nginx.conf @@ -6,6 +6,7 @@ http { upstream switchboard { server switchboard-1:80; server switchboard-2:80; + server switchboard-3:80; } # WebSocket upgrade map diff --git a/docker-compose-e2e.yml b/docker-compose-e2e.yml index 2708868..e3f66ab 100644 --- a/docker-compose-e2e.yml +++ b/docker-compose-e2e.yml @@ -1,12 +1,13 @@ # docker-compose-e2e.yml — Multi-replica E2E testing (Postgres) # -# Two Switchboard replicas behind an nginx load balancer, +# Three Switchboard replicas behind an nginx load balancer, # sharing a single Postgres instance. Tests cross-replica -# broadcast via pg_notify. +# broadcast via pg_notify and cluster registry (v0.6.0). # # Usage: # docker compose -f docker-compose-e2e.yml up --build -d -# ./ci/e2e-chat-test.sh +# ./ci/e2e-chat-test.sh # chat E2E +# ./ci/e2e-cluster-test.sh # cluster registry E2E # docker compose -f docker-compose-e2e.yml down -v services: @@ -44,6 +45,10 @@ services: LOG_FORMAT: text LOG_LEVEL: info SEED_USERS: "alice:password123:user,bob:password456:user,charlie:password789:user" + CLUSTER_NODE_ID: "node-1" + CLUSTER_HEARTBEAT_INTERVAL: "5s" + CLUSTER_STALE_THRESHOLD: "15s" + CLUSTER_ENDPOINT: "http://switchboard-1:8080" depends_on: postgres: condition: service_healthy @@ -70,12 +75,46 @@ services: LOG_FORMAT: text LOG_LEVEL: info SEED_USERS: "alice:password123:user,bob:password456:user,charlie:password789:user" + CLUSTER_NODE_ID: "node-2" + CLUSTER_HEARTBEAT_INTERVAL: "5s" + CLUSTER_STALE_THRESHOLD: "15s" + CLUSTER_ENDPOINT: "http://switchboard-2:8080" depends_on: postgres: condition: service_healthy ports: - "8082:80" + switchboard-3: + build: + context: . + dockerfile: Dockerfile + environment: + PORT: "8080" + BASE_PATH: "" + DB_DRIVER: postgres + DATABASE_URL: postgres://switchboard:e2e-password@postgres:5432/switchboard_e2e?sslmode=disable + JWT_SECRET: e2e-jwt-secret + ENCRYPTION_KEY: e2e-encryption-key-32chars!!!!! + SWITCHBOARD_ADMIN_USERNAME: admin + SWITCHBOARD_ADMIN_PASSWORD: admin + STORAGE_BACKEND: pvc + STORAGE_PATH: /data/storage + CORS_ALLOWED_ORIGINS: "*" + EXT_ALLOW_PRIVATE_IPS: "true" + LOG_FORMAT: text + LOG_LEVEL: info + SEED_USERS: "alice:password123:user,bob:password456:user,charlie:password789:user" + CLUSTER_NODE_ID: "node-3" + CLUSTER_HEARTBEAT_INTERVAL: "5s" + CLUSTER_STALE_THRESHOLD: "15s" + CLUSTER_ENDPOINT: "http://switchboard-3:8080" + depends_on: + postgres: + condition: service_healthy + ports: + - "8083:80" + lb: image: nginx:alpine volumes: @@ -85,3 +124,4 @@ services: depends_on: - switchboard-1 - switchboard-2 + - switchboard-3 diff --git a/packages/cluster-dashboard/css/main.css b/packages/cluster-dashboard/css/main.css new file mode 100644 index 0000000..f4cda2b --- /dev/null +++ b/packages/cluster-dashboard/css/main.css @@ -0,0 +1,100 @@ +.cluster-dashboard { + max-width: 1200px; + margin: 0 auto; + padding: 2rem 1.5rem; +} + +.cluster-header { + display: flex; + align-items: center; + gap: 1rem; + margin-bottom: 1.5rem; +} + +.cluster-header h1 { + margin: 0; + font-size: 1.5rem; + color: var(--text-primary, #111); +} + +.cluster-status { + font-size: 0.85rem; + color: var(--text-secondary, #666); + background: var(--bg-secondary, #f3f4f6); + padding: 0.25rem 0.75rem; + border-radius: 999px; +} + +.cluster-grid { + display: grid; + grid-template-columns: repeat(auto-fill, minmax(320px, 1fr)); + gap: 1rem; +} + +.cluster-card { + background: var(--bg-primary, #fff); + border: 1px solid var(--border, #e5e7eb); + border-radius: 8px; + padding: 1.25rem; +} + +.card-header { + display: flex; + justify-content: space-between; + align-items: center; + margin-bottom: 0.5rem; +} + +.node-id { + font-weight: 600; + font-size: 0.9rem; + font-family: var(--font-mono, monospace); + color: var(--text-primary, #111); +} + +.heartbeat-badge { + font-size: 0.75rem; + color: var(--text-secondary, #666); + background: var(--bg-secondary, #f3f4f6); + padding: 0.15rem 0.5rem; + border-radius: 4px; +} + +.node-endpoint { + font-size: 0.8rem; + color: var(--text-secondary, #666); + margin-bottom: 0.75rem; + font-family: var(--font-mono, monospace); +} + +.card-stats { + display: grid; + grid-template-columns: 1fr 1fr 1fr; + gap: 0.5rem; +} + +.stat { + display: flex; + flex-direction: column; + gap: 0.15rem; +} + +.stat-label { + font-size: 0.7rem; + text-transform: uppercase; + letter-spacing: 0.03em; + color: var(--text-secondary, #999); +} + +.stat-value { + font-size: 0.9rem; + font-weight: 500; + font-family: var(--font-mono, monospace); + color: var(--text-primary, #111); +} + +.cluster-empty { + color: var(--text-secondary, #666); + text-align: center; + padding: 3rem 1rem; +} diff --git a/packages/cluster-dashboard/js/main.js b/packages/cluster-dashboard/js/main.js new file mode 100644 index 0000000..efde770 --- /dev/null +++ b/packages/cluster-dashboard/js/main.js @@ -0,0 +1,124 @@ +/** + * Cluster Dashboard — admin surface showing registered cluster nodes. + * + * Polls GET /api/v1/admin/cluster every 10s. Renders one card per node + * with runtime stats from the JSONB stats column. New keys added to + * stats are rendered automatically — no code change required. + */ +(function () { + 'use strict'; + + var mount = document.getElementById('extension-mount'); + if (!mount) return; + + var REFRESH_MS = 10000; + var timer = null; + + mount.innerHTML = + '
' + + '
' + + '

Cluster

' + + 'loading...' + + '
' + + '
' + + '
'; + + load(); + timer = setInterval(load, REFRESH_MS); + + // Cleanup on navigation (SPA) + window.addEventListener('beforeunload', function () { + if (timer) clearInterval(timer); + }); + + function load() { + var api = (window.sw && window.sw.api) || (typeof API !== 'undefined' && API); + if (!api) return; + var get = api._get || api.get; + if (!get) return; + get.call(api, '/api/v1/admin/cluster').then(render).catch(function () { + // 404 = SQLite (no cluster route registered), or other fetch error + render({ data: [] }); + }); + } + + function render(resp) { + var nodes = (resp && resp.data) || []; + var status = document.getElementById('clusterStatus'); + var grid = document.getElementById('clusterGrid'); + + status.textContent = nodes.length + ' node' + (nodes.length !== 1 ? 's' : '') + ' registered'; + + if (nodes.length === 0) { + grid.innerHTML = '

No nodes registered. The cluster registry requires PostgreSQL — SQLite runs single-node only.

'; + return; + } + + grid.innerHTML = nodes.map(nodeCard).join(''); + } + + function nodeCard(node) { + var stats = {}; + try { stats = typeof node.stats === 'string' ? JSON.parse(node.stats) : (node.stats || {}); } catch (e) { /* ignore */ } + + var uptimeSec = stats.uptime_sec || 0; + var uptime = formatDuration(uptimeSec); + var heap = formatBytes(stats.heap_alloc || 0); + var gcPause = ((stats.gc_pause_ns || 0) / 1e6).toFixed(1) + ' ms'; + var wsClients = stats.ws_clients != null ? stats.ws_clients : '-'; + var goroutines = stats.goroutines != null ? stats.goroutines : '-'; + var gcCycles = stats.gc_cycles != null ? stats.gc_cycles : '-'; + var heartbeatAge = timeSince(node.heartbeat); + + return ( + '
' + + '
' + + '' + esc(node.node_id) + '' + + '' + heartbeatAge + '' + + '
' + + (node.endpoint ? '
' + esc(node.endpoint) + '
' : '') + + '
' + + stat('Uptime', uptime) + + stat('WS Clients', wsClients) + + stat('Goroutines', goroutines) + + stat('Heap', heap) + + stat('GC Pause', gcPause) + + stat('GC Cycles', gcCycles) + + '
' + + '
' + ); + } + + function stat(label, value) { + return '
' + label + '' + value + '
'; + } + + function formatBytes(b) { + if (b < 1024) return b + ' B'; + if (b < 1024 * 1024) return (b / 1024).toFixed(1) + ' KB'; + return (b / (1024 * 1024)).toFixed(1) + ' MB'; + } + + function formatDuration(sec) { + if (sec < 60) return Math.floor(sec) + 's'; + if (sec < 3600) return Math.floor(sec / 60) + 'm ' + Math.floor(sec % 60) + 's'; + var h = Math.floor(sec / 3600); + var m = Math.floor((sec % 3600) / 60); + return h + 'h ' + m + 'm'; + } + + function timeSince(iso) { + if (!iso) return '-'; + var ms = Date.now() - new Date(iso).getTime(); + if (ms < 0) ms = 0; + var sec = Math.floor(ms / 1000); + if (sec < 60) return sec + 's ago'; + return Math.floor(sec / 60) + 'm ago'; + } + + function esc(s) { + var d = document.createElement('div'); + d.appendChild(document.createTextNode(s || '')); + return d.innerHTML; + } +})(); diff --git a/packages/cluster-dashboard/manifest.json b/packages/cluster-dashboard/manifest.json new file mode 100644 index 0000000..003541b --- /dev/null +++ b/packages/cluster-dashboard/manifest.json @@ -0,0 +1,13 @@ +{ + "id": "cluster-dashboard", + "icon": "🔗", + "type": "surface", + "title": "Cluster", + "route": "/s/cluster-dashboard", + "auth": "admin", + "layout": "single", + "components": [], + "hooks": ["surface"], + "version": "0.6.0", + "description": "Cluster node registry — health, runtime stats, and peer visibility." +} diff --git a/server/cluster/registry.go b/server/cluster/registry.go new file mode 100644 index 0000000..9d236ed --- /dev/null +++ b/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 +} diff --git a/server/cluster/registry_test.go b/server/cluster/registry_test.go new file mode 100644 index 0000000..26d30d4 --- /dev/null +++ b/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") + } +} diff --git a/server/config/config.go b/server/config/config.go index 2b87845..dd8bb29 100644 --- a/server/config/config.go +++ b/server/config/config.go @@ -4,6 +4,7 @@ import ( "fmt" "os" "strconv" + "time" "github.com/joho/godotenv" ) @@ -85,6 +86,17 @@ type Config struct { OIDCRolesClaim string // JWT claim for role mapping (default "realm_access.roles") OIDCGroupsClaim string // JWT claim for group sync (default "groups") OIDCAdminRole string // IdP role value that maps to admin (default "admin") + + // Cluster registry (v0.6.0) + // PG-backed node registry for horizontal scaling. No-op on SQLite. + // CLUSTER_NODE_ID: override for deterministic identity (default: hostname-PID). + // CLUSTER_HEARTBEAT_INTERVAL: tick frequency (default "10s"). + // CLUSTER_STALE_THRESHOLD: 3x heartbeat = stale (default "30s"). + // CLUSTER_ENDPOINT: advertised address for peer mesh (Phase 2, auto-detect). + ClusterNodeID string + ClusterHeartbeatInterval time.Duration + ClusterStaleThreshold time.Duration + ClusterEndpoint string } // Load reads configuration from environment variables. @@ -143,6 +155,12 @@ func Load() *Config { OIDCRolesClaim: getEnv("OIDC_ROLES_CLAIM", "realm_access.roles"), OIDCGroupsClaim: getEnv("OIDC_GROUPS_CLAIM", "groups"), OIDCAdminRole: getEnv("OIDC_ADMIN_ROLE", "admin"), + + // Cluster + ClusterNodeID: getEnv("CLUSTER_NODE_ID", ""), + ClusterHeartbeatInterval: getEnvDuration("CLUSTER_HEARTBEAT_INTERVAL", 10*time.Second), + ClusterStaleThreshold: getEnvDuration("CLUSTER_STALE_THRESHOLD", 30*time.Second), + ClusterEndpoint: getEnv("CLUSTER_ENDPOINT", ""), } } @@ -204,6 +222,15 @@ func getEnvInt(key string, fallback int) int { return fallback } +func getEnvDuration(key string, fallback time.Duration) time.Duration { + if v := os.Getenv(key); v != "" { + if d, err := time.ParseDuration(v); err == nil { + return d + } + } + return fallback +} + func getEnvBool(key string, fallback bool) bool { if v := os.Getenv(key); v != "" { b, err := strconv.ParseBool(v) diff --git a/server/database/migrations/postgres/013_cluster_registry.sql b/server/database/migrations/postgres/013_cluster_registry.sql new file mode 100644 index 0000000..d0ff030 --- /dev/null +++ b/server/database/migrations/postgres/013_cluster_registry.sql @@ -0,0 +1,13 @@ +-- 013_cluster_registry.sql — v0.6.0 +-- PG-backed cluster registry for node self-assembly. +-- UNLOGGED: no WAL overhead, contents lost on PG crash (correct behavior — +-- all nodes are dead anyway and re-register on restart). + +CREATE UNLOGGED TABLE IF NOT EXISTS node_registry ( + node_id TEXT PRIMARY KEY, + endpoint TEXT NOT NULL DEFAULT '', + seq SERIAL, + registered_at TIMESTAMPTZ DEFAULT now(), + heartbeat TIMESTAMPTZ DEFAULT now(), + stats JSONB DEFAULT '{}' +); diff --git a/server/handlers/cluster.go b/server/handlers/cluster.go new file mode 100644 index 0000000..b63f7e0 --- /dev/null +++ b/server/handlers/cluster.go @@ -0,0 +1,32 @@ +package handlers + +import ( + "net/http" + + "github.com/gin-gonic/gin" + + "switchboard-core/store" +) + +// ClusterHandler serves the admin cluster API. +type ClusterHandler struct { + stores store.Stores +} + +func NewClusterHandler(stores store.Stores) *ClusterHandler { + return &ClusterHandler{stores: stores} +} + +// ListNodes returns all registered cluster nodes. +// GET /api/v1/admin/cluster +func (h *ClusterHandler) ListNodes(c *gin.Context) { + nodes, err := h.stores.Cluster.ListNodes(c.Request.Context()) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to list cluster nodes"}) + return + } + if nodes == nil { + nodes = []store.ClusterNode{} + } + c.JSON(http.StatusOK, gin.H{"data": nodes}) +} diff --git a/server/handlers/cluster_test.go b/server/handlers/cluster_test.go new file mode 100644 index 0000000..208ee07 --- /dev/null +++ b/server/handlers/cluster_test.go @@ -0,0 +1,114 @@ +package handlers + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/gin-gonic/gin" + + "switchboard-core/store" +) + +// mockClusterStore implements store.ClusterStore for handler tests. +type mockClusterStore struct { + nodes []store.ClusterNode +} + +func (m *mockClusterStore) Register(_ context.Context, _, _ string) error { return nil } +func (m *mockClusterStore) Heartbeat(_ context.Context, _ string, _ json.RawMessage) (int64, error) { + return 1, nil +} +func (m *mockClusterStore) SweepStale(_ context.Context, _ time.Duration) (int64, error) { 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 } + +func TestClusterListNodes(t *testing.T) { + gin.SetMode(gin.TestMode) + + mock := &mockClusterStore{ + nodes: []store.ClusterNode{ + { + NodeID: "node-1", + Endpoint: "http://node-1:8080", + Seq: 1, + RegisteredAt: time.Now(), + Heartbeat: time.Now(), + Stats: json.RawMessage(`{"ws_clients":3}`), + }, + { + NodeID: "node-2", + Endpoint: "http://node-2:8080", + Seq: 2, + RegisteredAt: time.Now(), + Heartbeat: time.Now(), + Stats: json.RawMessage(`{"ws_clients":7}`), + }, + }, + } + + stores := store.Stores{Cluster: mock} + h := NewClusterHandler(stores) + + r := gin.New() + r.GET("/api/v1/admin/cluster", h.ListNodes) + + req := httptest.NewRequest(http.MethodGet, "/api/v1/admin/cluster", nil) + w := httptest.NewRecorder() + r.ServeHTTP(w, req) + + if w.Code != http.StatusOK { + t.Fatalf("status = %d, want %d", w.Code, http.StatusOK) + } + + var resp struct { + Data []store.ClusterNode `json:"data"` + } + if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil { + t.Fatalf("unmarshal response: %v", err) + } + if len(resp.Data) != 2 { + t.Fatalf("data length = %d, want 2", len(resp.Data)) + } + if resp.Data[0].NodeID != "node-1" { + t.Errorf("data[0].node_id = %q, want %q", resp.Data[0].NodeID, "node-1") + } + if resp.Data[1].NodeID != "node-2" { + t.Errorf("data[1].node_id = %q, want %q", resp.Data[1].NodeID, "node-2") + } +} + +func TestClusterListNodesEmpty(t *testing.T) { + gin.SetMode(gin.TestMode) + + mock := &mockClusterStore{nodes: nil} + stores := store.Stores{Cluster: mock} + h := NewClusterHandler(stores) + + r := gin.New() + r.GET("/api/v1/admin/cluster", h.ListNodes) + + req := httptest.NewRequest(http.MethodGet, "/api/v1/admin/cluster", nil) + w := httptest.NewRecorder() + r.ServeHTTP(w, req) + + if w.Code != http.StatusOK { + t.Fatalf("status = %d, want %d", w.Code, http.StatusOK) + } + + var resp struct { + Data []json.RawMessage `json:"data"` + } + if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil { + t.Fatalf("unmarshal: %v", err) + } + // Should be empty array, not null + if resp.Data == nil { + t.Error("data should be [], not null") + } +} diff --git a/server/handlers/packages_bundled.go b/server/handlers/packages_bundled.go index be9a27e..306f556 100644 --- a/server/handlers/packages_bundled.go +++ b/server/handlers/packages_bundled.go @@ -23,13 +23,19 @@ import ( ) // defaultBundledPackages is the curated set of packages installed by default. +// This is the dev/test bundle — includes demos and workflow examples. // Other packages still ship in the Docker image but require BUNDLED_PACKAGES // to be set explicitly (or "*" for all). +// +// Recommended production override (lean): +// BUNDLED_PACKAGES=notes,chat,chat-core,cluster-dashboard var defaultBundledPackages = map[string]bool{ "notes": true, + "chat": true, "chat-core": true, "workflow-chat": true, "dashboard": true, + "cluster-dashboard": true, "workflow-demo": true, "bug-report-triage": true, "content-approval": true, diff --git a/server/main.go b/server/main.go index 7fd31dd..e7b055e 100644 --- a/server/main.go +++ b/server/main.go @@ -16,6 +16,7 @@ import ( "github.com/prometheus/client_golang/prometheus/promhttp" "switchboard-core/auth" + "switchboard-core/cluster" "switchboard-core/config" "switchboard-core/crypto" "switchboard-core/database" @@ -209,6 +210,27 @@ func main() { // ── WebSocket Hub ───────────────────────── hub := events.NewHub(bus, middleware.GetAllowedOrigins(cfg)) + // ── Cluster Registry (v0.6.0) ──────────── + // PG-backed node self-registration + heartbeat. No-op on SQLite. + var clusterReg *cluster.Registry + if database.IsPostgres() && stores.Cluster != nil { + nodeID := cfg.ClusterNodeID + if nodeID == "" { + hostname, _ := os.Hostname() + nodeID = fmt.Sprintf("%s-%d", hostname, os.Getpid()) + } + clusterReg = cluster.NewRegistry(nodeID, cfg.ClusterEndpoint, stores.Cluster, hub, cluster.RegistryConfig{ + HeartbeatInterval: cfg.ClusterHeartbeatInterval, + StaleThreshold: cfg.ClusterStaleThreshold, + }) + if err := clusterReg.Start(); err != nil { + log.Printf("⚠ Cluster registry failed to start: %v", err) + clusterReg = nil + } else { + defer clusterReg.Stop() + } + } + // ── WebSocket Ticket Store (v0.32.0: PG-backed for cross-pod) ───── ticketAdapter := &events.TicketValidatorAdapter{Store: stores.Tickets} @@ -244,13 +266,15 @@ func main() { // Health check (k8s probes hit this directly) base.GET("/health", func(c *gin.Context) { - c.JSON(200, gin.H{ + info := gin.H{ "status": "ok", "version": Version, "database": database.IsConnected(), "database_name": database.Name(), "schema_version": database.SchemaVersion(), - }) + } + appendClusterHealth(info, clusterReg, stores) + c.JSON(200, info) }) // v0.32.0: Kubernetes probe endpoints @@ -345,6 +369,7 @@ func main() { if database.IsConnected() { info["registration_enabled"] = handlers.IsRegistrationEnabled(stores) } + appendClusterHealth(info, clusterReg, stores) c.JSON(200, info) }) @@ -762,6 +787,12 @@ func main() { admin.PUT("/schedules/:id/disable", schedH.AdminDisableSchedule) admin.DELETE("/schedules/:id", schedH.AdminDeleteSchedule) + // ── Cluster (v0.6.0) ───────────────── + if stores.Cluster != nil { + clusterH := handlers.NewClusterHandler(stores) + admin.GET("/cluster", clusterH.ListNodes) + } + // Surface aliases (backward compat — same handlers) admin.GET("/surfaces", pkgAdm.ListPackages) admin.GET("/surfaces/:id", pkgAdm.GetPackage) @@ -841,6 +872,32 @@ func main() { } } +// appendClusterHealth adds cluster info to a health response if the registry is active. +func appendClusterHealth(info gin.H, reg *cluster.Registry, stores store.Stores) { + if reg == nil || stores.Cluster == nil { + return + } + nodes, err := stores.Cluster.ListNodes(context.Background()) + if err != nil { + return + } + var peers []string + var heartbeatAge time.Duration + for _, n := range nodes { + if n.NodeID != reg.NodeID() { + peers = append(peers, n.NodeID) + } else { + heartbeatAge = time.Since(n.Heartbeat) + } + } + info["node_id"] = reg.NodeID() + info["cluster"] = gin.H{ + "size": len(nodes), + "peers": peers, + "heartbeat_age_ms": heartbeatAge.Milliseconds(), + } +} + // ── Vault CLI Commands ────────────────────── func runVaultCommand(subcmd string) { diff --git a/server/store/cluster_iface.go b/server/store/cluster_iface.go new file mode 100644 index 0000000..c9d32fb --- /dev/null +++ b/server/store/cluster_iface.go @@ -0,0 +1,37 @@ +package store + +import ( + "context" + "encoding/json" + "time" +) + +// ClusterNode represents a registered node in the cluster registry. +type ClusterNode struct { + NodeID string `json:"node_id"` + Endpoint string `json:"endpoint"` + Seq int `json:"seq"` + RegisteredAt time.Time `json:"registered_at"` + Heartbeat time.Time `json:"heartbeat"` + Stats json.RawMessage `json:"stats"` +} + +// ClusterStore manages the node_registry table for cluster self-assembly. +// Postgres-only — SQLite deployments set this to nil. +type ClusterStore interface { + // Register inserts or re-registers a node (idempotent). + Register(ctx context.Context, nodeID, endpoint string) error + + // Heartbeat updates the node's heartbeat timestamp and stats. + // Returns rows affected — 0 means the node was swept (self-eviction). + Heartbeat(ctx context.Context, nodeID string, stats json.RawMessage) (int64, error) + + // SweepStale deletes nodes whose heartbeat is older than threshold. + SweepStale(ctx context.Context, threshold time.Duration) (int64, error) + + // ListNodes returns all registered nodes ordered by sequence. + ListNodes(ctx context.Context) ([]ClusterNode, error) + + // Deregister removes a node (best-effort cleanup on shutdown). + Deregister(ctx context.Context, nodeID string) error +} diff --git a/server/store/interfaces.go b/server/store/interfaces.go index 6e6b781..0351b6a 100644 --- a/server/store/interfaces.go +++ b/server/store/interfaces.go @@ -54,6 +54,7 @@ type Stores struct { RateLimits RateLimitStore // v0.32.0: Distributed rate limiting Triggers TriggerStore // v0.2.2: Extension event/webhook triggers ScheduledTasks ScheduledTaskStore // v0.2.2: User-created cron tasks + Cluster ClusterStore // v0.6.0: PG-backed cluster node registry (nil on SQLite) } // TeamAvailableModel is returned by CatalogStore.ListTeamAvailable. diff --git a/server/store/postgres/cluster.go b/server/store/postgres/cluster.go new file mode 100644 index 0000000..bcead43 --- /dev/null +++ b/server/store/postgres/cluster.go @@ -0,0 +1,78 @@ +package postgres + +import ( + "context" + "encoding/json" + "fmt" + "time" + + "switchboard-core/store" +) + +// ClusterStore manages node_registry — Postgres-only UNLOGGED table. +type ClusterStore struct{} + +func NewClusterStore() *ClusterStore { return &ClusterStore{} } + +func (s *ClusterStore) Register(ctx context.Context, nodeID, endpoint string) error { + _, err := DB.ExecContext(ctx, ` + INSERT INTO node_registry (node_id, endpoint) + VALUES ($1, $2) + ON CONFLICT (node_id) DO UPDATE + SET endpoint = EXCLUDED.endpoint, + registered_at = now(), + heartbeat = now(), + stats = '{}' + `, nodeID, endpoint) + return err +} + +func (s *ClusterStore) Heartbeat(ctx context.Context, nodeID string, stats json.RawMessage) (int64, error) { + result, err := DB.ExecContext(ctx, ` + UPDATE node_registry + SET heartbeat = now(), stats = $2 + WHERE node_id = $1 + `, nodeID, stats) + if err != nil { + return 0, err + } + return result.RowsAffected() +} + +func (s *ClusterStore) SweepStale(ctx context.Context, threshold time.Duration) (int64, error) { + result, err := DB.ExecContext(ctx, ` + DELETE FROM node_registry + WHERE heartbeat < now() - $1 * interval '1 second' + `, threshold.Seconds()) + if err != nil { + return 0, err + } + return result.RowsAffected() +} + +func (s *ClusterStore) ListNodes(ctx context.Context) ([]store.ClusterNode, error) { + rows, err := DB.QueryContext(ctx, ` + SELECT node_id, endpoint, seq, registered_at, heartbeat, stats + FROM node_registry + ORDER BY seq + `) + if err != nil { + return nil, err + } + defer rows.Close() + + var nodes []store.ClusterNode + for rows.Next() { + var n store.ClusterNode + if err := rows.Scan(&n.NodeID, &n.Endpoint, &n.Seq, &n.RegisteredAt, &n.Heartbeat, &n.Stats); err != nil { + return nil, fmt.Errorf("scan cluster node: %w", err) + } + nodes = append(nodes, n) + } + return nodes, rows.Err() +} + +func (s *ClusterStore) Deregister(ctx context.Context, nodeID string) error { + _, err := DB.ExecContext(ctx, `DELETE FROM node_registry WHERE node_id = $1`, nodeID) + return err +} diff --git a/server/store/postgres/stores.go b/server/store/postgres/stores.go index b8492d9..1f930f0 100644 --- a/server/store/postgres/stores.go +++ b/server/store/postgres/stores.go @@ -32,5 +32,6 @@ func NewStores(db *sql.DB) store.Stores { RateLimits: NewRateLimitStore(), Triggers: NewTriggerStore(), ScheduledTasks: NewScheduledTaskStore(), + Cluster: NewClusterStore(), } }