workflow

@falcondev-oss/workflow

メールを送信し、3 日間待機してからフォローアップを送信するような、長時間実行される非同期プロセスは、それを実行するプロセスとともに終了してしまいます。このライブラリを使えば、プロセスの終了後も処理を継続できます。ワークフローの本体は通常の非同期関数ですが、実行する step はすべて Redis に永続化されるため、クラッシュ、デプロイ、またはリトライが発生しても、関数を先頭から再実行し、完了済みのステップはそのままスキップします。

3 つのレイヤーがあり、それぞれひとつ上のレイヤーによって作成されます。

レイヤー	管理するもの
WorkflowNamespace	Redis 接続、Pub/Sub 接続、ワークフローをまたぐ同時実行数の上限
WorkflowDefinition	キューひとつとジョブタイプひとつ：入力スキーマとオプション。ジョブを登録、スケジュール、検査します
Workflow	定義とその run ハンドラー。work() も実行できます
WorkflowJob	登録された実行ひとつ：id、groupId、wait()

すべてのエクスポートはパッケージのルートから行います：WorkflowNamespace、createRedis、expBackoff、TimeoutError、ResultExpiredError、JobAlreadyExistsError、NonRecoverableError。ワークフローは必ず名前空間を通じて作成してください（createWorkflow ／ defineWorkflow）。new Workflow() は使用しないでください。

セットアップ

import { createRedis, WorkflowNamespace } from '@falcondev-oss/workflow'

const namespace = new WorkflowNamespace({
  id: 'app',
  redis: await createRedis({ url: process.env.REDIS_URL }),
  logger: console, // opt-in; no logging by default
})

名前空間はすべての設定を管理します：redis、logger（info ／ success ／ error ／ debug ／ warn。すべて省略可能）、およびワークフローが継承するオプションのセットです。redis が指定されていない場合、プロセス全体で共有される localhost:6379 への接続にフォールバックします。createRedis はライブラリに必要な接続オプション（maxRetriesPerRequest: null、オフラインキューの無効化、上限付きのリトライ戦略）を適用するため、new IORedis() ではなくこちらを使って接続を作成してください。

名前空間は独自の終了フックを登録するため、SIGTERM を受け取るとワーカーをドレインして切断します。テストでは autoClose: false を渡し、namespace.close() を自分で呼び出してください。

ワークフローの定義

const onboarding = namespace.createWorkflow({
  id: 'onboarding',
  schema: z.object({ userId: z.string(), name: z.string() }),
  getGroupId: (input) => input.userId,
  async run({ input, step }) {
    await step.do('welcome', () => sendWelcome(input.userId))
    await step.wait('settle', 3 * 24 * 60 * 60 * 1000)
    const engaged = await step.do('check engagement', () => isEngaged(input.userId))
    return { engaged }
  },
})

• id は、このワークフローに関するすべての Redis キーのスコープになります。変更すると、実行中のジョブが孤立します。
• schema は任意の Standard Schema（zod、arktype など）を指定でき、ジョブの登録時とワーカーでの実行時に検証されます。検証に失敗すると Invalid workflow input がスローされます。省略した場合、input の型は undefined になります。
• getGroupId は検証済みの入力からグループを導出します。実行呼び出しで明示的に指定した groupId が優先されます。
• run は { input, step, span } を受け取り、解決された値がジョブの出力になります。ペイロード、ステップの結果、出力はすべて superjson を通じて往復するため、Date、Map、Set、BigInt はそのまま保持されます。
• queueOptions ／ workerOptions ／ jobOptions はワークフローごとのデフォルトを設定します。名前空間の同じキーは、その下位の値としてシャロー結合されます。詳しくは reference/options.md を参照してください。

定義とハンドラー

createWorkflow はハンドラーをワークフローのモジュールに配置するため、これをエンキューするすべてのモジュールで、ハンドラーがインポートするものもすべて読み込まれます。エンキューするコードを軽量に保つ必要がある場合（クライアントモジュール、インポートの循環、遅いテスト起動など）は、分割します。

// jobs/onboarding.ts: imported wherever jobs are enqueued
export const onboarding = namespace
  .defineWorkflow({ id: 'onboarding', schema: z.object({ userId: z.string() }) })
  .returns<{ engaged: boolean }>()

// handlers/onboarding.ts: imported only by the worker process
export default onboarding.handle(async ({ input, step }) => ({ engaged: await check(input.userId) }))

// worker.ts
import onboarding from './handlers/onboarding'
await onboarding.work()

• defineWorkflow は、run を除くすべての createWorkflow オプションを受け取ります。この定義には、work() を除くすべてのワークフローメソッドがあります。
• returns<T>() は出力型を宣言するため、エンキューするコードで job.wait() の型が付きます。指定しない場合、定義の出力型は unknown です。
• handle(run) は完全な Workflow を返します。ハンドラーは宣言された型を返す必要があります。returns<T>() を指定しない場合、出力型はハンドラーから推論されます。
• すでにハンドラーがあるワークフローでの handle() と returns() は型エラーになります。ライブラリがハンドラーを import() することはないため、ワーカーのエントリーポイントでは、処理するすべてのハンドラーモジュールをインポートする必要があります。

await onboarding.work() // start a worker in this process

const job = await onboarding.run({ userId: 'u1', name: 'John' })
const { engaged } = await job.wait() // typed as the run() return

run(input, opts?) はすぐにエンキューし、runIn(input, delayMs, opts?) は遅延後に、runAt(input, date, opts?) は指定したタイムスタンプにエンキューします。3つとも、ジョブの実行を待つことなく、エンキューされるとすぐに WorkflowJob を返します。opts では groupId、priority、maxAttempts、jobId を指定できます。

job.wait(timeoutMs?) は pub/sub 通知を待ち、デシリアライズされた出力を返します。プロセス内にワーカーがいない純粋なプロデューサーからも使用できます。次の場合に拒否されます。

エラー	発生条件
ワークフロー独自の Error	run が例外をスローし、試行回数を使い果たした場合 — 元のメッセージが保持されます
TimeoutError	timeoutMs が先に経過した場合（デフォルト値はなく、指定しないと無期限に待機します）
ResultExpiredError	ジョブは完了したものの、結果レコードが resultTtl の期間（デフォルトは 300 s）を過ぎて期限切れになった場合

すでに使用されている jobId を明示的に渡すと、JobAlreadyExistsError で拒否されます。これは「最大 1 回だけエンキューする」ための重複排除ハンドルです。

ステップとリプレイ

step.do(name, run) は耐久性の単位です。完了時に結果をメモ化します。いったん戻ると、その結果はジョブのステップハッシュに書き込まれ、それ以降の試行では本文を再実行せず、キャッシュされた値を返します。そのため、再試行では完了済みのステップが即座にリプレイされ、失敗したステップだけが再実行されます。

const user = await step.do('load user', () => db.user.find(input.userId))
await step.do('charge', async ({ step, span }) => {
  span.setAttribute('user.plan', user.plan)
  await step.do('authorize', () => gateway.authorize(user)) // nested → 'charge|authorize'
  await step.do('capture', () => gateway.capture(user))
})

コールバックは { step, span } を受け取ります。ここには、名前に親の名前が付加される（parent|child）ネストされた step と、そのステップの OTel スパンが含まれます。ステップ名はメモ化キーです。ジョブ内で安定しており、一意になるようにしてください。また、ステップ名の変更はキャッシュ結果を無効にするものとして扱ってください。

step.wait(name, durationMs) は耐久性のあるスリープです。最初に実行したときに startedAt を保存するため、リプレイ時には期間全体ではなく、残りの時間だけスリープします。step.waitUntil(name, date) は絶対時刻を基準に同じ動作をします。

覚えておくべき重要な特性が 3 つあります。

• ステップ単位の再試行はありません。 step.do はオプションを受け取らず、例外がスローされるとジョブ全体が失敗します。再試行はジョブ単位（maxAttempts）です。
• キャンセルは協調的に行われます。 ワーカーが自身のロックを失うと（ロックの奪取、ハートビートの喪失など）、ジョブのシグナルを中止します。step.wait は直ちに拒否され、step.do は新しい処理の開始を拒否しますが、すでに実行中のステップは中断されません。CPU 負荷の高いステップは、処理が戻るまでスロットを占有します。
• ステップに対する Promise.all は、ほかの処理をキャンセルしません。 並列ステップの 1 つで例外がスローされても、ほかのステップは完了まで実行されます。ジョブの失敗は、それらの完了を待ちます（その際、警告がログに記録されます）。失敗後も実行を続けてはならないステップは、順番に実行してください。

リプレイでは、キャッシュされたステップを囲む関数本体が再実行されるため、ステップの外側はすべて純粋な処理にしてください。ステップ間の副作用は、試行のたびに実行されます。

順序制御、グループ化、並行処理

グループは、直列で処理されるレーンです。同じgroupIdを持つジョブはFIFO順で1つずつ実行され、ジョブが完了するまでそのグループをブロックします。groupId（またはgetGroupId）がない場合、各ジョブにはランダムなUUIDグループが割り当てられ、完全に並列で実行されます。エンティティごとに処理を直列化するには、userIdやtenantIdなどのエンティティIDをグループとして使用します。

priorityは0…2²¹-1の単純な数値で、値が大きいほど優先され、デフォルトは0です。'high' | 'normal'という列挙型はありません。

独立した上限は4つあり、いずれも任意です。

上限	設定箇所	デフォルト
1つの名前空間内のすべてのワークフロー全体	WorkflowNamespace({ concurrency })	無制限
1つのワークフローのすべてのワーカー全体	queueOptions.concurrency	無制限
1つのグループ内	queueOptions.groupConcurrency	1
1つのワーカープロセス内	workerOptions.concurrency / work({ concurrency })	1

ワーカーの同時実行数のデフォルトは1で、これがよくある意外な点です。work()だけを呼び出した場合、プロセスごとに一度に1つのジョブを処理します。

リトライ

maxAttemptsのデフォルトは1で、リトライは行われません。これはエンキュー時にジョブごとに適用されるため、workerOptionsに設定した値が有効になるのは、同じオプションを共有するプロデューサーによってエンキューされたジョブだけです。プロデューサーとワーカーが別々のプロセスの場合は、jobOptions.maxAttemptsまたは実行ごとのopts.maxAttemptsを使用してください。

バックオフはワーカー側でワークフローごとに設定します。workerOptions.backoff(attempt)は、指定された試行（1から始まります）までの待機時間をミリ秒で返します。デフォルトはexpBackoff()で、フルジッターを用いた指数バックオフです。baseは500 ms、factorは2、capは30 sです。テスト時の標準的な上書き方法はbackoff: () => 0です。

Cronスケジュール

await onboarding.upsertSchedule('nightly-digest', {
  pattern: '0 3 * * *',
  input: { userId: 'system', name: 'digest' },
  tz: 'Europe/Berlin',
})

upsertSchedule(scheduleId, opts)はべき等です。同じIDを再登録すると既存のスケジュールがその場で置き換わるため、起動のたびに安全に呼び出せます。inputは登録時にスキーマに照らして検証されるため、不正なペイロードは午前3時ではなく登録時にエラーになります。オプションは、pattern（Croner構文。cronパターンのみで、間隔形式はありません）、input、tz（IANA。デフォルトは登録プロセスのローカルタイムゾーン）、priority、groupId（デフォルトは scheduleIdで、各実行を直列化します）、skipIfRunning（デフォルトはtrue。時間のかかる実行中に次の実行時刻が来ると、ジョブが積み重なる代わりに次の実行をスキップします）です。未実行のスケジュールが複数回発生していた場合は、1回の実行にまとめられます。

removeSchedule(scheduleId)とgetSchedules()も用意されています。後者は、スケジュールごとに{ scheduleId, pattern, tz, nextRun, lastFireAt, lastJobId }を返します。

可観測性

OTel SDKが登録されている場合、トレーシングは自動的に行われます。プロデューサーのスパンがコンテキストをペイロードに伝播するため、ワーカーのジョブスパンと各ステップのスパンは、ジョブをエンキューしたスパンに関連付けられます。runとstep.doのコンテキストにあるspanは、その時点で有効なスパンです。属性は直接設定してください。

キュー深度のメトリクスは、workerOptions.metrics（{ meter, prefix }。名前空間に設定すると、すべてのワークフローに適用されます）でオプトインできます。メトリクスはwork()の呼び出し時に登録されます。エクスポートされるゲージと、ワーカーのチューニング設定全体（lockMs、maxStalledCount、stalledInterval、safetyTimeout、promoteBatchSize、keepFailed、onError、onFailed）については、reference/options.mdを参照してください。