ws-proto

README

ws-proto

A WebSocket transport for protobuf RPC services — the WebSocket analog of gRPC / Connect. One persistent connection multiplexes any number of concurrent RPC streams (unary, server-streaming, client-streaming, bidi) using a small protobuf framing envelope.

It ships two code generators and two runtimes:

Unit	Language	Analog	Provides
protoc-gen-go-ws	Go (protogen)	protoc-gen-go-grpc / connect-go	*_ws.pb.go: typed handler interface, typed client, stream wrappers, gRPC bridge
@gopherex/protoc-gen-ws-es	TypeScript (@bufbuild/protoplugin)	protoc-gen-connect-es	*_ws_pb.ts: typed service client
wsrpc (Go module)	Go	grpc-go / connect runtime	mux, framing, HTTP-upgrade server, dial client
@gopherex/ws-proto-transport	TypeScript	connect-web transport	mux, framing, browser/node client

The ES generator interoperates with protoc-gen-es v2 output (it imports the generated *_pb.ts message types and schemas).

---

Wire model

• One WebSocket connection per peer link, carrying binary frames.

• Each binary message is a marshaled Frame envelope:

  message Frame {
    uint32 stream_id = 1;             // client-allocated, monotonic odd ids
    Kind   kind      = 2;             // OPEN | MSG | HALF_CLOSE | END | RST | HEADER | WINDOW_UPDATE
    map<string,string> headers = 3;   // OPEN: metadata + deadline; END: trailers
    bytes  payload   = 4;             // MSG: one marshaled request/response message
    Status status    = 5;             // END only: gRPC-style code/message/details
    string method    = 6;             // OPEN only: "/pkg.Service/Method"
    uint32 window    = 8;             // WINDOW_UPDATE only: credit delta (bytes) returned to the sender
  }
  

• Streams are client-initiated, one stream per RPC. Flow: OPEN → MSG* → HALF_CLOSE → MSG* → END(status); RST aborts either side.

• The handshake negotiates the Sec-WebSocket-Protocol subprotocol wsrpc.v1.

• Per-RPC metadata and deadlines ride in-band on the OPEN frame. Connection-level concerns (auth, tenant routing) ride the HTTP Upgrade request headers, which proxies can inspect — see Deploying behind a proxy.

---

Flow control

Each stream has a per-stream credit window (default 256 KiB) for backpressure, symmetric in both directions. The receiver advertises credit; a sender's Send blocks (it never drops) until enough credit is available, so a well-behaved sender can't overrun a slow receiver:

• The receiver returns credit (a WINDOW_UPDATE frame carrying a byte delta) as it consumes messages — not merely as they arrive. That is what makes the window real backpressure.
• A single message larger than the whole window is still delivered: a MSG may be sent whenever the window is > 0, after which the window is allowed to go negative; subsequent sends then wait for credit. This matches gRPC/HTTP2 semantics and avoids a deadlock on an oversized message.
• The window assumes the same default on both peers with no handshake. Tune it with WithInitialWindow(n) (Go server), WithDialInitialWindow(n) (Go client), or initialWindow (TS WsTransportOptions); n <= 0 keeps the default. Mixing a windowed peer with a non-windowed one can stall a sender once the initial window is exhausted, so ship both sides together.

The bounded receive buffer (WithReceiveBuffer / maxReceiveQueue) remains a backstop: a misbehaving peer that ignores the window still can't grow memory without limit — its stream is reset with ResourceExhausted. A well-behaved peer never hits it.

---

Install

Go plugin + runtime

# the code generator (a protoc/easyp/buf plugin)
go install github.com/gopherex/ws-proto/cmd/protoc-gen-go-ws@latest

# the runtime, imported by generated code
go get github.com/gopherex/ws-proto/wsrpc

npm packages (GitHub Packages)

Both packages are published to GitHub Packages under the @gopherex scope. Point the scope at the GitHub registry and authenticate with a token that has read:packages (add to ~/.npmrc or the project .npmrc):

@gopherex:registry=https://npm.pkg.github.com
//npm.pkg.github.com/:_authToken=${GITHUB_TOKEN}

npm install @gopherex/ws-proto-transport
npm install -D @gopherex/protoc-gen-ws-es @bufbuild/protoc-gen-es @bufbuild/buf

---

Define your service

// echo/v1/echo.proto
syntax = "proto3";
package echo.v1;
option go_package = "example.com/gen/echo/v1;echov1";

service EchoService {
  rpc Unary(UnaryRequest) returns (UnaryResponse);
  rpc ServerStream(ServerStreamRequest) returns (stream ServerStreamResponse);
  rpc ClientStream(stream ClientStreamRequest) returns (ClientStreamResponse);
  rpc Bidi(stream BidiRequest) returns (stream BidiResponse);
}
// ... messages ...

No proto options are required; routes derive from /pkg.Service/Method.

---

Generate

Go (protoc-gen-go + protoc-gen-go-grpc + protoc-gen-go-ws)

Using easyp (see easyp.yaml):

generate:
  inputs:
    - directory: proto
  plugins:
    - { name: go,      out: gen, opts: { paths: source_relative } }
    - { name: go-grpc, out: gen, opts: { paths: source_relative } }   # only if you use the gRPC bridge
    - { name: go-ws,   out: gen, opts: { paths: source_relative } }

easyp generate   # emits *.pb.go, *_grpc.pb.go, *_ws.pb.go

TypeScript (protoc-gen-es + protoc-gen-ws-es)

buf.gen.yaml:

version: v2
plugins:
  - local: protoc-gen-es
    out: src/gen
    opt: [target=ts, import_extension=js]
  - local: protoc-gen-ws-es
    out: src/gen
    opt: [import_extension=js]
inputs:
  - directory: proto

npx buf generate   # emits *_pb.ts (messages) and *_ws_pb.ts (clients)

---

Use it

Go server

srv := wsrpc.NewServer(
    // Restrict browser origins (CSRF). Use your real frontend origins;
    // "*" disables the check — only safe if auth is enforced on the Upgrade request.
    wsrpc.WithOriginPatterns("app.example.com"),
    // Active keepalive pings keep the connection alive through proxy idle timeouts.
    wsrpc.WithKeepalive(20*time.Second, 10*time.Second),
    // Read auth/tenant info off the proxy-visible Upgrade request headers.
    wsrpc.WithConnContext(func(ctx context.Context, r *http.Request) context.Context {
        return context.WithValue(ctx, authKey{}, r.Header.Get("Authorization"))
    }),
)
echov1.RegisterEchoServiceHandler(srv, myHandler{})

http.Handle("/rpc", srv)          // srv is an http.Handler (a WebSocket upgrade endpoint)
log.Fatal(http.ListenAndServe(":8080", nil))

myHandler implements the generated EchoServiceHandler interface (Unary(ctx, *Req) (*Res, error), ServerStream(ctx, *Req, stream) error, …). You can also wrap an existing protoc-gen-go-grpc server with echov1.EchoServiceFromGRPC(grpcImpl).

Go client

cc, err := wsrpc.Dial(ctx, "ws://localhost:8080/rpc",
    wsrpc.WithHeader(http.Header{"Authorization": {"Bearer " + token}}),
)
defer cc.Close()

client := echov1.NewEchoServiceWSClient(cc)
res, err := client.Unary(ctx, &echov1.UnaryRequest{Name: "bob"})

TypeScript client (browser or node)

import { WsTransport } from "@gopherex/ws-proto-transport";
import { EchoServiceClient } from "./gen/echo_ws_pb.js";

const client = new EchoServiceClient(new WsTransport("wss://api.example.com/rpc"));

// unary
const res = await client.unary({ name: "bob" });

// server streaming
for await (const msg of client.serverStream({ count: 3 })) {
  console.log(msg.index);
}

Every method takes an optional CallOptions as its last argument:

const controller = new AbortController();

const res = await client.unary(
  { name: "bob" },
  {
    // request metadata -> headers on the opening frame
    headers: { authorization: "bearer …" },
    // cancellation: aborting sends RST; the call rejects with a WsStatusError
    signal: controller.signal,
    // optional leading response headers, sent before the first message
    onHeader: (header) => console.log(header["x-served-by"]),
    // response trailers, reported once the server ends the stream
    onTrailer: (trailer) => console.log(trailer["x-elapsed-ms"]),
  },
);

// later, to cancel an in-flight call:
controller.abort();

CallOptions (headers, signal, onHeader, onTrailer) is generated into each *_ws_pb.ts and accepted by all four RPC kinds. onHeader fires with the optional leading response metadata the server may send before the first message (or {} if none was sent); onTrailer fires with the END trailers after the response stream completes cleanly.

In Node, provide a WebSocket global (e.g. the ws package) before constructing the transport.

Auto-reconnect (opt-in)

By default a connection is a single socket: if it drops, every in-flight RPC fails and the connection stays down. You can opt into automatic reconnection of the connection (with exponential backoff + jitter) on both runtimes:

// Go: redial on connection loss. Backoff defaults to 100ms initial / 30s max.
cc, err := wsrpc.Dial(ctx, "ws://localhost:8080/rpc", wsrpc.WithReconnect())
// tune the schedule:
cc, err = wsrpc.Dial(ctx, "ws://localhost:8080/rpc",
    wsrpc.WithReconnect(wsrpc.WithBackoff(200*time.Millisecond, 10*time.Second)),
)

// TS: only WsTransport reconnects (it owns dialing); fromSocket() does not.
const transport = new WsTransport("wss://api.example.com/rpc", {
  reconnect: true,
  backoff: { initialMs: 200, maxMs: 10_000 }, // optional; defaults 100ms / 30s
});

There is no stream resume. When the connection drops, in-flight RPCs fail with codes.Unavailable (Go) / CODE_UNAVAILABLE (14) (TS) — message "connection lost" — and the caller must retry them; the server keeps no per-stream session state. New RPCs started after a drop wait for (or trigger) a reconnect and run on the fresh socket; RPCs opened during the gap buffer their frames and flush once the new socket opens. A deliberate Close() instead fails in-flight streams with codes.Canceled ("transport closed"), so an intentional shutdown stays distinguishable from a transport disconnect.

Without WithReconnect / { reconnect: true } the behavior is unchanged: a dropped connection fails everything with Unavailable and stays down.

---

Middleware & gRPC interceptors (Go)

Two complementary mechanisms add cross-cutting behavior on the server.

Native wsrpc middleware

wsrpc.WithMiddleware wraps every RPC (all four kinds) with a Middleware. The first listed runs outermost; a middleware can read the stream, short-circuit by returning an error, or wrap the call. Handler panics, middleware panics and interceptor panics are recovered and surface to the client as codes.Internal.

func Logging(next wsrpc.Handler) wsrpc.Handler {
    return func(ctx context.Context, s *wsrpc.Stream) error {
        start := time.Now()
        err := next(ctx, s)
        log.Printf("%s -> %v (%s)", s.Method(), wsrpc.FromError(err).Code, time.Since(start))
        return err
    }
}

func Auth(next wsrpc.Handler) wsrpc.Handler {
    return func(ctx context.Context, s *wsrpc.Stream) error {
        if s.Header()["x-token"] == "" {
            return wsrpc.Errorf(codes.Unauthenticated, "missing token")
        }
        return next(ctx, s)
    }
}

srv := wsrpc.NewServer(wsrpc.WithMiddleware(Logging, Auth)) // Logging outermost

Response headers & trailers (Go)

A server handler may send leading response metadata once, before the first message, with stream.SendHeader(map[string]string{...}), and trailers (flushed with the END frame) with stream.SetTrailer(map[string]string{...}):

func (impl) ServerStream(ctx context.Context, req *pb.Req, s *pb.Svc_StreamServerWS) error {
    _ = s.SendHeader(map[string]string{"x-served-by": "node-1"}) // before any Send
    defer s.SetTrailer(map[string]string{"x-elapsed-ms": "12"})
    return s.Send(&pb.Res{ /* … */ })
}

On the client, the leading headers are read with stream.Header() (returns {} if the server sent none) and the trailers with stream.Trailer() (populated once the stream ends). SendHeader after the first Send returns codes.FailedPrecondition.

Reusing existing gRPC interceptors via the bridge

When you serve an existing protoc-gen-go-grpc server through the bridge, you can run your real grpc.UnaryServerInterceptor / grpc.StreamServerInterceptor (auth, recovery, validation, otel, rate limiting, anything from go-grpc-middleware) — unchanged — over the WebSocket transport:

handler := echov1.EchoServiceFromGRPC(grpcImpl,
    wsrpc.WithUnaryInterceptor(grpc.ChainUnaryInterceptor(
        recovery.UnaryServerInterceptor(),
        auth.UnaryServerInterceptor(authFn),
        otelgrpc.UnaryServerInterceptor(),
    )),
    wsrpc.WithStreamInterceptor(grpc.ChainStreamInterceptor(
        recovery.StreamServerInterceptor(),
        auth.StreamServerInterceptor(authFn),
    )),
)
echov1.RegisterEchoServiceHandler(srv, handler)

The bridge mirrors grpc-go's own handler/interceptor flow (a generic grpc.ServerStream adapter wrapped in grpc.GenericServerStream[Req,Res]), so info.FullMethod is the real /pkg.Service/Method and interceptor chaining works as in a normal gRPC server. Streaming response metadata set on the grpc.ServerStream via SetHeader/SendHeader/SetTrailer is propagated: leading headers become a KIND_HEADER frame and trailers ride the END frame. Unary response metadata set via grpc.SetHeader/SendHeader/SetTrailer is propagated too: the unary path has no ServerStream, so the bridge installs a grpc.ServerTransportStream that forwards into a per-call metadata sink the generated registrar flushes after the handler returns (native unary handlers can set the same via wsrpc.SetHeader/wsrpc.SetTrailer). Limitation: a stream interceptor that wraps the ServerStream must delegate SendMsg/RecvMsg to the embedded stream (the idiomatic pattern).

---

Deploying behind a proxy

This is the part that breaks naive WebSocket deployments. Read it.

Transport facts to design around:

• One long-lived connection per client. Use wss:// (TLS) in production.
• Binary frames; negotiated subprotocol wsrpc.v1.
• The server sends keepalive pings every ~20s (WithKeepalive) so idle bidi/server-stream RPCs survive proxy idle timeouts. Browser clients cannot initiate WebSocket pings — they rely on the server's pings, so always keep server keepalive enabled.
• Default max message size is 16 MiB (WithReadLimit); keep it below the smallest message cap in your proxy chain (e.g. Cloudflare's 100 MB).
• Compression is off by default (payloads are already protobuf; permessage-deflate is occasionally mangled by intermediaries and costs memory). Enable per-message-deflate with wsrpc.WithCompression(...) on the server and wsrpc.WithDialCompression(...) on the client (e.g. wsrpc.CompressionContextTakeover). Browser clients negotiate it automatically when the server accepts it.
• The proxy → upstream hop must be HTTP/1.1 with Upgrade/Connection preserved. (coder/websocket uses the RFC 6455 HTTP/1.1 handshake, not RFC 8441 WebSocket-over-HTTP/2.)

Caddy

Works out of the box — reverse_proxy auto-detects the WebSocket upgrade and streams it with no idle timeout:

api.example.com {
    reverse_proxy localhost:8080
}

nginx

You must raise proxy_read_timeout above the keepalive interval and pass the upgrade headers:

location /rpc {
    proxy_pass http://upstream;
    proxy_http_version 1.1;
    proxy_set_header Upgrade $http_upgrade;
    proxy_set_header Connection "upgrade";
    proxy_set_header Host $host;
    proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
    proxy_set_header X-Forwarded-Proto $scheme;
    proxy_read_timeout 3600s;   # > server keepalive interval
    proxy_send_timeout 3600s;
    proxy_buffering off;
}

Cloudflare / AWS ALB / other LBs

• Cloudflare: WSS works on all plans; idle WS is dropped at ~100s and the message cap is 100 MB — the 20s keepalive covers the former, the 16 MiB read limit stays under the latter.
• AWS ALB: default idle timeout is 60s; the keepalive keeps it warm, but you may also raise the LB idle timeout above the ping interval.

Auth & origin behind a proxy

• Put connection-level auth (bearer token, cookie, tenant) on the Upgrade request headers — set them client-side with wsrpc.WithHeader(...) and read them server-side with wsrpc.WithConnContext(...). These are visible to proxies for routing/authorization; per-RPC metadata stays in-band and is not.
• Set wsrpc.WithOriginPatterns(...) to your real frontend origins. The default Origin check rejects cross-origin browser handshakes (correct CSRF behavior); "*" disables it and is only safe when auth is enforced on the Upgrade request.
• Trust X-Forwarded-For / X-Forwarded-Proto only when set by a proxy you control.

---

Compatibility & RFC notes

• RFC 6455 WebSocket: standard HTTP/1.1 Upgrade handshake (via coder/websocket); binary frames; Sec-WebSocket-Protocol: wsrpc.v1 negotiated; ping/pong control frames used for keepalive; close on graceful teardown. Per-RPC errors are carried in-band in the END/RST frame Status (not WebSocket close codes), because one socket is shared by many streams.
• RFC 8441 (WebSocket over HTTP/2) is not used; force HTTP/1.1 on the proxy→upstream hop.
• Read limits, keepalive interval/timeout, origin patterns, Upgrade headers and the gRPC bridge are all covered by tests, including a Go↔TypeScript cross-language integration test.

---

Development

make help          # list targets
make test          # gofmt + go vet + go test -race
make test-ts       # build + test the npm workspaces
make gen           # regenerate transport/*.pb.go
make gen-example   # rebuild the plugin and regenerate the golden example/
make release       # interactive, tag-driven release (Go module + npm share one version)

Pushing a vX.Y.Z tag (produced by make release) triggers .github/workflows/release.yml: it gates on CI, builds the protoc-gen-go-ws binaries for all platforms and attaches them to a GitHub Release, and publishes both npm packages to GitHub Packages.

---

License

MIT © yaroher