Perf: drop stream transform for certain cases

Build/build-chn-cidr.ts

@@ -1,12 +1,12 @@
-import { fetchRemoteTextByLine } from './lib/fetch-text-by-line';
+import { fetchRemoteTextLines } from './lib/fetch-text-by-line';
 import { SpanCategory, task } from './trace';
 
 import { IPListOutput } from './lib/rules/ip';
 import { createFileDescription } from './constants/description';
 
 const getChnCidrPromise = Promise.all([
-  fetchRemoteTextByLine('https://chnroutes2.cdn.skk.moe/chnroutes.txt', true).then(Array.fromAsync<string>),
+  fetchRemoteTextLines('https://chnroutes2.cdn.skk.moe/chnroutes.txt', true),
-  fetchRemoteTextByLine('https://gaoyifan.github.io/china-operator-ip/china6.txt', true).then(Array.fromAsync<string>)
+  fetchRemoteTextLines('https://gaoyifan.github.io/china-operator-ip/china6.txt', true)
 ]);
 
 export const buildChnCidr = task(require.main === module, __filename)(async (span) => {

Build/build-domestic-direct-lan-ruleset-dns-mapping-module.ts

@@ -3,7 +3,7 @@ import path from 'node:path';
 import { DOMESTICS, DOH_BOOTSTRAP, AdGuardHomeDNSMapping } from '../Source/non_ip/domestic';
 import { DIRECTS, HOSTS, LAN } from '../Source/non_ip/direct';
 import type { DNSMapping } from '../Source/non_ip/direct';
-import { fetchRemoteTextByLine, readFileIntoProcessedArray } from './lib/fetch-text-by-line';
+import { fetchRemoteTextLines, readFileIntoProcessedArray } from './lib/fetch-text-by-line';
 import { compareAndWriteFile } from './lib/create-file';
 import { SpanCategory, task } from './trace';
 import type { Span } from './trace';
@@ -386,9 +386,9 @@ async function buildLANCacheRuleset(span: Span) {
   const allDomains = (
     await Promise.all(
       allDomainFiles.map(
-        async (file) => childSpan.traceChildAsync(
+        (file) => childSpan.traceChildAsync(
           'download ' + file,
-          async () => Array.fromAsync(await fetchRemoteTextByLine('https://cdn.jsdelivr.net/gh/uklans/cache-domains@master/' + file, true)),
+          () => fetchRemoteTextLines('https://cdn.jsdelivr.net/gh/uklans/cache-domains@master/' + file, true),
           SpanCategory.Network
         )
       )

Build/lib/fetch-assets.ts

@@ -2,9 +2,8 @@ import picocolors from 'picocolors';
 import { $$fetch, defaultRequestInit, ResponseError } from './fetch-retry';
 import { waitWithAbort } from 'foxts/wait';
 import { nullthrow } from 'foxts/guard';
-import { TextLineStream } from 'foxts/text-line-stream';
+import { ignoreAdGuardUnsupportedLine } from './parse-filter/filters';
-import { ProcessLineStream } from './process-line';
+import { splitTextIntoLines } from './fetch-text-by-line';
-import { AdGuardFilterIgnoreUnsupportedLinesStream } from './parse-filter/filters';
 import { appendArrayInPlace } from 'foxts/append-array-in-place';
 
 import { newQueue } from '@henrygd/queue';
@@ -13,6 +12,7 @@ import { downloadTimestamp, recordExternalDownloadAttempt } from './download-sta
 import type { ExternalDownloadOutcome } from './download-stats';
 
 const reusedCustomAbortError = new AbortError();
+const textDecoder = new TextDecoder();
 
 const queue = newQueue(18);
 
@@ -183,34 +183,24 @@ export async function fetchAssets(
       });
       headersAt ??= downloadTimestamp();
 
-      let body = nullthrow(res.body, url + ' has an empty body');
+      nullthrow(res.body, url + ' has an empty body');
       if (isPrimary) {
         primaryProgress.headersReceived = true;
       }
-      body = body.pipeThrough(new TransformStream<Uint8Array, Uint8Array>({
-        transform(chunk, streamController) {
-          decodedBytes += chunk.byteLength;
-          streamController.enqueue(chunk);
-        }
-      }));
 
-      let stream = body
+      const arr = await queue.add(async () => {
-        .pipeThrough(new TextDecoderStream())
-        .pipeThrough(new TextLineStream({ skipEmptyLines: processLine }));
-      if (processLine) {
-        stream = stream.pipeThrough(new ProcessLineStream());
-      }
-      if (filterAdGuardUnsupportedLines) {
-        stream = stream.pipeThrough(new AdGuardFilterIgnoreUnsupportedLinesStream());
-      }
-
-      const arr = await queue.add(() => {
         decodedBodyStartedAt = downloadTimestamp();
         if (isPrimary) {
           primaryProgress.bodyConsumptionStartedAt = performance.now();
           primaryProgress.encodedBytesAtConsumptionStart = primaryProgress.encodedBytesReceived;
         }
-        return Array.fromAsync(stream);
+        const buf = await res.arrayBuffer();
+        decodedBytes = buf.byteLength;
+        return splitTextIntoLines(
+          textDecoder.decode(buf),
+          processLine,
+          filterAdGuardUnsupportedLines ? ignoreAdGuardUnsupportedLine : null
+        );
       });
 
       if (!allowEmpty && arr.length < 1) {

Build/lib/fetch-text-by-line.test.ts

@@ -0,0 +1,57 @@
+import { describe, it } from 'mocha';
+import { expect } from 'earl';
+import { TextLineStream } from 'foxts/text-line-stream';
+
+import { splitTextIntoLines } from './fetch-text-by-line';
+import { ProcessLineStream } from './process-line';
+
+/** The streaming pipeline splitTextIntoLines replaces, for equivalence checks */
+async function viaStream(text: string, processLine: boolean, chunkSize: number): Promise<string[]> {
+  let stream = new ReadableStream<string>({
+    start(controller) {
+      for (let i = 0; i < text.length; i += chunkSize) {
+        controller.enqueue(text.slice(i, i + chunkSize));
+      }
+      controller.close();
+    }
+  }).pipeThrough(new TextLineStream({ skipEmptyLines: processLine }));
+  if (processLine) {
+    stream = stream.pipeThrough(new ProcessLineStream());
+  }
+  return Array.fromAsync(stream);
+}
+
+const SAMPLES = [
+  '',
+  'single line without newline',
+  'a\nb\nc\n',
+  'a\nb\nc',
+  'a\r\nb\r\n\r\nc\r\n',
+  '\n\n\n',
+  '\r\n',
+  '# comment\n  padded  \n\n! adguard comment\n0.0.0.0 example.com\n\n'
+];
+
+const CHUNK_SIZES = [1, 3, 1024];
+
+describe('splitTextIntoLines', () => {
+  [false, true].forEach((processLine) => {
+    it(`matches the TextLineStream pipeline (processLine=${processLine})`, async () => {
+      for (let i = 0, len = SAMPLES.length; i < len; i++) {
+        for (let j = 0, jlen = CHUNK_SIZES.length; j < jlen; j++) {
+          // eslint-disable-next-line no-await-in-loop -- sequential equivalence checks
+          expect(splitTextIntoLines(SAMPLES[i], processLine)).toEqual(await viaStream(SAMPLES[i], processLine, CHUNK_SIZES[j]));
+        }
+      }
+    });
+  });
+
+  it('drops a lone trailing CR on an unterminated last line (TextLineStream keeps it -- a flush quirk, not a feature)', () => {
+    expect(splitTextIntoLines('a\r\nb\r', false)).toEqual(['a', 'b']);
+  });
+
+  it('applies the filter before processLine and lets it rewrite lines', () => {
+    const lines = splitTextIntoLines('keep\n  drop me \nrewrite\n', true, line => (line.includes('drop') ? null : line.toUpperCase()));
+    expect(lines).toEqual(['KEEP', 'REWRITE']);
+  });
+});

Build/lib/fetch-text-by-line.ts

@@ -4,7 +4,7 @@ import readline from 'node:readline';
 import { TextLineStream } from 'foxts/text-line-stream';
 import type { ReadableStream } from 'node:stream/web';
 import { TextDecoderStream } from 'node:stream/web';
-import { processLine, ProcessLineStream } from './process-line';
+import { processLine as processLineFn, ProcessLineStream } from './process-line';
 import { $$fetch } from './fetch-retry';
 import type { UndiciResponseData } from './fetch-retry';
 import type { Response as UnidiciWebResponse } from 'undici';
@@ -44,11 +44,59 @@ export function fetchRemoteTextByLine(url: string, processLine = false): Promise
   return $$fetch(url).then(resp => createReadlineInterfaceFromResponse(resp, processLine));
 }
 
+export function splitTextIntoLines(
+  text: string,
+  processLine = false,
+  filter: ((line: string) => string | null) | null = null
+): string[] {
+  const lines: string[] = [];
+  const len = text.length;
+  let start = 0;
+
+  while (start <= len) {
+    let end = text.indexOf('\n', start);
+    const next = end === -1 ? len + 1 : end + 1;
+    if (end === -1) {
+      end = len;
+    }
+    if (end > start && text.charCodeAt(end - 1) === 13 /* \r */) {
+      end--;
+    }
+
+    if (end > start || (!processLine && start < len)) {
+      let line: string | null = text.slice(start, end);
+      if (filter) {
+        line = filter(line);
+      }
+      if (line !== null && processLine) {
+        line = processLineFn(line);
+      }
+      if (line !== null) {
+        lines.push(line);
+      }
+    }
+
+    start = next;
+  }
+
+  return lines;
+}
+
+/**
+ * Download a text asset and return its lines. The body is buffered and split
+ * synchronously -- see {@link splitTextIntoLines} for why that beats streaming
+ * when the result is an array anyway.
+ */
+export async function fetchRemoteTextLines(url: string, processLine = false): Promise<string[]> {
+  const resp = await $$fetch(url);
+  return splitTextIntoLines(await resp.text(), processLine);
+}
+
 export async function readFileIntoProcessedArray(file: string /* | FileHandle */) {
   const results = [];
   let processed: string | null = '';
   for await (const line of readFileByLine(file)) {
-    processed = processLine(line);
+    processed = processLineFn(line);
     if (processed) {
       results.push(processed);
     }

Build/lib/parse-filter/filters.ts

@@ -200,62 +200,56 @@ const kwfilter = createKeywordFilter([
 ]);
 
 /**
- * The idea is that, TransformStream works kinda like a filter running on response. If we
+ * Drop filter lines that can never become a Surge/Clash rule (cosmetic, path
- * can filter lines before Array.fromAsync, we can create a smaller array, this saves memory
+ * rules, rules with browser-only modifiers) as early as possible, so the parser
- * and could improve performance.
+ * and the line array only ever see candidates. Returns the trimmed line, or null
+ * to drop it. Runs inline while the downloaded text is split into lines.
  */
-export class AdGuardFilterIgnoreUnsupportedLinesStream extends TransformStream<string, string> {
+export function ignoreAdGuardUnsupportedLine(line: string): string | null {
-  // private __buf = '';
+  let firstCharCode = line.charCodeAt(0);
-  constructor() {
+  if (
-    super({
+    // bail out path-like/cosmetic very early, even before trim
-      transform(line, controller) {
+    firstCharCode === 47 // /
-        let firstCharCode = line.charCodeAt(0);
+    || firstCharCode === 35 // #
-        if (
+    // doesn't include
-          // bail out path-like/cosmetic very early, even before trim
+    || !line.includes('.') // rule with out dot can not be a domain
-          firstCharCode === 47 // /
+    || kwfilter(line) // filter out some symbols/modifiers
-          || firstCharCode === 35 // #
+  ) {
-          // doesn't include
+    return null;
-          || !line.includes('.') // rule with out dot can not be a domain
-          || kwfilter(line) // filter out some symbols/modifiers
-        ) {
-          return;
-        }
-
-        line = line.trim();
-
-        if (line.length === 0) {
-          return;
-        }
-
-        firstCharCode = line.charCodeAt(0);
-        const lastCharCode = line.charCodeAt(line.length - 1);
-
-        if (
-          firstCharCode === 47 // 47 `/`
-          // ends with
-          // _160-600.
-          // -detect-adblock.
-          // _web-advert.
-          || lastCharCode === 46 // 46 `.`, line.endsWith('.')
-          || lastCharCode === 45 // 45 `-`, line.endsWith('-')
-          || lastCharCode === 95 // 95 `_`, line.endsWith('_')
-        ) {
-          return;
-        }
-
-        if ((line.includes('/') || line.includes(':')) && !line.includes('://')) {
-          // ignore any line that has "/" or ":" but not "://"
-          return;
-        }
-
-        controller.enqueue(line);
-      }
-    });
   }
+
+  line = line.trim();
+
+  if (line.length === 0) {
+    return null;
+  }
+
+  firstCharCode = line.charCodeAt(0);
+  const lastCharCode = line.charCodeAt(line.length - 1);
+
+  if (
+    firstCharCode === 47 // 47 `/`
+    // ends with
+    // _160-600.
+    // -detect-adblock.
+    // _web-advert.
+    || lastCharCode === 46 // 46 `.`, line.endsWith('.')
+    || lastCharCode === 45 // 45 `-`, line.endsWith('-')
+    || lastCharCode === 95 // 95 `_`, line.endsWith('_')
+  ) {
+    return null;
+  }
+
+  if ((line.includes('/') || line.includes(':')) && !line.includes('://')) {
+    // ignore any line that has "/" or ":" but not "://"
+    return null;
+  }
+
+  return line;
 }
 
 export function parse(line: string, result: [string, ParseType], includeThirdParty: boolean): [hostname: string, flag: ParseType] {
-  // We have already done this in AdGuardFilterIgnoreUnsupportedLinesStream
+  // We have already done this in ignoreAdGuardUnsupportedLine
 
   // if (
   //   // doesn't include

Build/tools-lum-apex-domains.ts

@@ -1,4 +1,4 @@
-import { fetchRemoteTextByLine } from './lib/fetch-text-by-line';
+import { fetchRemoteTextLines } from './lib/fetch-text-by-line';
 import tldts from 'tldts-experimental';
 import { HostnameSmolTrie } from 'hntrie/smol';
 import { domainToASCII } from 'node:url';
@@ -7,8 +7,8 @@ import { SOURCE_DIR } from './constants/dir';
 import runAgainstSourceFile from './lib/run-against-source-file';
 
 (async () => {
-  const lines1 = await Array.fromAsync(await fetchRemoteTextByLine('https://raw.githubusercontent.com/durablenapkin/block/master/luminati.txt', true));
+  const lines1 = await fetchRemoteTextLines('https://raw.githubusercontent.com/durablenapkin/block/master/luminati.txt', true);
-  const lines2 = await Array.fromAsync(await fetchRemoteTextByLine('https://raw.githubusercontent.com/durablenapkin/block/master/tvstream.txt', true));
+  const lines2 = await fetchRemoteTextLines('https://raw.githubusercontent.com/durablenapkin/block/master/tvstream.txt', true);
 
   const trie = new HostnameSmolTrie();