FazBrowse GitHub Viewer | Trending |
URL:
| Home
Tools: [Download Repo ZIP]   [Original HTTPS Page]

refactor(worker): extract heartbeat lifecycle · boringnode/queue@bf341bd · GitHub

/ queue Public

Commit bf341bd

Browse files
RomainLanz
committed
refactor(worker): extract heartbeat lifecycle
1 parent ae431d5 commit bf341bd

4 files changed

Lines changed: 255 additions & 186 deletions

File tree

‎src/worker_heartbeat.ts‎

Lines changed: 88 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,88 @@
1+
import debug from './debug.js'
2+
import type { Adapter } from './contracts/adapter.js'
3+
import type { JobPool } from './job_pool.js'
4+
5+
export type HeartbeatOperationWrapper = <T>(operation: () => Promise<T>) => Promise<T>
6+
7+
export interface WorkerHeartbeatOptions {
8+
workerId: string
9+
queues: readonly string[]
10+
interval: number
11+
pool: JobPool
12+
adapter: Adapter
13+
wrapInternal: HeartbeatOperationWrapper
14+
}
15+
16+
/**
17+
* Renews the leases of Jobs owned by one WorkerSession.
18+
*/
19+
export class WorkerHeartbeat {
20+
readonly #workerId: string
21+
readonly #queues: readonly string[]
22+
readonly #interval: number
23+
readonly #pool: JobPool
24+
readonly #adapter: Adapter
25+
readonly #wrapInternal: HeartbeatOperationWrapper
26+
27+
#timer?: NodeJS.Timeout
28+
#renewal?: Promise<void>
29+
30+
constructor(options: WorkerHeartbeatOptions) {
31+
this.#workerId = options.workerId
32+
this.#queues = Object.freeze([...options.queues])
33+
this.#interval = options.interval
34+
this.#pool = options.pool
35+
this.#adapter = options.adapter
36+
this.#wrapInternal = options.wrapInternal
37+
}
38+
39+
start(): void {
40+
this.#stopTimer()
41+
42+
this.#timer = setInterval(() => {
43+
if (this.#renewal) return
44+
45+
const renewal = this.#renewActiveJobs()
46+
this.#renewal = renewal
47+
48+
const clearRenewal = () => {
49+
if (this.#renewal === renewal) {
50+
this.#renewal = undefined
51+
}
52+
}
53+
54+
void renewal.then(clearRenewal, clearRenewal)
55+
}, this.#interval)
56+
57+
this.#timer.unref?.()
58+
}
59+
60+
async stop(): Promise<void> {
61+
this.#stopTimer()
62+
await this.#renewal
63+
}
64+
65+
#stopTimer(): void {
66+
if (!this.#timer) return
67+
68+
clearInterval(this.#timer)
69+
this.#timer = undefined
70+
}
71+
72+
async #renewActiveJobs(): Promise<void> {
73+
if (this.#pool.isEmpty()) return
74+
75+
const jobIdsByQueue = this.#pool.activeJobIdsByQueue()
76+
77+
for (const queue of this.#queues) {
78+
const jobIds = jobIdsByQueue.get(queue)
79+
if (!jobIds || jobIds.length === 0) continue
80+
81+
try {
82+
await this.#wrapInternal(() => this.#adapter.renewJobs(queue, jobIds))
83+
} catch (error) {
84+
debug('worker %s: failed to renew jobs on queue %s: %O', this.#workerId, queue, error)
85+
}
86+
}
87+
}
88+
}

‎src/worker_session.ts‎

Lines changed: 14 additions & 54 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ import debug from './debug.js'
33
import * as errors from './exceptions.js'
44
import { parse } from './utils.js'
55
import { JobPool } from './job_pool.js'
6+
import { WorkerHeartbeat, type HeartbeatOperationWrapper } from './worker_heartbeat.js'
67
import { Locator } from './locator.js'
78
import type { SingleJobDispatchRequest } from './job_dispatch_runtime.js'
89
import type { Adapter, AcquiredJob } from './contracts/adapter.js'
@@ -18,7 +19,7 @@ interface PoolFillResult {
1819
errors: unknown[]
1920
}
2021

21-
export type InternalOperationWrapper = <T>(operation: () => Promise<T>) => Promise<T>
22+
export type InternalOperationWrapper = HeartbeatOperationWrapper
2223
export type JobExecutor = Pick<JobExecutionRuntime, 'execute'>
2324
export interface ScheduleDispatcher {
2425
dispatch(request: SingleJobDispatchRequest): Promise<unknown>
@@ -58,6 +59,7 @@ export class WorkerSession {
5859
readonly #scheduleDispatcher: ScheduleDispatcher
5960
readonly #wrapInternal: InternalOperationWrapper
6061
readonly #settings: WorkerSessionSettings
62+
readonly #heartbeat: WorkerHeartbeat
6163

6264
#mode?: SessionMode
6365
#running = false
@@ -70,8 +72,6 @@ export class WorkerSession {
7072
#pool = new JobPool()
7173
#fillOperation?: Promise<PoolFillResult>
7274
#lastStalledCheck = 0
73-
#heartbeatTimer?: NodeJS.Timeout
74-
#heartbeatRenewal?: Promise<void>
7575
#delayController?: AbortController
7676

7777
constructor(options: WorkerSessionOptions) {
@@ -82,6 +82,14 @@ export class WorkerSession {
8282
this.#scheduleDispatcher = options.scheduleDispatcher
8383
this.#wrapInternal = options.wrapInternal
8484
this.#settings = options.settings
85+
this.#heartbeat = new WorkerHeartbeat({
86+
workerId: options.workerId,
87+
queues: this.#queues,
88+
interval: Math.max(Math.floor(options.settings.stalledThreshold / 2), 1),
89+
pool: this.#pool,
90+
adapter: options.adapter,
91+
wrapInternal: options.wrapInternal,
92+
})
8593
}
8694

8795
assertQueues(queues: readonly string[]): void {
@@ -192,8 +200,7 @@ export class WorkerSession {
192200
await this.#pool.drain()
193201
}
194202

195-
this.#stopHeartbeat()
196-
await this.#heartbeatRenewal
203+
await this.#heartbeat.stop()
197204
await this.#closeCycleGenerator()
198205

199206
if (this.#startCompletion) {
@@ -260,7 +267,7 @@ export class WorkerSession {
260267
}
261268

262269
async *#cycles(): AsyncGenerator<WorkerCycle, void, unknown> {
263-
this.#startHeartbeat()
270+
this.#heartbeat.start()
264271

265272
try {
266273
while (this.#running) {
@@ -326,7 +333,7 @@ export class WorkerSession {
326333
}
327334
} finally {
328335
if (!this.#stopping) {
329-
this.#stopHeartbeat()
336+
await this.#heartbeat.stop()
330337
}
331338
}
332339
}
@@ -468,53 +475,6 @@ export class WorkerSession {
468475
}
469476
}
470477

471-
#startHeartbeat(): void {
472-
this.#stopHeartbeat()
473-
474-
const interval = Math.max(Math.floor(this.#settings.stalledThreshold / 2), 1)
475-
476-
this.#heartbeatTimer = setInterval(() => {
477-
if (this.#heartbeatRenewal) return
478-
479-
const renewal = this.#renewActiveJobs()
480-
this.#heartbeatRenewal = renewal
481-
482-
const clearRenewal = () => {
483-
if (this.#heartbeatRenewal === renewal) {
484-
this.#heartbeatRenewal = undefined
485-
}
486-
}
487-
488-
void renewal.then(clearRenewal, clearRenewal)
489-
}, interval)
490-
491-
this.#heartbeatTimer.unref?.()
492-
}
493-
494-
#stopHeartbeat(): void {
495-
if (this.#heartbeatTimer) {
496-
clearInterval(this.#heartbeatTimer)
497-
this.#heartbeatTimer = undefined
498-
}
499-
}
500-
501-
async #renewActiveJobs(): Promise<void> {
502-
if (this.#pool.isEmpty()) return
503-
504-
const jobIdsByQueue = this.#pool.activeJobIdsByQueue()
505-
506-
for (const queue of this.#queues) {
507-
const jobIds = jobIdsByQueue.get(queue)
508-
if (!jobIds || jobIds.length === 0) continue
509-
510-
try {
511-
await this.#wrapInternal(() => this.#adapter.renewJobs(queue, jobIds))
512-
} catch (error) {
513-
debug('worker %s: failed to renew jobs on queue %s: %O', this.#workerId, queue, error)
514-
}
515-
}
516-
}
517-
518478
async #dispatchDueSchedules(): Promise<void> {
519479
while (this.#running) {
520480
const schedule = await this.#wrapInternal(() => this.#adapter.claimDueSchedule())

‎tests/worker_heartbeat.spec.ts‎

Lines changed: 153 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,153 @@
1+
import { setTimeout } from 'node:timers/promises'
2+
import { test } from '@japa/runner'
3+
import { JobPool } from '../src/job_pool.js'
4+
import { WorkerHeartbeat } from '../src/worker_heartbeat.js'
5+
import { ControllableAdapter } from './_mocks/controllable_adapter.js'
6+
import { trackPromise } from './_utils/track_promise.js'
7+
8+
function createHeartbeat(adapter: ControllableAdapter, pool: JobPool, interval = 10) {
9+
return new WorkerHeartbeat({
10+
workerId: 'test-worker',
11+
queues: ['default'],
12+
interval,
13+
pool,
14+
adapter,
15+
wrapInternal: (operation) => operation(),
16+
})
17+
}
18+
19+
async function acquireIntoPool(
20+
adapter: ControllableAdapter,
21+
pool: JobPool,
22+
id: string,
23+
execution: Promise<void>
24+
): Promise<void> {
25+
adapter.setWorkerId('test-worker')
26+
await adapter.pushOn('default', {
27+
id,
28+
name: 'TestJob',
29+
payload: {},
30+
attempts: 0,
31+
priority: 0,
32+
})
33+
34+
const job = await adapter.popFrom('default')
35+
if (!job) throw new Error('Expected an acquired Job')
36+
pool.add(job, 'default', execution)
37+
}
38+
39+
test.group('WorkerHeartbeat', () => {
40+
test('waits for an in-flight renewal before stopping', async ({ assert, cleanup }) => {
41+
const adapter = new ControllableAdapter()
42+
const pool = new JobPool()
43+
const execution = Promise.withResolvers<void>()
44+
await acquireIntoPool(adapter, pool, 'renewed-job', execution.promise)
45+
adapter.renewals.block(1)
46+
const heartbeat = createHeartbeat(adapter, pool, 1)
47+
48+
cleanup(async () => {
49+
adapter.releaseAll()
50+
execution.resolve()
51+
await heartbeat.stop()
52+
await pool.drain()
53+
})
54+
55+
heartbeat.start()
56+
await adapter.renewals.waitForStarted()
57+
58+
const stop = trackPromise(heartbeat.stop())
59+
assert.isFalse(stop.settled)
60+
61+
adapter.renewals.release(1)
62+
await stop.promise
63+
})
64+
65+
test('does not overlap renewals', async ({ assert, cleanup }) => {
66+
const adapter = new ControllableAdapter()
67+
const pool = new JobPool()
68+
const execution = Promise.withResolvers<void>()
69+
await acquireIntoPool(adapter, pool, 'renewed-job', execution.promise)
70+
adapter.renewals.block(1)
71+
const heartbeat = createHeartbeat(adapter, pool, 1)
72+
73+
cleanup(async () => {
74+
adapter.releaseAll()
75+
execution.resolve()
76+
await heartbeat.stop()
77+
await pool.drain()
78+
})
79+
80+
heartbeat.start()
81+
await adapter.renewals.waitForStarted()
82+
await setTimeout(20)
83+
84+
assert.equal(adapter.renewals.calls, 1)
85+
86+
adapter.renewals.release(1)
87+
await heartbeat.stop()
88+
})
89+
90+
test('absorbs renewal errors', async ({ assert, cleanup }) => {
91+
const adapter = new ControllableAdapter()
92+
const pool = new JobPool()
93+
const execution = Promise.withResolvers<void>()
94+
await acquireIntoPool(adapter, pool, 'renewed-job', execution.promise)
95+
adapter.renewals.fail(1, new Error('renewal failed'))
96+
const heartbeat = createHeartbeat(adapter, pool, 1)
97+
98+
cleanup(async () => {
99+
execution.resolve()
100+
await heartbeat.stop()
101+
await pool.drain()
102+
})
103+
104+
heartbeat.start()
105+
await adapter.renewals.waitForSettled(1)
106+
await heartbeat.stop()
107+
108+
assert.isAtLeast(adapter.renewals.calls, 1)
109+
})
110+
111+
test('prevents an active Job from being recovered as stalled', async ({ assert, cleanup }) => {
112+
const adapter = new ControllableAdapter()
113+
const pool = new JobPool()
114+
const execution = Promise.withResolvers<void>()
115+
await acquireIntoPool(adapter, pool, 'long-running-job', execution.promise)
116+
const heartbeat = createHeartbeat(adapter, pool, 20)
117+
118+
cleanup(async () => {
119+
execution.resolve()
120+
await heartbeat.stop()
121+
await pool.drain()
122+
})
123+
124+
heartbeat.start()
125+
await adapter.renewals.waitForStarted()
126+
await setTimeout(60)
127+
128+
const recovered = await adapter.recoverStalledJobs('default', 40, 1)
129+
assert.equal(recovered, 0)
130+
})
131+
132+
test('stops renewing after stop resolves', async ({ assert, cleanup }) => {
133+
const adapter = new ControllableAdapter()
134+
const pool = new JobPool()
135+
const execution = Promise.withResolvers<void>()
136+
await acquireIntoPool(adapter, pool, 'heartbeat-job', execution.promise)
137+
const heartbeat = createHeartbeat(adapter, pool)
138+
139+
cleanup(async () => {
140+
execution.resolve()
141+
await heartbeat.stop()
142+
await pool.drain()
143+
})
144+
145+
heartbeat.start()
146+
await adapter.renewals.waitForStarted()
147+
await heartbeat.stop()
148+
149+
const renewalsAtStop = adapter.renewals.calls
150+
await setTimeout(40)
151+
assert.equal(adapter.renewals.calls, renewalsAtStop)
152+
})
153+
})

0 commit comments

Comments
 (0)
Back | FazBrowse Home | New Git URL