diff --git a/.changeset/swift-parrots-hammer.md b/.changeset/swift-parrots-hammer.md index 1c7c178bdb..1b26742937 100644 --- a/.changeset/swift-parrots-hammer.md +++ b/.changeset/swift-parrots-hammer.md @@ -2,4 +2,8 @@ '@backstage/backend-tasks': patch --- -The task scheduler now has two new methods: `scheduleLocalTask` and `createScheduledLocalTaskRunner`, which can be used to perform local-only recurring tasks. This can be used to replace usages of `runPeriodically` helpers. +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 e884b95a0d..0c82387cc8 100644 --- a/packages/backend-tasks/api-report.md +++ b/packages/backend-tasks/api-report.md @@ -11,11 +11,7 @@ import { Logger } from 'winston'; // @public export interface PluginTaskScheduler { - createScheduledLocalTaskRunner(schedule: TaskScheduleDefinition): TaskRunner; createScheduledTaskRunner(schedule: TaskScheduleDefinition): TaskRunner; - scheduleLocalTask( - task: TaskScheduleDefinition & TaskInvocationDefinition, - ): Promise; scheduleTask( task: TaskScheduleDefinition & TaskInvocationDefinition, ): Promise; @@ -47,6 +43,7 @@ export interface TaskScheduleDefinition { } | Duration; initialDelay?: Duration; + scope?: 'global' | 'local'; timeout: Duration; } diff --git a/packages/backend-tasks/src/tasks/LocalTaskWorker.ts b/packages/backend-tasks/src/tasks/LocalTaskWorker.ts index dd6208df1a..89f314a1ec 100644 --- a/packages/backend-tasks/src/tasks/LocalTaskWorker.ts +++ b/packages/backend-tasks/src/tasks/LocalTaskWorker.ts @@ -22,6 +22,11 @@ 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; @@ -31,10 +36,6 @@ export class LocalTaskWorker { private readonly logger: Logger, ) {} - get id(): string { - return this.taskId; - } - start(settings: TaskSettingsV2, options?: { signal?: AbortSignal }) { this.logger.info( `Task worker starting: ${this.taskId}, ${JSON.stringify(settings)}`, @@ -43,19 +44,19 @@ export class LocalTaskWorker { (async () => { try { if (settings.initialDelayDuration) { - await sleep( + await this.sleep( Duration.fromISO(settings.initialDelayDuration), options?.signal, ); } while (!options?.signal?.aborted) { - const startTime = Date.now(); + const startTime = process.hrtime(); await this.runOnce(settings, options?.signal); - const endTime = Date.now(); + const timeTaken = process.hrtime(startTime); await this.waitUntilNext( settings, - endTime - startTime, + (timeTaken[0] + timeTaken[1] / 1e9) * 1000, options?.signal, ); } @@ -130,8 +131,15 @@ export class LocalTaskWorker { )}`, ); - this.abortWait = delegateAbortController(signal); - await sleep(Duration.fromMillis(dt), this.abortWait.signal); + 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 f9a4ca77bf..6db8b9f4f5 100644 --- a/packages/backend-tasks/src/tasks/PluginTaskSchedulerImpl.ts +++ b/packages/backend-tasks/src/tasks/PluginTaskSchedulerImpl.ts @@ -52,49 +52,46 @@ export class PluginTaskSchedulerImpl implements PluginTaskScheduler { 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); - async scheduleLocalTask( - task: TaskScheduleDefinition & TaskInvocationDefinition, - ): Promise { - validateId(task.id); + 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, + }, + ); - 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); + this.localTasksById.set(task.id, worker); + } } createScheduledTaskRunner(schedule: TaskScheduleDefinition): TaskRunner { @@ -104,12 +101,4 @@ export class PluginTaskSchedulerImpl implements PluginTaskScheduler { }, }; } - - createScheduledLocalTaskRunner(schedule: TaskScheduleDefinition): TaskRunner { - return { - run: async task => { - await this.scheduleLocalTask({ ...task, ...schedule }); - }, - }; - } } 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 0656746d37..a60a0a4730 100644 --- a/packages/backend-tasks/src/tasks/TaskWorker.ts +++ b/packages/backend-tasks/src/tasks/TaskWorker.ts @@ -28,7 +28,7 @@ import { delegateAbortController, nowPlus, sleep } from './util'; 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 */ diff --git a/packages/backend-tasks/src/tasks/types.ts b/packages/backend-tasks/src/tasks/types.ts index 39ac2eca6e..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 */ @@ -174,21 +196,6 @@ export interface PluginTaskScheduler { task: TaskScheduleDefinition & TaskInvocationDefinition, ): Promise; - /** - * Schedules a task function for running only locally on this machine, without - * any form of coordination, locking or concurrency control across hosts. This - * convenience method performs both the scheduling and invocation in one go. - * - * @remarks - * - * If a task with this ID was already registered, a ConflictError is thrown. - * - * @param task - The task definition - */ - scheduleLocalTask( - task: TaskScheduleDefinition & TaskInvocationDefinition, - ): Promise; - /** * Creates a scheduled but dormant recurring task, ready to be launched at a * later time. @@ -202,21 +209,6 @@ export interface PluginTaskScheduler { * @param schedule - The task schedule */ createScheduledTaskRunner(schedule: TaskScheduleDefinition): TaskRunner; - - /** - * Creates a local (without any form of coordination, locking or concurrency - * control across hosts) scheduled but dormant recurring task, ready to be - * launched at a later time. - * - * @remarks - * - * This method is useful for pre-creating a schedule in outer code to be - * passed into an inner implementation, such that the outer code controls - * scheduling while inner code controls implementation. - * - * @param schedule - The task schedule - */ - createScheduledLocalTaskRunner(schedule: TaskScheduleDefinition): TaskRunner; } function isValidOptionalDurationString(d: string | undefined): boolean {