Add rate-limit visibility feature

Author: sindresorhus

readme.md

@@ -241,6 +241,36 @@ await queue.onPendingZero();
 // All running tasks have finished, though the queue may still have items
 ```
 
+#### .onRateLimit()
+
+Returns a promise that settles when the queue becomes rate-limited due to `intervalCap`. If the queue is already rate-limited, the promise resolves immediately.
+
+Useful for implementing backpressure to prevent memory issues when producers are faster than consumers.
+
+```js
+const queue = new PQueue({intervalCap: 5, interval: 1000});
+
+// Add many tasks
+for (let index = 0; index < 10; index++) {
+	queue.add(() => someTask());
+}
+
+await queue.onRateLimit();
+console.log('Queue is now rate-limited - time for maintenance tasks');
+```
+
+#### .onRateLimitCleared()
+
+Returns a promise that settles when the queue is no longer rate-limited. If the queue is not currently rate-limited, the promise resolves immediately.
+
+```js
+const queue = new PQueue({intervalCap: 5, interval: 1000});
+
+// Wait for rate limiting to be cleared
+await queue.onRateLimitCleared();
+console.log('Rate limit cleared - can add more tasks');
+```
+
 #### .onSizeLessThan(limit)
 
 Returns a promise that settles when the queue size is less than the given limit: `queue.size < limit`.
@@ -329,6 +359,10 @@ Number of running items (no longer in the queue).
 
 Whether the queue is currently paused.
 
+#### .isRateLimited
+
+Whether the queue is currently rate-limited due to `intervalCap`. Returns `true` when the number of tasks executed in the current interval has reached the `intervalCap` and there are still tasks waiting to be processed.
+
 ## Events
 
 #### active
@@ -465,6 +499,56 @@ await queue.add(() => delay(600));
 //=> 'Task is completed.  Size: 0  Pending: 0'
 ```
 
+#### rateLimit
+
+Emitted when the queue becomes rate-limited due to `intervalCap`. This happens when the maximum number of tasks allowed per interval has been reached.
+
+Useful for implementing backpressure to prevent memory issues when producers are faster than consumers.
+
+```js
+import delay from 'delay';
+import PQueue from 'p-queue';
+
+const queue = new PQueue({
+	intervalCap: 2,
+	interval: 1000
+});
+
+queue.on('rateLimit', () => {
+	console.log('Queue is rate-limited - processing backlog or maintenance tasks');
+});
+
+// Add 3 tasks - third one triggers rate limiting
+queue.add(() => delay(100));
+queue.add(() => delay(100));
+queue.add(() => delay(100));
+```
+
+#### rateLimitCleared
+
+Emitted when the queue is no longer rate-limited—either because the interval reset and new tasks can start, or because the backlog was drained.
+
+```js
+import delay from 'delay';
+import PQueue from 'p-queue';
+
+const queue = new PQueue({
+	intervalCap: 1,
+	interval: 1000
+});
+
+queue.on('rateLimit', () => {
+	console.log('Rate limited - waiting for interval to reset');
+});
+
+queue.on('rateLimitCleared', () => {
+	console.log('Rate limit cleared - can process more tasks');
+});
+
+queue.add(() => delay(100));
+queue.add(() => delay(100)); // This triggers rate limiting
+```
+
 ## Advanced example
 
 A more advanced example to help you understand the flow.

source/index.ts

@@ -8,7 +8,7 @@ type Task<TaskResultType> =
 	| ((options: TaskOptions) => PromiseLike<TaskResultType>)
 	| ((options: TaskOptions) => TaskResultType);
 
-type EventName = 'active' | 'idle' | 'empty' | 'add' | 'next' | 'completed' | 'error' | 'pendingZero';
+type EventName = 'active' | 'idle' | 'empty' | 'add' | 'next' | 'completed' | 'error' | 'pendingZero' | 'rateLimit' | 'rateLimitCleared';
 
 /**
 Promise queue with concurrency control.
@@ -22,6 +22,9 @@ export default class PQueue<QueueType extends Queue<RunFunction, EnqueueOptionsT
 
 	readonly #intervalCap: number;
 
+	#rateLimitedInInterval = false;
+	#rateLimitFlushScheduled = false;
+
 	readonly #interval: number;
 
 	#intervalEnd = 0;
@@ -84,8 +87,15 @@ export default class PQueue<QueueType extends Queue<RunFunction, EnqueueOptionsT
 		this.#queue = new options.queueClass!();
 		this.#queueClass = options.queueClass!;
 		this.concurrency = options.concurrency!;
+
+		if (options.timeout !== undefined && !(Number.isFinite(options.timeout) && options.timeout > 0)) {
+			throw new TypeError(`Expected \`timeout\` to be a positive finite number, got \`${options.timeout}\` (${typeof options.timeout})`);
+		}
+
 		this.timeout = options.timeout;
 		this.#isPaused = options.autoStart === false;
+
+		this.#setupRateLimitTracking();
 	}
 
 	get #doesIntervalAllowAnother(): boolean {
@@ -108,7 +118,7 @@ export default class PQueue<QueueType extends Queue<RunFunction, EnqueueOptionsT
 	}
 
 	#onResumeInterval(): void {
-		this.#onInterval();
+		this.#onInterval(); // Already schedules update
 		this.#initializeIntervalIfNeeded();
 		this.#timeoutId = undefined;
 	}
@@ -183,6 +193,8 @@ export default class PQueue<QueueType extends Queue<RunFunction, EnqueueOptionsT
 			return false;
 		}
 
+		let taskStarted = false;
+
 		if (!this.#isPaused) {
 			const canInitializeInterval = !this.#isIntervalPaused;
 			if (this.#doesIntervalAllowAnother && this.#doesConcurrentAllowAnother) {
@@ -191,6 +203,7 @@ export default class PQueue<QueueType extends Queue<RunFunction, EnqueueOptionsT
 				// Increment interval count immediately to prevent race conditions
 				if (!this.#isIntervalIgnored) {
 					this.#intervalCount++;
+					this.#scheduleRateLimitUpdate();
 				}
 
 				this.emit('active');
@@ -201,11 +214,11 @@ export default class PQueue<QueueType extends Queue<RunFunction, EnqueueOptionsT
 					this.#initializeIntervalIfNeeded();
 				}
 
-				return true;
+				taskStarted = true;
 			}
 		}
 
-		return false;
+		return taskStarted;
 	}
 
 	#initializeIntervalIfNeeded(): void {
@@ -229,7 +242,9 @@ export default class PQueue<QueueType extends Queue<RunFunction, EnqueueOptionsT
 		}
 
 		this.#intervalCount = this.#carryoverConcurrencyCount ? this.#pending : 0;
+
 		this.#processQueue();
+		this.#scheduleRateLimitUpdate();
 	}
 
 	/**
@@ -299,6 +314,10 @@ export default class PQueue<QueueType extends Queue<RunFunction, EnqueueOptionsT
 	Here, the promise function with `id: '🦀'` executes last.
 	*/
 	setPriority(id: string, priority: number) {
+		if (typeof priority !== 'number' || !Number.isFinite(priority)) {
+			throw new TypeError(`Expected \`priority\` to be a finite number, got \`${priority}\` (${typeof priority})`);
+		}
+
 		this.#queue.setPriority(id, priority);
 	}
 
@@ -405,6 +424,8 @@ export default class PQueue<QueueType extends Queue<RunFunction, EnqueueOptionsT
 	*/
 	clear(): void {
 		this.#queue = new this.#queueClass();
+		// Force synchronous update since clear() should have immediate effect
+		this.#updateRateLimitState();
 	}
 
 	/**
@@ -464,6 +485,28 @@ export default class PQueue<QueueType extends Queue<RunFunction, EnqueueOptionsT
 		await this.#onEvent('pendingZero');
 	}
 
+	/**
+	@returns A promise that settles when the queue becomes rate-limited due to intervalCap.
+	*/
+	async onRateLimit(): Promise<void> {
+		if (this.isRateLimited) {
+			return;
+		}
+
+		await this.#onEvent('rateLimit');
+	}
+
+	/**
+	@returns A promise that settles when the queue is no longer rate-limited.
+	*/
+	async onRateLimitCleared(): Promise<void> {
+		if (!this.isRateLimited) {
+			return;
+		}
+
+		await this.#onEvent('rateLimitCleared');
+	}
+
 	async #onEvent(event: EventName, filter?: () => boolean): Promise<void> {
 		return new Promise(resolve => {
 			const listener = () => {
@@ -509,6 +552,57 @@ export default class PQueue<QueueType extends Queue<RunFunction, EnqueueOptionsT
 	get isPaused(): boolean {
 		return this.#isPaused;
 	}
+
+	#setupRateLimitTracking(): void {
+		// Only schedule updates when rate limiting is enabled
+		if (this.#isIntervalIgnored) {
+			return;
+		}
+
+		// Wire up to lifecycle events that affect rate limit state
+		// Only 'add' and 'next' can actually change rate limit state
+		this.on('add', () => {
+			if (this.#queue.size > 0) {
+				this.#scheduleRateLimitUpdate();
+			}
+		});
+
+		this.on('next', () => {
+			this.#scheduleRateLimitUpdate();
+		});
+	}
+
+	#scheduleRateLimitUpdate(): void {
+		// Skip if rate limiting is not enabled or already scheduled
+		if (this.#isIntervalIgnored || this.#rateLimitFlushScheduled) {
+			return;
+		}
+
+		this.#rateLimitFlushScheduled = true;
+		queueMicrotask(() => {
+			this.#rateLimitFlushScheduled = false;
+			this.#updateRateLimitState();
+		});
+	}
+
+	#updateRateLimitState(): void {
+		const previous = this.#rateLimitedInInterval;
+		const shouldBeRateLimited = !this.#isIntervalIgnored
+			&& this.#intervalCount >= this.#intervalCap
+			&& this.#queue.size > 0;
+
+		if (shouldBeRateLimited !== previous) {
+			this.#rateLimitedInInterval = shouldBeRateLimited;
+			this.emit(shouldBeRateLimited ? 'rateLimit' : 'rateLimitCleared');
+		}
+	}
+
+	/**
+	Whether the queue is currently rate-limited due to intervalCap.
+	*/
+	get isRateLimited(): boolean {
+		return this.#rateLimitedInInterval;
+	}
 }
 
 export type {Queue} from './queue.js';