From b45bb9adb4115b1970e17f5b0438ec292cf9e645 Mon Sep 17 00:00:00 2001 From: =?utf8?q?J=C3=A9r=C3=B4me=20Benoit?= Date: Tue, 20 Oct 2026 19:57:42 +0200 Subject: [PATCH] fix: recover runtime task function operations (#3369) * fix: recover runtime task function operations Closes #3365 Closes #3366 Closes #3367 * test: cover unsent task function targets Cover partial-send failures before uncontacted targets with native acknowledgements, failures and exits on both transports. Clarify that successful mutations await restoration attempts. * docs: clarify rejected task function calls --- docs/api.md | 8 +- src/pools/abstract-pool.ts | 154 ++++- tests/pools/task-function-recovery.test.mjs | 559 ++++++++++++++++++ .../taskFunctionRecoveryWorker.mjs | 49 ++ 4 files changed, 736 insertions(+), 34 deletions(-) create mode 100644 tests/pools/task-function-recovery.test.mjs create mode 100644 tests/worker-files/taskFunctionRecoveryWorker.mjs diff --git a/docs/api.md b/docs/api.md index b06f56dda..d5fed6045 100644 --- a/docs/api.md +++ b/docs/api.md @@ -130,13 +130,15 @@ With no workers, this method resolves to `true` without changing the default sel ### Runtime task function changes -Pool-wide calls to `addTaskFunction()`, `removeTaskFunction()` and `setDefaultTaskFunction()` run in call order. A failed call does not block later calls or undo changes already applied to workers. +Pool-wide calls to `addTaskFunction()`, `removeTaskFunction()` and `setDefaultTaskFunction()` run in call order. A rejected call does not prevent later calls or undo changes already applied to workers. -If workers are present when a change starts, each must respond or exit before the call settles. At least one successful result is required; any reported failure rejects the change. Thread pools preserve operation results sent before an unexpected worker exit. +Workers contacted by a change must respond or exit before the call settles, including when a later send fails. Send failures preserve the original error; otherwise, at least one successful response is required and any failed response rejects the change. Thread pools preserve operation results sent before an unexpected worker exit. New and replacement workers wait for ongoing task function changes. They accept tasks only after restoring successfully registered task functions, the selected default and scheduling options. Workers being terminated remain unavailable. -Custom worker subclasses that defer task function operations must echo the request's `taskFunctionOperationId` in their acknowledgement, or retain the original `taskFunctionProperties` object when using the standard transport. An acknowledgement without a recoverable UUID rejects the operation. Pool-wide changes still wait for all original targets to respond or exit. Status-only overrides that discard this correlation must be updated. +Successful calls resolve only after a restoration attempt completes for workers whose initialization failed. If restoration fails again, the change still succeeds, the worker remains unavailable, and the pool emits `error` when events are enabled. + +Custom worker subclasses that defer task function operations must echo the request's `taskFunctionOperationId` in their acknowledgement, or retain the original `taskFunctionProperties` object when using the standard transport. An acknowledgement without a recoverable UUID rejects the operation. Pool-wide changes still wait for all contacted workers to respond or exit. Status-only overrides that discard this correlation must be updated. ### Pool options diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts index ec606b87e..43de06fc8 100644 --- a/src/pools/abstract-pool.ts +++ b/src/pools/abstract-pool.ts @@ -446,6 +446,14 @@ export abstract class AbstractPool< */ private defaultTaskFunctionName?: string + /** + * Failed bootstraps retain native readiness outside active waits. + */ + private readonly failedWorkerTaskFunctionBootstraps = new Map< + IWorkerNode, + boolean + >() + private pendingTaskFunctionOperations = 0 /** @@ -753,15 +761,10 @@ export abstract class AbstractPool< /** @inheritDoc */ public listTaskFunctionsProperties (): TaskFunctionProperties[] { - let fallback: TaskFunctionProperties[] | undefined - for (const workerNode of this.workerNodes) { - const properties = workerNode.info.taskFunctionsProperties - if (Array.isArray(properties) && properties.length > 0) { - if (workerNode.info.ready) return properties - fallback ??= properties - } - } - return fallback ?? [] + return ( + this.getTaskFunctionsPropertiesWorkerNode()?.info + .taskFunctionsProperties ?? [] + ) } /** @inheritDoc */ @@ -1314,18 +1317,11 @@ export abstract class AbstractPool< this.taskFunctions.size > 0 || this.defaultTaskFunctionName != null ) { - const bootstrap: { promise?: Promise; ready: boolean } = { - ready: false, - } - workerNode.info.ready = false - this.workerTaskFunctionBootstraps.set(workerNode, bootstrap) - bootstrap.promise = this.replayTaskFunctionsAfterOperations( - workerNode, - bootstrap + this.startWorkerTaskFunctionBootstrap(workerNode).catch( + (error: unknown) => { + this.emitter?.emit(PoolEvents.error, error) + } ) - bootstrap.promise.catch((error: unknown) => { - this.emitter?.emit(PoolEvents.error, error) - }) } this.afterWorkerNodeSetup(workerNodeKey) return workerNodeKey @@ -2022,6 +2018,20 @@ export abstract class AbstractPool< `Task function operation acknowledgement from worker ${String(workerId)} is missing taskFunctionOperationId` ) + private getTaskFunctionsPropertiesWorkerNode (): + | IWorkerNode + | undefined { + let fallback: IWorkerNode | undefined + for (const workerNode of this.workerNodes) { + const properties = workerNode.info.taskFunctionsProperties + if (Array.isArray(properties) && properties.length > 0) { + if (workerNode.info.ready) return workerNode + fallback ??= workerNode + } + } + return fallback + } + /** * Gets task function worker choice strategy, if any. * @param name - The task function name. @@ -2049,11 +2059,20 @@ export abstract class AbstractPool< */ private readonly getTaskFunctionWorkerNodeKeys = ( name?: string, - taskFunctionsProperties: readonly TaskFunctionProperties[] = this.listTaskFunctionsProperties() + taskFunctionsProperties?: readonly TaskFunctionProperties[] ): readonly number[] | undefined => { name = name ?? DEFAULT_TASK_NAME + const propertiesWorkerNode = + taskFunctionsProperties == null || name === DEFAULT_TASK_NAME + ? this.getTaskFunctionsPropertiesWorkerNode() + : undefined + taskFunctionsProperties ??= + propertiesWorkerNode?.info.taskFunctionsProperties ?? [] if (name === DEFAULT_TASK_NAME) { - name = taskFunctionsProperties[1]?.name + name = + propertiesWorkerNode?.info.ready === true + ? (taskFunctionsProperties[1]?.name ?? this.defaultTaskFunctionName) + : (this.defaultTaskFunctionName ?? taskFunctionsProperties[1]?.name) } // Keep the affinity lookup free of iterator allocation. // eslint-disable-next-line @typescript-eslint/prefer-for-of @@ -2063,7 +2082,7 @@ export abstract class AbstractPool< return taskFunctionProperties.workerNodeKeys } } - return undefined + return this.taskFunctions.get(name)?.taskFunctionProperties.workerNodeKeys } private getTasksQueuePriority ( @@ -2409,8 +2428,13 @@ export abstract class AbstractPool< const workerNodeKey = this.getWorkerNodeKeyByWorkerId(workerId) const workerNode = this.workerNodes[workerNodeKey] const bootstrap = this.workerTaskFunctionBootstraps.get(workerNode) + const failedBootstrap = + this.failedWorkerTaskFunctionBootstraps.has(workerNode) if (bootstrap != null) bootstrap.ready = ready - workerNode.info.ready = bootstrap == null + else if (failedBootstrap) { + this.failedWorkerTaskFunctionBootstraps.set(workerNode, ready) + } + workerNode.info.ready = bootstrap == null && !failedBootstrap workerNode.info.taskFunctionsProperties = taskFunctionsProperties this.sendStatisticsMessageToWorker(workerNodeKey) this.setTasksQueuePriority(workerNodeKey) @@ -2609,7 +2633,11 @@ export abstract class AbstractPool< operation: () => Promise ): Promise { const result = (this.taskFunctionOperationsQueue ?? Promise.resolve()).then( - operation + async () => { + const result = await operation() + await this.retryFailedWorkerTaskFunctionBootstraps() + return result + } ) this.taskFunctionOperationsQueue = result.then( EMPTY_FUNCTION, @@ -2764,6 +2792,7 @@ export abstract class AbstractPool< */ private removeWorkerNode (workerNode: IWorkerNode): void { this.workerTaskFunctionBootstraps.delete(workerNode) + this.failedWorkerTaskFunctionBootstraps.delete(workerNode) const workerNodeKey = this.workerNodes.indexOf(workerNode) if (workerNodeKey !== -1) { this.workerNodes.splice(workerNodeKey, 1) @@ -2845,7 +2874,15 @@ export abstract class AbstractPool< this.checkAndEmitReadyEvent() } catch (error) { // Retired bootstraps no longer block waiting submissions. - if (this.workerNodes.includes(workerNode)) throw error + if (this.workerNodes.includes(workerNode)) { + if (!this.destroying && !this.terminatingWorkerNodes.has(workerNode)) { + this.failedWorkerTaskFunctionBootstraps.set( + workerNode, + bootstrap.ready + ) + } + throw error + } } finally { this.workerTaskFunctionBootstraps.delete(workerNode) } @@ -2874,6 +2911,23 @@ export abstract class AbstractPool< } } + private async retryFailedWorkerTaskFunctionBootstraps (): Promise { + if (this.destroying || this.failedWorkerTaskFunctionBootstraps.size === 0) { + return + } + const promises: Promise[] = [] + for (const [workerNode, ready] of this.failedWorkerTaskFunctionBootstraps) { + if (this.getWorkerNodeKeyForBootstrap(workerNode) !== -1) { + const promise = this.startWorkerTaskFunctionBootstrap(workerNode, ready) + promise.catch((error: unknown) => { + this.emitter?.emit(PoolEvents.error, error) + }) + promises.push(promise) + } + } + await Promise.allSettled(promises) + } + private async sendKillMessageToWorker ( workerNodeKey: number, timeout = 1000 @@ -3022,6 +3076,8 @@ export abstract class AbstractPool< let remainingResponses = responsesReceived.size let errorResponse: MessageValue | undefined let protocolError: Error | undefined + let sendError: unknown + let sendFailed = false const { promise, reject, resolve } = Promise.withResolvers() const listeners: { cleanup: () => void @@ -3030,7 +3086,9 @@ export abstract class AbstractPool< }[] = [] const complete = (): void => { if (remainingResponses !== 0) return - if (protocolError != null) { + if (sendFailed) { + reject(sendError) + } else if (protocolError != null) { reject(protocolError) } else if (errorResponse != null) { reject( @@ -3070,8 +3128,9 @@ export abstract class AbstractPool< complete() } } - try { - for (const workerId of responsesReceived.keys()) { + const workerIds = responsesReceived.keys() + for (const workerId of workerIds) { + try { const workerNodeKey = this.getWorkerNodeKeyByWorkerId(workerId) const workerNode = this.workerNodes[workerNodeKey] const cleanup = this.createWorkerMessageListenerCleanup( @@ -3096,9 +3155,26 @@ export abstract class AbstractPool< ...message, taskFunctionOperationId, }) + } catch (error) { + sendError = error + sendFailed = true + // Retain listeners for requests already sent, but retire unsent targets. + if ( + responsesReceived.has(workerId) && + responsesReceived.get(workerId) == null + ) { + --remainingResponses + } + responsesReceived.delete(workerId) + for (const unsentWorkerId of workerIds) { + if (responsesReceived.get(unsentWorkerId) == null) { + --remainingResponses + } + responsesReceived.delete(unsentWorkerId) + } + complete() + break } - } catch (error) { - reject(error) } try { return await promise @@ -3193,6 +3269,22 @@ export abstract class AbstractPool< } } + private startWorkerTaskFunctionBootstrap ( + workerNode: IWorkerNode, + ready = false + ): Promise { + const bootstrap: { promise?: Promise; ready: boolean } = { ready } + workerNode.info.ready = false + this.failedWorkerTaskFunctionBootstraps.delete(workerNode) + this.workerTaskFunctionBootstraps.set(workerNode, bootstrap) + const promise = this.replayTaskFunctionsAfterOperations( + workerNode, + bootstrap + ) + bootstrap.promise = promise + return promise + } + private readonly stealTask = ( sourceWorkerNode: IWorkerNode, destinationWorkerNodeKey: number diff --git a/tests/pools/task-function-recovery.test.mjs b/tests/pools/task-function-recovery.test.mjs new file mode 100644 index 000000000..533b8d86a --- /dev/null +++ b/tests/pools/task-function-recovery.test.mjs @@ -0,0 +1,559 @@ +import { once } from 'node:events' +import { describe, expect, it } from 'vitest' + +import { + DynamicClusterPool, + DynamicThreadPool, + FixedClusterPool, + FixedThreadPool, + PoolEvents, + WorkerChoiceStrategies, +} from '../../lib/index.mjs' +import { bounded, waitForMessage } from '../test-utils.cjs' + +const workerFile = './tests/worker-files/taskFunctionRecoveryWorker.mjs' + +const outcome = promise => + promise.then( + value => ({ value }), + error => ({ error }) + ) + +const identify = function (data) { + return { ...data, workerId: this.id } +} + +describe.each([ + ['thread', FixedThreadPool, DynamicThreadPool], + ['cluster', FixedClusterPool, DynamicClusterPool], +])('%s task function recovery', (workerType, FixedPool, DynamicPool) => { + const control = async (node, message) => { + const acknowledged = waitForMessage( + node.worker, + response => response.control === message.control + ) + message = { ...message, workerId: node.info.id } + if (workerType === 'thread') node.worker.postMessage(message) + else node.worker.send(message) + await bounded(acknowledged) + } + + it('uses captured affinity when an ineligible worker rejects replay', async () => { + const pool = new DynamicPool(0, 3, workerFile) + const errors = [] + pool.emitter.on(PoolEvents.error, error => errors.push(error)) + const send = pool.sendToWorker.bind(pool) + let injected = false + pool.sendToWorker = (key, message) => { + if ( + !injected && + key === 0 && + message.taskFunctionOperation === 'add' && + message.taskFunctionProperties.name === 'runtime' + ) { + injected = true + message = { + ...message, + taskFunctionProperties: { + ...message.taskFunctionProperties, + priority: 1000, + }, + } + } + send(key, message) + } + let execution + try { + await bounded( + pool.addTaskFunction('runtime', { + taskFunction: identify, + workerNodeKeys: [2], + }) + ) + const failed = once(pool.emitter, PoolEvents.error) + execution = outcome(pool.execute({ input: 42 }, 'runtime')) + const result = await bounded(execution) + await bounded(failed) + expect(result).toEqual({ + value: expect.objectContaining({ input: 42 }), + }) + expect(pool.workerNodes).toHaveLength(3) + expect(result.value.workerId).toBe(pool.workerNodes[2].info.id) + expect(pool.workerNodes[0].info.ready).toBe(false) + expect(errors).toEqual([expect.any(Error)]) + } finally { + await bounded(pool.destroy()) + await bounded(Promise.allSettled([execution])) + } + }) + + it('uses captured default affinity when only an unready worker has metadata', async () => { + const pool = new DynamicPool(1, 3, workerFile) + const errors = [] + pool.emitter.on(PoolEvents.error, error => errors.push(error)) + try { + await bounded(once(pool.emitter, PoolEvents.ready)) + await bounded( + pool.addTaskFunction('runtime', { + taskFunction: identify, + workerNodeKeys: [2], + }) + ) + await bounded(pool.setDefaultTaskFunction('runtime')) + await bounded(pool.destroy()) + const send = pool.sendToWorker.bind(pool) + let injected = false + pool.sendToWorker = (key, message) => { + if (!injected && key === 0 && message.taskFunctionOperation === 'add') { + injected = true + message = { + ...message, + taskFunctionProperties: { + ...message.taskFunctionProperties, + priority: 1000, + }, + } + } + send(key, message) + } + const failed = once(pool.emitter, PoolEvents.error) + pool.start() + await bounded(failed) + expect(pool.workerNodes[0].info.ready).toBe(false) + expect(await bounded(pool.execute({ input: 42 }))).toEqual({ + input: 42, + workerId: pool.workerNodes[2].info.id, + }) + expect(pool.workerNodes).toHaveLength(3) + expect(errors).toEqual([expect.any(Error)]) + } finally { + if (pool.info.started) await bounded(pool.destroy()) + } + }) + + it('corrects invalid captured source and restores every desired function', async () => { + const pool = new FixedPool(1, workerFile, { startWorkers: false }) + const errors = [] + pool.emitter.on(PoolEvents.error, error => errors.push(error)) + try { + const invalid = data => data + 1 + invalid.toString = () => 'null' + await bounded(pool.addTaskFunction('runtime', invalid)) + await bounded(pool.addTaskFunction('later', data => data + 2)) + const failed = once(pool.emitter, PoolEvents.error) + pool.start() + const node = pool.workerNodes[0] + await bounded(failed) + expect(node.info.ready).toBe(false) + expect( + await bounded(pool.addTaskFunction('runtime', data => data + 1)) + ).toBe(true) + expect(await bounded(pool.execute(42, 'runtime'))).toBe(43) + expect(await bounded(pool.execute(42, 'later'))).toBe(44) + expect(pool.workerNodes[0]).toBe(node) + expect(node.info.ready).toBe(true) + expect(errors).toEqual([expect.any(Error)]) + } finally { + await bounded(pool.destroy()) + } + }) + + it.each([ + ['acknowledgement', 2], + ['failure', 2], + ['exit', 2], + ['acknowledgement', 3], + ['failure', 3], + ['exit', 3], + ])( + 'keeps earlier sent mutations pending until their %s with %i workers', + async (completion, size) => { + const pool = new FixedPool(size, workerFile, { + workerChoiceStrategy: WorkerChoiceStrategies.ROUND_ROBIN, + }) + const errors = [] + pool.emitter.on(PoolEvents.error, error => errors.push(error)) + const originalError = new Error('Injected pre-dispatch send failure') + let removal + let addition + let heldNode + try { + await bounded(once(pool.emitter, PoolEvents.ready)) + await bounded(pool.addTaskFunction('runtime', () => 'old')) + if (size === 3) { + await bounded( + pool.addTaskFunction('inspect', { + taskFunction: function () { + return this.hasTaskFunction('runtime').status + }, + workerNodeKeys: [2], + }) + ) + } + heldNode = pool.workerNodes[0] + if (completion === 'failure') { + await bounded( + pool.sendTaskFunctionOperationToWorker(0, { + taskFunctionOperation: 'default', + taskFunctionProperties: { name: 'runtime' }, + }) + ) + } + await control(heldNode, { + control: 'hold', + name: 'runtime', + operation: 'remove', + }) + const held = waitForMessage( + heldNode.worker, + message => message.heldOperation === 'remove' + ) + const send = pool.sendToWorker.bind(pool) + let injected = false + pool.sendToWorker = (key, message) => { + if ( + !injected && + key === 1 && + message.taskFunctionOperation === 'remove' + ) { + injected = true + throw originalError + } + send(key, message) + } + let settled = false + removal = outcome(pool.removeTaskFunction('runtime')).then(result => { + settled = true + return result + }) + addition = outcome(pool.addTaskFunction('runtime', () => 'new')) + await bounded(held) + expect(await bounded(pool.execute({}, 'runtime'))).toBe('old') + expect(settled).toBe(false) + if (size === 3) { + expect(await bounded(pool.execute({}, 'inspect'))).toBe(true) + } + if (completion === 'exit') { + await bounded(heldNode.terminate()) + } else { + await control(heldNode, { control: 'release' }) + } + expect((await bounded(removal)).error).toBe(originalError) + expect(await bounded(addition)).toEqual({ value: true }) + for (let i = 0; i < size; ++i) { + expect(await bounded(pool.execute({}, 'runtime'))).toBe('new') + } + expect(errors).toEqual([]) + } finally { + if (heldNode != null && pool.workerNodes.includes(heldNode)) { + await control(heldNode, { control: 'release' }) + } + await bounded(pool.destroy()) + await bounded(Promise.allSettled([removal, addition])) + } + } + ) + + it('preserves an undefined send error when no target was contacted', async () => { + const pool = new FixedPool(1, workerFile) + try { + await bounded(once(pool.emitter, PoolEvents.ready)) + await bounded(pool.addTaskFunction('runtime', () => 'old')) + const send = pool.sendToWorker.bind(pool) + const originalError = undefined + let injected = false + pool.sendToWorker = (key, message) => { + if (!injected && message.taskFunctionOperation === 'remove') { + injected = true + throw originalError + } + send(key, message) + } + expect( + await bounded(outcome(pool.removeTaskFunction('runtime'))) + ).toEqual({ + error: originalError, + }) + expect(await bounded(pool.addTaskFunction('runtime', () => 'new'))).toBe( + true + ) + expect(await bounded(pool.execute({}, 'runtime'))).toBe('new') + } finally { + await bounded(pool.destroy()) + } + }) + + it('uses live worker defaults after a pool-wide default changes locally', async () => { + const pool = new FixedPool(2, workerFile, { + workerChoiceStrategy: WorkerChoiceStrategies.ROUND_ROBIN, + }) + try { + await bounded(once(pool.emitter, PoolEvents.ready)) + await bounded( + pool.addTaskFunction('first', { + taskFunction: identify, + workerNodeKeys: [0], + }) + ) + await bounded( + pool.addTaskFunction('second', { + taskFunction: identify, + workerNodeKeys: [1], + }) + ) + await bounded( + pool.addTaskFunction('select', function (name) { + return this.setDefaultTaskFunction(name).status + }) + ) + await bounded(pool.setDefaultTaskFunction('first')) + expect(await bounded(pool.execute({ before: true }))).toEqual({ + before: true, + workerId: pool.workerNodes[0].info.id, + }) + for (let i = 0; i < 2; ++i) { + expect(await bounded(pool.execute('second', 'select'))).toBe(true) + } + expect(await bounded(pool.execute({ after: true }))).toEqual({ + after: true, + workerId: pool.workerNodes[1].info.id, + }) + } finally { + await bounded(pool.destroy()) + } + }) + + it.each(['add', 'remove', 'default'])( + 'restores the full state before completing a corrective %s', + async operation => { + const pool = new FixedPool(1, workerFile) + const errors = [] + pool.emitter.on(PoolEvents.error, error => errors.push(error)) + let node + const pending = [] + try { + await bounded(once(pool.emitter, PoolEvents.ready)) + await bounded(pool.addTaskFunction('runtime', () => 'runtime')) + await bounded( + pool.addTaskFunction('later', { + priority: -10, + strategy: WorkerChoiceStrategies.FAIR_SHARE, + taskFunction: () => 'later', + }) + ) + await bounded(pool.setDefaultTaskFunction('later')) + await bounded(pool.destroy()) + const send = pool.sendToWorker.bind(pool) + let injected = false + pool.sendToWorker = (key, message) => { + if ( + !injected && + message.taskFunctionOperation === 'add' && + message.taskFunctionProperties.name === 'later' + ) { + injected = true + message = { + ...message, + taskFunctionProperties: { + ...message.taskFunctionProperties, + priority: 1000, + }, + } + } + send(key, message) + } + const failed = once(pool.emitter, PoolEvents.error) + pool.start() + node = pool.workerNodes[0] + await bounded(failed) + await control(node, { + control: 'hold', + name: 'later', + operation: 'add', + }) + const held = waitForMessage( + node.worker, + message => message.heldOperation === 'add' + ) + let corrected = false + const correction = outcome( + operation === 'add' + ? pool.addTaskFunction('runtime', () => 'updated') + : operation === 'remove' + ? pool.removeTaskFunction('runtime') + : pool.setDefaultTaskFunction('runtime') + ).then(result => { + corrected = true + return result + }) + let queuedDone = false + const queued = outcome( + pool.addTaskFunction('queued', () => 'queued') + ).then(result => { + queuedDone = true + return result + }) + pending.push(correction, queued) + await bounded(held) + expect(node.info.ready).toBe(false) + expect(corrected).toBe(false) + expect(queuedDone).toBe(false) + await control(node, { control: 'release' }) + expect(await bounded(correction)).toEqual({ value: true }) + expect(await bounded(queued)).toEqual({ value: true }) + expect(pool.workerNodes[0]).toBe(node) + expect(node.info.ready).toBe(true) + expect(await bounded(pool.execute({}, 'later'))).toBe('later') + expect(await bounded(pool.execute({}))).toBe( + operation === 'default' ? 'runtime' : 'later' + ) + expect(await bounded(pool.execute({}, 'queued'))).toBe('queued') + if (operation === 'remove') { + await expect(bounded(pool.execute({}, 'runtime'))).rejects.toThrow( + "Task function 'runtime' not found" + ) + } else { + expect(await bounded(pool.execute({}, 'runtime'))).toBe( + operation === 'add' ? 'updated' : 'runtime' + ) + } + expect(pool.listTaskFunctionsProperties()).toContainEqual({ + name: 'later', + priority: -10, + strategy: WorkerChoiceStrategies.FAIR_SHARE, + }) + expect(errors).toEqual([expect.any(Error)]) + } finally { + if (node != null && pool.workerNodes.includes(node)) { + await control(node, { control: 'release' }) + } + await bounded(pool.destroy()) + await bounded(Promise.allSettled(pending)) + } + } + ) + + it('keeps repeated replay failures recoverable without becoming ready early', async () => { + const pool = new FixedPool(1, workerFile, { startWorkers: false }) + const errors = [] + pool.emitter.on(PoolEvents.error, error => errors.push(error)) + try { + const invalid = () => 'invalid' + invalid.toString = () => 'null' + await bounded(pool.addTaskFunction('first', invalid)) + await bounded(pool.addTaskFunction('second', invalid)) + await bounded(pool.addTaskFunction('later', () => 'later')) + let failed = once(pool.emitter, PoolEvents.error) + pool.start() + const node = pool.workerNodes[0] + await bounded(failed) + failed = once(pool.emitter, PoolEvents.error) + expect(await bounded(pool.addTaskFunction('first', () => 'first'))).toBe( + true + ) + await bounded(failed) + expect(node.info.ready).toBe(false) + expect( + await bounded(pool.addTaskFunction('second', () => 'second')) + ).toBe(true) + expect(pool.workerNodes[0]).toBe(node) + expect(node.info.ready).toBe(true) + expect(await bounded(pool.execute({}, 'first'))).toBe('first') + expect(await bounded(pool.execute({}, 'second'))).toBe('second') + expect(await bounded(pool.execute({}, 'later'))).toBe('later') + expect(errors).toEqual([expect.any(Error), expect.any(Error)]) + } finally { + await bounded(pool.destroy()) + } + }) + + it.each(['exit', 'destroy'])( + 'settles corrective registration when %s interrupts recovery', + async transition => { + const pool = new FixedPool(1, workerFile, { startWorkers: false }) + const errors = [] + pool.emitter.on(PoolEvents.error, error => errors.push(error)) + let node + let correction + try { + const invalid = () => 'invalid' + invalid.toString = () => 'null' + await bounded(pool.addTaskFunction('runtime', invalid)) + await bounded(pool.addTaskFunction('later', () => 'later')) + const failed = once(pool.emitter, PoolEvents.error) + pool.start() + node = pool.workerNodes[0] + await bounded(failed) + await control(node, { + control: 'hold', + name: 'later', + operation: 'add', + }) + const held = waitForMessage( + node.worker, + message => message.heldOperation === 'add' + ) + correction = outcome(pool.addTaskFunction('runtime', () => 'corrected')) + await bounded(held) + expect(node.info.ready).toBe(false) + if (transition === 'exit') await bounded(node.terminate()) + else await bounded(pool.destroy()) + expect(await bounded(correction)).toEqual({ value: true }) + if (transition === 'exit') { + expect(pool.workerNodes[0]).not.toBe(node) + expect(await bounded(pool.execute({}, 'runtime'))).toBe('corrected') + expect(await bounded(pool.execute({}, 'later'))).toBe('later') + } else { + expect(pool.info.started).toBe(false) + expect(pool.workerNodes).toEqual([]) + } + expect(errors).toEqual([expect.any(Error)]) + } finally { + if (node != null && pool.workerNodes.includes(node)) { + await control(node, { control: 'release' }) + } + if (pool.info.started) await bounded(pool.destroy()) + await bounded(Promise.allSettled([correction])) + } + } + ) + + it('keeps a failed bootstrap unavailable when native readiness arrives later', async () => { + const pool = new FixedPool(1, workerFile, { startWorkers: false }) + const errors = [] + pool.emitter.on(PoolEvents.error, error => errors.push(error)) + try { + await bounded(pool.addTaskFunction('runtime', () => 'runtime')) + const send = pool.sendToWorker.bind(pool) + let injected = false + pool.sendToWorker = (key, message) => { + if (!injected && message.taskFunctionOperation === 'add') { + injected = true + throw new Error( + 'Injected replay send failure before native readiness' + ) + } + send(key, message) + } + const failed = once(pool.emitter, PoolEvents.error) + pool.start() + const node = pool.workerNodes[0] + const nativeReady = waitForMessage( + node.messageChannel?.port1 ?? node.worker, + message => message.ready === true + ) + await bounded(failed) + await bounded(nativeReady) + expect(node.info.ready).toBe(false) + expect( + await bounded(pool.addTaskFunction('runtime', () => 'corrected')) + ).toBe(true) + expect(pool.workerNodes[0]).toBe(node) + expect(node.info.ready).toBe(true) + expect(await bounded(pool.execute({}, 'runtime'))).toBe('corrected') + expect(errors).toEqual([expect.any(Error)]) + } finally { + await bounded(pool.destroy()) + } + }) +}) diff --git a/tests/worker-files/taskFunctionRecoveryWorker.mjs b/tests/worker-files/taskFunctionRecoveryWorker.mjs new file mode 100644 index 000000000..469e28580 --- /dev/null +++ b/tests/worker-files/taskFunctionRecoveryWorker.mjs @@ -0,0 +1,49 @@ +import { isMainThread, parentPort } from 'node:worker_threads' + +import { ClusterWorker, ThreadWorker } from '../../lib/index.mjs' + +const Worker = isMainThread ? ClusterWorker : ThreadWorker +const notify = message => { + if (parentPort != null) parentPort.postMessage(message) + else process.send(message) +} + +class TaskFunctionRecoveryWorker extends Worker { + constructor (...args) { + super(...args) + const controlPort = parentPort ?? process + controlPort.on('message', message => { + if (message.control === 'hold') { + this.heldOperation = { + name: message.name, + operation: message.operation, + } + } else if (message.control === 'release') { + const pending = this.pendingOperation + this.pendingOperation = undefined + if (pending != null) super.handleTaskFunctionOperationMessage(pending) + } else { + return + } + notify({ control: message.control, workerId: this.id }) + }) + } + + handleTaskFunctionOperationMessage (message) { + if ( + this.heldOperation?.operation === message.taskFunctionOperation && + this.heldOperation.name === message.taskFunctionProperties?.name + ) { + this.heldOperation = undefined + this.pendingOperation = message + notify({ + heldOperation: message.taskFunctionOperation, + workerId: this.id, + }) + return + } + super.handleTaskFunctionOperationMessage(message) + } +} + +export default new TaskFunctionRecoveryWorker({ echo: data => data }) -- 2.53.0