From a0778e9a42bcce7948922c1f74cf839a7ae331f3 Mon Sep 17 00:00:00 2001 From: jjaw Date: Mon, 13 Jul 2026 21:56:47 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E8=B0=83=E5=BA=A6=E5=99=A8?= =?UTF-8?q?=E4=BD=BF=E7=94=A8=E5=B7=B2=E6=B6=88=E8=B4=B9=E4=BA=8B=E4=BB=B6?= =?UTF-8?q?=E8=AE=A1=E7=AE=97=E5=BB=B6=E8=BF=9F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../repository/scheduler_outbox_repo.go | 18 +++++ .../repository/scheduler_outbox_repo_test.go | 52 ++++++++++++++ backend/internal/service/scheduler_outbox.go | 2 + .../scheduler_snapshot_outbox_cleanup_test.go | 71 ++++++++++++++++--- .../service/scheduler_snapshot_service.go | 31 +++++--- 5 files changed, 156 insertions(+), 18 deletions(-) diff --git a/backend/internal/repository/scheduler_outbox_repo.go b/backend/internal/repository/scheduler_outbox_repo.go index 59772fbb5a..500f6e850d 100644 --- a/backend/internal/repository/scheduler_outbox_repo.go +++ b/backend/internal/repository/scheduler_outbox_repo.go @@ -93,6 +93,24 @@ func (r *schedulerOutboxRepository) ListAfterAndReleaseDedup(ctx context.Context return events, nil } +func (r *schedulerOutboxRepository) FirstCreatedAtAfter(ctx context.Context, afterID int64) (time.Time, bool, error) { + var createdAt time.Time + err := r.db.QueryRowContext(ctx, ` + SELECT created_at + FROM scheduler_outbox + WHERE id > $1 + ORDER BY id ASC + LIMIT 1 + `, afterID).Scan(&createdAt) + if err == sql.ErrNoRows { + return time.Time{}, false, nil + } + if err != nil { + return time.Time{}, false, err + } + return createdAt, true, nil +} + func (r *schedulerOutboxRepository) MaxID(ctx context.Context) (int64, error) { var maxID int64 if err := r.db.QueryRowContext(ctx, "SELECT COALESCE(MAX(id), 0) FROM scheduler_outbox").Scan(&maxID); err != nil { diff --git a/backend/internal/repository/scheduler_outbox_repo_test.go b/backend/internal/repository/scheduler_outbox_repo_test.go index 619339d207..250014f6ee 100644 --- a/backend/internal/repository/scheduler_outbox_repo_test.go +++ b/backend/internal/repository/scheduler_outbox_repo_test.go @@ -4,11 +4,63 @@ import ( "context" "regexp" "testing" + "time" sqlmock "github.com/DATA-DOG/go-sqlmock" "github.com/stretchr/testify/require" ) +func TestSchedulerOutboxRepositoryFirstCreatedAtAfter(t *testing.T) { + db, mock, err := sqlmock.New() + require.NoError(t, err) + defer func() { _ = db.Close() }() + + repo := &schedulerOutboxRepository{db: db} + createdAt := time.Now().UTC().Truncate(time.Microsecond) + const expectedSQL = ` + SELECT created_at + FROM scheduler_outbox + WHERE id > $1 + ORDER BY id ASC + LIMIT 1 + ` + mock.ExpectQuery(regexp.QuoteMeta(expectedSQL)). + WithArgs(int64(42)). + WillReturnRows(sqlmock.NewRows([]string{"created_at"}).AddRow(createdAt)) + + got, ok, err := repo.FirstCreatedAtAfter(context.Background(), 42) + + require.NoError(t, err) + require.True(t, ok) + require.Equal(t, createdAt, got) + require.NoError(t, mock.ExpectationsWereMet()) +} + +func TestSchedulerOutboxRepositoryFirstCreatedAtAfterReturnsNotFound(t *testing.T) { + db, mock, err := sqlmock.New() + require.NoError(t, err) + defer func() { _ = db.Close() }() + + repo := &schedulerOutboxRepository{db: db} + const expectedSQL = ` + SELECT created_at + FROM scheduler_outbox + WHERE id > $1 + ORDER BY id ASC + LIMIT 1 + ` + mock.ExpectQuery(regexp.QuoteMeta(expectedSQL)). + WithArgs(int64(42)). + WillReturnRows(sqlmock.NewRows([]string{"created_at"})) + + got, ok, err := repo.FirstCreatedAtAfter(context.Background(), 42) + + require.NoError(t, err) + require.False(t, ok) + require.True(t, got.IsZero()) + require.NoError(t, mock.ExpectationsWereMet()) +} + func TestSchedulerOutboxRepositoryDeleteConsumedUpToUsesBoundedCTE(t *testing.T) { db, mock, err := sqlmock.New() require.NoError(t, err) diff --git a/backend/internal/service/scheduler_outbox.go b/backend/internal/service/scheduler_outbox.go index 2b7665ad78..a44f2a3a30 100644 --- a/backend/internal/service/scheduler_outbox.go +++ b/backend/internal/service/scheduler_outbox.go @@ -17,6 +17,8 @@ type SchedulerOutboxEvent struct { // SchedulerOutboxRepository 提供调度 outbox 的读取接口。 type SchedulerOutboxRepository interface { ListAfterAndReleaseDedup(ctx context.Context, afterID int64, limit int) ([]SchedulerOutboxEvent, error) + // FirstCreatedAtAfter 返回指定水位之后第一条待消费事件的创建时间，不领取事件或修改去重键。 + FirstCreatedAtAfter(ctx context.Context, afterID int64) (time.Time, bool, error) MaxID(ctx context.Context) (int64, error) DeleteConsumedUpTo(ctx context.Context, watermark int64, limit int) (int64, error) TryAcquireCleanupLock(ctx context.Context) (SchedulerOutboxCleanupLease, bool, error) diff --git a/backend/internal/service/scheduler_snapshot_outbox_cleanup_test.go b/backend/internal/service/scheduler_snapshot_outbox_cleanup_test.go index 535f8d54e2..91e2d36f76 100644 --- a/backend/internal/service/scheduler_snapshot_outbox_cleanup_test.go +++ b/backend/internal/service/scheduler_snapshot_outbox_cleanup_test.go @@ -6,12 +6,15 @@ import ( "reflect" "testing" "time" + + "github.com/Wei-Shaw/sub2api/internal/config" ) type outboxCleanupCache struct { - watermark int64 - setWatermarks []int64 - updateErr error + watermark int64 + setWatermarks []int64 + updateErr error + listBucketCalls int } func (c *outboxCleanupCache) GetSnapshot(ctx context.Context, bucket SchedulerBucket) ([]*Account, bool, error) { @@ -47,6 +50,7 @@ func (c *outboxCleanupCache) UnlockBucket(ctx context.Context, bucket SchedulerB } func (c *outboxCleanupCache) ListBuckets(ctx context.Context) ([]SchedulerBucket, error) { + c.listBucketCalls++ return nil, nil } @@ -66,12 +70,13 @@ type outboxCleanupDeleteCall struct { } type outboxCleanupRepo struct { - events []SchedulerOutboxEvent - rows []int64 - lockAcquired bool - lockAttempts int - releaseCount int - deleteCalls []outboxCleanupDeleteCall + events []SchedulerOutboxEvent + rows []int64 + lockAcquired bool + lockAttempts int + releaseCount int + deleteCalls []outboxCleanupDeleteCall + firstCreatedAfterID []int64 } func (r *outboxCleanupRepo) ListAfterAndReleaseDedup(ctx context.Context, afterID int64, limit int) ([]SchedulerOutboxEvent, error) { @@ -88,6 +93,16 @@ func (r *outboxCleanupRepo) ListAfterAndReleaseDedup(ctx context.Context, afterI return events, nil } +func (r *outboxCleanupRepo) FirstCreatedAtAfter(ctx context.Context, afterID int64) (time.Time, bool, error) { + r.firstCreatedAfterID = append(r.firstCreatedAfterID, afterID) + for _, event := range r.events { + if event.ID > afterID { + return event.CreatedAt, true, nil + } + } + return time.Time{}, false, nil +} + func (r *outboxCleanupRepo) MaxID(ctx context.Context) (int64, error) { var maxID int64 for _, id := range r.rows { @@ -240,6 +255,44 @@ func TestSchedulerSnapshotServicePollOutboxDoesNotCleanupOnHandleFailure(t *test } } +func TestSchedulerSnapshotServicePollOutboxDoesNotUseConsumedEventForLag(t *testing.T) { + cache := &outboxCleanupCache{} + repo := &outboxCleanupRepo{ + events: []SchedulerOutboxEvent{ + { + ID: 7, + EventType: SchedulerOutboxEventAccountLastUsed, + CreatedAt: time.Now().Add(-time.Hour), + }, + }, + } + cfg := &config.Config{ + Gateway: config.GatewayConfig{ + Scheduling: config.GatewaySchedulingConfig{ + OutboxLagWarnSeconds: 1, + OutboxLagRebuildSeconds: 1, + OutboxLagRebuildFailures: 1, + }, + }, + } + svc := NewSchedulerSnapshotService(cache, repo, nil, nil, cfg) + + svc.pollOutbox() + + if cache.watermark != 7 { + t.Fatalf("expected watermark 7, got %d", cache.watermark) + } + if !reflect.DeepEqual(repo.firstCreatedAfterID, []int64{7}) { + t.Fatalf("expected lag check after consumed watermark, got %#v", repo.firstCreatedAfterID) + } + if cache.listBucketCalls != 0 { + t.Fatalf("expected consumed event not to trigger full rebuild, got %d attempts", cache.listBucketCalls) + } + if svc.lagFailures != 0 { + t.Fatalf("expected lag failures to remain reset, got %d", svc.lagFailures) + } +} + func TestSchedulerSnapshotServiceCleanupSkipsNonPositiveWatermark(t *testing.T) { repo := &outboxCleanupRepo{ rows: []int64{1, 2, 3}, diff --git a/backend/internal/service/scheduler_snapshot_service.go b/backend/internal/service/scheduler_snapshot_service.go index dc514bc851..e19150ad02 100644 --- a/backend/internal/service/scheduler_snapshot_service.go +++ b/backend/internal/service/scheduler_snapshot_service.go @@ -254,7 +254,6 @@ func (s *SchedulerSnapshotService) pollOutbox() { return } - watermarkForCheck := watermark seen := make(map[batchSeenKey]struct{}) for _, event := range events { eventCtx, cancel := context.WithTimeout(context.Background(), outboxEventTimeout) @@ -281,12 +280,15 @@ func (s *SchedulerSnapshotService) pollOutbox() { } if wmErr != nil { logger.LegacyPrintf("service.scheduler_snapshot", "[Scheduler] outbox watermark write failed: %v", wmErr) - } else { - watermarkForCheck = lastID - s.cleanupConsumedOutbox(lastID) + return } + s.cleanupConsumedOutbox(lastID) - s.checkOutboxLag(ctx, events[0], watermarkForCheck) + // 只有 watermark 成功推进后，当前批次才算已消费。延迟必须按下一条待消费事件计算， + // 否则本批次处理越慢，越容易误触发一次更慢的全量重建，形成正反馈。 + lagCtx, lagCancel := context.WithTimeout(context.Background(), 5*time.Second) + s.checkOutboxLag(lagCtx, lastID) + lagCancel() } func (s *SchedulerSnapshotService) cleanupConsumedOutbox(watermark int64) { @@ -620,12 +622,23 @@ func (s *SchedulerSnapshotService) triggerFullRebuild(reason string) error { return s.rebuildBuckets(ctx, buckets, reason) } -func (s *SchedulerSnapshotService) checkOutboxLag(ctx context.Context, oldest SchedulerOutboxEvent, watermark int64) { - if oldest.CreatedAt.IsZero() || s.cfg == nil { +func (s *SchedulerSnapshotService) checkOutboxLag(ctx context.Context, watermark int64) { + if s.cfg == nil || s.outboxRepo == nil { + return + } + oldestCreatedAt, ok, err := s.outboxRepo.FirstCreatedAtAfter(ctx, watermark) + if err != nil { + logger.LegacyPrintf("service.scheduler_snapshot", "[Scheduler] outbox pending event read failed: %v", err) + return + } + if !ok || oldestCreatedAt.IsZero() { + s.lagMu.Lock() + s.lagFailures = 0 + s.lagMu.Unlock() return } - lag := time.Since(oldest.CreatedAt) + lag := time.Since(oldestCreatedAt) if lagSeconds := int(lag.Seconds()); lagSeconds >= s.cfg.Gateway.Scheduling.OutboxLagWarnSeconds && s.cfg.Gateway.Scheduling.OutboxLagWarnSeconds > 0 { logger.LegacyPrintf("service.scheduler_snapshot", "[Scheduler] outbox lag warning: %ds", lagSeconds) } @@ -652,7 +665,7 @@ func (s *SchedulerSnapshotService) checkOutboxLag(ctx context.Context, oldest Sc } threshold := s.cfg.Gateway.Scheduling.OutboxBacklogRebuildRows - if threshold <= 0 || s.outboxRepo == nil { + if threshold <= 0 { return } maxID, err := s.outboxRepo.MaxID(ctx)