From 36ac001d7009db2a8a7311415f33dc0cf5444f7a Mon Sep 17 00:00:00 2001 From: t8y2 Date: Sun, 4 Oct 2026 20:57:55 +0800 Subject: [PATCH] feat(docs): fetch custom bundle packages in parallel --- docs/app/[lang]/drivers/DriversClient.tsx | 29 ++--- docs/lib/agentBundle.test.ts | 24 +++- docs/lib/agentBundle.ts | 143 ++++++++++++++++------ 3 files changed, 136 insertions(+), 60 deletions(-) diff --git a/docs/app/[lang]/drivers/DriversClient.tsx b/docs/app/[lang]/drivers/DriversClient.tsx index 7412879feb..b7f899f0f3 100644 --- a/docs/app/[lang]/drivers/DriversClient.tsx +++ b/docs/app/[lang]/drivers/DriversClient.tsx @@ -51,8 +51,6 @@ const i18n = { customIncludeJre: `Include JRE 21`, customIncludeJreForced: "Included automatically — Java agents require the JRE", customSource: "Fetch source", - customSourceCnb: "CNB (direct)", - customSourceGithub: "GitHub (via this site)", customDownload: "Download bundle", customEstimate: "Estimated download", customSelected: "selected", @@ -61,7 +59,6 @@ const i18n = { customWorkersNote: "The SQLite SSH Worker Linux packages are always included — remote SSH hosts execute them.", customNoDrivers: "No drivers match this platform or search.", customPhaseDownload: "Downloading", - customPhaseUnpack: "Unpacking", customPhaseAssemble: "Assembling ZIP", customPhaseDone: "Custom bundle downloaded ({size}). Import it in DBX from Settings > Driver Manager > offline import.", customErrorTitle: "Bundle build failed", @@ -111,8 +108,6 @@ const i18n = { customIncludeJre: "包含 JRE 21", customIncludeJreForced: "已自动包含——Java 驱动必须搭配 JRE", customSource: "下载源", - customSourceCnb: "CNB（直连）", - customSourceGithub: "GitHub（经本站代理）", customDownload: "下载离线包", customEstimate: "预计下载量", customSelected: "已选", @@ -121,7 +116,6 @@ const i18n = { customWorkersNote: "SQLite SSH Worker 的 Linux 包会自动包含——它们在远端 SSH 主机上执行。", customNoDrivers: "没有匹配该平台或搜索词的驱动。", customPhaseDownload: "正在下载", - customPhaseUnpack: "正在解包", customPhaseAssemble: "正在打包", customPhaseDone: "自定义离线包已下载（{size}）。在 DBX 的“设置 > 驱动管理”中离线导入即可。", customErrorTitle: "打包失败", @@ -301,17 +295,16 @@ export function DriversClient({ initialCatalog, initialRegistry }: { initialCata }; const fetchBundlePackage = async (url: string, onProgress: (received: number) => void, signal: AbortSignal): Promise => { - const proxy = proxyUrl(window.location.origin, url); - const mirror = cnbMirrorUrl(url); + // The selected source is tried first; the other is a silent fallback. const attempts: Array<{ label: string; url: string }> = bundleSource === "cnb" ? [ - { label: t.customSourceCnb, url: mirror }, - { label: t.customSourceGithub, url: proxy }, + { label: "CNB", url: cnbMirrorUrl(url) }, + { label: "GitHub", url: proxyUrl(window.location.origin, url) }, ] : [ - { label: t.customSourceGithub, url: proxy }, - { label: t.customSourceCnb, url: mirror }, + { label: "GitHub", url: proxyUrl(window.location.origin, url) }, + { label: "CNB", url: cnbMirrorUrl(url) }, ]; const errors: string[] = []; for (const attempt of attempts) { @@ -388,12 +381,10 @@ export function DriversClient({ initialCatalog, initialRegistry }: { initialCata const bundleProgressText = useMemo(() => { if (!bundleProgress) return ""; if (bundleProgress.phase === "download") { - const expected = bundleProgress.expectedBytes > 0 ? ` / ${formatSize(bundleProgress.expectedBytes)}` : ""; - return `${t.customPhaseDownload} ${bundleProgress.label} (${bundleProgress.index}/${bundleProgress.total}) · ${formatSize(bundleProgress.received)}${expected}`; + const expected = bundleProgress.totalBytes > 0 ? ` / ${formatSize(bundleProgress.totalBytes)}` : ""; + return `${t.customPhaseDownload} ${bundleProgress.completed}/${bundleProgress.total} · ${formatSize(bundleProgress.receivedBytes)}${expected}`; } - if (bundleProgress.phase === "unpack") return `${t.customPhaseUnpack} ${bundleProgress.label}`; - if (bundleProgress.phase === "assemble") return t.customPhaseAssemble; - return ""; + return t.customPhaseAssemble; }, [bundleProgress, t]); const activeCount = activeTab === "bundles" ? filteredBundles.length : activeTab === "custom" ? filteredBundleOptions.length : activeTab === "drivers" ? filteredDrivers.length : activeTab === "native" ? filteredNativeGroups.length : activeTab === "jre" ? filteredJres.length : filteredJdbcPlugin.length; @@ -580,8 +571,8 @@ export function DriversClient({ initialCatalog, initialRegistry }: { initialCata value={bundleSource} options={[ - { value: "cnb", label: t.customSourceCnb }, - { value: "github", label: t.customSourceGithub }, + { value: "cnb", label: "CNB" }, + { value: "github", label: "GitHub" }, ]} onChange={setBundleSource} disabled={bundleBusy} diff --git a/docs/lib/agentBundle.test.ts b/docs/lib/agentBundle.test.ts index a949ea40d6..01a09a5eb1 100644 --- a/docs/lib/agentBundle.test.ts +++ b/docs/lib/agentBundle.test.ts @@ -285,8 +285,9 @@ test("assemble builds a driver-manager compatible offline ZIP end to end", async const assembled = await assembleCustomBundle(plan, fetchPackage, (event) => progress.push(event), signal); assert.equal(assembled.filename, "dbx-agents-offline-custom-windows-x64.zip"); - assert.ok(progress.some((event) => event.phase === "download" && event.label === "H2")); - assert.ok(progress.some((event) => event.phase === "unpack")); + const downloads = progress.filter((event): event is Extract => event.phase === "download"); + assert.ok(downloads.some((event) => event.completed === 4), "all four jobs must report completion"); + assert.equal(downloads.at(-1)?.receivedBytes, h2Package.length + workerX64Package.length + workerArmPackage.length + JRE_BYTES.length); assert.ok(progress.some((event) => event.phase === "assemble")); assert.ok(progress.some((event) => event.phase === "done")); @@ -317,6 +318,25 @@ test("assemble builds a driver-manager compatible offline ZIP end to end", async assert.equal(fetched.length, 4); }); +test("bundle downloads run in parallel under the concurrency limit", async () => { + const plan = computeBundlePlan(testRegistry, "windows-x64", new Set(["h2"]), false); + assert.equal(plan.drivers.length + plan.workers.length + (plan.jre ? 1 : 0), 4); + let inFlight = 0; + let maxInFlight = 0; + const fetchPackage: BundleFetch = async (url) => { + inFlight += 1; + maxInFlight = Math.max(maxInFlight, inFlight); + await new Promise((resolve) => setTimeout(resolve, 15)); + inFlight -= 1; + if (url.endsWith("dbx-agent-h2-1.0.0.tar.zst")) return h2Package; + if (url.endsWith("linux-x64.tar.zst")) return workerX64Package; + if (url.endsWith("linux-aarch64.tar.zst")) return workerArmPackage; + return JRE_BYTES; + }; + await assembleCustomBundle(plan, fetchPackage, () => {}, new AbortController().signal); + assert.equal(maxInFlight, 4); +}); + test("assembly rejects packages whose checksum does not match the registry", async () => { const corrupted: AgentRegistry = JSON.parse(JSON.stringify(testRegistry)); corrupted.drivers!.h2!.jar!.sha256 = "f".repeat(64); diff --git a/docs/lib/agentBundle.ts b/docs/lib/agentBundle.ts index b9189bbb76..3f189f6d51 100644 --- a/docs/lib/agentBundle.ts +++ b/docs/lib/agentBundle.ts @@ -404,8 +404,7 @@ export function buildStoredZipParts(entries: StoredZipEntry[], date = new Date() export type BundleFetch = (url: string, onProgress: (received: number) => void, signal: AbortSignal) => Promise; export type BundleProgress = - | { phase: "download"; label: string; index: number; total: number; received: number; expectedBytes: number } - | { phase: "unpack"; label: string } + | { phase: "download"; completed: number; total: number; receivedBytes: number; totalBytes: number } | { phase: "assemble" } | { phase: "done"; filename: string; size: number }; @@ -419,6 +418,33 @@ async function sha256Hex(bytes: Uint8Array): Promise { return Array.from(new Uint8Array(digest), (byte) => byte.toString(16).padStart(2, "0")).join(""); } +/** Parallel package downloads; 4 keeps the peak heap bounded by a few packages. */ +export const BUNDLE_DOWNLOAD_CONCURRENCY = 4; + +async function runPool(tasks: Array<() => Promise>, limit: number): Promise { + let cursor = 0; + const failure: { current: { error: unknown } | null } = { current: null }; + const worker = async () => { + for (;;) { + if (failure.current) return; + const index = cursor++; + if (index >= tasks.length) return; + try { + await tasks[index]!(); + } catch (error) { + failure.current ??= { error }; + return; + } + } + }; + await Promise.all(Array.from({ length: Math.max(1, Math.min(limit, tasks.length)) }, worker)); + if (failure.current) throw failure.current.error; +} + +function describe(error: unknown): string { + return error instanceof Error ? error.message : String(error); +} + function mergeDriverEntry(target: Record, extracted: ExtractedDriverPackage): void { const existing = target[extracted.driverKey]; if (!existing) { @@ -435,9 +461,10 @@ function mergeDriverEntry(target: Record, extracted: } /** - * Downloads every planned package, unpacks the raw artifacts, verifies the - * package-provided SHA-256 digests, and assembles the final driver-manager - * compatible ZIP. `fetchPackage` is injected so tests can stub the network. + * Downloads the planned packages in parallel (CNB first with a silent fallback + * handled by the injected `fetchPackage`), unpacks and verifies each raw + * artifact, then assembles the driver-manager compatible ZIP. The first + * failing package stops the pool; in-flight downloads still settle. */ export async function assembleCustomBundle( plan: BundlePlan, @@ -449,59 +476,97 @@ export async function assembleCustomBundle( if (plan.drivers.length === 0) throw new Error("Select at least one driver"); const hash = deps.sha256 ?? sha256Hex; const items = [...plan.drivers, ...plan.workers]; - const drivers: Record = {}; - const zipEntries: StoredZipEntry[] = []; - let rawBytes = 0; + const jre = plan.jre; + const total = items.length + (jre ? 1 : 0); + const totalBytes = items.reduce((sum, item) => sum + item.packageSize, 0) + (jre?.size ?? 0); + const extractedByUrl = new Map(); + const receivedByUrl = new Map(); + const jreResult: { bytes: Uint8Array | null } = { bytes: null }; + let completed = 0; - for (let index = 0; index < items.length; index += 1) { - const item = items[index]; + const reportDownload = () => { + let receivedBytes = 0; + for (const value of receivedByUrl.values()) receivedBytes += value; + onProgress({ phase: "download", completed, total, receivedBytes, totalBytes }); + }; + + const downloadOne = async (url: string, expectedBytes: number): Promise => { + const bytes = await fetchPackage( + url, + (received) => { + receivedByUrl.set(url, received); + reportDownload(); + }, + signal, + ); + receivedByUrl.set(url, bytes.length); + return bytes; + }; + + const tasks: Array<() => Promise> = items.map((item) => async () => { signal.throwIfAborted(); - onProgress({ phase: "download", label: item.label, index: index + 1, total: items.length + (plan.jre ? 1 : 0), received: 0, expectedBytes: item.packageSize }); let packageBytes: Uint8Array; try { - packageBytes = await fetchPackage(item.packageUrl, (received) => { - onProgress({ phase: "download", label: item.label, index: index + 1, total: items.length + (plan.jre ? 1 : 0), received, expectedBytes: item.packageSize }); - }, signal); + packageBytes = await downloadOne(item.packageUrl, item.packageSize); } catch (error) { if (signal.aborted) throw error; - throw new Error(`${item.label} — ${error instanceof Error ? error.message : String(error)}`); + throw new Error(`${item.label} — ${describe(error)}`); } if (item.packageSha256) { const actual = await hash(packageBytes); - if (actual !== item.packageSha256.toLowerCase()) throw new Error(`Package checksum mismatch for ${item.label}`); + if (actual !== item.packageSha256.toLowerCase()) throw new Error(`${item.label} — package checksum mismatch`); } - onProgress({ phase: "unpack", label: item.label }); - const extracted = extractDriverPackage(packageBytes, item.key); - if (extracted.artifactKind !== item.kind) throw new Error(`Package kind mismatch for ${item.label}: expected ${item.kind}`); + let extracted: ExtractedDriverPackage; + try { + extracted = extractDriverPackage(packageBytes, item.key); + } catch (error) { + throw new Error(`${item.label} — ${describe(error)}`); + } + if (extracted.artifactKind !== item.kind) throw new Error(`${item.label} — package kind mismatch: expected ${item.kind}`); if (extracted.sha256) { const actual = await hash(extracted.bytes); - if (actual !== extracted.sha256.toLowerCase()) throw new Error(`Artifact checksum mismatch for ${item.label}`); + if (actual !== extracted.sha256.toLowerCase()) throw new Error(`${item.label} — artifact checksum mismatch`); } + extractedByUrl.set(item.packageUrl, extracted); + completed += 1; + reportDownload(); + }); + if (jre) { + tasks.push(async () => { + signal.throwIfAborted(); + let bytes: Uint8Array; + try { + bytes = await downloadOne(jre.url, jre.size); + } catch (error) { + if (signal.aborted) throw error; + throw new Error(`JRE ${jre.key} — ${describe(error)}`); + } + if (jre.sha256) { + const actual = await hash(bytes); + if (actual !== jre.sha256.toLowerCase()) throw new Error(`JRE ${jre.key} — checksum mismatch`); + } + jreResult.bytes = bytes; + completed += 1; + reportDownload(); + }); + } + + await runPool(tasks, BUNDLE_DOWNLOAD_CONCURRENCY); + + const drivers: Record = {}; + const zipEntries: StoredZipEntry[] = []; + let rawBytes = 0; + for (const item of items) { + const extracted = extractedByUrl.get(item.packageUrl); + if (!extracted) throw new Error(`${item.label} — package was not downloaded`); mergeDriverEntry(drivers, extracted); zipEntries.push({ name: `drivers/${extracted.filename}`, data: extracted.bytes }); rawBytes += extracted.bytes.length; } let jres: Record }> = {}; - if (plan.jre) { - signal.throwIfAborted(); - const jre = plan.jre; - const index = items.length + 1; - const total = index; - onProgress({ phase: "download", label: `JRE ${jre.key}`, index, total, received: 0, expectedBytes: jre.size }); - let jreBytes: Uint8Array; - try { - jreBytes = await fetchPackage(jre.url, (received) => { - onProgress({ phase: "download", label: `JRE ${jre.key}`, index, total, received, expectedBytes: jre.size }); - }, signal); - } catch (error) { - if (signal.aborted) throw error; - throw new Error(`JRE ${jre.key} — ${error instanceof Error ? error.message : String(error)}`); - } - if (jre.sha256) { - const actual = await hash(jreBytes); - if (actual !== jre.sha256.toLowerCase()) throw new Error("JRE package checksum mismatch"); - } + const jreBytes = jreResult.bytes; + if (jre && jreBytes) { jres = { [jre.key]: { version: jre.version,