fix(privval): CV ignore duplicate privval listen when connected (back…

…port #3830) (#3896) Resolve #3828 On cometbft startup, duplicate privval request can enter the channel between the time listen begins and a send is attempted. The duplicate listen request will timeout since there is not another signer connecting, filling logs with errors. A duplicate listen request is undesired. --- #### PR checklist - [x] Tests written/updated - [x] Changelog entry added in `.changelog` (we use [unclog](https://github.com/informalsystems/unclog) to manage our changelog) - [ ] Updated relevant documentation (`docs/` or `spec/`) and code comments <hr>This is an automatic backport of pull request #3830 done by [Mergify](https://mergify.com). --------- Co-authored-by: Chill Validation <92176880+chillyvee@users.noreply.github.com> Co-authored-by: Anton Kaliaev <anton.kalyaev@gmail.com>
cometbft · faddat · Nov 17, 2023 · Nov 17, 2023 · Nov 17, 2023 · Nov 17, 2023
commit 221c7445ae7461bfc39dd526b88b4f92e2dd2c22
diff --git a/.changelog/unreleased/bug-fixes/3828-privval-drop-duplicate-listen.md b/.changelog/unreleased/bug-fixes/3828-privval-drop-duplicate-listen.md
@@ -0,0 +1 @@
+- `[privval]` Ignore duplicate privval listen when already connected ([\#3828](https://github.com/cometbft/cometbft/issues/3828)
@@ -53,11 +53,9 @@ func (se *signerEndpoint) GetAvailableConnection(connectionAvailableCh chan net.
 
 // TryGetConnection retrieves a connection if it is already available
 func (se *signerEndpoint) WaitConnection(connectionAvailableCh chan net.Conn, maxWait time.Duration) error {
-	se.connMtx.Lock()
-	defer se.connMtx.Unlock()
-
 	select {
-	case se.conn = <-connectionAvailableCh:
+	case conn := <-connectionAvailableCh:
+		se.SetConnection(conn)
 	case <-time.After(maxWait):
 		return ErrConnectionTimeout
 	}

@@ -3,6 +3,7 @@ package privval
 import (
 	"fmt"
 	"net"
+	"sync/atomic"
 	"time"
 
 	"github.com/cometbft/cometbft/libs/log"
@@ -34,9 +35,10 @@ type SignerListenerEndpoint struct {
 	connectRequestCh      chan struct{}
 	connectionAvailableCh chan net.Conn
 
-	timeoutAccept time.Duration
-	pingTimer     *time.Ticker
-	pingInterval  time.Duration
+	timeoutAccept   time.Duration
+	acceptFailCount atomic.Uint32
+	pingTimer       *time.Ticker
+	pingInterval    time.Duration
 
 	instanceMtx cmtsync.Mutex // Ensures instance public methods access, i.e. SendRequest
 }
@@ -159,9 +161,11 @@ func (sl *SignerListenerEndpoint) acceptNewConnection() (net.Conn, error) {
 	sl.Logger.Info("SignerListener: Listening for new connection")
 	conn, err := sl.listener.Accept()
 	if err != nil {
+		sl.acceptFailCount.Add(1)
 		return nil, err
 	}
 
+	sl.acceptFailCount.Store(0)
 	return conn, nil
 }
 
@@ -181,9 +185,17 @@ func (sl *SignerListenerEndpoint) serviceLoop() {
 	for {
 		select {
 		case <-sl.connectRequestCh:
+			// On start, listen timeouts can queue a duplicate connect request to queue
+			// while the first request connects.  Drop duplicate request.
+			if sl.IsConnected() {
+				sl.Logger.Debug("SignerListener: Connected. Drop Listen Request")
+				continue
+			}
+
+			// Listen for remote signer
 			conn, err := sl.acceptNewConnection()
 			if err != nil {
-				sl.Logger.Error("SignerListener: Error accepting connection", "err", err)
+				sl.Logger.Error("SignerListener: Error accepting connection", "err", err, "failures", sl.acceptFailCount.Load())
 				sl.triggerConnect()
 				continue
 			}

@@ -146,6 +146,89 @@ func TestRetryConnToRemoteSigner(t *testing.T) {
 	}
 }
 
+func TestDuplicateListenReject(t *testing.T) {
+	for _, tc := range getDialerTestCases(t) {
+		var (
+			logger           = log.TestingLogger()
+			chainID          = cmtrand.Str(12)
+			mockPV           = types.NewMockPV()
+			endpointIsOpenCh = make(chan struct{})
+			thisConnTimeout  = testTimeoutReadWrite
+			listenerEndpoint = newSignerListenerEndpoint(logger, tc.addr, thisConnTimeout)
+		)
+		listenerEndpoint.timeoutAccept = defaultTimeoutAcceptSeconds / 2 * time.Second
+
+		dialerEndpoint := NewSignerDialerEndpoint(
+			logger,
+			tc.dialer,
+		)
+		SignerDialerEndpointTimeoutReadWrite(testTimeoutReadWrite)(dialerEndpoint)
+		SignerDialerEndpointConnRetries(10)(dialerEndpoint)
+
+		signerServer := NewSignerServer(dialerEndpoint, chainID, mockPV)
+
+		startListenerEndpointAsync(t, listenerEndpoint, endpointIsOpenCh)
+		t.Cleanup(func() {
+			if err := listenerEndpoint.Stop(); err != nil {
+				t.Error(err)
+			}
+		})
+
+		require.NoError(t, signerServer.Start())
+		assert.True(t, signerServer.IsRunning())
+
+		<-endpointIsOpenCh
+		if err := signerServer.Stop(); err != nil {
+			t.Error(err)
+		}
+
+		dialerEndpoint2 := NewSignerDialerEndpoint(
+			logger,
+			tc.dialer,
+		)
+		signerServer2 := NewSignerServer(dialerEndpoint2, chainID, mockPV)
+
+		// let some pings pass
+		require.NoError(t, signerServer2.Start())
+		assert.True(t, signerServer2.IsRunning())
+
+		// wait for successful connection
+		for {
+			if listenerEndpoint.IsConnected() {
+				break
+			}
+		}
+
+		// simulate ensureConnection, bypass triggerConnect default drop with multiple messages
+		time.Sleep(100 * time.Millisecond)
+		listenerEndpoint.triggerConnect()
+		time.Sleep(100 * time.Millisecond)
+		listenerEndpoint.triggerConnect()
+		time.Sleep(100 * time.Millisecond)
+		listenerEndpoint.triggerConnect()
+
+		// simulate validator node running long enough for privval listen timeout multiple times
+		// up to 1 timeout error is possible due to timing differences
+		// Run 3 times longer than timeout to generate at least 2 accept errors
+		time.Sleep(3 * defaultTimeoutAcceptSeconds * time.Second)
+		t.Cleanup(func() {
+			if err := signerServer2.Stop(); err != nil {
+				t.Error(err)
+			}
+		})
+
+		// after connect, there should not be more than 1 accept fail
+		assert.LessOrEqual(t, listenerEndpoint.acceptFailCount.Load(), uint32(1))
+
+		// give the client some time to re-establish the conn to the remote signer
+		// should see sth like this in the logs:
+		//
+		// E[10016-01-10|17:12:46.128] Ping                                         err="remote signer timed out"
+		// I[10016-01-10|17:16:42.447] Re-created connection to remote signer       impl=SocketVal
+		time.Sleep(testTimeoutReadWrite * 2)
+	}
+}
+
 func newSignerListenerEndpoint(logger log.Logger, addr string, timeoutReadWrite time.Duration) *SignerListenerEndpoint {
 	proto, address := cmtnet.ProtocolAndAddress(addr)
Original file line number	Diff line number	Diff line change
		@@ -0,0 +1 @@
		- `[privval]` Ignore duplicate privval listen when already connected ([\#3828](https://github.com/cometbft/cometbft/issues/3828)