ModelRouter/internal/lua/plugin_runtime.go

// Package lua — 插件长期运行环境（R4-b / Phase 4）
//
// 用户需求：插件需要**长期运行环境**，以支持自动签到、token 轮换等需要
// "没有浏览器、没有人操作时也会发生"的能力。
//
// 模型（重要）：**一个插件 = 一个 Lua VM = 一个调度器**。
//
// 钩子（请求路径）和定时任务（内核 ticker 驱动）跑在**同一个** Lua 状态上，
// 由同一把锁串行化。这不是实现细节，而是正确性要求：
//
//	定时任务 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 },    -- 带抖动
//	}
//
// 崩溃隔离：任务函数出错只记 hookErr，不影响钩子与请求转发。但一个**永不返回**
// 的任务会永久持有该插件的锁（Lua 调用无法中断，见 checkWedged 的能力说明），
// 因此超时后该插件被**停用**而不是"换一个 VM 重来"—— 换 VM 会把累计状态
// （计费总额、已签到的天数）一起丢掉，那比停用更糟。
package lua

import (
	"errors"
	"fmt"
	"log"
	"sync"
	"time"

	golua "github.com/aarzilli/golua/lua"
)

// ScheduledTask is one declared periodic job.
type ScheduledTask struct {
	// Fn is the plugin function name to invoke.
	Fn string
	// Every is the interval in seconds. 0 disables the task.
	Every int `json:"every"`
	// At is an optional daily local time "HH:MM"; when set, Every is ignored
	// and the task fires once per day at that wall-clock time.
	At string `json:"at,omitempty"`
	// 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"`
	// FirstDueNow makes the FIRST run happen shortly after boot instead of after
	// a full period.
	//
	// Why it exists (measured defect, borrowed from the reference implementation's
	// post-mortem): with interval-only scheduling a restart leaves short-lived
	// credentials expired for up to a whole period. The reference's providers
	// have cline at 1h, codearts at ~2h and raccoon at 3h, so a 30-minute
	// refresh interval left them showing "已过期" and their quota rows returning
	// 401 after every host restart. A startup catch-up round fixes that — but it
	// must be opt-in, because a task that is genuinely periodic (a daily
	// check-in, say) should NOT fire again just because the host restarted.
	FirstDueNow bool `json:"first_due_now,omitempty"`
	// 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"`
}

// scheduleSpec is the parsed `PLUGIN.schedule` table.
type scheduleSpec struct {
	tasks []ScheduledTask
}

// parseSchedule extracts the schedule from a loaded plugin table. It is read
// from the already-compiled state's global, not by re-running the script, so a
// plugin's schedule is available the moment it loads.
func parseSchedule(L *golua.State) scheduleSpec {
	var spec scheduleSpec
	if L == nil {
		return spec
	}
	L.GetGlobal(pluginGlobal)
	if L.IsNil(-1) {
		L.Pop(1)
		return spec
	}
	if !getFieldBatch(L, "schedule") {
		L.Pop(1)
		return spec
	}
	// schedule is an array of tables
	if L.Type(-1) != golua.LUA_TTABLE {
		L.Pop(2)
		return spec
	}
	arr := L.GetTop()
	idx := 1
	for {
		// golua exposes indexed table access through RawGeti; GetField only
		// takes a string key.
		L.RawGeti(arr, idx)
		if L.IsNil(-1) {
			L.Pop(1)
			break
		}
		if L.Type(-1) == golua.LUA_TTABLE {
			var t ScheduledTask
			t.Fn = readString(L, -1, "fn")
			t.Every = readNumberInt(L, -1, "every")
			t.At = readString(L, -1, "at")
			t.Jitter = readNumberInt(L, -1, "jitter")
			t.Timeout = readNumberInt(L, -1, "timeout")
			t.FirstDueNow = readBoolField(L, -1, "first_due_now")
			if t.Fn != "" && (t.Every > 0 || t.At != "") {
				spec.tasks = append(spec.tasks, t)
			}
		}
		L.Pop(1)
		idx++
	}
	L.Pop(2) // schedule table + plugin global
	return spec
}

// PluginRuntime owns one plugin's periodic work.
//
// 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 *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

	// stopOnce makes Stop idempotent. Stop closes a channel and waits on done;
	// without this, a second Stop panics ("close of closed channel"). That is
	// reachable in practice now that BOTH Unload and Close stop schedulers:
	// unloading a plugin then closing the plugin set would otherwise abort the
	// process during shutdown — the worst possible moment (a panic here masks
	// whatever the operator was actually shutting down for).
	stopOnce sync.Once

	mu sync.Mutex
	// nextDue is when each task should next fire.
	nextDue map[string]time.Time
	// 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
	// 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

	stop chan struct{}
	done chan struct{}
}

// 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.
var wedgedAfter = 10 * time.Minute

// 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 attaches a scheduler to a plugin that declared one.
//
// 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:    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{}),
	}
	for _, t := range spec.tasks {
		rt.nextDue[t.Fn] = rt.firstDue(t)
	}
	go rt.loop()
	return rt
}

// 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 {
	now := time.Now()
	// A catch-up run: fire shortly after boot so a restart does not leave
	// short-lived credentials expired for a whole period (see FirstDueNow).
	// The short delay lets the rest of the plugin finish booting first.
	if t.FirstDueNow {
		return now.Add(catchUpDelay)
	}
	if t.At != "" {
		return nextDailyOccurrence(t.At, now)
	}
	d := time.Duration(t.Every) * time.Second
	if d <= 0 {
		d = time.Hour
	}
	if j := time.Duration(t.Jitter) * time.Second; j > 0 {
		d += time.Duration(randInt63n(int64(j)))
	}
	return now.Add(d)
}

// nextDailyOccurrence returns the next time "HH:MM" passes after now (today if
// it has not yet passed, else tomorrow).
func nextDailyOccurrence(at string, now time.Time) time.Time {
	var hh, mm int
	if _, err := fmt.Sscanf(at, "%d:%d", &hh, &mm); err != nil || hh < 0 || hh > 23 || mm < 0 || mm > 59 {
		// A malformed time must not wedge the task forever; fall back to hourly.
		return now.Add(time.Hour)
	}
	t := time.Date(now.Year(), now.Month(), now.Day(), hh, mm, 0, 0, now.Location())
	if !t.After(now) {
		t = t.Add(24 * time.Hour)
	}
	return t
}

// runtimeTick is the scheduler's resolution. A var so tests can shorten it
// without waiting whole seconds; production always uses 1s.
var runtimeTick = time.Second

// loop is the scheduler goroutine: one ticker drives every task of the plugin.
func (rt *PluginRuntime) loop() {
	defer close(rt.done)
	tick := time.NewTicker(runtimeTick)
	defer tick.Stop()
	for {
		select {
		case <-rt.stop:
			return
		case <-tick.C:
			rt.checkWedged()
			rt.runDue()
		}
	}
}

// checkWedged disarms a plugin whose task has overrun wedgedAfter.
//
// 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()
	if rt.disarmed {
		rt.mu.Unlock()
		return
	}
	var wedged []string
	for fn, at := range rt.startedAt {
		if time.Since(at) >= wedgedAfter {
			wedged = append(wedged, fn)
		}
	}
	if len(wedged) == 0 {
		rt.mu.Unlock()
		return
	}
	rt.disarmed = true
	rt.mu.Unlock()

	for _, fn := range wedged {
		ps := rt.ps
		if ps != nil {
			ps.hookErr.note(StageRequestEnd, fmt.Sprintf("%s: task %s overran %s; plugin disabled",
				rt.pluginInfoName(), fn, wedgedAfter))
		}
		log.Printf("[scheduler] plugin %s task %s exceeded %s; disabling the plugin "+
			"(reload it to retry)", rt.pluginInfoName(), fn, wedgedAfter)
	}
}

// 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.mu.Unlock()
		return
	}
	now := time.Now()
	var due []ScheduledTask
	for _, t := range rt.spec.tasks {
		if rt.running[t.Fn] {
			continue // still executing; skip this cycle
		}
		if now.Before(rt.nextDue[t.Fn]) {
			continue
		}
		rt.running[t.Fn] = true
		rt.active++
		rt.nextDue[t.Fn] = rt.nextAfter(t, now)
		due = append(due, t)
	}
	rt.mu.Unlock()
	for _, t := range due {
		go rt.runOne(t)
	}
}

// catchUpDelay is how long a FirstDueNow task waits after boot. Long enough for
// the plugin's own boot work to finish, short enough that a restart heals
// expired credentials promptly rather than after a full period.
const catchUpDelay = 5 * time.Second

// nextAfter computes the following due time for a task that just fired.
func (rt *PluginRuntime) nextAfter(t ScheduledTask, now time.Time) time.Time {
	if t.At != "" {
		return nextDailyOccurrence(t.At, now)
	}
	d := time.Duration(t.Every) * time.Second
	if d <= 0 {
		d = time.Hour
	}
	return now.Add(d)
}

// runOne executes one task on the plugin's own VM, under the plugin's own lock.
//
// 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(rt.running, t.Fn)
		delete(rt.startedAt, t.Fn)
		rt.active--
		rt.mu.Unlock()
	}()

	if p == nil {
		rt.recordFailure(t.Fn, "plugin is gone")
		return
	}

	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.mu.Unlock()
}

// 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()
	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)
}

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
	}
	rt.stopOnce.Do(func() {
		close(rt.stop)
		<-rt.done
	})
}

// callFnOn invokes a plugin function on an already-locked state.
//
// 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)
	L.GetGlobal(pluginGlobal)
	if L.IsNil(-1) {
		return fmt.Errorf("plugin table missing")
	}
	plug := L.GetTop()
	L.GetGlobal(hookGuardName)
	if L.IsNil(-1) {
		L.Pop(1)
	}
	L.GetField(plug, fn)
	if !L.IsFunction(-1) {
		return fmt.Errorf("task %s missing", fn)
	}
	L.PushString("{}")
	if err := L.Call(1, 2); err != nil {
		return err
	}
	if L.GetTop() >= 2 && !L.IsNil(-1) {
		msg := L.ToString(-1)
		L.Pop(1)
		if msg != "" {
			return fmt.Errorf("%s", msg)
		}
	}
	return nil
}

// 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 {
	if n <= 0 {
		return 0
	}
	seed := time.Now().UnixNano() ^ int64(n*2654435761)
	x := uint64(seed)
	x ^= x << 13
	x ^= x >> 7
	x ^= x << 17
	return int64(x % uint64(n))
}

// readNumberInt reads a numeric field as an int. Missing/zero yields 0, which
// every caller treats as "not set" (a task with every<=0 and no at is disabled).
// readBoolField reads a boolean field, treating only a real Lua boolean as set.
// A missing field (nil) means false — the caller's declared default.
func readBoolField(L *golua.State, idx int, field string) bool {
	L.GetField(idx, field)
	defer L.Pop(1)
	if L.Type(-1) != golua.LUA_TBOOLEAN {
		return false
	}
	return L.ToBoolean(-1)
}

func readNumberInt(L *golua.State, idx int, field string) int {
	L.GetField(idx, field)
	defer L.Pop(1)
	if L.Type(-1) != golua.LUA_TNUMBER {
		return 0
	}
	return int(L.ToNumber(-1))
}