ModelRouter/internal/lua/persist_test.go

package lua

import (
	"encoding/json"
	"os"
	"path/filepath"
	"testing"
	"time"
)

// These cover plugin state persistence: a plugin's totals live in the Lua VM,
// which dies with the process. Measured on the real gateway before this change:
// 218,241 requests and 26.2M prompt tokens gone after one systemctl restart.

func persistTestPlugin() string {
	return `
local plugin = { name = "counter", version = "1.0" }
plugin.state = { n = 0, prices_seen = 0 }
plugin.prices = { rate = 1 }
plugin.hooks = { request_end = "bump" }
function plugin.bump(payload)
  plugin.state.n = plugin.state.n + 1
  return nil
end
return plugin
`
}

// newPersistVM wires a plugin registry on a fresh dir, mirroring newPluginVM.
func newPersistVM(t *testing.T) (*Plugins, string) {
	t.Helper()
	dir := t.TempDir()
	vm := NewVM(filepath.Join(dir, "adapters"))
	if err := vm.Start(); err != nil {
		t.Fatalf("vm start: %v", err)
	}
	t.Cleanup(vm.Stop)
	pdir := filepath.Join(dir, "plugins")
	ps := NewPlugins(vm, pdir)
	// Shorten the saver debounce: Close() waits for the saver goroutine, so
	// with the production 2s interval every test would pay 2s on teardown.
	ps.save.setWakeDelayForTest(2 * time.Millisecond)
	t.Cleanup(ps.Close)
	return ps, pdir
}

func loadCounter(t *testing.T, ps *Plugins) {
	t.Helper()
	if err := ps.LoadSource("counter", persistTestPlugin()); err != nil {
		t.Fatalf("load: %v", err)
	}
}

func stateN(t *testing.T, ps *Plugins) float64 {
	t.Helper()
	st := ps.State("counter")
	m, ok := st.(map[string]interface{})
	if !ok {
		t.Fatalf("state is %T, want map", st)
	}
	n, ok := m["n"].(float64)
	if !ok {
		t.Fatalf("state.n is %T, want float64", m["n"])
	}
	return n
}

// TestStateSurvivesRestart is the defect itself: a rebuilt registry over the
// same plugin dir must come up with the previous totals, not at zero.
func TestStateSurvivesRestart(t *testing.T) {
	ps, pdir := newPersistVM(t)
	loadCounter(t, ps)
	for i := 0; i < 25; i++ {
		ps.Fire(StageRequestEnd, map[string]interface{}{"model": "m", "source": "s"})
	}
	if got := stateN(t, ps); got != 25 {
		t.Fatalf("in-process total = %v, want 25", got)
	}
	// Force the write the saver would do, so the test does not depend on timing.
	ps.save.flush()
	ps.Close()

	// A brand-new registry over the same dir: this is the restart.
	vm := NewVM(filepath.Join(t.TempDir(), "adapters"))
	if err := vm.Start(); err != nil {
		t.Fatal(err)
	}
	defer vm.Stop()
	ps2 := NewPlugins(vm, pdir)
	defer ps2.Close()
	loadCounter(t, ps2)

	if got := stateN(t, ps2); got != 25 {
		t.Fatalf("★ total after restart = %v, want 25 — this is the 'restart loses the books' bug", got)
	}
}

// TestPricesAndStateAreSeparate: restoring configuration over history (or the
// reverse) would either erase the totals or resurrect stale prices.
func TestPricesAndStateAreSeparate(t *testing.T) {
	ps, pdir := newPersistVM(t)
	loadCounter(t, ps)
	for i := 0; i < 7; i++ {
		ps.Fire(StageRequestEnd, map[string]interface{}{"model": "m", "source": "s"})
	}
	ps.save.flush()
	ps.Close()

	b, err := os.ReadFile(ps.stateFile("counter"))
	if err != nil {
		t.Fatalf("no state file: %v", err)
	}
	var saved struct {
		State  map[string]interface{} `json:"state"`
		Prices map[string]interface{} `json:"prices"`
	}
	if err := json.Unmarshal(b, &saved); err != nil {
		t.Fatalf("state file is not valid JSON: %v", err)
	}
	if saved.Prices == nil || saved.Prices["rate"] != float64(1) {
		t.Errorf("prices were not persisted separately: %v", saved.Prices)
	}
	if saved.State["n"] != float64(7) {
		t.Errorf("state.n = %v, want 7", saved.State["n"])
	}
	_ = pdir
}

// TestPricesOnlyUpdatePersists guards a real hole: SetState returns early for a
// prices-only payload (correctly leaving state alone), and the first version
// returned before the persistence write — so a reprice was durable in memory
// only and a restart silently reverted to the old prices.
func TestPricesOnlyUpdatePersists(t *testing.T) {
	ps, _ := newPersistVM(t)
	loadCounter(t, ps)
	for i := 0; i < 5; i++ {
		ps.Fire(StageRequestEnd, map[string]interface{}{"model": "m", "source": "s"})
	}
	ps.save.flush()

	err := ps.SetState("counter", map[string]interface{}{
		"prices": map[string]interface{}{"rate": 42},
	})
	if err != nil {
		t.Fatalf("SetState: %v", err)
	}
	if got := stateN(t, ps); got != 5 {
		t.Errorf("a prices-only payload must not touch state: n = %v, want 5", got)
	}

	b, err := os.ReadFile(ps.stateFile("counter"))
	if err != nil {
		t.Fatalf("no state file after a prices-only update: %v", err)
	}
	var saved struct {
		State  map[string]interface{} `json:"state"`
		Prices map[string]interface{} `json:"prices"`
	}
	if err := json.Unmarshal(b, &saved); err != nil {
		t.Fatal(err)
	}
	if saved.Prices["rate"] != float64(42) {
		t.Errorf("★ new price not persisted: %v — a restart would revert to the old price", saved.Prices)
	}
	if saved.State["n"] != float64(5) {
		t.Errorf("state was clobbered by a price change: %v", saved.State["n"])
	}
}

// TestHookReturnValueSurvivesSnapshot guards the ordering bug: snapshotting
// plugin.state resets the Lua stack, so doing it BEFORE reading the hook's
// return value silently turned every opinionated plugin into a silent one —
// breaking the documented "return a table to merge into payload" contract with
// no error anywhere.
func TestHookReturnValueSurvivesSnapshot(t *testing.T) {
	ps, _ := newPersistVM(t)
	code := `
local plugin = { name = "opinionated" }
plugin.state = { n = 0 }
plugin.hooks = { request_end = "tag" }
function plugin.tag(payload)
  plugin.state.n = plugin.state.n + 1
  return { cost_usd = 1.25, verdict = "billed" }
end
return plugin
`
	if err := ps.LoadSource("opinionated", code); err != nil {
		t.Fatal(err)
	}
	out := ps.Fire(StageRequestEnd, map[string]interface{}{"model": "m", "source": "s"})
	if out["cost_usd"] != 1.25 {
		t.Errorf("★ hook return value lost: payload = %v", out)
	}
	if out["verdict"] != "billed" {
		t.Errorf("merged field lost: %v", out)
	}
	ps.save.flush()
	st := ps.State("opinionated").(map[string]interface{})
	if st["n"] != float64(1) {
		t.Errorf("state.n = %v, want 1 (the hook still ran)", st["n"])
	}
}

// TestCorruptStateFileIsNotFatal: a truncated write must degrade to compiled-in
// defaults, never to a gateway that refuses to start.
func TestCorruptStateFileIsNotFatal(t *testing.T) {
	ps, _ := newPersistVM(t)
	loadCounter(t, ps)
	for i := 0; i < 3; i++ {
		ps.Fire(StageRequestEnd, map[string]interface{}{"model": "m", "source": "s"})
	}
	ps.save.flush()
	ps.Close()

	if err := os.WriteFile(ps.stateFile("counter"), []byte("{not json"), 0o644); err != nil {
		t.Fatal(err)
	}

	vm := NewVM(filepath.Join(t.TempDir(), "adapters"))
	if err := vm.Start(); err != nil {
		t.Fatal(err)
	}
	defer vm.Stop()
	ps2 := NewPlugins(vm, ps.dir)
	defer ps2.Close()
	var warned bool
	ps2.logf = func(string, ...interface{}) { warned = true }
	loadCounter(t, ps2)

	if got := stateN(t, ps2); got != 0 {
		t.Errorf("with a corrupt file the plugin must fall back to its defaults, got n = %v", got)
	}
	if !warned {
		t.Error("a corrupt state file must warn the operator, not fail silently")
	}
	// And it must still forward.
	ps2.Fire(StageRequestEnd, map[string]interface{}{"model": "m", "source": "s"})
}

// TestFlushIsCoalesced: N mutations must not become N writes. A synchronous
// per-request write would put a file write on the hot path.
func TestFlushIsCoalesced(t *testing.T) {
	ps, _ := newPersistVM(t)
	loadCounter(t, ps)
	for i := 0; i < 500; i++ {
		ps.Fire(StageRequestEnd, map[string]interface{}{"model": "m", "source": "s"})
	}
	ps.save.flush()
	// 500 mutations must collapse to a handful of writes, not 500. The bound is
	// loose on purpose: the saver also flushes on a timer, so how many writes
	// happen during the loop depends on how long 500 hook calls take. What must
	// never happen is one write per mutation.
	if w := ps.save.writes.Load(); w > 5 {
		t.Errorf("500 hook calls produced %d file writes; the saver must coalesce", w)
	}
	if got := stateN(t, ps); got != 500 {
		t.Fatalf("n = %v, want 500", got)
	}
}

// TestCloseFlushesTail: the shutdown path must not lose the last interval,
// which would reintroduce the same defect in a smaller window.
func TestCloseFlushesTail(t *testing.T) {
	ps, _ := newPersistVM(t)
	loadCounter(t, ps)
	ps.Fire(StageRequestEnd, map[string]interface{}{"model": "m", "source": "s"})
	ps.Close()
	// Close() already waits for the final flush; no sleep needed.

	b, err := os.ReadFile(ps.stateFile("counter"))
	if err != nil {
		t.Fatalf("★ Close() did not flush: %v — a shutdown loses the tail", err)
	}
	var saved struct {
		State map[string]interface{} `json:"state"`
	}
	if err := json.Unmarshal(b, &saved); err != nil {
		t.Fatal(err)
	}
	if saved.State["n"] != float64(1) {
		t.Errorf("flushed n = %v, want 1", saved.State["n"])
	}
}