gorums: harden message unmarshaling and address stream review findings

doc/dev-guide.md

@@ -151,7 +151,7 @@ A `Transport` is everything a call needs to reach one node.
 It holds a `channelRef`, an atomically replaceable reference to the node's current `Channel`.
 There are three channel kinds:
 
-* `OutboundChannel` runs over streams this node dials; it opens a new `session` for each stream and requeues pending calls when a stream is lost.
+* `OutboundChannel` runs over streams this node dials; it opens a new `session` for each stream and requeues pending calls when a stream is lost, as far as the send queue has space.
 * `InboundChannel` runs over one stream that the server accepted; its single `session` ends with the stream.
 * `LocalChannel` serves requests in-process through the handler, with no stream.
 

internal/conn/inbound_manager.go

@@ -108,9 +108,9 @@ const ClientIDStart = 1 << 20
 // installed on the self-node (if present) to enable in-process dispatch without
 // a network round-trip. Inbound channels use sendBuffer and dispatchSize as
 // their send queue and request dispatch capacities; a dispatchSize of 0
-// selects the default.
-// Panics on configuration errors (invalid addresses, duplicate nodes, etc.)
-func NewInboundManager(myID uint32, peerNodes NodeSource, sendBuffer, dispatchSize uint, onConfigChange func(Config), handler stream.RequestHandler) *InboundManager {
+// selects the default. It returns an error if peerNodes is invalid, for
+// example because of an invalid address or a duplicate node.
+func NewInboundManager(myID uint32, peerNodes NodeSource, sendBuffer, dispatchSize uint, onConfigChange func(Config), handler stream.RequestHandler) (*InboundManager, error) {
 	im := &InboundManager{
 		myID:           myID,
 		knownNodes:     make(map[uint32]*Node),
@@ -125,11 +125,11 @@ func NewInboundManager(myID uint32, peerNodes NodeSource, sendBuffer, dispatchSi
 	}
 	if peerNodes != nil {
 		if _, err := peerNodes.newConfig(im); err != nil {
-			panic("gorums: invalid peer configuration: " + err.Error())
+			return nil, err
 		}
 	}
 	im.rebuildConfig()
-	return im
+	return im, nil
 }
 
 // Nodes returns a slice of known peer nodes in order of their IDs.

internal/conn/inbound_manager_test.go

@@ -74,20 +74,15 @@ func (f requestHandlerFunc) HandleRequest(ctx context.Context, msg *stream.Messa
 	f(ctx, msg, release, send)
 }
 
-// shouldPanic asserts that fn panics with a message containing wantSubstr.
-func shouldPanic(t *testing.T, wantSubstr string, fn func()) {
+// newInboundManager returns an InboundManager for a test, failing the test if
+// [NewInboundManager] rejects peerNodes.
+func newInboundManager(t *testing.T, myID uint32, peerNodes NodeSource, onConfigChange func(Config), handler stream.RequestHandler) *InboundManager {
 	t.Helper()
-	defer func() {
-		r := recover()
-		if r == nil {
-			t.Fatalf("expected panic containing %q; got no panic", wantSubstr)
-		}
-		msg := fmt.Sprint(r)
-		if !strings.Contains(msg, wantSubstr) {
-			t.Fatalf("panic = %q; want it to contain %q", msg, wantSubstr)
-		}
-	}()
-	fn()
+	im, err := NewInboundManager(myID, peerNodes, 0, 0, onConfigChange, handler)
+	if err != nil {
+		t.Fatalf("NewInboundManager: %v", err)
+	}
+	return im
 }
 
 // testNode is a minimal NodeSource for use in tests.
@@ -103,11 +98,11 @@ var _ NodeSource = nodeMap[testNode](nil)
 // newTestInboundManager creates an InboundManager with myID and three known peers.
 func newTestInboundManager(t *testing.T, myID uint32) *InboundManager {
 	t.Helper()
-	im := NewInboundManager(myID, WithNodes(map[uint32]testNode{
+	im := newInboundManager(t, myID, WithNodes(map[uint32]testNode{
 		1: {"127.0.0.1:9081"},
 		2: {"127.0.0.1:9082"},
 		3: {"127.0.0.1:9083"},
-	}), 0, 0, nil, nil)
+	}), nil, nil)
 	return im
 }
 
@@ -117,7 +112,7 @@ func TestNewInboundManager(t *testing.T) {
 		opt        NodeSource
 		wantIDs    []uint32
 		wantCfgIDs []uint32 // expected Config IDs after construction
-		wantPanic  string   // if non-empty, expect panic containing this substring
+		wantErr    string   // if non-empty, expect an error containing this substring
 	}{
 		{
 			name: "ValidNodes",
@@ -130,32 +125,32 @@ func TestNewInboundManager(t *testing.T) {
 			wantCfgIDs: []uint32{1}, // only self-node until peers connect
 		},
 		{
-			name:      "EmptyMapRejected",
-			opt:       WithNodes(map[uint32]testNode{}),
-			wantPanic: "missing required node map",
+			name:    "EmptyMapRejected",
+			opt:     WithNodes(map[uint32]testNode{}),
+			wantErr: "missing required node map",
 		},
 		{
 			name: "NodeZeroRejected",
 			opt: WithNodes(map[uint32]testNode{
 				0: {"127.0.0.1:9080"},
 				1: {"127.0.0.1:9081"},
 			}),
-			wantPanic: "node 0 is reserved",
+			wantErr: "node 0 is reserved",
 		},
 		{
 			name: "DuplicateAddressRejected",
 			opt: WithNodes(map[uint32]testNode{
 				1: {"127.0.0.1:9081"},
 				2: {"127.0.0.1:9081"}, // same address as ID 1
 			}),
-			wantPanic: "already in use by node",
+			wantErr: "already in use by node",
 		},
 		{
 			name: "InvalidAddressRejected",
 			opt: WithNodes(map[uint32]testNode{
 				1: {"not-an-address"},
 			}),
-			wantPanic: "invalid address",
+			wantErr: "invalid address",
 		},
 		{
 			// WithNodeList assigns IDs starting at 1.
@@ -165,20 +160,23 @@ func TestNewInboundManager(t *testing.T) {
 			wantCfgIDs: []uint32{1}, // only self-node until peers connect
 		},
 		{
-			name:      "WithNodeListEmptyRejected",
-			opt:       WithNodeList([]string{}),
-			wantPanic: "missing required node addresses",
+			name:    "WithNodeListEmptyRejected",
+			opt:     WithNodeList([]string{}),
+			wantErr: "missing required node addresses",
 		},
 	}
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
-			if tc.wantPanic != "" {
-				shouldPanic(t, tc.wantPanic, func() {
-					NewInboundManager(1, tc.opt, 0, 0, nil, nil)
-				})
+			im, err := NewInboundManager(1, tc.opt, 0, 0, nil, nil)
+			if tc.wantErr != "" {
+				if err == nil || !strings.Contains(err.Error(), tc.wantErr) {
+					t.Fatalf("NewInboundManager() error = %v; want it to contain %q", err, tc.wantErr)
+				}
 				return
 			}
-			im := NewInboundManager(1, tc.opt, 0, 0, nil, nil)
+			if err != nil {
+				t.Fatalf("NewInboundManager() error = %v", err)
+			}
 			nodes := im.Nodes()
 			if len(nodes) != len(tc.wantIDs) {
 				t.Fatalf("len(im.Nodes()) = %d; want %d", len(nodes), len(tc.wantIDs))
@@ -196,19 +194,19 @@ func TestNewInboundManager(t *testing.T) {
 }
 
 func TestInboundManagerKeepsHighKnownPeerIDs(t *testing.T) {
-	im := NewInboundManager(ClientIDStart, WithNodes(map[uint32]testNode{
+	im := newInboundManager(t, ClientIDStart, WithNodes(map[uint32]testNode{
 		ClientIDStart: {"127.0.0.1:9081"},
-	}), 0, 0, nil, nil)
+	}), nil, nil)
 
 	checkIDs(t, im.ConnectedPeers(), []uint32{ClientIDStart}, "known peers")
 	checkIDs(t, im.ConnectedClients(), []uint32{}, "dynamic clients")
 }
 
 func TestInboundManagerDynamicClientIDSkipsKnownPeer(t *testing.T) {
-	im := NewInboundManager(1, WithNodes(map[uint32]testNode{
+	im := newInboundManager(t, 1, WithNodes(map[uint32]testNode{
 		1:             {"127.0.0.1:9081"},
 		ClientIDStart: {"127.0.0.1:9082"},
-	}), 0, 0, nil, nil)
+	}), nil, nil)
 	clientStream := newMockBidiStream()
 	t.Cleanup(clientStream.close)
 
@@ -607,10 +605,13 @@ func TestAcceptPeerReplyRidesReceivingStream(t *testing.T) {
 		defer release()
 		send(msg)
 	})
-	im := NewInboundManager(1, WithNodes(map[uint32]testNode{
+	im, err := NewInboundManager(1, WithNodes(map[uint32]testNode{
 		1: {"127.0.0.1:9081"},
 		2: {"127.0.0.1:9082"},
 	}), 4, 0, nil, echo)
+	if err != nil {
+		t.Fatalf("NewInboundManager: %v", err)
+	}
 
 	survivor := newRecordingBidiStream()
 	t.Cleanup(survivor.close)
@@ -655,11 +656,11 @@ func TestAcceptPeerReplyRidesReceivingStream(t *testing.T) {
 // self-node present in the initial configuration.
 func TestOnConfigChangeCallbackFiringOnConstruction(t *testing.T) {
 	var calls [][]uint32
-	NewInboundManager(1, WithNodes(map[uint32]testNode{
+	newInboundManager(t, 1, WithNodes(map[uint32]testNode{
 		1: {"127.0.0.1:9081"},
 		2: {"127.0.0.1:9082"},
 		3: {"127.0.0.1:9083"},
-	}), 0, 0, func(cfg Config) {
+	}), func(cfg Config) {
 		calls = append(calls, slices.Clone(cfg.NodeIDs()))
 	}, nil)
 
@@ -676,11 +677,11 @@ func TestOnConfigChangeCallbackFiringOnConstruction(t *testing.T) {
 // later disconnects.
 func TestOnConfigChangeCallbackPeerConnectDisconnect(t *testing.T) {
 	var snapshots [][]uint32
-	im := NewInboundManager(1, WithNodes(map[uint32]testNode{
+	im := newInboundManager(t, 1, WithNodes(map[uint32]testNode{
 		1: {"127.0.0.1:9081"},
 		2: {"127.0.0.1:9082"},
 		3: {"127.0.0.1:9083"},
-	}), 0, 0, func(cfg Config) {
+	}), func(cfg Config) {
 		snapshots = append(snapshots, slices.Clone(cfg.NodeIDs()))
 	}, nil)
 
@@ -711,11 +712,11 @@ func TestOnConfigChangeCallbackPeerConnectDisconnect(t *testing.T) {
 // fires in sorted ID order as multiple peers connect and disconnect.
 func TestOnConfigChangeCallbackMultiplePeers(t *testing.T) {
 	var snapshots [][]uint32
-	im := NewInboundManager(1, WithNodes(map[uint32]testNode{
+	im := newInboundManager(t, 1, WithNodes(map[uint32]testNode{
 		1: {"127.0.0.1:9081"},
 		2: {"127.0.0.1:9082"},
 		3: {"127.0.0.1:9083"},
-	}), 0, 0, func(cfg Config) {
+	}), func(cfg Config) {
 		snapshots = append(snapshots, slices.Clone(cfg.NodeIDs()))
 	}, nil)
 
@@ -753,10 +754,10 @@ func TestOnConfigChangeCallbackMultiplePeers(t *testing.T) {
 // function twice does not fire the callback a second time on the same disconnect.
 func TestOnConfigChangeCallbackIdempotentCleanup(t *testing.T) {
 	var callCount int
-	im := NewInboundManager(1, WithNodes(map[uint32]testNode{
+	im := newInboundManager(t, 1, WithNodes(map[uint32]testNode{
 		1: {"127.0.0.1:9081"},
 		2: {"127.0.0.1:9082"},
-	}), 0, 0, func(_ Config) {
+	}), func(_ Config) {
 		callCount++
 	}, nil)
 

internal/impl/marshaling.go

@@ -9,29 +9,42 @@ import (
 	"google.golang.org/protobuf/reflect/protoregistry"
 )
 
+// UnmarshalRequest unmarshals the request proto message from the message.
+// It uses the method name in the message to look up the Input type from the proto registry.
+func UnmarshalRequest(in *stream.Message) (proto.Message, error) {
+	return unmarshal(in, protoreflect.MethodDescriptor.Input, "request")
+}
+
 // unmarshalResponse unmarshals the response proto message from the message.
 // It uses the method name in the message to look up the Output type from the proto registry.
 func unmarshalResponse(out *stream.Message) (proto.Message, error) {
-	// get method descriptor from registry
-	desc, err := protoregistry.GlobalFiles.FindDescriptorByName(protoreflect.FullName(out.GetMethod()))
+	return unmarshal(out, protoreflect.MethodDescriptor.Output, "response")
+}
+
+// unmarshal unmarshals the payload of msg into a new message of the type that
+// msgDesc selects from the descriptor of the method msg names. The method
+// name comes from the peer, so it is checked to name a registered method.
+// kind names the message in errors.
+func unmarshal(msg *stream.Message, msgDesc func(protoreflect.MethodDescriptor) protoreflect.MessageDescriptor, kind string) (proto.Message, error) {
+	method := protoreflect.FullName(msg.GetMethod())
+	desc, err := protoregistry.GlobalFiles.FindDescriptorByName(method)
 	if err != nil {
-		return nil, fmt.Errorf("gorums: could not find method descriptor for %s", out.GetMethod())
+		return nil, fmt.Errorf("gorums: could not find method descriptor for %s: %w", method, err)
 	}
-	methodDesc := desc.(protoreflect.MethodDescriptor)
-
-	// get the response message type (Output type)
-	msgType, err := protoregistry.GlobalTypes.FindMessageByName(methodDesc.Output().FullName())
+	methodDesc, ok := desc.(protoreflect.MethodDescriptor)
+	if !ok {
+		return nil, fmt.Errorf("gorums: %s is not a method", method)
+	}
+	name := msgDesc(methodDesc).FullName()
+	msgType, err := protoregistry.GlobalTypes.FindMessageByName(name)
 	if err != nil {
-		return nil, fmt.Errorf("gorums: could not find message type %s", methodDesc.Output().FullName())
+		return nil, fmt.Errorf("gorums: could not find message type %s: %w", name, err)
 	}
-	resp := msgType.New().Interface()
-
-	// unmarshal message from the Message.Payload field
-	payload := out.GetPayload()
-	if len(payload) > 0 {
-		if err := proto.Unmarshal(payload, resp); err != nil {
-			return nil, fmt.Errorf("gorums: could not unmarshal response: %w", err)
+	m := msgType.New().Interface()
+	if payload := msg.GetPayload(); len(payload) > 0 {
+		if err := proto.Unmarshal(payload, m); err != nil {
+			return nil, fmt.Errorf("gorums: could not unmarshal %s: %w", kind, err)
 		}
 	}
-	return resp, nil
+	return m, nil
 }

internal/impl/marshaling_test.go

@@ -0,0 +1,61 @@
+package impl
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/relab/gorums/internal/stream"
+	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/reflect/protoregistry"
+)
+
+// TestMarshalingUnmarshalMethodName verifies that request and response
+// unmarshaling resolve the message type from the method name, and that a
+// method name that is unknown or names a registered descriptor other than a
+// method fails with an error instead of a panic.
+func TestMarshalingUnmarshalMethodName(t *testing.T) {
+	payload, err := proto.Marshal(stream.Message_builder{MessageSeqNo: 7}.Build())
+	if err != nil {
+		t.Fatal(err)
+	}
+	unmarshalers := []struct {
+		name      string
+		unmarshal func(*stream.Message) (proto.Message, error)
+	}{
+		{name: "Request", unmarshal: UnmarshalRequest},
+		{name: "Response", unmarshal: unmarshalResponse},
+	}
+	tests := []struct {
+		name         string
+		method       string
+		wantNotFound bool
+		wantErr      bool
+	}{
+		{name: "Method", method: "stream.Gorums.NodeStream"},
+		{name: "UnknownMethod", method: "stream.Gorums.Missing", wantNotFound: true, wantErr: true},
+		{name: "MessageName", method: "stream.MetadataEntry", wantErr: true},
+		{name: "ServiceName", method: "stream.Gorums", wantErr: true},
+		{name: "FieldName", method: "stream.Message.entry", wantErr: true},
+	}
+	for _, u := range unmarshalers {
+		for _, tt := range tests {
+			t.Run(u.name+"/"+tt.name, func(t *testing.T) {
+				in := stream.Message_builder{Method: tt.method, Payload: payload}.Build()
+				msg, err := u.unmarshal(in)
+				if (err != nil) != tt.wantErr {
+					t.Fatalf("unmarshal(%q) error = %v, want error %t", tt.method, err, tt.wantErr)
+				}
+				if got := errors.Is(err, protoregistry.NotFound); got != tt.wantNotFound {
+					t.Errorf("errors.Is(%v, protoregistry.NotFound) = %t, want %t", err, got, tt.wantNotFound)
+				}
+				if tt.wantErr {
+					return
+				}
+				got, ok := msg.(*stream.Message)
+				if !ok || got.GetMessageSeqNo() != 7 {
+					t.Errorf("unmarshal(%q) = %v, want a stream.Message with MessageSeqNo 7", tt.method, msg)
+				}
+			})
+		}
+	}
+}

internal/stream/channel.go

@@ -25,6 +25,8 @@ type Channel interface {
 	// the send queue was full or closed.
 	DroppedReplies() int64
 	// PendingCount returns the number of two-way calls awaiting responses.
+	// The count may include calls whose callers are done but that have not
+	// yet been removed.
 	PendingCount() int
 	// Close closes the channel and fails its queued and pending requests.
 	Close() error