diff --git a/src/backend/ingestion/index.ts b/src/backend/ingestion/index.ts index af951ae29..460df5110 100644 --- a/src/backend/ingestion/index.ts +++ b/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); } } diff --git a/src/backend/ingestion/list-objects.lambda.ts b/src/backend/ingestion/list-objects.lambda.ts new file mode 100644 index 000000000..521580742 --- /dev/null +++ b/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 { + 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; +} diff --git a/src/backend/ingestion/list-objects.ts b/src/backend/ingestion/list-objects.ts new file mode 100644 index 000000000..38ebff13c --- /dev/null +++ b/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')), + }); + } +}