Ensure that processing continues despite errors

Co-authored-by: Ben Lambert <ben@blam.sh>

Signed-off-by: Johan Haals <johan.haals@gmail.com>

plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts

@@ -97,4 +97,65 @@ describe('DefaultCatalogProcessingEngine', () => {
     });
     await engine.stop();
   });
+
+  it('should process stuff even if the first attempt fail', async () => {
+    orchestrator.process.mockResolvedValue({
+      ok: true,
+      completedEntity: {
+        apiVersion: '1',
+        kind: 'Location',
+        metadata: { name: 'test' },
+      },
+      relations: [],
+      errors: [],
+      deferredEntities: [],
+      state: new Map(),
+    });
+    const engine = new DefaultCatalogProcessingEngine(
+      getVoidLogger(),
+      [],
+      db,
+      orchestrator,
+      stitcher,
+    );
+
+    db.transaction.mockImplementation(cb => cb((() => {}) as any));
+
+    db.getProcessableEntities
+      .mockImplementation(async () => {
+        await engine.stop();
+        return { items: [] };
+      })
+      .mockRejectedValueOnce(new Error('I FAILED'))
+      .mockResolvedValueOnce({
+        items: [
+          {
+            entityRef: 'foo',
+            id: '1',
+            unprocessedEntity: {
+              apiVersion: '1',
+              kind: 'Location',
+              metadata: { name: 'test' },
+            },
+            state: new Map(),
+            nextUpdateAt: '',
+            lastDiscoveryAt: '',
+          },
+        ],
+      });
+
+    await engine.start();
+    await waitForExpect(() => {
+      expect(orchestrator.process).toBeCalledTimes(1);
+      expect(orchestrator.process).toBeCalledWith({
+        entity: {
+          apiVersion: '1',
+          kind: 'Location',
+          metadata: { name: 'test' },
+        },
+        state: expect.anything(),
+      });
+    });
+    await engine.stop();
+  });
 });

plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts

@@ -83,52 +83,69 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
 
   private async run() {
     while (this.running) {
-      const { items } = await this.processingDatabase.transaction(async tx => {
-        return this.processingDatabase.getProcessableEntities(tx, {
-          processBatchSize: 1,
+      try {
+        // TODO: We want to disconnect the queue popping and message processing
+        // so that if the queue popping fails we exponentially back off in order to give the DB room to sort itself out.
+        await this.process();
+      } catch (e) {
+        this.logger.warn('Processing failed with:', e);
+        // TODO: this can be a little smarter as mentioned in the above comment.
+        // But for now, if something fails, wait a brief time to pick up the next message.
+        await this.wait();
+      }
+    }
+  }
+
+  private async process() {
+    const { items } = await this.processingDatabase.transaction(async tx => {
+      return this.processingDatabase.getProcessableEntities(tx, {
+        processBatchSize: 1,
+      });
+    });
+
+    if (items.length) {
+      const { id, state, unprocessedEntity } = items[0];
+
+      const result = await this.orchestrator.process({
+        entity: unprocessedEntity,
+        state,
+      });
+      for (const error of result.errors) {
+        this.logger.warn(error.message);
+      }
+      if (!result.ok) {
+        return;
+      }
+
+      result.completedEntity.metadata.uid = id;
+      await this.processingDatabase.transaction(async tx => {
+        await this.processingDatabase.updateProcessedEntity(tx, {
+          id,
+          processedEntity: result.completedEntity,
+          state: result.state,
+          errors: JSON.stringify(result.errors),
+          relations: result.relations,
+          deferredEntities: result.deferredEntities,
         });
       });
 
-      if (items.length) {
-        const { id, state, unprocessedEntity } = items[0];
-
-        const result = await this.orchestrator.process({
-          entity: unprocessedEntity,
-          state,
-        });
-        for (const error of result.errors) {
-          this.logger.warn(error.message);
-        }
-        if (!result.ok) {
-          return;
-        }
-
-        result.completedEntity.metadata.uid = id;
-        await this.processingDatabase.transaction(async tx => {
-          await this.processingDatabase.updateProcessedEntity(tx, {
-            id,
-            processedEntity: result.completedEntity,
-            state: result.state,
-            errors: JSON.stringify(result.errors),
-            relations: result.relations,
-            deferredEntities: result.deferredEntities,
-          });
-        });
-
-        const setOfThingsToStitch = new Set<string>([
-          stringifyEntityRef(result.completedEntity),
-          ...result.relations.map(relation =>
-            stringifyEntityRef(relation.source),
-          ),
-        ]);
-        await this.stitcher.stitch(setOfThingsToStitch);
-      } else {
-        // Wait one second before fetching next item.
-        await new Promise<void>(resolve => setTimeout(resolve, 1000));
-      }
+      const setOfThingsToStitch = new Set<string>([
+        stringifyEntityRef(result.completedEntity),
+        ...result.relations.map(relation =>
+          stringifyEntityRef(relation.source),
+        ),
+      ]);
+      await this.stitcher.stitch(setOfThingsToStitch);
+    } else {
+      // No items to process, wait and try again.
+      await this.wait();
     }
   }
 
+  private async wait() {
+    await new Promise<void>(resolve => setTimeout(resolve, 1000));
+  }
+
   async stop() {
     this.running = false;
   }