Merge pull request #25609 from backstage/timbonicush/backend-tasks-manual-trigger

Allow manual trigger frequency for SchedulerService and backend-tasks

.changeset/hip-fishes-guess.md

@@ -0,0 +1,6 @@
+---
+'@backstage/backend-plugin-api': patch
+'@backstage/backend-defaults': patch
+---
+
+The `SchedulerService` now allows tasks with `frequency: { trigger: 'manual' }`. This means that the task will not be scheduled, but rather run only when manually triggered with `SchedulerService.triggerTask`.

.changeset/metal-planes-nail.md

@@ -0,0 +1,5 @@
+---
+'@backstage/backend-tasks': patch
+---
+
+The `PluginTaskScheduler` now allows tasks with `frequency: { trigger: 'manual' }`. This means that the task will not be scheduled, but rather run only when manually triggered with `PluginTaskScheduler.triggerTask`.

packages/backend-defaults/migrations/scheduler/20240712211735_nullable_next_run.js

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2024 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.
+ */
+
+// @ts-check
+
+/**
+ * @param { import("knex").Knex } knex
+ * @returns { Promise<void> }
+ */
+exports.up = async function up(knex) {
+  await knex.schema.alterTable('backstage_backend_tasks__tasks', table => {
+    table.setNullable('next_run_start_at');
+  });
+};
+
+/**
+ * @param { import("knex").Knex } knex
+ * @returns { Promise<void> }
+ */
+exports.down = async function down(knex) {
+  await knex
+    .delete()
+    .from('backstage_backend_tasks__tasks')
+    .where({ next_run_start_at: null });
+  await knex.schema.alterTable('backstage_backend_tasks__tasks', table => {
+    table.dropNullable('next_run_start_at');
+  });
+};

packages/backend-defaults/src/entrypoints/scheduler/lib/PluginTaskSchedulerImpl.test.ts

@@ -347,6 +347,7 @@ describe('PluginTaskManagerImpl', () => {
       expect(parseDuration({ milliseconds: 5000 })).toEqual('PT5S');
       expect(parseDuration(Duration.fromMillis(5000))).toEqual('PT5S');
       expect(parseDuration({ cron: '1 * * * *' })).toEqual('1 * * * *');
+      expect(parseDuration({ trigger: 'manual' })).toEqual('manual');
     });
   });
 });

packages/backend-defaults/src/entrypoints/scheduler/lib/PluginTaskSchedulerImpl.ts

@@ -155,6 +155,9 @@ export function parseDuration(
   if (typeof frequency === 'object' && 'cron' in frequency) {
     return frequency.cron;
   }
+  if (typeof frequency === 'object' && 'trigger' in frequency) {
+    return frequency.trigger;
+  }
 
   const parsed = Duration.isDuration(frequency)
     ? frequency

packages/backend-defaults/src/entrypoints/scheduler/lib/TaskWorker.test.ts

@@ -503,4 +503,32 @@ describe('TaskWorker', () => {
       await knex.destroy();
     },
   );
+
+  it.each(databases.eachSupportedId())(
+    'next_run_start_at is not set for manually-triggered tasks, %p',
+    async databaseId => {
+      const knex = await databases.init(databaseId);
+      await migrateBackendTasks(knex);
+
+      const fn = jest.fn(
+        async () => new Promise<void>(resolve => setTimeout(resolve, 50)),
+      );
+
+      const initialSettings: TaskSettingsV2 = {
+        version: 2,
+        cadence: 'manual',
+        timeoutAfterDuration: 'PT1M',
+      };
+
+      const worker = new TaskWorker('task99', fn, knex, logger);
+      await worker.persistTask(initialSettings);
+      await worker.tryClaimTask('ticket', initialSettings);
+      await worker.tryReleaseTask('ticket', initialSettings);
+
+      const row = (await knex<DbTasksRow>(DB_TASKS_TABLE))[0];
+      expect(row.next_run_start_at).toBeNull();
+
+      await knex.destroy();
+    },
+  );
 });

packages/backend-defaults/src/entrypoints/scheduler/lib/TaskWorker.ts

@@ -53,8 +53,8 @@ export class TaskWorker {
     );
 
     let workCheckFrequency = this.workCheckFrequency;
-    const isCron = !settings?.cadence.startsWith('P');
-    if (!isCron) {
+    const isDuration = settings?.cadence.startsWith('P');
+    if (isDuration) {
       const cadence = Duration.fromISO(settings.cadence);
       if (cadence < workCheckFrequency) {
         workCheckFrequency = cadence;
@@ -175,7 +175,9 @@ export class TaskWorker {
     // read it back again.
     taskSettingsV2Schema.parse(settings);
 
-    const isCron = !settings?.cadence.startsWith('P');
+    const isManual = settings?.cadence === 'manual';
+    const isDuration = settings?.cadence.startsWith('P');
+    const isCron = !isManual && !isDuration;
 
     let startAt: Knex.Raw | undefined;
     let nextStartAt: Knex.Raw | undefined;
@@ -194,6 +196,9 @@ export class TaskWorker {
 
       nextStartAt = this.nextRunAtRaw(time);
       startAt ||= nextStartAt;
+    } else if (isManual) {
+      nextStartAt = this.knex.raw('null');
+      startAt ||= nextStartAt;
     } else {
       startAt ||= this.knex.fn.now();
       nextStartAt = nowPlus(Duration.fromISO(settings.cadence), this.knex);
@@ -317,7 +322,9 @@ export class TaskWorker {
     ticket: string,
     settings: TaskSettingsV2,
   ): Promise<boolean> {
-    const isCron = !settings?.cadence.startsWith('P');
+    const isManual = settings?.cadence === 'manual';
+    const isDuration = settings?.cadence.startsWith('P');
+    const isCron = !isManual && !isDuration;
 
     let nextRun: Knex.Raw;
     if (isCron) {
@@ -325,6 +332,8 @@ export class TaskWorker {
       this.logger.debug(`task: ${this.taskId} will next occur around ${time}`);
 
       nextRun = this.nextRunAtRaw(time);
+    } else if (isManual) {
+      nextRun = this.knex.raw('null');
     } else {
       const dt = Duration.fromISO(settings.cadence).as('seconds');
       this.logger.debug(

packages/backend-defaults/src/entrypoints/scheduler/lib/types.ts

@@ -40,6 +40,10 @@ function isValidCronFormat(c: string | undefined): boolean {
   }
 }
 
+function isValidTrigger(t: string): boolean {
+  return t === 'manual';
+}
+
 export const taskSettingsV1Schema = z.object({
   version: z.literal(1),
   initialDelayDuration: z
@@ -68,6 +72,11 @@ export const taskSettingsV2Schema = z.object({
   cadence: z
     .string()
     .refine(isValidCronFormat, { message: 'Invalid cron' })
+    .or(
+      z.string().refine(isValidTrigger, {
+        message: "Invalid trigger, expecting 'manual'",
+      }),
+    )
     .or(
       z.string().refine(isValidOptionalDurationString, {
         message: 'Invalid duration, expecting ISO Period',

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

@@ -0,0 +1,123 @@
+/*
+ * Copyright 2024 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/scheduler`;
+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);
+  }
+}
+
+jest.setTimeout(60_000);
+
+describe('migrations', () => {
+  const databases = TestDatabases.create();
+
+  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);
+
+      // This looks odd - you might expect a .toThrow at the end but that
+      // actually is flaky for some reason specifically on sqlite when
+      // performing multiple runs in sequence
+      await expect(knex('backstage_backend_tasks__tasks')).rejects.toEqual(
+        expect.anything(),
+      );
+
+      await knex.destroy();
+    },
+  );
+
+  it.each(databases.eachSupportedId())(
+    '20240712211735_nullable_next_run.js, %p',
+    async databaseId => {
+      const knex = await databases.init(databaseId);
+
+      await migrateUntilBefore(knex, '20240712211735_nullable_next_run.js');
+      await migrateUpOnce(knex);
+
+      await knex('backstage_backend_tasks__tasks').insert({
+        id: 'test',
+        settings_json: '{}',
+        next_run_start_at: knex.raw('null'),
+      });
+
+      await expect(knex('backstage_backend_tasks__tasks')).resolves.toEqual([
+        {
+          id: 'test',
+          settings_json: '{}',
+          next_run_start_at: null,
+          current_run_ticket: null,
+          current_run_started_at: null,
+          current_run_expires_at: null,
+        },
+      ]);
+
+      await migrateDownOnce(knex);
+
+      await expect(
+        knex('backstage_backend_tasks__tasks').insert({
+          id: 'test',
+          settings_json: '{}',
+          next_run_start_at: knex.raw('null'),
+        }),
+      ).rejects.toEqual(expect.anything());
+
+      await knex.destroy();
+    },
+  );
+});

packages/backend-plugin-api/api-report.md

@@ -607,7 +607,10 @@ export interface SchedulerServiceTaskScheduleDefinition {
         cron: string;
       }
     | Duration
-    | HumanDuration;
+    | HumanDuration
+    | {
+        trigger: 'manual';
+      };
   initialDelay?: Duration | HumanDuration;
   scope?: 'global' | 'local';
   timeout: Duration | HumanDuration;
@@ -620,7 +623,14 @@ export interface SchedulerServiceTaskScheduleDefinitionConfig {
         cron: string;
       }
     | string
-    | HumanDuration;
+    | HumanDuration
+    /**
+     * This task will only run when manually triggered with the `triggerTask` method; no automatic
+     * scheduling. This is useful for locking of global tasks that should not be run concurrently.
+     */
+    | {
+        trigger: 'manual';
+      };
   initialDelay?: string | HumanDuration;
   scope?: 'global' | 'local';
   timeout: string | HumanDuration;

packages/backend-plugin-api/src/services/definitions/SchedulerService.ts

@@ -96,7 +96,8 @@ export interface SchedulerServiceTaskScheduleDefinition {
         cron: string;
       }
     | Duration
-    | HumanDuration;
+    | HumanDuration
+    | { trigger: 'manual' };
 
   /**
    * The maximum amount of time that a single task invocation can take, before
@@ -193,7 +194,12 @@ export interface SchedulerServiceTaskScheduleDefinitionConfig {
         cron: string;
       }
     | string
-    | HumanDuration;
+    | HumanDuration
+    /**
+     * This task will only run when manually triggered with the `triggerTask` method; no automatic
+     * scheduling. This is useful for locking of global tasks that should not be run concurrently.
+     */
+    | { trigger: 'manual' };
 
   /**
    * The maximum amount of time that a single task invocation can take, before
@@ -361,14 +367,20 @@ function readDuration(config: Config, key: string): HumanDuration {
   return readDurationFromConfig(config, { key });
 }
 
-function readCronOrDuration(
+function readFrequency(
   config: Config,
   key: string,
-): { cron: string } | HumanDuration {
+): { cron: string } | HumanDuration | { trigger: 'manual' } {
   const value = config.get(key);
   if (typeof value === 'object' && (value as { cron?: string }).cron) {
     return value as { cron: string };
   }
+  if (
+    typeof value === 'object' &&
+    (value as { trigger?: string }).trigger === 'manual'
+  ) {
+    return { trigger: 'manual' };
+  }
 
   return readDuration(config, key);
 }
@@ -383,7 +395,7 @@ function readCronOrDuration(
 export function readSchedulerServiceTaskScheduleDefinitionFromConfig(
   config: Config,
 ): SchedulerServiceTaskScheduleDefinition {
-  const frequency = readCronOrDuration(config, 'frequency');
+  const frequency = readFrequency(config, 'frequency');
   const timeout = readDuration(config, 'timeout');
 
   const initialDelay = config.has('initialDelay')

packages/backend-tasks/api-report.md

@@ -62,7 +62,14 @@ export interface TaskScheduleDefinition {
         cron: string;
       }
     | Duration
-    | HumanDuration_2;
+    | HumanDuration_2
+    /**
+     * This task will only run when manually triggered with the `triggerTask` method; no automatic
+     * scheduling. This is useful for locking of global tasks that should not be run concurrently.
+     */
+    | {
+        trigger: 'manual';
+      };
   initialDelay?: Duration | HumanDuration_2;
   scope?: 'global' | 'local';
   timeout: Duration | HumanDuration_2;

packages/backend-tasks/migrations/20240712211735_nullable_next_run.js

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2024 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.
+ */
+
+// @ts-check
+
+/**
+ * @param { import("knex").Knex } knex
+ * @returns { Promise<void> }
+ */
+exports.up = async function up(knex) {
+  await knex.schema.alterTable('backstage_backend_tasks__tasks', table => {
+    table.setNullable('next_run_start_at');
+  });
+};
+
+/**
+ * @param { import("knex").Knex } knex
+ * @returns { Promise<void> }
+ */
+exports.down = async function down(knex) {
+  await knex
+    .delete()
+    .from('backstage_backend_tasks__tasks')
+    .where({ next_run_start_at: null });
+  await knex.schema.alterTable('backstage_backend_tasks__tasks', table => {
+    table.dropNullable('next_run_start_at');
+  });
+};

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

@@ -81,4 +81,43 @@ describe('migrations', () => {
       await knex.destroy();
     },
   );
+
+  it.each(databases.eachSupportedId())(
+    '20240712211735_nullable_next_run.js, %p',
+    async databaseId => {
+      const knex = await databases.init(databaseId);
+
+      await migrateUntilBefore(knex, '20240712211735_nullable_next_run.js');
+      await migrateUpOnce(knex);
+
+      await knex('backstage_backend_tasks__tasks').insert({
+        id: 'test',
+        settings_json: '{}',
+        next_run_start_at: knex.raw('null'),
+      });
+
+      await expect(knex('backstage_backend_tasks__tasks')).resolves.toEqual([
+        {
+          id: 'test',
+          settings_json: '{}',
+          next_run_start_at: null,
+          current_run_ticket: null,
+          current_run_started_at: null,
+          current_run_expires_at: null,
+        },
+      ]);
+
+      await migrateDownOnce(knex);
+
+      await expect(
+        knex('backstage_backend_tasks__tasks').insert({
+          id: 'test',
+          settings_json: '{}',
+          next_run_start_at: knex.raw('null'),
+        }),
+      ).rejects.toEqual(expect.anything());
+
+      await knex.destroy();
+    },
+  );
 });

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

@@ -345,6 +345,7 @@ describe('PluginTaskManagerImpl', () => {
       expect(parseDuration({ milliseconds: 5000 })).toEqual('PT5S');
       expect(parseDuration(Duration.fromMillis(5000))).toEqual('PT5S');
       expect(parseDuration({ cron: '1 * * * *' })).toEqual('1 * * * *');
+      expect(parseDuration({ trigger: 'manual' })).toEqual('manual');
     });
   });
 });

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

@@ -152,6 +152,9 @@ export function parseDuration(
   if ('cron' in frequency) {
     return frequency.cron;
   }
+  if ('trigger' in frequency) {
+    return frequency.trigger;
+  }
 
   const parsed = Duration.isDuration(frequency)
     ? frequency

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

@@ -503,4 +503,32 @@ describe('TaskWorker', () => {
       await knex.destroy();
     },
   );
+
+  it.each(databases.eachSupportedId())(
+    'next_run_start_at is not set for manually-triggered tasks, %p',
+    async databaseId => {
+      const knex = await databases.init(databaseId);
+      await migrateBackendTasks(knex);
+
+      const fn = jest.fn(
+        async () => new Promise<void>(resolve => setTimeout(resolve, 50)),
+      );
+
+      const initialSettings: TaskSettingsV2 = {
+        version: 2,
+        cadence: 'manual',
+        timeoutAfterDuration: 'PT1M',
+      };
+
+      const worker = new TaskWorker('task99', fn, knex, logger);
+      await worker.persistTask(initialSettings);
+      await worker.tryClaimTask('ticket', initialSettings);
+      await worker.tryReleaseTask('ticket', initialSettings);
+
+      const row = (await knex<DbTasksRow>(DB_TASKS_TABLE))[0];
+      expect(row.next_run_start_at).toBeNull();
+
+      await knex.destroy();
+    },
+  );
 });

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

@@ -52,8 +52,8 @@ export class TaskWorker {
     );
 
     let workCheckFrequency = this.workCheckFrequency;
-    const isCron = !settings?.cadence.startsWith('P');
-    if (!isCron) {
+    const isDuration = settings?.cadence.startsWith('P');
+    if (isDuration) {
       const cadence = Duration.fromISO(settings.cadence);
       if (cadence < workCheckFrequency) {
         workCheckFrequency = cadence;
@@ -174,7 +174,9 @@ export class TaskWorker {
     // read it back again.
     taskSettingsV2Schema.parse(settings);
 
-    const isCron = !settings?.cadence.startsWith('P');
+    const isManual = settings?.cadence === 'manual';
+    const isDuration = settings?.cadence.startsWith('P');
+    const isCron = !isManual && !isDuration;
 
     let startAt: Knex.Raw | undefined;
     let nextStartAt: Knex.Raw | undefined;
@@ -193,6 +195,9 @@ export class TaskWorker {
 
       nextStartAt = this.nextRunAtRaw(time);
       startAt ||= nextStartAt;
+    } else if (isManual) {
+      nextStartAt = this.knex.raw('null');
+      startAt ||= nextStartAt;
     } else {
       startAt ||= this.knex.fn.now();
       nextStartAt = nowPlus(Duration.fromISO(settings.cadence), this.knex);
@@ -316,7 +321,9 @@ export class TaskWorker {
     ticket: string,
     settings: TaskSettingsV2,
   ): Promise<boolean> {
-    const isCron = !settings?.cadence.startsWith('P');
+    const isManual = settings?.cadence === 'manual';
+    const isDuration = settings?.cadence.startsWith('P');
+    const isCron = !isManual && !isDuration;
 
     let nextRun: Knex.Raw;
     if (isCron) {
@@ -324,6 +331,8 @@ export class TaskWorker {
       this.logger.debug(`task: ${this.taskId} will next occur around ${time}`);
 
       nextRun = this.nextRunAtRaw(time);
+    } else if (isManual) {
+      nextRun = this.knex.raw('null');
     } else {
       const dt = Duration.fromISO(settings.cadence).as('seconds');
       this.logger.debug(

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

@@ -32,14 +32,20 @@ function readDuration(config: Config, key: string): HumanDuration {
   return readDurationFromConfig(config, { key });
 }
 
-function readCronOrDuration(
+function readFrequency(
   config: Config,
   key: string,
-): { cron: string } | Duration | HumanDuration {
+): { cron: string } | Duration | HumanDuration | { trigger: 'manual' } {
   const value = config.get(key);
   if (typeof value === 'object' && (value as { cron?: string }).cron) {
     return value as { cron: string };
   }
+  if (
+    typeof value === 'object' &&
+    (value as { trigger?: string }).trigger === 'manual'
+  ) {
+    return { trigger: 'manual' };
+  }
 
   return readDuration(config, key);
 }
@@ -56,7 +62,7 @@ function readCronOrDuration(
 export function readTaskScheduleDefinitionFromConfig(
   config: Config,
 ): TaskScheduleDefinition {
-  const frequency = readCronOrDuration(config, 'frequency');
+  const frequency = readFrequency(config, 'frequency');
   const timeout = readDuration(config, 'timeout');
 
   const initialDelay = config.has('initialDelay')

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

@@ -100,7 +100,12 @@ export interface TaskScheduleDefinition {
         cron: string;
       }
     | Duration
-    | HumanDuration;
+    | HumanDuration
+    /**
+     * This task will only run when manually triggered with the `triggerTask` method; no automatic
+     * scheduling. This is useful for locking of global tasks that should not be run concurrently.
+     */
+    | { trigger: 'manual' };
 
   /**
    * The maximum amount of time that a single task invocation can take, before
@@ -373,6 +378,10 @@ function isValidCronFormat(c: string | undefined): boolean {
   }
 }
 
+function isValidTrigger(t: string): boolean {
+  return t === 'manual';
+}
+
 export const taskSettingsV1Schema = z.object({
   version: z.literal(1),
   initialDelayDuration: z
@@ -401,6 +410,11 @@ export const taskSettingsV2Schema = z.object({
   cadence: z
     .string()
     .refine(isValidCronFormat, { message: 'Invalid cron' })
+    .or(
+      z.string().refine(isValidTrigger, {
+        message: "Invalid trigger, expecting 'manual'",
+      }),
+    )
     .or(
       z.string().refine(isValidOptionalDurationString, {
         message: 'Invalid duration, expecting ISO Period',