From 34e66ec0a5453b54739d21fb3fd4196a183cc7cb Mon Sep 17 00:00:00 2001 From: jjaw Date: Fri, 12 Jun 2026 02:07:28 +0800 Subject: [PATCH 1/5] fix: outbox scheduler snapshot coalesce --- .../account_repo_integration_test.go | 31 +++++++++++++++++++ .../ops_write_pressure_integration_test.go | 21 +++++++++++-- .../repository/scheduler_outbox_repo.go | 2 +- 3 files changed, 51 insertions(+), 3 deletions(-) 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/ops_write_pressure_integration_test.go b/backend/internal/repository/ops_write_pressure_integration_test.go index ebb7a84226..4520891ec6 100644 --- a/backend/internal/repository/ops_write_pressure_integration_test.go +++ b/backend/internal/repository/ops_write_pressure_integration_test.go @@ -57,10 +57,27 @@ 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) + require.GreaterOrEqual(t, schedulerOutboxDedupWindow, 10*time.Second) + + time.Sleep(1200 * time.Millisecond) 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) + require.Equal(t, 1, count) +} + +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 dedup_window=%s", count, schedulerOutboxDedupWindow) + require.Equal(t, 1, count) } func TestEnqueueSchedulerOutbox_DoesNotDeduplicateLastUsed(t *testing.T) { diff --git a/backend/internal/repository/scheduler_outbox_repo.go b/backend/internal/repository/scheduler_outbox_repo.go index 4b9a9f58b1..586efd0582 100644 --- a/backend/internal/repository/scheduler_outbox_repo.go +++ b/backend/internal/repository/scheduler_outbox_repo.go @@ -13,7 +13,7 @@ type schedulerOutboxRepository struct { db *sql.DB } -const schedulerOutboxDedupWindow = time.Second +const schedulerOutboxDedupWindow = 10 * time.Second func NewSchedulerOutboxRepository(db *sql.DB) service.SchedulerOutboxRepository { return &schedulerOutboxRepository{db: db} From 3ef70b045d3e151c65e7574e50f184d77c56485f Mon Sep 17 00:00:00 2001 From: jjaw Date: Fri, 12 Jun 2026 12:38:34 +0800 Subject: [PATCH 2/5] fix: safely coalesce scheduler outbox events --- .../migrations_schema_integration_test.go | 4 ++ .../ops_write_pressure_integration_test.go | 24 ++++++-- .../repository/scheduler_outbox_repo.go | 55 ++++++++++++++----- backend/internal/service/scheduler_outbox.go | 1 + .../service/scheduler_snapshot_service.go | 10 ++++ .../151_scheduler_outbox_dedup_key.sql | 2 + ...er_outbox_pending_dedup_key_index_notx.sql | 3 + 7 files changed, 81 insertions(+), 18 deletions(-) create mode 100644 backend/migrations/151_scheduler_outbox_dedup_key.sql create mode 100644 backend/migrations/152_scheduler_outbox_pending_dedup_key_index_notx.sql 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 4520891ec6..91987c01eb 100644 --- a/backend/internal/repository/ops_write_pressure_integration_test.go +++ b/backend/internal/repository/ops_write_pressure_integration_test.go @@ -57,12 +57,13 @@ 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) - require.GreaterOrEqual(t, schedulerOutboxDedupWindow, 10*time.Second) + var firstID int64 + require.NoError(t, integrationDB.QueryRowContext(ctx, "SELECT id FROM scheduler_outbox WHERE event_type = $1", service.SchedulerOutboxEventAccountChanged).Scan(&firstID)) + require.NoError(t, NewSchedulerOutboxRepository(integrationDB).MarkProcessed(ctx, []int64{firstID})) - time.Sleep(1200 * time.Millisecond) 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, 1, count) + require.Equal(t, 2, count) } func TestEnqueueSchedulerOutbox_CoalescesAccountStateBurst(t *testing.T) { @@ -76,10 +77,25 @@ func TestEnqueueSchedulerOutbox_CoalescesAccountStateBurst(t *testing.T) { 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 dedup_window=%s", count, schedulerOutboxDedupWindow) + 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 586efd0582..b60d1648f5 100644 --- a/backend/internal/repository/scheduler_outbox_repo.go +++ b/backend/internal/repository/scheduler_outbox_repo.go @@ -2,19 +2,21 @@ package repository import ( "context" + "crypto/sha256" "database/sql" + "encoding/hex" "encoding/json" - "time" + "fmt" + "strconv" "github.com/Wei-Shaw/sub2api/internal/service" + "github.com/lib/pq" ) type schedulerOutboxRepository struct { db *sql.DB } -const schedulerOutboxDedupWindow = 10 * time.Second - func NewSchedulerOutboxRepository(db *sql.DB) service.SchedulerOutboxRepository { return &schedulerOutboxRepository{db: db} } @@ -79,17 +81,32 @@ func (r *schedulerOutboxRepository) MaxID(ctx context.Context) (int64, error) { return maxID, nil } +func (r *schedulerOutboxRepository) MarkProcessed(ctx context.Context, eventIDs []int64) error { + if len(eventIDs) == 0 { + return nil + } + _, err := r.db.ExecContext(ctx, ` + UPDATE scheduler_outbox + SET dedup_key = NULL + WHERE id = ANY($1) + AND dedup_key IS NOT NULL + `, pq.Array(eventIDs)) + return err +} + func enqueueSchedulerOutbox(ctx context.Context, exec sqlExecutor, eventType string, accountID *int64, groupID *int64, payload any) error { if exec == nil { 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..0fe977d25e 100644 --- a/backend/internal/service/scheduler_outbox.go +++ b/backend/internal/service/scheduler_outbox.go @@ -18,4 +18,5 @@ type SchedulerOutboxEvent struct { type SchedulerOutboxRepository interface { ListAfter(ctx context.Context, afterID int64, limit int) ([]SchedulerOutboxEvent, error) MaxID(ctx context.Context) (int64, error) + MarkProcessed(ctx context.Context, eventIDs []int64) error } diff --git a/backend/internal/service/scheduler_snapshot_service.go b/backend/internal/service/scheduler_snapshot_service.go index a68cdf0c77..23025ad719 100644 --- a/backend/internal/service/scheduler_snapshot_service.go +++ b/backend/internal/service/scheduler_snapshot_service.go @@ -253,6 +253,7 @@ func (s *SchedulerSnapshotService) pollOutbox() { watermarkForCheck := watermark seen := make(map[batchSeenKey]struct{}) + processedIDs := make([]int64, 0, len(events)) for _, event := range events { eventCtx, cancel := context.WithTimeout(context.Background(), outboxEventTimeout) err := s.handleOutboxEvent(eventCtx, event, seen) @@ -261,6 +262,15 @@ func (s *SchedulerSnapshotService) pollOutbox() { logger.LegacyPrintf("service.scheduler_snapshot", "[Scheduler] outbox handle failed: id=%d type=%s err=%v", event.ID, event.EventType, err) return } + processedIDs = append(processedIDs, event.ID) + } + + markCtx, markCancel := context.WithTimeout(context.Background(), 5*time.Second) + markErr := s.outboxRepo.MarkProcessed(markCtx, processedIDs) + markCancel() + if markErr != nil { + logger.LegacyPrintf("service.scheduler_snapshot", "[Scheduler] outbox mark processed failed: %v", markErr) + return } lastID := events[len(events)-1].ID diff --git a/backend/migrations/151_scheduler_outbox_dedup_key.sql b/backend/migrations/151_scheduler_outbox_dedup_key.sql new file mode 100644 index 0000000000..c1d320ef67 --- /dev/null +++ b/backend/migrations/151_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/152_scheduler_outbox_pending_dedup_key_index_notx.sql b/backend/migrations/152_scheduler_outbox_pending_dedup_key_index_notx.sql new file mode 100644 index 0000000000..4be7f610c9 --- /dev/null +++ b/backend/migrations/152_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; From 60cf89ae260c0c3330ef378f7deda06922b9d78b Mon Sep 17 00:00:00 2001 From: jjaw Date: Fri, 12 Jun 2026 13:51:27 +0800 Subject: [PATCH 3/5] fix: recover scheduler outbox invalid dedup index --- .../internal/repository/migrations_runner.go | 16 ++++++-- .../repository/migrations_runner_notx_test.go | 38 +++++++++++++++++++ 2 files changed, 50 insertions(+), 4 deletions(-) diff --git a/backend/internal/repository/migrations_runner.go b/backend/internal/repository/migrations_runner.go index 6dbb9fbd7c..155dfaa11a 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 = "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 } diff --git a/backend/internal/repository/migrations_runner_notx_test.go b/backend/internal/repository/migrations_runner_notx_test.go index b7cb396c47..f5453c4bd0 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("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) From b3ec6288adb263f8c8f9c4647502bc160a934f3e Mon Sep 17 00:00:00 2001 From: jjaw Date: Fri, 12 Jun 2026 16:22:43 +0800 Subject: [PATCH 4/5] fix: release scheduler outbox dedup on claim --- .../ops_write_pressure_integration_test.go | 27 ++++++++++++- .../repository/scheduler_outbox_repo.go | 40 +++++++++---------- backend/internal/service/scheduler_outbox.go | 3 +- .../service/scheduler_snapshot_service.go | 12 +----- 4 files changed, 48 insertions(+), 34 deletions(-) diff --git a/backend/internal/repository/ops_write_pressure_integration_test.go b/backend/internal/repository/ops_write_pressure_integration_test.go index 91987c01eb..0eb0385848 100644 --- a/backend/internal/repository/ops_write_pressure_integration_test.go +++ b/backend/internal/repository/ops_write_pressure_integration_test.go @@ -59,13 +59,38 @@ func TestEnqueueSchedulerOutbox_DeduplicatesIdempotentEvents(t *testing.T) { var firstID int64 require.NoError(t, integrationDB.QueryRowContext(ctx, "SELECT id FROM scheduler_outbox WHERE event_type = $1", service.SchedulerOutboxEventAccountChanged).Scan(&firstID)) - require.NoError(t, NewSchedulerOutboxRepository(integrationDB).MarkProcessed(ctx, []int64{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") diff --git a/backend/internal/repository/scheduler_outbox_repo.go b/backend/internal/repository/scheduler_outbox_repo.go index b60d1648f5..9edde6945c 100644 --- a/backend/internal/repository/scheduler_outbox_repo.go +++ b/backend/internal/repository/scheduler_outbox_repo.go @@ -10,7 +10,6 @@ import ( "strconv" "github.com/Wei-Shaw/sub2api/internal/service" - "github.com/lib/pq" ) type schedulerOutboxRepository struct { @@ -21,16 +20,30 @@ 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 @@ -81,19 +94,6 @@ func (r *schedulerOutboxRepository) MaxID(ctx context.Context) (int64, error) { return maxID, nil } -func (r *schedulerOutboxRepository) MarkProcessed(ctx context.Context, eventIDs []int64) error { - if len(eventIDs) == 0 { - return nil - } - _, err := r.db.ExecContext(ctx, ` - UPDATE scheduler_outbox - SET dedup_key = NULL - WHERE id = ANY($1) - AND dedup_key IS NOT NULL - `, pq.Array(eventIDs)) - return err -} - func enqueueSchedulerOutbox(ctx context.Context, exec sqlExecutor, eventType string, accountID *int64, groupID *int64, payload any) error { if exec == nil { return nil diff --git a/backend/internal/service/scheduler_outbox.go b/backend/internal/service/scheduler_outbox.go index 0fe977d25e..c138b7e5a5 100644 --- a/backend/internal/service/scheduler_outbox.go +++ b/backend/internal/service/scheduler_outbox.go @@ -16,7 +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) - MarkProcessed(ctx context.Context, eventIDs []int64) error } diff --git a/backend/internal/service/scheduler_snapshot_service.go b/backend/internal/service/scheduler_snapshot_service.go index 23025ad719..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 @@ -253,7 +253,6 @@ func (s *SchedulerSnapshotService) pollOutbox() { watermarkForCheck := watermark seen := make(map[batchSeenKey]struct{}) - processedIDs := make([]int64, 0, len(events)) for _, event := range events { eventCtx, cancel := context.WithTimeout(context.Background(), outboxEventTimeout) err := s.handleOutboxEvent(eventCtx, event, seen) @@ -262,15 +261,6 @@ func (s *SchedulerSnapshotService) pollOutbox() { logger.LegacyPrintf("service.scheduler_snapshot", "[Scheduler] outbox handle failed: id=%d type=%s err=%v", event.ID, event.EventType, err) return } - processedIDs = append(processedIDs, event.ID) - } - - markCtx, markCancel := context.WithTimeout(context.Background(), 5*time.Second) - markErr := s.outboxRepo.MarkProcessed(markCtx, processedIDs) - markCancel() - if markErr != nil { - logger.LegacyPrintf("service.scheduler_snapshot", "[Scheduler] outbox mark processed failed: %v", markErr) - return } lastID := events[len(events)-1].ID From 1fdbe52f9619611f459bcaeb1df5b950c4949b25 Mon Sep 17 00:00:00 2001 From: shaw Date: Tue, 16 Jun 2026 11:50:25 +0800 Subject: [PATCH 5/5] chore(migrations): renumber scheduler outbox dedup migrations 151/152 -> 152/153 #3232 (already merged) occupies 151. Renumber #3255's dedup migrations to avoid collision and keep linear ordering. Updated runner constant + tests. --- backend/internal/repository/migrations_runner.go | 2 +- backend/internal/repository/migrations_runner_notx_test.go | 6 +++--- ...box_dedup_key.sql => 152_scheduler_outbox_dedup_key.sql} | 0 ...> 153_scheduler_outbox_pending_dedup_key_index_notx.sql} | 0 4 files changed, 4 insertions(+), 4 deletions(-) rename backend/migrations/{151_scheduler_outbox_dedup_key.sql => 152_scheduler_outbox_dedup_key.sql} (100%) rename backend/migrations/{152_scheduler_outbox_pending_dedup_key_index_notx.sql => 153_scheduler_outbox_pending_dedup_key_index_notx.sql} (100%) diff --git a/backend/internal/repository/migrations_runner.go b/backend/internal/repository/migrations_runner.go index 155dfaa11a..285326537d 100644 --- a/backend/internal/repository/migrations_runner.go +++ b/backend/internal/repository/migrations_runner.go @@ -53,7 +53,7 @@ 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 schedulerOutboxPendingDedupKeyMigration = "153_scheduler_outbox_pending_dedup_key_index_notx.sql" const schedulerOutboxPendingDedupKeyIndex = "idx_scheduler_outbox_pending_dedup_key" type migrationChecksumCompatibilityRule struct { diff --git a/backend/internal/repository/migrations_runner_notx_test.go b/backend/internal/repository/migrations_runner_notx_test.go index f5453c4bd0..c9f6a2cdf1 100644 --- a/backend/internal/repository/migrations_runner_notx_test.go +++ b/backend/internal/repository/migrations_runner_notx_test.go @@ -201,7 +201,7 @@ func TestApplyMigrationsFS_SchedulerOutboxPendingDedupKeyMigration_DropsInvalidI prepareMigrationsBootstrapExpectations(mock) mock.ExpectQuery("SELECT checksum FROM schema_migrations WHERE filename = \\$1"). - WithArgs("152_scheduler_outbox_pending_dedup_key_index_notx.sql"). + WithArgs("153_scheduler_outbox_pending_dedup_key_index_notx.sql"). WillReturnError(sql.ErrNoRows) mock.ExpectQuery("SELECT EXISTS \\("). WithArgs("idx_scheduler_outbox_pending_dedup_key"). @@ -211,14 +211,14 @@ func TestApplyMigrationsFS_SchedulerOutboxPendingDedupKeyMigration_DropsInvalidI 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()). + 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{ - "152_scheduler_outbox_pending_dedup_key_index_notx.sql": &fstest.MapFile{ + "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) diff --git a/backend/migrations/151_scheduler_outbox_dedup_key.sql b/backend/migrations/152_scheduler_outbox_dedup_key.sql similarity index 100% rename from backend/migrations/151_scheduler_outbox_dedup_key.sql rename to backend/migrations/152_scheduler_outbox_dedup_key.sql diff --git a/backend/migrations/152_scheduler_outbox_pending_dedup_key_index_notx.sql b/backend/migrations/153_scheduler_outbox_pending_dedup_key_index_notx.sql similarity index 100% rename from backend/migrations/152_scheduler_outbox_pending_dedup_key_index_notx.sql rename to backend/migrations/153_scheduler_outbox_pending_dedup_key_index_notx.sql