[FAB-7676] Ignore read-only collections preimage pull

Currently, when the coordinator parses the block, it doesn't distinguish
between collections that have no private writes and collections that do.

This results in gossip attempting to retrieve from the transient store, or from
other peers - preimages for read-only collection RWSets, which don't exist anywhere
because they're not persisted in the transient store, and as a result - the attempt
times out.

This change set makes the logic skip collections that are read-only.

Change-Id: If41d2e0ebe093d567c480d17d9b7e26263659498
Signed-off-by: yacovm <yacovm@il.ibm.com>

Author: yacovm

gossip/privdata/coordinator.go

@@ -665,20 +665,23 @@ type transactionInspector struct {
 
 func (bi *transactionInspector) inspectTransaction(seqInBlock uint64, chdr *common.ChannelHeader, txRWSet *rwsetutil.TxRwSet, endorsers []*peer.Endorsement) {
 	for _, ns := range txRWSet.NsRwSets {
-		for _, hashed := range ns.CollHashedRwSets {
-			policy := bi.accessPolicyForCollection(chdr, ns.NameSpace, hashed.CollectionName)
+		for _, hashedCollection := range ns.CollHashedRwSets {
+			if !containsWrites(chdr.TxId, ns.NameSpace, hashedCollection) {
+				continue
+			}
+			policy := bi.accessPolicyForCollection(chdr, ns.NameSpace, hashedCollection.CollectionName)
 			if policy == nil {
 				continue
 			}
-			if !bi.isEligible(policy, ns.NameSpace, hashed.CollectionName) {
+			if !bi.isEligible(policy, ns.NameSpace, hashedCollection.CollectionName) {
 				continue
 			}
 			key := rwSetKey{
 				txID:       chdr.TxId,
 				seqInBlock: seqInBlock,
-				hash:       hex.EncodeToString(hashed.PvtRwSetHash),
+				hash:       hex.EncodeToString(hashedCollection.PvtRwSetHash),
 				namespace:  ns.NameSpace,
-				collection: hashed.CollectionName,
+				collection: hashedCollection.CollectionName,
 			}
 			bi.privateRWsetsInBlock[key] = struct{}{}
 			if _, exists := bi.ownedRWsets[key]; !exists {
@@ -687,7 +690,7 @@ func (bi *transactionInspector) inspectTransaction(seqInBlock uint64, chdr *comm
 					seqInBlock: seqInBlock,
 				}
 				bi.missingKeys[txAndSeq] = append(bi.missingKeys[txAndSeq], key)
-				bi.sources[key] = endorsersFromOrgs(ns.NameSpace, hashed.CollectionName, endorsers, policy.MemberOrgs())
+				bi.sources[key] = endorsersFromOrgs(ns.NameSpace, hashedCollection.CollectionName, endorsers, policy.MemberOrgs())
 			}
 		} // for all hashed RW sets
 	} // for all RW sets
@@ -814,3 +817,16 @@ func (c *coordinator) GetPvtDataAndBlockByNum(seqNum uint64, peerAuthInfo common
 
 	return blockAndPvtData.Block, seqs2Namespaces.asPrivateData(), nil
 }
+
+// containsWrites checks whether the given CollHashedRwSet contains writes
+func containsWrites(txID string, namespace string, colHashedRWSet *rwsetutil.CollHashedRwSet) bool {
+	if colHashedRWSet.HashedRwSet == nil {
+		logger.Warningf("HashedRWSet of tx %s, namespace %s, collection %s is nil", txID, namespace, colHashedRWSet.CollectionName)
+		return false
+	}
+	if len(colHashedRWSet.HashedRwSet.HashedWrites) == 0 {
+		logger.Debugf("HashedRWSet of tx %s, namespace %s, collection %s doesn't contain writes", txID, namespace, colHashedRWSet.CollectionName)
+		return false
+	}
+	return true
+}

gossip/privdata/coordinator_test.go

@@ -19,11 +19,13 @@ import (
 	util2 "github.com/hyperledger/fabric/common/util"
 	"github.com/hyperledger/fabric/core/common/privdata"
 	"github.com/hyperledger/fabric/core/ledger"
+	"github.com/hyperledger/fabric/core/ledger/kvledger/txmgmt/rwsetutil"
 	"github.com/hyperledger/fabric/core/transientstore"
 	"github.com/hyperledger/fabric/gossip/util"
 	"github.com/hyperledger/fabric/protos/common"
 	proto "github.com/hyperledger/fabric/protos/gossip"
 	"github.com/hyperledger/fabric/protos/ledger/rwset"
+	"github.com/hyperledger/fabric/protos/ledger/rwset/kvrwset"
 	"github.com/hyperledger/fabric/protos/msp"
 	"github.com/spf13/viper"
 	"github.com/stretchr/testify/assert"
@@ -748,7 +750,7 @@ func TestCoordinatorToFilterOutPvtRWSetsWithWrongHash(t *testing.T) {
 		channelID: "test",
 	}
 
-	block := bf.AddTxnWithEndorsement("tx1", "ns1", hash, "org1", "c1").create()
+	block := bf.AddTxnWithEndorsement("tx1", "ns1", hash, "org1", true, "c1").create()
 	store.On("GetTxPvtRWSetByTxid", "tx1", mock.Anything).Return((&mockRWSetScanner{}).withRWSet("ns1", "c1"), nil)
 
 	coordinator := NewCoordinator(Support{
@@ -837,7 +839,8 @@ func TestCoordinatorStoreBlock(t *testing.T) {
 	bf := &blockFactory{
 		channelID: "test",
 	}
-	block := bf.AddTxnWithEndorsement("tx1", "ns1", hash, "org1", "c1", "c2").AddTxnWithEndorsement("tx2", "ns2", hash, "org2", "c1").create()
+	block := bf.AddTxnWithEndorsement("tx1", "ns1", hash, "org1", true, "c1", "c2").
+		AddTxnWithEndorsement("tx2", "ns2", hash, "org2", true, "c1").create()
 
 	// Scenario I: Block we got has sufficient private data alongside it.
 	// If the coordinator tries fetching from the transientstore, or peers it would result in panic,
@@ -1223,3 +1226,71 @@ func TestCoordinatorStorePvtData(t *testing.T) {
 	assert.Error(t, err)
 	assert.Contains(t, err.Error(), "I/O error: file system full")
 }
+
+func TestContainsWrites(t *testing.T) {
+	// Scenario I: Nil HashedRwSet in collection
+	col := &rwsetutil.CollHashedRwSet{
+		CollectionName: "col1",
+	}
+	assert.False(t, containsWrites("tx", "ns", col))
+
+	// Scenario II: No writes in collection
+	col.HashedRwSet = &kvrwset.HashedRWSet{}
+	assert.False(t, containsWrites("tx", "ns", col))
+
+	// Scenario III: Some writes in collection
+	col.HashedRwSet.HashedWrites = append(col.HashedRwSet.HashedWrites, &kvrwset.KVWriteHash{})
+	assert.True(t, containsWrites("tx", "ns", col))
+}
+
+func TestIgnoreReadOnlyColRWSets(t *testing.T) {
+	// Scenario: The transaction has some ColRWSets that have only reads and no writes,
+	// These should be ignored and not considered as missing private data that needs to be retrieved
+	// from the transient store or other peers.
+	// The gossip and transient store mocks in this test aren't initialized with
+	// actions, so if the coordinator attempts to fetch private data from the
+	// transient store or other peers, the test would fail.
+	// Also - we check that at commit time - the coordinator concluded that
+	// no missing private data was found.
+	peerSelfSignedData := common.SignedData{
+		Identity:  []byte{0, 1, 2},
+		Signature: []byte{3, 4, 5},
+		Data:      []byte{6, 7, 8},
+	}
+	cs := createcollectionStore(peerSelfSignedData).thatAcceptsAll()
+	var commitHappened bool
+	assertCommitHappened := func() {
+		assert.True(t, commitHappened)
+		commitHappened = false
+	}
+	committer := &committerMock{}
+	committer.On("CommitWithPvtData", mock.Anything).Run(func(args mock.Arguments) {
+		blockAndPrivateData := args.Get(0).(*ledger.BlockAndPvtData)
+		// Ensure there is no private data to commit
+		assert.Empty(t, blockAndPrivateData.BlockPvtData)
+		// Ensure there is no missing private data
+		assert.Empty(t, blockAndPrivateData.Missing)
+		commitHappened = true
+	}).Return(nil)
+	store := &mockTransientStore{t: t}
+
+	fetcher := &fetcherMock{t: t}
+	hash := util2.ComputeSHA256([]byte("rws-pre-image"))
+	bf := &blockFactory{
+		channelID: "test",
+	}
+	// The block contains a read only private data transaction
+	block := bf.AddReadOnlyTxn("tx1", "ns3", hash, "c3", "c2").create()
+	coordinator := NewCoordinator(Support{
+		CollectionStore: cs,
+		Committer:       committer,
+		Fetcher:         fetcher,
+		TransientStore:  store,
+		Validator:       &validatorMock{},
+	}, peerSelfSignedData)
+	// We pass a nil private data slice to indicate no pre-images though the block contains
+	// private data reads.
+	err := coordinator.StoreBlock(block, nil)
+	assert.NoError(t, err)
+	assertCommitHappened()
+}

gossip/privdata/util.go

@@ -31,17 +31,25 @@ type blockFactory struct {
 }
 
 func (bf *blockFactory) AddTxn(txID string, nsName string, hash []byte, collections ...string) *blockFactory {
-	return bf.AddTxnWithEndorsement(txID, nsName, hash, "", collections...)
+	return bf.AddTxnWithEndorsement(txID, nsName, hash, "", true, collections...)
 }
 
-func (bf *blockFactory) AddTxnWithEndorsement(txID string, nsName string, hash []byte, org string, collections ...string) *blockFactory {
+func (bf *blockFactory) AddReadOnlyTxn(txID string, nsName string, hash []byte, collections ...string) *blockFactory {
+	return bf.AddTxnWithEndorsement(txID, nsName, hash, "", false, collections...)
+}
+
+func (bf *blockFactory) AddTxnWithEndorsement(txID string, nsName string, hash []byte, org string, hasWrites bool, collections ...string) *blockFactory {
 	txn := &peer.Transaction{
 		Actions: []*peer.TransactionAction{
 			{},
 		},
 	}
+	nsRWSet := sampleNsRwSet(nsName, hash, collections...)
+	if !hasWrites {
+		nsRWSet = sampleReadOnlyNsRwSet(nsName, hash, collections...)
+	}
 	txrws := rwsetutil.TxRwSet{
-		NsRwSets: []*rwsetutil.NsRwSet{sampleNsRwSet(nsName, hash, collections...)},
+		NsRwSets: []*rwsetutil.NsRwSet{nsRWSet},
 	}
 
 	b, err := txrws.ToProtoBytes()
@@ -167,7 +175,17 @@ func sampleNsRwSet(ns string, hash []byte, collections ...string) *rwsetutil.NsR
 		KvRwSet: sampleKvRwSet(),
 	}
 	for _, col := range collections {
-		nsRwSet.CollHashedRwSets = append(nsRwSet.CollHashedRwSets, sampleCollHashedRwSet(col, hash))
+		nsRwSet.CollHashedRwSets = append(nsRwSet.CollHashedRwSets, sampleCollHashedRwSet(col, hash, true))
+	}
+	return nsRwSet
+}
+
+func sampleReadOnlyNsRwSet(ns string, hash []byte, collections ...string) *rwsetutil.NsRwSet {
+	nsRwSet := &rwsetutil.NsRwSet{NameSpace: ns,
+		KvRwSet: sampleKvRwSet(),
+	}
+	for _, col := range collections {
+		nsRwSet.CollHashedRwSets = append(nsRwSet.CollHashedRwSets, sampleCollHashedRwSet(col, hash, false))
 	}
 	return nsRwSet
 }
@@ -188,21 +206,23 @@ func sampleKvRwSet() *kvrwset.KVRWSet {
 	}
 }
 
-func sampleCollHashedRwSet(collectionName string, hash []byte) *rwsetutil.CollHashedRwSet {
+func sampleCollHashedRwSet(collectionName string, hash []byte, hasWrites bool) *rwsetutil.CollHashedRwSet {
 	collHashedRwSet := &rwsetutil.CollHashedRwSet{
 		CollectionName: collectionName,
 		HashedRwSet: &kvrwset.HashedRWSet{
 			HashedReads: []*kvrwset.KVReadHash{
 				{KeyHash: []byte("Key-1-hash"), Version: &kvrwset.Version{BlockNum: 1, TxNum: 2}},
 				{KeyHash: []byte("Key-2-hash"), Version: &kvrwset.Version{BlockNum: 2, TxNum: 3}},
 			},
-			HashedWrites: []*kvrwset.KVWriteHash{
-				{KeyHash: []byte("Key-3-hash"), ValueHash: []byte("value-3-hash"), IsDelete: false},
-				{KeyHash: []byte("Key-4-hash"), ValueHash: []byte("value-4-hash"), IsDelete: true},
-			},
 		},
 		PvtRwSetHash: hash,
 	}
+	if hasWrites {
+		collHashedRwSet.HashedRwSet.HashedWrites = []*kvrwset.KVWriteHash{
+			{KeyHash: []byte("Key-3-hash"), ValueHash: []byte("value-3-hash"), IsDelete: false},
+			{KeyHash: []byte("Key-4-hash"), ValueHash: []byte("value-4-hash"), IsDelete: true},
+		}
+	}
 	return collHashedRwSet
 }