Maintenance: Add Batch E2E tests for Lambda Managed Instances

[svozza]

Summary

Add an end-to-end test validating Batch Processing's InvokeStore-backed state isolation on Lambda Managed Instances, where concurrent invocations are multiplexed into shared execution environments.

Part of #5092. This issue tracks the Batch portion; Logger is tracked in #5518 and Metrics in #5612.

Why is this needed?

BatchProcessingStore and SqsFifoProcessorStore scope records, the record handler, options, success and failure results, and FIFO failed-group tracking to the current invocation through InvokeStore. These code paths are currently exercised only by unit tests with mocked Lambda Managed Instances environments. Nothing proves that a BatchProcessor instance shared at module scope keeps its state isolated between concurrent invocations on the real platform.

Which area does this relate to?

• Tests
• Batch Processing

Solution

Add a Batch LMI end-to-end suite following the Logger and Metrics pattern from #5518 and #5612: a module-scoped promise barrier in the function code to guarantee an overlap, unique per-invocation identifiers, and many concurrent invocations held open so the scheduler multiplexes at least one pair into a shared execution environment. Reuse the shared capacity provider infrastructure and workflow support introduced by #5465.

Unlike Logger and Metrics there is no output stream to intercept: the observable state is the batchItemFailures response, so the handler simply returns it along with the message ids it processed.

Function code. Module-scoped BatchProcessor and SqsFifoPartialProcessor({ skipGroupOnError: true }) instances, as users write them. The record handler awaits the peer barrier on the first record of each batch only, so that overlapping invocations are both mid-process() with their own records registered in the store. That is the window in which a shared store would let one invocation's register() clobber the other's records and options, or mix failure messages.

const processor = new BatchProcessor(EventType.SQS);
const awaitPeer = createPeerBarrier();

export const handler = async (event: SQSEvent & { role: 'warmup' | 'test' }, context: Context) => {
  let first = true;
  let sawPeer = false;
  const recordHandler = async (record: SQSRecord) => {
    const { shouldFail } = JSON.parse(record.body);
    if (first && event.role === 'test') {
      first = false;
      sawPeer = await awaitPeer();
    }
    if (shouldFail) throw new Error(`Simulated failure for ${record.messageId}`);
  };
  const response = await processPartialResponse(event, recordHandler, processor, {
    context,
    throwOnFullBatchFailure: false,
  });
  return {
    invocationId: JSON.parse(event.Records[0].body).invocationId,
    executionEnvId,
    sawPeer,
    initializationType: process.env.AWS_LAMBDA_INITIALIZATION_TYPE ?? 'unset',
    maxConcurrency: process.env.AWS_LAMBDA_MAX_CONCURRENCY ?? 'unset',
    receivedMessageIds: event.Records.map((r) => r.messageId),
    failedMessageIds: response.batchItemFailures.map((f) => f.itemIdentifier),
  };
};

A second export does the same with the FIFO processor.

Test. Each invocation sends a synthetic SQS event whose message ids embed the invocation id (inv-3-msg-0) and whose bodies carry { invocationId, shouldFail }, with a failure pattern that differs between invocations. For the FIFO variant every invocation uses the same message group names (group-0, group-1), so a failed-group id leaked from a peer would make the processor skip this invocation's healthy records. Assertions per invocation, on the returned payload:

• receivedMessageIds equals the invocation's own ids: catches records or handler overwritten by a peer's register().
• failedMessageIds as a set equals the expected failures for that invocation's pattern (for FIFO, including the records short-circuited after a group failure): catches mixed failure messages, leaked failed-group ids, and failures lost to a peer's clear().
• every failed id starts with the invocation's own prefix.
• at least one result has sawPeer === true, and every result reports AWS_LAMBDA_INITIALIZATION_TYPE=lambda-managed-instances and AWS_LAMBDA_MAX_CONCURRENCY=10, as in the Metrics suite.

Run as describe.each over the standard and FIFO handlers, each attached to the capacity provider with perExecutionEnvironmentMaxConcurrency: 10.

Acknowledgment

• This request meets Powertools for AWS Lambda (TypeScript) Tenets
• Should this be considered in other Powertools for AWS Lambda languages? i.e. Python, Java, and .NET

Future readers

Please react with 👍 and your use case to help us understand customer demand.