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 #4203 from jianjianai/codex/fix-scheduler-pending-lag

Browse Source 
修复调度器使用已消费事件计算 outbox 延迟
This commit is contained in:
Wesley Liddick
2026-07-14 10:59:57 +08:00
committed by GitHub
parent 9c3c560d49 a0778e9a42
commit 24d908b257
5 changed files with 156 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/internal/repository/scheduler_outbox_repo.go
+18
View File
@@ -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 {
backend/internal/repository/scheduler_outbox_repo_test.go
+52
View File
@@ -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)
backend/internal/service/scheduler_outbox.go
+2
View File
@@ -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)
backend/internal/service/scheduler_snapshot_outbox_cleanup_test.go
+62 -9
View File
@@ -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},
backend/internal/service/scheduler_snapshot_service.go
+22 -9
View File
@@ -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)