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