github-actions[bot] commented on code in PR #4503: URL: https://github.com/apache/iggy/pull/4503#discussion_r4256816981
########## foreign/swift/Sources/Iggy/Transport/TCPConnection.swift: ########## @@ -0,0 +1,279 @@ +// 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 NIOCore +import NIOPosix +import NIOSSL +import NIOTLS + +/// One complete consensus frame as read off the wire. +struct RawFrame: Sendable { + let header: [UInt8] + let body: [UInt8] +} + +/// A transport that carries consensus frames to one endpoint. +/// +/// The protocol is lockstep: one request is written, exactly one reply is +/// read. The client serializes callers; the connection only moves bytes. +protocol IggyConnection: AnyObject, Sendable { + var localAddress: String? { get } + var remoteAddress: String? { get } + /// Writes a frame and reads the reply, bounded by `timeout`. A timeout + /// leaves the stream at an unknown boundary, so the connection closes + /// itself and every later call fails. + func exchange(_ frame: [UInt8], timeout: Duration) async throws -> RawFrame + func close() async +} + +/// A dialer that produces connections; swapped out by tests. +protocol IggyConnector: Sendable { + func connect(to endpoint: Endpoint, tls: TLSOptions?, timeout: Duration, noDelay: Bool) async throws -> any IggyConnection +} + +/// The NIO-backed dialer used in production. +struct NIOConnector: IggyConnector { + func connect(to endpoint: Endpoint, tls: TLSOptions?, timeout: Duration, noDelay: Bool) async throws -> any IggyConnection { + try await TCPConnection.connect(to: endpoint, tls: tls, timeout: timeout, noDelay: noDelay) + } +} + +/// A TCP connection, optionally wrapped in TLS, speaking the consensus +/// framing. Backed by SwiftNIO so the same code runs on Apple platforms and +/// Linux. +final class TCPConnection: IggyConnection, @unchecked Sendable { + private let channel: Channel + private let handler: ExchangeHandler + let localAddress: String? + let remoteAddress: String? + + private init(channel: Channel, handler: ExchangeHandler) { + self.channel = channel + self.handler = handler + localAddress = channel.localAddress.map(Self.describe) + remoteAddress = channel.remoteAddress.map(Self.describe) + } + + private static func describe(_ address: SocketAddress) -> String { + switch address { + case .v4(let v4): "\(v4.host):\(address.port ?? 0)" + case .v6(let v6): "[\(v6.host)]:\(address.port ?? 0)" + case .unixDomainSocket: address.description + } + } + + static func connect(to endpoint: Endpoint, tls: TLSOptions?, timeout: Duration, noDelay: Bool) async throws -> TCPConnection { + let sslContext: NIOSSLContext? + if let tls { + sslContext = try tls.makeContext() + } else { + sslContext = nil + } + let handler = ExchangeHandler() + // TCP_NODELAY lives at the TCP level; the socket-level shorthand would + // set SO_DEBUG instead, which Linux refuses without CAP_NET_ADMIN. + let noDelayValue: ChannelOptions.Types.SocketOption.Value = noDelay ? 1 : 0 + let bootstrap = ClientBootstrap(group: MultiThreadedEventLoopGroup.singleton) + .channelOption(.tcpOption(.tcp_nodelay), value: noDelayValue) + .connectTimeout(.nanoseconds(timeout.nanosecondsClamped)) + .channelInitializer { channel in + channel.eventLoop.makeCompletedFuture { + if let sslContext { + let serverHostname = tls?.serverName(for: endpoint) + let sslHandler = try NIOSSLClientHandler(context: sslContext, serverHostname: serverHostname) + try channel.pipeline.syncOperations.addHandler(sslHandler) + } + try channel.pipeline.syncOperations.addHandler(ByteToMessageHandler(FrameDecoder())) + try channel.pipeline.syncOperations.addHandler(handler) + } + } + let channel: Channel + do { + channel = try await bootstrap.connect(host: endpoint.host, port: endpoint.port).get() + } catch { + throw IggyError(.cannotEstablishConnection, context: "\(endpoint): \(error)") + } + if sslContext != nil { + // The handshake completes asynchronously after connect; waiting for + // it here means a failed handshake is reported by the dial rather + // than by the first request. + do { + try await handler.handshakeDone(on: channel) Review Comment: warning: `connectTimeout` reaches only `bootstrap.connectTimeout`, so the TLS handshake wait has no deadline, though `ClientConfiguration.connectTimeout` documents the bound as covering the handshake. A peer that accepts TCP and stalls the handshake leaves `connect()` suspended on `handshakeDone` forever. Bound the whole dial with one deadline, the handshake included. ########## 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 { Review Comment: critical: `sendRaw()` treats every `transientNotAccepted` as proof that the server never admitted the request, even after the same attempt saw `transientNotCommitted` for that request id. The failover re-issues it under a fresh session, which the dedup fence cannot match, so the write applies twice. Carry the uncertainty across attempts and answer later refusals with `transientNotCommitted`, as `RetryOutcome` does in `core/sdk/src/vsr.rs`. ########## foreign/swift/Sources/Iggy/Client/IggyClient.swift: ########## @@ -0,0 +1,242 @@ +// 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 Logging + +/// A client for an Apache Iggy server over TCP, with or without TLS. +/// +/// The client speaks the consensus wire protocol, the only protocol the +/// server accepts. One client owns one connection and is safe to share +/// between tasks: requests are serialized on the connection, so a single +/// client feeds many producers and consumers. +/// +/// ```swift +/// let client = IggyClient(address: "localhost:8090", autoLogin: .usernamePassword(username: "iggy", password: "iggy")) +/// try await client.connect() +/// let stream = try await client.createStream(name: "orders") +/// let topic = try await client.createTopic(streamID: "orders", name: "created", options: .init(partitionsCount: 3)) +/// try await client.shutdown() +/// ``` +/// +/// Lifecycle: construct, ``connect()``, use, ``shutdown()``. With +/// ``ClientConfiguration/autoLogin`` set, `connect()` also signs in and every +/// reconnect restores the session; otherwise call ``login(username:password:)`` +/// after connecting. ``disconnect()`` closes the connection but keeps the +/// client usable; a client that was shut down cannot reconnect. +/// +/// Every operation throws ``IggyError``, or `CancellationError` when its +/// task is cancelled. A request that hits a dropped connection is replayed +/// over a fresh one only when it provably never reached the log, so a +/// replicated write whose outcome is unknown surfaces the error instead. +public final class IggyClient: Sendable { + /// The configuration the client was created with. + public let configuration: ClientConfiguration + let core: ClientCore + let logger: Logger + + /// Creates a client that dials `configuration.address` on ``connect()``. + public convenience init(configuration: ClientConfiguration = ClientConfiguration(), logger: Logger = Logger(label: "org.apache.iggy")) { + self.init(configuration: configuration, connector: NIOConnector(), logger: logger) + } + + /// Creates a client for `address` (`host:port`). + public convenience init(address: String, autoLogin: Credentials? = nil, tls: TLSOptions? = nil) { + self.init(configuration: ClientConfiguration(address: address, tls: tls, autoLogin: autoLogin)) + } + + init(configuration: ClientConfiguration, connector: any IggyConnector, logger: Logger) { + self.configuration = configuration + self.logger = logger + core = ClientCore(configuration: configuration, connector: connector, logger: logger) + } + + // MARK: Lifecycle + + /// Connects to the server, signing in when auto-login is configured and + /// settling on the cluster leader. Does nothing when already connected. + public func connect() async throws { + try await core.connectByCaller() + await core.startHeartbeat { [weak self] in + await self?.heartbeat() + } + } + + /// Closes the connection and forgets the session. Requests fail with + /// ``IggyErrorCode/notConnected`` until ``connect()`` is called again; no + /// reconnect brings the client back on its own. + public func disconnect() async throws { Review Comment: nit: `disconnect()` and `shutdown()` declare `async throws`, but the core methods they call do not throw. Every caller writes `try` for an error that never comes. Drop `throws` from both while the API is new. ########## 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 reader in the package, and `state` already exposes the same test. Delete it. ########## foreign/swift/Sources/Iggy/Client/ClientConfiguration.swift: ########## @@ -0,0 +1,141 @@ +// 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. + +/// Credentials a client signs in with. +public enum Credentials: Sendable, Hashable { + case usernamePassword(username: String, password: String) + case personalAccessToken(String) +} + +extension Credentials: CustomStringConvertible { + /// Never includes the secret. + public var description: String { + switch self { + case .usernamePassword(let username, _): "username \(username)" + case .personalAccessToken: "a personal access token" + } + } +} + +/// How the client reconnects after losing its connection. +public struct ReconnectionOptions: Sendable, Hashable { + /// Whether to reconnect at all. Off, a dropped connection fails the + /// request that hit it and every later one until `connect()` is called. + public var enabled: Bool + /// How many rounds over the known endpoints to attempt; nil means forever. Review Comment: nit: The doc calls `maxRetries` the rounds to attempt, but the loop counts only the rounds after the first pass. A value of 0 still makes one pass, so say that the count applies after the first pass. ########## 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 { Review Comment: warning: A `shutdown()` during `establishBounded` goes unseen: line 436 tests only `disconnectedByCaller`, so the sweep installs the connection and sets `.connected` over `.shutdown`. The retry path checks `state == .shutdown` at line 492, but this path does not, so the client revives. Close the established connection and throw `.clientShutdown` when the state is `.shutdown`. ########## foreign/swift/Sources/Iggy/Transport/TCPConnection.swift: ########## @@ -0,0 +1,279 @@ +// 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 NIOCore +import NIOPosix +import NIOSSL +import NIOTLS + +/// One complete consensus frame as read off the wire. +struct RawFrame: Sendable { + let header: [UInt8] + let body: [UInt8] +} + +/// A transport that carries consensus frames to one endpoint. +/// +/// The protocol is lockstep: one request is written, exactly one reply is +/// read. The client serializes callers; the connection only moves bytes. +protocol IggyConnection: AnyObject, Sendable { + var localAddress: String? { get } + var remoteAddress: String? { get } Review Comment: simplification: `remoteAddress` has no reader: only the requirement at line 36, the stored property at line 63 and the assignment at line 69 mention it. Drop all three and keep `localAddress`. ########## 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 { Review Comment: simplification: `walkingRoster` becomes true only on a pass that already set `checkedLeader` at line 256, so `!walkingRoster` cannot change this guard. Keep `!checkedLeader` alone. ########## 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 Review Comment: warning: One caller's login replay sets `skipAutoLoginOnce`, but any other connect sweep can consume it, so a later reconnect skips auto-login and connects unauthenticated. Pass the skip intent into that replay's own connect call. ########## 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) + let old = connection + connection = nil + // A Register still in flight is retried on the next connection and + // must arrive under the identity and secret its body was built from. + if session.isBound { + session.reset() + } + await old?.close() + publish(.disconnected) + } + + /// An explicit disconnect ends the session on purpose: the remembered + /// sign-in goes with it so the next reconnect does not resurrect it. + func disconnect() async { + disconnectedByCaller = true + rememberedSignIn = nil + stopHeartbeat() + await disconnectTransport() + } + + func shutdown() async { + guard state != .shutdown else { + return + } + stopHeartbeat() + let old = connection + connection = nil + session.reset() + setState(.shutdown) + await old?.close() + publish(.shutdown) + for continuation in eventSubscribers.values { + continuation.finish() + } + eventSubscribers.removeAll() + } + + // MARK: Session + + /// Credentials to sign in with after connecting: the ones a sign-in on this + /// client last succeeded with, else the configured ones with any committed + /// password change applied. + private func signInCredentials() -> Credentials? { + if let rememberedSignIn { + return rememberedSignIn.credentials + } + switch configuration.autoLogin { + case .usernamePassword(let username, let password): + return .usernamePassword(username: configuredUsername ?? username, password: configuredPassword ?? password) + case .personalAccessToken(let token): + return .personalAccessToken(token) + case nil: + return nil + } + } + + func login(username: String, password: String) async throws -> IdentityInfo { + try Requests.validateUsername(username) + try Requests.validatePassword(password) + let response = try await register(code: CommandCode.loginRegister.rawValue) { bindSecret in + try Requests.loginRegister(username: username, password: password, bindSecret: bindSecret) + } + rememberSignIn(.usernamePassword(username: username, password: password), userID: response.userID) + await learnRosterOnce() + return IdentityInfo(userID: response.userID) + } + + func login(personalAccessToken token: String) async throws -> IdentityInfo { + try Requests.validatePersonalAccessToken(token) + let response = try await register(code: CommandCode.loginRegisterWithPersonalAccessToken.rawValue) { bindSecret in + try Requests.loginRegisterWithPersonalAccessToken(token: token, bindSecret: bindSecret) + } + rememberSignIn(.personalAccessToken(token), userID: response.userID) + await learnRosterOnce() + return IdentityInfo(userID: response.userID) + } + + private func register(code: UInt32, payload makePayload: (_ bindSecret: [UInt8]) throws -> [UInt8]) async throws -> Responses.LoginRegister { + // The server binds a connection to one session and a re-login on a + // bound connection is served as an idempotent replay of the old + // identity, so switching users needs a logout first. + if state == .authenticated { + try await logout() + } + // A connection lost under a bound session leaves that session behind. + // The sign-in on the next one registers a new session, not a replay. + if session.isBound { + session.reset() + } + // The login body carries the bind secret of the session the frame + // encoder registers, and both survive a retry on a new connection. + let payload = try makePayload(session.bindSecret) + let body: [UInt8] + do { + body = try await exchange(code: code, payload: payload) + } catch { + session.reset() + throw error + } + let response: Responses.LoginRegister + do { + response = try Responses.loginRegister(body[...]) + try session.bind(response.session) + } catch { + session.reset() + throw error + } + logger.debug( + "Authenticated against iggy server", + metadata: ["server_version": "\(response.serverVersion)", "server_protocol_version": "\(response.serverProtocolVersion)"]) + setState(.authenticated) + publish(.signedIn) + return response + } + + func logout() async throws { + guard state == .authenticated else { + throw IggyError(.unauthenticated) + } + _ = try await exchange(code: CommandCode.logoutUser.rawValue, payload: []) + rememberedSignIn = nil + session.reset() + setState(.connected) + publish(.signedOut) + } + + private func rememberSignIn(_ credentials: Credentials, userID: UInt32) { + rememberedSignIn = (credentials, userID) + } + + /// A committed password change for `user`: when it is the signed-in user + /// or the configured auto-login user, the credentials the next reconnect + /// signs in with switch to the new password. + func refreshSessionPassword(user: Identifier, newPassword: String) { + var rememberedUsername: (name: String, isSessionUser: Bool)? + if let remembered = rememberedSignIn, case .usernamePassword(let username, _) = remembered.credentials { + let targetsSessionUser: Bool + switch user.kind { + case .numeric(let id): targetsSessionUser = id == remembered.userID + case .named(let name): targetsSessionUser = name == username + } + if targetsSessionUser { + rememberedSignIn = (.usernamePassword(username: username, password: newPassword), remembered.userID) + } + rememberedUsername = (username, targetsSessionUser) + } + guard case .usernamePassword(let configuredAtStart, _) = configuration.autoLogin else { + return + } + let configured = configuredUsername ?? configuredAtStart + let targetsConfiguredUser: Bool + switch user.kind { + case .named(let name): targetsConfiguredUser = name == configured + case .numeric: targetsConfiguredUser = rememberedUsername.map { $0.isSessionUser && $0.name == configured } ?? false + } + if targetsConfiguredUser { + configuredPassword = newPassword + } + } + + /// A committed rename of `user`: the credentials the next reconnect signs + /// in with follow the new name when it is the signed-in user or the + /// configured auto-login user, so a rename does not strand the session. + func refreshSessionUsername(user: Identifier, newUsername: String) { + var renamed: String? + if let remembered = rememberedSignIn, case .usernamePassword(let username, let password) = remembered.credentials { + let targetsSessionUser: Bool + switch user.kind { + case .numeric(let id): targetsSessionUser = id == remembered.userID + case .named(let name): targetsSessionUser = name == username + } + if targetsSessionUser { + rememberedSignIn = (.usernamePassword(username: newUsername, password: password), remembered.userID) + renamed = username + } + } + guard case .usernamePassword(let configuredAtStart, _) = configuration.autoLogin else { + return + } + let configured = configuredUsername ?? configuredAtStart + if renamed == configured || user.name == configured { + configuredUsername = newUsername + } + } + + // MARK: Cluster awareness + + /// Reads the roster once, on the first sign-in a caller ran by hand: a + /// client with configured auto-login learns it from the connect flow's + /// leader check instead. + private func learnRosterOnce() async { + guard configuration.autoLogin == nil, state == .authenticated, !rosterLearned else { + return + } + rosterLearned = true + let endpoints = + await withTimeout(Self.rosterReadTimeout) { [self] in + await self.readTransportEndpoints() + } ?? [] + if !endpoints.isEmpty { + logger.info("Learned \(endpoints.count) endpoint(s) to fail over to") + rosterEndpoints = endpoints + } + } + + private func readTransportEndpoints() async -> [String] { + do { + let body = try await exchange(code: CommandCode.getClusterMetadata.rawValue, payload: []) + return Self.transportEndpoints(try Responses.clusterMetadata(body[...])) + } catch { + logger.debug("Failed to read the cluster roster: \(error)") + return [] + } + } + + static func transportEndpoints(_ metadata: ClusterMetadata) -> [String] { + metadata.nodes.filter { $0.endpoints.tcp != 0 }.map(\.tcpAddress) + } + + /// Checks cluster metadata and moves off a non-leader node. Returns true + /// when the client disconnected and the caller must reconnect. + func handleLeaderRedirection() async throws -> Bool { + let check = await checkLeader() + if !check.endpoints.isEmpty { + rosterEndpoints = check.endpoints + } + guard let leader = check.redirect else { + redirects = 0 + return false + } + guard redirects < Self.maxLeaderRedirects else { + logger.warning("Maximum leader redirections reached, continuing with the current connection") + return false + } + redirects += 1 + logger.info("Current node is not the leader, redirecting to \(leader)") + connectedAt = nil + await disconnectTransport() + currentAddress = leader + return true + } + + private struct LeaderCheck { + var redirect: String? + var endpoints: [String] + } + + private func checkLeader() async -> LeaderCheck { + let clock = ContinuousClock() + let deadline = clock.now + Self.leaderlessWaitBudget + while true { + let metadata: ClusterMetadata + do { + let body = try await exchange(code: CommandCode.getClusterMetadata.rawValue, payload: []) + metadata = try Responses.clusterMetadata(body[...]) + } catch { + logger.debug("Failed to get cluster metadata: \(error), continuing on \(currentAddress)") + return LeaderCheck(redirect: nil, endpoints: []) + } + let endpoints = Self.transportEndpoints(metadata) + if metadata.nodes.count <= 1 { + return LeaderCheck(redirect: nil, endpoints: endpoints) + } + guard let leader = metadata.nodes.first(where: { $0.role == .leader && $0.status == .healthy }) else { + if clock.now >= deadline { + logger.warning("No active leader in the cluster roster, continuing on \(currentAddress)") + return LeaderCheck(redirect: nil, endpoints: endpoints) + } + try? await Task.sleep(for: Self.leaderlessPollInterval) + continue + } + let leaderAddress = leader.tcpAddress + if isSameSpelling(currentAddress, leaderAddress) { + return LeaderCheck(redirect: nil, endpoints: endpoints) + } + if await isSameAddress(currentAddress, leaderAddress) { + return LeaderCheck(redirect: nil, endpoints: endpoints) + } + return LeaderCheck(redirect: leaderAddress, endpoints: endpoints) + } + } + + /// Moves the connection to another roster endpoint for a request the + /// current node keeps refusing to admit. + private func settle(on next: String) async throws { + logger.info("The request keeps being refused on \(currentAddress); trying the next cluster node at \(next)") + connectedAt = nil + await disconnectTransport() + currentAddress = next + } + + // MARK: Heartbeat + + func startHeartbeat(_ body: @escaping @Sendable () async -> Void) { + stopHeartbeat() + let interval = configuration.heartbeatInterval + guard interval > .zero else { + return + } + heartbeatTask = Task { + while !Task.isCancelled { Review Comment: warning: `startHeartbeat()` loops on `Task.isCancelled`, and nothing cancels it when the caller drops the client, because the package has no `deinit`. An abandoned client leaves a task that wakes every `heartbeatInterval` forever and calls nothing, because the body holds the client weakly. End the loop when the weak reference is nil. -- 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]
