@W-23692110: Multiple isolated DataWeave engines per process (Node + Python)

.github/actions/node/action.yml

@@ -52,6 +52,13 @@ runs:
 
     - name: Run Node.js TCK Conformance
       if: always() && inputs.run-tck == 'true'
+      # This is the dedicated (master-only) TCK lane. The corpus is staged once
+      # by the shared TCK staging step earlier in the job (see main.yml). Set the
+      # require-corpus flag so a missing/empty staged corpus fails this job loudly
+      # instead of skipping silently (review #10 #7); local dev without the flag
+      # still skips.
+      env:
+        DATAWEAVE_TCK_REQUIRE_CORPUS: '1'
       run: |
         cd native-lib/node && npm run test:tck
       shell: bash

.gitignore

@@ -32,3 +32,8 @@ grimoires/
 # Superpowers implementation plans are local scratch artifacts, never commit them.
 /docs/superpowers/plans/
 /docs/superpowers/plans/**/*.md
+
+# PR follow-up code-review notes are local scratch, keep them untracked.
+/docs/pr-*-follow-up-*code-review*.md
+# GA cleanup backlog is a local working note, keep it untracked.
+/docs/ga-cleanup-backlog.md

native-lib/README.md

@@ -22,14 +22,29 @@ The main purpose is to allow non-JVM consumers (most notably the Python package
 │  ┌────────────────────────────────────────┐ │
 │  │  Native Shared Library (dwlib)         │ │
 │  │  ┌──────────────────────────────────┐  │ │
-│  │  │  GraalVM Isolate                 │  │ │
-│  │  │  - NativeLib.run_script()        │  │ │
+│  │  │  GraalVM Isolate (process-wide)  │  │ │
+│  │  │  - create_engine /                │  │ │
+│  │  │    create_engine_with_resolver    │  │ │
+│  │  │  - run_script_engine /            │  │ │
+│  │  │    run_script_callback_engine /   │  │ │
+│  │  │    run_script_input_output_       │  │ │
+│  │  │    callback_engine                │  │ │
+│  │  │  - destroy_engine                 │  │ │
 │  │  │  - DataWeave script execution    │  │ │
 │  │  └──────────────────────────────────┘  │ │
 │  └────────────────────────────────────────┘ │
 └─────────────────────────────────────────────┘
 ```
 
+Each engine is a handle-addressed object created with `create_engine` (or
+`create_engine_with_resolver`, which additionally registers a module-resolve
+callback) and run via `run_script_engine`, `run_script_callback_engine`, or
+`run_script_input_output_callback_engine`, then released with
+`destroy_engine`. The underlying GraalVM isolate is a single process-wide
+isolate, created and attached via `graal_create_isolate` / `graal_attach_thread`
+on first use and torn down via `graal_tear_down_isolate` once the last engine
+across the process has been destroyed.
+
 ## Building with Gradle
 
 ### Prerequisites
@@ -450,14 +465,21 @@ import { DataWeave } from "@dataweave/native";
 
 const dw = new DataWeave();
 dw.initialize();
-
-const r1 = dw.run("2 + 2");
-const r2 = dw.run("x + y", { x: 10, y: 32 });
-
-console.log(r1.getString()); // "4"
-console.log(r2.getString()); // "42"
-
-dw.cleanup();
+try {
+  const r1 = dw.run("2 + 2");
+  const r2 = dw.run("x + y", { x: 10, y: 32 });
+
+  console.log(r1.getString()); // "4"
+  console.log(r2.getString()); // "42"
+} finally {
+  // cleanup() returns a Promise; await it. When this releases the FINAL shared
+  // native reference in the process, it drains any in-flight streaming/transform
+  // op and completes isolate teardown before resolving (so a subsequent
+  // initialize() does not race a still-tearing-down isolate). When other
+  // initialized instances remain, it resolves as soon as this instance is
+  // released, leaving the shared isolate live for them.
+  await dw.cleanup();
+}
 ```
 
 ### 5) Error handling
@@ -635,11 +657,20 @@ for await (const chunk of gen) {
 
 ### 9) Cleanup
 
-The module registers a `process.on('exit')` handler to clean up automatically. For explicit control:
+The module registers two process hooks to clean up automatically: `beforeExit`
+(async — it awaits cleanup so an in-flight streaming/transform op drains before
+the process exits normally) and `exit` (a synchronous best-effort fallback for
+`process.exit()` and uncaught exceptions, which cannot await the drain). Neither
+hook fires on `SIGTERM`/`SIGINT`/`SIGKILL`, so install your own signal handler
+that awaits `cleanup()` if you need a graceful drain on termination. For explicit
+control:
 
 ```typescript
 import { cleanup } from "@dataweave/native";
 
-// When done with all DataWeave operations
-cleanup();
+// When done with all DataWeave operations. cleanup() returns a Promise; await it.
+// Draining in-flight streaming/transform work and tearing down the isolate happen
+// only when this releases the final shared native reference; if other initialized
+// instances remain, it resolves as soon as this instance is released.
+await cleanup();
 ```

native-lib/node/README.md

@@ -128,13 +128,17 @@ const generator = runStreaming(
   '%dw 2.0\noutput application/json\n---\n[1, 2, 3, 4, 5]'
 );
 
-for await (const chunk of generator) {
-  console.log('Chunk:', chunk.toString());
+// Iterate manually with next() to capture the terminal return value. A
+// `for await` loop consumes the generator's return value internally, so a later
+// generator.return() would yield { value: undefined } -- drive next() yourself
+// and read the metadata off the terminal { done: true, value: StreamingResult }.
+let meta;
+while (true) {
+  const { value, done } = await generator.next();
+  if (done) { meta = value; break; }
+  console.log('Chunk:', value.toString());
 }
-
-// Generator return value contains metadata:
-const meta = await generator.return();
-console.log('MIME type:', meta.value.mimeType);
+console.log('MIME type:', meta.mimeType);
 ```
 
 **Parameters:**
@@ -156,12 +160,21 @@ Execute a DataWeave script with streaming input and output (bidirectional stream
 
 ```javascript
 import { runTransform } from '@dataweave/native';
-import { createReadStream } from 'fs';
+import { readFileSync } from 'fs';
+
+// The native read callback is synchronous, so an ASYNC input iterable (e.g.
+// fs.createReadStream) is fully pre-buffered into memory before the transform
+// starts. A SYNCHRONOUS iterable is instead consumed on demand -- one chunk at a
+// time -- so the transform makes no extra full copy of the input (it does NOT by
+// itself bound total memory: a source like readFileSync still holds the whole
+// input). (See "Sync vs async input and memory" below.)
+function* chunked(buf, size = 65536) {
+  for (let i = 0; i < buf.length; i += size) yield buf.subarray(i, i + size);
+}
 
-// Transform a large CSV file to JSON without loading it all into memory
 const generator = runTransform(
   '%dw 2.0\noutput application/json\n---\npayload',
-  createReadStream('large-file.csv'),
+  chunked(readFileSync('large-file.csv')),
   {
     inputName: 'payload',
     mimeType: 'application/csv',
@@ -175,28 +188,39 @@ for await (const chunk of generator) {
 }
 ```
 
+> **Sync vs async input and memory.** The native read callback runs synchronously
+> on the JS thread. **Synchronous** iterables (arrays, generators) are consumed
+> on demand — the transform holds only one chunk at a time and makes no extra
+> full copy of the input. This bounds the transform's *added* memory, not total
+> memory: if the source itself already holds the whole input (e.g. `readFileSync`),
+> that memory is still resident. **Async** iterables (e.g. `fs.createReadStream()`)
+> are **fully pre-buffered** into memory before the transform starts, because their
+> `.next()` returns a Promise that cannot be awaited inside the synchronous
+> callback. For large inputs, prefer a synchronous generator so the transform adds
+> no second copy.
+
 **Parameters:**
 - `script` (string): DataWeave script
 - `input` (AsyncIterable<Buffer> | Iterable<Buffer>): Streaming input data
 - `opts` (object, optional): Options
   - `inputName` (string): Name of input variable (default: "payload")
   - `mimeType` (string): Input MIME type (default: "application/json")
-  - `charset` (string | null): Input character encoding
+  - `charset` (string, optional): Input character encoding
   - `inputs` (object): Additional input variables
 
 **Yields:** `Buffer` chunks as they're produced
 
 **Returns:** `StreamingResult`
 
-#### `cleanup(): void`
+#### `cleanup(): Promise<void>`
 
-Clean up the global DataWeave runtime instance. Called automatically on process exit.
+Clean up the global DataWeave runtime instance. Called automatically on process shutdown via two hooks: `beforeExit` awaits it, so a streaming/transform operation still in flight drains gracefully before the process exits normally; `exit` is a synchronous last-ditch fallback for `process.exit()` and uncaught exceptions — cases where `beforeExit` never fires — and cannot await the drain. Neither hook fires on `SIGTERM`, `SIGINT`, or `SIGKILL` (Node does not emit `exit` for signals), so install your own signal handler that calls `cleanup()` if you need a graceful drain on termination. Called manually, it releases this instance's reference to the native runtime; the shared native isolate is torn down only when the **last** initialized instance in the process is released. When this call releases that final reference, it resolves once native teardown has actually finished, waiting for any still-in-flight streaming/transform operation to drain first; otherwise (other instances remain initialized) it resolves as soon as this instance is released, without draining process-wide work.
 
 ```javascript
 import { cleanup } from '@dataweave/native';
 
 // Manual cleanup (usually not needed)
-cleanup();
+await cleanup();
 ```
 
 ### Class-Based API
@@ -213,13 +237,13 @@ try {
   const result = dw.run('2 + 2');
   console.log(result.getString());
 } finally {
-  dw.cleanup();
+  await dw.cleanup();
 }
 ```
 
 **Methods:**
 - `initialize()`: Initialize the native library
-- `cleanup()`: Release native resources
+- `cleanup(): Promise<void>`: Release this instance's native resources. When it releases the last initialized instance in the process, it resolves once the shared isolate has finished tearing down (draining any in-flight streaming/transform op first); otherwise it resolves as soon as this instance is released, leaving the isolate live for other instances.
 - `run(script, inputs?, opts?)`: Same as module-level `run()`
 - `runStreaming(script, inputs?)`: Same as module-level `runStreaming()`
 - `runTransform(script, input, opts?)`: Same as module-level `runTransform()`
@@ -231,29 +255,42 @@ DataWeave scripts can import external modules using the `resolveModule` option.
 ```typescript
 import { DataWeave, composeResolvers, modulesFromDirectory, modulesFromJars } from '@dataweave/native';
 
+// Inside an async function (uses `await` for modulesFromJars and cleanup()).
 const dw = new DataWeave({
   resolveModule: composeResolvers(
     modulesFromDirectory('./my-modules'),
     await modulesFromJars(['./libs/dw-utils.jar'])
   )
 });
 dw.initialize();
-
-const result = dw.run(`
-  %dw 2.0
-  import org::company::utils
-  output application/json
-  ---
-  utils::doSomething()
-`);
-
-if (result.success) {
-  console.log(result.getString());
+try {
+  const result = dw.run(`
+    %dw 2.0
+    import org::company::utils
+    output application/json
+    ---
+    utils::doSomething()
+  `);
+
+  if (result.success) {
+    console.log(result.getString());
+  }
+} finally {
+  // Release the engine and resolver closure when done.
+  await dw.cleanup();
 }
 ```
 
 See [docs/external-modules.md](docs/external-modules.md) for complete documentation, resolver factories, error handling, and dependency management. Note: a resolver runs with full process permissions (no sandboxing) — see the "Security / Trust Model" section there before pointing one at untrusted sources.
 
+### Custom module resolution scope
+
+- A `resolveModule` you configure applies to `run()`.
+- Built-in modules (e.g. `dw::core::*`) resolve everywhere — `run()`, `runStreaming()`, and `runTransform()`.
+- Custom modules do **not** resolve inside `runStreaming()`/`runTransform()`: those execute on a background thread that must not call back into your resolver, so a streamed/transformed script that imports a custom module fails closed (reports the module as not found) rather than making an unsafe cross-thread call. If you need a custom module in a streamed/transform script, resolve it via `run()` instead, or inline the module into the script.
+
+See [docs/external-modules.md](docs/external-modules.md#multiple-independent-engines) for the full explanation, including the Worker-thread ownership rules.
+
 ### Input Formats
 
 Inputs can be provided in multiple formats:
@@ -366,7 +403,7 @@ console.log(result.getString());  // "300"
 
 ```javascript
 import { runTransform } from '@dataweave/native';
-import { createReadStream, createWriteStream } from 'fs';
+import { readFileSync, createWriteStream } from 'fs';
 
 const script = `
 %dw 2.0
@@ -375,9 +412,18 @@ output application/json
 payload filter $.amount > 1000
 `;
 
+// A synchronous generator is consumed on demand: the transform does not make a
+// second full copy of the input. Note readFileSync still holds the whole file in
+// memory, so this bounds the transform's *added* memory, not total memory -- the
+// native read callback is synchronous, so there is no fully-streaming-from-disk
+// path (an async createReadStream would instead be pre-buffered in full first).
+function* chunked(buf, size = 65536) {
+  for (let i = 0; i < buf.length; i += size) yield buf.subarray(i, i + size);
+}
+
 const generator = runTransform(
   script,
-  createReadStream('large-transactions.csv'),
+  chunked(readFileSync('large-transactions.csv')),
   { mimeType: 'application/csv' }
 );
 
@@ -422,12 +468,17 @@ try {
 ```javascript
 try {
   const generator = runStreaming('invalid syntax');
-  for await (const chunk of generator) {
-    // Process chunks
+  // Drive next() manually so the terminal { done: true, value: StreamingResult }
+  // is captured; a `for await` loop would consume it and a later
+  // generator.return() would give { value: undefined }.
+  let meta;
+  while (true) {
+    const { value, done } = await generator.next();
+    if (done) { meta = value; break; }
+    // Process chunk `value`
   }
-  const meta = await generator.return();
-  if (!meta.value.success) {
-    console.error('Streaming error:', meta.value.error);
+  if (!meta.success) {
+    console.error('Streaming error:', meta.error);
   }
 } catch (err) {
   console.error('Native error:', err);
@@ -440,20 +491,20 @@ The Node.js binding uses **N-API** (Node-API) for C addon integration:
 
 - **Thread-safe**: N-API calls are serialized on the Node.js event loop
 - **Async operations**: Streaming operations yield control to the event loop between chunks
-- **No blocking**: Long-running scripts execute on the native side without blocking the event loop
+- **No event-loop blocking for streaming**: `runStreaming`/`runTransform` execute on a background worker and yield to the event loop between chunks. Note the **synchronous** `run()` runs native work directly on the calling JS thread and *does* block it until the script completes — use the streaming methods for long-running work you cannot block on.
 
 **Important:** Do not share a single `DataWeave` instance across Worker threads. Use the module-level functions (which use a global singleton) or create separate instances per thread.
 
-**Custom module resolvers and Worker threads:** the native layer installs at
-most one resolver callback for the whole process lifetime, and it is bound to
-the Worker (main thread or a `worker_threads` Worker) that registered it
-first — see [External Modules: Multiple Resolvers](docs/external-modules.md#multiple-resolvers-in-one-process).
+**Custom module resolvers and Worker threads:** each resolver-backed
+`DataWeave` instance's native engine is bound to the thread that created it
+(main thread or a `worker_threads` Worker) — see
+[External Modules: Multiple Independent Engines](docs/external-modules.md#multiple-independent-engines).
 Custom-module resolution attempted from any *other* thread is not routed to
-that thread's own `resolveModule` callback; it silently falls back to
-built-in modules only (custom module paths resolve as "not found" rather than
-crashing or hanging). If you need per-Worker custom modules, resolve them on
-the thread that first constructs a resolver-backed `DataWeave` instance, or
-avoid resolver-backed instances in worker pools altogether.
+that engine's `resolveModule` callback; it silently falls back to built-in
+modules only (custom module paths resolve as "not found" rather than
+crashing or hanging). If you need custom modules on multiple Workers,
+construct and use a separate resolver-backed `DataWeave` instance on each
+Worker, created on that Worker itself.
 
 ## Platform Support
 
@@ -541,12 +592,12 @@ Tests use **Vitest** and cover:
 
 - **Buffered execution** (`run`): Best for small scripts with sub-MB outputs
 - **Streaming execution** (`runStreaming`): Best for large outputs (MB+), reduces memory footprint
-- **Bidirectional streaming** (`runTransform`): Best for large inputs and outputs, constant memory usage
+- **Bidirectional streaming** (`runTransform`): Best for large outputs; input memory is bounded only with a **synchronous** input iterable (async streams are pre-buffered — see the `runTransform` memory note above)
 
 Benchmark (1MB JSON transformation):
 - `run()`: ~50ms, 2MB peak memory
 - `runStreaming()`: ~55ms, 500KB peak memory
-- `runTransform()`: ~60ms, 256KB peak memory (streaming input)
+- `runTransform()`: ~60ms, 256KB peak memory (synchronous input iterable; an async stream is pre-buffered, so peak memory scales with input size)
 
 ## See Also