rollback: skip uncommittedPvtData of oldBlocks

As we are storing pvtData of both valid and invalid tx
in the pvtdataStore, we do store missing info for both
valid and invalid tx. When the reconciler fetches the
missing pvtData, it passes to the ledger coordinator.
As the pvtData contains data of both valid and invalid tx,
we need to pass both to pvtdataStore and only valid tx's
pvtData to the txmgr.

Similarly, during failure and recovery, we fetch the pvtData
of last updated old blocks and filter uncommitted tx pvtData
while syncing the stateDB

FAB-14861 #done

Change-Id: I858939beba2c702df7f197aa6a6c555f4655a83b
Signed-off-by: senthil <cendhu@gmail.com>
(cherry picked from commit df3616a)

Author: cendhu

core/ledger/kvledger/hashcheck_pvtdata.go

@@ -17,9 +17,9 @@ import (
 	"github.com/hyperledger/fabric/protoutil"
 )
 
-// ConstructValidAndInvalidPvtData computes the valid pvt data and hash mismatch list
+// constructValidAndInvalidPvtData computes the valid pvt data and hash mismatch list
 // from a received pvt data list of old blocks.
-func ConstructValidAndInvalidPvtData(blocksPvtData []*ledger.BlockPvtData, blockStore *ledgerstorage.Store) (
+func constructValidAndInvalidPvtData(blocksPvtData []*ledger.BlockPvtData, blockStore *ledgerstorage.Store) (
 	map[uint64][]*ledger.TxPvtData, []*ledger.PvtdataHashMismatch, error,
 ) {
 	// for each block, for each transaction, retrieve the txEnvelope to

core/ledger/kvledger/hashcheck_pvtdata_test.go

@@ -104,7 +104,7 @@ func TestConstructValidInvalidBlocksPvtData(t *testing.T) {
 		},
 	}
 
-	blocksValidPvtData, hashMismatched, err := ConstructValidAndInvalidPvtData(blocksPvtData, lg.(*kvLedger).blockStore)
+	blocksValidPvtData, hashMismatched, err := constructValidAndInvalidPvtData(blocksPvtData, lg.(*kvLedger).blockStore)
 	assert.NoError(t, err)
 	assert.Equal(t, len(expectedValidBlocksPvtData), len(blocksValidPvtData))
 	assert.ElementsMatch(t, expectedValidBlocksPvtData[1], blocksValidPvtData[1])
@@ -133,7 +133,7 @@ func TestConstructValidInvalidBlocksPvtData(t *testing.T) {
 		},
 	}
 
-	blocksValidPvtData, hashMismatches, err := ConstructValidAndInvalidPvtData(blocksPvtData, lg.(*kvLedger).blockStore)
+	blocksValidPvtData, hashMismatches, err := constructValidAndInvalidPvtData(blocksPvtData, lg.(*kvLedger).blockStore)
 	assert.NoError(t, err)
 	assert.Len(t, blocksValidPvtData, 0)
 

core/ledger/kvledger/kv_ledger.go

@@ -24,6 +24,7 @@ import (
 	"github.com/hyperledger/fabric/core/ledger/kvledger/txmgmt/txmgr/lockbasedtxmgr"
 	"github.com/hyperledger/fabric/core/ledger/ledgerstorage"
 	"github.com/hyperledger/fabric/core/ledger/pvtdatapolicy"
+	lutil "github.com/hyperledger/fabric/core/ledger/util"
 	"github.com/hyperledger/fabric/protos/common"
 	"github.com/hyperledger/fabric/protos/peer"
 	"github.com/hyperledger/fabric/protoutil"
@@ -245,22 +246,38 @@ func (l *kvLedger) syncStateDBWithPvtdatastore() error {
 	// of state update should not lie with the source (i.e., pvtdatastorage). A potential fix
 	// is mentioned in FAB-12731
 
-	// TODO: GetLastUpdatedOldBlocksPvtData would give the pvtData of both valid and
-	// invalid transactions. We would need only the valid transaction's pvtData. FAB-14861
 	blocksPvtData, err := l.blockStore.GetLastUpdatedOldBlocksPvtData()
 	if err != nil {
 		return err
 	}
-	if err := l.txtmgmt.RemoveStaleAndCommitPvtDataOfOldBlocks(blocksPvtData); err != nil {
-		return err
+
+	if l.blockStore.IsPvtStoreAheadOfBlockStore() {
+		if err := l.filterYetToCommitBlocks(blocksPvtData); err != nil {
+			return err
+		}
 	}
-	if err := l.blockStore.ResetLastUpdatedOldBlocksList(); err != nil {
+
+	if err = l.applyValidTxPvtDataOfOldBlocks(blocksPvtData); err != nil {
 		return err
 	}
 
 	return nil
 }
 
+func (l *kvLedger) filterYetToCommitBlocks(blocksPvtData map[uint64][]*ledger.TxPvtData) error {
+	info, err := l.blockStore.GetBlockchainInfo()
+	if err != nil {
+		return err
+	}
+	for blkNum := range blocksPvtData {
+		if blkNum > info.Height-1 {
+			logger.Infof("found pvtdata associated with yet to be committed block [%d]", blkNum)
+			delete(blocksPvtData, blkNum)
+		}
+	}
+	return nil
+}
+
 //recommitLostBlocks retrieves blocks in specified range and commit the write set to either
 //state DB or history DB or both
 func (l *kvLedger) recommitLostBlocks(firstBlockNum uint64, lastBlockNum uint64, recoverables ...recoverable) error {
@@ -537,35 +554,44 @@ func (l *kvLedger) GetConfigHistoryRetriever() (ledger.ConfigHistoryRetriever, e
 func (l *kvLedger) CommitPvtDataOfOldBlocks(pvtData []*ledger.BlockPvtData) ([]*ledger.PvtdataHashMismatch, error) {
 	logger.Debugf("[%s:] Comparing pvtData of [%d] old blocks against the hashes in transaction's rwset to find valid and invalid data",
 		l.ledgerID, len(pvtData))
-	// TODO: validPvtData consists of pvtData that is matching the hashed pvtData present in the
-	// block. Now, we store pvtData of both committed and failed transactions. Hence, we need to
-	// separate validPvtData into pvtData associated with committed tx (i.e., valid tx) and
-	// failed tx (i.e,. invalid tx). FAB-14861
-	validPvtData, hashMismatches, err := ConstructValidAndInvalidPvtData(pvtData, l.blockStore)
+
+	hashVerifiedPvtData, hashMismatches, err := constructValidAndInvalidPvtData(pvtData, l.blockStore)
 	if err != nil {
 		return nil, err
 	}
 
 	logger.Debugf("[%s:] Committing pvtData of [%d] old blocks to the pvtdatastore", l.ledgerID, len(pvtData))
-	err = l.blockStore.CommitPvtDataOfOldBlocks(validPvtData)
+	err = l.blockStore.CommitPvtDataOfOldBlocks(hashVerifiedPvtData)
 	if err != nil {
 		return nil, err
 	}
 
-	logger.Debugf("[%s:] Committing pvtData of [%d] old blocks to the stateDB", l.ledgerID, len(pvtData))
-	// TODO: As the validPvtData will be consisting of both committed and failed tx's pvtData,
-	// we need to pass only the pvtData of committed tx. FAB-14861
-	err = l.txtmgmt.RemoveStaleAndCommitPvtDataOfOldBlocks(validPvtData)
+	err = l.applyValidTxPvtDataOfOldBlocks(hashVerifiedPvtData)
 	if err != nil {
 		return nil, err
 	}
 
+	return hashMismatches, nil
+}
+
+func (l *kvLedger) applyValidTxPvtDataOfOldBlocks(hashVerifiedPvtData map[uint64][]*ledger.TxPvtData) error {
+	logger.Debugf("[%s:] Filtering pvtData of invalidation transactions", l.ledgerID)
+	committedPvtData, err := filterPvtDataOfInvalidTx(hashVerifiedPvtData, l.blockStore)
+	if err != nil {
+		return err
+	}
+
+	logger.Debugf("[%s:] Committing pvtData of [%d] old blocks to the stateDB", l.ledgerID, len(hashVerifiedPvtData))
+	err = l.txtmgmt.RemoveStaleAndCommitPvtDataOfOldBlocks(committedPvtData)
+	if err != nil {
+		return err
+	}
+
 	logger.Debugf("[%s:] Clearing the bookkeeping information from pvtdatastore", l.ledgerID)
 	if err := l.blockStore.ResetLastUpdatedOldBlocksList(); err != nil {
-		return nil, err
+		return err
 	}
-
-	return hashMismatches, nil
+	return nil
 }
 
 func (l *kvLedger) GetMissingPvtDataTracker() (ledger.MissingPvtDataTracker, error) {
@@ -630,3 +656,27 @@ func (a *ccEventListenerAdaptor) HandleChaincodeDeploy(chaincodeDefinition *ledg
 func (a *ccEventListenerAdaptor) ChaincodeDeployDone(succeeded bool) {
 	a.legacyEventListener.ChaincodeDeployDone(succeeded)
 }
+
+func filterPvtDataOfInvalidTx(hashVerifiedPvtData map[uint64][]*ledger.TxPvtData, blockStore *ledgerstorage.Store) (map[uint64][]*ledger.TxPvtData, error) {
+	committedPvtData := make(map[uint64][]*ledger.TxPvtData)
+	for blkNum, txsPvtData := range hashVerifiedPvtData {
+
+		// TODO: Instead of retrieving the whole block, we need to retireve only
+		// the TxValidationFlags from the block metadata. For that, we would need
+		// to add a new index for the block metadata. FAB- FAB-15808
+		block, err := blockStore.RetrieveBlockByNumber(blkNum)
+		if err != nil {
+			return nil, err
+		}
+		blockValidationFlags := lutil.TxValidationFlags(block.Metadata.Metadata[common.BlockMetadataIndex_TRANSACTIONS_FILTER])
+
+		var blksPvtData []*ledger.TxPvtData
+		for _, pvtData := range txsPvtData {
+			if blockValidationFlags.IsValid(int(pvtData.SeqInBlock)) {
+				blksPvtData = append(blksPvtData, pvtData)
+			}
+		}
+		committedPvtData[blkNum] = blksPvtData
+	}
+	return committedPvtData, nil
+}

core/ledger/kvledger/kv_ledger_test.go

@@ -457,11 +457,9 @@ func testSyncStateDBWithPvtdatastore(t *testing.T) {
 		map[string]uint64{"coll": 0},
 		conf,
 	)
-	defer provider.Close()
 	testLedgerid := "testLedger"
 	bg, gb := testutil.NewBlockGenerator(t, testLedgerid, false)
 	ledger, _ := provider.Create(gb)
-	defer ledger.Close()
 
 	// create and commit two data block (both with missing pvtdata)
 	blockAndPvtdata1, pvtdata1 := prepareNextBlockWithMissingPvtDataForTest(t, ledger, bg, "SimulateForBlk1",
@@ -476,28 +474,75 @@ func testSyncStateDBWithPvtdatastore(t *testing.T) {
 
 	assert.NoError(t, ledger.CommitWithPvtData(blockAndPvtdata2, &lgr.CommitOptions{}))
 
+	blockAndPvtdata3, pvtdata3 := prepareNextBlockWithMissingPvtDataForTest(t, ledger, bg, "SimulateForBlk3",
+		map[string]string{"key1": "value1.3", "key2": "value2.3", "key3": "value3.3"},
+		map[string]string{"key1": "pvtValue1.3", "key2": "pvtValue2.3", "key3": "pvtValue3.3"})
+
+	blockAndPvtdata3.Block.Metadata.Metadata[common.BlockMetadataIndex_TRANSACTIONS_FILTER][0] = byte(peer.TxValidationCode_PHANTOM_READ_CONFLICT)
+	assert.NoError(t, ledger.CommitWithPvtData(blockAndPvtdata3, &lgr.CommitOptions{}))
+
+	blockAndPvtdata4, pvtdata4 := prepareNextBlockWithMissingPvtDataForTest(t, ledger, bg, "SimulateForBlk4",
+		map[string]string{"key4": "value4"},
+		map[string]string{"key4": "pvtValue4"})
+
+	assert.NoError(t, ledger.CommitWithPvtData(blockAndPvtdata4, &lgr.CommitOptions{}))
+
+	blockAndPvtdata5, pvtdata5 := prepareNextBlockWithMissingPvtDataForTest(t, ledger, bg, "SimulateForBlk5",
+		map[string]string{"key5": "value4"},
+		map[string]string{"key5": "pvtValue5"})
+
+	assert.NoError(t, ledger.CommitWithPvtData(blockAndPvtdata5, &lgr.CommitOptions{}))
+
 	txSim, err := ledger.NewTxSimulator("test")
 	assert.NoError(t, err)
 	value, err := txSim.GetPrivateData("ns", "coll", "key1")
 	_, ok := err.(*txmgr.ErrPvtdataNotAvailable)
 	assert.True(t, ok)
 	assert.Nil(t, value)
 
+	value, err = txSim.GetPrivateData("ns", "coll", "key4")
+	_, ok = err.(*txmgr.ErrPvtdataNotAvailable)
+	assert.True(t, ok)
+	assert.Nil(t, value)
+
+	value, err = txSim.GetPrivateData("ns", "coll", "key5")
+	_, ok = err.(*txmgr.ErrPvtdataNotAvailable)
+	assert.True(t, ok)
+	assert.Nil(t, value)
+
 	blocksPvtData := map[uint64][]*lgr.TxPvtData{
 		1: {
 			pvtdata1,
 		},
 		2: {
 			pvtdata2,
 		},
+		3: {
+			pvtdata3,
+		},
+		4: {
+			pvtdata4,
+		},
+		5: {
+			pvtdata5,
+		},
 	}
 
 	assert.NoError(t, ledger.(*kvLedger).blockStore.CommitPvtDataOfOldBlocks(blocksPvtData))
+	// ensure that the pvtdata of the invalid transaction in block 3 got stored in the
+	// pvtdataStore
+	pvtdata, _ := ledger.GetPvtDataByNum(3, nil)
+	assert.NotNil(t, pvtdata)
+	assert.Equal(t, 1, len(pvtdata))
+	assert.True(t, pvtdata[0].Has("ns", "coll"))
 
 	// Now, assume that peer fails here before committing the pvtData to stateDB
 	ledger.Close()
 	provider.Close()
 
+	// Rollback to block 3
+	RollbackKVLedger(conf.RootFSPath, testLedgerid, 3)
+
 	// Here the peer comes online and calls NewKVLedger to get a handler for the ledger
 	// StateDB and HistoryDB should be recovered before returning from NewKVLedger call
 	provider = testutilNewProviderWithCollectionConfig(
@@ -506,13 +551,44 @@ func testSyncStateDBWithPvtdatastore(t *testing.T) {
 		map[string]uint64{"coll": 0},
 		conf,
 	)
-	ledger, _ = provider.Open(testLedgerid)
+	defer provider.Close()
+
+	ledger, err = provider.Open(testLedgerid)
+	assert.NoError(t, err)
+	defer ledger.Close()
 
 	txSim, err = ledger.NewTxSimulator("test")
 	assert.NoError(t, err)
 	value, err = txSim.GetPrivateData("ns", "coll", "key1")
 	assert.NoError(t, err)
+	// the value should match the string provided in the block 2 (i.e., pvtValue1.2)
+	// rather than block 3 (i.e., pvtValue1.3)
 	assert.Equal(t, value, []byte("pvtValue1.2"))
+
+	// block 4 is not committed yet
+	value, err = txSim.GetPrivateData("ns", "coll", "key4")
+	assert.Nil(t, err)
+	assert.Nil(t, value)
+
+	// block 5 is not committed yet
+	value, err = txSim.GetPrivateData("ns", "coll", "key5")
+	assert.Nil(t, err)
+	assert.Nil(t, value)
+	txSim.Done()
+
+	// recommit block 4 & 5
+	assert.NoError(t, ledger.CommitWithPvtData(blockAndPvtdata4, &lgr.CommitOptions{FetchPvtDataFromLedger: true}))
+	assert.NoError(t, ledger.CommitWithPvtData(blockAndPvtdata5, &lgr.CommitOptions{FetchPvtDataFromLedger: true}))
+
+	txSim, err = ledger.NewTxSimulator("test")
+	assert.NoError(t, err)
+	value, err = txSim.GetPrivateData("ns", "coll", "key4")
+	assert.NoError(t, err)
+	assert.Equal(t, value, []byte("pvtValue4"))
+
+	value, err = txSim.GetPrivateData("ns", "coll", "key5")
+	assert.NoError(t, err)
+	assert.Equal(t, value, []byte("pvtValue5"))
 }
 
 func TestLedgerWithCouchDbEnabledWithBinaryAndJSONData(t *testing.T) {