Change the postgres migrations in two ways:

1) Make each non-transactional migration step contain only a single migration operation. This ensures that if the operation fails, we don't end up "halfway" into a migration step. In the future, we'll enforce this via testing.
2) Change how concurrent indexes are created. See: https://www.shayon.dev/post/2024/225/stop-relying-on-if-not-exists-for-concurrent-index-creation-in-postgresql/ for why this is important

internal/datastore/postgres/migrations/zz_migration.0020_add_watch_api_index.go

@@ -7,11 +7,20 @@ import (
 	"github.com/jackc/pgx/v5"
 )
 
-const addWatchAPIIndexToRelationTupleTable = `CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_watch_index ON relation_tuple (created_xid);`
+const dropWatchAPIIndexFromRelationTupleTable = `
+	DROP INDEX CONCURRENTLY IF EXISTS ix_watch_index;
+`
+
+const addWatchAPIIndexToRelationTupleTable = `
+	CREATE INDEX CONCURRENTLY ix_watch_index 
+		   ON relation_tuple (created_xid);`
 
 func init() {
 	if err := DatabaseMigrations.Register("add-watch-api-index-to-relation-tuple-table", "add-metadata-to-transaction-table",
 		func(ctx context.Context, conn *pgx.Conn) error {
+			if _, err := conn.Exec(ctx, dropWatchAPIIndexFromRelationTupleTable); err != nil {
+				return fmt.Errorf("failed to drop watch API index to relation tuple table: %w", err)
+			}
 			if _, err := conn.Exec(ctx, addWatchAPIIndexToRelationTupleTable); err != nil {
 				return fmt.Errorf("failed to add watch API index to relation tuple table: %w", err)
 			}

internal/datastore/postgres/migrations/zz_migration.0021_add_expiration_support.go → internal/datastore/postgres/migrations/zz_migration.0021_add_expiration_column.go

@@ -12,24 +12,12 @@ const addExpirationColumn = `
 	ADD COLUMN expiration TIMESTAMPTZ DEFAULT NULL;
 `
 
-// Used for cleaning up expired relationships.
-const addExpiredRelationshipsIndex = `CREATE INDEX CONCURRENTLY
-	IF NOT EXISTS ix_relation_tuple_expired
-	ON relation_tuple (expiration)
-	WHERE expiration IS NOT NULL;
-`
-
 func init() {
 	if err := DatabaseMigrations.Register("add-expiration-support", "add-watch-api-index-to-relation-tuple-table",
 		func(ctx context.Context, conn *pgx.Conn) error {
 			if _, err := conn.Exec(ctx, addExpirationColumn); err != nil {
 				return fmt.Errorf("failed to add expiration column to relation tuple table: %w", err)
 			}
-
-			if _, err := conn.Exec(ctx, addExpiredRelationshipsIndex); err != nil {
-				return fmt.Errorf("failed to add expiration column to relation tuple table: %w", err)
-			}
-
 			return nil
 		},
 		noTxMigration); err != nil {

internal/datastore/postgres/migrations/zz_migration.0022_add_expiration_index.go

@@ -0,0 +1,38 @@
+package migrations
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/jackc/pgx/v5"
+)
+
+const dropExpiredRelationshipsIndex = `
+DROP INDEX CONCURRENTLY IF EXISTS ix_relation_tuple_expired;
+`
+
+// Used for cleaning up expired relationships.
+const addExpiredRelationshipsIndex = `
+CREATE INDEX CONCURRENTLY
+	ix_relation_tuple_expired
+	ON relation_tuple (expiration)
+	WHERE expiration IS NOT NULL;
+`
+
+func init() {
+	if err := DatabaseMigrations.Register("add-expiration-cleanup-index", "add-expiration-support",
+		func(ctx context.Context, conn *pgx.Conn) error {
+			if _, err := conn.Exec(ctx, dropExpiredRelationshipsIndex); err != nil {
+				return fmt.Errorf("failed to drop watch API index to relation tuple table: %w", err)
+			}
+
+			if _, err := conn.Exec(ctx, addExpiredRelationshipsIndex); err != nil {
+				return fmt.Errorf("failed to add expiration column to relation tuple table: %w", err)
+			}
+
+			return nil
+		},
+		noTxMigration); err != nil {
+		panic("failed to register migration: " + err.Error())
+	}
+}

internal/datastore/postgres/migrations/zz_migration.0022_add_index_for_transaction_gc.go → internal/datastore/postgres/migrations/zz_migration.0023_add_index_for_transaction_gc.go

@@ -21,13 +21,23 @@ import (
 // Planning Time: 0.098 ms
 // Execution Time: 5706.192 ms
 // (6 rows)
-const addGCIndexForRelationTupleTransaction = `CREATE INDEX CONCURRENTLY 
-	IF NOT EXISTS ix_relation_tuple_transaction_xid_desc_timestamp
+const addGCIndexForRelationTupleTransaction = `
+CREATE INDEX CONCURRENTLY 
+	ix_relation_tuple_transaction_xid_desc_timestamp
 	ON relation_tuple_transaction (xid DESC, timestamp);`
 
+const dropGCIndexForRelationTupleTransaction = `
+	DROP INDEX CONCURRENTLY IF EXISTS 
+	ix_relation_tuple_transaction_xid_desc_timestamp;
+`
+
 func init() {
-	if err := DatabaseMigrations.Register("add-index-for-transaction-gc", "add-expiration-support",
+	if err := DatabaseMigrations.Register("add-index-for-transaction-gc", "add-expiration-cleanup-index",
 		func(ctx context.Context, conn *pgx.Conn) error {
+			if _, err := conn.Exec(ctx, dropGCIndexForRelationTupleTransaction); err != nil {
+				return fmt.Errorf("failed to drop missing GC index: %w", err)
+			}
+
 			if _, err := conn.Exec(ctx, addGCIndexForRelationTupleTransaction); err != nil {
 				return fmt.Errorf("failed to add missing GC index: %w", err)
 			}