Publishes delivered hooks to a Google Cloud Pub/Sub topic (or to anything that speaks Pub/Sub's REST API, such as the floci-gcp emulator in docker-compose.integration.yml). The message data is Ankusa.Sink.Message, byte-identical to what Ankusa.Sink.Kafka, Ankusa.Sink.NATS and Ankusa.Sink.SQS publish: inline up to :inline_max_bytes, a claim ticket above it.

Dispatch's deliveries are batched (see "Batching"): up to 1 000 of them to one topic are published in one topics.publish POST. Anything else (direct publishing, batch: false) is one request per delivery. Both go through Ankusa.HttpClient (Req's connection pool, no hidden retries or redirects). The sink needs nothing beyond core's req; its batchers are started by Ankusa.Instance. Credentials are a bearer token from a caller-supplied :token_provider, the way Ankusa.BlobStore.GCS gets one: Ankusa bundles no OAuth client.

This sink never creates a topic or a subscription. Retention, schema, encryption and who may subscribe are an operator's contract, not something the first hook to arrive should decide. A topic that does not exist is {:error, {:permanent, {:config, {:pubsub, 404, "NOT_FOUND", message}}}}: dead-lettered at once, since only an operator creating the topic fixes it, and a DLQ replay then re-publishes the hooks. (Compare Ankusa.Sink.SQS, Ankusa.Sink.Kafka and Ankusa.Sink.NATS, which likewise never create their queue, topic or stream.)

Publishing is confirmed

deliver/3 returns :ok only on a 200 whose reply carries a messageIds entry for each message sent (one, or a batch's). Pub/Sub answers 200 once it has persisted the messages, so the sink keeps the default Ankusa.Sink.durable?/1 of true. Everything else is an error (see "Error classes" in Ankusa.Sink):

• {:permanent, {:message_too_large, size, max}} — data plus attributes plus ordering key over :max_message_bytes, refused before any request.
• {:permanent, {:pubsub, status, "INVALID_ARGUMENT", message}} — Pub/Sub refused the message itself. Dead-lettered at once.
• {:permanent, {:config, {:pubsub, status, google_status, message}}} — Pub/Sub refused this sink's configuration or credentials: NOT_FOUND (topic), PERMISSION_DENIED, UNAUTHENTICATED. Dead-lettered at once and counted in ankusa_dispatch_config_errors_total{sink}; fix the source, then replay the dead letters.
• {:retry_after, ms, {:pubsub, status, google_status, message}} — Pub/Sub throttled the publish (RESOURCE_EXHAUSTED, or any HTTP 429 or 503). ms is a full-jitter draw (Ankusa.Sink.throttle_after_ms/1).
• {:pubsub, status, google_status, message} — any other Pub/Sub error (UNAVAILABLE, INTERNAL, ...): transient, the retry policy.
• {:status, status, body} — a non-200 response that is not a Google error document (a proxy, a load balancer); a 429 or 503 is a {:retry_after, ms, {:status, status, body}}.
• {:unexpected_reply, body}, :no_credentials (the :token_provider answered :error; no request is sent) and transport errors — retried.

Caveats

• A topic with no subscription (and no topic message retention) still answers 200 and discards the message. The sink cannot detect this.
• There is no publish-side deduplication. Delivery is at-least-once, so consumers dedupe on the message's idempotency_key (the ankusa_idempotency_key attribute).
• An ordering key only orders messages on a subscription with message ordering enabled, published through one regional endpoint. Ankusa's own deliveries are unordered anyway (see Ankusa.Sink), so no key is sent unless :ordering_key is set; a key also caps that key's throughput at about 1 MB/s.

Batching

A delivery from dispatch waits in its topic's Ankusa.Sink.Batcher (one per topic and set of credentials) and is published in one topics.publish with up to 999 others, 20 ms after the first arrived or as soon as the batch is full (1,000 messages, 7.5 MB of data). deliver/3 returns when the request's response is known: the hook's row stays claimed until then, so a batcher that dies loses nothing and the attempt is retried.

publish has no per-message results: the topic accepts the request or refuses it as a whole. A transient error or throttle (RESOURCE_EXHAUSTED) fails every message of it, each retrying on its own schedule; so does a config error (NOT_FOUND, PERMISSION_DENIED, UNAUTHENTICATED), which no message caused. INVALID_ARGUMENT means one message was refused, and dead-lettering them all would lose the good ones, so the request is bisected (see Ankusa.Sink.GooglePubSub.Batcher): the one bad message is {:permanent, ...} and the others are delivered, for 2 * log2(n) extra requests. The size and attribute checks that make a message invalid run before it joins a batch, so this is the exception.

A batch may hold hooks of several tenants; entries are taken round-robin across tenants, and a delivery that would pass the queue's ceilings is refused with {:retry_after, ms, {:batch_queue_full, reason}} (see Ankusa.Sink.Batcher). With :ordering_key set, a key has one request in flight at a time and its messages leave in the order they were admitted.

What each message carries

• data — the Ankusa.Sink.Message JSON, base64 on the wire as the REST API requires.
• attributes — the list Ankusa.Sink.Message.attributes/2 builds: ankusa_id, ankusa_source_id, ankusa_tenant_id, ankusa_message_version, ankusa_idempotency_key, content_type (application/json), plus ankusa_dedupe_key and ankusa_replay_id when the hook carries them. ankusa_tenant_id is left out when the hook has no tenant.
• orderingKey — only when :ordering_key yields a non-empty string.

opts

• :project — required, the GCP project id
• :topic — required, the topic id (the short name, not

                     `projects/.../topics/...`)

• :endpoint — default "https://pubsub.googleapis.com". Set it

                     for a regional endpoint
                     (`"https://us-east1-pubsub.googleapis.com"`) or
                     `floci-gcp` (`"http://localhost:4588"`).

• :token_provider — {mod, fun, args}, applied once per request, and

                     answering `{:ok, token}` or `:error`. Omit it to
                     send no `authorization` header (an emulator only).

• :ordering_key — nil (default: no key), a string, or

                     `(Envelope.t() -> String.t())`

• :inline_max_bytes — default 64 KiB (65,536), configurable
• :max_message_bytes — the largest data plus attributes plus ordering key

                     sent; default `10_000_000`, Pub/Sub's 10 MB message
                     limit

• :timeout_ms — default 5_000, for both connect and response
• :req_options — transport options for the HTTP client (a Finch

                     pool, a proxy via `:connect_options`, `plug:` for
                     `Req.Test`); see `Ankusa.HttpClient`.

• :batch — default true; false publishes every delivery in

                     its own request and counts it against
                     `dispatch.sink_concurrency` as before

• :batch_linger_ms — default 20 (1 to 60,000), how long a batch waits to

                     fill

• :batch_max_entries — default and maximum 1,000
• :batch_max_bytes — default and maximum 7,500,000 of data before base64

                     (the REST body is 4/3 of it; Pub/Sub's request limit
                     is 10 MB)

• :batch_queue_max, :batch_queue_max_bytes, :batch_queue_tenant_max

                   — the queue's ceilings, default 4,096 entries,
                     128 MiB and 1,024 entries a tenant

• :batch_max_inflight — requests in flight per topic; default the

                     instance's `dispatch.sink_concurrency` (8)