package cluster // Phase-3.2: periodischer Heartbeat + Stale-Sweeper. // // Hintergrund: EnsureSelfRegistered schreibt last_seen einmal beim // API-Boot. Ohne periodisches Re-Schreiben friert last_seen auf den // Boot-Zeitpunkt ein — Peers (im Multi-Node-Setup) hätten keinen Weg // zu erkennen ob dieser Node noch lebt. Die Heartbeat-Goroutine in der // API bumpt das alle 30s; der Scheduler räumt mit SweepStaleNodes Peers // die länger als nicht gemeldet haben auf status='offline'. // // Single-Node-Effekt: Cluster-UI zeigt korrekt "last seen 12s" statt // "last seen 3h" weil last_seen frisch ist. UI-Drift-Banner-Logik im // ClusterHandler.Status nutzt die selben Felder. import ( "context" "fmt" "time" "github.com/jackc/pgx/v5/pgxpool" ) // Heartbeat bumpt last_seen + status='online' für die eigene Node-Row // und aktualisiert version + config_hash. Idempotent. UPDATE-only — die // Row muss existieren (wird via EnsureSelfRegistered beim Boot angelegt). // // Hash-Berechnung läuft synchron — typisch <50ms auf einer realistischen // DB-Größe; falls die compute-SQL fehlt (Migration im Flux) wird der // vorhandene config_hash via COALESCE behalten. func Heartbeat(ctx context.Context, pool *pgxpool.Pool, localID, version string) error { if pool == nil || localID == "" { return nil } hash, _ := ComputeConfigHash(ctx, pool) _, err := pool.Exec(ctx, ` UPDATE ha_nodes SET last_seen = NOW(), status = 'online', version = COALESCE(NULLIF($1, ''), version), config_hash = COALESCE(NULLIF($2, ''), config_hash), updated_at = NOW() WHERE id = $3`, version, hash, localID) return err } // SweepStaleNodes flippt status='online' → 'offline' für Peers deren // last_seen älter als threshold ist. Liefert die Anzahl gefliptpter // Rows zurück (für Logging). Idempotent — markiert keine Rows die // schon offline sind. // // Threshold-Konvention: 4× Heartbeat-Intervall = 2 min bei 30s-Tick. // Lässt Platz für eine verpasste API-Tick (Restart, GC-Pause, kurzer // Network-Glitch) ohne false-positive Offline. func SweepStaleNodes(ctx context.Context, pool *pgxpool.Pool, threshold time.Duration) (int64, error) { if pool == nil || threshold <= 0 { return 0, nil } // Wir bauen das Interval als String — pgx kann time.Duration nicht // direkt als INTERVAL serialisieren. interval := fmt.Sprintf("%d seconds", int(threshold.Seconds())) tag, err := pool.Exec(ctx, ` UPDATE ha_nodes SET status = 'offline', updated_at = NOW() WHERE last_seen < NOW() - $1::interval AND status = 'online'`, interval) if err != nil { return 0, err } return tag.RowsAffected(), nil }