temporal-golang-pro

Temporal Go SDK (temporal-golang-pro)

概要

Temporal Go SDK を使用して、回復性、スケーラビリティ、決定性に優れた分散システムを構築するためのエキスパート向けガイドです。このスキルでは、あいまいなオーケストレーション要件を、プロジェクト固有の検証を必要とする Go の実装案へと変換します。耐久性のある実行、厳格な決定性、エンタープライズ規模のワーカー設定に重点を置いています。

このスキルを使用する場面

• 分散システムの設計：永続的な状態と信頼性の高いオーケストレーションが必要なマイクロサービスを構築する場合。
• 複雑なワークフローの実装：Go SDK を使用して、長期間（数日／数か月）に及ぶプロセスや複雑な Saga パターンを処理する場合。
• パフォーマンスの最適化：ワーカーに細かな同時実行制御、mTLS セキュリティ、カスタムインターセプターが必要な場合。
• 信頼性の確保：冪等性のあるアクティビティ、適切なエラー処理、高度な再試行ポリシーを実装する場合。
• 保守と進化：実行中のワークフローをバージョン管理する場合や、ダウンタイムなしでワーカーを更新する場合。

このスキルを使用しない場面

• Temporal を他の SDK（Python、Java、TypeScript）で使用する場合は、それぞれの -pro スキルを参照してください。
• 耐久性や調整を必要としない、単純なリクエスト／レスポンスのタスクの場合。
• 実装を伴わない上位レベルの設計の場合（workflow-orchestration-patterns を使用してください）。

ステップごとのガイド

1. コンテキストの収集：次の情報を事前に確認してください。
   • 対象の Temporal Cluster（Cloud または Self-hosted）と Namespace。
   • Task Queue の名前と想定スループット。
   • セキュリティ要件（mTLS のパス、認証）。
   • 障害モードと、希望する再試行／タイムアウトポリシー。
2. 決定性の検証：ワークフローコードを提案する前に、次の 5 つのルールに照らして検証してください。
   • Go 標準の並行処理（goroutines）を使用しない。
   • Go 標準の時刻機能（time.Now、time.Sleep）を使用しない。
   • 決定性のない map の反復処理を行わない（キーをソートする必要があります）。
   • 外部 I/O やネットワーク呼び出しを直接行わない。
   • 決定性のない乱数を使用しない。
3. 段階的な実装：共有の Protobuf／データクラスから始め、次にアクティビティ、ワークフロー、最後にワーカーを実装してください。
4. リソースの活用：実装に高度なパターン（Saga、インターセプター、リプレイテスト）が必要な場合は、実装プレイブックとテスト戦略を明示的に参照してください。

機能

Go SDK の実装

• ワーカー管理：worker.Options に関する深い知識。MaxConcurrentActivityTaskPollers、WorkerStopTimeout、StickyScheduleToStartTimeout を含みます。
• インターセプター：横断的な関心事（ロギング、トレース、認証）に対応するクライアント、ワーカー、ワークフローのインターセプターを実装します。
• カスタムデータコンバーター：Protobuf、暗号化されたペイロード、カスタム JSON マーシャリングを統合します。

高度なワークフローパターン

• 耐久性のある並行処理：Go 標準のプリミティブの代わりに、workflow.Go、workflow.Channel、workflow.Selector を使用します。
• バージョン管理：workflow.GetVersion と workflow.GetLogger を使用して、安全なコードの進化を実装します。
• 大規模処理：履歴サイズの上限（デフォルト：50MB または 50K イベント）を管理するための ContinueAsNew パターン。
• 子ワークフロー：ライフサイクル、キャンセル、親子間のシグナル伝播を管理します。

テストと可観測性

• Testsuite の習熟：WorkflowTestSuite を使用して、決定論的な時刻制御によるユニットテストと機能テストを行います。
• モック：高度なアクティビティと子ワークフローのモック戦略。
• リプレイテスト：コード変更を本番環境のイベント履歴に照らして検証します。
• メトリクス：ワーカーのパフォーマンスを追跡するために、Prometheus／OpenTelemetry エクスポーターを設定します。

例

例 1：バージョン管理されたワークフロー（決定論的）

// Note: imports omitted. Requires 'go.temporal.io/sdk/workflow', 'go.temporal.io/sdk/temporal', and 'time'.
func SubscriptionWorkflow(ctx workflow.Context, userID string) error {
    // 1. Versioning for logic evolution (v1 = DefaultVersion)
    v := workflow.GetVersion(ctx, "billing_logic", workflow.DefaultVersion, 2)

    for i := 0; i < 12; i++ {
        ao := workflow.ActivityOptions{
            StartToCloseTimeout: 5 * time.Minute,
            RetryPolicy: &temporal.RetryPolicy{MaximumAttempts: 3},
        }
        ctx = workflow.WithActivityOptions(ctx, ao)

        // 2. Activity Execution (Always handle errors)
        err := workflow.ExecuteActivity(ctx, ChargePaymentActivity, userID).Get(ctx, nil)
        if err != nil {
            workflow.GetLogger(ctx).Error("Payment failed", "Error", err)
            return err
        }

        // 3. Durable Sleep (Time-skipping safe)
        sleepDuration := 30 * 24 * time.Hour
        if v >= 2 {
            sleepDuration = 28 * 24 * time.Hour
        }

        if err := workflow.Sleep(ctx, sleepDuration); err != nil {
            return err
        }
    }
    return nil
}

例 2：完全な mTLS ワーカー設定

func RunSecureWorker() error {
    // 1. Load Client Certificate and Key
    cert, err := tls.LoadX509KeyPair("client.pem", "client.key")
    if err != nil {
        return fmt.Errorf("failed to load client keys: %w", err)
    }

    // 2. Load CA Certificate for Server verification (Proper mTLS)
    caPem, err := os.ReadFile("ca.pem")
    if err != nil {
        return fmt.Errorf("failed to read CA cert: %w", err)
    }
    certPool := x509.NewCertPool()
    if !certPool.AppendCertsFromPEM(caPem) {
        return fmt.Errorf("failed to parse CA cert")
    }

    // 3. Dial Cluster with full TLS config
    c, err := client.Dial(client.Options{
        HostPort:  "temporal.example.com:7233",
        Namespace: "production",
        ConnectionOptions: client.ConnectionOptions{
            TLS: &tls.Config{
                Certificates: []tls.Certificate{cert},
                RootCAs:      certPool,
            },
        },
    })
    if err != nil {
        return fmt.Errorf("failed to dial temporal: %w", err)
    }
    defer c.Close()

    w := worker.New(c, "payment-queue", worker.Options{})
    w.RegisterWorkflow(SubscriptionWorkflow)
    w.RegisterActivity(ChargePaymentActivity) // implement with a stable per-charge idempotency key

    if err := w.Run(worker.InterruptCh()); err != nil {
        return fmt.Errorf("worker run failed: %w", err)
    }
    return nil
}

例 3：Selector とシグナルの統合

func ApprovalWorkflow(ctx workflow.Context) (string, error) {
    var approved bool
    signalCh := workflow.GetSignalChannel(ctx, "approval-signal")

    // Use Selector to wait for multiple async events
    s := workflow.NewSelector(ctx)
    s.AddReceive(signalCh, func(c workflow.ReceiveChannel, _ bool) {
        c.Receive(ctx, &approved)
    })

    // Add 72-hour timeout timer
    s.AddFuture(workflow.NewTimer(ctx, 72*time.Hour), func(f workflow.Future) {
        approved = false
    })

    s.Select(ctx)

    if !approved {
        return "rejected", nil
    }
    return "approved", nil
}

ベストプラクティス

• ✅ 推奨: ExecuteActivity と client.Dial のエラーは必ず処理してください。
• ✅ 推奨: 並行処理には workflow.Go と workflow.Channel を使用してください。
• ✅ 推奨: 決定性を保つため、マップのキーを並べ替えてから反復処理してください。
• ✅ 推奨: 1 分を超えるアクティビティには activity.RecordHeartbeat を使用してください。
• ✅ 推奨: replayer.ReplayWorkflowHistoryFromJSON を使用して、ロジックの互換性をテストしてください。
• ❌ 非推奨: 本番ワーカーで _ や log.Fatal を使ってエラーを握りつぶさないでください。
• ❌ 非推奨: ワークフロー関数内でネットワークやディスクへの直接 I/O を行わないでください。
• ❌ 非推奨: 標準の time.Now() や rand.Int() に依存しないでください。
• ❌ 非推奨: 耐久性を必要としない単純な cron ジョブには適用しないでください。

トラブルシューティング

• パニック：決定性の不一致：通常は、workflow.GetVersion を使わずにロジックを変更した場合や、決定性のないコード（例：標準のマップ）が原因です。
• エラー：履歴サイズの上限超過：履歴の上限（デフォルトで 50K イベント）に達しています。ContinueAsNew が実装されていることを確認してください。
• ワーカーのハング：WorkerStopTimeout を確認し、すべてのアクティビティがコンテキストのキャンセルを処理することを確認してください。

制限事項

• Temporal Cloud UI の操作や、TLS 証明書のプロビジョニングワークフローは対象外です。
• Temporal Java、Python、TypeScript SDK は対象外です。それぞれ専用の -pro スキルを参照してください。
• Temporal Server v1.20+ と Go SDK v1.25+ を前提としています。古い SDK バージョンでは API が異なる場合があります。
• Nexus やマルチクラスターのレプリケーションは対象外です。サポート状況は、固定された SDK／サーバーのドキュメントで確認してください。
• グローバルな名前空間の構成やマルチリージョンのフェイルオーバー設定は対象外です。
• 現行の Worker Deployment／Versioning API は対象外です。固定されたバージョンのドキュメントを参照して、アップグレード戦略を選択してください。

リソース

• 実装プレイブック - Go SDK のパターンを詳しく解説します。
• テスト戦略 - Go のユニットテスト、リプレイテスト、統合テストについて説明します。
• Temporal Go SDK リファレンス
• Temporal Go サンプル

関連スキル

• grpc-golang - 内部トランスポートプロトコルと Protobuf の設計。
• golang-pro - Go の一般的なパフォーマンスチューニングと高度な構文。
• workflow-orchestration-patterns - 言語に依存しないオーケストレーション戦略。

本番環境での注意点

例ではインポート、アクティビティ型、耐久性のあるストレージを省略しており、コンパイル可能なアプリケーションではありません。参照するすべてのアクティビティを登録してください。課金ループでは暦上の月ではなく、28 日／30 日の期間を使用しています。リトライによる二重課金や異なる期間の統合を防ぐため、明示的な論理課金 ID を、プロバイダー／データベースのアトミックな冪等性の契約に渡してください。Saga には補償処理の設計が必要であり、任意の外部作用のロールバックを保証することはできません。ハートビートに基づく検出／キャンセルには HeartbeatTimeout を設定し、ワーカー内のレート制限とタスクキュー全体のレート制限を区別してください。使用しないタイマーはキャンセルし、ワークフローのキャンセルを拒否とは別に処理してください。SDK／サーバーのバージョンを固定し、移行前に API、デプロイメント／バージョンの保持、リセット／リプレイの要件、実際の履歴上限を確認してください。現在実行中のインスタンスがないことだけでは、すべてのリプレイ互換性を破棄する根拠として不十分です。機密性の高い履歴ペイロードは暗号化して最小限にとどめ、エクスポートした履歴／証明書を保護してください。TLS 検証は決して無効にしないでください。