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..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'; /** @@ -111,15 +115,33 @@ 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)) { + // 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); } } +/** + * 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/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 new file mode 100644 index 0000000000..2fdda0f94f --- /dev/null +++ b/packages/batch/tests/unit/BatchProcessorSync.test.ts @@ -0,0 +1,67 @@ +import context from '@aws-lambda-powertools/testing-utils/context'; +import type { SQSRecord } from 'aws-lambda'; +import { describe, expect, it } from 'vitest'; +import { + AsyncHandlerNotSupportedError, + 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; + + 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 }; + const processor = new BatchProcessorSync(EventType.SQS); + + // Act & Assess + expect(() => + processPartialResponseSync(batch, asyncRecordHandler, processor, { + context, + }) + ).toThrow(AsyncHandlerNotSupportedError); + expect(processor.successMessages).toHaveLength(0); + }); + + 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 & 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, + }) + ).toThrow(AsyncHandlerNotSupportedError); + }); + + 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(AsyncHandlerNotSupportedError); + }); +});