jobs.js
6.28 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
import { randomUUID } from 'node:crypto'
const JOBS = 'vk_jobs'
const BATCH = 20
const LEASE_MS = 60_000
const MAX_ATTEMPTS = 5
const BACKOFF_BASE_MS = 30_000
/** 到点后直接映射为状态机 worker 动作的 job。 */
const TASK_ACTION_KINDS = Object.freeze(['open_expire', 'hold_timeout', 'checkpoint_miss', 'contest_close'])
const WARN_KINDS = Object.freeze(['checkpoint_warn', 'review_warn'])
/**
* vk_jobs 的领取与执行：`FOR UPDATE SKIP LOCKED` + 租约。每个 job 执行时重新读任务当前状态，
* 状态已变就无事可做，所以重复或过时的 job 无害。
*
* @param {{
* db: import('knex').Knex,
* tasks: {
* actAsWorker(ctx: object, input: { task_id: string, action: any }): Promise<{ ok: boolean }>,
* warn(ctx: object, input: { task_id: string, kind: any }): Promise<void>,
* expireTransfers(ctx: object, input: { task_id: string }): Promise<void>,
* autoSelectWinner(ctx: object, input: { task_id: string }): Promise<void>,
* arbitrate(ctx: object, input: { task_id: string }): Promise<void>,
* },
* help: { notify(ctx: object, input: { help_id: string }): Promise<void> },
* sendDigest(ctx: object, input: { user_id: string }): Promise<void>,
* runDailyGame(ctx: object): Promise<unknown>,
* conservationProblems(): Promise<Array<{ rule: string, id: string, detail: string }>>,
* logger: { error(message: string, fields: object): void },
* }} deps
*/
export function createJobs({ db, tasks, help, sendDigest, runDailyGame, conservationProblems, logger }) {
if (typeof db !== 'function') throw new Error('createJobs: db is required')
/**
* 领取并执行一批到期 job；返回本轮执行的条数。
* @param {object} ctx
* @returns {Promise<number>}
*/
async function runDue(ctx) {
const now = new Date()
const jobs = await db.transaction(async (trx) => {
const due = await trx(JOBS)
.where((q) => q.where({ status: 'pending' }).where('run_at', '<=', now).orWhere((r) => r.where({ status: 'running' }).where('lease_until', '<', now)))
.orderBy('run_at', 'asc')
.limit(BATCH)
.forUpdate()
.skipLocked()
if (due.length > 0) await trx(JOBS).whereIn('id', due.map((job) => job.id)).update({ status: 'running', lease_until: new Date(now.getTime() + LEASE_MS), updated_at: now })
return due
})
for (const job of jobs) {
// worker 的全局错误处理：单个 job 失败只记录并退避重试，不拖垮整轮。
try {
await execute(ctx, job)
await db(JOBS).where({ id: job.id }).update({ status: 'done', lease_until: null, updated_at: new Date() })
} catch (err) {
const attempts = job.attempts + 1
const message = err instanceof Error ? (err.stack || err.message) : String(err)
logger.error('job 执行失败', { job_id: job.id, kind: job.kind, ref_id: job.ref_id, attempts, error: message })
await db(JOBS).where({ id: job.id }).update({
status: attempts >= MAX_ATTEMPTS ? 'failed' : 'pending',
attempts,
run_at: new Date(Date.now() + BACKOFF_BASE_MS * 2 ** job.attempts),
lease_until: null,
last_error: message,
updated_at: new Date(),
})
}
}
return jobs.length
}
/**
* @param {object} ctx
* @param {{ id: string, kind: string, ref_id: string|null }} job
*/
async function execute(ctx, job) {
if (job.kind === 'verify_conservation') {
// 每日对账：有任何不守恒就让 job 失败（记错误日志即告警），并附上出问题的数据 id。
const problems = await conservationProblems()
if (problems.length > 0) throw new Error(`conservation check failed: ${JSON.stringify(problems.slice(0, 20))}`)
return
}
if (job.kind === 'daily_game') return void await runDailyGame(ctx)
if (job.ref_id === null) throw new Error(`vk_jobs.ref_id missing for ${job.kind}, id: ${job.id}`)
const task_id = job.ref_id
if (job.kind === 'daily_digest') return sendDigest(ctx, { user_id: job.ref_id })
if (TASK_ACTION_KINDS.includes(job.kind)) return void await tasks.actAsWorker(ctx, { task_id, action: { type: job.kind } })
if (WARN_KINDS.includes(job.kind)) return tasks.warn(ctx, { task_id, kind: job.kind })
if (job.kind === 'transfer_expire') return tasks.expireTransfers(ctx, { task_id })
if (job.kind === 'review_auto_accept') {
const row = await db('vk_tasks').where({ id: task_id }).first('template')
if (row === undefined) throw new Error(`vk_tasks row missing for review_auto_accept, id: ${task_id}`)
// 计件活到期直接结算；擂台赛由大模型按清单选赢家。
if (row.template === 'contest') return tasks.autoSelectWinner(ctx, { task_id })
return void await tasks.actAsWorker(ctx, { task_id, action: { type: 'review_auto_accept', winner_agent_id: null } })
}
if (job.kind === 'dispute_arbitrate') return tasks.arbitrate(ctx, { task_id })
if (job.kind === 'help_notify') return help.notify(ctx, { help_id: job.ref_id })
// 其余类型（Webhook、日报、对账）在 W4 接入。
throw new Error(`Unhandled job kind: ${job.kind}, id: ${job.id}`)
}
/**
* 排一条 job；同一类型、同一对象、同一时刻只排一次（dedupe_key UNIQUE），重复调用无害。
* @param {object} _ctx
* @param {{ kind: string, ref_id: string|null, run_at: Date }} input
*/
async function enqueue(_ctx, { kind, ref_id, run_at }) {
await db(JOBS)
.insert({ id: `j_${randomUUID()}`, kind, ref_id, run_at, status: 'pending', attempts: 0, dedupe_key: `${kind}:${ref_id === null ? '' : ref_id}:${run_at.toISOString()}`, created_at: new Date() })
.onConflict('dedupe_key')
.ignore()
}
/**
* 确保下一次每日对账（UTC 01:00）与每日称号结算（UTC 00:10）已经排上。
* @param {object} ctx
*/
async function ensureDaily(ctx) {
const now = new Date()
for (const [kind, hour, minute] of /** @type {const} */ ([['verify_conservation', 1, 0], ['daily_game', 0, 10]])) {
const next = new Date(Date.UTC(now.getUTCFullYear(), now.getUTCMonth(), now.getUTCDate(), hour, minute))
if (next <= now) next.setUTCDate(next.getUTCDate() + 1)
await enqueue(ctx, { kind, ref_id: null, run_at: next })
}
}
return { runDue, enqueue, ensureDaily }
}