mirror of
https://gitcode.com/JianFeeeee/ModelRouter.git
synced 2026-11-12 20:27:23 +00:00
feat(gateway): 对外序列化移交适配器,OpenAI 格式降级为最后容错
内核不再硬编码对外 wire。writeChatCompletion / pumpStream 里的 OpenAI 帧拼装抽到 Serializer 接口,适配器可声明钩子完全接管。 优先级契约(用户明确要求:冲突时兜底直接让步): 适配器声明 serialize_* -> 适配器完全接管 未声明 -> 兜底 openAISerializer 「让步」的含义:适配器接管后内核绝不往其输出补任何 OpenAI 字段。混合体 (半个 Anthropic + 半个 OpenAI)比任何一个单独结果都更难排查——客户端无法 解析,而排查时会指向错误的层。 接口把「流如何结束」也纳入:OpenAI 需 usage 帧 + [DONE],Anthropic 用 message_stop,Gemini 用空 candidates 帧。DoneMarker 让适配器声明自己的 终止标记,避免擅自给严格客户端加 [DONE]。 新增能力: - VM.HasFn(name, fn):不占 worker 探测适配器是否声明某函数 - VM.AdapterRevision(name):适配器源码每次替换 +1(接着旧值递增) 修掉两个开发中引入的真实缺陷: 1. 锁序倒置自死锁:LoadAdapterSource 里在已持有 v.mu.Lock() 时嵌套 v.mu.RLock() 取 revision,RWMutex 不可重入 -> 整个测试套件静默挂起。 症状是不报错也不超时,只是永不返回;靠 go test -timeout 的 goroutine 栈定位。 2. 缓存键漏掉 VM 维度:serializerCache 原按适配器名做键,导致同进程内两个 VM 共享判定,适配器重写后新钩子永不生效且不报错。键改为 (VM, revision, adapter),回归测试 TestSerializerCacheIsPerVM。 测试三件套: - 功能:适配器接管 / 兜底(无适配器)/ 钩子失败退回兜底 / 缓存随改写失效 - 压力:TestSerializerForConcurrent(16 worker × 40 次) - 回归:golden 字节级未漂移,Phase 0 安全网生效 验证:go test -tags luajit ./... 全绿(gateway 200s / lua 56s)。
This commit is contained in:
@ -114,6 +114,12 @@ type adapterPool struct {
|
||||
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
|
||||
@ -174,7 +180,7 @@ type PoolStats struct {
|
||||
}
|
||||
|
||||
func newAdapterPool(name, script string, static staticInfo) *adapterPool {
|
||||
p := &adapterPool{name: name, script: script, static: static, maxW: 1}
|
||||
p := &adapterPool{name: name, script: script, static: static, maxW: 1, revision: 1}
|
||||
p.cond = sync.NewCond(&p.mu)
|
||||
return p
|
||||
}
|
||||
@ -750,11 +756,28 @@ func (v *VM) LoadAdapterSource(name, code string) error {
|
||||
v.mu.RUnlock()
|
||||
|
||||
v.mu.Lock()
|
||||
if p, ok := v.pools[static.name]; ok {
|
||||
p.shutdown()
|
||||
}
|
||||
old := v.pools[static.name]
|
||||
p := newAdapterPool(static.name, 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()
|
||||
}
|
||||
if old != nil {
|
||||
old.shutdown()
|
||||
}
|
||||
v.pools[static.name] = p
|
||||
v.mu.Unlock()
|
||||
v.startJanitor()
|
||||
@ -950,6 +973,46 @@ func (v *VM) StreamEndpoint(name string) string {
|
||||
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) {
|
||||
@ -1295,3 +1358,25 @@ func jsonDecode(s string) (interface{}, error) {
|
||||
}
|
||||
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
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user