Skip to main content

Pub/Sub triggers

A Pub/Sub trigger subscribes to a topic — a named stream of messages — in your project, and pushes each published message to your function. Publish a message, the platform delivers it, your handler runs. Failed deliveries are retried; a message that keeps failing is parked in a dead-letter topic instead of being lost.

This page is the function-side guide. The end-to-end worked tutorial is event-driven functions; the messaging service itself is Pub/Sub.

What your handler receives​

Delivery is a POST in CloudEvents structured mode, which is the default and what you get unless you asked for something else. There are no ce-* headers. The whole event — envelope and payload together — is one JSON body under Content-Type: application/cloudevents+json, and the payload rides base64-encoded in data_base64, because a published message is arbitrary bytes and may not be valid UTF-8, let alone valid JSON:

POST / HTTP/1.1
Content-Type: application/cloudevents+json; charset=utf-8

{"specversion":"1.0",
 "type":"ai.crusoe.pubsub.message.v1",
 "source":"//pubsub.crusoe.ai/topics/orders",
 "id":"8f1c...",
 "time":"2026-08-27T14:03:00.123456789Z",
 "subject":"orders",
 "datacontenttype":"application/json",
 "subscription":"order-ingest",
 "deliveryattempt":1,
 "data_base64":"eyJvcmRlcl9pZCI6Im8tMTIzIiwidG90YWwiOjQyLjV9"}

Do not decode that yourself. The shim unwraps the envelope before calling your handler, in all four runtimes. Publish a JSON object and your handler's event is that object — {"order_id":"o-123","total":42.50}. Publish a bare string or a number and it arrives as event["data"]. Publish bytes that are not valid UTF-8 and it arrives as event["data_base64"], still encoded, for you to decode. A handler that reads event["data_base64"] on an ordinary JSON message finds nothing there.

The content type of the JSON body is what marks this a CloudEvent to the shim, so the CloudEvent rules apply: your return value is discarded, the shim ACKs 204 on success, and a handler exception becomes a 400 — which the subscription treats as terminal for that delivery attempt per the trigger retry rules.

So the handler contract for a Pub/Sub-fired function is: read the body, do the work, log what you need to see, raise only when you mean it.

Binary mode is opt-in, and only then do ce-* headers exist

Create the subscription with --push-content-mode cloudevents-binary and delivery flips: the payload becomes the raw body, and the envelope moves into headers — ce-id, ce-type, ce-source, ce-time, ce-subject, ce-subscription, ce-deliveryattempt, plus each publisher attribute as ce-attr-<name>. On the Python runtime those reach your handler as event["_cloudevent"]; Node.js, Go and Ruby drop them. A handler written against ce-* headers on a default subscription sees none of them, because that subscription is structured.

Note the trap on the other side: in binary mode the raw body is delivered as-is, so a message that is not JSON reaches a Node.js, Go or Ruby function as a 400 from the shim before your code runs. See HTTP and events.

subject is the topic name, not the message key, and source is //pubsub.crusoe.ai/topics/<topic>. deliveryattempt starts at 1 and counts up on redelivery, which is how a handler recognises a retry.

A complete worked example​

The function — one log line per order, observable through --history after the instance scales to zero. This is the event-logger example pattern extended with real work:

import json

def handle(event: dict) -> dict:
    order_id = event.get("order_id")
    print(f"event-logger received: {json.dumps(event)}")
    # ... process order_id ...
    return {"statusCode": 200, "body": "logged"}   # discarded on topic deliveries

Deploy, create the topic, wire the trigger:

platformctl functions deploy ./my-function --name order-worker

platformctl serverless triggers create order-ingest \
  --target order-worker \
  --type pubsub \
  --topic orders \
  --subscription ingest-orders

Then publish a message and watch it land:

# publish (any publisher works; the topic is the interface)
# ... then read what the function did:
platformctl logs order-worker --history

Fields that matter​

  • topic is required, and must be a topic in this project. Name one that does not exist and the trigger stays pending, with the reason written on the trigger's message: no topic named "orders" in this project.
  • subscription is optional. Leave it blank and the subscription takes the trigger's own name. Each subscription keeps its own place in the stream, so two triggers on one topic each read every message — they do not split the work. That is the fan-out pattern: one topic, a logging function and a processing function.
  • Suspending does not lose messages. Suspend a trigger and its subscription switches to pull mode: nothing is delivered until someone asks, messages pile up, and they drain when you resume.

Retries and dead-lettering​

The retry table is the trigger table: 2xx succeeds; 429, any 5xx, or no response retries up to retry.max_attempts (default 5); any other 4xx is terminal. Two function-specific consequences:

  • The shim turns a handler exception into a 400. A crashing handler is terminal for the trigger retry loop — exactly the behavior you want only when the crash is deterministic. Catch temporary failures inside the handler and return normally.
  • A message that exhausts its attempts is parked in a dead-letter topic rather than dropped. The event-driven functions tutorial deliberately breaks a function to watch this machinery work — that walkthrough, including the retry log lines and the dead-letter inspection, is the fastest way to build intuition here.

The headers, acknowledgement behavior, and dead-letter details belong to the messaging service: publish and consume and the Pub/Sub API reference.

Next steps​