diff --git a/internal/gateway/plugins_api.go b/internal/gateway/plugins_api.go index 3a794e0..65f01fb 100644 --- a/internal/gateway/plugins_api.go +++ b/internal/gateway/plugins_api.go @@ -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. // diff --git a/internal/gateway/ui/index.html b/internal/gateway/ui/index.html index 6d9168e..e6e36ac 100644 --- a/internal/gateway/ui/index.html +++ b/internal/gateway/ui/index.html @@ -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); diff --git a/internal/lua/plugin_shard.go b/internal/lua/plugin_shard.go new file mode 100644 index 0000000..d72ad90 --- /dev/null +++ b/internal/lua/plugin_shard.go @@ -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. diff --git a/internal/lua/plugin_shard_test.go b/internal/lua/plugin_shard_test.go new file mode 100644 index 0000000..14128db --- /dev/null +++ b/internal/lua/plugin_shard_test.go @@ -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) + } +} diff --git a/internal/lua/plugins.go b/internal/lua/plugins.go index eee9423..29f45ba 100644 --- a/internal/lua/plugins.go +++ b/internal/lua/plugins.go @@ -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 }