ovflowd · Nov 3, 2026
@@ -0,0 +1,6 @@
+---
+'@doc-kit/core': minor
+'@doc-kit/generator-react': minor
+---
+
+`metadata` and `section-pages` stream their input: with more than one thread, they start on their dependency's first chunk, so parsing, metadata and section pages overlap. The generated files are unchanged. `groupWholeModules`, from `@doc-kit/core/utils/generators.mjs`, regroups streamed entries so that each batch holds whole modules.
@@ -379,7 +379,10 @@ export async function* generate(input, worker) {
 - With more than one thread, `input` can be iterated only once, and passing it
   to `worker.stream` iterates it. If you need it twice, collect it into an
   array first.
-- The chunks are the dependency's own, in the order it yields them.
+- The chunks are the dependency's own, in the order it yields them. `metadata`'s
+  hold whole files, but chunks need not hold whole modules: a generator that
+  needs all of a module's entries at once should regroup them with
+  `groupWholeModules` from `@doc-kit/core/utils/generators.mjs`.
 - Work that needs the whole input, such as sorting or writing one file, has to
   wait for the last chunk.
 - Other generators reading the same dependency are not affected: each gets

@@ -16,6 +16,8 @@ export default {
 
   hasParallelProcessor: true,
 
+  streamsInput: true,
+
   markdown: {
     remarkPlugins: [
       'remark-parse',

@@ -3,7 +3,10 @@ import type { Position } from 'unist';
 
 export type Generator = GeneratorMetadata<
   {},
-  Generate<Array<Root>, AsyncGenerator<MetadataEntry>>,
+  Generate<
+    Array<Root> | AsyncIterable<Array<Root>>,
+    AsyncGenerator<MetadataEntry>
+  >,
   ProcessChunk<Root, MetadataEntry, Record<string, string>>
 >;
 

@@ -4,6 +4,7 @@ import { describe, it } from 'node:test';
 import {
   getEntryDescription,
   groupNodesByModule,
+  groupWholeModules,
   getVersionFromSemVer,
   coerceSemVer,
   getCompatibleVersions,
@@ -87,6 +88,97 @@ describe('groupNodesByModule', () => {
   });
 });
 
+describe('groupWholeModules', () => {
+  /**
+   * Collects the batches regrouped from the given input, each entry as its
+   * module and number.
+   *
+   * @param {Array<{ api: string }> | AsyncIterable<Array<{ api: string }>>} input
+   * @returns {Promise<string[][]>}
+   */
+  const collect = async input => {
+    const output = [];
+
+    for await (const batch of groupWholeModules(input)) {
+      output.push(batch.map(({ api, n }) => `${api}${n}`));
+    }
+
+    return output;
+  };
+
+  /**
+   * Regroups the given batches, streamed in, and collects the batches it
+   * yields.
+   *
+   * @param {Array<Array<{ api: string }>>} batches
+   * @returns {Promise<string[][]>}
+   */
+  const regroup = batches => {
+    async function* input() {
+      yield* batches;
+    }
+
+    return collect(input());
+  };
+
+  /**
+   * An entry of a module, told apart from the module's others by its number.
+   *
+   * @param {string} api
+   * @param {number} n
+   * @returns {{ api: string, n: number }}
+   */
+  const entry = (api, n) => ({ api, n });
+
+  it('passes the entries on in the same order, each batch ending once its last module is whole', async () => {
+    const output = await regroup([
+      [entry('a', 1), entry('a', 2), entry('b', 1)],
+      [entry('c', 1)],
+    ]);
+
+    assert.deepEqual(output, [['a1', 'a2'], ['b1'], ['c1']]);
+  });
+
+  it('holds back a module that continues in the next batch', async () => {
+    const output = await regroup([
+      [entry('a', 1), entry('b', 1)],
+      [entry('b', 2), entry('b', 3)],
+      [entry('b', 4), entry('c', 1)],
+    ]);
+
+    assert.deepEqual(output, [['a1'], ['b1', 'b2', 'b3', 'b4'], ['c1']]);
+  });
+
+  it('skips empty batches', async () => {
+    const output = await regroup([[], [entry('a', 1)], []]);
+
+    assert.deepEqual(output, [['a1']]);
+  });
+
+  it('takes entries given as one array as a single batch', async () => {
+    const output = await collect([entry('a', 1), entry('a', 2), entry('b', 1)]);
+
+    assert.deepEqual(output, [['a1', 'a2'], ['b1']]);
+  });
+
+  it('rejects a module whose entries arrive apart', async () => {
+    await assert.rejects(
+      regroup([
+        [entry('a', 1), entry('b', 1)],
+        [entry('c', 1), entry('a', 2)],
+      ]),
+      /"a" arrived apart/
+    );
+  });
+
+  it('rejects a module whose entries arrive apart in one batch too', async () => {
+    await assert.rejects(
+      regroup([[entry('a', 1), entry('b', 1), entry('a', 2)]]),
+      /"a" arrived apart/
+    );
+  });
+});
+
 describe('getVersionFromSemVer', () => {
   it('returns major.x for minor 0', () => {
     const version = { major: 18, minor: 0, patch: 0 };

@@ -59,6 +59,67 @@ export const toBatches = input => {
   return [input];
 };
 
+/**
+ * Regroups streamed entries so that each batch holds whole modules: a module
+ * whose entries continue in the next batch is held back and passed on with
+ * it. A module's entries must arrive together: one that comes back after
+ * another module's is an error, wherever the batches end, as its pages would
+ * be built without it. Entries given as one array, as a generator that streams
+ * its input gets them with one thread, are taken as a single batch.
+ *
+ * @template {{ api: string }} T
+ * @param {Array<T> | AsyncIterable<Array<T>>} batches - The entries, in
+ * batches as they are streamed, or in one array
+ * @returns {AsyncGenerator<Array<T>>} The same entries, in batches of whole
+ * modules
+ */
+export async function* groupWholeModules(batches) {
+  // The modules whose entries have all arrived
+  const finished = new Set();
+
+  // The module whose entries are arriving, and those of them held back
+  let current;
+  let held = [];
+
+  for await (const batch of toBatches(batches)) {
+    for (const { api } of batch) {
+      if (api !== current) {
+        if (finished.has(api)) {
+          throw new Error(
+            `The entries of "${api}" arrived apart: a module's entries must ` +
+              'arrive together, in one chunk or in consecutive ones'
+          );
+        }
+
+        if (current !== undefined) {
+          finished.add(current);
+        }
+
+        current = api;
+      }
+    }
+
+    const entries = held.length > 0 ? [...held, ...batch] : batch;
+
+    // The module the batch ends with may continue in the next one
+    let end = entries.length;
+
+    while (end > 0 && entries[end - 1].api === current) {
+      end--;
+    }
+
+    held = entries.slice(end);
+
+    if (end > 0) {
+      yield entries.slice(0, end);
+    }
+  }
+
+  if (held.length > 0) {
+    yield held;
+  }
+}
+
 /**
  * Parses the SemVer string into a Node.js-alike version
  *

@@ -5,18 +5,37 @@ import { join } from 'node:path';
 import { describe, it } from 'node:test';
 import { pathToFileURL } from 'node:url';
 
+import { collectAsyncGenerator } from '@doc-kit/core/caching.mjs';
 import { loadGenerator } from '@doc-kit/core/generators/loader.mjs';
 import { setConfig } from '@doc-kit/core/utils/configuration/index.mjs';
 import { loadMarkdownPlugins } from '@doc-kit/core/utils/markdown/plugins.mjs';
 import { jsx, toJs } from 'estree-util-to-js';
 
 import buildContent from '../../jsx-ast/utils/buildContent.mjs';
 import { buildNotFoundPage } from '../../jsx-ast/utils/synthetic/404.mjs';
-import { generate as chunk } from '../../section-pages/generate.mjs';
+import { generate as sectionPages } from '../../section-pages/generate.mjs';
 import { compile, createViteBundler } from '../bundlers/vite.mjs';
 import { FONTS } from '../constants.mjs';
 import { generate } from '../generate.mjs';
 
+/**
+ * Yields the given entries as one batch, as a dependency streaming them does.
+ *
+ * @param {Array<object>} entries
+ */
+async function* streamInOneBatch(entries) {
+  yield entries;
+}
+
+/**
+ * Runs `section-pages` over the given entries, streamed in as one batch, and
+ * collects its pages.
+ *
+ * @param {Array<object>} entries
+ */
+const chunk = entries =>
+  collectAsyncGenerator(sectionPages(streamInOneBatch(entries)));
+
 // Pages are rendered with the pipeline of `jsx-ast`
 await loadMarkdownPlugins(
   await loadGenerator(import.meta.resolve('../../jsx-ast/index.mjs'))