From 9725e789508ecafb8cf5c96e9a51d28c45cdeb4e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Pawe=C5=82=20Gronowski?= Date: Wed, 14 Oct 2026 17:16:53 +0200 Subject: [PATCH 1/7] cmd/mobyextgen: Copy interface doc comments into generated code MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The generated .proto now carries the contract interface's doc comment on the service and each method's doc comment on its rpc. The generated raw gRPC server and client interfaces repeat the method comments. Signed-off-by: Paweł Gronowski --- cmd/mobyextgen/model.go | 2 + cmd/mobyextgen/parse.go | 44 ++++++++++++++++++- cmd/mobyextgen/proto.go | 2 + cmd/mobyextgen/wire.go | 15 +++++++ example/greeter/v0/greeter.proto | 1 + internal/launcher/echo/v1/echo.proto | 2 + .../launcher/echo/v1/protogen/wire.gen.go | 2 + sdk/sdkapi/extension.proto | 11 +++++ sdk/sdkapi/protogen/wire.gen.go | 10 +++++ 9 files changed, 87 insertions(+), 2 deletions(-) diff --git a/cmd/mobyextgen/model.go b/cmd/mobyextgen/model.go index 8ad2217..0328b12 100644 --- a/cmd/mobyextgen/model.go +++ b/cmd/mobyextgen/model.go @@ -14,6 +14,7 @@ type point struct { iface string // Go service interface name isPoint bool // whether the contract declares an extensions.Point isSingle bool // whether the point was declared with DefineSinglePoint + doc string // documentation on the provider interface methods []method messages []message } @@ -28,6 +29,7 @@ type method struct { request string // request message name response string // response message name bareError bool // true for `M(ctx, *Req) error`; response is a generated empty message + doc string // method documentation } type message struct { diff --git a/cmd/mobyextgen/parse.go b/cmd/mobyextgen/parse.go index c3a7fce..a5580c6 100644 --- a/cmd/mobyextgen/parse.go +++ b/cmd/mobyextgen/parse.go @@ -77,6 +77,7 @@ func parseContract(dir, serviceIdentity string) (point, error) { if ifaceType == nil { return point{}, fmt.Errorf("interface %q not found", pt.iface) } + pt.doc = interfaceComment(files, pt.iface) // Inherit the contract's copyright and license rather than imposing the // generator's notices on output produced for other projects. for _, f := range files { @@ -220,6 +221,33 @@ func structHasPBTag(st *ast.StructType) bool { return false } +func interfaceComment(files []*ast.File, name string) string { + for _, file := range files { + for _, decl := range file.Decls { + group, ok := decl.(*ast.GenDecl) + if !ok || group.Tok != token.TYPE { + continue + } + for _, spec := range group.Specs { + typeSpec := spec.(*ast.TypeSpec) + if typeSpec.Name.Name != name { + continue + } + doc := typeSpec.Doc + // Like go/doc, a group comment documents only a single-type group. + if doc == nil && len(group.Specs) == 1 { + doc = group.Doc + } + if doc != nil { + return strings.TrimSpace(doc.Text()) + } + return "" + } + } + } + return "" +} + func parseMethods(iface *ast.InterfaceType) ([]method, error) { var methods []method for _, m := range iface.Methods.List { @@ -228,6 +256,7 @@ func parseMethods(iface *ast.InterfaceType) ([]method, error) { continue } name := m.Names[0].Name + doc := fieldComment(m) if ft.Params == nil || len(ft.Params.List) == 0 { return nil, fmt.Errorf("method %q: expected a request parameter", name) } @@ -239,13 +268,13 @@ func parseMethods(iface *ast.InterfaceType) ([]method, error) { res := results(ft) switch { case len(res) == 1 && isIdent(res[0], "error"): - methods = append(methods, method{name: name, request: req, response: name + "Response", bareError: true}) + methods = append(methods, method{name: name, request: req, response: name + "Response", bareError: true, doc: doc}) case len(res) == 2 && isIdent(res[1], "error"): resp, err := pointerIdent(res[0]) if err != nil { return nil, fmt.Errorf("method %q response: %w", name, err) } - methods = append(methods, method{name: name, request: req, response: resp}) + methods = append(methods, method{name: name, request: req, response: resp, doc: doc}) default: return nil, fmt.Errorf("method %q: result must be `error` or `(*Resp, error)`", name) } @@ -253,6 +282,17 @@ func parseMethods(iface *ast.InterfaceType) ([]method, error) { return methods, nil } +func fieldComment(field *ast.Field) string { + group := field.Doc + if group == nil { + group = field.Comment + } + if group == nil { + return "" + } + return strings.TrimSpace(group.Text()) +} + func parseMessages(files []*ast.File, msgNames map[string]bool) ([]message, error) { var messages []message for _, f := range files { diff --git a/cmd/mobyextgen/proto.go b/cmd/mobyextgen/proto.go index 4a7423a..20b6d15 100644 --- a/cmd/mobyextgen/proto.go +++ b/cmd/mobyextgen/proto.go @@ -21,8 +21,10 @@ func emitProto(pt point) ([]byte, error) { fmt.Fprintf(&b, "package %s;\n\n", pt.id) fmt.Fprintf(&b, "option go_package = %q;\n\n", pt.protogenImport()) + emitComment(&b, "", pt.doc) fmt.Fprintf(&b, "service %s {\n", pt.service) for _, m := range pt.methods { + emitComment(&b, " ", m.doc) fmt.Fprintf(&b, " rpc %s(%s) returns (%s);\n", m.name, m.request, m.response) } fmt.Fprintln(&b, "}") diff --git a/cmd/mobyextgen/wire.go b/cmd/mobyextgen/wire.go index 6af9ae3..9beefa8 100644 --- a/cmd/mobyextgen/wire.go +++ b/cmd/mobyextgen/wire.go @@ -59,6 +59,7 @@ func emitWire(pt point) ([]byte, error) { `, svc) } for _, m := range pt.methods { + emitComment(&b, "\t", m.doc) fmt.Fprintf(&b, "\t%s(context.Context, *%s) (*%s, error)\n", m.name, m.request, m.response) } fmt.Fprintln(&b, "}") @@ -115,6 +116,7 @@ func emitWire(pt point) ([]byte, error) { `, svc) } for _, m := range pt.methods { + emitComment(&b, "\t", m.doc) fmt.Fprintf(&b, "\t%s(ctx context.Context, in *%s, opts ...grpc.CallOption) (*%s, error)\n", m.name, m.request, m.response) } fmt.Fprintln(&b, "}") @@ -242,6 +244,19 @@ func emitWire(pt point) ([]byte, error) { return format.Source([]byte(b.String())) } +func emitComment(b *strings.Builder, indent, doc string) { + if doc == "" { + return + } + for line := range strings.SplitSeq(doc, "\n") { + if line == "" { + fmt.Fprintf(b, "%s//\n", indent) + continue + } + fmt.Fprintf(b, "%s// %s\n", indent, line) + } +} + func emitConversions(b *strings.Builder, cpkg string, msg message) { conv := lowerFirst(msg.name) diff --git a/example/greeter/v0/greeter.proto b/example/greeter/v0/greeter.proto index c153eb2..34c0ea6 100644 --- a/example/greeter/v0/greeter.proto +++ b/example/greeter/v0/greeter.proto @@ -9,6 +9,7 @@ package org.mobyproject.extension.example.greeter.v0; option go_package = "github.com/moby/extensions/example/greeter/v0/protogen"; +// Greeter is the example service. service Greeter { rpc Greet(HelloRequest) returns (HelloReply); } diff --git a/internal/launcher/echo/v1/echo.proto b/internal/launcher/echo/v1/echo.proto index abbb3f3..4cfb068 100644 --- a/internal/launcher/echo/v1/echo.proto +++ b/internal/launcher/echo/v1/echo.proto @@ -9,7 +9,9 @@ package moby.extensions.internal.launcher.echo.v1; option go_package = "github.com/moby/extensions/internal/launcher/echo/v1/protogen"; +// Echo is the provider interface for the echo test point. service Echo { + // Echo returns the request message or an error when it is empty. rpc Echo(EchoRequest) returns (EchoResponse); } diff --git a/internal/launcher/echo/v1/protogen/wire.gen.go b/internal/launcher/echo/v1/protogen/wire.gen.go index 250bb50..51c5ed1 100644 --- a/internal/launcher/echo/v1/protogen/wire.gen.go +++ b/internal/launcher/echo/v1/protogen/wire.gen.go @@ -24,6 +24,7 @@ const ( // proto-level shape of the point, not the point's Go interface: a contract // method returning a bare error returns an empty response message here. type EchoServer interface { + // Echo returns the request message or an error when it is empty. Echo(context.Context, *EchoRequest) (*EchoResponse, error) } @@ -57,6 +58,7 @@ func handleEcho(srv any, ctx context.Context, dec func(any) error, interceptor g // outside the framework - one calling a service an extension publishes on the // API socket - can reach it with a plain gRPC client. type EchoClient interface { + // Echo returns the request message or an error when it is empty. Echo(ctx context.Context, in *EchoRequest, opts ...grpc.CallOption) (*EchoResponse, error) } diff --git a/sdk/sdkapi/extension.proto b/sdk/sdkapi/extension.proto index 0a8603b..c237131 100644 --- a/sdk/sdkapi/extension.proto +++ b/sdk/sdkapi/extension.proto @@ -9,8 +9,19 @@ package moby.extension.runtime.v1; option go_package = "github.com/moby/extensions/sdk/sdkapi/protogen"; +// Extension is served by every out-of-process extension. +// +// The generator command names the gRPC service fully qualified because, unlike +// a point, there is no id to take the proto package from. Both halves are wire +// identifiers: extensions built against them are separate binaries, possibly +// built from an older tree, so neither may change without a new version. service Extension { + // Describe returns the extension's declaration. rpc Describe(DescribeRequest) returns (DescribeResponse); + // Initialize runs the extension's Init. The daemon calls it after connecting + // every extension, in dependency order, so that by the time an extension + // initializes its dependencies are already initialized and reachable over the + // callback channel. An Init error is returned as the RPC status. rpc Initialize(InitializeRequest) returns (InitializeResponse); } diff --git a/sdk/sdkapi/protogen/wire.gen.go b/sdk/sdkapi/protogen/wire.gen.go index 40d12b2..251efdf 100644 --- a/sdk/sdkapi/protogen/wire.gen.go +++ b/sdk/sdkapi/protogen/wire.gen.go @@ -22,7 +22,12 @@ const ( // proto-level shape of the service, not the handwritten Go interface: a contract // method returning a bare error returns an empty response message here. type ExtensionServer interface { + // Describe returns the extension's declaration. Describe(context.Context, *DescribeRequest) (*DescribeResponse, error) + // Initialize runs the extension's Init. The daemon calls it after connecting + // every extension, in dependency order, so that by the time an extension + // initializes its dependencies are already initialized and reachable over the + // callback channel. An Init error is returned as the RPC status. Initialize(context.Context, *InitializeRequest) (*InitializeResponse, error) } @@ -69,7 +74,12 @@ func handleInitialize(srv any, ctx context.Context, dec func(any) error, interce // ExtensionClient calls the contract's raw gRPC service. It is exported so a client // outside the framework can reach the service with a plain gRPC client. type ExtensionClient interface { + // Describe returns the extension's declaration. Describe(ctx context.Context, in *DescribeRequest, opts ...grpc.CallOption) (*DescribeResponse, error) + // Initialize runs the extension's Init. The daemon calls it after connecting + // every extension, in dependency order, so that by the time an extension + // initializes its dependencies are already initialized and reachable over the + // callback channel. An Init error is returned as the RPC status. Initialize(ctx context.Context, in *InitializeRequest, opts ...grpc.CallOption) (*InitializeResponse, error) } From 77cddb5982fd38ccdd4ca6d9325bba4131c19efa Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Pawe=C5=82=20Gronowski?= Date: Wed, 14 Oct 2026 17:16:54 +0200 Subject: [PATCH 2/7] stream: Add Sender and Receiver interfaces MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Paweł Gronowski --- stream/stream.go | 38 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 38 insertions(+) create mode 100644 stream/stream.go diff --git a/stream/stream.go b/stream/stream.go new file mode 100644 index 0000000..eac057d --- /dev/null +++ b/stream/stream.go @@ -0,0 +1,38 @@ +// SPDX-FileCopyrightText: Copyright The Moby Authors +// SPDX-License-Identifier: Apache-2.0 + +// Package stream defines typed, method-scoped streaming contracts. +// +// The owning method controls stream lifetime and its context controls +// cancellation. +package stream + +// Sender sends messages to the other side of a method-scoped stream. +// +// Calls to one Sender must be serialized. The caller must not mutate the +// message, or data reachable from it, until the owning method returns. +// Send must treat that data as read-only and may retain it until then. +// A nil message is invalid. +// +// A successful Send does not report final method success or durable storage. +// Implementations that block must cooperate with the context of the method +// that owns the stream. +type Sender[T any] interface { + Send(*T) error +} + +// Receiver receives messages from the other side of a method-scoped stream. +// +// Calls to one Receiver must be serialized. Recv returns [io.EOF] only for +// normal input completion. A successful call must return a non-nil message. +// Messages returned by Recv are read-only to the caller and must not be used +// after the method that owns the stream returns. +// +// Implementations that block must cooperate with the context of the method +// that owns the stream. +// The method cannot interrupt a blocked Recv in any other way. +// If the other side finishes early, for example with an error, the method +// sees that result only after Recv returns. +type Receiver[T any] interface { + Recv() (*T, error) +} From 6318f6a4cd74e0ef9736e4089ce68416d2ff5f78 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Pawe=C5=82=20Gronowski?= Date: Wed, 14 Oct 2026 17:16:54 +0200 Subject: [PATCH 3/7] stream: Add All and Copy iterator helpers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Paweł Gronowski --- stream/stream.go | 68 ++++++++++++ stream/stream_test.go | 234 ++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 302 insertions(+) create mode 100644 stream/stream_test.go diff --git a/stream/stream.go b/stream/stream.go index eac057d..e518a57 100644 --- a/stream/stream.go +++ b/stream/stream.go @@ -7,6 +7,12 @@ // cancellation. package stream +import ( + "errors" + "io" + "iter" +) + // Sender sends messages to the other side of a method-scoped stream. // // Calls to one Sender must be serialized. The caller must not mutate the @@ -36,3 +42,65 @@ type Sender[T any] interface { type Receiver[T any] interface { Recv() (*T, error) } + +var errNilMessage = errors.New("stream: nil message") + +// All adapts a Receiver to an iterator. +// +// Each successful message is yielded with a nil error. All stops at [io.EOF], +// including an error wrapping [io.EOF], without yielding the EOF. For any +// other receive error, All yields a nil message and that error once, then +// stops. +// +// A receiver that returns a nil message with a nil error is invalid. All +// reports that condition as an ordinary error and stops. +// +// If the yield callback returns false, All makes no more receive calls. +// Stopping the iterator does not cancel a receive that is already in progress; +// the caller must end or cancel the owning method. Neither All nor the stream +// interfaces can forcibly interrupt an interface callback. +func All[T any](receiver Receiver[T]) iter.Seq2[*T, error] { + return func(yield func(*T, error) bool) { + for { + message, err := receiver.Recv() + if err != nil { + if errors.Is(err, io.EOF) { + return + } + yield(nil, err) + return + } + if message == nil { + yield(nil, errNilMessage) + return + } + if !yield(message, nil) { + return + } + } + } +} + +// Copy sends each message from source through sender. +// +// Copy stops source and returns the first source error, nil-message error, or +// sender error. +// Use [All] to translate a receiver's normal EOF into iterator completion. +// +// Source must honor a false return from its yield callback. Copy can stop a +// source through that callback, but cannot forcibly interrupt source code that +// is already executing. +func Copy[T any](sender Sender[T], source iter.Seq2[*T, error]) error { + for message, err := range source { + if err != nil { + return err + } + if message == nil { + return errNilMessage + } + if err := sender.Send(message); err != nil { + return err + } + } + return nil +} diff --git a/stream/stream_test.go b/stream/stream_test.go new file mode 100644 index 0000000..f6369a5 --- /dev/null +++ b/stream/stream_test.go @@ -0,0 +1,234 @@ +// SPDX-FileCopyrightText: Copyright The Moby Authors +// SPDX-License-Identifier: Apache-2.0 + +package stream + +import ( + "errors" + "fmt" + "io" + "iter" + "testing" + + "gotest.tools/v3/assert" + is "gotest.tools/v3/assert/cmp" +) + +type testMessage struct { + Value int +} + +type testReceiver struct { + messages []*testMessage + terminalErr error + calls int +} + +func (r *testReceiver) Recv() (*testMessage, error) { + r.calls++ + if len(r.messages) == 0 { + return nil, r.terminalErr + } + message := r.messages[0] + r.messages = r.messages[1:] + return message, nil +} + +type testSender struct { + messages []*testMessage + err error + calls int +} + +func (s *testSender) Send(message *testMessage) error { + s.calls++ + if s.err != nil { + return s.err + } + s.messages = append(s.messages, message) + return nil +} + +var ( + _ Sender[testMessage] = (*testSender)(nil) + _ Receiver[testMessage] = (*testReceiver)(nil) +) + +func TestAllStopsAtEOF(t *testing.T) { + t.Parallel() + receiver := &testReceiver{ + messages: []*testMessage{{Value: 1}}, + terminalErr: fmt.Errorf("input complete: %w", io.EOF), + } + + var got []*testMessage + var gotErrs []error + for message, err := range All(receiver) { + if err != nil { + assert.Assert(t, is.Nil(message)) + gotErrs = append(gotErrs, err) + continue + } + got = append(got, message) + } + + assert.DeepEqual(t, got, []*testMessage{{Value: 1}}) + assert.Equal(t, len(gotErrs), 0) + assert.Equal(t, receiver.calls, 2) +} + +func TestAllPropagatesSourceError(t *testing.T) { + t.Parallel() + sourceErr := errors.New("source failed") + receiver := &testReceiver{ + messages: []*testMessage{{Value: 1}}, + terminalErr: sourceErr, + } + + var got []*testMessage + var gotErrs []error + for message, err := range All(receiver) { + if err != nil { + assert.Assert(t, is.Nil(message)) + gotErrs = append(gotErrs, err) + continue + } + got = append(got, message) + } + + assert.DeepEqual(t, got, []*testMessage{{Value: 1}}) + assert.Equal(t, len(gotErrs), 1) + assert.Assert(t, is.ErrorIs(gotErrs[0], sourceErr)) + assert.Equal(t, receiver.calls, 2) +} + +func TestAllStopsWhenIterationStops(t *testing.T) { + t.Parallel() + receiver := &testReceiver{ + messages: []*testMessage{{Value: 1}, {Value: 2}}, + } + + for message, err := range All(receiver) { + assert.Assert(t, is.Nil(err)) + assert.Equal(t, message.Value, 1) + break + } + + assert.Equal(t, receiver.calls, 1) +} + +func TestCopy(t *testing.T) { + t.Parallel() + t.Run("EOF is successful completion", func(t *testing.T) { + receiver := &testReceiver{ + messages: []*testMessage{{Value: 1}}, + terminalErr: fmt.Errorf("input complete: %w", io.EOF), + } + sender := &testSender{} + + err := Copy(sender, All(receiver)) + + assert.Assert(t, is.Nil(err)) + assert.DeepEqual(t, sender.messages, []*testMessage{{Value: 1}}) + assert.Equal(t, sender.calls, 1) + assert.Equal(t, receiver.calls, 2) + }) + + t.Run("source error is returned", func(t *testing.T) { + sourceErr := errors.New("source failed") + receiver := &testReceiver{ + messages: []*testMessage{{Value: 1}}, + terminalErr: sourceErr, + } + sender := &testSender{} + + err := Copy(sender, All(receiver)) + + assert.Assert(t, is.ErrorIs(err, sourceErr)) + assert.DeepEqual(t, sender.messages, []*testMessage{{Value: 1}}) + assert.Equal(t, sender.calls, 1) + assert.Equal(t, receiver.calls, 2) + }) + + t.Run("destination error is returned", func(t *testing.T) { + destinationErr := errors.New("destination failed") + receiver := &testReceiver{ + messages: []*testMessage{{Value: 1}, {Value: 2}}, + terminalErr: io.EOF, + } + sender := &testSender{err: destinationErr} + + err := Copy(sender, All(receiver)) + + assert.Assert(t, is.ErrorIs(err, destinationErr)) + assert.Equal(t, sender.calls, 1) + assert.Equal(t, receiver.calls, 1) + }) +} + +func TestCopyStopsSourceAfterDestinationError(t *testing.T) { + t.Parallel() + destinationErr := errors.New("destination failed") + sender := &testSender{err: destinationErr} + pulls := 0 + source := iter.Seq2[*testMessage, error](func(yield func(*testMessage, error) bool) { + for value := 1; value <= 3; value++ { + pulls++ + if !yield(&testMessage{Value: value}, nil) { + return + } + } + }) + + err := Copy(sender, source) + + assert.Assert(t, is.ErrorIs(err, destinationErr)) + assert.Equal(t, pulls, 1) + assert.Equal(t, sender.calls, 1) +} + +func TestHelpersRejectNilMessages(t *testing.T) { + t.Parallel() + t.Run("All", func(t *testing.T) { + receiver := &testReceiver{ + messages: []*testMessage{nil}, + terminalErr: io.EOF, + } + + var gotErrs []error + for message, err := range All(receiver) { + assert.Assert(t, is.Nil(message)) + gotErrs = append(gotErrs, err) + } + + assert.Equal(t, len(gotErrs), 1) + assert.Assert(t, is.ErrorContains(gotErrs[0], "nil message")) + assert.Equal(t, receiver.calls, 1) + }) + + t.Run("Copy", func(t *testing.T) { + sender := &testSender{} + source := iter.Seq2[*testMessage, error](func(yield func(*testMessage, error) bool) { + yield(nil, nil) + }) + + err := Copy(sender, source) + + assert.Assert(t, is.ErrorContains(err, "nil message")) + assert.Equal(t, sender.calls, 0) + }) +} + +func TestCopyPreservesIteratorErrors(t *testing.T) { + t.Parallel() + for _, sourceErr := range []error{io.EOF, errors.New("source failed")} { + sender := &testSender{} + stopped := false + source := iter.Seq2[*testMessage, error](func(yield func(*testMessage, error) bool) { + stopped = !yield(nil, sourceErr) + }) + assert.ErrorIs(t, Copy(sender, source), sourceErr) + assert.Assert(t, stopped) + assert.Equal(t, sender.calls, 0) + } +} From bd861887d5474d53292b50111c407c2e7d3b364e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Pawe=C5=82=20Gronowski?= Date: Wed, 14 Oct 2026 17:18:01 +0200 Subject: [PATCH 4/7] cmd/mobyextgen: List accepted method signatures when rejecting a method MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Contract methods with extra or missing parameters were accepted as long as the last parameter was a request pointer, and the generator emitted adapters that did not compile against the contract. The generator now classifies each parameter and accepts only methods with exactly two parameters, the second a request pointer, and an error or (*Reply, error) result. It rejects any other signature with one error that lists the accepted shapes. Signed-off-by: Paweł Gronowski --- cmd/mobyextgen/parse.go | 101 ++++++++++++++++++++++++++--------- cmd/mobyextgen/parse_test.go | 11 ++++ 2 files changed, 87 insertions(+), 25 deletions(-) diff --git a/cmd/mobyextgen/parse.go b/cmd/mobyextgen/parse.go index a5580c6..3dcc70b 100644 --- a/cmd/mobyextgen/parse.go +++ b/cmd/mobyextgen/parse.go @@ -256,32 +256,77 @@ func parseMethods(iface *ast.InterfaceType) ([]method, error) { continue } name := m.Names[0].Name - doc := fieldComment(m) - if ft.Params == nil || len(ft.Params.List) == 0 { - return nil, fmt.Errorf("method %q: expected a request parameter", name) - } - reqType := ft.Params.List[len(ft.Params.List)-1].Type - req, err := pointerIdent(reqType) - if err != nil { - return nil, fmt.Errorf("method %q request: %w", name, err) + paramFields := fieldsPerName(ft.Params) + params := make([]methodParameter, len(paramFields)) + for i, param := range paramFields { + var err error + params[i], err = classifyMethodParameter(param.Type, i) + if err != nil { + return nil, fmt.Errorf("method %q parameter %d: %w", name, i+1, err) + } } res := results(ft) + + doc := fieldComment(m) switch { - case len(res) == 1 && isIdent(res[0], "error"): - methods = append(methods, method{name: name, request: req, response: name + "Response", bareError: true, doc: doc}) - case len(res) == 2 && isIdent(res[1], "error"): - resp, err := pointerIdent(res[0]) - if err != nil { - return nil, fmt.Errorf("method %q response: %w", name, err) + case len(params) == 2 && params[0].kind == methodContextParameter && params[1].kind == methodMessageParameter: + var ( + response string + bareError bool + ) + switch { + case len(res) == 1 && isIdent(res[0], "error"): + response, bareError = name+"Response", true + case len(res) == 2 && isIdent(res[1], "error"): + var err error + response, err = pointerIdent(res[0]) + if err != nil { + return nil, methodShapeError(name) + } + default: + return nil, methodShapeError(name) } - methods = append(methods, method{name: name, request: req, response: resp, doc: doc}) + methods = append(methods, method{ + name: name, request: params[1].message, response: response, + bareError: bareError, doc: doc, + }) default: - return nil, fmt.Errorf("method %q: result must be `error` or `(*Resp, error)`", name) + return nil, methodShapeError(name) } } return methods, nil } +type methodParameterKind int + +const ( + methodContextParameter methodParameterKind = iota + 1 + methodMessageParameter +) + +type methodParameter struct { + kind methodParameterKind + message string +} + +const acceptedMethodShapes = "accepted shapes are M(context.Context, *Request) (*Reply, error) and " + + "M(context.Context, *Request) error" + +func methodShapeError(name string) error { + return fmt.Errorf("method %q: unsupported signature; %s", name, acceptedMethodShapes) +} + +func classifyMethodParameter(expr ast.Expr, index int) (methodParameter, error) { + if index == 0 { + return methodParameter{kind: methodContextParameter}, nil + } + message, err := pointerIdent(expr) + if err != nil { + return methodParameter{}, err + } + return methodParameter{kind: methodMessageParameter, message: message}, nil +} + func fieldComment(field *ast.Field) string { group := field.Doc if group == nil { @@ -514,17 +559,23 @@ func isIdent(expr ast.Expr, name string) bool { } func results(ft *ast.FuncType) []ast.Expr { - if ft.Results == nil { + var out []ast.Expr + for _, f := range fieldsPerName(ft.Results) { + out = append(out, f.Type) + } + return out +} + +// fieldsPerName repeats each field once per declared name, so a grouped +// declaration such as `a, b *T` counts as two entries. +func fieldsPerName(list *ast.FieldList) []*ast.Field { + if list == nil { return nil } - var out []ast.Expr - for _, f := range ft.Results.List { - if len(f.Names) == 0 { - out = append(out, f.Type) - continue - } - for range f.Names { - out = append(out, f.Type) + var out []*ast.Field + for _, f := range list.List { + for range max(1, len(f.Names)) { + out = append(out, f) } } return out diff --git a/cmd/mobyextgen/parse_test.go b/cmd/mobyextgen/parse_test.go index fcb52e1..e5eacd0 100644 --- a/cmd/mobyextgen/parse_test.go +++ b/cmd/mobyextgen/parse_test.go @@ -166,3 +166,14 @@ var Point = extensions.DefinePoint[Runtime]("example.api.v1") `, "example.api.v1.Runtime") assert.ErrorContains(t, err, "cannot be used with an ordinary Point") } + +func TestRejectsGroupedExtraParameter(t *testing.T) { + _, err := parseSource(t, `package p +import "github.com/moby/extensions" +type Req struct{} +type Resp struct{} +type S interface{ Do(ctx interface{}, first, second *Req) (*Resp, error) } +var Point = extensions.DefinePoint[S]("test.gen.v1") +`) + assert.ErrorContains(t, err, "unsupported signature") +} From 853e848e0ff0b845210d305bfab8b600be78f3ee Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Pawe=C5=82=20Gronowski?= Date: Wed, 14 Oct 2026 17:20:58 +0200 Subject: [PATCH 5/7] cmd/mobyextgen: Generate server-streaming RPCs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Contracts can declare server-streaming methods with this shape: M(ctx, *Req, stream.Sender[T]) error The generator emits them as protobuf server-streaming RPCs and adapts the gRPC stream to stream.Sender on both sides. Stream messages must be local, non-generic structs. A provider that sends a nil message gets codes.Internal. A runtime test generates fixture contracts and runs their generated code over a real gRPC connection and directly in process. Signed-off-by: Paweł Gronowski --- cmd/mobyextgen/descriptor.go | 8 +- cmd/mobyextgen/model.go | 10 +- cmd/mobyextgen/parse.go | 169 +++++++- cmd/mobyextgen/proto.go | 10 +- cmd/mobyextgen/runtime_norace_test.go | 8 + cmd/mobyextgen/runtime_race_test.go | 8 + cmd/mobyextgen/runtime_test.go | 104 +++++ cmd/mobyextgen/streaming_test.go | 196 +++++++++ .../testdata/runtime_behavior_test.go.txt | 390 ++++++++++++++++++ .../testdata/runtime_contract.go.txt | 33 ++ .../testdata/runtime_server_contract.go.txt | 22 + .../testdata/runtime_unary_contract.go.txt | 18 + cmd/mobyextgen/wire.go | 154 ++++++- docs/AUTHORING.md | 5 + docs/DESIGN.md | 41 ++ 15 files changed, 1151 insertions(+), 25 deletions(-) create mode 100644 cmd/mobyextgen/runtime_norace_test.go create mode 100644 cmd/mobyextgen/runtime_race_test.go create mode 100644 cmd/mobyextgen/runtime_test.go create mode 100644 cmd/mobyextgen/streaming_test.go create mode 100644 cmd/mobyextgen/testdata/runtime_behavior_test.go.txt create mode 100644 cmd/mobyextgen/testdata/runtime_contract.go.txt create mode 100644 cmd/mobyextgen/testdata/runtime_server_contract.go.txt create mode 100644 cmd/mobyextgen/testdata/runtime_unary_contract.go.txt diff --git a/cmd/mobyextgen/descriptor.go b/cmd/mobyextgen/descriptor.go index 813bfd2..7e266c2 100644 --- a/cmd/mobyextgen/descriptor.go +++ b/cmd/mobyextgen/descriptor.go @@ -80,11 +80,15 @@ func fileDescriptor(pt point) (*descriptorpb.FileDescriptorProto, error) { svc := &descriptorpb.ServiceDescriptorProto{Name: new(pt.service)} for _, m := range pt.methods { - svc.Method = append(svc.Method, &descriptorpb.MethodDescriptorProto{ + method := &descriptorpb.MethodDescriptorProto{ Name: new(m.name), InputType: new("." + pt.id + "." + m.request), OutputType: new("." + pt.id + "." + m.response), - }) + } + if m.kind == methodServerStreaming { + method.ServerStreaming = new(true) + } + svc.Method = append(svc.Method, method) } fd.Service = []*descriptorpb.ServiceDescriptorProto{svc} return fd, nil diff --git a/cmd/mobyextgen/model.go b/cmd/mobyextgen/model.go index 0328b12..0694c6e 100644 --- a/cmd/mobyextgen/model.go +++ b/cmd/mobyextgen/model.go @@ -25,13 +25,21 @@ func (p point) grpcService() string { return p.id + "." + p.service } func (p point) protogenImport() string { return p.importPath + "/protogen" } type method struct { + kind methodKind name string // method/rpc name request string // request message name response string // response message name - bareError bool // true for `M(ctx, *Req) error`; response is a generated empty message + bareError bool // true only for legacy unary `M(ctx, *Req) error` doc string // method documentation } +type methodKind int + +const ( + methodUnary methodKind = iota + methodServerStreaming +) + type message struct { name string fields []field diff --git a/cmd/mobyextgen/parse.go b/cmd/mobyextgen/parse.go index 3dcc70b..b394a91 100644 --- a/cmd/mobyextgen/parse.go +++ b/cmd/mobyextgen/parse.go @@ -10,6 +10,7 @@ import ( "go/parser" "go/token" "os" + "path" "reflect" "sort" "strconv" @@ -20,6 +21,8 @@ import ( // parsing +const streamImportPath = "github.com/moby/extensions/stream" + func parsePoint(dir string) (point, error) { return parseContract(dir, "") } @@ -99,7 +102,7 @@ func parseContract(dir, serviceIdentity string) (point, error) { if pt.licenseHeader != "" { pt.licenseHeader += "\n" } - pt.methods, err = parseMethods(ifaceType) + pt.methods, err = parseMethods(files, ifaceType) if err != nil { return point{}, err } @@ -248,7 +251,7 @@ func interfaceComment(files []*ast.File, name string) string { return "" } -func parseMethods(iface *ast.InterfaceType) ([]method, error) { +func parseMethods(files []*ast.File, iface *ast.InterfaceType) ([]method, error) { var methods []method for _, m := range iface.Methods.List { ft, ok := m.Type.(*ast.FuncType) @@ -260,12 +263,17 @@ func parseMethods(iface *ast.InterfaceType) ([]method, error) { params := make([]methodParameter, len(paramFields)) for i, param := range paramFields { var err error - params[i], err = classifyMethodParameter(param.Type, i) + params[i], err = classifyMethodParameter(files, param.Type, i) if err != nil { return nil, fmt.Errorf("method %q parameter %d: %w", name, i+1, err) } } res := results(ft) + for _, result := range res { + if _, isStream, _ := parseStreamParameter(files, result); isStream { + return nil, methodShapeError(name) + } + } doc := fieldComment(m) switch { @@ -290,6 +298,15 @@ func parseMethods(iface *ast.InterfaceType) ([]method, error) { name: name, request: params[1].message, response: response, bareError: bareError, doc: doc, }) + case len(params) == 3 && params[0].kind == methodContextParameter && + params[1].kind == methodMessageParameter && params[2].kind == methodSenderParameter: + if len(res) != 1 || !isIdent(res[0], "error") { + return nil, methodShapeError(name) + } + methods = append(methods, method{ + kind: methodServerStreaming, name: name, + request: params[1].message, response: params[2].message, doc: doc, + }) default: return nil, methodShapeError(name) } @@ -302,6 +319,7 @@ type methodParameterKind int const ( methodContextParameter methodParameterKind = iota + 1 methodMessageParameter + methodSenderParameter ) type methodParameter struct { @@ -309,15 +327,24 @@ type methodParameter struct { message string } -const acceptedMethodShapes = "accepted shapes are M(context.Context, *Request) (*Reply, error) and " + - "M(context.Context, *Request) error" +const acceptedMethodShapes = "accepted shapes are M(context.Context, *Request) (*Reply, error), " + + "M(context.Context, *Request) error, and " + + "M(context.Context, *Request, stream.Sender[Chunk]) error" func methodShapeError(name string) error { return fmt.Errorf("method %q: unsupported signature; %s", name, acceptedMethodShapes) } -func classifyMethodParameter(expr ast.Expr, index int) (methodParameter, error) { - if index == 0 { +func classifyMethodParameter(files []*ast.File, expr ast.Expr, index int) (methodParameter, error) { + param, isStream, err := parseStreamParameter(files, expr) + switch { + case err != nil: + return methodParameter{}, err + case isStream && index == 0: + return methodParameter{}, errors.New("stream parameters cannot replace context.Context") + case isStream: + return param, nil + case index == 0: return methodParameter{kind: methodContextParameter}, nil } message, err := pointerIdent(expr) @@ -327,6 +354,134 @@ func classifyMethodParameter(expr ast.Expr, index int) (methodParameter, error) return methodParameter{kind: methodMessageParameter, message: message}, nil } +// parseStreamParameter reports whether expr names stream.Sender and, if so, +// validates its import and message argument. +func parseStreamParameter(files []*ast.File, expr ast.Expr) (methodParameter, bool, error) { + var ( + selector *ast.SelectorExpr + args []ast.Expr + ) + switch expr := expr.(type) { + case *ast.SelectorExpr: + selector = expr + case *ast.IndexExpr: + selector, _ = expr.X.(*ast.SelectorExpr) + args = []ast.Expr{expr.Index} + case *ast.IndexListExpr: + selector, _ = expr.X.(*ast.SelectorExpr) + args = expr.Indices + default: + return methodParameter{}, false, nil + } + if selector == nil || selector.Sel.Name != "Sender" { + return methodParameter{}, false, nil + } + use := "stream." + selector.Sel.Name + file := fileForExpr(files, expr) + if file == nil { + return methodParameter{}, true, fmt.Errorf("%s must be imported from %q", use, streamImportPath) + } + if err := resolveStreamImport(file, selector.X, use); err != nil { + return methodParameter{}, true, err + } + if len(args) != 1 { + return methodParameter{}, true, fmt.Errorf("%s must have exactly one local message type argument", use) + } + message, err := streamMessageArgument(args[0]) + if err != nil { + return methodParameter{}, true, fmt.Errorf("%s: %w", use, err) + } + if err := validateStreamMessage(files, message); err != nil { + return methodParameter{}, true, fmt.Errorf("%s message %q: %w", use, message, err) + } + return methodParameter{kind: methodSenderParameter, message: message}, true, nil +} + +func fileForExpr(files []*ast.File, expr ast.Expr) *ast.File { + for _, file := range files { + if file.Pos() <= expr.Pos() && expr.End() <= file.End() { + return file + } + } + return nil +} + +func resolveStreamImport(file *ast.File, packageExpr ast.Expr, use string) error { + ident, ok := packageExpr.(*ast.Ident) + if !ok { + return fmt.Errorf("%s must use a package imported from %q", use, streamImportPath) + } + for _, spec := range file.Imports { + importPath, err := strconv.Unquote(spec.Path.Value) + if err != nil { + continue + } + name := path.Base(importPath) + if spec.Name != nil { + name = spec.Name.Name + } + if name != ident.Name { + continue + } + if importPath != streamImportPath { + return fmt.Errorf("%s must be imported from %q", use, streamImportPath) + } + return nil + } + return fmt.Errorf("%s package name %q is not imported from %q", use, ident.Name, streamImportPath) +} + +func streamMessageArgument(expr ast.Expr) (string, error) { + if ident, ok := expr.(*ast.Ident); ok { + return ident.Name, nil + } + switch expr.(type) { + case *ast.StarExpr: + return "", errors.New("message type argument must not be a pointer") + case *ast.SelectorExpr: + return "", errors.New("imported message types are unsupported; use a local struct message") + case *ast.IndexExpr, *ast.IndexListExpr: + return "", errors.New("message type argument must be a non-generic local named type") + default: + return "", errors.New("message type argument must be a local named type") + } +} + +func validateStreamMessage(files []*ast.File, name string) error { + var found *ast.TypeSpec + for _, file := range files { + for _, decl := range file.Decls { + group, ok := decl.(*ast.GenDecl) + if !ok || group.Tok != token.TYPE { + continue + } + for _, spec := range group.Specs { + typeSpec := spec.(*ast.TypeSpec) + if typeSpec.Name.Name != name { + continue + } + if found != nil { + return errors.New("is ambiguous; declare one local struct message") + } + found = typeSpec + } + } + } + if found == nil { + return errors.New("is unresolved") + } + if found.Assign.IsValid() { + return errors.New("is an alias") + } + if found.TypeParams != nil && len(found.TypeParams.List) != 0 { + return errors.New("must not be generic") + } + if _, ok := found.Type.(*ast.StructType); !ok { + return errors.New("must be a local struct declaration") + } + return nil +} + func fieldComment(field *ast.Field) string { group := field.Doc if group == nil { diff --git a/cmd/mobyextgen/proto.go b/cmd/mobyextgen/proto.go index 20b6d15..26f561f 100644 --- a/cmd/mobyextgen/proto.go +++ b/cmd/mobyextgen/proto.go @@ -25,7 +25,15 @@ func emitProto(pt point) ([]byte, error) { fmt.Fprintf(&b, "service %s {\n", pt.service) for _, m := range pt.methods { emitComment(&b, " ", m.doc) - fmt.Fprintf(&b, " rpc %s(%s) returns (%s);\n", m.name, m.request, m.response) + request, response := m.request, m.response + switch m.kind { + case methodUnary: + case methodServerStreaming: + response = "stream " + response + default: + return nil, fmt.Errorf("%s: unknown method kind %d", m.name, m.kind) + } + fmt.Fprintf(&b, " rpc %s(%s) returns (%s);\n", m.name, request, response) } fmt.Fprintln(&b, "}") diff --git a/cmd/mobyextgen/runtime_norace_test.go b/cmd/mobyextgen/runtime_norace_test.go new file mode 100644 index 0000000..78bfe01 --- /dev/null +++ b/cmd/mobyextgen/runtime_norace_test.go @@ -0,0 +1,8 @@ +// SPDX-FileCopyrightText: Copyright The Moby Authors +// SPDX-License-Identifier: Apache-2.0 + +//go:build !race + +package main + +const runtimeRace = false diff --git a/cmd/mobyextgen/runtime_race_test.go b/cmd/mobyextgen/runtime_race_test.go new file mode 100644 index 0000000..f99c688 --- /dev/null +++ b/cmd/mobyextgen/runtime_race_test.go @@ -0,0 +1,8 @@ +// SPDX-FileCopyrightText: Copyright The Moby Authors +// SPDX-License-Identifier: Apache-2.0 + +//go:build race + +package main + +const runtimeRace = true diff --git a/cmd/mobyextgen/runtime_test.go b/cmd/mobyextgen/runtime_test.go new file mode 100644 index 0000000..56ae931 --- /dev/null +++ b/cmd/mobyextgen/runtime_test.go @@ -0,0 +1,104 @@ +// SPDX-FileCopyrightText: Copyright The Moby Authors +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "os" + "os/exec" + "path" + "path/filepath" + "strings" + "testing" + + "gotest.tools/v3/assert" +) + +func TestGeneratedStreamingRuntime(t *testing.T) { + if testing.Short() { + t.Skip("builds and tests generated packages") + } + t.Parallel() + root := repositoryRoot(t) + // Use a directory inside the module because this package has no standalone + // go.mod; this resolves the current checkout's vendor tree. Under testdata, + // repository-wide ./... patterns skip it even if cleanup does not run. + dir, err := os.MkdirTemp(filepath.Join(root, "cmd", "mobyextgen", "testdata"), "runtime-") //nolint:usetesting // t.TempDir is outside the module. + assert.NilError(t, err) + t.Cleanup(func() { _ = os.RemoveAll(dir) }) + + writeFixture(t, filepath.Join(dir, "contract.go"), "runtime_contract.go.txt") + importPath, _, err := locate(dir) + assert.NilError(t, err) + behavior := strings.NewReplacer( + "__CONTRACT_IMPORT__", importPath, + "__PROTOGEN_IMPORT__", path.Join(importPath, "protogen"), + ).Replace(readFixture(t, "runtime_behavior_test.go.txt")) + assert.NilError(t, os.WriteFile(filepath.Join(dir, "runtime_test.go"), []byte(behavior), 0o644)) + assert.NilError(t, run(dir, "")) + + for _, fixture := range []struct { + name string + source string + id string + }{ + { + name: "unary", + source: "runtime_unary_contract.go.txt", + id: "org.example.generated.unary.v1.Service", + }, + { + name: "serverstream", + source: "runtime_server_contract.go.txt", + id: "org.example.generated.serverstream.v1.Service", + }, + } { + subdir := filepath.Join(dir, fixture.name) + assert.NilError(t, os.MkdirAll(subdir, 0o755)) + writeFixture(t, filepath.Join(subdir, "contract.go"), fixture.source) + assert.NilError(t, run(subdir, fixture.id)) + } + + args := []string{"test"} + if _, err := os.Stat(filepath.Join(root, "vendor", "modules.txt")); err == nil { + args = append(args, "-mod=vendor") + } + if runtimeRace { + args = append(args, "-race") + } + args = append(args, "./...") + cmd := exec.Command("go", args...) + cmd.Dir = dir + output, err := cmd.CombinedOutput() + if err != nil { + t.Fatalf("generated runtime test failed: %v\n%s", err, output) + } +} + +func repositoryRoot(t *testing.T) string { + t.Helper() + root, err := filepath.Abs(".") + assert.NilError(t, err) + for { + if _, err := os.Stat(filepath.Join(root, "go.mod")); err == nil { + return root + } + parent := filepath.Dir(root) + if parent == root { + t.Fatal("go.mod not found above generator package") + } + root = parent + } +} + +func readFixture(t *testing.T, name string) string { + t.Helper() + content, err := os.ReadFile(filepath.Join("testdata", name)) + assert.NilError(t, err) + return string(content) +} + +func writeFixture(t *testing.T, destination, name string) { + t.Helper() + assert.NilError(t, os.WriteFile(destination, []byte(readFixture(t, name)), 0o644)) +} diff --git a/cmd/mobyextgen/streaming_test.go b/cmd/mobyextgen/streaming_test.go new file mode 100644 index 0000000..3601065 --- /dev/null +++ b/cmd/mobyextgen/streaming_test.go @@ -0,0 +1,196 @@ +// SPDX-FileCopyrightText: Copyright The Moby Authors +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "strings" + "testing" + + "gotest.tools/v3/assert" + is "gotest.tools/v3/assert/cmp" +) + +func streamingPointSource(body string) string { + return `package contract + +import ( + stdctx "context" + ext "github.com/moby/extensions" + s "github.com/moby/extensions/stream" +) +` + body + ` +var Point = ext.DefinePoint[Service]("org.example.contract.v1") +` +} + +func TestParseContractStreamingContracts(t *testing.T) { + t.Parallel() + source := streamingPointSource(` +// Service documents its streaming operations. +type Service interface { + // Download sends chunks to the caller. + Download(stdctx.Context, *DownloadRequest, s.Sender[DownloadChunk]) error +} + +type DownloadRequest struct{} +type DownloadChunk struct{} +`) + point, err := parseSource(t, source) + assert.NilError(t, err) + assert.Equal(t, point.doc, "Service documents its streaming operations.") + assert.Equal(t, len(point.methods), 1) + assert.Equal(t, point.methods[0].kind, methodServerStreaming) + assert.Equal(t, point.methods[0].request, "DownloadRequest") + assert.Equal(t, point.methods[0].response, "DownloadChunk") + assert.Equal(t, point.methods[0].bareError, false) + assert.Equal(t, point.methods[0].doc, "Download sends chunks to the caller.") + messageNames := make(map[string]bool, len(point.messages)) + for _, message := range point.messages { + messageNames[message.name] = true + } + for _, name := range []string{"DownloadRequest", "DownloadChunk"} { + assert.Assert(t, messageNames[name], "missing message %s", name) + } +} + +func TestParseContractRejectsInvalidStreamingContracts(t *testing.T) { + t.Parallel() + tests := []struct { + name string + source string + want string + }{ + { + name: "server stream with a reply", + source: streamingPointSource(`type Service interface { + Download(stdctx.Context, *Request, s.Sender[Chunk]) (*Request, error) +} +type Request struct{} +type Chunk struct{} +`), + want: acceptedMethodShapes, + }, + { + name: "stream in a result", + source: streamingPointSource(`type Service interface { + Download(stdctx.Context, *Request) (s.Sender[Chunk], error) +} +type Request struct{} +type Chunk struct{} +`), + want: acceptedMethodShapes, + }, + { + name: "pointer stream message argument", + source: streamingPointSource(`type Service interface { + Download(stdctx.Context, *Request, s.Sender[*Chunk]) error +} +type Request struct{} +type Chunk struct{} +`), + want: "must not be a pointer", + }, + { + name: "imported stream message argument", + source: `package contract + +import ( + stdctx "context" + other "example/other" + ext "github.com/moby/extensions" + s "github.com/moby/extensions/stream" +) +type Service interface { + Download(stdctx.Context, *Request, s.Sender[other.Chunk]) error +} +type Request struct{} +var Point = ext.DefinePoint[Service]("org.example.contract.v1") +`, + want: "imported message types", + }, + { + name: "wrong stream import", + source: `package contract + +import ( + stdctx "context" + ext "github.com/moby/extensions" + s "example/stream" +) +type Service interface { + Download(stdctx.Context, *Request, s.Sender[Chunk]) error +} +type Request struct{} +type Chunk struct{} +var Point = ext.DefinePoint[Service]("org.example.contract.v1") +`, + want: "must be imported from", + }, + { + name: "generic stream message", + source: streamingPointSource(`type Service interface { + Download(stdctx.Context, *Request, s.Sender[Chunk]) error +} +type Request struct{} +type Chunk[T any] struct{} +`), + want: "must not be generic", + }, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + t.Parallel() + _, err := parseSource(t, test.source) + assert.Assert(t, is.ErrorContains(err, test.want)) + }) + } +} + +func TestStreamingProtoDescriptorAndWire(t *testing.T) { + t.Parallel() + source := streamingPointSource(` +// Service documents its streaming operations. +type Service interface { + // Download sends chunks to the caller. + Download(stdctx.Context, *DownloadRequest, s.Sender[DownloadChunk]) error +} + +type DownloadRequest struct{} +type DownloadChunk struct{} +`) + point, err := parseSource(t, source) + assert.NilError(t, err) + point.importPath = "example.test/contract" + point.protoPath = "contract/service.proto" + + protoFile, err := emitProto(point) + assert.NilError(t, err) + protoText := string(protoFile) + for _, declaration := range []string{ + "rpc Download(DownloadRequest) returns (stream DownloadChunk);", + } { + assert.Assert(t, strings.Contains(protoText, declaration), declaration) + } + + fd, err := fileDescriptor(point) + assert.NilError(t, err) + assert.Equal(t, len(fd.GetService()), 1) + methods := fd.GetService()[0].GetMethod() + assert.Equal(t, len(methods), 1) + assert.Equal(t, methods[0].GetName(), "Download") + assert.Equal(t, methods[0].GetServerStreaming(), true) + assert.Equal(t, methods[0].GetClientStreaming(), false) + assert.Assert(t, methods[0].ClientStreaming == nil) + + wire, err := emitWire(point) + assert.NilError(t, err) + wireText := string(wire) + for _, declaration := range []string{ + "Download(*DownloadRequest, grpc.ServerStreamingServer[DownloadChunk]) error", + "Streams: []grpc.StreamDesc{", + "ServerStreams: true", + } { + assert.Assert(t, strings.Contains(wireText, declaration), declaration) + } +} diff --git a/cmd/mobyextgen/testdata/runtime_behavior_test.go.txt b/cmd/mobyextgen/testdata/runtime_behavior_test.go.txt new file mode 100644 index 0000000..33d143a --- /dev/null +++ b/cmd/mobyextgen/testdata/runtime_behavior_test.go.txt @@ -0,0 +1,390 @@ +// SPDX-FileCopyrightText: Copyright The Moby Authors +// SPDX-License-Identifier: Apache-2.0 + +package contract_test + +import ( + "context" + "errors" + "net" + "sync" + "testing" + "time" + + contract "__CONTRACT_IMPORT__" + protogen "__PROTOGEN_IMPORT__" + + "github.com/moby/extensions/stream" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/status" + "google.golang.org/protobuf/types/known/wrapperspb" + "gotest.tools/v3/assert" +) + +type provider struct { + downloadValues []int32 + downloadNil bool + downloadErr error + downloadWait <-chan struct{} + downloadStart chan struct{} + downloadDone chan struct{} + + downloadStartOnce sync.Once + downloadDoneOnce sync.Once + + unaryErr error + legacyErr error +} + +var _ contract.Service = (*provider)(nil) + +func (p *provider) Unary(_ context.Context, req *contract.Request) (*contract.Reply, error) { + if p.unaryErr != nil { + return nil, p.unaryErr + } + return &contract.Reply{Values: []int32{req.Value}}, nil +} + +func (p *provider) Remove(context.Context, *contract.Request) error { + return p.legacyErr +} + +func (p *provider) Download(ctx context.Context, _ *contract.Request, sender stream.Sender[contract.Chunk]) error { + return p.download(ctx, sender) +} + +func (p *provider) DownloadAgain(ctx context.Context, _ *contract.Request, sender stream.Sender[contract.Chunk]) error { + return p.download(ctx, sender) +} + +func (p *provider) download(ctx context.Context, sender stream.Sender[contract.Chunk]) error { + p.signalDownloadStart() + defer p.signalDownloadDone() + if p.downloadNil { + return sender.Send(nil) + } + for _, value := range p.downloadValues { + message := &contract.Chunk{Value: value} + if err := sender.Send(message); err != nil { + return err + } + } + if p.downloadWait != nil { + select { + case <-p.downloadWait: + case <-ctx.Done(): + return ctx.Err() + } + } + return p.downloadErr +} + +func (p *provider) signalDownloadStart() { + if p.downloadStart != nil { + p.downloadStartOnce.Do(func() { close(p.downloadStart) }) + } +} + +func (p *provider) signalDownloadDone() { + if p.downloadDone != nil { + p.downloadDoneOnce.Do(func() { close(p.downloadDone) }) + } +} + +type runtimeMode struct { + name string + new func(*testing.T, *provider) contract.Service +} + +func runtimeModes() []runtimeMode { + return []runtimeMode{ + { + name: "direct", + new: func(_ *testing.T, p *provider) contract.Service { + return p + }, + }, + { + name: "remote", + new: func(t *testing.T, p *provider) contract.Service { + return start(t, p) + }, + }, + } +} + +func start(t *testing.T, p *provider) contract.Service { + t.Helper() + listener, err := net.Listen("tcp", "127.0.0.1:0") + assert.NilError(t, err) + server := grpc.NewServer() + protogen.ServerPoint.Register(server, p) + serveDone := make(chan error, 1) + go func() { serveDone <- server.Serve(listener) }() + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + conn, err := grpc.DialContext( + ctx, + listener.Addr().String(), + grpc.WithTransportCredentials(insecure.NewCredentials()), + grpc.WithBlock(), + ) + cancel() + if err != nil { + server.Stop() + <-serveDone + t.Fatal(err) + } + t.Cleanup(func() { + _ = conn.Close() + server.Stop() + <-serveDone + }) + return protogen.NewClient(conn) +} + +func await(t *testing.T, signal <-chan struct{}) { + t.Helper() + timer := time.NewTimer(10 * time.Second) + defer timer.Stop() + select { + case <-signal: + case <-timer.C: + t.Fatal("timed out waiting for handler cleanup") + } +} + +type release struct { + ch chan struct{} + once sync.Once +} + +func newRelease() *release { + return &release{ch: make(chan struct{})} +} + +func (r *release) Close() { + r.once.Do(func() { close(r.ch) }) +} + +func asyncErrorCall(t *testing.T, call func() error) <-chan error { + t.Helper() + result := make(chan error, 1) + done := make(chan struct{}) + go func() { + result <- call() + close(done) + }() + t.Cleanup(func() { + timer := time.NewTimer(10 * time.Second) + defer timer.Stop() + select { + case <-done: + case <-timer.C: + t.Errorf("timed out waiting for asynchronous call") + } + }) + return result +} + +type sink struct { + values []int32 + err error + mu sync.Mutex +} + +func (s *sink) Send(message *contract.Chunk) error { + if s.err != nil { + return s.err + } + s.mu.Lock() + s.values = append(s.values, message.Value) + s.mu.Unlock() + return nil +} + +func detailError(code codes.Code, message, reason string) error { + st, err := status.New(code, message).WithDetails(wrapperspb.String(reason)) + if err != nil { + panic(err) + } + return st.Err() +} + +func assertDetail(t *testing.T, err error, code codes.Code, message, reason string) { + t.Helper() + st, ok := status.FromError(err) + assert.Assert(t, ok) + assert.Equal(t, st.Code(), code) + assert.Equal(t, st.Message(), message) + details := st.Details() + assert.Equal(t, len(details), 1) + info, ok := details[0].(*wrapperspb.StringValue) + assert.Assert(t, ok) + assert.Equal(t, info.Value, reason) +} + +func TestUnaryAndBareError(t *testing.T) { + t.Parallel() + p := &provider{} + client := start(t, p) + reply, err := client.Unary(context.Background(), &contract.Request{Value: 7}) + assert.NilError(t, err) + assert.DeepEqual(t, reply.Values, []int32{7}) + assert.NilError(t, client.Remove(context.Background(), &contract.Request{})) + + p.unaryErr = errors.Join(errors.New("wrapped cancellation"), context.Canceled) + _, err = client.Unary(context.Background(), &contract.Request{}) + assert.Equal(t, status.Code(err), codes.Canceled) +} + +func TestDirectAndRemoteParity(t *testing.T) { + for _, mode := range runtimeModes() { + t.Run(mode.name, func(t *testing.T) { + t.Parallel() + t.Run("empty and multiple downloads", func(t *testing.T) { + p := &provider{} + client := mode.new(t, p) + got := &sink{} + assert.NilError(t, client.Download(context.Background(), &contract.Request{}, got)) + assert.Equal(t, len(got.values), 0) + + p = &provider{downloadValues: []int32{1, 2, 3}} + client = mode.new(t, p) + got = &sink{} + assert.NilError(t, client.Download(context.Background(), &contract.Request{}, got)) + assert.DeepEqual(t, got.values, []int32{1, 2, 3}) + + p = &provider{downloadValues: []int32{4, 5}} + client = mode.new(t, p) + got = &sink{} + assert.NilError(t, client.DownloadAgain(context.Background(), &contract.Request{}, got)) + assert.DeepEqual(t, got.values, []int32{4, 5}) + }) + + t.Run("final status follows delivered download data", func(t *testing.T) { + p := &provider{ + downloadValues: []int32{10, 20}, + downloadErr: detailError(codes.FailedPrecondition, "final rejection", "FINAL_REJECTION"), + } + client := mode.new(t, p) + got := &sink{} + err := client.Download(context.Background(), &contract.Request{}, got) + assert.DeepEqual(t, got.values, []int32{10, 20}) + assertDetail(t, err, codes.FailedPrecondition, "final rejection", "FINAL_REJECTION") + }) + }) + } +} + +func TestServerStreamingInvalidProviderMessage(t *testing.T) { + t.Parallel() + p := &provider{downloadNil: true} + client := start(t, p) + err := client.Download(context.Background(), &contract.Request{}, &sink{}) + assert.Equal(t, status.Code(err), codes.Internal) +} + +func TestDirectAndRemoteErrorIdentity(t *testing.T) { + for _, mode := range runtimeModes() { + t.Run(mode.name, func(t *testing.T) { + t.Parallel() + t.Run("destination error is local and exact", func(t *testing.T) { + p := &provider{ + downloadValues: []int32{1}, + downloadStart: make(chan struct{}), + downloadDone: make(chan struct{}), + } + client := mode.new(t, p) + destinationErr := errors.New("destination failed") + result := asyncErrorCall(t, func() error { + return client.Download(context.Background(), &contract.Request{}, &sink{err: destinationErr}) + }) + var started bool + select { + case <-p.downloadStart: + started = true + case err := <-result: + assert.Equal(t, err, destinationErr) + } + if started { + assert.Equal(t, <-result, destinationErr) + await(t, p.downloadDone) + } + }) + }) + } +} + +func TestServerStreamingCallbacksAndCancellation(t *testing.T) { + t.Parallel() + t.Run("canceled transport cleans up handler", func(t *testing.T) { + release := newRelease() + p := &provider{ + downloadWait: release.ch, + downloadStart: make(chan struct{}), + downloadDone: make(chan struct{}), + } + client := start(t, p) + ctx, cancel := context.WithCancel(context.Background()) + result := asyncErrorCall(t, func() error { + return client.Download(ctx, &contract.Request{}, &sink{}) + }) + t.Cleanup(cancel) + t.Cleanup(release.Close) + await(t, p.downloadStart) + cancel() + assert.Equal(t, status.Code(<-result), codes.Canceled) + await(t, p.downloadDone) + }) + + t.Run("deadline does not require an unstarted handler", func(t *testing.T) { + release := newRelease() + p := &provider{ + downloadWait: release.ch, + downloadStart: make(chan struct{}), + downloadDone: make(chan struct{}), + } + client := start(t, p) + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond) + result := asyncErrorCall(t, func() error { + return client.Download(ctx, &contract.Request{}, &sink{}) + }) + t.Cleanup(cancel) + t.Cleanup(release.Close) + + var started bool + select { + case <-p.downloadStart: + started = true + case err := <-result: + assert.Equal(t, status.Code(err), codes.DeadlineExceeded) + } + if started { + assert.Equal(t, status.Code(<-result), codes.DeadlineExceeded) + await(t, p.downloadDone) + } + }) +} + +func TestDirectContextCancellation(t *testing.T) { + t.Parallel() + release := newRelease() + p := &provider{ + downloadWait: release.ch, + downloadStart: make(chan struct{}), + downloadDone: make(chan struct{}), + } + ctx, cancel := context.WithCancel(context.Background()) + result := asyncErrorCall(t, func() error { + return p.Download(ctx, &contract.Request{}, &sink{}) + }) + t.Cleanup(cancel) + t.Cleanup(release.Close) + await(t, p.downloadStart) + cancel() + assert.Equal(t, <-result, context.Canceled) + await(t, p.downloadDone) +} diff --git a/cmd/mobyextgen/testdata/runtime_contract.go.txt b/cmd/mobyextgen/testdata/runtime_contract.go.txt new file mode 100644 index 0000000..8b71d39 --- /dev/null +++ b/cmd/mobyextgen/testdata/runtime_contract.go.txt @@ -0,0 +1,33 @@ +// SPDX-FileCopyrightText: Copyright The Moby Authors +// SPDX-License-Identifier: Apache-2.0 + +package contract + +import ( + "context" + + ext "github.com/moby/extensions" + stream "github.com/moby/extensions/stream" +) + +type Service interface { + Unary(context.Context, *Request) (*Reply, error) + Remove(context.Context, *Request) error + Download(context.Context, *Request, stream.Sender[Chunk]) error + DownloadAgain(context.Context, *Request, stream.Sender[Chunk]) error +} + +type Request struct { + Value int32 `pb:"1"` +} + +type Chunk struct { + Value int32 `pb:"1"` + Data []byte `pb:"2"` +} + +type Reply struct { + Values []int32 `pb:"1"` +} + +var Point = ext.DefinePoint[Service]("org.example.generated.runtime.v1") diff --git a/cmd/mobyextgen/testdata/runtime_server_contract.go.txt b/cmd/mobyextgen/testdata/runtime_server_contract.go.txt new file mode 100644 index 0000000..6a1271a --- /dev/null +++ b/cmd/mobyextgen/testdata/runtime_server_contract.go.txt @@ -0,0 +1,22 @@ +// SPDX-FileCopyrightText: Copyright The Moby Authors +// SPDX-License-Identifier: Apache-2.0 + +package contract + +import ( + "context" + + stream "github.com/moby/extensions/stream" +) + +type Service interface { + Download(context.Context, *Request, stream.Sender[Chunk]) error +} + +type Request struct { + Value int32 `pb:"1"` +} + +type Chunk struct { + Value int32 `pb:"1"` +} diff --git a/cmd/mobyextgen/testdata/runtime_unary_contract.go.txt b/cmd/mobyextgen/testdata/runtime_unary_contract.go.txt new file mode 100644 index 0000000..98e8442 --- /dev/null +++ b/cmd/mobyextgen/testdata/runtime_unary_contract.go.txt @@ -0,0 +1,18 @@ +// SPDX-FileCopyrightText: Copyright The Moby Authors +// SPDX-License-Identifier: Apache-2.0 + +package contract + +import "context" + +type Service interface { + Run(context.Context, *Request) (*Reply, error) +} + +type Request struct { + Value int32 `pb:"1"` +} + +type Reply struct { + Value int32 `pb:"1"` +} diff --git a/cmd/mobyextgen/wire.go b/cmd/mobyextgen/wire.go index 9beefa8..7b479e8 100644 --- a/cmd/mobyextgen/wire.go +++ b/cmd/mobyextgen/wire.go @@ -16,17 +16,33 @@ func emitWire(pt point) ([]byte, error) { var b strings.Builder const cpkg = "mobyextcontract" svc, iface := pt.service, pt.iface + streamIndices := make(map[string]int) + for _, m := range pt.methods { + if m.kind != methodUnary { + streamIndices[m.name] = len(streamIndices) + } + } + hasStreaming := len(streamIndices) > 0 b.WriteString(pt.licenseHeader) fmt.Fprintln(&b, "// Code generated by mobyextgen. DO NOT EDIT.") fmt.Fprintf(&b, "package %s\n\n", path.Base(pt.protogenImport())) fmt.Fprintln(&b, "import (") fmt.Fprintln(&b, ` context "context"`) + if hasStreaming { + fmt.Fprintln(&b, ` errors "errors"`) + fmt.Fprintln(&b, ` io "io"`) + } if pt.isPoint { fmt.Fprintln(&b, ` extensions "github.com/moby/extensions"`) fmt.Fprintln(&b, ` clientpoint "github.com/moby/extensions/clientpoint"`) fmt.Fprintln(&b, ` serverpoint "github.com/moby/extensions/serverpoint"`) } fmt.Fprintf(&b, "\t%s %q\n", cpkg, pt.importPath) + if hasStreaming { + fmt.Fprintln(&b, ` extensionstream "github.com/moby/extensions/stream"`) + fmt.Fprintln(&b, ` codes "google.golang.org/grpc/codes"`) + fmt.Fprintln(&b, ` status "google.golang.org/grpc/status"`) + } fmt.Fprintln(&b, ` grpc "google.golang.org/grpc"`) fmt.Fprintln(&b, ")") @@ -60,7 +76,12 @@ func emitWire(pt point) ([]byte, error) { } for _, m := range pt.methods { emitComment(&b, "\t", m.doc) - fmt.Fprintf(&b, "\t%s(context.Context, *%s) (*%s, error)\n", m.name, m.request, m.response) + switch m.kind { + case methodUnary: + fmt.Fprintf(&b, "\t%s(context.Context, *%s) (*%s, error)\n", m.name, m.request, m.response) + case methodServerStreaming: + fmt.Fprintf(&b, "\t%s(*%s, grpc.ServerStreamingServer[%s]) error\n", m.name, m.request, m.response) + } } fmt.Fprintln(&b, "}") @@ -79,12 +100,31 @@ func emitWire(pt point) ([]byte, error) { Methods: []grpc.MethodDesc{ `, svc, pt.protoPath) for _, m := range pt.methods { - fmt.Fprintf(&b, "\t\t{MethodName: %q, Handler: %s},\n", m.name, handlerName(m)) + if m.kind == methodUnary { + fmt.Fprintf(&b, "\t\t{MethodName: %q, Handler: %s},\n", m.name, handlerName(m)) + } } - fmt.Fprintln(&b, "\t},\n}") + fmt.Fprintln(&b, "\t},") + if hasStreaming { + fmt.Fprintln(&b, "\t\tStreams: []grpc.StreamDesc{") + for _, m := range pt.methods { + if m.kind == methodUnary { + continue + } + fmt.Fprintf(&b, "\t\t\t{StreamName: %q, Handler: %s", m.name, handlerName(m)) + if m.kind == methodServerStreaming { + fmt.Fprint(&b, ", ServerStreams: true") + } + fmt.Fprintln(&b, "},") + } + fmt.Fprintln(&b, "\t\t},") + } + fmt.Fprintln(&b, "}") for _, m := range pt.methods { - fmt.Fprintf(&b, ` + switch m.kind { + case methodUnary: + fmt.Fprintf(&b, ` func %[1]s(srv any, ctx context.Context, dec func(any) error, interceptor grpc.UnaryServerInterceptor) (any, error) { in := new(%[2]s) if err := dec(in); err != nil { @@ -99,6 +139,17 @@ func emitWire(pt point) ([]byte, error) { }) } `, handlerName(m), m.request, svc, m.name, methodConst(m)) + case methodServerStreaming: + fmt.Fprintf(&b, ` + func %[1]s(srv any, stream grpc.ServerStream) error { + in := new(%[2]s) + if err := stream.RecvMsg(in); err != nil { + return err + } + return srv.(%[3]sServer).%[4]s(in, &grpc.GenericServerStream[%[2]s, %[5]s]{ServerStream: stream}) + } + `, handlerName(m), m.request, svc, m.name, m.response) + } } if pt.isPoint { @@ -117,7 +168,12 @@ func emitWire(pt point) ([]byte, error) { } for _, m := range pt.methods { emitComment(&b, "\t", m.doc) - fmt.Fprintf(&b, "\t%s(ctx context.Context, in *%s, opts ...grpc.CallOption) (*%s, error)\n", m.name, m.request, m.response) + switch m.kind { + case methodUnary: + fmt.Fprintf(&b, "\t%s(ctx context.Context, in *%s, opts ...grpc.CallOption) (*%s, error)\n", m.name, m.request, m.response) + case methodServerStreaming: + fmt.Fprintf(&b, "\t%s(ctx context.Context, in *%s, opts ...grpc.CallOption) (grpc.ServerStreamingClient[%s], error)\n", m.name, m.request, m.response) + } } fmt.Fprintln(&b, "}") @@ -132,7 +188,9 @@ func emitWire(pt point) ([]byte, error) { `, svc) for _, m := range pt.methods { - fmt.Fprintf(&b, ` + switch m.kind { + case methodUnary: + fmt.Fprintf(&b, ` func (c *serviceClient) %[1]s(ctx context.Context, in *%[2]s, opts ...grpc.CallOption) (*%[3]s, error) { out := new(%[3]s) if err := c.cc.Invoke(ctx, %[4]s, in, out, append([]grpc.CallOption{grpc.StaticMethod()}, opts...)...); err != nil { @@ -141,6 +199,24 @@ func emitWire(pt point) ([]byte, error) { return out, nil } `, m.name, m.request, m.response, methodConst(m)) + case methodServerStreaming: + fmt.Fprintf(&b, ` + func (c *serviceClient) %[1]s(ctx context.Context, in *%[2]s, opts ...grpc.CallOption) (grpc.ServerStreamingClient[%[3]s], error) { + stream, err := c.cc.NewStream(ctx, &serviceDesc.Streams[%[4]d], %[5]s, append([]grpc.CallOption{grpc.StaticMethod()}, opts...)...) + if err != nil { + return nil, err + } + x := &grpc.GenericClientStream[%[2]s, %[3]s]{ClientStream: stream} + if err := x.ClientStream.SendMsg(in); err != nil { + return nil, err + } + if err := x.ClientStream.CloseSend(); err != nil { + return nil, err + } + return x, nil + } + `, m.name, m.request, m.response, streamIndices[m.name], methodConst(m)) + } } if pt.isPoint { @@ -189,8 +265,10 @@ func emitWire(pt point) ([]byte, error) { `, cpkg, iface) for _, m := range pt.methods { - if m.bareError { - fmt.Fprintf(&b, ` + switch m.kind { + case methodUnary: + if m.bareError { + fmt.Fprintf(&b, ` func (s *grpcServer) %[1]s(ctx context.Context, req *%[2]s) (*%[3]s, error) { if err := s.impl.%[1]s(ctx, %[4]sFromProto(req)); err != nil { return nil, err @@ -198,8 +276,8 @@ func emitWire(pt point) ([]byte, error) { return &%[3]s{}, nil } `, m.name, m.request, m.response, lowerFirst(m.request)) - } else { - fmt.Fprintf(&b, ` + } else { + fmt.Fprintf(&b, ` func (s *grpcServer) %[1]s(ctx context.Context, req *%[2]s) (*%[3]s, error) { resp, err := s.impl.%[1]s(ctx, %[4]sFromProto(req)) if err != nil { @@ -208,6 +286,25 @@ func emitWire(pt point) ([]byte, error) { return %[5]sToProto(resp), nil } `, m.name, m.request, m.response, lowerFirst(m.request), lowerFirst(m.response)) + } + case methodServerStreaming: + fmt.Fprintf(&b, ` + func (s *grpcServer) %[1]s(req *%[2]s, out grpc.ServerStreamingServer[%[3]s]) error { + sender := &mobyext%[1]sSender{stream: out} + return s.impl.%[1]s(out.Context(), %[5]sFromProto(req), sender) + } + + type mobyext%[1]sSender struct { + stream grpc.ServerStreamingServer[%[3]s] + } + + func (s *mobyext%[1]sSender) Send(message *%[6]s.%[7]s) error { + if message == nil { + return status.Error(codes.Internal, %[8]q) + } + return s.stream.Send(%[9]sToProto(message)) + } + `, m.name, m.request, m.response, lowerFirst(m.name), lowerFirst(m.request), cpkg, m.response, m.name+": nil message", lowerFirst(m.response)) } } @@ -217,15 +314,17 @@ func emitWire(pt point) ([]byte, error) { } `, svc) for _, m := range pt.methods { - if m.bareError { - fmt.Fprintf(&b, ` + switch m.kind { + case methodUnary: + if m.bareError { + fmt.Fprintf(&b, ` func (c *grpcClient) %[1]s(ctx context.Context, req *%[2]s.%[3]s) error { _, err := c.client.%[1]s(ctx, %[4]sToProto(req)) return err } `, m.name, cpkg, m.request, lowerFirst(m.request)) - } else { - fmt.Fprintf(&b, ` + } else { + fmt.Fprintf(&b, ` func (c *grpcClient) %[1]s(ctx context.Context, req *%[2]s.%[3]s) (*%[2]s.%[5]s, error) { resp, err := c.client.%[1]s(ctx, %[4]sToProto(req)) if err != nil { @@ -234,6 +333,33 @@ func emitWire(pt point) ([]byte, error) { return %[6]sFromProto(resp), nil } `, m.name, cpkg, m.request, lowerFirst(m.request), m.response, lowerFirst(m.response)) + } + case methodServerStreaming: + fmt.Fprintf(&b, ` + func (c *grpcClient) %[1]s(ctx context.Context, req *%[2]s.%[3]s, sink extensionstream.Sender[%[2]s.%[4]s]) error { + ctx, cancel := context.WithCancel(ctx) + defer cancel() + stream, err := c.client.%[1]s(ctx, %[5]sToProto(req)) + if err != nil { + return err + } + for { + if err := ctx.Err(); err != nil { + return status.FromContextError(err).Err() + } + message, err := stream.Recv() + if err != nil { + if errors.Is(err, io.EOF) { + return nil + } + return err + } + if err := sink.Send(%[6]sFromProto(message)); err != nil { + return err + } + } + } + `, m.name, cpkg, m.request, m.response, lowerFirst(m.request), lowerFirst(m.response)) } } diff --git a/docs/AUTHORING.md b/docs/AUTHORING.md index 4d86c43..22a8045 100644 --- a/docs/AUTHORING.md +++ b/docs/AUTHORING.md @@ -150,6 +150,11 @@ The generated package exports: - `ClientPoint`, for host wiring to a launched provider; - `NewClient(conn)`, which returns the handwritten Go interface over a gRPC connection. +A contract can also declare server-streaming methods with `stream.Sender[Message]`. +The handwritten provider interface keeps its Go types. +The generated adapters make these methods available as protobuf streaming RPCs. +The generator rejects all other stream parameter and result shapes. + The contract package does not import protobuf packages; generated code belongs in `protogen/`. The generator derives the import path from `go.mod` and the proto file name from the inferred service identity. diff --git a/docs/DESIGN.md b/docs/DESIGN.md index dca2f65..d30cdef 100644 --- a/docs/DESIGN.md +++ b/docs/DESIGN.md @@ -229,6 +229,47 @@ The Go interface and message structs are the source of truth. `mobyextgen` generates the proto, protobuf message code, gRPC service, client and server points, adapters, and conversions without `protoc`, plugins, or other `PATH` tools. In-process providers implement Go directly; out-of-process providers use gRPC. +The generator supports two method kinds: unary and server-streaming. +A unary method has the shape `M(context.Context, *Request) (*Reply, error)`. +The generator also accepts the legacy unary shape `M(context.Context, *Request) error`. +A server-streaming method has the shape `M(context.Context, *Request, stream.Sender[Chunk]) error`. +The generator rejects all other stream signatures. +A stream message type must be a local struct type that is not generic. + +### Why not channels? + +A Go channel carries values and a close signal. +A channel does not show receive errors, send errors, or the final RPC status. +A channel contract must define rules for these results. +It must also define rules for cancellation, buffers, message ownership, and which side closes the channel. +An adapter between an asynchronous channel and an RPC can need forwarding goroutines and careful cleanup. +The generator can support channels, but channels are not the standard contract. +If channels were the standard contract, each streaming method would include Go-specific concurrency rules. + +### Why not iterators? + +Iterators are a possible alternative for RPC adapters. +With iterators, a consumer can read values in sequence and stop early without forwarding goroutines. +But an `iter.Seq2[*T, error]` contract must still define rules for errors, cancellation, message ownership, and method completion. +The stream interfaces give the generator a small and explicit contract to send and receive messages. +This contract has a shape similar to the Go I/O interfaces. +`All` and `Copy` give the convenience of iterators, but iterator signatures do not become part of the protocol model. +The cost is two framework interfaces in addition to the standard library types. + +### What method scope provides + +The enclosing method stays active until the stream is complete. +The return value of the method sets the final success or failure. +A normal server-streaming EOF alone does not show success. +The final `OK` status of the provider shows success. +Adapters reject a nil stream message from a provider with `Internal`. + +Generated adapters return provider errors without changes. +gRPC converts these errors at the server boundary. +Explicit status errors keep their details. +Context cancellation errors and deadline errors become `Canceled` and `DeadlineExceeded`, also when they are wrapped. +All other errors become `Unknown`. + Messages use proto3. Supported fields are scalar values, `string`, `[]byte`, repeated scalars such as `[]string`, string-keyed scalar maps such as `map[string]string`, pointer nested messages such as `*Other`, and repeated nested messages such as `[]Other`. A single nested message must be a pointer; value nested messages are rejected. From 52b64a70b5573673287a36e9d92b6dc02a0c2f21 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Pawe=C5=82=20Gronowski?= Date: Wed, 14 Oct 2026 17:28:02 +0200 Subject: [PATCH 6/7] cmd/mobyextgen: Generate client-streaming RPCs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Contracts can declare client-streaming methods with this shape: M(ctx, stream.Receiver[T]) (*Resp, error) The generator emits them as protobuf client-streaming RPCs and adapts the gRPC stream to stream.Receiver on both sides. The client adapter reads the caller's Receiver synchronously, so a blocked transport send stops further reads. A normal io.EOF from the Receiver closes the stream and waits for the provider's reply. An io.EOF from Send means the provider finished early, so the adapter returns the provider's final status. A nil message from the caller fails with codes.InvalidArgument. Bidirectional streaming remains rejected. Signed-off-by: Paweł Gronowski --- cmd/mobyextgen/descriptor.go | 3 + cmd/mobyextgen/model.go | 1 + cmd/mobyextgen/parse.go | 30 +- cmd/mobyextgen/proto.go | 2 + cmd/mobyextgen/runtime_test.go | 10 + cmd/mobyextgen/streaming_test.go | 69 ++++- .../testdata/runtime_behavior_test.go.txt | 266 +++++++++++++++++- .../testdata/runtime_client_contract.go.txt | 22 ++ .../testdata/runtime_contract.go.txt | 7 + .../testdata/runtime_transport_test.go.txt | 127 +++++++++ cmd/mobyextgen/wire.go | 95 ++++++- docs/AUTHORING.md | 4 +- docs/DESIGN.md | 13 +- 13 files changed, 625 insertions(+), 24 deletions(-) create mode 100644 cmd/mobyextgen/testdata/runtime_client_contract.go.txt create mode 100644 cmd/mobyextgen/testdata/runtime_transport_test.go.txt diff --git a/cmd/mobyextgen/descriptor.go b/cmd/mobyextgen/descriptor.go index 7e266c2..0e263f8 100644 --- a/cmd/mobyextgen/descriptor.go +++ b/cmd/mobyextgen/descriptor.go @@ -85,6 +85,9 @@ func fileDescriptor(pt point) (*descriptorpb.FileDescriptorProto, error) { InputType: new("." + pt.id + "." + m.request), OutputType: new("." + pt.id + "." + m.response), } + if m.kind == methodClientStreaming { + method.ClientStreaming = new(true) + } if m.kind == methodServerStreaming { method.ServerStreaming = new(true) } diff --git a/cmd/mobyextgen/model.go b/cmd/mobyextgen/model.go index 0694c6e..5b1c3a2 100644 --- a/cmd/mobyextgen/model.go +++ b/cmd/mobyextgen/model.go @@ -38,6 +38,7 @@ type methodKind int const ( methodUnary methodKind = iota methodServerStreaming + methodClientStreaming ) type message struct { diff --git a/cmd/mobyextgen/parse.go b/cmd/mobyextgen/parse.go index b394a91..a1202de 100644 --- a/cmd/mobyextgen/parse.go +++ b/cmd/mobyextgen/parse.go @@ -307,6 +307,18 @@ func parseMethods(files []*ast.File, iface *ast.InterfaceType) ([]method, error) kind: methodServerStreaming, name: name, request: params[1].message, response: params[2].message, doc: doc, }) + case len(params) == 2 && params[0].kind == methodContextParameter && params[1].kind == methodReceiverParameter: + if len(res) != 2 || !isIdent(res[1], "error") { + return nil, methodShapeError(name) + } + response, err := pointerIdent(res[0]) + if err != nil { + return nil, methodShapeError(name) + } + methods = append(methods, method{ + kind: methodClientStreaming, name: name, + request: params[1].message, response: response, doc: doc, + }) default: return nil, methodShapeError(name) } @@ -320,6 +332,7 @@ const ( methodContextParameter methodParameterKind = iota + 1 methodMessageParameter methodSenderParameter + methodReceiverParameter ) type methodParameter struct { @@ -328,8 +341,9 @@ type methodParameter struct { } const acceptedMethodShapes = "accepted shapes are M(context.Context, *Request) (*Reply, error), " + - "M(context.Context, *Request) error, and " + - "M(context.Context, *Request, stream.Sender[Chunk]) error" + "M(context.Context, *Request) error, " + + "M(context.Context, *Request, stream.Sender[Chunk]) error, and " + + "M(context.Context, stream.Receiver[Chunk]) (*Reply, error)" func methodShapeError(name string) error { return fmt.Errorf("method %q: unsupported signature; %s", name, acceptedMethodShapes) @@ -354,8 +368,8 @@ func classifyMethodParameter(files []*ast.File, expr ast.Expr, index int) (metho return methodParameter{kind: methodMessageParameter, message: message}, nil } -// parseStreamParameter reports whether expr names stream.Sender and, if so, -// validates its import and message argument. +// parseStreamParameter reports whether expr names stream.Sender or +// stream.Receiver and, if so, validates its import and message argument. func parseStreamParameter(files []*ast.File, expr ast.Expr) (methodParameter, bool, error) { var ( selector *ast.SelectorExpr @@ -373,7 +387,7 @@ func parseStreamParameter(files []*ast.File, expr ast.Expr) (methodParameter, bo default: return methodParameter{}, false, nil } - if selector == nil || selector.Sel.Name != "Sender" { + if selector == nil || (selector.Sel.Name != "Sender" && selector.Sel.Name != "Receiver") { return methodParameter{}, false, nil } use := "stream." + selector.Sel.Name @@ -394,7 +408,11 @@ func parseStreamParameter(files []*ast.File, expr ast.Expr) (methodParameter, bo if err := validateStreamMessage(files, message); err != nil { return methodParameter{}, true, fmt.Errorf("%s message %q: %w", use, message, err) } - return methodParameter{kind: methodSenderParameter, message: message}, true, nil + kind := methodSenderParameter + if selector.Sel.Name == "Receiver" { + kind = methodReceiverParameter + } + return methodParameter{kind: kind, message: message}, true, nil } func fileForExpr(files []*ast.File, expr ast.Expr) *ast.File { diff --git a/cmd/mobyextgen/proto.go b/cmd/mobyextgen/proto.go index 26f561f..6a89606 100644 --- a/cmd/mobyextgen/proto.go +++ b/cmd/mobyextgen/proto.go @@ -30,6 +30,8 @@ func emitProto(pt point) ([]byte, error) { case methodUnary: case methodServerStreaming: response = "stream " + response + case methodClientStreaming: + request = "stream " + request default: return nil, fmt.Errorf("%s: unknown method kind %d", m.name, m.kind) } diff --git a/cmd/mobyextgen/runtime_test.go b/cmd/mobyextgen/runtime_test.go index 56ae931..ffe1cb7 100644 --- a/cmd/mobyextgen/runtime_test.go +++ b/cmd/mobyextgen/runtime_test.go @@ -35,6 +35,11 @@ func TestGeneratedStreamingRuntime(t *testing.T) { "__PROTOGEN_IMPORT__", path.Join(importPath, "protogen"), ).Replace(readFixture(t, "runtime_behavior_test.go.txt")) assert.NilError(t, os.WriteFile(filepath.Join(dir, "runtime_test.go"), []byte(behavior), 0o644)) + transport := strings.NewReplacer( + "__CONTRACT_IMPORT__", importPath, + "__PROTOGEN_IMPORT__", path.Join(importPath, "protogen"), + ).Replace(readFixture(t, "runtime_transport_test.go.txt")) + assert.NilError(t, os.WriteFile(filepath.Join(dir, "runtime_transport_test.go"), []byte(transport), 0o644)) assert.NilError(t, run(dir, "")) for _, fixture := range []struct { @@ -52,6 +57,11 @@ func TestGeneratedStreamingRuntime(t *testing.T) { source: "runtime_server_contract.go.txt", id: "org.example.generated.serverstream.v1.Service", }, + { + name: "clientstream", + source: "runtime_client_contract.go.txt", + id: "org.example.generated.clientstream.v1.Service", + }, } { subdir := filepath.Join(dir, fixture.name) assert.NilError(t, os.MkdirAll(subdir, 0o755)) diff --git a/cmd/mobyextgen/streaming_test.go b/cmd/mobyextgen/streaming_test.go index 3601065..2c33455 100644 --- a/cmd/mobyextgen/streaming_test.go +++ b/cmd/mobyextgen/streaming_test.go @@ -4,6 +4,8 @@ package main import ( + "go/parser" + "go/token" "strings" "testing" @@ -31,25 +33,33 @@ func TestParseContractStreamingContracts(t *testing.T) { type Service interface { // Download sends chunks to the caller. Download(stdctx.Context, *DownloadRequest, s.Sender[DownloadChunk]) error + // Upload receives chunks from the caller. + Upload(stdctx.Context, s.Receiver[UploadChunk]) (*UploadReply, error) } type DownloadRequest struct{} type DownloadChunk struct{} +type UploadChunk struct{} +type UploadReply struct{} `) point, err := parseSource(t, source) assert.NilError(t, err) assert.Equal(t, point.doc, "Service documents its streaming operations.") - assert.Equal(t, len(point.methods), 1) + assert.Equal(t, len(point.methods), 2) assert.Equal(t, point.methods[0].kind, methodServerStreaming) assert.Equal(t, point.methods[0].request, "DownloadRequest") assert.Equal(t, point.methods[0].response, "DownloadChunk") assert.Equal(t, point.methods[0].bareError, false) assert.Equal(t, point.methods[0].doc, "Download sends chunks to the caller.") + assert.Equal(t, point.methods[1].kind, methodClientStreaming) + assert.Equal(t, point.methods[1].request, "UploadChunk") + assert.Equal(t, point.methods[1].response, "UploadReply") + assert.Equal(t, point.methods[1].doc, "Upload receives chunks from the caller.") messageNames := make(map[string]bool, len(point.messages)) for _, message := range point.messages { messageNames[message.name] = true } - for _, name := range []string{"DownloadRequest", "DownloadChunk"} { + for _, name := range []string{"DownloadRequest", "DownloadChunk", "UploadChunk", "UploadReply"} { assert.Assert(t, messageNames[name], "missing message %s", name) } } @@ -62,11 +72,10 @@ func TestParseContractRejectsInvalidStreamingContracts(t *testing.T) { want string }{ { - name: "server stream with a reply", + name: "unsupported parameter shape", source: streamingPointSource(`type Service interface { - Download(stdctx.Context, *Request, s.Sender[Chunk]) (*Request, error) + Sync(stdctx.Context, s.Receiver[Chunk], s.Sender[Chunk]) error } -type Request struct{} type Chunk struct{} `), want: acceptedMethodShapes, @@ -147,17 +156,56 @@ type Chunk[T any] struct{} } } +func TestStreamingHelpersDoNotCollideWithHandlers(t *testing.T) { + t.Parallel() + tests := []struct { + name string + methods string + }{ + { + name: "server stream beside a unary Sender method", + methods: `Handle(stdctx.Context, *Request, s.Sender[Chunk]) error + Sender(stdctx.Context, *Request) error`, + }, + { + name: "client stream beside a unary Receiver method", + methods: `Handle(stdctx.Context, s.Receiver[Chunk]) (*Request, error) + Receiver(stdctx.Context, *Request) error`, + }, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + t.Parallel() + point, err := parseSource(t, streamingPointSource("type Service interface {\n\t"+test.methods+` +} +type Request struct{} +type Chunk struct{} +`)) + assert.NilError(t, err) + point.importPath = "example.test/contract" + wire, err := emitWire(point) + assert.NilError(t, err) + _, err = parser.ParseFile(token.NewFileSet(), "wire.gen.go", wire, parser.DeclarationErrors) + assert.NilError(t, err) + }) + } +} + func TestStreamingProtoDescriptorAndWire(t *testing.T) { t.Parallel() source := streamingPointSource(` -// Service documents its streaming operations. +// Service documents both streaming directions. type Service interface { // Download sends chunks to the caller. Download(stdctx.Context, *DownloadRequest, s.Sender[DownloadChunk]) error + // Upload receives chunks from the caller. + Upload(stdctx.Context, s.Receiver[UploadChunk]) (*UploadReply, error) } type DownloadRequest struct{} type DownloadChunk struct{} +type UploadChunk struct{} +type UploadReply struct{} `) point, err := parseSource(t, source) assert.NilError(t, err) @@ -169,6 +217,7 @@ type DownloadChunk struct{} protoText := string(protoFile) for _, declaration := range []string{ "rpc Download(DownloadRequest) returns (stream DownloadChunk);", + "rpc Upload(stream UploadChunk) returns (UploadReply);", } { assert.Assert(t, strings.Contains(protoText, declaration), declaration) } @@ -177,19 +226,25 @@ type DownloadChunk struct{} assert.NilError(t, err) assert.Equal(t, len(fd.GetService()), 1) methods := fd.GetService()[0].GetMethod() - assert.Equal(t, len(methods), 1) + assert.Equal(t, len(methods), 2) assert.Equal(t, methods[0].GetName(), "Download") assert.Equal(t, methods[0].GetServerStreaming(), true) assert.Equal(t, methods[0].GetClientStreaming(), false) assert.Assert(t, methods[0].ClientStreaming == nil) + assert.Equal(t, methods[1].GetName(), "Upload") + assert.Equal(t, methods[1].GetServerStreaming(), false) + assert.Equal(t, methods[1].GetClientStreaming(), true) + assert.Assert(t, methods[1].ServerStreaming == nil) wire, err := emitWire(point) assert.NilError(t, err) wireText := string(wire) for _, declaration := range []string{ "Download(*DownloadRequest, grpc.ServerStreamingServer[DownloadChunk]) error", + "Upload(grpc.ClientStreamingServer[UploadChunk, UploadReply]) error", "Streams: []grpc.StreamDesc{", "ServerStreams: true", + "ClientStreams: true", } { assert.Assert(t, strings.Contains(wireText, declaration), declaration) } diff --git a/cmd/mobyextgen/testdata/runtime_behavior_test.go.txt b/cmd/mobyextgen/testdata/runtime_behavior_test.go.txt index 33d143a..ee5040d 100644 --- a/cmd/mobyextgen/testdata/runtime_behavior_test.go.txt +++ b/cmd/mobyextgen/testdata/runtime_behavior_test.go.txt @@ -6,8 +6,10 @@ package contract_test import ( "context" "errors" + "io" "net" "sync" + "sync/atomic" "testing" "time" @@ -34,8 +36,18 @@ type provider struct { downloadStartOnce sync.Once downloadDoneOnce sync.Once - unaryErr error - legacyErr error + uploadErr error + uploadRejectAfter int + uploadWait <-chan struct{} + uploadStart chan struct{} + uploadDone chan struct{} + uploadObservedErr error + uploadDoneOnce sync.Once + uploadStartOnce sync.Once + uploadValues []int32 + uploadFinalized atomic.Bool + unaryErr error + legacyErr error } var _ contract.Service = (*provider)(nil) @@ -81,6 +93,55 @@ func (p *provider) download(ctx context.Context, sender stream.Sender[contract.C return p.downloadErr } +func (p *provider) Upload(ctx context.Context, receiver stream.Receiver[contract.Chunk]) (*contract.Reply, error) { + values, err := p.receive(ctx, receiver) + if err != nil { + return nil, err + } + return &contract.Reply{Values: values}, nil +} + +func (p *provider) UploadAgain(ctx context.Context, receiver stream.Receiver[contract.Chunk]) (*contract.AlternateReply, error) { + values, err := p.receive(ctx, receiver) + if err != nil { + return nil, err + } + return &contract.AlternateReply{Values: values, Marker: "alternate"}, nil +} + +func (p *provider) receive(ctx context.Context, receiver stream.Receiver[contract.Chunk]) ([]int32, error) { + p.signalUploadStart() + defer p.signalUploadDone() + for { + message, err := receiver.Recv() + if err != nil { + if !errors.Is(err, io.EOF) { + p.uploadObservedErr = err + return nil, err + } + p.uploadFinalized.Store(true) + return append([]int32(nil), p.uploadValues...), nil + } + if message == nil { + err := errors.New("provider received nil message") + p.uploadObservedErr = err + return nil, err + } + p.uploadValues = append(p.uploadValues, message.Value) + if p.uploadRejectAfter > 0 && len(p.uploadValues) >= p.uploadRejectAfter { + return nil, p.uploadErr + } + if p.uploadWait != nil { + select { + case <-p.uploadWait: + case <-ctx.Done(): + p.uploadObservedErr = ctx.Err() + return nil, ctx.Err() + } + } + } +} + func (p *provider) signalDownloadStart() { if p.downloadStart != nil { p.downloadStartOnce.Do(func() { close(p.downloadStart) }) @@ -93,6 +154,18 @@ func (p *provider) signalDownloadDone() { } } +func (p *provider) signalUploadStart() { + if p.uploadStart != nil { + p.uploadStartOnce.Do(func() { close(p.uploadStart) }) + } +} + +func (p *provider) signalUploadDone() { + if p.uploadDone != nil { + p.uploadDoneOnce.Do(func() { close(p.uploadDone) }) + } +} + type runtimeMode struct { name string new func(*testing.T, *provider) contract.Service @@ -189,6 +262,72 @@ func asyncErrorCall(t *testing.T, call func() error) <-chan error { return result } +type uploadResult struct { + reply *contract.Reply + err error +} + +func asyncUploadCall(t *testing.T, call func() (*contract.Reply, error)) <-chan uploadResult { + t.Helper() + result := make(chan uploadResult, 1) + done := make(chan struct{}) + go func() { + reply, err := call() + result <- uploadResult{reply: reply, err: err} + close(done) + }() + t.Cleanup(func() { + timer := time.NewTimer(10 * time.Second) + defer timer.Stop() + select { + case <-done: + case <-timer.C: + t.Errorf("timed out waiting for asynchronous upload") + } + }) + return result +} + +type source struct { + ready <-chan struct{} + values []*contract.Chunk + last error + index int +} + +func (s *source) Recv() (*contract.Chunk, error) { + if s.ready != nil { + ready := s.ready + s.ready = nil + <-ready + } + if s.index < len(s.values) { + message := s.values[s.index] + s.index++ + return message, nil + } + return nil, s.last +} + +type contextSource struct { + ctx context.Context + values []*contract.Chunk + index int + started chan struct{} + once sync.Once +} + +func (s *contextSource) Recv() (*contract.Chunk, error) { + if s.index < len(s.values) { + message := s.values[s.index] + s.index++ + return message, nil + } + s.once.Do(func() { close(s.started) }) + <-s.ctx.Done() + return nil, s.ctx.Err() +} + type sink struct { values []int32 err error @@ -264,6 +403,34 @@ func TestDirectAndRemoteParity(t *testing.T) { assert.DeepEqual(t, got.values, []int32{4, 5}) }) + t.Run("empty and multiple uploads", func(t *testing.T) { + p := &provider{} + client := mode.new(t, p) + reply, err := client.Upload(context.Background(), &source{last: io.EOF}) + assert.NilError(t, err) + assert.Equal(t, len(reply.Values), 0) + + p = &provider{} + client = mode.new(t, p) + reply, err = client.Upload(context.Background(), &source{ + values: []*contract.Chunk{{Value: 2}, {Value: 4}, {Value: 6}}, + last: errors.Join(errors.New("input complete"), io.EOF), + }) + assert.NilError(t, err) + assert.DeepEqual(t, reply.Values, []int32{2, 4, 6}) + assert.Assert(t, p.uploadFinalized.Load()) + + p = &provider{} + client = mode.new(t, p) + alternate, err := client.UploadAgain(context.Background(), &source{ + values: []*contract.Chunk{{Value: 8}}, + last: io.EOF, + }) + assert.NilError(t, err) + assert.DeepEqual(t, alternate.Values, []int32{8}) + assert.Equal(t, alternate.Marker, "alternate") + }) + t.Run("final status follows delivered download data", func(t *testing.T) { p := &provider{ downloadValues: []int32{10, 20}, @@ -314,6 +481,33 @@ func TestDirectAndRemoteErrorIdentity(t *testing.T) { await(t, p.downloadDone) } }) + + t.Run("source error is local and not a clean EOF", func(t *testing.T) { + sourceErr := errors.New("source failed") + p := &provider{ + uploadStart: make(chan struct{}), + uploadDone: make(chan struct{}), + } + client := mode.new(t, p) + source := &source{ + ready: p.uploadStart, + values: []*contract.Chunk{{Value: 3}}, + last: sourceErr, + } + result := asyncUploadCall(t, func() (*contract.Reply, error) { + return client.Upload(context.Background(), source) + }) + t.Cleanup(p.signalUploadStart) + await(t, p.uploadStart) + assert.Equal(t, (<-result).err, sourceErr) + await(t, p.uploadDone) + assert.Assert(t, !p.uploadFinalized.Load()) + if mode.name == "direct" { + assert.Equal(t, p.uploadObservedErr, sourceErr) + } else { + assert.Assert(t, p.uploadObservedErr != sourceErr) + } + }) }) } } @@ -369,6 +563,46 @@ func TestServerStreamingCallbacksAndCancellation(t *testing.T) { }) } +func TestClientStreamingInvalidInputAndProviderStatus(t *testing.T) { + t.Parallel() + t.Run("nil source message is invalid", func(t *testing.T) { + p := &provider{ + uploadStart: make(chan struct{}), + uploadDone: make(chan struct{}), + } + client := start(t, p) + source := &source{ + ready: p.uploadStart, + values: []*contract.Chunk{nil}, + last: io.EOF, + } + result := asyncUploadCall(t, func() (*contract.Reply, error) { + return client.Upload(context.Background(), source) + }) + t.Cleanup(p.signalUploadStart) + await(t, p.uploadStart) + assert.Equal(t, status.Code((<-result).err), codes.InvalidArgument) + await(t, p.uploadDone) + assert.Assert(t, !p.uploadFinalized.Load()) + }) + + t.Run("provider final status survives early rejection", func(t *testing.T) { + p := &provider{ + uploadErr: detailError(codes.ResourceExhausted, "quota", "UPLOAD_QUOTA"), + uploadRejectAfter: 1, + uploadDone: make(chan struct{}), + } + client := start(t, p) + _, err := client.Upload(context.Background(), &source{ + values: []*contract.Chunk{{Value: 1}, {Value: 2}, {Value: 3}}, + last: io.EOF, + }) + assertDetail(t, err, codes.ResourceExhausted, "quota", "UPLOAD_QUOTA") + await(t, p.uploadDone) + assert.Assert(t, !p.uploadFinalized.Load()) + }) +} + func TestDirectContextCancellation(t *testing.T) { t.Parallel() release := newRelease() @@ -388,3 +622,31 @@ func TestDirectContextCancellation(t *testing.T) { assert.Equal(t, <-result, context.Canceled) await(t, p.downloadDone) } + +func TestClientStreamingCancellationCleansUpHandler(t *testing.T) { + t.Parallel() + release := newRelease() + p := &provider{ + uploadStart: make(chan struct{}), + uploadDone: make(chan struct{}), + uploadWait: release.ch, + } + client := start(t, p) + ctx, cancel := context.WithCancel(context.Background()) + source := &contextSource{ + ctx: ctx, + values: []*contract.Chunk{{Value: 9}}, + started: make(chan struct{}), + } + result := asyncUploadCall(t, func() (*contract.Reply, error) { + return client.Upload(ctx, source) + }) + t.Cleanup(cancel) + t.Cleanup(release.Close) + t.Cleanup(p.signalUploadStart) + await(t, p.uploadStart) + await(t, source.started) + cancel() + assert.Equal(t, (<-result).err, context.Canceled) + await(t, p.uploadDone) +} diff --git a/cmd/mobyextgen/testdata/runtime_client_contract.go.txt b/cmd/mobyextgen/testdata/runtime_client_contract.go.txt new file mode 100644 index 0000000..e6345b7 --- /dev/null +++ b/cmd/mobyextgen/testdata/runtime_client_contract.go.txt @@ -0,0 +1,22 @@ +// SPDX-FileCopyrightText: Copyright The Moby Authors +// SPDX-License-Identifier: Apache-2.0 + +package contract + +import ( + "context" + + stream "github.com/moby/extensions/stream" +) + +type Service interface { + Upload(context.Context, stream.Receiver[Chunk]) (*Reply, error) +} + +type Chunk struct { + Value int32 `pb:"1"` +} + +type Reply struct { + Value int32 `pb:"1"` +} diff --git a/cmd/mobyextgen/testdata/runtime_contract.go.txt b/cmd/mobyextgen/testdata/runtime_contract.go.txt index 8b71d39..7eb015c 100644 --- a/cmd/mobyextgen/testdata/runtime_contract.go.txt +++ b/cmd/mobyextgen/testdata/runtime_contract.go.txt @@ -15,6 +15,8 @@ type Service interface { Remove(context.Context, *Request) error Download(context.Context, *Request, stream.Sender[Chunk]) error DownloadAgain(context.Context, *Request, stream.Sender[Chunk]) error + Upload(context.Context, stream.Receiver[Chunk]) (*Reply, error) + UploadAgain(context.Context, stream.Receiver[Chunk]) (*AlternateReply, error) } type Request struct { @@ -30,4 +32,9 @@ type Reply struct { Values []int32 `pb:"1"` } +type AlternateReply struct { + Values []int32 `pb:"1"` + Marker string `pb:"2"` +} + var Point = ext.DefinePoint[Service]("org.example.generated.runtime.v1") diff --git a/cmd/mobyextgen/testdata/runtime_transport_test.go.txt b/cmd/mobyextgen/testdata/runtime_transport_test.go.txt new file mode 100644 index 0000000..0b4ca7a --- /dev/null +++ b/cmd/mobyextgen/testdata/runtime_transport_test.go.txt @@ -0,0 +1,127 @@ +// SPDX-FileCopyrightText: Copyright The Moby Authors +// SPDX-License-Identifier: Apache-2.0 + +package contract_test + +import ( + "context" + "io" + "sync" + "sync/atomic" + "testing" + "testing/synctest" + + contract "__CONTRACT_IMPORT__" + protogen "__PROTOGEN_IMPORT__" + + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/metadata" + "google.golang.org/grpc/status" + "google.golang.org/protobuf/proto" + "gotest.tools/v3/assert" +) + +type transportSource struct { + reads atomic.Int32 +} + +func (s *transportSource) Recv() (*contract.Chunk, error) { + s.reads.Add(1) + return &contract.Chunk{Value: 1}, nil +} + +type transportStream struct { + ctx context.Context + entered chan struct{} + release <-chan struct{} + finalErr error + enteredOnce sync.Once +} + +func (s *transportStream) Header() (metadata.MD, error) { + return nil, nil +} + +func (s *transportStream) Trailer() metadata.MD { + return nil +} + +func (s *transportStream) CloseSend() error { + return nil +} + +func (s *transportStream) Context() context.Context { + return s.ctx +} + +func (s *transportStream) SendMsg(any) error { + s.enteredOnce.Do(func() { close(s.entered) }) + <-s.release + return io.EOF +} + +func (s *transportStream) RecvMsg(any) error { + return s.finalErr +} + +type transportConn struct { + stream *transportStream +} + +func (c *transportConn) Invoke(context.Context, string, any, any, ...grpc.CallOption) error { + return status.Error(codes.Unimplemented, "unexpected unary RPC") +} + +func (c *transportConn) NewStream(ctx context.Context, _ *grpc.StreamDesc, _ string, _ ...grpc.CallOption) (grpc.ClientStream, error) { + c.stream.ctx = ctx + return c.stream, nil +} + +func TestClientStreamingTransportBoundary(t *testing.T) { + t.Parallel() + synctest.Test(t, func(t *testing.T) { + parent, cancelParent := context.WithCancel(context.Background()) + t.Cleanup(cancelParent) + + release := newRelease() + finalErr := detailError(codes.PermissionDenied, "transport final status", "TRANSPORT_FINAL") + stream := &transportStream{ + entered: make(chan struct{}), + release: release.ch, + finalErr: finalErr, + } + conn := &transportConn{stream: stream} + client := protogen.NewClient(conn) + source := &transportSource{} + result := make(chan uploadResult, 1) + done := make(chan struct{}) + go func() { + reply, err := client.Upload(parent, source) + result <- uploadResult{reply: reply, err: err} + close(done) + }() + t.Cleanup(func() { + release.Close() + <-done + }) + + <-stream.entered + synctest.Wait() + assert.Equal(t, source.reads.Load(), int32(1)) + select { + case <-done: + t.Fatal("Upload completed while SendMsg was blocked") + default: + } + + release.Close() + got := <-result + assert.Assert(t, got.reply == nil) + assertDetail(t, got.err, codes.PermissionDenied, "transport final status", "TRANSPORT_FINAL") + assert.Assert(t, proto.Equal(status.Convert(got.err).Proto(), status.Convert(finalErr).Proto())) + assert.Equal(t, source.reads.Load(), int32(1)) + assert.Equal(t, stream.ctx.Err(), context.Canceled) + assert.NilError(t, parent.Err()) + }) +} diff --git a/cmd/mobyextgen/wire.go b/cmd/mobyextgen/wire.go index 7b479e8..a2262d7 100644 --- a/cmd/mobyextgen/wire.go +++ b/cmd/mobyextgen/wire.go @@ -81,6 +81,8 @@ func emitWire(pt point) ([]byte, error) { fmt.Fprintf(&b, "\t%s(context.Context, *%s) (*%s, error)\n", m.name, m.request, m.response) case methodServerStreaming: fmt.Fprintf(&b, "\t%s(*%s, grpc.ServerStreamingServer[%s]) error\n", m.name, m.request, m.response) + case methodClientStreaming: + fmt.Fprintf(&b, "\t%s(grpc.ClientStreamingServer[%s, %s]) error\n", m.name, m.request, m.response) } } fmt.Fprintln(&b, "}") @@ -114,6 +116,8 @@ func emitWire(pt point) ([]byte, error) { fmt.Fprintf(&b, "\t\t\t{StreamName: %q, Handler: %s", m.name, handlerName(m)) if m.kind == methodServerStreaming { fmt.Fprint(&b, ", ServerStreams: true") + } else { + fmt.Fprint(&b, ", ClientStreams: true") } fmt.Fprintln(&b, "},") } @@ -149,6 +153,12 @@ func emitWire(pt point) ([]byte, error) { return srv.(%[3]sServer).%[4]s(in, &grpc.GenericServerStream[%[2]s, %[5]s]{ServerStream: stream}) } `, handlerName(m), m.request, svc, m.name, m.response) + case methodClientStreaming: + fmt.Fprintf(&b, ` + func %[1]s(srv any, stream grpc.ServerStream) error { + return srv.(%[2]sServer).%[3]s(&grpc.GenericServerStream[%[4]s, %[5]s]{ServerStream: stream}) + } + `, handlerName(m), svc, m.name, m.request, m.response) } } @@ -173,6 +183,8 @@ func emitWire(pt point) ([]byte, error) { fmt.Fprintf(&b, "\t%s(ctx context.Context, in *%s, opts ...grpc.CallOption) (*%s, error)\n", m.name, m.request, m.response) case methodServerStreaming: fmt.Fprintf(&b, "\t%s(ctx context.Context, in *%s, opts ...grpc.CallOption) (grpc.ServerStreamingClient[%s], error)\n", m.name, m.request, m.response) + case methodClientStreaming: + fmt.Fprintf(&b, "\t%s(ctx context.Context, opts ...grpc.CallOption) (grpc.ClientStreamingClient[%s, %s], error)\n", m.name, m.request, m.response) } } fmt.Fprintln(&b, "}") @@ -215,6 +227,16 @@ func emitWire(pt point) ([]byte, error) { } return x, nil } + `, m.name, m.request, m.response, streamIndices[m.name], methodConst(m)) + case methodClientStreaming: + fmt.Fprintf(&b, ` + func (c *serviceClient) %[1]s(ctx context.Context, opts ...grpc.CallOption) (grpc.ClientStreamingClient[%[2]s, %[3]s], error) { + stream, err := c.cc.NewStream(ctx, &serviceDesc.Streams[%[4]d], %[5]s, append([]grpc.CallOption{grpc.StaticMethod()}, opts...)...) + if err != nil { + return nil, err + } + return &grpc.GenericClientStream[%[2]s, %[3]s]{ClientStream: stream}, nil + } `, m.name, m.request, m.response, streamIndices[m.name], methodConst(m)) } } @@ -291,20 +313,43 @@ func emitWire(pt point) ([]byte, error) { fmt.Fprintf(&b, ` func (s *grpcServer) %[1]s(req *%[2]s, out grpc.ServerStreamingServer[%[3]s]) error { sender := &mobyext%[1]sSender{stream: out} - return s.impl.%[1]s(out.Context(), %[5]sFromProto(req), sender) + return s.impl.%[1]s(out.Context(), %[4]sFromProto(req), sender) } type mobyext%[1]sSender struct { stream grpc.ServerStreamingServer[%[3]s] } - func (s *mobyext%[1]sSender) Send(message *%[6]s.%[7]s) error { + func (s *mobyext%[1]sSender) Send(message *%[5]s.%[6]s) error { if message == nil { - return status.Error(codes.Internal, %[8]q) + return status.Error(codes.Internal, %[7]q) + } + return s.stream.Send(%[8]sToProto(message)) + } + `, m.name, m.request, m.response, lowerFirst(m.request), cpkg, m.response, m.name+": nil message", lowerFirst(m.response)) + case methodClientStreaming: + fmt.Fprintf(&b, ` + func (s *grpcServer) %[1]s(stream grpc.ClientStreamingServer[%[2]s, %[3]s]) error { + receiver := &mobyext%[1]sReceiver{stream: stream} + resp, err := s.impl.%[1]s(stream.Context(), receiver) + if err != nil { + return err + } + return stream.SendAndClose(%[4]sToProto(resp)) + } + + type mobyext%[1]sReceiver struct { + stream grpc.ClientStreamingServer[%[2]s, %[3]s] + } + + func (r *mobyext%[1]sReceiver) Recv() (*%[5]s.%[6]s, error) { + message, err := r.stream.Recv() + if err != nil { + return nil, err } - return s.stream.Send(%[9]sToProto(message)) + return %[7]sFromProto(message), nil } - `, m.name, m.request, m.response, lowerFirst(m.name), lowerFirst(m.request), cpkg, m.response, m.name+": nil message", lowerFirst(m.response)) + `, m.name, m.request, m.response, lowerFirst(m.response), cpkg, m.request, lowerFirst(m.request)) } } @@ -360,6 +405,46 @@ func emitWire(pt point) ([]byte, error) { } } `, m.name, cpkg, m.request, m.response, lowerFirst(m.request), lowerFirst(m.response)) + case methodClientStreaming: + fmt.Fprintf(&b, ` + func (c *grpcClient) %[1]s(ctx context.Context, source extensionstream.Receiver[%[2]s.%[3]s]) (*%[2]s.%[4]s, error) { + ctx, cancel := context.WithCancel(ctx) + defer cancel() + stream, err := c.client.%[1]s(ctx) + if err != nil { + return nil, err + } + for { + message, err := source.Recv() + if err != nil { + if !errors.Is(err, io.EOF) { + return nil, err + } + resp, err := stream.CloseAndRecv() + if err != nil { + return nil, err + } + return %[5]sFromProto(resp), nil + } + if message == nil { + return nil, status.Error(codes.InvalidArgument, %[6]q) + } + if err := ctx.Err(); err != nil { + return nil, status.FromContextError(err).Err() + } + if err := stream.Send(%[7]sToProto(message)); err != nil { + if !errors.Is(err, io.EOF) { + return nil, err + } + resp, err := stream.CloseAndRecv() + if err != nil { + return nil, err + } + return %[5]sFromProto(resp), nil + } + } + } + `, m.name, cpkg, m.request, m.response, lowerFirst(m.response), m.name+": nil message", lowerFirst(m.request)) } } diff --git a/docs/AUTHORING.md b/docs/AUTHORING.md index 22a8045..7e82337 100644 --- a/docs/AUTHORING.md +++ b/docs/AUTHORING.md @@ -150,10 +150,10 @@ The generated package exports: - `ClientPoint`, for host wiring to a launched provider; - `NewClient(conn)`, which returns the handwritten Go interface over a gRPC connection. -A contract can also declare server-streaming methods with `stream.Sender[Message]`. +A contract can also declare server-streaming methods with `stream.Sender[Message]` and client-streaming methods with `stream.Receiver[Message]`. The handwritten provider interface keeps its Go types. The generated adapters make these methods available as protobuf streaming RPCs. -The generator rejects all other stream parameter and result shapes. +The generator rejects bidirectional streams and all other stream parameter and result shapes. The contract package does not import protobuf packages; generated code belongs in `protogen/`. The generator derives the import path from `go.mod` and the proto file name diff --git a/docs/DESIGN.md b/docs/DESIGN.md index d30cdef..d2fa4f3 100644 --- a/docs/DESIGN.md +++ b/docs/DESIGN.md @@ -229,11 +229,12 @@ The Go interface and message structs are the source of truth. `mobyextgen` generates the proto, protobuf message code, gRPC service, client and server points, adapters, and conversions without `protoc`, plugins, or other `PATH` tools. In-process providers implement Go directly; out-of-process providers use gRPC. -The generator supports two method kinds: unary and server-streaming. +The generator supports three method kinds: unary, server-streaming, and client-streaming. A unary method has the shape `M(context.Context, *Request) (*Reply, error)`. The generator also accepts the legacy unary shape `M(context.Context, *Request) error`. A server-streaming method has the shape `M(context.Context, *Request, stream.Sender[Chunk]) error`. -The generator rejects all other stream signatures. +A client-streaming method has the shape `M(context.Context, stream.Receiver[Chunk]) (*Reply, error)`. +The generator rejects bidirectional streams and all other stream signatures. A stream message type must be a local struct type that is not generic. ### Why not channels? @@ -260,8 +261,16 @@ The cost is two framework interfaces in addition to the standard library types. The enclosing method stays active until the stream is complete. The return value of the method sets the final success or failure. +Callbacks run synchronously. +Thus, when a transport send blocks, the adapter stops reading from the source. +The adapter does this without a background callback pump or an unlimited buffer. +A normal input EOF closes a client stream, and the adapter then waits for the final reply of the provider. +This EOF alone does not show that the method is successful. +If the provider stops a client stream early, the adapter sees the final status only when the source returns from `Recv`. +Thus, a source that can block must stop when the context of the caller is canceled. A normal server-streaming EOF alone does not show success. The final `OK` status of the provider shows success. +Adapters reject a nil stream message from the caller with `InvalidArgument`. Adapters reject a nil stream message from a provider with `Internal`. Generated adapters return provider errors without changes. From 59b1bee374ee15cce16762000df08fa4f6dcb284 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Pawe=C5=82=20Gronowski?= Date: Thu, 15 Oct 2026 20:32:06 +0200 Subject: [PATCH 7/7] cmd/mobyextgen: Reject methods whose first parameter is not a context MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The generator treated the first method parameter as the context without checking its type. It accepted contracts whose first parameter was, for example, a request pointer or an int. The generated adapters pass a context.Context, so they did not compile against such contracts. The first parameter must now name context.Context. The check resolves the file's imports, so it accepts a renamed import of the standard context package and rejects a Context type from any other package. Test contracts that declared ctx interface{} now use context.Context. Signed-off-by: Paweł Gronowski --- cmd/mobyextgen/generation_test.go | 21 +++++++++----- cmd/mobyextgen/parse.go | 47 +++++++++++++++++++++++++------ cmd/mobyextgen/parse_test.go | 15 ++++++---- cmd/mobyextgen/streaming_test.go | 39 +++++++++++++++++++++++++ 4 files changed, 101 insertions(+), 21 deletions(-) diff --git a/cmd/mobyextgen/generation_test.go b/cmd/mobyextgen/generation_test.go index 30a3a3b..2dccfd5 100644 --- a/cmd/mobyextgen/generation_test.go +++ b/cmd/mobyextgen/generation_test.go @@ -39,9 +39,11 @@ func TestGeneratedFilesInheritContractNotices(t *testing.T) { assert.NilError(t, os.WriteFile(filepath.Join(dir, "go.mod"), []byte("module example.com/contract\n"), 0o644)) const contract = `package contract +import "context" + // SPDX-FileCopyrightText: ignore comments after the package clause // SPDX-License-Identifier: ignore comments after the package clause -type Runtime interface{ Do(ctx interface{}, req *Req) (*Resp, error) } +type Runtime interface{ Do(ctx context.Context, req *Req) (*Resp, error) } type Req struct{} type Resp struct{} ` @@ -88,8 +90,9 @@ func TestSingleMessageField(t *testing.T) { func TestInitialismFieldBridgesNames(t *testing.T) { const src = `package p +import "context" import "github.com/moby/extensions" -type S interface{ Do(ctx interface{}, req *Req) (*Resp, error) } +type S interface{ Do(ctx context.Context, req *Req) (*Resp, error) } type Req struct{ ContainerID string ` + "`pb:\"1\"`" + ` } type Resp struct{ Ok bool ` + "`pb:\"1\"`" + ` } var Point = extensions.DefinePoint[S]("test.gen.v1") @@ -114,10 +117,11 @@ var Point = extensions.DefinePoint[S]("test.gen.v1") func TestServiceContractWithoutAPoint(t *testing.T) { const contract = `package p +import "context" type Req struct{ Name string ` + "`pb:\"1\"`" + ` } type Resp struct{ Ok bool ` + "`pb:\"1\"`" + ` } -type Runtime interface{ Do(ctx interface{}, req *Req) (*Resp, error) } +type Runtime interface{ Do(ctx context.Context, req *Req) (*Resp, error) } ` pt, err := parseServiceSource(t, contract, "my.proto.pkg.v1.Runtime") assert.NilError(t, err) @@ -140,13 +144,14 @@ type Runtime interface{ Do(ctx interface{}, req *Req) (*Resp, error) } func TestOrdinaryPointGeneratesPublicationAndClientAPIs(t *testing.T) { const contract = `package p +import "context" import "github.com/moby/extensions" type Req struct{ Name string ` + "`pb:\"1\"`" + ` } type Resp struct{ Ok bool ` + "`pb:\"1\"`" + ` } type Runtime interface { - Do(ctx interface{}, req *Req) (*Resp, error) - Delete(ctx interface{}, req *Req) error + Do(ctx context.Context, req *Req) (*Resp, error) + Delete(ctx context.Context, req *Req) error } var Point = extensions.DefinePoint[Runtime]("my.proto.pkg.v1") ` @@ -173,8 +178,9 @@ var Point = extensions.DefinePoint[Runtime]("my.proto.pkg.v1") func TestReservedFieldNumbers(t *testing.T) { const header = `package p +import "context" import "github.com/moby/extensions" -type S interface{ Do(ctx interface{}, req *Req) (*Resp, error) } +type S interface{ Do(ctx context.Context, req *Req) (*Resp, error) } type Resp struct{ Ok bool ` + "`pb:\"1\"`" + ` } var Point = extensions.DefinePoint[S]("test.gen.v1") @@ -224,8 +230,9 @@ var Point = extensions.DefinePoint[S]("test.gen.v1") func TestSinglePointCardinality(t *testing.T) { const contract = `package p +import "context" import "github.com/moby/extensions" -type S interface{ Do(ctx interface{}, req *Req) (*Resp, error) } +type S interface{ Do(ctx context.Context, req *Req) (*Resp, error) } type Req struct{ Name string ` + "`pb:\"1\"`" + ` } type Resp struct{ Ok bool ` + "`pb:\"1\"`" + ` } ` diff --git a/cmd/mobyextgen/parse.go b/cmd/mobyextgen/parse.go index a1202de..2fc052d 100644 --- a/cmd/mobyextgen/parse.go +++ b/cmd/mobyextgen/parse.go @@ -358,6 +358,8 @@ func classifyMethodParameter(files []*ast.File, expr ast.Expr, index int) (metho return methodParameter{}, errors.New("stream parameters cannot replace context.Context") case isStream: return param, nil + case index == 0 && !isContextType(files, expr): + return methodParameter{}, errors.New("must be context.Context") case index == 0: return methodParameter{kind: methodContextParameter}, nil } @@ -429,24 +431,51 @@ func resolveStreamImport(file *ast.File, packageExpr ast.Expr, use string) error if !ok { return fmt.Errorf("%s must use a package imported from %q", use, streamImportPath) } + importPath, ok := importedPath(file, ident.Name) + if !ok { + return fmt.Errorf("%s package name %q is not imported from %q", use, ident.Name, streamImportPath) + } + if importPath != streamImportPath { + return fmt.Errorf("%s must be imported from %q", use, streamImportPath) + } + return nil +} + +// importedPath returns the path of the package that file imports as name. +func importedPath(file *ast.File, name string) (string, bool) { for _, spec := range file.Imports { importPath, err := strconv.Unquote(spec.Path.Value) if err != nil { continue } - name := path.Base(importPath) + local := path.Base(importPath) if spec.Name != nil { - name = spec.Name.Name - } - if name != ident.Name { - continue + local = spec.Name.Name } - if importPath != streamImportPath { - return fmt.Errorf("%s must be imported from %q", use, streamImportPath) + if local == name { + return importPath, true } - return nil } - return fmt.Errorf("%s package name %q is not imported from %q", use, ident.Name, streamImportPath) + return "", false +} + +// isContextType reports whether expr names context.Context under any import +// name. +func isContextType(files []*ast.File, expr ast.Expr) bool { + selector, ok := expr.(*ast.SelectorExpr) + if !ok || selector.Sel.Name != "Context" { + return false + } + ident, ok := selector.X.(*ast.Ident) + if !ok { + return false + } + file := fileForExpr(files, expr) + if file == nil { + return false + } + importPath, ok := importedPath(file, ident.Name) + return ok && importPath == "context" } func streamMessageArgument(expr ast.Expr) (string, error) { diff --git a/cmd/mobyextgen/parse_test.go b/cmd/mobyextgen/parse_test.go index e5eacd0..7e98718 100644 --- a/cmd/mobyextgen/parse_test.go +++ b/cmd/mobyextgen/parse_test.go @@ -27,8 +27,9 @@ func parseServiceSource(t *testing.T, src, service string) (point, error) { func TestContractValidation(t *testing.T) { const header = `package p +import "context" import "github.com/moby/extensions" -type S interface{ Do(ctx interface{}, req *Req) (*Resp, error) } +type S interface{ Do(ctx context.Context, req *Req) (*Resp, error) } type Resp struct{ Ok bool ` + "`pb:\"1\"`" + ` } var Point = extensions.DefinePoint[S]("test.gen.v1") ` @@ -127,8 +128,9 @@ var Point = extensions.DefinePoint[S]("test.gen.v1") func TestPointServiceName(t *testing.T) { const contract = `package p +import "context" import "github.com/moby/extensions" -type S interface{ Do(ctx interface{}, req *Req) (*Resp, error) } +type S interface{ Do(ctx context.Context, req *Req) (*Resp, error) } type Req struct{ Name string ` + "`pb:\"1\"`" + ` } type Resp struct{ Ok bool ` + "`pb:\"1\"`" + ` } ` @@ -140,9 +142,10 @@ type Resp struct{ Ok bool ` + "`pb:\"1\"`" + ` } func TestExplicitNonPointService(t *testing.T) { const contract = `package p +import "context" type Req struct{ Name string ` + "`pb:\"1\"`" + ` } type Resp struct{ Ok bool ` + "`pb:\"1\"`" + ` } -type Runtime interface{ Do(ctx interface{}, req *Req) (*Resp, error) } +type Runtime interface{ Do(ctx context.Context, req *Req) (*Resp, error) } ` pt, err := parseServiceSource(t, contract, "example.api.v1.Runtime") assert.NilError(t, err) @@ -157,10 +160,11 @@ type Runtime interface{ Do(ctx interface{}, req *Req) (*Resp, error) } func TestExplicitServiceRejectsPointContract(t *testing.T) { _, err := parseServiceSource(t, `package p +import "context" import "github.com/moby/extensions" type Req struct{} type Resp struct{} -type Runtime interface{ Do(ctx interface{}, req *Req) (*Resp, error) } +type Runtime interface{ Do(ctx context.Context, req *Req) (*Resp, error) } var Point = extensions.DefinePoint[Runtime]("example.api.v1") `, "example.api.v1.Runtime") @@ -169,10 +173,11 @@ var Point = extensions.DefinePoint[Runtime]("example.api.v1") func TestRejectsGroupedExtraParameter(t *testing.T) { _, err := parseSource(t, `package p +import "context" import "github.com/moby/extensions" type Req struct{} type Resp struct{} -type S interface{ Do(ctx interface{}, first, second *Req) (*Resp, error) } +type S interface{ Do(ctx context.Context, first, second *Req) (*Resp, error) } var Point = extensions.DefinePoint[S]("test.gen.v1") `) assert.ErrorContains(t, err, "unsupported signature") diff --git a/cmd/mobyextgen/streaming_test.go b/cmd/mobyextgen/streaming_test.go index 2c33455..7b5f9a8 100644 --- a/cmd/mobyextgen/streaming_test.go +++ b/cmd/mobyextgen/streaming_test.go @@ -90,6 +90,45 @@ type Chunk struct{} `), want: acceptedMethodShapes, }, + { + name: "client stream without a context", + source: streamingPointSource(`type Service interface { + Upload(*Request, s.Receiver[Chunk]) (*Reply, error) +} +type Request struct{} +type Chunk struct{} +type Reply struct{} +`), + want: "parameter 1: must be context.Context", + }, + { + name: "server stream with a non-context first parameter", + source: streamingPointSource(`type Service interface { + Download(int, *Request, s.Sender[Chunk]) error +} +type Request struct{} +type Chunk struct{} +`), + want: "parameter 1: must be context.Context", + }, + { + name: "context from another package", + source: `package contract + +import ( + stdctx "example/context" + ext "github.com/moby/extensions" + s "github.com/moby/extensions/stream" +) +type Service interface { + Download(stdctx.Context, *Request, s.Sender[Chunk]) error +} +type Request struct{} +type Chunk struct{} +var Point = ext.DefinePoint[Service]("org.example.contract.v1") +`, + want: "parameter 1: must be context.Context", + }, { name: "pointer stream message argument", source: streamingPointSource(`type Service interface {