Fix rate limiter delaying tasks in a fresh window when intervalCap > 1

Fixes #251

Author: sindresorhus

source/index.ts

@@ -27,12 +27,15 @@ export default class PQueue<QueueType extends Queue<RunFunction, EnqueueOptionsT
 
 	readonly #interval: number;
 
+	/*
+	Timestamp when the current interval window ends. Kept accurate at every window transition: `#initializeIntervalIfNeeded` sets it when a window starts and `#onInterval` advances it as the recurring timer rolls into each new window. This is the source of truth for whether a task added after the queue goes idle belongs to a fresh window, so it must not go stale.
+	*/
 	#intervalEnd = 0;
 
-	#lastExecutionTime = 0;
-
+	// Recurring timer that drives windows while the queue is actively processing. Absent (`undefined`) once the queue goes idle.
 	#intervalId?: NodeJS.Timeout;
 
+	// One-shot timer used while idle to resume at the next window boundary, since `#intervalId` is cleared when idle.
 	#timeoutId?: NodeJS.Timeout;
 
 	readonly #strict: boolean;
@@ -234,26 +237,14 @@ export default class PQueue<QueueType extends Queue<RunFunction, EnqueueOptionsT
 			return false;
 		}
 
-		// Fixed window mode (original logic)
+		// Fixed window mode. While the recurring timer runs it already governs the windows, so pausing is only decided here when the queue is idle (the timer is absent) and we rely on `#intervalEnd`.
 		if (this.#intervalId === undefined) {
 			const delay = this.#intervalEnd - now;
 			if (delay < 0) {
-				// If the interval has expired while idle, check if we should enforce the interval
-				// from the last task execution. This ensures proper spacing between tasks even
-				// when the queue becomes empty and then new tasks are added.
-				if (this.#lastExecutionTime > 0) {
-					const timeSinceLastExecution = now - this.#lastExecutionTime;
-					if (timeSinceLastExecution < this.#interval) {
-						// Not enough time has passed since the last task execution
-						this.#createIntervalTimeout(this.#interval - timeSinceLastExecution);
-						return true;
-					}
-				}
-
-				// Enough time has passed or no previous execution, allow execution
-				this.#intervalCount = (this.#carryoverIntervalCount) ? this.#pending : 0;
+				// The interval expired while idle, so reset the count for the new window.
+				this.#intervalCount = this.#carryoverIntervalCount ? this.#pending : 0;
 			} else {
-				// Act as the interval is pending
+				// Still inside the previous window; wait out the remaining time before starting the next task.
 				this.#createIntervalTimeout(delay);
 				return true;
 			}
@@ -323,13 +314,13 @@ export default class PQueue<QueueType extends Queue<RunFunction, EnqueueOptionsT
 					this.#scheduleRateLimitUpdate();
 				}
 
-				this.emit('active');
-				job();
-
 				if (canInitializeInterval) {
 					this.#initializeIntervalIfNeeded();
 				}
 
+				this.emit('active');
+				job();
+
 				taskStarted = true;
 			}
 		}
@@ -360,8 +351,14 @@ export default class PQueue<QueueType extends Queue<RunFunction, EnqueueOptionsT
 	#onInterval(): void {
 		// Non-strict mode uses interval timers and intervalCount
 		if (!this.#strict) {
-			if (this.#intervalCount === 0 && this.#pending === 0 && this.#intervalId) {
-				this.#clearIntervalTimer();
+			// Only touch the recurring interval timer here. When resumed via a timeout instead, `#intervalId` is undefined and `#initializeIntervalIfNeeded` sets `#intervalEnd` right after.
+			if (this.#intervalId !== undefined) {
+				if (this.#intervalCount === 0 && this.#pending === 0) {
+					this.#clearIntervalTimer();
+				} else {
+					// The recurring timer fired, starting a new window. Keep the boundary accurate so tasks added after the queue later goes idle are scheduled against the correct window.
+					this.#intervalEnd = Date.now() + this.#interval;
+				}
 			}
 
 			this.#intervalCount = this.#carryoverIntervalCount ? this.#pending : 0;
@@ -489,8 +486,6 @@ export default class PQueue<QueueType extends Queue<RunFunction, EnqueueOptionsT
 						throw error;
 					}
 
-					this.#lastExecutionTime = Date.now();
-
 					let operation = function_({signal: options.signal});
 
 					if (options.timeout !== undefined) {

test/advanced.ts

@@ -999,6 +999,92 @@ test('interval should be maintained when using await between adds (issue #182)',
 	}
 });
 
+test('task added in a fresh window runs immediately after queue went idle mid-window (issue #251)', async () => {
+	const queue = new PQueue({intervalCap: 2, interval: 200});
+
+	// Window [0, 200): consume slot 1 at t≈0.
+	queue.add(() => undefined);
+
+	// Consume slot 2 later in the window at t≈120.
+	await delay(120);
+	queue.add(() => undefined);
+	await queue.onIdle();
+
+	// Now past the window boundary at t≈220. Both slots of the fresh window should be available immediately.
+	await delay(100);
+	const start = Date.now();
+	queue.add(() => undefined);
+	await queue.add(() => undefined);
+	const elapsed = Date.now() - start;
+
+	assert.ok(elapsed < 50, `Both tasks should start immediately in the new window, but waited ${elapsed}ms`);
+});
+
+test('fresh window after idle still enforces the cap for the following task (issue #251)', async () => {
+	const queue = new PQueue({intervalCap: 2, interval: 200});
+
+	// Consume both slots of the first window, then let the queue drain.
+	queue.add(() => undefined);
+	await delay(120);
+	queue.add(() => undefined);
+	await queue.onIdle();
+
+	// Past the boundary: the fresh window grants exactly two slots, so a third task must wait for the next window.
+	await delay(100);
+	const timestamps: number[] = [];
+	const start = Date.now();
+	queue.add(() => {
+		timestamps.push(Date.now() - start);
+	});
+	queue.add(() => {
+		timestamps.push(Date.now() - start);
+	});
+	await queue.add(() => {
+		timestamps.push(Date.now() - start);
+	});
+
+	assert.ok(timestamps[0] < 50, `First task should be immediate, waited ${timestamps[0]}ms`);
+	assert.ok(timestamps[1] < 50, `Second task should be immediate, waited ${timestamps[1]}ms`);
+	assert.ok(timestamps[2] >= 170, `Third task should wait for the next window, but ran after ${timestamps[2]}ms`);
+});
+
+test('carryoverIntervalCount does not break the fresh window after idle mid-window (issue #251)', async () => {
+	const queue = new PQueue({intervalCap: 2, interval: 200, carryoverIntervalCount: true});
+
+	queue.add(() => undefined);
+	await delay(120);
+	queue.add(() => undefined);
+	await queue.onIdle();
+
+	// With no pending tasks, carrying the count over resets it to zero, so the fresh window is fully available.
+	await delay(100);
+	const start = Date.now();
+	queue.add(() => undefined);
+	await queue.add(() => undefined);
+	const elapsed = Date.now() - start;
+
+	assert.ok(elapsed < 50, `Both tasks should start immediately in the new window, but waited ${elapsed}ms`);
+});
+
+test('task added after going idle but before the boundary still waits for the current window (issue #251)', async () => {
+	const queue = new PQueue({intervalCap: 2, interval: 200});
+	const windowStart = Date.now();
+
+	// Fill both slots spread across the window, after which the queue goes idle.
+	queue.add(() => undefined);
+	await delay(120);
+	queue.add(() => undefined);
+	await queue.onIdle();
+
+	// Added while still inside the window [0, 200): the cap is reached, so it must wait for the boundary, not run immediately. This guards against the reset firing before the window has actually expired.
+	const start = Date.now();
+	await queue.add(() => undefined);
+	const waited = Date.now() - start;
+
+	assert.ok(waited >= 40, `Task should wait for the current window to end, but only waited ${waited}ms`);
+	assert.ok(Date.now() - windowStart >= 190, 'Task should run at the window boundary');
+});
+
 test('interval maintained when queue becomes empty multiple times', async () => {
 	const queue = new PQueue({
 		intervalCap: 1,

test/rate-limit.ts

@@ -43,6 +43,56 @@ test('rate-limit rapid pause/start cycles', async () => {
 	assert.equal(results.length, 4);
 });
 
+test('rate-limit preserves the interval count for re-entrant task additions', async () => {
+	const queue = new PQueue({
+		interval: 100,
+		intervalCap: 2,
+	});
+
+	const starts: number[] = [];
+	const addTask = (index: number): void => {
+		queue.add(() => {
+			starts.push(index);
+
+			if (index < 4) {
+				addTask(index + 1);
+			}
+		});
+	};
+
+	addTask(0);
+	assert.deepEqual(starts, [0, 1]);
+
+	await queue.onIdle();
+	assert.deepEqual(starts, [0, 1, 2, 3, 4]);
+});
+
+test('rate-limit preserves the interval count for re-entrant active listeners', async () => {
+	const queue = new PQueue({
+		interval: 100,
+		intervalCap: 2,
+	});
+
+	const starts: number[] = [];
+	let nextTask = 0;
+	queue.on('active', () => {
+		if (nextTask < 5) {
+			const taskIndex = nextTask++;
+			queue.add(() => {
+				starts.push(taskIndex);
+			});
+		}
+	});
+
+	queue.add(() => {
+		starts.push(-1);
+	});
+	assert.equal(starts.length, 2);
+
+	await queue.onIdle();
+	assert.equal(starts.length, 6);
+});
+
 test('rate-limit edge case with zero-interval', async () => {
 	// Zero interval should effectively disable rate limiting
 	const queue = new PQueue({

test/strict.ts

@@ -507,6 +507,27 @@ test('strict mode after queue becomes idle', async () => {
 	assert.ok(elapsed < 50, `Second batch should execute immediately after idle, took ${elapsed}ms`);
 });
 
+test('strict mode frees a slot as soon as the oldest tick ages out, even with spread ticks', async () => {
+	const queue = new PQueue({
+		interval: 100,
+		intervalCap: 2,
+		strict: true,
+	});
+
+	// Consume both slots spread across the window: tick at ≈0 and tick at ≈60.
+	await queue.add(async () => undefined);
+	await delay(60);
+	await queue.add(async () => undefined);
+
+	// At ≈120ms the first tick (≈0) has aged out but the second (≈60) has not, so exactly one slot is free and the next task should run immediately.
+	await delay(60);
+	const start = Date.now();
+	await queue.add(async () => undefined);
+	const elapsed = Date.now() - start;
+
+	assert.ok(elapsed < 40, `Task should run as soon as the oldest tick ages out, but waited ${elapsed}ms`);
+});
+
 test('strict mode prevents boundary bursts', async () => {
 	// This test verifies that strict mode prevents the boundary burst problem
 	// that occurs with fixed window mode