Merge pull request #11009 from backstage/freben/local-tasks

Add local task running support to TaskScheduler

.changeset/swift-parrots-hammer.md

@@ -0,0 +1,9 @@
+---
+'@backstage/backend-tasks': patch
+---
+
+Scheduled tasks now have an optional `scope` field. If unset, or having the
+value `'global'`, the old behavior with cross-worker locking is retained. If
+having the value `'local'`, there is no coordination across workers and the
+behavior is more like `setInterval`. This can be used to replace usages of
+`runPeriodically` helpers.

packages/backend-tasks/api-report.md

@@ -43,6 +43,7 @@ export interface TaskScheduleDefinition {
       }
     | Duration;
   initialDelay?: Duration;
+  scope?: 'global' | 'local';
   timeout: Duration;
 }
 

packages/backend-tasks/src/tasks/LocalTaskWorker.test.ts

@@ -0,0 +1,105 @@
+/*
+ * Copyright 2022 The Backstage Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { getVoidLogger } from '@backstage/backend-common';
+import { AbortController } from 'node-abort-controller';
+import { LocalTaskWorker } from './LocalTaskWorker';
+
+describe('LocalTaskWorker', () => {
+  const logger = getVoidLogger();
+
+  it('runs the happy path (with iso duration) and handles cancellation', async () => {
+    const fn = jest.fn();
+    const controller = new AbortController();
+
+    const worker = new LocalTaskWorker('a', fn, logger);
+    worker.start(
+      {
+        version: 2,
+        initialDelayDuration: 'PT0.2S',
+        cadence: 'PT0.2S',
+        timeoutAfterDuration: 'PT1S',
+      },
+      { signal: controller.signal },
+    );
+
+    // TODO(freben): Rewrite to fake timers - tried, but it wouldn't work
+    expect(fn).toBeCalledTimes(0);
+    await new Promise(r => setTimeout(r, 100));
+    expect(fn).toBeCalledTimes(0);
+    await new Promise(r => setTimeout(r, 200));
+    expect(fn).toBeCalledTimes(1);
+    await new Promise(r => setTimeout(r, 200));
+    expect(fn).toBeCalledTimes(2);
+    controller.abort();
+    await new Promise(r => setTimeout(r, 200));
+    expect(fn).toBeCalledTimes(2);
+  });
+
+  it('runs the happy path (with a cron expression) and handles cancellation', async () => {
+    const fn = jest.fn();
+    const controller = new AbortController();
+
+    // Await until system time is just past a second boundary (since cron is
+    // wall clock based)
+    await new Promise(r => setTimeout(r, 1000 - (Date.now() % 1000) + 10));
+
+    const worker = new LocalTaskWorker('a', fn, logger);
+    worker.start(
+      {
+        version: 2,
+        initialDelayDuration: 'PT0.2S',
+        cadence: '* * * * * *',
+        timeoutAfterDuration: 'PT1S',
+      },
+      { signal: controller.signal },
+    );
+
+    // TODO(freben): Rewrite to fake timers - tried, but it wouldn't work
+    expect(fn).toBeCalledTimes(0);
+    await new Promise(r => setTimeout(r, 100));
+    expect(fn).toBeCalledTimes(0);
+    await new Promise(r => setTimeout(r, 200));
+    expect(fn).toBeCalledTimes(1);
+    await new Promise(r => setTimeout(r, 1000));
+    expect(fn).toBeCalledTimes(2);
+    controller.abort();
+    await new Promise(r => setTimeout(r, 1000));
+    expect(fn).toBeCalledTimes(2);
+  });
+
+  it('can trigger to abort wait', async () => {
+    const fn = jest.fn();
+
+    const worker = new LocalTaskWorker('a', fn, logger);
+    worker.start({
+      version: 2,
+      initialDelayDuration: 'PT0.2S',
+      cadence: 'PT0.2S',
+      timeoutAfterDuration: 'PT1S',
+    });
+
+    // TODO(freben): Rewrite to fake timers - tried, but it wouldn't work
+    expect(fn).toBeCalledTimes(0);
+    await new Promise(r => setTimeout(r, 100));
+    expect(fn).toBeCalledTimes(0);
+    await new Promise(r => setTimeout(r, 200));
+    expect(fn).toBeCalledTimes(1);
+    worker.trigger();
+    await new Promise(r => setTimeout(r, 10));
+    expect(fn).toBeCalledTimes(2);
+  });
+});

packages/backend-tasks/src/tasks/LocalTaskWorker.ts

@@ -0,0 +1,146 @@
+/*
+ * Copyright 2022 The Backstage Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { ConflictError } from '@backstage/errors';
+import { CronTime } from 'cron';
+import { DateTime, Duration } from 'luxon';
+import { AbortController, AbortSignal } from 'node-abort-controller';
+import { Logger } from 'winston';
+import { TaskFunction, TaskSettingsV2 } from './types';
+import { delegateAbortController, sleep } from './util';
+
+/**
+ * Implements tasks that run locally without cross-host collaboration.
+ *
+ * @private
+ */
+export class LocalTaskWorker {
+  private abortWait: AbortController | undefined;
+
+  constructor(
+    private readonly taskId: string,
+    private readonly fn: TaskFunction,
+    private readonly logger: Logger,
+  ) {}
+
+  start(settings: TaskSettingsV2, options?: { signal?: AbortSignal }) {
+    this.logger.info(
+      `Task worker starting: ${this.taskId}, ${JSON.stringify(settings)}`,
+    );
+
+    (async () => {
+      try {
+        if (settings.initialDelayDuration) {
+          await this.sleep(
+            Duration.fromISO(settings.initialDelayDuration),
+            options?.signal,
+          );
+        }
+
+        while (!options?.signal?.aborted) {
+          const startTime = process.hrtime();
+          await this.runOnce(settings, options?.signal);
+          const timeTaken = process.hrtime(startTime);
+          await this.waitUntilNext(
+            settings,
+            (timeTaken[0] + timeTaken[1] / 1e9) * 1000,
+            options?.signal,
+          );
+        }
+
+        this.logger.info(`Task worker finished: ${this.taskId}`);
+      } catch (e) {
+        this.logger.warn(`Task worker failed unexpectedly, ${e}`);
+      }
+    })();
+  }
+
+  trigger(): void {
+    if (!this.abortWait) {
+      throw new ConflictError(`Task ${this.taskId} is currently running`);
+    }
+    this.abortWait.abort();
+  }
+
+  /**
+   * Makes a single attempt at running the task to completion.
+   */
+  private async runOnce(
+    settings: TaskSettingsV2,
+    signal?: AbortSignal,
+  ): Promise<void> {
+    // Abort the task execution either if the worker is stopped, or if the
+    // task timeout is hit
+    const taskAbortController = delegateAbortController(signal);
+    const timeoutHandle = setTimeout(() => {
+      taskAbortController.abort();
+    }, Duration.fromISO(settings.timeoutAfterDuration).as('milliseconds'));
+
+    try {
+      await this.fn(taskAbortController.signal);
+    } catch (e) {
+      // ignore intentionally
+    }
+
+    // release resources
+    clearTimeout(timeoutHandle);
+    taskAbortController.abort();
+  }
+
+  /**
+   * Sleeps until it's time to run the task again.
+   */
+  private async waitUntilNext(
+    settings: TaskSettingsV2,
+    lastRunMillis: number,
+    signal?: AbortSignal,
+  ) {
+    if (signal?.aborted) {
+      return;
+    }
+
+    const isCron = !settings.cadence.startsWith('P');
+    let dt: number;
+
+    if (isCron) {
+      const nextRun = +new CronTime(settings.cadence).sendAt().toDate();
+      dt = nextRun - Date.now();
+    } else {
+      dt =
+        Duration.fromISO(settings.cadence).as('milliseconds') - lastRunMillis;
+    }
+
+    dt = Math.max(dt, 0);
+
+    this.logger.debug(
+      `task: ${this.taskId} will next occur around ${DateTime.now().plus(
+        Duration.fromMillis(dt),
+      )}`,
+    );
+
+    await this.sleep(Duration.fromMillis(dt), signal);
+  }
+
+  private async sleep(
+    duration: Duration,
+    abortSignal?: AbortSignal,
+  ): Promise<void> {
+    this.abortWait = delegateAbortController(abortSignal);
+    await sleep(duration, this.abortWait.signal);
+    this.abortWait.abort(); // cleans up resources
+    this.abortWait = undefined;
+  }
+}

packages/backend-tasks/src/tasks/PluginTaskSchedulerImpl.test.ts

@@ -16,11 +16,11 @@
 
 import { getVoidLogger } from '@backstage/backend-common';
 import { TestDatabaseId, TestDatabases } from '@backstage/backend-test-utils';
+import { ConflictError, NotFoundError } from '@backstage/errors';
 import { Duration } from 'luxon';
+import { AbortSignal } from 'node-abort-controller';
 import { migrateBackendTasks } from '../database/migrateBackendTasks';
 import { PluginTaskSchedulerImpl } from './PluginTaskSchedulerImpl';
-import { ConflictError, NotFoundError } from '@backstage/errors';
-import { AbortSignal } from 'node-abort-controller';
 
 jest.useFakeTimers();
 
@@ -49,7 +49,7 @@ describe('PluginTaskManagerImpl', () => {
 
   // This is just to test the wrapper code; most of the actual tests are in
   // TaskWorker.test.ts
-  describe('scheduleTask', () => {
+  describe('scheduleTask with global scope', () => {
     it.each(databases.eachSupportedId())(
       'can run the v1 happy path, %p',
       async databaseId => {
@@ -62,6 +62,7 @@ describe('PluginTaskManagerImpl', () => {
           timeout: Duration.fromMillis(5000),
           frequency: Duration.fromMillis(5000),
           fn,
+          scope: 'global',
         });
 
         await promise;
@@ -82,6 +83,7 @@ describe('PluginTaskManagerImpl', () => {
           timeout: Duration.fromMillis(5000),
           frequency: { cron: '* * * * * *' },
           fn,
+          scope: 'global',
         });
 
         await promise;
@@ -91,7 +93,7 @@ describe('PluginTaskManagerImpl', () => {
     );
   });
 
-  describe('triggerTask', () => {
+  describe('triggerTask with global scope', () => {
     it.each(databases.eachSupportedId())(
       'can manually trigger a task, %p',
       async databaseId => {
@@ -105,6 +107,7 @@ describe('PluginTaskManagerImpl', () => {
           frequency: Duration.fromObject({ years: 1 }),
           initialDelay: Duration.fromObject({ years: 1 }),
           fn,
+          scope: 'global',
         });
 
         await manager.triggerTask('task1');
@@ -127,6 +130,7 @@ describe('PluginTaskManagerImpl', () => {
           timeout: Duration.fromMillis(5000),
           frequency: Duration.fromObject({ years: 1 }),
           fn,
+          scope: 'global',
         });
 
         await expect(() => manager.triggerTask('task2')).rejects.toThrow(
@@ -151,6 +155,7 @@ describe('PluginTaskManagerImpl', () => {
             resolve();
             await new Promise(r => setTimeout(r, 20000));
           },
+          scope: 'global',
         });
 
         await promise;
@@ -162,6 +167,106 @@ describe('PluginTaskManagerImpl', () => {
     );
   });
 
+  // This is just to test the wrapper code; most of the actual tests are in
+  // TaskWorker.test.ts
+  describe('scheduleTask with local scope', () => {
+    it('can run the v1 happy path', async () => {
+      const { manager } = await init('SQLITE_3');
+
+      const fn = jest.fn();
+      const promise = new Promise(resolve => fn.mockImplementation(resolve));
+      await manager.scheduleTask({
+        id: 'task1',
+        timeout: Duration.fromMillis(5000),
+        frequency: Duration.fromMillis(5000),
+        fn,
+        scope: 'local',
+      });
+
+      await promise;
+      expect(fn).toHaveBeenCalledWith(expect.any(AbortSignal));
+    }, 60_000);
+
+    it('can run the v2 happy path', async () => {
+      const { manager } = await init('SQLITE_3');
+
+      const fn = jest.fn();
+      const promise = new Promise(resolve => fn.mockImplementation(resolve));
+      await manager.scheduleTask({
+        id: 'task2',
+        timeout: Duration.fromMillis(5000),
+        frequency: { cron: '* * * * * *' },
+        fn,
+        scope: 'local',
+      });
+
+      await promise;
+      expect(fn).toHaveBeenCalledWith(expect.any(AbortSignal));
+    }, 60_000);
+  });
+
+  describe('triggerTask with local scope', () => {
+    it('can manually trigger a task', async () => {
+      const { manager } = await init('SQLITE_3');
+
+      const fn = jest.fn();
+      const promise = new Promise(resolve => fn.mockImplementation(resolve));
+      await manager.scheduleTask({
+        id: 'task1',
+        timeout: Duration.fromMillis(5000),
+        frequency: Duration.fromObject({ years: 1 }),
+        initialDelay: Duration.fromObject({ years: 1 }),
+        fn,
+        scope: 'local',
+      });
+
+      await manager.triggerTask('task1');
+      jest.advanceTimersByTime(5000);
+
+      await promise;
+      expect(fn).toHaveBeenCalledWith(expect.any(AbortSignal));
+    }, 60_000);
+
+    it('cant trigger a non-existent task', 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 }),
+        fn,
+        scope: 'local',
+      });
+
+      await expect(() => manager.triggerTask('task2')).rejects.toThrow(
+        NotFoundError,
+      );
+    }, 60_000);
+
+    it('cant trigger a running task', async () => {
+      const { manager } = await init('SQLITE_3');
+
+      const { promise, resolve } = defer();
+
+      await manager.scheduleTask({
+        id: 'task1',
+        timeout: Duration.fromMillis(5000),
+        frequency: Duration.fromObject({ years: 1 }),
+        fn: async () => {
+          resolve();
+          await new Promise(r => setTimeout(r, 20000));
+        },
+        scope: 'local',
+      });
+
+      await promise;
+      await expect(() => manager.triggerTask('task1')).rejects.toThrow(
+        ConflictError,
+      );
+    }, 60_000);
+  });
+
   // This is just to test the wrapper code; most of the actual tests are in
   // TaskWorker.test.ts
   describe('createScheduledTaskRunner', () => {
@@ -176,6 +281,7 @@ describe('PluginTaskManagerImpl', () => {
           .createScheduledTaskRunner({
             timeout: Duration.fromMillis(5000),
             frequency: Duration.fromMillis(5000),
+            scope: 'global',
           })
           .run({
             id: 'task1',

packages/backend-tasks/src/tasks/PluginTaskSchedulerImpl.ts

@@ -16,6 +16,7 @@
 
 import { Knex } from 'knex';
 import { Logger } from 'winston';
+import { LocalTaskWorker } from './LocalTaskWorker';
 import { TaskWorker } from './TaskWorker';
 import {
   PluginTaskScheduler,
@@ -24,62 +25,73 @@ import {
   TaskScheduleDefinition,
 } from './types';
 import { validateId } from './util';
-import { DB_TASKS_TABLE, DbTasksRow } from '../database/tables';
-import { ConflictError, NotFoundError } from '@backstage/errors';
 
 /**
  * Implements the actual task management.
  */
 export class PluginTaskSchedulerImpl implements PluginTaskScheduler {
+  private readonly localTasksById = new Map<string, LocalTaskWorker>();
+
   constructor(
     private readonly databaseFactory: () => Promise<Knex>,
     private readonly logger: Logger,
   ) {}
 
   async triggerTask(id: string): Promise<void> {
+    const localTask = this.localTasksById.get(id);
+    if (localTask) {
+      localTask.trigger();
+      return;
+    }
+
     const knex = await this.databaseFactory();
-
-    // check if task exists
-    const rows = await knex<DbTasksRow>(DB_TASKS_TABLE)
-      .select(knex.raw(1))
-      .where('id', '=', id);
-    if (rows.length !== 1) {
-      throw new NotFoundError(`Task ${id} does not exist`);
-    }
-
-    const updatedRows = await knex<DbTasksRow>(DB_TASKS_TABLE)
-      .where('id', '=', id)
-      .whereNull('current_run_ticket')
-      .update({
-        next_run_start_at: knex.fn.now(),
-      });
-    if (updatedRows < 1) {
-      throw new ConflictError(`Task ${id} is currently running`);
-    }
+    await TaskWorker.trigger(knex, id);
   }
 
   async scheduleTask(
     task: TaskScheduleDefinition & TaskInvocationDefinition,
   ): Promise<void> {
     validateId(task.id);
+    const scope = task.scope ?? 'global';
 
-    const knex = await this.databaseFactory();
-    const worker = new TaskWorker(task.id, task.fn, knex, this.logger);
+    if (scope === 'global') {
+      const knex = await this.databaseFactory();
+      const worker = new TaskWorker(task.id, task.fn, knex, this.logger);
 
-    await worker.start(
-      {
-        version: 2,
-        cadence:
-          'cron' in task.frequency
-            ? task.frequency.cron
-            : task.frequency.toISO(),
-        initialDelayDuration: task.initialDelay?.toISO(),
-        timeoutAfterDuration: task.timeout.toISO(),
-      },
-      {
-        signal: task.signal,
-      },
-    );
+      await worker.start(
+        {
+          version: 2,
+          cadence:
+            'cron' in task.frequency
+              ? task.frequency.cron
+              : task.frequency.toISO(),
+          initialDelayDuration: task.initialDelay?.toISO(),
+          timeoutAfterDuration: task.timeout.toISO(),
+        },
+        {
+          signal: task.signal,
+        },
+      );
+    } else {
+      const worker = new LocalTaskWorker(task.id, task.fn, this.logger);
+
+      worker.start(
+        {
+          version: 2,
+          cadence:
+            'cron' in task.frequency
+              ? task.frequency.cron
+              : task.frequency.toISO(),
+          initialDelayDuration: task.initialDelay?.toISO(),
+          timeoutAfterDuration: task.timeout.toISO(),
+        },
+        {
+          signal: task.signal,
+        },
+      );
+
+      this.localTasksById.set(task.id, worker);
+    }
   }
 
   createScheduledTaskRunner(schedule: TaskScheduleDefinition): TaskRunner {

packages/backend-tasks/src/tasks/TaskScheduler.test.ts

@@ -17,8 +17,8 @@
 import { DatabaseManager, getVoidLogger } from '@backstage/backend-common';
 import { TestDatabaseId, TestDatabases } from '@backstage/backend-test-utils';
 import { Duration } from 'luxon';
-import { TaskScheduler } from './TaskScheduler';
 import waitForExpect from 'wait-for-expect';
+import { TaskScheduler } from './TaskScheduler';
 
 describe('TaskScheduler', () => {
   const logger = getVoidLogger();

packages/backend-tasks/src/tasks/TaskWorker.test.ts

@@ -14,10 +14,10 @@
  * limitations under the License.
  */
 
-import { AbortController } from 'node-abort-controller';
 import { getVoidLogger } from '@backstage/backend-common';
 import { TestDatabases } from '@backstage/backend-test-utils';
 import { Duration } from 'luxon';
+import { AbortController } from 'node-abort-controller';
 import waitForExpect from 'wait-for-expect';
 import { migrateBackendTasks } from '../database/migrateBackendTasks';
 import { DbTasksRow, DB_TASKS_TABLE } from '../database/tables';

packages/backend-tasks/src/tasks/TaskWorker.ts

@@ -14,6 +14,8 @@
  * limitations under the License.
  */
 
+import { ConflictError, NotFoundError } from '@backstage/errors';
+import { CronTime } from 'cron';
 import { Knex } from 'knex';
 import { DateTime, Duration } from 'luxon';
 import { AbortSignal } from 'node-abort-controller';
@@ -22,12 +24,11 @@ import { Logger } from 'winston';
 import { DbTasksRow, DB_TASKS_TABLE } from '../database/tables';
 import { TaskFunction, TaskSettingsV2, taskSettingsV2Schema } from './types';
 import { delegateAbortController, nowPlus, sleep } from './util';
-import { CronTime } from 'cron';
 
 const DEFAULT_WORK_CHECK_FREQUENCY = Duration.fromObject({ seconds: 5 });
 
 /**
- * Performs the actual work of a task.
+ * Implements tasks that run across worker hosts, with collaborative locking.
  *
  * @private
  */
@@ -76,12 +77,32 @@ export class TaskWorker {
     })();
   }
 
+  static async trigger(knex: Knex, taskId: string): Promise<void> {
+    // check if task exists
+    const rows = await knex<DbTasksRow>(DB_TASKS_TABLE)
+      .select(knex.raw(1))
+      .where('id', '=', taskId);
+    if (rows.length !== 1) {
+      throw new NotFoundError(`Task ${taskId} does not exist`);
+    }
+
+    const updatedRows = await knex<DbTasksRow>(DB_TASKS_TABLE)
+      .where('id', '=', taskId)
+      .whereNull('current_run_ticket')
+      .update({
+        next_run_start_at: knex.fn.now(),
+      });
+    if (updatedRows < 1) {
+      throw new ConflictError(`Task ${taskId} is currently running`);
+    }
+  }
+
   /**
    * Makes a single attempt at running the task to completion, if ready.
    *
    * @returns The outcome of the attempt
    */
-  async runOnce(
+  private async runOnce(
     signal?: AbortSignal,
   ): Promise<
     | { result: 'not-ready-yet' }

packages/backend-tasks/src/tasks/types.ts

@@ -14,10 +14,10 @@
  * limitations under the License.
  */
 
+import { CronTime } from 'cron';
 import { Duration } from 'luxon';
 import { AbortSignal } from 'node-abort-controller';
 import { z } from 'zod';
-import { CronTime } from 'cron';
 
 /**
  * A function that can be called as a scheduled task.
@@ -93,14 +93,37 @@ export interface TaskScheduleDefinition {
    * will happen as soon as possible according to the cadence.
    *
    * NOTE: This is a per-worker delay. If you have a cluster of workers all
-   * collaborating on a task, then you may still see the task being processed by
-   * other long-lived workers, while any given single worker is in its initial
-   * sleep delay time e.g. after a deployment. Therefore this parameter is not
-   * useful for "globally" pausing work; its main intended use is for individual
-   * machines to get a chance to reach some equilibrium at startup before
-   * triggering heavy batch workloads.
+   * collaborating on a task that has its `scope` field set to `'global'`, then
+   * you may still see the task being processed by other long-lived workers,
+   * while any given single worker is in its initial sleep delay time e.g. after
+   * a deployment. Therefore this parameter is not useful for "globally" pausing
+   * work; its main intended use is for individual machines to get a chance to
+   * reach some equilibrium at startup before triggering heavy batch workloads.
    */
   initialDelay?: Duration;
+
+  /**
+   * Sets the scope of concurrency control / locking to apply for invocations of
+   * this task.
+   *
+   * @remarks
+   *
+   * When the scope is set to the default value `'global'`, the scheduler will
+   * attempt to ensure that only one worker machine runs the task at a time,
+   * according to the given cadence. This means that as the number of worker
+   * hosts increases, the invocation frequency of this task will not go up.
+   * Instead the load is spread randomly across hosts. This setting is useful
+   * for tasks that access shared resources, for example catalog ingestion tasks
+   * where you do not want many machines to repeatedly import the same data and
+   * trample over each other.
+   *
+   * When the scope is set to `'local'`, there is no concurrency control across
+   * hosts. Each host runs the task according to the given cadence similarly to
+   * `setInterval`, but the runtime ensures that there are no overlapping runs.
+   *
+   * @defaultValue 'global'
+   */
+  scope?: 'global' | 'local';
 }
 
 /**
@@ -149,24 +172,23 @@ export interface PluginTaskScheduler {
   /**
    * Manually triggers a task by ID.
    *
-   * If the task doesn't exist, a NotFoundError is thrown.
-   * If the task is currently running, a ConflictError is thrown.
+   * If the task doesn't exist, a NotFoundError is thrown. If the task is
+   * currently running, a ConflictError is thrown.
    *
    * @param id - The task ID
-   *
    */
   triggerTask(id: string): Promise<void>;
 
   /**
-   * Schedules a task function for coordinated exclusive invocation across
-   * workers. This convenience method performs both the scheduling and
-   * invocation in one go.
+   * Schedules a task function for recurring runs.
    *
    * @remarks
    *
-   * If the task was already scheduled since before by us or by another party,
-   * its options are just overwritten with the given options, and things
-   * continue from there.
+   * The `scope` task field controls whether to use coordinated exclusive
+   * invocation across workers, or to just coordinate within the current worker.
+   *
+   * This convenience method performs both the scheduling and invocation in one
+   * go.
    *
    * @param task - The task definition
    */