backend-app-api: implement topological traversal for dependency tree

Signed-off-by: Patrik Oldsberg <poldsberg@gmail.com>

packages/backend-app-api/src/lib/DependencyTree.test.ts

@@ -121,4 +121,79 @@ describe('DependencyTree', () => {
       { id: '4', unsatisfied: ['c'] },
     ]);
   });
+
+  it('should traverse dependencies in topological order', async () => {
+    await expect(
+      DependencyTree.fromMap({
+        1: {},
+        2: {},
+        3: {},
+        4: {},
+      }).parallelTopologicalTraversal(async id => id),
+    ).resolves.toEqual(['1', '2', '3', '4']);
+
+    await expect(
+      DependencyTree.fromMap({
+        1: { produces: ['a'] },
+        2: { consumes: ['a'], produces: ['b', 'c'] },
+        3: { consumes: ['b'] },
+        4: { consumes: ['c'] },
+      }).parallelTopologicalTraversal(async id => id),
+    ).resolves.toEqual(['1', '2', '3', '4']);
+
+    await expect(
+      DependencyTree.fromMap({
+        1: { consumes: ['c'] },
+        2: { produces: ['c'], consumes: ['b'] },
+        3: { produces: ['b'], consumes: ['a'] },
+        4: { produces: ['a'] },
+      }).parallelTopologicalTraversal(async id => id),
+    ).resolves.toEqual(['4', '3', '2', '1']);
+
+    await expect(
+      DependencyTree.fromMap({
+        1: { produces: ['a'] },
+        2: { produces: ['b'], consumes: ['a'] },
+        3: { produces: ['c'], consumes: ['a'] },
+        4: { consumes: ['b'] },
+        5: { consumes: ['c'] },
+      }).parallelTopologicalTraversal(async id => id),
+    ).resolves.toEqual(['1', '2', '3', '4', '5']);
+
+    // Same as above, but with 2 being delayed
+    await expect(
+      DependencyTree.fromMap({
+        1: { produces: ['a'] },
+        2: { produces: ['b'], consumes: ['a'] },
+        3: { produces: ['c'], consumes: ['a'] },
+        4: { consumes: ['b'] },
+        5: { consumes: ['c'] },
+      }).parallelTopologicalTraversal(async id => {
+        // When delaying 2 we expect 3 and 5 to complete before 2 and 4
+        if (id === '2') {
+          await new Promise(resolve => setTimeout(resolve, 100));
+        }
+        return id;
+      }),
+    ).resolves.toEqual(['1', '3', '5', '2', '4']);
+
+    await expect(
+      DependencyTree.fromMap({
+        1: { produces: ['a'], consumes: ['a'] },
+      }).parallelTopologicalTraversal(async id => id),
+    ).rejects.toThrow('Circular dependency detected');
+    await expect(
+      DependencyTree.fromMap({
+        1: { produces: ['a'], consumes: ['b'] },
+        2: { produces: ['b'], consumes: ['a'] },
+      }).parallelTopologicalTraversal(async id => id),
+    ).rejects.toThrow('Circular dependency detected');
+    await expect(
+      DependencyTree.fromMap({
+        1: { produces: ['a'] },
+        2: { produces: ['c'], consumes: ['a', 'b'] },
+        3: { produces: ['b'], consumes: ['a', 'c'] },
+      }).parallelTopologicalTraversal(async id => id),
+    ).rejects.toThrow('Circular dependency detected');
+  });
 });

packages/backend-app-api/src/lib/DependencyTree.ts

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-import { ConflictError, InputError } from '@backstage/errors';
+import { ConflictError, ForwardedError, InputError } from '@backstage/errors';
 
 interface NodeInput {
   id: string;
@@ -136,4 +136,64 @@ export class DependencyTree {
     }
     return undefined;
   }
+
+  async parallelTopologicalTraversal<T>(
+    fn: (nodeId: string) => Promise<T>,
+  ): Promise<T[]> {
+    const allProduced = this.#allProduced;
+    const producedSoFar = new Set<string>();
+    const waiting = new Set(this.nodes.values());
+    const visited = new Set<string>();
+    const results = new Array<T>();
+    let inFlight = 0;
+
+    async function processMoreNodes() {
+      if (waiting.size === 0) {
+        return;
+      }
+      const nodesToProcess = [];
+      for (const node of waiting) {
+        let ready = true;
+        for (const consumed of node.consumes) {
+          if (allProduced.has(consumed) && !producedSoFar.has(consumed)) {
+            ready = false;
+            continue;
+          }
+        }
+        if (ready) {
+          nodesToProcess.push(node);
+        }
+      }
+
+      for (const node of nodesToProcess) {
+        waiting.delete(node);
+      }
+
+      if (nodesToProcess.length === 0 && inFlight === 0) {
+        // We expect the caller to check for circular dependencies before
+        // traversal, so this error should never happen
+        throw new Error('Circular dependency detected');
+      }
+
+      await Promise.all(nodesToProcess.map(processNode));
+    }
+
+    async function processNode(node: Node) {
+      visited.add(node.id);
+      inFlight += 1;
+      try {
+        const result = await fn(node.id);
+        results.push(result);
+      } catch (error) {
+        throw new ForwardedError(`Failed at ${node.id}`, error);
+      }
+      node.produces.forEach(produced => producedSoFar.add(produced));
+      inFlight -= 1;
+      await processMoreNodes();
+    }
+
+    await processMoreNodes();
+
+    return results;
+  }
 }