TypeScriptとXState v5によるアクターモデル：分散データパイプラインの構築

Table of Contents

アプローチの比較：複雑な非同期ロジックのオーケストレーション

バックエンドのマイクロサービスや高トラフィックなブラウザセッション全体で非同期状態を管理する場合、通常は「イベント駆動型Pub/Sub」と「アクターモデル」という2つのパターンのどちらかを選択することになります。

Redis Pub/Sub、Apache Kafka、Node.jsのEventEmitterなどに代表される標準的なPub/Subアーキテクチャでは、共有チャネルにイベントをブロードキャストし、独立したリスナーがそれぞれのペースでメッセージを消費します。この疎結合設計により水平スケーリングは容易になりますが、状態の不整合（ステートドリフト）を招きやすくなります。

たとえば、5つの分散リスナーが同時に同一のPostgreSQLレコードを変更しようとすると、競合状態（レースコンディション）が発生します。エラーハンドリングも個別のコールバックに分散してしまいます。エンティティの正確なライフサイクルを復元するためだけに、十数個のマイクロサービスにまたがるログを集約しなければならないケースも珍しくありません。

アクターモデルは、状態、振る舞い、そして専用のメッセージキューを「アクター」と呼ばれる独立したユニット内に集約（co-locate）することで、この課題を解決します。アクターは自身のローカル状態のみを変更し、外部とのやり取りは直接の非同期メッセージに厳密に限定されます。XState v5は、コアのプリミティブをこの標準に基づいて再設計しました。ステートマシン、Promise、Observable、コールバックがファーストクラスのアクターとなり、追加設定なしで子アクターの生成（spawn）、監視（supervise）、メッセージ送受信を行えます。

XState v5でアクターモデルを採用するメリットとデメリット

本番パイプラインを書き換える前に、XState v5をアクターランタイムとして採用する際の明確なアーキテクチャ上のトレードオフを評価しておくことが重要です。

メリット

プロセス内の競合状態を完全に排除：各アクターはメールボックスから順次メッセージを処理します。複雑な非同期操作中であっても、ミューテックスロックやセマフォ、Read-After-Writeの競合を気にする必要がありません。
階層的な監視（スーパービジョン）：親アクターは、チャンク単位のアップロードやソケットのハートビートといった独立したタスクのために専用の子アクターを生成できます。子アクターがクラッシュしても、スーパーバイザー（監視アクター）がプロセス全体を停止させることなく、クリーンに再起動、スロットリング、または終了させることができます。
コンパイル時の型安全性：XState v5はTypeScriptのテンプレートリテラル型や文脈的型付け（Contextual Typing）を活用しています。イベントスキーマ、コンテキストプロパティ、子アクターへの入力が、冗長なボイラープレートなしでエンドツーエンドに型検証されます。
視覚的な追跡可能性（トレーサビリティ）：ワークフローは宣言的に記述されます。定義したマシンをXState Statelyのビジュアライザに直接エクスポートできるため、本番環境にデプロイする前にエッジケースの検証や障害パスのシミュレーションが可能です。

デメリット

学習コストが高い：従来の命令型async/awaitスクリプトから、メッセージ駆動型のメールボックス思考へと切り替えるには慣れが必要です。イベントストリームと状態遷移をベースに設計する考え方をチーム全体で習得しなければなりません。
メモリフットプリント：XStateのアクターインスタンス1つにつき、約2.5 KB〜5 KBのヒープメモリを消費します。並行する10,000個のアクターを実行しても約35 MBのRAM消費で済みますが、単一のNode.jsランタイム上で100万個の常駐アクターをスケールさせるにはメモリプロファイリングが必要です。軽量さの面ではシンプルな非同期クロージャの方が有利です。
永続化のオーバーヘッド：サーバー再起動をまたぐような長時間実行ワークフローには、カスタムの永続化アダプタが必要です。アクターのコンテキストスナップショットをRedisやDynamoDBなどのストレージに保存し、起動時に状態を復元（リハイドレーション）する必要があります。

推奨セットアップ：プロジェクトの初期化とアーキテクチャ

単純なCRUD操作を超えて毎秒数千件のイベントを取り込む規模になると、スーパーバイザーと子アクター間の循環参照によって開発が停滞しがちです。マシン定義を個別のモジュールに分離しておくことで、疎結合な状態を維持できます。

厳格なESMサポートを有効にした最新のTypeScriptプロジェクトを初期化します：

mkdir xstate-actor-pipeline
cd xstate-actor-pipeline
npm init -y
npm install xstate@^5.18.0
npm install -D typescript @types/node tsx
npx tsc --init

最新のモジュール解決と厳格な型チェックを適用するためにtsconfig.jsonを設定します：

{
  "compilerOptions": {
    "target": "ES2022",
    "module": "NodeNext",
    "moduleResolution": "NodeNext",
    "strict": true,
    "esModuleInterop": true,
    "skipLibCheck": true
  }
}

実装ガイド：耐障害性に優れたデータ取り込みパイプラインの構築

以下の例では、堅牢なデータ取り込みパイプラインを構築します。ルートのpipelineSupervisorが、最大並行ワーカー数を2に制限しつつバッチ処理を調整し、子アクターであるdataWorkerに各チャンクを委譲します。

1. ワーカーアクターの定義

ワーカーは単一バッチの取り込み処理を担当します。20%の一時的な障害率でネットワークI/Oをシミュレートし、最大3回までの再試行を自身で管理して、結果をスーパーバイザーに報告します。

import { setup, fromPromise, sendParent, assign } from 'xstate';

export interface WorkerInput {
  batchId: string;
  items: string[];
}

export type WorkerEvents =
  | { type: 'PROCESS' }
  | { type: 'RETRY' };

const uploadBatchLogic = fromPromise(async ({ input }: { input: { batchId: string; items: string[] } }) => {
  // 20%の一時的な障害率でネットワークI/Oをシミュレート
  if (Math.random() < 0.2) {
    throw new Error(`バッチ処理中に一時的なネットワーク障害が発生しました: ${input.batchId}`);
  }
  return { batchId: input.batchId, processedCount: input.items.length };
});

export const workerMachine = setup({
  types: {
    context: {} as { batchId: string; items: string[]; attempts: number },
    input: {} as WorkerInput,
    events: {} as WorkerEvents,
  },
  actors: {
    uploadBatch: uploadBatchLogic,
  },
}).createMachine({
  id: 'dataWorker',
  initial: 'idle',
  context: ({ input }) => ({
    batchId: input.batchId,
    items: input.items,
    attempts: 0,
  }),
  states: {
    idle: {
      on: {
        PROCESS: 'processing',
      },
    },
    processing: {
      entry: assign({
        attempts: ({ context }) => context.attempts + 1,
      }),
      invoke: {
        id: 'uploadTask',
        src: 'uploadBatch',
        input: ({ context }) => ({ batchId: context.batchId, items: context.items }),
        onDone: {
          target: 'completed',
          actions: sendParent(({ event }) => ({
            type: 'WORKER_SUCCESS',
            batchId: event.output.batchId,
            processedCount: event.output.processedCount,
          })),
        },
        onError: {
          target: 'failed',
        },
      },
    },
    failed: {
      always: [
        {
          guard: ({ context }) => context.attempts < 3,
          target: 'processing',
        },
        {
          target: 'aborted',
          actions: sendParent(({ context }) => ({
            type: 'WORKER_FAILED',
            batchId: context.batchId,
          })),
        },
      ],
    },
    completed: { type: 'final' },
    aborted: { type: 'final' },
  },
});

2. スーパーバイザーアクターの実装

スーパーバイザーはキューの管理を担います。稼働中のワーカー数を最大2つに制限しながらオンデマンドでワーカーを生成し、ワーカーの完了または失敗に応じて内部レジストリを更新します。

import { setup, createActor, assign, type ActorRefFrom } from 'xstate';
import { workerMachine } from './worker.js';

type WorkerRef = ActorRefFrom<typeof workerMachine>;

interface SupervisorContext {
  queue: Array<{ batchId: string; items: string[] }>;
  activeWorkers: Record<string, WorkerRef>;
  completedBatches: string[];
  failedBatches: string[];
}

type SupervisorEvents =
  | { type: 'QUEUE_BATCH'; batchId: string; items: string[] }
  | { type: 'WORKER_SUCCESS'; batchId: string; processedCount: number }
  | { type: 'WORKER_FAILED'; batchId: string };

const MAX_CONCURRENT_WORKERS = 2;

export const supervisorMachine = setup({
  types: {
    context: {} as SupervisorContext,
    events: {} as SupervisorEvents,
  },
  actions: {
    spawnNextWorker: assign(({ context, spawn }) => {
      const activeCount = Object.keys(context.activeWorkers).length;
      if (activeCount >= MAX_CONCURRENT_WORKERS || context.queue.length === 0) {
        return {};
      }

      const [nextBatch, ...remainingQueue] = context.queue;
      const child = spawn(workerMachine, {
        id: `worker-${nextBatch.batchId}`,
        input: nextBatch,
      });

      child.send({ type: 'PROCESS' });

      return {
        queue: remainingQueue,
        activeWorkers: {
          ...context.activeWorkers,
          [nextBatch.batchId]: child,
        },
      };
    }),
  },
}).createMachine({
  id: 'supervisor',
  initial: 'active',
  context: {
    queue: [],
    activeWorkers: {},
    completedBatches: [],
    failedBatches: [],
  },
  states: {
    active: {
      on: {
        QUEUE_BATCH: {
          actions: [
            assign({
              queue: ({ context, event }) => [
                ...context.queue,
                { batchId: event.batchId, items: event.items },
              ],
            }),
            'spawnNextWorker',
          ],
        },
        WORKER_SUCCESS: {
          actions: [
            assign({
              completedBatches: ({ context, event }) => [
                ...context.completedBatches,
                event.batchId,
              ],
              activeWorkers: ({ context, event }) => {
                const updated = { ...context.activeWorkers };
                delete updated[event.batchId];
                return updated;
              },
            }),
            'spawnNextWorker',
          ],
        },
        WORKER_FAILED: {
          actions: [
            assign({
              failedBatches: ({ context, event }) => [
                ...context.failedBatches,
                event.batchId,
              ],
              activeWorkers: ({ context, event }) => {
                const updated = { ...context.activeWorkers };
                delete updated[event.batchId];
                return updated;
              },
            }),
            'spawnNextWorker',
          ],
        },
      },
    },
  },
});

3. システムの実行と監視

ルートのスーパーバイザーアクターを起動し、4つのテストペイロードを投入します：

import { createActor } from 'xstate';
import { supervisorMachine } from './supervisor.js';

const supervisor = createActor(supervisorMachine);

supervisor.subscribe((snapshot) => {
  const { activeWorkers, completedBatches, failedBatches, queue } = snapshot.context;
  console.log(
    `[ステータス] 稼働中: ${Object.keys(activeWorkers).length} | ` +
    `キュー内: ${queue.length} | ` +
    `完了: ${completedBatches.length} | ` +
    `失敗: ${failedBatches.length}`
  );
});

supervisor.start();

supervisor.send({ type: 'QUEUE_BATCH', batchId: 'b-101', items: ['payloadA', 'payloadB'] });
supervisor.send({ type: 'QUEUE_BATCH', batchId: 'b-102', items: ['payloadC'] });
supervisor.send({ type: 'QUEUE_BATCH', batchId: 'b-103', items: ['payloadD', 'payloadE'] });
supervisor.send({ type: 'QUEUE_BATCH', batchId: 'b-104', items: ['payloadF'] });

コンソール出力を確認してください。スーパーバイザーが並行処理数の上限を厳格に順守していることがわかります。バッチb-101とb-102は即座に開始され、b-103とb-104はキューで待機します。各ワーカーはネットワーク障害をローカルで処理し、ワーカーが完了するとスーパーバイザーが自動的にキューを消化していきます。ミューテックスやグローバル変数を一切使わずに、すべてが安全に制御されます。