mirror of
https://gitcode.com/JianFeeeee/ModelRouter.git
synced 2026-11-07 09:47:23 +00:00
- 新增唯一公开接口表 `modelrouter`(sdk_version=1.0):http / secrets / kernel / adapters / plugin_dir。成员与内部桥是**同一对象**而非副本 —— 复制会让凭据保险库分叉(插件写 A、内核读 B)。 - `llmsproxy` 与 `__llmsproxy_*` 降为同对象兼容别名,只为已装插件继续可用。 仓库内 4 个插件(channels / billing / quota / usage)全部改用 `modelrouter.*`:此前它们直接依赖内核私有全局,内核无法在不炸插件的前提下 改实现(重命名、加守卫、换实现都算破坏性变更)。 - 插件不再由网关安装:`SeedBundled()` 变彻底 no-op,`core.New` 不再调用; `ReadBundledPlugin` 改为按源文件定位(脱离 cwd),只服务测试与开发工具。 - 桌面端删除 `BUILTIN_PLUGINS` 与 `syncBundledPluginsOnUpgrade`:升级不再 删除、改名或覆盖用户插件文件。插件是运维资产,不是内核资源。 - `Builtin` 不再靠"源码与仓库示例逐字节相同"推断,插件一律视为外部文件。 - 判据:SDK 存在/版本/形状、别名与内部桥同一张表、适配器 VM 拿不到 SDK、 经 SDK 真实写读凭据并驱动内核动作、GUI 升级不动插件文件、release 包不含插件。 说明:channels.lua / quota.lua 同时带有此前的渠道与额度改动(含 OpenCode 免费池 监控),与 SDK 改名同处一个文件,无法文件级拆分,故一并纳入。
1699 lines
53 KiB
Go
1699 lines
53 KiB
Go
// Package lua implements the adapter runtime: bundles/loads *.lua adapter
|
||
// scripts, exposes json/string helpers, and lets Go call the protocol
|
||
// transform functions plus a build_headers signature hook.
|
||
//
|
||
// The runtime uses LuaJIT (github.com/aarzilli/golua) so every adapter runs on
|
||
// an independent interpreter state. Because a lua.State is NOT goroutine-safe,
|
||
// each adapter owns a pool of workers (states) and a worker is checked out for
|
||
// the duration of a single hook call.
|
||
//
|
||
// Pools are ELASTIC: the sum of max_concurrent over the sources using an adapter
|
||
// is a ceiling, not a preallocation. States are booted on demand in batches
|
||
// sized by that ceiling, and reclaimed by a single background janitor at a rate
|
||
// scaled inversely to the adapter's live connection count. An idle gateway
|
||
// therefore holds close to zero Lua states regardless of how much aggregate
|
||
// concurrency is configured. See adapterPool for the sizing rules.
|
||
package lua
|
||
|
||
import (
|
||
"crypto/hmac"
|
||
"crypto/sha256"
|
||
"embed"
|
||
"encoding/base64"
|
||
"encoding/hex"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"os"
|
||
"path/filepath"
|
||
"runtime"
|
||
"sort"
|
||
"strconv"
|
||
"strings"
|
||
"sync"
|
||
"time"
|
||
|
||
// golua exposes the LuaJIT C API. Build with -tags luajit so the cgo
|
||
// LDFLAGS resolve to -lluajit-5.1 (see golua's lua.go).
|
||
golua "github.com/aarzilli/golua/lua"
|
||
)
|
||
|
||
//go:embed adapters/*.lua
|
||
var bundledAdapters embed.FS
|
||
|
||
// adapterGlobal is the reserved global holding the adapter table after the
|
||
// worker boots the script. A reserved name avoids clashing with user code.
|
||
const adapterGlobal = "__llmsproxy_adapter"
|
||
|
||
type APIAdapter struct {
|
||
Name string `json:"name"`
|
||
Version string `json:"version"`
|
||
}
|
||
|
||
// worker wraps one LuaJIT interpreter state. Not goroutine-safe; each worker
|
||
// is used by exactly one goroutine at a time, checked out from its pool.
|
||
type worker struct {
|
||
L *golua.State
|
||
}
|
||
|
||
func (w *worker) close() {
|
||
if w.L != nil {
|
||
w.L.Close()
|
||
}
|
||
}
|
||
|
||
// staticInfo caches the immutable fields of an adapter script, extracted once
|
||
// at load time. Endpoint() and the static headers fallback read from here so
|
||
// they never churn a pooled worker.
|
||
type staticInfo struct {
|
||
name string
|
||
version string
|
||
endpoint string
|
||
// endpointStream is the streaming counterpart of endpoint. It exists so the
|
||
// KERNEL never has to know a vendor's streaming path: Gemini streams over the
|
||
// same path with a different verb (:generateContent -> :streamGenerateContent),
|
||
// and the verb is part of the path, so the two cannot both be static.
|
||
//
|
||
// Before this field the kernel carried `strings.Replace(ep,
|
||
// ":generateContent", ":streamGenerateContent", 1)` — vendor protocol
|
||
// knowledge inside the dispatcher. It now lives here, where it belongs:
|
||
// the adapter declares both endpoints and the kernel just picks one.
|
||
// 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
|
||
// ingress is the client-facing endpoint path this adapter CLAIMS to parse,
|
||
// e.g. "/v1/messages". It is the INBOUND dual of serialize_* (which is
|
||
// outbound): when an adapter declares it, the kernel YIELDS parsing of that
|
||
// endpoint's request body to the adapter's deserialize_request hook, instead
|
||
// of its own built-in parser. The kernel keeps a built-in parser as the
|
||
// fallback (R5), so an unconfigured deployment still serves /v1/messages —
|
||
// but once an adapter claims the endpoint, protocol iteration lives in the
|
||
// adapter, not the kernel. Empty means the adapter claims no ingress.
|
||
ingress string
|
||
// hidden marks an adapter that should NOT appear in the user-facing adapter
|
||
// list (GET /api/adapters → UI 的适配器页与源编辑器的适配器下拉)。隐藏的
|
||
// 适配器仍然正常加载、仍然可以被源引用——它只是不从选择列表里冒出来。
|
||
// 用途:厂商探针类/内部实验类适配器(如 opencodezen 的免费层门禁形态),
|
||
// 直接暴露会让运维误当成通用 OpenAI 适配器去选。声明方式:adapter.hidden = true。
|
||
hidden bool
|
||
}
|
||
|
||
// adapterPool owns the worker states for one adapter. Idle workers are held in
|
||
// a slice guarded by a cond; workers are created lazily on demand and reclaimed
|
||
// by the VM janitor once demand drops.
|
||
//
|
||
// Sizing is elastic rather than fixed (plan 阶段 2). Three inputs drive it:
|
||
//
|
||
// - maxW — the adapter's maximum concurrency (Σ max_concurrent of every source
|
||
// using it). It is the hard ceiling AND sets the growth step, so a hot,
|
||
// high-concurrency adapter ramps in batches while a low-concurrency one
|
||
// creeps up one state at a time.
|
||
// - created / len(idle) — how many states already exist, i.e. how much room is
|
||
// left to grow and how much is available to reclaim.
|
||
// - inUse — how many states are checked out right now, i.e. the live
|
||
// connection count for this adapter. It sets the shrink step: the fewer
|
||
// connections, the more aggressively idle states are released.
|
||
type adapterPool struct {
|
||
name string
|
||
script string
|
||
static staticInfo
|
||
|
||
mu sync.Mutex
|
||
cond *sync.Cond
|
||
idle []*worker
|
||
created int // live states: len(idle) + inUse (+ states being booted)
|
||
inUse int // states checked out right now == live connections
|
||
waiting int // goroutines blocked waiting for a free state
|
||
maxW int // hard ceiling = Σ max_concurrent of the sources using this adapter
|
||
used bool
|
||
closed bool
|
||
|
||
lastGrow time.Time
|
||
idleRounds int // consecutive janitor rounds that saw reclaimable slack
|
||
peakInUse int // high-water mark of inUse, for observability
|
||
// revision counts how many times this adapter's SOURCE was replaced (or the
|
||
// adapter was removed and re-added). Callers that cache a decision derived
|
||
// from the script's declared surface read it via AdapterRevision, so a
|
||
// runtime rewrite (a plugin installing a new version of its own adapter)
|
||
// cannot leave them acting on a stale answer.
|
||
revision uint64
|
||
// pluginMode makes boot() store the returned table under pluginGlobal
|
||
// instead of adapterGlobal. Everything else (elastic sizing, reclaim) is
|
||
// identical, which is why plugins reuse this pool rather than getting a
|
||
// second implementation.
|
||
pluginMode bool
|
||
}
|
||
|
||
const (
|
||
// growDivisor turns the adapter's max concurrency into the growth step CAP:
|
||
// step <= clamp(ceil(maxW / growDivisor), 1, growStepCap). The actual batch
|
||
// is additionally bounded by how many callers are queued (see growPlanLocked),
|
||
// so a wide adapter may ramp in big jumps but never warms states nobody is
|
||
// waiting for.
|
||
growDivisor = 8
|
||
growStepCap = 8
|
||
// growCooldown keeps a burst of misses from batching repeatedly while the
|
||
// previous batch is still booting.
|
||
growCooldown = time.Second
|
||
// maxPluginStates caps how many concurrent VM states ONE plugin may occupy.
|
||
// A plugin is third-party code on the request path, so its ceiling is much
|
||
// lower than an adapter's (which is sized from the sources'
|
||
// max_concurrent): a plugin hook is a short synchronous call, so a handful
|
||
// of states is already far more parallelism than any real hook needs, and a
|
||
// runaway plugin cannot balloon memory the way a per-source adapter pool can.
|
||
maxPluginStates = 4
|
||
// residentWorkers is how many states an adapter keeps warm once it has
|
||
// served at least one request. Booting is milliseconds, but keeping one warm
|
||
// removes that from the critical path of the next request. Adapters that
|
||
// were never used keep nothing.
|
||
residentWorkers = 1
|
||
// idleHeadroom is the spare idle state kept above the live connection count,
|
||
// so the next concurrent request does not have to boot.
|
||
idleHeadroom = 1
|
||
// shrinkInterval is how often the VM janitor reclaims idle states.
|
||
shrinkInterval = 30 * time.Second
|
||
// shrinkGraceRounds is how many consecutive janitor rounds must see slack
|
||
// before anything is released. It guards against tearing a pool down between
|
||
// two bursts — NOT against ordinary traffic: a pool serving one request at a
|
||
// time still has reclaimable slack, and requiring "no traffic at all" would
|
||
// pin a burst's leftover states forever on any busy gateway. The counter is
|
||
// therefore reset by GROWTH (real contention), not by a mere checkout.
|
||
shrinkGraceRounds = 2
|
||
)
|
||
|
||
// PoolStats is a snapshot of one adapter pool's elastic sizing, surfaced by the
|
||
// status API so the algorithm is observable instead of a black box.
|
||
type PoolStats struct {
|
||
Name string `json:"name"`
|
||
Created int `json:"created"`
|
||
Idle int `json:"idle"`
|
||
InUse int `json:"in_use"`
|
||
Waiting int `json:"waiting"`
|
||
Max int `json:"max"`
|
||
Resident int `json:"resident"`
|
||
GrowStep int `json:"grow_step"`
|
||
ShrinkStep int `json:"shrink_step"`
|
||
PeakInUse int `json:"peak_in_use"`
|
||
}
|
||
|
||
func newAdapterPool(name, script string, static staticInfo) *adapterPool {
|
||
p := &adapterPool{name: name, script: script, static: static, maxW: 1, revision: 1}
|
||
p.cond = sync.NewCond(&p.mu)
|
||
return p
|
||
}
|
||
|
||
// setMax updates the pool's hard ceiling (Σ max_concurrent of its sources).
|
||
// Lowering it does not kill live states: the janitor reclaims the excess.
|
||
func (p *adapterPool) setMax(n int) {
|
||
if n < 1 {
|
||
n = 1
|
||
}
|
||
p.mu.Lock()
|
||
p.maxW = n
|
||
p.cond.Broadcast()
|
||
p.mu.Unlock()
|
||
}
|
||
|
||
func ceilDiv(a, b int) int {
|
||
if b <= 0 {
|
||
return a
|
||
}
|
||
return (a + b - 1) / b
|
||
}
|
||
|
||
// growStepLocked is how many states to warm on a miss, derived from the
|
||
// adapter's maximum concurrency. Caller holds p.mu.
|
||
func (p *adapterPool) growStepLocked() int {
|
||
step := ceilDiv(p.maxW, growDivisor)
|
||
if step < 1 {
|
||
return 1
|
||
}
|
||
if step > growStepCap {
|
||
return growStepCap
|
||
}
|
||
return step
|
||
}
|
||
|
||
// residentLocked is the floor the janitor will not shrink below. Caller holds p.mu.
|
||
func (p *adapterPool) residentLocked() int {
|
||
if !p.used {
|
||
return 0
|
||
}
|
||
if residentWorkers > p.maxW {
|
||
return p.maxW
|
||
}
|
||
return residentWorkers
|
||
}
|
||
|
||
// shrinkStepLocked is how many idle states to release this round. The step is
|
||
// inversely proportional to the live connection count: with no connections the
|
||
// slack collapses in one round, while a busy adapter gives up one state at a
|
||
// time so the hot path keeps its warm states. Caller holds p.mu.
|
||
func (p *adapterPool) shrinkStepLocked() int {
|
||
excess := p.created - p.keepLocked()
|
||
if excess <= 0 {
|
||
return 0
|
||
}
|
||
step := ceilDiv(excess, 1+p.inUse)
|
||
if step > excess {
|
||
step = excess
|
||
}
|
||
if step > len(p.idle) {
|
||
step = len(p.idle)
|
||
}
|
||
if step < 0 {
|
||
return 0
|
||
}
|
||
return step
|
||
}
|
||
|
||
// keepLocked is the number of live states this pool should retain right now.
|
||
// Caller holds p.mu.
|
||
func (p *adapterPool) keepLocked() int {
|
||
keep := p.residentLocked()
|
||
if h := p.inUse + idleHeadroom; h > keep {
|
||
keep = h
|
||
}
|
||
if keep > p.maxW {
|
||
keep = p.maxW
|
||
}
|
||
return keep
|
||
}
|
||
|
||
func (p *adapterPool) stats() PoolStats {
|
||
p.mu.Lock()
|
||
defer p.mu.Unlock()
|
||
return PoolStats{
|
||
Name: p.name,
|
||
Created: p.created,
|
||
Idle: len(p.idle),
|
||
InUse: p.inUse,
|
||
Waiting: p.waiting,
|
||
Max: p.maxW,
|
||
Resident: p.residentLocked(),
|
||
GrowStep: p.growStepLocked(),
|
||
ShrinkStep: p.shrinkStepLocked(),
|
||
PeakInUse: p.peakInUse,
|
||
}
|
||
}
|
||
|
||
// boot creates a fresh LuaJIT state, registers the shared globals and executes
|
||
// the adapter script, storing the returned adapter table in the state.
|
||
func (p *adapterPool) boot() (*worker, error) {
|
||
L := golua.NewState()
|
||
L.OpenLibs()
|
||
setupGlobals(L)
|
||
if err := L.DoString(p.script); err != nil {
|
||
L.Close()
|
||
return nil, fmt.Errorf("compile adapter %s: %w", p.name, err)
|
||
}
|
||
if L.Type(-1) != golua.LUA_TTABLE {
|
||
L.Close()
|
||
return nil, fmt.Errorf("adapter %s must return a table", p.name)
|
||
}
|
||
// SetGlobal POPS the value off the stack, so it can only be called once per
|
||
// boot. Plugins therefore store under pluginGlobal only, and NOT under
|
||
// adapterGlobal: a second SetGlobal on the now-empty stack would assign nil.
|
||
if p.pluginMode {
|
||
L.SetGlobal(pluginGlobal)
|
||
} else {
|
||
L.SetGlobal(adapterGlobal)
|
||
}
|
||
L.SetTop(0)
|
||
return &worker{L: L}, nil
|
||
}
|
||
|
||
// growPlanLocked reserves capacity for a batch prewarm and returns how many
|
||
// EXTRA states to boot in the background (the caller already reserved one for
|
||
// itself).
|
||
//
|
||
// Two independent limits apply, and BOTH matter:
|
||
//
|
||
// - the adapter's max concurrency caps the step (growStepLocked), so a
|
||
// high-concurrency adapter is allowed to ramp in bigger jumps than a
|
||
// narrow one;
|
||
// - the number of goroutines actually BLOCKED waiting for a state bounds it
|
||
// to real demand.
|
||
//
|
||
// Sizing on the ceiling alone over-provisions badly: an adapter with
|
||
// max_concurrent=76 has a step of 8, so two concurrent requests would warm 8
|
||
// states, and the janitor would throw 7 of them away a minute later — boot,
|
||
// discard, repeat, with RSS oscillating for no benefit. Prewarming only for
|
||
// goroutines that are genuinely queued keeps the ramp cheap without the churn.
|
||
//
|
||
// Caller holds p.mu.
|
||
func (p *adapterPool) growPlanLocked(contended bool) int {
|
||
if !contended || p.waiting <= 0 {
|
||
return 0
|
||
}
|
||
if time.Since(p.lastGrow) < growCooldown {
|
||
return 0
|
||
}
|
||
extra := p.growStepLocked() - 1
|
||
if extra > p.waiting {
|
||
extra = p.waiting // never warm more than the queue needs
|
||
}
|
||
if extra <= 0 {
|
||
return 0
|
||
}
|
||
if room := p.maxW - p.created; extra > room {
|
||
extra = room
|
||
}
|
||
if extra <= 0 {
|
||
return 0
|
||
}
|
||
// Reserve immediately so concurrent acquires respect the ceiling while the
|
||
// batch is still booting.
|
||
p.created += extra
|
||
p.lastGrow = time.Now()
|
||
return extra
|
||
}
|
||
|
||
// prewarm boots n reserved states and publishes them as idle. A boot failure
|
||
// releases its reservation; it is not fatal, since the synchronous path
|
||
// surfaces adapter errors already.
|
||
func (p *adapterPool) prewarm(n int) {
|
||
for i := 0; i < n; i++ {
|
||
p.mu.Lock()
|
||
closed := p.closed
|
||
p.mu.Unlock()
|
||
if closed {
|
||
p.releaseReservation(n - i)
|
||
return
|
||
}
|
||
w, err := p.boot()
|
||
if err != nil {
|
||
p.releaseReservation(n - i)
|
||
return
|
||
}
|
||
p.mu.Lock()
|
||
if p.closed {
|
||
p.created--
|
||
p.mu.Unlock()
|
||
w.close()
|
||
p.releaseReservation(n - i - 1)
|
||
return
|
||
}
|
||
p.idle = append(p.idle, w)
|
||
p.cond.Signal()
|
||
p.mu.Unlock()
|
||
}
|
||
}
|
||
|
||
func (p *adapterPool) releaseReservation(n int) {
|
||
if n <= 0 {
|
||
return
|
||
}
|
||
p.mu.Lock()
|
||
p.created -= n
|
||
if p.created < 0 {
|
||
p.created = 0
|
||
}
|
||
p.cond.Broadcast()
|
||
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: %w", p.name, ErrPoolClosed)
|
||
}
|
||
if n := len(p.idle); n > 0 {
|
||
w := p.idle[n-1]
|
||
p.idle = p.idle[:n-1]
|
||
p.checkoutLocked()
|
||
p.mu.Unlock()
|
||
return w, nil
|
||
}
|
||
if p.created < p.maxW {
|
||
// Every existing state busy on a miss = real concurrency, not a
|
||
// sequential caller reusing one warm state.
|
||
contended := p.created > 0 && p.inUse >= p.created
|
||
p.created++
|
||
// Growing is the signal that capacity is genuinely short, so any
|
||
// pending shrink decision is stale.
|
||
p.idleRounds = 0
|
||
extra := p.growPlanLocked(contended)
|
||
p.mu.Unlock()
|
||
if extra > 0 {
|
||
go p.prewarm(extra)
|
||
}
|
||
w, err := p.boot()
|
||
if err != nil {
|
||
p.mu.Lock()
|
||
p.created--
|
||
p.cond.Signal()
|
||
p.mu.Unlock()
|
||
return nil, err
|
||
}
|
||
p.mu.Lock()
|
||
p.checkoutLocked()
|
||
p.mu.Unlock()
|
||
return w, nil
|
||
}
|
||
p.waiting++
|
||
p.cond.Wait()
|
||
p.waiting--
|
||
}
|
||
}
|
||
|
||
// checkoutLocked accounts one state as handed out. Caller holds p.mu.
|
||
func (p *adapterPool) checkoutLocked() {
|
||
p.inUse++
|
||
p.used = true
|
||
if p.inUse > p.peakInUse {
|
||
p.peakInUse = p.inUse
|
||
}
|
||
}
|
||
|
||
func (p *adapterPool) release(w *worker) {
|
||
w.L.SetTop(0)
|
||
p.mu.Lock()
|
||
if p.inUse > 0 {
|
||
p.inUse--
|
||
}
|
||
if p.closed {
|
||
p.created--
|
||
p.mu.Unlock()
|
||
w.close()
|
||
return
|
||
}
|
||
// A lowered ceiling is honoured on the spot rather than waiting for the
|
||
// janitor: returning a state the pool may no longer keep closes it.
|
||
if p.created > p.maxW {
|
||
p.created--
|
||
p.cond.Signal()
|
||
p.mu.Unlock()
|
||
w.close()
|
||
return
|
||
}
|
||
p.idle = append(p.idle, w)
|
||
p.cond.Signal()
|
||
p.mu.Unlock()
|
||
}
|
||
|
||
// reclaim releases idle states down to the current keep target, using a step
|
||
// scaled by the live connection count. It returns how many states were closed.
|
||
func (p *adapterPool) reclaim() int {
|
||
p.mu.Lock()
|
||
if p.closed {
|
||
p.mu.Unlock()
|
||
return 0
|
||
}
|
||
step := p.shrinkStepLocked()
|
||
if step <= 0 {
|
||
p.idleRounds = 0
|
||
p.mu.Unlock()
|
||
return 0
|
||
}
|
||
p.idleRounds++
|
||
if p.idleRounds < shrinkGraceRounds {
|
||
p.mu.Unlock()
|
||
return 0
|
||
}
|
||
victims := append([]*worker(nil), p.idle[len(p.idle)-step:]...)
|
||
p.idle = p.idle[:len(p.idle)-step]
|
||
p.created -= step
|
||
p.idleRounds = 0
|
||
p.mu.Unlock()
|
||
for _, w := range victims {
|
||
w.close()
|
||
}
|
||
return step
|
||
}
|
||
|
||
func (p *adapterPool) shutdown() {
|
||
p.mu.Lock()
|
||
p.closed = true
|
||
idle := p.idle
|
||
p.idle = nil
|
||
p.created -= len(idle)
|
||
p.cond.Broadcast()
|
||
p.mu.Unlock()
|
||
for _, w := range idle {
|
||
w.close()
|
||
}
|
||
}
|
||
|
||
// VM aggregates per-adapter worker pools. All exported methods are safe for
|
||
// concurrent use.
|
||
type VM struct {
|
||
mu sync.RWMutex
|
||
dir string
|
||
pools map[string]*adapterPool
|
||
|
||
// hiddenRefs 记录"被源引用的隐藏适配器"。hidden=true 的适配器默认不进
|
||
// ListAdapters,但被源引用时必须保持可见(源编辑器要能回显当前值)。
|
||
// 由内核在配置装载/源变更时调用 SetHiddenRefs 维护;nil = 未标记。
|
||
hiddenRefs map[string]bool
|
||
|
||
// janitor drives elastic shrinking. One goroutine serves every pool (not
|
||
// one per adapter), so an idle gateway costs a single sleeping goroutine.
|
||
janitorOnce sync.Once
|
||
janitorStop chan struct{}
|
||
janitorDone chan struct{}
|
||
}
|
||
|
||
// SetHiddenRefs 供内核同步"哪些隐藏适配器正被源引用"。传 nil/空集 = 没有引用。
|
||
func (v *VM) SetHiddenRefs(names []string) {
|
||
v.mu.Lock()
|
||
defer v.mu.Unlock()
|
||
if len(names) == 0 {
|
||
v.hiddenRefs = nil
|
||
return
|
||
}
|
||
m := make(map[string]bool, len(names))
|
||
for _, n := range names {
|
||
m[n] = true
|
||
}
|
||
v.hiddenRefs = m
|
||
}
|
||
|
||
func NewVM(dir string) *VM {
|
||
return &VM{
|
||
dir: dir,
|
||
pools: map[string]*adapterPool{},
|
||
janitorStop: make(chan struct{}),
|
||
janitorDone: make(chan struct{}),
|
||
}
|
||
}
|
||
|
||
// startJanitor launches the single shrink loop. Idempotent.
|
||
func (v *VM) startJanitor() {
|
||
v.janitorOnce.Do(func() {
|
||
go v.janitorLoop(shrinkInterval)
|
||
})
|
||
}
|
||
|
||
func (v *VM) janitorLoop(every time.Duration) {
|
||
defer close(v.janitorDone)
|
||
t := time.NewTicker(every)
|
||
defer t.Stop()
|
||
for {
|
||
select {
|
||
case <-v.janitorStop:
|
||
return
|
||
case <-t.C:
|
||
v.reclaimIdle()
|
||
}
|
||
}
|
||
}
|
||
|
||
// reclaimIdle runs one shrink round over every pool. Exported behaviour is
|
||
// tested through ReclaimIdleNow.
|
||
func (v *VM) reclaimIdle() int {
|
||
v.mu.RLock()
|
||
pools := make([]*adapterPool, 0, len(v.pools))
|
||
for _, p := range v.pools {
|
||
pools = append(pools, p)
|
||
}
|
||
v.mu.RUnlock()
|
||
n := 0
|
||
for _, p := range pools {
|
||
n += p.reclaim()
|
||
}
|
||
return n
|
||
}
|
||
|
||
// ReclaimIdleNow forces one shrink round and reports how many worker states
|
||
// were closed. Intended for tests and for an explicit "release idle memory"
|
||
// action; normal operation relies on the janitor.
|
||
func (v *VM) ReclaimIdleNow() int { return v.reclaimIdle() }
|
||
|
||
// PoolStats returns the elastic sizing snapshot of every loaded adapter,
|
||
// sorted by name.
|
||
func (v *VM) PoolStats() []PoolStats {
|
||
v.mu.RLock()
|
||
pools := make([]*adapterPool, 0, len(v.pools))
|
||
for _, p := range v.pools {
|
||
pools = append(pools, p)
|
||
}
|
||
v.mu.RUnlock()
|
||
out := make([]PoolStats, 0, len(pools))
|
||
for _, p := range pools {
|
||
out = append(out, p.stats())
|
||
}
|
||
sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name })
|
||
return out
|
||
}
|
||
|
||
func (v *VM) Start() error {
|
||
v.startJanitor()
|
||
if v.dir == "" {
|
||
return nil
|
||
}
|
||
// First-run seeding: a brand-new adapter dir is created and populated with
|
||
// the bundled adapters. If the dir already exists it is treated as
|
||
// authoritative and never rewritten — deleting a file there is a real delete.
|
||
firstRun := false
|
||
if _, err := os.Stat(v.dir); os.IsNotExist(err) {
|
||
firstRun = true
|
||
if err := os.MkdirAll(v.dir, 0755); err != nil {
|
||
return fmt.Errorf("mkdir adapter dir: %w", err)
|
||
}
|
||
}
|
||
if firstRun {
|
||
if err := v.writeBundledAdapters(); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
entries, err := os.ReadDir(v.dir)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
for _, e := range entries {
|
||
if filepath.Ext(e.Name()) != ".lua" {
|
||
continue
|
||
}
|
||
if err := v.LoadAdapter(filepath.Join(v.dir, e.Name())); err != nil {
|
||
fmt.Printf("[lua] preload %s: %v\n", e.Name(), err)
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// ReadBundledPlugin is retained only for source-level test fixtures and
|
||
// development tooling. It resolves relative to this source file, not the
|
||
// process working directory, and is never used by production plugin loading.
|
||
func ReadBundledPlugin(name string) (string, error) {
|
||
_, file, _, ok := runtime.Caller(0)
|
||
if !ok {
|
||
return "", fmt.Errorf("locate plugin source %s", name)
|
||
}
|
||
data, err := os.ReadFile(filepath.Join(filepath.Dir(file), "plugins", name+".lua"))
|
||
if err != nil {
|
||
return "", fmt.Errorf("plugin source %s: %w", name, err)
|
||
}
|
||
return string(data), nil
|
||
}
|
||
|
||
// writeBundledPlugins is intentionally a no-op: plugins are independent optional
|
||
// files and are never seeded by the gateway.
|
||
func writeBundledPlugins(string) error { return nil }
|
||
|
||
func (v *VM) Stop() {
|
||
select {
|
||
case <-v.janitorStop:
|
||
// already stopped
|
||
default:
|
||
close(v.janitorStop)
|
||
}
|
||
v.mu.Lock()
|
||
pools := v.pools
|
||
v.pools = map[string]*adapterPool{}
|
||
v.mu.Unlock()
|
||
for _, p := range pools {
|
||
p.shutdown()
|
||
}
|
||
}
|
||
|
||
// ConfigureConcurrency sets each adapter's worker CEILING to the sum of
|
||
// max_concurrent of every source using it. It is a ceiling, not a preallocation:
|
||
// states are booted on demand and reclaimed when demand drops, so a config with
|
||
// a large aggregate concurrency no longer implies a large resident pool.
|
||
// Values below one are clamped to 1.
|
||
func (v *VM) ConfigureConcurrency(adapterConcurrency map[string]int) {
|
||
v.mu.RLock()
|
||
defer v.mu.RUnlock()
|
||
for name, n := range adapterConcurrency {
|
||
if p, ok := v.pools[name]; ok {
|
||
p.setMax(n)
|
||
}
|
||
}
|
||
}
|
||
|
||
func (v *VM) ListAdapters() []APIAdapter {
|
||
v.mu.RLock()
|
||
defer v.mu.RUnlock()
|
||
// hidden=true 的适配器不进用户可见列表(仍正常加载、仍可被源引用)。
|
||
// 例外:**正在被某个源引用**的隐藏适配器必须仍出现在列表里——否则源编辑器
|
||
// 打开时找不到当前值,下拉会被重置成空,造成"保存后适配器悄悄丢了"。
|
||
// 引用关系由内核持有(core 在配置装载时标记),VM 只看这个位。
|
||
keepHidden := v.hiddenKeep()
|
||
list := make([]APIAdapter, 0, len(v.pools))
|
||
for name, p := range v.pools {
|
||
if p.static.hidden && !keepHidden[name] {
|
||
continue
|
||
}
|
||
list = append(list, APIAdapter{Name: name, Version: p.static.version})
|
||
}
|
||
sort.Slice(list, func(i, j int) bool { return list[i].Name < list[j].Name })
|
||
return list
|
||
}
|
||
|
||
// hiddenKeep returns the set of hidden adapter names that must STAY visible
|
||
// because at least one source references them. nil 表示内核未标记(测试环境/
|
||
// 未装载配置),此时隐藏适配器一律隐藏——与"新适配器默认隐藏"的预期一致。
|
||
func (v *VM) hiddenKeep() map[string]bool {
|
||
if v.hiddenRefs == nil {
|
||
return nil
|
||
}
|
||
return v.hiddenRefs
|
||
}
|
||
|
||
// AdapterForIngress returns the registered name of the adapter that CLAIMS the
|
||
// given client-facing endpoint path (its static.ingress), or "" when none does.
|
||
//
|
||
// This is the kernel-yields decision for an inbound endpoint (user requirement:
|
||
// "/v1/messages is the kernel's fallback endpoint; an adapter may declare it and
|
||
// the kernel actively yields"). The kernel keeps its own built-in parser as the
|
||
// fallback (R5) — AdapterForIngress returning "" means "no adapter claimed it,
|
||
// use the built-in", so an unconfigured deployment still serves the endpoint.
|
||
//
|
||
// Ties (two adapters claiming the same path) resolve to the lexically-first
|
||
// name so the choice is deterministic rather than map-iteration order. A real
|
||
// deployment should not have two adapters claiming one ingress; when it does,
|
||
// the plugin conflict-arbitration surface is the place to make it explicit, but
|
||
// a parse endpoint must still pick SOMETHING rather than vary per request.
|
||
func (v *VM) AdapterForIngress(path string) string {
|
||
if path == "" {
|
||
return ""
|
||
}
|
||
v.mu.RLock()
|
||
defer v.mu.RUnlock()
|
||
chosen := ""
|
||
for name, p := range v.pools {
|
||
p.mu.Lock()
|
||
claims := p.static.ingress == path
|
||
p.mu.Unlock()
|
||
if claims && (chosen == "" || name < chosen) {
|
||
chosen = name
|
||
}
|
||
}
|
||
return chosen
|
||
}
|
||
|
||
// LoadAdapter compiles and registers an adapter script from a file.
|
||
//
|
||
// The path is validated for SAFETY before reading (R6): it must sit inside the
|
||
// VM's adapter dir and the derived name must be a legal adapter name (no path
|
||
// traversal, absolute paths, or whitespace). It does NOT reject built-in names
|
||
// here: an operator placing openai.lua in adapter_dir to OVERRIDE the bundled
|
||
// openai is the documented hot-override mechanism. Only PLUGIN installs are
|
||
// barred from shadowing a built-in (see InstallPluginAdapter), because a
|
||
// third-party plugin clobbering the kernel's shipped adapter is tampering,
|
||
// whereas the operator editing their own deployment is not.
|
||
func (v *VM) LoadAdapter(path string) error {
|
||
name, err := ValidateAdapterPath(path, v.dir, nil)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
data, err := os.ReadFile(path)
|
||
if err != nil {
|
||
return fmt.Errorf("read adapter: %w", err)
|
||
}
|
||
return v.LoadAdapterSource(name, string(data))
|
||
}
|
||
|
||
// LoadAdapterSource registers an adapter from source code. Replacing an
|
||
// existing adapter shuts the old pool down and re-boots lazily, carrying the
|
||
// configured ceiling over so a hot-reload does not reset elastic sizing.
|
||
//
|
||
// The name is validated for path SAFETY only (no traversal/absolute/space).
|
||
// Built-in names are allowed: an operator override is legitimate. A plugin
|
||
// that must NOT shadow a built-in uses InstallPluginAdapter, which adds the
|
||
// reserved-name gate.
|
||
func (v *VM) LoadAdapterSource(name, code string) error {
|
||
if err := ValidateAdapterName(name, nil); err != nil {
|
||
return err
|
||
}
|
||
// 共享前导注入点 —— 只此一处。适配器源码有三个编译点
|
||
//(worker boot / HasFn 探针 / inspectScript),它们全部源自这里存进
|
||
// pool 的 script 文本,所以拼在这里就同时覆盖三条路径,且不改变 Lua
|
||
// 作用域语义(运行时 SetGlobal 会在热重载/池重建时丢失)。
|
||
// 见 adapter_prelude.go 的开篇:12 个适配器里有 12680 字节的逐字重复。
|
||
code = withAdapterPrelude(code)
|
||
static, err := inspectScript(name, code)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
maxW := 1
|
||
v.mu.RLock()
|
||
if p, ok := v.pools[static.name]; ok {
|
||
p.mu.Lock()
|
||
maxW = p.maxW
|
||
p.mu.Unlock()
|
||
}
|
||
v.mu.RUnlock()
|
||
|
||
v.mu.Lock()
|
||
// 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.
|
||
//
|
||
// The key can live under EITHER name (this is the "reload replaces the same
|
||
// pool entry rather than silently registering a second one" rule):
|
||
//
|
||
// - pools[static.name] — the pool filed under the DECLARED name (the
|
||
// common case, and the only case when a plugin installs under a key that
|
||
// matches its script's own name), or
|
||
// The pool registry is keyed by the DECLARED name (static.name), always:
|
||
// first install files pools[static.name], and a reload replaces that same
|
||
// slot. registeredAs only RECORDS the registration key for reverse lookups
|
||
// (EffectiveAdapterName scans for it); it is not a second key space.
|
||
//
|
||
// An earlier version redirected the key to prev.static.registeredAs here,
|
||
// assuming pools were filed under the registration key — but first installs
|
||
// file under the declared name, so that redirect created a SECOND pool and
|
||
// left the old one serving whenever key≠declared (a real bug: a v2 install
|
||
// via arbitration left v1 serving, and UpdateAdapter had the same hole).
|
||
key := static.name
|
||
old := v.pools[key]
|
||
// The new pool MUST be filed under the SAME slot as the old one (key), and
|
||
// named after that slot, so a later lookup by either the DECLARED name or
|
||
// the REGISTRATION KEY finds the one live pool. Naming it after static.name
|
||
// would move the pool to the declared-name slot when the incumbent lived
|
||
// under the registration key, leaving the registration-key slot pointing at
|
||
// the stale (old) pool that shutdown() then tears down — callers reading by
|
||
// the registration key see "adapter missing", callers by declared name see
|
||
// a pool that isn't the one serving. key is the stable identity; the
|
||
// declared name is reachable through static.name.
|
||
p := newAdapterPool(key, code, *static)
|
||
p.setMax(maxW)
|
||
// Carry the revision forward so the new source keeps counting ABOVE the old
|
||
// one. A cache keyed on (vm, revision, name) must see a value different
|
||
// from the previous generation's; restarting at 1 could collide with a
|
||
// value an earlier generation already published, which would resurrect a
|
||
// stale cache entry — exactly the bug this revision exists to prevent.
|
||
//
|
||
// The old pool is read while v.mu is HELD (a nested RLock here would
|
||
// self-deadlock: sync.RWMutex is not reentrant) and while its own mu is
|
||
// free, because shutdown() — which takes p.mu — happens after v.mu is
|
||
// released below. No other path can mutate p.revision while we hold v.mu,
|
||
// because every writer reaches the pool through v.mu first.
|
||
if old != nil {
|
||
old.mu.Lock()
|
||
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.startJanitor()
|
||
return nil
|
||
}
|
||
|
||
// RemoveAdapter evicts an adapter and shuts its workers down.
|
||
func (v *VM) RemoveAdapter(name string) {
|
||
v.mu.Lock()
|
||
p := v.pools[name]
|
||
delete(v.pools, name)
|
||
v.mu.Unlock()
|
||
if p != nil {
|
||
p.shutdown()
|
||
}
|
||
}
|
||
|
||
// poolGlobal returns the Lua global holding a pool's returned table: plugins
|
||
// store theirs under one global, adapters under another. Transform and friends
|
||
// must read the SAME one boot() wrote, or a plugin-mode pool answers
|
||
// "has no table" even though the state is perfectly healthy.
|
||
func poolGlobal(p *adapterPool) string {
|
||
if p != nil && p.pluginMode {
|
||
return pluginGlobal
|
||
}
|
||
return adapterGlobal
|
||
}
|
||
|
||
func (v *VM) pool(name string) *adapterPool {
|
||
v.mu.RLock()
|
||
defer v.mu.RUnlock()
|
||
return v.pools[name]
|
||
}
|
||
|
||
// bundledNames returns the set of adapter names that ship inside the kernel.
|
||
// It is derived from the embedded FS (same source writeBundledAdapters uses),
|
||
// so a new built-in adapter is automatically reserved and an operator cannot
|
||
// clobber it by importing a same-named external adapter.
|
||
// BundledAdapterNames is the exported counterpart of bundledNames for callers
|
||
// outside the lua package (core.UploadAdapter validates against it). It returns
|
||
// the set of adapter names shipped inside the kernel so an external import that
|
||
// shadows one is rejected up front (R6).
|
||
func (v *VM) BundledAdapterNames() []string { return v.bundledNames() }
|
||
|
||
func (v *VM) bundledNames() []string {
|
||
entries, err := bundledAdapters.ReadDir("adapters")
|
||
if err != nil {
|
||
return nil
|
||
}
|
||
out := make([]string, 0, len(entries))
|
||
for _, e := range entries {
|
||
if e.IsDir() {
|
||
continue
|
||
}
|
||
out = append(out, strings.TrimSuffix(e.Name(), ".lua"))
|
||
}
|
||
return out
|
||
}
|
||
|
||
func (v *VM) writeBundledAdapters() error {
|
||
// Derive the set from the embedded FS instead of a hardcoded name list so
|
||
// adding an adapter never requires maintaining a second list.
|
||
entries, err := bundledAdapters.ReadDir("adapters")
|
||
if err != nil {
|
||
return err
|
||
}
|
||
for _, e := range entries {
|
||
if e.IsDir() || !strings.HasSuffix(e.Name(), ".lua") {
|
||
continue
|
||
}
|
||
dst := filepath.Join(v.dir, e.Name())
|
||
if _, err := os.Stat(dst); err == nil {
|
||
continue
|
||
}
|
||
data, err := bundledAdapters.ReadFile("adapters/" + e.Name())
|
||
if err != nil {
|
||
continue
|
||
}
|
||
if err := os.WriteFile(dst, data, 0644); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// 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 := acquireAcrossSwap(v, name, p)
|
||
if err != nil {
|
||
return "", err
|
||
}
|
||
defer p.release(w)
|
||
L := w.L
|
||
|
||
L.SetTop(0)
|
||
defer L.SetTop(0)
|
||
|
||
L.GetGlobal(poolGlobal(p))
|
||
if L.IsNil(-1) {
|
||
return "", fmt.Errorf("adapter %s has no table", name)
|
||
}
|
||
L.GetField(-1, fn)
|
||
if L.IsNil(-1) || !L.IsFunction(-1) {
|
||
return "", fmt.Errorf("adapter %s missing %s", name, fn)
|
||
}
|
||
L.PushString(raw)
|
||
if err := L.Call(1, 1); err != nil {
|
||
return "", fmt.Errorf("%s: %w", fn, err)
|
||
}
|
||
return L.ToString(-1), nil
|
||
}
|
||
|
||
// BuildHeaders executes adapter.build_headers(meta). If the adapter defines no
|
||
// build_headers hook, it falls back to the static adapter.headers table.
|
||
func (v *VM) BuildHeaders(name string, meta map[string]interface{}) (map[string]string, error) {
|
||
p := v.pool(name)
|
||
if p == nil {
|
||
return nil, fmt.Errorf("adapter %s not loaded", name)
|
||
}
|
||
|
||
w, err := p.acquire()
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer p.release(w)
|
||
L := w.L
|
||
|
||
L.SetTop(0)
|
||
defer L.SetTop(0)
|
||
|
||
L.GetGlobal(adapterGlobal)
|
||
if L.IsNil(-1) {
|
||
return nil, fmt.Errorf("adapter %s has no table", name)
|
||
}
|
||
L.GetField(-1, "build_headers")
|
||
if !L.IsFunction(-1) {
|
||
// no hook: fall back to the static headers captured at load time
|
||
p.mu.Lock()
|
||
hdr := make(map[string]string, len(p.static.headers))
|
||
for k, h := range p.static.headers {
|
||
hdr[k] = h
|
||
}
|
||
p.mu.Unlock()
|
||
return hdr, nil
|
||
}
|
||
|
||
L.SetTop(0)
|
||
L.GetGlobal(adapterGlobal)
|
||
L.GetField(-1, "build_headers")
|
||
pushGoValue(L, meta)
|
||
if err := L.Call(1, 1); err != nil {
|
||
return nil, fmt.Errorf("build_headers: %w", err)
|
||
}
|
||
if L.Type(-1) != golua.LUA_TTABLE {
|
||
return nil, fmt.Errorf("build_headers returned non-table")
|
||
}
|
||
headers := map[string]string{}
|
||
luaTableToMap(L, -1, headers)
|
||
return headers, nil
|
||
}
|
||
|
||
// TransformError executes the optional adapter.transform_error(status, body)
|
||
// hook: per-source condensing of an upstream error response into a short
|
||
// client-facing reason. ok=false means the adapter defines no hook (or it
|
||
// failed) — the caller falls back to the generic Go-side condenser.
|
||
func (v *VM) TransformError(name string, status int, body string) (reason string, ok bool, err error) {
|
||
p := v.pool(name)
|
||
if p == nil {
|
||
return "", false, fmt.Errorf("adapter %s not loaded", name)
|
||
}
|
||
w, err := p.acquire()
|
||
if err != nil {
|
||
return "", false, err
|
||
}
|
||
defer p.release(w)
|
||
L := w.L
|
||
|
||
L.SetTop(0)
|
||
defer L.SetTop(0)
|
||
L.GetGlobal(adapterGlobal)
|
||
if L.IsNil(-1) {
|
||
return "", false, nil
|
||
}
|
||
L.GetField(-1, "transform_error")
|
||
if !L.IsFunction(-1) {
|
||
return "", false, nil
|
||
}
|
||
L.SetTop(0)
|
||
L.GetGlobal(adapterGlobal)
|
||
L.GetField(-1, "transform_error")
|
||
L.PushInteger(int64(status))
|
||
L.PushString(body)
|
||
if callErr := L.Call(2, 1); callErr != nil {
|
||
return "", false, fmt.Errorf("transform_error: %w", callErr)
|
||
}
|
||
if L.Type(-1) != golua.LUA_TSTRING {
|
||
return "", false, nil
|
||
}
|
||
return L.ToString(-1), true, nil
|
||
}
|
||
|
||
func (v *VM) Endpoint(name string) string {
|
||
p := v.pool(name)
|
||
if p == nil {
|
||
return ""
|
||
}
|
||
p.mu.Lock()
|
||
defer p.mu.Unlock()
|
||
return p.static.endpoint
|
||
}
|
||
|
||
// StreamEndpoint returns the adapter's streaming endpoint, or "" when it
|
||
// declares none (in which case the caller keeps using Endpoint).
|
||
//
|
||
// This is the whole mechanism that keeps vendor streaming verbs out of the
|
||
// kernel: an adapter whose stream path differs declares endpoint_stream, and
|
||
// the dispatcher only ever compares two opaque strings.
|
||
func (v *VM) StreamEndpoint(name string) string {
|
||
p := v.pool(name)
|
||
if p == nil {
|
||
return ""
|
||
}
|
||
p.mu.Lock()
|
||
defer p.mu.Unlock()
|
||
return p.static.endpointStream
|
||
}
|
||
|
||
// HasFn reports whether the named adapter declares a callable field fn,
|
||
// WITHOUT acquiring a worker and WITHOUT running the script.
|
||
//
|
||
// It exists for capability probing that runs on the request path: the
|
||
// out-bound serializer asks "does this adapter declare serialize_chunk?" once
|
||
// per adapter, not once per chunk. Probing by calling Transform would be
|
||
// wrong twice over — it costs a pool checkout, and a MISSING hook comes back
|
||
// as an error string rather than a boolean.
|
||
//
|
||
// The probe runs on a throwaway state built from the SAME script text, so the
|
||
// answer reflects what the script declares. A script whose table is built
|
||
// conditionally on its environment could differ per state; that is already
|
||
// true of every other staticInfo field (endpoint, headers), so the pool does
|
||
// not get a special exemption here.
|
||
func (v *VM) HasFn(name, fn string) bool {
|
||
p := v.pool(name)
|
||
if p == nil {
|
||
return false
|
||
}
|
||
p.mu.Lock()
|
||
script := p.script
|
||
p.mu.Unlock()
|
||
if script == "" {
|
||
return false
|
||
}
|
||
L := golua.NewState()
|
||
defer L.Close()
|
||
L.OpenLibs()
|
||
setupGlobals(L)
|
||
if err := L.DoString(script); err != nil {
|
||
return false
|
||
}
|
||
if L.Type(-1) != golua.LUA_TTABLE {
|
||
return false
|
||
}
|
||
L.GetField(-1, fn)
|
||
defer L.Pop(1)
|
||
return !L.IsNil(-1) && L.IsFunction(-1)
|
||
}
|
||
|
||
// ---- static inspection (compile-once at load time) ----
|
||
|
||
func inspectScript(name, code string) (*staticInfo, error) {
|
||
L := golua.NewState()
|
||
L.OpenLibs()
|
||
setupGlobals(L)
|
||
if err := L.DoString(code); err != nil {
|
||
L.Close()
|
||
return nil, fmt.Errorf("compile adapter: %w", err)
|
||
}
|
||
if L.Type(-1) != golua.LUA_TTABLE {
|
||
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")
|
||
info.ingress = readString(L, -1, "ingress")
|
||
// hidden:适配器可声明 adapter.hidden = true 退出用户可见列表(仍可被源引用)。
|
||
L.GetField(-1, "hidden")
|
||
info.hidden = L.ToBoolean(-1)
|
||
L.Pop(1)
|
||
if getFieldBatch(L, "headers") {
|
||
luaTableToMap(L, -1, info.headers)
|
||
L.Pop(1)
|
||
} else {
|
||
L.Pop(1)
|
||
}
|
||
L.Close()
|
||
return info, nil
|
||
}
|
||
|
||
// getFieldBatch fetches field of the table at the top of the stack and returns
|
||
// true when the fetched value (now at the top) is a table.
|
||
func getFieldBatch(L *golua.State, field string) bool {
|
||
L.GetField(-1, field)
|
||
return L.Type(-1) == golua.LUA_TTABLE
|
||
}
|
||
|
||
func readString(L *golua.State, tableIdx int, field string) string {
|
||
L.GetField(tableIdx, field)
|
||
defer L.Pop(1)
|
||
if L.Type(-1) == golua.LUA_TNIL || L.Type(-1) == golua.LUA_TNONE {
|
||
return ""
|
||
}
|
||
return normalizeString(L, -1)
|
||
}
|
||
|
||
func normalizeString(L *golua.State, idx int) string {
|
||
if L.Type(idx) == golua.LUA_TNUMBER {
|
||
n := L.ToNumber(idx)
|
||
return strconv.FormatFloat(n, 'f', -1, 64)
|
||
}
|
||
return L.ToString(idx)
|
||
}
|
||
|
||
// ---- shared globals installed into every worker state ----
|
||
|
||
// restorePcall puts the real pcall/xpcall back in place. golua's OpenLibs
|
||
// replaces them with unsafe_ variants (it captures errors via its own panic
|
||
// handler); the built-in adapters rely on pcall for soft error handling.
|
||
func restorePcall(L *golua.State) {
|
||
for _, name := range []string{"pcall", "xpcall"} {
|
||
L.GetGlobal("unsafe_" + name)
|
||
if !L.IsNil(-1) {
|
||
L.SetGlobal(name)
|
||
} else {
|
||
L.Pop(1)
|
||
}
|
||
}
|
||
}
|
||
|
||
// hookGuardName is the Lua global that wraps every plugin hook call.
|
||
const hookGuardName = "__llmsproxy_call_hook"
|
||
|
||
// hookGuardSrc defines the wrapper. It exists because of a hard constraint in
|
||
// the binding: golua's callEx calls L.StackTrace() on ANY pcall error, and
|
||
// StackTrace() calls lua_getinfo, which SIGSEGVs in this LuaJIT build once the
|
||
// stack is deep enough (a request_end payload with the AUTO chain trace does
|
||
// it). Go cannot recover from a C-level signal, so a single Lua mistake inside
|
||
// one plugin killed the whole gateway — the production outage this was written
|
||
// for. Repeated crashes proved the boundary is not theoretical.
|
||
//
|
||
// pcall INSIDE Lua catches the error before golua ever sees a non-zero
|
||
// pcall status, so the C stack-trace path is never entered. The failure comes
|
||
// back as an ordinary (nil, message) pair, which the Go side records in
|
||
// hook_errors and moves on from — the documented contract that "a broken plugin
|
||
// must not affect request forwarding" finally holds for script errors too, not
|
||
// just for Go panics.
|
||
//
|
||
// returns: (result, errorMessage) — both nil/"" on success.
|
||
const hookGuardSrc = `
|
||
function ` + hookGuardName + `(fn, payload)
|
||
local ok, res = pcall(fn, payload)
|
||
if not ok then
|
||
return nil, tostring(res)
|
||
end
|
||
return res, nil
|
||
end`
|
||
|
||
func registerHookGuard(L *golua.State) {
|
||
if err := L.DoString(hookGuardSrc); err != nil {
|
||
// Nothing useful to do here beyond leaving the global absent: invoke()
|
||
// checks for it and falls back to a direct call, which still works for
|
||
// correct plugins (only their ERRORS stop being survivable).
|
||
L.Pop(1)
|
||
}
|
||
}
|
||
|
||
func setupGlobals(L *golua.State) {
|
||
restorePcall(L)
|
||
buildJSONTable(L)
|
||
registerHookGuard(L)
|
||
|
||
registerFn(L, "hmac_sha256_hex", func(L *golua.State) int {
|
||
key := L.ToString(1)
|
||
data := L.ToString(2)
|
||
m := hmac.New(sha256.New, []byte(key))
|
||
m.Write([]byte(data))
|
||
L.PushString(hex.EncodeToString(m.Sum(nil)))
|
||
return 1
|
||
})
|
||
registerFn(L, "sha256_hex", func(L *golua.State) int {
|
||
h := sha256.Sum256([]byte(L.ToString(1)))
|
||
L.PushString(hex.EncodeToString(h[:]))
|
||
return 1
|
||
})
|
||
registerFn(L, "base64_encode", func(L *golua.State) int {
|
||
L.PushString(base64.StdEncoding.EncodeToString([]byte(L.ToString(1))))
|
||
return 1
|
||
})
|
||
registerFn(L, "tohex", func(L *golua.State) int {
|
||
L.PushString(hex.EncodeToString([]byte(L.ToString(1))))
|
||
return 1
|
||
})
|
||
registerFn(L, "log", func(L *golua.State) int {
|
||
fmt.Printf("[adapter/%s] %s\n", L.ToString(1), L.ToString(2))
|
||
return 0
|
||
})
|
||
}
|
||
|
||
func buildJSONTable(L *golua.State) {
|
||
L.NewTable()
|
||
for _, name := range []string{"encode", "decode"} {
|
||
var fn golua.LuaGoFunction
|
||
switch name {
|
||
case "encode":
|
||
fn = func(L *golua.State) int {
|
||
b, err := jsonEncode(luaValueToGo(L, 1))
|
||
if err != nil {
|
||
L.PushString("null")
|
||
return 1
|
||
}
|
||
L.PushString(string(b))
|
||
return 1
|
||
}
|
||
case "decode":
|
||
fn = func(L *golua.State) int {
|
||
v, err := jsonDecode(L.ToString(1))
|
||
if err != nil {
|
||
L.PushNil()
|
||
return 1
|
||
}
|
||
pushGoValue(L, v)
|
||
return 1
|
||
}
|
||
}
|
||
L.PushGoClosure(fn)
|
||
L.SetField(-2, name)
|
||
}
|
||
L.SetGlobal("json")
|
||
}
|
||
|
||
func registerFn(L *golua.State, name string, fn golua.LuaGoFunction) {
|
||
L.PushGoClosure(fn)
|
||
L.SetGlobal(name)
|
||
}
|
||
|
||
// ---- value conversions ----
|
||
|
||
func luaValueToGo(L *golua.State, idx int) interface{} {
|
||
switch L.Type(idx) {
|
||
case golua.LUA_TSTRING:
|
||
return L.ToString(idx)
|
||
case golua.LUA_TNUMBER:
|
||
return L.ToNumber(idx)
|
||
case golua.LUA_TBOOLEAN:
|
||
return L.ToBoolean(idx)
|
||
case golua.LUA_TTABLE:
|
||
return luaTableToGoValue(L, idx)
|
||
default:
|
||
return nil
|
||
}
|
||
}
|
||
|
||
// luaTableToGo value converts a Lua table to either []interface{} (array 1..n)
|
||
// or map[string]interface{}.
|
||
func luaTableToGoValue(L *golua.State, idx int) interface{} {
|
||
abs := idx
|
||
if idx < 0 {
|
||
abs = L.GetTop() + idx + 1
|
||
}
|
||
m := map[string]interface{}{}
|
||
allInt, n, maxK := true, 0, 0
|
||
L.PushNil()
|
||
for L.Next(abs) != 0 {
|
||
kt := L.Type(-2)
|
||
if kt == golua.LUA_TNUMBER {
|
||
k := int(L.ToNumber(-2))
|
||
if k > maxK {
|
||
maxK = k
|
||
}
|
||
} else {
|
||
allInt = false
|
||
}
|
||
m[keyToString(L, -2)] = luaValueToGo(L, -1)
|
||
L.Pop(1)
|
||
n++
|
||
}
|
||
if allInt && maxK > 0 && n == maxK {
|
||
arr := make([]interface{}, maxK)
|
||
for i := 1; i <= maxK; i++ {
|
||
arr[i-1] = m[strconv.Itoa(i)]
|
||
}
|
||
return arr
|
||
}
|
||
return m
|
||
}
|
||
|
||
func luaTableToMap(L *golua.State, idx int, dst map[string]string) {
|
||
abs := idx
|
||
if idx < 0 {
|
||
abs = L.GetTop() + idx + 1
|
||
}
|
||
L.PushNil()
|
||
for L.Next(abs) != 0 {
|
||
dst[keyToString(L, -2)] = normalizeString(L, -1)
|
||
L.Pop(1)
|
||
}
|
||
}
|
||
|
||
// keyToString reads the stack value at idx as a string without modifying the
|
||
// original (golua's ToString mutates numeric stack values in place, which
|
||
// would break the next lua_next call), then pops the temporary copy.
|
||
func keyToString(L *golua.State, idx int) string {
|
||
L.PushValue(idx)
|
||
defer L.Pop(1)
|
||
return normalizeString(L, -1)
|
||
}
|
||
|
||
func pushGoValue(L *golua.State, v interface{}) {
|
||
switch x := v.(type) {
|
||
case nil:
|
||
L.PushNil()
|
||
case string:
|
||
L.PushString(x)
|
||
case bool:
|
||
L.PushBoolean(x)
|
||
case int:
|
||
L.PushNumber(float64(x))
|
||
case int64:
|
||
L.PushNumber(float64(x))
|
||
case float64:
|
||
L.PushNumber(x)
|
||
case []interface{}:
|
||
L.NewTable()
|
||
for i, it := range x {
|
||
pushGoValue(L, it)
|
||
L.RawSeti(-2, i+1)
|
||
}
|
||
case map[string]interface{}:
|
||
L.NewTable()
|
||
for k, it := range x {
|
||
pushGoValue(L, it)
|
||
L.SetField(-2, k)
|
||
}
|
||
default:
|
||
if b, err := json.Marshal(x); err == nil {
|
||
var iv interface{}
|
||
if json.Unmarshal(b, &iv) == nil {
|
||
pushGoValue(L, iv)
|
||
return
|
||
}
|
||
}
|
||
L.PushNil()
|
||
}
|
||
}
|
||
|
||
func jsonEncode(v interface{}) ([]byte, error) { return json.Marshal(v) }
|
||
|
||
// luaToJSON converts the Lua value at idx into a Go value via json.encode, then
|
||
// unmarshals it into out. It is the bridge used by the plugin manifest/UI
|
||
// reader: the plugin returns a plain Lua table, and Go wants a typed struct.
|
||
//
|
||
// It goes through JSON rather than walking the Lua stack directly because the
|
||
// adapter/plugin boundary already speaks JSON everywhere else (transform_request
|
||
// gets a JSON string, hooks get a JSON string), so this keeps one representation
|
||
// instead of two.
|
||
func luaToJSON(L *golua.State, idx int, out interface{}) error {
|
||
if L.GetTop() < 1 {
|
||
return fmt.Errorf("empty stack")
|
||
}
|
||
abs := idx
|
||
if abs < 0 {
|
||
abs = L.GetTop() + 1 + abs
|
||
}
|
||
if abs < 1 || abs > L.GetTop() {
|
||
return fmt.Errorf("index %d out of range (top=%d)", idx, L.GetTop())
|
||
}
|
||
// Absolute indices throughout: this binding aborts the process (SIGABRT)
|
||
// on a bad index rather than panicking, so the stack is captured before any
|
||
// push instead of being addressed relative to a shifting top.
|
||
//
|
||
// json.encode is pushed onto the stack and the value is pushed AFTER it, so
|
||
// Call(1, 1) invokes it (Call takes no function index — it calls whatever
|
||
// sits below the nargs values).
|
||
L.GetGlobal("json")
|
||
if L.IsNil(-1) {
|
||
L.SetTop(0)
|
||
return fmt.Errorf("json global missing")
|
||
}
|
||
L.GetField(-1, "encode")
|
||
if L.Type(-1) != golua.LUA_TFUNCTION {
|
||
L.SetTop(0)
|
||
return fmt.Errorf("json.encode missing")
|
||
}
|
||
L.PushValue(abs)
|
||
if err := L.Call(1, 1); err != nil {
|
||
L.SetTop(0)
|
||
return err
|
||
}
|
||
if L.GetTop() < 1 || L.Type(-1) != golua.LUA_TSTRING {
|
||
L.SetTop(0)
|
||
return fmt.Errorf("json.encode did not return a string")
|
||
}
|
||
s := L.ToString(-1)
|
||
L.SetTop(0)
|
||
return json.Unmarshal([]byte(s), out)
|
||
}
|
||
func jsonDecode(s string) (interface{}, error) {
|
||
var v interface{}
|
||
if err := json.Unmarshal([]byte(s), &v); err != nil {
|
||
return nil, err
|
||
}
|
||
return v, nil
|
||
}
|
||
|
||
// AdapterRevision returns a monotonically increasing revision for the named
|
||
// adapter's CURRENT source. Every LoadAdapterSource / RemoveAdapter bumps it.
|
||
//
|
||
// Purpose: callers that cache a decision derived from an adapter's declared
|
||
// surface (e.g. "does this adapter declare serialize_chunk?") store the
|
||
// revision alongside the cached value and treat a mismatch as invalid. That
|
||
// removes the need for those callers to be told when to invalidate — which
|
||
// they previously were not, so a plugin rewriting an adapter at runtime could
|
||
// leave a stale decision in place forever.
|
||
//
|
||
// The revision is per-adapter rather than global so an unrelated adapter's
|
||
// reload does not discard a still-valid cache entry.
|
||
func (v *VM) AdapterRevision(name string) uint64 {
|
||
p := v.pool(name)
|
||
if p == nil {
|
||
return 0
|
||
}
|
||
p.mu.Lock()
|
||
defer p.mu.Unlock()
|
||
return p.revision
|
||
}
|
||
|
||
// InstallPluginAdapter registers an adapter ON BEHALF OF A PLUGIN, with the
|
||
// one extra rule plugins (and only plugins) are held to: the name must not
|
||
// shadow a built-in adapter.
|
||
//
|
||
// Why plugins are stricter than operators: an operator dropping openai.lua in
|
||
// adapter_dir is overriding their own deployment (documented, intended). A
|
||
// third-party plugin registering an adapter named "openai" would silently
|
||
// swap out the kernel's shipped protocol translation for code the operator
|
||
// never reviewed — tampering with the shipped artifact. R6 requires that to
|
||
// fail loudly with the conflict named, not succeed quietly.
|
||
//
|
||
// This is the entry point Phase 3's plugin adapter-install bridge calls.
|
||
func (v *VM) InstallPluginAdapter(name, code string) error {
|
||
if err := ValidateAdapterName(name, v.bundledNames()); err != nil {
|
||
return err
|
||
}
|
||
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
|
||
}
|
||
|
||
// AdapterIngress returns the client-facing endpoint path an adapter CLAIMS
|
||
// (its static.ingress), or "" when it claims none. Exported so operator-facing
|
||
// surfaces (and tests) can see which adapter, if any, has taken over parsing of
|
||
// an inbound endpoint — the same declaration AdapterForIngress consults.
|
||
func (v *VM) AdapterIngress(name string) string {
|
||
p := v.pool(name)
|
||
if p == nil {
|
||
return ""
|
||
}
|
||
p.mu.Lock()
|
||
defer p.mu.Unlock()
|
||
return p.static.ingress
|
||
}
|
||
|
||
// 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
|
||
}
|