diff --git a/docs/deploy/aws-lambda.mdx b/docs/deploy/aws-lambda.mdx index 5e7520a049..3a35e805cf 100644 --- a/docs/deploy/aws-lambda.mdx +++ b/docs/deploy/aws-lambda.mdx @@ -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 diff --git a/packages/engine/src/services/videoFrameExtractor.test.ts b/packages/engine/src/services/videoFrameExtractor.test.ts index e643be22b9..ad0f9c6dc1 100644 --- a/packages/engine/src/services/videoFrameExtractor.test.ts +++ b/packages/engine/src/services/videoFrameExtractor.test.ts @@ -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", () => { diff --git a/packages/engine/src/services/videoFrameExtractor.ts b/packages/engine/src/services/videoFrameExtractor.ts index d5de062553..4e196c1707 100644 --- a/packages/engine/src/services/videoFrameExtractor.ts +++ b/packages/engine/src/services/videoFrameExtractor.ts @@ -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 { + 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( diff --git a/packages/producer/src/services/distributed/planV2.test.ts b/packages/producer/src/services/distributed/planV2.test.ts index 7f2fbb8668..11fbe1ec85 100644 --- a/packages/producer/src/services/distributed/planV2.test.ts +++ b/packages/producer/src/services/distributed/planV2.test.ts @@ -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((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[] = []; diff --git a/packages/producer/src/services/distributed/planV2.ts b/packages/producer/src/services/distributed/planV2.ts index 7a8f1541a7..6233e383bd 100644 --- a/packages/producer/src/services/distributed/planV2.ts +++ b/packages/producer/src/services/distributed/planV2.ts @@ -593,14 +593,21 @@ export async function publishPlanV2FromExecutionPlan( ): Promise { 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) {