From f7280213f70f9978241652f3d98ded822dd9b190 Mon Sep 17 00:00:00 2001 From: Rain <94058511+Pluviobyte@users.noreply.github.com> Date: Sat, 19 Sep 2026 00:51:01 +0800 Subject: [PATCH] fix(proxies): skip expiry writes from stale snapshots --- .../proxy_expiry_renewal_integration_test.go | 84 +++++++++++++++++++ backend/internal/repository/proxy_repo.go | 39 ++++++--- 2 files changed, 110 insertions(+), 13 deletions(-) create mode 100644 backend/internal/repository/proxy_expiry_renewal_integration_test.go diff --git a/backend/internal/repository/proxy_expiry_renewal_integration_test.go b/backend/internal/repository/proxy_expiry_renewal_integration_test.go new file mode 100644 index 0000000000..434c6d4647 --- /dev/null +++ b/backend/internal/repository/proxy_expiry_renewal_integration_test.go @@ -0,0 +1,84 @@ +//go:build integration + +package repository + +import ( + "github.com/Wei-Shaw/sub2api/internal/service" + "time" +) + +// Reproduce the ordering deterministically without sleeps: scan snapshot, +// administrator edit, then the per-proxy transaction consuming that snapshot. +func (s *ProxyExpirySuite) TestSweep_SkipsChangedSnapshot() { + for _, edit := range []string{"renew", "clear expiry", "disable", "change mode", "change backup"} { + s.Run(edit, func() { + now := time.Now() + past := now.Add(-time.Hour) + future := now.Add(24 * time.Hour) + backup := s.mkProxy("snapshot-backup", service.FallbackModeNone, &future, nil) + source := s.mkProxy("snapshot-source", service.FallbackModeDirect, &past, nil) + if edit == "change backup" { + _, err := s.tx.ExecContext(s.ctx, `UPDATE proxies SET fallback_mode='proxy', backup_proxy_id=$1 WHERE id=$2`, backup, source) + s.Require().NoError(err) + } + account := s.mkAccountWithProxy(source) + snapshot, err := s.repo.GetByID(s.ctx, source) + s.Require().NoError(err) + target, change := service.ResolveProxyFallbackTarget(*snapshot, map[int64]service.Proxy{ + backup: {ID: backup, Status: service.StatusActive, ExpiresAt: &future}, + }, now) + s.Require().True(change) + updated := *snapshot + switch edit { + case "renew": + updated.ExpiresAt = &future + case "clear expiry": + updated.ExpiresAt = nil + case "disable": + updated.Status = "inactive" + case "change mode": + updated.FallbackMode = service.FallbackModeNone + case "change backup": + otherBackup := s.mkProxy("replacement-backup", service.FallbackModeNone, &future, nil) + updated.BackupProxyID = &otherBackup + } + s.Require().NoError(s.repo.Update(s.ctx, &updated)) + var eventsBefore int64 + s.Require().NoError(scanSingleRow(s.ctx, s.tx, `SELECT COUNT(*) FROM scheduler_outbox`, nil, &eventsBefore)) + changed, err := s.repo.sweepOneExpiredProxy(s.ctx, *snapshot, now, target, change) + s.Require().NoError(err) + s.Empty(changed) + got, err := s.repo.GetByID(s.ctx, source) + s.Require().NoError(err) + s.Equal(updated.Status, got.Status) + s.Equal(&source, s.accountProxyID(account)) + var origin *int64 + s.Require().NoError(scanSingleRow(s.ctx, s.tx, `SELECT proxy_fallback_origin_id FROM accounts WHERE id=$1`, []any{account}, &origin)) + s.Nil(origin) + var eventsAfter int64 + s.Require().NoError(scanSingleRow(s.ctx, s.tx, `SELECT COUNT(*) FROM scheduler_outbox`, nil, &eventsAfter)) + s.Equal(eventsBefore, eventsAfter, "stale sweep must not publish account changes") + }) + } +} + +func (s *ProxyExpirySuite) TestSweep_ChangedModeIsUsedOnNextScan() { + now := time.Now() + past := now.Add(-time.Hour) + source := s.mkProxy("mode-source", service.FallbackModeDirect, &past, nil) + account := s.mkAccountWithProxy(source) + snapshot, err := s.repo.GetByID(s.ctx, source) + s.Require().NoError(err) + updated := *snapshot + updated.FallbackMode = service.FallbackModeNone + s.Require().NoError(s.repo.Update(s.ctx, &updated)) + changed, err := s.repo.sweepOneExpiredProxy(s.ctx, *snapshot, now, nil, true) + s.Require().NoError(err) + s.Empty(changed) + _, err = s.repo.SweepExpiredProxies(s.ctx, now) + s.Require().NoError(err) + got, err := s.repo.GetByID(s.ctx, source) + s.Require().NoError(err) + s.Equal(service.StatusExpired, got.Status) + s.Equal(&source, s.accountProxyID(account), "new none policy must preserve account binding") +} diff --git a/backend/internal/repository/proxy_repo.go b/backend/internal/repository/proxy_repo.go index 07c92046bd..020acfd673 100644 --- a/backend/internal/repository/proxy_repo.go +++ b/backend/internal/repository/proxy_repo.go @@ -634,7 +634,7 @@ func (r *proxyRepository) ListAllForFallback(ctx context.Context) ([]service.Pro // 全部代理处理完后若有账号被改投，再统一 enqueue 一次 account_bulk_changed 事件——该 enqueue 在子事务之外 // （走 r.sql、失败仅记日志、由调度器周期性 full rebuild 兜底），故「改投 → 失效」整体并非原子。 func (r *proxyRepository) SweepExpiredProxies(ctx context.Context, now time.Time) (int64, error) { - // 快照读（事务前）：允许脏读不影响正确性，事务内已加锁写。 + // 快照用于选择候选；事务内条件更新再次校验有效期、状态及回退配置。 all, err := r.ListAllForFallback(ctx) if err != nil { return 0, err @@ -658,7 +658,7 @@ func (r *proxyRepository) SweepExpiredProxies(ctx context.Context, now time.Time logger.LegacyPrintf("repository.proxy", "[ProxyExpiry] proxy %d expired but fallback chain unresolved (cycle/all-expired); accounts kept", p.ID) } - changedAccountIDs, sweepErr := r.sweepOneExpiredProxy(ctx, p.ID, target, change) + changedAccountIDs, sweepErr := r.sweepOneExpiredProxy(ctx, p, now, target, change) if sweepErr != nil { return totalChanged, sweepErr } @@ -696,7 +696,7 @@ func sortedUniqueAccountIDs(accountIDs []int64) []int64 { // sweepOneExpiredProxy 在单事务内原子执行：标记代理 expired + 改投绑定账号。 // 若 r.client 已绑定事务（测试注入场景），直接在 r.sql 上执行，由外层事务保证原子性。 -func (r *proxyRepository) sweepOneExpiredProxy(ctx context.Context, proxyID int64, target *int64, change bool) ([]int64, error) { +func (r *proxyRepository) sweepOneExpiredProxy(ctx context.Context, snapshot service.Proxy, now time.Time, target *int64, change bool) ([]int64, error) { // 尝试开启子事务；若 r.client 已是事务 client，则返回 ErrTxStarted，退回使用 r.sql。 tx, txErr := r.client.Tx(ctx) if txErr != nil { @@ -704,13 +704,13 @@ func (r *proxyRepository) sweepOneExpiredProxy(ctx context.Context, proxyID int6 return nil, txErr } // 已在外层事务中（集成测试场景），直接用 r.sql 执行 - return r.sweepOneExpiredProxyOnExec(ctx, r.sql, proxyID, target, change) + return r.sweepOneExpiredProxyOnExec(ctx, r.sql, snapshot, now, target, change) } // 使用新事务执行 var accountIDs []int64 var err error - accountIDs, err = r.sweepOneExpiredProxyOnExec(ctx, tx, proxyID, target, change) + accountIDs, err = r.sweepOneExpiredProxyOnExec(ctx, tx, snapshot, now, target, change) if err != nil { _ = tx.Rollback() return nil, err @@ -722,12 +722,28 @@ func (r *proxyRepository) sweepOneExpiredProxy(ctx context.Context, proxyID int6 } // sweepOneExpiredProxyOnExec 在给定的 sqlExecutor 上执行：标记 expired + 改投账号。 -func (r *proxyRepository) sweepOneExpiredProxyOnExec(ctx context.Context, exec sqlExecutor, proxyID int64, target *int64, change bool) ([]int64, error) { - if _, err := exec.ExecContext(ctx, - `UPDATE proxies SET status=$1, updated_at=NOW() WHERE id=$2 AND deleted_at IS NULL`, - service.StatusExpired, proxyID); err != nil { +func (r *proxyRepository) sweepOneExpiredProxyOnExec(ctx context.Context, exec sqlExecutor, snapshot service.Proxy, now time.Time, target *int64, change bool) ([]int64, error) { + proxyID := snapshot.ID + // The UPDATE locks the row and rechecks its predicates after concurrent writes. + // If an administrator renewed, disabled or reconfigured the proxy after the + // snapshot, leave its accounts untouched and reconsider it on the next sweep. + result, err := exec.ExecContext(ctx, ` + UPDATE proxies SET status=$1, updated_at=NOW() + WHERE id=$2 AND deleted_at IS NULL AND status=$3 + AND expires_at <= $4 AND expires_at = $5 + AND fallback_mode=$6 AND backup_proxy_id IS NOT DISTINCT FROM $7`, + service.StatusExpired, proxyID, service.StatusActive, now, snapshot.ExpiresAt, + snapshot.FallbackMode, snapshot.BackupProxyID) + if err != nil { return nil, err } + changed, err := result.RowsAffected() + if err != nil { + return nil, err + } + if changed == 0 { + return nil, nil + } if !change { accountIDs, err := invalidateProxyProbeSnapshots(ctx, exec, proxyID) if err != nil { @@ -738,10 +754,7 @@ func (r *proxyRepository) sweepOneExpiredProxyOnExec(ctx context.Context, exec s } return nil, nil } - var ( - rows *sql.Rows - err error - ) + var rows *sql.Rows // Match the current proxy even after an earlier fallback. Keep the first // origin so manual revert still restores the originally assigned proxy. if target == nil {