How to intercept an error from an interceptor?

[lalloni]

We are trying to write an interceptor to detect errors that are not *connect.Error and convert them to that after doing some sanitization on them.

The above works well for unary methods, but we can't find a way to intercept errors in the streaming methods case via a connect.StreamingHandlerConn wrapper.

connect.StreamingHandlerConn is defined as:

type StreamingHandlerConn interface {
	Spec() Spec
	Peer() Peer

	Receive(any) error
	RequestHeader() http.Header

	Send(any) error
	ResponseHeader() http.Header
	ResponseTrailer() http.Header
}

And there are no apparent method/s to override to intercept errors (as returned from handlers) and transforming them to other errors forms (e.g. a *connect.Error)

Any clue would be very appreciated.

As context information, this is what we are using for unary methods:

func (e errorInterceptor) WrapUnary(next connect.UnaryFunc) connect.UnaryFunc {
	return func(ctx context.Context, request connect.AnyRequest) (connect.AnyResponse, error) {
		r, err := next(ctx, request)
		if err == nil {
			return r, nil
		}
		if errors.Is(err, &connect.Error{}) {
			return nil, err
		}
		return nil, sanitize(err)
	}
}

func sanitize(err error) *connect.Error {
  // log details and return sanitized error
}

[akshayjshah]

You'd implement StreamingHandlerConn, embedding the original connection and overriding Send and Receive. Here's a very similar example from inside the Connect runtime: https://github.com/connectrpc/connect-go/blob/main/protocol.go#L165.

[lalloni]

That is the approach we tried but unfortunately there seems to be no obvious way of capturing errors using the StreamingHandlerConn interface.

Overriding its Send(any) error would only allow us to capture the errors produced while trying to write/send to the client-side "wire", but what we need is to hook into the handler-side produced error to manipulate it.

I can see the in the internals of the Connect runtime the above is accomplished via the handlerConnCloser interface and its Close(error) error method, but apparently that is not exposed as public API.

We tried adding a Close(error) error method to our StreamingHandlerConn impl in hope that it would be picked up by the runtime as an implementor of handlerConnCloser and called but that didn't work.

I wonder what else we'd need to do be able to capture and mutate errors returned by handlers.

Any clues are welcome!

[jhump]

@lalloni, I don't quite follow why you can't use the same approach as for the unary case. I think it would be as simple as this:

func (e errorInterceptor) WrapStreamingHandler(next connect.StreamingHandlerFunc) connect.StreamingHandlerFunc {
	return func(ctx context.Context, conn connect.StreamingHandlerConn) error {
		err := next(ctx, conn)
		if err == nil {
			return nil
		}
		if errors.Is(err, &connect.Error{}) {
			return err
		}
		return sanitize(err)
	}
}

If that is not sufficient, can you elaborate on why not?

Wrapping the stream connection is only necessary from the client side, as that is the way that the client will observe the errors, in particular from the Receive method.

[lalloni]

Hey @jhump!

You are right, that works as we need.

Thanks a lot.