docs: Add deduplication sample for Cloud Scheduler job requests

cloud_scheduler/snippets/deduplicate_job_requests.py

@@ -0,0 +1,126 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Demonstrates how to deduplicate retried Cloud Scheduler HTTP requests.
+See https://docs.cloud.google.com/scheduler/docs/configuring/deduplicate-job-requests
+"""
+
+# [START cloudscheduler_deduplicate_job_requests]
+import datetime
+import re
+
+import flask
+from google.api_core.exceptions import AlreadyExists
+from google.cloud import firestore
+
+db = firestore.Client()
+JOB_ID_RE = re.compile(r"^[A-Za-z0-9_-]{1,500}$")
+# Set LOCK_TTL to at least your job's retry window (maxRetryDuration) so a
+# Firestore TTL policy can asynchronously clean up old records. Because
+# Firestore TTL deletion isn't instantaneous, if an instance terminates
+# unexpectedly (for example, due to an out-of-memory error or SIGKILL) before
+# doc_ref.delete() runs, the record remains IN_PROGRESS until cleaned up (unless
+# you also check a lease expiration timestamp in a transaction).
+LOCK_TTL = datetime.timedelta(hours=24)
+
+
+def run_job_logic() -> None:
+    """Replace with your job processing logic."""
+    pass
+
+
+def handle_scheduled_job(request: flask.Request) -> tuple[str, int]:
+    """Deduplicates retried Cloud Scheduler requests using Firestore.
+
+    Args:
+        request: The incoming Flask HTTP request object.
+
+    Returns:
+        A tuple containing the response body and HTTP status code.
+    """
+    # Assumes the endpoint requires authentication (such as OIDC); these headers
+    # aren't authenticated.
+    # request.headers in Flask and Functions Framework performs case-insensitive
+    # lookups automatically.
+    job_name = request.headers.get("X-CloudScheduler-JobName")
+    schedule_time = request.headers.get("X-CloudScheduler-ScheduleTime")
+
+    # X-CloudScheduler-ScheduleTime is only sent for scheduled runs in
+    # unix-cron format. Jobs using groc schedules omit this header, and manual
+    # triggers (gcloud scheduler jobs run or "Force a job run") might omit it
+    # or reuse a scheduled timestamp. To allow manual runs when schedule_time
+    # is omitted, run run_job_logic() directly when job_name is valid and
+    # schedule_time is None.
+    if not job_name or not schedule_time:
+        return ("Missing Cloud Scheduler headers", 400)
+
+    # Validate the job ID (prevent "/" path traversal) and normalize the
+    # RFC 3339 timestamp to UTC so equivalent offsets map to the same key.
+    if not JOB_ID_RE.fullmatch(job_name):
+        return ("Invalid X-CloudScheduler-JobName header", 400)
+    try:
+        parsed_time = datetime.datetime.fromisoformat(
+            schedule_time.replace("Z", "+00:00")
+        )
+        if parsed_time.tzinfo is None:
+            return (
+                "X-CloudScheduler-ScheduleTime must include a timezone",
+                400,
+            )
+        normalized_time = parsed_time.astimezone(
+            datetime.timezone.utc
+        ).strftime("%Y-%m-%dT%H:%M:%SZ")
+    except ValueError:
+        return ("Invalid X-CloudScheduler-ScheduleTime header", 400)
+
+    # If this endpoint receives jobs from multiple projects or regions, also
+    # validate and include the project ID or region in dedup_key.
+    dedup_key = f"{job_name}:{normalized_time}"
+    doc_ref = db.collection("scheduler_dedup").document(dedup_key)
+    now = datetime.datetime.now(datetime.timezone.utc)
+
+    try:
+        # create() is atomic and raises AlreadyExists if the document exists.
+        doc_ref.create(
+            {
+                "job_name": job_name,
+                "schedule_time": normalized_time,
+                "status": "IN_PROGRESS",
+                "created_at": now,
+                "expires_at": now + LOCK_TTL,
+            }
+        )
+    except AlreadyExists:
+        # Firestore returns ALREADY_EXISTS (HTTP 409 Conflict) if the document
+        # already exists.
+        doc = doc_ref.get()
+        if doc.exists and doc.to_dict().get("status") == "COMPLETED":
+            # A previous attempt already succeeded. Return 200 OK so Cloud
+            # Scheduler stops retrying.
+            return ("Duplicate job request ignored", 200)
+        # Another attempt is still running. Return 409 Conflict so Cloud
+        # Scheduler keeps its retry policy active if that attempt fails.
+        return ("Job execution already in progress", 409)
+
+    try:
+        run_job_logic()
+    except BaseException:
+        # If the job logic fails or is terminated (such as by SystemExit),
+        # delete the lock document and re-raise so a subsequent retry can run.
+        doc_ref.delete()
+        raise
+
+    doc_ref.update({"status": "COMPLETED"})
+    return ("Job completed", 200)
+# [END cloudscheduler_deduplicate_job_requests]

cloud_scheduler/snippets/deduplicate_job_requests_test.py

@@ -0,0 +1,166 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import datetime
+from typing import Iterator
+from unittest import mock
+import uuid
+
+import deduplicate_job_requests
+import flask
+import pytest
+
+SCHEDULE_TIME_UTC = "2026-10-16T12:00:00Z"
+SCHEDULE_TIME_OFFSET = "2026-10-16T05:00:00-07:00"
+
+
+@pytest.fixture
+def app() -> flask.Flask:
+    return flask.Flask(__name__)
+
+
+@pytest.fixture
+def job_name() -> Iterator[str]:
+    name = f"test-job-{uuid.uuid4().hex[:12]}"
+    yield name
+    dedup_key = f"{name}:{SCHEDULE_TIME_UTC}"
+    deduplicate_job_requests.db.collection("scheduler_dedup").document(
+        dedup_key
+    ).delete()
+
+
+def test_missing_or_invalid_headers(app: flask.Flask) -> None:
+    with app.test_request_context("/"):
+        body, status = deduplicate_job_requests.handle_scheduled_job(
+            flask.request
+        )
+        assert status == 400
+        assert "Missing" in body
+
+    with app.test_request_context(
+        "/",
+        headers={
+            "X-CloudScheduler-JobName": "../invalid/job",
+            "X-CloudScheduler-ScheduleTime": SCHEDULE_TIME_UTC,
+        },
+    ):
+        body, status = deduplicate_job_requests.handle_scheduled_job(
+            flask.request
+        )
+        assert status == 400
+        assert "Invalid X-CloudScheduler-JobName" in body
+
+    with app.test_request_context(
+        "/",
+        headers={
+            "X-CloudScheduler-JobName": "valid-job",
+            "X-CloudScheduler-ScheduleTime": "2026-10-16T12:00:00",
+        },
+    ):
+        body, status = deduplicate_job_requests.handle_scheduled_job(
+            flask.request
+        )
+        assert status == 400
+        assert "timezone" in body
+
+
+def test_first_and_duplicate_completed_request(
+    app: flask.Flask, job_name: str
+) -> None:
+    # First attempt succeeds and marks the record COMPLETED.
+    with app.test_request_context(
+        "/",
+        headers={
+            "X-CloudScheduler-JobName": job_name,
+            "X-CloudScheduler-ScheduleTime": SCHEDULE_TIME_UTC,
+        },
+    ):
+        body, status = deduplicate_job_requests.handle_scheduled_job(
+            flask.request
+        )
+        assert status == 200
+        assert body == "Job completed"
+
+    # Retried attempt with an equivalent RFC 3339 offset maps to the same UTC
+    # key and is ignored with 200 OK.
+    with app.test_request_context(
+        "/",
+        headers={
+            "X-CloudScheduler-JobName": job_name,
+            "X-CloudScheduler-ScheduleTime": SCHEDULE_TIME_OFFSET,
+        },
+    ):
+        body, status = deduplicate_job_requests.handle_scheduled_job(
+            flask.request
+        )
+        assert status == 200
+        assert body == "Duplicate job request ignored"
+
+
+def test_in_progress_request_returns_409(
+    app: flask.Flask, job_name: str
+) -> None:
+    dedup_key = f"{job_name}:{SCHEDULE_TIME_UTC}"
+    now = datetime.datetime.now(datetime.timezone.utc)
+    deduplicate_job_requests.db.collection("scheduler_dedup").document(
+        dedup_key
+    ).create(
+        {
+            "job_name": job_name,
+            "schedule_time": SCHEDULE_TIME_UTC,
+            "status": "IN_PROGRESS",
+            "created_at": now,
+            "expires_at": now + deduplicate_job_requests.LOCK_TTL,
+        }
+    )
+
+    with app.test_request_context(
+        "/",
+        headers={
+            "X-CloudScheduler-JobName": job_name,
+            "X-CloudScheduler-ScheduleTime": SCHEDULE_TIME_UTC,
+        },
+    ):
+        body, status = deduplicate_job_requests.handle_scheduled_job(
+            flask.request
+        )
+        assert status == 409
+        assert body == "Job execution already in progress"
+
+
+def test_failed_execution_releases_lock_for_retry(
+    app: flask.Flask, job_name: str
+) -> None:
+    headers = {
+        "X-CloudScheduler-JobName": job_name,
+        "X-CloudScheduler-ScheduleTime": SCHEDULE_TIME_UTC,
+    }
+
+    # If run_job_logic raises an exception, the lock document is deleted.
+    with mock.patch.object(
+        deduplicate_job_requests,
+        "run_job_logic",
+        side_effect=RuntimeError("Transient failure"),
+    ):
+        with app.test_request_context("/", headers=headers):
+            with pytest.raises(RuntimeError, match="Transient failure"):
+                deduplicate_job_requests.handle_scheduled_job(flask.request)
+
+    # A subsequent retry can acquire the lock and complete the job.
+    with app.test_request_context("/", headers=headers):
+        body, status = deduplicate_job_requests.handle_scheduled_job(
+            flask.request
+        )
+        assert status == 200
+        assert body == "Job completed"

cloud_scheduler/snippets/requirements.txt

@@ -1,4 +1,5 @@
 Flask==3.1.3; python_version >= '3.9'
 gunicorn==23.0.0
+google-cloud-firestore==2.19.0
 google-cloud-scheduler==2.14.1
 Werkzeug==3.1.9; python_version >= '3.10'