[FAB-6555] Purge manager for pvt statedb data

This CR introduces a purge manager that manages the
schedule of expiry of pvt data in the expiry keeper
and actual purging of the data from pvt statedb

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

Author: manish-sethi

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

@@ -84,7 +84,6 @@ func (s *CommonStorageDB) LoadCommittedVersionsOfPubAndHashedKeys(pubKeys []*sta
 	if !ok {
 		return nil
 	}
-
 	// Here, hashedKeys are merged into pubKeys to get a combined set of keys for combined loading
 	for _, key := range hashedKeys {
 		ns := deriveHashedDataNs(key.Namespace, key.CollectionName)

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

@@ -7,6 +7,8 @@ SPDX-License-Identifier: Apache-2.0
 package privacyenabledstate
 
 import (
+	"fmt"
+
 	"github.com/hyperledger/fabric/core/ledger/cceventmgmt"
 	"github.com/hyperledger/fabric/core/ledger/kvledger/txmgmt/statedb"
 	"github.com/hyperledger/fabric/core/ledger/kvledger/txmgmt/version"
@@ -37,6 +39,13 @@ type DB interface {
 	ApplyPrivacyAwareUpdates(updates *UpdateBatch, height *version.Height) error
 }
 
+// PvtdataCompositeKey encloses Namespace, CollectionName and Key components
+type PvtdataCompositeKey struct {
+	Namespace      string
+	CollectionName string
+	Key            string
+}
+
 // HashedCompositeKey encloses Namespace, CollectionName and KeyHash components
 type HashedCompositeKey struct {
 	Namespace      string
@@ -150,3 +159,33 @@ func (h HashedUpdateBatch) Put(ns, coll string, key []byte, value []byte, versio
 func (h HashedUpdateBatch) Delete(ns, coll string, key []byte, version *version.Height) {
 	h.UpdateMap.Delete(ns, coll, string(key), version)
 }
+
+// ToCompositeKeyMap rearranges the update batch data in the form of a single map
+func (h HashedUpdateBatch) ToCompositeKeyMap() map[HashedCompositeKey]*statedb.VersionedValue {
+	m := make(map[HashedCompositeKey]*statedb.VersionedValue)
+	for ns, nsBatch := range h.UpdateMap {
+		for _, coll := range nsBatch.GetCollectionNames() {
+			for key, vv := range nsBatch.GetUpdates(coll) {
+				m[HashedCompositeKey{ns, coll, key}] = vv
+			}
+		}
+	}
+	return m
+}
+
+// ToCompositeKeyMap rearranges the update batch data in the form of a single map
+func (p PvtUpdateBatch) ToCompositeKeyMap() map[PvtdataCompositeKey]*statedb.VersionedValue {
+	m := make(map[PvtdataCompositeKey]*statedb.VersionedValue)
+	for ns, nsBatch := range p.UpdateMap {
+		for _, coll := range nsBatch.GetCollectionNames() {
+			for key, vv := range nsBatch.GetUpdates(coll) {
+				m[PvtdataCompositeKey{ns, coll, key}] = vv
+			}
+		}
+	}
+	return m
+}
+
+func (hck *HashedCompositeKey) String() string {
+	return fmt.Sprintf("ns=%s, collection=%s, keyHash=%x", hck.Namespace, hck.CollectionName, hck.KeyHash)
+}

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

@@ -129,8 +129,6 @@ func testDB(t *testing.T, env TestEnv) {
 	assert.Nil(t, vv)
 }
 
-//TODO add tests for functions GetPrivateStateMultipleKeys and GetPrivateStateRangeScanIterator
-
 func TestGetStateMultipleKeys(t *testing.T) {
 	for _, env := range testEnvs {
 		t.Run(env.GetName(), func(t *testing.T) {
@@ -387,6 +385,26 @@ func testKey(i int) string {
 	return fmt.Sprintf("key%d", i)
 }
 
+func TestCompositeKeyMap(t *testing.T) {
+	b := NewPvtUpdateBatch()
+	b.Put("ns1", "coll1", "key1", []byte("testVal1"), nil)
+	b.Delete("ns1", "coll2", "key2", nil)
+	b.Put("ns2", "coll1", "key1", []byte("testVal3"), nil)
+	b.Put("ns2", "coll2", "key2", []byte("testVal4"), nil)
+	m := b.ToCompositeKeyMap()
+	testutil.AssertEquals(t, len(m), 4)
+	vv, ok := m[PvtdataCompositeKey{"ns1", "coll1", "key1"}]
+	testutil.AssertEquals(t, ok, true)
+	testutil.AssertEquals(t, vv.Value, []byte("testVal1"))
+	vv, ok = m[PvtdataCompositeKey{"ns1", "coll2", "key2"}]
+	testutil.AssertNil(t, vv.Value)
+	testutil.AssertEquals(t, ok, true)
+	_, ok = m[PvtdataCompositeKey{"ns2", "coll1", "key1"}]
+	testutil.AssertEquals(t, ok, true)
+	_, ok = m[PvtdataCompositeKey{"ns2", "coll2", "key2"}]
+	testutil.AssertEquals(t, ok, true)
+}
+
 func putPvtUpdates(t *testing.T, updates *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)

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

@@ -0,0 +1,84 @@
+/*
+Copyright IBM Corp. All Rights Reserved.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package pvtstatepurgemgmt
+
+import (
+	math "math"
+
+	"github.com/hyperledger/fabric/core/ledger/kvledger/txmgmt/privacyenabledstate"
+	"github.com/hyperledger/fabric/core/ledger/kvledger/txmgmt/statedb"
+	"github.com/hyperledger/fabric/core/ledger/pvtdatapolicy"
+	"github.com/hyperledger/fabric/core/ledger/util"
+)
+
+type expiryScheduleBuilder struct {
+	btlPolicy       pvtdatapolicy.BTLPolicy
+	scheduleEntries map[expiryInfoKey]*PvtdataKeys
+}
+
+func newExpiryScheduleBuilder(btlPolicy pvtdatapolicy.BTLPolicy) *expiryScheduleBuilder {
+	return &expiryScheduleBuilder{btlPolicy, make(map[expiryInfoKey]*PvtdataKeys)}
+}
+
+func (builder *expiryScheduleBuilder) add(ns, coll, key string, keyHash []byte, versionedValue *statedb.VersionedValue) {
+	committingBlk := versionedValue.Version.BlockNum
+	expiryBlk := builder.btlPolicy.GetExpiringBlock(ns, coll, committingBlk)
+	if isDelete(versionedValue) || neverExpires(expiryBlk) {
+		return
+	}
+	expinfoKey := expiryInfoKey{committingBlk: committingBlk, expiryBlk: expiryBlk}
+	pvtdataKeys, ok := builder.scheduleEntries[expinfoKey]
+	if !ok {
+		pvtdataKeys = newPvtdataKeys()
+		builder.scheduleEntries[expinfoKey] = pvtdataKeys
+	}
+	pvtdataKeys.add(ns, coll, key, keyHash)
+}
+
+func isDelete(versionedValue *statedb.VersionedValue) bool {
+	return versionedValue.Value == nil
+}
+
+func neverExpires(expiryBlk uint64) bool {
+	return expiryBlk == math.MaxUint64
+}
+
+func (builder *expiryScheduleBuilder) getExpiryInfo() []*expiryInfo {
+	var listExpinfo []*expiryInfo
+	for expinfoKey, pvtdataKeys := range builder.scheduleEntries {
+		expinfoKeyCopy := expinfoKey
+		listExpinfo = append(listExpinfo, &expiryInfo{expiryInfoKey: &expinfoKeyCopy, pvtdataKeys: pvtdataKeys})
+	}
+	return listExpinfo
+}
+
+func buildExpirySchedule(
+	btlPolicy pvtdatapolicy.BTLPolicy,
+	pvtUpdates *privacyenabledstate.PvtUpdateBatch,
+	hashedUpdates *privacyenabledstate.HashedUpdateBatch) []*expiryInfo {
+
+	hashedUpdateKeys := hashedUpdates.ToCompositeKeyMap()
+	expiryScheduleBuilder := newExpiryScheduleBuilder(btlPolicy)
+
+	// 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)
+		expiryScheduleBuilder.add(pvtUpdateKey.Namespace, pvtUpdateKey.CollectionName, pvtUpdateKey.Key, keyHash, vv)
+		delete(hashedUpdateKeys, privacyenabledstate.HashedCompositeKey{
+			Namespace: pvtUpdateKey.Namespace, CollectionName: pvtUpdateKey.CollectionName, KeyHash: string(keyHash)})
+	}
+
+	// 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 {
+		expiryScheduleBuilder.add(hashedUpdateKey.Namespace, hashedUpdateKey.CollectionName, "", []byte(hashedUpdateKey.KeyHash), vv)
+	}
+	return expiryScheduleBuilder.getExpiryInfo()
+}

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

@@ -0,0 +1,68 @@
+/*
+Copyright IBM Corp. All Rights Reserved.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package pvtstatepurgemgmt
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/davecgh/go-spew/spew"
+
+	"github.com/hyperledger/fabric/common/ledger/testutil"
+	"github.com/hyperledger/fabric/core/ledger/kvledger/txmgmt/privacyenabledstate"
+	"github.com/hyperledger/fabric/core/ledger/kvledger/txmgmt/version"
+	"github.com/hyperledger/fabric/core/ledger/pvtdatapolicy"
+	"github.com/hyperledger/fabric/core/ledger/util"
+	"github.com/spf13/viper"
+)
+
+func TestBuildExpirySchedule(t *testing.T) {
+	ledgerid := "testledger-BuildExpirySchedule"
+	viper.Set(fmt.Sprintf("ledger.pvtdata.btlpolicy.%s.ns1.coll1", ledgerid), 1)
+	viper.Set(fmt.Sprintf("ledger.pvtdata.btlpolicy.%s.ns1.coll2", ledgerid), 2)
+	viper.Set(fmt.Sprintf("ledger.pvtdata.btlpolicy.%s.ns2.coll3", ledgerid), 3)
+
+	btlPolicy, _ := pvtdatapolicy.GetBTLPolicy(ledgerid)
+
+	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))
+
+	listExpinfo := buildExpirySchedule(btlPolicy, updates.PvtUpdates, updates.HashUpdates)
+	t.Logf("listExpinfo=%s", spew.Sdump(listExpinfo))
+
+	pvtdataKeys1 := newPvtdataKeys()
+	pvtdataKeys1.add("ns1", "coll1", "pvtkey1", util.ComputeStringHash("pvtkey1"))
+
+	pvtdataKeys2 := newPvtdataKeys()
+	pvtdataKeys2.add("ns1", "coll2", "pvtkey2", util.ComputeStringHash("pvtkey2"))
+
+	pvtdataKeys3 := newPvtdataKeys()
+	pvtdataKeys3.add("ns2", "coll3", "pvtkey3", util.ComputeStringHash("pvtkey3"))
+
+	expectedListExpInfo := []*expiryInfo{
+		{expiryInfoKey: &expiryInfoKey{expiryBlk: 3, committingBlk: 1}, pvtdataKeys: pvtdataKeys1},
+		{expiryInfoKey: &expiryInfoKey{expiryBlk: 5, committingBlk: 2}, pvtdataKeys: pvtdataKeys2},
+		{expiryInfoKey: &expiryInfoKey{expiryBlk: 7, committingBlk: 3}, pvtdataKeys: pvtdataKeys3},
+	}
+
+	testutil.AssertEquals(t, len(listExpinfo), 3)
+	testutil.AssertContainsAll(t, listExpinfo, expectedListExpInfo)
+}
+
+func putPvtUpdates(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)
+}
+
+func deletePvtUpdates(t *testing.T, updates *privacyenabledstate.UpdateBatch, ns, coll, key string, ver *version.Height) {
+	updates.PvtUpdates.Delete(ns, coll, key, ver)
+	updates.HashUpdates.Delete(ns, coll, util.ComputeStringHash(key), ver)
+}