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/3] 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/3] 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); }); }); From e4571aca95bb460a59752e977c23ca601febe8d2 Mon Sep 17 00:00:00 2001 From: svozza Date: Tue, 22 Sep 2026 12:49:18 +0000 Subject: [PATCH 3/3] fix(batch): log the abandoned promise rejection and refine the sync handler guard Log the rejection reason of the abandoned promise via the processor logger instead of silently discarding it, so a failing async handler is not lost. Reword AsyncHandlerNotSupportedError to note the handler has already run, correct the processRecordSync JSDoc, make isThenable cast-free, and reuse the shared asyncSqsRecordHandler fixture in the tests. --- packages/batch/src/BatchProcessorSync.ts | 22 +++++++++---- packages/batch/src/errors.ts | 2 +- .../tests/unit/BatchProcessorSync.test.ts | 33 +++++++++---------- 3 files changed, 33 insertions(+), 24 deletions(-) diff --git a/packages/batch/src/BatchProcessorSync.ts b/packages/batch/src/BatchProcessorSync.ts index 55593cbf50..92fc2a48b3 100644 --- a/packages/batch/src/BatchProcessorSync.ts +++ b/packages/batch/src/BatchProcessorSync.ts @@ -108,7 +108,9 @@ import type { BaseRecord, FailureResponse, SuccessResponse } from './types.js'; * Then, it calls the handler function with the record data and context. * * If the handler function completes successfully, the method returns a success response. - * Otherwise, it returns a failure response with the error that occurred during processing. + * If it throws, the method returns a failure response with the error that occurred during processing. + * If it returns a promise, the method throws an {@link AsyncHandlerNotSupportedError}, since a + * synchronous processor cannot await it. * * @param record The record to be processed */ @@ -124,10 +126,15 @@ import type { BaseRecord, FailureResponse, SuccessResponse } from './types.js'; } if (isThenable(result)) { - // 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); + // The promise cannot be awaited here. Suppress its rejection so a failing handler + // doesn't also surface as Runtime.UnhandledPromiseRejection and mask the error + // thrown below, and log the reason so the handler's own failure is not lost. + result.then(undefined, (reason) => { + this.logger.error( + 'The record handler returned a promise to a synchronous batch processor and the promise later rejected', + toError(reason) + ); + }); throw new AsyncHandlerNotSupportedError(); } @@ -142,6 +149,9 @@ import type { BaseRecord, FailureResponse, SuccessResponse } from './types.js'; * @param value - The value returned by the record handler */ const isThenable = (value: unknown): value is PromiseLike => - typeof (value as PromiseLike | undefined)?.then === 'function'; + typeof value === 'object' && + value !== null && + 'then' in value && + typeof value.then === 'function'; export { BatchProcessorSync }; diff --git a/packages/batch/src/errors.ts b/packages/batch/src/errors.ts index 2c50efdb06..dbcc9c8f1e 100644 --- a/packages/batch/src/errors.ts +++ b/packages/batch/src/errors.ts @@ -17,7 +17,7 @@ class BatchProcessingError extends Error { 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.' + 'The record handler returned a promise, but this batch processor is synchronous and cannot await it. The handler has already been invoked, so any side effects of that call may still be in flight. Use BatchProcessor together with processPartialResponse(), or SqsFifoPartialProcessorAsync for FIFO queues.' ); this.name = 'AsyncHandlerNotSupportedError'; } diff --git a/packages/batch/tests/unit/BatchProcessorSync.test.ts b/packages/batch/tests/unit/BatchProcessorSync.test.ts index 2fdda0f94f..e0d61c2615 100644 --- a/packages/batch/tests/unit/BatchProcessorSync.test.ts +++ b/packages/batch/tests/unit/BatchProcessorSync.test.ts @@ -1,6 +1,5 @@ import context from '@aws-lambda-powertools/testing-utils/context'; -import type { SQSRecord } from 'aws-lambda'; -import { describe, expect, it } from 'vitest'; +import { describe, expect, it, vi } from 'vitest'; import { AsyncHandlerNotSupportedError, BatchProcessorSync, @@ -9,17 +8,9 @@ import { SqsFifoPartialProcessor, } from '../../src/index.js'; import { sqsRecordFactory } from '../helpers/factories.js'; +import { asyncSqsRecordHandler } from '../helpers/handlers.js'; describe('Class: BatchProcessorSync', () => { - const asyncRecordHandler = async (record: SQSRecord): Promise => - record.body; - - 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')]; @@ -28,27 +19,35 @@ describe('Class: BatchProcessorSync', () => { // Act & Assess expect(() => - processPartialResponseSync(batch, asyncRecordHandler, processor, { + processPartialResponseSync(batch, asyncSqsRecordHandler, processor, { context, }) ).toThrow(AsyncHandlerNotSupportedError); expect(processor.successMessages).toHaveLength(0); }); - it('leaves no unhandled rejection behind when the promise rejects', () => { + it('logs the rejection reason instead of leaving an unhandled rejection when the promise rejects', async () => { // Prepare - const records = [sqsRecordFactory('success'), sqsRecordFactory('success')]; + const records = [sqsRecordFactory('fail'), sqsRecordFactory('success')]; const batch = { Records: records }; const processor = new BatchProcessorSync(EventType.SQS); - // Act & Assess + // Act // 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, { + processPartialResponseSync(batch, asyncSqsRecordHandler, processor, { context, }) ).toThrow(AsyncHandlerNotSupportedError); + + // Assess + await vi.waitFor(() => + expect(console.error).toHaveBeenCalledWith( + 'The record handler returned a promise to a synchronous batch processor and the promise later rejected', + new Error('Failed to process record.') + ) + ); }); it('throws when a SqsFifoPartialProcessor record handler returns a promise', () => { @@ -59,7 +58,7 @@ describe('Class: BatchProcessorSync', () => { // Act & Assess expect(() => - processPartialResponseSync(batch, asyncRecordHandler, processor, { + processPartialResponseSync(batch, asyncSqsRecordHandler, processor, { context, }) ).toThrow(AsyncHandlerNotSupportedError);