ワーカー

ワーカーとは何か

ワーカーは、取得できるステップの本体を持ち、デーモンに仕事を求めます。あなたの コードが動く場所で、あなたの import、あなたのコネクションプール、あなたの資格 情報とともに動き、デーモンはそのどれも見ません。デーモンは、展開済みの入力と 先行ステップを添えてリースを渡します。ワーカーは本体を実行し、どうなったかを 報告します。関係はこれだけです。

どの SDK にもワーカーが付いています。持っているワークフローを登録し、リースを ロングポーリングで待ち、本体の実行中はハートビートを送り、すべての拒否に対して その理由ごとに応じます。以下の例は ワークフローのページに あるものと同じプログラムで、ここではそれを処理する側を示しています。

// Serve step bodies for as long as this process lives. The worker registers
// the definition, long-polls for leases, heartbeats while a body runs, and
// answers every 409 by its reason.
const worker = new Worker(mcpd, { concurrency: 8 }).register(support)
void worker.serve()

# Serve step bodies for as long as this process lives. serve() blocks, so it
# runs on a thread of its own when the process does other work too.
worker = Worker(mcpd, concurrency=8).register(support)
threading.Thread(target=worker.serve, daemon=True).start()

// Serve step bodies for as long as this process lives: register, long-poll
// for leases, heartbeat while a body runs, answer every 409 by its reason.
let worker = Worker::builder(&mcpd).concurrency(8).register(support).build();
let serving = {
    let worker = worker.clone();
    tokio::spawn(async move { worker.serve().await })
};

// Serve step bodies for as long as this process lives: register, long-poll
// for leases, heartbeat while a body runs, answer every 409 by its reason.
worker := mcpd.NewWorker(m, mcpd.WorkerConcurrency(8)).Register(support)
go func() { _ = worker.Serve(ctx) }()

// Serve step bodies for as long as this process lives: register, long-poll
// for leases, heartbeat while a body runs, answer every 409 by its reason.
val worker = Worker(mcpd, concurrency = 8).register(support)
launch { worker.serve() }

# Serve step bodies for as long as this node lives: register, long-poll for
# leases, heartbeat while a body runs, answer every 409 by its reason.
{:ok, worker} = StepWorker.start_link(mcpd, workflows: [support], concurrency: 8)
:ok = StepWorker.serve(worker)

プロトコル

呼び出しは 5 つで、これらを話すワーカーは SDK なしで HTTP API に対して書けます。 どの SDK のワーカーも、これらのクライアントであるにすぎません。

リース

リースとは、1 つのワーカーが限られた時間だけ保持する 1 回の試行です。展開済みの 入力、id で引ける先行ステップ、実行の状態と相関、このステップのリトライを またいで安定するキー、そしてデーモンが開いたステップスパンのトレース id と スパン id を含みます。ステップコンテキストを通じて自分のスパンを記録する本体は、 それらをそのスパンの下に置くので、ステップの中のモデル呼び出しは実行のトレースの 正しい深さに収まります。

ワーカーがハートビートを止めたリースは失効します。デーモンはその試行に孤立の 印を付け、回収の猶予を過ぎたら、at_least_once では再びキューに 入れ、at_most_once では失敗させます。停止から戻ってきたワーカーが 回収済みのリースに対して送っても、無視されはしません。そのことが伝えられます。

4 つの拒否

デーモンが受け付けられないワーカーの呼び出しは、4 つの理由のいずれかを伴う 409 で拒否されます。理由の集合は閉じており、どの SDK のワーカーもそれぞれを名前で 処理します。これらを 1 つの分岐にまとめてしまうワーカーは、すでに終えたステップを 捨てたり、成功を失敗として報告したりするからです。

本体

本体は、コンテキスト、展開済みの入力、先行ステップを受け取ります。コンテキストは、 デーモンに残してほしいものを報告する場所です。実行の状態への差分は complete のときに、最上位のキーを置き換える形で統合されます。 使ったトークンは、スパンに値付けするのと同じ価格表でデーモンが値付けします。 そしてキャンセルの合図は、リースを失ったときや実行が終わったときに発火するので、 それを自分の I/O に渡す本体は止まるべきときに止まります。トークンは自動的に 合算されます。本体がコンテキストを通じて記録したスパンと、その下のすべての スパンは、終了時に自分の使用量をステップの使用量に加えます。したがってモデル 呼び出しをスパンとして記録する本体が手で合計することはありません。何も記録 しない本体のための明示的な上書きが set_usage です。

本体は、出力を返すか、例外を投げて終わります。ステップの予算の内側でリトライ してほしいときは SDK のリトライ可能エラーを投げ、それ以外の分類を名指しする ときは型付きのステップ失敗を投げます。それ以外はすべて致命的です。メッセージは デーモンに送られ、ジャーナルに表示されます。スタックトレースは送られません。

ワーカーが持っていない定義、バージョン、ステップに対するリースは unservable として失敗し、本体は実行されません。その試行はリトライの 予算を使わずに、それを持つワーカーのためにキューへ戻ります。これがあるおかげで、 デプロイの最中に 2 つのバージョンのワーカーを並べて動かせます。

スケールアウト

1 つのデーモンに対して、好きなだけワーカーを動かせます。それぞれが自分の並列度 まで取得し、デーモンはすべての試行をちょうど 1 つのワーカーに渡します。2 つの ワーカーが何らかの理由で同じリースを持ってしまっても、フェンスによって勝者は 1 つに決まります。ワーカーの id は、それが取ったすべての試行に記録されるので、 どのプロセスがステップを実行したかはジャーナルが答えます。各プロセスに名前を 付けるか、SDK にホスト名から導出させてください。

取得はロングポーリングなので、待機中のワーカー群がデーモンにかけるコストは、 ワーカーごとに開いたリクエスト 1 つ分だけで、通信は発生しません。待っている 間、取得は、実行が始まったとき、ステップが終わったとき、期限が発火したとき、 イベントが待機を再開させたときに仕事を探し、それ以外では多くても毎秒 2 回 だけ探します。忙しいときのコストはワーカー自身の並列度で頭打ちになり、 1 時間かかる本体のコストは 10 秒ごとのハートビート 1 回だけで、ほかには 何もありません。

キー

ワーカーの呼び出しは worker スコープを取り、実行の開始、シグナル 送信、キャンセルは execute を取ります。どちらも未設定のときは write キーにフォールバックし、admin キーは両方を兼ねます。ステップを処理する だけのワーカーには実行を開始できるキーは要らず、イベントを発行するだけの webhook にはステップを取得できるキーは要りません。フラグについては 設定を参照してください。

リースはステップの入力と先行ステップの出力を持ちますが、ワーカーの呼び出しは 監査ログに記録されません。これは 1 分間に何度も取得する機械のループであり、 取得ごとに 1 行を書けば、人が行った読み取りが埋もれてしまいます。実行の エンドポイントを通じて実行を読み返す操作は記録されます。 セキュリティを参照してください。

次に

• 実行の開始と追跡、待機への応答。 実行。
• ワーカーが処理する定義。 ワークフロー。
• 仕様から生成されたワーカープロトコルの全エンドポイント。 HTTP API、Workers。