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

Author: denyeart

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
 }