Merge pull request #10525 from backstage/freben/initial-wait

fix initialDelay for tasks

.changeset/ninety-radios-protect.md

@@ -0,0 +1,14 @@
+---
+'@backstage/backend-tasks': patch
+---
+
+Fixed the `initialDelay` parameter of tasks to properly make task workers
+_always_ wait before the first invocations on startup, not just the very first
+time that the task is ever created. This behavior is more in line with
+expectations. Callers to not need to update their code.
+
+Also clarified in the doc comment for the field that this wait applies only on
+an individual worker level. That is, if you have a cluster of workers then each
+individual machine may postpone its first task invocation by the given amount of
+time to leave room for the service to settle, but _other_ workers may still
+continue to invoke the task on the regular cadence in the meantime.

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

@@ -14,6 +14,7 @@
  * limitations under the License.
  */
 
+import { AbortController } from 'node-abort-controller';
 import { getVoidLogger } from '@backstage/backend-common';
 import { TestDatabases } from '@backstage/backend-test-utils';
 import { Duration } from 'luxon';
@@ -257,4 +258,58 @@ describe('TaskWorker', () => {
     },
     60_000,
   );
+
+  it.each(databases.eachSupportedId())(
+    'respects initialDelayDuration per worker, %p',
+    async databaseId => {
+      const knex = await databases.init(databaseId);
+      await migrateBackendTasks(knex);
+
+      const abortFirst = new AbortController();
+      const settings: TaskSettingsV2 = {
+        version: 2,
+        initialDelayDuration: 'PT0.3S',
+        cadence: 'PT0.1S',
+        timeoutAfterDuration: 'PT10S',
+      };
+
+      // Start a single worker and make sure it waits and then goes to work
+      const fn1 = jest.fn(async () => {});
+      const worker1 = new TaskWorker(
+        'task1',
+        fn1,
+        knex,
+        logger,
+        Duration.fromMillis(10),
+      );
+      await worker1.start(settings, { signal: abortFirst.signal });
+
+      expect(fn1).toBeCalledTimes(0);
+      await new Promise(resolve => setTimeout(resolve, 250));
+      expect(fn1).toBeCalledTimes(0);
+      await new Promise(resolve => setTimeout(resolve, 100));
+      expect(fn1.mock.calls.length).toBeGreaterThan(0);
+
+      // Start a second worker and make sure it waits but the first worker still works along
+      const fn2 = jest.fn();
+      const promise2 = new Promise(resolve => fn2.mockImplementation(resolve));
+      const worker2 = new TaskWorker(
+        'task1',
+        fn2,
+        knex,
+        logger,
+        Duration.fromMillis(10),
+      );
+      await worker2.start(settings);
+
+      // We eventually abort the first worker just to make sure that the second
+      // one for sure will get a go at running the task
+      setTimeout(() => abortFirst.abort(), 1000);
+
+      const before = fn1.mock.calls.length;
+      await promise2;
+      expect(fn1.mock.calls.length).toBeGreaterThan(before);
+    },
+    60_000,
+  );
 });

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

@@ -53,6 +53,13 @@ export class TaskWorker {
 
     (async () => {
       try {
+        if (settings.initialDelayDuration) {
+          await sleep(
+            Duration.fromISO(settings.initialDelayDuration),
+            options?.signal,
+          );
+        }
+
         while (!options?.signal?.aborted) {
           const runResult = await this.runOnce(options?.signal);
           if (runResult.result === 'abort') {
@@ -61,6 +68,7 @@ export class TaskWorker {
 
           await sleep(this.workCheckFrequency, options?.signal);
         }
+
         this.logger.info(`Task worker finished: ${this.taskId}`);
       } catch (e) {
         this.logger.warn(`Task worker failed unexpectedly, ${e}`);
@@ -106,6 +114,7 @@ export class TaskWorker {
 
     try {
       await this.fn(taskAbortController.signal);
+      taskAbortController.abort(); // releases resources
     } catch (e) {
       await this.tryReleaseTask(ticket, taskSettings);
       return { result: 'failed' };

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

@@ -41,6 +41,8 @@ export interface TaskScheduleDefinition {
    * How often you want the task to run. The system does its best to avoid
    * overlapping invocations.
    *
+   * @remarks
+   *
    * This is a best effort value; under some circumstances there can be
    * deviations. For example, if the task runtime is longer than the frequency
    * and the timeout has not been given or not been exceeded yet, the next
@@ -54,6 +56,8 @@ export interface TaskScheduleDefinition {
         /**
          * A crontab style string.
          *
+         * @remarks
+         *
          * Overview:
          *
          * ```
@@ -82,11 +86,19 @@ export interface TaskScheduleDefinition {
   /**
    * The amount of time that should pass before the first invocation happens.
    *
-   * This can be useful in cold start scenarios to stagger or delay some heavy
-   * compute jobs.
+   * @remarks
    *
-   * If no value is given for this field then the first invocation will happen
-   * as soon as possible according to the cadence.
+   * This can be useful in cold start scenarios to stagger or delay some heavy
+   * compute jobs. If no value is given for this field then the first invocation
+   * will happen as soon as possible according to the cadence.
+   *
+   * NOTE: This is a per-worker delay. If you have a cluster of workers all
+   * collaborating on a task, then you may still see the task being processed by
+   * other long-lived workers, while any given single worker is in its initial
+   * sleep delay time e.g. after a deployment. Therefore this parameter is not
+   * useful for "globally" pausing work; its main intended use is for individual
+   * machines to get a chance to reach some equilibrium at startup before
+   * triggering heavy batch workloads.
    */
   initialDelay?: Duration;
 }