diff --git a/cmd/mobyextgen/descriptor.go b/cmd/mobyextgen/descriptor.go index 813bfd2..0e263f8 100644 --- a/cmd/mobyextgen/descriptor.go +++ b/cmd/mobyextgen/descriptor.go @@ -80,11 +80,18 @@ 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 == methodClientStreaming { + method.ClientStreaming = new(true) + } + 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/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/model.go b/cmd/mobyextgen/model.go index 8ad2217..5b1c3a2 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 } @@ -24,12 +25,22 @@ 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 + methodClientStreaming +) + type message struct { name string fields []field diff --git a/cmd/mobyextgen/parse.go b/cmd/mobyextgen/parse.go index c3a7fce..2fc052d 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, "") } @@ -77,6 +80,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 { @@ -98,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 } @@ -220,7 +224,34 @@ func structHasPBTag(st *ast.StructType) bool { return false } -func parseMethods(iface *ast.InterfaceType) ([]method, error) { +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(files []*ast.File, iface *ast.InterfaceType) ([]method, error) { var methods []method for _, m := range iface.Methods.List { ft, ok := m.Type.(*ast.FuncType) @@ -228,31 +259,287 @@ func parseMethods(iface *ast.InterfaceType) ([]method, error) { continue } name := m.Names[0].Name - 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(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 { - case len(res) == 1 && isIdent(res[0], "error"): - methods = append(methods, method{name: name, request: req, response: name + "Response", bareError: true}) - case len(res) == 2 && isIdent(res[1], "error"): - resp, err := pointerIdent(res[0]) + 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: 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, + }) + 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, fmt.Errorf("method %q response: %w", name, err) + return nil, methodShapeError(name) } - methods = append(methods, method{name: name, request: req, response: resp}) + methods = append(methods, method{ + kind: methodClientStreaming, name: name, + request: params[1].message, response: response, 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 + methodSenderParameter + methodReceiverParameter +) + +type methodParameter struct { + kind methodParameterKind + message string +} + +const acceptedMethodShapes = "accepted shapes are M(context.Context, *Request) (*Reply, 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) +} + +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 && !isContextType(files, expr): + return methodParameter{}, errors.New("must be context.Context") + case index == 0: + return methodParameter{kind: methodContextParameter}, nil + } + message, err := pointerIdent(expr) + if err != nil { + return methodParameter{}, err + } + return methodParameter{kind: methodMessageParameter, message: message}, nil +} + +// 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 + 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" && selector.Sel.Name != "Receiver") { + 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) + } + 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 { + 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) + } + 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 + } + local := path.Base(importPath) + if spec.Name != nil { + local = spec.Name.Name + } + if local == name { + return importPath, true + } + } + 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) { + 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 { + 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 { @@ -474,17 +761,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..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,12 +160,25 @@ 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") assert.ErrorContains(t, err, "cannot be used with an ordinary Point") } + +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 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/proto.go b/cmd/mobyextgen/proto.go index 4a7423a..6a89606 100644 --- a/cmd/mobyextgen/proto.go +++ b/cmd/mobyextgen/proto.go @@ -21,9 +21,21 @@ 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 { - fmt.Fprintf(&b, " rpc %s(%s) returns (%s);\n", m.name, m.request, m.response) + emitComment(&b, " ", m.doc) + request, response := m.request, m.response + switch m.kind { + 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) + } + 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..ffe1cb7 --- /dev/null +++ b/cmd/mobyextgen/runtime_test.go @@ -0,0 +1,114 @@ +// 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)) + 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 { + 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", + }, + { + 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)) + 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..7b5f9a8 --- /dev/null +++ b/cmd/mobyextgen/streaming_test.go @@ -0,0 +1,290 @@ +// SPDX-FileCopyrightText: Copyright The Moby Authors +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "go/parser" + "go/token" + "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 + // 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), 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", "UploadChunk", "UploadReply"} { + 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: "unsupported parameter shape", + source: streamingPointSource(`type Service interface { + Sync(stdctx.Context, s.Receiver[Chunk], s.Sender[Chunk]) error +} +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: "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 { + 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 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 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) + 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);", + "rpc Upload(stream UploadChunk) returns (UploadReply);", + } { + 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), 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 new file mode 100644 index 0000000..ee5040d --- /dev/null +++ b/cmd/mobyextgen/testdata/runtime_behavior_test.go.txt @@ -0,0 +1,652 @@ +// SPDX-FileCopyrightText: Copyright The Moby Authors +// SPDX-License-Identifier: Apache-2.0 + +package contract_test + +import ( + "context" + "errors" + "io" + "net" + "sync" + "sync/atomic" + "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 + + 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) + +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) 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) }) + } +} + +func (p *provider) signalDownloadDone() { + if p.downloadDone != nil { + p.downloadDoneOnce.Do(func() { close(p.downloadDone) }) + } +} + +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 +} + +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 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 + 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("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}, + 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) + } + }) + + 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) + } + }) + }) + } +} + +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 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() + 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) +} + +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 new file mode 100644 index 0000000..7eb015c --- /dev/null +++ b/cmd/mobyextgen/testdata/runtime_contract.go.txt @@ -0,0 +1,40 @@ +// 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 + Upload(context.Context, stream.Receiver[Chunk]) (*Reply, error) + UploadAgain(context.Context, stream.Receiver[Chunk]) (*AlternateReply, 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"` +} + +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_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_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/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 6af9ae3..a2262d7 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, ")") @@ -59,7 +75,15 @@ func emitWire(pt point) ([]byte, error) { `, svc) } for _, m := range pt.methods { - fmt.Fprintf(&b, "\t%s(context.Context, *%s) (*%s, error)\n", m.name, m.request, m.response) + emitComment(&b, "\t", m.doc) + 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) + case methodClientStreaming: + fmt.Fprintf(&b, "\t%s(grpc.ClientStreamingServer[%s, %s]) error\n", m.name, m.request, m.response) + } } fmt.Fprintln(&b, "}") @@ -78,12 +102,33 @@ 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},") + 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") + } else { + fmt.Fprint(&b, ", ClientStreams: true") + } + fmt.Fprintln(&b, "},") + } + fmt.Fprintln(&b, "\t\t},") } - fmt.Fprintln(&b, "\t},\n}") + 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 { @@ -98,6 +143,23 @@ 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) + 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) + } } if pt.isPoint { @@ -115,7 +177,15 @@ func emitWire(pt point) ([]byte, error) { `, svc) } for _, m := range pt.methods { - fmt.Fprintf(&b, "\t%s(ctx context.Context, in *%s, opts ...grpc.CallOption) (*%s, error)\n", m.name, m.request, m.response) + emitComment(&b, "\t", m.doc) + 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) + 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, "}") @@ -130,7 +200,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 { @@ -139,6 +211,34 @@ 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)) + 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)) + } } if pt.isPoint { @@ -187,8 +287,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 @@ -196,8 +298,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 { @@ -206,6 +308,48 @@ 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(), %[4]sFromProto(req), sender) + } + + type mobyext%[1]sSender struct { + stream grpc.ServerStreamingServer[%[3]s] + } + + func (s *mobyext%[1]sSender) Send(message *%[5]s.%[6]s) error { + if message == nil { + 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 %[7]sFromProto(message), nil + } + `, m.name, m.request, m.response, lowerFirst(m.response), cpkg, m.request, lowerFirst(m.request)) } } @@ -215,15 +359,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 { @@ -232,6 +378,73 @@ 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)) + 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)) } } @@ -242,6 +455,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/docs/AUTHORING.md b/docs/AUTHORING.md index 4d86c43..7e82337 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]` 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 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 from the inferred service identity. diff --git a/docs/DESIGN.md b/docs/DESIGN.md index dca2f65..d2fa4f3 100644 --- a/docs/DESIGN.md +++ b/docs/DESIGN.md @@ -229,6 +229,56 @@ 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 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`. +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? + +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. +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. +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. 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) } diff --git a/stream/stream.go b/stream/stream.go new file mode 100644 index 0000000..e518a57 --- /dev/null +++ b/stream/stream.go @@ -0,0 +1,106 @@ +// 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 + +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 +// 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) +} + +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) + } +}