Undefined behaviour due to error handing in xrpc-server

[matthieusieben]

The following lines in xrpc-server can cause an undefined behavior:

atproto/packages/xrpc-server/src/server.ts

Lines 240 to 243 in 30b05a7

	if (input?.body instanceof Readable) {
	// If the body stream errors at any time, abort the request
	input.body.once('error', next)
	}

The reason is that the next() function might get called multiple times. These lines also prevent the handler to catch and handle the error itself:

server.method(
  'io.example.blobTest',
  async (ctx: { input?: xrpcServer.HandlerInput }) => {
    if (!(ctx.input?.body instanceof Readable))
      throw new Error('Input not readable')

    try {
      const buffers: Buffer[] = []
      for await (const data of ctx.input.body) {
        buffers.push(data)
      }
      return {
        encoding: 'json',
        body: { data: Buffer.concat(buffers).toString('base64') },
      }
    } catch (er) {
      return {
        encoding: 'json',
        body: { error: er.message },
      }
    }
  }
)

In the previous example, an error in the body stream can either result in the handler's payload being sent ({ error: 'some message' }) OR result in Express' finalhandler sending a generir error response. Whichever occurs first depends on the order of async tasks which is not predictable by the user of xrpc-server.

IMHO xrpc-server should avoid this behavior and require that the handler handles any potential error when the body is a stream. For convenience, RouteOpts could be used in order to describe how the payload should be processed (e.g. allow defining acceptable content types).