From d4f412fcd3fbef498eeebfd2ea1647da56d5ced2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Thu, 30 Sep 2021 09:35:49 +0200 Subject: [PATCH] Get most of the task worker code into place MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Fredrik Adelöw --- packages/backend-common/api-report.md | 47 +++ .../migrations/20210928160613_init.js | 89 ++++-- packages/backend-common/package.json | 6 +- .../backend-common/src/database/tables.ts | 20 +- .../backend-common/src/tasks/CancelToken.ts | 48 +++ .../src/tasks/PluginTaskManagerImpl.test.ts | 67 +++++ .../src/tasks/PluginTaskManagerImpl.ts | 132 +++++++++ .../src/tasks/TaskManager.test.ts | 46 +-- .../backend-common/src/tasks/TaskManager.ts | 126 ++------ .../src/tasks/TaskWorker.test.ts | 257 ++++++++++++++++ .../backend-common/src/tasks/TaskWorker.ts | 277 ++++++++++++++++++ packages/backend-common/src/tasks/types.ts | 32 +- packages/backend-common/src/tasks/util.ts | 19 ++ packages/backend/src/index.ts | 4 +- yarn.lock | 5 - 15 files changed, 996 insertions(+), 179 deletions(-) create mode 100644 packages/backend-common/src/tasks/CancelToken.ts create mode 100644 packages/backend-common/src/tasks/PluginTaskManagerImpl.test.ts create mode 100644 packages/backend-common/src/tasks/PluginTaskManagerImpl.ts create mode 100644 packages/backend-common/src/tasks/TaskWorker.test.ts create mode 100644 packages/backend-common/src/tasks/TaskWorker.ts diff --git a/packages/backend-common/api-report.md b/packages/backend-common/api-report.md index a19447034e..93c25bca24 100644 --- a/packages/backend-common/api-report.md +++ b/packages/backend-common/api-report.md @@ -12,6 +12,7 @@ import { BitbucketIntegration } from '@backstage/integration'; import { Config } from '@backstage/config'; import cors from 'cors'; import Docker from 'dockerode'; +import { Duration } from 'luxon'; import { ErrorRequestHandler } from 'express'; import express from 'express'; import { GithubCredentialsProvider } from '@backstage/integration'; @@ -399,6 +400,37 @@ export type PluginEndpointDiscovery = { getExternalBaseUrl(pluginId: string): Promise; }; +// @public +export interface PluginTaskManager { + // (undocumented) + acquireLock( + id: string, + options: { + timeout: Duration; + }, + ): Promise< + | { + acquired: false; + } + | { + acquired: true; + release: () => void | Promise; + } + >; + // (undocumented) + scheduleTask( + id: string, + options: { + timeout: Duration; + frequency: Duration; + initialDelay?: Duration; + }, + fn: () => Promise, + ): Promise<{ + unschedule: () => Promise; + }>; +} + // @public export type ReaderFactory = (options: { config: Config; @@ -576,6 +608,21 @@ export interface StatusCheckHandlerOptions { statusCheck?: StatusCheck; } +// @public +export class TaskManager { + constructor(databaseManager: DatabaseManager, logger: Logger_2); + // (undocumented) + forPlugin(pluginId: string): PluginTaskManager; + // (undocumented) + static fromConfig( + config: Config, + options?: { + databaseManager?: DatabaseManager; + logger?: Logger_2; + }, + ): TaskManager; +} + // @public export type UrlReader = { read(url: string): Promise; diff --git a/packages/backend-common/migrations/20210928160613_init.js b/packages/backend-common/migrations/20210928160613_init.js index 663e2f8aeb..7a7590d903 100644 --- a/packages/backend-common/migrations/20210928160613_init.js +++ b/packages/backend-common/migrations/20210928160613_init.js @@ -21,32 +21,29 @@ */ exports.up = async function up(knex) { // - // locking + // mutexes // - await knex.schema.createTable( - 'backstage_backend_common__task_locks', - table => { - table.comment('Locks used for mutual exclusion among multiple workers'); - table - .text('id') - .primary() - .notNullable() - .comment('The unique id of this particular lock'); - table - .text('acquired_ticket') - .nullable() - .comment('A unique ticket for the current lock acquiral, if any'); - table - .dateTime('acquired_at') - .nullable() - .comment('The time when the lock was acquired, if locked'); - table - .dateTime('expires_at') - .nullable() - .comment('The time when an acquired lock will time out and expire'); - table.index('id', 'task_locks_id_idx'); - }, - ); + await knex.schema.createTable('backstage_backend_common__mutexes', table => { + table.comment('Locks used for mutual exclusion among multiple workers'); + table + .text('id') + .primary() + .notNullable() + .comment('The unique ID of this particular mutex'); + table + .text('current_lock_ticket') + .nullable() + .comment('A unique ticket for the current mutex lock'); + table + .dateTime('current_lock_acquired_at') + .nullable() + .comment('The time when the mutex was locked'); + table + .dateTime('current_lock_expires_at') + .nullable() + .comment('The time when a locked mutex will time out and auto-release'); + table.index(['id'], 'backstage_backend_common__mutexes__id_idx'); + }); // // tasks // @@ -56,7 +53,28 @@ exports.up = async function up(knex) { .text('id') .primary() .notNullable() - .comment('The unique id of this particular task'); + .comment('The unique ID of this particular task'); + table + .text('settings_json') + .notNullable() + .comment('JSON serialized object with properties for this task'); + table + .dateTime('next_run_start_at') + .nullable() + .comment('The next time that the task should be started'); + table + .text('current_run_ticket') + .nullable() + .comment('A unique ticket for the current task run'); + table + .dateTime('current_run_started_at') + .nullable() + .comment('The time that the current task run started'); + table + .dateTime('current_run_expires_at') + .nullable() + .comment('The time that the current task run will time out'); + table.index(['id'], 'backstage_backend_common__tasks__id_idx'); }); }; @@ -64,8 +82,21 @@ exports.up = async function up(knex) { * @param {import('knex').Knex} knex */ exports.down = async function down(knex) { - await knex.schema.alterTable('task_locks', table => { - table.dropIndex([], 'task_locks_id_idx'); + // + // tasks + // + await knex.schema.alterTable('backstage_backend_common__tasks', table => { + table.dropIndex([], 'backstage_backend_common__tasks__id_idx'); }); - await knex.schema.dropTable('task_locks'); + await knex.schema.dropTable('backstage_backend_common__tasks'); + // + // locks + // + await knex.schema.alterTable( + 'backstage_backend_common__task_locks', + table => { + table.dropIndex([], 'backstage_backend_common__task_locks__id_idx'); + }, + ); + await knex.schema.dropTable('backstage_backend_common__task_locks'); }; diff --git a/packages/backend-common/package.json b/packages/backend-common/package.json index c8e8d69ab5..a2255926e9 100644 --- a/packages/backend-common/package.json +++ b/packages/backend-common/package.json @@ -71,7 +71,8 @@ "unzipper": "^0.10.11", "uuid": "^8.0.0", "winston": "^3.2.1", - "yn": "^4.0.0" + "yn": "^4.0.0", + "zod": "^3.9.5" }, "peerDependencies": { "pg-connection-string": "^2.3.0" @@ -107,7 +108,8 @@ "msw": "^0.35.0", "mysql2": "^2.2.5", "recursive-readdir": "^2.2.2", - "supertest": "^6.1.3" + "supertest": "^6.1.3", + "wait-for-expect": "^3.0.2" }, "files": [ "dist", diff --git a/packages/backend-common/src/database/tables.ts b/packages/backend-common/src/database/tables.ts index 8efa393bb7..e566272ea6 100644 --- a/packages/backend-common/src/database/tables.ts +++ b/packages/backend-common/src/database/tables.ts @@ -14,9 +14,21 @@ * limitations under the License. */ -export type DbTaskLocksRow = { +export const DB_MUTEXES_TABLE = 'backstage_backend_common__mutexes'; +export const DB_TASKS_TABLE = 'backstage_backend_common__tasks'; + +export type DbMutexesRow = { id: string; - acquired_ticket?: string; - acquired_at?: Date; - expires_at?: Date; + current_lock_ticket?: string; + current_lock_acquired_at?: Date | string; + current_lock_expires_at?: Date | string; +}; + +export type DbTasksRow = { + id: string; + settings_json: string; + next_run_start_at?: Date | string; + current_run_ticket?: string; + current_run_started_at?: Date | string; + current_run_expires_at?: Date | string; }; diff --git a/packages/backend-common/src/tasks/CancelToken.ts b/packages/backend-common/src/tasks/CancelToken.ts new file mode 100644 index 0000000000..9ea9802e82 --- /dev/null +++ b/packages/backend-common/src/tasks/CancelToken.ts @@ -0,0 +1,48 @@ +/* + * Copyright 2021 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. + */ + +export class CancelToken { + // @ts-ignore: is actually assigned by the Promise constructor + #cancel: () => void; + #isCancelled: boolean; + #cancelPromise: Promise; + + static create(): CancelToken { + return new CancelToken(); + } + + private constructor() { + this.#isCancelled = false; + this.#cancelPromise = new Promise(resolve => { + this.#cancel = () => { + this.#isCancelled = true; + resolve(); + }; + }); + } + + cancel(): void { + this.#cancel(); + } + + get isCancelled(): boolean { + return this.#isCancelled; + } + + get promise(): Promise { + return this.#cancelPromise; + } +} diff --git a/packages/backend-common/src/tasks/PluginTaskManagerImpl.test.ts b/packages/backend-common/src/tasks/PluginTaskManagerImpl.test.ts new file mode 100644 index 0000000000..ed5bf0911e --- /dev/null +++ b/packages/backend-common/src/tasks/PluginTaskManagerImpl.test.ts @@ -0,0 +1,67 @@ +/* + * Copyright 2021 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 { TestDatabases } from '@backstage/backend-test-utils'; +import { Duration } from 'luxon'; +import { migrateBackendCommon } from '../database/migrateBackendCommon'; +import { getVoidLogger } from '../logging'; +import { PluginTaskManagerImpl } from './PluginTaskManagerImpl'; + +describe('PluginTaskManagerImpl', () => { + const databases = TestDatabases.create({ + ids: ['POSTGRES_13', 'POSTGRES_9', 'SQLITE_3'], + }); + + describe('locking', () => { + it.each(databases.eachSupportedId())( + 'can run the happy path, %p', + async databaseId => { + const knex = await databases.init(databaseId); + await migrateBackendCommon(knex); + + const manager = new PluginTaskManagerImpl( + async () => knex, + getVoidLogger(), + ); + + const lock1 = await manager.acquireLock('lock1', { + timeout: Duration.fromMillis(5000), + }); + const lock2 = await manager.acquireLock('lock2', { + timeout: Duration.fromMillis(5000), + }); + + expect(lock1.acquired).toBe(true); + expect(lock2.acquired).toBe(true); + + await expect( + manager.acquireLock('lock1', { + timeout: Duration.fromMillis(5000), + }), + ).resolves.toEqual({ acquired: false }); + + await (lock1 as any).release(); + await (lock2 as any).release(); + + const lock1Again = await manager.acquireLock('lock1', { + timeout: Duration.fromMillis(5000), + }); + expect(lock1Again.acquired).toBe(true); + await (lock1Again as any).release(); + }, + ); + }); +}); diff --git a/packages/backend-common/src/tasks/PluginTaskManagerImpl.ts b/packages/backend-common/src/tasks/PluginTaskManagerImpl.ts new file mode 100644 index 0000000000..8b9aad4388 --- /dev/null +++ b/packages/backend-common/src/tasks/PluginTaskManagerImpl.ts @@ -0,0 +1,132 @@ +/* + * Copyright 2021 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 { Knex } from 'knex'; +import { Duration } from 'luxon'; +import { v4 as uuid } from 'uuid'; +import { Logger } from 'winston'; +import { DbMutexesRow, DB_MUTEXES_TABLE } from '../database/tables'; +import { PluginTaskManagerJanitor } from './PluginTaskManagerJanitor'; +import { TaskWorker } from './TaskWorker'; +import { PluginTaskManager } from './types'; +import { nowPlus, validateId } from './util'; + +/** + * Implements the actual task management. + */ +export class PluginTaskManagerImpl implements PluginTaskManager { + private janitor: PluginTaskManagerJanitor | undefined; + + constructor( + private readonly databaseFactory: () => Promise, + private readonly logger: Logger, + ) {} + + async acquireLock( + id: string, + options: { + timeout: Duration; + }, + ): Promise< + | { acquired: false } + | { acquired: true; release: () => void | Promise } + > { + validateId(id); + + const knex = await this.databaseFactory(); + await this.ensureJanitor(knex); + + const ticket = uuid(); + + async function release() { + try { + await knex(DB_MUTEXES_TABLE) + .where('id', '=', id) + .where('current_lock_ticket', '=', ticket) + .delete(); + } catch { + // fail silently + } + } + + // First try to overwrite an existing lock, that has timed out + const stolen = await knex(DB_MUTEXES_TABLE) + .where('id', '=', id) + .whereNotNull('current_lock_ticket') + .where('current_lock_expires_at', '<', knex.fn.now()) + .update({ + current_lock_ticket: ticket, + current_lock_acquired_at: knex.fn.now(), + current_lock_expires_at: nowPlus(options.timeout, knex), + }); + + if (stolen) { + return { acquired: true, release }; + } + + try { + await knex(DB_MUTEXES_TABLE).insert({ + id, + current_lock_ticket: ticket, + current_lock_acquired_at: knex.fn.now(), + current_lock_expires_at: nowPlus(options.timeout, knex), + }); + return { acquired: true, release }; + } catch { + return { acquired: false }; + } + } + + async scheduleTask( + id: string, + options: { + timeout?: Duration; + frequency?: Duration; + initialDelay?: Duration; + }, + fn: () => void | Promise, + ): Promise<{ unschedule: () => Promise }> { + validateId(id); + + const knex = await this.databaseFactory(); + await this.ensureJanitor(knex); + + const task = new TaskWorker(id, fn, knex, this.logger); + await task.start({ + version: 1, + initialDelayDuration: options.initialDelay?.toISO(), + recurringAtMostEveryDuration: options.frequency?.toISO(), + timeoutAfterDuration: options.timeout?.toISO(), + }); + + return { + async unschedule() { + await task.stop(); + }, + }; + } + + private async ensureJanitor(knex: Knex) { + if (!this.janitor) { + this.janitor = new PluginTaskManagerJanitor({ + knex, + waitBetweenRuns: Duration.fromObject({ minutes: 1 }), + logger: this.logger, + }); + this.janitor.start(); + } + } +} diff --git a/packages/backend-common/src/tasks/TaskManager.test.ts b/packages/backend-common/src/tasks/TaskManager.test.ts index da9adb70f6..e6a34e58c1 100644 --- a/packages/backend-common/src/tasks/TaskManager.test.ts +++ b/packages/backend-common/src/tasks/TaskManager.test.ts @@ -17,9 +17,11 @@ import { TestDatabaseId, TestDatabases } from '@backstage/backend-test-utils'; import { Duration } from 'luxon'; import { DatabaseManager } from '../database'; +import { getVoidLogger } from '../logging'; import { TaskManager } from './TaskManager'; describe('TaskManager', () => { + const logger = getVoidLogger(); const databases = TestDatabases.create({ ids: ['POSTGRES_13', 'POSTGRES_9', 'SQLITE_3'], }); @@ -36,38 +38,16 @@ describe('TaskManager', () => { return databaseManager as DatabaseManager; } - describe('locking', () => { - it.each(databases.eachSupportedId())( - 'can run the happy path, %p', - async databaseId => { - const database = await createDatabase(databaseId); - const manager = new TaskManager(database).forPlugin('test'); + it.each(databases.eachSupportedId())( + 'can return a working plugin impl, %p', + async databaseId => { + const database = await createDatabase(databaseId); + const manager = new TaskManager(database, logger).forPlugin('test'); - const lock1 = await manager.acquireLock('lock1', { - timeout: Duration.fromMillis(5000), - }); - const lock2 = await manager.acquireLock('lock2', { - timeout: Duration.fromMillis(5000), - }); - - expect(lock1.acquired).toBe(true); - expect(lock2.acquired).toBe(true); - - await expect( - manager.acquireLock('lock1', { - timeout: Duration.fromMillis(5000), - }), - ).resolves.toEqual({ acquired: false }); - - await (lock1 as any).release(); - await (lock2 as any).release(); - - const lock1Again = await manager.acquireLock('lock1', { - timeout: Duration.fromMillis(5000), - }); - expect(lock1Again.acquired).toBe(true); - await (lock1Again as any).release(); - }, - ); - }); + const lock = await manager.acquireLock('lock1', { + timeout: Duration.fromMillis(5000), + }); + expect(lock.acquired).toBe(true); + }, + ); }); diff --git a/packages/backend-common/src/tasks/TaskManager.ts b/packages/backend-common/src/tasks/TaskManager.ts index 29741e1594..edb8e46609 100644 --- a/packages/backend-common/src/tasks/TaskManager.ts +++ b/packages/backend-common/src/tasks/TaskManager.ts @@ -15,104 +15,13 @@ */ import { Config } from '@backstage/config'; -import { Knex } from 'knex'; import { memoize } from 'lodash'; -import { Duration } from 'luxon'; -import { v4 as uuid } from 'uuid'; +import { Logger } from 'winston'; import { DatabaseManager } from '../database'; import { migrateBackendCommon } from '../database/migrateBackendCommon'; -import { DbTaskLocksRow } from '../database/tables'; +import { getRootLogger } from '../logging'; +import { PluginTaskManagerImpl } from './PluginTaskManagerImpl'; import { PluginTaskManager } from './types'; -import { validateId } from './util'; - -export class PluginTaskManagerImpl implements PluginTaskManager { - constructor( - private readonly pluginId: string, - private readonly databaseFactory: () => Promise, - ) {} - - async acquireLock( - idWithoutPrefix: string, - options: { - timeout: Duration; - }, - ): Promise< - | { acquired: false } - | { acquired: true; release: () => void | Promise } - > { - validateId(idWithoutPrefix); - - const knex = await this.databaseFactory(); - const id = `plugin:${this.pluginId}:${idWithoutPrefix}`; - const ticket = uuid(); - const timeout = options.timeout.as('seconds'); - - const release = async () => { - try { - await knex('backstage_backend_common__task_locks') - .where('id', '=', id) - .where('acquired_ticket', '=', ticket) - .delete(); - } catch { - // fail silently - } - }; - - // First try to overwrite an existing lock, that has timed out - const stolen = await knex( - 'backstage_backend_common__task_locks', - ) - .where('id', '=', id) - .whereNotNull('acquired_ticket') - .where('expires_at', '<', knex.fn.now()) - .update({ - acquired_ticket: ticket, - acquired_at: knex.fn.now(), - expires_at: - knex.client.config.client === 'sqlite3' - ? knex.raw(`datetime('now', ?)`, [`${timeout} seconds`]) - : knex.raw(`now() + interval '${timeout} seconds'`), - }); - - if (stolen) { - return { acquired: true, release }; - } - - try { - await knex('backstage_backend_common__task_locks').insert( - { - id, - acquired_ticket: ticket, - acquired_at: knex.fn.now(), - expires_at: - knex.client.config.client === 'sqlite3' - ? knex.raw(`datetime('now', ?)`, [`${timeout} seconds`]) - : knex.raw(`now() + interval '${timeout} seconds'`), - }, - ); - return { acquired: true, release }; - } catch { - return { acquired: false }; - } - } - - async scheduleTask( - idWithoutPrefix: string, - options: { - timeout: Duration; - frequency: Duration; - initialDelay?: Duration; - }, - fn: () => Promise, - ): Promise<{ release: () => Promise }> { - validateId(idWithoutPrefix); - - const knex = await this.databaseFactory(); - const id = `plugin:${this.pluginId}:${idWithoutPrefix}`; - - return {}; - } -} /** * Deals with management and locking related to distributed tasks. @@ -122,23 +31,34 @@ export class PluginTaskManagerImpl implements PluginTaskManager { export class TaskManager { static fromConfig( config: Config, - options?: { databaseManager?: DatabaseManager }, + options?: { + databaseManager?: DatabaseManager; + logger?: Logger; + }, ): TaskManager { const databaseManager = options?.databaseManager ?? DatabaseManager.fromConfig(config); - return new TaskManager(databaseManager); + const logger = (options?.logger || getRootLogger()).child({ + type: 'taskManager', + }); + return new TaskManager(databaseManager, logger); } - constructor(private readonly databaseManager: DatabaseManager) {} + constructor( + private readonly databaseManager: DatabaseManager, + private readonly logger: Logger, + ) {} forPlugin(pluginId: string): PluginTaskManager { + const databaseFactory = memoize(async () => { + const knex = await this.databaseManager.forPlugin(pluginId).getClient(); + await migrateBackendCommon(knex); + return knex; + }); + return new PluginTaskManagerImpl( - pluginId, - memoize(async () => { - const knex = await this.databaseManager.forPlugin(pluginId).getClient(); - await migrateBackendCommon(knex); - return knex; - }), + databaseFactory, + this.logger.child({ plugin: pluginId }), ); } } diff --git a/packages/backend-common/src/tasks/TaskWorker.test.ts b/packages/backend-common/src/tasks/TaskWorker.test.ts new file mode 100644 index 0000000000..ca0d624560 --- /dev/null +++ b/packages/backend-common/src/tasks/TaskWorker.test.ts @@ -0,0 +1,257 @@ +/* + * Copyright 2021 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 { TestDatabases } from '@backstage/backend-test-utils'; +import { Duration } from 'luxon'; +import waitForExpect from 'wait-for-expect'; +import { migrateBackendCommon } from '../database/migrateBackendCommon'; +import { DbTasksRow, DB_TASKS_TABLE } from '../database/tables'; +import { getVoidLogger } from '../logging'; +import { TaskWorker } from './TaskWorker'; +import { TaskSettingsV1 } from './types'; + +describe('TaskWorker', () => { + const logger = getVoidLogger(); + const databases = TestDatabases.create({ + ids: ['POSTGRES_13', 'POSTGRES_9', 'SQLITE_3'], + }); + + beforeEach(() => { + jest.resetAllMocks(); + }); + + it.each(databases.eachSupportedId())( + 'can run a single task to completion, %p', + async databaseId => { + const knex = await databases.init(databaseId); + await migrateBackendCommon(knex); + + const fn = jest.fn( + async () => new Promise(resolve => setTimeout(resolve, 50)), + ); + const settings: TaskSettingsV1 = { + version: 1, + }; + + const worker = new TaskWorker('task1', fn, knex, logger); + await worker.start(settings); + + waitForExpect(() => { + expect(fn).toBeCalledTimes(1); + }); + }, + 60_000, + ); + + it.each(databases.eachSupportedId())( + 'goes through the expected states for a single run, %p', + async databaseId => { + const knex = await databases.init(databaseId); + await migrateBackendCommon(knex); + + const fn = jest.fn( + async () => new Promise(resolve => setTimeout(resolve, 50)), + ); + const settings: TaskSettingsV1 = { + version: 1, + initialDelayDuration: Duration.fromObject({ seconds: 1 }).toISO(), + recurringAtMostEveryDuration: undefined, + timeoutAfterDuration: undefined, + }; + + const worker = new TaskWorker('task1', fn, knex, logger); + await worker.persistTask(settings); + + let row = (await knex(DB_TASKS_TABLE))[0]; + expect(row).toEqual( + expect.objectContaining({ + id: 'task1', + current_run_ticket: null, + current_run_started_at: null, + current_run_expires_at: null, + }), + ); + expect(JSON.parse(row.settings_json)).toEqual({ + version: 1, + initialDelayDuration: 'PT1S', + }); + + await expect(worker.findReadyTask()).resolves.toEqual({ + result: 'not ready yet', + }); + + waitForExpect(async () => { + await expect(worker.findReadyTask()).resolves.toEqual({ + result: 'ready', + }); + }); + + row = (await knex(DB_TASKS_TABLE))[0]; + expect(row).toEqual( + expect.objectContaining({ + id: 'task1', + current_run_ticket: null, + current_run_started_at: null, + current_run_expires_at: null, + }), + ); + + await expect(worker.tryClaimTask('ticket', settings)).resolves.toBe(true); + + row = (await knex(DB_TASKS_TABLE))[0]; + expect(row).toEqual( + expect.objectContaining({ + id: 'task1', + current_run_ticket: 'ticket', + current_run_started_at: expect.anything(), + current_run_expires_at: null, + }), + ); + + await expect(worker.tryReleaseTask('ticket', settings)).resolves.toBe( + true, + ); + + row = (await knex(DB_TASKS_TABLE))[0]; + expect(row).toBeUndefined(); + }, + 60_000, + ); + + it.each(databases.eachSupportedId())( + 'runs tasks more than once even when the task throws, %p', + async databaseId => { + const knex = await databases.init(databaseId); + await migrateBackendCommon(knex); + + const fn = jest.fn().mockRejectedValue(new Error('failed')); + const settings: TaskSettingsV1 = { + version: 1, + initialDelayDuration: undefined, + recurringAtMostEveryDuration: Duration.fromMillis(0).toISO(), + timeoutAfterDuration: undefined, + }; + + const worker = new TaskWorker('task1', fn, knex, logger); + worker.start(settings); + + waitForExpect(() => { + expect(fn).toBeCalledTimes(3); + }); + }, + 60_000, + ); + + it.each(databases.eachSupportedId())( + 'does not clobber ticket lock when stolen, %p', + async databaseId => { + const knex = await databases.init(databaseId); + await migrateBackendCommon(knex); + + const fn = jest.fn( + async () => new Promise(resolve => setTimeout(resolve, 50)), + ); + const settings: TaskSettingsV1 = { + version: 1, + recurringAtMostEveryDuration: Duration.fromMillis(0).toISO(), + }; + + const worker = new TaskWorker('task1', fn, knex, logger); + await worker.persistTask(settings); + await expect(worker.findReadyTask()).resolves.toEqual({ + result: 'ready', + settings, + }); + await expect(worker.tryClaimTask('ticket', settings)).resolves.toBe(true); + + let row = (await knex(DB_TASKS_TABLE))[0]; + expect(row).toEqual( + expect.objectContaining({ + id: 'task1', + current_run_ticket: 'ticket', + current_run_started_at: expect.anything(), + current_run_expires_at: null, + }), + ); + + await knex(DB_TASKS_TABLE) + .where('id', '=', 'task1') + .update({ current_run_ticket: 'stolen' }); + + await expect(worker.tryReleaseTask('ticket', settings)).resolves.toBe( + false, + ); + + row = (await knex(DB_TASKS_TABLE))[0]; + expect(row).toEqual( + expect.objectContaining({ + id: 'task1', + current_run_ticket: 'stolen', + current_run_started_at: expect.anything(), + current_run_expires_at: null, + }), + ); + }, + 60_000, + ); + + it.each(databases.eachSupportedId())( + 'gracefully handles a disappeared task row, %p', + async databaseId => { + const knex = await databases.init(databaseId); + await migrateBackendCommon(knex); + + const fn = jest.fn(async () => {}); + const settings: TaskSettingsV1 = { + version: 1, + recurringAtMostEveryDuration: Duration.fromMillis(0).toISO(), + }; + + const worker1 = new TaskWorker('task1', fn, knex, logger); + await worker1.persistTask(settings); + await knex(DB_TASKS_TABLE).where('id', '=', 'task1').delete(); + await expect(worker1.findReadyTask()).resolves.toEqual({ + result: 'abort', + }); + + const worker2 = new TaskWorker('task2', fn, knex, logger); + await worker2.persistTask(settings); + await expect(worker2.findReadyTask()).resolves.toEqual({ + result: 'ready', + settings, + }); + await knex(DB_TASKS_TABLE).where('id', '=', 'task2').delete(); + await expect(worker2.tryClaimTask('ticket', settings)).resolves.toBe( + false, + ); + + const worker3 = new TaskWorker('task3', fn, knex, logger); + await worker3.persistTask(settings); + await expect(worker3.findReadyTask()).resolves.toEqual({ + result: 'ready', + settings, + }); + await expect(worker3.tryClaimTask('ticket', settings)).resolves.toBe( + true, + ); + await knex(DB_TASKS_TABLE).where('id', '=', 'task3').delete(); + await expect(worker3.tryReleaseTask('ticket', settings)).resolves.toBe( + false, + ); + }, + 60_000, + ); +}); diff --git a/packages/backend-common/src/tasks/TaskWorker.ts b/packages/backend-common/src/tasks/TaskWorker.ts new file mode 100644 index 0000000000..3fd1f30716 --- /dev/null +++ b/packages/backend-common/src/tasks/TaskWorker.ts @@ -0,0 +1,277 @@ +/* + * Copyright 2021 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 { Knex } from 'knex'; +import { Duration } from 'luxon'; +import { v4 as uuid } from 'uuid'; +import { Logger } from 'winston'; +import { DbTasksRow, DB_TASKS_TABLE } from '../database/tables'; +import { CancelToken } from './CancelToken'; +import { TaskSettingsV1, taskSettingsV1Schema } from './types'; +import { nowPlus } from './util'; + +const WORK_CHECK_FREQUENCY = Duration.fromObject({ seconds: 5 }); + +/** + * Performs the actual work of a task. + * + * @private + */ +export class TaskWorker { + private readonly taskId: string; + private readonly fn: () => void | Promise; + private readonly knex: Knex; + private readonly logger: Logger; + private readonly cancelToken: CancelToken; + + constructor( + taskId: string, + fn: () => void | Promise, + knex: Knex, + logger: Logger, + ) { + this.taskId = taskId; + this.fn = fn; + this.knex = knex; + this.logger = logger; + this.cancelToken = CancelToken.create(); + } + + async start(settings: TaskSettingsV1) { + try { + await this.persistTask(settings); + } catch (e) { + throw new Error(`Failed to persist task, ${e}`); + } + + this.logger.info( + `Task worker starting: ${this.taskId}, ${JSON.stringify(settings)}`, + ); + + (async () => { + try { + while (!this.cancelToken.isCancelled) { + const runResult = await this.runOnce(); + if (runResult.result === 'abort') { + break; + } + if (!settings.recurringAtMostEveryDuration) { + break; + } + + await this.sleep(WORK_CHECK_FREQUENCY); + } + this.logger.info(`Task worker finished: ${this.taskId}`); + } catch (e) { + this.logger.warn(`Task worker failed unexpectedly, ${e}`); + } + })(); + } + + stop() { + this.cancelToken.cancel(); + } + + /** + * Makes a single attempt at running the task to completion, if ready. + * + * @returns The outcome of the attempt + */ + async runOnce(): Promise< + | { result: 'not ready yet' } + | { result: 'abort' } + | { result: 'failed' } + | { result: 'completed' } + > { + const findResult = await this.findReadyTask(); + if ( + findResult.result === 'not ready yet' || + findResult.result === 'abort' + ) { + return findResult; + } + + const taskSettings = findResult.settings; + const ticket = uuid(); + + const claimed = await this.tryClaimTask(ticket, taskSettings); + if (!claimed) { + return { result: 'not ready yet' }; + } + + try { + await this.fn(); + } catch (e) { + await this.tryReleaseTask(ticket, taskSettings); + return { result: 'failed' }; + } + + await this.tryReleaseTask(ticket, taskSettings); + return { result: 'completed' }; + } + + /** + * Sleep for the given duration, but abort sooner if the cancel token + * triggers. + * + * @param duration - The amount of time to sleep, at most + */ + private async sleep(duration: Duration): Promise { + await Promise.race([ + new Promise(resolve => setTimeout(resolve, duration.as('milliseconds'))), + this.cancelToken.promise, + ]); + } + + /** + * Perform the initial store of the task info + */ + async persistTask(settings: TaskSettingsV1) { + // Perform an initial parse to ensure that we will definitely be able to + // read it back again. + taskSettingsV1Schema.parse(settings); + + const settingsJson = JSON.stringify(settings); + const startAt = settings.initialDelayDuration + ? nowPlus(Duration.fromISO(settings.initialDelayDuration), this.knex) + : this.knex.fn.now(); + + // It's OK if the task already exists; if it does, just replace its + // settings with the new value and start the loop as usual. + await this.knex(DB_TASKS_TABLE) + .insert({ + id: this.taskId, + settings_json: settingsJson, + next_run_start_at: startAt, + }) + .onConflict('id') + .merge(['settings_json']); + } + + /** + * Check if the task is ready to run + */ + async findReadyTask(): Promise< + | { result: 'not ready yet' } + | { result: 'abort' } + | { result: 'ready'; settings: TaskSettingsV1 } + > { + const [row] = await this.knex(DB_TASKS_TABLE) + .where('id', '=', this.taskId) + .select({ + settingsJson: 'settings_json', + ready: this.knex.raw( + ` + CASE + WHEN next_run_start_at <= ? AND current_run_ticket IS NULL THEN TRUE + ELSE FALSE + END`, + [this.knex.fn.now()], + ), + }); + + if (!row) { + this.logger.info( + 'No longer able to find task; aborting and assuming that it has been unregistered or expired', + ); + return { result: 'abort' }; + } else if (!row.ready) { + return { result: 'not ready yet' }; + } + + try { + const settings = taskSettingsV1Schema.parse(JSON.parse(row.settingsJson)); + return { result: 'ready', settings }; + } catch (e) { + this.logger.info( + 'No longer able to parse task settings; aborting and assuming that a ' + + 'newer version of the task has been issued and being handled by ' + + `other workers, ${e}`, + ); + return { result: 'abort' }; + } + } + + /** + * Attempts to claim a task that's ready for execution, on this worker's + * behalf. We should not attempt to perform the work unless the claim really + * goes through. + * + * @param ticket - A globally unique string that changes for each invocation + * @param settings - The settings of the task to claim + * @returns True if it was successfully claimed + */ + async tryClaimTask( + ticket: string, + settings: TaskSettingsV1, + ): Promise { + const startedAt = this.knex.fn.now(); + const expiresAt = settings.timeoutAfterDuration + ? nowPlus(Duration.fromISO(settings.timeoutAfterDuration), this.knex) + : this.knex.raw('null'); + + const rows = await this.knex(DB_TASKS_TABLE) + .where('id', '=', this.taskId) + .whereNull('current_run_ticket') + .update({ + current_run_ticket: ticket, + current_run_started_at: startedAt, + current_run_expires_at: expiresAt, + }); + + return rows === 1; + } + + async tryReleaseTask( + ticket: string, + settings: TaskSettingsV1, + ): Promise { + const { recurringAtMostEveryDuration } = settings; + + // If this is not a recurring task, and we still have the current run + // ticket, delete it from the table + if (recurringAtMostEveryDuration === undefined) { + const rows = await this.knex(DB_TASKS_TABLE) + .where('id', '=', this.taskId) + .where('current_run_ticket', '=', ticket) + .delete(); + + return rows === 1; + } + + // We make an effort to keep the datetime calculations in the database + // layer, making sure to not have to perform conversions back and forth and + // leaning on the database as a central clock source + const dbNull = this.knex.raw('null'); + const dt = Duration.fromISO(recurringAtMostEveryDuration).as('seconds'); + const nextRun = + this.knex.client.config.client === 'sqlite3' + ? this.knex.raw('datetime(next_run_start_at, ?)', [`+${dt} seconds`]) + : this.knex.raw(`next_run_start_at + interval '${dt} seconds'`); + + const rows = await this.knex(DB_TASKS_TABLE) + .where('id', '=', this.taskId) + .where('current_run_ticket', '=', ticket) + .update({ + next_run_start_at: nextRun, + current_run_ticket: dbNull, + current_run_started_at: dbNull, + current_run_expires_at: dbNull, + }); + + return rows === 1; + } +} diff --git a/packages/backend-common/src/tasks/types.ts b/packages/backend-common/src/tasks/types.ts index 4a4fcb474e..4b061f9069 100644 --- a/packages/backend-common/src/tasks/types.ts +++ b/packages/backend-common/src/tasks/types.ts @@ -15,6 +15,7 @@ */ import { Duration } from 'luxon'; +import { z } from 'zod'; /** * Deals with management and locking related to distributed tasks, for a given @@ -41,5 +42,34 @@ export interface PluginTaskManager { initialDelay?: Duration; }, fn: () => Promise, - ): Promise<{ release: () => Promise }>; + ): Promise<{ unschedule: () => Promise }>; } + +function isValidOptionalDurationString(d: string | undefined): boolean { + try { + return !d || Duration.fromISO(d).isValid === true; + } catch { + return false; + } +} + +export const taskSettingsV1Schema = z.object({ + version: z.literal(1), + initialDelayDuration: z + .string() + .optional() + .refine(isValidOptionalDurationString, { message: 'Invalid duration' }), + recurringAtMostEveryDuration: z + .string() + .optional() + .refine(isValidOptionalDurationString, { message: 'Invalid duration' }), + timeoutAfterDuration: z + .string() + .optional() + .refine(isValidOptionalDurationString, { message: 'Invalid duration' }), +}); + +/** + * The properties that control a scheduled task (version 1). + */ +export type TaskSettingsV1 = z.infer; diff --git a/packages/backend-common/src/tasks/util.ts b/packages/backend-common/src/tasks/util.ts index 7bd4ff61b5..dd5c2d4b7c 100644 --- a/packages/backend-common/src/tasks/util.ts +++ b/packages/backend-common/src/tasks/util.ts @@ -15,6 +15,8 @@ */ import { InputError } from '@backstage/errors'; +import { Knex } from 'knex'; +import { DateTime, Duration } from 'luxon'; // Keep the IDs compatible with e.g. Prometheus export function validateId(id: string) { @@ -24,3 +26,20 @@ export function validateId(id: string) { ); } } + +export function dbTime(t: Date | string): DateTime { + if (typeof t === 'string') { + return DateTime.fromSQL(t); + } + return DateTime.fromJSDate(t); +} + +export function nowPlus(duration: Duration | undefined, knex: Knex) { + const seconds = duration?.as('seconds') ?? 0; + if (!seconds) { + return knex.fn.now(); + } + return knex.client.config.client === 'sqlite3' + ? knex.raw(`datetime('now', ?)`, [`${seconds} seconds`]) + : knex.raw(`now() + interval '${seconds} seconds'`); +} diff --git a/packages/backend/src/index.ts b/packages/backend/src/index.ts index 554c9a7d99..4e44729200 100644 --- a/packages/backend/src/index.ts +++ b/packages/backend/src/index.ts @@ -57,7 +57,7 @@ import techInsights from './plugins/techInsights'; import todo from './plugins/todo'; import { PluginEnvironment } from './types'; -async function makeCreateEnv(config: Config) { +function makeCreateEnv(config: Config) { const root = getRootLogger(); const reader = UrlReaders.default({ logger: root, config }); const discovery = SingleHostDiscovery.fromConfig(config); @@ -90,7 +90,7 @@ async function main() { argv: process.argv, logger, }); - const createEnv = await makeCreateEnv(config); + const createEnv = makeCreateEnv(config); const healthcheckEnv = useHotMemoize(module, () => createEnv('healthcheck')); const catalogEnv = useHotMemoize(module, () => createEnv('catalog')); diff --git a/yarn.lock b/yarn.lock index caba3db5fb..4c42941ad9 100644 --- a/yarn.lock +++ b/yarn.lock @@ -20895,11 +20895,6 @@ node-abi@^2.21.0: dependencies: semver "^5.4.1" -node-abort-controller@^3.0.0: - version "3.0.0" - resolved "https://registry.npmjs.org/node-abort-controller/-/node-abort-controller-3.0.0.tgz#e7b1f2b72f4c5a74b5594cc23fa8fe3b62010ab5" - integrity sha512-IqMCPbihDpbHV4bNws015hU0svIBGyzPjJearwXMGJyungWdblbBcboNojTz9bWOrrJD3zIwmcr5w5c+NH+2+A== - node-addon-api@^3.0.0: version "3.2.1" resolved "https://registry.npmjs.org/node-addon-api/-/node-addon-api-3.2.1.tgz#81325e0a2117789c0128dab65e7e38f07ceba161"