> Part of the walkerOS documentation. Project overview and full index: <https://www.walkeros.io/llms.txt>

# Google Pub/Sub

<!-- -->

[Server](#)[ ](https://github.com/elbwalker/walkerOS/tree/main/packages/server/sources/gcp)

<!-- -->

[Source code](https://github.com/elbwalker/walkerOS/tree/main/packages/server/sources/gcp)[ ](https://www.npmjs.com/package/@walkeros/server-source-gcp)

<!-- -->

[Package](https://www.npmjs.com/package/@walkeros/server-source-gcp)

Subscribes to a [Google Cloud Pub/Sub](https://cloud.google.com/pubsub) topic and forwards each delivered message to the walkerOS collector. Ships two delivery models from the same package: a long-running streaming pull subscriber for container deployments, and an HTTP push webhook handler for serverless. Decoders for JSON, text, and raw payloads. Idempotent subscription provisioning with optional dead-letter wiring. Optional OIDC verification on the push handler.

The source ships inside `@walkeros/server-source-gcp` alongside the Cloud Functions handler; install the package once and import `sourcePubSubPull` or `sourcePubSubPush`.

<!-- -->

Where this fits

Pub/Sub is a **server source** in the walkerOS flow:

Forwards each Pub/Sub message to the collector. Pull mode runs a streaming subscriber in long-lived containers; push mode receives HTTP envelopes in serverless deployments.

## Installation[​](#installation "Direct link to Installation")

```
npm install @walkeros/server-source-gcp
```

## Choosing pull vs push[​](#choosing-pull-vs-push "Direct link to Choosing pull vs push")

| Aspect            | Pull (`sourcePubSubPull`)                              | Push (`sourcePubSubPush`)                                   |
| ----------------- | ------------------------------------------------------ | ----------------------------------------------------------- |
| Process model     | Long-running streaming subscriber                      | Stateless HTTP request handler                              |
| Hosting           | Cloud Run with `min-instances >= 1`, GKE, GCE, on-prem | Cloud Run scale-to-zero, Cloud Functions, Lambda            |
| Idle cost         | Pay for always-on compute                              | Zero when idle                                              |
| Throughput        | Highest (gRPC streaming)                               | Limited by HTTP overhead                                    |
| Latency           | Lowest (warm stream, single-digit ms)                  | Higher (HTTP per message, cold-start adds 100-500ms)        |
| Auth direction    | Subscriber holds credentials, connects outbound        | Pub/Sub holds credentials, posts inbound (OIDC verify)      |
| Subscription type | `subscription.type = "pull"`                           | `subscription.type = "push"` with `pushConfig.pushEndpoint` |
| Best fit          | Long-running collector services, high-volume streams   | Serverless walkerOS endpoints, hybrid setups                |

Both modes share the same `@walkeros/server-source-gcp` package, the same auth resolution, the same decoders, and the same setup command. Pick the one that matches your hosting model.

## Pull subscriber[​](#pull-subscriber "Direct link to Pull subscriber")

```
import { sourcePubSubPull } from '@walkeros/server-source-gcp';
import { startFlow } from '@walkeros/collector';

await startFlow({
  sources: {
    pubsub: {
      code: sourcePubSubPull,
      config: {
        settings: {
          projectId: 'YOUR_PROJECT_ID',
          subscription: 'YOUR_SUBSCRIPTION_NAME',
        },
      },
    },
  },
  destinations: {
    // your destinations
  },
});
```

The pull source is event-driven: `init()` opens the streaming subscription and forwards each delivered message to the collector via `env.push`. The source's `push` method is a deliberate no-op stub. There is no external invocation; the subscriber pushes events autonomously. `destroy()` closes the subscriber gracefully, honoring `shutdownTimeoutMs` (default 30000).

### Pull settings[​](#pull-settings "Direct link to Pull settings")

| Field               | Type                                  | Default                                 | Description                                                                                               |
| ------------------- | ------------------------------------- | --------------------------------------- | --------------------------------------------------------------------------------------------------------- |
| `projectId`         | `string`                              | (required)                              | GCP project id.                                                                                           |
| `subscription`      | `string`                              | (required)                              | Subscription short name.                                                                                  |
| `topic`             | `string`                              | (optional)                              | Topic short name. Used by setup auto-create when `setup.createTopic` is on.                               |
| `credentials`       | `string \| ServiceAccountCredentials` | (ADC fallback)                          | Service account JSON (parsed object or JSON string). Set on `config.credentials` (sibling of `settings`). |
| `apiEndpoint`       | `string`                              | (SDK default)                           | Override for the emulator (e.g. `localhost:8085`).                                                        |
| `decoder`           | `'json' \| 'text' \| 'raw'`           | `'json'`                                | How the subscriber decodes message bodies before forwarding.                                              |
| `flowControl`       | `{ maxMessages?, maxBytes? }`         | `{ maxMessages: 100, maxBytes: 10 MB }` | Subscriber pull rate (in-flight cap).                                                                     |
| `ackDeadline`       | `number`                              | `60`                                    | Per-message ack window in seconds.                                                                        |
| `shutdownTimeoutMs` | `number`                              | `30000`                                 | Graceful drain budget on `destroy()`.                                                                     |
| `onPushError`       | `'nack' \| 'ack'`                     | `'nack'`                                | Behavior when forwarding to the collector throws.                                                         |

## Push webhook[​](#push-webhook "Direct link to Push webhook")

```
import express from 'express';
import { sourcePubSubPush } from '@walkeros/server-source-gcp';
import { startFlow } from '@walkeros/collector';

const { sources } = await startFlow({
  sources: {
    pubsub: {
      code: sourcePubSubPush,
      config: { settings: { decoder: 'json' } },
    },
  },
  destinations: {
    // your destinations
  },
});

const app = express();
app.use(express.json());
app.post('/pubsub-push', sources.pubsub.push);
app.listen(8080);
```

Pub/Sub posts each message envelope to the configured endpoint as a JSON body matching the [Pub/Sub push spec](https://cloud.google.com/pubsub/docs/push). The handler decodes the envelope, base64-decodes the `data` field, runs the configured decoder, and forwards to the collector. Status codes:

* `200`: forwarded successfully.
* `400`: the message is permanently invalid, either a malformed envelope (missing `message.data`, invalid base64) or an event the pipeline rejected. A retry cannot fix either, so the answer is a nack.
* `401`: OIDC verification failed (when enabled).
* `500`: collector push failed on an otherwise valid message; Pub/Sub will retry per the subscription's retry policy.

Because a permanently invalid message is nacked rather than acknowledged, Pub/Sub keeps redelivering it. Set a [dead-letter policy](https://cloud.google.com/pubsub/docs/handling-failures) on the subscription: that policy, not the source, is the poison-message guard.

### Push settings[​](#push-settings "Direct link to Push settings")

In addition to the shared base (`projectId`, `apiEndpoint`, `decoder` in `settings`, plus `config.credentials`):

| Field        | Type      | Default | Description                                                               |
| ------------ | --------- | ------- | ------------------------------------------------------------------------- |
| `verifyOidc` | `boolean` | `false` | When `true`, verifies the bearer token Pub/Sub attaches to push requests. |
| `audience`   | `string`  | -       | Required when `verifyOidc` is true. Must match your endpoint URL exactly. |

## Authentication[​](#authentication "Direct link to Authentication")

Three modes for the pull subscriber, evaluated in order:

1. **Application Default Credentials (ADC).** Default. Works on GCP-native runtimes (Cloud Run, Cloud Functions, GKE) and locally with `gcloud auth application-default login`.
2. **Service account JSON.** Pass `config.credentials`, either parsed or as JSON string (the source JSON-parses strings). Combine with `$env.NAME` to inject from an environment variable. (The package-specific `settings.credentials` still works but is deprecated.)
3. **Pre-configured client.** Pass an existing `PubSub` SDK instance as `settings.client`.

For the push handler, auth is reversed: Pub/Sub holds the identity and signs requests with an OIDC token. The handler verifies the token against GCP public keys when `verifyOidc: true` and `audience` is set. Off by default because misconfigured OIDC silently rejects all messages, so opt in once your endpoint is publicly reachable.

## Setup[​](#setup "Direct link to Setup")

Provision the subscription once per environment:

```
walkeros setup source.pubsub
```

Idempotent. Defaults: 60-second ack deadline, project-default retention, no filter, no dead-letter policy. Drift on `ackDeadlineSeconds`, `messageRetentionDuration`, `deadLetterPolicy`, or `filter` emits `WARN setup.drift {...}` and never auto-mutates.

`config.setup`:

* `false` (default): no provisioning.
* `true`: provisions the subscription against an existing topic with safe defaults.
* `{ ackDeadlineSeconds, messageRetentionDuration, filter, deadLetterPolicy, retryPolicy, ... }`: object form for explicit overrides.

Optional one-shot helpers:

* `setup.createTopic: true`: auto-create the topic if it does not exist (requires `settings.topic`). Useful when source and topic owners share a deployment.
* `setup.deadLetterPolicy.createDeadLetterTopic: true`: auto-create the dead-letter topic referenced in `deadLetterPolicy.deadLetterTopic`.

The provisioning identity needs `pubsub.subscriptions.create` (and `pubsub.topics.create` if auto-creating). The runtime pull identity needs `roles/pubsub.subscriber` on the subscription.

Topic provisioning is owned by the [Pub/Sub destination](https://www.walkeros.io/docs/destinations/server/pubsub.md) when the topic is the producer's responsibility; the source can also auto-create as a convenience for self-contained setups.

## Decoders[​](#decoders "Direct link to Decoders")

| Decoder          | Behavior                                  |
| ---------------- | ----------------------------------------- |
| `json` (default) | `JSON.parse(data.toString('utf8'))`       |
| `text`           | `data.toString('utf8')` (forwarded as-is) |
| `raw`            | The raw `Buffer` is forwarded unchanged   |

A decoder that throws on malformed payloads triggers `onPushError` (pull mode): `nack` redelivers, `ack` drops with a debug log.

## Emulator[​](#emulator "Direct link to Emulator")

Both pull and push honor `PUBSUB_EMULATOR_HOST`. For explicit configuration, set `settings.apiEndpoint` (e.g. `localhost:8085`).

```
gcloud beta emulators pubsub start --host-port=localhost:8085
export PUBSUB_EMULATOR_HOST=localhost:8085
```

## Lifecycle[​](#lifecycle "Direct link to Lifecycle")

The pull source is the canonical long-running listener pattern in walkerOS. Existing server sources (express, lambda, cloudfunction, fetch) are request-handlers; pull is fundamentally different:

1. `init()` starts the streaming subscriber. It returns the `Source.Instance` and the subscriber runs in the background.
2. The instance's `push` method is a no-op stub. The framework never calls it; the subscriber forwards events autonomously.
3. `destroy()` closes the subscriber. On shutdown the source stops accepting new messages, drains in-flight ack/nacks, and force-closes after `shutdownTimeoutMs`.

## Troubleshooting[​](#troubleshooting "Direct link to Troubleshooting")

* **`NOT_FOUND` on the subscription**: the subscription does not exist. Run `walkeros setup source.<id>` once or create it via `gcloud pubsub subscriptions create`.
* **`PERMISSION_DENIED` / `UNAUTHENTICATED`** on the pull subscriber: the runtime service account lacks `roles/pubsub.subscriber` on the subscription.
* **Push handler returns 401 for every message**: OIDC misconfiguration. Verify `settings.audience` matches your endpoint URL exactly, including scheme and path.
* **Messages decoded incorrectly**: switch the decoder. JSON is the default; use `text` for plain-text payloads, `raw` for binary.

## Next steps[​](#next-steps "Direct link to Next steps")

* [Pub/Sub destination](https://www.walkeros.io/docs/destinations/server/pubsub.md) for publishing events to a Pub/Sub topic
* [GCP source overview](https://www.walkeros.io/docs/sources/server/gcp.md) for the Cloud Functions HTTP handler