From dee8380d2295070353aac1f98deb28c030d86572 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Miguel=20=C3=81ngel?= Date: Wed, 21 Oct 2026 05:26:58 +0000 Subject: [PATCH] feat(aws-lambda): a cloud render mixes its audio beside its chunks (#5997) * feat(aws-lambda): a cloud render mixes its audio beside its chunks * fix(aws-lambda): the audio task keeps a worker free and runs only when the plan left audio out * test(aws-lambda): the audio task's free worker holds for a reserved pool of 1 and 2 * refactor(aws-lambda): one helper points every handler at the bundled Chrome --- examples/aws-lambda/template.yaml | 220 ++++++++++++------ .../HyperframesRenderStack.snapshot.test.ts | 77 ++++-- .../src/cdk/HyperframesRenderStack.ts | 49 +++- packages/aws-lambda/src/events.ts | 24 +- packages/aws-lambda/src/handler.test.ts | 157 ++++++++++++- packages/aws-lambda/src/handler.ts | 142 +++++++---- packages/aws-lambda/src/s3Stitch.test.ts | 24 +- packages/aws-lambda/src/s3Stitch.ts | 14 +- packages/producer/src/distributed.ts | 1 + .../src/services/distributed/assemble.ts | 6 +- 10 files changed, 558 insertions(+), 156 deletions(-) diff --git a/examples/aws-lambda/template.yaml b/examples/aws-lambda/template.yaml index bc8e38cd3f..3fb682404d 100644 --- a/examples/aws-lambda/template.yaml +++ b/examples/aws-lambda/template.yaml @@ -324,6 +324,7 @@ Resources: MaxChunkConcurrency.$: "$.Payload.MaxChunkConcurrency" Format.$: "$.Payload.Format" HasAudio.$: "$.Payload.HasAudio" + SeparateAudio.$: "$.Payload.SeparateAudio" ResultPath: $.Plan Retry: - ErrorEquals: @@ -395,7 +396,7 @@ Resources: Choices: - Variable: $.Plan.PlanProtocol StringEquals: v2 - Next: RenderChunksV2 + Next: RenderV2 Default: RenderChunks RenderChunks: @@ -511,74 +512,151 @@ Resources: MaxDelaySeconds: 60 End: true - RenderChunksV2: - Type: Map - Label: RenderChunksV2 - ItemsPath: $.Iterator.ChunkIndexes - ItemSelector: - ChunkIndex.$: "$$.Map.Item.Value" - PlanV2ManifestS3Uri.$: "$.Plan.PlanV2ManifestS3Uri" - PlanV2ArtifactS3Prefix.$: "$.Plan.PlanV2ArtifactS3Prefix" - PlanHash.$: "$.Plan.PlanHash" - ChunkOutputS3Prefix.$: "$.PlanOutputS3Prefix" - Format.$: "$.Plan.Format" - # Chunks past the plan's concurrency queue here instead of being throttled by Lambda. - MaxConcurrencyPath: $.Plan.MaxChunkConcurrency - ResultPath: $.Chunks - ItemProcessor: - ProcessorConfig: - Mode: DISTRIBUTED - ExecutionType: STANDARD - StartAt: RenderChunkV2 - States: - RenderChunkV2: - Type: Task - Resource: arn:aws:states:::lambda:invoke - Parameters: - FunctionName: !GetAtt RenderFunction.Arn - Payload: - Action: renderChunk - PlanProtocol: v2 - ChunkIndex.$: "$.ChunkIndex" - PlanV2ManifestS3Uri.$: "$.PlanV2ManifestS3Uri" - PlanV2ArtifactS3Prefix.$: "$.PlanV2ArtifactS3Prefix" - PlanHash.$: "$.PlanHash" - ChunkOutputS3Prefix.$: "$.ChunkOutputS3Prefix" - Format.$: "$.Format" - ResultSelector: - ChunkS3Uri.$: "$.Payload.ChunkS3Uri" - ChunkIndex.$: "$.Payload.ChunkIndex" - Sha256.$: "$.Payload.Sha256" - FramesEncoded.$: "$.Payload.FramesEncoded" - DurationMs.$: "$.Payload.DurationMs" - Retry: - - ErrorEquals: - - FFMPEG_VERSION_MISMATCH - - PLAN_HASH_MISMATCH - - S3_URI_NOT_ALLOWED - - BROWSER_GPU_NOT_SOFTWARE - - PLAN_TOO_LARGE - - PlanTooLargeError - - PLAN_PROTOCOL_UNSUPPORTED - - PlanProtocolUnsupportedError - - PLAN_V2_INTEGRITY_UNRECOVERABLE - - INVALID_VIDEO_METADATA - - PlanV2IntegrityError - - 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 - BackoffRate: 2 - MaxDelaySeconds: 60 - End: true + RenderV2: + # Chunks render video only while the audio task mixes the track AssembleV2 stitches in. + Type: Parallel + Branches: + - StartAt: RenderChunksV2 + States: + RenderChunksV2: + Type: Map + Label: RenderChunksV2 + ItemsPath: $.Iterator.ChunkIndexes + ItemSelector: + ChunkIndex.$: "$$.Map.Item.Value" + PlanV2ManifestS3Uri.$: "$.Plan.PlanV2ManifestS3Uri" + PlanV2ArtifactS3Prefix.$: "$.Plan.PlanV2ArtifactS3Prefix" + PlanHash.$: "$.Plan.PlanHash" + ChunkOutputS3Prefix.$: "$.PlanOutputS3Prefix" + Format.$: "$.Plan.Format" + # Chunks past the plan's concurrency queue here instead of being throttled by Lambda. + MaxConcurrencyPath: $.Plan.MaxChunkConcurrency + ResultPath: $.Chunks + ItemProcessor: + ProcessorConfig: + Mode: DISTRIBUTED + ExecutionType: STANDARD + StartAt: RenderChunkV2 + States: + RenderChunkV2: + Type: Task + Resource: arn:aws:states:::lambda:invoke + Parameters: + FunctionName: !GetAtt RenderFunction.Arn + Payload: + Action: renderChunk + PlanProtocol: v2 + ChunkIndex.$: "$.ChunkIndex" + PlanV2ManifestS3Uri.$: "$.PlanV2ManifestS3Uri" + PlanV2ArtifactS3Prefix.$: "$.PlanV2ArtifactS3Prefix" + PlanHash.$: "$.PlanHash" + ChunkOutputS3Prefix.$: "$.ChunkOutputS3Prefix" + Format.$: "$.Format" + ResultSelector: + ChunkS3Uri.$: "$.Payload.ChunkS3Uri" + ChunkIndex.$: "$.Payload.ChunkIndex" + Sha256.$: "$.Payload.Sha256" + FramesEncoded.$: "$.Payload.FramesEncoded" + DurationMs.$: "$.Payload.DurationMs" + Retry: + - ErrorEquals: + - FFMPEG_VERSION_MISMATCH + - PLAN_HASH_MISMATCH + - S3_URI_NOT_ALLOWED + - BROWSER_GPU_NOT_SOFTWARE + - PLAN_TOO_LARGE + - PlanTooLargeError + - PLAN_PROTOCOL_UNSUPPORTED + - PlanProtocolUnsupportedError + - PLAN_V2_INTEGRITY_UNRECOVERABLE + - INVALID_VIDEO_METADATA + - PlanV2IntegrityError + - 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 + BackoffRate: 2 + MaxDelaySeconds: 60 + End: true + End: true + - StartAt: AudioNeeded + States: + AudioNeeded: + Type: Choice + Choices: + - Variable: $.Plan.SeparateAudio + BooleanEquals: true + Next: Audio + Default: NoAudio + NoAudio: + Type: Pass + Result: + HasAudio: false + AudioS3Uri: null + ResultPath: $.Audio + End: true + Audio: + # The plan left its audio out: mix it here while the chunks render. + Type: Task + Resource: arn:aws:states:::lambda:invoke + Parameters: + FunctionName: !GetAtt RenderFunction.Arn + Payload: + Action: audio + ProjectS3Uri.$: "$.ProjectS3Uri" + PlanOutputS3Prefix.$: "$.PlanOutputS3Prefix" + Config.$: "$.Config" + ResultSelector: + HasAudio.$: "$.Payload.HasAudio" + AudioS3Uri.$: "$.Payload.AudioS3Uri" + ResultPath: $.Audio + Retry: + - ErrorEquals: + - FFMPEG_VERSION_MISMATCH + - PLAN_HASH_MISMATCH + - S3_URI_NOT_ALLOWED + - BROWSER_GPU_NOT_SOFTWARE + - FONT_FETCH_FAILED + - PLAN_TOO_LARGE + - PlanTooLargeError + - PLAN_PROTOCOL_UNSUPPORTED + - PlanProtocolUnsupportedError + - PLAN_V2_INTEGRITY_UNRECOVERABLE + - VIDEO_SOURCE_UNRENDERABLE + - INVALID_VIDEO_METADATA + - NOT_MEDIA_PAYLOAD + - NotMediaPayloadError + - PlanV2IntegrityError + - PLAN_ARTIFACT_DIGEST_MISMATCH + - FORMAT_NOT_SUPPORTED_IN_DISTRIBUTED + - CHUNK_EXCEEDS_WORKER_TIMEOUT + - ChromeBinaryUnavailableError + - SITE_MANIFEST_INVALID + MaxAttempts: 0 + - ErrorEquals: [Lambda.TooManyRequestsException] + IntervalSeconds: 5 + MaxAttempts: 10 + BackoffRate: 2 + MaxDelaySeconds: 60 + JitterStrategy: FULL + - ErrorEquals: [States.ALL] + IntervalSeconds: 2 + MaxAttempts: 4 + BackoffRate: 2 + MaxDelaySeconds: 60 + End: true + ResultSelector: + Chunks.$: "$[0].Chunks" + Audio.$: "$[1].Audio" + ResultPath: $.Render Next: AssembleV2 AssembleV2: @@ -592,8 +670,8 @@ Resources: PlanV2ManifestS3Uri.$: "$.Plan.PlanV2ManifestS3Uri" PlanV2ArtifactS3Prefix.$: "$.Plan.PlanV2ArtifactS3Prefix" PlanHash.$: "$.Plan.PlanHash" - ChunkS3Uris.$: "$.Chunks[*].ChunkS3Uri" - AudioS3Uri: null + ChunkS3Uris.$: "$.Render.Chunks[*].ChunkS3Uri" + AudioS3Uri.$: "$.Render.Audio.AudioS3Uri" OutputS3Uri.$: "$.OutputS3Uri" Format.$: "$.Plan.Format" ResultSelector: diff --git a/packages/aws-lambda/src/cdk/HyperframesRenderStack.snapshot.test.ts b/packages/aws-lambda/src/cdk/HyperframesRenderStack.snapshot.test.ts index 8179f9c0cf..0f22be4089 100644 --- a/packages/aws-lambda/src/cdk/HyperframesRenderStack.snapshot.test.ts +++ b/packages/aws-lambda/src/cdk/HyperframesRenderStack.snapshot.test.ts @@ -64,7 +64,7 @@ const EXPECTED_STATE_NAMES = [ "AssertChunkCount", "SelectWorkerProtocol", "RenderChunks", - "RenderChunksV2", + "RenderV2", "Assemble", "AssembleV2", "PlanProducedZeroChunks", @@ -190,15 +190,7 @@ describe("HyperframesRenderStack — snapshot", () => { | undefined; const innerStates = renderChunks?.Iterator?.States ?? renderChunks?.ItemProcessor?.States ?? {}; collectNonRetryableErrors(innerStates.RenderChunk, collected); - const renderChunksV2 = definition.States.RenderChunksV2 as - | { - Iterator?: { States?: Record }; - ItemProcessor?: { States?: Record }; - } - | undefined; - const innerStatesV2 = - renderChunksV2?.Iterator?.States ?? renderChunksV2?.ItemProcessor?.States ?? {}; - collectNonRetryableErrors(innerStatesV2.RenderChunkV2, collected); + collectNonRetryableErrors(getV2TaskStates(definition).RenderChunkV2, collected); for (const expected of EXPECTED_NON_RETRYABLE_ERRORS) { expect({ error: expected, present: collected.has(expected) }).toEqual({ @@ -234,7 +226,7 @@ describe("HyperframesRenderStack — snapshot", () => { const cdkTasks = getV2TaskStates(SYNTHED.definition); const samTasks = getV2TaskStates(readSamDefinition()); - for (const taskName of ["PlanV2", "RenderChunkV2", "AssembleV2"] as const) { + for (const taskName of ["PlanV2", "Audio", "RenderChunkV2", "AssembleV2"] as const) { const cdkErrors = new Set(); const samErrors = new Set(); collectNonRetryableErrors(cdkTasks[taskName], cdkErrors); @@ -265,6 +257,40 @@ describe("HyperframesRenderStack — snapshot", () => { } }); + it("renders v2 chunks beside the audio task and stitches both into AssembleV2, in CDK and SAM", () => { + for (const definition of [SYNTHED.definition, readSamDefinition()]) { + const renderV2 = requireRecord(definition.States.RenderV2, "RenderV2 state"); + expect([renderV2.Type, renderV2.Next, renderV2.ResultPath]).toEqual([ + "Parallel", + "AssembleV2", + "$.Render", + ]); + expect(renderV2.ResultSelector).toEqual({ + "Chunks.$": "$[0].Chunks", + "Audio.$": "$[1].Audio", + }); + expect(sentPayload(renderV2Branch(definition, "Audio"))).toMatchObject({ Action: "audio" }); + // Only a plan that left its audio out runs the task; any other skips its Lambda. + expect(renderV2Branch(definition, "AudioNeeded")).toMatchObject({ + Type: "Choice", + Choices: [{ Variable: "$.Plan.SeparateAudio", BooleanEquals: true, Next: "Audio" }], + Default: "NoAudio", + }); + expect(renderV2Branch(definition, "NoAudio")).toMatchObject({ + Type: "Pass", + Result: { HasAudio: false, AudioS3Uri: null }, + ResultPath: "$.Audio", + End: true, + }); + expect(requireRecord(definition.States.PlanV2, "PlanV2").ResultSelector).toMatchObject({ + "SeparateAudio.$": "$.Payload.SeparateAudio", + }); + expect(sentPayload(definition.States.AssembleV2)).toMatchObject({ + "ChunkS3Uris.$": "$.Render.Chunks[*].ChunkS3Uri", + }); + } + }); + it("keeps v1 and v2 locators disjoint across orchestration branches", () => { const { definition } = SYNTHED; const v1 = JSON.stringify({ @@ -274,7 +300,7 @@ describe("HyperframesRenderStack — snapshot", () => { }); const v2 = JSON.stringify({ plan: definition.States.PlanV2, - chunks: definition.States.RenderChunksV2, + chunks: renderV2Branch(definition, "RenderChunksV2"), assemble: definition.States.AssembleV2, }); expect(v1).toContain("PlanS3Uri"); @@ -351,7 +377,9 @@ describe("HyperframesRenderStack — snapshot", () => { 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 state = + name === "RenderChunksV2" ? renderV2Branch(definition, name) : definition.States[name]; + const map = requireRecord(state, `${name} state`); const processor = requireRecordProperty(map, "ItemProcessor", `${name} processor`); expect({ name, config: processor.ProcessorConfig }).toEqual({ name, @@ -468,23 +496,23 @@ describe("HyperframesRenderStack — snapshot", () => { ); }); - it("materializes null audio in the synthesized CDK v2 assembly payload", () => { + it("takes the audio task's track in the synthesized CDK v2 assembly payload", () => { const state = requireRecord(SYNTHED.definition.States.AssembleV2, "AssembleV2 state"); const parameters = requireRecordProperty(state, "Parameters", "AssembleV2 parameters"); const payload = requireRecordProperty(parameters, "Payload", "AssembleV2 payload"); - expect(payload["AudioS3Uri.$"]).toBe("States.StringToJson('null')"); + expect(payload["AudioS3Uri.$"]).toBe("$.Render.Audio.AudioS3Uri"); expect(Object.hasOwn(payload, "AudioS3Uri")).toBe(false); expect(payload.PlanProtocol).toBe("v2"); expect(payload["PlanV2ManifestS3Uri.$"]).toBe("$.Plan.PlanV2ManifestS3Uri"); }); - it("keeps SAM v2 assembly audio null and both v1 audio result paths", () => { + it("takes SAM v2 assembly audio from the audio task and keeps both v1 audio result paths", () => { const sam = readSamDefinition(); const assemble = requireRecord(sam.States.AssembleV2, "SAM AssembleV2 state"); const parameters = requireRecordProperty(assemble, "Parameters", "SAM AssembleV2 parameters"); const payload = requireRecordProperty(parameters, "Payload", "SAM AssembleV2 payload"); - expect(payload.AudioS3Uri).toBeNull(); + expect(payload["AudioS3Uri.$"]).toBe("$.Render.Audio.AudioS3Uri"); for (const definition of [SYNTHED.definition, sam]) { const v1 = requireRecord(definition.States.Assemble, "Assemble state"); @@ -588,16 +616,27 @@ function sentPayload(state: unknown): Record { ); } +/** The state named `name` in one of the v2 Parallel's branches. */ +function renderV2Branch(definition: { States: Record }, name: string): unknown { + const renderV2 = requireRecord(definition.States.RenderV2, "RenderV2 state"); + const branches: unknown[] = Array.isArray(renderV2.Branches) ? renderV2.Branches : []; + const found = branches + .map((branch) => requireRecordProperty(requireRecord(branch, "branch"), "States", "states")) + .find((states) => name in states); + return requireRecord(found?.[name], `RenderV2 branch state ${name}`); +} + function getV2TaskStates(definition: { States: Record; -}): Record<"PlanV2" | "RenderChunkV2" | "AssembleV2", unknown> { - const renderChunksV2 = requireRecord(definition.States.RenderChunksV2, "RenderChunksV2 state"); +}): Record<"PlanV2" | "Audio" | "RenderChunkV2" | "AssembleV2", unknown> { + const renderChunksV2 = requireRecord(renderV2Branch(definition, "RenderChunksV2"), "map"); const processor = isRecord(renderChunksV2.Iterator) ? renderChunksV2.Iterator : requireRecord(renderChunksV2.ItemProcessor, "RenderChunksV2 processor"); const innerStates = requireRecord(processor.States, "RenderChunksV2 processor states"); return { PlanV2: definition.States.PlanV2, + Audio: renderV2Branch(definition, "Audio"), RenderChunkV2: innerStates.RenderChunkV2, AssembleV2: definition.States.AssembleV2, }; diff --git a/packages/aws-lambda/src/cdk/HyperframesRenderStack.ts b/packages/aws-lambda/src/cdk/HyperframesRenderStack.ts index 08038d1432..864ab61b43 100644 --- a/packages/aws-lambda/src/cdk/HyperframesRenderStack.ts +++ b/packages/aws-lambda/src/cdk/HyperframesRenderStack.ts @@ -318,6 +318,7 @@ export class HyperframesRenderStack extends Construct { "MaxChunkConcurrency.$": "$.Payload.MaxChunkConcurrency", "Format.$": "$.Payload.Format", "HasAudio.$": "$.Payload.HasAudio", + "SeparateAudio.$": "$.Payload.SeparateAudio", }, resultPath: "$.Plan", }); @@ -479,8 +480,8 @@ export class HyperframesRenderStack extends Construct { "PlanV2ManifestS3Uri.$": "$.Plan.PlanV2ManifestS3Uri", "PlanV2ArtifactS3Prefix.$": "$.Plan.PlanV2ArtifactS3Prefix", "PlanHash.$": "$.Plan.PlanHash", - "ChunkS3Uris.$": "$.Chunks[*].ChunkS3Uri", - AudioS3Uri: sfn.JsonPath.stringToJson("null"), + "ChunkS3Uris.$": "$.Render.Chunks[*].ChunkS3Uri", + "AudioS3Uri.$": "$.Render.Audio.AudioS3Uri", "OutputS3Uri.$": "$.OutputS3Uri", "Format.$": "$.Plan.Format", }), @@ -501,11 +502,47 @@ export class HyperframesRenderStack extends Construct { maxDelay: Duration.seconds(60), }); + // The plan left its audio out: mix it while the chunks render. + const audio = new tasks.LambdaInvoke(this, "Audio", { + lambdaFunction: this.renderFunction, + payload: sfn.TaskInput.fromObject({ + Action: "audio", + "ProjectS3Uri.$": "$.ProjectS3Uri", + "PlanOutputS3Prefix.$": "$.PlanOutputS3Prefix", + "Config.$": "$.Config", + }), + resultSelector: { + "HasAudio.$": "$.Payload.HasAudio", + "AudioS3Uri.$": "$.Payload.AudioS3Uri", + }, + resultPath: "$.Audio", + }); + audio.addRetry({ errors: NON_RETRYABLE_PLAN, maxAttempts: 0 }); + audio.addRetry(THROTTLED_RETRY); + audio.addRetry({ + errors: ["States.ALL"], + interval: Duration.seconds(2), + maxAttempts: 4, + backoffRate: 2, + maxDelay: Duration.seconds(60), + }); + const audioNeeded = new sfn.Choice(this, "AudioNeeded") + .when(sfn.Condition.booleanEquals("$.Plan.SeparateAudio", true), audio) + .otherwise( + new sfn.Pass(this, "NoAudio", { + result: sfn.Result.fromObject({ HasAudio: false, AudioS3Uri: null }), + resultPath: "$.Audio", + }), + ); + const renderV2 = new sfn.Parallel(this, "RenderV2", { + resultSelector: { "Chunks.$": "$[0].Chunks", "Audio.$": "$[1].Audio" }, + resultPath: "$.Render", + }) + .branch(renderChunksV2) + .branch(audioNeeded); + const selectWorkerProtocol = new sfn.Choice(this, "SelectWorkerProtocol") - .when( - sfn.Condition.stringEquals("$.Plan.PlanProtocol", "v2"), - renderChunksV2.next(assembleV2), - ) + .when(sfn.Condition.stringEquals("$.Plan.PlanProtocol", "v2"), renderV2.next(assembleV2)) .otherwise(renderChunks.next(assemble)); const assertChunkCount = new sfn.Choice(this, "AssertChunkCount") .when(sfn.Condition.numberGreaterThan("$.Plan.ChunkCount", 0), selectWorkerProtocol) diff --git a/packages/aws-lambda/src/events.ts b/packages/aws-lambda/src/events.ts index b2fb2ecc90..384cf7a0c5 100644 --- a/packages/aws-lambda/src/events.ts +++ b/packages/aws-lambda/src/events.ts @@ -24,7 +24,7 @@ import type { export type { SerializableDistributedRenderConfig } from "@hyperframes/producer/distributed"; /** Discriminator for the three roles the one Lambda image fulfills. */ -export type LambdaAction = "plan" | "renderChunk" | "assemble"; +export type LambdaAction = "plan" | "audio" | "renderChunk" | "assemble"; /** Transport protocol selected for one complete distributed render. */ export type LambdaPlanProtocol = "v1" | "v2"; @@ -38,6 +38,7 @@ export type LambdaPlanProtocol = "v1" | "v2"; */ export type LambdaEvent = | PlanEvent + | AudioEvent | RenderChunkEvent | AssembleEvent | { Payload: LambdaEvent } @@ -70,6 +71,11 @@ export interface PlanV2Event extends PlanEventBase { export type PlanEvent = PlanV1Event | PlanV2Event; +/** Makes a v2 plan's audio beside its chunks, from the same project and config as the plan. */ +export interface AudioEvent extends Omit { + Action: "audio"; +} + /** Activity B: fetch planDir, render one chunk, upload result. */ interface RenderChunkEventBase { Action: "renderChunk"; @@ -190,10 +196,20 @@ export interface PlanV2LambdaResult extends PlanLambdaResultBase { PlanProtocol: "v2"; PlanV2ManifestS3Uri: string; PlanV2ArtifactS3Prefix: string; + /** The plan left its audio to the audio task, which has a worker kept free for it. */ + SeparateAudio: boolean; } export type PlanLambdaResult = PlanV1LambdaResult | PlanV2LambdaResult; +export interface AudioLambdaResult { + Action: "audio"; + HasAudio: boolean; + /** The track for the assembler to stitch in; `null` without audio or when the plan mixed it. */ + AudioS3Uri: string | null; + DurationMs: number; +} + /** Result of a `renderChunk` invocation. Sized ≤200 bytes per §2.4. */ export interface RenderChunkLambdaResult { Action: "renderChunk"; @@ -215,4 +231,8 @@ export interface AssembleLambdaResult { DurationMs: number; } -export type LambdaResult = PlanLambdaResult | RenderChunkLambdaResult | AssembleLambdaResult; +export type LambdaResult = + | PlanLambdaResult + | AudioLambdaResult + | RenderChunkLambdaResult + | AssembleLambdaResult; diff --git a/packages/aws-lambda/src/handler.test.ts b/packages/aws-lambda/src/handler.test.ts index 4731384d0c..c4ee380038 100644 --- a/packages/aws-lambda/src/handler.test.ts +++ b/packages/aws-lambda/src/handler.test.ts @@ -1012,7 +1012,11 @@ describe("handler dispatch", () => { expect(url.searchParams.get("X-Amz-Signature")).toMatch(/^[a-f0-9]{64}$/); }); - async function planStitched(chunkCount: number) { + async function planStitched( + chunkCount: number, + separateAudio = false, + workerConcurrency?: number, + ) { const tmpRoot = makeTmpRoot(); const s3 = new FakeS3Client(); s3.objects.set("s3://bucket/project.tar.gz", await makeMinimalProjectTar()); @@ -1025,7 +1029,11 @@ describe("handler dispatch", () => { const planV2WithPublisherMock = mock( async (_dir: string, _config: unknown, publisher: PlanV2ArtifactPublisher) => { const v1Dir = join(tmpRoot, "v1"); - makeMinimalV1PlanDir(v1Dir, true, { chunkCount, chunksCarryAudio: true }); + makeMinimalV1PlanDir(v1Dir, !separateAudio, { + chunkCount, + chunksCarryAudio: !separateAudio, + separateAudio, + }); return publishPlanV2FromExecutionPlan(v1Dir, publisher); }, ); @@ -1045,13 +1053,13 @@ describe("handler dispatch", () => { Action: "plan", ProjectS3Uri: "s3://bucket/project.tar.gz", PlanOutputS3Prefix: "s3://bucket/renders/f/", - Config: { fps: 30, width: 640, height: 360, format: "mp4" }, + Config: { fps: 30, width: 640, height: 360, format: "mp4", workerConcurrency }, }, deps, ); if (!("PlanProtocol" in planned) || planned.PlanProtocol !== "v2") throw new Error("expected v2"); - const assemble = (chunkS3Uris: string[]) => + const assemble = (chunkS3Uris: string[], audioS3Uri: string | null = null) => handler( { Action: "assemble", @@ -1059,15 +1067,42 @@ describe("handler dispatch", () => { PlanV2ArtifactS3Prefix: planned.PlanV2ArtifactS3Prefix, PlanHash: planned.PlanHash, ChunkS3Uris: chunkS3Uris, - AudioS3Uri: null, + AudioS3Uri: audioS3Uri, OutputS3Uri: "s3://bucket/renders/f/output.mp4", Format: "mp4", }, deps, ); - return { s3, assembleMock, assemble }; + return { s3, assembleMock, assemble, planV2WithPublisherMock, planned }; } + it("keeps a worker free for the audio task, and keeps the audio in a plan with one worker", async () => { + const planOptions = (run: Awaited>) => + (run.planV2WithPublisherMock.mock.calls[0] as unknown[])[3]; + const four = await planStitched(8, true, 4); + expect(planOptions(four)).toMatchObject({ deferAudio: true }); + expect([four.planned.SeparateAudio, four.planned.MaxChunkConcurrency]).toEqual([true, 3]); + const one = await planStitched(8, false, 1); + expect(planOptions(one)).toMatchObject({ deferAudio: false }); + expect([one.planned.SeparateAudio, one.planned.MaxChunkConcurrency]).toEqual([false, 1]); + for (const [reserved, separate, chunks] of [ + ["2", true, 1], + ["1", false, 1], + ] as const) { + process.env.HYPERFRAMES_LAMBDA_RESERVED_CONCURRENCY = reserved; + try { + const run = await planStitched(8, separate, undefined); + expect(planOptions(run)).toMatchObject({ deferAudio: separate }); + expect([run.planned.SeparateAudio, run.planned.MaxChunkConcurrency]).toEqual([ + separate, + chunks, + ]); + } finally { + delete process.env.HYPERFRAMES_LAMBDA_RESERVED_CONCURRENCY; + } + } + }); + it.skipIf(spawnSync("ffmpeg", ["-version"]).status !== 0)( "stitches mp4 chunks inside S3 instead of joining them locally", async () => { @@ -1096,6 +1131,55 @@ describe("handler dispatch", () => { }, ); + it.skipIf(spawnSync("ffmpeg", ["-version"]).status !== 0)( + "stitches the audio task's track in when the plan left its audio out", + async () => { + const { s3, assemble, planV2WithPublisherMock } = await planStitched(2, true); + expect((planV2WithPublisherMock.mock.calls[0] as unknown[])[3]).toMatchObject({ + chunksCarryAudio: true, + deferAudio: true, + }); + const ffmpeg = (args: string[], path: string) => { + const run = spawnSync("ffmpeg", ["-nostdin", "-loglevel", "error", ...args, "-y", path]); + if (run.status !== 0) throw new Error(String(run.stderr)); + return readFileSync(path); + }; + const chunks = [0, 1].map((index) => { + const uri = `s3://bucket/renders/f/chunks/000${index}.mp4`; + const lavfi = ["-f", "lavfi", "-i", "testsrc2=size=64x48:rate=30:duration=1"]; + s3.objects.set( + uri, + ffmpeg([...lavfi, "-c:v", "libx264", "-bf", "0"], join(makeTmpRoot(), "chunk.mp4")), + ); + return uri; + }); + const audioUri = "s3://bucket/renders/f/audio/audio.m4a"; + s3.objects.set( + audioUri, + ffmpeg( + ["-f", "lavfi", "-i", "sine=duration=2", "-c:a", "aac"], + join(makeTmpRoot(), "a.m4a"), + ), + ); + await assemble(chunks, audioUri); + const outputPath = join(makeTmpRoot(), "out.mp4"); + writeFileSync(outputPath, s3.objects.get("s3://bucket/renders/f/output.mp4")!); + const probe = spawnSync( + "ffprobe", + ["-v", "error", "-show_entries", "stream=codec_type", "-of", "csv=p=0", "--", outputPath], + { encoding: "utf-8" }, + ); + expect(probe.stdout.trim().split("\n").sort()).toEqual(["audio", "video"]); + }, + ); + + it("refuses an audio URI for a v2 plan that ships its own audio", async () => { + const { assemble } = await planStitched(1); + await expect( + assemble(["s3://bucket/renders/f/chunks/0000.mp4"], "s3://bucket/renders/f/audio/audio.m4a"), + ).rejects.toMatchObject({ name: "PLAN_PROTOCOL_UNSUPPORTED" }); + }); + it("rejects unknown plan protocol values", async () => { const tmpRoot = makeTmpRoot(); const s3 = new FakeS3Client(); @@ -1158,6 +1242,55 @@ describe("handler dispatch", () => { }); }); +describe("handler — audio", () => { + async function runAudio(config: Record) { + const tmpRoot = makeTmpRoot(); + const s3 = new FakeS3Client(); + s3.objects.set("s3://bucket/project.tar.gz", await makeMinimalProjectTar()); + const buildPlanAudio = mock( + async (_project: string, _config: unknown, _work: string, outputPath: string) => { + writeFileSync(outputPath, "AAC"); + return { hasAudio: true }; + }, + ); + const result = await handler( + { + Action: "audio", + ProjectS3Uri: "s3://bucket/project.tar.gz", + PlanOutputS3Prefix: "s3://bucket/renders/f/", + Config: { fps: 30, width: 640, height: 360, format: "mp4", ...config }, + } as LambdaEvent, + { + s3: s3 as unknown as import("@aws-sdk/client-s3").S3Client, + primitives: { buildPlanAudio } as unknown as NonNullable< + Parameters[1] + >["primitives"], + tmpRoot, + skipChromeResolution: true, + }, + ); + return { result, s3, buildPlanAudio }; + } + + it("mixes a plan's deferred audio and uploads it beside the plan", async () => { + const { result, s3, buildPlanAudio } = await runAudio({}); + expect(result).toMatchObject({ + Action: "audio", + HasAudio: true, + AudioS3Uri: "s3://bucket/renders/f/audio/audio.m4a", + }); + expect(s3.objects.get("s3://bucket/renders/f/audio/audio.m4a")?.toString()).toBe("AAC"); + expect(buildPlanAudio).toHaveBeenCalledTimes(1); + }); + + it("leaves the audio to the plan when the stitch cannot take it (cfr)", async () => { + const { result, s3, buildPlanAudio } = await runAudio({ cfr: true }); + expect(result).toMatchObject({ Action: "audio", HasAudio: false, AudioS3Uri: null }); + expect(buildPlanAudio).not.toHaveBeenCalled(); + expect(s3.ops).toHaveLength(0); + }); +}); + /** A real client (so it can pre-sign offline) whose requests go to the fake. */ function presigningClient(fake: FakeS3Client): import("@aws-sdk/client-s3").S3Client { const client = new S3Client({ @@ -1398,7 +1531,12 @@ async function makeMinimalPlanTar(): Promise { function makeMinimalV1PlanDir( dir: string, withAudio: boolean, - options: { chunkCount?: number; chunksCarryAudio?: boolean; withDeferredVideo?: boolean } = {}, + options: { + chunkCount?: number; + chunksCarryAudio?: boolean; + separateAudio?: boolean; + withDeferredVideo?: boolean; + } = {}, ): void { const chunkCount = options.chunkCount ?? 1; mkdirSync(join(dir, "meta"), { recursive: true }); @@ -1426,7 +1564,10 @@ function makeMinimalV1PlanDir( ); writeFileSync( join(dir, "meta", "encoder.json"), - options.chunksCarryAudio ? JSON.stringify({ chunksCarryAudio: true }) : "{}", + JSON.stringify({ + ...(options.chunksCarryAudio ? { chunksCarryAudio: true } : {}), + ...(options.separateAudio ? { separateAudio: true } : {}), + }), ); if (withAudio) writeFileSync(join(dir, "audio.m4a"), "AAC"); if (options.withDeferredVideo) writeDeferredVideo(dir); diff --git a/packages/aws-lambda/src/handler.ts b/packages/aws-lambda/src/handler.ts index 5e2bbc06be..a4820a86ca 100644 --- a/packages/aws-lambda/src/handler.ts +++ b/packages/aws-lambda/src/handler.ts @@ -18,7 +18,10 @@ import { GetAccountSettingsCommand, LambdaClient } from "@aws-sdk/client-lambda" import { S3Client } from "@aws-sdk/client-s3"; import { assemble, + buildPlanAudio, chunksAreStitchedMp4, + plansSeparateAudio, + stitchesSeparateAudio, type AssembleResult, type ChunkRenderer, type ChunkResult, @@ -46,6 +49,8 @@ import { type DistributedFormat, formatExtension } from "./formatExtension.js"; import type { AssembleEvent, AssembleLambdaResult, + AudioEvent, + AudioLambdaResult, LambdaAction, LambdaEvent, LambdaResult, @@ -95,6 +100,7 @@ export interface HandlerDeps { primitives?: { plan: typeof plan; planV2WithPublisher?: typeof planV2WithPublisher; + buildPlanAudio?: typeof buildPlanAudio; renderChunk: ChunkRenderer; assemble: typeof assemble; }; @@ -127,6 +133,8 @@ export async function handler(event: LambdaEvent, deps?: HandlerDeps): Promise; if (typeof obj.Action === "string" && isLambdaAction(obj.Action)) { - return cursor as PlanEvent | RenderChunkEvent | AssembleEvent; + return cursor as PlanEvent | AudioEvent | RenderChunkEvent | AssembleEvent; } if ("Payload" in obj) { cursor = obj.Payload as LambdaEvent; @@ -220,13 +230,15 @@ export function unwrapEvent(event: LambdaEvent): PlanEvent | RenderChunkEvent | } function isLambdaAction(value: string): value is LambdaAction { - return value === "plan" || value === "renderChunk" || value === "assemble"; + return value === "plan" || value === "audio" || value === "renderChunk" || value === "assemble"; } // This is the single fail-closed boundary for the wire union. Keeping all // forbidden locator combinations together makes mixed-protocol input auditable. // fallow-ignore-next-line complexity -function validatePlanProtocolShape(event: PlanEvent | RenderChunkEvent | AssembleEvent): void { +function validatePlanProtocolShape( + event: PlanEvent | AudioEvent | RenderChunkEvent | AssembleEvent, +): void { const raw = event as unknown as Record; const protocol = raw.PlanProtocol; if (protocol !== undefined && protocol !== "v1" && protocol !== "v2") { @@ -236,7 +248,7 @@ function validatePlanProtocolShape(event: PlanEvent | RenderChunkEvent | Assembl error.name = "PLAN_PROTOCOL_UNSUPPORTED"; throw error; } - if (event.Action === "plan") return; + if (event.Action === "plan" || event.Action === "audio") return; const effectiveProtocol = protocol ?? "v2"; const hasV1Locator = typeof raw.PlanS3Uri === "string"; @@ -253,11 +265,6 @@ function validatePlanProtocolShape(event: PlanEvent | RenderChunkEvent | Assembl error.name = "PLAN_PROTOCOL_UNSUPPORTED"; throw error; } - if (effectiveProtocol === "v2" && event.Action === "assemble" && event.AudioS3Uri !== null) { - const error = new Error("[handler] v2 assemble audio must be materialized from the manifest"); - error.name = "PLAN_PROTOCOL_UNSUPPORTED"; - throw error; - } } /** @@ -281,7 +288,7 @@ function logEvent(payload: Record): void { // Keep event variants together so logs share one redaction and summarization boundary. // fallow-ignore-next-line complexity function summarizeEvent( - event: PlanEvent | RenderChunkEvent | AssembleEvent, + event: PlanEvent | AudioEvent | RenderChunkEvent | AssembleEvent, ): Record { switch (event.Action) { case "plan": @@ -292,6 +299,13 @@ function summarizeEvent( format: event.Config.format, fps: event.Config.fps, }; + case "audio": + return { + projectS3Uri: event.ProjectS3Uri, + planOutputS3Prefix: event.PlanOutputS3Prefix, + format: event.Config.format, + fps: event.Config.fps, + }; case "renderChunk": return { planProtocol: event.PlanProtocol ?? "v2", @@ -394,16 +408,7 @@ async function handlePlan(event: PlanEvent, deps?: HandlerDeps): Promise 1 && plansSeparateAudio(config, { deferAudio: true }); const manifest: PlanV2Manifest = await primitive(projectDir, config, publisher, { stagingParentDir: work, remoteFiles, chunksCarryAudio: true, + deferAudio: separateAudio, }); return { @@ -519,7 +526,10 @@ async function handlePlanV2( PlanV2ArtifactS3Prefix: publisher.artifactPrefix, PlanHash: manifest.planHash, ChunkCount: manifest.chunkCount, - MaxChunkConcurrency: Math.min(config.workerConcurrency, manifest.chunkCount), + MaxChunkConcurrency: Math.min( + config.workerConcurrency - (separateAudio ? 1 : 0), + manifest.chunkCount, + ), TotalFrames: manifest.totalFrames, Fps: manifest.fps, Width: manifest.width, @@ -527,6 +537,7 @@ async function handlePlanV2( Format: manifest.format, HasAudio: manifest.artifacts.some((artifact) => isPlanAudioArtifactPath(artifact.path)), AudioS3Uri: null, + SeparateAudio: separateAudio, FfmpegVersion: manifest.ffmpegVersion, ProducerVersion: manifest.producerVersion, DurationMs: Date.now() - started, @@ -536,6 +547,51 @@ async function handlePlanV2( } } +/** + * Points the engine at the bundled Chrome (unpacked into /tmp once per container) before any stage + * launches it; a caller-set PRODUCER_HEADLESS_SHELL_PATH, such as the SAM-local smoke's, wins. + */ +async function ensureChromePath(deps?: HandlerDeps): Promise { + if (deps?.skipChromeResolution || process.env.PRODUCER_HEADLESS_SHELL_PATH) return; + process.env.PRODUCER_HEADLESS_SHELL_PATH = await resolveChromeExecutablePath(); +} + +// ── Audio ─────────────────────────────────────────────────────────────────── + +async function handleAudio(event: AudioEvent, deps?: HandlerDeps): Promise { + const started = Date.now(); + const none = (): AudioLambdaResult => ({ + Action: "audio", + HasAudio: false, + AudioS3Uri: null, + DurationMs: Date.now() - started, + }); + // The plan mixed it (non-mp4, cfr): the assembler reads it from the manifest. + if (!plansSeparateAudio(event.Config, { deferAudio: true })) return none(); + const s3 = deps?.s3 ?? getS3Client(); + await ensureChromePath(deps); + const work = mkdtempSync(join(deps?.tmpRoot ?? tmpdir(), "hf-lambda-audio-")); + try { + const { projectDir, remoteFiles } = await fetchProject(s3, event.ProjectS3Uri, work, true); + const audioPath = join(work, "audio.m4a"); + const build = deps?.primitives?.buildPlanAudio ?? buildPlanAudio; + const { hasAudio } = await build(projectDir, event.Config, join(work, "plan"), audioPath, { + remoteFiles, + }); + if (!hasAudio) return none(); + const audioUri = `${trimTrailingSlash(event.PlanOutputS3Prefix)}/audio/${PLAN_AUDIO_RELATIVE_PATH}`; + await uploadFileToS3(s3, audioPath, audioUri, "audio/mp4"); + return { + Action: "audio", + HasAudio: true, + AudioS3Uri: audioUri, + DurationMs: Date.now() - started, + }; + } finally { + cleanupDir(work); + } +} + // ── RenderChunk ───────────────────────────────────────────────────────────── async function handleRenderChunk( @@ -548,18 +604,7 @@ async function handleRenderChunk( const started = Date.now(); const s3 = deps?.s3 ?? getS3Client(); const primitive = deps?.primitives?.renderChunk ?? renderChunk; - - // Sparticuz decompresses Chromium into /tmp on first call; warm starts - // skip the work (path already cached). Guard the env-var mutation too so - // a caller-supplied PRODUCER_HEADLESS_SHELL_PATH (e.g. the SAM-local - // RIE smoke) wins over the auto-resolution. - if (!deps?.skipChromeResolution && !process.env.PRODUCER_HEADLESS_SHELL_PATH) { - const chromePath = await resolveChromeExecutablePath(); - // The OSS engine resolves Chrome via `PRODUCER_HEADLESS_SHELL_PATH` - // first (see `browserManager.resolveHeadlessShellPath`); set it before - // invoking the primitive so launch picks up the bundled binary. - process.env.PRODUCER_HEADLESS_SHELL_PATH = chromePath; - } + await ensureChromePath(deps); const work = mkdtempSync(join(deps?.tmpRoot ?? tmpdir(), "hf-lambda-chunk-")); const planTar = join(work, "plan.tar.gz"); @@ -617,9 +662,7 @@ async function handleRenderChunkV2( const started = Date.now(); const s3 = deps?.s3 ?? getS3Client(); const primitive = deps?.primitives?.renderChunk ?? renderChunk; - if (!deps?.skipChromeResolution && !process.env.PRODUCER_HEADLESS_SHELL_PATH) { - process.env.PRODUCER_HEADLESS_SHELL_PATH = await resolveChromeExecutablePath(); - } + await ensureChromePath(deps); const work = mkdtempSync(join(deps?.tmpRoot ?? tmpdir(), "hf-lambda-chunk-v2-")); try { const { planDir } = await downloadAndMaterializePlanV2( @@ -754,6 +797,12 @@ async function handleAssembleV2( { role: "assembler" }, work, ); + // Only a plan that left its audio to the audio task takes it by URI; any other v2 plan ships it. + if (event.AudioS3Uri !== null && !stitchesSeparateAudio(planDir)) { + const error = new Error("[handler] v2 assemble audio must be materialized from the manifest"); + error.name = "PLAN_PROTOCOL_UNSUPPORTED"; + throw error; + } if (chunksAreStitchedMp4(planDir, event.Cfr === true)) { return await stitchChunksInS3(s3, event, manifest, started); } @@ -809,7 +858,9 @@ async function stitchChunksInS3( ); } const stitchStarted = Date.now(); - const stitched = await stitchMp4InS3(s3, event.ChunkS3Uris, event.OutputS3Uri); + const stitched = await stitchMp4InS3(s3, event.ChunkS3Uris, event.OutputS3Uri, { + audioUri: event.AudioS3Uri ?? undefined, + }); logEvent({ event: "assemble_stages", mode: "s3-stitch", @@ -962,9 +1013,12 @@ async function downloadChunkObjects( /** Collect every S3 URI that the handler will touch for a given event. */ // This is an exhaustive event-union projection used only for safe log summaries. // fallow-ignore-next-line complexity -function getEventS3Uris(event: PlanEvent | RenderChunkEvent | AssembleEvent): string[] { +function getEventS3Uris( + event: PlanEvent | AudioEvent | RenderChunkEvent | AssembleEvent, +): string[] { switch (event.Action) { case "plan": + case "audio": return [event.ProjectS3Uri, event.PlanOutputS3Prefix]; case "renderChunk": return event.PlanProtocol !== "v1" @@ -990,7 +1044,9 @@ function getEventS3Uris(event: PlanEvent | RenderChunkEvent | AssembleEvent): st * Skipped when `HYPERFRAMES_RENDER_BUCKET` is unset so existing deployments * without the env var continue to work. */ -function validateEventS3Uris(event: PlanEvent | RenderChunkEvent | AssembleEvent): void { +function validateEventS3Uris( + event: PlanEvent | AudioEvent | RenderChunkEvent | AssembleEvent, +): void { const allowedBucket = process.env.HYPERFRAMES_RENDER_BUCKET?.trim(); if (!allowedBucket) return; diff --git a/packages/aws-lambda/src/s3Stitch.test.ts b/packages/aws-lambda/src/s3Stitch.test.ts index 74c5adf61e..a807ef8f6d 100644 --- a/packages/aws-lambda/src/s3Stitch.test.ts +++ b/packages/aws-lambda/src/s3Stitch.test.ts @@ -230,7 +230,9 @@ describe.skipIf(!hasFfmpeg)("stitchMp4InS3 on real chunks", () => { expect(tailAudio).toBe(rate === 30); const uris = CHUNKS.map((slice) => `s3://bucket/chunks/${slice.index}.mp4`); // A 16 KiB part minimum makes these small chunks take every path S3 takes at 5 MiB. - const result = await stitchMp4InS3(s3.asClient(), uris, "s3://bucket/out.mp4", 16 * 1024); + const result = await stitchMp4InS3(s3.asClient(), uris, "s3://bucket/out.mp4", { + minPartBytes: 16 * 1024, + }); const outputPath = join(dir, "out.mp4"); const output = s3.objects.get("s3://bucket/out.mp4")!; writeFileSync(outputPath, output); @@ -278,6 +280,26 @@ describe.skipIf(!hasFfmpeg)("stitchMp4InS3 on real chunks", () => { reference[0]!.duration, reference[1]!.duration, ]); + + // Video-only chunks with the plan audio as a separate object: the same bytes as on disk, + // and the same frames and audio as today's join. + s3.objects.set("s3://bucket/audio.m4a", readFileSync(audioPath)); + for (const [i, path] of videoPaths.entries()) { + s3.objects.set(`s3://bucket/video/${i}.mp4`, readFileSync(path)); + } + await stitchMp4InS3( + s3.asClient(), + videoPaths.map((_, i) => `s3://bucket/video/${i}.mp4`), + "s3://bucket/separate.mp4", + { audioUri: "s3://bucket/audio.m4a", minPartBytes: 16 * 1024 }, + ); + writeFileSync(join(planDir, "meta", "encoder.json"), JSON.stringify({ separateAudio: true })); + const separateLocal = join(dir, "separate-local.mp4"); + await assemble(planDir, videoPaths, audioPath, separateLocal); + const separate = s3.objects.get("s3://bucket/separate.mp4")!; + expect(separate.equals(readFileSync(separateLocal))).toBe(true); + expect(frames(separateLocal)).toEqual(frames(referencePath)); + expect(pcm(separateLocal) === pcm(referencePath)).toBe(true); }, 60_000, ); diff --git a/packages/aws-lambda/src/s3Stitch.ts b/packages/aws-lambda/src/s3Stitch.ts index 739102663a..2d3b23da6c 100644 --- a/packages/aws-lambda/src/s3Stitch.ts +++ b/packages/aws-lambda/src/s3Stitch.ts @@ -127,20 +127,24 @@ export interface StitchResult { downloadedBytes: number; } -/** One faststart mp4 at `destUri` from the stitched mp4 chunks at `chunkUris`, in order. */ +/** One faststart mp4 at `destUri` from the stitched mp4 chunks at `chunkUris`, plus `audioUri`'s track when given. */ export async function stitchMp4InS3( s3: S3Client, chunkUris: readonly string[], destUri: string, - minPartBytes = MIN_PART_BYTES, + { audioUri, minPartBytes = MIN_PART_BYTES }: { audioUri?: string; minPartBytes?: number } = {}, ): Promise { + const sourceUris = audioUri ? [audioUri, ...chunkUris] : chunkUris; const layouts = await Promise.all( - chunkUris.map((uri) => readMp4Layout((start, end) => getRange(s3, uri, start, end))), + sourceUris.map((uri) => readMp4Layout((start, end) => getRange(s3, uri, start, end))), ); + const header = audioUri + ? buildStitchedHeader(layouts.slice(1), layouts[0]) + : buildStitchedHeader(layouts); const segments: Segment[] = [ - { bytes: buildStitchedHeader(layouts) }, + { bytes: header }, ...layouts.map((layout, i) => ({ - uri: chunkUris[i]!, + uri: sourceUris[i]!, start: layout.payloadStart, end: layout.payloadEnd, })), diff --git a/packages/producer/src/distributed.ts b/packages/producer/src/distributed.ts index 45ca2520ad..ca7fec8171 100644 --- a/packages/producer/src/distributed.ts +++ b/packages/producer/src/distributed.ts @@ -117,6 +117,7 @@ export { export { assemble, chunksAreStitchedMp4, + stitchesSeparateAudio, type AssembleChunkPaths, type AssembleResult, } from "./services/distributed/assemble.js"; diff --git a/packages/producer/src/services/distributed/assemble.ts b/packages/producer/src/services/distributed/assemble.ts index 3cfc7b66fa..df76814284 100644 --- a/packages/producer/src/services/distributed/assemble.ts +++ b/packages/producer/src/services/distributed/assemble.ts @@ -147,7 +147,7 @@ export async function assemble( } if (chunksAreStitchedMp4(planDir, cfr)) { - const separateAudio = lockedAudio(planDir).separateAudio === true && audioPath !== null; + const separateAudio = stitchesSeparateAudio(planDir) && audioPath !== null; await writeStitchedMp4(chunkPaths, outputPath, separateAudio ? audioPath : undefined); return { outputPath, @@ -410,6 +410,10 @@ export function chunksAreStitchedMp4(planDir: string, cfr: boolean): boolean { return true; } +export function stitchesSeparateAudio(planDir: string): boolean { + return lockedAudio(planDir).separateAudio === true; +} + function lockedAudio( planDir: string, ): Pick {