From 9e737ee0a7d993ff5b3fa58e386ec336a3e09350 Mon Sep 17 00:00:00 2001 From: pallaoro Date: Sat, 10 Oct 2026 17:34:18 +0200 Subject: [PATCH] Footage: a file Drive won't hand over whole comes in a piece at a time Drive's per-file download limit answers a request for the whole file with its "Quota exceeded" page, but still serves byte ranges of the same file. Such a file used to wait up to a day. Now, up to the video host's 30 GB, it is read in 200 MiB ranges from its shared link and appended to a resumable upload on the media service, a few pieces per background delivery (never on a read). The upload's own offset is the cursor, so a piece that didn't land, a wait on Drive or two overlapping steps carry on where the upload stands. An upload about to expire starts again; a piece the host refuses outright fails the clip with its reason; deleting the project deletes an upload under way. Each clip shows how much is in. --- agent.md | 20 ++-- schema.sql | 11 ++ src/client/edit.tsx | 5 + src/server/footage.ts | 181 +++++++++++++++++++++++++++---- src/server/index.ts | 27 ++++- src/server/media.ts | 6 +- src/server/relay.ts | 211 ++++++++++++++++++++++++++++++++++++ test/e2e/run.mjs | 221 +++++++++++++++++++++++++++++++++++-- test/relay.test.ts | 246 ++++++++++++++++++++++++++++++++++++++++++ 9 files changed, 889 insertions(+), 39 deletions(-) create mode 100644 src/server/relay.ts create mode 100644 test/relay.test.ts diff --git a/agent.md b/agent.md index 326b07e..879a6ce 100644 --- a/agent.md +++ b/agent.md @@ -82,6 +82,7 @@ otherwise), `next_offset`, and `items`: { "id": "…", "name": "B_0012.MP4", "folder": "Day 1/Cam B", "status": "ready", // waiting | importing | ready | failed + "received": null, // while it comes in a piece at a time: { "bytes", "size" } "asset": { "id": "…", "duration": 41.2 }, // the EDL uses "asset:" "log_status": "done", // preparing | running | done | failed "log": { @@ -149,19 +150,24 @@ each file's Drive link. cut from the kept soundbites" places them with its `add_highlight` operation. Google Drive limits how often a file shared with the link can be downloaded, -and on a heavy day it refuses the files of a whole folder for a while. A clip it -refuses is not failed: it waits (`status: "waiting"` with `retry_at`, counted in +and on a heavy day it refuses the files of a whole folder for a while. It +refuses the whole file but still serves pieces of it, so a refused file of up +to 30 GB comes in a piece at a time instead, in the background steps only: +`status: "importing"`, with `received` saying how much is in. A big file takes +a while this way. A clip Drive won't serve even a piece of, or one over 30 GB, +is not failed: it waits (`status: "waiting"` with `retry_at`, counted in `counts.drive_waiting`) and is tried again by itself, further apart each time, -for about a day, after which it fails with the reason. `imports_paused` says so -when every waiting clip is waiting on Drive. A copy of the folder in another -Drive account has its own limit. +for about a day, after which it fails with the reason. One that was coming in +by pieces carries on from where it got to. `imports_paused` says so when every +waiting clip is waiting on Drive. A copy of the folder in another Drive account +has its own limit. When the org has a Google Drive connection whose account can open the folder, clips are downloaded through it instead, as that account, which Drive's limit on shared links doesn't touch. Those downloads run in the background steps only. A file too big for the connection's temporary storage -(seen at 9 GB; 3.4 GB passed) comes by the original's shared link, with the -waits above. It is never copied inside Google first: Drive answers a header +(seen at 9 GB; 3.4 GB passed) comes by the original's shared link, as above: +whole, or a piece at a time when Drive refuses it whole. It is never copied inside Google first: Drive answers a header check on a fresh copy of a big file with an empty page, and the video host refuses such a download. Copies an earlier version made are deleted once their import is over (or with the project). diff --git a/schema.sql b/schema.sql index 9193c5f..8b51b8b 100644 --- a/schema.sql +++ b/schema.sql @@ -219,3 +219,14 @@ ALTER TABLE project_footage ADD COLUMN copy_id TEXT; -- other refusal sends it back to be re-encoded. ALTER TABLE project_footage ADD COLUMN transcode INTEGER NOT NULL DEFAULT 0; ALTER TABLE project_footage ADD COLUMN transcode_job TEXT; + +-- A file Google Drive won't hand over whole (its download limit) still serves +-- byte ranges, so it comes in a piece at a time, into an upload on the media +-- service (src/server/relay.ts). Set while that upload is open: its id and +-- link, when the link expires, the file's size, and how much is in so far +-- (for showing; the upload itself says where it stands). +ALTER TABLE project_footage ADD COLUMN upload_uid TEXT; +ALTER TABLE project_footage ADD COLUMN upload_url TEXT; +ALTER TABLE project_footage ADD COLUMN upload_expires TEXT; +ALTER TABLE project_footage ADD COLUMN upload_size INTEGER; +ALTER TABLE project_footage ADD COLUMN upload_done INTEGER; diff --git a/src/client/edit.tsx b/src/client/edit.tsx index e43c465..2ec8199 100644 --- a/src/client/edit.tsx +++ b/src/client/edit.tsx @@ -149,6 +149,8 @@ interface FootageItem { error: string | null; /** Set while it waits on Google Drive: when it is tried again. */ retry_at?: string | null; + /** Set while a file Drive won't hand over whole comes in a piece at a time. */ + received?: { bytes: number; size: number } | null; asset: Asset | null; log_status: "preparing" | "running" | "done" | "failed" | null; log_error: string | null; @@ -2055,6 +2057,9 @@ function footageLine(i: FootageItem): string { return `Waiting for Google Drive, which is limiting downloads of this file. Trying again at ${at}`; } if (i.status === "waiting") return "Waiting to import"; + if (i.status === "importing" && i.received) { + return `Importing: ${Math.floor((i.received.bytes / i.received.size) * 100)}% of ${fmtBytes(i.received.size)}`; + } if (i.status === "importing") return "Importing"; if (i.status === "failed") return `Not imported: ${i.error ?? "unknown error"}`; if (i.log_status === "done" && i.log) return i.log.summary; diff --git a/src/server/footage.ts b/src/server/footage.ts index aa310c2..86d2a50 100644 --- a/src/server/footage.ts +++ b/src/server/footage.ts @@ -4,7 +4,9 @@ // A shoot arrives as a shared folder: a folder per camera, a few hundred clips // a day. Every video in it becomes a row of project_footage, which moves on by // itself: -// waiting → importing (the media service pulls it from Drive) → ready +// waiting → importing (the media service pulls it from Drive, or, for a +// file Drive won't hand over whole, this app sends it a piece at +// a time: src/server/relay.ts) → ready // and once ready, its log: // preparing (the platform makes the copy analysis reads) → running → done // @@ -16,9 +18,10 @@ import { enqueueJob, type QueueEnv } from "@clawnify/queue"; import { query, get, run } from "./db"; -import { deleteMedia, importMedia, mediaState, prepareMedia, startTranscode, transcodeState, type MediaConfig } from "./media"; +import { deleteMedia, importMedia, mediaState, openMediaUpload, prepareMedia, startTranscode, transcodeState, type MediaConfig } from "./media"; import { refusalDetail } from "./refusal"; import { directDownloadUrl, folderListingUrl, judgeLinkResponse, listFolderVideos, type FolderVideo } from "./drive-link"; +import { MAX_RELAY_BYTES, RELAY_AT_ONCE, RELAY_BUDGET_MS, RELAY_EXPIRY_MARGIN_MS, rangedSize, relayPieces } from "./relay"; const DEFAULT_SERVICES_URL = "https://services.clawnify.com"; @@ -324,6 +327,10 @@ interface WorkRow { copy_id: string | null; transcode: number; transcode_job: string | null; + upload_uid: string | null; + upload_url: string | null; + upload_expires: string | null; + upload_size: number | null; } /** @@ -354,6 +361,8 @@ export interface StepOptions { drive?: DriveSource; /** False: start no imports this step (a read, while imports go through the connection, which is slow). */ startImports?: boolean; + /** Move files coming in a piece at a time a few pieces on: deliveries only, as it takes a while. */ + relay?: boolean; } export interface StepOutcome { @@ -409,7 +418,8 @@ export async function stepFootage(cfg: MediaConfig, projectId: string, opts: Ste } const rows = await query( `SELECT f.id, f.status, f.drive_file_id, f.name, f.folder, f.language, f.asset_id, f.log_status, f.log_job, - f.updated_at, f.retry_at, f.drive_tries, f.link_only, f.copy_id, f.transcode, f.transcode_job, a.media_uid, a.duration + f.updated_at, f.retry_at, f.drive_tries, f.link_only, f.copy_id, f.transcode, f.transcode_job, + f.upload_uid, f.upload_url, f.upload_expires, f.upload_size, a.media_uid, a.duration FROM project_footage f LEFT JOIN assets a ON a.id = f.asset_id WHERE f.project_id = ? AND (f.status IN ('waiting', 'importing') @@ -427,6 +437,7 @@ export async function stepFootage(cfg: MediaConfig, projectId: string, opts: Ste // 1. Imports that finished, or never will. const importing = rows.filter((r) => r.status === "importing"); + const relaying: WorkRow[] = []; await Promise.all( importing.map(async (r) => { // Being re-encoded: once it has a media id it imports like any other. @@ -452,6 +463,12 @@ export async function stepFootage(cfg: MediaConfig, projectId: string, opts: Ste } return; } + // Coming in a piece at a time: moved on below, on deliveries. It has no + // media id until the last piece is in, and is not a stale claim. + if (r.upload_uid) { + relaying.push(r); + return; + } if (!r.media_uid) { if (age(r.updated_at) > STALE_CLAIM_MS) await setRow(r.id, { status: "waiting" }, " AND status = 'importing'"); return; @@ -484,6 +501,13 @@ export async function stepFootage(cfg: MediaConfig, projectId: string, opts: Ste }), ); + // 1b. Files Drive won't hand over whole: a few more pieces each. The time + // counts from here, so it doesn't matter what ran before. + if (opts.relay && relaying.length) { + const until = Date.now() + RELAY_BUDGET_MS; + await Promise.all(relaying.slice(0, RELAY_AT_ONCE).map((r) => relayClip(cfg, r, until))); + } + // 2. Start imports while there is room, together. A refusal that is about // the org (storage full) puts its clip back in line and starts no more. const room = opts.startImports === false ? 0 : STEP_LIMITS.importing - importing.filter((r) => r.status === "importing").length; @@ -499,15 +523,7 @@ export async function stepFootage(cfg: MediaConfig, projectId: string, opts: Ste const started = await startImport(cfg, r, opts.drive); if (started === true) return; if (started.driveLimit) { - // Drive lifts its limit within a day: the clip waits and is tried - // again by itself, further apart each time, then gives up. - const tries = r.drive_tries + 1; - if (tries > DRIVE_RETRY_MS.length) { - await setRow(r.id, { status: "failed", error: DRIVE_QUOTA, drive_tries: tries }); - } else { - const at = new Date(Date.now() + DRIVE_RETRY_MS[tries - 1]).toISOString(); - await setRow(r.id, { status: "waiting", error: DRIVE_WAIT, drive_tries: tries, retry_at: at }); - } + await waitOnDrive(cfg, r); return; } if (started.orgWide) { @@ -640,6 +656,120 @@ const DRIVE_WAIT_ALL = /** How long a clip Drive refused waits before each new try: about a day in all. */ const DRIVE_RETRY_MS = [30, 60, 120, 240, 240, 240, 240, 240].map((m) => m * 60_000); +/** + * Drive refused the clip. It lifts its limit within a day, so the clip waits + * and is tried again by itself, further apart each time, then gives up. A clip + * coming in a piece at a time keeps its upload while it waits, and carries on + * from there. + */ +async function waitOnDrive(cfg: MediaConfig, r: WorkRow): Promise { + const tries = r.drive_tries + 1; + if (tries > DRIVE_RETRY_MS.length) { + // An upload that can't be deleted now stays on the row: retrying the clip + // or deleting the project deletes it. + await dropUpload(cfg, r); + await setRow(r.id, { status: "failed", error: DRIVE_QUOTA, drive_tries: tries }); + r.status = "failed"; + } else { + const at = new Date(Date.now() + DRIVE_RETRY_MS[tries - 1]).toISOString(); + await setRow(r.id, { status: "waiting", error: DRIVE_WAIT, drive_tries: tries, retry_at: at }); + Object.assign(r, { status: "waiting", drive_tries: tries, retry_at: at }); + } +} + +const NO_UPLOAD = { upload_uid: null, upload_url: null, upload_expires: null, upload_size: null, upload_done: null }; + +/** + * Delete a clip's open upload on the media service, then forget it. False + * when the service wouldn't: it stays on the row, to be deleted later, rather + * than forgotten while it still holds storage. + */ +async function dropUpload(cfg: MediaConfig, r: WorkRow): Promise { + if (!r.upload_uid) return true; + try { + await deleteMedia(cfg, r.upload_uid); + } catch { + return false; + } + await setRow(r.id, NO_UPLOAD); + Object.assign(r, NO_UPLOAD); + return true; +} + +/** + * Drive won't hand the file over whole: open an upload for it to come in a + * piece at a time instead. Null when that can't help, so the clip waits on + * Drive: it won't serve even a piece, the file is over the video host's limit, + * or the clip is marked for a re-encode, which reads the whole file. + */ +async function openRelay(cfg: MediaConfig, r: WorkRow): Promise { + if (r.transcode) return null; + const size = await rangedSize(r.drive_file_id); + if (!size || size > MAX_RELAY_BYTES) return null; + await dropFailedCopy(cfg, r); + const opened = await openMediaUpload(cfg, size, r.name); + if ("failure" in opened) return ORG_LIMITS.has(opened.failure.error) ? { orgWide: true, detail: opened.failure.detail } : null; + await setRow(r.id, { + upload_uid: opened.id, + upload_url: opened.uploadUrl, + upload_expires: opened.expiresAt, + upload_size: size, + upload_done: 0, + }); + return true; +} + +/** Move a clip coming in a piece at a time on, and settle it once it is all in. */ +async function relayClip(cfg: MediaConfig, r: WorkRow, until: number): Promise { + const uid = r.upload_uid!; + const expiring = r.upload_expires !== null && Date.parse(r.upload_expires) - Date.now() < RELAY_EXPIRY_MARGIN_MS; + const result = expiring + ? ({ state: "gone" } as const) + : await relayPieces({ url: r.upload_url!, fileId: r.drive_file_id, size: r.upload_size! }, until, async (received) => { + await setRow(r.id, { upload_done: received }); + }); + if (result.state === "moving") return; + if (result.state === "done") return finishRelay(r, uid, result.contentType); + if (result.state === "drive") return waitOnDrive(cfg, r); + if (result.state === "refused") { + // Sending it again would be refused again: the clip fails, saying why. + await dropUpload(cfg, r); + await setRow(r.id, { status: "failed", error: `the video host refused a piece of this file (${result.detail})` }); + r.status = "failed"; + return; + } + // The upload can't take more. One whose last piece landed without being + // recorded (the step ended first) is the video: it is recorded now. + const s = await mediaState(cfg, uid); + if (!("failure" in s) && s.media.state !== "pendingupload") return finishRelay(r, uid, "video/mp4"); + // Otherwise it starts again with a new upload, counted like a refusal so a + // file that never gets in gives up in the end. + if (!(await dropUpload(cfg, r))) return; + const tries = r.drive_tries + 1; + if (tries > DRIVE_RETRY_MS.length) { + await setRow(r.id, { status: "failed", error: "the import did not finish", drive_tries: tries }); + r.status = "failed"; + } else { + await setRow(r.id, { status: "waiting", drive_tries: tries }); + Object.assign(r, { status: "waiting", drive_tries: tries }); + } +} + +/** The last piece is in: the upload is the clip's video, imported like any other. */ +async function finishRelay(r: WorkRow, uid: string, contentType: string): Promise { + const key = `media/${uid}`; + // Two steps that overlap can both get here: the key is unique, so one asset. + await run("INSERT INTO assets (key, name, content_type, size, media_uid) VALUES (?, ?, ?, ?, ?) ON CONFLICT(key) DO NOTHING", [ + key, + r.name, + contentType, + r.upload_size ?? 0, + uid, + ]); + const asset = await get<{ id: string }>("SELECT id FROM assets WHERE key = ?", [key]); + await setRow(r.id, { asset_id: asset!.id, ...NO_UPLOAD }); +} + /** * Check the Drive file serves video, then have the media service pull it. * @@ -654,6 +784,9 @@ async function startImport( r: WorkRow, drive?: DriveSource, ): Promise { + // Already coming in a piece at a time, back from a wait on Drive: it + // carries on from where its upload got to, on the next delivery. + if (r.upload_uid) return true; // Through the org's connection first. A file it can't hand over (too big for // the connector's temporary storage) comes by the original's shared link, // with its waits. Not by a copy in the connected account: Drive answers a @@ -674,7 +807,10 @@ async function startImport( if (!page) await probe.body?.cancel(); const verdict = judgeLinkResponse(probe.status, served, probe.headers.get("content-range"), probe.headers.get("content-length")); if (!verdict.ok) { - if (/[^<]*quota exceeded/i.test(page)) return { orgWide: false, detail: DRIVE_QUOTA, driveLimit: true }; + // Drive's limit refuses the whole file but still serves pieces of it. + if (/<title>[^<]*quota exceeded/i.test(page)) { + return (await openRelay(cfg, r)) ?? { orgWide: false, detail: DRIVE_QUOTA, driveLimit: true }; + } return { orgWide: false, detail: verdict.reason ?? "that file isn't a video" }; } @@ -684,6 +820,17 @@ async function startImport( /** The video host's refusal of a source over its bitrate cap. */ const BITRATE_REFUSED = /bitrate exceeded/i; +/** + * A copy from an attempt that failed is of no use and still counts against + * the org's storage: it goes before the clip is imported again. + */ +async function dropFailedCopy(cfg: MediaConfig, r: WorkRow): Promise<void> { + if (!r.asset_id) return; + if (r.media_uid) await deleteMedia(cfg, r.media_uid).catch(() => {}); + await run("DELETE FROM assets WHERE id = ?", [r.asset_id]); + await setRow(r.id, { asset_id: null }); +} + /** Have the media service pull the clip from `url`, and record it as an asset. */ async function importFrom( cfg: MediaConfig, @@ -692,13 +839,7 @@ async function importFrom( type: string, size: number, ): Promise<true | { orgWide: boolean; detail: string }> { - // A copy from an attempt that failed is of no use and still counts against - // the org's storage: it goes before the clip is imported again. - if (r.asset_id) { - if (r.media_uid) await deleteMedia(cfg, r.media_uid).catch(() => {}); - await run("DELETE FROM assets WHERE id = ?", [r.asset_id]); - await setRow(r.id, { asset_id: null }); - } + await dropFailedCopy(cfg, r); // Marked for re-encoding: the media id comes later, from the re-encode. if (r.transcode) { const started = await startTranscode(cfg, url, r.name); diff --git a/src/server/index.ts b/src/server/index.ts index 12a5bc4..75be045 100644 --- a/src/server/index.ts +++ b/src/server/index.ts @@ -830,8 +830,14 @@ app.delete("/api/projects/:id", async (c) => { // service, counted against the org's storage until it is deleted there, so // a big shoot is deleted a batch per call: 202 says how many are left, and // calling again carries on. A clip another project uses stays, in the library. - const footage = await query<{ id: string; asset_id: string | null; media_uid: string | null; copy_id: string | null }>( - `SELECT f.id, f.asset_id, a.media_uid, f.copy_id FROM project_footage f LEFT JOIN assets a ON a.id = f.asset_id + const footage = await query<{ + id: string; + asset_id: string | null; + media_uid: string | null; + copy_id: string | null; + upload_uid: string | null; + }>( + `SELECT f.id, f.asset_id, a.media_uid, f.copy_id, f.upload_uid FROM project_footage f LEFT JOIN assets a ON a.id = f.asset_id WHERE f.project_id = ? LIMIT ?`, [id, FOOTAGE_DELETE_BATCH], ); @@ -855,6 +861,15 @@ app.delete("/api/projects/:id", async (c) => { await Promise.all( footage.map(async (f) => { if (copyLeft.has(f.id)) return; + // A file coming in a piece at a time has an upload open, holding + // storage until it is deleted. + if (f.upload_uid) { + try { + await deleteMedia(mediaCfg(c.env), f.upload_uid); + } catch { + return; // kept for the next call + } + } if (f.asset_id) { const usedElsewhere = await get( "SELECT 1 AS used FROM edit_projects WHERE id != ? AND edl LIKE ? LIMIT 1", @@ -963,6 +978,8 @@ async function advanceFootage( const outcome = await stepFootage(mediaCfg(c.env), projectId, { drive: after === "delivery" ? drive : undefined, startImports: !(drive && after === "read"), + // Pieces of a file Drive won't hand over whole take a while: deliveries only. + relay: after === "delivery", }); // Highlights are read only on a delivery: a model call takes up to half a // minute, and a read answers at once. A read still keeps the chain booked. @@ -995,6 +1012,8 @@ interface FootageItemRow { size: number | null; duration: number | null; media_uid: string | null; + upload_size: number | null; + upload_done: number | null; } function footageOut(r: FootageItemRow) { @@ -1006,6 +1025,8 @@ function footageOut(r: FootageItemRow) { error: r.error, // Set while the clip waits on Google Drive: when it is tried again. retry_at: r.retry_at, + // Set while a file Drive won't hand over whole comes in a piece at a time. + received: r.upload_size ? { bytes: r.upload_done ?? 0, size: r.upload_size } : null, asset: r.asset_id ? { id: r.asset_id, @@ -1024,7 +1045,7 @@ function footageOut(r: FootageItemRow) { } const FOOTAGE_COLUMNS = `f.id, f.name, f.folder, f.status, f.error, f.retry_at, f.log_status, f.log_error, f.asset_id, - a.key, a.content_type, a.size, a.duration, a.media_uid`; + a.key, a.content_type, a.size, a.duration, a.media_uid, f.upload_size, f.upload_done`; function intParam(v: string | undefined, fallback: number, min: number, max: number): number { const n = Number.parseInt(v ?? "", 10); diff --git a/src/server/media.ts b/src/server/media.ts index 0d8f572..c5e4c47 100644 --- a/src/server/media.ts +++ b/src/server/media.ts @@ -144,12 +144,12 @@ export async function openMediaUpload( size: number, name: string, maxDurationSeconds?: number, -): Promise<{ id: string; uploadUrl: string } | { failure: MediaFailure }> { - const res = await call<{ id: string; upload_url: string }>(cfg, "/uploads", { +): Promise<{ id: string; uploadUrl: string; expiresAt: string | null } | { failure: MediaFailure }> { + const res = await call<{ id: string; upload_url: string; expires_at?: string }>(cfg, "/uploads", { method: "POST", body: JSON.stringify({ size, name, ...(maxDurationSeconds ? { max_duration_seconds: maxDurationSeconds } : {}) }), }); - return "failure" in res ? res : { id: res.data.id, uploadUrl: res.data.upload_url }; + return "failure" in res ? res : { id: res.data.id, uploadUrl: res.data.upload_url, expiresAt: res.data.expires_at ?? null }; } export async function mediaState(cfg: MediaConfig, uid: string): Promise<{ media: MediaState } | { failure: MediaFailure }> { diff --git a/src/server/relay.ts b/src/server/relay.ts new file mode 100644 index 0000000..9eb47c2 --- /dev/null +++ b/src/server/relay.ts @@ -0,0 +1,211 @@ +// A file Google Drive won't hand over whole still comes in, a piece at a time. +// +// Drive's per-file download limit ("Too many users have viewed or downloaded +// this file recently") answers a request for the whole file with its "Quota +// exceeded" page, yet still serves byte ranges of that same file. The media +// service only pulls whole files, so for such a file this app reads it in +// ranges from the shared link and appends each one to a resumable (tus) upload +// on the media service, a few pieces per background step. +// +// The upload's own offset is the cursor. It is read from the upload at the +// start of every step and never trusted from a row of ours, so a step that +// dies mid-piece, or two steps that overlap, can't leave a gap or send a piece +// twice: tus refuses a piece sent at any offset but the current one. + +import { directDownloadUrl, judgeLinkResponse } from "./drive-link"; + +/** + * Bytes per piece. The media service's video host takes chunks of at most + * 200 MiB, each but the last a multiple of 256 KiB. + */ +export const PIECE_BYTES = 200 * 1024 * 1024; + +/** The largest file the video host takes. A larger one waits on Drive, as before. */ +export const MAX_RELAY_BYTES = 30 * 1024 ** 3; + +/** + * Start no new piece once a step has relayed this long. The platform's queue + * delivers a batch of jobs one after another, so a long step holds up other + * apps' deliveries. At Drive's pace (about 10 MB/s a range) a piece takes some + * 20 s, so a step relays for under a minute. + */ +export const RELAY_BUDGET_MS = 30_000; + +/** Clips moved on at once in a step: each holds two connections while a piece is on its way. */ +export const RELAY_AT_ONCE = 2; + +/** An upload whose link expires sooner than this is opened again: a piece started on it might not land. */ +export const RELAY_EXPIRY_MARGIN_MS = 10 * 60_000; + +/** + * A piece still on its way after this long is given up (both ends are cut), + * so a stalled read can't hold a step open: 200 MiB at under 2 MB/s. + */ +const PIECE_TIMEOUT_MS = 120_000; + +// The video host's upload endpoint turned away a request with Python's +// default user agent ("error code: 1010"); it takes a named one. +const TUS = { "Tus-Resumable": "1.0.0", "User-Agent": "OpenVideo/1.0" }; + +/** The bytes of the next piece from `offset`, inclusive, as a Range header names them. Null once the file is all in. */ +export function nextPiece(offset: number, size: number): { start: number; end: number } | null { + if (offset >= size) return null; + return { start: offset, end: Math.min(offset + PIECE_BYTES, size) - 1 }; +} + +/** + * What Drive's answer to a ranged read means. + * - "ok": exactly the bytes asked for, of a video. + * - "refused": Drive won't serve them (its quota page, no access, no file, or + * other bytes than asked for), so the clip waits on Drive. + * - "retry": a hiccup (rate limited, an outage, the whole file instead of a + * piece), asked again on the next step. + */ +export function pieceVerdict( + status: number, + contentType: string | null, + contentRange: string | null, + piece: { start: number; end: number }, + size: number, +): "ok" | "refused" | "retry" { + if (status === 429 || status >= 500) return "retry"; + if (!judgeLinkResponse(status, contentType, contentRange, null).ok) return "refused"; + if (status !== 206) return "retry"; + const m = /^bytes (\d+)-(\d+)\/(\d+)$/.exec((contentRange ?? "").trim()); + return m && Number(m[1]) === piece.start && Number(m[2]) === piece.end && Number(m[3]) === size ? "ok" : "refused"; +} + +/** + * The file's length, read off a one-byte range, which Drive serves even for a + * file it won't hand over whole. Null when it won't serve even that. + */ +export async function rangedSize(fileId: string): Promise<number | null> { + const res = await fetch(directDownloadUrl(fileId), { + headers: { Range: "bytes=0-0" }, + signal: AbortSignal.timeout(30_000), + }).catch(() => null); + if (!res) return null; + await res.body?.cancel().catch(() => {}); + if (res.status !== 206) return null; + const v = judgeLinkResponse(res.status, res.headers.get("content-type"), res.headers.get("content-range"), null); + return v.ok && v.size ? v.size : null; +} + +/** How many bytes the upload holds. "gone": it can't take more. Null: it couldn't be asked. */ +async function uploadOffset(url: string): Promise<number | "gone" | null> { + const res = await fetch(url, { method: "HEAD", headers: TUS, signal: AbortSignal.timeout(15_000) }).catch(() => null); + if (!res) return null; + if (res.status === 404 || res.status === 410) return "gone"; + const offset = res.headers.get("Upload-Offset"); + if (!res.ok || offset === null) return null; + const n = Number(offset); + return Number.isInteger(n) && n >= 0 ? n : null; +} + +/** + * Append one piece at `offset`, streamed through without holding it. The new + * offset; "conflict" when the upload stands elsewhere; "gone"; `refused` with + * the host's reason when it turns the piece away for good; null when it didn't + * land this time (asked again on the next step). + */ +async function sendPiece( + url: string, + offset: number, + body: ReadableStream, + length: number, + signal: AbortSignal, +): Promise<number | "conflict" | "gone" | { refused: string } | null> { + // The video host refuses a piece sent without its length (400), and a + // streamed body goes without one unless it has a fixed length. (The test + // harness runs on Node, which has no FixedLengthStream.) + let sent: ReadableStream = body; + if (typeof FixedLengthStream === "function") { + const fixed = new FixedLengthStream(length); + body.pipeTo(fixed.writable).catch(() => {}); + sent = fixed.readable; + } + const res = await fetch(url, { + method: "PATCH", + headers: { ...TUS, "Upload-Offset": String(offset), "Content-Type": "application/offset+octet-stream" }, + body: sent, + signal, + }).catch(() => null); + if (!res) return null; + if (res.status === 409) return "conflict"; + if (res.status === 404 || res.status === 410) return "gone"; + // A malformed piece is refused the same way every time: retrying can't help. + if (res.status >= 400 && res.status < 500 && res.status !== 408 && res.status !== 429) { + const said = (await res.text().catch(() => "")).trim().slice(0, 200); + return { refused: `${res.status}${said ? `: ${said}` : ""}` }; + } + await res.body?.cancel().catch(() => {}); + const next = Number(res.headers.get("Upload-Offset")); + return res.ok && Number.isInteger(next) && next > offset ? next : null; +} + +export type RelayResult = + /** Every byte is in. `contentType` is what Drive said the file is. */ + | { state: "done"; contentType: string } + /** Some or no pieces landed this time; the next step carries on. */ + | { state: "moving" } + /** Drive refused a piece: the clip waits on Drive, and carries on from here after. */ + | { state: "drive" } + /** The upload can't take more (gone, or its link expired). */ + | { state: "gone" } + /** The video host turned a piece away for good, saying why. */ + | { state: "refused"; detail: string }; + +/** + * Append pieces of a Drive file to its open upload until the file is all in + * or `until` (a Date.now() time) passes; a piece already on its way finishes. + * `progress` hears each new offset. + */ +export async function relayPieces( + upload: { url: string; fileId: string; size: number }, + until: number, + progress: (received: number) => Promise<void>, +): Promise<RelayResult> { + let at = await uploadOffset(upload.url); + if (at === "gone") return { state: "gone" }; + if (at === null) return { state: "moving" }; + let contentType = "video/mp4"; + while (at < upload.size && Date.now() < until) { + const piece = nextPiece(at, upload.size)!; + // One signal cuts both ends of the piece: the read from Drive and the send. + const cut = new AbortController(); + const timer = setTimeout(() => cut.abort(), PIECE_TIMEOUT_MS); + let sent: Awaited<ReturnType<typeof sendPiece>>; + try { + const res = await fetch(directDownloadUrl(upload.fileId), { + headers: { Range: `bytes=${piece.start}-${piece.end}` }, + signal: cut.signal, + }).catch(() => null); + if (!res) break; + const verdict = pieceVerdict(res.status, res.headers.get("content-type"), res.headers.get("content-range"), piece, upload.size); + if (verdict !== "ok" || !res.body) { + cut.abort(); + if (verdict === "refused") return { state: "drive" }; + break; + } + contentType = (res.headers.get("content-type") ?? "").split(";")[0].trim() || contentType; + sent = await sendPiece(upload.url, at, res.body, piece.end - piece.start + 1, cut.signal); + if (typeof sent !== "number") cut.abort(); + } finally { + clearTimeout(timer); + } + if (sent === "gone") return { state: "gone" }; + if (sent === null) break; + if (typeof sent === "object") return { state: "refused", detail: sent.refused }; + if (sent === "conflict") { + // Someone else's piece landed first: carry on from where the upload stands. + const now = await uploadOffset(upload.url); + if (now === "gone") return { state: "gone" }; + if (now === null || now === at) break; + at = now; + continue; + } + at = sent; + await progress(at); + } + return at >= upload.size ? { state: "done", contentType } : { state: "moving" }; +} diff --git a/test/e2e/run.mjs b/test/e2e/run.mjs index 71f0234..907113d 100644 --- a/test/e2e/run.mjs +++ b/test/e2e/run.mjs @@ -38,7 +38,9 @@ const fid = (s) => (s + "_".repeat(28)).slice(0, 28); const world = { drive: {}, - media: new Map(), // uid → { polls, name, deleted } + media: new Map(), // uid → { polls, name, deleted, pendingUpload } + uploads: new Map(), // uid → { size, at, name, patches }: uploads opened on the media service + big: {}, // Drive file id → { size, refuseFrom }: files Drive serves only in pieces jobs: new Map(), calls: [], importRefusal: null, @@ -68,6 +70,24 @@ setDrive(); const json = (status, body) => new Response(JSON.stringify(body), { status, headers: { "content-type": "application/json" } }); let uidN = 0; +// `n` bytes, made as they are read: a piece of a big file costs no memory. +const ZEROS = new Uint8Array(1 << 20); +const zeros = (n) => { + let left = n; + return new ReadableStream({ + pull(c) { + if (left <= 0) return c.close(); + const k = Math.min(left, ZEROS.length); + c.enqueue(ZEROS.subarray(0, k)); + left -= k; + }, + }); +}; +const quotaPage = () => + new Response("<html><head><title>Google Drive - Quota exceededToo many users have viewed or downloaded this file recently.", { + status: 200, + headers: { "content-type": "text/html; charset=utf-8" }, + }); globalThis.fetch = async (input, init = {}) => { const url = typeof input === "string" ? input : input.url; const method = init.method ?? "GET"; @@ -78,17 +98,70 @@ globalThis.fetch = async (input, init = {}) => { return html ? new Response(html, { status: 200, headers: { "content-type": "text/html" } }) : new Response("no", { status: 404 }); } if (u.hostname === "drive.usercontent.google.com") { - const ranged = new Headers(init.headers ?? {}).has("range"); + const range = new Headers(init.headers ?? {}).get("range"); + const ranged = range !== null; const id = u.searchParams.get("id"); - if ((id === fid("q1") && !world.quotaLifted || (id.startsWith("conn") && id !== "conn_big___________________")) && !ranged) { + // Over Drive's daily download limit: the whole file gets the quota page, + // a piece its bytes, unless Drive is refusing pieces from some point on. + const big = world.big[id]; + if (big) { + if (!ranged) { + (world.wholeAsks ??= []).push(id); + return quotaPage(); + } + const [, a, b] = /^bytes=(\d+)-(\d+)$/.exec(range); + const start = Number(a); + const end = Math.min(Number(b), big.size - 1); + if (big.refuseFrom !== undefined && start >= big.refuseFrom) return quotaPage(); + return new Response(zeros(end - start + 1), { status: 206, headers: { "content-type": "video/mp4", "content-range": `bytes ${start}-${end}/${big.size}` } }); + } + // Over the limit and refusing pieces too: nothing gets it in but waiting. + if (id === fid("q1") && !world.quotaLifted) return quotaPage(); + if (id.startsWith("conn") && id !== "conn_big___________________" && !ranged) { // Over Drive's daily download limit: ranged reads still work, the whole file doesn't. - return new Response("Google Drive - Quota exceededToo many users have viewed or downloaded this file recently.", { status: 200, headers: { "content-type": "text/html; charset=utf-8" } }); + return quotaPage(); } return ranged ? new Response("xx", { status: 206, headers: { "content-type": "video/mp4", "content-range": "bytes 0-1/123456789" } }) : new Response("x".repeat(64), { status: 200, headers: { "content-type": "video/mp4", "content-length": "123456789" } }); } + // The media service's resumable upload, with the video host's rules: a + // named user agent ("error code: 1010" otherwise), a piece only at the + // upload's own offset, at most 200 MiB, and every piece but the last at + // least 5 MiB and a multiple of 256 KiB. + if (u.hostname === "upload.test") { + const id = u.pathname.split("/").pop(); + const up = world.uploads.get(id); + const h = new Headers(init.headers ?? {}); + if (!h.get("user-agent")) return new Response("error code: 1010", { status: 403 }); + if (h.get("tus-resumable") !== "1.0.0") return new Response(null, { status: 412 }); + if (!up || world.media.get(id)?.deleted) return new Response(null, { status: 404 }); + if (method === "HEAD") return new Response(null, { status: 200, headers: { "Upload-Offset": String(up.at), "Upload-Length": String(up.size) } }); + if (method === "PATCH") { + if (world.refusePatches === id) return new Response("Decoding Error", { status: 400 }); + if (world.patchFails?.(id, up.at)) return new Response("busy", { status: 503 }); + if (Number(h.get("upload-offset")) !== up.at) return new Response(null, { status: 409 }); + if (h.get("content-type") !== "application/offset+octet-stream") return new Response(null, { status: 415 }); + let n = 0; + for await (const chunk of init.body) n += chunk.byteLength; + const last = up.at + n === up.size; + if (n > 200 * 2 ** 20 || up.at + n > up.size || (!last && (n < 5 * 2 ** 20 || n % (256 * 1024) !== 0))) { + return new Response("bad chunk", { status: 400 }); + } + up.patches.push(up.at); + up.at += n; + if (last) world.media.get(id).pendingUpload = false; + return new Response(null, { status: 204, headers: { "Upload-Offset": String(up.at) } }); + } + } if (u.hostname === "svc.test") { + if (u.pathname === "/media/uploads" && method === "POST") { + const body = JSON.parse(init.body); + const id = (++uidN).toString(16).padStart(32, "0"); + world.media.set(id, { polls: 0, name: body.name, pendingUpload: true }); + world.uploads.set(id, { size: body.size, at: 0, name: body.name, patches: [] }); + return json(201, { id, upload_url: `https://upload.test/tus/${id}?tusv2=true`, expires_at: new Date(Date.now() + 6 * 3600_000).toISOString() }); + } if (u.pathname === "/media/import") { if (world.importRefusal) return json(409, world.importRefusal); const id = (++uidN).toString(16).padStart(32, "0"); @@ -124,6 +197,7 @@ globalThis.fetch = async (input, init = {}) => { if (m) { const v = world.media.get(m[1]); if (!v || v.deleted) return json(404, { error: "not_found" }); + if (v.pendingUpload) return json(200, { id: m[1], state: "pendingupload", ready: false, duration: null, error: null }); v.polls++; if (v.name === "HIGH.MP4") return json(200, { id: m[1], state: "error", ready: false, duration: null, error: "The video bitrate exceeded the maximum acceptable value of 200 Mbps." }); const ready = v.polls >= 2; @@ -361,11 +435,13 @@ const probes = () => world.calls.filter((c) => c.includes(fid("q1")) && c.includ const probesBefore = probes(); await statuses(); assert.equal(probes(), probesBefore, "a read before it is due leaves Drive alone"); -// "Try Google Drive again now" asks at once. +// "Try Google Drive again now" asks at once: the whole file, then one byte to +// see whether it would come in pieces (this file's pieces are refused too). r = await call("POST", `/api/projects/${pid}/footage/retry`); assert.equal(r.data.imports, 1); f = await statuses(); -assert.equal(probes(), probesBefore + 1); +assert.equal(probes(), probesBefore + 2); +assert.ok(![...world.uploads.values()].some((x) => x.name === "B_0009.MP4"), "Drive refusing pieces too: no upload is opened"); // Drive lifts the limit: when it is due, it imports. world.quotaLifted = true; db.prepare("UPDATE project_footage SET retry_at = ? WHERE name = 'B_0009.MP4'").run(new Date(Date.now() - 1000).toISOString()); @@ -385,6 +461,139 @@ db.prepare("DELETE FROM assets WHERE id = (SELECT asset_id FROM project_footage db.prepare("DELETE FROM project_footage WHERE name = 'B_0009.MP4'").run(); world.quotaLifted = true; +// 7c. A file Drive won't hand over whole still serves pieces: it comes in a +// piece at a time, on deliveries only, into an upload on the media service, +// and carries on from where the upload stands after a piece that didn't land +// or a wait on Drive. An upload about to expire starts again on a new one; a +// file over the video host's 30 GB limit waits on Drive as before. +{ + const P = 200 * 2 ** 20; + const A = 2 * P + 3 * 2 ** 20; + world.big[fid("big1")] = { size: A }; + world.big[fid("big2")] = { size: 12 * 2 ** 20 }; + world.big[fid("huge")] = { size: 31 * 2 ** 30 }; + world.drive[fid("day4")] = page("Day 4", [ + fileEntry(fid("big1"), "INTERVIEW_A.MP4"), + fileEntry(fid("big2"), "INTERVIEW_B.MP4"), + fileEntry(fid("huge"), "STATIC.MP4"), + ]); + r = await call("POST", "/api/projects", { folder: `https://drive.google.com/drive/folders/${fid("day4")}` }); + assert.equal(r.status, 201, JSON.stringify(r.data)); + const p4 = r.data.id; + const items4 = async () => Object.fromEntries((await call("GET", `/api/projects/${p4}/footage?logs=0`)).data.items.map((i) => [i.name, i])); + const deliver4 = async () => { + const body = JSON.stringify({ project_id: p4 }); + const res = await app.request("https://open-video.apps.clawnify.com/api/footage/step", { method: "POST", headers: { "content-type": "application/json", ...(await signed(body)) }, body }, env, ctx); + assert.equal(res.status, 200, await res.text()); + }; + const uploadOf = (name) => db.prepare("SELECT upload_uid FROM project_footage WHERE project_id = ? AND name = ?").get(p4, name).upload_uid; + const pieces = () => world.calls.filter((c) => c.startsWith("PATCH https://upload.test/")).length; + + // A read opens the uploads and sends no pieces. + let it = await items4(); + assert.equal(it["INTERVIEW_A.MP4"].status, "importing", JSON.stringify(it["INTERVIEW_A.MP4"])); + assert.deepEqual(it["INTERVIEW_A.MP4"].received, { bytes: 0, size: A }); + assert.equal(pieces(), 0, "a read sends no pieces"); + const a1 = uploadOf("INTERVIEW_A.MP4"); + assert.deepEqual([world.uploads.get(a1).size, world.uploads.get(a1).name], [A, "INTERVIEW_A.MP4"]); + assert.equal(it["STATIC.MP4"].status, "waiting", "over 30 GB: it waits on Drive, as before"); + assert.match(it["STATIC.MP4"].error, /Waiting for Google Drive/); + assert.ok(![...world.uploads.values()].some((x) => x.name === "STATIC.MP4"), "and no upload is opened for it"); + + // First delivery: INTERVIEW_A's second piece doesn't land, and INTERVIEW_B's + // upload is about to expire, so it starts again on a new one. + world.patchFails = (id, at) => id === a1 && at === P && !world.patchFailed && (world.patchFailed = true); + const b1 = uploadOf("INTERVIEW_B.MP4"); + db.prepare("UPDATE project_footage SET upload_expires = ? WHERE project_id = ? AND name = 'INTERVIEW_B.MP4'").run(new Date(Date.now() + 60_000).toISOString(), p4); + await deliver4(); + assert.deepEqual(world.uploads.get(a1).patches, [0], "one piece in; the next didn't land"); + it = await items4(); + assert.deepEqual(it["INTERVIEW_A.MP4"].received, { bytes: P, size: A }); + assert.equal(world.media.get(b1).deleted, true, "the upload about to expire is deleted"); + const b2 = uploadOf("INTERVIEW_B.MP4"); + assert.ok(b2 && b2 !== b1, "and a new one opened"); + + // Second delivery: A carries on from where its upload stands, until Drive + // refuses a piece; it then waits on Drive and keeps its upload. B is all in. + world.big[fid("big1")].refuseFrom = 2 * P; + await deliver4(); + assert.deepEqual(world.uploads.get(a1).patches, [0, P], "carried on from where the upload stood"); + it = await items4(); + assert.equal(it["INTERVIEW_A.MP4"].status, "waiting", JSON.stringify(it["INTERVIEW_A.MP4"])); + assert.ok(it["INTERVIEW_A.MP4"].retry_at); + assert.equal(uploadOf("INTERVIEW_A.MP4"), a1, "its upload is kept while it waits"); + assert.equal(it["INTERVIEW_B.MP4"].asset?.media_uid, b2, "INTERVIEW_B is in, as its upload's video"); + + // Asking Drive again now: it carries on without asking for the whole file again. + world.big[fid("big1")].refuseFrom = undefined; + const wholeAsks = () => world.wholeAsks.filter((x) => x === fid("big1")).length; + const asked = wholeAsks(); + r = await call("POST", `/api/projects/${p4}/footage/retry`); + assert.equal(r.data.imports, 2, "A, and the file over the limit"); + it = await items4(); + assert.equal(it["INTERVIEW_A.MP4"].status, "importing"); + assert.equal(wholeAsks(), asked, "an upload under way carries on: Drive isn't asked for the whole file"); + await deliver4(); + assert.deepEqual(world.uploads.get(a1).patches, [0, P, 2 * P], "every piece once, in order"); + it = await items4(); + assert.equal(it["INTERVIEW_A.MP4"].received, null); + assert.deepEqual([it["INTERVIEW_A.MP4"].asset?.media_uid, it["INTERVIEW_A.MP4"].asset?.size], [a1, A]); + + // From there it is imported like any other: ready, then logged. + for (let i = 0; i < 10 && !["INTERVIEW_A.MP4", "INTERVIEW_B.MP4"].every((n) => it[n].log_status === "done"); i++) it = await items4(); + assert.equal(it["INTERVIEW_A.MP4"].status, "ready"); + assert.equal(it["INTERVIEW_A.MP4"].log_status, "done"); + assert.equal(it["INTERVIEW_B.MP4"].log_status, "done"); + r = await call("DELETE", `/api/projects/${p4}`); + assert.equal(r.status, 200, JSON.stringify(r.data)); + assert.ok(world.media.get(a1).deleted && world.media.get(b2).deleted); + console.log("7c ok: a file Drive won't hand over whole comes in by pieces on deliveries, carries on after a failed piece and a wait on Drive, restarts an expiring upload; over 30 GB still waits"); +} + +// 7d. Deleting a project deletes an upload still under way; one the service +// won't delete keeps its clip for the next call. +{ + world.big[fid("big5")] = { size: 2 * 2 ** 20 }; + world.drive[fid("day5")] = page("Day 5", [fileEntry(fid("big5"), "INTERVIEW_C.MP4")]); + r = await call("POST", "/api/projects", { folder: `https://drive.google.com/drive/folders/${fid("day5")}` }); + const p5 = r.data.id; + await call("GET", `/api/projects/${p5}/footage?logs=0`); + const u5 = db.prepare("SELECT upload_uid FROM project_footage WHERE project_id = ?").get(p5).upload_uid; + assert.ok(u5, "the read opened an upload"); + world.refuseMediaDeletes = 1; + r = await call("DELETE", `/api/projects/${p5}`); + assert.equal(r.status, 202, JSON.stringify(r.data)); + assert.ok(!world.media.get(u5).deleted, "refused: the clip stays, with its upload"); + r = await call("DELETE", `/api/projects/${p5}`); + assert.equal(r.status, 200, JSON.stringify(r.data)); + assert.equal(world.media.get(u5).deleted, true); + console.log("7d ok: deleting a project deletes an upload under way, after a refusal too"); +} + +// 7e. A piece the video host turns away for good fails the clip with the +// host's reason, and its upload is deleted: sending it again can't help. +{ + world.big[fid("big6")] = { size: 3 * 2 ** 20 }; + world.drive[fid("day6")] = page("Day 6", [fileEntry(fid("big6"), "INTERVIEW_D.MP4")]); + r = await call("POST", "/api/projects", { folder: `https://drive.google.com/drive/folders/${fid("day6")}` }); + const p6 = r.data.id; + await call("GET", `/api/projects/${p6}/footage?logs=0`); + const u6 = db.prepare("SELECT upload_uid FROM project_footage WHERE project_id = ?").get(p6).upload_uid; + assert.ok(u6, "the read opened an upload"); + world.refusePatches = u6; + const body = JSON.stringify({ project_id: p6 }); + const res = await app.request("https://open-video.apps.clawnify.com/api/footage/step", { method: "POST", headers: { "content-type": "application/json", ...(await signed(body)) }, body }, env, ctx); + assert.equal(res.status, 200, await res.text()); + const row = db.prepare("SELECT status, error, upload_uid FROM project_footage WHERE project_id = ?").get(p6); + assert.equal(row.status, "failed", JSON.stringify(row)); + assert.equal(row.error, "the video host refused a piece of this file (400: Decoding Error)"); + assert.equal(row.upload_uid, null); + assert.equal(world.media.get(u6).deleted, true, "its upload is deleted"); + r = await call("DELETE", `/api/projects/${p6}`); + assert.equal(r.status, 200, JSON.stringify(r.data)); + console.log("7e ok: a piece the video host refuses for good fails the clip with its reason, and deletes the upload"); +} + // 8. Retry puts failures back in line. db.prepare("UPDATE project_footage SET status = 'failed', error = 'x' WHERE project_id = ? AND name = 'B_0002.MP4'").run(pid); r = await call("POST", `/api/projects/${pid}/footage/retry`); diff --git a/test/relay.test.ts b/test/relay.test.ts new file mode 100644 index 0000000..cb3f067 --- /dev/null +++ b/test/relay.test.ts @@ -0,0 +1,246 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { MAX_RELAY_BYTES, PIECE_BYTES, nextPiece, pieceVerdict, rangedSize, relayPieces } from "../src/server/relay"; + +afterEach(() => vi.unstubAllGlobals()); + +const MiB = 1024 * 1024; +const UPLOAD = "https://upload.test/tus/abc"; + +/** `n` bytes, made as they are read. */ +const ZEROS = new Uint8Array(MiB); +function bytes(n: number): ReadableStream { + let left = n; + return new ReadableStream({ + pull(c) { + if (left <= 0) return c.close(); + const k = Math.min(left, ZEROS.length); + c.enqueue(ZEROS.subarray(0, k)); + left -= k; + }, + }); +} + +async function count(body: unknown): Promise { + let n = 0; + for await (const chunk of body as AsyncIterable) n += chunk.byteLength; + return n; +} + +/** + * A Drive file of `size` bytes behind its shared link, and an open upload for + * it that holds `at` bytes. The upload counts what it is sent, like the real + * one, and refuses a piece sent anywhere but where it stands. + */ +function fakeUpload(size: number, at = 0) { + const state = { + at, + /** Where each piece that landed started. */ + patches: [] as number[], + headers: [] as Headers[], + refuseRange: (_start: number): "quota" | number | null => null, + patchFails: (_at: number): number | "throw" | null => null, + headStatus: 200, + }; + vi.stubGlobal( + "fetch", + vi.fn(async (input: string, init: RequestInit = {}) => { + const url = new URL(input); + const h = new Headers(init.headers); + if (url.hostname === "drive.usercontent.google.com") { + const [, a, b] = /^bytes=(\d+)-(\d+)$/.exec(h.get("range") ?? "")!; + const start = Number(a); + const end = Number(b); + const refusal = state.refuseRange(start); + if (refusal === "quota") { + return new Response("Google Drive - Quota exceeded", { + status: 200, + headers: { "content-type": "text/html; charset=utf-8" }, + }); + } + if (typeof refusal === "number") return new Response("busy", { status: refusal }); + return new Response(bytes(end - start + 1), { + status: 206, + headers: { "content-type": "video/mp4", "content-range": `bytes ${start}-${end}/${size}` }, + }); + } + if (url.href === UPLOAD && init.method === "HEAD") { + if (state.headStatus !== 200) return new Response(null, { status: state.headStatus }); + return new Response(null, { status: 200, headers: { "Upload-Offset": String(state.at), "Upload-Length": String(size) } }); + } + if (url.href === UPLOAD && init.method === "PATCH") { + state.headers.push(h); + const fail = state.patchFails(state.at); + if (fail === "throw") throw new TypeError("network connection lost"); + if (typeof fail === "number") return new Response(fail === 400 ? "Decoding Error" : null, { status: fail }); + if (Number(h.get("upload-offset")) !== state.at) return new Response(null, { status: 409 }); + state.patches.push(state.at); + state.at += await count(init.body); + return new Response(null, { status: 204, headers: { "Upload-Offset": String(state.at) } }); + } + throw new Error(`unexpected ${init.method ?? "GET"} ${input}`); + }), + ); + return state; +} + +describe("the video host's limits", () => { + it("a piece is its largest chunk, a multiple of 256 KiB, and a file at most its largest upload", () => { + expect(PIECE_BYTES).toBe(209_715_200); + expect(PIECE_BYTES % (256 * 1024)).toBe(0); + expect(MAX_RELAY_BYTES).toBe(30 * 1024 ** 3); + }); +}); + +describe("nextPiece", () => { + it("cuts a file into 200 MiB pieces, the last one shorter", () => { + const size = 2 * PIECE_BYTES + 3 * MiB; + expect(nextPiece(0, size)).toEqual({ start: 0, end: PIECE_BYTES - 1 }); + expect(nextPiece(PIECE_BYTES, size)).toEqual({ start: PIECE_BYTES, end: 2 * PIECE_BYTES - 1 }); + expect(nextPiece(2 * PIECE_BYTES, size)).toEqual({ start: 2 * PIECE_BYTES, end: size - 1 }); + expect(nextPiece(size, size)).toBeNull(); + }); + + it("sends a small file in one piece", () => { + expect(nextPiece(0, 4 * MiB)).toEqual({ start: 0, end: 4 * MiB - 1 }); + }); + + it("carries on from any offset, every piece but the last a multiple of 256 KiB", () => { + const size = 3 * PIECE_BYTES + 12_345; + let at = 7; // a piece that half landed + const lengths: number[] = []; + for (let p = nextPiece(at, size); p; p = nextPiece(at, size)) { + lengths.push(p.end - p.start + 1); + at = p.end + 1; + } + expect(at).toBe(size); + for (const n of lengths.slice(0, -1)) expect(n % (256 * 1024)).toBe(0); + }); +}); + +describe("pieceVerdict", () => { + const piece = { start: 100, end: 199 }; + + it("takes exactly the bytes asked for, of a video", () => { + expect(pieceVerdict(206, "video/mp4", "bytes 100-199/1000", piece, 1000)).toBe("ok"); + }); + + it("is refused by Drive's quota page, a closed or missing file, or other bytes than asked for", () => { + expect(pieceVerdict(200, "text/html; charset=utf-8", null, piece, 1000)).toBe("refused"); + expect(pieceVerdict(403, "text/html", null, piece, 1000)).toBe("refused"); + expect(pieceVerdict(404, null, null, piece, 1000)).toBe("refused"); + expect(pieceVerdict(206, "video/mp4", "bytes 100-150/1000", piece, 1000)).toBe("refused"); + expect(pieceVerdict(206, "video/mp4", "bytes 100-199/999", piece, 1000)).toBe("refused"); + }); + + it("asks again after a hiccup, or when the whole file comes instead of a piece", () => { + expect(pieceVerdict(429, null, null, piece, 1000)).toBe("retry"); + expect(pieceVerdict(503, "text/html", null, piece, 1000)).toBe("retry"); + expect(pieceVerdict(200, "video/mp4", null, piece, 1000)).toBe("retry"); + }); +}); + +describe("rangedSize", () => { + it("reads the length off a one-byte piece, which Drive serves even when it refuses the whole file", async () => { + const fetch = vi.fn(async () => new Response("x", { status: 206, headers: { "content-type": "video/mp4", "content-range": "bytes 0-0/15000000000" } })); + vi.stubGlobal("fetch", fetch); + expect(await rangedSize("file-id")).toBe(15_000_000_000); + const [url, init] = fetch.mock.calls[0] as unknown as [string, RequestInit]; + expect(url).toBe("https://drive.usercontent.google.com/download?id=file-id&export=download&confirm=t"); + expect(new Headers(init.headers).get("range")).toBe("bytes=0-0"); + }); + + it("is null when Drive won't serve even that, or can't be reached", async () => { + vi.stubGlobal("fetch", vi.fn(async () => new Response("Google Drive - Quota exceeded", { status: 200, headers: { "content-type": "text/html" } }))); + expect(await rangedSize("file-id")).toBeNull(); + vi.stubGlobal("fetch", vi.fn(async () => Promise.reject(new TypeError("offline")))); + expect(await rangedSize("file-id")).toBeNull(); + }); +}); + +describe("relayPieces", () => { + const SIZE = 2 * PIECE_BYTES + 3 * MiB; + const file = { url: UPLOAD, fileId: "file-id", size: SIZE }; + const later = () => Date.now() + 60_000; + const quiet = async () => {}; + + it("sends the file in order, a piece at a time, and says when it is all in", async () => { + const up = fakeUpload(SIZE); + const seen: number[] = []; + const result = await relayPieces(file, later(), async (n) => { + seen.push(n); + }); + expect(result).toEqual({ state: "done", contentType: "video/mp4" }); + expect(up.patches).toEqual([0, PIECE_BYTES, 2 * PIECE_BYTES]); + expect(up.at).toBe(SIZE); + expect(seen).toEqual([PIECE_BYTES, 2 * PIECE_BYTES, SIZE]); + const h = up.headers[0]; + expect(h.get("tus-resumable")).toBe("1.0.0"); + expect(h.get("content-type")).toBe("application/offset+octet-stream"); + expect(h.get("user-agent")).toMatch(/^OpenVideo\//); + }); + + it("carries on from where the upload stands, not from a count of its own", async () => { + const up = fakeUpload(SIZE, PIECE_BYTES + 7); + expect((await relayPieces(file, later(), quiet)).state).toBe("done"); + expect(up.patches).toEqual([PIECE_BYTES + 7, 2 * PIECE_BYTES + 7]); + expect(up.at).toBe(SIZE); + }); + + it("starts no piece once its time is up", async () => { + const up = fakeUpload(SIZE); + expect(await relayPieces(file, Date.now() - 1, quiet)).toEqual({ state: "moving" }); + expect(up.patches).toEqual([]); + }); + + it("stops for Drive when it refuses a piece, keeping what landed", async () => { + const up = fakeUpload(SIZE); + up.refuseRange = (start) => (start >= PIECE_BYTES ? "quota" : null); + expect(await relayPieces(file, later(), quiet)).toEqual({ state: "drive" }); + expect(up.at).toBe(PIECE_BYTES); + }); + + it("leaves a hiccup to the next step: Drive busy, a lost connection, a piece the upload turned away", async () => { + let up = fakeUpload(SIZE); + up.refuseRange = () => 503; + expect(await relayPieces(file, later(), quiet)).toEqual({ state: "moving" }); + + up = fakeUpload(SIZE); + up.patchFails = (at) => (at === PIECE_BYTES ? "throw" : null); + expect(await relayPieces(file, later(), quiet)).toEqual({ state: "moving" }); + expect(up.at).toBe(PIECE_BYTES); + + up = fakeUpload(SIZE); + up.patchFails = () => 500; + expect(await relayPieces(file, later(), quiet)).toEqual({ state: "moving" }); + expect(up.at).toBe(0); + }); + + it("follows the upload when another step's piece landed first", async () => { + const up = fakeUpload(SIZE); + let raced = false; + up.patchFails = (at) => { + if (raced || at !== 0) return null; + raced = true; + up.at = PIECE_BYTES; // someone else's first piece + return 409; + }; + expect((await relayPieces(file, later(), quiet)).state).toBe("done"); + expect(up.patches).toEqual([PIECE_BYTES, 2 * PIECE_BYTES]); + }); + + it("gives up on a piece the upload turns away for good, with the host's reason", async () => { + const up = fakeUpload(SIZE); + up.patchFails = () => 400; + expect(await relayPieces(file, later(), quiet)).toEqual({ state: "refused", detail: "400: Decoding Error" }); + expect(up.at).toBe(0); + }); + + it("says when the upload can't take more", async () => { + let up = fakeUpload(SIZE); + up.headStatus = 404; + expect(await relayPieces(file, later(), quiet)).toEqual({ state: "gone" }); + up = fakeUpload(SIZE); + up.patchFails = () => 410; + expect(await relayPieces(file, later(), quiet)).toEqual({ state: "gone" }); + }); +});