mirror of
https://github.com/heygen-com/hyperframes.git
synced 2026-10-30 09:16:55 +00:00
fix(engine): extract long video ranges in segments and keep plan uploads flowing (#5280)
* fix(engine): extract long video ranges in segments and keep plan uploads flowing A distributed render's plan step extracted a whole video in one ffmpeg process, so a long clip hit ffmpegProcessTimeout (300 s) however much of Lambda's 15-minute limit was left, and every frame then uploaded in batches of 16 that each waited for their slowest blob. Ranges longer than two minutes now extract as two-minute segments that run side by side, each numbered after the last. They sample the same absolute times as a single pass, so the frames are byte-identical (checked on a 26-minute, 37,440 frame clip). Plan v2 publishing keeps 32 uploads in flight instead of batching. On a 10 GB Lambda the 26-minute clip's plan step fell from 395 s to 200 s. The Lambda docs now state the plan-step length bound. Fixes #5260 Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> * fix(engine): split extraction on an integer frame count The split compared and subtracted float seconds, so 200.3 - 80.3 became 120.00000000000001 s: a second segment got `-t 1.42e-14`, which ffmpeg rejects, failing an ordinary two-minute range. The frame count now comes from extractionFrameCountForDuration, the rule a single pass follows, and every segment takes an exact frame count. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
James Russo
co-authored by
Claude Opus 5.5
Claude Opus 5.5
parent
e07b64101b
commit
de1ef5a05a
@@ -223,6 +223,14 @@ The render bucket is created with CloudFormation `Retain` on delete — `hyperfr
|
||||
- Each `--region` is an independent stack. Cross-region failover is not built
|
||||
in.
|
||||
- The distributed AWS path is SDR-only.
|
||||
- Video length is bounded by the plan step. One Lambda invocation extracts
|
||||
every frame of every video and uploads them before any chunk starts, and it
|
||||
must finish inside Lambda's 15-minute limit. On a 10,240 MB function, a
|
||||
26-minute 576×1024 24 fps H.264 clip plans in about 3.5 minutes. Larger
|
||||
frames, high-bitrate footage and more videos take proportionally longer.
|
||||
ffmpeg extracts long clips in 2-minute segments, so `FFMPEG_PROCESS_TIMEOUT_MS`
|
||||
applies to each segment, not to the whole clip. Split edits that approach the
|
||||
limit into separate renders.
|
||||
|
||||
## Related topics
|
||||
|
||||
|
||||
@@ -1879,6 +1879,53 @@ describe.skipIf(!HAS_FFMPEG)("frame sampling at the output frame rate", () => {
|
||||
},
|
||||
30_000,
|
||||
);
|
||||
|
||||
// #5260: one ffmpeg process over a 26-minute clip hit ffmpegProcessTimeout.
|
||||
it.each([
|
||||
// 260 frames: one 240-frame segment, then 20 more.
|
||||
{ startTime: 0.5, duration: 130, frames: 260 },
|
||||
// 200.3 - 80.3 is 120.00000000000001 s: exactly one segment, not a second near-empty one.
|
||||
{ startTime: 80.3, duration: 200.3 - 80.3, frames: 240 },
|
||||
])(
|
||||
"extracts $duration s from $startTime s at 2 fps across segments without a gap or repeat",
|
||||
async (c) => {
|
||||
const source = join(FIXTURE_DIR, "index-2fps-long.mp4");
|
||||
if (!existsSync(source)) {
|
||||
const synth = await runFfmpeg([
|
||||
"-y",
|
||||
"-hide_banner",
|
||||
"-loglevel",
|
||||
"error",
|
||||
"-f",
|
||||
"lavfi",
|
||||
"-i",
|
||||
`nullsrc=s=${WIDTH}x${HEIGHT}:r=2:d=205,geq=lum='16+2*mod(N\\,100)':cb=128:cr=128`,
|
||||
"-c:v",
|
||||
"libx264",
|
||||
"-qp",
|
||||
"0",
|
||||
"-pix_fmt",
|
||||
"yuv420p",
|
||||
source,
|
||||
]);
|
||||
if (!synth.success) throw new Error(`long fixture synthesis failed: ${synth.stderr}`);
|
||||
}
|
||||
const extracted = await extractVideoFramesRange(
|
||||
source,
|
||||
`long-2fps-${c.startTime}`,
|
||||
c.startTime,
|
||||
c.duration,
|
||||
{ fps: 2, outputDir: FIXTURE_DIR, format: "png" },
|
||||
);
|
||||
const onScreen = Array.from(
|
||||
{ length: c.frames },
|
||||
(_, i) => Math.floor((c.startTime + i / 2) * 2 + 1e-9) % 100,
|
||||
);
|
||||
expect(extracted.totalFrames).toBe(c.frames);
|
||||
expect(sourceIndexes(extracted)).toEqual(onScreen);
|
||||
},
|
||||
60_000,
|
||||
);
|
||||
});
|
||||
|
||||
describe.skipIf(!HAS_FFMPEG)("held tails on sparse-timestamp sources", () => {
|
||||
|
||||
@@ -9,6 +9,7 @@
|
||||
import { audioGroupsById, isMemberGroupHidden, isSelfOrAncestorHidden } from "./mediaHidden.js";
|
||||
import { copyFileSync, existsSync, linkSync, mkdirSync, rmSync } from "fs";
|
||||
import { join } from "path";
|
||||
import { cpus } from "os";
|
||||
import { parseHTML } from "linkedom";
|
||||
import { resolveProjectRelativeSrc } from "@hyperframes/parsers/asset-resolution";
|
||||
import {
|
||||
@@ -856,6 +857,84 @@ function restoreSourceColourFilter(metadata: VideoMetadata): string[] {
|
||||
: [];
|
||||
}
|
||||
|
||||
/**
|
||||
* Longest stretch of output one ffmpeg process extracts. `ffmpegProcessTimeout` bounds a
|
||||
* single process, so one process over a long clip timed out however much time the render
|
||||
* had left. Longer ranges split into segments that run side by side. Segments sample the
|
||||
* same absolute times as a single pass, so their frames are byte-identical to it.
|
||||
*/
|
||||
const EXTRACTION_SEGMENT_SECONDS = 120;
|
||||
|
||||
interface SegmentedExtraction {
|
||||
decodeArgs: string[];
|
||||
filterAndEncodeArgs: string[];
|
||||
videoPath: string;
|
||||
startTime: number;
|
||||
totalFrames: number;
|
||||
fps: number;
|
||||
segmentFrames: number;
|
||||
runOptions: { signal?: AbortSignal; timeout: number };
|
||||
}
|
||||
|
||||
async function runSegmentedExtraction(job: SegmentedExtraction): Promise<RunFfmpegResult> {
|
||||
const { decodeArgs, filterAndEncodeArgs, videoPath, startTime, totalFrames, fps, segmentFrames } =
|
||||
job;
|
||||
const segmentCount = Math.ceil(totalFrames / segmentFrames);
|
||||
// One failed segment fails the range, so stop the others instead of finishing them.
|
||||
const failed = new AbortController();
|
||||
const signal = job.runOptions.signal
|
||||
? AbortSignal.any([job.runOptions.signal, failed.signal])
|
||||
: failed.signal;
|
||||
const results: RunFfmpegResult[] = [];
|
||||
let next = 0;
|
||||
const worker = async () => {
|
||||
while (next < segmentCount && !signal.aborted) {
|
||||
const index = next++;
|
||||
const firstFrame = index * segmentFrames;
|
||||
const frames = Math.min(segmentFrames, totalFrames - firstFrame);
|
||||
const result = await runFfmpeg(
|
||||
[
|
||||
...decodeArgs,
|
||||
"-noaccurate_seek",
|
||||
"-ss",
|
||||
String(startTime + firstFrame / fps),
|
||||
"-i",
|
||||
videoPath,
|
||||
// Read one frame past the segment and keep exactly its own frames.
|
||||
"-t",
|
||||
String((frames + 1) / fps),
|
||||
"-frames:v",
|
||||
String(frames),
|
||||
"-start_number",
|
||||
String(firstFrame + 1),
|
||||
...filterAndEncodeArgs,
|
||||
],
|
||||
{ ...job.runOptions, signal },
|
||||
);
|
||||
results[index] = result;
|
||||
if (!result.success) failed.abort();
|
||||
}
|
||||
};
|
||||
const workers = Math.min(segmentCount, Math.max(1, Math.floor(cpus().length / 2)));
|
||||
await Promise.all(Array.from({ length: workers }, worker));
|
||||
const ran = results.filter(Boolean);
|
||||
// Report the segment that failed, not the ones stopped because of it.
|
||||
const failure =
|
||||
ran.find((r) => !r.success && r.terminationReason !== "abort") ?? ran.find((r) => !r.success);
|
||||
if (failure) return failure;
|
||||
// A cancel between segments leaves later ones unrun; their frames are missing, not done.
|
||||
if (ran.length < segmentCount) {
|
||||
return {
|
||||
success: false,
|
||||
exitCode: null,
|
||||
stderr: "",
|
||||
durationMs: 0,
|
||||
terminationReason: "abort",
|
||||
};
|
||||
}
|
||||
return ran[ran.length - 1]!;
|
||||
}
|
||||
|
||||
export async function extractVideoFramesRange(
|
||||
videoPath: string,
|
||||
videoId: string,
|
||||
@@ -929,6 +1008,7 @@ export async function extractVideoFramesRange(
|
||||
if (codecMayHaveAlpha(metadata.videoCodec)) {
|
||||
args.push("-c:v", decoderForCodec(metadata.videoCodec));
|
||||
}
|
||||
const decodeArgs = [...args];
|
||||
if (options.finalFrameOnly) {
|
||||
// Output-side seek decodes from the start before selecting the final
|
||||
// sample. This is intentionally reserved for the one-frame path: input
|
||||
@@ -1008,8 +1088,22 @@ export async function extractVideoFramesRange(
|
||||
runOptions,
|
||||
);
|
||||
} else {
|
||||
if (vfFilters.length > 0) args.push("-vf", vfFilters.join(","));
|
||||
processResult = await runFfmpeg([...args, ...encodeArgs], runOptions);
|
||||
const filterArgs = vfFilters.length > 0 ? ["-vf", vfFilters.join(",")] : [];
|
||||
const segmentFrames = Math.round(EXTRACTION_SEGMENT_SECONDS * fps);
|
||||
const totalFrames = extractionFrameCountForDuration(duration, normalizedFps, false);
|
||||
processResult =
|
||||
sampleCfrAtOutputFps && totalFrames > segmentFrames
|
||||
? await runSegmentedExtraction({
|
||||
decodeArgs,
|
||||
filterAndEncodeArgs: [...filterArgs, ...encodeArgs],
|
||||
videoPath,
|
||||
startTime,
|
||||
totalFrames,
|
||||
fps,
|
||||
segmentFrames,
|
||||
runOptions,
|
||||
})
|
||||
: await runFfmpeg([...args, ...filterArgs, ...encodeArgs], runOptions);
|
||||
}
|
||||
if (processResult.failureReason === "external_interruption") {
|
||||
throw new VideoSourceExtractionError(
|
||||
|
||||
@@ -874,6 +874,33 @@ describe("Plan v2 artifact publisher", () => {
|
||||
await publisher.abort();
|
||||
});
|
||||
|
||||
it("keeps publishing the other blobs while one upload is slow", async () => {
|
||||
const root = tempPath("hf-plan-v2-publisher-slow-");
|
||||
const v1 = createV1Plan(root);
|
||||
for (let i = 0; i < 40; i++)
|
||||
writeFileSync(join(v1, "compiled", `frame-${i}.txt`), `frame ${i}`);
|
||||
refreshV1PlanHash(v1);
|
||||
const manifest = await publishPlanV2FromV1(v1, {
|
||||
async commitManifest() {},
|
||||
async abort() {},
|
||||
putBlob: (() => {
|
||||
let started = 0;
|
||||
let releaseSlow: (byOthers: boolean) => void = () => {};
|
||||
const slow = new Promise<boolean>((resolve) => (releaseSlow = resolve));
|
||||
setTimeout(() => releaseSlow(false), 2_000);
|
||||
return async () => {
|
||||
started += 1;
|
||||
if (started === 1) {
|
||||
expect(await slow).toBe(true);
|
||||
return;
|
||||
}
|
||||
if (started === 41) releaseSlow(true);
|
||||
};
|
||||
})(),
|
||||
});
|
||||
expect(manifest.artifacts.length).toBeGreaterThan(40);
|
||||
});
|
||||
|
||||
it("aborts without committing a manifest when a blob publish fails", async () => {
|
||||
const root = tempPath("hf-plan-v2-publisher-failure-");
|
||||
const calls: string[] = [];
|
||||
|
||||
@@ -593,14 +593,21 @@ export async function publishPlanV2FromExecutionPlan(
|
||||
): Promise<PlanV2Manifest> {
|
||||
try {
|
||||
const publication = buildPlanV2Publication(executionPlanDir);
|
||||
const concurrency = 16;
|
||||
for (let offset = 0; offset < publication.blobs.length; offset += concurrency) {
|
||||
const batch = publication.blobs.slice(offset, offset + concurrency);
|
||||
const results = await Promise.allSettled(batch.map((blob) => publisher.putBlob(blob)));
|
||||
for (const result of results) {
|
||||
if (result.status === "rejected") throw result.reason;
|
||||
const uploadSlots = 32;
|
||||
let next = 0;
|
||||
let failure: { reason: unknown } | undefined;
|
||||
const upload = async () => {
|
||||
while (!failure && next < publication.blobs.length) {
|
||||
const blob = publication.blobs[next++]!;
|
||||
try {
|
||||
await publisher.putBlob(blob);
|
||||
} catch (reason) {
|
||||
failure ??= { reason };
|
||||
}
|
||||
}
|
||||
}
|
||||
};
|
||||
await Promise.all(Array.from({ length: uploadSlots }, upload));
|
||||
if (failure) throw failure.reason;
|
||||
await publisher.commitManifest(canonicalJsonStringify(publication.manifest));
|
||||
return publication.manifest;
|
||||
} catch (error) {
|
||||
|
||||
Reference in New Issue
Block a user