fix(api): a read-side failure waits for its event instead of losing it

The Redis consumer used to leave a failed event pending and go on to the
next ones, retrying it only at the next start. With Mongo down for a
while, a later change of the same ticket was then projected first, and
the version check dropped the earlier one for good (a status change
could vanish from the read model), besides the read side lagging until
a restart.

Now a failed event is retried in place, waiting longer each time (up to
30 s), and the next one waits for it. Failures of the infrastructure
(Mongo, Redis, the network: isTransientError) wait as long as it takes,
so the read side catches up by itself; an event that keeps failing on
its own goes, after 5 attempts, to <stream>:dead with its error, so it
does not hold up the rest.

GET /api/health reports the way from write to read (events waiting in
the outbox, pending to be projected, set aside) and says `degraded`,
still with 200, when something is stuck.

Tried for real: with Mongo stopped, two status changes went in; the
first failed to project with ECONNREFUSED, and once Mongo was back the
ticket's history had both steps, in order, without restarting the API.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

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 `<stream>: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).

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';

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<EventPipelineReading> {
+    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<number> {
+    try {
+      const summary = (await this.redis.xpending(
+        this.stream.streamKey,
+        EVENTS_BRIDGE_GROUP,
+      )) as [number, ...unknown[]];
+      return Number(summary[0] ?? 0);
+    } catch {
+      return 0;
+    }
+  }
+}

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),

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<string, string>[] = [];
+  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<string, string> = { 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<void>,
+) {
+  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);
+  });
+});

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: `<key>: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 (`<key>: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<void>;
+  /** 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<void> {
     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<boolean> {
     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<void> {
+    return new Promise((resolve) => {
+      const timer = setTimeout(done, ms);
+      function done() {
+        clearTimeout(timer);
+        resolve();
+      }
+      this.wake = done;
+    });
+  }
 }

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;
+}

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,
       ],
     };
   }