REST API

authを設定すると、Tsumugiは/api以下のエンドポイントへのリクエストを処理します。
設定していない場合、すべてのリクエストに404で応答します。
認証は/api/*に限って適用されます。
ダッシュボードのHTMLはジョブのデータを含まないので、認証を要求せずに配信されます。
ダッシュボードは、認証されていない場合にトークンの入力欄を表示します。
OpenAPIの定義は、GET /api/openapi.jsonで取得します。

共通のレスポンス

503の本文には、理由が含まれます。
未適用のマイグレーションが存在する場合は、未適用のファイル名が本文に含まれます。
bindingが不足している場合は、不足しているbindingの一覧が本文に含まれます。
一時的に利用できない場合は、時間を置いて再試行してください。

Flowを登録していない構成では、Runの開始, 再開, 取り消しのエンドポイントは501で応答します。
/api/openapi.jsonは、構成が完了していなくても応答します。
runIdの形式が不正な場合、/api/runs/:idを含むエンドポイントは400で応答します。

ジョブ

GET /api/jobs

ジョブの一覧を取得します。
レスポンスは、ジョブの配列と総件数を含みます。

クエリパラメータ

複数の条件を指定した場合は、すべての条件を満たすジョブが一覧に含まれます。
キーの検索は完全一致で、部分一致は使用できません。
sortとorderに不正な値を指定した場合は、400ではなく既定の並び順で応答します。
created_fromとcreated_toに整数以外を指定した場合は、その条件は無視されます。

{
  "jobs": [
    {
      "id": "MAIL#0:xxxxxxxxxxxxxxxxxxxxxxxx",
      "binding": "MAIL",
      "state": "FAILED",
      "priority": 0,
      "attempts": 3,
      "max_attempts": 3,
      "created_at": 1753000000000,
      "updated_at": 1753000060000,
      "dispatched_at": 1753000030000,
      "retryable": true
    }
  ],
  "total": 1
}

retryableは、保持期間に基づいて算出された目安です。
リトライの可否は、リトライ要求時に判定されます。
保持期間を過ぎたジョブのリトライには、410で応答します。

GET /api/jobs/:id

ジョブ1件の詳細を取得します。
ジョブが見つからない場合は、404で応答します。

{
  "job": {
    "id": "MAIL#0:xxxxxxxxxxxxxxxxxxxxxxxx",
    "binding": "MAIL",
    "state": "FAILED",
    "payload": "{\"to\":\"a@example.com\"}",
    "result": null,
    "priority": 0,
    "attempts": 3,
    "max_attempts": 3,
    "progress": null,
    "concurrency_key": "domain:example.com",
    "unique_key": null,
    "guarantee": "at-least-once",
    "created_at": 1753000000000,
    "updated_at": 1753000060000,
    "dispatched_at": 1753000030000,
    "run_after": null,
    "expires_at": null,
    "cancel_requested_at": null,
    "traceparent": "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01",
    "retryable": true,
    "logs": [
      {
        "attempt": 1,
        "timestamp": 1753000011000,
        "message": "Payment requested"
      }
    ],
    "attempts_log": [
      {
        "attempt": 1,
        "state": "FAILED",
        "started_at": 1753000010000,
        "finished_at": 1753000012000,
        "error": "payment failed: 502"
      }
    ]
  }
}

POST /api/jobs

ジョブを投入します。
リクエストの本文は、次の形式のJSONです。

{
  "binding": "MAIL",
  "payload": { "to": "a@example.com", "subject": "hi" },
  "maxAttempts": 5,
  "delayMs": 60000,
  "expiresInMs": 600000,
  "priority": 10,
  "concurrencyKey": "domain:example.com",
  "uniqueKey": "mail:a@example.com:hi",
  "partitionKey": "tenant-42",
  "traceparent": "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"
}

bindingとpayloadは必須で、performersに登録されたbindingを指定します。
maxAttempts, delayMs, expiresInMs, priorityは整数にしてください。
上の例にない項目(timeoutMs, backoffなど)は無視され、既定値が使われます。

expiresInMsは、ジョブの投入時刻から数えた実行期限です。
期限を過ぎたジョブは実行されず、CANCELLEDへ遷移します。
partitionKeyは、shardsを2以上に設定したbindingにおいて、投入先のシャードを決定するためのキーであり、そのbindingでは必須です。

traceparentは任意です。
本文で指定しない場合はtraceparentヘッダの値が使われ、両方を指定した場合は本文の値が優先されます。
値には、ジョブの投入で説明している、W3C Trace Contextのversion 00の書式を使用してください。
指定した値は、リトライ時にも引き継がれ、performerのctx.traceparentに渡されます。

uniqueKeyが既存のジョブと同じ場合も201で応答し、新しいジョブは作成されず、本文には既存のジョブIDが含まれます。

POST /api/jobs/:id/retry

FAILEDまたはSTALLEDのジョブをSCHEDULED状態に遷移させます。
保持期間を過ぎて削除されたジョブは、リトライできません。

409の場合は、ジョブの状態が変化してから、再び要求してください。

POST /api/jobs/:id/cancel

ジョブを取り消します。
対象は、SCHEDULED, QUEUED, RUNNINGのジョブです。

SCHEDULEDのジョブは、直ちにCANCELLEDに遷移します。
QUEUEDとRUNNINGのジョブは、既に実行されている可能性があるので、取り消しの要求として記録されます。
この場合、レスポンスにrequested: trueが含まれます。
ジョブは、performerが要求を確認して終了した場合か、要求からreaperGraceMs(既定30秒)が過ぎても結果が報告されない場合に、CANCELLEDに遷移します。
詳細は、実行中の取り消しを参照してください。

{ "ok": true, "requested": true }

取り消しを要求済みのジョブに再び要求した場合も、requested: trueで応答します。
終了したジョブへの要求には、409で応答します。
そのほかのレスポンスは、retryと同一です。

POST /api/jobs/bulk-retry

POST /api/jobs/bulk-cancel

複数のジョブをまとめてリトライまたは取り消します。
対象は、ジョブIDの配列、またはフィルタ条件のいずれかで指定します。

ジョブIDで指定する場合はidsを指定し、フィルタ条件で指定する場合はidsを省略します。

{ "ids": ["MAIL#0:xxxxxxxxxxxxxxxxxxxxxxxx", "MAIL#0:yyyyyyyyyyyyyyyyyyyyyyyy"] }

{ "binding": "MAIL", "state": "FAILED", "created_from": 1753000000000, "limit": 200 }

idsを指定した場合、他の条件は無視されます。
created_from, created_to, limitには整数を指定してください。

条件でstateを省略した場合、bulk-retryの対象はFAILEDとSTALLEDのジョブ、bulk-cancelの対象はSCHEDULED, QUEUED, RUNNINGのジョブです。
これ以外の状態を指定した場合は、400で応答します。
bulk-retryでは、failedRetentionMsを過ぎたジョブは対象から除外されます。
bulk-cancelでは、すでに取り消しを要求済みのジョブは、CANCELLEDが確定するまで対象から除外されます。
idsでジョブを直接指定した場合は、すでに要求済みのジョブに対して再度要求しても、二重に処理されることはありません。

{
  "ok": ["MAIL#0:xxxxxxxxxxxxxxxxxxxxxxxx"],
  "failed": [{ "id": "MAIL#0:yyyyyyyyyyyyyyyyyyyyyyyy", "reason": "gone" }],
  "remaining": 320
}

一部のジョブの処理に失敗した場合も、ステータスコードは200です。
bulk-cancelでは、取り消しの要求として記録されたジョブもokに含まれます。
failedのreasonは、1件ずつのリトライと取り消しのエンドポイントが応答するエラーに対応します。
invalid-stateとgoneは、それぞれ409と410に相当します。
idsにジョブIDとして不正な値が含まれている場合は、その値がinvalid-idとしてfailedに含まれます。

remainingは、条件で指定したときにlimitを超えて処理されなかった件数です。
0になるまで、同じ要求を再送してください。
idsで指定した場合は、remainingは常に0です。
要求の時点で条件に合わないジョブは、一覧に表示されていてもfailedに含まれます。

POST /api/jobs/:id/reschedule

予約したジョブの実行時刻を変更します。対象は、SCHEDULEDのジョブです。

{ "runAt": 1753003600000, "priority": 10 }

runAtとdelayMsは、いずれか一方を指定してください。
両方を指定した場合と、どちらも指定しない場合は、400で応答します。
runAt, delayMs, priorityは整数にしてください。

ジョブIDは変わりません。
uniqueKeyによる重複の判定も継続します。
ステータスコードは、retryと同一です。

集計と診断

GET /api/stats

状態別の件数を取得します。

{
  "byState": { "SCHEDULED": 12, "RUNNING": 3, "COMPLETED": 480, "FAILED": 2 },
  "oldestScheduledMs": 45000
}

oldestScheduledMsは、最も古いSCHEDULEDのジョブが投入されてからの経過時間をミリ秒で表します。
SCHEDULEDのジョブが存在しない場合、値はnullです。

GET /api/metrics

記録したメトリクスをもとに、bindingごとの失敗率と所要時間を取得します。

クエリパラメータ

{
  "hours": 24,
  "bindings": [
    {
      "binding": "MAIL",
      "total": 1200,
      "failed": 18,
      "failureRate": 0.015,
      "avgDurationMs": 820,
      "maxDurationMs": 5400,
      "p95DurationMs": 2100,
      "avgAttempts": 1.04
    }
  ],
  "series": [{ "at": 1753000000000, "total": 50, "failed": 1, "avgDurationMs": 810 }]
}

メトリクスがサンプリングされている期間も、実際の件数に近い値が含まれます。
hoursが範囲外の場合と、bindingが不正な名前の場合は、400で応答します。
metricsを設定していない構成では、501で応答します。
502は、メトリクスを取得できなかった場合の応答で、本文に理由が含まれます。

GET /api/diagnostics

bindingごとに、実行待ちのジョブが実行されない原因を取得します。実行中のジョブの件数と、現在の実行量の設定も含まれます。

{
  "bindings": {
    "MAIL": {
      "active": 2,
      "blocked": { "paused": false, "capacity": true, "tokens": false, "perKey": false, "perKeyTokens": false },
      "policy": {
        "paused": false,
        "concurrency": 100,
        "perKeyConcurrency": 1,
        "rate": null,
        "perKeyRate": null,
        "agingIntervalMs": 60000,
        "reaperGraceMs": 30000
      }
    }
  }
}

blocked

GET /api/bindings

ジョブを投入するbinding名の一覧を取得します。
一度も実行されていないbindingを含め、performersに登録したすべてのbinding名が取得対象です。

{ "bindings": ["CHARGE", "MAIL"] }

実行量の設定

POST /api/bindings/:binding/policy

実行量の設定を、実行時に変更します。
変更は、bindingsの設定より優先されます。

{ "paused": true, "concurrency": 5 }

指定した項目のみが変更され、省略した項目は維持されます。
成功時には、変更後の設定と適用されたshardの数がレスポンスに含まれます。
以下は、pausedとconcurrencyを変更した場合の例です。

{
  "binding": "MAIL",
  "shards": 1,
  "policy": {
    "paused": true,
    "concurrency": 5,
    "perKeyConcurrency": 1,
    "rate": null,
    "perKeyRate": null,
    "agingIntervalMs": 60000,
    "reaperGraceMs": 30000
  }
}

変更は、対象のbindingに属するすべてのshardに適用されます。
shardを分割している場合、ネットワークの分断や一時的な障害により、一部のshardへの変更の適用に失敗する可能性があります。
その場合は、HTTPステータスコード500で応答し、本文に失敗したshardの番号が含まれます。

{ "error": "the change reached 3 of 4 shards, send the same request again", "binding": "MAIL", "shards": 4, "failed": [2] }

同じ要求を再び送信すると、残りのshardに変更が適用されます。
何度送信しても、設定の結果は同一です。
変更後は、bindingsの設定が適用されなくなります。

POST /api/bindings/:binding/policy/reset

実行時の変更を破棄し、bindingsの設定に戻します。
次の投入から、bindingsの設定が再び適用されます。

{ "ok": true }

一部のshardへの適用に失敗した場合は、policyの変更と同様に、500で応答し、failedが本文に含まれます。

定期実行

schedulesを定義していない構成では、このセクションのエンドポイントは501で応答します。
詳細は、定期実行を参照してください。

GET /api/schedules

定期実行の一覧を取得します。
ジョブの一覧とは異なり、状態はリアルタイムで反映されます。

{
  "schedules": [
    {
      "name": "nightly",
      "kind": "flow",
      "target": "REPORT",
      "every_ms": null,
      "cron": "0 3 * * *",
      "time_zone": "Asia/Tokyo",
      "overlap": "skip",
      "paused": false,
      "next_occurrence_at": 1767722400000,
      "last_occurrence_at": 1767636000000,
      "last_fired_at": 1767636001200,
      "last_job_id": null,
      "last_run_id": "REPORT:nightly-1767636000000",
      "last_skipped_at": null,
      "skipped_count": 0,
      "last_error": null
    }
  ]
}

末尾が_atの項目は、epochミリ秒です。

POST /api/schedules/:name/pause

一時停止中は、定期的な実行が行われません。
停止前に投入されたジョブは、影響を受けません。

{ "ok": true }

POST /api/schedules/:name/resume

一時停止中のスケジュールを再開します。
停止中に経過した回は、実行されません。
再開後の次の実行は、再開した時点より後の最初の予定時刻です。

{ "ok": true }

POST /api/schedules/:name/trigger

スケジュールを直ちに1回実行します。

{ "id": "REPORT:nightly-1767650000000-1-manual", "kind": "flow" }

idは、投入したジョブのIDまたは開始したRunのIDです。
kindは、idがジョブとRunのどちらを示すかを表します。
一時停止中でも実行され、overlapの判定は行われません。
次の定期実行の時刻は、変更されません。
実行に失敗した場合は、500とエラーの詳細で応答します。
一覧のlast_errorの値は、このエラーの詳細と同一です。

Run

GET /api/flows

登録済みのFlow名を取得します。
実行の有無にかかわらず、flowsに登録したすべてのFlow名が対象です。

{ "flows": ["GREETINGS", "PIPELINE"] }

GET /api/runs

Runの一覧を取得します。
各Runについて、ノードの進捗状況が含まれます。

クエリパラメータ

{
  "runs": [
    {
      "id": "GREETINGS:cl9x0a1b2c3d",
      "flow": "GREETINGS",
      "state": "RUNNING",
      "node_total": 6,
      "node_done": 4,
      "node_failed": 0,
      "created_at": 1767225600000,
      "updated_at": 1767225603000
    }
  ],
  "total": 1
}

GET /api/runs/:id

Run1件と、そのRunに含まれるノードを取得します。
ノードには、事前に定義された静的なノードと、実行時に動的に追加されたノードの両方が含まれます。

{
  "run": {
    "id": "GREETINGS:cl9x0a1b2c3d",
    "flow": "GREETINGS",
    "state": "RUNNING",
    "input": "{\"prefix\":\"hello\"}",
    "node_total": 6,
    "node_done": 4,
    "node_failed": 0,
    "created_at": 1767225600000,
    "updated_at": 1767225603000,
    "parent_run_id": null,
    "parent_node_id": null,
    "retryable": false
  },
  "nodes": [
    {
      "id": "greet",
      "binding": "GREET",
      "state": "RUNNING",
      "container": true,
      "parent": null,
      "origin": "static",
      "after": ["list"],
      "job_id": null,
      "child_run_id": null,
      "result": null,
      "error": null,
      "position": 1,
      "created_at": 1767225600000,
      "updated_at": 1767225603000
    }
  ]
}

containerがtrueのノードは、fan-outのノードです。
fan-outのノードはジョブを持たないため、job_idはnullです。
resultは、子ノードの集計値を含みます。
parentを持つノードは、実行時に追加されたノードです。
originはfanOutまたはspawnです。
subflowとして開始されたRunでは、parent_run_idとparent_node_idに、親のRunとノードのIDが含まれます。
subflowのノードでは、child_run_idに、開始した子のRunのrunIdが含まれます。

POST /api/runs

Runを開始します。
リクエストの本文は、Flow名と入力を含むJSONです。

{ "flow": "GREETINGS", "input": { "prefix": "hello" }, "id": "order-1234" }

成功すると、201で応答し、本文は開始したRunのrunIdを含みます。
次の例のrunIdは、flowとidから作られたGREETINGS:order-1234です。

{ "id": "GREETINGS:order-1234" }

POST /api/runs/:id/retry

失敗したRunを再開します。
完了したノードの結果は、そのまま引き継ぎます。
FAILED, STALLED, SKIPPED, CANCELLEDのノードは、未実行として再度実行します。
fan-outで作成された子ノードのうち、再実行されるのは、Runの期限を過ぎて中断された子ノードに限られます。
実行して失敗した子ノードは、再実行しません。

再開の対象は、FAILEDのRunに限られます。
それ以外の状態のRunには、409で応答します。
存在しないRunと保持期間を過ぎたRunには、410で応答します。

POST /api/runs/:id/cancel

Runを取り消します。
未実行のノードは開始されず、実行中のノードには取り消しの要求が送信されます。
実行中のジョブは停止されません。
performerが要求を確認して終了した場合か、要求からreaperGraceMs(既定30秒)が過ぎても結果が報告されない場合に、RunはCANCELLEDに遷移します。
取り消しの対象は、RUNNINGのRunに限られます。

GET /api/openapi.json

すべてのエンドポイントのパラメータと応答の形式を記述したOpenAPI 3.1の定義を取得します。