fix(go): tear down only the connection a request failed on

foreign/go/client/tcp/tcp_core.go

@@ -100,6 +100,11 @@ type IggyTcpClient struct {
 	// connectAttempt is the attempt a Connect is running, shared with every
 	// caller that arrives while it is in progress. Guarded by c.mtx.
 	connectAttempt *connectAttempt
+	// connGeneration counts the connections this client installed. A request
+	// carries the generation it ran on, so the teardown that follows its
+	// failure closes that connection rather than one another caller
+	// established meanwhile. Guarded by c.mtx.
+	connGeneration uint64
 	// rememberedLogin holds the credentials a manual sign-in succeeded with,
 	// so a reconnect -- on this node or, after a failover, another one -- can
 	// re-establish the session instead of surfacing an unauthenticated error.
@@ -505,7 +510,7 @@ func (e *localPreconditionError) Unwrap() error { return e.err }
 // exchange runs one request to completion, reconnecting and replaying it when
 // the failure is one a fresh connection recovers from.
 func (c *IggyTcpClient) exchange(ctx context.Context, code uint32, frame []byte) ([]byte, error) {
-	response, err := c.sendFrame(ctx, code, frame)
+	response, generation, err := c.sendFrame(ctx, code, frame)
 	if err == nil || !isReconnectable(err) {
 		return response, err
 	}
@@ -549,7 +554,7 @@ func (c *IggyTcpClient) exchange(ctx context.Context, code uint32, frame []byte)
 		return nil, err
 	}
 
-	if disconnectErr := c.disconnect(); disconnectErr != nil {
+	if _, disconnectErr := c.disconnectGeneration(generation); disconnectErr != nil {
 		return nil, disconnectErr
 	}
 	reconnectCtx := ctx
@@ -568,7 +573,8 @@ func (c *IggyTcpClient) exchange(ctx context.Context, code uint32, frame []byte)
 	if reconnectErr := c.Connect(reconnectCtx); reconnectErr != nil {
 		return nil, reconnectErr
 	}
-	return c.sendFrame(ctx, code, frame)
+	response, _, err = c.sendFrame(ctx, code, frame)
+	return response, err
 }
 
 // canReplay reports whether re-issuing the request over a fresh connection
@@ -622,12 +628,17 @@ func isReconnectable(err error) bool {
 
 // sendFrame runs the request against the current connection. One deadline
 // bounds it across every same-connection replay and every leader failover.
-func (c *IggyTcpClient) sendFrame(ctx context.Context, code uint32, frame []byte) ([]byte, error) {
+// It reports the connection generation the last attempt ran on, so a caller
+// tearing the connection down after a failure can tell whether that
+// connection is still the installed one.
+func (c *IggyTcpClient) sendFrame(
+	ctx context.Context, code uint32, frame []byte,
+) ([]byte, uint64, error) {
 	if ctx == nil {
-		return nil, ierror.ErrNilContext
+		return nil, 0, ierror.ErrNilContext
 	}
 	if err := ctx.Err(); err != nil {
-		return nil, err
+		return nil, 0, err
 	}
 
 	deadline := time.Now().Add(responseReadTimeout)
@@ -644,21 +655,21 @@ func (c *IggyTcpClient) sendFrame(ctx context.Context, code uint32, frame []byte
 			}
 		}
 
-		response, attemptStamped, err := c.attempt(
+		response, attemptStamped, generation, err := c.attempt(
 			ctx, code, frame, stamped, transientDeadline, deadline)
 		stamped = attemptStamped
 
 		switch {
 		case err == nil:
-			return response, nil
+			return response, generation, nil
 		case errors.Is(err, ierror.ErrTransientNotAccepted) &&
 			!isRegisterCode(code) && time.Now().Before(deadline):
 			// The server never admitted the request, so re-issuing it cannot
 			// double-apply. A same-connection replay keeps the stamped request
 			// id; a redirect registers again, so the frame is stamped afresh.
-			redirect, redirectErr := c.HandleLeaderRedirection(ctx)
+			redirect, redirectErr := c.redirectToLeader(ctx, generation)
 			if redirectErr != nil {
-				return nil, redirectErr
+				return nil, generation, redirectErr
 			}
 			if redirect {
 				redirectCtx := ctx
@@ -670,41 +681,43 @@ func (c *IggyTcpClient) sendFrame(ctx context.Context, code uint32, frame []byte
 					redirectCtx = suppressAutoLogin(ctx)
 				}
 				if connectErr := c.Connect(redirectCtx); connectErr != nil {
-					return nil, connectErr
+					return nil, generation, connectErr
 				}
 				stamped = false
 			}
 		default:
-			return nil, err
+			return nil, generation, err
 		}
 	}
 }
 
 // attempt stamps the frame if it is not stamped yet and exchanges it once,
-// replaying in place while the server answers transiently.
+// replaying in place while the server answers transiently. It reports the
+// connection generation it ran on alongside the outcome.
 func (c *IggyTcpClient) attempt(
 	ctx context.Context,
 	code uint32,
 	frame []byte,
 	stamped bool,
 	transientDeadline, readDeadline time.Time,
-) ([]byte, bool, error) {
+) ([]byte, bool, uint64, error) {
 	c.mtx.Lock()
 	defer c.mtx.Unlock()
 
+	generation := c.connGeneration
 	switch c.transportState {
 	case iggcon.TransportStateShutdown:
 		c.logger.Debug("Cannot send data. Client is shutdown.")
-		return nil, stamped, ierror.ErrClientShutdown
+		return nil, stamped, generation, ierror.ErrClientShutdown
 	case iggcon.TransportStateDisconnected:
 		c.logger.Debug("Cannot send data. Client is not connected.")
-		return nil, stamped, ierror.ErrNotConnected
+		return nil, stamped, generation, ierror.ErrNotConnected
 	case iggcon.TransportStateConnecting:
 		c.logger.Debug("Cannot send data. Client is still connecting.")
-		return nil, stamped, ierror.ErrNotConnected
+		return nil, stamped, generation, ierror.ErrNotConnected
 	}
 	if c.conn == nil {
-		return nil, stamped, ierror.ErrNotConnected
+		return nil, stamped, generation, ierror.ErrNotConnected
 	}
 
 	if !stamped {
@@ -713,7 +726,7 @@ func (c *IggyTcpClient) attempt(
 		// A stamp failure is local and pre-write, so it is marked as such:
 		// the connection is healthy and must not be torn down for it.
 		if err := vsr.StampRequestHeader(c.session, code, frame); err != nil {
-			return nil, false, &localPreconditionError{err}
+			return nil, false, generation, &localPreconditionError{err}
 		}
 		stamped = true
 	}
@@ -748,10 +761,10 @@ func (c *IggyTcpClient) attempt(
 
 	if err != nil {
 		if ctxErr := ctx.Err(); ctxErr != nil {
-			return nil, stamped, ctxErr
+			return nil, stamped, generation, ctxErr
 		}
 	}
-	return response, stamped, err
+	return response, stamped, generation, err
 }
 
 // exchangeLocked writes the frame and reads its reply, resending the identical
@@ -1093,6 +1106,7 @@ func (c *IggyTcpClient) Connect(ctx context.Context) (err error) {
 	}
 	c.conn = conn
 	c.reader = bufio.NewReaderSize(conn, 64*1024)
+	c.connGeneration++
 	c.transportState = iggcon.TransportStateConnected
 	c.connectedAt = time.Now()
 	// The server fence does not survive the old socket, so the new connection
@@ -1347,7 +1361,33 @@ func (c *IggyTcpClient) createTLSConfig(address string) (*tls.Config, error) {
 func (c *IggyTcpClient) disconnect() error {
 	c.mtx.Lock()
 	defer c.mtx.Unlock()
+	return c.disconnectLocked()
+}
+
+// disconnectGeneration tears down the connection a request ran on, and only
+// while that connection is still the installed one. It reports whether the
+// generation still was.
+//
+// A caller cannot go by the transport state alone: Connect marks the client
+// connected before it signs in, so a caller parked on c.mtx for the length of
+// that sign-in wakes to a state that looks healthy and would close the socket
+// the reconnect just established -- leaving the requests replaying over it
+// with nothing to send on, and starting a second attempt for a reconnect that
+// already happened.
+func (c *IggyTcpClient) disconnectGeneration(generation uint64) (bool, error) {
+	c.mtx.Lock()
+	defer c.mtx.Unlock()
+
+	if c.connGeneration != generation {
+		c.logger.Debug("Not disconnecting; the connection was already replaced.",
+			slog.Uint64("request_generation", generation),
+			slog.Uint64("current_generation", c.connGeneration))
+		return false, nil
+	}
+	return true, c.disconnectLocked()
+}
 
+func (c *IggyTcpClient) disconnectLocked() error {
 	if c.transportState == iggcon.TransportStateDisconnected || c.transportState == iggcon.TransportStateShutdown {
 		return nil
 	}

foreign/go/client/tcp/tcp_failover_test.go

@@ -687,6 +687,41 @@ func TestConnect_ConcurrentReconnectsThroughExchangeShareOneAttempt(t *testing.T
 		"the two failing requests reconnected separately")
 }
 
+// The teardown that follows a failure belongs to the connection the request
+// ran on. Connect marks the client connected before it signs in, so a caller
+// parked on c.mtx for that sign-in wakes to a healthy-looking state that says
+// nothing about which connection is installed.
+func TestDisconnect_DoesNotCloseAConnectionItDidNotFailOn(t *testing.T) {
+	var server *testListener
+	server = listenVSR(t, nil, singleNodeHandler(t, func() string { return server.address() }))
+
+	client := newDialingClient(t, server.address(),
+		WithAutoLogin(NewUsernamePasswordCredentials("iggy", "iggy")))
+	require.NoError(t, client.Connect(context.Background()))
+
+	// What a request that fails on this connection carries with it.
+	failed := client.connGeneration
+
+	require.NoError(t, client.disconnect())
+	require.NoError(t, client.Connect(context.Background()))
+	require.NotEqual(t, failed, client.connGeneration,
+		"a reconnect installs a connection of its own")
+
+	torn, err := client.disconnectGeneration(failed)
+	require.NoError(t, err)
+	assert.False(t, torn, "the stale generation reported a teardown it did not make")
+	assert.Equal(t, iggcon.TransportStateConnected, client.transportState)
+	require.NoError(t, client.Ping(context.Background()),
+		"the reconnected client was torn down by a stale failure")
+	assert.Equal(t, 2, server.connections(), "the stale teardown forced a third connection")
+
+	// The connection the caller did fail on is still torn down.
+	torn, err = client.disconnectGeneration(client.connGeneration)
+	require.NoError(t, err)
+	assert.True(t, torn)
+	assert.Equal(t, iggcon.TransportStateDisconnected, client.transportState)
+}
+
 // The sign-in transaction holds registerMtx across its reconnect, and an
 // attempt started by a plain request ends in a sign-in that needs that same
 // lock. Waiting for that attempt closes a cycle -- the owner blocked on

foreign/go/client/tcp/tcp_session_management.go

@@ -167,8 +167,11 @@ func (c *IggyTcpClient) settleOnLeader(ctx context.Context, code uint32, body []
 		// The roster read runs while register holds the sign-in lock, so it must
 		// not enter the reconnect path: the reconnect's automatic sign-in would
 		// deadlock on that lock. The connect scope fails it fast instead.
-		redirect, err := c.HandleLeaderRedirection(
-			context.WithValue(ctx, connectScoped{}, struct{}{}))
+		c.mtx.Lock()
+		generation := c.connGeneration
+		c.mtx.Unlock()
+		redirect, err := c.redirectToLeader(
+			context.WithValue(ctx, connectScoped{}, struct{}{}), generation)
 		if err != nil || !redirect {
 			return settled, err
 		}
@@ -244,7 +247,24 @@ func (c *IggyTcpClient) LogoutUser(ctx context.Context) error {
 	return nil
 }
 
+// HandleLeaderRedirection moves the client to the leader the cluster roster
+// names, tearing down whichever connection it is on.
 func (c *IggyTcpClient) HandleLeaderRedirection(ctx context.Context) (bool, error) {
+	c.mtx.Lock()
+	generation := c.connGeneration
+	c.mtx.Unlock()
+	return c.redirectToLeader(ctx, generation)
+}
+
+// redirectToLeader moves the client to the leader, tearing down the
+// connection generation the redirect was decided on.
+//
+// A caller whose connection was replaced while the roster was being read has
+// nothing left to redirect: closing the replacement would strand the requests
+// running on it, and reporting a redirect would have the caller replay on a
+// node it never chose. It is told no redirect happened, and its re-attempt
+// reads the roster over the connection it now has.
+func (c *IggyTcpClient) redirectToLeader(ctx context.Context, generation uint64) (bool, error) {
 	// Clone current address
 	c.mtx.Lock()
 	currentAddress := c.currentServerAddress
@@ -283,9 +303,14 @@ func (c *IggyTcpClient) HandleLeaderRedirection(ctx context.Context) (bool, erro
 	}
 	c.mtx.Unlock()
 
-	if err = c.disconnect(); err != nil {
+	torn, err := c.disconnectGeneration(generation)
+	if err != nil {
 		return false, err
 	}
+	if !torn {
+		c.logger.Debug("Dropping a redirect decided on a replaced connection.")
+		return false, nil
+	}
 
 	c.mtx.Lock()
 	c.leaderRedirectionState.IncrementRedirect(leaderAddress)