Serve handler from cloudflare workers

[robbertvanginkel]

Is your feature request related to a problem? Please describe. I want to use connect-es to serve a connect/grpc-web handler from a cloudflare worker, filing an issue as suggested in https://github.com/bufbuild/connect-es#other-platforms.

Describe the solution you'd like Similar to the adaptor that exists for node, something that I can plug into the worker's https://developers.cloudflare.com/workers/runtime-apis/fetch-event/. It would only need to support the connect & grpc-web protocols, as it seems that the workers runtime does not expose enough to implement the gRPC protocol.

Describe alternatives you've considered Not using connect-es and manually writing the grpc-web/connect protocol output.

Additional context I've gotten some basics to work with the following snippet:

import { createConnectRouter } from "@bufbuild/connect";
import { UniversalHandler } from "@bufbuild/connect/protocol";
import { ElizaService } from "./gen/buf/connect/demo/eliza/v1/eliza_connect";

export default {
    async fetch(request: Request, env: Env, ctx: ExecutionContext ): Promise<Response> {
        const url = new URL(request.url);
        let router = createConnectRouter({ grpc: false, grpcWeb: true, connect: true})
        router.service(ElizaService, {
            async say(req) {
                return {sentence: `You said ${req.sentence}`};
            }
        })
        const paths = new Map<string, UniversalHandler>();
        for (const uHandler of router.handlers) {
            paths.set(uHandler.requestPath, uHandler);
        }
        const uHandler = paths.get(url.pathname ?? "");
        if (!uHandler) {
            return new Response(undefined, { status: 404 });
        }

        let response = await uHandler({
            url: url,
            header: request.headers,
            httpVersion: "2",
            method: request.method,
            body: request.body,
        })
        return new Response(
            readableStreamFromIterable(wrapIterable(response.body ?? new Uint8Array())),
            { status: response.status, headers: response.header }
        )
    },
};

function wrapIterable(x: AsyncIterable<Uint8Array> | Uint8Array): AsyncIterable<Uint8Array> {
    if (x instanceof Uint8Array) {
        return { async *[Symbol.asyncIterator]() { yield x } }
    }
    return x;
}

function readableStreamFromIterable<T>(iterable: Iterable<T> | AsyncIterable<T>): ReadableStream<T> {
    // kindly borrowing from https://github.com/denoland/deno_std/blob/2a60184084ae98ab18960209fb1e79ba1a815bfa/streams/readable_stream_from_iterable.ts#L40
}

However, when running this on cloudflare (either production or using wrangler dev --experimental-local) I'm met with the following error:

service core:user:: Uncaught Error: Some functionality, such as asynchronous I/O, timeouts, and generating random values, can only be performed while handling a request.
  at index.js:24:19

where the stack trace points to the following part of connect-es: https://github.com/bufbuild/connect-es/blob/c18d298d026a769b4075aac3d03f78bcb6b60e11/packages/connect/src/protocol/universal-handler.ts#L142

The cloudflare docs at https://developers.cloudflare.com/workers/runtime-apis/web-standards#abortcontroller-and-abortsignal don't seem to indicate any limits on where new AbortController can be used, but when I manually modify my bundled output to inline that toplevel variable into the place it's used the above snippet works.

I'm sure there's more things problematic about the snippet: the way to convert UniversalHandler's response.body's type into a Response's body param seems off and creating the router each request is probably not ideal. I'm new to typescript so I'm still figuring things out, it'd be great if connect-es could support workers out of the box!

[timostamm]

Hey Robbert :wave:

The latest release v0.8.5 includes a fix for the AbortController issue. Let me know if that works for you.

Yes, creating the entire router for each request does not seem ideal. I think it should be possible to move the first 11 lines from the fetch function body up, so that they are above the export default statement.

For converting request / response types, we have something in the works. Here is an early version: https://gist.github.com/timostamm/5ca423155a0ddf03678ddc61f08cf4bd

[robbertvanginkel]

Tried the latest version and the AbortController issue is indeed fixed! Thanks for that.

Yes, creating the entire router for each request does not seem ideal. I think it should be possible to move the first 11 lines from the fetch function body up, so that they are above the export default statement.

This does work, but I haven't figured out how to pass fetch's env and ctx parameters which are needed to interact with the bindings to r2/kv etc from the handlers. According to https://developers.cloudflare.com/workers/runtime-apis/fetch-event/#parameters these (could) vary on each call, so they need to be available in the hander somehow. I looked at passing something through HandlerContext or when calling UniversalHandler, but did not see a way to do so.

If something like the convert request/response gist eventually ends up in connect-es that'd be neat!

[timostamm]

Robbert, the (improved) code from the request/response gist landed in https://github.com/bufbuild/connect-es/pull/575. Unfortunately, it uncovered an underlying issue: Our handlers return the response before the implementation had a chance to populate the response headers. We need to address this in our handlers before this is really usable.

For passing around env and ctx parameters, it looks like you need something similar to https://github.com/bufbuild/connect-es/issues/586?

[robbertvanginkel]

I'm not familiar with express but based on my reading of https://expressjs.com/en/api.html#res.locals it looks like something I could have reached to context for in Go. Anything similar in connect-es that lets me pass a value into a handler (ideally typesafe without any) would work.

[timdp]

Just familiarizing myself with Connect in the hope that I'll be able to contribute, so please bear with me.

Our handlers return the response before the implementation had a chance to populate the response headers. We need to address this in our handlers before this is really usable.

@timostamm Are you saying that when we convert a universal response to a fetch response, we do so before the headers on the universal response have been populated? If so, is the aim to eventually stream the headers and body once they become available?

And if that's the case, could we perhaps focus on an implementation that receives the finalized universal response (post-streaming), and synchronously converts it to a fetch response? Presumably, that would already support a lot of implementations.

Again, sorry if I'm way off.

[timostamm]

Hey @timdp 👋

Are you saying that when we convert a universal response to a fetch response, we do so before the headers on the universal response have been populated?

That is correct. Consider this server-streaming RPC implementation, which sets a response header before it streams two response messages:

async function* streamingRpc(req, ctx) {
  ctx.responseHeaders.set("Foo", "Bar") // should set response headers
  await delay(150);
  yield new Resp(); // sends some bytes in the response body
  await delay(150);
  yield new Resp(); // sends some bytes in the response body
}

The handler (the function that processes this RPC) returns a Promise<UniversalServerResponse>. With some details elided it is:

interface UniversalServerResponse {
  status: number;
  header: Headers;
  body: AsyncIterable<Uint8Array>;
}

We produced this object too early, before the first line in the implementation block had a chance to set headers. We fixed the issue in https://github.com/bufbuild/connect-es/pull/588, by waiting for the first bytes of the response body before resolving.

To really support Cloudflare Workers and the Vercel Edge runtime, we need automated test coverage. Both Vercel and Cloudflare offer tooling to run in a simulated environment, which is useful to surface some issues. I don't think we can avoid running tests on the actual platform though. Happy to bounce some thoughts how to tackle this 🙂

[timdp]

Thanks for clarifying, and great to read this in #588:

The fix is to wait until the first response body byte is produced before resolving the response.

because that would have been my suggestion too. 😁

So it sounds like the current status is:

• There are currently no known blockers.
• Therefore we should be able to run a Connect router on Workers (essentially using #575).
• However, as there are no integration tests using e.g., Wrangler/Miniflare, calling it "supported" would be a stretch.

Is this accurate?

I do have quite a bit of experience developing Workers and running them at scale, although I'm just jumping into Connect. I'd be more than willing to have a chat about how I (or our team) can help connect those dots—no pun intended.

[marekbuild]

Hi @timdp, are you on our Public Slack? Trying to figure out a good way to chat...

[timdp]

I am now. Come find me!

[robbertvanginkel]

@timostamm any thoughts on how env/ctx could be passed to a UniversalHandler and used in a user provided implementation? I haven't been able to find a way to do this other than creating the handlers on each request and accessing them from the closure as in the example, but that's not scaling so well to having many rpcs.

[timdp]

@robbertvanginkel I think this might be a good use case for the emerging AsyncLocalStorage API that's already supported in Workers today.

[marekbuild]

Related to the question of env/ctx is https://github.com/bufbuild/connect-es/issues/527

[srikrsna-buf]

Just a small update, with the release of v1.1.0 we now have the ability to pass custom values to handlers. The release also exports helpful functions to interact with the fetch API.

Here's a simple cloudflare worker example for handlers: https://github.com/connectrpc/examples-es/pull/1068. This demonstrates using KV to implement a service as a cloudflare worker.

See https://github.com/connectrpc/connect-es/discussions/829 for the high level plan for edge runtimes.