From: Jérôme Benoit Date: Tue, 20 Oct 2026 20:43:29 +0000 (+0200) Subject: fix: remove failed bootstrap registrations (#3371) X-Git-Url: https://git.piment-noir.org/?a=commitdiff_plain;h=1ce0d759b6b6726fff1432ddd171fda414727f9a;p=poolifier.git fix: remove failed bootstrap registrations (#3371) * fix: remove failed bootstrap registrations Accept only worker-confirmed absence after failed replay so captured registrations can be discarded without hiding genuine removal failures. Refs #3370 * fix: bind removal absence to requested name Keep base-result provenance tied to the original request name so unrelated prerequisite failures remain refusals. --- diff --git a/docs/api.md b/docs/api.md index d5fed6045..deddf72e5 100644 --- a/docs/api.md +++ b/docs/api.md @@ -116,6 +116,8 @@ The function's source and options, including `workerNodeKeys`, are captured at c This method is available on both pool implementations and returns a boolean promise. See [Runtime task function changes](#runtime-task-function-changes). +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 the task function properties. @@ -132,13 +134,13 @@ With no workers, this method resolves to `true` without changing the default sel 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. -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. +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 accepted response is required and any other 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. 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. +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. 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. ### Pool options diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts index 43de06fc8..9064825ab 100644 --- a/src/pools/abstract-pool.ts +++ b/src/pools/abstract-pool.ts @@ -3065,6 +3065,7 @@ export abstract class AbstractPool< if (this.workerNodes.length === 0) { return true } + const { taskFunctionOperation } = message const taskFunctionOperationId = randomUUID() const responsesReceived = new Map< number | undefined, @@ -3122,7 +3123,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 e1bb84084..870246399 100644 --- a/src/utility-types.ts +++ b/src/utility-types.ts @@ -49,6 +49,10 @@ export interface MessageValue * Task function operation 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 6207a7b9f..1ad9cceb0 100644 --- a/src/worker/abstract-worker.ts +++ b/src/worker/abstract-worker.ts @@ -100,6 +100,12 @@ export abstract class AbstractWorker< */ protected taskFunctions!: Map> + /** + * The base removal result and name that confirm absence for the current operation. + */ + private missingTaskFunctionRemoval?: TaskFunctionOperationResult + private missingTaskFunctionRemovalName?: string + private readonly taskFunctionOperationIds = new WeakMap< TaskFunctionProperties, UUID @@ -238,7 +244,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 { error: error as Error, status: false } } @@ -370,6 +381,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': @@ -398,7 +410,20 @@ 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 = { @@ -416,6 +441,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 533b8d86a..c6fc62f89 100644 --- a/tests/pools/task-function-recovery.test.mjs +++ b/tests/pools/task-function-recovery.test.mjs @@ -158,6 +158,240 @@ describe.each([ } }) + it.each([false, true])( + 'discards failed registrations with reconstructed requests: %s', + async legacy => { + const pool = new FixedPool( + 1, + legacy + ? workerType === 'thread' + ? './tests/worker-files/thread/taskLifecycleWorker.mjs' + : './tests/worker-files/cluster/legacyTaskFunctionWorker.mjs' + : workerFile, + { + startWorkers: false, + ...(legacy && + workerType === 'thread' && { + workerOptions: { + workerData: { legacyTaskFunctionMessages: true }, + }, + }), + } + ) + const errors = [] + pool.emitter.on(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.emitter, 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 FixedPool(2, workerFile, { + startWorkers: false, + workerChoiceStrategy: WorkerChoiceStrategies.ROUND_ROBIN, + }) + const errors = [] + pool.emitter.on(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.emitter, 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 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('bad', invalid)) + await bounded(pool.addTaskFunction('discard', () => 'discard')) + 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.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 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('bad', invalid)) + await bounded(pool.addTaskFunction('echo', data => data)) + const failed = once(pool.emitter, 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 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('bad', invalid)) + const failed = once(pool.emitter, 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 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('kept', () => 42)) + await bounded(pool.addTaskFunction('broken', invalid)) + const failed = once(pool.emitter, 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 FixedPool(1, workerFile) + try { + await bounded(once(pool.emitter, 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()) + } + }) + it.each([ ['acknowledgement', 2], ['failure', 2], diff --git a/tests/worker-files/taskFunctionRecoveryWorker.mjs b/tests/worker-files/taskFunctionRecoveryWorker.mjs index 469e28580..25cc4933d 100644 --- a/tests/worker-files/taskFunctionRecoveryWorker.mjs +++ b/tests/worker-files/taskFunctionRecoveryWorker.mjs @@ -22,6 +22,13 @@ class TaskFunctionRecoveryWorker extends Worker { 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 } @@ -44,6 +51,15 @@ class TaskFunctionRecoveryWorker extends Worker { } 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 }) diff --git a/tests/worker-files/thread/taskLifecycleWorker.mjs b/tests/worker-files/thread/taskLifecycleWorker.mjs index 7031b734b..64b10292c 100644 --- a/tests/worker-files/thread/taskLifecycleWorker.mjs +++ b/tests/worker-files/thread/taskLifecycleWorker.mjs @@ -33,12 +33,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 &&