add test for assembler-consenter reconnection (#126)

* add test for assembler-consenter reconnection

Signed-off-by: Natalie Morad <natalie.morad@ibm.com>

* fix review

Signed-off-by: Natalie Morad <natalie.morad@ibm.com>

---------

Signed-off-by: Natalie Morad <natalie.morad@ibm.com>

Author: moradna

node/assembler/assembler_batcher_consenter_test.go

@@ -16,65 +16,84 @@ import (
 	"github.com/hyperledger/fabric-x-orderer/node/assembler"
 	"github.com/hyperledger/fabric-x-orderer/node/comm/tlsgen"
 	"github.com/hyperledger/fabric-x-orderer/node/config"
+	"github.com/hyperledger/fabric-x-orderer/node/consensus/state"
 	"github.com/hyperledger/fabric-x-orderer/testutil"
 
 	"github.com/hyperledger/fabric-protos-go-apiv2/common"
 	"github.com/hyperledger/fabric-protos-go-apiv2/orderer"
 	"github.com/stretchr/testify/require"
 )
 
-func TestAssemblerAppendBlockAndIgnoreDuplicate(t *testing.T) {
+func TestAssemblerHandlesConsenterReconnect(t *testing.T) {
 	ca, err := tlsgen.NewCA()
 	require.NoError(t, err)
 
-	batcherStub := NewStubBatcher(t, types.ShardID(1), types.PartyID(1), ca)
-	defer batcherStub.Stop()
+	numParties := 4
+	partyID := types.PartyID(1)
+	shardID := types.ShardID(1)
 
-	consenterStub := NewStubConsenter(t, types.PartyID(1), ca)
-	defer consenterStub.Stop()
+	batchersStub, batcherInfos, cleanup := createStubBatchersAndInfos(t, numParties, shardID, ca)
+	defer cleanup()
 
-	assembler, clean := newAssemblerTest(t, 1, 1, ca, batcherStub.batcherInfo, consenterStub.consenterInfo)
-	defer clean()
+	consenterStub := NewStubConsenter(t, partyID, ca)
+	defer consenterStub.Shutdown()
 
-	// genesis block added
+	assembler := newAssemblerTest(t, partyID, shardID, ca, batcherInfos, consenterStub.consenterInfo)
+	defer assembler.Stop()
+
+	// wait for genesis block
 	require.Eventually(t, func() bool {
 		return assembler.GetTxCount() == 1
 	}, 3*time.Second, 100*time.Millisecond)
 
 	obaCreator, _ := NewOrderedBatchAttestationCreator()
 
-	// create batch with 1 req, send from batcher and consenter
+	// send batch and matching decision
 	batch1 := testutil.CreateMockBatch(1, 1, 1, []int{1})
-	batcherStub.SetNextBatch(batch1)
-
+	batchersStub[0].SetNextBatch(batch1)
 	oba1 := obaCreator.Append(batch1, 1, 1, 1)
-	consenterStub.SetNextDecision(oba1)
+	consenterStub.SetNextDecision(oba1.(*state.AvailableBatchOrdered))
 
 	require.Eventually(t, func() bool {
 		return assembler.GetTxCount() == 2
 	}, 3*time.Second, 100*time.Millisecond)
 
-	// create batch with 2 reqs, send from batcher and consenter
+	// stop consenter and send next batch
+	consenterStub.Stop()
+
 	batch2 := testutil.CreateMockBatch(1, 1, 2, []int{2, 3})
-	batcherStub.SetNextBatch(batch2)
+	batchersStub[0].SetNextBatch(batch2)
+
+	// restart consenter and send matching decision
+	consenterStub.Restart()
 
 	oba2 := obaCreator.Append(batch2, 2, 1, 1)
-	consenterStub.SetNextDecision(oba2)
+	consenterStub.SetNextDecision(oba2.(*state.AvailableBatchOrdered))
 
 	require.Eventually(t, func() bool {
 		return assembler.GetTxCount() == 4
 	}, 3*time.Second, 100*time.Millisecond)
 
-	// send duplicate batch+oba, should be ignored
-	batcherStub.SetNextBatch(batch2)
-	consenterStub.SetNextDecision(oba2)
+	// send next decision and restart consenter
+	batch3 := testutil.CreateMockBatch(1, 1, 3, []int{4})
+	oba3 := obaCreator.Append(batch3, 3, 1, 1)
+	consenterStub.SetNextDecision(oba3.(*state.AvailableBatchOrdered))
+
+	// wait for decistion will be sent
+	time.Sleep(3 * time.Second)
+
+	consenterStub.Stop()
+	consenterStub.Restart()
+
+	// send matching batch
+	batchersStub[0].SetNextBatch(batch3)
 
-	require.Never(t, func() bool {
-		return assembler.GetTxCount() > 4
-	}, 3*time.Millisecond, 100*time.Millisecond)
+	require.Eventually(t, func() bool {
+		return assembler.GetTxCount() == 5
+	}, 3*time.Second, 100*time.Millisecond)
 }
 
-func newAssemblerTest(t *testing.T, partyID int, shardID int, ca tlsgen.CA, batcherInfo config.BatcherInfo, consenterInfo config.ConsenterInfo) (*assembler.Assembler, func()) {
+func newAssemblerTest(t *testing.T, partyID types.PartyID, shardID types.ShardID, ca tlsgen.CA, batchersInfo []config.BatcherInfo, consenterInfo config.ConsenterInfo) *assembler.Assembler {
 	genesisBlock := utils.EmptyGenesisBlock("arma")
 	genesisBlock.Metadata = &common.BlockMetadata{
 		Metadata: [][]byte{nil, nil, []byte("dummy"), []byte("dummy")},
@@ -85,19 +104,17 @@ func newAssemblerTest(t *testing.T, partyID int, shardID int, ca tlsgen.CA, batc
 
 	shards := []config.ShardInfo{
 		{
-			ShardId: types.ShardID(shardID),
-			Batchers: []config.BatcherInfo{
-				batcherInfo,
-			},
+			ShardId:  shardID,
+			Batchers: batchersInfo,
 		},
 	}
 
 	nodeConfig := &config.AssemblerNodeConfig{
 		TLSPrivateKeyFile:         ckp.Key,
 		TLSCertificateFile:        ckp.Cert,
-		PartyId:                   types.PartyID(partyID),
+		PartyId:                   partyID,
 		Directory:                 t.TempDir(),
-		ListenAddress:             "0.0.0.0:0",
+		ListenAddress:             "127.0.0.1:0",
 		PrefetchBufferMemoryBytes: 1 * 1024 * 1024 * 1024,
 		RestartLedgerScanTimeout:  5 * time.Second,
 		PrefetchEvictionTtl:       time.Hour,
@@ -111,15 +128,30 @@ func newAssemblerTest(t *testing.T, partyID int, shardID int, ca tlsgen.CA, batc
 
 	assemblerGRPC := node.CreateGRPCAssembler(nodeConfig)
 
-	assembler := assembler.NewAssembler(nodeConfig, assemblerGRPC, genesisBlock, testutil.CreateLogger(t, partyID))
+	assembler := assembler.NewAssembler(nodeConfig, assemblerGRPC, genesisBlock, testutil.CreateLogger(t, int(partyID)))
 
 	orderer.RegisterAtomicBroadcastServer(assemblerGRPC.Server(), assembler)
 	go func() {
 		err := assemblerGRPC.Start()
 		require.NoError(t, err)
 	}()
 
-	return assembler, func() {
-		assembler.Stop()
+	return assembler
+}
+
+func createStubBatchersAndInfos(t *testing.T, numParties int, shardID types.ShardID, ca tlsgen.CA) ([]*stubBatcher, []config.BatcherInfo, func()) {
+	var batchers []*stubBatcher
+	var batcherInfos []config.BatcherInfo
+
+	for i := 1; i <= numParties; i++ {
+		b := NewStubBatcher(t, shardID, types.PartyID(i), ca)
+		batchers = append(batchers, b)
+		batcherInfos = append(batcherInfos, b.batcherInfo)
+	}
+
+	return batchers, batcherInfos, func() {
+		for _, b := range batchers {
+			b.Shutdown()
+		}
 	}
 }

node/assembler/stub_batcher_test.go

@@ -8,7 +8,6 @@ package assembler_test
 
 import (
 	"fmt"
-	"sync"
 	"testing"
 
 	"github.com/hyperledger/fabric-x-orderer/common/types"
@@ -31,8 +30,7 @@ type stubBatcher struct {
 	cert        []byte
 	key         []byte
 	batcherInfo config.BatcherInfo
-	storedBatch *common.Block
-	blockLock   sync.Mutex
+	batches     chan *common.Block
 }
 
 func NewStubBatcher(t *testing.T, shardID types.ShardID, partyID types.PartyID, ca tlsgen.CA) *stubBatcher {
@@ -64,6 +62,7 @@ func NewStubBatcher(t *testing.T, shardID types.ShardID, partyID types.PartyID,
 		cert:        certKeyPair.Cert,
 		key:         certKeyPair.Key,
 		batcherInfo: batcherInfo,
+		batches:     make(chan *common.Block, 100),
 	}
 
 	orderer.RegisterAtomicBroadcastServer(server.Server(), stubBatcher)
@@ -79,21 +78,58 @@ func (sb *stubBatcher) Stop() {
 	sb.server.Stop()
 }
 
-func (sb *stubBatcher) Broadcast(stream orderer.AtomicBroadcast_BroadcastServer) error {
-	return fmt.Errorf("not implemented")
+func (sb *stubBatcher) Shutdown() {
+	close(sb.batches)
+	sb.server.Stop()
 }
 
-func (sb *stubBatcher) Deliver(stream orderer.AtomicBroadcast_DeliverServer) error {
-	sb.blockLock.Lock()
-	defer sb.blockLock.Unlock()
-	return stream.Send(&orderer.DeliverResponse{
-		Type: &orderer.DeliverResponse_Block{Block: sb.storedBatch},
+func (sb *stubBatcher) Restart() {
+	server, err := comm.NewGRPCServer(sb.endpoint, comm.ServerConfig{
+		SecOpts: comm.SecureOptions{
+			UseTLS:      true,
+			Certificate: sb.cert,
+			Key:         sb.key,
+		},
 	})
+	if err != nil {
+		panic(fmt.Sprintf("failed to restart gRPC server: %v", err))
+	}
+
+	sb.server = server
+
+	orderer.RegisterAtomicBroadcastServer(server.Server(), sb)
+
+	go func() {
+		if err := sb.server.Start(); err != nil {
+			panic(err)
+		}
+	}()
+}
+
+func (sb *stubBatcher) Deliver(stream orderer.AtomicBroadcast_DeliverServer) error {
+	for {
+		select {
+		case b := <-sb.batches:
+			if b == nil {
+				return nil
+			}
+			err := stream.Send(&orderer.DeliverResponse{
+				Type: &orderer.DeliverResponse_Block{Block: b},
+			})
+			if err != nil {
+				return err
+			}
+		case <-stream.Context().Done():
+			return stream.Context().Err()
+		}
+	}
+}
+
+func (sb *stubBatcher) Broadcast(stream orderer.AtomicBroadcast_BroadcastServer) error {
+	return fmt.Errorf("not implemented")
 }
 
 func (sb *stubBatcher) SetNextBatch(batch core.Batch) {
 	block, _ := ledger.NewFabricBatchFromRequests(sb.partyID, sb.shardID, batch.Seq(), batch.Requests(), []byte(""))
-	sb.blockLock.Lock()
-	defer sb.blockLock.Unlock()
-	sb.storedBatch = (*common.Block)(block)
+	sb.batches <- (*common.Block)(block)
 }

node/assembler/stub_consenter_test.go

@@ -9,11 +9,9 @@ package assembler_test
 import (
 	"encoding/asn1"
 	"fmt"
-	"sync"
 	"testing"
 
 	"github.com/hyperledger/fabric-x-orderer/common/types"
-	"github.com/hyperledger/fabric-x-orderer/core"
 	"github.com/hyperledger/fabric-x-orderer/node/comm"
 	"github.com/hyperledger/fabric-x-orderer/node/comm/tlsgen"
 	"github.com/hyperledger/fabric-x-orderer/node/config"
@@ -32,8 +30,7 @@ type stubConsenter struct {
 	cert          []byte
 	key           []byte
 	consenterInfo config.ConsenterInfo
-	storedBlock   *common.Block
-	blockLock     sync.Mutex
+	decisions     chan *common.Block
 }
 
 func NewStubConsenter(t *testing.T, partyID types.PartyID, ca tlsgen.CA) *stubConsenter {
@@ -63,6 +60,7 @@ func NewStubConsenter(t *testing.T, partyID types.PartyID, ca tlsgen.CA) *stubCo
 		key:           certKeyPair.Key,
 		endpoint:      server.Address(),
 		consenterInfo: consenterInfo,
+		decisions:     make(chan *common.Block, 100),
 	}
 
 	orderer.RegisterAtomicBroadcastServer(server.Server(), stubConsenter)
@@ -75,25 +73,62 @@ func NewStubConsenter(t *testing.T, partyID types.PartyID, ca tlsgen.CA) *stubCo
 	return stubConsenter
 }
 
-func (sc *stubConsenter) Deliver(stream orderer.AtomicBroadcast_DeliverServer) error {
-	sc.blockLock.Lock()
-	defer sc.blockLock.Unlock()
-	return stream.Send(&orderer.DeliverResponse{
-		Type: &orderer.DeliverResponse_Block{Block: sc.storedBlock},
-	})
+func (sc *stubConsenter) Stop() {
+	sc.server.Stop()
 }
 
-func (sc *stubConsenter) Broadcast(stream orderer.AtomicBroadcast_BroadcastServer) error {
-	return fmt.Errorf("not implemented")
+func (sc *stubConsenter) Shutdown() {
+	close(sc.decisions)
+	sc.server.Stop()
 }
 
-func (sc *stubConsenter) Stop() {
-	sc.server.Stop()
+func (sc *stubConsenter) Restart() {
+	server, err := comm.NewGRPCServer(sc.endpoint, comm.ServerConfig{
+		SecOpts: comm.SecureOptions{
+			UseTLS:      true,
+			Certificate: sc.cert,
+			Key:         sc.key,
+		},
+	})
+	if err != nil {
+		panic(fmt.Sprintf("failed to restart gRPC server: %v", err))
+	}
+
+	sc.server = server
+
+	orderer.RegisterAtomicBroadcastServer(server.Server(), sc)
+
+	go func() {
+		if err := sc.server.Start(); err != nil {
+			panic(err)
+		}
+	}()
+}
+
+func (sc *stubConsenter) Deliver(stream orderer.AtomicBroadcast_DeliverServer) error {
+	for {
+		select {
+		case b := <-sc.decisions:
+			if b == nil {
+				return nil
+			}
+			err := stream.Send(&orderer.DeliverResponse{
+				Type: &orderer.DeliverResponse_Block{Block: b},
+			})
+			if err != nil {
+				return err
+			}
+		case <-stream.Context().Done():
+			return stream.Context().Err()
+		}
+	}
 }
 
-func (sc *stubConsenter) SetNextDecision(oba core.OrderedBatchAttestation) {
-	ba := oba.(*state.AvailableBatchOrdered)
+func (sc *stubConsenter) Broadcast(stream orderer.AtomicBroadcast_BroadcastServer) error {
+	return fmt.Errorf("not implemented")
+}
 
+func (sc *stubConsenter) SetNextDecision(ba *state.AvailableBatchOrdered) {
 	proposal := smartbft_types.Proposal{
 		Header: (&state.Header{
 			Num: ba.OrderingInformation.DecisionNum,
@@ -113,9 +148,11 @@ func (sc *stubConsenter) SetNextDecision(oba core.OrderedBatchAttestation) {
 	signatures := []smartbft_types.Signature{{Value: sigBytes}}
 	bytes := state.DecisionToBytes(proposal, signatures)
 
-	sc.blockLock.Lock()
-	defer sc.blockLock.Unlock()
-	sc.storedBlock = &common.Block{
+	sc.decisions <- &common.Block{
+		Header: &common.BlockHeader{
+			Number:       uint64(ba.OrderingInformation.DecisionNum),
+			PreviousHash: ba.OrderingInformation.PrevHash,
+		},
 		Data: &common.BlockData{Data: [][]byte{bytes}},
 	}
 }