[FAB-10366]: Prevent pulling form peer purged data

While seeking for missing private data lagging behind peer could try to
pull peaces from the peers which already purged it due to BTL policy.
This in particular might be a problem while joining a new peer to the
network after a while. This commit adds routing filter to prevent from
pulling private data from peers which already purged it, thus not
wasting time while committing block.

Change-Id: I8ccfc186d69c12e5ae933455640b721454a2e80a
Signed-off-by: Artem Barger <bartem@il.ibm.com>

Author: C0rWin

gossip/privdata/coordinator.go

@@ -104,10 +104,17 @@ func (d2s dig2sources) keys() []*gossip2.PvtDataDigest {
 	return res
 }
 
+// FetchedPvtDataContainer container for pvt data elements
+// returned by Fetcher
+type FetchedPvtDataContainer struct {
+	AvailableElemenets []*gossip2.PvtDataElement
+	PurgedElements     []*gossip2.PvtDataDigest
+}
+
 // Fetcher interface which defines API to fetch missing
 // private data elements
 type Fetcher interface {
-	fetch(dig2src dig2sources) ([]*gossip2.PvtDataElement, error)
+	fetch(dig2src dig2sources, blockSeq uint64) (*FetchedPvtDataContainer, error)
 }
 
 // Support encapsulates set of interfaces to
@@ -262,14 +269,14 @@ func (c *coordinator) fetchFromPeers(blockSeq uint64, ownedRWsets map[rwSetKey][
 		}
 		dig2src[dig] = privateInfo.sources[k]
 	})
-	fetchedData, err := c.fetch(dig2src)
+	fetchedData, err := c.fetch(dig2src, blockSeq)
 	if err != nil {
 		logger.Warning("Failed fetching private data for block", blockSeq, "from peers:", err)
 		return
 	}
 
 	// Iterate over data fetched from peers
-	for _, element := range fetchedData {
+	for _, element := range fetchedData.AvailableElemenets {
 		dig := element.Digest
 		for _, rws := range element.Payload {
 			hash := hex.EncodeToString(util2.ComputeSHA256(rws))
@@ -292,6 +299,18 @@ func (c *coordinator) fetchFromPeers(blockSeq uint64, ownedRWsets map[rwSetKey][
 			logger.Debug("Fetched", key)
 		}
 	}
+	// Iterate over purged data
+	for _, dig := range fetchedData.PurgedElements {
+		// delete purged key from missing keys
+		for missingPvtRWKey := range privateInfo.missingKeys {
+			if missingPvtRWKey.namespace == dig.Namespace &&
+				missingPvtRWKey.collection == dig.Collection &&
+				missingPvtRWKey.txID == dig.TxId {
+				delete(privateInfo.missingKeys, missingPvtRWKey)
+				logger.Debug(missingPvtRWKey, "was purged or will soon be purged, skipping fetch")
+			}
+		}
+	}
 }
 
 func (c *coordinator) fetchMissingFromTransientStore(missing rwSetKeysByTxIDs, ownedRWsets map[rwSetKey][]byte) {

gossip/privdata/coordinator_test.go

@@ -293,8 +293,7 @@ func (f *fetcherMock) On(methodName string, arguments ...interface{}) *fetchCall
 	}
 }
 
-func (f *fetcherMock) fetch(dig2src dig2sources) ([]*proto.PvtDataElement, error) {
-	fmt.Println("XXX: Expected endorsers", f.expectedEndorsers)
+func (f *fetcherMock) fetch(dig2src dig2sources, _ uint64) (*FetchedPvtDataContainer, error) {
 	for _, endorsements := range dig2src {
 		for _, endorsement := range endorsements {
 			_, exists := f.expectedEndorsers[string(endorsement.Endorser)]
@@ -309,7 +308,7 @@ func (f *fetcherMock) fetch(dig2src dig2sources) ([]*proto.PvtDataElement, error
 	assert.Empty(f.t, f.expectedEndorsers)
 	args := f.Called(dig2src)
 	if args.Get(1) == nil {
-		return args.Get(0).([]*proto.PvtDataElement), nil
+		return args.Get(0).(*FetchedPvtDataContainer), nil
 	}
 	return nil, args.Get(1).(error)
 }
@@ -821,15 +820,17 @@ func TestCoordinatorToFilterOutPvtRWSetsWithWrongHash(t *testing.T) {
 		{
 			TxId: "tx1", Namespace: "ns1", Collection: "c1", BlockSeq: 1,
 		},
-	}).expectingEndorsers("org1").Return([]*proto.PvtDataElement{
-		{
-			Digest: &proto.PvtDataDigest{
-				BlockSeq:   1,
-				Collection: "c1",
-				Namespace:  "ns1",
-				TxId:       "tx1",
+	}).expectingEndorsers("org1").Return(&FetchedPvtDataContainer{
+		AvailableElemenets: []*proto.PvtDataElement{
+			{
+				Digest: &proto.PvtDataDigest{
+					BlockSeq:   1,
+					Collection: "c1",
+					Namespace:  "ns1",
+					TxId:       "tx1",
+				},
+				Payload: [][]byte{[]byte("rws-original")},
 			},
-			Payload: [][]byte{[]byte("rws-original")},
 		},
 	}, nil)
 	store.On("Persist", mock.Anything, uint64(1), mock.Anything).
@@ -945,25 +946,27 @@ func TestCoordinatorStoreBlock(t *testing.T) {
 		{
 			TxId: "tx2", Namespace: "ns2", Collection: "c1", BlockSeq: 1, SeqInBlock: 1,
 		},
-	}).expectingEndorsers("org1").Return([]*proto.PvtDataElement{
-		{
-			Digest: &proto.PvtDataDigest{
-				BlockSeq:   1,
-				Collection: "c2",
-				Namespace:  "ns1",
-				TxId:       "tx1",
+	}).expectingEndorsers("org1").Return(&FetchedPvtDataContainer{
+		AvailableElemenets: []*proto.PvtDataElement{
+			{
+				Digest: &proto.PvtDataDigest{
+					BlockSeq:   1,
+					Collection: "c2",
+					Namespace:  "ns1",
+					TxId:       "tx1",
+				},
+				Payload: [][]byte{[]byte("rws-pre-image")},
 			},
-			Payload: [][]byte{[]byte("rws-pre-image")},
-		},
-		{
-			Digest: &proto.PvtDataDigest{
-				SeqInBlock: 1,
-				BlockSeq:   1,
-				Collection: "c1",
-				Namespace:  "ns2",
-				TxId:       "tx2",
+			{
+				Digest: &proto.PvtDataDigest{
+					SeqInBlock: 1,
+					BlockSeq:   1,
+					Collection: "c1",
+					Namespace:  "ns2",
+					TxId:       "tx2",
+				},
+				Payload: [][]byte{[]byte("rws-pre-image")},
 			},
-			Payload: [][]byte{[]byte("rws-pre-image")},
 		},
 	}, nil)
 	store.On("Persist", mock.Anything, uint64(1), mock.Anything).
@@ -994,15 +997,17 @@ func TestCoordinatorStoreBlock(t *testing.T) {
 		{
 			TxId: "tx3", Namespace: "ns3", Collection: "c3", BlockSeq: 1,
 		},
-	}).Return([]*proto.PvtDataElement{
-		{
-			Digest: &proto.PvtDataDigest{
-				BlockSeq:   1,
-				Collection: "c3",
-				Namespace:  "ns3",
-				TxId:       "tx3",
+	}).Return(&FetchedPvtDataContainer{
+		AvailableElemenets: []*proto.PvtDataElement{
+			{
+				Digest: &proto.PvtDataDigest{
+					BlockSeq:   1,
+					Collection: "c3",
+					Namespace:  "ns3",
+					TxId:       "tx3",
+				},
+				Payload: [][]byte{[]byte("rws-pre-image")},
 			},
-			Payload: [][]byte{[]byte("rws-pre-image")},
 		},
 	}, nil)
 	store = &mockTransientStore{t: t}
@@ -1123,15 +1128,17 @@ func TestProceedWithoutPrivateData(t *testing.T) {
 		{
 			TxId: "tx1", Namespace: "ns3", Collection: "c2", BlockSeq: 1,
 		},
-	}).Return([]*proto.PvtDataElement{
-		{
-			Digest: &proto.PvtDataDigest{
-				BlockSeq:   1,
-				Collection: "c2",
-				Namespace:  "ns3",
-				TxId:       "tx1",
+	}).Return(&FetchedPvtDataContainer{
+		AvailableElemenets: []*proto.PvtDataElement{
+			{
+				Digest: &proto.PvtDataDigest{
+					BlockSeq:   1,
+					Collection: "c2",
+					Namespace:  "ns3",
+					TxId:       "tx1",
+				},
+				Payload: [][]byte{[]byte("wrong pre-image")},
 			},
-			Payload: [][]byte{[]byte("wrong pre-image")},
 		},
 	}, nil)