背景と課題：大規模環境における分散ビジネスロジックの管理
半年前、私たちの注文処理パイプラインは、1日あたり約45,000件のトランザクション負荷に耐えきれなくなっていました。多くのマイクロサービス構成と同様に、決済フローはKafkaトピック、Postgresのステータスフラグ、そして5分間隔のcronバッチ処理という場当たり的な組み合わせに依存していました。
チェックアウト処理中にStripeがタイムアウトしたり在庫サービスが再起動したりするたびに、注文データは不整合な状態のまま停止してしまいました。オンコール担当のエンジニアは毎週10〜15時間も費やし、手動の補正スクリプトを実行したり、デッドレターキューを整理したり、取り残された孤立レコードの整合性を合わせたりしていました。
分散トランザクションは非常に壊れやすいことで知られています。ネットワークでのパケットロス、コンテナの予期せぬ再起動、アップストリームのサードパーティAPIによる一時的な502エラーなど日常茶飯事です。これらを手動で対処しようとすると、SQL上で独自のステートマシンを構築したり、多数のサービス間で冪等性キー（Idempotency Key）を引き回したり、すべてのHTTPクライアントに個別のリトライループを組み込んだりすることになります。気がつけば、そうした防御的なボイラープレートコードがコードベースの大半を埋め尽くしてしまいます。
Temporalは「耐久性のある実行（Durable Execution）」によってこの問題を解決します。ステップの遷移やリトライ回数を自前で追跡する代わりに、シンプルで順次実行されるGoコードを書くだけで済みます。内部では、Temporalクラスターが状態遷移を追加専用（append-only）のイベントログに記録します。関数実行の途中でワーカープロセスがクラッシュしたり、基盤となるVMが消失したりしても、別のワーカーが即座に実行を引き継ぎます。実行状態が再構築され、処理が中断されたコードのまさにその行から処理が再開されます。
決済パイプラインをTemporalに移行した結果、月に約120件発生していたゴーストトランザクション（不整合な注文）はゼロになりました。注文ルーティング関連のオンコール通知も事実上消失しました。さらに素晴らしいことに、インフラの一時的な障害に対してエンジニアが個別の補償処理ルートを設計する必要がなくなったため、プルリクエストのレビューサイクルが30%高速化されました。
インストール：TemporalサーバーとGo SDKのセットアップ
ローカルのTemporalクラスターの立ち上げは2分もかかりません。Temporal CLIには、組み込みのSQLiteデータストアとWebコンソールを備えた自己完結型の開発サーバーが同梱されています。
公式インストールスクリプトまたはHomebrewを使用してCLIを取得します：
# curlを使用したLinuxおよびmacOS向けインストール
curl -sSf https://temporal.download/cli.sh | sh
# Homebrewを使用したmacOS向けインストール
brew install temporalローカルサーバーのインスタンスを起動します：
temporal server start-devこのコマンドにより、127.0.0.1:7233でgRPCエンジンが起動し、http://localhost:8233でWeb UIが開始されます。ブラウザでそのURLを開くと、実行中のワークフローの確認、イベントタイムラインの監査、タスクキューのスループットの確認が行えます。
サーバーが起動したら、Goモジュールを初期化し、公式SDKをインストールします：
mkdir temporal-go-orders
cd temporal-go-orders
go mod init temporal-go-orders
go get go.temporal.io/sdk設定：Goにおけるワークフロー、アクティビティ、ワーカーの構造化
Temporalのオーケストレーションは、主に「ワークフロー（Workflows）」と「アクティビティ（Activities）」という2つの構成要素を中心に構築されます。
- ワークフロー（Workflows）： 純粋なオーケストレーションロジックを記述します。決定論的（Deterministic）である必要があり、ワークフロー関数内から直接ネットワーク呼び出しを行ったり、システムクロックを参照したり、シードなしの乱数を生成したりしてはいけません。
- アクティビティ（Activities）： クレジットカード決済、SQLの更新、外部APIへのリクエストなど、副作用を伴う処理を実行します。Temporalは設定されたバックオフポリシーに従ってアクティビティを自動的にリトライします。
1. アクティビティの定義
activities.goで、外部との対話を行う処理を定義します：
package main
import (
"context"
"errors"
"fmt"
"math/rand"
)
type OrderRequest struct {
OrderID string
Amount float64
Email string
}
type OrderActivities struct{}
func (a *OrderActivities) ProcessPayment(ctx context.Context, order OrderRequest) (string, error) {
// 決済ゲートウェイのレイテンシと一時的な障害（失敗率30%）をシミュレート
if rand.Float32() < 0.3 {
return "", errors.New("決済ゲートウェイのタイムアウト：ピアにより接続がリセットされました")
}
txnID := fmt.Sprintf("TXN-%s-OK", order.OrderID)
return txnID, nil
}
func (a *OrderActivities) ReserveInventory(ctx context.Context, orderID string) error {
// 内部の在庫サービスを呼び出す
return nil
}
func (a *OrderActivities) SendConfirmationEmail(ctx context.Context, email string, orderID string) error {
// ResendまたはSendGrid経由で確認メールを送信
return nil
}2. 決定論的ワークフローの実装
次に、workflow.goでビジネスフローを定義します。workflow.ActivityOptionsで明示的なタイムアウトと指数バックオフのリトライポリシーを設定している点に注目してください：
package main
import (
"time"
"go.temporal.io/sdk/temporal"
"go.temporal.io/sdk/workflow"
)
func OrderWorkflow(ctx workflow.Context, order OrderRequest) (string, error) {
retryPolicy := &temporal.RetryPolicy{
InitialInterval: time.Second * 1,
BackoffCoefficient: 2.0,
MaximumInterval: time.Second * 30,
MaximumAttempts: 5,
}
actOptions := workflow.ActivityOptions{
StartToCloseTimeout: time.Minute * 2,
RetryPolicy: retryPolicy,
}
ctx = workflow.WithActivityOptions(ctx, actOptions)
var activities *OrderActivities
// ステップ1: 自動リトライ付きで顧客のクレジットカードに請求
var transactionID string
err := workflow.ExecuteActivity(ctx, activities.ProcessPayment, order).Get(ctx, &transactionID)
if err != nil {
return "", err
}
// ステップ2: 在庫アイテムを確保
err = workflow.ExecuteActivity(ctx, activities.ReserveInventory, order.OrderID).Get(ctx, nil)
if err != nil {
return "", err
}
// ステップ3: 確認メールを送信
err = workflow.ExecuteActivity(ctx, activities.SendConfirmationEmail, order.Email, order.OrderID).Get(ctx, nil)
if err != nil {
return "", err
}
return transactionID, nil
}3. ワーカープロセスの構築
ワーカーは、gRPC経由でTemporalのタスクキューをポーリングする常駐型のGoバイナリです。main.goに以下のワーカー実装を記述します：
package main
import (
"context"
"log"
"go.temporal.io/sdk/client"
"go.temporal.io/sdk/worker"
)
const TaskQueueName = "orders-queue"
func main() {
c, err := client.Dial(client.Options{
HostPort: client.DefaultHostPort,
})
if err != nil {
log.Fatalln("Temporalクライアントの作成に失敗しました:", err)
}
defer c.Close()
w := worker.New(c, TaskQueueName, worker.Options{})
act := &OrderActivities{}
w.RegisterWorkflow(OrderWorkflow)
w.RegisterActivity(act)
go func() {
// ワーカー起動後にサンプルワークフローの実行をトリガー
workflowOptions := client.StartWorkflowOptions{
ID: "order-sample-1001",
TaskQueue: TaskQueueName,
}
req := OrderRequest{
OrderID: "1001",
Amount: 89.99,
Email: "[email protected]",
}
we, startErr := c.ExecuteWorkflow(context.Background(), workflowOptions, OrderWorkflow, req)
if startErr != nil {
log.Println("ワークフローの開始に失敗しました:", startErr)
return
}
log.Printf("ワークフローを開始しました WorkflowID: %s, RunID: %s\n", we.GetID(), we.GetRunID())
}()
log.Println("タスクキューをリッスン中:", TaskQueueName)
err = w.Run(worker.InterruptCh())
if err != nil {
log.Fatalln("ワーカーで致命的なエラーが発生しました:", err)
}
}検証とモニタリング：障害テストとオブザーバビリティ
プロセスの突然のクラッシュ時にシステムがどのように動作するか確認してみましょう。ワーカーを起動します：
go run .ブラウザでhttp://localhost:8233を開き、order-sample-1001というタイトルの実行を選択します。
耐障害性の検証
ターミナルから直接、プロセスの強制終了をシミュレートします：
go run .を実行してワークフローを開始します。ProcessPaymentの実行中にCtrl+Cを押して、ワーカープロセスを即座にクラッシュさせます。- TemporalのWebコンソールを確認します。ワークフローは失敗することなく、オープンで正常な状態を維持しています。Temporalは単にアクティブなワーカーがキューをポーリングするのを待機している状態です。
go run .でワーカーを再起動します。Temporalがキューに入ったタスクを再割り当てします。ワークフローは過去の処理を重複実行することなく、中断されたまさにその箇所から再開されます。
本番環境でのモニタリングと重要メトリクス
このアーキテクチャを高スループットな環境で運用する場合、2つの極めて重要なシグナルを監視する必要があります：
- Schedule-To-Startレイテンシ： Temporalがタスクをエンキューしてから、ワーカーがそれを取得するまでの時間差を追跡します。正常なクラスターでは通常50ミリ秒未満を維持します。これが1〜2秒を超えて上昇し始めた場合は、ワーカーインスタンスをスケールアウトしてください。
- アクティビティの失敗率： アップストリームのプロバイダーで障害が発生した際は、Temporal UIの履歴ツリーを検査します。各リトライの完全なスタックトレース、入力ペイロード、失敗理由が確認できます。非構造化ログを掘り起こすことなく、問題を即座に診断可能です。
Temporalのイベントソーシング型ステートエンジンとGoの軽量な並行処理モデルを組み合わせることで、ネットワークの切断やサーバーのクラッシュにも標準で耐えうる、クリーンで堅牢なワークフローが実現できます。
