[FAB-1384]: Change ValidatedLedger APIs

In order to provide support for JIRA [FAB-1038], validated
ledger should be able to receive block to commit with list
of invalid transactions to execute MVCC validation as part
of the flow described in FAB-1038.

Change-Id: Ica9c48c036ecd97a17484e4954a85d1b7abccf2a
Signed-off-by: Artem Barger <bartem@il.ibm.com>
Signed-off-by: Yacov Manevich <yacovm@il.ibm.com>
Signed-off-by: Artem Barger <bartem@il.ibm.com>

core/chaincode/exectransaction_test.go

@@ -191,11 +191,8 @@ func endTxSimulation(chainID string, txsim ledger.TxSimulator, payload []byte, c
 			//create the block with 1 transaction
 			block := common.NewBlock(1, []byte{})
 			block.Data.Data = [][]byte{envBytes}
-			if _, _, err = lgr.RemoveInvalidTransactionsAndPrepare(block); err != nil {
-				return err
-			}
 			//commit the block
-			if err := lgr.Commit(); err != nil {
+			if err := lgr.Commit(block); err != nil {
 				return err
 			}
 		}

core/committer/committer_impl.go

@@ -48,10 +48,7 @@ func NewLedgerCommitter(ledger ledger.ValidatedLedger) *LedgerCommitter {
 
 // CommitBlock commits block to into the ledger
 func (lc *LedgerCommitter) CommitBlock(block *common.Block) error {
-	if _, _, err := lc.ledger.RemoveInvalidTransactionsAndPrepare(block); err != nil {
-		return err
-	}
-	if err := lc.ledger.Commit(); err != nil {
+	if err := lc.ledger.Commit(block); err != nil {
 		return err
 	}
 	return nil

core/endorser/endorser.go

@@ -334,11 +334,7 @@ func (e *Endorser) commitTxSimulation(proposal *pb.Proposal, chainID string, sig
 	block := common.NewBlock(1, []byte{})
 	block.Data.Data = [][]byte{txBytes}
 	block.Header.DataHash = block.Data.Hash()
-	if _, _, err = lgr.RemoveInvalidTransactionsAndPrepare(block); err != nil {
-		return err
-	}
-
-	if err = lgr.Commit(); err != nil {
+	if err = lgr.Commit(block); err != nil {
 		return err
 	}
 

core/ledger/kvledger/example/committer.go

@@ -20,7 +20,6 @@ import (
 	"github.com/hyperledger/fabric/core/ledger"
 
 	"github.com/hyperledger/fabric/protos/common"
-	pb "github.com/hyperledger/fabric/protos/peer"
 )
 
 // Committer a toy committer
@@ -34,17 +33,10 @@ func ConstructCommitter(ledger ledger.ValidatedLedger) *Committer {
 }
 
 // CommitBlock commits the block
-func (c *Committer) CommitBlock(rawBlock *common.Block) (*common.Block, []*pb.InvalidTransaction, error) {
-	var validBlock *common.Block
-	var invalidTxs []*pb.InvalidTransaction
-	var err error
+func (c *Committer) CommitBlock(rawBlock *common.Block) error {
 	logger.Debugf("Committer validating the block...")
-	if validBlock, invalidTxs, err = c.ledger.RemoveInvalidTransactionsAndPrepare(rawBlock); err != nil {
-		return nil, nil, err
+	if err := c.ledger.Commit(rawBlock); err != nil {
+		return err
 	}
-	logger.Debugf("Committer committing the block...")
-	if err = c.ledger.Commit(); err != nil {
-		return nil, nil, err
-	}
-	return validBlock, invalidTxs, err
+	return nil
 }

core/ledger/kvledger/example/main/example.go

@@ -24,8 +24,8 @@ import (
 	"github.com/hyperledger/fabric/core/ledger/kvledger"
 	"github.com/hyperledger/fabric/core/ledger/kvledger/example"
 	"github.com/hyperledger/fabric/core/ledger/testutil"
+	"github.com/hyperledger/fabric/core/ledger/util"
 	"github.com/hyperledger/fabric/protos/common"
-	pb "github.com/hyperledger/fabric/protos/peer"
 )
 
 const (
@@ -97,9 +97,9 @@ func initApp() {
 		accounts[3]: 100})
 	handleError(err, true)
 	rawBlock := consenter.ConstructBlock(tx)
-	finalBlock, invalidTx, err := committer.CommitBlock(rawBlock)
+	err = committer.CommitBlock(rawBlock)
 	handleError(err, true)
-	printBlocksInfo(rawBlock, finalBlock, invalidTx)
+	printBlocksInfo(rawBlock)
 }
 
 func transferFunds() {
@@ -112,9 +112,9 @@ func transferFunds() {
 	rawBlock := consenter.ConstructBlock(tx1, tx2)
 
 	// act as committing peer to commit the Raw Block
-	finalBlock, invalidTx, err := committer.CommitBlock(rawBlock)
+	err = committer.CommitBlock(rawBlock)
 	handleError(err, true)
-	printBlocksInfo(rawBlock, finalBlock, invalidTx)
+	printBlocksInfo(rawBlock)
 }
 
 func tryInvalidTransfer() {
@@ -128,14 +128,23 @@ func tryDoubleSpend() {
 	tx2, err := app.TransferFunds("account1", "account4", 50)
 	handleError(err, true)
 	rawBlock := consenter.ConstructBlock(tx1, tx2)
-	finalBlock, invalidTx, err := committer.CommitBlock(rawBlock)
+	err = committer.CommitBlock(rawBlock)
 	handleError(err, true)
-	printBlocksInfo(rawBlock, finalBlock, invalidTx)
+	printBlocksInfo(rawBlock)
 }
 
-func printBlocksInfo(rawBlock *common.Block, finalBlock *common.Block, invalidTxs []*pb.InvalidTransaction) {
+func printBlocksInfo(block *common.Block) {
+	// Read invalid transactions filter
+	txsFltr := util.NewFilterBitArrayFromBytes(block.Metadata.Metadata[common.BlockMetadataIndex_TRANSACTIONS_FILTER])
+	numOfInvalid := 0
+	// Count how many transaction indeed invalid
+	for i := 0; i < len(block.Data.Data); i++ {
+		if txsFltr.IsSet(uint(i)) {
+			numOfInvalid++
+		}
+	}
 	fmt.Printf("Num txs in rawBlock = [%d], num invalidTxs = [%d]\n",
-		len(rawBlock.Data.Data), len(invalidTxs))
+		len(block.Data.Data), numOfInvalid)
 }
 
 func printBalances() {

core/ledger/kvledger/kv_ledger.go

@@ -27,12 +27,12 @@ import (
 	"github.com/hyperledger/fabric/core/ledger/history"
 	"github.com/hyperledger/fabric/core/ledger/kvledger/txmgmt/couchdbtxmgmt"
 	"github.com/hyperledger/fabric/core/ledger/kvledger/txmgmt/statedb/stateleveldb"
+	"github.com/hyperledger/fabric/core/ledger/kvledger/txmgmt/txmgr"
 	"github.com/hyperledger/fabric/core/ledger/kvledger/txmgmt/txmgr/lockbasedtxmgr"
 	"github.com/hyperledger/fabric/core/ledger/ledgerconfig"
 
 	logging "github.com/op/go-logging"
 
-	"github.com/hyperledger/fabric/core/ledger/kvledger/txmgmt/txmgr"
 	"github.com/hyperledger/fabric/protos/common"
 	pb "github.com/hyperledger/fabric/protos/peer"
 )
@@ -60,10 +60,9 @@ func NewConf(filesystemPath string, maxBlockfileSize int) *Conf {
 // KVLedger provides an implementation of `ledger.ValidatedLedger`.
 // This implementation provides a key-value based data model
 type KVLedger struct {
-	blockStore           blkstorage.BlockStore
-	txtmgmt              txmgr.TxMgr
-	historymgmt          history.HistMgr
-	pendingBlockToCommit *common.Block
+	blockStore  blkstorage.BlockStore
+	txtmgmt     txmgr.TxMgr
+	historymgmt history.HistMgr
 }
 
 // NewKVLedger constructs new `KVLedger`
@@ -114,10 +113,13 @@ func NewKVLedger(conf *Conf) (*KVLedger, error) {
 			couchDBDef.Username, //enter couchDB id here
 			couchDBDef.Password) //enter couchDB pw here
 	}
-	l := &KVLedger{blockStore, txmgmt, historymgmt, nil}
+
+	l := &KVLedger{blockStore, txmgmt, historymgmt}
+
 	if err := recoverStateDB(l); err != nil {
 		panic(fmt.Errorf(`Error during state DB recovery:%s`, err))
 	}
+
 	return l, nil
 }
 
@@ -135,7 +137,7 @@ func recoverStateDB(l *KVLedger) error {
 	if savepointValue, err = l.txtmgmt.GetBlockNumFromSavepoint(); err != nil {
 		return err
 	}
-	logger.Debugf("savepointValue=%d, info.Height=%d", savepointValue, info.Height)
+
 	//Checking whether the savepointValue is in sync with block storage height
 	if savepointValue == info.Height {
 		return nil
@@ -145,19 +147,20 @@ func recoverStateDB(l *KVLedger) error {
 
 	//Compute updateSet for each missing savepoint and commit to state DB
 	for blockNumber := savepointValue + 1; blockNumber <= info.Height; blockNumber++ {
-		if l.pendingBlockToCommit, err = l.GetBlockByNumber(blockNumber); err != nil {
+		var block *common.Block
+		if block, err = l.GetBlockByNumber(blockNumber); err != nil {
 			return err
 		}
 		logger.Debugf("Constructing updateSet for the block %d", blockNumber)
-		if _, _, err = l.txtmgmt.ValidateAndPrepare(l.pendingBlockToCommit, false); err != nil {
+		if err = l.txtmgmt.ValidateAndPrepare(block, false); err != nil {
 			return err
 		}
 		logger.Debugf("Committing block %d to state database", blockNumber)
 		if err = l.txtmgmt.Commit(); err != nil {
 			return err
 		}
 	}
-	l.pendingBlockToCommit = nil
+
 	return nil
 }
 
@@ -208,54 +211,38 @@ func (l *KVLedger) NewQueryExecutor() (ledger.QueryExecutor, error) {
 	return l.txtmgmt.NewQueryExecutor()
 }
 
-// RemoveInvalidTransactionsAndPrepare validates all the transactions in the given block
-// and returns a block that contains only valid transactions and a list of transactions that are invalid
-func (l *KVLedger) RemoveInvalidTransactionsAndPrepare(block *common.Block) (*common.Block, []*pb.InvalidTransaction, error) {
-	var validBlock *common.Block
-	var invalidTxs []*pb.InvalidTransaction
+// Commit commits the valid block (returned in the method RemoveInvalidTransactionsAndPrepare) and related state changes
+func (l *KVLedger) Commit(block *common.Block) error {
 	var err error
-	validBlock, invalidTxs, err = l.txtmgmt.ValidateAndPrepare(block, true)
-	if err == nil {
-		l.pendingBlockToCommit = validBlock
-	}
-	return validBlock, invalidTxs, err
-}
 
-// Commit commits the valid block (returned in the method RemoveInvalidTransactionsAndPrepare) and related state changes
-func (l *KVLedger) Commit() error {
-	if l.pendingBlockToCommit == nil {
-		panic(fmt.Errorf(`Nothing to commit. RemoveInvalidTransactionsAndPrepare() method should have been called and should not have thrown error`))
+	logger.Debugf("Validating block")
+	err = l.txtmgmt.ValidateAndPrepare(block, true)
+	if err != nil {
+		return err
 	}
 
 	logger.Debugf("Committing block to storage")
-	if err := l.blockStore.AddBlock(l.pendingBlockToCommit); err != nil {
+	if err = l.blockStore.AddBlock(block); err != nil {
 		return err
 	}
 
 	logger.Debugf("Committing block to state database")
-	if err := l.txtmgmt.Commit(); err != nil {
+	if err = l.txtmgmt.Commit(); err != nil {
 		panic(fmt.Errorf(`Error during commit to txmgr:%s`, err))
 	}
 
 	//TODO future will want to run async with state db writes.  History needs to wait for chain (FSBlock) to write but not the state db
 	logger.Debugf("===HISTORYDB=== Commit() will write to hisotry if enabled else will be by-passed if not enabled: vledgerconfig.IsHistoryDBEnabled(): %v\n", ledgerconfig.IsHistoryDBEnabled())
 	if ledgerconfig.IsHistoryDBEnabled() == true {
 		logger.Debugf("Committing transactions to history database")
-		if err := l.historymgmt.Commit(l.pendingBlockToCommit); err != nil {
+		if err := l.historymgmt.Commit(block); err != nil {
 			panic(fmt.Errorf(`Error during commit to txthistory:%s`, err))
 		}
 	}
 
-	l.pendingBlockToCommit = nil
 	return nil
 }
 
-// Rollback rollbacks the changes caused by the last invocation to method `RemoveInvalidTransactionsAndPrepare`
-func (l *KVLedger) Rollback() {
-	l.txtmgmt.Rollback()
-	l.pendingBlockToCommit = nil
-}
-
 // Close closes `KVLedger`
 func (l *KVLedger) Close() {
 	l.blockStore.Shutdown()

core/ledger/kvledger/kv_ledger_test.go

@@ -22,6 +22,7 @@ import (
 	"github.com/hyperledger/fabric/core/ledger/ledgerconfig"
 	"github.com/hyperledger/fabric/core/ledger/testutil"
 	pb "github.com/hyperledger/fabric/protos/peer"
+	"github.com/stretchr/testify/assert"
 )
 
 func TestKVLedgerBlockStorage(t *testing.T) {
@@ -41,8 +42,7 @@ func TestKVLedgerBlockStorage(t *testing.T) {
 	simRes, _ := simulator.GetTxSimulationResults()
 	bg := testutil.NewBlockGenerator(t)
 	block1 := bg.NextBlock([][]byte{simRes}, false)
-	ledger.RemoveInvalidTransactionsAndPrepare(block1)
-	ledger.Commit()
+	ledger.Commit(block1)
 
 	bcInfo, _ = ledger.GetBlockchainInfo()
 	block1Hash := block1.Header.Hash()
@@ -56,8 +56,7 @@ func TestKVLedgerBlockStorage(t *testing.T) {
 	simulator.Done()
 	simRes, _ = simulator.GetTxSimulationResults()
 	block2 := bg.NextBlock([][]byte{simRes}, false)
-	ledger.RemoveInvalidTransactionsAndPrepare(block2)
-	ledger.Commit()
+	ledger.Commit(block2)
 
 	bcInfo, _ = ledger.GetBlockchainInfo()
 	block2Hash := block2.Header.Hash()
@@ -99,9 +98,7 @@ func TestKVLedgerStateDBRecovery(t *testing.T) {
 	bg := testutil.NewBlockGenerator(t)
 	block1 := bg.NextBlock([][]byte{simRes}, false)
 	//performing validation of read and write set to find valid transactions
-	ledger.RemoveInvalidTransactionsAndPrepare(block1)
-	//writing the validated block to block storage and committing the transaction to state DB
-	ledger.Commit()
+	ledger.Commit(block1)
 
 	bcInfo, _ = ledger.GetBlockchainInfo()
 	block1Hash := block1.Header.Hash()
@@ -119,10 +116,11 @@ func TestKVLedgerStateDBRecovery(t *testing.T) {
 	//generating a block based on the simulation result
 	block2 := bg.NextBlock([][]byte{simRes}, false)
 	//performing validation of read and write set to find valid transactions
-	ledger.RemoveInvalidTransactionsAndPrepare(block2)
+	ledger.txtmgmt.ValidateAndPrepare(block2, true)
 	//writing the validated block to block storage but not committing the transaction to state DB
-	ledger.blockStore.AddBlock(ledger.pendingBlockToCommit)
+	err := ledger.blockStore.AddBlock(block2)
 	//assume that peer fails here before committing the transaction
+	assert.NoError(t, err)
 
 	bcInfo, _ = ledger.GetBlockchainInfo()
 	block2Hash := block2.Header.Hash()
@@ -186,8 +184,7 @@ func TestLedgerWithCouchDbEnabledWithBinaryAndJSONData(t *testing.T) {
 	bg := testutil.NewBlockGenerator(t)
 	block1 := bg.NextBlock([][]byte{simRes}, false)
 
-	ledger.RemoveInvalidTransactionsAndPrepare(block1)
-	ledger.Commit()
+	ledger.Commit(block1)
 
 	bcInfo, _ = ledger.GetBlockchainInfo()
 	block1Hash := block1.Header.Hash()
@@ -214,8 +211,7 @@ func TestLedgerWithCouchDbEnabledWithBinaryAndJSONData(t *testing.T) {
 	simulationResults = append(simulationResults, simRes2)
 
 	block2 := bg.NextBlock(simulationResults, false)
-	ledger.RemoveInvalidTransactionsAndPrepare(block2)
-	ledger.Commit()
+	ledger.Commit(block2)
 
 	bcInfo, _ = ledger.GetBlockchainInfo()
 	block2Hash := block2.Header.Hash()

core/ledger/kvledger/marble_example/main/marble_example.go

@@ -24,8 +24,8 @@ import (
 	"github.com/hyperledger/fabric/core/ledger/kvledger"
 	"github.com/hyperledger/fabric/core/ledger/kvledger/example"
 	"github.com/hyperledger/fabric/core/ledger/testutil"
+	"github.com/hyperledger/fabric/core/ledger/util"
 	"github.com/hyperledger/fabric/protos/common"
-	pb "github.com/hyperledger/fabric/protos/peer"
 	logging "github.com/op/go-logging"
 )
 
@@ -81,24 +81,33 @@ func initApp() {
 	tx, err := marbleApp.CreateMarble(marble)
 	handleError(err, true)
 	rawBlock := consenter.ConstructBlock(tx)
-	finalBlock, invalidTx, err := committer.CommitBlock(rawBlock)
+	err = committer.CommitBlock(rawBlock)
 	handleError(err, true)
-	printBlocksInfo(rawBlock, finalBlock, invalidTx)
+	printBlocksInfo(rawBlock)
 }
 
 func transferMarble() {
 	logger.Debugf("===COUCHDB=== Marble Example transferMarble()")
 	tx1, err := marbleApp.TransferMarble([]string{"marble1", "jerry"})
 	handleError(err, true)
 	rawBlock := consenter.ConstructBlock(tx1)
-	finalBlock, invalidTx, err := committer.CommitBlock(rawBlock)
+	err = committer.CommitBlock(rawBlock)
 	handleError(err, true)
-	printBlocksInfo(rawBlock, finalBlock, invalidTx)
+	printBlocksInfo(rawBlock)
 }
 
-func printBlocksInfo(rawBlock *common.Block, finalBlock *common.Block, invalidTxs []*pb.InvalidTransaction) {
+func printBlocksInfo(block *common.Block) {
+	// Read invalid transactions filter
+	txsFltr := util.NewFilterBitArrayFromBytes(block.Metadata.Metadata[common.BlockMetadataIndex_TRANSACTIONS_FILTER])
+	numOfInvalid := 0
+	// Count how many transaction indeed invalid
+	for i := 0; i < len(block.Data.Data); i++ {
+		if txsFltr.IsSet(uint(i)) {
+			numOfInvalid++
+		}
+	}
 	fmt.Printf("Num txs in rawBlock = [%d], num invalidTxs = [%d]\n",
-		len(rawBlock.Data.Data), len(invalidTxs))
+		len(block.Data.Data), numOfInvalid)
 }
 
 func handleError(err error, quit bool) {