fix: publish to bus only if l2 write is successful

Author: Julien-R44

packages/bentocache/src/cache/cache_stack.ts

@@ -142,12 +142,19 @@ export class CacheStack extends BaseDriver {
     /**
      * Store the serialized value in the remote cache
      */
+    let l2Success = false
     if (this.l2 && options.skipL2Write !== true) {
       const l2Item = this.options.serializeL1 ? l1Item : this.options.serializer.serialize(rawItem)
-      await this.l2?.set(key, l2Item as any, options)
+      l2Success = await this.l2?.set(key, l2Item as any, options)
+    }
+
+    /**
+     * Publish only if the remote cache write was successful.
+     */
+    if ((this.l2 && l2Success) || !this.l2) {
+      await this.publish({ type: CacheBusMessageType.Set, keys: [key] }, options)
     }
 
-    await this.publish({ type: CacheBusMessageType.Set, keys: [key] }, options)
     this.emit(cacheEvents.written(key, value, this.name))
     return true
   }

packages/bentocache/tests/cache/two_tier.spec.ts

@@ -3,6 +3,7 @@ import { sleep } from '@julr/utils/misc'
 import { MemoryTransport } from '@boringnode/bus/transports/memory'
 
 import { RedisDriver } from '../../src/drivers/redis.js'
+import { ChaosBus } from '../helpers/chaos/chaos_bus.js'
 import { NullDriver } from '../helpers/null/null_driver.js'
 import { ChaosCache } from '../helpers/chaos/chaos_cache.js'
 import { CacheFactory } from '../../factories/cache_factory.js'
@@ -773,4 +774,32 @@ test.group('Cache', () => {
 
     assert.isUndefined(r1)
   })
+
+  test('should not publish to bus if l2 could not be written', async ({ assert }) => {
+    const redis1 = new ChaosCache(new RedisDriver({ connection: REDIS_CREDENTIALS }))
+    const redis2 = new ChaosCache(new RedisDriver({ connection: REDIS_CREDENTIALS }))
+    const bus1 = new ChaosBus(new MemoryTransport())
+    const bus2 = new ChaosBus(new MemoryTransport())
+
+    const [cache1] = new CacheFactory()
+      .merge({ l2Driver: redis1, busDriver: bus1 })
+      .withL1L2Config()
+      .create()
+    const [cache2] = new CacheFactory()
+      .merge({ l2Driver: redis2, busDriver: bus2 })
+      .withL1L2Config()
+      .create()
+
+    await cache1.set({ key: 'foo', value: 'bar' })
+    const r1 = await cache2.get({ key: 'foo' })
+
+    redis1.alwaysThrow()
+    redis2.alwaysThrow()
+
+    await cache1.set({ key: 'foo', value: 'baz' })
+    const r2 = await cache2.get({ key: 'foo' })
+
+    assert.deepEqual(r1, 'bar')
+    assert.deepEqual(r2, 'bar')
+  })
 })

pnpm-lock.yaml

@@ -44,7 +44,7 @@
     "@julr/utils": "^1.8.0",
     "@primevue/themes": "^4.2.5",
     "@vinejs/vine": "^3.0.0",
-    "bentocache": "^1.2.0",
+    "bentocache": "workspace:*",
     "edge.js": "^6.2.1",
     "primeicons": "^7.0.0",
     "primevue": "^4.2.5",

simulator/package.json

@@ -59,17 +59,19 @@ for (let i = 0; i < kNodesCount; i++) {
 }
 
 router.get('/', async ({ inertia }) => {
-  const results = nodes.entries().map(async ([key, cache]) => {
-    return {
-      name: key,
-      result: await cache.bento.get({ key: 'value', defaultValue: 0 }),
-      busId: cache.bus.id,
-    }
-  })
+  const results = await Promise.all(
+    nodes.entries().map(async ([key, cache]) => {
+      return {
+        name: key,
+        result: await cache.bento.get({ key: 'value', defaultValue: 0 }),
+        busId: cache.bus.id,
+      }
+    }),
+  )
 
   return inertia.render('home', {
     correctValue: await trueCache.get({ key: 'value', defaultValue: 0 }),
-    caches: await Promise.all(results),
+    caches: results,
     state,
     sentMessages: [...nodes.entries()]
       .map(([key, cache]) =>