diff --git a/foreign/go/client/tcp/tcp_core.go b/foreign/go/client/tcp/tcp_core.go index 057907d61b..651c6475c9 100644 --- a/foreign/go/client/tcp/tcp_core.go +++ b/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 } diff --git a/foreign/go/client/tcp/tcp_failover_test.go b/foreign/go/client/tcp/tcp_failover_test.go index e7946a462e..c070cc32e5 100644 --- a/foreign/go/client/tcp/tcp_failover_test.go +++ b/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 diff --git a/foreign/go/client/tcp/tcp_session_management.go b/foreign/go/client/tcp/tcp_session_management.go index 4da1fb23b5..6b585fc220 100644 --- a/foreign/go/client/tcp/tcp_session_management.go +++ b/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)