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 cd5efee509..92fc2a48b3 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'; /** @@ -104,22 +108,50 @@ 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 */ 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)) { + // 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(); + } + + 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 === 'object' && + value !== null && + 'then' in value && + typeof value.then === 'function'; + export { BatchProcessorSync }; 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..dbcc9c8f1e 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. 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'; + } +} + /** * 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 new file mode 100644 index 0000000000..e0d61c2615 --- /dev/null +++ b/packages/batch/tests/unit/BatchProcessorSync.test.ts @@ -0,0 +1,66 @@ +import context from '@aws-lambda-powertools/testing-utils/context'; +import { describe, expect, it, vi } from 'vitest'; +import { + AsyncHandlerNotSupportedError, + BatchProcessorSync, + EventType, + processPartialResponseSync, + SqsFifoPartialProcessor, +} from '../../src/index.js'; +import { sqsRecordFactory } from '../helpers/factories.js'; +import { asyncSqsRecordHandler } from '../helpers/handlers.js'; + +describe('Class: BatchProcessorSync', () => { + 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, asyncSqsRecordHandler, processor, { + context, + }) + ).toThrow(AsyncHandlerNotSupportedError); + expect(processor.successMessages).toHaveLength(0); + }); + + it('logs the rejection reason instead of leaving an unhandled rejection when the promise rejects', async () => { + // Prepare + const records = [sqsRecordFactory('fail'), sqsRecordFactory('success')]; + const batch = { Records: records }; + const processor = new BatchProcessorSync(EventType.SQS); + + // Act + // The abandoned promise rejects after the throw, and vitest fails this file + // if nothing has taken ownership of that rejection. + expect(() => + 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', () => { + // Prepare + const records = [sqsRecordFactory('success'), sqsRecordFactory('success')]; + const batch = { Records: records }; + const processor = new SqsFifoPartialProcessor(); + + // Act & Assess + expect(() => + processPartialResponseSync(batch, asyncSqsRecordHandler, processor, { + context, + }) + ).toThrow(AsyncHandlerNotSupportedError); + }); +});