From 96b819665096d34b6a948f0d06427fc98c302100 Mon Sep 17 00:00:00 2001 From: neon <18559658+neonsolstice@users.noreply.github.com> Date: Thu, 6 Aug 2026 04:11:52 +0000 Subject: [PATCH] perf(server): stream zip archive rewrites sequentially to prevent main thread blocking Stream ZIP archive entry rewrites sequentially in writeZipArchive, opening sources lazily to keep file descriptors and memory flat during file metadata operations. Closes #893 --- .../formats/cbx/cbz-zip-patcher.test.ts | 34 ++- .../file-write/formats/cbx/cbz-zip-patcher.ts | 33 +-- .../formats/cbx/cbz-zip-roundtrip.test.ts | 88 +++++++ .../formats/epub/epub-zip-patcher.test.ts | 34 ++- .../formats/epub/epub-zip-patcher.ts | 53 ++--- .../formats/shared/zip-rewrite.test.ts | 219 ++++++++++++++++++ .../file-write/formats/shared/zip-rewrite.ts | 55 +++++ 7 files changed, 453 insertions(+), 63 deletions(-) create mode 100644 server/src/modules/file-write/formats/cbx/cbz-zip-roundtrip.test.ts create mode 100644 server/src/modules/file-write/formats/shared/zip-rewrite.test.ts create mode 100644 server/src/modules/file-write/formats/shared/zip-rewrite.ts diff --git a/server/src/modules/file-write/formats/cbx/cbz-zip-patcher.test.ts b/server/src/modules/file-write/formats/cbx/cbz-zip-patcher.test.ts index 00f013c95..db2538d59 100644 --- a/server/src/modules/file-write/formats/cbx/cbz-zip-patcher.test.ts +++ b/server/src/modules/file-write/formats/cbx/cbz-zip-patcher.test.ts @@ -46,6 +46,8 @@ let lastArchive: EventEmitter & { finalize: vi.Mock; }; +const processedEntries: string[] = []; + vi.mock('archiver', () => ({ ZipArchive: vi.fn(function () { let output: EventEmitter | null = null; @@ -54,7 +56,15 @@ vi.mock('archiver', () => ({ pipe: vi.Mock; finalize: vi.Mock; }; - emitter.append = vi.fn(); + // The real archiver emits `entry` once it has consumed and written an appended source. + emitter.append = vi.fn((source: unknown, data: { name: string }) => { + if (source instanceof Readable) source.resume(); + setImmediate(() => { + processedEntries.push(data.name); + emitter.emit('entry', data); + }); + return emitter; + }); emitter.pipe = vi.fn((out: EventEmitter) => { output = out; }); @@ -76,10 +86,11 @@ const streamFrom = (value: string) => Readable.from([Buffer.from(value)]); describe('cbz-zip-patcher', () => { beforeEach(() => { vi.clearAllMocks(); + processedEntries.length = 0; mockRandomUuid.mockReturnValue('uuid-1234'); mockRename.mockResolvedValue(undefined); mockUnlink.mockResolvedValue(undefined); - mockCreateWriteStream.mockImplementation(() => new EventEmitter() as never); + mockCreateWriteStream.mockImplementation(() => new PassThrough() as never); }); it('readComicInfoFromZip finds comicinfo case-insensitively in root or nested paths', async () => { @@ -119,6 +130,25 @@ describe('cbz-zip-patcher', () => { expect(mockRename).toHaveBeenCalledWith('/books/.cbx-write-uuid-1234', '/books/a.cbz'); }); + it('opens each page stream only after the previous page was written', async () => { + const openedAfter: string[][] = []; + const pages = ['pages/001.jpg', 'pages/002.jpg', 'pages/003.jpg'].map((path) => ({ + path, + stream: vi.fn(() => { + openedAfter.push([...processedEntries]); + return streamFrom(path); + }), + })); + + mockOpenFile.mockResolvedValue({ files: pages } as never); + + await writeComicInfoToZip('/books/a.cbz', ''); + + expect(openedAfter).toEqual([[], ['pages/001.jpg'], ['pages/001.jpg', 'pages/002.jpg']]); + expect(processedEntries).toEqual(['pages/001.jpg', 'pages/002.jpg', 'pages/003.jpg', 'ComicInfo.xml']); + for (const page of pages) expect(page.stream).toHaveBeenCalledTimes(1); + }); + it('rejects when a source entry stream errors while patching', async () => { const sourceError = Object.assign(new Error('ENOENT: no such file or directory, open /books/a.cbz'), { code: 'ENOENT' }); const imageEntry = { diff --git a/server/src/modules/file-write/formats/cbx/cbz-zip-patcher.ts b/server/src/modules/file-write/formats/cbx/cbz-zip-patcher.ts index ebe43fa59..04ea02123 100644 --- a/server/src/modules/file-write/formats/cbx/cbz-zip-patcher.ts +++ b/server/src/modules/file-write/formats/cbx/cbz-zip-patcher.ts @@ -1,10 +1,8 @@ -import { createWriteStream } from 'fs'; import { dirname, join } from 'path'; import { randomUUID } from 'crypto'; -import type { Readable } from 'stream'; import * as unzipper from 'unzipper'; -import { ZipArchive, type Archiver } from 'archiver'; import { replaceFileAtomically } from '../shared/atomic-file-replace'; +import { writeZipArchive, type ZipRewriteEntry } from '../shared/zip-rewrite'; function isComicInfoEntry(entryPath: string): boolean { const normalized = entryPath.replace(/\\/g, '/').toLowerCase(); @@ -24,29 +22,16 @@ export async function writeComicInfoToZip(filePath: string, xmlContent: string): const xmlEntryPath = existing?.path ?? 'ComicInfo.xml'; const tmpPath = join(dirname(filePath), `.cbx-write-${randomUUID()}`); - const archive = new ZipArchive({ zlib: { level: 6 } }); - const output = createWriteStream(tmpPath); - - await new Promise((resolve, reject) => { - output.on('close', resolve); - output.on('error', reject); - archive.on('error', reject); - archive.pipe(output); - - for (const entry of zip.files) { - if (isComicInfoEntry(entry.path)) continue; - appendEntryStream(archive, entry, reject); - } - - archive.append(Buffer.from(xmlContent, 'utf-8'), { name: xmlEntryPath }); - void archive.finalize(); - }); + await writeZipArchive(tmpPath, rewriteEntries(zip.files, xmlEntryPath, xmlContent)); await replaceFileAtomically(tmpPath, filePath); } -function appendEntryStream(archive: Archiver, entry: { path: string; stream: () => Readable }, reject: (error: Error) => void): void { - const source = entry.stream(); - source.once('error', reject); - archive.append(source, { name: entry.path }); +function* rewriteEntries(files: readonly unzipper.File[], xmlEntryPath: string, xmlContent: string): Generator { + for (const entry of files) { + if (isComicInfoEntry(entry.path)) continue; + yield { name: entry.path, source: () => entry.stream() }; + } + + yield { name: xmlEntryPath, source: Buffer.from(xmlContent, 'utf-8') }; } diff --git a/server/src/modules/file-write/formats/cbx/cbz-zip-roundtrip.test.ts b/server/src/modules/file-write/formats/cbx/cbz-zip-roundtrip.test.ts new file mode 100644 index 000000000..b11d77dfe --- /dev/null +++ b/server/src/modules/file-write/formats/cbx/cbz-zip-roundtrip.test.ts @@ -0,0 +1,88 @@ +import { ZipArchive } from 'archiver'; +import { createWriteStream } from 'fs'; +import { mkdtemp, readdir, rm } from 'fs/promises'; +import { tmpdir } from 'os'; +import { join } from 'path'; +import * as unzipper from 'unzipper'; + +import { readComicInfoFromZip, writeComicInfoToZip } from './cbz-zip-patcher'; + +let testRoot: string; + +beforeEach(async () => { + testRoot = await mkdtemp(join(tmpdir(), 'bookorbit-cbz-roundtrip-')); +}); + +afterEach(async () => { + await rm(testRoot, { recursive: true, force: true }); +}); + +function pageBytes(index: number): Buffer { + const bytes = Buffer.allocUnsafe(512); + for (let offset = 0; offset < bytes.length; offset++) bytes[offset] = (index * 31 + offset) % 256; + return bytes; +} + +async function writeArchive(filePath: string, entries: Array<{ path: string; content: Buffer }>): Promise { + const output = createWriteStream(filePath); + const archive = new ZipArchive({ zlib: { level: 6 } }); + + await new Promise((resolve, reject) => { + output.on('close', resolve); + output.on('error', reject); + archive.on('error', reject); + archive.pipe(output); + for (const entry of entries) archive.append(entry.content, { name: entry.path }); + void archive.finalize(); + }); +} + +async function readArchive(filePath: string): Promise> { + const zip = await unzipper.Open.file(filePath); + return Promise.all(zip.files.map(async (entry) => ({ path: entry.path, content: await entry.buffer() }))); +} + +function comicPages(count: number): Array<{ path: string; content: Buffer }> { + return Array.from({ length: count }, (_unused, index) => ({ + path: `pages/${String(index + 1).padStart(3, '0')}.jpg`, + content: pageBytes(index), + })); +} + +describe('cbz zip round trip', () => { + it('preserves every page byte-for-byte and appends comic info when absent', async () => { + const target = join(testRoot, 'volume.cbz'); + const pages = comicPages(120); + await writeArchive(target, pages); + + await writeComicInfoToZip(target, 'Saga'); + + const written = await readArchive(target); + expect(written.map((entry) => entry.path)).toEqual([...pages.map((page) => page.path), 'ComicInfo.xml']); + for (const [index, page] of pages.entries()) { + expect(written[index]!.content.equals(page.content)).toBe(true); + } + await expect(readComicInfoFromZip(target)).resolves.toBe('Saga'); + }); + + it('replaces existing comic info in place without duplicating it', async () => { + const target = join(testRoot, 'volume.cbz'); + const pages = comicPages(5); + await writeArchive(target, [{ path: 'metadata/ComicInfo.XML', content: Buffer.from('Old') }, ...pages]); + + await writeComicInfoToZip(target, 'New'); + + const written = await readArchive(target); + expect(written.map((entry) => entry.path)).toEqual([...pages.map((page) => page.path), 'metadata/ComicInfo.XML']); + await expect(readComicInfoFromZip(target)).resolves.toBe('New'); + }); + + it('leaves no temporary archive behind', async () => { + const target = join(testRoot, 'volume.cbz'); + await writeArchive(target, comicPages(3)); + + await writeComicInfoToZip(target, ''); + + await expect(readdir(testRoot)).resolves.toEqual(['volume.cbz']); + }); +}); diff --git a/server/src/modules/file-write/formats/epub/epub-zip-patcher.test.ts b/server/src/modules/file-write/formats/epub/epub-zip-patcher.test.ts index ea624ade5..e53362331 100644 --- a/server/src/modules/file-write/formats/epub/epub-zip-patcher.test.ts +++ b/server/src/modules/file-write/formats/epub/epub-zip-patcher.test.ts @@ -37,6 +37,8 @@ let lastArchive: EventEmitter & { finalize: vi.Mock; }; +const processedEntries: string[] = []; + vi.mock('archiver', () => ({ ZipArchive: vi.fn(function () { let output: EventEmitter | null = null; @@ -45,7 +47,15 @@ vi.mock('archiver', () => ({ pipe: vi.Mock; finalize: vi.Mock; }; - emitter.append = vi.fn(); + // The real archiver emits `entry` once it has consumed and written an appended source. + emitter.append = vi.fn((source: unknown, data: { name: string }) => { + if (source instanceof Readable) source.resume(); + setImmediate(() => { + processedEntries.push(data.name); + emitter.emit('entry', data); + }); + return emitter; + }); emitter.pipe = vi.fn((out: EventEmitter) => { output = out; }); @@ -66,7 +76,8 @@ const streamFrom = (value: string) => Readable.from([Buffer.from(value)]); describe('epub-zip-patcher', () => { beforeEach(() => { vi.clearAllMocks(); - mockCreateWriteStream.mockImplementation(() => new EventEmitter() as never); + processedEntries.length = 0; + mockCreateWriteStream.mockImplementation(() => new PassThrough() as never); mockRename.mockResolvedValue(undefined); mockUnlink.mockResolvedValue(undefined); }); @@ -116,6 +127,25 @@ describe('epub-zip-patcher', () => { expect(mockRename).toHaveBeenCalledWith('/book.epub.tmp', '/book.epub'); }); + it('opens each carried-over entry stream only after the previous entry was written', async () => { + const openedAfter: string[][] = []; + const carried = ['OPS/ch1.xhtml', 'OPS/ch2.xhtml', 'OPS/ch3.xhtml'].map((path) => ({ + path, + stream: vi.fn(() => { + openedAfter.push([...processedEntries]); + return streamFrom(path); + }), + })); + + mockOpenFile.mockResolvedValue({ files: carried } as never); + + await patch('/book.epub', new Map()); + + expect(openedAfter).toEqual([['mimetype'], ['mimetype', 'OPS/ch1.xhtml'], ['mimetype', 'OPS/ch1.xhtml', 'OPS/ch2.xhtml']]); + expect(processedEntries).toEqual(['mimetype', 'OPS/ch1.xhtml', 'OPS/ch2.xhtml', 'OPS/ch3.xhtml']); + for (const entry of carried) expect(entry.stream).toHaveBeenCalledTimes(1); + }); + it('rejects when a source entry stream errors while patching', async () => { const sourceError = Object.assign(new Error('ENOENT: no such file or directory, open /book.epub'), { code: 'ENOENT' }); const missingEntry = { diff --git a/server/src/modules/file-write/formats/epub/epub-zip-patcher.ts b/server/src/modules/file-write/formats/epub/epub-zip-patcher.ts index 410d0ccc2..97c747de5 100644 --- a/server/src/modules/file-write/formats/epub/epub-zip-patcher.ts +++ b/server/src/modules/file-write/formats/epub/epub-zip-patcher.ts @@ -1,8 +1,8 @@ -import { createWriteStream } from 'fs'; -import type { Readable } from 'stream'; import * as unzipper from 'unzipper'; -import { ZipArchive, type Archiver } from 'archiver'; import { replaceFileAtomically } from '../shared/atomic-file-replace'; +import { writeZipArchive, type ZipRewriteEntry } from '../shared/zip-rewrite'; + +const MIMETYPE_ENTRY_PATH = 'mimetype'; export async function readEntry(filePath: string, entryPath: string): Promise { const zip = await unzipper.Open.file(filePath); @@ -19,41 +19,24 @@ export async function listEntryPaths(filePath: string): Promise { export async function patch(filePath: string, patches: Map): Promise { const tmpPath = filePath + '.tmp'; const zip = await unzipper.Open.file(filePath); - const archive = new ZipArchive({ zlib: { level: 6 } }); - const output = createWriteStream(tmpPath); - await new Promise((resolve, reject) => { - output.on('close', resolve); - output.on('error', reject); - archive.on('error', reject); - archive.pipe(output); - - archive.append(Buffer.from('application/epub+zip'), { name: 'mimetype', store: true }); - - for (const entry of zip.files) { - if (entry.path === 'mimetype') continue; - const patched = patches.get(entry.path); - if (patched) { - archive.append(patched, { name: entry.path }); - } else { - appendEntryStream(archive, entry, reject); - } - } - - for (const [path, content] of patches) { - if (!zip.files.some((e) => e.path === path)) { - archive.append(content, { name: path }); - } - } - - void archive.finalize(); - }); + await writeZipArchive(tmpPath, rewriteEntries(zip.files, patches)); await replaceFileAtomically(tmpPath, filePath); } -function appendEntryStream(archive: Archiver, entry: { path: string; stream: () => Readable }, reject: (error: Error) => void): void { - const source = entry.stream(); - source.once('error', reject); - archive.append(source, { name: entry.path }); +function* rewriteEntries(files: readonly unzipper.File[], patches: Map): Generator { + yield { name: MIMETYPE_ENTRY_PATH, source: Buffer.from('application/epub+zip'), store: true }; + + for (const entry of files) { + if (entry.path === MIMETYPE_ENTRY_PATH) continue; + const patched = patches.get(entry.path); + yield { name: entry.path, source: patched ?? (() => entry.stream()) }; + } + + for (const [path, content] of patches) { + if (!files.some((entry) => entry.path === path)) { + yield { name: path, source: content }; + } + } } diff --git a/server/src/modules/file-write/formats/shared/zip-rewrite.test.ts b/server/src/modules/file-write/formats/shared/zip-rewrite.test.ts new file mode 100644 index 000000000..e994fc181 --- /dev/null +++ b/server/src/modules/file-write/formats/shared/zip-rewrite.test.ts @@ -0,0 +1,219 @@ +import { mkdtemp, rm } from 'fs/promises'; +import { tmpdir } from 'os'; +import { join } from 'path'; +import { PassThrough, Readable } from 'stream'; +import * as unzipper from 'unzipper'; + +import { writeZipArchive, type ZipRewriteEntry } from './zip-rewrite'; + +const STORED_COMPRESSION_METHOD = 0; + +let testRoot: string; + +beforeEach(async () => { + testRoot = await mkdtemp(join(tmpdir(), 'bookorbit-zip-rewrite-')); +}); + +afterEach(async () => { + await rm(testRoot, { recursive: true, force: true }); +}); + +function chunkedStream(value: string, chunks = 4): Readable { + const size = Math.max(1, Math.ceil(value.length / chunks)); + const parts: Buffer[] = []; + for (let offset = 0; offset < value.length; offset += size) parts.push(Buffer.from(value.slice(offset, offset + size))); + return Readable.from(parts); +} + +/** Tracks how many entry sources are open at once, which is what the eager-append regression cost. */ +function trackedSources(count: number): { entries: ZipRewriteEntry[]; openOrder: string[]; maxConcurrentOpen: () => number } { + const openOrder: string[] = []; + let open = 0; + let maxOpen = 0; + + const entries = Array.from({ length: count }, (_unused, index): ZipRewriteEntry => { + const name = `pages/${String(index).padStart(4, '0')}.txt`; + return { + name, + source: () => { + openOrder.push(name); + open += 1; + maxOpen = Math.max(maxOpen, open); + const stream = chunkedStream(`contents of ${name}`); + stream.once('end', () => { + open -= 1; + }); + return stream; + }, + }; + }); + + return { entries, openOrder, maxConcurrentOpen: () => maxOpen }; +} + +async function readArchive(filePath: string): Promise> { + const zip = await unzipper.Open.file(filePath); + return Promise.all( + zip.files.map(async (entry) => ({ + path: entry.path, + compressionMethod: entry.compressionMethod, + content: (await entry.buffer()).toString('utf-8'), + })), + ); +} + +describe('writeZipArchive', () => { + it('writes buffer and stream sources in the order they are yielded', async () => { + const target = join(testRoot, 'out.zip'); + + await writeZipArchive(target, [ + { name: 'mimetype', source: Buffer.from('application/epub+zip'), store: true }, + { name: 'META-INF/container.xml', source: () => chunkedStream('') }, + { name: 'OPS/content.opf', source: Buffer.from('') }, + ]); + + const written = await readArchive(target); + expect(written.map((entry) => entry.path)).toEqual(['mimetype', 'META-INF/container.xml', 'OPS/content.opf']); + expect(written.map((entry) => entry.content)).toEqual(['application/epub+zip', '', '']); + }); + + it('stores entries flagged with store and compresses the rest', async () => { + const target = join(testRoot, 'out.zip'); + const compressible = 'a'.repeat(4096); + + await writeZipArchive(target, [ + { name: 'mimetype', source: Buffer.from('application/epub+zip'), store: true }, + { name: 'OPS/ch1.xhtml', source: Buffer.from(compressible) }, + ]); + + const written = await readArchive(target); + expect(written[0]!.compressionMethod).toBe(STORED_COMPRESSION_METHOD); + expect(written[1]!.compressionMethod).not.toBe(STORED_COMPRESSION_METHOD); + expect(written[1]!.content).toBe(compressible); + }); + + it('opens at most one entry source at a time', async () => { + const target = join(testRoot, 'out.zip'); + const { entries, openOrder, maxConcurrentOpen } = trackedSources(150); + + await writeZipArchive(target, entries); + + expect(maxConcurrentOpen()).toBe(1); + expect(openOrder).toEqual(entries.map((entry) => entry.name)); + const written = await readArchive(target); + expect(written).toHaveLength(150); + expect(written[149]!.content).toBe('contents of pages/0149.txt'); + }); + + it('does not open a source until the preceding sources have been consumed', async () => { + const target = join(testRoot, 'out.zip'); + const drained: string[] = []; + const seenWhenOpened = new Map(); + const lazySource = (name: string) => () => { + seenWhenOpened.set(name, [...drained]); + const stream = chunkedStream(`contents of ${name}`); + stream.once('end', () => { + drained.push(name); + }); + return stream; + }; + + await writeZipArchive(target, [ + { name: 'first', source: lazySource('first') }, + { name: 'second', source: lazySource('second') }, + { name: 'third', source: lazySource('third') }, + ]); + + expect(seenWhenOpened.get('first')).toEqual([]); + expect(seenWhenOpened.get('second')).toEqual(['first']); + expect(seenWhenOpened.get('third')).toEqual(['first', 'second']); + }); + + it('accepts an iterable that is consumed lazily', async () => { + const target = join(testRoot, 'out.zip'); + let yielded = 0; + function* entries(): Generator { + for (const name of ['a.txt', 'b.txt', 'c.txt']) { + yielded += 1; + yield { name, source: Buffer.from(name) }; + } + } + + await writeZipArchive(target, entries()); + + expect(yielded).toBe(3); + expect((await readArchive(target)).map((entry) => entry.path)).toEqual(['a.txt', 'b.txt', 'c.txt']); + }); + + it('writes zero-length entries without stalling the append loop', async () => { + const target = join(testRoot, 'out.zip'); + + await writeZipArchive(target, [ + { name: 'empty.txt', source: Buffer.alloc(0) }, + { name: 'after-empty.txt', source: Buffer.from('still written') }, + ]); + + const written = await readArchive(target); + expect(written.map((entry) => entry.path)).toEqual(['empty.txt', 'after-empty.txt']); + expect(written.map((entry) => entry.content)).toEqual(['', 'still written']); + }); + + it('writes an empty archive when there are no entries', async () => { + const target = join(testRoot, 'empty.zip'); + + await writeZipArchive(target, []); + + await expect(readArchive(target)).resolves.toEqual([]); + }); + + it('rejects with the source error and stops opening further sources', async () => { + const target = join(testRoot, 'out.zip'); + const sourceError = Object.assign(new Error('ENOENT: no such file or directory, open /book.cbz'), { code: 'ENOENT' }); + const laterSource = vi.fn(() => chunkedStream('never reached')); + + const failing = writeZipArchive(target, [ + { + name: 'pages/001.jpg', + source: () => { + const source = new PassThrough(); + process.nextTick(() => source.emit('error', sourceError)); + return source; + }, + }, + { name: 'pages/002.jpg', source: laterSource }, + ]); + + await expect(failing).rejects.toThrow('ENOENT'); + expect(laterSource).not.toHaveBeenCalled(); + }); + + it('rejects when the archive rejects an entry instead of hanging', async () => { + const target = join(testRoot, 'out.zip'); + + await expect(writeZipArchive(target, [{ name: '', source: Buffer.from('x') }])).rejects.toThrow(/ENTRYNAMEREQUIRED|entry name/i); + }); + + it('rejects when the output file cannot be opened', async () => { + const target = join(testRoot, 'missing-directory', 'out.zip'); + + await expect(writeZipArchive(target, [{ name: 'a.txt', source: Buffer.from('a') }])).rejects.toThrow(/ENOENT/); + }); + + it('releases the output stream when a source fails', async () => { + const target = join(testRoot, 'out.zip'); + + const failing = writeZipArchive(target, [ + { + name: 'a.txt', + source: () => { + const source = new PassThrough(); + process.nextTick(() => source.emit('error', new Error('read failed'))); + return source; + }, + }, + ]); + + await expect(failing).rejects.toThrow('read failed'); + await expect(rm(target, { force: true })).resolves.toBeUndefined(); + }); +}); diff --git a/server/src/modules/file-write/formats/shared/zip-rewrite.ts b/server/src/modules/file-write/formats/shared/zip-rewrite.ts new file mode 100644 index 000000000..6f67f81f9 --- /dev/null +++ b/server/src/modules/file-write/formats/shared/zip-rewrite.ts @@ -0,0 +1,55 @@ +import { createWriteStream } from 'fs'; +import type { Readable } from 'stream'; +import { ZipArchive, type Archiver } from 'archiver'; + +const ZLIB_COMPRESSION_LEVEL = 6; + +export interface ZipRewriteEntry { + name: string; + source: Buffer | (() => Readable); + store?: boolean; +} + +/** + * Archiver pipes an appended stream into its queue immediately, so appending every entry of a large + * archive up front holds one descriptor and one decompression buffer per entry until the rewrite + * finishes. Entries are appended one at a time and their sources opened only once archiver is ready + * for them, which keeps descriptors and memory flat regardless of entry count. + */ +export async function writeZipArchive(tmpPath: string, entries: Iterable): Promise { + const archive = new ZipArchive({ zlib: { level: ZLIB_COMPRESSION_LEVEL } }); + const output = createWriteStream(tmpPath); + + let fail!: (error: Error) => void; + const failure = new Promise((_resolve, reject) => { + fail = reject; + }); + // The losing side of every race below stays rejected; keep it handled so it can never surface as + // an unhandled rejection. + failure.catch(() => {}); + + archive.on('error', fail); + output.on('error', fail); + const closed = new Promise((resolve) => output.once('close', resolve)); + archive.pipe(output); + + try { + for (const entry of entries) { + await Promise.race([appendEntry(archive, entry, fail), failure]); + } + void archive.finalize(); + await Promise.race([closed, failure]); + } catch (error) { + output.destroy(); + throw error; + } +} + +function appendEntry(archive: Archiver, entry: ZipRewriteEntry, fail: (error: Error) => void): Promise { + return new Promise((resolve) => { + const source = typeof entry.source === 'function' ? entry.source() : entry.source; + if (!Buffer.isBuffer(source)) source.once('error', fail); + archive.once('entry', () => resolve()); + archive.append(source, entry.store ? { name: entry.name, store: true } : { name: entry.name }); + }); +}