diff --git a/packages/cli/src/cloud/offload.progress.test.ts b/packages/cli/src/cloud/offload.progress.test.ts new file mode 100644 index 00000000000..ab7c7059a17 --- /dev/null +++ b/packages/cli/src/cloud/offload.progress.test.ts @@ -0,0 +1,53 @@ +import { execFileSync } from "node:child_process"; +import { mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterAll, describe, expect, it } from "vitest"; +import { findFfBinary } from "@hyperframes/parsers/ff-binaries"; +import { buildPublishFileMap } from "../utils/publishProject.js"; +import { OFFLOAD_MIN_BYTES, offloadHeavyMedia, type ReencodeProgress } from "./offload.js"; + +const ffmpeg = findFfBinary("ffmpeg"); +const ffprobe = findFfBinary("ffprobe"); +const dir = mkdtempSync(join(tmpdir(), "hf-offload-progress-")); +afterAll(() => rmSync(dir, { recursive: true, force: true })); + +/** Two seconds of video with sound, past the offload cutoff, under a name the asset API refuses. */ +function clipNeedingReencode(project: string): void { + const made = join(dir, "made.mov"); + execFileSync(ffmpeg!, [ + ...["-v", "error", "-y", "-f", "lavfi", "-i", "testsrc2=s=160x90:d=2"], + ...["-f", "lavfi", "-i", "sine=f=440:d=2"], + ...["-c:v", "libx264", "-pix_fmt", "yuv420p", "-c:a", "aac", "-shortest", made], + ]); + const free = Buffer.alloc(OFFLOAD_MIN_BYTES); + free.writeUInt32BE(free.length); + free.write("free", 4); + writeFileSync(join(project, "take.mkv"), Buffer.concat([readFileSync(made), free])); +} + +describe.skipIf(!ffmpeg || !ffprobe)("a re-encode before upload", () => { + it("reports the output seconds it has made, against the clip's length", async () => { + const project = mkdtempSync(join(dir, "project-")); + writeFileSync( + join(project, "index.html"), + ``, + ); + clipNeedingReencode(project); + const heard: ReencodeProgress[] = []; + + const [item] = await offloadHeavyMedia(project, buildPublishFileMap(project), { + apiBaseUrl: "https://api.example", + cachePath: join(project, "cache.json"), + upload: async () => ({ asset_id: "a", url: "https://assets.example/a.mp4" }), + onReencodeProgress: (event) => heard.push(event), + }); + + expect(item).toMatchObject({ action: "reencode" }); + expect(heard.length).toBeGreaterThan(0); + const lastHeard = heard.at(-1)!; + expect(lastHeard.path).toBe("take.mkv"); + expect(lastHeard.doneSeconds).toBeGreaterThan(1.5); + expect(lastHeard.totalSeconds).toBeCloseTo(2, 0); + }); +}); diff --git a/packages/cli/src/cloud/offload.test.ts b/packages/cli/src/cloud/offload.test.ts index f515e222f77..b6f3bd46308 100644 --- a/packages/cli/src/cloud/offload.test.ts +++ b/packages/cli/src/cloud/offload.test.ts @@ -14,6 +14,7 @@ import { tmpdir } from "node:os"; import { dirname, extname, join } from "node:path"; import { buildPublishFileMap, stagePublishFiles } from "../utils/publishProject.js"; import { + CACHE_CHECK_TIMEOUT_MS, type GifPrep, OFFLOAD_MIN_BYTES, offloadHeavyMedia, @@ -305,6 +306,43 @@ describe("offloadHeavyMedia", () => { expect(report[0]).toMatchObject({ action: "upload", url: "https://assets.example/2/clip.mp4" }); }); + it("uploads again when the cached asset URL never answers", async () => { + const dir = project({ + "index.html": html(``), + "clip.mp4": HEAVY, + }); + const opts = { + apiBaseUrl: "https://api.example", + upload: fakeUpload(), + cachePath: cachePath(), + }; + await offloadHeavyMedia(dir, buildPublishFileMap(dir), opts); + const silentHead = vi.fn( + (_url: string, init?: RequestInit) => + new Promise((_resolve, reject) => + init?.signal?.addEventListener("abort", () => reject(init.signal!.reason)), + ), + ) as unknown as typeof fetch; + + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] }); + try { + const pending = offloadHeavyMedia(dir, buildPublishFileMap(dir), { + ...opts, + fetchImpl: silentHead, + }); + while (vi.mocked(silentHead).mock.calls.length === 0) { + await new Promise((resolve) => setImmediate(resolve)); + } + await vi.advanceTimersByTimeAsync(CACHE_CHECK_TIMEOUT_MS); + const report = await pending; + + expect(report[0]).toMatchObject({ action: "upload" }); + expect(opts.upload).toHaveBeenCalledTimes(2); + } finally { + vi.useRealTimers(); + } + }); + it("re-encodes media the asset API refuses and uploads the accepted format", async () => { const dir = project({ "index.html": html(``), @@ -328,7 +366,12 @@ describe("offloadHeavyMedia", () => { cachePath: cachePath(), }); - expect(reencoder.run).toHaveBeenCalledWith(staged("take.mkv"), "video", undefined); + expect(reencoder.run).toHaveBeenCalledWith( + staged("take.mkv"), + "video", + undefined, + undefined, + ); expect(upload.mock.calls[0]![0]).toMatchObject({ filename: "take.mp4", contentType: "video/mp4", @@ -600,7 +643,12 @@ describe("offloadHeavyMedia", () => { cachePath: cachePath(), }); - expect(run).toHaveBeenCalledWith(join(dir, "take.mov"), "video", undefined); + expect(run).toHaveBeenCalledWith( + join(dir, "take.mov"), + "video", + undefined, + undefined, + ); expect(item).toMatchObject({ path: "take.mov", action: "reencode", size_bytes: 2 ** 32 + 1 }); expect(fileMap.has("take.mov")).toBe(false); }, 60_000); @@ -694,10 +742,15 @@ describe("offloadHeavyMedia", () => { cachePath: cachePath(), }); - expect(run).toHaveBeenCalledWith(staged("long.mkv"), "video", [ - { start: 9.5, end: 12.5 }, - { start: 100, end: 105 }, - ]); + expect(run).toHaveBeenCalledWith( + staged("long.mkv"), + "video", + [ + { start: 9.5, end: 12.5 }, + { start: 100, end: 105 }, + ], + undefined, + ); expect(report[0]).toMatchObject({ path: "long.mkv", trimmed_to_seconds: 8 }); const index = fileMap.get("index.html")!.toString(); expect(index).toMatch(/id="a"[^>]*data-media-start="0.5"/); @@ -823,7 +876,12 @@ describe("offloadHeavyMedia", () => { reencoder: { choose: async () => "video", trimGrid: async () => 30, run }, cachePath: cachePath(), }); - expect(run).toHaveBeenCalledWith(staged("long.mkv"), "video", undefined); + expect(run).toHaveBeenCalledWith( + staged("long.mkv"), + "video", + undefined, + undefined, + ); expect(fileMap.get("index.html")!.toString()).toContain(`data-media-start="10"`); } }); diff --git a/packages/cli/src/cloud/offload.ts b/packages/cli/src/cloud/offload.ts index cb9952135d2..e03c33d3441 100644 --- a/packages/cli/src/cloud/offload.ts +++ b/packages/cli/src/cloud/offload.ts @@ -3,7 +3,7 @@ * staged copy's references at the asset URLs. The project on disk is never touched. */ -import { execFile } from "node:child_process"; +import { execFile, spawn } from "node:child_process"; import { createHash } from "node:crypto"; import { closeSync, @@ -38,6 +38,7 @@ import { CONVERTIBLE_IMAGE_EXTS, VIDEO_EXTS, } from "@hyperframes/studio-server/outside-file"; +import { followFfmpegProgress } from "@hyperframes/parsers/ff-binaries"; import { type PlayedSegment, planTrim } from "./offloadTrim.js"; import { archiveKey, @@ -60,6 +61,7 @@ export const OFFLOAD_MIN_BYTES = 5 * 1024 * 1024; const MiB = 1024 * 1024; const AUDIO_MAX_BYTES = 1_500_000_000; +export const CACHE_CHECK_TIMEOUT_MS = 10_000; // What POST /v3/assets/direct-uploads takes as-is, and its per-type cap. const ACCEPTED: Record = { @@ -144,9 +146,17 @@ export interface OffloadOptions { /** Overrides the renderer's GIF preparation for tests. */ gifPrep?: GifPrep; onFile?: (item: OffloadedMedia) => void; + /** Hears a re-encode's output seconds made so far; `totalSeconds` is null when the length is unknown. */ + onReencodeProgress?: (event: ReencodeProgress) => void; variablesText?: string; } +export interface ReencodeProgress { + path: string; + doneSeconds: number; + totalSeconds: number | null; +} + type MediaKind = "video" | "audio" | "image"; export interface Reencoder { @@ -160,6 +170,7 @@ export interface Reencoder { sourcePath: string, preset: PresetName, segments?: PlayedSegment[], + onProgress?: (doneSeconds: number, totalSeconds: number | null) => void, ) => Promise; } @@ -202,11 +213,15 @@ function writeCache(path: string, cache: Record): void { // Asset URLs are public and unsigned, so HEAD needs no auth; a deleted or replaced asset fails it. async function stillServed(entry: CacheEntry, fetchImpl: typeof fetch): Promise { + const controller = new AbortController(); + const timer = setTimeout(() => controller.abort(), CACHE_CHECK_TIMEOUT_MS); try { - const res = await fetchImpl(entry.url, { method: "HEAD" }); + const res = await fetchImpl(entry.url, { method: "HEAD", signal: controller.signal }); return res.ok && Number(res.headers.get("content-length")) === entry.size_bytes; } catch { return false; + } finally { + clearTimeout(timer); } } @@ -364,6 +379,9 @@ const capOf = (presetName: PresetName) => ACCEPTED[REENCODE_PRESETS[presetName]. const playedSeconds = (segments: PlayedSegment[]) => segments.reduce((sum, s) => sum + s.end - s.start, 0); +const outputSeconds = (sourcePath: string, segments: PlayedSegment[] | undefined) => + segments?.length ? Promise.resolve(playedSeconds(segments)) : durationSeconds(sourcePath); + export async function limitFor( sourcePath: string, presetName: PresetName, @@ -371,7 +389,7 @@ export async function limitFor( maxBytes = capOf(presetName), ): Promise { if (mediaOf(presetName) !== "video") return []; - const seconds = segments?.length ? playedSeconds(segments) : await durationSeconds(sourcePath); + const seconds = await outputSeconds(sourcePath, segments); const limit = sizeLimitArgs({ maxBytes, seconds: seconds ?? 0, @@ -381,10 +399,36 @@ export async function limitFor( return presetName === "alphaVideo" && limit.length ? [...limit, "-b:v", limit[1]!] : limit; } +/** ffmpeg with `-progress` on stdout, so `onSeconds` hears the output time made; fails with ffmpeg's own stderr. */ +function runFfmpeg( + ffmpeg: string, + args: string[], + onSeconds: (seconds: number) => void, +): Promise { + return new Promise((resolve, reject) => { + const proc = spawn(ffmpeg, ["-progress", "pipe:1", "-nostats", ...args], { + stdio: ["ignore", "pipe", "pipe"], + windowsHide: true, + }); + followFfmpegProgress(proc.stdout, onSeconds); + let stderr = ""; + proc.stderr.on("data", (chunk: Buffer) => { + stderr = (stderr + chunk.toString()).slice(-16_384); + }); + proc.on("error", reject); + proc.on("close", (code, signal) => { + if (code === 0) return resolve(); + const ended = signal ? `was ended by ${signal}` : `exited with code ${code}`; + reject(new Error(`ffmpeg ${ended}: ${stderr.trim()}`)); + }); + }); +} + async function ffmpegReencode( sourcePath: string, presetName: PresetName, segments?: PlayedSegment[], + onProgress?: (doneSeconds: number, totalSeconds: number | null) => void, ): Promise { const ffmpeg = findFFmpeg(); if (!ffmpeg) throw new Error(`Re-encoding needs ffmpeg. Install it: ${getFFmpegInstallHint()}`); @@ -392,10 +436,11 @@ async function ffmpegReencode( const probe = preset.contentType.startsWith("video/") ? await probeVideo(sourcePath) : undefined; const decoder = probe ? alphaDecoderArgs(probe) : []; const color = presetName === "hdrVideo" && probe ? hdrColorArgs(probe) : []; + const totalSeconds = onProgress ? await outputSeconds(sourcePath, segments) : null; const workDir = mkdtempSync(join(tmpdir(), "hf-offload-")); const outPath = join(workDir, `${basename(sourcePath, extname(sourcePath))}${preset.ext}`); try { - await execFileAsync( + await runFfmpeg( ffmpeg, [ "-v", @@ -408,7 +453,7 @@ async function ffmpegReencode( ...preset.audio, outPath, ], - { maxBuffer: 16 * MiB }, + (doneSeconds) => onProgress?.(doneSeconds, totalSeconds), ); // Never read an output the asset API would refuse: it can exceed what one read can hold. return { @@ -700,8 +745,14 @@ async function uploadPlanned( plan: FilePlan, ): Promise<{ url: string; asset_id: string }> { const filename = basename(path); + const report = ctx.opts.onReencodeProgress; const file: UploadableFile = plan.preset - ? await ctx.reencoder.run(plan.sourcePath, plan.preset, plan.segments) + ? await ctx.reencoder.run( + plan.sourcePath, + plan.preset, + plan.segments, + report && ((doneSeconds, totalSeconds) => report({ path, doneSeconds, totalSeconds })), + ) : Buffer.isBuffer(staged) ? { bytes: staged, filename, contentType: plan.contentType! } : { path: staged.absPath, size: staged.size, filename, contentType: plan.contentType! }; diff --git a/packages/cli/src/commands/cloud/render.test.ts b/packages/cli/src/commands/cloud/render.test.ts index 050899f65f1..9dc9944d19b 100644 --- a/packages/cli/src/commands/cloud/render.test.ts +++ b/packages/cli/src/commands/cloud/render.test.ts @@ -10,6 +10,7 @@ import { logsTick, resolveCloudFpsArg, parseAspectRatioFlag, + reencodeProgressLines, resolveAspectRatioForSubmit, validateDryRunSource, type ProjectInputSource, @@ -410,3 +411,53 @@ describe("resolveAspectRatioForSubmit — local dir", () => { expect(output).toContain(`project directory ${resolve(dir)}`); }); }); + +describe("reencodeProgressLines", () => { + it("writes one JSON line on stderr per whole percent, and per second when the length is unknown", () => { + const lines: string[] = []; + vi.spyOn(console, "error").mockImplementation((line: string) => void lines.push(line)); + let clock = 0; + const report = reencodeProgressLines(() => clock); + + report({ path: "take.mkv", doneSeconds: 0.5, totalSeconds: 100 }); + report({ path: "take.mkv", doneSeconds: 0.9, totalSeconds: 100 }); + report({ path: "take.mkv", doneSeconds: 1.2, totalSeconds: 100 }); + report({ path: "raw.mxf", doneSeconds: 3, totalSeconds: null }); + clock = 400; + report({ path: "raw.mxf", doneSeconds: 3.5, totalSeconds: null }); + clock = 1200; + report({ path: "raw.mxf", doneSeconds: 4, totalSeconds: null }); + + expect(lines.map((line) => JSON.parse(line))).toEqual([ + { + event: "reencode_progress", + file: "take.mkv", + percent: 0, + done_seconds: 0.5, + total_seconds: 100, + }, + { + event: "reencode_progress", + file: "take.mkv", + percent: 1, + done_seconds: 1.2, + total_seconds: 100, + }, + { + event: "reencode_progress", + file: "raw.mxf", + percent: null, + done_seconds: 3, + total_seconds: null, + }, + { + event: "reencode_progress", + file: "raw.mxf", + percent: null, + done_seconds: 4, + total_seconds: null, + }, + ]); + vi.restoreAllMocks(); + }); +}); diff --git a/packages/cli/src/commands/cloud/render.ts b/packages/cli/src/commands/cloud/render.ts index c072d54db2c..d1b08af8f8f 100644 --- a/packages/cli/src/commands/cloud/render.ts +++ b/packages/cli/src/commands/cloud/render.ts @@ -788,6 +788,7 @@ async function offloadMedia( variablesText, onFile: (item) => (asJson ? console.error : console.log)(c.dim(` ${describeOffload(item, !upload)}`)), + onReencodeProgress: asJson ? reencodeProgressLines() : undefined, }); } catch (err) { reportApiError("Media upload failed", err); @@ -829,6 +830,32 @@ export function assertArchiveFits(sizeBytes: number, largestFiles: ArchiveFileSu failCommand(); } +/** Under --json, a line on stderr per whole percent a re-encode makes (per second when its length is unknown), + * so a caller can tell a long encode from a dead one. */ +export function reencodeProgressLines( + now: () => number = Date.now, +): NonNullable { + let last = { path: "", percent: -1, at: 0 }; + return ({ path, doneSeconds, totalSeconds }) => { + const percent = totalSeconds + ? Math.min(100, Math.floor((doneSeconds / totalSeconds) * 100)) + : null; + const at = now(); + const same = path === last.path; + if (same && (percent === null ? at - last.at < 1000 : percent === last.percent)) return; + last = { path, percent: percent ?? -1, at }; + console.error( + JSON.stringify({ + event: "reencode_progress", + file: path, + percent, + done_seconds: Math.round(doneSeconds * 10) / 10, + total_seconds: totalSeconds === null ? null : Math.round(totalSeconds * 10) / 10, + }), + ); + }; +} + /** A line per PUT progress event: JSON on stderr under --json, so a caller can tell a slow upload from a dead one. */ function reportUploadProgress(name: string, asJson: boolean): (event: UploadProgressEvent) => void { return (event) => { diff --git a/packages/parsers/src/ffBinaries.ts b/packages/parsers/src/ffBinaries.ts index 68cc47a9873..2dda2fe81b8 100644 --- a/packages/parsers/src/ffBinaries.ts +++ b/packages/parsers/src/ffBinaries.ts @@ -9,6 +9,7 @@ import { } from "node:fs"; import { createRequire } from "node:module"; import { basename, dirname, join, resolve } from "node:path"; +import type { Readable } from "node:stream"; import { isTrustedExecutablePath, trustedPathDirs } from "./trustedExecutable.js"; /** @@ -243,3 +244,23 @@ export function clearFfBinaryLookupCache(): void { lastMissAt.clear(); skippedX86Binaries.clear(); } + +const OUT_TIME = /^out_time_(?:us|ms)=(\d+)$/; + +/** Hears the furthest output time, in seconds, that ffmpeg's `-progress pipe:1` lines on `stream` report. */ +export function followFfmpegProgress(stream: Readable, onSeconds: (seconds: number) => void): void { + let furthest = 0; + let lineTail = ""; + stream.on("data", (chunk: Buffer) => { + const lines = (lineTail + chunk.toString()).split("\n"); + lineTail = lines.pop() ?? ""; + for (const line of lines) { + const reached = OUT_TIME.exec(line.trim()); + if (!reached) continue; + const seconds = Number(reached[1]) / 1e6; + if (seconds <= furthest) continue; + furthest = seconds; + onSeconds(seconds); + } + }); +} diff --git a/packages/studio-server/src/helpers/proxyTranscoder.test.ts b/packages/studio-server/src/helpers/proxyTranscoder.test.ts index 6cf66981659..e8a54cbe4f5 100644 --- a/packages/studio-server/src/helpers/proxyTranscoder.test.ts +++ b/packages/studio-server/src/helpers/proxyTranscoder.test.ts @@ -169,7 +169,8 @@ async function loadModule( const mocked = { spawn }; return { ...mocked, default: mocked }; }); - vi.doMock("@hyperframes/parsers/ff-binaries", () => ({ + vi.doMock("@hyperframes/parsers/ff-binaries", async (importOriginal) => ({ + ...(await importOriginal()), findFfBinary: () => ffmpegPath, })); vi.doMock("./mediaMetadata.js", () => ({ @@ -1774,7 +1775,10 @@ describe("resolveProxy", () => { const mocked = { spawn }; return { ...mocked, default: mocked }; }); - vi.doMock("@hyperframes/parsers/ff-binaries", () => ({ findFfBinary: ffmpegPath })); + vi.doMock("@hyperframes/parsers/ff-binaries", async (importOriginal) => ({ + ...(await importOriginal()), + findFfBinary: ffmpegPath, + })); vi.doMock("./mediaMetadata.js", () => ({ probeMediaMetadata: () => new Promise((resolveProbe) => setTimeout(resolveProbe, 5)).then(probe), diff --git a/packages/studio-server/src/helpers/proxyTranscoder.ts b/packages/studio-server/src/helpers/proxyTranscoder.ts index 559bcc2219e..8448a01ca5d 100644 --- a/packages/studio-server/src/helpers/proxyTranscoder.ts +++ b/packages/studio-server/src/helpers/proxyTranscoder.ts @@ -12,7 +12,7 @@ import { import os, { availableParallelism, tmpdir, totalmem } from "node:os"; import { basename, dirname, extname, isAbsolute, join, relative, resolve, sep } from "node:path"; import { hdrToSdrToneMapFilter, type PreviewProxyBox } from "@hyperframes/core"; -import { findFfBinary } from "@hyperframes/parsers/ff-binaries"; +import { findFfBinary, followFfmpegProgress } from "@hyperframes/parsers/ff-binaries"; import { alphaDecoderFor } from "@hyperframes/parsers/media-alpha"; import { ffmpegHasHdrFilters } from "./hdrFilters.js"; import { probeFirstFrameColour, probeMediaMetadata } from "./mediaMetadata.js"; @@ -814,9 +814,6 @@ function concatArgs(listPath: string, target: string, variant: ProxyVariant): st ]; } -/** ffmpeg's `-progress` lines name the output time reached, in microseconds. */ -const OUT_TIME = /^out_time_(?:us|ms)=(\d+)$/; - /** Runs ffmpeg and ends it when it makes no progress for `stallMs`; a slow encode that moves on is left to finish. * `onProgress` hears the seconds of output made so far (from `-progress pipe:1`). */ function lowerPriorityUnlessExited(pid: number): void { @@ -855,21 +852,12 @@ function spawnFfmpeg( timer.unref(); }; armStall(); - let furthest = 0; - let lineTail = ""; - proc.stdout?.on("data", (chunk: Buffer) => { - const lines = (lineTail + chunk.toString()).split("\n"); - lineTail = lines.pop() ?? ""; - for (const line of lines) { - const reached = OUT_TIME.exec(line.trim()); - if (!reached) continue; - const seconds = Number(reached[1]) / 1e6; - if (seconds <= furthest) continue; - furthest = seconds; + if (proc.stdout) { + followFfmpegProgress(proc.stdout, (seconds) => { armStall(); onProgress?.(seconds); - } - }); + }); + } let stderrTail = ""; proc.stderr?.on("data", (chunk: Buffer) => { stderrTail = (stderrTail + chunk.toString()).slice(-STDERR_TAIL_MAX_CHARS);