[FAB-1840] Use block cutter to create batches

https://jira.hyperledger.org/browse/FAB-1840

Change-Id: If53dbe94604ae3cca1083bb31d40798a16adee02
Signed-off-by: Gabor Hosszu <[email protected]>

Author: gaborh-da

orderer/multichain/chainsupport.go

@@ -242,12 +242,10 @@ func (cs *chainSupport) WriteBlock(block *cb.Block, committers []filter.Committe
 	for _, committer := range committers {
 		committer.Commit()
 	}
-
 	// Set the orderer-related metadata field
 	if encodedMetadataValue != nil {
 		block.Metadata.Metadata[cb.BlockMetadataIndex_ORDERER] = utils.MarshalOrPanic(&cb.Metadata{Value: encodedMetadataValue})
 	}
-
 	cs.addBlockSignature(block)
 	cs.addLastConfigSignature(block)
 

orderer/sbft/backend/backend.go

@@ -40,6 +40,7 @@ import (
 	"encoding/gob"
 
 	"github.com/golang/protobuf/proto"
+	"github.com/hyperledger/fabric/orderer/common/filter"
 	commonfilter "github.com/hyperledger/fabric/orderer/common/filter"
 	"github.com/hyperledger/fabric/orderer/multichain"
 	"github.com/hyperledger/fabric/orderer/sbft/connection"
@@ -246,6 +247,51 @@ func (b *Backend) AddSbftPeer(chainID string, support multichain.ConsenterSuppor
 	return s.New(b.GetMyId(), chainID, config, b)
 }
 
+func (b *Backend) Ordered(chainID string, req *s.Request) ([][]*s.Request, [][]filter.Committer, bool) {
+	// ([][]*cb.Envelope, [][]filter.Committer, bool) {
+	// If the message is a valid normal message and fills a batch, the batch, committers, true is returned
+	// If the message is a valid special message (like a config message) it terminates the current batch
+	// and returns the current batch and committers (if it is not empty), plus a second batch containing the special transaction and commiter, and true
+	env := &cb.Envelope{}
+	err := proto.Unmarshal(req.Payload, env)
+	if err != nil {
+		logger.Panicf("Request format error: %s", err)
+	}
+	envbatch, committers, accepted := b.supports[chainID].BlockCutter().Ordered(env)
+	if accepted {
+		if len(envbatch) == 1 {
+			rb1 := toRequestBatch(envbatch[0])
+			return [][]*s.Request{rb1}, committers, true
+		}
+		if len(envbatch) == 2 {
+			rb1 := toRequestBatch(envbatch[0])
+			rb2 := toRequestBatch(envbatch[1])
+			return [][]*s.Request{rb1, rb2}, committers, true
+		}
+
+		return nil, nil, true
+	}
+	return nil, nil, false
+}
+
+func (b *Backend) Cut(chainID string) ([]*s.Request, []filter.Committer) {
+	envbatch, committers := b.supports[chainID].BlockCutter().Cut()
+	return toRequestBatch(envbatch), committers
+}
+
+func toRequestBatch(envelopes []*cb.Envelope) []*s.Request {
+	rqs := make([]*s.Request, 0, len(envelopes))
+	for _, e := range envelopes {
+		requestbytes, err := proto.Marshal(e)
+		if err != nil {
+			logger.Panicf("Cannot marshal envelope: %s", err)
+		}
+		rq := &s.Request{Payload: requestbytes}
+		rqs = append(rqs, rq)
+	}
+	return rqs
+}
+
 // Consensus implements the SBFT consensus gRPC interface
 func (c *consensusConn) Consensus(_ *Handshake, srv Consensus_ConsensusServer) error {
 	pi := connection.GetPeerInfo(srv)
@@ -333,7 +379,7 @@ func (b *Backend) Timer(d time.Duration, tf func()) s.Canceller {
 }
 
 // Deliver writes a block
-func (b *Backend) Deliver(chainId string, batch *s.Batch) {
+func (b *Backend) Deliver(chainId string, batch *s.Batch, committers []commonfilter.Committer) {
 	blockContents := make([]*cb.Envelope, 0, len(batch.Payloads))
 	for _, p := range batch.Payloads {
 		envelope := &cb.Envelope{}
@@ -354,7 +400,6 @@ func (b *Backend) Deliver(chainId string, batch *s.Batch) {
 	metadata[signaturesIndex] = encodeSignatures(batch.Signatures)
 	block.Metadata.Metadata = metadata
 	b.lastBatches[chainId] = batch
-	committers := []commonfilter.Committer{}
 	b.supports[chainId].WriteBlock(block, committers, nil)
 }
 

orderer/sbft/backend/backend_test.go

@@ -25,6 +25,7 @@ import (
 	"testing"
 
 	"github.com/golang/protobuf/proto"
+	"github.com/hyperledger/fabric/orderer/common/filter"
 	"github.com/hyperledger/fabric/orderer/mocks/multichain"
 	mc "github.com/hyperledger/fabric/orderer/multichain"
 	"github.com/hyperledger/fabric/orderer/sbft/simplebft"
@@ -88,10 +89,10 @@ func TestLedgerReadWrite(t *testing.T) {
 	sgns[uint64(22)] = []byte("sgn22")
 	batch := simplebft.Batch{Header: header, Payloads: data, Signatures: sgns}
 
-	b.Deliver(testChainID1, &batch)
+	b.Deliver(testChainID1, &batch, []filter.Committer{})
 	batch1 := b.LastBatch(testChainID1)
 	batch2 := b.LastBatch(testChainID2)
-	b.Deliver(testChainID3, &batch)
+	b.Deliver(testChainID3, &batch, []filter.Committer{})
 	batch3 := b.LastBatch(testChainID3)
 
 	if !reflect.DeepEqual(batch, *batch1) {

orderer/sbft/simplebft/batch.go

@@ -51,15 +51,15 @@ func (s *SBFT) checkBatch(b *Batch, checkData bool, needSigs bool) (*BatchHeader
 	if checkData {
 		datahash := merkleHashData(b.Payloads)
 		if !reflect.DeepEqual(datahash, batchheader.DataHash) {
-			return nil, fmt.Errorf("malformed batch: invalid hash")
+			return nil, fmt.Errorf("malformed batches: invalid hash")
 		}
 	}
 
 	if batchheader.PrevHash == nil {
 		// TODO check against root hash, which should be part of constructor
 	} else if needSigs {
 		if len(b.Signatures) < s.oneCorrectQuorum() {
-			return nil, fmt.Errorf("insufficient number of signatures on batch: need %d, got %d", s.oneCorrectQuorum(), len(b.Signatures))
+			return nil, fmt.Errorf("insufficient number of signatures on batches: need %d, got %d", s.oneCorrectQuorum(), len(b.Signatures))
 		}
 	}
 

orderer/sbft/simplebft/checkpoint.go

@@ -97,7 +97,7 @@ func (s *SBFT) handleCheckpoint(c *Checkpoint, src uint64) {
 	// ignore null requests
 	batch := *s.cur.preprep.Batch
 	batch.Signatures = cpset
-	s.deliverBatch(&batch)
+	s.deliverBatch(&batch, s.cur.committers)
 	log.Infof("replica %d: request %s %s delivered on %d (completed common case)", s.id, s.cur.subject.Seq, hash2str(s.cur.subject.Digest), s.id)
 	s.maybeSendNextBatch()
 	s.processBacklog()

orderer/sbft/simplebft/connection.go

@@ -35,14 +35,14 @@ func (s *SBFT) Connection(replica uint64) {
 	}
 
 	// A reconnecting replica can play forward its blockchain to
-	// the batch listed in the hello message.  However, the
-	// currently in-flight batch will not be reflected in the
+	// the batches listed in the hello message.  However, the
+	// currently in-flight batches will not be reflected in the
 	// Hello message, nor will all messages be present to actually
-	// commit the in-flight batch at the reconnecting replica.
+	// commit the in-flight batches at the reconnecting replica.
 	//
 	// Therefore we also send the most recent (pre)prepare,
 	// commit, checkpoint so that the reconnecting replica can
-	// catch up on the in-flight batch.
+	// catch up on the in-flight batches.
 
 	batchheader, err := s.checkBatch(&batch, false, true)
 	if err != nil {
@@ -66,16 +66,21 @@ func (s *SBFT) Connection(replica uint64) {
 
 func (s *SBFT) handleHello(h *Hello, src uint64) {
 	bh, err := s.checkBatch(h.Batch, false, true)
-	log.Debugf("replica %d: got hello for batch %d from replica %d", s.id, bh.Seq, src)
+	log.Debugf("replica %d: got hello for batches %d from replica %d", s.id, bh.Seq, src)
 
 	if err != nil {
-		log.Warningf("replica %d: invalid hello batch from %d: %s", s.id, src, err)
+		log.Warningf("replica %d: invalid hello batches from %d: %s", s.id, src, err)
 		return
 	}
 
 	if s.sys.LastBatch(s.chainId).DecodeHeader().Seq < bh.Seq {
-		log.Debugf("replica %d: delivering batch %d after hello from replica %d", s.id, bh.Seq, src)
-		s.deliverBatch(h.Batch)
+		log.Debugf("replica %d: delivering batches %d after hello from replica %d", s.id, bh.Seq, src)
+		blockOK, committers := s.getCommittersFromBlockCutter(h.Batch)
+		if blockOK {
+			s.deliverBatch(h.Batch, committers)
+		} else {
+			log.Debugf("replica %d: we got a hello from %d with an erroneous block", s.id, src)
+		}
 	}
 
 	s.handleNewView(h.NewView, src)

orderer/sbft/simplebft/newview.go

@@ -45,7 +45,7 @@ func (s *SBFT) maybeSendNewView() {
 
 	var batch *Batch
 	if xset == nil {
-		// no need for batch, it is contained in the vset
+		// no need for batches, it is contained in the vset
 	} else if reflect.DeepEqual(s.cur.subject.Digest, xset.Digest) {
 		batch = s.cur.preprep.Batch
 	} else {
@@ -122,12 +122,12 @@ func (s *SBFT) handleNewView(nv *NewView, src uint64) {
 
 	if nv.Xset == nil {
 		if nv.Batch != nil {
-			log.Warningf("replica %d: invalid new view from %d: null request should come with null batch", s.id, src)
+			log.Warningf("replica %d: invalid new view from %d: null request should come with null batches", s.id, src)
 			s.sendViewChange()
 			return
 		}
 	} else if nv.Batch == nil || !bytes.Equal(nv.Batch.Hash(), nv.Xset.Digest) {
-		log.Warningf("replica %d: invalid new view from %d: batch head hash does not match xset: %x, %x, %v",
+		log.Warningf("replica %d: invalid new view from %d: batches head hash does not match xset: %x, %x, %v",
 			s.id, src, hash(nv.Batch.Header), nv.Xset.Digest, nv)
 		s.sendViewChange()
 		return
@@ -136,7 +136,7 @@ func (s *SBFT) handleNewView(nv *NewView, src uint64) {
 	if nv.Batch != nil {
 		_, err = s.checkBatch(nv.Batch, true, false)
 		if err != nil {
-			log.Warningf("replica %d: invalid new view from %d: invalid batch, %s",
+			log.Warningf("replica %d: invalid new view from %d: invalid batches, %s",
 				s.id, src, err)
 			s.sendViewChange()
 			return
@@ -146,19 +146,25 @@ func (s *SBFT) handleNewView(nv *NewView, src uint64) {
 	s.view = nv.View
 	s.discardBacklog(s.primaryID())
 
-	// maybe deliver previous batch
+	// maybe deliver previous batches
 	if s.sys.LastBatch(s.chainId).DecodeHeader().Seq < prevBatch.DecodeHeader().Seq {
 		if prevBatch.DecodeHeader().Seq == s.cur.subject.Seq.Seq {
 			// we just received a signature set for a request which we preprepared, but never delivered.
 			// check first if the locally preprepared request matches the signature set
 			if !reflect.DeepEqual(prevBatch.DecodeHeader().DataHash, s.cur.preprep.Batch.DecodeHeader().DataHash) {
-				log.Warningf("replica %d: [seq %d] request checkpointed in a previous view does not match locally preprepared one, delivering batch without payload", s.id, s.cur.subject.Seq.Seq)
+				log.Warningf("replica %d: [seq %d] request checkpointed in a previous view does not match locally preprepared one, delivering batches without payload", s.id, s.cur.subject.Seq.Seq)
 			} else {
-				log.Debugf("replica %d: [seq %d] request checkpointed in a previous view with matching preprepare, completing and delivering the batch with payload", s.id, s.cur.subject.Seq.Seq)
+				log.Debugf("replica %d: [seq %d] request checkpointed in a previous view with matching preprepare, completing and delivering the batches with payload", s.id, s.cur.subject.Seq.Seq)
 				prevBatch.Payloads = s.cur.preprep.Batch.Payloads
 			}
 		}
-		s.deliverBatch(prevBatch)
+		// TODO we should not do this here, as prevBatch was already delivered
+		blockOK, committers := s.getCommittersFromBlockCutter(prevBatch)
+		if !blockOK {
+			log.Panic("Replica %d: our last checkpointed batch is erroneous (block cutter).", s.id)
+		}
+		// TODO what should we do with the remaining?
+		s.deliverBatch(prevBatch, committers)
 	}
 
 	// after a new-view message, prepare to accept new requests.
@@ -174,10 +180,13 @@ func (s *SBFT) handleNewView(nv *NewView, src uint64) {
 			Seq:   &SeqView{Seq: nv.Batch.DecodeHeader().Seq, View: s.view},
 			Batch: nv.Batch,
 		}
+		blockOK, committers := s.getCommittersFromBlockCutter(nv.Batch)
+		if !blockOK {
+			log.Panic("Replica %d: new view %d batch erroneous (block cutter).", s.id, nv.View)
+		}
 
-		s.handleCheckedPreprepare(pp)
+		s.handleCheckedPreprepare(pp, committers)
 	} else {
-		log.Debugf("replica %d: %+v", s.id, s)
 		s.cancelViewChangeTimer()
 		s.maybeSendNextBatch()
 	}

orderer/sbft/simplebft/preprepare.go

@@ -19,9 +19,11 @@ package simplebft
 import (
 	"bytes"
 	"time"
+
+	"github.com/hyperledger/fabric/orderer/common/filter"
 )
 
-func (s *SBFT) sendPreprepare(batch []*Request) {
+func (s *SBFT) sendPreprepare(batch []*Request, committers []filter.Committer) {
 	seq := s.nextSeq()
 
 	data := make([][]byte, len(batch))
@@ -38,7 +40,8 @@ func (s *SBFT) sendPreprepare(batch []*Request) {
 
 	s.sys.Persist(s.chainId, preprepared, m)
 	s.broadcast(&Msg{&Msg_Preprepare{m}})
-	s.handleCheckedPreprepare(m)
+	log.Infof("replica %d: sendPreprepare", s.id)
+	s.handleCheckedPreprepare(m, committers)
 }
 
 func (s *SBFT) handlePreprepare(pp *Preprepare, src uint64) {
@@ -60,26 +63,27 @@ func (s *SBFT) handlePreprepare(pp *Preprepare, src uint64) {
 		return
 	}
 	if pp.Batch == nil {
-		log.Infof("replica %d: preprepare without batch", s.id)
+		log.Infof("replica %d: preprepare without batches", s.id)
 		return
 	}
 
 	batchheader, err := s.checkBatch(pp.Batch, true, false)
 	if err != nil || batchheader.Seq != pp.Seq.Seq {
-		log.Infof("replica %d: preprepare %v batch head inconsistent from %d: %s", s.id, pp.Seq, src, err)
+		log.Infof("replica %d: preprepare %v batches head inconsistent from %d: %s", s.id, pp.Seq, src, err)
 		return
 	}
 
 	prevhash := s.sys.LastBatch(s.chainId).Hash()
 	if !bytes.Equal(batchheader.PrevHash, prevhash) {
-		log.Infof("replica %d: preprepare batch prev hash does not match expected %s, got %s", s.id, hash2str(batchheader.PrevHash), hash2str(prevhash))
+		log.Infof("replica %d: preprepare batches prev hash does not match expected %s, got %s", s.id, hash2str(batchheader.PrevHash), hash2str(prevhash))
 		return
 	}
-
-	s.handleCheckedPreprepare(pp)
+	committers := s.getCommitters(pp)
+	log.Infof("replica %d: handlePrepare", s.id)
+	s.handleCheckedPreprepare(pp, committers)
 }
 
-func (s *SBFT) acceptPreprepare(pp *Preprepare) {
+func (s *SBFT) acceptPreprepare(pp *Preprepare, committers []filter.Committer) {
 	sub := Subject{Seq: pp.Seq, Digest: pp.Batch.Hash()}
 
 	log.Infof("replica %d: accepting preprepare for %v, %x", s.id, sub.Seq, sub.Digest)
@@ -92,11 +96,32 @@ func (s *SBFT) acceptPreprepare(pp *Preprepare) {
 		prep:       make(map[uint64]*Subject),
 		commit:     make(map[uint64]*Subject),
 		checkpoint: make(map[uint64]*Checkpoint),
+		committers: committers,
+	}
+}
+
+func (s *SBFT) getCommitters(pp *Preprepare) []filter.Committer {
+	// if we are the primary, we can be sure the block is OK
+	// and we also have the committers
+	// TODO what to do with the remaining ones???
+	// how to mantain the mapping between batches and committers?
+	var committers []filter.Committer
+
+	if !s.isPrimary() {
+		blockOK, allcommitters := s.getCommittersFromBlockCutter(pp.Batch)
+		if !blockOK {
+			log.Panicf("Replica %d found Byzantine block, Seq: %d View: %d", s.id, pp.Seq.Seq, pp.Seq.View)
+		}
+		committers = allcommitters
+	} else {
+		committers = s.primarycommitters[0]
+		s.primarycommitters = s.primarycommitters[1:]
 	}
+	return committers
 }
 
-func (s *SBFT) handleCheckedPreprepare(pp *Preprepare) {
-	s.acceptPreprepare(pp)
+func (s *SBFT) handleCheckedPreprepare(pp *Preprepare, committers []filter.Committer) {
+	s.acceptPreprepare(pp, committers)
 	if !s.isPrimary() {
 		s.sendPrepare()
 		s.processBacklog()
@@ -105,6 +130,34 @@ func (s *SBFT) handleCheckedPreprepare(pp *Preprepare) {
 	s.maybeSendCommit()
 }
 
+func (s *SBFT) getCommittersFromBlockCutter(reqBatch *Batch) (bool, []filter.Committer) {
+	reqs := make([]*Request, 0, len(reqBatch.Payloads))
+	for _, pl := range reqBatch.Payloads {
+		req := &Request{Payload: pl}
+		reqs = append(reqs, req)
+	}
+	batches := make([][]*Request, 0, 1)
+	comms := [][]filter.Committer{}
+	for _, r := range reqs {
+		b, c, accepted := s.sys.Ordered(s.chainId, r)
+		if !accepted {
+			return false, nil
+		}
+		batches = append(batches, b...)
+		comms = append(comms, c...)
+	}
+	if len(batches) > 1 || len(batches) != len(comms) {
+		return false, nil
+	}
+
+	if len(batches) == 0 {
+		_, committer := s.sys.Cut(s.chainId)
+		return true, committer
+	} else {
+		return true, comms[0]
+	}
+}
+
 ////////////////////////////////////////////////
 
 func (s *SBFT) requestTimeout() {