[v2024.3-2025.1]: feature: headers in case of worker error

proxy/proxy.go

@@ -17,6 +17,7 @@ import (
 	"github.com/roadrunner-server/pool/pool/static_pool"
 	"github.com/roadrunner-server/pool/worker"
 	"go.opentelemetry.io/otel/propagation"
+	"go.uber.org/zap"
 	"golang.org/x/net/context"
 	spb "google.golang.org/genproto/googleapis/rpc/status"
 	"google.golang.org/grpc"
@@ -33,6 +34,7 @@ const (
 	peerAuthType string = ":peer.auth-type"
 	delimiter    string = "|:|"
 	apiErr       string = "error"
+	headers      string = "headers"
 )
 
 type Pool interface {
@@ -65,6 +67,7 @@ type rpcContext struct {
 // Proxy manages GRPC/RoadRunner bridge.
 type Proxy struct {
 	mu       *sync.RWMutex
+	log      *zap.Logger
 	prop     propagation.TextMapPropagator
 	grpcPool Pool
 	name     string
@@ -75,8 +78,9 @@ type Proxy struct {
 }
 
 // NewProxy creates a new service proxy object.
-func NewProxy(name string, metadata string, grpcPool Pool, mu *sync.RWMutex, prop propagation.TextMapPropagator) *Proxy {
+func NewProxy(name string, metadata string, log *zap.Logger, grpcPool Pool, mu *sync.RWMutex, prop propagation.TextMapPropagator) *Proxy {
 	return &Proxy{
+		log:      log,
 		mu:       mu,
 		prop:     prop,
 		grpcPool: grpcPool,
@@ -190,8 +194,10 @@ func (p *Proxy) invoke(ctx context.Context, method string, in *codec.RawMessage)
 
 	md, err := p.responseMetadata(r)
 	if err != nil {
+		_ = grpc.SetHeader(ctx, md)
 		return nil, err
 	}
+
 	ctx = metadata.NewIncomingContext(ctx, md)
 	err = grpc.SetHeader(ctx, md)
 	if err != nil {
@@ -215,29 +221,54 @@ func (p *Proxy) responseMetadata(resp *payload.Payload) (metadata.MD, error) {
 	}
 
 	if len(rpcMetadata) > 0 {
+		// new meta should be used
+		mdn := metadata.New(map[string]string{})
+		// old meta should not be used in response
 		md = metadata.New(rpcMetadata)
 
+		if len(md.Get(headers)) > 0 {
+			mdh := make(map[string]any)
+			err = json.Unmarshal([]byte(md.Get(headers)[0]), &mdh)
+			if err != nil {
+				// we don't need to return this error, log it
+				p.log.Error("error unmarshalling headers", zap.Error(err))
+				goto api
+			}
+
+			for k, v := range mdh {
+				switch tt := v.(type) {
+				case string:
+					mdn.Append(k, tt)
+				case int:
+					mdn.Append(k, strconv.Itoa(tt))
+				default:
+					p.log.Warn("skipping header with unsupported type", zap.String("key", k), zap.Any("value", v))
+				}
+			}
+		}
+
 		/*
 			we have an error
-			actually, if code is OK, status.ErrorProto will be nil
+			actually if the code is OK, status.ErrorProto will be nil
 			but, we use this only in case of PHP exception happened
 
 		*/
+	api:
 		if len(md.Get(apiErr)) > 0 {
 			st := &spb.Status{}
 
 			// get an error
 			data, err := base64.StdEncoding.DecodeString(md.Get(apiErr)[0])
 			if err != nil {
-				return nil, err
+				return mdn, err
 			}
 
 			err = proto.Unmarshal(data, st)
 			if err != nil {
-				return nil, err
+				return mdn, err
 			}
 
-			return md, status.ErrorProto(st)
+			return mdn, status.ErrorProto(st)
 		}
 	}
 

server.go

@@ -62,7 +62,7 @@ func (p *Plugin) createGRPCserver(interceptors map[string]common.Interceptor) (*
 		}
 
 		for _, service := range services {
-			px := proxy.NewProxy(fmt.Sprintf("%s.%s", service.Package, service.Name), p.config.Proto[i], p.gPool, p.mu, p.prop)
+			px := proxy.NewProxy(fmt.Sprintf("%s.%s", service.Package, service.Name), p.config.Proto[i], p.log.Named(service.Name), p.gPool, p.mu, p.prop)
 			for _, m := range service.Methods {
 				px.RegisterMethod(m.Name)
 			}