Revert "feat: expose invalid argument error to clients in bidirectional streaming (#4795) (#4819)" (#5303)

This reverts commit 830ba27.
Unfortunately this suffered from a bug that could cause
a panic. Revert until we can reintroduce it without the
bug.

Author: johanbrandhorst

examples/internal/integration/BUILD.bazel

@@ -25,7 +25,6 @@ go_test(
         "@org_golang_google_protobuf//encoding/protojson",
         "@org_golang_google_protobuf//proto",
         "@org_golang_google_protobuf//testing/protocmp",
-        "@org_golang_google_protobuf//types/known/durationpb",
         "@org_golang_google_protobuf//types/known/emptypb",
         "@org_golang_google_protobuf//types/known/fieldmaskpb",
         "@org_golang_google_protobuf//types/known/structpb",

examples/internal/integration/integration_test.go

@@ -28,7 +28,6 @@ import (
 	"google.golang.org/protobuf/encoding/protojson"
 	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/testing/protocmp"
-	"google.golang.org/protobuf/types/known/durationpb"
 	"google.golang.org/protobuf/types/known/emptypb"
 	fieldmaskpb "google.golang.org/protobuf/types/known/fieldmaskpb"
 	"google.golang.org/protobuf/types/known/structpb"
@@ -523,7 +522,6 @@ func TestABE(t *testing.T) {
 	testABEDownload(t, 8088)
 	testABEBulkEcho(t, 8088)
 	testABEBulkEchoZeroLength(t, 8088)
-	testABEBulkEchoDurationError(t, 8088)
 	testAdditionalBindings(t, 8088)
 	testABERepeated(t, 8088)
 	testABEExists(t, 8088)
@@ -1451,98 +1449,6 @@ func testABEBulkEchoZeroLength(t *testing.T, port int) {
 	}
 }
 
-func testABEBulkEchoDurationError(t *testing.T, port int) {
-	reqr, reqw := io.Pipe()
-	var wg sync.WaitGroup
-	var want []*durationpb.Duration
-	wg.Add(1)
-	go func() {
-		defer wg.Done()
-		defer reqw.Close()
-		for i := 0; i < 10; i++ {
-			s := fmt.Sprintf("%d.123s", i)
-			if i == 5 {
-				s = "invalidDurationFormat"
-			}
-			buf, err := marshaler.Marshal(s)
-			if err != nil {
-				t.Errorf("marshaler.Marshal(%v) failed with %v; want success", s, err)
-				return
-			}
-			if _, err = reqw.Write(buf); err != nil {
-				t.Errorf("reqw.Write(%q) failed with %v; want success", string(buf), err)
-				return
-			}
-			want = append(want, &durationpb.Duration{Seconds: int64(i), Nanos: int32(0.123 * 1e9)})
-		}
-	}()
-	apiURL := fmt.Sprintf("http://localhost:%d/v1/example/a_bit_of_everything/echo_duration", port)
-	req, err := http.NewRequest("POST", apiURL, reqr)
-	if err != nil {
-		t.Errorf("http.NewRequest(%q, %q, reqr) failed with %v; want success", "POST", apiURL, err)
-		return
-	}
-	req.Header.Set("Content-Type", "application/json")
-	req.Header.Set("Transfer-Encoding", "chunked")
-	resp, err := http.DefaultClient.Do(req)
-	if err != nil {
-		t.Errorf("http.Post(%q, %q, req) failed with %v; want success", apiURL, "application/json", err)
-		return
-	}
-	defer resp.Body.Close()
-	if got, want := resp.StatusCode, http.StatusOK; got != want {
-		t.Errorf("resp.StatusCode = %d; want %d", got, want)
-	}
-
-	var got []*durationpb.Duration
-	var invalidArgumentCount int
-	wg.Add(1)
-	go func() {
-		defer wg.Done()
-
-		dec := marshaler.NewDecoder(resp.Body)
-		for i := 0; ; i++ {
-			var item struct {
-				Result json.RawMessage        `json:"result"`
-				Error  map[string]interface{} `json:"error"`
-			}
-			err := dec.Decode(&item)
-			if err == io.EOF {
-				break
-			}
-			if err != nil {
-				t.Errorf("dec.Decode(&item) failed with %v; want success; i = %d", err, i)
-			}
-			if len(item.Error) != 0 {
-				code, ok := item.Error["code"].(float64)
-				if !ok {
-					t.Errorf("item.Error[code] not found or not a number: %#v; i = %d", item.Error, i)
-				} else if int32(code) == 3 {
-					invalidArgumentCount++
-				} else {
-					t.Errorf("item.Error[code] = %v; want 3; i = %d", code, i)
-				}
-				continue
-			}
-
-			msg := new(durationpb.Duration)
-			if err := marshaler.Unmarshal(item.Result, msg); err != nil {
-				t.Errorf("marshaler.Unmarshal(%q, msg) failed with %v; want success", item.Result, err)
-			}
-			got = append(got, msg)
-		}
-
-		if invalidArgumentCount != 1 {
-			t.Errorf("got %d errors with code 3; want exactly 1", invalidArgumentCount)
-		}
-	}()
-
-	wg.Wait()
-	if diff := cmp.Diff(got, want[:5], protocmp.Transform()); diff != "" {
-		t.Error(diff)
-	}
-}
-
 func testAdditionalBindings(t *testing.T, port int) {
 	for i, f := range []func() *http.Response{
 		func() *http.Response {

examples/internal/proto/examplepb/flow_combination.pb.gw.go

@@ -272,7 +272,7 @@ var (
 
 	_ = template.Must(handlerTemplate.New("request-func-signature").Parse(strings.ReplaceAll(`
 {{ if and .Method.GetClientStreaming .Method.GetServerStreaming }}
-func request_{{ .Method.Service.GetName }}_{{ .Method.GetName }}_{{ .Index }}(ctx context.Context, marshaler runtime.Marshaler, client {{ .Method.Service.InstanceName }}Client, req *http.Request, pathParams map[string]string) ({{ .Method.Service.InstanceName }}_{{ .Method.GetName }}Client, runtime.ServerMetadata, chan error, error)
+func request_{{ .Method.Service.GetName }}_{{ .Method.GetName }}_{{ .Index }}(ctx context.Context, marshaler runtime.Marshaler, client {{ .Method.Service.InstanceName }}Client, req *http.Request, pathParams map[string]string) ({{ .Method.Service.InstanceName }}_{{ .Method.GetName }}Client, runtime.ServerMetadata, error)
 {{ else if .Method.GetServerStreaming }}
 func request_{{ .Method.Service.GetName }}_{{ .Method.GetName }}_{{ .Index }}(ctx context.Context, marshaler runtime.Marshaler, client {{ .Method.Service.InstanceName }}Client, req *http.Request, pathParams map[string]string) ({{ .Method.Service.InstanceName }}_{{ .Method.GetName }}Client, runtime.ServerMetadata, error)
 {{ else }}
@@ -457,12 +457,10 @@ var filter_{{ .Method.Service.GetName }}_{{ .Method.GetName }}_{{ .Index }} = {{
 	_ = template.Must(handlerTemplate.New("bidi-streaming-request-func").Parse(`
 {{ template "request-func-signature" . }} {
 	var metadata runtime.ServerMetadata
-	errChan := make(chan error, 1)
 	stream, err := client.{{ .Method.GetName }}(ctx)
 	if err != nil {
 		grpclog.Errorf("Failed to start streaming: %v", err)
-		close(errChan)
-		return nil, metadata, errChan, err
+		return nil, metadata, err
 	}
 	dec := marshaler.NewDecoder(req.Body)
 	handleSend := func() error {
@@ -482,10 +480,8 @@ var filter_{{ .Method.Service.GetName }}_{{ .Method.GetName }}_{{ .Index }} = {{
 		return nil
 	}
 	go func() {
-		defer close(errChan)
 		for {
 			if err := handleSend(); err != nil {
-				errChan <- err
 				break
 			}
 		}
@@ -496,10 +492,10 @@ var filter_{{ .Method.Service.GetName }}_{{ .Method.GetName }}_{{ .Index }} = {{
 	header, err := stream.Header()
 	if err != nil {
 		grpclog.Errorf("Failed to get header from client: %v", err)
-		return nil, metadata, errChan, err
+		return nil, metadata, err
 	}
 	metadata.HeaderMD = header
-	return stream, metadata, errChan, nil
+	return stream, metadata, nil
 }
 `))
 
@@ -741,25 +737,12 @@ func Register{{ $svc.GetName }}{{ $.RegisterFuncSuffix }}Client(ctx context.Cont
 			runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
 			return
 		}
-		{{if and $m.GetClientStreaming $m.GetServerStreaming }}
-		resp, md, reqErrChan, err := request_{{ $svc.GetName }}_{{ $m.GetName }}_{{ $b.Index }}(annotatedContext, inboundMarshaler, client, req, pathParams)
-		{{- else -}}
 		resp, md, err := request_{{ $svc.GetName }}_{{ $m.GetName }}_{{ $b.Index }}(annotatedContext, inboundMarshaler, client, req, pathParams)
-		{{- end }}
 		annotatedContext = runtime.NewServerMetadataContext(annotatedContext, md)
 		if err != nil {
 			runtime.HTTPError(annotatedContext, mux, outboundMarshaler, w, req, err)
 			return
 		}
-		{{- if and $m.GetClientStreaming $m.GetServerStreaming }}
-		go func() {
-			for err := range reqErrChan {
-				if err != nil && !errors.Is(err, io.EOF) {
-					runtime.HTTPStreamError(annotatedContext, mux, outboundMarshaler, w, req, err)
-				}
-			}
-		}()
-		{{- end }}
 		{{- if $m.GetServerStreaming }}
 		{{- if $b.ResponseBody }}
 		forward_{{ $svc.GetName }}_{{ $m.GetName }}_{{ $b.Index }}(annotatedContext, mux, outboundMarshaler, w, req, func() (proto.Message, error) {

examples/internal/proto/examplepb/stream.pb.gw.go

@@ -322,7 +322,7 @@ func TestApplyTemplateRequestWithClientStreaming(t *testing.T) {
 		},
 		{
 			serverStreaming: true,
-			sigWant:         `func request_ExampleService_Echo_0(ctx context.Context, marshaler runtime.Marshaler, client ExampleServiceClient, req *http.Request, pathParams map[string]string) (ExampleService_EchoClient, runtime.ServerMetadata, chan error, error) {`,
+			sigWant:         `func request_ExampleService_Echo_0(ctx context.Context, marshaler runtime.Marshaler, client ExampleServiceClient, req *http.Request, pathParams map[string]string) (ExampleService_EchoClient, runtime.ServerMetadata, error) {`,
 		},
 	} {
 		meth.ServerStreaming = proto.Bool(spec.serverStreaming)