Repository: goadesign/goa (go)

Issue:
grpc stream.Recv doesn't decode errors
The error handling for unary vs streaming gRPC clients differ significantly. The upshot is if a service returns a `*goa.ServiceError` from a server-streaming method, the Recv method returns instead a `*grpc.StatusError`. If I've traced it back correctly, grpc/codegen/service_data.go's streamRecvT and grpc/codegen/client.go's clientEndpointInitT show the difference. From the clientEndpointInitT code I see something like this:

```go
		res, err := inv.Invoke(ctx, v)
		if err != nil {
			resp := goagrpc.DecodeError(err)
			switch message := resp.(type) {
			case *goapb.ErrorResponse:
				return nil, goagrpc.NewServiceError(message)
			default:
				return nil, goa.Fault(err.Error())
			}
		}
```

whereas from the streamRecvT code I see something like this:

```go
	v, err := s.stream.Recv()
	if err != nil {
		return res, err
	}
```

I'm willing to craft a PR for this with a little guidance, or totally willing to leverage someone else's hard work. :) The guidance I would need is mostly whether it's safe and how much of the clientEndpointInitT's error handling can be copied into streamRecvT; the template's data type differ significantly. Assuming this is shared with server-side receive, that deserves additional attention; I've only looked at the client-side.

Pull request description:
fix(grpc): decode errors in streaming Recv() methods
  This commit fixes issue #3320 where gRPC streaming Recv() methods were not
  decoding errors properly, unlike unary methods which correctly decode custom
  error types.

  Changes:
  - Updated stream_recv.go.tpl template to add error decoding for client streaming
  - Added DecodeError call and type switching for custom errors
  - Added validation for custom errors that have validation rules
  - Fixed code generation issues (indentation and argument passing)

  The fix ensures consistent error handling between unary and streaming gRPC
  methods, allowing clients to properly handle custom service errors defined
  in the DSL for all streaming patterns (server, client, and bidirectional).

  Tests added to verify:
  - Custom errors are properly decoded in streaming recv methods
  - Validation is applied to custom errors
  - All streaming patterns handle errors consistently

  Fixes #3320

Patch:
diff --git a/grpc/codegen/templates/stream_recv.go.tpl b/grpc/codegen/templates/stream_recv.go.tpl
index 93215633..bc2a5534 100644
--- a/grpc/codegen/templates/stream_recv.go.tpl
+++ b/grpc/codegen/templates/stream_recv.go.tpl
@@ -3,7 +3,28 @@ func (s *{{ .VarName }}) {{ .RecvName }}() ({{ .RecvRef }}, error) {
 	var res {{ .RecvRef }}
 	v, err := s.stream.{{ .RecvName }}()
 	if err != nil {
+	{{- if and .Endpoint .Endpoint.Errors (eq .Type "client") }}
+		resp := goagrpc.DecodeError(err)
+		switch message := resp.(type) {
+		{{- range .Endpoint.Errors }}
+			{{- if .Response.ClientConvert }}
+		case {{ .Response.ClientConvert.SrcRef }}:
+			{{- if .Response.ClientConvert.Validation }}
+			if err := {{ .Response.ClientConvert.Validation.Name }}(message); err != nil {
+				return res, err
+			}
+			{{- end }}
+			return res, {{ .Response.ClientConvert.Init.Name }}({{ range .Response.ClientConvert.Init.Args }}{{ .Name }}, {{ end }})
+			{{- end }}
+		{{- end }}
+		case *goapb.ErrorResponse:
+			return res, goagrpc.NewServiceError(message)
+		default:
+			return res, err
+		}
+	{{- else }}
 		return res, err
+	{{- end }}
 	}
 {{- if and .Endpoint.Method.ViewedResult (eq .Type "client") }}
 	proj := {{ .RecvConvert.Init.Name }}({{ range .RecvConvert.Init.Args }}{{ .Name }}, {{ end }})


Test patch:
diff --git a/grpc/codegen/streaming_errors_test.go b/grpc/codegen/streaming_errors_test.go
new file mode 100644
index 00000000..5e4b188d
--- /dev/null
+++ b/grpc/codegen/streaming_errors_test.go
@@ -0,0 +1,191 @@
+package codegen
+
+import (
+	"strings"
+	"testing"
+
+	. "goa.design/goa/v3/dsl"
+	"goa.design/goa/v3/grpc/codegen/testdata"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// TestStreamingWithErrors tests that streaming endpoints properly handle
+// custom errors defined in the service DSL.
+func TestStreamingWithErrors(t *testing.T) {
+	cases := []struct {
+		name     string
+		dsl      func()
+		testFunc func(t *testing.T, code string)
+	}{
+		{
+			name: "server streaming with custom errors",
+			dsl:  testdata.ServerStreamingWithCustomErrorsDSL,
+			testFunc: func(t *testing.T, code string) {
+				// Verify error decoding is present
+				assert.Contains(t, code, "goagrpc.DecodeError(err)",
+					"should decode errors from stream")
+				
+				// Verify custom error types are handled
+				assert.Contains(t, code, "case *streaming_error_servicepb.ServerStreamCustomErrorError:",
+					"should handle custom error type")
+				assert.Contains(t, code, "case *streaming_error_servicepb.ServerStreamValidationErrorError:",
+					"should handle validation error type")
+				
+				// Verify generic errors are handled
+				assert.Contains(t, code, "case *goapb.ErrorResponse:",
+					"should handle generic goa errors")
+				
+				// Verify proper error construction
+				assert.Contains(t, code, "NewServerStreamCustomErrorError(message",
+					"should construct custom error")
+				assert.Contains(t, code, "NewServerStreamValidationErrorError(message",
+					"should construct validation error")
+			},
+		},
+		{
+			name: "bidirectional streaming with errors",
+			dsl:  testdata.BidirectionalStreamingRPCWithErrorsDSL,
+			testFunc: func(t *testing.T, code string) {
+				// Bidirectional streaming with simple errors should still decode
+				assert.Contains(t, code, "goagrpc.DecodeError(err)",
+					"should decode errors from bidirectional stream")
+				assert.Contains(t, code, "case *goapb.ErrorResponse:",
+					"should handle generic errors in bidirectional streaming")
+			},
+		},
+	}
+
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			root := RunGRPCDSL(t, c.dsl)
+			services := CreateGRPCServices(root)
+			clientfs := ClientFiles("", services)
+			require.Greater(t, len(clientfs), 0)
+
+			// Get recv method implementations
+			recvSections := clientfs[0].Section("client-stream-recv")
+			require.Greater(t, len(recvSections), 0)
+
+			// Build complete recv method code
+			var codeBuilder strings.Builder
+			for _, section := range recvSections {
+				require.NoError(t, section.Write(&codeBuilder))
+			}
+			code := codeBuilder.String()
+
+			// Run test-specific assertions
+			c.testFunc(t, code)
+		})
+	}
+}
+
+// TestStreamingErrorsWithValidation verifies that custom errors with
+// validation rules are properly validated in streaming recv methods.
+func TestStreamingErrorsWithValidation(t *testing.T) {
+	root := RunGRPCDSL(t, testdata.ServerStreamingWithCustomErrorsDSL)
+	services := CreateGRPCServices(root)
+	
+	// Verify the DSL has errors with validation
+	require.Len(t, root.Services, 1)
+	svc := root.Services[0]
+	require.Len(t, svc.Methods, 1)
+	method := svc.Methods[0]
+	require.Greater(t, len(method.Errors), 0, "method should have errors defined")
+
+	// Generate client code
+	clientfs := ClientFiles("", services)
+	require.Greater(t, len(clientfs), 0)
+
+	// Check recv implementations
+	recvSections := clientfs[0].Section("client-stream-recv")
+	var code strings.Builder
+	for _, section := range recvSections {
+		require.NoError(t, section.Write(&code))
+	}
+	recvCode := code.String()
+
+	// For errors with validation, verify validation is called
+	if strings.Contains(recvCode, "ValidateServerStreamCustomErrorError") {
+		assert.Contains(t, recvCode, "if err := ValidateServerStreamCustomErrorError(message); err != nil {",
+			"should validate custom error before returning")
+	}
+}
+
+// TestStreamingErrorComparison compares error handling between unary and
+// streaming methods to ensure consistency.
+func TestStreamingErrorComparison(t *testing.T) {
+	// DSL with both unary and streaming methods with errors
+	dsl := func() {
+		var CustomError = Type("CustomError", func() {
+			ErrorName("name", String, "error name")
+			Attribute("message", String, "error message")
+			Required("name", "message")
+		})
+
+		Service("MixedService", func() {
+			// Unary method with custom error
+			Method("UnaryMethod", func() {
+				Payload(String)
+				Result(String)
+				Error("custom_error", CustomError)
+				GRPC(func() {
+					Response("custom_error", CodeInvalidArgument)
+				})
+			})
+
+			// Streaming method with same error
+			Method("StreamingMethod", func() {
+				Payload(String)
+				StreamingResult(String)
+				Error("custom_error", CustomError)
+				GRPC(func() {
+					Response("custom_error", CodeInvalidArgument)
+				})
+			})
+		})
+	}
+
+	root := RunGRPCDSL(t, dsl)
+	services := CreateGRPCServices(root)
+	clientfs := ClientFiles("", services)
+	require.Greater(t, len(clientfs), 0, "should have client files")
+
+	// Find unary and streaming code in different sections
+	var unaryCode, streamCode string
+	
+	// For unary, look in client-endpoint-init
+	if sections := clientfs[0].Section("client-endpoint-init"); len(sections) > 0 {
+		var code strings.Builder
+		for _, section := range sections {
+			require.NoError(t, section.Write(&code))
+		}
+		unaryCode = code.String()
+	}
+	
+	// For streaming, look in client-stream-recv
+	if sections := clientfs[0].Section("client-stream-recv"); len(sections) > 0 {
+		var code strings.Builder
+		for _, section := range sections {
+			require.NoError(t, section.Write(&code))
+		}
+		streamCode = code.String()
+	}
+
+	// If no sections found, skip test with explanation
+	if unaryCode == "" || streamCode == "" {
+		t.Skip("Cannot compare unary and streaming - sections not found in generated code")
+	}
+
+	// Both should decode errors
+	assert.Contains(t, unaryCode, "goagrpc.DecodeError(err)",
+		"unary methods should decode errors")
+	assert.Contains(t, streamCode, "goagrpc.DecodeError(err)",
+		"streaming methods should decode errors")
+
+	// Both should handle the custom error type
+	assert.Contains(t, unaryCode, "case *mixed_servicepb.",
+		"unary should handle custom error types")
+	assert.Contains(t, streamCode, "case *mixed_servicepb.",
+		"streaming should handle custom error types")
+}
\ No newline at end of file
diff --git a/grpc/codegen/testdata/dsls.go b/grpc/codegen/testdata/dsls.go
index 8bdd24b8..df7f3d7c 100644
--- a/grpc/codegen/testdata/dsls.go
+++ b/grpc/codegen/testdata/dsls.go
@@ -321,6 +321,40 @@ var BidirectionalStreamingRPCWithErrorsDSL = func() {
 	})
 }
 
+var ServerStreamingWithCustomErrorsDSL = func() {
+	// Custom error types for testing error handling in streaming
+	var CustomError = Type("CustomError", func() {
+		ErrorName("name", String, "error name")
+		Attribute("message", String, "error message")
+		Attribute("code", Int, "error code")
+		Required("name", "message", "code")
+	})
+
+	var ValidationError = Type("ValidationError", func() {
+		ErrorName("name", String, "error name")
+		Attribute("field", String, "field that failed validation")
+		Attribute("reason", String, "validation failure reason")
+		Required("name", "field", "reason")
+	})
+
+	Service("StreamingErrorService", func() {
+		Method("ServerStream", func() {
+			Payload(String)
+			StreamingResult(String)
+			Error("custom_error", CustomError, "Custom application error")
+			Error("validation_error", ValidationError, "Validation error")
+			Error("internal_error", func() {
+				Description("Internal server error")
+			})
+			GRPC(func() {
+				Response("custom_error", CodeInvalidArgument)
+				Response("validation_error", CodeFailedPrecondition)
+				Response("internal_error", CodeInternal)
+			})
+		})
+	})
+}
+
 var BidirectionalStreamingRPCSameTypeDSL = func() {
 	var T = Type("UserType", func() {
 		Field(1, "a", Int)