Get most of the task worker code into place

Signed-off-by: Fredrik Adelöw <freben@gmail.com>

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<string>;
 };
 
+// @public
+export interface PluginTaskManager {
+  // (undocumented)
+  acquireLock(
+    id: string,
+    options: {
+      timeout: Duration;
+    },
+  ): Promise<
+    | {
+        acquired: false;
+      }
+    | {
+        acquired: true;
+        release: () => void | Promise<void>;
+      }
+  >;
+  // (undocumented)
+  scheduleTask(
+    id: string,
+    options: {
+      timeout: Duration;
+      frequency: Duration;
+      initialDelay?: Duration;
+    },
+    fn: () => Promise<void>,
+  ): Promise<{
+    unschedule: () => Promise<void>;
+  }>;
+}
+
 // @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<Buffer>;

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');
 };

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",

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;
 };

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<void>;
+
+  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<void> {
+    return this.#cancelPromise;
+  }
+}

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();
+      },
+    );
+  });
+});

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<Knex>,
+    private readonly logger: Logger,
+  ) {}
+
+  async acquireLock(
+    id: string,
+    options: {
+      timeout: Duration;
+    },
+  ): Promise<
+    | { acquired: false }
+    | { acquired: true; release: () => void | Promise<void> }
+  > {
+    validateId(id);
+
+    const knex = await this.databaseFactory();
+    await this.ensureJanitor(knex);
+
+    const ticket = uuid();
+
+    async function release() {
+      try {
+        await knex<DbMutexesRow>(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<DbMutexesRow>(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<DbMutexesRow>(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<void>,
+  ): Promise<{ unschedule: () => Promise<void> }> {
+    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();
+    }
+  }
+}

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);
+    },
+  );
 });

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<Knex>,
-  ) {}
-
-  async acquireLock(
-    idWithoutPrefix: string,
-    options: {
-      timeout: Duration;
-    },
-  ): Promise<
-    | { acquired: false }
-    | { acquired: true; release: () => void | Promise<void> }
-  > {
-    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<DbTaskLocksRow>('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<DbTaskLocksRow>(
-      '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<DbTaskLocksRow>('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<void>,
-  ): Promise<{ release: () => Promise<void> }> {
-    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 }),
     );
   }
 }

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<void>(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<void>(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<DbTasksRow>(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<DbTasksRow>(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<DbTasksRow>(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<DbTasksRow>(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<void>(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<DbTasksRow>(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<DbTasksRow>(DB_TASKS_TABLE)
+        .where('id', '=', 'task1')
+        .update({ current_run_ticket: 'stolen' });
+
+      await expect(worker.tryReleaseTask('ticket', settings)).resolves.toBe(
+        false,
+      );
+
+      row = (await knex<DbTasksRow>(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<DbTasksRow>(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<DbTasksRow>(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<DbTasksRow>(DB_TASKS_TABLE).where('id', '=', 'task3').delete();
+      await expect(worker3.tryReleaseTask('ticket', settings)).resolves.toBe(
+        false,
+      );
+    },
+    60_000,
+  );
+});

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<void>;
+  private readonly knex: Knex;
+  private readonly logger: Logger;
+  private readonly cancelToken: CancelToken;
+
+  constructor(
+    taskId: string,
+    fn: () => void | Promise<void>,
+    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<void> {
+    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<DbTasksRow>(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<DbTasksRow>(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<boolean> {
+    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<DbTasksRow>(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<boolean> {
+    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<DbTasksRow>(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<DbTasksRow>(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;
+  }
+}

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<void>,
-  ): Promise<{ release: () => Promise<void> }>;
+  ): Promise<{ unschedule: () => Promise<void> }>;
 }
+
+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<typeof taskSettingsV1Schema>;

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'`);
+}

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'));

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"