redis-stream.ts

redis-stream.tsapps/worker/src/lib/redis-stream.tsraw ← lib7,566 B← Back to the app
1/**
2 * apps/worker — domain-event stream consumer (Redis Streams, XREADGROUP).
3 *
4 * The API publishes every domain event to the stream `factory:events` as a
5 * flat field map `{event, payload (JSON string), ts}`. Each queue gets its
6 * OWN consumer group (`worker-<queue>`), so queues can lag independently
7 * without head-of-line blocking each other.
8 *
9 * Delivery semantics:
10 *  - XGROUP CREATE ... MKSTREAM (tolerates BUSYGROUP on restart),
11 *  - loop: XREADGROUP GROUP g c COUNT 50 BLOCK 5000 → onEvent(id, envelope)
12 *    → XACK on success; on error the entry stays PENDING (not acked) and is
13 *    redelivered by the XAUTOCLAIM loop below — at-least-once, exactly-once
14 *    is the processor's job (all handlers must be idempotent).
15 *  - XAUTOCLAIM loop every 60s claims entries idle > 60s (crashed consumer /
16 *    died between processing and ACK), max 20 per pass.
17 *  - stop() resolves cleanly; in-flight entries finish or stay pending.
18 */
19import type { Redis } from "ioredis";
20import { getLogger } from "../logger.js";
21 
22export interface StreamEntry {
23  id: string;
24  fields: Record<string, string | undefined>;
25}
26 
27/**
28 * XREADGROUP reply: an array of [streamName, entries] pairs where entries is
29 * a list of [id, [field, value, …]] pairs — or `null` when the BLOCK window
30 * expires with no new entries. (ioredis types the command loosely; we model
31 * the shape we actually consume.)
32 */
33type ReadStreamResult = [string, [string, string[]][]][] | null;
34 
35export type StreamEventHandler = (entry: StreamEntry) => Promise<void>;
36 
37interface ClaimedEntry {
38  id: string;
39  fields: Record<string, string | undefined>;
40}
41 
42const COUNT = 50;
43const BLOCK_MS = 5000;
44const CLAIM_INTERVAL_MS = 60_000;
45const CLAIM_MIN_IDLE_MS = 60_000;
46const CLAIM_BATCH = 20;
47 
48const log = getLogger("stream");
49 
50export class StreamConsumer {
51  private stopped = false;
52  private processing = 0;
53  private claimTimer: NodeJS.Timeout | null = null;
54  // Reserved for unblocking an in-flight XREADGROUP on stop(). Currently
55  // always null: the read loop exits via the `stopped` flag + the command's
56  // own BLOCK expiry, so no resolver is wired. Kept so `stop()`'s guard
57  // compiles and a future fast-stop can assign a resolver here.
58  private readResolve: (() => void) | null = null;
59 
60  constructor(
61    private redis: Redis,
62    private stream: string,
63    private group: string,
64    private consumerId: string,
65    private onEvent: StreamEventHandler
66  ) {}
67 
68  private get log() {
69    return log.child({ group: this.group });
70  }
71 
72  /** Idempotent: safe to call stop() multiple times. */
73  get isStopped(): boolean {
74    return this.stopped;
75  }
76 
77  async start(): Promise<void> {
78    // 1) Ensure the consumer group exists (BUSYGROUP = already there).
79    try {
80      await this.redis.xgroup("CREATE", this.stream, this.group, "0", "MKSTREAM");
81      this.log.info(`consumer group created: ${this.group} @ ${this.stream}`);
82    } catch (err) {
83      const msg = String((err as Error).message ?? err);
84      if (!msg.includes("BUSYGROUP")) throw err;
85    }
86 
87    // 2) XAUTOCLAIM maintenance loop.
88    this.claimTimer = setInterval(() => {
89      void this.claimStale();
90    }, CLAIM_INTERVAL_MS);
91    if (typeof this.claimTimer.unref === "function") this.claimTimer.unref();
92 
93    // 3) Read loop.
94    void this.readLoop();
95  }
96 
97  async stop(): Promise<void> {
98    if (this.stopped) return;
99    this.stopped = true;
100    if (this.claimTimer) clearInterval(this.claimTimer);
101    // Unblock an in-flight XREADGROUP so the loop can exit.
102    if (this.readResolve) {
103      this.readResolve();
104      this.readResolve = null;
105    }
106    // Give in-flight handlers a moment to finish (they XACK or stay pending).
107    for (let i = 0; i < 50 && this.processing > 0; i++) {
108      await new Promise((r) => setTimeout(r, 100));
109    }
110    this.log.info("consumer stopped");
111  }
112 
113  private async readLoop(): Promise<void> {
114    const lastId = ">";
115    while (!this.stopped) {
116      let entries: [string, string[]][];
117      try {
118        const res = await this.readWithStopSignal(lastId);
119        // ioredis returns [[streamName, [[id, [k,v,...]], ...]]] (one pair)
120        // or null when BLOCK expires with no new entries.
121        entries = res?.[0]?.[1] ?? [];
122      } catch (err) {
123        if (this.stopped) break;
124        this.log.warn(
125          { err: (err as Error).message },
126          "XREADGROUP failed — backoff 2s"
127        );
128        await sleep(2000);
129        continue;
130      }
131      if (entries.length === 0) continue;
132 
133      for (const [id, fields] of entries) {
134        if (this.stopped) break;
135        const map: Record<string, string | undefined> = {};
136        for (let i = 0; i < fields.length; i += 2) {
137          const key = fields[i];
138          if (key === undefined) continue; // defensive: malformed field list
139          map[key] = fields[i + 1];
140        }
141        const entry: StreamEntry = { id, fields: map };
142        this.processing++;
143        try {
144          await this.onEvent(entry);
145          await this.redis.xack(this.stream, this.group, id);
146        } catch (err) {
147          // NOT acked: entry stays pending → redelivered by XAUTOCLAIM.
148          this.log.error(
149            { err: (err as Error).message, id, group: this.group },
150            "handler failed — leaving entry pending for redelivery"
151          );
152        } finally {
153          this.processing--;
154        }
155      }
156    }
157  }
158 
159  /**
160   * XREADGROUP with the stop-signal trick: a racing `stop()` can't wedge the
161   * loop on a 5s BLOCK, because we race the command against a stop flag
162   * (the command itself keeps running on the socket, harmlessly).
163   */
164  private readWithStopSignal(lastId: string): Promise<ReadStreamResult> {
165    return new Promise((resolve, reject) => {
166      if (this.stopped) {
167        resolve(null);
168        return;
169      }
170      const cmd = this.redis.xreadgroup(
171        "GROUP",
172        this.group,
173        this.consumerId,
174        "COUNT",
175        COUNT,
176        "BLOCK",
177        BLOCK_MS,
178        "STREAMS",
179        this.stream,
180        lastId
181      ) as Promise<ReadStreamResult>;
182      cmd.then(resolve).catch(reject);
183    });
184  }
185 
186  /**
187   * Claim entries that went idle (crashed / pending > 60s) and re-run them
188   * through the same handler. XAUTOCLAIM returns [nextStartId, entries,
189   * deletedIds].
190   */
191  private async claimStale(): Promise<void> {
192    if (this.stopped) return;
193    try {
194      const res = await this.redis.call(
195        "XAUTOCLAIM",
196        this.stream,
197        this.group,
198        this.consumerId,
199        CLAIM_MIN_IDLE_MS,
200        "0",
201        "COUNT",
202        CLAIM_BATCH
203      ) as [string, ClaimedEntry[], (string | null)[]];
204      const entries = Array.isArray(res?.[1]) ? res[1] : [];
205      if (entries.length === 0) return;
206      this.log.warn({ count: entries.length }, "claiming stale pending entries");
207      for (const e of entries) {
208        if (this.stopped) break;
209        this.processing++;
210        try {
211          await this.onEvent(e);
212          await this.redis.xack(this.stream, this.group, e.id);
213        } catch (err) {
214          this.log.error(
215            { err: (err as Error).message, id: e.id },
216            "claimed entry failed again — stays pending"
217          );
218        } finally {
219          this.processing--;
220        }
221      }
222    } catch (err) {
223      this.log.warn(
224        { err: (err as Error).message },
225        "XAUTOCLAIM failed (stream may not exist yet)"
226      );
227    }
228  }
229}
230 
231function sleep(ms: number): Promise<void> {
232  return new Promise((r) => setTimeout(r, ms));
233}
234