fix: stop a malformed SQS body from poisoning the whole batch

lambdas/functions/control-plane/src/lambda.test.ts

@@ -116,6 +116,49 @@ describe('Test scale up lambda wrapper.', () => {
       vi.clearAllMocks();
     });
 
+    const malformedRecord = (messageId: string): SQSRecord =>
+      ({ ...sqsRecord, eventSource: 'aws:sqs', messageId, body: 'not-json{' }) as SQSRecord;
+
+    it('Should not let one malformed message fail the whole invocation', async () => {
+      // Previously JSON.parse ran outside the try/catch, so an unparseable body threw
+      // straight out of the handler, failing the invocation and making SQS redeliver
+      // every message in the batch.
+      const records = [...createMultipleRecords(2), malformedRecord('message-bad')];
+      vi.mocked(scaleUp).mockResolvedValue([]);
+
+      await expect(scaleUpHandler({ Records: records }, context)).resolves.not.toThrow();
+    });
+
+    it('Should report only the malformed message as a batch item failure', async () => {
+      const records = [...createMultipleRecords(2), malformedRecord('message-bad')];
+      vi.mocked(scaleUp).mockResolvedValue([]);
+
+      await expect(scaleUpHandler({ Records: records }, context)).resolves.toEqual({
+        batchItemFailures: [{ itemIdentifier: 'message-bad' }],
+      });
+    });
+
+    it('Should still process the valid messages alongside a malformed one', async () => {
+      const records = [...createMultipleRecords(2), malformedRecord('message-bad')];
+      vi.mocked(scaleUp).mockResolvedValue([]);
+
+      await scaleUpHandler({ Records: records }, context);
+
+      expect(scaleUp).toHaveBeenCalledWith([
+        expect.objectContaining({ messageId: 'message-0' }),
+        expect.objectContaining({ messageId: 'message-1' }),
+      ]);
+    });
+
+    it('Should combine malformed and rejected messages in batch item failures', async () => {
+      const records = [...createMultipleRecords(2), malformedRecord('message-bad')];
+      vi.mocked(scaleUp).mockResolvedValue(['message-1']);
+
+      await expect(scaleUpHandler({ Records: records }, context)).resolves.toEqual({
+        batchItemFailures: [{ itemIdentifier: 'message-bad' }, { itemIdentifier: 'message-1' }],
+      });
+    });
+
     const createMultipleRecords = (count: number, eventSource = 'aws:sqs'): SQSRecord[] => {
       return Array.from({ length: count }, (_, i) => ({
         ...sqsRecord,

lambdas/functions/control-plane/src/lambda.ts

@@ -16,6 +16,7 @@ export async function scaleUpHandler(event: SQSEvent, context: Context): Promise
 
   const sqsMessages: ActionRequestMessageSQS[] = [];
   const warnedEventSources = new Set<string>();
+  const malformedMessageIds: string[] = [];
 
   for (const { body, eventSource, messageId } of event.Records) {
     if (eventSource !== 'aws:sqs') {
@@ -27,7 +28,25 @@ export async function scaleUpHandler(event: SQSEvent, context: Context): Promise
       continue;
     }
 
-    const payload = JSON.parse(body) as ActionRequestMessage;
+    let payload: ActionRequestMessage;
+    try {
+      payload = JSON.parse(body) as ActionRequestMessage;
+    } catch (e) {
+      // Parsing happens outside the try/catch below, so an unparseable body used to
+      // throw straight out of the handler. That failed the whole invocation and made
+      // SQS redeliver the entire batch, so one malformed message held up every valid
+      // message alongside it, indefinitely.
+      //
+      // Report it as an individual failure instead: the rest of the batch proceeds,
+      // and the malformed message exhausts maxReceiveCount on its own. It cannot be
+      // acknowledged and discarded here — reporting it is the only way to single it
+      // out — so configure redrive_build_queue if these should be captured rather
+      // than expire.
+      logger.error(`Ignoring message ${messageId}, body is not valid JSON`, { error: e, messageId });
+      malformedMessageIds.push(messageId);
+
+      continue;
+    }
     sqsMessages.push({ ...payload, messageId });
   }
 
@@ -38,7 +57,7 @@ export async function scaleUpHandler(event: SQSEvent, context: Context): Promise
     return (l.retryCounter ?? 0) - (r.retryCounter ?? 0);
   });
 
-  const batchItemFailures: SQSBatchItemFailure[] = [];
+  const batchItemFailures: SQSBatchItemFailure[] = malformedMessageIds.map((itemIdentifier) => ({ itemIdentifier }));
 
   try {
     const rejectedMessageIds = await scaleUp(sqsMessages);