Fix flaky session test

The TestSessionLifecycleConcurrent tests if messages can be enqueued on
a closed session. As `enqueue` should drop messages when a session is
closed, the test checked that the underlying chan is empty. However,
this test is flaky as it depends on the consumer goroutine to consume
all messages.

To simplify this test, `enqueue` now returns true if a message was
enqueued in the session, or false if the message is dropped. The test
can now check the returned value.

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

Author: mbrandenburger

platform/view/services/comm/master.go

@@ -48,7 +48,9 @@ func (p *P2PNode) getOrCreateSession(sessionID, endpointAddress, contextID, call
 
 	if msg != nil {
 		logger.Debugf("pushing first message to [%s], [%s]", internalSessionID, msg)
-		s.enqueue(msg)
+		if ok := s.enqueue(msg); !ok {
+			panic("programming error: can not enqueue message in newly created session")
+		}
 	} else {
 		logger.Debugf("no first message to push to [%s]", internalSessionID)
 	}

platform/view/services/comm/p2p.go

@@ -147,8 +147,11 @@ func (p *P2PNode) dispatchMessages(ctx context.Context) {
 			}
 			p.dispatchMutex.Unlock()
 
-			logger.Debugf("pushing message to [%s], [%s]", internalSessionID, msg.message)
-			session.enqueue(msg.message)
+			if ok = session.enqueue(msg.message); ok {
+				logger.Debugf("pushing message to [%s], [%s]", internalSessionID, msg.message)
+			} else {
+				logger.Warnf("dropping message from %s for closed session [%s]", msg.message.Caller, msg.message.SessionID)
+			}
 		case <-ctx.Done():
 			logger.Info("closing p2p comm...")
 			return

platform/view/services/comm/session.go

@@ -86,15 +86,14 @@ func (n *NetworkStreamSession) Receive() <-chan *view.Message {
 }
 
 // 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 the session is closed, the message will be dropped and false returned, otherwise true is returned.
+func (n *NetworkStreamSession) enqueue(msg *view.Message) bool {
 	if msg == nil {
-		return
+		return false
 	}
 
 	if n.isClosed() {
-		logger.Warnf("dropping message from %s for closed session [%s]", msg.Caller, msg.SessionID)
-		return
+		return false
 	}
 
 	n.wg.Add(1)
@@ -103,8 +102,9 @@ func (n *NetworkStreamSession) enqueue(msg *view.Message) {
 	select {
 	case <-n.ctx.Done():
 		logger.Warnf("dropping message from %s for closed session [%s]", msg.Caller, msg.SessionID)
-		return
+		return false
 	case n.incoming <- msg:
+		return true
 	}
 }
 

platform/view/services/comm/session_test.go

@@ -14,6 +14,7 @@ import (
 	"time"
 
 	"github.com/hyperledger-labs/fabric-smart-client/pkg/utils/proto"
+	"github.com/hyperledger-labs/fabric-smart-client/platform/common/services/logging"
 	host2 "github.com/hyperledger-labs/fabric-smart-client/platform/view/services/comm/host"
 	"github.com/hyperledger-labs/fabric-smart-client/platform/view/view"
 	"github.com/stretchr/testify/assert"
@@ -34,6 +35,10 @@ func (m *mockSender) sendTo(ctx context.Context, info host2.StreamInfo, msg prot
 }
 
 func setup() *NetworkStreamSession {
+	logging.Init(logging.Config{
+		LogSpec: "fsc.view.services.comm=error",
+	})
+
 	sessionID := "someSessionID"
 	contextID := "someContextID"
 	endpointAddress := "someEndpointAddress"
@@ -132,16 +137,15 @@ func TestSessionLifecycleConcurrent(t *testing.T) {
 
 		// send a few messages
 		for i := range numMessage {
-			s.enqueue(&view.Message{Payload: []byte(fmt.Sprintf("msg #%v", i))})
+			assert.True(t, s.enqueue(&view.Message{Payload: []byte(fmt.Sprintf("msg #%v", i))}))
 		}
 
 		// once we delivered all our messages we close
 		sess.Close()
 
 		// try to send more but the session should not accept them
 		for i := range numMessage {
-			s.enqueue(&view.Message{Payload: []byte(fmt.Sprintf("msg #%v", i))})
-			assert.Empty(t, s.incoming)
+			assert.False(t, s.enqueue(&view.Message{Payload: []byte(fmt.Sprintf("msg #%v", i))}))
 		}
 	}()
 
@@ -157,6 +161,7 @@ func TestSessionLifecycleConcurrent(t *testing.T) {
 		}
 		assert.Equal(t, numMessage, cnt)
 		assert.True(t, sess.Info().Closed)
+		assert.Empty(t, s.incoming)
 	}()
 
 	wg.Wait()