P2P comm session no more panic

- multiplexed websockets do not panic anymore
- safe send and receive on sessions
- add tests

Signed-off-by: Marcus Brandenburger <bur@zurich.ibm.com>

Author: mbrandenburger

platform/view/services/comm/host/rest/websocket/multiplexed_provider.go

@@ -63,7 +63,7 @@ func (c *MultiplexedProvider) KillAll() error {
 	}
 
 	// cleanup clients
-	c.clients = make(map[string]*multiplexedClientConn)
+	clear(c.clients)
 
 	return err
 }
@@ -227,11 +227,14 @@ func (c *multiplexedClientConn) readIncoming() {
 		c.mu.RUnlock()
 
 		if !ok && mm.Err == "" {
-			panic("subconn not found")
+			// it might happen that we receive a message from the server after we have already closed the sub-connection
+			// in this case we just ignore the message and drop it
+			logger.Warnf("client sub-connection does not exist mmId=%v, dropping message", mm.ID)
+			logger.Debugf("dropping message: `%s`", string(mm.Msg))
 		} else if !ok && mm.Err != "" {
-			logger.Debugf("Client subconnection errored: %v", mm.Err)
+			logger.Debugf("client sub-connection does not exist mmId=%v, errored: %v", mm.ID, mm.Err)
 		} else if mm.Err != "" {
-			logger.Debugf("Client subconn errored: %v", mm.Err)
+			logger.Debugf("client sub-connection mmId=%v errored: %v", mm.ID, mm.Err)
 		} else {
 			sc.receiverChan <- result{value: mm.Msg}
 		}

platform/view/services/comm/host/rest/websocket/multiplexed_provider_test.go

@@ -22,10 +22,12 @@ import (
 	"testing"
 	"time"
 
+	"github.com/gorilla/websocket"
 	"github.com/hyperledger-labs/fabric-smart-client/platform/common/services/logging"
 	"github.com/hyperledger-labs/fabric-smart-client/platform/view/services/comm/host"
 	"github.com/hyperledger-labs/fabric-smart-client/platform/view/services/metrics/disabled"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/otel/trace/noop"
 	"go.uber.org/goleak"
 )
@@ -147,6 +149,94 @@ func TestConnections(t *testing.T) {
 	p.mu.RUnlock()
 }
 
+func TestSendingOnClosedSubConnections(t *testing.T) {
+	testSetup(t)
+
+	// let check that at the end of this test all our go routines are stopped
+	defer goleak.VerifyNone(t, goleak.IgnoreCurrent())
+
+	p := NewMultiplexedProvider(noop.NewTracerProvider(), &disabled.Provider{})
+	var wg sync.WaitGroup
+
+	wait := make(chan struct{})
+
+	// server
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		err := p.NewServerStream(w, r, func(s host.P2PStream) {
+			wg.Add(1)
+			go func(srv host.P2PStream) {
+				defer wg.Done()
+				serverLogger.Debugf("[server] new stream established with %v (ID=%v) sessionID=%v", srv.RemotePeerID(), srv.RemotePeerID(), srv.Hash())
+				serverLogger.Debugf("[server] reading ...")
+
+				// server receives first message
+				answer, err := readMsg(srv)
+				require.NoError(t, err)
+				require.EqualValues(t, []byte("ping"), answer)
+
+				// sends back a message
+				err = sendMsg(srv, []byte("pong"))
+				require.NoError(t, err)
+
+				// we wait for next orders
+				<-wait
+
+				// we send a few messages at a high rate; eventually we should receive a channel closed message
+				require.EventuallyWithT(t, func(c *assert.CollectT) {
+					err = sendMsg(srv, []byte("pong again"))
+					require.ErrorIs(c, err, websocket.ErrCloseSent)
+				}, 100*time.Millisecond, time.Nanosecond)
+			}(s)
+		})
+		require.NoError(t, err)
+	}))
+
+	srvEndpoint := strings.TrimPrefix(srv.URL, "http://")
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	info := host.StreamInfo{
+		RemotePeerID:      "serverID",
+		RemotePeerAddress: srvEndpoint,
+		ContextID:         "someContextID",
+		SessionID:         "someSessionID",
+	}
+	src := host.PeerID("somePeerID")
+	config := &tls.Config{}
+
+	client, err := p.NewClientStream(info, ctx, src, config)
+	require.NoError(t, err)
+
+	// send ping
+	err = sendMsg(client, []byte("ping"))
+	require.NoError(t, err)
+
+	// expects pong
+	answer, err := readMsg(client)
+	require.NoError(t, err)
+	require.EqualValues(t, []byte("pong"), answer)
+
+	// now the client is actually done and closes the subconn
+	err = client.Close()
+	require.NoError(t, err)
+
+	//time.Sleep(snoozeTime)
+	// now we use our superpowers to let the server send another message
+	wait <- struct{}{}
+
+	wg.Wait()
+
+	// close server
+	srv.Close()
+
+	err = p.KillAll()
+	require.NoError(t, err)
+
+	p.mu.RLock()
+	require.Equal(t, 0, len(p.clients))
+	p.mu.RUnlock()
+}
+
 func testSetup(_ *testing.T) {
 	logSpec := cmp.Or(
 		os.Getenv("FABRIC_LOGGING_SPEC"),

platform/view/services/comm/master.go

@@ -31,21 +31,24 @@ func (p *P2PNode) getOrCreateSession(sessionID, endpointAddress, contextID, call
 		return session, nil
 	}
 
+	ctx, cancel := context.WithCancel(context.Background())
 	s := &NetworkStreamSession{
+		node:            p,
 		endpointID:      endpointID,
 		endpointAddress: endpointAddress,
 		contextID:       contextID,
-		callerViewID:    callerViewID,
-		caller:          caller,
 		sessionID:       sessionID,
-		node:            p,
+		caller:          caller,
+		callerViewID:    callerViewID,
 		incoming:        make(chan *view.Message, 1),
 		streams:         make(map[*streamHandler]struct{}),
+		ctx:             ctx,
+		cancel:          cancel,
 	}
 
 	if msg != nil {
 		logger.Debugf("pushing first message to [%s], [%s]", internalSessionID, msg)
-		s.incoming <- msg
+		s.enqueue(msg)
 	} else {
 		logger.Debugf("no first message to push to [%s]", internalSessionID)
 	}

platform/view/services/comm/p2p.go

@@ -148,7 +148,7 @@ func (p *P2PNode) dispatchMessages(ctx context.Context) {
 			p.dispatchMutex.Unlock()
 
 			logger.Debugf("pushing message to [%s], [%s]", internalSessionID, msg.message)
-			session.incoming <- msg.message
+			session.enqueue(msg.message)
 		case <-ctx.Done():
 			logger.Info("closing p2p comm...")
 			return

platform/view/services/comm/session.go

@@ -9,15 +9,25 @@ package comm
 import (
 	"context"
 	"sync"
+	"sync/atomic"
 
+	"github.com/hyperledger-labs/fabric-smart-client/pkg/utils/errors"
+	"github.com/hyperledger-labs/fabric-smart-client/pkg/utils/proto"
 	"github.com/hyperledger-labs/fabric-smart-client/platform/view/services/comm/host"
 	"github.com/hyperledger-labs/fabric-smart-client/platform/view/view"
 	"go.uber.org/zap/zapcore"
 )
 
+// ErrSessionClosed is returned when a message is sent when the session is closed.
+var ErrSessionClosed = errors.New("session closed")
+
+type sender interface {
+	sendTo(ctx context.Context, info host.StreamInfo, msg proto.Message) error
+}
+
 // NetworkStreamSession implements view.Session
 type NetworkStreamSession struct {
-	node            *P2PNode
+	node            sender
 	endpointID      []byte
 	endpointAddress string
 	contextID       string
@@ -26,10 +36,16 @@ type NetworkStreamSession struct {
 	callerViewID    string
 	incoming        chan *view.Message
 	streams         map[*streamHandler]struct{}
-	closed          bool
 	mutex           sync.RWMutex
+
+	closed    atomic.Bool
+	closeOnce sync.Once
+	ctx       context.Context
+	cancel    context.CancelFunc
+	wg        sync.WaitGroup
 }
 
+// Info returns a view.SessionInfo.
 func (n *NetworkStreamSession) Info() view.SessionInfo {
 	n.mutex.RLock()
 	defer n.mutex.RUnlock()
@@ -39,25 +55,27 @@ func (n *NetworkStreamSession) Info() view.SessionInfo {
 		CallerViewID: n.callerViewID,
 		Endpoint:     n.endpointAddress,
 		EndpointPKID: n.endpointID,
-		Closed:       n.closed,
+		Closed:       n.isClosed(),
 	}
 	return ret
 }
 
-// Send sends the payload to the endpoint
+// Send sends the payload to the endpoint.
 func (n *NetworkStreamSession) Send(payload []byte) error {
 	return n.SendWithContext(context.TODO(), payload)
 }
 
+// SendWithContext sends the payload to the endpoint with the passed context.Context.
 func (n *NetworkStreamSession) SendWithContext(ctx context.Context, payload []byte) error {
 	return n.sendWithStatus(ctx, payload, view.OK)
 }
 
-// SendError sends an error to the endpoint with the passed payload
+// SendError sends an error to the endpoint with the passed payload.
 func (n *NetworkStreamSession) SendError(payload []byte) error {
 	return n.SendErrorWithContext(context.TODO(), payload)
 }
 
+// SendErrorWithContext sends an error to the endpoint with the passed context.Context and payload.
 func (n *NetworkStreamSession) SendErrorWithContext(ctx context.Context, payload []byte) error {
 	return n.sendWithStatus(ctx, payload, view.ERROR)
 }
@@ -67,45 +85,77 @@ func (n *NetworkStreamSession) Receive() <-chan *view.Message {
 	return n.incoming
 }
 
+// enqueue enqueues a message into the session's incoming channel.
+// If the session is closed, the message will be dropped and a warning is logged.
+func (n *NetworkStreamSession) enqueue(msg *view.Message) {
+	if msg == nil {
+		return
+	}
+
+	if n.isClosed() {
+		logger.Warnf("dropping message from %s for closed session [%s]", msg.Caller, msg.SessionID)
+		return
+	}
+
+	n.wg.Add(1)
+	defer n.wg.Done()
+
+	select {
+	case <-n.ctx.Done():
+		logger.Warnf("dropping message from %s for closed session [%s]", msg.Caller, msg.SessionID)
+		return
+	case n.incoming <- msg:
+	}
+}
+
 // Close releases all the resources allocated by this session
 func (n *NetworkStreamSession) Close() {
-	n.node.sessionsMutex.Lock()
-	defer n.node.sessionsMutex.Unlock()
-
 	n.closeInternal()
 }
 
 func (n *NetworkStreamSession) closeInternal() {
-	if n.closed {
-		return
-	}
-
-	logger.Debugf("closing session [%s] with [%d] streams", n.sessionID, len(n.streams))
-	toClose := make([]*streamHandler, 0, len(n.streams))
-	for stream := range n.streams {
-		if logger.IsEnabledFor(zapcore.DebugLevel) {
-			logger.Debugf("session [%s], stream [%s], refCtr [%d]", n.sessionID, stream.stream.Hash(), stream.refCtr)
+	n.closeOnce.Do(func() {
+		n.mutex.Lock()
+		defer n.mutex.Unlock()
+
+		logger.Debugf("closing session [%s] with [%d] streams", n.sessionID, len(n.streams))
+		toClose := make([]*streamHandler, 0, len(n.streams))
+		for stream := range n.streams {
+			if logger.IsEnabledFor(zapcore.DebugLevel) {
+				logger.Debugf("session [%s], stream [%s], refCtr [%d]", n.sessionID, stream.stream.Hash(), stream.refCtr)
+			}
+			stream.refCtr--
+			if stream.refCtr == 0 {
+				toClose = append(toClose, stream)
+			}
 		}
-		stream.refCtr--
-		if stream.refCtr == 0 {
-			toClose = append(toClose, stream)
+
+		logger.Debugf("closing session [%s]'s streams [%d]", n.sessionID, len(toClose))
+		for _, stream := range toClose {
+			stream.close(context.TODO())
 		}
-	}
+		logger.Debugf("closing session [%s]'s streams [%d] done", n.sessionID, len(toClose))
 
-	logger.Debugf("closing session [%s]'s streams [%d]", n.sessionID, len(toClose))
-	for _, stream := range toClose {
-		stream.close(context.TODO())
-	}
+		// next we are closing the incoming and the closing signal channel to drain the receivers;
+		n.closed.Store(true)
+		n.cancel()
+		n.wg.Wait()
+		close(n.incoming)
+		n.streams = make(map[*streamHandler]struct{})
 
-	logger.Debugf("closing session [%s]'s streams [%d] done", n.sessionID, len(toClose))
-	close(n.incoming)
-	n.closed = true
-	n.streams = make(map[*streamHandler]struct{})
+		logger.Debugf("closing session [%s] done", n.sessionID)
+	})
+}
 
-	logger.Debugf("closing session [%s] done", n.sessionID)
+func (n *NetworkStreamSession) isClosed() bool {
+	return n.closed.Load()
 }
 
 func (n *NetworkStreamSession) sendWithStatus(ctx context.Context, payload []byte, status int32) error {
+	if n.isClosed() {
+		return ErrSessionClosed
+	}
+
 	n.mutex.RLock()
 	info := host.StreamInfo{
 		RemotePeerID:      string(n.endpointID),
@@ -123,11 +173,12 @@ func (n *NetworkStreamSession) sendWithStatus(ctx context.Context, payload []byt
 	n.mutex.RUnlock()
 
 	err := n.node.sendTo(ctx, info, packet)
-	logger.Debugf("sent message [len:%d] to [%s:%s] from [%s] with err [%v]",
+	logger.Debugf("sent message [len:%d] to [%s:%s] from [%s] [status:%v] with err [%v]",
 		len(payload),
 		info.RemotePeerID,
 		info.RemotePeerAddress,
 		packet.Caller,
+		status,
 		err,
 	)
 	return err