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 d5fed6045485410e27b27ede2aa4b797ac8a91ff..deddf72e5df1bf996e8b380a087c29de614b274d 100644 (file)
--- 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 43de06fc8f357c97a291c375770e95801589ff67..9064825ab96bd96c92d64cbbc4dfd6bc51925b7a 100644 (file)
--- 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 e1bb84084b14b043a9c41dd44911d7b2e25bfefe..870246399cbd4ef40b25bc75e68caa88cc093dc2 100644 (file)
--- a/src/utility-types.ts
+++ b/src/utility-types.ts
@@ -49,6 +49,10 @@ export interface MessageValue<Data = unknown, ErrorData = unknown>
    * 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 6207a7b9fe700f715c3011539e108fe74b52e870..1ad9cceb0bcec3ced0bc8c98b48108f9c417fd52 100644 (file)
--- a/src/worker/abstract-worker.ts
+++ b/src/worker/abstract-worker.ts
@@ -100,6 +100,12 @@ export abstract class AbstractWorker<
    */
   protected taskFunctions!: Map<string, TaskFunctionObject<Data, Response>>
 
+  /**
+   * 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 533b8d86a8fc4bf36f1908acd46da0b6357ed079..c6fc62f897c9012b0b361d87f262eea338f9fbbc 100644 (file)
--- 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 469e2858001cfb8506fb2f9742b7c289af4ea02a..25cc4933d45b3bd9e5abf6987e14ac4aac18d240 100644 (file)
--- 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 7031b734b60c20210bf85037037b5f130dfb53e0..64b10292c41ddfb713250308d86db67171e170dc 100644 (file)
--- 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 &&