diff --git a/docs/deploy/aws-lambda.mdx b/docs/deploy/aws-lambda.mdx index 3a35e805cf..dc59d80158 100644 --- a/docs/deploy/aws-lambda.mdx +++ b/docs/deploy/aws-lambda.mdx @@ -223,14 +223,16 @@ 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. +- Each render chunk extracts only the video frames it shows, so the plan step + no longer grows with video length. On a 10,240 MB function, a 26-minute + 576×1024 24 fps H.264 clip plans in about 20 seconds. Each chunk downloads + the whole source file of every video it shows. Variable-frame-rate and HDR + sources are the exception: the plan step still extracts every frame of those, + inside Lambda's 15-minute limit, in 2-minute segments, so + `FFMPEG_PROCESS_TIMEOUT_MS` applies to each segment, not to the whole clip. +- A chunk must also finish inside 15 minutes. For long renders, raise + `--max-parallel-chunks` (`maxParallelChunks` in the SDK) so each chunk covers + fewer frames. ## Related topics diff --git a/packages/engine/src/index.ts b/packages/engine/src/index.ts index bb01945821..0ff6cbf4af 100644 --- a/packages/engine/src/index.ts +++ b/packages/engine/src/index.ts @@ -221,6 +221,8 @@ export { type VideoElement, type ImageElement, type ExtractedFrames, + type DeferredFrameRange, + type FrameRange, type ExtractionOptions, type ExtractionResult, type ExtractionPhaseBreakdown, diff --git a/packages/engine/src/services/extractedFrameIndex.ts b/packages/engine/src/services/extractedFrameIndex.ts index af6e1bde54..a041033ac1 100644 --- a/packages/engine/src/services/extractedFrameIndex.ts +++ b/packages/engine/src/services/extractedFrameIndex.ts @@ -27,6 +27,7 @@ export function extractedFrameIndex(file: string, format: ExtractedFrameFormat): export function framePathsFromDirectory( outputDir: string, format: ExtractedFrameFormat, + allowGaps = false, ): Map { const suffix = `.${format}`; const indexed = new Map(); @@ -40,6 +41,7 @@ export function framePathsFromDirectory( } indexed.set(index, join(outputDir, file)); } + if (allowGaps) return new Map([...indexed].sort(([a], [b]) => a - b)); const ordered = new Map(); for (let index = 0; index < indexed.size; index += 1) { diff --git a/packages/engine/src/services/videoFrameExtractor.test.ts b/packages/engine/src/services/videoFrameExtractor.test.ts index ad0f9c6dc1..2f917e9a3d 100644 --- a/packages/engine/src/services/videoFrameExtractor.test.ts +++ b/packages/engine/src/services/videoFrameExtractor.test.ts @@ -1926,6 +1926,112 @@ describe.skipIf(!HAS_FFMPEG)("frame sampling at the output frame rate", () => { }, 60_000, ); + + // A distributed chunk extracts only the frames it renders; they must be the full extraction's. + describe("partial extraction", () => { + const gopSource = join(FIXTURE_DIR, "gop-30fps.mp4"); + + beforeAll(async () => { + // Real motion and long GOPs, so range seeks land mid-GOP. + const synth = await runFfmpeg([ + "-y", + "-hide_banner", + "-loglevel", + "error", + "-f", + "lavfi", + "-i", + "testsrc2=s=64x36:r=30:d=20", + "-c:v", + "libx264", + "-g", + "250", + "-pix_fmt", + "yuv420p", + gopSource, + ]); + if (!synth.success) throw new Error(`gop fixture synthesis failed: ${synth.stderr}`); + }, 30_000); + + function frameBytes(dir: string): Map { + return new Map(readdirSync(dir).map((name) => [name, readFileSync(join(dir, name))])); + } + + it.each([ + { fps: 24, startTime: 1.3, duration: 17 }, + { fps: 60, startTime: 0, duration: 20 }, + ])( + "frameRanges at $fps fps write the same files as a full extraction", + async (c) => { + const full = await extractVideoFramesRange( + gopSource, + `full-${c.fps}`, + c.startTime, + c.duration, + { fps: c.fps, outputDir: FIXTURE_DIR, format: "jpg" }, + ); + const last = full.totalFrames; + const ranges = [ + { firstFrame: 0, frames: 5 }, + { firstFrame: 97, frames: 40 }, + { firstFrame: last - 3, frames: 3 }, + ]; + const partial = await extractVideoFramesRange( + gopSource, + `partial-${c.fps}`, + c.startTime, + c.duration, + { fps: c.fps, outputDir: FIXTURE_DIR, format: "jpg", frameRanges: ranges }, + ); + const fullFrames = frameBytes(full.outputDir); + const partialFrames = frameBytes(partial.outputDir); + expect(partialFrames.size).toBe(48); + for (const [name, bytes] of partialFrames) { + expect(fullFrames.get(name)?.equals(bytes), name).toBe(true); + } + }, + 60_000, + ); + + it.each([ + { fps: 24, mediaStart: 3.3 }, + { fps: 30, mediaStart: 0 }, + { fps: 60, mediaStart: 7.77 }, + ])( + "a deferred window to the source end at $fps fps reports the full frame count", + async (c) => { + const video = (): VideoElement => ({ + id: `defer-${c.fps}`, + src: gopSource, + start: 0, + end: Infinity, + mediaStart: c.mediaStart, + loop: false, + hasAudio: false, + }); + const full = await extractAllVideoFrames([video()], FIXTURE_DIR, { + fps: c.fps, + outputDir: join(FIXTURE_DIR, `defer-full-${c.fps}`), + }); + const deferred = await extractAllVideoFrames([video()], FIXTURE_DIR, { + fps: c.fps, + outputDir: join(FIXTURE_DIR, `defer-${c.fps}`), + deferRangeExtraction: true, + }); + expect(deferred.errors).toEqual([]); + const result = deferred.extracted[0]!; + expect(result.totalFrames).toBe(full.extracted[0]!.totalFrames); + expect(result.framePaths.size).toBe(0); + expect(readdirSync(result.outputDir)).toEqual([]); + expect(result.deferredRange).toEqual({ + startTime: c.mediaStart, + durationSeconds: expect.any(Number), + format: "jpg", + }); + }, + 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 4e196c1707..1dfee36d99 100644 --- a/packages/engine/src/services/videoFrameExtractor.ts +++ b/packages/engine/src/services/videoFrameExtractor.ts @@ -97,6 +97,7 @@ export interface ExtractedFrames { totalFrames: number; metadata: VideoMetadata; framePaths: Map; + deferredRange?: DeferredFrameRange; /** * True when the extractor owns `outputDir` and cleanup should rm it when * the render ends. Cache hits set this to false so the shared entry isn't @@ -194,6 +195,18 @@ export function extractionFrameCountForDuration( return Math.max(1, Number.isSafeInteger(frames) ? frames : Number.MAX_SAFE_INTEGER); } +export interface FrameRange { + firstFrame: number; + frames: number; +} + +/** The window a deferred extraction samples; frame k is the source frame on screen at `startTime + k / fps`. */ +export interface DeferredFrameRange { + startTime: number; + durationSeconds: number; + format: "jpg" | "png"; +} + export interface ExtractionOptions { /** Exact configured rate. Rational rates are passed to FFmpeg verbatim. */ fps: FpsInput; @@ -222,6 +235,13 @@ export interface ExtractionOptions { * candidate enforce lane may opt into typed aggregation. */ collectProbeFailures?: boolean; + /** Extract only these runs of output frames, named as a full extraction names them. CFR only. */ + frameRanges?: readonly FrameRange[]; + /** + * Leave CFR windows longer than a second unextracted, returning `deferredRange` and an exact + * `totalFrames` so another process can extract just the frames it needs with `frameRanges`. + */ + deferRangeExtraction?: boolean; } export const EXTRACT_CACHE_MIN_AGE_MS = 60 * 60 * 1000; @@ -870,16 +890,28 @@ interface SegmentedExtraction { filterAndEncodeArgs: string[]; videoPath: string; startTime: number; - totalFrames: number; fps: number; - segmentFrames: number; + segments: readonly FrameRange[]; runOptions: { signal?: AbortSignal; timeout: number }; } +/** Split runs so no single ffmpeg process extracts more than `segmentFrames` frames. */ +function splitFrameRanges(ranges: readonly FrameRange[], segmentFrames: number): FrameRange[] { + const segments: FrameRange[] = []; + for (const range of ranges) { + for (let offset = 0; offset < range.frames; offset += segmentFrames) { + segments.push({ + firstFrame: range.firstFrame + offset, + frames: Math.min(segmentFrames, range.frames - offset), + }); + } + } + return segments; +} + async function runSegmentedExtraction(job: SegmentedExtraction): Promise { - const { decodeArgs, filterAndEncodeArgs, videoPath, startTime, totalFrames, fps, segmentFrames } = - job; - const segmentCount = Math.ceil(totalFrames / segmentFrames); + const { decodeArgs, filterAndEncodeArgs, videoPath, startTime, fps, segments } = job; + const segmentCount = segments.length; // One failed segment fails the range, so stop the others instead of finishing them. const failed = new AbortController(); const signal = job.runOptions.signal @@ -890,8 +922,7 @@ async function runSegmentedExtraction(job: SegmentedExtraction): Promise { while (next < segmentCount && !signal.aborted) { const index = next++; - const firstFrame = index * segmentFrames; - const frames = Math.min(segmentFrames, totalFrames - firstFrame); + const { firstFrame, frames } = segments[index]!; const result = await runFfmpeg( [ ...decodeArgs, @@ -991,6 +1022,9 @@ export async function extractVideoFramesRange( const decodeWithVideoToolbox = isHdr && isMacOS && !toneMappedToSdr; const sampleCfrAtOutputFps = !options.finalFrameOnly && !metadata.isVFR; + if (options.frameRanges && !sampleCfrAtOutputFps) { + throw new Error("frameRanges extraction needs a constant-frame-rate source"); + } const args: string[] = []; if (decodeWithVideoToolbox) { @@ -1092,15 +1126,17 @@ export async function extractVideoFramesRange( const segmentFrames = Math.round(EXTRACTION_SEGMENT_SECONDS * fps); const totalFrames = extractionFrameCountForDuration(duration, normalizedFps, false); processResult = - sampleCfrAtOutputFps && totalFrames > segmentFrames + sampleCfrAtOutputFps && (options.frameRanges || totalFrames > segmentFrames) ? await runSegmentedExtraction({ decodeArgs, filterAndEncodeArgs: [...filterArgs, ...encodeArgs], videoPath, startTime, - totalFrames, fps, - segmentFrames, + segments: splitFrameRanges( + options.frameRanges ?? [{ firstFrame: 0, frames: totalFrames }], + segmentFrames, + ), runOptions, }) : await runFfmpeg([...args, ...filterArgs, ...encodeArgs], runOptions); @@ -1154,7 +1190,7 @@ export async function extractVideoFramesRange( ); } - const framePaths = framePathsFromDirectory(videoOutputDir, format); + const framePaths = framePathsFromDirectory(videoOutputDir, format, !!options.frameRanges); if (framePaths.size === 0 && duration > 0) { throw new VideoSourceExtractionError( "zero_output", @@ -1168,7 +1204,7 @@ export async function extractVideoFramesRange( // frames predicted by the probed duration, the duration is likely // inflated (container duration includes a longer audio track). Correct // it so downstream coverage accounting uses the actual video extent. - if (framePaths.size > 0 && duration > 0 && !options.finalFrameOnly) { + if (framePaths.size > 0 && duration > 0 && !options.finalFrameOnly && !options.frameRanges) { const expectedAtFps = extractionFrameCountForDuration(duration, normalizedFps, metadata.isVFR); if (expectedAtFps > 0 && framePaths.size <= expectedAtFps * 0.6) { const correctedDuration = startTime + framePaths.size / fps; @@ -2186,6 +2222,68 @@ export async function extractAllVideoFrames( return rehydratePublishedCache(work, cacheTarget); } + /** + * Every output slot before the source's end holds a frame, so decoding the last second gives + * the exact count. Anything unusual returns null and takes the normal path. + */ + async function deferExtraction(work: PreparedExtraction): Promise { + if ( + work.finalFrameOnly || + work.metadata.isVFR || + work.sdrToHdrTransfer !== undefined || + work.hdrToSdrTransformKey !== undefined + ) { + return null; + } + const expectedFrames = extractionFrameCountForDuration( + work.videoDuration, + configuredFps, + false, + ); + const tailFrames = Math.min(expectedFrames, Math.ceil(fps)); + if (expectedFrames <= tailFrames) return null; + const probeDir = join(options.outputDir, `${work.video.id}.tail-probe`); + try { + const tail = await extractVideoFramesRange( + work.videoPath, + work.video.id, + work.extractionMediaStart, + work.videoDuration, + { + ...scopedExtractionOptions(work), + frameRanges: [{ firstFrame: expectedFrames - tailFrames, frames: tailFrames }], + }, + signal, + config, + probeDir, + ); + const totalFrames = expectedFrames - tailFrames + tail.totalFrames; + // A full extraction this short rewrites the probed duration; let it. + if (totalFrames <= expectedFrames * 0.6) return null; + const outputDir = join(options.outputDir, work.video.id); + mkdirSync(outputDir, { recursive: true }); + return { + videoId: work.video.id, + srcPath: work.videoPath, + outputDir, + framePattern: `${FRAME_FILENAME_PREFIX}%05d.${work.format}`, + fps, + totalFrames, + metadata: work.metadata, + framePaths: new Map(), + deferredRange: { + startTime: work.extractionMediaStart, + durationSeconds: work.videoDuration, + format: work.format, + }, + }; + } catch { + return null; + } finally { + rmSync(probeDir, { recursive: true, force: true }); + } + } + async function executeDirectMiss( miss: UniqueExtractionMiss, maxTransientRetries = options.maxTransientRetries ?? 0, @@ -2350,8 +2448,17 @@ export async function extractAllVideoFrames( } const uniqueOutcomes = new Map(); + if (options.deferRangeExtraction) { + const works = [...uniqueWorks.values()]; + const deferred = await Promise.all(works.map(deferExtraction)); + works.forEach((work, i) => { + const result = deferred[i]; + if (result) uniqueOutcomes.set(work.dedupeKey, { result }); + }); + } const cacheMisses: UniqueExtractionMiss[] = []; for (const work of uniqueWorks.values()) { + if (uniqueOutcomes.has(work.dedupeKey)) continue; const lookup = lookupCacheFor(work); if ("work" in lookup) { cacheMisses.push(lookup); @@ -2419,7 +2526,7 @@ export async function extractAllVideoFrames( errors.push(item.error); } else if ("result" in item) { extracted.push(item.result); - totalFramesExtracted += item.result.totalFrames; + if (!item.result.deferredRange) totalFramesExtracted += item.result.totalFrames; } } diff --git a/packages/producer/src/services/distributed/plan.ts b/packages/producer/src/services/distributed/plan.ts index 93edeafa31..fe7a6db4b8 100644 --- a/packages/producer/src/services/distributed/plan.ts +++ b/packages/producer/src/services/distributed/plan.ts @@ -26,8 +26,17 @@ * never have to handle them. */ -import { cpSync, existsSync, mkdirSync, renameSync, rmSync, writeFileSync } from "node:fs"; -import { join, relative, sep } from "node:path"; +import { + copyFileSync, + cpSync, + existsSync, + linkSync, + mkdirSync, + renameSync, + rmSync, + writeFileSync, +} from "node:fs"; +import { extname, isAbsolute, join, relative, sep } from "node:path"; import { type CanvasResolution, fpsToNumber } from "@hyperframes/core"; import { type EngineConfig, @@ -276,6 +285,12 @@ export interface BuildLocalExecutionPlanOptions { * transport cap because it publishes role-scoped content-addressed objects. */ readonly executionPlanSizeLimitBytes?: number; + /** + * Leave long constant-frame-rate video windows unextracted and ship their + * sources, so each chunk extracts only the frames it renders. Plan v2 only: + * chunks write those frames into their own work directory. + */ + readonly deferVideoExtraction?: boolean; } /** @@ -825,6 +840,53 @@ export function resolveDistributedEngineConfig(config: DistributedRenderConfig): }; } +function pathInside(root: string, path: string): string | null { + const rel = relative(root, path); + return rel === "" || rel.startsWith("..") || isAbsolute(rel) ? null : rel; +} + +/** + * Return a planDir-relative path chunks can extract a deferred video from. + * Sources the plan already ships under `compiled/` are reused; anything else + * (downloads, files outside the project) is linked into `video-sources/`. + * Call after the staged `compiled/` and `video-frames/` trees are promoted. + */ +function createVideoSourceShipper(input: { + planDir: string; + projectDir: string; + workCompiledDir: string; + workVideoFramesDir: string; +}): (srcPath: string) => string { + const shipped = new Map(); + return (srcPath) => { + const known = shipped.get(srcPath); + if (known) return known; + // Downloads were staged under the frame tree, which now lives at video-frames/. + const stagedRel = pathInside(input.workVideoFramesDir, srcPath); + const current = stagedRel === null ? srcPath : join(input.planDir, "video-frames", stagedRel); + let result: string | null = null; + for (const root of stagedRel === null ? [input.workCompiledDir, input.projectDir] : []) { + const rel = pathInside(root, srcPath); + if (rel !== null && existsSync(join(input.planDir, "compiled", rel))) { + result = ["compiled", ...rel.split(sep)].join("/"); + break; + } + } + if (result === null) { + result = `video-sources/${shipped.size}${extname(current)}`; + const destination = join(input.planDir, ...result.split("/")); + mkdirSync(join(input.planDir, "video-sources"), { recursive: true }); + try { + linkSync(current, destination); + } catch { + copyFileSync(current, destination); + } + } + shipped.set(srcPath, result); + return result; + }; +} + /** * Build the shared local execution representation used by both transport * protocols. See the module docstring for the directory layout. @@ -1043,6 +1105,7 @@ export async function buildLocalExecutionPlan( abortSignal, assertNotAborted, materializeSymlinks: true, + deferRangeExtraction: options.deferVideoExtraction, }); if (extractResult.failureToEnforce) throw extractResult.failureToEnforce; if (extractResult.extractionResult) { @@ -1082,6 +1145,7 @@ export async function buildLocalExecutionPlan( const stagedVideoFrames = join(compiledDir, "__hyperframes_video_frames"); const videoFramesDst = join(planDir, "video-frames"); if (existsSync(videoFramesDst)) rmSync(videoFramesDst, { recursive: true, force: true }); + rmSync(join(planDir, "video-sources"), { recursive: true, force: true }); if (existsSync(stagedVideoFrames)) { renameSync(stagedVideoFrames, videoFramesDst); } else { @@ -1097,6 +1161,12 @@ export async function buildLocalExecutionPlan( // page's native `