From 362f9e77bf189a17ff12970b6dee6332a4f9fefb Mon Sep 17 00:00:00 2001 From: visa2 Date: Fri, 5 Jun 2026 21:15:43 +0800 Subject: [PATCH] fix(service): gate periodic background jobs with a leader lock for multi-instance MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three periodic background jobs ran on every instance with no cross-instance coordination, multiplying their cost (and side effects) by the replica count: - DashboardAggregationService.runScheduledAggregation: N× heavy GROUP BY aggregation queries every minute plus watermark write races. - PaymentOrderExpiryService.runOnce: N× upstream payment-provider reconcile/ expiry API calls per pending order. - SubscriptionExpiryService.sendExpiryReminders: N× full active-subscription scans every minute and potential duplicate reminder emails. Add a LeaderLockCache abstraction so only one instance runs each job per cycle: - The interface lives in the service layer; the Redis-backed implementation (SetNX + compare-and-delete release) lives in the repository layer, so the service package keeps its depguard "must not import redis" boundary intact. - tryAcquireSingletonLeaderLock prefers the cache and falls back to a Postgres advisory lock when Redis errors, mirroring the Ops background services. When neither backend is configured the job runs ungated, preserving single-instance and test behavior (no self-lockout: the lock is released every cycle). Co-Authored-By: Claude Opus 4.8 --- backend/cmd/server/wire_gen.go | 7 +- .../internal/repository/leader_lock_cache.go | 42 +++++ .../repository/leader_lock_cache_test.go | 79 ++++++++++ backend/internal/repository/wire.go | 1 + .../service/dashboard_aggregation_service.go | 33 ++++ backend/internal/service/leader_lock.go | 68 +++++++++ backend/internal/service/leader_lock_test.go | 144 ++++++++++++++++++ .../service/payment_order_expiry_service.go | 38 +++++ .../service/subscription_expiry_service.go | 36 +++++ backend/internal/service/wire.go | 9 +- 10 files changed, 451 insertions(+), 6 deletions(-) create mode 100644 backend/internal/repository/leader_lock_cache.go create mode 100644 backend/internal/repository/leader_lock_cache_test.go create mode 100644 backend/internal/service/leader_lock.go create mode 100644 backend/internal/service/leader_lock_test.go diff --git a/backend/cmd/server/wire_gen.go b/backend/cmd/server/wire_gen.go index 814c07e6b1..806d31301c 100644 --- a/backend/cmd/server/wire_gen.go +++ b/backend/cmd/server/wire_gen.go @@ -167,7 +167,8 @@ func initializeApplication(buildInfo handler.BuildInfo) (*Application, error) { dashboardAggregationRepository := repository.NewDashboardAggregationRepository(db) dashboardStatsCache := repository.NewDashboardCache(redisClient, configConfig) dashboardService := service.NewDashboardService(usageLogRepository, dashboardAggregationRepository, dashboardStatsCache, configConfig) - dashboardAggregationService := service.ProvideDashboardAggregationService(dashboardAggregationRepository, timingWheelService, configConfig) + leaderLockCache := repository.NewLeaderLockCache(redisClient) + dashboardAggregationService := service.ProvideDashboardAggregationService(dashboardAggregationRepository, timingWheelService, leaderLockCache, db, configConfig) dashboardHandler := admin.NewDashboardHandler(dashboardService, dashboardAggregationService) proxyExitInfoProber := repository.NewProxyExitInfoProber(configConfig) proxyLatencyCache := repository.NewProxyLatencyCache(redisClient) @@ -265,9 +266,9 @@ func initializeApplication(buildInfo handler.BuildInfo) (*Application, error) { opsScheduledReportService := service.ProvideOpsScheduledReportService(opsService, userService, emailService, redisClient, configConfig) tokenRefreshService := service.ProvideTokenRefreshService(accountRepository, oAuthService, openAIOAuthService, geminiOAuthService, antigravityOAuthService, compositeTokenCacheInvalidator, schedulerCache, configConfig, tempUnschedCache, privacyClientFactory, proxyRepository, oAuthRefreshAPI, openAIGatewayService) accountExpiryService := service.ProvideAccountExpiryService(accountRepository) - subscriptionExpiryService := service.ProvideSubscriptionExpiryService(userSubscriptionRepository, settingRepository, notificationEmailService) + subscriptionExpiryService := service.ProvideSubscriptionExpiryService(userSubscriptionRepository, settingRepository, notificationEmailService, leaderLockCache, db) scheduledTestRunnerService := service.ProvideScheduledTestRunnerService(scheduledTestPlanRepository, scheduledTestService, accountTestService, rateLimitService, configConfig) - paymentOrderExpiryService := service.ProvidePaymentOrderExpiryService(paymentService) + paymentOrderExpiryService := service.ProvidePaymentOrderExpiryService(paymentService, leaderLockCache, db) channelMonitorRunner := service.ProvideChannelMonitorRunner(channelMonitorService, settingService) userPlatformQuotaUsageFlusher := service.ProvideUserPlatformQuotaUsageFlusher(configConfig, billingCache, serviceUserPlatformQuotaRepository, timingWheelService) v := provideCleanup(client, redisClient, opsMetricsCollector, opsAggregationService, opsAlertEvaluatorService, opsCleanupService, opsScheduledReportService, opsSystemLogSink, schedulerSnapshotService, tokenRefreshService, accountExpiryService, subscriptionExpiryService, usageCleanupService, idempotencyCleanupService, pricingService, emailQueueService, billingCacheService, usageRecordWorkerPool, subscriptionService, oAuthService, openAIOAuthService, geminiOAuthService, antigravityOAuthService, openAIGatewayService, scheduledTestRunnerService, backupService, paymentOrderExpiryService, channelMonitorRunner, userPlatformQuotaUsageFlusher) diff --git a/backend/internal/repository/leader_lock_cache.go b/backend/internal/repository/leader_lock_cache.go new file mode 100644 index 0000000000..0a9f985aab --- /dev/null +++ b/backend/internal/repository/leader_lock_cache.go @@ -0,0 +1,42 @@ +package repository + +import ( + "context" + "time" + + "github.com/Wei-Shaw/sub2api/internal/service" + + "github.com/redis/go-redis/v9" +) + +const leaderLockKeyPrefix = "leader:lock:" + +// leaderLockReleaseScript releases a leader lock only when the caller still owns +// it (compare-and-delete by owner token). This prevents a previous holder whose +// lock already expired — and was re-acquired by another instance — from deleting +// the new owner's lock. +var leaderLockReleaseScript = redis.NewScript(` +if redis.call("GET", KEYS[1]) == ARGV[1] then + return redis.call("DEL", KEYS[1]) +end +return 0 +`) + +type leaderLockCache struct { + rdb *redis.Client +} + +// NewLeaderLockCache returns a Redis-backed implementation of +// service.LeaderLockCache used by periodic background jobs to elect a single +// runner across instances. +func NewLeaderLockCache(rdb *redis.Client) service.LeaderLockCache { + return &leaderLockCache{rdb: rdb} +} + +func (c *leaderLockCache) TryAcquireLeaderLock(ctx context.Context, key, owner string, ttl time.Duration) (bool, error) { + return c.rdb.SetNX(ctx, leaderLockKeyPrefix+key, owner, ttl).Result() +} + +func (c *leaderLockCache) ReleaseLeaderLock(ctx context.Context, key, owner string) error { + return leaderLockReleaseScript.Run(ctx, c.rdb, []string{leaderLockKeyPrefix + key}, owner).Err() +} diff --git a/backend/internal/repository/leader_lock_cache_test.go b/backend/internal/repository/leader_lock_cache_test.go new file mode 100644 index 0000000000..4abd7b3fe8 --- /dev/null +++ b/backend/internal/repository/leader_lock_cache_test.go @@ -0,0 +1,79 @@ +//go:build unit + +package repository + +import ( + "context" + "testing" + "time" + + "github.com/alicebob/miniredis/v2" + "github.com/redis/go-redis/v9" + "github.com/stretchr/testify/require" +) + +func newLeaderLockTestCache(t *testing.T) (*leaderLockCache, *miniredis.Miniredis) { + t.Helper() + mr := miniredis.RunT(t) + rdb := redis.NewClient(&redis.Options{Addr: mr.Addr()}) + t.Cleanup(func() { _ = rdb.Close() }) + return &leaderLockCache{rdb: rdb}, mr +} + +func TestLeaderLockCache_AcquireContendedRelease(t *testing.T) { + cache, _ := newLeaderLockTestCache(t) + ctx := context.Background() + const key = "dashboard:aggregation:leader" + + ok, err := cache.TryAcquireLeaderLock(ctx, key, "A", time.Minute) + require.NoError(t, err) + require.True(t, ok, "first owner should acquire") + + ok, err = cache.TryAcquireLeaderLock(ctx, key, "B", time.Minute) + require.NoError(t, err) + require.False(t, ok, "peer must be locked out while held") + + require.NoError(t, cache.ReleaseLeaderLock(ctx, key, "A")) + + ok, err = cache.TryAcquireLeaderLock(ctx, key, "B", time.Minute) + require.NoError(t, err) + require.True(t, ok, "peer should acquire after release") +} + +// A stale owner whose lock expired and was re-acquired by a peer must not delete +// the peer's lock when its late release fires (compare-and-delete by owner). +func TestLeaderLockCache_ReleaseIsCompareAndDelete(t *testing.T) { + cache, _ := newLeaderLockTestCache(t) + ctx := context.Background() + const key = "payment:order:expiry:leader" + + ok, err := cache.TryAcquireLeaderLock(ctx, key, "A", time.Minute) + require.NoError(t, err) + require.True(t, ok) + + // Simulate A's lock expiring and peer B taking ownership. + require.NoError(t, cache.rdb.Set(ctx, leaderLockKeyPrefix+key, "B", time.Minute).Err()) + + // A's stale release must be a no-op against B's lock. + require.NoError(t, cache.ReleaseLeaderLock(ctx, key, "A")) + + val, err := cache.rdb.Get(ctx, leaderLockKeyPrefix+key).Result() + require.NoError(t, err) + require.Equal(t, "B", val, "stale owner must not delete the new owner's lock") +} + +func TestLeaderLockCache_TTLExpires(t *testing.T) { + cache, mr := newLeaderLockTestCache(t) + ctx := context.Background() + const key = "subscription:expiry:reminder:leader" + + ok, err := cache.TryAcquireLeaderLock(ctx, key, "A", time.Minute) + require.NoError(t, err) + require.True(t, ok) + + mr.FastForward(2 * time.Minute) + + ok, err = cache.TryAcquireLeaderLock(ctx, key, "B", time.Minute) + require.NoError(t, err) + require.True(t, ok, "lock should be re-acquirable after the TTL expires") +} diff --git a/backend/internal/repository/wire.go b/backend/internal/repository/wire.go index 2d1b04e393..630c27ed47 100644 --- a/backend/internal/repository/wire.go +++ b/backend/internal/repository/wire.go @@ -115,6 +115,7 @@ var ProviderSet = wire.NewSet( NewRedeemCache, NewUpdateCache, NewGeminiTokenCache, + NewLeaderLockCache, ProvideSchedulerCache, NewSchedulerOutboxRepository, NewProxyLatencyCache, diff --git a/backend/internal/service/dashboard_aggregation_service.go b/backend/internal/service/dashboard_aggregation_service.go index b58a1ea93f..5dfa02a824 100644 --- a/backend/internal/service/dashboard_aggregation_service.go +++ b/backend/internal/service/dashboard_aggregation_service.go @@ -2,6 +2,7 @@ package service import ( "context" + "database/sql" "errors" "log/slog" "sync/atomic" @@ -9,12 +10,20 @@ import ( "github.com/Wei-Shaw/sub2api/internal/config" "github.com/Wei-Shaw/sub2api/internal/pkg/logger" + "github.com/google/uuid" ) const ( defaultDashboardAggregationTimeout = 2 * time.Minute defaultDashboardAggregationBackfillTimeout = 30 * time.Minute dashboardAggregationRetentionInterval = 6 * time.Hour + + // dashboardAggregationLeaderLockKey gates the periodic scheduled aggregation so + // that only one instance runs it per cycle in a multi-replica deployment. + dashboardAggregationLeaderLockKey = "dashboard:aggregation:leader" + // dashboardAggregationLeaderLockTTL must exceed the job's worst-case runtime + // (defaultDashboardAggregationTimeout) so the lock never expires mid-run. + dashboardAggregationLeaderLockTTL = 5 * time.Minute ) var ( @@ -46,6 +55,10 @@ type DashboardAggregationService struct { cfg config.DashboardAggregationConfig running int32 lastRetentionCleanup atomic.Value // time.Time + + lockCache LeaderLockCache + db *sql.DB + instanceID string } // NewDashboardAggregationService 创建聚合服务。 @@ -58,9 +71,21 @@ func NewDashboardAggregationService(repo DashboardAggregationRepository, timingW repo: repo, timingWheel: timingWheel, cfg: aggCfg, + instanceID: uuid.NewString(), } } +// SetLeaderLock injects the leader-lock cache and DB used to elect a single +// instance for the periodic scheduled aggregation. When both are nil the job runs +// ungated (single-instance / test behavior). +func (s *DashboardAggregationService) SetLeaderLock(lockCache LeaderLockCache, db *sql.DB) { + if s == nil { + return + } + s.lockCache = lockCache + s.db = db +} + // Start 启动定时聚合作业（重启生效配置）。 func (s *DashboardAggregationService) Start() { if s == nil || s.repo == nil || s.timingWheel == nil { @@ -197,6 +222,14 @@ func (s *DashboardAggregationService) runScheduledAggregation() { ctx, cancel := context.WithTimeout(context.Background(), defaultDashboardAggregationTimeout) defer cancel() + // Multi-instance guard: only the leader runs the periodic aggregation; peers + // skip this cycle to avoid N× redundant GROUP BY queries and watermark races. + release, ok := tryAcquireSingletonLeaderLock(ctx, s.lockCache, s.db, dashboardAggregationLeaderLockKey, s.instanceID, dashboardAggregationLeaderLockTTL) + if !ok { + return + } + defer release() + now := time.Now().UTC() last, err := s.repo.GetAggregationWatermark(ctx) if err != nil { diff --git a/backend/internal/service/leader_lock.go b/backend/internal/service/leader_lock.go new file mode 100644 index 0000000000..a0c774a653 --- /dev/null +++ b/backend/internal/service/leader_lock.go @@ -0,0 +1,68 @@ +package service + +import ( + "context" + "database/sql" + "time" +) + +// LeaderLockCache provides cross-instance mutual exclusion for periodic background +// jobs. It is implemented in the repository layer (Redis-backed) so the service +// layer never depends on Redis directly. Release is a compare-and-delete keyed by +// owner so a stale holder can never delete a peer's lock. +type LeaderLockCache interface { + // TryAcquireLeaderLock sets key=owner with the given TTL iff key is absent. + // It returns true when the caller becomes the owner. + TryAcquireLeaderLock(ctx context.Context, key, owner string, ttl time.Duration) (bool, error) + // ReleaseLeaderLock deletes key iff it is still owned by owner. + ReleaseLeaderLock(ctx context.Context, key, owner string) error +} + +// tryAcquireSingletonLeaderLock provides best-effort single-flight execution of a +// periodic background job across multiple instances. It prefers the Redis-backed +// LeaderLockCache and falls back to a Postgres advisory lock when the cache is +// unavailable or errors, mirroring the approach used by the Ops background +// services. +// +// Semantics: +// - acquired -> returns a non-nil release func and true; callers should +// defer the release once the job finishes. +// - held by peer -> returns (nil, false); callers should skip this cycle. +// - no backend -> when neither the cache nor a DB is configured (e.g. unit +// tests, or a single-instance deployment without Redis) it runs without +// gating, returning a no-op release and true, so the job is never silently +// starved. +// +// The TTL is purely a crash-safety bound: callers release the lock as soon as the +// job completes, so leadership is re-contested every cycle rather than pinned to +// one instance. The TTL must therefore be larger than the job's worst-case +// runtime so the lock does not expire mid-run. +func tryAcquireSingletonLeaderLock(ctx context.Context, cache LeaderLockCache, db *sql.DB, key, owner string, ttl time.Duration) (func(), bool) { + if ctx == nil { + ctx = context.Background() + } + + if cache != nil { + ok, err := cache.TryAcquireLeaderLock(ctx, key, owner, ttl) + if err == nil { + if !ok { + return nil, false + } + release := func() { + ctx2, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + _ = cache.ReleaseLeaderLock(ctx2, key, owner) + } + return release, true + } + // Cache error: fall through to the DB advisory lock so a flaky Redis does + // not stampede the job across every instance. + } + + if db != nil { + return tryAcquireDBAdvisoryLock(ctx, db, hashAdvisoryLockID(key)) + } + + // No coordination backend available: run without gating. + return func() {}, true +} diff --git a/backend/internal/service/leader_lock_test.go b/backend/internal/service/leader_lock_test.go new file mode 100644 index 0000000000..aba1c62c3e --- /dev/null +++ b/backend/internal/service/leader_lock_test.go @@ -0,0 +1,144 @@ +package service + +import ( + "context" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// fakeLeaderLockCache is an in-memory LeaderLockCache for unit tests. It models the +// compare-and-delete release semantics of the real Redis-backed implementation. +type fakeLeaderLockCache struct { + mu sync.Mutex + owners map[string]string + acquireErr error +} + +func (f *fakeLeaderLockCache) TryAcquireLeaderLock(_ context.Context, key, owner string, _ time.Duration) (bool, error) { + if f.acquireErr != nil { + return false, f.acquireErr + } + f.mu.Lock() + defer f.mu.Unlock() + if f.owners == nil { + f.owners = map[string]string{} + } + if _, held := f.owners[key]; held { + return false, nil + } + f.owners[key] = owner + return true, nil +} + +func (f *fakeLeaderLockCache) ReleaseLeaderLock(_ context.Context, key, owner string) error { + f.mu.Lock() + defer f.mu.Unlock() + if f.owners[key] == owner { + delete(f.owners, key) + } + return nil +} + +func (f *fakeLeaderLockCache) heldBy(key string) string { + f.mu.Lock() + defer f.mu.Unlock() + return f.owners[key] +} + +func TestTryAcquireSingletonLeaderLock_NoBackendRunsUngated(t *testing.T) { + release, ok := tryAcquireSingletonLeaderLock(context.Background(), nil, nil, "k", "inst", time.Minute) + require.True(t, ok) + require.NotNil(t, release) + require.NotPanics(t, release) +} + +func TestTryAcquireSingletonLeaderLock_ContendedThenReleased(t *testing.T) { + cache := &fakeLeaderLockCache{} + ctx := context.Background() + const key = "leader:test:contended" + + releaseA, ok := tryAcquireSingletonLeaderLock(ctx, cache, nil, key, "A", time.Minute) + require.True(t, ok, "first instance should acquire") + require.Equal(t, "A", cache.heldBy(key)) + + _, okB := tryAcquireSingletonLeaderLock(ctx, cache, nil, key, "B", time.Minute) + require.False(t, okB, "peer must be locked out while the lock is held") + + releaseA() + require.Empty(t, cache.heldBy(key), "release must free the lock") + + releaseB, okB := tryAcquireSingletonLeaderLock(ctx, cache, nil, key, "B", time.Minute) + require.True(t, okB, "peer should acquire after the holder releases") + releaseB() +} + +// When the cache errors, the helper must fall through rather than acquire via the +// cache. With no DB configured it runs ungated so the job is never starved by a +// flaky Redis. +func TestTryAcquireSingletonLeaderLock_CacheErrorFallsThrough(t *testing.T) { + cache := &fakeLeaderLockCache{acquireErr: context.DeadlineExceeded} + release, ok := tryAcquireSingletonLeaderLock(context.Background(), cache, nil, "k", "inst", time.Minute) + require.True(t, ok, "cache error with no DB must run ungated, not skip") + require.NotNil(t, release) + require.NotPanics(t, release) +} + +func TestSubscriptionExpiryService_ReminderSkipsScanWhenNotLeader(t *testing.T) { + cache := &fakeLeaderLockCache{} + // A peer already holds the reminder leader lock. + _, _ = cache.TryAcquireLeaderLock(context.Background(), subscriptionExpiryReminderLeaderLockKey, "peer", time.Minute) + + repo := &subscriptionExpiryRepoStub{} + settingRepo := &subscriptionExpirySettingRepoStub{values: map[string]string{}} + svc := NewSubscriptionExpiryService(repo, time.Minute) + svc.SetSettingRepository(settingRepo) + svc.SetNotificationEmailService(NewNotificationEmailService(settingRepo, nil)) + svc.SetLeaderLock(cache, nil) + + svc.sendExpiryReminders(context.Background()) + + require.Zero(t, repo.listCalls, "non-leader must not scan active subscriptions") +} + +func TestSubscriptionExpiryService_ReminderScansWhenLeader(t *testing.T) { + repo := &subscriptionExpiryRepoStub{} + settingRepo := &subscriptionExpirySettingRepoStub{values: map[string]string{}} + svc := NewSubscriptionExpiryService(repo, time.Minute) + svc.SetSettingRepository(settingRepo) + svc.SetNotificationEmailService(NewNotificationEmailService(settingRepo, nil)) + svc.SetLeaderLock(&fakeLeaderLockCache{}, nil) + + svc.sendExpiryReminders(context.Background()) + + require.Equal(t, 1, repo.listCalls, "leader should scan active subscriptions once") +} + +// Single-instance correctness: the lock is released at the end of each cycle, so +// the same instance must re-acquire it and run on every subsequent cycle (no +// self-lockout). Covers both the cache-backed path and the no-backend path. +func TestSubscriptionExpiryService_ReminderRunsEveryCycleSingleInstance(t *testing.T) { + cases := map[string]LeaderLockCache{ + "with_cache": &fakeLeaderLockCache{}, + "no_backend": nil, + } + for name, cache := range cases { + t.Run(name, func(t *testing.T) { + repo := &subscriptionExpiryRepoStub{} + settingRepo := &subscriptionExpirySettingRepoStub{values: map[string]string{}} + svc := NewSubscriptionExpiryService(repo, time.Minute) + svc.SetSettingRepository(settingRepo) + svc.SetNotificationEmailService(NewNotificationEmailService(settingRepo, nil)) + svc.SetLeaderLock(cache, nil) + + // Three consecutive cycles, mimicking the ticker loop. + svc.sendExpiryReminders(context.Background()) + svc.sendExpiryReminders(context.Background()) + svc.sendExpiryReminders(context.Background()) + + require.Equal(t, 3, repo.listCalls, "single instance must run every cycle") + }) + } +} diff --git a/backend/internal/service/payment_order_expiry_service.go b/backend/internal/service/payment_order_expiry_service.go index 32e51d7fb2..b292ef89ae 100644 --- a/backend/internal/service/payment_order_expiry_service.go +++ b/backend/internal/service/payment_order_expiry_service.go @@ -2,13 +2,25 @@ package service import ( "context" + "database/sql" "log/slog" "sync" "time" + + "github.com/google/uuid" ) const expiryCheckTimeout = 30 * time.Second +const ( + // paymentOrderExpiryLeaderLockKey gates the periodic reconcile + expiry sweep so + // that only one instance issues the upstream payment-provider calls per cycle. + paymentOrderExpiryLeaderLockKey = "payment:order:expiry:leader" + // paymentOrderExpiryLeaderLockTTL must exceed the combined reconcile + expiry + // timeouts (2 * expiryCheckTimeout) so the lock never expires mid-run. + paymentOrderExpiryLeaderLockTTL = 3 * time.Minute +) + // PaymentOrderExpiryService periodically expires timed-out payment orders. type PaymentOrderExpiryService struct { paymentSvc *PaymentService @@ -16,6 +28,10 @@ type PaymentOrderExpiryService struct { stopCh chan struct{} stopOnce sync.Once wg sync.WaitGroup + + lockCache LeaderLockCache + db *sql.DB + instanceID string } func NewPaymentOrderExpiryService(paymentSvc *PaymentService, interval time.Duration) *PaymentOrderExpiryService { @@ -23,9 +39,21 @@ func NewPaymentOrderExpiryService(paymentSvc *PaymentService, interval time.Dura paymentSvc: paymentSvc, interval: interval, stopCh: make(chan struct{}), + instanceID: uuid.NewString(), } } +// SetLeaderLock injects the leader-lock cache and DB used to elect a single +// instance for the periodic reconcile/expiry sweep. When both are nil the job +// runs ungated (single-instance / test behavior). +func (s *PaymentOrderExpiryService) SetLeaderLock(lockCache LeaderLockCache, db *sql.DB) { + if s == nil { + return + } + s.lockCache = lockCache + s.db = db +} + func (s *PaymentOrderExpiryService) Start() { if s == nil || s.paymentSvc == nil || s.interval <= 0 { return @@ -59,6 +87,16 @@ func (s *PaymentOrderExpiryService) Stop() { } func (s *PaymentOrderExpiryService) runOnce() { + // Multi-instance guard: only the leader reconciles/expires orders per cycle, + // avoiding N× upstream payment-provider API calls and update races. + lockCtx, lockCancel := context.WithTimeout(context.Background(), 2*time.Second) + release, ok := tryAcquireSingletonLeaderLock(lockCtx, s.lockCache, s.db, paymentOrderExpiryLeaderLockKey, s.instanceID, paymentOrderExpiryLeaderLockTTL) + lockCancel() + if !ok { + return + } + defer release() + reconcileCtx, cancel := context.WithTimeout(context.Background(), expiryCheckTimeout) recovered, err := s.paymentSvc.ReconcilePendingWxpayOrders(reconcileCtx) cancel() diff --git a/backend/internal/service/subscription_expiry_service.go b/backend/internal/service/subscription_expiry_service.go index a9ec90425a..c93c763f02 100644 --- a/backend/internal/service/subscription_expiry_service.go +++ b/backend/internal/service/subscription_expiry_service.go @@ -2,6 +2,7 @@ package service import ( "context" + "database/sql" "errors" "fmt" "log" @@ -10,6 +11,17 @@ import ( "time" "github.com/Wei-Shaw/sub2api/internal/pkg/pagination" + "github.com/google/uuid" +) + +const ( + // subscriptionExpiryReminderLeaderLockKey gates the per-cycle reminder scan so + // that only one instance walks all active subscriptions and sends reminder + // emails, avoiding redundant full scans and duplicate emails. + subscriptionExpiryReminderLeaderLockKey = "subscription:expiry:reminder:leader" + // subscriptionExpiryReminderLeaderLockTTL bounds crash recovery; the scan can + // page through many subscriptions, so keep it comfortably above one cycle. + subscriptionExpiryReminderLeaderLockTTL = 5 * time.Minute ) // SubscriptionExpiryService periodically updates expired subscription status. @@ -21,6 +33,10 @@ type SubscriptionExpiryService struct { stopCh chan struct{} stopOnce sync.Once wg sync.WaitGroup + + lockCache LeaderLockCache + db *sql.DB + instanceID string } func NewSubscriptionExpiryService(userSubRepo UserSubscriptionRepository, interval time.Duration) *SubscriptionExpiryService { @@ -28,9 +44,21 @@ func NewSubscriptionExpiryService(userSubRepo UserSubscriptionRepository, interv userSubRepo: userSubRepo, interval: interval, stopCh: make(chan struct{}), + instanceID: uuid.NewString(), } } +// SetLeaderLock injects the leader-lock cache and DB used to elect a single +// instance for the periodic expiry-reminder scan. When both are nil the scan runs +// ungated (single-instance / test behavior). +func (s *SubscriptionExpiryService) SetLeaderLock(lockCache LeaderLockCache, db *sql.DB) { + if s == nil { + return + } + s.lockCache = lockCache + s.db = db +} + func (s *SubscriptionExpiryService) SetSettingRepository(settingRepo SettingRepository) { s.settingRepo = settingRepo } @@ -93,6 +121,14 @@ func (s *SubscriptionExpiryService) sendExpiryReminders(ctx context.Context) { if !s.expiryReminderEnabled(ctx) { return } + + // Multi-instance guard: only the leader walks every active subscription and + // sends reminders, avoiding N× full scans and duplicate reminder emails. + release, ok := tryAcquireSingletonLeaderLock(ctx, s.lockCache, s.db, subscriptionExpiryReminderLeaderLockKey, s.instanceID, subscriptionExpiryReminderLeaderLockTTL) + if !ok { + return + } + defer release() for page := 1; ; page++ { subs, pag, err := s.userSubRepo.List(ctx, pagination.PaginationParams{Page: page, PageSize: 200}, nil, nil, SubscriptionStatusActive, "", "expires_at", "asc") if err != nil { diff --git a/backend/internal/service/wire.go b/backend/internal/service/wire.go index fbee1b056c..7eada0b3c2 100644 --- a/backend/internal/service/wire.go +++ b/backend/internal/service/wire.go @@ -143,8 +143,9 @@ func ProvideAntigravityTokenProvider( } // ProvideDashboardAggregationService 创建并启动仪表盘聚合服务 -func ProvideDashboardAggregationService(repo DashboardAggregationRepository, timingWheel *TimingWheelService, cfg *config.Config) *DashboardAggregationService { +func ProvideDashboardAggregationService(repo DashboardAggregationRepository, timingWheel *TimingWheelService, lockCache LeaderLockCache, db *sql.DB, cfg *config.Config) *DashboardAggregationService { svc := NewDashboardAggregationService(repo, timingWheel, cfg) + svc.SetLeaderLock(lockCache, db) svc.Start() return svc } @@ -164,10 +165,11 @@ func ProvideAccountExpiryService(accountRepo AccountRepository) *AccountExpirySe } // ProvideSubscriptionExpiryService creates and starts SubscriptionExpiryService. -func ProvideSubscriptionExpiryService(userSubRepo UserSubscriptionRepository, settingRepo SettingRepository, notificationEmailService *NotificationEmailService) *SubscriptionExpiryService { +func ProvideSubscriptionExpiryService(userSubRepo UserSubscriptionRepository, settingRepo SettingRepository, notificationEmailService *NotificationEmailService, lockCache LeaderLockCache, db *sql.DB) *SubscriptionExpiryService { svc := NewSubscriptionExpiryService(userSubRepo, time.Minute) svc.SetSettingRepository(settingRepo) svc.SetNotificationEmailService(notificationEmailService) + svc.SetLeaderLock(lockCache, db) svc.Start() return svc } @@ -613,8 +615,9 @@ func ProvidePaymentService(entClient *dbent.Client, registry *payment.Registry, } // ProvidePaymentOrderExpiryService creates and starts PaymentOrderExpiryService. -func ProvidePaymentOrderExpiryService(paymentSvc *PaymentService) *PaymentOrderExpiryService { +func ProvidePaymentOrderExpiryService(paymentSvc *PaymentService, lockCache LeaderLockCache, db *sql.DB) *PaymentOrderExpiryService { svc := NewPaymentOrderExpiryService(paymentSvc, 60*time.Second) + svc.SetLeaderLock(lockCache, db) svc.Start() return svc }