自律エージェントの流量を制御する — バックプレッシャーと待ち行列で本番を壊さない設計

// Before: 到着した仕事を全部いっぺんに下流へ流す
async function runAgentTasks(tasks: AgentTask[]): Promise<void> {
  // tasks.length が 4 でも 400 でも、全部同時に走り出す
  await Promise.all(
    tasks.map(async (task) => {
      const result = await agent.run(task);   // 生成は速い
      await pushToGitHub(result);              // ここで下流が詰まる
    }),
  );
}

// 上限つきの待ち行列。詰まったら enqueue 側を待たせる
class BoundedQueue<T> {
  private items: T[] = [];
  private waiters: Array<() => void> = [];
 
  constructor(private readonly maxLength: number) {}
 
  // 満杯なら、空きが出るまで待たせる（ここがバックプレッシャーの起点）
  async enqueue(item: T): Promise<void> {
    while (this.items.length >= this.maxLength) {
      await new Promise<void>((resolve) => this.waiters.push(resolve));
    }
    this.items.push(item);
  }
 
  dequeue(): T | undefined {
    const item = this.items.shift();
    const waiter = this.waiters.shift();
    if (waiter) waiter(); // 空きが出たので待っている enqueue を1つ起こす
    return item;
  }
 
  get length(): number {
    return this.items.length;
  }
}

// 同時に通れる本数を N に制限するセマフォ
class Semaphore {
  private available: number;
  private queue: Array<() => void> = [];
 
  constructor(permits: number) {
    this.available = permits;
  }
 
  async acquire(): Promise<void> {
    if (this.available > 0) {
      this.available -= 1;
      return;
    }
    await new Promise<void>((resolve) => this.queue.push(resolve));
  }
 
  release(): void {
    const next = this.queue.shift();
    if (next) {
      next(); // 待っていた1件を通す（permit は移譲）
    } else {
      this.available += 1;
    }
  }
}

// 一定レートで補充されるトークンを消費して通す。バーストもある程度許容する
class TokenBucket {
  private tokens: number;
  private lastRefill = Date.now();
 
  constructor(
    private readonly capacity: number,     // バケットの最大容量（バースト許容量）
    private readonly refillPerSec: number, // 1秒あたりの補充トークン数
  ) {
    this.tokens = capacity;
  }
 
  private refill(): void {
    const now = Date.now();
    const elapsedSec = (now - this.lastRefill) / 1000;
    this.tokens = Math.min(this.capacity, this.tokens + elapsedSec * this.refillPerSec);
    this.lastRefill = now;
  }
 
  // トークンが貯まるまで待ってから1つ消費する
  async take(): Promise<void> {
    this.refill();
    while (this.tokens < 1) {
      const waitMs = ((1 - this.tokens) / this.refillPerSec) * 1000;
      await new Promise((r) => setTimeout(r, Math.ceil(waitMs)));
      this.refill();
    }
    this.tokens -= 1;
  }
}

// 部品を組み立てた流量制御つきパイプライン
class FlowControlledPipeline<T> {
  private readonly queue: BoundedQueue<T>;
  private readonly semaphore: Semaphore;
  private readonly bucket: TokenBucket;
  private running = true;
  private inFlight = 0;
 
  constructor(opts: {
    maxQueueLength: number;
    maxConcurrency: number;
    refillPerSec: number;
    burstCapacity: number;
    handle: (item: T) => Promise<void>;
    onOverflow: (item: T, reason: string) => Promise<void>;
  }) {
    this.queue = new BoundedQueue<T>(opts.maxQueueLength);
    this.semaphore = new Semaphore(opts.maxConcurrency);
    this.bucket = new TokenBucket(opts.burstCapacity, opts.refillPerSec);
    this.opts = opts;
    void this.loop();
  }
 
  private opts: {
    handle: (item: T) => Promise<void>;
    onOverflow: (item: T, reason: string) => Promise<void>;
  } & Record<string, unknown>;
 
  // エージェントはこれを呼ぶだけ。満杯ならここで待たされる（=バックプレッシャー）
  async submit(item: T): Promise<void> {
    await this.queue.enqueue(item);
  }
 
  private async loop(): Promise<void> {
    while (this.running) {
      const item = this.queue.dequeue();
      if (item === undefined) {
        await new Promise((r) => setTimeout(r, 20));
        continue;
      }
      await this.semaphore.acquire();
      await this.bucket.take();
      this.inFlight += 1;
      void this.process(item);
    }
  }
 
  private async process(item: T): Promise<void> {
    try {
      await this.opts.handle(item);
    } catch (err) {
      await this.opts.onOverflow(item, String(err));
    } finally {
      this.inFlight -= 1;
      this.semaphore.release();
    }
  }
}

async function handleWithDeadLetter(item: AgentTask): Promise<void> {
  const maxAttempts = 3;
  for (let attempt = 1; attempt <= maxAttempts; attempt += 1) {
    try {
      await pushToGitHub(item.result);
      return;
    } catch (err) {
      if (attempt === maxAttempts) {
        // 退避先（KV や R2、ログ）に理由つきで残す
        await deadLetterStore.put(item.id, {
          item,
          reason: String(err),
          failedAt: new Date().toISOString(),
        });
        return;
      }
      // 指数バックオフ。下流が回復する時間を与える
      await new Promise((r) => setTimeout(r, 2 ** attempt * 500));
    }
  }
}