Merge pull request #28527 from backstage/rugvip/procfix

catalog-backend: fix catalog processing not replacing all existing references

.changeset/fair-mangos-sleep.md

@@ -2,4 +2,4 @@
 '@backstage/plugin-catalog-backend': patch
 ---
 
-Cleanup `refresh_state_references` for providers that are no longer in control of a `refresh_state` row for entity
+Cleanup `refresh_state_references` for entity processors and providers that are no longer in control of a `refresh_state` row for entity

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

@@ -383,9 +383,12 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
       }
     }
 
-    // Replace all references for the originating entity or source and then create new ones
+    // Lastly, replace refresh state references for the originating entity and any successfully added entities
     await tx<DbRefreshStateReferencesRow>('refresh_state_references')
-      .andWhere({ source_entity_ref: options.sourceEntityRef })
+      // Remove all existing references from the originating entity
+      .where({ source_entity_ref: options.sourceEntityRef })
+      // And remove any existing references to entities that we're inserting new references for
+      .orWhereIn('target_entity_ref', stateReferences)
       .delete();
     await tx.batchInsert(
       'refresh_state_references',

plugins/catalog-backend/src/tests/integration.test.ts

@@ -31,7 +31,8 @@ import {
 } from '@backstage/plugin-catalog-node';
 import { PermissionEvaluator } from '@backstage/plugin-permission-common';
 import { createHash } from 'crypto';
-import knexFactory, { Knex } from 'knex';
+import { Knex } from 'knex';
+import merge from 'lodash/merge';
 import { EntitiesCatalog } from '../catalog/types';
 import { DefaultCatalogDatabase } from '../database/DefaultCatalogDatabase';
 import { DefaultProcessingDatabase } from '../database/DefaultProcessingDatabase';
@@ -55,7 +56,7 @@ import { mockServices } from '@backstage/backend-test-utils';
 import { LoggerService } from '@backstage/backend-plugin-api';
 import { DatabaseManager } from '@backstage/backend-common';
 import { entitiesResponseToObjects } from '../service/response';
-import { DbRefreshStateReferencesRow } from '../database/tables';
+import { deleteOrphanedEntities } from '../database/operations/util/deleteOrphanedEntities';
 
 const voidLogger = mockServices.logger.mock();
 
@@ -197,6 +198,7 @@ class TestHarness {
   readonly #refresh: RefreshService;
   readonly #provider: TestProvider;
   readonly #proxyProgressTracker: ProxyProgressTracker;
+  readonly #db: Knex;
 
   static async create(options?: {
     disableRelationsCompatibility?: boolean;
@@ -326,6 +328,7 @@ class TestHarness {
       refresh,
       provider,
       proxyProgressTracker,
+      db,
     );
   }
 
@@ -335,12 +338,14 @@ class TestHarness {
     refresh: RefreshService,
     provider: TestProvider,
     proxyProgressTracker: ProxyProgressTracker,
+    db: Knex,
   ) {
     this.#catalog = catalog;
     this.#engine = engine;
     this.#refresh = refresh;
     this.#provider = provider;
     this.#proxyProgressTracker = proxyProgressTracker;
+    this.#db = db;
   }
 
   async process(entityRefs?: Set<string>) {
@@ -382,6 +387,55 @@ class TestHarness {
   async refresh(options: RefreshOptions) {
     return this.#refresh.refresh(options);
   }
+
+  async removeOrphanedEntities() {
+    await deleteOrphanedEntities({
+      knex: this.#db,
+      strategy: { mode: 'immediate' },
+    });
+  }
+
+  async getRefreshState(): Promise<
+    Record<
+      string,
+      {
+        id: string;
+        unprocessedEntity: Entity;
+        processedEntity: Entity;
+        locationKey: string | null;
+      }
+    >
+  > {
+    const result = await this.#db('refresh_state').select('*');
+    return Object.fromEntries(
+      result.map(r => [
+        r.entity_ref,
+        {
+          id: r.entity_id,
+          unprocessedEntity: JSON.parse(r.unprocessed_entity),
+          processedEntity: r.processed_entity
+            ? JSON.parse(r.processed_entity)
+            : undefined,
+          locationKey: r.location_key,
+        },
+      ]),
+    );
+  }
+
+  async getRefreshStateReferences(): Promise<
+    Array<{
+      sourceKey: string | null;
+      sourceEntityRef: string | null;
+      targetEntityRef: string;
+    }>
+  > {
+    const result = await this.#db('refresh_state_references').select('*');
+    return result.map(r => ({
+      sourceKey: r.source_key ?? undefined,
+      sourceEntityRef: r.source_entity_ref ?? undefined,
+      targetEntityRef: r.target_entity_ref,
+    }));
+  }
 }
 
 describe('Catalog Backend Integration', () => {
@@ -847,12 +901,6 @@ describe('Catalog Backend Integration', () => {
   });
 
   it('should replace any refresh_state_references that are dangling after claiming an entityRef with locationKey', async () => {
-    const db = knexFactory({
-      client: 'better-sqlite3',
-      connection: ':memory:',
-      useNullAsDefault: true,
-    });
-
     const firstProvider = new TestProvider();
     firstProvider.getProviderName = () => 'first';
 
@@ -861,7 +909,6 @@ describe('Catalog Backend Integration', () => {
 
     const harness = await TestHarness.create({
       additionalProviders: [firstProvider, secondProvider],
-      db,
     });
 
     await firstProvider.getConnection().applyMutation({
@@ -933,15 +980,11 @@ describe('Catalog Backend Integration', () => {
       }),
     });
 
-    await expect(
-      db<DbRefreshStateReferencesRow>('refresh_state_references')
-        .where({ target_entity_ref: 'component:default/component-1' })
-        .select(),
-    ).resolves.toEqual([
-      expect.objectContaining({
-        source_key: 'second',
-        target_entity_ref: 'component:default/component-1',
-      }),
+    await expect(harness.getRefreshStateReferences()).resolves.toEqual([
+      {
+        sourceKey: 'second',
+        targetEntityRef: 'component:default/component-1',
+      },
     ]);
 
     await secondProvider.getConnection().applyMutation({
@@ -970,11 +1013,12 @@ describe('Catalog Backend Integration', () => {
 
     await expect(harness.process()).resolves.toEqual({});
 
-    await expect(
-      db<DbRefreshStateReferencesRow>('refresh_state_references')
-        .where({ target_entity_ref: 'component:default/component-1' })
-        .select(),
-    ).resolves.toEqual([]);
+    await expect(harness.getRefreshStateReferences()).resolves.toEqual([
+      {
+        sourceKey: 'second',
+        targetEntityRef: 'component:default/component-2',
+      },
+    ]);
 
     await expect(harness.getOutputEntities()).resolves.toEqual({
       'component:default/component-2': expect.objectContaining({
@@ -985,4 +1029,139 @@ describe('Catalog Backend Integration', () => {
       }),
     });
   });
+
+  function withOutputFields(entity: Entity) {
+    return {
+      ...entity,
+      metadata: {
+        ...entity.metadata,
+        etag: expect.any(String),
+        uid: expect.any(String),
+      },
+      relations: [],
+    };
+  }
+
+  it('should fully replace existing entities when emitting override entities during processing', async () => {
+    const baseEntity = {
+      apiVersion: 'backstage.io/v1alpha1',
+      kind: 'Component',
+      metadata: {
+        annotations: {
+          'backstage.io/managed-by-location': 'url:.',
+          'backstage.io/managed-by-origin-location': 'url:.',
+        },
+      },
+    };
+    const entityA = merge({ metadata: { name: 'a' } }, baseEntity);
+    const entityB = merge({ metadata: { name: 'b' } }, baseEntity);
+    const entityBOverride = merge({ metadata: { override: true } }, entityB);
+
+    const processEntity = jest.fn(
+      async (
+        entity: Entity,
+        _location: LocationSpec,
+        _emit: CatalogProcessorEmit,
+      ) => entity,
+    );
+    const harness = await TestHarness.create({ processEntity });
+
+    processEntity.mockImplementation(async (entity, location, emit) => {
+      if (entity.metadata.name === entityA.metadata.name) {
+        emit(processingResult.entity(location, entityBOverride));
+      }
+      return entity;
+    });
+
+    // A and B are added to the catalog, but the processor emits B' from A that overrides B
+    await harness.setInputEntities([entityA, entityB]);
+    await expect(harness.process()).resolves.toEqual({});
+
+    // Expect to find A and B' in the catalog
+    await expect(harness.getRefreshStateReferences()).resolves.toEqual([
+      {
+        sourceKey: 'test',
+        targetEntityRef: 'component:default/a',
+      },
+      {
+        sourceEntityRef: 'component:default/a',
+        targetEntityRef: 'component:default/b',
+      },
+    ]);
+    await expect(harness.getRefreshState()).resolves.toEqual({
+      'component:default/a': expect.objectContaining({
+        locationKey: null,
+        unprocessedEntity: entityA,
+      }),
+      'component:default/b': expect.objectContaining({
+        locationKey: 'url:.',
+        unprocessedEntity: entityBOverride,
+      }),
+    });
+    await expect(harness.getOutputEntities()).resolves.toEqual({
+      'component:default/a': withOutputFields(entityA),
+      'component:default/b': withOutputFields(entityBOverride),
+    });
+
+    // Stop emitting B' from A, then do a full sync with A and B
+    processEntity.mockImplementation(async entity => entity);
+    await harness.setInputEntities([entityA, entityB]);
+
+    // At this point we should still have A and B' in the catalog
+    await expect(harness.getRefreshStateReferences()).resolves.toEqual([
+      {
+        sourceKey: 'test',
+        targetEntityRef: 'component:default/a',
+      },
+      {
+        sourceEntityRef: 'component:default/a',
+        targetEntityRef: 'component:default/b',
+      },
+    ]);
+    await expect(harness.getRefreshState()).resolves.toEqual({
+      'component:default/a': expect.objectContaining({
+        locationKey: null,
+        unprocessedEntity: entityA,
+      }),
+      'component:default/b': expect.objectContaining({
+        locationKey: 'url:.',
+        unprocessedEntity: entityBOverride,
+      }),
+    });
+    await expect(harness.getOutputEntities()).resolves.toEqual({
+      'component:default/a': withOutputFields(entityA),
+      'component:default/b': withOutputFields(entityBOverride),
+    });
+
+    // Once we process, B' should be orphaned
+    await expect(harness.process()).resolves.toEqual({});
+    // This is expected to remove B'
+    await harness.removeOrphanedEntities();
+
+    // At this point only A is left in the catalog
+    await expect(harness.getRefreshStateReferences()).resolves.toEqual([
+      {
+        sourceKey: 'test',
+        targetEntityRef: 'component:default/a',
+      },
+    ]);
+    await expect(harness.getRefreshState()).resolves.toEqual({
+      'component:default/a': expect.objectContaining({
+        locationKey: null,
+        unprocessedEntity: entityA,
+      }),
+    });
+    await expect(harness.getOutputEntities()).resolves.toEqual({
+      'component:default/a': withOutputFields(entityA),
+    });
+
+    // Next time the provider runs and does a full sync we should now be able to add back B
+    await harness.setInputEntities([entityA, entityB]);
+    await expect(harness.process()).resolves.toEqual({});
+
+    await expect(harness.getOutputEntities()).resolves.toEqual({
+      'component:default/a': withOutputFields(entityA),
+      'component:default/b': withOutputFields(entityB),
+    });
+  });
 });