feat(docs): fetch custom bundle packages in parallel

2026-10-21 06:00:39 +00:00 · 2026-10-04 20:57:55 +08:00
parent 22c384517e
commit 36ac001d70
3 changed files with 136 additions and 60 deletions
@@ -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<Uint8Array> => {
-    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
                      <LandingSelect<BundleSource>
                        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}
@@ -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<BundleProgress, { phase: "download" }> => 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);
@@ -404,8 +404,7 @@ export function buildStoredZipParts(entries: StoredZipEntry[], date = new Date()
 export type BundleFetch = (url: string, onProgress: (received: number) => void, signal: AbortSignal) => Promise<Uint8Array>;

 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<string> {
  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<void>>, limit: number): Promise<void> {
+  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<string, PackageDriverEntry>, extracted: ExtractedDriverPackage): void {
  const existing = target[extracted.driverKey];
  if (!existing) {
@@ -435,9 +461,10 @@ function mergeDriverEntry(target: Record<string, PackageDriverEntry>, 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<string, PackageDriverEntry> = {};
-  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<string, ExtractedDriverPackage>();
+  const receivedByUrl = new Map<string, number>();
+  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<Uint8Array> => {
+    const bytes = await fetchPackage(
+      url,
+      (received) => {
+        receivedByUrl.set(url, received);
+        reportDownload();
+      },
+      signal,
+    );
+    receivedByUrl.set(url, bytes.length);
+    return bytes;
+  };
+
+  const tasks: Array<() => Promise<void>> = 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<string, PackageDriverEntry> = {};
+  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<string, { version: string; platforms: Record<string, { url: string; sha256?: string; size: number; format: "tar_zstd" }> }> = {};
-  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,