fix(control-plane): discard malformed SQS messages (github-aws-runners#5219)

## Description

Treat malformed SQS message bodies as permanent, non-retryable failures.
The scale-up handler logs and acknowledges malformed records so they do
not block valid messages in the same batch or repeatedly return to the
queue.

Document the SQS event source mapping contract near `batchItemFailures`
and update the focused handler tests for the new behavior.

## Test Plan

1. Send a scale-up SQS batch containing one valid message and one
malformed JSON body.
2. Confirm the valid message is passed to `scaleUp`.
3. Confirm the malformed message is logged but omitted from
`batchItemFailures`.
4. Confirm a message rejected by `scaleUp` is still returned in
`batchItemFailures`.
5. Run `yarn vitest run functions/control-plane/src/lambda.test.ts
--config functions/control-plane/vitest.config.ts --reporter=dot`.

## Related Issues

None.

Author: edersonbrilhante

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

@@ -129,12 +129,12 @@ describe('Test scale up lambda wrapper.', () => {
       await expect(scaleUpHandler({ Records: records }, context)).resolves.not.toThrow();
     });
 
-    it('Should report only the malformed message as a batch item failure', async () => {
+    it('Should acknowledge a malformed message without retrying it', async () => {
       const records = [...createMultipleRecords(2), malformedRecord('message-bad')];
       vi.mocked(scaleUp).mockResolvedValue([]);
 
       await expect(scaleUpHandler({ Records: records }, context)).resolves.toEqual({
-        batchItemFailures: [{ itemIdentifier: 'message-bad' }],
+        batchItemFailures: [],
       });
     });
 
@@ -150,12 +150,12 @@ describe('Test scale up lambda wrapper.', () => {
       ]);
     });
 
-    it('Should combine malformed and rejected messages in batch item failures', async () => {
+    it('Should report rejected messages but acknowledge malformed messages', 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' }],
+        batchItemFailures: [{ itemIdentifier: 'message-1' }],
       });
     });
 

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

@@ -16,7 +16,6 @@ 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') {
@@ -32,18 +31,10 @@ export async function scaleUpHandler(event: SQSEvent, context: Context): Promise
     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.
+      // A malformed body is a permanent, non-retryable failure. Keep it out of
+      // batchItemFailures so the event source mapping acknowledges and deletes it,
+      // while valid records in the same batch continue to scale up normally.
       logger.error(`Ignoring message ${messageId}, body is not valid JSON`, { error: e, messageId });
-      malformedMessageIds.push(messageId);
 
       continue;
     }
@@ -57,7 +48,11 @@ export async function scaleUpHandler(event: SQSEvent, context: Context): Promise
     return (l.retryCounter ?? 0) - (r.retryCounter ?? 0);
   });
 
-  const batchItemFailures: SQSBatchItemFailure[] = malformedMessageIds.map((itemIdentifier) => ({ itemIdentifier }));
+  // The SQS event source mapping owns message acknowledgement and deletion. Because
+  // ReportBatchItemFailures is enabled, a successful handler response makes Lambda
+  // delete every record not listed here; listed records remain in SQS and become
+  // available for retry after their visibility timeout expires.
+  const batchItemFailures: SQSBatchItemFailure[] = [];
 
   try {
     const rejectedMessageIds = await scaleUp(sqsMessages);