make startTaskPipeline iterative

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

.changeset/cold-apricots-poke.md

@@ -0,0 +1,5 @@
+---
+'@backstage/plugin-catalog-backend': patch
+---
+
+Changed the processing loop task pipeline implementation from recursive to iterative

plugins/catalog-backend/src/processing/TaskPipeline.ts

@@ -14,6 +14,8 @@
  * limitations under the License.
  */
 
+import { createBarrier } from './util';
+
 const DEFAULT_POLLING_INTERVAL_MS = 1000;
 
 type Options<T> = {
@@ -72,50 +74,58 @@ export function startTaskPipeline<T>(options: Options<T>) {
     throw new Error('lowWatermark must be lower than highWatermark');
   }
 
-  let loading = false;
-  let stopped = false;
-  let inFlightCount = 0;
+  // State is in an object so that it can be stably referenced from within
+  // callbacks below
+  const state = { inFlightCount: 0 };
+  const abortController = new AbortController();
+  const abortSignal = abortController.signal;
 
-  async function maybeLoadMore() {
-    if (stopped || loading || inFlightCount > lowWatermark) {
-      return;
-    }
-
-    // Once we hit the low watermark we load in enough items to reach the high watermark
-    loading = true;
-    const loadCount = highWatermark - inFlightCount;
-    const loadedItems = await loadTasks(loadCount);
-    loading = false;
-
-    // We might not reach the high watermark here, in case there weren't enough items to load
-    inFlightCount += loadedItems.length;
-    loadedItems.forEach(item => {
-      processTask(item).finally(() => {
-        if (stopped) {
-          return;
-        }
-
-        // For each item we complete we check if it's time to load more
-        inFlightCount -= 1;
-        maybeLoadMore();
-      });
+  async function pipelineLoop() {
+    const barrier = createBarrier({
+      waitTimeoutMillis: pollingIntervalMs,
+      signal: abortSignal,
     });
 
-    // We might have processed some tasks while we where loading, so check if we can load more
-    if (loadedItems.length > 1) {
-      maybeLoadMore();
+    while (!abortSignal.aborted) {
+      if (state.inFlightCount <= lowWatermark) {
+        const loadCount = highWatermark - state.inFlightCount;
+        const loadedItems = await Promise.resolve()
+          .then(() => loadTasks(loadCount))
+          .catch(() => {
+            // Silently swallow errors and go back to sleep to try again; we
+            // delegate to the loadTasks function itself to catch errors and log
+            // if it so desires
+            return [];
+          });
+        if (loadedItems.length && !abortSignal.aborted) {
+          state.inFlightCount += loadedItems.length;
+          for (const item of loadedItems) {
+            Promise.resolve()
+              .then(() => processTask(item))
+              .catch(() => {
+                // Silently swallow errors and go back to sleep to try again; we
+                // delegate to the processTask function itself to catch errors
+                // and log if it so desires
+              })
+              .finally(() => {
+                state.inFlightCount -= 1;
+                barrier.release();
+              });
+          }
+        }
+      }
+
+      await barrier.wait();
     }
   }
 
-  // This interval makes sure that we load in new items if the loop runs
-  // dry because of the lack of available tasks. As long as there are
-  // enough items to process this will be a noop.
-  const intervalId = setInterval(() => {
-    maybeLoadMore();
-  }, pollingIntervalMs);
+  pipelineLoop().catch(_error => {
+    // This should be impossible, but if it did happen, it would signal a
+    // programming error inside the loop (errors should definitely be caught
+    // inside of it).
+  });
 
   return () => {
-    stopped = true;
-    clearInterval(intervalId);
+    abortController.abort();
   };
 }

plugins/catalog-backend/src/processing/util.test.ts

@@ -0,0 +1,102 @@
+/*
+ * Copyright 2023 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 { createBarrier } from './util';
+
+describe('createBarrier', () => {
+  const tick = (millis: number) =>
+    new Promise(resolve => setTimeout(resolve, millis));
+
+  it('abandons a wait after the timeout expires', async () => {
+    const abortController = new AbortController();
+    const signal = abortController.signal;
+    const barrier = createBarrier({ waitTimeoutMillis: 100, signal });
+
+    const fn1 = jest.fn();
+    barrier.wait().then(fn1);
+
+    await tick(0);
+    expect(fn1).not.toHaveBeenCalled();
+
+    await tick(50);
+    expect(fn1).not.toHaveBeenCalled();
+
+    // start a new wait mid-way through the timeout
+    // should NOT resolve when the first one times out
+    const fn2 = jest.fn();
+    barrier.wait().then(fn2);
+
+    await tick(0);
+    expect(fn2).not.toHaveBeenCalled();
+
+    await tick(50);
+    expect(fn1).toHaveBeenCalledTimes(1);
+    expect(fn2).not.toHaveBeenCalled();
+
+    await tick(50);
+    expect(fn1).toHaveBeenCalledTimes(1);
+    expect(fn2).toHaveBeenCalledTimes(1);
+  });
+
+  it('abandons a wait after aborted', async () => {
+    const abortController = new AbortController();
+    const signal = abortController.signal;
+    const barrier = createBarrier({ waitTimeoutMillis: 100, signal });
+
+    const fn1 = jest.fn();
+    barrier.wait().then(fn1);
+
+    // should resolve immediately, not after timeout
+    await tick(0);
+    expect(fn1).not.toHaveBeenCalled();
+    abortController.abort();
+    await tick(0);
+    expect(fn1).toHaveBeenCalledTimes(1);
+
+    // subsequent waits should be immediate no matter what
+    const fn2 = jest.fn();
+    barrier.wait().then(fn2);
+    await tick(0);
+    expect(fn2).toHaveBeenCalledTimes(1);
+  });
+
+  it('release immediately unblocks all waits', async () => {
+    const abortController = new AbortController();
+    const signal = abortController.signal;
+    const barrier = createBarrier({ waitTimeoutMillis: 100, signal });
+
+    const fn1 = jest.fn();
+    barrier.wait().then(fn1);
+
+    await tick(50);
+    expect(fn1).not.toHaveBeenCalled();
+
+    // start a new wait mid-way through the timeout
+    // SHOULD resolve when releasing
+    const fn2 = jest.fn();
+    barrier.wait().then(fn2);
+
+    await tick(0);
+    expect(fn1).not.toHaveBeenCalled();
+    expect(fn2).not.toHaveBeenCalled();
+
+    barrier.release();
+
+    await tick(0);
+    expect(fn1).toHaveBeenCalledTimes(1);
+    expect(fn2).toHaveBeenCalledTimes(1);
+  });
+});

plugins/catalog-backend/src/processing/util.ts

@@ -86,3 +86,50 @@ export function isObject(value: JsonValue | undefined): value is JsonObject {
 export const validateEntity = entitySchemaValidator();
 
 export const validateEntityEnvelope = entityEnvelopeSchemaValidator();
+
+/**
+ * Creates a barrier with a timeout, that can be awaited or prematurely
+ * released either manually or by an abort signal.
+ */
+export function createBarrier(options: {
+  waitTimeoutMillis: number;
+  signal: AbortSignal;
+}): {
+  wait: () => Promise<void>;
+  release: () => void;
+} {
+  const { waitTimeoutMillis, signal } = options;
+  const resolvers: Array<() => void> = [];
+
+  function wait() {
+    if (signal.aborted || !(waitTimeoutMillis > 0)) {
+      return Promise.resolve();
+    }
+
+    return new Promise<void>(resolve => {
+      const timeoutHandle = setTimeout(done, waitTimeoutMillis);
+
+      function done() {
+        const index = resolvers.indexOf(done);
+        if (index !== -1) {
+          resolvers.splice(index, 1);
+        }
+        clearTimeout(timeoutHandle);
+        resolve();
+      }
+
+      resolvers.push(done);
+    });
+  }
+
+  function release() {
+    const resolversToCall = resolvers.splice(0, resolvers.length);
+    for (const resolver of resolversToCall) {
+      resolver();
+    }
+  }
+
+  signal.addEventListener('abort', release);
+
+  return { wait, release };
+}