diff --git a/.changeset/multipart-collect-linear.md b/.changeset/multipart-collect-linear.md new file mode 100644 index 00000000000..c28d408f5a9 --- /dev/null +++ b/.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. diff --git a/packages/effect/benchmark/http/multipart.ts b/packages/effect/benchmark/http/multipart.ts new file mode 100644 index 00000000000..22ef22378db --- /dev/null +++ b/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 => { + 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 = [] + for (let i = 0; i < payload.length; i += chunkSize) { + chunks.push(payload.subarray(i, i + chunkSize)) + } + return chunks +} + +const fieldsPayload = (fieldCount: number): Array => { + 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) => { + 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) => + 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) => + 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) => + 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()) diff --git a/packages/effect/src/Channel.ts b/packages/effect/src/Channel.ts index c4ee3591e84..a9d09d3c1b9 100644 --- a/packages/effect/src/Channel.ts +++ b/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 = ( + self: Channel, OutErr, OutDone, unknown, unknown, unknown, Env> +): Effect.Effect, OutErr, Env> => + Effect.map( + runFold( + self, + (): { + bytes: number + readonly arrays: Array + } => ({ + 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. * diff --git a/packages/effect/src/Stream.ts b/packages/effect/src/Stream.ts index 272b8a9e2cd..90231576640 100644 --- a/packages/effect/src/Stream.ts +++ b/packages/effect/src/Stream.ts @@ -10858,35 +10858,7 @@ export const mkString = (self: Stream): Effect.Effect(self: Stream): Effect.Effect => - Effect.map( - Channel.runFold( - self.channel, - (): { - bytes: number - readonly arrays: Array - } => ({ - 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 = (self: Stream): Effect.Effe * @since 4.0.0 */ export const mkUint8Array = (self: Stream): Effect.Effect => - Effect.map(mkArrayBuffer(self), (buffer) => new Uint8Array(buffer)) + Channel.mkUint8Array(self.channel) /** * Converts the stream to a `ReadableStream` using the provided services. diff --git a/packages/effect/src/unstable/http/Multipart.ts b/packages/effect/src/unstable/http/Multipart.ts index f147cd4f196..344efc4bf78 100644 --- a/packages/effect/src/unstable/http/Multipart.ts +++ b/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 = ( self: Channel.Channel, OE, OD, unknown, unknown, unknown, R> -): Effect.Effect, 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, OE, R> => Channel.mkUint8Array(self) /** * Persists a stream of multipart parts into a record. diff --git a/packages/effect/src/unstable/http/MultipartParser/internal/headers.ts b/packages/effect/src/unstable/http/MultipartParser/internal/headers.ts index 0e4a03a7a44..312d2dc4af5 100644 --- a/packages/effect/src/unstable/http/MultipartParser/internal/headers.ts +++ b/packages/effect/src/unstable/http/MultipartParser/internal/headers.ts @@ -12,394 +12,23 @@ type State = (typeof State)[keyof typeof State] const constContinue: Continue = { _tag: "Continue" } -const constNameChars = [ - 0, - 0, - 0, - 0, - 0, - 0, - 0, - 0, - 0, - 0, - 0, - 0, - 0, - 0, - 0, - 0, - 0, - 0, - 0, - 0, - 0, - 0, - 0, - 0, - 0, - 0, - 0, - 0, - 0, - 0, - 0, - 0, - 0, - 1, - 0, - 1, - 1, - 1, - 1, - 1, - 0, - 0, - 1, - 1, - 0, - 1, - 1, - 0, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 0, - 0, - 0, - 0, - 0, - 0, - 0, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 0, - 0, - 0, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 0, - 1, - 0, - 1 -] +// RFC 7230 token characters, allowed in header names +// The previous table stopped at byte 126; zero-filled entries for bytes 127-255 +// preserve its rejection behavior because callers accept only entries equal to 1. +const constNameChars = new Uint8Array(256) +for (const char of "!#$%&'*+-.^_`|~0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ") { + constNameChars[char.charCodeAt(0)] = 1 +} -const constValueChars = [ - 0, - 0, - 0, - 0, - 0, - 0, - 0, - 0, - 0, - 1, - 0, - 0, - 0, - 0, - 0, - 0, - 0, - 0, - 0, - 0, - 0, - 0, - 0, - 0, - 0, - 0, - 0, - 0, - 0, - 0, - 0, - 0, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 0, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1, - 1 -] +// HTAB, visible ASCII and obs-text, allowed in header values +const constValueChars = new Uint8Array(256) +constValueChars[9] = 1 +for (let i = 32; i <= 126; i++) { + constValueChars[i] = 1 +} +for (let i = 128; i <= 255; i++) { + constValueChars[i] = 1 +} export function make() { const decoder = new TextDecoder() diff --git a/packages/effect/test/Channel.test.ts b/packages/effect/test/Channel.test.ts index 9950043bcf0..2d160f81034 100644 --- a/packages/effect/test/Channel.test.ts +++ b/packages/effect/test/Channel.test.ts @@ -170,6 +170,20 @@ describe("Channel", () => { })) }) + describe("destructors", () => { + it.effect("mkUint8Array", () => + Effect.gen(function*() { + const bytes = yield* Channel.fromArray( + [ + [new Uint8Array([1, 2])], + [new Uint8Array([3]), new Uint8Array([4, 5])] + ] as const + ).pipe(Channel.mkUint8Array) + + assert.deepStrictEqual(bytes, new Uint8Array([1, 2, 3, 4, 5])) + })) + }) + describe("mapping", () => { it.effect("map", () => Effect.gen(function*() { diff --git a/packages/effect/test/unstable/http/Multipart.test.ts b/packages/effect/test/unstable/http/Multipart.test.ts index fd2d6f58d13..ba482f7701d 100644 --- a/packages/effect/test/unstable/http/Multipart.test.ts +++ b/packages/effect/test/unstable/http/Multipart.test.ts @@ -38,6 +38,34 @@ describe("Multipart", () => { ]) })) + it.effect("collects file content across pulls and a split trailing boundary", () => + Effect.gen(function*() { + const boundary = "----testboundary" + const encoder = new TextEncoder() + const boundarySplit = 8 + const chunks = [ + encoder.encode( + `--${boundary}\r\n` + + `Content-Disposition: form-data; name="file"; filename="file.txt"\r\n` + + `Content-Type: text/plain\r\n\r\n` + + "abc" + ), + encoder.encode(`def\r\n--${boundary.slice(0, boundarySplit)}`), + encoder.encode(`${boundary.slice(boundarySplit)}--\r\n`) + ] + + const contents = yield* Stream.fromArray(chunks).pipe( + Stream.rechunk(1), + Stream.pipeThroughChannel( + Multipart.makeChannel({ "content-type": `multipart/form-data; boundary=${boundary}` }) + ), + Stream.mapEffect((part) => part._tag === "File" ? part.contentEffect : Effect.die("expected file")), + Stream.runCollect + ) + + deepStrictEqual(contents, [encoder.encode("abcdef")]) + })) + it.effect("parses non-Latin-1 filenames", () => Effect.gen(function*() { const data = new globalThis.FormData()