From e9d61bcbeba2d202be39b456e3ed46640ad1a91c Mon Sep 17 00:00:00 2001 From: Johan Haals Date: Mon, 3 May 2021 14:30:40 +0200 Subject: [PATCH] Ensure that processing continues despite errors Co-authored-by: Ben Lambert Signed-off-by: Johan Haals --- .../DefaultCatalogProcessingEngine.test.ts | 61 ++++++++++++ .../next/DefaultCatalogProcessingEngine.ts | 97 +++++++++++-------- 2 files changed, 118 insertions(+), 40 deletions(-) diff --git a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts index b6363a697d..376af6e748 100644 --- a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts +++ b/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(); + }); }); diff --git a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts index 6bee54f153..877c3075aa 100644 --- a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts +++ b/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([ - 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(resolve => setTimeout(resolve, 1000)); - } + const setOfThingsToStitch = new Set([ + 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(resolve => setTimeout(resolve, 1000)); + } + async stop() { this.running = false; }