From 71eb63817f3f715e82adc0f6717a0cd413abda28 Mon Sep 17 00:00:00 2001 From: Jennifer Davis Date: Tue, 10 Nov 2026 17:29:12 +0000 Subject: [PATCH 1/2] fix(cloud_scheduler): key deduplication on the scheduled minute The deduplication and receiver samples identified a scheduled run by the raw X-CloudScheduler-ScheduleTime header. A forced run sent before a job's first scheduled run carries the schedule time without Cloud Scheduler's per-job offset of a few seconds, so the same run could get two keys and be processed twice. Key on the scheduled minute in UTC instead, and reject a schedule time that can't be parsed with HTTP 400. Add unit tests for a delivery with and without the offset, and for an invalid schedule time. --- .../deduplicate_job_requests/main.py | 12 ++++++--- .../tests/integration_test.py | 7 +++--- .../tests/main_test.py | 25 +++++++++++++++++++ cloud_scheduler/receiver/main.py | 12 +++++++-- cloud_scheduler/receiver/tests/main_test.py | 21 ++++++++++++++++ 5 files changed, 69 insertions(+), 8 deletions(-) diff --git a/cloud_scheduler/deduplicate_job_requests/main.py b/cloud_scheduler/deduplicate_job_requests/main.py index d29fcf0206..3c60c15ba7 100644 --- a/cloud_scheduler/deduplicate_job_requests/main.py +++ b/cloud_scheduler/deduplicate_job_requests/main.py @@ -85,9 +85,15 @@ def handle_scheduled_job(request: flask.Request) -> tuple[str, int]: 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() + # A forced run sent before the job's first scheduled run omits Cloud + # Scheduler's few-second offset, so key on the scheduled minute. A forced + # run uses the next scheduled time, so that scheduled run is then skipped. + try: + scheduled = datetime.datetime.fromisoformat(schedule_time) + except ValueError: + return "error: invalid X-CloudScheduler-ScheduleTime header", 400 + minute = scheduled.astimezone(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%MZ") + run_id = hashlib.sha256(f"{job_name}/{minute}".encode()).hexdigest() run_ref = db.collection("scheduler_runs").document(run_id) now = datetime.datetime.now(datetime.timezone.utc) attempt_id = uuid.uuid4().hex diff --git a/cloud_scheduler/deduplicate_job_requests/tests/integration_test.py b/cloud_scheduler/deduplicate_job_requests/tests/integration_test.py index cc7500cc9c..15035bdad6 100644 --- a/cloud_scheduler/deduplicate_job_requests/tests/integration_test.py +++ b/cloud_scheduler/deduplicate_job_requests/tests/integration_test.py @@ -42,10 +42,11 @@ 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']}" + scheduled = datetime.datetime.fromisoformat( + headers["X-CloudScheduler-ScheduleTime"] ) + minute = scheduled.astimezone(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%MZ") + key = f"{headers['X-CloudScheduler-JobName']}/{minute}" run_id = hashlib.sha256(key.encode()).hexdigest() return main.db.collection("scheduler_runs").document(run_id) diff --git a/cloud_scheduler/deduplicate_job_requests/tests/main_test.py b/cloud_scheduler/deduplicate_job_requests/tests/main_test.py index 6a643ff495..482aab0a20 100644 --- a/cloud_scheduler/deduplicate_job_requests/tests/main_test.py +++ b/cloud_scheduler/deduplicate_job_requests/tests/main_test.py @@ -186,6 +186,31 @@ def test_different_schedule_times_are_different_runs(fake_db: FakeDatabase) -> N assert len(fake_db.store) == 2 +def test_schedule_time_with_and_without_offset_is_one_run( + fake_db: FakeDatabase, +) -> None: + # A forced run before the job's first scheduled run omits the offset. + without_offset = dict( + HEADERS, **{"X-CloudScheduler-ScheduleTime": "2026-11-06T19:29:00-08:00"} + ) + call_function(without_offset) + + with mock.patch.object(main, "run_job_logic") as run_job_logic: + body, status = call_function(HEADERS) + + assert (body, status) == ("Duplicate run skipped", 200) + run_job_logic.assert_not_called() + + +def test_invalid_schedule_time_is_rejected() -> None: + body, status = call_function( + dict(HEADERS, **{"X-CloudScheduler-ScheduleTime": "soon"}) + ) + + assert status == 400 + assert body.startswith("error:") + + def test_missing_scheduler_headers_are_rejected() -> None: body, status = call_function() diff --git a/cloud_scheduler/receiver/main.py b/cloud_scheduler/receiver/main.py index e31317b294..cca8b18dc2 100644 --- a/cloud_scheduler/receiver/main.py +++ b/cloud_scheduler/receiver/main.py @@ -13,6 +13,7 @@ # limitations under the License. # [START cloudscheduler_receiver] +import datetime import json import threading @@ -51,8 +52,15 @@ def handle_scheduled_run(request: flask.Request) -> tuple[str, int]: if not job_name or not schedule_time: return "error: missing Cloud Scheduler headers", 400 - # Retries keep the same schedule time, so this identifies one run. - run_key = f"{job_name}/{schedule_time}" + # A forced run sent before the job's first scheduled run omits Cloud + # Scheduler's few-second offset, so key on the scheduled minute. A forced + # run uses the next scheduled time, so that scheduled run is then skipped. + try: + scheduled = datetime.datetime.fromisoformat(schedule_time) + except ValueError: + return "error: invalid X-CloudScheduler-ScheduleTime header", 400 + minute = scheduled.astimezone(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%MZ") + run_key = f"{job_name}/{minute}" # Check and claim in one step so concurrent duplicates can't both run. with claimed_runs_lock: diff --git a/cloud_scheduler/receiver/tests/main_test.py b/cloud_scheduler/receiver/tests/main_test.py index 74e9552c19..ba1aab5140 100644 --- a/cloud_scheduler/receiver/tests/main_test.py +++ b/cloud_scheduler/receiver/tests/main_test.py @@ -69,6 +69,27 @@ def test_failed_run_releases_claim_for_retry() -> None: assert body == "Run processed" +def test_schedule_time_with_and_without_offset_is_one_run() -> None: + # A forced run before the job's first scheduled run omits the offset. + call_function( + dict(HEADERS, **{"X-CloudScheduler-ScheduleTime": "2026-11-06T09:00:00.25Z"}) + ) + + body, status = call_function(HEADERS) + + assert status == 200 + assert body == "Duplicate run skipped" + + +def test_invalid_schedule_time_is_rejected() -> None: + body, status = call_function( + dict(HEADERS, **{"X-CloudScheduler-ScheduleTime": "soon"}) + ) + + assert status == 400 + assert body.startswith("error:") + + def test_missing_scheduler_headers_are_rejected() -> None: body, status = call_function() From 34fb9dd8be41e376f228548ecacffa91737793bc Mon Sep 17 00:00:00 2001 From: Jennifer Davis Date: Tue, 10 Nov 2026 18:02:57 +0000 Subject: [PATCH 2/2] fix(cloud_scheduler): shorten scheduled-minute comments Replace the three-line explanation in the deduplication and receiver samples with a single line, and drop redundant comments from the tests whose names already describe the behavior. --- cloud_scheduler/deduplicate_job_requests/main.py | 4 +--- cloud_scheduler/deduplicate_job_requests/tests/main_test.py | 1 - cloud_scheduler/receiver/main.py | 4 +--- cloud_scheduler/receiver/tests/main_test.py | 1 - 4 files changed, 2 insertions(+), 8 deletions(-) diff --git a/cloud_scheduler/deduplicate_job_requests/main.py b/cloud_scheduler/deduplicate_job_requests/main.py index 3c60c15ba7..3951882cc6 100644 --- a/cloud_scheduler/deduplicate_job_requests/main.py +++ b/cloud_scheduler/deduplicate_job_requests/main.py @@ -85,9 +85,7 @@ def handle_scheduled_job(request: flask.Request) -> tuple[str, int]: if not job_name or not schedule_time: return "error: missing Cloud Scheduler headers", 400 - # A forced run sent before the job's first scheduled run omits Cloud - # Scheduler's few-second offset, so key on the scheduled minute. A forced - # run uses the next scheduled time, so that scheduled run is then skipped. + # Key on the minute; some deliveries omit the few-second offset. try: scheduled = datetime.datetime.fromisoformat(schedule_time) except ValueError: diff --git a/cloud_scheduler/deduplicate_job_requests/tests/main_test.py b/cloud_scheduler/deduplicate_job_requests/tests/main_test.py index 482aab0a20..e5247a01ef 100644 --- a/cloud_scheduler/deduplicate_job_requests/tests/main_test.py +++ b/cloud_scheduler/deduplicate_job_requests/tests/main_test.py @@ -189,7 +189,6 @@ def test_different_schedule_times_are_different_runs(fake_db: FakeDatabase) -> N def test_schedule_time_with_and_without_offset_is_one_run( fake_db: FakeDatabase, ) -> None: - # A forced run before the job's first scheduled run omits the offset. without_offset = dict( HEADERS, **{"X-CloudScheduler-ScheduleTime": "2026-11-06T19:29:00-08:00"} ) diff --git a/cloud_scheduler/receiver/main.py b/cloud_scheduler/receiver/main.py index cca8b18dc2..d2e9c076d8 100644 --- a/cloud_scheduler/receiver/main.py +++ b/cloud_scheduler/receiver/main.py @@ -52,9 +52,7 @@ def handle_scheduled_run(request: flask.Request) -> tuple[str, int]: if not job_name or not schedule_time: return "error: missing Cloud Scheduler headers", 400 - # A forced run sent before the job's first scheduled run omits Cloud - # Scheduler's few-second offset, so key on the scheduled minute. A forced - # run uses the next scheduled time, so that scheduled run is then skipped. + # Key on the minute; some deliveries omit the few-second offset. try: scheduled = datetime.datetime.fromisoformat(schedule_time) except ValueError: diff --git a/cloud_scheduler/receiver/tests/main_test.py b/cloud_scheduler/receiver/tests/main_test.py index ba1aab5140..ab7ac32dfe 100644 --- a/cloud_scheduler/receiver/tests/main_test.py +++ b/cloud_scheduler/receiver/tests/main_test.py @@ -70,7 +70,6 @@ def test_failed_run_releases_claim_for_retry() -> None: def test_schedule_time_with_and_without_offset_is_one_run() -> None: - # A forced run before the job's first scheduled run omits the offset. call_function( dict(HEADERS, **{"X-CloudScheduler-ScheduleTime": "2026-11-06T09:00:00.25Z"}) )