[FAB-10381] Purge orphaned private keys

This CR fixes a bug that in the pvt data purge code
that could leave the orphaned pvt keys un-purged.
See Jira for a more detailed description of the bug.

Also, a number of tests has been added to verify the above bug fix.

Change-Id: I830ec704192977084398d91c29ddb7d364901640
Signed-off-by: manish <manish.sethi@gmail.com>

Author: manish-sethi

core/ledger/kvledger/txmgmt/privacyenabledstate/db.go

@@ -114,7 +114,7 @@ func (b UpdateMap) Put(ns, coll, key string, value []byte, version *version.Heig
 	b.getOrCreateNsBatch(ns).Put(coll, key, value, version)
 }
 
-// Delete removes the entry from the batch for a given combination of namespace and collection name
+// Delete adds a delete marker in the batch for a given combination of namespace and collection name
 func (b UpdateMap) Delete(ns, coll, key string, version *version.Height) {
 	b.getOrCreateNsBatch(ns).Delete(coll, key, version)
 }
@@ -128,6 +128,15 @@ func (b UpdateMap) Get(ns, coll, key string) *statedb.VersionedValue {
 	return nsPvtBatch.Get(coll, key)
 }
 
+// Contains returns true if the given <ns,coll,key> tuple is present in the batch
+func (b UpdateMap) Contains(ns, coll, key string) bool {
+	nsBatch, ok := b[ns]
+	if !ok {
+		return false
+	}
+	return nsBatch.Exists(coll, key)
+}
+
 func (nsb nsBatch) GetCollectionNames() []string {
 	return nsb.GetUpdatedNamespaces()
 }
@@ -143,11 +152,7 @@ func (b UpdateMap) getOrCreateNsBatch(ns string) nsBatch {
 
 // Contains returns true if the given <ns,coll,keyHash> tuple is present in the batch
 func (h HashedUpdateBatch) Contains(ns, coll string, keyHash []byte) bool {
-	nsBatch, ok := h.UpdateMap[ns]
-	if !ok {
-		return false
-	}
-	return nsBatch.Exists(coll, string(keyHash))
+	return h.UpdateMap.Contains(ns, coll, string(keyHash))
 }
 
 // Put overrides the function in UpdateMap for allowing the key to be a []byte instead of a string

core/ledger/kvledger/txmgmt/pvtstatepurgemgmt/expiry_schedule_builder.go

@@ -68,22 +68,30 @@ func buildExpirySchedule(
 	hashedUpdateKeys := hashedUpdates.ToCompositeKeyMap()
 	expiryScheduleBuilder := newExpiryScheduleBuilder(btlPolicy)
 
+	logger.Debugf("Building the expiry schedules based on the update batch")
+
 	// Iterate through the private data updates and for each key add into the expiry schedule
 	// i.e., when these private data key and it's hashed-keys are going to be expired
 	// Note that the 'hashedUpdateKeys'  may be superset of the pvtUpdates. This is because,
 	// the peer may not receive all the private data either because the peer is not eligible for certain private data
 	// or because we allow proceeding with the missing private data data
 	for pvtUpdateKey, vv := range pvtUpdates.ToCompositeKeyMap() {
 		keyHash := util.ComputeStringHash(pvtUpdateKey.Key)
+		hashedCompisiteKey := privacyenabledstate.HashedCompositeKey{
+			Namespace:      pvtUpdateKey.Namespace,
+			CollectionName: pvtUpdateKey.CollectionName,
+			KeyHash:        string(keyHash),
+		}
+		logger.Debugf("Adding expiry schedule for key and key hash [%s]", &hashedCompisiteKey)
 		if err := expiryScheduleBuilder.add(pvtUpdateKey.Namespace, pvtUpdateKey.CollectionName, pvtUpdateKey.Key, keyHash, vv); err != nil {
 			return nil, err
 		}
-		delete(hashedUpdateKeys, privacyenabledstate.HashedCompositeKey{
-			Namespace: pvtUpdateKey.Namespace, CollectionName: pvtUpdateKey.CollectionName, KeyHash: string(keyHash)})
+		delete(hashedUpdateKeys, hashedCompisiteKey)
 	}
 
 	// Add entries for the leftover key hashes i.e., the hashes corresponding to which there is not private key is present
 	for hashedUpdateKey, vv := range hashedUpdateKeys {
+		logger.Debugf("Adding expiry schedule for key hash [%s]", &hashedUpdateKey)
 		if err := expiryScheduleBuilder.add(hashedUpdateKey.Namespace, hashedUpdateKey.CollectionName, "", []byte(hashedUpdateKey.KeyHash), vv); err != nil {
 			return nil, err
 		}

core/ledger/kvledger/txmgmt/pvtstatepurgemgmt/expiry_schedule_builder_test.go

@@ -29,10 +29,10 @@ func TestBuildExpirySchedule(t *testing.T) {
 	btlPolicy := pvtdatapolicy.ConstructBTLPolicy(cs)
 	updates := privacyenabledstate.NewUpdateBatch()
 	updates.PubUpdates.Put("ns1", "pubkey1", []byte("pubvalue1"), version.NewHeight(1, 1))
-	putPvtUpdates(t, updates, "ns1", "coll1", "pvtkey1", []byte("pvtvalue1"), version.NewHeight(1, 1))
-	putPvtUpdates(t, updates, "ns1", "coll2", "pvtkey2", []byte("pvtvalue2"), version.NewHeight(2, 1))
-	putPvtUpdates(t, updates, "ns2", "coll3", "pvtkey3", []byte("pvtvalue3"), version.NewHeight(3, 1))
-	putPvtUpdates(t, updates, "ns3", "coll4", "pvtkey4", []byte("pvtvalue4"), version.NewHeight(4, 1))
+	putPvtAndHashUpdates(t, updates, "ns1", "coll1", "pvtkey1", []byte("pvtvalue1"), version.NewHeight(1, 1))
+	putPvtAndHashUpdates(t, updates, "ns1", "coll2", "pvtkey2", []byte("pvtvalue2"), version.NewHeight(2, 1))
+	putPvtAndHashUpdates(t, updates, "ns2", "coll3", "pvtkey3", []byte("pvtvalue3"), version.NewHeight(3, 1))
+	putPvtAndHashUpdates(t, updates, "ns3", "coll4", "pvtkey4", []byte("pvtvalue4"), version.NewHeight(4, 1))
 
 	listExpinfo, err := buildExpirySchedule(btlPolicy, updates.PvtUpdates, updates.HashUpdates)
 	testutil.AssertNoError(t, err, "")
@@ -57,12 +57,67 @@ func TestBuildExpirySchedule(t *testing.T) {
 	testutil.AssertContainsAll(t, listExpinfo, expectedListExpInfo)
 }
 
-func putPvtUpdates(t *testing.T, updates *privacyenabledstate.UpdateBatch, ns, coll, key string, value []byte, ver *version.Height) {
+func TestBuildExpiryScheduleWithMissingPvtdata(t *testing.T) {
+	cs := btltestutil.NewMockCollectionStore()
+	cs.SetBTL("ns1", "coll1", 1)
+	cs.SetBTL("ns1", "coll2", 2)
+	cs.SetBTL("ns2", "coll3", 3)
+	cs.SetBTL("ns3", "coll4", 0)
+	cs.SetBTL("ns3", "coll5", 20)
+	btlPolicy := pvtdatapolicy.ConstructBTLPolicy(cs)
+	updates := privacyenabledstate.NewUpdateBatch()
+
+	// This update should appear in the expiry schedule with both the key and the hash
+	putPvtAndHashUpdates(t, updates, "ns1", "coll1", "pvtkey1", []byte("pvtvalue1"), version.NewHeight(50, 1))
+
+	// This update should appear in the expiry schedule with only the key-hash
+	putHashUpdates(updates, "ns1", "coll2", "pvtkey2", []byte("pvtvalue2"), version.NewHeight(50, 2))
+
+	// This update should appear in the expiry schedule with only the key-hash
+	putHashUpdates(updates, "ns2", "coll3", "pvtkey3", []byte("pvtvalue3"), version.NewHeight(50, 3))
+
+	// this update is not expectd to appear in the expiry schdule as this collection is configured to expire - 'never'
+	putPvtAndHashUpdates(t, updates, "ns3", "coll4", "pvtkey4", []byte("pvtvalue4"), version.NewHeight(50, 4))
+
+	// the following two updates are not expected to appear in the expiry schdule as they are deletes
+	deletePvtAndHashUpdates(t, updates, "ns3", "coll5", "pvtkey5", version.NewHeight(50, 5))
+	deleteHashUpdates(updates, "ns3", "coll5", "pvtkey6", version.NewHeight(50, 6))
+
+	listExpinfo, err := buildExpirySchedule(btlPolicy, updates.PvtUpdates, updates.HashUpdates)
+	testutil.AssertNoError(t, err, "")
+	t.Logf("listExpinfo=%s", spew.Sdump(listExpinfo))
+
+	pvtdataKeys1 := newPvtdataKeys()
+	pvtdataKeys1.add("ns1", "coll1", "pvtkey1", util.ComputeStringHash("pvtkey1"))
+	pvtdataKeys2 := newPvtdataKeys()
+	pvtdataKeys2.add("ns1", "coll2", "", util.ComputeStringHash("pvtkey2"))
+	pvtdataKeys3 := newPvtdataKeys()
+	pvtdataKeys3.add("ns2", "coll3", "", util.ComputeStringHash("pvtkey3"))
+
+	expectedListExpInfo := []*expiryInfo{
+		{expiryInfoKey: &expiryInfoKey{expiryBlk: 52, committingBlk: 50}, pvtdataKeys: pvtdataKeys1},
+		{expiryInfoKey: &expiryInfoKey{expiryBlk: 53, committingBlk: 50}, pvtdataKeys: pvtdataKeys2},
+		{expiryInfoKey: &expiryInfoKey{expiryBlk: 54, committingBlk: 50}, pvtdataKeys: pvtdataKeys3},
+	}
+
+	testutil.AssertEquals(t, len(listExpinfo), 3)
+	testutil.AssertContainsAll(t, listExpinfo, expectedListExpInfo)
+}
+
+func putPvtAndHashUpdates(t *testing.T, updates *privacyenabledstate.UpdateBatch, ns, coll, key string, value []byte, ver *version.Height) {
 	updates.PvtUpdates.Put(ns, coll, key, value, ver)
-	updates.HashUpdates.Put(ns, coll, util.ComputeStringHash(key), util.ComputeHash(value), ver)
+	putHashUpdates(updates, ns, coll, key, value, ver)
 }
 
-func deletePvtUpdates(t *testing.T, updates *privacyenabledstate.UpdateBatch, ns, coll, key string, ver *version.Height) {
+func deletePvtAndHashUpdates(t *testing.T, updates *privacyenabledstate.UpdateBatch, ns, coll, key string, ver *version.Height) {
 	updates.PvtUpdates.Delete(ns, coll, key, ver)
+	deleteHashUpdates(updates, ns, coll, key, ver)
+}
+
+func putHashUpdates(updates *privacyenabledstate.UpdateBatch, ns, coll, key string, value []byte, ver *version.Height) {
+	updates.HashUpdates.Put(ns, coll, util.ComputeStringHash(key), util.ComputeHash(value), ver)
+}
+
+func deleteHashUpdates(updates *privacyenabledstate.UpdateBatch, ns, coll, key string, ver *version.Height) {
 	updates.HashUpdates.Delete(ns, coll, util.ComputeStringHash(key), ver)
 }

core/ledger/kvledger/txmgmt/pvtstatepurgemgmt/purge_mgr.go

@@ -12,6 +12,7 @@ import (
 
 	"github.com/hyperledger/fabric/core/ledger/kvledger/bookkeeping"
 	"github.com/hyperledger/fabric/core/ledger/kvledger/txmgmt/privacyenabledstate"
+	"github.com/hyperledger/fabric/core/ledger/kvledger/txmgmt/statedb"
 	"github.com/hyperledger/fabric/core/ledger/kvledger/txmgmt/version"
 	"github.com/hyperledger/fabric/core/ledger/pvtdatapolicy"
 )
@@ -31,6 +32,7 @@ type PurgeMgr interface {
 type keyAndVersion struct {
 	key             string
 	committingBlock uint64
+	purgeKeyOnly    bool
 }
 
 type expiryInfoMap map[*privacyenabledstate.HashedCompositeKey]*keyAndVersion
@@ -85,29 +87,37 @@ func (p *purgeMgr) DeleteExpiredAndUpdateBookkeeping(
 	if p.workingset.err != nil {
 		return p.workingset.err
 	}
+
+	listExpiryInfo, err := buildExpirySchedule(p.btlPolicy, pvtUpdates, hashedUpdates)
+	if err != nil {
+		return err
+	}
+
 	// For each key selected for purging, check if the key is not getting updated in the current block,
 	// add its deletion in the update batches for pvt and hashed updates
 	for compositeHashedKey, keyAndVersion := range p.workingset.toPurge {
 		ns := compositeHashedKey.Namespace
 		coll := compositeHashedKey.CollectionName
 		keyHash := []byte(compositeHashedKey.KeyHash)
 		key := keyAndVersion.key
-		if hashedUpdates.Contains(ns, coll, keyHash) {
-			logger.Debugf("Skipping the key [%s] from purging because it is already updated in the current batch", compositeHashedKey)
-			continue
-		}
+		purgeKeyOnly := keyAndVersion.purgeKeyOnly
+		hashUpdated := hashedUpdates.Contains(ns, coll, keyHash)
+		pvtKeyUpdated := pvtUpdates.Contains(ns, coll, key)
+
+		logger.Debugf("Checking whether the key [ns=%s, coll=%s, keyHash=%x, purgeKeyOnly=%t] "+
+			"is updated in the update batch for the committing block - hashUpdated=%t, and pvtKeyUpdated=%t",
+			ns, coll, keyHash, purgeKeyOnly, hashUpdated, pvtKeyUpdated)
+
 		expiringTxVersion := version.NewHeight(p.workingset.expiringBlk, math.MaxUint64)
-		logger.Debugf("Purging the hashed key [%s]", compositeHashedKey)
-		hashedUpdates.Delete(ns, coll, keyHash, expiringTxVersion)
-		if keyAndVersion.key != "" {
-			logger.Debugf("Purging the pvt key corresponding to hashed key [%s]", compositeHashedKey)
+		if !hashUpdated && !purgeKeyOnly {
+			logger.Debugf("Adding the hashed key to be purged to the delete list in the update batch")
+			hashedUpdates.Delete(ns, coll, keyHash, expiringTxVersion)
+		}
+		if key != "" && !pvtKeyUpdated {
+			logger.Debugf("Adding the pvt key to be purged to the delete list in the update batch")
 			pvtUpdates.Delete(ns, coll, key, expiringTxVersion)
 		}
 	}
-	listExpiryInfo, err := buildExpirySchedule(p.btlPolicy, pvtUpdates, hashedUpdates)
-	if err != nil {
-		return err
-	}
 	return p.expKeeper.updateBookkeeping(listExpiryInfo, nil)
 }
 
@@ -127,6 +137,7 @@ func (p *purgeMgr) BlockCommitDone() {
 // prepareWorkingsetFor returns a working set for a given expiring block 'expiringAtBlk'.
 // This working set contains the pvt data keys that will expire with the commit of block 'expiringAtBlk'.
 func (p *purgeMgr) prepareWorkingsetFor(expiringAtBlk uint64) *workingset {
+	logger.Debugf("Preparing potential purge list working-set for expiringAtBlk [%d]", expiringAtBlk)
 	workingset := &workingset{expiringBlk: expiringAtBlk}
 	// Retrieve the keys from bookkeeper
 	expiryInfo, err := p.expKeeper.retrieve(expiringAtBlk)
@@ -140,18 +151,49 @@ func (p *purgeMgr) prepareWorkingsetFor(expiringAtBlk uint64) *workingset {
 	p.preloadCommittedVersionsInCache(toPurge)
 	var expiryInfoKeysToClear []*expiryInfoKey
 
-	// for each hashed key, check whether the committed version is still the same (i.e, the key was not overwritten already)
-	for hashedKey, keyAndVersion := range toPurge {
-		expiryInfoKeysToClear = append(expiryInfoKeysToClear, &expiryInfoKey{committingBlk: keyAndVersion.committingBlock, expiryBlk: expiringAtBlk})
-		currentVersion, err := p.db.GetKeyHashVersion(hashedKey.Namespace, hashedKey.CollectionName, []byte(hashedKey.KeyHash))
+	if len(toPurge) == 0 {
+		logger.Debugf("No expiry entry found for expiringAtBlk [%d]", expiringAtBlk)
+		return workingset
+	}
+	logger.Debugf("Total [%d] expiring entries found. Evaluaitng whether some of these keys have been overwritten in later blocks...", len(toPurge))
+
+	for purgeEntryK, purgeEntryV := range toPurge {
+		logger.Debugf("Evaluating for hashedKey [%s]", purgeEntryK)
+		expiryInfoKeysToClear = append(expiryInfoKeysToClear, &expiryInfoKey{committingBlk: purgeEntryV.committingBlock, expiryBlk: expiringAtBlk})
+		currentVersion, err := p.db.GetKeyHashVersion(purgeEntryK.Namespace, purgeEntryK.CollectionName, []byte(purgeEntryK.KeyHash))
 		if err != nil {
 			workingset.err = err
 			return workingset
 		}
-		// if the committed version is different from the expiry entry in the bookkeeper, remove the key from the purge-list (the key was overwritten already)
-		if !sameVersion(currentVersion, keyAndVersion.committingBlock) {
-			delete(toPurge, hashedKey)
+
+		if sameVersion(currentVersion, purgeEntryV.committingBlock) {
+			logger.Debugf(
+				"The version of the hashed key in the committed state and in the expiry entry is same " +
+					"hence, keeping the entry in the purge list")
+			continue
+		}
+
+		logger.Debugf("The version of the hashed key in the committed state and in the expiry entry is different")
+		if purgeEntryV.key != "" {
+			logger.Debugf("The expiry entry also contains the raw key along with the key hash")
+			committedPvtVerVal, err := p.db.GetPrivateData(purgeEntryK.Namespace, purgeEntryK.CollectionName, purgeEntryV.key)
+			if err != nil {
+				workingset.err = err
+				return workingset
+			}
+
+			if sameVersionFromVal(committedPvtVerVal, purgeEntryV.committingBlock) {
+				logger.Debugf(
+					"The version of the pvt key in the committed state and in the expiry entry is same" +
+						"Including only key in the purge list and not the hashed key")
+				purgeEntryV.purgeKeyOnly = true
+				continue
+			}
 		}
+
+		// If we reached here, the keyhash and private key (if present, in the expiry entry) have been updated in a later block, therefore remove from current purge list
+		logger.Debugf("Removing from purge list - the key hash and key (if present, in the expiry entry)")
+		delete(toPurge, purgeEntryK)
 	}
 	// Final keys to purge from state
 	workingset.toPurge = toPurge
@@ -189,3 +231,7 @@ func transformToExpiryInfoMap(expiryInfo []*expiryInfo) expiryInfoMap {
 func sameVersion(version *version.Height, blockNum uint64) bool {
 	return version != nil && version.BlockNum == blockNum
 }
+
+func sameVersionFromVal(vv *statedb.VersionedValue, blockNum uint64) bool {
+	return vv != nil && sameVersion(vv.Version, blockNum)
+}