zib/Veskslop-t
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add daemon restart, preview diagnostics, browse caching, and rate-limit pauses

Browse Source
This commit is contained in:
Fisto
2026-09-17 07:33:35 +02:00
parent cba7ae8143
commit dea1724eb3
13 changed files with 367 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files
media-scrape/src/download.ts
+22 -1
View File
@@ -80,12 +80,14 @@ export interface DownloadOptions {
export interface DownloadOutcome {
id: number;
state: "downloaded" | "skipped" | "failed";
state: "downloaded" | "skipped" | "failed" | "ratelimited";
path?: string;
hash?: string;
bytes?: number;
/** File extension of the stored bytes, when known. */
ext?: string;
/** Seconds the server asked us to wait, when it is known. */
retryAfterMs?: number;
error?: string;
}
@@ -157,6 +159,16 @@ export async function downloadOne(
const response = await fetchImpl(media.url);
if (!response.ok) {
if (response.status === 429) {
const after = Number(response.headers.get("retry-after") ?? 0);
return {
id: media.id,
state: "ratelimited",
retryAfterMs: Number.isFinite(after) && after > 0 ? after * 1000 : undefined,
error: "http 429"
};
}
if (SKIP_STATUS.has(response.status)) {
return { id: media.id, state: "skipped", error: `http ${response.status}` };
}
@@ -252,6 +264,15 @@ export async function downloadPreview(
try {
const response = await fetchImpl(target);
if (response.status === 429) {
const after = Number(response.headers.get("retry-after") ?? 0);
return {
id: media.id,
state: "ratelimited",
retryAfterMs: Number.isFinite(after) && after > 0 ? after * 1000 : undefined,
error: "http 429"
};
}
if (!response.ok) return { id: media.id, state: "skipped", error: `http ${response.status}` };
const buffer = Buffer.from(await response.arrayBuffer());
media-scrape/src/engine.ts
+75 -4
View File
@@ -64,6 +64,8 @@ export interface ScrapeEngineOptions {
previews?: boolean;
/** Walk pages between preview cycles. 0 defers all previews to the end. Default 5. */
walkCyclePages?: number;
/** Pause after a rate limit is detected. Default 60 s. */
rateLimitPauseMs?: number;
/** Read dimensions and duration from a downloaded file. Defaults to ffprobe. */
probe?: (
path: string
@@ -88,11 +90,17 @@ function oldestId(messages: CoreMessage[]): string | null {
return oldest;
}
function isRateLimited(error: unknown): boolean {
const anyError = error as { status?: number; message?: string } | undefined;
return anyError?.status === 429 || /\b429\b|rate.?limit/i.test(String(anyError?.message ?? error));
}
export class ScrapeEngine {
private readonly queue: QueuedJob[] = [];
private readonly active = new Map<string, ScrapeMeta>();
private readonly stopRequested = new Set<string>();
private readonly pageTimes: number[] = [];
private rateLimitUntil = 0;
private processingPromise: Promise<void> | null = null;
constructor(private readonly options: ScrapeEngineOptions) {}
@@ -264,6 +272,7 @@ export class ScrapeEngine {
break;
}
await this.respectRateLimit();
const page = await this.fetchWithBackoff(meta, backend);
if (page.rateLimitedMs) await this.sleep(page.rateLimitedMs);
@@ -366,6 +375,8 @@ export class ScrapeEngine {
return;
}
await this.respectRateLimit();
const outcome = await downloadOne(row, targetDir, {
...options,
log: line => this.log(`${meta.id}: ${line}`)
@@ -394,6 +405,10 @@ export class ScrapeEngine {
meta.previewed = (meta.previewed ?? 0) + 1;
}
}
} else if (outcome.state === "ratelimited") {
// Leave the row pending and hold off: the server asked for a break.
this.registerRateLimit(outcome.retryAfterMs);
this.log(`${meta.id}: download #${row.id} rate limited (${outcome.error ?? "429"})`);
} else {
store.markDownloadError(row.id, outcome.error ?? outcome.state);
if (outcome.state === "failed") meta.downloadErrors = (meta.downloadErrors ?? 0) + 1;
@@ -412,6 +427,11 @@ export class ScrapeEngine {
destinationDir: meta.downloadDir ?? null
};
const pending = store.pendingPreviews(limit > 0 ? limit : 5000, true);
let cached = 0;
let skipped = 0;
let failed = 0;
if (pending.length) this.log(`${meta.id}: preview pass: ${pending.length} pending`);
const batchSize = Math.max(1, options.previewBatchSize ?? 25);
const batchPauseMs = Math.max(0, options.previewBatchPauseMs ?? 60_000);
let batch = 0;
@@ -432,6 +452,8 @@ export class ScrapeEngine {
return;
}
await this.respectRateLimit();
// Audio has no image proxy: render a waveform strip with ffmpeg.
if (row.kind === "audio" && row.source !== "embed") {
const waveform = await remoteFrame(row.url, join(dir, "thumbs"), { waveform: true });
@@ -444,6 +466,8 @@ export class ScrapeEngine {
meta.previewed = (meta.previewed ?? 0) + 1;
} else {
store.markPreviewError(row.id, "waveform failed");
failed++;
this.log(`${meta.id}: preview #${row.id} audio waveform failed (${row.url.slice(0, 60)})`);
}
this.persist(dir, meta);
await this.sleep(previewGapMs(options));
@@ -458,6 +482,8 @@ export class ScrapeEngine {
const resolved = await resolveEmbedPreview(row, options);
if (!resolved) {
store.markPreviewError(row.id, "no preview source");
skipped++;
this.log(`${meta.id}: preview #${row.id} ${row.kind}/${row.provider ?? "?"} has no preview source`);
this.persist(dir, meta);
await this.sleep(previewGapMs(options));
await afterRow();
@@ -473,19 +499,22 @@ export class ScrapeEngine {
let bytes = outcome.bytes ?? 0;
let ext = outcome.ext ?? "webp";
// Animated GIFs are kept static in the grid; hovering plays the original.
// Animated GIFs show a static frame in the grid; hovering plays the cached
// original from disk, so browsing never hotlinks the provider.
let animHash: string | undefined;
if (ext === "gif") {
const poster = await gifPoster(join(dir, "thumbs", `${hash}.gif`), join(dir, "thumbs"));
if (poster) {
rmSync(join(dir, "thumbs", `${hash}.gif`), { force: true });
animHash = hash;
hash = poster.hash;
bytes = poster.bytes;
ext = "webp";
}
}
store.markPreview(row.id, { hash, bytes, ext });
store.markPreview(row.id, { hash, bytes, ext, animHash });
meta.previewed = (meta.previewed ?? 0) + 1;
cached++;
// Embed thumbnails also land in the custom destination, so the download has a
// viewable poster for embeds while the media tab keeps its local copy.
@@ -506,18 +535,57 @@ export class ScrapeEngine {
meta.previewed = (meta.previewed ?? 0) + 1;
} else {
store.markPreviewError(row.id, "too large");
skipped++;
this.log(`${meta.id}: preview #${row.id} gif too large and no frame (${row.url.slice(0, 60)})`);
}
} else if (outcome.state === "ratelimited") {
// Leave the row pending and hold off: the server asked for a break.
this.registerRateLimit(outcome.retryAfterMs);
this.log(`${meta.id}: preview #${row.id} rate limited (${outcome.error ?? "429"})`);
} else if (outcome.state === "failed") {
// Transient network failure: leave the row pending for the next pass.
this.log(`${meta.id}: preview ${row.id} failed: ${outcome.error ?? "unknown"}`);
failed++;
this.log(
`${meta.id}: preview #${row.id} ${row.kind}/${row.provider ?? "?"} failed (transient): ` +
`${outcome.error ?? "unknown"} (${row.url.slice(0, 60)})`
);
} else {
store.markPreviewError(row.id, outcome.error ?? outcome.state);
skipped++;
this.log(
`${meta.id}: preview #${row.id} ${row.kind}/${row.provider ?? "?"} skipped: ` +
`${outcome.error ?? outcome.state} (${row.url.slice(0, 60)})`
);
}
this.persist(dir, meta);
await this.sleep(previewGapMs(options));
await afterRow();
}
const remaining = store.pendingPreviews(5000, true).length;
if (pending.length) {
this.log(
`${meta.id}: preview pass done: ${cached} cached, ${skipped} skipped, ${failed} failed, ` +
`${remaining} still pending`
);
}
}
/** A rate limit was seen: hold all walk and preview traffic for a while. */
private registerRateLimit(retryAfterMs?: number): void {
const pause = Math.max(this.options.rateLimitPauseMs ?? 60_000, retryAfterMs ?? 0);
const until = this.now() + pause;
if (until > this.rateLimitUntil) {
this.rateLimitUntil = until;
this.log(`rate limit detected; pausing media traffic until ${new Date(until).toISOString()}`);
}
}
private async respectRateLimit(): Promise<void> {
const wait = this.rateLimitUntil - this.now();
if (wait > 0) await this.sleep(wait);
}
private async fetchWithBackoff(meta: ScrapeMeta, backend: ScrapeBackend): Promise<FetchPage> {
@@ -530,6 +598,9 @@ export class ScrapeEngine {
return await backend.fetchPage(meta.channelId, meta.cursor, pageSize);
} catch (error) {
meta.errors++;
if (isRateLimited(error)) {
this.registerRateLimit((error as { retryAfterMs?: number }).retryAfterMs);
}
if (attempt >= maxRetries) throw error;
const retryAfter = (error as { retryAfterMs?: number }).retryAfterMs;
media-scrape/src/store.ts
+38 -1
View File
@@ -135,6 +135,7 @@ export interface MediaRow {
file_hash: string | null;
preview_hash: string | null;
preview_ext: string | null;
anim_hash: string | null;
file_bytes: number | null;
preview_bytes: number | null;
file_path: string | null;
@@ -173,6 +174,7 @@ CREATE TABLE IF NOT EXISTS media (
file_hash TEXT,
preview_hash TEXT,
preview_ext TEXT,
anim_hash TEXT,
file_bytes INTEGER,
preview_bytes INTEGER,
file_path TEXT,
@@ -217,6 +219,7 @@ export class ScrapeStore {
if (!names.has("download_error")) this.db.exec("ALTER TABLE media ADD COLUMN download_error TEXT");
if (!names.has("preview_error")) this.db.exec("ALTER TABLE media ADD COLUMN preview_error TEXT");
if (!names.has("preview_ext")) this.db.exec("ALTER TABLE media ADD COLUMN preview_ext TEXT");
if (!names.has("anim_hash")) this.db.exec("ALTER TABLE media ADD COLUMN anim_hash TEXT");
if (!names.has("play_url")) this.db.exec("ALTER TABLE media ADD COLUMN play_url TEXT");
if (!names.has("aspect_ratio")) this.db.exec("ALTER TABLE media ADD COLUMN aspect_ratio REAL");
}
@@ -355,7 +358,16 @@ export class ScrapeStore {
.all(limit) as unknown as MediaRow[];
}
markPreview(id: number, preview: { hash: string; bytes: number; ext?: string }): void {
markPreview(id: number, preview: { hash: string; bytes: number; ext?: string; animHash?: string }): void {
if (preview.animHash) {
this.db
.prepare(
"UPDATE media SET preview_hash = ?, preview_bytes = ?, preview_ext = ?, anim_hash = ?, preview_error = NULL WHERE id = ?"
)
.run(preview.hash, preview.bytes, preview.ext ?? "webp", preview.animHash, id);
return;
}
this.db
.prepare(
"UPDATE media SET preview_hash = ?, preview_bytes = ?, preview_ext = ?, preview_error = NULL WHERE id = ?"
@@ -385,6 +397,31 @@ export class ScrapeStore {
this.db.prepare(`UPDATE media SET ${sets.join(", ")} WHERE id = ?`).run(...values, id);
}
/** Rows whose preview failed, with the reason, newest first. */
previewFailures(
limit = 50
): Array<{ id: number; kind: MediaKind; provider: string | null; url: string; error: string }> {
const rows = this.db
.prepare(
"SELECT id, kind, provider, url, preview_error FROM media WHERE preview_error IS NOT NULL ORDER BY id DESC LIMIT ?"
)
.all(limit) as unknown as Array<{
id: number;
kind: MediaKind;
provider: string | null;
url: string;
preview_error: string;
}>;
return rows.map(row => ({
id: row.id,
kind: row.kind,
provider: row.provider,
url: row.url,
error: row.preview_error
}));
}
/** The message a media row came from, for URL refreshes. */
sourceMessageId(mediaId: number): string | null {
const row = this.db
media-scrape/test/run.ts
+73
View File
@@ -877,6 +877,79 @@ void (async () => {
assert.equal(existsSync(join(dir, `${poster?.hash}.webp`)), true);
});
await test("markPreview stores the animated gif reference", () => {
const dir = join(root, "anim-store");
mkdirSync(dir, { recursive: true });
const store = new ScrapeStore(dir);
store.insertPage([
message("1200", { attachments: [attachment({ filename: "loop.gif" })] })
]);
const row = store.queryMedia({ kind: "gif" })[0];
store.markPreview(row.id, { hash: "static-hash", bytes: 10, ext: "webp", animHash: "anim-hash" });
const updated = store.getMedia(row.id);
assert.equal(updated?.preview_hash, "static-hash");
assert.equal(updated?.anim_hash, "anim-hash");
store.close();
});
await test("rate limits pause media traffic", async () => {
const backend = new FakeBackend([
{
messages: [
message("1300", {
attachments: [
attachment({ filename: "a.png", contentType: "image/png" }),
attachment({ filename: "b.png", contentType: "image/png" })
]
})
]
}
]);
const sleeps: number[] = [];
const engine = new ScrapeEngine({
rootDir: root,
pageSize: 10,
pageMinMs: 0,
pageMaxMs: 0,
sleep: async (ms: number) => {
sleeps.push(ms);
},
random: () => 0,
log: () => {},
previews: true,
rateLimitPauseMs: 30_000,
download: {
fetchImpl: async () =>
new Response("slow down", { status: 429, headers: { "retry-after": "10" } }),
minGapMs: 0,
maxGapMs: 0,
previewMinGapMs: 0,
previewMaxGapMs: 0,
previewBatchSize: 100,
sleep: async () => {}
}
});
const started = await engine.start({ channelId: "160", guildId: null, mode: "links", backend });
await engine.processQueue();
const store = new ScrapeStore(join(root, started.id));
for (const row of store.queryMedia({ kind: "image" })) {
assert.equal(row.preview_hash, null);
assert.equal(row.preview_error, null, "rate limited rows stay pending");
}
store.close();
assert.ok(
sleeps.some(ms => ms > 25_000 && ms <= 30_000),
"the engine paused after the rate limit"
);
});
await test("previews run in cycles during the walk", async () => {
const events: string[] = [];
const pages: Array<{ messages: CoreMessage[] }> = [