Multi-step Processes (HelloInterview) のメモ

この記事で学んだことをメモる。

Patterns

• Real-time Updates のメモ
• Dealing with Contention のメモ
• Multi-step Processes のメモ <--- これ 🟢
• Scaling Reads のメモ
• Scaling Writes のメモ
• Handling Large Blobs のメモ

Architectures

• Message Passing
• Managed Workflow Systems

Pattern 1: Message Passing

todo

Pattern 2: Managed Workflow Systems

todo

Temporal (code base) vs Apache airflow (config base)

todo

DD "How will you handle updates to the workflow?" 🔥

Question: "You have 10,000 running workflows for loan approvals. You need to add a new compliance check. How do you update the workflow without breaking existing executions?"

The challenge is that workflows can run for days or weeks. You can't just deploy new code and expect running workflows to handle it correctly. If a workflow started with 5 steps and you deploy a version with 6 steps, what happens when it resumes?

🟢 (解決策1) Workflow Versioning

今後新しく実行するワークフローのみにステップを追加するので良い場合はこちらを使う。

In workflow versioning, we simply create a new version of the workflow code and deploy it separately. Old workflows will continue to run with the old version of the code, and new workflows will run with the new version of the code.

🟢 (解決策2) Workflow Migration

実行中のワークフローに新しいステップを含める場合にはこちらを使わざるを得ない。

// Temporal workflow 
public void loanApprovalWorkflow(LoanApplication app) {
    
    // Step 1 & 2 は共通
    activities.apply(app);
    activities.creditCheck(app);

    // Step 3: ここで1万件のジョブが数日間ポーズしていた
    activities.waitApproval(app);

    // 🔥 Versioning Guard (Migration Logic)
    // "compliance-check-v1" という識別子でバージョンを取得
    int version = Workflow.getVersion("compliance-check-v1", Workflow.DEFAULT_VERSION, 1);

    if (version == 1) {
        // 新しいロジック：移行対象のジョブも含め、ここから新ステップを実行
        activities.complianceCheck(app);
    } else {
        // 過去にすでに Step 4 まで完了して終了したログをリプレイする時は、
        // version = DEFAULT_VERSION が返るため、ここはスキップされる
    }

    // Step 4: 最終ステップ
    activities.disburse(app);
}

ワークフローコードの中に Versioning Guard (if文) を入れる必要が発生する。

これはワークフローが途中で失敗した際のリトライ時の Replay Safety のためである。
ワークフローエンジンはリトライ時、ワークフローコードを上からなぞり、すでに実行済みの場合はログと照らし合わせてそのステップの終了時の状態を再現する。もし Versioning Guard がないと、ステップの再現時に、「新しく追加されたコードはあるが過去のイベントログが見つからない」ため、エラーを起こしてしまう。

DD "How do we keep the workflow state size in check?" 🔥

• input と result はできるだけ小さくする。生データ本体は DB や object storage に置き、workflow には identifier だけ持たせる。
• 1つの long-running workflow に step を増やし続けない。一定の区切りで後続の別の workflow に必要最小限の state だけを渡して続行する。

DD "How do we deal with external events?" 🔥🔥

Question: "Your workflow needs to wait for a customer to sign documents. They might take 5 minutes or 5 days. How do you handle this efficiently?"

(この質問は多分聞かれないと思う)

A. Use signals for external events.
In this way, workflow is not actively running code, but it is not gone. Its state is durably saved, and the engine is waiting for something like a signal, callback, or timeout. When the event arrives, it resumes from that waiting point instead of starting over.

DD "How can we ensure X step runs exactly once?" 🔥🔥🔥

タスクの処理を冪等 (Idempotent) にする。

• ❌ DB に is_done カラムを作り、実行前にチェック && 実行後に更新する。→ これだと複数のワーカーによる race condition に対応できない。
• ❌ DB に is_done カラムを作り、SELECT FOR UPDATEによる Row Lockを用いて、実行前にチェック && 実行後に更新する。→ タスク実行中にDBのコネクションを占有され続け、DBのコネクションプールが枯渇して可用性が悪化する。
• ⭕️ State Machine: DB に status カラムを作り、OCC (WHERE status = ...) を使って状態をPENDING -> PROCESSING -> SUCCESSと遷移させていく。(OCC で良いのは、他の競合リクエストを順番待ちさせる必要がないから。なお、PROCESSING中にワーカーが死んだ場合のゾンビタスクの対策が必要。)
• ⭕️ Distributed Lock: Redis に is_doneフラグを持たせ、読み取りと書き込みを直列化する。

もし、そもそも外部システムが冪等性の担保をしてくれるならば、

• ⭕️ DB に idempotentcy key をDB保存し、外部API(email送信、refund)を叩く際にこれを指定する。

なお、結局その外部システムは、内部で Distributed Lock を使ったタスク実行の直列化を使っていることが多い。(ref. 広告基盤におけるクリック計測の重複除去)

TODO: State Machine と Distributed Lock のPros/Consの比較をする。

メモ、結局flag/status更新と処理をアトミックにしないとexactly once は無理では？アウトボックスパターン？
あとはそもそもタスクを最小粒度にしてワークフローにすると、マネージドワークフローエンジンを使えて楽になりそう？