RustToMetal commented on code in PR #4503: URL: https://github.com/apache/iggy/pull/4503#discussion_r4261344793
########## 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: Done, one deadline bounds the whole dial and the handshake wait gets what is left of it, with a test against a peer that accepts and never answers. ########## 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: Done, the sweep checks for shutdown before each dial and after it: a connection that completes late is closed before it is installed and the sweep throws clientShutdown. ########## 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: Done, the shared flag is gone. The skip is a parameter of the retried login's own connect call. ########## 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: Done, both dropped throws. ########## 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: Done, the doc says the count applies after the first round and that zero still makes one. ########## 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: Done, removed. ########## 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: Done, all three are gone. ########## 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: Done, the flag is gone. The walk keeps only checkedLeader and the visited endpoints. -- 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]
