diff --git a/.changeset/cold-apricots-poke.md b/.changeset/cold-apricots-poke.md new file mode 100644 index 0000000000..b050c8082b --- /dev/null +++ b/.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 diff --git a/plugins/catalog-backend/src/processing/TaskPipeline.ts b/plugins/catalog-backend/src/processing/TaskPipeline.ts index 4afe2408a8..24b4ec5ae3 100644 --- a/plugins/catalog-backend/src/processing/TaskPipeline.ts +++ b/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 = { @@ -72,50 +74,58 @@ export function startTaskPipeline(options: Options) { 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(); }; } diff --git a/plugins/catalog-backend/src/processing/util.test.ts b/plugins/catalog-backend/src/processing/util.test.ts new file mode 100644 index 0000000000..b8d10f7db6 --- /dev/null +++ b/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); + }); +}); diff --git a/plugins/catalog-backend/src/processing/util.ts b/plugins/catalog-backend/src/processing/util.ts index cd93be3c8b..2e1d3a7093 100644 --- a/plugins/catalog-backend/src/processing/util.ts +++ b/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; + release: () => void; +} { + const { waitTimeoutMillis, signal } = options; + const resolvers: Array<() => void> = []; + + function wait() { + if (signal.aborted || !(waitTimeoutMillis > 0)) { + return Promise.resolve(); + } + + return new Promise(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 }; +}