diff --git a/charts/terdut-server/templates/deployment.yaml b/charts/terdut-server/templates/deployment.yaml index f394ff1..e12ec2f 100644 --- a/charts/terdut-server/templates/deployment.yaml +++ b/charts/terdut-server/templates/deployment.yaml @@ -11,11 +11,10 @@ spec: matchLabels: {{- include "terdut-server.selectorLabels" . | nindent 6 }} # Recreate, not RollingUpdate, even though the PVC that forced it is gone: the - # sweeper and notifier now take a Postgres advisory lock for each pass, so two - # replicas overlapping during a rollout no longer both page for the same - # incident, but DB migrations and new-incident creation on first webhook are - # still unguarded — a second replica starting concurrently with the first can - # still race either of those. + # sweeper, notifier and migration runner now take a Postgres advisory lock + # each, so two replicas overlapping during a rollout no longer both page for + # the same incident or race applying a migration, but new-incident creation + # on the first webhook for a brand-new groupKey is still unguarded. strategy: type: Recreate template: diff --git a/internal/db/db.go b/internal/db/db.go index acb5120..c43bce8 100644 --- a/internal/db/db.go +++ b/internal/db/db.go @@ -1,6 +1,7 @@ package db import ( + "context" "database/sql" "embed" "fmt" @@ -62,6 +63,16 @@ func Open(dsn string) (*sql.DB, error) { } } +// migrationLockKey is the Postgres advisory lock Migrate holds for its whole +// run. Two replicas starting at once would otherwise race the check-then-apply +// loop below against schema_migrations: the loser could crash on a +// duplicate-key insert, or contend with the winner's uncommitted DDL. Blocking +// (pg_advisory_lock, not pg_try_advisory_lock as the archiver and notifier +// use): on boot there is no later tick to defer to, so the right behaviour is +// to wait for the other replica to finish migrating, not to skip ahead and +// start serving against an unmigrated schema. +const migrationLockKey int64 = 7265_0003 + // Migrate applies every embedded migration that has not been applied yet, in // filename order, recording each in schema_migrations. // @@ -69,6 +80,22 @@ func Open(dsn string) (*sql.DB, error) { // migration that failed half way used to leave the schema in whatever state it // had reached. Postgres has transactional DDL, so the rollback is real. func Migrate(db *sql.DB) error { + ctx := context.Background() + conn, err := db.Conn(ctx) + if err != nil { + return fmt.Errorf("migrate: acquire connection: %w", err) + } + defer conn.Close() + + if _, err := conn.ExecContext(ctx, "SELECT pg_advisory_lock($1)", migrationLockKey); err != nil { + return fmt.Errorf("migrate: acquire advisory lock: %w", err) + } + defer func() { + if _, err := conn.ExecContext(ctx, "SELECT pg_advisory_unlock($1)", migrationLockKey); err != nil { + log.Printf("migrate: release advisory lock: %v", err) + } + }() + if _, err := db.Exec(`CREATE TABLE IF NOT EXISTS schema_migrations ( version TEXT PRIMARY KEY, applied_at BIGINT NOT NULL DEFAULT FLOOR(EXTRACT(EPOCH FROM now()))::bigint diff --git a/internal/db/db_test.go b/internal/db/db_test.go new file mode 100644 index 0000000..b50af8b --- /dev/null +++ b/internal/db/db_test.go @@ -0,0 +1,125 @@ +package db_test + +import ( + "database/sql" + "fmt" + "net/url" + "os" + "strings" + "sync" + "testing" + + "git.ryuvia.com/niklas/terdut-server/internal/db" + + _ "github.com/jackc/pgx/v5/stdlib" +) + +// TERDUT_TEST_DSN must point at a database the test role may create schemas +// in; see internal/api/testdb_test.go for the fuller rationale this mirrors. +// An unset DSN fails rather than skips, deliberately. +const testDSNEnv = "TERDUT_TEST_DSN" + +// TestMigrate_ConcurrentCallersDoNotRace reproduces two replicas starting at +// once against a brand-new, unmigrated schema: both call db.Migrate at the +// same time. Before migrationLockKey, the loser could crash on a +// duplicate-key insert into schema_migrations, or contend with the winner's +// uncommitted DDL; with the advisory lock, one blocks until the other +// finishes and both return cleanly. +func TestMigrate_ConcurrentCallersDoNotRace(t *testing.T) { + dsn := os.Getenv(testDSNEnv) + if dsn == "" { + t.Fatalf("%s is not set: these tests need Postgres.\n"+ + "Run `make test-db` for a local one, then\n"+ + " export %s=postgres://terdut:terdut@localhost:5432/terdut_test?sslmode=disable", + testDSNEnv, testDSNEnv) + } + + schema := fmt.Sprintf("migrate_race_%d", os.Getpid()) + admin, err := sql.Open("pgx", dsn) + if err != nil { + t.Fatalf("connect to %s: %v", testDSNEnv, err) + } + defer admin.Close() + if _, err := admin.Exec("CREATE SCHEMA " + schema); err != nil { + t.Fatalf("create schema %s: %v", schema, err) + } + t.Cleanup(func() { + cleanup, err := sql.Open("pgx", dsn) + if err != nil { + return + } + defer cleanup.Close() + if _, err := cleanup.Exec("DROP SCHEMA " + schema + " CASCADE"); err != nil { + t.Logf("drop schema %s: %v", schema, err) + } + }) + + scoped := withSearchPath(dsn, schema) + + const callers = 2 + errs := make([]error, callers) + var wg sync.WaitGroup + for i := range callers { + wg.Add(1) + go func(i int) { + defer wg.Done() + database, err := db.Open(scoped) + if err != nil { + errs[i] = fmt.Errorf("open: %w", err) + return + } + defer database.Close() + errs[i] = db.Migrate(database) + }(i) + } + wg.Wait() + + for i, err := range errs { + if err != nil { + t.Fatalf("Migrate #%d: %v", i, err) + } + } + + entries, err := os.ReadDir("migrations") + if err != nil { + t.Fatalf("read migrations dir: %v", err) + } + var want int + for _, e := range entries { + if !e.IsDir() && strings.HasSuffix(e.Name(), ".sql") { + want++ + } + } + + check, err := sql.Open("pgx", scoped) + if err != nil { + t.Fatalf("connect for verification: %v", err) + } + defer check.Close() + + var got int + if err := check.QueryRow("SELECT COUNT(*) FROM schema_migrations").Scan(&got); err != nil { + t.Fatalf("count schema_migrations: %v", err) + } + if got != want { + t.Fatalf("schema_migrations has %d row(s) after two concurrent Migrate calls, want %d (one per migration file, no duplicates)", got, want) + } +} + +// withSearchPath pins a DSN to one schema. Copied from +// internal/api/testdb_test.go rather than shared: that helper lives in the +// api_test package, a separate compiled package this one cannot import. +func withSearchPath(dsn, schema string) string { + opt := "-csearch_path=" + schema + + if strings.HasPrefix(dsn, "postgres://") || strings.HasPrefix(dsn, "postgresql://") { + u, err := url.Parse(dsn) + if err == nil { + q := u.Query() + q.Set("options", opt) + u.RawQuery = q.Encode() + return u.String() + } + } + return dsn + " options='" + opt + "'" +}