fix(core): make generator output ordering deterministic (#1193)

Author: OKAN-SOFTWARE

.changeset/stable-generator-order.md

@@ -0,0 +1,5 @@
+---
+'@doc-kit/core': patch
+---
+
+Keep worker results and discovered source files in deterministic order so repeated documentation builds produce stable output.

packages/core/src/generators/ast-js/__tests__/generate.test.mjs

@@ -0,0 +1,41 @@
+import assert from 'node:assert/strict';
+import { mkdtemp, rm, writeFile } from 'node:fs/promises';
+import { tmpdir } from 'node:os';
+import { join, sep } from 'node:path';
+import { it } from 'node:test';
+
+import { setConfig } from '#utils/configuration/index.mjs';
+
+import { generate } from '../generate.mjs';
+
+it('submits JavaScript files in sorted path order', async () => {
+  const dir = await mkdtemp(join(tmpdir(), 'ast-js-order-'));
+
+  try {
+    for (const name of ['z.js', 'a.js', 'm.js', 'ignore.md']) {
+      await writeFile(join(dir, name), '');
+    }
+
+    const input = join(dir, '*').split(sep).join('/');
+    await setConfig({ target: ['ast-js'], input: [input] });
+
+    let received;
+    const worker = {
+      async *stream(items) {
+        received = items;
+        yield [];
+      },
+    };
+
+    for await (const chunk of generate(undefined, worker)) {
+      assert.deepStrictEqual(chunk, []);
+    }
+
+    assert.deepStrictEqual(
+      received.map(path => path.replaceAll('\\', '/').split('/').at(-1)),
+      ['a.js', 'm.js', 'z.js']
+    );
+  } finally {
+    await rm(dir, { recursive: true, force: true });
+  }
+});

packages/core/src/generators/ast-js/generate.mjs

@@ -44,9 +44,9 @@ export async function processChunk(inputSlice, itemIndices) {
 export async function* generate(_, worker) {
   const config = getConfig('ast-js');
 
-  const files = findFiles(config.input, { ignore: config.ignore }).filter(
-    p => extname(p) === '.js'
-  );
+  const files = findFiles(config.input, { ignore: config.ignore })
+    .filter(p => extname(p) === '.js')
+    .sort();
 
   // Parse the Javascript sources into ASTs in parallel using worker threads
   // source is both the items list and the fullInput since we use sliceInput

packages/core/src/generators/ast/__tests__/generate.test.mjs

@@ -1,15 +1,16 @@
 import assert from 'node:assert/strict';
-import { mkdtemp, writeFile, rm } from 'node:fs/promises';
+import { mkdir, mkdtemp, writeFile, rm } from 'node:fs/promises';
 import { tmpdir } from 'node:os';
 import { join, sep } from 'node:path';
 import { after, before, describe, it } from 'node:test';
 import { pathToFileURL } from 'node:url';
 
 import { loadGenerator } from '#generators/loader.mjs';
+import { setConfig } from '#utils/configuration/index.mjs';
 import { loadMarkdownPlugins } from '#utils/markdown/plugins.mjs';
 
 import { STABILITY_INDEX_URL } from '../constants.mjs';
-import { processChunk } from '../generate.mjs';
+import { generate, processChunk } from '../generate.mjs';
 
 const ast = await loadGenerator(import.meta.resolve('../index.mjs'));
 
@@ -216,3 +217,46 @@ describe('processChunk', () => {
     });
   });
 });
+
+describe('generate', () => {
+  it('sorts files inside each pattern without reordering configured patterns', async () => {
+    const first = join(dir, 'ordered-first');
+    const second = join(dir, 'ordered-second');
+    await mkdir(first);
+    await mkdir(second);
+
+    for (const name of ['z.md', 'a.md', 'm.md']) {
+      await writeFile(join(first, name), '# Title\n');
+      await writeFile(join(second, name), '# Title\n');
+    }
+
+    await setConfig({
+      target: ['ast'],
+      input: [
+        toPosixPath(join(second, '*.md')),
+        toPosixPath(join(first, '*.md')),
+      ],
+    });
+
+    let received;
+    const worker = {
+      async *stream(items) {
+        received = items;
+        yield [];
+      },
+    };
+
+    for await (const chunk of generate(undefined, worker)) {
+      assert.deepStrictEqual(chunk, []);
+    }
+
+    assert.deepStrictEqual(
+      received.map(([child]) =>
+        child.replaceAll('\\', '/').split('/').slice(-2).join('/')
+      ),
+      ['ordered-second', 'ordered-first'].flatMap(parent =>
+        ['a.md', 'm.md', 'z.md'].map(name => `${parent}/${name}`)
+      )
+    );
+  });
+});

packages/core/src/generators/ast/generate.mjs

@@ -116,10 +116,9 @@ export async function* generate(_, worker) {
   const files = config.input.flatMap(input => {
     const parent = globParent(input);
 
-    return findFiles(input, { ignore: config.ignore }).map(child => [
-      child,
-      parent,
-    ]);
+    return findFiles(input, { ignore: config.ignore })
+      .sort()
+      .map(child => [child, parent]);
   });
 
   // Parse markdown files in parallel using worker threads

packages/core/src/threading/__tests__/parallel.test.mjs

@@ -1,4 +1,4 @@
-import { deepStrictEqual, ok, strictEqual } from 'node:assert';
+import { deepStrictEqual, ok, rejects, strictEqual } from 'node:assert';
 import { describe, it } from 'node:test';
 
 import { loadGenerator } from '../../generators/loader.mjs';
@@ -106,7 +106,7 @@ describe('createParallelWorker', () => {
     await pool.destroy();
   });
 
-  it('should yield results as chunks complete', async () => {
+  it('should yield one result per input chunk', async () => {
     const pool = createWorkerPool(2);
     const worker = createParallelWorker('metadata', metadataGenerator, pool, {
       threads: 2,
@@ -177,4 +177,56 @@ describe('createParallelWorker', () => {
 
     await pool.destroy();
   });
+
+  it('yields concurrent worker results in submission order', async () => {
+    const resolvers = [];
+    const pool = {
+      run: task =>
+        new Promise(resolve => {
+          resolvers.push(() => resolve(task.input));
+        }),
+    };
+    const worker = createParallelWorker('ast-js', astJsGenerator, pool, {
+      threads: 2,
+      chunkSize: 1,
+    });
+
+    const result = collectStream(worker.stream([0, 1, 2, 3, 4], {}));
+    await Promise.resolve();
+
+    // All five tasks start before any result is yielded, but finish in reverse.
+    strictEqual(resolvers.length, 5);
+    for (const resolve of resolvers.toReversed()) {
+      resolve();
+    }
+
+    deepStrictEqual(await result, [0, 1, 2, 3, 4]);
+  });
+
+  it('propagates a later worker failure without an unhandled rejection', async () => {
+    const tasks = [];
+    const pool = {
+      run: task =>
+        new Promise((resolve, reject) => {
+          tasks.push({ resolve, reject, input: task.input });
+        }),
+    };
+    const worker = createParallelWorker('ast-js', astJsGenerator, pool, {
+      threads: 2,
+      chunkSize: 1,
+    });
+
+    const result = collectStream(worker.stream([0, 1, 2, 3], {}));
+    await Promise.resolve();
+    strictEqual(tasks.length, 4);
+
+    tasks[2].reject(new Error('chunk failed'));
+    for (const [index, task] of tasks.entries()) {
+      if (index !== 2) {
+        task.resolve(task.input);
+      }
+    }
+
+    await rejects(result, /chunk failed/);
+  });
 });

packages/core/src/threading/parallel.mjs

@@ -77,12 +77,12 @@ export default function createParallelWorker(
 
   return {
     /**
-     * Processes items in parallel, yielding results as chunks complete.
+     * Processes items in parallel, yielding results in submission order.
      *
      * @template T
      * @param {T[]} items - Items to process
      * @param {object} extra - Extra options
-     * @yields {R[]} Chunk results as they complete
+     * @yields {R[]} Chunk results in input order
      */
     async *stream(items, extra) {
       if (items.length === 0) {
@@ -98,42 +98,34 @@ export default function createParallelWorker(
 
       const runInOneGo = threads <= 1 || items.length <= 2;
 
-      // Submit all tasks to Piscina - each promise resolves to itself for removal
-      const pending = new Set(
-        chunks.map(indices => {
-          if (runInOneGo) {
-            const promise = generator
-              .processChunk(items, indices, extra)
-              .then(result => ({ promise, result }));
-
-            return promise;
-          }
-
-          const promise = pool
-            .run(
+      // Start every chunk immediately, but emit results in submission order.
+      // Handle errors as each task settles so a later failure cannot become an
+      // unhandled rejection while earlier chunks are still running.
+      const tasks = chunks.map(indices => {
+        const task = runInOneGo
+          ? generator.processChunk(items, indices, extra)
+          : pool.run(
               createTask(items, indices, extra, configuration, specifier, name)
-            )
-            .then(result => ({ promise, result }));
-
-          return promise;
-        })
-      );
-
-      // Yield results as they complete (true parallel collection)
-      let completed = 0;
+            );
 
-      while (pending.size > 0) {
-        const { promise, result } = await Promise.race(pending);
+        return task.then(
+          result => ({ result }),
+          error => ({ error })
+        );
+      });
 
-        pending.delete(promise);
+      for (const [index, task] of tasks.entries()) {
+        const outcome = await task;
 
-        completed++;
+        if ('error' in outcome) {
+          throw outcome.error;
+        }
 
-        parallelLogger.debug(`Chunk ${completed}/${chunks.length} completed`, {
+        parallelLogger.debug(`Chunk ${index + 1}/${chunks.length} completed`, {
           generator: name,
         });
 
-        yield result;
+        yield outcome.result;
       }
     },
   };