add migrations test, return tasks async

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

packages/backend-tasks/api-report.md

@@ -7,6 +7,7 @@ import { Config } from '@backstage/config';
 import { DatabaseManager } from '@backstage/backend-common';
 import { Duration } from 'luxon';
 import { HumanDuration as HumanDuration_2 } from '@backstage/types';
+import { JsonObject } from '@backstage/types';
 import { Logger } from 'winston';
 import { PluginDatabaseManager } from '@backstage/backend-common';
 
@@ -16,7 +17,7 @@ export type HumanDuration = HumanDuration_2;
 // @public
 export interface PluginTaskScheduler {
   createScheduledTaskRunner(schedule: TaskScheduleDefinition): TaskRunner;
-  getScheduledTasks(): TaskDescriptor[];
+  getScheduledTasks(): Promise<TaskDescriptor[]>;
   scheduleTask(
     task: TaskScheduleDefinition & TaskInvocationDefinition,
   ): Promise<void>;
@@ -29,8 +30,13 @@ export function readTaskScheduleDefinitionFromConfig(
 ): TaskScheduleDefinition;
 
 // @public
-export type TaskDescriptor = TaskScheduleDefinition &
-  Exclude<TaskInvocationDefinition, 'fn' | 'signal'>;
+export type TaskDescriptor = {
+  id: string;
+  scope: 'global' | 'local';
+  settings: {
+    version: number;
+  } & JsonObject;
+};
 
 // @public
 export type TaskFunction =

packages/backend-tasks/src/migrations.test.ts

@@ -0,0 +1,82 @@
+/*
+ * Copyright 2022 The Backstage Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { Knex } from 'knex';
+import { TestDatabases } from '@backstage/backend-test-utils';
+import fs from 'fs';
+
+const migrationsDir = `${__dirname}/../migrations`;
+const migrationsFiles = fs.readdirSync(migrationsDir).sort();
+
+async function migrateUpOnce(knex: Knex): Promise<void> {
+  await knex.migrate.up({ directory: migrationsDir });
+}
+
+async function migrateDownOnce(knex: Knex): Promise<void> {
+  await knex.migrate.down({ directory: migrationsDir });
+}
+
+async function migrateUntilBefore(knex: Knex, target: string): Promise<void> {
+  const index = migrationsFiles.indexOf(target);
+  if (index === -1) {
+    throw new Error(`Migration ${target} not found`);
+  }
+  for (let i = 0; i < index; i++) {
+    await migrateUpOnce(knex);
+  }
+}
+
+describe('migrations', () => {
+  const databases = TestDatabases.create({
+    ids: ['POSTGRES_13', 'POSTGRES_9', 'SQLITE_3'],
+  });
+
+  it.each(databases.eachSupportedId())(
+    '20210928160613_init.js, %p',
+    async databaseId => {
+      const knex = await databases.init(databaseId);
+
+      await migrateUntilBefore(knex, '20210928160613_init.js');
+      await migrateUpOnce(knex);
+
+      await knex('backstage_backend_tasks__tasks').insert({
+        id: 'test',
+        settings_json: '{}',
+        next_run_start_at: knex.fn.now(),
+      });
+
+      await expect(knex('backstage_backend_tasks__tasks')).resolves.toEqual([
+        {
+          id: 'test',
+          settings_json: '{}',
+          next_run_start_at: expect.anything(),
+          current_run_ticket: null,
+          current_run_started_at: null,
+          current_run_expires_at: null,
+        },
+      ]);
+
+      await migrateDownOnce(knex);
+
+      await expect(knex('backstage_backend_tasks__tasks')).rejects.toThrow(
+        /backstage_backend_tasks__tasks/,
+      );
+
+      await knex.destroy();
+    },
+    60_000,
+  );
+});

packages/backend-tasks/src/tasks/LocalTaskWorker.ts

@@ -39,8 +39,9 @@ export class LocalTaskWorker {
     this.logger.info(
       `Task worker starting: ${this.taskId}, ${JSON.stringify(settings)}`,
     );
-    let attemptNum = 1;
+
     (async () => {
+      let attemptNum = 1;
       for (;;) {
         try {
           if (settings.initialDelayDuration) {
@@ -60,6 +61,7 @@ export class LocalTaskWorker {
               options?.signal,
             );
           }
+
           this.logger.info(`Task worker finished: ${this.taskId}`);
           attemptNum = 0;
           break;

packages/backend-tasks/src/tasks/PluginTaskSchedulerImpl.test.ts

@@ -309,9 +309,8 @@ describe('PluginTaskManagerImpl', () => {
       'can fetch both global and local task ids, %p',
       async databaseId => {
         const { manager } = await init(databaseId);
-
         const fn = jest.fn();
-        const promise = new Promise(resolve => fn.mockImplementation(resolve));
+
         await manager.scheduleTask({
           id: 'task1',
           timeout: Duration.fromMillis(5000),
@@ -328,18 +327,18 @@ describe('PluginTaskManagerImpl', () => {
           scope: 'local',
         });
 
-        await promise;
-
-        const tasks = manager.getScheduledTasks();
-        expect(tasks.length).toEqual(2);
-        expect(tasks[0].id).toEqual('task1');
-        expect(tasks[1].id).toEqual('task2');
-        expect(tasks[0].scope).toEqual('global');
-        expect(tasks[1].scope).toEqual('local');
-        expect(tasks[0].fn).toBeUndefined();
-        expect(tasks[1].fn).toBeUndefined();
-        expect(tasks[0].signal).toBeUndefined();
-        expect(tasks[1].signal).toBeUndefined();
+        await expect(manager.getScheduledTasks()).resolves.toEqual([
+          {
+            id: 'task1',
+            scope: 'global',
+            settings: expect.objectContaining({ cadence: 'PT5S' }),
+          },
+          {
+            id: 'task2',
+            scope: 'local',
+            settings: expect.objectContaining({ cadence: 'PT5S' }),
+          },
+        ]);
       },
       60_000,
     );

packages/backend-tasks/src/tasks/PluginTaskSchedulerImpl.ts

@@ -25,6 +25,7 @@ import {
   TaskInvocationDefinition,
   TaskRunner,
   TaskScheduleDefinition,
+  TaskSettingsV2,
 } from './types';
 import { validateId } from './util';
 
@@ -57,6 +58,14 @@ export class PluginTaskSchedulerImpl implements PluginTaskScheduler {
     validateId(task.id);
     const scope = task.scope ?? 'global';
 
+    const settings: TaskSettingsV2 = {
+      version: 2,
+      cadence: parseDuration(task.frequency),
+      initialDelayDuration:
+        task.initialDelay && parseDuration(task.initialDelay),
+      timeoutAfterDuration: parseDuration(task.timeout),
+    };
+
     if (scope === 'global') {
       const knex = await this.databaseFactory();
       const worker = new TaskWorker(
@@ -65,39 +74,22 @@ export class PluginTaskSchedulerImpl implements PluginTaskScheduler {
         knex,
         this.logger.child({ task: task.id }),
       );
-
-      await worker.start(
-        {
-          version: 2,
-          cadence: parseDuration(task.frequency),
-          initialDelayDuration:
-            task.initialDelay && parseDuration(task.initialDelay),
-          timeoutAfterDuration: parseDuration(task.timeout),
-        },
-        {
-          signal: task.signal,
-        },
-      );
+      await worker.start(settings, { signal: task.signal });
     } else {
-      const worker = new LocalTaskWorker(task.id, task.fn, this.logger);
-
-      worker.start(
-        {
-          version: 2,
-          cadence: parseDuration(task.frequency),
-          initialDelayDuration:
-            task.initialDelay && parseDuration(task.initialDelay),
-          timeoutAfterDuration: parseDuration(task.timeout),
-        },
-        {
-          signal: task.signal,
-        },
+      const worker = new LocalTaskWorker(
+        task.id,
+        task.fn,
+        this.logger.child({ task: task.id }),
       );
-
+      worker.start(settings, { signal: task.signal });
       this.localTasksById.set(task.id, worker);
     }
-    const { fn: _, signal: __, ...descriptor } = task;
-    this.allScheduledTasks.push(descriptor as TaskDescriptor);
+
+    this.allScheduledTasks.push({
+      id: task.id,
+      scope: scope,
+      settings: settings,
+    });
   }
 
   createScheduledTaskRunner(schedule: TaskScheduleDefinition): TaskRunner {
@@ -108,7 +100,7 @@ export class PluginTaskSchedulerImpl implements PluginTaskScheduler {
     };
   }
 
-  getScheduledTasks(): TaskDescriptor[] {
+  async getScheduledTasks(): Promise<TaskDescriptor[]> {
     return this.allScheduledTasks;
   }
 }

packages/backend-tasks/src/tasks/types.ts

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-import { HumanDuration } from '@backstage/types';
+import { HumanDuration, JsonObject } from '@backstage/types';
 import { CronTime } from 'cron';
 import { Duration } from 'luxon';
 import { z } from 'zod';
@@ -32,12 +32,26 @@ export type TaskFunction =
   | (() => void | Promise<void>);
 
 /**
- * A type to describe a scheduled task.
+ * A semi-opaque type to describe an actively scheduled task.
  *
  * @public
  */
-export type TaskDescriptor = TaskScheduleDefinition &
-  Exclude<TaskInvocationDefinition, 'fn' | 'signal'>;
+export type TaskDescriptor = {
+  /**
+   * The unique identifier of the task.
+   */
+  id: string;
+  /**
+   * The scope of the task.
+   */
+  scope: 'global' | 'local';
+  /**
+   * The settings that control the task flow. This is a semi-opaque structure
+   * that is mainly there for debugging purposes. Do not make any assumptions
+   * about the contents of this field.
+   */
+  settings: { version: number } & JsonObject;
+};
 
 /**
  * Options that control the scheduling of a task.
@@ -327,7 +341,7 @@ export interface PluginTaskScheduler {
    *
    * @returns Scheduled tasks
    */
-  getScheduledTasks(): TaskDescriptor[];
+  getScheduledTasks(): Promise<TaskDescriptor[]>;
 }
 
 function isValidOptionalDurationString(d: string | undefined): boolean {