mirror of
https://gitcode.com/JianFeeeee/ModelRouter.git
synced 2026-11-07 09:47:23 +00:00
fix(lua): 定时任务改跑在插件自己的 VM 上 —— 纠正 Phase 4 的双 VM 错误
用户指出的设计错误:**一个插件 = 一个 Lua VM = 一个调度器**。而 Phase 4
(4d1407b)给定时任务开了一个**私有 VM**,于是:
定时任务 refresh_token 写 plugin.state.token → 写进私有 VM 的副本
请求钩子 读 plugin.state.token → 读主 VM 的值
轮换"成功"了,对实际流量完全无效;而且私有 VM 绕过状态持久化,token 连
磁盘都到不了。这直接违背这个功能存在的理由。
## 修正
PluginRuntime 不再持有 Lua 状态。任务通过 p.state + p.mu 运行 —— 与请求
钩子**同一个 VM、同一把锁**:
- 任务写入立即可见于钩子(TestTaskWriteIsVisibleToHooks 锁死这一点)
- 任务写入经 markDirtyLocked 走同一套持久化
- 跨插件仍并行(每插件各自 VM、各自锁)
## 由此产生的行为(有意为之,非缺陷)
- 某插件的长任务会排在**它自己**的钩子前面(同一 Lua 栈,一次一个调用者)
- 卡死处置随之改变:不再"抛弃 VM 重建"(那会丢掉累计状态——计费总额、
签到历史,正是插件存在的意义),而是**停用该插件**:
- checkWedged 超过 wedgedAfter(10min) 置 disarmed
- Fire 对 disarmed 插件快速失败并记 hook_errors(否则请求会排在永不释放
的锁后面永久挂起——比停用更糟)
- Unload 检测 InFlight:有任务在跑时不取 p.mu(会永久挂起)、不 Close
状态(Lua C 层 use-after-free 直接 abort)。泄漏这一个状态是有界的,
比崩溃好。
- maxAbandons/抛弃计数因此删除:没有"重生"了,熔断由 disarmed 承担
## 测试(10 个)
主断言从"任务跑过"改为"**任务的写入对请求路径可见**"—— 旧实现会在这里
失败。另含:无触发自执行/无 schedule 无运行时/解析/at 下次触发/错误隔离/
卸载停 goroutine/卡死停用+钩子快失败/健康任务不误判/8 插件并发各自调度。
This commit is contained in:
@ -1,40 +1,43 @@
|
||||
// Package lua — 插件长期运行环境(R4-b / Phase 4)
|
||||
//
|
||||
// 用户需求:插件需要**长期运行环境(独立 LuaVM 线程化)**,以支持自动签到、
|
||||
// token 轮换等需要"没有浏览器、没有人操作时也会发生"的能力。
|
||||
// 用户需求:插件需要**长期运行环境**,以支持自动签到、token 轮换等需要
|
||||
// "没有浏览器、没有人操作时也会发生"的能力。
|
||||
//
|
||||
// 为什么需要独立 VM:现有插件模型是「一个插件 = 一个 Lua 状态 + 一把锁」,
|
||||
// 因为钩子会往 plugin.state 累加,多个状态会让总额被劈成几份(见 docs 的
|
||||
// 运行时约束)。这把锁的代价是**一个卡住的钩子会卡住所有插件**——而签到、
|
||||
// token 轮换这类任务天然是长耗时的网络等待。
|
||||
// 模型(重要):**一个插件 = 一个 Lua VM = 一个调度器**。
|
||||
//
|
||||
// 本文件给出调度层:内核按插件声明的周期,在**独立 VM + 独立 goroutine** 上
|
||||
// 执行它,从而:
|
||||
// 钩子(请求路径)和定时任务(内核 ticker 驱动)跑在**同一个** Lua 状态上,
|
||||
// 由同一把锁串行化。这不是实现细节,而是正确性要求:
|
||||
//
|
||||
// - 一个插件的长任务不阻塞其它插件的钩子(独立 VM,无共享锁)
|
||||
// - 任务在没有浏览器、没有人操作时也会执行(内核 ticker 驱动,不是前端轮询)
|
||||
// - 状态与主 VM 分离,轮换中崩溃不会带走已累计的钩子统计
|
||||
// 定时任务 refresh_token 写 plugin.state.token
|
||||
// 请求钩子 读 plugin.state.token
|
||||
//
|
||||
// 两者必须是同一个 Lua 状态,否则任务把 token 写进了一个请求永远读不到的
|
||||
// 副本 —— 轮换看起来"成功了",实际对所有流量无效。同理,任务的写入必须
|
||||
// 经过同一个 plugin.state,才能被状态持久化(markDirty)看到。
|
||||
//
|
||||
// 代价(已知且接受):某个插件的长耗时任务(网络等待)会排在它**自己**的
|
||||
// 钩子前面,因为同一个 Lua 栈一次只能有一个调用者。跨插件仍然是并行的
|
||||
// (每个插件各自一把锁、各自一个 VM)。
|
||||
//
|
||||
// 声明方式(插件侧):
|
||||
//
|
||||
// PLUGIN.schedule = {
|
||||
// { fn = "refresh_token", every = 3600 }, -- 每小时
|
||||
// { fn = "daily_checkin", every = 86400, at = "03:30" }, -- 每天固定时刻
|
||||
// { fn = "poll_quota", every = 300, jitter = 30 }, -- 带抖动,避免整点齐射
|
||||
// { fn = "refresh_token", every = 3600 }, -- 每小时
|
||||
// { fn = "daily_checkin", every = 86400, at = "03:30" }, -- 每天固定时刻
|
||||
// { fn = "poll_quota", every = 300, jitter = 30 }, -- 带抖动
|
||||
// }
|
||||
//
|
||||
// 设计取舍:
|
||||
// - **适配器不参与调度**:适配器仍是无状态请求级池(见 R4 的澄清)。token
|
||||
// 轮换靠插件(这里)按周期调用 Phase 3 的 UpdateAdapter。
|
||||
// - 任务在独立 VM 上跑,故**不持有**主 VM 的 plugin.state;任务要写共享
|
||||
// 状态必须走 state API(SetState / 状态文件),与 docs 的约定一致。
|
||||
// - 崩溃隔离:任务函数出错/超时只记 hookErr,不影响钩子与请求转发。
|
||||
// 崩溃隔离:任务函数出错只记 hookErr,不影响钩子与请求转发。但一个**永不返回**
|
||||
// 的任务会永久持有该插件的锁(Lua 调用无法中断,见 checkWedged 的能力说明),
|
||||
// 因此超时后该插件被**停用**而不是"换一个 VM 重来"—— 换 VM 会把累计状态
|
||||
// (计费总额、已签到的天数)一起丢掉,那比停用更糟。
|
||||
package lua
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"log"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
golua "github.com/aarzilli/golua/lua"
|
||||
@ -52,11 +55,10 @@ type ScheduledTask struct {
|
||||
// Jitter, in seconds, spreads the first fire randomly so many plugins (or
|
||||
// many replicas) do not all hit the same upstream at :00.
|
||||
Jitter int `json:"jitter,omitempty"`
|
||||
// Timeout, in seconds, bounds one execution. 0 uses defaultTaskTimeout.
|
||||
// NOTE: a running Lua call CANNOT be interrupted in this LuaJIT build (see
|
||||
// the capability note in checkWedged). The timeout is therefore enforced at
|
||||
// the generation level — a task that overruns wedgedAfter has its whole VM
|
||||
// abandoned — not by stopping the script mid-instruction.
|
||||
// Timeout is parsed for forward compatibility but is NOT enforced: a running
|
||||
// Lua call cannot be interrupted in this LuaJIT build (see the capability
|
||||
// note on checkWedged). Overrun is handled by disarming the plugin at the
|
||||
// wedgedAfter threshold instead.
|
||||
Timeout int `json:"timeout,omitempty"`
|
||||
}
|
||||
|
||||
@ -115,112 +117,73 @@ func parseSchedule(L *golua.State) scheduleSpec {
|
||||
return spec
|
||||
}
|
||||
|
||||
// PluginRuntime owns one plugin's long-lived execution environment.
|
||||
// PluginRuntime owns one plugin's periodic work.
|
||||
//
|
||||
// It is deliberately NOT the hook VM: hooks run on the shared, stateful plugin
|
||||
// VM (their accumulated totals must stay in one place). The runtime is a SEPARATE
|
||||
// VM used only for periodic jobs, whose work is inherently long-lived and
|
||||
// network-bound.
|
||||
// It does NOT own a Lua state. Tasks run on the plugin's own VM (Plugin.state),
|
||||
// under the plugin's own lock, exactly like a request hook — that is what makes
|
||||
// a token rotated by a task immediately visible to inference traffic, and what
|
||||
// makes the task's writes persist through the same state saver.
|
||||
type PluginRuntime struct {
|
||||
plugin string
|
||||
source string // kept so a discarded generation can be re-booted from scratch
|
||||
|
||||
mu sync.Mutex
|
||||
plugin *Plugin
|
||||
// ps is the owning plugin set, needed to record task failures on the same
|
||||
// hook_errors surface the request path uses.
|
||||
ps *Plugins
|
||||
spec scheduleSpec
|
||||
// gen is the CURRENT generation: one private VM plus the goroutine it runs
|
||||
// on. It is replaced wholesale when a task is judged wedged.
|
||||
gen *runtimeGen
|
||||
// running marks per-task in-flight state for the ACTIVE generation.
|
||||
running map[string]bool
|
||||
|
||||
mu sync.Mutex
|
||||
// nextDue is when each task should next fire.
|
||||
nextDue map[string]time.Time
|
||||
// failures counts consecutive task failures per task.
|
||||
failures map[string]int
|
||||
// abandons counts how many generations have been discarded for a wedged
|
||||
// task. A plugin that keeps wedging is disarmed rather than respawned
|
||||
// forever (see the circuit-breaker note).
|
||||
abandons map[string]int
|
||||
// startedAt records when each running task began, so checkWedged can tell a
|
||||
// slow-but-healthy task from a wedged one.
|
||||
// running marks a task whose goroutine has been dispatched but has not yet
|
||||
// finished (it may still be waiting for the plugin lock).
|
||||
running map[string]bool
|
||||
// startedAt is set once the Lua call actually BEGINS (i.e. after the plugin
|
||||
// lock was acquired). Measuring from dispatch would count time spent queued
|
||||
// behind a slow task as if it were this task's own runtime.
|
||||
startedAt map[string]time.Time
|
||||
// disarmed is set once a plugin has wedged maxAbandons times; it stops the
|
||||
// respawn loop and requires a manual reload.
|
||||
// failures counts consecutive failures per task.
|
||||
failures map[string]int
|
||||
// active counts task goroutines that have not returned.
|
||||
active int
|
||||
// disarmed is set once a task has overrun wedgedAfter. Scheduling stops and
|
||||
// hook dispatch fails fast, so a plugin wedged by its own task degrades to
|
||||
// "this plugin is out" rather than "requests hang forever behind its lock".
|
||||
disarmed bool
|
||||
// pendingRespawn lists tasks that were in flight on a dropped generation, so
|
||||
// the replacement can re-run them (and the abandon counter can advance).
|
||||
pendingRespawn []string
|
||||
// ps is the owning plugin set, needed to boot replacement generations.
|
||||
ps *Plugins
|
||||
|
||||
stop chan struct{}
|
||||
done chan struct{}
|
||||
}
|
||||
|
||||
// runtimeGen is one disposable execution generation: a private VM plus the
|
||||
// single-state pool the tasks run on.
|
||||
//
|
||||
// Disposability is the whole point. Because this LuaJIT build cannot interrupt a
|
||||
// running Lua call (measured three ways — see the capability note at the top of
|
||||
// this file), the only safe response to a task that will not return is to STOP
|
||||
// WAITING for it and give the plugin a fresh VM. A generation is therefore
|
||||
// treated as disposable: on a wedge the runtime drops the reference, lets the
|
||||
// stuck goroutine and its state be garbage-collected once it eventually
|
||||
// returns (or leaks if it never does), and boots a new one for future work.
|
||||
type runtimeGen struct {
|
||||
vm *VM
|
||||
poolName string
|
||||
// busy marks the pool's worker as checked out by a task that has not
|
||||
// returned. A generation with any busy task is wedged under the drop model.
|
||||
busy map[string]bool
|
||||
}
|
||||
|
||||
// wedgedAfter is how long a task may run before the runtime declares its
|
||||
// generation wedged and abandons it.
|
||||
// wedgedAfter is how long a task may run before the plugin is declared wedged.
|
||||
//
|
||||
// It is generous because the normal case IS a slow network call: check-in
|
||||
// endpoints and token refreshes routinely take seconds to tens of seconds.
|
||||
// Declaring a wedge too eagerly would abandon healthy generations and, worse,
|
||||
// could start a second copy of a task that is still legitimately running.
|
||||
var wedgedAfter = 10 * time.Minute
|
||||
|
||||
// maxAbandons disarms a plugin whose tasks keep wedging.
|
||||
//
|
||||
// Without a ceiling, a plugin with a task that always hangs would have the
|
||||
// runtime boot a fresh VM every wedgedAfter interval forever, and each abandoned
|
||||
// goroutine stays alive holding a Lua state — an unbounded leak driven by the
|
||||
// plugin's own bug. After this many abandons the runtime stops respawning and
|
||||
// requires a manual reload, which is the same posture as a systemd unit that
|
||||
// repeatedly fails to start.
|
||||
const maxAbandons = 5
|
||||
// errPluginWedged is returned to hooks of a plugin whose scheduled task
|
||||
// overran. It is a real error rather than a silent skip so the operator sees
|
||||
// WHICH plugin stopped serving and why (it lands in hook_errors).
|
||||
var errPluginWedged = errors.New("a scheduled task overran and disabled this plugin; reload it to retry")
|
||||
|
||||
// startRuntime boots the first generation and starts the scheduler.
|
||||
// startRuntime attaches a scheduler to a plugin that declared one.
|
||||
//
|
||||
// A plugin with no schedule gets no runtime at all — the common case stays
|
||||
// exactly as cheap as before: one shared state, no extra VM, no goroutine.
|
||||
func (ps *Plugins) startRuntime(name, source string, spec scheduleSpec) *PluginRuntime {
|
||||
if len(spec.tasks) == 0 || ps.vm == nil {
|
||||
// The scheduler runs on p.state — the SAME state a hook runs on. A plugin with
|
||||
// no schedule gets no runtime at all, so the common case stays exactly as cheap
|
||||
// as before: one VM, no extra goroutine.
|
||||
func (ps *Plugins) startRuntime(p *Plugin, spec scheduleSpec) *PluginRuntime {
|
||||
if len(spec.tasks) == 0 {
|
||||
return nil
|
||||
}
|
||||
rt := &PluginRuntime{
|
||||
plugin: name,
|
||||
source: source,
|
||||
ps: ps,
|
||||
spec: spec,
|
||||
running: make(map[string]bool, len(spec.tasks)),
|
||||
nextDue: make(map[string]time.Time, len(spec.tasks)),
|
||||
failures: make(map[string]int, len(spec.tasks)),
|
||||
abandons: make(map[string]int, len(spec.tasks)),
|
||||
stop: make(chan struct{}),
|
||||
done: make(chan struct{}),
|
||||
plugin: p,
|
||||
ps: ps,
|
||||
spec: spec,
|
||||
nextDue: make(map[string]time.Time, len(spec.tasks)),
|
||||
running: make(map[string]bool, len(spec.tasks)),
|
||||
startedAt: make(map[string]time.Time, len(spec.tasks)),
|
||||
failures: make(map[string]int, len(spec.tasks)),
|
||||
stop: make(chan struct{}),
|
||||
done: make(chan struct{}),
|
||||
}
|
||||
gen := rt.bootGeneration(ps)
|
||||
if gen == nil {
|
||||
ps.logf("[scheduler] plugin %s: no generation could be started", name)
|
||||
return nil
|
||||
}
|
||||
// Store it, or every later reader (checkWedged, runDue, dropGeneration) sees
|
||||
// a nil generation and silently does nothing — the runtime would look alive
|
||||
// while never running a single task.
|
||||
rt.gen = gen
|
||||
for _, t := range spec.tasks {
|
||||
rt.nextDue[t.Fn] = rt.firstDue(t)
|
||||
}
|
||||
@ -228,42 +191,6 @@ func (ps *Plugins) startRuntime(name, source string, spec scheduleSpec) *PluginR
|
||||
return rt
|
||||
}
|
||||
|
||||
// bootGeneration creates a fresh disposable generation: a private VM with the
|
||||
// plugin compiled into a single-state pool.
|
||||
//
|
||||
// Splitting this out is what makes the drop-and-respawn model work: a wedged
|
||||
// generation is simply replaced by calling this again.
|
||||
func (rt *PluginRuntime) bootGeneration(ps *Plugins) *runtimeGen {
|
||||
vm := &VM{
|
||||
dir: "",
|
||||
pools: map[string]*adapterPool{},
|
||||
// No janitor: this pool holds one permanent worker that we manage
|
||||
// directly, so a reclaimer would be pure overhead.
|
||||
janitorStop: make(chan struct{}),
|
||||
}
|
||||
w, err := bootPluginState(rt.source, rt.plugin, ps.outboundAllowlist())
|
||||
if err != nil {
|
||||
ps.logf("[scheduler] plugin %s: private VM boot failed: %v", rt.plugin, err)
|
||||
return nil
|
||||
}
|
||||
// The Cond MUST be built from this pool's own mu. Handing it a fresh mutex
|
||||
// makes cond.Wait unlock a mutex it never locked, which Go reports as
|
||||
// "unlock of unlocked mutex" — a fatal error, not a recoverable one.
|
||||
pool := &adapterPool{
|
||||
name: "__pluginrt_" + rt.plugin,
|
||||
script: rt.source,
|
||||
static: staticInfo{name: rt.plugin, headers: map[string]string{}},
|
||||
pluginMode: true,
|
||||
maxW: 1,
|
||||
revision: 1,
|
||||
}
|
||||
pool.cond = sync.NewCond(&pool.mu)
|
||||
pool.idle = []*worker{w}
|
||||
pool.created = 1
|
||||
vm.pools[pool.name] = pool
|
||||
return &runtimeGen{vm: vm, poolName: pool.name, busy: map[string]bool{}}
|
||||
}
|
||||
|
||||
// firstDue computes when a task first runs. Jitter spreads the first fire so
|
||||
// plugins do not all wake at process start.
|
||||
func (rt *PluginRuntime) firstDue(t ScheduledTask) time.Time {
|
||||
@ -308,136 +235,70 @@ func (rt *PluginRuntime) loop() {
|
||||
for {
|
||||
select {
|
||||
case <-rt.stop:
|
||||
// Abandon any in-flight generation rather than waiting for it: a
|
||||
// wedged task will never return, and Stop must not block forever.
|
||||
rt.dropGeneration()
|
||||
return
|
||||
case <-tick.C:
|
||||
rt.checkWedged()
|
||||
rt.ensureGeneration(rt.ps)
|
||||
rt.runDue()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// checkWedged abandons a generation whose task has overrun wedgedAfter.
|
||||
// checkWedged disarms a plugin whose task has overrun wedgedAfter.
|
||||
//
|
||||
// This is the heart of the model: instead of trying (and failing) to interrupt
|
||||
// the Lua call, the runtime stops trusting that generation and boots a clean
|
||||
// one. The abandoned goroutine keeps its own VM and will never touch the new
|
||||
// generation's state.
|
||||
// Capability note (measured, not assumed): a running Lua call CANNOT be
|
||||
// interrupted in this golua+LuaJIT build.
|
||||
//
|
||||
// 1. SetExecutionLimit(200000) against a tight loop — still running after 10s.
|
||||
// 2. A count hook + RaiseError against a tight loop — the hook never fired
|
||||
// (LuaJIT compiles the loop to machine code; instruction hooks only fire on
|
||||
// the interpreter path).
|
||||
// 3. A count hook against a loop with calls/table writes — fired 20971 times,
|
||||
// but RaiseError still did not return: it calls L.StackTrace(), which is the
|
||||
// known SIGSEGV path in vm.go.
|
||||
//
|
||||
// So the response cannot be "stop it". It is: stop trusting it, stop scheduling
|
||||
// more work, and make the condition visible. The state is deliberately NOT
|
||||
// rebuilt — a fresh VM would discard exactly the accumulated state (billing
|
||||
// totals, check-in history) that makes the plugin worth running.
|
||||
func (rt *PluginRuntime) checkWedged() {
|
||||
rt.mu.Lock()
|
||||
gen := rt.gen
|
||||
if gen == nil {
|
||||
if rt.disarmed {
|
||||
rt.mu.Unlock()
|
||||
return
|
||||
}
|
||||
// Collect the tasks that are both in flight AND past the wedge threshold.
|
||||
// A task that is merely slow (in flight but within wedgedAfter) must NOT
|
||||
// count: treating it as wedged would discard a healthy generation and could
|
||||
// start a second copy of work that is still legitimately running.
|
||||
var wedged []string
|
||||
for fn, busy := range gen.busy {
|
||||
if !busy {
|
||||
continue
|
||||
for fn, at := range rt.startedAt {
|
||||
if time.Since(at) >= wedgedAfter {
|
||||
wedged = append(wedged, fn)
|
||||
}
|
||||
started, ok := rt.startedAt[fn]
|
||||
if !ok || time.Since(started) < wedgedAfter {
|
||||
continue
|
||||
}
|
||||
wedged = append(wedged, fn)
|
||||
}
|
||||
if len(wedged) == 0 {
|
||||
rt.mu.Unlock()
|
||||
return
|
||||
}
|
||||
var disarmedNow bool
|
||||
rt.disarmed = true
|
||||
rt.mu.Unlock()
|
||||
|
||||
for _, fn := range wedged {
|
||||
rt.abandons[fn]++
|
||||
rt.failures[fn]++
|
||||
n := rt.abandons[fn]
|
||||
if n >= maxAbandons {
|
||||
rt.disarmed = true
|
||||
disarmedNow = true
|
||||
ps := rt.ps
|
||||
if ps != nil {
|
||||
ps.hookErr.note(StageRequestEnd, fmt.Sprintf("%s: task %s overran %s; plugin disabled",
|
||||
rt.pluginInfoName(), fn, wedgedAfter))
|
||||
}
|
||||
rt.logf("[scheduler] plugin %s task %s exceeded %s; abandoning its VM (abandon #%d/%d)",
|
||||
rt.plugin, fn, wedgedAfter, n, maxAbandons)
|
||||
log.Printf("[scheduler] plugin %s task %s exceeded %s; disabling the plugin "+
|
||||
"(reload it to retry)", rt.pluginInfoName(), fn, wedgedAfter)
|
||||
}
|
||||
rt.mu.Unlock()
|
||||
if disarmedNow {
|
||||
rt.logf("[scheduler] plugin %s disarmed after %d wedges; reload the plugin to retry",
|
||||
rt.plugin, maxAbandons)
|
||||
}
|
||||
// Drop the generation: the stuck goroutine keeps its own VM, and a clean one
|
||||
// is booted by ensureGeneration on the next tick. The in-flight tasks are
|
||||
// carried over so the counter can advance on the replacement.
|
||||
rt.dropGeneration()
|
||||
}
|
||||
|
||||
// dropGeneration replaces the current generation with a fresh one, leaving the
|
||||
// old (possibly wedged) goroutine holding its own VM.
|
||||
//
|
||||
// The wedged tasks are carried over and marked due NOW: the point of a fresh
|
||||
// generation is to re-run that work against clean state, and if we simply forgot
|
||||
// them the abandon counter would never advance and the circuit breaker could
|
||||
// never trip. The old generation's goroutine is deliberately NOT closed — the
|
||||
// stuck Lua call still owns its state, and closing it underneath would be a
|
||||
// use-after-free rather than a clean teardown.
|
||||
func (rt *PluginRuntime) dropGeneration() {
|
||||
rt.mu.Lock()
|
||||
old := rt.gen
|
||||
if old == nil {
|
||||
rt.mu.Unlock()
|
||||
return
|
||||
}
|
||||
// Collect the tasks that were in flight on that generation.
|
||||
var carry []string
|
||||
for fn, busy := range old.busy {
|
||||
if busy {
|
||||
carry = append(carry, fn)
|
||||
}
|
||||
}
|
||||
rt.gen = nil
|
||||
rt.running = make(map[string]bool, len(rt.spec.tasks))
|
||||
rt.startedAt = map[string]time.Time{}
|
||||
// Re-arm the carried tasks for immediate dispatch on the new generation.
|
||||
for _, fn := range carry {
|
||||
rt.nextDue[fn] = time.Now()
|
||||
}
|
||||
rt.pendingRespawn = carry
|
||||
rt.mu.Unlock()
|
||||
_ = old // kept alive by the stuck goroutine; the GC reclaims it if it returns
|
||||
}
|
||||
|
||||
// ensureGeneration boots a replacement when the previous one was dropped.
|
||||
func (rt *PluginRuntime) ensureGeneration(ps *Plugins) {
|
||||
rt.mu.Lock()
|
||||
need := rt.gen == nil && !rt.disarmed
|
||||
rt.mu.Unlock()
|
||||
if !need {
|
||||
return
|
||||
}
|
||||
gen := rt.bootGeneration(ps)
|
||||
if gen == nil {
|
||||
return
|
||||
}
|
||||
rt.mu.Lock()
|
||||
if rt.gen == nil {
|
||||
rt.gen = gen
|
||||
}
|
||||
rt.mu.Unlock()
|
||||
}
|
||||
|
||||
// runDue launches every task whose due time has arrived. Each task runs in its
|
||||
// own goroutine, so one long task never delays another's due time.
|
||||
// runDue dispatches every task whose due time has arrived. Each task runs in its
|
||||
// own goroutine so that one task waiting on the plugin lock does not delay the
|
||||
// due time of another — the lock, not the scheduler, is what serialises them.
|
||||
func (rt *PluginRuntime) runDue() {
|
||||
rt.mu.Lock()
|
||||
if rt.disarmed || rt.gen == nil {
|
||||
if rt.disarmed {
|
||||
rt.mu.Unlock()
|
||||
return
|
||||
}
|
||||
gen := rt.gen
|
||||
now := time.Now()
|
||||
var due []ScheduledTask
|
||||
for _, t := range rt.spec.tasks {
|
||||
@ -448,16 +309,13 @@ func (rt *PluginRuntime) runDue() {
|
||||
continue
|
||||
}
|
||||
rt.running[t.Fn] = true
|
||||
if rt.startedAt == nil {
|
||||
rt.startedAt = map[string]time.Time{}
|
||||
}
|
||||
rt.startedAt[t.Fn] = now
|
||||
rt.active++
|
||||
rt.nextDue[t.Fn] = rt.nextAfter(t, now)
|
||||
due = append(due, t)
|
||||
}
|
||||
rt.mu.Unlock()
|
||||
for _, t := range due {
|
||||
go rt.runOne(gen, t)
|
||||
go rt.runOne(t)
|
||||
}
|
||||
}
|
||||
|
||||
@ -473,80 +331,104 @@ func (rt *PluginRuntime) nextAfter(t ScheduledTask, now time.Time) time.Time {
|
||||
return now.Add(d)
|
||||
}
|
||||
|
||||
// runOne executes one task on the given generation.
|
||||
// runOne executes one task on the plugin's own VM, under the plugin's own lock.
|
||||
//
|
||||
// A task that never returns leaves this goroutine (and the generation it holds)
|
||||
// behind; that is the accepted cost of the drop model, bounded by the abandon
|
||||
// counter.
|
||||
func (rt *PluginRuntime) runOne(gen *runtimeGen, t ScheduledTask) {
|
||||
if gen == nil {
|
||||
// The generation was already abandoned between selection and dispatch.
|
||||
// Nothing to run on — clear the in-flight flag so the next cycle can
|
||||
// schedule this task against the replacement generation.
|
||||
rt.mu.Lock()
|
||||
delete(rt.running, t.Fn)
|
||||
delete(rt.startedAt, t.Fn)
|
||||
rt.mu.Unlock()
|
||||
return
|
||||
}
|
||||
rt.mu.Lock()
|
||||
gen.busy[t.Fn] = true
|
||||
rt.running[t.Fn] = true
|
||||
if rt.startedAt == nil {
|
||||
rt.startedAt = map[string]time.Time{}
|
||||
}
|
||||
// Recorded here, not in runDue: any path that dispatches a task must have it
|
||||
// tracked, or checkWedged cannot tell "still running" from "never started".
|
||||
rt.startedAt[t.Fn] = time.Now()
|
||||
rt.mu.Unlock()
|
||||
// Holding that lock for the whole call is what makes state sharing correct — and
|
||||
// it is why a task that never returns wedges its own plugin. That trade is
|
||||
// deliberate: the alternative (a private VM per task) silently breaks the state
|
||||
// sharing this feature exists for.
|
||||
func (rt *PluginRuntime) runOne(t ScheduledTask) {
|
||||
p := rt.plugin
|
||||
defer func() {
|
||||
rt.mu.Lock()
|
||||
delete(gen.busy, t.Fn)
|
||||
delete(rt.running, t.Fn)
|
||||
delete(rt.startedAt, t.Fn)
|
||||
rt.active--
|
||||
rt.mu.Unlock()
|
||||
}()
|
||||
|
||||
p := gen.vm.pool(gen.poolName)
|
||||
if p == nil {
|
||||
rt.recordFailure(t.Fn, "no pool")
|
||||
rt.recordFailure(t.Fn, "plugin is gone")
|
||||
return
|
||||
}
|
||||
w, err := p.acquire()
|
||||
if err != nil {
|
||||
rt.recordFailure(t.Fn, err.Error())
|
||||
return
|
||||
}
|
||||
defer p.release(w)
|
||||
|
||||
// No interrupt is attempted: see the capability note. The deadline is
|
||||
// enforced by checkWedged at the generation level instead.
|
||||
if err := callFnOn(w.L, t.Fn); err != nil {
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
if p.state == nil || p.state.L == nil {
|
||||
rt.recordFailure(t.Fn, "no plugin state")
|
||||
return
|
||||
}
|
||||
|
||||
// Recorded only now: the time before the lock was acquired was spent queued
|
||||
// behind another caller, and counting it would report a healthy hook as a
|
||||
// wedged task.
|
||||
rt.mu.Lock()
|
||||
rt.startedAt[t.Fn] = time.Now()
|
||||
rt.mu.Unlock()
|
||||
|
||||
// No interrupt is attempted; see the capability note on checkWedged.
|
||||
if err := callFnOn(p.state.L, t.Fn); err != nil {
|
||||
rt.recordFailure(t.Fn, err.Error())
|
||||
return
|
||||
}
|
||||
// A task mutates plugin.state exactly as a hook does, so it must dirty the
|
||||
// same way — otherwise a rotated token would live only in memory until some
|
||||
// unrelated request happened to flush it.
|
||||
if ps := rt.ps; ps != nil {
|
||||
ps.markDirtyLocked(p)
|
||||
}
|
||||
|
||||
rt.mu.Lock()
|
||||
rt.failures[t.Fn] = 0
|
||||
rt.abandons[t.Fn] = 0
|
||||
rt.mu.Unlock()
|
||||
}
|
||||
|
||||
// recordFailure notes a task error. A failure is visible in the hook-error
|
||||
// surface so a silently broken task cannot hide.
|
||||
// recordFailure notes a task error so a silently broken task cannot hide.
|
||||
func (rt *PluginRuntime) recordFailure(fn, msg string) {
|
||||
rt.mu.Lock()
|
||||
rt.failures[fn]++
|
||||
rt.mu.Unlock()
|
||||
rt.logf("[scheduler] plugin %s task %s failed: %s", rt.plugin, fn, msg)
|
||||
}
|
||||
|
||||
func (rt *PluginRuntime) logf(format string, args ...interface{}) {
|
||||
if f := activeLogger.Load(); f != nil {
|
||||
(*f)(format, args...)
|
||||
if ps := rt.ps; ps != nil {
|
||||
ps.hookErr.note(StageRequestEnd, fmt.Sprintf("%s: task %s failed: %s",
|
||||
rt.pluginInfoName(), fn, msg))
|
||||
}
|
||||
log.Printf("[scheduler] plugin %s task %s failed: %s", rt.pluginInfoName(), fn, msg)
|
||||
}
|
||||
|
||||
// Stop halts the scheduler and abandons the current generation.
|
||||
func (rt *PluginRuntime) pluginInfoName() string {
|
||||
if rt.plugin == nil {
|
||||
return "?"
|
||||
}
|
||||
return rt.plugin.Info.Name
|
||||
}
|
||||
|
||||
// InFlight reports whether any task goroutine has not yet returned. Unload uses
|
||||
// it to decide whether closing the Lua state is safe: a task holds the state for
|
||||
// the duration of its call, and a wedged one never releases it, so closing
|
||||
// underneath would be a use-after-free in the Lua C layer (which aborts the
|
||||
// process rather than returning an error). Skipping the close leaks that one
|
||||
// state instead — bounded by the wedge, and strictly better than crashing.
|
||||
func (rt *PluginRuntime) InFlight() bool {
|
||||
if rt == nil {
|
||||
return false
|
||||
}
|
||||
rt.mu.Lock()
|
||||
defer rt.mu.Unlock()
|
||||
return rt.active > 0
|
||||
}
|
||||
|
||||
// Disarmed reports whether the plugin was disabled by an overrunning task.
|
||||
func (rt *PluginRuntime) Disarmed() bool {
|
||||
if rt == nil {
|
||||
return false
|
||||
}
|
||||
rt.mu.Lock()
|
||||
defer rt.mu.Unlock()
|
||||
return rt.disarmed
|
||||
}
|
||||
|
||||
// Stop halts the scheduler. It does NOT wait for in-flight tasks: a wedged task
|
||||
// will never return, and Stop must not block unload forever.
|
||||
func (rt *PluginRuntime) Stop() {
|
||||
if rt == nil {
|
||||
return
|
||||
@ -555,26 +437,12 @@ func (rt *PluginRuntime) Stop() {
|
||||
<-rt.done
|
||||
}
|
||||
|
||||
// activeLogger is installed by the Plugins layer so a runtime goroutine can log
|
||||
// without capturing the parent (which would keep it alive past Close).
|
||||
var activeLogger atomic.Pointer[func(string, ...interface{})]
|
||||
|
||||
// defaultTaskTimeout documents the intended bound for one scheduled task. It is
|
||||
// NOT enforced by the Lua runtime (see the capability note at the top of this
|
||||
// file); it exists so the declaration and the plan agree on the number a future
|
||||
// implementation should honour.
|
||||
const defaultTaskTimeout = 5 * time.Minute
|
||||
|
||||
// callFnOn invokes a plugin function on an ALREADY-CHECKED-OUT state.
|
||||
// callFnOn invokes a plugin function on an already-locked state.
|
||||
//
|
||||
// It is the runtime's counterpart to VM.Transform: Transform takes its own
|
||||
// checkout, which would deadlock here because the runtime holds the only worker
|
||||
// of a single-state pool. The stack discipline mirrors Transform exactly:
|
||||
// [global, guard, fn] then call with one argument.
|
||||
//
|
||||
// A return value other than a string is deliberately discarded: a scheduled
|
||||
// task's output is its side effect (a token written, an adapter updated), not a
|
||||
// value the scheduler forwards anywhere.
|
||||
// The stack discipline mirrors Plugins.invoke: [global, guard, fn] then call
|
||||
// with one argument. A return value other than an error string is discarded —
|
||||
// a scheduled task's output is its side effect (a token written, an adapter
|
||||
// updated), not a value the scheduler forwards anywhere.
|
||||
func callFnOn(L *golua.State, fn string) error {
|
||||
L.SetTop(0)
|
||||
defer L.SetTop(0)
|
||||
@ -605,16 +473,6 @@ func callFnOn(L *golua.State, fn string) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// runInterruptible runs fn. It currently performs NO interruption — see the
|
||||
// capability note at the top of this file for the three measurements showing
|
||||
// that neither SetExecutionLimit nor a count-hook RaiseError stops a LuaJIT
|
||||
// loop in this build. The name and signature are kept so that the call site
|
||||
// stays correct once a real interrupt mechanism exists: only this function
|
||||
// would change.
|
||||
func runInterruptible(L *golua.State, timeout time.Duration, fn func() error) error {
|
||||
return fn()
|
||||
}
|
||||
|
||||
// randInt63n returns a pseudo-random value in [0,n) for schedule jitter. Quality
|
||||
// is irrelevant here — it only needs to differ across processes.
|
||||
func randInt63n(n int64) int64 {
|
||||
|
||||
@ -2,19 +2,16 @@ package lua
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// 本文件验证 R4-b 的核心主张:**插件声明的周期任务在没有任何浏览器、
|
||||
// 没有任何外部触发时会自己执行**。
|
||||
// 没有任何外部触发时会自己执行**,并且跑在**插件自己的那个 VM** 上。
|
||||
//
|
||||
// 这正是「自动签到」这类能力的死穴:旧模型下插件的周期工作只能靠前端
|
||||
// pluginAPI.call 驱动,页面没开就一次都不跑。这里用「内核 ticker + 独立 VM」
|
||||
// 替换那个机制。
|
||||
// 第二点是这个文件最重要的断言。早期实现给定时任务开了一个私有 VM,结果
|
||||
// 任务写 plugin.state 时写进了一个请求路径永远读不到的副本 —— token 轮换
|
||||
// 看起来成功了,对实际流量完全无效。TestTaskWriteIsVisibleToHooks 锁死这一点。
|
||||
|
||||
func shortTick(t *testing.T) {
|
||||
t.Helper()
|
||||
@ -23,166 +20,131 @@ func shortTick(t *testing.T) {
|
||||
t.Cleanup(func() { runtimeTick = old })
|
||||
}
|
||||
|
||||
// runtimePlugins 建一个带插件目录的 Plugins,并把 outbound allowlist 设好,
|
||||
// 让插件的 HTTP 调用在白名单内可通。
|
||||
// runtimePlugins 建一个带插件目录的 Plugins。
|
||||
func runtimePlugins(t *testing.T) (*VM, *Plugins, string) {
|
||||
t.Helper()
|
||||
dir := filepath.Join(t.TempDir(), "adapters")
|
||||
dir := fmt.Sprintf("%s/adapters", t.TempDir())
|
||||
vm := NewVM(dir)
|
||||
if err := vm.Start(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(vm.Stop)
|
||||
pdir := filepath.Join(t.TempDir(), "plugins")
|
||||
pdir := fmt.Sprintf("%s/plugins", t.TempDir())
|
||||
ps := NewPlugins(vm, pdir)
|
||||
ps.SetOutboundHosts([]string{"127.0.0.1"})
|
||||
return vm, ps, pdir
|
||||
}
|
||||
|
||||
// runtimeOf 取某个插件的 Runtime,最多等 wait。
|
||||
func runtimeOf(t *testing.T, ps *Plugins, name string, wait time.Duration) *PluginRuntime {
|
||||
t.Helper()
|
||||
deadline := time.Now().Add(wait)
|
||||
for time.Now().Before(deadline) {
|
||||
ps.mu.RLock()
|
||||
for _, p := range ps.plugins {
|
||||
if p.Info.Name == name {
|
||||
rt := p.Runtime
|
||||
ps.mu.RUnlock()
|
||||
if rt != nil {
|
||||
return rt
|
||||
}
|
||||
goto sleep
|
||||
}
|
||||
}
|
||||
ps.mu.RUnlock()
|
||||
sleep:
|
||||
time.Sleep(5 * time.Millisecond)
|
||||
}
|
||||
t.Fatalf("插件 %s 未获得运行时", name)
|
||||
return nil
|
||||
}
|
||||
|
||||
// TestScheduledTaskFiresWithoutAnyTrigger 是主断言。
|
||||
//
|
||||
// 插件声明 1 秒一次的任务,测试**不做任何外部调用**,只等 —— 任务必须自己跑。
|
||||
// 插件声明一个极短周期的任务,测试**不做任何外部调用**,只等 —— 任务必须自己跑。
|
||||
func TestScheduledTaskFiresWithoutAnyTrigger(t *testing.T) {
|
||||
shortTick(t)
|
||||
_, ps, _ := runtimePlugins(t)
|
||||
|
||||
// 用一个可观测的副作用:任务把标记写进自己的状态文件(独立 VM 不共享
|
||||
// plugin.state,所以走文件是最忠实的观测方式)。
|
||||
marker := filepath.Join(t.TempDir(), "fired.marker")
|
||||
code := `
|
||||
local plugin = { name = "checkin", version = "1.0" }
|
||||
plugin.state = { fired = 0 }
|
||||
plugin.schedule = { { fn = "do_checkin", every = 1 } }
|
||||
local MARK = ` + luaQuote(marker) + `
|
||||
function plugin.do_checkin()
|
||||
local f = __llmsproxy_http and nil
|
||||
-- 用 log 之外的可观测副作用:追加一行到一个文件不可用(无 io 库),
|
||||
-- 所以用 state —— 见下面的断言方式。
|
||||
plugin.state = plugin.state or { fired = 0 }
|
||||
plugin.state.fired = (plugin.state.fired or 0) + 1
|
||||
return nil
|
||||
plugin.state.fired = plugin.state.fired + 1
|
||||
end
|
||||
return plugin
|
||||
`
|
||||
if err := ps.LoadSource("checkin", code); err != nil {
|
||||
t.Fatalf("load: %v", err)
|
||||
}
|
||||
// 等到运行时就位
|
||||
rt := runtimeOf(t, ps, "checkin", 3*time.Second)
|
||||
|
||||
// 只等,不触发任何东西。
|
||||
deadline := time.Now().Add(3 * time.Second)
|
||||
var rt *PluginRuntime
|
||||
for time.Now().Before(deadline) {
|
||||
ps.mu.RLock()
|
||||
for _, p := range ps.plugins {
|
||||
if p.Info.Name == "checkin" {
|
||||
rt = p.Runtime
|
||||
rt.mu.Lock()
|
||||
n := rt.failures["do_checkin"]
|
||||
rt.mu.Unlock()
|
||||
if n > 0 {
|
||||
t.Fatalf("任务执行失败 %d 次", n)
|
||||
}
|
||||
// 用插件的状态作为"真的跑了"的证据。
|
||||
if st, ok := ps.State("checkin").(map[string]interface{}); ok {
|
||||
if f, ok := st["fired"].(float64); ok && f >= 1 {
|
||||
return // 通过:没有任何外部触发,任务自己跑了
|
||||
}
|
||||
}
|
||||
ps.mu.RUnlock()
|
||||
if rt != nil {
|
||||
break
|
||||
}
|
||||
time.Sleep(5 * time.Millisecond)
|
||||
}
|
||||
if rt == nil {
|
||||
t.Fatal("声明了 schedule 的插件没有获得运行时")
|
||||
}
|
||||
|
||||
// 不做任何外部触发,只等任务自己跑。用 failures 与 running 状态推断
|
||||
// 它至少被执行过一次:running 为空且时间已过 due。
|
||||
if len(rt.spec.tasks) != 1 {
|
||||
t.Fatalf("任务数 = %d, want 1", len(rt.spec.tasks))
|
||||
}
|
||||
if rt.spec.tasks[0].Fn != "do_checkin" {
|
||||
t.Fatalf("任务名 = %q", rt.spec.tasks[0].Fn)
|
||||
}
|
||||
// 等到第一次 due 通过(every=1s,但我们用短 tick 只影响轮询频率;
|
||||
// every 仍是 1,所以这里等 1.2s)
|
||||
time.Sleep(1200 * time.Millisecond)
|
||||
rt.mu.Lock()
|
||||
fails := rt.failures["do_checkin"]
|
||||
rt.mu.Unlock()
|
||||
if fails > 0 {
|
||||
t.Fatalf("任务失败了 %d 次(应为 0)", fails)
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
}
|
||||
t.Fatal("任务在没有任何外部触发的情况下没有执行")
|
||||
}
|
||||
|
||||
// TestScheduledTaskActuallyRuns 用独立的可观测面(task 计数器经 state 暴露)
|
||||
// 证明任务体真的被执行了,而不是仅仅"没有失败"。
|
||||
func TestScheduledTaskActuallyRuns(t *testing.T) {
|
||||
// TestTaskWriteIsVisibleToHooks 是这个模型的关键正确性断言。
|
||||
//
|
||||
// 定时任务写 plugin.state.token,请求路径(同一个 VM 上的钩子)必须能读到
|
||||
// 它。早期"每个任务一个私有 VM"的实现会在这里失败 —— 任务写进副本,钩子
|
||||
// 看到的还是旧值,token 轮换静默失效。
|
||||
func TestTaskWriteIsVisibleToHooks(t *testing.T) {
|
||||
shortTick(t)
|
||||
_, ps, _ := runtimePlugins(t)
|
||||
|
||||
// 任务把执行次数写进插件 state;runtime 是独立 VM,故我们从它的 VM 读回。
|
||||
code := `
|
||||
local plugin = { name = "counter", version = "1.0" }
|
||||
plugin.schedule = { { fn = "tick", every = 1 } }
|
||||
function plugin.tick()
|
||||
plugin.state = plugin.state or { n = 0 }
|
||||
plugin.state.n = (plugin.state.n or 0) + 1
|
||||
return plugin.state
|
||||
local plugin = { name = "rotator", version = "1.0" }
|
||||
plugin.state = { token = "OLD" }
|
||||
plugin.schedule = { { fn = "rotate", every = 1 } }
|
||||
function plugin.rotate()
|
||||
plugin.state.token = "NEW"
|
||||
end
|
||||
-- keep 返回字符串,便于测试从 Transform 的字符串返回值里读回计数
|
||||
function plugin.keep()
|
||||
return tostring((plugin.state and plugin.state.n) or 0)
|
||||
plugin.hooks = { routed = "read_token" }
|
||||
function plugin.read_token(payload)
|
||||
payload.token = plugin.state.token -- 请求路径读到的值
|
||||
return payload
|
||||
end
|
||||
return plugin
|
||||
`
|
||||
if err := ps.LoadSource("counter", code); err != nil {
|
||||
if err := ps.LoadSource("rotator", code); err != nil {
|
||||
t.Fatalf("load: %v", err)
|
||||
}
|
||||
time.Sleep(1400 * time.Millisecond)
|
||||
runtimeOf(t, ps, "rotator", 3*time.Second)
|
||||
|
||||
// 直接在运行时的私有 VM 上调用一次 tick 读回计数
|
||||
// (独立 VM 的状态不在共享 plugin.state 里,这是设计使然)。
|
||||
ps.mu.RLock()
|
||||
var rt *PluginRuntime
|
||||
for _, p := range ps.plugins {
|
||||
if p.Info.Name == "counter" {
|
||||
rt = p.Runtime
|
||||
// 等任务把 token 轮换掉。
|
||||
deadline := time.Now().Add(3 * time.Second)
|
||||
for time.Now().Before(deadline) {
|
||||
out := ps.Fire(StageRouted, map[string]interface{}{})
|
||||
if got, _ := out["token"].(string); got == "NEW" {
|
||||
return // 通过:钩子读到了任务写进去的值 —— 同一个 VM
|
||||
}
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
}
|
||||
ps.mu.RUnlock()
|
||||
if rt == nil {
|
||||
t.Fatal("无运行时")
|
||||
}
|
||||
|
||||
// Transform 只返回 Lua **字符串** 返回值;tick 返回的是 table,故 out 为空
|
||||
// 不代表任务没跑。真正的观测点是 failures:任务体若抛错会被计入。
|
||||
rt.mu.Lock()
|
||||
gen := rt.gen
|
||||
rt.mu.Unlock()
|
||||
if gen == nil {
|
||||
t.Fatal("无 generation")
|
||||
}
|
||||
if _, err := gen.vm.Transform(gen.poolName, "tick", "{}"); err != nil {
|
||||
t.Fatalf("tick: %v", err)
|
||||
}
|
||||
rt.mu.Lock()
|
||||
fails := rt.failures["tick"]
|
||||
rt.mu.Unlock()
|
||||
if fails > 0 {
|
||||
t.Errorf("tick 失败 %d 次", fails)
|
||||
}
|
||||
// 用 keep 函数读回计数:任务把 n 累加在 state 里,读它证明任务体确实执行过。
|
||||
out, err := gen.vm.Transform(gen.poolName, "keep", "{}")
|
||||
if err != nil {
|
||||
t.Fatalf("keep: %v", err)
|
||||
}
|
||||
var n int
|
||||
if _, err := fmt.Sscanf(out, "%d", &n); err != nil {
|
||||
t.Fatalf("keep 返回不可解析: %q (%v)", out, err)
|
||||
}
|
||||
if n < 1 {
|
||||
t.Errorf("任务执行计数 n = %d, want >=1(说明任务体从未真正运行)", n)
|
||||
}
|
||||
t.Logf("任务已执行 %d 次", n)
|
||||
out := ps.Fire(StageRouted, map[string]interface{}{})
|
||||
t.Fatalf("定时任务写的 token 对请求钩子不可见(钩子读到 %v)—— 说明任务跑在了另一个 VM 上", out["token"])
|
||||
}
|
||||
|
||||
// TestNoScheduleMeansNoRuntime:没声明 schedule 的插件不该有额外 VM/goroutine。
|
||||
// TestNoScheduleMeansNoRuntime:没声明 schedule 就不该有额外 VM/goroutine。
|
||||
func TestNoScheduleMeansNoRuntime(t *testing.T) {
|
||||
_, ps, _ := runtimePlugins(t)
|
||||
code := `
|
||||
local plugin = { name = "plain", version = "1.0" }
|
||||
function plugin.on_request_end(p) return nil end
|
||||
return plugin
|
||||
`
|
||||
if err := ps.LoadSource("plain", code); err != nil {
|
||||
@ -191,374 +153,219 @@ return plugin
|
||||
ps.mu.RLock()
|
||||
defer ps.mu.RUnlock()
|
||||
for _, p := range ps.plugins {
|
||||
if p.Info.Name == "plain" {
|
||||
if p.Runtime != nil {
|
||||
t.Error("未声明 schedule 的插件不应有运行时(零开销路径)")
|
||||
}
|
||||
if len(p.Schedule) != 0 {
|
||||
t.Errorf("Schedule 应为空, got %v", p.Schedule)
|
||||
}
|
||||
return
|
||||
if p.Info.Name == "plain" && p.Runtime != nil {
|
||||
t.Error("未声明 schedule 的插件不应有运行时")
|
||||
}
|
||||
}
|
||||
t.Fatal("插件未加载")
|
||||
}
|
||||
|
||||
// TestDailyAtScheduleParsed:`at = "HH:MM"` 的解析与首次触发时间计算。
|
||||
func TestDailyAtScheduleParsed(t *testing.T) {
|
||||
// 直接测 nextDailyOccurrence:今天的时刻若已过,应落到明天。
|
||||
now := time.Date(2026, 3, 10, 14, 0, 0, 0, time.Local)
|
||||
got := nextDailyOccurrence("03:30", now)
|
||||
if got.Day() != 11 || got.Hour() != 3 || got.Minute() != 30 {
|
||||
t.Errorf("过了今天 03:30,应落到明天 03:30,实得 %v", got)
|
||||
}
|
||||
got2 := nextDailyOccurrence("23:00", now)
|
||||
if got2.Day() != 10 || got2.Hour() != 23 {
|
||||
t.Errorf("未到 23:00,应在今天,实得 %v", got2)
|
||||
}
|
||||
// 畸形输入必须退化为每小时,而不是卡死
|
||||
got3 := nextDailyOccurrence("nonsense", now)
|
||||
if !got3.After(now) {
|
||||
t.Errorf("畸形时刻应退化为未来某个时间,实得 %v", got3)
|
||||
}
|
||||
}
|
||||
|
||||
// TestScheduleParsingRejectsIncomplete:缺 fn / 缺周期 的条目被忽略。
|
||||
func TestScheduleParsingRejectsIncomplete(t *testing.T) {
|
||||
// TestScheduleParsing:every / at / jitter / timeout 解析。
|
||||
func TestScheduleParsing(t *testing.T) {
|
||||
_, ps, _ := runtimePlugins(t)
|
||||
code := `
|
||||
local plugin = { name = "partial", version = "1.0" }
|
||||
local plugin = { name = "sched", version = "1.0" }
|
||||
plugin.schedule = {
|
||||
{ fn = "good", every = 60 },
|
||||
{ every = 60 }, -- 缺 fn
|
||||
{ fn = "noPeriod" }, -- 缺 every 且无 at
|
||||
{ fn = "alsoGood", at = "03:00" },
|
||||
{ fn = "a", every = 60 },
|
||||
{ fn = "b", at = "03:30" },
|
||||
{ fn = "c", every = 300, jitter = 30 },
|
||||
{ fn = "d" }, -- 无 every 无 at → 丢弃
|
||||
{ every = 10 }, -- 无 fn → 丢弃
|
||||
}
|
||||
return plugin
|
||||
`
|
||||
if err := ps.LoadSource("partial", code); err != nil {
|
||||
if err := ps.LoadSource("sched", code); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ps.mu.RLock()
|
||||
defer ps.mu.RUnlock()
|
||||
var sched []ScheduledTask
|
||||
for _, p := range ps.plugins {
|
||||
if p.Info.Name == "partial" {
|
||||
if len(p.Schedule) != 2 {
|
||||
t.Fatalf("应只接受 2 条合法任务, got %d: %+v", len(p.Schedule), p.Schedule)
|
||||
}
|
||||
names := p.Schedule
|
||||
if names[0].Fn != "good" || names[1].Fn != "alsoGood" {
|
||||
t.Errorf("接受的任务 = %+v", names)
|
||||
}
|
||||
return
|
||||
if p.Info.Name == "sched" {
|
||||
sched = p.Schedule
|
||||
}
|
||||
}
|
||||
t.Fatal("插件未加载")
|
||||
ps.mu.RUnlock()
|
||||
if len(sched) != 3 {
|
||||
t.Fatalf("应解析出 3 个任务(a/b/c),实得 %d: %+v", len(sched), sched)
|
||||
}
|
||||
if sched[0].Every != 60 || sched[1].At != "03:30" || sched[2].Jitter != 30 {
|
||||
t.Errorf("字段解析错误: %+v", sched)
|
||||
}
|
||||
}
|
||||
|
||||
// TestScheduledTaskFailureIsIsolated:任务抛错被记录,且不影响其它任务。
|
||||
// TestDailyAtNextOccurrence:at="HH:MM" 的下一次触发时刻。
|
||||
func TestDailyAtNextOccurrence(t *testing.T) {
|
||||
now := time.Date(2026, 10, 11, 12, 0, 0, 0, time.Local)
|
||||
// 今天还没到 → 今天
|
||||
if got := nextDailyOccurrence("23:00", now); got.Day() != 11 || got.Hour() != 23 {
|
||||
t.Errorf("23:00 应为今天 23:00, 实得 %v", got)
|
||||
}
|
||||
// 今天已过 → 明天
|
||||
if got := nextDailyOccurrence("03:30", now); got.Day() != 12 || got.Hour() != 3 {
|
||||
t.Errorf("03:30 应为明天 03:30, 实得 %v", got)
|
||||
}
|
||||
// 非法值 → 兜底一小时(不能永久卡住任务)
|
||||
if got := nextDailyOccurrence("99:99", now); got.Sub(now) > 2*time.Hour {
|
||||
t.Errorf("非法 at 应兜底为约一小时, 实得 %v", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestScheduledTaskFailureIsIsolated:任务出错只记账,不影响钩子。
|
||||
func TestScheduledTaskFailureIsIsolated(t *testing.T) {
|
||||
shortTick(t)
|
||||
_, ps, _ := runtimePlugins(t)
|
||||
code := `
|
||||
local plugin = { name = "mixed", version = "1.0" }
|
||||
plugin.schedule = {
|
||||
{ fn = "boom", every = 1 },
|
||||
{ fn = "ok", every = 1 },
|
||||
}
|
||||
function plugin.boom() error("deliberate") end
|
||||
function plugin.ok() return nil end
|
||||
local plugin = { name = "flaky", version = "1.0" }
|
||||
plugin.schedule = { { fn = "boom", every = 1 } }
|
||||
function plugin.boom() error("任务故意失败") end
|
||||
return plugin
|
||||
`
|
||||
if err := ps.LoadSource("mixed", code); err != nil {
|
||||
if err := ps.LoadSource("flaky", code); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
time.Sleep(1300 * time.Millisecond)
|
||||
ps.mu.RLock()
|
||||
var rt *PluginRuntime
|
||||
for _, p := range ps.plugins {
|
||||
if p.Info.Name == "mixed" {
|
||||
rt = p.Runtime
|
||||
rt := runtimeOf(t, ps, "flaky", 3*time.Second)
|
||||
deadline := time.Now().Add(3 * time.Second)
|
||||
for time.Now().Before(deadline) {
|
||||
rt.mu.Lock()
|
||||
n := rt.failures["boom"]
|
||||
rt.mu.Unlock()
|
||||
if n > 0 {
|
||||
// 失败必须被记账,且插件未被熔断(错误 ≠ 卡死)
|
||||
if rt.Disarmed() {
|
||||
t.Error("普通错误不应熔断插件")
|
||||
}
|
||||
return
|
||||
}
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
}
|
||||
ps.mu.RUnlock()
|
||||
if rt == nil {
|
||||
t.Fatal("无运行时")
|
||||
}
|
||||
rt.mu.Lock()
|
||||
defer rt.mu.Unlock()
|
||||
if rt.failures["boom"] == 0 {
|
||||
t.Error("失败的任务应被计数(否则静默的坏任务是不可见的)")
|
||||
}
|
||||
if rt.failures["ok"] != 0 {
|
||||
t.Errorf("成功的任务不该被记为失败: %d", rt.failures["ok"])
|
||||
}
|
||||
t.Fatal("任务错误未被记账")
|
||||
}
|
||||
|
||||
// TestUnloadStopsRuntime:卸载插件必须停掉它的运行时(不泄漏 goroutine/VM)。
|
||||
// TestUnloadStopsRuntime:卸载后调度器 goroutine 必须停止。
|
||||
func TestUnloadStopsRuntime(t *testing.T) {
|
||||
shortTick(t)
|
||||
_, ps, _ := runtimePlugins(t)
|
||||
code := `
|
||||
local plugin = { name = "temp", version = "1.0" }
|
||||
plugin.schedule = { { fn = "t", every = 3600 } }
|
||||
function plugin.t() return nil end
|
||||
local plugin = { name = "stoppable", version = "1.0" }
|
||||
plugin.state = { n = 0 }
|
||||
plugin.schedule = { { fn = "tick", every = 1 } }
|
||||
function plugin.tick() plugin.state.n = plugin.state.n + 1 end
|
||||
return plugin
|
||||
`
|
||||
if err := ps.LoadSource("temp", code); err != nil {
|
||||
if err := ps.LoadSource("stoppable", code); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ps.mu.RLock()
|
||||
var rt *PluginRuntime
|
||||
for _, p := range ps.plugins {
|
||||
if p.Info.Name == "temp" {
|
||||
rt = p.Runtime
|
||||
}
|
||||
rt := runtimeOf(t, ps, "stoppable", 3*time.Second)
|
||||
if err := ps.Unload("stoppable"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ps.mu.RUnlock()
|
||||
if rt == nil {
|
||||
t.Fatal("无运行时")
|
||||
}
|
||||
if err := ps.Unload("temp"); err != nil {
|
||||
t.Fatalf("unload: %v", err)
|
||||
}
|
||||
// done 应已关闭
|
||||
// Stop 已返回 ⇒ done 已关闭 ⇒ loop 已退出
|
||||
select {
|
||||
case <-rt.done:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Error("卸载后调度 goroutine 未退出(泄漏)")
|
||||
default:
|
||||
t.Error("卸载后调度器 loop 未退出")
|
||||
}
|
||||
}
|
||||
|
||||
// TestConcurrentScheduledTasksStress 是压力测试:多个插件 × 多个任务并发跑,
|
||||
// 验证不 panic、不串任务、失败计数各自独立。
|
||||
func TestConcurrentScheduledTasksStress(t *testing.T) {
|
||||
shortTick(t)
|
||||
_, ps, _ := runtimePlugins(t)
|
||||
var total atomic.Int64
|
||||
for i := 0; i < 6; i++ {
|
||||
name := "stress" + string(rune('a'+i))
|
||||
code := `
|
||||
local plugin = { name = "` + name + `", version = "1.0" }
|
||||
plugin.schedule = {
|
||||
{ fn = "a", every = 1 },
|
||||
{ fn = "b", every = 1 },
|
||||
{ fn = "c", every = 1 },
|
||||
}
|
||||
function plugin.a() return nil end
|
||||
function plugin.b() return nil end
|
||||
function plugin.c() return nil end
|
||||
return plugin
|
||||
`
|
||||
if err := ps.LoadSource(name, code); err != nil {
|
||||
t.Fatalf("load %s: %v", name, err)
|
||||
}
|
||||
total.Add(1)
|
||||
}
|
||||
time.Sleep(1200 * time.Millisecond)
|
||||
// 所有任务都不该失败
|
||||
ps.mu.RLock()
|
||||
defer ps.mu.RUnlock()
|
||||
for _, p := range ps.plugins {
|
||||
if p.Runtime == nil {
|
||||
continue
|
||||
}
|
||||
p.Runtime.mu.Lock()
|
||||
for fn, n := range p.Runtime.failures {
|
||||
if n > 0 {
|
||||
t.Errorf("插件 %s 任务 %s 失败 %d 次", p.Info.Name, fn, n)
|
||||
}
|
||||
}
|
||||
p.Runtime.mu.Unlock()
|
||||
}
|
||||
}
|
||||
|
||||
func luaQuote(s string) string {
|
||||
out := []rune{'"'}
|
||||
for _, r := range s {
|
||||
if r == '"' || r == '\\' {
|
||||
out = append(out, '\\')
|
||||
}
|
||||
out = append(out, r)
|
||||
}
|
||||
return string(append(out, '"'))
|
||||
}
|
||||
|
||||
var _ = os.TempDir
|
||||
|
||||
// TestWedgedTaskAbandonsItsVM 是「抛弃式隔离」的核心测试。
|
||||
//
|
||||
// 前提(实测三次):本 LuaJIT 构建**无法中断**运行中的 Lua 调用。所以模型
|
||||
// 不是「中断它」,而是「不再等它」——把整个 generation(私有 VM + goroutine)
|
||||
// 丢掉,重新起一个干净的。卡死的 goroutine 连同它自己的 VM 一起被丢弃,
|
||||
// 永远不会碰到新一代的状态。
|
||||
func TestWedgedTaskAbandonsItsVM(t *testing.T) {
|
||||
// TestWedgedTaskDisarmsPlugin:任务超时后插件被停用(不重建 VM —— 重建会丢掉
|
||||
// 累计状态,那正是这个插件存在的意义)。
|
||||
func TestWedgedTaskDisarmsPlugin(t *testing.T) {
|
||||
shortTick(t)
|
||||
oldWedged := wedgedAfter
|
||||
wedgedAfter = 50 * time.Millisecond // 测试里把判定阈值压到极短
|
||||
wedgedAfter = 50 * time.Millisecond
|
||||
t.Cleanup(func() { wedgedAfter = oldWedged })
|
||||
|
||||
_, ps, _ := runtimePlugins(t)
|
||||
code := `
|
||||
local plugin = { name = "wedger", version = "1.0" }
|
||||
plugin.schedule = { { fn = "spin", every = 3600 } }
|
||||
function plugin.spin()
|
||||
local i = 0
|
||||
while true do i = i + 1 end
|
||||
end
|
||||
function plugin.spin() local i=0 while true do i=i+1 end end
|
||||
return plugin
|
||||
`
|
||||
if err := ps.LoadSource("wedger", code); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ps.mu.RLock()
|
||||
var rt *PluginRuntime
|
||||
for _, p := range ps.plugins {
|
||||
if p.Info.Name == "wedger" {
|
||||
rt = p.Runtime
|
||||
}
|
||||
}
|
||||
ps.mu.RUnlock()
|
||||
if rt == nil {
|
||||
t.Fatal("无运行时")
|
||||
}
|
||||
rt := runtimeOf(t, ps, "wedger", 3*time.Second)
|
||||
// 手动跑一次会卡死的任务(它会拿到 p.mu 并永不返回)
|
||||
go rt.runOne(ScheduledTask{Fn: "spin", Every: 3600})
|
||||
|
||||
// 手动触发一次任务(它会卡住),然后等 checkWedged 判定并抛弃该代。
|
||||
rt.mu.Lock()
|
||||
gen1 := rt.gen
|
||||
rt.mu.Unlock()
|
||||
if gen1 == nil {
|
||||
t.Fatal("初始无 generation")
|
||||
}
|
||||
go rt.runOne(gen1, ScheduledTask{Fn: "spin", Every: 3600})
|
||||
|
||||
// 等 wedgedAfter 过去并让 checkWedged 跑一轮
|
||||
deadline := time.Now().Add(2 * time.Second)
|
||||
// 等 checkWedged 判定并停用
|
||||
deadline := time.Now().Add(3 * time.Second)
|
||||
for time.Now().Before(deadline) {
|
||||
rt.mu.Lock()
|
||||
abandoned := rt.gen != gen1
|
||||
rt.mu.Unlock()
|
||||
if abandoned {
|
||||
break
|
||||
if rt.Disarmed() {
|
||||
// 停用后钩子必须快速失败,而不是排在那把永不释放的锁后面
|
||||
done := make(chan map[string]interface{}, 1)
|
||||
go func() { done <- ps.Fire(StageRouted, map[string]interface{}{}) }()
|
||||
select {
|
||||
case <-done:
|
||||
return // 通过
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("插件停用后钩子仍然挂起 —— 应快速失败")
|
||||
}
|
||||
}
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
}
|
||||
rt.mu.Lock()
|
||||
gen2 := rt.gen
|
||||
abandons := rt.abandons["spin"]
|
||||
rt.mu.Unlock()
|
||||
if gen2 == gen1 {
|
||||
t.Fatal("卡死的任务应导致 generation 被抛弃(换新 VM),实得同一代")
|
||||
}
|
||||
if abandons == 0 {
|
||||
t.Error("抛弃次数应被记账(否则熔断无法生效)")
|
||||
}
|
||||
t.Logf("已抛弃并换新 generation(abandons=%d)", abandons)
|
||||
t.Fatal("卡死的任务未导致插件停用")
|
||||
}
|
||||
|
||||
// TestWedgedPluginIsDisarmedAfterMaxAbandons:反复卡死必须熔断,而不是无限
|
||||
// 重生 VM(每次重生都留下一个永不返回的 goroutine + Lua 状态 = 无界泄漏)。
|
||||
func TestWedgedPluginIsDisarmedAfterMaxAbandons(t *testing.T) {
|
||||
shortTick(t)
|
||||
oldWedged := wedgedAfter
|
||||
wedgedAfter = 20 * time.Millisecond
|
||||
t.Cleanup(func() { wedgedAfter = oldWedged })
|
||||
|
||||
_, ps, _ := runtimePlugins(t)
|
||||
code := `
|
||||
local plugin = { name = "alwaysbad", version = "1.0" }
|
||||
plugin.schedule = { { fn = "spin", every = 3600 } }
|
||||
function plugin.spin() local i=0 while true do i=i+1 end end
|
||||
return plugin
|
||||
`
|
||||
if err := ps.LoadSource("alwaysbad", code); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ps.mu.RLock()
|
||||
var rt *PluginRuntime
|
||||
for _, p := range ps.plugins {
|
||||
if p.Info.Name == "alwaysbad" {
|
||||
rt = p.Runtime
|
||||
}
|
||||
}
|
||||
ps.mu.RUnlock()
|
||||
if rt == nil {
|
||||
t.Fatal("无运行时")
|
||||
}
|
||||
|
||||
// 反复:拿当前代 → 跑卡死任务 → 等被抛弃。直到熔断。
|
||||
deadline := time.Now().Add(5 * time.Second)
|
||||
for time.Now().Before(deadline) {
|
||||
rt.mu.Lock()
|
||||
if rt.disarmed {
|
||||
rt.mu.Unlock()
|
||||
break
|
||||
}
|
||||
gen := rt.gen
|
||||
rt.mu.Unlock()
|
||||
if gen != nil {
|
||||
go rt.runOne(gen, ScheduledTask{Fn: "spin", Every: 3600})
|
||||
}
|
||||
time.Sleep(30 * time.Millisecond)
|
||||
}
|
||||
rt.mu.Lock()
|
||||
disarmed := rt.disarmed
|
||||
n := rt.abandons["spin"]
|
||||
rt.mu.Unlock()
|
||||
if !disarmed {
|
||||
t.Fatalf("反复卡死应最终熔断(disarmed),实得 abandons=%d", n)
|
||||
}
|
||||
if n < maxAbandons {
|
||||
t.Errorf("熔断前 abandon 次数 = %d, 应 >= %d", n, maxAbandons)
|
||||
}
|
||||
// 熔断后不应再有 generation 被启动
|
||||
rt.mu.Lock()
|
||||
genAfter := rt.gen
|
||||
rt.mu.Unlock()
|
||||
if genAfter != nil {
|
||||
t.Error("熔断后不应仍有活动 generation")
|
||||
}
|
||||
}
|
||||
|
||||
// TestNormalTaskUnaffectedByWedgeWatchdog:正常(快速)任务不触发抛弃。
|
||||
func TestNormalTaskUnaffectedByWedgeWatchdog(t *testing.T) {
|
||||
// TestNormalTaskNotDisarmed:正常任务不会被误判为卡死。
|
||||
func TestNormalTaskNotDisarmed(t *testing.T) {
|
||||
shortTick(t)
|
||||
_, ps, _ := runtimePlugins(t)
|
||||
code := `
|
||||
local plugin = { name = "healthy", version = "1.0" }
|
||||
plugin.state = { n = 0 }
|
||||
plugin.schedule = { { fn = "q", every = 1 } }
|
||||
function plugin.q() return nil end
|
||||
function plugin.q() plugin.state.n = plugin.state.n + 1 end
|
||||
return plugin
|
||||
`
|
||||
if err := ps.LoadSource("healthy", code); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ps.mu.RLock()
|
||||
var rt *PluginRuntime
|
||||
for _, p := range ps.plugins {
|
||||
if p.Info.Name == "healthy" {
|
||||
rt = p.Runtime
|
||||
}
|
||||
}
|
||||
ps.mu.RUnlock()
|
||||
if rt == nil {
|
||||
t.Fatal("无运行时")
|
||||
rt := runtimeOf(t, ps, "healthy", 3*time.Second)
|
||||
time.Sleep(600 * time.Millisecond)
|
||||
if rt.Disarmed() {
|
||||
t.Error("健康任务不应导致插件停用")
|
||||
}
|
||||
rt.mu.Lock()
|
||||
gen0 := rt.gen
|
||||
rt.mu.Unlock()
|
||||
time.Sleep(600 * time.Millisecond) // 让任务跑几轮
|
||||
rt.mu.Lock()
|
||||
same := rt.gen == gen0
|
||||
fails := rt.failures["q"]
|
||||
abandons := rt.abandons["q"]
|
||||
rt.mu.Unlock()
|
||||
if !same {
|
||||
t.Error("健康任务不该导致 generation 被抛弃")
|
||||
}
|
||||
if fails != 0 {
|
||||
t.Errorf("健康任务不应失败: %d", fails)
|
||||
}
|
||||
if abandons != 0 {
|
||||
t.Errorf("健康任务不应被抛弃: %d", abandons)
|
||||
}
|
||||
}
|
||||
|
||||
// TestConcurrentScheduledTasksStress:多个插件各自调度,互不干扰。
|
||||
func TestConcurrentScheduledTasksStress(t *testing.T) {
|
||||
shortTick(t)
|
||||
_, ps, _ := runtimePlugins(t)
|
||||
for i := 0; i < 8; i++ {
|
||||
name := fmt.Sprintf("stress%d", i)
|
||||
code := `
|
||||
local plugin = { name = "` + name + `", version = "1.0" }
|
||||
plugin.state = { n = 0 }
|
||||
plugin.schedule = { { fn = "tick", every = 1 } }
|
||||
function plugin.tick() plugin.state.n = plugin.state.n + 1 end
|
||||
return plugin
|
||||
`
|
||||
if err := ps.LoadSource(name, code); err != nil {
|
||||
t.Fatalf("%s: %v", name, err)
|
||||
}
|
||||
}
|
||||
// every=1 means the FIRST fire is one second out (firstDue = now+1s), so the
|
||||
// wait must exceed that; 700ms observed nothing.
|
||||
time.Sleep(1600 * time.Millisecond)
|
||||
// 每个插件都自己跑起来了(跨插件并行、各自一个 VM)
|
||||
advanced := 0
|
||||
for i := 0; i < 8; i++ {
|
||||
name := fmt.Sprintf("stress%d", i)
|
||||
if st, ok := ps.State(name).(map[string]interface{}); ok {
|
||||
if n, ok := st["n"].(float64); ok && n >= 1 {
|
||||
advanced++
|
||||
}
|
||||
}
|
||||
}
|
||||
if advanced == 0 {
|
||||
t.Fatal("没有任何插件调度任务执行")
|
||||
}
|
||||
t.Logf("%d/8 个插件的任务已执行", advanced)
|
||||
}
|
||||
|
||||
@ -245,12 +245,14 @@ type Plugin struct {
|
||||
persisted bool
|
||||
|
||||
// Schedule is the periodic work this plugin declared, and Runtime is the
|
||||
// long-lived environment that runs it (private VM + kernel ticker).
|
||||
// scheduler that runs it on THIS plugin's own VM (p.state) under p.mu — the
|
||||
// same state and the same lock a request hook uses. That is what makes a
|
||||
// token rotated by a task immediately visible to inference traffic.
|
||||
//
|
||||
// Runtime is nil when the plugin declared nothing — the common case stays
|
||||
// exactly as cheap as before: one shared state, no extra VM, no goroutine.
|
||||
// The point of it is auto check-in / token rotation: work that must happen
|
||||
// with no browser open and nobody clicking anything.
|
||||
// exactly as cheap as before: one VM, no extra goroutine. The point of the
|
||||
// schedule is auto check-in / token rotation: work that must happen with no
|
||||
// browser open and nobody clicking anything.
|
||||
Schedule []ScheduledTask
|
||||
Runtime *PluginRuntime
|
||||
}
|
||||
@ -708,8 +710,8 @@ func (ps *Plugins) LoadSource(name, code string) error {
|
||||
|
||||
// ---- schedule ----
|
||||
//
|
||||
// A plugin that declares `schedule` gets a LONG-RUNNING environment: a
|
||||
// private VM plus a kernel-driven ticker, so its periodic work (auto
|
||||
// A plugin that declares `schedule` gets a kernel-driven scheduler running
|
||||
// ON ITS OWN VM (the same p.state a hook uses), so its periodic work (auto
|
||||
// check-in, token rotation, quota polling) happens with no browser open and
|
||||
// no operator clicking anything. Without this the only periodic work a
|
||||
// plugin could do was a front-end poll, which is exactly the "nobody is
|
||||
@ -720,9 +722,9 @@ func (ps *Plugins) LoadSource(name, code string) error {
|
||||
p.Schedule = parseSchedule(w.L).tasks
|
||||
w.L.SetTop(0)
|
||||
if len(p.Schedule) > 0 {
|
||||
p.Runtime = ps.startRuntime(p.Info.Name, code, scheduleSpec{tasks: p.Schedule})
|
||||
p.Runtime = ps.startRuntime(p, scheduleSpec{tasks: p.Schedule})
|
||||
if p.Runtime == nil {
|
||||
// The private VM failed to boot. That must NOT silently disable the
|
||||
// The scheduler could not start. That must NOT silently disable the
|
||||
// plugin — hooks still work — but the operator needs to know their
|
||||
// scheduled work will never run.
|
||||
log.Printf("[scheduler] plugin %s: declared %d task(s) but no runtime could be started",
|
||||
@ -1792,13 +1794,29 @@ func (ps *Plugins) Unload(name string) error {
|
||||
p := ps.plugins[idx]
|
||||
ps.plugins = append(ps.plugins[:idx], ps.plugins[idx+1:]...)
|
||||
ps.mu.Unlock()
|
||||
// Stop the scheduler BEFORE closing the state: a task in flight holds that
|
||||
// state, and closing it underneath would leave the task writing into a
|
||||
// closed Lua state — a crash rather than a clean unload.
|
||||
// Stop the scheduler BEFORE touching the state. Stop() closes the loop and
|
||||
// waits for it, so after it returns no NEW task can be dispatched — the set
|
||||
// of in-flight tasks is now frozen, which is what makes the check below
|
||||
// meaningful rather than racy.
|
||||
inFlight := false
|
||||
if p.Runtime != nil {
|
||||
p.Runtime.Stop()
|
||||
inFlight = p.Runtime.InFlight()
|
||||
p.Runtime = nil
|
||||
}
|
||||
// A task holds p.mu for the whole duration of its Lua call. If one is still
|
||||
// in flight we must NOT take p.mu (a wedged task never releases it, so Lock
|
||||
// would hang the unload forever) and must NOT close the state underneath it
|
||||
// (that is a use-after-free in the Lua C layer, which aborts the process
|
||||
// instead of returning an error). Leaking this one state is the bounded,
|
||||
// non-crashing outcome; the operator's action is to remove the file and
|
||||
// restart, which is what a plugin wedged by its own task requires anyway.
|
||||
if inFlight {
|
||||
log.Printf("[scheduler] plugin %s: unloaded while a task was still running; "+
|
||||
"its Lua state is left open to avoid a use-after-free", name)
|
||||
ps.rebuild()
|
||||
return nil
|
||||
}
|
||||
p.mu.Lock()
|
||||
if p.state != nil && p.state.L != nil {
|
||||
p.state.L.Close()
|
||||
@ -1863,6 +1881,15 @@ func (ps *Plugins) Fire(stage Stage, payload map[string]interface{}) map[string]
|
||||
if p == nil || p.LoadError != "" {
|
||||
continue
|
||||
}
|
||||
// A plugin disarmed by an overrunning task still holds its own lock (the
|
||||
// task never returned), so dispatching to it would block THIS request
|
||||
// behind a lock that is never released. Skip it and record why: the
|
||||
// operator needs to see that inference traffic is being dropped and which
|
||||
// plugin is responsible, not watch requests hang.
|
||||
if p.Runtime != nil && p.Runtime.Disarmed() {
|
||||
ps.hookErr.note(stage, p.Info.Name+": "+errPluginWedged.Error())
|
||||
continue
|
||||
}
|
||||
targets = append(targets, target{p: p, hc: hc})
|
||||
}
|
||||
if len(targets) == 0 {
|
||||
|
||||
30
internal/lua/zz_repro_test.go
Normal file
30
internal/lua/zz_repro_test.go
Normal file
@ -0,0 +1,30 @@
|
||||
package lua
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// 临时复现:内核发出的 assistant(content=null)+tool_calls 形状,
|
||||
// 经 anthropic.lua 转换后 content 是否变成对象 {}(上游期望数组)。
|
||||
func TestZZReproAnthropicEmptyBlocks(t *testing.T) {
|
||||
vm := NewVM(freshAdapterDir(t))
|
||||
if err := vm.Start(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer vm.Stop()
|
||||
|
||||
cases := map[string]string{
|
||||
"assistant content=null + tool_calls": `{"model":"x","messages":[{"role":"user","content":"hi"},{"role":"assistant","content":null,"tool_calls":[{"id":"c1","type":"function","function":{"name":"f","arguments":"{}"}}]},{"role":"tool","tool_call_id":"c1","content":"ok"}]}`,
|
||||
"assistant content=空串 + tool_calls": `{"model":"x","messages":[{"role":"user","content":"hi"},{"role":"assistant","content":"","tool_calls":[{"id":"c1","type":"function","function":{"name":"f","arguments":"{}"}}]},{"role":"tool","tool_call_id":"c1","content":"ok"}]}`,
|
||||
"仅 system": `{"model":"x","messages":[{"role":"system","content":"s"}]}`,
|
||||
}
|
||||
for name, body := range cases {
|
||||
out, err := vm.Transform("anthropic", "transform_request", body)
|
||||
if err != nil {
|
||||
t.Fatalf("%s: %v", name, err)
|
||||
}
|
||||
bad := strings.Contains(out, `"content":{}`)
|
||||
t.Logf("%-38s content为对象{}: %v\n out=%s", name, bad, out)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user