Merge pull request #33241 from backstage/freben/scheduler-cancel-task

Add cancelTask to SchedulerService for cancelling running tasks

.changeset/add-scheduler-cancel-task-test-utils.md

@@ -0,0 +1,5 @@
+---
+'@backstage/backend-test-utils': patch
+---
+
+Added `cancelTask` to `MockSchedulerService` and mock scheduler service factory.

.changeset/add-scheduler-cancel-task.md

@@ -0,0 +1,6 @@
+---
+'@backstage/backend-plugin-api': minor
+'@backstage/backend-defaults': patch
+---
+
+Added `cancelTask` method to the `SchedulerService` interface and implementation, allowing cancellation of currently running scheduled tasks. For global tasks, the database lock is released and a periodic liveness check aborts the running task function. For local tasks, the task's abort signal is triggered directly. A new `POST /.backstage/scheduler/v1/tasks/:id/cancel` endpoint is also available.

.github/vale/config/vocabularies/Backstage/accept.txt

@@ -246,6 +246,7 @@ Levenshtein
 lightbox
 Lightsail
 limitranges
+liveness
 LocalStack
 lockdown
 lockfile

packages/backend-defaults/src/entrypoints/scheduler/lib/LocalTaskWorker.test.ts

@@ -16,6 +16,7 @@
 
 import { LocalTaskWorker } from './LocalTaskWorker';
 import { mockServices } from '@backstage/backend-test-utils';
+import { ConflictError } from '@backstage/errors';
 import waitFor from 'wait-for-expect';
 
 jest.setTimeout(10_000);
@@ -110,6 +111,54 @@ describe('LocalTaskWorker', () => {
     controller.abort();
   });
 
+  it('can cancel a running task', async () => {
+    let receivedSignal: AbortSignal | undefined;
+    const fn = jest.fn(async (signal: AbortSignal) => {
+      receivedSignal = signal;
+      await new Promise(r => setTimeout(r, 5000));
+    });
+    const controller = new AbortController();
+
+    const worker = new LocalTaskWorker('a', fn, logger);
+    worker.start(
+      {
+        version: 2,
+        cadence: 'PT10S',
+        timeoutAfterDuration: 'PT10S',
+      },
+      { signal: controller.signal },
+    );
+
+    await waitFor(() => {
+      expect(fn).toHaveBeenCalledTimes(1);
+    });
+
+    expect(receivedSignal?.aborted).toBe(false);
+    worker.cancel();
+    expect(receivedSignal?.aborted).toBe(true);
+
+    controller.abort();
+  });
+
+  it('cannot cancel a task that is not running', async () => {
+    const fn = jest.fn();
+    const controller = new AbortController();
+
+    const worker = new LocalTaskWorker('a', fn, logger);
+    worker.start(
+      {
+        version: 2,
+        initialDelayDuration: 'PT1000S',
+        cadence: 'PT10S',
+        timeoutAfterDuration: 'PT10S',
+      },
+      { signal: controller.signal },
+    );
+
+    expect(() => worker.cancel()).toThrow(ConflictError);
+    controller.abort();
+  });
+
   it('goes through the expected states', async () => {
     const fn = jest
       .fn()

packages/backend-defaults/src/entrypoints/scheduler/lib/LocalTaskWorker.ts

@@ -29,6 +29,7 @@ import { delegateAbortController, serializeError, sleep } from './util';
  */
 export class LocalTaskWorker {
   private abortWait: AbortController | undefined;
+  private taskAbortController: AbortController | undefined;
   #taskState: Exclude<TaskApiTasksResponse['taskState'], null> = {
     status: 'idle',
   };
@@ -93,6 +94,13 @@ export class LocalTaskWorker {
     this.abortWait.abort();
   }
 
+  cancel(): void {
+    if (!this.taskAbortController) {
+      throw new ConflictError(`Task ${this.taskId} is not running`);
+    }
+    this.taskAbortController.abort();
+  }
+
   taskState(): TaskApiTasksResponse['taskState'] {
     return this.#taskState;
   }
@@ -134,10 +142,10 @@ export class LocalTaskWorker {
   ): Promise<void> {
     // Abort the task execution either if the worker is stopped, or if the
     // task timeout is hit
-    const taskAbortController = delegateAbortController(signal);
+    this.taskAbortController = delegateAbortController(signal);
     const timeoutDuration = Duration.fromISO(settings.timeoutAfterDuration);
     const timeoutHandle = setTimeout(() => {
-      taskAbortController.abort();
+      this.taskAbortController?.abort();
     }, timeoutDuration.as('milliseconds'));
 
     this.#taskState = {
@@ -152,7 +160,7 @@ export class LocalTaskWorker {
     };
 
     try {
-      await this.fn(taskAbortController.signal);
+      await this.fn(this.taskAbortController.signal);
       this.#taskState.lastRunEndedAt = DateTime.utc().toISO()!;
       this.#taskState.lastRunError = undefined;
     } catch (e) {
@@ -162,7 +170,8 @@ export class LocalTaskWorker {
 
     // release resources
     clearTimeout(timeoutHandle);
-    taskAbortController.abort();
+    this.taskAbortController.abort();
+    this.taskAbortController = undefined;
   }
 
   /**

packages/backend-defaults/src/entrypoints/scheduler/lib/PluginTaskSchedulerImpl.test.ts

@@ -399,6 +399,100 @@ describe('PluginTaskManagerImpl', () => {
     );
   });
 
+  describe('cancelTask with local scope', () => {
+    it('can cancel a running task', async () => {
+      const { manager } = await init('SQLITE_3');
+
+      const promise = createDeferred();
+
+      await manager.scheduleTask({
+        id: 'task1',
+        timeout: Duration.fromMillis(5000),
+        frequency: Duration.fromObject({ years: 1 }),
+        fn: async () => {
+          promise.resolve();
+          await new Promise(r => setTimeout(r, 20000));
+        },
+        scope: 'local',
+      });
+
+      await promise;
+      await expect(manager.cancelTask('task1')).resolves.toBeUndefined();
+    }, 60_000);
+
+    it('cannot cancel a task that is not running', async () => {
+      const { manager } = await init('SQLITE_3');
+
+      const fn = jest.fn();
+      await manager.scheduleTask({
+        id: 'task1',
+        timeout: Duration.fromMillis(5000),
+        frequency: Duration.fromObject({ years: 1 }),
+        initialDelay: Duration.fromObject({ years: 1 }),
+        fn,
+        scope: 'local',
+      });
+
+      await expect(manager.cancelTask('task1')).rejects.toThrow(ConflictError);
+    }, 60_000);
+  });
+
+  describe('cancelTask with global scope', () => {
+    it.each(databases.eachSupportedId())(
+      'can cancel a running task, %p',
+      async databaseId => {
+        const { manager } = await init(databaseId);
+
+        const promise = createDeferred();
+
+        await manager.scheduleTask({
+          id: 'task1',
+          timeout: Duration.fromMillis(5000),
+          frequency: Duration.fromObject({ years: 1 }),
+          fn: async () => {
+            promise.resolve();
+            await new Promise(r => setTimeout(r, 20000));
+          },
+          scope: 'global',
+        });
+
+        await promise;
+        await expect(manager.cancelTask('task1')).resolves.toBeUndefined();
+      },
+    );
+
+    it.each(databases.eachSupportedId())(
+      'cannot cancel a non-existent task, %p',
+      async databaseId => {
+        const { manager } = await init(databaseId);
+
+        await expect(manager.cancelTask('nonexistent')).rejects.toThrow(
+          NotFoundError,
+        );
+      },
+    );
+
+    it.each(databases.eachSupportedId())(
+      'cannot cancel a task that is not running, %p',
+      async databaseId => {
+        const { manager } = await init(databaseId);
+
+        await manager.scheduleTask({
+          id: 'task1',
+          timeout: Duration.fromMillis(5000),
+          frequency: Duration.fromObject({ years: 1 }),
+          initialDelay: Duration.fromObject({ years: 1 }),
+          fn: jest.fn(),
+          scope: 'global',
+        });
+
+        await expect(manager.cancelTask('task1')).rejects.toThrow(
+          ConflictError,
+        );
+      },
+    );
+  });
+
   describe('parseDuration', () => {
     it('should parse durations', () => {
       expect(parseDuration({ milliseconds: 5000 })).toEqual('PT5S');

packages/backend-defaults/src/entrypoints/scheduler/lib/PluginTaskSchedulerImpl.ts

@@ -107,6 +107,17 @@ export class PluginTaskSchedulerImpl implements SchedulerService {
     await TaskWorker.trigger(knex, id);
   }
 
+  async cancelTask(id: string): Promise<void> {
+    const localTask = this.localWorkersById.get(id);
+    if (localTask) {
+      localTask.cancel();
+      return;
+    }
+
+    const knex = await this.databaseFactory();
+    await TaskWorker.cancel(knex, id);
+  }
+
   async scheduleTask(
     task: SchedulerServiceTaskScheduleDefinition &
       SchedulerServiceTaskInvocationDefinition,
@@ -206,6 +217,15 @@ export class PluginTaskSchedulerImpl implements SchedulerService {
       },
     );
 
+    router.post(
+      '/.backstage/scheduler/v1/tasks/:id/cancel',
+      async (req, res) => {
+        const { id } = req.params;
+        await this.cancelTask(id);
+        res.status(200).end();
+      },
+    );
+
     return router;
   }
 

packages/backend-defaults/src/entrypoints/scheduler/lib/TaskWorker.test.ts

@@ -15,6 +15,7 @@
  */
 
 import { TestDatabases, mockServices } from '@backstage/backend-test-utils';
+import { ConflictError, NotFoundError } from '@backstage/errors';
 import { DateTime, Duration } from 'luxon';
 import waitForExpect from 'wait-for-expect';
 import { migrateBackendTasks } from '../database/migrateBackendTasks';
@@ -584,4 +585,79 @@ describe('TaskWorker', () => {
       await knex.destroy();
     },
   );
+
+  it.each(databases.eachSupportedId())(
+    'can cancel a running task, %p',
+    async databaseId => {
+      const knex = await databases.init(databaseId);
+      await migrateBackendTasks(knex);
+
+      const fn = jest.fn(async () => {});
+      const settings: TaskSettingsV2 = {
+        version: 2,
+        cadence: '* * * * * *',
+        initialDelayDuration: undefined,
+        timeoutAfterDuration: Duration.fromObject({ minutes: 1 }).toISO()!,
+      };
+
+      const worker = new TaskWorker('task1', fn, knex, logger);
+      await worker.persistTask(settings);
+      await worker.tryClaimTask('ticket', settings);
+
+      // Verify the task is running
+      let row = (await knex<DbTasksRow>(DB_TASKS_TABLE))[0];
+      expect(row.current_run_ticket).toBe('ticket');
+
+      await TaskWorker.cancel(knex, 'task1');
+
+      // Verify the task is now idle with a cancellation error recorded
+      row = (await knex<DbTasksRow>(DB_TASKS_TABLE))[0];
+      expect(row.current_run_ticket).toBeNull();
+      expect(row.current_run_started_at).toBeNull();
+      expect(row.current_run_expires_at).toBeNull();
+      expect(row.last_run_ended_at).not.toBeNull();
+      expect(row.last_run_error_json).toContain('Task was cancelled');
+
+      await knex.destroy();
+    },
+  );
+
+  it.each(databases.eachSupportedId())(
+    'cannot cancel a non-existent task, %p',
+    async databaseId => {
+      const knex = await databases.init(databaseId);
+      await migrateBackendTasks(knex);
+
+      await expect(TaskWorker.cancel(knex, 'nonexistent')).rejects.toThrow(
+        NotFoundError,
+      );
+
+      await knex.destroy();
+    },
+  );
+
+  it.each(databases.eachSupportedId())(
+    'cannot cancel a task that is not running, %p',
+    async databaseId => {
+      const knex = await databases.init(databaseId);
+      await migrateBackendTasks(knex);
+
+      const fn = jest.fn(async () => {});
+      const settings: TaskSettingsV2 = {
+        version: 2,
+        cadence: '* * * * * *',
+        initialDelayDuration: undefined,
+        timeoutAfterDuration: Duration.fromObject({ minutes: 1 }).toISO()!,
+      };
+
+      const worker = new TaskWorker('task1', fn, knex, logger);
+      await worker.persistTask(settings);
+
+      await expect(TaskWorker.cancel(knex, 'task1')).rejects.toThrow(
+        ConflictError,
+      );
+
+      await knex.destroy();
+    },
+  );
 });

packages/backend-defaults/src/entrypoints/scheduler/lib/TaskWorker.ts

@@ -152,6 +152,36 @@ export class TaskWorker {
     }
   }
 
+  static async cancel(knex: Knex, taskId: string): Promise<void> {
+    const [row] = await knex<DbTasksRow>(DB_TASKS_TABLE)
+      .where('id', '=', taskId)
+      .select('settings_json', 'current_run_ticket');
+    if (!row) {
+      throw new NotFoundError(`Task ${taskId} does not exist`);
+    }
+    if (!row.current_run_ticket) {
+      throw new ConflictError(`Task ${taskId} is not running`);
+    }
+
+    const settings = taskSettingsV2Schema.parse(JSON.parse(row.settings_json));
+    const nextRun = TaskWorker.computeNextRunStartAt(knex, settings);
+
+    const updatedRows = await knex<DbTasksRow>(DB_TASKS_TABLE)
+      .where('id', '=', taskId)
+      .where('current_run_ticket', '=', row.current_run_ticket)
+      .update({
+        next_run_start_at: nextRun,
+        current_run_ticket: knex.raw('null'),
+        current_run_started_at: knex.raw('null'),
+        current_run_expires_at: knex.raw('null'),
+        last_run_ended_at: knex.fn.now(),
+        last_run_error_json: serializeError(new Error('Task was cancelled')),
+      });
+    if (updatedRows < 1) {
+      throw new ConflictError(`Task ${taskId} is not running`);
+    }
+  }
+
   static async taskStates(
     knex: Knex,
   ): Promise<Map<string, TaskApiTasksResponse['taskState']>> {
@@ -227,11 +257,22 @@ export class TaskWorker {
     }
 
     // Abort the task execution either if the worker is stopped, or if the
-    // task timeout is hit
+    // task timeout is hit, or if the task ticket was lost (e.g. due to
+    // cancellation from another host)
     const taskAbortController = delegateAbortController(signal);
     const timeoutHandle = setTimeout(() => {
       taskAbortController.abort();
     }, Duration.fromISO(taskSettings.timeoutAfterDuration).as('milliseconds'));
+    let livenessHandle: ReturnType<typeof setTimeout> | undefined;
+    const scheduleLivenessCheck = () => {
+      livenessHandle = setTimeout(async () => {
+        await this.checkLiveness(ticket, taskAbortController);
+        if (!taskAbortController.signal.aborted) {
+          scheduleLivenessCheck();
+        }
+      }, this.workCheckFrequency.as('milliseconds'));
+    };
+    scheduleLivenessCheck();
 
     try {
       this.#workerState = {
@@ -248,6 +289,7 @@ export class TaskWorker {
         status: 'idle',
       };
       clearTimeout(timeoutHandle);
+      clearTimeout(livenessHandle);
     }
 
     await this.tryReleaseTask(ticket, taskSettings);
@@ -283,7 +325,7 @@ export class TaskWorker {
       // We make a conversion here to make typescript happy, because the luxon versions of the cron library and here may not be the same
       const timeConverted = DateTime.fromJSDate(time.toJSDate());
 
-      nextStartAt = this.nextRunAtRaw(timeConverted);
+      nextStartAt = TaskWorker.nextRunAtRaw(this.knex, timeConverted);
       startAt ||= nextStartAt;
     } else if (isManual) {
       nextStartAt = this.knex.raw('null');
@@ -334,6 +376,33 @@ export class TaskWorker {
       );
   }
 
+  /**
+   * Checks whether the current task ticket is still valid in the database.
+   * If the ticket has been cleared (e.g. by cancellation or janitor cleanup),
+   * aborts the task execution.
+   */
+  private async checkLiveness(
+    ticket: string,
+    taskAbortController: AbortController,
+  ): Promise<void> {
+    try {
+      const [row] = await this.knex<DbTasksRow>(DB_TASKS_TABLE)
+        .where('id', '=', this.taskId)
+        .select('current_run_ticket');
+
+      if (!row || row.current_run_ticket !== ticket) {
+        this.logger.info(
+          `Task ticket for "${this.taskId}" is no longer valid; aborting execution`,
+        );
+        taskAbortController.abort();
+      }
+    } catch (e) {
+      this.logger.warn(
+        `Failed to check liveness for task "${this.taskId}", ${e}`,
+      );
+    }
+  }
+
   /**
    * Check if the task is ready to run
    */
@@ -407,48 +476,49 @@ export class TaskWorker {
     return rows === 1;
   }
 
+  private static computeNextRunStartAt(
+    knex: Knex,
+    settings: TaskSettingsV2,
+  ): Knex.Raw {
+    const isManual = settings?.cadence === 'manual';
+    const isDuration = settings?.cadence.startsWith('P');
+    const isCron = !isManual && !isDuration;
+
+    if (isCron) {
+      const time = new CronTime(settings.cadence).sendAt().toUTC();
+      const timeConverted = DateTime.fromJSDate(time.toJSDate());
+      return TaskWorker.nextRunAtRaw(knex, timeConverted);
+    }
+
+    if (isManual) {
+      return knex.raw('null');
+    }
+
+    const dt = Duration.fromISO(settings.cadence).as('seconds');
+
+    if (knex.client.config.client.includes('sqlite3')) {
+      return knex.raw(`max(datetime(next_run_start_at, ?), datetime('now'))`, [
+        `+${dt} seconds`,
+      ]);
+    }
+
+    if (knex.client.config.client.includes('mysql')) {
+      return knex.raw(
+        `greatest(next_run_start_at + interval ${dt} second, now())`,
+      );
+    }
+
+    return knex.raw(
+      `greatest(next_run_start_at + interval '${dt} seconds', now())`,
+    );
+  }
+
   async tryReleaseTask(
     ticket: string,
     settings: TaskSettingsV2,
     error?: Error,
   ): Promise<boolean> {
-    const isManual = settings?.cadence === 'manual';
-    const isDuration = settings?.cadence.startsWith('P');
-    const isCron = !isManual && !isDuration;
-
-    let nextRun: Knex.Raw;
-    if (isCron) {
-      const time = new CronTime(settings.cadence).sendAt().toUTC();
-      this.logger.debug(`task: ${this.taskId} will next occur around ${time}`);
-      // We make a conversion here to make typescript happy, because the luxon versions of the cron library and here may not be the same
-      const timeConverted = DateTime.fromJSDate(time.toJSDate());
-
-      nextRun = this.nextRunAtRaw(timeConverted);
-    } else if (isManual) {
-      nextRun = this.knex.raw('null');
-    } else {
-      const dt = Duration.fromISO(settings.cadence).as('seconds');
-      this.logger.debug(
-        `task: ${this.taskId} will next occur around ${DateTime.now().plus({
-          seconds: dt,
-        })}`,
-      );
-
-      if (this.knex.client.config.client.includes('sqlite3')) {
-        nextRun = this.knex.raw(
-          `max(datetime(next_run_start_at, ?), datetime('now'))`,
-          [`+${dt} seconds`],
-        );
-      } else if (this.knex.client.config.client.includes('mysql')) {
-        nextRun = this.knex.raw(
-          `greatest(next_run_start_at + interval ${dt} second, now())`,
-        );
-      } else {
-        nextRun = this.knex.raw(
-          `greatest(next_run_start_at + interval '${dt} seconds', now())`,
-        );
-      }
-    }
+    const nextRun = TaskWorker.computeNextRunStartAt(this.knex, settings);
 
     const rows = await this.knex<DbTasksRow>(DB_TASKS_TABLE)
       .where('id', '=', this.taskId)
@@ -467,12 +537,13 @@ export class TaskWorker {
     return rows === 1;
   }
 
-  private nextRunAtRaw(time: DateTime): Knex.Raw {
-    if (this.knex.client.config.client.includes('sqlite3')) {
-      return this.knex.raw('datetime(?)', [time.toISO()]);
-    } else if (this.knex.client.config.client.includes('mysql')) {
-      return this.knex.raw(`?`, [time.toSQL({ includeOffset: false })]);
+  private static nextRunAtRaw(knex: Knex, time: DateTime): Knex.Raw {
+    if (knex.client.config.client.includes('sqlite3')) {
+      return knex.raw('datetime(?)', [time.toISO()]);
     }
-    return this.knex.raw(`?`, [time.toISO()]);
+    if (knex.client.config.client.includes('mysql')) {
+      return knex.raw(`?`, [time.toSQL({ includeOffset: false })]);
+    }
+    return knex.raw(`?`, [time.toISO()]);
   }
 }

packages/backend-plugin-api/report.api.md

@@ -643,6 +643,7 @@ export interface RootServiceFactoryOptions<
 
 // @public
 export interface SchedulerService {
+  cancelTask(id: string): Promise<void>;
   createScheduledTaskRunner(
     schedule: SchedulerServiceTaskScheduleDefinition,
   ): SchedulerServiceTaskRunner;

packages/backend-plugin-api/src/services/definitions/SchedulerService.ts

@@ -304,6 +304,16 @@ export interface SchedulerService {
    */
   triggerTask(id: string): Promise<void>;
 
+  /**
+   * Cancels a currently running task by ID, marking it as idle.
+   *
+   * If the task doesn't exist, a NotFoundError is thrown. If the task is
+   * not currently running, a ConflictError is thrown.
+   *
+   * @param id - The task ID
+   */
+  cancelTask(id: string): Promise<void>;
+
   /**
    * Schedules a task function for recurring runs.
    *

packages/backend-test-utils/src/services/MockSchedulerService.test.ts

@@ -206,6 +206,68 @@ describe('MockSchedulerService', () => {
     await expect(isDone()).resolves.toBe(true);
   });
 
+  it('should cancel a running task and allow re-triggering with a fresh signal', async () => {
+    const scheduler = new MockSchedulerService();
+    const signals: AbortSignal[] = [];
+
+    scheduler.scheduleTask({
+      ...baseOpts,
+      id: 'test',
+      fn: async signal => {
+        signals.push(signal);
+        // Simulate long-running work that respects cancellation
+        await new Promise<void>((resolve, reject) => {
+          if (signal.aborted) {
+            reject(new Error('aborted'));
+            return;
+          }
+          signal.addEventListener('abort', () => reject(new Error('aborted')));
+          setTimeout(1).then(resolve);
+        });
+      },
+    });
+
+    // First run completes normally
+    await scheduler.triggerTask('test');
+    expect(signals).toHaveLength(1);
+    expect(signals[0].aborted).toBe(false);
+
+    // Start a task that will block until cancelled
+    const blockingScheduler = new MockSchedulerService();
+    let resolveBlock: (() => void) | undefined;
+    blockingScheduler.scheduleTask({
+      ...baseOpts,
+      id: 'blocking',
+      fn: async signal => {
+        signals.push(signal);
+        await new Promise<void>((resolve, reject) => {
+          signal.addEventListener('abort', () => reject(new Error('aborted')));
+          resolveBlock = resolve;
+        });
+      },
+    });
+
+    const triggerPromise = blockingScheduler.triggerTask('blocking');
+    // Give the task fn time to start
+    await setTimeout(1);
+
+    await blockingScheduler.cancelTask('blocking');
+    await triggerPromise.catch(() => {});
+
+    expect(signals).toHaveLength(2);
+    expect(signals[1].aborted).toBe(true);
+
+    // Re-trigger should get a fresh non-aborted signal
+    resolveBlock = undefined;
+    const triggerPromise2 = blockingScheduler.triggerTask('blocking');
+    await setTimeout(1);
+    resolveBlock!();
+    await triggerPromise2;
+
+    expect(signals).toHaveLength(3);
+    expect(signals[2].aborted).toBe(false);
+  });
+
   it('should abort tasks when shutting down', async () => {
     let taskSignal: AbortSignal | undefined;
 

packages/backend-test-utils/src/services/MockSchedulerService.ts

@@ -23,6 +23,7 @@ import {
   SchedulerServiceTaskRunner,
   SchedulerServiceTaskScheduleDefinition,
 } from '@backstage/backend-plugin-api';
+import { ConflictError, NotFoundError } from '@backstage/errors';
 import { createDeferred, DeferredPromise } from '@backstage/types';
 
 export class MockSchedulerService implements SchedulerService {
@@ -95,10 +96,22 @@ export class MockSchedulerService implements SchedulerService {
     });
   }
 
+  async cancelTask(id: string): Promise<void> {
+    const task = this.#tasks.get(id);
+    if (!task) {
+      throw new NotFoundError(`Task ${id} not found`);
+    }
+    if (!this.#runningTasks.has(id)) {
+      throw new ConflictError(`Task ${id} is not running`);
+    }
+    task.abortControllers.abort();
+    task.abortControllers = new AbortController();
+  }
+
   async triggerTask(id: string): Promise<void> {
     const task = this.#tasks.get(id);
     if (!task) {
-      throw new Error(`Task ${id} not found`);
+      throw new NotFoundError(`Task ${id} not found`);
     }
     if (this.#runningTasks.has(id)) {
       return;

packages/backend-test-utils/src/services/mockServices.ts

@@ -526,6 +526,7 @@ export namespace mockServices {
       getScheduledTasks: jest.fn(),
       scheduleTask: jest.fn(),
       triggerTask: jest.fn(),
+      cancelTask: jest.fn(),
     }));
   }
 

plugins/catalog-backend-module-gitea/src/providers/GiteaEntityProvider.test.ts

@@ -34,6 +34,7 @@ describe('GiteaEntityProvider', () => {
     triggerTask: jest.fn(),
     scheduleTask: jest.fn(),
     getScheduledTasks: jest.fn(),
+    cancelTask: jest.fn(),
   };
   const mockTaskRunner = {
     run: jest.fn(),