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
This commit is contained in:
Miguel Ángel
2026-10-21 05:26:58 +00:00
committed by GitHub
parent b16d5a7063
commit dee8380d22
10 changed files with 558 additions and 156 deletions
+149 -71
View File
@@ -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:
@@ -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<string, unknown> };
ItemProcessor?: { States?: Record<string, unknown> };
}
| 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<string>();
const samErrors = new Set<string>();
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<string, unknown> {
);
}
/** The state named `name` in one of the v2 Parallel's branches. */
function renderV2Branch(definition: { States: Record<string, unknown> }, 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<string, unknown>;
}): 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,
};
@@ -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)
+22 -2
View File
@@ -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<PlanEventBase, "Action"> {
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;
+149 -8
View File
@@ -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<ReturnType<typeof planStitched>>) =>
(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<string, unknown>) {
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<typeof handler>[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<Buffer> {
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);
+99 -43
View File
@@ -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<L
switch (unwrapped.Action) {
case "plan":
return await handlePlan(unwrapped, deps);
case "audio":
return await handleAudio(unwrapped, deps);
case "renderChunk":
return await handleRenderChunk(unwrapped, deps);
case "assemble":
@@ -195,13 +203,15 @@ function normalizeTerminalErrorName(error: unknown): void {
// configurations and prevents infinite loops on malformed input.
const MAX_ENVELOPE_DEPTH = 4;
export function unwrapEvent(event: LambdaEvent): PlanEvent | RenderChunkEvent | AssembleEvent {
export function unwrapEvent(
event: LambdaEvent,
): PlanEvent | AudioEvent | RenderChunkEvent | AssembleEvent {
let cursor: LambdaEvent = event;
for (let i = 0; i < MAX_ENVELOPE_DEPTH; i++) {
if (cursor && typeof cursor === "object") {
const obj = cursor as Record<string, unknown>;
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<string, unknown>;
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<string, unknown>): 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<string, unknown> {
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<PlanLam
const config = await planConfig(event, deps);
const s3 = deps?.s3 ?? getS3Client();
const primitive = deps?.primitives?.plan ?? plan;
// The producer's probe stage launches Chromium whenever the composition
// needs a runtime duration probe or has unresolved sub-compositions, so
// plan has to resolve Chrome the same way renderChunk does. Without this
// the probe throws "An `executablePath` or `channel` must be specified
// for `puppeteer-core`" the moment runProbeStage calls puppeteer.launch.
if (!deps?.skipChromeResolution && !process.env.PRODUCER_HEADLESS_SHELL_PATH) {
const chromePath = await resolveChromeExecutablePath();
process.env.PRODUCER_HEADLESS_SHELL_PATH = chromePath;
}
await ensureChromePath(deps);
const work = mkdtempSync(join(deps?.tmpRoot ?? tmpdir(), "hf-lambda-plan-"));
const planDir = join(work, "plan");
@@ -494,9 +499,7 @@ async function handlePlanV2(
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) {
process.env.PRODUCER_HEADLESS_SHELL_PATH = await resolveChromeExecutablePath();
}
await ensureChromePath(deps);
const work = mkdtempSync(join(deps?.tmpRoot ?? tmpdir(), "hf-lambda-plan-v2-"));
try {
@@ -506,10 +509,14 @@ async function handlePlanV2(
planOutputS3Prefix: event.PlanOutputS3Prefix,
temporaryRoot: work,
});
// The audio task shares the chunks' workers: it runs beside them only with one kept free.
const separateAudio =
config.workerConcurrency > 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<void> {
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<AudioLambdaResult> {
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;
+23 -1
View File
@@ -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,
);
+9 -5
View File
@@ -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<StitchResult> {
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,
})),
+1
View File
@@ -117,6 +117,7 @@ export {
export {
assemble,
chunksAreStitchedMp4,
stitchesSeparateAudio,
type AssembleChunkPaths,
type AssembleResult,
} from "./services/distributed/assemble.js";
@@ -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<LockedRenderConfig, "chunksCarryAudio" | "separateAudio"> {