From cce2559760e3d3ba8189766987138d5e4461eab0 Mon Sep 17 00:00:00 2001 From: Antoine Toussaint <312536060+antoinetoussaint-byte@users.noreply.github.com> Date: Sat, 26 Sep 2026 18:30:51 -0400 Subject: [PATCH] feat: pass a consumed module's server-streaming methods through to the page MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The passthrough refused every streaming method ("only unary methods pass through"), so a page could not follow a module's live state. A server-streaming method is now served as a Connect server-streaming procedure at the same /modules/// path, under exactly the unary rules: - the authority minted for that method (or the bearer alone, for a ViewerBearer module); - the declared response fields applied to every streamed message. Both bounds are declared per method: MaxStreamDuration (default 5 min, at most 30 min; the stream then ends with deadline_exceeded) and MaxStreamMessageBytes (default 1 MiB; a larger message ends the stream with resource_exhausted and is never truncated). The page disconnecting cancels the module request. Client-streaming methods, and stream bounds on a unary method or out of range, are refused at boot. A module answers a streamed method over its google.api.http binding with grpc-gateway's server-streaming wire: newline-delimited JSON, {"result": …} or {"error": {"code", "message"}} per line (application/x-ndjson, or application/json). Gateway.TranscodedStream is the same call for a solution handler. The interface artifact marks a streamed operation (x-streaming: server). Co-Authored-By: Claude Opus 5.5 (1M context) --- README.md | 44 +++++- operations.go | 7 + passthrough.go | 109 ++++++++++---- stream.go | 310 ++++++++++++++++++++++++++++++++++++++++ stream_test.go | 353 ++++++++++++++++++++++++++++++++++++++++++++++ transcode_test.go | 6 + 6 files changed, 803 insertions(+), 26 deletions(-) create mode 100644 stream.go create mode 100644 stream_test.go diff --git a/README.md b/README.md index ca4fc5e..ff361c4 100644 --- a/README.md +++ b/README.md @@ -456,7 +456,7 @@ module's, and `ViewerBearer` takes none at either level. Least privilege by default. A method that is not declared is `not_found`. At boot, `Serve` refuses a module whose `as` is not in the solution's api.consumes, a method missing from the registry or with no binding under `/v1/`, a -streaming method, a method with no declared response fields, a method with no +client-streaming method, stream bounds on a unary method or out of range, a method with no declared response fields, a method with no scopes of its own or its module's (unless the module is `ViewerBearer`), scopes on a `ViewerBearer` module or method, and a scope with no resource kind, no action, or a blank or repeated action. `Response` @@ -473,6 +473,48 @@ artifact, and `InterfaceArtifact(dir, info, ops, modules...)` renders a backend's whole artifact — its own operations and the passthrough — at the version its `service.codefly.yaml` declares. +#### Streamed methods + +A server-streaming method passes through too, as a Connect server-streaming +procedure at the same path, so connect-es's generated client consumes it +unmodified. It takes the same authority as a unary call — the viewer's Work +Context minted for that method, or the bearer alone for a `ViewerBearer` +module — and the declared response fields (`Response`, or `WholeResponse`) +apply to **every** message of the stream. + +Every stream is bounded, and both bounds are declared per method: + +| Field | Default | Refused at boot | +| --- | --- | --- | +| `MaxStreamDuration` | `DefaultStreamDuration` (5 min) | negative, or over `MaxStreamDurationLimit` (30 min) | +| `MaxStreamMessageBytes` | `DefaultStreamMessageLimit` (1 MiB) | negative | + +A stream that reaches its duration ends with `deadline_exceeded`; a message +over its bound ends it with `resource_exhausted` and is never truncated. For a +streaming method `Timeout` bounds only the mint and the module's first answer +(status line and headers), and `MaxResponseBytes` is refused (bound each +message instead). The page disconnecting cancels the module request at once. + +**The wire a module answers a streamed method with** is grpc-gateway's for a +server-streaming method, over the method's `google.api.http` binding: a 2xx +response with content type `application/x-ndjson` (`application/json` is +accepted too), one JSON object per line, each flushed as it is ready: + +``` +{"result": } +{"error": {"code": , "message": ""}} +``` + +Blank lines are ignored, so a module may send them to keep an idle connection +open. An error line ends the stream with that code and message; the end of the +body ends it normally. A non-2xx answer before the stream starts is the +module's refusal, relayed like a unary one. `Gateway.TranscodedStream` is the +same call for a handler of the solution's own. + +In the interface artifact a streamed operation says so in its behaviour and +carries `x-streaming: server` (Swagger 2 has no stream; its 200 schema is one +message of it). + ### Generated messages in a response A handler's value is encoded with `encoding/json`, which reads a generated diff --git a/operations.go b/operations.go index c552dfe..1a1a326 100644 --- a/operations.go +++ b/operations.go @@ -48,6 +48,9 @@ type Operation struct { // the artifact's schema is derived from its type. A generated protobuf // message is carried as Message[T] and described from its descriptor. Response any + // Streaming marks an operation answered as a stream of Response messages + // (a Connect server-streaming procedure) rather than one of them. + Streaming bool // Exactly one of Handler and RequestHandler answers the route. Handler Handler RequestHandler RequestHandler @@ -127,6 +130,10 @@ func swaggerDocument(info InterfaceInfo, ops []Operation) (map[string]any, error "tags": []string{info.Tag}, "responses": operationResponses(op, definitions), } + if op.Streaming { + // Swagger 2 has no stream: the 200 schema is one message of it. + endpoint["x-streaming"] = "server" + } parameters := []any{} if op.Request != nil { parameters = append(parameters, map[string]any{"name": "body", "in": "body", "required": true, "schema": valueSchema(op.Request, definitions)}) diff --git a/passthrough.go b/passthrough.go index 0bb2889..4f35f20 100644 --- a/passthrough.go +++ b/passthrough.go @@ -37,7 +37,9 @@ import ( // /modules/// // // so the module's own generated client (connect-es, connect-go) is used -// unmodified with the base URL `/modules/`. Each call is answered +// unmodified with the base URL `/modules/`. A server-streaming +// method is served the same way as a Connect server-streaming procedure; see +// "Streamed methods" in stream.go for the wire a module answers it with. Each call is answered // by the module, as the viewer: the runtime mints the viewer's Work Context for // the module (Gateway.ForModule) — or forwards only the bearer, for a module // that authenticates the viewer itself — and forwards the request over the @@ -105,8 +107,22 @@ type ConsumedMethod struct { // DefaultTranscodedResponseLimit. MaxResponseBytes int64 // Timeout bounds one call, the mint included; zero is - // DefaultPassthroughTimeout. + // DefaultPassthroughTimeout. For a server-streaming method it bounds the + // mint and the module's first answer (its status line and headers); the + // stream itself is bounded by MaxStreamDuration. Timeout time.Duration + // MaxStreamDuration bounds how long one stream of a server-streaming + // method stays open, from the call to its last message; zero is + // DefaultStreamDuration, and no stream may be declared longer than + // MaxStreamDurationLimit. A stream that reaches it ends with + // deadline_exceeded. Only for a server-streaming method. + MaxStreamDuration time.Duration + // MaxStreamMessageBytes bounds each message of a server-streaming method's + // stream, as the module sends it (one line of its answer); zero is + // DefaultStreamMessageLimit. A larger message ends the stream with + // resource_exhausted: it is never truncated. Only for a server-streaming + // method. + MaxStreamMessageBytes int64 } const ( @@ -168,8 +184,11 @@ func resolvePassthrough(modules []ConsumedModule) (map[string]passthroughRoute, if err != nil { return nil, fmt.Errorf("consumed module %q: %w", module.As, err) } - if md.IsStreamingClient() || md.IsStreamingServer() { - return nil, fmt.Errorf("consumed module %q: %s streams; only unary methods pass through", module.As, md.FullName()) + if md.IsStreamingClient() { + return nil, fmt.Errorf("consumed module %q: %s streams its requests; only unary and server-streaming methods pass through", module.As, md.FullName()) + } + if err := checkStreamBounds(md, method); err != nil { + return nil, fmt.Errorf("consumed module %q: %w", module.As, err) } _, verb, template, err := bindingUnder(md, prefix) if err != nil { @@ -310,6 +329,16 @@ func (s *Server) passthroughRoute(route passthroughRoute) http.Handler { limit = DefaultPassthroughRequestLimit } procedure := "/" + string(md.Parent().FullName()) + "/" + string(md.Name()) + if md.IsStreamingServer() { + handler := connect.NewServerStreamHandler(procedure, func(ctx context.Context, req *connect.Request[dynamicpb.Message], stream *connect.ServerStream[dynamicpb.Message]) error { + return s.forwardStream(ctx, route, req, stream) + }, + connect.WithSchema(md), + connect.WithRequestInitializer(initialize), + connect.WithReadMaxBytes(limit), + ) + return http.StripPrefix(PassthroughPathPrefix+route.module.As, handler) + } handler := connect.NewUnaryHandler(procedure, func(ctx context.Context, req *connect.Request[dynamicpb.Message]) (*connect.Response[dynamicpb.Message], error) { resp, err := s.forward(ctx, route, req) if err != nil { @@ -338,7 +367,29 @@ func (s *Server) forward(ctx context.Context, route passthroughRoute, req *conne } ctx, cancel := context.WithTimeout(ctx, timeout) defer cancel() - gw := newGateway(s.cfg.gatewayURL, bearer, req.Header().Get(orgHeader), req.Header().Get(sessionHeader)) + gw, err := s.authorize(ctx, route, req.Header(), req.Msg) + if err != nil { + return nil, err + } + out := dynamicpb.NewMessage(route.md.Output()) + var opts []TranscodeOption + if route.method.MaxResponseBytes > 0 { + opts = append(opts, MaxResponseBytes(route.method.MaxResponseBytes)) + } + if err := gw.Transcoded(ctx, "/v1/"+route.module.As, route.method.Name, req.Msg, out, opts...); err != nil { + return nil, relayedError(err) + } + if !route.method.WholeResponse { + out = Apply(route.method.Response, out) + } + return out, nil +} + +// authorize prepares one call as the viewer: the declared authority minted for +// this method (or the bearer alone, for a ViewerBearer module), and the +// declared pin merged into the request. +func (s *Server) authorize(ctx context.Context, route passthroughRoute, header http.Header, msg *dynamicpb.Message) (*Gateway, error) { + gw := newGateway(s.cfg.gatewayURL, header.Get("authorization"), header.Get(orgHeader), header.Get(sessionHeader)) if !route.module.ViewerBearer { acting, err := gw.ForModule(ctx, route.module.As, route.scopes()...) if err != nil { @@ -353,22 +404,11 @@ func (s *Server) forward(ctx context.Context, route passthroughRoute, req *conne if err != nil { return nil, connect.NewError(connect.CodeInternal, errors.New("the declared pin cannot be encoded")) } - if err := (proto.UnmarshalOptions{Merge: true}).Unmarshal(pinned, req.Msg); err != nil { + if err := (proto.UnmarshalOptions{Merge: true}).Unmarshal(pinned, msg); err != nil { return nil, connect.NewError(connect.CodeInternal, errors.New("the declared pin cannot be applied")) } } - out := dynamicpb.NewMessage(route.md.Output()) - var opts []TranscodeOption - if route.method.MaxResponseBytes > 0 { - opts = append(opts, MaxResponseBytes(route.method.MaxResponseBytes)) - } - if err := gw.Transcoded(ctx, "/v1/"+route.module.As, route.method.Name, req.Msg, out, opts...); err != nil { - return nil, relayedError(err) - } - if !route.method.WholeResponse { - out = Apply(route.method.Response, out) - } - return out, nil + return gw, nil } // relayedError turns a failed call into the Connect error the page receives. @@ -395,6 +435,17 @@ func relayedError(err error) error { } return connect.NewError(code, errors.New(message)) } + var streamErr *StreamError + if errors.As(err, &streamErr) && streamErr != nil { + code := connect.CodeUnknown + if streamErr.Code > 0 && streamErr.Code <= 16 { + code = connect.Code(streamErr.Code) + } + return connect.NewError(code, errors.New(truncated(streamErr.Message))) + } + if errors.Is(err, ErrStreamMessageTooLarge) { + return connect.NewError(connect.CodeResourceExhausted, errors.New("a message of the module's stream exceeds its declared bound")) + } var clientErr *ClientError if errors.As(err, &clientErr) && clientErr != nil { return connect.NewError(codeForStatus(clientErr.StatusCode), errors.New(clientErr.Message)) @@ -486,6 +537,10 @@ func PassthroughOperations(modules ...ConsumedModule) ([]Operation, error) { callers = fmt.Sprintf("Signed-in viewers. The %s module authenticates the viewer's bearer and authorizes every call itself.", route.module.As) } behavior := fmt.Sprintf("A Connect unary call, forwarded to the module's %s %s binding. ", route.verb, route.template) + if route.md.IsStreamingServer() { + behavior = fmt.Sprintf("A Connect server-streaming call, forwarded to the module's %s %s binding, which answers as newline-delimited JSON. "+ + "The stream stays open at most %s and each message is at most %d bytes. ", route.verb, route.template, route.streamDuration(), route.streamMessageLimit()) + } if route.method.Pin != nil { pin, _ := protojson.Marshal(route.method.Pin) behavior += fmt.Sprintf("Every request is merged with the declared pin %s first. ", pin) @@ -495,14 +550,18 @@ func PassthroughOperations(modules ...ConsumedModule) ([]Operation, error) { } else { behavior += "Only these response fields are returned: " + strings.Join(route.method.Response.paths(), ", ") + "." } + if route.md.IsStreamingServer() { + behavior += " The response fields apply to every streamed message." + } ops = append(ops, Operation{ - Path: path, - Method: "post", - Summary: fmt.Sprintf("Call %s on the %s module as the viewer", route.md.FullName(), route.module.As), - Callers: callers, - Behavior: behavior, - Request: describedMessage{route.md.Input()}, - Response: describedMessage{route.md.Output()}, + Path: path, + Method: "post", + Summary: fmt.Sprintf("Call %s on the %s module as the viewer", route.md.FullName(), route.module.As), + Streaming: route.md.IsStreamingServer(), + Callers: callers, + Behavior: behavior, + Request: describedMessage{route.md.Input()}, + Response: describedMessage{route.md.Output()}, // Documentation only: InterfaceDocument requires a handler, and the // passthrough, not Operations, serves these routes. Handler: func(context.Context, *Gateway) (any, error) { return nil, errors.New("served by the passthrough") }, diff --git a/stream.go b/stream.go new file mode 100644 index 0000000..7ae008c --- /dev/null +++ b/stream.go @@ -0,0 +1,310 @@ +package solution + +import ( + "bufio" + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "mime" + "net/http" + "sync/atomic" + "time" + + "connectrpc.com/connect" + "google.golang.org/protobuf/encoding/protojson" + "google.golang.org/protobuf/proto" + "google.golang.org/protobuf/reflect/protoreflect" + "google.golang.org/protobuf/types/dynamicpb" +) + +// --- Streamed methods --- +// +// A consumed module's server-streaming method passes through as a Connect +// server-streaming procedure, under the same authority as a unary one: the +// viewer's Work Context minted for that method (or the bearer alone, for a +// ViewerBearer module), and the declared response fields applied to every +// message before it reaches the page. +// +// The module answers the method's google.api.http binding the way grpc-gateway +// answers a server-streaming method: a 2xx response, content type +// application/x-ndjson (application/json is accepted too), whose body is one +// JSON object per line, flushed as each is ready: +// +// {"result": } +// {"error": {"code": , "message": ""}} +// +// Blank lines are ignored (a module may send them to keep an idle connection +// alive). An error line ends the stream with that code and message; the end of +// the body ends it normally. A non-2xx answer before the stream starts is the +// module's refusal, relayed like a unary one. +// +// Every stream is bounded: its duration by the method's MaxStreamDuration, and +// each message by MaxStreamMessageBytes. The page disconnecting cancels the +// module request at once. + +const ( + // DefaultStreamDuration bounds a stream unless the method declares its own. + DefaultStreamDuration = 5 * time.Minute + // MaxStreamDurationLimit is the longest stream a method may declare. + MaxStreamDurationLimit = 30 * time.Minute + // DefaultStreamMessageLimit bounds one streamed message unless the method + // declares its own. + DefaultStreamMessageLimit = 1 << 20 +) + +// ErrStreamMessageTooLarge is a streamed message over its declared bound. +var ErrStreamMessageTooLarge = errors.New("a streamed message exceeds its bound") + +// StreamError is an error line a module's stream ended with. +type StreamError struct { + Code int32 + Message string +} + +func (e *StreamError) Error() string { + return fmt.Sprintf("the module's stream ended with code %d: %s", e.Code, e.Message) +} + +// checkStreamBounds refuses stream bounds on a unary method (they would never +// apply) and bounds a streaming method cannot keep. +func checkStreamBounds(md protoreflect.MethodDescriptor, method ConsumedMethod) error { + if !md.IsStreamingServer() { + if method.MaxStreamDuration != 0 || method.MaxStreamMessageBytes != 0 { + return fmt.Errorf("%s is unary, but declares stream bounds", md.FullName()) + } + return nil + } + switch { + case method.MaxStreamDuration < 0 || method.MaxStreamDuration > MaxStreamDurationLimit: + return fmt.Errorf("%s: MaxStreamDuration %s is outside (0, %s]", md.FullName(), method.MaxStreamDuration, MaxStreamDurationLimit) + case method.MaxStreamMessageBytes < 0: + return fmt.Errorf("%s: MaxStreamMessageBytes %d is negative", md.FullName(), method.MaxStreamMessageBytes) + case method.MaxResponseBytes != 0: + return fmt.Errorf("%s streams: bound each message with MaxStreamMessageBytes, not MaxResponseBytes", md.FullName()) + } + return nil +} + +func (r passthroughRoute) streamDuration() time.Duration { + if r.method.MaxStreamDuration > 0 { + return r.method.MaxStreamDuration + } + return DefaultStreamDuration +} + +func (r passthroughRoute) streamMessageLimit() int64 { + if r.method.MaxStreamMessageBytes > 0 { + return r.method.MaxStreamMessageBytes + } + return DefaultStreamMessageLimit +} + +// forwardStream answers one streamed call as the viewer: the declared +// authority, the module's stream over its binding, and only the declared +// response fields of each message. +func (s *Server) forwardStream(ctx context.Context, route passthroughRoute, req *connect.Request[dynamicpb.Message], stream *connect.ServerStream[dynamicpb.Message]) error { + if req.Header().Get("authorization") == "" { + return connect.NewError(connect.CodeUnauthenticated, errors.New("missing bearer")) + } + // The whole stream, from the call to its last message, is bounded; the + // page going away cancels it (ctx is the page's request). + ctx, cancel := context.WithTimeout(ctx, route.streamDuration()) + defer cancel() + timeout := route.method.Timeout + if timeout <= 0 { + timeout = DefaultPassthroughTimeout + } + setup, cancelSetup := context.WithTimeout(ctx, timeout) + gw, err := s.authorize(setup, route, req.Header(), req.Msg) + cancelSetup() + if err != nil { + return err + } + err = gw.TranscodedStream(ctx, "/v1/"+route.module.As, route.method.Name, req.Msg, func() proto.Message { + return dynamicpb.NewMessage(route.md.Output()) + }, func(msg proto.Message) error { + out := msg.(*dynamicpb.Message) + if !route.method.WholeResponse { + out = Apply(route.method.Response, out) + } + return stream.Send(out) + }, StreamFirstAnswerWithin(timeout), MaxStreamMessageBytes(route.streamMessageLimit())) + if err == nil { + return nil + } + if errors.Is(err, context.DeadlineExceeded) && ctx.Err() != nil { + return connect.NewError(connect.CodeDeadlineExceeded, fmt.Errorf("the stream reached its declared duration (%s)", route.streamDuration())) + } + var connectErr *connect.Error + if errors.As(err, &connectErr) { + return err + } + return relayedError(err) +} + +// StreamOption adjusts one TranscodedStream call. +type StreamOption func(*streamOptions) + +type streamOptions struct { + maxMessageBytes int64 + firstAnswer time.Duration +} + +// MaxStreamMessageBytes bounds each message of the stream. A larger one ends +// the stream with ErrStreamMessageTooLarge, never a truncated message. +func MaxStreamMessageBytes(n int64) StreamOption { + return func(o *streamOptions) { o.maxMessageBytes = n } +} + +// StreamFirstAnswerWithin bounds how long the module takes to answer (its +// status line and headers); the stream after that is bounded by ctx. +func StreamFirstAnswerWithin(d time.Duration) StreamOption { + return func(o *streamOptions) { o.firstAnswer = d } +} + +// TranscodedStream calls one server-streaming RPC of a composed module through +// the gateway, over the module's HTTP/JSON binding, and hands each streamed +// message to each in order: newMsg makes the (empty) message each line decodes +// into. It returns when the stream ends: nil at the end of the body, a +// *StreamError for an error line, the error each returned, or ctx's error. See +// "Streamed methods" for the wire. +func (g *Gateway) TranscodedStream(ctx context.Context, prefix, method string, req proto.Message, newMsg func() proto.Message, each func(proto.Message) error, opts ...StreamOption) error { + options := streamOptions{maxMessageBytes: DefaultStreamMessageLimit} + for _, opt := range opts { + opt(&options) + } + md, err := methodDescriptor(method) + if err != nil { + return err + } + if !md.IsStreamingServer() || md.IsStreamingClient() { + return fmt.Errorf("transcoded stream %s: the method is not server-streaming", md.FullName()) + } + call, err := transcodeRequest(prefix, method, req, newMsg()) + if err != nil { + return err + } + var body io.Reader + if call.body != nil { + body = bytes.NewReader(call.body) + } + // The module request lives as long as the stream; it is cancelled when the + // stream ends for any reason, the page going away included. + reqCtx, cancelReq := context.WithCancel(ctx) + defer cancelReq() + httpReq, err := http.NewRequestWithContext(reqCtx, call.verb, g.baseURL+call.path, body) + if err != nil { + return err + } + httpReq.Header.Set("accept", "application/x-ndjson, application/json") + if call.body != nil { + httpReq.Header.Set("content-type", "application/json") + } + var late atomic.Bool + if options.firstAnswer > 0 { + // A module that never starts its stream is abandoned; the stream + // itself, once started, is bounded only by ctx. + timer := time.AfterFunc(options.firstAnswer, func() { late.Store(true); cancelReq() }) + defer timer.Stop() + httpResp, err := g.HTTPClient().Do(httpReq) + timer.Stop() + if err != nil { + if late.Load() && ctx.Err() == nil { + return fmt.Errorf("%s %s: the module did not start its stream within %s: %w", call.verb, call.template, options.firstAnswer, context.DeadlineExceeded) + } + return fmt.Errorf("%s %s: %w", call.verb, call.template, err) + } + return readStream(ctx, call, httpResp, options, newMsg, each) + } + httpResp, err := g.HTTPClient().Do(httpReq) + if err != nil { + return fmt.Errorf("%s %s: %w", call.verb, call.template, err) + } + return readStream(ctx, call, httpResp, options, newMsg, each) +} + +// readStream reads a started stream to its end. +func readStream(ctx context.Context, call transcodedCall, httpResp *http.Response, options streamOptions, newMsg func() proto.Message, each func(proto.Message) error) error { + // Closed, never drained: a stream has no end to drain to, and the + // request's context cancellation has already stopped it on the way out. + defer httpResp.Body.Close() + if httpResp.StatusCode < 200 || httpResp.StatusCode > 299 { + detail, _ := io.ReadAll(io.LimitReader(httpResp.Body, gatewayErrorDetailLimit)) + return fmt.Errorf("%s %s: %w", call.verb, call.template, &GatewayError{StatusCode: httpResp.StatusCode, detail: detail}) + } + if media, _, err := mime.ParseMediaType(httpResp.Header.Get("content-type")); err != nil || (media != "application/x-ndjson" && media != "application/json") { + return fmt.Errorf("%s %s: the module answered %q, not a newline-delimited JSON stream", call.verb, call.template, httpResp.Header.Get("content-type")) + } + // A line is its message plus the envelope around it. + reader := bufio.NewReaderSize(httpResp.Body, 64<<10) + lineLimit := options.maxMessageBytes + 64 + for { + line, err := readLine(reader, lineLimit) + if errors.Is(err, io.EOF) { + return nil + } + if err != nil { + if ctx.Err() != nil { + return ctx.Err() + } + if errors.Is(err, ErrStreamMessageTooLarge) { + return err + } + return fmt.Errorf("%s %s: %w", call.verb, call.template, err) + } + line = bytes.TrimSpace(line) + if len(line) == 0 { + continue + } + var envelope struct { + Result json.RawMessage `json:"result"` + Error *struct { + Code int32 `json:"code"` + Message string `json:"message"` + } `json:"error"` + } + if err := json.Unmarshal(line, &envelope); err != nil { + return fmt.Errorf("%s %s: an undecodable line in the stream: %w", call.verb, call.template, err) + } + switch { + case envelope.Error != nil: + return &StreamError{Code: envelope.Error.Code, Message: envelope.Error.Message} + case envelope.Result == nil: + return fmt.Errorf("%s %s: a line in the stream carries neither a result nor an error", call.verb, call.template) + case int64(len(envelope.Result)) > options.maxMessageBytes: + return ErrStreamMessageTooLarge + } + msg := newMsg() + if err := (protojson.UnmarshalOptions{DiscardUnknown: true}).Unmarshal(envelope.Result, msg); err != nil { + return fmt.Errorf("%s %s: invalid %s in the stream: %w", call.verb, call.template, msg.ProtoReflect().Descriptor().FullName(), err) + } + if err := each(msg); err != nil { + return err + } + } +} + +// readLine reads one line, refusing one longer than limit rather than +// buffering it whole. +func readLine(r *bufio.Reader, limit int64) ([]byte, error) { + var line []byte + for { + chunk, isPrefix, err := r.ReadLine() + if err != nil { + if errors.Is(err, io.EOF) && len(line) > 0 { + return line, nil + } + return nil, err + } + line = append(line, chunk...) + if int64(len(line)) > limit { + return nil, ErrStreamMessageTooLarge + } + if !isPrefix { + return line, nil + } + } +} diff --git a/stream_test.go b/stream_test.go new file mode 100644 index 0000000..431a57f --- /dev/null +++ b/stream_test.go @@ -0,0 +1,353 @@ +package solution + +import ( + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + "connectrpc.com/connect" + codefly "github.com/codefly-dev/sdk-go" + "google.golang.org/protobuf/reflect/protoreflect" + "google.golang.org/protobuf/types/dynamicpb" +) + +const watchName = "/things.v1.Things/Watch" + +// streamModule declares the things module's Watch stream, returning only +// entry_id and big of each message, with Search as a unary method beside it. +func streamModule(watch ConsumedMethod) ConsumedModule { + module := passthroughModule() + watch.Name = watchName + if watch.Response.desc == nil && !watch.WholeResponse { + watch.Response = MustFieldMask(thingMessage("Thing"), "entry_id", "big") + } + watch.Scopes = []Scope{{ResourceKind: "things", Actions: []string{"watch"}}} + module.Methods = append(module.Methods, watch) + return module +} + +// streamGateway is the host gateway with a module whose Watch answers with the +// lines its test writes, flushed one by one; the unary routes answer as +// moduleGateway does. +type streamGateway struct { + *httptest.Server + lines chan string + contentType string + status int + + mu sync.Mutex + mints []mintRequest + watches []*http.Request + canceled chan struct{} +} + +func newStreamGateway(t *testing.T) *streamGateway { + t.Helper() + g := &streamGateway{lines: make(chan string, 16), contentType: "application/x-ndjson", status: http.StatusOK, canceled: make(chan struct{}, 1)} + g.Server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, _ := io.ReadAll(r.Body) + if r.URL.Path == workContextStartTaskProcedure { + var mint mintRequest + _ = json.Unmarshal(body, &mint) + g.mu.Lock() + g.mints = append(g.mints, mint) + g.mu.Unlock() + writeJSON(w, http.StatusOK, map[string]any{ + "token": "context-" + mint.Audience + ".1", "orgId": mint.OrgID, + "ownerPrincipalId": "viewer", "currentActorPrincipalId": "viewer", + "expiresAt": time.Now().Add(5 * time.Minute).UTC().Format(time.RFC3339Nano), + }) + return + } + g.mu.Lock() + g.watches = append(g.watches, r.Clone(context.Background())) + g.mu.Unlock() + if g.status != http.StatusOK { + w.WriteHeader(g.status) + _, _ = io.WriteString(w, `{"code":7,"message":"not yours"}`) + return + } + w.Header().Set("content-type", g.contentType) + w.WriteHeader(http.StatusOK) + w.(http.Flusher).Flush() + for { + select { + case line, ok := <-g.lines: + if !ok { + return + } + _, _ = io.WriteString(w, line+"\n") + w.(http.Flusher).Flush() + case <-r.Context().Done(): + g.canceled <- struct{}{} + return + } + } + })) + t.Cleanup(g.Close) + return g +} + +// watchClient is the page: a Connect server-streaming client over JSON, as +// connect-es is, against the solution's passthrough served for real. +func watchClient(t *testing.T, s *Server) *connect.Client[dynamicpb.Message, dynamicpb.Message] { + t.Helper() + routes, err := s.validatePassthroughRoutesForTest() + if err != nil { + t.Fatal(err) + } + srv := httptest.NewServer(s.passthroughHandler(routes)) + t.Cleanup(srv.Close) + md := transcodeTestFile.Services().ByName("Things").Methods().ByName("Watch") + return connect.NewClient[dynamicpb.Message, dynamicpb.Message](srv.Client(), srv.URL+"/modules/things/things.v1.Things/Watch", + connect.WithSchema(md), connect.WithProtoJSON(), + connect.WithResponseInitializer(func(_ connect.Spec, msg any) error { + *msg.(*dynamicpb.Message) = *dynamicpb.NewMessage(md.Output()) + return nil + })) +} + +func (s *Server) validatePassthroughRoutesForTest() (map[string]passthroughRoute, error) { + return resolvePassthrough(s.consumed) +} + +func watchRequest(entry string) *connect.Request[dynamicpb.Message] { + msg := thingMessage("GetThingRequest") + setField(msg, "entry_id", protoreflect.ValueOfString(entry)) + req := connect.NewRequest(msg) + req.Header().Set("authorization", "Bearer viewer") + req.Header().Set(orgHeader, "org-1") + req.Header().Set(sessionHeader, "session-1") + return req +} + +func field(m *dynamicpb.Message, name string) protoreflect.Value { + return m.Get(m.Descriptor().Fields().ByName(protoreflect.Name(name))) +} + +func has(m *dynamicpb.Message, name string) bool { + return m.Has(m.Descriptor().Fields().ByName(protoreflect.Name(name))) +} + +func TestPassthroughStreamsEachMessageMaskedAsTheViewer(t *testing.T) { + gw := newStreamGateway(t) + s := passthroughServer(gw.URL, streamModule(ConsumedMethod{})) + gw.lines <- `{"result":{"entryId":"e1","big":"1","secret":"s1"}}` + gw.lines <- `` + gw.lines <- `{"result":{"entryId":"e1","big":"2","secret":"s2","sub":{"name":"x"}}}` + close(gw.lines) + stream, err := watchClient(t, s).CallServerStream(context.Background(), watchRequest("e1")) + if err != nil { + t.Fatal(err) + } + var got []string + for stream.Receive() { + msg := stream.Msg() + // The declared fields reach the page, on every message; nothing else does. + if has(msg, "secret") || has(msg, "sub") { + t.Fatalf("an undeclared field reached the page: %v", msg) + } + got = append(got, fmt.Sprintf("%s/%d", field(msg, "entry_id").String(), field(msg, "big").Int())) + } + if err := stream.Err(); err != nil { + t.Fatalf("stream: %v", err) + } + if strings.Join(got, ",") != "e1/1,e1/2" { + t.Fatalf("messages = %v", got) + } + // Called over the binding, with the authority minted for this method alone. + if len(gw.watches) != 1 || gw.watches[0].Method != http.MethodGet || gw.watches[0].URL.Path != "/v1/things/e1/watch" { + t.Fatalf("module calls = %v", gw.watches) + } + if gw.watches[0].Header.Get(codefly.WorkContextHeaderName) != "context-things.1" || !strings.Contains(gw.watches[0].Header.Get("accept"), "application/x-ndjson") { + t.Fatalf("headers = %v", gw.watches[0].Header) + } + if len(gw.mints) != 1 || mintedActions(gw.mints[0]) != "things:watch" { + t.Fatalf("mints = %+v, want the Watch method's scope only", gw.mints) + } +} + +func TestPassthroughStreamForwardsOnlyTheBearerForAModuleThatAuthenticatesTheViewer(t *testing.T) { + gw := newStreamGateway(t) + module := streamModule(ConsumedMethod{}) + module.Scopes, module.ViewerBearer = nil, true + for i := range module.Methods { + module.Methods[i].Scopes = nil + } + s := passthroughServer(gw.URL, module) + close(gw.lines) + stream, err := watchClient(t, s).CallServerStream(context.Background(), watchRequest("e1")) + if err != nil { + t.Fatal(err) + } + for stream.Receive() { + } + if err := stream.Err(); err != nil { + t.Fatal(err) + } + if len(gw.mints) != 0 || gw.watches[0].Header.Get(codefly.WorkContextHeaderName) != "" || gw.watches[0].Header.Get("authorization") != "Bearer viewer" { + t.Fatalf("want the viewer's bearer and no capability: mints %d, headers %v", len(gw.mints), gw.watches[0].Header) + } +} + +func receiveAll(stream *connect.ServerStreamForClient[dynamicpb.Message]) (int, error) { + n := 0 + for stream.Receive() { + n++ + } + return n, stream.Err() +} + +func TestPassthroughStreamEndsWithTheModulesErrorLine(t *testing.T) { + gw := newStreamGateway(t) + s := passthroughServer(gw.URL, streamModule(ConsumedMethod{})) + gw.lines <- `{"result":{"entryId":"e1"}}` + gw.lines <- `{"error":{"code":9,"message":"the turn was deleted"}}` + stream, err := watchClient(t, s).CallServerStream(context.Background(), watchRequest("e1")) + if err != nil { + t.Fatal(err) + } + n, err := receiveAll(stream) + if n != 1 || connect.CodeOf(err) != connect.CodeFailedPrecondition || !strings.Contains(err.Error(), "the turn was deleted") { + t.Fatalf("got %d messages, err %v; want one message then failed_precondition", n, err) + } +} + +func TestPassthroughStreamRelaysARefusalBeforeTheStream(t *testing.T) { + gw := newStreamGateway(t) + gw.status = http.StatusForbidden + s := passthroughServer(gw.URL, streamModule(ConsumedMethod{})) + stream, err := watchClient(t, s).CallServerStream(context.Background(), watchRequest("e1")) + if err != nil { + t.Fatal(err) + } + if _, err := receiveAll(stream); connect.CodeOf(err) != connect.CodePermissionDenied { + t.Fatalf("err = %v, want the module's permission_denied", err) + } +} + +func TestPassthroughStreamRefusesAnOversizedMessageWithoutTruncating(t *testing.T) { + gw := newStreamGateway(t) + s := passthroughServer(gw.URL, streamModule(ConsumedMethod{MaxStreamMessageBytes: 64})) + gw.lines <- `{"result":{"entryId":"e1"}}` + gw.lines <- `{"result":{"entryId":"` + strings.Repeat("x", 200) + `"}}` + stream, err := watchClient(t, s).CallServerStream(context.Background(), watchRequest("e1")) + if err != nil { + t.Fatal(err) + } + n, err := receiveAll(stream) + if n != 1 || connect.CodeOf(err) != connect.CodeResourceExhausted { + t.Fatalf("got %d messages, err %v; want one message then resource_exhausted", n, err) + } +} + +func TestPassthroughStreamEndsAtItsDeclaredDuration(t *testing.T) { + gw := newStreamGateway(t) + s := passthroughServer(gw.URL, streamModule(ConsumedMethod{MaxStreamDuration: 300 * time.Millisecond})) + gw.lines <- `{"result":{"entryId":"e1"}}` + stream, err := watchClient(t, s).CallServerStream(context.Background(), watchRequest("e1")) + if err != nil { + t.Fatal(err) + } + start := time.Now() + n, err := receiveAll(stream) + if n != 1 || connect.CodeOf(err) != connect.CodeDeadlineExceeded || !strings.Contains(err.Error(), "declared duration") { + t.Fatalf("got %d messages, err %v; want one message then deadline_exceeded", n, err) + } + if elapsed := time.Since(start); elapsed > 5*time.Second { + t.Fatalf("the stream outlived its duration: %s", elapsed) + } + select { + case <-gw.canceled: + case <-time.After(5 * time.Second): + t.Fatal("the module's request was not cancelled when the stream ended") + } +} + +func TestPassthroughStreamCancelsTheModuleWhenThePageGoesAway(t *testing.T) { + gw := newStreamGateway(t) + s := passthroughServer(gw.URL, streamModule(ConsumedMethod{})) + gw.lines <- `{"result":{"entryId":"e1"}}` + ctx, cancel := context.WithCancel(context.Background()) + stream, err := watchClient(t, s).CallServerStream(ctx, watchRequest("e1")) + if err != nil { + t.Fatal(err) + } + if !stream.Receive() { + t.Fatalf("first message: %v", stream.Err()) + } + cancel() + _ = stream.Close() + select { + case <-gw.canceled: + case <-time.After(5 * time.Second): + t.Fatal("the module's request outlived the page") + } +} + +func TestPassthroughStreamRefusesAnAnswerThatIsNotAStream(t *testing.T) { + gw := newStreamGateway(t) + gw.contentType = "text/html" + s := passthroughServer(gw.URL, streamModule(ConsumedMethod{})) + close(gw.lines) + stream, err := watchClient(t, s).CallServerStream(context.Background(), watchRequest("e1")) + if err != nil { + t.Fatal(err) + } + if _, err := receiveAll(stream); connect.CodeOf(err) != connect.CodeUnavailable { + t.Fatalf("err = %v, want unavailable, never the body as a message", err) + } +} + +func TestPassthroughRefusesStreamBoundsItCannotKeep(t *testing.T) { + search := passthroughModule().Methods[0] + for name, module := range map[string]ConsumedModule{ + "a client stream": streamModule(ConsumedMethod{}), + "bounds on a unary method": {As: "things", ViewerBearer: true, Methods: []ConsumedMethod{{Name: search.Name, Response: search.Response, MaxStreamDuration: time.Minute}}}, + "a negative duration": streamModule(ConsumedMethod{MaxStreamDuration: -time.Second}), + "a duration over the limit": streamModule(ConsumedMethod{MaxStreamDuration: MaxStreamDurationLimit + time.Second}), + "a negative message bound": streamModule(ConsumedMethod{MaxStreamMessageBytes: -1}), + "a response bound on streams": streamModule(ConsumedMethod{MaxResponseBytes: 10}), + } { + if name == "a client stream" { + module.Methods = append(module.Methods, ConsumedMethod{Name: "/things.v1.Things/Upload", WholeResponse: true}) + } + if _, err := resolvePassthrough([]ConsumedModule{module}); err == nil { + t.Errorf("%s: accepted", name) + } + } + if _, err := resolvePassthrough([]ConsumedModule{streamModule(ConsumedMethod{MaxStreamDuration: time.Minute, MaxStreamMessageBytes: 4096})}); err != nil { + t.Fatalf("a bounded stream: %v", err) + } +} + +func TestPassthroughOperationsDocumentAStreamedMethod(t *testing.T) { + ops, err := PassthroughOperations(streamModule(ConsumedMethod{MaxStreamDuration: time.Minute, MaxStreamMessageBytes: 4096})) + if err != nil { + t.Fatal(err) + } + var watch *Operation + for i := range ops { + if strings.HasSuffix(ops[i].Path, "/Watch") { + watch = &ops[i] + } + } + if watch == nil || !watch.Streaming || !strings.Contains(watch.Behavior, "server-streaming") || + !strings.Contains(watch.Behavior, "1m0s") || !strings.Contains(watch.Behavior, "4096 bytes") || !strings.Contains(watch.Behavior, "every streamed message") { + t.Fatalf("watch operation = %+v", watch) + } + doc, err := InterfaceDocument(InterfaceInfo{Title: "t", Version: "1", Tag: "t", OperationPrefix: "T"}, ops) + if err != nil { + t.Fatal(err) + } + if !strings.Contains(string(doc), `"x-streaming": "server"`) { + t.Fatalf("the document does not mark the stream: %s", doc) + } +} diff --git a/transcode_test.go b/transcode_test.go index 2dc95ce..395fed9 100644 --- a/transcode_test.go +++ b/transcode_test.go @@ -76,6 +76,12 @@ var transcodeTestFile = func() protoreflect.FileDescriptor { {Name: proto.String("Unbound"), InputType: proto.String(".things.v1.GetThingRequest"), OutputType: proto.String(".things.v1.Thing")}, {Name: proto.String("Wide"), InputType: proto.String(".things.v1.GetThingRequest"), OutputType: proto.String(".things.v1.Thing"), Options: http(&annotations.HttpRule{Pattern: &annotations.HttpRule_Get{Get: "/v1/things/{entry_id=**}"}})}, + {Name: proto.String("Watch"), InputType: proto.String(".things.v1.GetThingRequest"), OutputType: proto.String(".things.v1.Thing"), + ServerStreaming: proto.Bool(true), + Options: http(&annotations.HttpRule{Pattern: &annotations.HttpRule_Get{Get: "/v1/things/{entry_id}/watch"}})}, + {Name: proto.String("Upload"), InputType: proto.String(".things.v1.GetThingRequest"), OutputType: proto.String(".things.v1.Thing"), + ClientStreaming: proto.Bool(true), + Options: http(&annotations.HttpRule{Pattern: &annotations.HttpRule_Post{Post: "/v1/things/upload"}, Body: "*"})}, }, }}, }