diff --git a/package.json b/package.json index 96752db..505a396 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@profullstack/player", - "version": "0.6.0", + "version": "0.7.0", "description": "One web player for every source a Profullstack site serves: MP4, HLS, MPEG-2 transport streams and audio, with one control bar, on desktop, mobile, PWA and television.", "keywords": [ "video", diff --git a/src/m3u.ts b/src/m3u.ts index 4f01a71..5c2a028 100644 --- a/src/m3u.ts +++ b/src/m3u.ts @@ -45,7 +45,15 @@ export interface M3uEntry { } export interface ParseOptions { - /** Stop after this many entries. */ + /** + * Stop after this many entries. + * + * `Infinity` -- or `0`, which is what an unset environment variable parses to + * -- means no ceiling. A ceiling is a memory bound and nothing else, so it + * belongs to whoever owns the memory: a browser holding the result in an array + * wants one, a server streaming entries into Postgres through {@link + * ParseOptions.onEntry} does not, and neither can be guessed from here. + */ max?: number; } @@ -106,9 +114,22 @@ export interface M3uParser { * because the bytes behind them usually still have to be hashed. */ push(line: string): boolean; - /** Everything kept so far. The same array throughout; not copied per push. */ + /** Everything held right now. The same array throughout; not copied per push. */ readonly entries: M3uEntry[]; - /** True once `max` entries have been kept. */ + /** + * Take everything held and forget it, so the parser can keep going in bounded + * memory. Returns a fresh array; `entries` is empty afterwards. + */ + drain(): M3uEntry[]; + /** + * How many entries have been kept in total, drained or not. + * + * This is what `max` is measured against, and why it is counted rather than + * read off `entries.length` -- a drained parser would otherwise forget it was + * ever full and start again from zero. + */ + readonly kept: number; + /** True once `max` entries have been kept. False when `max` is unlimited. */ readonly full: boolean; } @@ -125,7 +146,22 @@ export interface M3uParser { * at, because a playlist that half-parses is worse than one that does not. */ export function createM3uParser({ max = MAX_CHANNELS }: ParseOptions = {}): M3uParser { - const entries: M3uEntry[] = []; + let entries: M3uEntry[] = []; + /* + * Counted rather than read off `entries.length`, which drops to zero on every + * drain. Both ceiling checks below go through this, so the two cannot drift. + */ + let kept = 0; + /* + * 0 and NaN mean no ceiling, not "keep nothing". + * + * `num('PLAYLIST_MAX_CHANNELS', 0)` is how a site says unlimited, and an unset + * or unparseable variable arrives the same way. Treating that literally would + * turn a missing config line into a playlist with no channels in it -- an + * import that succeeds and stores nothing, which is the worst of the three + * outcomes because nothing reports it. + */ + const ceiling = Number.isFinite(max) && max > 0 ? max : Number.POSITIVE_INFINITY; /** `#EXTGRP:` is the other way providers state a group; it applies until changed. */ let currentGroup: string | null = null; @@ -136,11 +172,19 @@ export function createM3uParser({ max = MAX_CHANNELS }: ParseOptions = {}): M3uP get entries() { return entries; }, + drain() { + const taken = entries; + entries = []; + return taken; + }, + get kept() { + return kept; + }, get full() { - return entries.length >= max; + return kept >= ceiling; }, push(raw: string): boolean { - if (entries.length >= max) return false; + if (kept >= ceiling) return false; const line = raw.trim(); if (line.startsWith('#EXTGRP:')) { @@ -172,6 +216,7 @@ export function createM3uParser({ max = MAX_CHANNELS }: ParseOptions = {}): M3uP const group = attrGroup || currentGroup || null; entries.push({ title: name, group, url, kind: entryKind({ url, group }) }); + kept += 1; return true; } @@ -223,13 +268,35 @@ export interface StreamOptions extends ParseOptions { * belongs -- the policy and its error message are the caller's, not ours. */ onChunk?: (chunk: Uint8Array | string) => void; + /** + * Take the entries parsed so far, as they are parsed, so the parse holds none. + * + * Called after each chunk with what that chunk produced -- often zero entries, + * sometimes a few hundred -- and then the parser forgets them. It is awaited, + * which is the point: the consumer sets the pace, so a caller writing to a + * database in batches bounds this parse to one batch of memory no matter how + * large the list is. + * + * Without it, "no ceiling" is not really no ceiling: the result array becomes + * the ceiling, and on the 583MB catalogue that prompted this -- roughly 2.6 + * million entries -- that array alone is more heap than the container has. + * + * `entries` in the result is empty when this is given. Count with `kept`. + * + * Throwing (or rejecting) aborts the parse and cancels the stream, same as + * {@link StreamOptions.onChunk}. + */ + onEntries?: (entries: M3uEntry[]) => void | Promise; } export interface StreamResult { + /** Empty when `onEntries` took them; use {@link StreamResult.kept} to count. */ entries: M3uEntry[]; + /** How many entries were kept, whether retained or handed over. */ + kept: number; /** Bytes seen. String chunks are counted by length, having no encoding here. */ bytes: number; - /** True if `max` was reached and later entries were dropped. */ + /** True if `max` was reached and later entries were dropped. Never with no ceiling. */ truncated: boolean; } @@ -245,10 +312,16 @@ export interface StreamResult { * because `onChunk` is usually a hash and a hash of most of a file is worth * nothing. Past that point the decoding and splitting stop, so the tail of an * oversized list costs only the read. + * + * Pass `onEntries` and the parse holds nothing across chunks: entries are handed + * over as they are found and `entries` comes back empty. That is the form a + * server ingesting a catalogue into a database wants, and it is what makes + * `max: Infinity` a sensible thing to ask for -- without it, "no ceiling" only + * moves the ceiling to the result array. */ export async function parseM3uStream( chunks: AsyncIterable, - { max = MAX_CHANNELS, onChunk }: StreamOptions = {} + { max = MAX_CHANNELS, onEntries, onChunk }: StreamOptions = {} ): Promise { const parser = createM3uParser({ max }); const decoder = new TextDecoder('utf-8'); @@ -277,6 +350,11 @@ export async function parseM3uStream( break; } } + + // Handed over and forgotten, once per chunk. Awaited here rather than per + // entry because `push` is synchronous and a consumer that writes somewhere + // needs to be able to make the parse wait for it. + if (onEntries) await onEntries(parser.drain()); } if (!truncated) { @@ -286,5 +364,10 @@ export async function parseM3uStream( if (tail) parser.push(tail); } - return { entries: parser.entries, bytes, truncated: truncated || parser.full }; + // The last entry lands on the flushed tail, after the final chunk was drained, + // so a consumer that is not offered this one loses exactly one channel -- the + // ordinary shape of a file with no trailing newline. + if (onEntries) await onEntries(parser.drain()); + + return { entries: parser.entries, kept: parser.kept, bytes, truncated: truncated || parser.full }; } diff --git a/test/m3u.test.ts b/test/m3u.test.ts index b00fad1..c96ee2a 100644 --- a/test/m3u.test.ts +++ b/test/m3u.test.ts @@ -228,6 +228,68 @@ describe('parseM3uStream', () => { expect(got.entries.map((c) => c.title)).toEqual(['A', 'B']); }); + it('holds nothing across chunks when onEntries takes them, and still hashes it all', async () => { + // No trailing newline: the last entry lands on the flushed tail, after the + // final chunk has been drained. Losing it is the obvious bug here. + const many = Array.from({ length: 2000 }, (_, i) => + [`#EXTINF:-1,Ch ${i}`, `https://ok.test/${i}.ts`].join('\n') + ).join('\n'); + const taken: string[] = []; + let biggestHeld = 0; + let seen = 0; + const got = await parseM3uStream(inChunks(many, 64), { + max: Number.POSITIVE_INFINITY, + onEntries: (batch) => { + biggestHeld = Math.max(biggestHeld, batch.length); + for (const e of batch) taken.push(e.title); + }, + onChunk: (c) => { + seen += typeof c === 'string' ? c.length : c.byteLength; + }, + }); + expect(taken).toHaveLength(2000); + expect(taken[0]).toBe('Ch 0'); + expect(taken[1999]).toBe('Ch 1999'); + expect(taken).toEqual(parseM3u(many).map((e) => e.title)); + // The whole point: what the parse held at once is a chunk's worth, not 2000. + expect(biggestHeld).toBeLessThan(20); + expect(got.kept).toBe(2000); + expect(got.entries).toEqual([]); + expect(got.truncated).toBe(false); + expect(seen).toBe(new TextEncoder().encode(many).byteLength); + }); + + it('lets onEntries set the pace, so a slow consumer is not raced', async () => { + const list = Array.from({ length: 500 }, (_, i) => + [`#EXTINF:-1,Ch ${i}`, `https://ok.test/${i}.ts`].join('\n') + ).join('\n'); + const order: string[] = []; + let inFlight = 0; + await parseM3uStream(inChunks(list, 128), { + max: Number.POSITIVE_INFINITY, + onEntries: async (batch) => { + // Overlapping calls would mean the parse ran ahead of the writer, which + // is exactly the unbounded memory this exists to prevent. + expect(inFlight).toBe(0); + inFlight += 1; + await new Promise((r) => setTimeout(r, 0)); + for (const e of batch) order.push(e.title); + inFlight -= 1; + }, + }); + expect(order).toEqual(parseM3u(list).map((e) => e.title)); + }); + + it('aborts the parse when onEntries rejects', async () => { + await expect( + parseM3uStream(inChunks(big, 64), { + onEntries: async () => { + throw new Error('no room'); + }, + }) + ).rejects.toThrow('no room'); + }); + it('lets the caller abort mid-stream by throwing from onChunk', async () => { // Where a size ceiling belongs: the policy and the wording are the caller's. let read = 0; @@ -259,4 +321,39 @@ describe('createM3uParser', () => { expect(p.push('#EXTINF:-1,Two')).toBe(false); expect(p.entries).toHaveLength(1); }); + + it('hands entries over on drain and forgets them', () => { + const p = createM3uParser(); + p.push('#EXTINF:-1,One'); + p.push('https://ok.test/1.ts'); + expect(p.drain().map((e) => e.title)).toEqual(['One']); + expect(p.entries).toEqual([]); + p.push('#EXTINF:-1,Two'); + p.push('https://ok.test/2.ts'); + expect(p.drain().map((e) => e.title)).toEqual(['Two']); + }); + + it('counts kept across drains, so max still bites', () => { + const p = createM3uParser({ max: 2 }); + for (const line of ['#EXTINF:-1,One', 'https://ok.test/1.ts']) p.push(line); + // The array is empty again, but the ceiling must not reset with it. + p.drain(); + expect(p.kept).toBe(1); + expect(p.full).toBe(false); + for (const line of ['#EXTINF:-1,Two', 'https://ok.test/2.ts']) p.push(line); + expect(p.kept).toBe(2); + expect(p.full).toBe(true); + expect(p.push('#EXTINF:-1,Three')).toBe(false); + }); + + it.each([Number.POSITIVE_INFINITY, 0, Number.NaN])( + 'treats max=%p as no ceiling rather than as keep-nothing', + (max) => { + const p = createM3uParser({ max }); + p.push('#EXTINF:-1,One'); + p.push('https://ok.test/1.ts'); + expect(p.kept).toBe(1); + expect(p.full).toBe(false); + } + ); });