diff --git a/apps/api/openapi.json b/apps/api/openapi.json index 297b9d6..c419afa 100644 --- a/apps/api/openapi.json +++ b/apps/api/openapi.json @@ -3,6 +3,7 @@ "paths": { "/api/health": { "get": { + "description": "503 when a database is down. With the databases up, `status` is `degraded` (still 200, so a lagging read side does not get the container restarted) when events wait in the outbox over a minute, 500 or more wait to be projected, or some were set aside after failing again and again (`task-stream:events:dead`); `events.problems` says which.", "operationId": "HealthController_check", "parameters": [], "responses": { @@ -15,7 +16,14 @@ "status": "ok", "postgres": "up", "mongo": "up", - "redis": "up" + "redis": "up", + "events": { + "unpublished": 0, + "oldestUnpublishedSeconds": null, + "pending": 0, + "dead": 0, + "problems": [] + } } } } @@ -25,7 +33,7 @@ "description": "A database is down" } }, - "summary": "API and databases status", + "summary": "API, databases and the way from write to read", "tags": [ "Health" ], diff --git a/apps/api/src/app/health.controller.ts b/apps/api/src/app/health.controller.ts index d2c50b3..fd4b0df 100644 --- a/apps/api/src/app/health.controller.ts +++ b/apps/api/src/app/health.controller.ts @@ -12,7 +12,11 @@ import { ApiTags, } from '@nestjs/swagger'; import { Public } from '@task-stream/api/identity/presenters'; -import { REDIS_CLIENT } from '@task-stream/api/shared/infrastructure'; +import { + EventPipelineReading, + EventPipelineStatus, + REDIS_CLIENT, +} from '@task-stream/api/shared/infrastructure'; import Redis from 'ioredis'; import { Connection } from 'mongoose'; import { DataSource } from 'typeorm'; @@ -24,14 +28,31 @@ export class HealthController { private readonly postgres: DataSource, @InjectConnection() private readonly mongo: Connection, @Inject(REDIS_CLIENT) private readonly redis: Redis, + private readonly events: EventPipelineStatus, ) {} @Public() @Get() - @ApiOperation({ summary: 'API and databases status' }) + @ApiOperation({ + summary: 'API, databases and the way from write to read', + description: + '503 when a database is down. With the databases up, `status` is `degraded` (still 200, so a lagging read side does not get the container restarted) when events wait in the outbox over a minute, 500 or more wait to be projected, or some were set aside after failing again and again (`task-stream:events:dead`); `events.problems` says which.', + }) @ApiOkResponse({ schema: { - example: { status: 'ok', postgres: 'up', mongo: 'up', redis: 'up' }, + example: { + status: 'ok', + postgres: 'up', + mongo: 'up', + redis: 'up', + events: { + unpublished: 0, + oldestUnpublishedSeconds: null, + pending: 0, + dead: 0, + problems: [], + }, + }, }, }) @ApiServiceUnavailableResponse({ description: 'A database is down' }) @@ -44,10 +65,38 @@ export class HealthController { const [postgres, mongo, redis] = checks.map((c) => c.status === 'fulfilled' ? 'up' : 'down', ); - const result = { status: 'ok', postgres, mongo, redis }; + const events = await this.events.read().then( + (reading) => ({ ...reading, problems: problemsOf(reading) }), + () => null, + ); + const result = { + status: events?.problems.length ? 'degraded' : 'ok', + postgres, + mongo, + redis, + events, + }; if (checks.some((c) => c.status === 'rejected')) { throw new ServiceUnavailableException({ ...result, status: 'degraded' }); } return result; } } + +/** What is wrong on the way from write to read, if anything. */ +function problemsOf(reading: EventPipelineReading): string[] { + const problems: string[] = []; + if ((reading.oldestUnpublishedSeconds ?? 0) > 60) + problems.push( + `${reading.unpublished} events wait in the outbox (the oldest, ${reading.oldestUnpublishedSeconds} s): is Redis reachable?`, + ); + if (reading.pending >= 500) + problems.push( + `${reading.pending} events wait to be projected: is Mongo reachable?`, + ); + if (reading.dead > 0) + problems.push( + `${reading.dead} events were set aside after failing again and again (task-stream:events:dead)`, + ); + return problems; +} diff --git a/apps/mcp/src/app/api/taskstream.schema.ts b/apps/mcp/src/app/api/taskstream.schema.ts index 11402d1..ab8ac41 100644 --- a/apps/mcp/src/app/api/taskstream.schema.ts +++ b/apps/mcp/src/app/api/taskstream.schema.ts @@ -11,7 +11,10 @@ export interface paths { path?: never; cookie?: never; }; - /** API and databases status */ + /** + * API, databases and the way from write to read + * @description 503 when a database is down. With the databases up, `status` is `degraded` (still 200, so a lagging read side does not get the container restarted) when events wait in the outbox over a minute, 500 or more wait to be projected, or some were set aside after failing again and again (`task-stream:events:dead`); `events.problems` says which. + */ get: operations["HealthController_check"]; put?: never; post?: never; diff --git a/docs/02-arquitectura.md b/docs/02-arquitectura.md index ca16b90..d1a301d 100644 --- a/docs/02-arquitectura.md +++ b/docs/02-arquitectura.md @@ -125,12 +125,26 @@ sequenceDiagram ### De escritura a lectura: Redis +- La tabla `events` es la **bandeja de salida (outbox)**: cada evento entra en la misma transacción que la escritura, + con `published_at` vacío. Si Postgres falla, la escritura falla entera y no queda nada a medias. - El **`EventStoreRelay`** copia los eventos nuevos (`published_at IS NULL`) al Redis Stream `task-stream:events`, - en orden y con `FOR UPDATE SKIP LOCKED` (varias instancias no duplican). La entrega es _at-least-once_. + en orden y con `FOR UPDATE SKIP LOCKED` (varias instancias no duplican). La entrega es _at-least-once_. Si Redis + falla, los eventos esperan en la tabla y salen al volver. - El **`EventsBridge`** consume el stream con un _consumer group_, deserializa cada evento (`EventClsRegistry` + `@AutowiredEvent`) y lo entrega a los `@EventsHandler` registrados para su clase. A diferencia de `eventBus.subject$.next()`, **espera** a cada handler y procesa los eventos en orden; solo hace `XACK` cuando todos - terminan. Si uno falla, el evento queda pendiente y se reintenta (y `XAUTOCLAIM` recupera los de instancias caídas). + terminan. El grupo de consumidores hace de **bandeja de entrada (inbox)**, y la idempotencia la ponen las + proyecciones (versión, abajo). +- **Si un evento falla, se reintenta en su sitio** (`EventStreamConsumer`), esperando cada vez más (hasta 30 s), y el + siguiente espera: saltárselo aplicaría antes un cambio posterior y la comprobación de versión descartaría después el + anterior para siempre. Si el fallo es de la infraestructura (Mongo, Redis o la red caídos, ver `isTransientError`), + espera lo que haga falta y se pone al día solo al volver, sin reiniciar. Si es del propio evento (un fallo del + código), tras 5 intentos va al stream `task-stream:events:dead` con el error, para no bloquear el resto. Lo que + queda pendiente al parar se reintenta al arrancar (y `XAUTOCLAIM` recupera lo de instancias caídas). +- `GET /api/health` dice en `events` cuántos eventos esperan en la bandeja de salida, cuántos están pendientes de + proyectar y cuántos se apartaron; con alguno de más, `status: degraded` (pero 200: un retraso de la lectura no + reinicia el contenedor). El stream se recorta a `EVENT_STREAM_MAX_LENGTH` (un millón): solo un corte larguísimo + podría recortar algo pendiente, y Postgres siempre permite reconstruir la lectura. - Después de los handlers, el bridge ejecuta también las **sagas** (ver abajo). ### Lectura: MongoDB diff --git a/docs/07-despliegue.md b/docs/07-despliegue.md index 443abb3..71185de 100644 --- a/docs/07-despliegue.md +++ b/docs/07-despliegue.md @@ -36,7 +36,10 @@ genéricos, el DNS de Docker devuelve también los suyos y nginx reparte las pet de las llamadas a `/api/*` responden `404 Cannot GET …` y el login dice «no configurado»). Si añades un servicio, dale su alias `task-stream-*` y úsalo en las URLs. -Todos tienen healthcheck; `web` espera a que `api` esté sana, y `api` a las tres bases de datos. +Todos tienen healthcheck; `web` espera a que `api` esté sana, y `api` a las tres bases de datos. `GET /api/health` responde 503 si cae una base de datos; con +las tres en pie, `status: degraded` (200) avisa de eventos que no llegan a la lectura (`events.problems`). Los eventos +apartados por fallar una y otra vez se ven con `redis-cli XRANGE task-stream:events:dead - +`; arreglado el fallo, se +reconstruye la lectura (abajo) o se vuelven a publicar. ### Imágenes diff --git a/docs/10-futuro.md b/docs/10-futuro.md index b3fb632..0c29d8e 100644 --- a/docs/10-futuro.md +++ b/docs/10-futuro.md @@ -132,3 +132,6 @@ Mucho trabajo para lo que aportan hoy; se pueden recuperar si alguien los necesi directo y con la hora. - Panel para clientes: un enlace de solo lectura, sin cuenta, con los bloques que elijan el propietario y los admins (flujo, sprint, releases, épicas, actividad); se renueva o se retira desde Configuración. +- De escritura a lectura: si un evento no se puede proyectar, se reintenta en su sitio (el siguiente espera) hasta + que Mongo vuelve, sin reiniciar ni perder cambios; los que fallan por sí mismos se apartan a + `task-stream:events:dead`, y `GET /api/health` avisa de eventos atascados o apartados. diff --git a/docs/TaskStream-Documentacion.docx b/docs/TaskStream-Documentacion.docx index 81f02ca..f2ffb21 100644 Binary files a/docs/TaskStream-Documentacion.docx and b/docs/TaskStream-Documentacion.docx differ diff --git a/docs/TaskStream-Documentacion.pdf b/docs/TaskStream-Documentacion.pdf index a25a0bc..1e2734c 100644 Binary files a/docs/TaskStream-Documentacion.pdf and b/docs/TaskStream-Documentacion.pdf differ diff --git a/libs/api/shared/infrastructure/README.md b/libs/api/shared/infrastructure/README.md index 878e46f..1bfe9b3 100644 --- a/libs/api/shared/infrastructure/README.md +++ b/libs/api/shared/infrastructure/README.md @@ -12,6 +12,9 @@ Event store en PostgreSQL y el camino de los eventos hasta la lectura. - `EventStoreRelay`: copia los eventos nuevos al Redis Stream (`FOR UPDATE SKIP LOCKED`). - `EventsBridge`: consume el stream y entrega cada evento a sus `@EventsHandler`, en orden y esperando (ACK al final); después ejecuta los `@Saga()` (el `EventBus` en memoria nunca ve los eventos guardados) y sus comandos, uno a uno. +- `EventStreamConsumer`: reintenta en su sitio el evento que falla (el siguiente espera); los fallos de + infraestructura (`isTransientError`) sin límite, los del propio evento 5 veces y luego a `:dead`. +- `EventPipelineStatus`: bandeja de salida, pendientes y apartados, para `GET /api/health`. - Conexiones Redis (`REDIS_CLIENT`, `GRAPHQL_PUB_SUB`) y `SharedInfrastructureModule.forRootAsync`. Flujo completo: [Arquitectura](../../../../docs/02-arquitectura.md#22-flujo-completo-de-una-escritura). diff --git a/libs/api/shared/infrastructure/src/index.ts b/libs/api/shared/infrastructure/src/index.ts index eaccae2..0dc9d09 100644 --- a/libs/api/shared/infrastructure/src/index.ts +++ b/libs/api/shared/infrastructure/src/index.ts @@ -6,3 +6,4 @@ export * from './lib/event-stream/event-stream.consumer'; export * from './lib/event-store/entities/event.entity'; export * from './lib/event-store/deserializers/event.deserializer'; export * from './lib/event-store/serializers/event.serializer'; +export * from './lib/event-store/event-pipeline.status'; diff --git a/libs/api/shared/infrastructure/src/lib/event-store/event-pipeline.status.ts b/libs/api/shared/infrastructure/src/lib/event-store/event-pipeline.status.ts new file mode 100644 index 0000000..29d4df2 --- /dev/null +++ b/libs/api/shared/infrastructure/src/lib/event-store/event-pipeline.status.ts @@ -0,0 +1,66 @@ +import { Inject, Injectable } from '@nestjs/common'; +import Redis from 'ioredis'; +import { DataSource } from 'typeorm'; +import { deadLetterKey } from '../event-stream/event-stream.consumer'; +import { + EVENT_STREAM_OPTIONS, + EventStreamOptions, +} from '../event-stream/event-stream.options'; +import { REDIS_CLIENT } from '../redis/redis.tokens'; +import { EVENTS_BRIDGE_GROUP } from './events-bridge'; + +/** How write reaches read, for the health check. */ +export interface EventPipelineReading { + /** Events in Postgres not yet copied to the Redis stream (the outbox). */ + unpublished: number; + /** Seconds the oldest of them has waited (null: none). */ + oldestUnpublishedSeconds: number | null; + /** Entries handed to the read side and not yet projected (the inbox). */ + pending: number; + /** Events set aside after failing again and again. */ + dead: number; +} + +/** + * Reads the backlog at each step from the event store to the read model: + * a growing one means the read side is falling behind. + */ +@Injectable() +export class EventPipelineStatus { + constructor( + private readonly dataSource: DataSource, + @Inject(REDIS_CLIENT) private readonly redis: Redis, + @Inject(EVENT_STREAM_OPTIONS) private readonly stream: EventStreamOptions, + ) {} + + async read(): Promise { + const [outbox, pending, dead] = await Promise.all([ + this.dataSource.query( + `SELECT count(*)::int AS count, + EXTRACT(EPOCH FROM now() - min(created_at))::int AS age + FROM events WHERE published_at IS NULL`, + ) as Promise<{ count: number; age: number | null }[]>, + this.pending(), + this.redis.xlen(deadLetterKey(this.stream.streamKey)), + ]); + return { + unpublished: outbox[0]?.count ?? 0, + oldestUnpublishedSeconds: outbox[0]?.age ?? null, + pending, + dead, + }; + } + + /** XPENDING's summary; no group yet (nothing consumed) means none. */ + private async pending(): Promise { + try { + const summary = (await this.redis.xpending( + this.stream.streamKey, + EVENTS_BRIDGE_GROUP, + )) as [number, ...unknown[]]; + return Number(summary[0] ?? 0); + } catch { + return 0; + } + } +} diff --git a/libs/api/shared/infrastructure/src/lib/event-store/events-bridge.ts b/libs/api/shared/infrastructure/src/lib/event-store/events-bridge.ts index 0313f37..9b96913 100644 --- a/libs/api/shared/infrastructure/src/lib/event-store/events-bridge.ts +++ b/libs/api/shared/infrastructure/src/lib/event-store/events-bridge.ts @@ -30,6 +30,8 @@ import { RedisConnectionFactory } from '../redis/redis-connection.factory'; import { EventDeserializer } from './deserializers/event.deserializer'; export const EVENTS_BRIDGE_ENABLED = Symbol('EVENTS_BRIDGE_ENABLED'); +/** The Redis consumer group of the read side. */ +export const EVENTS_BRIDGE_GROUP = 'events-bridge'; /** Metadata written by @EventsHandler(...events) and @Saga() (@nestjs/cqrs). */ const EVENTS_HANDLER_METADATA = '__eventsHandler__'; @@ -84,7 +86,7 @@ export class EventsBridge this.connections.create('events-bridge'), { streamKey: this.stream.streamKey, - group: 'events-bridge', + group: EVENTS_BRIDGE_GROUP, consumer: hostname(), }, (event) => this.handleEvent(event), diff --git a/libs/api/shared/infrastructure/src/lib/event-stream/event-stream.consumer.spec.ts b/libs/api/shared/infrastructure/src/lib/event-stream/event-stream.consumer.spec.ts new file mode 100644 index 0000000..742d30c --- /dev/null +++ b/libs/api/shared/infrastructure/src/lib/event-stream/event-stream.consumer.spec.ts @@ -0,0 +1,178 @@ +import { SerializableEvent } from '@task-stream/api/shared/domain'; +import Redis from 'ioredis'; +import { deadLetterKey, EventStreamConsumer } from './event-stream.consumer'; +import { isTransientError } from './transient-error'; + +const KEY = 'test:events'; + +/** Just enough of a Redis stream with a consumer group. */ +class FakeRedis { + acked: string[] = []; + dead: Record[] = []; + private handedOut = false; + + constructor(private readonly entries: [string, string[]][]) {} + + async xgroup() { + return 'OK'; + } + async xautoclaim() { + return ['0-0', []]; + } + async xreadgroup(...args: string[]) { + const cursor = args[args.length - 1]; + if (cursor === '>' && !this.handedOut) { + this.handedOut = true; + return [[KEY, this.entries]]; + } + await new Promise((resolve) => setTimeout(resolve, 2)); + return null; + } + async xack(_key: string, _group: string, id: string) { + this.acked.push(id); + return 1; + } + async xadd(key: string, _id: string, ...fields: string[]) { + const entry: Record = { key }; + for (let i = 0; i < fields.length; i += 2) entry[fields[i]] = fields[i + 1]; + this.dead.push(entry); + return '1-0'; + } + disconnect() { + return undefined; + } +} + +const entry = (id: string, position: number): [string, string[]] => [ + id, + [ + 'event', + JSON.stringify({ streamId: 's', type: 'Moved', position, data: {} }), + ], +]; + +async function until(done: () => boolean) { + const deadline = Date.now() + 2000; + while (!done()) { + if (Date.now() > deadline) throw new Error('timed out'); + await new Promise((resolve) => setTimeout(resolve, 2)); + } +} + +function consumer( + redis: FakeRedis, + handle: (event: SerializableEvent) => Promise, +) { + return new EventStreamConsumer( + redis as unknown as Redis, + { + streamKey: KEY, + group: 'g', + consumer: 'c', + retryBaseMs: 1, + maxRetryMs: 2, + maxAttempts: 3, + }, + handle, + ); +} + +const mongoDown = () => + Object.assign(new Error('connection refused'), { + name: 'MongoNetworkError', + }); + +describe('EventStreamConsumer', () => { + it('retries a failing event in place: the next one waits for it', async () => { + const redis = new FakeRedis([entry('1-0', 1), entry('2-0', 2)]); + const handled: number[] = []; + let failures = 2; + const c = consumer(redis, async (event) => { + if (event.position === 1 && failures-- > 0) throw mongoDown(); + handled.push(event.position); + }); + await c.start(); + await until(() => redis.acked.length === 2); + await c.stop(); + + expect(handled).toEqual([1, 2]); + expect(redis.acked).toEqual(['1-0', '2-0']); + expect(redis.dead).toEqual([]); + }); + + it('waits out the infrastructure however long: it never sets it aside', async () => { + const redis = new FakeRedis([entry('1-0', 1)]); + let failures = 10; // well past maxAttempts + const c = consumer(redis, async () => { + if (failures-- > 0) throw mongoDown(); + }); + await c.start(); + await until(() => redis.acked.length === 1); + await c.stop(); + + expect(redis.dead).toEqual([]); + }); + + it('sets aside an event that keeps failing on its own, and goes on', async () => { + const redis = new FakeRedis([entry('1-0', 1), entry('2-0', 2)]); + const handled: number[] = []; + let calls = 0; + const c = consumer(redis, async (event) => { + if (event.position === 1) { + calls++; + throw new Error('Cannot read properties of undefined'); + } + handled.push(event.position); + }); + await c.start(); + await until(() => redis.acked.length === 2); + await c.stop(); + + expect(calls).toBe(3); + expect(handled).toEqual([2]); + expect(redis.dead).toHaveLength(1); + expect(redis.dead[0]).toMatchObject({ + key: deadLetterKey(KEY), + attempts: '3', + sourceId: '1-0', + error: 'Cannot read properties of undefined', + }); + expect(JSON.parse(redis.dead[0]['event'])).toMatchObject({ position: 1 }); + }); + + it('stopping while it retries leaves the event pending', async () => { + const redis = new FakeRedis([entry('1-0', 1)]); + let calls = 0; + const c = consumer(redis, async () => { + calls++; + throw mongoDown(); + }); + await c.start(); + await until(() => calls >= 2); + await c.stop(); + + expect(redis.acked).toEqual([]); + expect(redis.dead).toEqual([]); + }); +}); + +describe('isTransientError', () => { + it('tells the infrastructure being down from a fault of its own', () => { + expect(isTransientError(mongoDown())).toBe(true); + expect( + isTransientError(Object.assign(new Error('x'), { code: 'ECONNRESET' })), + ).toBe(true); + expect( + isTransientError( + Object.assign(new Error('write failed'), { + cause: Object.assign(new Error('y'), { + name: 'MongoServerSelectionError', + }), + }), + ), + ).toBe(true); + expect(isTransientError(new Error('Connection is closed.'))).toBe(true); + expect(isTransientError(new TypeError('x is undefined'))).toBe(false); + expect(isTransientError(new Error('E11000 duplicate key'))).toBe(false); + }); +}); diff --git a/libs/api/shared/infrastructure/src/lib/event-stream/event-stream.consumer.ts b/libs/api/shared/infrastructure/src/lib/event-stream/event-stream.consumer.ts index c1d33d3..f16c8d7 100644 --- a/libs/api/shared/infrastructure/src/lib/event-stream/event-stream.consumer.ts +++ b/libs/api/shared/infrastructure/src/lib/event-stream/event-stream.consumer.ts @@ -1,6 +1,7 @@ import { Logger } from '@nestjs/common'; import Redis from 'ioredis'; import { SerializableEvent } from '@task-stream/api/shared/domain'; +import { isTransientError } from './transient-error'; type StreamEntry = [id: string, fields: string[]]; type StreamReply = [stream: string, entries: StreamEntry[]][] | null; @@ -11,17 +12,35 @@ export interface EventStreamConsumerOptions { consumer: string; batchSize?: number; blockMs?: number; + /** First wait before retrying an event; it doubles up to `maxRetryMs`. */ + retryBaseMs?: number; + maxRetryMs?: number; + /** Attempts of an event that keeps failing for a reason of its own. */ + maxAttempts?: number; } +/** Where events that keep failing go, next to the stream: `:dead`. */ +export const deadLetterKey = (streamKey: string) => `${streamKey}:dead`; + /** - * Reads the event stream with a Redis consumer group. Entries are ACKed only - * after the handler succeeds; failed ones stay pending and are retried on + * Reads the event stream with a Redis consumer group (the read side's inbox). + * Entries are ACKed only after the handler succeeds. + * + * An event that fails is retried in place, waiting longer each time, and the + * next one waits for it: skipping ahead would apply a later change first, and + * the version check of the projections would then drop the earlier one for + * good. When the infrastructure is down (Mongo, Redis, the network) it waits + * as long as it takes; when the event itself keeps failing, after + * `maxAttempts` it goes to the dead-letter stream (`:dead`) so that it + * does not hold up the rest. Entries still pending at a stop are retried on * the next start (or by another consumer of the group). */ export class EventStreamConsumer { private readonly logger: Logger; private running = false; private loop?: Promise; + /** Ends the current wait between retries (on stop). */ + private wake?: () => void; constructor( private readonly redis: Redis, @@ -39,6 +58,7 @@ export class EventStreamConsumer { async stop(): Promise { this.running = false; + this.wake?.(); this.redis.disconnect(); await this.loop?.catch(() => undefined); } @@ -108,15 +128,98 @@ export class EventStreamConsumer { return; } const event = JSON.parse(fields[index + 1]) as SerializableEvent; + const name = `${event.type} (${event.streamId}#${event.position})`; + const maxAttempts = this.options.maxAttempts ?? 5; + + for (let attempt = 1; this.running; attempt++) { + try { + await this.handle(event); + await this.redis.xack(this.options.streamKey, this.options.group, id); + if (attempt > 1) + this.logger.log(`${name} handled on attempt ${attempt}`); + return; + } catch (error) { + const transient = isTransientError(error); + if (!transient && attempt >= maxAttempts) { + if ( + await this.deadLetter(id, fields[index + 1], name, error, attempt) + ) + return; + } + const wait = this.backoff(attempt); + const reason = (error as Error).message; + if (attempt === 1) + this.logger.error( + `Failed to handle ${name}: ${reason}; retrying in ${wait} ms`, + (error as Error).stack, + ); + else + this.logger.warn( + `${name} still failing (attempt ${attempt}${transient ? ', infrastructure' : `/${maxAttempts}`}): ${reason}; retrying in ${wait} ms`, + ); + await this.sleep(wait); + } + } + // Stopped while retrying: it stays pending for the next start. + } + + /** + * Moves an event that keeps failing aside, with why, and lets the stream + * go on. False if even that failed (Redis down): it is then retried. + */ + private async deadLetter( + id: string, + payload: string, + name: string, + error: unknown, + attempts: number, + ): Promise { try { - await this.handle(event); - await this.redis.xack(this.options.streamKey, this.options.group, id); - } catch (error) { - // Left pending: it will be retried after a restart. - this.logger.error( - `Failed to handle ${event.type} (${event.streamId}#${event.position}): ${(error as Error).message}`, - (error as Error).stack, + await this.redis.xadd( + deadLetterKey(this.options.streamKey), + '*', + 'event', + payload, + 'error', + (error as Error).message ?? String(error), + 'attempts', + String(attempts), + 'group', + this.options.group, + 'sourceId', + id, + 'failedAt', + new Date().toISOString(), ); + await this.redis.xack(this.options.streamKey, this.options.group, id); + this.logger.error( + `${name} failed ${attempts} times and went to ${deadLetterKey(this.options.streamKey)}: ${(error as Error).message}`, + ); + return true; + } catch (failure) { + this.logger.error( + `Could not set ${name} aside: ${(failure as Error).message}`, + ); + return false; } } + + private backoff(attempt: number): number { + const base = this.options.retryBaseMs ?? 500; + return Math.min( + this.options.maxRetryMs ?? 30_000, + base * 2 ** (attempt - 1), + ); + } + + private sleep(ms: number): Promise { + return new Promise((resolve) => { + const timer = setTimeout(done, ms); + function done() { + clearTimeout(timer); + resolve(); + } + this.wake = done; + }); + } } diff --git a/libs/api/shared/infrastructure/src/lib/event-stream/transient-error.ts b/libs/api/shared/infrastructure/src/lib/event-stream/transient-error.ts new file mode 100644 index 0000000..c6ecf02 --- /dev/null +++ b/libs/api/shared/infrastructure/src/lib/event-stream/transient-error.ts @@ -0,0 +1,36 @@ +/** Errors of the infrastructure itself: waiting is the cure, not skipping. */ +const TRANSIENT_NAMES = + /^(Mongo(Network|NetworkTimeout|ServerSelection|NotConnected|TopologyClosed|PoolCleared)Error|MaxRetriesPerRequestError)$/; +const TRANSIENT_CODES = new Set([ + 'ECONNREFUSED', + 'ECONNRESET', + 'ETIMEDOUT', + 'EPIPE', + 'EHOSTUNREACH', + 'ENETUNREACH', + 'ENOTFOUND', + 'EAI_AGAIN', +]); +const TRANSIENT_MESSAGES = + /connection (is )?closed|connection refused|timed out|getaddrinfo|socket hang up/i; + +/** + * Whether an error comes from a database or the network being down (and + * will go away by itself) rather than from the event or the code that + * handles it. Follows `cause`, as drivers wrap the original error. + */ +export function isTransientError(error: unknown): boolean { + for (let current = error, depth = 0; current && depth < 5; depth++) { + const e = current as { + name?: string; + code?: string; + message?: string; + cause?: unknown; + }; + if (e.name && TRANSIENT_NAMES.test(e.name)) return true; + if (e.code && TRANSIENT_CODES.has(e.code)) return true; + if (e.message && TRANSIENT_MESSAGES.test(e.message)) return true; + current = e.cause; + } + return false; +} diff --git a/libs/api/shared/infrastructure/src/lib/shared-infrastructure.module.ts b/libs/api/shared/infrastructure/src/lib/shared-infrastructure.module.ts index f2b920e..4ee1ecf 100644 --- a/libs/api/shared/infrastructure/src/lib/shared-infrastructure.module.ts +++ b/libs/api/shared/infrastructure/src/lib/shared-infrastructure.module.ts @@ -6,6 +6,7 @@ import { EventStore } from '@task-stream/api/shared/application'; import { RedisPubSub } from 'graphql-redis-subscriptions'; import { EventDeserializer } from './event-store/deserializers/event.deserializer'; import { EventEntity } from './event-store/entities/event.entity'; +import { EventPipelineStatus } from './event-store/event-pipeline.status'; import { EventStoreRelay } from './event-store/event-store.relay'; import { EventStoreSignal } from './event-store/event-store.signal'; import { @@ -103,6 +104,7 @@ export class SharedInfrastructureModule { EventStreamPublisher, EventStoreRelay, EventsBridge, + EventPipelineStatus, ], exports: [ EventStore, @@ -111,6 +113,7 @@ export class SharedInfrastructureModule { REDIS_CLIENT, GRAPHQL_PUB_SUB, EVENT_STREAM_OPTIONS, + EventPipelineStatus, ], }; }