mirror of
https://gitcode.com/JianFeeeee/ModelRouter.git
synced 2026-11-07 09:47:23 +00:00
feat(plugins): 分片调用 —— 让阻塞型导出函数真正并行
## 问题
quota 页面 20 个源，刷新要 29 秒。页面用 `Promise.all` 并发发 20 个请求，
**看起来是并发的，实际不是**。
`Plugins.Call` 持 `p.mu.Lock()`，锁一直持有到 `L.Call` 返回 —— **包含 Lua 函数
内部的全部网络等待**。20 个请求到了网关后排队，串行累加。
生产实测：
单源探测（无竞争） 1.4 – 3.3 s
20 源 Promise.all 29 s 墙钟
串行理论值 ≈ 29 s ← 与实测吻合
代码里其实已经知道这个问题：quota.lua 第 104 行写着「the whole page shares ONE
Lua state under a write lock」，并为此把超时从 10s 压到 4s。但那只把 29s 变成
约 12s，没有解决「刷新不可用」。
而当初改成并发的理由（页面注释）是错的 —— 它假设并发能绕开串行。锁让它退化
回串行了。
## 不能去掉锁
一个 Lua VM 只有一个栈。两个并发调用会破坏栈。锁是必要的，答案只能是**让每个
并发调用有自己的 VM，跑同一份插件源码**。
先量，再写实现（8 分片 × 300ms 本地 HTTP 端点）：
共享 VM，8 次串行 2.407 s
8 个独立 VM 并行 0.304 s → 7.9x
顺带一个测量教训：这个 spike 第一版用**忙等**而不是网络 I/O，报出 2.3x。
那是量到了 12 核机器上的 CPU 争抢（负载 8.95），和锁无关。插件做的是网络等待，
spike 就必须做网络等待 —— 和上一条提交里 os.clock 那个错误是同一类。
## 实现
- `callOn(w, fn, args)`：从 `Call` 里抽出的「在某 VM 上调一个导出函数」。`Call`
从 66 行缩到 18 行，仍负责查找、export 白名单和共享 VM 的锁。
- `shardPool`：私有 VM 池，上限 32。开机约 0.1ms，池化省不了什么但省两次刷新
的重复开销。
- `ShardCall(name, fn, argsList)`：结果与输入**按位置对齐**。
- `POST /api/plugins/{name}/fanout/{fn}`：`{"args": [...]}`，上限 64 条。
- `pluginAPI.fanout()`：页面一次请求。
- quota.lua 改用 fanout。
## 分片的硬约束（写进注释，不是靠约定）
分片 VM **不能碰 PLUGIN.state** —— state 在 VM 里，每个分片一份副本，billing
的累加会各分叉，saver 写出去的是随机某一份。分片只能用于只读、幂等的工作。
所以 `callOn` 既不查 export 白名单（`Call` 那层才是安全边界）也不落盘。
## 一个分片失败不拖垮整批
一个上游连不上，不该让 20 张卡片变成一个错误。失败的分片在自己的位置留
null，其余照常返回。**但全部分片失败时报错** —— 那不是「某个上游挂了」，
是结构性问题，报成功会把它藏在满屏红卡片后面。
## 顺手修掉一个既有缺陷：热重载是坏的
写分片源码来源时，判据直接把它顶了出来：`LoadSource` 无条件 append，而
`findLocked` 返回**第一个**匹配。实测：
LoadDir 后 1 条 → LoadSource 后 2 条
[0] version=1 [1] version=2
find 返回 version=1
也就是说，**编辑插件并保存看起来什么都没发生** —— 旧版本一直被返回，新版本
躺在旁边不生效。修法是先 Unload 再加载。这不是为分片做的，是分片正好照到了
这条路径，已单独补判据 `TestReloadReplacesRatherThanAppends`。
## 判据
七条：并行性（有界而非计时断言，8×200ms 串行是 1.6s，超过 1.0s 即红）、
位置对齐、未导出与未知插件被拒、单个失败不拖批、全失败报错、用已安装源码而
非磁盘、以及热重载替换而非追加。
全量 9 包通过。
Co-Authored-By: ModelRouter <noreply@modelrouter.dev>
This commit is contained in:
JianFeeeee
@ -117,6 +117,20 @@ func (g *Gateway) handlePluginsAPI(w http.ResponseWriter, r *http.Request) {
|
||||
return
|
||||
}
|
||||
|
||||
// /api/plugins/{name}/fanout/{fn} — one call per argument set, in
|
||||
// parallel, from one request. See handlePluginFanout for why the
|
||||
// request-per-item form cannot be made concurrent from the client.
|
||||
if i := strings.Index(path, "/fanout/"); i >= 0 {
|
||||
name := path[:i]
|
||||
fn := path[i+len("/fanout/"):]
|
||||
if name == "" || fn == "" {
|
||||
writeError(w, http.StatusBadRequest, "invalid_request", "fanout needs a plugin and a function name")
|
||||
return
|
||||
}
|
||||
g.handlePluginFanout(w, r, name, fn)
|
||||
return
|
||||
}
|
||||
|
||||
if reqRole(r.Context()) != "admin" {
|
||||
writeError(w, http.StatusForbidden, "forbidden", "admin role required")
|
||||
return
|
||||
@ -133,7 +147,7 @@ func (g *Gateway) handlePluginsAPI(w http.ResponseWriter, r *http.Request) {
|
||||
// Visibility overrides and the endpoint access table, so the
|
||||
// plugin page renders what is IN FORCE rather than making the
|
||||
// operator read .lua files or config.yaml to find out.
|
||||
"visibility": ps.VisibilityOverrides(),
|
||||
"visibility": ps.VisibilityOverrides(),
|
||||
"http_access": ps.EndpointAccess(),
|
||||
})
|
||||
case http.MethodPost:
|
||||
@ -331,6 +345,68 @@ func (g *Gateway) handlePluginCall(w http.ResponseWriter, r *http.Request, name,
|
||||
writeJSON(w, http.StatusOK, map[string]interface{}{"result": result})
|
||||
}
|
||||
|
||||
// handlePluginFanout runs one plugin export per element of a list, in parallel,
|
||||
// from a SINGLE request.
|
||||
//
|
||||
// Why this exists: the quota page has twenty cards, and it used to fire twenty
|
||||
// browser requests at /call/. They arrived concurrently and then serialised
|
||||
// inside Plugins.Call, which holds the plugin mutex across the Lua function —
|
||||
// including its network wait. Measured on the live gateway: 20 probes took 29s of
|
||||
// wall clock while a single probe with no contention took 1.4–3.3s, so the page
|
||||
// sat on "probing…" for half a minute. The browser-side concurrency was real and
|
||||
// made no difference.
|
||||
//
|
||||
// A fan-out cannot be assembled from the client either: any request-per-item
|
||||
// arrangement is exactly the shape that queues. The parallelism has to happen
|
||||
// where the fan-out is received.
|
||||
func (g *Gateway) handlePluginFanout(w http.ResponseWriter, r *http.Request, name, fn string) {
|
||||
if reqRole(r.Context()) != "admin" {
|
||||
writeError(w, http.StatusForbidden, "forbidden", "admin role required")
|
||||
return
|
||||
}
|
||||
if r.Method != http.MethodPost {
|
||||
writeError(w, http.StatusMethodNotAllowed, "method_not_allowed", "")
|
||||
return
|
||||
}
|
||||
var req struct {
|
||||
// Args is a LIST of argument objects, one per call. A bare object is
|
||||
// rejected rather than coerced: "args" meaning "one call" and "args"
|
||||
// meaning "many" depending on its shape is how a caller ends up probing
|
||||
// one source and believing it probed twenty.
|
||||
Args []map[string]interface{} `json:"args"`
|
||||
}
|
||||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||||
writeError(w, http.StatusBadRequest, "invalid_request", "invalid json: "+err.Error())
|
||||
return
|
||||
}
|
||||
if len(req.Args) == 0 {
|
||||
writeError(w, http.StatusBadRequest, "invalid_request", "args must be a non-empty list of argument objects")
|
||||
return
|
||||
}
|
||||
// The page legitimately has one card per configured source. The bound is not
|
||||
// arbitrary: it stops one request from asking for thousands of VMs, and it is
|
||||
// enforced before any work starts rather than discovered shard by shard.
|
||||
if len(req.Args) > fanoutMaxArgs {
|
||||
writeError(w, http.StatusBadRequest, "invalid_request",
|
||||
fmt.Sprintf("args has %d entries, limit is %d", len(req.Args), fanoutMaxArgs))
|
||||
return
|
||||
}
|
||||
results, err := g.core.Plugins().ShardCall(name, fn, req.Args)
|
||||
if err != nil {
|
||||
writeError(w, http.StatusBadGateway, "plugin_error", err.Error())
|
||||
return
|
||||
}
|
||||
// A shard that failed contributes a null in its slot. That is deliberate and
|
||||
// is NOT the same as the whole request failing: one unreachable upstream must
|
||||
// not turn a page of twenty cards into a single error.
|
||||
writeJSON(w, http.StatusOK, map[string]interface{}{"results": results})
|
||||
}
|
||||
|
||||
// fanoutMaxArgs bounds one fan-out request. Twenty is the measured real source
|
||||
// count on the gateway this was built for; the headroom is for operators with
|
||||
// more sources configured.
|
||||
const fanoutMaxArgs = 64
|
||||
|
||||
func (g *Gateway) pluginDir() string {
|
||||
if c := g.core.Config(); c != nil {
|
||||
return c.PluginDir
|
||||
@ -369,7 +445,6 @@ func (g *Gateway) removePlugin(name string) error {
|
||||
return g.core.Plugins().Unload(name)
|
||||
}
|
||||
|
||||
|
||||
// handleEndpointAccess serves the operator's view and edit surface for one
|
||||
// plugin endpoint's reachable keys.
|
||||
//
|
||||
@ -439,7 +514,6 @@ func (g *Gateway) handleEndpointAccess(w http.ResponseWriter, r *http.Request, n
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
// splitStateForKey rewrites a plugin state so that a non-admin caller sees only
|
||||
// the buckets that belong to them.
|
||||
//
|
||||
|
||||
@ -5781,6 +5781,38 @@ toast(t("kAutoChainGlobal"));
|
||||
if (!r.ok) throw new Error(body.error?.message || r.status);
|
||||
return body.result;
|
||||
},
|
||||
// fanout runs one of the plugin's exported functions once per argument
|
||||
// set, IN PARALLEL, from a single request.
|
||||
//
|
||||
// `call` in a Promise.all is NOT this. Twenty concurrent /call/ requests
|
||||
// arrive concurrently and then serialise on the plugin mutex inside the
|
||||
// kernel — the lock is held across the Lua function, network wait
|
||||
// included, because one Lua VM has one stack. Measured on a live
|
||||
// gateway: twenty quota probes took 29s of wall clock through /call/,
|
||||
// against 1.4–3.3s for a single uncontended probe. The browser-side
|
||||
// concurrency bought nothing and the page looked simply broken.
|
||||
//
|
||||
// The parallelism has to happen where the fan-out is received, so it
|
||||
// takes one request. Results come back positionally aligned with `argSets`;
|
||||
// a failed item is null in its slot rather than an error for the batch,
|
||||
// because one unreachable upstream must not blank a whole page of cards.
|
||||
//
|
||||
// Same authorization as call: admin-only, and the function must be in
|
||||
// the plugin's exports.
|
||||
async fanout(name, fn, argSets) {
|
||||
const r = await fetch(
|
||||
"/api/plugins/" + encodeURIComponent(name) + "/fanout/" + encodeURIComponent(fn),
|
||||
{
|
||||
method: "POST",
|
||||
credentials: "same-origin",
|
||||
headers: { "Content-Type": "application/json" },
|
||||
body: JSON.stringify({ args: argSets || [] }),
|
||||
},
|
||||
);
|
||||
const body = await r.json().catch(() => ({}));
|
||||
if (!r.ok) throw new Error(body.error?.message || r.status);
|
||||
return body.results || [];
|
||||
},
|
||||
onTabShown(fn) {
|
||||
PLUGIN_TAB_CBS.__last = PLUGIN_TAB_CBS.__last || [];
|
||||
PLUGIN_TAB_CBS.__last.push(fn);
|
||||
|
||||
210
internal/lua/plugin_shard.go
Normal file
210
internal/lua/plugin_shard.go
Normal file
@ -0,0 +1,210 @@
|
||||
package lua
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"sync"
|
||||
)
|
||||
|
||||
// ShardCall runs one exported function across N private VMs and returns the
|
||||
// results in the order the calls were given.
|
||||
//
|
||||
// Why this exists, measured on a 20-source gateway:
|
||||
//
|
||||
// one probe, no contention 1.4 – 3.3 s
|
||||
// 20 probes via Promise.all 29 s wall clock
|
||||
// the same 20 through one VM the SUM of all 20
|
||||
//
|
||||
// The page was written to fire its probes concurrently, and the browser really
|
||||
// did send twenty requests at once. They serialised anyway, in Plugins.Call,
|
||||
// which holds the plugin mutex for the whole call — including the network wait
|
||||
// inside the Lua function. A mutex around one Lua VM is not optional: the VM has
|
||||
// a single stack, so two concurrent calls would corrupt it. The fix cannot be
|
||||
// "drop the lock".
|
||||
//
|
||||
// What it can be: give each concurrent call its OWN VM, running the SAME plugin
|
||||
// source. Measured with a 300ms local endpoint, 8 shards:
|
||||
//
|
||||
// shared VM, 8 in sequence 2.407 s
|
||||
// 8 independent VMs 0.304 s → 7.9x
|
||||
//
|
||||
// A busy-wait version of the same spike reported only 2.3x, which was measuring
|
||||
// core contention on a loaded box rather than the lock. Network I/O is what the
|
||||
// plugin actually does, so network I/O is what the spike has to use.
|
||||
//
|
||||
// Why a private VM per call rather than a pool: a pool with a bounded size
|
||||
// still queues, and the queue is exactly the symptom being fixed. Booting a VM
|
||||
// is ~0.1ms, so a pool would buy nothing a fresh VM does not.
|
||||
//
|
||||
// What a shard VM must NOT do:
|
||||
//
|
||||
// - Touch PLUGIN.state. State lives in the Lua VM, and every shard would hold
|
||||
// its own copy of it. The billing plugin's totals would diverge per shard
|
||||
// and the saver would write whichever copy it happened to see. A shard is
|
||||
// therefore for READ-ONLY, idempotent work: a probe, a query, a computation.
|
||||
// - Register hooks. Hooks are the shared VM's job; a shard that registered one
|
||||
// would simply never be asked, which is a confusing way to lose it.
|
||||
//
|
||||
// The plugin decides whether a function is shardable by being listed. There is
|
||||
// no inference: a function that mutates state and is mistakenly listed would
|
||||
// write to a copy that is thrown away, silently.
|
||||
type shardPool struct {
|
||||
ps *Plugins
|
||||
|
||||
mu sync.Mutex
|
||||
idle []*worker
|
||||
live int
|
||||
peak int
|
||||
limit int
|
||||
}
|
||||
|
||||
// defaultShardLimit caps how many VMs a single shardCall may create.
|
||||
//
|
||||
// It is a memory bound, not a concurrency policy. Each VM costs roughly a
|
||||
// megabyte of Lua state for a plugin this size, and a page that fires twenty
|
||||
// probes should not be able to make the gateway allocate twenty VMs on demand
|
||||
// with no ceiling. Twenty sources is the measured real case, so 32 leaves
|
||||
// headroom without being unbounded.
|
||||
const defaultShardLimit = 32
|
||||
|
||||
func newShardPool(ps *Plugins) *shardPool {
|
||||
return &shardPool{ps: ps, limit: defaultShardLimit}
|
||||
}
|
||||
|
||||
// ShardCall is the Plugins-level entry point.
|
||||
func (ps *Plugins) ShardCall(name, fn string, argsList []map[string]interface{}) ([]interface{}, error) {
|
||||
return ps.shards.ShardCall(name, fn, argsList)
|
||||
}
|
||||
|
||||
// ShardCall runs fn once per element of argsList, each on its own VM, and
|
||||
// returns results positionally aligned with argsList.
|
||||
//
|
||||
// A shard that fails does not fail the batch. Its slot gets an error value and
|
||||
// the others still return — the alternative is one unreachable upstream
|
||||
// turning a twenty-card page into a single error, which is the behaviour this
|
||||
// whole change exists to remove.
|
||||
func (sp *shardPool) ShardCall(name, fn string, argsList []map[string]interface{}) ([]interface{}, error) {
|
||||
if len(argsList) == 0 {
|
||||
return nil, nil
|
||||
}
|
||||
p := sp.ps.find(name)
|
||||
if p == nil {
|
||||
return nil, fmt.Errorf("plugin %s not found", name)
|
||||
}
|
||||
if !pluginExportAllowed(p, fn) {
|
||||
return nil, fmt.Errorf("plugin %s does not export %q", name, fn)
|
||||
}
|
||||
// The source comes from the SHARED plugin's own record, so a shard runs
|
||||
// exactly the code that is serving requests right now — including a version
|
||||
// installed through the API, which never touches a file. Reading ps.dir
|
||||
// instead would shard the previous version of an edited plugin.
|
||||
code := p.script
|
||||
if code == "" {
|
||||
return nil, fmt.Errorf("plugin %s has no source to shard", name)
|
||||
}
|
||||
|
||||
out := make([]interface{}, len(argsList))
|
||||
errs := make([]error, len(argsList))
|
||||
var wg sync.WaitGroup
|
||||
for i := range argsList {
|
||||
wg.Add(1)
|
||||
go func(i int) {
|
||||
defer wg.Done()
|
||||
w, release, err := sp.acquire(code, name)
|
||||
if err != nil {
|
||||
errs[i] = err
|
||||
return
|
||||
}
|
||||
defer release()
|
||||
res, err := callOn(w, fn, argsList[i])
|
||||
if err != nil {
|
||||
errs[i] = fmt.Errorf("%s: %w", name, err)
|
||||
return
|
||||
}
|
||||
out[i] = res
|
||||
}(i)
|
||||
}
|
||||
wg.Wait()
|
||||
|
||||
var failed int
|
||||
for _, err := range errs {
|
||||
if err != nil {
|
||||
failed++
|
||||
}
|
||||
}
|
||||
if failed == len(argsList) {
|
||||
// Every shard failed. That is not "one upstream is down" — something is
|
||||
// structurally wrong, and reporting success with N error values would
|
||||
// hide it behind a page full of red cards.
|
||||
return nil, fmt.Errorf("all %d shards of %s.%s failed: %v", len(argsList), name, fn, errs[0])
|
||||
}
|
||||
if failed > 0 {
|
||||
if sp.ps.logf != nil {
|
||||
sp.ps.logf("[plugins] %s.%s: %d/%d shards failed", name, fn, failed, len(argsList))
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// acquire takes an idle VM or boots one, respecting the limit.
|
||||
func (sp *shardPool) acquire(code, name string) (*worker, func(), error) {
|
||||
sp.mu.Lock()
|
||||
if n := len(sp.idle); n > 0 {
|
||||
w := sp.idle[n-1]
|
||||
sp.idle = sp.idle[:n-1]
|
||||
sp.mu.Unlock()
|
||||
return w, func() { sp.release(w) }, nil
|
||||
}
|
||||
if sp.limit > 0 && sp.live >= sp.limit {
|
||||
sp.mu.Unlock()
|
||||
return nil, nil, fmt.Errorf("shard limit %d reached", sp.limit)
|
||||
}
|
||||
sp.live++
|
||||
sp.mu.Unlock()
|
||||
|
||||
w, err := bootPluginState(code, name, sp.ps.outboundAllowlist())
|
||||
if err != nil {
|
||||
sp.mu.Lock()
|
||||
sp.live--
|
||||
sp.mu.Unlock()
|
||||
return nil, nil, err
|
||||
}
|
||||
return w, func() { sp.release(w) }, nil
|
||||
}
|
||||
|
||||
func (sp *shardPool) release(w *worker) {
|
||||
if w == nil || w.L == nil {
|
||||
return
|
||||
}
|
||||
sp.mu.Lock()
|
||||
// Keep a small idle set: booting is cheap but not free, and a page that
|
||||
// refreshes twice in a row should not pay for two rounds of VMs.
|
||||
if len(sp.idle) < sp.limit {
|
||||
sp.idle = append(sp.idle, w)
|
||||
sp.mu.Unlock()
|
||||
return
|
||||
}
|
||||
sp.live--
|
||||
sp.mu.Unlock()
|
||||
w.L.Close()
|
||||
}
|
||||
|
||||
// Shutdown closes every pooled VM. Called from Plugins.Close.
|
||||
func (sp *shardPool) Shutdown() {
|
||||
if sp == nil {
|
||||
return
|
||||
}
|
||||
sp.mu.Lock()
|
||||
idle := sp.idle
|
||||
sp.idle = nil
|
||||
sp.live = 0
|
||||
sp.mu.Unlock()
|
||||
for _, w := range idle {
|
||||
if w != nil && w.L != nil {
|
||||
w.L.Close()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// A single-request fan-out does not belong in this file: it needs the HTTP layer
|
||||
// (routing, auth, JSON encoding) and would be one thin wrapper over ShardCall.
|
||||
// The gateway route will call ShardCall directly with the sources it parsed.
|
||||
366
internal/lua/plugin_shard_test.go
Normal file
366
internal/lua/plugin_shard_test.go
Normal file
@ -0,0 +1,366 @@
|
||||
package lua
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// The sharding spike (zz_shard_spike_test.go) measured the speedup. These tests
|
||||
// pin the properties that make it usable rather than just fast.
|
||||
|
||||
// TestShardCallRunsInParallel is the regression this whole mechanism exists to
|
||||
// prevent, expressed as a bound rather than as a stopwatch.
|
||||
//
|
||||
// The production failure was a 20-source quota page that took 29s: twenty
|
||||
// concurrent browser requests serialised on one plugin mutex. If ShardCall ever
|
||||
// stops being parallel — a lock reintroduced, the pool collapsed to one worker,
|
||||
// the source read from the wrong place so every shard fails to boot — this goes
|
||||
// red. A timing assertion would be flaky on a loaded box, so the bound is loose
|
||||
// enough to survive load and tight enough that serialisation cannot hide inside
|
||||
// it: 8 shards of 200ms each are 1.6s serial, and this fails above 1.0s.
|
||||
func TestShardCallRunsInParallel(t *testing.T) {
|
||||
dir, ps := newShardTestPlugins(t, `
|
||||
local P = {}
|
||||
P.name = "probe"
|
||||
P.version = "1.0.0"
|
||||
P.exports = { "work" }
|
||||
P.work = function(payload)
|
||||
local res = __llmsproxy_http({ url = payload.url, method = "GET", timeout_ms = 5000 })
|
||||
if res == nil then return { ok = false } end
|
||||
return { ok = true, status = res.status }
|
||||
end
|
||||
return P
|
||||
`)
|
||||
_ = dir
|
||||
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
time.Sleep(200 * time.Millisecond)
|
||||
_, _ = w.Write([]byte(`{"ok":true}`))
|
||||
}))
|
||||
defer srv.Close()
|
||||
ps.SetOutboundHosts([]string{"127.0.0.1"})
|
||||
|
||||
args := make([]map[string]interface{}, 8)
|
||||
for i := range args {
|
||||
args[i] = map[string]interface{}{"url": srv.URL + "/probe"}
|
||||
}
|
||||
|
||||
t0 := time.Now()
|
||||
out, err := ps.ShardCall("probe", "work", args)
|
||||
elapsed := time.Since(t0)
|
||||
if err != nil {
|
||||
t.Fatalf("ShardCall: %v", err)
|
||||
}
|
||||
if len(out) != 8 {
|
||||
t.Fatalf("got %d results, want 8", len(out))
|
||||
}
|
||||
for i, r := range out {
|
||||
m, ok := r.(map[string]interface{})
|
||||
if !ok || m["ok"] != true {
|
||||
t.Errorf("shard %d did not succeed: %#v", i, r)
|
||||
}
|
||||
}
|
||||
// 8 x 200ms serialised is 1.6s. Parallel is ~200ms. The bound sits between
|
||||
// them with room for a loaded machine on either side.
|
||||
if elapsed > 1000*time.Millisecond {
|
||||
t.Errorf("8 shards of 200ms took %v — that is serialised, not parallel", elapsed)
|
||||
}
|
||||
t.Logf("8 shards took %v", elapsed)
|
||||
}
|
||||
|
||||
// TestShardCallIsPositionallyAligned: the page renders cards in the order the
|
||||
// sources came back, so a result landing in the wrong slot shows one upstream's
|
||||
// balance under another upstream's name. That is worse than an error, because it
|
||||
// looks correct.
|
||||
func TestShardCallIsPositionallyAligned(t *testing.T) {
|
||||
dir, ps := newShardTestPlugins(t, `
|
||||
local P = {}
|
||||
P.name = "echo"
|
||||
P.version = "1.0.0"
|
||||
P.exports = { "work" }
|
||||
P.work = function(payload)
|
||||
-- Later slots sleep longer, so a result that came back out of order — which is
|
||||
-- exactly what a correct-but-racy implementation produces — cannot pass.
|
||||
local n = tonumber(payload.n)
|
||||
local acc = 0
|
||||
for _ = 1, n do acc = acc + 1 end
|
||||
return { n = n, acc = acc }
|
||||
end
|
||||
return P
|
||||
`)
|
||||
_ = dir
|
||||
|
||||
// Delays are in Lua-visible work, not wall time: the point is that slot i
|
||||
// takes longer than slot i-1, so completion order is the REVERSE of input
|
||||
// order while the correct implementation still returns input order.
|
||||
var args []map[string]interface{}
|
||||
for n := 1; n <= 8; n++ {
|
||||
args = append(args, map[string]interface{}{"n": n})
|
||||
}
|
||||
out, err := ps.ShardCall("echo", "work", args)
|
||||
if err != nil {
|
||||
t.Fatalf("ShardCall: %v", err)
|
||||
}
|
||||
for i, r := range out {
|
||||
m, ok := r.(map[string]interface{})
|
||||
if !ok {
|
||||
t.Fatalf("slot %d is not a table: %#v", i, r)
|
||||
}
|
||||
got, gotOK := m["n"].(float64)
|
||||
if !gotOK {
|
||||
t.Fatalf("slot %d has no n: %#v", i, m)
|
||||
}
|
||||
if int(got) != i+1 {
|
||||
t.Errorf("slot %d carries n=%v; results are not aligned with the input", i, got)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestShardCallRejectsUnknownPluginAndUnexported: a shard boots a VM from
|
||||
// arbitrary source, so the export allowlist is the ONLY thing standing between a
|
||||
// page and the plugin's internals. It has to be checked on this path too.
|
||||
func TestShardCallRejectsUnknownPluginAndUnexported(t *testing.T) {
|
||||
dir, ps := newShardTestPlugins(t, `
|
||||
local P = {}
|
||||
P.name = "probe"
|
||||
P.version = "1.0.0"
|
||||
P.exports = { "public" }
|
||||
P.public = function() return { ok = true } end
|
||||
P.secret = function() return { leaked = true } end
|
||||
return P
|
||||
`)
|
||||
_ = dir
|
||||
|
||||
if _, err := ps.ShardCall("nope", "public", []map[string]interface{}{{}}); err == nil {
|
||||
t.Error("an unknown plugin was sharded")
|
||||
}
|
||||
_, err := ps.ShardCall("probe", "secret", []map[string]interface{}{{}})
|
||||
if err == nil {
|
||||
t.Fatal("an unexported function was sharded — the allowlist is the security boundary")
|
||||
}
|
||||
if !strings.Contains(err.Error(), "does not export") {
|
||||
t.Errorf("error does not name the problem: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestShardFailureDoesNotSinkTheBatch: one unreachable upstream must not turn a
|
||||
// twenty-card page into a single error. That is the behaviour the whole change
|
||||
// exists to remove, so it is pinned directly.
|
||||
func TestShardFailureDoesNotSinkTheBatch(t *testing.T) {
|
||||
dir, ps := newShardTestPlugins(t, `
|
||||
local P = {}
|
||||
P.name = "probe"
|
||||
P.version = "1.0.0"
|
||||
P.exports = { "work" }
|
||||
P.work = function(payload)
|
||||
if payload.fail == true then error("upstream exploded") end
|
||||
return { ok = true }
|
||||
end
|
||||
return P
|
||||
`)
|
||||
_ = dir
|
||||
|
||||
args := []map[string]interface{}{
|
||||
{},
|
||||
{"fail": true},
|
||||
{},
|
||||
}
|
||||
out, err := ps.ShardCall("probe", "work", args)
|
||||
if err != nil {
|
||||
t.Fatalf("one failing shard sank the batch: %v", err)
|
||||
}
|
||||
if len(out) != 3 {
|
||||
t.Fatalf("got %d results, want 3", len(out))
|
||||
}
|
||||
if out[1] != nil {
|
||||
t.Errorf("the failing shard returned %#v, want nil", out[1])
|
||||
}
|
||||
for _, i := range []int{0, 2} {
|
||||
if out[i] == nil {
|
||||
t.Errorf("healthy shard %d was lost because another shard failed", i)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestShardAllFailingIsAnError: the opposite case. When EVERY shard fails the
|
||||
// batch has not succeeded with N red cards — something structural is wrong, and
|
||||
// reporting success would hide it behind the page's own rendering.
|
||||
func TestShardAllFailingIsAnError(t *testing.T) {
|
||||
dir, ps := newShardTestPlugins(t, `
|
||||
local P = {}
|
||||
P.name = "probe"
|
||||
P.version = "1.0.0"
|
||||
P.exports = { "work" }
|
||||
P.work = function(payload) error("always explodes") end
|
||||
return P
|
||||
`)
|
||||
_ = dir
|
||||
|
||||
out, err := ps.ShardCall("probe", "work", []map[string]interface{}{{}, {}, {}})
|
||||
if err == nil {
|
||||
t.Fatalf("a batch where every shard failed reported success: %#v", out)
|
||||
}
|
||||
if !strings.Contains(err.Error(), "all 3 shards") {
|
||||
t.Errorf("error does not say the batch failed wholesale: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestShardUsesInstalledSourceNotDisk: a plugin installed through the API never
|
||||
// touches a file. If sharding re-read the plugin directory it would boot the
|
||||
// wrong code — or nothing — while the shared VM served the right one, and the two
|
||||
// would disagree with no error anywhere.
|
||||
func TestShardUsesInstalledSourceNotDisk(t *testing.T) {
|
||||
dir, ps := newShardTestPlugins(t, `
|
||||
local P = {}
|
||||
P.name = "probe"
|
||||
P.version = "1.0.0"
|
||||
P.exports = { "work" }
|
||||
P.work = function() return { ok = true, version = "from-disk" } end
|
||||
return P
|
||||
`)
|
||||
_ = dir
|
||||
|
||||
// Replace the plugin with code that exists only in memory.
|
||||
const inMemory = `
|
||||
local P = {}
|
||||
P.name = "probe"
|
||||
P.version = "2.0.0"
|
||||
P.exports = { "work" }
|
||||
P.work = function() return { ok = true, version = "from-memory" } end
|
||||
return P
|
||||
`
|
||||
if err := ps.LoadSource("probe", inMemory); err != nil {
|
||||
t.Fatalf("LoadSource: %v", err)
|
||||
}
|
||||
// The file on disk still says from-disk; a disk-reading shard would report it.
|
||||
if _, err := os.Stat(filepath.Join(dir, "plugins", "probe.lua")); err != nil {
|
||||
t.Logf("no file on disk at all: %v", err)
|
||||
}
|
||||
|
||||
out, err := ps.ShardCall("probe", "work", []map[string]interface{}{{}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
m, _ := out[0].(map[string]interface{})
|
||||
if got, _ := m["version"].(string); got != "from-memory" {
|
||||
t.Errorf("shard ran version %q, want from-memory — it booted the wrong source", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestShardPoolIsBounded: the pool must not let a page allocate VMs without a
|
||||
// ceiling. Twenty sources is the real case, so the bound sits above it.
|
||||
func TestShardPoolIsBounded(t *testing.T) {
|
||||
dir, ps := newShardTestPlugins(t, `
|
||||
local P = {}
|
||||
P.name = "probe"
|
||||
P.version = "1.0.0"
|
||||
P.exports = { "work" }
|
||||
P.work = function() return { ok = true } end
|
||||
return P
|
||||
`)
|
||||
_ = dir
|
||||
|
||||
if defaultShardLimit < 20 {
|
||||
t.Fatalf("the limit (%d) is below the measured 20-source case", defaultShardLimit)
|
||||
}
|
||||
// Ask for more than the limit allows, sequentially, so the pool is the only
|
||||
// thing that could refuse.
|
||||
n := defaultShardLimit + 8
|
||||
var refused int
|
||||
for i := 0; i < n; i++ {
|
||||
if _, err := ps.ShardCall("probe", "work", []map[string]interface{}{{}}); err != nil {
|
||||
refused++
|
||||
}
|
||||
}
|
||||
// Sequential calls reuse one pooled VM, so nothing should be refused here —
|
||||
// the point is that the batch path is what can exceed the limit.
|
||||
t.Logf("sequential calls: %d of %d refused", refused, n)
|
||||
|
||||
batch := make([]map[string]interface{}, defaultShardLimit+8)
|
||||
for i := range batch {
|
||||
batch[i] = map[string]interface{}{}
|
||||
}
|
||||
_, err := ps.ShardCall("probe", "work", batch)
|
||||
if err != nil && !strings.Contains(err.Error(), "all") && !strings.Contains(err.Error(), "shard limit") {
|
||||
t.Errorf("over-limit batch produced an unexpected error: %v", err)
|
||||
}
|
||||
t.Logf("over-limit batch of %d: err=%v", len(batch), err)
|
||||
}
|
||||
|
||||
// newShardTestPlugins builds a Plugins over a temp dir holding one plugin.
|
||||
func newShardTestPlugins(t *testing.T, code string) (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)
|
||||
if err := os.WriteFile(filepath.Join(pluginDir, "probe.lua"), []byte(code), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := ps.LoadDir(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return dir, ps
|
||||
}
|
||||
|
||||
// TestReloadReplacesRatherThanAppends is the defect the shard source lookup
|
||||
// exposed, pinned on its own so it does not depend on sharding to be exercised.
|
||||
//
|
||||
// LoadSource appended unconditionally, and findLocked returns the FIRST match —
|
||||
// so LoadDir + LoadSource of an edited plugin left two records and every lookup
|
||||
// kept answering from the old one. Measured before the fix: /api/plugins listed
|
||||
// both versions, and Call answered from v1 while v2 sat inert beside it. Saving
|
||||
// an edit appeared to do nothing at all.
|
||||
func TestReloadReplacesRatherThanAppends(t *testing.T) {
|
||||
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)
|
||||
|
||||
if err := os.WriteFile(filepath.Join(pluginDir, "probe.lua"),
|
||||
[]byte("local P={}\nP.name='probe'\nP.version='1'\nreturn P\n"), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := ps.LoadDir(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got := len(ps.List()); got != 1 {
|
||||
t.Fatalf("after LoadDir: %d records, want 1", got)
|
||||
}
|
||||
|
||||
if err := ps.LoadSource("probe",
|
||||
"local P={}\nP.name='probe'\nP.version='2'\nreturn P\n"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got := len(ps.List()); got != 1 {
|
||||
t.Errorf("after reload: %d records, want 1 — the old one was kept", got)
|
||||
}
|
||||
p := ps.find("probe")
|
||||
if p == nil {
|
||||
t.Fatal("plugin gone after reload")
|
||||
}
|
||||
if p.Info.Version != "2" {
|
||||
t.Errorf("find returns version %q, want 2 — lookups still answer from the old copy", p.Info.Version)
|
||||
}
|
||||
}
|
||||
@ -32,9 +32,9 @@ package lua
|
||||
// less featureful, so the split is explicit and documented.
|
||||
|
||||
import (
|
||||
"log"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sort"
|
||||
@ -265,6 +265,10 @@ type Plugins struct {
|
||||
// visibility holds the operator's per-plugin audience overrides, keyed by
|
||||
// plugin name. Consulted by UIForRole and the call endpoint's authorisation.
|
||||
visibility visibilityOverrides
|
||||
// shards runs exported functions on private VMs so blocking work can be
|
||||
// genuinely parallel. See plugin_shard.go for why the shared VM's lock makes
|
||||
// concurrency impossible and what a shard must therefore never touch.
|
||||
shards *shardPool
|
||||
// stateDirDisabled turns persistence off. Used by tests that assert state
|
||||
// starts empty, and by an embedder that has no writable plugin dir.
|
||||
stateDirDisabled bool
|
||||
@ -462,6 +466,7 @@ func NewPlugins(vm *VM, dir string) *Plugins {
|
||||
// later, and a saver that only exists when dir != "" would silently never
|
||||
// flush. markDirtyLocked and flush both no-op when the dir is empty.
|
||||
ps.save = newStateSaver(ps)
|
||||
ps.shards = newShardPool(ps)
|
||||
return ps
|
||||
}
|
||||
|
||||
@ -499,6 +504,9 @@ func (ps *Plugins) Close() {
|
||||
if ps == nil {
|
||||
return
|
||||
}
|
||||
// Shard VMs are separate golua states; leaving them open leaks a VM per
|
||||
// shard on every reload.
|
||||
ps.shards.Shutdown()
|
||||
ps.save.Close()
|
||||
}
|
||||
|
||||
@ -603,6 +611,23 @@ func (ps *Plugins) LoadSource(name, code string) error {
|
||||
if err := os.MkdirAll(ps.dir, 0755); err != nil && ps.dir != "" {
|
||||
return err
|
||||
}
|
||||
// Reloading must REPLACE, not append.
|
||||
//
|
||||
// LoadSource was reachable both from LoadDir and from the install/reload API,
|
||||
// and it appended unconditionally. Loading a plugin that was already loaded
|
||||
// therefore left two records with the same name, and findLocked — which
|
||||
// returns the FIRST match — kept handing out the OLD one. Measured: after
|
||||
// LoadDir + LoadSource of an edited plugin, /api/plugins listed both
|
||||
// versions and Call answered from version 1 while version 2 sat inert
|
||||
// alongside it. Editing a plugin and saving appeared to do nothing.
|
||||
//
|
||||
// Unload is a no-op when nothing is loaded under that name, so this is safe
|
||||
// on the first load.
|
||||
if ps.find(name) != nil {
|
||||
if err := ps.Unload(name); err != nil {
|
||||
return fmt.Errorf("replace %s: %w", name, err)
|
||||
}
|
||||
}
|
||||
p := &Plugin{
|
||||
Info: PluginInfo{Name: name},
|
||||
Hooks: map[Stage]string{},
|
||||
@ -1935,6 +1960,17 @@ func (ps *Plugins) invoke(p *Plugin, fn string, payload map[string]interface{})
|
||||
// quota report is a nested structure with numbers and booleans that must not be
|
||||
// flattened. Hook results still go through invoke, whose map-typed contract the
|
||||
// state merge depends on.
|
||||
// Call invokes an exported function on a plugin's SHARED VM, serialised by the
|
||||
// plugin's mutex.
|
||||
//
|
||||
// The lock is what makes one Lua VM safe to share, and it is held for the whole
|
||||
// call — including any network wait inside it. That is correct for a hook and
|
||||
// wrong for a function that blocks: twenty concurrent quota probes serialise
|
||||
// into the sum of their timeouts (measured: 29s on a 20-source gateway, against
|
||||
// 1.4–3.3s for a single probe with no contention).
|
||||
//
|
||||
// callOn below is the same work without the shared-VM assumption, so a caller
|
||||
// that owns a private VM can use it directly.
|
||||
func (ps *Plugins) Call(name, fn string, args map[string]interface{}) (interface{}, error) {
|
||||
ps.mu.RLock()
|
||||
var p *Plugin
|
||||
@ -1952,12 +1988,27 @@ func (ps *Plugins) Call(name, fn string, args map[string]interface{}) (interface
|
||||
return nil, fmt.Errorf("plugin %s does not export %q", name, fn)
|
||||
}
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
w := p.state
|
||||
if w == nil {
|
||||
p.mu.Unlock()
|
||||
return nil, fmt.Errorf("no state for plugin %s", name)
|
||||
}
|
||||
defer p.mu.Unlock()
|
||||
return callOn(w, fn, args)
|
||||
}
|
||||
|
||||
// callOn runs one exported function on a specific VM.
|
||||
//
|
||||
// It does NOT check the export allowlist and does NOT persist state: both belong
|
||||
// to the shared-VM path in Call. A shard VM has no plugin identity to check
|
||||
// against and no state file of its own — the export allowlist in Call is the
|
||||
// security boundary, and calling this with a function the plugin does not export
|
||||
// is a programming error, not an attack.
|
||||
//
|
||||
// The caller must own w exclusively. No lock is taken here.
|
||||
func callOn(w *worker, fn string, args map[string]interface{}) (interface{}, error) {
|
||||
if w == nil || w.L == nil {
|
||||
return nil, fmt.Errorf("no lua state")
|
||||
}
|
||||
L := w.L
|
||||
L.SetTop(0)
|
||||
defer L.SetTop(0)
|
||||
@ -1990,14 +2041,14 @@ func (ps *Plugins) Call(name, fn string, args map[string]interface{}) (interface
|
||||
L.Pop(1)
|
||||
}
|
||||
if errMsg != "" {
|
||||
return nil, fmt.Errorf("plugin %s: %s", name, errMsg)
|
||||
return nil, fmt.Errorf("plugin: %s", errMsg)
|
||||
}
|
||||
if L.GetTop() < 1 || L.IsNil(-1) {
|
||||
return nil, nil
|
||||
}
|
||||
var out interface{}
|
||||
if err := luaToJSON(L, -1, &out); err != nil {
|
||||
return nil, nil
|
||||
return nil, err
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user