Extract and append config block in assembler (#330)

Signed-off-by: Yoav Tock <[email protected]>

Author: tock-ibm

node/assembler/assembler_role.go

@@ -11,8 +11,10 @@ import (
 	"sync"
 	"time"
 
+	"github.com/hyperledger/fabric-protos-go-apiv2/common"
 	"github.com/hyperledger/fabric-x-orderer/common/types"
 	"github.com/hyperledger/fabric-x-orderer/common/utils"
+	"github.com/hyperledger/fabric-x-orderer/node/consensus/state"
 )
 
 type AssemblerIndex interface {
@@ -23,6 +25,7 @@ type AssemblerIndex interface {
 
 type AssemblerLedgerWriter interface {
 	Append(batch types.Batch, orderingInfo types.OrderingInfo)
+	AppendConfig(configBlock *common.Block, decisionNum types.DecisionNum)
 	Close()
 }
 
@@ -62,7 +65,13 @@ func (a *AssemblerRole) processOrderedBatchAttestations() {
 		a.Logger.Infof("Received ordered batch attestation with BatchID: %s; OrderingInfo: %s", types.BatchIDToString(oba.BatchAttestation()), oba.OrderingInfo().String())
 
 		if oba.BatchAttestation().Shard() == types.ShardIDConsensus {
-			a.Logger.Infof("Config decision: shard: %d, primary: %d, Ignoring!", oba.BatchAttestation().Shard(), oba.BatchAttestation().Primary())
+			orderingInfo := oba.OrderingInfo()
+			a.Logger.Infof("Config decision: shard: %d, Ordering Info: %s", oba.BatchAttestation().Shard(), oba.OrderingInfo().String())
+			// TODO break the abstraction of oba.OrderingInfo().String()
+			ordInfo := orderingInfo.(*state.OrderingInformation)
+			block := ordInfo.CommonBlock
+			a.Ledger.AppendConfig(block, ordInfo.DecisionNum)
+			// TODO apply new config
 			return
 		}
 

node/assembler/assembler_role_test.go

@@ -7,17 +7,20 @@ SPDX-License-Identifier: Apache-2.0
 package assembler_test
 
 import (
-	"encoding/binary"
 	"fmt"
 	"sync"
 	"testing"
 	"time"
 
+	"github.com/hyperledger/fabric-protos-go-apiv2/common"
+	"github.com/hyperledger/fabric-x-common/protoutil"
 	"github.com/hyperledger/fabric-x-orderer/common/types"
+	"github.com/hyperledger/fabric-x-orderer/common/utils"
 	"github.com/hyperledger/fabric-x-orderer/node/assembler"
 	"github.com/hyperledger/fabric-x-orderer/node/consensus/state"
+	node_ledger "github.com/hyperledger/fabric-x-orderer/node/ledger"
 	"github.com/hyperledger/fabric-x-orderer/testutil"
-	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 type naiveOrderedBatchAttestationReplicator chan types.OrderedBatchAttestation
@@ -52,108 +55,199 @@ func (n *naiveIndex) PopOrWait(batchId types.BatchID) (types.Batch, error) {
 
 func (n *naiveIndex) Stop() {}
 
-type naiveAssemblerLedger chan types.OrderedBatchAttestation
+func TestAssemblerRole_Batches(t *testing.T) {
+	shardCount := 4
+	batchNum := 20
+	primaryID := types.PartyID(1)
 
-func (n naiveAssemblerLedger) Append(batch types.Batch, orderingInfo types.OrderingInfo) {
-	noba := &state.AvailableBatchOrdered{
-		AvailableBatch:      state.NewAvailableBatch(batch.Primary(), batch.Shard(), batch.Seq(), batch.Digest()),
-		OrderingInformation: orderingInfo.(*state.OrderingInformation),
-	}
+	// create test batches from all shards
+	digestsSet, batches := createTestBatches(t, shardCount, batchNum, primaryID)
+	// create test setup
+	index, ledger, ordBARep, assemblerRole := createAssemblerRole(t, shardCount)
 
-	n <- noba
-}
+	assemblerRole.Run()
+
+	totalOrder := make(chan types.OrderedBatchAttestation)
+
+	genBlock, err := ledger.LedgerReader().RetrieveBlockByNumber(0)
+	prevBlock := genBlock
+	require.NoError(t, err)
+	require.True(t, protoutil.IsConfigBlock(prevBlock), "ledger includes an empty genesis block")
+	decNum := types.DecisionNum(1)
+
+	simulateDecisions(t, batches, index, prevBlock, decNum, totalOrder)
+
+	go func() {
+		for ordererBatchAttestation := range totalOrder {
+			ordBARep <- ordererBatchAttestation
+		}
+	}()
+
+	require.Eventually(t, func() bool {
+		return ledger.LedgerReader().Height() == uint64(shardCount)*uint64(batchNum)+1
+	}, 10*time.Second, 100*time.Millisecond)
+
+	// verify all batches received
+	for blockNUm := uint64(0); blockNUm <= uint64(batchNum*shardCount); blockNUm++ {
+		block, err := ledger.LedgerReader().RetrieveBlockByNumber(blockNUm)
+		require.NoError(t, err)
+		if blockNUm == 0 {
+			require.True(t, protoutil.IsConfigBlock(block), "ledger includes an empty genesis block")
+			continue
+		}
+
+		batchID, ordInfo, _, _ := node_ledger.AssemblerBatchIdOrderingInfoAndTxCountFromBlock(block)
+
+		expectedBatch := batches[batchID.Shard()][batchID.Seq()]
+		require.Equal(t, types.BatchIDToString(expectedBatch), types.BatchIDToString(batchID))
+		require.Contains(t, ordInfo.String(), fmt.Sprintf("DecisionNum: %d, BatchIndex: 0, BatchCount: 1; No. Sigs: 0, BlockHeader: Number: %d", blockNUm, blockNUm), ordInfo.String())
 
-func (n naiveAssemblerLedger) Close() {
-	close(n)
+		delete(digestsSet, string(batchID.Digest()))
+	}
+
+	require.Len(t, digestsSet, 0)
 }
 
-func TestAssembler(t *testing.T) {
+func TestAssemblerRole_Config(t *testing.T) {
 	shardCount := 4
-	batchNum := 20
+	batchNum := 3
+	primaryID := types.PartyID(1)
 
-	digests := make(map[string]struct{})
+	// create test batches from all shards
+	_, batches := createTestBatches(t, shardCount, batchNum, primaryID)
+	// create test setup
+	index, ledger, ordBARep, assemblerRole := createAssemblerRole(t, shardCount)
 
-	var batches [][]types.Batch
-	for shardID := types.ShardID(0); shardID < types.ShardID(shardCount); shardID++ {
-		var batchesForShard []types.Batch
-		for seq := types.BatchSequence(0); seq < types.BatchSequence(batchNum); seq++ {
-			buff := make([]byte, 1024)
-			binary.BigEndian.PutUint16(buff, uint16(shardID))
-			binary.BigEndian.PutUint16(buff[100:], uint16(seq))
-			batch := types.NewSimpleBatch(shardID, 1, seq, [][]byte{buff}, 0)
-			digests[string(batch.Digest())] = struct{}{}
-			batchesForShard = append(batchesForShard, batch)
+	assemblerRole.Run()
+
+	totalOrder := make(chan types.OrderedBatchAttestation)
+
+	genBlock, err := ledger.LedgerReader().RetrieveBlockByNumber(0)
+	prevBlock := genBlock
+	require.NoError(t, err)
+	require.True(t, protoutil.IsConfigBlock(prevBlock), "ledger includes an empty genesis block")
+	decNum := types.DecisionNum(1)
+
+	simulateDecisions(t, batches, index, prevBlock, decNum, totalOrder)
+
+	go func() {
+		for ordererBatchAttestation := range totalOrder {
+			ordBARep <- ordererBatchAttestation
 		}
-		batches = append(batches, batchesForShard)
+	}()
+
+	require.Eventually(t, func() bool {
+		return ledger.LedgerReader().Height() == uint64(shardCount)*uint64(batchNum)+1
+	}, 10*time.Second, 100*time.Millisecond)
+
+	// consensus emits a config decision
+	lastBlock, err := ledger.LedgerReader().RetrieveBlockByNumber(ledger.LedgerReader().Height() - 1)
+	require.NoError(t, err)
+	configBlock := protoutil.UnmarshalBlockOrPanic(protoutil.MarshalOrPanic(genBlock)) // clone the block
+	configBlock.Header.PreviousHash = protoutil.BlockHeaderHash(lastBlock.Header)
+	configBlock.Header.Number = lastBlock.Header.Number + 1
+
+	configABO := &state.AvailableBatchOrdered{
+		AvailableBatch: state.NewAvailableBatch(0, types.ShardIDConsensus, 0, []byte{}),
+		OrderingInformation: &state.OrderingInformation{
+			CommonBlock: configBlock,
+			DecisionNum: decNum,
+			BatchIndex:  0,
+			BatchCount:  1,
+		},
 	}
+	totalOrder <- configABO
 
-	index, ledger, nobar, assembler := createAssemblerRole(t, shardCount)
+	require.Eventually(t, func() bool {
+		return ledger.LedgerReader().Height() == uint64(shardCount)*uint64(batchNum)+2
+	}, 10*time.Second, 100*time.Millisecond)
 
-	assembler.Run()
+	lastBlock, err = ledger.LedgerReader().RetrieveBlockByNumber(ledger.LedgerReader().Height() - 1)
+	require.NoError(t, err)
+	require.True(t, protoutil.IsConfigBlock(lastBlock))
+}
 
-	totalOrder := make(chan *state.AvailableBatch)
+func simulateDecisions(
+	t *testing.T,
+	batches [][]types.Batch,
+	index *naiveIndex,
+	prevBlock *common.Block,
+	decNum types.DecisionNum,
+	totalOrder chan types.OrderedBatchAttestation,
+) {
+	go func() {
+		for shardID := 0; shardID < len(batches); shardID++ {
+			for _, batch := range batches[shardID] {
 
-	for shardID := 0; shardID < shardCount; shardID++ {
-		go func(shard int) {
-			for _, batch := range batches[shard] {
 				index.Put(batch)
-				nba := state.NewAvailableBatch(batch.Primary(), batch.Shard(), batch.Seq(), batch.Digest())
-				totalOrder <- nba
-			}
-		}(shardID)
-	}
 
-	go func() {
-		var num uint64
-
-		for nba := range totalOrder {
-			noba := &state.AvailableBatchOrdered{
-				AvailableBatch: nba,
-				OrderingInformation: &state.OrderingInformation{
-					BlockHeader: &state.BlockHeader{Number: num, PrevHash: []byte{0x08}, Digest: []byte{0x09}},
-					DecisionNum: types.DecisionNum(num),
-					BatchIndex:  0,
-					BatchCount:  1,
-				},
+				ab := state.NewAvailableBatch(batch.Primary(), batch.Shard(), batch.Seq(), batch.Digest())
+				block := protoutil.NewBlock(prevBlock.Header.Number+1, protoutil.BlockHeaderHash(prevBlock.Header))
+				block.Header.DataHash = ab.Digest()
+
+				abo := &state.AvailableBatchOrdered{
+					AvailableBatch: ab,
+					OrderingInformation: &state.OrderingInformation{
+						CommonBlock: block,
+						DecisionNum: decNum,
+						BatchIndex:  0,
+						BatchCount:  1,
+					},
+				}
+				totalOrder <- abo
+
+				prevBlock = block
+				decNum++
 			}
-			nobar <- noba
-			num++
 		}
+
+		t.Logf("Simulated %d decisions", decNum)
 	}()
+}
 
-	for i := uint64(0); i < uint64(batchNum*shardCount); i++ {
-		noba := <-ledger
-		assert.Equal(t, fmt.Sprintf("DecisionNum: %d, BatchIndex: 0, BatchCount: 1; No. Sigs: 0, BlockHeader: Number: %d, PrevHash: 08, Digest: 09, Common Block: <nil>", i, i), noba.OrderingInfo().String())
-		delete(digests, string(noba.BatchAttestation().Digest()))
+func createTestBatches(t *testing.T, shardCount int, batchNum int, primaryID types.PartyID) (map[string]bool, [][]types.Batch) {
+	digestsSet := make(map[string]bool)
+	var batches [][]types.Batch
+	for shardID := types.ShardID(0); shardID < types.ShardID(shardCount); shardID++ {
+		var batchesForShard []types.Batch
+		for seq := types.BatchSequence(0); seq < types.BatchSequence(batchNum); seq++ {
+			buff := generateRandomBytes(t, 1024)
+			batch := types.NewSimpleBatch(shardID, primaryID, seq, [][]byte{buff}, 0)
+			digestsSet[string(batch.Digest())] = true
+			batchesForShard = append(batchesForShard, batch)
+		}
+		batches = append(batches, batchesForShard)
 	}
 
-	assert.Len(t, digests, 0)
+	return digestsSet, batches
 }
 
-func createAssemblerRole(t *testing.T, shardCount int) (
-	*naiveIndex,
-	naiveAssemblerLedger,
-	naiveOrderedBatchAttestationReplicator,
-	*assembler.AssemblerRole,
-) {
+func createAssemblerRole(t *testing.T, shardCount int) (*naiveIndex, node_ledger.AssemblerLedgerReaderWriter, naiveOrderedBatchAttestationReplicator, *assembler.AssemblerRole) {
+	tempDir := t.TempDir()
+
+	logger := testutil.CreateLogger(t, 0)
+
 	index := &naiveIndex{}
 
 	var shards []types.ShardID
 	for i := 0; i < shardCount; i++ {
 		shards = append(shards, types.ShardID(i))
 	}
 
-	ledger := make(naiveAssemblerLedger, 10)
+	ledgerFactory := &node_ledger.DefaultAssemblerLedgerFactory{}
+	ledger, err := ledgerFactory.Create(logger, tempDir)
+	require.NoError(t, err)
+	ledger.AppendConfig(utils.EmptyGenesisBlock("test"), 0)
 
-	nobar := make(naiveOrderedBatchAttestationReplicator)
+	ordBARep := make(naiveOrderedBatchAttestationReplicator)
 
-	assembler := &assembler.AssemblerRole{
+	assemblerRole := &assembler.AssemblerRole{
 		Shards:                            shards,
 		Logger:                            testutil.CreateLogger(t, 0),
 		Ledger:                            ledger,
 		ShardCount:                        shardCount,
-		OrderedBatchAttestationReplicator: nobar,
+		OrderedBatchAttestationReplicator: ordBARep,
 		Index:                             index,
 	}
-	return index, ledger, nobar, assembler
+	return index, ledger, ordBARep, assemblerRole
 }