From 19c53d2cc43bc81a4e6d19ab2ef2a09f65a47c0e Mon Sep 17 00:00:00 2001 From: Muhammad Zayyad Mukhtar <95658387+El-swaggerito@users.noreply.github.com> Date: Sun, 30 Aug 2026 18:48:14 +0100 Subject: [PATCH] Implemented the generic in-flight request deduplication primitive --- src/index.ts | 1 + src/utils/index.ts | 1 + src/utils/requestDeduplicator.ts | 178 +++++++++++ tests/unit/requestDeduplicator.test.ts | 420 +++++++++++++++++++++++++ 4 files changed, 600 insertions(+) create mode 100644 src/utils/requestDeduplicator.ts create mode 100644 tests/unit/requestDeduplicator.test.ts diff --git a/src/index.ts b/src/index.ts index 801c3a2..3740fd9 100644 --- a/src/index.ts +++ b/src/index.ts @@ -12,6 +12,7 @@ export * from "./types/index.js"; export * from "./utils/querySerializer.js"; export * from "./utils/batchPlanner.js"; export * from "./utils/pagination.js"; +export * from "./utils/requestDeduplicator.js"; export * from "./time/index.js"; export * from "./stellar/index.js"; export * from "./transport/index.js"; diff --git a/src/utils/index.ts b/src/utils/index.ts index d6095d9..a9254f5 100644 --- a/src/utils/index.ts +++ b/src/utils/index.ts @@ -1,3 +1,4 @@ export * from "./querySerializer.js"; export * from "./batchPlanner.js"; export * from "./pagination.js"; +export * from "./requestDeduplicator.js"; diff --git a/src/utils/requestDeduplicator.ts b/src/utils/requestDeduplicator.ts new file mode 100644 index 0000000..30b2e03 --- /dev/null +++ b/src/utils/requestDeduplicator.ts @@ -0,0 +1,178 @@ +/** + * Configuration options for {@link RequestDeduplicator}. + */ +export interface RequestDeduplicatorOptions { + /** + * Maximum number of concurrent in-flight requests. + * When exceeded, new requests throw a {@link CapacityExceededError}. + * @default 100 + */ + readonly maxInFlight?: number; + + /** + * Optional retention time in milliseconds for completed results. + * If set, successful results are cached for this duration after completion. + * This is separate from in-flight deduplication and does not affect failure handling. + * Set to 0 (default) to disable retention and remove entries immediately after completion. + * @default 0 + */ + readonly retentionMs?: number; +} + +/** + * An asynchronous producer function that generates a result. + */ +export type AsyncProducer = () => Promise; + +/** + * Error thrown when the deduplicator exceeds its configured capacity. + */ +export class CapacityExceededError extends Error { + constructor(message: string) { + super(message); + this.name = "CapacityExceededError"; + } +} + +/** + * A generic in-flight request deduplication primitive. + * + * Ensures concurrent callers using the same deduplication key share one underlying + * asynchronous execution. Entries are removed after completion (success or failure) + * unless retention is configured. + * + * @example + * ```ts + * const deduplicator = new RequestDeduplicator(); + * + * // Both calls share the same execution + * const [result1, result2] = await Promise.all([ + * deduplicator.execute("user:123", () => fetchUser("123")), + * deduplicator.execute("user:123", () => fetchUser("123")), + * ]); + * ``` + * + * @typeParam T - The type of value produced by the async producer. + */ +export class RequestDeduplicator { + private readonly inFlight = new Map>(); + private readonly retainedResults = new Map(); + private readonly maxInFlight: number; + private readonly retentionMs: number; + + constructor(options: RequestDeduplicatorOptions = {}) { + this.maxInFlight = options.maxInFlight ?? 100; + this.retentionMs = options.retentionMs ?? 0; + + if (this.maxInFlight <= 0 || !Number.isInteger(this.maxInFlight)) { + throw new Error("maxInFlight must be a positive integer"); + } + + if (this.retentionMs < 0 || !Number.isInteger(this.retentionMs)) { + throw new Error("retentionMs must be a non-negative integer"); + } + } + + /** + * Executes the producer for the given key, deduplicating concurrent calls. + * + * If a request with the same key is already in-flight, the existing promise + * is returned. Otherwise, a new promise is created and stored. + * + * @param key - A deterministic string key identifying the logical operation. + * @param producer - An async function that produces the result. + * @returns A promise that resolves to the producer's result. + * @throws {CapacityExceededError} When maxInFlight capacity is exceeded. + */ + async execute(key: string, producer: AsyncProducer): Promise { + // Check for retained result first + const retained = this.retainedResults.get(key); + if (retained) { + if (Date.now() < retained.expiresAt) { + return retained.result; + } + this.retainedResults.delete(key); + } + + // Check for existing in-flight request + const existing = this.inFlight.get(key); + if (existing) { + return existing; + } + + // Check capacity before creating new entry + if (this.inFlight.size >= this.maxInFlight) { + throw new CapacityExceededError( + `Maximum in-flight capacity (${this.maxInFlight}) exceeded for key "${key}"`, + ); + } + + // Create and store the new promise + const promise = producer() + .then((result) => { + this.inFlight.delete(key); + + // Schedule retention if configured + if (this.retentionMs > 0) { + this.retainedResults.set(key, { + result, + expiresAt: Date.now() + this.retentionMs, + }); + } + + return result; + }) + .catch((error) => { + // Always clean up on failure to allow retries + this.inFlight.delete(key); + throw error; + }); + + this.inFlight.set(key, promise); + return promise; + } + + /** + * Returns the current number of in-flight requests. + * + * This is useful for diagnostics and monitoring. + */ + getInFlightCount(): number { + return this.inFlight.size; + } + + /** + * Returns the current number of retained results. + * + * This is useful for diagnostics when retention is enabled. + */ + getRetainedCount(): number { + // Clean up expired entries before reporting + this.cleanupExpiredRetained(); + return this.retainedResults.size; + } + + /** + * Clears all in-flight requests and retained results. + * + * In-flight promises are not cancelled, but their entries are removed. + * This should be used with caution as it may lead to duplicate executions. + */ + clear(): void { + this.inFlight.clear(); + this.retainedResults.clear(); + } + + /** + * Cleans up expired retained entries. + * This is called automatically by getRetainedCount() and can be called manually. + */ + private cleanupExpiredRetained(): void { + const now = Date.now(); + for (const [key, entry] of this.retainedResults) { + if (now >= entry.expiresAt) { + this.retainedResults.delete(key); + } + } + } +} diff --git a/tests/unit/requestDeduplicator.test.ts b/tests/unit/requestDeduplicator.test.ts new file mode 100644 index 0000000..8f14ccb --- /dev/null +++ b/tests/unit/requestDeduplicator.test.ts @@ -0,0 +1,420 @@ +import { describe, it, expect, vi, beforeEach } from 'vitest'; +import { + RequestDeduplicator, + RequestDeduplicatorOptions, + AsyncProducer, + CapacityExceededError, +} from '../../src/utils/requestDeduplicator.js'; + +describe('RequestDeduplicator', () => { + describe('constructor validation', () => { + it('should accept default options', () => { + const deduplicator = new RequestDeduplicator(); + expect(deduplicator.getInFlightCount()).toBe(0); + }); + + it('should accept valid maxInFlight', () => { + const deduplicator = new RequestDeduplicator({ maxInFlight: 50 }); + expect(deduplicator.getInFlightCount()).toBe(0); + }); + + it('should throw for non-positive maxInFlight', () => { + expect(() => new RequestDeduplicator({ maxInFlight: 0 })).toThrow('maxInFlight must be a positive integer'); + expect(() => new RequestDeduplicator({ maxInFlight: -5 })).toThrow('maxInFlight must be a positive integer'); + }); + + it('should throw for non-integer maxInFlight', () => { + expect(() => new RequestDeduplicator({ maxInFlight: 1.5 })).toThrow('maxInFlight must be a positive integer'); + }); + + it('should accept valid retentionMs', () => { + const deduplicator = new RequestDeduplicator({ retentionMs: 1000 }); + expect(deduplicator.getRetainedCount()).toBe(0); + }); + + it('should throw for negative retentionMs', () => { + expect(() => new RequestDeduplicator({ retentionMs: -1 })).toThrow('retentionMs must be a non-negative integer'); + }); + + it('should throw for non-integer retentionMs', () => { + expect(() => new RequestDeduplicator({ retentionMs: 1.5 })).toThrow('retentionMs must be a non-negative integer'); + }); + }); + + describe('exactly-once execution', () => { + it('should execute producer exactly once for concurrent calls with same key', async () => { + const deduplicator = new RequestDeduplicator(); + const producer = vi.fn().mockResolvedValue('result'); + + const [result1, result2] = await Promise.all([ + deduplicator.execute('key1', producer), + deduplicator.execute('key1', producer), + ]); + + expect(result1).toBe('result'); + expect(result2).toBe('result'); + expect(producer).toHaveBeenCalledTimes(1); + }); + + it('should execute producer exactly once for three concurrent calls', async () => { + const deduplicator = new RequestDeduplicator(); + const producer = vi.fn().mockResolvedValue('result'); + + const results = await Promise.all([ + deduplicator.execute('key1', producer), + deduplicator.execute('key1', producer), + deduplicator.execute('key1', producer), + ]); + + expect(results).toEqual(['result', 'result', 'result']); + expect(producer).toHaveBeenCalledTimes(1); + }); + + it('should execute producer for different keys independently', async () => { + const deduplicator = new RequestDeduplicator(); + const producer1 = vi.fn().mockResolvedValue('result1'); + const producer2 = vi.fn().mockResolvedValue('result2'); + + const [result1, result2] = await Promise.all([ + deduplicator.execute('key1', producer1), + deduplicator.execute('key2', producer2), + ]); + + expect(result1).toBe('result1'); + expect(result2).toBe('result2'); + expect(producer1).toHaveBeenCalledTimes(1); + expect(producer2).toHaveBeenCalledTimes(1); + }); + + it('should allow retry after completion with same key', async () => { + const deduplicator = new RequestDeduplicator(); + const producer = vi.fn().mockResolvedValue('result'); + + const result1 = await deduplicator.execute('key1', producer); + expect(result1).toBe('result'); + expect(producer).toHaveBeenCalledTimes(1); + + const result2 = await deduplicator.execute('key1', producer); + expect(result2).toBe('result'); + expect(producer).toHaveBeenCalledTimes(2); // Executed again + }); + }); + + describe('success handling', () => { + it('should return successful result to all callers', async () => { + const deduplicator = new RequestDeduplicator(); + const producer = vi.fn().mockResolvedValue('success'); + + const [result1, result2] = await Promise.all([ + deduplicator.execute('key1', producer), + deduplicator.execute('key1', producer), + ]); + + expect(result1).toBe('success'); + expect(result2).toBe('success'); + }); + + it('should remove entry after success by default', async () => { + const deduplicator = new RequestDeduplicator(); + const producer = vi.fn().mockResolvedValue('result'); + + await deduplicator.execute('key1', producer); + expect(deduplicator.getInFlightCount()).toBe(0); + + // Should execute again on next call + await deduplicator.execute('key1', producer); + expect(producer).toHaveBeenCalledTimes(2); + }); + + it('should retain result for configured retention period', async () => { + const deduplicator = new RequestDeduplicator({ retentionMs: 100 }); + const producer = vi.fn().mockResolvedValue('result'); + + const result1 = await deduplicator.execute('key1', producer); + expect(result1).toBe('result'); + expect(producer).toHaveBeenCalledTimes(1); + + // Should return retained result without executing producer + const result2 = await deduplicator.execute('key1', producer); + expect(result2).toBe('result'); + expect(producer).toHaveBeenCalledTimes(1); + }); + + it('should expire retained result after retention period', async () => { + const deduplicator = new RequestDeduplicator({ retentionMs: 10 }); + const producer = vi.fn().mockResolvedValue('result'); + + await deduplicator.execute('key1', producer); + expect(producer).toHaveBeenCalledTimes(1); + + // Wait for retention to expire + await new Promise((resolve) => setTimeout(resolve, 15)); + + // Should execute producer again + await deduplicator.execute('key1', producer); + expect(producer).toHaveBeenCalledTimes(2); + }); + }); + + describe('failure handling', () => { + it('should propagate failure to all callers', async () => { + const deduplicator = new RequestDeduplicator(); + const error = new Error('producer failed'); + const producer = vi.fn().mockRejectedValue(error); + + await expect( + Promise.all([ + deduplicator.execute('key1', producer), + deduplicator.execute('key1', producer), + ]) + ).rejects.toThrow('producer failed'); + + expect(producer).toHaveBeenCalledTimes(1); + }); + + it('should remove entry after failure', async () => { + const deduplicator = new RequestDeduplicator(); + const producer = vi.fn().mockRejectedValue(new Error('failed')); + + await expect(deduplicator.execute('key1', producer)).rejects.toThrow('failed'); + expect(deduplicator.getInFlightCount()).toBe(0); + + // Should allow retry after failure + producer.mockResolvedValue('success'); + const result = await deduplicator.execute('key1', producer); + expect(result).toBe('success'); + expect(producer).toHaveBeenCalledTimes(2); + }); + + it('should not retain failed results', async () => { + const deduplicator = new RequestDeduplicator({ retentionMs: 100 }); + const producer = vi.fn().mockRejectedValue(new Error('failed')); + + await expect(deduplicator.execute('key1', producer)).rejects.toThrow('failed'); + expect(deduplicator.getRetainedCount()).toBe(0); + }); + }); + + describe('capacity limits', () => { + it('should enforce maxInFlight capacity', async () => { + const deduplicator = new RequestDeduplicator({ maxInFlight: 2 }); + const producer1 = vi.fn().mockImplementation(() => new Promise((resolve) => setTimeout(() => resolve('result1'), 100))); + const producer2 = vi.fn().mockImplementation(() => new Promise((resolve) => setTimeout(() => resolve('result2'), 100))); + const producer3 = vi.fn().mockResolvedValue('result3'); + + // Start two in-flight requests + const promise1 = deduplicator.execute('key1', producer1); + const promise2 = deduplicator.execute('key2', producer2); + + expect(deduplicator.getInFlightCount()).toBe(2); + + // Third request should exceed capacity + await expect(deduplicator.execute('key3', producer3)).rejects.toThrow(CapacityExceededError); + + // Clean up + await Promise.all([promise1, promise2]); + }); + + it('should allow new request after completion', async () => { + const deduplicator = new RequestDeduplicator({ maxInFlight: 1 }); + const producer = vi.fn().mockResolvedValue('result'); + + await deduplicator.execute('key1', producer); + expect(deduplicator.getInFlightCount()).toBe(0); + + // Should allow new request after completion + await deduplicator.execute('key2', producer); + expect(producer).toHaveBeenCalledTimes(2); + }); + + it('should allow new request after failure', async () => { + const deduplicator = new RequestDeduplicator({ maxInFlight: 1 }); + const producer = vi.fn().mockRejectedValue(new Error('failed')); + + await expect(deduplicator.execute('key1', producer)).rejects.toThrow('failed'); + expect(deduplicator.getInFlightCount()).toBe(0); + + // Should allow new request after failure + producer.mockResolvedValue('success'); + await deduplicator.execute('key2', producer); + expect(producer).toHaveBeenCalledTimes(2); + }); + }); + + describe('diagnostics', () => { + it('should report in-flight count accurately', async () => { + const deduplicator = new RequestDeduplicator(); + const producer = vi.fn().mockImplementation(() => new Promise((resolve) => setTimeout(() => resolve('result'), 100))); + + expect(deduplicator.getInFlightCount()).toBe(0); + + const promise1 = deduplicator.execute('key1', producer); + expect(deduplicator.getInFlightCount()).toBe(1); + + const promise2 = deduplicator.execute('key2', producer); + expect(deduplicator.getInFlightCount()).toBe(2); + + await Promise.all([promise1, promise2]); + expect(deduplicator.getInFlightCount()).toBe(0); + }); + + it('should report retained count accurately', async () => { + const deduplicator = new RequestDeduplicator({ retentionMs: 1000 }); + const producer = vi.fn().mockResolvedValue('result'); + + expect(deduplicator.getRetainedCount()).toBe(0); + + await deduplicator.execute('key1', producer); + expect(deduplicator.getRetainedCount()).toBe(1); + + await deduplicator.execute('key2', producer); + expect(deduplicator.getRetainedCount()).toBe(2); + }); + + it('should clean up expired entries when reporting retained count', async () => { + const deduplicator = new RequestDeduplicator({ retentionMs: 10 }); + const producer = vi.fn().mockResolvedValue('result'); + + await deduplicator.execute('key1', producer); + await deduplicator.execute('key2', producer); + expect(deduplicator.getRetainedCount()).toBe(2); + + // Wait for expiration + await new Promise((resolve) => setTimeout(resolve, 15)); + + // getRetainedCount should clean up expired entries + expect(deduplicator.getRetainedCount()).toBe(0); + }); + }); + + describe('clear', () => { + it('should clear all in-flight requests', async () => { + const deduplicator = new RequestDeduplicator(); + const producer = vi.fn().mockImplementation(() => new Promise((resolve) => setTimeout(() => resolve('result'), 100))); + + const promise1 = deduplicator.execute('key1', producer); + const promise2 = deduplicator.execute('key2', producer); + + expect(deduplicator.getInFlightCount()).toBe(2); + + deduplicator.clear(); + + expect(deduplicator.getInFlightCount()).toBe(0); + + // Promises should still resolve + await Promise.all([promise1, promise2]); + }); + + it('should clear all retained results', async () => { + const deduplicator = new RequestDeduplicator({ retentionMs: 1000 }); + const producer = vi.fn().mockResolvedValue('result'); + + await deduplicator.execute('key1', producer); + await deduplicator.execute('key2', producer); + + expect(deduplicator.getRetainedCount()).toBe(2); + + deduplicator.clear(); + + expect(deduplicator.getRetainedCount()).toBe(0); + }); + }); + + describe('controlled promise execution', () => { + it('should prove exactly-once execution with controlled promises', async () => { + const deduplicator = new RequestDeduplicator(); + let executionCount = 0; + + const createControlledProducer = (): AsyncProducer => { + return () => { + executionCount++; + return new Promise((resolve) => { + setTimeout(() => resolve(`execution-${executionCount}`), 10); + }); + }; + }; + + const producer = createControlledProducer(); + + const [result1, result2, result3] = await Promise.all([ + deduplicator.execute('key1', producer), + deduplicator.execute('key1', producer), + deduplicator.execute('key1', producer), + ]); + + expect(executionCount).toBe(1); + expect(result1).toBe('execution-1'); + expect(result2).toBe('execution-1'); + expect(result3).toBe('execution-1'); + }); + + it('should handle sequential calls with controlled promises', async () => { + const deduplicator = new RequestDeduplicator(); + let executionCount = 0; + + const createControlledProducer = (): AsyncProducer => { + return () => { + executionCount++; + return Promise.resolve(`execution-${executionCount}`); + }; + }; + + const producer = createControlledProducer(); + + const result1 = await deduplicator.execute('key1', producer); + expect(result1).toBe('execution-1'); + expect(executionCount).toBe(1); + + const result2 = await deduplicator.execute('key1', producer); + expect(result2).toBe('execution-2'); + expect(executionCount).toBe(2); + }); + }); + + describe('edge cases', () => { + it('should handle empty string key', async () => { + const deduplicator = new RequestDeduplicator(); + const producer = vi.fn().mockResolvedValue('result'); + + const result = await deduplicator.execute('', producer); + expect(result).toBe('result'); + expect(producer).toHaveBeenCalledTimes(1); + }); + + it('should handle special characters in key', async () => { + const deduplicator = new RequestDeduplicator(); + const producer = vi.fn().mockResolvedValue('result'); + + const result = await deduplicator.execute('key:with/special-chars_123', producer); + expect(result).toBe('result'); + expect(producer).toHaveBeenCalledTimes(1); + }); + + it('should handle producer that returns undefined', async () => { + const deduplicator = new RequestDeduplicator(); + const producer = vi.fn().mockResolvedValue(undefined); + + const result = await deduplicator.execute('key1', producer); + expect(result).toBeUndefined(); + expect(producer).toHaveBeenCalledTimes(1); + }); + + it('should handle producer that returns null', async () => { + const deduplicator = new RequestDeduplicator(); + const producer = vi.fn().mockResolvedValue(null); + + const result = await deduplicator.execute('key1', producer); + expect(result).toBeNull(); + expect(producer).toHaveBeenCalledTimes(1); + }); + + it('should handle producer that returns complex objects', async () => { + const deduplicator = new RequestDeduplicator<{ id: number; data: string }>(); + const producer = vi.fn().mockResolvedValue({ id: 123, data: 'test' }); + + const result = await deduplicator.execute('key1', producer); + expect(result).toEqual({ id: 123, data: 'test' }); + expect(producer).toHaveBeenCalledTimes(1); + }); + }); +});