Skip to main content

Publish and consume

This page shows every way to get messages into and out of a topic — REST API, platformctl, and the console — and spells out the delivery guarantees you are building on. It assumes a topic and subscription already exist (see the quickstart).

All curl examples use the same setup as the quickstart:

kubectl -n cai-system port-forward svc/pubsub-api 8080:8080 &
export API=http://localhost:8080
export CAI_TOKEN="<your-api-key-or-session-token>"
export CAI_PROJECT="<your-project-id>"

Publishing​

With the API​

Publishing uses a GCP-style custom method — the :publish suffix on the topic URL. Any project member can publish.

curl -sX POST "$API/v1/projects/$CAI_PROJECT/topics/orders:publish" \
  -H "Authorization: Bearer $CAI_TOKEN" -H "Content-Type: application/json" \
  -d '{"messages":[
        {"text":"hello","attributes":{"region":"eu"},"key":"customer-42"},
        {"data":"aGVsbG8gYWdhaW4=","attributes":{"datacontenttype":"text/plain"}}
      ]}'

You should see:

{"message_ids":["1234:0","1234:1"]}

Per message:

  • Exactly one of text (a plain string) or data (base64 bytes). Setting both is a 400: messages[i]: set either data (base64) or text, not both. Both empty is legal — an attributes-only message.
  • attributes are string key-value pairs; names must be non-empty and single-line.
  • key is the ordering key — on a key-shared subscription, all messages with the same key go to the same reader, in order.

Limits: request body at most 4 MiB, at most 100 messages per request (400: a publish carries at most 100 messages (got N)).

The message_ids array (in request order, ledger:entry form) is your proof of storage — counters elsewhere lag by up to about 5 minutes, but the publish response is the authority.

Two publish-time 409s to know about:

  • this topic is not ready yet: <reason> — the topic was just created; retry in a few seconds.
  • this topic is full and is configured to refuse new messages (discard: new). Consume its backlog, raise its max_bytes, or set discard: old to drop the oldest instead.

If some messages of a batch were stored before an error, you get 207 with the ids that made it plus "published" and "requested" counts — treat anything not in message_ids as unpublished.

Label your payloads

Push delivery wraps messages in CloudEvents. Publish with a datacontenttype attribute (for example application/json) or JSON payloads may arrive labelled application/octet-stream.

With the CLI​

platformctl pubsub topics publish orders --message "hello" --attribute region=eu

You should see:

1234:0

--message takes a literal string, @file to read a file, or - for stdin; --attribute KEY=VALUE repeats. The CLI sends the message as text. A partial batch failure prints partial publish: N of M stored (...).

With the console​

On your project's Pub/Sub page, each topic row has a Publish action: message text plus attributes, one KEY=VALUE per line. The console warns you before sending if the topic has no subscriptions (the message would be stranded), and the success toast names the returned message ID.

Consuming: pull​

Pull means your code asks for messages, then acknowledges each one it has finished with. It works on shared and key-shared subscriptions.

Pull messages​

curl -sX POST "$API/v1/projects/$CAI_PROJECT/topics/orders/subscriptions/workers:pull" \
  -H "Authorization: Bearer $CAI_TOKEN" -H "Content-Type: application/json" \
  -d '{"max_messages":10,"timeout_ms":2000}'

You should see:

{"messages":[{"ack_id":"...","id":"1234:0","data":"aGVsbG8=",
  "attributes":{"region":"eu"},"key":"customer-42",
  "publish_time":"...","delivery_attempt":1}]}
  • max_messages: default 10, silently capped at 100. Payloads come back base64-encoded in data.
  • timeout_ms: how long the server long-polls waiting for the first message; default 2000, capped at 20000 (20 s). An empty messages array after the wait just means nothing was ready.
  • delivery_attempt counts how many times this message has been delivered — your signal to stop retrying a poison message on a pull subscription.

Acknowledge​

Acknowledging tells the platform a message is done, so it is never redelivered. You have ack_deadline_seconds (default 30) from the pull to do it; miss the deadline and the message goes back in the queue.

curl -sX POST "$API/v1/projects/$CAI_PROJECT/topics/orders/subscriptions/workers:acknowledge" \
  -H "Authorization: Bearer $CAI_TOKEN" -H "Content-Type: application/json" \
  -d '{"ack_ids":["<ack_id-from-the-pull>"]}'

You should see:

{"acknowledged":1}

:ack is an accepted alias for :acknowledge. One request takes 1 to 1000 ack IDs, and you may send it to any API replica. An unrecognized ID is a 400: one or more acknowledgement ids are not valid for this subscription.

auto_ack is lossy

The pull body accepts "auto_ack": true, which acknowledges messages before you receive them. If your client crashes after the response, those messages are gone. It is off by default and the console deliberately doesn't offer it. Use it only for messages you can afford to lose.

With the CLI​

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

You should see:

ID      KEY          DATA
1234:0  customer-42  hello
acknowledged 1 message(s)

The table goes to stdout (payloads base64-decoded, newlines collapsed); the acknowledgement note goes to stderr. --ack runs a separate :acknowledge call after the pull; if that call fails you see pulled N message(s) but acknowledging them failed (they will redeliver): .... Without --ack, everything you pulled redelivers after the ack deadline.

Pulls the API refuses​

Two 409s tell you to use a different consumption path:

  • Pulling a push subscription: this is a push subscription; its messages are delivered to <url>. Create a separate pull subscription on the same topic to read them here.
  • Pulling an ordered subscription (exclusive/failover): it admits one reader at a time and cannot be served by a load-balanced API — use push delivery, a direct client, or a shared subscription on the same topic.

Consuming: push​

Push means the platform POSTs each message to an HTTP service inside your project and retries until your service accepts it.

Create a push subscription:

curl -sX POST "$API/v1/projects/$CAI_PROJECT/topics/orders/subscriptions" \
  -H "Authorization: Bearer $CAI_TOKEN" -H "Content-Type: application/json" \
  -d '{"name":"pusher","type":"shared",
       "deliver":{"mode":"push","push":{"url":"http://my-worker.<project-namespace>.svc.cluster.local:8080/events",
                                        "content_mode":"cloudevents-binary"}},
       "dead_letter":{"topic":"orders-dead","after_attempts":5}}'

Push target rules​

The URL is strictly validated (this prevents the platform being used to attack other systems):

  • http:// only, no userinfo, and redirects are never followed — a redirecting target fails delivery.
  • The host must be a Service in your project's own namespace: exactly <service>.<namespace>, <service>.<namespace>.svc, or <service>.<namespace>.svc.cluster.local.

A violation is a 400: deliver.push.url is not allowed: <reason>. A push target must be a Service in this project's own namespace (<ns>), e.g. http://my-worker.<ns>.svc.cluster.local:8080/events.

What your handler receives​

Every message arrives as a CloudEvent — a standard envelope for events. Two bindings:

  • cloudevents-structured (default): one JSON body carrying both metadata and payload.
  • cloudevents-binary: the raw payload as the body, metadata in headers — ce-type: ai.crusoe.pubsub.message.v1, ce-id, ce-source, ce-subject, ce-subscription, ce-deliveryattempt, and your message attributes as ce-attr-<name>. Attributes with control characters in name or value are silently dropped from binary-mode headers.

Respond with any 2xx status to acknowledge. Anything else — including a timeout — counts as a failed attempt.

Retries, timeouts, and dead letters​

  • Failed deliveries retry with exponential backoff: 1 second, doubling, capped at 60 seconds.
  • The subscription's ack_deadline_seconds doubles as the per-request timeout. If your target scales to zero, allow for a cold start when choosing it.
  • After max_deliver failed attempts, the message is republished to the dead_letter.topic with cai-dead-letter-* forensic attributes.
No dead letter means dropped

A push message that exhausts its attempts with no dead-letter topic configured is dropped (logged server-side only). Configure dead_letter for anything you cannot lose.

Push targets' response bodies are read to at most 64 KiB.

Delivery semantics at a glance​

QuestionAnswer
How many times does a message arrive?At-least-once. Duplicates are possible; make consumers idempotent.
Can I get exactly-once?No. Effectively-once publishing is not available through the REST API.
Is order preserved?Only if you choose it: shared = none; key-shared = per key; exclusive/failover = total order with one active reader.
When does a pulled message come back?When you neither acknowledge it nor extend past ack_deadline_seconds.
When does retry stop?Push: after max_deliver attempts (then dead-letter or drop). Pull: never — watch delivery_attempt.
Which counter do I trust?The publish response. Dashboard counters lag up to ~5 minutes and reset when a broker restarts.

Direct broker access (advanced)​

Project admins can fetch a credential for connecting a native Pulsar client:

curl -s "$API/v1/projects/$CAI_PROJECT/pubsub/credentials" \
  -H "Authorization: Bearer $CAI_TOKEN"

You should see:

{"service_url":"pulsar://pulsar-broker.cai-services.svc.cluster.local:6650",
 "token":"<JWT>",
 "topic_prefix":"persistent://p-<short>/main/",
 "note":"this credential can produce and consume only inside this project. It does not expire; ..."}

The console exposes the same thing behind the admin-only "Connect from outside the platform" panel — never fetched until you click.

Honest caveats, straight from the response's own note:

  • The token is scoped to your project's tenant but never expires. Revoking it means the platform operator rotating the signing key for every project. Treat it like a root credential for your project's messaging.
  • When an external_endpoint (a pulsar+ssl:// URL) is present, its reachable_scope is "vpc" — and by default only connect and topic lookup are proven from outside; producing and consuming need the in-cluster service_url.

Next steps​