chore: reduce memory footprint

src/__tests__/caching.test.mjs

@@ -0,0 +1,110 @@
+import { deepStrictEqual, strictEqual } from 'node:assert';
+import { describe, it } from 'node:test';
+
+import { collectAsyncGenerator, createCache } from '../caching.mjs';
+
+describe('caching', () => {
+  describe('collectAsyncGenerator', () => {
+    it('should collect all chunks into a flat array', async () => {
+      async function* gen() {
+        yield [1, 2];
+        yield [3, 4];
+        yield [5];
+      }
+
+      const result = await collectAsyncGenerator(gen());
+
+      deepStrictEqual(result, [1, 2, 3, 4, 5]);
+    });
+
+    it('should return empty array for empty generator', async () => {
+      async function* gen() {
+        // empty generator
+      }
+
+      const result = await collectAsyncGenerator(gen());
+
+      deepStrictEqual(result, []);
+    });
+
+    it('should handle empty chunks', async () => {
+      async function* gen() {
+        yield [];
+        yield [1];
+        yield [];
+        yield [2, 3];
+      }
+
+      const result = await collectAsyncGenerator(gen());
+
+      deepStrictEqual(result, [1, 2, 3]);
+    });
+  });
+
+  describe('createCache', () => {
+    it('should report whether a generator is stored', () => {
+      const cache = createCache();
+
+      strictEqual(cache.has('a'), false);
+      cache.store('a', Promise.resolve(1));
+      strictEqual(cache.has('a'), true);
+    });
+
+    it('should pass through non-streaming results', async () => {
+      const cache = createCache();
+
+      cache.store('a', Promise.resolve({ value: 42 }));
+
+      deepStrictEqual(await cache.consume('a'), { value: 42 });
+    });
+
+    it('should collect a streaming result only once for all consumers', async () => {
+      const cache = createCache();
+
+      let iterations = 0;
+
+      async function* gen() {
+        iterations++;
+        yield [1, 2];
+        yield [3];
+      }
+
+      cache.store('a', gen());
+
+      const first = await cache.consume('a');
+      const second = await cache.consume('a');
+
+      deepStrictEqual(first, [1, 2, 3]);
+      // The same collected array is shared, collection happened a single time
+      strictEqual(first, second);
+      strictEqual(iterations, 1);
+    });
+
+    it('should count consumers across a dependency closure', async () => {
+      const cache = createCache();
+
+      // graph: a -> b -> c and d -> b; requested targets are a and d
+      const graph = { a: 'b', b: 'c', d: 'b' };
+
+      cache.populateConsumerCounts(['a', 'd'], name => graph[name]);
+
+      // `b` is depended on by both `a` and `d`, so it needs two reads before
+      // it is evicted.
+      cache.store('b', Promise.resolve('B'));
+      await cache.consume('b');
+      strictEqual(cache.has('b'), true);
+      await cache.consume('b');
+      strictEqual(cache.has('b'), false);
+
+      // `c` is depended on only by `b`, so a single read evicts it.
+      cache.store('c', Promise.resolve('C'));
+      await cache.consume('c');
+      strictEqual(cache.has('c'), false);
+
+      // `a` is a requested target, consumed exactly once by the final read.
+      cache.store('a', Promise.resolve('A'));
+      await cache.consume('a');
+      strictEqual(cache.has('a'), false);
+    });
+  });
+});

src/__tests__/generators.test.mjs

@@ -0,0 +1,140 @@
+import assert from 'node:assert/strict';
+import { describe, it, mock, beforeEach } from 'node:test';
+
+// Tracks how many times each synthetic generator actually executed so we can
+// assert that a shared dependency runs exactly once even after its result is
+// evicted from the cache once its last consumer has read it.
+const runs = {};
+
+const record = name => {
+  runs[name] = (runs[name] ?? 0) + 1;
+};
+
+// Yields a single chunk as an async generator (mirrors streaming generators)
+const streamOf = chunk =>
+  (async function* () {
+    yield chunk;
+  })();
+
+mock.module('../generators/index.mjs', {
+  namedExports: {
+    allGenerators: {
+      // Root streaming generator (no dependency)
+      ast: {
+        name: 'ast',
+        hasParallelProcessor: true,
+        generate: async () => {
+          record('ast');
+          return streamOf([{ ast: true }]);
+        },
+      },
+      // Streaming generator shared by multiple consumers
+      metadata: {
+        name: 'metadata',
+        dependsOn: 'ast',
+        hasParallelProcessor: true,
+        generate: async input => {
+          record('metadata');
+          return streamOf([{ meta: input.length }]);
+        },
+      },
+      // Two non-streaming consumers of the shared `metadata` result
+      'gen-a': {
+        name: 'gen-a',
+        dependsOn: 'metadata',
+        generate: async input => {
+          record('gen-a');
+          return { a: input };
+        },
+      },
+      'gen-b': {
+        name: 'gen-b',
+        dependsOn: 'metadata',
+        generate: async input => {
+          record('gen-b');
+          return { b: input };
+        },
+      },
+      // A target that is itself depended upon by another target
+      'gen-c': {
+        name: 'gen-c',
+        dependsOn: 'metadata',
+        hasParallelProcessor: true,
+        generate: async () => {
+          record('gen-c');
+          return streamOf([{ c: true }]);
+        },
+      },
+      'gen-c-all': {
+        name: 'gen-c-all',
+        dependsOn: 'gen-c',
+        generate: async input => {
+          record('gen-c-all');
+          return { all: input };
+        },
+      },
+    },
+  },
+});
+
+mock.module('../threading/index.mjs', {
+  defaultExport: () => ({
+    run: async () => undefined,
+    destroy: async () => undefined,
+  }),
+});
+
+mock.module('../threading/parallel.mjs', {
+  defaultExport: () => ({
+    async *stream() {
+      // Unused: the mocked generators return their own async generators
+    },
+  }),
+});
+
+const createGenerator = (await import('../generators.mjs')).default;
+
+describe('createGenerator orchestration', () => {
+  beforeEach(() => {
+    for (const key of Object.keys(runs)) {
+      delete runs[key];
+    }
+  });
+
+  it('runs a shared dependency exactly once and returns correct results', async () => {
+    const { runGenerators } = createGenerator();
+
+    const results = await runGenerators({
+      target: ['gen-a', 'gen-b'],
+      threads: 1,
+    });
+
+    // ast -> metadata are shared and must each execute a single time
+    assert.equal(runs.ast, 1);
+    assert.equal(runs.metadata, 1);
+    assert.equal(runs['gen-a'], 1);
+    assert.equal(runs['gen-b'], 1);
+
+    // metadata collected one entry whose `meta` is the ast array length (1)
+    assert.deepStrictEqual(results, [
+      { a: [{ meta: 1 }] },
+      { b: [{ meta: 1 }] },
+    ]);
+  });
+
+  it('does not re-run a target that is also another target dependency', async () => {
+    const { runGenerators } = createGenerator();
+
+    const results = await runGenerators({
+      target: ['gen-c', 'gen-c-all'],
+      threads: 1,
+    });
+
+    // gen-c is both a requested target and a dependency of gen-c-all; eviction
+    // must not cause it to be scheduled (and run) twice.
+    assert.equal(runs['gen-c'], 1);
+    assert.equal(runs['gen-c-all'], 1);
+
+    assert.deepStrictEqual(results, [[{ c: true }], { all: [{ c: true }] }]);
+  });
+});