fix(nav): bound section-scan concurrency to keep zip.js writers from ERRORED-ing (#4528)

On Tauri, section.loadText() drives a plugin-fs open/read/close round trip per call. The unbounded Promise.all in computeBookNav and enrichTocFromNavElements can fire 200+ concurrent IPC chains against a long-spine EPUB, saturate the JS↔Rust bridge or the fd pool, and cause individual reads to reject. The zip.js TextWriter then transitions to ERRORED, surfacing as 'Cannot close a ERRORED writable stream' and silently dropping TOC fragments for the affected sections. In the worst case the rejection propagates through Promise.all and prevents the reader from opening the book.

Hoist the OPDS module's runWithConcurrency to utils/concurrency.ts (zero behaviour change for OPDS) and reuse it in computeBookNav and enrichTocFromNavElements, capped at 128. The cap was binary-searched against the worst-case repro (Android emulator + dev mode + 250-section EPUB): 30/64/128 pass, 200 fails. Section-internal loadText/createDocument dedupe is unchanged.

The worker pool also isolates per-section failures: the outcome shape ({item,result}|{item,error}) lets us log and skip the offending section instead of aborting the entire build as Promise.all did. Even if a future workload pushes past the cap, the reader still opens.

apps/readest-app/src/__tests__/utils/concurrency.test.ts

@@ -0,0 +1,102 @@
+import { describe, expect, test } from 'vitest';
+import { runWithConcurrency } from '@/utils/concurrency';
+
+/**
+ * Tests for the shared bounded-concurrency runner used by nav build,
+ * embedded-nav enrichment, and OPDS auto-download. Three properties matter
+ * to those callers and are pinned down here:
+ *
+ *   1. In-flight task count never exceeds the requested cap. The whole
+ *      reason this primitive exists is to stop nav build from saturating
+ *      the Tauri IPC bridge / fd pool, so a regression here would silently
+ *      reintroduce the "Cannot close a ERRORED writable stream" failure
+ *      mode on Android.
+ *   2. Output is positionally aligned with input even when tasks complete
+ *      out of order. OPDS callers index into the result array assuming
+ *      `results[i]` is the outcome of `items[i]`.
+ *   3. A throwing task does not abort sibling tasks — failures are
+ *      surfaced as `{ item, error }` outcomes alongside successes. This
+ *      is what lets nav build keep going when one section's inflate
+ *      genuinely fails.
+ */
+
+const wait = (ms: number) => new Promise<void>((resolve) => setTimeout(resolve, ms));
+
+describe('runWithConcurrency', () => {
+  test('caps simultaneous in-flight tasks at the requested concurrency', async () => {
+    const N = 30;
+    const CAP = 4;
+    let active = 0;
+    let peak = 0;
+
+    await runWithConcurrency(
+      Array.from({ length: N }, (_, i) => i),
+      CAP,
+      async () => {
+        active += 1;
+        if (active > peak) peak = active;
+        // A short await is enough to exercise the bound: every worker
+        // hits this point once before it has a chance to release, so
+        // peak observably reaches `CAP` (and never exceeds it).
+        await wait(5);
+        active -= 1;
+      },
+    );
+
+    expect(peak).toBe(CAP);
+    expect(active).toBe(0);
+  });
+
+  test('preserves positional alignment with the input array', async () => {
+    const items = Array.from({ length: 10 }, (_, i) => i);
+
+    // Reverse the natural completion order: large indices finish first,
+    // small indices last. If the runner appended results in completion
+    // order, the assertion below would fail.
+    const outcomes = await runWithConcurrency(items, 4, async (i) => {
+      await wait((10 - i) * 2);
+      return i * 10;
+    });
+
+    expect(outcomes).toHaveLength(items.length);
+    outcomes.forEach((o, i) => {
+      expect(o.item).toBe(items[i]);
+      expect('result' in o ? o.result : null).toBe(i * 10);
+    });
+  });
+
+  test('isolates failures so siblings still run to completion', async () => {
+    const items = ['ok-1', 'fail', 'ok-2'];
+
+    const outcomes = await runWithConcurrency(items, 2, async (item) => {
+      if (item === 'fail') throw new Error('boom');
+      return item.toUpperCase();
+    });
+
+    expect(outcomes.map((o) => ('error' in o ? 'fail' : o.result))).toEqual([
+      'OK-1',
+      'fail',
+      'OK-2',
+    ]);
+    const failed = outcomes.find((o) => 'error' in o);
+    expect(failed?.error).toBeInstanceOf(Error);
+    expect((failed?.error as Error).message).toBe('boom');
+  });
+
+  test('returns immediately on an empty input', async () => {
+    const outcomes = await runWithConcurrency<number, number>([], 4, async (x) => x);
+    expect(outcomes).toEqual([]);
+  });
+
+  test('normalises non-positive concurrency to a single worker', async () => {
+    let peak = 0;
+    let active = 0;
+    await runWithConcurrency([1, 2, 3, 4], 0, async () => {
+      active += 1;
+      if (active > peak) peak = active;
+      await wait(2);
+      active -= 1;
+    });
+    expect(peak).toBe(1);
+  });
+});

apps/readest-app/src/services/nav/enrichment.ts

@@ -8,11 +8,21 @@
 // elements and merge their links as ordinary top-level TOC items.
 
 import { BookDoc, TOCItem } from '@/libs/document';
+import { runWithConcurrency } from '@/utils/concurrency';
 import { collectAllTocItems } from './grouping';
 
 const NAV_ENRICH_MIN_SECTIONS = 64;
 const NAV_ENRICH_TOC_RATIO = 8;
 const EPUB_NS = 'http://www.idpf.org/2007/ops';
+/**
+ * Mirror `NAV_BUILD_CONCURRENCY` from ./index.ts. Sparse-ncx EPUBs that
+ * trigger this enrichment are exactly the ones with a long spine
+ * (NAV_ENRICH_MIN_SECTIONS = 64+ entries) — i.e. the same workload that
+ * blew up the unbounded Promise.all-based scan on Tauri Android. Cap to
+ * keep simultaneous in-flight section reads bounded; see the docblock
+ * on `NAV_BUILD_CONCURRENCY` for the failure mode that drove this.
+ */
+const NAV_ENRICH_CONCURRENCY = 128;
 
 interface ExtractedNavItem {
   label: string;
@@ -104,19 +114,21 @@ export const enrichTocFromNavElements = async (
 
   const sectionIdSet = new Set(sections.map((s) => s.id));
 
-  // Two-phase scan:
-  //   1) Concurrently `loadText` every section and quick-filter on
+  // Two-phase scan, each phase bounded by NAV_ENRICH_CONCURRENCY:
+  //   1) `loadText` every section and quick-filter on
   //      `content.includes('<nav')` — cheap (ASCII substring on the raw
   //      HTML) and immediately discards 90%+ of chapters that have no
-  //      embedded nav. Concurrency lets the zip inflates overlap.
-  //   2) For survivors, concurrently `createDocument` + walk `<nav>`
-  //      elements. The makeZipLoader dedupes in-flight loadText calls by
-  //      name, so phase 2's `createDocument()` reuses phase 1's inflated
-  //      bytes when the two awaits overlap.
+  //      embedded nav. Bounded concurrency lets the zip inflates overlap
+  //      without saturating the Tauri IPC bridge / fd pool.
+  //   2) For survivors, `createDocument` + walk `<nav>` elements. The
+  //      makeZipLoader dedupes in-flight loadText calls by name, so when
+  //      phase 2 fires inside the same microtask span as phase 1's tail
+  //      its `createDocument()` reuses phase 1's inflated bytes.
   type Survivor = { section: (typeof sections)[number] };
-  const survivors: Survivor[] = [];
-  const phase1 = await Promise.all(
-    sections.map(async (section) => {
+  const phase1Outcomes = await runWithConcurrency(
+    sections,
+    NAV_ENRICH_CONCURRENCY,
+    async (section): Promise<Survivor | null> => {
       if (!section.loadText) return null;
       try {
         const content = await section.loadText();
@@ -125,12 +137,18 @@ export const enrichTocFromNavElements = async (
       } catch {
         return null;
       }
-    }),
+    },
   );
-  for (const s of phase1) if (s) survivors.push(s);
+  const survivors: Survivor[] = [];
+  for (const o of phase1Outcomes) {
+    if ('result' in o && o.result) survivors.push(o.result);
+  }
 
-  const phase2 = await Promise.all(
-    survivors.map(async ({ section }) => {
+  type NavMatch = { sectionId: string; navs: Element[] };
+  const phase2Outcomes = await runWithConcurrency(
+    survivors,
+    NAV_ENRICH_CONCURRENCY,
+    async ({ section }): Promise<NavMatch | null> => {
       let doc: Document;
       try {
         doc = await section.createDocument();
@@ -140,14 +158,14 @@ export const enrichTocFromNavElements = async (
       const navs = Array.from(doc.getElementsByTagName('nav'));
       if (navs.length === 0) return null;
       return { sectionId: section.id, navs };
-    }),
+    },
   );
 
   const collected: ExtractedNavItem[] = [];
   const seenHref = new Set<string>();
-  for (const result of phase2) {
-    if (!result) continue;
-    const { sectionId, navs } = result;
+  for (const o of phase2Outcomes) {
+    if (!('result' in o) || !o.result) continue;
+    const { sectionId, navs } = o.result;
     for (const navEl of navs) {
       const extracted = extractTocFromNav(navEl, sectionId);
       for (const item of extracted) {

apps/readest-app/src/services/nav/index.ts

@@ -1,6 +1,7 @@
 import { ConvertChineseVariant } from '@/types/book';
 import { BookDoc, SectionFragment, TOCItem } from '@/libs/document';
 import { initSimpleCC, runSimpleCC } from '@/utils/simplecc';
+import { runWithConcurrency } from '@/utils/concurrency';
 import {
   cloneSectionFragments,
   cloneTocItems,
@@ -12,6 +13,43 @@ import { bakeLocationsAndCfis, sortTocItems } from './locations';
 import { buildSectionFragments } from './fragments';
 import { enrichTocFromNavElements } from './enrichment';
 
+/**
+ * Bound on simultaneous in-flight section reads during nav build.
+ *
+ * The original perf commit fanned every section out via Promise.all under
+ * the assumption that `section.loadText()` is a cheap in-memory inflate.
+ * That holds for pure-web targets (the EPUB Blob is an in-memory File
+ * backed by IndexedDB, slice/arrayBuffer is synchronous), but on Tauri
+ * each loadText drives a `@tauri-apps/plugin-fs` open/read/close round
+ * trip across the JS<->Rust IPC bridge: see `utils/file.ts::NativeFile`.
+ * Past a per-platform threshold the bridge or the fd pool saturates,
+ * individual reads reject, and the zip.js TextWriter driving that
+ * entry's inflate transitions to ERRORED, surfacing as
+ *   "Cannot close a ERRORED writable stream"
+ * with the affected sections silently losing their TOC fragments.
+ *
+ * The cap was picked empirically against the worst-case combination we
+ * could reproduce — Android emulator + dev mode + a 250-section EPUB —
+ * binary-searching the threshold:
+ *   - 30 ✅, 64 ✅, 128 ✅
+ *   - 200 ❌ (≈ 30 contiguous sections fail with the ERRORED message)
+ * Real devices (release build, native fd pool, no Next.js dev tooling
+ * sharing the event loop) sit well above this, so 128 leaves a healthy
+ * margin while preserving close-to-unbounded cross-section inflate
+ * overlap during cold nav build (CPU `parseFromString` for one section
+ * runs while the next several inflate). Higher caps yield diminishing
+ * returns: parseFromString is CPU-bound on the main thread and
+ * dominates total nav-build time once IPC latency is paid in parallel.
+ *
+ * Failure-isolation note: `runWithConcurrency` swallows per-task
+ * exceptions into the result tuple, so even if a future workload pushes
+ * past this cap on an unfamiliar platform, individual section failures
+ * degrade gracefully (the caller logs them and skips the offending
+ * fragments) instead of aborting the whole nav build the way the
+ * original Promise.all path did.
+ */
+const NAV_BUILD_CONCURRENCY = 128;
+
 export { findParentPath, findTocItemBS } from './lookup';
 export type { SectionFragment };
 
@@ -97,27 +135,28 @@ export const computeBookNav = async (bookDoc: BookDoc): Promise<BookNav> => {
   const groups = groupItemsBySection(bookDoc, allItems);
   const splitHref = (href: string) => bookDoc.splitTOCHref(href);
 
-  // Process sections concurrently. Each section's work is independent: the
-  // only shared writes are into `sections` (keyed by sectionId, no
-  // collisions) and into the local `bookSections` later (after all promises
-  // resolve). Concurrency here lets the zip-text inflate calls overlap with
-  // each other and with the `parseFromString` of earlier sections — on iOS
-  // WebView this turns the previously-serial chapter loop into something
-  // bounded by the slowest single inflate + parse rather than their sum.
-  //
-  // We don't bound concurrency: a typical EPUB has ≲ 200 sections and the
-  // zip.js loader is happy to interleave them; if memory pressure ever
-  // matters we can drop in a small p-limit-style throttle.
-  const sectionResults = await Promise.all(
-    Array.from(groups.entries()).map(async ([sectionId, { base, fragments }]) => {
+  // Process sections with a bounded worker pool. Each section's work is
+  // independent: the only shared writes are into `sections` (keyed by
+  // sectionId, no collisions) and into `bookSections` later (after all
+  // workers settle). The cross-section inflate overlap that motivated the
+  // original parallel rewrite is preserved — we just cap simultaneous
+  // in-flight reads at NAV_BUILD_CONCURRENCY so we don't saturate the
+  // platform's IPC bridge / fd pool. See the constant's docblock for the
+  // failure mode that drove this cap.
+  type SectionEntry = [string, BookNavSection];
+  const taskOutcomes = await runWithConcurrency(
+    Array.from(groups.entries()),
+    NAV_BUILD_CONCURRENCY,
+    async ([sectionId, { base, fragments }]): Promise<SectionEntry | null> => {
       const section = sectionMap.get(sectionId);
       if (!section || fragments.length === 0) return null;
       if (!section.loadText) return null;
 
-      // Issue both the raw-text and DOM-parse loads concurrently. The
-      // makeZipLoader now dedupes in-flight loadText calls by name, so the
-      // two awaits below resolve from a single zip inflate per section
-      // instead of two.
+      // Issue both the raw-text and DOM-parse loads concurrently within
+      // this section. makeZipLoader's in-flight loadText dedupe (see
+      // libs/document.ts) collapses these onto a single zip inflate when
+      // both await the same href in the same microtask span, so the cost
+      // is one read + one parse, not two reads.
       const contentP = section.loadText();
       const docP = section.createDocument().catch((e: unknown) => {
         console.warn(`Failed to parse section ${sectionId} for fragment CFIs:`, e);
@@ -136,12 +175,20 @@ export const computeBookNav = async (bookDoc: BookDoc): Promise<BookNav> => {
         splitHref,
       );
       if (sectionFragments.length === 0) return null;
-      return [sectionId, { id: sectionId, fragments: sectionFragments }] as const;
-    }),
+      return [sectionId, { id: sectionId, fragments: sectionFragments }];
+    },
   );
 
-  for (const r of sectionResults) {
-    if (r) sections[r[0]] = r[1];
+  for (const outcome of taskOutcomes) {
+    if ('error' in outcome) {
+      const [sectionId] = outcome.item;
+      console.warn(`Failed to build section ${sectionId} for fragment CFIs:`, outcome.error);
+      continue;
+    }
+    if (outcome.result) {
+      const [id, section] = outcome.result;
+      sections[id] = section;
+    }
   }
 
   // Attach the freshly computed fragments to live sections so the bake can see

apps/readest-app/src/services/opds/autoDownload.ts

@@ -16,6 +16,7 @@ import {
 import { upsertOPDSSourceMapping } from './sourceMap';
 import { isRetryEligible, DOWNLOAD_CONCURRENCY, MAX_RETRY_ATTEMPTS } from './types';
 import type { PendingItem, SyncResult, OPDSSubscriptionState, FailedEntry } from './types';
+import { runWithConcurrency } from '@/utils/concurrency';
 
 /**
  * Download a single item and import it into the library.
@@ -100,35 +101,6 @@ async function downloadAndImport(
   return book;
 }
 
-/**
- * Run a batch of async tasks with bounded concurrency.
- */
-async function runWithConcurrency<T, R>(
-  items: T[],
-  concurrency: number,
-  fn: (item: T) => Promise<R>,
-): Promise<Array<{ item: T; result: R } | { item: T; error: unknown }>> {
-  const results: Array<{ item: T; result: R } | { item: T; error: unknown }> = [];
-  let index = 0;
-
-  async function worker() {
-    while (index < items.length) {
-      const currentIndex = index++;
-      const item = items[currentIndex]!;
-      try {
-        const result = await fn(item);
-        results[currentIndex] = { item, result };
-      } catch (error) {
-        results[currentIndex] = { item, error };
-      }
-    }
-  }
-
-  const workers = Array.from({ length: Math.min(concurrency, items.length) }, () => worker());
-  await Promise.all(workers);
-  return results;
-}
-
 /**
  * Sync a single catalog: discover new items, retry failed, download, update state.
  */

apps/readest-app/src/utils/concurrency.ts

@@ -0,0 +1,65 @@
+/**
+ * Generic bounded-concurrency runner.
+ *
+ * Spawns up to `concurrency` workers that pull from a shared cursor over
+ * `items` and invoke `fn` on each. Each worker awaits its own task before
+ * pulling the next, so at any instant the in-flight count is bounded by
+ * `min(concurrency, items.length)` — the natural primitive for "I have a
+ * list of N independent async tasks and want to fan them out to a fixed
+ * pool" without pulling in `p-limit`.
+ *
+ * Why not throw on the first failure (Promise.all semantics):
+ *   - Several call sites (OPDS sync, nav build) treat the per-item
+ *     failure as a recoverable, individually-loggable event. A single
+ *     bad EPUB section, or a single 503'd OPDS download, must not
+ *     abort the rest of the batch.
+ *   - The shape `{ item, result } | { item, error }` lets callers
+ *     post-process success and failure separately without an extra
+ *     try/catch around every worker invocation.
+ *
+ * Behavioural guarantees:
+ *   - `results` is positionally aligned with `items` (results[i] is the
+ *     outcome of items[i]), even though tasks complete out of order.
+ *   - `concurrency <= 0` is normalised to a single worker so callers
+ *     don't have to special-case empty pools.
+ *   - When `items` is empty the function resolves immediately with `[]`
+ *     without spawning a worker.
+ */
+export interface ConcurrencyTaskSuccess<T, R> {
+  item: T;
+  result: R;
+}
+
+export interface ConcurrencyTaskFailure<T> {
+  item: T;
+  error: unknown;
+}
+
+export type ConcurrencyTaskOutcome<T, R> = ConcurrencyTaskSuccess<T, R> | ConcurrencyTaskFailure<T>;
+
+export async function runWithConcurrency<T, R>(
+  items: T[],
+  concurrency: number,
+  fn: (item: T) => Promise<R>,
+): Promise<ConcurrencyTaskOutcome<T, R>[]> {
+  if (items.length === 0) return [];
+  const effective = Math.max(1, Math.min(concurrency, items.length));
+  const results: ConcurrencyTaskOutcome<T, R>[] = new Array(items.length);
+  let index = 0;
+
+  async function worker() {
+    while (index < items.length) {
+      const currentIndex = index++;
+      const item = items[currentIndex]!;
+      try {
+        const result = await fn(item);
+        results[currentIndex] = { item, result };
+      } catch (error) {
+        results[currentIndex] = { item, error };
+      }
+    }
+  }
+
+  await Promise.all(Array.from({ length: effective }, () => worker()));
+  return results;
+}