remove naive batch (#243)

Signed-off-by: Yoav Tock <tock@il.ibm.com>

Author: tock-ibm

node/assembler/assembler_role.go

@@ -7,7 +7,6 @@ SPDX-License-Identifier: Apache-2.0
 package assembler
 
 import (
-	"encoding/hex"
 	"errors"
 	"sync"
 	"time"
@@ -88,7 +87,6 @@ func (a *AssemblerRole) processOrderedBatchAttestations() {
 			return
 		}
 
-		t1 := time.Now()
 		batch, err := a.collateAttestationWithBatch(oba.BatchAttestation())
 		if err != nil {
 			if errors.Is(err, utils.ErrOperationCancelled) {
@@ -97,7 +95,6 @@ func (a *AssemblerRole) processOrderedBatchAttestations() {
 			}
 			a.Logger.Panicf("Something went wrong while fetching the batch %v", oba.BatchAttestation())
 		}
-		a.Logger.Infof("Located batch for digest %s within %v", ShortDigestString(oba.BatchAttestation().Digest()), time.Since(t1))
 		a.Ledger.Append(batch, oba.OrderingInfo())
 	}
 	a.Logger.Infof("Finished processing incoming OrderedBatchAttestations from consensus")
@@ -109,18 +106,6 @@ func (a *AssemblerRole) collateAttestationWithBatch(ba types.BatchAttestation) (
 	if err != nil {
 		return nil, err
 	}
-	a.Logger.Infof("Retrieved batch with %d requests for attestation %s from index within %v", len(batch.Requests()), ShortDigestString(ba.Digest()), time.Since(t1))
+	a.Logger.Infof("Retrieved full batch with %d requests from index within %s, BatchID: %s", len(batch.Requests()), time.Since(t1), types.BatchIDToString(ba))
 	return batch, nil
 }
-
-// ShortDigestString provides a short string from a potentially long (32B) digest.
-func ShortDigestString(digest []byte) string {
-	if digest == nil {
-		return "nil"
-	}
-	if len(digest) <= 8 {
-		return hex.EncodeToString(digest)
-	}
-
-	return hex.EncodeToString(digest[:8])
-}

node/assembler/assembler_role_test.go

@@ -8,8 +8,6 @@ package assembler_test
 
 import (
 	"encoding/binary"
-	"encoding/hex"
-	"encoding/json"
 	"fmt"
 	"sync"
 	"sync/atomic"
@@ -18,6 +16,7 @@ import (
 
 	"github.com/hyperledger/fabric-x-orderer/common/types"
 	"github.com/hyperledger/fabric-x-orderer/node/assembler"
+	"github.com/hyperledger/fabric-x-orderer/node/consensus/state"
 	"github.com/hyperledger/fabric-x-orderer/testutil"
 	"github.com/stretchr/testify/assert"
 )
@@ -31,39 +30,13 @@ func (n naiveOrderedBatchAttestationReplicator) Replicate() <-chan types.Ordered
 type naiveReplication struct {
 	subscribers []chan types.Batch
 	i           uint32
-	stopped     int32
 }
 
 func (r *naiveReplication) Replicate(_ types.ShardID) <-chan types.Batch {
 	j := atomic.AddUint32(&r.i, 1)
 	return r.subscribers[j-1]
 }
 
-func (r *naiveReplication) PullBatches(_ types.PartyID) <-chan types.Batch {
-	j := atomic.AddUint32(&r.i, 1)
-	return r.subscribers[j-1]
-}
-
-func (r *naiveReplication) Stop() {
-	atomic.StoreInt32(&r.stopped, 0x1)
-}
-
-func (r *naiveReplication) Append(partyID types.PartyID, batchSeq types.BatchSequence, batchedRequests types.BatchedRequests) {
-	for _, s := range r.subscribers {
-		s <- types.NewSimpleBatch(0, partyID, batchSeq, batchedRequests, 0)
-	}
-}
-
-func (r *naiveReplication) Height(partyID types.PartyID) uint64 {
-	// TODO use in test
-	return 0
-}
-
-func (r *naiveReplication) RetrieveBatchByNumber(partyID types.PartyID, seq uint64) types.Batch {
-	// TODO use in test
-	return nil
-}
-
 type naiveIndex struct {
 	sync.Map
 }
@@ -90,132 +63,21 @@ func (n *naiveIndex) PopOrWait(batchId types.BatchID) (types.Batch, error) {
 
 func (n *naiveIndex) Stop() {}
 
-type naiveBatchAttestation struct {
-	primary types.PartyID
-	seq     types.BatchSequence
-	node    types.PartyID
-	shard   types.ShardID
-	digest  []byte
-}
-
-func (nba *naiveBatchAttestation) Epoch() uint64 {
-	return 0
-}
-
-func (nba *naiveBatchAttestation) GarbageCollect() [][]byte {
-	return nil
-}
-
-func (nba *naiveBatchAttestation) Signer() types.PartyID {
-	return nba.node
-}
-
-func (nba *naiveBatchAttestation) Primary() types.PartyID {
-	return nba.primary
-}
-
-func (nba *naiveBatchAttestation) Seq() types.BatchSequence {
-	return nba.seq
-}
-
-func (nba *naiveBatchAttestation) Shard() types.ShardID {
-	return nba.shard
-}
-
-func (nba *naiveBatchAttestation) Digest() []byte {
-	return nba.digest
-}
-
-func (nba *naiveBatchAttestation) Fragments() []types.BatchAttestationFragment {
-	return nil
-}
-
-func (nba *naiveBatchAttestation) Serialize() []byte {
-	m := make(map[string]interface{})
-	m["seq"] = nba.seq
-	m["node"] = nba.node
-	m["shard"] = nba.shard
-	m["digest"] = hex.EncodeToString(nba.digest)
-
-	bytes, err := json.Marshal(m)
-	if err != nil {
-		panic(err)
-	}
-
-	return bytes
-}
-
-func (nba *naiveBatchAttestation) Deserialize(bytes []byte) error {
-	m := make(map[string]interface{})
-	if err := json.Unmarshal(bytes, &m); err != nil {
-		panic(err)
-	}
-
-	node := m["node"]
-	shard := m["shard"]
-	seq := m["seq"]
-	dig := m["digest"]
-
-	nba.node = types.PartyID(node.(float64))
-	nba.shard = types.ShardID(shard.(float64))
-	nba.seq = types.BatchSequence(seq.(float64))
-	nba.digest, _ = hex.DecodeString(dig.(string))
-
-	return nil
-}
-
-type naiveOrderingInfo struct {
-	num uint64
-}
-
-func (noi *naiveOrderingInfo) String() string {
-	return fmt.Sprintf("BlockNumber: %d", noi.num)
-}
-
-type naiveOrderedBatchAttestation struct {
-	ba           types.BatchAttestation
-	orderingInfo types.OrderingInfo
-}
-
-func (noba *naiveOrderedBatchAttestation) BatchAttestation() types.BatchAttestation {
-	return noba.ba
-}
-
-func (noba *naiveOrderedBatchAttestation) OrderingInfo() types.OrderingInfo {
-	return noba.orderingInfo
-}
-
 type naiveAssemblerLedger chan types.OrderedBatchAttestation
 
 func (n naiveAssemblerLedger) Append(batch types.Batch, orderingInfo types.OrderingInfo) {
-	noba := &naiveOrderedBatchAttestation{
-		ba: &naiveBatchAttestation{
-			primary: batch.Primary(),
-			seq:     batch.Seq(),
-			node:    batch.Primary(),
-			shard:   batch.Shard(),
-			digest:  batch.Digest(),
-		},
-		orderingInfo: orderingInfo,
+	noba := &state.AvailableBatchOrdered{
+		AvailableBatch:      state.NewAvailableBatch(batch.Primary(), batch.Shard(), batch.Seq(), batch.Digest()),
+		OrderingInformation: orderingInfo.(*state.OrderingInformation),
 	}
+
 	n <- noba
 }
 
 func (n naiveAssemblerLedger) Close() {
 	close(n)
 }
 
-func TestNaive(t *testing.T) {
-	nba := &naiveBatchAttestation{}
-	nba.seq = 100
-	nba.node = 1
-	nba.digest = []byte{1, 2, 3, 4, 5, 6, 7, 8}
-
-	nba2 := &naiveBatchAttestation{}
-	nba2.Deserialize(nba.Serialize())
-	assert.Equal(t, nba, nba2)
-}
-
 func TestAssembler(t *testing.T) {
 	shardCount := 4
 	batchNum := 20
@@ -236,7 +98,7 @@ func TestAssembler(t *testing.T) {
 		batches = append(batches, batchesForShard)
 	}
 
-	r, ledger, nobar, assembler := createAssembler(t, shardCount)
+	r, ledger, nobar, assembler := createAssemblerRole(t, shardCount)
 
 	assembler.Run()
 
@@ -245,20 +107,15 @@ func TestAssembler(t *testing.T) {
 		time.Sleep(time.Millisecond)
 	}
 
-	totalOrder := make(chan *naiveBatchAttestation)
+	totalOrder := make(chan *state.AvailableBatch)
 
 	for shardID := 0; shardID < shardCount; shardID++ {
 		go func(shard int) {
 			for _, batch := range batches[shard] {
 
 				r.subscribers[shard] <- batch
 
-				nba := &naiveBatchAttestation{
-					primary: batch.Primary(),
-					seq:     batch.Seq(),
-					shard:   batch.Shard(),
-					digest:  batch.Digest(),
-				}
+				nba := state.NewAvailableBatch(batch.Primary(), batch.Shard(), batch.Seq(), batch.Digest())
 				totalOrder <- nba
 			}
 		}(shardID)
@@ -268,9 +125,14 @@ func TestAssembler(t *testing.T) {
 		var num uint64
 
 		for nba := range totalOrder {
-			noba := &naiveOrderedBatchAttestation{
-				ba:           nba,
-				orderingInfo: &naiveOrderingInfo{num: num},
+			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,
+				},
 			}
 			nobar <- noba
 			num++
@@ -279,14 +141,14 @@ func TestAssembler(t *testing.T) {
 
 	for i := uint64(0); i < uint64(batchNum*shardCount); i++ {
 		noba := <-ledger
-		assert.Equal(t, fmt.Sprintf("BlockNumber: %d", i), noba.OrderingInfo().String())
+		assert.Equal(t, fmt.Sprintf("DecisionNum: %d, BatchIndex: 0, BatchCount: 1; No. Sigs: 0, BlockHeader: Number: %d, PrevHash: 08, Digest: 09", i, i), noba.OrderingInfo().String())
 		delete(digests, string(noba.BatchAttestation().Digest()))
 	}
 
 	assert.Len(t, digests, 0)
 }
 
-func createAssembler(t *testing.T, shardCount int) (*naiveReplication, naiveAssemblerLedger, naiveOrderedBatchAttestationReplicator, *assembler.AssemblerRole) {
+func createAssemblerRole(t *testing.T, shardCount int) (*naiveReplication, naiveAssemblerLedger, naiveOrderedBatchAttestationReplicator, *assembler.AssemblerRole) {
 	index := &naiveIndex{}
 
 	r := &naiveReplication{}