Let a caller take entries as they parse, instead of holding the catalogue

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",

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<void>;
 }
 
 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<Uint8Array | string>,
-  { max = MAX_CHANNELS, onChunk }: StreamOptions = {}
+  { max = MAX_CHANNELS, onEntries, onChunk }: StreamOptions = {}
 ): Promise<StreamResult> {
   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 };
 }

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);
+    }
+  );
 });