diff --git a/backend/internal/repository/account_repo_integration_test.go b/backend/internal/repository/account_repo_integration_test.go index c4d65a665c..b216ecd04d 100644 --- a/backend/internal/repository/account_repo_integration_test.go +++ b/backend/internal/repository/account_repo_integration_test.go @@ -752,6 +752,37 @@ func (s *AccountRepoSuite) TestTempUnschedulableFieldsLoadedByGetByIDAndGetByIDs s.Require().Equal("", cacheRecorder.setAccounts[0].TempUnschedulableReason) } +func (s *AccountRepoSuite) TestSetTempUnschedulableSkipsOutboxWhenWindowDoesNotExtend() { + account := mustCreateAccount(s.T(), s.client, &service.Account{Name: "acc-temp-noop"}) + cacheRecorder := &schedulerCacheRecorder{} + s.repo.schedulerCache = cacheRecorder + + _, err := s.repo.sql.ExecContext(s.ctx, "TRUNCATE scheduler_outbox") + s.Require().NoError(err) + + until := time.Now().Add(30 * time.Minute).UTC().Truncate(time.Second) + s.Require().NoError(s.repo.SetTempUnschedulable(s.ctx, account.ID, until, "first")) + + var count int + err = scanSingleRow(s.ctx, s.repo.sql, "SELECT COUNT(*) FROM scheduler_outbox", nil, &count) + s.Require().NoError(err) + s.Require().Equal(1, count) + s.Require().Len(cacheRecorder.setAccounts, 1) + + s.Require().NoError(s.repo.SetTempUnschedulable(s.ctx, account.ID, until.Add(-5*time.Minute), "older")) + + err = scanSingleRow(s.ctx, s.repo.sql, "SELECT COUNT(*) FROM scheduler_outbox", nil, &count) + s.Require().NoError(err) + s.Require().Equal(1, count) + s.Require().Len(cacheRecorder.setAccounts, 1) + + got, err := s.repo.GetByID(s.ctx, account.ID) + s.Require().NoError(err) + s.Require().Equal("first", got.TempUnschedulableReason) + s.Require().NotNil(got.TempUnschedulableUntil) + s.Require().WithinDuration(until, *got.TempUnschedulableUntil, time.Second) +} + func (s *AccountRepoSuite) TestClearModelRateLimits_SyncsSchedulerSnapshot() { account := mustCreateAccount(s.T(), s.client, &service.Account{ Name: "acc-clear-model-rate", diff --git a/backend/internal/repository/migrations_runner.go b/backend/internal/repository/migrations_runner.go index 6dbb9fbd7c..285326537d 100644 --- a/backend/internal/repository/migrations_runner.go +++ b/backend/internal/repository/migrations_runner.go @@ -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 = "153_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 } diff --git a/backend/internal/repository/migrations_runner_notx_test.go b/backend/internal/repository/migrations_runner_notx_test.go index b7cb396c47..c9f6a2cdf1 100644 --- a/backend/internal/repository/migrations_runner_notx_test.go +++ b/backend/internal/repository/migrations_runner_notx_test.go @@ -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("153_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("153_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{ + "153_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) diff --git a/backend/internal/repository/migrations_schema_integration_test.go b/backend/internal/repository/migrations_schema_integration_test.go index 7d5f66c7f5..b1ea0f990f 100644 --- a/backend/internal/repository/migrations_schema_integration_test.go +++ b/backend/internal/repository/migrations_schema_integration_test.go @@ -97,6 +97,10 @@ func TestMigrationsRunner_IsIdempotent_AndSchemaIsUpToDate(t *testing.T) { require.NoError(t, tx.QueryRowContext(context.Background(), "SELECT to_regclass('public.security_secrets')").Scan(&securitySecretsRegclass)) require.True(t, securitySecretsRegclass.Valid, "expected security_secrets table to exist") + // scheduler_outbox pending dedup support + requireColumn(t, tx, "scheduler_outbox", "dedup_key", "text", 0, true) + requireIndex(t, tx, "scheduler_outbox", "idx_scheduler_outbox_pending_dedup_key") + // user_allowed_groups table should exist var uagRegclass sql.NullString require.NoError(t, tx.QueryRowContext(context.Background(), "SELECT to_regclass('public.user_allowed_groups')").Scan(&uagRegclass)) diff --git a/backend/internal/repository/ops_write_pressure_integration_test.go b/backend/internal/repository/ops_write_pressure_integration_test.go index ebb7a84226..0eb0385848 100644 --- a/backend/internal/repository/ops_write_pressure_integration_test.go +++ b/backend/internal/repository/ops_write_pressure_integration_test.go @@ -57,12 +57,70 @@ func TestEnqueueSchedulerOutbox_DeduplicatesIdempotentEvents(t *testing.T) { require.NoError(t, integrationDB.QueryRowContext(ctx, "SELECT COUNT(*) FROM scheduler_outbox WHERE event_type = $1", service.SchedulerOutboxEventAccountChanged).Scan(&count)) require.Equal(t, 1, count) - time.Sleep(schedulerOutboxDedupWindow + 150*time.Millisecond) + var firstID int64 + require.NoError(t, integrationDB.QueryRowContext(ctx, "SELECT id FROM scheduler_outbox WHERE event_type = $1", service.SchedulerOutboxEventAccountChanged).Scan(&firstID)) + events, err := NewSchedulerOutboxRepository(integrationDB).ListAfterAndReleaseDedup(ctx, 0, 100) + require.NoError(t, err) + require.Len(t, events, 1) + require.Equal(t, firstID, events[0].ID) + require.NoError(t, enqueueSchedulerOutbox(ctx, integrationDB, service.SchedulerOutboxEventAccountChanged, &accountID, nil, nil)) require.NoError(t, integrationDB.QueryRowContext(ctx, "SELECT COUNT(*) FROM scheduler_outbox WHERE event_type = $1", service.SchedulerOutboxEventAccountChanged).Scan(&count)) require.Equal(t, 2, count) } +func TestSchedulerOutbox_ListAfterAndReleaseDedup_AllowsSameKeyWhileEventInFlight(t *testing.T) { + ctx := context.Background() + _, _ = integrationDB.ExecContext(ctx, "TRUNCATE scheduler_outbox RESTART IDENTITY") + + accountID := int64(17345) + require.NoError(t, enqueueSchedulerOutbox(ctx, integrationDB, service.SchedulerOutboxEventAccountChanged, &accountID, nil, nil)) + + events, err := NewSchedulerOutboxRepository(integrationDB).ListAfterAndReleaseDedup(ctx, 0, 100) + require.NoError(t, err) + require.Len(t, events, 1) + + require.NoError(t, enqueueSchedulerOutbox(ctx, integrationDB, service.SchedulerOutboxEventAccountChanged, &accountID, nil, nil)) + + var count int + require.NoError(t, integrationDB.QueryRowContext(ctx, "SELECT COUNT(*) FROM scheduler_outbox WHERE event_type = $1", service.SchedulerOutboxEventAccountChanged).Scan(&count)) + require.Equal(t, 2, count) + + var pendingKeys int + require.NoError(t, integrationDB.QueryRowContext(ctx, "SELECT COUNT(*) FROM scheduler_outbox WHERE dedup_key IS NOT NULL").Scan(&pendingKeys)) + require.Equal(t, 1, pendingKeys) +} + +func TestEnqueueSchedulerOutbox_CoalescesAccountStateBurst(t *testing.T) { + ctx := context.Background() + _, _ = integrationDB.ExecContext(ctx, "TRUNCATE scheduler_outbox RESTART IDENTITY") + + accountID := int64(22345) + for range 50 { + require.NoError(t, enqueueSchedulerOutbox(ctx, integrationDB, service.SchedulerOutboxEventAccountChanged, &accountID, nil, nil)) + } + + var count int + require.NoError(t, integrationDB.QueryRowContext(ctx, "SELECT COUNT(*) FROM scheduler_outbox WHERE event_type = $1", service.SchedulerOutboxEventAccountChanged).Scan(&count)) + t.Logf("same-account account_changed burst: calls=50 inserted=%d", count) + require.Equal(t, 1, count) +} + +func TestEnqueueSchedulerOutbox_DoesNotDeduplicateDifferentPayload(t *testing.T) { + ctx := context.Background() + _, _ = integrationDB.ExecContext(ctx, "TRUNCATE scheduler_outbox RESTART IDENTITY") + + accountID := int64(32345) + payload1 := map[string]any{"group_ids": []int64{1}} + payload2 := map[string]any{"group_ids": []int64{2}} + require.NoError(t, enqueueSchedulerOutbox(ctx, integrationDB, service.SchedulerOutboxEventAccountChanged, &accountID, nil, payload1)) + require.NoError(t, enqueueSchedulerOutbox(ctx, integrationDB, service.SchedulerOutboxEventAccountChanged, &accountID, nil, payload2)) + + var count int + require.NoError(t, integrationDB.QueryRowContext(ctx, "SELECT COUNT(*) FROM scheduler_outbox WHERE event_type = $1", service.SchedulerOutboxEventAccountChanged).Scan(&count)) + require.Equal(t, 2, count) +} + func TestEnqueueSchedulerOutbox_DoesNotDeduplicateLastUsed(t *testing.T) { ctx := context.Background() _, _ = integrationDB.ExecContext(ctx, "TRUNCATE scheduler_outbox RESTART IDENTITY") diff --git a/backend/internal/repository/scheduler_outbox_repo.go b/backend/internal/repository/scheduler_outbox_repo.go index 4b9a9f58b1..9edde6945c 100644 --- a/backend/internal/repository/scheduler_outbox_repo.go +++ b/backend/internal/repository/scheduler_outbox_repo.go @@ -2,9 +2,12 @@ package repository import ( "context" + "crypto/sha256" "database/sql" + "encoding/hex" "encoding/json" - "time" + "fmt" + "strconv" "github.com/Wei-Shaw/sub2api/internal/service" ) @@ -13,22 +16,34 @@ type schedulerOutboxRepository struct { db *sql.DB } -const schedulerOutboxDedupWindow = time.Second - func NewSchedulerOutboxRepository(db *sql.DB) service.SchedulerOutboxRepository { return &schedulerOutboxRepository{db: db} } -func (r *schedulerOutboxRepository) ListAfter(ctx context.Context, afterID int64, limit int) ([]service.SchedulerOutboxEvent, error) { +func (r *schedulerOutboxRepository) ListAfterAndReleaseDedup(ctx context.Context, afterID int64, limit int) ([]service.SchedulerOutboxEvent, error) { if limit <= 0 { limit = 100 } rows, err := r.db.QueryContext(ctx, ` - SELECT id, event_type, account_id, group_id, payload, created_at - FROM scheduler_outbox - WHERE id > $1 - ORDER BY id ASC - LIMIT $2 + WITH selected AS MATERIALIZED ( + SELECT id, event_type, account_id, group_id, payload, created_at + FROM scheduler_outbox + WHERE id > $1 + ORDER BY id ASC + LIMIT $2 + FOR UPDATE + ), released AS ( + UPDATE scheduler_outbox AS o + SET dedup_key = NULL + FROM selected AS s + WHERE o.id = s.id + AND o.dedup_key IS NOT NULL + RETURNING o.id + ) + SELECT s.id, s.event_type, s.account_id, s.group_id, s.payload, s.created_at + FROM selected AS s + CROSS JOIN (SELECT COUNT(*) FROM released) AS release_barrier + ORDER BY s.id ASC `, afterID, limit) if err != nil { return nil, err @@ -84,12 +99,14 @@ func enqueueSchedulerOutbox(ctx context.Context, exec sqlExecutor, eventType str return nil } var payloadArg any + var payloadJSON []byte if payload != nil { encoded, err := json.Marshal(payload) if err != nil { return err } payloadArg = encoded + payloadJSON = encoded } query := ` INSERT INTO scheduler_outbox (event_type, account_id, group_id, payload) @@ -97,24 +114,34 @@ func enqueueSchedulerOutbox(ctx context.Context, exec sqlExecutor, eventType str ` args := []any{eventType, accountID, groupID, payloadArg} if schedulerOutboxEventSupportsDedup(eventType) { + dedupKey := schedulerOutboxDedupKey(eventType, accountID, groupID, payloadJSON) query = ` - INSERT INTO scheduler_outbox (event_type, account_id, group_id, payload) - SELECT $1, $2, $3, $4 - WHERE NOT EXISTS ( - SELECT 1 - FROM scheduler_outbox - WHERE event_type = $1 - AND account_id IS NOT DISTINCT FROM $2 - AND group_id IS NOT DISTINCT FROM $3 - AND created_at >= NOW() - make_interval(secs => $5) - ) + INSERT INTO scheduler_outbox (event_type, account_id, group_id, payload, dedup_key) + VALUES ($1, $2, $3, $4, $5) + ON CONFLICT (dedup_key) WHERE dedup_key IS NOT NULL DO NOTHING ` - args = append(args, schedulerOutboxDedupWindow.Seconds()) + args = append(args, dedupKey) } _, err := exec.ExecContext(ctx, query, args...) return err } +func schedulerOutboxDedupKey(eventType string, accountID *int64, groupID *int64, payloadJSON []byte) string { + h := sha256.New() + _, _ = h.Write([]byte(eventType)) + _, _ = h.Write([]byte{0}) + if accountID != nil { + _, _ = h.Write([]byte(strconv.FormatInt(*accountID, 10))) + } + _, _ = h.Write([]byte{0}) + if groupID != nil { + _, _ = h.Write([]byte(strconv.FormatInt(*groupID, 10))) + } + _, _ = h.Write([]byte{0}) + _, _ = h.Write(payloadJSON) + return fmt.Sprintf("scheduler_outbox:%s", hex.EncodeToString(h.Sum(nil))) +} + func schedulerOutboxEventSupportsDedup(eventType string) bool { switch eventType { case service.SchedulerOutboxEventAccountChanged, diff --git a/backend/internal/service/scheduler_outbox.go b/backend/internal/service/scheduler_outbox.go index 32bfcfaaa1..c138b7e5a5 100644 --- a/backend/internal/service/scheduler_outbox.go +++ b/backend/internal/service/scheduler_outbox.go @@ -16,6 +16,6 @@ type SchedulerOutboxEvent struct { // SchedulerOutboxRepository 提供调度 outbox 的读取接口。 type SchedulerOutboxRepository interface { - ListAfter(ctx context.Context, afterID int64, limit int) ([]SchedulerOutboxEvent, error) + ListAfterAndReleaseDedup(ctx context.Context, afterID int64, limit int) ([]SchedulerOutboxEvent, error) MaxID(ctx context.Context) (int64, error) } diff --git a/backend/internal/service/scheduler_snapshot_service.go b/backend/internal/service/scheduler_snapshot_service.go index a68cdf0c77..6b25c0430f 100644 --- a/backend/internal/service/scheduler_snapshot_service.go +++ b/backend/internal/service/scheduler_snapshot_service.go @@ -242,7 +242,7 @@ func (s *SchedulerSnapshotService) pollOutbox() { return } - events, err := s.outboxRepo.ListAfter(ctx, watermark, 200) + events, err := s.outboxRepo.ListAfterAndReleaseDedup(ctx, watermark, 200) if err != nil { logger.LegacyPrintf("service.scheduler_snapshot", "[Scheduler] outbox poll failed: %v", err) return diff --git a/backend/migrations/152_scheduler_outbox_dedup_key.sql b/backend/migrations/152_scheduler_outbox_dedup_key.sql new file mode 100644 index 0000000000..c1d320ef67 --- /dev/null +++ b/backend/migrations/152_scheduler_outbox_dedup_key.sql @@ -0,0 +1,2 @@ +ALTER TABLE scheduler_outbox + ADD COLUMN IF NOT EXISTS dedup_key TEXT; diff --git a/backend/migrations/153_scheduler_outbox_pending_dedup_key_index_notx.sql b/backend/migrations/153_scheduler_outbox_pending_dedup_key_index_notx.sql new file mode 100644 index 0000000000..4be7f610c9 --- /dev/null +++ b/backend/migrations/153_scheduler_outbox_pending_dedup_key_index_notx.sql @@ -0,0 +1,3 @@ +CREATE UNIQUE INDEX CONCURRENTLY IF NOT EXISTS idx_scheduler_outbox_pending_dedup_key + ON scheduler_outbox (dedup_key) + WHERE dedup_key IS NOT NULL;