A Fast GraphQL API With Fastify and Yoga · Part 6 of 8 in the series
Part 6: Subscriptions Across Instances With Valkey
Run a second instance of the API, watch its subscribers miss the events of the first, and share the events through Valkey, so every instance hears every change, even after Valkey restarts.
Quest objectives
- Understand why an in-memory PubSub stops at the edge of its process
- Run Valkey next to PostgreSQL
- Send the events of Yoga through Valkey with the Redis event target
- Keep the API running when Valkey goes away
One Node.js process runs JavaScript on one CPU core. To use more cores, to deploy without downtime, or to spread the load over several machines, you run the API more than once, with a load balancer in front. Then a subscription can land on one instance and the mutation it waits for on another.
Two instances
Reset the board, so your output matches the one here:
$ LOG_SQL=false pnpm db:seedSeeded 3 guilds, 9 adventurers and 10 quests.
Keep pnpm dev running on port 4000 and start a second instance on port 4001 in another terminal. A variable set in the shell wins over the one in .env:
$ PORT=4001 pnpm start{"level":30,"time":1793720723417,"pid":58,"hostname":"laptop","msg":"Server listening at http://[::1]:4001"}{"level":30,"time":1793720723418,"pid":58,"hostname":"laptop","msg":"Server listening at http://127.0.0.1:4001"}
Subscribe on the second instance:
$ curl -sN localhost:4001/graphql -H 'accept: text/event-stream' -H 'content-type: application/json' -d '{"query":"subscription { questPosted { title } }"}':
And post a quest on the first:
$ curl -s localhost:4000/graphql -H 'content-type: application/json' -d '{"query":"mutation { postQuest(input: { guildId: 3, title: \"Count the acorns\", reward: 5 }) { id } }"}'{"data":{"postQuest":{"id":"11"}}}
The quest is posted, and the subscriber on port 4001 hears nothing. The PubSub from the last part is a list of listeners in the memory of its process, and the instance on port 4000 has no way of reaching the listeners of the one on 4001.
Valkey
The instances need something in the middle that every one of them can reach. Valkey is an open source key-value store that grew out of Redis and speaks the same protocol, so every Redis client works with it. Its pub/sub sends a message published on a channel to every connection that subscribed to it. Add it to compose.yml:
services:
postgres:
image: postgres:18-alpine
environment:
POSTGRES_USER: guild
POSTGRES_PASSWORD: guild
POSTGRES_DB: guild
ports:
- 127.0.0.1:5432:5432
volumes:
- postgres-data:/var/lib/postgresql
valkey:
image: valkey/valkey:9-alpine
ports:
- 127.0.0.1:6379:6379
volumes:
postgres-data:$ docker compose up -d Container guild-api-postgres-1 Running Container guild-api-valkey-1 Started
Add its address to .env:
REDIS_URL=redis://localhost:6379Yoga's PubSub takes an event target, the part that actually delivers the events, and there is one for Redis. Install it with ioredis, the client it uses:
$ pnpm add ioredis@5 @graphql-yoga/redis-event-targetPackages: +8dependencies:+ @graphql-yoga/redis-event-target 4.0.0+ ioredis 5.11.1Done in 416ms using pnpm v12.16.1
The @5 matters. ioredis 6 is out, but the event target still expects ioredis 5. Now give the PubSub its event target in src/pubsub.ts:
import { createPubSub } from 'graphql-yoga';
import { createRedisEventTarget } from '@graphql-yoga/redis-event-target';
import { Redis } from 'ioredis';
import type { QuestRow } from './db/schema.ts';
export const pubsub = createPubSub<{
'quest:posted': [quest: QuestRow];
'quest:updated': [id: number, quest: QuestRow];
}>({
eventTarget: createRedisEventTarget({
publishClient: new Redis(process.env.REDIS_URL!),
subscribeClient: new Redis(process.env.REDIS_URL!),
}),
});
export type PubSub = typeof pubsub;It needs two connections, because a Redis connection that has subscribed to a channel can do nothing else, not even publish. Nothing else changes. The resolvers publish and subscribe as before, and the event target sends the events through Valkey instead of memory.
pnpm dev restarted by itself. Stop the second instance with Ctrl+C, start it again, subscribe on port 4001 again and post the same quest on port 4000:
$ curl -sN localhost:4001/graphql -H 'accept: text/event-stream' -H 'content-type: application/json' -d '{"query":"subscription { questPosted { title } }"}':event: nextdata: {"data":{"questPosted":{"title":"Count the acorns"}}}
The event crossed from one instance to the other. While a subscription is open, you can see its channels in Valkey. With one board on questPosted and one on questUpdated(id: 4):
$ docker compose exec valkey valkey-cli pubsub channelsquest:updated:4quest:posted
A topic becomes a channel of its own, quest:updated:4, so Valkey only sends an update to the instances that have someone waiting for it.
When Valkey goes away
Valkey is now something the API depends on, so find out what happens without it. Stop it:
$ docker compose stop valkey Container guild-api-valkey-1 Stopping Container guild-api-valkey-1 Stopped
Then post a quest to the second instance, on port 4001. The mutation answers as usual. Seven seconds later, the instance is gone. This is its terminal, shortened, because ioredis repeats its warning on every attempt to reconnect:
[ioredis] Unhandled error event: AggregateError [ECONNREFUSED]: at internalConnectMultiple (node:net:1492:18) at afterConnectMultiple (node:net:2161:7)[ioredis] Unhandled error event: AggregateError [ECONNREFUSED]: at internalConnectMultiple (node:net:1492:18) at afterConnectMultiple (node:net:2161:7){"level":30,"time":1793719497516,"pid":8,"hostname":"laptop","requestId":"req-1","method":"POST","path":"/graphql","operationType":"mutation","status":200,"errorCount":0,"durationMs":34,"msg":"Request processed"}[ioredis] Unhandled error event: AggregateError [ECONNREFUSED]: at internalConnectMultiple (node:net:1492:18) at afterConnectMultiple (node:net:2161:7)/home/robin/projects/guild-api/node_modules/.pnpm/ioredis@5.11.1/node_modules/ioredis/built/redis/event_handler.js:207 self.flushQueue(new errors_1.MaxRetriesPerRequestError(maxRetriesPerRequest)); ^MaxRetriesPerRequestError: Reached the max retries per request limit (which is 20). Refer to "maxRetriesPerRequest" option for details.Node.js v24.22.2
While it is disconnected, ioredis keeps every command in a queue and tries to reconnect. After 20 attempts it gives up on a command and rejects it. The event target never waits for publish, so nobody catches that rejection, and Node.js ends a process with an unhandled rejection. A Valkey restart during a deploy is enough to take the whole API down with it, one instance after the other.
Two changes fix it. Change src/pubsub.ts:
import { createPubSub } from 'graphql-yoga';
import { createRedisEventTarget } from '@graphql-yoga/redis-event-target';
import { Redis } from 'ioredis';
import type { QuestRow } from './db/schema.ts';
function connect() {
// Keep commands until Valkey is back, instead of failing them after 20 retries.
const client = new Redis(process.env.REDIS_URL!, { maxRetriesPerRequest: null });
client.on('error', (error: NodeJS.ErrnoException) => {
console.warn(`Valkey: ${error.code ?? error.message}`);
});
return client;
}
export const pubsub = createPubSub<{
'quest:posted': [quest: QuestRow];
'quest:updated': [id: number, quest: QuestRow];
}>({
eventTarget: createRedisEventTarget({
publishClient: connect(),
subscribeClient: connect(),
}),
});
export type PubSub = typeof pubsub;maxRetriesPerRequest: null keeps the commands in the queue until Valkey is back, however long that takes, so nothing is ever rejected. The listener on error replaces the stack traces with one short line per attempt.
Try it again. Start Valkey with docker compose start valkey, subscribe on port 4000, stop Valkey, post a quest, and start Valkey again after a while. The terminal of pnpm dev tells the story:
Valkey: ECONNREFUSEDValkey: ECONNREFUSEDValkey: ECONNREFUSED[15:26:06.932] INFO (66): Request processed requestId: "req-2" method: "POST" path: "/graphql" operationType: "mutation" status: 200 errorCount: 0 durationMs: 29Valkey: ECONNREFUSEDValkey: ECONNREFUSEDValkey: ECONNREFUSED
And the subscriber gets its quest once Valkey is back:
$ curl -sN localhost:4000/graphql -H 'accept: text/event-stream' -H 'content-type: application/json' -d '{"query":"subscription { questPosted { title } }"}'::event: nextdata: {"data":{"questPosted":{"title":"Posted while Valkey sleeps"}}}:
The API kept running, the mutation was answered, the event waited in the queue, and ioredis subscribed to the channels again by itself when it reconnected.
Valkey is now part of the setup, and it can do more than carry events. In the next part it holds a cache of whole responses that all instances share.