fix(batch): throw when a synchronous processor gets a promise-returning handler

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"

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<unknown> =>
+  typeof (value as PromiseLike<unknown> | undefined)?.then === 'function';
+
 export { BatchProcessorSync };

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<void> => {
+ * const recordHandler = (record: SQSRecord): void => {
  *   const payload = JSON.parse(record.body);
  * };
  *

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,

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,

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<void> => {
+ * 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<void> => {
+ * 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<void> => {
+ * const recordHandler = (record: SQSRecord): void => {
  *   const payload = JSON.parse(record.body);
  * };
  *

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<string> =>
+    record.body;
+
+  const rejectingRecordHandler = async (
+    _record: SQSRecord
+  ): Promise<string> => {
+    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);
+  });
+});