From e70fe6b6bb32709fb30ddf1530d4f4da082dee6e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C3=A9r=C3=B4me=20Benoit?= Date: Tue, 20 Oct 2026 22:45:54 +0200 Subject: [PATCH] fix(pool): remove failed bootstrap registrations Port poolifier/poolifier#3371 at 8adc6887, retaining Web lifecycle and acknowledgement contracts. --- docs/api.md | 29 +- src/pools/abstract-pool.ts | 15 +- src/utility-types.ts | 4 + src/worker/abstract-worker.ts | 29 +- tests/pools/task-function-recovery.test.mjs | 257 +++++++++++++++++- .../thread/taskContractWorker.mjs | 8 +- .../thread/taskFunctionRecoveryWorker.mjs | 16 ++ 7 files changed, 335 insertions(+), 23 deletions(-) diff --git a/docs/api.md b/docs/api.md index 53205c80..569bbdd8 100644 --- a/docs/api.md +++ b/docs/api.md @@ -190,12 +190,12 @@ 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. +otherwise, at least one accepted response is required and any other 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 @@ -210,12 +210,13 @@ 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. Deno and Bun preserve -incoming asynchronous context; browsers preserve only synchronous context. -Deferred browser handlers must retain the UUID or original properties object. +original `taskFunctionProperties` object when using the standard transport. +Preserve result metadata when forwarding acknowledgements. 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. Deno and Bun preserve incoming asynchronous +context; browsers preserve only synchronous context. Deferred browser handlers +must retain the UUID or original properties object. ### `pool.removeTaskFunction(name)` @@ -224,6 +225,10 @@ Deferred browser handlers must retain the UUID or original properties object. This method is available on both pool implementations and returns a boolean promise. +For workers whose initialization failed, confirmed absence of the requested +function also counts as successful removal. Other removal errors still reject +the call. + ### `pool.listTaskFunctionsProperties()` This method is available on both pool implementations and returns an array of diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts index b578de41..c8da9523 100644 --- a/src/pools/abstract-pool.ts +++ b/src/pools/abstract-pool.ts @@ -947,6 +947,7 @@ export abstract class AbstractPool< if (this.workerNodes.length === 0) { return true } + const { taskFunctionOperation } = message const taskFunctionOperationId = crypto.randomUUID() const responsesReceived = new Map< WorkerInfo['id'] | undefined, @@ -1011,7 +1012,19 @@ export abstract class AbstractPool< protocolError ??= this.getTaskFunctionOperationAcknowledgementError( message.workerId, ) - } else if (!message.taskFunctionOperationStatus) { + } else if ( + !message.taskFunctionOperationStatus && + !( + taskFunctionOperation === 'remove' && + message.taskFunctionOperationNotFound === true && + message.workerError == null && + this.failedWorkerTaskFunctionBootstraps.has( + this.workerNodes[ + this.getWorkerNodeKeyByWorkerId(message.workerId) + ], + ) + ) + ) { errorResponse ??= message } complete() diff --git a/src/utility-types.ts b/src/utility-types.ts index 58610e7a..f5062066 100644 --- a/src/utility-types.ts +++ b/src/utility-types.ts @@ -188,6 +188,10 @@ export interface MessageValue * Task function operation request UUID. */ readonly taskFunctionOperationId?: UUID + /** + * Whether removal confirmed that the requested task function was absent. + */ + readonly taskFunctionOperationNotFound?: boolean /** * Whether the task function operation is successful or not. */ diff --git a/src/worker/abstract-worker.ts b/src/worker/abstract-worker.ts index 8306151f..ff7de7f4 100644 --- a/src/worker/abstract-worker.ts +++ b/src/worker/abstract-worker.ts @@ -131,6 +131,12 @@ export abstract class AbstractWorker< */ protected activeInterval?: ReturnType + /** + * The base removal result and name that confirm absence for the current operation. + */ + private missingTaskFunctionRemoval?: TaskFunctionOperationResult + private missingTaskFunctionRemovalName?: string + /** * Correlation of task function operations across protected handler overrides. */ @@ -308,7 +314,12 @@ export abstract class AbstractWorker< } const deleteStatus = this.taskFunctions.delete(name) this.sendTaskFunctionsPropertiesToMainWorker() - return { status: deleteStatus } + const response = { status: deleteStatus } + if (!deleteStatus) { + this.missingTaskFunctionRemoval = response + this.missingTaskFunctionRemovalName = name + } + return response } catch (error) { return { status: false, error: error as Error } } @@ -466,6 +477,7 @@ export abstract class AbstractWorker< 'Cannot handle task function operation message without task function properties', ) } + let taskFunctionOperationNotFound = false let response: TaskFunctionOperationResult switch (taskFunctionOperation) { case 'add': @@ -493,7 +505,19 @@ export abstract class AbstractWorker< response = this.setDefaultTaskFunction(taskFunctionProperties.name) break case 'remove': - response = this.removeTaskFunction(taskFunctionProperties.name) + this.missingTaskFunctionRemoval = undefined + this.missingTaskFunctionRemovalName = undefined + try { + const name = taskFunctionProperties.name + response = this.removeTaskFunction(name) + taskFunctionOperationNotFound = !response.status && + response.error == null && + Object.is(response, this.missingTaskFunctionRemoval) && + Object.is(name, this.missingTaskFunctionRemovalName) + } finally { + this.missingTaskFunctionRemoval = undefined + this.missingTaskFunctionRemovalName = undefined + } break default: response = { @@ -510,6 +534,7 @@ export abstract class AbstractWorker< taskFunctionOperationId: this.getTaskFunctionOperationId(message), taskFunctionOperationStatus: status, taskFunctionProperties, + ...(taskFunctionOperationNotFound && { taskFunctionOperationNotFound }), ...(!status && error != null && { workerError: { diff --git a/tests/pools/task-function-recovery.test.mjs b/tests/pools/task-function-recovery.test.mjs index 451d408b..3747cec1 100644 --- a/tests/pools/task-function-recovery.test.mjs +++ b/tests/pools/task-function-recovery.test.mjs @@ -2,11 +2,10 @@ 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 { each, FixedThreadPool, once } from '../contract-test-utils.mjs' import { bounded, waitForMessage } from '../test-utils.mjs' const workerFile = new URL( @@ -170,6 +169,260 @@ describe({ } }) + each([false, true])( + 'discards failed registrations with reconstructed requests: %s', + async (legacy) => { + const pool = new FixedThreadPool( + 1, + legacy + ? new URL( + '../worker-files/thread/taskContractWorker.mjs', + import.meta.url, + ) + : workerFile, + { + startWorkers: false, + ...(legacy && { + workerOptions: { + workerData: { legacyTaskFunctionMessages: true }, + }, + }), + }, + ) + const errors = [] + pool.eventTarget.addEventListener( + PoolEvents.error, + ({ error }) => errors.push(error), + ) + try { + const invalid = () => 'invalid' + invalid.toString = () => 'null' + await bounded(pool.addTaskFunction('bad', 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(await bounded(pool.removeTaskFunction('bad'))).toBe(true) + expect(pool.workerNodes[0]).toBe(node) + expect(node.info.ready).toBe(true) + expect(await bounded(pool.execute(40, 'later'))).toBe(42) + await expect(bounded(pool.execute({}, 'bad'))).rejects.toThrow(Error) + await bounded(pool.destroy()) + pool.start() + expect(await bounded(pool.execute(40, 'later'))).toBe(42) + expect(pool.hasTaskFunction('bad')).toBe(false) + expect(errors).toEqual([expect.any(Error)]) + } finally { + if (pool.info.started) await bounded(pool.destroy()) + } + }, + ) + + it('removes a registration from mixed ready and failed workers', async () => { + const pool = new FixedThreadPool(2, workerFile, { + startWorkers: false, + workerChoiceStrategy: WorkerChoiceStrategies.ROUND_ROBIN, + }) + const errors = [] + pool.eventTarget.addEventListener( + PoolEvents.error, + ({ error }) => errors.push(error), + ) + try { + await bounded(pool.addTaskFunction('bad', (data) => data)) + await bounded(pool.addTaskFunction('later', identify)) + const send = pool.sendToWorker.bind(pool) + let injected = false + pool.sendToWorker = (key, message) => { + if ( + !injected && + key === 0 && + message.taskFunctionOperation === 'add' && + message.taskFunctionProperties.name === 'bad' + ) { + injected = true + message = { + ...message, + taskFunctionProperties: { + ...message.taskFunctionProperties, + priority: 1000, + }, + } + } + send(key, message) + } + const failed = once(pool.eventTarget, PoolEvents.error) + pool.start() + const nodes = [...pool.workerNodes] + await bounded(failed) + expect(await bounded(pool.removeTaskFunction('bad'))).toBe(true) + const results = [] + for (let i = 0; i < 2; ++i) { + results.push(await bounded(pool.execute({ restored: true }, 'later'))) + } + expect(new Set(results.map((result) => result.workerId))).toEqual( + new Set(nodes.map((node) => node.info.id)), + ) + expect(results.every((result) => result.restored)).toBe(true) + expect(pool.workerNodes).toEqual(nodes) + await expect(bounded(pool.execute({}, 'bad'))).rejects.toThrow(Error) + expect(errors).toEqual([expect.any(Error)]) + } finally { + await bounded(pool.destroy()) + } + }) + + it('discards skipped registrations without losing repeated recovery failures', 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('bad', invalid)) + await bounded(pool.addTaskFunction('discard', () => 'discard')) + 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.removeTaskFunction('discard'))).toBe(true) + await bounded(failed) + expect(node.info.ready).toBe(false) + expect(await bounded(pool.removeTaskFunction('bad'))).toBe(true) + expect(pool.workerNodes[0]).toBe(node) + expect(await bounded(pool.execute({}, 'later'))).toBe('later') + await expect(bounded(pool.execute({}, 'discard'))).rejects.toThrow( + Error, + ) + expect(errors).toEqual([expect.any(Error), expect.any(Error)]) + } finally { + await bounded(pool.destroy()) + } + }) + + it('preserves selected-default removal errors on a failed worker', 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('bad', invalid)) + await bounded(pool.addTaskFunction('echo', (data) => data)) + const failed = once(pool.eventTarget, PoolEvents.error) + pool.start() + const node = pool.workerNodes[0] + await bounded(failed) + await expect(bounded(pool.removeTaskFunction('echo'))).rejects.toThrow( + Error, + ) + expect(node.info.ready).toBe(false) + expect(await bounded(pool.removeTaskFunction('bad'))).toBe(true) + expect(pool.workerNodes[0]).toBe(node) + expect(await bounded(pool.execute(42))).toBe(42) + expect(errors).toEqual([expect.any(Error)]) + } finally { + await bounded(pool.destroy()) + } + }) + + it('preserves custom errorless removal refusals after failed replay', 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('bad', invalid)) + const failed = once(pool.eventTarget, PoolEvents.error) + pool.start() + const node = pool.workerNodes[0] + await bounded(failed) + await control(node, { control: 'refuseRemoval', name: 'bad' }) + await expect(bounded(pool.removeTaskFunction('bad'))).rejects.toThrow( + Error, + ) + expect(node.info.ready).toBe(false) + await control(node, { control: 'refuseRemoval' }) + expect(await bounded(pool.removeTaskFunction('bad'))).toBe(true) + expect(pool.workerNodes[0]).toBe(node) + expect(await bounded(pool.execute(42))).toBe(42) + expect(errors).toEqual([expect.any(Error)]) + } finally { + await bounded(pool.destroy()) + } + }) + + it('rejects absence results for a different task function', 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('kept', () => 42)) + await bounded(pool.addTaskFunction('broken', invalid)) + const failed = once(pool.eventTarget, PoolEvents.error) + pool.start() + const node = pool.workerNodes[0] + await bounded(failed) + await control(node, { + control: 'refuseRemoval', + name: 'kept', + prerequisite: 'missingPrerequisite', + }) + await expect(bounded(pool.removeTaskFunction('kept'))).rejects.toThrow( + Error, + ) + expect(node.info.ready).toBe(false) + expect(await bounded(pool.removeTaskFunction('broken'))).toBe(true) + expect(pool.workerNodes[0]).toBe(node) + expect(await bounded(pool.execute({}, 'kept'))).toBe(42) + await bounded(pool.destroy()) + pool.start() + expect(await bounded(pool.execute({}, 'kept'))).toBe(42) + expect(errors).toEqual([expect.any(Error)]) + } finally { + if (pool.info.started) await bounded(pool.destroy()) + } + }) + + it('preserves missing-registration failures on ready workers', async () => { + const pool = new FixedThreadPool(1, workerFile) + try { + await bounded(once(pool.eventTarget, PoolEvents.ready)) + await bounded(pool.addTaskFunction('bad', (data) => data)) + await bounded( + pool.addTaskFunction('removeLocal', function () { + return this.removeTaskFunction('bad').status + }), + ) + expect(await bounded(pool.execute({}, 'removeLocal'))).toBe(true) + await expect(bounded(pool.removeTaskFunction('bad'))).rejects.toThrow( + Error, + ) + expect(await bounded(pool.addTaskFunction('bad', () => 43))).toBe(true) + expect(await bounded(pool.execute({}, 'bad'))).toBe(43) + } finally { + await bounded(pool.destroy()) + } + }) + each([ ['acknowledgement', 2], ['failure', 2], diff --git a/tests/worker-files/thread/taskContractWorker.mjs b/tests/worker-files/thread/taskContractWorker.mjs index 89d37157..f76b2f5f 100644 --- a/tests/worker-files/thread/taskContractWorker.mjs +++ b/tests/worker-files/thread/taskContractWorker.mjs @@ -38,12 +38,8 @@ class TaskLifecycleWorker extends ThreadWorker { message.taskFunctionOperationStatus != null && workerData?.legacyTaskFunctionMessages === true ) { - message = { - taskFunctionOperation: message.taskFunctionOperation, - taskFunctionOperationStatus: message.taskFunctionOperationStatus, - taskFunctionProperties: message.taskFunctionProperties, - workerError: message.workerError, - } + message = { ...message } + delete message.taskFunctionOperationId } if ( !statusOnlySent && diff --git a/tests/worker-files/thread/taskFunctionRecoveryWorker.mjs b/tests/worker-files/thread/taskFunctionRecoveryWorker.mjs index 23ec83eb..abbd586e 100644 --- a/tests/worker-files/thread/taskFunctionRecoveryWorker.mjs +++ b/tests/worker-files/thread/taskFunctionRecoveryWorker.mjs @@ -15,6 +15,13 @@ class TaskFunctionRecoveryWorker extends ThreadWorker { const pending = this.pendingOperation this.pendingOperation = undefined if (pending != null) super.handleTaskFunctionOperationMessage(pending) + } else if (message.control === 'refuseRemoval') { + this.refusedRemoval = message.name + this.refusedRemovalPrerequisite = message.prerequisite + this.refusedRemovalResult = + message.name == null || message.prerequisite != null + ? undefined + : super.removeTaskFunction(message.name) } else { return } @@ -37,6 +44,15 @@ class TaskFunctionRecoveryWorker extends ThreadWorker { } super.handleTaskFunctionOperationMessage(message) } + + removeTaskFunction(name) { + if (this.refusedRemoval != null && name === this.refusedRemoval) { + return this.refusedRemovalPrerequisite == null + ? this.refusedRemovalResult + : super.removeTaskFunction(this.refusedRemovalPrerequisite) + } + return super.removeTaskFunction(name) + } } export default new TaskFunctionRecoveryWorker({ echo: (data) => data })