chii
9b689b2476
公开页抓取改为 Chrome 136 自洽身份 + 每节点 SQLite 养罐，Trafilatura 走 curl_cffi；MCP/README/接手说明与 09-02 现网复测对齐，避免消费方继续抄过期的站点三分表。 Co-authored-by: Cursor <cursoragent@cursor.com>
316 lines
11 KiB
Go
316 lines
11 KiB
Go
// core.go：Scheduler 核心（队列/worker 池/reaper/能力路由/降级链/背压）。
|
||
//
|
||
// 权威：docs/plan-final-20260901.md §2.1（入队路径/抢单/租约/reaper 定义）、
|
||
// docs/design-arch-20260901.md §3（降级链）、§4.3（ADMIT_MAX=60 背压）、
|
||
// §4.4（能力路由标签制）、§4.5（渲染互斥 P2-AR2：shell_active ⇒ panda 停新）。
|
||
package scheduler
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"log"
|
||
"sync"
|
||
"sync/atomic"
|
||
"time"
|
||
|
||
"onesvm.com/onesvm/browser-server/internal/config"
|
||
"onesvm.com/onesvm/browser-server/internal/contract"
|
||
"onesvm.com/onesvm/browser-server/internal/dock"
|
||
"onesvm.com/onesvm/browser-server/internal/session"
|
||
"onesvm.com/onesvm/browser-server/internal/store"
|
||
)
|
||
|
||
// 默认旋钮（env 可调，非密钥）。
|
||
const (
|
||
DefaultAdmitMax = 60 // ADMIT_MAX（running+queued 合计）
|
||
DefaultWorkers = 8 // worker 池
|
||
LeaseFor = 45 * time.Second // 抢单租约
|
||
ReaperEvery = 5 * time.Second // reaper 周期（A3.1：每 5s）
|
||
MaxAttempts = 2 // attempts≤2 瞬时错误重试
|
||
BackoffBase = 5 * time.Second // 退避基（available_at = now + 2^attempts*5s 语义，见 store.RetryWithBackoff）
|
||
proxyProbeInterval = 30 * time.Second // proxymanager 探活间隔
|
||
healthWatchEvery = 30 * time.Second // 适配器健康重探周期（ITER-1 F2：Init 一次性探活竞态的恢复路径）
|
||
|
||
healthWatchBootProbes = 10 // 冷启动收敛：最多短重探次数（ITER-2）
|
||
healthWatchBootInterval = 1 * time.Second // 冷启动短重探间隔（ITER-2）
|
||
)
|
||
|
||
// ResultStore 任务结果存储接口（store.DB 满足；测试可替换）。
|
||
type ResultStore interface {
|
||
EnqueueJob(reqID, intent string, payload []byte, priority int) (int64, error)
|
||
ClaimNext(worker string, leaseFor time.Duration) (*store.Job, error)
|
||
RenewLease(id int64, until time.Time) error
|
||
FinishJob(id int64, status, errMsg string) error
|
||
RetryWithBackoff(id int64, lastErr string, maxAttempts int, backoffBase time.Duration) (bool, error)
|
||
ReapExpired(maxAttempts int) (int, error)
|
||
CountRunning() (int, error)
|
||
CountQueued() (int, error)
|
||
JobByRequestID(reqID string) (*store.Job, error)
|
||
JobEnvelopeFromPayload(payload []byte) (contract.JobEnvelope, error)
|
||
}
|
||
|
||
// ShellHold 渲染互斥状态（shell 活跃时 lightpanda 停新；design §4.5 条款 6）。
|
||
type ShellHold struct{ active atomic.Bool }
|
||
|
||
// Set 标记 shell 活跃位。
|
||
func (h *ShellHold) Set(v bool) { h.active.Store(v) }
|
||
|
||
// Active shell 是否活跃。
|
||
func (h *ShellHold) Active() bool { return h.active.Load() }
|
||
|
||
// Stats 运行统计（/pressure /metrics 用；原子计数）。
|
||
type Stats struct {
|
||
Admitted atomic.Int64 // admitted_total
|
||
Rejected atomic.Int64 // rejected_total{reason}（合计）
|
||
ShellAct atomic.Int64 // rejected{reason=shell_active_panda_hold}
|
||
Recycles atomic.Int64 // recycles_total
|
||
WaitingSum atomic.Int64 // 排队等待 ms 累计（avg 用）
|
||
DoneCnt atomic.Int64
|
||
RunningG atomic.Int64 // 进程内在途（适配器执行中）
|
||
}
|
||
|
||
// st 内嵌统计（Core.stats() 返回）。
|
||
// Core 内嵌字段。
|
||
|
||
// Core scheduler 核心。
|
||
type Core struct {
|
||
db ResultStore
|
||
reg *dock.Registry
|
||
tmpl *Template
|
||
proxy *ProxyClient
|
||
log *log.Logger
|
||
hold ShellHold
|
||
workers int
|
||
admitMax int
|
||
|
||
// 快通道：bounded chan（design §4.2：热路径快通道；ACK 以 SQLite 落盘为准，
|
||
// chan 仅加速唤醒，worker 空闲时兜底轮询 ClaimNext）。
|
||
fast chan int64
|
||
|
||
results *resultCache // done 信封进程内缓存（/result 直读）
|
||
|
||
// proxy 可达性（overseas fail-closed：不可达 → 排队等待不降级直连，O3）。
|
||
proxyOK atomic.Bool
|
||
|
||
jar *session.Jar // 可选；nil 时不养罐（单测默认）
|
||
|
||
st *Stats
|
||
|
||
stopCh chan struct{}
|
||
stopOnce sync.Once
|
||
wg sync.WaitGroup
|
||
}
|
||
|
||
// NewCore 构造。db 为唯一写者；reg 已注册五适配器。
|
||
func NewCore(db ResultStore, reg *dock.Registry, tmpl *Template, proxyURL string, workers, admitMax int, logger *log.Logger) *Core {
|
||
if workers <= 0 {
|
||
workers = DefaultWorkers
|
||
}
|
||
if admitMax <= 0 {
|
||
admitMax = DefaultAdmitMax
|
||
}
|
||
if logger == nil {
|
||
logger = log.Default()
|
||
}
|
||
return &Core{
|
||
db: db,
|
||
reg: reg,
|
||
tmpl: tmpl,
|
||
proxy: NewProxyClient(proxyURL),
|
||
log: logger,
|
||
workers: workers,
|
||
admitMax: admitMax,
|
||
fast: make(chan int64, admitMax),
|
||
results: newResultCache(),
|
||
st: &Stats{},
|
||
stopCh: make(chan struct{}),
|
||
}
|
||
}
|
||
|
||
// SetJar 注入每节点 Cookie 罐（生产 main 调用；单测可不设）。
|
||
func (c *Core) SetJar(j *session.Jar) { c.jar = j }
|
||
|
||
// AdmitMax 背压上限。
|
||
func (c *Core) AdmitMax() int { return c.admitMax }
|
||
|
||
// Start 起 worker 池 + reaper + proxy 探活。
|
||
func (c *Core) Start(ctx context.Context) {
|
||
for i := 0; i < c.workers; i++ {
|
||
c.wg.Add(1)
|
||
go c.workerLoop(ctx, i)
|
||
}
|
||
c.wg.Add(1)
|
||
go c.reaperLoop(ctx)
|
||
c.wg.Add(1)
|
||
go c.proxyWatchLoop(ctx)
|
||
c.wg.Add(1)
|
||
go c.healthWatchLoop(ctx) // ITER-1 F2：适配器健康周期重探（摘除→恢复闭环）
|
||
if c.jar != nil {
|
||
c.wg.Add(1)
|
||
go c.jarReaperLoop(ctx)
|
||
}
|
||
}
|
||
|
||
// Stop 优雅停机（等待在途任务完成；reaper/worker 退出）。
|
||
func (c *Core) Stop() {
|
||
c.stopOnce.Do(func() { close(c.stopCh) })
|
||
c.wg.Wait()
|
||
}
|
||
|
||
// Enqueue 入队：ADMIT_MAX 检查（running+queued 合计）→ 渲染互斥检查 →
|
||
// SQLite 落盘（WAL）→ 快通道唤醒 → ACK {job_id, request_id}。
|
||
// 超限 503 + Retry-After + {running,queued} 现状；shell 活跃时 lightpanda 意图
|
||
// （render=light 的 read）入队侧拒绝（503 + reason=shell_active_panda_hold，P2-AR2）。
|
||
func (c *Core) Enqueue(job contract.JobEnvelope) (int64, *EnqueueReject) {
|
||
// 渲染互斥（入队侧执行，design §4.5 条款 6）：shell 活跃 → lightpanda 意图拒绝。
|
||
if job.Intent == "read" && job.Read != nil && job.Read.Region != contract.RegionOverseas {
|
||
// render=light 意图即「非降级直达 read」——首版入队侧以 intent=read 判定，
|
||
// shell 活跃期间暂停所有新渲染类（保守口径：read 意图全部暂停新入队）。
|
||
if c.hold.Active() && job.Read != nil && job.Read.Formats != nil && false {
|
||
// 保留分支位：特权 full 渲染显式通道（首版未启用）。
|
||
}
|
||
}
|
||
if c.hold.Active() && job.Intent == "read" && !isFallbackRead(job) {
|
||
return 0, &EnqueueReject{
|
||
HTTPStatus: 503,
|
||
Reason: "shell_active_panda_hold",
|
||
RetryAfterS: 5,
|
||
}
|
||
}
|
||
running, err := c.db.CountRunning()
|
||
if err != nil {
|
||
return 0, &EnqueueReject{HTTPStatus: 503, Reason: "store_unavailable", RetryAfterS: 2, Message: err.Error()}
|
||
}
|
||
queued, err := c.db.CountQueued()
|
||
if err != nil {
|
||
return 0, &EnqueueReject{HTTPStatus: 503, Reason: "store_unavailable", RetryAfterS: 2, Message: err.Error()}
|
||
}
|
||
if running+queued >= c.admitMax {
|
||
c.stats().Rejected.Add(1)
|
||
return 0, &EnqueueReject{
|
||
HTTPStatus: 429,
|
||
Reason: "admit_max",
|
||
RetryAfterS: 5,
|
||
Running: running,
|
||
Queued: queued,
|
||
}
|
||
}
|
||
payload, err := json.Marshal(job)
|
||
if err != nil {
|
||
return 0, &EnqueueReject{HTTPStatus: 503, Reason: "payload_marshal", Message: err.Error()}
|
||
}
|
||
id, err := c.db.EnqueueJob(job.RequestID, job.Intent, payload, job.Priority)
|
||
if err != nil {
|
||
c.stats().Rejected.Add(1)
|
||
return 0, &EnqueueReject{HTTPStatus: 503, Reason: "store_write", RetryAfterS: 2, Message: err.Error()}
|
||
}
|
||
c.stats().Admitted.Add(1)
|
||
// 快通道：非阻塞唤醒（chan 满则 worker 兜底轮询会取到）。
|
||
select {
|
||
case c.fast <- id:
|
||
default:
|
||
}
|
||
return id, nil
|
||
}
|
||
|
||
// isFallbackRead 降级链回灌的 read（lightpanda/panda 升级路径）不在互斥限制内——
|
||
// 降级是 shell_active 的结果，若 shell 刚回收则放行；仍活跃时统一走 shell 队列。
|
||
// 标记法：payload 由降级链重入队时 Priority=200（>100）识别。
|
||
func isFallbackRead(job contract.JobEnvelope) bool { return job.Priority >= 200 }
|
||
|
||
// EnqueueReject 拒绝详情（HTTP 面在 server.go 组装）。
|
||
type EnqueueReject struct {
|
||
HTTPStatus int // 429 | 503
|
||
Reason string // admit_max | shell_active_panda_hold | store_write...
|
||
RetryAfterS int
|
||
Running int
|
||
Queued int
|
||
Message string
|
||
}
|
||
|
||
// stats 统计指针。
|
||
func (c *Core) stats() *Stats { return c.st }
|
||
|
||
// Pressure /pressure 形状（design §7.1，抄 Browserless）。
|
||
type Pressure struct {
|
||
CPU float64 `json:"cpu"`
|
||
MemoryPct float64 `json:"memory_pct"`
|
||
Running int `json:"running"`
|
||
Queued int `json:"queued"`
|
||
RecentlyRejected int64 `json:"recently_rejected"`
|
||
IsAvailable bool `json:"is_available"`
|
||
Reason string `json:"reason"` // ok | full | store
|
||
}
|
||
|
||
// PressureData /pressure 输出（内存口径：本进程 RSS 比例经 cgroup 不可读时给 0，
|
||
// 注明估算；60 接纳上限是反滥用阀非内存指标）。
|
||
func (c *Core) PressureData() Pressure {
|
||
running, _ := c.db.CountRunning()
|
||
queued, _ := c.db.CountQueued()
|
||
p := Pressure{
|
||
Running: running,
|
||
Queued: queued,
|
||
RecentlyRejected: c.st.Rejected.Load(),
|
||
IsAvailable: running+queued < c.admitMax,
|
||
}
|
||
if !p.IsAvailable {
|
||
p.Reason = "full"
|
||
}
|
||
return p
|
||
}
|
||
|
||
// runJob 单任务执行：路由 → 降级链 → 模版层 → 结果落盘。
|
||
// 返回 error 仅租约续期/存储层故障；业务失败写 jobs.result/error。
|
||
func (c *Core) runJob(ctx context.Context, job *store.Job) {
|
||
env, err := c.db.JobEnvelopeFromPayload(job.Payload)
|
||
if err != nil {
|
||
_ = c.db.FinishJob(job.ID, "failed", "payload 解析失败: "+err.Error())
|
||
return
|
||
}
|
||
// 续租兜底：长任务每 lease/2 续一次（执行在适配器内，无法精确续）。
|
||
leaseCtx, cancel := context.WithTimeout(ctx, 120*time.Second)
|
||
defer cancel()
|
||
go c.leaseKeeper(leaseCtx, job.ID)
|
||
|
||
res, errBody, rr := c.executeWithRouting(leaseCtx, env)
|
||
if errBody != nil {
|
||
// 瞬时错误（timeout/upstream/unavailable）重试；其余直接 failed。
|
||
if contract.Retryable(errBody.Code) || errBody.Code == contract.CodeUnavailable {
|
||
ok, rerr := c.db.RetryWithBackoff(job.ID, errBody.Code+": "+errBody.Message, MaxAttempts, BackoffBase)
|
||
if rerr != nil {
|
||
c.log.Printf("job %d retry 写入失败: %v", job.ID, rerr)
|
||
}
|
||
_ = ok // false = RetryWithBackoff 已入死信并置 dead
|
||
return
|
||
}
|
||
_ = c.db.FinishJob(job.ID, "failed", errBody.Code+": "+errBody.Message)
|
||
// 失败信封也进缓存（/result 可见失败态）。
|
||
c.results.put(env.RequestID, &contract.Envelope{
|
||
OK: false, Kind: env.Intent, RequestID: env.RequestID,
|
||
TookMs: rr.tookMs, Error: errBody,
|
||
})
|
||
return
|
||
}
|
||
// 成功：模版层封装（guard → safetyscan → score 归一 → 信封）。
|
||
envelope := c.tmpl.Build(TemplateInput{
|
||
Job: env, Adapter: rr.adapter, ProxyExit: rr.proxy,
|
||
Raw: res, Warnings: rr.warnings, TookMs: rr.tookMs,
|
||
})
|
||
c.results.put(env.RequestID, envelope)
|
||
_ = c.db.FinishJob(job.ID, "done", "")
|
||
c.st.DoneCnt.Add(1)
|
||
}
|
||
|
||
// leaseKeeper 租约守护：每 10s 续租直至 ctx 结束（防长任务被 reaper 误收）。
|
||
func (c *Core) leaseKeeper(ctx context.Context, id int64) {
|
||
t := time.NewTicker(10 * time.Second)
|
||
defer t.Stop()
|
||
for {
|
||
select {
|
||
case <-ctx.Done():
|
||
return
|
||
case <-t.C:
|
||
_ = c.db.RenewLease(id, config.Now().Add(LeaseFor))
|
||
}
|
||
}
|
||
}
|