hubcio commented on code in PR #4503: URL: https://github.com/apache/iggy/pull/4503#discussion_r4256654800
########## foreign/swift/Sources/Iggy/Client/ClientCore.swift: ########## @@ -0,0 +1,956 @@ +// 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. + +import Foundation +import Logging + +/// The connection state machine behind ``IggyClient``: one lockstep +/// connection, its consensus session, the credentials to restore it with, +/// and the cluster roster to fail over to. Modelled on the Rust `TcpClient` +/// and `leader_aware`. +/// +/// One deliberate difference: a reconnect registers a new session instead of +/// resuming the old one, so a replicated write whose outcome is unknown is +/// never replayed across connections. +actor ClientCore { + /// Backoff before replaying a request the server answered with an explicit + /// transient frame. The reply arrives promptly, so a short pause keeps the + /// replay from spinning while the primary catches up. + static let notReadyRetryInterval: Duration = .milliseconds(50) + /// How long a request replays "not accepted" on the same connection before + /// the client re-checks cluster leadership. + static let transientFailoverCheckInterval: Duration = .seconds(2) + /// Bound on one dial while other endpoints are queued behind it. + static let failoverDialTimeout: Duration = .seconds(2) + /// Bound on the roster read that follows a manual sign-in. + static let rosterReadTimeout: Duration = .seconds(5) + /// How long to wait for a transiently leaderless cluster to elect. + static let leaderlessWaitBudget: Duration = .seconds(5) + static let leaderlessPollInterval: Duration = .milliseconds(250) + static let maxLeaderRedirects = 3 + + let configuration: ClientConfiguration + let connector: any IggyConnector + let logger: Logger + + private(set) var state: ClientState = .disconnected + private var connection: (any IggyConnection)? + /// Counts the connections installed so far, so a request that failed on + /// an old one does not tear down the one that replaced it. + private var connectionEpoch: UInt64 = 0 + private var session = ConsensusSession() + private var currentAddress: String + private var rosterEndpoints: [String] = [] + private var rosterLearned = false + private var rememberedSignIn: (credentials: Credentials, userID: UInt32)? + /// The password a committed change gave the configured auto-login user; + /// the configuration cannot be rewritten and the old password is dead. + private var configuredPassword: String? + /// The name a committed rename gave the configured auto-login user. + private var configuredUsername: String? + private var connectedAt: ContinuousClock.Instant? + private var redirects = 0 + private var skipAutoLoginOnce = false + /// Set by an explicit ``disconnect()`` and cleared by the caller's next + /// ``connectByCaller()``: in between, nothing may bring the client back. + private var disconnectedByCaller = false + private var eventSubscribers: [UUID: AsyncStream<DiagnosticEvent>.Continuation] = [:] + private let exchangeLock = AsyncMutex() + private let routingLock = AsyncMutex() + private let connectGate = SingleFlight() + private var heartbeatTask: Task<Void, Never>? + + init(configuration: ClientConfiguration, connector: any IggyConnector, logger: Logger) { + self.configuration = configuration + self.connector = connector + self.logger = logger + currentAddress = configuration.address + } + + // MARK: State and events + + var clientAddress: String? { connection?.localAddress } + var serverAddress: String { currentAddress } + var isSignedIn: Bool { state == .authenticated } + + private func setState(_ newState: ClientState) { + state = newState + } + + func subscribeEvents() -> AsyncStream<DiagnosticEvent> { + let id = UUID() + let (stream, continuation) = AsyncStream<DiagnosticEvent>.makeStream(bufferingPolicy: .bufferingNewest(256)) + eventSubscribers[id] = continuation Review Comment: warning: event subscriptions created after shutdown never finish, so callers can wait forever for their first event. finish new streams immediately when the client is shut down. ########## foreign/swift/Sources/Iggy/Client/ClientCore.swift: ########## @@ -0,0 +1,956 @@ +// 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. + +import Foundation +import Logging + +/// The connection state machine behind ``IggyClient``: one lockstep +/// connection, its consensus session, the credentials to restore it with, +/// and the cluster roster to fail over to. Modelled on the Rust `TcpClient` +/// and `leader_aware`. +/// +/// One deliberate difference: a reconnect registers a new session instead of +/// resuming the old one, so a replicated write whose outcome is unknown is +/// never replayed across connections. +actor ClientCore { + /// Backoff before replaying a request the server answered with an explicit + /// transient frame. The reply arrives promptly, so a short pause keeps the + /// replay from spinning while the primary catches up. + static let notReadyRetryInterval: Duration = .milliseconds(50) + /// How long a request replays "not accepted" on the same connection before + /// the client re-checks cluster leadership. + static let transientFailoverCheckInterval: Duration = .seconds(2) + /// Bound on one dial while other endpoints are queued behind it. + static let failoverDialTimeout: Duration = .seconds(2) + /// Bound on the roster read that follows a manual sign-in. + static let rosterReadTimeout: Duration = .seconds(5) + /// How long to wait for a transiently leaderless cluster to elect. + static let leaderlessWaitBudget: Duration = .seconds(5) + static let leaderlessPollInterval: Duration = .milliseconds(250) + static let maxLeaderRedirects = 3 + + let configuration: ClientConfiguration + let connector: any IggyConnector + let logger: Logger + + private(set) var state: ClientState = .disconnected + private var connection: (any IggyConnection)? + /// Counts the connections installed so far, so a request that failed on + /// an old one does not tear down the one that replaced it. + private var connectionEpoch: UInt64 = 0 + private var session = ConsensusSession() + private var currentAddress: String + private var rosterEndpoints: [String] = [] + private var rosterLearned = false + private var rememberedSignIn: (credentials: Credentials, userID: UInt32)? + /// The password a committed change gave the configured auto-login user; + /// the configuration cannot be rewritten and the old password is dead. + private var configuredPassword: String? + /// The name a committed rename gave the configured auto-login user. + private var configuredUsername: String? + private var connectedAt: ContinuousClock.Instant? + private var redirects = 0 + private var skipAutoLoginOnce = false + /// Set by an explicit ``disconnect()`` and cleared by the caller's next + /// ``connectByCaller()``: in between, nothing may bring the client back. + private var disconnectedByCaller = false + private var eventSubscribers: [UUID: AsyncStream<DiagnosticEvent>.Continuation] = [:] + private let exchangeLock = AsyncMutex() + private let routingLock = AsyncMutex() + private let connectGate = SingleFlight() + private var heartbeatTask: Task<Void, Never>? + + init(configuration: ClientConfiguration, connector: any IggyConnector, logger: Logger) { + self.configuration = configuration + self.connector = connector + self.logger = logger + currentAddress = configuration.address + } + + // MARK: State and events + + var clientAddress: String? { connection?.localAddress } + var serverAddress: String { currentAddress } + var isSignedIn: Bool { state == .authenticated } + + private func setState(_ newState: ClientState) { + state = newState + } + + func subscribeEvents() -> AsyncStream<DiagnosticEvent> { + let id = UUID() + let (stream, continuation) = AsyncStream<DiagnosticEvent>.makeStream(bufferingPolicy: .bufferingNewest(256)) + eventSubscribers[id] = continuation + continuation.onTermination = { [weak self] _ in + guard let self else { return } + Task { await self.removeSubscriber(id) } + } + return stream + } + + private func removeSubscriber(_ id: UUID) { + eventSubscribers[id] = nil + } + + private func publish(_ event: DiagnosticEvent) { + for continuation in eventSubscribers.values { + continuation.yield(event) + } + } + + // MARK: Exchange + + /// Sends a request and returns its reply body, reconnecting and replaying + /// when the failure is one a fresh connection recovers from. + /// + /// A cancelled caller gets `CancellationError`. A request it already wrote + /// takes the connection with it, since its reply can no longer be told + /// from the next one, and the next caller reconnects. + func exchange(code: UInt32, payload: [UInt8]) async throws -> [UInt8] { + if disconnectedByCaller { + throw IggyError(state == .shutdown ? .clientShutdown : .notConnected) + } + var failedEpoch = connectionEpoch + do { + return try await sendRaw(code: code, payload: payload, epoch: &failedEpoch) + } catch let error as IggyError { + // A cancelled caller cannot use a new connection: every exchange + // it starts is abandoned at once. + try Task.checkCancellation() + guard error.isReconnectable else { + throw error + } + if code == CommandCode.getClusterMetadata.rawValue { + // An unauthenticated roster read cannot be repaired by a + // reconnect; it would re-issue the same read forever. + throw error + } + guard configuration.reconnection.enabled else { + throw IggyError(.disconnected, context: "automatic reconnection is disabled") + } + let isLogin = CommandCode.isRegister(code) + if !isLogin, signInCredentials() == nil { + // With no credentials a reconnect cannot restore the session, so + // anything but a sign-in fails instead of replaying + // unauthenticated. The sign-in itself is the exception: the + // server stays silent on a transient register failure and + // relies on the client replaying it. + throw error + } + if isLogin, await connectGate.isActive { + // The sign-in ran inside a connect sweep, which owns the + // endpoint walk; the sweep reports the failure itself. + throw error + } + // A replay over a new connection runs under a new session, and the + // server deduplicates by (client, request). Replay only what + // provably never reached the log. + let replay = replayAfterSessionResetIsSafe(code: code, error: error) + await routingLock.lock() + // Another caller may have reconnected while this one waited for + // the lock; its connection is not the one that failed. + if failedEpoch == connectionEpoch { + await disconnectTransport() + } + if isLogin { + skipAutoLoginOnce = true + } + logger.info("Reconnecting to the server", metadata: ["server_address": "\(currentAddress)", "error": "\(error)"]) + do { + try await connect() Review Comment: warning: reconnect attempts outlive the request deadline, so requests can remain pending forever despite `requestTimeout`. carry one deadline through reconnection, authentication, leader discovery, and replay. ########## foreign/swift/Sources/Iggy/Utilities/AsyncMutex.swift: ########## @@ -0,0 +1,63 @@ +// 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. + +/// A FIFO mutex that can be held across suspension points. +/// +/// Actors release their isolation at every `await`, so an actor method that +/// writes a request and then awaits the reply would let a second caller +/// interleave on the same connection. The lockstep exchange holds this +/// instead. +actor AsyncMutex { + private var locked = false + private var waiters: [CheckedContinuation<Void, Never>] = [] + + func lock() async { + if !locked { + locked = true + return + } + await withCheckedContinuation { waiters.append($0) } + } + + /// Hands the lock to the longest waiter, or frees it. + func unlock() { + if waiters.isEmpty { + locked = false + return + } + waiters.removeFirst().resume() + } +} + +/// A single-flight gate: the first caller runs the work, every caller that +/// arrives meanwhile awaits the same outcome. +actor SingleFlight { + private var inFlight: Task<Void, Error>? + + var isActive: Bool { inFlight != nil } + + func run(_ work: @Sendable @escaping () async throws -> Void) async throws { + if let inFlight { + try await inFlight.value + return + } + let task = Task { try await work() } + inFlight = task + defer { inFlight = nil } + try await task.value Review Comment: warning: `connect()` ignores caller cancellation and can wait forever with unlimited retries. let canceled callers exit promptly and cancel the shared connection task when no callers remain. ########## foreign/swift/Sources/Iggy/Client/ClientCore.swift: ########## @@ -0,0 +1,956 @@ +// 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. + +import Foundation +import Logging + +/// The connection state machine behind ``IggyClient``: one lockstep +/// connection, its consensus session, the credentials to restore it with, +/// and the cluster roster to fail over to. Modelled on the Rust `TcpClient` +/// and `leader_aware`. +/// +/// One deliberate difference: a reconnect registers a new session instead of +/// resuming the old one, so a replicated write whose outcome is unknown is +/// never replayed across connections. +actor ClientCore { + /// Backoff before replaying a request the server answered with an explicit + /// transient frame. The reply arrives promptly, so a short pause keeps the + /// replay from spinning while the primary catches up. + static let notReadyRetryInterval: Duration = .milliseconds(50) + /// How long a request replays "not accepted" on the same connection before + /// the client re-checks cluster leadership. + static let transientFailoverCheckInterval: Duration = .seconds(2) + /// Bound on one dial while other endpoints are queued behind it. + static let failoverDialTimeout: Duration = .seconds(2) + /// Bound on the roster read that follows a manual sign-in. + static let rosterReadTimeout: Duration = .seconds(5) + /// How long to wait for a transiently leaderless cluster to elect. + static let leaderlessWaitBudget: Duration = .seconds(5) + static let leaderlessPollInterval: Duration = .milliseconds(250) + static let maxLeaderRedirects = 3 + + let configuration: ClientConfiguration + let connector: any IggyConnector + let logger: Logger + + private(set) var state: ClientState = .disconnected + private var connection: (any IggyConnection)? + /// Counts the connections installed so far, so a request that failed on + /// an old one does not tear down the one that replaced it. + private var connectionEpoch: UInt64 = 0 + private var session = ConsensusSession() + private var currentAddress: String + private var rosterEndpoints: [String] = [] + private var rosterLearned = false + private var rememberedSignIn: (credentials: Credentials, userID: UInt32)? + /// The password a committed change gave the configured auto-login user; + /// the configuration cannot be rewritten and the old password is dead. + private var configuredPassword: String? + /// The name a committed rename gave the configured auto-login user. + private var configuredUsername: String? + private var connectedAt: ContinuousClock.Instant? + private var redirects = 0 + private var skipAutoLoginOnce = false + /// Set by an explicit ``disconnect()`` and cleared by the caller's next + /// ``connectByCaller()``: in between, nothing may bring the client back. + private var disconnectedByCaller = false + private var eventSubscribers: [UUID: AsyncStream<DiagnosticEvent>.Continuation] = [:] + private let exchangeLock = AsyncMutex() + private let routingLock = AsyncMutex() + private let connectGate = SingleFlight() + private var heartbeatTask: Task<Void, Never>? + + init(configuration: ClientConfiguration, connector: any IggyConnector, logger: Logger) { + self.configuration = configuration + self.connector = connector + self.logger = logger + currentAddress = configuration.address + } + + // MARK: State and events + + var clientAddress: String? { connection?.localAddress } + var serverAddress: String { currentAddress } + var isSignedIn: Bool { state == .authenticated } Review Comment: simplification: `isSignedIn` has no callers. remove the unused property. ########## foreign/swift/Sources/Iggy/Utilities/AsyncMutex.swift: ########## @@ -0,0 +1,63 @@ +// 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. + +/// A FIFO mutex that can be held across suspension points. +/// +/// Actors release their isolation at every `await`, so an actor method that +/// writes a request and then awaits the reply would let a second caller +/// interleave on the same connection. The lockstep exchange holds this +/// instead. +actor AsyncMutex { + private var locked = false + private var waiters: [CheckedContinuation<Void, Never>] = [] + + func lock() async { + if !locked { + locked = true + return + } + await withCheckedContinuation { waiters.append($0) } Review Comment: warning: canceled callers stay queued behind active requests or unlimited reconnect attempts. remove canceled waiters and resume them with `CancellationError` without releasing another caller's lock. ########## foreign/swift/Sources/Iggy/Client/ClientCore.swift: ########## @@ -0,0 +1,956 @@ +// 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. + +import Foundation +import Logging + +/// The connection state machine behind ``IggyClient``: one lockstep +/// connection, its consensus session, the credentials to restore it with, +/// and the cluster roster to fail over to. Modelled on the Rust `TcpClient` +/// and `leader_aware`. +/// +/// One deliberate difference: a reconnect registers a new session instead of +/// resuming the old one, so a replicated write whose outcome is unknown is +/// never replayed across connections. +actor ClientCore { + /// Backoff before replaying a request the server answered with an explicit + /// transient frame. The reply arrives promptly, so a short pause keeps the + /// replay from spinning while the primary catches up. + static let notReadyRetryInterval: Duration = .milliseconds(50) + /// How long a request replays "not accepted" on the same connection before + /// the client re-checks cluster leadership. + static let transientFailoverCheckInterval: Duration = .seconds(2) + /// Bound on one dial while other endpoints are queued behind it. + static let failoverDialTimeout: Duration = .seconds(2) + /// Bound on the roster read that follows a manual sign-in. + static let rosterReadTimeout: Duration = .seconds(5) + /// How long to wait for a transiently leaderless cluster to elect. + static let leaderlessWaitBudget: Duration = .seconds(5) + static let leaderlessPollInterval: Duration = .milliseconds(250) + static let maxLeaderRedirects = 3 + + let configuration: ClientConfiguration + let connector: any IggyConnector + let logger: Logger + + private(set) var state: ClientState = .disconnected + private var connection: (any IggyConnection)? + /// Counts the connections installed so far, so a request that failed on + /// an old one does not tear down the one that replaced it. + private var connectionEpoch: UInt64 = 0 + private var session = ConsensusSession() + private var currentAddress: String + private var rosterEndpoints: [String] = [] + private var rosterLearned = false + private var rememberedSignIn: (credentials: Credentials, userID: UInt32)? + /// The password a committed change gave the configured auto-login user; + /// the configuration cannot be rewritten and the old password is dead. + private var configuredPassword: String? + /// The name a committed rename gave the configured auto-login user. + private var configuredUsername: String? + private var connectedAt: ContinuousClock.Instant? + private var redirects = 0 + private var skipAutoLoginOnce = false + /// Set by an explicit ``disconnect()`` and cleared by the caller's next + /// ``connectByCaller()``: in between, nothing may bring the client back. + private var disconnectedByCaller = false + private var eventSubscribers: [UUID: AsyncStream<DiagnosticEvent>.Continuation] = [:] + private let exchangeLock = AsyncMutex() + private let routingLock = AsyncMutex() + private let connectGate = SingleFlight() + private var heartbeatTask: Task<Void, Never>? + + init(configuration: ClientConfiguration, connector: any IggyConnector, logger: Logger) { + self.configuration = configuration + self.connector = connector + self.logger = logger + currentAddress = configuration.address + } + + // MARK: State and events + + var clientAddress: String? { connection?.localAddress } + var serverAddress: String { currentAddress } + var isSignedIn: Bool { state == .authenticated } + + private func setState(_ newState: ClientState) { Review Comment: simplification: `setState()` only assigns the property and adds no invariant. assign `state` directly at its call sites. ########## foreign/swift/Sources/Iggy/Client/ClientCore.swift: ########## @@ -0,0 +1,956 @@ +// 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. + +import Foundation +import Logging + +/// The connection state machine behind ``IggyClient``: one lockstep +/// connection, its consensus session, the credentials to restore it with, +/// and the cluster roster to fail over to. Modelled on the Rust `TcpClient` +/// and `leader_aware`. +/// +/// One deliberate difference: a reconnect registers a new session instead of +/// resuming the old one, so a replicated write whose outcome is unknown is +/// never replayed across connections. +actor ClientCore { + /// Backoff before replaying a request the server answered with an explicit + /// transient frame. The reply arrives promptly, so a short pause keeps the + /// replay from spinning while the primary catches up. + static let notReadyRetryInterval: Duration = .milliseconds(50) + /// How long a request replays "not accepted" on the same connection before + /// the client re-checks cluster leadership. + static let transientFailoverCheckInterval: Duration = .seconds(2) + /// Bound on one dial while other endpoints are queued behind it. + static let failoverDialTimeout: Duration = .seconds(2) + /// Bound on the roster read that follows a manual sign-in. + static let rosterReadTimeout: Duration = .seconds(5) + /// How long to wait for a transiently leaderless cluster to elect. + static let leaderlessWaitBudget: Duration = .seconds(5) + static let leaderlessPollInterval: Duration = .milliseconds(250) + static let maxLeaderRedirects = 3 + + let configuration: ClientConfiguration + let connector: any IggyConnector + let logger: Logger + + private(set) var state: ClientState = .disconnected + private var connection: (any IggyConnection)? + /// Counts the connections installed so far, so a request that failed on + /// an old one does not tear down the one that replaced it. + private var connectionEpoch: UInt64 = 0 + private var session = ConsensusSession() + private var currentAddress: String + private var rosterEndpoints: [String] = [] + private var rosterLearned = false + private var rememberedSignIn: (credentials: Credentials, userID: UInt32)? + /// The password a committed change gave the configured auto-login user; + /// the configuration cannot be rewritten and the old password is dead. + private var configuredPassword: String? + /// The name a committed rename gave the configured auto-login user. + private var configuredUsername: String? + private var connectedAt: ContinuousClock.Instant? + private var redirects = 0 + private var skipAutoLoginOnce = false + /// Set by an explicit ``disconnect()`` and cleared by the caller's next + /// ``connectByCaller()``: in between, nothing may bring the client back. + private var disconnectedByCaller = false + private var eventSubscribers: [UUID: AsyncStream<DiagnosticEvent>.Continuation] = [:] + private let exchangeLock = AsyncMutex() + private let routingLock = AsyncMutex() + private let connectGate = SingleFlight() + private var heartbeatTask: Task<Void, Never>? + + init(configuration: ClientConfiguration, connector: any IggyConnector, logger: Logger) { + self.configuration = configuration + self.connector = connector + self.logger = logger + currentAddress = configuration.address + } + + // MARK: State and events + + var clientAddress: String? { connection?.localAddress } + var serverAddress: String { currentAddress } + var isSignedIn: Bool { state == .authenticated } + + private func setState(_ newState: ClientState) { + state = newState + } + + func subscribeEvents() -> AsyncStream<DiagnosticEvent> { + let id = UUID() + let (stream, continuation) = AsyncStream<DiagnosticEvent>.makeStream(bufferingPolicy: .bufferingNewest(256)) + eventSubscribers[id] = continuation + continuation.onTermination = { [weak self] _ in + guard let self else { return } + Task { await self.removeSubscriber(id) } + } + return stream + } + + private func removeSubscriber(_ id: UUID) { + eventSubscribers[id] = nil + } + + private func publish(_ event: DiagnosticEvent) { + for continuation in eventSubscribers.values { + continuation.yield(event) + } + } + + // MARK: Exchange + + /// Sends a request and returns its reply body, reconnecting and replaying + /// when the failure is one a fresh connection recovers from. + /// + /// A cancelled caller gets `CancellationError`. A request it already wrote + /// takes the connection with it, since its reply can no longer be told + /// from the next one, and the next caller reconnects. + func exchange(code: UInt32, payload: [UInt8]) async throws -> [UInt8] { + if disconnectedByCaller { + throw IggyError(state == .shutdown ? .clientShutdown : .notConnected) + } + var failedEpoch = connectionEpoch + do { + return try await sendRaw(code: code, payload: payload, epoch: &failedEpoch) + } catch let error as IggyError { + // A cancelled caller cannot use a new connection: every exchange + // it starts is abandoned at once. + try Task.checkCancellation() + guard error.isReconnectable else { + throw error + } + if code == CommandCode.getClusterMetadata.rawValue { + // An unauthenticated roster read cannot be repaired by a + // reconnect; it would re-issue the same read forever. + throw error + } + guard configuration.reconnection.enabled else { + throw IggyError(.disconnected, context: "automatic reconnection is disabled") + } + let isLogin = CommandCode.isRegister(code) + if !isLogin, signInCredentials() == nil { + // With no credentials a reconnect cannot restore the session, so + // anything but a sign-in fails instead of replaying + // unauthenticated. The sign-in itself is the exception: the + // server stays silent on a transient register failure and + // relies on the client replaying it. + throw error + } + if isLogin, await connectGate.isActive { + // The sign-in ran inside a connect sweep, which owns the + // endpoint walk; the sweep reports the failure itself. + throw error + } + // A replay over a new connection runs under a new session, and the + // server deduplicates by (client, request). Replay only what + // provably never reached the log. + let replay = replayAfterSessionResetIsSafe(code: code, error: error) + await routingLock.lock() + // Another caller may have reconnected while this one waited for + // the lock; its connection is not the one that failed. + if failedEpoch == connectionEpoch { + await disconnectTransport() + } + if isLogin { + skipAutoLoginOnce = true + } + logger.info("Reconnecting to the server", metadata: ["server_address": "\(currentAddress)", "error": "\(error)"]) + do { + try await connect() + } catch { + skipAutoLoginOnce = false + await routingLock.unlock() + throw error + } + // Released before the replay, which takes the lock itself when the + // new connection keeps refusing the request. + await routingLock.unlock() + guard replay else { + logger.warning( + "Reconnected, but the outcome of command \(code) is unknown: replaying it could apply it twice, so the original error is returned instead") + throw error + } + return try await sendRaw(code: code, payload: payload, epoch: &failedEpoch) + } + } + + /// Whether replaying `code` after reconnecting with a new session cannot + /// apply it twice. + private func replayAfterSessionResetIsSafe(code: UInt32, error: IggyError) -> Bool { + if CommandCode.isRegister(code) { + return true + } + switch error.code { + case .notConnected, .cannotEstablishConnection, .unauthenticated: + return true + default: + break + } + let operation = ConsensusOperation.forCode(code) + return operation == .nonReplicated || operation == .logout + } + + /// One request against the current connection, bounded by one deadline + /// across every same-connection replay and every leader failover. `epoch` + /// reports the connection the last attempt ran on. + private func sendRaw(code: UInt32, payload: [UInt8], epoch: inout UInt64) async throws -> [UInt8] { + switch state { + case .shutdown: + throw IggyError(.clientShutdown) + case .disconnected, .connecting: + throw IggyError(.notConnected) + default: + break + } + let clock = ContinuousClock() + let overallDeadline = clock.now + configuration.requestTimeout + let isLogin = CommandCode.isRegister(code) + var request: VSRFrame.EncodedRequest? + var visited = Set<String>() + var walkingRoster = false + var checkedLeader = false + while true { + let transientDeadline = isLogin ? overallDeadline : min(overallDeadline, clock.now + Self.transientFailoverCheckInterval) + let outcome = try await attempt( + code: code, payload: payload, request: &request, epoch: &epoch, transientDeadline: transientDeadline, readDeadline: overallDeadline) + switch outcome { + case .success(let body): + return body + case .failure(let error): + guard error.code == .transientNotAccepted, !isLogin, clock.now < overallDeadline else { + if error.code == .disconnected, epoch == connectionEpoch { + // The reply stream is at an unknown boundary; a late + // reply would desync framing for the next request. + connection = nil + setState(.disconnected) + publish(.disconnected) + } + throw error + } + if code == CommandCode.getClusterMetadata.rawValue { + throw error + } + // The server never admitted the request, so re-issuing it + // cannot double-apply: keep the stamped id for same-session + // replays; a redirect registers again, so the frame is stamped + // afresh. + await routingLock.lock() + var moved = false + do { + if !walkingRoster, !checkedLeader { + checkedLeader = true + visited.insert(currentAddress) + if try await handleLeaderRedirection() { + try await connect() + moved = true + } + } + if !moved, let next = nextRosterEndpoint(excluding: visited) { + // The roster names this node as the metadata leader, + // yet it keeps refusing: its replica of the target + // partition group is not that group's primary. Walk + // the roster instead of replaying into the same refusal. + visited.insert(next) + walkingRoster = true + try await settle(on: next) + try await connect(settleOffLeader: true) + moved = true + } + } catch { + await routingLock.unlock() + throw error + } + await routingLock.unlock() + if moved { + request = nil + } else if rosterEndpoints.count <= 1 { + // A single node: partition materialisation can outlast the + // short retry window, so keep replaying until the budget ends. + continue + } else { + throw error + } + } + } + } + + /// Holds the exchange lock for one request, so the lockstep stream sees + /// one request at a time. Throws only for a caller cancelled before its + /// turn, which leaves the connection untouched. + private func attempt( + code: UInt32, payload: [UInt8], request: inout VSRFrame.EncodedRequest?, epoch: inout UInt64, transientDeadline: ContinuousClock.Instant, + readDeadline: ContinuousClock.Instant + ) async throws -> Result<[UInt8], IggyError> { + await exchangeLock.lock() + if Task.isCancelled { + await exchangeLock.unlock() + throw CancellationError() + } + let outcome = await exchangeLocked( + code: code, payload: payload, request: &request, epoch: &epoch, transientDeadline: transientDeadline, readDeadline: readDeadline) + await exchangeLock.unlock() + return outcome + } + + /// Stamps the request if it is not stamped yet and exchanges it, replaying + /// in place while the server answers transiently. + private func exchangeLocked( + code: UInt32, payload: [UInt8], request: inout VSRFrame.EncodedRequest?, epoch: inout UInt64, transientDeadline: ContinuousClock.Instant, + readDeadline: ContinuousClock.Instant + ) async -> Result<[UInt8], IggyError> { + guard let connection else { + return .failure(IggyError(.notConnected)) + } + epoch = connectionEpoch + let stamped: VSRFrame.EncodedRequest + if let request { + stamped = request + } else { + do { + stamped = try VSRFrame.encodeRequest(session: &session, code: code, payload: payload) + } catch let error as IggyError { + return .failure(error) + } catch { + return .failure(IggyError(.invalidConfiguration, context: "\(error)")) + } + request = stamped + } + let clock = ContinuousClock() + while true { + let remaining = readDeadline - clock.now + guard remaining > .zero else { + await connection.close() + return .failure(IggyError(.disconnected, context: "timed out waiting for the reply to command \(code)")) + } + let raw: RawFrame + do { + raw = try await connection.exchange(stamped.bytes, timeout: remaining) + } catch let error as IggyError { + logger.error("Request failed", metadata: ["code": "\(code)", "error": "\(error)"]) + return .failure(error.code == .disconnected || error.code == .tcpError ? error : IggyError(.disconnected, context: "\(error)")) + } catch { + return .failure(IggyError(.disconnected, context: "\(error)")) + } + let header = raw.header[...] + if VSRFrame.peekCommand(header) == .reply, VSRFrame.readReplyRequestID(header) != stamped.requestID { + // A reply must answer the request in flight; an unexpected echo + // means every later reply would pair off by one. + await connection.close() + let answered = VSRFrame.readReplyRequestID(header).map(String.init) ?? "unknown" + return .failure(IggyError(.disconnected, context: "the reply answers request \(answered), expected \(stamped.requestID)")) + } + do { + return .success(Array(try VSRFrame.decodeReply(header: header, body: raw.body[...]))) + } catch let error as IggyError { + switch error.code { + case .transientNotCommitted where clock.now < readDeadline: + // The outcome is unknown; only a same-session replay of the + // same request id is safe, which the client table answers + // from its reply cache if it did commit. + try? await Task.sleep(for: min(Self.notReadyRetryInterval, readDeadline - clock.now)) + case .transientNotAccepted where clock.now < transientDeadline: + try? await Task.sleep(for: min(Self.notReadyRetryInterval, transientDeadline - clock.now)) + default: + if error.code == .staleClient || error.code == .unauthenticated, VSRFrame.peekCommand(header) == .eviction { + // A session-terminal eviction: the connection is done. + await connection.close() + } + return .failure(error) + } + } catch { + return .failure(IggyError(.invalidCommand, context: "\(error)")) + } + } + } + + // MARK: Connecting + + /// The connect a caller asked for, which also lifts an earlier explicit + /// disconnect. + func connectByCaller() async throws { + disconnectedByCaller = false + try await connect() + } + + func connect(settleOffLeader: Bool = false) async throws { + try await connectGate.run { [self] in + try await self.connectInner(settleOffLeader: settleOffLeader) + } + } + + private func connectInner(settleOffLeader: Bool) async throws { + while true { + switch state { + case .shutdown: + throw IggyError(.clientShutdown) + case .connected, .authenticating, .authenticated: + return + case .connecting: + return + case .disconnected: + try checkNotDisconnectedByCaller() + setState(.connecting) + } + + var candidates = dialCandidates() + let pacedEndpoint = currentAddress + if candidates.count > 1, reestablishWait() != nil { + // The paced endpoint goes last; the others owe no cooldown. + candidates.append(candidates.removeFirst()) + } + let skipAutoLogin = skipAutoLoginOnce + skipAutoLoginOnce = false + + var retries: UInt32 = 0 + var candidateIndex = 0 + var configurationFault: IggyError? + var signInFailure: IggyError? + var dialFailure: IggyError? + var shouldRedirect = false + sweep: while true { + try abandonSweepIfDisconnectedByCaller() + let address = candidates[candidateIndex] + if address == pacedEndpoint, let remaining = reestablishWait() { + logger.info("Trying to connect to the server \(address) in \(remaining)") + try await Task.sleep(for: remaining) + } + try abandonSweepIfDisconnectedByCaller() + logger.info("Connecting to server \(address)") + do { + let established = try await establishBounded(address, candidates: candidates) + if disconnectedByCaller { + // The caller disconnected while this was being dialed; + // the check at the top of the sweep ends it. + await established.close() + continue sweep + } + currentAddress = address + connection = established + connectionEpoch += 1 + setState(.connected) + connectedAt = ContinuousClock().now + publish(.connected) + logger.info( + "Connected to server", metadata: ["server_address": "\(address)", "client_address": "\(established.localAddress ?? "unknown")"]) + do { + shouldRedirect = try await establishSession(skipAutoLogin: skipAutoLogin, settleOffLeader: settleOffLeader) + break sweep + } catch let failure as SignInFailure { + if failure.connectionLost { + logger.warning("The sign-in on \(address) did not complete: \(failure.error)") + signInFailure = failure.error + setState(.connecting) + } else { + // The connection stands and only the session is + // missing; no other endpoint would answer differently. + throw failure.error + } + } + } catch let error as IggyError where error.code == .cannotEstablishConnection { + logger.warning("Cannot connect to \(address): \(error)") + dialFailure = error + } catch let error as IggyError { + configurationFault = error + } + + candidateIndex += 1 + if candidateIndex < candidates.count { + continue + } + candidateIndex = 0 + + if let configurationFault { + failConnect() + throw configurationFault + } + guard configuration.reconnection.enabled else { + logger.warning("Automatic reconnection is disabled") + failConnect() + throw signInFailure ?? dialFailure ?? IggyError(.cannotEstablishConnection, context: currentAddress) + } + let maxRetries = configuration.reconnection.maxRetries + if maxRetries.map({ retries < $0 }) ?? true { + retries += 1 + let limit = maxRetries.map(String.init) ?? "unlimited" + logger.info("Retrying to connect (\(retries)/\(limit)), \(candidates.count) endpoint(s) in \(configuration.reconnection.interval)") + try await Task.sleep(for: configuration.reconnection.interval) + if state == .shutdown { + throw IggyError(.clientShutdown) + } + continue + } + failConnect() + throw signInFailure ?? dialFailure ?? IggyError(.cannotEstablishConnection, context: currentAddress) + } + if shouldRedirect { + continue + } + return + } + } + + private struct SignInFailure: Error { + let error: IggyError + let connectionLost: Bool + } + + /// Re-establishes the session on a connection that just came up and + /// settles it on the leader. Reports whether the leader check asks for a + /// redirect. + private func establishSession(skipAutoLogin: Bool, settleOffLeader: Bool) async throws -> Bool { + guard let credentials = signInCredentials() else { + logger.info("No credentials to sign in with") + return false + } + if skipAutoLogin { + logger.info("Skipping automatic sign-in for a retried login request") + return false + } + setState(.authenticating) + do { + switch credentials { + case .usernamePassword(let username, let password): + _ = try await login(username: username, password: password) + case .personalAccessToken(let token): + _ = try await login(personalAccessToken: token) + } + logger.info("Signed in with \(credentials)") + } catch let error as IggyError { + throw await failSignIn(error) + } + if settleOffLeader { + // A failover walking the roster past the metadata leader stays + // where it dialed; the leader settlement would put it straight + // back on the node whose partition replica keeps refusing. + return false + } + do { + return try await handleLeaderRedirection() + } catch let error as IggyError { + throw SignInFailure(error: error, connectionLost: false) + } + } + + private func failSignIn(_ error: IggyError) async -> SignInFailure { + let connectionLost = error.isConnectionLoss + if connectionLost { + await disconnectTransport() + } else if state == .authenticating { + setState(.connected) + } + if error.isCredentialRejection { + // A rejected credential does not become valid on the next reconnect + // and replaying it costs an argon2 on the server every time. + rememberedSignIn = nil + } + return SignInFailure(error: error, connectionLost: connectionLost) + } + + private func failConnect() { + setState(.disconnected) + publish(.disconnected) + } + + private func checkNotDisconnectedByCaller() throws { + if disconnectedByCaller { + throw IggyError(.notConnected, context: "the client was disconnected by its caller") + } + } + + /// A sweep checks this before each dial and before each retry pause: once + /// the caller disconnects, the sweep must not bring the client back. + private func abandonSweepIfDisconnectedByCaller() throws { + if disconnectedByCaller { + failConnect() + try checkNotDisconnectedByCaller() + } + } + + /// Endpoints to dial for one connect, likeliest first: where the client + /// currently is, the configured address, then the roster it learned. + private func dialCandidates() -> [String] { + var candidates = [currentAddress] + for endpoint in [configuration.address] + rosterEndpoints where !candidates.contains(where: { isSameSpelling($0, endpoint) }) { + candidates.append(endpoint) + } + return candidates + } + + private func nextRosterEndpoint(excluding visited: Set<String>) -> String? { + rosterEndpoints.first { candidate in !visited.contains { isSameSpelling($0, candidate) } } + } + + private func establishBounded(_ address: String, candidates: [String]) async throws -> any IggyConnection { + let endpoint = try Endpoint(parsing: address) + let timeout = candidates.count > 1 ? min(Self.failoverDialTimeout, configuration.connectTimeout) : configuration.connectTimeout + return try await connector.connect(to: endpoint, tls: configuration.tls, timeout: timeout, noDelay: configuration.noDelay) + } + + private func reestablishWait() -> Duration? { + guard let connectedAt else { + return nil + } + let elapsed = ContinuousClock().now - connectedAt + let interval = configuration.reconnection.reestablishAfter + return elapsed < interval ? interval - elapsed : nil + } + + /// Tears the connection down without touching the remembered sign-in, so + /// a reconnect can restore the session. + func disconnectTransport() async { + switch state { + case .disconnected: + return + case .connecting: + // A connect is already sweeping; tearing down under it would take + // the stream it just installed. + return + default: + break + } + logger.info("Disconnecting from server \(currentAddress)") + setState(.disconnected) Review Comment: critical: `disconnect()` or an unfinished dial can overwrite `.shutdown` and reopen the client. preserve the terminal state during teardown and reject connections that complete after shutdown. -- 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]
