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 21:28:56 UTC
(iggy) 02/02: fix(sdk): preserve Swift call lineage and offset replay
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 dc4be4c922061bb7c0182ec8daa0299969e8d9a3
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Wed Oct 21 23:28:26 2026 +0200
fix(sdk): preserve Swift call lineage and offset replay
End disconnected callers with their lineage refusal and retain offset
write identity across a lost data connection. Session evictions remain
final outcomes rather than reconnect triggers.
---
foreign/swift/Sources/Iggy/Client/ClientCore.swift | 26 +++++++-----
foreign/swift/Sources/Iggy/Client/PollRouter.swift | 9 +++--
.../Tests/IggyTests/ClientTransportTests.swift | 47 ++++++++++++++++++++++
foreign/swift/Tests/IggyTests/MessagingTests.swift | 6 +--
4 files changed, 72 insertions(+), 16 deletions(-)
diff --git a/foreign/swift/Sources/Iggy/Client/ClientCore.swift b/foreign/swift/Sources/Iggy/Client/ClientCore.swift
index cb2bb95ed..932cb9f27 100644
--- a/foreign/swift/Sources/Iggy/Client/ClientCore.swift
+++ b/foreign/swift/Sources/Iggy/Client/ClientCore.swift
@@ -185,10 +185,10 @@ actor ClientCore {
code: UInt32, payload: [UInt8], context: PartitionContext = .empty, deadline limit: ContinuousClock.Instant? = nil,
lineage captured: UInt64? = nil
) async throws -> [UInt8] {
- if disconnectedByCaller {
- throw IggyError(state == .shutdown ? .clientShutdown : .notConnected)
- }
+ if state == .shutdown { throw IggyError(.clientShutdown) }
let lineage = captured ?? lineage
+ if let refusal = lineageRefusal(lineage, code: code) { throw refusal }
+ if disconnectedByCaller { throw IggyError(.notConnected) }
let deadline = requestDeadline(within: limit)
var outcome = RetryOutcome()
var backoff = PollRouter.Backoff()
@@ -308,9 +308,9 @@ actor ClientCore {
func exchangeOnce(
code: UInt32, payload: [UInt8], context: PartitionContext, deadline limit: ContinuousClock.Instant? = nil, lineage captured: UInt64? = nil
) async throws -> [UInt8] {
- if disconnectedByCaller || state == .shutdown {
- throw IggyError(state == .shutdown ? .clientShutdown : .notConnected)
- }
+ if state == .shutdown { throw IggyError(.clientShutdown) }
+ if let refusal = lineageRefusal(captured ?? lineage, code: code) { throw refusal }
+ if disconnectedByCaller { throw IggyError(.notConnected) }
var request: VSRFrame.EncodedRequest?
var epoch = connectionEpoch
var outcome = RetryOutcome()
@@ -381,11 +381,12 @@ actor ClientCore {
/// The refusal of a call that started under `captured` once the caller
/// signed in or out since: nothing of it may be written any more.
- func lineageRefusal(_ captured: UInt64, code: UInt32) -> IggyError? {
+ func lineageRefusal(_ captured: UInt64, code: UInt32? = nil) -> IggyError? {
guard captured != lineage else {
return nil
}
- return IggyError(.unauthenticated, context: "the caller signed in or out after command \(code) was issued")
+ let context = code.map { "after command \($0) was issued" } ?? "while the call was waiting"
+ return IggyError(.unauthenticated, context: "the caller signed in or out \(context)")
}
/// Dials `address` with the client's TLS, for a data connection. A host
@@ -423,9 +424,14 @@ actor ClientCore {
reconnected = try await withDeadline(ContinuousClock.now + configuration.requestTimeout) { [self] in
try await self.connect()
}
- } catch let error as IggyError where error.isCredentialRejection {
- throw started == lineage ? Self.signInRefused(error) : error
+ } catch {
+ if let refusal = lineageRefusal(started) { throw refusal }
+ if let refusal = error as? IggyError, refusal.isCredentialRejection {
+ throw Self.signInRefused(refusal)
+ }
+ throw error
}
+ if let refusal = lineageRefusal(started) { throw refusal }
if reconnected == nil {
throw IggyError(.transientNotAccepted, context: "could not reconnect within the request timeout")
}
diff --git a/foreign/swift/Sources/Iggy/Client/PollRouter.swift b/foreign/swift/Sources/Iggy/Client/PollRouter.swift
index 8efd8c898..70fad6d9a 100644
--- a/foreign/swift/Sources/Iggy/Client/PollRouter.swift
+++ b/foreign/swift/Sources/Iggy/Client/PollRouter.swift
@@ -455,8 +455,11 @@ final class PollRouter: Sendable {
data.consumerSession = nil
}
slot.connection.withLockedValue { $0 = data }
- if replay.outcome.isUncertain, error.isConnectionLoss, error.code != .staleClient {
- throw error
+ if error.isConnectionLoss, error.code != .staleClient {
+ if replay.frame?.operation.isPartition == true, error.code != .notConnected, error.code != .cannotEstablishConnection {
+ replay.outcome.markUncertain()
+ }
+ if replay.outcome.isUncertain { throw error }
}
throw Self.gradedLoss(ClientCore.graded(error), on: endpoint)
}
@@ -567,7 +570,7 @@ final class PollRouter: Sendable {
// its session: either way the request is lost with the
// connection, not refused.
await connection.close()
- throw IggyError(.disconnected, context: "\(refusal)")
+ throw IggyError(.transientNotCommitted, context: "\(refusal)")
}
answer = refusal
}
diff --git a/foreign/swift/Tests/IggyTests/ClientTransportTests.swift b/foreign/swift/Tests/IggyTests/ClientTransportTests.swift
index 8cefbe7f1..54c2de57c 100644
--- a/foreign/swift/Tests/IggyTests/ClientTransportTests.swift
+++ b/foreign/swift/Tests/IggyTests/ClientTransportTests.swift
@@ -115,6 +115,53 @@ private func pingCount(_ server: FakeVSRServer) async -> Int {
@Suite("Client transport", .serialized)
struct ClientTransportTests {
+ @Test(arguments: [false, true])
+ func aCallStartedBeforeDisconnectEndsUnauthenticated(singleAttempt: Bool) async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ let client = makeClient(server)
+ defer { Task { await client.shutdown() } }
+ try await client.connect()
+ let lineage = await client.core.lineage
+ await client.disconnect()
+
+ await expectCode(.unauthenticated) {
+ if singleAttempt {
+ _ = try await client.core.exchangeOnce(code: CommandCode.ping.rawValue, payload: [], context: .empty, lineage: lineage)
+ } else {
+ _ = try await client.core.exchange(code: CommandCode.ping.rawValue, payload: [], lineage: lineage)
+ }
+ }
+ await expectCode(.notConnected) { try await client.ping() }
+ #expect(await pingCount(server) == 0)
+ }
+
+ @Test func aCallWaitingForReconnectEndsUnauthenticatedWhenTheCallerDisconnects() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ let client = makeClient(server)
+ defer { Task { await client.shutdown() } }
+ try await client.connect()
+ await client.core.disconnectTransport()
+ let signingIn = Locked(false)
+ let released = Locked(false)
+ await server.respond { request in
+ if request.code == CommandCode.bindSession.rawValue { return .status(IggyErrorCode.unauthenticated.rawValue) }
+ if request.operation == .register {
+ signingIn.withLock { $0 = true }
+ _ = await eventually { released.withLock { $0 } }
+ }
+ return FakeVSRServer.defaultReply(request, address: server.address)
+ }
+ let call = Task { try await client.getStreams() }
+ #expect(await eventually { signingIn.withLock { $0 } })
+ await client.disconnect()
+ released.withLock { $0 = true }
+
+ await expectCode(.unauthenticated) { _ = try await call.value }
+ #expect(await server.requests.allSatisfy { $0.code != CommandCode.getStreams.rawValue })
+ }
+
@Test func loginBindsTheSessionAndSequencesRequests() async throws {
let server = try await FakeVSRServer.started()
defer { Task { await server.stop() } }
diff --git a/foreign/swift/Tests/IggyTests/MessagingTests.swift b/foreign/swift/Tests/IggyTests/MessagingTests.swift
index 4ec147cf6..0569156b8 100644
--- a/foreign/swift/Tests/IggyTests/MessagingTests.swift
+++ b/foreign/swift/Tests/IggyTests/MessagingTests.swift
@@ -1776,8 +1776,8 @@ struct MessagingTests {
#expect(Set(attempts.map(\.requestID)).count == 1)
}
- @Test(arguments: [1, 3])
- func anUncommittedOffsetWriteReplaysAcrossLostDataConnections(_ losses: Int) async throws {
+ @Test(arguments: [1, 3], [false, true])
+ func anOffsetWriteReplaysAcrossLostDataConnections(_ losses: Int, refusedFirst: Bool) async throws {
let primary = try await FakeVSRServer.started()
let coordinator = try await Self.coordinator(routingTo: primary)
defer {
@@ -1791,7 +1791,7 @@ struct MessagingTests {
let attempt = attempts.withLock {
$0 += 1; return $0
}
- if attempt <= losses { return .statusThenClose(IggyErrorCode.transientNotCommitted.rawValue) }
+ if attempt <= losses { return refusedFirst ? .statusThenClose(IggyErrorCode.transientNotCommitted.rawValue) : .close }
}
return FakeVSRServer.defaultReply(request, address: primary.address)
}