perf(cli): cloud render downloads the finished video over parallel connections (#5864)

* perf(cli): cloud render checks render status every 2 seconds instead of 10 A finished render was noticed up to 10 s late. Polling every 2 s cuts that wait to at most 2 s and stays well under the status endpoint's rate limit. `--poll-interval` still overrides it. * perf(cli): cloud render downloads the finished video over parallel connections The first request asks for the first 8 MB. When the server answers 206 with the full size, the rest of the file is fetched in up to 7 more ranged requests at once and each part is written at its offset into the staged file. A server that ignores Range (200), a small file, or an empty file is still fetched in one request. A failed part cancels the others and the existing output is kept, as before.
2026-10-30 17:27:09 +00:00 · 2026-10-19 00:12:57 +00:00
parent d369159305
commit cddb34feac
8 changed files with 204 additions and 45 deletions
@@ -1827,7 +1827,7 @@ Lifecycle and control:
 | `--callback-id`     | Opaque tracking ID echoed in webhook payloads                                                                   |
 | `--asset-id`        | Skip zip and upload; submit an already-uploaded composition. Mutually exclusive with the project dir and `--url`. |
 | `--url`             | Submit a public HTTPS zip URL. Same mutual exclusion as `--asset-id`.                                            |
-| `--poll-interval`   | Poll cadence in seconds (default 10)                                                                            |
+| `--poll-interval`   | Poll cadence in seconds (default 2)                                                                             |
 | `--max-wait`        | Maximum poll duration in minutes (default 60)                                                                   |
 | `--idempotency-key` | `Idempotency-Key` for safe retries. 1–255 characters from `[A-Za-z0-9_:.-]`.                                     |
 | `--json`            | Machine-readable JSON instead of human progress                                                                 |
@@ -10,6 +10,8 @@ import {
  symlinkSync,
  writeFileSync,
 } from "node:fs";
+import { createServer, type Server } from "node:http";
+import type { AddressInfo } from "node:net";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
 import { afterEach, beforeEach, describe, expect, it } from "vitest";
@@ -30,6 +32,57 @@ function makeErrorFetch(status: number, statusText = "Not Found"): typeof fetch
  return (async () => new Response("nope", { status, statusText })) as unknown as typeof fetch;
 }

+interface RangeServer {
+  url: string;
+  ranges: (string | undefined)[];
+  close: () => Promise<void>;
+}
+
+async function serveBytes(
+  body: Buffer,
+  options: { honourRange: boolean; failRangeStart?: number },
+): Promise<RangeServer> {
+  const ranges: (string | undefined)[] = [];
+  const server: Server = createServer((req, res) => {
+    ranges.push(req.headers.range);
+    const match = /^bytes=(\d+)-(\d+)$/.exec(req.headers.range ?? "");
+    if (!options.honourRange || !match) {
+      res.writeHead(200, { "content-length": body.length }).end(body);
+      return;
+    }
+    const start = Number(match[1]);
+    if (start >= body.length) {
+      res.writeHead(416, { "content-range": `bytes */${body.length}` }).end();
+      return;
+    }
+    const end = Math.min(Number(match[2]), body.length - 1);
+    res.writeHead(206, {
+      "content-length": end - start + 1,
+      "content-range": `bytes ${start}-${end}/${body.length}`,
+    });
+    if (start === options.failRangeStart) {
+      res.write(body.subarray(start, start + 1024), () => res.destroy());
+      return;
+    }
+    res.end(body.subarray(start, end + 1));
+  });
+  await new Promise<void>((resolve) => server.listen(0, "127.0.0.1", resolve));
+  const { port } = server.address() as AddressInfo;
+  return {
+    url: `http://127.0.0.1:${port}/video.mp4`,
+    ranges,
+    close: () => new Promise((resolve) => server.close(() => resolve())),
+  };
+}
+
+const MiB = 1024 * 1024;
+
+function patternBytes(length: number): Buffer {
+  const bytes = Buffer.alloc(length);
+  for (let i = 0; i < length; i++) bytes[i] = (i * 31 + (i >> 13)) & 0xff;
+  return bytes;
+}
+
 describe("cloud/download", () => {
  let dir: string;
  beforeEach(() => {
@@ -221,4 +274,60 @@ describe("cloud/download", () => {
      expect(readdirSync(join(dir, "root")).sort()).toEqual(["sub"]);
    },
  );
+
+  describe("over HTTP", () => {
+    const servers: RangeServer[] = [];
+    afterEach(async () => {
+      await Promise.all(servers.splice(0).map((server) => server.close()));
+    });
+    async function serve(...args: Parameters<typeof serveBytes>) {
+      const server = await serveBytes(...args);
+      servers.push(server);
+      return server;
+    }
+
+    it("splits a large file into parallel ranged requests", async () => {
+      const body = patternBytes(60 * MiB + 123);
+      const server = await serve(body, { honourRange: true });
+      const dest = join(dir, "out.mp4");
+      const result = await downloadToFile(server.url, dest);
+      expect(result.bytes).toBe(body.length);
+      expect(readFileSync(dest).equals(body)).toBe(true);
+      expect(server.ranges).toHaveLength(8);
+      expect(server.ranges[0]).toBe(`bytes=0-${8 * MiB - 1}`);
+      expect(readdirSync(dir)).toEqual(["out.mp4"]);
+    });
+
+    it("fetches small and empty files in one request", async () => {
+      for (const size of [1000, 0]) {
+        const body = patternBytes(size);
+        const server = await serve(body, { honourRange: true });
+        const dest = join(dir, `out-${size}.mp4`);
+        await downloadToFile(server.url, dest);
+        expect(readFileSync(dest).equals(body)).toBe(true);
+        expect(server.ranges[0]).toBe(`bytes=0-${8 * MiB - 1}`);
+        expect(server.ranges.slice(1)).toEqual(size === 0 ? [undefined] : []);
+      }
+    });
+
+    it("streams the whole file in one request when the server ignores Range", async () => {
+      const body = patternBytes(20 * MiB);
+      const server = await serve(body, { honourRange: false });
+      const dest = join(dir, "out.mp4");
+      await downloadToFile(server.url, dest);
+      expect(readFileSync(dest).equals(body)).toBe(true);
+      expect(server.ranges).toEqual([`bytes=0-${8 * MiB - 1}`]);
+    });
+
+    it("keeps the previous output and leaves no staged file when one part fails", async () => {
+      const body = patternBytes(50 * MiB);
+      const server = await serve(body, { honourRange: true, failRangeStart: 8 * MiB });
+      const dest = join(dir, "out.mp4");
+      writeFileSync(dest, "previous render");
+      await expect(downloadToFile(server.url, dest)).rejects.toThrow();
+      expect(server.ranges.length).toBeGreaterThan(1);
+      expect(readFileSync(dest, "utf8")).toBe("previous render");
+      expect(readdirSync(dir)).toEqual(["out.mp4"]);
+    });
+  });
 });
@@ -13,18 +13,15 @@
 * existing destination is replaced only after the download succeeds.
 */

-import {
-  chmodSync,
-  createWriteStream,
-  mkdirSync,
-  mkdtempSync,
-  renameSync,
-  rmSync,
-  statSync,
-} from "node:fs";
+import { chmodSync, mkdirSync, mkdtempSync, renameSync, rmSync, statSync } from "node:fs";
+import { open } from "node:fs/promises";
 import { dirname, join } from "node:path";
 import { resolveWritePath } from "@hyperframes/core/atomic-file";
-import { pipeline } from "node:stream/promises";
+
+// The first request asks for PART_BYTES; a larger file is split into up to
+// CONNECTIONS ranged GETs. A server that ignores Range answers 200 and is streamed.
+const PART_BYTES = 8 * 1024 * 1024;
+const CONNECTIONS = 8;

 export interface DownloadOptions {
  signal?: AbortSignal;
@@ -54,44 +51,78 @@ export async function downloadToFile(
  options: DownloadOptions = {},
 ): Promise<DownloadResult> {
  const fetchImpl = options.fetchImpl ?? fetch;
-  const res = await fetchImpl(url, { signal: options.signal });
-  if (!res.ok) {
-    throw new Error(`Failed to download ${url}: HTTP ${res.status} ${res.statusText}`);
-  }
-  if (!res.body) {
-    throw new Error(`Failed to download ${url}: empty response body`);
+  // Cancels the sibling parts when one fails.
+  const failFast = new AbortController();
+  const signal = options.signal
+    ? AbortSignal.any([options.signal, failFast.signal])
+    : failFast.signal;
+  const get = async (range?: string) => fetchImpl(url, { signal, headers: range ? { range } : {} });
+
+  let first = await get(`bytes=0-${PART_BYTES - 1}`);
+  // 416 (empty file) or a 206 without a total cannot be split: fetch it whole.
+  if (first.status === 416 || (first.status === 206 && contentRangeTotal(first) === undefined)) {
+    await first.body?.cancel();
+    first = await get();
  }
+  assertOk(first, url);
+  const ranged = first.status === 206;
+  const total = ranged ? contentRangeTotal(first) : contentLength(first);

  mkdirSync(dirname(destPath), { recursive: true });

-  const totalHeader = res.headers.get("content-length");
-  const total = totalHeader ? Number.parseInt(totalHeader, 10) : undefined;
-  const totalOpt = total !== undefined && Number.isFinite(total) ? total : undefined;
-
  // Writes through a caller-selected symlink, as before.
  const destination = resolveWritePath(destPath);
  const previous = statSync(destination, { throwIfNoEntry: false });
  const stage = mkdtempSync(join(dirname(destination), ".hf-download-"));
  const stagedFile = join(stage, "download");
  let bytes = 0;
+  let firstError: unknown;
  try {
-    await pipeline(
-      async function* () {
-        for await (const chunk of res.body as unknown as AsyncIterable<Uint8Array>) {
-          bytes += chunk.byteLength;
-          options.onProgress?.(bytes, totalOpt);
-          yield chunk;
-        }
-        if (totalOpt !== undefined && bytes !== totalOpt) {
-          throw new Error(
-            `Truncated download: got ${bytes} bytes, expected ${totalOpt} (content-length). ` +
-              `The presigned URL may have expired mid-transfer — refetch via \`hyperframes cloud get\`.`,
-          );
-        }
-      },
-      createWriteStream(stagedFile, { flags: "wx" }),
-      { signal: options.signal },
+    const file = await open(stagedFile, "wx");
+    const writePart = async (res: Response, start: number, length: number | undefined) => {
+      let written = 0;
+      for await (const chunk of res.body as unknown as AsyncIterable<Uint8Array>) {
+        await file.write(chunk, 0, chunk.byteLength, start + written);
+        written += chunk.byteLength;
+        bytes += chunk.byteLength;
+        signal.throwIfAborted();
+        options.onProgress?.(bytes, total);
+      }
+      if (length !== undefined && written !== length) {
+        throw new Error(
+          `Truncated download: got ${written} bytes, expected ${length} (content-length). ` +
+            `The presigned URL may have expired mid-transfer — refetch via \`hyperframes cloud get\`.`,
+        );
+      }
+    };
+    const parts: Promise<void>[] = [];
+    if (ranged && total !== undefined) {
+      parts.push(writePart(first, 0, Math.min(PART_BYTES, total)));
+      const rest = total - PART_BYTES;
+      const size = Math.ceil(rest / Math.min(CONNECTIONS - 1, Math.ceil(rest / PART_BYTES)));
+      for (let start = PART_BYTES; start < total; start += size) {
+        const end = Math.min(start + size, total) - 1;
+        parts.push(
+          get(`bytes=${start}-${end}`).then((res) => {
+            assertOk(res, url);
+            if (res.status !== 206) throw new Error(`Failed to download ${url}: range ignored`);
+            return writePart(res, start, end - start + 1);
+          }),
+        );
+      }
+    } else {
+      parts.push(writePart(first, 0, total));
+    }
+    await Promise.all(
+      parts.map((part) =>
+        part.catch((error: unknown) => {
+          firstError ??= error;
+          failFast.abort(error);
+        }),
+      ),
    );
+    await file.close();
+    if (firstError !== undefined) throw firstError;
    options.signal?.throwIfAborted();
    if (previous?.isFile()) chmodSync(stagedFile, previous.mode & 0o777);
    renameSync(stagedFile, destination);
@@ -104,3 +135,22 @@ export async function downloadToFile(
  }
  return { path: destPath, bytes };
 }
+
+function assertOk(res: Response, url: string): void {
+  if (!res.ok) {
+    throw new Error(`Failed to download ${url}: HTTP ${res.status} ${res.statusText}`);
+  }
+  if (!res.body) {
+    throw new Error(`Failed to download ${url}: empty response body`);
+  }
+}
+
+function contentLength(res: Response): number | undefined {
+  const total = Number.parseInt(res.headers.get("content-length") ?? "", 10);
+  return Number.isFinite(total) ? total : undefined;
+}
+
+function contentRangeTotal(res: Response): number | undefined {
+  const match = /^bytes \d+-\d+\/(\d+)$/.exec(res.headers.get("content-range") ?? "");
+  return match ? Number(match[1]) : undefined;
+}
@@ -43,8 +43,8 @@ describe("cloud/poll", () => {
  });

  describe("defaults", () => {
-    it("matches the documented 10s / 60min defaults", () => {
-      expect(DEFAULT_POLL_INTERVAL_MS).toBe(10_000);
+    it("matches the documented 2s / 60min defaults", () => {
+      expect(DEFAULT_POLL_INTERVAL_MS).toBe(2_000);
      expect(DEFAULT_MAX_WAIT_MS).toBe(60 * 60 * 1000);
    });
  });
@@ -1,8 +1,8 @@
 /**
 * Poll a HyperFrames render until it reaches a terminal state.
 *
- * Defaults: 10s interval, 60min cap — confirmed with the API owner. The
- * `start_to_close` timeout on the underlying Temporal workflow is the
+ * Defaults: 2s interval (well under the status GET's per-user rate limit), 60min cap.
+ * The `start_to_close` timeout on the underlying Temporal workflow is the
 * same order of magnitude, so polling longer would only hide a stuck
 * workflow rather than recover from one.
 *
@@ -25,7 +25,7 @@ export interface PollOptions {
  signal?: AbortSignal;
 }

-export const DEFAULT_POLL_INTERVAL_MS = 10_000;
+export const DEFAULT_POLL_INTERVAL_MS = 2_000;
 export const DEFAULT_MAX_WAIT_MS = 60 * 60 * 1000;

 const TERMINAL_STATUSES: ReadonlySet<HyperframesRenderStatus> = new Set(["completed", "failed"]);
@@ -16,7 +16,7 @@ import { failCommand } from "../../utils/commandResult.js";
 *      `project: {type:"asset_id", asset_id}` shape.
 *   5. If `--no-wait`: print the `render_id` and exit immediately.
 *      Otherwise poll `GET /v3/hyperframes/renders/{id}` every
- *      `--poll-interval` (default 10s, max 60min). `--callback-url`
+ *      `--poll-interval` (default 2s, max 60min). `--callback-url`
 *      can be combined with either mode: the webhook always fires when
 *      the server-side render terminates, independent of whether the
 *      CLI is still polling.
@@ -30,7 +30,7 @@
      "files": 8
    },
    "hyperframes-cli": {
-      "hash": "3c8d1fac9e0a02fe",
+      "hash": "35688a15a6d0617b",
      "files": 11
    },
    "hyperframes-core": {
@@ -137,7 +137,7 @@ npx hyperframes cloud render --callback-url https://example.com/hf-hook --no-wai
 | `--no-wait`       | Submit and exit immediately; print the `render_id`. |
 | `--callback-url`  | HTTPS webhook fired when the render terminates.     |
 | `--callback-id`   | Opaque tracking ID echoed in webhook payloads.      |
-| `--poll-interval` | Poll cadence in seconds (default `10`).             |
+| `--poll-interval` | Poll cadence in seconds (default `2`).              |
 | `--max-wait`      | Max poll duration in minutes (default `60`).        |

 ## Managing renders