Improve worker resilience with RabbitMQ reconnects, structured ingest logging, retry vs DLQ error handling, and safer backend delivery logic

Author: minkinad

.env.example

@@ -1,8 +1,18 @@
-NODE_ENV=development
 RABBITMQ_URL=amqp://app:app@localhost:5672
-REDIS_URL=redis://localhost:6379
-API_GRAPHQL_URL=http://localhost:3000/graphql
 
 QUEUE_RAW_EVENT=source.raw.v1
+QUEUE_RETRY_EVENT=source.raw.retry.v1
+QUEUE_DEAD_LETTER_EVENT=source.raw.dlq.v1
 QUEUE_NORMALIZED_EVENT=source.normalized.v1
+
+API_BASE_URL=http://localhost:3000
+GRAPHQL_PATH=/graphql
+# API_GRAPHQL_URL=http://localhost:3000/graphql
+API_INGEST_TOKEN=replace_me_ingest_token
+
 SHARED_CONTRACTS_DIR=../shared-contracts
+
+RETRY_ATTEMPTS=5
+RETRY_BASE_DELAY_MS=5000
+PREFETCH=10
+LOG_LEVEL=info

Dockerfile

@@ -1,7 +1,7 @@
 FROM node:20-alpine AS deps
 WORKDIR /app
 COPY package.json package-lock.json* ./
-RUN npm install
+RUN npm ci --fetch-retries 5 --fetch-retry-mintimeout 20000 --fetch-retry-maxtimeout 120000
 
 FROM deps AS build
 WORKDIR /app

README.md

@@ -11,7 +11,9 @@
 - читает `source.raw.v1` из RabbitMQ;
 - валидирует raw-события;
 - нормализует payload до формата `source.normalized.v1`;
-- публикует нормализованные события и отправляет ingest mutation в `backend-api`.
+- отправляет ingest mutation `ingestNormalizedItem` в `backend-api` с заголовком `x-ingest-token`;
+- после успешного ingest публикует нормализованные события в `source.normalized.v1`;
+- при transient ошибках отправляет сообщение в retry queue, а poison messages складывает в DLQ.
 
 ## Черновая реализация
 
@@ -21,6 +23,20 @@
 - GraphQL-клиент отправки в backend (`src/backend-client.ts`);
 - Dockerfile и CI workflow.
 
+## Переменные окружения
+
+- `RABBITMQ_URL` - адрес RabbitMQ.
+- `QUEUE_RAW_EVENT` - основная очередь входящих raw-событий, по умолчанию `source.raw.v1`.
+- `QUEUE_RETRY_EVENT` - очередь retry, по умолчанию `source.raw.retry.v1`.
+- `QUEUE_DEAD_LETTER_EVENT` - очередь dead-letter, по умолчанию `source.raw.dlq.v1`.
+- `QUEUE_NORMALIZED_EVENT` - очередь нормализованных событий, по умолчанию `source.normalized.v1`.
+- `API_BASE_URL` - базовый URL `backend-api`, по умолчанию `http://localhost:3000`.
+- `GRAPHQL_PATH` - GraphQL path, по умолчанию `/graphql`.
+- `API_GRAPHQL_URL` - явный GraphQL URL. Если задан, имеет приоритет над `API_BASE_URL + GRAPHQL_PATH`.
+- `API_INGEST_TOKEN` - токен для заголовка `x-ingest-token`. Должен совпадать с `INGEST_API_TOKEN` в `backend-api`.
+- `SHARED_CONTRACTS_DIR` - путь к `shared-contracts`.
+- `RETRY_ATTEMPTS`, `RETRY_BASE_DELAY_MS`, `PREFETCH`, `LOG_LEVEL` - параметры обработки и логирования.
+
 ## Локальный запуск
 
 ```bash
@@ -29,13 +45,26 @@ npm install
 npm run start:dev
 ```
 
-## Важные переменные
+Для Docker Compose сервис ожидает:
+
+- RabbitMQ на `RABBITMQ_URL`;
+- `backend-api` на `API_BASE_URL` или `API_GRAPHQL_URL`;
+- смонтированный `shared-contracts` в `SHARED_CONTRACTS_DIR`.
+
+## Ожидаемые очереди
+
+- `source.raw.v1` - входящие события.
+- `source.raw.retry.v1` - delayed retry queue.
+- `source.raw.dlq.v1` - dead-letter queue.
+- `source.normalized.v1` - успешно обработанные нормализованные события.
+
+## Как проверить обработку
 
-- `RABBITMQ_URL`
-- `QUEUE_RAW_EVENT`
-- `QUEUE_NORMALIZED_EVENT`
-- `API_GRAPHQL_URL`
-- `SHARED_CONTRACTS_DIR`
+1. Опубликуй валидное сообщение в `source.raw.v1`.
+2. В логах worker должны появиться строки `connected to rabbitmq`, `consuming queue`, `raw event validated`, `normalized event created`, `ingest success`, `published normalized event`, `message acknowledged`.
+3. Проверь, что сообщение появилось в `source.normalized.v1`, а запись создалась в `backend-api`.
+4. Для transient ошибки временно выключи `backend-api`: worker должен логировать `ingest failed` и `retry scheduled`, а не падать.
+5. Для poison message отправь невалидный JSON или payload, нарушающий schema: worker должен логировать `message dead-lettered`, сообщение должно уйти в `source.raw.dlq.v1`.
 
 ## Связи с другими репозиториями
 

src/backend-client.ts

@@ -1,3 +1,4 @@
+import { PoisonMessageError, TransientMessageError, toErrorMessage } from "./errors";
 import type { NormalizedSourceEvent } from "./types";
 
 const INGEST_MUTATION = `
@@ -15,45 +16,83 @@ export async function sendToBackend(
   ingestToken: string,
   event: NormalizedSourceEvent
 ): Promise<void> {
-  const response = await fetch(graphqlUrl, {
-    method: "POST",
-    headers: {
-      "content-type": "application/json",
-      "x-ingest-token": ingestToken
-    },
-    body: JSON.stringify({
-      query: INGEST_MUTATION,
-      variables: {
-        input: {
-          externalId: event.externalId,
-          source: event.source,
-          title: event.title,
-          description: event.description,
-          customer: event.customer,
-          supplier: event.supplier,
-          amount: event.amount,
-          currency: event.currency,
-          publishedAt: event.publishedAt,
-          deadlineAt: event.deadlineAt,
-          payloadVersion: event.payloadVersion,
-          sourceUrl: event.sourceUrl,
-          status: event.status,
-          rawPayload: { rawRef: event.rawRef, eventId: event.eventId },
-          rawEvent: event.rawEvent
+  let response: Response;
+
+  try {
+    response = await fetch(graphqlUrl, {
+      method: "POST",
+      signal: AbortSignal.timeout(10000),
+      headers: {
+        "content-type": "application/json",
+        "x-ingest-token": ingestToken
+      },
+      body: JSON.stringify({
+        query: INGEST_MUTATION,
+        variables: {
+          input: {
+            externalId: event.externalId,
+            source: event.source,
+            title: event.title,
+            description: event.description,
+            customer: event.customer,
+            supplier: event.supplier,
+            amount: event.amount,
+            currency: event.currency,
+            publishedAt: event.publishedAt,
+            deadlineAt: event.deadlineAt,
+            payloadVersion: event.payloadVersion,
+            sourceUrl: event.sourceUrl,
+            status: event.status,
+            rawPayload: { rawRef: event.rawRef, eventId: event.eventId },
+            rawEvent: event.rawEvent
+          }
         }
-      }
-    })
-  });
+      })
+    });
+  } catch (error) {
+    throw new TransientMessageError(
+      `Backend API is unavailable at ${graphqlUrl}: ${toErrorMessage(error)}`,
+      { cause: error }
+    );
+  }
 
   if (!response.ok) {
-    throw new Error(`Backend API ответил с кодом ${response.status}`);
+    const message = `Backend API responded with HTTP ${response.status}`;
+    if (response.status >= 500 || response.status === 429 || response.status === 408) {
+      throw new TransientMessageError(message);
+    }
+
+    throw new PoisonMessageError(message);
   }
 
-  const payload = (await response.json()) as {
-    errors?: Array<{ message: string }>;
-  };
+  let payload: { errors?: Array<{ message: string }> };
+
+  try {
+    payload = (await response.json()) as {
+      errors?: Array<{ message: string }>;
+    };
+  } catch (error) {
+    throw new TransientMessageError("Backend API returned invalid JSON", { cause: error });
+  }
 
   if (payload.errors?.length) {
-    throw new Error(`Ошибка GraphQL ingest: ${payload.errors.map((e) => e.message).join("; ")}`);
+    const message = `GraphQL ingest failed: ${payload.errors.map((entry) => entry.message).join("; ")}`;
+    if (isPoisonGraphqlError(message)) {
+      throw new PoisonMessageError(message);
+    }
+
+    throw new TransientMessageError(message);
   }
 }
+
+function isPoisonGraphqlError(message: string): boolean {
+  const normalized = message.toLowerCase();
+
+  return (
+    normalized.includes("unauthorized") ||
+    normalized.includes("forbidden") ||
+    normalized.includes("invalid") ||
+    normalized.includes("bad request") ||
+    normalized.includes("validation")
+  );
+}

src/config.ts

@@ -3,7 +3,9 @@ import { z } from "zod";
 
 const envSchema = z.object({
   RABBITMQ_URL: z.string().default("amqp://app:app@localhost:5672"),
-  API_GRAPHQL_URL: z.string().default("http://localhost:3000/graphql"),
+  API_BASE_URL: z.string().url().optional(),
+  API_GRAPHQL_URL: z.string().url().optional(),
+  GRAPHQL_PATH: z.string().default("/graphql"),
   API_INGEST_TOKEN: z.string().default("replace_me_ingest_token"),
   QUEUE_RAW_EVENT: z.string().default("source.raw.v1"),
   QUEUE_RETRY_EVENT: z.string().default("source.raw.retry.v1"),
@@ -15,4 +17,15 @@ const envSchema = z.object({
   PREFETCH: z.coerce.number().int().positive().default(10)
 });
 
-export const config = envSchema.parse(process.env);
+const parsedEnv = envSchema.parse(process.env);
+
+const apiBaseUrl = parsedEnv.API_BASE_URL ?? "http://localhost:3000";
+const graphqlPath = parsedEnv.GRAPHQL_PATH.startsWith("/")
+  ? parsedEnv.GRAPHQL_PATH
+  : `/${parsedEnv.GRAPHQL_PATH}`;
+
+export const config = {
+  ...parsedEnv,
+  API_BASE_URL: apiBaseUrl,
+  API_GRAPHQL_URL: parsedEnv.API_GRAPHQL_URL ?? new URL(graphqlPath, apiBaseUrl).toString()
+};

src/contracts/schema-validator.ts

@@ -1,7 +1,8 @@
 import { readFileSync } from "node:fs";
 import { join } from "node:path";
-import Ajv from "ajv";
+import Ajv2020 from "ajv/dist/2020";
 import addFormats from "ajv-formats";
+import { PoisonMessageError } from "../errors";
 import type { NormalizedSourceEvent, RawSourceEvent } from "../types";
 
 export function createSchemaValidators(sharedContractsDir: string) {
@@ -12,7 +13,7 @@ export function createSchemaValidators(sharedContractsDir: string) {
     readFileSync(join(sharedContractsDir, "events", "source-normalized.v1.json"), "utf-8")
   );
 
-  const ajv = new Ajv({ allErrors: true });
+  const ajv = new Ajv2020({ allErrors: true });
   addFormats(ajv);
   const rawValidate = ajv.compile<RawSourceEvent>(rawSchema);
   const normalizedValidate = ajv.compile<NormalizedSourceEvent>(normalizedSchema);
@@ -22,7 +23,7 @@ export function createSchemaValidators(sharedContractsDir: string) {
       if (rawValidate(event)) {
         return;
       }
-      throw new Error(
+      throw new PoisonMessageError(
         `Raw schema validation failed: ${(rawValidate.errors ?? [])
           .map((err) => `${err.instancePath || "/"} ${err.message}`)
           .join("; ")}`
@@ -32,7 +33,7 @@ export function createSchemaValidators(sharedContractsDir: string) {
       if (normalizedValidate(event)) {
         return;
       }
-      throw new Error(
+      throw new PoisonMessageError(
         `Normalized schema validation failed: ${(normalizedValidate.errors ?? [])
           .map((err) => `${err.instancePath || "/"} ${err.message}`)
           .join("; ")}`

src/errors.ts

@@ -0,0 +1,38 @@
+export class WorkerError extends Error {
+  constructor(
+    message: string,
+    public readonly kind: "poison" | "transient",
+    options?: { cause?: unknown }
+  ) {
+    super(message, options);
+    this.name = new.target.name;
+  }
+}
+
+export class PoisonMessageError extends WorkerError {
+  constructor(message: string, options?: { cause?: unknown }) {
+    super(message, "poison", options);
+  }
+}
+
+export class TransientMessageError extends WorkerError {
+  constructor(message: string, options?: { cause?: unknown }) {
+    super(message, "transient", options);
+  }
+}
+
+export function isPoisonMessageError(error: unknown): error is PoisonMessageError {
+  return error instanceof WorkerError && error.kind === "poison";
+}
+
+export function isTransientMessageError(error: unknown): error is TransientMessageError {
+  return error instanceof WorkerError && error.kind === "transient";
+}
+
+export function toErrorMessage(error: unknown): string {
+  if (error instanceof Error) {
+    return error.message;
+  }
+
+  return String(error);
+}