mirror of
https://gitcode.com/JianFeeeee/ModelRouter.git
synced 2026-11-12 04:07:25 +00:00
feat(lua): 插件适配器管理桥接(R4-a)—— 安装/改写/删除,支撑 token 轮换
用户需求:插件可安装、运行时修改适配器以支持复杂认证与 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 拒绝在脏工作区运行(设计如此),
故本提交后它才可运行。
This commit is contained in:
@ -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
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user