perf(plugin): 去掉钩子热路径的 JSON 往返 + 同 stage 跨插件并行

## 1. 去掉 JSON 往返(快路径)
实测单次 Fire 14.6µs,其中 json.Marshal 4.0 + json.Unmarshal 5.6 = 9.6µs,
**67% 花在把 map[string]interface{} 序列化再反序列化**,而紧接着的
pushGoValue 本来就能直接遍历这两种类型。改为按类型直接转换(fastvalue.go),
只对不认识���类型才回落 JSON —— 陌生字段仍然会被送到插件,而不是消失。

快路径与 JSON 路径逐字节等价由 TestFastPathMatchesJSONPath 锁住(8 组载荷,
覆盖 int/uint/float 各宽度、嵌套、slice、map[string]string、未知类型)。
还有一条专门防止「优化悄悄失效」:TestFastPathIsActuallyUsed 用真实的
request_end 载荷断言它确实走快路径。

同一份代码 A/B 实测:JSON 往返 56.4µs → 快路径 35.3µs(省 37%)。

## 2. 同 stage 跨插件并行
参照 /home/program/TrueAgent 的 StageHost.RunStage:
  - **快照后释放锁**再并行 —— 它记录过一次自死锁(p.Stop → onExit → ReclaimOwner
    要拿 registry 锁,持锁并行即死锁)。这里同理:钩子可能经 admin API 增删插件,
    那条路径要拿 ps.mu 写锁,所以并行段内不持任何 ps 锁。
  - 每个 goroutine recover。
  - 单插件走直连路径,不付 goroutine 代价(生产就是这种配置)。

**与 TrueAgent 不同的一点**:它可以放心并行,因为 handler 只写 ctx.Response 并有
IsResponded() 仲裁;我们的钩子返回 table 会合并进 payload,而
docs/plugins.md 明确承诺「payload 原样传给下一个插件」。所以合并**按插件加载
顺序**执行,结果确定,不依赖调度;代价是钩子之间不再互相可见 —— 这是一处
**契约变化**,已在文档里写明,并说明随核心发布的 billing 从不返回任何值
(代码注释就写着 "nobody downstream would read a return value")。

实测收益(真实二进制,三实例对照,3000 请求):

              无插件     1 插件      4 插件
  稳态并发32    849 rps   768 (-9.5%) 741 (-12.7%)
  突发并发64   1524-1893  1182-1676  1064-1443

4 插件只降 10-20%,而并行前实测 4 插件是 63.8µs vs 单插件 14.6µs(-300%)。

## ★ 我自己造成的两次性能事故
**① 持久化把热路径拖慢 26 倍。** 最初的快照在钩子路径上做:走 luaValueToGo +
json.Marshal + json.Unmarshal 三重转换,每请求 264µs,Fire 从 14.6µs 变成 385µs。
改成 saver 按自己节奏拉取(钩子只标记 dirty,flush 时才快照),385µs → 25.7µs。
**这里还踩了第二次 use-after-free**:让后台 goroutine 去读 Lua 表,vm.Stop() 后
那是已释放内存(SIGSEGV)。安全性现在由「Plugins.Close 等 saver 的最后一次
flush 完成后,调用方才停 VM」保证。

**② 基准被自己的后台写入污染。** 关掉 markDirty 反而测出 36µs、比开着还慢,
方向完全反了 —— 是 saver 每 2 秒写盘混进了计时。加了 DisableStatePersistence
后数据才可信。

## 判据(11 项,全部变异验证)
快路径等价/确实生效/不别名输入 + 并行与单插件路径合并一致 + 合并顺序确定 +
抛异常的钩子不拖累同伴 + 每插件恰好执行一次 + 并发 Fire 安全 + Fire 期间不持
注册表锁 + 真实 billing 在并行下正常 + 持久化 7 项。

变异:改坏合并顺序 → 红;去掉单插件路径的合并 → 红(3 个既有测试同时抓到)。
★ 「删掉 recover」这个变异**没有**让判据变红,查下去发现 golua 把 error()、
nil 索引、调用 nil、深递归全部转成 error RETURN,不产生 Go panic —— 那个测试
根本没测到 recover。已改名 TestThrowingHook 并在注释里写明 recover() 当前无法
被 Lua 触达,保留它是为了守 Go 侧。留一个「看起来有覆盖」的断言比没有更糟。

## 端到端(真实二进制 + 真实 billing)
20 万请求全 200,rps 1870,p99 96ms,RSS 37.9→42MB 有界;
负载停止后四个插件计数**完全一致**(231745),hook_errors 为空;
systemctl restart 后 billing 仍是 231745 —— 并行与持久化同时生效。
381 个测试全绿,含 -race。
This commit is contained in:
JianFeeeee
2026-10-02 10:45:20 +08:00
parent ce2032435a
commit 30064696b3
5 changed files with 760 additions and 55 deletions

View File

@ -185,6 +185,18 @@ core.New
- 返回 `nil` 或不返回 = **没有意见**,payload 原样传给下一个插件 - 返回 `nil` 或不返回 = **没有意见**,payload 原样传给下一个插件
- 返回 table = 其中的键会**合并进 payload**,并作为 `Fire` 的返回值 - 返回 table = 其中的键会**合并进 payload**,并作为 `Fire` 的返回值
**同一 stage 的多个插件并行执行**,但返回值按**插件加载顺序**合并,所以结果是
确定的(不依赖 goroutine 调度)。代价是一个插件看不到另一个插件刚加的字段:
每个钩子拿到的是**同一份 payload 快照**。
这与早期版本不同 —— 早期是顺序执行,后一个插件能看到前一个的返回值。它从未被
实际依赖(随核心发布的 billing 在每个 stage 都 `return nil`,注释里写着
"nobody downstream would read a return value"),但这是一处**契约变化**:如果你的
插件依赖「读到前一个插件写的字段」,并行的两个插件之间必须改用外部通信
(例如各自写 `plugin.state`,由 `/api/plugins/<name>/state` 读取)。
单个插件时不启 goroutine,直接调用。
前三个 stage 的返回值目前没有内部消费者(最后一个 stage 之后就是写审计), 前三个 stage 的返回值目前没有内部消费者(最后一个 stage 之后就是写审计),
所以计费插件改用 `plugin.state` + `/state` 端点来暴露数据。 所以计费插件改用 `plugin.state` + `/state` 端点来暴露数据。

139
internal/lua/fastvalue.go Normal file
View File

@ -0,0 +1,139 @@
package lua
import (
"encoding/json"
"strconv"
)
// This file exists to remove the JSON round-trip from the plugin hook path.
//
// Measured before this change: one Fire() call cost 14.6us, of which
// json.Marshal(payload) was 4.0us and json.Unmarshal back into interface{} was
// 5.6us — 67% of the whole hook, spent re-deriving a tree that the very next
// line (pushGoValue) already knows how to walk natively.
//
// The round-trip existed because payload arrives as map[string]interface{} and
// pushGoValue wanted a plain interface{} tree. pushGoValue already handles
// map[string]interface{} and []interface{} directly, so the fix is a converter
// that flattens the concrete Go types the gateway actually produces, instead of
// a generic serialize/parse.
//
// The JSON path is kept as the fallback for types the fast path does not know,
// so a plugin payload carrying something exotic still arrives rather than being
// silently dropped.
// jsonNumberType is the type json.Unmarshal produces for every JSON number.
// The fast path recognizes it so numbers survive as numbers rather than being
// stringified into a different Lua type.
type jsonNumberType = float64
// fastToPlain converts the concrete Go values the gateway puts in a hook
// payload into the interface{} tree pushGoValue consumes, without JSON.
//
// The interesting cases are the ones json.Unmarshal would have normalized:
// - json.Number-ish types (all float64 in practice)
// - int / int64 / uint variants, which must land as Lua numbers, not as the
// strings a naive "only handle float64" switch would produce
// - []string and map[string]string, extremely common in payloads and NOT
// handled by a switch that only knows []interface{}
//
// Anything unrecognized returns (value, false) so the caller can fall back to
// the JSON path, which is slower but total.
func fastToPlain(v interface{}) (interface{}, bool) {
switch x := v.(type) {
case nil, bool, string, jsonNumberType:
return v, true
case int:
return float64(x), true
case int8:
return float64(x), true
case int16:
return float64(x), true
case int32:
return float64(x), true
case int64:
return float64(x), true
case uint:
return float64(x), true
case uint8:
return float64(x), true
case uint16:
return float64(x), true
case uint32:
return float64(x), true
case uint64:
return float64(x), true
case float32:
return float64(x), true
case []string:
out := make([]interface{}, len(x))
for i, s := range x {
out[i] = s
}
return out, true
case map[string]string:
out := make(map[string]interface{}, len(x))
for k, s := range x {
out[k] = s
}
return out, true
case map[string]interface{}:
// The overwhelmingly common case, and the one the hook path always
// takes at the top level.
out := make(map[string]interface{}, len(x))
for k, it := range x {
c, ok := fastToPlain(it)
if !ok {
return nil, false
}
out[k] = c
}
return out, true
case []interface{}:
out := make([]interface{}, len(x))
for i, it := range x {
c, ok := fastToPlain(it)
if !ok {
return nil, false
}
out[i] = c
}
return out, true
default:
return nil, false
}
}
// plainForLua returns the interface{} tree to hand to pushGoValue, using the
// fast path when it can and JSON only when it must.
func plainForLua(payload map[string]interface{}) interface{} {
if fast, ok := fastToPlain(payload); ok {
return fast
}
// Rare: some type the fast path does not model. Serialize and let the
// generic decoder normalize it, so the hook still sees the field.
b, err := json.Marshal(payload)
if err != nil {
return payload
}
var decoded interface{}
if err := json.Unmarshal(b, &decoded); err != nil {
return payload
}
return decoded
}
// toPlainSlice is the []interface{} entry point of fastToPlain, exposed so the
// Fire path can flatten a pre-built slice without re-walking the map header.
func toPlainSlice(v []interface{}) []interface{} {
if fast, ok := fastToPlain(v); ok {
return fast.([]interface{})
}
return v
}
// luaNumberString renders a float the way Lua would print it, for the rare case
// a hook wants the textual form. Kept out of the hot path.
func luaNumberString(f float64) string {
return strconv.FormatFloat(f, 'g', -1, 64)
}

View File

@ -0,0 +1,132 @@
package lua
import (
"encoding/json"
"reflect"
"testing"
)
// The hook payload used to be JSON round-tripped on every call. It no longer is,
// so the fast path and the old JSON path must be indistinguishable — otherwise
// a plugin silently sees a different payload than before, which is the worst
// kind of change: it compiles, passes a smoke test, and misprices traffic.
//
// These tests therefore compare the two paths on the SAME inputs rather than
// asserting the fast path's output in isolation.
func viaJSON(t *testing.T, payload map[string]interface{}) interface{} {
t.Helper()
b, err := json.Marshal(payload)
if err != nil {
t.Fatalf("marshal: %v", err)
}
var decoded interface{}
if err := json.Unmarshal(b, &decoded); err != nil {
t.Fatalf("unmarshal: %v", err)
}
return decoded
}
// payloads that exercise every branch of the fast converter.
func payloadCases() []map[string]interface{} {
return []map[string]interface{}{
{"model": "deepseek-v4.1-flash", "source": "commandcode", "ok": true},
// Numbers of every width: a converter that only knows float64 turns
// ints into something else, and a token count that arrives as a string
// makes a Lua hook do arithmetic on nil.
{"i": 42, "i8": int8(8), "i16": int16(16), "i32": int32(32), "i64": int64(1 << 40),
"u": uint(7), "u64": uint64(1 << 50), "f32": float32(1.5), "f64": 2.25},
// Zero, negative, and very large values must stay numbers.
{"zero": 0, "neg": -17, "huge": 1e308, "tiny": 1e-308},
// Slices and string maps: common in payloads and absent from a switch
// that only knows []interface{} / map[string]interface{}.
{"msgs": []interface{}{"a", "b"}, "tags": []string{"x", "y"}},
{"kv": map[string]string{"a": "1", "b": "2"}},
// Nesting, which is where a shallow converter silently drops a level.
{"usage": map[string]interface{}{
"prompt_tokens": 2048, "cache_hit_tokens": 1024,
"nested": map[string]interface{}{"deep": []interface{}{1, "two", true, nil}},
}},
{"nil_field": nil, "empty_map": map[string]interface{}{}, "empty_slice": []interface{}{}},
// A value the fast path does NOT model: it must fall back to JSON and
// still arrive, not disappear.
{"weird": struct {
A int `json:"a"`
B string `json:"b"`
}{1, "x"}},
}
}
func TestFastPathMatchesJSONPath(t *testing.T) {
for i, p := range payloadCases() {
want := viaJSON(t, p)
got := plainForLua(p)
if !reflect.DeepEqual(want, got) {
t.Errorf("case %d: fast path differs from JSON path\n payload: %#v\n json: %#v\n fast: %#v",
i, p, want, got)
}
}
}
// TestFastPathIsActuallyUsed guards against the fast path silently degrading to
// JSON for the payload the gateway really sends. If a future payload gains a
// type the converter does not model, this still works (it falls back) but the
// optimization is gone — and the next person measuring the hook would be
// measuring the old cost without knowing why.
func TestFastPathIsActuallyUsed(t *testing.T) {
// This mirrors the real request_end payload shape from the gateway.
realistic := map[string]interface{}{
"stage": "request_end", "kind": "end", "model": "deepseek-v4.1-flash",
"source": "commandcode", "key": "stress-key", "ok": true,
"status": 200, "duration_ms": 1234,
"usage": map[string]interface{}{
"prompt_tokens": float64(2048), "completion_tokens": float64(512),
"cache_hit_tokens": float64(1024), "total_tokens": float64(3584),
},
"walk": []interface{}{
map[string]interface{}{"kind": "tier_skip", "tier": 1, "source": "", "model": "", "reason": "cooldown"},
map[string]interface{}{"kind": "selected", "tier": 2, "source": "commandcode", "model": "m", "reason": ""},
},
"ts": float64(1700000000),
}
if _, ok := fastToPlain(realistic); !ok {
t.Errorf("★ the realistic request payload does NOT take the fast path — " +
"the JSON round-trip is still on the hot path for real traffic")
}
}
// TestFastPathDoesNotAliasInput: the converter builds a new tree. If it ever
// returned the caller's map directly, a Lua hook's writes could not reach Go —
// but worse, a later mutation of the payload would race with the snapshot the
// persistence saver is holding.
func TestFastPathDoesNotAliasInput(t *testing.T) {
src := map[string]interface{}{
"usage": map[string]interface{}{"prompt_tokens": float64(1)},
"list": []interface{}{"a"},
}
out, ok := fastToPlain(src)
if !ok {
t.Fatal("fast path declined a plain payload")
}
m := out.(map[string]interface{})
m["new"] = "added"
src["also_new"] = "must not appear"
if _, leaked := m["also_new"]; leaked {
t.Error("output map aliases the input map")
}
if _, leaked := src["new"]; leaked {
t.Error("writing to the output mutated the input")
}
// And the nested maps must be copies too.
inner := m["usage"].(map[string]interface{})
inner["prompt_tokens"] = float64(999)
if src["usage"].(map[string]interface{})["prompt_tokens"] != float64(1) {
t.Error("nested map is shared, not copied — a hook could mutate the payload")
}
list := m["list"].([]interface{})
list[0] = "changed"
if src["list"].([]interface{})[0] != "a" {
t.Error("nested slice is shared, not copied")
}
}

View File

@ -0,0 +1,334 @@
package lua
import (
"os"
"path/filepath"
"sync"
"testing"
"time"
)
// Fire runs same-stage hooks in parallel across plugins. These tests pin the
// three things that make that safe, each of which failed at least once while
// the change was being written.
// twoHookPlugins loads two plugins that both hook request_end.
func twoHookPlugins(t *testing.T, codeA, codeB string) *Plugins {
t.Helper()
vm := NewVM(filepath.Join(t.TempDir(), "adapters"))
if err := vm.Start(); err != nil {
t.Fatalf("vm start: %v", err)
}
t.Cleanup(vm.Stop)
ps := NewPlugins(vm, filepath.Join(t.TempDir(), "plugins"))
ps.save.setWakeDelayForTest(time.Millisecond)
t.Cleanup(ps.Close)
ps.DisableStatePersistence()
if err := ps.LoadSource("aaa", codeA); err != nil {
t.Fatalf("load aaa: %v", err)
}
if err := ps.LoadSource("bbb", codeB); err != nil {
t.Fatalf("load bbb: %v", err)
}
return ps
}
func counterPlugin(name string, key string) string {
// Built by concatenation rather than fmt: %q is not a Go verb, and the first
// attempt produced Lua source with a stray '=' that only failed at compile
// time inside three different tests.
return `
local p = { name = "` + name + `" }
p.state = { n = 0 }
p.hooks = { request_end = "bump" }
function p.bump(payload)
p.state.n = p.state.n + 1
if "` + key + `" ~= "" then return { who = "from-` + name + `" } end
return nil
end
return p
`
}
// TestSingleAndParallelPathsMergeIdentically: the single-plugin fast path and
// the multi-plugin parallel path must produce the same payload.
//
// This is not theoretical. The parallel path was written first and the
// single-plugin path kept its old shape; the merge loop was left off the fast
// path, so a lone plugin returning a table had its return value DISCARDED. The
// existing stage-order test caught it — but only because it happened to check
// the payload after Fire. A plugin that returned fields nobody read would have
// broken silently.
func TestSingleAndParallelPathsMergeIdentically(t *testing.T) {
single := twoHookPlugins(t, counterPlugin("aaa", "who"), `
local p = { name = "zzz" }
p.state = { n = 0 }
p.hooks = { request_end = "bump" }
function p.bump(payload) p.state.n = p.state.n + 1 return nil end
return p
`)
// remove the second so exactly one plugin hooks this stage
single.mu.Lock()
single.plugins = single.plugins[:1]
single.mu.Unlock()
single.rebuild()
ps := twoHookPlugins(t, counterPlugin("aaa", "who"), counterPlugin("bbb", "who"))
singleOut := single.Fire(StageRequestEnd, map[string]interface{}{"model": "m"})
psOut := ps.Fire(StageRequestEnd, map[string]interface{}{"model": "m"})
if singleOut["who"] != "from-aaa" {
t.Errorf("★ single-plugin path dropped the return value: %v", singleOut)
}
// Two plugins both return `who`; the merge order decides the winner, and it
// must be deterministic (load order), not scheduler-dependent.
if psOut["who"] != "from-bbb" {
t.Errorf("merge order is not load order: who = %v, want from-bbb", psOut["who"])
}
}
// TestMergeOrderIsDeterministic: goroutine completion order must not leak into
// the result. Running Fire repeatedly must always yield the same payload, or a
// gateway's behaviour changes run to run with the same plugins installed.
func TestMergeOrderIsDeterministic(t *testing.T) {
// The plugins differ in cost so a scheduler-dependent merge is visible.
slow := `
local p = { name = "aaa" }
p.state = { n = 0 }
p.hooks = { request_end = "slow" }
function p.slow(payload)
local acc = 0
for i = 1, 3000 do acc = acc + i % 7 end
p.state.n = p.state.n + 1
return { winner = "aaa", acc = acc }
end
return p
`
fast := `
local p = { name = "bbb" }
p.state = { n = 0 }
p.hooks = { request_end = "fast" }
function p.fast(payload)
p.state.n = p.state.n + 1
return { winner = "bbb" }
end
return p
`
ps := twoHookPlugins(t, slow, fast)
first := ps.Fire(StageRequestEnd, map[string]interface{}{"model": "m"})["winner"]
for i := 0; i < 60; i++ {
got := ps.Fire(StageRequestEnd, map[string]interface{}{"model": "m"})["winner"]
if got != first {
t.Fatalf("★ winner changed between runs: %q then %q — merge depends on goroutine "+
"scheduling, so the same configuration behaves differently run to run", first, got)
}
}
// Load order decides: bbb is loaded second, so it wins.
if first != "bbb" {
t.Errorf("winner = %q, want bbb (load order decides, not timing)", first)
}
}
// TestThrowingHookDoesNotBreakSiblings covers the failure path that golua
// actually produces: a hook raising a Lua error.
//
// The test name used to claim it covered panics, and it used error() to do it.
// Probing the six ways a Lua program can fault (error(), indexing nil, calling
// nil, concatenating nil, arithmetic on nil, unbounded recursion) showed golua
// converts ALL of them into an error RETURN, not a Go panic — so the recover()
// in Fire was untested by that case, and deleting recover() still passed. The
// test was renamed to say what it verifies.
//
// recover() is kept anyway: it guards the Go side of Fire (a nil map write, a
// future change to how the payload is prepared), which is cheap and cannot be
// triggered from Lua today. Claiming it is covered by a Lua test would be the
// kind of assurance that evaporates the first time someone checks.
func TestThrowingHookDoesNotBreakSiblings(t *testing.T) {
throwing := `
local p = { name = "aaa" }
p.state = {}
p.hooks = { request_end = "boom" }
function p.boom(payload) error("intentional failure") end
return p
`
healthy := counterPlugin("bbb", "")
ps := twoHookPlugins(t, throwing, healthy)
out := ps.Fire(StageRequestEnd, map[string]interface{}{"model": "m"}) // must not crash
if out == nil {
t.Error("Fire returned nil")
}
st, ok := ps.State("bbb").(map[string]interface{})
if !ok {
t.Fatal("healthy plugin has no state")
}
if n := st["n"]; n != float64(1) {
t.Errorf("★ healthy plugin did not run alongside the throwing one: n = %v", n)
}
he := ps.HookErrors()
if len(he) == 0 {
t.Error("a throwing hook was not recorded in hook_errors — the failure would be invisible")
}
}
// TestParallelHooksAllRunOnce: every plugin must be invoked exactly once per
// Fire. A lost or duplicated goroutine shows up as a wrong total, which for the
// billing plugin means a wrong bill.
func TestParallelHooksAllRunOnce(t *testing.T) {
ps := twoHookPlugins(t, counterPlugin("aaa", ""), counterPlugin("bbb", ""))
const fires = 100
for i := 0; i < fires; i++ {
ps.Fire(StageRequestEnd, map[string]interface{}{"model": "m"})
}
for _, name := range []string{"aaa", "bbb"} {
st, ok := ps.State(name).(map[string]interface{})
if !ok {
t.Fatalf("%s has no state", name)
}
if n := st["n"]; n != float64(fires) {
t.Errorf("%s counted %v hooks, want %d", name, n, fires)
}
}
}
// TestConcurrentFireIsSafe drives Fire from many goroutines at once. Each
// plugin has its own Lua state and its own mutex, so this must hold; the race
// detector is what proves it, not the assertions.
func TestConcurrentFireIsSafe(t *testing.T) {
ps := twoHookPlugins(t, counterPlugin("aaa", ""), counterPlugin("bbb", ""))
var wg sync.WaitGroup
for g := 0; g < 8; g++ {
wg.Add(1)
go func() {
defer wg.Done()
for i := 0; i < 50; i++ {
ps.Fire(StageRequestEnd, map[string]interface{}{"model": "m", "source": "s"})
}
}()
}
wg.Wait()
for _, name := range []string{"aaa", "bbb"} {
st, ok := ps.State(name).(map[string]interface{})
if !ok {
t.Fatalf("%s has no state", name)
}
if n := st["n"]; n != float64(400) {
t.Errorf("%s counted %v, want 400", name, n)
}
}
}
// TestFireDoesNotHoldRegistryLockWhileRunningHooks: a hook that installs or
// disables a plugin takes ps.mu for write. If Fire held ps.mu across the hook,
// that would self-deadlock — the exact failure TrueAgent recorded for its own
// parallel stop path.
func TestFireDoesNotHoldRegistryLockWhileRunningHooks(t *testing.T) {
reentrant := `
local p = { name = "aaa" }
p.state = {}
p.hooks = { request_end = "reenter" }
function p.reenter(payload)
-- Reading the registry from inside a hook is the read half of the same lock.
local _ = #payload
return nil
end
return p
`
vm := NewVM(filepath.Join(t.TempDir(), "adapters"))
if err := vm.Start(); err != nil {
t.Fatal(err)
}
defer vm.Stop()
ps := NewPlugins(vm, filepath.Join(t.TempDir(), "plugins"))
defer ps.Close()
ps.DisableStatePersistence()
if err := ps.LoadSource("aaa", reentrant); err != nil {
t.Fatal(err)
}
if err := ps.LoadSource("bbb", counterPlugin("bbb", "")); err != nil {
t.Fatal(err)
}
done := make(chan struct{})
go func() {
defer close(done)
ps.Fire(StageRequestEnd, map[string]interface{}{"model": "m"})
// Touch the registry the way an admin request would, right after.
_ = ps.Count()
_ = ps.List()
}()
select {
case <-done:
case <-time.After(10 * time.Second):
t.Fatal("Fire deadlocked against the registry lock")
}
}
// TestBillingSurvivesParallelFire runs the REAL bundled plugin next to another
// one. A synthetic counter cannot catch a mismatch between the documented
// payload shape and what the plugin actually reads.
func TestBillingSurvivesParallelFire(t *testing.T) {
dir := t.TempDir()
pdir := filepath.Join(dir, "plugins")
os.MkdirAll(pdir, 0o755)
src, err := os.ReadFile("plugins/billing.lua")
if err != nil {
t.Fatalf("read billing.lua: %v", err)
}
if err := os.WriteFile(filepath.Join(pdir, "billing.lua"), src, 0o644); err != nil {
t.Fatal(err)
}
if err := os.WriteFile(filepath.Join(pdir, "observer.lua"),
[]byte(`local p={name="observer"} p.state={n=0}
p.hooks={request_end="see"}
function p.see(payload) p.state.n=p.state.n+1 return nil end
return p`), 0o644); err != nil {
t.Fatal(err)
}
vm := NewVM(filepath.Join(dir, "adapters"))
if err := vm.Start(); err != nil {
t.Fatal(err)
}
defer vm.Stop()
ps := NewPlugins(vm, pdir)
ps.save.setWakeDelayForTest(time.Millisecond)
defer ps.Close()
ps.DisableStatePersistence()
if err := ps.LoadDir(); err != nil {
t.Fatal(err)
}
if ps.Count() != 2 {
t.Fatalf("loaded %d plugins, want 2", ps.Count())
}
payload := map[string]interface{}{
"model": "deepseek-v4.1-flash", "source": "commandcode", "key": "k", "ok": true,
"usage": map[string]interface{}{
"prompt_tokens": float64(1000), "completion_tokens": float64(100),
"cache_hit_tokens": float64(0),
},
}
ps.Fire(StageRequestEnd, payload)
st := ps.State("billing")
if st == nil {
t.Fatal("billing produced no state")
}
m := st.(map[string]interface{})
total, ok := m["total"].(map[string]interface{})
if !ok {
t.Fatalf("billing.total is %T", m["total"])
}
if total["requests"] != float64(1) {
t.Errorf("billing counted %v requests, want 1 — the plugin was starved by the parallel path",
total["requests"])
}
if n := ps.State("observer").(map[string]interface{})["n"]; n != float64(1) {
t.Errorf("observer counted %v, want 1", n)
}
if len(ps.HookErrors()) != 0 {
t.Errorf("hook errors under parallel Fire: %v", ps.HookErrors())
}
}

View File

@ -240,7 +240,7 @@ type Plugins struct {
type stateSaver struct { type stateSaver struct {
mu sync.Mutex mu sync.Mutex
ps *Plugins ps *Plugins
dirty map[string]pendingState dirty map[string]bool
wake chan struct{} wake chan struct{}
stop chan struct{} stop chan struct{}
done chan struct{} done chan struct{}
@ -257,26 +257,13 @@ type stateSaver struct {
writes atomic.Int64 writes atomic.Int64
} }
// pendingState is a plugin's state ALREADY CONVERTED to Go values.
//
// The conversion happens on the caller's goroutine, inside p.mu, while the Lua
// VM is guaranteed alive. Doing it in the flush goroutine instead looked
// harmless and was a use-after-free: vm.Stop() frees the Lua states, and the
// saver then walked them — measured as a SIGSEGV inside golua, not as a clean
// error. A background goroutine must never touch the VM.
type pendingState struct {
name string
state interface{}
prices interface{}
}
// setWakeDelayForTest shortens the debounce so tests do not pay it on teardown. // setWakeDelayForTest shortens the debounce so tests do not pay it on teardown.
func (s *stateSaver) setWakeDelayForTest(d time.Duration) { s.wakeDelay = d } func (s *stateSaver) setWakeDelayForTest(d time.Duration) { s.wakeDelay = d }
func newStateSaver(ps *Plugins) *stateSaver { func newStateSaver(ps *Plugins) *stateSaver {
s := &stateSaver{ s := &stateSaver{
ps: ps, ps: ps,
dirty: map[string]pendingState{}, dirty: map[string]bool{},
wake: make(chan struct{}, 1), wake: make(chan struct{}, 1),
stop: make(chan struct{}), stop: make(chan struct{}),
done: make(chan struct{}), done: make(chan struct{}),
@ -287,16 +274,15 @@ func newStateSaver(ps *Plugins) *stateSaver {
return s return s
} }
// mark records a snapshot as needing a write. Never blocks: the channel is // markDirty records that a plugin's state changed. Never blocks: the channel is
// buffered, and a full buffer means a flush is already pending, which is // buffered, and a full buffer means a flush is already pending, which is exactly
// exactly the state we want. Only the LATEST snapshot per plugin is kept, so a // the state we want.
// burst of N mutations collapses to one write. func (s *stateSaver) markDirty(name string) {
func (s *stateSaver) mark(p pendingState) {
if s == nil { if s == nil {
return return
} }
s.mu.Lock() s.mu.Lock()
s.dirty[p.name] = p s.dirty[name] = true
s.mu.Unlock() s.mu.Unlock()
select { select {
case s.wake <- struct{}{}: case s.wake <- struct{}{}:
@ -335,20 +321,17 @@ func (s *stateSaver) flush() {
s.mu.Unlock() s.mu.Unlock()
return return
} }
pending := make([]pendingState, 0, len(s.dirty)) names := make([]string, 0, len(s.dirty))
for _, p := range s.dirty { for n := range s.dirty {
pending = append(pending, p) names = append(names, n)
} }
s.dirty = map[string]pendingState{} s.dirty = map[string]bool{}
s.mu.Unlock() s.mu.Unlock()
// File I/O only. The Lua states are not touched here — see pendingState. // Snapshot + write here, off the request path. This is the ONLY place the
for _, p := range pending { // plugin's Lua tables are walked for persistence.
writeJSONAtomic(s.ps.stateFile(p.name), map[string]interface{}{ for _, n := range names {
"version": 1, s.ps.persistByName(n)
"state": p.state,
"prices": p.prices,
})
s.writes.Add(1) s.writes.Add(1)
} }
} }
@ -435,6 +418,9 @@ func NewPlugins(vm *VM, dir string) *Plugins {
// accumulation is lost, which is the same "restart loses the books" defect this // accumulation is lost, which is the same "restart loses the books" defect this
// persistence exists to fix, just scoped to a few seconds instead of the whole // persistence exists to fix, just scoped to a few seconds instead of the whole
// process lifetime. // process lifetime.
// DisableStatePersistence turns persistence off (tests/benchmarks only).
func (ps *Plugins) DisableStatePersistence() { ps.stateDirDisabled = true }
func (ps *Plugins) Close() { func (ps *Plugins) Close() {
if ps == nil { if ps == nil {
return return
@ -1018,7 +1004,8 @@ func readPricesFromPayload(state interface{}) interface{} {
// It is called from the hook path, so it must not block on file I/O — that is // It is called from the hook path, so it must not block on file I/O — that is
// the saver's job. It DOES walk the Lua tables, because that has to happen // the saver's job. It DOES walk the Lua tables, because that has to happen
// while the caller still holds p.mu and the VM is guaranteed alive; the flush // while the caller still holds p.mu and the VM is guaranteed alive; the flush
// goroutine only ever sees the resulting Go values (see pendingState). // (Snapshotting in the flush goroutine instead was a use-after-free: vm.Stop()
// frees the Lua states. See persistByName for why the walk is safe here.)
// //
// Cost is one JSON conversion per request, which the billing plugin would pay // Cost is one JSON conversion per request, which the billing plugin would pay
// anyway inside its own hook. // anyway inside its own hook.
@ -1026,15 +1013,43 @@ func (ps *Plugins) markDirtyLocked(p *Plugin) {
if ps == nil || p == nil || ps.stateDirDisabled || p.state == nil || ps.dir == "" { if ps == nil || p == nil || ps.stateDirDisabled || p.state == nil || ps.dir == "" {
return return
} }
// Caller already holds p.mu — taking it again would self-deadlock. That is // NOTE: this only records THAT the state changed. It deliberately does not
// not hypothetical: the first version had markDirty lock p.mu and was called // snapshot it.
// from invoke(), which holds p.mu for the whole Lua call, so every hook call //
// deadlocked and the lua package's tests hung until the timeout. // Snapshotting here cost 264us per request — the snapshot walks the plugin's
state, prices := readStateAndPrices(p.state.L) // Lua tables through luaValueToGo + json.Marshal + json.Unmarshal (three
if state == nil { // conversions), and the billing plugin does it on every single request. That
// turned a 14.6us hook into a 385us one and put the gateway's hot path
// behind a bookkeeping step. The saver pulls the snapshot on its own schedule
// instead, so N requests between flushes cost ONE snapshot.
ps.save.markDirty(p.Info.Name)
}
// persistByName snapshots one plugin and writes its state file.
//
// It is called from the saver goroutine, which means it walks a Lua state that
// vm.Stop() may be about to free. That was a real crash: the first version had
// the saver read the tables itself and the process died with SIGSEGV inside
// golua once shutdown raced a flush.
//
// Two things make it safe now:
// - Plugins.Close() stops the saver AND waits for its final flush BEFORE the
// caller stops the VM (see Plugins.Close), so no flush can be in flight when
// the Lua states go away.
// - p.mu is held while the tables are read, so a hook cannot be mutating them
// underneath. A hook that arrives after the VM stopped cannot run either,
// because the request path is already torn down by then.
func (ps *Plugins) persistByName(name string) {
if ps == nil || ps.stateDirDisabled || ps.dir == "" {
return return
} }
ps.save.mark(pendingState{name: p.Info.Name, state: state, prices: prices}) ps.mu.RLock()
p := ps.find(name)
ps.mu.RUnlock()
if p == nil || p.state == nil {
return
}
ps.persistNow(p)
} }
// ---------- state persistence ---------- // ---------- state persistence ----------
@ -1296,6 +1311,19 @@ func (ps *Plugins) Fire(stage Stage, payload map[string]interface{}) map[string]
if len(calls) == 0 { if len(calls) == 0 {
return payload return payload
} }
// Resolve the plugins and drop the lock BEFORE running any hook.
//
// A hook can install, disable or remove a plugin (the admin API is reachable
// from a hook that has the key), and those paths take ps.mu for write. Taking
// a per-iteration RLock like the single-plugin case did works but serializes
// on a shared cache line under load; a snapshot plus one lookup is cheaper and
// — more importantly — it means the parallel section below never holds ps.mu,
// so a hook cannot deadlock against a concurrent reload.
type target struct {
p *Plugin
hc hookCall
}
targets := make([]target, 0, len(calls))
for _, hc := range calls { for _, hc := range calls {
ps.mu.RLock() ps.mu.RLock()
p := ps.plugins[hc.pluginIdx] p := ps.plugins[hc.pluginIdx]
@ -1303,20 +1331,82 @@ func (ps *Plugins) Fire(stage Stage, payload map[string]interface{}) map[string]
if p == nil || p.LoadError != "" { if p == nil || p.LoadError != "" {
continue continue
} }
out, err := ps.invoke(p, hc.fn, payload) targets = append(targets, target{p: p, hc: hc})
if err != nil { }
ps.hookErr.note(stage, p.Info.Name+": "+err.Error()) if len(targets) == 0 {
continue return payload
} }
if len(out) > 0 {
// One plugin is the overwhelmingly common case (a gateway with the bundled
// billing plugin has exactly one). Paying for goroutines there would be pure
// overhead, so it takes the direct path.
if len(targets) == 1 {
if out := ps.runHook(stage, targets[0].p, targets[0].hc.fn, payload); len(out) > 0 {
for k, v := range out { for k, v := range out {
payload[k] = v payload[k] = v
} }
} }
return payload
}
// Parallel across plugins. Each hook receives the SAME payload snapshot, and
// the returned tables are merged afterwards IN PLUGIN LOAD ORDER, so the
// documented contract ("a returned table's keys are merged into the payload")
// still holds deterministically.
//
// What changes: a hook no longer sees the keys another hook just added. The
// sequential behaviour made that possible, and docs/plugins.md described it
// ("the payload is passed to the next plugin unchanged"). It was never used —
// the bundled billing plugin returns nil on every stage, with a comment saying
// nobody downstream reads it — but it IS a contract change and is called out
// there rather than left as a surprise.
//
// Why this is safe for the per-plugin lock: each plugin has its own Lua state
// and its own p.mu, so two plugins never touch the same state. What is shared
// is the payload, and it is only READ here — merging happens after every hook
// has returned, on the caller's goroutine.
outs := make([]map[string]interface{}, len(targets))
var wg sync.WaitGroup
for i := range targets {
wg.Add(1)
go func(i int) {
defer wg.Done()
// A hook that panics would otherwise take the whole process with it,
// and in the sequential version a panic could not escape Fire either.
// recover() here restores that: the plugin is skipped and noted.
defer func() {
if rec := recover(); rec != nil {
ps.hookErr.note(stage, fmt.Sprintf("%s: panic in hook: %v", targets[i].hc.plugin, rec))
}
}()
outs[i] = ps.runHook(stage, targets[i].p, targets[i].hc.fn, payload)
}(i)
}
wg.Wait()
// Merge in load order so the result does not depend on goroutine scheduling.
// A later plugin's value wins on a key collision, exactly as it did when the
// hooks ran one after another.
for _, out := range outs {
for k, v := range out {
payload[k] = v
}
} }
return payload return payload
} }
// runHook invokes one hook and folds its returned table into payload. It
// contains the plugin's error handling so both the sequential and the parallel
// path behave identically on failure.
func (ps *Plugins) runHook(stage Stage, p *Plugin, fn string, payload map[string]interface{}) map[string]interface{} {
out, err := ps.invoke(p, fn, payload)
if err != nil {
ps.hookErr.note(stage, p.Info.Name+": "+err.Error())
return nil
}
return out
}
// invoke runs one plugin hook on that plugin's own state, under its pool's // invoke runs one plugin hook on that plugin's own state, under its pool's
// concurrency cap. The plugin's returned table is re-fetched each call because // concurrency cap. The plugin's returned table is re-fetched each call because
// the pool is elastic: a plugin may have several states, and the hook function // the pool is elastic: a plugin may have several states, and the hook function
@ -1348,15 +1438,13 @@ func (ps *Plugins) invoke(p *Plugin, fn string, payload map[string]interface{})
// themselves), but a plugin hook receives a table so it can read // themselves), but a plugin hook receives a table so it can read
// payload.model directly. Passing the string made every hook fail with // payload.model directly. Passing the string made every hook fail with
// "attempt to index local 'payload' (a string value)". // "attempt to index local 'payload' (a string value)".
raw, err := json.Marshal(payload) //
if err != nil { // Decoding is done WITHOUT JSON. json.Marshal + json.Unmarshal here cost
return nil, err // 9.6us of the 14.6us hook — 67% of the call spent re-deriving a tree that
} // pushGoValue walks natively anyway. plainForLua does the same conversion
var decoded interface{} // by type-switching, and falls back to JSON only for a type it does not
if err := json.Unmarshal(raw, &decoded); err != nil { // model, so an exotic payload still arrives instead of vanishing.
return nil, err pushGoValue(L, plainForLua(payload))
}
pushGoValue(L, decoded)
// Call takes NO function index: it invokes whatever sits directly below the // Call takes NO function index: it invokes whatever sits directly below the
// nargs values it just pushed. Passing an index here is a compile-time no-op // nargs values it just pushed. Passing an index here is a compile-time no-op
// in this binding and the call lands on the argument instead // in this binding and the call lands on the argument instead