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>

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;