From b9a025efbab879648929a020e8b7b324b1858177 Mon Sep 17 00:00:00 2001 From: JianFeeeee Date: Mon, 5 Oct 2026 07:36:40 +0800 Subject: [PATCH] =?UTF-8?q?feat(quota):=20=E5=88=A4=E5=AE=9A=E4=B8=8E?= =?UTF-8?q?=E8=AE=B0=E5=BD=95=E5=9B=9E=E5=88=B0=E5=85=B1=E4=BA=AB=20VM?= =?UTF-8?q?=EF=BC=8C=E5=8F=AA=E6=9C=89=E7=BD=91=E7=BB=9C=E7=AD=89=E5=BE=85?= =?UTF-8?q?=E6=89=8D=E5=88=86=E7=89=87?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## 我引入的自相矛盾 分片是为解决这个量出来的： 单源探测（无竞争） 1.4 – 3.3 s 20 源 Promise.all 29 s 20 源串行理论值 ≈ 29 s ← 吻合 根因是 Plugins.Call 持 p.mu 到 Lua 函数返回，**含全部网络等待**；一个 Lua VM 只有一个栈，锁不能去掉。于是每个并发调用用自己的 VM 跑同一份源码（实测 8 分片 2.407s → 0.304s）。 但分片 VM 有自己的 PLUGIN.state —— 而**退避状态就存在 state 里**： fanout（分片） → memo 0 条 → 退避永不生效，每轮都付满 12s call（共享 VM） → memo 17 条 → 13.1s → 4.8s 我设计分片时写了「分片不碰 state」这条硬约束（防 billing 的累加在各分片分叉、 saver 写出随机一份），却没意识到探测的退避状态恰好住在那里。约束和功能打架。 ## 解法：按职责切开，不搬家 把退避表移到网关（B）能解决，但 quota 的 adapter memo 也在 state 里，而且 「插件自包含」是这个插件机制的前提。搬一半比不搬更糟。 所以判定与记录留在插件里，共享 VM 负责；只有网络等待分片： due_sources 共享 VM，纯判定，不发请求（微秒级） fanout 只对 due 里的源分片探测 memo_record 共享 VM，把这轮结果（成功和失败都记）写回 一个分片也不碰 state —— 约束保住了。 ## 实测（隔离实例，18 源，与生产同配置） 第1轮 10163 ms 退避跳过 0 分片 18 答了 4 第2轮 1299 ms 退避跳过 14 分片 4 答了 4 第3轮 1288 ms 退避跳过 14 分片 4 答了 4 第4轮 1316 ms 退避跳过 14 分片 4 答了 4 **29s → 1.3s，且 4 个有余额的源每次都探到** —— 不是靠少显示换来的快。 ## 几个细节 - due_sources 的 body 里不能出现 collectSource：一旦它自己探测，过滤就没有意义 了（成本在分片之前就付掉了）。有判据守着。 - memo_record 忽略 reason 以 "backing off" 开头的行 —— 那是被跳过的探���， 把它记成失败会让 streak 在一次没发生的探测上增长。 - 成功写成 vendor 字符串，streak 自然重置，不需要显式清零。 ## 判据（新增 4 条，共 11 条读实际发布的 quota.lua） - 两个导出存在且在 exports 里 - due_sources 不调 collectSource - memo_record 让 streak 从分片失败里累加，成功后重置 - due_sources 把退避中的 host 挡在分片之外，且零网络请求 写判据时踩了一次：两个源用了同一台 httptest server，于是被判为同一个 host， 两个都退避 —— 这是按 host 记忆的**正确**行为，是我的判据写错了。 Co-Authored-By: ModelRouter --- internal/lua/plugins/quota.lua | 119 +++++++++++++++++++++- internal/lua/quota_backoff_test.go | 153 +++++++++++++++++++++++++++++ 2 files changed, 268 insertions(+), 4 deletions(-) diff --git a/internal/lua/plugins/quota.lua b/internal/lua/plugins/quota.lua index 3a22d09..7148279 100644 --- a/internal/lua/plugins/quota.lua +++ b/internal/lua/plugins/quota.lua @@ -43,7 +43,7 @@ PLUGIN.name = "quota" PLUGIN.version = "1.0.0" PLUGIN.description = "多源套餐余量（余额 / 用量 / 滚动窗口）" -PLUGIN.exports = { "collect_source", "collect_all" } +PLUGIN.exports = { "collect_source", "collect_all", "due_sources", "memo_record" } -- plugin.state must exist before anything indexes it. The kernel restores -- persisted state OVER this table, so declaring it here is safe: a restart with @@ -501,6 +501,92 @@ end -- and doing 21 sources in sequence overran the HTTP call's own budget — the -- whole page returned 502 while individual probes were still in flight. The -- sequential entry point stays for tests and for anyone scripting the plugin. +-- due_sources answers, on the SHARED VM, which sources are worth probing right +-- now — and it makes NO network requests. +-- +-- This exists because of a conflict between two facts: +-- +-- - Probing is network-bound, and Plugins.Call holds the plugin mutex for the +-- whole Lua function, so twenty probes through /call/ serialise. Measured: +-- 29s on a 20-source gateway. The fix is ShardCall, which runs each call on +-- its own VM — and a shard VM has its own PLUGIN.state. +-- - The backoff lives in PLUGIN.state (HOST_MEMO). A shard therefore cannot +-- see it, cannot record it, and a host that fails keeps being probed on every +-- page open. Measured: fanout left the memo at 0 entries, collect_all reached +-- 17. +-- +-- So the split is: the DECISION runs where the state is (one shared-VM call, +-- microseconds), and only the hosts that pass it get sharded. +-- +-- Cheap on purpose. It reads HOST_MEMO and nothing else; if the state has not +-- been restored yet, every host is due, which is the correct answer for a first +-- call. +-- memo_record writes the outcome of a sharded pass into the SHARED VM's memo. +-- +-- A shard runs on its own VM and therefore cannot record anything: measured, a +-- page of fanouts left the memo at 0 entries, so the backoff never accumulated +-- and every refresh re-probed the dead hosts. This is the other half of +-- due_sources — the decision is taken here, the outcome is written back here, and +-- only the network wait happens on the shards. +-- +-- It records FAILURES as well as successes. Recording only successes is what +-- left every source with no balance endpoint re-probing three adapters forever. +PLUGIN.memo_record = function(payload) + if type(payload) ~= "table" then return nil, "expected table" end + if not next(HOST_MEMO) then + local st = PLUGIN.state + if type(st) == "table" and type(st.probe_memo) == "table" then + for k, v in pairs(st.probe_memo) do HOST_MEMO[k] = v end + end + end + local rows = payload.results + if type(rows) ~= "table" then return nil, "missing results" end + local now = os.time() + for _, r in ipairs(rows) do + if type(r) == "table" and type(r.host) == "string" and r.host ~= "" then + local host = hostOf(r.host) + if r.vendor then + -- A source that answered resets its streak: the memo becomes a vendor + -- string, which every reader already treats as a positive memo. + HOST_MEMO[host] = r.vendor + else + local prev = HOST_MEMO[host] + local streak = 1 + if type(prev) == "table" then streak = (prev.streak or 1) + 1 end + HOST_MEMO[host] = { at = now, streak = streak } + end + end + end + PLUGIN.state.probe_memo = HOST_MEMO + return true +end + +PLUGIN.due_sources = function(payload) + if type(payload) ~= "table" then return nil, "expected table" end + if not next(HOST_MEMO) then + local st = PLUGIN.state + if type(st) == "table" and type(st.probe_memo) == "table" then + for k, v in pairs(st.probe_memo) do HOST_MEMO[k] = v end + end + end + local srcs = payload.sources + if type(srcs) ~= "table" then return nil, "missing sources" end + local due, backing = {}, {} + for _, s in ipairs(srcs) do + if type(s) == "table" and type(s.base_url) == "string" and s.base_url ~= "" then + local base = (s.base_url:gsub("/+$", "")) + local known = HOST_MEMO[hostOf(base)] + if type(known) == "table" and backoffSeconds(known.at or 0, known.streak) then + backing[#backing + 1] = { name = s.name, host = hostOf(base), + reason = "backing off after " .. tostring(known.streak) .. " failed probes" } + else + due[#due + 1] = s + end + end + end + return { due = due, backing_off = backing } +end + PLUGIN.collect_all = function(payload) if type(payload) ~= "table" then return nil, "expected table" end if not next(HOST_MEMO) then @@ -830,9 +916,22 @@ PLUGIN.ui = { // overrun. It could not have: the requests were serialised, so the budget they // overran was the very same one. It is a single call now because a // request-per-item fan-out cannot be parallel from the client at all. - out = await pluginAPI.fanout("quota","collect_source",srcs.map(function(s){ - return {sources:[s],name:s.name}; - })); + // Which sources are worth probing is decided on the SHARED VM, where the + // backoff state lives; only those get sharded. + // + // Asking each shard to decide for itself does not work: a shard has its own + // PLUGIN.state, so it cannot see the memo and a host that keeps failing is + // re-probed on every refresh forever. Measured before this split: fanout left + // the memo at 0 entries and every refresh cost the full 12s. + var plan = await pluginAPI.call("quota","due_sources",{sources:srcs}); + var due = (plan && plan.due) || []; + var out = (plan && plan.backing_off) || []; + + if (due.length) { + out = out.concat(await pluginAPI.fanout("quota","collect_source",due.map(function(s){ + return {sources:[s],name:s.name}; + }))); + } // Mirror the host→vendor map the plugin keeps: a source that answered is // worth probing first next time, and the page has everything it needs to // record that without another round trip. @@ -841,6 +940,18 @@ PLUGIN.ui = { if(r && !r.skipped && !r.error && r.host && r.vendor) memo[r.host] = r.vendor; }); window.__qaMemo = memo; + // Hand the whole pass back to the SHARED VM so the backoff has something to + // work from next time. Without this the shards' results die with their VMs: + // measured, the memo stayed at 0 entries and every refresh paid full price. + await pluginAPI.call("quota","memo_record",{results:(out||[]).map(function(r){ + if(!r) return null; + // A host that is merely BACKING OFF has not just failed; recording it as a + // failure would grow the streak on a probe that never happened. + if(r.reason && String(r.reason).indexOf("backing off")===0) return null; + if(r.error) return {host:r.host}; + if(r.skipped) return {host:r.host}; + return {host:r.host, vendor:r.vendor}; + }).filter(Boolean)}); }catch(e){ if(btn) btn.disabled=false; if(sub) sub.textContent="探测失败: "+e.message; diff --git a/internal/lua/quota_backoff_test.go b/internal/lua/quota_backoff_test.go index 3a95ab6..c26d34c 100644 --- a/internal/lua/quota_backoff_test.go +++ b/internal/lua/quota_backoff_test.go @@ -401,3 +401,156 @@ func newCountingServer(t *testing.T) *countingServer { t.Cleanup(cs.Close) return cs } + +// TestQuotaDueSourcesAndMemoRecordCoverTheShardGap pins the split that makes the +// backoff work at all under sharding. +// +// The conflict, measured on a live 20-source gateway: +// - probing is network-bound and Plugins.Call holds the plugin mutex for the +// whole Lua call, so /call/ serialises (29s) +// - so the page uses ShardCall — and a shard has its own PLUGIN.state +// - and the backoff lives in PLUGIN.state: fanout left the memo at 0 entries, +// so dead hosts were re-probed forever +// +// The resolution is to put the DECISION and the RECORDING back on the shared VM +// (microseconds, no network) and shard only the network wait. These two exports +// are that seam, and without them the fanout path silently defeats the backoff. +func TestQuotaDueSourcesAndMemoRecordCoverTheShardGap(t *testing.T) { + src := readShippedPlugin(t, "quota.lua") + for _, want := range []string{"due_sources", "memo_record"} { + if !strings.Contains(src, want) { + t.Errorf("quota.lua does not export %s — a sharded pass cannot see the "+ + "backoff state, nor leave any behind", want) + } + if !strings.Contains(src, `"`+want+`"`) { + t.Errorf("%s is not listed in PLUGIN.exports, so the page cannot call it", want) + } + } + // due_sources must not probe. If it did, the whole point of asking it first + // would be lost: the cost would be paid before any shard runs. + if strings.Contains(src, "due_sources = function(payload)") && + strings.Contains(src, "due_sources") { + // The body is small; check it does not call collectSource. + i := strings.Index(src, "due_sources = function(payload)") + j := strings.Index(src[i:], "\nend\n") + if j > 0 { + body := src[i : i+j] + if strings.Contains(body, "collectSource(") { + t.Error("due_sources calls collectSource — the filter must make no " + + "network request, or it costs the same as probing everything") + } + } + } +} + +// TestQuotaMemoRecordGrowsTheStreakFromShards covers the recording half: a host +// that failed inside a shard must end up in the SHARED memo with a streak, or the +// next refresh has nothing to back off from. +func TestQuotaMemoRecordGrowsTheStreakFromShards(t *testing.T) { + srv := newCountingServer(t) + ps := newQuotaPlugins(t, srv.URL) + host := hostOnly(srv.URL) + + read := func() map[string]interface{} { + st, _ := ps.State("quota").(map[string]interface{}) + memo, _ := st["probe_memo"].(map[string]interface{}) + return memo + } + + // A shard reports a host that answered with nothing. + for i := 1; i <= 3; i++ { + out, err := ps.Call("quota", "memo_record", map[string]interface{}{ + "results": []interface{}{ + map[string]interface{}{"host": srv.URL + "/v1"}, + }, + }) + if err != nil { + t.Fatalf("memo_record %d: %v", i, err) + } + if out != true { + t.Fatalf("memo_record returned %#v", out) + } + } + memo := read() + e, ok := memo[host].(map[string]interface{}) + if !ok { + t.Fatalf("memo has no entry for %s: %#v", host, memo) + } + if got, _ := e["streak"].(float64); got != 3 { + t.Errorf("streak after three shard failures = %.0f, want 3", got) + } + + // A host that answers is recorded as a positive memo, which resets the streak. + if _, err := ps.Call("quota", "memo_record", map[string]interface{}{ + "results": []interface{}{ + map[string]interface{}{"host": srv.URL + "/v1", "vendor": "deepseek"}, + }, + }); err != nil { + t.Fatal(err) + } + if got := read()[host]; got != "deepseek" { + t.Errorf("a host that answered is memoised as %#v, want the vendor name — "+ + "the streak must not survive a success", got) + } +} + +// TestQuotaDueSourcesSkipsBackedOffHosts is the behaviour the page depends on: +// the filter must keep a backed-off host out of the sharded set. +func TestQuotaDueSourcesSkipsBackedOffHosts(t *testing.T) { + srv := newCountingServer(t) + ps := newQuotaPlugins(t, srv.URL) + host := hostOnly(srv.URL) + + // One failure, which is enough to put the host inside its window. + if _, err := ps.Call("quota", "memo_record", map[string]interface{}{ + "results": []interface{}{map[string]interface{}{"host": srv.URL + "/v1"}}, + }); err != nil { + t.Fatal(err) + } + + // A SECOND host, so the two cases are genuinely independent. The memo is keyed + // by host, so two paths on one host are one entry — using the same server here + // would have asserted that a per-host memo behaves per-path. + other := newCountingServer(t) + srcs := []interface{}{ + map[string]interface{}{"name": "dead", "base_url": srv.URL + "/v1", "api_key": "k"}, + map[string]interface{}{"name": "fresh", "base_url": other.URL + "/v1", "api_key": "k"}, + } + out, err := ps.Call("quota", "due_sources", map[string]interface{}{"sources": srcs}) + if err != nil { + t.Fatal(err) + } + m, ok := out.(map[string]interface{}) + if !ok { + t.Fatalf("due_sources returned %#v", out) + } + due, _ := m["due"].([]interface{}) + backing, _ := m["backing_off"].([]interface{}) + if len(backing) != 1 { + t.Errorf("backing_off = %#v, want exactly the host that failed", backing) + } else if b, _ := backing[0].(map[string]interface{}); b["host"] != host { + t.Errorf("backing off %v, want %s — the key must be the host, or it will "+ + "never match on the next refresh", b["host"], host) + } + if len(due) != 1 { + t.Errorf("due = %#v, want only the unprobed host", due) + } else if d, _ := due[0].(map[string]interface{}); d["name"] != "fresh" { + t.Errorf("due contains %v, want fresh", d["name"]) + } + + // And it must make no network request — the whole saving depends on it. + if n := srv.Hits(); n != 0 { + t.Errorf("due_sources made %d requests; it must decide without probing", n) + } +} + +// hostOnly mirrors the plugin's own hostOf. Used only to CHECK what the plugin +// wrote — seeding a memo under a different key produces a memo the plugin never +// looks up, which cost this file a couple of rounds before it was noticed. +func hostOnly(baseURL string) string { + u := strings.TrimPrefix(strings.TrimPrefix(baseURL, "https://"), "http://") + if i := strings.Index(u, "/"); i >= 0 { + u = u[:i] + } + return u +}