Eine schnelle GraphQL-API mit Fastify und Yoga · Teil 6 von 8 der Serie
Teil 6: Subscriptions über mehrere Instanzen mit Valkey
Starte eine zweite Instanz der API, sieh zu, wie ihre Abonnenten die Events der ersten verpassen, und teil die Events über Valkey, damit jede Instanz jede Änderung mitbekommt, selbst nach einem Neustart von Valkey.
Quest-Ziele
- Verstehen, warum ein PubSub im Speicher an der Grenze seines Prozesses endet
- Valkey neben PostgreSQL betreiben
- Die Events von Yoga mit dem Redis-Event-Target über Valkey schicken
- Die API am Laufen halten, wenn Valkey weg ist
Ein Node.js-Prozess führt JavaScript auf einem CPU-Kern aus. Um mehr Kerne zu nutzen, ohne Ausfallzeit zu deployen oder die Last auf mehrere Maschinen zu verteilen, startest du die API mehrmals, mit einem Load Balancer davor. Dann kann eine Subscription bei einer Instanz landen und die Mutation, auf die sie wartet, bei einer anderen.
Zwei Instanzen
Setz das Brett zurück, damit deine Ausgabe zu der hier passt:
$ LOG_SQL=false pnpm db:seedSeeded 3 guilds, 9 adventurers and 10 quests.
Lass pnpm dev auf Port 4000 laufen und starte in einem anderen Terminal eine zweite Instanz auf Port 4001. Eine Variable aus der Shell gewinnt gegen die aus der .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"}
Abonnier bei der zweiten Instanz:
$ curl -sN localhost:4001/graphql -H 'accept: text/event-stream' -H 'content-type: application/json' -d '{"query":"subscription { questPosted { title } }"}':
Und schreib bei der ersten eine Quest aus:
$ 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"}}}
Die Quest ist ausgeschrieben, und der Abonnent auf Port 4001 hört nichts. Der PubSub aus dem letzten Teil ist eine Liste von Zuhörern im Speicher seines Prozesses, und die Instanz auf Port 4000 hat keinen Weg zu den Zuhörern der Instanz auf 4001.
Valkey
Die Instanzen brauchen etwas in der Mitte, das jede von ihnen erreicht. Valkey ist ein quelloffener Key-Value-Speicher, der aus Redis hervorgegangen ist und dasselbe Protokoll spricht, jeder Redis-Client funktioniert also damit. Sein Pub/Sub schickt eine Nachricht, die auf einem Kanal veröffentlicht wird, an jede Verbindung, die ihn abonniert hat. Ergänz es in der 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
Schreib seine Adresse in die .env:
REDIS_URL=redis://localhost:6379Der PubSub von Yoga nimmt ein Event-Target, den Teil, der die Events tatsächlich zustellt, und es gibt eins für Redis. Installier es mit ioredis, dem Client, den es nutzt:
$ 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
Das @5 ist wichtig. ioredis 6 ist erschienen, aber das Event-Target erwartet noch ioredis 5. Gib dem PubSub jetzt in src/pubsub.ts sein Event-Target:
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;Es braucht zwei Verbindungen, weil eine Redis-Verbindung, die einen Kanal abonniert hat, nichts anderes mehr kann, nicht einmal veröffentlichen. Sonst ändert sich nichts. Die Resolver veröffentlichen und abonnieren wie vorher, und das Event-Target schickt die Events über Valkey statt über den Speicher.
pnpm dev hat von selbst neu gestartet. Stopp die zweite Instanz mit Strg+C, start sie wieder, abonnier noch einmal auf Port 4001 und schreib dieselbe Quest auf Port 4000 aus:
$ 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"}}}
Das Event ist von einer Instanz zur anderen gelangt. Solange eine Subscription offen ist, siehst du ihre Kanäle in Valkey. Mit einem Brett auf questPosted und einem auf questUpdated(id: 4):
$ docker compose exec valkey valkey-cli pubsub channelsquest:updated:4quest:posted
Ein Topic wird zu einem eigenen Kanal, quest:updated:4, Valkey schickt ein Update also nur an die Instanzen, bei denen jemand darauf wartet.
Wenn Valkey weg ist
Die API hängt jetzt von Valkey ab, also finde heraus, was ohne passiert. Stopp es:
$ docker compose stop valkey Container guild-api-valkey-1 Stopping Container guild-api-valkey-1 Stopped
Dann schreib eine Quest bei der zweiten Instanz aus, auf Port 4001. Die Mutation antwortet wie immer. Sieben Sekunden später ist die Instanz weg. Das ist ihr Terminal, gekürzt, weil ioredis seine Warnung bei jedem Versuch wiederholt, sich neu zu verbinden:
[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
Solange die Verbindung weg ist, hält ioredis jeden Befehl in einer Warteschlange und versucht, sich neu zu verbinden. Nach 20 Versuchen gibt es einen Befehl auf und lehnt ihn ab. Das Event-Target wartet nie auf publish, also fängt niemand diese Ablehnung, und Node.js beendet einen Prozess mit einer unbehandelten Ablehnung. Ein Neustart von Valkey während eines Deployments reicht, um die ganze API mitzureißen, eine Instanz nach der anderen.
Zwei Änderungen beheben das. Änder 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 hält die Befehle in der Warteschlange, bis Valkey zurück ist, egal wie lange das dauert, es wird also nie etwas abgelehnt. Der Listener auf error ersetzt die Stacktraces durch eine kurze Zeile pro Versuch.
Probier es noch einmal. Start Valkey mit docker compose start valkey, abonnier auf Port 4000, stopp Valkey, schreib eine Quest aus und start Valkey nach einer Weile wieder. Das Terminal von pnpm dev erzählt die Geschichte:
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
Und der Abonnent bekommt seine Quest, sobald Valkey zurück ist:
$ 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"}}}:
Die API lief weiter, die Mutation wurde beantwortet, das Event hat in der Warteschlange gewartet, und ioredis hat die Kanäle beim Neuverbinden von selbst wieder abonniert.
Valkey gehört jetzt zum Aufbau, und es kann mehr als Events tragen. Im nächsten Teil hält es einen Cache ganzer Antworten, den sich alle Instanzen teilen.