29436437414

Committed 15 Jul 2026 05:25PM UTC coverage: 95.8% (-0.08%) from 95.882%

Build # 29436437414

Build Type

Pull #101

github

Committed by

web-flow

Commit Message

Merge cf2ff24f6 into e5107fec2

Pull Request Pull Request #101: Fix cache security hardening

Coverage Stats

2009 of 2163 branches covered (92.88%)

Branch coverage included in aggregate %.

401 of 420 new or added lines in 25 files covered. (95.48%)

2 existing lines in 1 file now uncovered.

3534 of 3623 relevant lines covered (97.54%)

391.04 hits per line

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

92.68

/src/internal/CacheStackMaintenance.ts

import { type CacheWriteBehindOptions, type CacheWriteCoordinationOptions, CacheWriteSaturationError } from '../types'

type WriteBehindOperation = () => Promise<void>
type FlushWriteBehindBatch = (batch: WriteBehindOperation[]) => Promise<void>
type GenerationCleanupTask = (generation: number) => Promise<void>
type GenerationCleanupErrorHandler = (generation: number, error: unknown) => void

const MAX_KEY_EPOCHS = 50_000
const DEFAULT_MAX_PENDING_WRITES = 10_000
const DEFAULT_MAX_ACTIVE_WRITE_KEYS = 10_000
const DEFAULT_MAX_PENDING_WRITES_PER_KEY = 1_000

export class CacheStackMaintenance {
  private readonly keyEpochs = new Map<string, number>()
  private readonly writeBehindQueue: WriteBehindOperation[] = []
  private writeBehindTimer?: ReturnType<typeof setInterval>
  private writeBehindFlushPromise?: Promise<void>
  private generationCleanupPromise?: Promise<void>
  private readonly writeChains = new Map<string, Promise<void>>()
  private readonly pendingWritesByKey = new Map<string, number>()
  private readonly maxPendingWrites: number
  private readonly maxActiveWriteKeys: number
  private readonly maxPendingWritesPerKey: number
  private pendingWriteUnits = 0
  private nextKeyEpoch = 0
  private absentKeyEpoch = 0
  private clearEpoch = 0

  constructor(options: CacheWriteCoordinationOptions = {}) {
    this.maxPendingWrites = options.maxPendingWrites ?? DEFAULT_MAX_PENDING_WRITES
    this.maxActiveWriteKeys = options.maxActiveKeys ?? DEFAULT_MAX_ACTIVE_WRITE_KEYS
    this.maxPendingWritesPerKey = options.maxPendingWritesPerKey ?? DEFAULT_MAX_PENDING_WRITES_PER_KEY
  }

  initializeWriteBehindTimer(
    writeStrategy: 'write-through' | 'write-behind' | undefined,
    options: CacheWriteBehindOptions | undefined,
    flush: () => Promise<void>
  ): void {
    if (writeStrategy !== 'write-behind') {
      return
    }

    const flushIntervalMs = options?.flushIntervalMs
    if (!flushIntervalMs || flushIntervalMs <= 0) {
      return
    }

    this.disposeWriteBehindTimer()
    this.writeBehindTimer = setInterval(() => {
      void flush()
    }, flushIntervalMs)
    this.writeBehindTimer.unref?.()
  }

  disposeWriteBehindTimer(): void {
    if (!this.writeBehindTimer) {
      return
    }

    clearInterval(this.writeBehindTimer)
    this.writeBehindTimer = undefined
  }

  beginClearEpoch(): void {
    this.clearEpoch += 1
    this.keyEpochs.clear()
    this.writeBehindQueue.length = 0
  }

  currentClearEpoch(): number {
    return this.clearEpoch
  }

  currentKeyEpoch(key: string): number {
    return this.keyEpochs.get(key) ?? this.absentKeyEpoch
  }

  bumpKeyEpochs(keys: string[]): void {
    for (const key of keys) {
      this.keyEpochs.delete(key)
      this.keyEpochs.set(key, this.allocateKeyEpoch())
    }
    this.pruneKeyEpochsIfNeeded()
  }

  isWriteOutdated(key: string, expectedClearEpoch?: number, expectedKeyEpoch?: number): boolean {
    if (expectedClearEpoch !== undefined && expectedClearEpoch !== this.clearEpoch) {
      return true
    }

    if (expectedKeyEpoch !== undefined && expectedKeyEpoch !== this.currentKeyEpoch(key)) {
      return true
    }

    return false
  }

  async runFencedWrite(
    key: string,
    expectedClearEpoch: number,
    expectedKeyEpoch: number,
    operation: () => Promise<void>,
    cleanup: () => Promise<void>
  ): Promise<boolean> {
    const run = async (): Promise<boolean> => {
      if (this.isWriteOutdated(key, expectedClearEpoch, expectedKeyEpoch)) return false
      await operation()
      if (!this.isWriteOutdated(key, expectedClearEpoch, expectedKeyEpoch)) return true
      await cleanup()
      return false
    }
    return this.runSerializedWrites([key], run)
  }

  async runSerializedWrites<T>(keys: string[], operation: () => Promise<T>): Promise<T> {
    // One shared tail per touched key is the ordering boundary: it keeps bulk
    // and single-key paths from bypassing each other while admission limits
    // bound the promises and key strings retained by that boundary.
    const uniqueKeys = [...new Set(keys)].sort()
    if (uniqueKeys.length === 0) {
      return operation()
    }

    const writeUnits = uniqueKeys.length
    if (this.pendingWriteUnits + writeUnits > this.maxPendingWrites) {
      throw new CacheWriteSaturationError('pending-writes', this.maxPendingWrites)
    }

    const newActiveKeys = uniqueKeys.filter((key) => !this.writeChains.has(key)).length
    if (this.writeChains.size + newActiveKeys > this.maxActiveWriteKeys) {
      throw new CacheWriteSaturationError('active-keys', this.maxActiveWriteKeys)
    }

    for (const key of uniqueKeys) {
      if ((this.pendingWritesByKey.get(key) ?? 0) + 1 > this.maxPendingWritesPerKey) {
        throw new CacheWriteSaturationError('per-key', this.maxPendingWritesPerKey)
      }
    }

    this.pendingWriteUnits += writeUnits
    for (const key of uniqueKeys) {
      this.pendingWritesByKey.set(key, (this.pendingWritesByKey.get(key) ?? 0) + 1)
    }

    const previousTails = [
      ...new Set(
        uniqueKeys.map((key) => this.writeChains.get(key)).filter((tail): tail is Promise<void> => tail !== undefined)
      )
    ]
    const result = Promise.all(previousTails).then(operation)
    const tail = result.then(
      () => undefined,
      () => undefined
    )
    for (const key of uniqueKeys) {
      this.writeChains.set(key, tail)
    }
    void tail.finally(() => {
      this.pendingWriteUnits -= writeUnits
      for (const key of uniqueKeys) {
        const depth = (this.pendingWritesByKey.get(key) ?? 1) - 1
        if (depth === 0) {
          this.pendingWritesByKey.delete(key)
        } else {
          this.pendingWritesByKey.set(key, depth)
        }
        if (this.writeChains.get(key) === tail) this.writeChains.delete(key)
      }
    })
    return result
  }

  async enqueueWriteBehind(
    operation: WriteBehindOperation,
    options: CacheWriteBehindOptions | undefined,
    flushBatch: FlushWriteBehindBatch
  ): Promise<void> {
    const batchSize = options?.batchSize ?? 100
    const maxQueueSize = options?.maxQueueSize ?? batchSize * 10
    if (this.writeBehindQueue.length >= maxQueueSize) {
      throw new Error(`Write-behind queue limit (${maxQueueSize}) exceeded.`)
    }
    this.writeBehindQueue.push(operation)

    if (this.writeBehindQueue.length >= maxQueueSize) {
      await this.flushWriteBehindQueue(options, flushBatch)
      return
    }

    if (this.writeBehindQueue.length >= batchSize) {
      await this.flushWriteBehindQueue(options, flushBatch)
      return
    }
  }

  async flushWriteBehindQueue(
    options: CacheWriteBehindOptions | undefined,
    flushBatch: FlushWriteBehindBatch
  ): Promise<void> {
    if (this.writeBehindFlushPromise || this.writeBehindQueue.length === 0) {
      await this.writeBehindFlushPromise
      return
    }

    const batchSize = options?.batchSize ?? 100
    const batch = this.writeBehindQueue.splice(0, batchSize)
    this.writeBehindFlushPromise = flushBatch(batch)

    try {
      await this.writeBehindFlushPromise
    } finally {
      this.writeBehindFlushPromise = undefined
    }

    if (this.writeBehindQueue.length > 0) {
      await this.flushWriteBehindQueue(options, flushBatch)
    }
  }

  scheduleGenerationCleanup(
    generation: number,
    task: GenerationCleanupTask,
    onError: GenerationCleanupErrorHandler
  ): void {
    const scheduledTask = (this.generationCleanupPromise ?? Promise.resolve())
      .then(() => task(generation))
      .catch((error) => {
        onError(generation, error)
      })

    this.generationCleanupPromise = scheduledTask.finally(() => {
      if (this.generationCleanupPromise === scheduledTask) {
        this.generationCleanupPromise = undefined
      }
    })
  }

  async waitForGenerationCleanup(): Promise<void> {
    await this.generationCleanupPromise
  }

  private pruneKeyEpochsIfNeeded(): void {
    if (this.keyEpochs.size <= MAX_KEY_EPOCHS) {
      return
    }

    const toDelete = Math.ceil(this.keyEpochs.size * 0.1)
    let deleted = 0
    for (let i = 0; i < toDelete; i++) {
      const oldestKey = this.keyEpochs.keys().next().value
      if (oldestKey === undefined) {
        break
      }
      this.keyEpochs.delete(oldestKey)
      deleted += 1
    }
    // Pruned keys become indistinguishable from never-seen keys. Rotating the
    // absent token invalidates any operation holding their old token instead
    // of letting a pruned key pass an ABA-style stale-write check.
    if (deleted > 0) this.absentKeyEpoch = this.allocateKeyEpoch()
  }

  private allocateKeyEpoch(): number {
    if (this.nextKeyEpoch >= Number.MAX_SAFE_INTEGER) {
      this.clearEpoch += 1
      this.keyEpochs.clear()
      this.nextKeyEpoch = 0
      this.absentKeyEpoch = 0
    }
    this.nextKeyEpoch += 1
    return this.nextKeyEpoch
  }
}

1	import { type CacheWriteBehindOptions, type CacheWriteCoordinationOptions, CacheWriteSaturationError } from '../types'
2
3	type WriteBehindOperation = () => Promise<void>
4	type FlushWriteBehindBatch = (batch: WriteBehindOperation[]) => Promise<void>
5	type GenerationCleanupTask = (generation: number) => Promise<void>
6	type GenerationCleanupErrorHandler = (generation: number, error: unknown) => void
7
8	const MAX_KEY_EPOCHS = 50_000	17✔
9	const DEFAULT_MAX_PENDING_WRITES = 10_000	17✔
10	const DEFAULT_MAX_ACTIVE_WRITE_KEYS = 10_000	17✔
11	const DEFAULT_MAX_PENDING_WRITES_PER_KEY = 1_000	17✔
12
13	export class CacheStackMaintenance {
14	private readonly keyEpochs = new Map<string, number>()	358✔
15	private readonly writeBehindQueue: WriteBehindOperation[] = []	358✔
16	private writeBehindTimer?: ReturnType<typeof setInterval>
17	private writeBehindFlushPromise?: Promise<void>
18	private generationCleanupPromise?: Promise<void>
19	private readonly writeChains = new Map<string, Promise<void>>()	358✔
20	private readonly pendingWritesByKey = new Map<string, number>()	358✔
21	private readonly maxPendingWrites: number
22	private readonly maxActiveWriteKeys: number
23	private readonly maxPendingWritesPerKey: number
24	private pendingWriteUnits = 0	358✔
25	private nextKeyEpoch = 0	358✔
26	private absentKeyEpoch = 0	358✔
27	private clearEpoch = 0	358✔
28
29	constructor(options: CacheWriteCoordinationOptions = {}) {	358✔
30	this.maxPendingWrites = options.maxPendingWrites ?? DEFAULT_MAX_PENDING_WRITES	358✔
31	this.maxActiveWriteKeys = options.maxActiveKeys ?? DEFAULT_MAX_ACTIVE_WRITE_KEYS	358✔
32	this.maxPendingWritesPerKey = options.maxPendingWritesPerKey ?? DEFAULT_MAX_PENDING_WRITES_PER_KEY	358✔
33	}
34
35	initializeWriteBehindTimer(
36	writeStrategy: 'write-through' \| 'write-behind' \| undefined,
37	options: CacheWriteBehindOptions \| undefined,
38	flush: () => Promise<void>
39	): void {
40	if (writeStrategy !== 'write-behind') {	274✔
41	return	267✔
42	}
43
44	const flushIntervalMs = options?.flushIntervalMs	7✔
45	if (!flushIntervalMs \|\| flushIntervalMs <= 0) {	274✔
46	return	5✔
47	}
48
49	this.disposeWriteBehindTimer()	2✔
50	this.writeBehindTimer = setInterval(() => {	2✔
51	void flush()	1✔
52	}, flushIntervalMs)
53	this.writeBehindTimer.unref?.()	2✔
54	}
55
56	disposeWriteBehindTimer(): void {
57	if (!this.writeBehindTimer) {	31✔
58	return	29✔
59	}
60
61	clearInterval(this.writeBehindTimer)	2✔
62	this.writeBehindTimer = undefined	2✔
63	}
64
65	beginClearEpoch(): void {
66	this.clearEpoch += 1	9✔
67	this.keyEpochs.clear()	9✔
68	this.writeBehindQueue.length = 0	9✔
69	}
70
71	currentClearEpoch(): number {
72	return this.clearEpoch	1,263✔
73	}
74
75	currentKeyEpoch(key: string): number {
76	return this.keyEpochs.get(key) ?? this.absentKeyEpoch	2,283✔
77	}
78
79	bumpKeyEpochs(keys: string[]): void {
80	for (const key of keys) {	66✔
81	this.keyEpochs.delete(key)	100,081✔
82	this.keyEpochs.set(key, this.allocateKeyEpoch())	100,081✔
83	}
84	this.pruneKeyEpochsIfNeeded()	66✔
85	}
86
87	isWriteOutdated(key: string, expectedClearEpoch?: number, expectedKeyEpoch?: number): boolean {
88	if (expectedClearEpoch !== undefined && expectedClearEpoch !== this.clearEpoch) {	993✔
89	return true	3✔
90	}
91
92	if (expectedKeyEpoch !== undefined && expectedKeyEpoch !== this.currentKeyEpoch(key)) {	990✔
93	return true	10✔
94	}
95
96	return false	980✔
97	}
98
99	async runFencedWrite(
100	key: string,
101	expectedClearEpoch: number,
102	expectedKeyEpoch: number,
103	operation: () => Promise<void>,
104	cleanup: () => Promise<void>
105	): Promise<boolean> {
106	const run = async (): Promise<boolean> => {	221✔
107	if (this.isWriteOutdated(key, expectedClearEpoch, expectedKeyEpoch)) return false	219✔
108	await operation()	218✔
109	if (!this.isWriteOutdated(key, expectedClearEpoch, expectedKeyEpoch)) return true	216✔
110	await cleanup()	4✔
111	return false	4✔
112	}
113	return this.runSerializedWrites([key], run)	221✔
114	}
115
116	async runSerializedWrites<T>(keys: string[], operation: () => Promise<T>): Promise<T> {
117	// One shared tail per touched key is the ordering boundary: it keeps bulk
118	// and single-key paths from bypassing each other while admission limits
119	// bound the promises and key strings retained by that boundary.
120	const uniqueKeys = [...new Set(keys)].sort()	246✔
121	if (uniqueKeys.length === 0) {	246!
NEW 122	return operation()	×
123	}
124
125	const writeUnits = uniqueKeys.length	246✔
126	if (this.pendingWriteUnits + writeUnits > this.maxPendingWrites) {	246✔
127	throw new CacheWriteSaturationError('pending-writes', this.maxPendingWrites)	1✔
128	}
129
130	const newActiveKeys = uniqueKeys.filter((key) => !this.writeChains.has(key)).length	266✔
131	if (this.writeChains.size + newActiveKeys > this.maxActiveWriteKeys) {	245✔
132	throw new CacheWriteSaturationError('active-keys', this.maxActiveWriteKeys)	1✔
133	}
134
135	for (const key of uniqueKeys) {	244✔
136	if ((this.pendingWritesByKey.get(key) ?? 0) + 1 > this.maxPendingWritesPerKey) {	265✔
137	throw new CacheWriteSaturationError('per-key', this.maxPendingWritesPerKey)	1✔
138	}
139	}
140
141	this.pendingWriteUnits += writeUnits	243✔
142	for (const key of uniqueKeys) {	243✔
143	this.pendingWritesByKey.set(key, (this.pendingWritesByKey.get(key) ?? 0) + 1)	264✔
144	}
145
146	const previousTails = [	243✔
147	...new Set(
148	uniqueKeys.map((key) => this.writeChains.get(key)).filter((tail): tail is Promise<void> => tail !== undefined)	264✔
149	)
150	]
151	const result = Promise.all(previousTails).then(operation)	243✔
152	const tail = result.then(	243✔
153	() => undefined,	240✔
154	() => undefined	3✔
155	)
156	for (const key of uniqueKeys) {	243✔
157	this.writeChains.set(key, tail)	264✔
158	}
159	void tail.finally(() => {	243✔
160	this.pendingWriteUnits -= writeUnits	243✔
161	for (const key of uniqueKeys) {	243✔
162	const depth = (this.pendingWritesByKey.get(key) ?? 1) - 1	264!
163	if (depth === 0) {	264✔
164	this.pendingWritesByKey.delete(key)	262✔
165	} else {
166	this.pendingWritesByKey.set(key, depth)	2✔
167	}
168	if (this.writeChains.get(key) === tail) this.writeChains.delete(key)	264✔
169	}
170	})
171	return result	243✔
172	}
173
174	async enqueueWriteBehind(
175	operation: WriteBehindOperation,
176	options: CacheWriteBehindOptions \| undefined,
177	flushBatch: FlushWriteBehindBatch
178	): Promise<void> {
179	const batchSize = options?.batchSize ?? 100	19!
180	const maxQueueSize = options?.maxQueueSize ?? batchSize * 10	19✔
181	if (this.writeBehindQueue.length >= maxQueueSize) {	19✔
182	throw new Error(`Write-behind queue limit (${maxQueueSize}) exceeded.`)	1✔
183	}
184	this.writeBehindQueue.push(operation)	18✔
185
186	if (this.writeBehindQueue.length >= maxQueueSize) {	18✔
187	await this.flushWriteBehindQueue(options, flushBatch)	2✔
188	return	2✔
189	}
190
191	if (this.writeBehindQueue.length >= batchSize) {	16✔
192	await this.flushWriteBehindQueue(options, flushBatch)	5✔
193	return	5✔
194	}
195	}
196
197	async flushWriteBehindQueue(
198	options: CacheWriteBehindOptions \| undefined,
199	flushBatch: FlushWriteBehindBatch
200	): Promise<void> {
201	if (this.writeBehindFlushPromise \|\| this.writeBehindQueue.length === 0) {	41✔
202	await this.writeBehindFlushPromise	31✔
203	return	31✔
204	}
205
206	const batchSize = options?.batchSize ?? 100	10!
207	const batch = this.writeBehindQueue.splice(0, batchSize)	41✔
208	this.writeBehindFlushPromise = flushBatch(batch)	41✔
209
210	try {	41✔
211	await this.writeBehindFlushPromise	41✔
212	} finally {
213	this.writeBehindFlushPromise = undefined	10✔
214	}
215
216	if (this.writeBehindQueue.length > 0) {	10✔
217	await this.flushWriteBehindQueue(options, flushBatch)	1✔
218	}
219	}
220
221	scheduleGenerationCleanup(
222	generation: number,
223	task: GenerationCleanupTask,
224	onError: GenerationCleanupErrorHandler
225	): void {
226	const scheduledTask = (this.generationCleanupPromise ?? Promise.resolve())	6✔
227	.then(() => task(generation))	6✔
228	.catch((error) => {
229	onError(generation, error)	2✔
230	})
231
232	this.generationCleanupPromise = scheduledTask.finally(() => {	6✔
233	if (this.generationCleanupPromise === scheduledTask) {	6!
234	this.generationCleanupPromise = undefined	×
235	}
236	})
237	}
238
239	async waitForGenerationCleanup(): Promise<void> {
240	await this.generationCleanupPromise	30✔
241	}
242
243	private pruneKeyEpochsIfNeeded(): void {
244	if (this.keyEpochs.size <= MAX_KEY_EPOCHS) {	66✔
245	return	62✔
246	}
247
248	const toDelete = Math.ceil(this.keyEpochs.size * 0.1)	4✔
249	let deleted = 0	4✔
250	for (let i = 0; i < toDelete; i++) {	4✔
251	const oldestKey = this.keyEpochs.keys().next().value	20,004✔
252	if (oldestKey === undefined) {	20,004!
253	break	×
254	}
255	this.keyEpochs.delete(oldestKey)	20,004✔
256	deleted += 1	20,004✔
257	}
258	// Pruned keys become indistinguishable from never-seen keys. Rotating the
259	// absent token invalidates any operation holding their old token instead
260	// of letting a pruned key pass an ABA-style stale-write check.
261	if (deleted > 0) this.absentKeyEpoch = this.allocateKeyEpoch()	4!
262	}
263
264	private allocateKeyEpoch(): number {
265	if (this.nextKeyEpoch >= Number.MAX_SAFE_INTEGER) {	100,085!
NEW 266	this.clearEpoch += 1	×
NEW 267	this.keyEpochs.clear()	×
NEW 268	this.nextKeyEpoch = 0	×
NEW 269	this.absentKeyEpoch = 0	×
270	}
271	this.nextKeyEpoch += 1	100,085✔
272	return this.nextKeyEpoch	100,085✔
273	}
274	}

flyingsquirrel0419 / layercache / 29436437414

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