Make multipart content collection linear

.changeset/multipart-collect-linear.md

@@ -0,0 +1,5 @@
+---
+"effect": patch
+---
+
+Add `Channel.mkUint8Array` and reuse it from `Stream` and multipart file collection, keeping streamed byte concatenation linear across upstream pulls.

packages/effect/benchmark/http/multipart.ts

@@ -0,0 +1,122 @@
+import { Effect, Stream } from "effect"
+import * as Multipart from "effect/unstable/http/Multipart"
+import * as MultipartParser from "effect/unstable/http/MultipartParser"
+import { Bench } from "tinybench"
+
+const bench = new Bench()
+
+const boundary = "----WebKitFormBoundary7MA4YWxkTrZu0gW"
+const headers = {
+  "content-type": `multipart/form-data; boundary=${boundary}`
+}
+const encoder = new TextEncoder()
+
+const filePayload = (fileSize: number, chunkSize: number): Array<Uint8Array> => {
+  const head = encoder.encode(
+    `--${boundary}\r\ncontent-disposition: form-data; name="file"; filename="file.bin"\r\ncontent-type: application/octet-stream\r\n\r\n`
+  )
+  const body = new Uint8Array(fileSize)
+  for (let i = 0; i < fileSize; i++) {
+    body[i] = i % 251
+  }
+  const tail = encoder.encode(`\r\n--${boundary}--\r\n`)
+  const payload = new Uint8Array(head.length + body.length + tail.length)
+  payload.set(head, 0)
+  payload.set(body, head.length)
+  payload.set(tail, head.length + body.length)
+  const chunks: Array<Uint8Array> = []
+  for (let i = 0; i < payload.length; i += chunkSize) {
+    chunks.push(payload.subarray(i, i + chunkSize))
+  }
+  return chunks
+}
+
+const fieldsPayload = (fieldCount: number): Array<Uint8Array> => {
+  let out = ""
+  for (let i = 0; i < fieldCount; i++) {
+    out += `--${boundary}\r\ncontent-disposition: form-data; name="field${i}"\r\n\r\nvalue of field number ${i}\r\n`
+  }
+  out += `--${boundary}--\r\n`
+  return [encoder.encode(out)]
+}
+
+const file16MiB64KiB = filePayload(16 * 1024 * 1024, 64 * 1024)
+const file16MiB4KiB = filePayload(16 * 1024 * 1024, 4 * 1024)
+const file1MiB64KiB = filePayload(1024 * 1024, 64 * 1024)
+const fields100 = fieldsPayload(100)
+
+const runParser = (chunks: Array<Uint8Array>) => {
+  const parser = MultipartParser.make({
+    headers,
+    onField() {},
+    onFile() {
+      return () => {}
+    },
+    onError(error) {
+      throw new Error(`unexpected error: ${error._tag}`)
+    },
+    onDone() {}
+  })
+  for (let i = 0; i < chunks.length; i++) {
+    parser.write(chunks[i])
+  }
+  parser.end()
+}
+
+const runChannelDrain = (chunks: Array<Uint8Array>) =>
+  Effect.runPromise(
+    Stream.fromArray(chunks).pipe(
+      Stream.pipeThroughChannel(Multipart.makeChannel(headers)),
+      Stream.mapEffect((part) => part._tag === "File" ? Stream.runDrain(part.content) : Effect.void),
+      Stream.runDrain
+    )
+  )
+
+// Stream.fromArray emits the chunks in one batch, so these are single-pull controls.
+const runChannelCollect = (chunks: Array<Uint8Array>) =>
+  Effect.runPromise(
+    Stream.fromArray(chunks).pipe(
+      Stream.pipeThroughChannel(Multipart.makeChannel(headers)),
+      Stream.mapEffect((part) => part._tag === "File" ? part.contentEffect : Effect.void),
+      Stream.runDrain
+    )
+  )
+
+// rechunk(1) forces one upstream pull per chunk, simulating network reads and
+// guarding against quadratic accumulation across pulls.
+const runChannelCollectStreaming = (chunks: Array<Uint8Array>) =>
+  Effect.runPromise(
+    Stream.fromArray(chunks).pipe(
+      Stream.rechunk(1),
+      Stream.pipeThroughChannel(Multipart.makeChannel(headers)),
+      Stream.mapEffect((part) => part._tag === "File" ? part.contentEffect : Effect.void),
+      Stream.runDrain
+    )
+  )
+
+bench
+  .add("parser: 16MiB file / 64KiB chunks", function() {
+    runParser(file16MiB64KiB)
+  })
+  .add("parser: 16MiB file / 4KiB chunks", function() {
+    runParser(file16MiB4KiB)
+  })
+  .add("parser: 100 small fields", function() {
+    runParser(fields100)
+  })
+  .add("channel drain: 16MiB file / 64KiB chunks", async function() {
+    await runChannelDrain(file16MiB64KiB)
+  })
+  .add("channel collect: 16MiB file / 64KiB chunks", async function() {
+    await runChannelCollect(file16MiB64KiB)
+  })
+  .add("channel collect: 1MiB file / 64KiB chunks", async function() {
+    await runChannelCollect(file1MiB64KiB)
+  })
+  .add("channel collect streaming: 16MiB file / 64KiB chunks", async function() {
+    await runChannelCollectStreaming(file16MiB64KiB)
+  })
+
+await bench.run()
+
+console.table(bench.table())

packages/effect/src/Channel.ts

@@ -7965,6 +7965,64 @@ export const runForEachWhile: {
       ))
 )
 
+/**
+ * Concatenates a channel's `Uint8Array` chunks into a single `Uint8Array`.
+ *
+ * **Example** (Joining channel byte chunks)
+ *
+ * ```ts import.meta.vitest
+ * import { Channel, Effect } from "effect"
+ *
+ * const channel = Channel.fromArray([
+ *   [new Uint8Array([1, 2])],
+ *   [new Uint8Array([3, 4])]
+ * ] as const)
+ *
+ * const bytes = Effect.runSync(Channel.mkUint8Array(channel))
+ * Array.from(bytes) // => [1, 2, 3, 4]
+ * ```
+ *
+ * **Gotchas**
+ *
+ * This materializes the full content in memory. The source channel must not
+ * reuse or mutate emitted buffers, which are retained until collection completes.
+ *
+ * @category running
+ * @since 4.0.0
+ */
+export const mkUint8Array = <OutErr, OutDone, Env>(
+  self: Channel<Arr.NonEmptyReadonlyArray<Uint8Array>, OutErr, OutDone, unknown, unknown, unknown, Env>
+): Effect.Effect<Uint8Array<ArrayBuffer>, OutErr, Env> =>
+  Effect.map(
+    runFold(
+      self,
+      (): {
+        bytes: number
+        readonly arrays: Array<Uint8Array>
+      } => ({
+        bytes: 0,
+        arrays: []
+      }),
+      (acc, chunk) => {
+        for (let i = 0; i < chunk.length; i++) {
+          acc.bytes += chunk[i].length
+          acc.arrays.push(chunk[i])
+        }
+        return acc
+      }
+    ),
+    ({ arrays, bytes }) => {
+      const result = new Uint8Array(bytes)
+      let offset = 0
+      for (let i = 0; i < arrays.length; i++) {
+        const array = arrays[i]
+        result.set(array, offset)
+        offset += array.length
+      }
+      return result
+    }
+  )
+
 /**
  * Runs a channel and collects all output elements into an array.
  *

packages/effect/src/Stream.ts

@@ -10858,35 +10858,7 @@ export const mkString = <E, R>(self: Stream<string, E, R>): Effect.Effect<string
  * @since 4.0.0
  */
 export const mkArrayBuffer = <E, R>(self: Stream<Uint8Array, E, R>): Effect.Effect<ArrayBuffer, E, R> =>
-  Effect.map(
-    Channel.runFold(
-      self.channel,
-      (): {
-        bytes: number
-        readonly arrays: Array<Uint8Array>
-      } => ({
-        bytes: 0,
-        arrays: []
-      }),
-      (acc, chunk) => {
-        for (let i = 0; i < chunk.length; i++) {
-          acc.bytes += chunk[i].length
-          acc.arrays.push(chunk[i])
-        }
-        return acc
-      }
-    ),
-    ({ arrays, bytes }) => {
-      const result = new Uint8Array(bytes)
-      let offset = 0
-      for (let i = 0; i < arrays.length; i++) {
-        const array = arrays[i]
-        result.set(array, offset)
-        offset += array.length
-      }
-      return result.buffer
-    }
-  )
+  Effect.map(Channel.mkUint8Array(self.channel), (bytes) => bytes.buffer)
 
 /**
  * Concatenates the stream's `Uint8Array` chunks into a single `Uint8Array`.
@@ -10909,7 +10881,7 @@ export const mkArrayBuffer = <E, R>(self: Stream<Uint8Array, E, R>): Effect.Effe
  * @since 4.0.0
  */
 export const mkUint8Array = <E, R>(self: Stream<Uint8Array, E, R>): Effect.Effect<Uint8Array, E, R> =>
-  Effect.map(mkArrayBuffer(self), (buffer) => new Uint8Array(buffer))
+  Channel.mkUint8Array(self.channel)
 
 /**
  * Converts the stream to a `ReadableStream` using the provided services.

packages/effect/src/unstable/http/Multipart.ts

@@ -602,7 +602,7 @@ class FileImpl extends PartBase implements File {
     this.contentType = info.contentType
     this.content = Stream.fromChannel(channel)
     this.contentEffect = channel.pipe(
-      collectUint8Array,
+      Channel.mkUint8Array,
       Effect.mapError((cause) => MultipartError.fromReason("InternalError", cause))
     )
   }
@@ -635,24 +635,15 @@ const defaultWriteFile = (path: string, file: File) =>
  * **Gotchas**
  *
  * This materializes the full content in memory.
+ * The source channel must not reuse or mutate emitted buffers, which are retained
+ * until collection completes.
  *
  * @category converting
  * @since 4.0.0
  */
 export const collectUint8Array = <OE, OD, R>(
   self: Channel.Channel<Arr.NonEmptyReadonlyArray<Uint8Array>, OE, OD, unknown, unknown, unknown, R>
-): Effect.Effect<Uint8Array<ArrayBuffer>, OE, R> =>
-  Channel.runFold(self, constant(new Uint8Array(0)), (accumulator, chunk) => {
-    const totalLength = chunk.reduce((sum, element) => sum + element.length, accumulator.length)
-    const newAccumulator = new Uint8Array(totalLength)
-    newAccumulator.set(accumulator, 0)
-    let offset = accumulator.length
-    for (const element of chunk) {
-      newAccumulator.set(element, offset)
-      offset += element.length
-    }
-    return newAccumulator
-  })
+): Effect.Effect<Uint8Array<ArrayBuffer>, OE, R> => Channel.mkUint8Array(self)
 
 /**
  * Persists a stream of multipart parts into a record.