Missingdata-recon: Handle coll eligibility change

This CR handles the event when a peer becomes eligible
for receiving data for an existing collection. All the
missing data entries for the collection that were previously
 marked as 'ineligible' are converted to 'eligible' in a
background goroutine so that the query results for reporting
missing data also include these entries for previous blocks

FAB-11437 #done

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

Author: manish-sethi

common/ledger/util/leveldbhelper/leveldb_provider.go

@@ -136,6 +136,11 @@ func (batch *UpdateBatch) Delete(key []byte) {
 	batch.KVs[string(key)] = nil
 }
 
+// Len returns the number of entries in the batch
+func (batch *UpdateBatch) Len() int {
+	return len(batch.KVs)
+}
+
 // Iterator extends actual leveldb iterator
 type Iterator struct {
 	iterator.Iterator

core/ledger/ledgerconfig/ledger_config.go

@@ -38,6 +38,9 @@ const confMaxBatchSize = "ledger.state.couchDBConfig.maxBatchUpdateSize"
 const confAutoWarmIndexes = "ledger.state.couchDBConfig.autoWarmIndexes"
 const confWarmIndexesAfterNBlocks = "ledger.state.couchDBConfig.warmIndexesAfterNBlocks"
 
+var confCollElgProcMaxDbBatchSize = &conf{"ledger.pvtdataStore.collElgProcMaxDbBatchSize", 5000}
+var confCollElgProcDbBatchesInterval = &conf{"ledger.pvtdataStore.collElgProcDbBatchesInterval", 1000}
+
 // GetRootPath returns the filesystem path.
 // All ledger related contents are expected to be stored under this path
 func GetRootPath() string {
@@ -85,7 +88,7 @@ func GetMaxBlockfileSize() int {
 	return 64 * 1024 * 1024
 }
 
-//GetTotalLimit exposes the totalLimit variable
+// GetTotalQueryLimit exposes the totalLimit variable
 func GetTotalQueryLimit() int {
 	totalQueryLimit := viper.GetInt(confTotalQueryLimit)
 	// if queryLimit was unset, default to 10000
@@ -95,7 +98,7 @@ func GetTotalQueryLimit() int {
 	return totalQueryLimit
 }
 
-//GetQueryLimit exposes the queryLimit variable
+// GetInternalQueryLimit exposes the queryLimit variable
 func GetInternalQueryLimit() int {
 	internalQueryLimit := viper.GetInt(confInternalQueryLimit)
 	// if queryLimit was unset, default to 1000
@@ -125,6 +128,26 @@ func GetPvtdataStorePurgeInterval() uint64 {
 	return uint64(purgeInterval)
 }
 
+// GetPvtdataStoreCollElgProcMaxDbBatchSize returns the maximum db batch size for converting
+// the ineligible missing data entries to eligible missing data entries
+func GetPvtdataStoreCollElgProcMaxDbBatchSize() int {
+	collElgProcMaxDbBatchSize := viper.GetInt(confCollElgProcMaxDbBatchSize.Name)
+	if collElgProcMaxDbBatchSize <= 0 {
+		collElgProcMaxDbBatchSize = confCollElgProcMaxDbBatchSize.DefaultVal
+	}
+	return collElgProcMaxDbBatchSize
+}
+
+// GetPvtdataStoreCollElgProcDbBatchesInterval returns the minimum duration (in milliseconds) between writing
+// two consecutive db batches for converting the ineligible missing data entries to eligible missing data entries
+func GetPvtdataStoreCollElgProcDbBatchesInterval() int {
+	collElgProcDbBatchesInterval := viper.GetInt(confCollElgProcDbBatchesInterval.Name)
+	if collElgProcDbBatchesInterval <= 0 {
+		collElgProcDbBatchesInterval = confCollElgProcDbBatchesInterval.DefaultVal
+	}
+	return collElgProcDbBatchesInterval
+}
+
 //IsHistoryDBEnabled exposes the historyDatabase variable
 func IsHistoryDBEnabled() bool {
 	return viper.GetBool(confEnableHistoryDatabase)
@@ -162,3 +185,8 @@ func GetWarmIndexesAfterNBlocks() int {
 	}
 	return warmAfterNBlocks
 }
+
+type conf struct {
+	Name       string
+	DefaultVal int
+}

core/ledger/ledgerconfig/ledger_config_test.go

@@ -158,6 +158,22 @@ func TestPvtdataStorePurgeInterval(t *testing.T) {
 	assert.Equal(t, uint64(1000), updatedValue) //test config returns 1000
 }
 
+func TestPvtdataStoreCollElgProcMaxDbBatchSize(t *testing.T) {
+	defaultVal := confCollElgProcMaxDbBatchSize.DefaultVal
+	testVal := defaultVal + 1
+	assert.Equal(t, defaultVal, GetPvtdataStoreCollElgProcMaxDbBatchSize())
+	viper.Set("ledger.pvtdataStore.collElgProcMaxDbBatchSize", testVal)
+	assert.Equal(t, testVal, GetPvtdataStoreCollElgProcMaxDbBatchSize())
+}
+
+func TestCollElgProcDbBatchesInterval(t *testing.T) {
+	defaultVal := confCollElgProcDbBatchesInterval.DefaultVal
+	testVal := defaultVal + 1
+	assert.Equal(t, defaultVal, GetPvtdataStoreCollElgProcDbBatchesInterval())
+	viper.Set("ledger.pvtdataStore.collElgProcDbBatchesInterval", testVal)
+	assert.Equal(t, testVal, GetPvtdataStoreCollElgProcDbBatchesInterval())
+}
+
 func TestIsHistoryDBEnabledDefault(t *testing.T) {
 	setUpCoreYAMLConfig()
 	defaultValue := IsHistoryDBEnabled()

core/ledger/pvtdatastorage/expiry_data.pb.go

@@ -8,11 +8,13 @@ package pvtdatastorage
 
 import (
 	"bytes"
+	"math"
 
 	"github.com/golang/protobuf/proto"
 	"github.com/hyperledger/fabric/core/ledger/kvledger/txmgmt/version"
 	"github.com/hyperledger/fabric/core/ledger/util"
 	"github.com/hyperledger/fabric/protos/ledger/rwset"
+	"github.com/pkg/errors"
 	"github.com/willf/bitset"
 )
 
@@ -23,6 +25,7 @@ var (
 	expiryKeyPrefix                = []byte{3}
 	eligibleMissingDataKeyPrefix   = []byte{4}
 	ineligibleMissingDataKeyPrefix = []byte{5}
+	collElgKeyPrefix               = []byte{6}
 
 	nilByte    = byte(0)
 	emptyValue = []byte{}
@@ -145,9 +148,51 @@ func decodeMissingDataValue(bitmapBytes []byte) (*bitset.BitSet, error) {
 	return bitmap, nil
 }
 
+func encodeCollElgKey(blkNum uint64) []byte {
+	return append(collElgKeyPrefix, util.EncodeReverseOrderVarUint64(blkNum)...)
+}
+
+func decodeCollElgKey(b []byte) uint64 {
+	blkNum, _ := util.DecodeReverseOrderVarUint64(b[1:])
+	return blkNum
+}
+
+func encodeCollElgVal(m *CollElgInfo) ([]byte, error) {
+	return proto.Marshal(m)
+}
+
+func decodeCollElgVal(b []byte) (*CollElgInfo, error) {
+	m := &CollElgInfo{}
+	if err := proto.Unmarshal(b, m); err != nil {
+		return nil, errors.WithStack(err)
+	}
+	return m, nil
+}
+
 func createRangeScanKeysForEligibleMissingDataEntries(blkNum uint64) (startKey, endKey []byte) {
 	startKey = append(eligibleMissingDataKeyPrefix, util.EncodeReverseOrderVarUint64(blkNum)...)
 	endKey = append(eligibleMissingDataKeyPrefix, util.EncodeReverseOrderVarUint64(0)...)
 
 	return startKey, endKey
 }
+
+func createRangeScanKeysForIneligibleMissingData(maxBlkNum uint64, ns, coll string) (startKey, endKey []byte) {
+	startKey = encodeMissingDataKey(
+		&missingDataKey{
+			nsCollBlk:  nsCollBlk{ns: ns, coll: coll, blkNum: maxBlkNum},
+			isEligible: false,
+		},
+	)
+	endKey = encodeMissingDataKey(
+		&missingDataKey{
+			nsCollBlk:  nsCollBlk{ns: ns, coll: coll, blkNum: 0},
+			isEligible: false,
+		},
+	)
+	return
+}
+
+func createRangeScanKeysForCollElg() (startKey, endKey []byte) {
+	return encodeCollElgKey(math.MaxUint64),
+		encodeCollElgKey(0)
+}