You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by hg...@apache.org on 2026/08/31 06:31:29 UTC
(iggy) branch fix/go-sdk-stale-reconnect-teardown updated (1ac46f1c0 -> dc3db1bf3)
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a change to branch fix/go-sdk-stale-reconnect-teardown
in repository https://gitbox.apache.org/repos/asf/iggy.git
omit 1ac46f1c0 fix(go): tear down only the connection a request failed on
add db14a4b17 fix(cluster): make the advertised leader accept partition writes (#3985)
add 90e38b10f feat(node): support connection string syntax for nodejs client creation (#3917)
add 996ac04c8 fix(cluster): recover writes after a fast partition primary rejoin (#3987)
add 5916e6f58 feat(partitions): persist log and index concurrently, harden recovery (#3970)
add eb410a96a fix(server): apply thread pool limit carve-out to all of macOS (#3994)
new dc3db1bf3 fix(go): tear down only the connection a request failed on
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (1ac46f1c0)
\
N -- N -- N refs/heads/fix/go-sdk-stale-reconnect-teardown (dc3db1bf3)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
Cargo.lock | 20 +-
Cargo.toml | 8 +-
bdd/python/uv.lock | 2 +-
bdd/rust/tests/steps/streams.rs | 22 +-
core/ai/mcp/Cargo.toml | 2 +-
core/bench/Cargo.toml | 2 +-
core/bench/report/src/types/group_metrics.rs | 28 +-
core/bench/report/src/types/individual_metrics.rs | 94 +-
core/bench/report/src/utils.rs | 30 +-
core/bench/src/analytics/metrics/group.rs | 48 +-
core/bench/src/analytics/metrics/individual.rs | 65 +-
core/bench/src/args/common.rs | 17 +-
core/bench/src/args/examples.rs | 6 +
core/bench/src/args/kinds/balanced/producer.rs | 19 +-
core/bench/src/benchmarks/benchmark.rs | 13 +-
core/binary_protocol/Cargo.toml | 2 +-
core/cli/Cargo.toml | 2 +-
core/common/Cargo.toml | 2 +-
core/connectors/runtime/Cargo.toml | 2 +-
core/consensus/src/impls.rs | 116 +-
core/harness_derive/src/attrs.rs | 12 +
core/harness_derive/src/codegen.rs | 6 +
core/integration/src/harness/disk.rs | 19 +-
.../src/harness/handle/client_builder.rs | 61 +-
.../src/harness/orchestrator/harness.rs | 145 +-
.../tests/cluster/crash_recovery_corruption.rs | 269 ++
.../tests/cluster/fast_primary_rejoin.rs | 472 ++++
core/integration/tests/cluster/mod.rs | 3 +
.../tests/cluster/partition_primary_routing.rs | 198 ++
.../tests/cluster/staggered_bootstrap.rs | 180 ++
core/integration/tests/server/mod.rs | 2 +-
core/partitions/src/iggy_index.rs | 6 +
core/partitions/src/iggy_index_writer.rs | 24 +-
core/partitions/src/iggy_partition.rs | 906 +++++--
core/partitions/src/journal.rs | 14 +
core/partitions/src/lib.rs | 6 +-
core/partitions/src/log.rs | 74 +-
core/partitions/src/messages_writer.rs | 35 +-
core/partitions/src/poll_plan.rs | 72 +-
core/partitions/src/state_transfer.rs | 8 +
core/partitions/src/types.rs | 35 +-
core/sdk/Cargo.toml | 2 +-
core/sdk/src/leader_aware.rs | 521 ++++
core/sdk/src/quic/quic_client.rs | 370 ++-
core/sdk/src/tcp/tcp_client.rs | 373 ++-
core/sdk/src/vsr.rs | 35 +-
core/sdk/src/websocket/websocket_client.rs | 423 +++-
core/server/Cargo.toml | 2 +-
core/server/config.toml | 77 +-
core/server/src/bootstrap.rs | 87 +-
core/server/src/cluster_meta.rs | 8 +-
core/server/src/dispatch.rs | 9 +-
core/server/src/http.rs | 37 +-
core/server/src/http/forward.rs | 201 +-
core/server/src/partition_helpers.rs | 45 +-
core/server/src/partition_reconciler.rs | 170 +-
core/server/src/responses.rs | 2 +-
core/server/src/segment_recovery.rs | 2566 ++++++++++++++++----
core/server/src/server_error.rs | 133 +-
core/server_common/src/executor.rs | 2 +-
core/shard/src/lib.rs | 236 +-
core/shard/src/router.rs | 110 +-
core/simulator/src/lib.rs | 150 +-
examples/node/package-lock.json | 2 +-
examples/python/uv.lock | 2 +-
foreign/cpp/Cargo.toml | 2 +-
foreign/cpp/MODULE.bazel | 2 +-
.../Implementations/TcpMessageStream.Vsr.cs | 85 +-
.../IggyClient/Implementations/TcpMessageStream.cs | 17 +-
foreign/csharp/Iggy_SDK/Iggy_SDK.csproj | 2 +-
.../VsrTests/EndpointFailoverTests.cs | 128 +-
foreign/go/client/tcp/tcp_core.go | 47 +-
foreign/go/client/tcp/tcp_failover_test.go | 61 +
foreign/go/client/tcp/tcp_session_management.go | 70 +
foreign/go/contracts/version.go | 2 +-
.../iggy/client/async/tcp/AsyncIggyTcpClient.java | 85 +-
.../iggy/client/async/tcp/AsyncTcpConnection.java | 55 +-
.../iggy/client/async/tcp/LeaderAwareness.java | 33 +
.../AsyncIggyTcpClientTransientFailoverTest.java | 126 +-
.../iggy/client/async/tcp/LeaderAwarenessTest.java | 42 +
foreign/node/README.md | 39 +
foreign/node/package-lock.json | 4 +-
foreign/node/package.json | 2 +-
foreign/node/src/client/client.config.test.ts | 44 +
foreign/node/src/client/client.config.ts | 33 +-
.../src/client/client.connection-string.test.ts | 310 +++
.../node/src/client/client.connection-string.ts | 226 ++
foreign/node/src/client/client.connection.test.ts | 144 +-
foreign/node/src/client/client.connection.ts | 71 +-
foreign/node/src/client/client.socket.test.ts | 139 ++
foreign/node/src/client/client.socket.ts | 103 +-
foreign/node/src/client/client.ts | 57 +-
foreign/node/src/client/client.type.ts | 25 +-
foreign/node/src/client/index.ts | 1 +
.../theme/mod.rs => foreign/node/src/constant.ts | 3 +-
foreign/node/src/duration.utils.test.ts | 180 ++
foreign/node/src/duration.utils.ts | 393 +++
foreign/node/src/e2e/tcp.connection-string.e2e.ts | 146 ++
foreign/node/src/e2e/tls.system.e2e.ts | 64 +
foreign/node/src/stream/consumer-stream.ts | 6 +-
foreign/php/Cargo.toml | 2 +-
foreign/python/Cargo.toml | 4 +-
foreign/python/pyproject.toml | 2 +-
foreign/python/uv.lock | 2 +-
104 files changed, 10091 insertions(+), 1276 deletions(-)
create mode 100644 core/integration/tests/cluster/fast_primary_rejoin.rs
create mode 100644 core/integration/tests/cluster/partition_primary_routing.rs
create mode 100644 core/integration/tests/cluster/staggered_bootstrap.rs
create mode 100644 foreign/node/src/client/client.connection-string.test.ts
create mode 100644 foreign/node/src/client/client.connection-string.ts
copy core/bench/dashboard/frontend/src/components/theme/mod.rs => foreign/node/src/constant.ts (89%)
create mode 100644 foreign/node/src/duration.utils.test.ts
create mode 100644 foreign/node/src/duration.utils.ts
create mode 100644 foreign/node/src/e2e/tcp.connection-string.e2e.ts
(iggy) 01/01: fix(go): tear down only the connection a request failed on
Posted by hg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch fix/go-sdk-stale-reconnect-teardown
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit dc3db1bf38bd7d0d5483c270c40505b410172f40
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Fri Aug 28 12:16:09 2026 +0200
fix(go): tear down only the connection a request failed on
Two requests failing at once are one reconnect, but the
teardown that precedes it went by transport state alone.
Connect marks the client connected before it signs in, so
the second caller -- parked on c.mtx for the length of that
sign-in -- woke to a healthy-looking state and closed the
socket the first caller's reconnect had just established.
That left the first caller replaying over nothing, failing
it with ErrNotConnected, and started a second attempt for a
reconnect that had already happened.
Stamp every installed connection with a generation and carry
it out of the attempt that ran on it, so the teardown is a
no-op once the connection has been replaced. The leader
redirect took the same stale teardown and now drops a
redirect decided on a connection that is gone.
Surfaced as a flaky
TestConnect_ConcurrentReconnectsThroughExchangeShareOneAttempt.
---
foreign/go/client/tcp/tcp_core.go | 84 ++++++++++++++++++-------
foreign/go/client/tcp/tcp_failover_test.go | 35 +++++++++++
foreign/go/client/tcp/tcp_session_management.go | 31 ++++++++-
3 files changed, 125 insertions(+), 25 deletions(-)
diff --git a/foreign/go/client/tcp/tcp_core.go b/foreign/go/client/tcp/tcp_core.go
index 0bd8b3600..f52a9d861 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.
@@ -514,7 +519,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
}
@@ -558,7 +563,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
@@ -577,7 +582,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
@@ -631,12 +637,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)
@@ -659,13 +670,13 @@ 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
@@ -675,9 +686,9 @@ func (c *IggyTcpClient) sendFrame(ctx context.Context, code uint32, frame []byte
walked := false
if !walkingRoster {
var redirectErr error
- redirect, redirectErr = c.HandleLeaderRedirection(ctx)
+ redirect, redirectErr = c.redirectToLeader(ctx, generation)
if redirectErr != nil {
- return nil, redirectErr
+ return nil, generation, redirectErr
}
}
if !redirect {
@@ -691,7 +702,7 @@ func (c *IggyTcpClient) sendFrame(ctx context.Context, code uint32, frame []byte
var walkErr error
walked, walkErr = c.settleOnNextEndpoint(visitedRosterEndpoints)
if walkErr != nil {
- return nil, walkErr
+ return nil, generation, walkErr
}
if walked {
walkingRoster = true
@@ -711,41 +722,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 {
@@ -754,7 +767,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
}
@@ -789,10 +802,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
@@ -1134,6 +1147,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
@@ -1388,7 +1402,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 1a035deee..9a0ca663e 100644
--- a/foreign/go/client/tcp/tcp_failover_test.go
+++ b/foreign/go/client/tcp/tcp_failover_test.go
@@ -748,6 +748,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 a34d05565..e62fbc965 100644
--- a/foreign/go/client/tcp/tcp_session_management.go
+++ b/foreign/go/client/tcp/tcp_session_management.go
@@ -176,8 +176,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
}
@@ -253,7 +256,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
@@ -292,9 +312,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)