From ab033fbc81b37bc720ebef5ef79936554f4d10f2 Mon Sep 17 00:00:00 2001 From: Juan Leni Date: Tue, 19 Mar 2019 11:27:02 +0100 Subject: [PATCH] Improve reconnections + fixing unit tests --- node/node.go | 9 ++++-- node/node_test.go | 5 ++- privval/errors.go | 2 ++ privval/signer_client.go | 1 + privval/signer_dialer_endpoint.go | 29 ++++++++++-------- privval/signer_listener_endpoint.go | 39 +++++++++++++----------- privval/signer_listener_endpoint_test.go | 14 ++++----- 7 files changed, 56 insertions(+), 43 deletions(-) diff --git a/node/node.go b/node/node.go index 1bec6323a..c0ae728f6 100644 --- a/node/node.go +++ b/node/node.go @@ -296,18 +296,23 @@ func NewNode(config *cfg.Config, } } + pubKey := privValidator.GetPubKey() + if pubKey == nil { + return nil, fmt.Errorf("could not retrieve pubkey") + } + // Decide whether to fast-sync or not // We don't fast-sync when the only validator is us. fastSync := config.FastSync if state.Validators.Size() == 1 { addr, _ := state.Validators.GetByIndex(0) - privValAddr := privValidator.GetPubKey().Address() + + privValAddr := pubKey.Address() if bytes.Equal(privValAddr, addr) { fastSync = false } } - pubKey := privValidator.GetPubKey() addr := pubKey.Address() // Log whether this node is a validator or an observer if state.Validators.HasAddress(addr) { diff --git a/node/node_test.go b/node/node_test.go index 76a88fcf7..b6d846f93 100644 --- a/node/node_test.go +++ b/node/node_test.go @@ -138,7 +138,7 @@ func TestNodeSetPrivValTCP(t *testing.T) { types.NewMockPV(), dialer, ) - privval.SignerServiceEndpointTimeoutReadWrite(100 * time.Millisecond)(pvsc) + privval.SignerDialerEndpointTimeoutReadWrite(100 * time.Millisecond)(pvsc) go func() { err := pvsc.Start() @@ -180,7 +180,7 @@ func TestNodeSetPrivValIPC(t *testing.T) { types.NewMockPV(), dialer, ) - privval.SignerServiceEndpointTimeoutReadWrite(100 * time.Millisecond)(pvsc) + privval.SignerDialerEndpointTimeoutReadWrite(100 * time.Millisecond)(pvsc) go func() { err := pvsc.Start() @@ -191,7 +191,6 @@ func TestNodeSetPrivValIPC(t *testing.T) { n, err := DefaultNewNode(config, log.TestingLogger()) require.NoError(t, err) assert.IsType(t, &privval.SignerClient{}, n.PrivValidator()) - } // testFreeAddr claims a free port so we don't block on listener being ready. diff --git a/privval/errors.go b/privval/errors.go index 13687f384..db7d2782f 100644 --- a/privval/errors.go +++ b/privval/errors.go @@ -17,6 +17,8 @@ var ( ErrListenerTimeout = ListenerTimeoutError{} ErrListenerNoConnection = fmt.Errorf("signer listening endpoint is not connected") ErrDialerTimeout = fmt.Errorf("signer dialer endpoint timed out") + ErrDialerReadTimeout = fmt.Errorf("signer dialer endpoint read timed out") + ErrDialerWriteTimeout = fmt.Errorf("signer dialer endpoint write timed out") ) // RemoteSignerError allows (remote) validators to include meaningful error descriptions in their reply. diff --git a/privval/signer_client.go b/privval/signer_client.go index 740314ec0..f46a1f035 100644 --- a/privval/signer_client.go +++ b/privval/signer_client.go @@ -54,6 +54,7 @@ func (sc *SignerClient) WaitForConnection(maxWait time.Duration) error { func (sc *SignerClient) GetPubKey() crypto.PubKey { response, err := sc.endpoint.SendRequest(&PubKeyRequest{}) if err != nil { + sc.endpoint.Logger.Error("error sending request", "err", err) return nil } diff --git a/privval/signer_dialer_endpoint.go b/privval/signer_dialer_endpoint.go index e6aca4568..7afa71972 100644 --- a/privval/signer_dialer_endpoint.go +++ b/privval/signer_dialer_endpoint.go @@ -13,27 +13,24 @@ import ( ) const ( - defaultMaxDialRetries = 10 + defaultMaxDialRetries = 10 + defaultRetryWaitMilliseconds = 100 ) // SignerServiceEndpointOption sets an optional parameter on the SignerDialerEndpoint. type SignerServiceEndpointOption func(*SignerDialerEndpoint) -// SignerServiceEndpointTimeoutReadWrite sets the read and write timeout for connections +// SignerDialerEndpointTimeoutReadWrite sets the read and write timeout for connections // from external signing processes. -func SignerServiceEndpointTimeoutReadWrite(timeout time.Duration) SignerServiceEndpointOption { +func SignerDialerEndpointTimeoutReadWrite(timeout time.Duration) SignerServiceEndpointOption { return func(ss *SignerDialerEndpoint) { ss.timeoutReadWrite = timeout } } -// SignerServiceEndpointConnRetries sets the amount of attempted retries to AcceptNewConnection. -func SignerServiceEndpointConnRetries(retries int) SignerServiceEndpointOption { +// SignerDialerEndpointConnRetries sets the amount of attempted retries to AcceptNewConnection. +func SignerDialerEndpointConnRetries(retries int) SignerServiceEndpointOption { return func(ss *SignerDialerEndpoint) { ss.maxConnRetries = retries } } -// TODO(jleni): Create a common type for a signerEndpoint (common for both listener/dialer) -// read -// write - // SignerDialerEndpoint dials using its dialer and responds to any // signature requests using its privVal. type SignerDialerEndpoint struct { @@ -44,6 +41,7 @@ type SignerDialerEndpoint struct { conn net.Conn timeoutReadWrite time.Duration + retryWait time.Duration maxConnRetries int chainID string @@ -66,6 +64,7 @@ func NewSignerDialerEndpoint( se := &SignerDialerEndpoint{ dialer: dialer, timeoutReadWrite: defaultTimeoutReadWriteSeconds * time.Second, + retryWait: defaultRetryWaitMilliseconds * time.Millisecond, maxConnRetries: defaultMaxDialRetries, chainID: chainID, @@ -91,11 +90,12 @@ func (ss *SignerDialerEndpoint) OnStart() error { // OnStop implements cmn.Service. func (ss *SignerDialerEndpoint) OnStop() { ss.Logger.Debug("SignerDialerEndpoint: OnStop calling Close") - _ = ss.Close() // Stop service loop close(ss.stopCh) <-ss.stoppedCh + + _ = ss.Close() } // Close closes the underlying net.Conn. @@ -142,7 +142,7 @@ func (ss *SignerDialerEndpoint) readMessage() (msg RemoteSignerMsg, err error) { const maxRemoteSignerMsgSize = 1024 * 10 _, err = cdc.UnmarshalBinaryLengthPrefixedReader(ss.conn, &msg, maxRemoteSignerMsgSize) if _, ok := err.(timeoutError); ok { - err = cmn.ErrorWrap(ErrDialerTimeout, err.Error()) + err = cmn.ErrorWrap(ErrDialerReadTimeout, err.Error()) } return @@ -165,7 +165,7 @@ func (ss *SignerDialerEndpoint) writeMessage(msg RemoteSignerMsg) (err error) { _, err = cdc.MarshalBinaryLengthPrefixedWriter(ss.conn, msg) if _, ok := err.(timeoutError); ok { - err = cmn.ErrorWrap(ErrDialerTimeout, err.Error()) + err = cmn.ErrorWrap(ErrDialerWriteTimeout, err.Error()) } return @@ -224,10 +224,13 @@ func (ss *SignerDialerEndpoint) serviceLoop() { if ss.conn == nil { ss.conn, err = ss.dialer() - if err != nil { + ss.Logger.Info("Try connect", "err", err) ss.conn = nil // Explicitly set to nil because dialer returns an interface (https://golang.org/doc/faq#nil_error) retries++ + + // Wait between retries + time.Sleep(ss.retryWait) continue } } diff --git a/privval/signer_listener_endpoint.go b/privval/signer_listener_endpoint.go index 007c3d9b9..0d712b58d 100644 --- a/privval/signer_listener_endpoint.go +++ b/privval/signer_listener_endpoint.go @@ -18,7 +18,7 @@ type SignerValidatorEndpointOption func(*SignerListenerEndpoint) type SignerListenerEndpoint struct { cmn.BaseService - extMtx sync.Mutex + mtx sync.Mutex listener net.Listener conn net.Conn @@ -71,43 +71,39 @@ func (ve *SignerListenerEndpoint) OnStop() { } } - ve.Logger.Debug("SignerListenerEndpoint: OnStop close stopCh") // Stop service loop - ve.stopCh <- struct{}{} <-ve.stoppedCh } // Close closes the underlying net.Conn. func (ve *SignerListenerEndpoint) Close() error { - ve.extMtx.Lock() - defer ve.extMtx.Unlock() + ve.mtx.Lock() + defer ve.mtx.Unlock() ve.Logger.Debug("SignerListenerEndpoint: Close") ve.dropConnection() - - ve.Logger.Debug("SignerListenerEndpoint: Closed") return nil } // IsConnected indicates if there is an active connection func (ve *SignerListenerEndpoint) IsConnected() bool { - ve.extMtx.Lock() - defer ve.extMtx.Unlock() + ve.mtx.Lock() + defer ve.mtx.Unlock() return ve.isConnected() } // WaitForConnection waits maxWait for a connection or returns a timeout error func (ve *SignerListenerEndpoint) WaitForConnection(maxWait time.Duration) error { - ve.extMtx.Lock() - defer ve.extMtx.Unlock() + ve.mtx.Lock() + defer ve.mtx.Unlock() return ve.ensureConnection(maxWait) } // SendRequest sends a request and waits for a response func (ve *SignerListenerEndpoint) SendRequest(request RemoteSignerMsg) (RemoteSignerMsg, error) { - ve.extMtx.Lock() - defer ve.extMtx.Unlock() + ve.mtx.Lock() + defer ve.mtx.Unlock() // TODO: Add retries.. that include dropping the connection and @@ -237,19 +233,16 @@ func (ve *SignerListenerEndpoint) ensureConnection(maxWait time.Duration) error // dropConnection closes the current connection but does not touch the listening socket func (ve *SignerListenerEndpoint) dropConnection() { - ve.Logger.Debug("SignerListenerEndpoint: dropConnection") if ve.conn != nil { if err := ve.conn.Close(); err != nil { ve.Logger.Error("SignerListenerEndpoint::dropConnection", "err", err) } ve.conn = nil } - ve.Logger.Debug("SignerListenerEndpoint: dropConnection DONE") } func (ve *SignerListenerEndpoint) serviceLoop() { defer close(ve.stoppedCh) - defer ve.Logger.Debug("SignerListenerEndpoint::serviceLoop EXIT") ve.Logger.Debug("SignerListenerEndpoint::serviceLoop") for { @@ -261,8 +254,18 @@ func (ve *SignerListenerEndpoint) serviceLoop() { conn, err := ve.acceptNewConnection() if err == nil { ve.Logger.Info("Connected") - ve.connectedCh <- conn - break + + select { + case ve.connectedCh <- conn: + { + ve.Logger.Debug("SignerListenerEndpoint: connection relayed") + } + case <-ve.stopCh: + { + ve.Logger.Debug("SignerListenerEndpoint: stopping") + return + } + } } } } diff --git a/privval/signer_listener_endpoint_test.go b/privval/signer_listener_endpoint_test.go index 405746eb3..9890969e9 100644 --- a/privval/signer_listener_endpoint_test.go +++ b/privval/signer_listener_endpoint_test.go @@ -66,8 +66,8 @@ func TestSignerRemoteRetryTCPOnly(t *testing.T) { ) defer serviceEndpoint.Stop() - SignerServiceEndpointTimeoutReadWrite(time.Millisecond)(serviceEndpoint) - SignerServiceEndpointConnRetries(retries)(serviceEndpoint) + SignerDialerEndpointTimeoutReadWrite(time.Millisecond)(serviceEndpoint) + SignerDialerEndpointConnRetries(retries)(serviceEndpoint) err = serviceEndpoint.Start() assert.NoError(t, err) @@ -75,7 +75,7 @@ func TestSignerRemoteRetryTCPOnly(t *testing.T) { select { case attempts := <-attemptCh: assert.Equal(t, retries, attempts) - case <-time.After(100 * time.Millisecond): + case <-time.After(1500 * time.Millisecond): t.Error("expected remote to observe connection attempts") } } @@ -98,8 +98,8 @@ func TestRetryConnToRemoteSigner(t *testing.T) { validatorEndpoint = newSignerValidatorEndpoint(logger, tc.addr, thisConnTimeout) ) - SignerServiceEndpointTimeoutReadWrite(testTimeoutReadWrite)(serviceEndpoint) - SignerServiceEndpointConnRetries(10)(serviceEndpoint) + SignerDialerEndpointTimeoutReadWrite(testTimeoutReadWrite)(serviceEndpoint) + SignerDialerEndpointConnRetries(10)(serviceEndpoint) getStartEndpoint(t, readyCh, validatorEndpoint) defer validatorEndpoint.Stop() @@ -188,8 +188,8 @@ func getMockEndpoints( validatorEndpoint = newSignerValidatorEndpoint(logger, addr, testTimeoutReadWrite) ) - SignerServiceEndpointTimeoutReadWrite(testTimeoutReadWrite)(serviceEndpoint) - SignerServiceEndpointConnRetries(1e6)(serviceEndpoint) + SignerDialerEndpointTimeoutReadWrite(testTimeoutReadWrite)(serviceEndpoint) + SignerDialerEndpointConnRetries(1e6)(serviceEndpoint) getStartEndpoint(t, readyCh, validatorEndpoint)