fix(cloud_scheduler): key deduplication on the scheduled minute

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

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)
 

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()
 

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:

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()