numinnex commented on code in PR #4503: URL: https://github.com/apache/iggy/pull/4503#discussion_r4273275633
########## foreign/swift/Sources/Iggy/Client/IggyClient.swift: ########## @@ -0,0 +1,246 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +import Logging + +/// A client for an Apache Iggy server over TCP, with or without TLS. +/// +/// The client speaks the consensus wire protocol, the only protocol the +/// server accepts. One client owns one connection and is safe to share +/// between tasks: requests are serialized on the connection, so a single +/// client feeds many producers and consumers. +/// +/// ```swift +/// let client = IggyClient(address: "localhost:8090", autoLogin: .usernamePassword(username: "iggy", password: "iggy")) +/// try await client.connect() +/// let stream = try await client.createStream(name: "orders") +/// let topic = try await client.createTopic(streamID: "orders", name: "created", options: .init(partitionsCount: 3)) +/// try await client.shutdown() +/// ``` +/// +/// Lifecycle: construct, ``connect()``, use, ``shutdown()``. With +/// ``ClientConfiguration/autoLogin`` set, `connect()` also signs in and every +/// reconnect restores the session; otherwise call ``login(username:password:)`` +/// after connecting. ``disconnect()`` closes the connection but keeps the +/// client usable; a client that was shut down cannot reconnect. +/// +/// Every operation throws ``IggyError``, or `CancellationError` when its +/// task is cancelled. A request that hits a dropped connection is replayed +/// over a fresh one only when it provably never reached the log, so a +/// replicated write whose outcome is unknown surfaces the error instead. +public final class IggyClient: Sendable { + /// The configuration the client was created with. + public let configuration: ClientConfiguration + let core: ClientCore + let logger: Logger + + /// Creates a client that dials `configuration.address` on ``connect()``. + public convenience init(configuration: ClientConfiguration = ClientConfiguration(), logger: Logger = Logger(label: "org.apache.iggy")) { + self.init(configuration: configuration, connector: NIOConnector(), logger: logger) + } + + /// Creates a client for `address` (`host:port`). + public convenience init(address: String, autoLogin: Credentials? = nil, tls: TLSOptions? = nil) { + self.init(configuration: ClientConfiguration(address: address, tls: tls, autoLogin: autoLogin)) + } + + init(configuration: ClientConfiguration, connector: any IggyConnector, logger: Logger) { + self.configuration = configuration + self.logger = logger + core = ClientCore(configuration: configuration, connector: connector, logger: logger) + } + + // MARK: Lifecycle + + /// Connects to the server, signing in when auto-login is configured and + /// settling on the cluster leader. Does nothing when already connected. + public func connect() async throws { + try await core.connectByCaller() + await core.startHeartbeat { [weak self] in + guard let self else { + return false + } + await self.heartbeat() + return true + } + } + + /// Closes the connection and forgets the session. Requests fail with + /// ``IggyErrorCode/notConnected`` until ``connect()`` is called again; no + /// reconnect brings the client back on its own. + public func disconnect() async { + await core.disconnect() + } + + /// Closes the connection and releases every resource. The client cannot + /// be used afterwards. + public func shutdown() async { + await core.shutdown() + } + + /// Where the client is in its lifecycle. + public var state: ClientState { + get async { await core.state } + } + + /// A new subscription to lifecycle events. Each access creates an + /// independent stream that ends when the client shuts down. + public var events: AsyncStream<DiagnosticEvent> { + get async { await core.subscribeEvents() } + } + + /// Local address of the current connection. + public var clientAddress: String? { + get async { await core.clientAddress } + } + + /// The endpoint the client is connected to, or will dial next. + public var serverAddress: String { + get async { await core.serverAddress } + } + + private func heartbeat() async { + guard await core.state == .authenticated else { + return + } + do { + try await ping() + } catch { + logger.debug("Heartbeat ping failed: \(error)") + } + } + + // MARK: Authentication + + /// Signs in with a username and password. Any existing session on this + /// connection is ended first. Remembers the credentials so a reconnect + /// restores the session. + public func login(username: String, password: String) async throws -> IdentityInfo { + let identity = try await core.login(username: username, password: password) + try await settleOnLeaderAfterLogin() + return identity + } + + /// Signs in with a personal access token. + public func login(personalAccessToken token: String) async throws -> IdentityInfo { + let identity = try await core.login(personalAccessToken: token) + try await settleOnLeaderAfterLogin() + return identity + } + + /// Ends the session. The next reconnect stays signed out unless + /// auto-login is configured. + public func logout() async throws { + try await core.logout() + } + + /// After a manual sign-in the roster is consulted once so later + /// replicated writes land on the leader. The reconnect signs in again + /// with the credentials just remembered. + private func settleOnLeaderAfterLogin() async throws { + if try await core.handleLeaderRedirection() { + logger.info("Redirected to the leader, reconnecting") + try await core.connect() + } + } + + // MARK: Raw requests + + /// Sends a command code with an already-encoded payload and returns the + /// raw reply body. For commands without a typed method; the caller owns + /// the wire format. Session-control codes are rejected because they must + /// go through ``login(username:password:)`` and ``logout()``. + public func sendRawRequest(code: UInt32, payload: [UInt8]) async throws -> [UInt8] { + guard !CommandCode.isSessionControl(code) else { + throw IggyError(.invalidCommand, context: "code \(code) controls the session; use login and logout") + } + return try await core.exchange(code: code, payload: payload) + } + + // MARK: Internals shared by the domain extensions + + func send(_ code: CommandCode, _ payload: [UInt8] = []) async throws -> [UInt8] { + try await core.exchange(code: code.rawValue, payload: payload) + } + + /// Fails fast with the same error the Rust SDK raises for the state. + func requireAuthenticated() async throws { + switch await core.state { + case .shutdown: + throw IggyError(.clientShutdown) + case .disconnected, .connecting, .authenticating: + throw IggyError(.disconnected) Review Comment: Once the state drops to `.disconnected` nothing brings it back for typed calls. One reply timeout (`ClientCore.swift:365` -> `:257`), a cancelled typed call (`:145` throws before the reconnect) or an outage longer than `requestTimeout` (`failConnect`) all land here. After that every typed call throws here without dialing, and the heartbeat at `:117` skips its ping because the state isn't `.authenticated`. Only `ping()` or a manual `connect()` heals it. Rust heals because its heartbeat pings in every state (`core/sdk/src/clients/client.rs:1005`). Suggest pinging in every state except shutdown/caller-disconnected, or reconnecting here when credentials exist, plus a test where the server returns and `getStreams()` succeeds without `connect()`. ########## foreign/swift/Sources/Iggy/Client/ClientCore.swift: ########## @@ -0,0 +1,1027 @@ +// 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 + + /// Shutdown is final: a teardown or a dial that finishes afterwards must + /// not make the client usable again, so no later write replaces it. + private(set) var state: ClientState = .disconnected { + didSet { + if oldValue == .shutdown { + state = .shutdown + } + } + } + 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 + /// 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 } + + func subscribeEvents() -> AsyncStream<DiagnosticEvent> { + let id = UUID() + let (stream, continuation) = AsyncStream<DiagnosticEvent>.makeStream(bufferingPolicy: .bufferingNewest(256)) + guard state != .shutdown else { + // Nothing is published after shutdown, so the stream ends here + // instead of leaving its reader waiting for a first event. + continuation.finish() + return stream + } + 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. One deadline, + /// `requestTimeout` from now, covers the request, the reconnect, the + /// sign-in, the leader check, and the replay. + /// + /// 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) + } + let deadline = ContinuousClock.now + configuration.requestTimeout + var outcome = RetryOutcome() + var failedEpoch = connectionEpoch + do { + return try await sendRaw(code: code, payload: payload, deadline: deadline, epoch: &failedEpoch, outcome: &outcome) + } 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, 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) + guard ContinuousClock.now < deadline else { + // Nothing is left of the deadline to reconnect in; the next + // request does it. + throw error + } + let epoch = failedEpoch + let reconnected: Void? = try await withDeadline(deadline) { [self] in + // The retried sign-in is the sign-in: the reconnect must not + // run another one ahead of it. + try await self.reconnect(failedEpoch: epoch, skipAutoLogin: isLogin, cause: error) + } + guard reconnected != nil else { + throw IggyError(error.code, context: "the request deadline passed while reconnecting after: \(error)") + } + guard replay else { Review Comment: A replicated write with an unknown outcome is rethrown as `.disconnected`, and the deadline branch at `:367` returns `.disconnected` as well (`RetryOutcome.observe` at `:1021` doesn't convert it). That's the same error `requireAuthenticated` throws for a request that was never sent, so callers can't tell "maybe committed" from "not sent". Retry-on-disconnect can then double-apply createPartitions/deletePartitions/deleteSegments. Rust reports `TransientNotCommitted` here (`core/sdk/src/vsr.rs:214-219`). Suggest the same, and update the tests at `ClientTransportTests.swift:230` and `:430`. ########## foreign/swift/Sources/Iggy/Client/ClientCore.swift: ########## @@ -0,0 +1,1027 @@ +// 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 + + /// Shutdown is final: a teardown or a dial that finishes afterwards must + /// not make the client usable again, so no later write replaces it. + private(set) var state: ClientState = .disconnected { + didSet { + if oldValue == .shutdown { + state = .shutdown + } + } + } + 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 + /// 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 } + + func subscribeEvents() -> AsyncStream<DiagnosticEvent> { + let id = UUID() + let (stream, continuation) = AsyncStream<DiagnosticEvent>.makeStream(bufferingPolicy: .bufferingNewest(256)) + guard state != .shutdown else { + // Nothing is published after shutdown, so the stream ends here + // instead of leaving its reader waiting for a first event. + continuation.finish() + return stream + } + 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. One deadline, + /// `requestTimeout` from now, covers the request, the reconnect, the + /// sign-in, the leader check, and the replay. + /// + /// 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) + } + let deadline = ContinuousClock.now + configuration.requestTimeout + var outcome = RetryOutcome() + var failedEpoch = connectionEpoch + do { + return try await sendRaw(code: code, payload: payload, deadline: deadline, epoch: &failedEpoch, outcome: &outcome) + } 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, 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) + guard ContinuousClock.now < deadline else { + // Nothing is left of the deadline to reconnect in; the next + // request does it. + throw error + } + let epoch = failedEpoch + let reconnected: Void? = try await withDeadline(deadline) { [self] in + // The retried sign-in is the sign-in: the reconnect must not + // run another one ahead of it. + try await self.reconnect(failedEpoch: epoch, skipAutoLogin: isLogin, cause: error) + } + guard reconnected != nil else { + throw IggyError(error.code, context: "the request deadline passed while reconnecting after: \(error)") + } + 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, deadline: deadline, epoch: &failedEpoch, outcome: &outcome) + } + } + + /// Replaces the connection a request failed on. Serialized, so callers + /// that failed together reconnect once. + private func reconnect(failedEpoch: UInt64, skipAutoLogin: Bool, cause: IggyError) async throws { + try await routingLock.lock() + defer { routingLock.unlock() } + // 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() + } + logger.info("Reconnecting to the server", metadata: ["server_address": "\(currentAddress)", "error": "\(cause)"]) + try await connect(skipAutoLogin: skipAutoLogin) + } + + /// 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 `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], deadline: ContinuousClock.Instant, epoch: inout UInt64, outcome: inout RetryOutcome + ) async throws -> [UInt8] { + switch state { + case .shutdown: + throw IggyError(.clientShutdown) + case .disconnected, .connecting: + throw outcome.observe(IggyError(.notConnected)) + default: + break + } + let clock = ContinuousClock() + let isLogin = CommandCode.isRegister(code) + var request: VSRFrame.EncodedRequest? + var walk = RosterWalk() + while true { + let transientDeadline = isLogin ? deadline : min(deadline, clock.now + Self.transientFailoverCheckInterval) + let attempted = try await attempt( + code: code, payload: payload, request: &request, epoch: &epoch, outcome: &outcome, transientDeadline: transientDeadline, + readDeadline: deadline) + switch attempted { + case .success(let body): + return body + case .failure(let failure): + let error = outcome.observe(failure) + guard error.code == .transientNotAccepted, !isLogin, clock.now < deadline else { + if error.code == .disconnected, epoch == connectionEpoch, state != .shutdown { + // The reply stream is at an unknown boundary; a late + // reply would desync framing for the next request. + connection = nil + state = .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. + let from = walk + guard let step = try await withDeadline(deadline, { [self] in try await self.failOver(from) }) else { + throw error + } + walk = step.walk + if step.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 Review Comment: This `continue` keeps request id N, but `attempt` releases `exchangeLock`, so a queued caller can stamp N+1 and commit first. Replaying N then hits `AlreadyApplied` in `check_request` (`core/consensus/src/client_table.rs:986-1002`) and surfaces as `RequestTooOld` ("outcome unknown") for a write that provably never reached the log. Reachable on a single node during NotCaughtUp after a restart with concurrent callers. `request = nil` here is safe: NotAccepted proves non-admission, and the guard at `:256` already blocks this path once the outcome is uncertain. ########## foreign/swift/Sources/Iggy/Client/IggyClient.swift: ########## @@ -0,0 +1,246 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +import Logging + +/// A client for an Apache Iggy server over TCP, with or without TLS. +/// +/// The client speaks the consensus wire protocol, the only protocol the +/// server accepts. One client owns one connection and is safe to share +/// between tasks: requests are serialized on the connection, so a single +/// client feeds many producers and consumers. +/// +/// ```swift +/// let client = IggyClient(address: "localhost:8090", autoLogin: .usernamePassword(username: "iggy", password: "iggy")) +/// try await client.connect() +/// let stream = try await client.createStream(name: "orders") +/// let topic = try await client.createTopic(streamID: "orders", name: "created", options: .init(partitionsCount: 3)) +/// try await client.shutdown() +/// ``` +/// +/// Lifecycle: construct, ``connect()``, use, ``shutdown()``. With +/// ``ClientConfiguration/autoLogin`` set, `connect()` also signs in and every +/// reconnect restores the session; otherwise call ``login(username:password:)`` +/// after connecting. ``disconnect()`` closes the connection but keeps the +/// client usable; a client that was shut down cannot reconnect. +/// +/// Every operation throws ``IggyError``, or `CancellationError` when its +/// task is cancelled. A request that hits a dropped connection is replayed +/// over a fresh one only when it provably never reached the log, so a +/// replicated write whose outcome is unknown surfaces the error instead. +public final class IggyClient: Sendable { + /// The configuration the client was created with. + public let configuration: ClientConfiguration + let core: ClientCore + let logger: Logger + + /// Creates a client that dials `configuration.address` on ``connect()``. + public convenience init(configuration: ClientConfiguration = ClientConfiguration(), logger: Logger = Logger(label: "org.apache.iggy")) { + self.init(configuration: configuration, connector: NIOConnector(), logger: logger) + } + + /// Creates a client for `address` (`host:port`). + public convenience init(address: String, autoLogin: Credentials? = nil, tls: TLSOptions? = nil) { + self.init(configuration: ClientConfiguration(address: address, tls: tls, autoLogin: autoLogin)) + } + + init(configuration: ClientConfiguration, connector: any IggyConnector, logger: Logger) { + self.configuration = configuration + self.logger = logger + core = ClientCore(configuration: configuration, connector: connector, logger: logger) + } + + // MARK: Lifecycle + + /// Connects to the server, signing in when auto-login is configured and + /// settling on the cluster leader. Does nothing when already connected. + public func connect() async throws { + try await core.connectByCaller() + await core.startHeartbeat { [weak self] in Review Comment: The doc says this does nothing when already connected, but `startHeartbeat` always calls `stopHeartbeat()` -> `heartbeatTask.cancel()`. If a ping is in flight, the cancel reaches the onCancel in `TCPConnection.swift:133-138` and aborts the channel. `sendRaw` then sets `.disconnected`, and with the issue at `:186` typed calls stay dead. Even once that's fixed it costs a spurious reconnect and an argon2 re-login. Suggest starting a heartbeat only when `heartbeatTask == nil`, or ending the loop with a flag instead of `Task.cancel()`. ########## foreign/swift/Sources/Iggy/Client/ClientCore.swift: ########## @@ -0,0 +1,1027 @@ +// 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 + + /// Shutdown is final: a teardown or a dial that finishes afterwards must + /// not make the client usable again, so no later write replaces it. + private(set) var state: ClientState = .disconnected { + didSet { + if oldValue == .shutdown { + state = .shutdown + } + } + } + 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 + /// 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 } + + func subscribeEvents() -> AsyncStream<DiagnosticEvent> { + let id = UUID() + let (stream, continuation) = AsyncStream<DiagnosticEvent>.makeStream(bufferingPolicy: .bufferingNewest(256)) + guard state != .shutdown else { + // Nothing is published after shutdown, so the stream ends here + // instead of leaving its reader waiting for a first event. + continuation.finish() + return stream + } + 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. One deadline, + /// `requestTimeout` from now, covers the request, the reconnect, the + /// sign-in, the leader check, and the replay. + /// + /// 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) + } + let deadline = ContinuousClock.now + configuration.requestTimeout + var outcome = RetryOutcome() + var failedEpoch = connectionEpoch + do { + return try await sendRaw(code: code, payload: payload, deadline: deadline, epoch: &failedEpoch, outcome: &outcome) + } 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, 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) + guard ContinuousClock.now < deadline else { + // Nothing is left of the deadline to reconnect in; the next + // request does it. + throw error + } + let epoch = failedEpoch + let reconnected: Void? = try await withDeadline(deadline) { [self] in + // The retried sign-in is the sign-in: the reconnect must not + // run another one ahead of it. + try await self.reconnect(failedEpoch: epoch, skipAutoLogin: isLogin, cause: error) + } + guard reconnected != nil else { + throw IggyError(error.code, context: "the request deadline passed while reconnecting after: \(error)") + } + 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, deadline: deadline, epoch: &failedEpoch, outcome: &outcome) + } + } + + /// Replaces the connection a request failed on. Serialized, so callers + /// that failed together reconnect once. + private func reconnect(failedEpoch: UInt64, skipAutoLogin: Bool, cause: IggyError) async throws { + try await routingLock.lock() + defer { routingLock.unlock() } + // 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() + } + logger.info("Reconnecting to the server", metadata: ["server_address": "\(currentAddress)", "error": "\(cause)"]) + try await connect(skipAutoLogin: skipAutoLogin) + } + + /// 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 `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], deadline: ContinuousClock.Instant, epoch: inout UInt64, outcome: inout RetryOutcome + ) async throws -> [UInt8] { + switch state { + case .shutdown: + throw IggyError(.clientShutdown) + case .disconnected, .connecting: + throw outcome.observe(IggyError(.notConnected)) + default: + break + } + let clock = ContinuousClock() + let isLogin = CommandCode.isRegister(code) + var request: VSRFrame.EncodedRequest? + var walk = RosterWalk() + while true { + let transientDeadline = isLogin ? deadline : min(deadline, clock.now + Self.transientFailoverCheckInterval) + let attempted = try await attempt( + code: code, payload: payload, request: &request, epoch: &epoch, outcome: &outcome, transientDeadline: transientDeadline, + readDeadline: deadline) + switch attempted { + case .success(let body): + return body + case .failure(let failure): + let error = outcome.observe(failure) + guard error.code == .transientNotAccepted, !isLogin, clock.now < deadline else { Review Comment: On master since #4506 a normal backup refuses metadata writes and DeleteSegments with `NotPrimary` (89) instead of `TransientNotAccepted` (`core/metadata/src/impls/metadata.rs:3110`). Here 89 decodes to an untyped `.error`, skips this failover guard and isn't reconnectable, so after a view change every replicated write fails until disconnect + connect. The merge base predates #4506, so CI is green and the merge is clean, but it breaks on merge. Please rebase, add `notPrimary`, route it to `failOver` like `transientNotAccepted` (never admitted, safe to re-stamp) and cover it in FakeVSRServer. -- 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]
