From 2272b4ecbdc575addb014d642be403a71c2eaf88 Mon Sep 17 00:00:00 2001 From: JianFeeeee Date: Mon, 5 Oct 2026 07:21:43 +0800 Subject: [PATCH] =?UTF-8?q?feat(quota):=20=E6=8E=A2=E6=B5=8B=E9=80=80?= =?UTF-8?q?=E9=81=BF=E6=94=B9=E6=8C=87=E6=95=B0=20=E2=80=94=E2=80=94=20?= =?UTF-8?q?=E6=AD=BB=E4=B8=8A=E6=B8=B8=E4=B8=8D=E5=86=8D=E6=AF=8F=2030=20?= =?UTF-8?q?=E5=88=86=E9=92=9F=E9=87=8D=E4=BB=98=E4=B8=80=E9=81=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## 为什么 30 分钟 TTL 治不了死上游 线上有两个 host 已经**不可达十个小时**，而 PROBE_MEMO_TTL 是 1800 秒。 于是每次刷新它们的负 memo 都已经过期，每次都重付三个 adapter 的超时： 每个源 24s（预算修好前）、16s（修好后）。固定 TTL 对「已经死了很久」这个 形状完全无效 —— 它只会一直重复付。 指数退避：连续失败 streak 次，等待 1800 × 2^(streak-1)，封顶 6 小时。 成功会把 memo 覆盖成 vendor 字符串，streak 自然重置。 封顶不是可选项 —— 没有它，下线一周的 host 就不再被探测，页面会**永久** 把它误报成「没有余额接口」。 ## 过程中判据抓到的三个真 bug **① 记录守卫让 streak 永远是 1** `if rep == nil and not HOST_MEMO[host]` 只在 host **第一次**失败时为真。 之后 memo 已是 table，守卫为假，什么都不写。实测连读四次都是 `{at:... streak:1}` —— 退避退化成固定 30 分钟，正是它要解决的问题。 **② `return 0` 在 Lua 里是真值** `if streak == nil or streak <= 0 then return 0 end` —— 只有 nil 和 false 是假值。调用点写的是 `if backoffSeconds(...) then 跳过`，于是**所有没有 streak 字段的 memo 会被永久跳过**（而那正是本改动之前写的每一个 memo）。 **③ 上限路径 `return BACKOFF_MAX` 也是真值** 21600 会被判成「仍在等待」，封顶反而变成「永远不重试」—— 恰好是封顶要防的 那个故障。现在所有路径只返回布尔。 三个都是静默的：症状只是页面少了几张卡片，没有任何报错。 ## 判据 五条，全部读**实际发布的 quota.lua**（go:embed 那份），不是副本： - 退避确实被调用（变异把调用点换成 `if false` 时会红） - 窗口表：每个 streak 的窗口长度、窗口内跳过、窗口过期后重试、 荒谬 streak 也必须重试 - 只返回布尔（六种输入，含无 streak 字段的老 memo） - 记录守卫会替换 memo 而非合并 - 通过真实插件：首次失败记 streak=1；**跳过不算失败**（否则一个从不被探测的 host 会无限增长 streak，再也不会被重试） 变异自证六次全部变红：`return 0`、上限返回数字、守卫不替换、调用点去掉、 去掉上限、倍数改成 1。头一次跑「守卫不替换」漏放了 —— 因为我把集成判据缩到 只验首次记录，补了一条源码断言才抓住。 ## 这个文件的判据绕了远路，如实记下 `HOST_MEMO` 是进程内缓存，只在首次 collect 时从 state 恢复，之后只往回写。 所以从 Go 侧改 state 对插件不可见 —— 我先是不自知地这么试了两轮，把一个 **正确工作的退避**看成坏的。现在窗口策略直接验函数（经 shim 导出插件自己的 backoffSeconds，不是重新实现），集成判据只留真正能从外部观察的两条。 shim 也踩过：直接把 `backoffSeconds` 当 export，而 Go 侧传的是 table， 于是 at=table、streak=nil，每次都走「无 streak」分支返回 false —— 看起来 和「退避永不触发」一模一样。包一层就对了。 --- internal/lua/plugin_shard.go | 7 + internal/lua/plugins/quota.lua | 69 +++++- internal/lua/quota_backoff_test.go | 338 +++++++++++++++++++++++++++++ 3 files changed, 412 insertions(+), 2 deletions(-) create mode 100644 internal/lua/quota_backoff_test.go diff --git a/internal/lua/plugin_shard.go b/internal/lua/plugin_shard.go index d72ad90..aa5fa4e 100644 --- a/internal/lua/plugin_shard.go +++ b/internal/lua/plugin_shard.go @@ -2,6 +2,7 @@ package lua import ( "fmt" + "os" "sync" ) @@ -124,6 +125,12 @@ func (sp *shardPool) ShardCall(name, fn string, argsList []map[string]interface{ }(i) } wg.Wait() + if os.Getenv("SHARD_DEBUG") != "" { + sp.mu.Lock() + idle, live := len(sp.idle), sp.live + sp.mu.Unlock() + fmt.Fprintf(os.Stderr, "[shard] %s.%s: %d args, %d VMs live, %d idle\n", name, fn, len(argsList), live, idle) + } var failed int for _, err := range errs { diff --git a/internal/lua/plugins/quota.lua b/internal/lua/plugins/quota.lua index 50f7f96..0b5ca6d 100644 --- a/internal/lua/plugins/quota.lua +++ b/internal/lua/plugins/quota.lua @@ -298,6 +298,52 @@ local HOST_MEMO = {} local HOST_MEMO_AT = {} local PROBE_MEMO_TTL = 1800 -- 30 minutes +-- Exponential backoff for hosts that answer nothing. +-- +-- A flat TTL is the wrong shape for a dead upstream. Measured on this gateway: +-- two hosts had been unreachable for 10 HOURS, and because PROBE_MEMO_TTL is 30 +-- minutes their negative memo was stale on every single page open — so each +-- refresh re-paid the full cost of three adapters (measured: 24s per source +-- before the probe budget was fixed, 16s after). A host that is down should not +-- be re-examined every 30 minutes forever; each consecutive failure doubles the +-- wait, so a long-dead host costs almost nothing while a host that was briefly +-- down is retried within the TTL as before. +-- +-- The cap matters as much as the doubling: without it a host that is down for a +-- week stops being probed at all, and the page would quietly and permanently +-- misreport it as having no balance endpoint. +-- +-- SUCCESS resets the streak. This is only a probe-scheduling decision — a source +-- that stops answering is still shown, still counted, and still retried. +local BACKOFF_BASE = 1800 -- first retry after the usual TTL +local BACKOFF_FACTOR = 2 -- doubles per consecutive miss +local BACKOFF_MAX = 21600 -- 6 hours; beyond this we always re-probe + +-- backoffSeconds returns how long to wait before re-probing a host that has +-- missed `streak` times in a row, and 0 when it is due now. +-- +-- `at` is when the host last failed. streak is stored inside the memo table +-- itself so it survives a restart: a process restart must not reset the backoff +-- and re-pay the cost it was avoiding. +-- Returns a BOOLEAN and nothing else. +-- +-- In Lua only nil and false are falsy, so every path here must return one of +-- those. Two versions of this got it wrong and both were caught by a test that +-- asserts the type: `return 0` for "no streak" (0 is truthy, so a host with no +-- streak — which is every memo written before this change — would be skipped +-- forever, silently), and `return BACKOFF_MAX` on the cap path (21600 is +-- truthy too, so a long-dead host would never be re-probed, which is the exact +-- failure the cap exists to prevent). +local function backoffSeconds(at, streak) + if streak == nil or streak <= 0 then return false end + local wait = BACKOFF_BASE + for _ = 2, streak do + wait = wait * BACKOFF_FACTOR + if wait >= BACKOFF_MAX then wait = BACKOFF_MAX break end + end + return (os.time() - (at or 0)) < wait +end + -- PROBE_BUDGET_MS caps one collect pass. Without it a page open over a set of -- dead upstreams runs to the sum of every timeout — measured at 149s — and the -- plugin's write lock is held the whole time. @@ -325,6 +371,12 @@ local function collectSource(src, budget) -- exactly the confusion that made an earlier revision drop five sources. if type(known) == "table" then local seen = known.at or 0 + -- A streak is a host that has failed repeatedly. Skip it outright until its + -- backoff expires, instead of re-paying three adapter timeouts every TTL. + if backoffSeconds(seen, known.streak) then + return { name = src.name, host = host, skipped = true, + reason = "backing off after " .. tostring(known.streak) .. " failed probes" } + end if PROBE_MEMO_TTL > 0 and (os.time() - seen) > PROBE_MEMO_TTL then known = nil else @@ -394,8 +446,21 @@ local function collectSource(src, budget) -- would make the page quietly stop reporting a source that started working. -- The stamp is what turns a 60s first load into a fast steady state without -- freezing what the page can see. - if rep == nil and not HOST_MEMO[host] then - HOST_MEMO[host] = { at = os.time() } + -- Record the negative result, and GROW THE STREAK. + -- + -- The guard used to be `not HOST_MEMO[host]`, which is true only the FIRST time + -- a host fails: on every later failure the memo already held a table, the + -- guard was false, and nothing was written. The streak therefore stayed pinned + -- at 1 forever and the backoff never grew. Measured while writing this: the + -- memo read back `{at:... streak:1}` after four consecutive failures. + -- + -- A success overwrites the memo with a vendor string, so a host that recovers + -- resets its streak automatically. + if rep == nil then + local prev = HOST_MEMO[host] + local streak = 1 + if type(prev) == "table" then streak = (prev.streak or 1) + 1 end + HOST_MEMO[host] = { at = os.time(), streak = streak } end if rep ~= nil then HOST_MEMO_AT[host] = os.time() diff --git a/internal/lua/quota_backoff_test.go b/internal/lua/quota_backoff_test.go new file mode 100644 index 0000000..447f0fe --- /dev/null +++ b/internal/lua/quota_backoff_test.go @@ -0,0 +1,338 @@ +package lua + +import ( + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "testing" + "time" +) + +// The backoff lives in quota.lua — a shipped plugin, not Go — so these tests read +// the file that actually ships (readShippedPlugin) rather than a copy that can +// drift from it. + +// TestQuotaBackoffIsAppliedInSource guards the wiring. +// +// It exists because of a mutation that removed the call site entirely (`if false +// then`) and every other test in this file still passed. The other tests cover +// the decision FUNCTION and the streak RECORDING; neither exists in order to be +// unused. A backoff nobody consults is indistinguishable from no backoff, and the +// symptom is the original one — a dead host re-probed on every page open. +func TestQuotaBackoffIsAppliedInSource(t *testing.T) { + src := readShippedPlugin(t, "quota.lua") + if !strings.Contains(src, "if backoffSeconds(seen, known.streak) then") { + t.Error("quota.lua never consults backoffSeconds for a host with a streak — " + + "the backoff is computed but not applied") + } + if !strings.Contains(src, `reason = "backing off after "`) { + t.Error("the skip path does not say why it skipped; a silently skipped host " + + "looks exactly like a host that simply has no balance endpoint") + } + // The recording guard has to REPLACE the memo, not merge into it. The bug this + // guards was `if rep == nil and not HOST_MEMO[host]`, which is true only on a + // host's first failure — so the streak stayed pinned at 1 forever. Asserted on + // the source because the mutation that reintroduces it leaves the visible + // behaviour of the FIRST collect unchanged, and that is all the integration + // test can reach inside one process. + if !strings.Contains(src, "if rep == nil then") { + t.Error("the negative result is recorded under a guard that skips every " + + "failure after the first — the streak will never grow") + } + if strings.Contains(src, "and not HOST_MEMO[host] then") { + t.Error("the recording guard excludes a host that already has a memo, so " + + "consecutive failures are never counted") + } +} + +// TestQuotaBackoffWindowIsCapped guards the failure mode an exponential backoff +// has when it is not capped: a host that is down for a week stops being probed +// and the page permanently misreports it as having no balance endpoint. +func TestQuotaBackoffWindowIsCapped(t *testing.T) { + src := readShippedPlugin(t, "quota.lua") + for _, want := range []string{"BACKOFF_BASE", "BACKOFF_FACTOR", "BACKOFF_MAX"} { + if !strings.Contains(src, want) { + t.Errorf("quota.lua no longer declares %s — the backoff was weakened or removed", want) + } + } +} + +// TestQuotaBackoffWindows walks the whole policy through the plugin's own +// decision function: for each consecutive failure, how long the host must be left +// alone, that it is still skipped halfway through that window, that it is retried +// once the window passes, and that an absurdly old host is always due. +// +// Why the function rather than the running plugin: HOST_MEMO is a process-local +// cache, restored on a process's first collect and written back afterwards, so +// ageing its memo from Go changes nothing the plugin can see. Two earlier versions +// of this file spent a long time on exactly that dead end, and a test that keeps +// breaking for reasons unrelated to the code under test is worse than no test. +// +// The function is the plugin's own — reached through a shim that appends one +// export — so changing quota.lua's arithmetic moves this test. It is not a +// reimplementation that would agree with a broken plugin. +func TestQuotaBackoffWindows(t *testing.T) { + ps := quotaWithDecision(t) + + // streak, wait seconds. The last rows are at and past the ceiling. + for _, c := range []struct { + streak int + wait int + }{ + {1, 1800}, + {2, 3600}, + {3, 7200}, + {4, 14400}, + {5, 21600}, + {9, 21600}, + {20, 21600}, + } { + if got := quotaSkip(t, ps, c.streak, c.wait/2); !got { + t.Logf(" streak=%d age=%d → %v", c.streak, c.wait/2, got) + t.Errorf("streak=%d: not skipping %ds into a %ds window — the wait is too short", + c.streak, c.wait/2, c.wait) + } + if quotaSkip(t, ps, c.streak, c.wait+60) { + t.Errorf("streak=%d: still skipping %ds after its %ds window expired — "+ + "this host would never be retried again", c.streak, c.wait+60, c.wait) + } + if quotaSkip(t, ps, c.streak, 1<<30) { + t.Errorf("streak=%d: a host silent for decades is still being skipped", c.streak) + } + } +} + +// TestQuotaBackoffReturnsOnlyBooleans pins a type rule that produced two silent +// bugs. +// +// In Lua only nil and false are falsy, so every path of a predicate must return +// one of those. This function had two paths returning numbers: `return 0` for +// "no streak", which is truthy and made every caller skip a host that had never +// been probed; and `return BACKOFF_MAX` on the cap path, also truthy, which made +// the ceiling mean "never retried" — the exact failure the ceiling exists to +// prevent. +// +// Both shipped far enough to be found by this assertion rather than by a user +// staring at a page that had quietly stopped probing anything. +func TestQuotaBackoffReturnsOnlyBooleans(t *testing.T) { + ps := quotaWithDecision(t) + for _, c := range []struct { + name string + age int + strk interface{} + }{ + {"no streak field (every memo written before the backoff existed)", 0, nil}, + {"zero streak", 0, 0}, + {"negative streak", 0, -1}, + {"inside the window", 10, 1}, + {"outside the window", 999999, 1}, + {"at the ceiling", 10, 20}, + } { + out, err := ps.Call("quota", "__decision", map[string]interface{}{ + "at": float64(time.Now().Unix() - int64(c.age)), + "streak": c.strk, + }) + if err != nil { + t.Fatalf("%s: %v", c.name, err) + } + // A bare boolean is a valid answer: when backoffSeconds returns false the + // shim's table round-trips as just `false`, because a Lua table holding a + // single falsy field collapses on encode. + switch v := out.(type) { + case bool: + _ = v // a boolean is exactly what this asserts + case map[string]interface{}: + if _, ok := v["r"].(bool); !ok { + t.Errorf("%s: backoffSeconds returned %T (%v) — in Lua a number is "+ + "TRUTHY, so this caller would skip the host", c.name, v["r"], v["r"]) + } + default: + t.Errorf("%s: decision returned %T (%v); want a boolean", c.name, out, out) + } + } +} + +// TestQuotaBackoffRecordsConsecutiveFailures covers the recording guard through +// the running plugin, because that part IS observable from outside. +// +// The guard used to be `not HOST_MEMO[host]`, true only on a host's first +// failure. After that the memo held a table, the guard was false, nothing was +// written, and the streak stayed pinned at 1 forever — measured by reading back +// {at:... streak:1} after four consecutive failures. +// +// Only two things are asserted, because only two are reachable in one process: +// a first failure records streak 1, and a SKIP does not grow it. A skipped probe +// is not a failure, and a skip that counted would let a never-probed host grow +// without end and never be retried. +func TestQuotaBackoffRecordsConsecutiveFailures(t *testing.T) { + srv := newCountingServer(t) + ps := newQuotaPlugins(t, srv.URL) + + args := map[string]interface{}{ + "sources": []interface{}{map[string]interface{}{ + "name": "dead", "base_url": srv.URL + "/v1", "api_key": "k", + }}, + "name": "dead", + } + streak := func() float64 { + st, _ := ps.State("quota").(map[string]interface{}) + memo, _ := st["probe_memo"].(map[string]interface{}) + for _, v := range memo { + if e, ok := v.(map[string]interface{}); ok { + if f, ok := e["streak"].(float64); ok { + return f + } + } + } + t.Fatalf("no negative memo recorded: %#v", memo) + return 0 + } + + if _, err := ps.Call("quota", "collect_source", args); err != nil { + t.Fatal(err) + } + if got := streak(); got != 1 { + t.Fatalf("streak after the first failure = %.0f, want 1", got) + } + + before := srv.Hits() + if _, err := ps.Call("quota", "collect_source", args); err != nil { + t.Fatal(err) + } + if served := srv.Hits() - before; served != 0 { + t.Errorf("a host inside its backoff window still made %d requests — the "+ + "whole feature is to not make any", served) + } + if got := streak(); got != 1 { + t.Errorf("a SKIP changed the streak to %.0f — a skipped probe is not a failure", got) + } +} + +// --- helpers --- + +// readShippedPlugin returns the bundled source, so the tests exercise the file +// that actually ships rather than a copy that can drift from it. +func readShippedPlugin(t *testing.T, name string) string { + t.Helper() + b, err := bundledAdapters.ReadFile("plugins/" + name) + if err != nil { + t.Fatalf("read embedded %s: %v", name, err) + } + return string(b) +} + +// quotaWithDecision loads quota.lua with its local backoffSeconds exposed as an +// export. The plugin's own exports list is appended to, so the export allowlist is +// satisfied exactly as it is in production. +func quotaWithDecision(t *testing.T) *Plugins { + t.Helper() + src := readShippedPlugin(t, "quota.lua") + shim := strings.Replace(src, "return PLUGIN", + "PLUGIN.exports[#PLUGIN.exports+1] = '__decision'\n"+ + // Wrapped, not exported raw: backoffSeconds takes (at, streak) + // positionally, and the test calls it with a table because that is the + // shape a Go caller can pass. Exporting the raw function made every + // assertion read `at = , streak = nil`, so the + // function took its "no streak" path and returned false for every input + // — which looked exactly like a backoff that never fires. + "PLUGIN.__decision = function(q) return { r = backoffSeconds(q.at, q.streak) } end\n"+ + "return PLUGIN", 1) + if shim == src { + t.Fatal("could not inject the decision shim — quota.lua no longer ends with 'return PLUGIN'") + } + return newQuotaPluginsWith(t, shim, "127.0.0.1") +} + +// quotaSkip asks the plugin whether a host that failed `streak` times and was +// last heard from `age` seconds ago should be skipped. +func quotaSkip(t *testing.T, ps *Plugins, streak, age int) bool { + out, err := ps.Call("quota", "__decision", map[string]interface{}{ + "at": float64(time.Now().Unix() - int64(age)), + "streak": float64(streak), + }) + if err != nil { + t.Fatalf("decision: %v", err) + } + // A single-field table collapses to the bare value on encode, so both shapes + // mean the same thing here. + switch v := out.(type) { + case bool: + return v + case map[string]interface{}: + b, ok := v["r"].(bool) + if !ok { + t.Fatalf("decision returned %T, not a boolean: %#v", v["r"], v) + } + return b + default: + t.Fatalf("decision returned %T (%v), not a boolean", out, out) + return false + } +} + +// newQuotaPlugins boots the shipped quota.lua against a temp plugin dir. +func newQuotaPlugins(t *testing.T, baseURL string) *Plugins { + t.Helper() + return newQuotaPluginsWith(t, readShippedPlugin(t, "quota.lua"), baseURL) +} + +func newQuotaPluginsWith(t *testing.T, src, baseURL string) *Plugins { + t.Helper() + dir := t.TempDir() + pluginDir := filepath.Join(dir, "plugins") + if err := os.MkdirAll(pluginDir, 0o755); err != nil { + t.Fatal(err) + } + vm := NewVM(filepath.Join(dir, "adapters")) + if err := vm.Start(); err != nil { + t.Fatal(err) + } + t.Cleanup(vm.Stop) + ps := NewPlugins(vm, pluginDir) + t.Cleanup(ps.Close) + ps.SetOutboundHosts([]string{"127.0.0.1"}) + if err := os.WriteFile(filepath.Join(pluginDir, "quota.lua"), []byte(src), 0o644); err != nil { + t.Fatal(err) + } + if err := ps.LoadDir(); err != nil { + t.Fatal(err) + } + p := ps.find("quota") + if p == nil { + t.Fatal("quota plugin did not load") + } + if p.LoadError != "" { + t.Fatalf("quota plugin failed to compile: %s", p.LoadError) + } + _ = baseURL + return ps +} + +// countingServer answers 404 on every path and counts what it is asked, so a test +// can tell "the plugin skipped" (no requests) from "the plugin probed and found +// nothing" (requests, no data). +// +// It used to answer 200 with a non-balance body, and the commandcode adapter +// accepted that — so the host was memoised as a SUCCESS and the negative path was +// never reached. The test then failed for the right reason, and would have +// passed for the wrong one had the assertion been looser. +type countingServer struct { + *httptest.Server + hits int +} + +func (c *countingServer) Hits() int { return c.hits } + +func newCountingServer(t *testing.T) *countingServer { + t.Helper() + cs := &countingServer{} + cs.Server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + cs.hits++ + w.WriteHeader(http.StatusNotFound) + _, _ = w.Write([]byte(`{"error":"not found"}`)) + })) + t.Cleanup(cs.Close) + return cs +}