Skip to main content

Publish and consume

This page covers publishing and consuming messages across Pub/Sub, including pull and push delivery modes, acknowledgement mechanics, and native wire protocol integration.

Environment configuration​

export CAI_PUBSUB_API="https://api.codyhill.dev"
export CAI_TOKEN="<your-api-key-or-session-token>"
export CAI_PROJECT="<your-project-id>"

Publishing messages​

Publishers send payloads to topics using string attributes, ordering keys, and binary or text data.

Publish payload structure​

  • Payload Body: Set either text (plain text string) or data (base64 encoded bytes).
  • Attributes: Key-value pairs for metadata or consumer filtering.
  • Ordering Key: String key used by key-shared subscriptions to guarantee in-order delivery per key.
  • Batch Limits: Up to 100 messages or 4 MiB per publish request.
platformctl pubsub topics publish orders --message "order-created" --attribute region=eu

Output:

1234:0

Consuming with Pull delivery​

Pull subscriptions allow consumers to fetch messages on demand and acknowledge them upon successful processing.

Pulling messages​

platformctl pubsub subscriptions pull workers --topic orders --max 10

Acknowledging messages​

Acknowledge messages before the ack_deadline_seconds expires to prevent automatic redelivery.

platformctl pubsub subscriptions ack workers --topic orders CAEQAB...

Or pull and acknowledge automatically:

platformctl pubsub subscriptions pull workers --topic orders --max 10 --ack

Consuming with Push delivery​

Push subscriptions deliver messages automatically via HTTP POST requests to an endpoint within your project.

Creating a push subscription​

# WORKER_URL is the "url" field that 'platformctl status order-worker' prints
platformctl pubsub subscriptions create pusher --topic orders \
  --push-url "$WORKER_URL:8080/events" \
  --push-content-mode cloudevents-binary \
  --dead-letter-topic orders-dead --dead-letter-after-attempts 5

Do not compose that address by hand. Copy the url field that platformctl status order-worker prints and append the port and path your handler listens on.

The push target is validated against your own project, at create time

The check is an allowlist, not a blocklist of bad addresses. The host has to be the private address of a workload in this subscription's own project — the address the platform already hands you, printed as the url field of platformctl status <workload> (the shortened leading form of that same address is accepted too). Anything else is refused with 400:

deliver.push.url is outside this project. A push target must be a workload in this
project, and the address to use is the one the platform already gives you: the "url"
field on that workload, which 'platformctl status <workload>' prints.

An invented hostname such as my-service.project.internal, a bare hostname, an IP literal, localhost, an external domain, and a workload in someone else's project all fail that way — as does the delivery machinery's own address, so a push target cannot be turned back on the platform. The scheme must be http — https to a private address has no per-project trust material to verify against, so it is refused rather than delivered unverified — and userinfo in the URL is refused, because it would be written into the subscription object and its logs.

CloudEvent delivery bindings​

  • cloudevents-structured (the default): Delivers a single JSON payload containing the CloudEvent metadata envelope and the message data, under Content-Type: application/cloudevents+json. No ce-* headers are sent.
  • cloudevents-binary: Delivers the raw message body with CloudEvent metadata passed via HTTP headers (ce-id, ce-type, ce-source).

Omit --push-content-mode and you get structured, so a handler written to read ce-* headers sees none of them. Ask for binary explicitly if that is what your code expects.

Native wire protocol integration​

Crusoe AI Platform Pub/Sub speaks the Pulsar wire protocol, so a stock Pulsar client library lets your own code produce and consume over the binary protocol instead of the REST API.

service_url is a private address: it resolves only from a workload running inside this project, and it is the supported path for direct clients. Reaching the broker from outside the platform is a separate thing you have to ask for — nothing here answers on the internet until this project's Pub/Sub endpoint is published, and until then the credentials response carries no external address at all, with a note saying so.

Project administrators can request the project's client credentials:

platformctl pubsub credentials

Publishing from a deployed workload​

Everything above authenticates as you, against the public API host. A function or agent running on the platform does neither, and copying these examples into a handler produces a failure that points in the wrong direction.

Use the injected private address, not the public hostname. Every workload gets CAI_PUBSUB_URL and CAI_PROJECT_ID in its environment — read the variables, do not hard-code what they hold. The public API host, https://api.codyhill.dev, does not route from inside a project; a handler that reaches for it hangs until its own timeout.

Use a service-account key, not the workload's own key. CAI_PROJECT_KEY is also injected, and it is refused on every project route by design — its one power is minting a token to read this project's secrets. The refusal is 404 not found, not 403, because the platform does not disclose whether a project exists to a caller with no grant on it. That means a wrong credential looks exactly like a misspelled topic. Check the credential first.

So: create a service account, give it the member role (enough to publish; creating topics and subscriptions needs admin), mint a key, store it as a project secret, bind the secret to the function, and Apply the binding — a bound secret only reaches the workload on the next revision.

import json, os, urllib.request

PUBSUB = os.environ['CAI_PUBSUB_URL'].rstrip('/')
PROJECT = os.environ['CAI_PROJECT_ID']
KEY = os.environ['PIPELINE_KEY']      # the bound service-account key

def publish(topic, payload):
    req = urllib.request.Request(
        '%s/v1/projects/%s/topics/%s:publish' % (PUBSUB, PROJECT, topic),
        data=json.dumps({'messages': [{'text': json.dumps(payload)}]}).encode('utf-8'),
        headers={'Content-Type': 'application/json', 'Authorization': 'Bearer ' + KEY},
        method='POST')
    with urllib.request.urlopen(req, timeout=30) as resp:
        return json.loads(resp.read().decode('utf-8') or '{}')

A handler that reads os.environ['PIPELINE_KEY'] at import time fails its first revision every time: bind the secret, then Apply, and the revision that comes up after the Apply is the one that has it.

Delivery guarantees summary​

GuaranteeDetails
Delivery CountAt-least-once delivery guaranteed across active subscriptions.
OrderingConfigurable per subscription (shared un-ordered, key-shared per-key ordered).
RedeliveryTriggered if an unacknowledged pull message exceeds ack_deadline_seconds.
Push RetriesExponential backoff retries up to max_deliver attempts before dead-letter routing.