feat: add circuit breaker system for l2 cache

Author: Julien-R44

.changeset/good-candles-trade.md

@@ -0,0 +1,9 @@
+---
+'bentocache': minor
+---
+
+Added a super simple circuit breaker system to the L2 Cache : 
+- a `l2CircuitBreakerDuration` parameter to set the duration of the circuit breaker. How many seconds the circuit breaker will stay open.
+- If defined, the circuit breaker will open when a call to our distributed cache fails. It will stay open for `l2CircuitBreakerDuration` seconds. 
+
+We may introduce more sophisticated circuit breaker system in the future, but for now, this simple system should be enough.

packages/bentocache/factories/cache_factory.ts

@@ -32,16 +32,7 @@ export class CacheFactory {
    */
   #createCacheStack() {
     const options = new BentoCacheOptions({
-      ttl: this.#parameters.ttl,
-      grace: this.#parameters.grace,
-      graceBackoff: this.#parameters.graceBackoff,
-      timeout: this.#parameters.timeout,
-      hardTimeout: this.#parameters.hardTimeout,
-      logger: this.#parameters.logger,
-      emitter: this.#parameters.emitter,
-      lockTimeout: this.#parameters.lockTimeout,
-      serializer: this.#parameters.serializer,
-      onFactoryError: this.#parameters.onFactoryError,
+      ...this.#parameters,
     }).serializeL1Cache(this.#l1Options.serialize ?? true)
 
     const stack = new CacheStack('primary', options, {

packages/bentocache/src/bento_cache_options.ts

@@ -3,6 +3,7 @@ import lodash from '@poppinss/utils/lodash'
 import string from '@poppinss/utils/string'
 import { noopLogger } from '@julr/utils/logger'
 
+import { resolveTtl } from './helpers.js'
 import type { FactoryError } from './errors.js'
 import { JsonSerializer } from './serializers/json.js'
 import type {
@@ -73,6 +74,12 @@ export class BentoCacheOptions {
    */
   lockTimeout?: Duration = null
 
+  /**
+   * Duration for the circuit breaker to stay open
+   * if l2 cache fails
+   */
+  l2CircuitBreakerDuration: number | undefined
+
   /**
    * If the L1 cache should be serialized
    */
@@ -93,6 +100,7 @@ export class BentoCacheOptions {
 
     this.emitter = this.#options.emitter!
     this.serializer = this.#options.serializer ?? defaultSerializer
+    this.l2CircuitBreakerDuration = resolveTtl(this.#options.l2CircuitBreakerDuration, null)
 
     this.logger = this.#options.logger!.child({ pkg: 'bentocache' })
     this.onFactoryError = this.#options.onFactoryError

packages/bentocache/src/cache/cache_stack.ts

@@ -45,7 +45,7 @@ export class CacheStack extends BaseDriver {
         this.options.serializeL1 ? this.options.serializer : undefined,
       )
     if (drivers.l2Driver)
-      this.l2 = new RemoteCache(drivers.l2Driver, this.logger, this.options.serializer, !!this.l1)
+      this.l2 = new RemoteCache(drivers.l2Driver, this.logger, !!this.l1, this.options)
 
     this.bus = bus ? bus : this.#createBus(drivers.busDriver, drivers.busOptions)
     if (this.l1) this.bus?.manageCache(this.prefix, this.l1)

packages/bentocache/src/cache/facades/remote_cache.ts

@@ -1,8 +1,10 @@
 import { is } from '@julr/utils/is'
 
 import { CacheEntry } from '../cache_entry/cache_entry.js'
+import { CircuitBreaker } from '../../circuit_breaker/index.js'
+import type { L2CacheDriver, Logger } from '../../types/main.js'
+import type { BentoCacheOptions } from '../../bento_cache_options.js'
 import type { CacheEntryOptions } from '../cache_entry/cache_entry_options.js'
-import type { CacheSerializer, L2CacheDriver, Logger } from '../../types/main.js'
 
 /**
  * RemoteCache is a wrapper around a L2 Cache Driver that provides
@@ -11,18 +13,23 @@ import type { CacheSerializer, L2CacheDriver, Logger } from '../../types/main.js
 export class RemoteCache {
   #driver: L2CacheDriver
   #logger: Logger
-  #serializer: CacheSerializer
   #hasL1Backup: boolean
+  #circuitBreaker?: CircuitBreaker
+  #options: BentoCacheOptions
 
   constructor(
     driver: L2CacheDriver,
     logger: Logger,
-    serializer: CacheSerializer,
     hasL1Backup: boolean,
+    options: BentoCacheOptions,
   ) {
-    this.#hasL1Backup = hasL1Backup
     this.#driver = driver
-    this.#serializer = serializer
+    this.#options = options
+    this.#hasL1Backup = hasL1Backup
+    this.#circuitBreaker = options.l2CircuitBreakerDuration
+      ? new CircuitBreaker({ breakDuration: options.l2CircuitBreakerDuration })
+      : undefined
+
     this.#logger = logger.child({ context: 'bentocache.remoteCache' })
   }
 
@@ -36,11 +43,18 @@ export class RemoteCache {
     fallbackValue: unknown,
     fn: () => any,
   ) {
+    if (this.#circuitBreaker?.isOpen()) {
+      this.#logger.error({ opId: options.id }, `circuit breaker is open. ignoring operation`)
+      return fallbackValue
+    }
+
     try {
       return await fn()
     } catch (error) {
       this.#logger.error({ error, opId: options.id }, `(${operation}) failed on remote cache`)
 
+      this.#circuitBreaker?.open()
+
       /**
        * SuppressL2Errors is enabled automatically if undefined and we have a L1 backup
        * Otherwise, we need to check what the user set
@@ -64,7 +78,7 @@ export class RemoteCache {
       const value = await this.#driver.get(key)
       if (value === undefined) return
 
-      return CacheEntry.fromDriver(key, value, this.#serializer)
+      return CacheEntry.fromDriver(key, value, this.#options.serializer)
     })
   }
 

packages/bentocache/src/circuit_breaker/index.ts

@@ -0,0 +1,69 @@
+import { InvalidArgumentsException } from '@poppinss/utils/exceptions'
+
+const CircuitBreakerState = { Closed: 0, Open: 1 }
+
+interface CircuitBreakerOptions {
+  /**
+   * In milliseconds, how long the circuit breaker should stay open
+   */
+  breakDuration: number | undefined
+}
+
+/**
+ * Super simple circuit breaker implementation
+ */
+export class CircuitBreaker {
+  #state = CircuitBreakerState.Closed
+  #willCloseAt: number | null = null
+  #breakDuration: number
+
+  constructor(options: CircuitBreakerOptions) {
+    this.#breakDuration = options.breakDuration ?? 0
+    if (this.#breakDuration < 0) {
+      throw new InvalidArgumentsException('breakDuration must be a positive number')
+    }
+
+    this.#state = CircuitBreakerState.Closed
+  }
+
+  /**
+   * Check if the circuit breaker should change state
+   */
+  #checkState() {
+    if (this.#willCloseAt && this.#willCloseAt < Date.now()) this.close()
+  }
+
+  /**
+   * Check if the circuit breaker is open
+   */
+  isOpen() {
+    this.#checkState()
+    return this.#state === CircuitBreakerState.Open
+  }
+
+  /**
+   * Check if the circuit breaker is closed
+   */
+  isClosed() {
+    this.#checkState()
+    return this.#state === CircuitBreakerState.Closed
+  }
+
+  /**
+   * Open the circuit breaker
+   */
+  open() {
+    if (this.#state === CircuitBreakerState.Open) return
+
+    this.#state = CircuitBreakerState.Open
+    this.#willCloseAt = Date.now() + this.#breakDuration
+  }
+
+  /**
+   * Close the circuit breaker
+   */
+  close() {
+    this.#state = CircuitBreakerState.Closed
+    this.#willCloseAt = null
+  }
+}

packages/bentocache/src/types/options/options.ts

@@ -56,6 +56,14 @@ export type RawCommonOptions = {
    * throws an error
    */
   onFactoryError?: (error: FactoryError) => void
+
+  /**
+   * Duration for the circuit breaker to stay open
+   * if l2 cache fails
+   *
+   * @default null Means, no circuit breaker
+   */
+  l2CircuitBreakerDuration?: Duration
 }
 
 /**
@@ -82,7 +90,7 @@ export type RawBentoCacheOptions = {
   emitter?: Emitter
 
   /**
-   * Custom serialiser
+   * Custom serializer
    */
   serializer?: CacheSerializer
 } & RawCommonOptions

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

@@ -4,15 +4,15 @@ import { testLogger } from '@julr/utils/logger'
 import { REDIS_CREDENTIALS } from '../helpers/index.js'
 import { RedisDriver } from '../../src/drivers/redis.js'
 import { ChaosCache } from '../helpers/chaos/chaos_cache.js'
-import { JsonSerializer } from '../../src/serializers/json.js'
+import { BentoCacheOptions } from '../../src/bento_cache_options.js'
 import { RemoteCache } from '../../src/cache/facades/remote_cache.js'
 import { createCacheEntryOptions } from '../../src/cache/cache_entry/cache_entry_options.js'
 
 test.group('Remote Cache', () => {
   test('should rethrows errors if suppressL2Errors is disabled', async ({ assert, cleanup }) => {
     const logger = testLogger()
     const chaosCacheDriver = new ChaosCache(new RedisDriver({ connection: REDIS_CREDENTIALS }))
-    const cache = new RemoteCache(chaosCacheDriver, logger, new JsonSerializer(), true)
+    const cache = new RemoteCache(chaosCacheDriver, logger, true, new BentoCacheOptions({}))
 
     cleanup(() => chaosCacheDriver.disconnect())
 
@@ -32,7 +32,7 @@ test.group('Remote Cache', () => {
   test('should ignore errors if suppressL2Errors is enabled', async ({ assert, cleanup }) => {
     const logger = testLogger()
     const chaosCacheDriver = new ChaosCache(new RedisDriver({ connection: REDIS_CREDENTIALS }))
-    const cache = new RemoteCache(chaosCacheDriver, logger, new JsonSerializer(), true)
+    const cache = new RemoteCache(chaosCacheDriver, logger, true, new BentoCacheOptions({}))
 
     cleanup(() => chaosCacheDriver.disconnect())
 
@@ -55,7 +55,7 @@ test.group('Remote Cache', () => {
   }) => {
     const logger = testLogger()
     const chaosCacheDriver = new ChaosCache(new RedisDriver({ connection: REDIS_CREDENTIALS }))
-    const cache = new RemoteCache(chaosCacheDriver, logger, new JsonSerializer(), false)
+    const cache = new RemoteCache(chaosCacheDriver, logger, false, new BentoCacheOptions({}))
 
     cleanup(() => chaosCacheDriver.disconnect())
 
@@ -78,7 +78,7 @@ test.group('Remote Cache', () => {
   }) => {
     const logger = testLogger()
     const chaosCacheDriver = new ChaosCache(new RedisDriver({ connection: REDIS_CREDENTIALS }))
-    const cache = new RemoteCache(chaosCacheDriver, logger, new JsonSerializer(), false)
+    const cache = new RemoteCache(chaosCacheDriver, logger, false, new BentoCacheOptions({}))
 
     cleanup(() => chaosCacheDriver.disconnect())
 

packages/bentocache/tests/circuit_breaking.spec.ts

@@ -0,0 +1,83 @@
+import { mock } from 'node:test'
+import { test } from '@japa/runner'
+import { sleep } from '@julr/utils/misc'
+
+import { NullDriver } from './helpers/null/null_driver.js'
+import { CacheFactory } from '../factories/cache_factory.js'
+import { CircuitBreaker } from '../src/circuit_breaker/index.js'
+
+test.group('Circuit breaking', () => {
+  test('Simple circuit breaker should works', async ({ assert }) => {
+    const cb = new CircuitBreaker({ breakDuration: 400 })
+
+    cb.open()
+    cb.open()
+    cb.open()
+
+    assert.isTrue(cb.isOpen())
+    await sleep(200)
+    assert.isTrue(cb.isOpen())
+    await sleep(200)
+    assert.isFalse(cb.isOpen())
+    assert.isTrue(cb.isClosed())
+  })
+
+  test('Circuit breaker should open when remote cache call fail', async ({ assert }) => {
+    const get = mock.fn(() => {
+      throw new Error('Unable to connect to remote cache')
+    })
+
+    class L2Driver extends NullDriver {
+      type = 'l2' as const
+
+      get() {
+        return get()
+      }
+    }
+
+    const { cache } = new CacheFactory()
+      .withMemoryL1()
+      .merge({ l2Driver: new L2Driver({}), l2CircuitBreakerDuration: '200ms' })
+      .create()
+
+    await cache.get({ key: 'foo' })
+    await cache.get({ key: 'foo' })
+    await cache.get({ key: 'foo' })
+
+    assert.deepEqual(get.mock.callCount(), 1)
+
+    // Wait for the circuit breaker to close
+    await sleep(200)
+
+    await cache.get({ key: 'foo' })
+
+    assert.deepEqual(get.mock.callCount(), 2)
+  }).disableTimeout()
+
+  test('should not have circuit breaker if l2CircuitBreakerDuration is not set', async ({
+    assert,
+  }) => {
+    const get = mock.fn(() => {
+      throw new Error('Unable to connect to remote cache')
+    })
+
+    class L2Driver extends NullDriver {
+      type = 'l2' as const
+
+      get() {
+        return get()
+      }
+    }
+
+    const { cache } = new CacheFactory()
+      .withMemoryL1()
+      .merge({ l2Driver: new L2Driver({}) })
+      .create()
+
+    await cache.get({ key: 'foo' })
+    await cache.get({ key: 'foo' })
+    await cache.get({ key: 'foo' })
+
+    assert.deepEqual(get.mock.callCount(), 3)
+  })
+})