ジョブリクエストの重複除去

Cloud Scheduler には、HTTP リクエストと App Engine リクエストにジョブ ID とスケジュール時間の HTTP ヘッダーが含まれています。これにより、再試行された実行の重複を排除し、ターゲット ハンドラをべき等にできます。Cloud Scheduler は少なくとも 1 回の配信を行うため、リクエストの重複除去により、タイムアウトや一時的なネットワークの中断が発生した場合に、ハンドラまたはダウンストリームの Google Cloud API が同じスケジュールされた実行を複数回処理することを防止できます。

リクエストの重複除去の仕組み

同じスケジュール実行に対して重複するジョブリクエストは、次の状況で発生する可能性があります。

  • 長時間実行されるターゲット: ターゲット サービスがレスポンスを返すのに、ジョブの試行期限（attemptDeadline）よりも時間がかかり、Cloud Scheduler がタイムアウトして、最初の試行がまだ実行されている間にジョブを再試行します。HTTP ターゲットの場合、attemptDeadline のデフォルトは 3 分で、最大 30 分まで構成できます。attemptDeadline をターゲットのタイムアウトに合わせるか、実行時間が 30 分を超える可能性がある場合は、長時間実行される処理を非同期で処理します。
  • ネットワーク確認応答の喪失または一時的なプロキシの再試行: ターゲットはリクエストを受信して処理しますが、一時的なネットワークの中断またはプロキシの切断により、Cloud Scheduler が HTTP 2xx レスポンスを受信できません。retryCount を 0 に設定した場合でも、1 回以上の配信では、一時的な転送エラー中にターゲットが重複したリクエストを受信する可能性があります。

ジョブハンドラをべき等にして、再試行されたリクエストの重複を排除するには、Cloud Scheduler が HTTP ターゲットと App Engine ターゲットへのリクエストに自動的に含める HTTP ヘッダーを使用します。

ヘッダー 説明
X-CloudScheduler-JobName projects/PROJECT_ID/locations/LOCATION_ID/jobs/JOB_ID の短いジョブ ID（JOB_ID）。
X-CloudScheduler-ScheduleTime

unix-cron 形式で指定されたジョブの場合、RFC 3339 に従ってフォーマットされた UTC のオフセットとしてのスケジュールされた実行時間（例: 2026-10-16T12:00:00Z または 2026-10-16T05:00:00-07:00）。この値は、同じスケジュールされた実行のすべての再試行で一定です。

このヘッダーは、every 5 minutes などの代替の groc スケジュール形式を使用するジョブでは保証されません。ジョブを手動でトリガーする場合（gcloud scheduler jobs run を実行するか、 Google Cloud コンソールで [ジョブを強制実行] をクリックするなど）、このヘッダーは、次にスケジュールされた実行のタイムスタンプを使用するか、省略される可能性があります。その結果、ヘッダーベースの重複除去により、今後のスケジュールされた実行と、そのスケジュールされた時刻より前にトリガーされた追加の手動実行がスキップされます。

ターゲット エンドポイントで認証を必須にする

X-CloudScheduler-* ヘッダーは認証されておらず、任意の呼び出し元で設定できます。これらのヘッダーを認証に使用しないでください。代わりに、ターゲット タイプに基づいてエンドポイントを保護します。

大文字と小文字を区別しないヘッダー検索を使用する

ヘッダーの大文字と小文字の区別は、HTTP バージョンとインフラストラクチャによって異なる場合があります（たとえば、HTTP/1.1 では X-Cloudscheduler-Jobname、Cloud Run などの HTTP/2 環境では小文字の x-cloudscheduler-jobname と x-cloudscheduler-scheduletime）。ハンドラでは、常に大文字と小文字を区別しないヘッダー ルックアップを使用します。

プロジェクトまたはリージョン間でジョブ ID の曖昧さを解消する

Cloud Scheduler は、プロジェクト ID ヘッダーまたはリージョン ヘッダーを自動的に送信しません。1 つのハンドラ エンドポイントが、同じ JOB_ID を共有する複数のプロジェクトまたはリージョンからリクエストを受信する場合があります。この場合は、プロジェクト ID またはリージョンを HttpTarget.headers、AppEngineHttpTarget.headers、またはターゲット URL パスまたはクエリ パラメータに渡し、それらの値を重複除去キーに含めます。

HTTP ハンドラでリクエストの重複を除去する

ハンドラにリクエストの重複除去を実装する前に、次の設定の前提条件を完了します。

  • Cloud Scheduler が X-CloudScheduler-ScheduleTime ヘッダーを送信するように、ジョブのスケジュールを unix-cron 形式（例: every 5 minutes ではなく */5 * * * *）で構成します。
  • トランザクションまたはアトミックな条件付き書き込みをサポートする永続データストア（Firestore、Spanner、Memorystore など）を設定します。コレクションまたはテーブル（expire_at フィールドなど）に有効期間（TTL）ポリシーを構成して、ジョブの再試行ウィンドウ（maxRetryDuration）後に古い重複除去レコードを削除します。Firestore TTL は、期限切れのレコードを非同期で削除します。通常は 24 ～ 72 時間以内に削除されます。TTL クリーンアップの前に、クラッシュしたインスタンスから再試行で実行を再利用できるようにするには、リース有効期限のタイムスタンプ（lease_expires など、少なくともターゲットのリクエスト タイムアウトと attemptDeadline に設定）を保存し、トランザクションで評価します。

HTTP ハンドラでスケジュールされた各実行を 1 回だけ処理するには、次の手順を完了します。

  1. 受信リクエスト ヘッダーから X-CloudScheduler-JobName と X-CloudScheduler-ScheduleTime を読み取り、固定長の重複除去キー（run_id）にハッシュします。
  2. データストアでトランザクションを実行して、重複除去キーのレコードを確認して要求します。
    • レコードが存在し、status が completed の場合は、ジョブロジックを実行せずに HTTP 200 OK（または 204 No Content）ステータス コードをすぐに返します。これにより、Cloud Scheduler は再試行を停止します。
    • レコードが存在し、status が in_progress で、リース（lease_expires）が期限切れになっていない場合は、409 Conflict などの再試行可能なエラー ステータス コードを返します。これにより、処理中の試行が失敗した場合に、Cloud Scheduler は再試行ポリシーを有効に保ちます。App Engine ターゲットから 503 Service Unavailable を返さないようにしてください。App Engine は 503 をシステムエラーとして扱い、retryCount にカウントせずにリクエストを再試行します。
    • レコードが存在しないか、リースが期限切れになっている場合は、status を in_progress に設定し、一意の試行 ID（attempt_id）、リース有効期限のタイムスタンプ（lease_expires）、TTL 有効期限のタイムスタンプ（expire_at）を設定してレコードを書き込みます。

  3. レコードを要求したら、ジョブ ロジックを実行します。トランザクションで、レコードを変更する前に、レコードの attempt_id が現在の試行と一致していることを確認します。

    • ジョブ ロジックが成功した場合は、status を completed に更新し、HTTP 200 OK ステータス コードを返します。
    • 再試行可能なエラーでジョブ ロジックが失敗した場合は、5xx ステータス コードを返す前に重複除去レコードを削除して、Cloud Scheduler がリースの有効期限を待たずに実行を再試行できるようにします。

HTTP ハンドラの例

次の Python の例（Cloud Run または Cloud Run functions 用）では、リース付きの Firestore トランザクションを使用して、実行を要求し、再試行されたジョブリクエストの重複を排除します。

このサンプルはプロダクション レディではありません。デプロイする前に、ニーズに合わせて確認して調整します。

import datetime
import hashlib
import uuid

import flask
import functions_framework
from google.cloud import firestore

# Longer than the work takes, and at least the job's attempt deadline.
LEASE = datetime.timedelta(minutes=5)
# Keep records past the job's retry window. A TTL policy on expire_at deletes them.
RETENTION = datetime.timedelta(days=1)

db = firestore.Client()


def run_job_logic() -> None:
    """Does the scheduled work. Replace the body with your own task."""


@firestore.transactional
def claim_run(
    transaction: firestore.Transaction,
    run_ref: firestore.DocumentReference,
    attempt_id: str,
    now: datetime.datetime,
) -> str:
    """Claims the run unless it's done or another attempt holds the lease."""
    snapshot = run_ref.get(transaction=transaction)
    if snapshot.exists:
        run = snapshot.to_dict()
        if run["status"] == "completed":
            return "completed"
        if run["lease_expires"] > now:
            return "in_progress"
    transaction.set(
        run_ref,
        {
            "status": "in_progress",
            "attempt_id": attempt_id,
            "lease_expires": now + LEASE,
            "expire_at": now + RETENTION,
        },
    )
    return "claimed"


@firestore.transactional
def finish_run(
    transaction: firestore.Transaction,
    run_ref: firestore.DocumentReference,
    attempt_id: str,
    succeeded: bool,
) -> None:
    """Completes or releases the run, if this attempt still holds the claim."""
    snapshot = run_ref.get(transaction=transaction)
    if not snapshot.exists or snapshot.to_dict().get("attempt_id") != attempt_id:
        return  # A retry claimed the run after this attempt's lease expired.
    if succeeded:
        transaction.update(run_ref, {"status": "completed"})
    else:
        transaction.delete(run_ref)


@functions_framework.http
def handle_scheduled_job(request: flask.Request) -> tuple[str, int]:
    """Runs a job's work once per scheduled run, even if the run is delivered twice."""
    job_name = request.headers.get("X-CloudScheduler-JobName")
    schedule_time = request.headers.get("X-CloudScheduler-ScheduleTime")
    if not job_name or not schedule_time:
        return "error: missing Cloud Scheduler headers", 400

    # Retries keep the same schedule time. A forced run uses the next
    # scheduled time, so that scheduled run is then skipped.
    run_id = hashlib.sha256(f"{job_name}/{schedule_time}".encode()).hexdigest()
    run_ref = db.collection("scheduler_runs").document(run_id)
    now = datetime.datetime.now(datetime.timezone.utc)
    attempt_id = uuid.uuid4().hex

    status = claim_run(db.transaction(), run_ref, attempt_id, now)
    if status == "completed":
        return "Duplicate run skipped", 200
    if status == "in_progress":
        # A non-2xx response keeps retries going in case that attempt fails.
        return "Run in progress", 409

    try:
        run_job_logic()
    except Exception:
        # Releases the claim so that a retry can process the run.
        finish_run(db.transaction(), run_ref, attempt_id, succeeded=False)
        raise
    finish_run(db.transaction(), run_ref, attempt_id, succeeded=True)
    return "Run processed", 200

Google Cloud API ターゲットへの重複リクエストを防止する

Cloud Scheduler が HTTP 経由で Google Cloud API エンドポイント（*.googleapis.com）を直接呼び出すと、ターゲット API は X-CloudScheduler-JobName ヘッダーと X-CloudScheduler-ScheduleTime ヘッダーを無視します。たとえば、workflowexecutions.googleapis.com を呼び出して Workflows の実行を開始する、run.googleapis.com を呼び出して Cloud Run ジョブを実行する、cloudbuild.googleapis.com を呼び出して Cloud Build トリガーを実行すると、再試行のたびに新しい実行が開始されます。

本番環境の信頼性を確保するため、状態を変更する最も下流のサービスでべき等性を適用します。アップストリーム レイヤでのみリクエストの重複除去を行うと、再試行ウィンドウが狭くなります。ただし、ダウンストリーム リクエストが成功し、HTTP レスポンスがネットワーク経由で失われた場合、アップストリームの呼び出し元はオペレーションが完了したかどうかを判断できず、リクエストを再試行する可能性があります。

Google Cloud API ターゲットをスケジュールするときに重複実行を防ぐには、次のアプローチを組み合わせてエンドツーエンドのべき等性を実現します。

  • べき等プロキシを介してリクエストをルーティングする: Cloud Scheduler から *.googleapis.com エンドポイントを直接呼び出す代わりに、独自の Cloud Run サービスまたは Cloud Run functions プロキシをデプロイし、そのプロキシを呼び出すように Cloud Scheduler を構成します。プロキシで、このドキュメントの HTTP ハンドラでリクエストの重複を排除するで説明されている Python と Firestore のアトミック トランザクション クレーム パターンを実装します。

    プロキシは X-CloudScheduler-JobName と X-CloudScheduler-ScheduleTime を読み取り、再試行されたリクエストの重複を除去し、重複除去キーをリクエスト ペイロードのべき等トークンとして渡します（たとえば、Workflows の実行を開始するときに argument フィールドに渡します）。

    プロキシで認証を要求し（たとえば、--no-allow-unauthenticated を使用してデプロイする）、OIDC トークンで認証するように Cloud Scheduler ジョブを構成します。必要最小限の Identity and Access Management（IAM）ロール（ターゲット ワークフローの roles/workflows.invoker や重複除去レコードの roles/datastore.user など）のみを持つ専用のサービス アカウントでプロキシを実行します。

  • ダウンストリーム サービスでべき等性を適用する: べき等性トークンを使用して、状態変更ステップを実行する前に、Firestore などのデータストアでアトミック チェックまたはトランザクション チェックを実行します。サービスが呼び出すダウンストリーム HTTP エンドポイントまたは Cloud Run functions にトークンを渡して、それらのターゲットでも再試行の重複除去を行えるようにします。Workflows、Firestore トランザクション、Cloud Run functions を使用したエンドツーエンドの例については、 Google Cloudのブログ投稿「Workflows の使用時に API 呼び出しを正確に 1 回だけ行う方法」をご覧ください。