Skip to content

Commit afe1d11

Browse files
committed
feat(pool): persist cooldowns, honour retryAfterSeconds, show pool state in console
- pool state (cooldown_until + last_error) persisted to <dataRoot>/pool_state.json and restored on startup, so a restart does not immediately retry a known-dead account - LoadState runs after all slots are added (constructing an empty pool and loading matched nothing) - CooldownFor prefers upstream retryAfterSeconds (e.g. 10605 gateway-busy queue) capped at 5m instead of cooling the account for 30m - console account list shows a 可用/冷却中 <remaining> badge with the last error, plus a manual 冷却 button
1 parent 4361bf0 commit afe1d11

4 files changed

Lines changed: 178 additions & 8 deletions

File tree

‎internal/bridge/pool.go‎

Lines changed: 131 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,11 @@
11
package bridge
22

33
import (
4+
"encoding/json"
45
"errors"
56
"net/http"
7+
"os"
8+
"strconv"
69
"strings"
710
"sync"
811
"time"
@@ -72,13 +75,99 @@ func (s *Slot) noteSuccess() {
7275

7376
// Pool 是账号池:轮询选号,跳过冷却中与已试过的账号。
7477
type Pool struct {
75-
mu sync.Mutex
76-
slots []*Slot
77-
cursor int
78+
mu sync.Mutex
79+
slots []*Slot
80+
cursor int
81+
statePath string // 非空时把冷却状态持久化到该文件(重启不丢)
7882
}
7983

8084
func NewPool(slots ...*Slot) *Pool { return &Pool{slots: slots} }
8185

86+
// NewPoolWithState 创建带持久化的账号池:先从 statePath 载入冷却状态,
87+
// 之后每次冷却 / 恢复都会写回,使重启不丢失冷却(避免重启后立刻再撞
88+
// 已知失效的账号)。
89+
func NewPoolWithState(statePath string, slots ...*Slot) *Pool {
90+
p := &Pool{slots: slots, statePath: statePath}
91+
p.load()
92+
return p
93+
}
94+
95+
// poolState 是落盘的池状态(仅保存需要跨重启保留的字段)。
96+
type poolState struct {
97+
Slots map[string]slotState `json:"slots"`
98+
}
99+
100+
type slotState struct {
101+
CooldownUntil time.Time `json:"cooldown_until"`
102+
LastError string `json:"last_error,omitempty"`
103+
}
104+
105+
// LoadState 从磁盘恢复冷却状态。必须在账号全部 Add 之后调用——
106+
// 构造时空池 load 匹配不到任何 ID(这正是先前重启丢冷却的原因)。
107+
func (p *Pool) LoadState() { p.load() }
108+
109+
// load 从磁盘恢复冷却状态(仅恢复尚未过期的冷却)。
110+
func (p *Pool) load() {
111+
if p == nil || p.statePath == "" {
112+
return
113+
}
114+
data, err := os.ReadFile(p.statePath)
115+
if err != nil {
116+
return
117+
}
118+
var st poolState
119+
if json.Unmarshal(data, &st) != nil {
120+
return
121+
}
122+
now := time.Now()
123+
restored := 0
124+
for _, s := range p.snapshot() {
125+
e, ok := st.Slots[s.ID]
126+
if !ok {
127+
continue
128+
}
129+
s.mu.Lock()
130+
if e.CooldownUntil.After(now) {
131+
s.cooldownUntil = e.CooldownUntil
132+
restored++
133+
}
134+
s.lastErr = e.LastError
135+
s.mu.Unlock()
136+
}
137+
if restored > 0 {
138+
logger.Info("pool: restored %d cooling account(s) from %s", restored, p.statePath)
139+
}
140+
}
141+
142+
// save 原子写回池状态(临时文件 + rename)。
143+
func (p *Pool) save() {
144+
if p == nil || p.statePath == "" {
145+
return
146+
}
147+
st := poolState{Slots: map[string]slotState{}}
148+
for _, s := range p.snapshot() {
149+
s.mu.Lock()
150+
until := s.cooldownUntil
151+
lastErr := s.lastErr
152+
s.mu.Unlock()
153+
if !until.IsZero() || lastErr != "" {
154+
st.Slots[s.ID] = slotState{CooldownUntil: until, LastError: lastErr}
155+
}
156+
}
157+
data, err := json.MarshalIndent(st, "", " ")
158+
if err != nil {
159+
return
160+
}
161+
tmp := p.statePath + ".tmp"
162+
if err := os.WriteFile(tmp, data, 0600); err != nil {
163+
logger.Error("pool: save state: %v", err)
164+
return
165+
}
166+
if err := os.Rename(tmp, p.statePath); err != nil {
167+
logger.Error("pool: rename state: %v", err)
168+
}
169+
}
170+
82171
func (p *Pool) Add(s *Slot) {
83172
if p == nil {
84173
return
@@ -143,6 +232,7 @@ func (p *Pool) Cool(id string, d time.Duration, reason string) {
143232
if s.ID == id {
144233
s.cool(d, reason)
145234
logger.Info("pool: account %s cooled for %s (%s)", id, d, reason)
235+
p.save()
146236
return
147237
}
148238
}
@@ -153,6 +243,7 @@ func (p *Pool) NoteSuccess(id string) {
153243
for _, s := range p.snapshot() {
154244
if s.ID == id {
155245
s.noteSuccess()
246+
p.save()
156247
return
157248
}
158249
}
@@ -204,6 +295,11 @@ func isAccountExhaustedDetail(detail string) bool {
204295
func CooldownFor(err error) time.Duration {
205296
var ue *UpstreamError
206297
if errors.As(err, &ue) {
298+
// 上游给了明确重试时间(如 10605 排队 retryAfterSeconds)时优先用它,
299+
// 上限 5 分钟,避免异常大值把账号长期雪藏。
300+
if d, ok := retryAfterFromDetail(ue.Detail); ok {
301+
return d
302+
}
207303
switch {
208304
case ue.Status == http.StatusUnauthorized || ue.Status == http.StatusForbidden:
209305
return CooldownQuota
@@ -216,6 +312,38 @@ func CooldownFor(err error) time.Duration {
216312
return CooldownTransient
217313
}
218314

315+
// retryAfterFromDetail 从上游详情解析 retryAfterSeconds(10605 排队等场景)。
316+
func retryAfterFromDetail(detail string) (time.Duration, bool) {
317+
idx := strings.Index(detail, "retryAfterSeconds")
318+
if idx < 0 {
319+
return 0, false
320+
}
321+
rest := detail[idx+len("retryAfterSeconds"):]
322+
start := -1
323+
for i := 0; i < len(rest); i++ {
324+
if rest[i] >= '0' && rest[i] <= '9' {
325+
start = i
326+
break
327+
}
328+
}
329+
if start < 0 {
330+
return 0, false
331+
}
332+
end := start
333+
for end < len(rest) && rest[end] >= '0' && rest[end] <= '9' {
334+
end++
335+
}
336+
n, err := strconv.Atoi(rest[start:end])
337+
if err != nil || n <= 0 {
338+
return 0, false
339+
}
340+
d := time.Duration(n) * time.Second
341+
if d > 5*time.Minute {
342+
d = 5 * time.Minute
343+
}
344+
return d, true
345+
}
346+
219347
// SlotStatus 是给控制台/运维看的账号池运行时状态。
220348
type SlotStatus struct {
221349
ID string `json:"id"`

‎internal/bridge/pool_test.go‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -215,3 +215,18 @@ func TestNoAccountErrorMapsTo503(t *testing.T) {
215215
t.Errorf("FriendlyError(ErrNoAccount) = (%q, %q)", msg, typ)
216216
}
217217
}
218+
219+
func TestCooldownForRetryAfterSeconds(t *testing.T) {
220+
detail := `{"code":"403","message":"{\"code\":\"10605\",\"retryAfterSeconds\":30}"}`
221+
if got := CooldownFor(NewUpstreamError(403, detail)); got != 30*time.Second {
222+
t.Errorf("retryAfterSeconds=30 -> %v, want 30s", got)
223+
}
224+
if got := CooldownFor(NewUpstreamError(403, `{"code":"112"}`)); got != CooldownQuota {
225+
t.Errorf("plain 403 -> %v, want %v", got, CooldownQuota)
226+
}
227+
// 超大值被截到 5 分钟
228+
big := `{"retryAfterSeconds":99999}`
229+
if got := CooldownFor(NewUpstreamError(403, big)); got != 5*time.Minute {
230+
t.Errorf("huge retryAfter -> %v, want 5m", got)
231+
}
232+
}

‎service.go‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@ import (
55
"encoding/json"
66
"fmt"
77
"net/http"
8+
"path/filepath"
89
"strings"
910
"sync"
1011
"time"
@@ -271,7 +272,7 @@ func (s *Service) startBridgeWithAccount(acct *account.Account) error {
271272
// buildAccountPool 把所有有 secret 的账号建成运行时账号池。
272273
// 单个账号建会话失败(凭证失效/网络等)只跳过该账号,不阻断整池启动。
273274
func (s *Service) buildAccountPool() *bridge.Pool {
274-
pool := bridge.NewPool()
275+
pool := bridge.NewPoolWithState(filepath.Join(account.DataRoot(), "pool_state.json"))
275276
accounts, err := account.List()
276277
if err != nil {
277278
logger.Error("buildAccountPool: list accounts: %v", err)
@@ -294,6 +295,8 @@ func (s *Service) buildAccountPool() *bridge.Pool {
294295
pool.Add(slot)
295296
logger.Info("buildAccountPool: added account %s (region=%s)", a.Name, a.Region)
296297
}
298+
// 账号全部入池后再恢复冷却状态(构造时空池 load 匹配不到 ID)。
299+
pool.LoadState()
297300
return pool
298301
}
299302

‎webconsole/index.html‎

Lines changed: 28 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -437,6 +437,12 @@ <h3>NewAPI / 中转站接入</h3>
437437
</table>`;
438438
}
439439

440+
function fmtDur(ms) {
441+
const sec = Math.max(0, Math.ceil(ms / 1000));
442+
const m = Math.floor(sec / 60), r = sec % 60;
443+
return m > 0 ? m + 'm' + String(r).padStart(2, '0') + 's' : r + 's';
444+
}
445+
440446
function fmtNum(n) {
441447
if (n == null || Number.isNaN(n)) return '-';
442448
return Number(n).toLocaleString(undefined, { maximumFractionDigits: 2 });
@@ -497,19 +503,29 @@ <h3>NewAPI / 中转站接入</h3>
497503

498504

499505

500-
function renderAccounts(list) {
501-
506+
function renderAccounts(list, pool) {
507+
const poolById = {};
508+
((pool && pool.slots) || []).forEach(s => { poolById[s.id] = s; });
502509
if (!list || !list.length) {
503510
$('accountBody').innerHTML = '<tr><td colspan="4" class="muted">暂无账号</td></tr>';
504511
return;
505512
}
506513
$('accountBody').innerHTML = list.map(a => {
507514
const cachedQuota = quotaCache[a.id];
515+
const slot = poolById[a.id];
516+
const coolMs = slot ? (slot.cooling_remaining_ms || 0) : 0;
517+
const poolBadge = __omp_shell("slot ? ''")
518+
: coolMs > 0
519+
? `<span class="pill" style="border-color:#e6b800;background:#fff8e1;color:#8a6d00">冷却中 ${fmtDur(coolMs)}</span>`
520+
: `<span class="pill" style="border-color:#8bc34a;background:#f1f8e9;color:#33691e">可用</span>`;
521+
const poolNote = (slot && coolMs > 0 && slot.last_error)
522+
? `<div class="muted" style="font-size:11px">${slot.last_error}</div>` : '';
508523
return `
509524
<tr data-id="${a.id}">
510525
<td>
511-
<div><strong>${a.name || a.id}</strong> ${a.active ? '<span class="pill">active</span>' : ''}</div>
526+
<div><strong>${a.name || a.id}</strong> ${a.active ? '<span class="pill">active</span>' : ''} ${poolBadge}</div>
512527
<div class="muted">${a.email || '-'}</div>
528+
${poolNote}
513529
</td>
514530
<td>
515531
<div>${a.plan || '-'}</div>
@@ -520,6 +536,7 @@ <h3>NewAPI / 中转站接入</h3>
520536
<button class="secondary" data-act="activate" data-id="${a.id}" ${a.active ? 'disabled' : ''}>激活</button>
521537
<button class="secondary" data-act="quota" data-id="${a.id}">刷新额度</button>
522538
<button class="secondary" data-act="checkin" data-id="${a.id}">🎁 签到</button>
539+
<button class="secondary" data-act="cool" data-id="${a.id}">冷却</button>
523540
<button class="danger" data-act="delete" data-id="${a.id}">删除</button>
524541
</td>
525542
</tr>
@@ -548,8 +565,10 @@ <h3>NewAPI / 中转站接入</h3>
548565
await loadConnection();
549566

550567
try {
568+
let pool = null;
569+
try { pool = await api('/api/pool'); } catch (e) {}
551570
const accounts = await api('/api/accounts');
552-
renderAccounts(accounts);
571+
renderAccounts(accounts, pool);
553572
// 并行拉额度
554573
(accounts || []).forEach(a => loadQuota(a.id));
555574
} catch (e) {
@@ -829,6 +848,11 @@ <h3>NewAPI / 中转站接入</h3>
829848
btn.disabled = false;
830849
btn.textContent = oldText;
831850
}
851+
} else if (act === 'cool') {
852+
const secs = parseInt(prompt('冷却该账号多少秒?(默认 300)', '300'), 10);
853+
if (!secs || secs <= 0) return;
854+
await api('/api/pool/cool', { method: 'POST', body: JSON.stringify({ id, seconds: secs, reason: 'manual' }) });
855+
await refresh();
832856
} else if (act === 'delete') {
833857
if (!confirm('确认删除该账号?')) return;
834858
await api('/api/accounts/delete', { method: 'POST', body: JSON.stringify({ id }) });

0 commit comments

Comments
 (0)