package main import ( "fmt" "os" "sort" "strings" ) // cmdClusterReconcileReplication bringt Publication, Grants und Subscription // idempotent in den Soll-Zustand. Verhindert die zwei Fehlermodi, die sich // beim Nachrüsten von Features zeigen: // - Eine `FOR TABLE`-Publication nimmt später per Migration hinzugekommene // Shared-Tables NICHT automatisch auf → sie replizieren nie (Standby // läuft nach Failover ohne WAF/OIDC/DHCP/RADIUS-Config). // - Der GRANT SELECT für den Replikations-User ist ein Snapshot bei Setup; // neue Tabellen fehlen → tablesync hängt in `d` (Permission). // // Rollen-Selbsterkennung (idempotent, läuft im postinst nach migrate): // // Publisher (hat Publication): // - GRANT SELECT auf ALLE Tabellen (+ DEFAULT PRIVILEGES) für den // Replikations-User. // - Publication-Mitgliedschaft angleichen: fehlende Shared-Tables ADD, // fälschlich enthaltene node-lokale (localOnlyTables) DROP. // Subscriber (hat Subscription): // - Frisch zu synchronisierende Shared-Tables lokal TRUNCATE (Primary = // Source of Truth; verhindert Duplicate-Key beim Initial-COPY einer // per Migration seed-befüllten Singleton-Tabelle), dann REFRESH. // Single-Node (weder noch): nichts zu tun. // // Best-effort: Fehler werden geloggt, brechen aber ein Paket-Upgrade nie ab. func cmdClusterReconcileReplication(_ []string) int { hasPub := psqlDBBool("edgeguard", fmt.Sprintf("SELECT EXISTS(SELECT 1 FROM pg_publication WHERE pubname='%s')", egPubName)) hasSub := psqlDBBool("edgeguard", fmt.Sprintf("SELECT EXISTS(SELECT 1 FROM pg_subscription WHERE subname='%s')", egSubName)) switch { case hasPub: reconcilePublisher() case hasSub: reconcileSubscriber() default: // Standalone-Node — keine Replikation eingerichtet. } return 0 } // reconcilePublisher gleicht Grants + Publication-Mitgliedschaft an. func reconcilePublisher() { // 1. Grants IMMER neu setzen (idempotent, deckt neue Tabellen ab). grantSQL := fmt.Sprintf( "GRANT SELECT ON ALL TABLES IN SCHEMA public TO %s;\n"+ "ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT ON TABLES TO %s;", egReplUser, egReplUser) if err := psqlDBExec("edgeguard", grantSQL); err != nil { fmt.Fprintln(os.Stderr, "reconcile: GRANT SELECT fehlgeschlagen:", err) } else { fmt.Printf("✓ reconcile: SELECT-Grants für %q aktualisiert\n", egReplUser) } // 2. Publication-Mitgliedschaft angleichen. desired, err := desiredSharedTables() if err != nil { fmt.Fprintln(os.Stderr, "reconcile: Tabellen-Liste:", err) return } current, err := publicationTables() if err != nil { fmt.Fprintln(os.Stderr, "reconcile: Publication-Liste:", err) return } desiredSet := toSet(desired) currentSet := toSet(current) var toAdd, toDrop []string for _, t := range desired { if !currentSet[t] { toAdd = append(toAdd, t) } } for _, t := range current { if !desiredSet[t] { toDrop = append(toDrop, t) // node-lokale, die fälschlich drin sind } } sort.Strings(toAdd) sort.Strings(toDrop) if len(toAdd) > 0 { if err := psqlDBExec("edgeguard", fmt.Sprintf( "ALTER PUBLICATION %s ADD TABLE %s;", egPubName, strings.Join(toAdd, ", "))); err != nil { fmt.Fprintln(os.Stderr, "reconcile: ADD TABLE fehlgeschlagen:", err) } else { fmt.Printf("✓ reconcile: %d Tabelle(n) zur Publication hinzugefügt: %s\n", len(toAdd), strings.Join(toAdd, ", ")) } } if len(toDrop) > 0 { if err := psqlDBExec("edgeguard", fmt.Sprintf( "ALTER PUBLICATION %s DROP TABLE %s;", egPubName, strings.Join(toDrop, ", "))); err != nil { fmt.Fprintln(os.Stderr, "reconcile: DROP TABLE fehlgeschlagen:", err) } else { fmt.Printf("✓ reconcile: %d node-lokale Tabelle(n) aus Publication entfernt: %s\n", len(toDrop), strings.Join(toDrop, ", ")) } } if len(toAdd) == 0 && len(toDrop) == 0 { fmt.Println("✓ reconcile: Publication bereits im Soll-Zustand") } } // reconcileSubscriber zieht neu publizierte Tabellen nach: erst lokal leeren // (Primary = Source of Truth, verhindert Duplicate-Key beim Initial-COPY), // dann REFRESH PUBLICATION. Bereits synchronisierte Tabellen bleiben unberührt. func reconcileSubscriber() { desired, err := desiredSharedTables() if err != nil { fmt.Fprintln(os.Stderr, "reconcile: Tabellen-Liste:", err) return } synced, err := subscriptionRelTables() if err != nil { fmt.Fprintln(os.Stderr, "reconcile: subscription_rel-Liste:", err) return } syncedSet := toSet(synced) var fresh []string for _, t := range desired { if !syncedSet[t] { fresh = append(fresh, t) } } sort.Strings(fresh) if len(fresh) > 0 { // Nur frisch zu synchronisierende Shared-Tables leeren — nie eine // bereits replizierte oder node-lokale Tabelle. if err := psqlDBExec("edgeguard", fmt.Sprintf("TRUNCATE %s;", strings.Join(fresh, ", "))); err != nil { fmt.Fprintln(os.Stderr, "reconcile: TRUNCATE (neue Tabellen) fehlgeschlagen:", err) } else { fmt.Printf("✓ reconcile: %d neue Tabelle(n) für Initial-Sync geleert: %s\n", len(fresh), strings.Join(fresh, ", ")) } } // REFRESH ist NICHT transaktionssicher → einzelnes Statement, autocommit. if err := psqlDBExec("edgeguard", fmt.Sprintf("ALTER SUBSCRIPTION %s REFRESH PUBLICATION;", egSubName)); err != nil { fmt.Fprintln(os.Stderr, "reconcile: REFRESH PUBLICATION fehlgeschlagen:", err) } else { fmt.Printf("✓ reconcile: Subscription %q refresht\n", egSubName) } } // desiredSharedTables = alle public-Tabellen minus localOnlyTables. func desiredSharedTables() ([]string, error) { out, err := psqlDBRun("edgeguard", []string{"-tA", "-c", `SELECT tablename FROM pg_tables WHERE schemaname='public'`}) if err != nil { return nil, fmt.Errorf("list tables: %w", err) } return filterSharedTables(splitLines(string(out))), nil } // filterSharedTables entfernt localOnlyTables aus der Tabellenliste. Pure // Funktion — unit-testbar. func filterSharedTables(all []string) []string { excluded := toSet(localOnlyTables) var out []string for _, t := range all { if t != "" && !excluded[t] { out = append(out, t) } } sort.Strings(out) return out } // publicationTables listet die aktuell in edgeguard_shared publizierten Tabellen. func publicationTables() ([]string, error) { out, err := psqlDBRun("edgeguard", []string{"-tA", "-c", fmt.Sprintf("SELECT tablename FROM pg_publication_tables WHERE pubname='%s'", egPubName)}) if err != nil { return nil, err } return splitLines(string(out)), nil } // subscriptionRelTables listet die Tabellen, die die Subscription bereits kennt. func subscriptionRelTables() ([]string, error) { out, err := psqlDBRun("edgeguard", []string{"-tA", "-c", fmt.Sprintf(`SELECT c.relname FROM pg_subscription_rel r JOIN pg_class c ON c.oid = r.srrelid JOIN pg_subscription s ON s.oid = r.srsubid WHERE s.subname = '%s'`, egSubName)}) if err != nil { return nil, err } return splitLines(string(out)), nil } func psqlDBBool(db, sql string) bool { out, err := psqlDBRun(db, []string{"-tA", "-c", sql}) if err != nil { return false } return strings.TrimSpace(string(out)) == "t" } func splitLines(s string) []string { var out []string for _, l := range strings.Split(strings.TrimSpace(s), "\n") { if l = strings.TrimSpace(l); l != "" { out = append(out, l) } } return out } func toSet(items []string) map[string]bool { m := make(map[string]bool, len(items)) for _, it := range items { m[it] = true } return m }