turbopack-node: linear IPC frame parsing with eager UTF-8 decoding

@@ -0,0 +1,217 @@+/* eslint-env jest */+import { PacketReader } from '../../turbopack/crates/turbopack-node/js/src/child_process/packet-reader'++function frame(payload: string): Buffer {+  const body = Buffer.from(payload, 'utf8')+  const header = Buffer.alloc(4)+  header.writeUInt32BE(body.length, 0)+  return Buffer.concat([header, body])+}++const messages: unknown[] = [+  { a: 1 },+  'é'.repeat(1000),+  [1, 2, 3],+  { big: 'x'.repeat(70000) },+  null,+  'z',+]+const stream = Buffer.concat(messages.map((m) => frame(JSON.stringify(m))))++function splitEvery(buffer: Buffer, size: number): Buffer[] {+  const chunks: Buffer[] = []+  for (let i = 0; i < buffer.length; i += size) {+    chunks.push(buffer.subarray(i, i + size))+  }+  return chunks+}++function splitRandomly(buffer: Buffer, seed: number): Buffer[] {+  const chunks: Buffer[] = []+  for (let i = 0; i < buffer.length; ) {+    seed = (seed * 1103515245 + 12345) & 0x7fffffff+    const size = 1 + (seed % 9000)+    chunks.push(buffer.subarray(i, i + size))+    i += size+  }+  return chunks+}++async function receiveAll(+  reader: PacketReader<unknown>,+  count: number+): Promise<unknown[]> {+  const received: unknown[] = []+  for (let i = 0; i < count; i++) {+    received.push(await reader.recv())+  }+  return received+}++describe('turbopack-node PacketReader', () => {+  it.each([+    ['in one chunk', () => [stream]],+    ['byte by byte', () => splitEvery(stream, 1)],+    ['in 3 byte chunks', () => splitEvery(stream, 3)],+    ['in random chunks', () => splitRandomly(stream, 42)],+  ])('decodes frames delivered %s', async (_name, split) => {+    const reader = new PacketReader<unknown>()+    for (const chunk of split()) {+      reader.push(chunk)+    }+    expect(await receiveAll(reader, messages.length)).toEqual(messages)+  })++  it('handles a chunk that ends one frame and starts the next header', async () => {+    const first = frame('"first"')+    const second = frame('"second"')+    const reader = new PacketReader<unknown>()+    reader.push(first.subarray(0, 5))+    reader.push(Buffer.concat([first.subarray(5), second.subarray(0, 2)]))+    reader.push(second.subarray(2))+    expect(await receiveAll(reader, 2)).toEqual(['first', 'second'])+  })++  it('resolves recv() calls that were pending before data arrived in order', async () => {+    const reader = new PacketReader<unknown>()+    const pending = messages.map(() => reader.recv())+    for (const chunk of splitRandomly(stream, 7)) {+      reader.push(chunk)+    }+    expect(await Promise.all(pending)).toEqual(messages)+  })++  it('keeps framing after a queued zero-length frame', async () => {+    const reader = new PacketReader<unknown>()+    reader.push(Buffer.concat([frame(''), frame('"after"')]))+    await expect(reader.recv()).rejects.toThrow(SyntaxError)+    expect(await reader.recv()).toBe('after')+  })++  it('throws from push() when a zero-length frame completes a pending recv()', () => {+    const reader = new PacketReader<unknown>()+    void reader.recv()+    expect(() => reader.push(frame(''))).toThrow(SyntaxError)+  })++  it('decodes chunks that are consumed whole without slicing them', async () => {+    const payload = Buffer.from('{"exact":true}', 'utf8')+    const header = Buffer.alloc(4)+    header.writeUInt32BE(payload.length, 0)+    const reader = new PacketReader<unknown>()+    const subarray = jest.spyOn(Buffer.prototype, 'subarray')+    // The payload chunk itself (not a copy or view of it) is what gets decoded.+    const toString = jest.spyOn(payload, 'toString')+    try {+      reader.push(header)+      reader.push(payload)+      expect(subarray).not.toHaveBeenCalled()+      expect(toString.mock.calls).toEqual([['utf8', 0, payload.length]])+    } finally {+      subarray.mockRestore()+      toString.mockRestore()+    }+    expect(await reader.recv()).toEqual({ exact: true })+  })++  describe('matches Buffer#toString decoding for every split point', () => {+    // Captures the decoded strings by bypassing JSON.parse, so invalid JSON+    // (e.g. a truncated sequence) can be compared too. All payloads are+    // sent as consecutive frames, so decoder state must not leak between them.+    async function check(...payloads: Buffer[]) {+      const stream = Buffer.concat(+        payloads.map((payload) => {+          const header = Buffer.alloc(4)+          header.writeUInt32BE(payload.length, 0)+          return Buffer.concat([header, payload])+        })+      )+      const expected = payloads.map((payload) => payload.toString('utf8'))+      for (let split = 1; split < stream.length; split++) {+        const reader = new PacketReader<unknown>()+        reader.push(stream.subarray(0, split))+        reader.push(stream.subarray(split))+        const parse = jest.spyOn(JSON, 'parse').mockImplementation((s) => s)+        let decoded: unknown[]+        try {+          decoded = await receiveAll(reader, payloads.length)+        } finally {+          parse.mockRestore()+        }+        expect({ split, decoded }).toEqual({ split, decoded: expected })+      }+    }++    it('multi-byte characters', async () => {+      await check(Buffer.from(JSON.stringify('aé€😀bé€😀'), 'utf8'))+    })++    it('invalid UTF-8 sequences', async () => {+      await check(+        Buffer.from([+          0x22, 0xff, 0x41, 0xe2, 0x82, 0x41, 0xf0, 0x9f, 0x98, 0x41, 0xc0,+          0xaf, 0xed, 0xa0, 0x80, 0xf4, 0x90, 0x80, 0x80, 0x22,+        ])+      )+    })++    it.each([+      ['2-byte', [0xc3]],+      ['3-byte', [0xe2, 0x82]],+      ['4-byte', [0xf0, 0x9f, 0x98]],+    ])(+      'a truncated %s sequence at the end of a frame',+      async (_name, truncated) => {+        await check(+          Buffer.from([0x22, 0x61, ...truncated]),+          Buffer.from('"b"', 'utf8')+        )+      }+    )+  })++  it('does not retain received chunks in queued packets', async () => {+    const first = frame('"first"')+    const second = frame('"second"')+    const reader = new PacketReader<unknown>()+    // One frame decoded from a single chunk, one spanning two chunks.+    const chunks = [first, second.subarray(0, 6), second.subarray(6)].map(+      (chunk) => Buffer.from(chunk)+    )+    for (const chunk of chunks) {+      reader.push(chunk)+    }+    for (const chunk of chunks) {+      chunk.fill(0x20)+    }+    expect(await receiveAll(reader, 2)).toEqual(['first', 'second'])+  })++  it('does not assemble fragmented payloads into a Buffer', async () => {+    const size = 4 * 1024 * 1024+    const payload = Buffer.alloc(size, 0x78)+    payload[0] = 0x22+    payload[size - 1] = 0x22+    const header = Buffer.alloc(4)+    header.writeUInt32BE(size, 0)+    const chunks = splitEvery(Buffer.concat([header, payload]), 64 * 1024)++    const copy = jest.spyOn(Buffer.prototype, 'copy')+    const concat = jest.spyOn(Buffer, 'concat')+    const allocUnsafe = jest.spyOn(Buffer, 'allocUnsafe')+    const reader = new PacketReader<string>()+    try {+      for (const chunk of chunks) {+        reader.push(chunk)+      }+      expect(copy).not.toHaveBeenCalled()+      expect(concat).not.toHaveBeenCalled()+      expect(allocUnsafe).not.toHaveBeenCalled()+    } finally {+      copy.mockRestore()+      concat.mockRestore()+      allocUnsafe.mockRestore()+    }+    expect(await reader.recv()).toBe('x'.repeat(size - 2))+  })+})

@@ -2,15 +2,7 @@ import { writeSync } from 'node:fs' import { createConnection } from 'node:net' import { Writable } from 'node:stream' import { structuredError } from '../error'--type State =-  | {-      type: 'waiting'-    }-  | {-      type: 'packet'-      length: number-    }+import { PacketReader } from './packet-reader'  export type Ipc<TIncoming, TOutgoing> = {   recv(): Promise<TIncoming>@@ -65,54 +57,12 @@ function createIpc<TIncoming, TOutgoing>(     },   }) -  const packetQueue: Buffer[] = []-  const recvPromiseResolveQueue: Array<(message: TIncoming) => void> = []+  const reader = new PacketReader<TIncoming>() -  function pushPacket(packet: Buffer) {-    const recvPromiseResolve = recvPromiseResolveQueue.shift()-    if (recvPromiseResolve != null) {-      recvPromiseResolve(JSON.parse(packet.toString('utf8')) as TIncoming)-    } else {-      packetQueue.push(packet)-    }-  }--  let state: State = { type: 'waiting' }-  let buffer: Buffer = Buffer.alloc(0)   socket.once('connect', () => {     socket.off('error', onConnectError)     socket.setNoDelay(true)-    socket.on('data', (chunk) => {-      buffer = Buffer.concat([buffer, chunk])--      loop: while (true) {-        switch (state.type) {-          case 'waiting': {-            if (buffer.length >= 4) {-              const length = buffer.readUInt32BE(0)-              buffer = buffer.subarray(4)-              state = { type: 'packet', length }-            } else {-              break loop-            }-            break-          }-          case 'packet': {-            if (buffer.length >= state.length) {-              const packet = buffer.subarray(0, state.length)-              buffer = buffer.subarray(state.length)-              state = { type: 'waiting' }-              pushPacket(packet)-            } else {-              break loop-            }-            break-          }-          default:-            invariant(state, (state) => `Unknown state type: ${state?.type}`)-        }-      }-    })+    socket.on('data', (chunk) => reader.push(chunk))   })   // When the socket is closed, this process is no longer needed.   // This might happen e. g. when parent process is killed or@@ -148,19 +98,8 @@ function createIpc<TIncoming, TOutgoing>(   }    return {-    async recv() {-      const packet = packetQueue.shift()-      if (packet != null) {-        return JSON.parse(packet.toString('utf8')) as TIncoming-      }--      const result = await new Promise<TIncoming>((resolve) => {-        recvPromiseResolveQueue.push((result) => {-          resolve(result)-        })-      })--      return result+    recv() {+      return reader.recv()     },      send(message: TOutgoing) {@@ -234,10 +173,3 @@ improveConsole('timeEnd', 'stdout', true) improveConsole('timeLog', 'stdout', true) improveConsole('timeStamp', 'stdout', true) improveConsole('assert', 'stderr', true)--/**- * Utility function to ensure all variants of an enum are handled.- */-function invariant(never: never, computeMessage: (arg: any) => string): never {-  throw new Error(`Invariant: ${computeMessage(never)}`)-}

@@ -0,0 +1,186 @@+import { TextDecoder } from 'node:util'++type State =+  | {+      type: 'waiting'+    }+  | {+      type: 'packet'+      length: number+    }++/**+ * Decodes length-prefixed packets (a u32 big-endian byte length followed by a+ * UTF-8 JSON payload) from a stream of chunks, and hands the decoded messages+ * to `recv()` callers in order.+ */+export class PacketReader<TIncoming> {+  private state: State = { type: 'waiting' }+  // Received chunks that have not been consumed yet. We keep them as a list+  // (instead of concatenating on every chunk) so that a packet that arrives in+  // many chunks can be decoded straight from them.+  private readonly chunks: Buffer[] = []+  private bufferedLength = 0+  // Decoded packets that no `recv()` call has consumed yet. Packets are+  // decoded as soon as they are complete so that queued packets don't retain+  // the received chunks (which may be views into larger Buffers).+  private readonly packetQueue: string[] = []+  // `recv()` calls waiting for the next packet.+  private readonly recvPromiseResolveQueue: Array<+    (message: TIncoming) => void+  > = []++  /**+   * Feeds a chunk of received bytes into the reader. Each packet it completes+   * is decoded for the oldest waiting `recv()` call, or queued until the next+   * `recv()` call if none is waiting.+   */+  push(chunk: Buffer): void {+    this.chunks.push(chunk)+    this.bufferedLength += chunk.length++    loop: while (true) {+      switch (this.state.type) {+        case 'waiting': {+          if (this.bufferedLength >= 4) {+            const length = this.takeBytes(4).readUInt32BE(0)+            this.state = { type: 'packet', length }+          } else {+            break loop+          }+          break+        }+        case 'packet': {+          if (this.bufferedLength >= this.state.length) {+            const packet = this.takeString(this.state.length)+            this.state = { type: 'waiting' }+            this.pushPacket(packet)+          } else {+            break loop+          }+          break+        }+        default:+          invariant(this.state, (state) => `Unknown state type: ${state?.type}`)+      }+    }+  }++  /**+   * Resolves with the next decoded message.+   */+  async recv(): Promise<TIncoming> {+    const packet = this.packetQueue.shift()+    if (packet != null) {+      return JSON.parse(packet) as TIncoming+    }++    return new Promise<TIncoming>((resolve) => {+      this.recvPromiseResolveQueue.push(resolve)+    })+  }++  private pushPacket(packet: string): void {+    const recvPromiseResolve = this.recvPromiseResolveQueue.shift()+    if (recvPromiseResolve != null) {+      recvPromiseResolve(JSON.parse(packet) as TIncoming)+    } else {+      this.packetQueue.push(packet)+    }+  }++  /**+   * Removes the first `length` bytes from `chunks` and returns them decoded as+   * UTF-8, without first assembling them into a single Buffer.+   * The caller must ensure that `bufferedLength >= length`.+   */+  private takeString(length: number): string {+    if (length === 0) {+      return ''+    }+    this.bufferedLength -= length+    const chunks = this.chunks+    const first = chunks[0]+    if (first.length >= length) {+      // Fast path: the bytes are contained in the first chunk, decode them in+      // place.+      const result = first.toString('utf8', 0, length)+      if (first.length === length) {+        chunks.shift()+      } else {+        chunks[0] = first.subarray(length)+      }+      return result+    }+    // The bytes span multiple chunks. A streaming decoder carries multi-byte+    // characters that are split across chunk boundaries.+    const decoder = new TextDecoder('utf-8')+    let result = ''+    let remaining = length+    let consumed = 0+    while (remaining > 0) {+      const chunk = chunks[consumed]+      if (chunk.length <= remaining) {+        result += decoder.decode(chunk, { stream: true })+        remaining -= chunk.length+        consumed+++      } else {+        result += decoder.decode(chunk.subarray(0, remaining), {+          stream: true,+        })+        chunks[consumed] = chunk.subarray(remaining)+        remaining = 0+      }+    }+    chunks.splice(0, consumed)+    return result + decoder.decode()+  }++  /**+   * Removes the first `length` bytes from `chunks` and returns them.+   * The caller must ensure that `bufferedLength >= length`.+   */+  private takeBytes(length: number): Buffer {+    if (length === 0) {+      return Buffer.alloc(0)+    }+    this.bufferedLength -= length+    const chunks = this.chunks+    const first = chunks[0]+    if (first.length === length) {+      // Fast path: the first chunk is exactly the requested bytes.+      chunks.shift()+      return first+    }+    if (first.length > length) {+      // The bytes are contained in the first chunk, no copy needed.+      chunks[0] = first.subarray(length)+      return first.subarray(0, length)+    }+    const result = Buffer.allocUnsafe(length)+    let offset = 0+    let consumed = 0+    while (offset < length) {+      const chunk = chunks[consumed]+      const remaining = length - offset+      if (chunk.length <= remaining) {+        chunk.copy(result, offset)+        offset += chunk.length+        consumed+++      } else {+        chunk.copy(result, offset, 0, remaining)+        chunks[consumed] = chunk.subarray(remaining)+        offset += remaining+      }+    }+    chunks.splice(0, consumed)+    return result+  }+}++/**+ * Utility function to ensure all variants of an enum are handled.+ */+function invariant(never: never, computeMessage: (arg: any) => string): never {+  throw new Error(`Invariant: ${computeMessage(never)}`)+}