Merge pull request #17363 from backstage/freben/hack-orphan

Add a formal builtin orphan deletion mechanism

.changeset/odd-falcons-shout.md

@@ -0,0 +1,38 @@
+---
+'@backstage/plugin-catalog-backend': minor
+---
+
+The catalog now has a new, optional `catalog.orphanStrategy` app-config parameter, which can have the string values `'keep'` (default) or `'delete'`.
+
+If set to `'keep'` or left unset, the old behavior is maintained of keeping orphaned entities around until manually deleted.
+
+If set to `'delete'`, the catalog will attempt to automatically clean out orphaned entities without manual intervention. Note that there are no guarantees that this process is instantaneous, so there may be some delay before orphaned items disappear.
+
+For context, the [Life of an Entity](https://backstage.io/docs/features/software-catalog/life-of-an-entity/#orphaning) article goes into some more details on how the nature of orphaning works.
+
+To enable the new behavior, you will need to pass the plugin task scheduler to your catalog backend builder. If your code already looks like this, you don't need to change it:
+
+```ts
+// in packages/backend/src/plugins/catalog.ts
+export default async function createPlugin(
+  env: PluginEnvironment,
+): Promise<Router> {
+  const builder = await CatalogBuilder.create(env);
+```
+
+But if you pass things into the catalog builder one by one, you'll need to add the new field:
+
+```diff
+ // in packages/backend/src/plugins/catalog.ts
+   const builder = await CatalogBuilder.create({
+     // ... other dependencies
++    scheduler: env.scheduler,
+   });
+```
+
+Finally adjust your app-config:
+
+```yaml
+catalog:
+  orphanStrategy: delete
+```

plugins/catalog-backend/api-report.md

@@ -46,6 +46,7 @@ import { PermissionRule } from '@backstage/plugin-permission-node';
 import { PermissionRuleParams } from '@backstage/plugin-permission-common';
 import { PluginDatabaseManager } from '@backstage/backend-common';
 import { PluginEndpointDiscovery } from '@backstage/backend-common';
+import { PluginTaskScheduler } from '@backstage/backend-tasks';
 import { Router } from 'express';
 import { ScmIntegrationRegistry } from '@backstage/integration';
 import { TokenManager } from '@backstage/backend-common';
@@ -172,6 +173,7 @@ export type CatalogEnvironment = {
   config: Config;
   reader: UrlReader;
   permissions: PermissionEvaluator | PermissionAuthorizer;
+  scheduler?: PluginTaskScheduler;
 };
 
 // @public

plugins/catalog-backend/config.d.ts

@@ -81,7 +81,6 @@ export interface Config {
      * be used in combination with static locations to only serve operator
      * provided locations. Effectively this removes the ability to register new
      * components to a running backstage instance.
-     *
      */
     readonly?: boolean;
 
@@ -136,5 +135,12 @@ export interface Config {
         allow: Array<string>;
       }>;
     }>;
+
+    /**
+     * The strategy to use for entities that are orphaned, i.e. no longer have
+     * any other entities or providers referencing them. The default value is
+     * "keep".
+     */
+    orphanStrategy?: 'keep' | 'delete';
   };
 }

plugins/catalog-backend/package.json

@@ -47,6 +47,7 @@
   "dependencies": {
     "@backstage/backend-common": "workspace:^",
     "@backstage/backend-plugin-api": "workspace:^",
+    "@backstage/backend-tasks": "workspace:^",
     "@backstage/catalog-client": "workspace:^",
     "@backstage/catalog-model": "workspace:^",
     "@backstage/config": "workspace:^",

plugins/catalog-backend/src/alpha.ts

@@ -15,7 +15,7 @@
  */
 
 // TODO(Rugvip): Re-exported for alpha types as the API report will otherwise
-//               produce warnings due to the indirect dependency. We be nice to avoid.
+//               produce warnings due to the indirect dependency. Would be nice to avoid.
 import type { EntitiesSearchFilter } from './catalog/types';
 
 export type { /** @alpha */ EntitiesSearchFilter };

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

@@ -16,6 +16,7 @@
 
 import { Entity, stringifyEntityRef } from '@backstage/catalog-model';
 import { ConflictError } from '@backstage/errors';
+import { DeferredEntity } from '@backstage/plugin-catalog-node';
 import { Knex } from 'knex';
 import lodash from 'lodash';
 import type { Logger } from 'winston';
@@ -38,11 +39,10 @@ import {
   UpdateEntityCacheOptions,
   UpdateProcessedEntityOptions,
 } from './types';
-
-import { DeferredEntity } from '@backstage/plugin-catalog-node';
 import { checkLocationKeyConflict } from './operations/refreshState/checkLocationKeyConflict';
 import { insertUnprocessedEntity } from './operations/refreshState/insertUnprocessedEntity';
 import { updateUnprocessedEntity } from './operations/refreshState/updateUnprocessedEntity';
+import { deleteOrphanedEntities } from './operations/util/deleteOrphanedEntities';
 import { generateStableHash } from './util';
 
 // The number of items that are sent per batch to the database layer, when
@@ -269,6 +269,11 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
     return { entityRefs };
   }
 
+  async deleteOrphanedEntities(txOpaque: Transaction): Promise<number> {
+    const tx = txOpaque as Knex.Transaction;
+    return await deleteOrphanedEntities({ tx });
+  }
+
   async transaction<T>(fn: (tx: Transaction) => Promise<T>): Promise<T> {
     try {
       let result: T | undefined = undefined;

plugins/catalog-backend/src/database/operations/util/deleteOrphanedEntities.test.ts

@@ -0,0 +1,196 @@
+/*
+ * Copyright 2023 The Backstage Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { TestDatabaseId, TestDatabases } from '@backstage/backend-test-utils';
+import { Knex } from 'knex';
+import * as uuid from 'uuid';
+import { applyDatabaseMigrations } from '../../migrations';
+import {
+  DbFinalEntitiesRow,
+  DbRefreshStateReferencesRow,
+  DbRefreshStateRow,
+  DbRelationsRow,
+} from '../../tables';
+import { deleteOrphanedEntities } from './deleteOrphanedEntities';
+
+jest.setTimeout(60_000);
+
+describe('deleteOrphanedEntities', () => {
+  const databases = TestDatabases.create({
+    ids: ['MYSQL_8', 'POSTGRES_13', 'POSTGRES_9', 'SQLITE_3'],
+  });
+
+  async function createDatabase(databaseId: TestDatabaseId) {
+    const knex = await databases.init(databaseId);
+    await applyDatabaseMigrations(knex);
+    return knex;
+  }
+
+  async function run(knex: Knex): Promise<number> {
+    let result: number;
+    await knex.transaction(
+      async tx => {
+        // We can't return here, as knex swallows the return type in case the
+        // transaction is rolled back:
+        // https://github.com/knex/knex/blob/e37aeaa31c8ef9c1b07d2e4d3ec6607e557d800d/lib/transaction.js#L136
+        result = await deleteOrphanedEntities({ tx });
+      },
+      {
+        // If we explicitly trigger a rollback, don't fail.
+        doNotRejectOnRollback: true,
+      },
+    );
+    return result!;
+  }
+
+  async function insertEntity(knex: Knex, ...entityRefs: string[]) {
+    for (const ref of entityRefs) {
+      const entityId = uuid.v4();
+      await knex<DbRefreshStateRow>('refresh_state').insert({
+        entity_id: entityId,
+        entity_ref: ref,
+        unprocessed_entity: '{}',
+        processed_entity: '{}',
+        errors: '[]',
+        next_update_at: '2021-04-01 13:37:00',
+        last_discovery_at: '2021-04-01 13:37:00',
+        result_hash: 'original',
+      });
+      await knex<DbFinalEntitiesRow>('final_entities').insert({
+        entity_id: entityId,
+        hash: 'original',
+        stitch_ticket: '',
+      });
+    }
+  }
+
+  async function insertReference(
+    knex: Knex,
+    ...refs: DbRefreshStateReferencesRow[]
+  ) {
+    await knex<DbRefreshStateReferencesRow>('refresh_state_references').insert(
+      refs,
+    );
+  }
+
+  async function insertRelation(knex: Knex, fromRef: string, toRef: string) {
+    const orig = await knex
+      .select('entity_id')
+      .from('refresh_state')
+      .where('entity_ref', fromRef);
+    await knex<DbRelationsRow>('relations').insert({
+      originating_entity_id: orig[0].entity_id,
+      type: 'fake',
+      source_entity_ref: fromRef,
+      target_entity_ref: toRef,
+    });
+  }
+
+  async function refreshState(knex: Knex) {
+    return await knex<DbRefreshStateRow>('refresh_state')
+      .orderBy('entity_ref')
+      .select('entity_ref', 'result_hash');
+  }
+
+  async function finalEntities(knex: Knex) {
+    return await knex<DbFinalEntitiesRow>('final_entities')
+      .join(
+        'refresh_state',
+        'final_entities.entity_id',
+        'refresh_state.entity_id',
+      )
+      .orderBy('refresh_state.entity_ref')
+      .select({
+        entity_ref: 'refresh_state.entity_ref',
+        hash: 'final_entities.hash',
+      });
+  }
+
+  it.each(databases.eachSupportedId())(
+    'works for some mixed paths, %p',
+    async databaseId => {
+      /*
+          In this graph, edges represent refresh state references, not entity relations:
+
+          P1 - E1 -- E2
+                    /
+                  E3
+                 /
+               E4
+                 \
+                  E5
+                 /
+               E6
+                 \
+                  E7
+                 /
+          P2 - E8
+
+          P3 - E9
+
+               E10
+
+          Result: E3, E4, E5, E6, and E10 deleted; others remain
+                  Entities that had relations pointing at orphans are marked for reprocessing
+       */
+      const knex = await createDatabase(databaseId);
+      await insertEntity(
+        knex,
+        'E1',
+        'E2',
+        'E3',
+        'E4',
+        'E5',
+        'E6',
+        'E7',
+        'E8',
+        'E9',
+        'E10',
+      );
+      await insertReference(
+        knex,
+        { source_key: 'P1', target_entity_ref: 'E1' },
+        { source_entity_ref: 'E1', target_entity_ref: 'E2' },
+        { source_entity_ref: 'E3', target_entity_ref: 'E2' },
+        { source_entity_ref: 'E4', target_entity_ref: 'E3' },
+        { source_entity_ref: 'E4', target_entity_ref: 'E5' },
+        { source_entity_ref: 'E6', target_entity_ref: 'E5' },
+        { source_entity_ref: 'E6', target_entity_ref: 'E7' },
+        { source_key: 'P2', target_entity_ref: 'E8' },
+        { source_entity_ref: 'E8', target_entity_ref: 'E7' },
+        { source_key: 'P3', target_entity_ref: 'E9' },
+      );
+      await insertRelation(knex, 'E1', 'E2');
+      await insertRelation(knex, 'E2', 'E3');
+      await insertRelation(knex, 'E10', 'E6');
+      await expect(run(knex)).resolves.toEqual(5);
+      await expect(refreshState(knex)).resolves.toEqual([
+        { entity_ref: 'E1', result_hash: 'original' },
+        { entity_ref: 'E2', result_hash: 'orphan-relation-deleted' },
+        { entity_ref: 'E7', result_hash: 'original' },
+        { entity_ref: 'E8', result_hash: 'original' },
+        { entity_ref: 'E9', result_hash: 'original' },
+      ]);
+      await expect(finalEntities(knex)).resolves.toEqual([
+        { entity_ref: 'E1', hash: 'original' },
+        { entity_ref: 'E2', hash: 'orphan-relation-deleted' },
+        { entity_ref: 'E7', hash: 'original' },
+        { entity_ref: 'E8', hash: 'original' },
+        { entity_ref: 'E9', hash: 'original' },
+      ]);
+    },
+  );
+});

plugins/catalog-backend/src/database/operations/util/deleteOrphanedEntities.ts

@@ -0,0 +1,95 @@
+/*
+ * Copyright 2023 The Backstage Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { Knex } from 'knex';
+import uniq from 'lodash/uniq';
+import { DbFinalEntitiesRow, DbRefreshStateRow } from '../../tables';
+
+/**
+ * Finds and deletes all orphaned entities, i.e. entities that do not have any
+ * incoming references to them, and also eagerly deletes all of their children
+ * that would otherwise become orphaned.
+ */
+export async function deleteOrphanedEntities(options: {
+  tx: Knex.Transaction | Knex;
+}): Promise<number> {
+  const { tx } = options;
+
+  let total = 0;
+
+  // Limit iterations for sanity
+  for (let i = 0; i < 100; ++i) {
+    const candidates = await tx
+      .with('orphans', orphans =>
+        orphans
+          .from<DbRefreshStateRow>('refresh_state')
+          .select('entity_id', 'entity_ref')
+          .whereNotIn('entity_ref', keep =>
+            keep.distinct('target_entity_ref').from('refresh_state_references'),
+          ),
+      )
+      .select({
+        entityId: 'orphans.entity_id',
+        relationSourceId: 'refresh_state.entity_id',
+      })
+      .from('orphans')
+      .leftOuterJoin(
+        'relations',
+        'relations.target_entity_ref',
+        'orphans.entity_ref',
+      )
+      .leftOuterJoin(
+        'refresh_state',
+        'refresh_state.entity_ref',
+        'relations.source_entity_ref',
+      );
+
+    if (!candidates.length) {
+      break;
+    }
+
+    const orphanIds: string[] = uniq(candidates.map(r => r.entityId));
+    const orphanRelationIds: string[] = uniq(
+      candidates.map(r => r.relationSourceId).filter(Boolean),
+    );
+
+    total += orphanIds.length;
+
+    // Delete the orphans themselves
+    await tx
+      .table<DbRefreshStateRow>('refresh_state')
+      .delete()
+      .whereIn('entity_id', orphanIds);
+
+    // Mark all of things that the orphans had relations to for processing and
+    // stitching
+    await tx
+      .table<DbFinalEntitiesRow>('final_entities')
+      .update({
+        hash: 'orphan-relation-deleted',
+      })
+      .whereIn('entity_id', orphanRelationIds);
+    await tx
+      .table<DbRefreshStateRow>('refresh_state')
+      .update({
+        result_hash: 'orphan-relation-deleted',
+        next_update_at: tx.fn.now(),
+      })
+      .whereIn('entity_id', orphanRelationIds);
+  }
+
+  return total;
+}

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

@@ -151,6 +151,8 @@ export interface ProcessingDatabase {
     txOpaque: Transaction,
     options: ListParentsOptions,
   ): Promise<ListParentsResult>;
+
+  deleteOrphanedEntities(txOpaque: Transaction): Promise<number>;
 }
 
 /**

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

@@ -277,18 +277,19 @@ class TestHarness {
       new NoopProgressTracker(),
     );
 
-    const engine = new DefaultCatalogProcessingEngine(
+    const engine = new DefaultCatalogProcessingEngine({
+      config: new ConfigReader({}),
       logger,
       processingDatabase,
       orchestrator,
       stitcher,
-      () => createHash('sha1'),
-      50,
-      event => {
+      createHash: () => createHash('sha1'),
+      pollingIntervalMs: 50,
+      onProcessingError: event => {
         proxyProgressTracker.reportError(event.unprocessedEntity, event.errors);
       },
-      proxyProgressTracker,
-    );
+      tracker: proxyProgressTracker,
+    });
 
     const refresh = new DefaultRefreshService({ database: catalogDatabase });
 

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

@@ -22,6 +22,7 @@ import { DefaultProcessingDatabase } from '../database/DefaultProcessingDatabase
 import { DefaultCatalogProcessingEngine } from './DefaultCatalogProcessingEngine';
 import { CatalogProcessingOrchestrator } from './types';
 import { Stitcher } from '../stitching/Stitcher';
+import { ConfigReader } from '@backstage/config';
 
 describe('DefaultCatalogProcessingEngine', () => {
   const db = {
@@ -61,13 +62,14 @@ describe('DefaultCatalogProcessingEngine', () => {
       state: {},
       refreshKeys: [],
     });
-    const engine = new DefaultCatalogProcessingEngine(
-      getVoidLogger(),
-      db,
-      orchestrator,
-      stitcher,
-      () => hash,
-    );
+    const engine = new DefaultCatalogProcessingEngine({
+      config: new ConfigReader({}),
+      logger: getVoidLogger(),
+      processingDatabase: db,
+      orchestrator: orchestrator,
+      stitcher: stitcher,
+      createHash: () => hash,
+    });
 
     db.transaction.mockImplementation(cb => cb((() => {}) as any));
 
@@ -127,13 +129,14 @@ describe('DefaultCatalogProcessingEngine', () => {
       state: {},
       refreshKeys: [],
     });
-    const engine = new DefaultCatalogProcessingEngine(
-      getVoidLogger(),
-      db,
-      orchestrator,
-      stitcher,
-      () => hash,
-    );
+    const engine = new DefaultCatalogProcessingEngine({
+      config: new ConfigReader({}),
+      logger: getVoidLogger(),
+      processingDatabase: db,
+      orchestrator: orchestrator,
+      stitcher: stitcher,
+      createHash: () => hash,
+    });
 
     db.transaction.mockImplementation(cb => cb((() => {}) as any));
 
@@ -209,13 +212,14 @@ describe('DefaultCatalogProcessingEngine', () => {
       refreshKeys: [],
     });
 
-    const engine = new DefaultCatalogProcessingEngine(
-      getVoidLogger(),
-      db,
-      orchestrator,
-      stitcher,
-      () => hash,
-    );
+    const engine = new DefaultCatalogProcessingEngine({
+      config: new ConfigReader({}),
+      logger: getVoidLogger(),
+      processingDatabase: db,
+      orchestrator: orchestrator,
+      stitcher: stitcher,
+      createHash: () => hash,
+    });
 
     db.transaction.mockImplementation(cb => cb((() => {}) as any));
 
@@ -284,13 +288,14 @@ describe('DefaultCatalogProcessingEngine', () => {
       errors: [],
     });
 
-    const engine = new DefaultCatalogProcessingEngine(
-      getVoidLogger(),
-      db,
-      orchestrator,
-      stitcher,
-      () => hash,
-    );
+    const engine = new DefaultCatalogProcessingEngine({
+      config: new ConfigReader({}),
+      logger: getVoidLogger(),
+      processingDatabase: db,
+      orchestrator: orchestrator,
+      stitcher: stitcher,
+      createHash: () => hash,
+    });
 
     db.transaction.mockImplementation(cb => cb((() => {}) as any));
 
@@ -341,14 +346,15 @@ describe('DefaultCatalogProcessingEngine', () => {
   });
 
   it('should stitch both the previous and new sources when relations change', async () => {
-    const engine = new DefaultCatalogProcessingEngine(
-      getVoidLogger(),
-      db,
-      orchestrator,
-      stitcher,
-      () => hash,
-      100,
-    );
+    const engine = new DefaultCatalogProcessingEngine({
+      config: new ConfigReader({}),
+      logger: getVoidLogger(),
+      processingDatabase: db,
+      orchestrator: orchestrator,
+      stitcher: stitcher,
+      createHash: () => hash,
+      pollingIntervalMs: 100,
+    });
 
     db.transaction.mockImplementation(cb => cb((() => {}) as any));
 
@@ -454,14 +460,15 @@ describe('DefaultCatalogProcessingEngine', () => {
   });
 
   it('should not stitch sources entities when relations are the same', async () => {
-    const engine = new DefaultCatalogProcessingEngine(
-      getVoidLogger(),
-      db,
-      orchestrator,
-      stitcher,
-      () => hash,
-      100,
-    );
+    const engine = new DefaultCatalogProcessingEngine({
+      config: new ConfigReader({}),
+      logger: getVoidLogger(),
+      processingDatabase: db,
+      orchestrator: orchestrator,
+      stitcher: stitcher,
+      createHash: () => hash,
+      pollingIntervalMs: 100,
+    });
 
     db.transaction.mockImplementation(cb => cb((() => {}) as any));
 
@@ -535,14 +542,15 @@ describe('DefaultCatalogProcessingEngine', () => {
   });
 
   it('should stitch sources entities when new relation of different type added', async () => {
-    const engine = new DefaultCatalogProcessingEngine(
-      getVoidLogger(),
-      db,
-      orchestrator,
-      stitcher,
-      () => hash,
-      100,
-    );
+    const engine = new DefaultCatalogProcessingEngine({
+      config: new ConfigReader({}),
+      logger: getVoidLogger(),
+      processingDatabase: db,
+      orchestrator: orchestrator,
+      stitcher: stitcher,
+      createHash: () => hash,
+      pollingIntervalMs: 100,
+    });
 
     db.transaction.mockImplementation(cb => cb((() => {}) as any));
 
@@ -621,14 +629,15 @@ describe('DefaultCatalogProcessingEngine', () => {
   });
 
   it('should stitch sources entities when relation is removed', async () => {
-    const engine = new DefaultCatalogProcessingEngine(
-      getVoidLogger(),
-      db,
-      orchestrator,
-      stitcher,
-      () => hash,
-      100,
-    );
+    const engine = new DefaultCatalogProcessingEngine({
+      config: new ConfigReader({}),
+      logger: getVoidLogger(),
+      processingDatabase: db,
+      orchestrator: orchestrator,
+      stitcher: stitcher,
+      createHash: () => hash,
+      pollingIntervalMs: 100,
+    });
 
     db.transaction.mockImplementation(cb => cb((() => {}) as any));
 

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

@@ -33,34 +33,85 @@ import {
 } from './types';
 import { Stitcher } from '../stitching/Stitcher';
 import { startTaskPipeline } from './TaskPipeline';
+import { PluginTaskScheduler } from '@backstage/backend-tasks';
+import { Config } from '@backstage/config';
 
 const CACHE_TTL = 5;
 
 export type ProgressTracker = ReturnType<typeof progressTracker>;
 
 export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
+  private readonly config: Config;
+  private readonly scheduler?: PluginTaskScheduler;
+  private readonly logger: Logger;
+  private readonly processingDatabase: ProcessingDatabase;
+  private readonly orchestrator: CatalogProcessingOrchestrator;
+  private readonly stitcher: Stitcher;
+  private readonly createHash: () => Hash;
+  private readonly pollingIntervalMs: number;
+  private readonly orphanCleanupIntervalMs: number;
+  private readonly onProcessingError?: (event: {
+    unprocessedEntity: Entity;
+    errors: Error[];
+  }) => Promise<void> | void;
+  private readonly tracker: ProgressTracker;
+
   private stopFunc?: () => void;
 
-  constructor(
-    private readonly logger: Logger,
-    private readonly processingDatabase: ProcessingDatabase,
-    private readonly orchestrator: CatalogProcessingOrchestrator,
-    private readonly stitcher: Stitcher,
-    private readonly createHash: () => Hash,
-    private readonly pollingIntervalMs: number = 1000,
-    private readonly onProcessingError?: (event: {
+  constructor(options: {
+    config: Config;
+    scheduler?: PluginTaskScheduler;
+    logger: Logger;
+    processingDatabase: ProcessingDatabase;
+    orchestrator: CatalogProcessingOrchestrator;
+    stitcher: Stitcher;
+    createHash: () => Hash;
+    pollingIntervalMs?: number;
+    orphanCleanupIntervalMs?: number;
+    onProcessingError?: (event: {
       unprocessedEntity: Entity;
       errors: Error[];
-    }) => Promise<void> | void,
-    private readonly tracker: ProgressTracker = progressTracker(),
-  ) {}
+    }) => Promise<void> | void;
+    tracker?: ProgressTracker;
+  }) {
+    this.config = options.config;
+    this.scheduler = options.scheduler;
+    this.logger = options.logger;
+    this.processingDatabase = options.processingDatabase;
+    this.orchestrator = options.orchestrator;
+    this.stitcher = options.stitcher;
+    this.createHash = options.createHash;
+    this.pollingIntervalMs = options.pollingIntervalMs ?? 1_000;
+    this.orphanCleanupIntervalMs = options.orphanCleanupIntervalMs ?? 30_000;
+    this.onProcessingError = options.onProcessingError;
+    this.tracker = options.tracker ?? progressTracker();
+
+    this.stopFunc = undefined;
+  }
 
   async start() {
     if (this.stopFunc) {
       throw new Error('Processing engine is already started');
     }
 
-    this.stopFunc = startTaskPipeline<RefreshStateItem>({
+    const stopPipeline = this.startPipeline();
+    const stopCleanup = this.startOrphanCleanup();
+
+    this.stopFunc = () => {
+      stopPipeline();
+      stopCleanup();
+    };
+  }
+
+  async stop() {
+    if (this.stopFunc) {
+      this.stopFunc();
+      this.stopFunc = undefined;
+    }
+  }
+
+  private startPipeline(): () => void {
+    return startTaskPipeline<RefreshStateItem>({
       lowWatermark: 5,
       highWatermark: 10,
       pollingIntervalMs: this.pollingIntervalMs,
@@ -252,11 +303,44 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
     });
   }
 
-  async stop() {
-    if (this.stopFunc) {
-      this.stopFunc();
-      this.stopFunc = undefined;
+  private startOrphanCleanup(): () => void {
+    const strategy =
+      this.config.getOptionalString('catalog.orphanStrategy') ?? 'keep';
+    if (strategy !== 'delete') {
+      return () => {};
     }
+
+    const runOnce = async () => {
+      try {
+        await this.processingDatabase.transaction(async tx => {
+          const n = await this.processingDatabase.deleteOrphanedEntities(tx);
+          this.logger.info(`Deleted ${n} orphaned entities`);
+        });
+      } catch (error) {
+        this.logger.warn(`Failed to delete orphaned entities`, error);
+      }
+    };
+
+    if (this.scheduler) {
+      const abortController = new AbortController();
+
+      this.scheduler.scheduleTask({
+        id: 'catalog_orphan_cleanup',
+        frequency: { milliseconds: this.orphanCleanupIntervalMs },
+        timeout: { milliseconds: this.orphanCleanupIntervalMs * 0.8 },
+        fn: runOnce,
+        signal: abortController.signal,
+      });
+
+      return () => {
+        abortController.abort();
+      };
+    }
+
+    const intervalKey = setInterval(runOnce, this.orphanCleanupIntervalMs);
+    return () => {
+      clearInterval(intervalKey);
+    };
   }
 }
 
@@ -290,9 +374,7 @@ function progressTracker() {
   const meter = metrics.getMeter('default');
   const stitchedEntities = meter.createCounter(
     'catalog.stitched.entities.count',
-    {
-      description: 'Amount of entities stitched',
-    },
+    { description: 'Amount of entities stitched' },
   );
 
   const processedEntities = meter.createCounter(

plugins/catalog-backend/src/service/CatalogBuilder.ts

@@ -15,6 +15,7 @@
  */
 
 import { PluginDatabaseManager, UrlReader } from '@backstage/backend-common';
+import { PluginTaskScheduler } from '@backstage/backend-tasks';
 import {
   DefaultNamespaceEntityPolicy,
   Entity,
@@ -115,6 +116,7 @@ export type CatalogEnvironment = {
   config: Config;
   reader: UrlReader;
   permissions: PermissionEvaluator | PermissionAuthorizer;
+  scheduler?: PluginTaskScheduler;
 };
 
 /**
@@ -424,7 +426,7 @@ export class CatalogBuilder {
     processingEngine: CatalogProcessingEngine;
     router: Router;
   }> {
-    const { config, database, logger, permissions } = this.env;
+    const { config, database, logger, permissions, scheduler } = this.env;
 
     const policy = this.buildEntityPolicy();
     const processors = this.buildProcessors();
@@ -517,17 +519,19 @@ export class CatalogBuilder {
       provider => provider.getProviderName(),
     );
 
-    const processingEngine = new DefaultCatalogProcessingEngine(
+    const processingEngine = new DefaultCatalogProcessingEngine({
+      config,
+      scheduler,
       logger,
       processingDatabase,
       orchestrator,
       stitcher,
-      () => createHash('sha1'),
-      1000,
-      event => {
+      createHash: () => createHash('sha1'),
+      pollingIntervalMs: 1000,
+      onProcessingError: event => {
         this.onProcessingError?.(event);
       },
-    );
+    });
 
     const locationAnalyzer =
       this.locationAnalyzer ??

plugins/catalog-backend/src/service/CatalogPlugin.ts

@@ -76,6 +76,7 @@ export const catalogPlugin = createBackendPlugin({
         database: coreServices.database,
         httpRouter: coreServices.httpRouter,
         lifecycle: coreServices.lifecycle,
+        scheduler: coreServices.scheduler,
       },
       async init({
         logger,
@@ -85,6 +86,7 @@ export const catalogPlugin = createBackendPlugin({
         permissions,
         httpRouter,
         lifecycle,
+        scheduler,
       }) {
         const winstonLogger = loggerToWinstonLogger(logger);
         const builder = await CatalogBuilder.create({
@@ -92,6 +94,7 @@ export const catalogPlugin = createBackendPlugin({
           reader,
           permissions,
           database,
+          scheduler,
           logger: winstonLogger,
         });
         builder.addProcessor(...processingExtensions.processors);

plugins/catalog-backend/src/service/DefaultRefreshService.test.ts

@@ -33,6 +33,7 @@ import { DefaultCatalogProcessingEngine } from '../processing/DefaultCatalogProc
 import { EntityProcessingRequest } from '../processing/types';
 import { Stitcher } from '../stitching/Stitcher';
 import { DefaultRefreshService } from './DefaultRefreshService';
+import { ConfigReader } from '@backstage/config';
 
 jest.setTimeout(60_000);
 
@@ -108,10 +109,11 @@ describe('DefaultRefreshService', () => {
       }
     }
 
-    const engine = new DefaultCatalogProcessingEngine(
-      defaultLogger,
-      db,
-      {
+    const engine = new DefaultCatalogProcessingEngine({
+      config: new ConfigReader({}),
+      logger: defaultLogger,
+      processingDatabase: db,
+      orchestrator: {
         async process(request: EntityProcessingRequest) {
           const entityRef = stringifyEntityRef(request.entity);
           const entity = entityMap.get(entityRef);
@@ -149,10 +151,10 @@ describe('DefaultRefreshService', () => {
           };
         },
       },
-      new Stitcher(knex, defaultLogger),
-      () => createHash('sha1'),
-      50,
-    );
+      stitcher: new Stitcher(knex, defaultLogger),
+      createHash: () => createHash('sha1'),
+      pollingIntervalMs: 50,
+    });
 
     return engine;
   };

plugins/catalog-backend/src/service/createRouter.ts

@@ -52,6 +52,7 @@ import {
 } from './util';
 import type { ApiRouter } from '@backstage/backend-openapi-utils';
 import spec from '../schema/openapi';
+import { PluginTaskScheduler } from '@backstage/backend-tasks';
 
 /**
  * Options used by {@link createRouter}.
@@ -64,6 +65,7 @@ export interface RouterOptions {
   locationService: LocationService;
   orchestrator?: CatalogProcessingOrchestrator;
   refreshService?: RefreshService;
+  scheduler?: PluginTaskScheduler;
   logger: Logger;
   config: Config;
   permissionIntegrationRouter?: express.Router;

yarn.lock

@@ -5436,6 +5436,7 @@ __metadata:
     "@backstage/backend-common": "workspace:^"
     "@backstage/backend-openapi-utils": "workspace:^"
     "@backstage/backend-plugin-api": "workspace:^"
+    "@backstage/backend-tasks": "workspace:^"
     "@backstage/backend-test-utils": "workspace:^"
     "@backstage/catalog-client": "workspace:^"
     "@backstage/catalog-model": "workspace:^"