catalog-backend: switch cache to persist items unless new ones are written

Co-authored-by: blam <ben@blam.sh>
Co-authored-by: Johan Haals <johan.haals@gmail.com>
Signed-off-by: Patrik Oldsberg <poldsberg@gmail.com>

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

@@ -61,23 +61,30 @@ 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 generated = (await new Promise<CatalogProcessorResult>(emit =>
-      processor.readLocation(
-        spec,
-        false,
-        emit,
-        defaultEntityDataParser,
-        mockCache,
-      ),
-    )) as CatalogProcessorEntityResult;
+    const emitted = new Array<CatalogProcessorResult>();
+    await processor.readLocation(
+      spec,
+      false,
+      result => emitted.push(result),
+      defaultEntityDataParser,
+      mockCache,
+    );
 
-    expect(generated.type).toBe('entity');
-    expect(generated.location).toEqual(spec);
-    expect(generated.entity).toEqual({ mock: 'entity' });
+    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 () => {
@@ -119,8 +126,7 @@ describe('UrlReaderProcessor', () => {
     expect(generated.entity).toEqual({ mock: 'entity' });
     expect(mockCache.get).toBeCalledWith('v1');
     expect(mockCache.get).toBeCalledTimes(1);
-    expect(mockCache.set).toBeCalledWith('v1', cacheItem);
-    expect(mockCache.set).toBeCalledTimes(1);
+    expect(mockCache.set).toBeCalledTimes(0);
   });
 
   it('should fail load from url with error', async () => {

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

@@ -91,7 +91,6 @@ export class UrlReaderProcessor implements CatalogProcessor {
         for (const parseResult of cacheItem.value) {
           emit(parseResult);
         }
-        await cache.set<CacheItem>(CACHE_KEY, cacheItem);
       } else if (error.name === 'NotFoundError') {
         if (!optional) {
           emit(result.notFoundError(location, message));

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

@@ -137,8 +137,8 @@ export type CatalogProcessorParser = (options: {
  * 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, meaning existing values
- * are removed and need to be set again for them to remain in the cache.
+ * 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
  */

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

@@ -43,10 +43,25 @@ describe('ProcessorCacheManager', () => {
     await expect(processorCache.get<string>('my-key')).resolves.toBe(
       'my-value',
     );
-    processorCache.set('my-key', 'my-new-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(
@@ -56,7 +71,7 @@ describe('ProcessorCacheManager', () => {
     // 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-key'),
+      newCache.forProcessor(myProcessor).get<string>('my-new-key'),
     ).resolves.toBe('my-new-value');
   });
 });

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

@@ -21,12 +21,13 @@ import { isObject } from './util';
 
 class SingleProcessorCache implements CatalogProcessorCache {
   private newState?: JsonObject;
-  constructor(private readonly existingState: JsonObject) {}
+
+  constructor(private readonly existingState?: JsonObject) {}
 
   async get<ItemType extends JsonValue>(
     key: string,
   ): Promise<ItemType | undefined> {
-    return this.existingState[key] as ItemType | undefined;
+    return this.existingState?.[key] as ItemType | undefined;
   }
 
   async set<ItemType extends JsonValue>(
@@ -41,7 +42,7 @@ class SingleProcessorCache implements CatalogProcessorCache {
   }
 
   collect(): JsonObject | undefined {
-    return this.newState;
+    return this.newState ?? this.existingState;
   }
 }
 
@@ -61,7 +62,7 @@ export class ProcessorCacheManager {
     const existing = this.existingState[name];
 
     const newCache = new SingleProcessorCache(
-      isObject(existing) ? existing : {},
+      isObject(existing) ? existing : undefined,
     );
     this.caches.set(name, newCache);
     return newCache;