From a8768f49edbbb6c3c3c2418b755bd3ad86633181 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C3=A9r=C3=B4me=20Benoit?= Date: Tue, 20 Oct 2026 20:10:16 +0200 Subject: [PATCH] fix(pool): recover runtime task function operations Port poolifier/poolifier#3369 at 23357872, retaining synchronous Web lifecycle safeguards. --- docs/api.md | 28 +- src/pools/abstract-pool.ts | 210 ++++-- tests/pools/task-function-recovery.test.mjs | 676 ++++++++++++++++++ .../thread/taskFunctionRecoveryWorker.mjs | 42 ++ 4 files changed, 895 insertions(+), 61 deletions(-) create mode 100644 tests/pools/task-function-recovery.test.mjs create mode 100644 tests/worker-files/thread/taskFunctionRecoveryWorker.mjs diff --git a/docs/api.md b/docs/api.md index 45150678..53205c80 100644 --- a/docs/api.md +++ b/docs/api.md @@ -185,16 +185,17 @@ Functions registered through this method are installed on newly created workers, including replacements for fixed and minimum workers. 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. - -If workers are present when a change starts, each original target must respond -or exit before the call settles. At least one successful result is required; any -reported failure rejects the change. An unanswered target's exit does not reject -a change with a successful peer, including synchronous exits from protected -dispatch hooks. All targets exiting unanswered rejects the change. Deno and Bun -preserve buffered operation results before an observed unexpected exit; browsers -preserve only already-delivered results. +`setDefaultTaskFunction()` run in call order. A rejected call does not prevent +later calls or undo changes already applied to workers. + +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. Without a send failure, an unanswered target's exit does not +reject a change with a successful peer, including synchronous exits from +protected dispatch hooks. All targets exiting unanswered rejects the change. +Deno and Bun preserve buffered operation results before an observed unexpected +exit; browsers preserve only already-delivered results. New and replacement workers wait for ongoing task function changes. They accept tasks only after restoring successfully registered task functions, the selected @@ -202,11 +203,16 @@ default and scheduling options. Workers being terminated remain unavailable. Replay remains bound to its original worker when protected listener-registration hooks synchronously retire or reindex worker nodes. +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 original targets to respond or exit. Status-only +changes still wait for all contacted workers to respond or exit. Status-only overrides that discard this correlation must be updated. Deno and Bun preserve incoming asynchronous context; browsers preserve only synchronous context. Deferred browser handlers must retain the UUID or original properties object. diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts index 555035bd..b578de41 100644 --- a/src/pools/abstract-pool.ts +++ b/src/pools/abstract-pool.ts @@ -151,6 +151,15 @@ export abstract class AbstractPool< * The last explicit default acknowledged by an operation's worker cohort. */ private defaultTaskFunctionName?: string + + /** + * Failed bootstraps retain native readiness outside active waits. + */ + private readonly failedWorkerTaskFunctionBootstraps = new Map< + IWorkerNode, + boolean + >() + private pendingTaskFunctionOperations = 0 private resolveTaskFunctionOperations?: (value: undefined) => void private taskFunctionOperationsFinished?: Promise @@ -846,7 +855,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, @@ -945,6 +958,10 @@ export abstract class AbstractPool< let remainingResponses = responsesReceived.size let errorResponse: MessageValue | undefined let protocolError: Error | undefined + let sendError: unknown + let sendFailed = false + // Web exit hooks can settle targets synchronously before a send throws. + let dispatching = true const { promise, reject, resolve } = Promise.withResolvers() const listeners: { cleanup: () => void @@ -952,8 +969,10 @@ export abstract class AbstractPool< workerNode: IWorkerNode }[] = [] const complete = (): void => { - if (remainingResponses !== 0) return - if (protocolError != null) { + if (dispatching || remainingResponses !== 0) return + if (sendFailed) { + reject(sendError) + } else if (protocolError != null) { reject(protocolError) } else if (errorResponse != null) { reject( @@ -1020,29 +1039,59 @@ export abstract class AbstractPool< listeners.push({ cleanup, exitListener, workerNode }) this.registerTaskFunctionOperationExitListener(workerNode, exitListener) } - for (const { workerNode } of listeners) { - let workerNodeKey = this.workerNodes.indexOf(workerNode) - if ( - workerNodeKey === -1 || - !responsesReceived.has(workerNode.info.id) - ) { - continue - } - this.registerWorkerMessageListener(workerNodeKey, listener) - workerNodeKey = this.workerNodes.indexOf(workerNode) - if ( - workerNodeKey === -1 || - !responsesReceived.has(workerNode.info.id) - ) { - continue + const workerListeners = listeners.values() + for (const { workerNode } of workerListeners) { + const workerId = workerNode.info.id + try { + let workerNodeKey = this.workerNodes.indexOf(workerNode) + if ( + workerNodeKey === -1 || + !responsesReceived.has(workerNode.info.id) + ) { + continue + } + this.registerWorkerMessageListener(workerNodeKey, listener) + workerNodeKey = this.workerNodes.indexOf(workerNode) + if ( + workerNodeKey === -1 || + !responsesReceived.has(workerNode.info.id) + ) { + continue + } + this.sendToWorker(workerNodeKey, { + ...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 { workerNode: unsentWorkerNode } of workerListeners) { + const unsentWorkerId = unsentWorkerNode.info.id + if ( + responsesReceived.has(unsentWorkerId) && + responsesReceived.get(unsentWorkerId) == null + ) { + --remainingResponses + } + responsesReceived.delete(unsentWorkerId) + } + complete() + break } - this.sendToWorker(workerNodeKey, { - ...message, - taskFunctionOperationId, - }) } } catch (error) { reject(error) + } finally { + dispatching = false + complete() } try { return await promise @@ -1162,15 +1211,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 ?? [] + ) } private readonly getAbortError = ( @@ -1249,6 +1293,20 @@ export abstract class AbstractPool< )?.strategy } + 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 node keys affinity array, if any. * @param name - The task function name. @@ -1257,20 +1315,29 @@ 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 for (let i = 0; i < taskFunctionsProperties.length; i++) { const taskFunctionProperties = taskFunctionsProperties[i] if (taskFunctionProperties.name === name) { return taskFunctionProperties.workerNodeKeys } } - return undefined + return this.taskFunctions.get(name)?.taskFunctionProperties.workerNodeKeys } /** @@ -2068,20 +2135,13 @@ 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.eventTarget?.dispatchEvent( + new ErrorEvent(PoolEvents.error, { error }), + ) + }, ) - bootstrap.promise.catch((error: unknown) => { - this.eventTarget?.dispatchEvent( - new ErrorEvent(PoolEvents.error, { error }), - ) - }) } this.afterWorkerNodeSetup(workerNodeKey) return workerNodeKey @@ -2715,8 +2775,14 @@ 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) @@ -2984,6 +3050,7 @@ export abstract class AbstractPool< */ protected 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) @@ -3220,7 +3287,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) } @@ -3235,6 +3310,41 @@ export abstract class AbstractPool< return this.workerNodes.indexOf(workerNode) } + 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.eventTarget?.dispatchEvent( + new ErrorEvent(PoolEvents.error, { error }), + ) + }) + promises.push(promise) + } + } + await Promise.allSettled(promises) + } + + 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 async waitForWorkerTaskFunctionBootstraps( abortSignal?: AbortSignal, name?: string, diff --git a/tests/pools/task-function-recovery.test.mjs b/tests/pools/task-function-recovery.test.mjs new file mode 100644 index 00000000..451d408b --- /dev/null +++ b/tests/pools/task-function-recovery.test.mjs @@ -0,0 +1,676 @@ +import { expect } from '@std/expect' +import { describe, it } from '@std/testing/bdd' +import { + DynamicThreadPool, + FixedThreadPool, + PoolEvents, + WorkerChoiceStrategies, +} from '../../src/mod.ts' +import { each, once } from '../contract-test-utils.mjs' +import { bounded, waitForMessage } from '../test-utils.mjs' + +const workerFile = new URL( + '../worker-files/thread/taskFunctionRecoveryWorker.mjs', + import.meta.url, +) + +const outcome = (promise) => + promise.then( + (value) => ({ value }), + (error) => ({ error }), + ) + +const identify = function (data) { + return { ...data, workerId: this.id } +} + +describe({ + name: 'Task function recovery', + sanitizeOps: false, + sanitizeResources: false, + fn: () => { + const control = async (node, message) => { + const acknowledged = waitForMessage( + node.worker, + (response) => response.control === message.control, + ) + message = { ...message, workerId: node.info.id } + node.worker.postMessage(message) + await bounded(acknowledged) + } + + it('uses captured affinity when an ineligible worker rejects replay', async () => { + const pool = new DynamicThreadPool(0, 3, workerFile) + const errors = [] + pool.eventTarget.addEventListener( + 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.eventTarget, 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 DynamicThreadPool(1, 3, workerFile) + const errors = [] + pool.eventTarget.addEventListener( + PoolEvents.error, + ({ error }) => errors.push(error), + ) + try { + await bounded(once(pool.eventTarget, 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.eventTarget, 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 FixedThreadPool(1, workerFile, { startWorkers: false }) + const errors = [] + pool.eventTarget.addEventListener( + 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.eventTarget, 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()) + } + }) + + 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 FixedThreadPool(size, workerFile, { + workerChoiceStrategy: WorkerChoiceStrategies.ROUND_ROBIN, + }) + const errors = [] + pool.eventTarget.addEventListener( + 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.eventTarget, 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 FixedThreadPool(1, workerFile) + try { + await bounded(once(pool.eventTarget, 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 FixedThreadPool(2, workerFile, { + workerChoiceStrategy: WorkerChoiceStrategies.ROUND_ROBIN, + }) + try { + await bounded(once(pool.eventTarget, 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()) + } + }) + + each(['add', 'remove', 'default'])( + 'restores the full state before completing a corrective %s', + async (operation) => { + const pool = new FixedThreadPool(1, workerFile) + const errors = [] + pool.eventTarget.addEventListener( + PoolEvents.error, + ({ error }) => errors.push(error), + ) + let node + const pending = [] + try { + await bounded(once(pool.eventTarget, 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.eventTarget, 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 FixedThreadPool(1, workerFile, { startWorkers: false }) + const errors = [] + pool.eventTarget.addEventListener( + 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.eventTarget, PoolEvents.error) + pool.start() + const node = pool.workerNodes[0] + await bounded(failed) + failed = once(pool.eventTarget, 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()) + } + }) + + each(['exit', 'destroy'])( + 'settles corrective registration when %s interrupts recovery', + async (transition) => { + const pool = new FixedThreadPool(1, workerFile, { startWorkers: false }) + const errors = [] + pool.eventTarget.addEventListener( + 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.eventTarget, 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 FixedThreadPool(1, workerFile, { startWorkers: false }) + const errors = [] + pool.eventTarget.addEventListener( + 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.eventTarget, 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()) + } + }) + + it('preserves the send error when its hook synchronously retires every target', async () => { + const pool = new FixedThreadPool(2, workerFile) + const originalError = new Error( + 'Send failed after synchronous cohort retirement', + ) + try { + await bounded(once(pool.eventTarget, PoolEvents.ready)) + await bounded(pool.addTaskFunction('runtime', () => 'old')) + const send = pool.sendToWorker.bind(pool) + let injected = false + pool.sendToWorker = (key, message) => { + if (!injected && message.taskFunctionOperation === 'remove') { + injected = true + for (const node of [...pool.workerNodes]) node.terminate() + 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('waits for contacted targets when a throwing hook synchronously retires an unsent peer', async () => { + const pool = new FixedThreadPool(3, workerFile) + const originalError = new Error( + 'Send failed after unsent peer retirement', + ) + let heldNode + let removal + let addition + try { + await bounded(once(pool.eventTarget, PoolEvents.ready)) + await bounded(pool.addTaskFunction('runtime', () => 'old')) + heldNode = pool.workerNodes[0] + 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 + pool.workerNodes[2].terminate() + 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) + await control(heldNode, { control: 'release' }) + expect((await bounded(removal)).error).toBe(originalError) + expect(await bounded(addition)).toEqual({ value: true }) + expect(await bounded(pool.execute({}, 'runtime'))).toBe('new') + } finally { + if (heldNode != null && pool.workerNodes.includes(heldNode)) { + await control(heldNode, { control: 'release' }) + } + await bounded(pool.destroy()) + await bounded(Promise.allSettled([removal, addition])) + } + }) + }, +}) diff --git a/tests/worker-files/thread/taskFunctionRecoveryWorker.mjs b/tests/worker-files/thread/taskFunctionRecoveryWorker.mjs new file mode 100644 index 00000000..23ec83eb --- /dev/null +++ b/tests/worker-files/thread/taskFunctionRecoveryWorker.mjs @@ -0,0 +1,42 @@ +import { ThreadWorker } from '../../../src/mod.ts' + +const notify = (message) => self.postMessage(message) + +class TaskFunctionRecoveryWorker extends ThreadWorker { + constructor(...args) { + super(...args) + self.addEventListener('message', ({ data: 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 })