diff --git a/bun.lock b/bun.lock index 69e180ee89..4bf3159aea 100644 --- a/bun.lock +++ b/bun.lock @@ -26,8 +26,9 @@ }, "packages/aws-lambda": { "name": "@hyperframes/aws-lambda", - "version": "0.8.150", + "version": "0.8.169", "dependencies": { + "@aws-sdk/client-lambda": "^3.1146.0", "@aws-sdk/client-s3": "^3.1146.0", "@aws-sdk/client-sfn": "^3.1146.0", "@hyperframes/producer": "workspace:^", @@ -59,7 +60,7 @@ }, "packages/cli": { "name": "@hyperframes/cli", - "version": "0.8.150", + "version": "0.8.169", "bin": { "hyperframes": "./bin/hyperframes.mjs", "hyperframes-localize-fonts": "./bin/hyperframes-localize-fonts.mjs", @@ -112,7 +113,7 @@ }, "packages/core": { "name": "@hyperframes/core", - "version": "0.8.150", + "version": "0.8.169", "dependencies": { "@chenglou/pretext": "^0.0.5", "@hyperframes/lint": "workspace:*", @@ -157,7 +158,7 @@ }, "packages/engine": { "name": "@hyperframes/engine", - "version": "0.8.150", + "version": "0.8.169", "dependencies": { "@hono/node-server": "^2.0.5", "@hyperframes/core": "workspace:^", @@ -176,7 +177,7 @@ }, "packages/gcp-cloud-run": { "name": "@hyperframes/gcp-cloud-run", - "version": "0.8.150", + "version": "0.8.169", "dependencies": { "@google-cloud/storage": "^7.14.0", "@google-cloud/workflows": "^4.2.0", @@ -197,7 +198,7 @@ }, "packages/inspector": { "name": "@hyperframes/inspector", - "version": "0.8.150", + "version": "0.8.169", "dependencies": { "@hyperframes/core": "workspace:*", "@hyperframes/parsers": "workspace:*", @@ -216,7 +217,7 @@ }, "packages/lint": { "name": "@hyperframes/lint", - "version": "0.8.150", + "version": "0.8.169", "dependencies": { "@hyperframes/parsers": "workspace:*", "acorn": "^8.17.0", @@ -236,7 +237,7 @@ }, "packages/parsers": { "name": "@hyperframes/parsers", - "version": "0.8.150", + "version": "0.8.169", "dependencies": { "@babel/parser": "^7.27.0", "acorn": "^8.17.0", @@ -257,7 +258,7 @@ }, "packages/player": { "name": "@hyperframes/player", - "version": "0.8.150", + "version": "0.8.169", "dependencies": { "@hyperframes/core": "workspace:*", }, @@ -272,7 +273,7 @@ }, "packages/producer": { "name": "@hyperframes/producer", - "version": "0.8.150", + "version": "0.8.169", "dependencies": { "@hono/node-server": "^2.0.5", "@hyperframes/core": "workspace:*", @@ -298,7 +299,7 @@ }, "packages/sdk": { "name": "@hyperframes/sdk", - "version": "0.8.150", + "version": "0.8.169", "dependencies": { "@hyperframes/core": "workspace:*", "@hyperframes/parsers": "workspace:*", @@ -328,7 +329,7 @@ }, "packages/shader-transitions": { "name": "@hyperframes/shader-transitions", - "version": "0.8.150", + "version": "0.8.169", "dependencies": { "html2canvas": "^1.4.1", }, @@ -340,7 +341,7 @@ }, "packages/studio": { "name": "@hyperframes/studio", - "version": "0.8.150", + "version": "0.8.169", "dependencies": { "@base-ui/react": "1.7.0", "@codemirror/autocomplete": "^6.20.1", @@ -393,7 +394,7 @@ }, "packages/studio-server": { "name": "@hyperframes/studio-server", - "version": "0.8.150", + "version": "0.8.169", "dependencies": { "@hyperframes/core": "workspace:*", "@hyperframes/parsers": "workspace:*", @@ -437,6 +438,8 @@ "@aws-sdk/checksums": ["@aws-sdk/checksums@3.1001.1", "", { "dependencies": { "@aws-sdk/core": "^3.978.1", "@aws-sdk/types": "^3.974.6", "@smithy/core": "^3.35.0", "@smithy/types": "^4.19.0", "tslib": "^2.6.2" } }, "sha512-x12Q17KYlJAd3nKf8LV5LV0vt8sh8/6YfQLGPtrGnQf/tW4jqxPGq5GPpuVitpQYM3eUR4XB7CbxZf751NMbLw=="], + "@aws-sdk/client-lambda": ["@aws-sdk/client-lambda@3.1146.0", "", { "dependencies": { "@aws-sdk/core": "^3.978.1", "@aws-sdk/credential-provider-node": "^3.972.84", "@aws-sdk/types": "^3.974.6", "@smithy/core": "^3.35.0", "@smithy/fetch-http-handler": "^5.8.0", "@smithy/node-http-handler": "^4.12.1", "@smithy/types": "^4.19.0", "tslib": "^2.6.2" } }, "sha512-qCeVHGhkI5aw0Lvexhwuf1H41L2sCm+9C4K+IOUhErY/nx5ioCU0dFyXF4+P4m4Nk23uiTbON7lI7nmlPNncHw=="], + "@aws-sdk/client-s3": ["@aws-sdk/client-s3@3.1146.0", "", { "dependencies": { "@aws-sdk/checksums": "^3.1001.1", "@aws-sdk/core": "^3.978.1", "@aws-sdk/credential-provider-node": "^3.972.84", "@aws-sdk/middleware-sdk-s3": "^3.972.77", "@aws-sdk/signature-v4-multi-region": "^3.996.47", "@aws-sdk/types": "^3.974.6", "@smithy/core": "^3.35.0", "@smithy/fetch-http-handler": "^5.8.0", "@smithy/node-http-handler": "^4.12.1", "@smithy/types": "^4.19.0", "tslib": "^2.6.2" } }, "sha512-WY0YCBzxc4muFfY6UbGpA+oib0nGT/Px2aEPGYX7pp1la2OmDLChnK81mu8H7LdYvwGU1pLAsLIa/qgad0Wd7g=="], "@aws-sdk/client-sfn": ["@aws-sdk/client-sfn@3.1146.0", "", { "dependencies": { "@aws-sdk/core": "^3.978.1", "@aws-sdk/credential-provider-node": "^3.972.84", "@aws-sdk/types": "^3.974.6", "@smithy/core": "^3.35.0", "@smithy/fetch-http-handler": "^5.8.0", "@smithy/node-http-handler": "^4.12.1", "@smithy/types": "^4.19.0", "tslib": "^2.6.2" } }, "sha512-u4IjYTOm0Vo7mPxZdx11mOMUsMqcclXSoh5zpCcka/N38sEIYVLSU92ugHwQicg5njePkysvNc1SXXhP7DejkA=="], @@ -1319,7 +1322,6 @@ "bowser": ["bowser@2.14.1", "", {}, "sha512-tzPjzCxygAKWFOJP011oxFHs57HzIhOEracIgAePE4pqB3LikALKnSzUyU4MGs9/iCEUuHlAJTjTc5M+u7YEGg=="], - "brace-expansion": ["brace-expansion@5.0.6", "", { "dependencies": { "balanced-match": "^4.0.2" } }, "sha512-kLpxurY4Z4r9sgMsyG0Z9uzsBlgiU/EFKhj/h91/8yHu0edo7XuixOIH3VcJ8kkxs6/jPzoI6U9Vj3WqbMQ94g=="], "brotli": ["brotli@1.3.3", "", { "dependencies": { "base64-js": "^1.1.2" } }, "sha512-oTKjJdShmDuGW94SyyaoQvAjf30dZaHnjJ8uAF+u2/vGJkJbJPJAT1gDiOJP5v1Zb6f9KEyW/1HpuaWIXtGHPg=="], diff --git a/docs/deploy/aws-lambda.mdx b/docs/deploy/aws-lambda.mdx index dc59d80158..dd4a0e00c2 100644 --- a/docs/deploy/aws-lambda.mdx +++ b/docs/deploy/aws-lambda.mdx @@ -60,14 +60,20 @@ The CLI is a thin wrapper around the SAM template + the `@hyperframes/aws-lambda hyperframes lambda deploy \ --stack-name=hyperframes-prod \ --region=us-east-1 \ - --concurrency=8 \ --memory=10240 ``` -The default `--concurrency=8` is deliberately conservative for first-time -users. It limits how many Lambda workers the deployed stack can run at once. -Raise it only after you have measured a typical render and checked the account's -limits and budget. +By default the stack reserves no concurrency. Each render reads the account's +unreserved Lambda concurrency when it plans, and runs up to 90% of it at once +(at most 256 chunks), leaving the rest for other functions in the account. The +plan logs the number it chose. Pass `--concurrency=N` to reserve `N` executions +for the function instead; every render then runs at most `N` chunks at once. + +Upgrading: redeploying an existing stack without `--concurrency` keeps the +reservation it already has, and `deploy` prints it. Stacks deployed by earlier +CLI versions reserved 8 by default, so they stay at 8 until you pass +`--concurrency=N`. Pass `--concurrency=none` to remove the reservation so each +render sizes itself to the account. After `deploy`, render anything with: @@ -117,7 +123,7 @@ The template emits three CloudFormation outputs you'll need to invoke renders: - `RenderFunctionArn` — the single Lambda function the state machine dispatches against. -The SAM template's own default for `ReservedConcurrency` is `-1` (unreserved, account-default). The Path 1 CLI overrides it to `8` to keep first-time spend bounded; if you drop `ReservedConcurrency` from `--parameter-overrides` here, you get the unreserved default. Set it explicitly unless you've already sized your typical render's fan-out. +The SAM template's default for `ReservedConcurrency` is `-1` (unreserved), the same as the Path 1 CLI: each render runs up to 90% of the account's unreserved concurrency. Set it to bound how many chunks run at once. ### Path 3 — CDK construct diff --git a/docs/deploy/migrating-to-hyperframes-lambda.mdx b/docs/deploy/migrating-to-hyperframes-lambda.mdx index 9ff021f4cf..7301df8033 100644 --- a/docs/deploy/migrating-to-hyperframes-lambda.mdx +++ b/docs/deploy/migrating-to-hyperframes-lambda.mdx @@ -58,7 +58,7 @@ Most adopters' render config maps directly: | Output format | `--format=mp4 / mov / webm / png-sequence` | webm uses libvpx-vp9 + closed-GOP concat-copy. Distributed mode still refuses HDR mp4 at plan time. | | Quality preset | `--quality=draft / standard / high` | Maps onto ffmpeg encoder presets. | | Chunk size in frames | `--chunk-size=240` (default: auto) | By default each chunk is sized to finish inside the Lambda timeout with headroom; a render that cannot fails at plan time with `CHUNK_EXCEEDS_WORKER_TIMEOUT`. | -| Max parallel chunks | `--max-parallel-chunks=16` (default: auto) | By default one wave across the deploy's `--concurrency` (16 if unknown), widened when a chunk would outrun the Lambda timeout. | +| Max parallel chunks | `--max-parallel-chunks=16` (default: auto) | By default one wave across the deploy's `--concurrency` (or 90% of the account's unreserved concurrency), widened when a chunk would outrun the Lambda timeout. | | Per-chunk frame ceiling | `--target-chunk-frames=N` (optional) | Caps frames per chunk so one chunk can't run past Lambda's 15-min cap on a long video: the planner adds chunks (up to `--max-parallel-chunks`) to keep each at or below `N`, and short videos still collapse to fewer chunks. A ceiling, not a fixed size; ignored when `--chunk-size` is set. | | Bitrate / CRF | `config.bitrate` or `config.crf` in the SDK | Mutually exclusive; the current Lambda CLI does not expose these two fields. | diff --git a/docs/packages/cli.mdx b/docs/packages/cli.mdx index d8d1dcc1b5..c64636273c 100644 --- a/docs/packages/cli.mdx +++ b/docs/packages/cli.mdx @@ -1932,7 +1932,7 @@ deploy: | `--stack-name` | `hyperframes-default` | CloudFormation stack name | | `--region` | `AWS_REGION`, else `us-east-1` | AWS region | | `--profile` | `AWS_PROFILE` | AWS profile name | -| `--concurrency` | `8` | Lambda reserved concurrency | +| `--concurrency` | none | Lambda reserved concurrency. Without it, a new stack reserves none and each render runs up to 90% of the account's unreserved concurrency; a redeploy keeps the stack's current reservation; `none` removes it | | `--memory` | `10240` | Lambda memory in MB | | `--chrome-source` | `sparticuz` | `sparticuz` or `chrome-headless-shell` | | `--skip-build` | off | Reuse the existing `handler.zip` | @@ -1949,7 +1949,7 @@ And these drive `render` and `render-batch`: | `--quality` | `standard` | `draft`, `standard`, or `high` | | `--output-resolution` | — | Supersample via Chrome `deviceScaleFactor` to a preset (`1080p`, `4k`, `uhd`, `hd`, `landscape-4k`, …) without changing layout | | `--chunk-size` | auto | Frames per chunk. By default each chunk is sized to finish inside the Lambda timeout | -| `--max-parallel-chunks` | auto | Chunks per render. By default one wave across the deploy's `--concurrency` (16 if unknown), widened when a chunk would outrun the Lambda timeout | +| `--max-parallel-chunks` | auto | Chunks per render. By default one wave across the deploy's `--concurrency` (or 90% of the account's unreserved concurrency), widened when a chunk would outrun the Lambda timeout | | `--target-chunk-frames` | — | Ceiling on frames per chunk. The planner adds chunks — up to `--max-parallel-chunks` — to stay under it, and short videos still collapse to fewer. Ignored when `--chunk-size` is set. | | `--variables` | — | JSON object reaching every chunk worker as `window.__hfVariables` | | `--variables-file` | — | Read those from a JSON file | @@ -2055,7 +2055,8 @@ batch does not try to spawn 10,000 executions and trip your account limits. It is orchestrator-side only, and cannot enforce your account's Lambda concurrency quota — pick a value from that quota and the reserved concurrency you set at `lambda deploy --concurrency=`. It is not `--max-parallel-chunks`, which caps -chunks inside one render. +chunks inside one render. Without a reserved concurrency every render in the +batch sizes itself to the whole account, so reserve one before a large batch. ### `lambda progress ` diff --git a/examples/aws-lambda/scripts/_semantic-compare.sh b/examples/aws-lambda/scripts/_semantic-compare.sh index d90be6292a..cd57e9f6a4 100755 --- a/examples/aws-lambda/scripts/_semantic-compare.sh +++ b/examples/aws-lambda/scripts/_semantic-compare.sh @@ -57,15 +57,15 @@ hf_decode_pcm() { -ac 2 -ar 48000 -c:a pcm_s16le -f s16le "$2" } +# Chunks run in Distributed Map child executions; the Map's exit output carries every result. hf_extract_chunk_hashes() { local history="$1" output="$2" jq '[ .events[] | - .taskSucceededEventDetails.output? | - select(type == "string") | - (try fromjson catch empty) | - .Payload | - select(type == "object" and .Action == "renderChunk") | + select(.type == "MapStateExited") | + .stateExitedEventDetails | + select(.name == "RenderChunks" or .name == "RenderChunksV2") | + (.output | fromjson | .Chunks // [])[] | {ChunkIndex, Sha256, FramesEncoded} ] | sort_by(.ChunkIndex)' "$history" > "$output" } diff --git a/examples/aws-lambda/scripts/semantic-compare.test.sh b/examples/aws-lambda/scripts/semantic-compare.test.sh index 99de4c5f5c..22841d357c 100755 --- a/examples/aws-lambda/scripts/semantic-compare.test.sh +++ b/examples/aws-lambda/scripts/semantic-compare.test.sh @@ -39,28 +39,25 @@ jq -e ' .encoded.equal == false ' "$WORK/equal.json" >/dev/null -cat > "$WORK/v1-history.json" <<'JSON' -{"events":[ - {"taskSucceededEventDetails":{"output":"{\"Payload\":{\"Action\":\"renderChunk\",\"ChunkIndex\":1,\"Sha256\":\"bbb\",\"FramesEncoded\":12}}"}}, - {"taskSucceededEventDetails":{"output":"{\"Payload\":{\"Action\":\"plan\",\"ChunkCount\":2}}"}}, - {"taskSucceededEventDetails":{"output":"{\"Payload\":{\"Action\":\"renderChunk\",\"ChunkIndex\":0,\"Sha256\":\"aaa\",\"FramesEncoded\":12}}"}} -]} -JSON -cat > "$WORK/v2-history.json" <<'JSON' -{"events":[ - {"taskSucceededEventDetails":{"output":"{\"Payload\":{\"Action\":\"renderChunk\",\"ChunkIndex\":0,\"Sha256\":\"aaa\",\"FramesEncoded\":12}}"}}, - {"taskSucceededEventDetails":{"output":"{\"Payload\":{\"Action\":\"renderChunk\",\"ChunkIndex\":1,\"Sha256\":\"bbb\",\"FramesEncoded\":12}}"}} -]} -JSON +# Map exit output as Step Functions records it: the whole state, chunk results under Chunks. +map_exit() { + jq -nc --arg name "$1" --argjson chunks "$2" \ + '{type: "MapStateExited", stateExitedEventDetails: {name: $name, output: ({Plan: {ChunkCount: 2}, Chunks: $chunks} | tojson)}}' +} +CHUNKS='[{"ChunkIndex":1,"Sha256":"bbb","FramesEncoded":12,"DurationMs":9},{"ChunkIndex":0,"Sha256":"aaa","FramesEncoded":12,"DurationMs":8}]' +jq -n --argjson exit "$(map_exit RenderChunks "$CHUNKS")" \ + '{events: [{type: "TaskSucceeded", taskSucceededEventDetails: {output: "{}"}}, $exit]}' \ + > "$WORK/v1-history.json" +jq -n --argjson exit "$(map_exit RenderChunksV2 "$CHUNKS")" '{events: [$exit]}' \ + > "$WORK/v2-history.json" hf_compare_render_semantics \ "$WORK/original.mp4" "$WORK/rewrapped.mp4" "$WORK/chunks-equal" \ "$WORK/v1-history.json" "$WORK/v2-history.json" jq -e '.semanticEqual == true and .chunks == {checked:true,equal:true,v1Count:2,v2Count:2}' \ "$WORK/chunks-equal.json" >/dev/null -jq '(.events[0].taskSucceededEventDetails.output) = - "{\"Payload\":{\"Action\":\"renderChunk\",\"ChunkIndex\":0,\"Sha256\":\"different\",\"FramesEncoded\":12}}"' \ - "$WORK/v2-history.json" > "$WORK/v2-history-mismatch.json" +jq -n --argjson exit "$(map_exit RenderChunksV2 "$(jq -c '.[1].Sha256 = "different"' <<<"$CHUNKS")")" \ + '{events: [$exit]}' > "$WORK/v2-history-mismatch.json" if hf_compare_render_semantics \ "$WORK/original.mp4" "$WORK/rewrapped.mp4" "$WORK/chunks-mismatch" \ "$WORK/v1-history.json" "$WORK/v2-history-mismatch.json"; then diff --git a/examples/aws-lambda/template.yaml b/examples/aws-lambda/template.yaml index 45475d7fe8..d44d00d437 100644 --- a/examples/aws-lambda/template.yaml +++ b/examples/aws-lambda/template.yaml @@ -43,7 +43,8 @@ Parameters: Description: >- Lambda reserved concurrency cap. Set to a positive integer to bound simultaneous chunk renders (e.g. 50 to limit cost). -1 means - unreserved (account default). + unreserved: each render runs up to 90% of the account's unreserved + concurrency. ChromeSource: Type: String @@ -159,6 +160,7 @@ Resources: # clarity + so users can override during local SAM invoke. TMPDIR: /tmp HYPERFRAMES_RENDER_BUCKET: !Ref RenderBucket + HYPERFRAMES_LAMBDA_RESERVED_CONCURRENCY: !Ref ReservedConcurrency Policies: - S3CrudPolicy: BucketName: !Ref RenderBucket @@ -166,6 +168,11 @@ Resources: - Effect: Allow Action: s3:AbortMultipartUpload Resource: !Sub "arn:${AWS::Partition}:s3:::${RenderBucket}/*" + # Plan sizes an unreserved render to the account's concurrency. + - Statement: + - Effect: Allow + Action: lambda:GetAccountSettings + Resource: "*" # CloudWatch Logs perms are covered by SAM's default # AWSLambdaBasicExecutionRole — explicit `CloudWatchLogsFullAccess` # would be overscope (`logs:*` on `*`, including DeleteLogGroup + @@ -249,6 +256,7 @@ Resources: PlanS3Uri.$: "$.Payload.PlanS3Uri" PlanHash.$: "$.Payload.PlanHash" ChunkCount.$: "$.Payload.ChunkCount" + MaxChunkConcurrency.$: "$.Payload.MaxChunkConcurrency" Format.$: "$.Payload.Format" HasAudio.$: "$.Payload.HasAudio" AudioS3Uri.$: "$.Payload.AudioS3Uri" @@ -274,7 +282,15 @@ Resources: - NotMediaPayloadError - PLAN_ARTIFACT_DIGEST_MISMATCH - FORMAT_NOT_SUPPORTED_IN_DISTRIBUTED + - CHUNK_EXCEEDS_WORKER_TIMEOUT MaxAttempts: 0 + # Another render on the same function can still take the reserved slots; wait minutes. + - ErrorEquals: [Lambda.TooManyRequestsException] + IntervalSeconds: 5 + MaxAttempts: 10 + BackoffRate: 2 + MaxDelaySeconds: 60 + JitterStrategy: FULL - ErrorEquals: [States.ALL] IntervalSeconds: 2 MaxAttempts: 4 @@ -299,6 +315,7 @@ Resources: PlanV2ArtifactS3Prefix.$: "$.Payload.PlanV2ArtifactS3Prefix" PlanHash.$: "$.Payload.PlanHash" ChunkCount.$: "$.Payload.ChunkCount" + MaxChunkConcurrency.$: "$.Payload.MaxChunkConcurrency" Format.$: "$.Payload.Format" HasAudio.$: "$.Payload.HasAudio" ResultPath: $.Plan @@ -321,8 +338,15 @@ Resources: - PlanV2IntegrityError - PLAN_ARTIFACT_DIGEST_MISMATCH - FORMAT_NOT_SUPPORTED_IN_DISTRIBUTED + - CHUNK_EXCEEDS_WORKER_TIMEOUT - ChromeBinaryUnavailableError MaxAttempts: 0 + - ErrorEquals: [Lambda.TooManyRequestsException] + IntervalSeconds: 5 + MaxAttempts: 10 + BackoffRate: 2 + MaxDelaySeconds: 60 + JitterStrategy: FULL - ErrorEquals: [States.ALL] IntervalSeconds: 2 MaxAttempts: 4 @@ -369,6 +393,7 @@ Resources: RenderChunks: Type: Map + Label: RenderChunks ItemsPath: $.Iterator.ChunkIndexes ItemSelector: ChunkIndex.$: "$$.Map.Item.Value" @@ -376,17 +401,15 @@ Resources: PlanHash.$: "$.Plan.PlanHash" ChunkOutputS3Prefix.$: "$.PlanOutputS3Prefix" Format.$: "$.Plan.Format" - # Map fan-out cap derives from the Plan's chunkCount so - # caller-supplied `Config.maxParallelChunks` (which - # `plan()` honours when sizing the chunk list) is the - # single source of truth. A hardcoded value here would - # silently throttle adopters who scale up the chunk count - # in their event payload. - MaxConcurrencyPath: $.Plan.ChunkCount + # Chunks past the plan's concurrency queue here instead of being throttled by Lambda. + MaxConcurrencyPath: $.Plan.MaxChunkConcurrency ResultPath: $.Chunks ItemProcessor: ProcessorConfig: - Mode: INLINE + # Inline Map runs at most 40 iterations at once. STANDARD children + # because EXPRESS ones stop at 5 minutes and a chunk may run 15. + Mode: DISTRIBUTED + ExecutionType: STANDARD StartAt: RenderChunk States: RenderChunk: @@ -406,6 +429,8 @@ Resources: ChunkS3Uri.$: "$.Payload.ChunkS3Uri" ChunkIndex.$: "$.Payload.ChunkIndex" Sha256.$: "$.Payload.Sha256" + FramesEncoded.$: "$.Payload.FramesEncoded" + DurationMs.$: "$.Payload.DurationMs" Retry: - ErrorEquals: - FFMPEG_VERSION_MISMATCH @@ -418,6 +443,12 @@ Resources: - INVALID_VIDEO_METADATA - PLAN_ARTIFACT_DIGEST_MISMATCH MaxAttempts: 0 + - ErrorEquals: [Lambda.TooManyRequestsException] + IntervalSeconds: 5 + MaxAttempts: 10 + BackoffRate: 2 + MaxDelaySeconds: 60 + JitterStrategy: FULL - ErrorEquals: [States.ALL] IntervalSeconds: 2 MaxAttempts: 4 @@ -460,6 +491,12 @@ Resources: - PlanProtocolUnsupportedError - PLAN_ARTIFACT_DIGEST_MISMATCH MaxAttempts: 0 + - ErrorEquals: [Lambda.TooManyRequestsException] + IntervalSeconds: 5 + MaxAttempts: 10 + BackoffRate: 2 + MaxDelaySeconds: 60 + JitterStrategy: FULL - ErrorEquals: [States.ALL] IntervalSeconds: 2 MaxAttempts: 4 @@ -469,6 +506,7 @@ Resources: RenderChunksV2: Type: Map + Label: RenderChunksV2 ItemsPath: $.Iterator.ChunkIndexes ItemSelector: ChunkIndex.$: "$$.Map.Item.Value" @@ -477,11 +515,13 @@ Resources: PlanHash.$: "$.Plan.PlanHash" ChunkOutputS3Prefix.$: "$.PlanOutputS3Prefix" Format.$: "$.Plan.Format" - MaxConcurrencyPath: $.Plan.ChunkCount + # Chunks past the plan's concurrency queue here instead of being throttled by Lambda. + MaxConcurrencyPath: $.Plan.MaxChunkConcurrency ResultPath: $.Chunks ItemProcessor: ProcessorConfig: - Mode: INLINE + Mode: DISTRIBUTED + ExecutionType: STANDARD StartAt: RenderChunkV2 States: RenderChunkV2: @@ -502,6 +542,8 @@ Resources: ChunkS3Uri.$: "$.Payload.ChunkS3Uri" ChunkIndex.$: "$.Payload.ChunkIndex" Sha256.$: "$.Payload.Sha256" + FramesEncoded.$: "$.Payload.FramesEncoded" + DurationMs.$: "$.Payload.DurationMs" Retry: - ErrorEquals: - FFMPEG_VERSION_MISMATCH @@ -518,6 +560,12 @@ Resources: - PLAN_ARTIFACT_DIGEST_MISMATCH - ChromeBinaryUnavailableError MaxAttempts: 0 + - ErrorEquals: [Lambda.TooManyRequestsException] + IntervalSeconds: 5 + MaxAttempts: 10 + BackoffRate: 2 + MaxDelaySeconds: 60 + JitterStrategy: FULL - ErrorEquals: [States.ALL] IntervalSeconds: 2 MaxAttempts: 4 @@ -561,6 +609,12 @@ Resources: - PLAN_ARTIFACT_DIGEST_MISMATCH - ChromeBinaryUnavailableError MaxAttempts: 0 + - ErrorEquals: [Lambda.TooManyRequestsException] + IntervalSeconds: 5 + MaxAttempts: 10 + BackoffRate: 2 + MaxDelaySeconds: 60 + JitterStrategy: FULL - ErrorEquals: [States.ALL] IntervalSeconds: 2 MaxAttempts: 4 @@ -587,6 +641,20 @@ Resources: - Effect: Allow Action: lambda:InvokeFunction Resource: !GetAtt RenderFunction.Arn + # A Distributed Map runs each chunk as a child execution of this state machine. + # ARNs built by name: !Ref RenderStateMachine here would be a dependency cycle. + - PolicyName: RunDistributedMapChildren + PolicyDocument: + Version: "2012-10-17" + Statement: + - Effect: Allow + Action: states:StartExecution + Resource: !Sub "arn:${AWS::Partition}:states:${AWS::Region}:${AWS::AccountId}:stateMachine:${ProjectName}-render" + - Effect: Allow + Action: + - states:DescribeExecution + - states:StopExecution + Resource: !Sub "arn:${AWS::Partition}:states:${AWS::Region}:${AWS::AccountId}:execution:${ProjectName}-render:*" - PolicyName: WriteCloudwatchLogs PolicyDocument: Version: "2012-10-17" diff --git a/packages/aws-lambda/build.mjs b/packages/aws-lambda/build.mjs index 93532e654b..d639aca5c5 100644 --- a/packages/aws-lambda/build.mjs +++ b/packages/aws-lambda/build.mjs @@ -30,6 +30,7 @@ const sharedOpts = { format: "esm", minify: false, external: [ + "@aws-sdk/client-lambda", "@aws-sdk/client-s3", "@aws-sdk/client-sfn", "@hyperframes/producer", diff --git a/packages/aws-lambda/package.json b/packages/aws-lambda/package.json index 7f8b9f0986..17a5b76bf1 100644 --- a/packages/aws-lambda/package.json +++ b/packages/aws-lambda/package.json @@ -69,6 +69,7 @@ "verify:zip-size": "tsx scripts/verify-zip-size.ts" }, "dependencies": { + "@aws-sdk/client-lambda": "^3.1146.0", "@aws-sdk/client-s3": "^3.1146.0", "@aws-sdk/client-sfn": "^3.1146.0", "@hyperframes/producer": "workspace:^", diff --git a/packages/aws-lambda/scripts/build-zip.ts b/packages/aws-lambda/scripts/build-zip.ts index 1a46bef044..2856c6fb09 100644 --- a/packages/aws-lambda/scripts/build-zip.ts +++ b/packages/aws-lambda/scripts/build-zip.ts @@ -228,6 +228,7 @@ async function bundleHandler(stagingDir: string): Promise { "puppeteer", // AWS SDK v3 is pre-installed in the Lambda Node 22 runtime; mark // external so we don't double-bundle 3+ MiB of SDK. + "@aws-sdk/client-lambda", "@aws-sdk/client-s3", ], plugins: [workspaceAliasPlugin], diff --git a/packages/aws-lambda/src/cdk/HyperframesRenderStack.snapshot.test.ts b/packages/aws-lambda/src/cdk/HyperframesRenderStack.snapshot.test.ts index b9cc767b39..e1493ae7b4 100644 --- a/packages/aws-lambda/src/cdk/HyperframesRenderStack.snapshot.test.ts +++ b/packages/aws-lambda/src/cdk/HyperframesRenderStack.snapshot.test.ts @@ -42,9 +42,9 @@ const EXPECTED_RESOURCE_COUNTS: Record = { "AWS::CloudWatch::Alarm": 3, "AWS::Logs::LogGroup": 1, // CDK emits IAM roles for both the function and the state machine, plus - // a managed policy for the bucket grant. + // a managed policy for the bucket grant and one for the Distributed Map children. "AWS::IAM::Role": 2, - "AWS::IAM::Policy": 2, + "AWS::IAM::Policy": 3, }; // Top-level state names emitted by ASL. The Map state's inner @@ -86,7 +86,7 @@ const EXPECTED_NON_RETRYABLE_ERRORS = new Set([ "ChromeBinaryUnavailableError", ]); -function doSynth(): { +function doSynth(reservedConcurrency?: number): { template: Template; definition: { States: Record; StartAt: string }; } { @@ -94,7 +94,10 @@ function doSynth(): { writeFileSync(join(zipDir, "handler.zip"), "fake zip bytes"); const app = new App(); const stack = new Stack(app, "TestStack"); - new HyperframesRenderStack(stack, "Render", { handlerZipPath: join(zipDir, "handler.zip") }); + new HyperframesRenderStack(stack, "Render", { + handlerZipPath: join(zipDir, "handler.zip"), + reservedConcurrency, + }); const template = Template.fromStack(stack); const stateMachine = Object.values( template.findResources("AWS::StepFunctions::StateMachine"), @@ -308,6 +311,126 @@ describe("HyperframesRenderStack — snapshot", () => { expect(sentPayload(tasks["Fan/1/RenderChunk"]).Format).toBe("webm"); }); + it("runs every chunk in a Distributed Map with STANDARD children in CDK and SAM", () => { + for (const definition of [SYNTHED.definition, readSamDefinition()]) { + for (const name of ["RenderChunks", "RenderChunksV2"]) { + const map = requireRecord(definition.States[name], `${name} state`); + const processor = requireRecordProperty(map, "ItemProcessor", `${name} processor`); + expect({ name, config: processor.ProcessorConfig }).toEqual({ + name, + config: { Mode: "DISTRIBUTED", ExecutionType: "STANDARD" }, + }); + expect(map.Label).toBe(name); + expect(map.MaxConcurrencyPath).toBe("$.Plan.MaxChunkConcurrency"); + // Any failed chunk must still fail the render. + expect(Object.keys(map).filter((key) => key.startsWith("ToleratedFailure"))).toEqual([]); + } + } + }); + + it("caps chunk fan-out at the concurrency the plan returns, in CDK and SAM", () => { + for (const definition of [SYNTHED.definition, readSamDefinition()]) { + for (const name of ["Plan", "PlanV2"]) { + const selector = requireRecordProperty( + requireRecord(definition.States[name], `${name} state`), + "ResultSelector", + `${name} result selector`, + ); + expect(selector["MaxChunkConcurrency.$"]).toBe("$.Payload.MaxChunkConcurrency"); + } + } + }); + + it("tells the plan the reserved concurrency and lets it read the account's, in CDK and SAM", () => { + const cdkFunction = (synthed: ReturnType) => + Object.values(synthed.template.findResources("AWS::Lambda::Function"))[0] as { + Properties: { Environment: { Variables: Record } }; + }; + const reservedEnv = (synthed: ReturnType) => + cdkFunction(synthed).Properties.Environment.Variables.HYPERFRAMES_LAMBDA_RESERVED_CONCURRENCY; + expect(reservedEnv(SYNTHED)).toBe("-1"); + expect(reservedEnv(doSynth(300))).toBe("300"); + const cdkActions = Object.values(SYNTHED.template.findResources("AWS::IAM::Policy")) + .flatMap(policyStatements) + .flatMap((statement) => statement.Action); + expect(cdkActions).toContain("lambda:GetAccountSettings"); + + const samFunction = requireRecordProperty( + requireRecordProperty(readSamResources(), "RenderFunction", "SAM RenderFunction"), + "Properties", + "SAM function properties", + ) as { Environment: { Variables: Record }; Policies: unknown[] }; + // The YAML parser drops the !Ref tag and keeps its argument. + expect(samFunction.Environment.Variables.HYPERFRAMES_LAMBDA_RESERVED_CONCURRENCY).toBe( + "ReservedConcurrency", + ); + const samActions = samFunction.Policies.flatMap((policy) => + ((policy as { Statement?: Record[] }).Statement ?? []).map((s) => s.Action), + ); + expect(samActions).toContain("lambda:GetAccountSettings"); + }, 30000); + + it("fails at once when a chunk cannot fit the worker timeout, in CDK and SAM", () => { + for (const definition of [SYNTHED.definition, readSamDefinition()]) { + for (const name of ["Plan", "PlanV2"]) { + const errors = new Set(); + collectNonRetryableErrors(definition.States[name], errors); + expect({ name, terminal: errors.has("CHUNK_EXCEEDS_WORKER_TIMEOUT") }).toEqual({ + name, + terminal: true, + }); + } + } + }); + + it("gives Lambda throttles a minutes-long backoff ahead of the catch-all, in CDK and SAM", () => { + for (const definition of [SYNTHED.definition, readSamDefinition()]) { + for (const [taskName, task] of Object.entries(lambdaTasks(definition.States))) { + expect({ taskName, ...throttleRetry(task) }).toEqual({ + taskName, + beforeCatchAll: true, + MaxAttempts: 10, + JitterStrategy: "FULL", + }); + } + } + }); + + it("lets the state machine role start, describe and stop its own child executions", () => { + const cdkActions = Object.values(SYNTHED.template.findResources("AWS::IAM::Policy")).flatMap( + (policy) => policyStatements(policy).flatMap((statement) => [statement.Action].flat()), + ); + const samRole = requireRecord(readSamResources().RenderStateMachineRole, "SAM role"); + const samPolicies = requireRecordProperty(samRole, "Properties", "SAM role properties") + .Policies as Record[]; + const samStatements = samPolicies.flatMap( + (policy) => + requireRecordProperty(policy, "PolicyDocument", "SAM policy document").Statement as Record< + string, + unknown + >[], + ); + const samActions = samStatements.flatMap((statement) => [statement.Action].flat()); + for (const action of [ + "states:StartExecution", + "states:DescribeExecution", + "states:StopExecution", + ]) { + expect({ + action, + cdk: cdkActions.includes(action), + sam: samActions.includes(action), + }).toEqual({ + action, + cdk: true, + sam: true, + }); + } + expect(samStatements.map((statement) => statement.Resource)).toContainEqual( + "arn:${AWS::Partition}:states:${AWS::Region}:${AWS::AccountId}:stateMachine:${ProjectName}-render", + ); + }); + it("materializes null audio in the synthesized CDK v2 assembly payload", () => { const state = requireRecord(SYNTHED.definition.States.AssembleV2, "AssembleV2 state"); const parameters = requireRecordProperty(state, "Parameters", "AssembleV2 parameters"); @@ -345,6 +468,19 @@ function collectNonRetryableErrors(state: unknown, out: Set): void { } } +/** Where a task's throttle retrier sits and how hard it retries. */ +function throttleRetry(task: unknown): Record { + const retriers = requireRecord(task, "task").Retry as Record[]; + const errors = retriers.map((retrier) => (retrier.ErrorEquals as string[]).join()); + const at = errors.indexOf("Lambda.TooManyRequestsException"); + const { MaxAttempts, JitterStrategy } = retriers[at] ?? {}; + return { + beforeCatchAll: at >= 0 && at < errors.indexOf("States.ALL"), + MaxAttempts, + JitterStrategy, + }; +} + function isRecord(value: unknown): value is Record { return typeof value === "object" && value !== null && !Array.isArray(value); } @@ -445,6 +581,12 @@ function getV1TaskStates(definition: { }; } +function policyStatements(policy: unknown): Record[] { + const properties = requireRecordProperty(requireRecord(policy, "policy"), "Properties", "props"); + const document = requireRecordProperty(properties, "PolicyDocument", "policy document"); + return document.Statement as Record[]; +} + function readSamResources(): Record { const source = readFileSync( new URL("../../../../examples/aws-lambda/template.yaml", import.meta.url), diff --git a/packages/aws-lambda/src/cdk/HyperframesRenderStack.ts b/packages/aws-lambda/src/cdk/HyperframesRenderStack.ts index 0f0cb49cde..5e6031a0c1 100644 --- a/packages/aws-lambda/src/cdk/HyperframesRenderStack.ts +++ b/packages/aws-lambda/src/cdk/HyperframesRenderStack.ts @@ -23,6 +23,7 @@ import { dirname, resolve } from "node:path"; import { fileURLToPath } from "node:url"; import { Duration, RemovalPolicy, Size } from "aws-cdk-lib"; import * as cloudwatch from "aws-cdk-lib/aws-cloudwatch"; +import * as iam from "aws-cdk-lib/aws-iam"; import * as lambda from "aws-cdk-lib/aws-lambda"; import * as logs from "aws-cdk-lib/aws-logs"; import * as s3 from "aws-cdk-lib/aws-s3"; @@ -111,6 +112,7 @@ export class HyperframesRenderStack extends Construct { TMPDIR: "/tmp", HYPERFRAMES_LAMBDA_CHROME_SOURCE: chromeSource, HYPERFRAMES_RENDER_BUCKET: this.bucket.bucketName, + HYPERFRAMES_LAMBDA_RESERVED_CONCURRENCY: String(props.reservedConcurrency ?? -1), }, }); @@ -119,6 +121,9 @@ export class HyperframesRenderStack extends Construct { // SAM's AWSLambdaBasicExecutionRole equivalent is included by the // default `new lambda.Function` execution role. this.bucket.grantReadWrite(this.renderFunction); + this.renderFunction.addToRolePolicy( + new iam.PolicyStatement({ actions: ["lambda:GetAccountSettings"], resources: ["*"] }), + ); const stateMachineLogGroup = new logs.LogGroup(this, "RenderStateMachineLogGroup", { logGroupName: `/aws/states/${projectName}-render`, @@ -211,6 +216,7 @@ export class HyperframesRenderStack extends Construct { "PlanV2IntegrityError", "PLAN_ARTIFACT_DIGEST_MISMATCH", "FORMAT_NOT_SUPPORTED_IN_DISTRIBUTED", + "CHUNK_EXCEEDS_WORKER_TIMEOUT", "ChromeBinaryUnavailableError", ]; const NON_RETRYABLE_CHUNK = [ @@ -243,6 +249,15 @@ export class HyperframesRenderStack extends Construct { "ChromeBinaryUnavailableError", ]; + const THROTTLED_RETRY: sfn.RetryProps = { + errors: ["Lambda.TooManyRequestsException"], + interval: Duration.seconds(5), + maxAttempts: 10, + backoffRate: 2, + maxDelay: Duration.seconds(60), + jitterStrategy: sfn.JitterType.FULL, + }; + const plan = new tasks.LambdaInvoke(this, "Plan", { lambdaFunction: this.renderFunction, payload: sfn.TaskInput.fromObject({ @@ -257,6 +272,7 @@ export class HyperframesRenderStack extends Construct { "PlanS3Uri.$": "$.Payload.PlanS3Uri", "PlanHash.$": "$.Payload.PlanHash", "ChunkCount.$": "$.Payload.ChunkCount", + "MaxChunkConcurrency.$": "$.Payload.MaxChunkConcurrency", "Format.$": "$.Payload.Format", "HasAudio.$": "$.Payload.HasAudio", "AudioS3Uri.$": "$.Payload.AudioS3Uri", @@ -267,6 +283,7 @@ export class HyperframesRenderStack extends Construct { errors: NON_RETRYABLE_PLAN, maxAttempts: 0, }); + plan.addRetry(THROTTLED_RETRY); plan.addRetry({ errors: ["States.ALL"], interval: Duration.seconds(2), @@ -290,12 +307,14 @@ export class HyperframesRenderStack extends Construct { "PlanV2ArtifactS3Prefix.$": "$.Payload.PlanV2ArtifactS3Prefix", "PlanHash.$": "$.Payload.PlanHash", "ChunkCount.$": "$.Payload.ChunkCount", + "MaxChunkConcurrency.$": "$.Payload.MaxChunkConcurrency", "Format.$": "$.Payload.Format", "HasAudio.$": "$.Payload.HasAudio", }, resultPath: "$.Plan", }); planV2.addRetry({ errors: NON_RETRYABLE_PLAN, maxAttempts: 0 }); + planV2.addRetry(THROTTLED_RETRY); planV2.addRetry({ errors: ["States.ALL"], interval: Duration.seconds(2), @@ -331,12 +350,15 @@ export class HyperframesRenderStack extends Construct { "ChunkS3Uri.$": "$.Payload.ChunkS3Uri", "ChunkIndex.$": "$.Payload.ChunkIndex", "Sha256.$": "$.Payload.Sha256", + "FramesEncoded.$": "$.Payload.FramesEncoded", + "DurationMs.$": "$.Payload.DurationMs", }, }); renderChunkTask.addRetry({ errors: NON_RETRYABLE_CHUNK, maxAttempts: 0, }); + renderChunkTask.addRetry(THROTTLED_RETRY); renderChunkTask.addRetry({ errors: ["States.ALL"], interval: Duration.seconds(2), @@ -345,7 +367,10 @@ export class HyperframesRenderStack extends Construct { maxDelay: Duration.seconds(60), }); - const renderChunks = new sfn.Map(this, "RenderChunks", { + const renderChunks = new sfn.DistributedMap(this, "RenderChunks", { + label: "RenderChunks", + // Standard-type children: express ones stop at 5 minutes and a chunk may run the Lambda's full 15. + mapExecutionType: sfn.StateMachineType.STANDARD, itemsPath: "$.Iterator.ChunkIndexes", itemSelector: { "ChunkIndex.$": "$$.Map.Item.Value", @@ -354,7 +379,8 @@ export class HyperframesRenderStack extends Construct { "ChunkOutputS3Prefix.$": "$.PlanOutputS3Prefix", "Format.$": "$.Plan.Format", }, - maxConcurrencyPath: "$.Plan.ChunkCount", + // Chunks past the plan's concurrency queue here instead of being throttled by Lambda. + maxConcurrencyPath: "$.Plan.MaxChunkConcurrency", resultPath: "$.Chunks", }); renderChunks.itemProcessor(renderChunkTask); @@ -381,6 +407,7 @@ export class HyperframesRenderStack extends Construct { errors: NON_RETRYABLE_ASSEMBLE, maxAttempts: 0, }); + assemble.addRetry(THROTTLED_RETRY); assemble.addRetry({ errors: ["States.ALL"], interval: Duration.seconds(2), @@ -405,9 +432,12 @@ export class HyperframesRenderStack extends Construct { "ChunkS3Uri.$": "$.Payload.ChunkS3Uri", "ChunkIndex.$": "$.Payload.ChunkIndex", "Sha256.$": "$.Payload.Sha256", + "FramesEncoded.$": "$.Payload.FramesEncoded", + "DurationMs.$": "$.Payload.DurationMs", }, }); renderChunkV2Task.addRetry({ errors: NON_RETRYABLE_CHUNK, maxAttempts: 0 }); + renderChunkV2Task.addRetry(THROTTLED_RETRY); renderChunkV2Task.addRetry({ errors: ["States.ALL"], interval: Duration.seconds(2), @@ -416,7 +446,9 @@ export class HyperframesRenderStack extends Construct { maxDelay: Duration.seconds(60), }); - const renderChunksV2 = new sfn.Map(this, "RenderChunksV2", { + const renderChunksV2 = new sfn.DistributedMap(this, "RenderChunksV2", { + label: "RenderChunksV2", + mapExecutionType: sfn.StateMachineType.STANDARD, itemsPath: "$.Iterator.ChunkIndexes", itemSelector: { "ChunkIndex.$": "$$.Map.Item.Value", @@ -426,7 +458,7 @@ export class HyperframesRenderStack extends Construct { "ChunkOutputS3Prefix.$": "$.PlanOutputS3Prefix", "Format.$": "$.Plan.Format", }, - maxConcurrencyPath: "$.Plan.ChunkCount", + maxConcurrencyPath: "$.Plan.MaxChunkConcurrency", resultPath: "$.Chunks", }); renderChunksV2.itemProcessor(renderChunkV2Task); @@ -452,6 +484,7 @@ export class HyperframesRenderStack extends Construct { resultPath: "$.Output", }); assembleV2.addRetry({ errors: NON_RETRYABLE_ASSEMBLE, maxAttempts: 0 }); + assembleV2.addRetry(THROTTLED_RETRY); assembleV2.addRetry({ errors: ["States.ALL"], interval: Duration.seconds(2), diff --git a/packages/aws-lambda/src/events.ts b/packages/aws-lambda/src/events.ts index c15b52d679..b2fb2ecc90 100644 --- a/packages/aws-lambda/src/events.ts +++ b/packages/aws-lambda/src/events.ts @@ -163,6 +163,7 @@ interface PlanLambdaResultBase { Action: "plan"; PlanHash: string; ChunkCount: number; + MaxChunkConcurrency: number; TotalFrames: number; Fps: SerializableDistributedRenderConfig["fps"]; Width: number; diff --git a/packages/aws-lambda/src/handler.test.ts b/packages/aws-lambda/src/handler.test.ts index c2c46f441c..86fd043434 100644 --- a/packages/aws-lambda/src/handler.test.ts +++ b/packages/aws-lambda/src/handler.test.ts @@ -26,11 +26,13 @@ import { PlanVideosMetadataError, type AssembleResult, type ChunkResult, + type DistributedRenderConfig, type PlanResult, type PlanV2ArtifactPublisher, type PlanV2Manifest, publishPlanV2FromExecutionPlan, } from "@hyperframes/producer/distributed"; +import { resolveMaxParallelChunks } from "../../producer/src/services/distributed/plan.js"; import { recomputePlanHashFromPlanDir } from "../../producer/src/services/render/stages/freezePlan.js"; import type { AssembleEvent, LambdaEvent, PlanEvent, RenderChunkEvent } from "./events.js"; import { handler, unwrapEvent } from "./handler.js"; @@ -322,6 +324,131 @@ describe("handler dispatch", () => { expect(seen).toEqual([900, 300]); }); + describe("chunk concurrency when the render does not set one", () => { + const THIRTY_MIN_FRAMES = 54_000; + const MP_1080P = (1920 * 1080) / 1_000_000; + + function fakeLambda(reply: () => Promise) { + const send = mock(reply); + return { send, client: { send } as unknown as import("@aws-sdk/client-lambda").LambdaClient }; + } + + async function planThirtyMinutes(lambda: { client: unknown }, workerConcurrency?: number) { + const tmpRoot = makeTmpRoot(); + const s3 = new FakeS3Client(); + s3.objects.set("s3://bucket/project.tar.gz", await makeMinimalProjectTar()); + const warnings: string[] = []; + const log = { warn: (m: string) => warnings.push(m), error() {}, info() {}, debug() {} }; + let planned: DistributedRenderConfig | undefined; + const result = await handler( + { + Action: "plan", + PlanProtocol: "v1", + ProjectS3Uri: "s3://bucket/project.tar.gz", + PlanOutputS3Prefix: "s3://bucket/renders/conc/", + Config: { fps: 30, width: 1920, height: 1080, format: "mp4", workerConcurrency }, + }, + { + s3: s3 as unknown as import("@aws-sdk/client-s3").S3Client, + lambda: lambda.client as import("@aws-sdk/client-lambda").LambdaClient, + primitives: { + plan: mock(async (_dir: string, config: DistributedRenderConfig, planDir: string) => { + planned = config; + mkdirSync(planDir, { recursive: true }); + writeFileSync(join(planDir, "plan.json"), "{}"); + return { + planDir, + planProtocol: CURRENT_PLAN_PROTOCOL, + planHash: "h", + // The real sizing rule, so the cap and the chunk count are tested together. + chunkCount: resolveMaxParallelChunks(config, THIRTY_MIN_FRAMES, MP_1080P, log), + totalFrames: THIRTY_MIN_FRAMES, + fps: 30 as const, + width: 1920, + height: 1080, + format: "mp4" as const, + ffmpegVersion: "6.0", + producerVersion: "0.0.0-test", + }; + }) as unknown as typeof import("@hyperframes/producer/distributed").plan, + renderChunk: mock(async () => { + throw new Error("unused"); + }) as unknown as typeof import("@hyperframes/producer/distributed").renderChunk, + assemble: mock(async () => { + throw new Error("unused"); + }) as unknown as typeof import("@hyperframes/producer/distributed").assemble, + }, + tmpRoot, + skipChromeResolution: true, + getRemainingTimeInMillis: () => 900_000, + }, + ); + if (result.Action !== "plan") throw new Error("unreachable"); + return { result, planned, warnings }; + } + + it("runs every chunk at once when the account has room", async () => { + const lambda = fakeLambda(async () => ({ + AccountLimit: { UnreservedConcurrentExecutions: 1000 }, + })); + const { result, planned, warnings } = await planThirtyMinutes(lambda); + expect(planned?.workerConcurrency).toBe(900); + expect(result.ChunkCount).toBe(204); + expect(result.MaxChunkConcurrency).toBe(result.ChunkCount); + expect(warnings).toEqual([]); + }); + + it("caps at 90% of a small account and says the render runs in waves", async () => { + const lambda = fakeLambda(async () => ({ + AccountLimit: { UnreservedConcurrentExecutions: 10 }, + })); + const { result, warnings } = await planThirtyMinutes(lambda); + expect(result.MaxChunkConcurrency).toBe(9); + expect(result.ChunkCount).toBe(21); + expect(warnings).toHaveLength(1); + expect(warnings[0]).toContain("in 3 waves"); + expect(warnings[0]).toContain("--concurrency 21"); + }); + + it("falls back to 16 at once and logs why when the account cannot be read", async () => { + const lambda = fakeLambda(async () => { + const denied = new Error("not authorized to perform: lambda:GetAccountSettings"); + denied.name = "AccessDeniedException"; + throw denied; + }); + const logged = spyOn(console, "log").mockImplementation(() => undefined); + try { + const { result, planned } = await planThirtyMinutes(lambda); + expect(planned?.workerConcurrency).toBe(16); + expect(result.MaxChunkConcurrency).toBe(16); + const lines = logged.mock.calls.map((call) => String(call[0])); + const fallback = lines.find((line) => line.includes('"source":"fallback"')); + expect(fallback).toContain('"level":"WARN"'); + expect(fallback).toContain("AccessDeniedException"); + expect(fallback).toContain("planning for 16 chunks at once"); + } finally { + logged.mockRestore(); + } + }); + + it("keeps an explicit concurrency, then the reserved one, without asking the account", async () => { + const lambda = fakeLambda(async () => ({ + AccountLimit: { UnreservedConcurrentExecutions: 1000 }, + })); + const explicit = await planThirtyMinutes(lambda, 64); + expect(explicit.planned?.workerConcurrency).toBe(64); + expect(explicit.result.MaxChunkConcurrency).toBe(64); + process.env.HYPERFRAMES_LAMBDA_RESERVED_CONCURRENCY = "32"; + try { + const reserved = await planThirtyMinutes(lambda); + expect(reserved.result.MaxChunkConcurrency).toBe(32); + } finally { + delete process.env.HYPERFRAMES_LAMBDA_RESERVED_CONCURRENCY; + } + expect(lambda.send).not.toHaveBeenCalled(); + }); + }); + it("normalizes producer workflow codes to Step Functions error names", async () => { for (const code of [ "PLAN_TOO_LARGE", diff --git a/packages/aws-lambda/src/handler.ts b/packages/aws-lambda/src/handler.ts index f3f6e4fecd..f0fa4831fe 100644 --- a/packages/aws-lambda/src/handler.ts +++ b/packages/aws-lambda/src/handler.ts @@ -14,6 +14,7 @@ import { existsSync, mkdirSync, mkdtempSync, readFileSync, rmSync, statSync } from "node:fs"; import { tmpdir } from "node:os"; import { basename, join } from "node:path"; +import { GetAccountSettingsCommand, LambdaClient } from "@aws-sdk/client-lambda"; import { S3Client } from "@aws-sdk/client-s3"; import { assemble, @@ -22,6 +23,7 @@ import { type ChunkRenderer, type ChunkResult, type DistributedRenderConfig, + DEFAULT_MAX_PARALLEL_CHUNKS, listPlanV2ArtifactsForTarget, materializePlanV2Target, plan, @@ -83,6 +85,8 @@ function getS3Client(): S3Client { */ export interface HandlerDeps { s3?: S3Client; + /** Reads the account's concurrency when the render does not set one. */ + lambda?: LambdaClient; primitives?: { plan: typeof plan; planV2WithPublisher?: typeof planV2WithPublisher; @@ -325,12 +329,54 @@ function primeRuntimeEnv(): void { // ── Plan ──────────────────────────────────────────────────────────────────── -function planConfig(event: PlanEvent, deps?: HandlerDeps): DistributedRenderConfig { - const remainingMs = deps?.getRemainingTimeInMillis?.(); - if (event.Config.workerTimeoutSeconds !== undefined || remainingMs === undefined) { - return { ...event.Config }; +// Share of the account's unreserved concurrency a render plans for. The rest stays free for other +// functions; at the default 1000 quota that is the 100 AWS keeps unreserved anyway. +const ACCOUNT_CONCURRENCY_SHARE = 0.9; + +async function resolveWorkerConcurrency(event: PlanEvent, deps?: HandlerDeps): Promise { + if (event.Config.workerConcurrency !== undefined) return event.Config.workerConcurrency; + try { + return await functionConcurrency(deps?.lambda); + } catch (err) { + logEvent({ + event: "worker_concurrency", + level: "WARN", + source: "fallback", + concurrency: DEFAULT_MAX_PARALLEL_CHUNKS, + message: + `Could not read the account's Lambda concurrency (${String(err)}); ` + + `planning for ${DEFAULT_MAX_PARALLEL_CHUNKS} chunks at once. Allow lambda:GetAccountSettings ` + + "on the function role, or set the concurrency explicitly.", + }); + return DEFAULT_MAX_PARALLEL_CHUNKS; } - return { ...event.Config, workerTimeoutSeconds: Math.ceil(remainingMs / 1000) }; +} + +/** The function's reserved concurrency, else its share of the account's unreserved pool. */ +async function functionConcurrency(lambda = new LambdaClient({})): Promise { + const reserved = Number(process.env.HYPERFRAMES_LAMBDA_RESERVED_CONCURRENCY); + if (reserved > 0) return reserved; + const { AccountLimit } = await lambda.send(new GetAccountSettingsCommand({})); + const unreserved = AccountLimit?.UnreservedConcurrentExecutions; + if (unreserved === undefined) throw new Error("no UnreservedConcurrentExecutions in the reply"); + const concurrency = Math.max(1, Math.floor(unreserved * ACCOUNT_CONCURRENCY_SHARE)); + logEvent({ event: "worker_concurrency", source: "account", unreserved, concurrency }); + return concurrency; +} + +async function planConfig( + event: PlanEvent, + deps?: HandlerDeps, +): Promise { + const config = { + ...event.Config, + workerConcurrency: await resolveWorkerConcurrency(event, deps), + }; + const remainingMs = deps?.getRemainingTimeInMillis?.(); + if (config.workerTimeoutSeconds === undefined && remainingMs !== undefined) { + config.workerTimeoutSeconds = Math.ceil(remainingMs / 1000); + } + return config; } // The v1 handler owns one transactional download, plan, archive, upload, and cleanup lifecycle. @@ -340,7 +386,7 @@ async function handlePlan(event: PlanEvent, deps?: HandlerDeps): Promise> { const started = Date.now(); - const config = planConfig(event, deps); + const config = await planConfig(event, deps); const s3 = deps?.s3 ?? getS3Client(); const primitive = deps?.primitives?.planV2WithPublisher ?? planV2WithPublisher; if (!deps?.skipChromeResolution && !process.env.PRODUCER_HEADLESS_SHELL_PATH) { @@ -455,6 +502,7 @@ async function handlePlanV2( PlanV2ArtifactS3Prefix: publisher.artifactPrefix, PlanHash: manifest.planHash, ChunkCount: manifest.chunkCount, + MaxChunkConcurrency: Math.min(config.workerConcurrency, manifest.chunkCount), TotalFrames: manifest.totalFrames, Fps: manifest.fps, Width: manifest.width, diff --git a/packages/aws-lambda/src/sdk/getRenderProgress.test.ts b/packages/aws-lambda/src/sdk/getRenderProgress.test.ts index 7f0982511c..3d2802ae39 100644 --- a/packages/aws-lambda/src/sdk/getRenderProgress.test.ts +++ b/packages/aws-lambda/src/sdk/getRenderProgress.test.ts @@ -5,7 +5,8 @@ import { type HistoryEvent, type SFNClient, } from "@aws-sdk/client-sfn"; -import { getRenderProgress } from "./getRenderProgress.js"; +import { computeRenderCost } from "./costAccounting.js"; +import { createProgressSfnClient, getRenderProgress } from "./getRenderProgress.js"; interface DescribeShape { status?: string; @@ -13,28 +14,65 @@ interface DescribeShape { stopDate?: Date; } +interface FakeInput { + executionArn?: string; + mapRunArn?: string; + statusFilter?: string; + nextToken?: string; +} + +interface FakeChild { + status: string; + history: HistoryEvent[]; +} + class FakeSFN { describe: DescribeShape = {}; // Pages of history events; FakeSFN walks them in order, paginating with `nextToken`. historyPages: HistoryEvent[][] = []; + // Distributed Map: map run ARN -> item counts and child executions (child ARN -> status + history). + mapRunCounts: Record> = {}; + mapRunChildren: Record> = {}; + calls: string[] = []; + // Command name -> error name the command throws, as the SDK names service errors. + failures: Record = {}; async send(command: unknown): Promise { - const cmdName = (command as { constructor: { name: string } }).constructor.name; - if (cmdName === "DescribeExecutionCommand") { - return { - status: this.describe.status ?? "RUNNING", - startDate: this.describe.startDate ?? new Date("2026-05-16T00:00:00Z"), - stopDate: this.describe.stopDate, - }; + const { constructor, input } = command as { constructor: { name: string }; input: FakeInput }; + this.calls.push(constructor.name); + const failure = this.failures[constructor.name]; + if (failure) + throw Object.assign(new Error(`${failure} from ${constructor.name}`), { name: failure }); + if (constructor.name === "DescribeExecutionCommand") return this.describeResult(); + if (constructor.name === "DescribeMapRunCommand") { + return { itemCounts: this.mapRunCounts[input.mapRunArn ?? ""] }; } - if (cmdName === "GetExecutionHistoryCommand") { - const input = (command as { input: { nextToken?: string } }).input; - const idx = input.nextToken ? Number.parseInt(input.nextToken, 10) : 0; - const page = this.historyPages[idx] ?? []; - const nextToken = idx + 1 < this.historyPages.length ? String(idx + 1) : undefined; - return { events: page, nextToken }; - } - throw new Error(`FakeSFN: unexpected command ${cmdName}`); + if (constructor.name === "ListExecutionsCommand") return this.listResult(input); + if (constructor.name === "GetExecutionHistoryCommand") return this.historyResult(input); + throw new Error(`FakeSFN: unexpected command ${constructor.name}`); } + private describeResult(): unknown { + return { + status: this.describe.status ?? "RUNNING", + startDate: this.describe.startDate ?? new Date("2026-05-16T00:00:00Z"), + stopDate: this.describe.stopDate, + }; + } + private listResult({ mapRunArn = "", statusFilter }: FakeInput): unknown { + const children = Object.entries(this.mapRunChildren[mapRunArn] ?? {}); + const matching = children.filter(([, c]) => !statusFilter || c.status === statusFilter); + return { executions: matching.map(([executionArn]) => ({ executionArn })) }; + } + private historyResult({ executionArn = "", nextToken }: FakeInput): unknown { + const child = Object.values(this.mapRunChildren) + .map((children) => children[executionArn]) + .find(Boolean); + return historyPage(child ? [child.history] : this.historyPages, Number(nextToken ?? 0)); + } +} + +function historyPage(pages: HistoryEvent[][], idx: number): unknown { + const nextToken = idx + 1 < pages.length ? String(idx + 1) : undefined; + return { events: pages[idx] ?? [], nextToken }; } function lambdaSucceeded(payload: unknown): HistoryEvent { @@ -291,6 +329,176 @@ describe("getRenderProgress", () => { expect(progress.overallProgress).toBe(1); }); + describe("Distributed Map chunks", () => { + const mapRunArn = "arn:aws:states:us-east-2:0:mapRun:hf-render/RenderChunksV2:run"; + const CHUNKS = 225; + + function parentUntilMapRun(totalFrames: number): HistoryEvent[] { + return [ + stateEntered("PlanV2"), + taskScheduled(), + taskSucceeded({ TotalFrames: totalFrames, DurationMs: 1000 }), + { + type: "MapStateEntered", + id: 1, + timestamp: new Date(), + stateEnteredEventDetails: { name: "RenderChunksV2" }, + } as HistoryEvent, + { + type: "MapRunStarted", + id: 1, + timestamp: new Date(), + mapRunStartedEventDetails: { mapRunArn }, + } as HistoryEvent, + ]; + } + + function chunkChild(status: string, last: HistoryEvent): FakeChild { + return { status, history: [stateEntered("RenderChunkV2"), taskScheduled(), last] }; + } + + // Children with real histories, so a per-child reader would find (and pay for) all of them. + function fakeWithChildren(statuses: string[]): FakeSFN { + const sfn = new FakeSFN(); + sfn.historyPages = [parentUntilMapRun(3600)]; + sfn.mapRunChildren[mapRunArn] = Object.fromEntries( + statuses.map((status, i) => [ + `child-${i}`, + status === "FAILED" + ? chunkChild(status, taskFailed("PLAN_HASH_MISMATCH", `chunk ${i} hash drift`)) + : chunkChild(status, taskSucceeded({ FramesEncoded: 16, DurationMs: 4000 })), + ]), + ); + return sfn; + } + + it("polls 225 running chunks with a constant number of calls", async () => { + const statuses = Array.from({ length: CHUNKS }, (_, i) => + i < 125 ? "SUCCEEDED" : "RUNNING", + ); + const sfn = fakeWithChildren(statuses); + sfn.mapRunCounts[mapRunArn] = { total: CHUNKS, pending: 0, running: 100, succeeded: 125 }; + const p = await getRenderProgress({ + executionArn: "arn:x", + sfn: sfn as unknown as SFNClient, + }); + expect(sfn.calls).toEqual([ + "DescribeExecutionCommand", + "GetExecutionHistoryCommand", + "DescribeMapRunCommand", + ]); + expect(p.framesRendered).toBe(Math.floor((3600 * 125) / CHUNKS)); + expect(p.overallProgress).toBeCloseTo(0.1 + 0.8 * (125 / CHUNKS), 5); + expect(p.lambdasInvoked).toBe(1 + CHUNKS); + expect(p.errors).toEqual([]); + }); + + // Callers on the policy from before the Distributed Map lack states:DescribeMapRun. + it("falls back to the parent snapshot with a hint when DescribeMapRun is denied", async () => { + const sfn = fakeWithChildren(["SUCCEEDED", "RUNNING"]); + sfn.failures.DescribeMapRunCommand = "AccessDeniedException"; + const p = await getRenderProgress({ + executionArn: "arn:x", + sfn: sfn as unknown as SFNClient, + }); + expect(p.status).toBe("RUNNING"); + expect(p.totalFrames).toBe(3600); + expect(p.overallProgress).toBeCloseTo(0.1, 5); + expect(p.warnings).toEqual([expect.stringContaining("hyperframes lambda policies user")]); + }); + + it("still throws other map-run read errors", async () => { + const sfn = fakeWithChildren(["RUNNING"]); + sfn.failures.DescribeMapRunCommand = "InternalServerError"; + await expect( + getRenderProgress({ executionArn: "arn:x", sfn: sfn as unknown as SFNClient }), + ).rejects.toThrow("InternalServerError"); + }); + + it("reads exact chunk frames and cost from the Map output once it exits", async () => { + const sfn = new FakeSFN(); + const chunks = Array.from({ length: CHUNKS }, (_, i) => ({ + ChunkIndex: i, + FramesEncoded: 16, + DurationMs: 4000, + })); + sfn.historyPages = [ + [ + ...parentUntilMapRun(3600), + { + type: "MapStateExited", + id: 1, + timestamp: new Date(), + stateExitedEventDetails: { + name: "RenderChunksV2", + output: JSON.stringify({ Chunks: chunks }), + }, + } as HistoryEvent, + ], + ]; + const p = await getRenderProgress({ + executionArn: "arn:x", + sfn: sfn as unknown as SFNClient, + }); + expect(sfn.calls).toEqual(["DescribeExecutionCommand", "GetExecutionHistoryCommand"]); + expect(p.framesRendered).toBe(3600); + expect(p.lambdasInvoked).toBe(1 + CHUNKS); + const billed = [1000, ...chunks.map((c) => c.DurationMs)].map((billedDurationMs) => ({ + billedDurationMs, + memorySizeMb: 10240, + estimated: false, + })); + expect(p.costs.breakdown.lambdaUsd).toBe(computeRenderCost(billed, 0).breakdown.lambdaUsd); + }); + + it("surfaces a failed chunk's own error, reading only failed children", async () => { + const statuses = Array.from({ length: CHUNKS }, (_, i) => + i === 7 ? "FAILED" : i < 100 ? "SUCCEEDED" : "ABORTED", + ); + const sfn = fakeWithChildren(statuses); + sfn.mapRunCounts[mapRunArn] = { total: CHUNKS, succeeded: 99, failed: 1, aborted: 125 }; + const p = await getRenderProgress({ + executionArn: "arn:x", + sfn: sfn as unknown as SFNClient, + }); + expect(p.errors).toEqual([ + { state: "RenderChunkV2", error: "PLAN_HASH_MISMATCH", cause: "chunk 7 hash drift" }, + ]); + expect(sfn.calls).toHaveLength(5); + }); + }); + + it("backs off on ThrottlingException and still returns progress", async () => { + let throttlesLeft = 3; + const targets: string[] = []; + const json = (statusCode: number, body: unknown) => ({ + response: { + statusCode, + headers: { "content-type": "application/x-amz-json-1.0" }, + body: new TextEncoder().encode(JSON.stringify(body)), + }, + }); + const sfn = createProgressSfnClient({ + region: "us-east-2", + credentials: { accessKeyId: "a", secretAccessKey: "b" }, + requestHandler: { + async handle(req: { headers: Record }) { + targets.push(req.headers["x-amz-target"] ?? ""); + if (throttlesLeft-- > 0) { + return json(400, { __type: "ThrottlingException", message: "Rate exceeded" }); + } + if (req.headers["x-amz-target"]?.endsWith("DescribeExecution")) { + return json(200, { status: "RUNNING", startDate: 1_760_000_000 }); + } + return json(200, { events: [] }); + }, + }, + }); + const p = await getRenderProgress({ executionArn: "arn:x", sfn }); + expect(p.status).toBe("RUNNING"); + expect(targets).toHaveLength(5); + }, 20_000); + it("requires executionArn", async () => { await expect(getRenderProgress({ executionArn: "" })).rejects.toThrow(/executionArn/); }); diff --git a/packages/aws-lambda/src/sdk/getRenderProgress.ts b/packages/aws-lambda/src/sdk/getRenderProgress.ts index 6d17736ab3..1029766276 100644 --- a/packages/aws-lambda/src/sdk/getRenderProgress.ts +++ b/packages/aws-lambda/src/sdk/getRenderProgress.ts @@ -2,10 +2,10 @@ * `getRenderProgress` — read-only progress + cost snapshot for a single * render started by {@link renderToLambda}. * - * Pulls one `DescribeExecution` + one `GetExecutionHistory` per call. The - * history is paginated server-side; the helper loops until exhausted so a - * 1,000-event Step Functions execution still produces a single - * `RenderProgress` snapshot. + * Pulls one `DescribeExecution` + one `GetExecutionHistory` per call. While + * the chunk Distributed Map runs, one `DescribeMapRun` adds its item counts; + * once it exits, chunk frames and cost come from the Map's output. Failed + * chunks' histories are read only when the counts show a failure. * * Progress math: * - 0 before Plan completes (no frame count is known yet) @@ -22,9 +22,12 @@ import { DescribeExecutionCommand, + DescribeMapRunCommand, GetExecutionHistoryCommand, type HistoryEvent, + ListExecutionsCommand, SFNClient, + type SFNClientConfig, } from "@aws-sdk/client-sfn"; import { type BilledLambdaInvocation, @@ -81,18 +84,25 @@ export interface RenderProgress { errors: RenderError[]; /** `true` once the execution has terminated in a non-`SUCCEEDED` state. */ fatalErrorEncountered: boolean; + /** Problems reading progress that leave the render itself untouched, such as a missing IAM permission. */ + warnings: string[]; startedAt: string; endedAt: string | null; } const DEFAULT_MEMORY_MB = 10240; +/** Rides out `ThrottlingException` with the SDK's backoff instead of throwing after the default 3 tries. */ +export function createProgressSfnClient(config: SFNClientConfig = {}): SFNClient { + return new SFNClient({ maxAttempts: 8, ...config }); +} + /** Pull a current progress snapshot for one render. */ export async function getRenderProgress(opts: GetRenderProgressOptions): Promise { if (!opts.executionArn) { throw new Error("[getRenderProgress] executionArn is required"); } - const sfn = opts.sfn ?? new SFNClient({ region: opts.region }); + const sfn = opts.sfn ?? createProgressSfnClient({ region: opts.region }); const memoryMb = opts.defaultMemorySizeMb ?? DEFAULT_MEMORY_MB; const describe = await sfn.send( @@ -104,6 +114,15 @@ export async function getRenderProgress(opts: GetRenderProgressOptions): Promise const history = await loadFullHistory(sfn, opts.executionArn); const summary = summarizeHistory(history, memoryMb); + const mapRun = findChunkMapRun(history); + const warnings: string[] = []; + if (mapRun?.exited) addChunkResults(summary, mapRun.output, memoryMb); + else if (mapRun) { + await addRunningMapRun(sfn, mapRun.arn, summary, memoryMb).catch((err: unknown) => { + if ((err as { name?: unknown } | null)?.name !== "AccessDeniedException") throw err; + warnings.push(MAP_RUN_ACCESS_DENIED); + }); + } const costs = computeRenderCost(summary.lambdaInvocations, summary.stateTransitions); const overallProgress = computeOverallProgress({ @@ -123,6 +142,7 @@ export async function getRenderProgress(opts: GetRenderProgressOptions): Promise outputFile: summary.outputFile, errors: summary.errors, fatalErrorEncountered: isTerminalFailure(status), + warnings, startedAt, endedAt, }; @@ -147,6 +167,88 @@ async function loadFullHistory(sfn: SFNClient, executionArn: string): Promise