ステートフル ワークフロー向け Durable Functions

Durable Functions を使う理由

3 種類の関数

// Client function (HTTP trigger) — starts the orchestration
module.exports = async function (context, req) {
  const client = df.getClient(context);
  const orderId = req.body.orderId;
  const instanceId = await client.startNew('OrderOrchestrator', undefined, { orderId });
  return client.createCheckStatusResponse(context.bindingData.req, instanceId);
};

チェイニング パターン

// Orchestrator: chaining pattern
const df = require('durable-functions');
module.exports = df.orchestrator(function* (context) {
  const orderId = context.df.getInput().orderId;

  const validated = yield context.df.callActivity('ValidateOrder', orderId);
  const charged   = yield context.df.callActivity('ChargePayment', validated);
  const shipped   = yield context.df.callActivity('ShipOrder', charged);

  return { status: 'shipped', trackingId: shipped.trackingId };
});

ファンアウト／ファンイン パターン

// Orchestrator: fan-out / fan-in
module.exports = df.orchestrator(function* (context) {
  const items = context.df.getInput().items;

  // Fan-out: start all tasks in parallel
  const tasks = items.map(item => context.df.callActivity('ProcessItem', item));

  // Fan-in: wait for all tasks to complete
  const results = yield context.df.Task.all(tasks);

  return results;
});

モニター パターン

// Orchestrator: monitor pattern (poll until job completes)
module.exports = df.orchestrator(function* (context) {
  const jobId = context.df.getInput().jobId;
  const expiry = new Date(context.df.currentUtcDateTime);
  expiry.setHours(expiry.getHours() + 24); // 24-hour timeout

  while (context.df.currentUtcDateTime < expiry) {
    const status = yield context.df.callActivity('GetJobStatus', jobId);
    if (status === 'completed') return { jobId, status };
    if (status === 'failed') throw new Error('Job failed');
    // Wait 30 seconds before next poll
    const nextCheck = new Date(context.df.currentUtcDateTime);
    nextCheck.setSeconds(nextCheck.getSeconds() + 30);
    yield context.df.createTimer(nextCheck);
  }
  throw new Error('Workflow timed out');
});

人間による操作パターン

// Orchestrator: wait for human approval with timeout
module.exports = df.orchestrator(function* (context) {
  const request = context.df.getInput();

  yield context.df.callActivity('SendApprovalEmail', request);

  const timeout = df.Task.createTimer(context, new Date(Date.now() + 48 * 3600 * 1000));
  const approval = context.df.waitForExternalEvent('ApprovalResponse');

  const winner = yield context.df.Task.any([approval, timeout]);

  if (winner === approval) {
    const approved = winner.result;
    return approved ? 'Approved' : 'Rejected';
  } else {
    return 'Timed out — escalated';
  }
});

オーケストレーターの制約

// WRONG — non-deterministic, will cause replay bugs
module.exports = df.orchestrator(function* (context) {
  const now = new Date();          // Don't use Date()
  const rand = Math.random();       // Don't use Math.random()
  const data = await fetch('/api'); // Don't make HTTP calls directly
});

// CORRECT
module.exports = df.orchestrator(function* (context) {
  const now = context.df.currentUtcDateTime; // OK
  const data = yield context.df.callActivity('FetchData', null); // OK
});

インスタンスの管理：状態の確認と終了

// Client function: check orchestration status
module.exports = async function (context, req) {
  const client = df.getClient(context);
  const instanceId = req.params.instanceId;

  const status = await client.getStatus(instanceId, true, true, true);
  return {
    status: 200,
    body: {
      instanceId,
      runtimeStatus: status.runtimeStatus,
      customStatus: status.customStatus,
      output: status.output
    }
  };
};

ストレージ バックエンドとパフォーマンス

// host.json: configure the Durable Task storage provider
{
  'version': '2.0',
  'extensions': {
    'durableTask': {
      'hubName': 'MyTaskHub',
      'storageProvider': {
        'type': 'azure',
        'connectionStringName': 'AzureWebJobsStorage',
        'controlQueueBatchSize': 32,
        'maxQueuePollingInterval': '00:00:02'
      }
    }
  }
}

エラー処理と再試行

// Orchestrator: retry an activity with exponential backoff
module.exports = df.orchestrator(function* (context) {
  const retryOptions = new df.RetryOptions(
    5000,  // firstRetryIntervalInMilliseconds
    3      // maxNumberOfAttempts
  );
  retryOptions.backoffCoefficient = 2; // 5s, 10s, 20s

  try {
    const result = yield context.df.callActivityWithRetry(
      'CallExternalAPI',
      retryOptions,
      context.df.getInput()
    );
    return result;
  } catch (e) {
    yield context.df.callActivity('SendFailureAlert', e.message);
    throw e;
  }
});

Durable Entities

// Counter entity function
const df = require('durable-functions');
module.exports = df.entity(function (context) {
  let count = context.df.getState(() => 0);
  const operation = context.df.operationName;

  if (operation === 'add') count += context.df.getInput();
  if (operation === 'reset') count = 0;
  if (operation === 'get') context.df.return(count);

  context.df.setState(count);
});

// From orchestrator, increment counter entity
// const entityId = new df.EntityId('Counter', 'myCounter');
// yield context.df.callEntity(entityId, 'add', 1);

クイックチェック

レッスンのまとめ