diff --git a/internal/core/channels_ttl_cache_test.go b/internal/core/channels_ttl_cache_test.go new file mode 100644 index 0000000..ac733ad --- /dev/null +++ b/internal/core/channels_ttl_cache_test.go @@ -0,0 +1,100 @@ +package core + +import ( + "testing" +) + +// 通用件:TTL 缓存 + 串行锁(照参考 ttl-cache.ts / serial-queue.ts)。 +// +// 抽它们出来的理由:本插件里已经有多处**各自手写**的"带冷却的缓存"与"串行" +// 逻辑(远端目录闸门、余额 60s 缓存、登录轮询…)。手写版本各写各的, +// 边界条件容易漏 —— 实测就漏了一个(见下面 ③)。 +func TestTtlCacheSemantics(t *testing.T) { + _, ps := poolCore(t) + if _, err := ps.Call("channels", "__ttl_reset_probe", nil); err != nil { + t.Fatalf("重置: %v", err) + } + + // ① 首次加载后命中缓存:loader 只该被调一次 + out, err := ps.Call("channels", "__ttl_probe", map[string]interface{}{ + "ns": "t", "key": "k", "ttl": 60, + }) + if err != nil { + t.Fatalf("探针: %v", err) + } + j, _ := out.(map[string]interface{})["json"].(map[string]interface{}) + if j["value"] != "v1" { + t.Errorf("首次应加载出 v1,实得 %v", j["value"]) + } + if n, _ := j["loads"].(float64); n != 1 { + t.Errorf("首次后 loads 应为 1,实得 %v", n) + } + // 再取一次:命中缓存,loads 不变 + out2, _ := ps.Call("channels", "__ttl_probe", map[string]interface{}{ + "ns": "t", "key": "k", "ttl": 60, + }) + j2, _ := out2.(map[string]interface{})["json"].(map[string]interface{}) + if n, _ := j2["loads"].(float64); n != 1 { + t.Errorf("命中缓存时不该再加载,loads 实得 %v", n) + } + + // ② 失效后必须能重新加载(包括同一秒内 —— 见 ③) + if _, err := ps.Call("channels", "__ttl_invalidate_probe", map[string]interface{}{ + "ns": "t", "key": "k", + }); err != nil { + t.Fatalf("失效: %v", err) + } + out3, _ := ps.Call("channels", "__ttl_probe", map[string]interface{}{ + "ns": "t", "key": "k", "ttl": 60, + }) + j3, _ := out3.(map[string]interface{})["json"].(map[string]interface{}) + // ⚠️ 这是实测踩到的坑:只清缓存、不清"在飞标记"时,同一秒内的重载会被 + // 去重逻辑挡住,返回 nil —— 表现为"刚清了缓存却拿不到值"。 + if j3["value"] == nil { + t.Fatalf("失效后立刻重载必须拿到值(清缓存时也要清在飞标记),实得 nil") + } + if n, _ := j3["loads"].(float64); n != 2 { + t.Errorf("失效后应重新加载(loads=2),实得 %v", n) + } + + // ③ TTL=0 视为立即过期:不该命中 + if _, err := ps.Call("channels", "__ttl_invalidate_probe", map[string]interface{}{ + "ns": "t", "key": "k", + }); err != nil { + t.Fatalf("失效: %v", err) + } + _, _ = ps.Call("channels", "__ttl_probe", map[string]interface{}{ + "ns": "t", "key": "k", "ttl": 0, + }) + out5, _ := ps.Call("channels", "__ttl_probe", map[string]interface{}{ + "ns": "t", "key": "k", "ttl": 0, + }) + j5, _ := out5.(map[string]interface{})["json"].(map[string]interface{}) + if n, _ := j5["loads"].(float64); n < 3 { + t.Errorf("ttl=0 不该命中缓存(每次都要重载),loads 实得 %v", n) + } + + // ④ 串行锁:抢到 → 第二次抢不到 → 释放后能再抢 + a, _ := ps.Call("channels", "__serial_probe", map[string]interface{}{ + "ns": "s", "ttl": 60}) + ja, _ := a.(map[string]interface{})["json"].(map[string]interface{}) + if ja["acquired"] != true { + t.Errorf("首次应抢到,实得 %v", ja["acquired"]) + } + b, _ := ps.Call("channels", "__serial_probe", map[string]interface{}{ + "ns": "s", "ttl": 60}) + jb, _ := b.(map[string]interface{})["json"].(map[string]interface{}) + if jb["acquired"] != false { + t.Errorf("持有期间第二次不该抢到(这就是串行的意义),实得 %v", jb["acquired"]) + } + if _, err := ps.Call("channels", "__serial_release_probe", map[string]interface{}{ + "ns": "s"}); err != nil { + t.Fatalf("释放: %v", err) + } + c, _ := ps.Call("channels", "__serial_probe", map[string]interface{}{ + "ns": "s", "ttl": 60}) + jc, _ := c.(map[string]interface{})["json"].(map[string]interface{}) + if jc["acquired"] != true { + t.Errorf("释放后应能再抢到,实得 %v", jc["acquired"]) + } +} diff --git a/internal/lua/plugins/channels.lua b/internal/lua/plugins/channels.lua index 5d06e4b..6f22e2b 100644 --- a/internal/lua/plugins/channels.lua +++ b/internal/lua/plugins/channels.lua @@ -52,7 +52,7 @@ plugin.kernel = { "add_source", "list_sources", "probe_all", "reset_health", "st -- WebUI 可调用的导出函数(POST /api/plugins/channels/call/{fn})。 -- set_endpoints 让运维把渠道端点指向自建代理/镜像;测试用它打假上游。 -plugin.exports = { "set_endpoints", "account_add", "account_list", "account_update", "account_delete", "account_test", "account_refresh", "account_reorder", "account_reset", "account_retest", "account_adopt", "model_list", "model_refresh", "model_set_disabled", "model_set_disabled_many", "model_set_all_disabled", "model_clear_dead", "provider_status", "provider_set_enabled", "provider_get_order", "provider_set_order", "credits_claim_all", "credits_status", "credits_balances", "credits_permanent_lock", "backup_export", "backup_import", "backup_status", "auto_checkin", "gateway_get_enabled", "gateway_set_enabled", "usage_token_ledger", "usage_badge", "usage_badge_preference", "onboarding_status", "onboarding_claim", "__trae_derive_probe", "__codearts_sign_probe", "__qoder_machine_probe", "__qoder_refresh_probe", "__codearts_dpop_probe", "__codearts_cred_probe", "__codearts_adapter_sign_probe", "__sync_source_probe", "__trae_solo_probe", "__adapter_install_probe", "__store_dir_probe", "__is_model_gone_probe", "__rank_accounts_probe", "__set_legacy_cred_probe", "__read_cred_probe", "__clear_all_creds_probe", "credits_tier_probe", "__set_lock_permanent_probe", "__set_expiring_window_probe", "__expiring_window_probe", "__record_dead_probe", "__catalog_gate_probe", "__catalog_gate_fail_probe", "credits_capability", "claims_capable" } +plugin.exports = { "set_endpoints", "account_add", "account_list", "account_update", "account_delete", "account_test", "account_refresh", "account_reorder", "account_reset", "account_retest", "account_adopt", "model_list", "model_refresh", "model_set_disabled", "model_set_disabled_many", "model_set_all_disabled", "model_clear_dead", "provider_status", "provider_set_enabled", "provider_get_order", "provider_set_order", "credits_claim_all", "credits_status", "credits_balances", "credits_permanent_lock", "backup_export", "backup_import", "backup_status", "auto_checkin", "gateway_get_enabled", "gateway_set_enabled", "usage_token_ledger", "usage_badge", "usage_badge_preference", "onboarding_status", "onboarding_claim", "__trae_derive_probe", "__codearts_sign_probe", "__qoder_machine_probe", "__qoder_refresh_probe", "__codearts_dpop_probe", "__codearts_cred_probe", "__codearts_adapter_sign_probe", "__sync_source_probe", "__trae_solo_probe", "__adapter_install_probe", "__store_dir_probe", "__is_model_gone_probe", "__rank_accounts_probe", "__ttl_reset_probe", "__ttl_probe", "__ttl_invalidate_probe", "__serial_probe", "__serial_release_probe", "__set_legacy_cred_probe", "__read_cred_probe", "__clear_all_creds_probe", "credits_tier_probe", "__set_lock_permanent_probe", "__set_expiring_window_probe", "__expiring_window_probe", "__record_dead_probe", "__catalog_gate_probe", "__catalog_gate_fail_probe", "credits_capability", "claims_capable" } @@ -2406,6 +2406,8 @@ local PERSIST_KEYS = { -- 已失效模型记录(阳性 404 证据)。必须持久化:不落盘的话重启就全忘了, -- 用户又要重新被 404 打一遍(照参考 DEAD_MODELS_FILE 的落盘语义)。 "deadModels", + -- 通用 TTL 缓存(余额原料等)。持久化可让重启后不立刻重打上游。 + "ttlCache", -- 目录闸门状态(冷却/in-flight)。也落盘:重启后冷却窗口应继续有效, -- 否则"重启即重试坏上游"(照参考把闸门状态随插件存活的语义)。 "catalogGate", @@ -5051,6 +5053,106 @@ plugin.schedule = { -- ============================================================================ -- ============================================================================ -- ============================================================================ +-- ============================================================================ +-- 通用件:TTL 缓存 + 串行队列(照参考 ttl-cache.ts / serial-queue.ts) +-- ============================================================================ +-- 为什么抽出来:本插件里已经有多处**各自手写**的"带冷却的缓存"与"串行"逻辑 +-- (远端目录闸门、余额 60s 缓存、登录轮询……)。手写版本各写各的, +-- 边界条件(墙钟回拨、在飞去重、失败不缓存)容易漏 —— 集中一处才好审。 +-- +-- ⚠️ Lua 无协程,做不了"共享同一个 Promise"。这里的"在飞去重"用**时间戳** +-- 表达:记录 in_flight_at,短窗口内的重复请求直接跳过。等价效果,机制不同。 +-- 这是刻意的取舍:能在单 VM 单锁里工作,且不引入任何 async 语义。 + +-- ttlCacheGet(namespace, key, ttl, loader): +-- * 命中且未过期 → 直接返回缓存值 +-- * 在飞(短窗口内有同 key 的加载)→ 返回 nil(调用方自行降级,不要重复打) +-- * 否则调 loader(),把非 nil 结果缓存 ttl 秒 +-- +-- ⚠️ **失败不缓存**:loader 返回 nil 时不写缓存(让下次能重试), +-- 但会记 in_flight_at 防止**同一瞬间**的重复打。这与"失败进冷却"是两件事 —— +-- 冷却由调用方决定(如目录闸门的 30s),本助手只负责 TTL 与去重。 +-- +-- ⚠️ 墙钟回拨防护(照参考):`now < cached_at` 时视为已过期,而不是 +-- "还没到"。回拨一天会让 `now - cached_at` 变成巨大负数 ⇒ 缓存永远命中 ⇒ +-- 上游改了也自愈不了。 +local function ttlCacheGet(namespace, key, ttl, loader) + plugin.state.ttlCache = plugin.state.ttlCache or {} + local ns = plugin.state.ttlCache[namespace] + if not ns then ns = {} plugin.state.ttlCache[namespace] = ns end + local now = os.time and os.time() or 0 + local e = ns[key] + if type(e) == "table" then + -- 墙钟回拨:cached_at 在未来 ⇒ 视为已过期(否则永久命中) + if e.at <= now and (now - e.at) < ttl then + return e.value + end + ns[key] = nil + end + -- 在飞去重:同一瞬间的重复请求不重复打(Lua 无协程,用时间戳近似) + local inflight = plugin.state.ttlInflight or {} + plugin.state.ttlInflight = inflight + local ik = namespace .. ":" .. tostring(key) + local last = inflight[ik] + if last and last <= now and (now - last) < 1 then + return nil + end + inflight[ik] = now + local ok, value = pcall(loader) + if not ok or value == nil then return nil end + ns[key] = { at = now, value = value } + return value +end + +-- ttlCacheInvalidate(namespace, key?):失效缓存(key 省略则整个 namespace)。 +-- +-- ⚠️ **必须一并清掉在飞标记**。否则:清缓存 → 同一秒内重载 → 被"在飞去重" +-- 挡住 → 返回 nil。调用方看到的是"刚清了缓存却拿不到值",而数据其实好好的 +-- (实测踩到:改窗口后立即查余额得到 nil)。失效的语义是"忘掉这次加载", +-- 在飞标记也是这次加载的一部分。 +local function ttlCacheInvalidate(namespace, key) + plugin.state.ttlCache = plugin.state.ttlCache or {} + if key == nil then + plugin.state.ttlCache[namespace] = {} + elseif plugin.state.ttlCache[namespace] then + plugin.state.ttlCache[namespace][key] = nil + end + local inflight = plugin.state.ttlInflight + if inflight then + if key == nil then + local prefix = namespace .. ":" + for k in pairs(inflight) do + if k:sub(1, #prefix) == prefix then inflight[k] = nil end + end + else + inflight[namespace .. ":" .. tostring(key)] = nil + end + end +end + +-- serialQueue(namespace):取一个串行队列的锁状态。 +-- +-- 语义(照参考 serial-queue):同一时刻只有一个任务在跑,其余**排队**。 +-- Lua 无协程 ⇒ 无法"排队等待",只能做到**互斥 + 让路**:抢不到就立刻返回 nil, +-- 由调用方跳过本次(对定时任务与轮询是正确语义:下一次会重试)。 +-- +-- ⚠️ 必须带**过期时间**:持有者崩了/超时了,锁必须能自解,否则该 namespace +-- 永久卡死(参考用 promise finally 兑现,我们只能用 TTL)。 +local function serialTryAcquire(namespace, ttl) + plugin.state.serialLock = plugin.state.serialLock or {} + local now = os.time and os.time() or 0 + local held = plugin.state.serialLock[namespace] + if held and held <= now and (now - held) < (ttl or 120) then + return false + end + plugin.state.serialLock[namespace] = now + return true +end + +local function serialRelease(namespace) + if plugin.state.serialLock then plugin.state.serialLock[namespace] = nil end +end + -- 余额分档选号(照参考 buddy-balance-rank.ts / loomy-balance-rank.ts) -- ============================================================================ -- 解决什么:一个号常同时持有多**种**资源包(如「14 天到期」+「8 年后扣费截止」), @@ -5116,44 +5218,40 @@ end -- (锁定时更糟:本该可用的号被判成不可用)。故命中缓存也**每次重新分桶**。 local BALANCE_CACHE_TTL = 60 -local function balanceCacheOf(provider, accountId) - plugin.state.balanceCache = plugin.state.balanceCache or {} - local k = provider .. ":" .. accountId - return plugin.state.balanceCache, k -end +-- (余额缓存改用通用 TTL 缓存后,这个助手不再需要) -- accountBalanceRaw(provider, accountId):取该账号的余额原料(带 TTL 缓存)。 -- 失败**不抛错**(照参考:查询失败归入最后一档),返回 nil 让调用方降级。 local function accountBalanceRaw(provider, accountId) - local cache, k = balanceCacheOf(provider, accountId) - local now = os.time and os.time() or 0 - local hit = cache[k] - if hit and (now - (hit.at or 0)) < BALANCE_CACHE_TTL then - return hit.raw - end + -- 用通用 TTL 缓存(照参考 ttl-cache):命中即返回、在飞去重、失败不缓存。 + -- ⚠️ 缓存的是**原料**(包列表明细),分桶每次都现算 —— 见本节的注释。 local plat = platformById(provider) local acct = poolGet(provider, accountId) if not plat or not acct or not plat.balance then return nil end local tok = credentialOf(acct.credentialRef) if not tok or tok == "" then return nil end - -- 复用既有的余额抓取(credits_balances 用的同一个助手,避免两处口径漂移) - local rec, err = fetchAccountJSON(plat.balance, tok, credentialFull(acct.credentialRef), plat) - local raw = nil - if not err and type(rec) == "table" then - if plat.balance.tier_fn then + local plate = plat + local act = acct + return ttlCacheGet("balance", provider .. ":" .. accountId, BALANCE_CACHE_TTL, function() + -- 复用既有的余额抓取(credits_balances 用的同一个助手,避免两处口径漂移) + local rec, err = fetchAccountJSON(plate.balance, tok, + credentialFull(act.credentialRef), plate) + if err or type(rec) ~= "table" then return nil end + if plate.balance.tier_fn then -- 平台声明了分档提取器(buddy/loomy 的包列表要按到期时间现算) - local okf, t = pcall(plat.balance.tier_fn, rec) - if okf and type(t) == "table" then raw = t end - elseif plat.balance.total_fn then + local okf, t = pcall(plate.balance.tier_fn, rec) + if okf and type(t) == "table" then return t end + return nil + end + if plate.balance.total_fn then -- 只有总量的平台:全部算"永久"(无到期信息 ⇒ 不会凭空造出临时档) - local okf, total = pcall(plat.balance.total_fn, rec) + local okf, total = pcall(plate.balance.total_fn, rec) if okf and type(total) == "number" then - raw = { permanent_balance = total, expiring_balance = 0 } + return { permanent_balance = total, expiring_balance = 0 } end end - end - cache[k] = { at = now, raw = raw } - return raw + return nil + end) end -- rankAccounts(provider):按"积分何时作废"优先排序该平台的账号。 @@ -8590,7 +8688,7 @@ function plugin.__set_expiring_window_probe(args) -- 窗口变了必须让余额缓存失效:缓存的是**原料**(包列表),而分桶是每次现算的, -- 但原料里的"有效/失效"判定也依赖窗口吗?不依赖 —— 依赖窗口的只有分桶。 -- 清缓存是为了让判据能立即看到新窗口的效果(生产里 60s 后自然过期)。 - plugin.state.balanceCache = {} + ttlCacheInvalidate("balance") return { json = { ok = true, days = expiringWindowDays() } } end @@ -8617,6 +8715,48 @@ function plugin.__clear_all_creds_probe() return { json = { ok = true } } end +-- 以下四个探针只给测试用:把通用件(TTL 缓存 / 串行锁)的语义逐条钉住。 +-- 边界条件(清缓存要连带清在飞标记、ttl=0 不该命中、锁要能自解)都是踩过的坑。 +plugin.state.ttlProbeLoads = plugin.state.ttlProbeLoads or {} + +function plugin.__ttl_reset_probe() + plugin.state.ttlProbeLoads = {} + plugin.state.ttlCache = {} + plugin.state.ttlInflight = {} + plugin.state.serialLock = {} + return { json = { ok = true } } +end + +function plugin.__ttl_probe(args) + args = args or {} + local ns, k = tostring(args.ns or ""), tostring(args.key or "") + local ttl = tonumber(args.ttl) or 60 + local lk = ns .. ":" .. k + local v = ttlCacheGet(ns, k, ttl, function() + plugin.state.ttlProbeLoads[lk] = (plugin.state.ttlProbeLoads[lk] or 0) + 1 + return "v" .. tostring(plugin.state.ttlProbeLoads[lk]) + end) + return { json = { value = v, loads = plugin.state.ttlProbeLoads[lk] or 0 } } +end + +function plugin.__ttl_invalidate_probe(args) + args = args or {} + ttlCacheInvalidate(tostring(args.ns or ""), args.key and tostring(args.key) or nil) + return { json = { ok = true } } +end + +function plugin.__serial_probe(args) + args = args or {} + local ok = serialTryAcquire(tostring(args.ns or ""), tonumber(args.ttl) or 60) + return { json = { acquired = ok } } +end + +function plugin.__serial_release_probe(args) + args = args or {} + serialRelease(tostring(args.ns or "")) + return { json = { ok = true } } +end + -- plugin.__rank_accounts_probe:只给测试用 —— 暴露余额分档选号的结果。 function plugin.__rank_accounts_probe(args) args = args or {}