Lesson 5.3: Promise Queues and Rate Limiting

Why Rate Limiting?

Rate limiting prevents:

• API throttling and bans
• Server overload
• Resource exhaustion
• Excessive costs

Simple Concurrency Limiter

async function limitConcurrency(tasks, limit) {
  const results = [];
  const executing = [];
  
  for (const task of tasks) {
    const promise = Promise.resolve().then(() => task());
    results.push(promise);
    
    if (limit <= tasks.length) {
      const e = promise.then(() => {
        executing.splice(executing.indexOf(e), 1);
      });
      executing.push(e);
      
      if (executing.length >= limit) {
        await Promise.race(executing);
      }
    }
  }
  
  return Promise.all(results);
}

// Usage: Max 3 concurrent operations
const tasks = userIds.map(id => () => fetchUser(id));
const results = await limitConcurrency(tasks, 3);

Promise Queue

class PromiseQueue {
  constructor(concurrency = 1) {
    this.concurrency = concurrency;
    this.running = 0;
    this.queue = [];
  }
  
  async add(fn) {
    while (this.running >= this.concurrency) {
      await new Promise(resolve => this.queue.push(resolve));
    }
    
    this.running++;
    
    try {
      return await fn();
    } finally {
      this.running--;
      const resolve = this.queue.shift();
      if (resolve) resolve();
    }
  }
}

// Usage
const queue = new PromiseQueue(3);

const results = await Promise.all(
  userIds.map(id => queue.add(() => fetchUser(id)))
);

Rate Limiter (Requests per Second)

class RateLimiter {
  constructor(requestsPerSecond) {
    this.requestsPerSecond = requestsPerSecond;
    this.interval = 1000 / requestsPerSecond;
    this.lastCallTime = 0;
  }
  
  async execute(fn) {
    const now = Date.now();
    const timeSinceLastCall = now - this.lastCallTime;
    
    if (timeSinceLastCall < this.interval) {
      const delay = this.interval - timeSinceLastCall;
      await new Promise(resolve => setTimeout(resolve, delay));
    }
    
    this.lastCallTime = Date.now();
    return await fn();
  }
}

// Usage: 10 requests per second
const limiter = new RateLimiter(10);

for (const id of userIds) {
  const user = await limiter.execute(() => fetchUser(id));
  console.log(user);
}

Token Bucket Algorithm

class TokenBucket {
  constructor(capacity, refillRate) {
    this.capacity = capacity;
    this.tokens = capacity;
    this.refillRate = refillRate; // tokens per second
    this.lastRefill = Date.now();
  }
  
  refill() {
    const now = Date.now();
    const timePassed = (now - this.lastRefill) / 1000;
    const tokensToAdd = timePassed * this.refillRate;
    
    this.tokens = Math.min(this.capacity, this.tokens + tokensToAdd);
    this.lastRefill = now;
  }
  
  async consume(tokens = 1) {
    this.refill();
    
    while (this.tokens < tokens) {
      const waitTime = ((tokens - this.tokens) / this.refillRate) * 1000;
      await new Promise(resolve => setTimeout(resolve, waitTime));
      this.refill();
    }
    
    this.tokens -= tokens;
  }
  
  async execute(fn, tokens = 1) {
    await this.consume(tokens);
    return await fn();
  }
}

// Usage: 100 token capacity, refill 10 per second
const bucket = new TokenBucket(100, 10);

for (const id of userIds) {
  const user = await bucket.execute(() => fetchUser(id));
  console.log(user);
}

Sliding Window Rate Limiter

class SlidingWindowLimiter {
  constructor(maxRequests, windowMs) {
    this.maxRequests = maxRequests;
    this.windowMs = windowMs;
    this.requests = [];
  }
  
  async execute(fn) {
    const now = Date.now();
    
    // Remove old requests outside window
    this.requests = this.requests.filter(
      time => now - time < this.windowMs
    );
    
    // Wait if at limit
    while (this.requests.length >= this.maxRequests) {
      const oldestRequest = this.requests[0];
      const waitTime = this.windowMs - (now - oldestRequest);
      
      if (waitTime > 0) {
        await new Promise(resolve => setTimeout(resolve, waitTime));
      }
      
      // Refresh after waiting
      const currentTime = Date.now();
      this.requests = this.requests.filter(
        time => currentTime - time < this.windowMs
      );
    }
    
    this.requests.push(Date.now());
    return await fn();
  }
}

// Usage: 100 requests per minute
const limiter = new SlidingWindowLimiter(100, 60000);

for (const id of userIds) {
  const user = await limiter.execute(() => fetchUser(id));
  console.log(user);
}

Combined Queue and Rate Limiter

class ThrottledQueue {
  constructor(concurrency, requestsPerSecond) {
    this.queue = new PromiseQueue(concurrency);
    this.limiter = new RateLimiter(requestsPerSecond);
  }
  
  async add(fn) {
    return await this.queue.add(async () => {
      return await this.limiter.execute(fn);
    });
  }
}

// Usage: Max 5 concurrent, 10 per second
const throttledQueue = new ThrottledQueue(5, 10);

const results = await Promise.all(
  userIds.map(id => throttledQueue.add(() => fetchUser(id)))
);

Real-World Example: API Client with Rate Limiting

class RateLimitedAPIClient {
  constructor(baseURL, options = {}) {
    this.baseURL = baseURL;
    this.limiter = new TokenBucket(
      options.capacity || 100,
      options.refillRate || 10
    );
  }
  
  async request(endpoint, options = {}) {
    return await this.limiter.execute(async () => {
      const url = `${this.baseURL}${endpoint}`;
      const response = await fetch(url, options);
      
      if (!response.ok) {
        throw new Error(`HTTP ${response.status}`);
      }
      
      return await response.json();
    });
  }
  
  async batchRequest(endpoints, batchSize = 10) {
    const results = [];
    
    for (let i = 0; i < endpoints.length; i += batchSize) {
      const batch = endpoints.slice(i, i + batchSize);
      
      const batchResults = await Promise.all(
        batch.map(endpoint => this.request(endpoint))
      );
      
      results.push(...batchResults);
    }
    
    return results;
  }
}

// Usage
const api = new RateLimitedAPIClient('https://api.example.com', {
  capacity: 100,
  refillRate: 10
});

const users = await api.batchRequest(
  userIds.map(id => `/users/${id}`),
  10
);

Adaptive Rate Limiting

class AdaptiveRateLimiter {
  constructor(initialRate = 10) {
    this.rate = initialRate;
    this.minRate = 1;
    this.maxRate = 100;
    this.successCount = 0;
    this.failureCount = 0;
  }
  
  async execute(fn) {
    const delay = 1000 / this.rate;
    await new Promise(resolve => setTimeout(resolve, delay));
    
    try {
      const result = await fn();
      this.onSuccess();
      return result;
    } catch (error) {
      if (error.status === 429) { // Rate limited
        this.onRateLimit();
      } else {
        this.onFailure();
      }
      throw error;
    }
  }
  
  onSuccess() {
    this.successCount++;
    
    // Gradually increase rate after successes
    if (this.successCount >= 10) {
      this.rate = Math.min(this.rate * 1.1, this.maxRate);
      this.successCount = 0;
      console.log(`Rate increased to ${this.rate.toFixed(1)}/s`);
    }
  }
  
  onRateLimit() {
    // Aggressively decrease on rate limit
    this.rate = Math.max(this.rate * 0.5, this.minRate);
    this.successCount = 0;
    console.log(`Rate limited! Decreased to ${this.rate.toFixed(1)}/s`);
  }
  
  onFailure() {
    this.failureCount++;
    
    if (this.failureCount >= 3) {
      this.rate = Math.max(this.rate * 0.8, this.minRate);
      this.failureCount = 0;
      console.log(`Rate decreased to ${this.rate.toFixed(1)}/s`);
    }
  }
}

// Usage
const limiter = new AdaptiveRateLimiter(10);

for (const id of userIds) {
  try {
    const user = await limiter.execute(() => fetchUser(id));
    console.log(user);
  } catch (error) {
    console.error(`Failed to fetch user ${id}`);
  }
}

Priority Queue

class PriorityQueue {
  constructor(concurrency = 1) {
    this.concurrency = concurrency;
    this.running = 0;
    this.queue = [];
  }
  
  async add(fn, priority = 0) {
    return new Promise((resolve, reject) => {
      this.queue.push({ fn, priority, resolve, reject });
      this.queue.sort((a, b) => b.priority - a.priority);
      this.process();
    });
  }
  
  async process() {
    if (this.running >= this.concurrency || this.queue.length === 0) {
      return;
    }
    
    this.running++;
    const { fn, resolve, reject } = this.queue.shift();
    
    try {
      const result = await fn();
      resolve(result);
    } catch (error) {
      reject(error);
    } finally {
      this.running--;
      this.process();
    }
  }
}

// Usage
const queue = new PriorityQueue(3);

// High priority tasks
queue.add(() => fetchCriticalData(), 10);

// Normal priority
queue.add(() => fetchUser(1), 5);

// Low priority
queue.add(() => fetchAnalytics(), 1);

Best Practices

1. Respect API limits - Stay below documented limits
2. Use exponential backoff on rate limit errors
3. Batch requests when possible
4. Cache results to reduce requests
5. Monitor rate limit headers from APIs
6. Implement adaptive limiting for dynamic adjustment
7. Use priority queues for important tasks

Next Steps

Next, we'll learn about cancellable Promises using AbortController!