src/worker/abstract-worker.ts

   1 import type { Worker } from 'node:cluster'
   2 import type { UUID } from 'node:crypto'
   3 import type { MessagePort } from 'node:worker_threads'
   4
   5 import { AsyncLocalStorage } from 'node:async_hooks'
   6 import { performance } from 'node:perf_hooks'
   7 import { isPromise } from 'node:util/types'
   8
   9 import type {
  10   MessageValue,
  11   Task,
  12   TaskFunctionProperties,
  13   TaskPerformance,
  14   WorkerStatistics,
  15 } from '../utility-types.js'
  16 import type {
  17   TaskAsyncFunction,
  18   TaskFunction,
  19   TaskFunctionObject,
  20   TaskFunctionOperationResult,
  21   TaskFunctions,
  22   TaskSyncFunction,
  23 } from './task-functions.js'
  24
  25 import {
  26   buildTaskFunctionProperties,
  27   DEFAULT_TASK_NAME,
  28   EMPTY_FUNCTION,
  29   isAsyncFunction,
  30   isPlainObject,
  31 } from '../utils.js'
  32 import { AbortError } from './abort-error.js'
  33 import {
  34   checkTaskFunctionName,
  35   checkValidTaskFunctionObjectEntry,
  36   checkValidWorkerOptions,
  37 } from './utils.js'
  38 import { KillBehaviors, type WorkerOptions } from './worker-options.js'
  39
  40 const DEFAULT_MAX_INACTIVE_TIME = 60000
  41 const DEFAULT_WORKER_OPTIONS: Readonly<WorkerOptions> = Object.freeze({
  42   /**
  43    * The kill behavior option on this worker or its default value.
  44    */
  45   killBehavior: KillBehaviors.SOFT,
  46   /**
  47    * The function to call when the worker is killed.
  48    */
  49   killHandler: EMPTY_FUNCTION,
  50   /**
  51    * The maximum time to keep this worker active while idle.
  52    * The pool automatically checks and terminates this worker when the time expires.
  53    */
  54   maxInactiveTime: DEFAULT_MAX_INACTIVE_TIME,
  55 })
  56
  57 /**
  58  * Base class that implements some shared logic for all poolifier workers.
  59  * @template MainWorker - Type of main worker.
  60  * @template Data - Type of data this worker receives from pool's execution. This can only be structured-cloneable data.
  61  * @template Response - Type of response the worker sends back to the main worker. This can only be structured-cloneable data.
  62  */
  63 export abstract class AbstractWorker<
  64   MainWorker extends MessagePort | Worker,
  65   Data = unknown,
  66   Response = unknown
  67 > {
  68   /**
  69    * Handler id of the `activeInterval` worker activity check.
  70    */
  71   protected activeInterval?: NodeJS.Timeout
  72   /**
  73    * Worker id.
  74    */
  75   protected abstract readonly id: number
  76   /**
  77    * Timestamp of the last task processed by this worker.
  78    */
  79   protected lastTaskTimestamp!: number
  80
  81   /**
  82    * Performance statistics computation requirements.
  83    */
  84   protected statistics?: WorkerStatistics
  85
  86   /**
  87    * Task abort functions processed by the worker when task operation 'abort' is received.
  88    */
  89   protected taskAbortFunctions: Map<UUID, () => void>
  90
  91   /**
  92    * Correlation of task function operations across protected handler overrides.
  93    */
  94   protected readonly taskFunctionOperationContext: AsyncLocalStorage<
  95     undefined | UUID
  96   > = new AsyncLocalStorage<undefined | UUID>()
  97
  98   /**
  99    * Task function object(s) processed by the worker when the pool's `execute` method is invoked.
 100    */
 101   protected taskFunctions!: Map<string, TaskFunctionObject<Data, Response>>
 102
 103   /**
 104    * The base removal result and name that confirm absence for the current operation.
 105    */
 106   private missingTaskFunctionRemoval?: TaskFunctionOperationResult
 107   private missingTaskFunctionRemovalName?: string
 108
 109   private readonly taskFunctionOperationIds = new WeakMap<
 110     TaskFunctionProperties,
 111     UUID
 112   >()
 113
 114   /**
 115    * Constructs a new poolifier worker.
 116    * @param isMain - Whether this is the main worker or not.
 117    * @param mainWorker - Reference to main worker.
 118    * @param taskFunctions - Task function(s) processed by the worker when the pool's `execute` method is invoked. The first function is the default function.
 119    * @param opts - Options for the worker.
 120    */
 121   public constructor (
 122     protected readonly isMain: boolean | undefined,
 123     private readonly mainWorker: MainWorker | null | undefined,
 124     taskFunctions: TaskFunction<Data, Response> | TaskFunctions<Data, Response>,
 125     protected opts: WorkerOptions = DEFAULT_WORKER_OPTIONS
 126   ) {
 127     if (this.isMain == null) {
 128       throw new Error('isMain parameter is mandatory')
 129     }
 130     this.checkTaskFunctions(taskFunctions)
 131     this.taskAbortFunctions = new Map<UUID, () => void>()
 132     this.checkWorkerOptions(this.opts)
 133     if (!this.isMain) {
 134       this.getMainWorker().once('message', this.handleReadyMessage.bind(this))
 135     }
 136   }
 137
 138   /**
 139    * Adds a task function to the worker.
 140    * If a task function with the same name already exists, it is replaced.
 141    * @param name - The name of the task function to add.
 142    * @param fn - The task function to add.
 143    * @returns Whether the task function was added or not.
 144    */
 145   public addTaskFunction (
 146     name: string,
 147     fn: TaskFunction<Data, Response> | TaskFunctionObject<Data, Response>
 148   ): TaskFunctionOperationResult {
 149     try {
 150       checkTaskFunctionName(name)
 151       if (name === DEFAULT_TASK_NAME) {
 152         throw new Error(
 153           'Cannot add a task function with the default reserved name'
 154         )
 155       }
 156       if (typeof fn === 'function') {
 157         fn = { taskFunction: fn } satisfies TaskFunctionObject<Data, Response>
 158       }
 159       checkValidTaskFunctionObjectEntry<Data, Response>(name, fn)
 160       fn.taskFunction = fn.taskFunction.bind(this)
 161       if (
 162         this.taskFunctions.get(name) ===
 163         this.taskFunctions.get(DEFAULT_TASK_NAME)
 164       ) {
 165         this.taskFunctions.set(DEFAULT_TASK_NAME, fn)
 166       }
 167       this.taskFunctions.set(name, fn)
 168       this.sendTaskFunctionsPropertiesToMainWorker()
 169       return { status: true }
 170     } catch (error) {
 171       return { error: error as Error, status: false }
 172     }
 173   }
 174
 175   /**
 176    * Checks if the worker has a task function with the given name.
 177    * @param name - The name of the task function to check.
 178    * @returns Whether the worker has a task function with the given name or not.
 179    */
 180   public hasTaskFunction (name: string): TaskFunctionOperationResult {
 181     try {
 182       checkTaskFunctionName(name)
 183     } catch (error) {
 184       return { error: error as Error, status: false }
 185     }
 186     return { status: this.taskFunctions.has(name) }
 187   }
 188
 189   /**
 190    * Lists the properties of the worker's task functions.
 191    * @returns The properties of the worker's task functions.
 192    */
 193   public listTaskFunctionsProperties (): TaskFunctionProperties[] {
 194     let defaultTaskFunctionName = DEFAULT_TASK_NAME
 195     for (const [name, fnObj] of this.taskFunctions) {
 196       if (
 197         name !== DEFAULT_TASK_NAME &&
 198         fnObj === this.taskFunctions.get(DEFAULT_TASK_NAME)
 199       ) {
 200         defaultTaskFunctionName = name
 201         break
 202       }
 203     }
 204     const taskFunctionsProperties: TaskFunctionProperties[] = []
 205     for (const [name, fnObj] of this.taskFunctions) {
 206       if (name === DEFAULT_TASK_NAME || name === defaultTaskFunctionName) {
 207         continue
 208       }
 209       taskFunctionsProperties.push(buildTaskFunctionProperties(name, fnObj))
 210     }
 211     return [
 212       buildTaskFunctionProperties(
 213         DEFAULT_TASK_NAME,
 214         this.taskFunctions.get(DEFAULT_TASK_NAME)
 215       ),
 216       buildTaskFunctionProperties(
 217         defaultTaskFunctionName,
 218         this.taskFunctions.get(defaultTaskFunctionName)
 219       ),
 220       ...taskFunctionsProperties,
 221     ]
 222   }
 223
 224   /**
 225    * Removes a task function from the worker.
 226    * @param name - The name of the task function to remove.
 227    * @returns Whether the task function existed and was removed or not.
 228    */
 229   public removeTaskFunction (name: string): TaskFunctionOperationResult {
 230     try {
 231       checkTaskFunctionName(name)
 232       if (name === DEFAULT_TASK_NAME) {
 233         throw new Error(
 234           'Cannot remove the task function with the default reserved name'
 235         )
 236       }
 237       if (
 238         this.taskFunctions.get(name) ===
 239         this.taskFunctions.get(DEFAULT_TASK_NAME)
 240       ) {
 241         throw new Error(
 242           'Cannot remove the task function used as the default task function'
 243         )
 244       }
 245       const deleteStatus = this.taskFunctions.delete(name)
 246       this.sendTaskFunctionsPropertiesToMainWorker()
 247       const response = { status: deleteStatus }
 248       if (!deleteStatus) {
 249         this.missingTaskFunctionRemoval = response
 250         this.missingTaskFunctionRemovalName = name
 251       }
 252       return response
 253     } catch (error) {
 254       return { error: error as Error, status: false }
 255     }
 256   }
 257
 258   /**
 259    * Sets the default task function to use in the worker.
 260    * @param name - The name of the task function to use as default task function.
 261    * @returns Whether the default task function was set or not.
 262    */
 263   public setDefaultTaskFunction (name: string): TaskFunctionOperationResult {
 264     try {
 265       checkTaskFunctionName(name)
 266       if (name === DEFAULT_TASK_NAME) {
 267         throw new Error(
 268           'Cannot set the default task function reserved name as the default task function'
 269         )
 270       }
 271       if (!this.taskFunctions.has(name)) {
 272         throw new Error(
 273           'Cannot set the default task function to a non-existing task function'
 274         )
 275       }
 276       // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
 277       this.taskFunctions.set(DEFAULT_TASK_NAME, this.taskFunctions.get(name)!)
 278       this.sendTaskFunctionsPropertiesToMainWorker()
 279       return { status: true }
 280     } catch (error) {
 281       return { error: error as Error, status: false }
 282     }
 283   }
 284
 285   /**
 286    * Returns the main worker.
 287    * @returns Reference to the main worker.
 288    * @throws {Error} If the main worker is not set.
 289    */
 290   protected getMainWorker (): MainWorker {
 291     if (this.mainWorker == null) {
 292       throw new Error('Main worker not set')
 293     }
 294     return this.mainWorker
 295   }
 296
 297   /**
 298    * Gets the task function operation UUID for a request or response.
 299    * @param message - The task function operation message.
 300    * @returns The task function operation UUID if available.
 301    */
 302   protected getTaskFunctionOperationId (
 303     message: MessageValue
 304   ): undefined | UUID {
 305     return (
 306       message.taskFunctionOperationId ??
 307       (message.taskFunctionProperties != null
 308         ? this.taskFunctionOperationIds.get(message.taskFunctionProperties)
 309         : undefined) ??
 310       this.taskFunctionOperationContext.getStore()
 311     )
 312   }
 313
 314   /**
 315    * Handles a worker error.
 316    * @param error - The error raised by the worker.
 317    * @returns The worker error object.
 318    */
 319   protected abstract handleError (error: Error): {
 320     aborted: boolean
 321     error?: Error
 322     message: string
 323     stack?: string
 324   }
 325
 326   /**
 327    * Handles a kill message sent by the main worker.
 328    * @param message - The kill message.
 329    */
 330   protected handleKillMessage (message: MessageValue<Data>): void {
 331     this.stopCheckActive()
 332     let promise: Promise<void> | undefined
 333     try {
 334       promise = this.getResultPromise(this.opts.killHandler?.())
 335     } catch {
 336       this.sendToMainWorker({ kill: 'failure' })
 337       return
 338     }
 339     if (promise != null) {
 340       promise
 341         .then(
 342           () => {
 343             this.sendToMainWorker({ kill: 'success' })
 344             return undefined
 345           },
 346           () => {
 347             this.sendToMainWorker({ kill: 'failure' })
 348             return undefined
 349           }
 350         )
 351         .catch((error: unknown) => {
 352           queueMicrotask(() => {
 353             throw error
 354           })
 355         })
 356     } else {
 357       this.sendToMainWorker({ kill: 'success' })
 358     }
 359   }
 360
 361   /**
 362    * Handles the ready message sent by the main worker.
 363    * @param message - The ready message.
 364    */
 365   protected abstract handleReadyMessage (message: MessageValue<Data>): void
 366
 367   /**
 368    * Handles a task function operation.
 369    * Deferred overrides must retain the task function properties reference or the
 370    * operation UUID when forwarding outside the incoming asynchronous context.
 371    * An acknowledgement without a recoverable UUID rejects the pool operation.
 372    * @param message - The task function operation request.
 373    */
 374   protected handleTaskFunctionOperationMessage (
 375     message: MessageValue<Data>
 376   ): void {
 377     const { taskFunction, taskFunctionOperation, taskFunctionProperties } =
 378       message
 379     if (taskFunctionProperties == null) {
 380       throw new Error(
 381         'Cannot handle task function operation message without task function properties'
 382       )
 383     }
 384     let taskFunctionOperationNotFound = false
 385     let response: TaskFunctionOperationResult
 386     switch (taskFunctionOperation) {
 387       case 'add':
 388         if (typeof taskFunction !== 'string') {
 389           throw new Error(
 390             `Cannot handle task function operation ${taskFunctionOperation} message without task function`
 391           )
 392         }
 393         response = this.addTaskFunction(taskFunctionProperties.name, {
 394           // eslint-disable-next-line @typescript-eslint/no-implied-eval, no-new-func, @typescript-eslint/no-unsafe-call
 395           taskFunction: new Function(
 396             `return (${taskFunction})`
 397           )() as TaskFunction<Data, Response>,
 398           ...(taskFunctionProperties.priority != null && {
 399             priority: taskFunctionProperties.priority,
 400           }),
 401           ...(taskFunctionProperties.strategy != null && {
 402             strategy: taskFunctionProperties.strategy,
 403           }),
 404           ...(taskFunctionProperties.workerNodeKeys != null && {
 405             workerNodeKeys: taskFunctionProperties.workerNodeKeys,
 406           }),
 407         })
 408         break
 409       case 'default':
 410         response = this.setDefaultTaskFunction(taskFunctionProperties.name)
 411         break
 412       case 'remove':
 413         this.missingTaskFunctionRemoval = undefined
 414         this.missingTaskFunctionRemovalName = undefined
 415         try {
 416           const name = taskFunctionProperties.name
 417           response = this.removeTaskFunction(name)
 418           taskFunctionOperationNotFound =
 419             !response.status &&
 420             response.error == null &&
 421             Object.is(response, this.missingTaskFunctionRemoval) &&
 422             Object.is(name, this.missingTaskFunctionRemovalName)
 423         } finally {
 424           this.missingTaskFunctionRemoval = undefined
 425           this.missingTaskFunctionRemovalName = undefined
 426         }
 427         break
 428       default:
 429         response = {
 430           error: new Error(
 431             // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
 432             `Unknown task function operation: ${taskFunctionOperation!}`
 433           ),
 434           status: false,
 435         }
 436         break
 437     }
 438     const { error, status } = response
 439     this.sendToMainWorker({
 440       taskFunctionOperation,
 441       taskFunctionOperationId: this.getTaskFunctionOperationId(message),
 442       taskFunctionOperationStatus: status,
 443       taskFunctionProperties,
 444       ...(taskFunctionOperationNotFound && { taskFunctionOperationNotFound }),
 445       ...(!status &&
 446         error != null && {
 447         workerError: {
 448           name: taskFunctionProperties.name,
 449           ...this.handleError(error),
 450         },
 451       }),
 452     })
 453   }
 454
 455   /**
 456    * Worker message listener.
 457    * @param message - The received message.
 458    */
 459   protected messageListener (message: MessageValue<Data>): void {
 460     this.checkMessageWorkerId(message)
 461     const {
 462       checkActive,
 463       data,
 464       kill,
 465       statistics,
 466       taskFunctionOperation,
 467       taskId,
 468       taskOperation,
 469     } = message
 470     if (statistics != null) {
 471       // Statistics message received
 472       this.statistics = statistics
 473     } else if (checkActive != null) {
 474       // Check active message received
 475       checkActive ? this.startCheckActive() : this.stopCheckActive()
 476     } else if (taskFunctionOperation != null) {
 477       // Task function operation message received
 478       if (
 479         message.taskFunctionProperties != null &&
 480         typeof message.taskFunctionProperties === 'object' &&
 481         message.taskFunctionOperationId != null
 482       ) {
 483         this.taskFunctionOperationIds.set(
 484           message.taskFunctionProperties,
 485           message.taskFunctionOperationId
 486         )
 487       }
 488       this.taskFunctionOperationContext.run(
 489         message.taskFunctionOperationId,
 490         this.runTaskFunctionOperation,
 491         message
 492       )
 493     } else if (taskId != null && data != null) {
 494       // Task message received
 495       this.run(message)
 496     } else if (taskOperation === 'abort' && taskId != null) {
 497       // Abort task operation message received
 498       if (this.taskAbortFunctions.has(taskId)) {
 499         this.taskAbortFunctions.get(taskId)?.()
 500       }
 501     } else if (kill === true) {
 502       // Kill message received
 503       this.handleKillMessage(message)
 504     }
 505   }
 506
 507   /**
 508    * Runs the given task.
 509    * @param task - The task to execute.
 510    */
 511   protected readonly run = (task: Task<Data>): void => {
 512     const { abortable, name, taskId } = task
 513     const taskFunctionName = name ?? DEFAULT_TASK_NAME
 514     if (!this.taskFunctions.has(taskFunctionName)) {
 515       this.sendTaskError(
 516         task,
 517         new Error(`Task function '${taskFunctionName}' not found`)
 518       )
 519       return
 520     }
 521     let fn: TaskFunction<Data, Response>
 522     if (abortable === true) {
 523       // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
 524       fn = this.getAbortableTaskFunction(taskFunctionName, taskId!)
 525     } else {
 526       // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
 527       fn = this.taskFunctions.get(taskFunctionName)!.taskFunction
 528     }
 529     if (isAsyncFunction(fn)) {
 530       this.runAsync(fn as TaskAsyncFunction<Data, Response>, task)
 531     } else {
 532       this.runSync(fn as TaskSyncFunction<Data, Response>, task)
 533     }
 534   }
 535
 536   /**
 537    * Runs the given task function asynchronously.
 538    * @param fn - Task function that will be executed.
 539    * @param task - Input data for the task function.
 540    */
 541   protected readonly runAsync = (
 542     fn: TaskAsyncFunction<Data, Response>,
 543     task: Task<Data>
 544   ): void => {
 545     try {
 546       const taskPerformance = this.beginTaskPerformance(task.name)
 547       const result = fn(task.data)
 548       this.completeAsyncTask(
 549         this.getResultPromise(result) ?? Promise.resolve(result),
 550         task,
 551         taskPerformance
 552       )
 553     } catch (error) {
 554       try {
 555         this.sendTaskError(task, error as Error)
 556       } finally {
 557         this.finishTask(task)
 558       }
 559     }
 560   }
 561
 562   /**
 563    * Runs the given task function synchronously.
 564    * @param fn - Task function that will be executed.
 565    * @param task - Input data for the task function.
 566    */
 567   protected readonly runSync = (
 568     fn: TaskSyncFunction<Data, Response>,
 569     task: Task<Data>
 570   ): void => {
 571     const { data, name, taskId } = task
 572     let asynchronous = false
 573     try {
 574       let taskPerformance = this.beginTaskPerformance(name)
 575       const res = fn(data)
 576       const promise = this.getResultPromise(res)
 577       if (promise != null) {
 578         this.completeAsyncTask(promise, task, taskPerformance)
 579         asynchronous = true
 580         return
 581       }
 582       taskPerformance = this.endTaskPerformance(taskPerformance)
 583       this.sendToMainWorker({
 584         data: res,
 585         taskId,
 586         taskPerformance,
 587       })
 588     } catch (error) {
 589       this.sendTaskError(task, error as Error)
 590     } finally {
 591       if (!asynchronous) this.finishTask(task)
 592     }
 593   }
 594
 595   /**
 596    * Sends task functions properties to the main worker.
 597    */
 598   protected sendTaskFunctionsPropertiesToMainWorker (): void {
 599     this.sendToMainWorker({
 600       taskFunctionsProperties: this.listTaskFunctionsProperties(),
 601     })
 602   }
 603
 604   /**
 605    * Sends a message to main worker.
 606    * @param message - The response message.
 607    */
 608   protected abstract sendToMainWorker (
 609     message: MessageValue<Response, Data>
 610   ): void
 611
 612   private beginTaskPerformance (name?: string): TaskPerformance {
 613     if (this.statistics == null) {
 614       throw new Error('Performance statistics computation requirements not set')
 615     }
 616     return {
 617       name: name ?? DEFAULT_TASK_NAME,
 618       timestamp: performance.now(),
 619       ...(this.statistics.elu && {
 620         elu: performance.eventLoopUtilization(),
 621       }),
 622     }
 623   }
 624
 625   /**
 626    * Checks if the worker should be terminated, because its living too long.
 627    */
 628   private checkActive (): void {
 629     if (
 630       performance.now() - this.lastTaskTimestamp >
 631       (this.opts.maxInactiveTime ?? DEFAULT_MAX_INACTIVE_TIME)
 632     ) {
 633       this.sendToMainWorker({ kill: this.opts.killBehavior })
 634     }
 635   }
 636
 637   /**
 638    * Check if the message worker id is set and matches the worker id.
 639    * @param message - The message to check.
 640    * @throws {Error} If the message worker id is not set or does not match the worker id.
 641    */
 642   private checkMessageWorkerId (message: MessageValue<Data>): void {
 643     if (message.workerId == null) {
 644       throw new Error(
 645         `Message worker id is not set: ${JSON.stringify(message)}`
 646       )
 647     }
 648     if (message.workerId !== this.id) {
 649       throw new Error(
 650         `Message worker id ${message.workerId.toString()} does not match the worker id ${this.id.toString()}: ${JSON.stringify(message)}`
 651       )
 652     }
 653   }
 654
 655   /**
 656    * Checks if the `taskFunctions` parameter is passed to the constructor and valid.
 657    * @param taskFunctions - The task function(s) parameter that should be checked.
 658    */
 659   private checkTaskFunctions (
 660     taskFunctions:
 661       | TaskFunction<Data, Response>
 662       | TaskFunctions<Data, Response>
 663       | undefined
 664   ): void {
 665     if (taskFunctions == null) {
 666       throw new Error('taskFunctions parameter is mandatory')
 667     }
 668     this.taskFunctions = new Map<string, TaskFunctionObject<Data, Response>>()
 669     if (typeof taskFunctions === 'function') {
 670       const fnObj = { taskFunction: taskFunctions.bind(this) }
 671       this.taskFunctions.set(DEFAULT_TASK_NAME, fnObj)
 672       this.taskFunctions.set(
 673         typeof taskFunctions.name === 'string' &&
 674           taskFunctions.name.trim().length > 0
 675           ? taskFunctions.name
 676           : 'fn1',
 677         fnObj
 678       )
 679     } else if (isPlainObject(taskFunctions)) {
 680       let firstEntry = true
 681       for (let [name, fnObj] of Object.entries(taskFunctions)) {
 682         if (typeof fnObj === 'function') {
 683           fnObj = { taskFunction: fnObj } satisfies TaskFunctionObject<
 684             Data,
 685             Response
 686           >
 687         }
 688         checkValidTaskFunctionObjectEntry<Data, Response>(name, fnObj)
 689         fnObj.taskFunction = fnObj.taskFunction.bind(this)
 690         if (firstEntry) {
 691           this.taskFunctions.set(DEFAULT_TASK_NAME, fnObj)
 692           firstEntry = false
 693         }
 694         this.taskFunctions.set(name, fnObj)
 695       }
 696       if (firstEntry) {
 697         throw new Error('taskFunctions parameter object is empty')
 698       }
 699     } else {
 700       throw new TypeError(
 701         'taskFunctions parameter is not a function or a plain object'
 702       )
 703     }
 704   }
 705
 706   private checkWorkerOptions (opts: WorkerOptions): void {
 707     checkValidWorkerOptions(opts)
 708     this.opts = { ...DEFAULT_WORKER_OPTIONS, ...opts }
 709   }
 710
 711   private completeAsyncTask (
 712     promise: Promise<Response>,
 713     task: Task<Data>,
 714     taskPerformance: TaskPerformance
 715   ): void {
 716     const { taskId } = task
 717     promise
 718       .then(res => {
 719         taskPerformance = this.endTaskPerformance(taskPerformance)
 720         this.sendToMainWorker({
 721           data: res,
 722           taskId,
 723           taskPerformance,
 724         })
 725         return undefined
 726       })
 727       .catch((error: unknown) => {
 728         this.sendTaskError(task, error as Error)
 729       })
 730       .finally(() => {
 731         this.finishTask(task)
 732       })
 733       .catch((error: unknown) => {
 734         queueMicrotask(() => {
 735           throw error
 736         })
 737       })
 738   }
 739
 740   private endTaskPerformance (
 741     taskPerformance: TaskPerformance
 742   ): TaskPerformance {
 743     if (this.statistics == null) {
 744       throw new Error('Performance statistics computation requirements not set')
 745     }
 746     return {
 747       ...taskPerformance,
 748       ...(this.statistics.runTime && {
 749         runTime: performance.now() - taskPerformance.timestamp,
 750       }),
 751       ...(this.statistics.elu && {
 752         elu: performance.eventLoopUtilization(taskPerformance.elu),
 753       }),
 754     }
 755   }
 756
 757   private finishTask (task: Task<Data>): void {
 758     this.updateLastTaskTimestamp()
 759     if (task.abortable === true && task.taskId != null) {
 760       this.taskAbortFunctions.delete(task.taskId)
 761     }
 762   }
 763
 764   /**
 765    * Gets abortable task function.
 766    * An abortable promise is built to permit the task to be aborted.
 767    * @param name - The name of the task.
 768    * @param taskId - The task id.
 769    * @returns The abortable task function.
 770    */
 771   private getAbortableTaskFunction (
 772     name: string,
 773     taskId: UUID
 774   ): TaskAsyncFunction<Data, Response> {
 775     return async (data?: Data): Promise<Response> => {
 776       let result: PromiseLike<Response> | Response | undefined
 777       // Keep cancellation active while awaiting the task. Resolve only this gate,
 778       // not the task result, to avoid an extra promise assimilation.
 779       await new Promise<void>((resolve, reject: (reason?: unknown) => void) => {
 780         this.taskAbortFunctions.set(taskId, () => {
 781           reject(new AbortError(`Task '${name}' id '${taskId}' aborted`))
 782         })
 783         // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
 784         const taskFunction = this.taskFunctions.get(name)!.taskFunction
 785         const value = taskFunction(data)
 786         const promise = this.getResultPromise(value)
 787         if (promise != null) {
 788           // eslint-disable-next-line promise/catch-or-return -- The second callback handles rejection.
 789           promise.then(value => {
 790             result = value
 791             resolve()
 792             return undefined
 793           }, reject)
 794         } else {
 795           result = value
 796           resolve()
 797         }
 798       })
 799       return result as Response
 800     }
 801   }
 802
 803   private getResultPromise<Result>(
 804     result: PromiseLike<Result> | Result
 805   ): Promise<Result> | undefined {
 806     if (
 807       result == null ||
 808       (typeof result !== 'object' && typeof result !== 'function')
 809     ) {
 810       return
 811     }
 812     if (
 813       isPromise(result) &&
 814       Object.getPrototypeOf(result) === Promise.prototype &&
 815       !Object.hasOwn(result, 'then')
 816     ) {
 817       return result as Promise<Result>
 818     }
 819     const then: unknown = Reflect.get(result, 'then')
 820     if (typeof then === 'function') {
 821       // The declared result type determines the callable's fulfillment type.
 822       const promiseThen = then as PromiseLike<Result>['then']
 823       return Promise.resolve({
 824         then (...args: Parameters<typeof promiseThen>): void {
 825           Reflect.apply(promiseThen, result, args)
 826         },
 827       })
 828     }
 829     return undefined
 830   }
 831
 832   private readonly runTaskFunctionOperation = (
 833     message: MessageValue<Data>
 834   ): void => {
 835     this.handleTaskFunctionOperationMessage(message)
 836   }
 837
 838   private sendTaskError (task: Task<Data>, error: Error): void {
 839     const message = {
 840       taskId: task.taskId,
 841       workerError: {
 842         ...((task.transferList == null || task.transferList.length === 0) && {
 843           data: task.data,
 844         }),
 845         name: task.name,
 846         ...this.handleError(error),
 847       },
 848     } satisfies MessageValue<Response, Data>
 849     try {
 850       this.sendToMainWorker(message)
 851     } catch (sendError) {
 852       if (
 853         !(sendError instanceof Error) ||
 854         sendError.name !== 'DataCloneError' ||
 855         !Object.hasOwn(message.workerError, 'data')
 856       ) {
 857         throw sendError
 858       }
 859       delete message.workerError.data
 860       this.sendToMainWorker(message)
 861     }
 862   }
 863
 864   /**
 865    * Starts the worker check active interval.
 866    */
 867   private startCheckActive (): void {
 868     this.lastTaskTimestamp = performance.now()
 869     this.activeInterval = setInterval(
 870       this.checkActive.bind(this),
 871       (this.opts.maxInactiveTime ?? DEFAULT_MAX_INACTIVE_TIME) / 2
 872     )
 873     this.activeInterval.unref()
 874   }
 875
 876   /**
 877    * Stops the worker check active interval.
 878    */
 879   private stopCheckActive (): void {
 880     if (this.activeInterval != null) {
 881       clearInterval(this.activeInterval)
 882       this.activeInterval = undefined
 883     }
 884   }
 885
 886   private updateLastTaskTimestamp (): void {
 887     if (this.activeInterval != null) {
 888       this.lastTaskTimestamp = performance.now()
 889     }
 890   }
 891 }