diff --git a/lambdas/functions/control-plane/src/lambda.test.ts b/lambdas/functions/control-plane/src/lambda.test.ts index 2c9a98e420..12653ceaa9 100644 --- a/lambdas/functions/control-plane/src/lambda.test.ts +++ b/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, diff --git a/lambdas/functions/control-plane/src/lambda.ts b/lambdas/functions/control-plane/src/lambda.ts index e2a0451c95..404cafc44a 100644 --- a/lambdas/functions/control-plane/src/lambda.ts +++ b/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(); + 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);