Merge "[FAB-16821] Use skipPullingInvalidTransactions flag"

seuzht · Oct 21, 2019 · 7a41685 · 7a41685
2 parents a427a02 + 4cab077
commit 7a41685
Show file tree

Hide file tree

Showing 3 changed files with 261 additions and 154 deletions.
diff --git a/gossip/privdata/coordinator.go b/gossip/privdata/coordinator.go
@@ -484,7 +484,11 @@ func (s rwsetByKeys) bySeqsInBlock() map[uint64]readWriteSets {
 	return res
 }
 
-type rwsetKeys map[rwSetKey]struct{}
+type rwsetInfo struct {
+	invalid bool
+}
+
+type rwsetKeys map[rwSetKey]rwsetInfo
 
 // String returns a string representation of the rwsetKeys
 func (s rwsetKeys) String() string {
@@ -519,10 +523,10 @@ type txAndSeqInBlock struct {
 type rwSetKeysByTxIDs map[txAndSeqInBlock][]rwSetKey
 
 func (s rwSetKeysByTxIDs) flatten() rwsetKeys {
-	m := make(map[rwSetKey]struct{})
+	m := make(rwsetKeys)
 	for _, keys := range s {
 		for _, k := range keys {
-			m[k] = struct{}{}
+			m[k] = rwsetInfo{}
 		}
 	}
 	return m
@@ -680,7 +684,7 @@ func (c *coordinator) listMissingPrivateData(block *common.Block, ownedRWsets ma
 	}
 
 	sources := make(map[rwSetKey][]*peer.Endorsement)
-	requestedEligiblePrivateRWSets := make(map[rwSetKey]struct{})
+	requestedEligiblePrivateRWSets := make(rwsetKeys)
 	eligibleMissingKeysByTxIDs := make(rwSetKeysByTxIDs)
 	ineligibleMissingKeysByTxIDs := make(rwSetKeysByTxIDs)
 	data := blockData(block.Data.Data)
@@ -780,7 +784,7 @@ func (bi *transactionInspector) inspectTransaction(seqInBlock uint64, chdr *comm
 				continue
 			}
 
-			bi.requestedEligiblePrivateRWSets[key] = struct{}{}
+			bi.requestedEligiblePrivateRWSets[key] = rwsetInfo{}
 			if _, exists := bi.ownedRWsets[key]; !exists {
 				bi.eligibleMissingKeysByTxIDs[txAndSeq] = append(bi.eligibleMissingKeysByTxIDs[txAndSeq], key)
 				bi.sources[key] = endorsersFromOrgs(ns.NameSpace, hashedCollection.CollectionName, endorsers, policy.MemberOrgs())

diff --git a/gossip/privdata/pvtdataprovider.go b/gossip/privdata/pvtdataprovider.go
@@ -124,12 +124,15 @@ func (ec *eligibilityComputer) computeEligibility(txPvtdataQuery []*ledger.TxPvt
 				ec.logger.Debugf("Peer is not eligible for collection: chaincode [%s], "+
 					"collection name [%s], txID [%s] the policy is [%#v]. Skipping.",
 					ns, col, txID, policy)
-				ineligibleMissingKeys[key] = struct{}{}
+				ineligibleMissingKeys[key] = rwsetInfo{}
 				continue
 			}
 
 			// treat all eligible keys as missing
-			eligibleMissingKeys[key] = struct{}{}
+			eligibleMissingKeys[key] = rwsetInfo{
+				invalid: invalid,
+			}
+
 			sources[key] = endorsersFromOrgs(ns, col, endorsers, policy.MemberOrgs())
 		}
 	}
@@ -155,14 +158,15 @@ type PvtdataProvider struct {
 	channelID                               string
 	blockNum                                uint64
 	storePvtdataOfInvalidTx                 bool
+	skipPullingInvalidTransactions          bool
 	idDeserializerFactory                   IdentityDeserializerFactory
 	fetcher                                 Fetcher
 
 	sleeper sleeper
 }
 
 // RetrievePvtdata retrieves the private data for the given txs containing private data
-func (pdp *PvtdataProvider) RetrievePrivatedata(txPvtdataQuery []*ledger.TxPvtdataInfo) (*RetrievedPvtdata, error) {
+func (pdp *PvtdataProvider) RetrievePvtdata(txPvtdataQuery []*ledger.TxPvtdataInfo) (*RetrievedPvtdata, error) {
 	retrievedPvtdata := &RetrievedPvtdata{
 		transientStore:          pdp.transientStore,
 		logger:                  pdp.logger,
@@ -339,7 +343,11 @@ func (pdp *PvtdataProvider) populateFromRemotePeers(pvtdata rwsetByKeys, private
 	pdp.logger.Debugf("Attempting to retrieve %d private write sets from remote peers.", len(privateInfo.eligibleMissingKeys))
 
 	dig2src := make(map[pvtdatacommon.DigKey][]*peer.Endorsement)
-	for k := range privateInfo.eligibleMissingKeys {
+	for k, v := range privateInfo.eligibleMissingKeys {
+		if v.invalid && pdp.skipPullingInvalidTransactions {
+			pdp.logger.Debugf("Skipping invalid key [%v] because peer is configured to skip pulling rwsets of invalid transactions.", k)
+			continue
+		}
 		pdp.logger.Debugf("Fetching [%v] from remote peers", k)
 		dig := pvtdatacommon.DigKey{
 			TxId:       k.txID,