diff --git a/cloud_scheduler/deduplicate_job_requests/main.py b/cloud_scheduler/deduplicate_job_requests/main.py index d29fcf0206..3951882cc6 100644 --- a/cloud_scheduler/deduplicate_job_requests/main.py +++ b/cloud_scheduler/deduplicate_job_requests/main.py @@ -85,9 +85,13 @@ 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() + # Key on the minute; some deliveries omit the few-second offset. + 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..e5247a01ef 100644 --- a/cloud_scheduler/deduplicate_job_requests/tests/main_test.py +++ b/cloud_scheduler/deduplicate_job_requests/tests/main_test.py @@ -186,6 +186,30 @@ 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: + 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..d2e9c076d8 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,13 @@ 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}" + # Key on the minute; some deliveries omit the few-second offset. + 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..ab7ac32dfe 100644 --- a/cloud_scheduler/receiver/tests/main_test.py +++ b/cloud_scheduler/receiver/tests/main_test.py @@ -69,6 +69,26 @@ 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: + 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()