ModelRouter/internal/lua/plugin_runtime.go

// Package lua — 插件长期运行环境（R4-b / Phase 4）
//
// 用户需求：插件需要**长期运行环境（独立 LuaVM 线程化）**，以支持自动签到、
// token 轮换等需要"没有浏览器、没有人操作时也会发生"的能力。
//
// 为什么需要独立 VM：现有插件模型是「一个插件 = 一个 Lua 状态 + 一把锁」，
// 因为钩子会往 plugin.state 累加，多个状态会让总额被劈成几份（见 docs 的
// 运行时约束）。这把锁的代价是**一个卡住的钩子会卡住所有插件**——而签到、
// token 轮换这类任务天然是长耗时的网络等待。
//
// 本文件给出调度层：内核按插件声明的周期，在**独立 VM + 独立 goroutine** 上
// 执行它，从而：
//
//   - 一个插件的长任务不阻塞其它插件的钩子（独立 VM，无共享锁）
//   - 任务在没有浏览器、没有人操作时也会执行（内核 ticker 驱动，不是前端轮询）
//   - 状态与主 VM 分离，轮换中崩溃不会带走已累计的钩子统计
//
// 声明方式（插件侧）：
//
//	PLUGIN.schedule = {
//	    { 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，不影响钩子与请求转发。
package lua

import (
	"fmt"
	"sort"
	"sync"
	"sync/atomic"
	"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"`
	// Timeout, in seconds, bounds one execution. 0 uses defaultTaskTimeout.
	// A task that exceeds it is interrupted at the Lua instruction level — the
	// only way to stop a runaway script on the C stack.
	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")
			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 long-lived execution environment.
//
// 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.
type PluginRuntime struct {
	plugin string
	source string // kept so a discarded generation can be re-booted from scratch

	mu   sync.Mutex
	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
	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.
	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.
	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.
//
// 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

// startRuntime boots the first generation and starts the scheduler.
//
// 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 {
		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{}),
	}
	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)
	}
	go rt.loop()
	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 {
	now := time.Now()
	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:
			// 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.
//
// 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.
func (rt *PluginRuntime) checkWedged() {
	rt.mu.Lock()
	gen := rt.gen
	if gen == nil {
		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
		}
		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
	for _, fn := range wedged {
		rt.abandons[fn]++
		rt.failures[fn]++
		n := rt.abandons[fn]
		if n >= maxAbandons {
			rt.disarmed = true
			disarmedNow = true
		}
		rt.logf("[scheduler] plugin %s task %s exceeded %s; abandoning its VM (abandon #%d/%d)",
			rt.plugin, fn, wedgedAfter, n, maxAbandons)
	}
	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.
func (rt *PluginRuntime) runDue() {
	rt.mu.Lock()
	if rt.disarmed || rt.gen == nil {
		rt.mu.Unlock()
		return
	}
	gen := rt.gen
	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
		if rt.startedAt == nil {
			rt.startedAt = map[string]time.Time{}
		}
		rt.startedAt[t.Fn] = now
		rt.nextDue[t.Fn] = rt.nextAfter(t, now)
		due = append(due, t)
	}
	rt.mu.Unlock()
	for _, t := range due {
		go rt.runOne(gen, t)
	}
}

// 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 given generation.
//
// 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()
	defer func() {
		rt.mu.Lock()
		delete(gen.busy, t.Fn)
		delete(rt.running, t.Fn)
		delete(rt.startedAt, t.Fn)
		rt.mu.Unlock()
	}()

	p := gen.vm.pool(gen.poolName)
	if p == nil {
		rt.recordFailure(t.Fn, "no pool")
		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 {
		rt.recordFailure(t.Fn, err.Error())
		return
	}
	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.
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...)
	}
}

// Stop halts the scheduler and abandons the current generation.
func (rt *PluginRuntime) Stop() {
	if rt == nil {
		return
	}
	close(rt.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.
//
// 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.
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
}

// 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 {
	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).
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))
}

// sortedTaskNames returns the task function names of a spec (for diagnostics).
func (s scheduleSpec) sortedTaskNames() []string {
	out := make([]string, 0, len(s.tasks))
	for _, t := range s.tasks {
		out = append(out, t.Fn)
	}
	sort.Strings(out)
	return out
}

// PluginTasks lists the scheduled task names a plugin declared (for diagnostics
// and for the WebUI to show what periodic work exists).
func (p *Plugin) PluginTasks() []string {
	out := make([]string, 0, len(p.Schedule))
	for _, t := range p.Schedule {
		out = append(out, t.Fn)
	}
	return out
}