add test

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

plugins/catalog-backend/migrations/20210813143113_add_refresh_state_hash.js

@@ -22,7 +22,7 @@
 exports.up = async function up(knex) {
   await knex.schema.alterTable('refresh_state', table => {
     table
-      .text('hash')
+      .text('result_hash')
       .nullable()
       .comment(
         'A hash of the processed contents, used to avoid duplicate work',
@@ -35,6 +35,6 @@ exports.up = async function up(knex) {
  */
 exports.down = async function down(knex) {
   await knex.schema.alterTable('refresh_state', table => {
-    table.dropColumn('hash');
+    table.dropColumn('result_hash');
   });
 };

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

@@ -15,6 +15,7 @@
  */
 
 import { getVoidLogger } from '@backstage/backend-common';
+import { Hash } from 'crypto';
 import waitForExpect from 'wait-for-expect';
 import { DefaultProcessingDatabase } from './database/DefaultProcessingDatabase';
 import { DefaultCatalogProcessingEngine } from './DefaultCatalogProcessingEngine';
@@ -33,6 +34,10 @@ describe('DefaultCatalogProcessingEngine', () => {
   const stitcher = {
     stitch: jest.fn(),
   } as unknown as jest.Mocked<Stitcher>;
+  const hash = {
+    update: () => hash,
+    digest: jest.fn(),
+  } as unknown as jest.Mocked<Hash>;
 
   beforeEach(() => {
     jest.resetAllMocks();
@@ -57,6 +62,7 @@ describe('DefaultCatalogProcessingEngine', () => {
       db,
       orchestrator,
       stitcher,
+      () => hash,
     );
 
     db.transaction.mockImplementation(cb => cb((() => {}) as any));
@@ -76,7 +82,7 @@ describe('DefaultCatalogProcessingEngine', () => {
               kind: 'Location',
               metadata: { name: 'test' },
             },
-            hash: '',
+            resultHash: '',
             state: new Map(),
             nextUpdateAt: '',
             lastDiscoveryAt: '',
@@ -118,6 +124,7 @@ describe('DefaultCatalogProcessingEngine', () => {
       db,
       orchestrator,
       stitcher,
+      () => hash,
     );
 
     db.transaction.mockImplementation(cb => cb((() => {}) as any));
@@ -138,7 +145,7 @@ describe('DefaultCatalogProcessingEngine', () => {
               kind: 'Location',
               metadata: { name: 'test' },
             },
-            hash: '',
+            resultHash: '',
             state: new Map(),
             nextUpdateAt: '',
             lastDiscoveryAt: '',
@@ -160,4 +167,71 @@ describe('DefaultCatalogProcessingEngine', () => {
     });
     await engine.stop();
   });
+
+  it('runs fully when hash mismatches, early-outs when hash matches', async () => {
+    const entity = {
+      apiVersion: '1',
+      kind: 'Location',
+      metadata: { name: 'test' },
+    };
+
+    const refreshState = {
+      id: '',
+      entityRef: '',
+      unprocessedEntity: entity,
+      resultHash: 'the matching hash',
+      state: new Map(),
+      nextUpdateAt: '',
+      lastDiscoveryAt: '',
+    };
+
+    hash.digest.mockReturnValue('the matching hash');
+
+    orchestrator.process.mockResolvedValue({
+      ok: true,
+      completedEntity: entity,
+      relations: [],
+      errors: [],
+      deferredEntities: [],
+      state: new Map(),
+    });
+
+    const engine = new DefaultCatalogProcessingEngine(
+      getVoidLogger(),
+      [],
+      db,
+      orchestrator,
+      stitcher,
+      () => hash,
+    );
+
+    db.transaction.mockImplementation(cb => cb((() => {}) as any));
+
+    db.getProcessableEntities
+      .mockResolvedValueOnce({
+        items: [{ ...refreshState, resultHash: 'NOT RIGHT' }],
+      })
+      .mockResolvedValue({ items: [] });
+
+    await engine.start();
+
+    await waitForExpect(() => {
+      expect(orchestrator.process).toBeCalledTimes(1);
+      expect(hash.digest).toBeCalledTimes(1);
+      expect(db.updateProcessedEntity).toBeCalledTimes(1);
+    });
+
+    db.getProcessableEntities
+      .mockReset()
+      .mockResolvedValueOnce({ items: [refreshState] })
+      .mockResolvedValue({ items: [] });
+
+    await waitForExpect(() => {
+      expect(orchestrator.process).toBeCalledTimes(2);
+      expect(hash.digest).toBeCalledTimes(2);
+      expect(db.updateProcessedEntity).toBeCalledTimes(1);
+    });
+
+    await engine.stop();
+  });
 });

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

@@ -20,7 +20,7 @@ import {
   stringifyEntityRef,
 } from '@backstage/catalog-model';
 import { serializeError } from '@backstage/errors';
-import { createHash } from 'crypto';
+import { Hash } from 'crypto';
 import stableStringify from 'fast-json-stable-stringify';
 import { Logger } from 'winston';
 import { ProcessingDatabase, RefreshStateItem } from './database/types';
@@ -91,6 +91,7 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
     private readonly processingDatabase: ProcessingDatabase,
     private readonly orchestrator: CatalogProcessingOrchestrator,
     private readonly stitcher: Stitcher,
+    private readonly createHash: () => Hash,
   ) {}
 
   async start() {
@@ -133,7 +134,7 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
             unprocessedEntity,
             entityRef,
             locationKey,
-            hash: previousHash,
+            resultHash: previousResultHash,
           } = item;
           const result = await this.orchestrator.process({
             entity: unprocessedEntity,
@@ -151,17 +152,17 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
             result.errors.map(e => serializeError(e)),
           );
 
-          let hashBuilder = createHash('sha1').update(errorsString);
+          let hashBuilder = this.createHash().update(errorsString);
           if (result.ok) {
             hashBuilder = hashBuilder
               .update(stableStringify({ ...result.completedEntity }))
               .update(stableStringify([...result.deferredEntities]))
               .update(stableStringify([...result.relations]))
-              .update(stableStringify({ ...result.state }));
+              .update(stableStringify(Object.fromEntries(result.state)));
           }
 
-          const hash = hashBuilder.digest('hex');
-          if (hash === previousHash) {
+          const resultHash = hashBuilder.digest('hex');
+          if (resultHash === previousResultHash) {
             // If nothing changed in our produced outputs, we cannot have any
             // significant effect on our surroundings; therefore, we just abort
             // without any updates / stitching.
@@ -180,7 +181,7 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
               await this.processingDatabase.updateProcessedEntityErrors(tx, {
                 id,
                 errors: errorsString,
-                hash,
+                resultHash,
               });
             });
             await this.stitcher.stitch(
@@ -194,7 +195,7 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
             await this.processingDatabase.updateProcessedEntity(tx, {
               id,
               processedEntity: result.completedEntity,
-              hash,
+              resultHash,
               state: result.state,
               errors: errorsString,
               relations: result.relations,

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

@@ -31,6 +31,7 @@ import {
 } from '@backstage/catalog-model';
 import { Config } from '@backstage/config';
 import { ScmIntegrations } from '@backstage/integration';
+import { createHash } from 'crypto';
 import lodash from 'lodash';
 import { Logger } from 'winston';
 import {
@@ -310,6 +311,7 @@ export class NextCatalogBuilder {
       processingDatabase,
       orchestrator,
       stitcher,
+      () => createHash('sha1'),
     );
 
     const locationsCatalog = new DatabaseLocationsCatalog(db);

plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts

@@ -19,8 +19,8 @@ import { TestDatabaseId, TestDatabases } from '@backstage/backend-test-utils';
 import { Entity, stringifyEntityRef } from '@backstage/catalog-model';
 import { JsonObject } from '@backstage/config';
 import { Knex } from 'knex';
-import { Logger } from 'winston';
 import * as uuid from 'uuid';
+import { Logger } from 'winston';
 import { DatabaseManager } from './DatabaseManager';
 import { DefaultProcessingDatabase } from './DefaultProcessingDatabase';
 import {
@@ -206,7 +206,7 @@ describe('Default Processing Database', () => {
             db.updateProcessedEntity(tx, {
               id,
               processedEntity,
-              hash: '',
+              resultHash: '',
               state: new Map<string, JsonObject>(),
               relations: [],
               deferredEntities: [],
@@ -225,7 +225,7 @@ describe('Default Processing Database', () => {
         const options = {
           id,
           processedEntity,
-          hash: '',
+          resultHash: '',
           state: new Map<string, JsonObject>(),
           relations: [],
           deferredEntities: [],
@@ -254,7 +254,7 @@ describe('Default Processing Database', () => {
           expect(
             db.updateProcessedEntity(tx, {
               ...options,
-              hash: '',
+              resultHash: '',
               locationKey: 'fail',
             }),
           ).rejects.toThrow(
@@ -286,7 +286,7 @@ describe('Default Processing Database', () => {
           db.updateProcessedEntity(tx, {
             id,
             processedEntity,
-            hash: '',
+            resultHash: '',
             state,
             relations: [],
             deferredEntities: [],
@@ -302,7 +302,9 @@ describe('Default Processing Database', () => {
         expect(entities[0].processed_entity).toEqual(
           JSON.stringify(processedEntity),
         );
-        expect(entities[0].cache).toEqual(JSON.stringify(state));
+        expect(entities[0].cache).toEqual(
+          JSON.stringify(Object.fromEntries(state)),
+        );
         expect(entities[0].errors).toEqual("['something broke']");
         expect(entities[0].location_key).toEqual('key');
       },
@@ -343,7 +345,7 @@ describe('Default Processing Database', () => {
           db.updateProcessedEntity(tx, {
             id,
             processedEntity,
-            hash: '',
+            resultHash: '',
             state: new Map<string, JsonObject>(),
             relations: relations,
             deferredEntities: [],
@@ -395,7 +397,7 @@ describe('Default Processing Database', () => {
           db.updateProcessedEntity(tx, {
             id,
             processedEntity,
-            hash: '',
+            resultHash: '',
             state: new Map<string, JsonObject>(),
             relations: [],
             deferredEntities,

plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts

@@ -60,7 +60,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
     const {
       id,
       processedEntity,
-      hash,
+      resultHash,
       state,
       errors,
       relations,
@@ -70,8 +70,8 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
     const refreshResult = await tx<DbRefreshStateRow>('refresh_state')
       .update({
         processed_entity: JSON.stringify(processedEntity),
-        hash,
-        cache: JSON.stringify(state),
+        result_hash: resultHash,
+        cache: JSON.stringify(Object.fromEntries(state || [])),
         errors,
         location_key: locationKey,
       })
@@ -124,12 +124,12 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
     options: UpdateProcessedEntityOptions,
   ): Promise<void> {
     const tx = txOpaque as Knex.Transaction;
-    const { id, errors, hash } = options;
+    const { id, errors, resultHash } = options;
 
     await tx<DbRefreshStateRow>('refresh_state')
       .update({
         errors,
-        hash,
+        result_hash: resultHash,
       })
       .where('entity_id', id);
   }
@@ -495,7 +495,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
             processedEntity: i.processed_entity
               ? (JSON.parse(i.processed_entity) as Entity)
               : undefined,
-            hash: i.hash || '',
+            resultHash: i.result_hash || '',
             nextUpdateAt: i.next_update_at,
             lastDiscoveryAt: i.last_discovery_at,
             state: i.cache

plugins/catalog-backend/src/next/database/tables.ts

@@ -25,7 +25,7 @@ export type DbRefreshStateRow = {
   entity_ref: string;
   unprocessed_entity: string;
   processed_entity?: string;
-  hash?: string;
+  result_hash?: string;
   cache?: string;
   next_update_at: string;
   last_discovery_at: string; // remove?

plugins/catalog-backend/src/next/database/types.ts

@@ -34,7 +34,7 @@ export type AddUnprocessedEntitiesResult = {};
 export type UpdateProcessedEntityOptions = {
   id: string;
   processedEntity: Entity;
-  hash: string;
+  resultHash: string;
   state?: Map<string, JsonObject>;
   errors?: string;
   relations: EntityRelationSpec[];
@@ -45,7 +45,7 @@ export type UpdateProcessedEntityOptions = {
 export type UpdateProcessedEntityErrorsOptions = {
   id: string;
   errors?: string;
-  hash: string;
+  resultHash: string;
 };
 
 export type RefreshStateItem = {
@@ -53,7 +53,7 @@ export type RefreshStateItem = {
   entityRef: string;
   unprocessedEntity: Entity;
   processedEntity?: Entity;
-  hash: string;
+  resultHash: string;
   nextUpdateAt: string;
   lastDiscoveryAt: string; // remove?
   state: Map<string, JsonObject>;