Google Cloud Tasks Execution¶

CloudTasksExecutionBackend (configured via CloudTasksExecutionConfig) provides scale-to-zero HTTP push execution on Google Cloud Run without running an always-on polling worker process.

ID-Only Wire Contract & At-Least-Once Delivery¶

Every Cloud Tasks HTTP delivery body carries only the protocol version and the record's id:

{
  "version": 1,
  "task_id": "2c8d8f72-6a4b-4c5c-9c82-5e3f21a09b41"
}

Task arguments, keyword arguments, result payloads, retry state, and metadata stay in your storage backend and never cross the network inside the Cloud Tasks message. When Cloud Tasks posts the delivery payload to your consumer service, the route handler delegates to consume_one(service, task_id), which claims the authoritative record from storage by ID and executes the attempt.

Google Cloud Tasks provides an at-least-once delivery guarantee. Although consume_one() fences task claims on expected_retry_count in your storage backend, transient network drops after external side effects can trigger redelivery or retry, so task handlers with external side effects must be idempotent.

Runnable Configuration Example¶

"""Configuring CloudTasksExecutionConfig for consumer and producer-only services."""

from litestar_queues import QueueConfig, WorkerConfig
from litestar_queues.backends.redis import RedisBackendConfig
from litestar_queues.execution.cloudtasks import CloudTasksExecutionConfig

DISPATCH_DEADLINE_SECONDS = 900


def build_consumer_cloud_tasks_config() -> QueueConfig:
    """Configure a Cloud Run service that both enqueues and receives Cloud Tasks HTTP deliveries."""
    return QueueConfig(
        queue_backend=RedisBackendConfig(url="redis://redis.internal:6379/0"),
        execution_backend=CloudTasksExecutionConfig(
            project_id="my-gcp-project",
            location="us-central1",
            queue_id="primary-tasks",
            service_url="https://worker-service-uc.a.run.app",
            service_account_email="cloud-tasks-invoker@my-gcp-project.iam.gserviceaccount.com",
            dispatch_deadline=DISPATCH_DEADLINE_SECONDS,
            response_margin=30.0,
            default_task_timeout=840.0,
            mount_route=True,
            trust_platform_auth=True,
            verify_oidc=True,
        ),
        worker=WorkerConfig(placement="external", stale_after=930.0),
    )


def build_producer_only_cloud_tasks_config() -> QueueConfig:
    """Configure a public web tier that dispatches to Cloud Tasks without mounting the delivery route."""
    return QueueConfig(
        queue_backend=RedisBackendConfig(url="redis://redis.internal:6379/0"),
        execution_backend=CloudTasksExecutionConfig(
            project_id="my-gcp-project",
            location="us-central1",
            queue_id="primary-tasks",
            service_url="https://worker-service-uc.a.run.app",
            service_account_email="cloud-tasks-invoker@my-gcp-project.iam.gserviceaccount.com",
            mount_route=False,
        ),
        worker=WorkerConfig(placement="external"),
    )


async def main() -> None:
    """Verify consumer and producer-only CloudTasksExecutionConfig settings."""
    consumer = build_consumer_cloud_tasks_config()
    producer = build_producer_only_cloud_tasks_config()
    assert isinstance(consumer.execution_backend, CloudTasksExecutionConfig)
    assert isinstance(producer.execution_backend, CloudTasksExecutionConfig)
    assert consumer.execution_backend.mount_route is True
    assert consumer.execution_backend.verify_oidc is True
    assert consumer.execution_backend.dispatch_deadline == DISPATCH_DEADLINE_SECONDS
    assert producer.execution_backend.mount_route is False

Provisioning Cloud Tasks & Cloud Run¶

The package never creates queues, Cloud Run services, or IAM bindings at runtime; operators provision them via Terraform or gcloud before enqueuing records.

Create the Cloud Tasks Queue¶

gcloud tasks queues create primary-tasks \
  --location=us-central1 \
  --max-dispatches-per-second=50 \
  --max-concurrent-dispatches=100 \
  --max-attempts=5 \
  --min-backoff=10s \
  --max-backoff=300s

Deploy the Cloud Run Consumer Service¶

Because Cloud Tasks pushes each task as an active HTTP request, the Cloud Run service can scale to zero when idle and keep standard request-scoped CPU allocation (--cpu-throttling):

gcloud run deploy worker-service \
  --image=us-central1-docker.pkg.dev/my-gcp-project/apps/worker:latest \
  --region=us-central1 \
  --no-allow-unauthenticated \
  --cpu-throttling \
  --concurrency=8 \
  --min-instances=0 \
  --timeout=900

Required IAM Roles¶

Route Mounting & OIDC Authentication¶

When CloudTasksExecutionConfig.mount_route=True (the default), QueuePlugin registers the delivery endpoint at /_litestar-queues/cloud-tasks (or /_<namespace>/cloud-tasks when a custom QueueNamespace is configured) with opt={"exclude_from_auth": True, "exclude_from_csrf": True} so application-wide session or CSRF middleware does not intercept Cloud Tasks deliveries.

To prevent unauthenticated delivery spoofing, CloudTasksExecutionConfig requires at least one route protection layer when mount_route=True:

1. In-Process OIDC Verification (verify_oidc=True): Attaches CloudTasksOIDCGuard, which verifies the Authorization: Bearer <token> JWT signature against Google's certs, checks that the token aud matches audience (defaulting to the service_url origin), and verifies that email matches service_account_email.
2. Platform IAM (trust_platform_auth=True): Accepts Cloud Run's --no-allow-unauthenticated IAM perimeter (Combine both trust_platform_auth=True and verify_oidc=True for defense in depth).
3. Custom Litestar Guards (guards=(...)): Attaches application-supplied Litestar Guard callables to the route.

For public web tiers that only enqueue tasks and never execute them, set mount_route=False.

Delivery HTTP Response Semantics¶

The /_litestar-queues/cloud-tasks endpoint translates consume_one() outcomes into deterministic HTTP status codes understood by Cloud Tasks:

Retry Layers & Inline Stale Recovery¶

1. Application Task Retries (@task(retries=..., retry_backoff=...)): When a task raises a Python exception, QueueService catches it, increments retry_count in storage, schedules a new Cloud Task at scheduled_at = now + backoff, and returns 204 No Content so the completed HTTP delivery is acknowledged cleanly.
2. Transport Redelivery & Inline Stale Recovery (WorkerConfig.stale_after): If a Cloud Run container crashes or hits an OOM kill while a task is running, Cloud Tasks receives a 5xx or connection reset and redelivers the same {version, task_id} payload. When WorkerConfig.stale_after is configured on the consumer service, the delivery route automatically checks whether the existing running claim's heartbeat_at has exceeded stale_after and recovers the stale claim inline before executing the redelivered attempt.

Deadlines & the 30-Day Scheduling Horizon¶

• Timeout Budget Equation: Keep task timeouts, response margins, Cloud Tasks dispatch deadlines (15 seconds to 1800 seconds), and Cloud Run request timeouts ordered as:

$$\text{task.timeout} + \text{response_margin} \le \text{dispatch_deadline} < \text{Cloud Run } \texttt{--timeout}$$

• 30-Day Horizon (CLOUD_TASKS_MAX_SCHEDULE_HORIZON): Google Cloud Tasks accepts schedule_time up to 30 days in the future. Enqueuing a task or recurring schedule whose scheduled_at exceeds 30 days on "cloudtasks" raises QueueConfigurationError before mutating active schedule records.

Cost Model & Scale-to-Zero Trade-Offs (2026)¶

Each successful task execution performs two operations in Cloud Tasks (one create and one HTTP delivery), billed after the monthly free tier while letting the Cloud Run consumer service scale to zero idle instances (--min-instances=0 --cpu-throttling). Note that scaling the Cloud Run worker tier to zero removes idle compute charges for the worker container, not the always-on cost of the backing database (such as Cloud SQL or Memorystore) unless paired with a serverless database tier.

Hybrid Routing with Cloud Run Jobs¶

Because Cloud Tasks caps dispatch_deadline at 30 minutes (1800 seconds), combine CloudTasksExecutionConfig and CloudRunExecutionConfig in QueueConfig.execution_backends to run fast tasks over HTTP push and multi-hour jobs on Cloud Run Jobs:

config = QueueConfig(
    queue_backend=backend_config,
    execution_backend=cloud_tasks_config,
    execution_backends=(cloud_tasks_config, cloud_run_config),
    worker=WorkerConfig(placement="external"),
)

Delivery Repair & Remote Cancellation¶

• Two-Phase Enqueue Reservation & Repair: QueueService.enqueue() writes an external-dispatch:<uuid> reservation sentinel on QueuedTaskRecord.execution_ref before calling the Cloud Tasks API, then finalizes execution_ref with the created Cloud Task resource path. If a producer process crashes between storage commit and the Cloud Tasks API call, litestar queues run-maintenance --phase external (or QueueService.reconcile_external_tasks()) performs maintenance repair for the missing Cloud Task automatically.
• Remote Cancellation: Calling await queue_service.cancel_task(task_id) deletes the scheduled Cloud Task in Google Cloud Tasks (delete_task) before marking the storage record cancelled.
• Local Testing Without an Emulator: Google does not ship an official Cloud Tasks emulator. Because Cloud Tasks routes through consume_one(service, task_id), unit and integration tests can test the exact execution path with execution_backend="local" and await consume_one(service, handle.id) without running a Cloud Tasks emulator.

Choosing Between Google Cloud Execution Products¶