Tsumugi

外観

Flow ​

複数のジョブを依存関係付きで実行する場合はFlowを定義します。
Flowは有向非巡回グラフ(DAG)の定義であり、実行単位はRunです。

定義 ​

createFlowにperformersを渡すと、Flowを定義する関数が返されます。
binding名とpayloadの型は単発のジョブと同じように検査されます。

ts
import { createFlow, defineTsumugi } from 'tsumugi';

const performers = { ListNames, Greet, Report };
const flow = createFlow(performers);

const flows = {
  GREETINGS: flow<{ prefix: string }>((f) => {
    const list = f.node('list', 'ListNames', {
      input: (i) => ({ prefix: i.prefix }),
    });

    const each = f.fanOut('greet', 'Greet', {
      after: { list },
      over: (_i, d) => d.list.names,
      input: (name) => ({ name }),
    });

    f.node('report', 'Report', {
      after: { each },
      input: (_i, d) => ({ total: d.each.total, failed: d.each.failed }),
    });
  }),
};

const tsumugi = defineTsumugi({ performers, flows, auth: /* ... */ });

binding名はexportした名前で解決されるため、performersのキーはexport名と一致させる必要があります。

f.nodeの第1引数はノードID、第2引数はbinding、inputは前段の戻り値からpayloadを組み立てる関数です。
afterに渡したオブジェクトのキーが、inputの第2引数のプロパティ名になります。

ノードIDに使用できる文字は英数字/ハイフン/アンダースコアに限定されます。

ノードにはmaxAttempts、backoff、timeoutMs、priority、concurrencyKeyなど、投入時と同じオプションの指定が可能です。

開始 ​

ts
const runId = await tsumugi.start(env, 'GREETINGS', { prefix: 'hello' });

inputの型はFlowの型引数から決定されます。

startの第4引数に{ id }を指定するとrunIdが<flow>:<id>に固定され、同じIDでの2回目の開始は既存のrunIdを返します。
リトライを行うHTTPハンドラから呼び出してもRunが重複することはありません。

同じく第4引数の{ deadlineMs }でRun全体の期限を指定できます。

fan-out ​

実行時に件数が決まる並列処理はf.fanOutで定義します。
overが返した配列の要素ごとに子ノードが1つ作成され、子のノードIDには要素の添字が使われます(greet:0, greet:1)。
keyを指定した場合、その戻り値が使用されます。使用できる文字はノードIDと同じです。

後続のノードが受け取るのは集計値です。

ts
{ total: 3, succeeded: 2, failed: 1 }

子ノードごとの戻り値は後続のノードへ渡されません。個別の結果が必要な場合は、performerからR2やD1へ書き込み、参照を戻り値としてください。

子ノードの失敗は親ノードの失敗として扱われません。後続のノードはfailedの値で判断されます。

perform内からの追加 ​

ctx.spawnは、実行中のノードの下に子ノードを追加します。

ts
class Crawl extends Performer<{ url: string }, void, {}, Env> {
  async perform(payload: { url: string }, ctx: JobContext): Promise<void> {
    for (const found of await this.discover(payload.url)) {
      ctx.spawn(found.id, 'CRAWL', { url: found.url });
    }
  }
}

第1引数のIDは必須です。同じIDで二度要求しても子ノードは1つだけ作成されます。
使用できる文字はノードIDと同じで、子のノードIDは<親のノードID>:<指定したID>になります。

spawnには型検査が適用されません。binding名もpayloadも実行時の値として渡します。

performが失敗した場合、その試行で要求した子ノードは作成されません。再実行時に改めて要求してください。
service binding越しのリモートperformerからもspawnを呼び出せますが、awaitが必要です。

別のFlowの起動 ​

f.subflowは、別のFlowをRunとして起動し、その終端を待機します。

ts
const REPORTING = flow<{ ids: string[] }>((f) => {
  // ...
});

const PIPELINE = flow<{ prefix: string }>((f) => {
  const list = f.node('list', 'LIST', { input: (i) => ({ prefix: i.prefix }) });
  const reported = f.subflow('report', REPORTING, { input: (_i, d) => ({ ids: d.list.ids }) });
  f.node('notify', 'NOTIFY', { after: { reported }, input: () => ({}) });
});

第2引数にはFlowの定義そのものを渡します。inputの型は渡したFlowの型引数から決定されます。
起動先はflowsに登録されている必要があります。登録されていない場合は起動時にエラーがthrowされます。

子のrunIdは<子のFlow名>:<親のrunIdのローカル部>-<ノードID>です。

子の状態がそのままノードの状態になり、COMPLETED、FAILED、CANCELLEDのいずれかです。
子の戻り値は受け取りません。結果が必要な場合は、performerからR2やD1へ書き込んでください。

この状態は親に依存しており、親を取り消すことで子も取り消されます。

入れ子は既定で3段までに制限されています。これは、defineTsumugiのruns.maxDepthで任意値へ変更可能です。

待ち合わせ ​

ノードは、自身と子孫のすべてが終わった時点で完了として扱われます。
afterで親ノードを指定した後続のノードは、fan-outで展開された子ノードとspawnで追加された子孫の完了も待機します。

発火条件 ​

既定では、依存がすべて成功した場合にだけノードが実行されます。
triggerを指定すると、依存が失敗した場合の動作を変更できます。

値実行される条件
'success'依存がすべて成功。既定値です
'failure'依存のうち1つ以上がFAILED STALLED CANCELLEDのいずれか
'always'依存がすべて終了。成否は問いません
ts
const flows = {
  BACKUP: flow<{ target: string }>((f) => {
    const dump = f.node('dump', 'DUMP', { input: (i) => ({ target: i.target }) });

    // 失敗したときだけ実行する後始末
    f.node('alert', 'ALERT', {
      after: { dump },
      trigger: 'failure',
      input: (i) => ({ message: `backup failed: ${i.target}` }),
    });

    // 成否に関わらず必ず実行する
    f.node('unlock', 'UNLOCK', { after: { dump }, trigger: 'always', input: (i) => ({ target: i.target }) });
  }),
};

triggerは依存を持つノードにのみ指定可能です。afterが無いノードへの指定はエラーになります。

'failure'が数えるのは実際に失敗したノードに限定されます。
発火条件やwhenで実行されずSKIPPEDになった依存は失敗として数えないため、後始末のノードも実行されません。

'failure'と'always'では、失敗した依存に戻り値がありません。
そのため受け取り口の型がundefinedを含むようになり、値の欠落を扱う必要があります。

ts
f.node('alert', 'ALERT', {
  after: { dump },
  trigger: 'always',
  // d.dumpはundefinedの可能性がある
  input: (_i, d) => ({ size: d.dump?.size ?? 0 }),
});

条件分岐 ​

入力や前段の結果で経路を選ぶ場合はwhenを指定します。
falseを返したノードはSKIPPEDになり、それを待機する下流のノードも実行されません。

ts
f.node('detail', 'DETAIL', {
  after: { list },
  when: (i, d) => i.verbose && d.list.items.length > 0,
  input: (_i, d) => ({ items: d.list.items }),
});

whenはノードを起動する直前に評価されます。
inputと同じくFlowの定義から呼び出されるため、外部への問い合わせを行わず、渡された値だけで判断してください。

whenで例外が発生した場合、実行の可否が決まらないためノードはFAILEDになります。

失敗時の動作 ​

ノードが失敗した場合、それをsuccess(既定)で待つ下流のノードがSKIPPEDになります。
triggerに'failure'か'always'を指定した下流は実行されます。
依存関係のないノードは最後まで実行され、すべてのノードが終わった時点でRunがFAILEDになります。

RunがFAILEDになるのは、失敗したノードがある場合に限り、発火条件やwhenで実行されなかっただけのノードは失敗としてカウントされません。
trigger: 'failure'の後始末が成功しても、上流が失敗していればRunはFAILEDのまま終了されます。

SKIPPEDになった理由はノードのerrorに残ります。ダッシュボードの詳細から確認できます。

FAILEDのRunはダッシュボードとREST APIから再開可能です。
成功済みのノードの結果はそのまま使われ、それ以外のノードは未実行の状態に戻って改めて実行されます。

取り消し ​

cancelは未実行のノードを停止します。
実行中のジョブは停止されないため、完了/失敗時点でRunがCANCELLEDになります。

期限 ​

Run全体の期限は、Flowの定義の第2引数で指定します。

ts
const GREETINGS = flow<{ prefix: string }>(
  (f) => {
    // ...
  },
  { deadlineMs: 10 * 60 * 1000 },
);

deadlineMsは正の整数(ミリ秒)を許容します。
startの第4引数の{ deadlineMs }を指定した場合はそちらが優先されます。

期限を超過したRunは取り消しと同じ手順で中断されます。
未実行のノードはrun deadline exceededのエラーでFAILEDになり、実行中のジョブは停止しないため、完了/失敗時点でRunがFAILEDになります。
また、実行中の子のRunは取り消されます。

期限の時点で終わっていなかったRunは、残りのノードがすべて成功している場合であってもFAILED扱いとなります。

FAILEDになったRunは通常の失敗と同じように再開可能であり、期限は再開の時点から再計算されます。

保持期間 ​

終了したRunはruns.sweepAfterMs(既定5分)、FAILEDのRunはruns.failedRetentionMs(既定7日)の経過後に削除されます。
詳細はRunSettingsを参照してください。

テスト ​

simulateFlowは、ある入力に対してノードがどの順序で実行され、各ノードにどのpayloadが渡るかを検証するための関数です。performerは実行されません。

ts
import { simulateFlow } from 'tsumugi/testing';

const result = simulateFlow(flows.GREETINGS, { prefix: 'hello' }, { results: { list: { names: ['a', 'b'] } } });

expect(result.nodes.map((node) => node.id)).toEqual(['list', 'greet:0', 'greet:1', 'greet', 'report']);
expect(result.nodes[1].payload).toEqual({ name: 'a' });

各ノードの戻り値はresultsにノードIDとの対応で指定し、関数も指定可能です。
指定の無いノードの戻り値はundefinedとなります。

failsに渡したノードは失敗します。下流はSKIPPED、RunはFAILEDになります。

fan-outはoverの結果に従って展開されます。ctx.spawnによる追加は含まれません。

制約 ​

  • ノードはuniqueKeyを受け付けません。uniqueKeyを必須と宣言したperformerをノードに指定すると型エラーになります
  • ノードの戻り値が8,192文字を超えた場合と直列化できない場合、その試行は失敗として扱われ、maxAttemptsまで再試行されます。大きい結果はR2等へ書き込み、参照を戻り値としてください
  • 1つのRunに含められるノード数は既定で10,000件です。defineTsumugiのruns.maxNodesで変更可能です
  • subflowの入れ子は既定で3段までです。defineTsumugiのruns.maxDepthで変更可能です

設定 ​

flowsを指定する場合、wranglerの設定に2箇所追記します。

jsonc
{
  "durable_objects": {
    "bindings": [
      { "name": "JOB_SHARD", "class_name": "TsumugiJobShard" },
      { "name": "RUN", "class_name": "TsumugiRun" },
    ],
  },
  "migrations": [
    { "tag": "v1", "new_sqlite_classes": ["TsumugiJobShard"] },
    { "tag": "v2", "new_sqlite_classes": ["TsumugiRun"] },
  ],
}

TsumugiRunはdefineTsumugiの戻り値から取り出してエクスポートします。

ts
export { TsumugiJobShard } from 'tsumugi';
export class TsumugiRun extends tsumugi.runClass {}

D1のマイグレーションも適用し直してください。

bash
pnpm wrangler d1 migrations apply my-jobs --remote

flowsを指定しない構成では、どちらも不要です。

デプロイと実行中のRun ​

実行中のRunは開始時の構造のまま進みます。
inputなどの関数を修正した場合、次に実行されるノードから反映されます。

定義からノードを削除すると、実行中のRunはそのノードで失敗します。
構造を変更する場合は、実行中のRunがすべて終わってから削除してください。

Copyright © 2026 chan-mai All Rights Reserved.