opensource/sub2api
3
0
Fork 0
mirror of https://github.com/Wei-Shaw/sub2api.git synced 2026-09-24 16:05:44 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix: recover scheduler outbox invalid dedup index

Browse Source
This commit is contained in:
jjaw
2026-06-16 11:48:13 +08:00
committed by shaw
parent 3ef70b045d
commit 60cf89ae26
2 changed files with 50 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/internal/repository/migrations_runner.go
+12 -4
View File
@@ -53,6 +53,8 @@ const migrationsLockRetryInterval = 500 * time.Millisecond
const nonTransactionalMigrationSuffix = "_notx.sql"
const paymentOrdersOutTradeNoUniqueMigration = "120_enforce_payment_orders_out_trade_no_unique_notx.sql"
const paymentOrdersOutTradeNoUniqueIndex = "paymentorder_out_trade_no_unique"
const schedulerOutboxPendingDedupKeyMigration = "152_scheduler_outbox_pending_dedup_key_index_notx.sql"
const schedulerOutboxPendingDedupKeyIndex = "idx_scheduler_outbox_pending_dedup_key"
type migrationChecksumCompatibilityRule struct {
fileChecksum string
@@ -258,6 +260,8 @@ func prepareNonTransactionalMigration(ctx context.Context, db *sql.DB, name stri
switch name {
case paymentOrdersOutTradeNoUniqueMigration:
return preparePaymentOrdersOutTradeNoUniqueMigration(ctx, db)
case schedulerOutboxPendingDedupKeyMigration:
return dropInvalidIndexIfPresent(ctx, db, schedulerOutboxPendingDedupKeyIndex)
default:
return nil
}
@@ -276,16 +280,20 @@ func preparePaymentOrdersOutTradeNoUniqueMigration(ctx context.Context, db *sql.
)
}
invalid, err := indexIsInvalid(ctx, db, paymentOrdersOutTradeNoUniqueIndex)
return dropInvalidIndexIfPresent(ctx, db, paymentOrdersOutTradeNoUniqueIndex)
}
func dropInvalidIndexIfPresent(ctx context.Context, db *sql.DB, indexName string) error {
invalid, err := indexIsInvalid(ctx, db, indexName)
if err != nil {
return fmt.Errorf("check invalid index %s: %w", paymentOrdersOutTradeNoUniqueIndex, err)
return fmt.Errorf("check invalid index %s: %w", indexName, err)
}
if !invalid {
return nil
}
if _, err := db.ExecContext(ctx, fmt.Sprintf("DROP INDEX CONCURRENTLY IF EXISTS %s", paymentOrdersOutTradeNoUniqueIndex)); err != nil {
return fmt.Errorf("drop invalid index %s: %w", paymentOrdersOutTradeNoUniqueIndex, err)
if _, err := db.ExecContext(ctx, fmt.Sprintf("DROP INDEX CONCURRENTLY IF EXISTS %s", indexName)); err != nil {
return fmt.Errorf("drop invalid index %s: %w", indexName, err)
}
return nil
}
backend/internal/repository/migrations_runner_notx_test.go
+38
View File
@@ -194,6 +194,44 @@ DROP INDEX CONCURRENTLY IF EXISTS paymentorder_out_trade_no;
require.NoError(t, mock.ExpectationsWereMet())
}
func TestApplyMigrationsFS_SchedulerOutboxPendingDedupKeyMigration_DropsInvalidIndexBeforeRetry(t *testing.T) {
db, mock, err := sqlmock.New()
require.NoError(t, err)
defer func() { _ = db.Close() }()
prepareMigrationsBootstrapExpectations(mock)
mock.ExpectQuery("SELECT checksum FROM schema_migrations WHERE filename = \\$1").
WithArgs("152_scheduler_outbox_pending_dedup_key_index_notx.sql").
WillReturnError(sql.ErrNoRows)
mock.ExpectQuery("SELECT EXISTS \\(").
WithArgs("idx_scheduler_outbox_pending_dedup_key").
WillReturnRows(sqlmock.NewRows([]string{"exists"}).AddRow(true))
mock.ExpectExec("DROP INDEX CONCURRENTLY IF EXISTS idx_scheduler_outbox_pending_dedup_key").
WillReturnResult(sqlmock.NewResult(0, 0))
mock.ExpectExec("CREATE UNIQUE INDEX CONCURRENTLY IF NOT EXISTS idx_scheduler_outbox_pending_dedup_key").
WillReturnResult(sqlmock.NewResult(0, 0))
mock.ExpectExec("INSERT INTO schema_migrations \\(filename, checksum\\) VALUES \\(\\$1, \\$2\\)").
WithArgs("152_scheduler_outbox_pending_dedup_key_index_notx.sql", sqlmock.AnyArg()).
WillReturnResult(sqlmock.NewResult(1, 1))
mock.ExpectExec("SELECT pg_advisory_unlock\\(\\$1\\)").
WithArgs(migrationsAdvisoryLockID).
WillReturnResult(sqlmock.NewResult(0, 1))
fsys := fstest.MapFS{
"152_scheduler_outbox_pending_dedup_key_index_notx.sql": &fstest.MapFile{
Data: []byte(`
CREATE UNIQUE INDEX CONCURRENTLY IF NOT EXISTS idx_scheduler_outbox_pending_dedup_key
ON scheduler_outbox (dedup_key)
WHERE dedup_key IS NOT NULL;
`),
},
}
err = applyMigrationsFS(context.Background(), db, fsys)
require.NoError(t, err)
require.NoError(t, mock.ExpectationsWereMet())
}
func TestApplyMigrationsFS_TransactionalMigration(t *testing.T) {
db, mock, err := sqlmock.New()
require.NoError(t, err)