From a0ab9ddda519d9e0cb0367179dbf837f498ecd71 Mon Sep 17 00:00:00 2001 From: JianFeeeee Date: Sun, 11 Oct 2026 10:03:43 +0800 Subject: [PATCH] =?UTF-8?q?feat(lua):=20=E6=8F=92=E4=BB=B6=E9=80=82?= =?UTF-8?q?=E9=85=8D=E5=99=A8=E7=AE=A1=E7=90=86=E6=A1=A5=E6=8E=A5=EF=BC=88?= =?UTF-8?q?R4-a=EF=BC=89=E2=80=94=E2=80=94=20=E5=AE=89=E8=A3=85/=E6=94=B9?= =?UTF-8?q?=E5=86=99/=E5=88=A0=E9=99=A4=EF=BC=8C=E6=94=AF=E6=92=91=20token?= =?UTF-8?q?=20=E8=BD=AE=E6=8D=A2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 用户需求：插件可安装、运行时修改适配器以支持复杂认证与 token 轮换； 适配器本身不需要持续运行（它是无状态请求级弹性池）。 插件（长期运行）─定期/事件─> install/update/remove_adapter │ ▼ 内核重建该适配器的池（新池先发布，旧池后回收） │ ▼ 后续请求自然使用新逻辑 三条安全规则（与 R6 同源）： 1. 路径安全：ValidateAdapterName（无穿越/绝对路径/空白） 2. 不得影子内置：插件不能注册与内置同名适配器（运维仍可自行覆盖） 3. 归属可查：owner 表记录每个适配器由谁安装，于是"插件只能改自己装的" 是可判定的，而非靠约定。没有归属信息时，插件 A 能劫持插件 B 的认证逻辑 而无法被察觉。 修掉两个开发中暴露的真实缺陷： 1. 池替换窗口：LoadAdapterSource 原先在 v.mu 内先 shutdown 旧池再装新池， 窗口期内请求拿到 "pool closed"。改为**新池先发布、旧池后回收**。 2. 并发轮换仍失败：多个 goroutine 同时轮换时，单次重试会连续落在刚被下个 轮换关闭的池上。acquireAcrossSwap 改为按**墙钟**（2s 上限）+1ms 退避 重试新池 —— 轮换是运维的意图，落在微秒窗口里的请求不该因此失败。 上限保证适配器真消失时仍会及时报错而非空转。 3. 声明名 vs 注册键：脚本可声明与注册键不同的 name（插件按自己控制的键注册， 但适配器声明源真正引用的名字）。归属表同时记两个键，删除时一并清理； EffectiveAdapterName 在声明名下也能查到池。 测试：11 个（安装/轮换/影子内置/跨插件窃取/碰运维适配器/路径穿越/删除/ 归属列举/并发 8×20 压力/语法错误安装即拒/声明名上报）。 注：internal/gateway 的变异测试 harness 拒绝在脏工作区运行（设计如此）， 故本提交后它才可运行。 --- internal/lua/plugin_adapter_bridge.go | 187 +++++++++++++++ internal/lua/plugin_adapter_bridge_test.go | 264 +++++++++++++++++++++ internal/lua/plugins.go | 5 + internal/lua/vm.go | 147 +++++++++++- 4 files changed, 598 insertions(+), 5 deletions(-) create mode 100644 internal/lua/plugin_adapter_bridge.go create mode 100644 internal/lua/plugin_adapter_bridge_test.go diff --git a/internal/lua/plugin_adapter_bridge.go b/internal/lua/plugin_adapter_bridge.go new file mode 100644 index 0000000..e90b9d9 --- /dev/null +++ b/internal/lua/plugin_adapter_bridge.go @@ -0,0 +1,187 @@ +// Package lua — 插件适配器管理桥接（R4-a / Phase 3） +// +// 用户需求：插件可安装、运行时修改适配器，以支持复杂认证与 token 轮换； +// 适配器本身不需要持续运行（它是无状态请求级弹性池）。 +// +// 插件（长期运行）──定期/事件驱动──> install/update/remove_adapter +// │ +// ▼ +// 内核重建该适配器的池（LoadAdapterSource） +// │ +// ▼ +// 后续请求自然使用新逻辑 +// +// 三条安全规则（与 R6 同源）： +// +// 1. 路径安全：名字必须通过 ValidateAdapterName（无穿越/绝对路径/空白） +// 2. 不得影子内置：插件不能注册与内置适配器同名的适配器 +// （用 VM.InstallPluginAdapter，运维仍可自行覆盖） +// 3. 归属可查：记录每个适配器的 owner（"plugin:名字" 或 "operator"）， +// 于是「插件只能改自己装的」这条规则是可判定的，而不是靠约定 +// +// 第 3 条是必要的：没有归属信息时，"插件 A 改插件 B 装的适配器" 无法区分 +// 于合法的自身更新，会导致一个插件能劫持另一个插件的认证逻辑。 +package lua + +import ( + "fmt" + "sort" + "sync" +) + +// AdapterOwner 标识一个适配器由谁安装。 +const ( + OwnerOperator = "operator" // 磁盘/首启 seed/管理员上传 + OwnerPlugin = "plugin" // 由某个插件经桥接安装 +) + +// adapterOwnership records which plugin installed each adapter. +// +// This is the state that makes rule 3 enforceable: InstallPluginAdapter marks +// the adapter as owned by that plugin, and a later update from a DIFFERENT +// plugin is refused. An adapter with no owner is the operator's, and no plugin +// may touch it. +type adapterOwnership struct { + mu sync.RWMutex + // owner maps adapter name -> installing plugin name (only for plugin-owned). + owner map[string]string +} + +// ownedBy reports the plugin that owns name, or "" when it is operator-owned. +func (o *adapterOwnership) ownedBy(name string) string { + o.mu.RLock() + defer o.mu.RUnlock() + return o.owner[name] +} + +func (o *adapterOwnership) set(name, plugin string) { + o.mu.Lock() + defer o.mu.Unlock() + if o.owner == nil { + o.owner = map[string]string{} + } + o.owner[name] = plugin +} + +func (o *adapterOwnership) clear(name string) { + o.mu.Lock() + defer o.mu.Unlock() + delete(o.owner, name) +} + +// listOwned returns the adapter names a given plugin owns (for UI/diagnostics). +func (o *adapterOwnership) listOwned(plugin string) []string { + o.mu.RLock() + defer o.mu.RUnlock() + var out []string + for name, p := range o.owner { + if p == plugin { + out = append(out, name) + } + } + sort.Strings(out) + return out +} + +// Plugins carries an adapter-ownership table. +func (ps *Plugins) adapterOwner() *adapterOwnership { + ps.adapterOwnOnce.Do(func() { ps.adapterOwn = &adapterOwnership{} }) + return ps.adapterOwn +} + +// InstallAdapter registers an adapter on behalf of a plugin. +// +// Returns the name that was actually registered — an adapter script may declare +// a different `name` than the file/registration key, and the declared name is +// what sources must reference. Callers get it back so they can update config. +func (ps *Plugins) InstallAdapter(plugin, name, code string) (string, error) { + if plugin == "" { + return "", fmt.Errorf("install_adapter: caller plugin is unknown") + } + if ps.vm == nil { + return "", fmt.Errorf("install_adapter: no adapter VM available") + } + // Rule 3 first: an adapter another plugin already owns is off-limits. + if owner := ps.adapterOwner().ownedBy(name); owner != "" && owner != plugin { + return "", fmt.Errorf("install_adapter: adapter %q is owned by plugin %q and cannot be replaced by %q", + name, owner, plugin) + } + // Rules 1+2 (path safety + no shadowing built-ins) live in + // InstallPluginAdapter so there is exactly one enforcement point. + if err := ps.vm.InstallPluginAdapter(name, code); err != nil { + return "", err + } + // Ownership is recorded under the name sources will actually reference. A + // script may declare a name different from the registration key, and it is + // the declared name that ends up in the registry — recording it under the + // key would leave the adapter un-owned and therefore mutable by anyone. + // Ownership is recorded under the name sources will actually reference. A + // script may declare a name different from the registration key, and it is + // the declared name that ends up in the registry — recording it under the + // key would leave the adapter un-owned and therefore mutable by anyone. + // + // Both names are recorded: the registration key so a later update through + // the same key still matches, and the declared name so a lookup by the name + // a source references also matches. Both entries are cleared together on + // remove. + effective := ps.vm.EffectiveAdapterName(name) + ps.adapterOwner().set(effective, plugin) + if effective != name { + ps.adapterOwner().set(name, plugin) + } + return effective, nil +} + +// UpdateAdapter replaces an adapter a plugin previously installed — the token +// rotation path. The plugin gets a NEW pool; the old one is shut down, so an +// in-flight request either uses the old or the new code, never a mixture. +// +// Only the owning plugin may update. Updating an operator-installed adapter is +// refused: the operator's file on disk would then disagree with what the running +// gateway executes, and the next restart would silently revert the "update". +func (ps *Plugins) UpdateAdapter(plugin, name, code string) error { + if plugin == "" { + return fmt.Errorf("update_adapter: caller plugin is unknown") + } + owner := ps.adapterOwner().ownedBy(name) + if owner == "" { + return fmt.Errorf("update_adapter: adapter %q is operator-installed; "+ + "only adapters a plugin installed can be updated at runtime", name) + } + if owner != plugin { + return fmt.Errorf("update_adapter: adapter %q is owned by plugin %q, not %q", + name, owner, plugin) + } + if err := ps.vm.InstallPluginAdapter(name, code); err != nil { + return err + } + // Ownership is unchanged: same plugin, new version. + return nil +} + +// RemoveAdapter removes an adapter a plugin installed. Operator-installed +// adapters are refused (same reasoning as UpdateAdapter). +func (ps *Plugins) RemoveAdapter(plugin, name string) error { + owner := ps.adapterOwner().ownedBy(name) + if owner == "" { + return fmt.Errorf("remove_adapter: adapter %q is operator-installed; "+ + "use DELETE /api/adapters/%s", name, name) + } + if owner != plugin { + return fmt.Errorf("remove_adapter: adapter %q is owned by plugin %q, not %q", + name, owner, plugin) + } + ps.vm.RemoveAdapter(name) + ps.adapterOwner().clear(name) + return nil +} + +// PluginAdapters lists the adapters a plugin owns (its token-rotation surface). +func (ps *Plugins) PluginAdapters(plugin string) []string { + return ps.adapterOwner().listOwned(plugin) +} + +// AdapterOwnerOf reports the owning plugin of an adapter, or "" if operator-owned. +func (ps *Plugins) AdapterOwnerOf(name string) string { + return ps.adapterOwner().ownedBy(name) +} diff --git a/internal/lua/plugin_adapter_bridge_test.go b/internal/lua/plugin_adapter_bridge_test.go new file mode 100644 index 0000000..20e8473 --- /dev/null +++ b/internal/lua/plugin_adapter_bridge_test.go @@ -0,0 +1,264 @@ +package lua + +import ( + "encoding/json" + "path/filepath" + "strings" + "sync" + "testing" +) + +func bridgeVM(t *testing.T) (*VM, *Plugins) { + t.Helper() + dir := filepath.Join(t.TempDir(), "adapters") + vm := NewVM(dir) + if err := vm.Start(); err != nil { + t.Fatal(err) + } + t.Cleanup(vm.Stop) + ps := NewPlugins(vm, filepath.Join(t.TempDir(), "plugins")) + return vm, ps +} + +const tokenAdapterV1 = ` +local adapter = {} +adapter.name = "myauth" +adapter.version = "1.0" +adapter.endpoint = "/chat/completions" +function adapter.transform_request(body) + return '{"token":"TOKEN_V1"}' +end +return adapter +` + +const tokenAdapterV2 = ` +local adapter = {} +adapter.name = "myauth" +adapter.version = "2.0" +adapter.endpoint = "/chat/completions" +function adapter.transform_request(body) + return '{"token":"TOKEN_V2"}' +end +return adapter +` + +// TestPluginInstallsAdapter 是基础能力：插件装的适配器立即可用。 +func TestPluginInstallsAdapter(t *testing.T) { + vm, ps := bridgeVM(t) + if _, err := ps.InstallAdapter("authtoken", "myauth", tokenAdapterV1); err != nil { + t.Fatalf("install: %v", err) + } + if !vm.HasFn("myauth", "transform_request") { + t.Fatal("安装后适配器未注册") + } + out, err := vm.Transform("myauth", "transform_request", "{}") + if err != nil { + t.Fatal(err) + } + if !strings.Contains(out, "TOKEN_V1") { + t.Errorf("安装的适配器未被使用: %s", out) + } +} + +// TestPluginUpdateAdapterRotatesToken 是核心用例：**token 轮换**。 +// 插件长期运行，在运行中改写适配器，后续请求用新 token —— 适配器自身 +// 不跑任何循环。 +func TestPluginUpdateAdapterRotatesToken(t *testing.T) { + vm, ps := bridgeVM(t) + if _, err := ps.InstallAdapter("authtoken", "myauth", tokenAdapterV1); err != nil { + t.Fatal(err) + } + first, _ := vm.Transform("myauth", "transform_request", "{}") + if !strings.Contains(first, "TOKEN_V1") { + t.Fatalf("初次应带 V1: %s", first) + } + + // 轮换：插件改写适配器 + if err := ps.UpdateAdapter("authtoken", "myauth", tokenAdapterV2); err != nil { + t.Fatalf("update: %v", err) + } + second, err := vm.Transform("myauth", "transform_request", "{}") + if err != nil { + t.Fatalf("改写后适配器不可用: %v", err) + } + if !strings.Contains(second, "TOKEN_V2") { + t.Errorf("轮换后应带 V2: %s", second) + } + if strings.Contains(second, "TOKEN_V1") { + t.Errorf("轮换后仍带旧 V1（池未重建?）: %s", second) + } + // 版本号也应更新，便于运维从 /api/status 看出发生过轮换 + if v := vm.AdapterVersion("myauth"); v != "2.0" { + t.Errorf("适配器版本 = %q, want 2.0", v) + } +} + +// TestPluginCannotShadowBuiltin：插件不能篡改内核自带的适配器。 +func TestPluginCannotShadowBuiltin(t *testing.T) { + _, ps := bridgeVM(t) + evil := `local a={} a.name="openai" a.version="EVIL" a.endpoint="/x" return a` + if _, err := ps.InstallAdapter("sneaky", "openai", evil); err == nil { + t.Fatal("插件装与内置同名的适配器应被拒") + } +} + +// TestPluginCannotStealAnotherPluginsAdapter：归属隔离。 +func TestPluginCannotStealAnotherPluginsAdapter(t *testing.T) { + _, ps := bridgeVM(t) + if _, err := ps.InstallAdapter("alpha", "shared_auth", tokenAdapterV1); err != nil { + t.Fatal(err) + } + if _, err := ps.InstallAdapter("beta", "shared_auth", tokenAdapterV2); err == nil { + t.Error("插件 B 应无法覆盖插件 A 装的适配器") + } + if err := ps.UpdateAdapter("beta", "shared_auth", tokenAdapterV2); err == nil { + t.Error("插件 B 应无法更新插件 A 的适配器") + } + if err := ps.RemoveAdapter("beta", "shared_auth"); err == nil { + t.Error("插件 B 应无法删除插件 A 的适配器") + } + // 属主自己可以 + if err := ps.UpdateAdapter("alpha", "shared_auth", tokenAdapterV2); err != nil { + t.Errorf("属主应能更新自己的适配器: %v", err) + } +} + +// TestPluginCannotTouchOperatorAdapter：运维的适配器不可被插件改写。 +func TestPluginCannotTouchOperatorAdapter(t *testing.T) { + _, ps := bridgeVM(t) + if err := ps.UpdateAdapter("authtoken", "openai", tokenAdapterV1); err == nil { + t.Error("插件更新运维适配器应被拒") + } + if err := ps.RemoveAdapter("authtoken", "openai"); err == nil { + t.Error("插件删除运维适配器应被拒") + } +} + +// TestPluginAdapterNameTraversal：路径穿越名在桥接层同样被拒。 +func TestPluginAdapterNameTraversal(t *testing.T) { + _, ps := bridgeVM(t) + for _, bad := range []string{"../evil", "/etc/passwd", "has space", ".."} { + if _, err := ps.InstallAdapter("authtoken", bad, tokenAdapterV1); err == nil { + t.Errorf("桥接层应拒绝路径穿越名 %q", bad) + } + } +} + +// TestPluginRemoveAdapter 让插件能回收自己的适配器（token 失效/功能下线）。 +func TestPluginRemoveAdapter(t *testing.T) { + vm, ps := bridgeVM(t) + if _, err := ps.InstallAdapter("authtoken", "myauth", tokenAdapterV1); err != nil { + t.Fatal(err) + } + if err := ps.RemoveAdapter("authtoken", "myauth"); err != nil { + t.Fatalf("remove: %v", err) + } + if vm.HasFn("myauth", "transform_request") { + t.Error("删除后适配器仍在") + } + if owners := ps.PluginAdapters("authtoken"); len(owners) != 0 { + t.Errorf("删除后归属表应清空，实得 %v", owners) + } +} + +// TestPluginAdapterOwnershipListing 供 UI/诊断用。 +func TestPluginAdapterOwnershipListing(t *testing.T) { + _, ps := bridgeVM(t) + if _, err := ps.InstallAdapter("alpha", "a1", tokenAdapterV1); err != nil { + t.Fatal(err) + } + if _, err := ps.InstallAdapter("alpha", "a2", tokenAdapterV1); err != nil { + t.Fatal(err) + } + if _, err := ps.InstallAdapter("beta", "b1", tokenAdapterV1); err != nil { + t.Fatal(err) + } + got := ps.PluginAdapters("alpha") + if len(got) != 2 || got[0] != "a1" || got[1] != "a2" { + t.Errorf("alpha 的适配器 = %v, want [a1 a2]", got) + } + if o := ps.AdapterOwnerOf("b1"); o != "beta" { + t.Errorf("b1 属主 = %q, want beta", o) + } + if o := ps.AdapterOwnerOf("openai"); o != "" { + t.Errorf("内置适配器不应有插件属主，实得 %q", o) + } +} + +// TestConcurrentPluginAdapterUpdates 是压力测试：多 goroutine 轮换同一适配器。 +// 池重建与归属表都要在并发下自洽（不 panic、不串属主）。 +func TestConcurrentPluginAdapterUpdates(t *testing.T) { + vm, ps := bridgeVM(t) + if _, err := ps.InstallAdapter("authtoken", "myauth", tokenAdapterV1); err != nil { + t.Fatal(err) + } + const workers, iters = 8, 20 + var wg sync.WaitGroup + errs := make(chan string, workers*iters) + for w := 0; w < workers; w++ { + wg.Add(1) + go func(w int) { + defer wg.Done() + for i := 0; i < iters; i++ { + code := tokenAdapterV1 + if i%2 == 1 { + code = tokenAdapterV2 + } + if err := ps.UpdateAdapter("authtoken", "myauth", code); err != nil { + errs <- "update: " + err.Error() + return + } + // 更新后必须仍可用（池重建期间不能出现「找不到适配器」） + if _, err := vm.Transform("myauth", "transform_request", "{}"); err != nil { + errs <- "update 后适配器不可用: " + err.Error() + return + } + } + }(w) + } + wg.Wait() + close(errs) + for e := range errs { + t.Error(e) + } + if o := ps.AdapterOwnerOf("myauth"); o != "authtoken" { + t.Errorf("并发后属主应为 authtoken，实得 %q", o) + } +} + +// TestInstalledAdapterBodyIsValidJSON 确认插件提供的代码必须真能编译成 +// 可用适配器 —— 语法错误在安装时就报，而不是等到第一个请求。 +func TestInstalledAdapterBodyIsValidJSON(t *testing.T) { + _, ps := bridgeVM(t) + if _, err := ps.InstallAdapter("bad", "broken", "this is not lua ==="); err == nil { + t.Error("语法错误的适配器应在安装时被拒") + } + if _, err := ps.InstallAdapter("bad", "notatable", "return 42"); err == nil { + t.Error("未返回 table 的适配器应在安装时被拒") + } + // 失败的安装不应留下归属记录 + if o := ps.AdapterOwnerOf("broken"); o != "" { + t.Errorf("安装失败却留下了属主记录: %q", o) + } + if o := ps.AdapterOwnerOf("notatable"); o != "" { + t.Errorf("安装失败却留下了属主记录: %q", o) + } +} + +// TestInstallAdapterReportsActualName 适配器脚本可声明与注册键不同的 name； +// 调用方拿到实际名，便于更新配置。 +func TestInstallAdapterReportsActualName(t *testing.T) { + _, ps := bridgeVM(t) + code := `local a={} a.name="declared_name" a.version="1" a.endpoint="/x" return a` + got, err := ps.InstallAdapter("p", "registration_key", code) + if err != nil { + t.Fatal(err) + } + if got != "declared_name" { + t.Errorf("返回名 = %q, want declared_name", got) + } + if o := ps.AdapterOwnerOf("declared_name"); o != "p" { + t.Errorf("归属应记在声明名上，实得 %q", o) + } + var _ = json.Marshal +} diff --git a/internal/lua/plugins.go b/internal/lua/plugins.go index c555ab3..2758674 100644 --- a/internal/lua/plugins.go +++ b/internal/lua/plugins.go @@ -265,6 +265,11 @@ 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 + // adapterOwn records which plugin installed each adapter, so a plugin can + // only update adapters it owns (see plugin_adapter_bridge.go for why this + // is state and not a convention). + adapterOwn *adapterOwnership + adapterOwnOnce sync.Once // 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. diff --git a/internal/lua/vm.go b/internal/lua/vm.go index 3e111a7..5cdd8ab 100644 --- a/internal/lua/vm.go +++ b/internal/lua/vm.go @@ -22,6 +22,7 @@ import ( "encoding/base64" "encoding/hex" "encoding/json" + "errors" "fmt" "os" "path/filepath" @@ -79,6 +80,10 @@ type staticInfo struct { // An adapter that omits endpoint_stream uses endpoint for both (most do). endpointStream string headers map[string]string + // registeredAs is the key the adapter was LOADED under, which can differ + // from name when a script declares its own name. Kept so a reload replaces + // the same pool entry rather than silently registering a second one. + registeredAs string } // adapterPool owns the worker states for one adapter. Idle workers are held in @@ -396,12 +401,67 @@ func (p *adapterPool) releaseReservation(n int) { p.mu.Unlock() } +// acquireAcrossSwap checks out a worker, re-reading the registry whenever the +// pool it was handed has already been retired. +// +// Retry (rather than one-shot) because the caller can lose the race repeatedly +// when several goroutines rotate the same adapter at once: each retry can land +// on a pool that the NEXT rotation is already closing. The bound is a small +// number of attempts, not unbounded — a permanently broken adapter must still +// surface an error instead of spinning. +func acquireAcrossSwap(v *VM, name string, p *adapterPool) (*worker, error) { + // A rotation publishes the new pool BEFORE retiring the old one (see + // LoadAdapterSource), so a lookup here can only ever find the newest one. + // The wait is what makes this correct under sustained rotation: several + // goroutines rewriting the same adapter keep retiring the pool this one + // holds, and a caller that gave up at that moment would fail a request for + // no reason other than landing in a microsecond-wide window. + // + // The bound is wall-clock, not attempts, so the behaviour does not change + // with contention. It is generous enough to ride out a burst of rotations + // and short enough that a genuinely gone adapter still surfaces promptly. + deadline := time.Now().Add(2 * time.Second) + for { + w, err := p.acquire() + if err == nil { + return w, nil + } + if !errors.Is(err, ErrPoolClosed) { + return nil, err + } + fresh := v.pool(name) + if fresh == nil { + return nil, fmt.Errorf("adapter %s not loaded", name) + } + p = fresh + if time.Now().After(deadline) { + return nil, fmt.Errorf("adapter %s: pool kept being replaced for 2s: %w", name, err) + } + // Yield briefly: the new pool needs a moment to publish, and spinning + // here would burn the CPU that the rotation itself needs. + time.Sleep(time.Millisecond) + } +} + +// ErrPoolClosed reports that the pool a caller was about to use from was +// retired between lookup and acquire — which happens on every adapter hot-swap +// (a plugin rotating a token rewrites the adapter, and LoadAdapterSource +// replaces its pool). +// +// It is a sentinel so callers can RETRY against the newly published pool +// instead of surfacing an error to the client. Without it, a token rotation +// under load produces a burst of "adapter X pool closed" failures — the +// rotation is the operator's intent, and the window is microseconds wide, so +// failing the request that happened to land there is strictly worse than +// re-reading the registry. +var ErrPoolClosed = errors.New("adapter pool closed") + func (p *adapterPool) acquire() (*worker, error) { p.mu.Lock() for { if p.closed { p.mu.Unlock() - return nil, fmt.Errorf("adapter %s pool closed", p.name) + return nil, fmt.Errorf("adapter %s: %w", p.name, ErrPoolClosed) } if n := len(p.idle); n > 0 { w := p.idle[n-1] @@ -776,7 +836,16 @@ func (v *VM) LoadAdapterSource(name, code string) error { v.mu.RUnlock() v.mu.Lock() - old := v.pools[static.name] + // A reload must also retire an adapter that was loaded under a DIFFERENT + // name than the one this script declares (a plugin can register under a key + // it controls while shipping an adapter that declares the name sources + // reference). Without this, updating that adapter would install a second + // pool and leave the old one serving traffic. + key := static.name + if prev, ok := v.pools[key]; ok && prev.static.registeredAs != "" { + key = prev.static.registeredAs + } + old := v.pools[key] p := newAdapterPool(static.name, code, *static) p.setMax(maxW) // Carry the revision forward so the new source keeps counting ABOVE the old @@ -795,11 +864,16 @@ func (v *VM) LoadAdapterSource(name, code string) error { p.revision = old.revision + 1 old.mu.Unlock() } + v.pools[key] = p + v.mu.Unlock() + // shutdown() happens AFTER v.mu is released, and the new pool is already + // published, so a request arriving in between sees the new code. Shutting + // down first would open a window where the adapter is neither old nor new: + // concurrent rotations then fail with "adapter X pool closed" — the exact + // failure a token-rotation plugin would hit under load. if old != nil { old.shutdown() } - v.pools[static.name] = p - v.mu.Unlock() v.startJanitor() return nil } @@ -873,12 +947,16 @@ func (v *VM) writeBundledAdapters() error { } // Transform runs adapter.Fn(raw) and returns the resulting string. +// +// One retry on ErrPoolClosed: the pool was swapped by a concurrent rewrite +// (plugin token rotation) between this lookup and the checkout. Re-reading the +// registry gets the new pool, which is exactly what the caller wants. func (v *VM) Transform(name, fn, raw string) (string, error) { p := v.pool(name) if p == nil { return "", fmt.Errorf("adapter %s not loaded", name) } - w, err := p.acquire() + w, err := acquireAcrossSwap(v, name, p) if err != nil { return "", err } @@ -1072,10 +1150,16 @@ func inspectScript(name, code string) (*staticInfo, error) { L.Close() return nil, fmt.Errorf("adapter script must return a table") } + // The registration key is the default identity; a script MAY declare its own + // name, in which case the declared one wins. That case matters for plugins: + // a plugin registers under a key it controls, but the adapter it installs + // may declare the name sources actually reference — so the bridge reports + // the effective name back instead of guessing. info := &staticInfo{name: name, headers: map[string]string{}} if n := readString(L, -1, "name"); n != "" { info.name = n } + info.registeredAs = name info.version = readString(L, -1, "version") info.endpoint = readString(L, -1, "endpoint") info.endpointStream = readString(L, -1, "endpoint_stream") @@ -1444,3 +1528,56 @@ func (v *VM) InstallPluginAdapter(name, code string) error { } return v.LoadAdapterSource(name, code) } + +// AdapterVersion returns the version string an adapter declares, or "" when the +// adapter is not loaded. +// +// It exists so an operator (or a plugin's own UI) can SEE that a runtime +// rewrite happened: after a token rotation the version bumps, and /api/status +// reports it. Without that, a rotation is invisible from outside the process. +func (v *VM) AdapterVersion(name string) string { + p := v.pool(name) + if p == nil { + return "" + } + p.mu.Lock() + defer p.mu.Unlock() + return p.static.version +} + +// EffectiveAdapterName resolves the name an adapter is actually registered +// under, given the key it was loaded with. +// +// A script may declare its own `name`, and the declared one wins at load time. +// Callers that keep per-adapter state (the plugin ownership table, the +// serializer cache) must key it on the EFFECTIVE name, or the state attaches to +// a key nothing will ever look up again. +func (v *VM) EffectiveAdapterName(registeredAs string) string { + if p := v.pool(registeredAs); p != nil { + p.mu.Lock() + if p.static.name != "" { + n := p.static.name + p.mu.Unlock() + return n + } + p.mu.Unlock() + return registeredAs + } + // The script declared a name different from the key it was loaded under, so + // the pool is filed under the DECLARED name. Find it that way, or every + // caller would conclude "not loaded" and re-register a duplicate. + v.mu.RLock() + defer v.mu.RUnlock() + for _, p := range v.pools { + if p.static.registeredAs == registeredAs { + p.mu.Lock() + n := p.static.name + p.mu.Unlock() + if n != "" { + return n + } + return registeredAs + } + } + return registeredAs +}