diff --git a/privval/errors.go b/privval/errors.go index 5260e4faa..4d3481207 100644 --- a/privval/errors.go +++ b/privval/errors.go @@ -6,9 +6,10 @@ import ( // Socket errors. var ( - ErrUnexpectedResponse = fmt.Errorf("received unexpected response") - ErrListenerTimeout = fmt.Errorf("signer listening endpoint timed out") - ErrDialerTimeout = fmt.Errorf("signer dialer endpoint timed out") + ErrUnexpectedResponse = fmt.Errorf("received unexpected response") + ErrListenerTimeout = fmt.Errorf("signer listening endpoint timed out") + ErrListenerNoConnection = fmt.Errorf("signer listening endpoint is not connected") + ErrDialerTimeout = fmt.Errorf("signer dialer endpoint timed out") ) // RemoteSignerError allows (remote) validators to include meaningful error descriptions in their reply. diff --git a/privval/file_test.go b/privval/file_test.go index a82911443..45079dce3 100644 --- a/privval/file_test.go +++ b/privval/file_test.go @@ -58,7 +58,7 @@ func TestResetValidator(t *testing.T) { // priv val after signing is not same as empty assert.NotEqual(t, privVal.LastSignState, emptyState) - // priv val after tryConnect is same as empty + // priv val after tryAcceptConnection is same as empty privVal.Reset() assert.Equal(t, privVal.LastSignState, emptyState) } diff --git a/privval/signer_dialer_endpoint.go b/privval/signer_dialer_endpoint.go index 9d1d80c83..988d5ddfc 100644 --- a/privval/signer_dialer_endpoint.go +++ b/privval/signer_dialer_endpoint.go @@ -20,7 +20,7 @@ func SignerServiceEndpointTimeoutReadWrite(timeout time.Duration) SignerServiceE return func(ss *SignerDialerEndpoint) { ss.timeoutReadWrite = timeout } } -// SignerServiceEndpointConnRetries sets the amount of attempted retries to tryConnect. +// SignerServiceEndpointConnRetries sets the amount of attempted retries to tryAcceptConnection. func SignerServiceEndpointConnRetries(retries int) SignerServiceEndpointOption { return func(ss *SignerDialerEndpoint) { ss.connRetries = retries } } diff --git a/privval/signer_listener_endpoint.go b/privval/signer_listener_endpoint.go index 3c50ce16a..9ac8a5e8f 100644 --- a/privval/signer_listener_endpoint.go +++ b/privval/signer_listener_endpoint.go @@ -30,7 +30,7 @@ func SignerValidatorEndpointSetHeartbeat(period time.Duration) SignerValidatorEn // TODO: Add a type for SignerEndpoints // getConnection -// tryConnect +// tryAcceptConnection // read // write // close @@ -46,6 +46,8 @@ type SignerListenerEndpoint struct { listener net.Listener conn net.Conn + timeoutReadWrite time.Duration + // ping cancelPingCh chan struct{} pingTicker *time.Ticker @@ -56,6 +58,7 @@ type SignerListenerEndpoint struct { func NewSignerListenerEndpoint(logger log.Logger, listener net.Listener) *SignerListenerEndpoint { sc := &SignerListenerEndpoint{ listener: listener, + timeoutReadWrite : defaultTimeoutReadWriteSeconds * time.Second, heartbeatPeriod: heartbeatPeriod, } @@ -68,47 +71,55 @@ func NewSignerListenerEndpoint(logger log.Logger, listener net.Listener) *Signer func (ve *SignerListenerEndpoint) OnStart() error { ve.Logger.Debug("SignerListenerEndpoint: OnStart") - err := ve.tryConnect() + err := ve.tryAcceptConnection() if err != nil { ve.Logger.Error("OnStart", "err", err) return err } + ve.Logger.Info("OnStart", "connected", ve.isConnected()) + // Start a routine to keep the connection alive + go ve.pingLoop() + + return nil +} + +func (ve *SignerListenerEndpoint)pingLoop() { ve.cancelPingCh = make(chan struct{}, 1) ve.pingTicker = time.NewTicker(ve.heartbeatPeriod) - // TODO: Move subroutine to another place? - go func() { - for { - select { - case <-ve.pingTicker.C: - ve.Logger.Debug("SignerListenerEndpoint: ping ticker ch") - err := ve.ping() - if err != nil { - ve.Logger.Error("Ping", "err", err) - if err == ErrUnexpectedResponse { - return - } + fmt.Printf("ONSTART out -> %p\n", ve) - err := ve.tryConnect() - if err != nil { - ve.Logger.Error("Reconnecting to remote signer failed", "err", err) - continue - } + for { + fmt.Printf("ONSTART in -> %p\n", ve) - ve.Logger.Info("Re-created connection to remote signer", "impl", ve) + select { + case <-ve.pingTicker.C: + fmt.Printf("ONSTART before -> %p\n", ve) + err := ve.ping() + + if err != nil { + ve.Logger.Error("Ping", "err", err) + if err == ErrUnexpectedResponse { + return } - case <-ve.cancelPingCh: - ve.Logger.Debug("SignerListenerEndpoint: cancel ping ch") - ve.pingTicker.Stop() - return + err := ve.tryAcceptConnection() + if err != nil { + ve.Logger.Error("Connection from remote signer not available", "err", err) + continue + } + + ve.Logger.Info("Connection from remote signer available", "impl", ve) } - } - }() + case <-ve.cancelPingCh: + ve.Logger.Debug("SignerListenerEndpoint: cancel ping ch") - return nil + ve.pingTicker.Stop() + return + } + } } // OnStop implements cmn.Service. @@ -150,6 +161,7 @@ func (ve *SignerListenerEndpoint) Close() error { ve.Logger.Error("Closing connection", "err", err) return err } + ve.Logger.Debug("SignerListenerEndpoint: set ve.conn Nil") ve.conn = nil } @@ -168,15 +180,21 @@ func (ve *SignerListenerEndpoint) SendRequest(request RemoteSignerMsg) (RemoteSi ve.mtx.Lock() defer ve.mtx.Unlock() + ve.Logger.Debug("SignerListenerEndpoint: Send request", "connected", ve.isConnected()) + if !ve.isConnected() { - return nil, cmn.ErrorWrap(ErrListenerTimeout, "endpoint is not connected") + return nil, cmn.ErrorWrap(ErrListenerNoConnection, "endpoint is not connected") } + ve.Logger.Debug("Send request. Write") + err := ve.writeMessage(request) if err != nil { return nil, err } + ve.Logger.Debug("Send request. Read") + res, err := ve.readMessage() if err != nil { return nil, err @@ -187,12 +205,21 @@ func (ve *SignerListenerEndpoint) SendRequest(request RemoteSignerMsg) (RemoteSi // IsConnected indicates if there is an active connection func (ve *SignerListenerEndpoint) isConnected() bool { - return ve.IsRunning() && ve.conn != nil +// return ve.IsRunning() && ve.conn != nil + return ve.conn != nil } func (ve *SignerListenerEndpoint) readMessage() (msg RemoteSignerMsg, err error) { if !ve.isConnected() { - return nil, cmn.ErrorWrap(ErrListenerTimeout, "endpoint is not connected") + return nil, cmn.ErrorWrap(ErrListenerNoConnection, "endpoint is not connected") + } + + // Reset read deadline + deadline := time.Now().Add(ve.timeoutReadWrite) + ve.Logger.Debug("SignerListenerEndpoint: readMessage", "deadline", deadline) + err = ve.conn.SetReadDeadline(deadline) + if err != nil { + return } const maxRemoteSignerMsgSize = 1024 * 10 @@ -206,7 +233,17 @@ func (ve *SignerListenerEndpoint) readMessage() (msg RemoteSignerMsg, err error) func (ve *SignerListenerEndpoint) writeMessage(msg RemoteSignerMsg) (err error) { if !ve.isConnected() { - return cmn.ErrorWrap(ErrListenerTimeout, "endpoint is not connected") + return cmn.ErrorWrap(ErrListenerNoConnection, "endpoint is not connected") + } + + fmt.Printf("writemessage -> %p\n", ve) + + // Reset read deadline + deadline := time.Now().Add(ve.timeoutReadWrite) + ve.Logger.Debug("SignerListenerEndpoint: writeMessage", "deadline", deadline) + err = ve.conn.SetWriteDeadline(deadline) + if err != nil { + return } _, err = cdc.MarshalBinaryLengthPrefixedWriter(ve.conn, msg) @@ -217,12 +254,12 @@ func (ve *SignerListenerEndpoint) writeMessage(msg RemoteSignerMsg) (err error) return } -// tryConnect waits to accept a new connection. -func (ve *SignerListenerEndpoint) tryConnect() error { +// tryAcceptConnection waits to accept a new connection. +func (ve *SignerListenerEndpoint) tryAcceptConnection() error { ve.mtx.Lock() defer ve.mtx.Unlock() - ve.Logger.Debug("SignerListenerEndpoint: tryConnect") + ve.Logger.Debug("SignerListenerEndpoint: tryAcceptConnection") if !ve.IsRunning() || ve.listener == nil { return fmt.Errorf("endpoint is closing") @@ -231,22 +268,25 @@ func (ve *SignerListenerEndpoint) tryConnect() error { // if the conn already exists and close it. if ve.conn != nil { if tmpErr := ve.conn.Close(); tmpErr != nil { - ve.Logger.Error("error closing socket val connection during tryConnect", "err", tmpErr) + ve.Logger.Error("tryAcceptConnection: error closing old connection", "err", tmpErr) } } // Forget old connection ve.conn = nil + ve.Logger.Debug("SignerListenerEndpoint: set ve.conn Nil") // wait for a new conn conn, err := ve.listener.Accept() if err != nil { + ve.Logger.Debug("listener accept failed", "err", err) return err } - ve.Logger.Debug("SignerListenerEndpoint: New connection") - ve.conn = conn + ve.Logger.Debug("SignerListenerEndpoint: New connection", "connected", ve.isConnected()) + + fmt.Printf("tryAcceptConnection: %p\n", ve) // TODO: maybe we need to inform the owner that a connection has been received return nil @@ -254,6 +294,8 @@ func (ve *SignerListenerEndpoint) tryConnect() error { // Ping is used to check connection health. func (ve *SignerListenerEndpoint) ping() error { + ve.Logger.Debug("SignerListenerEndpoint: PING", "connected", ve.isConnected()) + response, err := ve.SendRequest(&PingRequest{}) if err != nil { diff --git a/privval/signer_listener_endpoint_test.go b/privval/signer_listener_endpoint_test.go index 8a550c099..de266cab6 100644 --- a/privval/signer_listener_endpoint_test.go +++ b/privval/signer_listener_endpoint_test.go @@ -31,7 +31,7 @@ type dialerTestCase struct { // TestSignerRemoteRetryTCPOnly will test connection retry attempts over TCP. We // don't need this for Unix sockets because the OS instantly knows the state of // both ends of the socket connection. This basically causes the -// SignerDialerEndpoint.dialer() call inside SignerDialerEndpoint.tryConnect() to return +// SignerDialerEndpoint.dialer() call inside SignerDialerEndpoint.tryAcceptConnection() to return // successfully immediately, putting an instant stop to any retry attempts. func TestSignerRemoteRetryTCPOnly(t *testing.T) { var (