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/10/21 22:22:49 UTC
(iggy) branch feat/spread-placement updated (e019f7d59 -> 949a55e02)
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a change to branch feat/spread-placement
in repository https://gitbox.apache.org/repos/asf/iggy.git
from e019f7d59 fix(server): preserve read fences and forwarding refusals
new 58b58dedd fix(csharp): recover sent offsets during disposal
new eac054bdd test(csharp): await counters and assert final checkpoints
new 1d873975e fix(go): preserve dial retries and fence closed logins
new 4a4bc762c fix(java): fence read replies and release session waiters
new 5294276bf refactor(java): separate session reply bookkeeping
new 35d332a19 fix(sdk): retain replay identity through session resets
new 6e71a6941 fix(go): close detached sockets outside the session lock
new 949a55e02 fix(swift): refresh routes beyond refused primary views
The 8 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:
core/sdk/src/binary/mod.rs | 5 +-
core/sdk/src/session_owner/lanes.rs | 20 +++++++-
.../Fixtures/IggyServerFixture.cs | 16 ++++++
.../IggyConsumerTests.cs | 9 ++--
.../IggyTypedConsumerTests.cs | 3 +-
.../Iggy_SDK.Tests.Integration/StreamsTests.cs | 4 +-
.../Iggy_SDK.Tests.Integration/TopicsTests.cs | 5 +-
foreign/csharp/Iggy_SDK/Consumers/IggyConsumer.cs | 3 ++
.../PartitionIncarnationTests.Disposal.cs | 22 ++++++--
foreign/go/client/tcp/tcp_connect_test.go | 39 ++++++++++++++
foreign/go/client/tcp/tcp_core.go | 15 +++---
foreign/go/client/tcp/tcp_replicas.go | 20 ++++----
foreign/go/client/tcp/tcp_replicas_test.go | 46 +++++++++++++++++
foreign/go/client/tcp/tcp_session_management.go | 33 ++++++++++--
.../iggy/client/async/tcp/AsyncTcpConnection.java | 8 +--
.../client/async/tcp/vsr/ConsensusSession.java | 19 ++++---
.../client/async/tcp/vsr/VsrResponseHandler.java | 33 ++++++++----
.../tcp/AsyncTcpConnectionConcurrencyTest.java | 10 +++-
.../client/async/tcp/vsr/ConsensusSessionTest.java | 26 ++++++++++
.../async/tcp/vsr/VsrResponseHandlerTest.java | 11 ++++
foreign/swift/Sources/Iggy/Client/PollRouter.swift | 59 +++++++++++++++++++---
.../swift/Sources/Iggy/Protocol/Responses.swift | 8 +--
foreign/swift/Tests/IggyTests/MessagingTests.swift | 59 +++++++++++++++++++++-
.../Tests/IggyTests/Support/FakeVSRServer.swift | 4 +-
24 files changed, 404 insertions(+), 73 deletions(-)
(iggy) 08/08: fix(swift): refresh routes beyond refused primary views
Posted by hg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch feat/spread-placement
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 949a55e023b6878eae92f5c7805af8e3cf1b7127
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Thu Oct 22 00:21:38 2026 +0200
fix(swift): refresh routes beyond refused primary views
A metadata replica can still advertise the primary that refused a
request. Ask that primary for its newer view and fence every cached
alias against older routes while retaining uncertain write frames.
---
foreign/swift/Sources/Iggy/Client/PollRouter.swift | 59 +++++++++++++++++++---
.../swift/Sources/Iggy/Protocol/Responses.swift | 8 +--
foreign/swift/Tests/IggyTests/MessagingTests.swift | 59 +++++++++++++++++++++-
.../Tests/IggyTests/Support/FakeVSRServer.swift | 4 +-
4 files changed, 114 insertions(+), 16 deletions(-)
diff --git a/foreign/swift/Sources/Iggy/Client/PollRouter.swift b/foreign/swift/Sources/Iggy/Client/PollRouter.swift
index 70fad6d9a..14002d00a 100644
--- a/foreign/swift/Sources/Iggy/Client/PollRouter.swift
+++ b/foreign/swift/Sources/Iggy/Client/PollRouter.swift
@@ -62,6 +62,9 @@ final class PollRouter: Sendable {
/// watermark a data connection binds at.
let consumerSession: Responses.SessionIdentity
let context: PartitionContext
+ let view: UInt32
+ var minimumView: UInt32 = 0
+ var askFirst: UInt8?
}
private struct DataConnection {
@@ -175,7 +178,6 @@ final class PollRouter: Sendable {
guard error.code == .transientNotAccepted || reconnects else {
throw replay.outcome.observe(error)
}
- removeRoute(key)
refusal = replay.outcome.observe(error)
if !replay.outcome.isUncertain {
replay.frame = nil
@@ -367,6 +369,10 @@ final class PollRouter: Sendable {
removeRoute(request.key)
throw unsent.refusal
} catch let error as IggyError {
+ if error.code == .notPrimary {
+ markNotPrimary(route, partitionID: request.key.partitionID)
+ throw ClientCore.graded(error)
+ }
removeRoute(request.key)
let source = MissSource.replica(identity: route.consumerSession, lineage: request.lineage, fence: route.fence)
if error.isNamespaceMiss, await !isMetadataPrimary(route.primaryReplica, answeredOn: source, coordinator: coordinator, deadline: request.deadline) {
@@ -455,6 +461,7 @@ final class PollRouter: Sendable {
data.consumerSession = nil
}
slot.connection.withLockedValue { $0 = data }
+ if error.code == .notPrimary { throw error }
if error.isConnectionLoss, error.code != .staleClient {
if replay.frame?.operation.isPartition == true, error.code != .notConnected, error.code != .cannotEstablishConnection {
replay.outcome.markUncertain()
@@ -574,13 +581,13 @@ final class PollRouter: Sendable {
}
answer = refusal
}
+ if answer.code == .notPrimary, !VSRFrame.hasCommittedResult(header) {
+ throw answer
+ }
guard request.operation.isPartition, replay.outcome.isUncertain || answer.code == .transientNotCommitted else {
throw ClientCore.graded(answer)
}
let error = replay.outcome.observe(ClientCore.graded(answer), committed: VSRFrame.hasCommittedResult(header))
- if answer.code == .notPrimary, !VSRFrame.hasCommittedResult(header) {
- throw answer
- }
// Once the outcome is unknown, any answer but a transient one ends
// the replay, as the same request id gets the same answer again.
// A replay sent with a sliver of the budget would time out, and
@@ -604,18 +611,24 @@ final class PollRouter: Sendable {
) async throws -> Route {
let watermark = await coordinator.metadataWatermark
let (fence, cached) = state.withLockedValue { ($0.fence, $0.routes[key]) }
- if let cached, cached.fence == fence, cached.context.metadataOp >= watermark {
+ if let cached, cached.fence == fence, cached.askFirst == nil, cached.view >= cached.minimumView, cached.context.metadataOp >= watermark {
return cached
}
let response = try await lookUp(
- key, clustered: clustered, fence: fence, watermark: watermark, lineage: lineage, coordinator: coordinator, deadline: deadline)
+ key, clustered: clustered, fence: fence, watermark: watermark, lineage: lineage, coordinator: coordinator, deadline: deadline, preferred: cached)
guard let session = await coordinator.sessionIdentity(at: response.context.metadataOp) else {
throw IggyError(.transientNotAccepted, context: "no session is bound to route partition \(key.partitionID)")
}
- let route = Route(fence: fence, primaryReplica: response.primaryReplica, consumerSession: session, context: response.context)
+ var route = Route(fence: fence, primaryReplica: response.primaryReplica, consumerSession: session, context: response.context, view: response.view)
try await validate(route, coordinator: coordinator)
try state.withLockedValue { state in
try Self.checkFence(route.fence, in: state)
+ for (alias, known) in state.routes where alias.partitionID == key.partitionID && known.context.incarnation == route.context.incarnation {
+ route.minimumView = max(route.minimumView, known.minimumView)
+ }
+ guard route.view >= route.minimumView else {
+ throw IggyError(.transientNotAccepted, context: "the partition route predates a primary refusal")
+ }
if state.routes.count >= Self.maxCachedRoutes {
state.routes.removeAll()
}
@@ -628,10 +641,30 @@ final class PollRouter: Sendable {
/// the coordinator on a single node.
private func lookUp(
_ key: RouteKey, clustered: Bool, fence: UInt128Value, watermark: UInt64, lineage: UInt64, coordinator: ClientCore,
- deadline: ContinuousClock.Instant
+ deadline: ContinuousClock.Instant, preferred: Route? = nil
) async throws -> Responses.PartitionRoute {
let payload = try Requests.partitionRoute(streamID: key.stream, topicID: key.topic, partitionID: key.partitionID)
let context = ClientCore.readContext(watermark)
+ if clustered, let preferred, let replica = preferred.askFirst,
+ let identity = await coordinator.sessionIdentity(at: watermark)
+ {
+ do {
+ let body = try await exchange(
+ onReplica: replica, code: CommandCode.getPartitionRoute.rawValue, payload: payload, context: context,
+ identity: identity, lineage: lineage, fence: fence, coordinator: coordinator, deadline: deadline,
+ dialDeadline: min(deadline, .now + ClientCore.failoverDialTimeout))
+ let response = try Responses.partitionRoute(body[...])
+ if response.context.incarnation != preferred.context.incarnation || response.view >= preferred.minimumView {
+ return response
+ }
+ } catch let unsent as UnsentRequest {
+ throw unsent.refusal
+ } catch let error as IggyError
+ where error.code == .transientNotAccepted || error.code == .transientNotCommitted || error.isNamespaceMiss || error.isConnectionLoss
+ {
+ // A failed hint leaves the metadata primary as the fallback.
+ }
+ }
let body =
clustered
? try await lookUpInCluster(payload, context: context, fence: fence, lineage: lineage, coordinator: coordinator, deadline: deadline)
@@ -804,4 +837,14 @@ final class PollRouter: Sendable {
private func removeRoute(_ key: RouteKey) {
_ = state.withLockedValue { $0.routes.removeValue(forKey: key) }
}
+
+ private func markNotPrimary(_ route: Route, partitionID: UInt32) {
+ state.withLockedValue { state in
+ for key in state.routes.keys where key.partitionID == partitionID && state.routes[key]?.context.incarnation == route.context.incarnation {
+ let floor = max(state.routes[key]?.minimumView ?? 0, route.view == .max ? .max : route.view + 1)
+ state.routes[key]?.minimumView = floor
+ state.routes[key]?.askFirst = route.primaryReplica
+ }
+ }
+ }
}
diff --git a/foreign/swift/Sources/Iggy/Protocol/Responses.swift b/foreign/swift/Sources/Iggy/Protocol/Responses.swift
index 91d4d88df..83772cc6c 100644
--- a/foreign/swift/Sources/Iggy/Protocol/Responses.swift
+++ b/foreign/swift/Sources/Iggy/Protocol/Responses.swift
@@ -434,6 +434,7 @@ enum Responses {
struct PartitionRoute: Equatable {
/// The context a request routed by the reply carries.
let context: PartitionContext
+ let view: UInt32
/// The configured replica id of the partition primary, as
/// `GetClusterMetadata` reports it, never a roster position.
let primaryReplica: UInt8
@@ -449,10 +450,9 @@ enum Responses {
}
return try decode(body) { reader in
let context = PartitionContext(incarnation: try reader.readUInt64(), ownerGeneration: 0, metadataOp: try reader.readUInt64())
- // The view and the partition count, which a client that asks only
- // the coordinator for routes does not use.
- try reader.skip(2 * MemoryLayout<UInt32>.size)
- return PartitionRoute(context: context, primaryReplica: try reader.readUInt8())
+ let view = try reader.readUInt32()
+ try reader.skip(MemoryLayout<UInt32>.size)
+ return PartitionRoute(context: context, view: view, primaryReplica: try reader.readUInt8())
}
}
diff --git a/foreign/swift/Tests/IggyTests/MessagingTests.swift b/foreign/swift/Tests/IggyTests/MessagingTests.swift
index 0569156b8..26e999c04 100644
--- a/foreign/swift/Tests/IggyTests/MessagingTests.swift
+++ b/foreign/swift/Tests/IggyTests/MessagingTests.swift
@@ -958,6 +958,9 @@ struct MessagingTests {
}
let refused = Locked(false)
await primary.respond { request in
+ if request.code == CommandCode.getPartitionRoute.rawValue {
+ return .ok(FakeVSRServer.partitionRouteBody(FakeVSRServer.routeContext, primaryReplica: 1, view: 3))
+ }
if request.code == CommandCode.pollMessages.rawValue, !refused.exchange(true) {
return .status(IggyErrorCode.notPrimary.rawValue)
}
@@ -965,7 +968,8 @@ struct MessagingTests {
}
let client = try await connectedClient(coordinator)
_ = try await client.pollMessages(streamID: 1, topicID: 2, partitionID: 0, strategy: .first, count: 1, autoCommit: false)
- #expect(await requests(coordinator, .getPartitionRoute).count == 2)
+ #expect(await requests(coordinator, .getPartitionRoute).count == 1)
+ #expect(await requests(primary, .getPartitionRoute).count == 1)
#expect(await requests(primary, .pollMessages).count == 2)
// The refusal left the data connection at a frame boundary.
#expect(await primary.connectionCount == 1)
@@ -1165,6 +1169,56 @@ struct MessagingTests {
// MARK: Namespace misses
+ @Test func aPrimaryRefusalRefreshesEveryAliasPastTheRejectedView() async throws {
+ let coordinator = try await FakeVSRServer.started()
+ let previous = try await FakeVSRServer.started()
+ let current = try await FakeVSRServer.started()
+ defer {
+ Task {
+ await coordinator.stop()
+ await previous.stop()
+ await current.stop()
+ }
+ }
+ let moved = Locked(false)
+ let hints = Locked(0)
+ await coordinator.respond { request in
+ switch request.code.flatMap(CommandCode.init(rawValue:)) {
+ case .getClusterMetadata:
+ return .ok(FakeVSRServer.rosterBody([coordinator.address, previous.address, current.address]))
+ case .getPartitionRoute:
+ return .ok(FakeVSRServer.partitionRouteBody(FakeVSRServer.routeContext, primaryReplica: 1))
+ default:
+ return FakeVSRServer.defaultReply(request, address: coordinator.address)
+ }
+ }
+ await previous.respond { request in
+ if request.code == CommandCode.pollMessages.rawValue, moved.withLock({ $0 }) {
+ return .status(IggyErrorCode.notPrimary.rawValue)
+ }
+ if request.code == CommandCode.getPartitionRoute.rawValue {
+ let ready = hints.withLock {
+ $0 += 1; return $0 > 1
+ }
+ return .ok(FakeVSRServer.partitionRouteBody(FakeVSRServer.routeContext, primaryReplica: ready ? 2 : 1, view: ready ? 2 : 1))
+ }
+ return FakeVSRServer.defaultReply(request, address: previous.address)
+ }
+ let client = try await connectedClient(coordinator)
+ let aliases: [Identifier] = [1, "same-stream"]
+ for stream in aliases {
+ _ = try await client.pollMessages(streamID: stream, topicID: 2, partitionID: 0, strategy: .first, count: 1, autoCommit: false)
+ }
+ moved.withLock { $0 = true }
+ for stream in aliases {
+ _ = try await client.pollMessages(streamID: stream, topicID: 2, partitionID: 0, strategy: .first, count: 1, autoCommit: false)
+ }
+ #expect(await requests(previous, .pollMessages).count == 3)
+ #expect(await requests(current, .pollMessages).count == 2)
+ #expect(hints.withLock { $0 } == 3)
+ await client.shutdown()
+ }
+
/// A backup primary that has not applied the topic yet answers "not
/// found". The route is looked up again on the metadata primary and the
/// poll goes again.
@@ -1827,7 +1881,8 @@ struct MessagingTests {
case .getClusterMetadata:
return .ok(FakeVSRServer.rosterBody([coordinator.address, first.address, second.address]))
case .getPartitionRoute:
- return .ok(FakeVSRServer.partitionRouteBody(FakeVSRServer.routeContext, primaryReplica: moved.withLock { $0 } ? 2 : 1))
+ let hasMoved = moved.withLock { $0 }
+ return .ok(FakeVSRServer.partitionRouteBody(FakeVSRServer.routeContext, primaryReplica: hasMoved ? 2 : 1, view: hasMoved ? 2 : 1))
default:
return FakeVSRServer.defaultReply(request, address: coordinator.address)
}
diff --git a/foreign/swift/Tests/IggyTests/Support/FakeVSRServer.swift b/foreign/swift/Tests/IggyTests/Support/FakeVSRServer.swift
index 1d5ad081e..ae8018de8 100644
--- a/foreign/swift/Tests/IggyTests/Support/FakeVSRServer.swift
+++ b/foreign/swift/Tests/IggyTests/Support/FakeVSRServer.swift
@@ -174,11 +174,11 @@ final class FakeVSRServer: @unchecked Sendable {
}
/// The reply to `GetPartitionRoute`.
- static func partitionRouteBody(_ context: PartitionContext, partitionsCount: UInt32 = 3, primaryReplica: UInt8 = 0) -> [UInt8] {
+ static func partitionRouteBody(_ context: PartitionContext, partitionsCount: UInt32 = 3, primaryReplica: UInt8 = 0, view: UInt32 = 1) -> [UInt8] {
var writer = ByteWriter()
writer.write(context.incarnation)
writer.write(context.metadataOp)
- writer.write(UInt32(1))
+ writer.write(view)
writer.write(partitionsCount)
writer.write(primaryReplica)
return writer.bytes
(iggy) 01/08: fix(csharp): recover sent offsets during disposal
Posted by hg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch feat/spread-placement
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 58b58dedd17bc04b84ac30567ae2bc759877ed27
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Thu Oct 22 00:00:15 2026 +0200
fix(csharp): recover sent offsets during disposal
A cancelled after-receive store can retain an uncertain write. Let the
shutdown checkpoint resolve that write before storing progress and
leaving the group.
---
foreign/csharp/Iggy_SDK/Consumers/IggyConsumer.cs | 3 +++
.../PartitionIncarnationTests.Disposal.cs | 22 ++++++++++++++++++----
2 files changed, 21 insertions(+), 4 deletions(-)
diff --git a/foreign/csharp/Iggy_SDK/Consumers/IggyConsumer.cs b/foreign/csharp/Iggy_SDK/Consumers/IggyConsumer.cs
index 9c98f2b38..7d0a465c7 100644
--- a/foreign/csharp/Iggy_SDK/Consumers/IggyConsumer.cs
+++ b/foreign/csharp/Iggy_SDK/Consumers/IggyConsumer.cs
@@ -424,6 +424,9 @@ public partial class IggyConsumer : IAsyncDisposable
catch (OperationCanceledException) when (afterReceive && IsDisposed && !ct.IsCancellationRequested)
{
}
+ catch (VsrRequestOutcomeUnknownException) when (afterReceive && IsDisposed && !ct.IsCancellationRequested)
+ {
+ }
}
/// <summary>
diff --git a/foreign/csharp/Iggy_SDK_Tests/ConsumerTests/PartitionIncarnationTests.Disposal.cs b/foreign/csharp/Iggy_SDK_Tests/ConsumerTests/PartitionIncarnationTests.Disposal.cs
index d6a6e2da9..54bff3e4a 100644
--- a/foreign/csharp/Iggy_SDK_Tests/ConsumerTests/PartitionIncarnationTests.Disposal.cs
+++ b/foreign/csharp/Iggy_SDK_Tests/ConsumerTests/PartitionIncarnationTests.Disposal.cs
@@ -18,6 +18,7 @@
using System.Collections.Concurrent;
using Apache.Iggy.Consumers;
using Apache.Iggy.Contracts;
+using Apache.Iggy.Exceptions;
using Apache.Iggy.IggyClient;
using Apache.Iggy.Kinds;
using Microsoft.Extensions.Logging.Abstractions;
@@ -164,8 +165,10 @@ public sealed partial class PartitionIncarnationTests
Assert.Equal(["store", "leave"], events);
}
- [Fact]
- public async Task given_an_after_receive_store_in_flight_when_disposed_should_drain_it_before_checkpointing_and_leaving()
+ [Theory]
+ [InlineData(false)]
+ [InlineData(true)]
+ public async Task given_an_after_receive_store_in_flight_when_disposed_should_drain_it_before_checkpointing_and_leaving(bool sent)
{
var events = new ConcurrentQueue<string>();
var client = ClientMock(new());
@@ -177,18 +180,29 @@ public sealed partial class PartitionIncarnationTests
var cancelled = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
var attempts = 0;
+ client.Setup(mock => mock.ResolveOperationAsync(1, It.IsAny<TimeSpan>(), It.IsAny<CancellationToken>()))
+ .Callback(() => events.Enqueue("resolve")).Returns(Task.CompletedTask);
client.Setup(mock => mock.StoreOffsetAsync(It.IsAny<Consumer>(), It.IsAny<Identifier>(), It.IsAny<Identifier>(),
It.IsAny<ulong>(), It.IsAny<uint>(), It.IsAny<PartitionContext>(), It.IsAny<CancellationToken>()))
.Returns(async (Consumer _, Identifier _, Identifier _, ulong _, uint _, PartitionContext _, CancellationToken token) =>
{
- if (Interlocked.Increment(ref attempts) == 1)
+ var attempt = Interlocked.Increment(ref attempts);
+ if (attempt == 1)
{
storing.TrySetResult();
using var registration = token.Register(() => cancelled.TrySetResult());
await release.Task;
events.Enqueue("drained");
+ if (sent)
+ {
+ throw new VsrRequestOutcomeUnknownException(new OperationCanceledException(token), 1);
+ }
token.ThrowIfCancellationRequested();
}
+ else if (sent && attempt == 2)
+ {
+ throw new VsrLaneBlockedException(1);
+ }
else
{
events.Enqueue("checkpoint");
@@ -217,7 +231,7 @@ public sealed partial class PartitionIncarnationTests
}
await dispose.WaitAsync(timeout.Token);
Assert.False(await receive);
- Assert.Equal(["drained", "checkpoint", "leave"], events);
+ Assert.Equal(sent ? ["drained", "resolve", "checkpoint", "leave"] : ["drained", "checkpoint", "leave"], events);
}
private static void SetupDisposalGroup(Mock<IIggyClient> client, ConcurrentQueue<string> events)
(iggy) 07/08: fix(go): close detached sockets outside the session lock
Posted by hg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch feat/spread-placement
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 6e71a6941cfbdc5ddb7dc49a9394bd230547828a
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Thu Oct 22 00:21:28 2026 +0200
fix(go): close detached sockets outside the session lock
TLS close_notify can wait for an unresponsive peer. Detach ownership
under the lock, then close sockets and drain tasks after releasing it.
---
foreign/go/client/tcp/tcp_core.go | 2 +-
foreign/go/client/tcp/tcp_replicas.go | 20 +++++------
foreign/go/client/tcp/tcp_replicas_test.go | 46 +++++++++++++++++++++++++
foreign/go/client/tcp/tcp_session_management.go | 22 +++++++++---
4 files changed, 74 insertions(+), 16 deletions(-)
diff --git a/foreign/go/client/tcp/tcp_core.go b/foreign/go/client/tcp/tcp_core.go
index ac38cf2e0..5a73cc937 100644
--- a/foreign/go/client/tcp/tcp_core.go
+++ b/foreign/go/client/tcp/tcp_core.go
@@ -522,7 +522,7 @@ func (c *IggyTcpClient) Disconnect() error {
c.loggedOut = c.autoLoginBlocked
stopped := c.links.detach()
c.mtx.Unlock()
- stopped()
+ stopped(true)
c.connectMu.Lock()
attempt := c.connectionAttempt
c.connectMu.Unlock()
diff --git a/foreign/go/client/tcp/tcp_replicas.go b/foreign/go/client/tcp/tcp_replicas.go
index aec789eb0..3da47d471 100644
--- a/foreign/go/client/tcp/tcp_replicas.go
+++ b/foreign/go/client/tcp/tcp_replicas.go
@@ -564,14 +564,11 @@ func (r *replicaLinks) stopHeartbeat() {
}
func (r *replicaLinks) disconnect(wait bool) {
- stopped := r.detach()
- if wait {
- stopped()
- }
+ r.detach()(wait)
}
-// detach separates ownership changes from waiting for the old link tasks.
-func (r *replicaLinks) detach() func() {
+// detach removes ownership; its result closes sockets outside the session lock.
+func (r *replicaLinks) detach() func(bool) {
r.mu.Lock()
if r.cancel != nil {
r.cancel()
@@ -597,10 +594,13 @@ func (r *replicaLinks) detach() func() {
r.hasPrimary = false
r.hasRoster.Store(false)
r.mu.Unlock()
- for link := range links {
- link.close(ierror.ErrClientShutdown)
- }
- return func() {
+ return func(wait bool) {
+ for link := range links {
+ link.close(ierror.ErrClientShutdown)
+ }
+ if !wait {
+ return
+ }
if attempt != nil {
<-attempt.done
}
diff --git a/foreign/go/client/tcp/tcp_replicas_test.go b/foreign/go/client/tcp/tcp_replicas_test.go
index ea0ee649d..5557aac85 100644
--- a/foreign/go/client/tcp/tcp_replicas_test.go
+++ b/foreign/go/client/tcp/tcp_replicas_test.go
@@ -134,6 +134,7 @@ type managedLinkTestConnection struct {
closed chan<- struct{}
readStopped chan<- struct{}
resumeRead <-chan struct{}
+ resumeClose <-chan struct{}
}
func (connection *managedLinkTestConnection) Read(buffer []byte) (int, error) {
@@ -151,9 +152,54 @@ func (connection *managedLinkTestConnection) Close() error {
case connection.closed <- struct{}{}:
default:
}
+ if connection.resumeClose != nil {
+ <-connection.resumeClose
+ }
return err
}
+func TestSlowSocketCloseDoesNotHoldSessionLock(t *testing.T) {
+ for _, action := range []string{"disconnect", "close", "failed_login"} {
+ t.Run(action, func(t *testing.T) {
+ peer, connection := net.Pipe()
+ defer func() { _ = peer.Close() }()
+ closing, resume := make(chan struct{}, 1), make(chan struct{})
+ unblock := sync.OnceFunc(func() { close(resume) })
+ defer unblock()
+ client := newTestClient(t, &managedLinkTestConnection{Conn: connection, closed: closing, resumeClose: resume})
+ done := make(chan error, 1)
+ go func() {
+ switch action {
+ case "disconnect":
+ done <- client.Disconnect()
+ case "close":
+ done <- client.Close()
+ default:
+ _, err := client.LoginUser(context.Background(), "", "password")
+ done <- err
+ }
+ }()
+ select {
+ case <-closing:
+ case <-time.After(managedLinkTestBudget):
+ t.Fatal("the socket did not start closing")
+ }
+ locked := client.mtx.TryLock()
+ if locked {
+ client.mtx.Unlock()
+ }
+ require.True(t, locked, "socket shutdown blocked the session lock")
+ require.Nil(t, client.links.any())
+ unblock()
+ if action == "failed_login" {
+ require.Error(t, <-done)
+ } else {
+ require.NoError(t, <-done)
+ }
+ })
+ }
+}
+
func TestManagedLinks_StopHeartbeatAndReconnectOnSessionEnd(t *testing.T) {
for _, action := range []string{"logout", "disconnect", "close"} {
t.Run(action, func(t *testing.T) {
diff --git a/foreign/go/client/tcp/tcp_session_management.go b/foreign/go/client/tcp/tcp_session_management.go
index 611e184b1..7bd71b3c0 100644
--- a/foreign/go/client/tcp/tcp_session_management.go
+++ b/foreign/go/client/tcp/tcp_session_management.go
@@ -88,11 +88,15 @@ func (c *IggyTcpClient) register(ctx context.Context, code command.Code, credent
c.rememberedLogin = AutoLogin{}
c.loggedOut = true
c.invalidateSessionLocked()
+ var stopped func(bool)
if len(c.registerGate) != 0 {
c.session.Reset()
- c.links.disconnect(false)
+ stopped = c.links.detach()
}
c.mtx.Unlock()
+ if stopped != nil {
+ stopped(false)
+ }
callerCtx := ctx
defer func() {
if err == nil {
@@ -104,13 +108,17 @@ func (c *IggyTcpClient) register(ctx context.Context, code command.Code, credent
err = callerCtx.Err()
}
c.mtx.Lock()
- defer c.mtx.Unlock()
+ var stopped func(bool)
if c.epoch.Load() == epoch {
c.session.Reset()
c.invalidateSessionLocked()
c.rememberedLogin = AutoLogin{}
c.loggedOut = true
- c.links.disconnect(false)
+ stopped = c.links.detach()
+ }
+ c.mtx.Unlock()
+ if stopped != nil {
+ stopped(false)
}
}()
if ctx == nil {
@@ -154,12 +162,16 @@ func (c *IggyTcpClient) register(ctx context.Context, code command.Code, credent
if err = epoch.err(); err == nil {
err = ctx.Err()
}
+ var stopped func(bool)
if err == nil {
c.session.Reset()
c.invalidateSessionLocked()
- c.links.disconnect(false)
+ stopped = c.links.detach()
}
c.mtx.Unlock()
+ if stopped != nil {
+ stopped(false)
+ }
if err != nil {
return nil, err
}
@@ -584,7 +596,7 @@ func (c *IggyTcpClient) finishLogout(epoch *signInEpoch) {
c.resetSessionLocked(true)
stopped := c.links.detach()
c.mtx.Unlock()
- stopped()
+ stopped(true)
}
// signInCredentials are the last sign-in, else the configured AutoLogin. A
(iggy) 05/08: refactor(java): separate session reply bookkeeping
Posted by hg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch feat/spread-placement
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 5294276bf6c1bc8a1dd9706e5be1a9bd9008a80e
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Thu Oct 22 00:16:07 2026 +0200
refactor(java): separate session reply bookkeeping
---
.../iggy/client/async/tcp/AsyncTcpConnection.java | 5 ++---
.../client/async/tcp/vsr/VsrResponseHandler.java | 26 ++++++++++++++--------
2 files changed, 19 insertions(+), 12 deletions(-)
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncTcpConnection.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncTcpConnection.java
index 0c86e6d8d..092aecb4f 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncTcpConnection.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncTcpConnection.java
@@ -821,9 +821,8 @@ public class AsyncTcpConnection {
}
/**
- * Decides what follows a failed attempt of a call: the routed caller gets the failure, a call that has not sent
- * its frame starts again once a session is bound, a lost link of an uncertain call or a 58 or 89 refusal moves the
- * call to the next attempt after a leader recheck, and anything else ends the call.
+ * Retries unsent calls after session recovery and transient refusals after a leader recheck.
+ * Routed calls and terminal failures return to their caller.
*/
@SuppressWarnings({"checkstyle:CyclomaticComplexity", "checkstyle:NPathComplexity"})
private void completeFailedRequest(
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/VsrResponseHandler.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/VsrResponseHandler.java
index f42adf4b6..94f433006 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/VsrResponseHandler.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/VsrResponseHandler.java
@@ -226,13 +226,20 @@ public class VsrResponseHandler extends SimpleChannelInboundHandler<ByteBuf> {
&& resultCode != VsrHeaders.ERROR_TRANSIENT_NOT_COMMITTED
&& resultCode != VsrHeaders.ERROR_TRANSIENT_NOT_ACCEPTED
&& resultCode != IggyErrorCode.NOT_PRIMARY.getCode();
+ recordSessionReply(frame, pending, operation, resultCode, bodyStart, bodyLength);
+ if (resultCode != 0) {
+ throw IggyServerException.fromTcpResponse(resultCode, new byte[0]);
+ }
+
+ return frame.retainedSlice(bodyStart, bodyLength);
+ }
+
+ private void recordSessionReply(
+ ByteBuf frame, PendingRequest pending, int operation, long resultCode, int bodyStart, int bodyLength) {
boolean sessionBound = false;
synchronized (session) {
if (pending.identity.matches(session)) {
- if (pending.committed
- && (VsrOperation.isMetadata(operation)
- || operation == VsrOperation.REGISTER
- || operation == VsrOperation.LOGOUT)) {
+ if (pending.committed && advancesMetadata(operation)) {
session.observeMetadata(frame.getLongLE(frame.readerIndex() + VsrHeaders.REPLY_COMMIT_OFFSET));
}
if (resultCode == 0) {
@@ -245,11 +252,6 @@ public class VsrResponseHandler extends SimpleChannelInboundHandler<ByteBuf> {
session.owner().changed();
}
resetLogoutSession(pending, operation, resultCode);
- if (resultCode != 0) {
- throw IggyServerException.fromTcpResponse(resultCode, new byte[0]);
- }
-
- return frame.retainedSlice(bodyStart, bodyLength);
}
private int validatedOperation(ByteBuf frame) {
@@ -290,6 +292,12 @@ public class VsrResponseHandler extends SimpleChannelInboundHandler<ByteBuf> {
}
}
+ private static boolean advancesMetadata(int operation) {
+ return VsrOperation.isMetadata(operation)
+ || operation == VsrOperation.REGISTER
+ || operation == VsrOperation.LOGOUT;
+ }
+
private void applySessionEffects(ByteBuf frame, int operation, int bodyStart, int bodyLength) {
if (operation == VsrOperation.REGISTER) {
// A terminal register failure ships an empty body (no result
(iggy) 04/08: fix(java): fence read replies and release session waiters
Posted by hg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch feat/spread-placement
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 4a4bc762c4a77fd7a93471fa01f78410ee3a84ef
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Thu Oct 22 00:05:29 2026 +0200
fix(java): fence read replies and release session waiters
Register completed waiters while holding the session monitor, and a
late read could succeed after its caller signed out. Signal after
unlocking, fence read completion, and classify unsent resets as
disconnected instead of a lost retained operation.
---
.../iggy/client/async/tcp/AsyncTcpConnection.java | 3 ++-
.../client/async/tcp/vsr/ConsensusSession.java | 19 +++++++++++-----
.../client/async/tcp/vsr/VsrResponseHandler.java | 7 +++++-
.../tcp/AsyncTcpConnectionConcurrencyTest.java | 10 ++++++++-
.../client/async/tcp/vsr/ConsensusSessionTest.java | 26 ++++++++++++++++++++++
.../async/tcp/vsr/VsrResponseHandlerTest.java | 11 +++++++++
6 files changed, 67 insertions(+), 9 deletions(-)
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncTcpConnection.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncTcpConnection.java
index 266dddd57..0c86e6d8d 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncTcpConnection.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncTcpConnection.java
@@ -794,7 +794,8 @@ public class AsyncTcpConnection {
try {
synchronized (consensusSession) {
synchronized (consensusSession.owner()) {
- if (isLoginCommand && failoverState.signedOut()) {
+ if ((isLoginCommand || VsrOperation.operationForCode(commandCode) == VsrOperation.NON_REPLICATED)
+ && failoverState.signedOut()) {
throw failoverState.lineageEnded(error);
}
handlePostResponse(channel, commandCode, isLoginCommand, response, error);
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/ConsensusSession.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/ConsensusSession.java
index 294d0386f..2fd93ef60 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/ConsensusSession.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/ConsensusSession.java
@@ -100,6 +100,12 @@ public final class ConsensusSession {
/** Binds the fence epoch returned by a committed Register reply. */
void bind(long sessionEpoch) {
+ bindWithoutSignal(sessionEpoch);
+ owner().changed();
+ }
+
+ /** The reply handler signals after releasing the identity-check monitor. */
+ void bindWithoutSignal(long sessionEpoch) {
ConsensusSession owner;
synchronized (this) {
owner = parent;
@@ -115,11 +121,8 @@ public final class ConsensusSession {
}
}
if (owner != null) {
- owner.bind(sessionEpoch);
- return;
+ owner.bindWithoutSignal(sessionEpoch);
}
- // Wakes the calls that wait for a session, see Ticket#awaitSession.
- changed();
}
/**
@@ -601,7 +604,7 @@ public final class ConsensusSession {
}
}
- private void changed() {
+ void changed() {
CompletableFuture<Void> signal;
synchronized (this) {
signal = admissionChanged;
@@ -809,12 +812,14 @@ public final class ConsensusSession {
CompletableFuture<Void> changed = null;
boolean signedOut;
boolean ended;
+ boolean written;
synchronized (owner) {
if (result.isDone()) {
return;
}
signedOut = owner.lineage != lineage;
ended = completed;
+ written = sent;
if (!signedOut && !ended && owner.session == null) {
changed = owner.admissionChanged;
}
@@ -823,7 +828,9 @@ public final class ConsensusSession {
result.completeExceptionally(owner.lineageEnded(lineage, null));
} else if (ended) {
result.completeExceptionally(
- new IggyOperationException(IggyErrorCode.OPERATION_NOT_RETAINED, operationId));
+ written
+ ? new IggyOperationException(IggyErrorCode.OPERATION_NOT_RETAINED, operationId)
+ : new IggyNotConnectedException("The connection closed before the request was sent"));
} else if (changed != null) {
changed.thenRun(() -> awaitSession(lineage, result));
} else {
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/VsrResponseHandler.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/VsrResponseHandler.java
index b81d48246..f42adf4b6 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/VsrResponseHandler.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/VsrResponseHandler.java
@@ -226,6 +226,7 @@ public class VsrResponseHandler extends SimpleChannelInboundHandler<ByteBuf> {
&& resultCode != VsrHeaders.ERROR_TRANSIENT_NOT_COMMITTED
&& resultCode != VsrHeaders.ERROR_TRANSIENT_NOT_ACCEPTED
&& resultCode != IggyErrorCode.NOT_PRIMARY.getCode();
+ boolean sessionBound = false;
synchronized (session) {
if (pending.identity.matches(session)) {
if (pending.committed
@@ -236,9 +237,13 @@ public class VsrResponseHandler extends SimpleChannelInboundHandler<ByteBuf> {
}
if (resultCode == 0) {
applySessionEffects(frame, operation, bodyStart, bodyLength);
+ sessionBound = operation == VsrOperation.REGISTER;
}
}
}
+ if (sessionBound) {
+ session.owner().changed();
+ }
resetLogoutSession(pending, operation, resultCode);
if (resultCode != 0) {
throw IggyServerException.fromTcpResponse(resultCode, new byte[0]);
@@ -293,7 +298,7 @@ public class VsrResponseHandler extends SimpleChannelInboundHandler<ByteBuf> {
if (bodyLength < REGISTER_BODY_MIN_LEN) {
throw IggyServerException.fromTcpResponse(VsrHeaders.ERROR_UNAUTHENTICATED, new byte[0]);
}
- session.bind(frame.getLongLE(bodyStart + 4));
+ session.bindWithoutSignal(frame.getLongLE(bodyStart + 4));
}
}
diff --git a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncTcpConnectionConcurrencyTest.java b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncTcpConnectionConcurrencyTest.java
index b5ed7a0c7..e27e187e8 100644
--- a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncTcpConnectionConcurrencyTest.java
+++ b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncTcpConnectionConcurrencyTest.java
@@ -26,6 +26,7 @@ import io.netty.buffer.UnpooledHeapByteBuf;
import org.apache.iggy.client.async.tcp.vsr.ConsensusSession;
import org.apache.iggy.exception.IggyErrorCode;
import org.apache.iggy.exception.IggyOperationException;
+import org.apache.iggy.exception.IggyServerException;
import org.apache.iggy.partition.PartitionContext;
import org.apache.iggy.serde.CommandCode;
import org.junit.jupiter.api.Test;
@@ -665,6 +666,7 @@ class AsyncTcpConnectionConcurrencyTest {
try (ServerSocket socket = new ServerSocket(0, 1, InetAddress.getLoopbackAddress())) {
CompletableFuture<Request> observed = new CompletableFuture<>();
CompletableFuture<Request> registered = new CompletableFuture<>();
+ CompletableFuture<Void> releaseReply = new CompletableFuture<>();
var serving = CompletableFuture.runAsync(() -> {
try (Socket peer = socket.accept()) {
Request register = readRequest(peer.getInputStream());
@@ -672,6 +674,7 @@ class AsyncTcpConnectionConcurrencyTest {
writeResponse(peer.getOutputStream(), register, registerBody());
Request read = readRequest(peer.getInputStream());
observed.complete(read);
+ releaseReply.join();
writeResponse(peer.getOutputStream(), read, new byte[0]);
} catch (IOException error) {
throw new IllegalStateException("Mock VSR server failed", error);
@@ -704,7 +707,11 @@ class AsyncTcpConnectionConcurrencyTest {
}
payload.releaseEncoding.complete(null);
changed.get(5, TimeUnit.SECONDS);
- read.get(5, TimeUnit.SECONDS).release();
+ releaseReply.complete(null);
+ assertThatThrownBy(() -> read.get(5, TimeUnit.SECONDS))
+ .hasCauseInstanceOf(IggyServerException.class)
+ .satisfies(error -> assertThat(((IggyServerException) error.getCause()).getRawErrorCode())
+ .isEqualTo(IggyErrorCode.UNAUTHENTICATED.getCode()));
Request request = observed.get(5, TimeUnit.SECONDS);
Request register = registered.get(5, TimeUnit.SECONDS);
assertThat(Arrays.copyOfRange(
@@ -714,6 +721,7 @@ class AsyncTcpConnectionConcurrencyTest {
register.header(), REQUEST_CLIENT_OFFSET, REQUEST_CLIENT_OFFSET + CLIENT_ID_BYTES));
serving.get(5, TimeUnit.SECONDS);
} finally {
+ releaseReply.complete(null);
payload.releaseEncoding.complete(null);
connection.close().get(5, TimeUnit.SECONDS);
}
diff --git a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/vsr/ConsensusSessionTest.java b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/vsr/ConsensusSessionTest.java
index 0dfca3aaa..a59e942ab 100644
--- a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/vsr/ConsensusSessionTest.java
+++ b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/vsr/ConsensusSessionTest.java
@@ -24,6 +24,7 @@ import io.netty.buffer.PooledByteBufAllocator;
import io.netty.buffer.Unpooled;
import io.netty.channel.embedded.EmbeddedChannel;
import org.apache.iggy.exception.IggyErrorCode;
+import org.apache.iggy.exception.IggyNotConnectedException;
import org.apache.iggy.exception.IggyOperationException;
import org.apache.iggy.exception.IggyServerException;
import org.junit.jupiter.api.Test;
@@ -37,6 +38,31 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy;
class ConsensusSessionTest {
private static final long DEADLINE = TimeUnit.MINUTES.toNanos(1);
+ @Test
+ void shouldReportAnUnsentSessionWaitAsDisconnectedAfterReset() {
+ ConsensusSession session = new ConsensusSession();
+ EmbeddedChannel channel = new EmbeddedChannel();
+ long deadline = System.nanoTime() + DEADLINE;
+ try {
+ var ticket = session.acquire(
+ ConsensusSession.LaneKey.METADATA,
+ true,
+ VsrHeaders.HEADER_SIZE,
+ deadline,
+ channel.eventLoop())
+ .join();
+ var waiting = ticket.awaitSession(session.lineage(), deadline, channel.eventLoop());
+ assertThat(waiting).isNotDone();
+
+ session.reset();
+
+ assertThatThrownBy(waiting::join).hasCauseInstanceOf(IggyNotConnectedException.class);
+ } finally {
+ session.reset();
+ channel.finishAndReleaseAll();
+ }
+ }
+
@Test
void shouldAdmitLaneWaitersInOrderAfterThreeAdmissionChanges() {
ConsensusSession session = new ConsensusSession();
diff --git a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/vsr/VsrResponseHandlerTest.java b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/vsr/VsrResponseHandlerTest.java
index b9b581657..372081af5 100644
--- a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/vsr/VsrResponseHandlerTest.java
+++ b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/vsr/VsrResponseHandlerTest.java
@@ -148,6 +148,14 @@ class VsrResponseHandlerTest {
@Test
void shouldBindSessionOnRegisterReply() throws Exception {
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
+ var ticket = session.acquire(
+ ConsensusSession.LaneKey.METADATA, true, VsrHeaders.HEADER_SIZE, deadline, channel.eventLoop())
+ .join();
+ var heldMonitor = new AtomicBoolean();
+ var waiting = ticket.awaitSession(session.lineage(), deadline, channel.eventLoop())
+ .thenRun(() -> heldMonitor.set(Thread.holdsLock(session)));
+ assertThat(waiting).isNotDone();
CompletableFuture<ByteBuf> future = enqueue(VsrOperation.REGISTER, 0);
ByteBuf body = Unpooled.buffer();
body.writeIntLE(0); // result section: success
@@ -163,7 +171,10 @@ class VsrResponseHandlerTest {
assertThat(response.readUnsignedIntLE()).isEqualTo(1);
assertThat(session.isBound()).isTrue();
assertThat(session.boundSession()).isEqualTo(99);
+ waiting.get();
+ assertThat(heldMonitor).isFalse();
} finally {
+ ticket.complete();
response.release();
}
}
(iggy) 03/08: fix(go): preserve dial retries and fence closed logins
Posted by hg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch feat/spread-placement
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 1d873975effe979f2c149eef4d93c19a0a108ca2
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Thu Oct 22 00:02:14 2026 +0200
fix(go): preserve dial retries and fence closed logins
Connect spent its sign-in deadline while waiting to dial, and a racing
login could replace the epoch that Close needed to detach links. Scope
the deadline to registration and reject login after shutdown begins.
---
foreign/go/client/tcp/tcp_connect_test.go | 39 +++++++++++++++++++++++++
foreign/go/client/tcp/tcp_core.go | 13 +++++----
foreign/go/client/tcp/tcp_session_management.go | 11 +++++++
3 files changed, 58 insertions(+), 5 deletions(-)
diff --git a/foreign/go/client/tcp/tcp_connect_test.go b/foreign/go/client/tcp/tcp_connect_test.go
index 89463fd84..5a670012d 100644
--- a/foreign/go/client/tcp/tcp_connect_test.go
+++ b/foreign/go/client/tcp/tcp_connect_test.go
@@ -103,6 +103,45 @@ func TestConnectBoundsSilentSignInWithoutCallerDeadline(t *testing.T) {
})
}
+func TestConnectDoesNotSpendSignInBudgetWaitingForDial(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ client := newDialingClient(t, "127.0.0.1:1")
+ for range cap(client.links.permits) {
+ client.links.permits <- struct{}{}
+ }
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ done := make(chan error, 1)
+ go func() { done <- client.Connect(ctx) }()
+ time.Sleep(responseReadTimeout + time.Second)
+ synctest.Wait()
+ select {
+ case err := <-done:
+ t.Fatalf("Connect consumed the sign-in budget before dialing: %v", err)
+ default:
+ }
+ cancel()
+ require.ErrorIs(t, <-done, context.Canceled)
+ })
+}
+
+func TestExplicitSignInCannotReplaceEpochAfterCloseStarts(t *testing.T) {
+ serverConnection, connection := net.Pipe()
+ defer func() { _ = serverConnection.Close() }()
+ client := newTestClient(t, connection)
+ server := serve(serverConnection, func(_ int, read request) []byte {
+ return registerReplyFrame(signedInUserID, managedLinkSession)
+ })
+ epoch := client.epoch.Load()
+ client.closeOnce.Do(func() { close(client.closed) })
+ _, err := client.LoginUser(context.Background(), "user", "password")
+ require.ErrorIs(t, err, ierror.ErrClientShutdown)
+ require.Same(t, epoch, client.epoch.Load(), "a new epoch would cancel Close's pending detach")
+ require.Empty(t, server.recorded())
+ require.NoError(t, client.Close())
+ require.False(t, linkReady(client.links.bootstrap))
+}
+
// testListener is a VSR server on a loopback port. The handler is called with
// the zero-based index of the accepted connection and of the frame within it.
type testListener struct {
diff --git a/foreign/go/client/tcp/tcp_core.go b/foreign/go/client/tcp/tcp_core.go
index 6fca25740..ac38cf2e0 100644
--- a/foreign/go/client/tcp/tcp_core.go
+++ b/foreign/go/client/tcp/tcp_core.go
@@ -429,8 +429,11 @@ func (c *IggyTcpClient) Connect(ctx context.Context) error {
return ierror.ErrInvalidConfiguration
}
epoch := c.callEpoch(ctx)
- ctx, endSignIn := signInContext(ctx, epoch)
- defer endSignIn()
+ ctx, cancel := context.WithCancel(ctx)
+ stop := context.AfterFunc(epoch.ctx, cancel)
+ defer cancel()
+ defer stop()
+ ctx = context.WithValue(ctx, operationScope{}, epoch)
c.connectMu.Lock()
if attempt := c.connectionAttempt; attempt != nil {
c.connectMu.Unlock()
@@ -452,11 +455,11 @@ func (c *IggyTcpClient) Connect(ctx context.Context) error {
if lifetime == nil {
err = ierror.ErrNotConnected
} else {
- sessionCtx, cancel := context.WithCancel(ctx)
- stop := context.AfterFunc(lifetime, cancel)
+ sessionCtx, endSignIn := signInContext(ctx, epoch)
+ stop := context.AfterFunc(lifetime, endSignIn)
err = c.establishSession(sessionCtx, link)
stop()
- cancel()
+ endSignIn()
}
}
if ended := epoch.err(); ended != nil {
diff --git a/foreign/go/client/tcp/tcp_session_management.go b/foreign/go/client/tcp/tcp_session_management.go
index 1f8c52245..611e184b1 100644
--- a/foreign/go/client/tcp/tcp_session_management.go
+++ b/foreign/go/client/tcp/tcp_session_management.go
@@ -75,6 +75,12 @@ func (c *IggyTcpClient) acquireSignIn(ctx context.Context) error {
func (c *IggyTcpClient) register(ctx context.Context, code command.Code, credentials Credentials) (identity *iggcon.IdentityInfo, err error) {
// Even an invalid or cancelled explicit attempt leaves the previous owner.
c.mtx.Lock()
+ select {
+ case <-c.closed:
+ c.mtx.Unlock()
+ return nil, ierror.ErrClientShutdown
+ default:
+ }
c.endRecoveryLocked()
c.endSignInEpochLocked(nil)
epoch := c.epoch.Load()
@@ -175,6 +181,11 @@ func (c *IggyTcpClient) register(ctx context.Context, code command.Code, credent
}
c.mtx.Lock()
defer c.mtx.Unlock()
+ select {
+ case <-c.closed:
+ return nil, ierror.ErrClientShutdown
+ default:
+ }
if err = epoch.err(); err != nil {
return nil, err
}
(iggy) 02/08: test(csharp): await counters and assert final checkpoints
Posted by hg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch feat/spread-placement
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit eac054bdd1db9a57f260284a0486fde4b970b5af
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Thu Oct 22 00:00:41 2026 +0200
test(csharp): await counters and assert final checkpoints
Disposal checkpoints the last delivered message, and node-local
statistics can trail committed writes. Assert exactly five deliveries
and wait for replicated counters before checking their values.
---
.../Fixtures/IggyServerFixture.cs | 16 ++++++++++++++++
.../Iggy_SDK.Tests.Integration/IggyConsumerTests.cs | 9 +++++----
.../Iggy_SDK.Tests.Integration/IggyTypedConsumerTests.cs | 3 ++-
.../csharp/Iggy_SDK.Tests.Integration/StreamsTests.cs | 4 +++-
foreign/csharp/Iggy_SDK.Tests.Integration/TopicsTests.cs | 5 +++--
5 files changed, 29 insertions(+), 8 deletions(-)
diff --git a/foreign/csharp/Iggy_SDK.Tests.Integration/Fixtures/IggyServerFixture.cs b/foreign/csharp/Iggy_SDK.Tests.Integration/Fixtures/IggyServerFixture.cs
index ac28ebb1a..6b1afdbc8 100644
--- a/foreign/csharp/Iggy_SDK.Tests.Integration/Fixtures/IggyServerFixture.cs
+++ b/foreign/csharp/Iggy_SDK.Tests.Integration/Fixtures/IggyServerFixture.cs
@@ -72,6 +72,22 @@ public class IggyServerFixture : IAsyncInitializer, IAsyncDisposable
/// </summary>
public static readonly TimeSpan SessionReportInterval = TimeSpan.FromMilliseconds(500);
+ private static readonly TimeSpan StatsConvergenceTimeout = TimeSpan.FromSeconds(10);
+ private static readonly TimeSpan StatsPollInterval = TimeSpan.FromMilliseconds(100);
+
+ /// <summary>Node-local counters catch up after partition commit broadcasts.</summary>
+ public static async Task<T> ReadConvergedStatsAsync<T>(Func<Task<T>> read, Func<T, bool> converged)
+ {
+ var deadline = Environment.TickCount64 + (long)StatsConvergenceTimeout.TotalMilliseconds;
+ var result = await read();
+ while (!converged(result) && Environment.TickCount64 < deadline)
+ {
+ await Task.Delay(StatsPollInterval);
+ result = await read();
+ }
+ return result;
+ }
+
/// <summary>
/// Extra environment variables for every node, layered over the cluster's base configuration.
/// Heartbeat verification is on for every fixture so the eviction tests share the general server
diff --git a/foreign/csharp/Iggy_SDK.Tests.Integration/IggyConsumerTests.cs b/foreign/csharp/Iggy_SDK.Tests.Integration/IggyConsumerTests.cs
index ef7c062f0..a5b2d3660 100644
--- a/foreign/csharp/Iggy_SDK.Tests.Integration/IggyConsumerTests.cs
+++ b/foreign/csharp/Iggy_SDK.Tests.Integration/IggyConsumerTests.cs
@@ -422,13 +422,12 @@ public class IggyConsumerTests
await foreach (var message in consumer.ReceiveAsync(cts.Token))
{
+ lastOffset = message.CurrentOffset;
+ receivedCount++;
if (receivedCount >= 5)
{
break;
}
-
- lastOffset = message.CurrentOffset;
- receivedCount++;
}
await consumer.DisposeAsync();
@@ -439,7 +438,9 @@ public class IggyConsumerTests
1u);
offset.ShouldNotBeNull();
- offset.StoredOffset.ShouldBe(4ul);
+ receivedCount.ShouldBe(5);
+ lastOffset.ShouldBe(4ul);
+ offset.StoredOffset.ShouldBe(lastOffset);
}
[Test]
diff --git a/foreign/csharp/Iggy_SDK.Tests.Integration/IggyTypedConsumerTests.cs b/foreign/csharp/Iggy_SDK.Tests.Integration/IggyTypedConsumerTests.cs
index cf2799672..a1cf6421b 100644
--- a/foreign/csharp/Iggy_SDK.Tests.Integration/IggyTypedConsumerTests.cs
+++ b/foreign/csharp/Iggy_SDK.Tests.Integration/IggyTypedConsumerTests.cs
@@ -128,7 +128,8 @@ public class IggyTypedConsumerTests
1u);
offset.ShouldNotBeNull();
- offset.StoredOffset.ShouldBe(3ul);
+ count.ShouldBe(5);
+ offset.StoredOffset.ShouldBe(4ul);
}
[Test]
diff --git a/foreign/csharp/Iggy_SDK.Tests.Integration/StreamsTests.cs b/foreign/csharp/Iggy_SDK.Tests.Integration/StreamsTests.cs
index 460bf6d42..648b25f0f 100644
--- a/foreign/csharp/Iggy_SDK.Tests.Integration/StreamsTests.cs
+++ b/foreign/csharp/Iggy_SDK.Tests.Integration/StreamsTests.cs
@@ -153,7 +153,9 @@ public class StreamsTests
new Message(Guid.NewGuid(), "Test message 7"u8.ToArray())
]);
- var response = await client.GetStreamByIdAsync(Identifier.String(streamName));
+ var response = await IggyServerFixture.ReadConvergedStatsAsync(
+ () => client.GetStreamByIdAsync(Identifier.String(streamName)),
+ stream => stream?.MessagesCount == 7);
response.ShouldNotBeNull();
response.Name.ShouldBe(streamName);
response.Size.ShouldBeGreaterThan(0u);
diff --git a/foreign/csharp/Iggy_SDK.Tests.Integration/TopicsTests.cs b/foreign/csharp/Iggy_SDK.Tests.Integration/TopicsTests.cs
index d0e3aca53..87ee53581 100644
--- a/foreign/csharp/Iggy_SDK.Tests.Integration/TopicsTests.cs
+++ b/foreign/csharp/Iggy_SDK.Tests.Integration/TopicsTests.cs
@@ -161,8 +161,9 @@ public class TopicsTests
Partitioning.None(), GetMessages(i + 2));
}
- var response = await client.GetTopicByIdAsync(Identifier.String(streamName),
- Identifier.String("Parts Topic"));
+ var response = await IggyServerFixture.ReadConvergedStatsAsync(
+ () => client.GetTopicByIdAsync(Identifier.String(streamName), Identifier.String("Parts Topic")),
+ topic => topic?.MessagesCount == 9);
response.ShouldNotBeNull();
response.Name.ShouldBe("Parts Topic");
(iggy) 06/08: fix(sdk): retain replay identity through session resets
Posted by hg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch feat/spread-placement
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 35d332a19ebb9afee3ff8c751f8dbd23410d2db3
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Thu Oct 22 00:21:00 2026 +0200
fix(sdk): retain replay identity through session resets
A replay could lose its retained operation while routing across a
session reset and report a safe retry for an uncertain write. Preserve
its identity and end disconnect lineage before closing links.
---
core/sdk/src/binary/mod.rs | 5 +++--
core/sdk/src/session_owner/lanes.rs | 20 +++++++++++++++++++-
2 files changed, 22 insertions(+), 3 deletions(-)
diff --git a/core/sdk/src/binary/mod.rs b/core/sdk/src/binary/mod.rs
index 0ded804bc..00172b281 100644
--- a/core/sdk/src/binary/mod.rs
+++ b/core/sdk/src/binary/mod.rs
@@ -44,9 +44,10 @@ macro_rules! impl_binary_client {
}
async fn disconnect(&self) -> Result<(), iggy_common::IggyError> {
- self.driver.links().disconnect();
self.driver.session().forget();
- self.driver.session().reset().map(|_| ())
+ let result = self.driver.session().reset().map(|_| ());
+ self.driver.links().disconnect();
+ result
}
async fn shutdown(&self) -> Result<(), iggy_common::IggyError> {
diff --git a/core/sdk/src/session_owner/lanes.rs b/core/sdk/src/session_owner/lanes.rs
index 0a9e47f61..0e8058d7c 100644
--- a/core/sdk/src/session_owner/lanes.rs
+++ b/core/sdk/src/session_owner/lanes.rs
@@ -312,7 +312,7 @@ impl LaneTicket {
{
return Ok(());
}
- if self.sent && self.operation.is_some() {
+ if (self.sent || self.resumed) && self.operation.is_some() {
return Err(IggyError::Unauthenticated);
}
self.operation = None;
@@ -1187,6 +1187,24 @@ mod tests {
);
}
+ #[tokio::test(start_paused = true)]
+ async fn given_a_resumed_write_when_reset_during_routing_should_preserve_its_unknown_outcome() {
+ let owner = bound_owner();
+ let op = retain_write(&owner, partition_lane()).await;
+ let (mut replay, _retained) = owner.resume(op, deadline()).await.unwrap();
+ owner.reset().unwrap();
+
+ assert_eq!(
+ replay.reserve(SEND_MESSAGES_CODE, 0, deadline()).await,
+ Err(IggyError::Unauthenticated)
+ );
+ assert_eq!(replay.retain(), IggyError::OperationUncertain(op));
+ assert!(matches!(
+ owner.resume(op, deadline()).await,
+ Err(IggyError::OperationNotRetained(missing)) if missing == op
+ ));
+ }
+
#[tokio::test(start_paused = true)]
async fn given_a_resumed_op_when_its_replay_ends_should_stay_retained_until_completed() {
let owner = bound_owner();