From 4528b457987910263673025c4cff945b969a2016 Mon Sep 17 00:00:00 2001 From: Camie Kim Date: Fri, 30 Oct 2026 08:32:22 -0400 Subject: [PATCH 1/6] docs: Add deduplication logic for Cloud Scheduler jobs This script demonstrates how to deduplicate retried Cloud Scheduler HTTP requests using Firestore, ensuring that duplicate job requests are handled appropriately. Per b/574556578 --- cloud_scheduler/snippets/deduplicate_job.py | 124 ++++++++++++++++++++ 1 file changed, 124 insertions(+) create mode 100644 cloud_scheduler/snippets/deduplicate_job.py diff --git a/cloud_scheduler/snippets/deduplicate_job.py b/cloud_scheduler/snippets/deduplicate_job.py new file mode 100644 index 0000000000..db49ed59cf --- /dev/null +++ b/cloud_scheduler/snippets/deduplicate_job.py @@ -0,0 +1,124 @@ +# 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.""" + +# [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] From bb0b6b13d03e24bc88399a0813cec5cd81550906 Mon Sep 17 00:00:00 2001 From: Camie Kim Date: Fri, 30 Oct 2026 08:35:35 -0400 Subject: [PATCH 2/6] docs: Implement tests for job request deduplication Add unit tests for deduplicating job requests handling. --- .../snippets/deduplicate_job_requests_test.py | 167 ++++++++++++++++++ 1 file changed, 167 insertions(+) create mode 100644 cloud_scheduler/snippets/deduplicate_job_requests_test.py diff --git a/cloud_scheduler/snippets/deduplicate_job_requests_test.py b/cloud_scheduler/snippets/deduplicate_job_requests_test.py new file mode 100644 index 0000000000..a1954cbdda --- /dev/null +++ b/cloud_scheduler/snippets/deduplicate_job_requests_test.py @@ -0,0 +1,167 @@ +# 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 flask +import pytest + +import deduplicate_job_requests + +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" From c25ced36f69854acf6184f9e0d6bb7b4de57c532 Mon Sep 17 00:00:00 2001 From: Camie Kim Date: Fri, 30 Oct 2026 08:36:32 -0400 Subject: [PATCH 3/6] Rename deduplicate_job.py to deduplicate_job_requests.py --- .../snippets/{deduplicate_job.py => deduplicate_job_requests.py} | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename cloud_scheduler/snippets/{deduplicate_job.py => deduplicate_job_requests.py} (100%) diff --git a/cloud_scheduler/snippets/deduplicate_job.py b/cloud_scheduler/snippets/deduplicate_job_requests.py similarity index 100% rename from cloud_scheduler/snippets/deduplicate_job.py rename to cloud_scheduler/snippets/deduplicate_job_requests.py From ebe13267bfd0e0fa5eb322828682591e0c5e796d Mon Sep 17 00:00:00 2001 From: Camie Kim Date: Fri, 30 Oct 2026 08:45:39 -0400 Subject: [PATCH 4/6] docs: Add google-cloud-firestore to requirements --- cloud_scheduler/snippets/requirements.txt | 1 + 1 file changed, 1 insertion(+) diff --git a/cloud_scheduler/snippets/requirements.txt b/cloud_scheduler/snippets/requirements.txt index 0d8b19c3f0..cf9a7e190f 100644 --- a/cloud_scheduler/snippets/requirements.txt +++ b/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' From df6aa3a0b6ce63a011d4e9db75a52b3f63ddd2a2 Mon Sep 17 00:00:00 2001 From: Camie Kim Date: Fri, 30 Oct 2026 09:21:39 -0400 Subject: [PATCH 5/6] Rearrange import statements in test file Reordered import statements for consistency. --- cloud_scheduler/snippets/deduplicate_job_requests_test.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/cloud_scheduler/snippets/deduplicate_job_requests_test.py b/cloud_scheduler/snippets/deduplicate_job_requests_test.py index a1954cbdda..4ff455ec10 100644 --- a/cloud_scheduler/snippets/deduplicate_job_requests_test.py +++ b/cloud_scheduler/snippets/deduplicate_job_requests_test.py @@ -17,11 +17,10 @@ from unittest import mock import uuid +import deduplicate_job_requests import flask import pytest -import deduplicate_job_requests - SCHEDULE_TIME_UTC = "2026-10-16T12:00:00Z" SCHEDULE_TIME_OFFSET = "2026-10-16T05:00:00-07:00" From ee1185a6c463d6325fa5043f335b9f5663cb5aa7 Mon Sep 17 00:00:00 2001 From: Camie Kim Date: Fri, 30 Oct 2026 09:46:42 -0400 Subject: [PATCH 6/6] Update docstring to include documentation link Added a link to the Cloud Scheduler documentation for deduplication. --- cloud_scheduler/snippets/deduplicate_job_requests.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/cloud_scheduler/snippets/deduplicate_job_requests.py b/cloud_scheduler/snippets/deduplicate_job_requests.py index db49ed59cf..c0b50b08a9 100644 --- a/cloud_scheduler/snippets/deduplicate_job_requests.py +++ b/cloud_scheduler/snippets/deduplicate_job_requests.py @@ -12,7 +12,9 @@ # See the License for the specific language governing permissions and # limitations under the License. -"""Demonstrates how to deduplicate retried Cloud Scheduler HTTP requests.""" +"""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