Fix interval timing when queue becomes empty between task additions

Fixes #163

Author: sindresorhus

readme.md

@@ -46,7 +46,7 @@ npm install p-queue
 
 ## Usage
 
-Here we run only one promise at the time. For example, set `concurrency` to 4 to run four promises at the same time.
+Here we run only one promise at a time. For example, set `concurrency` to 4 to run four promises at the same time.
 
 ```js
 import PQueue from 'p-queue';

source/index.ts

@@ -26,6 +26,8 @@ export default class PQueue<QueueType extends Queue<RunFunction, EnqueueOptionsT
 
 	#intervalEnd = 0;
 
+	#lastExecutionTime = 0;
+
 	#intervalId?: NodeJS.Timeout;
 
 	#timeoutId?: NodeJS.Timeout;
@@ -114,40 +116,64 @@ export default class PQueue<QueueType extends Queue<RunFunction, EnqueueOptionsT
 		if (this.#intervalId === undefined) {
 			const delay = this.#intervalEnd - now;
 			if (delay < 0) {
-				// Act as the interval was done
-				// We don't need to resume it here because it will be resumed on line 160
+				// 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.#carryoverConcurrencyCount) ? this.#pending : 0;
 			} else {
 				// Act as the interval is pending
-				if (this.#timeoutId === undefined) {
-					this.#timeoutId = setTimeout(
-						() => {
-							this.#onResumeInterval();
-						},
-						delay,
-					);
-				}
-
+				this.#createIntervalTimeout(delay);
 				return true;
 			}
 		}
 
 		return false;
 	}
 
+	#createIntervalTimeout(delay: number): void {
+		if (this.#timeoutId !== undefined) {
+			return;
+		}
+
+		this.#timeoutId = setTimeout(() => {
+			this.#onResumeInterval();
+		}, delay);
+	}
+
+	#clearIntervalTimer(): void {
+		if (this.#intervalId) {
+			clearInterval(this.#intervalId);
+			this.#intervalId = undefined;
+		}
+	}
+
+	#clearTimeoutTimer(): void {
+		if (this.#timeoutId) {
+			clearTimeout(this.#timeoutId);
+			this.#timeoutId = undefined;
+		}
+	}
+
 	#tryToStartAnother(): boolean {
 		if (this.#queue.size === 0) {
 			// We can clear the interval ("pause")
 			// Because we can redo it later ("resume")
-			if (this.#intervalId) {
-				clearInterval(this.#intervalId);
-			}
-
-			this.#intervalId = undefined;
-
+			this.#clearIntervalTimer();
 			this.emit('empty');
 
 			if (this.#pending === 0) {
+				// Clear timeout as well when completely idle
+				this.#clearTimeoutTimer();
 				this.emit('idle');
 			}
 
@@ -160,6 +186,7 @@ export default class PQueue<QueueType extends Queue<RunFunction, EnqueueOptionsT
 				const job = this.#queue.dequeue()!;
 
 				this.emit('active');
+				this.#lastExecutionTime = Date.now();
 				job();
 
 				if (canInitializeInterval) {
@@ -190,8 +217,7 @@ export default class PQueue<QueueType extends Queue<RunFunction, EnqueueOptionsT
 
 	#onInterval(): void {
 		if (this.#intervalCount === 0 && this.#pending === 0 && this.#intervalId) {
-			clearInterval(this.#intervalId);
-			this.#intervalId = undefined;
+			this.#clearIntervalTimer();
 		}
 
 		this.#intervalCount = this.#carryoverConcurrencyCount ? this.#pending : 0;

test/test.ts

@@ -1313,6 +1313,157 @@ test('.setPriority() - execute a promise before planned', async t => {
 	t.deepEqual(result, ['🐌', '🐢', '🦆']);
 });
 
+test('interval should be maintained when using await between adds (issue #182)', async t => {
+	const queue = new PQueue({
+		intervalCap: 1,
+		interval: 100,
+	});
+
+	const timestamps: number[] = [];
+
+	// Add first 3 tasks without await
+	queue.add(() => {
+		timestamps.push(Date.now());
+		return 'task1';
+	});
+	queue.add(() => {
+		timestamps.push(Date.now());
+		return 'task2';
+	});
+	queue.add(() => {
+		timestamps.push(Date.now());
+		return 'task3';
+	});
+
+	// Add task 4 with await
+	await queue.add(() => {
+		timestamps.push(Date.now());
+		return 'task4';
+	});
+
+	// Add task 5 with await - this should still respect interval
+	await queue.add(() => {
+		timestamps.push(Date.now());
+		return 'task5';
+	});
+
+	// Add task 6 with await
+	await queue.add(() => {
+		timestamps.push(Date.now());
+		return 'task6';
+	});
+
+	// Check intervals between tasks
+	for (let i = 1; i < timestamps.length; i++) {
+		const interval = timestamps[i] - timestamps[i - 1];
+		// Allow 10ms tolerance for timing
+		t.true(interval >= 90, `Interval between task ${i} and ${i + 1} was ${interval}ms, expected >= 90ms`);
+	}
+});
+
+test('interval maintained when queue becomes empty multiple times', async t => {
+	const queue = new PQueue({
+		intervalCap: 1,
+		interval: 100,
+	});
+
+	const timestamps: number[] = [];
+
+	// First batch
+	await queue.add(() => {
+		timestamps.push(Date.now());
+		return 'task1';
+	});
+	await queue.add(() => {
+		timestamps.push(Date.now());
+		return 'task2';
+	});
+
+	// Queue is empty, wait a bit
+	await delay(50);
+
+	// Second batch - should still respect interval from task 2
+	await queue.add(() => {
+		timestamps.push(Date.now());
+		return 'task3';
+	});
+	await queue.add(() => {
+		timestamps.push(Date.now());
+		return 'task4';
+	});
+
+	// Check all intervals
+	for (let i = 1; i < timestamps.length; i++) {
+		const interval = timestamps[i] - timestamps[i - 1];
+		t.true(interval >= 90, `Interval between task ${i} and ${i + 1} was ${interval}ms, expected >= 90ms`);
+	}
+});
+
+test('interval reset after long idle period', async t => {
+	const queue = new PQueue({
+		intervalCap: 1,
+		interval: 100,
+	});
+
+	const timestamps: number[] = [];
+
+	// Run first task
+	await queue.add(() => {
+		timestamps.push(Date.now());
+		return 'task1';
+	});
+
+	// Wait much longer than interval
+	await delay(250);
+
+	// This task should run immediately since enough time has passed
+	await queue.add(() => {
+		timestamps.push(Date.now());
+		return 'task2';
+	});
+
+	// But this one should wait for interval
+	await queue.add(() => {
+		timestamps.push(Date.now());
+		return 'task3';
+	});
+
+	const interval1to2 = timestamps[1] - timestamps[0];
+	const interval2to3 = timestamps[2] - timestamps[1];
+
+	t.true(interval1to2 >= 240, `Task 2 ran after ${interval1to2}ms, expected >= 240ms`);
+	t.true(interval2to3 >= 90, `Task 3 should respect interval: ${interval2to3}ms`);
+});
+
+test('interval with carryoverConcurrencyCount after queue empty', async t => {
+	const queue = new PQueue({
+		intervalCap: 1,
+		interval: 100,
+		carryoverConcurrencyCount: true,
+	});
+
+	const timestamps: number[] = [];
+
+	// Run first task
+	await queue.add(() => {
+		timestamps.push(Date.now());
+		return 'task1';
+	});
+
+	// Queue becomes empty
+	t.is(queue.size, 0);
+	t.is(queue.pending, 0);
+
+	// Add new task - should respect interval
+	await queue.add(() => {
+		timestamps.push(Date.now());
+		return 'task2';
+	});
+
+	const interval = timestamps[1] - timestamps[0];
+	t.true(interval >= 90, `Interval was ${interval}ms, expected >= 90ms`);
+});
+
 test('.setPriority() - execute a promise after planned', async t => {
 	const result: string[] = [];
 	const queue = new PQueue({concurrency: 1});
@@ -1453,3 +1604,24 @@ test('.setPriority() - execute a promise before planned - concurrency 3 and unsp
 	await queue.onIdle();
 	t.deepEqual(result, ['🐌', '🦆', '🐢', '🦀', '⚡️']);
 });
+
+test('process exits cleanly after interval tasks complete', async t => {
+	const queue = new PQueue({
+		concurrency: 100,
+		intervalCap: 500,
+		interval: 60 * 1000,
+	});
+
+	// Execute tasks that complete quickly with long interval
+	const tasks = [];
+	for (let i = 0; i < 4; i++) {
+		tasks.push(queue.add(() => `result-${i}`));
+	}
+
+	await Promise.all(tasks);
+	await queue.onIdle();
+
+	// Test that no timers are hanging by checking process can exit naturally
+	// This ensures both intervalId and timeoutId are cleared when idle
+	t.pass();
+});