Fix concurrent limiter accounting and bounded waiting

.github/workflows/ci.yml

@@ -43,3 +43,6 @@ jobs:
 
       - name: Run tests
         run: yarn test
+
+      - name: Build distribution
+        run: yarn prepack

CHANGELOG.md

@@ -1,5 +1,19 @@
 # Changelog
 
+## Unreleased
+
+### Fixed
+- Reserve interval allowance before asynchronous calls yield; recheck it after waiting.
+- Debit parent and child buckets together, avoiding double charges under contention.
+- Serialize waiting requests per instance with one active timer, and reject cyclic parents.
+- Reject invalid token values and intervals; cap long timers and avoid NaN balances for very short intervals.
+- Correct the byte-throttling example and document interval, queue, zero-value, and floating-point behavior.
+
+### Compatibility
+- Invalid numeric inputs now throw `RangeError` (or reject the returned promise).
+- Async requests on one instance now run FIFO; synchronous requests can still consume capacity first.
+- Fractional millisecond clock precision is retained, so remaining balances may differ slightly from earlier versions.
+
 ## [3.0.0] - 2025-01-24
 ### Added
 - **Dual Module Support:** Distributed as both CommonJS and ES Module, with separate `package.json` files for each format (no need for custom transformers). This improves compatibility with modern bundlers and frameworks.

README.md

@@ -22,15 +22,15 @@ A simple example allowing 150 requests per hour:
 ```javascript
 import { RateLimiter } from "limiter";
 
-// Allow 150 requests per hour (the Twitter search limit). Also understands
+// Allow 150 requests per hour. Also understands
 // 'second', 'minute', 'day', or a number of milliseconds
 const limiter = new RateLimiter({ tokensPerInterval: 150, interval: "hour" });
 
 async function sendRequest() {
   // This call will throw if we request more than the maximum number of requests
   // that were set in the constructor
-  // remainingRequests tells us how many additional requests could be sent
-  // right this moment
+  // remainingRequests is the underlying bucket balance; the interval counter
+  // may impose a lower allowance.
   const remainingRequests = await limiter.removeTokens(1);
   callMyRequestSendingFunction(...);
 }
@@ -108,14 +108,14 @@ Using the token bucket directly to throttle at the byte level:
 ```javascript
 import { TokenBucket } from "limiter";
 
-const BURST_RATE = 1024 * 1024 * 150; // 150KB/sec burst rate
-const FILL_RATE = 1024 * 1024 * 50; // 50KB/sec sustained rate
+const BURST_CAPACITY = 1024 * 150; // 150 KiB maximum burst
+const FILL_RATE = 1024 * 50; // 50 KiB per second sustained rate
 
 // We could also pass a parent token bucket in to create a hierarchical token
 // bucket
 // bucketSize, tokensPerInterval, interval
 const bucket = new TokenBucket({
-  bucketSize: BURST_RATE,
+  bucketSize: BURST_CAPACITY,
   tokensPerInterval: FILL_RATE,
   interval: "second"
 });
@@ -128,11 +128,41 @@ async function handleData(myData) {
 
 ## Additional Notes
 
-Both the token bucket and rate limiter should be used with a message queue or 
-some way of preventing multiple simultaneous calls to removeTokens(). 
-Otherwise, earlier messages may get held up for long periods of time if more 
-recent messages are continually draining the token bucket. This can lead to 
-out of order messages or the appearance of "lost" messages under heavy load.
+Waiting calls to `removeTokens()` are processed in FIFO order on each instance,
+using one active timer per instance. Concurrent calls are supported. A request
+is charged only when it can succeed; hierarchical buckets debit the child and
+all parents together. Independent children sharing a parent do not have a global
+FIFO order. `tryRemoveTokens()` and `fireImmediately` requests do not join the
+waiting queue and may consume capacity ahead of waiting requests.
+
+`RateLimiter` combines a continuously refilled token bucket with an interval
+counter. Its interval starts at construction and resets on the first removal
+attempt at or after the previous interval expires. It is **not** a rolling-window
+limiter: traffic around an interval boundary can exceed the configured count in
+a sliding window. It does not track when your asynchronous work finishes, and it
+does not limit simultaneous in-flight operations. Use a concurrency limiter or
+rolling-window algorithm separately when those are your requirements.
+
+`getTokensRemaining()` reports the underlying bucket's balance (possibly
+fractional), not the interval counter's remaining allowance. Use
+`tryRemoveTokens()` to test whether a request can proceed immediately.
+
+Token counts and capacities must be finite, non-negative numbers at most
+`Number.MAX_SAFE_INTEGER`; fractional tokens are supported. Numeric intervals
+must be finite and greater than zero. Invalid input throws `RangeError` (an
+async call rejects). An oversized request returns `false` from `tryRemoveTokens()`
+and rejects from `removeTokens()`. A standalone `TokenBucket` starts empty;
+`RateLimiter` starts full. For compatibility, `bucketSize: 0` means unlimited
+capacity and bypasses parents, and `tokensPerInterval: 0` refills a finite bucket
+to capacity on every attempt. These zero values do **not** disable all traffic.
+`RateLimiter` with `tokensPerInterval: 0` accepts only zero-token requests.
+
+Timing uses a monotonic clock. Timers can run late when the event loop is busy;
+availability is rechecked after every wait. Balances use JavaScript floating-point
+numbers, so fractional results can have normal rounding error. State is local to
+the instance and is not shared across processes, workers, or machines. Configure
+instances at the scope of the resource you want to limit; creating a new limiter
+for every request defeats a shared rate limit.
 
 ## License
 

src/RateLimiter.ts

@@ -1,4 +1,5 @@
-import { Interval, TokenBucket } from "./TokenBucket.js";
+import { RequestQueue } from "./RequestQueue.js";
+import { Interval, TokenBucket, validateTokens } from "./TokenBucket.js";
 import { getMilliseconds, wait } from "./clock.js";
 
 export type RateLimiterOpts = {
@@ -20,6 +21,7 @@ export type RateLimiterOpts = {
  *  immediately when rate limiting is in effect (default is false).
  */
 export class RateLimiter {
+  private readonly requests = new RequestQueue();
   tokenBucket: TokenBucket;
   curIntervalStart: number;
   tokensThisInterval: number;
@@ -49,40 +51,25 @@ export class RateLimiter {
    * @returns A promise for the remainingTokens count.
    */
   async removeTokens(count: number): Promise<number> {
-    // Make sure the request isn't for more than we can handle
+    validateTokens(count, "count");
     if (count > this.tokenBucket.bucketSize) {
-      throw new Error(
+      throw new RangeError(
         `Requested tokens ${count} exceeds maximum tokens per interval ${this.tokenBucket.bucketSize}`,
       );
     }
-
-    const now = getMilliseconds();
-
-    // Advance the current interval and reset the current interval token count
-    // if needed
-    if (now < this.curIntervalStart || now - this.curIntervalStart >= this.tokenBucket.interval) {
-      this.curIntervalStart = now;
-      this.tokensThisInterval = 0;
+    if (this.fireImmediately) {
+      return this.tryRemoveTokens(count) ? this.tokenBucket.content : -1;
     }
-
-    // If we don't have enough tokens left in this interval, wait until the
-    // next interval
-    if (count > this.tokenBucket.tokensPerInterval - this.tokensThisInterval) {
-      if (this.fireImmediately) {
-        return -1;
-      } else {
-        const waitMs = Math.ceil(this.curIntervalStart + this.tokenBucket.interval - now);
-        await wait(waitMs);
-        const remainingTokens = await this.tokenBucket.removeTokens(count);
-        this.tokensThisInterval += count;
-        return remainingTokens;
+    return this.requests.run(async () => {
+      while (true) {
+        if (this.tryRemoveTokens(count)) return this.tokenBucket.content;
+        const intervalDelay =
+          count > this.tokenBucket.tokensPerInterval - this.tokensThisInterval
+            ? this.curIntervalStart + this.tokenBucket.interval - getMilliseconds()
+            : 0;
+        await wait(Math.max(1, Math.ceil(intervalDelay), this.tokenBucket.getWaitTime(count)));
       }
-    }
-
-    // Remove the requested number of tokens from the token bucket
-    const remainingTokens = await this.tokenBucket.removeTokens(count);
-    this.tokensThisInterval += count;
-    return remainingTokens;
+    });
   }
 
   /**
@@ -95,6 +82,7 @@ export class RateLimiter {
    *  false.
    */
   tryRemoveTokens(count: number): boolean {
+    validateTokens(count, "count");
     // Make sure the request isn't for more than we can handle
     if (count > this.tokenBucket.bucketSize) return false;
 

src/RequestQueue.ts

@@ -0,0 +1,17 @@
+/** Serialize waiting requests so a backlog needs only one active timer. */
+export class RequestQueue {
+  private tail?: Promise<void>;
+
+  run<T>(request: () => Promise<T>): Promise<T> {
+    const result = this.tail ? this.tail.then(request) : request();
+    const tail = result.then(
+      () => undefined,
+      () => undefined,
+    );
+    this.tail = tail;
+    void tail.then(() => {
+      if (this.tail === tail) this.tail = undefined;
+    });
+    return result;
+  }
+}

src/TokenBucket.ts

@@ -1,3 +1,4 @@
+import { RequestQueue } from "./RequestQueue.js";
 import { getMilliseconds, wait } from "./clock.js";
 
 export type Interval = number | "second" | "sec" | "minute" | "min" | "hour" | "hr" | "day";
@@ -24,6 +25,7 @@ export type TokenBucketOpts = {
  *  the parent of this bucket.
  */
 export class TokenBucket {
+  private readonly requests = new RequestQueue();
   bucketSize: number;
   tokensPerInterval: number;
   interval: number;
@@ -32,6 +34,8 @@ export class TokenBucket {
   lastDrip: number;
 
   constructor({ bucketSize, tokensPerInterval, interval, parentBucket }: TokenBucketOpts) {
+    validateTokens(bucketSize, "bucketSize");
+    validateTokens(tokensPerInterval, "tokensPerInterval");
     this.bucketSize = bucketSize;
     this.tokensPerInterval = tokensPerInterval;
 
@@ -59,6 +63,9 @@ export class TokenBucket {
       this.interval = interval;
     }
 
+    if (!Number.isFinite(this.interval) || this.interval <= 0) {
+      throw new RangeError("interval must be a finite positive number of milliseconds");
+    }
     this.parentBucket = parentBucket;
     this.content = 0;
     this.lastDrip = getMilliseconds();
@@ -72,47 +79,42 @@ export class TokenBucket {
    * @returns A promise for the remainingTokens count.
    */
   async removeTokens(count: number): Promise<number> {
-    // Is this an infinite size bucket?
-    if (this.bucketSize === 0) {
-      return Number.POSITIVE_INFINITY;
-    }
+    validateTokens(count, "count");
+    this.getWaitTime(count);
+    return this.requests.run(async () => {
+      while (true) {
+        if (this.tryRemoveTokens(count)) return this.remainingTokens();
+        await wait(Math.max(1, this.getWaitTime(count)));
+      }
+    });
+  }
 
-    // Make sure the bucket can hold the requested number of tokens
-    if (count > this.bucketSize) {
-      throw new Error(`Requested tokens ${count} exceeds bucket size ${this.bucketSize}`);
+  /** Return the wait required by this bucket and all of its ancestors. */
+  getWaitTime(count: number): number {
+    validateTokens(count, "count");
+    let delay = 0;
+    for (const bucket of hierarchy(this)) {
+      if (count > bucket.bucketSize) {
+        throw new RangeError(`Requested tokens ${count} exceeds bucket size ${bucket.bucketSize}`);
+      }
+      bucket.drip();
+      if (count > bucket.content) {
+        delay = Math.max(
+          delay,
+          Math.ceil((count - bucket.content) / (bucket.tokensPerInterval / bucket.interval)),
+        );
+      }
     }
+    return delay;
+  }
 
-    // Drip new tokens into this bucket
-    this.drip();
-
-    const comeBackLater = async () => {
-      // How long do we need to wait to make up the difference in tokens?
-      const waitMs = Math.ceil((count - this.content) * (this.interval / this.tokensPerInterval));
-      await wait(waitMs);
-      return this.removeTokens(count);
-    };
-
-    // If we don't have enough tokens in this bucket, come back later
-    if (count > this.content) return comeBackLater();
-
-    if (this.parentBucket != undefined) {
-      // Remove the requested from the parent bucket first
-      const remainingTokens = await this.parentBucket.removeTokens(count);
-
-      // Check that we still have enough tokens in this bucket
-      if (count > this.content) return comeBackLater();
-
-      // Tokens were removed from the parent bucket, now remove them from
-      // this bucket. Note that we look at the current bucket and parent
-      // bucket's remaining tokens and return the smaller of the two values
-      this.content -= count;
-
-      return Math.min(remainingTokens, this.content);
-    } else {
-      // Remove the requested tokens from this bucket
-      this.content -= count;
-      return this.content;
+  private remainingTokens(): number {
+    if (!this.parentBucket) return this.bucketSize === 0 ? Number.POSITIVE_INFINITY : this.content;
+    let remaining = Number.POSITIVE_INFINITY;
+    for (const bucket of hierarchy(this)) {
+      remaining = Math.min(remaining, bucket.content);
     }
+    return remaining;
   }
 
   /**
@@ -124,23 +126,24 @@ export class TokenBucket {
    *  false.
    */
   tryRemoveTokens(count: number): boolean {
-    // Is this an infinite size bucket?
-    if (!this.bucketSize) return true;
-
-    // Make sure the bucket can hold the requested number of tokens
-    if (count > this.bucketSize) return false;
-
-    // Drip new tokens into this bucket
-    this.drip();
-
-    // If we don't have enough tokens in this bucket, return false
-    if (count > this.content) return false;
-
-    // Try to remove the requested tokens from the parent bucket
-    if (this.parentBucket && !this.parentBucket.tryRemoveTokens(count)) return false;
-
-    // Remove the requested tokens from this bucket and return
-    this.content -= count;
+    validateTokens(count, "count");
+    // Check the whole hierarchy before charging any bucket. No await may occur
+    // between this check and the debit: competing callers must see the debit.
+    if (this.bucketSize === 0) return true;
+    if (!this.parentBucket) {
+      if (count > this.bucketSize) return false;
+      this.drip();
+      if (count > this.content) return false;
+      this.content -= count;
+      return true;
+    }
+    const buckets = hierarchy(this);
+    for (const bucket of buckets) {
+      if (count > bucket.bucketSize) return false;
+      bucket.drip();
+      if (count > bucket.content) return false;
+    }
+    for (const bucket of buckets) bucket.content -= count;
     return true;
   }
 
@@ -159,9 +162,37 @@ export class TokenBucket {
     const deltaMS = Math.max(now - this.lastDrip, 0);
     this.lastDrip = now;
 
-    const dripAmount = deltaMS * (this.tokensPerInterval / this.interval);
+    // An extremely short interval can overflow the rate even though both
+    // inputs are finite. In particular, 0 * Infinity would poison the bucket.
+    const rate = this.tokensPerInterval / this.interval;
+    const dripAmount = Number.isFinite(rate)
+      ? deltaMS * rate
+      : (deltaMS * this.tokensPerInterval) / this.interval;
     const prevContent = this.content;
     this.content = Math.min(this.content + dripAmount, this.bucketSize);
     return Math.floor(this.content) > Math.floor(prevContent);
   }
 }
+
+export function validateTokens(value: number, name: string): void {
+  if (!Number.isFinite(value) || value < 0 || value > Number.MAX_SAFE_INTEGER) {
+    throw new RangeError(
+      `${name} must be a finite non-negative number no greater than Number.MAX_SAFE_INTEGER`,
+    );
+  }
+}
+
+function hierarchy(start: TokenBucket): TokenBucket[] {
+  const buckets: TokenBucket[] = [];
+  const visited = new Set<TokenBucket>();
+  for (
+    let bucket: TokenBucket | undefined = start;
+    bucket && bucket.bucketSize !== 0;
+    bucket = bucket.parentBucket
+  ) {
+    if (visited.has(bucket)) throw new Error("Circular parentBucket hierarchy");
+    visited.add(bucket);
+    buckets.push(bucket);
+  }
+  return buckets;
+}