Repository: un-ts/synckit (ts)

Issue:
[feat] override stdout & stderr in worker_thread for debug worker 
According to the research based on the issue at https://github.com/jimmywarting/await-sync/issues/1#issuecomment-1538762664

Worker threads in `Node.js` send their standard IO to the main thread for processing. When synchronous operations are implemented, the blocking of these operations can cause the logs of these worker threads to be synchronized and output only after the computation results are completed.

Similarly, following the implementation of `whatwg-workers`, by rewriting the write methods of `stdout` and `stderr` streams within worker threads, it is possible to allow the standard input and standard error within the worker threads to be directly written into the streams.

``` typescript
// caused by https://github.com/nodejs/node/blob/0b3fcfcf351fba9f29234976eeec4afb09ae2cc0/lib/internal/worker/io.js#L360
// override process write to make stdout and stderr work in worker thread
function overrideProcess4WorkThread() {
  for (const [i, stream] of [process.stdout, process.stderr].entries()) {
    // process.stdout.fd === 1 & process.stderr.fd === 2 . In Worker threads, this field does not exist.
    const fd = i + 1
    const streamOriginWritev = stream._writev?.bind(stream)
    stream._writev = (chunks, cb) => {
      if (chunks.length === 0) {
        return
      }
      const chunk = chunks.pop()!
      // type-coverage:ignore-next-line -- we can't control
      fs.write(fd, chunk.chunk as string, null, chunk.encoding, err => {
        if (err) cb(err)
        else if (chunks.length === 0) cb()
        else streamOriginWritev?.(chunks, cb)
      })
    }
  }
}
```
and test demo output 
![image](https://github.com/un-ts/synckit/assets/24679861/fa879886-6472-4dda-9f83-8823bc3e914d)

If allowed, I'm glad to raise a pr

Pull request:
feat: support `stdio` overriding in worker
close #161

Patch:
diff --git a/.changeset/rich-cooks-clean.md b/.changeset/rich-cooks-clean.md
new file mode 100644
index 0000000..1d16fba
--- /dev/null
+++ b/.changeset/rich-cooks-clean.md
@@ -0,0 +1,5 @@
+---
+"synckit": minor
+---
+
+feat: support `stdio` overriding in worker, what means `stdio` will be streamed in correct order now
diff --git a/README.md b/README.md
index 8327a78..e301c7d 100644
--- a/README.md
+++ b/README.md
@@ -11,7 +11,7 @@
 [![JavaScript Style Guide](https://img.shields.io/badge/code_style-standard-brightgreen.svg)](https://standardjs.com)
 [![Code Style: Prettier](https://img.shields.io/badge/code_style-prettier-ff69b4.svg)](https://github.com/prettier/prettier)
 
-Perform async work synchronously in Node.js using `worker_threads` with first-class TypeScript and Yarn P'n'P support.
+Perform async work synchronously in Node.js/Bun using `worker_threads` with first-class TypeScript and Yarn P'n'P support.
 
 ## TOC <!-- omit in toc -->
 
diff --git a/src/index.ts b/src/index.ts
index 95e8e44..c0b0d32 100644
--- a/src/index.ts
+++ b/src/index.ts
@@ -22,6 +22,7 @@ import type {
   MainToWorkerCommandMessage,
   MainToWorkerMessage,
   PackageJson,
+  StdioChunk,
   Syncify,
   ValueOf,
   WorkerData,
@@ -685,12 +686,16 @@ function startWorkerThread<T extends AnyFn, R = Awaited<ReturnType<T>>>(
 
     worker.postMessage(msg)
 
-    const { result, error, properties } = receiveMessageWithId(
+    const { result, error, properties, stdio } = receiveMessageWithId(
       mainPort,
       id,
       timeout,
     )
 
+    for (const { type, chunk, encoding } of stdio) {
+      process[type].write(chunk, encoding)
+    }
+
     if (error) {
       // eslint-disable-next-line @typescript-eslint/only-throw-error
       throw Object.assign(error as object, properties)
@@ -713,6 +718,27 @@ export function runAsWorker<T extends AnyFn<Promise<R> | R>, R = ReturnType<T>>(
     return
   }
 
+  const stdio: StdioChunk[] = []
+
+  // https://github.com/nodejs/node/blob/66556f53a7b36384bce305865c30ca43eaa0874b/lib/internal/worker/io.js#L369
+  for (const type of ['stdout', 'stderr'] as const) {
+    process[type]._writev = (chunks, callback) => {
+      for (const {
+        // type-coverage:ignore-next-line -- we can't control
+        chunk,
+        encoding,
+      } of chunks) {
+        stdio.push({
+          type,
+          // eslint-disable-next-line @typescript-eslint/no-unsafe-assignment -- we can't control
+          chunk,
+          encoding,
+        })
+      }
+      callback()
+    }
+  }
+
   const { workerPort, sharedBuffer, pnpLoaderPath } = workerData as WorkerData
 
   if (
@@ -739,18 +765,23 @@ export function runAsWorker<T extends AnyFn<Promise<R> | R>, R = ReturnType<T>>(
         workerPort.on('message', handleAbortMessage)
         let msg: WorkerToMainMessage<Awaited<R>>
         try {
-          msg = { id, result: await fn(...args) }
+          msg = { id, stdio, result: await fn(...args) }
         } catch (error: unknown) {
-          msg = { id, error, properties: extractProperties(error) }
+          msg = { id, stdio, error, properties: extractProperties(error) }
         }
         workerPort.off('message', handleAbortMessage)
         // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition -- false positive for `handleAbortMessage`
         if (isAborted) {
+          stdio.length = 0
           return
         }
-        workerPort.postMessage(msg)
-        Atomics.add(sharedBufferView, 0, 1)
-        Atomics.notify(sharedBufferView, 0)
+        try {
+          workerPort.postMessage(msg)
+          Atomics.add(sharedBufferView, 0, 1)
+          Atomics.notify(sharedBufferView, 0)
+        } finally {
+          stdio.length = 0
+        }
       })()
     },
   )
diff --git a/src/types.ts b/src/types.ts
index a146d71..66352a5 100644
--- a/src/types.ts
+++ b/src/types.ts
@@ -31,8 +31,15 @@ export interface DataMessage<T> {
   properties?: unknown
 }
 
+export interface StdioChunk {
+  type: 'stderr' | 'stdout'
+  chunk: Uint8Array | string
+  encoding: BufferEncoding
+}
+
 export interface WorkerToMainMessage<T> extends DataMessage<T> {
   id: number
+  stdio: StdioChunk[]
 }
 
 export interface GlobalShim {


Test patch:
diff --git a/test/fixtures/__snapshots__/fixtures.spec.ts.snap b/test/fixtures/__snapshots__/fixtures.spec.ts.snap
index d47e00c..f4d9b6a 100644
--- a/test/fixtures/__snapshots__/fixtures.spec.ts.snap
+++ b/test/fixtures/__snapshots__/fixtures.spec.ts.snap
@@ -4,7 +4,10 @@ exports[`bun 1`] = `
 {
   "exitCode": 0,
   "stderr": "",
-  "stdout": "1 2 5
+  "stdout": "Worker: 1
+Worker: 2
+Worker: 5
+1 2 5
 ",
 }
 `;
@@ -13,7 +16,10 @@ exports[`yarn-pnp 1`] = `
 {
   "exitCode": 0,
   "stderr": "",
-  "stdout": "1 2 5
+  "stdout": "Worker: 1
+Worker: 2
+Worker: 5
+1 2 5
 ",
 }
 `;
diff --git a/test/fixtures/bun/worker.ts b/test/fixtures/bun/worker.ts
index 89d8305..2b7c137 100644
--- a/test/fixtures/bun/worker.ts
+++ b/test/fixtures/bun/worker.ts
@@ -2,5 +2,10 @@ import { runAsWorker } from 'synckit'
 
 runAsWorker(
   (result: number, timeout: number = result) =>
-    new Promise<number>(resolve => setTimeout(() => resolve(result), timeout)),
+    new Promise<number>(resolve =>
+      setTimeout(() => {
+        console.log('Worker:', result)
+        return resolve(result)
+      }, timeout),
+    ),
 )
diff --git a/test/fixtures/yarn-pnp/worker.js b/test/fixtures/yarn-pnp/worker.js
index a005813..2a62e23 100644
--- a/test/fixtures/yarn-pnp/worker.js
+++ b/test/fixtures/yarn-pnp/worker.js
@@ -2,5 +2,10 @@ import { runAsWorker } from 'synckit'
 
 runAsWorker(
   (result, timeout = result) =>
-    new Promise(resolve => setTimeout(() => resolve(result), timeout)),
+    new Promise(resolve =>
+      setTimeout(() => {
+        console.log('Worker:', result)
+        return resolve(result)
+      }, timeout),
+    ),
 )
diff --git a/test/fn.spec.ts b/test/fn.spec.ts
index 9280a3d..ce38279 100644
--- a/test/fn.spec.ts
+++ b/test/fn.spec.ts
@@ -1,8 +1,8 @@
-import { createRequire } from 'node:module'
 import path from 'node:path'
 import { pathToFileURL } from 'node:url'
 
 import { jest } from '@jest/globals'
+import { cjsRequire } from '@pkgr/core'
 
 import {
   _dirname,
@@ -12,7 +12,7 @@ import {
 } from './helpers.js'
 import type { AsyncWorkerFn } from './types.js'
 
-import { createSyncFn } from 'synckit'
+import { type StdioChunk, createSyncFn } from 'synckit'
 
 const { SYNCKIT_TIMEOUT } = process.env
 
@@ -29,8 +29,6 @@ beforeEach(() => {
   }
 })
 
-const cjsRequire = createRequire(import.meta.url)
-
 const workerCjsTsPath = path.resolve(_dirname, 'cjs/worker-cjs.ts')
 const workerEsmTsPath = path.resolve(_dirname, 'esm/worker-esm.ts')
 const workerNoExtAsJsPath = path.resolve(_dirname, 'worker-js')
@@ -130,6 +128,8 @@ test('subsequent executions after timeout', async () => {
   expect(syncFn(3, 1)).toBe(3)
 })
 
+const stdio: StdioChunk[] = []
+
 test('handling of outdated message from worker', async () => {
   const executionTimeout = 60
   process.env.SYNCKIT_TIMEOUT = executionTimeout.toString()
@@ -138,8 +138,8 @@ test('handling of outdated message from worker', async () => {
   jest.spyOn(Atomics, 'wait').mockReturnValue('ok')
 
   receiveMessageOnPortMock
-    .mockReturnValueOnce({ message: { id: -1 } })
-    .mockReturnValueOnce({ message: { id: 0, result: 1 } })
+    .mockReturnValueOnce({ message: { id: -1, stdio } })
+    .mockReturnValueOnce({ message: { id: 0, stdio, result: 1 } })
 
   const { createSyncFn } = await import('synckit')
   const syncFn = createSyncFn<AsyncWorkerFn>(workerCjsPath)
@@ -154,8 +154,8 @@ test('propagation of undefined timeout', async () => {
   const atomicsWaitSpy = jest.spyOn(Atomics, 'wait').mockReturnValue('ok')
 
   receiveMessageOnPortMock
-    .mockReturnValueOnce({ message: { id: -1 } })
-    .mockReturnValueOnce({ message: { id: 0, result: 1 } })
+    .mockReturnValueOnce({ message: { id: -1, stdio } })
+    .mockReturnValueOnce({ message: { id: 0, stdio, result: 1 } })
 
   const { createSyncFn } = await import('synckit')
   const syncFn = createSyncFn<AsyncWorkerFn>(workerCjsPath)
@@ -187,8 +187,8 @@ test('reduction of waiting time', async () => {
   })
 
   receiveMessageOnPortMock
-    .mockReturnValueOnce({ message: { id: -1 } })
-    .mockReturnValueOnce({ message: { id: 0, result: 1 } })
+    .mockReturnValueOnce({ message: { id: -1, stdio } })
+    .mockReturnValueOnce({ message: { id: 0, stdio, result: 1 } })
 
   const { createSyncFn } = await import('synckit')
   const syncFn = createSyncFn<AsyncWorkerFn>(workerCjsPath)
@@ -210,7 +210,7 @@ test('unexpected message from worker', async () => {
   jest.spyOn(Atomics, 'wait').mockReturnValue('ok')
 
   const receiveMessageOnPortMock = await setupReceiveMessageOnPortMock()
-  receiveMessageOnPortMock.mockReturnValueOnce({ message: { id: 100 } })
+  receiveMessageOnPortMock.mockReturnValueOnce({ message: { id: 100, stdio } })
 
   const { createSyncFn } = await import('synckit')
   const syncFn = createSyncFn<AsyncWorkerFn>(workerCjsPath)
diff --git a/test/helpers.ts b/test/helpers.ts
index a84ca35..b8747de 100644
--- a/test/helpers.ts
+++ b/test/helpers.ts
@@ -5,9 +5,10 @@ import WorkerThreads from 'node:worker_threads'
 import { jest } from '@jest/globals'
 
 import {
-  compareVersion,
+  type WorkerToMainMessage,
   MTS_SUPPORTED_NODE_VERSION,
   NODE_VERSION,
+  compareVersion,
 } from 'synckit'
 
 export const _dirname = path.dirname(fileURLToPath(import.meta.url))
@@ -22,22 +23,16 @@ export const tsUseEsmSupported =
 export const testIf = (condition: boolean) => (condition ? it : it.skip)
 
 type ReceiveMessageOnPortMock = jest.Mock<
-  typeof WorkerThreads.receiveMessageOnPort
+  <T>(port: WorkerThreads.MessagePort) => { message: WorkerToMainMessage<T> }
 >
 export const setupReceiveMessageOnPortMock =
   async (): Promise<ReceiveMessageOnPortMock> => {
-    jest.unstable_mockModule('node:worker_threads', () => {
-      return {
-        ...WorkerThreads,
-        receiveMessageOnPort: jest.fn(WorkerThreads.receiveMessageOnPort),
-      }
-    })
-
-    const { receiveMessageOnPort: receiveMessageOnPortMock } = (await import(
-      'node:worker_threads'
-    )) as unknown as {
-      receiveMessageOnPort: ReceiveMessageOnPortMock
-    }
-
-    return receiveMessageOnPortMock
+    jest.unstable_mockModule('node:worker_threads', () => ({
+      ...WorkerThreads,
+      receiveMessageOnPort: jest.fn(WorkerThreads.receiveMessageOnPort),
+    }))
+
+    const { receiveMessageOnPort } = await import('node:worker_threads')
+
+    return receiveMessageOnPort as ReceiveMessageOnPortMock
   }