hubcio commented on code in PR #3944:
URL: https://github.com/apache/iggy/pull/3944#discussion_r3851566653
##########
core/sdk/src/tcp/tcp_client.rs:
##########
@@ -387,6 +441,10 @@ impl TcpClient {
error!("Failed to establish TCP connection to the server:
{error}",);
IggyError::CannotEstablishConnection
})?;
+ // The endpoint that answered is where this client now lives:
+ // the leader check compares against it, and the next
+ // reconnect starts from it.
+ *self.current_server_address.lock().await =
server_address.clone();
Review Comment:
this runs before the tls handshake, so a node that accepts tcp but fails tls
becomes sticky: the `?` on the handshake aborts the sweep and the next
`connect()` leads with the same node, survivors never get dialed. set it where
the loop breaks out with a working stream and treat a handshake failure as a
failed dial (still return `CannotEstablishConnection` once the sweep is
exhausted, otherwise `max_retries = None` loops forever on a bad ca).
##########
core/sdk/src/tcp/tcp_client.rs:
##########
@@ -155,20 +178,30 @@ impl BinaryTransport for TcpClient {
return Err(error);
}
+ // A stale-client eviction is the server ending this session
+ // authoritatively, like a logout: the remembered sign-in ends with it,
+ // so only a configured auto-login may bring the session back.
+ // Remembered credentials exist for transport loss, where the session
+ // died with the socket rather than by anyone's decision.
+ if matches!(error, IggyError::StaleClient) {
+ self.forget_session_credentials().await;
+ }
+
if !self.config.reconnection.enabled {
return Err(IggyError::Disconnected);
}
- if matches!(self.config.auto_login, AutoLogin::Disabled) &&
!is_login_register_code(code) {
- // Without auto-login a reconnect cannot re-establish the session,
- // so non-login requests fail fast. Login/register itself is the
+ if !is_login_register_code(code) &&
self.sign_in_credentials().await.is_none() {
Review Comment:
the replay after `Disconnected` / `EmptyResponse` runs under a fresh session
(new `client_id` from `reset_vsr_session`), so the server's dedup fence can't
match it - a `send_messages` that committed before the reply was lost applies
twice on the survivor. this used to be limited to `AutoLogin::Enabled`, now
it's every signed-in client. replay only for pre-write errors (`NotConnected`,
`CannotEstablishConnection`) and non-replicated ops, or document at-least-once
on failover.
##########
core/sdk/src/tcp/tcp_client.rs:
##########
@@ -320,34 +363,36 @@ impl TcpClient {
}
self.set_state(ClientState::Connecting).await;
- if let Some(connected_at) =
self.connected_at.lock().await.as_ref() {
- let now = IggyTimestamp::now();
- let elapsed = now.as_micros() - connected_at.as_micros();
- let interval =
self.config.reconnection.reestablish_after.as_micros();
- trace!(
- "Elapsed time since last connection: {}",
- IggyDuration::from(elapsed)
- );
- if elapsed < interval {
- let remaining = IggyDuration::from(interval - elapsed);
- info!("Trying to connect to the server in: {remaining}",);
- sleep(remaining.get_duration()).await;
- }
+ let candidates = self.dial_candidates().await;
+ // The reestablish delay paces reconnects to the one endpoint a
+ // single-address client has. With other endpoints known there is
+ // somewhere else to go, and pausing first only pushes the
+ // failover past the window the caller is willing to wait; the
+ // retry interval still paces the loop.
+ let reestablish_wait = if candidates.len() > 1 {
Review Comment:
`reestablish_after` is silently skipped whenever there is more than one
candidate - every clustered client after its first leader check, even when it
redials the same node first. `with_reestablish_after` still promises a
cooldown. apply the wait to the current endpoint only, or document it.
##########
core/sdk/src/tcp/tcp_client.rs:
##########
@@ -155,20 +178,30 @@ impl BinaryTransport for TcpClient {
return Err(error);
}
+ // A stale-client eviction is the server ending this session
+ // authoritatively, like a logout: the remembered sign-in ends with it,
+ // so only a configured auto-login may bring the session back.
+ // Remembered credentials exist for transport loss, where the session
+ // died with the socket rather than by anyone's decision.
+ if matches!(error, IggyError::StaleClient) {
+ self.forget_session_credentials().await;
Review Comment:
why forget here? `StaleClient` is only ever sent by the heartbeat verifier,
whose comment says it evicts so the client can fail fast and reconnect - a gc
pause or a laptop sleep is not caller intent. after this an `IggyConsumer` on a
manually signed-in client sets `can_poll = false` and waits for a sign-in that
never comes, while an auto-login client recovers. that is the same asymmetry
this PR removes elsewhere.
##########
core/sdk/src/tcp/tcp_client.rs:
##########
@@ -358,6 +403,15 @@ impl TcpClient {
return Err(IggyError::CannotEstablishConnection);
Review Comment:
with `reconnection.enabled = false` this returns on the first failed dial,
before `candidate += 1`, so `failover_addresses` are never tried with
auto-reconnect off. sweep every candidate once, then bail. also every early
return here leaves the state at `Connecting`, so a later `connect()` is a no-op
- it returns ok at the top without dialing.
##########
foreign/go/client/tcp/tcp_core.go:
##########
@@ -994,6 +991,47 @@ func (c *IggyTcpClient) Connect(ctx context.Context) error
{
return nil
}
+// dialCandidate opens one connection, wrapping it in TLS when configured, and
+// records the endpoint that answered: the leader check compares against it and
+// the next reconnect starts from it.
+func (c *IggyTcpClient) dialCandidate(ctx context.Context, address string)
(net.Conn, error) {
+ c.logger.Info("Iggy client is connecting to server...",
slog.String("server_address", address))
+ connection, err := (&net.Dialer{}).DialContext(ctx, "tcp", address)
+ if err != nil {
+ c.logger.Error("Failed to establish TCP connection to the
server", slog.Any("error", err))
+ return nil, ierror.ErrCannotEstablishConnection
+ }
+
+ tc := connection.(*net.TCPConn)
+ if err := tc.SetNoDelay(c.config.noDelay); err != nil {
+ c.logger.Error("Failed to set the nodelay option on the client,
continuing...", slog.Any("error", err))
+ }
+
+ c.mtx.Lock()
+ c.clientAddress = tc.LocalAddr().String()
+ c.currentServerAddress = address
Review Comment:
written before the tls handshake, so when a candidate fails the handshake
the client ends up on it and the next pass leads with it. move this after the
handshake.
##########
foreign/go/client/tcp/tcp_failover_test.go:
##########
@@ -0,0 +1,190 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package tcp
+
+import (
+ "context"
+ "log/slog"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/apache/iggy/foreign/go/internal/command"
+ "github.com/apache/iggy/foreign/go/internal/vsr"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+// The node a client signed in on dies; its next request has to complete on a
+// survivor the roster named, under the identity a fresh sign-in binds there.
+// Mirrors `core/integration/tests/cluster/failover_client_continuity.rs`.
+func TestFailover_ResumesOnASurvivorAfterTheSignedInNodeDies(t *testing.T) {
+ var survivor *testListener
+ var primary *testListener
+ var primaryDead atomic.Bool
+
+ survivor = listenVSR(t, nil, func(_, _ int, read request) []byte {
+ switch {
+ case read.code() == uint32(command.GetClusterMetadataCode):
+ return clusterMetadataFrame(t, 1, primary.address(),
survivor.address())
+ case read.operation() == vsr.OperationRegister:
+ return registerReplyFrame(7, 512)
+ default:
+ return replyFrame(vsr.OperationNonReplicated, nil)
+ }
+ })
+
+ primary = listenVSR(t, nil, func(_, _ int, read request) []byte {
+ // A dead node answers nothing; returning nil drops the
connection the
+ // way a killed process does.
+ if primaryDead.Load() {
+ return nil
+ }
+ switch {
+ case read.code() == uint32(command.GetClusterMetadataCode):
+ // The primary leads, so the sign-in settles here and
the roster is
+ // only remembered -- not acted on -- until the node
dies.
+ return clusterMetadataFrame(t, 0, primary.address(),
survivor.address())
+ case read.operation() == vsr.OperationRegister:
+ return registerReplyFrame(7, 128)
+ default:
+ return replyFrame(vsr.OperationNonReplicated, nil)
+ }
+ })
+
+ // No auto-login: the credentials come from the caller's own sign-in,
which
+ // is the shape that could not reconnect at all before.
+ client := newDialingClient(t, primary.address())
+ ctx := context.Background()
+ require.NoError(t, client.Connect(ctx))
+ _, err := client.LoginUser(ctx, "iggy", "iggy")
+ require.NoError(t, err)
+ require.NoError(t, client.Ping(ctx), "the live primary answers")
+ require.Equal(t, primary.address(), client.currentServerAddress)
+
+ primaryDead.Store(true)
+ require.NoError(t, primary.listener.Close(), "stop accepting, so a
redial is refused")
+
+ require.NoError(t, client.Ping(ctx),
+ "the client has to resume on the survivor the roster named")
+
+ assert.Equal(t, survivor.address(), client.currentServerAddress,
+ "the client moved off the dead endpoint")
+ assert.True(t, client.session.Bound(), "the session was re-established")
+
+ var registers int
+ for _, read := range survivor.recorded() {
+ if read.operation() == vsr.OperationRegister {
+ registers++
+ }
+ }
+ assert.Equal(t, 1, registers,
+ "the remembered credentials signed in again on the survivor")
+}
+
+// Without any credentials there is nothing to sign in with, so a request on a
+// dead node fails instead of reconnecting into an unauthenticated session.
+func TestFailover_FailsFastWhenNothingEverSignedIn(t *testing.T) {
+ var server *testListener
+ var dead atomic.Bool
+ server = listenVSR(t, nil, func(_, _ int, read request) []byte {
+ if dead.Load() {
+ return nil
+ }
+ return singleNodeHandler(t, func() string { return
server.address() })(0, 0, read)
+ })
+
+ client := newDialingClient(t, server.address())
+ ctx := context.Background()
+ require.NoError(t, client.Connect(ctx))
+ require.NoError(t, client.Ping(ctx))
+
+ dead.Store(true)
+ require.NoError(t, server.listener.Close())
+
+ assert.Error(t, client.Ping(ctx),
+ "a client that never signed in cannot restore a session by
reconnecting")
+}
+
+// A stale-client eviction is the server ending the session authoritatively,
+// like a logout: the remembered sign-in must not resurrect it, so the evicted
+// request surfaces the loss instead of reconnecting into a fresh session.
+func TestFailover_ServerEvictionForgetsTheRememberedSignIn(t *testing.T) {
+ var server *testListener
+ var evict atomic.Bool
+ server = listenVSR(t, nil, func(_, _ int, read request) []byte {
+ if read.operation() == vsr.OperationRegister {
+ return registerReplyFrame(7, 128)
+ }
+ if evict.Load() {
+ return evictionFrame(vsr.EvictionStaleClient, 0, 0)
+ }
+ if read.code() == uint32(command.GetClusterMetadataCode) {
+ return clusterMetadataFrame(t, 0, server.address())
+ }
+ return replyFrame(vsr.OperationNonReplicated, nil)
+ })
+
+ client := newDialingClient(t, server.address())
+ ctx := context.Background()
+ require.NoError(t, client.Connect(ctx))
+ _, err := client.LoginUser(ctx, "iggy", "iggy")
+ require.NoError(t, err)
+ connectionsBefore := server.connections()
+
+ evict.Store(true)
+ require.Error(t, client.Ping(ctx), "the evicted request surfaces the
loss")
+
+ _, remembered := client.signInCredentials()
+ assert.False(t, remembered, "the eviction forgot the remembered
sign-in")
+ assert.Equal(t, connectionsBefore, server.connections(),
+ "no reconnect dial resurrected the evicted session")
+}
+
+// An explicit sign-out is caller intent: the reconnect must not sign back in
+// with the credentials the earlier sign-in used.
+func TestFailover_DoesNotResurrectASignedOutSession(t *testing.T) {
+ var server *testListener
+ server = listenVSR(t, nil, singleNodeHandler(t, func() string { return
server.address() }))
+
+ client := newDialingClient(t, server.address())
+ ctx := context.Background()
+ require.NoError(t, client.Connect(ctx))
+ _, err := client.LoginUser(ctx, "iggy", "iggy")
+ require.NoError(t, err)
+ require.NoError(t, client.LogoutUser(ctx))
+
+ credentials, ok := client.signInCredentials()
+ assert.False(t, ok, "the sign-out forgot them")
+ assert.Empty(t, credentials.username)
+}
+
+func TestFailover_LeavesTheReestablishPauseToSingleEndpointClients(t
*testing.T) {
Review Comment:
this never calls `Connect`, so it doesn't pin the pause skip: deleting `&&
len(candidates) == 1` in `tcp_core.go` keeps the whole package green. drive
`Connect` under a short deadline and assert the elapsed time.
##########
foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClient.java:
##########
@@ -585,7 +616,7 @@ private CompletableFuture<Void> redialAttempt(int attempt,
RetryPolicy policy) {
log.error("Redial gave up after {} attempts, next request will
fail fast", policy.getMaxRetries());
return CompletableFuture.completedFuture(null);
}
- ConnectionInfo target = ReconnectPlan.target(connectionInfo,
seedConnectionInfo, attempt);
+ ConnectionInfo target = ReconnectPlan.target(redialCandidates(),
attempt);
Review Comment:
one candidate per attempt, with `ReconnectPlan.delay` slept before every
dial and every dial counted against `maxRetries`. with the default 12 x 5s
policy a 2-node survivor is first dialed after two delays, and a 3-node roster
gets four rotations in total. the other sdks sweep all candidates inside one
attempt and sleep once per rotation - do the same here.
##########
foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClient.java:
##########
@@ -585,7 +616,7 @@ private CompletableFuture<Void> redialAttempt(int attempt,
RetryPolicy policy) {
log.error("Redial gave up after {} attempts, next request will
fail fast", policy.getMaxRetries());
return CompletableFuture.completedFuture(null);
}
- ConnectionInfo target = ReconnectPlan.target(connectionInfo,
seedConnectionInfo, attempt);
+ ConnectionInfo target = ReconnectPlan.target(redialCandidates(),
attempt);
Duration delay = ReconnectPlan.delay(policy, attempt);
Review Comment:
`redialCandidates()` on the line above runs on the netty event loop
(`onConnectionFailure` -> `redialAttempt(1)` is synchronous) and
`resolveToSameHost` does a blocking `InetAddress.getAllByName` per seed/roster
pair. dedup without dns like the rust `is_same_address`; moving it into the
delayed stage would only put blocking io on the common pool.
##########
foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClient.java:
##########
@@ -617,6 +648,12 @@ private CompletableFuture<Void> redialAttempt(int attempt,
RetryPolicy policy) {
* again before Register when the redialed node is not the leader.
*/
private CompletableFuture<Void> replayLogin() {
+ Supplier<CompletableFuture<IdentityInfo>> replay = rememberedLogin;
Review Comment:
remembered wins over builder credentials here; rust and go take the
configured ones first. flip it or say why.
##########
foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClient.java:
##########
@@ -638,6 +675,9 @@ CompletableFuture<IdentityInfo>
loginOnLeader(Supplier<CompletableFuture<Identit
CompletableFuture<IdentityInfo> callerFuture = new
CompletableFuture<>();
transaction.whenComplete((identity, error) -> {
gate.complete(null);
+ if (error == null) {
+ rememberedLogin = loginAttempt;
Review Comment:
nothing clears this on a `StaleClient` eviction; the eviction reaches the
client only through the reason-blind `sessionResetListener`. and even with a
forget here, `AsyncTcpConnection` keeps its `loginPayload` (released only on
login failure or close) and replays it on the next channel, so the evicted
session is resurrected either way. plumb `REASON_STALE_CLIENT` to the listener
and drop both there.
##########
foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/LeaderAwarenessTest.java:
##########
@@ -276,6 +281,22 @@ void shouldGiveUpWhenMetadataFetchThrowsSynchronously() {
assertThat(leader).isEmpty();
}
+ @Test
+ void shouldRememberEveryNodeTheRosterNamesEvenWhileLeaderless() {
Review Comment:
no test for the port-0 skip in `nodeTargets`, nor for an inconclusive lookup
keeping the last roster (that one lives in `findLeaderElsewhere` on the client,
so it needs a package-private accessor).
##########
foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.cs:
##########
@@ -1206,18 +1296,29 @@ private async Task<IMemoryOwner<byte>>
SendWithResponseAsync(int code, ReadOnlyM
catch (Exception e) when (IsLostConnection(e) && !IsConnecting &&
!_disposed)
{
_logger.LogWarning("Connection lost");
+
+ // A server-side eviction is the server ending this session
authoritatively, like a logout: the
+ // remembered sign-in ends with it, so only a configured auto
login may bring the session back.
+ // Remembered credentials exist for transport loss, where the
session died with the socket rather
+ // than by anyone's decision.
+ if (e is IggyInvalidStatusCodeException { StatusCode:
VsrError.STALE_CLIENT, FromServer: true })
Review Comment:
this never runs when the eviction lands on an in-flight replicated write:
`TcpMessageStream.Vsr.cs` wraps it in `VsrRequestOutcomeUnknownException`,
which matches neither arm of `IsLostConnection`. `_rememberedLogin` survives,
the next request gets `NotConnectedException`, and the reconnect resurrects the
evicted session. also forget in the `VsrSessionEvictedException` branch when
`Verdict is { StatusCode: STALE_CLIENT, FromServer: true }`.
##########
foreign/csharp/Iggy_SDK_Tests/VsrTests/DialCandidatesTests.cs:
##########
@@ -0,0 +1,64 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+using Apache.Iggy.IggyClient.Implementations;
+
+namespace Apache.Iggy.Tests.VsrTests;
+
+/// <summary>
+/// Mirrors the Rust SDK's <c>dial_candidates</c>: a client that loses the
node it is on has to dial the rest
+/// of the cluster, and the two SDKs have to agree on which endpoints
those are and in what order.
+/// </summary>
+public sealed class DialCandidatesTests
+{
+ [Fact]
+ public void LeadsWithTheCurrentEndpointThenNamesEachOtherOneOnce()
+ {
+ var candidates = TcpMessageStream.DialCandidates(
Review Comment:
`localhost:8090` normalizes to the current endpoint, so no test pins
base-before-roster: swapping `Prepend` for `Append` in `DialCandidates` keeps
all four green. add a distinct base with a non-empty roster.
##########
foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.cs:
##########
@@ -1103,6 +1134,17 @@ private async Task TryEstablishConnectionAsync(bool
autoLogin, CancellationToken
throw;
}
+ // Every other endpoint gets its turn before the retry delay:
the node just lost may be gone for
+ // good, and pausing on it helps nothing.
+ if (++candidate < candidates.Length)
Review Comment:
the `MaxRetries` / `!Enabled` checks above run per failed dial, before this
advance, so the last round dials only `candidates[0]` (the dead node) and
`Enabled = false` never sweeps at all. move both checks down next to
`retryCount++`.
##########
foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.cs:
##########
@@ -725,6 +744,9 @@ public async Task LogoutUserAsync(CancellationToken token =
default)
{
await ResetConsensusSessionAsync();
+ // An explicit sign-out leaves no session to restore, and a
reconnect must not resurrect one.
Review Comment:
only true for the remembered login: with `AutoLoginSettings.Enabled`
configured, `SignInSettings()` still returns the configured credentials after a
logout and the next lost-connection reconnect signs in again. pre-existing, but
the comment claims more than the code does.
##########
foreign/csharp/Iggy_SDK_Tests/VsrTests/EndpointFailoverTests.cs:
##########
@@ -0,0 +1,428 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+using System.Buffers.Binary;
+using System.Net;
+using System.Net.Sockets;
+using System.Text;
+using Apache.Iggy.Configuration;
+using Apache.Iggy.Contracts.Tcp;
+using Apache.Iggy.Enums;
+using Apache.Iggy.IggyClient.Implementations;
+using Microsoft.Extensions.Logging.Abstractions;
+
+namespace Apache.Iggy.Tests.VsrTests;
+
+/// <summary>
+/// The node a client signed in on dies; its next request has to complete
on a survivor the roster named,
+/// under a session established there. Mirrors
+/// <c>core/integration/tests/cluster/failover_client_continuity.rs</c>.
+/// </summary>
+public sealed class EndpointFailoverTests
+{
+ private const int HeaderSize = 256;
+ private const int SizeOffset = 48;
+ private const int CommandOffset = 60;
+ private const int RequestIdOffset = 168;
+ private const int RequestOperationOffset = 176;
+ private const int RequestReservedOffset = 196;
+ private const int ReplyRequestIdOffset = 200;
+ private const int ReplyOperationOffset = 208;
+ private const int ReplyStatusOffset = 216;
+
+ private const byte CommandReply = 8;
+ private const byte CommandEviction = 13;
+ private const int EvictionReasonOffset = 255;
+ private const byte EvictionStaleClient = 13;
+ private const byte OperationRegister = 1;
+ private const byte OperationNonReplicated = 2;
+ private const int GetClusterMetadataCode = 12;
+ private const int PingCode = 1;
+
+ [Fact]
+ public async Task ResumesOnASurvivorAfterTheSignedInNodeDies()
+ {
+ using var primary = new MockNode();
+ using var survivor = new MockNode();
+
+ // The primary leads, so the sign-in settles there and the roster is
only remembered - not acted on -
+ // until the node dies.
+ primary.Serve(request => request.Code == GetClusterMetadataCode
+ ? Reply(OperationNonReplicated, ClusterMetadata(primary.Port,
survivor.Port, primary.Port))
+ : Answer(request));
+ survivor.Serve(request => request.Code == GetClusterMetadataCode
+ ? Reply(OperationNonReplicated, ClusterMetadata(primary.Port,
survivor.Port, survivor.Port))
+ : Answer(request));
+
+ var configuration = new IggyClientConfigurator
+ {
+ BaseAddress = $"127.0.0.1:{primary.Port}",
+ Protocol = Protocol.Tcp,
+ ReconnectionSettings = new ReconnectionSettings
+ {
+ Enabled = true,
+ MaxRetries = 4,
+ InitialDelay = TimeSpan.FromMilliseconds(20)
+ }
+ };
+ using var client = new TcpMessageStream(configuration,
NullLoggerFactory.Instance);
+
+ await client.ConnectAsync(TestContext.Current.CancellationToken);
+ // No auto login: the credentials come from the caller's own sign-in,
which is the shape that could not
+ // reconnect at all before.
+ await client.LoginUserAsync("iggy", "iggy",
TestContext.Current.CancellationToken);
+ await client.PingAsync(TestContext.Current.CancellationToken);
+ Assert.Equal(1, primary.Pings);
+
+ primary.Kill();
+
+ // The request in flight when the node died is allowed to fail; what
is not allowed is never completing
+ // one, which is what a client that only knows the dead endpoint does.
+ var (resumed, lastError) = await ResumedWithin(client,
TimeSpan.FromSeconds(10));
+ Assert.True(resumed,
+ $"the client has to resume on the survivor the roster named
({lastError}, survivor saw " +
+ $"{survivor.Registrations} registrations and {survivor.Pings}
pings)");
+ Assert.True(survivor.Registrations >= 1, "the remembered credentials
signed in again on the survivor");
+ Assert.True(survivor.Pings >= 1, "the request landed on the survivor");
+ }
+
+ /// <summary>
+ /// Mirrors the integration contract (HeartbeatTests
+ /// EvictedClient_WithoutAutoLogin_Should_FailFast_And_NotReconnect):
a server-side eviction ends the
+ /// session authoritatively, so the credentials a manual sign-in
remembered must not resurrect it - the
+ /// evicted request surfaces the loss with no reconnect attempt.
+ /// </summary>
+ [Fact]
+ public async Task ServerEvictionForgetsTheRememberedSignIn()
+ {
+ using var node = new MockNode();
+ var evict = false;
+ node.Serve(request =>
+ {
+ if (request.Operation == OperationRegister)
+ {
+ return Reply(OperationRegister, RegisterBody(session: 128));
+ }
+
+ return evict
+ ? EvictionFrame(EvictionStaleClient)
+ : Reply(OperationNonReplicated, request.Code ==
GetClusterMetadataCode
+ ? ClusterMetadata(node.Port, node.Port, node.Port)
+ : []);
+ });
+
+ var configuration = new IggyClientConfigurator
+ {
+ BaseAddress = $"127.0.0.1:{node.Port}",
+ Protocol = Protocol.Tcp,
+ ReconnectionSettings = new ReconnectionSettings
+ {
+ Enabled = true,
+ MaxRetries = 2,
+ InitialDelay = TimeSpan.FromMilliseconds(20)
+ }
+ };
+ using var client = new TcpMessageStream(configuration,
NullLoggerFactory.Instance);
+
+ await client.ConnectAsync(TestContext.Current.CancellationToken);
+ await client.LoginUserAsync("iggy", "iggy",
TestContext.Current.CancellationToken);
+ await client.PingAsync(TestContext.Current.CancellationToken);
+ var connectionsBeforeEviction = node.Connections;
+
+ evict = true;
+ await Assert.ThrowsAnyAsync<Exception>(() =>
client.PingAsync(TestContext.Current.CancellationToken));
+ Assert.Equal(connectionsBeforeEviction, node.Connections);
+
+ // The dropped connection leaves the next call transport-shaped, but
the eviction forgot the remembered
+ // sign-in, so it must fail fast instead of reconnecting into a
resurrected session.
+ await Assert.ThrowsAnyAsync<Exception>(() =>
client.PingAsync(TestContext.Current.CancellationToken));
+ Assert.Equal(connectionsBeforeEviction, node.Connections);
+ }
+
+ private static byte[] EvictionFrame(byte reason)
+ {
+ var frame = new byte[HeaderSize];
+ BinaryPrimitives.WriteUInt32LittleEndian(frame.AsSpan(SizeOffset, 4),
HeaderSize);
+ frame[CommandOffset] = CommandEviction;
+ frame[EvictionReasonOffset] = reason;
+ return frame;
+ }
+
+ [Fact]
+ public async Task FailsFastWhenNothingEverSignedIn()
+ {
+ using var node = new MockNode();
+ node.Serve(request => request.Code == GetClusterMetadataCode
+ ? Reply(OperationNonReplicated, ClusterMetadata(node.Port,
node.Port, node.Port))
+ : Answer(request));
+
+ var configuration = new IggyClientConfigurator
+ {
+ BaseAddress = $"127.0.0.1:{node.Port}",
+ Protocol = Protocol.Tcp,
+ ReconnectionSettings = new ReconnectionSettings
+ {
+ Enabled = true,
+ MaxRetries = 2,
+ InitialDelay = TimeSpan.FromMilliseconds(20)
+ }
+ };
+ using var client = new TcpMessageStream(configuration,
NullLoggerFactory.Instance);
+
+ await client.ConnectAsync(TestContext.Current.CancellationToken);
+ await client.PingAsync(TestContext.Current.CancellationToken);
+
+ node.Kill();
+
+ var (resumed, _) = await ResumedWithin(client,
TimeSpan.FromSeconds(2));
Review Comment:
2s of polling per tfm to prove a negative. assert no `Connecting` transition
via `SubscribeConnectionEvents` instead, like `HeartbeatTests` does.
##########
foreign/node/src/client/client.connection.test.ts:
##########
@@ -393,6 +393,31 @@ describe('IggyConnection', () => {
}
);
+ it('rotates a redial through the roster it learned while connected',
Review Comment:
never redials and the seed is the current endpoint, so order, spelling dedup
and destroy-mid-pass are all untested here. add those cases - the destroy one
would have caught the leaked socket.
##########
foreign/node/src/client/client.socket.test.ts:
##########
@@ -504,6 +504,111 @@ describe('VSR client socket', () => {
}
});
+ // The node a client authenticated on dies; its next command has to complete
+ // on a survivor the roster named, under a session established there.
+ // Mirrors `core/integration/tests/cluster/failover_client_continuity.rs`.
+ it('resumes on a survivor after the node it authenticated on dies',
+ async () => {
+ const primarySockets = new Set<Socket>();
+ let primaryDead = false;
+
+ const survivor = await startVsrServer((frame, socket) => {
+ const operation = frame.readUInt8(REQUEST_OFFSET.operation);
+ if (operation === Operation.Register) {
+ socket.write(replyFrame(Operation.Register, registerReplyBody()));
+ return;
+ }
+ const code = frame.readUInt32LE(REQUEST_OFFSET.reserved);
+ if (code === COMMAND_CODE.GetClusterMetadata) {
+ // The survivor leads once the primary is gone.
+ socket.write(replyFrame(
+ Operation.NonReplicated,
+ twoNodeMetadataBody(primary.port, survivor.port)
+ ));
+ return;
+ }
+ socket.write(replyFrame(operation));
+ });
+
+ const primary = await startVsrServer((frame, socket) => {
+ primarySockets.add(socket);
+ if (primaryDead) {
+ socket.destroy();
+ return;
+ }
+ const operation = frame.readUInt8(REQUEST_OFFSET.operation);
+ if (operation === Operation.Register) {
+ socket.write(replyFrame(Operation.Register, registerReplyBody()));
+ return;
+ }
+ const code = frame.readUInt32LE(REQUEST_OFFSET.reserved);
+ if (code === COMMAND_CODE.GetClusterMetadata) {
+ // The primary leads, so the login settles here and the roster is
+ // only remembered, not acted on, until the node dies.
+ socket.write(replyFrame(
+ Operation.NonReplicated,
+ twoNodeMetadataBody(survivor.port, primary.port)
+ ));
+ return;
+ }
+ socket.write(replyFrame(operation));
+ });
+
+ const config: ClientConfig = {
+ ...vsrConfig(primary.port),
+ reconnect: { enabled: true, interval: 1, maxRetries: 3 }
+ };
+ const client = new CommandResponseStream(config);
+ try {
+ await client.authenticate(config.credentials);
+ await client.sendCommand(60_021, Buffer.alloc(0));
+ assert.ok(
+ primary.frames.some(
+ (frame) => frame.readUInt32LE(REQUEST_OFFSET.reserved) === 60_021
+ ),
+ 'the live primary answered the first command'
+ );
+
+ primaryDead = true;
+ for (const socket of primarySockets)
+ socket.destroy();
+ await primary.close();
+
+ // The attempt in flight when the socket died is allowed to fail; what
+ // is not allowed is never completing one, which is what a client that
+ // only knows the dead endpoint does.
+ let resumed = false;
+ let lastError: unknown;
+ for (let attempt = 0; attempt < 20 && !resumed; attempt += 1) {
Review Comment:
20 attempts with 10ms sleeps passes even if 19 calls fail. attempt 0 fails
and attempt 1 resumes every time here, so cap it at 2 - that also pins the 'at
most one failed submission' promise in the comment.
##########
core/common/src/traits/binary_impls/users.rs:
##########
@@ -218,6 +218,11 @@ impl<B: BinaryClient> UserClient for B {
"authenticated against iggy server"
);
self.set_state(ClientState::Authenticated).await;
+ self.remember_session_credentials(Credentials::UsernamePassword(
Review Comment:
`change_password` never refreshes this, so after a password change the next
drop re-logs in with the old password and an unrelated op fails with
`InvalidCredentials`. on a successful change for the signed-in user swap in the
new password (check it is the same user - `change_password` can target someone
else with `manage_users`).
##########
core/sdk/src/leader_aware.rs:
##########
@@ -162,7 +217,7 @@ fn process_cluster_metadata(
/// Check if two addresses refer to the same endpoint
/// Handles various formats like 127.0.0.1:8090 vs localhost:8090
-fn is_same_address(addr1: &str, addr2: &str) -> bool {
+pub(crate) fn is_same_address(addr1: &str, addr2: &str) -> bool {
Review Comment:
hostname and ip spellings of one node are not deduped (`iggy-server:8090` vs
`10.0.0.5:8090`). a client configured by hostname whose roster reports the
advertised ip gets two candidates for a single node: the dead node is dialed
twice per sweep, and a single-node deployment loses `reestablish_after` and
gets the bounded dial. resolve the host name before giving up on equality.
##########
foreign/go/client/tcp/tcp_core.go:
##########
@@ -923,46 +943,23 @@ func (c *IggyTcpClient) Connect(ctx context.Context)
error {
}),
).Do(
func() error {
- address := candidates[candidateIndex%len(candidates)]
- candidateIndex++
- c.logger.Info("Iggy client is connecting to server...",
slog.String("server_address", address))
- connection, err := (&net.Dialer{}).DialContext(ctx,
"tcp", address)
- if err != nil {
- c.logger.Error("Failed to establish TCP
connection to the server", slog.Any("error", err))
- return ierror.ErrCannotEstablishConnection
- }
-
- tc := connection.(*net.TCPConn)
- if err := tc.SetNoDelay(c.config.noDelay); err != nil {
- c.logger.Error("Failed to set the nodelay
option on the client, continuing...", slog.Any("error", err))
- }
-
- c.mtx.Lock()
- c.clientAddress = tc.LocalAddr().String()
- c.currentServerAddress = address
- c.mtx.Unlock()
+ // Every endpoint gets its turn inside one attempt, so
a full pass
+ // over the cluster costs one retry rather than one per
endpoint:
+ // a pass that stopped at the first refusal would never
reach the
+ // survivors of a client configured for a single retry.
+ var lastErr error
+ for _, address := range candidates {
Review Comment:
with `WithServerAddress("")` `candidates` is empty, `lastErr` stays nil and
`Connect` reports connected with `c.conn == nil`; every request then returns
`ErrNotConnected` while `Connect` keeps answering already connected. reject an
empty candidate list.
##########
foreign/go/client/tcp/tcp_core.go:
##########
@@ -1025,16 +1063,16 @@ func (c *IggyTcpClient) connectionCandidates() []string
{
// backup: once the caller signs in, the first replicated request fails over
// through the transient-deny path.
func (c *IggyTcpClient) establishSession(ctx context.Context, skipAutoLogin
bool) error {
- if !c.config.autoLogin.enabled {
- c.logger.Info("Automatic sign-in is disabled.")
+ credentials, ok := c.signInCredentials()
Review Comment:
self-deadlock on a manual re-login over a dropped transport: `register`
holds `registerMtx` -> `endBoundSession` -> `LogoutUser` -> `exchange`;
`LogoutUserCode` is not in `replicatedOperation` so `canReplay` is true ->
`Connect` -> this finds the remembered login -> `LoginUser` -> `register` ->
`registerMtx.Lock()` forever. reproduced with a fake server that drops the
logout frame; same hang when it answers with a `NoSession` eviction. auto-login
clients already had this, remembered credentials extend it to manual logins.
simplest fix: let `endBoundSession` swallow reconnectable logout errors (the
session died with the socket anyway) and go straight to the sign-in, whose own
replay reconnects.
##########
core/integration/tests/sdk/disconnect_relogin.rs:
##########
@@ -0,0 +1,65 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! An explicit disconnect ends the session for good: the credentials a manual
+//! sign-in remembered (for reconnecting across involuntary drops and
+//! failovers) must not resurrect it. Pins at the Rust layer the contract the
+//! C++ e2e suite asserts through the FFI
(`DisconnectThenReconnectWithoutRelogin`,
+//! `GetStatsBeforeLoginThrows`), so a regression fails here first instead of
+//! three suites downstream.
+
+use iggy::prelude::*;
+use integration::iggy_harness;
+
+#[iggy_harness]
+async fn
given_a_logged_in_client_when_explicitly_disconnected_should_require_a_fresh_login(
+ harness: &TestHarness,
+) {
+ let client = harness.new_client().await.unwrap();
+ client
+ .login_user(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD)
+ .await
+ .unwrap();
+ client.get_me().await.expect("authenticated get_me works");
+
+ client.disconnect().await.unwrap();
+ client.connect().await.unwrap();
+ assert!(
+ client.get_me().await.is_err(),
Review Comment:
`is_err()` passes for any error, a connect failure included. assert
`Unauthenticated` here and `NotConnected` at line 49.
##########
foreign/go/client/tcp/tcp_core.go:
##########
@@ -480,12 +488,22 @@ func (c *IggyTcpClient) exchange(ctx context.Context,
code uint32, frame []byte)
return nil, err
}
- // Without auto-login a reconnect cannot restore the session, so
anything
- // but a sign-in fails here instead of replaying unauthenticated. The
- // sign-in itself is the exception: the server stays silent on a
transient
+ // A stale-client eviction is the server ending this session
+ // authoritatively, like a logout: the remembered sign-in ends with it,
so
+ // only a configured auto-login may bring the session back. Remembered
+ // credentials exist for transport loss, where the session died with the
+ // socket rather than by anyone's decision.
+ if errors.Is(err, ierror.ErrStaleClient) {
Review Comment:
this sits after the `!reconnection.enabled` return, so with reconnection
disabled the eviction never forgets the login and the next manual `Connect`
signs in with the evicted session's credentials. move it right after the
`isReconnectable` check.
##########
core/sdk/src/tcp/tcp_client.rs:
##########
@@ -485,24 +543,24 @@ impl TcpClient {
};
// Handle auto-login
- let should_redirect = match &self.config.auto_login {
- AutoLogin::Disabled => {
- info!("Automatic sign-in is disabled.");
+ let should_redirect = match self.sign_in_credentials().await {
+ None => {
+ info!("No credentials to sign in with.");
// Only `IggyClient` redirects after a manual sign-in, so
// a raw transport can stay on a backup: its first
// replicated write gets `TransientNotAccepted`, the
// redirect drops the session, and the retry fails
// `Unauthenticated` until the caller signs in again.
false
}
- AutoLogin::Enabled(credentials) => {
+ Some(credentials) => {
if skip_auto_login {
info!("Skipping automatic sign-in for a retried
login/register request.");
false
} else {
info!("{NAME} client: {client_address} is signing
in...");
self.set_state(ClientState::Authenticating).await;
- match credentials {
+ match &credentials {
Credentials::UsernamePassword(username, password)
=> {
Review Comment:
after a redirect this now signs in with the remembered credentials, then the
outer `login_user` runs `logout_before_relogin` and logs in again: login,
logout, login and three metadata reads per redirect, each login an argon2 on
the server, where it used to be two logins. most clients seeded with one
address land on a backup first, so this is the common path. skip the re-login
when the state is already `Authenticated` and `auto_login` is disabled (with
`AutoLogin::Enabled(A)` plus a manual `login_user(B)` a plain short-circuit
would leave the client as A).
also, if this login fails the state stays `Authenticating`: every gated op
then fails client-side with `Disconnected`, `connect()` is a no-op and only an
explicit `login_user` recovers. reachable now whenever the remembered password
went stale. reset the state on failure.
##########
foreign/go/client/tcp/tcp_core.go:
##########
@@ -994,6 +991,47 @@ func (c *IggyTcpClient) Connect(ctx context.Context) error
{
return nil
}
+// dialCandidate opens one connection, wrapping it in TLS when configured, and
+// records the endpoint that answered: the leader check compares against it and
+// the next reconnect starts from it.
+func (c *IggyTcpClient) dialCandidate(ctx context.Context, address string)
(net.Conn, error) {
+ c.logger.Info("Iggy client is connecting to server...",
slog.String("server_address", address))
+ connection, err := (&net.Dialer{}).DialContext(ctx, "tcp", address)
Review Comment:
`net.Dialer{}` has no timeout and nothing up the stack adds a deadline, so a
node whose syns are dropped blocks here for the whole kernel connect timeout
(minutes) before the survivor is tried - and the dead node is `candidates[0]`
on every pass. tried it: black-holed address first, live survivor second,
`Connect` under a 4s context expired without ever dialing the survivor; a
refused address first fails over immediately. use `net.Dialer{Timeout:
failoverDialTimeout}` (2s, like rust) when `len(candidates) > 1`.
##########
foreign/go/client/tcp/tcp_failover_test.go:
##########
@@ -0,0 +1,190 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package tcp
+
+import (
+ "context"
+ "log/slog"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/apache/iggy/foreign/go/internal/command"
+ "github.com/apache/iggy/foreign/go/internal/vsr"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+// The node a client signed in on dies; its next request has to complete on a
+// survivor the roster named, under the identity a fresh sign-in binds there.
+// Mirrors `core/integration/tests/cluster/failover_client_continuity.rs`.
+func TestFailover_ResumesOnASurvivorAfterTheSignedInNodeDies(t *testing.T) {
+ var survivor *testListener
+ var primary *testListener
+ var primaryDead atomic.Bool
+
+ survivor = listenVSR(t, nil, func(_, _ int, read request) []byte {
+ switch {
+ case read.code() == uint32(command.GetClusterMetadataCode):
+ return clusterMetadataFrame(t, 1, primary.address(),
survivor.address())
+ case read.operation() == vsr.OperationRegister:
+ return registerReplyFrame(7, 512)
+ default:
+ return replyFrame(vsr.OperationNonReplicated, nil)
+ }
+ })
+
+ primary = listenVSR(t, nil, func(_, _ int, read request) []byte {
+ // A dead node answers nothing; returning nil drops the
connection the
+ // way a killed process does.
+ if primaryDead.Load() {
+ return nil
+ }
+ switch {
+ case read.code() == uint32(command.GetClusterMetadataCode):
+ // The primary leads, so the sign-in settles here and
the roster is
+ // only remembered -- not acted on -- until the node
dies.
+ return clusterMetadataFrame(t, 0, primary.address(),
survivor.address())
+ case read.operation() == vsr.OperationRegister:
+ return registerReplyFrame(7, 128)
+ default:
+ return replyFrame(vsr.OperationNonReplicated, nil)
+ }
+ })
+
+ // No auto-login: the credentials come from the caller's own sign-in,
which
+ // is the shape that could not reconnect at all before.
+ client := newDialingClient(t, primary.address())
+ ctx := context.Background()
+ require.NoError(t, client.Connect(ctx))
+ _, err := client.LoginUser(ctx, "iggy", "iggy")
+ require.NoError(t, err)
+ require.NoError(t, client.Ping(ctx), "the live primary answers")
+ require.Equal(t, primary.address(), client.currentServerAddress)
+
+ primaryDead.Store(true)
+ require.NoError(t, primary.listener.Close(), "stop accepting, so a
redial is refused")
+
+ require.NoError(t, client.Ping(ctx),
+ "the client has to resume on the survivor the roster named")
+
+ assert.Equal(t, survivor.address(), client.currentServerAddress,
+ "the client moved off the dead endpoint")
+ assert.True(t, client.session.Bound(), "the session was re-established")
+
+ var registers int
+ for _, read := range survivor.recorded() {
+ if read.operation() == vsr.OperationRegister {
+ registers++
+ }
+ }
+ assert.Equal(t, 1, registers,
+ "the remembered credentials signed in again on the survivor")
+}
+
+// Without any credentials there is nothing to sign in with, so a request on a
+// dead node fails instead of reconnecting into an unauthenticated session.
+func TestFailover_FailsFastWhenNothingEverSignedIn(t *testing.T) {
+ var server *testListener
+ var dead atomic.Bool
+ server = listenVSR(t, nil, func(_, _ int, read request) []byte {
+ if dead.Load() {
+ return nil
+ }
+ return singleNodeHandler(t, func() string { return
server.address() })(0, 0, read)
+ })
+
+ client := newDialingClient(t, server.address())
+ ctx := context.Background()
+ require.NoError(t, client.Connect(ctx))
+ require.NoError(t, client.Ping(ctx))
+
+ dead.Store(true)
+ require.NoError(t, server.listener.Close())
+
+ assert.Error(t, client.Ping(ctx),
+ "a client that never signed in cannot restore a session by
reconnecting")
+}
+
+// A stale-client eviction is the server ending the session authoritatively,
+// like a logout: the remembered sign-in must not resurrect it, so the evicted
+// request surfaces the loss instead of reconnecting into a fresh session.
+func TestFailover_ServerEvictionForgetsTheRememberedSignIn(t *testing.T) {
+ var server *testListener
+ var evict atomic.Bool
+ server = listenVSR(t, nil, func(_, _ int, read request) []byte {
+ if read.operation() == vsr.OperationRegister {
+ return registerReplyFrame(7, 128)
+ }
+ if evict.Load() {
+ return evictionFrame(vsr.EvictionStaleClient, 0, 0)
+ }
+ if read.code() == uint32(command.GetClusterMetadataCode) {
+ return clusterMetadataFrame(t, 0, server.address())
+ }
+ return replyFrame(vsr.OperationNonReplicated, nil)
+ })
+
+ client := newDialingClient(t, server.address())
+ ctx := context.Background()
+ require.NoError(t, client.Connect(ctx))
+ _, err := client.LoginUser(ctx, "iggy", "iggy")
+ require.NoError(t, err)
+ connectionsBefore := server.connections()
+
+ evict.Store(true)
+ require.Error(t, client.Ping(ctx), "the evicted request surfaces the
loss")
+
+ _, remembered := client.signInCredentials()
+ assert.False(t, remembered, "the eviction forgot the remembered
sign-in")
+ assert.Equal(t, connectionsBefore, server.connections(),
+ "no reconnect dial resurrected the evicted session")
+}
+
+// An explicit sign-out is caller intent: the reconnect must not sign back in
+// with the credentials the earlier sign-in used.
+func TestFailover_DoesNotResurrectASignedOutSession(t *testing.T) {
Review Comment:
never drives a reconnect. drop the conn after the logout, `Connect` +
`Ping`, and assert no second `OperationRegister` reached the server.
##########
foreign/go/client/tcp/tcp_session_management.go:
##########
@@ -33,15 +33,25 @@ func (c *IggyTcpClient) LoginUser(ctx context.Context,
username string, password
if err != nil {
return nil, err
}
- return c.register(ctx, uint32(command.LoginRegisterCode), body)
+ identity, err := c.register(ctx, uint32(command.LoginRegisterCode),
body)
+ if err != nil {
+ return nil, err
+ }
+ c.rememberLogin(NewUsernamePasswordCredentials(username, password))
Review Comment:
`rememberLogin` runs outside `registerMtx` (here and in the pat variant), so
two concurrent sign-ins can remember A while the session is B and the next
reconnect signs in as A. pass the credentials into `register` and remember them
after `settleOnLeader`, under the lock.
##########
foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClient.java:
##########
@@ -129,10 +132,38 @@ public class AsyncIggyTcpClient {
private final Optional<File> tlsCertificate;
private final TcpConnectionPoolConfig poolConfig;
private final ClientRoutingState routingState = new ClientRoutingState();
+ private final LoginRoutingHook loginRoutingHook = new LoginRoutingHook() {
+
+ @Override
+ public CompletableFuture<IdentityInfo>
loginOnLeader(Supplier<CompletableFuture<IdentityInfo>> loginAttempt) {
+ return AsyncIggyTcpClient.this.loginOnLeader(loginAttempt);
+ }
+
+ @Override
+ public void forgetLogin() {
+ rememberedLogin = null;
Review Comment:
`close()` doesn't call this, and `connect()` resets `closed = false`, so
`login(A) -> close() -> connect() -> loss` replays A. null it in `close()` too
(rust `disconnect` forgets).
##########
foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClient.java:
##########
@@ -585,7 +616,7 @@ private CompletableFuture<Void> redialAttempt(int attempt,
RetryPolicy policy) {
log.error("Redial gave up after {} attempts, next request will
fail fast", policy.getMaxRetries());
return CompletableFuture.completedFuture(null);
}
- ConnectionInfo target = ReconnectPlan.target(connectionInfo,
seedConnectionInfo, attempt);
+ ConnectionInfo target = ReconnectPlan.target(redialCandidates(),
attempt);
Duration delay = ReconnectPlan.delay(policy, attempt);
Executor delayedExecutor =
CompletableFuture.delayedExecutor(delay.toMillis(), TimeUnit.MILLISECONDS);
Review Comment:
below, a rejected replay login (rotated password, expired pat) lands in the
same `handle` as a transport failure and triggers `redialAttempt(attempt + 1)`:
the freshly published good connection is torn down by the next retarget, up to
12 times, and the loop ends with the client connected but unauthenticated. stop
redialing when the retarget succeeded and only the login failed, and drop
`rememberedLogin` on a non-transient login error.
##########
foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/LeaderAwareness.java:
##########
@@ -248,6 +265,24 @@ private static boolean
reachesOnlyLocalMachine(InetAddress[] addresses) {
/**
Review Comment:
orphaned: this javadoc belonged to `LeaderCheck` below, the new record got
inserted between them.
##########
foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.cs:
##########
@@ -82,6 +82,17 @@ public sealed partial class TcpMessageStream : IIggyClient
private DateTimeOffset _lastConnectionTime;
private int _stateValue = (int)ConnectionState.Disconnected;
+ // Every node the roster named on the last read, kept as dial candidates.
A node dies together with its
+ // address, and the roster is unreachable exactly when it is needed, so
the client has to have remembered it
+ // while the connection was still healthy. Written by the leader probe,
read by the connect loop.
+ private string[] _rosterAddresses = [];
+
+ // The credentials a sign-in succeeded with, so a reconnect - on this node
or, after a failover, another one -
+ // can re-establish the session instead of leaving every later request
unauthenticated. A caller that signs in
+ // by hand is otherwise less reconnectable than one that configures auto
login, which is a surprising
+ // difference between two ways of doing the same thing. Cleared on
sign-out and on a server eviction.
+ private AutoLoginSettings? _rememberedLogin;
Review Comment:
`Dispose()` doesn't null this. minor, but it holds a plain-string password /
pat.
##########
foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.cs:
##########
@@ -999,15 +1027,18 @@ private async Task TryEstablishConnectionAsync(bool
autoLogin, CancellationToken
var retryCount = 0;
var redirects = 0;
var delay = _configuration.ReconnectionSettings.InitialDelay;
+
+ if (string.IsNullOrEmpty(_currentAddress))
+ {
+ _currentAddress = _configuration.BaseAddress;
+ }
+
+ var candidates = DialCandidates();
Review Comment:
`socket.ConnectAsync(host, port, token)` at line 1060 has no timeout and
nothing up the stack adds one, so a node whose syns are dropped blocks the
sweep for the whole kernel connect timeout (minutes) before a survivor is tried
- and the dead node is `candidates[0]` on every pass, with
`_connectionSemaphore` held. dial under a linked cts with `CancelAfter(2s)`
when `candidates.Length > 1`. the timeout surfaces as
`OperationCanceledException`, which the catch filter at 1111 treats as fatal,
so gate on `!token.IsCancellationRequested` to advance the candidate instead.
##########
foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClient.java:
##########
@@ -617,6 +648,12 @@ private CompletableFuture<Void> redialAttempt(int attempt,
RetryPolicy policy) {
* again before Register when the redialed node is not the leader.
Review Comment:
stale: this replays the remembered login first, and the javadoc on
`redialAttempt` still says it alternates current and seed and that pat logins
can't be replayed - neither is true anymore.
##########
foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClientEndpointFailoverTest.java:
##########
@@ -0,0 +1,318 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iggy.client.async.tcp;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import org.apache.iggy.config.RetryPolicy;
+import org.junit.jupiter.api.Test;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.InetAddress;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * The node a client signed in on dies; its next request has to complete on a
+ * survivor the roster named, under a session established there. Mirrors
+ * {@code core/integration/tests/cluster/failover_client_continuity.rs}. The
+ * mock VSR framing matches {@link AsyncIggyTcpClientTransientFailoverTest},
+ * kept separate so a death mid-connection cannot disturb that suite's server.
+ */
+class AsyncIggyTcpClientEndpointFailoverTest {
+ private static final int HEADER_SIZE = 256;
+ private static final int SIZE_OFFSET = 48;
+ private static final int COMMAND_OFFSET = 60;
+ private static final int REQUEST_ID_OFFSET = 168;
+ private static final int REQUEST_OPERATION_OFFSET = 176;
+ private static final int REQUEST_CODE_OFFSET = 196;
+ private static final int REPLY_REQUEST_ID_OFFSET = 200;
+ private static final int REPLY_OPERATION_OFFSET = 208;
+ private static final int REPLY_STATUS_OFFSET = 216;
+
+ private static final int COMMAND_REPLY = 8;
+ private static final int OPERATION_REGISTER = 1;
+ private static final int OPERATION_NON_REPLICATED = 2;
+ private static final int GET_CLUSTER_METADATA_CODE = 12;
+ private static final int PING_CODE = 1;
+
+ @Test
+ void shouldResumeOnASurvivorAfterTheSignedInNodeDies() throws Exception {
+ InetAddress loopback = InetAddress.getLoopbackAddress();
+ try (ServerSocket primarySocket = new ServerSocket(0, 4, loopback);
+ ServerSocket survivorSocket = new ServerSocket(0, 4,
loopback)) {
+ int primaryPort = primarySocket.getLocalPort();
+ int survivorPort = survivorSocket.getLocalPort();
+ AtomicInteger survivorRegistrations = new AtomicInteger();
+ AtomicInteger survivorPings = new AtomicInteger();
+
+ // The primary leads, so the sign-in settles there and the roster
is
+ // only remembered -- not acted on -- until the node dies.
+ MockNode primary = MockNode.serve(primarySocket, request -> {
+ if (request.is(GET_CLUSTER_METADATA_CODE,
OPERATION_NON_REPLICATED)) {
+ return Response.success(
+ OPERATION_NON_REPLICATED,
clusterMetadata(primaryPort, survivorPort, primaryPort));
+ }
+ if (request.operation() == OPERATION_REGISTER) {
+ return Response.success(OPERATION_REGISTER,
registerBody(1));
+ }
+ return Response.success(OPERATION_NON_REPLICATED,
Unpooled.EMPTY_BUFFER);
+ });
+ MockNode survivor = MockNode.serve(survivorSocket, request -> {
+ if (request.is(GET_CLUSTER_METADATA_CODE,
OPERATION_NON_REPLICATED)) {
+ return Response.success(
+ OPERATION_NON_REPLICATED,
clusterMetadata(primaryPort, survivorPort, survivorPort));
+ }
+ if (request.operation() == OPERATION_REGISTER) {
+ survivorRegistrations.incrementAndGet();
+ return Response.success(OPERATION_REGISTER,
registerBody(2));
+ }
+ if (request.is(PING_CODE, OPERATION_NON_REPLICATED)) {
+ survivorPings.incrementAndGet();
+ }
+ return Response.success(OPERATION_NON_REPLICATED,
Unpooled.EMPTY_BUFFER);
+ });
+
+ AsyncIggyTcpClient client = AsyncIggyTcpClient.builder()
+ .host(loopback.getHostAddress())
+ .port(primaryPort)
+ .credentials("iggy", "iggy")
Review Comment:
with builder credentials `replayLogin` falls back to them, so this passes
even with `rememberedLogin = null` (tried it, still green). sign in through
`users().login()` on a credential-less client, and add a logout-then-kill case
asserting no registration reached the survivor.
##########
foreign/csharp/Iggy_SDK_Tests/VsrTests/EndpointFailoverTests.cs:
##########
@@ -0,0 +1,428 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+using System.Buffers.Binary;
+using System.Net;
+using System.Net.Sockets;
+using System.Text;
+using Apache.Iggy.Configuration;
+using Apache.Iggy.Contracts.Tcp;
+using Apache.Iggy.Enums;
+using Apache.Iggy.IggyClient.Implementations;
+using Microsoft.Extensions.Logging.Abstractions;
+
+namespace Apache.Iggy.Tests.VsrTests;
+
+/// <summary>
+/// The node a client signed in on dies; its next request has to complete
on a survivor the roster named,
+/// under a session established there. Mirrors
+/// <c>core/integration/tests/cluster/failover_client_continuity.rs</c>.
+/// </summary>
+public sealed class EndpointFailoverTests
+{
+ private const int HeaderSize = 256;
+ private const int SizeOffset = 48;
+ private const int CommandOffset = 60;
+ private const int RequestIdOffset = 168;
+ private const int RequestOperationOffset = 176;
+ private const int RequestReservedOffset = 196;
+ private const int ReplyRequestIdOffset = 200;
+ private const int ReplyOperationOffset = 208;
+ private const int ReplyStatusOffset = 216;
+
+ private const byte CommandReply = 8;
+ private const byte CommandEviction = 13;
+ private const int EvictionReasonOffset = 255;
+ private const byte EvictionStaleClient = 13;
+ private const byte OperationRegister = 1;
+ private const byte OperationNonReplicated = 2;
+ private const int GetClusterMetadataCode = 12;
+ private const int PingCode = 1;
+
+ [Fact]
+ public async Task ResumesOnASurvivorAfterTheSignedInNodeDies()
+ {
+ using var primary = new MockNode();
+ using var survivor = new MockNode();
+
+ // The primary leads, so the sign-in settles there and the roster is
only remembered - not acted on -
+ // until the node dies.
+ primary.Serve(request => request.Code == GetClusterMetadataCode
+ ? Reply(OperationNonReplicated, ClusterMetadata(primary.Port,
survivor.Port, primary.Port))
+ : Answer(request));
+ survivor.Serve(request => request.Code == GetClusterMetadataCode
+ ? Reply(OperationNonReplicated, ClusterMetadata(primary.Port,
survivor.Port, survivor.Port))
+ : Answer(request));
+
+ var configuration = new IggyClientConfigurator
+ {
+ BaseAddress = $"127.0.0.1:{primary.Port}",
+ Protocol = Protocol.Tcp,
+ ReconnectionSettings = new ReconnectionSettings
+ {
+ Enabled = true,
+ MaxRetries = 4,
+ InitialDelay = TimeSpan.FromMilliseconds(20)
+ }
+ };
+ using var client = new TcpMessageStream(configuration,
NullLoggerFactory.Instance);
+
+ await client.ConnectAsync(TestContext.Current.CancellationToken);
+ // No auto login: the credentials come from the caller's own sign-in,
which is the shape that could not
+ // reconnect at all before.
+ await client.LoginUserAsync("iggy", "iggy",
TestContext.Current.CancellationToken);
+ await client.PingAsync(TestContext.Current.CancellationToken);
+ Assert.Equal(1, primary.Pings);
+
+ primary.Kill();
+
+ // The request in flight when the node died is allowed to fail; what
is not allowed is never completing
+ // one, which is what a client that only knows the dead endpoint does.
+ var (resumed, lastError) = await ResumedWithin(client,
TimeSpan.FromSeconds(10));
+ Assert.True(resumed,
+ $"the client has to resume on the survivor the roster named
({lastError}, survivor saw " +
+ $"{survivor.Registrations} registrations and {survivor.Pings}
pings)");
+ Assert.True(survivor.Registrations >= 1, "the remembered credentials
signed in again on the survivor");
+ Assert.True(survivor.Pings >= 1, "the request landed on the survivor");
+ }
+
+ /// <summary>
+ /// Mirrors the integration contract (HeartbeatTests
+ /// EvictedClient_WithoutAutoLogin_Should_FailFast_And_NotReconnect):
a server-side eviction ends the
+ /// session authoritatively, so the credentials a manual sign-in
remembered must not resurrect it - the
+ /// evicted request surfaces the loss with no reconnect attempt.
+ /// </summary>
+ [Fact]
+ public async Task ServerEvictionForgetsTheRememberedSignIn()
+ {
+ using var node = new MockNode();
+ var evict = false;
+ node.Serve(request =>
+ {
+ if (request.Operation == OperationRegister)
+ {
+ return Reply(OperationRegister, RegisterBody(session: 128));
+ }
+
+ return evict
+ ? EvictionFrame(EvictionStaleClient)
+ : Reply(OperationNonReplicated, request.Code ==
GetClusterMetadataCode
+ ? ClusterMetadata(node.Port, node.Port, node.Port)
+ : []);
+ });
+
+ var configuration = new IggyClientConfigurator
+ {
+ BaseAddress = $"127.0.0.1:{node.Port}",
+ Protocol = Protocol.Tcp,
+ ReconnectionSettings = new ReconnectionSettings
+ {
+ Enabled = true,
+ MaxRetries = 2,
+ InitialDelay = TimeSpan.FromMilliseconds(20)
+ }
+ };
+ using var client = new TcpMessageStream(configuration,
NullLoggerFactory.Instance);
+
+ await client.ConnectAsync(TestContext.Current.CancellationToken);
+ await client.LoginUserAsync("iggy", "iggy",
TestContext.Current.CancellationToken);
+ await client.PingAsync(TestContext.Current.CancellationToken);
+ var connectionsBeforeEviction = node.Connections;
+
+ evict = true;
+ await Assert.ThrowsAnyAsync<Exception>(() =>
client.PingAsync(TestContext.Current.CancellationToken));
Review Comment:
`ThrowsAnyAsync<Exception>` accepts anything. assert
`IggyInvalidStatusCodeException` here and `NotConnectedException` at 152.
##########
foreign/node/src/client/client.connection.ts:
##########
@@ -312,24 +319,28 @@ export class IggyConnection extends EventEmitter {
if (this.connected || this.socket !== expectedSocket)
return this.connect();
- const options = this._reconnectTarget(attempt);
- attempt += 1;
- const socket = this._installSocket(
- getTransport({ ...this.config, options })
- );
- this.socket = socket;
- expectedSocket = socket;
- try {
- await this._waitForConnection(socket);
- if (this.socket !== socket)
- return this.connect();
- this.config.options = options;
- return this;
- } catch (error) {
- lastError = error instanceof Error
- ? error
- : new Error(String(error));
- debug('reconnect attempt failed', lastError);
+ // Every endpoint gets its turn inside one attempt, so a full pass over
+ // the cluster costs one retry rather than one per endpoint: a pass that
+ // stopped at the first refusal would never reach the survivors of a
+ // client configured for a single retry.
+ for (const options of this._redialCandidates()) {
+ const socket = this._installSocket(
+ getTransport({ ...this.config, options })
+ );
+ this.socket = socket;
+ expectedSocket = socket;
+ try {
+ await this._waitForConnection(socket);
Review Comment:
no dial bound: this waits for the os connect timeout, there is no
`'timeout'` listener, and the dead node is candidate 0 on every pass
(`config.options` only changes on success). with a black-holed roster node
ahead of a live survivor the survivor is never dialed while the first connect
hangs. race it against a 2s timer when there is more than one candidate and
`socket.destroy()` on expiry (rust bounds this at 2s).
##########
foreign/node/src/client/client.connection.ts:
##########
@@ -341,16 +352,43 @@ export class IggyConnection extends EventEmitter {
}
/**
- * Alternates reconnect dials between the current endpoint and the
- * configured seed. After a leader redirect the current endpoint may die
- * with the leader, and the seed is the way back to the rest of the cluster.
+ * Records the cluster roster as redial candidates.
+ *
+ * Replaced wholesale rather than merged: the roster is the cluster's own
+ * answer about where its nodes are, so a node it dropped stops being
+ * dialed. The configured seed is kept separately and outlives it.
+ */
+ rememberRoster(endpoints: { host: string, port: number }[]): void {
+ if (endpoints.length === 0)
+ return;
+ this.rosterEndpoints = endpoints;
+ }
+
+ /**
+ * Endpoints a redial rotates through, likeliest first: where the client
+ * currently is, the endpoint it was configured with, then the roster it
+ * learned while connected. After a leader redirect the current endpoint may
+ * die with the leader, and the rest of the list is the way back to the
+ * cluster. Duplicates are dropped, so an endpoint the roster merely spells
Review Comment:
`normalizeHost` only knows localhost / `::1` / `::ffff:`, so a seed given as
a dns name is not deduped against the roster's ip for the same node - the dead
node gets dialed twice per pass. resolve, or soften this sentence.
##########
foreign/node/src/client/client.connection.ts:
##########
@@ -312,24 +319,28 @@ export class IggyConnection extends EventEmitter {
if (this.connected || this.socket !== expectedSocket)
return this.connect();
- const options = this._reconnectTarget(attempt);
- attempt += 1;
- const socket = this._installSocket(
- getTransport({ ...this.config, options })
- );
- this.socket = socket;
- expectedSocket = socket;
- try {
- await this._waitForConnection(socket);
- if (this.socket !== socket)
- return this.connect();
- this.config.options = options;
- return this;
- } catch (error) {
- lastError = error instanceof Error
- ? error
- : new Error(String(error));
- debug('reconnect attempt failed', lastError);
+ // Every endpoint gets its turn inside one attempt, so a full pass over
+ // the cluster costs one retry rather than one per endpoint: a pass that
+ // stopped at the first refusal would never reach the survivors of a
+ // client configured for a single retry.
+ for (const options of this._redialCandidates()) {
Review Comment:
`_destroy()` mid-pass no longer stops the pass: the `ending` and
supersession checks sit above this loop, so after the destroyed candidate
rejects the loop dials the rest, and one that answers leaves `ending = true`,
`connected = true`, an open socket nobody closes and a `'connect'` event after
destroy - the leaked socket keeps the process alive. reproduced. re-check
`this.ending` / `this.socket !== expectedSocket` at the top of each iteration,
and destroy + throw after `_waitForConnection` if `this.ending`.
##########
foreign/node/src/client/client.connection.ts:
##########
@@ -341,16 +352,43 @@ export class IggyConnection extends EventEmitter {
}
/**
- * Alternates reconnect dials between the current endpoint and the
- * configured seed. After a leader redirect the current endpoint may die
- * with the leader, and the seed is the way back to the rest of the cluster.
+ * Records the cluster roster as redial candidates.
+ *
+ * Replaced wholesale rather than merged: the roster is the cluster's own
+ * answer about where its nodes are, so a node it dropped stops being
+ * dialed. The configured seed is kept separately and outlives it.
+ */
+ rememberRoster(endpoints: { host: string, port: number }[]): void {
+ if (endpoints.length === 0)
+ return;
+ this.rosterEndpoints = endpoints;
+ }
+
+ /**
+ * Endpoints a redial rotates through, likeliest first: where the client
+ * currently is, the endpoint it was configured with, then the roster it
+ * learned while connected. After a leader redirect the current endpoint may
+ * die with the leader, and the rest of the list is the way back to the
+ * cluster. Duplicates are dropped, so an endpoint the roster merely spells
+ * differently does not earn a second attempt.
*/
- private _reconnectTarget(attempt: number): ClientConfig['options'] {
- const current = this.config.options;
- if (this.seedOptions.host === current.host &&
- this.seedOptions.port === current.port)
- return current;
- return attempt % 2 === 0 ? current : this.seedOptions;
+ _redialCandidates(): ClientConfig['options'][] {
+ const candidates = [this.config.options];
+ const known = [
+ this.seedOptions,
+ ...this.rosterEndpoints.map(
+ ({ host, port }) => ({ ...this.config.options, host, port })
+ )
+ ];
+ for (const candidate of known) {
+ const duplicate = candidates.some(
+ (known) => known.port === candidate.port &&
Review Comment:
`known` shadows the `known` array above; rename to `existing`. also `{ host:
string, port: number }` is now spelled inline three times - export one
`Endpoint` type.
##########
foreign/node/src/client/client.connection.ts:
##########
@@ -300,7 +308,6 @@ export class IggyConnection extends EventEmitter {
): Promise<this> {
let lastError = initialError;
let expectedSocket = this.socket;
- let attempt = 0;
while (enabled && this.reconnectCount < maxRetries) {
this.connecting = true;
this.reconnectCount += 1;
Review Comment:
the sleep on the next line runs before the first pass even with a roster
known, so failover to a live survivor takes at least `interval` (5s by
default). skip the first sleep when there is more than one candidate -
single-node backoff stays and the 'does not reconnect after destruction' test
keeps working.
##########
foreign/node/src/client/client.socket.ts:
##########
@@ -478,6 +478,13 @@ export class CommandResponseStream extends EventEmitter {
{ last: false }
);
const metadata = GET_CLUSTER_METADATA.deserialize(response);
+ // Every read feeds the redial candidates, leaderless ones included: a
+ // roster with no leader still names where the nodes are.
+ this.connection.rememberRoster(
Review Comment:
this is the only feed and it runs only inside `_settleOnLeader` after a
login, so the roster goes stale between logins. and `_processVsr` replays 57/58
on the same connection for 30s with no leader recheck, so a leader move with
the socket still up never fails over. feed it from every `GetClusterMetadata`
reply and recheck the leader on 58 like the rust client does (57 stays
same-session, its outcome is unknown).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
