25367879152

Committed 05 May 2026 09:14AM UTC coverage: 95.634% (-0.5%) from 96.091%

Build # 25367879152

Build Type

Pull #71

github

Committed by

web-flow

Commit Message

Merge 641ee0845 into 57c878ba7

Pull Request Pull Request #71: Merge security, cache reliability, and docs updates

Coverage Stats

1727 of 1860 branches covered (92.85%)

Branch coverage included in aggregate %.

160 of 173 new or added lines in 11 files covered. (92.49%)

2 existing lines in 1 file now uncovered.

3136 of 3225 relevant lines covered (97.24%)

331.34 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

97.81

/src/internal/CacheStackReader.ts

import type { StampedeGuard } from '../stampede/StampedeGuard'
import type {
  CacheCircuitBreakerOptions,
  CacheFetcher,
  CacheFetcherContext,
  CacheGetOptions,
  CacheLayer,
  CacheLogger,
  CacheRateLimitOptions,
  CacheSingleFlightCoordinator,
  CacheSingleFlightExecutionOptions,
  CacheStackEvents,
  CacheTagIndex,
  CacheWriteOptions,
  LayerTtlMap
} from '../types'
import type { CacheWriteKind } from './CacheStackLayerWriter'
import type { CacheStackMaintenance } from './CacheStackMaintenance'
import { planFreshReadPolicies, shouldStartBackgroundRefresh } from './CacheStackRuntimePolicy'
import type { CircuitBreakerManager } from './CircuitBreakerManager'
import type { FetchRateLimiter } from './FetchRateLimiter'
import type { MetricsCollector } from './MetricsCollector'
import { isStoredValueEnvelope, remainingStoredTtlMs, resolveStoredValue } from './StoredValue'
import type { TtlResolver } from './TtlResolver'

const DEFAULT_SINGLE_FLIGHT_LEASE_MS = 30_000
const DEFAULT_SINGLE_FLIGHT_TIMEOUT_MS = 5_000
const DEFAULT_SINGLE_FLIGHT_POLL_MS = 50
const DEFAULT_BACKGROUND_REFRESH_TIMEOUT_MS = 30_000

type ReadMode = 'allow-stale' | 'fresh-only'

type ReadHit<T> =
  | {
      found: true
      value: T | null
      stored: unknown
      state: 'fresh' | 'stale-while-revalidate' | 'stale-if-error'
      layerIndex: number
      layerName: string
    }
  | { found: false; value: null; stored: null; state: 'miss' }

interface CacheStackReaderOptions {
  // Direct service objects
  layers: CacheLayer[]
  metricsCollector: MetricsCollector
  maintenance: CacheStackMaintenance
  tagIndex: CacheTagIndex
  circuitBreakerManager: CircuitBreakerManager
  fetchRateLimiter: FetchRateLimiter
  stampedeGuard: StampedeGuard
  ttlResolver: TtlResolver
  logger: CacheLogger

  // CacheStack method callbacks
  shouldSkipLayer: (layer: CacheLayer) => boolean
  handleLayerFailure: (layer: CacheLayer, operation: string, error: unknown) => Promise<null>
  emit: <K extends keyof CacheStackEvents>(event: K, data: CacheStackEvents[K]) => boolean
  emitError: (operation: string, context: Record<string, unknown>) => void
  formatError: (error: unknown) => string
  storeEntry: (key: string, kind: CacheWriteKind, value: unknown, options?: CacheWriteOptions) => Promise<void>
  recordCircuitFailure: (key: string, options: CacheCircuitBreakerOptions | undefined, error: unknown) => void
  resolveLayerMs: (
    layerName: string,
    override: number | LayerTtlMap | undefined,
    globalDefault?: number | LayerTtlMap,
    fallback?: number
  ) => number | undefined
  sleep: (ms: number) => Promise<void>
  withTimeout: <T>(promise: Promise<T>, timeoutMs: number, createError: () => Error) => Promise<T>
  isDisconnecting: () => boolean
  isGracefulDegradationEnabled: () => boolean
  scheduleBackgroundRefreshDispatch: <T>(
    key: string,
    fetcher: CacheFetcher<T>,
    options?: CacheGetOptions,
    fetcherContext?: CacheFetcherContext<T>
  ) => void

  // Config values
  stampedePrevention?: boolean
  singleFlightCoordinator?: CacheSingleFlightCoordinator
  singleFlightLeaseMs?: number
  singleFlightTimeoutMs?: number
  singleFlightPollMs?: number
  singleFlightRenewIntervalMs?: number
  backgroundRefreshTimeoutMs?: number
  negativeCaching?: boolean
  refreshAhead?: number | LayerTtlMap
  circuitBreaker?: CacheCircuitBreakerOptions
  fetcherRateLimit?: CacheRateLimitOptions
}

export class CacheStackReader {
  private readonly backgroundRefreshes = new Map<string, Promise<void>>()
  private readonly backgroundRefreshAbort = new Map<string, boolean>()

  constructor(private readonly options: CacheStackReaderOptions) {}

  get activeRefreshCount(): number {
    return this.backgroundRefreshes.size
  }

  async getPrepared<T>(normalizedKey: string, fetcher?: CacheFetcher<T>, options?: CacheGetOptions): Promise<T | null> {
    const hit = await this.readFromLayers<T>(normalizedKey, options, 'allow-stale')
    if (hit.found) {
      this.options.ttlResolver.recordAccess(normalizedKey)
      if (this.isNegativeStoredValue(hit.stored)) {
        this.options.metricsCollector.increment('negativeCacheHits')
      }

      if (hit.state === 'fresh') {
        this.options.metricsCollector.increment('hits')
        await this.applyFreshReadPolicies(normalizedKey, hit, options, fetcher)
        return hit.value
      }

      if (hit.state === 'stale-while-revalidate') {
        this.options.metricsCollector.increment('hits')
        this.options.metricsCollector.increment('staleHits')
        this.options.emit('stale-serve', { key: normalizedKey, state: hit.state, layer: hit.layerName })
        if (fetcher) {
          this.scheduleBackgroundRefresh(normalizedKey, fetcher, options, this.createFetcherContext(normalizedKey, hit))
        }
        return hit.value
      }

      if (!fetcher) {
        this.options.metricsCollector.increment('hits')
        this.options.metricsCollector.increment('staleHits')
        this.options.emit('stale-serve', { key: normalizedKey, state: hit.state, layer: hit.layerName })
        return hit.value
      }

      try {
        return await this.fetchWithGuards(
          normalizedKey,
          fetcher,
          options,
          undefined,
          undefined,
          false,
          this.createFetcherContext(normalizedKey, hit)
        )
      } catch (error) {
        this.options.metricsCollector.increment('staleHits')
        this.options.metricsCollector.increment('refreshErrors')
        this.options.logger.debug?.('stale-if-error', {
          key: normalizedKey,
          error: this.options.formatError(error)
        })
        return hit.value
      }
    }

    this.options.metricsCollector.increment('misses')
    if (!fetcher) {
      return null
    }

    return this.fetchWithGuards(normalizedKey, fetcher, options, undefined, undefined, true, {
      key: normalizedKey,
      currentValue: undefined,
      state: 'miss'
    })
  }

  async readLayerEntry(layer: CacheLayer, key: string): Promise<unknown | null> {
    if (this.options.shouldSkipLayer(layer)) {
      return null
    }

    if (layer.getEntry) {
      try {
        return await layer.getEntry(key)
      } catch (error) {
        return this.options.handleLayerFailure(layer, 'read', error)
      }
    }

    try {
      return await layer.get(key)
    } catch (error) {
      return this.options.handleLayerFailure(layer, 'read', error)
    }
  }

  async backfill(key: string, stored: unknown, upToIndex: number, options?: CacheGetOptions): Promise<void> {
    if (upToIndex < 0) {
      return
    }

    const operations: Array<Promise<void>> = []

    for (let index = 0; index <= upToIndex; index += 1) {
      const layer = this.options.layers[index]
      if (!layer || this.options.shouldSkipLayer(layer)) {
        continue
      }

      const ttl =
        remainingStoredTtlMs(stored) ??
        this.options.resolveLayerMs(layer.name, options?.ttl, undefined, layer.defaultTtl)
      operations.push(
        (async () => {
          try {
            await layer.set(key, stored, ttl)
          } catch (error) {
            await this.options.handleLayerFailure(layer, 'backfill', error)
            return
          }
          this.options.metricsCollector.increment('backfills')
          this.options.logger.debug?.('backfill', { key, layer: layer.name })
          this.options.emit('backfill', { key, layer: layer.name })
        })()
      )
    }

    await Promise.all(operations)
  }

  abortAllRefreshes(): void {
    for (const key of this.backgroundRefreshAbort.keys()) {
      this.backgroundRefreshAbort.set(key, true)
    }
  }

  getAllRefreshPromises(): Promise<void>[] {
    return [...this.backgroundRefreshes.values()]
  }

  private async readFromLayers<T>(
    key: string,
    options: CacheGetOptions | undefined,
    mode: ReadMode
  ): Promise<ReadHit<T>> {
    let sawRetainableValue = false

    for (let index = 0; index < this.options.layers.length; index += 1) {
      const layer = this.options.layers[index]
      if (!layer) continue
      const readStart = performance.now()
      const stored = await this.readLayerEntry(layer, key)
      const readDuration = performance.now() - readStart
      this.options.metricsCollector.recordLatency(layer.name, readDuration)
      if (stored === null) {
        this.options.metricsCollector.incrementLayer('missesByLayer', layer.name)
        continue
      }

      const resolved = resolveStoredValue<T>(stored)
      if (resolved.state === 'expired') {
        await layer.delete(key)
        continue
      }

      sawRetainableValue = true

      if (mode === 'fresh-only' && resolved.state !== 'fresh') {
        continue
      }

      await this.options.tagIndex.touch(key)
      await this.backfill(key, stored, index - 1, options)
      this.options.metricsCollector.incrementLayer('hitsByLayer', layer.name)
      this.options.logger.debug?.('hit', { key, layer: layer.name, state: resolved.state })
      this.options.emit('hit', {
        key,
        layer: layer.name,
        state: resolved.state as CacheStackEvents['hit']['state']
      })
      return {
        found: true,
        value: resolved.value,
        stored,
        state: resolved.state,
        layerIndex: index,
        layerName: layer.name
      }
    }

    if (!sawRetainableValue) {
      await this.options.tagIndex.remove(key)
    }

    this.options.logger.debug?.('miss', { key, mode })
    this.options.emit('miss', { key, mode })
    return { found: false, value: null, stored: null, state: 'miss' }
  }

  private async fetchWithGuards<T>(
    key: string,
    fetcher: CacheFetcher<T>,
    options?: CacheGetOptions,
    expectedClearEpoch?: number,
    expectedKeyEpoch?: number,
    initialMissConfirmed = false,
    fetcherContext: CacheFetcherContext<T> = {
      key,
      currentValue: undefined,
      state: 'miss'
    }
  ): Promise<T | null> {
    const fetchTask = async (): Promise<T | null> => {
      const shouldRecheckFreshLayers = !(initialMissConfirmed && this.options.singleFlightCoordinator)
      if (shouldRecheckFreshLayers) {
        const secondHit = await this.readFromLayers<T>(key, options, 'fresh-only')
        if (secondHit.found) {
          this.options.metricsCollector.increment('hits')
          return secondHit.value
        }
      }

      return this.fetchAndPopulate(key, fetcher, options, expectedClearEpoch, expectedKeyEpoch, fetcherContext)
    }

    const singleFlightTask = async (): Promise<T | null> => {
      if (!this.options.singleFlightCoordinator) {
        return fetchTask()
      }

      try {
        return await this.options.singleFlightCoordinator.execute(
          key,
          this.resolveSingleFlightOptions(),
          fetchTask,
          () => this.waitForFreshValue(key, fetcher, options, expectedClearEpoch, expectedKeyEpoch, fetcherContext)
        )
      } catch (error) {
        if (!this.options.isGracefulDegradationEnabled()) {
          throw error
        }

        this.options.metricsCollector.increment('degradedOperations')
        this.options.logger.warn?.('single-flight-coordinator-degraded', {
          key,
          error: this.options.formatError(error)
        })
        this.options.emitError('single-flight', {
          key,
          degraded: true,
          error: this.options.formatError(error)
        })
        return fetchTask()
      }
    }

    if (this.options.stampedePrevention === false) {
      return singleFlightTask()
    }

    return this.options.stampedeGuard.execute(key, singleFlightTask)
  }

  private async waitForFreshValue<T>(
    key: string,
    fetcher: CacheFetcher<T>,
    options?: CacheGetOptions,
    expectedClearEpoch?: number,
    expectedKeyEpoch?: number,
    fetcherContext: CacheFetcherContext<T> = {
      key,
      currentValue: undefined,
      state: 'miss'
    }
  ): Promise<T | null> {
    const timeoutMs = this.options.singleFlightTimeoutMs ?? DEFAULT_SINGLE_FLIGHT_TIMEOUT_MS
    const pollIntervalMs = this.options.singleFlightPollMs ?? DEFAULT_SINGLE_FLIGHT_POLL_MS
    const deadline = Date.now() + timeoutMs

    this.options.metricsCollector.increment('singleFlightWaits')
    this.options.emit('stampede-dedupe', { key })

    while (Date.now() < deadline) {
      const hit = await this.readFromLayers<T>(key, options, 'fresh-only')
      if (hit.found) {
        this.options.metricsCollector.increment('hits')
        return hit.value
      }
      await this.options.sleep(pollIntervalMs)
    }

    if (!this.options.singleFlightCoordinator) {
      return this.fetchAndPopulate(key, fetcher, options, expectedClearEpoch, expectedKeyEpoch, fetcherContext)
    }

    return this.options.singleFlightCoordinator.execute(
      key,
      this.resolveSingleFlightOptions(),
      () => this.fetchAndPopulate(key, fetcher, options, expectedClearEpoch, expectedKeyEpoch, fetcherContext),
      () => this.waitForFreshValue(key, fetcher, options, expectedClearEpoch, expectedKeyEpoch, fetcherContext)
    )
  }

  private async fetchAndPopulate<T>(
    key: string,
    fetcher: CacheFetcher<T>,
    options?: CacheGetOptions,
    expectedClearEpoch?: number,
    expectedKeyEpoch?: number,
    fetcherContext: CacheFetcherContext<T> = {
      key,
      currentValue: undefined,
      state: 'miss'
    }
  ): Promise<T | null> {
    this.options.circuitBreakerManager.assertClosed(key, options?.circuitBreaker ?? this.options.circuitBreaker)
    this.options.metricsCollector.increment('fetches')
    const fetchStart = Date.now()
    let fetched: T

    try {
      fetched = await this.options.fetchRateLimiter.schedule(
        options?.fetcherRateLimit ?? this.options.fetcherRateLimit,
        { key, fetcher },
        () => fetcher(fetcherContext)
      )
      this.options.circuitBreakerManager.recordSuccess(key)
      this.options.logger.debug?.('fetch', { key, durationMs: Date.now() - fetchStart })
    } catch (error) {
      this.options.recordCircuitFailure(key, options?.circuitBreaker ?? this.options.circuitBreaker, error)
      throw error
    }

    if (fetched === null || fetched === undefined) {
      if (!this.shouldNegativeCache(options)) {
        return null
      }

      if (this.options.maintenance.isWriteOutdated(key, expectedClearEpoch, expectedKeyEpoch)) {
        this.options.logger.debug?.('skip-negative-store-after-invalidation', {
          key,
          expectedClearEpoch,
          clearEpoch: this.options.maintenance.currentClearEpoch(),
          expectedKeyEpoch,
          keyEpoch: this.options.maintenance.currentKeyEpoch(key)
        })
        return null
      }

      await this.options.storeEntry(key, 'empty', null, options)
      return null
    }

    // Conditional caching: skip storage if shouldCache returns false
    if (options?.shouldCache) {
      try {
        if (!options.shouldCache(fetched)) {
          return fetched
        }
      } catch (error) {
        this.options.logger.warn?.('shouldCache-error', {
          key,
          error: this.options.formatError(error)
        })
      }
    }

    if (this.options.maintenance.isWriteOutdated(key, expectedClearEpoch, expectedKeyEpoch)) {
      this.options.logger.debug?.('skip-store-after-invalidation', {
        key,
        expectedClearEpoch,
        clearEpoch: this.options.maintenance.currentClearEpoch(),
        expectedKeyEpoch,
        keyEpoch: this.options.maintenance.currentKeyEpoch(key)
      })
      return fetched
    }

    await this.options.storeEntry(key, 'value', fetched, options)
    return fetched
  }

  runScheduleBackgroundRefresh<T>(
    key: string,
    fetcher: CacheFetcher<T>,
    options?: CacheGetOptions,
    fetcherContext?: CacheFetcherContext<T>
  ): void {
    this.scheduleBackgroundRefresh(key, fetcher, options, fetcherContext)
  }

  private scheduleBackgroundRefresh<T>(
    key: string,
    fetcher: CacheFetcher<T>,
    options?: CacheGetOptions,
    fetcherContext: CacheFetcherContext<T> = {
      key,
      currentValue: undefined,
      state: 'miss'
    }
  ): void {
    if (
      !shouldStartBackgroundRefresh({
        isDisconnecting: this.options.isDisconnecting(),
        hasRefreshInFlight: this.backgroundRefreshes.has(key)
      })
    ) {
      return
    }

    const clearEpoch = this.options.maintenance.currentClearEpoch()
    const keyEpoch = this.options.maintenance.currentKeyEpoch(key)
    this.backgroundRefreshAbort.set(key, false)
    const refresh = (async () => {
      this.options.metricsCollector.increment('refreshes')
      try {
        if (this.backgroundRefreshAbort.get(key)) return
        await this.runBackgroundRefresh(key, fetcher, options, clearEpoch, keyEpoch, fetcherContext)
      } catch (error) {
        if (this.backgroundRefreshAbort.get(key)) return
        this.options.metricsCollector.increment('refreshErrors')
        this.options.logger.warn?.('background-refresh-error', {
          key,
          error: this.options.formatError(error)
        })
      } finally {
        this.backgroundRefreshes.delete(key)
        this.backgroundRefreshAbort.delete(key)
      }
    })()

    this.backgroundRefreshes.set(key, refresh)
  }

  private async runBackgroundRefresh<T>(
    key: string,
    fetcher: CacheFetcher<T>,
    options?: CacheGetOptions,
    expectedClearEpoch?: number,
    expectedKeyEpoch?: number,
    fetcherContext: CacheFetcherContext<T> = {
      key,
      currentValue: undefined,
      state: 'miss'
    }
  ): Promise<void> {
    const timeoutMs = this.options.backgroundRefreshTimeoutMs ?? DEFAULT_BACKGROUND_REFRESH_TIMEOUT_MS
    await this.fetchWithGuards(
      key,
      (context) =>
        this.options.withTimeout(fetcher(context), timeoutMs, () => {
          return new Error(`Background refresh timed out after ${timeoutMs}ms for key "${key}".`)
        }),
      options,
      expectedClearEpoch,
      expectedKeyEpoch,
      false,
      fetcherContext
    )
  }

  async runApplyFreshReadPolicies<T>(
    key: string,
    hit: {
      found: true
      value: T | null
      stored: unknown
      state: 'fresh' | 'stale-while-revalidate' | 'stale-if-error'
      layerIndex: number
      layerName: string
    },
    options: CacheGetOptions | undefined,
    fetcher?: CacheFetcher<T>
  ): Promise<void> {
    return this.applyFreshReadPolicies(key, hit as Extract<ReadHit<T>, { found: true }>, options, fetcher)
  }

  private async applyFreshReadPolicies<T>(
    key: string,
    hit: Extract<ReadHit<T>, { found: true }>,
    options: CacheGetOptions | undefined,
    fetcher?: CacheFetcher<T>
  ): Promise<void> {
    const plan = planFreshReadPolicies({
      stored: hit.stored,
      hasFetcher: Boolean(fetcher),
      slidingTtl: options?.slidingTtl ?? false,
      refreshAheadMs:
        this.options.resolveLayerMs(hit.layerName, options?.refreshAhead, this.options.refreshAhead, 0) ?? 0
    })

    if (plan.refreshedStored) {
      for (let index = 0; index <= hit.layerIndex; index += 1) {
        const layer = this.options.layers[index]
        if (!layer || this.options.shouldSkipLayer(layer)) {
          continue
        }

        try {
          await layer.set(key, plan.refreshedStored, plan.refreshedStoredTtl)
        } catch (error) {
          await this.options.handleLayerFailure(layer, 'sliding-ttl', error)
        }
      }
    }

    if (fetcher && plan.shouldScheduleBackgroundRefresh) {
      this.options.scheduleBackgroundRefreshDispatch(key, fetcher, options, this.createFetcherContext(key, hit))
    }
  }

  private createFetcherContext<T>(key: string, hit: Extract<ReadHit<T>, { found: true }>): CacheFetcherContext<T> {
    return {
      key,
      currentValue: hit.value === null ? undefined : hit.value,
      state: hit.state,
      layer: hit.layerName
    }
  }

  private resolveSingleFlightOptions(): CacheSingleFlightExecutionOptions {
    return {
      leaseMs: this.options.singleFlightLeaseMs ?? DEFAULT_SINGLE_FLIGHT_LEASE_MS,
      waitTimeoutMs: this.options.singleFlightTimeoutMs ?? DEFAULT_SINGLE_FLIGHT_TIMEOUT_MS,
      pollIntervalMs: this.options.singleFlightPollMs ?? DEFAULT_SINGLE_FLIGHT_POLL_MS,
      renewIntervalMs: this.options.singleFlightRenewIntervalMs
    }
  }

  private shouldNegativeCache(options?: CacheGetOptions): boolean {
    return options?.negativeCache ?? this.options.negativeCaching ?? false
  }

  private isNegativeStoredValue(stored: unknown): boolean {
    return isStoredValueEnvelope(stored) && stored.kind === 'empty'
  }
}

1	import type { StampedeGuard } from '../stampede/StampedeGuard'
2	import type {
3	CacheCircuitBreakerOptions,
4	CacheFetcher,
5	CacheFetcherContext,
6	CacheGetOptions,
7	CacheLayer,
8	CacheLogger,
9	CacheRateLimitOptions,
10	CacheSingleFlightCoordinator,
11	CacheSingleFlightExecutionOptions,
12	CacheStackEvents,
13	CacheTagIndex,
14	CacheWriteOptions,
15	LayerTtlMap
16	} from '../types'
17	import type { CacheWriteKind } from './CacheStackLayerWriter'
18	import type { CacheStackMaintenance } from './CacheStackMaintenance'
19	import { planFreshReadPolicies, shouldStartBackgroundRefresh } from './CacheStackRuntimePolicy'
20	import type { CircuitBreakerManager } from './CircuitBreakerManager'
21	import type { FetchRateLimiter } from './FetchRateLimiter'
22	import type { MetricsCollector } from './MetricsCollector'
23	import { isStoredValueEnvelope, remainingStoredTtlMs, resolveStoredValue } from './StoredValue'
24	import type { TtlResolver } from './TtlResolver'
25
26	const DEFAULT_SINGLE_FLIGHT_LEASE_MS = 30_000	14✔
27	const DEFAULT_SINGLE_FLIGHT_TIMEOUT_MS = 5_000	14✔
28	const DEFAULT_SINGLE_FLIGHT_POLL_MS = 50	14✔
29	const DEFAULT_BACKGROUND_REFRESH_TIMEOUT_MS = 30_000	14✔
30
31	type ReadMode = 'allow-stale' \| 'fresh-only'
32
33	type ReadHit<T> =
34	\| {
35	found: true
36	value: T \| null
37	stored: unknown
38	state: 'fresh' \| 'stale-while-revalidate' \| 'stale-if-error'
39	layerIndex: number
40	layerName: string
41	}
42	\| { found: false; value: null; stored: null; state: 'miss' }
43
44	interface CacheStackReaderOptions {
45	// Direct service objects
46	layers: CacheLayer[]
47	metricsCollector: MetricsCollector
48	maintenance: CacheStackMaintenance
49	tagIndex: CacheTagIndex
50	circuitBreakerManager: CircuitBreakerManager
51	fetchRateLimiter: FetchRateLimiter
52	stampedeGuard: StampedeGuard
53	ttlResolver: TtlResolver
54	logger: CacheLogger
55
56	// CacheStack method callbacks
57	shouldSkipLayer: (layer: CacheLayer) => boolean
58	handleLayerFailure: (layer: CacheLayer, operation: string, error: unknown) => Promise<null>
59	emit: <K extends keyof CacheStackEvents>(event: K, data: CacheStackEvents[K]) => boolean
60	emitError: (operation: string, context: Record<string, unknown>) => void
61	formatError: (error: unknown) => string
62	storeEntry: (key: string, kind: CacheWriteKind, value: unknown, options?: CacheWriteOptions) => Promise<void>
63	recordCircuitFailure: (key: string, options: CacheCircuitBreakerOptions \| undefined, error: unknown) => void
64	resolveLayerMs: (
65	layerName: string,
66	override: number \| LayerTtlMap \| undefined,
67	globalDefault?: number \| LayerTtlMap,
68	fallback?: number
69	) => number \| undefined
70	sleep: (ms: number) => Promise<void>
71	withTimeout: <T>(promise: Promise<T>, timeoutMs: number, createError: () => Error) => Promise<T>
72	isDisconnecting: () => boolean
73	isGracefulDegradationEnabled: () => boolean
74	scheduleBackgroundRefreshDispatch: <T>(
75	key: string,
76	fetcher: CacheFetcher<T>,
77	options?: CacheGetOptions,
78	fetcherContext?: CacheFetcherContext<T>
79	) => void
80
81	// Config values
82	stampedePrevention?: boolean
83	singleFlightCoordinator?: CacheSingleFlightCoordinator
84	singleFlightLeaseMs?: number
85	singleFlightTimeoutMs?: number
86	singleFlightPollMs?: number
87	singleFlightRenewIntervalMs?: number
88	backgroundRefreshTimeoutMs?: number
89	negativeCaching?: boolean
90	refreshAhead?: number \| LayerTtlMap
91	circuitBreaker?: CacheCircuitBreakerOptions
92	fetcherRateLimit?: CacheRateLimitOptions
93	}
94
95	export class CacheStackReader {
96	private readonly backgroundRefreshes = new Map<string, Promise<void>>()	296✔
97	private readonly backgroundRefreshAbort = new Map<string, boolean>()	296✔
98
99	constructor(private readonly options: CacheStackReaderOptions) {}	296✔
100
101	get activeRefreshCount(): number {
102	return this.backgroundRefreshes.size	34✔
103	}
104
105	async getPrepared<T>(normalizedKey: string, fetcher?: CacheFetcher<T>, options?: CacheGetOptions): Promise<T \| null> {
106	const hit = await this.readFromLayers<T>(normalizedKey, options, 'allow-stale')	333✔
107	if (hit.found) {	333✔
108	this.options.ttlResolver.recordAccess(normalizedKey)	106✔
109	if (this.isNegativeStoredValue(hit.stored)) {	106✔
110	this.options.metricsCollector.increment('negativeCacheHits')	2✔
111	}
112
113	if (hit.state === 'fresh') {	106✔
114	this.options.metricsCollector.increment('hits')	84✔
115	await this.applyFreshReadPolicies(normalizedKey, hit, options, fetcher)	84✔
116	return hit.value	84✔
117	}
118
119	if (hit.state === 'stale-while-revalidate') {	22✔
120	this.options.metricsCollector.increment('hits')	17✔
121	this.options.metricsCollector.increment('staleHits')	17✔
122	this.options.emit('stale-serve', { key: normalizedKey, state: hit.state, layer: hit.layerName })	17✔
123	if (fetcher) {	17✔
124	this.scheduleBackgroundRefresh(normalizedKey, fetcher, options, this.createFetcherContext(normalizedKey, hit))	15✔
125	}
126	return hit.value	17✔
127	}
128
129	if (!fetcher) {	5✔
130	this.options.metricsCollector.increment('hits')	1✔
131	this.options.metricsCollector.increment('staleHits')	1✔
132	this.options.emit('stale-serve', { key: normalizedKey, state: hit.state, layer: hit.layerName })	1✔
133	return hit.value	1✔
134	}
135
136	try {	4✔
137	return await this.fetchWithGuards(	4✔
138	normalizedKey,
139	fetcher,
140	options,
141	undefined,
142	undefined,
143	false,
144	this.createFetcherContext(normalizedKey, hit)
145	)
146	} catch (error) {
147	this.options.metricsCollector.increment('staleHits')	3✔
148	this.options.metricsCollector.increment('refreshErrors')	3✔
149	this.options.logger.debug?.('stale-if-error', {	3✔
150	key: normalizedKey,
151	error: this.options.formatError(error)
152	})
153	return hit.value	3✔
154	}
155	}
156
157	this.options.metricsCollector.increment('misses')	227✔
158	if (!fetcher) {	227✔
159	return null	74✔
160	}
161
162	return this.fetchWithGuards(normalizedKey, fetcher, options, undefined, undefined, true, {	153✔
163	key: normalizedKey,
164	currentValue: undefined,
165	state: 'miss'
166	})
167	}
168
169	async readLayerEntry(layer: CacheLayer, key: string): Promise<unknown \| null> {
170	if (this.options.shouldSkipLayer(layer)) {	497✔
171	return null	3✔
172	}
173
174	if (layer.getEntry) {	494✔
175	try {	430✔
176	return await layer.getEntry(key)	430✔
177	} catch (error) {
178	return this.options.handleLayerFailure(layer, 'read', error)	2✔
179	}
180	}
181
182	try {	64✔
183	return await layer.get(key)	64✔
184	} catch (error) {
185	return this.options.handleLayerFailure(layer, 'read', error)	3✔
186	}
187	}
188
189	async backfill(key: string, stored: unknown, upToIndex: number, options?: CacheGetOptions): Promise<void> {
190	if (upToIndex < 0) {	126✔
191	return	106✔
192	}
193
194	const operations: Array<Promise<void>> = []	20✔
195
196	for (let index = 0; index <= upToIndex; index += 1) {	20✔
197	const layer = this.options.layers[index]	24✔
198	if (!layer \|\| this.options.shouldSkipLayer(layer)) {	24✔
199	continue	4✔
200	}
201
202	const ttl =	20✔
203	remainingStoredTtlMs(stored) ??
204	this.options.resolveLayerMs(layer.name, options?.ttl, undefined, layer.defaultTtl)
205	operations.push(	24✔
206	(async () => {
207	try {	20✔
208	await layer.set(key, stored, ttl)	20✔
209	} catch (error) {
210	await this.options.handleLayerFailure(layer, 'backfill', error)	1✔
211	return	1✔
212	}
213	this.options.metricsCollector.increment('backfills')	19✔
214	this.options.logger.debug?.('backfill', { key, layer: layer.name })	19✔
215	this.options.emit('backfill', { key, layer: layer.name })	20✔
216	})()
217	)
218	}
219
220	await Promise.all(operations)	24✔
221	}
222
223	abortAllRefreshes(): void {
224	for (const key of this.backgroundRefreshAbort.keys()) {	29✔
225	this.backgroundRefreshAbort.set(key, true)	1✔
226	}
227	}
228
229	getAllRefreshPromises(): Promise<void>[] {
230	return [...this.backgroundRefreshes.values()]	31✔
231	}
232
233	private async readFromLayers<T>(
234	key: string,
235	options: CacheGetOptions \| undefined,
236	mode: ReadMode
237	): Promise<ReadHit<T>> {
238	let sawRetainableValue = false	435✔
239
240	for (let index = 0; index < this.options.layers.length; index += 1) {	435✔
241	const layer = this.options.layers[index]	457✔
242	if (!layer) continue	457!
243	const readStart = performance.now()	457✔
244	const stored = await this.readLayerEntry(layer, key)	457✔
245	const readDuration = performance.now() - readStart	457✔
246	this.options.metricsCollector.recordLatency(layer.name, readDuration)	457✔
247	if (stored === null) {	457✔
248	this.options.metricsCollector.incrementLayer('missesByLayer', layer.name)	325✔
249	continue	325✔
250	}
251
252	const resolved = resolveStoredValue<T>(stored)	132✔
253	if (resolved.state === 'expired') {	132✔
254	await layer.delete(key)	2✔
255	continue	2✔
256	}
257
258	sawRetainableValue = true	130✔
259
260	if (mode === 'fresh-only' && resolved.state !== 'fresh') {	130✔
261	continue	20✔
262	}
263
264	await this.options.tagIndex.touch(key)	110✔
265	await this.backfill(key, stored, index - 1, options)	110✔
266	this.options.metricsCollector.incrementLayer('hitsByLayer', layer.name)	110✔
267	this.options.logger.debug?.('hit', { key, layer: layer.name, state: resolved.state })	110✔
268	this.options.emit('hit', {	457✔
269	key,
270	layer: layer.name,
271	state: resolved.state as CacheStackEvents['hit']['state']
272	})
273	return {	457✔
274	found: true,
275	value: resolved.value,
276	stored,
277	state: resolved.state,
278	layerIndex: index,
279	layerName: layer.name
280	}
281	}
282
283	if (!sawRetainableValue) {	325✔
284	await this.options.tagIndex.remove(key)	306✔
285	}
286
287	this.options.logger.debug?.('miss', { key, mode })	325✔
288	this.options.emit('miss', { key, mode })	435✔
289	return { found: false, value: null, stored: null, state: 'miss' }	435✔
290	}
291
292	private async fetchWithGuards<T>(
293	key: string,
294	fetcher: CacheFetcher<T>,
295	options?: CacheGetOptions,
296	expectedClearEpoch?: number,
297	expectedKeyEpoch?: number,
298	initialMissConfirmed = false,	175✔
299	fetcherContext: CacheFetcherContext<T> = {	175✔
300	key,
301	currentValue: undefined,
302	state: 'miss'
303	}
304	): Promise<T \| null> {
305	const fetchTask = async (): Promise<T \| null> => {	175✔
306	const shouldRecheckFreshLayers = !(initialMissConfirmed && this.options.singleFlightCoordinator)	96✔
307	if (shouldRecheckFreshLayers) {	96✔
308	const secondHit = await this.readFromLayers<T>(key, options, 'fresh-only')	90✔
309	if (secondHit.found) {	90✔
310	this.options.metricsCollector.increment('hits')	2✔
311	return secondHit.value	2✔
312	}
313	}
314
315	return this.fetchAndPopulate(key, fetcher, options, expectedClearEpoch, expectedKeyEpoch, fetcherContext)	94✔
316	}
317
318	const singleFlightTask = async (): Promise<T \| null> => {	175✔
319	if (!this.options.singleFlightCoordinator) {	101✔
320	return fetchTask()	90✔
321	}
322
323	try {	11✔
324	return await this.options.singleFlightCoordinator.execute(	11✔
325	key,
326	this.resolveSingleFlightOptions(),
327	fetchTask,
328	() => this.waitForFreshValue(key, fetcher, options, expectedClearEpoch, expectedKeyEpoch, fetcherContext)	4✔
329	)
330	} catch (error) {
331	if (!this.options.isGracefulDegradationEnabled()) {	3✔
332	throw error	1✔
333	}
334
335	this.options.metricsCollector.increment('degradedOperations')	2✔
336	this.options.logger.warn?.('single-flight-coordinator-degraded', {	2✔
337	key,
338	error: this.options.formatError(error)
339	})
340	this.options.emitError('single-flight', {	3✔
341	key,
342	degraded: true,
343	error: this.options.formatError(error)
344	})
345	return fetchTask()	3✔
346	}
347	}
348
349	if (this.options.stampedePrevention === false) {	175✔
350	return singleFlightTask()	3✔
351	}
352
353	return this.options.stampedeGuard.execute(key, singleFlightTask)	172✔
354	}
355
356	private async waitForFreshValue<T>(
357	key: string,
358	fetcher: CacheFetcher<T>,
359	options?: CacheGetOptions,
360	expectedClearEpoch?: number,
361	expectedKeyEpoch?: number,
362	fetcherContext: CacheFetcherContext<T> = {	4✔
363	key,
364	currentValue: undefined,
365	state: 'miss'
366	}
367	): Promise<T \| null> {
368	const timeoutMs = this.options.singleFlightTimeoutMs ?? DEFAULT_SINGLE_FLIGHT_TIMEOUT_MS	4✔
369	const pollIntervalMs = this.options.singleFlightPollMs ?? DEFAULT_SINGLE_FLIGHT_POLL_MS	4✔
370	const deadline = Date.now() + timeoutMs	4✔
371
372	this.options.metricsCollector.increment('singleFlightWaits')	4✔
373	this.options.emit('stampede-dedupe', { key })	4✔
374
375	while (Date.now() < deadline) {	4✔
376	const hit = await this.readFromLayers<T>(key, options, 'fresh-only')	12✔
377	if (hit.found) {	12✔
378	this.options.metricsCollector.increment('hits')	2✔
379	return hit.value	2✔
380	}
381	await this.options.sleep(pollIntervalMs)	10✔
382	}
383
384	if (!this.options.singleFlightCoordinator) {	2!
NEW 385	return this.fetchAndPopulate(key, fetcher, options, expectedClearEpoch, expectedKeyEpoch, fetcherContext)	×
386	}
387
388	return this.options.singleFlightCoordinator.execute(	2✔
389	key,
390	this.resolveSingleFlightOptions(),
391	() => this.fetchAndPopulate(key, fetcher, options, expectedClearEpoch, expectedKeyEpoch, fetcherContext),	2✔
NEW 392	() => this.waitForFreshValue(key, fetcher, options, expectedClearEpoch, expectedKeyEpoch, fetcherContext)	×
393	)
394	}
395
396	private async fetchAndPopulate<T>(
397	key: string,
398	fetcher: CacheFetcher<T>,
399	options?: CacheGetOptions,
400	expectedClearEpoch?: number,
401	expectedKeyEpoch?: number,
402	fetcherContext: CacheFetcherContext<T> = {	96✔
403	key,
404	currentValue: undefined,
405	state: 'miss'
406	}
407	): Promise<T \| null> {
408	this.options.circuitBreakerManager.assertClosed(key, options?.circuitBreaker ?? this.options.circuitBreaker)	96✔
409	this.options.metricsCollector.increment('fetches')	96✔
410	const fetchStart = Date.now()	96✔
411	let fetched: T
412
413	try {	96✔
414	fetched = await this.options.fetchRateLimiter.schedule(	96✔
415	options?.fetcherRateLimit ?? this.options.fetcherRateLimit,	183✔
416	{ key, fetcher },
417	() => fetcher(fetcherContext)	93✔
418	)
419	this.options.circuitBreakerManager.recordSuccess(key)	77✔
420	this.options.logger.debug?.('fetch', { key, durationMs: Date.now() - fetchStart })	77✔
421	} catch (error) {
422	this.options.recordCircuitFailure(key, options?.circuitBreaker ?? this.options.circuitBreaker, error)	13✔
423	throw error	13✔
424	}
425
426	if (fetched === null \|\| fetched === undefined) {	77✔
427	if (!this.shouldNegativeCache(options)) {	7✔
428	return null	4✔
429	}
430
431	if (this.options.maintenance.isWriteOutdated(key, expectedClearEpoch, expectedKeyEpoch)) {	3✔
432	this.options.logger.debug?.('skip-negative-store-after-invalidation', {	1✔
433	key,
434	expectedClearEpoch,
435	clearEpoch: this.options.maintenance.currentClearEpoch(),
436	expectedKeyEpoch,
437	keyEpoch: this.options.maintenance.currentKeyEpoch(key)
438	})
439	return null	1✔
440	}
441
442	await this.options.storeEntry(key, 'empty', null, options)	2✔
443	return null	2✔
444	}
445
446	// Conditional caching: skip storage if shouldCache returns false
447	if (options?.shouldCache) {	70✔
448	try {	4✔
449	if (!options.shouldCache(fetched)) {	4✔
450	return fetched	2✔
451	}
452	} catch (error) {
453	this.options.logger.warn?.('shouldCache-error', {	2✔
454	key,
455	error: this.options.formatError(error)
456	})
457	}
458	}
459
460	if (this.options.maintenance.isWriteOutdated(key, expectedClearEpoch, expectedKeyEpoch)) {	68✔
461	this.options.logger.debug?.('skip-store-after-invalidation', {	2✔
462	key,
463	expectedClearEpoch,
464	clearEpoch: this.options.maintenance.currentClearEpoch(),
465	expectedKeyEpoch,
466	keyEpoch: this.options.maintenance.currentKeyEpoch(key)
467	})
468	return fetched	2✔
469	}
470
471	await this.options.storeEntry(key, 'value', fetched, options)	66✔
472	return fetched	61✔
473	}
474
475	runScheduleBackgroundRefresh<T>(
476	key: string,
477	fetcher: CacheFetcher<T>,
478	options?: CacheGetOptions,
479	fetcherContext?: CacheFetcherContext<T>
480	): void {
481	this.scheduleBackgroundRefresh(key, fetcher, options, fetcherContext)	5✔
482	}
483
484	private scheduleBackgroundRefresh<T>(
485	key: string,
486	fetcher: CacheFetcher<T>,
487	options?: CacheGetOptions,
488	fetcherContext: CacheFetcherContext<T> = {	20✔
489	key,
490	currentValue: undefined,
491	state: 'miss'
492	}
493	): void {
494	if (	20✔
495	!shouldStartBackgroundRefresh({
496	isDisconnecting: this.options.isDisconnecting(),
497	hasRefreshInFlight: this.backgroundRefreshes.has(key)
498	})
499	) {
500	return	2✔
501	}
502
503	const clearEpoch = this.options.maintenance.currentClearEpoch()	18✔
504	const keyEpoch = this.options.maintenance.currentKeyEpoch(key)	18✔
505	this.backgroundRefreshAbort.set(key, false)	18✔
506	const refresh = (async () => {	18✔
507	this.options.metricsCollector.increment('refreshes')	18✔
508	try {	18✔
509	if (this.backgroundRefreshAbort.get(key)) return	18!
510	await this.runBackgroundRefresh(key, fetcher, options, clearEpoch, keyEpoch, fetcherContext)	18✔
511	} catch (error) {
512	if (this.backgroundRefreshAbort.get(key)) return	4!
513	this.options.metricsCollector.increment('refreshErrors')	4✔
514	this.options.logger.warn?.('background-refresh-error', {	4✔
515	key,
516	error: this.options.formatError(error)
517	})
518	} finally {
519	this.backgroundRefreshes.delete(key)	16✔
520	this.backgroundRefreshAbort.delete(key)	16✔
521	}
522	})()
523
524	this.backgroundRefreshes.set(key, refresh)	18✔
525	}
526
527	private async runBackgroundRefresh<T>(
528	key: string,
529	fetcher: CacheFetcher<T>,
530	options?: CacheGetOptions,
531	expectedClearEpoch?: number,
532	expectedKeyEpoch?: number,
533	fetcherContext: CacheFetcherContext<T> = {	18✔
534	key,
535	currentValue: undefined,
536	state: 'miss'
537	}
538	): Promise<void> {
539	const timeoutMs = this.options.backgroundRefreshTimeoutMs ?? DEFAULT_BACKGROUND_REFRESH_TIMEOUT_MS	18✔
540	await this.fetchWithGuards(	18✔
541	key,
542	(context) =>
543	this.options.withTimeout(fetcher(context), timeoutMs, () => {	18✔
544	return new Error(`Background refresh timed out after ${timeoutMs}ms for key "${key}".`)	4✔
545	}),
546	options,
547	expectedClearEpoch,
548	expectedKeyEpoch,
549	false,
550	fetcherContext
551	)
552	}
553
554	async runApplyFreshReadPolicies<T>(
555	key: string,
556	hit: {
557	found: true
558	value: T \| null
559	stored: unknown
560	state: 'fresh' \| 'stale-while-revalidate' \| 'stale-if-error'
561	layerIndex: number
562	layerName: string
563	},
564	options: CacheGetOptions \| undefined,
565	fetcher?: CacheFetcher<T>
566	): Promise<void> {
567	return this.applyFreshReadPolicies(key, hit as Extract<ReadHit<T>, { found: true }>, options, fetcher)	4✔
568	}
569
570	private async applyFreshReadPolicies<T>(
571	key: string,
572	hit: Extract<ReadHit<T>, { found: true }>,
573	options: CacheGetOptions \| undefined,
574	fetcher?: CacheFetcher<T>
575	): Promise<void> {
576	const plan = planFreshReadPolicies({	88✔
577	stored: hit.stored,
578	hasFetcher: Boolean(fetcher),
579	slidingTtl: options?.slidingTtl ?? false,	169✔
580	refreshAheadMs:
581	this.options.resolveLayerMs(hit.layerName, options?.refreshAhead, this.options.refreshAhead, 0) ?? 0	96✔
582	})
583
584	if (plan.refreshedStored) {	88✔
585	for (let index = 0; index <= hit.layerIndex; index += 1) {	7✔
586	const layer = this.options.layers[index]	10✔
587	if (!layer \|\| this.options.shouldSkipLayer(layer)) {	10✔
588	continue	1✔
589	}
590
591	try {	9✔
592	await layer.set(key, plan.refreshedStored, plan.refreshedStoredTtl)	9✔
593	} catch (error) {
594	await this.options.handleLayerFailure(layer, 'sliding-ttl', error)	1✔
595	}
596	}
597	}
598
599	if (fetcher && plan.shouldScheduleBackgroundRefresh) {	88✔
600	this.options.scheduleBackgroundRefreshDispatch(key, fetcher, options, this.createFetcherContext(key, hit))	2✔
601	}
602	}
603
604	private createFetcherContext<T>(key: string, hit: Extract<ReadHit<T>, { found: true }>): CacheFetcherContext<T> {
605	return {	21✔
606	key,
607	currentValue: hit.value === null ? undefined : hit.value,	21!
608	state: hit.state,
609	layer: hit.layerName
610	}
611	}
612
613	private resolveSingleFlightOptions(): CacheSingleFlightExecutionOptions {
614	return {	13✔
615	leaseMs: this.options.singleFlightLeaseMs ?? DEFAULT_SINGLE_FLIGHT_LEASE_MS,	25✔
616	waitTimeoutMs: this.options.singleFlightTimeoutMs ?? DEFAULT_SINGLE_FLIGHT_TIMEOUT_MS,	20✔
617	pollIntervalMs: this.options.singleFlightPollMs ?? DEFAULT_SINGLE_FLIGHT_POLL_MS,	20✔
618	renewIntervalMs: this.options.singleFlightRenewIntervalMs
619	}
620	}
621
622	private shouldNegativeCache(options?: CacheGetOptions): boolean {
623	return options?.negativeCache ?? this.options.negativeCaching ?? false	7✔
624	}
625
626	private isNegativeStoredValue(stored: unknown): boolean {
627	return isStoredValueEnvelope(stored) && stored.kind === 'empty'	106✔
628	}
629	}

flyingsquirrel0419 / layercache / 25367879152

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous