|
| 1 | +--- |
| 2 | +icon: tabler:refresh |
| 3 | +--- |
| 4 | + |
| 5 | +# Sync Backplane |
| 6 | + |
| 7 | +> [!IMPORTANT] |
| 8 | +> Experimental! The `sync` option and `crossws/sync` API may change between 0.x versions. |
| 9 | +
|
| 10 | +crossws [pub/sub](/guide/pubsub) is in-memory and **local to one instance**. When you run more than one instance — multiple regions, processes, or replicas behind a load balancer — a `peer.publish("chat", ...)` only reaches the peers connected to _that_ instance. A subscriber connected elsewhere never sees it. |
| 11 | + |
| 12 | +A **sync adapter** bridges that gap: it relays published messages between instances over a shared backplane (e.g. Redis pub/sub) and fans inbound messages out to local subscribers. You don't need to change your hooks — `subscribe` and `publish` keep working, now across the cluster. |
| 13 | + |
| 14 | +```js |
| 15 | +import nodeAdapter from "crossws/adapters/node"; |
| 16 | +import { redis } from "crossws/sync"; |
| 17 | +import Redis from "ioredis"; |
| 18 | + |
| 19 | +const ws = nodeAdapter({ |
| 20 | + hooks, |
| 21 | + sync: redis({ client: new Redis(), channel: "my-app" }), |
| 22 | +}); |
| 23 | +``` |
| 24 | + |
| 25 | +Each driver takes a single options object. The `channel` name is **required** — it scopes the cluster, so unrelated servers don't silently bridge into each other. |
| 26 | + |
| 27 | +crossws subscribes to the backplane automatically. On shutdown, call `await ws.close()` — it closes every connected peer and tears down the backplane in one step (leaving any Redis/Postgres client you passed in connected; its lifecycle stays yours). Any server you created (e.g. the `http.Server`) is also yours to close. |
| 28 | + |
| 29 | +## Delivery semantics |
| 30 | + |
| 31 | +The cross-instance relay is **best-effort and fire-and-forget** — design your topics to tolerate that: |
| 32 | + |
| 33 | +- **At-most-once and unordered.** A relayed message reaches remote subscribers at most once, and ordering across instances is not guaranteed. If you need to detect a gap, carry your own sequence/version in the payload rather than assuming a reliable stream. |
| 34 | +- **No replay or buffering.** Messages published while an instance is disconnected from the backplane — or before it finishes subscribing on startup — are not queued; that instance simply misses them (see the ioredis vs node-redis [reconnect note](#redis-client-channel-connector)). |
| 35 | +- **Local delivery is independent.** `peer.publish()` always reaches _local_ subscribers synchronously, even when the backplane is down. Only the cross-instance relay is best-effort. |
| 36 | +- **Failures never throw into your `publish()`.** A backplane error is isolated so it can't crash the process. Pass `onError` to observe a degraded backplane (for logging, metrics, or alerting): |
| 37 | + |
| 38 | +```js |
| 39 | +const ws = nodeAdapter({ |
| 40 | + hooks, |
| 41 | + sync: redis({ client: new Redis(), channel: "my-app" }), |
| 42 | + onError(error, { stage }) { |
| 43 | + // stage: "subscribe" (initial connect) | "publish" (relay out) | "delivery" (fan-in) |
| 44 | + metrics.increment(`crossws.sync.error.${stage}`); |
| 45 | + console.error("[crossws] sync error", stage, error); |
| 46 | + }, |
| 47 | +}); |
| 48 | +``` |
| 49 | + |
| 50 | +Without `onError`, failures are logged to `console.error`. |
| 51 | + |
| 52 | +## Choosing a driver |
| 53 | + |
| 54 | +| Driver | Backplane | Reach | When to use | |
| 55 | +| --- | --- | --- | --- | |
| 56 | +| [`redis`](#redis-client-channel-connector) | Redis pub/sub | Multi-region / multi-host | The general-purpose choice for a real cluster. | |
| 57 | +| [`pgsql`](#pgsql-client-channel-connector) | Postgres `LISTEN`/`NOTIFY` | Multi-region / multi-host | You already run Postgres and would rather not add Redis (payloads must stay under 8 KB). | |
| 58 | +| [`cluster`](#cluster-channel) | Node.js `cluster` IPC | Single host (forked processes) | Multiple processes on one host (Node `cluster` / PM2) without a network broker. | |
| 59 | +| [`broadcastChannel`](#broadcastchannel-channel) | `BroadcastChannel` | Single process (worker threads) | Local fan-out, tests, in-process worker threads. | |
| 60 | + |
| 61 | +All four ship in `crossws/sync` and have no third-party dependencies — `redis` and `pgsql` take a client you bring; `cluster` and `broadcastChannel` need nothing. |
| 62 | + |
| 63 | +> [!NOTE] |
| 64 | +> On **Cloudflare** the model is different — a single Durable Object is already cluster-global, so a backplane is only relevant for multi-instance sharding or the fallback path. See [Cloudflare → Sync across instances](/adapters/cloudflare#sync-across-instances). |
| 65 | +
|
| 66 | +## Sync Drivers |
| 67 | + |
| 68 | +### `redis({ client, channel, connector? })` |
| 69 | + |
| 70 | +Works out of the box with both [ioredis](https://github.com/redis/ioredis) and [node-redis](https://github.com/redis/node-redis). The client flavor is auto-detected; pass `connector: "ioredis" | "node-redis"` to override detection. |
| 71 | + |
| 72 | +```js [ioredis] |
| 73 | +import { redis } from "crossws/sync"; |
| 74 | +import Redis from "ioredis"; |
| 75 | + |
| 76 | +const ws = nodeAdapter({ |
| 77 | + hooks, |
| 78 | + sync: redis({ client: new Redis(), channel: "my-app" }), |
| 79 | +}); |
| 80 | +``` |
| 81 | + |
| 82 | +```js [node-redis] |
| 83 | +import { redis } from "crossws/sync"; |
| 84 | +import { createClient } from "redis"; |
| 85 | + |
| 86 | +const client = await createClient().connect(); |
| 87 | + |
| 88 | +const ws = nodeAdapter({ |
| 89 | + hooks, |
| 90 | + sync: redis({ client, channel: "my-app" }), |
| 91 | +}); |
| 92 | +``` |
| 93 | + |
| 94 | +crossws derives a dedicated `SUBSCRIBE` connection from your client via `duplicate()` (for node-redis it also `connect()`s it), so a single connected client is all you pass in. |
| 95 | + |
| 96 | +> [!NOTE] |
| 97 | +> On a dropped connection, ioredis automatically re-subscribes its channels; node-redis does not restore subscriptions the same way, so after a transient outage a node-redis-backed instance may stop receiving relayed messages until the client reconnects and re-subscribes. Prefer ioredis if resilience to flaky connections matters. |
| 98 | +
|
| 99 | +### `pgsql({ client, channel, connector? })` |
| 100 | + |
| 101 | +Relays over PostgreSQL [`LISTEN`/`NOTIFY`](https://www.postgresql.org/docs/current/sql-notify.html) — handy for clusters that already run Postgres and would rather not add Redis. Works with both [node-postgres](https://github.com/brianc/node-postgres) (`pg`) and [postgres.js](https://github.com/porsager/postgres); the flavor is auto-detected, with `connector: "pg" | "postgres.js"` to override. |
| 102 | + |
| 103 | +```js [node-postgres] |
| 104 | +import { pgsql } from "crossws/sync"; |
| 105 | +import { Client } from "pg"; |
| 106 | + |
| 107 | +const client = new Client(); |
| 108 | +await client.connect(); |
| 109 | + |
| 110 | +const ws = nodeAdapter({ |
| 111 | + hooks, |
| 112 | + sync: pgsql({ client, channel: "my-app" }), |
| 113 | +}); |
| 114 | +``` |
| 115 | + |
| 116 | +```js [postgres.js] |
| 117 | +import { pgsql } from "crossws/sync"; |
| 118 | +import postgresjs from "postgres"; |
| 119 | + |
| 120 | +const sql = postgresjs(); |
| 121 | + |
| 122 | +const ws = nodeAdapter({ |
| 123 | + hooks, |
| 124 | + sync: pgsql({ client: sql, channel: "my-app" }), |
| 125 | +}); |
| 126 | +``` |
| 127 | + |
| 128 | +Unlike Redis `SUBSCRIBE`, Postgres `LISTEN` doesn't block the connection, so no duplicate connection is needed (for node-postgres the same client both listens and notifies; postgres.js reserves its own dedicated connection internally). |
| 129 | + |
| 130 | +> [!IMPORTANT] |
| 131 | +> Pass a dedicated `Client`, **not** a `Pool`. Pool connections rotate per query, so a `LISTEN` lands on a backend that's then returned to the pool and notifications never reach a stable listener. A `Pool` is detected and rejected at construction. For the same reason, give this driver its own client rather than sharing one across two `pgsql()` instances on the same channel — closing one issues `UNLISTEN` and silences the others. |
| 132 | +
|
| 133 | +Two transport limits to keep in mind: the `channel` name maps to a Postgres identifier, capped at **63 bytes** (rejected at construction if longer), and a `NOTIFY` payload is capped at **8000 bytes** (base64 inflates binary ~33%), so keep relayed messages small. |
| 134 | + |
| 135 | +### `cluster({ channel })` |
| 136 | + |
| 137 | +Relays over Node.js [`cluster`](https://nodejs.org/api/cluster.html) IPC — bridges the **forked processes on a single host** (Node `cluster`, or PM2 `instances`) without standing up a network broker. This is the gap [`broadcastChannel`](#broadcastchannel-channel) leaves: its registry is per-process and silently won't sync across forks. For multiple hosts or regions you still want [`redis`](#redis-client-channel-connector) or [`pgsql`](#pgsql-client-channel-connector). |
| 138 | + |
| 139 | +The driver runs in the **workers**; the **primary** needs a one-line relay because cluster workers can't message each other directly — IPC only flows between each worker and the primary, which rebroadcasts. Call `setupPrimaryCluster()` once in the primary (it's a no-op in workers, so guarding with `cluster.isPrimary` is optional): |
| 140 | + |
| 141 | +```js |
| 142 | +import cluster from "node:cluster"; |
| 143 | +import { availableParallelism } from "node:os"; |
| 144 | +import { setupPrimaryCluster, cluster as clusterSync } from "crossws/sync"; |
| 145 | + |
| 146 | +if (cluster.isPrimary) { |
| 147 | + setupPrimaryCluster(); |
| 148 | + for (let i = 0; i < availableParallelism(); i++) cluster.fork(); |
| 149 | +} else { |
| 150 | + const ws = nodeAdapter({ |
| 151 | + hooks, |
| 152 | + sync: clusterSync({ channel: "my-app" }), |
| 153 | + }); |
| 154 | + // ... start your server |
| 155 | +} |
| 156 | +``` |
| 157 | + |
| 158 | +Binary payloads are base64-encoded, so default IPC serialization is enough — you don't need `serialization: "advanced"`. Calling `clusterSync()` outside a forked worker (no `process.send`) throws on `subscribe`, surfacing the misconfiguration rather than silently not syncing. |
| 159 | + |
| 160 | +### `broadcastChannel({ channel })` |
| 161 | + |
| 162 | +Bridges instances that share a `BroadcastChannel` registry. On Node.js, Deno and Bun that registry is scoped to a **single process** — it spans the main thread and its worker threads, but **not** separate OS processes (e.g. Node `cluster`/PM2 forks), which each get an isolated registry and silently won't sync. For forked processes on one host use [`cluster`](#cluster-channel); across hosts or regions use [`redis`](#redis-client-channel-connector) or [`pgsql`](#pgsql-client-channel-connector). (Deno Deploy is the exception — its `BroadcastChannel` spans isolates.) |
| 163 | + |
| 164 | +```js |
| 165 | +import { broadcastChannel } from "crossws/sync"; |
| 166 | + |
| 167 | +const ws = nodeAdapter({ |
| 168 | + hooks, |
| 169 | + sync: broadcastChannel({ channel: "my-app" }), |
| 170 | +}); |
| 171 | +``` |
| 172 | + |
| 173 | +## Writing a driver |
| 174 | + |
| 175 | +Any transport that can fan a message out to your other instances can back a sync driver. A driver is a `SyncAdapter`: a factory crossws calls **once per instance**, passing a stable per-instance `id`. It returns a `SyncDriver` with three methods: |
| 176 | + |
| 177 | +**`publish(msg)`** — called for every local `peer.publish()` / `adapter.publish()`. Relay `msg` to the other instances over your backplane. A `msg` is `{ namespace, topic, data }`, where `data` is a `string | Uint8Array`. |
| 178 | + |
| 179 | +**`subscribe(deliver)`** — called once on startup. Listen for messages from the other instances and hand each one to `deliver`, which fans it out to this instance's local subscribers. |
| 180 | + |
| 181 | +**`close()`** _(optional)_ — called when the adapter shuts down. Release any connection you opened. |
| 182 | + |
| 183 | +Two things are your responsibility on the wire: **suppress your own echo** (most backplanes deliver a publisher its own messages) and **preserve binary payloads** (most transports are text-only). Both are handled by a small envelope that stamps the sender `id` and base64-encodes binary `data`. crossws exports the same `encodeEnvelope` / `decodeEnvelope` helpers its built-in drivers use, so you don't have to reimplement them: |
| 184 | + |
| 185 | +```ts |
| 186 | +import { encodeEnvelope, decodeEnvelope } from "crossws/sync"; |
| 187 | +import type { SyncAdapter } from "crossws/sync"; |
| 188 | + |
| 189 | +const mySync: SyncAdapter = ({ id }) => ({ |
| 190 | + subscribe(deliver) { |
| 191 | + backplane.on("message", (raw) => { |
| 192 | + const envelope = decodeEnvelope(raw); |
| 193 | + if (!envelope || envelope.id === id) { |
| 194 | + return; // ignore malformed messages and our own echo |
| 195 | + } |
| 196 | + deliver(envelope.msg); |
| 197 | + }); |
| 198 | + }, |
| 199 | + publish(msg) { |
| 200 | + backplane.send(encodeEnvelope(id, msg)); // stamps `id` so we can filter it back out |
| 201 | + }, |
| 202 | + close() { |
| 203 | + backplane.disconnect(); // optional: release the connection on shutdown |
| 204 | + }, |
| 205 | +}); |
| 206 | +``` |
| 207 | + |
| 208 | +`encodeEnvelope(id, msg)` returns a JSON string; `decodeEnvelope(raw)` parses it back to `{ id, msg }` (or `undefined` for anything malformed). Binary `data` is base64-encoded inside the envelope and restored on decode. |
| 209 | + |
| 210 | +**Notes:** |
| 211 | +- **Skip the echo check and every publish is delivered twice locally** — once from your own loopback, once from `deliver`. The `envelope.id === id` guard is what prevents it. |
| 212 | +- **Reuse the envelope helpers** unless your transport carries structured data natively. `BroadcastChannel`, for example, can pass `{ id, msg }` (binary `Uint8Array` and all) straight through, so its built-in driver skips the JSON/base64 envelope entirely and filters on `id` directly. |
0 commit comments