TypeScriptでdeadline伝播を実装する：AbortSignalでリアルタイム処理を止める設計

本稿のコードは Node.js 20 以降を前提にしています。AbortSignal.timeout() と AbortSignal.any() の仕様を使い、リアルタイム音声処理を題材にしています。

結論

リアルタイムの音声AIやWebSocket APIでは、各処理に個別のタイムアウトを置くだけでは遅延を抑え切れません。リクエストの開始時に絶対時刻の deadline を1つ決め、すべての下流処理へ AbortSignal と一緒に渡すのが扱いやすい設計です。

この方式には次の利点があります。

• STT、検索、LLMの合計時間がユーザー向けの応答予算を超えません。
• クライアント切断時に、下流の不要な処理をまとめて止められます。
• 「どの段で時間を使い切ったか」をログとメトリクスで観測できます。
• 面接のシステム設計では、timeout を足し合わせないという説明につながります。

この記事では、Node.js 20+ / TypeScriptだけで、子処理に残り時間を伝播する最小実装を作ります。

個別 timeout だけでは何が壊れるか

たとえば音声チャンクを処理する際、以下の上限をそれぞれ設定したとします。

処理	個別上限
STT	1.2秒
意図推定・検索	0.25秒
LLM	0.7秒

順番に実行されれば、最悪で 2.15 秒です。しかしUIが許容できる待ち時間が 1.7 秒なら、すでに契約を破っています。各処理が「自分は timeout 内だった」と成功しても、ユーザーは遅い応答を受け取ります。

必要なのは相対的な「あと700ms待つ」ではなく、リクエスト全体で共有する deadlineAt です。子処理の予算は毎回、次で求めます。

childBudget = min(その処理の上限, deadlineAt - 現在時刻)

残り時間が0以下なら、新しい外部呼び出しを開始しません。

実装：deadline と AbortSignal を一緒に渡す

以下はそのまま実行できる例です。親の AbortSignal は「クライアント切断」「サーバーの明示的なキャンセル」を表します。AbortSignal.timeout() は段ごとの残り時間を表し、AbortSignal.any() で2つを合成します。

class DeadlineExceeded extends Error {
  constructor(message: string) {
    super(message);
    this.name = "DeadlineExceeded";
  }
}

type DeadlineContext = {
  signal: AbortSignal;
  deadlineAt: number;
};

function abortError(signal: AbortSignal, label: string): Error {
  if (signal.reason?.name === "TimeoutError") {
    return new DeadlineExceeded(`${label}: deadline exceeded`);
  }
  return signal.reason instanceof Error
    ? signal.reason
    : new Error(`${label}: aborted`);
}

function throwIfAborted(signal: AbortSignal, label: string): void {
  if (signal.aborted) throw abortError(signal, label);
}

function childSignal(
  parent: AbortSignal,
  deadlineAt: number,
  limitMs: number,
): AbortSignal {
  const remainingMs = Math.min(limitMs, deadlineAt - Date.now());

  if (remainingMs <= 0) {
    return AbortSignal.abort(new DeadlineExceeded("request deadline exceeded"));
  }

  return AbortSignal.any([parent, AbortSignal.timeout(remainingMs)]);
}

async function runStep<T>(
  label: string,
  context: DeadlineContext,
  limitMs: number,
  action: (signal: AbortSignal) => Promise<T>,
): Promise<T> {
  const signal = childSignal(context.signal, context.deadlineAt, limitMs);
  throwIfAborted(signal, label);

  try {
    const value = await action(signal);
    throwIfAborted(signal, label);
    return value;
  } catch (error) {
    if (signal.aborted) throw abortError(signal, label);
    throw error;
  }
}

function sleep(ms: number, signal: AbortSignal): Promise<void> {
  return new Promise((resolve, reject) => {
    const timer = setTimeout(done, ms);

    function done() {
      signal.removeEventListener("abort", onAbort);
      resolve();
    }

    function onAbort() {
      clearTimeout(timer);
      signal.removeEventListener("abort", onAbort);
      reject(abortError(signal, "sleep"));
    }

    if (signal.aborted) onAbort();
    else signal.addEventListener("abort", onAbort, { once: true });
  });
}

async function fakeRemoteCall(
  name: string,
  latencyMs: number,
  signal: AbortSignal,
): Promise<string> {
  await sleep(latencyMs, signal);
  return `${name}: ok`;
}

async function processAudioChunk(context: DeadlineContext) {
  const transcript = await runStep(
    "STT",
    context,
    1_200,
    (signal) => fakeRemoteCall("transcript", 900, signal),
  );

  const [intent, retrieval] = await Promise.all([
    runStep(
      "intent",
      context,
      250,
      (signal) => fakeRemoteCall("intent", 100, signal),
    ),
    runStep(
      "retrieval",
      context,
      250,
      (signal) => fakeRemoteCall("retrieval", 120, signal),
    ),
  ]);

  return runStep(
    "LLM",
    context,
    700,
    (signal) => fakeRemoteCall(
      `${transcript} / ${intent} / ${retrieval}`,
      500,
      signal,
    ),
  );
}

const controller = new AbortController();
const context = {
  signal: controller.signal,
  deadlineAt: Date.now() + 1_700,
};

console.log(await processAudioChunk(context));

実行結果は次のようになります。

transcript: ok / intent: ok / retrieval: ok: ok

ここで重要なのは、processAudioChunk の中に「残り時間を計算する処理」を重複して書いていない点です。runStep に境界を集約しているため、新しい依存先を追加しても同じ制約を守れます。

並列処理でも deadline は1つにする

意図推定と検索は Promise.all で並列にしています。並列化すると速くなりますが、予算が増えるわけではありません。両方とも同じ deadlineAt を参照するため、どちらかが長引いてもリクエスト全体の締切を越えて走り続けません。

一方で、並列タスクの失敗方針は別途決める必要があります。

• 必須のタスク: STTや認可のように欠けると結果を返せないもの。失敗したら全体を失敗させます。
• 任意のタスク: 補助検索や要約候補のように、なくても縮退応答を返せるもの。Promise.allSettled で個別に記録し、期限内の結果だけを採用します。

「任意だから timeout を付けない」ではなく、「任意でも deadline の中でだけ実行する」がポイントです。期限後に完了した検索結果は、そのターンの応答には価値がありません。

fetch、WebSocket、SDKへ渡すときの注意

fetch は標準で signal を受け取れます。

const response = await fetch(url, {
  headers: { authorization: `Bearer ${token}` },
  signal,
});

一部のSDKは signal、abortSignal、requestOptions など名前が異なります。SDKがキャンセルを受け取れない場合でも、少なくとも呼び出し元では deadline を確認して、結果を採用しないようにします。ただし、これは接続や課金を止めるものではありません。高頻度のリアルタイム処理では、キャンセルを伝播できるクライアントを優先した方がよいです。

WebSocketは少し事情が違います。ソケット自体をリクエストごとに閉じるのではなく、メッセージ単位の requestId と deadlineAt を持たせます。サーバー側でその requestId の AbortController を管理し、切断・置換・期限切れで abort します。これなら同一接続上の別メッセージまで巻き込みません。

観測する値を先に決める

deadline を導入すると、失敗を単に「timeout」と一括りにしなくて済みます。少なくとも次を記録すると原因を切り分けやすくなります。

指標	例
残り予算	LLM開始時に何ms残っていたか
中断理由	client_disconnect / request_deadline / step_cap
段ごとの所要時間	STT・検索・LLMのp50/p95/p99
縮退率	任意タスクを捨てて応答した割合

特に「LLMのp95が遅い」の前に、LLMを開始する時点で残り時間がどれだけあるかを見ると、前段のSTTやキュー滞留が本当の原因だと分かることがあります。

面接で説明するなら

システム設計の面接では、次の順番で説明すると伝わりやすいです。

1. ユーザー体験から、エンドツーエンドのSLOを決めます。たとえば1.7秒です。
2. リクエスト開始時に絶対 deadline を生成し、下流サービスへ伝播します。
3. 各サービスは min(局所上限, 残り時間) で timeout を設定します。
4. クライアント切断や期限切れは同じ cancellation 経路で下流へ通知します。
5. 必須処理と任意処理を分け、期限内に縮退できるようにします。
6. 残り予算と中断理由を計測し、予算配分を調整します。

個別の timeout は安全装置として必要です。しかし、リアルタイム系で守りたい契約は「各サービスが何秒待ったか」ではなく「ユーザーへいつ返せたか」です。だからこそ deadline は、最初の入口で1回だけ決めて最後まで渡します。