BFT chain unit tests: successful tx propagation

Signed-off-by: May Rosenbaum <[email protected]>

orderer/consensus/smartbft/chain.go

@@ -35,7 +35,7 @@ import (
 	"go.uber.org/zap"
 )
 
-//go:generate counterfeiter -o mocks/mock_blockpuller.go . BlockPuller
+//go:generate mockery --dir . --name BlockPuller --case underscore --with-expecter=true --output mocks
 
 // BlockPuller is used to pull blocks from other OSN
 type BlockPuller interface {
@@ -52,6 +52,8 @@ type WALConfig struct {
 }
 
 // ConfigValidator interface
+//
+//go:generate mockery --dir . --name ConfigValidator --case underscore --with-expecter=true --output mocks
 type ConfigValidator interface {
 	ValidateConfig(env *cb.Envelope) error
 }
@@ -100,12 +102,13 @@ func NewChain(
 	blockPuller BlockPuller,
 	comm cluster.Communicator,
 	signerSerializer signerSerializer,
-	policyManager policies.Manager,
+	policyManager PolicyManager,
 	support consensus.ConsenterSupport,
 	metrics *Metrics,
 	metricsBFT *api.Metrics,
 	metricsWalBFT *wal.Metrics,
 	bccsp bccsp.BCCSP,
+	egressCommFactory EgressCommFactory,
 ) (*BFTChain, error) {
 	logger := flogging.MustGetLogger("orderer.consensus.smartbft.chain").With(zap.String("channel", support.ChannelID()))
 
@@ -159,7 +162,7 @@ func NewChain(
 	c.RuntimeConfig.Store(rtc)
 
 	c.verifier = buildVerifier(cv, c.RuntimeConfig, support, requestInspector, policyManager)
-	c.consensus = bftSmartConsensusBuild(c, requestInspector)
+	c.consensus = bftSmartConsensusBuild(c, requestInspector, egressCommFactory)
 
 	// Setup communication with list of remotes notes for the new channel
 	c.Comm.Configure(c.support.ChannelID(), rtc.RemoteNodes)
@@ -176,6 +179,7 @@ func NewChain(
 func bftSmartConsensusBuild(
 	c *BFTChain,
 	requestInspector *RequestInspector,
+	egressCommFactory EgressCommFactory,
 ) *smartbft.Consensus {
 	var err error
 
@@ -258,18 +262,7 @@ func bftSmartConsensusBuild(
 		Assembler:         c.assembler,
 		RequestInspector:  requestInspector,
 		Synchronizer:      sync,
-		Comm: &Egress{
-			RuntimeConfig: c.RuntimeConfig,
-			Channel:       c.support.ChannelID(),
-			Logger:        flogging.MustGetLogger("orderer.consensus.smartbft.egress").With(channelDecorator),
-			RPC: &cluster.RPC{
-				Logger:        flogging.MustGetLogger("orderer.consensus.smartbft.rpc").With(channelDecorator),
-				Channel:       c.support.ChannelID(),
-				StreamsByType: cluster.NewStreamsByType(),
-				Comm:          c.Comm,
-				Timeout:       5 * time.Minute, // Externalize configuration
-			},
-		},
+		Comm:              egressCommFactory(c.RuntimeConfig, c.Channel, c.Comm),
 		Scheduler:         time.NewTicker(time.Second).C,
 		ViewChangerTicker: time.NewTicker(time.Second).C,
 	}

orderer/consensus/smartbft/chain_test.go

@@ -5,7 +5,18 @@
 package smartbft_test
 
 import (
+	"bytes"
+	"encoding/binary"
 	"testing"
+
+	cb "github.com/hyperledger/fabric-protos-go/common"
+	"github.com/hyperledger/fabric-protos-go/msp"
+	"github.com/hyperledger/fabric/protoutil"
+	"github.com/stretchr/testify/require"
+)
+
+var (
+	nonce uint64 = 0
 )
 
 // Scenario:
@@ -17,6 +28,49 @@ func TestSuccessfulTxPropagation(t *testing.T) {
 	channelId := "testchannel"
 
 	networkSetupInfo := NewNetworkSetupInfo(t, channelId, dir)
-	_ = networkSetupInfo.CreateNodes(4)
+	nodeMap := networkSetupInfo.CreateNodes(4)
 	networkSetupInfo.StartAllNodes()
+
+	for _, node := range nodeMap {
+		node.State.WaitLedgerHeightToBe(1)
+	}
+
+	env := createEndorserTxEnvelope("TEST_MESSAGE #1", channelId)
+	err := networkSetupInfo.SendTxToAllAvailableNodes(env)
+	require.NoError(t, err)
+	for _, node := range nodeMap {
+		node.State.WaitLedgerHeightToBe(2)
+	}
+}
+
+func createEndorserTxEnvelope(message string, channelId string) *cb.Envelope {
+	return &cb.Envelope{
+		Payload: protoutil.MarshalOrPanic(&cb.Payload{
+			Header: &cb.Header{
+				ChannelHeader: protoutil.MarshalOrPanic(&cb.ChannelHeader{
+					Type:      int32(cb.HeaderType_ENDORSER_TRANSACTION),
+					ChannelId: channelId,
+				}),
+				SignatureHeader: protoutil.MarshalOrPanic(&cb.SignatureHeader{
+					Creator: protoutil.MarshalOrPanic(&msp.SerializedIdentity{
+						Mspid:   "mockMSP",
+						IdBytes: []byte("mockClient"),
+					}),
+					Nonce: generateNonce(),
+				}),
+			},
+			Data: []byte(message),
+		}),
+		Signature: []byte{1, 2, 3},
+	}
+}
+
+func generateNonce() []byte {
+	nonceBuf := new(bytes.Buffer)
+	err := binary.Write(nonceBuf, binary.LittleEndian, nonce)
+	if err != nil {
+		panic("Cannot generate nonce")
+	}
+	nonce++
+	return nonceBuf.Bytes()
 }

orderer/consensus/smartbft/consenter.go

@@ -14,6 +14,10 @@ import (
 	"encoding/pem"
 	"path"
 	"reflect"
+	"sync/atomic"
+	"time"
+
+	"go.uber.org/zap"
 
 	"github.com/hyperledger/fabric/orderer/consensus/smartbft/util"
 
@@ -203,7 +207,23 @@ func (c *Consenter) HandleChain(support consensus.ConsenterSupport, metadata *cb
 		Logger:               c.Logger,
 	}
 
-	chain, err := NewChain(configValidator, (uint64)(selfID), config, path.Join(c.WALBaseDir, support.ChannelID()), puller, c.Comm, c.SignerSerializer, c.GetPolicyManager(support.ChannelID()), support, c.Metrics, c.MetricsBFT, c.MetricsWalBFT, c.BCCSP)
+	egressCommFactory := func(runtimeConfig *atomic.Value, channelId string, comm Communicator) EgressComm {
+		channelDecorator := zap.String("channel", channelId)
+		return &Egress{
+			RuntimeConfig: runtimeConfig,
+			Channel:       channelId,
+			Logger:        flogging.MustGetLogger("orderer.consensus.smartbft.egress").With(channelDecorator),
+			RPC: &cluster.RPC{
+				Logger:        flogging.MustGetLogger("orderer.consensus.smartbft.rpc").With(channelDecorator),
+				Channel:       channelId,
+				StreamsByType: cluster.NewStreamsByType(),
+				Comm:          comm,
+				Timeout:       5 * time.Minute, // Externalize configuration
+			},
+		}
+	}
+
+	chain, err := NewChain(configValidator, (uint64)(selfID), config, path.Join(c.WALBaseDir, support.ChannelID()), puller, c.Comm, c.SignerSerializer, c.GetPolicyManager(support.ChannelID()), support, c.Metrics, c.MetricsBFT, c.MetricsWalBFT, c.BCCSP, egressCommFactory)
 	if err != nil {
 		return nil, errors.Wrap(err, "failed creating a new BFTChain")
 	}

orderer/consensus/smartbft/interfaces.go

@@ -7,6 +7,10 @@ SPDX-License-Identifier: Apache-2.0
 package smartbft
 
 import (
+	"sync/atomic"
+
+	protos "github.com/SmartBFT-Go/consensus/smartbftprotos"
+	"github.com/hyperledger/fabric/common/policies"
 	"github.com/hyperledger/fabric/orderer/common/cluster"
 	"github.com/hyperledger/fabric/orderer/consensus"
 )
@@ -24,3 +28,32 @@ type ConsenterSupport interface {
 type Communicator interface {
 	cluster.Communicator
 }
+
+//go:generate mockery --dir . --name PolicyManager --case underscore --with-expecter=true --output mocks
+
+// PolicyManager is a read only subset of the policy ManagerImpl
+type PolicyManager interface {
+	policies.Manager
+}
+
+//go:generate mockery --dir . --name Policy --case underscore --with-expecter=true --output mocks
+
+// Policy is used to determine if a signature is valid
+type Policy interface {
+	policies.Policy
+}
+
+//go:generate mockery --dir . --name EgressComm --case underscore --with-expecter=true --output mocks
+
+type EgressCommFactory func(runtimeConfig *atomic.Value, channelId string, comm Communicator) EgressComm
+
+// Comm enables the communications between the nodes.
+type EgressComm interface {
+	// SendConsensus sends the consensus protocol related message m to the node with id targetID.
+	SendConsensus(targetID uint64, m *protos.Message)
+	// SendTransaction sends the given client's request to the node with id targetID.
+	SendTransaction(targetID uint64, request []byte)
+	// Nodes returns a set of ids of participating nodes.
+	// In case you need to change or keep this slice, create a copy.
+	Nodes() []uint64
+}