add config-puller delivery client to router, and pull config blocks for consensus

add new config blocks to the config-store

add test

Signed-off-by: Dor.Katzelnick <Dor.Katzelnick@ibm.com>

Author: DorKatzelnick

node/delivery/consensus_config_deliver_client.go

@@ -0,0 +1,97 @@
+/*
+Copyright IBM Corp. All Rights Reserved.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package delivery
+
+import (
+	"context"
+
+	"github.com/hyperledger/fabric-protos-go-apiv2/common"
+	"github.com/hyperledger/fabric-protos-go-apiv2/orderer"
+	"github.com/hyperledger/fabric-x-orderer/common/types"
+	"github.com/hyperledger/fabric-x-orderer/node/comm"
+	node_config "github.com/hyperledger/fabric-x-orderer/node/config"
+	"github.com/hyperledger/fabric/protoutil"
+)
+
+const (
+	configBlocksChanSize = 100
+)
+
+// ConsensusConfigPuller pulls decisions from consensus and consumes config blocks.
+type ConsensusConfigPuller struct {
+	tlsKey, tlsCert []byte
+	endpoint        string
+	cc              comm.ClientConfig
+	logger          types.Logger
+	seekInfo        *orderer.SeekInfo
+	cancelCtx       context.Context
+	ctxCancelFunc   context.CancelFunc
+}
+
+func NewConsensusConfigPuller(config *node_config.RouterNodeConfig, logger types.Logger, seekInfo *orderer.SeekInfo) *ConsensusConfigPuller {
+	ctx, cancelFunc := context.WithCancel(context.Background())
+	configPuller := &ConsensusConfigPuller{
+		cc:            clientConfig(config.Consenter.TLSCACerts, config.TLSPrivateKeyFile, config.TLSCertificateFile),
+		endpoint:      config.Consenter.Endpoint,
+		logger:        logger,
+		seekInfo:      seekInfo,
+		tlsKey:        config.TLSPrivateKeyFile,
+		tlsCert:       config.TLSCertificateFile,
+		cancelCtx:     ctx,
+		ctxCancelFunc: cancelFunc,
+	}
+	return configPuller
+}
+
+func (ccp *ConsensusConfigPuller) PullConfigBlocks() <-chan *common.Block {
+	endpoint := func() string {
+		return ccp.endpoint
+	}
+
+	requestEnvelopeFactoryFunc := func() *common.Envelope {
+		requestEnvelope, err := protoutil.CreateSignedEnvelopeWithTLSBinding(
+			common.HeaderType_DELIVER_SEEK_INFO,
+			"consensus",
+			nil,
+			ccp.seekInfo,
+			int32(0),
+			uint64(0),
+			nil,
+		)
+		if err != nil {
+			ccp.logger.Panicf("Failed creating signed envelope: %v", err)
+		}
+
+		return requestEnvelope
+	}
+
+	res := make(chan *common.Block, configBlocksChanSize)
+
+	blockHandlerFunc := func(block *common.Block) {
+		// check if the decision contains a config block. then extract the config block and send it on the res channel
+		header, _, err := extractHeaderAndSigsFromBlock(block)
+		if err != nil {
+			ccp.logger.Panicf("Failed extracting header from decision: %s", err)
+		}
+
+		if header.Num == header.DecisionNumOfLastConfigBlock {
+			res <- header.AvailableCommonBlocks[len(header.AvailableCommonBlocks)-1]
+		}
+	}
+
+	onClose := func() {
+		close(res)
+	}
+
+	go Pull(ccp.cancelCtx, "consensus-configBlock-pull", ccp.logger, endpoint, requestEnvelopeFactoryFunc, ccp.cc, blockHandlerFunc, onClose)
+
+	return res
+}
+
+func (ccp *ConsensusConfigPuller) Stop() {
+	ccp.ctxCancelFunc()
+}

node/router/router.go

@@ -18,6 +18,7 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/hyperledger/fabric-protos-go-apiv2/common"
 	"github.com/hyperledger/fabric-protos-go-apiv2/orderer"
 	"github.com/hyperledger/fabric-x-common/common/policies"
 	"github.com/hyperledger/fabric-x-orderer/common/configstore"
@@ -26,6 +27,8 @@ import (
 	"github.com/hyperledger/fabric-x-orderer/config"
 	"github.com/hyperledger/fabric-x-orderer/node"
 	nodeconfig "github.com/hyperledger/fabric-x-orderer/node/config"
+	"github.com/hyperledger/fabric-x-orderer/node/delivery"
+	"github.com/hyperledger/fabric-x-orderer/node/ledger"
 	protos "github.com/hyperledger/fabric-x-orderer/node/protos/comm"
 )
 
@@ -34,6 +37,11 @@ type Net interface {
 	Address() string
 }
 
+type ConfigPuller interface {
+	PullConfigBlocks() <-chan *common.Block
+	Stop()
+}
+
 type Router struct {
 	mapper           ShardMapper
 	net              Net
@@ -46,6 +54,7 @@ type Router struct {
 	stopChan         chan struct{}
 	configStore      *configstore.Store
 	configSubmitter  ConfigurationSubmitter
+	configPuller     ConfigPuller
 }
 
 func NewRouter(config *nodeconfig.RouterNodeConfig, logger types.Logger) *Router {
@@ -81,13 +90,43 @@ func NewRouter(config *nodeconfig.RouterNodeConfig, logger types.Logger) *Router
 	configSubmitter := NewConfigSubmitter(config.Consenter.Endpoint, tlsCAsOfConsenter,
 		config.TLSCertificateFile, config.TLSPrivateKeyFile, logger)
 
+	configStore, err := configstore.NewStore(config.ConfigStorePath)
+	if err != nil {
+		logger.Panicf("Failed creating router config store: %s", err)
+	}
+
+	seekInfo := NextSeekInfoFromConfigStore(configStore, logger)
+
+	// TODO - pull cofig blocks from all consenter nodes, not only the one in party
+	configPuller := delivery.NewConsensusConfigPuller(config, logger, seekInfo)
+
 	verifier := createVerifier(config)
 
-	r := createRouter(shardIDs, batcherEndpoints, tlsCAsOfBatchers, config, logger, verifier, configSubmitter)
+	r := createRouter(shardIDs, batcherEndpoints, tlsCAsOfBatchers, config, logger, configStore, verifier, configSubmitter, configPuller)
 	r.init()
 	return r
 }
 
+// NextSeekInfoFromConfigStore creates a SeekInfo to start pulling config blocks from consensus, based on the last connfig block stored in the config store.
+func NextSeekInfoFromConfigStore(configStore *configstore.Store, logger types.Logger) *orderer.SeekInfo {
+	lastBlock, err := configStore.Last()
+	if err != nil {
+		logger.Panicf("Failed getting last config block from config store: %s", err)
+	}
+
+	// check if last block is genesis block
+	if lastBlock.GetHeader().GetNumber() == 0 {
+		return delivery.NextSeekInfo(1)
+	}
+
+	ordererBlockMetadata := lastBlock.Metadata.Metadata[common.BlockMetadataIndex_ORDERER]
+	_, _, _, lastDecisionNumber, _, _, _, err := ledger.AssemblerBlockMetadataFromBytes(ordererBlockMetadata)
+	if err != nil {
+		logger.Panicf("Failed extracting decision number from last config block: %s", err)
+	}
+	return delivery.NextSeekInfo(uint64(lastDecisionNumber) + 1)
+}
+
 func (r *Router) StartRouterService() <-chan struct{} {
 	srv := node.CreateGRPCRouter(r.routerNodeConfig)
 
@@ -108,6 +147,8 @@ func (r *Router) StartRouterService() <-chan struct{} {
 
 	r.configSubmitter.Start()
 
+	go r.pullAndProcessConfigBlocks()
+
 	return stop
 }
 
@@ -184,7 +225,7 @@ func (r *Router) Deliver(server orderer.AtomicBroadcast_DeliverServer) error {
 	return fmt.Errorf("not implemented")
 }
 
-func createRouter(shardIDs []types.ShardID, batcherEndpoints map[types.ShardID]string, batcherRootCAs map[types.ShardID][][]byte, rconfig *nodeconfig.RouterNodeConfig, logger types.Logger, verifier *requestfilter.RulesVerifier, configSubmitter ConfigurationSubmitter) *Router {
+func createRouter(shardIDs []types.ShardID, batcherEndpoints map[types.ShardID]string, batcherRootCAs map[types.ShardID][][]byte, rconfig *nodeconfig.RouterNodeConfig, logger types.Logger, configStore *configstore.Store, verifier *requestfilter.RulesVerifier, configSubmitter ConfigurationSubmitter, configPuller ConfigPuller) *Router {
 	if rconfig.NumOfConnectionsForBatcher == 0 {
 		rconfig.NumOfConnectionsForBatcher = config.DefaultRouterParams.NumberOfConnectionsPerBatcher
 	}
@@ -193,11 +234,6 @@ func createRouter(shardIDs []types.ShardID, batcherEndpoints map[types.ShardID]s
 		rconfig.NumOfgRPCStreamsPerConnection = config.DefaultRouterParams.NumberOfStreamsPerConnection
 	}
 
-	configStore, err := configstore.NewStore(rconfig.ConfigStorePath)
-	if err != nil {
-		logger.Panicf("Failed creating router config store: %s", err)
-	}
-
 	r := &Router{
 		mapper: MapperCRC64{
 			Logger:     logger,
@@ -211,6 +247,7 @@ func createRouter(shardIDs []types.ShardID, batcherEndpoints map[types.ShardID]s
 		stopChan:         make(chan struct{}),
 		configStore:      configStore,
 		configSubmitter:  configSubmitter,
+		configPuller:     configPuller,
 	}
 
 	for _, shardId := range shardIDs {
@@ -363,6 +400,37 @@ func createVerifier(config *nodeconfig.RouterNodeConfig) *requestfilter.RulesVer
 	return rv
 }
 
+// pullAndProcessConfigBlocks pulls config blocks from consensus and processes them. this function should be run as a goroutine.
+func (r *Router) pullAndProcessConfigBlocks() {
+	configBlocksChan := r.configPuller.PullConfigBlocks()
+	defer func() {
+		r.configPuller.Stop()
+		r.logger.Infof("Stopped config puller")
+	}()
+
+	for {
+		select {
+		case configBlock, ok := <-configBlocksChan:
+			if !ok {
+				r.logger.Infof("Config blocks channel closed, stopping config blocks processing")
+				return
+			}
+			r.logger.Infof("Received new config block from consensus with block number %d", configBlock.GetHeader().GetNumber())
+
+			// TODO process the config block. store in config store and apply.
+			if err := r.configStore.Add(configBlock); err != nil {
+				r.logger.Errorf("Failed adding config block to config store: %s", err)
+				continue
+			}
+			r.logger.Infof("Added config block %d to config store", configBlock.GetHeader().GetNumber())
+
+		case <-r.stopChan:
+			r.logger.Infof("Stopping config blocks processing")
+			return
+		}
+	}
+}
+
 // IsAllStreamsOK checks that all the streams accross all shard-routers are non-faulty.
 // Use for testing only.
 func (r *Router) IsAllStreamsOK() bool {
@@ -384,3 +452,13 @@ func (r *Router) IsAllConnectionsDown() bool {
 	}
 	return true
 }
+
+// GetConfigStoreSize returns the number of config blocks stored in the config store.
+// Use for testing only.
+func (r *Router) GetConfigStoreSize() int {
+	list, err := r.configStore.ListBlockNumbers()
+	if err != nil {
+		r.logger.Panicf("Failed listing config store block numbers: %s", err)
+	}
+	return len(list)
+}

node/router/router_test.go

@@ -17,11 +17,13 @@ import (
 
 	"github.com/hyperledger/fabric-protos-go-apiv2/common"
 	ab "github.com/hyperledger/fabric-protos-go-apiv2/orderer"
+	"github.com/hyperledger/fabric-x-orderer/common/configstore"
 	policyMocks "github.com/hyperledger/fabric-x-orderer/common/policy/mocks"
 	"github.com/hyperledger/fabric-x-orderer/common/types"
 	"github.com/hyperledger/fabric-x-orderer/node/comm"
 	"github.com/hyperledger/fabric-x-orderer/node/comm/tlsgen"
 	"github.com/hyperledger/fabric-x-orderer/node/config"
+	"github.com/hyperledger/fabric-x-orderer/node/consensus/state"
 	protos "github.com/hyperledger/fabric-x-orderer/node/protos/comm"
 	"github.com/hyperledger/fabric-x-orderer/node/router"
 	configMocks "github.com/hyperledger/fabric-x-orderer/test/mocks"
@@ -59,7 +61,9 @@ func (r *routerTestSetup) Close() {
 		batcher.server.Stop()
 	}
 
-	r.consenter.Stop()
+	if r.consenter != nil {
+		r.consenter.Stop()
+	}
 }
 
 func (r *routerTestSetup) isReconnectComplete() bool {
@@ -86,6 +90,7 @@ func createRouterTestSetup(t *testing.T, partyID types.PartyID, numOfShards int,
 	for _, batcher := range batchers {
 		batcher.Start()
 	}
+
 	// create and start stub-consenter
 	stubConsenter := router.NewStubConsenter(t, ca, partyID)
 	stubConsenter.Start()
@@ -510,6 +515,39 @@ func submitConfigRequest(t *testing.T, conn *grpc.ClientConn) error {
 	return nil
 }
 
+func TestConfigPullFromConsensus(t *testing.T) {
+	testSetup := createRouterTestSetup(t, types.PartyID(1), 1, true, false)
+	err := createServerTLSClientConnection(testSetup, testSetup.ca)
+	require.NoError(t, err)
+	require.NotNil(t, testSetup.clientConn)
+	sc := testSetup.consenter
+	defer testSetup.Close()
+
+	initialConfigStoreSize := testSetup.router.GetConfigStoreSize()
+
+	// create a decision with a config block
+	configBlock := common.Block{Header: &common.BlockHeader{Number: 999}, Data: &common.BlockData{}}
+	acb := make([]*common.Block, 2)
+	acb[len(acb)-1] = &configBlock
+	err = sc.DeliverDecisionFromHeader(&state.Header{Num: 1, DecisionNumOfLastConfigBlock: 1, AvailableCommonBlocks: acb})
+	require.NoError(t, err)
+
+	// check if config block is stored in router's config store
+	require.Eventually(t, func() bool {
+		return testSetup.router.GetConfigStoreSize() == initialConfigStoreSize+1
+	}, 10*time.Second, 10*time.Millisecond)
+
+	// create a decision, with no config block
+	configBlock = common.Block{Header: &common.BlockHeader{Number: 1000}, Data: &common.BlockData{}}
+	acb = make([]*common.Block, 6)
+	acb[len(acb)-1] = &configBlock
+	err = sc.DeliverDecisionFromHeader(&state.Header{Num: 2, DecisionNumOfLastConfigBlock: 1, AvailableCommonBlocks: acb})
+	require.NoError(t, err)
+
+	time.Sleep(100 * time.Millisecond)
+	require.Equal(t, initialConfigStoreSize+1, testSetup.router.GetConfigStoreSize(), "no new config block should be stored")
+}
+
 func createServerTLSClientConnection(testSetup *routerTestSetup, ca tlsgen.CA) error {
 	cc := comm.ClientConfig{
 		SecOpts: comm.SecureOptions{
@@ -686,13 +724,17 @@ func createAndStartRouter(t *testing.T, partyID types.PartyID, ca tlsgen.CA, bat
 
 	stubConsenterInfo := config.ConsenterInfo{PartyID: partyID, Endpoint: consenter.GetConsenterEndpoint(), TLSCACerts: []config.RawBytes{ca.CertBytes()}}
 
+	configStorePath := t.TempDir()
+	cs, _ := configstore.NewStore(configStorePath)
+	cs.Add(&common.Block{Header: &common.BlockHeader{Number: 0}, Data: &common.BlockData{}}) // add dummy genesis block
+
 	conf := &config.RouterNodeConfig{
 		PartyID:                             partyID,
 		TLSCertificateFile:                  ckp.Cert,
 		UseTLS:                              useTLS,
 		TLSPrivateKeyFile:                   ckp.Key,
 		ListenAddress:                       "127.0.0.1:0",
-		ConfigStorePath:                     t.TempDir(),
+		ConfigStorePath:                     configStorePath,
 		ClientAuthRequired:                  clientAuthRequired,
 		Shards:                              shards,
 		Consenter:                           stubConsenterInfo,