OPDATA-6402 update comparison in CompositeTransport CompareResponseCache (#823)

* OPDATA-6402 update comparison in CompositeTransport CompareResponseCache

* rework placing local cache staleness check inside the CompareResponseCache and leaving shouldUpdate as-is

* remove localCache from CompareResponseCache, remove most of the updated code, falls back to responseCache/redis

* caps

* lint

Author: mmcallister-cll

src/cache/response-cache/compare.ts

@@ -14,10 +14,6 @@ export class CompareResponseCache<
   readonly transportName: string
   // The actual cache where responses are written to
   responseCache: ResponseCache<T>
-  // A local map to keep track of the most recent entries written to the responseCache
-  // We compare with this first before comparing with value in cache
-  // so that we can reduce cache reads
-  localCache: Map<string, AdapterResponse<T['Response']>>
   // True if next should replace current in cache
   shouldUpdate: (
     next: AdapterResponse<T['Response']>,
@@ -41,7 +37,6 @@ export class CompareResponseCache<
     })
     this.transportName = transportName
     this.responseCache = responseCache
-    this.localCache = new Map()
     this.shouldUpdate = shouldUpdate
   }
 
@@ -60,11 +55,8 @@ export class CompareResponseCache<
     const filteredEntries = (
       await Promise.all(
         entries.flatMap(async ({ key, value }) => {
-          if (!this.shouldUpdate(value, this.localCache.get(key))) {
-            return []
-          }
-          const entryInCache = await this.get(key)
-          if (!this.shouldUpdate(value, entryInCache)) {
+          const current = await this.get(key)
+          if (!this.shouldUpdate(value, current)) {
             return []
           }
           return [{ key, value }]
@@ -74,10 +66,6 @@ export class CompareResponseCache<
 
     if (filteredEntries.length > 0) {
       await this.responseCache.writeEntries(filteredEntries)
-
-      filteredEntries.forEach(({ key, value }) => {
-        this.localCache.set(key, value)
-      })
     }
   }
 

src/transports/composite.ts

@@ -28,6 +28,8 @@ export class CompositeTransport<T extends TransportGenerics> implements Transpor
       transportName,
       this.responseCache,
       (next, current) =>
+        // Write if the incoming value has a newer provider timestamp
+        // Also writes if the previous cache entry is stale/expired
         (next.timestamps?.providerIndicatedTimeUnixMs ?? 0) >
         (current?.timestamps?.providerIndicatedTimeUnixMs ?? 0),
     )
@@ -61,6 +63,10 @@ export class CompositeTransport<T extends TransportGenerics> implements Transpor
 
   async backgroundExecute(context: EndpointContext<T>): Promise<void> {
     const entries = Object.entries(this.transports)
+
+    // Note that this will wait for the slowest transport to resolve before completing
+    // this shared backgroundExecute loop and stalling the faster transport(s).
+    // Consider setting lower timeouts on the individual transports if this becomes an issue.
     const results = await Promise.allSettled(entries.map(([, t]) => t.backgroundExecute?.(context)))
 
     results.forEach((r, i) => {

test/cache/response-cache/compare.test.ts

@@ -57,7 +57,7 @@ test('writes under CompareResponseCache transportName', async (t) => {
   t.is(entry?.meta?.transportName, 'ws')
 })
 
-test('second write override first write', async (t) => {
+test('second write overrides first write', async (t) => {
   const compareCache = new CompareResponseCache('merged', buildSimpleCache(), () => true)
 
   const params = { base: 'ETH', factor: 1 }
@@ -81,12 +81,10 @@ test('shouldUpdate can block write when new value is not fresher than cache', as
 
   await compareCache.write('merged', [providerResult(params, 25)])
   t.is((await compareCache.get(compareCache.getCacheKey('merged', params)))?.result, 50)
-  t.is(compareCache.localCache.size, 1)
 })
 
-test('shouldUpdate can block write without old value in localCache', async (t) => {
+test('shouldUpdate checks the backing cache when no prior write through compareCache', async (t) => {
   const simpleCache = buildSimpleCache()
-
   const compareCache = new CompareResponseCache(
     'merged',
     simpleCache,
@@ -95,10 +93,11 @@ test('shouldUpdate can block write without old value in localCache', async (t) =
 
   const params = { base: 'ETH', factor: 1 }
 
+  // Write directly to the backing cache, bypassing compareCache
   await simpleCache.write('merged', [providerResult(params, 100)])
   t.is((await compareCache.get(compareCache.getCacheKey('merged', params)))?.result, 100)
 
+  // Sees the backing cache value and blocks the lower write
   await compareCache.write('merged', [providerResult(params, 25)])
   t.is((await compareCache.get(compareCache.getCacheKey('merged', params)))?.result, 100)
-  t.is(compareCache.localCache.size, 0)
 })

test/transports/composite.test.ts

@@ -140,7 +140,7 @@ test.serial(
 )
 
 test.serial(
-  'composite transport merges child writes by providerIndicatedTimeUnixMs when run under an adapter',
+  'composite transport merges child writes by providerIndicatedTimeUnixMs when run under an adapter, takes newest timestamped value',
   async (t) => {
     axiosMock
       .onGet(`${WS_PROVIDER}/price`, { params: { base: 'BTC', factor: 3 } })
@@ -161,6 +161,50 @@ test.serial(
   },
 )
 
+test.serial(
+  'composite transport refreshes TTL when both transports return same value at same timestamp',
+  async (t) => {
+    axiosMock
+      .onGet(`${WS_PROVIDER}/price`, { params: { base: 'XRP', factor: 1 } })
+      .reply(200, { result: 42, ts: 100 })
+    axiosMock
+      .onGet(`${REST_PROVIDER}/price`, { params: { base: 'XRP', factor: 1 } })
+      .reply(200, { result: 42, ts: 100 })
+
+    const res = await t.context.testAdapter.request({ base: 'XRP', factor: 1 })
+
+    // How to test whether the params were refreshed in the cache?
+    // we can check that the response is correct and then check that the ws transport was called again
+    // on a subsequent request, indicating that the cache entry was refreshed and the ws transport was used again instead of the rest transport
+
+    t.is(res.statusCode, 200)
+    t.is(res.json().result, 42)
+  },
+)
+
+test.serial(
+  'composite transport does not rubberband when transports return conflicting values at same timestamp',
+  async (t) => {
+    // Send first request
+    axiosMock
+      .onGet(`${WS_PROVIDER}/price`, { params: { base: 'ABC', factor: 1 } })
+      .reply(200, { result: 10, ts: 200 })
+
+    const res = await t.context.testAdapter.request({ base: 'ABC', factor: 1 })
+    t.is(res.statusCode, 200)
+    t.is(res.json().result, 10)
+
+    // Send second request with same timestamp but different value
+    axiosMock
+      .onGet(`${REST_PROVIDER}/price`, { params: { base: 'ABC', factor: 1 } })
+      .reply(200, { result: 99, ts: 200 })
+
+    const res2 = await t.context.testAdapter.request({ base: 'ABC', factor: 1 })
+    t.is(res2.statusCode, 200)
+    t.is(res2.json().result, 10)
+  },
+)
+
 class ThrowingTransport<T extends TransportGenerics> implements Transport<T> {
   name!: string
   responseCache!: ResponseCache<T>