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

Merge pull request #3255 from jianjianai/fix1/outbox-scheduler-snapshot-coalesce

Browse Source 
降低高频账号状态变更下的 outbox 写入压力 / safely coalesce scheduler outbox events

Migrations renumbered 151/152 -> 152/153 to avoid collision with #3232.
This commit is contained in:
shaw
2026-06-16 11:50:35 +08:00
parent 45f3b0dd74 1fdbe52f96
commit 4254dcb352
10 changed files with 198 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/internal/repository/account_repo_integration_test.go
+31
View File
@@ -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",
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 = "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
}
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("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)
backend/internal/repository/migrations_schema_integration_test.go
+4
View File
@@ -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))
backend/internal/repository/ops_write_pressure_integration_test.go
+59 -1
View File
@@ -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")
backend/internal/repository/scheduler_outbox_repo.go
+47 -20
View File
@@ -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,
backend/internal/service/scheduler_outbox.go
+1 -1
View File
@@ -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)
}
backend/internal/service/scheduler_snapshot_service.go
+1 -1
View File
@@ -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
backend/migrations/152_scheduler_outbox_dedup_key.sql
+2
View File
@@ -0,0 +1,2 @@
ALTER TABLE scheduler_outbox
ADD COLUMN IF NOT EXISTS dedup_key TEXT;
backend/migrations/153_scheduler_outbox_pending_dedup_key_index_notx.sql
+3
View File
@@ -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;