onesvm-browser-server/server/internal/scheduler/scheduler_test.go

// scheduler_test.go：调度语义单测（无外网；真实 store.DB + httptest stub 适配器）。
//
// 覆盖（A3.5）：抢单并发不重不漏（调度层）、ADMIT_MAX 背压 429、reaper 收割、
// 渲染互斥 shell_active 拒新、降级链 trafilatura 空正文 → lightpanda、
// 模版层 blocked/redact/golden 信封。
package scheduler

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"net/http"
	"net/http/httptest"
	"path/filepath"
	"strings"
	"sync"
	"testing"
	"time"

	_ "modernc.org/sqlite" // driver 注册（同 store 包测试探针惯例）

	"onesvm.com/onesvm/browser-server/internal/contract"
	"onesvm.com/onesvm/browser-server/internal/dock"
	"onesvm.com/onesvm/browser-server/internal/store"
)

// newTestCore 构造真实 SQLite + 注册表 + Core（不依赖任何外部网络）。
func newTestCore(t *testing.T, admitMax int) (*Core, *store.DB) {
	t.Helper()
	db, err := store.Open(filepath.Join(t.TempDir(), "t.db"))
	if err != nil {
		t.Fatalf("Open: %v", err)
	}
	t.Cleanup(func() { db.Close() })
	if err := db.Migrate(); err != nil {
		t.Fatalf("Migrate: %v", err)
	}
	tmpl, err := NewTemplate()
	if err != nil {
		t.Fatalf("NewTemplate: %v", err)
	}
	core := NewCore(db, dock.NewRegistry(), tmpl, "http://127.0.0.1:1", // 不可达 proxy 端点
		4, admitMax, log.New(&discardLogger{}, "", 0))
	return core, db
}

// discardLogger 丢弃日志。
type discardLogger struct{}

func (d *discardLogger) Write(p []byte) (int, error) { return len(p), nil }

// stubSearchAdapter httptest searxng stub（样本形状）。
func stubSearchAdapter(t *testing.T) *httptest.Server {
	return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
		_, _ = w.Write([]byte(`{"query":"q","unresponsive_engines":[],"results":[
			{"title":"T1","url":"https://a.test/1","content":"内容一","engine":"baidu"},
			{"title":"T2","url":"https://a.test/2","content":"内容二","engine":"bing"}]}`))
	}))
}

// TestEnqueueAndResult 入队 → 执行 → /result 链路（search 快任务）。
func TestEnqueueAndResult(t *testing.T) {
	srv := stubSearchAdapter(t)
	defer srv.Close()
	core, _ := newTestCore(t, 60)
	core.reg.Register(dock.NewSearx("searxng-cn", srv.URL, "zh-CN"))
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	core.Start(ctx)
	defer core.Stop()

	env := contract.JobEnvelope{Intent: "search", RequestID: "req-1",
		Search: &contract.SearchInput{Query: "跨境电商"}}
	id, rej := core.Enqueue(env)
	if rej != nil {
		t.Fatalf("入队被拒: %+v", rej)
	}
	if id <= 0 {
		t.Fatalf("job_id 非法: %d", id)
	}
	// 轮询至 done（上限 5s）。
	deadline := time.Now().Add(5 * time.Second)
	for time.Now().Before(deadline) {
		job, err := core.db.JobByRequestID("req-1")
		if err == nil && job.Status == "done" {
			if got, ok := core.results.get("req-1"); ok {
				if !got.OK || got.Kind != "search" {
					t.Fatalf("信封不符: %+v", got)
				}
				sp := got.Data.(*contract.SearchPayload)
				if len(sp.Results) != 2 || sp.Results[0].Score != 1.0 {
					t.Fatalf("搜索结果不符: %+v", sp.Results)
				}
				return
			}
		}
		time.Sleep(20 * time.Millisecond)
	}
	t.Fatal("任务未在 5s 内完成")
}

// TestAdmitMaxBackpressure ADMIT_MAX 背压：无 worker 消费 → 排满 → 429 + Retry-After。
func TestAdmitMaxBackpressure(t *testing.T) {
	core, _ := newTestCore(t, 3) // 缩表：admit_max=3
	for i := 0; i < 3; i++ {
		_, rej := core.Enqueue(contract.JobEnvelope{Intent: "search", RequestID: fmt.Sprintf("r%d", i),
			Search: &contract.SearchInput{Query: "q"}})
		if rej != nil {
			t.Fatalf("前 3 个应入队成功: %+v", rej)
		}
	}
	id, rej := core.Enqueue(contract.JobEnvelope{Intent: "search", RequestID: "r-overflow",
		Search: &contract.SearchInput{Query: "q"}})
	if rej == nil {
		t.Fatalf("第 4 个应被拒: id=%d", id)
	}
	if rej.HTTPStatus != 429 || rej.RetryAfterS <= 0 {
		t.Errorf("拒绝应为 429 + Retry-After: %+v", rej)
	}
	if rej.Running+rej.Queued != 3 {
		t.Errorf("现状应 3，得 running=%d queued=%d", rej.Running, rej.Queued)
	}
}

// TestShellHoldMutualExclusive 渲染互斥：shell 活跃 → read 入队侧拒绝
// 503 + reason=shell_active_panda_hold；shell 回收后恢复。
func TestShellHoldMutualExclusive(t *testing.T) {
	core, _ := newTestCore(t, 60)
	core.hold.Set(true)
	_, rej := core.Enqueue(contract.JobEnvelope{Intent: "read", RequestID: "r-hold",
		Read: &contract.ReadInput{URL: "https://x.test/"}})
	if rej == nil {
		t.Fatal("shell 活跃期 read 新任务应被拒")
	}
	if rej.HTTPStatus != 503 || rej.Reason != "shell_active_panda_hold" {
		t.Errorf("拒绝形状不符: %+v", rej)
	}
	core.hold.Set(false)
	id, rej := core.Enqueue(contract.JobEnvelope{Intent: "read", RequestID: "r-hold2",
		Read: &contract.ReadInput{URL: "https://x.test/"}})
	if rej != nil {
		t.Fatalf("shell 回收后应放行: %+v", rej)
	}
	if id <= 0 {
		t.Errorf("job_id 非法: %d", id)
	}
}

// TestReaper 收割：租约过期任务回队（重试）→ 耗尽入死信（store 层语义经 Core 参数）。
func TestReaper(t *testing.T) {
	_, db := newTestCore(t, 60)
	payload, _ := json.Marshal(contract.JobEnvelope{Intent: "read", RequestID: "r-reap",
		Read: &contract.ReadInput{URL: "https://x.test/"}})
	id, err := db.EnqueueJob("r-reap", "read", payload, 100)
	if err != nil {
		t.Fatal(err)
	}
	// 手动抢单后不续租 → 租约过期。
	if _, err := db.ClaimNext("ghost", -time.Second); err != nil {
		t.Fatalf("抢单: %v", err)
	}
	n, err := db.ReapExpired(MaxAttempts)
	if err != nil || n != 1 {
		t.Fatalf("reaper 应收 1: n=%d err=%v", n, err)
	}
	// attempts=1 ≤ 2 → 回队（退避 available_at>now）。
	job, err := db.JobByID(id)
	if err != nil {
		t.Fatal(err)
	}
	if job.Status != "queued" || job.Attempts != 1 {
		t.Fatalf("应回队 attempts=1: %+v", job)
	}
	// 再耗尽：退避 available_at 在未来（base 30s）→ 期间 ClaimNext 恒 ErrNotFound。
	// reaper 只能收「running 且租约过期」——退避期任务卡在 queued 属 T6 设计口径
	//（available_at 兜底回抢），此处直改 available_at 模拟退避期满后再收割两次入死信。
	if _, err := db.Raw().Exec(`UPDATE jobs SET available_at = ? WHERE id = ?`,
		time.Now().Add(-time.Second).Format(time.RFC3339), id); err != nil {
		t.Fatal(err)
	}
	if _, err := db.ClaimNext("ghost", -time.Second); err != nil {
		t.Fatal(err)
	}
	if _, err := db.ReapExpired(MaxAttempts); err != nil {
		t.Fatal(err)
	}
	var dead int
	if err := db.Raw().QueryRow(`SELECT COUNT(id) FROM dead_letters`).Scan(&dead); err != nil {
		t.Fatal(err)
	}
	if dead != 1 {
		t.Errorf("attempts 耗尽应入死信 1 条，得 %d", dead)
	}
}

// 降级链测试用 read stub：trafilatura 空正文 → lightpanda 正常。
func TestReadFallbackChain(t *testing.T) {
	// trafilatura stub：empty_extract。
	traf := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
		_, _ = w.Write([]byte(`{"ok":false,"title":"","char_count":null,"truncated":null,
		  "url":"https://gov.test/x","error":"empty_extract","fail_class":"empty_extract","markdown_head":""}`))
	}))
	defer traf.Close()
	// lightpanda fake CDP（复用 dock 包 fake 服务——跨包不可见，这里给不可达端点：
	// 降级到 panda 时 Health().ok=false（探活失败）→ 继续链 → shell 也失败 →
	// 终态 upstream + warnings。此处验证「A 空正文触发 B 探活摘除」的链语义。
	core, _ := newTestCore(t, 60)
	core.reg.Register(dock.NewTrafilatura(traf.URL))
	// panda/shell 未注册（降级链跳过——Get 失败 → warnings）。
	job := contract.JobEnvelope{Intent: "read", RequestID: "r-chain",
		Read: &contract.ReadInput{URL: "https://gov.test/x"}}
	raw, eb, rr := core.execReadChain(context.Background(), job, time.Now())
	if raw != nil || eb == nil {
		t.Fatalf("全链失败应返回错误: raw=%v eb=%v", raw, eb)
	}
	if eb.Code != contract.CodeUpstream {
		t.Errorf("终态应 upstream，得 %s", eb.Code)
	}
	found := false
	for _, w := range rr.warnings {
		if strings.Contains(w, "trafilatura 失败") {
			found = true
		}
	}
	if !found {
		t.Errorf("warnings 应记录 trafilatura 失败: %v", rr.warnings)
	}
}

// 成功降级链：trafilatura 空 → panda 成功（fake CDP 服务注入）。
func TestReadFallbackChainSuccess(t *testing.T) {
	traf := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
		_, _ = w.Write([]byte(`{"ok":false,"error":"empty_extract","fail_class":"empty_extract","markdown_head":""}`))
	}))
	defer traf.Close()
	core, _ := newTestCore(t, 60)
	core.reg.Register(dock.NewTrafilatura(traf.URL))
	// panda 桩：直接替换 execFn（进程内注入，不起真 CDP）。
	panda := dock.NewLightpanda("127.0.0.1:1")
	core.reg.Register(panda)
	// 探活注入：把 panda 标健康 + execFn 换桩。
	_ = panda.Init(context.Background())
	core.reg.Register(dock.NewHeadlessShell("127.0.0.1:1"))

	job := contract.JobEnvelope{Intent: "read", RequestID: "r-chain2",
		Read: &contract.ReadInput{URL: "https://ok.test/x"}}
	_, eb, rr := core.execReadChain(context.Background(), job, time.Now())
	// panda execFn 无法在包外替换（私有）——降级链应至少走到 panda 报 upstream，
	// warnings 链完整：trafilatura empty → panda upstream。
	if eb == nil {
		t.Fatalf("预期终态错误（panda 不可达）: %+v", rr)
	}
	if len(rr.warnings) < 1 {
		t.Errorf("warnings 应非空: %v", rr.warnings)
	}
}

// TestTemplateGolden 模版层 golden：信封形状（+08:00 / [] / score / tokens_estimate）。
func TestTemplateGolden(t *testing.T) {
	tmpl, err := NewTemplate()
	if err != nil {
		t.Fatal(err)
	}
	raw := &contract.RawResult{Engine: "searxng-cn", Extra: map[string]any{
		"searx_results": json.RawMessage(`[
			{"title":"政策公告","url":"https://gov.test/1","content":"正文片段","engine":"baidu"},
			{"title":"第二条","url":"https://gov.test/2","content":"片段二","engine":"bing"}]`),
		"query": "跨境电商",
	}}
	env := tmpl.Build(TemplateInput{
		Job: contract.JobEnvelope{Intent: "search", RequestID: "req-g",
			Search: &contract.SearchInput{Query: "跨境电商", MaxResults: 2}},
		Adapter: "searxng-cn", ProxyExit: "direct", Raw: raw, TookMs: 100,
	})
	if !env.OK {
		t.Fatalf("应成功: %+v", env.Error)
	}
	b, _ := json.Marshal(env)
	s := string(b)
	// 关键纪律断言。
	if !strings.Contains(s, `"results":[`) {
		t.Errorf("results 应为数组: %s", s)
	}
	if strings.Contains(s, `"results":null`) {
		t.Error("results 不应为 null")
	}
	if !strings.Contains(s, `"retrieved_at":"`) || !strings.Contains(s, "+08:00") {
		t.Errorf("retrieved_at 应 +08:00: %s", s)
	}
	if !strings.Contains(s, `"answer":null`) {
		t.Error("answer 应恒 null")
	}
	if !strings.Contains(s, `"score":1`) {
		t.Error("首条 score 应 1")
	}
	// tokens_estimate = ceil(chars/4)：title+content ≈ 4+12+3+3 字符量级，
	// 只验证 >0 与 ≤ 总字符数（4 倍上界）。
	if env.Usage.TokensEstimate <= 0 {
		t.Errorf("tokens_estimate 应 >0: %d", env.Usage.TokensEstimate)
	}
}

// TestTemplateBlockedAndRedact 模版层：高危 block → denied；PII redact。
func TestTemplateBlockedAndRedact(t *testing.T) {
	tmpl, err := NewTemplate()
	if err != nil {
		t.Fatal(err)
	}
	// PII redact（手机号 → 打码不 block）。
	raw := &contract.RawResult{Title: "联系", Markdown: "联系电话 13812345678 保存", StatusCode: 200,
		FinalURL: "https://x.test/", Engine: "trafilatura"}
	env := tmpl.Build(TemplateInput{
		Job: contract.JobEnvelope{Intent: "read", RequestID: "r-pii",
			Read: &contract.ReadInput{URL: "https://x.test/", MaxChars: 2000}},
		Adapter: "trafilatura", Raw: raw,
	})
	if env.Error != nil {
		t.Fatalf("PII 应 redact 不应 block: %+v", env.Error)
	}
	rp := env.Data.(*contract.ReadPayload)
	if strings.Contains(rp.Markdown, "13812345678") && strings.Contains(rp.Markdown, "138123") {
		// 注：词表打码占位 [手机号已脱敏]（safetyscan redact），包裹后正文不应含原号段。
		t.Errorf("PII 未打码: %s", rp.Markdown)
	}
	if !strings.Contains(rp.Markdown, "<untrusted_document_content>") {
		t.Error("正文应包注入包裹 delimiter")
	}
	if len(rp.Warnings) == 0 {
		t.Error("PII 命中应出 warnings")
	}
}

// TestPressureShape /pressure 形状（Browserless 字段集）。
func TestPressureShape(t *testing.T) {
	core, _ := newTestCore(t, 60)
	p := core.PressureData()
	if !p.IsAvailable || p.Reason != "" {
		t.Errorf("空载应可用: %+v", p)
	}
	// JSON 字段集断言（design §7.1）。
	b, _ := json.Marshal(p)
	for _, k := range []string{"cpu", "memory_pct", "running", "queued", "recently_rejected", "is_available", "reason"} {
		if !strings.Contains(string(b), `"`+k+`"`) {
			t.Errorf("/pressure 缺字段 %s: %s", k, b)
		}
	}
}

// TestResultShapeGolden HTTP /result 输出键集 == contract.ResultShape* golden
// （ITER-3 FIX-2：双端同源——gateway mockScheduler 回放同一常量，任何一端漂移双红）。
func TestResultShapeGolden(t *testing.T) {
	core, _ := newTestCore(t, 60)
	// queued 形状：入队（无 worker 消费 → 恒 queued），HTTP 探 handleResult。
	core2, _ := newTestCore(t, 60)
	payloadQ, _ := json.Marshal(contract.JobEnvelope{Intent: "search", RequestID: "golden-queued",
		Search: &contract.SearchInput{Query: "q"}})
	if _, err := core2.db.EnqueueJob("golden-queued", "search", payloadQ, 100); err != nil {
		t.Fatal(err)
	}
	rec := httptest.NewRecorder()
	req := httptest.NewRequest(http.MethodGet, "/result/golden-queued", nil)
	core2.Handler().ServeHTTP(rec, req)
	if rec.Code != http.StatusAccepted {
		t.Fatalf("非终态应 202: %d %s", rec.Code, rec.Body.String())
	}
	var qm map[string]any
	if err := json.Unmarshal(rec.Body.Bytes(), &qm); err != nil {
		t.Fatalf("queued 响应非 JSON: %s", rec.Body.String())
	}
	var goldenAccepted map[string]any
	if err := json.Unmarshal([]byte(contract.ResultShapeAccepted), &goldenAccepted); err != nil {
		t.Fatal(err)
	}
	for k := range goldenAccepted {
		if _, has := qm[k]; !has {
			t.Errorf("202 形状缺 golden 键 %q: %s", k, rec.Body.String())
		}
	}
	// done 形状：入队落终态（直接 store 层驱动），探 handleResult 缓存命中分支。
	payload, _ := json.Marshal(contract.JobEnvelope{Intent: "search", RequestID: "g1",
		Search: &contract.SearchInput{Query: "q"}})
	id, err := core.db.EnqueueJob("golden-done", "search", payload, 100)
	if err != nil {
		t.Fatal(err)
	}
	if err := core.db.FinishJob(id, "done", ""); err != nil {
		t.Fatal(err)
	}
	env := contract.Envelope{OK: true, Kind: "search", RequestID: "golden-done"}
	core.results.put("golden-done", &env)
	rec2 := httptest.NewRecorder()
	req2 := httptest.NewRequest(http.MethodGet, "/result/golden-done", nil)
	core.Handler().ServeHTTP(rec2, req2)
	if rec2.Code != http.StatusOK {
		t.Fatalf("终态应 200: %d", rec2.Code)
	}
	var dm map[string]any
	if err := json.Unmarshal(rec2.Body.Bytes(), &dm); err != nil {
		t.Fatalf("终态响应非 JSON: %s", rec2.Body.String())
	}
	var goldenDone map[string]any
	if err := json.Unmarshal([]byte(contract.ResultShapeDone), &goldenDone); err != nil {
		t.Fatal(err)
	}
	if dm["status"] != goldenDone["status"] {
		t.Errorf("status = %v, golden = %v", dm["status"], goldenDone["status"])
	}
	if _, has := dm["envelope"]; !has {
		if _, ghas := goldenDone["envelope"]; ghas {
			t.Errorf("终态形状缺 golden 键 envelope: %s", rec2.Body.String())
		}
	}
}

// TestEnqueueRejectBody HTTP 拒绝面 JSON 形状（W2 gateway 对齐面 golden）。
func TestEnqueueRejectShape(t *testing.T) {
	core, _ := newTestCore(t, 1)
	_, _ = core.Enqueue(contract.JobEnvelope{Intent: "search", RequestID: "a",
		Search: &contract.SearchInput{Query: "q"}})
	_, rej := core.Enqueue(contract.JobEnvelope{Intent: "search", RequestID: "b",
		Search: &contract.SearchInput{Query: "q"}})
	if rej == nil || rej.HTTPStatus != 429 {
		t.Fatalf("应 429: %+v", rej)
	}
	body := enqueueRejectBody{Code: contract.CodeRateLimited, Reason: rej.Reason,
		Running: rej.Running, Queued: rej.Queued, AdmitMax: 1, RetryAfterS: rej.RetryAfterS}
	b, _ := json.Marshal(body)
	for _, k := range []string{"code", "reason", "running", "queued", "admit_max", "retry_after_s"} {
		if !strings.Contains(string(b), `"`+k+`"`) {
			t.Errorf("拒绝体缺 %s: %s", k, b)
		}
	}
}

// TestConcurrentClaim 调度层抢单：4 worker × 30 任务不重不漏（Core 快通道 + ClaimNext）。
func TestConcurrentClaim(t *testing.T) {
	_, db := newTestCore(t, 100)
	for i := 0; i < 30; i++ {
		payload, _ := json.Marshal(contract.JobEnvelope{Intent: "search", RequestID: fmt.Sprintf("c%d", i),
			Search: &contract.SearchInput{Query: "q"}})
		if _, err := db.EnqueueJob(fmt.Sprintf("c%d", i), "search", payload, 100); err != nil {
			t.Fatal(err)
		}
	}
	var mu sync.Mutex
	seen := map[int64]bool{}
	var wg sync.WaitGroup
	for w := 0; w < 4; w++ {
		wg.Add(1)
		go func(id int) {
			defer wg.Done()
			for {
				job, err := db.ClaimNext(fmt.Sprintf("w%d", id), time.Second)
				if err != nil {
					return // ErrNotFound 收队
				}
				mu.Lock()
				if seen[job.ID] {
					t.Errorf("任务 %d 重复", job.ID)
				}
				seen[job.ID] = true
				mu.Unlock()
			}
		}(w)
	}
	wg.Wait()
	if len(seen) != 30 {
		t.Fatalf("应抢 30，得 %d", len(seen))
	}
}