fix(ingestion): replace SDK integration with Lambda for S3 listing

src/backend/ingestion/index.ts

@@ -28,13 +28,13 @@ import {
   WaitTime,
 } from 'aws-cdk-lib/aws-stepfunctions';
 import {
-  CallAwsService,
   LambdaInvoke,
   StepFunctionsStartExecution,
 } from 'aws-cdk-lib/aws-stepfunctions-tasks';
 import { Construct } from 'constructs';
 import { MetricName, METRICS_NAMESPACE } from './constants';
 import { Ingestion as Handler } from './ingestion';
+import { ListObjects } from './list-objects';
 import { ReIngest } from './re-ingest';
 import { Repository } from '../../codeartifact/repository';
 import {
@@ -474,23 +474,39 @@ class ReprocessIngestionWorkflow extends Construct {
     // Need to physical-name the state machine so it can self-invoke.
     const stateMachineName = stateMachineNameFrom(this.node.path);
 
-    const listObjectsWithMaxKeys = (
-      name: string,
-      maxKeys: number,
-      token?: string
-    ) =>
-      new CallAwsService(this, name, {
-        service: 's3',
-        action: 'listObjectsV2',
-        iamAction: 's3:ListBucket',
-        iamResources: [props.bucket.bucketArn],
-        parameters: {
-          Bucket: props.bucket.bucketName,
-          ContinuationToken: token,
-          Prefix: STORAGE_KEY_PREFIX,
-          MaxKeys: maxKeys,
-        },
+    // Use a Lambda function to list objects instead of the Step Functions SDK
+    // integration. The Lambda returns only the Key field for each object,
+    // eliminating the 256KB payload limit issue that occurs when S3 returns
+    // large responses with per-object metadata (ETag, LastModified, Size, etc.).
+    const listObjectsFunction = new ListObjects(this, 'ListObjectsFunction', {
+      architecture: gravitonLambdaIfAvailable(this),
+      description:
+        '[ConstructHub/Ingestion/ListObjects] Lists S3 objects for reprocessing workflow',
+      environment: {
+        BUCKET_NAME: props.bucket.bucketName,
+        PREFIX: STORAGE_KEY_PREFIX,
+        MAX_KEYS: '1000',
+      },
+      memorySize: 256,
+      tracing: Tracing.ACTIVE,
+      timeout: Duration.minutes(1),
+    });
+
+    props.bucket.grantRead(listObjectsFunction);
+
+    const listObjects = (name: string, token?: string) =>
+      new LambdaInvoke(this, name, {
+        lambdaFunction: listObjectsFunction,
+        payload: token
+          ? TaskInput.fromObject({ ContinuationToken: token })
+          : TaskInput.fromObject({}),
+        payloadResponseOnly: true,
         resultPath: '$.response',
+      }).addRetry({
+        errors: ['Lambda.TooManyRequestsException'],
+        backoffRate: 2,
+        interval: Duration.seconds(10),
+        maxAttempts: 6,
       });
 
     const process = new Map(this, 'Process Result', {
@@ -542,60 +558,17 @@ class ReprocessIngestionWorkflow extends Construct {
       .afterwards({ includeOtherwise: true })
       .next(process);
 
-    const listObjects = (name: string, token?: string) => {
-      // Start at 750 (not 1000) to avoid an uncatchable edge case where
-      // the S3 response is just under 256KB, the state reports success,
-      // but Step Functions aborts the execution at the state transition
-      // boundary when internal metadata pushes the total over the limit.
-      const startMaxKeysValue = 750;
-      const minMaxKeysValue = 150;
-      const decrement = 100;
-
-      // Create the first task with maximum MaxKeys value
-      const firstTask = listObjectsWithMaxKeys(
-        `${name}Try${startMaxKeysValue}`,
-        startMaxKeysValue,
-        token
-      ).addRetry({ errors: ['S3.SdkClientException'] });
-
-      firstTask.next(isThereMore);
-
-      // Chain tasks with decreasing MaxKeys values
-      let lastTask = firstTask;
-      for (
-        let maxKeys = startMaxKeysValue - decrement;
-        maxKeys >= minMaxKeysValue;
-        maxKeys -= decrement
-      ) {
-        const nextTask = listObjectsWithMaxKeys(
-          `${name}Try${maxKeys}`,
-          maxKeys,
-          token
-        ).addRetry({ errors: ['S3.SdkClientException'] });
-
-        nextTask.next(isThereMore);
-
-        // Chain this task to the previous one using DataLimitExceeded catch
-        lastTask.addCatch(nextTask, {
-          errors: ['States.DataLimitExceeded'],
-          resultPath: JsonPath.DISCARD,
-        });
-
-        lastTask = nextTask;
-      }
-
-      return firstTask;
-    };
-
     const listBucket = new Choice(this, 'Has a ContinuationToken?')
       .when(
         Condition.isPresent('$.ContinuationToken'),
         listObjects(
           'S3.ListObjectsV2(NextPage)',
           JsonPath.stringAt('$.ContinuationToken')
-        )
+        ).next(isThereMore)
+      )
+      .otherwise(
+        listObjects('S3.ListObjectsV2(FirstPage)').next(isThereMore)
       )
-      .otherwise(listObjects('S3.ListObjectsV2(FirstPage)'))
       .afterwards();
 
     this.stateMachine = new StateMachine(this, 'StateMachine', {
@@ -604,7 +577,6 @@ class ReprocessIngestionWorkflow extends Construct {
       timeout: Duration.hours(1),
     });
 
-    props.bucket.grantRead(this.stateMachine);
     props.queue.grantSendMessages(this.stateMachine);
   }
 }

src/backend/ingestion/list-objects.lambda.ts

@@ -0,0 +1,62 @@
+import {
+  ListObjectsV2Command,
+  ListObjectsV2CommandOutput,
+} from '@aws-sdk/client-s3';
+import { S3_CLIENT } from '../shared/aws.lambda-shared';
+import { requireEnv } from '../shared/env.lambda-shared';
+
+interface Input {
+  ContinuationToken?: string;
+}
+
+interface Output {
+  Contents: Array<{ Key: string }>;
+  NextContinuationToken?: string;
+}
+
+/**
+ * Lists objects in the storage bucket, returning only the Key field for each
+ * object. This avoids the Step Functions 256KB payload limit that the direct
+ * SDK integration can hit when S3 returns large responses with per-object
+ * metadata (ETag, LastModified, Size, StorageClass, etc.).
+ */
+export async function handler(event: Input): Promise<Output> {
+  console.log('Event: ', JSON.stringify(event, null, 2));
+
+  const bucket = requireEnv('BUCKET_NAME');
+  const prefix = requireEnv('PREFIX');
+  const maxKeys = Number(requireEnv('MAX_KEYS'));
+
+  const params: {
+    Bucket: string;
+    Prefix: string;
+    MaxKeys: number;
+    ContinuationToken?: string;
+  } = {
+    Bucket: bucket,
+    Prefix: prefix,
+    MaxKeys: maxKeys,
+  };
+
+  if (event.ContinuationToken) {
+    params.ContinuationToken = event.ContinuationToken;
+  }
+
+  const response: ListObjectsV2CommandOutput = await S3_CLIENT.send(
+    new ListObjectsV2Command(params)
+  );
+
+  const output: Output = {
+    Contents: (response.Contents ?? []).map((obj) => ({ Key: obj.Key! })),
+  };
+
+  if (response.NextContinuationToken) {
+    output.NextContinuationToken = response.NextContinuationToken;
+  }
+
+  console.log(
+    `Listed ${output.Contents.length} objects, has more: ${!!output.NextContinuationToken}`
+  );
+
+  return output;
+}

src/backend/ingestion/list-objects.ts

@@ -0,0 +1,20 @@
+// ~~ Generated by projen. To modify, edit .projenrc.ts and run "npx projen".
+import * as path from 'path';
+import * as lambda from 'aws-cdk-lib/aws-lambda';
+import { Construct } from 'constructs';
+
+export interface ListObjectsProps extends lambda.FunctionOptions {
+}
+
+export class ListObjects extends lambda.Function {
+  constructor(scope: Construct, id: string, props?: ListObjectsProps) {
+    super(scope, id, {
+      description: 'backend/ingestion/list-objects.lambda.ts',
+      ...props,
+      architecture: lambda.Architecture.ARM_64,
+      runtime: lambda.Runtime.NODEJS_22_X,
+      handler: 'index.handler',
+      code: lambda.Code.fromAsset(path.join(__dirname, '/list-objects.lambda.bundle')),
+    });
+  }
+}