Byte stream inputs

An input of experimental_byteStream() is the request body itself. The client passes a ReadableStream<Uint8Array>, httpLink sends it alone as an application/octet-stream body, and the resolver gets an experimental_ByteStream: an Effect Stream<Uint8Array> that reads the body as the resolver pulls it, so nothing is buffered first.

import { function experimental_byteStream(opts?: experimental_ByteStreamOptions): experimental_ByteStreamSchemaexperimental_byteStream } from '@trpc12/trpc/server';
import { import EffectEffect, import StreamStream } from 'effect';

const const upload: Procedure<{
    type: "mutation";
    ctx: Context;
    meta: {};
    input: ReadableStream<Uint8Array<ArrayBufferLike>>;
    output: void;
    errors: experimental_ByteStreamError | SchemaValidationError<experimental_ByteStreamSchema>;
    services: never;
}>upload = const t: TrpcRoot<Context, {}, RootDef<Context, {}>>t.TrpcRoot<Context, {}, RootDef<Context, {}>>.procedure: ProcedureBuilder<RootDef<Context, {}>>procedure
  .ProcedureBuilder<RootDef<Context, {}>>.input<experimental_ByteStreamSchema>(schema: experimental_ByteStreamSchema): ProcedureBuilder<InputDef<RootDef<Context, {}>, experimental_ByteStreamSchema>> (+1 overload)input(function experimental_byteStream(opts?: experimental_ByteStreamOptions): experimental_ByteStreamSchemaexperimental_byteStream({ experimental_ByteStreamOptions.maxBytes?: number | undefinedmaxBytes: 100_000_000 }))
  .ProcedureBuilder<InputDef<RootDef<Context, {}>, experimental_ByteStreamSchema>>.mutation<Effect.Effect<void, experimental_ByteStreamError, never>>(resolver: (opts: ResolverOpts<InputDef<RootDef<Context, {}>, experimental_ByteStreamSchema>>) => Effect.Effect<void, experimental_ByteStreamError, never>): Procedure<{
    type: "mutation";
    ctx: Context;
    meta: {};
    input: ReadableStream<Uint8Array<ArrayBufferLike>>;
    output: void;
    errors: experimental_ByteStreamError | SchemaValidationError<...>;
    services: never;
}>mutation((opts: ResolverOpts<InputDef<RootDef<Context, {}>, experimental_ByteStreamSchema>>opts) =>
    import StreamStream.const runForEach: <Uint8Array<ArrayBufferLike>, experimental_ByteStreamError, never, void, never, never>(self: Stream.Stream<Uint8Array<ArrayBufferLike>, experimental_ByteStreamError, never>, f: (a: Uint8Array<ArrayBufferLike>) => Effect.Effect<void, never, never>) => Effect.Effect<void, experimental_ByteStreamError, never> (+1 overload)runForEach(opts: ResolverOpts<InputDef<RootDef<Context, {}>, experimental_ByteStreamSchema>>opts.ResolverOpts<InputDef<RootDef<Context, {}>, experimental_ByteStreamSchema>>.input: experimental_ByteStreaminput, (chunk: Uint8Array<ArrayBufferLike>chunk) => const storage: {
    append(chunk: Uint8Array): Effect.Effect<void>;
}storage.function append(chunk: Uint8Array): Effect.Effect<void>append(chunk: Uint8Array<ArrayBufferLike>chunk)),
  );

await const client: TrpcClient<{
    upload: Procedure<{
        type: "mutation";
        ctx: Context;
        meta: {};
        input: ReadableStream<Uint8Array<ArrayBufferLike>>;
        output: void;
        errors: experimental_ByteStreamError | SchemaValidationError<experimental_ByteStreamSchema>;
        services: never;
    }>;
}, {}, undefined, LinksTypes<readonly [TrpcLink<{
    form: JsonForm;
    provides: "response";
    builtinErrors: "NETWORK_ERROR" | "BAD_RESPONSE" | "STREAM_INTERRUPTED";
}, any>]>>client.upload: {
    mutate(input: ReadableStream<Uint8Array<ArrayBufferLike>>, opts?: CallOptions<{}> | undefined): TrpcPromise<undefined, CallError<{
        type: "mutation";
        ctx: Context;
        meta: {};
        input: ReadableStream<Uint8Array<ArrayBufferLike>>;
        output: void;
        errors: experimental_ByteStreamError | SchemaValidationError<experimental_ByteStreamSchema>;
        services: never;
    }, JsonForm, LinksTypes<...>>>;
}upload.function mutate(input: ReadableStream<Uint8Array<ArrayBufferLike>>, opts?: CallOptions<{}> | undefined): TrpcPromise<undefined, CallError<{
    type: "mutation";
    ctx: Context;
    meta: {};
    input: ReadableStream<Uint8Array<ArrayBufferLike>>;
    output: void;
    errors: experimental_ByteStreamError | SchemaValidationError<experimental_ByteStreamSchema>;
    services: never;
}, JsonForm, LinksTypes<...>>>mutate(const file: Filefile.Blob.stream(): ReadableStream<Uint8Array<ArrayBuffer>>stream());

Only the whole input can be a stream. Send what goes with it (a name, a type) in headers or in a call before it.

Reading the stream

maxBytes fails the stream with PAYLOAD_TOO_LARGE once more bytes than that have arrived. There's no limit by default: the handler's maxBodySize covers JSON and multipart bodies, not a stream, which is read only as fast as the resolver pulls it.
A body that can't be read (a connection reset) fails the stream with BAD_REQUEST. A client that goes away interrupts the call, as it does any other.
The body is read once. Reading the stream a second time fails it with BAD_REQUEST.
A resolver that stops early (Stream.take, a failure) still answers: on Node, the rest of the body is read and dropped rather than cut off, which would drop the connection before the response. A resolver that doesn't read the stream at all leaves the body to the server.

Both failures are in the procedure's error union, so the client can match PAYLOAD_TOO_LARGE as a typed error.

Streaming back

The output can stream while the body is still arriving: a mutation with .output(stream(Item)) can send progress for the chunks it has read.

import { function experimental_byteStream(opts?: experimental_ByteStreamOptions): experimental_ByteStreamSchemaexperimental_byteStream, function stream<TItem extends AnySchema, TReturn extends AnySchema | undefined = undefined>(item: TItem, returns?: TReturn): StreamSchema<TItem, TReturn>stream } from '@trpc12/trpc/server';
import { import SchemaSchema, import StreamStream } from 'effect';

const const progress: Procedure<{
    type: "mutation";
    ctx: Context;
    meta: {};
    input: ReadableStream<Uint8Array<ArrayBufferLike>>;
    output: AsyncIterable<number>;
    errors: SchemaValidationError<experimental_ByteStreamSchema>;
    services: never;
}>progress = const t: TrpcRoot<Context, {}, RootDef<Context, {}>>t.TrpcRoot<Context, {}, RootDef<Context, {}>>.procedure: ProcedureBuilder<RootDef<Context, {}>>procedure
  .ProcedureBuilder<RootDef<Context, {}>>.input<experimental_ByteStreamSchema>(schema: experimental_ByteStreamSchema): ProcedureBuilder<InputDef<RootDef<Context, {}>, experimental_ByteStreamSchema>> (+1 overload)input(function experimental_byteStream(opts?: experimental_ByteStreamOptions): experimental_ByteStreamSchemaexperimental_byteStream())
  .ProcedureBuilder<InputDef<RootDef<Context, {}>, experimental_ByteStreamSchema>>.output<StreamSchema<Schema.Number, undefined>>(schema: StreamSchema<Schema.Number, undefined>, opts?: {
    validate?: true;
}): ProcedureBuilder<With<InputDef<RootDef<Context, {}>, experimental_ByteStreamSchema>, {
    outputIn: StreamOutput<number, any>;
    outputOut: StreamOutput<number, any>;
    requires: never;
}>> (+1 overload)output(stream<Schema.Number, undefined>(item: Schema.Number, returns?: undefined): StreamSchema<Schema.Number, undefined>stream(import SchemaSchema.const Number: Schema.NumberNumber))
  .ProcedureBuilder<With<InputDef<RootDef<Context, {}>, experimental_ByteStreamSchema>, { outputIn: StreamOutput<number, any>; outputOut: StreamOutput<...>; requires: never; }>>.mutation<Stream.Stream<number, experimental_ByteStreamError, never>>(resolver: (opts: ResolverOpts<With<InputDef<RootDef<Context, {}>, experimental_ByteStreamSchema>, {
    outputIn: StreamOutput<number, any>;
    outputOut: StreamOutput<number, any>;
    requires: never;
}>>) => Stream.Stream<number, experimental_ByteStreamError, never> & (AsyncIterable<number> | Stream.Stream<...>)): Procedure<...>mutation((opts: ResolverOpts<With<InputDef<RootDef<Context, {}>, experimental_ByteStreamSchema>, {
    outputIn: StreamOutput<number, any>;
    outputOut: StreamOutput<number, any>;
    requires: never;
}>>opts) =>
    opts: ResolverOpts<With<InputDef<RootDef<Context, {}>, experimental_ByteStreamSchema>, {
    outputIn: StreamOutput<number, any>;
    outputOut: StreamOutput<number, any>;
    requires: never;
}>>opts.ResolverOpts<With<InputDef<RootDef<Context, {}>, experimental_ByteStreamSchema>, { outputIn: StreamOutput<number, any>; outputOut: StreamOutput<...>; requires: never; }>>.input: experimental_ByteStreaminput.Pipeable.pipe<experimental_ByteStream, Stream.Stream<number, experimental_ByteStreamError, never>>(this: experimental_ByteStream, ab: (_: experimental_ByteStream) => Stream.Stream<number, experimental_ByteStreamError, never>): Stream.Stream<number, experimental_ByteStreamError, never> (+21 overloads)pipe(
      import StreamStream.const mapAccum: <number, Uint8Array<ArrayBufferLike>, number>(initial: LazyArg<number>, f: (s: number, a: Uint8Array<ArrayBufferLike>) => readonly [state: number, values: readonly number[]], options?: {
    readonly onHalt?: ((state: number) => readonly number[]) | undefined;
} | undefined) => <E, R>(self: Stream.Stream<Uint8Array<ArrayBufferLike>, E, R>) => Stream.Stream<number, E, R> (+1 overload)mapAccum(
        () => 0,
        (read: numberread, chunk: Uint8Array<ArrayBufferLike>chunk) => {
          const const total: numbertotal = read: numberread + chunk: Uint8Array<ArrayBufferLike>chunk.Uint8Array<ArrayBufferLike>.byteLength: numberbyteLength;
          return [const total: numbertotal, [const total: numbertotal]] as type const = readonly [number, readonly [number]]const;
        },
      ),
    ),
  );

Node's fetch reads the response while it sends the body. Browsers send stream bodies only over HTTP/2 and later, read the response once the body is sent, and not all of them can yet; where fetch can't, the call fails before anything is sent. An input with a File in it, sent as multipart, works everywhere.

Links and handlers

httpLink sends the call alone, outside any batch, as a POST (queries too). It isn't resumed when the connection drops, and retryLink doesn't retry it by default, since the stream was read.
wsLink and the other socket links fail the call with BAD_REQUEST. localLink and createRouterClient pass the stream through as it is.
The handler takes an application/octet-stream body only for a procedure whose input is experimental_byteStream(), and only with the trpc-version header that httpLink sends, so a page on another site can't post one (UNSUPPORTED_MEDIA_TYPE otherwise). Any other input in it fails validation with BAD_REQUEST.
It streams on Node, Bun, Deno and fetch runtimes. AWS Lambda buffers event bodies, so the stream reads from memory, within Lambda's payload limit. Behind Express, a body express.raw() read is replayed; one that another parser read and dropped fails the call with INTERNAL_SERVER_ERROR.

experimental_byteStream() is an Effect Schema that decodes the ReadableStream to a Stream and encodes one back, so in-process calls decode it as the handler does.