25300270651

Committed 04 May 2026 03:57AM UTC coverage: 95.729%. First build

Build # 25300270651

Build Type

Pull #65

github

Committed by

web-flow

Commit Message

Merge 4de86efa1 into 04d66ec8b

Pull Request Pull Request #65: Reacquire distributed single-flight lock after waiter timeout

Coverage Stats

1715 of 1845 branches covered (92.95%)

Branch coverage included in aggregate %.

3 of 5 new or added lines in 1 file covered. (60.0%)

3104 of 3189 relevant lines covered (97.33%)

323.71 hits per line

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

97.79

/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
    }

    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)
      try {
        await layer.set(key, stored, ttl)
      } catch (error) {
        await this.options.handleLayerFailure(layer, 'backfill', error)
        continue
      }
      this.options.metricsCollector.increment('backfills')
      this.options.logger.debug?.('backfill', { key, layer: layer.name })
      this.options.emit('backfill', { key, layer: layer.name })
    }
  }

  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	13✔
27	const DEFAULT_SINGLE_FLIGHT_TIMEOUT_MS = 5_000	13✔
28	const DEFAULT_SINGLE_FLIGHT_POLL_MS = 50	13✔
29	const DEFAULT_BACKGROUND_REFRESH_TIMEOUT_MS = 30_000	13✔
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>>()	294✔
97	private readonly backgroundRefreshAbort = new Map<string, boolean>()	294✔
98
99	constructor(private readonly options: CacheStackReaderOptions) {}	294✔
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)) {	498✔
171	return null	3✔
172	}
173
174	if (layer.getEntry) {	495✔
175	try {	431✔
176	return await layer.getEntry(key)	431✔
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) {	125✔
191	return	106✔
192	}
193
194	for (let index = 0; index <= upToIndex; index += 1) {	19✔
195	const layer = this.options.layers[index]	22✔
196	if (!layer \|\| this.options.shouldSkipLayer(layer)) {	22✔
197	continue	4✔
198	}
199
200	const ttl =	18✔
201	remainingStoredTtlMs(stored) ??
202	this.options.resolveLayerMs(layer.name, options?.ttl, undefined, layer.defaultTtl)
203	try {	22✔
204	await layer.set(key, stored, ttl)	22✔
205	} catch (error) {
206	await this.options.handleLayerFailure(layer, 'backfill', error)	1✔
207	continue	1✔
208	}
209	this.options.metricsCollector.increment('backfills')	17✔
210	this.options.logger.debug?.('backfill', { key, layer: layer.name })	17✔
211	this.options.emit('backfill', { key, layer: layer.name })	22✔
212	}
213	}
214
215	abortAllRefreshes(): void {
216	for (const key of this.backgroundRefreshAbort.keys()) {	29✔
217	this.backgroundRefreshAbort.set(key, true)	1✔
218	}
219	}
220
221	getAllRefreshPromises(): Promise<void>[] {
222	return [...this.backgroundRefreshes.values()]	31✔
223	}
224
225	private async readFromLayers<T>(
226	key: string,
227	options: CacheGetOptions \| undefined,
228	mode: ReadMode
229	): Promise<ReadHit<T>> {
230	let sawRetainableValue = false	436✔
231
232	for (let index = 0; index < this.options.layers.length; index += 1) {	436✔
233	const layer = this.options.layers[index]	458✔
234	if (!layer) continue	458!
235	const readStart = performance.now()	458✔
236	const stored = await this.readLayerEntry(layer, key)	458✔
237	const readDuration = performance.now() - readStart	458✔
238	this.options.metricsCollector.recordLatency(layer.name, readDuration)	458✔
239	if (stored === null) {	458✔
240	this.options.metricsCollector.incrementLayer('missesByLayer', layer.name)	326✔
241	continue	326✔
242	}
243
244	const resolved = resolveStoredValue<T>(stored)	132✔
245	if (resolved.state === 'expired') {	132✔
246	await layer.delete(key)	2✔
247	continue	2✔
248	}
249
250	sawRetainableValue = true	130✔
251
252	if (mode === 'fresh-only' && resolved.state !== 'fresh') {	130✔
253	continue	20✔
254	}
255
256	await this.options.tagIndex.touch(key)	110✔
257	await this.backfill(key, stored, index - 1, options)	110✔
258	this.options.metricsCollector.incrementLayer('hitsByLayer', layer.name)	110✔
259	this.options.logger.debug?.('hit', { key, layer: layer.name, state: resolved.state })	110✔
260	this.options.emit('hit', {	458✔
261	key,
262	layer: layer.name,
263	state: resolved.state as CacheStackEvents['hit']['state']
264	})
265	return {	458✔
266	found: true,
267	value: resolved.value,
268	stored,
269	state: resolved.state,
270	layerIndex: index,
271	layerName: layer.name
272	}
273	}
274
275	if (!sawRetainableValue) {	326✔
276	await this.options.tagIndex.remove(key)	307✔
277	}
278
279	this.options.logger.debug?.('miss', { key, mode })	326✔
280	this.options.emit('miss', { key, mode })	436✔
281	return { found: false, value: null, stored: null, state: 'miss' }	436✔
282	}
283
284	private async fetchWithGuards<T>(
285	key: string,
286	fetcher: CacheFetcher<T>,
287	options?: CacheGetOptions,
288	expectedClearEpoch?: number,
289	expectedKeyEpoch?: number,
290	initialMissConfirmed = false,	175✔
291	fetcherContext: CacheFetcherContext<T> = {	175✔
292	key,
293	currentValue: undefined,
294	state: 'miss'
295	}
296	): Promise<T \| null> {
297	const fetchTask = async (): Promise<T \| null> => {	175✔
298	const shouldRecheckFreshLayers = !(initialMissConfirmed && this.options.singleFlightCoordinator)	96✔
299	if (shouldRecheckFreshLayers) {	96✔
300	const secondHit = await this.readFromLayers<T>(key, options, 'fresh-only')	90✔
301	if (secondHit.found) {	90✔
302	this.options.metricsCollector.increment('hits')	2✔
303	return secondHit.value	2✔
304	}
305	}
306
307	return this.fetchAndPopulate(key, fetcher, options, expectedClearEpoch, expectedKeyEpoch, fetcherContext)	94✔
308	}
309
310	const singleFlightTask = async (): Promise<T \| null> => {	175✔
311	if (!this.options.singleFlightCoordinator) {	101✔
312	return fetchTask()	90✔
313	}
314
315	try {	11✔
316	return await this.options.singleFlightCoordinator.execute(	11✔
317	key,
318	this.resolveSingleFlightOptions(),
319	fetchTask,
320	() => this.waitForFreshValue(key, fetcher, options, expectedClearEpoch, expectedKeyEpoch, fetcherContext)	4✔
321	)
322	} catch (error) {
323	if (!this.options.isGracefulDegradationEnabled()) {	3✔
324	throw error	1✔
325	}
326
327	this.options.metricsCollector.increment('degradedOperations')	2✔
328	this.options.logger.warn?.('single-flight-coordinator-degraded', {	2✔
329	key,
330	error: this.options.formatError(error)
331	})
332	this.options.emitError('single-flight', {	3✔
333	key,
334	degraded: true,
335	error: this.options.formatError(error)
336	})
337	return fetchTask()	3✔
338	}
339	}
340
341	if (this.options.stampedePrevention === false) {	175✔
342	return singleFlightTask()	3✔
343	}
344
345	return this.options.stampedeGuard.execute(key, singleFlightTask)	172✔
346	}
347
348	private async waitForFreshValue<T>(
349	key: string,
350	fetcher: CacheFetcher<T>,
351	options?: CacheGetOptions,
352	expectedClearEpoch?: number,
353	expectedKeyEpoch?: number,
354	fetcherContext: CacheFetcherContext<T> = {	4✔
355	key,
356	currentValue: undefined,
357	state: 'miss'
358	}
359	): Promise<T \| null> {
360	const timeoutMs = this.options.singleFlightTimeoutMs ?? DEFAULT_SINGLE_FLIGHT_TIMEOUT_MS	4✔
361	const pollIntervalMs = this.options.singleFlightPollMs ?? DEFAULT_SINGLE_FLIGHT_POLL_MS	4✔
362	const deadline = Date.now() + timeoutMs	4✔
363
364	this.options.metricsCollector.increment('singleFlightWaits')	4✔
365	this.options.emit('stampede-dedupe', { key })	4✔
366
367	while (Date.now() < deadline) {	4✔
368	const hit = await this.readFromLayers<T>(key, options, 'fresh-only')	13✔
369	if (hit.found) {	13✔
370	this.options.metricsCollector.increment('hits')	2✔
371	return hit.value	2✔
372	}
373	await this.options.sleep(pollIntervalMs)	11✔
374	}
375
376	if (!this.options.singleFlightCoordinator) {	2!
NEW 377	return this.fetchAndPopulate(key, fetcher, options, expectedClearEpoch, expectedKeyEpoch, fetcherContext)	×
378	}
379
380	return this.options.singleFlightCoordinator.execute(	2✔
381	key,
382	this.resolveSingleFlightOptions(),
383	() => this.fetchAndPopulate(key, fetcher, options, expectedClearEpoch, expectedKeyEpoch, fetcherContext),	2✔
NEW 384	() => this.waitForFreshValue(key, fetcher, options, expectedClearEpoch, expectedKeyEpoch, fetcherContext)	×
385	)
386	}
387
388	private async fetchAndPopulate<T>(
389	key: string,
390	fetcher: CacheFetcher<T>,
391	options?: CacheGetOptions,
392	expectedClearEpoch?: number,
393	expectedKeyEpoch?: number,
394	fetcherContext: CacheFetcherContext<T> = {	96✔
395	key,
396	currentValue: undefined,
397	state: 'miss'
398	}
399	): Promise<T \| null> {
400	this.options.circuitBreakerManager.assertClosed(key, options?.circuitBreaker ?? this.options.circuitBreaker)	96✔
401	this.options.metricsCollector.increment('fetches')	96✔
402	const fetchStart = Date.now()	96✔
403	let fetched: T
404
405	try {	96✔
406	fetched = await this.options.fetchRateLimiter.schedule(	96✔
407	options?.fetcherRateLimit ?? this.options.fetcherRateLimit,	183✔
408	{ key, fetcher },
409	() => fetcher(fetcherContext)	93✔
410	)
411	this.options.circuitBreakerManager.recordSuccess(key)	77✔
412	this.options.logger.debug?.('fetch', { key, durationMs: Date.now() - fetchStart })	77✔
413	} catch (error) {
414	this.options.recordCircuitFailure(key, options?.circuitBreaker ?? this.options.circuitBreaker, error)	13✔
415	throw error	13✔
416	}
417
418	if (fetched === null \|\| fetched === undefined) {	77✔
419	if (!this.shouldNegativeCache(options)) {	7✔
420	return null	4✔
421	}
422
423	if (this.options.maintenance.isWriteOutdated(key, expectedClearEpoch, expectedKeyEpoch)) {	3✔
424	this.options.logger.debug?.('skip-negative-store-after-invalidation', {	1✔
425	key,
426	expectedClearEpoch,
427	clearEpoch: this.options.maintenance.currentClearEpoch(),
428	expectedKeyEpoch,
429	keyEpoch: this.options.maintenance.currentKeyEpoch(key)
430	})
431	return null	1✔
432	}
433
434	await this.options.storeEntry(key, 'empty', null, options)	2✔
435	return null	2✔
436	}
437
438	// Conditional caching: skip storage if shouldCache returns false
439	if (options?.shouldCache) {	70✔
440	try {	4✔
441	if (!options.shouldCache(fetched)) {	4✔
442	return fetched	2✔
443	}
444	} catch (error) {
445	this.options.logger.warn?.('shouldCache-error', {	2✔
446	key,
447	error: this.options.formatError(error)
448	})
449	}
450	}
451
452	if (this.options.maintenance.isWriteOutdated(key, expectedClearEpoch, expectedKeyEpoch)) {	68✔
453	this.options.logger.debug?.('skip-store-after-invalidation', {	2✔
454	key,
455	expectedClearEpoch,
456	clearEpoch: this.options.maintenance.currentClearEpoch(),
457	expectedKeyEpoch,
458	keyEpoch: this.options.maintenance.currentKeyEpoch(key)
459	})
460	return fetched	2✔
461	}
462
463	await this.options.storeEntry(key, 'value', fetched, options)	66✔
464	return fetched	61✔
465	}
466
467	runScheduleBackgroundRefresh<T>(
468	key: string,
469	fetcher: CacheFetcher<T>,
470	options?: CacheGetOptions,
471	fetcherContext?: CacheFetcherContext<T>
472	): void {
473	this.scheduleBackgroundRefresh(key, fetcher, options, fetcherContext)	5✔
474	}
475
476	private scheduleBackgroundRefresh<T>(
477	key: string,
478	fetcher: CacheFetcher<T>,
479	options?: CacheGetOptions,
480	fetcherContext: CacheFetcherContext<T> = {	20✔
481	key,
482	currentValue: undefined,
483	state: 'miss'
484	}
485	): void {
486	if (	20✔
487	!shouldStartBackgroundRefresh({
488	isDisconnecting: this.options.isDisconnecting(),
489	hasRefreshInFlight: this.backgroundRefreshes.has(key)
490	})
491	) {
492	return	2✔
493	}
494
495	const clearEpoch = this.options.maintenance.currentClearEpoch()	18✔
496	const keyEpoch = this.options.maintenance.currentKeyEpoch(key)	18✔
497	this.backgroundRefreshAbort.set(key, false)	18✔
498	const refresh = (async () => {	18✔
499	this.options.metricsCollector.increment('refreshes')	18✔
500	try {	18✔
501	if (this.backgroundRefreshAbort.get(key)) return	18!
502	await this.runBackgroundRefresh(key, fetcher, options, clearEpoch, keyEpoch, fetcherContext)	18✔
503	} catch (error) {
504	if (this.backgroundRefreshAbort.get(key)) return	4!
505	this.options.metricsCollector.increment('refreshErrors')	4✔
506	this.options.logger.warn?.('background-refresh-error', {	4✔
507	key,
508	error: this.options.formatError(error)
509	})
510	} finally {
511	this.backgroundRefreshes.delete(key)	16✔
512	this.backgroundRefreshAbort.delete(key)	16✔
513	}
514	})()
515
516	this.backgroundRefreshes.set(key, refresh)	18✔
517	}
518
519	private async runBackgroundRefresh<T>(
520	key: string,
521	fetcher: CacheFetcher<T>,
522	options?: CacheGetOptions,
523	expectedClearEpoch?: number,
524	expectedKeyEpoch?: number,
525	fetcherContext: CacheFetcherContext<T> = {	18✔
526	key,
527	currentValue: undefined,
528	state: 'miss'
529	}
530	): Promise<void> {
531	const timeoutMs = this.options.backgroundRefreshTimeoutMs ?? DEFAULT_BACKGROUND_REFRESH_TIMEOUT_MS	18✔
532	await this.fetchWithGuards(	18✔
533	key,
534	(context) =>
535	this.options.withTimeout(fetcher(context), timeoutMs, () => {	18✔
536	return new Error(`Background refresh timed out after ${timeoutMs}ms for key "${key}".`)	4✔
537	}),
538	options,
539	expectedClearEpoch,
540	expectedKeyEpoch,
541	false,
542	fetcherContext
543	)
544	}
545
546	async runApplyFreshReadPolicies<T>(
547	key: string,
548	hit: {
549	found: true
550	value: T \| null
551	stored: unknown
552	state: 'fresh' \| 'stale-while-revalidate' \| 'stale-if-error'
553	layerIndex: number
554	layerName: string
555	},
556	options: CacheGetOptions \| undefined,
557	fetcher?: CacheFetcher<T>
558	): Promise<void> {
559	return this.applyFreshReadPolicies(key, hit as Extract<ReadHit<T>, { found: true }>, options, fetcher)	4✔
560	}
561
562	private async applyFreshReadPolicies<T>(
563	key: string,
564	hit: Extract<ReadHit<T>, { found: true }>,
565	options: CacheGetOptions \| undefined,
566	fetcher?: CacheFetcher<T>
567	): Promise<void> {
568	const plan = planFreshReadPolicies({	88✔
569	stored: hit.stored,
570	hasFetcher: Boolean(fetcher),
571	slidingTtl: options?.slidingTtl ?? false,	169✔
572	refreshAheadMs:
573	this.options.resolveLayerMs(hit.layerName, options?.refreshAhead, this.options.refreshAhead, 0) ?? 0	96✔
574	})
575
576	if (plan.refreshedStored) {	88✔
577	for (let index = 0; index <= hit.layerIndex; index += 1) {	7✔
578	const layer = this.options.layers[index]	10✔
579	if (!layer \|\| this.options.shouldSkipLayer(layer)) {	10✔
580	continue	1✔
581	}
582
583	try {	9✔
584	await layer.set(key, plan.refreshedStored, plan.refreshedStoredTtl)	9✔
585	} catch (error) {
586	await this.options.handleLayerFailure(layer, 'sliding-ttl', error)	1✔
587	}
588	}
589	}
590
591	if (fetcher && plan.shouldScheduleBackgroundRefresh) {	88✔
592	this.options.scheduleBackgroundRefreshDispatch(key, fetcher, options, this.createFetcherContext(key, hit))	2✔
593	}
594	}
595
596	private createFetcherContext<T>(key: string, hit: Extract<ReadHit<T>, { found: true }>): CacheFetcherContext<T> {
597	return {	21✔
598	key,
599	currentValue: hit.value === null ? undefined : hit.value,	21!
600	state: hit.state,
601	layer: hit.layerName
602	}
603	}
604
605	private resolveSingleFlightOptions(): CacheSingleFlightExecutionOptions {
606	return {	13✔
607	leaseMs: this.options.singleFlightLeaseMs ?? DEFAULT_SINGLE_FLIGHT_LEASE_MS,	25✔
608	waitTimeoutMs: this.options.singleFlightTimeoutMs ?? DEFAULT_SINGLE_FLIGHT_TIMEOUT_MS,	20✔
609	pollIntervalMs: this.options.singleFlightPollMs ?? DEFAULT_SINGLE_FLIGHT_POLL_MS,	20✔
610	renewIntervalMs: this.options.singleFlightRenewIntervalMs
611	}
612	}
613
614	private shouldNegativeCache(options?: CacheGetOptions): boolean {
615	return options?.negativeCache ?? this.options.negativeCaching ?? false	7✔
616	}
617
618	private isNegativeStoredValue(stored: unknown): boolean {
619	return isStoredValueEnvelope(stored) && stored.kind === 'empty'	106✔
620	}
621	}

flyingsquirrel0419 / layercache / 25300270651

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