Merge pull request #7266 from backstage/mob/catalog-cache

catalog-backend: add processing cache

.changeset/seven-bulldogs-grin.md

@@ -0,0 +1,11 @@
+---
+'@backstage/plugin-catalog-backend': minor
+---
+
+Introduced a new `CatalogProcessorCache` that is available to catalog processors. It allows arbitrary values to be saved that will then be visible during the next run. The cache is scoped to each individual processor and entity, but is shared across processing steps in a single processor.
+
+The cache is available as a new argument to each of the processing steps, except for `validateEntityKind` and `handleError`.
+
+This also introduces an optional `getProcessorName` to the `CatalogProcessor` interface, which is used to provide a stable identifier for the processor. While it is currently optional it will move to be required in the future.
+
+The breaking part of this change is the modification of the `state` field in the `EntityProcessingRequest` and `EntityProcessingResult` types. This is unlikely to have any impact as the `state` field was previously unused, but could require some minor updates.

plugins/catalog-backend/api-report.md

@@ -302,23 +302,27 @@ export interface CatalogProcessingOrchestrator {
 //
 // @public (undocumented)
 export type CatalogProcessor = {
+  getProcessorName?(): string;
   readLocation?(
     location: LocationSpec,
     optional: boolean,
     emit: CatalogProcessorEmit,
     parser: CatalogProcessorParser,
+    cache: CatalogProcessorCache,
   ): Promise<boolean>;
   preProcessEntity?(
     entity: Entity,
     location: LocationSpec,
     emit: CatalogProcessorEmit,
     originLocation: LocationSpec,
+    cache: CatalogProcessorCache,
   ): Promise<Entity>;
   validateEntityKind?(entity: Entity): Promise<boolean>;
   postProcessEntity?(
     entity: Entity,
     location: LocationSpec,
     emit: CatalogProcessorEmit,
+    cache: CatalogProcessorCache,
   ): Promise<Entity>;
   handleError?(
     error: Error,
@@ -327,6 +331,12 @@ export type CatalogProcessor = {
   ): Promise<void>;
 };
 
+// @public
+export interface CatalogProcessorCache {
+  get<ItemType extends JsonValue>(key: string): Promise<ItemType | undefined>;
+  set<ItemType extends JsonValue>(key: string, value: ItemType): Promise<void>;
+}
+
 // Warning: (ae-missing-release-tag) "CatalogProcessorEmit" is exported by the package, but it is missing a release tag (@alpha, @beta, @public, or @internal)
 //
 // @public (undocumented)
@@ -868,7 +878,7 @@ export type EntityPagination = {
 // @public (undocumented)
 export type EntityProcessingRequest = {
   entity: Entity;
-  state: Map<string, JsonObject>;
+  state?: JsonObject;
 };
 
 // Warning: (ae-missing-release-tag) "EntityProcessingResult" is exported by the package, but it is missing a release tag (@alpha, @beta, @public, or @internal)
@@ -877,7 +887,7 @@ export type EntityProcessingRequest = {
 export type EntityProcessingResult =
   | {
       ok: true;
-      state: Map<string, JsonObject>;
+      state: JsonObject;
       completedEntity: Entity;
       deferredEntities: DeferredEntity[];
       relations: EntityRelationSpec[];
@@ -1478,11 +1488,14 @@ export class UrlReaderProcessor implements CatalogProcessor {
   // Warning: (ae-forgotten-export) The symbol "Options" needs to be exported by the entry point index.d.ts
   constructor(options: Options_3);
   // (undocumented)
+  getProcessorName(): string;
+  // (undocumented)
   readLocation(
     location: LocationSpec,
     optional: boolean,
     emit: CatalogProcessorEmit,
     parser: CatalogProcessorParser,
+    cache: CatalogProcessorCache,
   ): Promise<boolean>;
 }
 
@@ -1505,21 +1518,6 @@ export class UrlReaderProcessor implements CatalogProcessor {
 // src/database/types.d.ts:164:8 - (tsdoc-param-tag-missing-hyphen) The @param block should be followed by a parameter name and then a hyphen
 // src/database/types.d.ts:165:8 - (tsdoc-param-tag-missing-hyphen) The @param block should be followed by a parameter name and then a hyphen
 // src/ingestion/processors/GithubMultiOrgReaderProcessor.d.ts:23:9 - (ae-forgotten-export) The symbol "GithubMultiOrgConfig" needs to be exported by the entry point index.d.ts
-// src/ingestion/processors/types.d.ts:7:8 - (tsdoc-param-tag-missing-hyphen) The @param block should be followed by a parameter name and then a hyphen
-// src/ingestion/processors/types.d.ts:8:8 - (tsdoc-param-tag-missing-hyphen) The @param block should be followed by a parameter name and then a hyphen
-// src/ingestion/processors/types.d.ts:9:8 - (tsdoc-param-tag-missing-hyphen) The @param block should be followed by a parameter name and then a hyphen
-// src/ingestion/processors/types.d.ts:10:8 - (tsdoc-param-tag-missing-hyphen) The @param block should be followed by a parameter name and then a hyphen
-// src/ingestion/processors/types.d.ts:23:8 - (tsdoc-param-tag-missing-hyphen) The @param block should be followed by a parameter name and then a hyphen
-// src/ingestion/processors/types.d.ts:24:8 - (tsdoc-param-tag-missing-hyphen) The @param block should be followed by a parameter name and then a hyphen
-// src/ingestion/processors/types.d.ts:25:8 - (tsdoc-param-tag-missing-hyphen) The @param block should be followed by a parameter name and then a hyphen
-// src/ingestion/processors/types.d.ts:26:8 - (tsdoc-param-tag-missing-hyphen) The @param block should be followed by a parameter name and then a hyphen
-// src/ingestion/processors/types.d.ts:36:8 - (tsdoc-param-tag-missing-hyphen) The @param block should be followed by a parameter name and then a hyphen
-// src/ingestion/processors/types.d.ts:47:8 - (tsdoc-param-tag-missing-hyphen) The @param block should be followed by a parameter name and then a hyphen
-// src/ingestion/processors/types.d.ts:48:8 - (tsdoc-param-tag-missing-hyphen) The @param block should be followed by a parameter name and then a hyphen
-// src/ingestion/processors/types.d.ts:49:8 - (tsdoc-param-tag-missing-hyphen) The @param block should be followed by a parameter name and then a hyphen
-// src/ingestion/processors/types.d.ts:56:8 - (tsdoc-param-tag-missing-hyphen) The @param block should be followed by a parameter name and then a hyphen
-// src/ingestion/processors/types.d.ts:57:8 - (tsdoc-param-tag-missing-hyphen) The @param block should be followed by a parameter name and then a hyphen
-// src/ingestion/processors/types.d.ts:58:8 - (tsdoc-param-tag-missing-hyphen) The @param block should be followed by a parameter name and then a hyphen
 // src/ingestion/types.d.ts:17:8 - (tsdoc-param-tag-missing-hyphen) The @param block should be followed by a parameter name and then a hyphen
 // src/ingestion/types.d.ts:41:8 - (tsdoc-param-tag-missing-hyphen) The @param block should be followed by a parameter name and then a hyphen
 ```

plugins/catalog-backend/src/ingestion/LocationReaders.ts

@@ -52,6 +52,13 @@ type Options = {
   policy: EntityPolicy;
 };
 
+const noopCache = {
+  async get() {
+    return undefined;
+  },
+  async set() {},
+};
+
 /**
  * Implements the reading of a location through a series of processor tasks.
  */
@@ -167,6 +174,7 @@ export class LocationReaders implements LocationReader {
               item.optional,
               validatedEmit,
               this.options.parser,
+              noopCache,
             )
           ) {
             return;
@@ -215,6 +223,7 @@ export class LocationReaders implements LocationReader {
             item.location,
             emit,
             originLocation,
+            noopCache,
           );
         } catch (e) {
           const message = `Processor ${
@@ -285,6 +294,7 @@ export class LocationReaders implements LocationReader {
             current,
             item.location,
             emit,
+            noopCache,
           );
         } catch (e) {
           const message = `Processor ${

plugins/catalog-backend/src/ingestion/processors/UrlReaderProcessor.test.ts

@@ -24,6 +24,7 @@ import { msw } from '@backstage/test-utils';
 import { rest } from 'msw';
 import { setupServer } from 'msw/node';
 import {
+  CatalogProcessorCache,
   CatalogProcessorEntityResult,
   CatalogProcessorErrorResult,
   CatalogProcessorResult,
@@ -33,10 +34,17 @@ import { defaultEntityDataParser } from './util/parse';
 
 describe('UrlReaderProcessor', () => {
   const mockApiOrigin = 'http://localhost';
-
+  const mockCache: jest.Mocked<CatalogProcessorCache> = {
+    get: jest.fn(),
+    set: jest.fn(),
+  };
   const server = setupServer();
   msw.setupDefaultHandlers(server);
 
+  beforeEach(() => {
+    jest.resetAllMocks();
+  });
+
   it('should load from url', async () => {
     const logger = getVoidLogger();
     const reader = UrlReaders.default({
@@ -53,17 +61,72 @@ describe('UrlReaderProcessor', () => {
 
     server.use(
       rest.get(`${mockApiOrigin}/component.yaml`, (_, res, ctx) =>
-        res(ctx.json({ mock: 'entity' })),
+        res(ctx.set({ ETag: 'my-etag' }), ctx.json({ mock: 'entity' })),
       ),
     );
 
+    const emitted = new Array<CatalogProcessorResult>();
+    await processor.readLocation(
+      spec,
+      false,
+      result => emitted.push(result),
+      defaultEntityDataParser,
+      mockCache,
+    );
+
+    expect(emitted.length).toBe(1);
+    expect(emitted[0]).toEqual({
+      type: 'entity',
+      location: spec,
+      entity: { mock: 'entity' },
+    });
+    expect(mockCache.set).toBeCalledWith('v1', {
+      etag: 'my-etag',
+      value: [{ type: 'entity', location: spec, entity: { mock: 'entity' } }],
+    });
+    expect(mockCache.set).toBeCalledTimes(1);
+  });
+
+  it('should use cached data when available', async () => {
+    const logger = getVoidLogger();
+    const reader = UrlReaders.default({
+      logger,
+      config: new ConfigReader({
+        backend: { reading: { allow: [{ host: 'localhost' }] } },
+      }),
+    });
+    server.use(
+      rest.get(`${mockApiOrigin}/component.yaml`, (_, res, ctx) =>
+        res(ctx.status(304)),
+      ),
+    );
+    const spec = {
+      type: 'url',
+      target: `${mockApiOrigin}/component.yaml`,
+    };
+    const cacheItem = {
+      etag: 'my-etag',
+      value: [{ type: 'entity', location: spec, entity: { mock: 'entity' } }],
+    };
+    mockCache.get.mockResolvedValue(cacheItem);
+    const processor = new UrlReaderProcessor({ reader, logger });
+
     const generated = (await new Promise<CatalogProcessorResult>(emit =>
-      processor.readLocation(spec, false, emit, defaultEntityDataParser),
+      processor.readLocation(
+        spec,
+        false,
+        emit,
+        defaultEntityDataParser,
+        mockCache,
+      ),
     )) as CatalogProcessorEntityResult;
 
     expect(generated.type).toBe('entity');
     expect(generated.location).toEqual(spec);
     expect(generated.entity).toEqual({ mock: 'entity' });
+    expect(mockCache.get).toBeCalledWith('v1');
+    expect(mockCache.get).toBeCalledTimes(1);
+    expect(mockCache.set).toBeCalledTimes(0);
   });
 
   it('should fail load from url with error', async () => {
@@ -87,7 +150,13 @@ describe('UrlReaderProcessor', () => {
     );
 
     const generated = (await new Promise<CatalogProcessorResult>(emit =>
-      processor.readLocation(spec, false, emit, defaultEntityDataParser),
+      processor.readLocation(
+        spec,
+        false,
+        emit,
+        defaultEntityDataParser,
+        mockCache,
+      ),
     )) as CatalogProcessorErrorResult;
 
     expect(generated.type).toBe('error');
@@ -116,6 +185,7 @@ describe('UrlReaderProcessor', () => {
       false,
       emit,
       defaultEntityDataParser,
+      mockCache,
     );
 
     expect(reader.search).toBeCalledTimes(1);

plugins/catalog-backend/src/ingestion/processors/UrlReaderProcessor.ts

@@ -15,49 +15,88 @@
  */
 
 import { UrlReader } from '@backstage/backend-common';
-import { LocationSpec } from '@backstage/catalog-model';
+import { Entity, LocationSpec } from '@backstage/catalog-model';
 import parseGitUrl from 'git-url-parse';
 import limiterFactory from 'p-limit';
 import { Logger } from 'winston';
 import * as result from './results';
 import {
   CatalogProcessor,
+  CatalogProcessorCache,
   CatalogProcessorEmit,
+  CatalogProcessorEntityResult,
   CatalogProcessorParser,
+  CatalogProcessorResult,
 } from './types';
 
+const CACHE_KEY = 'v1';
+
 type Options = {
   reader: UrlReader;
   logger: Logger;
 };
 
+// WARNING: If you change this type, you likely need to bump the CACHE_KEY as well
+type CacheItem = {
+  etag: string;
+  value: {
+    type: 'entity';
+    entity: Entity;
+    location: LocationSpec;
+  }[];
+};
+
 export class UrlReaderProcessor implements CatalogProcessor {
   constructor(private readonly options: Options) {}
 
+  getProcessorName() {
+    return 'url-reader';
+  }
+
   async readLocation(
     location: LocationSpec,
     optional: boolean,
     emit: CatalogProcessorEmit,
     parser: CatalogProcessorParser,
+    cache: CatalogProcessorCache,
   ): Promise<boolean> {
     if (location.type !== 'url') {
       return false;
     }
 
+    const cacheItem = await cache.get<CacheItem>(CACHE_KEY);
+
     try {
-      const output = await this.doRead(location.target);
-      for (const item of output) {
+      const { response, etag: newEtag } = await this.doRead(
+        location.target,
+        cacheItem?.etag,
+      );
+
+      const parseResults: CatalogProcessorResult[] = [];
+      for (const item of response) {
         for await (const parseResult of parser({
           data: item.data,
           location: { type: location.type, target: item.url },
         })) {
+          parseResults.push(parseResult);
           emit(parseResult);
         }
       }
+
+      const isOnlyEntities = parseResults.every(r => r.type === 'entity');
+      if (newEtag && isOnlyEntities) {
+        await cache.set<CacheItem>(CACHE_KEY, {
+          etag: newEtag,
+          value: parseResults as CatalogProcessorEntityResult[],
+        });
+      }
     } catch (error) {
       const message = `Unable to read ${location.type}, ${error}`;
-
-      if (error.name === 'NotFoundError') {
+      if (error.name === 'NotModifiedError' && cacheItem) {
+        for (const parseResult of cacheItem.value) {
+          emit(parseResult);
+        }
+      } else if (error.name === 'NotFoundError') {
         if (!optional) {
           emit(result.notFoundError(location, message));
         }
@@ -71,27 +110,31 @@ export class UrlReaderProcessor implements CatalogProcessor {
 
   private async doRead(
     location: string,
-  ): Promise<{ data: Buffer; url: string }[]> {
+    etag?: string,
+  ): Promise<{ response: { data: Buffer; url: string }[]; etag?: string }> {
     // Does it contain globs? I.e. does it contain asterisks or question marks
     // (no curly braces for now)
     const { filepath } = parseGitUrl(location);
     if (filepath?.match(/[*?]/)) {
       const limiter = limiterFactory(5);
-      const response = await this.options.reader.search(location);
+      const response = await this.options.reader.search(location, { etag });
       const output = response.files.map(async file => ({
         url: file.url,
         data: await limiter(file.content),
       }));
-      return Promise.all(output);
+      return { response: await Promise.all(output), etag: response.etag };
     }
 
     // Otherwise do a plain read, prioritizing readUrl if available
     if (this.options.reader.readUrl) {
-      const data = await this.options.reader.readUrl(location);
-      return [{ url: location, data: await data.buffer() }];
+      const data = await this.options.reader.readUrl(location, { etag });
+      return {
+        response: [{ url: location, data: await data.buffer() }],
+        etag: data.etag,
+      };
     }
 
     const data = await this.options.reader.read(location);
-    return [{ url: location, data }];
+    return { response: [{ url: location, data }] };
   }
 }

plugins/catalog-backend/src/ingestion/processors/types.ts

@@ -19,16 +19,24 @@ import {
   EntityRelationSpec,
   LocationSpec,
 } from '@backstage/catalog-model';
+import { JsonValue } from '@backstage/config';
 
 export type CatalogProcessor = {
+  /**
+   * A unique identifier for the Catalog Processor.
+   * It's strongly recommended to implement getProcessorName as this method will be required in the future.
+   */
+  getProcessorName?(): string;
+
   /**
    * Reads the contents of a location.
    *
-   * @param location The location to read
-   * @param optional Whether a missing target should trigger an error
-   * @param emit A sink for items resulting from the read
-   * @param parser A parser, that is able to take the raw catalog descriptor
+   * @param location - The location to read
+   * @param optional - Whether a missing target should trigger an error
+   * @param emit - A sink for items resulting from the read
+   * @param parser - A parser, that is able to take the raw catalog descriptor
    *               data and turn it into the actual result pieces.
+   * @param cache - A cache for storing values local to this processor and the current entity.
    * @returns True if handled by this processor, false otherwise
    */
   readLocation?(
@@ -36,6 +44,7 @@ export type CatalogProcessor = {
     optional: boolean,
     emit: CatalogProcessorEmit,
     parser: CatalogProcessorParser,
+    cache: CatalogProcessorCache,
   ): Promise<boolean>;
 
   /**
@@ -46,12 +55,13 @@ export type CatalogProcessor = {
    * additional data, and the input entity may actually still be incomplete
    * when the processor is invoked.
    *
-   * @param entity The (possibly partial) entity to process
-   * @param location The location that the entity came from
-   * @param emit A sink for auxiliary items resulting from the processing
-   * @param originLocation The location that the entity originally came from.
+   * @param entity - The (possibly partial) entity to process
+   * @param location - The location that the entity came from
+   * @param emit - A sink for auxiliary items resulting from the processing
+   * @param originLocation - The location that the entity originally came from.
    *   While location resolves to the direct parent location, originLocation
    *   tells which location was used to start the ingestion loop.
+   * @param cache - A cache for storing values local to this processor and the current entity.
    * @returns The same entity or a modified version of it
    */
   preProcessEntity?(
@@ -59,13 +69,14 @@ export type CatalogProcessor = {
     location: LocationSpec,
     emit: CatalogProcessorEmit,
     originLocation: LocationSpec,
+    cache: CatalogProcessorCache,
   ): Promise<Entity>;
 
   /**
    * Validates the entity as a known entity kind, after it has been pre-
    * processed and has passed through basic overall validation.
    *
-   * @param entity The entity to validate
+   * @param entity - The entity to validate
    * @returns Resolves to true, if the entity was of a kind that was known and
    *   handled by this processor, and was found to be valid. Resolves to false,
    *   if the entity was not of a kind that was known by this processor.
@@ -77,23 +88,25 @@ export type CatalogProcessor = {
   /**
    * Post-processes an emitted entity, after it has been validated.
    *
-   * @param entity The entity to process
-   * @param location The location that the entity came from
-   * @param emit A sink for auxiliary items resulting from the processing
+   * @param entity - The entity to process
+   * @param location - The location that the entity came from
+   * @param emit - A sink for auxiliary items resulting from the processing
+   * @param cache - A cache for storing values local to this processor and the current entity.
    * @returns The same entity or a modified version of it
    */
   postProcessEntity?(
     entity: Entity,
     location: LocationSpec,
     emit: CatalogProcessorEmit,
+    cache: CatalogProcessorCache,
   ): Promise<Entity>;
 
   /**
    * Handles an emitted error.
    *
-   * @param error The error
-   * @param location The location where the error occurred
-   * @param emit A sink for items resulting from this handling
+   * @param error - The error
+   * @param location - The location where the error occurred
+   * @param emit - A sink for items resulting from this handling
    * @returns Nothing
    */
   handleError?(
@@ -113,6 +126,34 @@ export type CatalogProcessorParser = (options: {
   location: LocationSpec;
 }) => AsyncIterable<CatalogProcessorResult>;
 
+/**
+ * A cache for storing data during processing.
+ *
+ * The values stored in the cache are always local to each processor, meaning
+ * no processor can see cache values from other processors.
+ *
+ * The cache instance provided to the CatalogProcessor is also scoped to the
+ * entity being processed, meaning that each processor run can't see cache
+ * values from processing runs for other entities.
+ *
+ * Values that are set during a processing run will only be visible in the directly
+ * following run. The cache will be overwritten every run unless no new cache items
+ * are written, in which case the existing values remain in the cache.
+ *
+ * @public
+ */
+export interface CatalogProcessorCache {
+  /**
+   * Retrieve a value from the cache.
+   */
+  get<ItemType extends JsonValue>(key: string): Promise<ItemType | undefined>;
+
+  /**
+   * Store a value in the cache.
+   */
+  set<ItemType extends JsonValue>(key: string, value: ItemType): Promise<void>;
+}
+
 export type CatalogProcessorEmit = (generated: CatalogProcessorResult) => void;
 
 export type CatalogProcessorLocationResult = {

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

@@ -28,6 +28,7 @@ describe('DefaultCatalogProcessingEngine', () => {
     transaction: jest.fn(),
     getProcessableEntities: jest.fn(),
     updateProcessedEntity: jest.fn(),
+    updateEntityCache: jest.fn(),
   } as unknown as jest.Mocked<DefaultProcessingDatabase>;
   const orchestrator: jest.Mocked<CatalogProcessingOrchestrator> = {
     process: jest.fn(),
@@ -55,7 +56,7 @@ describe('DefaultCatalogProcessingEngine', () => {
       relations: [],
       errors: [],
       deferredEntities: [],
-      state: new Map(),
+      state: {},
     });
     const engine = new DefaultCatalogProcessingEngine(
       getVoidLogger(),
@@ -84,7 +85,7 @@ describe('DefaultCatalogProcessingEngine', () => {
               metadata: { name: 'test' },
             },
             resultHash: '',
-            state: new Map(),
+            state: [] as any,
             nextUpdateAt: DateTime.now(),
             lastDiscoveryAt: DateTime.now(),
           },
@@ -100,7 +101,7 @@ describe('DefaultCatalogProcessingEngine', () => {
           kind: 'Location',
           metadata: { name: 'test' },
         },
-        state: expect.anything(),
+        state: [], // State is forwarded as is, even if it's a bad format
       });
     });
     await engine.stop();
@@ -117,7 +118,7 @@ describe('DefaultCatalogProcessingEngine', () => {
       relations: [],
       errors: [],
       deferredEntities: [],
-      state: new Map(),
+      state: {},
     });
     const engine = new DefaultCatalogProcessingEngine(
       getVoidLogger(),
@@ -147,7 +148,7 @@ describe('DefaultCatalogProcessingEngine', () => {
               metadata: { name: 'test' },
             },
             resultHash: '',
-            state: new Map(),
+            state: { cache: { myProcessor: { myKey: 'myValue' } } },
             nextUpdateAt: DateTime.now(),
             lastDiscoveryAt: DateTime.now(),
           },
@@ -163,7 +164,7 @@ describe('DefaultCatalogProcessingEngine', () => {
           kind: 'Location',
           metadata: { name: 'test' },
         },
-        state: expect.anything(),
+        state: { cache: { myProcessor: { myKey: 'myValue' } } },
       });
     });
     await engine.stop();
@@ -181,7 +182,7 @@ describe('DefaultCatalogProcessingEngine', () => {
       entityRef: '',
       unprocessedEntity: entity,
       resultHash: 'the matching hash',
-      state: new Map(),
+      state: {},
       nextUpdateAt: DateTime.now(),
       lastDiscoveryAt: DateTime.now(),
     };
@@ -194,7 +195,7 @@ describe('DefaultCatalogProcessingEngine', () => {
       relations: [],
       errors: [],
       deferredEntities: [],
-      state: new Map(),
+      state: {},
     });
 
     const engine = new DefaultCatalogProcessingEngine(
@@ -221,16 +222,100 @@ describe('DefaultCatalogProcessingEngine', () => {
       expect(hash.digest).toBeCalledTimes(1);
       expect(db.updateProcessedEntity).toBeCalledTimes(1);
     });
+    expect(db.updateEntityCache).not.toHaveBeenCalled();
 
     db.getProcessableEntities
       .mockReset()
-      .mockResolvedValueOnce({ items: [refreshState] })
+      .mockResolvedValueOnce({
+        items: [{ ...refreshState, state: { something: 'different' } }],
+      })
       .mockResolvedValue({ items: [] });
 
     await waitForExpect(() => {
       expect(orchestrator.process).toBeCalledTimes(2);
       expect(hash.digest).toBeCalledTimes(2);
       expect(db.updateProcessedEntity).toBeCalledTimes(1);
+      expect(db.updateEntityCache).toBeCalledTimes(1);
+    });
+    expect(db.updateEntityCache).toHaveBeenCalledWith(expect.anything(), {
+      id: '',
+      state: { ttl: 5 },
+    });
+    await engine.stop();
+  });
+
+  it('should decrease the state ttl if there are errors', async () => {
+    const entity = {
+      apiVersion: '1',
+      kind: 'Location',
+      metadata: { name: 'test' },
+    };
+
+    const refreshState = {
+      id: '',
+      entityRef: '',
+      unprocessedEntity: entity,
+      resultHash: 'the matching hash',
+      state: { some: 'value', ttl: 1 },
+      nextUpdateAt: DateTime.now(),
+      lastDiscoveryAt: DateTime.now(),
+    };
+
+    hash.digest.mockReturnValue('the matching hash');
+
+    orchestrator.process.mockResolvedValue({
+      ok: false,
+      errors: [],
+    });
+
+    const engine = new DefaultCatalogProcessingEngine(
+      getVoidLogger(),
+      [],
+      db,
+      orchestrator,
+      stitcher,
+      () => hash,
+    );
+
+    db.transaction.mockImplementation(cb => cb((() => {}) as any));
+
+    await engine.start();
+
+    db.getProcessableEntities
+      .mockResolvedValueOnce({
+        items: [refreshState],
+      })
+      .mockResolvedValue({ items: [] });
+
+    await waitForExpect(() => {
+      expect(db.updateEntityCache).toBeCalledTimes(1);
+    });
+
+    expect(db.updateEntityCache).toHaveBeenCalledWith(expect.anything(), {
+      id: '',
+      state: { some: 'value', ttl: 0 },
+    });
+
+    // Second run, the TTL should now reach 0 and the cache should be cleared
+    db.getProcessableEntities
+      .mockResolvedValueOnce({
+        items: [
+          {
+            ...refreshState,
+            state: db.updateEntityCache.mock.calls[0][1].state,
+          },
+        ],
+      })
+      .mockResolvedValue({ items: [] });
+
+    db.updateEntityCache.mockReset();
+    await waitForExpect(() => {
+      expect(db.updateEntityCache).toBeCalledTimes(1);
+    });
+
+    expect(db.updateEntityCache).toHaveBeenCalledWith(expect.anything(), {
+      id: '',
+      state: {},
     });
 
     await engine.stop();

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

@@ -38,6 +38,8 @@ import {
   EntityProviderMutation,
 } from './types';
 
+const CACHE_TTL = 5;
+
 class Connection implements EntityProviderConnection {
   readonly validateEntityEnvelope = entityEnvelopeSchemaValidator();
 
@@ -151,6 +153,29 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
 
           track.markProcessorsCompleted(result);
 
+          if (result.ok) {
+            if (stableStringify(state) !== stableStringify(result.state)) {
+              await this.processingDatabase.transaction(async tx => {
+                await this.processingDatabase.updateEntityCache(tx, {
+                  id,
+                  state: {
+                    ttl: CACHE_TTL,
+                    ...result.state,
+                  },
+                });
+              });
+            }
+          } else {
+            const maybeTtl = state?.ttl;
+            const ttl = Number.isInteger(maybeTtl) ? (maybeTtl as number) : 0;
+            await this.processingDatabase.transaction(async tx => {
+              await this.processingDatabase.updateEntityCache(tx, {
+                id,
+                state: ttl > 0 ? { ...state, ttl: ttl - 1 } : {},
+              });
+            });
+          }
+
           for (const error of result.errors) {
             // TODO(freben): Try to extract the location out of the unprocessed
             // entity and add as meta to the log lines
@@ -167,8 +192,7 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
             hashBuilder = hashBuilder
               .update(stableStringify({ ...result.completedEntity }))
               .update(stableStringify([...result.deferredEntities]))
-              .update(stableStringify([...result.relations]))
-              .update(stableStringify(Object.fromEntries(result.state)));
+              .update(stableStringify([...result.relations]));
           }
 
           const resultHash = hashBuilder.digest('hex');
@@ -208,7 +232,6 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
               id,
               processedEntity: result.completedEntity,
               resultHash,
-              state: result.state,
               errors: errorsString,
               relations: result.relations,
               deferredEntities: result.deferredEntities,

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

@@ -39,7 +39,7 @@ describe('DefaultLocationServiceTest', () => {
       store.listLocations.mockResolvedValueOnce([]);
       orchestrator.process.mockResolvedValueOnce({
         ok: true,
-        state: new Map(),
+        state: {},
         completedEntity: {
           apiVersion: 'backstage.io/v1alpha1',
           kind: 'Location',
@@ -65,7 +65,7 @@ describe('DefaultLocationServiceTest', () => {
 
       orchestrator.process.mockResolvedValueOnce({
         ok: true,
-        state: new Map(),
+        state: {},
         completedEntity: {
           apiVersion: 'backstage.io/v1alpha1',
           kind: 'Component',
@@ -124,7 +124,7 @@ describe('DefaultLocationServiceTest', () => {
       };
       orchestrator.process.mockResolvedValueOnce({
         ok: true,
-        state: new Map(),
+        state: {},
         completedEntity: {
           apiVersion: 'backstage.io/v1alpha1',
           kind: 'Component',
@@ -151,7 +151,7 @@ describe('DefaultLocationServiceTest', () => {
     it('should return exists false when the location does not exist beforehand', async () => {
       orchestrator.process.mockResolvedValueOnce({
         ok: true,
-        state: new Map(),
+        state: {},
         completedEntity: {
           apiVersion: 'backstage.io/v1alpha1',
           kind: 'Component',

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

@@ -87,7 +87,6 @@ export class DefaultLocationService implements LocationService {
       { entity, locationKey: `${spec.type}:${spec.target}` },
     ];
     const entities: Entity[] = [];
-    const state = new Map(); // ignored
     while (unprocessedEntities.length) {
       const currentEntity = unprocessedEntities.pop();
       if (!currentEntity) {
@@ -95,7 +94,7 @@ export class DefaultLocationService implements LocationService {
       }
       const processed = await this.orchestrator.process({
         entity: currentEntity.entity,
-        state,
+        state: {}, // we process without the existing cache
       });
 
       if (processed.ok) {

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

@@ -138,7 +138,7 @@ describe('Refresh integration', () => {
             relations: [],
             errors: [],
             deferredEntities,
-            state: new Map(),
+            state: {},
           };
         },
       },

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

@@ -17,7 +17,6 @@
 import { getVoidLogger } from '@backstage/backend-common';
 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 * as uuid from 'uuid';
 import { Logger } from 'winston';
@@ -213,7 +212,6 @@ describe('Default Processing Database', () => {
               id,
               processedEntity,
               resultHash: '',
-              state: new Map<string, JsonObject>(),
               relations: [],
               deferredEntities: [],
             }),
@@ -232,7 +230,6 @@ describe('Default Processing Database', () => {
           id,
           processedEntity,
           resultHash: '',
-          state: new Map<string, JsonObject>(),
           relations: [],
           deferredEntities: [],
           locationKey: 'key',
@@ -285,15 +282,11 @@ describe('Default Processing Database', () => {
           last_discovery_at: '2021-04-01 13:37:00',
         });
 
-        const state = new Map<string, JsonObject>();
-        state.set('hello', { t: 'something' });
-
         await db.transaction(tx =>
           db.updateProcessedEntity(tx, {
             id,
             processedEntity,
             resultHash: '',
-            state,
             relations: [],
             deferredEntities: [],
             locationKey: 'key',
@@ -308,9 +301,6 @@ describe('Default Processing Database', () => {
         expect(entities[0].processed_entity).toEqual(
           JSON.stringify(processedEntity),
         );
-        expect(entities[0].cache).toEqual(
-          JSON.stringify(Object.fromEntries(state)),
-        );
         expect(entities[0].errors).toEqual("['something broke']");
         expect(entities[0].location_key).toEqual('key');
       },
@@ -352,7 +342,6 @@ describe('Default Processing Database', () => {
             id,
             processedEntity,
             resultHash: '',
-            state: new Map<string, JsonObject>(),
             relations: relations,
             deferredEntities: [],
           }),
@@ -404,7 +393,6 @@ describe('Default Processing Database', () => {
             id,
             processedEntity,
             resultHash: '',
-            state: new Map<string, JsonObject>(),
             relations: [],
             deferredEntities,
           }),
@@ -422,6 +410,54 @@ describe('Default Processing Database', () => {
     );
   });
 
+  describe('updateEntityCache', () => {
+    it.each(databases.eachSupportedId())(
+      'updates the entityCache, %p',
+      async databaseId => {
+        const { knex, db } = await createDatabase(databaseId);
+        const id = '123';
+        await insertRefreshStateRow(knex, {
+          entity_id: id,
+          entity_ref: 'location:default/fakelocation',
+          unprocessed_entity: '{}',
+          processed_entity: '{}',
+          errors: '[]',
+          next_update_at: '2021-04-01 13:37:00',
+          last_discovery_at: '2021-04-01 13:37:00',
+        });
+
+        const state = { hello: { t: 'something' } };
+
+        await db.transaction(tx =>
+          db.updateEntityCache(tx, {
+            id,
+            state,
+          }),
+        );
+
+        const entities = await knex<DbRefreshStateRow>(
+          'refresh_state',
+        ).select();
+        expect(entities.length).toBe(1);
+        expect(entities[0].cache).toEqual(JSON.stringify(state));
+
+        await db.transaction(tx =>
+          db.updateEntityCache(tx, {
+            id,
+            state: undefined,
+          }),
+        );
+
+        const entities2 = await knex<DbRefreshStateRow>(
+          'refresh_state',
+        ).select();
+        expect(entities2.length).toBe(1);
+        expect(entities2[0].cache).toEqual('{}');
+      },
+      60_000,
+    );
+  });
+
   describe('replaceUnprocessedEntities', () => {
     const createLocations = async (db: Knex, entityRefs: string[]) => {
       for (const ref of entityRefs) {

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

@@ -15,7 +15,6 @@
  */
 
 import { Entity, stringifyEntityRef } from '@backstage/catalog-model';
-import { JsonObject } from '@backstage/config';
 import { ConflictError, NotFoundError } from '@backstage/errors';
 import { Knex } from 'knex';
 import lodash from 'lodash';
@@ -41,6 +40,7 @@ import {
   UpdateProcessedEntityOptions,
   ListAncestorsOptions,
   ListAncestorsResult,
+  UpdateEntityCacheOptions,
 } from './types';
 
 // The number of items that are sent per batch to the database layer, when
@@ -70,7 +70,6 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
       id,
       processedEntity,
       resultHash,
-      state,
       errors,
       relations,
       deferredEntities,
@@ -80,7 +79,6 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
       .update({
         processed_entity: JSON.stringify(processedEntity),
         result_hash: resultHash,
-        cache: JSON.stringify(Object.fromEntries(state || [])),
         errors,
         location_key: locationKey,
       })
@@ -141,6 +139,18 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
       .where('entity_id', id);
   }
 
+  async updateEntityCache(
+    txOpaque: Transaction,
+    options: UpdateEntityCacheOptions,
+  ): Promise<void> {
+    const tx = txOpaque as Knex.Transaction;
+    const { id, state } = options;
+
+    await tx<DbRefreshStateRow>('refresh_state')
+      .update({ cache: JSON.stringify(state ?? {}) })
+      .where('entity_id', id);
+  }
+
   private deduplicateRelations(rows: DbRelationsRow[]): DbRelationsRow[] {
     return lodash.uniqBy(
       rows,
@@ -508,9 +518,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
             resultHash: i.result_hash || '',
             nextUpdateAt: timestampToDateTime(i.next_update_at),
             lastDiscoveryAt: timestampToDateTime(i.last_discovery_at),
-            state: i.cache
-              ? JSON.parse(i.cache)
-              : new Map<string, JsonObject>(),
+            state: i.cache ? JSON.parse(i.cache) : undefined,
             errors: i.errors,
             locationKey: i.location_key,
           } as RefreshStateItem),

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

@@ -36,13 +36,17 @@ export type UpdateProcessedEntityOptions = {
   id: string;
   processedEntity: Entity;
   resultHash: string;
-  state?: Map<string, JsonObject>;
   errors?: string;
   relations: EntityRelationSpec[];
   deferredEntities: DeferredEntity[];
   locationKey?: string;
 };
 
+export type UpdateEntityCacheOptions = {
+  id: string;
+  state?: JsonObject;
+};
+
 export type UpdateProcessedEntityErrorsOptions = {
   id: string;
   errors?: string;
@@ -57,7 +61,7 @@ export type RefreshStateItem = {
   resultHash: string;
   nextUpdateAt: DateTime;
   lastDiscoveryAt: DateTime; // remove?
-  state: Map<string, JsonObject>;
+  state?: JsonObject;
   errors?: string;
   locationKey?: string;
 };
@@ -118,6 +122,14 @@ export interface ProcessingDatabase {
     options: UpdateProcessedEntityOptions,
   ): Promise<void>;
 
+  /**
+   * Updates the cache associated with an entity.
+   */
+  updateEntityCache(
+    txOpaque: Transaction,
+    options: UpdateEntityCacheOptions,
+  ): Promise<void>;
+
   /**
    * Updates only the errors of a processed entity
    */

plugins/catalog-backend/src/next/processing/DefaultCatalogProcessingOrchestrator.test.ts

@@ -16,71 +16,227 @@
 
 import { getVoidLogger } from '@backstage/backend-common';
 import {
-  EntityPolicy,
+  Entity,
+  EntityPolicies,
   LocationEntity,
+  LocationSpec,
   LOCATION_ANNOTATION,
   ORIGIN_LOCATION_ANNOTATION,
 } from '@backstage/catalog-model';
-import { ScmIntegrationRegistry } from '@backstage/integration';
+import { ScmIntegrations } from '@backstage/integration';
 import {
   CatalogProcessor,
+  CatalogProcessorCache,
+  CatalogProcessorEmit,
   CatalogProcessorParser,
   results,
 } from '../../ingestion';
 import { CatalogRulesEnforcer } from '../../ingestion/CatalogRules';
 import { DefaultCatalogProcessingOrchestrator } from './DefaultCatalogProcessingOrchestrator';
+import { defaultEntityDataParser } from '../../ingestion/processors/util/parse';
+import { ConfigReader } from '@backstage/config';
+
+class FooBarProcessor implements CatalogProcessor {
+  getProcessorName = () => 'foo-bar';
+
+  async validateEntityKind(entity: Entity) {
+    return entity.kind.toLocaleLowerCase('en-US') === 'foobar';
+  }
+
+  async postProcessEntity(
+    entity: Entity,
+    _location: LocationSpec,
+    emit: CatalogProcessorEmit,
+    cache: CatalogProcessorCache,
+  ) {
+    if (await cache.get('emit')) {
+      emit(
+        results.entity(
+          { type: 'url', target: './new-place' },
+          {
+            apiVersion: 'my-api/v1',
+            kind: 'FooBar',
+            metadata: {
+              name: 'my-new-foo-bar',
+            },
+          },
+        ),
+      );
+      emit(
+        results.relation({
+          type: 'my-type',
+          source: { kind: 'foobar', name: 'my-source', namespace: 'default' },
+          target: { kind: 'foobar', name: 'my-target', namespace: 'default' },
+        }),
+      );
+    }
+    return entity;
+  }
+}
 
 describe('DefaultCatalogProcessingOrchestrator', () => {
-  it('enforces catalog rules', async () => {
-    const entity: LocationEntity = {
-      apiVersion: 'backstage.io/v1beta1',
-      kind: 'Location',
+  describe('basic processing', () => {
+    const entity = {
+      apiVersion: 'my-api/v1',
+      kind: 'FooBar',
       metadata: {
-        name: 'l',
+        name: 'my-foo-bar',
         annotations: {
-          [ORIGIN_LOCATION_ANNOTATION]: 'url:https://example.com/origin.yaml',
-          [LOCATION_ANNOTATION]: 'url:https://example.com/origin.yaml',
+          [LOCATION_ANNOTATION]: 'url:./here',
+          [ORIGIN_LOCATION_ANNOTATION]: 'url:./there',
         },
       },
-      spec: {
-        type: 'url',
-        target: 'http://example.com/entity.yaml',
-      },
-    };
-
-    const processor: jest.Mocked<CatalogProcessor> = {
-      validateEntityKind: jest.fn(async () => true),
-      readLocation: jest.fn(async (_l, _o, emit) => {
-        emit(results.entity({ type: 't', target: 't' }, entity));
-        return true;
-      }),
-    };
-    const integrations: jest.Mocked<ScmIntegrationRegistry> = {} as any;
-    const parser: CatalogProcessorParser = jest.fn();
-    const policy: jest.Mocked<EntityPolicy> = {
-      enforce: jest.fn(async x => x),
-    };
-    const rulesEnforcer: jest.Mocked<CatalogRulesEnforcer> = {
-      isAllowed: jest.fn(),
     };
 
     const orchestrator = new DefaultCatalogProcessingOrchestrator({
-      processors: [processor],
-      integrations,
+      processors: [new FooBarProcessor()],
+      integrations: ScmIntegrations.fromConfig(new ConfigReader({})),
       logger: getVoidLogger(),
-      parser,
-      policy,
-      rulesEnforcer,
+      parser: defaultEntityDataParser,
+      policy: EntityPolicies.allOf([]),
+      rulesEnforcer: { isAllowed: () => true },
     });
 
-    rulesEnforcer.isAllowed.mockReturnValueOnce(true);
-    await expect(
-      orchestrator.process({ entity, state: new Map() }),
-    ).resolves.toEqual(expect.objectContaining({ ok: true }));
+    it('runs a minimal processing', async () => {
+      await expect(orchestrator.process({ entity })).resolves.toEqual({
+        ok: true,
+        completedEntity: entity,
+        deferredEntities: [],
+        errors: [],
+        relations: [],
+        state: {
+          cache: {},
+        },
+      });
+    });
 
-    rulesEnforcer.isAllowed.mockReturnValueOnce(false);
-    await expect(
-      orchestrator.process({ entity, state: new Map() }),
-    ).resolves.toEqual(expect.objectContaining({ ok: false }));
+    it('emits some things', async () => {
+      await expect(
+        orchestrator.process({
+          entity,
+          state: { cache: { 'foo-bar': { emit: true } } },
+        }),
+      ).resolves.toEqual({
+        ok: true,
+        completedEntity: entity,
+        deferredEntities: [
+          {
+            locationKey: 'url:./new-place',
+            entity: {
+              apiVersion: 'my-api/v1',
+              kind: 'FooBar',
+              metadata: {
+                name: 'my-new-foo-bar',
+                annotations: {
+                  [LOCATION_ANNOTATION]: 'url:./new-place',
+                  [ORIGIN_LOCATION_ANNOTATION]: 'url:./there',
+                },
+              },
+            },
+          },
+        ],
+        errors: [],
+        relations: [
+          {
+            type: 'my-type',
+            source: { kind: 'foobar', name: 'my-source', namespace: 'default' },
+            target: { kind: 'foobar', name: 'my-target', namespace: 'default' },
+          },
+        ],
+        state: {
+          cache: { 'foo-bar': { emit: true } },
+        },
+      });
+    });
+
+    it('accepts any state input', async () => {
+      await expect(
+        orchestrator.process({ entity, state: null as any }),
+      ).resolves.toMatchObject({
+        ok: true,
+      });
+      await expect(
+        orchestrator.process({ entity, state: [] as any }),
+      ).resolves.toMatchObject({
+        ok: true,
+      });
+      await expect(
+        orchestrator.process({ entity, state: Symbol() as any }),
+      ).resolves.toMatchObject({
+        ok: true,
+      });
+      await expect(
+        orchestrator.process({ entity, state: undefined }),
+      ).resolves.toMatchObject({
+        ok: true,
+      });
+      await expect(
+        orchestrator.process({ entity, state: 3 as any }),
+      ).resolves.toMatchObject({
+        ok: true,
+      });
+      await expect(
+        orchestrator.process({ entity, state: '}{' as any }),
+      ).resolves.toMatchObject({
+        ok: true,
+      });
+      await expect(
+        orchestrator.process({ entity, state: { cache: null } }),
+      ).resolves.toMatchObject({
+        ok: true,
+      });
+    });
+  });
+
+  describe('rules', () => {
+    it('enforces catalog rules', async () => {
+      const entity: LocationEntity = {
+        apiVersion: 'backstage.io/v1beta1',
+        kind: 'Location',
+        metadata: {
+          name: 'l',
+          annotations: {
+            [ORIGIN_LOCATION_ANNOTATION]: 'url:https://example.com/origin.yaml',
+            [LOCATION_ANNOTATION]: 'url:https://example.com/origin.yaml',
+          },
+        },
+        spec: {
+          type: 'url',
+          target: 'http://example.com/entity.yaml',
+        },
+      };
+
+      const integrations = ScmIntegrations.fromConfig(new ConfigReader({}));
+      const processor: jest.Mocked<CatalogProcessor> = {
+        validateEntityKind: jest.fn(async () => true),
+        readLocation: jest.fn(async (_l, _o, emit) => {
+          emit(results.entity({ type: 't', target: 't' }, entity));
+          return true;
+        }),
+      };
+      const parser: CatalogProcessorParser = jest.fn();
+      const rulesEnforcer: jest.Mocked<CatalogRulesEnforcer> = {
+        isAllowed: jest.fn(),
+      };
+
+      const orchestrator = new DefaultCatalogProcessingOrchestrator({
+        processors: [processor],
+        integrations,
+        logger: getVoidLogger(),
+        parser,
+        policy: EntityPolicies.allOf([]),
+        rulesEnforcer,
+      });
+
+      rulesEnforcer.isAllowed.mockReturnValueOnce(true);
+      await expect(
+        orchestrator.process({ entity, state: {} }),
+      ).resolves.toEqual(expect.objectContaining({ ok: true }));
+
+      rulesEnforcer.isAllowed.mockReturnValueOnce(false);
+      await expect(
+        orchestrator.process({ entity, state: {} }),
+      ).resolves.toEqual(expect.objectContaining({ ok: false }));
+    });
   });
 });

plugins/catalog-backend/src/next/processing/DefaultCatalogProcessingOrchestrator.ts

@@ -24,6 +24,7 @@ import {
   stringifyLocationReference,
 } from '@backstage/catalog-model';
 import { ConflictError, InputError, NotAllowedError } from '@backstage/errors';
+import { JsonValue } from '@backstage/config';
 import { ScmIntegrationRegistry } from '@backstage/integration';
 import path from 'path';
 import { Logger } from 'winston';
@@ -45,14 +46,17 @@ import {
   toAbsoluteUrl,
   validateEntity,
   validateEntityEnvelope,
+  isObject,
 } from './util';
 import { CatalogRulesEnforcer } from '../../ingestion/CatalogRules';
+import { ProcessorCacheManager } from './ProcessorCacheManager';
 
 type Context = {
   entityRef: string;
   location: LocationSpec;
   originLocation: LocationSpec;
   collector: ProcessorOutputCollector;
+  cache: ProcessorCacheManager;
 };
 
 export class DefaultCatalogProcessingOrchestrator
@@ -72,17 +76,23 @@ export class DefaultCatalogProcessingOrchestrator
   async process(
     request: EntityProcessingRequest,
   ): Promise<EntityProcessingResult> {
-    return this.processSingleEntity(request.entity);
+    return this.processSingleEntity(request.entity, request.state);
   }
 
   private async processSingleEntity(
     unprocessedEntity: Entity,
+    state: JsonValue | undefined,
   ): Promise<EntityProcessingResult> {
     const collector = new ProcessorOutputCollector(
       this.options.logger,
       unprocessedEntity,
     );
 
+    // Cache that is scoped to the entity and processor
+    const cache = new ProcessorCacheManager(
+      isObject(state) && isObject(state.cache) ? state.cache : {},
+    );
+
     try {
       // This will be checked and mutated step by step below
       let entity: Entity = unprocessedEntity;
@@ -108,6 +118,7 @@ export class DefaultCatalogProcessingOrchestrator
         originLocation: parseLocationReference(
           getEntityOriginLocationRef(entity),
         ),
+        cache,
         collector,
       };
 
@@ -145,7 +156,7 @@ export class DefaultCatalogProcessingOrchestrator
       return {
         ...collectorResults,
         completedEntity: entity,
-        state: new Map(),
+        state: { cache: cache.collect() },
         ok: true,
       };
     } catch (error) {
@@ -173,6 +184,7 @@ export class DefaultCatalogProcessingOrchestrator
             context.location,
             context.collector.onEmit,
             context.originLocation,
+            context.cache.forProcessor(processor),
           );
         } catch (e) {
           throw new InputError(
@@ -306,6 +318,7 @@ export class DefaultCatalogProcessingOrchestrator
               false,
               context.collector.onEmit,
               this.options.parser,
+              context.cache.forProcessor(processor, target),
             );
             if (read) {
               didRead = true;
@@ -343,6 +356,7 @@ export class DefaultCatalogProcessingOrchestrator
             result,
             context.location,
             context.collector.onEmit,
+            context.cache.forProcessor(processor),
           );
         } catch (e) {
           throw new InputError(

plugins/catalog-backend/src/next/processing/ProcessorCacheManager.test.ts

@@ -0,0 +1,118 @@
+/*
+ * Copyright 2021 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 { CatalogProcessor } from '../../ingestion/processors';
+import { ProcessorCacheManager } from './ProcessorCacheManager';
+
+class MyProcessor implements CatalogProcessor {
+  getProcessorName = () => 'my-processor';
+}
+
+class OtherProcessor implements CatalogProcessor {}
+
+describe('ProcessorCacheManager', () => {
+  const myProcessor = new MyProcessor();
+  const otherProcessor = new OtherProcessor();
+
+  it('should forward existing state and collect new state', async () => {
+    const cache = new ProcessorCacheManager({
+      'my-processor': { 'my-key': 'my-value' },
+    });
+
+    // Should be empty to begin with
+    expect(cache.collect()).toEqual({});
+
+    // instance should be cached
+    const processorCache = cache.forProcessor(myProcessor);
+    expect(processorCache).toBe(cache.forProcessor(myProcessor));
+
+    // Initial values should be visible, writes should not
+    await expect(processorCache.get<string>('my-key')).resolves.toBe(
+      'my-value',
+    );
+
+    // If set hasn't been called yet we should get the existing data
+    expect(cache.collect()).toEqual({
+      'my-processor': { 'my-key': 'my-value' },
+    });
+
+    processorCache.set('my-new-key', 'my-new-value');
+    // Once set has been called the old values should disappear
+    expect(cache.collect()).toEqual({
+      'my-processor': { 'my-new-key': 'my-new-value' },
+    });
+
+    // Getting the cache should return the initial state value
+    await expect(processorCache.get<string>('my-key')).resolves.toBe(
+      'my-value',
+    );
+    await expect(
+      processorCache.get<string>('my-new-key'),
+    ).resolves.toBeUndefined();
+
+    // There should be isolation between processors
+    await expect(
+      cache.forProcessor(otherProcessor).get<string>('my-key'),
+    ).resolves.toBeUndefined();
+
+    // Collecting the state and passing it to a new manager should make the new values visible
+    const newCache = new ProcessorCacheManager(cache.collect());
+    await expect(
+      newCache.forProcessor(myProcessor).get<string>('my-new-key'),
+    ).resolves.toBe('my-new-value');
+  });
+});
+
+describe('ScopedProcessorCache', () => {
+  const myProcessor = new MyProcessor();
+
+  it('should forward existing state and collect new state', async () => {
+    const cache = new ProcessorCacheManager({
+      'my-processor': { 'scope-1': { 'my-key': 'my-value' } },
+    });
+
+    const scopedCache1 = cache.forProcessor(myProcessor, 'scope-1');
+    const scopedCache2 = cache.forProcessor(myProcessor, 'scope-2');
+
+    // Should be empty to begin with
+    expect(cache.collect()).toEqual({
+      'my-processor': { 'scope-1': { 'my-key': 'my-value' } },
+    });
+
+    await scopedCache2.set('my-new-key-2', 'my-new-value-2');
+
+    expect(cache.collect()).toEqual({
+      'my-processor': {
+        'scope-1': { 'my-key': 'my-value' },
+        'scope-2': { 'my-new-key-2': 'my-new-value-2' },
+      },
+    });
+
+    await scopedCache1.set('my-new-key', 'my-new-value');
+
+    await expect(scopedCache1.get<string>('my-key')).resolves.toBe('my-value');
+    await expect(
+      scopedCache1.get<string>('my-new-key'),
+    ).resolves.toBeUndefined();
+
+    expect(cache.collect()).toEqual({
+      'my-processor': {
+        'scope-1': { 'my-new-key': 'my-new-value' },
+        'scope-2': { 'my-new-key-2': 'my-new-value-2' },
+      },
+    });
+  });
+});

plugins/catalog-backend/src/next/processing/ProcessorCacheManager.ts

@@ -0,0 +1,130 @@
+/*
+ * Copyright 2021 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 { JsonObject, JsonValue } from '@backstage/config';
+import { CatalogProcessor } from '../../ingestion/processors';
+import { CatalogProcessorCache } from '../../ingestion/processors/types';
+import { isObject } from './util';
+
+class SingleProcessorSubCache implements CatalogProcessorCache {
+  private newState?: JsonObject;
+
+  constructor(private readonly existingState?: JsonObject) {}
+
+  async get<ItemType extends JsonValue>(
+    key: string,
+  ): Promise<ItemType | undefined> {
+    return this.existingState?.[key] as ItemType | undefined;
+  }
+
+  async set<ItemType extends JsonValue>(
+    key: string,
+    value: ItemType,
+  ): Promise<void> {
+    if (!this.newState) {
+      this.newState = {};
+    }
+
+    this.newState[key] = value;
+  }
+
+  collect(): JsonObject | undefined {
+    return this.newState ?? this.existingState;
+  }
+}
+
+class SingleProcessorCache implements CatalogProcessorCache {
+  private newState?: JsonObject;
+  private subCaches: Map<string, SingleProcessorSubCache> = new Map();
+
+  constructor(private readonly existingState?: JsonObject) {}
+
+  async get<ItemType extends JsonValue>(
+    key: string,
+  ): Promise<ItemType | undefined> {
+    return this.existingState?.[key] as ItemType | undefined;
+  }
+
+  async set<ItemType extends JsonValue>(
+    key: string,
+    value: ItemType,
+  ): Promise<void> {
+    if (!this.newState) {
+      this.newState = {};
+    }
+
+    this.newState[key] = value;
+  }
+
+  withKey(key: string) {
+    const existingSubCache = this.subCaches.get(key);
+    if (existingSubCache) {
+      return existingSubCache;
+    }
+    const existing = this.existingState?.[key];
+    const subCache = new SingleProcessorSubCache(
+      isObject(existing) ? existing : undefined,
+    );
+    this.subCaches.set(key, subCache);
+    return subCache;
+  }
+
+  collect(): JsonObject | undefined {
+    let obj = this.newState ?? this.existingState;
+    for (const [key, subCache] of this.subCaches) {
+      const subCacheValue = subCache.collect();
+      if (subCacheValue) {
+        obj = { ...obj, [key]: subCacheValue };
+      }
+    }
+    return obj;
+  }
+}
+
+export class ProcessorCacheManager {
+  private caches = new Map<string, SingleProcessorCache>();
+
+  constructor(private readonly existingState: JsonObject) {}
+
+  forProcessor(
+    processor: CatalogProcessor,
+    key?: string,
+  ): CatalogProcessorCache {
+    // constructor name will be deprecated in the future when we make `getProcessorName` required in the implementation
+    const name = processor.getProcessorName?.() ?? processor.constructor.name;
+    const cache = this.caches.get(name);
+    if (cache) {
+      return key ? cache.withKey(key) : cache;
+    }
+
+    const existing = this.existingState[name];
+
+    const newCache = new SingleProcessorCache(
+      isObject(existing) ? existing : undefined,
+    );
+    this.caches.set(name, newCache);
+    return key ? newCache.withKey(key) : newCache;
+  }
+
+  collect(): JsonObject {
+    const result: JsonObject = {};
+    for (const [key, value] of this.caches.entries()) {
+      result[key] = value.collect();
+    }
+
+    return result;
+  }
+}

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

@@ -19,13 +19,13 @@ import { JsonObject } from '@backstage/config';
 
 export type EntityProcessingRequest = {
   entity: Entity;
-  state: Map<string, JsonObject>; // Versions for multiple deployments etc
+  state?: JsonObject; // Versions for multiple deployments etc
 };
 
 export type EntityProcessingResult =
   | {
       ok: true;
-      state: Map<string, JsonObject>;
+      state: JsonObject;
       completedEntity: Entity;
       deferredEntities: DeferredEntity[];
       relations: EntityRelationSpec[];

plugins/catalog-backend/src/next/processing/util.ts

@@ -24,6 +24,7 @@ import {
   ORIGIN_LOCATION_ANNOTATION,
   stringifyEntityRef,
 } from '@backstage/catalog-model';
+import { JsonObject, JsonValue } from '@backstage/config';
 import { InputError } from '@backstage/errors';
 import { ScmIntegrationRegistry } from '@backstage/integration';
 import path from 'path';
@@ -76,6 +77,10 @@ export function toAbsoluteUrl(
   }
 }
 
+export function isObject(value: JsonValue | undefined): value is JsonObject {
+  return typeof value === 'object' && value !== null && !Array.isArray(value);
+}
+
 export const validateEntity = entitySchemaValidator();
 
 export const validateEntityEnvelope = entityEnvelopeSchemaValidator();