From 2b380c30041cc53a4fc9ce12ffbebd20b39ef35e Mon Sep 17 00:00:00 2001 From: SHAIK VAHID <38548782+vahidshaik1901@users.noreply.github.com> Date: Sun, 6 Sep 2026 11:55:03 +0530 Subject: [PATCH 1/2] fix(batch): throw when a synchronous processor gets a promise-returning handler `BatchProcessorSync.processRecordSync()` passed the handler's return value straight to `successHandler()`. When the handler is `async`, that value is a pending promise, so every record was recorded as a success, `batchItemFailures` came back empty, and the event source deleted messages that were never really processed. Any rejection surfaced later as an unhandled rejection, after the response had already been returned. The record handler is typed as `CallableFunction`, so TypeScript does not reject an `async` function here, and there was no runtime guard. `SqsFifoPartialProcessor` delegates to `processRecordSync()`, so the non-deprecated FIFO class was affected as well. Add a thenable guard after invoking the handler and throw a `BatchProcessingError` pointing at `BatchProcessor` / `SqsFifoPartialProcessorAsync`. The throw happens outside the try/catch so it fails the whole invocation rather than being recorded as a per-record failure, which is the right outcome for a programming error and mirrors how `BatchProcessor.processRecordSync()` already rejects misuse. `BatchProcessorSync` had no unit tests, which is why this went unnoticed; this adds coverage for the sync processor and the FIFO subclass. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_018W34ieE3NNGaRQBxj5CKkH --- packages/batch/src/BatchProcessorSync.ts | 21 ++++++- .../tests/unit/BatchProcessorSync.test.ts | 63 +++++++++++++++++++ 2 files changed, 81 insertions(+), 3 deletions(-) create mode 100644 packages/batch/tests/unit/BatchProcessorSync.test.ts diff --git a/packages/batch/src/BatchProcessorSync.ts b/packages/batch/src/BatchProcessorSync.ts index cd5efee509..90469ff5e8 100644 --- a/packages/batch/src/BatchProcessorSync.ts +++ b/packages/batch/src/BatchProcessorSync.ts @@ -111,15 +111,30 @@ import type { BaseRecord, FailureResponse, SuccessResponse } from './types.js'; public processRecordSync( record: BaseRecord ): SuccessResponse | FailureResponse { + let result: unknown; try { const data = this.toBatchType(record, this.eventType); - const result = this.handler(data, this.options?.context); - - return this.successHandler(record, result); + result = this.handler(data, this.options?.context); } catch (error) { return this.failureHandler(record, toError(error)); } + + if (isThenable(result)) { + throw new BatchProcessingError( + 'The record handler returned a promise, but this batch processor is synchronous and cannot await it. Use `BatchProcessor` together with `processPartialResponse()`, or `SqsFifoPartialProcessorAsync` for FIFO queues.' + ); + } + + return this.successHandler(record, result); } } +/** + * Type guard to detect a thenable (promise-like) value returned by a record handler. + * + * @param value - The value returned by the record handler + */ +const isThenable = (value: unknown): value is PromiseLike => + typeof (value as PromiseLike | undefined)?.then === 'function'; + export { BatchProcessorSync }; diff --git a/packages/batch/tests/unit/BatchProcessorSync.test.ts b/packages/batch/tests/unit/BatchProcessorSync.test.ts new file mode 100644 index 0000000000..78d10d8875 --- /dev/null +++ b/packages/batch/tests/unit/BatchProcessorSync.test.ts @@ -0,0 +1,63 @@ +import context from '@aws-lambda-powertools/testing-utils/context'; +import type { SQSRecord } from 'aws-lambda'; +import { describe, expect, it } from 'vitest'; +import { + BatchProcessingError, + BatchProcessorSync, + EventType, + processPartialResponseSync, + SqsFifoPartialProcessor, +} from '../../src/index.js'; +import { sqsRecordFactory } from '../helpers/factories.js'; + +describe('Class: BatchProcessorSync', () => { + const asyncRecordHandler = async (record: SQSRecord): Promise => + record.body; + + it('throws when the record handler returns a Promise', () => { + // Prepare + const records = [sqsRecordFactory('success'), sqsRecordFactory('success')]; + const batch = { Records: records }; + const processor = new BatchProcessorSync(EventType.SQS); + + // Act & Assess + expect(() => + processPartialResponseSync(batch, asyncRecordHandler, processor, { + context, + }) + ).toThrow(BatchProcessingError); + }); + + it('does not report records as processed when the handler returns a Promise', () => { + // Prepare + const records = [sqsRecordFactory('success'), sqsRecordFactory('success')]; + const batch = { Records: records }; + const processor = new BatchProcessorSync(EventType.SQS); + + // Act + try { + processPartialResponseSync(batch, asyncRecordHandler, processor, { + context, + }); + } catch { + // the throw itself is asserted in the test above + } + + // Assess + expect(processor.successMessages).toHaveLength(0); + }); + + it('throws when a SqsFifoPartialProcessor record handler returns a Promise', () => { + // Prepare + const records = [sqsRecordFactory('success'), sqsRecordFactory('success')]; + const batch = { Records: records }; + const processor = new SqsFifoPartialProcessor(); + + // Act & Assess + expect(() => + processPartialResponseSync(batch, asyncRecordHandler, processor, { + context, + }) + ).toThrow(BatchProcessingError); + }); +}); From 8c978786e949e6ee1086d10f29966ab919c621eb Mon Sep 17 00:00:00 2001 From: SHAIK VAHID <38548782+vahidshaik1901@users.noreply.github.com> Date: Sun, 6 Sep 2026 15:00:24 +0530 Subject: [PATCH 2/2] fix(batch): address review on the synchronous handler guard MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Introduce AsyncHandlerNotSupportedError so users have something to match on, and take ownership of the abandoned promise's rejection before throwing — otherwise a rejecting handler also produces an unhandled rejection, which the Lambda runtime reports in place of the useful error. Cover the rejecting-handler case in the tests, fold the successMessages check into the first test, fix the four JSDoc examples that paired an async record handler with a synchronous processor, and document the new error in the SQS and FIFO sections of the batch docs. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01YSkXPgm2uonSJnja7ZHdwM --- docs/features/batch.md | 5 ++- packages/batch/src/BatchProcessorSync.ts | 15 +++++--- packages/batch/src/SqsFifoPartialProcessor.ts | 7 ++-- packages/batch/src/errors.ts | 14 ++++++++ packages/batch/src/index.ts | 1 + .../batch/src/processPartialResponseSync.ts | 10 +++--- .../tests/unit/BatchProcessorSync.test.ts | 36 ++++++++++--------- 7 files changed, 58 insertions(+), 30 deletions(-) diff --git a/docs/features/batch.md b/docs/features/batch.md index c19df73b3a..30e96bf47d 100644 --- a/docs/features/batch.md +++ b/docs/features/batch.md @@ -87,7 +87,7 @@ Processing batches from SQS works in three stages: By default, the batch processor will process messages in parallel, which does not guarantee the order of processing. If you need to process messages in order, set the [`processInParallel` option to `false`](#sequential-processing), or use [`SqsFifoPartialProcessor` for SQS FIFO queues](#fifo-queues). !!! note - If you're migrating from `BatchProcessorSync` to `BatchProcessor`, note that `processPartialResponse` is async and returns a promise. + If you're migrating from `BatchProcessorSync` to `BatchProcessor`, note that `processPartialResponse` is async and returns a promise. The synchronous processors have no way to await a record handler, so they throw `AsyncHandlerNotSupportedError` when the handler returns one. === "index.ts" @@ -125,6 +125,9 @@ By default, we will stop processing at the first failure and mark unprocessed me Enable the `skipGroupOnError` option for seamless processing of messages from various group IDs. This setup ensures that messages from a failed group ID are sent back to SQS, enabling uninterrupted processing of messages from the subsequent group ID. +!!! note + `SqsFifoPartialProcessor` is synchronous and throws `AsyncHandlerNotSupportedError` when the record handler returns a promise. Use `SqsFifoPartialProcessorAsync` together with `processPartialResponse` for asynchronous record handlers. + === "index.ts" ```typescript hl_lines="1-4 8 20" diff --git a/packages/batch/src/BatchProcessorSync.ts b/packages/batch/src/BatchProcessorSync.ts index 90469ff5e8..55593cbf50 100644 --- a/packages/batch/src/BatchProcessorSync.ts +++ b/packages/batch/src/BatchProcessorSync.ts @@ -1,5 +1,9 @@ import { BasePartialBatchProcessor } from './BasePartialBatchProcessor.js'; -import { BatchProcessingError, toError } from './errors.js'; +import { + AsyncHandlerNotSupportedError, + BatchProcessingError, + toError, +} from './errors.js'; import type { BaseRecord, FailureResponse, SuccessResponse } from './types.js'; /** @@ -120,9 +124,12 @@ import type { BaseRecord, FailureResponse, SuccessResponse } from './types.js'; } if (isThenable(result)) { - throw new BatchProcessingError( - 'The record handler returned a promise, but this batch processor is synchronous and cannot await it. Use `BatchProcessor` together with `processPartialResponse()`, or `SqsFifoPartialProcessorAsync` for FIFO queues.' - ); + // The promise is abandoned here, so take ownership of its rejection first: + // otherwise a rejecting handler also surfaces as an unhandled rejection, which + // the Lambda runtime reports instead of the error thrown below. + result.then(undefined, () => undefined); + + throw new AsyncHandlerNotSupportedError(); } return this.successHandler(record, result); diff --git a/packages/batch/src/SqsFifoPartialProcessor.ts b/packages/batch/src/SqsFifoPartialProcessor.ts index d7752b83f8..6c0fffdeb5 100644 --- a/packages/batch/src/SqsFifoPartialProcessor.ts +++ b/packages/batch/src/SqsFifoPartialProcessor.ts @@ -27,15 +27,14 @@ import type { * @example * ```typescript * import { - * BatchProcessor, - * EventType, * processPartialResponseSync, + * SqsFifoPartialProcessor, * } from '@aws-lambda-powertools/batch'; * import type { SQSRecord, SQSHandler } from 'aws-lambda'; * - * const processor = new BatchProcessor(EventType.SQS); + * const processor = new SqsFifoPartialProcessor(); * - * const recordHandler = async (record: SQSRecord): Promise => { + * const recordHandler = (record: SQSRecord): void => { * const payload = JSON.parse(record.body); * }; * diff --git a/packages/batch/src/errors.ts b/packages/batch/src/errors.ts index 7179ba7ba0..2c50efdb06 100644 --- a/packages/batch/src/errors.ts +++ b/packages/batch/src/errors.ts @@ -10,6 +10,19 @@ class BatchProcessingError extends Error { } } +/** + * Error thrown by the Batch Processing utility when a record handler returns a promise + * to a synchronous batch processor, which has no way to await it. + */ +class AsyncHandlerNotSupportedError extends BatchProcessingError { + public constructor() { + super( + 'The record handler returned a promise, but this batch processor is synchronous and cannot await it. Use BatchProcessor together with processPartialResponse(), or SqsFifoPartialProcessorAsync for FIFO queues.' + ); + this.name = 'AsyncHandlerNotSupportedError'; + } +} + /** * Error thrown by the Batch Processing utility when all batch records failed to be processed */ @@ -111,6 +124,7 @@ const toError = (value: unknown): Error => { }; export { + AsyncHandlerNotSupportedError, BatchProcessingError, FullBatchFailureError, ParsingError, diff --git a/packages/batch/src/index.ts b/packages/batch/src/index.ts index fb0a86a751..3e418f8063 100644 --- a/packages/batch/src/index.ts +++ b/packages/batch/src/index.ts @@ -3,6 +3,7 @@ export { BatchProcessor } from './BatchProcessor.js'; export { BatchProcessorSync } from './BatchProcessorSync.js'; export { EventType } from './constants.js'; export { + AsyncHandlerNotSupportedError, BatchProcessingError, FullBatchFailureError, ParsingError, diff --git a/packages/batch/src/processPartialResponseSync.ts b/packages/batch/src/processPartialResponseSync.ts index d694f440eb..1e8e3392db 100644 --- a/packages/batch/src/processPartialResponseSync.ts +++ b/packages/batch/src/processPartialResponseSync.ts @@ -26,15 +26,15 @@ import type { * @example * ```typescript * import { - * BatchProcessor, + * BatchProcessorSync, * EventType, * processPartialResponseSync, * } from '@aws-lambda-powertools/batch'; * import type { SQSRecord, SQSHandler } from 'aws-lambda'; * - * const processor = new BatchProcessor(EventType.SQS); + * const processor = new BatchProcessorSync(EventType.SQS); * - * const recordHandler = async (record: SQSRecord): Promise => { + * const recordHandler = (record: SQSRecord): void => { * const payload = JSON.parse(record.body); * }; * @@ -59,7 +59,7 @@ import type { * * const processor = new SqsFifoPartialProcessor(); * - * const recordHandler = async (record: SQSRecord): Promise => { + * const recordHandler = (record: SQSRecord): void => { * const payload = JSON.parse(record.body); * }; * @@ -83,7 +83,7 @@ import type { * * const processor = new SqsFifoPartialProcessor(); * - * const recordHandler = async (record: SQSRecord): Promise => { + * const recordHandler = (record: SQSRecord): void => { * const payload = JSON.parse(record.body); * }; * diff --git a/packages/batch/tests/unit/BatchProcessorSync.test.ts b/packages/batch/tests/unit/BatchProcessorSync.test.ts index 78d10d8875..2fdda0f94f 100644 --- a/packages/batch/tests/unit/BatchProcessorSync.test.ts +++ b/packages/batch/tests/unit/BatchProcessorSync.test.ts @@ -2,7 +2,7 @@ import context from '@aws-lambda-powertools/testing-utils/context'; import type { SQSRecord } from 'aws-lambda'; import { describe, expect, it } from 'vitest'; import { - BatchProcessingError, + AsyncHandlerNotSupportedError, BatchProcessorSync, EventType, processPartialResponseSync, @@ -14,7 +14,13 @@ describe('Class: BatchProcessorSync', () => { const asyncRecordHandler = async (record: SQSRecord): Promise => record.body; - it('throws when the record handler returns a Promise', () => { + const rejectingRecordHandler = async ( + _record: SQSRecord + ): Promise => { + throw new Error('failed'); + }; + + it('throws when the record handler returns a promise', () => { // Prepare const records = [sqsRecordFactory('success'), sqsRecordFactory('success')]; const batch = { Records: records }; @@ -25,29 +31,27 @@ describe('Class: BatchProcessorSync', () => { processPartialResponseSync(batch, asyncRecordHandler, processor, { context, }) - ).toThrow(BatchProcessingError); + ).toThrow(AsyncHandlerNotSupportedError); + expect(processor.successMessages).toHaveLength(0); }); - it('does not report records as processed when the handler returns a Promise', () => { + it('leaves no unhandled rejection behind when the promise rejects', () => { // Prepare const records = [sqsRecordFactory('success'), sqsRecordFactory('success')]; const batch = { Records: records }; const processor = new BatchProcessorSync(EventType.SQS); - // Act - try { - processPartialResponseSync(batch, asyncRecordHandler, processor, { + // Act & Assess + // The abandoned promise rejects after the throw, and vitest fails this file + // if nothing has taken ownership of that rejection. + expect(() => + processPartialResponseSync(batch, rejectingRecordHandler, processor, { context, - }); - } catch { - // the throw itself is asserted in the test above - } - - // Assess - expect(processor.successMessages).toHaveLength(0); + }) + ).toThrow(AsyncHandlerNotSupportedError); }); - it('throws when a SqsFifoPartialProcessor record handler returns a Promise', () => { + it('throws when a SqsFifoPartialProcessor record handler returns a promise', () => { // Prepare const records = [sqsRecordFactory('success'), sqsRecordFactory('success')]; const batch = { Records: records }; @@ -58,6 +62,6 @@ describe('Class: BatchProcessorSync', () => { processPartialResponseSync(batch, asyncRecordHandler, processor, { context, }) - ).toThrow(BatchProcessingError); + ).toThrow(AsyncHandlerNotSupportedError); }); });