Add support for passing per-message metadata.MD

async.go

@@ -46,7 +46,7 @@ type asyncCallState struct {
 // This function should only be used by generated code.
 func (c RawConfiguration) AsyncCall(ctx context.Context, d QuorumCallData) *Async {
 	expectedReplies := len(c)
-	md := ordering.Metadata_builder{MessageID: c.getMsgID(), Method: d.Method}.Build()
+	md := ordering.NewGorumsMetadata(ctx, c.getMsgID(), d.Method)
 	replyChan := make(chan response, expectedReplies)
 
 	for _, n := range c {

channel_test.go

@@ -44,16 +44,17 @@ func TestChannelCreation(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	//the node should be closed manually because it isn't added to the configuration
+	// the node should be closed manually because it isn't added to the configuration
 	defer node.close()
 	mgr := dummyMgr()
 	// a proper connection should NOT be established here
 	node.connect(mgr)
 
 	replyChan := make(chan response, 1)
 	go func() {
-		md := ordering.Metadata_builder{MessageID: 1, Method: handlerName}.Build()
-		req := request{ctx: context.Background(), msg: &Message{Metadata: md, Message: &mock.Request{}}}
+		ctx := context.Background()
+		md := ordering.NewGorumsMetadata(ctx, 1, handlerName)
+		req := request{ctx: ctx, msg: &Message{Metadata: md, Message: &mock.Request{}}}
 		node.channel.enqueue(req, replyChan, false)
 	}()
 	select {
@@ -118,7 +119,7 @@ func TestChannelReconnection(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	//the node should be closed manually because it isn't added to the configuration
+	// the node should be closed manually because it isn't added to the configuration
 	defer node.close()
 	mgr := dummyMgr()
 	// a proper connection should NOT be established here because server is not started
@@ -127,8 +128,9 @@ func TestChannelReconnection(t *testing.T) {
 	// send first message when server is down
 	replyChan1 := make(chan response, 1)
 	go func() {
-		md := ordering.Metadata_builder{MessageID: 1, Method: handlerName}.Build()
-		req := request{ctx: context.Background(), msg: &Message{Metadata: md, Message: &mock.Request{}}}
+		ctx := context.Background()
+		md := ordering.NewGorumsMetadata(ctx, 1, handlerName)
+		req := request{ctx: ctx, msg: &Message{Metadata: md, Message: &mock.Request{}}}
 		node.channel.enqueue(req, replyChan1, false)
 	}()
 
@@ -147,8 +149,9 @@ func TestChannelReconnection(t *testing.T) {
 	// send second message when server is up
 	replyChan2 := make(chan response, 1)
 	go func() {
-		md := ordering.Metadata_builder{MessageID: 2, Method: handlerName}.Build()
-		req := request{ctx: context.Background(), msg: &Message{Metadata: md, Message: &mock.Request{}}, opts: getCallOptions(E_Multicast, nil)}
+		ctx := context.Background()
+		md := ordering.NewGorumsMetadata(ctx, 2, handlerName)
+		req := request{ctx: ctx, msg: &Message{Metadata: md, Message: &mock.Request{}}, opts: getCallOptions(E_Multicast, nil)}
 		node.channel.enqueue(req, replyChan2, false)
 	}()
 
@@ -167,8 +170,9 @@ func TestChannelReconnection(t *testing.T) {
 	// send third message when server has been previously up but is now down
 	replyChan3 := make(chan response, 1)
 	go func() {
-		md := ordering.Metadata_builder{MessageID: 3, Method: handlerName}.Build()
-		req := request{ctx: context.Background(), msg: &Message{Metadata: md, Message: &mock.Request{}}}
+		ctx := context.Background()
+		md := ordering.NewGorumsMetadata(ctx, 3, handlerName)
+		req := request{ctx: ctx, msg: &Message{Metadata: md, Message: &mock.Request{}}}
 		node.channel.enqueue(req, replyChan3, false)
 	}()
 

correctable.go

@@ -102,7 +102,7 @@ type correctableCallState struct {
 // This method should only be used by generated code.
 func (c RawConfiguration) CorrectableCall(ctx context.Context, d CorrectableCallData) *Correctable {
 	expectedReplies := len(c)
-	md := ordering.Metadata_builder{MessageID: c.getMsgID(), Method: d.Method}.Build()
+	md := ordering.NewGorumsMetadata(ctx, c.getMsgID(), d.Method)
 
 	replyChan := make(chan response, expectedReplies)
 	for _, n := range c {

multicast.go

@@ -10,9 +10,11 @@ import (
 // By default this function returns once the message has been sent to all nodes.
 // Providing the call option WithNoSendWaiting, the function may return
 // before the message has been sent.
+//
+// This method should be used by generated code only.
 func (c RawConfiguration) Multicast(ctx context.Context, d QuorumCallData, opts ...CallOption) {
 	o := getCallOptions(E_Multicast, opts)
-	md := ordering.Metadata_builder{MessageID: c.getMsgID(), Method: d.Method}.Build()
+	md := ordering.NewGorumsMetadata(ctx, c.getMsgID(), d.Method)
 	sentMsgs := 0
 
 	var replyChan chan response

ordering/gorums_metadata.go

@@ -0,0 +1,40 @@
+package ordering
+
+import (
+	"context"
+
+	"google.golang.org/grpc/metadata"
+)
+
+// NewGorumsMetadata creates a new Gorums metadata object for the given method
+// and client message ID. It also appends any client-specific metadata from the
+// context to the Gorums metadata object.
+//
+// This is used to pass client-specific metadata to the server via Gorums.
+// This method should be used by generated code only.
+func NewGorumsMetadata(ctx context.Context, msgID uint64, method string) *Metadata {
+	gorumsMetadata := Metadata_builder{MessageID: msgID, Method: method}
+	md, _ := metadata.FromOutgoingContext(ctx)
+	for k, vv := range md {
+		for _, v := range vv {
+			entry := MetadataEntry_builder{Key: k, Value: v}.Build()
+			gorumsMetadata.Entry = append(gorumsMetadata.Entry, entry)
+		}
+	}
+	return gorumsMetadata.Build()
+}
+
+// AppendToIncomingContext appends client-specific metadata from the
+// Gorums metadata object to the incoming context.
+//
+// This is used to pass client-specific metadata from the Gorums runtime
+// to the server implementation.
+// This method should be used by generated code only.
+func (x *Metadata) AppendToIncomingContext(ctx context.Context) context.Context {
+	existingMD, _ := metadata.FromIncomingContext(ctx)
+	newMD := existingMD.Copy() // copy to avoid mutating the original
+	for _, entry := range x.GetEntry() {
+		newMD.Append(entry.GetKey(), entry.GetValue())
+	}
+	return metadata.NewIncomingContext(ctx, newMD)
+}

ordering/ordering.proto

@@ -20,4 +20,11 @@ message Metadata {
   uint64 MessageID = 1;
   string Method = 2;
   google.rpc.Status Status = 3;
+  // per message client-generated metadata
+  repeated MetadataEntry Entry = 4;
+}
+
+message MetadataEntry {
+  string Key = 1;
+  string Value = 2;
 }

quorumcall.go

@@ -24,7 +24,7 @@ type QuorumCallData struct {
 // This method should be used by generated code only.
 func (c RawConfiguration) QuorumCall(ctx context.Context, d QuorumCallData) (resp protoreflect.ProtoMessage, err error) {
 	expectedReplies := len(c)
-	md := ordering.Metadata_builder{MessageID: c.getMsgID(), Method: d.Method}.Build()
+	md := ordering.NewGorumsMetadata(ctx, c.getMsgID(), d.Method)
 
 	replyChan := make(chan response, expectedReplies)
 	for _, n := range c {

rpc.go

@@ -19,7 +19,7 @@ type CallData struct {
 //
 // This method should be used by generated code only.
 func (n *RawNode) RPCCall(ctx context.Context, d CallData) (protoreflect.ProtoMessage, error) {
-	md := ordering.Metadata_builder{MessageID: n.mgr.getMsgID(), Method: d.Method}.Build()
+	md := ordering.NewGorumsMetadata(ctx, n.mgr.getMsgID(), d.Method)
 	replyChan := make(chan response, 1)
 	n.channel.enqueue(request{ctx: ctx, msg: &Message{Metadata: md, Message: d.Message}}, replyChan, false)
 

server.go

@@ -95,7 +95,7 @@ func (s *orderingServer) NodeStream(srv ordering.Gorums_NodeStreamServer) error
 			// We start the handler in a new goroutine in order to allow multiple handlers to run concurrently.
 			// However, to preserve request ordering, the handler must unlock the shared mutex when it has either
 			// finished, or when it is safe to start processing the next request.
-			go handler(ServerCtx{Context: ctx, once: new(sync.Once), mut: &mut}, req, finished)
+			go handler(ServerCtx{Context: req.Metadata.AppendToIncomingContext(ctx), once: new(sync.Once), mut: &mut}, req, finished)
 			// Wait until the handler releases the mutex.
 			mut.Lock()
 		}
@@ -142,9 +142,7 @@ type Server struct {
 	grpcServer *grpc.Server
 }
 
-// NewServer returns a new instance of GorumsServer.
-// This function is intended for internal Gorums use.
-// You should call `NewServer` in the generated code instead.
+// NewServer returns a new instance of [gorums.Server].
 func NewServer(opts ...ServerOption) *Server {
 	var serverOpts serverOptions
 	for _, opt := range opts {