fix(service): gate periodic background jobs with a leader lock for multi-instance

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 <noreply@anthropic.com>

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)

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()
+}

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")
+}

backend/internal/repository/wire.go

@@ -115,6 +115,7 @@ var ProviderSet = wire.NewSet(
 	NewRedeemCache,
 	NewUpdateCache,
 	NewGeminiTokenCache,
+	NewLeaderLockCache,
 	ProvideSchedulerCache,
 	NewSchedulerOutboxRepository,
 	NewProxyLatencyCache,

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 {

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
+}

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")
+		})
+	}
+}

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()

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 {

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
 }