OpenAI 429/Rate Limit 재시도·큐잉 패턴 7가지

type RetryOptions = {
  maxAttempts: number;
  baseDelayMs: number;
  maxDelayMs: number;
};

function sleep(ms: number) {
  return new Promise((r) => setTimeout(r, ms));
}

function isRetryable429(err: any) {
  const status = err?.status ?? err?.response?.status;
  return status === 429;
}

function clamp(n: number, min: number, max: number) {
  return Math.max(min, Math.min(max, n));
}

export async function withRetry429<T>(
  fn: () => Promise<T>,
  opt: RetryOptions
): Promise<T> {
  let lastErr: any;

  for (let attempt = 0; attempt < opt.maxAttempts; attempt++) {
    try {
      return await fn();
    } catch (err: any) {
      lastErr = err;
      if (!isRetryable429(err)) throw err;

      const exp = opt.baseDelayMs * Math.pow(2, attempt);
      const backoff = clamp(exp, 0, opt.maxDelayMs);
      const jitter = Math.floor(Math.random() * backoff);

      await sleep(jitter);
    }
  }

  throw lastErr;
}

import time
import random

def retry_after_seconds(headers: dict) -> float | None:
  ra = headers.get("retry-after") or headers.get("Retry-After")
  if not ra:
    return None
  try:
    return float(ra)
  except:
    return None

def call_with_retry(fn, max_attempts=5, base=0.5, cap=10.0):
  last = None
  for attempt in range(max_attempts):
    try:
      return fn()
    except Exception as e:
      last = e
      status = getattr(e, "status", None)
      headers = getattr(e, "headers", {}) or {}
      if status != 429:
        raise

      ra = retry_after_seconds(headers)
      if ra is not None:
        time.sleep(ra)
        continue

      backoff = min(cap, base * (2 ** attempt))
      time.sleep(random.random() * backoff)

  raise last

class TokenBucket {
  private tokens: number;
  private lastRefill: number;

  constructor(
    private capacity: number,
    private refillPerSec: number
  ) {
    this.tokens = capacity;
    this.lastRefill = Date.now();
  }

  private refill() {
    const now = Date.now();
    const elapsed = (now - this.lastRefill) / 1000;
    const add = elapsed * this.refillPerSec;
    this.tokens = Math.min(this.capacity, this.tokens + add);
    this.lastRefill = now;
  }

  async take(n = 1) {
    while (true) {
      this.refill();
      if (this.tokens >= n) {
        this.tokens -= n;
        return;
      }
      await new Promise((r) => setTimeout(r, 50));
    }
  }
}

const bucket = new TokenBucket(10, 5); // 최대 10, 초당 5개 충전

async function callOpenAI(req: any) {
  await bucket.take(1);
  // 여기서 실제 OpenAI 호출
}

import { Queue, Worker } from "bullmq";

const queue = new Queue("openai-jobs", {
  connection: { host: "localhost", port: 6379 },
});

export async function enqueueJob(payload: any) {
  const job = await queue.add("chat", payload, {
    attempts: 5,
    backoff: { type: "exponential", delay: 500 },
    removeOnComplete: true,
  });
  return job.id;
}

const worker = new Worker(
  "openai-jobs",
  async (job) => {
    // 동시성은 Worker 옵션으로 제한
    // 여기서 OpenAI 호출 + 429 재시도 래핑
    return { ok: true };
  },
  {
    connection: { host: "localhost", port: 6379 },
    concurrency: 8,
  }
);

await queue.add("chat", payload, {
  priority: 1, // 숫자가 낮을수록 우선
});

await queue.add("chat", payload, {
  priority: 10,
});

class Semaphore {
  private inFlight = 0;
  private queue: Array<() => void> = [];

  constructor(public limit: number) {}

  async acquire() {
    if (this.inFlight < this.limit) {
      this.inFlight++;
      return;
    }
    await new Promise<void>((resolve) => this.queue.push(resolve));
    this.inFlight++;
  }

  release() {
    this.inFlight--;
    const next = this.queue.shift();
    if (next) next();
  }
}

const sem = new Semaphore(8);

async function guardedCall(fn: () => Promise<any>) {
  await sem.acquire();
  try {
    const res = await fn();
    // additive increase
    sem.limit = Math.min(64, sem.limit + 1);
    return res;
  } catch (e: any) {
    const status = e?.status ?? e?.response?.status;
    if (status === 429) {
      // multiplicative decrease
      sem.limit = Math.max(1, Math.floor(sem.limit * 0.5));
    }
    throw e;
  } finally {
    sem.release();
  }
}

const inFlight = new Map<string, Promise<any>>();

function keyOf(payload: any) {
  // 실제로는 stable stringify 또는 해시를 권장
  return JSON.stringify(payload);
}

export async function coalescedCall(payload: any, fn: () => Promise<any>) {
  const key = keyOf(payload);
  const existing = inFlight.get(key);
  if (existing) return existing;

  const p = (async () => {
    try {
      return await fn();
    } finally {
      inFlight.delete(key);
    }
  })();

  inFlight.set(key, p);
  return p;
}