recon: construct valid/invalid set from pvt data

This CR computes the valid pvt data and hash mismatch list from a received pvt data list of old blocks. To do that, we compare the hash of passed pvt data with the hash present in the block. FAB-11388 #done Change-Id: I6cf9f7c0cdae50451121d148c54d3eaa67dac268 Signed-off-by: senthil <[email protected]> Signed-off-by: manish <[email protected]>
hyperledger · Nov 15, 2018 · 1afa0f8 · 1afa0f8
1 parent e7a9624
commit 1afa0f8
Show file tree

Hide file tree

Showing 10 changed files with 479 additions and 12 deletions.
diff --git a/core/ledger/kvledger/hashcheck_pvtdata.go b/core/ledger/kvledger/hashcheck_pvtdata.go
@@ -0,0 +1,144 @@
+/*
+Copyright IBM Corp. All Rights Reserved.
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package kvledger
+
+import (
+	"bytes"
+
+	"github.com/hyperledger/fabric/common/util"
+	"github.com/hyperledger/fabric/core/ledger"
+	"github.com/hyperledger/fabric/core/ledger/kvledger/txmgmt/rwsetutil"
+	"github.com/hyperledger/fabric/core/ledger/ledgerstorage"
+	"github.com/hyperledger/fabric/protos/ledger/rwset"
+	"github.com/hyperledger/fabric/protos/utils"
+)
+
+// 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) (
+	map[uint64][]*ledger.TxPvtData, []*ledger.PvtdataHashMismatch, error,
+) {
+	// for each block, for each transaction, retrieve the txEnvelope to
+	// compare the hash of pvtRwSet in the block and the hash of the received
+	// txPvtData. On a mismatch, add an entry to hashMismatch list.
+	// On a match, add the pvtData to the validPvtData list
+	validPvtData := make(map[uint64][]*ledger.TxPvtData)
+	var invalidPvtData []*ledger.PvtdataHashMismatch
+
+	for _, blockPvtData := range blocksPvtData {
+		validData, invalidData, err := findValidAndInvalidBlockPvtData(blockPvtData, blockStore)
+		if err != nil {
+			return nil, nil, err
+		}
+		if len(validData) > 0 {
+			validPvtData[blockPvtData.BlockNum] = validData
+		}
+		invalidPvtData = append(invalidPvtData, invalidData...)
+	} // for each block's pvtData
+	return validPvtData, invalidPvtData, nil
+}
+
+func findValidAndInvalidBlockPvtData(blockPvtData *ledger.BlockPvtData, blockStore *ledgerstorage.Store) (
+	[]*ledger.TxPvtData, []*ledger.PvtdataHashMismatch, error,
+) {
+	var validPvtData []*ledger.TxPvtData
+	var invalidPvtData []*ledger.PvtdataHashMismatch
+	for _, txPvtData := range blockPvtData.WriteSets {
+		// (1) retrieve the txrwset from the blockstore
+		txRWSet, err := retrieveRwsetForTx(blockPvtData.BlockNum, txPvtData.SeqInBlock, blockStore)
+		if err != nil {
+			return nil, nil, err
+		}
+
+		// (2) validate passed pvtData against the pvtData hash in the tx rwset.
+		validData, invalidData := findValidAndInvalidTxPvtData(txPvtData, txRWSet, blockPvtData.BlockNum)
+
+		// (3) append validData to validPvtDataPvt list of this block and
+		// invalidData to invalidPvtData list
+		if validData != nil {
+			validPvtData = append(validPvtData, validData)
+		}
+		invalidPvtData = append(invalidPvtData, invalidData...)
+	} // for each tx's pvtData
+	return validPvtData, invalidPvtData, nil
+}
+
+func retrieveRwsetForTx(blkNum uint64, txNum uint64, blockStore *ledgerstorage.Store) (*rwsetutil.TxRwSet, error) {
+	// retrieve the txEnvelope from the block store so that the hash of
+	// the pvtData can be retrieved for comparison
+	txEnvelope, err := blockStore.RetrieveTxByBlockNumTranNum(blkNum, txNum)
+	if err != nil {
+		return nil, err
+	}
+	// retrieve pvtRWset hash from the txEnvelope
+	responsePayload, err := utils.GetActionFromEnvelopeMsg(txEnvelope)
+	if err != nil {
+		return nil, err
+	}
+	txRWSet := &rwsetutil.TxRwSet{}
+	if err := txRWSet.FromProtoBytes(responsePayload.Results); err != nil {
+		return nil, err
+	}
+	return txRWSet, nil
+}
+
+func findValidAndInvalidTxPvtData(txPvtData *ledger.TxPvtData, txRWSet *rwsetutil.TxRwSet, blkNum uint64) (
+	*ledger.TxPvtData, []*ledger.PvtdataHashMismatch,
+) {
+	var invalidPvtData []*ledger.PvtdataHashMismatch
+	var toDeleteNsColl []*nsColl
+	// Compare the hash of pvtData with the hash present in the rwset to
+	// find valid and invalid pvt data
+	for _, nsRwset := range txPvtData.WriteSet.NsPvtRwset {
+		txNum := txPvtData.SeqInBlock
+		invalidData, invalidNsColl := findInvalidNsPvtData(nsRwset, txRWSet, blkNum, txNum)
+		invalidPvtData = append(invalidPvtData, invalidData...)
+		toDeleteNsColl = append(toDeleteNsColl, invalidNsColl...)
+	}
+	for _, nsColl := range toDeleteNsColl {
+		txPvtData.WriteSet.Remove(nsColl.ns, nsColl.coll)
+	}
+	if len(txPvtData.WriteSet.NsPvtRwset) == 0 {
+		// denotes that all namespaces had
+		// invalid pvt data
+		return nil, invalidPvtData
+	}
+	return txPvtData, invalidPvtData
+}
+
+type nsColl struct {
+	ns, coll string
+}
+
+func findInvalidNsPvtData(nsRwset *rwset.NsPvtReadWriteSet, txRWSet *rwsetutil.TxRwSet, blkNum, txNum uint64) (
+	[]*ledger.PvtdataHashMismatch, []*nsColl,
+) {
+	var invalidPvtData []*ledger.PvtdataHashMismatch
+	var invalidNsColl []*nsColl
+
+	ns := nsRwset.Namespace
+	for _, collPvtRwset := range nsRwset.CollectionPvtRwset {
+		coll := collPvtRwset.CollectionName
+		rwsetHash := txRWSet.GetPvtDataHash(ns, coll)
+		if rwsetHash == nil {
+			logger.Warningf("namespace: %s collection: %s was not accessed by txNum %d in BlkNum %d. "+
+				"Unnecessary pvtdata has been passed", ns, coll, txNum, blkNum)
+			invalidNsColl = append(invalidNsColl, &nsColl{ns, coll})
+			continue
+		}
+
+		if !bytes.Equal(util.ComputeSHA256(collPvtRwset.Rwset), rwsetHash) {
+			invalidPvtData = append(invalidPvtData, &ledger.PvtdataHashMismatch{
+				BlockNum:     blkNum,
+				TxNum:        txNum,
+				Namespace:    ns,
+				Collection:   coll,
+				ExpectedHash: rwsetHash})
+			invalidNsColl = append(invalidNsColl, &nsColl{ns, coll})
+		}
+	}
+	return invalidPvtData, invalidNsColl
+}
diff --git a/core/ledger/kvledger/hashcheck_pvtdata_test.go b/core/ledger/kvledger/hashcheck_pvtdata_test.go
@@ -0,0 +1,157 @@
+/*
+Copyright IBM Corp. All Rights Reserved.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package kvledger
+
+import (
+	"fmt"
+	"strings"
+	"testing"
+
+	"github.com/golang/protobuf/proto"
+	"github.com/hyperledger/fabric/common/ledger/testutil"
+	"github.com/hyperledger/fabric/core/ledger"
+	"github.com/hyperledger/fabric/core/ledger/kvledger/txmgmt/rwsetutil"
+	"github.com/hyperledger/fabric/protos/ledger/rwset"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestConstructValidInvalidBlocksPvtData(t *testing.T) {
+	env := newTestEnv(t)
+	defer env.cleanup()
+	provider := testutilNewProvider(t)
+	defer provider.Close()
+
+	_, gb := testutil.NewBlockGenerator(t, "testLedger", false)
+	gbHash := gb.Header.Hash()
+	lg, _ := provider.Create(gb)
+	defer lg.Close()
+
+	// construct pvtData and pubRwSet (i.e., hashed rw set)
+	v0 := []byte{0}
+	pvtDataBlk1Tx0, pubSimResBytesBlk1Tx0 := produceSamplePvtdata(t, 0, []string{"ns-1:coll-1", "ns-1:coll-2"}, [][]byte{v0, v0})
+	v1 := []byte{1}
+	pvtDataBlk1Tx1, pubSimResBytesBlk1Tx1 := produceSamplePvtdata(t, 1, []string{"ns-1:coll-1", "ns-1:coll-2"}, [][]byte{v1, v1})
+	v2 := []byte{2}
+	pvtDataBlk1Tx2, pubSimResBytesBlk1Tx2 := produceSamplePvtdata(t, 2, []string{"ns-1:coll-1", "ns-2:coll-2"}, [][]byte{v2, v2})
+	v3 := []byte{3}
+	pvtDataBlk1Tx3, pubSimResBytesBlk1Tx3 := produceSamplePvtdata(t, 3, []string{"ns-1:coll-1", "ns-1:coll-2"}, [][]byte{v3, v3})
+	v4 := []byte{4}
+	pvtDataBlk1Tx4, pubSimResBytesBlk1Tx4 := produceSamplePvtdata(t, 4, []string{"ns-1:coll-1", "ns-4:coll-2"}, [][]byte{v4, v4})
+	v5 := []byte{5}
+	pvtDataBlk1Tx5, pubSimResBytesBlk1Tx5 := produceSamplePvtdata(t, 5, []string{"ns-1:coll-1", "ns-1:coll-2"}, [][]byte{v5, v5})
+	v6 := []byte{6}
+	pvtDataBlk1Tx6, pubSimResBytesBlk1Tx6 := produceSamplePvtdata(t, 6, []string{"ns-6:coll-2"}, [][]byte{v6})
+	v7 := []byte{7}
+	_, pubSimResBytesBlk1Tx7 := produceSamplePvtdata(t, 7, []string{"ns-1:coll-2"}, [][]byte{v7})
+	wrongPvtDataBlk1Tx7, _ := produceSamplePvtdata(t, 7, []string{"ns-6:coll-2"}, [][]byte{v6})
+
+	pubSimulationResults := &rwset.TxReadWriteSet{}
+	err := proto.Unmarshal(pubSimResBytesBlk1Tx7, pubSimulationResults)
+	assert.NoError(t, err)
+	tx7PvtdataHash := pubSimulationResults.NsRwset[0].CollectionHashedRwset[0].PvtRwsetHash
+
+	// construct block1
+	simulationResultsBlk1 := [][]byte{pubSimResBytesBlk1Tx0, pubSimResBytesBlk1Tx1, pubSimResBytesBlk1Tx2,
+		pubSimResBytesBlk1Tx3, pubSimResBytesBlk1Tx4, pubSimResBytesBlk1Tx5,
+		pubSimResBytesBlk1Tx6, pubSimResBytesBlk1Tx7}
+	blk1 := testutil.ConstructBlock(t, 1, gbHash, simulationResultsBlk1, false)
+
+	// construct a pvtData list for block1
+	pvtDataBlk1 := map[uint64]*ledger.TxPvtData{
+		0: pvtDataBlk1Tx0,
+		1: pvtDataBlk1Tx1,
+		2: pvtDataBlk1Tx2,
+		4: pvtDataBlk1Tx4,
+		5: pvtDataBlk1Tx5,
+	}
+
+	// construct a missingData list for block1
+	missingData := &ledger.MissingPrivateDataList{}
+	missingData.Add("", 3, "ns-1", "coll-1", true)
+	missingData.Add("", 3, "ns-1", "coll-2", true)
+	missingData.Add("", 6, "ns-6", "coll-2", true)
+	missingData.Add("", 7, "ns-1", "coll-2", true)
+
+	// commit block1
+	blockAndPvtData1 := &ledger.BlockAndPvtData{
+		Block:        blk1,
+		BlockPvtData: pvtDataBlk1,
+		Missing:      missingData}
+	assert.NoError(t, lg.(*kvLedger).blockStore.CommitWithPvtData(blockAndPvtData1))
+
+	// construct pvtData from missing data in tx3, tx6, and tx7
+	blocksPvtData := []*ledger.BlockPvtData{
+		{
+			BlockNum: 1,
+			WriteSets: map[uint64]*ledger.TxPvtData{
+				3: pvtDataBlk1Tx3,
+				6: pvtDataBlk1Tx6,
+				7: wrongPvtDataBlk1Tx7,
+				// ns-6:coll-2 does not present in tx7
+			},
+		},
+	}
+
+	expectedValidBlocksPvtData := map[uint64][]*ledger.TxPvtData{
+		1: {
+			pvtDataBlk1Tx3,
+			pvtDataBlk1Tx6,
+		},
+	}
+
+	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])
+	// should not include the pvtData passed for the tx7 even in hashmismatched as ns-6:coll-2 does not exist in tx7
+	assert.Len(t, hashMismatched, 0)
+
+	// construct pvtData from missing data in tx7 with wrong pvtData
+	wrongPvtDataBlk1Tx7, pubSimResBytesBlk1Tx7 = produceSamplePvtdata(t, 7, []string{"ns-1:coll-2"}, [][]byte{v6})
+	blocksPvtData = []*ledger.BlockPvtData{
+		{
+			BlockNum: 1,
+			WriteSets: map[uint64]*ledger.TxPvtData{
+				7: wrongPvtDataBlk1Tx7,
+				// ns-1:coll-1 exists in tx7 but the passed pvtData is incorrect
+			},
+		},
+	}
+
+	expectedHashMismatches := []*ledger.PvtdataHashMismatch{
+		{
+			BlockNum:     1,
+			TxNum:        7,
+			Namespace:    "ns-1",
+			Collection:   "coll-2",
+			ExpectedHash: tx7PvtdataHash,
+		},
+	}
+
+	blocksValidPvtData, hashMismatches, err := ConstructValidAndInvalidPvtData(blocksPvtData, lg.(*kvLedger).blockStore)
+	assert.NoError(t, err)
+	assert.Len(t, blocksValidPvtData, 0)
+
+	assert.ElementsMatch(t, expectedHashMismatches, hashMismatches)
+}
+
+func produceSamplePvtdata(t *testing.T, txNum uint64, nsColls []string, values [][]byte) (*ledger.TxPvtData, []byte) {
+	builder := rwsetutil.NewRWSetBuilder()
+	for index, nsColl := range nsColls {
+		nsCollSplit := strings.Split(nsColl, ":")
+		ns := nsCollSplit[0]
+		coll := nsCollSplit[1]
+		key := fmt.Sprintf("key-%s-%s", ns, coll)
+		value := values[index]
+		builder.AddToPvtAndHashedWriteSet(ns, coll, key, value)
+	}
+	simRes, err := builder.GetTxSimulationResults()
+	assert.NoError(t, err)
+	pubSimulationResultsBytes, err := proto.Marshal(simRes.PubSimulationResults)
+	assert.NoError(t, err)
+	return &ledger.TxPvtData{SeqInBlock: txNum, WriteSet: simRes.PvtSimulationResults}, pubSimulationResultsBytes
+}
diff --git a/core/ledger/kvledger/kv_ledger.go b/core/ledger/kvledger/kv_ledger.go
@@ -7,7 +7,6 @@ SPDX-License-Identifier: Apache-2.0
 package kvledger
 
 import (
-	"fmt"
 	"sync"
 	"time"
 
@@ -337,8 +336,29 @@ func (l *kvLedger) GetConfigHistoryRetriever() (ledger.ConfigHistoryRetriever, e
 	return l.configHistoryRetriever, nil
 }
 
+// TODO: FAB-12849 rename CommitPvtData() ledger API to CommitPvtDataOfOldBlocks()
 func (l *kvLedger) CommitPvtData(pvtData []*ledger.BlockPvtData) ([]*ledger.PvtdataHashMismatch, error) {
-	return nil, fmt.Errorf("not yet implemented")
+	validPvtData, hashMismatches, err := ConstructValidAndInvalidPvtData(pvtData, l.blockStore)
+	if err != nil {
+		return nil, err
+	}
+
+	err = l.blockStore.CommitPvtDataOfOldBlocks(validPvtData)
+	if err != nil {
+		return nil, err
+	}
+
+	// TODO: call txmgr with pvtData list to check whether any states need
+	// to be updated. if so, update the state.
+	// We need to compare write set version in the committedPvtData with the
+	// version in the stateDB. For a tx, if all versions matches, we need to
+	// committ the pvtData. This will be addressed by FAB-11765.
+
+	if err := l.blockStore.ResetLastUpdatedOldBlocksList(); err != nil {
+		return nil, err
+	}
+
+	return hashMismatches, nil
 }
 
 func (l *kvLedger) GetMissingPvtDataTracker() (ledger.MissingPvtDataTracker, error) {

diff --git a/core/ledger/kvledger/txmgmt/rwsetutil/rwset_proto_util.go b/core/ledger/kvledger/txmgmt/rwsetutil/rwset_proto_util.go
@@ -47,6 +47,32 @@ type CollHashedRwSet struct {
 	PvtRwSetHash   []byte
 }
 
+// GetPvtDataHash returns the PvtRwSetHash for a given namespace and collection
+func (txRwSet *TxRwSet) GetPvtDataHash(ns, coll string) []byte {
+	// we could build and use a map to reduce the number of lookup
+	// in the future call. However, we decided to defer such optimization
+	// due to the following assumptions (mainly to avoid additioan LOC).
+	// we assume that the number of namespaces and collections in a txRWSet
+	// to be very minimal (in a single digit),
+	for _, nsRwSet := range txRwSet.NsRwSets {
+		if nsRwSet.NameSpace != ns {
+			continue
+		}
+		return nsRwSet.getPvtDataHash(coll)
+	}
+	return nil
+}
+
+func (nsRwSet *NsRwSet) getPvtDataHash(coll string) []byte {
+	for _, collHashedRwSet := range nsRwSet.CollHashedRwSets {
+		if collHashedRwSet.CollectionName != coll {
+			continue
+		}
+		return collHashedRwSet.PvtRwSetHash
+	}
+	return nil
+}
+
 /////////////////////////////////////////////////////////////////
 // Messages related to PRIVATE read-write set
 /////////////////////////////////////////////////////////////////

diff --git a/core/ledger/ledger_interface.go b/core/ledger/ledger_interface.go
@@ -450,9 +450,9 @@ func (e *InvalidCollNameError) Error() string {
 // does not match the corresponding hash present in the block
 // See function `PeerLedger.CommitPvtData` for the usages
 type PvtdataHashMismatch struct {
-	BlockNum, TxNum               uint64
-	ChaincodeName, CollectionName string
-	ExpectedHash                  []byte
+	BlockNum, TxNum       uint64
+	Namespace, Collection string
+	ExpectedHash          []byte
 }
 
 // DeployedChaincodeInfoProvider is a dependency that is used by ledger to build collection config history