From 1e26c5d765bf842bffa0c225f69817cecfbc9859 Mon Sep 17 00:00:00 2001 From: loveheaven Date: Thu, 11 Jun 2026 13:42:39 +0800 Subject: [PATCH] fix(nav): bound section-scan concurrency to keep zip.js writers from ERRORED-ing (#4528) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- .../src/__tests__/utils/concurrency.test.ts | 102 ++++++++++++++++++ .../src/services/nav/enrichment.ts | 54 ++++++---- apps/readest-app/src/services/nav/index.ts | 89 +++++++++++---- .../src/services/opds/autoDownload.ts | 30 +----- apps/readest-app/src/utils/concurrency.ts | 65 +++++++++++ 5 files changed, 272 insertions(+), 68 deletions(-) create mode 100644 apps/readest-app/src/__tests__/utils/concurrency.test.ts create mode 100644 apps/readest-app/src/utils/concurrency.ts diff --git a/apps/readest-app/src/__tests__/utils/concurrency.test.ts b/apps/readest-app/src/__tests__/utils/concurrency.test.ts new file mode 100644 index 00000000..d8ac2e7e --- /dev/null +++ b/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((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([], 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); + }); +}); diff --git a/apps/readest-app/src/services/nav/enrichment.ts b/apps/readest-app/src/services/nav/enrichment.ts index 7a705e8d..8f4afc84 100644 --- a/apps/readest-app/src/services/nav/enrichment.ts +++ b/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('` - // 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 `