rollback: fix capability usage at gossip

Currently, we pass a static capability to the gossip which uses it to check whether to store the pvtData of invalid tx. With a channel update, the capability does not get updated. Hence, we need to use the channel with bundle listener so that the capability gets updated after a commit of a config update transaction. FAB-16114 #done Change-Id: I19d769432ce5783a1afdded5e7d4be9d682c2672 Signed-off-by: senthil <[email protected]> Signed-off-by: manish <[email protected]>
hyperledger · Aug 1, 2019 · ad802f9 · ad802f9
1 parent ea4fb38
commit ad802f9
Show file tree

Hide file tree

Showing 9 changed files with 455 additions and 192 deletions.
diff --git a/core/peer/peer.go b/core/peer/peer.go
@@ -414,12 +414,15 @@ func createChain(cid string, ledger ledger.PeerLedger, cb *common.Block, ccp ccp
 		cs.Resources = bundle
 	}
 
+	cp := &capabilityProvider{}
+
 	cs.bundleSource = channelconfig.NewBundleSource(
 		bundle,
 		gossipCallbackWrapper,
 		trustedRootsCallbackWrapper,
 		mspCallback,
 		peerSingletonCallback,
+		cp.updateChannelConfig,
 	)
 
 	vcs := struct {
@@ -460,6 +463,7 @@ func createChain(cid string, ledger ledger.PeerLedger, cb *common.Block, ccp ccp
 	if err != nil {
 		return errors.Wrapf(err, "[channel %s] failed opening transient store", bundle.ConfigtxValidator().ChainID())
 	}
+
 	csStoreSupport := &CollectionSupport{
 		PeerLedger: ledger,
 	}
@@ -476,7 +480,7 @@ func createChain(cid string, ledger ledger.PeerLedger, cb *common.Block, ccp ccp
 		Store:                store,
 		Cs:                   simpleCollectionStore,
 		IdDeserializeFactory: csStoreSupport,
-		Capabilities:         cs.Application.Capabilities(),
+		CapabilityProvider:   cp,
 	})
 
 	chains.Lock()
@@ -843,3 +847,24 @@ func (*configSupport) GetChannelConfig(channel string) cc.Config {
 	}
 	return chain.cs.bundleSource.ConfigtxValidator()
 }
+
+type capabilityProvider struct {
+	lock   sync.Mutex
+	bundle *channelconfig.Bundle
+}
+
+func (cp *capabilityProvider) Capabilities() channelconfig.ApplicationCapabilities {
+	cp.lock.Lock()
+	defer cp.lock.Unlock()
+	ac, ok := cp.bundle.ApplicationConfig()
+	if !ok {
+		return nil
+	}
+	return ac.Capabilities()
+}
+
+func (cp *capabilityProvider) updateChannelConfig(bundle *channelconfig.Bundle) {
+	cp.lock.Lock()
+	defer cp.lock.Unlock()
+	cp.bundle = bundle
+}
diff --git a/gossip/mocks/app_capabilities.go b/gossip/mocks/app_capabilities.go
diff --git a/gossip/privdata/coordinator.go b/gossip/privdata/coordinator.go
@@ -13,6 +13,7 @@ import (
 	"time"
 
 	"github.com/golang/protobuf/proto"
+	"github.com/hyperledger/fabric/common/channelconfig"
 	vsccErrors "github.com/hyperledger/fabric/common/errors"
 	util2 "github.com/hyperledger/fabric/common/util"
 	"github.com/hyperledger/fabric/core/committer"
@@ -39,7 +40,6 @@ var logger = util.GetLogger(util.PrivateDataLogger, "")
 
 //go:generate mockery -dir ../../core/common/privdata/ -name CollectionStore -case underscore -output mocks/
 //go:generate mockery -dir ../../core/committer/ -name Committer -case underscore -output mocks/
-//go:generate mockery -dir ./ -name AppCapabilities -case underscore -output ../mocks/
 
 // TransientStore holds private data that the corresponding blocks haven't been committed yet into the ledger
 type TransientStore interface {
@@ -106,11 +106,12 @@ type Fetcher interface {
 	fetch(dig2src dig2sources) (*privdatacommon.FetchedPvtDataContainer, error)
 }
 
-// AppCapabilities defines the capabilities for the application portion of a channel
-type AppCapabilities interface {
-	// StorePvtDataOfInvalidTx() returns true if the peer needs to store the pvtData of
-	// invalid transactions.
-	StorePvtDataOfInvalidTx() bool
+//go:generate mockery -dir ./ -name CapabilityProvider -case underscore -output mocks/
+
+// CapabilityProvider contains functions to retrieve capability information for a channel
+type CapabilityProvider interface {
+	// Capabilities defines the capabilities for the application portion of this channel
+	Capabilities() channelconfig.ApplicationCapabilities
 }
 
 // Support encapsulates set of interfaces to
@@ -122,7 +123,7 @@ type Support struct {
 	committer.Committer
 	TransientStore
 	Fetcher
-	AppCapabilities
+	CapabilityProvider
 }
 
 type coordinator struct {
@@ -705,7 +706,8 @@ func (c *coordinator) listMissingPrivateData(block *common.Block, ownedRWsets ma
 		privateRWsetsInBlock: privateRWsetsInBlock,
 		coordinator:          c,
 	}
-	txList, err := data.forEachTxn(c.Support.StorePvtDataOfInvalidTx(), txsFilter, bi.inspectTransaction)
+	storePvtDataOfInvalidTx := c.Support.CapabilityProvider.Capabilities().StorePvtDataOfInvalidTx()
+	txList, err := data.forEachTxn(storePvtDataOfInvalidTx, txsFilter, bi.inspectTransaction)
 	if err != nil {
 		return nil, err
 	}
@@ -882,7 +884,8 @@ func (c *coordinator) GetPvtDataAndBlockByNum(seqNum uint64, peerAuthInfo common
 
 	seqs2Namespaces := aggregatedCollections(make(map[seqAndDataModel]map[string][]*rwset.CollectionPvtReadWriteSet))
 	data := blockData(blockAndPvtData.Block.Data.Data)
-	data.forEachTxn(c.Support.StorePvtDataOfInvalidTx(), make(txValidationFlags, len(data)),
+	storePvtDataOfInvalidTx := c.Support.CapabilityProvider.Capabilities().StorePvtDataOfInvalidTx()
+	data.forEachTxn(storePvtDataOfInvalidTx, make(txValidationFlags, len(data)),
 		func(seqInBlock uint64, chdr *common.ChannelHeader, txRWSet *rwsetutil.TxRwSet, _ []*peer.Endorsement) error {
 			item, exists := blockAndPvtData.PvtData[seqInBlock]
 			if !exists {