check request type in stream, and then forward to batcher or config-submitter

add config submitter tests to router
fix some other tests

Signed-off-by: Dor.Katzelnick <[email protected]>

Author: DorKatzelnick

node/router/config_submitter.go

@@ -8,6 +8,7 @@ package router
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"time"
 
@@ -21,7 +22,7 @@ type ConfigurationSubmitter interface {
 	Start()
 	Stop()
 	// Update() // TODO implement a thread-safe update method for config submitter
-	Forward(tr *TrackedRequest) error
+	Forward(tr *TrackedRequest)
 }
 
 type configSubmitter struct {
@@ -102,6 +103,11 @@ func (cs *configSubmitter) forwardRequest(tr *TrackedRequest) error {
 		feedback.SubmitResponse = resp
 		if err != nil {
 			feedback.err = fmt.Errorf("error forwarding config request to consenter: %v", err)
+		} else {
+			feedback.SubmitResponse = resp
+			if resp.Error != "" {
+				feedback.err = errors.New(resp.Error)
+			}
 		}
 	}
 

node/router/config_submitter_test.go

@@ -19,7 +19,7 @@ import (
 )
 
 type configSubmitTestSetup struct {
-	stubConsenter   *stubConsenter
+	stubConsenter   *StubConsenter
 	configSubmitter *configSubmitter
 }
 
@@ -38,7 +38,7 @@ func createconfigSubmitTestSetup(t *testing.T) configSubmitTestSetup {
 	ca, err := tlsgen.NewCA()
 	require.NoError(t, err)
 
-	stubConsenter := NewStubConsenter(t, ca, types.PartyID(1), logger)
+	stubConsenter := NewStubConsenter(t, ca, types.PartyID(1))
 
 	ckp, err := ca.NewServerCertKeyPair("127.0.0.1")
 	require.NoError(t, err)

node/router/router.go

@@ -45,7 +45,7 @@ type Router struct {
 	verifier         *requestfilter.RulesVerifier
 	stopChan         chan struct{}
 	configStore      *configstore.Store
-	configSubmitter  *configSubmitter
+	configSubmitter  ConfigurationSubmitter
 }
 
 func NewRouter(config *nodeconfig.RouterNodeConfig, logger types.Logger) *Router {
@@ -184,7 +184,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 *configSubmitter) *Router {
+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 {
 	if rconfig.NumOfConnectionsForBatcher == 0 {
 		rconfig.NumOfConnectionsForBatcher = config.DefaultRouterParams.NumberOfConnectionsPerBatcher
 	}

node/router/router_test.go

@@ -41,6 +41,7 @@ func init() {
 type routerTestSetup struct {
 	ca         tlsgen.CA
 	batchers   []*stubBatcher
+	consenter  *router.StubConsenter
 	clientConn *grpc.ClientConn
 	router     *router.Router
 }
@@ -57,6 +58,8 @@ func (r *routerTestSetup) Close() {
 	for _, batcher := range r.batchers {
 		batcher.server.Stop()
 	}
+
+	r.consenter.Stop()
 }
 
 func (r *routerTestSetup) isReconnectComplete() bool {
@@ -83,14 +86,18 @@ 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()
 
 	// create and start router
-	router := createAndStartRouter(t, partyID, ca, batchers, useTLS, clientAuthRequired)
+	router := createAndStartRouter(t, partyID, ca, batchers, &stubConsenter, useTLS, clientAuthRequired)
 
 	return &routerTestSetup{
-		ca:       ca,
-		batchers: batchers,
-		router:   router,
+		ca:        ca,
+		batchers:  batchers,
+		consenter: &stubConsenter,
+		router:    router,
 	}
 }
 
@@ -428,6 +435,87 @@ func TestRequestFilters(t *testing.T) {
 	// 5) send request with invalid signature. Not implemented
 }
 
+// Scenario:
+// 1) Start a client, router and stub consenter
+// 2) Send valid config request, expect response from stub consenter
+func TestConfigSubmitter(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)
+
+	defer testSetup.Close()
+
+	err = submitConfigRequest(t, testSetup.clientConn)
+	require.NoError(t, err)
+
+	require.Eventually(t, func() bool {
+		return testSetup.consenter.ReceivedMessageCount() == uint32(1)
+	}, 10*time.Second, 10*time.Millisecond)
+}
+
+func TestConfigSubmitterConsenterDown(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)
+
+	defer testSetup.Close()
+
+	// submit one request, and wait for the response
+	err = submitConfigRequest(t, testSetup.clientConn)
+	require.NoError(t, err)
+
+	require.Eventually(t, func() bool {
+		return testSetup.consenter.ReceivedMessageCount() == uint32(1)
+	}, 10*time.Second, 10*time.Millisecond)
+
+	// stop the consenter and wait until it is down
+	testSetup.consenter.Stop()
+	time.Sleep(250 * time.Millisecond)
+
+	// wait and restart the consenter
+	go func() {
+		time.Sleep(250 * time.Millisecond)
+		testSetup.consenter.Restart()
+	}()
+
+	// meanwhile, forward another request
+	err = submitConfigRequest(t, testSetup.clientConn)
+	require.NoError(t, err)
+
+	require.Eventually(t, func() bool {
+		return testSetup.consenter.ReceivedMessageCount() == uint32(2)
+	}, 10*time.Second, 10*time.Millisecond)
+}
+
+func submitConfigRequest(t *testing.T, conn *grpc.ClientConn) error {
+	cl := ab.NewAtomicBroadcastClient(conn)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+
+	stream, err := cl.Broadcast(ctx)
+	require.NoError(t, err)
+	var wg sync.WaitGroup
+	wg.Add(1)
+
+	go func() {
+		defer wg.Done()
+		env := tx.CreateStructuredConfigEnvelope([]byte("123"))
+		err := stream.Send(env)
+		require.NoError(t, err)
+	}()
+
+	resp, err := stream.Recv()
+	require.NoError(t, err)
+	require.Equal(t, common.Status_INTERNAL_SERVER_ERROR, resp.Status)
+	require.Equal(t, "dummy submit config", resp.Info)
+	wg.Wait()
+
+	return nil
+}
+
 func createServerTLSClientConnection(testSetup *routerTestSetup, ca tlsgen.CA) error {
 	cc := comm.ClientConfig{
 		SecOpts: comm.SecureOptions{
@@ -585,7 +673,7 @@ func submitRequest(conn *grpc.ClientConn) error {
 	return nil
 }
 
-func createAndStartRouter(t *testing.T, partyID types.PartyID, ca tlsgen.CA, batchers []*stubBatcher, useTLS bool, clientAuthRequired bool) *router.Router {
+func createAndStartRouter(t *testing.T, partyID types.PartyID, ca tlsgen.CA, batchers []*stubBatcher, consenter *router.StubConsenter, useTLS bool, clientAuthRequired bool) *router.Router {
 	ckp, err := ca.NewServerCertKeyPair("127.0.0.1")
 	require.NoError(t, err)
 
@@ -602,6 +690,8 @@ func createAndStartRouter(t *testing.T, partyID types.PartyID, ca tlsgen.CA, bat
 	configtxValidator.ChannelIDReturns("arma")
 	bundle.ConfigtxValidatorReturns(configtxValidator)
 
+	stubConsenterInfo := config.ConsenterInfo{PartyID: partyID, Endpoint: consenter.GetConsenterEndpoint(), TLSCACerts: []config.RawBytes{ca.CertBytes()}}
+
 	conf := &config.RouterNodeConfig{
 		PartyID:                             partyID,
 		TLSCertificateFile:                  ckp.Cert,
@@ -611,6 +701,7 @@ func createAndStartRouter(t *testing.T, partyID types.PartyID, ca tlsgen.CA, bat
 		ConfigStorePath:                     t.TempDir(),
 		ClientAuthRequired:                  clientAuthRequired,
 		Shards:                              shards,
+		Consenter:                           stubConsenterInfo,
 		RequestMaxBytes:                     1 << 10,
 		ClientSignatureVerificationRequired: false,
 		Bundle:                              bundle,

node/router/shard_router.go

@@ -71,7 +71,7 @@ type ShardRouter struct {
 	reconnectRequests            chan reconnectReq
 	closeReconnect               chan bool
 	verifier                     *requestfilter.RulesVerifier
-	configSubmitter              *configSubmitter
+	configSubmitter              ConfigurationSubmitter
 }
 
 func NewShardRouter(l types.Logger,
@@ -82,7 +82,7 @@ func NewShardRouter(l types.Logger,
 	numOfConnectionsForBatcher int,
 	numOfgRPCStreamsPerConnection int,
 	verifier *requestfilter.RulesVerifier,
-	configSubmitter *configSubmitter,
+	configSubmitter ConfigurationSubmitter,
 ) *ShardRouter {
 	cc := comm.ClientConfig{
 		AsyncConnect: false,

node/router/stream.go

@@ -35,7 +35,7 @@ type stream struct {
 	srReconnectChan                   chan reconnectReq
 	notifiedReconnect                 bool
 	verifier                          *requestfilter.RulesVerifier
-	configSubmitter                   *configSubmitter
+	configSubmitter                   ConfigurationSubmitter
 }
 
 // readResponses listens for responses from the batcher.
@@ -83,26 +83,29 @@ func (s *stream) sendRequests() {
 				continue
 			}
 
-			// TODO: forward request to batcher or consenter
-			_ = reqType
+			if reqType == common.HeaderType_CONFIG_UPDATE {
+				s.logger.Debugf("received request with type %s, forwarding to consenter", reqType)
+				// forward to consenter. configSubmitter will handle sending response to client.
+				s.configSubmitter.Forward(tr)
+			} else {
+				s.logger.Debugf("received request with type %s, forwarding to batcher %s", reqType, s.endpoint)
+				err = s.requestTransmitSubmitStreamClient.Send(tr.request)
+				if err != nil {
+					s.logger.Errorf("Failed sending request to batcher %s", s.endpoint)
+					if tr.trace == nil {
+						// send error to client, in case request is not traced.
+						tr.responses <- Response{err: fmt.Errorf("server error: could not establish connection between router and batcher %s", s.endpoint)}
+					}
+					s.cancelOnServerError()
+					return
+				}
 
-			s.logger.Debugf("send request with trace id %x to batcher %s", tr.trace, s.endpoint)
-			err = s.requestTransmitSubmitStreamClient.Send(tr.request)
-			if err != nil {
-				s.logger.Errorf("Failed sending request to batcher %s", s.endpoint)
+				// send fast response to client for untraced requests.
+				// traced requests get their response from readResponses goroutine.
 				if tr.trace == nil {
-					// send error to client, in case request is not traced.
-					tr.responses <- Response{err: fmt.Errorf("server error: could not establish connection between router and batcher %s", s.endpoint)}
+					tr.responses <- Response{err: nil}
 				}
-				s.cancelOnServerError()
-				return
-			}
-			// fast response to client
-			if tr.trace == nil {
-				// request is untraced, send no-error to client.
-				tr.responses <- Response{err: nil}
 			}
-
 		}
 	}
 }

node/router/stub_consenter_test.go

@@ -19,17 +19,16 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-type stubConsenter struct {
+type StubConsenter struct {
 	ca          tlsgen.CA // Certificate authority that issues a certificate for the consenter
 	certificate []byte
 	key         []byte
 	server      *comm.GRPCServer // GRPCServer instance represents the consenter
 	txs         uint32           // Number of txs received from router
 	partyID     types.PartyID
-	logger      types.Logger
 }
 
-func NewStubConsenter(t *testing.T, ca tlsgen.CA, partyID types.PartyID, logger types.Logger) stubConsenter {
+func NewStubConsenter(t *testing.T, ca tlsgen.CA, partyID types.PartyID) StubConsenter {
 	// create a (cert,key) pair for the consenter
 	certKeyPair, err := ca.NewServerCertKeyPair("127.0.0.1")
 	require.NoError(t, err)
@@ -45,18 +44,17 @@ func NewStubConsenter(t *testing.T, ca tlsgen.CA, partyID types.PartyID, logger
 	require.NoError(t, err)
 
 	// return a stub consenter that includes all server setup
-	stubConsenter := stubConsenter{
+	stubConsenter := StubConsenter{
 		ca:          ca,
 		certificate: certKeyPair.Cert,
 		key:         certKeyPair.Key,
 		server:      server,
 		partyID:     partyID,
-		logger:      logger,
 	}
 	return stubConsenter
 }
 
-func (sc *stubConsenter) Start() {
+func (sc *StubConsenter) Start() {
 	protos.RegisterConsensusServer(sc.server.Server(), sc)
 	go func() {
 		if err := sc.server.Start(); err != nil {
@@ -65,11 +63,11 @@ func (sc *stubConsenter) Start() {
 	}()
 }
 
-func (sc *stubConsenter) Stop() {
+func (sc *StubConsenter) Stop() {
 	sc.server.Stop()
 }
 
-func (sc *stubConsenter) Restart() {
+func (sc *StubConsenter) Restart() {
 	// save the current server address
 	addr := sc.server.Address()
 
@@ -96,20 +94,20 @@ func (sc *stubConsenter) Restart() {
 	}()
 }
 
-func (sc *stubConsenter) ReceivedMessageCount() uint32 {
+func (sc *StubConsenter) ReceivedMessageCount() uint32 {
 	receivedTxs := atomic.LoadUint32(&sc.txs)
 	return receivedTxs
 }
 
-func (sc *stubConsenter) GetConsenterEndpoint() string {
+func (sc *StubConsenter) GetConsenterEndpoint() string {
 	return sc.server.Address()
 }
 
-func (sc *stubConsenter) NotifyEvent(stream protos.Consensus_NotifyEventServer) error {
+func (sc *StubConsenter) NotifyEvent(stream protos.Consensus_NotifyEventServer) error {
 	return fmt.Errorf("NotifyEvent not implemented")
 }
 
-func (sc *stubConsenter) SubmitConfig(ctx context.Context, request *protos.Request) (*protos.SubmitResponse, error) {
+func (sc *StubConsenter) SubmitConfig(ctx context.Context, request *protos.Request) (*protos.SubmitResponse, error) {
 	resp := &protos.SubmitResponse{
 		Error:   "dummy submit config",
 		ReqID:   request.Identity,