From 248383aab1dc5e3539fe4019fc7e3a695613ad34 Mon Sep 17 00:00:00 2001 From: Juergen Kunz Date: Sun, 25 Jan 2026 23:29:00 +0000 Subject: [PATCH] BREAKING CHANGE(taskbuffer): Change default Task error handling: trigger() now rejects when taskFunction throws; add catchErrors option (default false) to preserve previous swallow behavior; track errors (lastError, errorCount) and expose them in metadata; improve error propagation and logging across runners, chains, parallels and debounced tasks; add tests and documentation for new behavior. --- changelog.md | 11 + npmextra.json | 22 +- readme.hints.md | 29 +- test/test.11.errorhandling.ts | 363 +++++++++++++++++++++++++ test/test.9.steps.ts | 2 +- ts/00_commitinfo_data.ts | 2 +- ts/taskbuffer.classes.bufferrunner.ts | 17 +- ts/taskbuffer.classes.cyclecounter.ts | 12 + ts/taskbuffer.classes.task.ts | 85 ++++-- ts/taskbuffer.classes.taskchain.ts | 27 +- ts/taskbuffer.classes.taskdebounced.ts | 14 +- ts/taskbuffer.classes.taskmanager.ts | 17 +- ts/taskbuffer.classes.taskparallel.ts | 4 +- ts/taskbuffer.classes.taskrunner.ts | 29 +- ts/taskbuffer.interfaces.ts | 2 + ts_web/00_commitinfo_data.ts | 2 +- 16 files changed, 575 insertions(+), 63 deletions(-) create mode 100644 test/test.11.errorhandling.ts diff --git a/changelog.md b/changelog.md index 2cced2d..4f36a62 100644 --- a/changelog.md +++ b/changelog.md @@ -1,5 +1,16 @@ # Changelog +## 2026-01-25 - 4.0.0 - BREAKING CHANGE(taskbuffer) +Change default Task error handling: trigger() now rejects when taskFunction throws; add catchErrors option (default false) to preserve previous swallow behavior; track errors (lastError, errorCount) and expose them in metadata; improve error propagation and logging across runners, chains, parallels and debounced tasks; add tests and documentation for new behavior. + +- Introduce catchErrors option on Task (default: false) — previously errors were swallowed by default +- Tasks now set lastError and increment errorCount when failures occur; clearError() added to reset error state +- getMetadata() now reports status 'failed' and includes lastError and errorCount +- Task.run flow updated to reset error state at start, log errors, and either swallow or rethrow based on catchErrors +- BufferRunner, TaskRunner, Taskchain, Taskparallel, TaskDebounced and TaskManager updated to handle errors, avoid hanging promises, and use logger instead of console +- Added comprehensive tests (test/test.11.errorhandling.ts) and readme hints documenting the new error-handling behavior (v3.6.0+) +- npmextra.json updated for @git.zone/cli and release registries + ## 2025-12-04 - 3.5.0 - feat(core) Add debounced tasks and step-based progress tracking; upgrade deps and improve dashboard and scheduling diff --git a/npmextra.json b/npmextra.json index d5fae86..66724e3 100644 --- a/npmextra.json +++ b/npmextra.json @@ -1,10 +1,5 @@ { - "npmci": { - "npmGlobalTools": [], - "npmAccessLevel": "public", - "npmRegistryUrl": "registry.npmjs.org" - }, - "gitzone": { + "@git.zone/cli": { "projectType": "npm", "module": { "githost": "code.foss.global", @@ -25,9 +20,20 @@ "debounced tasks", "distributed coordination" ] + }, + "release": { + "registries": [ + "https://verdaccio.lossless.digital", + "https://registry.npmjs.org" + ], + "accessLevel": "public" } }, - "tsdoc": { + "@git.zone/tsdoc": { "legal": "\n## License and Legal Information\n\nThis repository contains open-source code that is licensed under the MIT License. A copy of the MIT License can be found in the [license](license) file within this repository. \n\n**Please note:** The MIT License does not grant permission to use the trade names, trademarks, service marks, or product names of the project, except as required for reasonable and customary use in describing the origin of the work and reproducing the content of the NOTICE file.\n\n### Trademarks\n\nThis project is owned and maintained by Task Venture Capital GmbH. The names and logos associated with Task Venture Capital GmbH and any related products or services are trademarks of Task Venture Capital GmbH and are not included within the scope of the MIT license granted herein. Use of these trademarks must comply with Task Venture Capital GmbH's Trademark Guidelines, and any usage must be approved in writing by Task Venture Capital GmbH.\n\n### Company Information\n\nTask Venture Capital GmbH \nRegistered at District court Bremen HRB 35230 HB, Germany\n\nFor any legal inquiries or if you require further information, please contact us via email at hello@task.vc.\n\nBy using this repository, you acknowledge that you have read this section, agree to comply with its terms, and understand that the licensing of the code does not imply endorsement by Task Venture Capital GmbH of any derivative works.\n" + }, + "@ship.zone/szci": { + "npmGlobalTools": [], + "npmRegistryUrl": "registry.npmjs.org" } -} +} \ No newline at end of file diff --git a/readme.hints.md b/readme.hints.md index 0519ecb..b9e6906 100644 --- a/readme.hints.md +++ b/readme.hints.md @@ -1 +1,28 @@ - \ No newline at end of file +# Taskbuffer Hints + +## Error Handling (v3.6.0+) +- `Task` now has `catchErrors` constructor option (default: `false`) +- Default behavior: `trigger()` rejects when taskFunction throws (breaking change from pre-3.6) +- Set `catchErrors: true` to swallow errors (old behavior) - returns `undefined` on error +- Error state tracked via `lastError?: Error`, `errorCount: number`, `clearError()` +- `getMetadata()` status uses all four values: `'idle'` | `'running'` | `'completed'` | `'failed'` +- All peripheral classes (Taskchain, Taskparallel, TaskRunner, BufferRunner, TaskDebounced, TaskManager) have proper error propagation/handling +- `console.log` calls replaced with `logger.log()` throughout + +## Breaking API Rename (TaskRunner) +- `maxParrallelJobs` → `maxParallelJobs` +- `qeuedTasks` → `queuedTasks` +- JSDoc typos fixed: "qeue" → "queue", "wether" → "whether", "loose" → "lose" +- The `setMaxParallelJobs()` parameter also renamed from `maxParrallelJobsArg` to `maxParallelJobsArg` + +## Error Context Improvements +- **TaskChain**: Errors now wrap the original with context: chain name, failing task name, and task index. Original error preserved via `.cause` +- **BufferRunner**: When `catchErrors: false`, buffered task errors now reject the trigger promise (via `CycleCounter.informOfCycleError`) instead of silently resolving with `undefined` +- **TaskChain stubs completed**: `removeTask(task)` returns `boolean`, `shiftTask()` returns `Task | undefined` + +## Project Structure +- Source in `ts/`, web components in `ts_web/` +- Tests in `test/` - naming: `*.node.ts`, `*.browser.ts`, `*.both.ts` +- Logger: `ts/taskbuffer.logging.ts` exports `logger` (ConsoleLog from smartlog) +- Build: `pnpm build` (tsbuild tsfolders) +- Test: `pnpm test` or `tstest test/test.XX.name.ts --verbose` diff --git a/test/test.11.errorhandling.ts b/test/test.11.errorhandling.ts new file mode 100644 index 0000000..a812537 --- /dev/null +++ b/test/test.11.errorhandling.ts @@ -0,0 +1,363 @@ +import { expect, tap } from '@git.zone/tstest/tapbundle'; +import * as taskbuffer from '../ts/index.js'; +import * as smartdelay from '@push.rocks/smartdelay'; + +// Test 1: Default task rejects on error (catchErrors: false) +tap.test('should reject when taskFunction throws (default catchErrors: false)', async () => { + const failingTask = new taskbuffer.Task({ + name: 'failing-task-default', + taskFunction: async () => { + throw new Error('intentional failure'); + }, + }); + + let didReject = false; + try { + await failingTask.trigger(); + } catch (err) { + didReject = true; + expect(err).toBeInstanceOf(Error); + expect((err as Error).message).toEqual('intentional failure'); + } + expect(didReject).toBeTrue(); +}); + +// Test 2: Task with catchErrors: true resolves on error +tap.test('should resolve with undefined when catchErrors is true', async () => { + const failingTask = new taskbuffer.Task({ + name: 'failing-task-catch', + catchErrors: true, + taskFunction: async () => { + throw new Error('swallowed failure'); + }, + }); + + const result = await failingTask.trigger(); + expect(result).toBeUndefined(); +}); + +// Test 3: lastError and errorCount are set after failure +tap.test('should set lastError and errorCount after failure', async () => { + const failingTask = new taskbuffer.Task({ + name: 'failing-task-state', + catchErrors: true, + taskFunction: async () => { + throw new Error('tracked failure'); + }, + }); + + await failingTask.trigger(); + expect(failingTask.lastError).toBeInstanceOf(Error); + expect(failingTask.lastError.message).toEqual('tracked failure'); + expect(failingTask.errorCount).toEqual(1); + + // Run again to verify errorCount increments + await failingTask.trigger(); + expect(failingTask.errorCount).toEqual(2); +}); + +// Test 4: Error state resets on successful re-run +tap.test('should reset lastError on successful re-run', async () => { + let shouldFail = true; + const task = new taskbuffer.Task({ + name: 'intermittent-task', + catchErrors: true, + taskFunction: async () => { + if (shouldFail) { + throw new Error('first run fails'); + } + return 'success'; + }, + }); + + // First run: fail + await task.trigger(); + expect(task.lastError).toBeInstanceOf(Error); + expect(task.errorCount).toEqual(1); + + // Second run: succeed + shouldFail = false; + const result = await task.trigger(); + expect(result).toEqual('success'); + expect(task.lastError).toBeUndefined(); + // errorCount should still be 1 since the second run succeeded + expect(task.errorCount).toEqual(1); +}); + +// Test 5: Taskchain rejects when a child task throws +tap.test('should reject Taskchain when a child task throws', async () => { + const goodTask = new taskbuffer.Task({ + name: 'good-chain-task', + taskFunction: async () => 'ok', + }); + const badTask = new taskbuffer.Task({ + name: 'bad-chain-task', + taskFunction: async () => { + throw new Error('chain failure'); + }, + }); + + const chain = new taskbuffer.Taskchain({ + name: 'test-chain', + taskArray: [goodTask, badTask], + }); + + let didReject = false; + try { + await chain.trigger(); + } catch (err) { + didReject = true; + } + expect(didReject).toBeTrue(); +}); + +// Test 6: Taskparallel rejects when a child task throws +tap.test('should reject Taskparallel when a child task throws', async () => { + const goodTask = new taskbuffer.Task({ + name: 'good-parallel-task', + taskFunction: async () => 'ok', + }); + const badTask = new taskbuffer.Task({ + name: 'bad-parallel-task', + taskFunction: async () => { + throw new Error('parallel failure'); + }, + }); + + const parallel = new taskbuffer.Taskparallel({ + taskArray: [goodTask, badTask], + }); + + let didReject = false; + try { + await parallel.trigger(); + } catch (err) { + didReject = true; + } + expect(didReject).toBeTrue(); +}); + +// Test 7: TaskRunner continues processing after a task error +tap.test('should continue TaskRunner queue after a task error', async () => { + const runner = new taskbuffer.TaskRunner(); + const executionOrder: string[] = []; + + const badTask = new taskbuffer.Task({ + name: 'runner-bad-task', + taskFunction: async () => { + executionOrder.push('bad'); + throw new Error('runner task failure'); + }, + }); + const goodTask = new taskbuffer.Task({ + name: 'runner-good-task', + taskFunction: async () => { + executionOrder.push('good'); + }, + }); + + await runner.start(); + runner.addTask(badTask); + runner.addTask(goodTask); + + // Wait for both tasks to be processed + await smartdelay.delayFor(500); + await runner.stop(); + + expect(executionOrder).toContain('bad'); + expect(executionOrder).toContain('good'); +}); + +// Test 8: BufferRunner handles errors without hanging +tap.test('should handle BufferRunner errors without hanging', async () => { + let callCount = 0; + const bufferedTask = new taskbuffer.Task({ + name: 'buffer-error-task', + buffered: true, + bufferMax: 3, + catchErrors: true, + taskFunction: async () => { + callCount++; + throw new Error('buffer failure'); + }, + }); + + await bufferedTask.trigger(); + // The task should have executed and not hung + expect(callCount).toBeGreaterThan(0); +}); + +// Test 9: clearError() resets error state +tap.test('should reset error state with clearError()', async () => { + const task = new taskbuffer.Task({ + name: 'clearable-task', + catchErrors: true, + taskFunction: async () => { + throw new Error('to be cleared'); + }, + }); + + await task.trigger(); + expect(task.lastError).toBeInstanceOf(Error); + + task.clearError(); + expect(task.lastError).toBeUndefined(); + // errorCount should remain unchanged + expect(task.errorCount).toEqual(1); +}); + +// Test 10: getMetadata() reflects error state +tap.test('should reflect error state in getMetadata()', async () => { + const task = new taskbuffer.Task({ + name: 'metadata-error-task', + catchErrors: true, + taskFunction: async () => { + throw new Error('metadata failure'); + }, + }); + + // Before any run + let metadata = task.getMetadata(); + expect(metadata.status).toEqual('idle'); + expect(metadata.errorCount).toEqual(0); + + // After failing + await task.trigger(); + metadata = task.getMetadata(); + expect(metadata.status).toEqual('failed'); + expect(metadata.lastError).toEqual('metadata failure'); + expect(metadata.errorCount).toEqual(1); +}); + +// Test 11: TaskChain error includes task name and index context +tap.test('should include task name and index in TaskChain error message', async () => { + const goodTask = new taskbuffer.Task({ + name: 'chain-step-ok', + taskFunction: async () => 'ok', + }); + const badTask = new taskbuffer.Task({ + name: 'chain-step-fail', + taskFunction: async () => { + throw new Error('step exploded'); + }, + }); + + const chain = new taskbuffer.Taskchain({ + name: 'context-chain', + taskArray: [goodTask, badTask], + }); + + let caughtError: Error | null = null; + try { + await chain.trigger(); + } catch (err) { + caughtError = err as Error; + } + expect(caughtError).toBeInstanceOf(Error); + expect(caughtError.message).toInclude('context-chain'); + expect(caughtError.message).toInclude('chain-step-fail'); + expect(caughtError.message).toInclude('index 1'); + expect((caughtError as any).cause).toBeInstanceOf(Error); + expect(((caughtError as any).cause as Error).message).toEqual('step exploded'); +}); + +// Test 12: BufferRunner error propagation with catchErrors: false +tap.test('should reject buffered task when catchErrors is false', async () => { + const bufferedTask = new taskbuffer.Task({ + name: 'buffer-reject-task', + buffered: true, + bufferMax: 3, + catchErrors: false, + taskFunction: async () => { + throw new Error('buffer reject failure'); + }, + }); + + let didReject = false; + try { + await bufferedTask.trigger(); + } catch (err) { + didReject = true; + expect(err).toBeInstanceOf(Error); + expect((err as Error).message).toEqual('buffer reject failure'); + } + expect(didReject).toBeTrue(); +}); + +// Test 13: Taskchain removeTask removes and returns true +tap.test('should remove a task from Taskchain', async () => { + const task1 = new taskbuffer.Task({ + name: 'removable-1', + taskFunction: async () => 'a', + }); + const task2 = new taskbuffer.Task({ + name: 'removable-2', + taskFunction: async () => 'b', + }); + + const chain = new taskbuffer.Taskchain({ + name: 'remove-chain', + taskArray: [task1, task2], + }); + + const removed = chain.removeTask(task1); + expect(removed).toBeTrue(); + expect(chain.taskArray.length).toEqual(1); + expect(chain.taskArray[0] === task2).toBeTrue(); +}); + +// Test 14: Taskchain removeTask returns false for unknown task +tap.test('should return false when removing a task not in Taskchain', async () => { + const task1 = new taskbuffer.Task({ + name: 'existing', + taskFunction: async () => 'a', + }); + const unknown = new taskbuffer.Task({ + name: 'unknown', + taskFunction: async () => 'b', + }); + + const chain = new taskbuffer.Taskchain({ + name: 'remove-false-chain', + taskArray: [task1], + }); + + const removed = chain.removeTask(unknown); + expect(removed).toBeFalse(); + expect(chain.taskArray.length).toEqual(1); +}); + +// Test 15: Taskchain shiftTask returns first task and shortens array +tap.test('should shift the first task from Taskchain', async () => { + const task1 = new taskbuffer.Task({ + name: 'shift-1', + taskFunction: async () => 'a', + }); + const task2 = new taskbuffer.Task({ + name: 'shift-2', + taskFunction: async () => 'b', + }); + + const chain = new taskbuffer.Taskchain({ + name: 'shift-chain', + taskArray: [task1, task2], + }); + + const shifted = chain.shiftTask(); + expect(shifted === task1).toBeTrue(); + expect(chain.taskArray.length).toEqual(1); + expect(chain.taskArray[0] === task2).toBeTrue(); +}); + +// Test 16: Taskchain shiftTask returns undefined on empty array +tap.test('should return undefined when shifting from empty Taskchain', async () => { + const chain = new taskbuffer.Taskchain({ + name: 'empty-shift-chain', + taskArray: [], + }); + + const shifted = chain.shiftTask(); + expect(shifted).toBeUndefined(); +}); + +export default tap.start(); diff --git a/test/test.9.steps.ts b/test/test.9.steps.ts index b13491c..85d91e3 100644 --- a/test/test.9.steps.ts +++ b/test/test.9.steps.ts @@ -147,7 +147,7 @@ tap.test('Task should provide complete metadata', async () => { // Get metadata after execution metadata = task.getMetadata(); - expect(metadata.status).toEqual('idle'); + expect(metadata.status).toEqual('completed'); expect(metadata.runCount).toEqual(1); expect(metadata.currentProgress).toEqual(100); }); diff --git a/ts/00_commitinfo_data.ts b/ts/00_commitinfo_data.ts index de407da..9d0be6b 100644 --- a/ts/00_commitinfo_data.ts +++ b/ts/00_commitinfo_data.ts @@ -3,6 +3,6 @@ */ export const commitinfo = { name: '@push.rocks/taskbuffer', - version: '3.5.0', + version: '4.0.0', description: 'A flexible task management library supporting TypeScript, allowing for task buffering, scheduling, and execution with dependency management.' } diff --git a/ts/taskbuffer.classes.bufferrunner.ts b/ts/taskbuffer.classes.bufferrunner.ts index 9303627..58e0c3c 100644 --- a/ts/taskbuffer.classes.bufferrunner.ts +++ b/ts/taskbuffer.classes.bufferrunner.ts @@ -1,4 +1,5 @@ import { Task } from './taskbuffer.classes.task.js'; +import { logger } from './taskbuffer.logging.js'; export class BufferRunner { public task: Task; @@ -24,9 +25,19 @@ export class BufferRunner { private async _run(x: any) { this.task.running = true; while (this.bufferCounter > 0) { - const result = await Task.runTask(this.task, { x: x }); - this.bufferCounter--; - this.task.cycleCounter.informOfCycle(result); + try { + const result = await Task.runTask(this.task, { x: x }); + this.bufferCounter--; + this.task.cycleCounter.informOfCycle(result); + } catch (err) { + logger.log('error', `BufferRunner: task "${this.task.name || 'unnamed'}" failed: ${err instanceof Error ? err.message : String(err)}`); + this.bufferCounter--; + if (this.task.catchErrors) { + this.task.cycleCounter.informOfCycle(undefined); + } else { + this.task.cycleCounter.informOfCycleError(err instanceof Error ? err : new Error(String(err))); + } + } } this.task.running = false; } diff --git a/ts/taskbuffer.classes.cyclecounter.ts b/ts/taskbuffer.classes.cyclecounter.ts index be3d951..e8e0ecd 100644 --- a/ts/taskbuffer.classes.cyclecounter.ts +++ b/ts/taskbuffer.classes.cyclecounter.ts @@ -33,4 +33,16 @@ export class CycleCounter { }); this.cycleObjectArray = newCycleObjectArray; } + public informOfCycleError(err: Error) { + const newCycleObjectArray: ICycleObject[] = []; + this.cycleObjectArray.forEach((cycleObjectArg) => { + cycleObjectArg.cycleCounter--; + if (cycleObjectArg.cycleCounter <= 0) { + cycleObjectArg.deferred.reject(err); + } else { + newCycleObjectArray.push(cycleObjectArg); + } + }); + this.cycleObjectArray = newCycleObjectArray; + } } diff --git a/ts/taskbuffer.classes.task.ts b/ts/taskbuffer.classes.task.ts index 63ddf31..0a6f5af 100644 --- a/ts/taskbuffer.classes.task.ts +++ b/ts/taskbuffer.classes.task.ts @@ -87,24 +87,37 @@ export class Task { - taskToRun.running = false; - - // Complete all steps when task finishes - taskToRun.completeAllSteps(); + done.promise + .then(async () => { + taskToRun.running = false; - // When the task has finished running, resolve the finished promise - taskToRun.resolveFinished(); + // Complete all steps when task finishes + taskToRun.completeAllSteps(); - // Create a new finished promise for the next run - taskToRun.finished = new Promise((resolve) => { - taskToRun.resolveFinished = resolve; + // When the task has finished running, resolve the finished promise + taskToRun.resolveFinished(); + + // Create a new finished promise for the next run + taskToRun.finished = new Promise((resolve) => { + taskToRun.resolveFinished = resolve; + }); + }) + .catch((err) => { + taskToRun.running = false; + + // Resolve finished so blocking dependants don't hang + taskToRun.resolveFinished(); + + // Create a new finished promise for the next run + taskToRun.finished = new Promise((resolve) => { + taskToRun.resolveFinished = resolve; + }); }); - }); const options = { ...{ x: undefined, touchedTasksArray: [] }, @@ -133,7 +146,13 @@ export class Task { @@ -155,10 +174,18 @@ export class Task { - console.log(err); + done.reject(err); }); localDeferred.resolve(); - return await done.promise; + + try { + return await done.promise; + } catch (err) { + if (taskToRun.catchErrors) { + return undefined; + } + throw err; + } }; public name: string; @@ -187,10 +214,19 @@ export class Task; public setupValue: T; @@ -210,6 +246,7 @@ export class Task; steps?: TSteps; + catchErrors?: boolean; }) { this.taskFunction = optionsArg.taskFunction; this.preTask = optionsArg.preTask; @@ -219,6 +256,7 @@ export class Task 0) { + status = 'completed'; + } else { + status = 'idle'; + } + return { name: this.name || 'unnamed', version: this.version, - status: this.running ? 'running' : 'idle', + status, steps: this.getStepsMetadata(), currentStep: this.currentStepName, currentProgress: this.getProgress(), @@ -318,6 +367,8 @@ export class Task { if (typeof this.taskArray[taskCounter] !== 'undefined') { - console.log( - this.name + ' running: Task' + this.taskArray[taskCounter].name, - ); + logger.log('info', `${this.name} running: Task ${this.taskArray[taskCounter].name}`); this.taskArray[taskCounter].trigger(x).then((x) => { logger.log('info', this.taskArray[taskCounter].name); taskCounter++; iterateTasks(x); + }).catch((err) => { + const chainError = new Error( + `Taskchain "${this.name}": task "${this.taskArray[taskCounter].name || 'unnamed'}" (index ${taskCounter}) failed: ${err instanceof Error ? err.message : String(err)}` + ); + (chainError as any).cause = err; + done.reject(chainError); }); } else { - console.log( - 'Taskchain "' + this.name + '" completed successfully', - ); + logger.log('info', `Taskchain "${this.name}" completed successfully`); done.resolve(x); } }; @@ -53,10 +55,15 @@ export class Taskchain extends Task { addTask(taskArg: Task) { this.taskArray.push(taskArg); } - removeTask(taskArg: Task) { - // TODO: + removeTask(taskArg: Task): boolean { + const index = this.taskArray.indexOf(taskArg); + if (index === -1) { + return false; + } + this.taskArray.splice(index, 1); + return true; } - shiftTask() { - // TODO: + shiftTask(): Task | undefined { + return this.taskArray.shift(); } } diff --git a/ts/taskbuffer.classes.taskdebounced.ts b/ts/taskbuffer.classes.taskdebounced.ts index ddb64a2..cfc3e37 100644 --- a/ts/taskbuffer.classes.taskdebounced.ts +++ b/ts/taskbuffer.classes.taskdebounced.ts @@ -1,6 +1,7 @@ import * as plugins from './taskbuffer.plugins.js'; import { Task, type ITaskFunction } from './taskbuffer.classes.task.js'; +import { logger } from './taskbuffer.logging.js'; export class TaskDebounced extends Task { private _debouncedTaskFunction: ITaskFunction; @@ -22,8 +23,17 @@ export class TaskDebounced extends Task { .pipe( plugins.smartrx.rxjs.ops.debounceTime(optionsArg.debounceTimeInMillis), ) - .subscribe((x) => { - this.taskFunction(x); + .subscribe({ + next: async (x) => { + try { + await this.taskFunction(x); + } catch (err) { + logger.log('error', `TaskDebounced "${this.name || 'unnamed'}" failed: ${err instanceof Error ? err.message : String(err)}`); + } + }, + error: (err) => { + logger.log('error', `TaskDebounced "${this.name || 'unnamed'}" observable error: ${err instanceof Error ? err.message : String(err)}`); + }, }); } } diff --git a/ts/taskbuffer.classes.taskmanager.ts b/ts/taskbuffer.classes.taskmanager.ts index dfda59b..0b12053 100644 --- a/ts/taskbuffer.classes.taskmanager.ts +++ b/ts/taskbuffer.classes.taskmanager.ts @@ -5,6 +5,7 @@ import { type IDistributedTaskRequestResult, } from './taskbuffer.classes.distributedcoordinator.js'; import type { ITaskMetadata, ITaskExecutionReport, IScheduledTaskInfo } from './taskbuffer.interfaces.js'; +import { logger } from './taskbuffer.logging.js'; export interface ICronJob { cronString: string; @@ -75,31 +76,35 @@ export class TaskManager { triggerTime, ); if (!announcementResult.shouldTrigger) { - console.log('Distributed coordinator result: NOT EXECUTING'); + logger.log('info', 'Distributed coordinator result: NOT EXECUTING'); return; } else { - console.log('Distributed coordinator result: CHOSEN AND EXECUTING'); + logger.log('info', 'Distributed coordinator result: CHOSEN AND EXECUTING'); } } - await task.trigger(); + try { + await task.trigger(); + } catch (err) { + logger.log('error', `TaskManager: scheduled task "${task.name || 'unnamed'}" failed: ${err instanceof Error ? err.message : String(err)}`); + } }, ); task.cronJob = cronJob; } private logTaskState(task: Task) { - console.log(`Taskbuffer schedule triggered task >>${task.name}<<`); + logger.log('info', `Taskbuffer schedule triggered task >>${task.name}<<`); const bufferState = task.buffered ? `buffered with max ${task.bufferMax} buffered calls` : `unbuffered`; - console.log(`Task >>${task.name}<< is ${bufferState}`); + logger.log('info', `Task >>${task.name}<< is ${bufferState}`); } private async performDistributedConsultation( task: Task, triggerTime: number, ): Promise { - console.log('Found a distributed coordinator, performing consultation.'); + logger.log('info', 'Found a distributed coordinator, performing consultation.'); return this.options.distributedCoordinator.fireDistributedTaskRequest({ submitterId: this.randomId, diff --git a/ts/taskbuffer.classes.taskparallel.ts b/ts/taskbuffer.classes.taskparallel.ts index 984a0fb..8282025 100644 --- a/ts/taskbuffer.classes.taskparallel.ts +++ b/ts/taskbuffer.classes.taskparallel.ts @@ -13,7 +13,9 @@ export class Taskparallel extends Task { this.taskArray.forEach(function (taskArg) { promiseArray.push(taskArg.trigger()); }); - Promise.all(promiseArray).then(done.resolve); + Promise.all(promiseArray) + .then((results) => done.resolve(results)) + .catch((err) => done.reject(err)); return done.promise; }, }, diff --git a/ts/taskbuffer.classes.taskrunner.ts b/ts/taskbuffer.classes.taskrunner.ts index f81dd53..0250664 100644 --- a/ts/taskbuffer.classes.taskrunner.ts +++ b/ts/taskbuffer.classes.taskrunner.ts @@ -1,13 +1,14 @@ import * as plugins from './taskbuffer.plugins.js'; import { Task } from './taskbuffer.classes.task.js'; +import { logger } from './taskbuffer.logging.js'; export class TaskRunner { - public maxParrallelJobs: number = 1; + public maxParallelJobs: number = 1; public status: 'stopped' | 'running' = 'stopped'; public runningTasks: plugins.lik.ObjectMap = new plugins.lik.ObjectMap(); - public qeuedTasks: Task[] = []; + public queuedTasks: Task[] = []; constructor() { this.runningTasks.eventSubject.subscribe(async (eventArg) => { @@ -16,19 +17,19 @@ export class TaskRunner { } /** - * adds a task to the qeue + * adds a task to the queue */ public addTask(taskArg: Task) { - this.qeuedTasks.push(taskArg); + this.queuedTasks.push(taskArg); this.checkExecution(); } /** * set amount of parallel tasks - * be careful, you might loose dependability of tasks + * be careful, you might lose dependability of tasks */ - public setMaxParallelJobs(maxParrallelJobsArg: number) { - this.maxParrallelJobs = maxParrallelJobsArg; + public setMaxParallelJobs(maxParallelJobsArg: number) { + this.maxParallelJobs = maxParallelJobsArg; } /** @@ -39,17 +40,21 @@ export class TaskRunner { } /** - * checks wether execution is on point + * checks whether execution is on point */ public async checkExecution() { if ( - this.runningTasks.getArray().length < this.maxParrallelJobs && + this.runningTasks.getArray().length < this.maxParallelJobs && this.status === 'running' && - this.qeuedTasks.length > 0 + this.queuedTasks.length > 0 ) { - const nextJob = this.qeuedTasks.shift(); + const nextJob = this.queuedTasks.shift(); this.runningTasks.add(nextJob); - await nextJob.trigger(); + try { + await nextJob.trigger(); + } catch (err) { + logger.log('error', `TaskRunner: task "${nextJob.name || 'unnamed'}" failed: ${err instanceof Error ? err.message : String(err)}`); + } this.runningTasks.remove(nextJob); this.checkExecution(); } diff --git a/ts/taskbuffer.interfaces.ts b/ts/taskbuffer.interfaces.ts index 8e3a3fd..f8f1542 100644 --- a/ts/taskbuffer.interfaces.ts +++ b/ts/taskbuffer.interfaces.ts @@ -15,6 +15,8 @@ export interface ITaskMetadata { buffered?: boolean; bufferMax?: number; timeout?: number; + lastError?: string; + errorCount?: number; } export interface ITaskExecutionReport { diff --git a/ts_web/00_commitinfo_data.ts b/ts_web/00_commitinfo_data.ts index de407da..9d0be6b 100644 --- a/ts_web/00_commitinfo_data.ts +++ b/ts_web/00_commitinfo_data.ts @@ -3,6 +3,6 @@ */ export const commitinfo = { name: '@push.rocks/taskbuffer', - version: '3.5.0', + version: '4.0.0', description: 'A flexible task management library supporting TypeScript, allowing for task buffering, scheduling, and execution with dependency management.' }