serve streaming Connect RPCs natively

[hugowetterberg]

Summary

A service that never serves Twirp can implement connect-go's own handler interface rather than the plain protobuf one, and that is what lets it declare streaming methods: protoc-gen-elephant-rpc and protoc-gen-twirp both fail generation on a stream. Five things in the library were wrong for a streaming handler and invisible for a unary one, so none of them was caught by the dual-stack tests.

Message size limits are per message on a Connect mount. APIServer no longer wraps a Connect subtree's request body in an http.MaxBytesReader, which counts the bytes read over the life of the request — and for a client or bidirectional stream the request body is the stream, so such a stream would have died once its cumulative traffic passed 8 MiB. ServiceOptions gains MaxMessageBytes, defaulting to DefaultMaxBodyBytes, and MaxSendMessageBytes, which is off unless set. Every non-Connect mount keeps the stream-level limit, and the 413 for an oversized declared Content-Length stays on every path.

rpc.DrainInterceptor ends streaming handlers with unavailable at shutdown. http.Server.Shutdown does not cancel request contexts, so a streaming handler previously learned nothing about a deploy, held shutdown open until the timeout, and then had its connection closed underneath it — a truncated stream with no code, and no cleanup. APIServer starts the drain before Shutdown, and the wait is now APIServerShutdownTimeout rather than a hard-coded ten seconds.

Streams have their own metric series. rpc_stream_duration_seconds and rpc_streams_active; rpc_duration_seconds is unary only, since a subscription's lifetime in a histogram whose top bucket is thirty seconds lands in +Inf and drags every quantile computed over the service with it. No existing series was renamed or relabelled.

rpc.ContextWithTokenExpiry ends a stream with unauthenticated when the caller's token expires. A helper a handler calls rather than a default, since whether a subscription may outlive the authorization that opened it differs from stream to stream.

The request-scoped log metadata map is mutex-guarded, and GetLogMetadata returns a copy. A streaming handler that runs a producer goroutine alongside the one writing to the stream is the normal shape, and two of them calling SetLogMetadata is a concurrent map write.

Why MaxSendMessageBytes defaults to off

A read limit protects the service from its callers and replaces one that was already there. A send limit protects the caller from the service, and nothing has ever bounded what an elephant service may answer with — connect-go leaves it off, and the Twirp mount of a dual-stack service has no response limit at all. Defaulting it on would refuse responses the same service already serves, and refuse them unpredictably: connect-go checks the limit against the bytes on the wire, so a response is measured compressed when the caller accepts compression, and the identical call passes for one caller and is refused for another.

Testing

The fixture gains stream.proto with one method per stream type, generated by a second buf run with only protoc-gen-go and protoc-gen-connect-go — which is what a native service's generation looks like. Before this the streaming interceptor wrappers, written deliberately rather than with connect.UnaryInterceptorFunc, were covered by nothing at all.

Also

buf.yaml turns on the STANDARD lint rules, grandfathering rpc/errormeta.proto: elephantine.rpc.ErrorMeta is the error-detail type name on the wire in every error body in the fleet, so its package can never move. The internal fixture service was renamed Test to TestService to satisfy SERVICE_SUFFIX.

docs/connect.md now covers the native shape and streaming alongside the dual-stack material it already had.