diff --git a/.changeset/swift-parrots-hammer.md b/.changeset/swift-parrots-hammer.md new file mode 100644 index 0000000000..1b26742937 --- /dev/null +++ b/.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. diff --git a/packages/backend-tasks/api-report.md b/packages/backend-tasks/api-report.md index 9f02088bdf..0c82387cc8 100644 --- a/packages/backend-tasks/api-report.md +++ b/packages/backend-tasks/api-report.md @@ -43,6 +43,7 @@ export interface TaskScheduleDefinition { } | Duration; initialDelay?: Duration; + scope?: 'global' | 'local'; timeout: Duration; } diff --git a/packages/backend-tasks/src/tasks/LocalTaskWorker.test.ts b/packages/backend-tasks/src/tasks/LocalTaskWorker.test.ts new file mode 100644 index 0000000000..bdbd976def --- /dev/null +++ b/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); + }); +}); diff --git a/packages/backend-tasks/src/tasks/LocalTaskWorker.ts b/packages/backend-tasks/src/tasks/LocalTaskWorker.ts new file mode 100644 index 0000000000..89f314a1ec --- /dev/null +++ b/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 { + // 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 { + this.abortWait = delegateAbortController(abortSignal); + await sleep(duration, this.abortWait.signal); + this.abortWait.abort(); // cleans up resources + this.abortWait = undefined; + } +} diff --git a/packages/backend-tasks/src/tasks/PluginTaskSchedulerImpl.test.ts b/packages/backend-tasks/src/tasks/PluginTaskSchedulerImpl.test.ts index 004426211c..1204431434 100644 --- a/packages/backend-tasks/src/tasks/PluginTaskSchedulerImpl.test.ts +++ b/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', diff --git a/packages/backend-tasks/src/tasks/PluginTaskSchedulerImpl.ts b/packages/backend-tasks/src/tasks/PluginTaskSchedulerImpl.ts index 44f46e7237..6db8b9f4f5 100644 --- a/packages/backend-tasks/src/tasks/PluginTaskSchedulerImpl.ts +++ b/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(); + constructor( private readonly databaseFactory: () => Promise, private readonly logger: Logger, ) {} async triggerTask(id: string): Promise { + const localTask = this.localTasksById.get(id); + if (localTask) { + localTask.trigger(); + return; + } + const knex = await this.databaseFactory(); - - // check if task exists - const rows = await knex(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(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 { 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 { diff --git a/packages/backend-tasks/src/tasks/TaskScheduler.test.ts b/packages/backend-tasks/src/tasks/TaskScheduler.test.ts index 2a5deddb19..e59f24c277 100644 --- a/packages/backend-tasks/src/tasks/TaskScheduler.test.ts +++ b/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(); diff --git a/packages/backend-tasks/src/tasks/TaskWorker.test.ts b/packages/backend-tasks/src/tasks/TaskWorker.test.ts index 0c64d06287..14688d0f98 100644 --- a/packages/backend-tasks/src/tasks/TaskWorker.test.ts +++ b/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'; diff --git a/packages/backend-tasks/src/tasks/TaskWorker.ts b/packages/backend-tasks/src/tasks/TaskWorker.ts index 02bc115ddd..a60a0a4730 100644 --- a/packages/backend-tasks/src/tasks/TaskWorker.ts +++ b/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 { + // check if task exists + const rows = await knex(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(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' } diff --git a/packages/backend-tasks/src/tasks/types.ts b/packages/backend-tasks/src/tasks/types.ts index 1e09e52cd0..f9ada1dd17 100644 --- a/packages/backend-tasks/src/tasks/types.ts +++ b/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; /** - * 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 */