[iennae]

Description

Adds cloudscheduler_deduplicate_job_requests, a Cloud Run function that does a Cloud Scheduler job's work once per scheduled run, even when Cloud Scheduler delivers the same run more than once. This rewrites #14702 by @camiekim, who is credited as co-author, and keeps that PR's region tag so the in-progress "Deduplicate job requests" docs page doesn't need to change.

How it works

• Run identity: The job name plus X-CloudScheduler-ScheduleTime, hashed into a Firestore document ID. The schedule time stays the same across retries, so the header is used exactly as received.
• Transactional claim with a lease:
  • A completed run is skipped and returns 200.
  • A run whose claim is still active returns 409, so retries continue in case that attempt fails.
  • A run whose claim has expired, for example after an attempt crashed, can be claimed again.
• Failure handling: If the work raises an exception, the function deletes the claim and re-raises, so Cloud Scheduler retries the run.
• Cleanup: Records include an expire_at field for a Firestore TTL policy.

Changes from #14702

• Adds a lease, so a crashed attempt no longer blocks retries until TTL cleanup, which can take more than 24 hours. The check runs in a transaction, which avoids the race condition raised in that PR's review.
• Moves the sample to its own directory, cloud_scheduler/deduplicate_job_requests/, as a functions-framework function with its own requirements (google-cloud-firestore 2.34.1), instead of adding Flask and Firestore to snippets/.
• Hashes the run key instead of validating the job name and normalizing timestamps.
• Uses shorter comments. Target-specific details belong on the docs page.

Scope. In testing, HTTP targets received X-CloudScheduler-JobName (the job ID) and X-CloudScheduler-ScheduleTime. The API reference documents the same headers for App Engine targets, so this sample covers both. Pub/Sub messages from Cloud Scheduler carry no run identifier, and Google API targets such as Workflows have no user handler, so those targets are better covered on the docs page. A forced run sends the next scheduled time, so header-based deduplication then skips that scheduled run. A comment in the sample notes this.

Tests (in tests/)

• Unit tests with an in-memory fake Firestore cover these cases, and run without credentials:
  • first delivery
  • duplicate after completion
  • active lease
  • expired lease
  • failed work
  • separate runs
  • missing headers
• Integration tests run the handler against a real Firestore database in Native mode. They're skipped unless RUN_INTEGRATION_TESTS=true; set FIRESTORE_DATABASE to use a named database.

Checklist

Testing

• I have tested this change on a live environment and verified it works as intended. Unit and integration tests pass on Python 3.13, including against a live Firestore database.

Compliance & Style

• I have followed the Sample Guidelines.
• The README is updated with all relevant information (if applicable). Not applicable.
• If this PR adds a new sample directory: I updated the CODEOWNERS file. Not applicable: the new directory is inside cloud_scheduler, which the default owners cover.

---

Post-Approval Actions

• Please merge this PR for me once it is approved

[snippet-bot]

Here is the summary of changes.

You are about to add 1 region tag.

• cloud_scheduler/deduplicate_job_requests/main.py:15, tag cloudscheduler_deduplicate_job_requests

---

This comment is generated by snippet-bot.
If you find problems with this result, please file an issue at:
https://github.com/googleapis/repo-automation-bots/issues.
To update this comment, add snippet-bot:force-run label or use the checkbox below:

• Refresh this comment

[gemini-code-assist]

Code Review

This pull request introduces a Cloud Function to deduplicate Cloud Scheduler job requests using Firestore for distributed locking and lease management. The review feedback highlights a potential race condition where a slow worker whose lease has expired could inadvertently overwrite or delete a lease claimed by a subsequent retry. To address this, the reviewer suggests using the lease expiration timestamp as a fencing token and performing both the completion and release operations transactionally, providing concrete code suggestions for the implementation and corresponding test updates.

[gemini-code-assist]

To prevent race conditions where a slow worker's attempt completes or fails after its lease has expired and been claimed by another worker, we should perform both the completion and release operations transactionally. By using the lease_expires timestamp as a fencing token, we can ensure that we only update or delete the document if our specific attempt still holds the active lease.

Let's update claim_run to accept lease_expires and add transactional helper functions complete_run and release_run.

@firestore.transactional
def claim_run(
    transaction: firestore.Transaction,
    run_ref: firestore.DocumentReference,
    now: datetime.datetime,
    lease_expires: datetime.datetime,
) -> str:
    """Claims the run unless it's done or another attempt holds the lease."""
    snapshot = run_ref.get(transaction=transaction)
    if snapshot.exists:
        run = snapshot.to_dict()
        if run["status"] == "completed":
            return "completed"
        if run["lease_expires"] > now:
            return "in_progress"
    transaction.set(
        run_ref,
        {
            "status": "in_progress",
            "lease_expires": lease_expires,
            "expire_at": now + RETENTION,
        },
    )
    return "claimed"


@firestore.transactional
def complete_run(
    transaction: firestore.Transaction,
    run_ref: firestore.DocumentReference,
    lease_expires: datetime.datetime,
) -> None:
    """Marks the run as completed if the lease is still held by this attempt."""
    snapshot = run_ref.get(transaction=transaction)
    if snapshot.exists:
        run = snapshot.to_dict()
        if run.get("lease_expires") == lease_expires and run.get("status") == "in_progress":
            transaction.update(run_ref, {"status": "completed"})


@firestore.transactional
def release_run(
    transaction: firestore.Transaction,
    run_ref: firestore.DocumentReference,
    lease_expires: datetime.datetime,
) -> None:
    """Releases the claim if the lease is still held by this attempt."""
    snapshot = run_ref.get(transaction=transaction)
    if snapshot.exists:
        run = snapshot.to_dict()
        if run.get("lease_expires") == lease_expires and run.get("status") == "in_progress":
            transaction.delete(run_ref)

References

1. When handling scheduled runs, check and claim the run atomically under a lock before performing the work. If the work raises an exception, ensure the claim is released so that subsequent retries can process the run.

[gemini-code-assist]

Update handle_scheduled_job to calculate lease_expires and use the new transactional complete_run and release_run helpers. This ensures that if this invocation runs slowly and its lease expires, it won't accidentally overwrite or delete a lease claimed by a subsequent retry.

Suggested change

	now = datetime.datetime.now(datetime.timezone.utc)
	status = claim_run(db.transaction(), run_ref, now)
	if status == "completed":
	return "Duplicate run skipped", 200
	if status == "in_progress":
	# A non-2xx response keeps retries going in case that attempt fails.
	return "Run in progress", 409
	try:
	run_job_logic()
	except Exception:
	run_ref.delete() # Lets a retry claim the run.
	raise
	run_ref.update({"status": "completed"})
	return "Run processed", 200
	now = datetime.datetime.now(datetime.timezone.utc)
	lease_expires = now + LEASE
	status = claim_run(db.transaction(), run_ref, now, lease_expires)
	if status == "completed":
	return "Duplicate run skipped", 200
	if status == "in_progress":
	# A non-2xx response keeps retries going in case that attempt fails.
	return "Run in progress", 409
	try:
	run_job_logic()
	except Exception:
	release_run(db.transaction(), run_ref, lease_expires) # Lets a retry claim the run.
	raise
	complete_run(db.transaction(), run_ref, lease_expires)
	return "Run processed", 200

References

1. When handling scheduled runs, check and claim the run atomically under a lock before performing the work. If the work raises an exception, ensure the claim is released so that subsequent retries can process the run.

[gemini-code-assist]

To support the transactional update and delete operations in our unit tests, we need to mock these methods on the FakeDatabase transaction object.

Suggested change

	def transaction(self) -> mock.MagicMock:
	# firestore.transactional begins, commits, and retries the transaction.
	transaction = mock.MagicMock(_max_attempts=1, _id=b"test", _read_only=False)
	transaction.set.side_effect = lambda ref, data: self.store.__setitem__(
	ref.id, dict(data)
	)
	return transaction
	def transaction(self) -> mock.MagicMock:
	# firestore.transactional begins, commits, and retries the transaction.
	transaction = mock.MagicMock(_max_attempts=1, _id=b"test", _read_only=False)
	transaction.set.side_effect = lambda ref, data: self.store.__setitem__(
	ref.id, dict(data)
	)
	transaction.update.side_effect = lambda ref, data: self.store[ref.id].update(data)
	transaction.delete.side_effect = lambda ref: self.store.pop(ref.id, None)
	return transaction