diff --git a/cloud_scheduler/snippets/deduplicate_job_requests.py b/cloud_scheduler/snippets/deduplicate_job_requests.py new file mode 100644 index 0000000000..c0b50b08a9 --- /dev/null +++ b/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] 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..4ff455ec10 --- /dev/null +++ b/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" 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'