jobs.js
3.35 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
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>,
* },
* logger: { error(message: string, fields: object): void },
* }} deps
*/
export function createJobs({ db, tasks, 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 (TASK_ACTION_KINDS.includes(job.kind) || WARN_KINDS.includes(job.kind)) {
if (job.ref_id === null) throw new Error(`vk_jobs.ref_id missing for ${job.kind}, id: ${job.id}`)
if (WARN_KINDS.includes(job.kind)) await tasks.warn(ctx, { task_id: job.ref_id, kind: job.kind })
else await tasks.actAsWorker(ctx, { task_id: job.ref_id, action: { type: job.kind } })
return
}
// 其余类型（自动确认、仲裁、Webhook、日报、对账）在 W3–W4 接入。
throw new Error(`Unhandled job kind: ${job.kind}, id: ${job.id}`)
}
return { runDue }
}