Skip to content

Commit 234b153

Browse files
committed
Add persisted download job queue and Download Center UI
download-jobs-store.ts backs jobs with a JSON store; download-worker.ts/native-downloader.ts drive the actual transfer via the Rust download lib; DownloadCenter.tsx surfaces job progress/pause/resume/retry/cancel.
1 parent 2e3fd7e commit 234b153

5 files changed

Lines changed: 459 additions & 5 deletions

File tree

‎app/src/download-jobs-store.ts‎

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -46,13 +46,18 @@ export interface DownloadJob {
4646
modelName: string;
4747
publisher: string;
4848
quantization?: string;
49-
backend: "llamacpp" | "mlx" | "vllm" | "ollama";
49+
backend: "llamacpp" | "mlx" | "vllm" | "ollama" | "transformers";
5050
destinationDir: string;
5151
modelId: string;
5252
shards: DownloadShard[];
5353
state: DownloadJobState;
5454
error?: DownloadJobError;
5555
retryCount: number;
56+
jobReceivedBytes?: number;
57+
totalBytes?: number;
58+
bytesPerSecond?: number;
59+
etaSeconds?: number;
60+
recoveredAtStartup?: boolean;
5661
createdAt: string;
5762
updatedAt: string;
5863
}
@@ -95,7 +100,7 @@ export function createJob(
95100
// benefit.
96101
export function updateJob(
97102
id: string,
98-
partial: Partial<Pick<DownloadJob, "state" | "error" | "retryCount" | "shards">>
103+
partial: Partial<Pick<DownloadJob, "state" | "error" | "retryCount" | "shards" | "jobReceivedBytes" | "totalBytes" | "bytesPerSecond" | "etaSeconds" | "recoveredAtStartup">>
99104
): DownloadJob | null {
100105
const all = listJobs();
101106
const idx = all.findIndex((j) => j.id === id);

‎app/src/download-queue.ts‎

Lines changed: 73 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,14 @@
11
import type { BrowserWindow } from "electron";
2-
import { listJobs, updateJob } from "./download-jobs-store";
2+
import * as downloadWorker from "./download-worker";
3+
import * as fs from "node:fs";
4+
import { createJob, deleteJob, getJob, listJobs, updateJob, type DownloadJob } from "./download-jobs-store";
5+
import { getDownloadManager } from "./native-downloader";
6+
import { detectModelFormat, getSpecs, resolveAutomaticRuntime } from "./system-specs";
7+
8+
export interface DownloadControls { concurrency: number; bandwidthMbps: number }
9+
export interface RecoveryStatus { recoveredJobs: number; recoveredAt: string | null }
10+
let recoveryStatus: RecoveryStatus = { recoveredJobs: 0, recoveredAt: null };
11+
export function getJobs(): DownloadJob[] { return listJobs(); }
312

413
// Deliberately a getWindow() closure, not a captured event.sender from one
514
// ipcMain.handle invocation — jobs push progress at times disconnected from
@@ -28,10 +37,71 @@ export function broadcast(): void {
2837
// resuming is exactly what downloadGgufFile's Range-request logic already
2938
// does for a partial file, whatever caused the interruption).
3039
export function resumeInterruptedJobs(): void {
40+
let recoveredJobs = 0;
3141
for (const job of listJobs()) {
32-
if (job.state === "downloading" || job.state === "resolving") {
33-
updateJob(job.id, { state: "queued" });
42+
if (["downloading", "resolving", "verifying", "installing"].includes(job.state)) {
43+
updateJob(job.id, { state: "queued", recoveredAtStartup: true });
44+
recoveredJobs++;
3445
}
3546
}
3647
broadcast();
48+
downloadWorker.wake();
49+
recoveryStatus = { recoveredJobs, recoveredAt: new Date().toISOString() };
50+
}
51+
52+
export function getRecoveryStatus(): RecoveryStatus { return recoveryStatus; }
53+
54+
export function configure(controls: DownloadControls): DownloadControls {
55+
const concurrency = Math.max(1, Math.min(8, Math.floor(controls.concurrency)));
56+
const bandwidthMbps = Math.max(0, Number(controls.bandwidthMbps) || 0);
57+
const manager = getDownloadManager();
58+
manager.setGlobalConcurrency(concurrency);
59+
manager.setBandwidthLimit(bandwidthMbps > 0 ? bandwidthMbps * 1024 * 1024 / 8 : undefined);
60+
return { concurrency, bandwidthMbps };
61+
}
62+
63+
export async function createHuggingFaceJob(input: { modelId: string; filename: string; expectedBytes: number; destinationDir: string; backend?: "automatic" | DownloadJob["backend"] }): Promise<DownloadJob> {
64+
const filename = input.filename.replace(/\\/g, "/").split("/").pop() || input.filename;
65+
// "automatic" (the default) resolves the same way the model-recommendation
66+
// flow does: format inferred from the filename/repo id, matched against
67+
// this machine's detected hardware.
68+
const backend = !input.backend || input.backend === "automatic"
69+
? resolveAutomaticRuntime(detectModelFormat(input.filename) === "unknown" ? detectModelFormat(input.modelId) : detectModelFormat(input.filename), await getSpecs())
70+
: input.backend;
71+
const job = createJob({
72+
kind: "huggingface", modelName: input.modelId.split("/").pop() || input.modelId,
73+
publisher: input.modelId.split("/")[0] || "Hugging Face", backend,
74+
destinationDir: input.destinationDir, modelId: input.modelId,
75+
quantization: filename.match(/(Q\d(?:_[A-Z0-9]+)*)/i)?.[1],
76+
shards: [{ filename: input.filename, path: `${input.destinationDir}/${filename}`, expectedBytes: Math.max(0, input.expectedBytes), receivedBytes: 0, state: "queued" }],
77+
});
78+
broadcast(); downloadWorker.wake(); return job;
79+
}
80+
81+
export function pauseJob(id: string): void {
82+
const job = getJob(id); if (!job || !["queued", "resolving", "downloading"].includes(job.state)) return;
83+
downloadWorker.pause(id); updateJob(id, { state: "paused", etaSeconds: undefined, bytesPerSecond: undefined }); broadcast();
84+
}
85+
export function resumeJob(id: string): void {
86+
const job = getJob(id); if (!job || !["paused", "failed", "cancelled"].includes(job.state)) return;
87+
downloadWorker.clearRetryTimer(id); updateJob(id, { state: "queued", error: undefined, recoveredAtStartup: false }); broadcast(); downloadWorker.wake();
88+
}
89+
export function retryJob(id: string): void { resumeJob(id); }
90+
export function cancelJob(id: string): void {
91+
const job = getJob(id); if (!job || ["ready", "cancelled"].includes(job.state)) return;
92+
downloadWorker.cancel(id); updateJob(id, { state: "cancelled", etaSeconds: undefined, bytesPerSecond: undefined }); broadcast();
93+
}
94+
export function removeJob(id: string): void {
95+
const job = getJob(id); if (!job) return;
96+
downloadWorker.cancel(id);
97+
for (const shard of job.shards) {
98+
for (const suffix of [".part", ".part.json"]) { try { fs.rmSync(`${shard.path}${suffix}`, { force: true }); } catch { /* best effort */ } }
99+
}
100+
deleteJob(id); broadcast();
101+
}
102+
export function diskForecast(id: string): { requiredBytes: number; availableBytes: number | null; enough: boolean | null } {
103+
const job = getJob(id); if (!job) throw new Error("Download job not found");
104+
const requiredBytes = job.shards.reduce((sum, shard) => sum + Math.max(0, shard.expectedBytes - shard.receivedBytes), 0);
105+
try { const stats = fs.statfsSync(job.destinationDir); const availableBytes = stats.bavail * stats.bsize; return { requiredBytes, availableBytes, enough: availableBytes >= requiredBytes }; }
106+
catch { return { requiredBytes, availableBytes: null, enough: null }; }
37107
}

‎app/src/download-worker.ts‎

Lines changed: 166 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,166 @@
1+
import * as accounts from "./accounts";
2+
import * as downloadQueue from "./download-queue";
3+
import { getJob, listJobs, updateJob, type DownloadErrorKind, type DownloadJob, type DownloadShard } from "./download-jobs-store";
4+
import { logger } from "./logger";
5+
import { getDownloadManager, type JobEvent, type JsDownloadJob } from "./native-downloader";
6+
7+
const MAX_AUTO_RETRIES = 3;
8+
9+
// How often raw JobEvents get coalesced into a `downloads:update` broadcast
10+
// — matches the ~100ms cadence the Rust side's own progress ticker already
11+
// uses, so this never becomes the bottleneck. A job's terminal state always
12+
// broadcasts immediately regardless (see `driveJob`'s `finally`), so the UI
13+
// never has to wait out this throttle to see a job actually finish.
14+
const BROADCAST_THROTTLE_MS = 100;
15+
let lastBroadcastAt = 0;
16+
17+
function throttledBroadcast(): void {
18+
const now = Date.now();
19+
if (now - lastBroadcastAt < BROADCAST_THROTTLE_MS) return;
20+
lastBroadcastAt = now;
21+
downloadQueue.broadcast();
22+
}
23+
24+
// Job ids currently being driven — guards against `wake()` picking up the
25+
// same "queued" job twice; the real concurrency gate is the Rust manager's
26+
// own semaphore, this is purely a JS-side duplicate-start guard.
27+
const active = new Set<string>();
28+
const retryTimers = new Map<string, NodeJS.Timeout>();
29+
30+
export function clearRetryTimer(jobId: string): void {
31+
const timer = retryTimers.get(jobId);
32+
if (timer) clearTimeout(timer);
33+
retryTimers.delete(jobId);
34+
}
35+
36+
function toJsDownloadJob(job: DownloadJob): JsDownloadJob {
37+
return {
38+
id: job.id,
39+
modelId: job.modelId,
40+
token: accounts.getAccountToken("huggingface") ?? undefined,
41+
destinationDir: job.destinationDir,
42+
shards: job.shards.map((s) => ({
43+
filename: s.filename,
44+
path: s.path,
45+
expectedBytes: s.expectedBytes,
46+
receivedBytes: s.receivedBytes,
47+
sha256: s.sha256,
48+
})),
49+
};
50+
}
51+
52+
function withShardPatch(job: DownloadJob, filename: string, patch: Partial<DownloadShard>): DownloadShard[] {
53+
return job.shards.map((s) => (s.filename === filename ? { ...s, ...patch } : s));
54+
}
55+
56+
function handleEvent(event: JobEvent): void {
57+
const job = getJob(event.jobId);
58+
if (!job) return; // job was deleted while its download was in flight
59+
60+
switch (event.kind) {
61+
case "shard_state":
62+
if (event.shardFilename && event.shardState) {
63+
updateJob(job.id, { shards: withShardPatch(job, event.shardFilename, { state: event.shardState as DownloadShard["state"] }) });
64+
}
65+
break;
66+
case "shard_progress":
67+
if (event.shardFilename && event.receivedBytes !== undefined) {
68+
updateJob(job.id, {
69+
shards: withShardPatch(job, event.shardFilename, { receivedBytes: event.receivedBytes }),
70+
jobReceivedBytes: event.jobReceivedBytes,
71+
totalBytes: event.totalBytes,
72+
bytesPerSecond: event.bytesPerSec,
73+
etaSeconds: event.etaSeconds,
74+
});
75+
}
76+
break;
77+
case "job_state":
78+
if (event.jobState) {
79+
updateJob(job.id, { state: event.jobState as DownloadJob["state"] });
80+
}
81+
break;
82+
case "job_error":
83+
updateJob(job.id, {
84+
state: "failed",
85+
error: {
86+
message: event.errorMessage ?? "Download failed.",
87+
kind: (event.errorKind ?? "unknown") as DownloadErrorKind,
88+
retryable: event.retryable ?? false,
89+
},
90+
});
91+
break;
92+
}
93+
}
94+
95+
// A `"failed"` job whose error is marked retryable gets a small number of
96+
// automatic requeue attempts with exponential backoff before it's left for
97+
// a manual retry — distinct from the Rust side's own per-chunk retry
98+
// (MAX_ATTEMPTS_PER_CHUNK), which handles much shorter-lived transient
99+
// failures within a single shard transfer.
100+
function maybeAutoRetry(jobId: string): void {
101+
const job = getJob(jobId);
102+
if (!job || job.state !== "failed" || !job.error?.retryable || job.retryCount >= MAX_AUTO_RETRIES) return;
103+
104+
const backoffMs = 1000 * 2 ** job.retryCount;
105+
if (retryTimers.has(jobId)) return;
106+
const expectedRetryCount = job.retryCount;
107+
const timer = setTimeout(() => {
108+
retryTimers.delete(jobId);
109+
const current = getJob(jobId);
110+
if (!current || current.state !== "failed" || !current.error?.retryable || current.retryCount !== expectedRetryCount) return;
111+
updateJob(jobId, { state: "queued", retryCount: expectedRetryCount + 1 });
112+
wake();
113+
}, backoffMs);
114+
timer.unref();
115+
retryTimers.set(jobId, timer);
116+
}
117+
118+
async function driveJob(job: DownloadJob): Promise<void> {
119+
active.add(job.id);
120+
try {
121+
await getDownloadManager().startJob(toJsDownloadJob(job), (err, event) => {
122+
if (err) {
123+
logger.error(`Download job ${job.id}: event callback error: ${err.message}`);
124+
return;
125+
}
126+
handleEvent(event);
127+
throttledBroadcast();
128+
});
129+
} catch (err) {
130+
// The rejection carries the same message the "job_error" event
131+
// already recorded via handleEvent — this catch exists only so an
132+
// unexpected throw here doesn't become an unhandled rejection.
133+
logger.error(`Download job ${job.id}: ${(err as Error).message}`);
134+
} finally {
135+
active.delete(job.id);
136+
downloadQueue.broadcast();
137+
maybeAutoRetry(job.id);
138+
}
139+
}
140+
141+
// Picks up every currently-"queued" Hugging Face job not already being
142+
// driven. Safe to call repeatedly/redundantly — `active` prevents double
143+
// starts. Called once at app-ready and again whenever something might have
144+
// made a "queued" job appear (a fresh `downloads:create`, a manual
145+
// retry/resume, or `download-queue.ts`'s startup requeue).
146+
export function wake(): void {
147+
for (const job of listJobs()) {
148+
if (job.state === "queued" && job.kind === "huggingface" && !active.has(job.id)) {
149+
void driveJob(job);
150+
}
151+
}
152+
}
153+
154+
export function start(): void {
155+
wake();
156+
}
157+
158+
export function pause(jobId: string): void {
159+
clearRetryTimer(jobId);
160+
getDownloadManager().pauseJob(jobId);
161+
}
162+
163+
export function cancel(jobId: string): void {
164+
clearRetryTimer(jobId);
165+
getDownloadManager().cancelJob(jobId);
166+
}

‎app/src/native-downloader.ts‎

Lines changed: 99 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,99 @@
1+
import * as path from "node:path";
2+
3+
export interface NativeDownloadProgress {
4+
receivedBytes: number;
5+
totalBytes?: number;
6+
}
7+
8+
export interface JobEvent {
9+
jobId: string;
10+
kind: "shard_progress" | "shard_state" | "job_state" | "job_error";
11+
shardFilename?: string;
12+
shardState?: string;
13+
jobState?: string;
14+
receivedBytes?: number;
15+
jobReceivedBytes?: number;
16+
totalBytes?: number;
17+
bytesPerSec?: number;
18+
etaSeconds?: number;
19+
errorMessage?: string;
20+
errorKind?: string;
21+
retryable?: boolean;
22+
}
23+
24+
export interface JsShard {
25+
filename: string;
26+
path: string;
27+
expectedBytes: number;
28+
receivedBytes: number;
29+
sha256?: string;
30+
}
31+
32+
export interface JsDownloadJob {
33+
id: string;
34+
modelId: string;
35+
token?: string;
36+
destinationDir: string;
37+
shards: JsShard[];
38+
}
39+
40+
export interface NativeDownloadManager {
41+
setGlobalConcurrency(limit: number): void;
42+
setBandwidthLimit(bytesPerSec: number | undefined | null): void;
43+
pauseJob(jobId: string): void;
44+
cancelJob(jobId: string): void;
45+
startJob(job: JsDownloadJob, onEvent: (err: Error | null, event: JobEvent) => void): Promise<void>;
46+
}
47+
48+
interface NativeAddon {
49+
downloadGgufFile(
50+
modelId: string,
51+
filename: string,
52+
destPath: string,
53+
token: string | undefined | null,
54+
onProgress: (err: Error | null, progress: NativeDownloadProgress) => void
55+
): Promise<void>;
56+
DownloadManager: new () => NativeDownloadManager;
57+
}
58+
59+
// The Rust addon is built by `lib`'s `napi build` step into app/native/ as a
60+
// plain filesystem artifact (not an npm dependency under node_modules —
61+
// see the plan for why: a `file:` dependency would need `npm ci` to
62+
// reconcile a different lockfile integrity hash per platform's .node
63+
// binary). `dist/native-downloader.js` sits one level under app/, so
64+
// app/native is a sibling reached by going up just one directory. Electron
65+
// transparently redirects native-module requires from inside app.asar to
66+
// the asarUnpack'd app.asar.unpacked counterpart at packaging time.
67+
let nativeAddon: NativeAddon | undefined;
68+
69+
function getNativeAddon(): NativeAddon {
70+
if (!nativeAddon) {
71+
nativeAddon = require(path.join(__dirname, "..", "native")) as NativeAddon;
72+
}
73+
return nativeAddon;
74+
}
75+
76+
export function downloadGgufFileNative(
77+
modelId: string,
78+
filename: string,
79+
destPath: string,
80+
token: string | undefined | null,
81+
onProgress: (err: Error | null, progress: NativeDownloadProgress) => void
82+
): Promise<void> {
83+
return getNativeAddon().downloadGgufFile(modelId, filename, destPath, token, onProgress);
84+
}
85+
86+
// One instance for the app's lifetime — its global concurrency semaphore
87+
// and bandwidth limiter are only meaningfully "global" if nothing else
88+
// constructs a second one. Purely additive to `downloadGgufFileNative`
89+
// above, which stays exactly as-is for the existing single-file `hf:
90+
// downloadFile` flow.
91+
let downloadManager: NativeDownloadManager | undefined;
92+
93+
export function getDownloadManager(): NativeDownloadManager {
94+
if (!downloadManager) {
95+
const { DownloadManager } = getNativeAddon();
96+
downloadManager = new DownloadManager();
97+
}
98+
return downloadManager;
99+
}

0 commit comments

Comments
 (0)