mcpField
C
cloudflare-workflows

cloudflare-workflows

github.com/jackspaceクラウド・DevOps1636 回インストール12 週間前に更新
リポジトリ

cloudflare-workflows について

cloudflare-workflowsは、Cloudflare Workersのワークフロー構築を支援するスキルです。ステップ実行、再試行、待機、イベント処理、起動、インスタンス管理を説明します。

主な機能

  1. 01WorkflowEntrypointによる定義
  2. 02再試行設定付きの名前付きステップ
  3. 03時間、日時、イベントによる待機
  4. 04ワークフローインスタンスの起動と管理
  5. 05Wranglerでのインスタンス確認

活用シーン

  1. 01再試行可能なステップを含む処理の構築
  2. 02指定時刻までのワークフロー待機
  3. 03外部イベント後の処理再開
  4. 04実行中インスタンスの確認や管理

インストール

コマンドをコピーしてターミナルで実行します。

$npx skills add https://github.com/jackspace/claudeskillz --skill cloudflare-workflows

インストール後、Claude Code で/cloudflare-workflowsを入力して読み込まれたか確認します。

同梱ファイルREADME.mdSKILL.json

詳細

Cloudflare Workflows

ステータス:本番環境で利用可能 ✅ 最終更新日:2025-10-22 依存関係:cloudflare-worker-base(Worker のセットアップ用) 最新バージョン:[email protected], @cloudflare/[email protected]


クイックスタート(10 分)

1. Workflow を作成する

Cloudflare Workflows のスターターテンプレートを使用します。

npm create cloudflare@latest my-workflow -- --template cloudflare/workflows-starter --git --deploy false
cd my-workflow

作成されるもの:

  • WorkflowEntrypoint クラスのテンプレート
  • ワークフローをトリガーする Worker
  • 完全な wrangler.jsonc 設定
2. 基本構造を理解する

src/index.ts:

import { WorkflowEntrypoint, WorkflowStep, WorkflowEvent } from 'cloudflare:workers';

type Env = {
  MY_WORKFLOW: Workflow;
};

type Params = {
  userId: string;
  email: string;
};

export class MyWorkflow extends WorkflowEntrypoint<Env, Params> {
  async run(event: WorkflowEvent<Params>, step: WorkflowStep) {
    // Access params from event.payload
    const { userId, email } = event.payload;

    // Step 1: Do some work
    const result = await step.do('process user', async () => {
      return { processed: true, userId };
    });

    // Step 2: Wait before next action
    await step.sleep('wait 1 hour', '1 hour');

    // Step 3: Continue workflow
    await step.do('send email', async () => {
      // Send email logic
      return { sent: true, email };
    });

    // Optional: return final state
    return { completed: true, userId };
  }
}

// Worker to trigger workflow
export default {
  async fetch(req: Request, env: Env): Promise<Response> {
    // Create new workflow instance
    const instance = await env.MY_WORKFLOW.create({
      params: { userId: '123', email: '[email protected]' }
    });

    return Response.json({
      id: instance.id,
      status: await instance.status()
    });
  }
};
3. Wrangler を設定する

wrangler.jsonc:

{
  "name": "my-workflow",
  "main": "src/index.ts",
  "compatibility_date": "2025-10-22",
  "workflows": [
    {
      "name": "my-workflow",
      "binding": "MY_WORKFLOW",
      "class_name": "MyWorkflow"
    }
  ]
}
4. デプロイしてテストする
# Deploy workflow
npm run deploy

# Trigger workflow (visit in browser or curl)
curl https://my-workflow.<subdomain>.workers.dev/

# View workflow instances
npx wrangler workflows instances list my-workflow

# Check instance status
npx wrangler workflows instances describe my-workflow <instance-id>

WorkflowEntrypoint クラス

WorkflowEntrypoint を拡張する

すべての Workflow は WorkflowEntrypoint を拡張し、run() メソッドを実装する必要があります。

export class MyWorkflow extends WorkflowEntrypoint<Env, Params> {
  async run(event: WorkflowEvent<Params>, step: WorkflowStep) {
    // Workflow steps here
  }
}

型パラメーター:

  • Env - 環境バインディング(KV、D1、R2 など)
  • Params - event.payload を介して渡されるワークフローパラメーターの型
run() メソッド
async run(
  event: WorkflowEvent<Params>,
  step: WorkflowStep
): Promise<T | void>

パラメーター:

  • event - ワークフローのメタデータとペイロードを含みます
  • step - ステップメソッド(do、sleep、sleepUntil、waitForEvent)を提供します

戻り値:

  • 省略可能な戻り値(シリアル化可能である必要があります)
  • 戻り値は instance.status() から取得できます

例:

export class OrderWorkflow extends WorkflowEntrypoint<Env, OrderParams> {
  async run(event: WorkflowEvent<OrderParams>, step: WorkflowStep) {
    const { orderId, customerId } = event.payload;

    // Access bindings via this.env
    const order = await this.env.DB.prepare(
      'SELECT * FROM orders WHERE id = ?'
    ).bind(orderId).first();

    const result = await step.do('process payment', async () => {
      // Payment processing
      return { paid: true, amount: order.total };
    });

    // Return final state
    return {
      orderId,
      status: 'completed',
      paidAmount: result.amount
    };
  }
}

ステップメソッド

step.do() - 作業を実行する
step.do<T>(
  name: string,
  config?: WorkflowStepConfig,
  callback: () => Promise<T>
): Promise<T>

または(config は省略可能です):

step.do<T>(
  name: string,
  callback: () => Promise<T>
): Promise<T>

パラメーター:

  • name - ステップ名(可観測性のため)
  • config(省略可能)- 再試行の設定
  • callback - 作業を実行する非同期関数

戻り値:

  • callback から返される値(シリアル化可能である必要があります)

例:

// Simple step
const files = await step.do('fetch files', async () => {
  const response = await fetch('https://api.example.com/files');
  return await response.json();
});

// Step with retry config
const result = await step.do(
  'call payment API',
  {
    retries: {
      limit: 10,
      delay: '10 seconds',
      backoff: 'exponential'
    },
    timeout: '5 minutes'
  },
  async () => {
    const response = await fetch('https://payment-api.example.com/charge', {
      method: 'POST',
      body: JSON.stringify({ amount: 100 })
    });
    return await response.json();
  }
);

重要 - シリアル化:

  • 戻り値は JSON シリアル化可能である必要があります
  • ✅ 使用可能:string、number、boolean、Array、Object、null
  • ❌ 使用不可:Function、Symbol、循環参照、undefined
  • 戻り値がシリアル化可能でない場合、ステップはエラーをスローします

step.sleep() - 相対時間でスリープする
step.sleep(name: string, duration: WorkflowDuration): Promise<void>

パラメーター:

  • name - ステップ名
  • duration - 数値(ミリ秒)または人が読みやすい形式の文字列

使用できる単位:

  • "second" / "seconds"
  • "minute" / "minutes"
  • "hour" / "hours"
  • "day" / "days"
  • "week" / "weeks"
  • "month" / "months"
  • "year" / "years"

例:

// Sleep for 5 minutes
await step.sleep('wait 5 minutes', '5 minutes');

// Sleep for 1 hour
await step.sleep('hourly delay', '1 hour');

// Sleep for 2 days
await step.sleep('wait 2 days', '2 days');

// Sleep using milliseconds
await step.sleep('wait 30 seconds', 30000);

// Common pattern: schedule daily task
await step.do('send daily report', async () => {
  // Send report
});
await step.sleep('wait until tomorrow', '1 day');
// Workflow continues next day

優先順位:

  • スリープから再開するワークフローは、新しいインスタンスより優先されます
  • 新しいワークフローの開始前に、古いワークフローが完了するようにします

step.sleepUntil() - 指定した日時までスリープする
step.sleepUntil(
  name: string,
  timestamp: Date | number
): Promise<void>

パラメーター:

  • name - ステップ名
  • timestamp - Date オブジェクトまたは UNIX タイムスタンプ(ミリ秒)

例:

// Sleep until specific date
const launchDate = new Date('2025-12-25T00:00:00Z');
await step.sleepUntil('wait for launch', launchDate);

// Sleep until UNIX timestamp
const timestamp = Date.parse('24 Oct 2024 13:00:00 UTC');
await step.sleepUntil('wait until time', timestamp);

// Sleep until next Monday 9am UTC
const nextMonday = new Date();
nextMonday.setDate(nextMonday.getDate() + ((1 + 7 - nextMonday.getDay()) % 7 || 7));
nextMonday.setUTCHours(9, 0, 0, 0);
await step.sleepUntil('wait until Monday 9am', nextMonday);

// Schedule work at specific time
await step.do('prepare campaign', async () => {
  // Prepare marketing campaign
});

const campaignLaunch = new Date('2025-11-01T12:00:00Z');
await step.sleepUntil('wait for campaign launch', campaignLaunch);

await step.do('launch campaign', async () => {
  // Launch campaign
});

step.waitForEvent() - 外部イベントを待機する
step.waitForEvent<T>(
  name: string,
  options: { type: string; timeout?: string | number }
): Promise<T>

パラメーター:

  • name - ステップ名
  • options.type - 一致させるイベントの種類
  • options.timeout(省略可能)- 最大待機時間(デフォルト:24 時間)

戻り値:

  • instance.sendEvent() を介して送信されたイベントペイロード

例:

export class PaymentWorkflow extends WorkflowEntrypoint<Env, Params> {
  async run(event: WorkflowEvent<Params>, step: WorkflowStep) {
    // Create payment intent
    await step.do('create payment intent', async () => {
      // Call Stripe API
    });

    // Wait for webhook from Stripe (max 1 hour)
    const webhookData = await step.waitForEvent<StripeWebhook>(
      'wait for payment confirmation',
      { type: 'stripe-webhook', timeout: '1 hour' }
    );

    // Continue based on webhook
    if (webhookData.status === 'succeeded') {
      await step.do('fulfill order', async () => {
        // Fulfill order
      });
    } else {
      await step.do('handle failed payment', async () => {
        // Handle failure
      });
    }
  }
}

// Worker receives webhook and sends event to workflow
export default {
  async fetch(req: Request, env: Env): Promise<Response> {
    if (req.url.includes('/webhook/stripe')) {
      const webhookData = await req.json();

      // Get workflow instance by ID (stored when created)
      const instance = await env.PAYMENT_WORKFLOW.get(instanceId);

      // Send event to waiting workflow
      await instance.sendEvent({
        type: 'stripe-webhook',
        payload: webhookData
      });

      return new Response('OK');
    }
  }
};

タイムアウト時の動作:

  • タイムアウトになるとエラーがスローされ、ワークフローは再試行または失敗します
  • タイムアウトによってワークフローを失敗させない場合は、try-catch で囲みます
try {
  const event = await step.waitForEvent('wait for user input', {
    type: 'user-submitted',
    timeout: '10 minutes'
  });
} catch (error) {
  // Timeout occurred - handle gracefully
  await step.do('send reminder', async () => {
    // Send reminder to user
  });
}

WorkflowStepConfig

個々のステップの再試行動作を設定します。

interface WorkflowStepConfig {
  retries?: {
    limit: number;          // Max retry attempts (Infinity allowed)
    delay: string | number; // Delay between retries
    backoff?: 'constant' | 'linear' | 'exponential';
  };
  timeout?: string | number; // Max time per attempt
}
デフォルト設定

設定が指定されていない場合、Workflows は次を使用します。

{
  retries: {
    limit: 5,
    delay: 10000,      // 10 seconds
    backoff: 'exponential'
  },
  timeout: '10 minutes'
}
再試行の例

一定のバックオフ(毎回同じ間隔):

await step.do(
  'send email',
  {
    retries: {
      limit: 3,
      delay: '30 seconds',
      backoff: 'constant'  // Always wait 30 seconds
    }
  },
  async () => {
    // Send email
  }
);

線形バックオフ(間隔を徐々に延長):

await step.do(
  'poll API',
  {
    retries: {
      limit: 5,
      delay: '1 minute',
      backoff: 'linear'  // 1m, 2m, 3m, 4m, 5m
    }
  },
  async () => {
    // Poll API
  }
);

指数バックオフ(ほとんどの場合に推奨):

await step.do(
  'call rate-limited API',
  {
    retries: {
      limit: 10,
      delay: '10 seconds',
      backoff: 'exponential'  // 10s, 20s, 40s, 80s, 160s, ...
    },
    timeout: '5 minutes'
  },
  async () => {
    // API call
  }
);

無制限の再試行:

await step.do(
  'critical operation',
  {
    retries: {
      limit: Infinity,  // Retry forever
      delay: '1 minute',
      backoff: 'exponential'
    }
  },
  async () => {
    // Operation that must succeed eventually
  }
);

再試行なし:

await step.do(
  'non-idempotent operation',
  {
    retries: {
      limit: 0  // Fail immediately on error
    }
  },
  async () => {
    // One-time operation
  }
);

エラー処理

NonRetryableError

再試行せずにワークフローを即座に失敗させます。

import { WorkflowEntrypoint, WorkflowStep, WorkflowEvent } from 'cloudflare:workers';
import { NonRetryableError } from 'cloudflare:workflows';

export class MyWorkflow extends WorkflowEntrypoint<Env, Params> {
  async run(event: WorkflowEvent<Params>, step: WorkflowStep) {
    await step.do('validate input', async () => {
      if (!event.payload.userId) {
        throw new NonRetryableError('userId is required');
      }

      // Validate user exists
      const user = await this.env.DB.prepare(
        'SELECT * FROM users WHERE id = ?'
      ).bind(event.payload.userId).first();

      if (!user) {
        // Terminal error - retrying won't help
        throw new NonRetryableError('User not found');
      }

      return user;
    });
  }
}

NonRetryableError を使用する場合:

  • ✅ 認証/認可の失敗
  • ✅ 変更されることのない無効な入力
  • ✅ リソースが存在しない(404)
  • ✅ 検証エラー
  • ❌ ネットワーク障害(再試行する必要があります)
  • ❌ レート制限(バックオフして再試行する必要があります)
  • ❌ 一時的なサービス停止(再試行する必要があります)

エラーを捕捉してワークフローを続行する

ステップのエラーを捕捉して、ワークフロー全体が失敗しないようにします。

export class MyWorkflow extends WorkflowEntrypoint<Env, Params> {
  async run(event: WorkflowEvent<Params>, step: WorkflowStep) {
    // Critical step - workflow fails if this fails
    await step.do('process payment', async () => {
      // Payment processing
    });

    // Optional step - workflow continues even if it fails
    try {
      await step.do('send confirmation email', async () => {
        // Email sending
      });
    } catch (error) {
      console.log(`Email failed: ${error.message}`);

      // Do cleanup or alternative action
      await step.do('log email failure', async () => {
        await this.env.DB.prepare(
          'INSERT INTO failed_emails (user_id, error) VALUES (?, ?)'
        ).bind(event.payload.userId, error.message).run();
      });
    }

    // Workflow continues
    await step.do('update order status', async () => {
      // Update status
    });
  }
}

パターン:柔軟な縮退動作:

// Try primary service, fall back to secondary
let result;

try {
  result = await step.do('call primary API', async () => {
    return await callPrimaryAPI();
  });
} catch (error) {
  console.log('Primary API failed, trying backup');

  result = await step.do('call backup API', async () => {
    return await callBackupAPI();
  });
}

ワークフローをトリガーする

Workers から

wrangler.jsonc でバインディングを設定します:

{
  "name": "trigger-worker",
  "main": "src/index.ts",
  "compatibility_date": "2025-10-22",
  "workflows": [
    {
      "name": "my-workflow",
      "binding": "MY_WORKFLOW",
      "class_name": "MyWorkflow",
      "script_name": "workflow-worker"  // If workflow is in different Worker
    }
  ]
}

Worker からトリガーします:

type Env = {
  MY_WORKFLOW: Workflow;
};

export default {
  async fetch(req: Request, env: Env): Promise<Response> {
    // Create new workflow instance
    const instance = await env.MY_WORKFLOW.create({
      params: {
        userId: '123',
        email: '[email protected]'
      }
    });

    // Return instance ID
    return Response.json({
      id: instance.id,
      status: await instance.status()
    });
  }
};
インスタンスのステータスを取得する
// Get instance by ID
const instance = await env.MY_WORKFLOW.get(instanceId);

// Get status
const status = await instance.status();

console.log(status);
// {
//   status: 'running' | 'complete' | 'errored' | 'queued' | 'unknown',
//   error: string | null,
//   output: any  // Return value from run() if complete
// }
実行中のインスタンスにイベントを送信する
// Get instance
const instance = await env.MY_WORKFLOW.get(instanceId);

// Send event (will be received by step.waitForEvent)
await instance.sendEvent({
  type: 'user-action',
  payload: { action: 'approved' }
});
一時停止と再開
// Pause instance
await instance.pause();

// Resume instance
await instance.resume();

// Terminate instance
await instance.terminate();

ワークフローパターン

パターン 1:長時間実行プロセス
export class VideoProcessingWorkflow extends WorkflowEntrypoint<Env, VideoParams> {
  async run(event: WorkflowEvent<VideoParams>, step: WorkflowStep) {
    const { videoId } = event.payload;

    // Step 1: Upload to processing service
    const uploadResult = await step.do('upload video', async () => {
      const video = await this.env.MY_BUCKET.get(`videos/${videoId}`);
      const response = await fetch('https://processor.example.com/upload', {
        method: 'POST',
        body: video?.body
      });
      return await response.json();
    });

    // Step 2: Wait for processing (could take hours)
    await step.sleep('wait for initial processing', '10 minutes');

    // Step 3: Poll for completion
    let processed = false;
    let attempts = 0;

    while (!processed && attempts < 20) {
      const status = await step.do(`check status attempt ${attempts}`, async () => {
        const response = await fetch(
          `https://processor.example.com/status/${uploadResult.jobId}`
        );
        return await response.json();
      });

      if (status.complete) {
        processed = true;
      } else {
        attempts++;
        await step.sleep(`wait before retry ${attempts}`, '5 minutes');
      }
    }

    // Step 4: Download processed video
    await step.do('download processed video', async () => {
      const response = await fetch(uploadResult.downloadUrl);
      const processed = await response.blob();
      await this.env.MY_BUCKET.put(`processed/${videoId}`, processed);
    });

    return { videoId, status: 'complete' };
  }
}

パターン 2:イベント駆動型の承認フロー
export class ApprovalWorkflow extends WorkflowEntrypoint<Env, ApprovalParams> {
  async run(event: WorkflowEvent<ApprovalParams>, step: WorkflowStep) {
    const { requestId, requesterId } = event.payload;

    // Step 1: Create approval request
    await step.do('create approval request', async () => {
      await this.env.DB.prepare(
        'INSERT INTO approvals (id, requester_id, status) VALUES (?, ?, ?)'
      ).bind(requestId, requesterId, 'pending').run();
    });

    // Step 2: Send notification to approvers
    await step.do('notify approvers', async () => {
      await sendNotification(requestId);
    });

    // Step 3: Wait for approval (max 7 days)
    let approvalEvent;

    try {
      approvalEvent = await step.waitForEvent<ApprovalEvent>(
        'wait for approval decision',
        { type: 'approval-decision', timeout: '7 days' }
      );
    } catch (error) {
      // Timeout - auto-reject
      await step.do('auto-reject due to timeout', async () => {
        await this.env.DB.prepare(
          'UPDATE approvals SET status = ? WHERE id = ?'
        ).bind('rejected', requestId).run();
      });

      return { requestId, status: 'rejected', reason: 'timeout' };
    }

    // Step 4: Process decision
    await step.do('process approval decision', async () => {
      await this.env.DB.prepare(
        'UPDATE approvals SET status = ?, approver_id = ? WHERE id = ?'
      ).bind(approvalEvent.approved ? 'approved' : 'rejected', approvalEvent.approverId, requestId).run();
    });

    // Step 5: Execute approved action (if approved)
    if (approvalEvent.approved) {
      await step.do('execute approved action', async () => {
        // Execute the action
      });
    }

    return { requestId, status: approvalEvent.approved ? 'approved' : 'rejected' };
  }
}

パターン 3:スケジュールされたワークフロー
export class DailyReportWorkflow extends WorkflowEntrypoint<Env, ReportParams> {
  async run(event: WorkflowEvent<ReportParams>, step: WorkflowStep) {
    // Calculate next 9am UTC
    const now = new Date();
    const tomorrow9am = new Date();
    tomorrow9am.setUTCDate(tomorrow9am.getUTCDate() + 1);
    tomorrow9am.setUTCHours(9, 0, 0, 0);

    // Sleep until tomorrow 9am
    await step.sleepUntil('wait until 9am tomorrow', tomorrow9am);

    // Generate report
    const report = await step.do('generate daily report', async () => {
      const results = await this.env.DB.prepare(
        'SELECT * FROM metrics WHERE date = ?'
      ).bind(now.toISOString().split('T')[0]).all();

      return {
        date: now.toISOString().split('T')[0],
        metrics: results.results
      };
    });

    // Send report
    await step.do('send report', async () => {
      await sendEmail({
        to: event.payload.recipients,
        subject: `Daily Report - ${report.date}`,
        body: formatReport(report.metrics)
      });
    });

    return { sent: true, date: report.date };
  }
}

パターン 4:ワークフローの連鎖
export class OrderWorkflow extends WorkflowEntrypoint<Env, OrderParams> {
  async run(event: WorkflowEvent<OrderParams>, step: WorkflowStep) {
    const { orderId } = event.payload;

    // Step 1: Process payment
    const paymentResult = await step.do('process payment', async () => {
      return await processPayment(orderId);
    });

    // Step 2: Trigger fulfillment workflow
    const fulfillmentInstance = await step.do('start fulfillment', async () => {
      return await this.env.FULFILLMENT_WORKFLOW.create({
        params: {
          orderId,
          paymentId: paymentResult.id
        }
      });
    });

    // Step 3: Wait for fulfillment to complete
    await step.sleep('wait for fulfillment', '5 minutes');

    // Step 4: Check fulfillment status
    const fulfillmentStatus = await step.do('check fulfillment', async () => {
      const instance = await this.env.FULFILLMENT_WORKFLOW.get(fulfillmentInstance.id);
      return await instance.status();
    });

    if (fulfillmentStatus.status === 'complete') {
      // Step 5: Send confirmation
      await step.do('send order confirmation', async () => {
        await sendConfirmation(orderId);
      });
    }

    return { orderId, status: 'complete' };
  }
}

Wrangler コマンド

ワークフローインスタンスの一覧表示
# List all instances of a workflow
npx wrangler workflows instances list my-workflow

# Filter by status
npx wrangler workflows instances list my-workflow --status running
npx wrangler workflows instances list my-workflow --status complete
npx wrangler workflows instances list my-workflow --status errored
インスタンスの詳細表示
# Get detailed info about specific instance
npx wrangler workflows instances describe my-workflow <instance-id>

# Output shows:
# - Current status (running/complete/errored)
# - Each step with start/end times
# - Step outputs
# - Retry history
# - Any errors
# - Sleep state (if sleeping)
ワークフローをトリガーする(開発環境)
# Deploy workflow
npx wrangler deploy

# Trigger via HTTP (if Worker is set up to trigger)
curl https://my-workflow.<subdomain>.workers.dev/

状態の永続化

永続化できるもの

ワークフローは、step.do() から返された状態を自動的に永続化します。

✅ シリアル化可能な型:

  • プリミティブ:string、number、boolean、null
  • 配列:[1, 2, 3]、['a', 'b', 'c']
  • オブジェクト:{ key: 'value' }、{ nested: { data: true } }
  • ネストされた構造:{ users: [{ id: 1, name: 'Alice' }] }

❌ シリアル化できない型:

  • 関数:() => {}
  • シンボル:Symbol('key')
  • 循環参照:const obj = {}; obj.self = obj;
  • undefined(代わりに null を使用します)
  • クラスインスタンス(プレーンオブジェクトにシリアル化します)

例 - 正しいシリアル化:

// ✅ Good - all values serializable
const result = await step.do('fetch data', async () => {
  return {
    users: [
      { id: 1, name: 'Alice', active: true },
      { id: 2, name: 'Bob', active: false }
    ],
    timestamp: Date.now(),
    metadata: null
  };
});

// ❌ Bad - contains function
const bad = await step.do('bad example', async () => {
  return {
    data: [1, 2, 3],
    transform: (x) => x * 2  // ❌ Function not serializable
  };
});
// This will throw an error!
ステップ間で状態にアクセスする
export class MyWorkflow extends WorkflowEntrypoint<Env, Params> {
  async run(event: WorkflowEvent<Params>, step: WorkflowStep) {
    // Step 1: Get data
    const userData = await step.do('fetch user', async () => {
      return { id: 123, email: '[email protected]' };
    });

    // Step 2: Use data from step 1
    const orderData = await step.do('create order', async () => {
      return {
        userId: userData.id,      // ✅ Access previous step's data
        userEmail: userData.email,
        orderId: 'ORD-456'
      };
    });

    // Step 3: Use data from step 1 and 2
    await step.do('send confirmation', async () => {
      await sendEmail({
        to: userData.email,       // ✅ Still accessible
        subject: `Order ${orderData.orderId} confirmed`
      });
    });
  }
}

オブザーバビリティ

組み込みメトリクス

ワークフローでは、以下が自動的に記録されます。

  • インスタンスのステータス:キュー、実行中、完了、エラー、一時停止
  • ステップの実行:開始/終了時刻、実行時間、成功/失敗
  • 再試行の履歴:試行回数、エラー、遅延
  • スリープ状態:ワークフローが再開する時刻
  • 出力:ステップおよび run() からの戻り値
ダッシュボードでメトリクスを表示する

Cloudflare ダッシュボードからアクセスします。

  1. Workers & Pages
  2. ワークフローを選択します
  3. インスタンスとメトリクスを表示します

メトリクスには以下が含まれます:

  • 作成されたインスタンスの合計数
  • 成功率/エラー率
  • 平均実行時間
  • ステップ単位のパフォーマンス
プログラムからのアクセス
// Get instance status
const instance = await env.MY_WORKFLOW.get(instanceId);
const status = await instance.status();

console.log(status);
// {
//   status: 'complete',
//   error: null,
//   output: { userId: '123', status: 'processed' }
// }

制限

機能制限
ワークフローの最大実行時間30日
ワークフローあたりの最大ステップ数10,000
sleep/sleepUntil の最大時間30日
ステップの最大タイムアウト15分
同時実行できるインスタンスの最大数無制限(自動スケール)
ペイロードの最大サイズ128 KB
ステップ出力の最大サイズ128 KB
waitForEvent の最大タイムアウト30日
最大再試行回数無限(設定可能)

注:

  • step.sleep() と step.sleepUntil() は、10,000 ステップの上限にカウントされません
  • ワークフローは合計で最大 30 日間実行できます
  • 各ステップの実行時間は最大 15 分に制限されます
  • 再試行は別のステップではなく、別の試行としてカウントされます

料金

Workers 有料プランが必要です(月額 $5)

ワークフローの実行:

  • 月間最初の 10,000,000 ステップ実行:無料
  • 以降:100 万ステップ実行あたり $0.30

ステップ実行としてカウントされるもの:

  • step.do() の呼び出しごと
  • ステップの再試行ごと
  • step.sleep()、step.sleepUntil()、step.waitForEvent() はカウントされません

料金の例:

  • 5 ステップのワークフローで再試行なし:5 ステップ実行
  • 3 ステップのワークフローで、1 つのステップを 2 回再試行:5 ステップ実行(3 + 2)
  • 月間 1,000 万件のシンプルなワークフロー(各 5 ステップ):((50M - 10M) / 1M) × $0.30 = 月額 $12

必ず実施すること ✅

  1. ステップ名を具体的にする - 「step 1」ではなく「ユーザーデータを取得」
  2. シリアル化可能な値のみを返す - プリミティブ、配列、プレーンオブジェクト
  3. 終了エラーには NonRetryableError を使用する - 認証エラー、無効な入力
  4. 再試行回数を設定する - 必要な場合を除き、無限の再試行を避ける
  5. 任意のステップではエラーを捕捉する - ステップが問題なく失敗を処理できる場合は try-catch を使用する
  6. 再試行には指数バックオフを使用する - ほとんどの場合はデフォルトのバックオフを使用する
  7. 入力を早い段階で検証する - 無効な場合は NonRetryableError で速やかに失敗させる
  8. ワークフローインスタンス ID を保存する - 後でステータスを照会できるよう、DB/KV に保存する
  9. 人間が介在する処理には waitForEvent を使用する - 承認、外部からの確認
  10. ワークフローのメトリクスを監視する - 成功率とエラーを追跡する

禁止事項 ❌

  1. ステップから関数を返さないでください - シリアライズエラーが発生します
  2. 循環参照を作成しないでください - シリアライズに失敗します
  3. ステップがすぐに実行されると想定しないでください - 再試行されたり、一時停止したりする場合があります
  4. ブロッキング操作を使用しないでください - 非同期処理には step.do() を使用してください
  5. ペイロード/出力のサイズを 128 KB を超えないでください - 失敗します
  6. 冪等でない操作を無限に再試行しないでください - 再試行の上限を設定してください
  7. シリアライズエラーを無視しないでください - データ構造を修正してください
  8. リアルタイム操作にワークフローを使用しないでください - 代わりに Durable Objects を使用してください
  9. 重要なステップのエラー処理を省略しないでください - try-catch で囲むか、NonRetryableError を使用してください
  10. 再試行時にステップの順序が保証されると想定しないでください - 各ステップは独立しています

トラブルシューティング

問題:「Cannot perform I/O on behalf of a different request」

原因: あるリクエストコンテキストで作成した I/O オブジェクトを、別のリクエストハンドラーから使用しようとしています

解決策: I/O は必ず step.do() のコールバック内で実行してください

// ❌ Bad - I/O outside step
const response = await fetch('https://api.example.com/data');
const data = await response.json();

await step.do('use data', async () => {
  // Using data from outside step's I/O context
  return data;  // This will fail!
});

// ✅ Good - I/O inside step
const data = await step.do('fetch data', async () => {
  const response = await fetch('https://api.example.com/data');
  return await response.json();  // ✅ Correct
});

問題:NonRetryableError の動作が開発環境と本番環境で異なる

既知の問題: 開発モードで空のメッセージを指定して NonRetryableError をスローすると再試行されますが、本番環境では正しく動作します

回避策: NonRetryableError には必ずメッセージを指定してください

// ❌ May retry in dev
throw new NonRetryableError();

// ✅ Works consistently
throw new NonRetryableError('User not found');

出典: workers-sdk#10113


問題:「The requested module 'cloudflare:workers' does not provide an export named 'WorkflowEvent'」

原因: インポートが正しくないか、@cloudflare/workers-types が古くなっています

解決策:

# Update types
npm install -D @cloudflare/workers-types@latest

# Ensure correct import
import { WorkflowEntrypoint, WorkflowStep, WorkflowEvent } from 'cloudflare:workers';
import { NonRetryableError } from 'cloudflare:workflows';

問題:ステップが期待した値ではなく undefined を返す

原因: ステップのコールバックが値を返していません

解決策: ステップのコールバックでは必ず値を返してください

// ❌ Bad - no return
const result = await step.do('get data', async () => {
  const data = await fetchData();
  // Missing return!
});
console.log(result);  // undefined

// ✅ Good - explicit return
const result = await step.do('get data', async () => {
  const data = await fetchData();
  return data;  // ✅
});

問題:ワークフローインスタンスが「running」状態のまま停止する

考えられる原因:

  1. ステップが長時間スリープしている
  2. ステップが到着しないイベントを待っている
  3. ステップが長いバックオフ間隔で再試行している

解決策:

# Check instance details
npx wrangler workflows instances describe my-workflow <instance-id>

# Look for:
# - Sleep state (will show wake time)
# - Waiting for event (will show event type and timeout)
# - Retry history (will show attempts and delays)

本番環境へのデプロイ前チェックリスト

ワークフローを本番環境にデプロイする前に、以下を確認してください。

  • すべてのステップに内容が分かる名前が付いている
  • すべてのステップに再試行の上限が設定されている
  • 終端エラーに NonRetryableError を使用している
  • 重要なステップにエラー処理がある
  • 任意のステップが try-catch で囲まれている
  • シリアライズできない値を返していない
  • ペイロードのサイズが 128 KB 未満である
  • ワークフローの実行時間が 30 日未満である
  • ステータス照会用にインスタンス ID を保存している
  • モニタリングとアラートが設定されている
  • waitForEvent のタイムアウトが設定されている
  • 開発環境でテストしている
  • 再試行の動作をテストしている
  • エラーの発生ケースをテストしている

関連ドキュメント


最終更新日:2025-10-22 バージョン:1.0.0 メンテナー:Jeremy Dawes | [email protected]

コメント