feat(quota): 判定与记录回到共享 VM，只有网络等待才分片

## 我引入的自相矛盾分片是为解决这个量出来的：单源探测（无竞争） 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 <noreply@modelrouter.dev>
2026-11-07 09:47:23 +00:00 · 2026-10-05 07:36:40 +08:00
parent 4a0a1f96e4
commit 06f8c5dba4
2 changed files with 268 additions and 4 deletions
--- 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;
--- 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
+}