diff options
author | Paul Okstad <pokstad@gitlab.com> | 2020-07-03 18:55:05 +0300 |
---|---|---|
committer | Paul Okstad <pokstad@gitlab.com> | 2020-07-03 18:55:05 +0300 |
commit | 4411f8ef16dc6a77c705e048e10dfc924ad48a34 (patch) | |
tree | 871d1734201384d836d37057d3a5b22d3e674faa /internal/praefect/grpc-proxy | |
parent | de246fc4268b1db216a348f01f37f54907b21635 (diff) |
Error forwarded mutator RPCs on replication job enqueue failure
Diffstat (limited to 'internal/praefect/grpc-proxy')
-rw-r--r-- | internal/praefect/grpc-proxy/proxy/director.go | 9 | ||||
-rw-r--r-- | internal/praefect/grpc-proxy/proxy/handler.go | 9 |
2 files changed, 12 insertions, 6 deletions
diff --git a/internal/praefect/grpc-proxy/proxy/director.go b/internal/praefect/grpc-proxy/proxy/director.go index 9c613dd11..386d4ad44 100644 --- a/internal/praefect/grpc-proxy/proxy/director.go +++ b/internal/praefect/grpc-proxy/proxy/director.go @@ -28,7 +28,7 @@ type StreamDirector func(ctx context.Context, fullMethodName string, peeker Stre // proxy handler type StreamParameters struct { primary Destination - reqFinalizer func() + reqFinalizer func() error callOptions []grpc.CallOption secondaries []Destination } @@ -41,7 +41,7 @@ type Destination struct { } // NewStreamParameters returns a new instance of StreamParameters -func NewStreamParameters(primary Destination, secondaries []Destination, reqFinalizer func(), callOpts []grpc.CallOption) *StreamParameters { +func NewStreamParameters(primary Destination, secondaries []Destination, reqFinalizer func() error, callOpts []grpc.CallOption) *StreamParameters { return &StreamParameters{ primary: primary, secondaries: secondaries, @@ -59,10 +59,11 @@ func (s *StreamParameters) Secondaries() []Destination { } // RequestFinalizer calls the request finalizer -func (s *StreamParameters) RequestFinalizer() { +func (s *StreamParameters) RequestFinalizer() error { if s.reqFinalizer != nil { - s.reqFinalizer() + return s.reqFinalizer() } + return nil } // CallOptions returns call options diff --git a/internal/praefect/grpc-proxy/proxy/handler.go b/internal/praefect/grpc-proxy/proxy/handler.go index cc596b537..73730b2c7 100644 --- a/internal/praefect/grpc-proxy/proxy/handler.go +++ b/internal/praefect/grpc-proxy/proxy/handler.go @@ -99,7 +99,7 @@ type streamAndMsg struct { // handler is where the real magic of proxying happens. // It is invoked like any gRPC server stream and uses the gRPC server framing to get and receive bytes from the wire, // forwarding it to a ClientStream established against the relevant ClientConn. -func (s *handler) handler(srv interface{}, serverStream grpc.ServerStream) error { +func (s *handler) handler(srv interface{}, serverStream grpc.ServerStream) (finalErr error) { // little bit of gRPC internals never hurt anyone fullMethodName, ok := grpc.MethodFromServerStream(serverStream) if !ok { @@ -114,7 +114,12 @@ func (s *handler) handler(srv interface{}, serverStream grpc.ServerStream) error return err } - defer params.RequestFinalizer() + defer func() { + err := params.RequestFinalizer() + if finalErr == nil { + finalErr = err + } + }() clientCtx, clientCancel := context.WithCancel(params.Primary().Ctx) defer clientCancel() |