From cfd779a9bc165d109cdf50adc22a1de404de8727 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Thu, 21 Apr 2022 14:18:15 +0200 Subject: [PATCH 1/2] Add local task running support to TaskScheduler MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Fredrik Adelöw --- .changeset/swift-parrots-hammer.md | 5 + packages/backend-tasks/api-report.md | 4 + .../src/tasks/LocalTaskWorker.test.ts | 105 +++++++++++++ .../src/tasks/LocalTaskWorker.ts | 138 ++++++++++++++++++ .../src/tasks/PluginTaskSchedulerImpl.ts | 63 +++++--- .../backend-tasks/src/tasks/TaskWorker.ts | 25 +++- packages/backend-tasks/src/tasks/types.ts | 30 ++++ 7 files changed, 348 insertions(+), 22 deletions(-) create mode 100644 .changeset/swift-parrots-hammer.md create mode 100644 packages/backend-tasks/src/tasks/LocalTaskWorker.test.ts create mode 100644 packages/backend-tasks/src/tasks/LocalTaskWorker.ts diff --git a/.changeset/swift-parrots-hammer.md b/.changeset/swift-parrots-hammer.md new file mode 100644 index 0000000000..1c7c178bdb --- /dev/null +++ b/.changeset/swift-parrots-hammer.md @@ -0,0 +1,5 @@ +--- +'@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. diff --git a/packages/backend-tasks/api-report.md b/packages/backend-tasks/api-report.md index 9f02088bdf..e884b95a0d 100644 --- a/packages/backend-tasks/api-report.md +++ b/packages/backend-tasks/api-report.md @@ -11,7 +11,11 @@ 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; 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..dd6208df1a --- /dev/null +++ b/packages/backend-tasks/src/tasks/LocalTaskWorker.ts @@ -0,0 +1,138 @@ +/* + * 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'; + +export class LocalTaskWorker { + private abortWait: AbortController | undefined; + + constructor( + private readonly taskId: string, + private readonly fn: TaskFunction, + 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)}`, + ); + + (async () => { + try { + if (settings.initialDelayDuration) { + await sleep( + Duration.fromISO(settings.initialDelayDuration), + options?.signal, + ); + } + + while (!options?.signal?.aborted) { + const startTime = Date.now(); + await this.runOnce(settings, options?.signal); + const endTime = Date.now(); + await this.waitUntilNext( + settings, + endTime - startTime, + 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), + )}`, + ); + + this.abortWait = delegateAbortController(signal); + await sleep(Duration.fromMillis(dt), this.abortWait.signal); + this.abortWait.abort(); // cleans up resources + this.abortWait = undefined; + } +} diff --git a/packages/backend-tasks/src/tasks/PluginTaskSchedulerImpl.ts b/packages/backend-tasks/src/tasks/PluginTaskSchedulerImpl.ts index 44f46e7237..f9a4ca77bf 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,38 +25,27 @@ 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( @@ -82,6 +72,31 @@ export class PluginTaskSchedulerImpl implements PluginTaskScheduler { ); } + async scheduleLocalTask( + task: TaskScheduleDefinition & TaskInvocationDefinition, + ): Promise { + validateId(task.id); + + 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 { return { run: async task => { @@ -89,4 +104,12 @@ 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/TaskWorker.ts b/packages/backend-tasks/src/tasks/TaskWorker.ts index 02bc115ddd..0656746d37 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,7 +24,6 @@ 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 }); @@ -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..39ac2eca6e 100644 --- a/packages/backend-tasks/src/tasks/types.ts +++ b/packages/backend-tasks/src/tasks/types.ts @@ -174,6 +174,21 @@ 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. @@ -187,6 +202,21 @@ 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 { From d63133e521476415825cf558ef6e0e8f93692f9c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Fri, 22 Apr 2022 20:06:31 +0200 Subject: [PATCH 2/2] change to using a scope parameter instead MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Fredrik Adelöw --- .changeset/swift-parrots-hammer.md | 6 +- packages/backend-tasks/api-report.md | 5 +- .../src/tasks/LocalTaskWorker.ts | 28 +++-- .../src/tasks/PluginTaskSchedulerImpl.test.ts | 114 +++++++++++++++++- .../src/tasks/PluginTaskSchedulerImpl.ts | 83 ++++++------- .../src/tasks/TaskScheduler.test.ts | 2 +- .../src/tasks/TaskWorker.test.ts | 2 +- .../backend-tasks/src/tasks/TaskWorker.ts | 2 +- packages/backend-tasks/src/tasks/types.ts | 84 ++++++------- 9 files changed, 211 insertions(+), 115 deletions(-) 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 {