CCM-22692: Send letters to DLQ if supplier allocation fails

infrastructure/terraform/components/api/module_lambda_supplier_allocator.tf

@@ -35,8 +35,9 @@ module "supplier_allocator" {
   log_subscription_role_arn = local.acct.log_subscription_role_arn
 
   lambda_env_vars = merge(local.common_lambda_env_vars, {
-    UPSERT_LETTERS_QUEUE_URL = module.sqs_letter_updates.sqs_queue_url,
-    IDEMPOTENCY_TABLE_NAME   = aws_dynamodb_table.idempotency.name
+    UPSERT_LETTERS_QUEUE_URL   = module.sqs_letter_updates.sqs_queue_url,
+    SUPPLIER_ALLOCATOR_DLQ_URL = module.sqs_supplier_allocator.sqs_dlq_url,
+    IDEMPOTENCY_TABLE_NAME     = aws_dynamodb_table.idempotency.name
   })
 }
 
@@ -83,6 +84,19 @@ data "aws_iam_policy_document" "supplier_allocator_lambda" {
     ]
   }
 
+  statement {
+    sid    = "AllowSupplierAllocatorDLQWrite"
+    effect = "Allow"
+
+    actions = [
+      "sqs:SendMessage"
+    ]
+
+    resources = [
+      module.sqs_supplier_allocator.sqs_dlq_arn
+    ]
+  }
+
   statement {
     sid    = "AllowConfigDynamoDBAccess"
     effect = "Allow"

lambdas/supplier-allocator/README.md

@@ -12,21 +12,23 @@ Consumes `LetterRequestPrepared` events (v1 and v2) from an SQS queue, chooses a
 2. Each record body is parsed and validated as either `$LetterRequestPreparedEventV2` or `$LetterRequestPreparedEvent` (v1 fallback).
 3. The allocator loads the relevant supplier configuration from `SUPPLIER_CONFIG_TABLE`, including the letter variant, active volume group, candidate suppliers, and compatible pack details.
 4. Candidate suppliers are filtered using pack support and daily capacity, then ranked using quota data from `SUPPLIER_QUOTAS_TABLE`.
-5. On success, the handler produces an allocation with `allocationStatus.status = "PENDING"`. If allocation cannot be completed, it produces a REJECTED allocation with a failure reason instead of dropping the message.
-6. Each record produces a `{ letterEvent, allocationDetails }` message sent to `UPSERT_LETTERS_QUEUE_URL`.
-7. After the batch completes, allocation counters are written back to `SUPPLIER_QUOTAS_TABLE`, and only genuine processing failures are returned as `batchItemFailures`.
+5. On success, the handler produces an allocation with `allocationStatus.status = "PENDING"` and sends `{ letterEvent, allocationDetails }` to `UPSERT_LETTERS_QUEUE_URL`.
+6. If a `SupplierConfigError` is raised, the original record is sent directly to `SUPPLIER_ALLOCATOR_DLQ_URL` and acknowledged so it is not retried.
+7. Any other processing error is returned in `batchItemFailures` so SQS retries the record based on the source queue redrive policy.
+8. After the batch completes, allocation counters are written back to `SUPPLIER_QUOTAS_TABLE`.
 
 ## Key Integration Points
 
-- **SQS**: Input from EventSub, output to the upsert-letter queue (`UPSERT_LETTERS_QUEUE`).
+- **SQS**: Input from EventSub, output to the upsert-letter queue (`UPSERT_LETTERS_QUEUE`), and direct publish to the allocator DLQ (`SUPPLIER_ALLOCATOR_DLQ_URL`) for `SupplierConfigError`.
 - **`SupplierConfigRepository`** from `@internal/datastore` (`SUPPLIER_CONFIG_TABLE`): reads letter variants, volume groups, supplier allocations, pack specifications, and supplier packs.
 - **`SupplierQuotasRepository`** from `@internal/datastore` (`SUPPLIER_QUOTAS_TABLE`): reads and writes daily and overall allocation counts per volume group and supplier.
 - **Event schemas**: `@nhsdigital/nhs-notify-event-schemas-letter-rendering` (v2) and `@nhsdigital/nhs-notify-event-schemas-letter-rendering-v1` (v1).
-- **Downstream consumer**: `upsert-letter` receives `{ letterEvent, allocationDetails }` and persists either a PENDING or REJECTED letter.
+- **Downstream consumer**: `upsert-letter` receives `{ letterEvent, allocationDetails }` and persists PENDING letters.
 
 ## Nuances and Peculiarities
 
-- **Failed allocations produce REJECTED letters, not dropped messages.** If the config lookup chain fails for any reason, the handler still sends a message to the upsert queue with `allocationStatus.status = "REJECTED"` and `supplierId = "unknown"`. No letters are silently lost.
+- **`SupplierConfigError` is treated as terminal for retries.** The handler sends the original message directly to the allocator DLQ and acknowledges the source record.
+- **All other failures retain normal retry semantics.** Non-`SupplierConfigError` records are returned in `batchItemFailures` and retried according to queue configuration.
 - **The factor algorithm is a running weighted average across the lifetime of the system, not per-batch.** The `overallAllocation` table accumulates counts since deployment. A supplier that handled a disproportionate share yesterday will have a high factor today and be deprioritised, allowing others to catch up to their target percentage.
 - **Daily capacity check uses London timezone.** `format(toZonedTime(new Date(), "Europe/London"), "yyyy-MM-dd")` determines the date key. Capacity resets at midnight London time, not UTC.
 - **Quota updates happen after the entire batch completes**, not per-record. This optimisation means concurrent Lambda invocations can transiently over-allocate a supplier before quotas are reconciled.

lambdas/supplier-allocator/src/errors/supplier-config-error.ts

@@ -0,0 +1,9 @@
+/**
+ * Error thrown when a supplier cannot be allocated due to incorrect supplier config
+ */
+export default class SupplierConfigError extends Error {
+  constructor(public readonly message: string) {
+    super(message);
+    this.name = "SupplierConfigError";
+  }
+}

lambdas/supplier-allocator/src/handler/__tests__/allocate-handler.test.ts

@@ -8,11 +8,12 @@
   LetterStatusChangeEvent,
 } from "@nhsdigital/nhs-notify-event-schemas-supplier-api/src/events/letter-events";
 import { makeIdempotent } from "@aws-lambda-powertools/idempotency";
 import createSupplierAllocatorHandler from "../allocate-handler";
 import * as supplierConfig from "../../services/supplier-config";
 import * as supplierQuotas from "../../services/supplier-quotas";
 import * as allocationConfig from "../allocation-config";
 import { Deps } from "../../config/deps";
+import SupplierConfigError from "../../errors/supplier-config-error";
 import packageJson from "../../../package.json";
 
 const renderingSchemaVersion: string =
@@ -527,56 +528,13 @@
     expect(sendCall.input.QueueUrl).toBe(queueUrl);
   });
 
-  test("logs error when supplier config retrieval fails", async () => {
-    const preparedEvent = createPreparedV2Event();
-
-    const evt: SQSEvent = createSQSEvent([
-      createSqsRecord("msg1", JSON.stringify(preparedEvent)),
-    ]);
-
-    process.env.UPSERT_LETTERS_QUEUE_URL = "https://sqs.test.queue";
-    const configError = new Error("Failed to retrieve supplier config");
-    (supplierConfig.getVariantDetails as jest.Mock).mockRejectedValueOnce(
-      configError,
-    );
-
-    const handler = createSupplierAllocatorHandler(mockedDeps);
-    const result = await handler(evt, {} as any, {} as any);
-    if (!result) throw new Error("expected BatchResponse, got void");
-    expect((mockedDeps.logger.error as jest.Mock).mock.calls).toHaveLength(1);
-    expect((mockedDeps.logger.error as jest.Mock).mock.calls[0][0]).toEqual(
-      expect.objectContaining({
-        description: "Error fetching supplier from config",
-        err: configError,
-        variantId: "lv1",
-      }),
-    );
-    expect(mockedDeps.sqsClient.send).toHaveBeenCalledTimes(1);
-    const sendCall = (mockedDeps.sqsClient.send as jest.Mock).mock.calls[0][0];
-    expect(sendCall).toBeInstanceOf(SendMessageCommand);
-
-    const messageBody = JSON.parse(sendCall.input.MessageBody);
-    expect(messageBody.letterEvent).toEqual(preparedEvent);
-    expect(messageBody.allocationDetails.supplierSpec).toEqual({
-      supplierId: "unknown",
-      specId: "unknown",
-      priority: 0,
-      billingId: "unknown",
-    });
-    expect(messageBody.allocationDetails.allocationStatus).toEqual({
-      status: "REJECTED",
-      reasonCode: "NO_SUPPLIERS_AVAILABLE",
-      reasonText: "Failed to retrieve supplier config",
-    });
-  });
-
   const rejectWith = (mock: jest.Mock, errorMessage: string) =>
     mock.mockRejectedValueOnce(new Error(errorMessage));
 
   const throwAny = (mock: jest.Mock) =>
     mock.mockRejectedValueOnce("anything that is not an Error");
 
-  const supplierConfigErrorCases = [
+  const nonSupplierConfigErrorCases = [
     {
       name: "getVolumeGroupDetails",
       errorMessage: "Volume group retrieval failed",
@@ -639,15 +597,17 @@
     },
   ];
 
-  test.each(supplierConfigErrorCases)(
-    "logs error when %s rejects during supplier config resolution",
+  test.each(nonSupplierConfigErrorCases)(
+    "returns batch failure when %s rejects with a non-SupplierConfigError",
     async ({ errorMessage, setup }) => {
       const preparedEvent = createPreparedV2Event();
       const evt: SQSEvent = createSQSEvent([
         createSqsRecord("msg1", JSON.stringify(preparedEvent)),
       ]);
 
       process.env.UPSERT_LETTERS_QUEUE_URL = "https://sqs.test.queue";
+      process.env.SUPPLIER_ALLOCATOR_DLQ_URL =
+        "https://sqs.test.queue/supplier-allocator-dlq";
       setup();
 
       const handler = createSupplierAllocatorHandler(mockedDeps);
@@ -657,71 +617,118 @@
       expect((mockedDeps.logger.error as jest.Mock).mock.calls).toHaveLength(1);
       expect((mockedDeps.logger.error as jest.Mock).mock.calls[0][0]).toEqual(
         expect.objectContaining({
-          description: "Error fetching supplier from config",
-          variantId: "lv1",
+          description: "Error processing allocation of record",
         }),
       );
+      expect(result.batchItemFailures).toHaveLength(1);
+      expect(result.batchItemFailures[0].itemIdentifier).toBe("msg1");
+      expect(mockedDeps.sqsClient.send).not.toHaveBeenCalled();
+
+      expect(errorMessage).toBeDefined();
+    },
+  );
+
+  describe("Dead letter queue", () => {
+    it("places the record on the DLQ when no suppliers are found for pack specification", async () => {
+      const preparedEvent = createPreparedV2Event();
+      const evt: SQSEvent = createSQSEvent([
+        createSqsRecord("msg1", JSON.stringify(preparedEvent)),
+      ]);
+
+      process.env.UPSERT_LETTERS_QUEUE_URL = "https://sqs.test.queue";
+      process.env.SUPPLIER_ALLOCATOR_DLQ_URL =
+        "https://sqs.test.queue/supplier-allocator-dlq";
+
+      setupDefaultMocks();
+      (allocationConfig.suppliersWithValidPack as jest.Mock).mockResolvedValue(
+        [],
+      );
+
+      const handler = createSupplierAllocatorHandler(mockedDeps);
+      const result = await handler(evt, {} as any, {} as any);
+      if (!result) throw new Error("expected BatchResponse, got void");
+
+      expect((mockedDeps.logger.error as jest.Mock).mock.calls).toHaveLength(1);
+      expect((mockedDeps.logger.error as jest.Mock).mock.calls[0][0]).toEqual(
+        expect.objectContaining({
+          description: "Error processing allocation of record",
+        }),
+      );
+
+      expect(result.batchItemFailures).toHaveLength(0);
       expect(mockedDeps.sqsClient.send).toHaveBeenCalledTimes(1);
+
       const sendCall = (mockedDeps.sqsClient.send as jest.Mock).mock
         .calls[0][0];
       expect(sendCall).toBeInstanceOf(SendMessageCommand);
+      expect(sendCall.input.QueueUrl).toBe(
+        "https://sqs.test.queue/supplier-allocator-dlq",
+      );
+      expect(sendCall.input.MessageBody).toBe(JSON.stringify(preparedEvent));
+    });
 
-      const messageBody = JSON.parse(sendCall.input.MessageBody);
-      expect(messageBody.letterEvent).toEqual(preparedEvent);
-      expect(messageBody.allocationDetails.supplierSpec).toEqual({
-        supplierId: "unknown",
-        specId: "unknown",
-        priority: 0,
-        billingId: "unknown",
-      });
-      expect(messageBody.allocationDetails.allocationStatus).toEqual({
-        status: "REJECTED",
-        reasonCode: "NO_SUPPLIERS_AVAILABLE",
-        reasonText: errorMessage,
-      });
-    },
-  );
+    it("places the record on the DLQ when supplier config retrieval fails", async () => {
+      const preparedEvent = createPreparedV2Event();
 
-  test("returns batch failure when no suppliers are found for pack specification", async () => {
-    const preparedEvent = createPreparedV2Event();
-    const evt: SQSEvent = createSQSEvent([
-      createSqsRecord("msg1", JSON.stringify(preparedEvent)),
-    ]);
+      const evt: SQSEvent = createSQSEvent([
+        createSqsRecord("msg1", JSON.stringify(preparedEvent)),
+      ]);
 
-    process.env.UPSERT_LETTERS_QUEUE_URL = "https://sqs.test.queue";
+      process.env.UPSERT_LETTERS_QUEUE_URL = "https://sqs.test.queue";
+      process.env.SUPPLIER_ALLOCATOR_DLQ_URL =
+        "https://sqs.test.queue/supplier-allocator-dlq";
+      const configError = new SupplierConfigError(
+        "Failed to retrieve supplier config",
+      );
+      (supplierConfig.getVariantDetails as jest.Mock).mockRejectedValueOnce(
+        configError,
+      );
 
-    setupDefaultMocks();
-    (allocationConfig.suppliersWithValidPack as jest.Mock).mockResolvedValue(
-      [],
-    );
+      const handler = createSupplierAllocatorHandler(mockedDeps);
+      const result = await handler(evt, {} as any, {} as any);
+      if (!result) throw new Error("expected BatchResponse, got void");
+      expect((mockedDeps.logger.error as jest.Mock).mock.calls).toHaveLength(1);
+      expect((mockedDeps.logger.error as jest.Mock).mock.calls[0][0]).toEqual(
+        expect.objectContaining({
+          description: "Error processing allocation of record",
+          err: configError,
+        }),
+      );
+      expect(result.batchItemFailures).toHaveLength(0);
+      expect(mockedDeps.sqsClient.send).toHaveBeenCalledTimes(1);
+      const sendCall = (mockedDeps.sqsClient.send as jest.Mock).mock
+        .calls[0][0];
+      expect(sendCall.input.QueueUrl).toBe(
+        "https://sqs.test.queue/supplier-allocator-dlq",
+      );
+      expect(sendCall.input.MessageBody).toBe(JSON.stringify(preparedEvent));
+    });
 
-    const handler = createSupplierAllocatorHandler(mockedDeps);
-    const result = await handler(evt, {} as any, {} as any);
-    if (!result) throw new Error("expected BatchResponse, got void");
+    it("returns batch failure when sending to DLQ fails", async () => {
+      const preparedEvent = createPreparedV2Event();
+      const evt: SQSEvent = createSQSEvent([
+        createSqsRecord("msg1", JSON.stringify(preparedEvent)),
+      ]);
 
-    expect((mockedDeps.logger.error as jest.Mock).mock.calls).toHaveLength(1);
-    expect((mockedDeps.logger.error as jest.Mock).mock.calls[0][0]).toEqual(
-      expect.objectContaining({
-        description: "Error fetching supplier from config",
-        variantId: "lv1",
-      }),
-    );
-    expect(mockedDeps.sqsClient.send).toHaveBeenCalledTimes(1);
-    const sendCall = (mockedDeps.sqsClient.send as jest.Mock).mock.calls[0][0];
-    expect(sendCall).toBeInstanceOf(SendMessageCommand);
+      process.env.UPSERT_LETTERS_QUEUE_URL = "https://sqs.test.queue";
+      process.env.SUPPLIER_ALLOCATOR_DLQ_URL =
+        "https://sqs.test.queue/supplier-allocator-dlq";
 
-    const messageBody = JSON.parse(sendCall.input.MessageBody);
-    expect(messageBody.letterEvent).toEqual(preparedEvent);
-    expect(messageBody.allocationDetails.supplierSpec).toEqual({
-      supplierId: "unknown",
-      specId: "unknown",
-      priority: 0,
-      billingId: "unknown",
-    });
-    expect(messageBody.allocationDetails.allocationStatus).toEqual({
-      status: "REJECTED",
-      reasonCode: "NO_SUPPLIERS_AVAILABLE",
-      reasonText: "No suppliers found for pack specification spec1",
+      setupDefaultMocks();
+      (allocationConfig.suppliersWithValidPack as jest.Mock).mockResolvedValue(
+        [],
+      );
+      (mockedDeps.sqsClient.send as jest.Mock).mockRejectedValueOnce(
+        new Error("DLQ send failed"),
+      );
+
+      const handler = createSupplierAllocatorHandler(mockedDeps);
+      const result = await handler(evt, {} as any, {} as any);
+      if (!result) throw new Error("expected BatchResponse, got void");
+
+      expect(result.batchItemFailures).toHaveLength(1);
+      expect(result.batchItemFailures[0].itemIdentifier).toBe("msg1");
+      expect((mockedDeps.logger.error as jest.Mock).mock.calls).toHaveLength(2);
     });
   });
 

lambdas/supplier-allocator/src/handler/__tests__/allocation-config.test.ts

@@ -21,6 +21,15 @@ import * as supplierQuotasService from "../../services/supplier-quotas";
 jest.mock("../../services/supplier-config");
 jest.mock("../../services/supplier-quotas");
 
+async function expectSupplierConfigError(
+  promise: Promise<unknown>,
+  message: string | RegExp,
+): Promise<void> {
+  await expect(promise).rejects.toThrow(message);
+  await expect(promise).rejects.toMatchObject({
+    name: "SupplierConfigError",
+  });
+}
 describe("eligibleSuppliers", () => {
   let mockDeps: jest.Mocked<Deps>;
   let mockVolumeGroup: VolumeGroup;
@@ -970,14 +979,14 @@ describe("selectSupplierByFactor", () => {
       } as SupplierAllocation,
     ];
 
-    await expect(
+    await expectSupplierConfigError(
       selectSupplierByFactor(
         mockSuppliers,
         zeroAllocations,
         domainId,
         mockDeps,
       ),
-    ).rejects.toThrow(
+
       "No valid supplier allocations found for suppliers with valid pack",
     );
   });
@@ -1155,13 +1164,14 @@ describe("selectSupplierByFactor", () => {
       supplierQuotasService.calculateSupplierAllocatedFactor as jest.Mock
     ).mockResolvedValue([]);
 
-    await expect(
+    await expectSupplierConfigError(
       selectSupplierByFactor(
         mockSuppliers,
         mockSupplierAllocations,
         domainId,
         mockDeps,
       ),
-    ).rejects.toThrow("No supplier factors could be calculated for allocation");
+      "No supplier factors could be calculated for allocation",
+    );
   });
 });