From 1041557621dffa4b0195c9157b40e7d29848f4d0 Mon Sep 17 00:00:00 2001 From: mlischetti Date: Wed, 15 Jul 2026 12:34:29 -0300 Subject: [PATCH] fix(node-binding): resolve teardown crash, streaming race, and swallowed read errors Extracted from the native-bindings work in PR #119 so the fixes to the pre-existing Node binding can be reviewed and merged independently of the new C/Go/Rust bindings. addon.c: - Fix a teardown crash: graal_tear_down_isolate() must run on a thread attached to the isolate. The bootstrap thread from graal_create_isolate() is joined and exits, so its IsolateThread is invalid at cleanup and also leaves a phantom attached thread that blocks teardown forever. Detach the bootstrap thread after create, and attach the cleanup thread before tear down. - Surface the pending JS read-callback exception (message + stack) to stderr and signal an error (bytes_read = -1) instead of silently swallowing it. index.ts: - Fix a streaming race: replace the single resolveChunk slot with a pendingResolves queue so multiple waiting consumers are all woken; the metadata resolution drains the whole queue. - Release consumed pre-buffered input buffers so memory is freed as the async input is drained. Co-Authored-By: Claude Opus 4.8 (1M context) --- native-lib/node/src/addon.c | 65 +++++++++++++++++++++++++++++++++--- native-lib/node/src/index.ts | 42 +++++++++++++---------- 2 files changed, 85 insertions(+), 22 deletions(-) diff --git a/native-lib/node/src/addon.c b/native-lib/node/src/addon.c index b18c2eba..ed61f455 100644 --- a/native-lib/node/src/addon.c +++ b/native-lib/node/src/addon.c @@ -68,13 +68,26 @@ static void init_thread_fn(void* arg) { return; } - rc = fn_create_isolate(NULL, &g_isolate, &g_thread); + void* boot_thread = NULL; + rc = fn_create_isolate(NULL, &g_isolate, &boot_thread); if (rc != 0) { snprintf(args->error, sizeof(args->error), "graal_create_isolate failed with code %d", rc); args->result = rc; return; } + // Detach the bootstrap thread immediately. This init OS thread is joined and + // exits right after, so leaving it attached would leave a phantom attached + // thread on the isolate — and graal_tear_down_isolate() blocks forever waiting + // for every other attached thread to reach a safepoint (the dead init thread + // never will). Subsequent calls (run/streaming/transform, and cleanup) attach + // their own OS thread on demand and detach when done. Mirrors the Go binding, + // which likewise detaches the bootstrap thread after graal_create_isolate. + if (fn_detach_thread) { + fn_detach_thread(boot_thread); + } + g_thread = NULL; + args->result = 0; } @@ -392,7 +405,40 @@ static void call_js_read(napi_env env, napi_value js_callback, void* context, vo req->bytes_read = 0; } } else { - req->bytes_read = 0; + // Clear pending exception to prevent propagation + if (status == napi_pending_exception) { + napi_value exception; + napi_get_and_clear_last_exception(env, &exception); + + // Extract and log exception details before discarding + napi_value message_prop, stack_prop; + char message_buf[512] = {0}; + char stack_buf[2048] = {0}; + size_t message_len = 0, stack_len = 0; + + // Try to get the message property + if (napi_get_named_property(env, exception, "message", &message_prop) == napi_ok) { + napi_get_value_string_utf8(env, message_prop, message_buf, sizeof(message_buf), &message_len); + } + + // Try to get the stack property + if (napi_get_named_property(env, exception, "stack", &stack_prop) == napi_ok) { + napi_get_value_string_utf8(env, stack_prop, stack_buf, sizeof(stack_buf), &stack_len); + } + + // Log the exception to stderr for diagnostics + fprintf(stderr, "[DataWeave Node addon] Read callback threw exception:\n"); + if (message_len > 0) { + fprintf(stderr, " Message: %s\n", message_buf); + } + if (stack_len > 0) { + fprintf(stderr, " Stack:\n%s\n", stack_buf); + } + if (message_len == 0 && stack_len == 0) { + fprintf(stderr, " (Unable to extract exception details)\n"); + } + } + req->bytes_read = -1; // Signal error } uv_mutex_lock(&req->mutex); @@ -586,9 +632,20 @@ static napi_value napi_run_script_transform(napi_env env, napi_callback_info inf static void cleanup_thread_fn(void* arg) { (void)arg; - if (fn_tear_down_isolate && g_thread) { - fn_tear_down_isolate(g_thread); + // graal_tear_down_isolate() must be passed the IsolateThread belonging to the + // *calling* OS thread. g_thread was created by graal_create_isolate() on the + // (now-exited, already-joined) init thread, so it is invalid here — passing it + // trips GraalVM's "wrong IsolateThread" guard and aborts with a fatal + // StackOverflowError during teardown. Attach this cleanup thread to the isolate + // to obtain a valid local IsolateThread, then tear down with that. + if (!fn_tear_down_isolate || !fn_attach_thread || !g_isolate) { + return; + } + void* local_thread = NULL; + if (fn_attach_thread(g_isolate, &local_thread) != 0 || local_thread == NULL) { + return; } + fn_tear_down_isolate(local_thread); } static napi_value napi_cleanup(napi_env env, napi_callback_info info) { diff --git a/native-lib/node/src/index.ts b/native-lib/node/src/index.ts index 15451701..2902003e 100644 --- a/native-lib/node/src/index.ts +++ b/native-lib/node/src/index.ts @@ -149,24 +149,26 @@ export class DataWeave { const inputsJson = buildInputsJson(inputs ?? {}); const chunks: Buffer[] = []; - let resolveChunk: (() => void) | null = null; + const pendingResolves: Array<() => void> = []; let done = false; let metaRaw: string | null = null; const chunkCb = (chunk: Buffer) => { chunks.push(chunk); - if (resolveChunk) { - resolveChunk(); - resolveChunk = null; + // Resolve one waiting consumer if any + const resolve = pendingResolves.shift(); + if (resolve) { + resolve(); } }; const metaPromise = ffi.runScriptStreaming(script, inputsJson, chunkCb).then((raw) => { metaRaw = raw; done = true; - if (resolveChunk) { - resolveChunk(); - resolveChunk = null; + // Wake all waiting consumers + while (pendingResolves.length > 0) { + const resolve = pendingResolves.shift(); + if (resolve) resolve(); } }); @@ -176,7 +178,7 @@ export class DataWeave { continue; } if (done) break; - await new Promise((resolve) => { resolveChunk = resolve; }); + await new Promise((resolve) => { pendingResolves.push(resolve); }); } // Drain remaining chunks @@ -208,7 +210,7 @@ export class DataWeave { if (isAsync) { // Async iterables must be pre-buffered because the native read callback // is invoked synchronously on the JS main thread and cannot await. - const inputBuffers: Buffer[] = []; + const inputBuffers: (Buffer | null)[] = []; const asyncIter = (input as AsyncIterable)[Symbol.asyncIterator](); try { while (true) { @@ -235,7 +237,9 @@ export class DataWeave { return Buffer.from(slice); } if (bufIdx < inputBuffers.length) { - currentBuf = inputBuffers[bufIdx++]; + currentBuf = inputBuffers[bufIdx]; + inputBuffers[bufIdx] = null; // Release memory as we consume + bufIdx++; readOffset = 0; continue; } @@ -274,15 +278,16 @@ export class DataWeave { } const chunks: Buffer[] = []; - let resolveChunk: (() => void) | null = null; + const pendingResolves: Array<() => void> = []; let done = false; let metaRaw: string | null = null; const writeCb = (chunk: Buffer) => { chunks.push(chunk); - if (resolveChunk) { - resolveChunk(); - resolveChunk = null; + // Resolve one waiting consumer if any + const resolve = pendingResolves.shift(); + if (resolve) { + resolve(); } }; @@ -291,9 +296,10 @@ export class DataWeave { ).then((raw) => { metaRaw = raw; done = true; - if (resolveChunk) { - resolveChunk(); - resolveChunk = null; + // Wake all waiting consumers + while (pendingResolves.length > 0) { + const resolve = pendingResolves.shift(); + if (resolve) resolve(); } }); @@ -303,7 +309,7 @@ export class DataWeave { continue; } if (done) break; - await new Promise((resolve) => { resolveChunk = resolve; }); + await new Promise((resolve) => { pendingResolves.push(resolve); }); } while (chunks.length > 0) {