commit d29a47e556c2f76f349ffe336551af8f860b00bb
parent 53896f7b6e47f77c1244722dd410617ef94e69c0
Author: I Mean I'm Just Saying <imeanimjustsaying@kiwifarms.st>
Date: Sun, 4 Oct 2026 16:34:05 -0400
editor: persist-videos job kind, per-channel action, replay
persistChannelVideosAction runs one persist-videos job on the channel's
download queue (persist-kept's), carrying ids, format, replace, gapMs and
minFreeMemMb in its spec so a retry re-runs the same list. The job fails when
the run stopped at the disk floor or a rate limit, or a video failed, so a
--wait exits non-zero and the re-run resumes. persistVideosAction plans a list
across channels (a dry run returns the buckets) and starts one job per channel
with something to fetch, reporting the channels it refused. The kind is
drainable and needsMedia, and counts as store-touching.
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Diffstat:
6 files changed, 214 insertions(+), 13 deletions(-)
diff --git a/common/controller/persistVideos.test.ts b/common/controller/persistVideos.test.ts
@@ -88,7 +88,12 @@ async function seedSaved(
paths: Paths,
slug: string,
id: string,
- opts: { height?: number; file?: string; keepReason?: "keep-latest" | "pin" } = {},
+ opts: {
+ height?: number;
+ file?: string;
+ keepReason?: "keep-latest" | "pin";
+ requestedBy?: string;
+ } = {},
): Promise<void> {
await seedVideo(paths, slug, id);
const dir = videoDir(paths, slug, id);
@@ -99,6 +104,7 @@ async function seedSaved(
sourceFilename: file,
storeDir: path.join(paths.savedVideosDir, slug, id),
keepReason: opts.keepReason ?? "keep-latest",
+ ...(opts.requestedBy ? { origin: { requestedBy: opts.requestedBy } } : {}),
format: {
preset: "original",
...(opts.height ? { height: opts.height } : {}),
@@ -240,7 +246,7 @@ test("format defaults to each channel's quality, and an explicit one wins", asyn
test("above-height replaces a too-tall or unmeasured container, keeping its provenance", async () => {
await withPaths(async (paths) => {
await seedSaved(paths, "demo-channel", "tall111", { height: 1080, file: "source-media.webm" });
- await seedSaved(paths, "demo-channel", "nohgt22");
+ await seedSaved(paths, "demo-channel", "nohgt22", { keepReason: "pin", requestedBy: "umtool" });
await seedSaved(paths, "demo-channel", "ok33333", { height: 720 });
const items = [
{ slug: "demo-channel", id: "tall111" },
@@ -275,6 +281,10 @@ test("above-height replaces a too-tall or unmeasured container, keeping its prov
assert.equal(tall?.format?.height, 720);
// The replacement is still the keep-latest container it replaced.
assert.equal(tall?.keepReason, "keep-latest");
+ const nohgt = await loadSavedVideo(videoDir(paths, "demo-channel", "nohgt22"));
+ assert.equal(nohgt?.keepReason, "pin");
+ assert.deepEqual(nohgt?.origin, { requestedBy: "umtool" });
+ assert.equal(nohgt?.format?.height, 720);
// The old, differently-named container went only after the new one landed.
await assert.rejects(stat(path.join(oldTall!.dir, oldTall!.file)));
assert.equal(
diff --git a/common/jobs/jobKinds.test.ts b/common/jobs/jobKinds.test.ts
@@ -82,6 +82,8 @@ const ADDED_KINDS: Record<string, { label: string; drainable: boolean }> = {
label: "Build & deploy homepage",
drainable: false,
},
+ // A list of videos persisted one at a time: a drain stops between them.
+ "persist-videos": { label: "Persist videos", drainable: true },
};
test("added kinds carry their pinned label and drainability", () => {
@@ -162,6 +164,7 @@ const STILL_MEDIA = [
"redownload-incomplete-bucket",
"retry-bucket",
"persist-kept",
+ "persist-videos",
"whisper-video",
"transcribe-one",
"download-one-pipeline",
diff --git a/common/jobs/jobKinds.ts b/common/jobs/jobKinds.ts
@@ -388,6 +388,17 @@ const JOB_KINDS: Record<string, JobKindMeta> = {
queueKeyStrategy: "custom",
needsMedia: true,
},
+ // Persist an explicit list of videos (controller/persistVideos.ts): one job
+ // per channel, on the channel's download queue as persist-kept is. Drainable:
+ // it stops between videos and leaves the rest for a re-run.
+ "persist-videos": {
+ kind: "persist-videos",
+ label: "Persist videos",
+ drainable: true,
+ replayable: true,
+ queueKeyStrategy: "custom",
+ needsMedia: true,
+ },
"backup-saved-videos": {
kind: "backup-saved-videos",
label: "Back up saved videos",
diff --git a/editor/app/channels/[slug]/persistActions.ts b/editor/app/channels/[slug]/persistActions.ts
@@ -12,10 +12,38 @@ import {
import { readChannelConfig } from "yt-dlp-transcript-common/controller/channels";
import { persistKept } from "yt-dlp-transcript-common/controller/persistKept";
import {
+ persistVideos,
+ type PersistReplacePolicy,
+ type PersistVideoItem,
+ type PersistVideosPlan,
+} from "yt-dlp-transcript-common/controller/persistVideos";
+import type { SourceVideoQuality } from "yt-dlp-transcript-common/ytdlp/downloadFormat";
+import {
runManagedFunction,
type StreamActionResult,
} from "yt-dlp-transcript-common/jobs/streamCommand";
+export type PersistVideosOptions = {
+ // Absent = each channel's effective source-video quality.
+ format?: SourceVideoQuality;
+ replace?: PersistReplacePolicy;
+ gapMs?: number;
+ minFreeMemMb?: number;
+ queueKey?: string;
+};
+
+export type PersistVideosActionResult =
+ | { ok: false; error: string }
+ | { ok: true; dryRun: true; plan: PersistVideosPlan }
+ | {
+ ok: true;
+ dryRun: false;
+ plan: PersistVideosPlan;
+ jobs: { slug: string; jobId: string }[];
+ jobIds: string[];
+ skipped: { slug: string; reason: string }[];
+ };
+
// Bulk catch-up: ensure every video in the channel's keep-latest window has its
// source container saved to the store. Re-downloads only those not already
// saved. Runs on the channel's download queue since it issues real downloads.
@@ -29,16 +57,8 @@ export async function persistKeptAction(
// The operator clicked this: only the floor applies and the shared hysteresis
// latch is left alone (see diskGate). This is a preflight only — persistKept
// re-checks per video, because it writes full containers in a loop.
- const disk = await diskGate(paths, getSettings(), { mode: "manual" });
- if (!disk.ok) {
- return {
- ok: false,
- error:
- `Low disk space: ${formatBytes(disk.freeBytes)} free, ` +
- `${formatBytes(disk.thresholdBytes)} required. Free up space or ` +
- `lower the floor in Settings.`,
- };
- }
+ const disk = await persistDiskPreflight();
+ if (disk) return { ok: false, error: disk };
return runManagedFunction({
kind: "persist-kept",
queueKey: resolveQueueKey(downloadQueueKey(config), queueKey),
@@ -57,3 +77,142 @@ export async function persistKeptAction(
},
});
}
+
+// Persist an explicit list of one channel's videos (controller/persistVideos.ts)
+// as one job on the channel's download queue — persist-kept's queue, since it
+// issues real downloads. The job carries its ids and options in its spec, so a
+// replay re-runs the same list, and a re-run fetches only what is still not
+// saved.
+export async function persistChannelVideosAction(
+ slug: string,
+ ids: string[],
+ opts: PersistVideosOptions = {},
+): Promise<StreamActionResult> {
+ const wanted = [...new Set(ids)];
+ if (wanted.length === 0) return { ok: false, error: "No video ids to persist." };
+ const paths = getPaths();
+ const config = await readChannelConfig(paths, slug);
+ if (!config) return { ok: false, error: `Channel "${slug}" not found` };
+ const disk = await persistDiskPreflight();
+ if (disk) return { ok: false, error: disk };
+ const { format, replace, gapMs, minFreeMemMb, queueKey } = opts;
+ return runManagedFunction({
+ kind: "persist-videos",
+ queueKey: resolveQueueKey(downloadQueueKey(config), queueKey),
+ paths,
+ channelSlug: slug,
+ spec: {
+ kind: "persist-videos",
+ slug,
+ params: { queueKey, ids: wanted, format, replace, gapMs, minFreeMemMb },
+ },
+ fn: async (onLog, signal, _progress, ctx) => {
+ const result = await persistVideos({
+ paths,
+ items: wanted.map((id) => ({ slug, id })),
+ format,
+ replace,
+ gapMs,
+ minFreeMemMb,
+ onLog,
+ signal,
+ drainSignal: ctx.drainSignal,
+ });
+ safeRevalidate([`/channels/${slug}`, "/saved-videos"]);
+ // A run that stopped short or lost a video did not do what it was asked:
+ // the job says so, and the re-run picks up the rest.
+ if (result.stopped === "low-disk" || result.stopped === "rate-limit") {
+ throw new Error(
+ `Stopped (${result.stopped}) with ${result.notAttempted.length} video(s) not attempted — run it again later.`,
+ );
+ }
+ if (result.failed.length > 0) {
+ throw new Error(`${result.failed.length} video(s) failed to persist.`);
+ }
+ },
+ });
+}
+
+// The disk floor a persist asks before it queues (manual mode: the operator
+// asked). A preflight only — the controllers re-check per video.
+async function persistDiskPreflight(): Promise<string | null> {
+ const disk = await diskGate(getPaths(), getSettings(), { mode: "manual" });
+ if (disk.ok) return null;
+ return (
+ `Low disk space: ${formatBytes(disk.freeBytes)} free, ` +
+ `${formatBytes(disk.thresholdBytes)} required. Free up space or ` +
+ `lower the floor in Settings.`
+ );
+}
+
+// Persist a list of videos ACROSS channels. A dry run answers with the buckets
+// and starts nothing. A real run starts one persist-videos job per channel that
+// has something to fetch — so each runs on its channel's download queue, behind
+// that platform's other downloads, under the channel's media guard — and
+// reports the channels it refused, with the action's own sentence.
+export async function persistVideosAction(
+ input: { items: PersistVideoItem[]; dryRun?: boolean } & PersistVideosOptions,
+): Promise<PersistVideosActionResult> {
+ const { items, dryRun, ...opts } = input;
+ if (items.length === 0) return { ok: false, error: "No videos to persist." };
+ const paths = getPaths();
+ const planned = await persistVideos({
+ paths,
+ items,
+ format: opts.format,
+ replace: opts.replace,
+ dryRun: true,
+ });
+ const plan = planned.plan;
+ if (dryRun) return { ok: true, dryRun: true, plan };
+ if (plan.willFetch === 0) {
+ return { ok: true, dryRun: false, plan, jobs: [], jobIds: [], skipped: [] };
+ }
+ const disk = await persistDiskPreflight();
+ if (disk) return { ok: false, error: disk };
+ // Only the channels with work get a job, and each job gets that channel's
+ // WHOLE slice of the list: its own re-run skips what is saved.
+ const due = new Set(
+ [
+ ...plan.toFetch.items,
+ ...(opts.replace === "above-height" ? plan.wrongHeight.items : []),
+ ].map((i) => i.slug),
+ );
+ const bySlug = new Map<string, string[]>();
+ for (const { slug, id } of items) {
+ if (!due.has(slug)) continue;
+ bySlug.set(slug, [...(bySlug.get(slug) ?? []), id]);
+ }
+ const jobs: { slug: string; jobId: string }[] = [];
+ const skipped: { slug: string; reason: string }[] = [];
+ for (const [slug, ids] of bySlug) {
+ let result: StreamActionResult;
+ try {
+ result = await persistChannelVideosAction(slug, ids, opts);
+ } catch (e) {
+ skipped.push({ slug, reason: (e as Error).message });
+ continue;
+ }
+ if (!result.ok) {
+ skipped.push({ slug, reason: result.error });
+ continue;
+ }
+ // Nobody reads the stream: the job's log is on disk (see the ops layer).
+ void result.stream.cancel();
+ jobs.push({ slug, jobId: result.jobId });
+ }
+ if (jobs.length === 0) {
+ return {
+ ok: false,
+ error: skipped.map((s) => `${s.slug}: ${s.reason}`).join("; "),
+ };
+ }
+ return {
+ ok: true,
+ dryRun: false,
+ plan,
+ jobs,
+ jobIds: jobs.map((j) => j.jobId),
+ skipped,
+ };
+}
diff --git a/editor/app/jobs/jobReplayRegistry.ts b/editor/app/jobs/jobReplayRegistry.ts
@@ -12,6 +12,8 @@ import { readChannelSnapshot } from "yt-dlp-transcript-common/controller/channel
import type { JobSpec, ReplayBucket } from "yt-dlp-transcript-common/jobs/jobSpec";
import type { StreamActionResult } from "yt-dlp-transcript-common/jobs/streamCommand";
import type { AudioFormat } from "yt-dlp-transcript-common/lib/channelConfig";
+import { isSourceVideoQuality } from "yt-dlp-transcript-common/ytdlp/downloadFormat";
+import { isPersistReplacePolicy } from "yt-dlp-transcript-common/controller/persistVideos";
import {
downloadAction,
downloadMissingAction,
@@ -38,7 +40,10 @@ import {
type DigestLaneChoice,
} from "../channels/[slug]/digestActions";
import { backfillChannelAction } from "../channels/[slug]/backfillActions";
-import { persistKeptAction } from "../channels/[slug]/persistActions";
+import {
+ persistChannelVideosAction,
+ persistKeptAction,
+} from "../channels/[slug]/persistActions";
import {
capturePostsAction,
checkPostAvailabilityAction,
@@ -341,4 +346,16 @@ export const JOB_REPLAY_HANDLERS: Record<string, ReplayHandler> = {
const { queueKey } = params(spec);
return persistKeptAction(spec.slug, queueKey);
},
+ // The same list, the same options: a replay fetches whatever of it is still
+ // not saved (the controller skips the rest).
+ "persist-videos": (spec) => {
+ const { p, queueKey } = params(spec);
+ return persistChannelVideosAction(spec.slug, strings(p.ids) ?? [], {
+ queueKey,
+ format: isSourceVideoQuality(p.format) ? p.format : undefined,
+ replace: isPersistReplacePolicy(p.replace) ? p.replace : undefined,
+ gapMs: num(p.gapMs),
+ minFreeMemMb: num(p.minFreeMemMb),
+ });
+ },
};
diff --git a/editor/app/storage/lib/storeBusy.ts b/editor/app/storage/lib/storeBusy.ts
@@ -47,6 +47,7 @@ const STORE_TOUCHING_KINDS = new Set([
"sync",
// The store's own jobs.
"persist-kept",
+ "persist-videos",
"check-kept-deleted",
"backup-saved-videos",
"verify-saved-video-backup",
