Error forwarded mutator RPCs on replication job enqueue failure

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()