commit db0387bd840b65ed295c18cf6b9afc145bcac3ab
parent 6d04f7ed3f1917924607080667accf845e954931
Author: I Mean I'm Just Saying <imeanimjustsaying@kiwifarms.st>
Date: Thu, 24 Sep 2026 21:57:03 -0400
common: a 429 mid-sweep is "incomplete", not a listing and not a failure
enumeratePlaylistUrls classifies a non-zero exit's stderr; rate_limit throws
EnumerationIncompleteError {platform, pagesReached, count} (page from the
extractor's last "Downloading page N"). syncFullSweep catches it before
acceptEnumeration: records the platform cooldown through the same
onPlatformBackoff a download's 429 uses, logs one "sweep incomplete" line, and
runs the paged walk for this sync. lastFullSweepAt is never stamped and the
stored playlist is untouched. Any other non-zero exit throws as before.
fullSweepDue is now async and false while the channel's platform (the Sync
gate's key, detectPlatform(url) ?? "unknown") is cooling down.
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Diffstat:
2 files changed, 211 insertions(+), 12 deletions(-)
diff --git a/common/ytdlp/runYtdlp.ts b/common/ytdlp/runYtdlp.ts
@@ -53,6 +53,7 @@ import {
isFullSweepDue,
resolveFullSweepIntervalMinutes,
} from "../jobs/deepSync";
+import { platformCooldownRemainingMs } from "../jobs/downloadBackoff";
import { computeKeepWindow, type KeepWindow } from "../controller/keptVideos";
import { downloadOneManaged } from "./downloadOneManaged";
import {
@@ -315,6 +316,56 @@ export function outputArgsForUrl(
// Enumerate a channel's video URLs via `--flat-playlist --print url` (metadata
// only, no downloads). Pass `range` to fetch a single newest-first page via
// `-I start:end`; sync uses this to walk the channel incrementally.
+// A full enumeration that the platform rate-limited part-way through. What it
+// printed is a PREFIX of the listing, not a listing: acting on it as one would
+// flag every entry past the cut-off as missing, and failing the sync on it
+// throws away a download walk that would have worked. Thrown only for a
+// `rate_limit`-classified exit (HTTP 429 and friends, classifyDownloadFailure);
+// every other non-zero exit still throws the plain error.
+//
+// `platform` is the cooldown key the Sync gate and the auto-download runner
+// share (`detectPlatform(url) ?? "unknown"`); `pagesReached` is the last
+// `Downloading page N` the extractor logged (null for an extractor that does
+// not log pages); `count` is how many entries were printed before the cut.
+export class EnumerationIncompleteError extends Error {
+ readonly platform: string;
+ readonly pagesReached: number | null;
+ readonly count: number;
+ constructor(
+ exitCode: number | undefined,
+ platform: string,
+ pagesReached: number | null,
+ count: number,
+ ) {
+ super(
+ `yt-dlp exited with code ${exitCode} (rate-limited${
+ pagesReached !== null ? ` at page ${pagesReached}` : ""
+ } of the listing, ${count} entries)`,
+ );
+ this.name = "EnumerationIncompleteError";
+ this.platform = platform;
+ this.pagesReached = pagesReached;
+ this.count = count;
+ }
+}
+
+// The last `Downloading page N` an extractor logged — yt-dlp's paged channel
+// extractors print `[RumbleChannel] <name>: Downloading page 155` per page.
+export function lastListingPage(stderr: string): number | null {
+ let last: number | null = null;
+ for (const m of stderr.matchAll(/Downloading page (\d+)/g)) {
+ last = Number(m[1]);
+ }
+ return last;
+}
+
+// The per-platform cooldown key: the one the Sync gate (pipelineActions) and
+// the auto-download runner both use, so a sweep's 429 pauses exactly what
+// theirs would.
+function cooldownPlatformKey(config: ChannelConfig): string {
+ return detectPlatform(config.url) ?? "unknown";
+}
+
async function enumeratePlaylistUrls(
opts: RunYtdlpOpts,
root: string,
@@ -349,6 +400,10 @@ async function enumeratePlaylistUrls(
});
child.stderr?.on("data", (c: Buffer) => opts.onLog(c.toString("utf8")));
const result = await child;
+ const urls = String(result.stdout ?? "")
+ .split("\n")
+ .map((s) => s.trim())
+ .filter(Boolean);
// yt-dlp exit code convention: 101 = "break-on-existing" / "max-downloads"
// (clean stop, not an error). Treat it the same as 0. A non-zero/101 exit
@@ -358,16 +413,22 @@ async function enumeratePlaylistUrls(
result.exitCode !== 101 &&
!opts.signal.aborted
) {
+ const stderr = String(result.stderr ?? "");
+ if (classifyDownloadFailure(stderr, undefined) === "rate_limit") {
+ throw new EnumerationIncompleteError(
+ result.exitCode,
+ cooldownPlatformKey(opts.channelConfig),
+ lastListingPage(stderr),
+ urls.length,
+ );
+ }
throw new Error(`yt-dlp exited with code ${result.exitCode}`);
}
if (result.exitCode === 101) {
opts.onLog(`yt-dlp stopped on existing entry (exit 101).\n`);
}
- return String(result.stdout ?? "")
- .split("\n")
- .map((s) => s.trim())
- .filter(Boolean);
+ return urls;
}
// Thin exported wrapper over the module-private enumeratePlaylistUrls for
@@ -1332,23 +1393,33 @@ const SYNC_PAGE_SIZE = 50;
// more expensive on a large channel, so it runs at most once per interval.
// `forceFullSweep` overrides the gate in both directions.
async function sync(opts: RunYtdlpOpts): Promise<void> {
- const sweep = opts.forceFullSweep ?? fullSweepDue(opts);
+ const sweep = opts.forceFullSweep ?? (await fullSweepDue(opts));
return sweep ? syncFullSweep(opts) : syncPaged(opts);
}
// Whether this sync should upgrade itself to a full sweep, resolved from the
-// per-channel override, the global cadence and the channel's lastFullSweepAt.
-function fullSweepDue(opts: RunYtdlpOpts): boolean {
+// per-channel override, the global cadence and the channel's lastFullSweepAt —
+// and never while the channel's platform is in a rate-limit cooldown: a sweep
+// is the longest request run a channel makes, and re-trying one into the
+// window that just 429'd is how the cooldown gets extended. The paged walk
+// still runs; `forceFullSweep` still overrides.
+export async function fullSweepDue(
+ opts: Pick<RunYtdlpOpts, "channelConfig" | "paths">,
+ now: number = Date.now(),
+): Promise<boolean> {
const scheduler = getSettings().syncScheduler;
const interval = resolveFullSweepIntervalMinutes(
opts.channelConfig,
scheduler,
);
- return isFullSweepDue(
- opts.channelConfig.lastFullSweepAt,
- interval,
- Date.now(),
+ if (!isFullSweepDue(opts.channelConfig.lastFullSweepAt, interval, now)) {
+ return false;
+ }
+ const cooling = await platformCooldownRemainingMs(
+ cooldownPlatformKey(opts.channelConfig),
+ opts.paths,
);
+ return cooling <= 0;
}
// The per-page download filter, shared by both passes so they can never drift:
@@ -1556,7 +1627,24 @@ async function syncFullSweep(opts: RunYtdlpOpts): Promise<void> {
// 1. One enumeration, no range, then the gate: record everything it saw in
// the roster (additive, so this is safe unconditionally) and decide
// whether the listing itself may be acted on.
- const urls = await enumeratePlaylistUrls(opts, root);
+ let urls: string[];
+ try {
+ urls = await enumeratePlaylistUrls(opts, root);
+ } catch (err) {
+ if (!(err instanceof EnumerationIncompleteError)) throw err;
+ // A 429 part-way through the listing: what was printed is a prefix, not a
+ // listing, so it never reaches acceptEnumeration (no playlist rewrite, no
+ // missing set, no lastFullSweepAt). Record the platform's cooldown the way
+ // a download's 429 does, then do this sync's job the cheap way.
+ await opts.onPlatformBackoff?.("rate_limit");
+ opts.onLog(
+ `Full sweep incomplete: 429 at ${
+ err.pagesReached !== null ? `page ${err.pagesReached}` : "an unknown page"
+ } of the listing, ${err.count} entries — not a listing; next syncs are paged walks until the ${err.platform} cooldown ends.\n`,
+ );
+ if (opts.signal.aborted) return;
+ return syncPaged(opts);
+ }
if (opts.signal.aborted) return;
const { decision, roster, listedIds, now } = await acceptEnumeration(
diff --git a/common/ytdlp/sweepIncomplete.test.ts b/common/ytdlp/sweepIncomplete.test.ts
@@ -0,0 +1,111 @@
+import { test } from "node:test";
+import assert from "node:assert/strict";
+import { mkdtempSync, writeFileSync, chmodSync, mkdirSync } from "node:fs";
+import os from "node:os";
+import path from "node:path";
+
+// Run with:
+// pnpm --filter yt-dlp-transcript-common exec tsx --test ytdlp/sweepIncomplete.test.ts
+//
+// A 429 part-way through a full enumeration is "incomplete", not a listing and
+// not a failure (release 5 slice R). getPaths() memoizes, so the env is set
+// before anything imports it.
+const ROOT = mkdtempSync(path.join(os.tmpdir(), "sweep-incomplete-"));
+process.env.TRANSCRIPTS_DIR = ROOT;
+process.env.SETTINGS_FILE = path.join(ROOT, "settings.json");
+writeFileSync(process.env.SETTINGS_FILE, "{}\n");
+mkdirSync(path.join(ROOT, "channels", "c"), { recursive: true });
+
+const {
+ fetchFlatPlaylistUrls,
+ EnumerationIncompleteError,
+ lastListingPage,
+ fullSweepDue,
+} = await import("./runYtdlp");
+const { getPaths } = await import("../lib/paths");
+const { recordDownloadBackoff } = await import("../jobs/downloadBackoff");
+
+// A stand-in yt-dlp: prints `urls` on stdout, `stderr` on stderr, exits `code`.
+function fakeBin(name: string, urls: string[], stderr: string, code: number) {
+ const bin = path.join(ROOT, name);
+ writeFileSync(
+ bin,
+ `#!/bin/sh\ncat <<'OUT'\n${urls.join("\n")}\nOUT\ncat >&2 <<'ERR'\n${stderr}\nERR\nexit ${code}\n`,
+ );
+ chmodSync(bin, 0o755);
+ return bin;
+}
+
+function run(bin: string) {
+ return fetchFlatPlaylistUrls({
+ channelConfig: { handling: "youtube", url: "https://rumble.com/c/x" } as never,
+ paths: { ...getPaths(), ytdlpBin: bin },
+ channelSlug: "c",
+ onLog: () => {},
+ signal: new AbortController().signal,
+ });
+}
+
+const PAGES = [1, 2, 3]
+ .map((n) => `[RumbleChannel] x: Downloading page ${n}`)
+ .join("\n");
+const URLS = Array.from({ length: 15 }, (_, i) => `https://rumble.com/v${i}-t.html`);
+
+test("lastListingPage reads the extractor's last page line", () => {
+ assert.equal(lastListingPage(PAGES), 3);
+ assert.equal(lastListingPage("nothing here"), null);
+});
+
+test("a 429 mid-listing throws EnumerationIncompleteError with page and count", async () => {
+ const bin = fakeBin(
+ "ytdlp-429",
+ URLS,
+ `${PAGES}\nERROR: x: Unable to download webpage: HTTP Error 429: Too Many Requests`,
+ 1,
+ );
+ await assert.rejects(run(bin), (err: unknown) => {
+ assert.ok(err instanceof EnumerationIncompleteError);
+ assert.equal(err.platform, "rumble");
+ assert.equal(err.pagesReached, 3);
+ assert.equal(err.count, 15);
+ assert.match(err.message, /^yt-dlp exited with code 1/);
+ return true;
+ });
+});
+
+test("any other non-zero exit keeps throwing the plain error", async () => {
+ const bin = fakeBin("ytdlp-other", URLS, "ERROR: Unsupported URL", 1);
+ await assert.rejects(run(bin), (err: unknown) => {
+ assert.ok(err instanceof Error);
+ assert.ok(!(err instanceof EnumerationIncompleteError));
+ assert.equal(err.message, "yt-dlp exited with code 1");
+ return true;
+ });
+});
+
+test("fullSweepDue is false while the channel's platform cools down", async () => {
+ const paths = getPaths();
+ const due = {
+ channelConfig: {
+ handling: "youtube",
+ url: "https://rumble.com/c/x",
+ fullSweepIntervalMinutes: 60,
+ lastFullSweepAt: "2020-01-01T00:00:00.000Z",
+ } as never,
+ paths,
+ };
+ const youtube = {
+ channelConfig: {
+ handling: "youtube",
+ url: "https://www.youtube.com/@x",
+ fullSweepIntervalMinutes: 60,
+ lastFullSweepAt: "2020-01-01T00:00:00.000Z",
+ } as never,
+ paths,
+ };
+ assert.equal(await fullSweepDue(due), true);
+ await recordDownloadBackoff("rumble", paths);
+ assert.equal(await fullSweepDue(due), false);
+ // Another platform's sweep is unaffected.
+ assert.equal(await fullSweepDue(youtube), true);
+});
