Use golang-migrate-compatible watermark tracking in migrations

coordinator/internal/setup/setup_integration_test.go

@@ -68,12 +68,12 @@ func TestRunProvisionsDatabaseSchemaAndEnvFile(t *testing.T) {
 		t.Fatalf("connect to provisioned database: %v", err)
 	}
 	defer func() { _ = conn.Close(ctx) }()
-	var count int
+	var watermark int64
-	if err := conn.QueryRow(ctx, "SELECT count(*) FROM schema_migrations").Scan(&count); err != nil {
+	if err := conn.QueryRow(ctx, "SELECT COALESCE(MAX(version), 0) FROM schema_migrations").Scan(&watermark); err != nil {
 		t.Fatalf("read schema_migrations: %v", err)
 	}
-	if count < 13 {
+	if watermark < 13 {
-		t.Errorf("schema_migrations has %d rows, want >= 13", count)
+		t.Errorf("schema watermark = %d, want >= 13", watermark)
 	}
 	envContent, err := os.ReadFile(envPath)
 	if err != nil {

coordinator/internal/storage/postgres/integration_test.go

@@ -672,16 +672,16 @@ func TestMigrateProvisionsAndIsIdempotent(t *testing.T) {
 		t.Fatalf("second migrate (idempotent): %v", err)
 	}
 	pool := testPool(t)
-	var count int
-	if err := pool.QueryRow(ctx, "SELECT count(*) FROM schema_migrations").Scan(&count); err != nil {
-		t.Fatalf("read schema_migrations: %v", err)
-	}
 	migrations, err := listMigrations()
 	if err != nil {
 		t.Fatal(err)
 	}
-	if count != len(migrations) {
+	var watermark int64
-		t.Errorf("schema_migrations has %d rows, want %d", count, len(migrations))
+	if err := pool.QueryRow(ctx, "SELECT COALESCE(MAX(version), 0) FROM schema_migrations").Scan(&watermark); err != nil {
+		t.Fatalf("read schema_migrations: %v", err)
+	}
+	if watermark != int64(len(migrations)) {
+		t.Errorf("schema watermark = %d, want %d", watermark, len(migrations))
 	}
 	var hasJobs bool
 	if err := pool.QueryRow(ctx,

coordinator/internal/storage/postgres/migrate.go

@@ -72,12 +72,13 @@ func listMigrations() ([]migration, error) {
 	return migrations, nil
 }
 
-// Migrate applies every embedded migration that is not yet recorded in the
+// Migrate applies every embedded migration above the recorded schema version,
-// schema_migrations table, so the binary provisions its own schema. It is
+// so the binary provisions its own schema. It is idempotent and interoperates
-// idempotent and safe to run concurrently: a PostgreSQL advisory lock
+// with the golang-migrate CLI: both tools use the same schema_migrations
-// serializes migrators, and each migration file runs as its own transaction
+// watermark table (single row: version + dirty flag), and a PostgreSQL
-// (the files carry explicit BEGIN/COMMIT, matching the golang-migrate format
+// advisory lock serializes concurrent migrators. Each migration file runs as
-// the CLI and CI still use).
+// its own transaction (the files carry explicit BEGIN/COMMIT, matching the
+// golang-migrate format the CLI and CI still use).
 func Migrate(ctx context.Context, databaseURL string, log *slog.Logger) error {
 	migrations, err := listMigrations()
 	if err != nil {
@@ -102,31 +103,20 @@ func Migrate(ctx context.Context, databaseURL string, log *slog.Logger) error {
 	defer func() { _, _ = conn.Exec(ctx, "SELECT pg_advisory_unlock(82473911)") }()
 
 	if _, err := conn.Exec(ctx,
-		"CREATE TABLE IF NOT EXISTS schema_migrations (version bigint PRIMARY KEY, applied_at timestamptz NOT NULL DEFAULT now())",
+		"CREATE TABLE IF NOT EXISTS schema_migrations (version bigint PRIMARY KEY, dirty boolean NOT NULL DEFAULT false)",
 	); err != nil {
 		return fmt.Errorf("ensure schema_migrations: %w", err)
 	}
 
-	applied := map[int64]bool{}
+	var applied int64
-	rows, err := conn.Query(ctx, "SELECT version FROM schema_migrations")
+	if err := conn.QueryRow(ctx,
-	if err != nil {
+		"SELECT COALESCE(MAX(version), 0) FROM schema_migrations",
-		return fmt.Errorf("read applied migrations: %w", err)
+	).Scan(&applied); err != nil {
-	}
+		return fmt.Errorf("read applied schema version: %w", err)
-	for rows.Next() {
-		var version int64
-		if err := rows.Scan(&version); err != nil {
-			rows.Close()
-			return fmt.Errorf("scan applied migration: %w", err)
-		}
-		applied[version] = true
-	}
-	rows.Close()
-	if err := rows.Err(); err != nil {
-		return fmt.Errorf("read applied migrations: %w", err)
 	}
 
 	for _, item := range migrations {
-		if applied[int64(item.version)] {
+		if int64(item.version) <= applied {
 			continue
 		}
 		if log != nil {
@@ -135,11 +125,20 @@ func Migrate(ctx context.Context, databaseURL string, log *slog.Logger) error {
 		if _, err := conn.Exec(ctx, item.sql); err != nil {
 			return fmt.Errorf("apply migration %s: %w", item.name, err)
 		}
-		if _, err := conn.Exec(ctx,
+		// Advance the watermark to the single-row golang-migrate layout.
-			"INSERT INTO schema_migrations (version) VALUES ($1)", item.version,
+		tag, err := conn.Exec(ctx,
-		); err != nil {
+			"UPDATE schema_migrations SET version = $1, dirty = false", item.version)
+		if err != nil {
 			return fmt.Errorf("record migration %s: %w", item.name, err)
 		}
+		if tag.RowsAffected() == 0 {
+			if _, err := conn.Exec(ctx,
+				"INSERT INTO schema_migrations (version, dirty) VALUES ($1, false)",
+				item.version,
+			); err != nil {
+				return fmt.Errorf("record migration %s: %w", item.name, err)
+			}
+		}
 	}
 	return nil
 }