Managed Agent をコスト上限つきの使い捨てワーカーとして扱う設計 — 外部入力を本番から隔離する

結果の取り込みは本番と切り離す

使い捨てワーカーが返した結果を、そのまま本番のデータベースに書き込んではいけません。ワーカーは隔離されていても、その出力は信頼できない外部入力の加工物だからです。

私が挟んでいるのは、次の三段階です。

1. ワーカーの出力をいったん検疫用の領域に置く
2. スキーマ検証で形式を機械的にチェックする（想定外のキーや型は弾く）
3. 検証を通ったものだけを本番に取り込む

def adopt_result(raw: dict) -> bool:
    # 期待するスキーマに合致しないものは本番に入れない
    required = {"title", "summary", "source_url"}
    if not required.issubset(raw.keys()):
        return False
    if len(raw["summary"]) > 2000:  # 異常に長い出力は弾く
        return False
    commit_to_production(raw)
    return True

この検疫を挟むだけで、ワーカーが妙な出力を返したときの被害が「本番が汚れる」から「1 件が弾かれる」に下がります。隔離は入力側だけでなく、出力側にも必要だというのが、運用してみての実感です。

同じ入力には同じ結果を返させる

使い捨てワーカーは失敗したら気軽に再実行できますが、再実行で副作用が二重に起きると意味がありません。そこで、取り込みを冪等にしておきます。

具体的には、外部入力からハッシュキーを作り、そのキーで取り込み済みかどうかを判定します。

import hashlib
 
def idempotency_key(untrusted_input: str) -> str:
    return hashlib.sha256(untrusted_input.encode()).hexdigest()[:16]
 
def adopt_once(key: str, raw: dict) -> bool:
    if already_adopted(key):   # 同じ入力は二度取り込まない
        return True
    return adopt_result(raw)

こうしておくと、ワーカーがタイムアウトして再起動しても、同じ URL を二度要約して二重に保存する、といった事故が起きません。使い捨てだからこそ再実行が前提になり、再実行が前提だからこそ冪等性が効いてきます。

落とし穴と回避策

運用していて踏んだ注意点を挙げます。

一つ目は、コスト上限を低くしすぎると途中で打ち切られることです。max_cost_usd を 0.05 のように絞りすぎると、複雑なタスクが完了前に止まり、不完全な出力が返ります。私は処理の重さに応じて 0.10〜0.30 の範囲を推奨しており、打ち切りが頻発するタスクだけ上限を上げるようにしています。

二つ目は、プレビュー版のレート制限です。公開プレビューの段階では、短時間に大量起動するとスロットルされます。使い捨てを並列で大量に走らせる設計にする場合は、起動側にキューを置いて流量を絞る必要があります。これを怠ると、本番のバッチが半分ほど失敗する事態になりました。

三つ目は、サンドボックスの起動コストです。1 リクエストごとにサンドボックスを立てるので、ごく軽い処理を使い捨てワーカーに投げると、処理本体よりも起動のオーバーヘッドが大きくなります。1 秒で終わるような整形は、わざわざ Managed Agent に出さず手元で処理するほうが速くて安いと判断しています。

次に試すこと

まずは、いま自分が「本番環境で直接動かすのが少し怖い」と感じている処理を一つ思い浮かべてください。外部から来る URL や JSON を扱う処理がきっとあるはずです。それを使い捨てワーカーに切り出すところから始めると、Managed Agent の使いどころが体感できます。

私もまだプレビュー版と付き合いながら設計を詰めている途中ですが、永続させずに使い捨てると割り切ったことで、コストも権限も見通しがよくなりました。同じように外部入力の扱いに悩んでいる方の参考になれば幸いです。