TypeScriptで同じキーの非同期処理を直列化する：Keyed Serial Queue

同じキーの非同期処理を直列化する：TypeScriptで実装する Keyed Serial Queue

元となる背景解説: https://www.aceround.app/ja/blog/ai-interview-transcript-review/

リアルタイムの文字起こしでは、同じ segmentId に対して「途中結果の保存」「revision の反映」「最終結果の確定」が短時間に到着します。この3つが非同期に走ると、古い更新が新しい更新を上書きすることがあります。

解決したい条件は次の2つです。

• 同じキーのタスクは登録順に一つずつ実行する
• 異なるキーのタスクは互いを待たず並列に実行する

この記事では、依存ライブラリなしで使える KeyedSerialQueue を実装し、失敗後の継続とキューの掃除まで扱います。面接なら「グローバルな1本のキューではなぜ不足か」「どこで順序を保証するか」を説明できる題材です。

なぜグローバルキューではないのか

すべての更新を1本の Promise チェーンに積めば順序は守れます。しかし segment-a の遅い保存が、関係ない segment-b の処理まで止めます。逆に何も制御せず Promise.all に渡すと、同じ segment の順序が壊れます。

キーごとに「最後に終わる Promise」を持てば、この2つを同時に満たせます。

ここでの不変条件は明確です。各 key について、tails.get(key) は、そのキーで登録済みの全タスクが完了または失敗処理済みになった後に完了する Promise を指します。

実装

type Task<T> = () => Promise<T>;

class KeyedSerialQueue {
  private readonly tails = new Map<string, Promise<void>>();

  run<T>(key: string, task: Task<T>): Promise<T> {
    const previous = this.tails.get(key) ?? Promise.resolve();

    // 前の失敗を次のタスクへ伝播させない。
    const current = previous.catch(() => undefined).then(task);

    // tail 自体は常に解決させ、後続タスクの入口にする。
    const tail = current.then(
      () => undefined,
      () => undefined,
    );

    this.tails.set(key, tail);

    // その後に同じキーへ積まれていなければ Map から除去する。
    void tail.finally(() => {
      if (this.tails.get(key) === tail) this.tails.delete(key);
    });

    return current;
  }
}

current は呼び出し元に本来の成功値または例外を返します。一方で tail は必ず解決する Promise です。この分離が重要です。前のタスクの例外をそのまま tails に置くと、同じキーの後続タスクまで実行されません。

また、finally 内の同一性チェックを外してはいけません。タスク A の終了を待つ間にタスク B が同じキーへ積まれている場合、A が Map を消すと B の順序保証が失われます。

動作確認

以下は Bun で実行した最小テストです。テストの狙いは実装詳細ではなく、外から観測できる3つの契約を固定することです。

const sleep = (ms: number) =>
  new Promise<void>((resolve) => setTimeout(resolve, ms));

async function testSameKeyIsSerial() {
  const queue = new KeyedSerialQueue();
  const events: string[] = [];

  const first = queue.run("segment-a", async () => {
    events.push("a1:start");
    await sleep(20);
    events.push("a1:end");
  });
  const second = queue.run("segment-a", async () => {
    events.push("a2:start");
    events.push("a2:end");
  });

  await Promise.all([first, second]);
  console.assert(
    events.join(",") === "a1:start,a1:end,a2:start,a2:end",
    events,
  );
}

async function testDifferentKeysOverlap() {
  const queue = new KeyedSerialQueue();
  const events: string[] = [];

  const a = queue.run("segment-a", async () => {
    events.push("a:start");
    await sleep(20);
    events.push("a:end");
  });
  const b = queue.run("segment-b", async () => {
    events.push("b:start");
    await sleep(1);
    events.push("b:end");
  });

  await Promise.all([a, b]);
  console.assert(events.indexOf("b:end") < events.indexOf("a:end"), events);
}

async function testFailureDoesNotBlockNextTask() {
  const queue = new KeyedSerialQueue();

  const failed = queue.run("segment-a", async () => {
    throw new Error("temporary failure");
  });
  const next = queue.run("segment-a", async () => "processed");

  await failed.catch(() => undefined);
  console.assert((await next) === "processed");
}

実行結果は 3 tests passed でした。

どこまで保証し、どこから別の仕組みが必要か

このキューが保証するのは、同一プロセス内で同じキーに登録された処理の開始順序です。複数の Node.js プロセスや複数コンテナが同じ segmentId を扱う場合、各プロセスの Map は共有されません。

その場合は、処理の所有権を一つのパーティションへ寄せる、DB の条件付き更新で revision を比較する、あるいはメッセージキューのキー付きパーティションを使う、といった分散側の順序保証が別途必要です。アプリ内キューと分散ロックを混ぜる前に、順序が必要な境界が「プロセス内」か「プロセス間」かを切り分けるのが先です。

まとめ

• キーごとに最後の Promise を保持すると、同一キーを直列、異なるキーを並列にできます
• 呼び出し元へ返す Promise と、後続をつなぐ失敗吸収済みの tail を分けます
• Map の削除には同一性チェックが必要です
• 分散環境では、この実装だけでは順序を保証できません

この小さな実装は、文字起こしの更新、同一注文のイベント処理、同一ドキュメントの保存など、同じリソースへの非同期更新がある場面にそのまま適用できます。