diff --git a/packages/backend-tasks/README.md b/packages/backend-tasks/README.md index abb25079e3..e98c997af0 100644 --- a/packages/backend-tasks/README.md +++ b/packages/backend-tasks/README.md @@ -19,7 +19,7 @@ import { TaskManager } from '@backstage/backend-tasks'; const manager = TaskManager.fromConfig(rootConfig).forPlugin('my-plugin'); -const { unschedule } = await manager.scheduleTask({ +await manager.scheduleTask({ id: 'refresh-things', frequency: Duration.fromObject({ minutes: 10 }), fn: async () => { diff --git a/packages/backend-tasks/api-report.md b/packages/backend-tasks/api-report.md index 68c4358348..8f7209b9ed 100644 --- a/packages/backend-tasks/api-report.md +++ b/packages/backend-tasks/api-report.md @@ -11,9 +11,7 @@ import { Logger as Logger_2 } from 'winston'; // @public export interface PluginTaskManager { - scheduleTask(task: TaskDefinition): Promise<{ - unschedule: () => Promise; - }>; + scheduleTask(task: TaskDefinition): Promise; } // @public @@ -22,6 +20,7 @@ export interface TaskDefinition { frequency: Duration; id: string; initialDelay?: Duration; + signal?: AbortSignal_2; timeout: Duration; } diff --git a/packages/backend-tasks/migrations/20210928160613_init.js b/packages/backend-tasks/migrations/20210928160613_init.js index c5028e2e93..cd812b59e8 100644 --- a/packages/backend-tasks/migrations/20210928160613_init.js +++ b/packages/backend-tasks/migrations/20210928160613_init.js @@ -36,7 +36,7 @@ exports.up = async function up(knex) { .comment('JSON serialized object with properties for this task'); table .dateTime('next_run_start_at') - .nullable() + .notNullable() .comment('The next time that the task should be started'); table .text('current_run_ticket') diff --git a/packages/backend-tasks/src/database/tables.ts b/packages/backend-tasks/src/database/tables.ts index 4777c4e60c..63aad6e42a 100644 --- a/packages/backend-tasks/src/database/tables.ts +++ b/packages/backend-tasks/src/database/tables.ts @@ -20,7 +20,7 @@ export const DB_TASKS_TABLE = 'backstage_backend_tasks__tasks'; export type DbTasksRow = { id: string; settings_json: string; - next_run_start_at?: Date | string; + next_run_start_at: Date; current_run_ticket?: string; current_run_started_at?: Date | string; current_run_expires_at?: Date | string; diff --git a/packages/backend-tasks/src/tasks/PluginTaskManagerImpl.test.ts b/packages/backend-tasks/src/tasks/PluginTaskManagerImpl.test.ts index bbb54a2aa0..27fd1f9f3b 100644 --- a/packages/backend-tasks/src/tasks/PluginTaskManagerImpl.test.ts +++ b/packages/backend-tasks/src/tasks/PluginTaskManagerImpl.test.ts @@ -45,7 +45,7 @@ describe('PluginTaskManagerImpl', () => { const { manager } = await init(databaseId); const fn = jest.fn(); - const { unschedule } = await manager.scheduleTask({ + await manager.scheduleTask({ id: 'task1', timeout: Duration.fromMillis(5000), frequency: Duration.fromMillis(5000), @@ -55,8 +55,6 @@ describe('PluginTaskManagerImpl', () => { await waitForExpect(() => { expect(fn).toBeCalled(); }); - - await unschedule(); }, ); }); diff --git a/packages/backend-tasks/src/tasks/PluginTaskManagerImpl.ts b/packages/backend-tasks/src/tasks/PluginTaskManagerImpl.ts index fd543cf828..497a58b348 100644 --- a/packages/backend-tasks/src/tasks/PluginTaskManagerImpl.ts +++ b/packages/backend-tasks/src/tasks/PluginTaskManagerImpl.ts @@ -29,25 +29,22 @@ export class PluginTaskManagerImpl implements PluginTaskManager { private readonly logger: Logger, ) {} - async scheduleTask( - task: TaskDefinition, - ): Promise<{ unschedule: () => Promise }> { + async scheduleTask(task: TaskDefinition): Promise { validateId(task.id); const knex = await this.databaseFactory(); const worker = new TaskWorker(task.id, task.fn, knex, this.logger); - await worker.start({ - version: 1, - initialDelayDuration: task.initialDelay?.toISO(), - recurringAtMostEveryDuration: task.frequency.toISO(), - timeoutAfterDuration: task.timeout.toISO(), - }); - - return { - async unschedule() { - await worker.stop(); + await worker.start( + { + version: 1, + initialDelayDuration: task.initialDelay?.toISO(), + recurringAtMostEveryDuration: task.frequency.toISO(), + timeoutAfterDuration: task.timeout.toISO(), }, - }; + { + signal: task.signal, + }, + ); } } diff --git a/packages/backend-tasks/src/tasks/PluginTaskManagerJanitor.ts b/packages/backend-tasks/src/tasks/PluginTaskManagerJanitor.ts index ec65eadf1e..6594243243 100644 --- a/packages/backend-tasks/src/tasks/PluginTaskManagerJanitor.ts +++ b/packages/backend-tasks/src/tasks/PluginTaskManagerJanitor.ts @@ -16,7 +16,7 @@ import { Knex } from 'knex'; import { Duration } from 'luxon'; -import { AbortController, AbortSignal } from 'node-abort-controller'; +import { AbortSignal } from 'node-abort-controller'; import { Logger } from 'winston'; import { DbTasksRow, DB_TASKS_TABLE } from '../database/tables'; import { sleep } from './util'; @@ -29,8 +29,6 @@ export class PluginTaskManagerJanitor { private readonly knex: Knex; private readonly waitBetweenRuns: Duration; private readonly logger: Logger; - private readonly abortController: AbortController; - private readonly abortSignal: AbortSignal; constructor(options: { knex: Knex; @@ -40,26 +38,20 @@ export class PluginTaskManagerJanitor { this.knex = options.knex; this.waitBetweenRuns = options.waitBetweenRuns; this.logger = options.logger; - this.abortController = new AbortController(); - this.abortSignal = this.abortController.signal; } - async start() { - while (!this.abortSignal.aborted) { + async start(abortSignal?: AbortSignal) { + while (!abortSignal?.aborted) { try { await this.runOnce(); } catch (e) { this.logger.warn(`Error while performing janitorial tasks, ${e}`); } - await sleep(this.waitBetweenRuns, this.abortSignal); + await sleep(this.waitBetweenRuns, abortSignal); } } - async stop() { - this.abortController.abort(); - } - private async runOnce() { // SQLite currently (Oct 1 2021) returns a number for returning() // statements, effectively ignoring them and instead returning the outcome @@ -68,9 +60,15 @@ export class PluginTaskManagerJanitor { // https://github.com/knex/knex/issues/4370 // https://github.com/mapbox/node-sqlite3/issues/1453 + const dbNull = this.knex.raw('null'); + const tasksQuery = this.knex(DB_TASKS_TABLE) .where('current_run_expires_at', '<', this.knex.fn.now()) - .delete(); + .update({ + current_run_ticket: dbNull, + current_run_started_at: dbNull, + current_run_expires_at: dbNull, + }); if (this.knex.client.config.client === 'sqlite3') { const tasks = await tasksQuery; diff --git a/packages/backend-tasks/src/tasks/TaskManager.test.ts b/packages/backend-tasks/src/tasks/TaskManager.test.ts index c81cfc8ee9..43fe79709c 100644 --- a/packages/backend-tasks/src/tasks/TaskManager.test.ts +++ b/packages/backend-tasks/src/tasks/TaskManager.test.ts @@ -18,6 +18,7 @@ import { DatabaseManager, getVoidLogger } from '@backstage/backend-common'; import { TestDatabaseId, TestDatabases } from '@backstage/backend-test-utils'; import { Duration } from 'luxon'; import { TaskManager } from './TaskManager'; +import waitForExpect from 'wait-for-expect'; describe('TaskManager', () => { const logger = getVoidLogger(); @@ -42,14 +43,18 @@ describe('TaskManager', () => { async databaseId => { const database = await createDatabase(databaseId); const manager = new TaskManager(database, logger).forPlugin('test'); + const fn = jest.fn(); - const task = await manager.scheduleTask({ + await manager.scheduleTask({ id: 'task1', timeout: Duration.fromMillis(5000), frequency: Duration.fromMillis(5000), - fn: () => {}, + fn, + }); + + await waitForExpect(() => { + expect(fn).toBeCalled(); }); - expect(task.unschedule).toBeDefined(); }, ); }); diff --git a/packages/backend-tasks/src/tasks/TaskWorker.ts b/packages/backend-tasks/src/tasks/TaskWorker.ts index 9be1f2c1dc..991d97b956 100644 --- a/packages/backend-tasks/src/tasks/TaskWorker.ts +++ b/packages/backend-tasks/src/tasks/TaskWorker.ts @@ -16,7 +16,7 @@ import { Knex } from 'knex'; import { Duration } from 'luxon'; -import { AbortController } from 'node-abort-controller'; +import { AbortSignal } from 'node-abort-controller'; import { v4 as uuid } from 'uuid'; import { Logger } from 'winston'; import { DbTasksRow, DB_TASKS_TABLE } from '../database/tables'; @@ -35,17 +35,15 @@ export class TaskWorker { private readonly fn: TaskFunction; private readonly knex: Knex; private readonly logger: Logger; - private readonly abortController: AbortController; constructor(taskId: string, fn: TaskFunction, knex: Knex, logger: Logger) { this.taskId = taskId; this.fn = fn; this.knex = knex; this.logger = logger; - this.abortController = new AbortController(); } - async start(settings: TaskSettingsV1) { + async start(settings: TaskSettingsV1, options?: { signal?: AbortSignal }) { try { await this.persistTask(settings); } catch (e) { @@ -58,13 +56,13 @@ export class TaskWorker { (async () => { try { - while (!this.abortController.signal.aborted) { - const runResult = await this.runOnce(); + while (!options?.signal?.aborted) { + const runResult = await this.runOnce(options?.signal); if (runResult.result === 'abort') { break; } - await sleep(WORK_CHECK_FREQUENCY, this.abortController.signal); + await sleep(WORK_CHECK_FREQUENCY, options?.signal); } this.logger.info(`Task worker finished: ${this.taskId}`); } catch (e) { @@ -73,16 +71,14 @@ export class TaskWorker { })(); } - stop() { - this.abortController.abort(); - } - /** * Makes a single attempt at running the task to completion, if ready. * * @returns The outcome of the attempt */ - async runOnce(): Promise< + async runOnce( + signal?: AbortSignal, + ): Promise< | { result: 'not-ready-yet' } | { result: 'abort' } | { result: 'failed' } @@ -106,9 +102,7 @@ export class TaskWorker { // Abort the task execution either if the worker is stopped, or if the // task timeout is hit - const taskAbortController = delegateAbortController( - this.abortController.signal, - ); + const taskAbortController = delegateAbortController(signal); const timeoutHandle = setTimeout(() => { taskAbortController.abort(); }, Duration.fromISO(taskSettings.timeoutAfterDuration).as('milliseconds')); diff --git a/packages/backend-tasks/src/tasks/types.ts b/packages/backend-tasks/src/tasks/types.ts index 3be6591887..4cd33fd3e1 100644 --- a/packages/backend-tasks/src/tasks/types.ts +++ b/packages/backend-tasks/src/tasks/types.ts @@ -38,6 +38,12 @@ export interface TaskDefinition { */ fn: TaskFunction; + /** + * An abort signal that, when triggered, will stop the recurring execution of + * the task. + */ + signal?: AbortSignal; + /** * The maximum amount of time that a single task invocation can take, before * it's considered timed out and gets "released" such that a new invocation @@ -93,13 +99,8 @@ export interface PluginTaskManager { * continue from there. * * @param definition - The task definition - * @returns An `unschedule` function that can be used to stop the task - * invocations later on. This removes the task entirely from storage - * and stops its invocations across all workers. */ - scheduleTask( - task: TaskDefinition, - ): Promise<{ unschedule: () => Promise }>; + scheduleTask(task: TaskDefinition): Promise; } function isValidOptionalDurationString(d: string | undefined): boolean { diff --git a/packages/backend-tasks/src/tasks/util.ts b/packages/backend-tasks/src/tasks/util.ts index 378224ed57..0509f29363 100644 --- a/packages/backend-tasks/src/tasks/util.ts +++ b/packages/backend-tasks/src/tasks/util.ts @@ -83,22 +83,24 @@ export async function sleep( * * @param parent - The "parent" signal that can trigger the delegate */ -export function delegateAbortController(parent: AbortSignal): AbortController { +export function delegateAbortController(parent?: AbortSignal): AbortController { const delegate = new AbortController(); - if (parent.aborted) { - delegate.abort(); - } else { - const onParentAborted = () => { + if (parent) { + if (parent.aborted) { delegate.abort(); - }; + } else { + const onParentAborted = () => { + delegate.abort(); + }; - const onChildAborted = () => { - parent.removeEventListener('abort', onParentAborted); - }; + const onChildAborted = () => { + parent.removeEventListener('abort', onParentAborted); + }; - parent.addEventListener('abort', onParentAborted, { once: true }); - delegate.signal.addEventListener('abort', onChildAborted, { once: true }); + parent.addEventListener('abort', onParentAborted, { once: true }); + delegate.signal.addEventListener('abort', onChildAborted, { once: true }); + } } return delegate;