seewe/sub2api
1
0
Fork 0
mirror of https://github.com/Wei-Shaw/sub2api.git synced 2026-10-14 03:52:26 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(proxies): skip expiry writes from stale snapshots

Browse Source
This commit is contained in:
Rain
2026-09-19 00:51:01 +08:00
parent efe9aab1e4
commit f7280213f7
2 changed files with 110 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/internal/repository/proxy_expiry_renewal_integration_test.go
+84
View File
@@ -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")
}
backend/internal/repository/proxy_repo.go
+26 -13
View File
@@ -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 {