feat(cloud_scheduler): add sample that prevents duplicate work from job requests

cloud_scheduler/deduplicate_job_requests/.python-version

@@ -0,0 +1 @@
+3.13

cloud_scheduler/deduplicate_job_requests/__init__.py

@@ -0,0 +1,13 @@
+# 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.

cloud_scheduler/deduplicate_job_requests/main.py

@@ -0,0 +1,90 @@
+# 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.
+
+# [START cloudscheduler_deduplicate_job_requests]
+import datetime
+import hashlib
+
+import flask
+import functions_framework
+from google.cloud import firestore
+
+# Match the job's attempt deadline, so a crashed attempt's claim expires.
+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,
+    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",
+            "lease_expires": now + LEASE,
+            "expire_at": now + RETENTION,
+        },
+    )
+    return "claimed"
+
+
+@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)
+
+    status = claim_run(db.transaction(), run_ref, 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:
+        run_ref.delete()  # Lets a retry claim the run.
+        raise
+    run_ref.update({"status": "completed"})
+    return "Run processed", 200
+
+
+# [END cloudscheduler_deduplicate_job_requests]

cloud_scheduler/deduplicate_job_requests/noxfile_config.py

@@ -0,0 +1,42 @@
+# 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.
+
+# Default TEST_CONFIG_OVERRIDE for python repos.
+
+# You can copy this file into your directory, then it will be imported from
+# the noxfile.py.
+
+# The source of truth:
+# https://github.com/GoogleCloudPlatform/python-docs-samples/blob/main/noxfile_config.py
+
+TEST_CONFIG_OVERRIDE = {
+    # You can opt out from the test for specific Python versions.
+    "ignored_versions": ["2.7", "3.7", "3.8", "3.9", "3.10", "3.11", "3.12"],
+    # Old samples are opted out of enforcing Python type hints
+    # All new samples should feature them
+    "enforce_type_hints": True,
+    # An envvar key for determining the project id to use. Change it
+    # to 'BUILD_SPECIFIC_GCLOUD_PROJECT' if you want to opt in using a
+    # build specific Cloud project. You can also use your own string
+    # to use your own Cloud project.
+    "gcloud_project_env": "GOOGLE_CLOUD_PROJECT",
+    # 'gcloud_project_env': 'BUILD_SPECIFIC_GCLOUD_PROJECT',
+    # If you need to use a specific version of pip,
+    # change pip_version_override to the string representation
+    # of the version number, for example, "20.2.4"
+    "pip_version_override": None,
+    # A dictionary you want to inject into your test. Don't put any
+    # secrets here. These values will override predefined values.
+    "envs": {},
+}

cloud_scheduler/deduplicate_job_requests/requirements-test.txt

@@ -0,0 +1 @@
+pytest==9.0.3

cloud_scheduler/deduplicate_job_requests/requirements.txt

@@ -0,0 +1,2 @@
+functions-framework==3.10.2
+google-cloud-firestore==2.34.1

cloud_scheduler/deduplicate_job_requests/tests/__init__.py

@@ -0,0 +1,13 @@
+# 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.

cloud_scheduler/deduplicate_job_requests/tests/conftest.py

@@ -0,0 +1,22 @@
+# 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 os
+from unittest import mock
+
+# main.py creates a Firestore client when it's imported. Without credentials
+# that fails, so unit-test runs replace the client class. The unit tests then
+# swap in a fake database, and the integration tests use a real one.
+if os.environ.get("RUN_INTEGRATION_TESTS") != "true":
+    mock.patch("google.cloud.firestore.Client").start()

cloud_scheduler/deduplicate_job_requests/tests/integration_test.py

@@ -0,0 +1,95 @@
+# 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.
+
+"""Integration tests that run the handler against a real Firestore database.
+
+These tests are skipped unless RUN_INTEGRATION_TESTS is "true". They need a
+Firestore database in Native mode; set FIRESTORE_DATABASE to use a named
+database instead of "(default)".
+"""
+
+import datetime
+import hashlib
+import os
+from typing import Iterator
+from unittest import mock
+import uuid
+
+import flask
+from google.cloud import firestore
+import pytest
+
+from .. import main
+
+pytestmark = pytest.mark.skipif(
+    os.environ.get("RUN_INTEGRATION_TESTS") != "true",
+    reason="Set RUN_INTEGRATION_TESTS=true to run integration tests.",
+)
+
+app = flask.Flask(__name__)
+
+
+def run_ref(headers: dict[str, str]) -> firestore.DocumentReference:
+    """Returns the record that the handler keeps for a run."""
+    key = (
+        f"{headers['X-CloudScheduler-JobName']}/"
+        f"{headers['X-CloudScheduler-ScheduleTime']}"
+    )
+    run_id = hashlib.sha256(key.encode()).hexdigest()
+    return main.db.collection("scheduler_runs").document(run_id)
+
+
+@pytest.fixture
+def headers(monkeypatch: pytest.MonkeyPatch) -> Iterator[dict[str, str]]:
+    """Uses a real database and a unique run, and deletes the run's record."""
+    database = firestore.Client(
+        database=os.environ.get("FIRESTORE_DATABASE", "(default)")
+    )
+    monkeypatch.setattr(main, "db", database)
+    run_headers = {
+        "X-CloudScheduler-JobName": f"test-job-{uuid.uuid4().hex[:12]}",
+        "X-CloudScheduler-ScheduleTime": "2026-11-06T19:29:01.336893-08:00",
+    }
+    yield run_headers
+    run_ref(run_headers).delete()
+
+
+def call_function(headers: dict[str, str]) -> tuple[str, int]:
+    with app.test_request_context("/", method="POST", headers=headers):
+        return main.handle_scheduled_job(flask.request)
+
+
+def test_duplicate_delivery_is_skipped(headers: dict[str, str]) -> None:
+    assert call_function(headers) == ("Run processed", 200)
+    assert call_function(headers) == ("Duplicate run skipped", 200)
+    assert run_ref(headers).get().to_dict()["status"] == "completed"
+
+
+def test_failed_work_releases_the_claim(headers: dict[str, str]) -> None:
+    with mock.patch.object(main, "run_job_logic", side_effect=RuntimeError("boom")):
+        with pytest.raises(RuntimeError):
+            call_function(headers)
+
+    assert not run_ref(headers).get().exists
+    assert call_function(headers) == ("Run processed", 200)
+
+
+def test_active_and_expired_leases(headers: dict[str, str]) -> None:
+    now = datetime.datetime.now(datetime.timezone.utc)
+    run_ref(headers).set({"status": "in_progress", "lease_expires": now + main.LEASE})
+    assert call_function(headers) == ("Run in progress", 409)
+
+    expired = now - datetime.timedelta(seconds=1)
+    run_ref(headers).set({"status": "in_progress", "lease_expires": expired})
+    assert call_function(headers) == ("Run processed", 200)