You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by gk...@apache.org on 2026/10/14 17:19:11 UTC
(iggy) branch master updated: feat(swift): add the TCP and TLS client (#4503)
This is an automated email from the ASF dual-hosted git repository.
numinnex pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new f0d10d46b feat(swift): add the TCP and TLS client (#4503)
f0d10d46b is described below
commit f0d10d46bcc00d93bfd98ad4d49fabaee702e653
Author: RustToMetal <th...@gmail.com>
AuthorDate: Wed Oct 14 12:19:06 2026 -0500
feat(swift): add the TCP and TLS client (#4503)
## Which issue does this PR address?
Relates to #4001
## Rationale
The Swift SDK has the codecs and the protocol layer (#4355) but nothing
that opens a connection. This is the next step of #4001: the client
itself.
## What changed?
The SDK could encode requests and decode replies but had no way to reach
a server. `IggyClient` now owns one lockstep connection over SwiftNIO,
with optional TLS through NIOSSL: it signs in, heartbeats, reconnects
with the credentials that last succeeded, follows the cluster leader,
and exposes the system, stream, topic, partition, segment, user, and
token commands.
A reconnect registers a new session, so only requests that provably
never reached the log are replayed; a replicated write with an unknown
outcome returns the error instead. Sending, polling, consumer groups,
and offsets are not in this PR: the server now fences them on a
partition context and a routing lookup, so they follow in the next one
together with the end-to-end suite.
---
foreign/swift/Package.resolved | 60 +
foreign/swift/Package.swift | 24 +-
foreign/swift/README.md | 74 +-
.../Sources/Iggy/Client/ClientConfiguration.swift | 142 +++
foreign/swift/Sources/Iggy/Client/ClientCore.swift | 1082 ++++++++++++++++++
.../Sources/Iggy/Client/IggyClient+Resources.swift | 195 ++++
foreign/swift/Sources/Iggy/Client/IggyClient.swift | 239 ++++
foreign/swift/Sources/Iggy/Errors/IggyError.swift | 28 +
.../swift/Sources/Iggy/Models/TopicOptions.swift | 3 +
.../swift/Sources/Iggy/Protocol/CommandCode.swift | 12 +
foreign/swift/Sources/Iggy/Protocol/Requests.swift | 6 +-
.../swift/Sources/Iggy/Transport/Endpoint.swift | 155 +++
.../Sources/Iggy/Transport/TCPConnection.swift | 295 +++++
.../swift/Sources/Iggy/Transport/TLSOptions.swift | 67 ++
.../swift/Sources/Iggy/Utilities/AsyncMutex.swift | 201 ++++
.../Tests/IggyTests/ClientTransportTests.swift | 1168 ++++++++++++++++++++
.../swift/Tests/IggyTests/ConcurrencyTests.swift | 209 ++++
.../swift/Tests/IggyTests/ConfigurationTests.swift | 120 ++
.../Tests/IggyTests/Support/FakeVSRServer.swift | 331 ++++++
.../Tests/IggyTests/Support/TestHelpers.swift | 64 ++
20 files changed, 4459 insertions(+), 16 deletions(-)
diff --git a/foreign/swift/Package.resolved b/foreign/swift/Package.resolved
new file mode 100644
index 000000000..762d62b53
--- /dev/null
+++ b/foreign/swift/Package.resolved
@@ -0,0 +1,60 @@
+{
+ "originHash" : "62946d564ad9c5d0b9eff5c2610c835de3d4e4e178e822878bdce27e328763ee",
+ "pins" : [
+ {
+ "identity" : "swift-atomics",
+ "kind" : "remoteSourceControl",
+ "location" : "https://github.com/apple/swift-atomics.git",
+ "state" : {
+ "revision" : "0442cb5a3f98ab802acb777929fdb446bda11a34",
+ "version" : "1.3.1"
+ }
+ },
+ {
+ "identity" : "swift-collections",
+ "kind" : "remoteSourceControl",
+ "location" : "https://github.com/apple/swift-collections.git",
+ "state" : {
+ "revision" : "a0cb0954ecb21e4e31b0070e6ed5674e8556685a",
+ "version" : "1.6.0"
+ }
+ },
+ {
+ "identity" : "swift-log",
+ "kind" : "remoteSourceControl",
+ "location" : "https://github.com/apple/swift-log.git",
+ "state" : {
+ "revision" : "9c6fb14227f55d8f711ce3847dc2f419fb0ecacb",
+ "version" : "1.15.1"
+ }
+ },
+ {
+ "identity" : "swift-nio",
+ "kind" : "remoteSourceControl",
+ "location" : "https://github.com/apple/swift-nio.git",
+ "state" : {
+ "revision" : "a931f2c1de8dd49381ce3bf2e279d033f68d8865",
+ "version" : "2.102.0"
+ }
+ },
+ {
+ "identity" : "swift-nio-ssl",
+ "kind" : "remoteSourceControl",
+ "location" : "https://github.com/apple/swift-nio-ssl.git",
+ "state" : {
+ "revision" : "03827c1a9fdb2b6b00a4e93ede8861520263af8c",
+ "version" : "2.37.4"
+ }
+ },
+ {
+ "identity" : "swift-system",
+ "kind" : "remoteSourceControl",
+ "location" : "https://github.com/apple/swift-system.git",
+ "state" : {
+ "revision" : "869129b7bf4ecc57b97d0193ad29690ca2134750",
+ "version" : "1.8.1"
+ }
+ }
+ ],
+ "version" : 3
+}
diff --git a/foreign/swift/Package.swift b/foreign/swift/Package.swift
index 27af6f7ae..66fe250f1 100644
--- a/foreign/swift/Package.swift
+++ b/foreign/swift/Package.swift
@@ -30,11 +30,31 @@ let package = Package(
products: [
.library(name: "Iggy", targets: ["Iggy"])
],
+ dependencies: [
+ .package(url: "https://github.com/apple/swift-nio.git", from: "2.80.0"),
+ .package(url: "https://github.com/apple/swift-nio-ssl.git", from: "2.29.0"),
+ .package(url: "https://github.com/apple/swift-log.git", from: "1.6.0"),
+ ],
targets: [
- .target(name: "Iggy"),
+ .target(
+ name: "Iggy",
+ dependencies: [
+ .product(name: "NIOConcurrencyHelpers", package: "swift-nio"),
+ .product(name: "NIOCore", package: "swift-nio"),
+ .product(name: "NIOPosix", package: "swift-nio"),
+ .product(name: "NIOTLS", package: "swift-nio"),
+ .product(name: "NIOSSL", package: "swift-nio-ssl"),
+ .product(name: "Logging", package: "swift-log"),
+ ]
+ ),
.testTarget(
name: "IggyTests",
- dependencies: ["Iggy"],
+ dependencies: [
+ "Iggy",
+ .product(name: "NIOCore", package: "swift-nio"),
+ .product(name: "NIOPosix", package: "swift-nio"),
+ .product(name: "Logging", package: "swift-log"),
+ ],
resources: [.copy("Fixtures")]
),
],
diff --git a/foreign/swift/README.md b/foreign/swift/README.md
index 7b3d31111..0d356dab7 100644
--- a/foreign/swift/README.md
+++ b/foreign/swift/README.md
@@ -10,18 +10,22 @@
Official Swift client SDK for [Apache Iggy](https://iggy.apache.org) message streaming.
-The SDK is built up in stages. This stage carries the package scaffold, the error table,
-the byte codec the wire protocol is encoded with, native XXH3-64 and XXH32 hashes pinned
-to the Rust crates by golden vectors, the domain models (streams, topics, messages,
-user headers, options, permissions) with their wire encodings, and the protocol layer:
-the consensus frame, the session and request id rules, the request encoders, the reply
-decoders, and the message batch format. The TCP and TLS client, the producer and
-consumer, examples, and BDD scenarios follow in later changes.
+The SDK is built up in stages. This stage adds the client: a TCP connection with
+optional TLS that speaks the consensus protocol, signs in, keeps the session alive with
+a heartbeat, reconnects, follows the cluster leader, and manages streams, topics,
+partitions, segments, users, and personal access tokens. Sending and polling messages,
+consumer groups and offsets, the producer and consumer, examples, and BDD scenarios
+follow in later changes.
## Requirements
- Swift 6.0 or later (Swift 6 language mode, strict concurrency)
- macOS 13, iOS 16, tvOS 16, watchOS 9, visionOS 1, or Linux with a Swift 6 toolchain
+- An Iggy server that speaks the consensus protocol (`iggy-server` from this repository)
+
+The SDK depends on [SwiftNIO](https://github.com/apple/swift-nio),
+[swift-nio-ssl](https://github.com/apple/swift-nio-ssl), and
+[swift-log](https://github.com/apple/swift-log).
## Installation
@@ -52,13 +56,57 @@ targets: [
## Usage
-Every operation of the SDK throws `IggyError`, a value carrying the typed `code`
-(`IggyErrorCode`, one case per server error code), the raw wire code, and optional
-context:
+Create a client, connect, and manage resources. With `autoLogin` set, `connect()` signs
+in and every reconnect restores the session:
```swift
import Iggy
+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))
+await client.shutdown()
+```
+
+One client owns one connection and is safe to share between tasks: requests are
+serialized on the connection. `disconnect()` closes the connection until the next
+`connect()`, `shutdown()` is final.
+
+### Configuration
+
+`ClientConfiguration` carries the address, TLS, the credentials to sign in with, the
+reconnection policy, the heartbeat interval, and the connect and request timeouts:
+
+```swift
+var configuration = ClientConfiguration(address: "iggy.example:8090")
+configuration.autoLogin = .personalAccessToken(token)
+configuration.tls = TLSOptions(caFile: "/etc/iggy/ca.pem")
+configuration.reconnection = ReconnectionOptions(maxRetries: 5, interval: .seconds(2))
+let client = IggyClient(configuration: configuration)
+```
+
+The server certificate is verified against the system roots, or against `caFile` when
+set, and against the host of the address unless `TLSOptions.domain` names another one.
+
+### Reconnection
+
+A request that hits a dropped connection reconnects, signs in again, and is replayed,
+all within its `requestTimeout`, and only when it provably never reached the log:
+reads, pings, and requests the server never admitted. A replicated write whose outcome
+is unknown fails with `transientNotCommitted` instead of being applied twice. A client
+left disconnected by a timeout or an outage reconnects on its next call or heartbeat
+when it has credentials to sign in with. In a cluster the client reads the roster after
+signing in, moves to the leader, and moves on when a node answers that it is not the
+primary.
+
+### Errors
+
+Every operation throws `IggyError`, a value carrying the typed `code`
+(`IggyErrorCode`, one case per server error code), the raw wire code, and optional
+context. A cancelled task gets `CancellationError`:
+
+```swift
func handle(_ error: IggyError) {
switch error.code {
case .streamNameNotFound:
@@ -69,9 +117,6 @@ func handle(_ error: IggyError) {
}
```
-The client that raises these errors lands in the next changes of the series; see the
-[Rust SDK](../../core/sdk) for the operations it mirrors.
-
## Testing
```bash
@@ -79,6 +124,9 @@ cd foreign/swift
swift test
```
+The client tests run against an in-process server that speaks the consensus framing
+and is scripted per test, so they need no running `iggy-server`.
+
The hash, error-code, model, request, reply, and consensus header tests compare against
`Tests/IggyTests/Fixtures/golden.json`, a fixture dumped from the Rust protocol crates. Regenerate it after a protocol change, from
`foreign/swift`, and commit the result; CI fails when the checked-in copy is stale:
diff --git a/foreign/swift/Sources/Iggy/Client/ClientConfiguration.swift b/foreign/swift/Sources/Iggy/Client/ClientConfiguration.swift
new file mode 100644
index 000000000..50dca2bb5
--- /dev/null
+++ b/foreign/swift/Sources/Iggy/Client/ClientConfiguration.swift
@@ -0,0 +1,142 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+/// Credentials a client signs in with.
+public enum Credentials: Sendable, Hashable {
+ case usernamePassword(username: String, password: String)
+ case personalAccessToken(String)
+}
+
+extension Credentials: CustomStringConvertible {
+ /// Never includes the secret.
+ public var description: String {
+ switch self {
+ case .usernamePassword(let username, _): "username \(username)"
+ case .personalAccessToken: "a personal access token"
+ }
+ }
+}
+
+/// How the client reconnects after losing its connection.
+public struct ReconnectionOptions: Sendable, Hashable {
+ /// Whether to reconnect at all. Off, a dropped connection fails the
+ /// request that hit it and every later one until `connect()` is called.
+ public var enabled: Bool
+ /// How many more rounds over the known endpoints to attempt after the
+ /// first one fails; nil means forever, and zero still makes one round.
+ public var maxRetries: UInt32?
+ /// Pause between rounds.
+ public var interval: Duration
+ /// Grace period after a successful connection before redialing the same
+ /// endpoint, so a flapping server is not hammered.
+ public var reestablishAfter: Duration
+
+ public init(enabled: Bool = true, maxRetries: UInt32? = nil, interval: Duration = .seconds(1), reestablishAfter: Duration = .seconds(5)) {
+ self.enabled = enabled
+ self.maxRetries = maxRetries
+ self.interval = interval
+ self.reestablishAfter = reestablishAfter
+ }
+}
+
+/// Configuration of a TCP client.
+///
+/// ```swift
+/// var configuration = ClientConfiguration(address: "localhost:8090")
+/// configuration.autoLogin = .usernamePassword(username: "iggy", password: "iggy")
+/// configuration.tls = TLSOptions(caFile: "core/certs/iggy_ca_cert.pem", domain: "localhost")
+/// ```
+public struct ClientConfiguration: Sendable, Hashable {
+ /// `host:port` of the server, or of any node of a cluster.
+ public var address: String
+ /// TLS settings; nil for plain TCP.
+ public var tls: TLSOptions?
+ /// Credentials to sign in with after every connection, including the
+ /// ones a reconnect establishes. Without them a reconnect cannot restore
+ /// the session, so a request that hits a dropped connection fails
+ /// instead of replaying.
+ public var autoLogin: Credentials?
+ /// How the client reconnects after losing its connection.
+ public var reconnection: ReconnectionOptions
+ /// Period of the ping the client sends to stay alive; the server evicts
+ /// clients that miss too many. Zero turns the heartbeat off.
+ public var heartbeatInterval: Duration
+ /// Disables Nagle's algorithm. On by default: the lockstep request model
+ /// stalls a multi-segment frame behind segment coalescing.
+ public var noDelay: Bool
+ /// Upper bound on one dial, including the TLS handshake.
+ public var connectTimeout: Duration
+ /// Upper bound on waiting for one reply across every replay and failover.
+ /// Far beyond any healthy round trip; it only trips when the server loses
+ /// the reply entirely.
+ public var requestTimeout: Duration
+
+ public init(
+ address: String = "127.0.0.1:8090", tls: TLSOptions? = nil, autoLogin: Credentials? = nil,
+ reconnection: ReconnectionOptions = ReconnectionOptions(), heartbeatInterval: Duration = .seconds(5),
+ noDelay: Bool = true, connectTimeout: Duration = .seconds(30), requestTimeout: Duration = .seconds(30)
+ ) {
+ self.address = address
+ self.tls = tls
+ self.autoLogin = autoLogin
+ self.reconnection = reconnection
+ self.heartbeatInterval = heartbeatInterval
+ self.noDelay = noDelay
+ self.connectTimeout = connectTimeout
+ self.requestTimeout = requestTimeout
+ }
+}
+
+/// Where the client is in its lifecycle.
+public enum ClientState: Sendable, Hashable, CustomStringConvertible {
+ case shutdown
+ case disconnected
+ case connecting
+ case connected
+ case authenticating
+ case authenticated
+
+ public var description: String {
+ switch self {
+ case .shutdown: "shutdown"
+ case .disconnected: "disconnected"
+ case .connecting: "connecting"
+ case .connected: "connected"
+ case .authenticating: "authenticating"
+ case .authenticated: "authenticated"
+ }
+ }
+}
+
+/// Lifecycle events a client publishes through ``IggyClient/events``.
+public enum DiagnosticEvent: Sendable, Hashable, CustomStringConvertible {
+ case shutdown
+ case disconnected
+ case connected
+ case signedIn
+ case signedOut
+
+ public var description: String {
+ switch self {
+ case .shutdown: "shutdown"
+ case .disconnected: "disconnected"
+ case .connected: "connected"
+ case .signedIn: "signed_in"
+ case .signedOut: "signed_out"
+ }
+ }
+}
diff --git a/foreign/swift/Sources/Iggy/Client/ClientCore.swift b/foreign/swift/Sources/Iggy/Client/ClientCore.swift
new file mode 100644
index 000000000..7cc9b57b0
--- /dev/null
+++ b/foreign/swift/Sources/Iggy/Client/ClientCore.swift
@@ -0,0 +1,1082 @@
+// 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 replicated write that may have reached the log before the connection
+ /// was lost is not replayed and fails with
+ /// ``IggyErrorCode/transientNotCommitted``, as in the Rust SDK, so it is
+ /// never mistaken for a request that was not sent.
+ ///
+ /// 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
+ }
+ // 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)
+ let failure = replay ? error : IggyError(.transientNotCommitted, context: "command \(code) may have committed before: \(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 replay ? IggyError(.disconnected, context: "automatic reconnection is disabled") : failure
+ }
+ 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 failure
+ }
+ 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
+ }
+ guard ContinuousClock.now < deadline else {
+ // Nothing is left of the deadline to reconnect in; the next
+ // request does it.
+ throw failure
+ }
+ 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(failure.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")
+ throw failure
+ }
+ return try await sendRaw(code: code, payload: payload, deadline: deadline, epoch: &failedEpoch, outcome: &outcome)
+ }
+ }
+
+ /// Fails fast with the error the Rust SDK raises for the state, after
+ /// bringing back a connection that dropped on its own. A reply timeout, a
+ /// cancelled request, or an outage longer than the request timeout leaves
+ /// the client disconnected, and with credentials to sign in again the
+ /// next typed call reconnects instead of failing until ``connect()``.
+ func requireAuthenticated() async throws {
+ switch state {
+ case .disconnected, .connecting, .authenticating:
+ if !disconnectedByCaller, configuration.reconnection.enabled, signInCredentials() != nil {
+ let reconnected: Void? = try await withDeadline(ContinuousClock.now + configuration.requestTimeout) { [self] in
+ try await self.connect()
+ }
+ if reconnected == nil {
+ throw IggyError(.disconnected, context: "could not reconnect within the request timeout")
+ }
+ }
+ default:
+ break
+ }
+ switch state {
+ case .shutdown:
+ throw IggyError(.clientShutdown)
+ case .disconnected, .connecting, .authenticating:
+ throw IggyError(.disconnected)
+ case .connected:
+ throw IggyError(.unauthenticated)
+ case .authenticated:
+ return
+ }
+ }
+
+ /// 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(Self.graded(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
+ // under a new request id cannot double-apply it.
+ let from = walk
+ guard let step = try await withDeadline(deadline, { [self] in try await self.failOver(from) }) else {
+ throw error
+ }
+ walk = step.walk
+ guard step.moved || rosterEndpoints.count <= 1 else {
+ throw error
+ }
+ // A caller queued behind this one may have spent the next
+ // request id while the exchange lock was free. Replaying the
+ // old id after it would be refused as too old, so the frame is
+ // stamped afresh.
+ request = nil
+ if !step.moved {
+ // A single node: partition materialisation can outlast the
+ // short retry window, so keep replaying until the budget ends.
+ try? await Task.sleep(until: min(clock.now + Self.notReadyRetryInterval, deadline), clock: clock)
+ }
+ }
+ }
+ }
+
+ /// `notPrimary` takes the `transientNotAccepted` path, as in the Rust SDK:
+ /// neither was admitted, and the primary is found by moving to another
+ /// node. Unlike a not-accepted refusal it is not replayed on the same node
+ /// first, since a backup keeps refusing until the view changes.
+ private static func graded(_ failure: IggyError) -> IggyError {
+ guard failure.code == .notPrimary else {
+ return failure
+ }
+ return IggyError(.transientNotAccepted, context: "\(failure)")
+ }
+
+ /// The cluster endpoints one request has already been refused on.
+ private struct RosterWalk: Sendable {
+ var visited = Set<String>()
+ var checkedLeader = false
+ }
+
+ /// Moves the connection for a request the current node keeps refusing to
+ /// admit: to the metadata leader once, then along the roster. Reports
+ /// whether the client moved.
+ private func failOver(_ from: RosterWalk) async throws -> (moved: Bool, walk: RosterWalk) {
+ var walk = from
+ try await routingLock.lock()
+ defer { routingLock.unlock() }
+ if !walk.checkedLeader {
+ walk.checkedLeader = true
+ walk.visited.insert(currentAddress)
+ if try await handleLeaderRedirection() {
+ try await connect()
+ return (true, walk)
+ }
+ }
+ guard let next = nextRosterEndpoint(excluding: walk.visited) else {
+ return (false, walk)
+ }
+ // 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.
+ walk.visited.insert(next)
+ try await settle(on: next)
+ try await connect(settleOffLeader: true)
+ return (true, walk)
+ }
+
+ /// 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, outcome: inout RetryOutcome,
+ transientDeadline: ContinuousClock.Instant, readDeadline: ContinuousClock.Instant
+ ) async throws -> Result<[UInt8], IggyError> {
+ try await exchangeLock.lock()
+ defer { exchangeLock.unlock() }
+ return await exchangeLocked(
+ code: code, payload: payload, request: &request, epoch: &epoch, outcome: &outcome, transientDeadline: transientDeadline,
+ readDeadline: readDeadline)
+ }
+
+ /// 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, outcome: inout RetryOutcome,
+ 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()
+ // The answer to the last attempt. At the deadline it is the error, as
+ // in the Rust SDK, and with no attempt written the request was never
+ // admitted. Either way the stream sits at a frame boundary, so the
+ // connection stays.
+ var refusal: IggyError?
+ while true {
+ let remaining = readDeadline - clock.now
+ guard remaining > .zero else {
+ return .failure(refusal ?? IggyError(.transientNotAccepted, context: "the request deadline passed before command \(code) was sent"))
+ }
+ 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 answer as IggyError {
+ // Once one attempt left the outcome unknown, a later refusal
+ // of the same request no longer proves it never committed.
+ let error = outcome.observe(answer)
+ refusal = error
+ 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()
+ }
+
+ /// Connects unless a connection stands. Callers share one sweep; a caller
+ /// that is cancelled leaves it, and the sweep ends when none is left.
+ ///
+ /// `skipAutoLogin` belongs to the caller whose own request is the sign-in.
+ func connect(skipAutoLogin: Bool = false, settleOffLeader: Bool = false) async throws {
+ try await connectGate.run { [self] in
+ try await self.connectInner(skipAutoLogin: skipAutoLogin, settleOffLeader: settleOffLeader)
+ }
+ }
+
+ private func connectInner(skipAutoLogin: Bool, settleOffLeader: Bool) async throws {
+ do {
+ try await sweep(skipAutoLogin: skipAutoLogin, settleOffLeader: settleOffLeader)
+ } catch {
+ // A sweep that ends early must not leave a half-open state behind.
+ // `.connecting` would make the next connect return at the top
+ // without dialing, and a connection whose sign-in was abandoned
+ // with its last caller is closed already.
+ switch state {
+ case .connecting:
+ failConnect()
+ case .connected, .authenticating:
+ if error is CancellationError {
+ await disconnectTransport()
+ }
+ default:
+ break
+ }
+ throw error
+ }
+ }
+
+ private func sweep(skipAutoLogin: Bool, 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()
+ state = .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())
+ }
+ var retries: UInt32 = 0
+ var candidateIndex = 0
+ var configurationFault: IggyError?
+ var signInFailure: IggyError?
+ var dialFailure: IggyError?
+ var shouldRedirect = false
+ sweep: while true {
+ try endSweepIfStopped()
+ 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 endSweepIfStopped()
+ logger.info("Connecting to server \(address)")
+ do {
+ let established = try await establishBounded(address, candidates: candidates)
+ if state == .shutdown || disconnectedByCaller || Task.isCancelled {
+ // The client was stopped while this was being dialed:
+ // the connection is dropped before it is installed,
+ // and the check at the top of the sweep ends it.
+ await established.close()
+ continue sweep
+ }
+ currentAddress = address
+ connection = established
+ connectionEpoch += 1
+ state = .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
+ state = .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)
+ 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
+ }
+ state = .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 {
+ state = .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() {
+ state = .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: once the client is shut down,
+ /// disconnected by its caller, or left without a caller waiting for it,
+ /// the sweep must not bring it back.
+ private func endSweepIfStopped() throws {
+ if state == .shutdown {
+ throw IggyError(.clientShutdown)
+ }
+ try checkNotDisconnectedByCaller()
+ try Task.checkCancellation()
+ }
+
+ /// 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, .shutdown:
+ 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)")
+ state = .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()
+ state = .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)"])
+ state = .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()
+ state = .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 =
+ (try? await withDeadline(ContinuousClock.now + 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
+
+ /// Runs `body` every heartbeat interval until it returns false, which is
+ /// how a client that was dropped without a shutdown ends its heartbeat.
+ /// A heartbeat already running is kept: cancelling it could abort a ping
+ /// in flight and take the connection with it.
+ func startHeartbeat(_ body: @escaping @Sendable () async -> Bool) {
+ let interval = configuration.heartbeatInterval
+ guard heartbeatTask == nil, interval > .zero, state != .shutdown else {
+ return
+ }
+ heartbeatTask = Task {
+ while !Task.isCancelled {
+ try? await Task.sleep(for: interval)
+ if Task.isCancelled {
+ return
+ }
+ guard await body() else {
+ return
+ }
+ }
+ }
+ }
+
+ func stopHeartbeat() {
+ heartbeatTask?.cancel()
+ heartbeatTask = nil
+ }
+}
+
+/// Whether an earlier attempt of a request may have committed, a port of
+/// `RetryOutcome` in `core/sdk/src/vsr.rs`.
+///
+/// `transientNotCommitted` leaves the outcome unknown. A later refusal of the
+/// same request, on this node or the next, cannot establish that the earlier
+/// attempt did not commit, so it is reported as unknown too instead of as
+/// proof that re-issuing the request under a new session is safe.
+struct RetryOutcome {
+ private(set) var isUncertain = false
+
+ mutating func observe(_ error: IggyError) -> IggyError {
+ if error.code == .transientNotCommitted {
+ isUncertain = true
+ }
+ guard isUncertain else {
+ return error
+ }
+ switch error.code {
+ case .transientNotAccepted, .historyUnavailable, .unauthorized, .unauthenticated, .staleClient, .notConnected, .cannotEstablishConnection:
+ return IggyError(.transientNotCommitted, context: "an earlier attempt may have committed, then: \(error)")
+ default:
+ return error
+ }
+ }
+}
diff --git a/foreign/swift/Sources/Iggy/Client/IggyClient+Resources.swift b/foreign/swift/Sources/Iggy/Client/IggyClient+Resources.swift
new file mode 100644
index 000000000..a57284887
--- /dev/null
+++ b/foreign/swift/Sources/Iggy/Client/IggyClient+Resources.swift
@@ -0,0 +1,195 @@
+// 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.
+
+// MARK: - Streams
+
+extension IggyClient {
+ /// A stream by id or name, with its topics, or nil if it does not exist.
+ public func getStream(_ streamID: Identifier) async throws -> StreamDetails? {
+ try await requireAuthenticated()
+ let body = try await send(.getStream, Requests.identifierOnly(streamID))
+ return body.isEmpty ? nil : try Responses.streamDetails(body[...])
+ }
+
+ /// Every stream.
+ public func getStreams() async throws -> [StreamInfo] {
+ try await requireAuthenticated()
+ let body = try await send(.getStreams)
+ return body.isEmpty ? [] : try Responses.streams(body[...])
+ }
+
+ /// Creates a stream and returns it.
+ public func createStream(name: String) async throws -> StreamDetails {
+ try await requireAuthenticated()
+ return try Responses.streamDetails(try await send(.createStream, try Requests.createStream(name: name))[...])
+ }
+
+ /// Renames a stream.
+ public func updateStream(_ streamID: Identifier, name: String, options: StreamUpdateOptions = StreamUpdateOptions()) async throws {
+ try await requireAuthenticated()
+ _ = try await send(.updateStream, try Requests.updateStream(streamID: streamID, name: name, options: try options.encode()))
+ }
+
+ /// Deletes a stream with every topic and message in it.
+ public func deleteStream(_ streamID: Identifier) async throws {
+ try await requireAuthenticated()
+ _ = try await send(.deleteStream, Requests.identifierOnly(streamID))
+ }
+}
+
+// MARK: - Topics
+
+extension IggyClient {
+ /// A topic by id or name, with its partitions, or nil if it does not exist.
+ public func getTopic(streamID: Identifier, topicID: Identifier) async throws -> TopicDetails? {
+ try await requireAuthenticated()
+ let body = try await send(.getTopic, Requests.streamAndTopic(streamID, topicID))
+ return body.isEmpty ? nil : try Responses.topicDetails(body[...])
+ }
+
+ /// Every topic of a stream.
+ public func getTopics(streamID: Identifier) async throws -> [Topic] {
+ try await requireAuthenticated()
+ let body = try await send(.getTopics, Requests.identifierOnly(streamID))
+ return body.isEmpty ? [] : try Responses.topics(body[...])
+ }
+
+ /// Creates a topic and returns it. Every knob rides `options`; an absent
+ /// key resolves against the server's defaults.
+ public func createTopic(streamID: Identifier, name: String, options: TopicCreateOptions = TopicCreateOptions()) async throws -> TopicDetails {
+ try await requireAuthenticated()
+ let payload = try Requests.createTopic(
+ streamID: streamID, partitionsCount: options.partitionsCount ?? TopicCreateOptions.defaultPartitionsCount,
+ name: name, options: try options.encode())
+ return try Responses.topicDetails(try await send(.createTopic, payload)[...])
+ }
+
+ /// Renames a topic and updates its updatable options.
+ public func updateTopic(streamID: Identifier, topicID: Identifier, name: String, options: TopicUpdateOptions = TopicUpdateOptions()) async throws {
+ try await requireAuthenticated()
+ _ = try await send(.updateTopic, try Requests.updateTopic(streamID: streamID, topicID: topicID, name: name, options: try options.encode()))
+ }
+
+ /// Deletes a topic with every message in it.
+ public func deleteTopic(streamID: Identifier, topicID: Identifier) async throws {
+ try await requireAuthenticated()
+ _ = try await send(.deleteTopic, Requests.streamAndTopic(streamID, topicID))
+ }
+}
+
+// MARK: - Partitions and segments
+
+extension IggyClient {
+ /// Appends `count` partitions to a topic.
+ public func createPartitions(streamID: Identifier, topicID: Identifier, count: UInt32) async throws {
+ try await requireAuthenticated()
+ _ = try await send(.createPartitions, Requests.partitions(streamID: streamID, topicID: topicID, partitionsCount: count))
+ }
+
+ /// Removes the last `count` partitions of a topic.
+ public func deletePartitions(streamID: Identifier, topicID: Identifier, count: UInt32) async throws {
+ try await requireAuthenticated()
+ _ = try await send(.deletePartitions, Requests.partitions(streamID: streamID, topicID: topicID, partitionsCount: count))
+ }
+
+ /// Removes the oldest `count` closed segments of a partition.
+ public func deleteSegments(streamID: Identifier, topicID: Identifier, partitionID: UInt32, count: UInt32) async throws {
+ try await requireAuthenticated()
+ _ = try await send(.deleteSegments, Requests.deleteSegments(streamID: streamID, topicID: topicID, partitionID: partitionID, segmentsCount: count))
+ }
+}
+
+// MARK: - Users
+
+extension IggyClient {
+ /// A user by id or username, with its permissions, or nil if unknown.
+ public func getUser(_ userID: Identifier) async throws -> UserInfoDetails? {
+ try await requireAuthenticated()
+ let body = try await send(.getUser, Requests.identifierOnly(userID))
+ return body.isEmpty ? nil : try Responses.userDetails(body[...])
+ }
+
+ /// Every user.
+ public func getUsers() async throws -> [UserInfo] {
+ try await requireAuthenticated()
+ let body = try await send(.getUsers)
+ return body.isEmpty ? [] : try Responses.users(body[...])
+ }
+
+ /// Creates a user and returns it.
+ public func createUser(username: String, password: String, status: UserStatus = .active, permissions: Permissions? = nil) async throws -> UserInfoDetails {
+ try await requireAuthenticated()
+ let payload = try Requests.createUser(username: username, password: password, status: status, permissions: permissions)
+ return try Responses.userDetails(try await send(.createUser, payload)[...])
+ }
+
+ /// Deletes a user and the tokens it owns.
+ public func deleteUser(_ userID: Identifier) async throws {
+ try await requireAuthenticated()
+ _ = try await send(.deleteUser, Requests.identifierOnly(userID))
+ }
+
+ /// Changes a user's name and/or status; nil leaves a field as is.
+ public func updateUser(
+ _ userID: Identifier, username: String? = nil, status: UserStatus? = nil, options: UserUpdateOptions = UserUpdateOptions()
+ ) async throws {
+ try await requireAuthenticated()
+ _ = try await send(.updateUser, try Requests.updateUser(userID: userID, username: username, status: status, options: try options.encode()))
+ if let username {
+ await core.refreshSessionUsername(user: userID, newUsername: username)
+ }
+ }
+
+ /// Replaces a user's permissions; nil removes them all.
+ public func updatePermissions(_ userID: Identifier, permissions: Permissions?) async throws {
+ try await requireAuthenticated()
+ _ = try await send(.updatePermissions, Requests.updatePermissions(userID: userID, permissions: permissions))
+ }
+
+ /// Changes a password. When it is the signed-in or auto-login user, the
+ /// credentials used by later reconnects switch to the new password.
+ public func changePassword(_ userID: Identifier, currentPassword: String, newPassword: String) async throws {
+ try await requireAuthenticated()
+ _ = try await send(.changePassword, try Requests.changePassword(userID: userID, currentPassword: currentPassword, newPassword: newPassword))
+ await core.refreshSessionPassword(user: userID, newPassword: newPassword)
+ }
+}
+
+// MARK: - Personal access tokens
+
+extension IggyClient {
+ /// The signed-in user's tokens. The secrets are not included.
+ public func getPersonalAccessTokens() async throws -> [PersonalAccessTokenInfo] {
+ try await requireAuthenticated()
+ let body = try await send(.getPersonalAccessTokens)
+ return body.isEmpty ? [] : try Responses.personalAccessTokens(body[...])
+ }
+
+ /// Creates a token for the signed-in user. The secret is only returned
+ /// here.
+ public func createPersonalAccessToken(name: String, expiry: PersonalAccessTokenExpiry = .never) async throws -> RawPersonalAccessToken {
+ try await requireAuthenticated()
+ return try Responses.rawPersonalAccessToken(
+ try await send(.createPersonalAccessToken, try Requests.createPersonalAccessToken(name: name, expiry: expiry))[...])
+ }
+
+ /// Deletes one of the signed-in user's tokens by name.
+ public func deletePersonalAccessToken(name: String) async throws {
+ try await requireAuthenticated()
+ _ = try await send(.deletePersonalAccessToken, try Requests.nameOnly(name))
+ }
+}
diff --git a/foreign/swift/Sources/Iggy/Client/IggyClient.swift b/foreign/swift/Sources/Iggy/Client/IggyClient.swift
new file mode 100644
index 000000000..cee5cd4ad
--- /dev/null
+++ b/foreign/swift/Sources/Iggy/Client/IggyClient.swift
@@ -0,0 +1,239 @@
+// 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 fails with
+/// ``IggyErrorCode/transientNotCommitted`` 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 }
+ }
+
+ /// Pings in every state but shutdown, as the Rust SDK does: a ping on a
+ /// client whose connection dropped is what reconnects it between calls.
+ private func heartbeat() async {
+ guard await core.state != .shutdown 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)
+ }
+
+ func requireAuthenticated() async throws {
+ try await core.requireAuthenticated()
+ }
+}
+
+// MARK: - System
+
+extension IggyClient {
+ /// Checks that the server is alive. Works before signing in.
+ public func ping() async throws {
+ _ = try await send(.ping)
+ }
+
+ /// Server statistics: process, memory, disk, counts.
+ public func getStats() async throws -> Stats {
+ try Responses.stats(try await send(.getStats)[...])
+ }
+
+ /// The server's view of this client.
+ public func getMe() async throws -> ClientInfoDetails {
+ try await requireAuthenticated()
+ return try Responses.clientDetails(try await send(.getMe)[...])
+ }
+
+ /// A connected client by its server-assigned id, or nil if unknown.
+ public func getClient(id: UInt32) async throws -> ClientInfoDetails? {
+ try await requireAuthenticated()
+ let body = try await send(.getClient, Requests.getClient(clientID: id))
+ return body.isEmpty ? nil : try Responses.clientDetails(body[...])
+ }
+
+ /// Every connected client.
+ public func getClients() async throws -> [ClientInfo] {
+ try await requireAuthenticated()
+ let body = try await send(.getClients)
+ return body.isEmpty ? [] : try Responses.clients(body[...])
+ }
+
+ /// The option catalog of a resource scope: the keys create accepts, their
+ /// kinds, and this server's defaults.
+ public func describeOptions(scope: OptionsScope) async throws -> [OptionSpec] {
+ try await requireAuthenticated()
+ return try Responses.optionSpecs(try await send(.describeOptions, Requests.describeOptions(scope: scope))[...])
+ }
+
+ /// Captures a snapshot archive of the server state.
+ public func snapshot(compression: SnapshotCompression = .deflated, types: [SystemSnapshotType] = [.all]) async throws -> Snapshot {
+ try await requireAuthenticated()
+ return Snapshot(data: try await send(.getSnapshotFile, try Requests.getSnapshot(compression: compression, types: types)))
+ }
+
+ /// The cluster roster: every node, its role, and its endpoints. Requires
+ /// a signed-in session.
+ public func getClusterMetadata() async throws -> ClusterMetadata {
+ try Responses.clusterMetadata(try await send(.getClusterMetadata)[...])
+ }
+}
diff --git a/foreign/swift/Sources/Iggy/Errors/IggyError.swift b/foreign/swift/Sources/Iggy/Errors/IggyError.swift
index 96d4460bc..2f8fc8a3c 100644
--- a/foreign/swift/Sources/Iggy/Errors/IggyError.swift
+++ b/foreign/swift/Sources/Iggy/Errors/IggyError.swift
@@ -76,3 +76,31 @@ public struct IggyError: Error, Sendable, Hashable, CustomStringConvertible {
hasher.combine(rawCode)
}
}
+
+extension IggyError {
+ /// Whether the failure means the connection itself is gone.
+ var isConnectionLoss: Bool {
+ switch code {
+ case .disconnected, .emptyResponse, .notConnected, .cannotEstablishConnection, .tcpError, .staleClient:
+ true
+ default:
+ false
+ }
+ }
+
+ /// Whether a fresh connection and sign-in can recover from this failure.
+ var isReconnectable: Bool {
+ isConnectionLoss || code == .unauthenticated
+ }
+
+ /// Whether a sign-in that failed this way would fail the same way on a
+ /// retry, so the remembered credentials are dropped.
+ var isCredentialRejection: Bool {
+ switch code {
+ case .invalidCredentials, .invalidUsername, .invalidPassword, .unauthenticated:
+ true
+ default:
+ false
+ }
+ }
+}
diff --git a/foreign/swift/Sources/Iggy/Models/TopicOptions.swift b/foreign/swift/Sources/Iggy/Models/TopicOptions.swift
index ab24e77c0..33b6f6620 100644
--- a/foreign/swift/Sources/Iggy/Models/TopicOptions.swift
+++ b/foreign/swift/Sources/Iggy/Models/TopicOptions.swift
@@ -86,6 +86,9 @@ public struct TopicCreateOptions: Sendable, Hashable {
self.raw = raw
}
+ /// The default partition count when none is given.
+ public static let defaultPartitionsCount: UInt32 = 1
+
func toResourceOptions() throws -> ResourceOptions {
var options = try TopicUpdateOptions(
compressionAlgorithm: compressionAlgorithm, messageExpiry: messageExpiry, maxTopicSize: maxTopicSize, raw: raw
diff --git a/foreign/swift/Sources/Iggy/Protocol/CommandCode.swift b/foreign/swift/Sources/Iggy/Protocol/CommandCode.swift
index db853ac59..99673ea0a 100644
--- a/foreign/swift/Sources/Iggy/Protocol/CommandCode.swift
+++ b/foreign/swift/Sources/Iggy/Protocol/CommandCode.swift
@@ -135,6 +135,18 @@ public enum CommandCode: UInt32, Sendable, Hashable, CaseIterable {
}
}
+ /// Codes that drive the session handshake and therefore cannot be sent
+ /// through the raw request path.
+ static func isSessionControl(_ code: UInt32) -> Bool {
+ switch CommandCode(rawValue: code) {
+ case .loginUser, .logoutUser, .loginRegister, .loginWithPersonalAccessToken,
+ .loginRegisterWithPersonalAccessToken:
+ true
+ default:
+ false
+ }
+ }
+
/// Codes that carry the sign-in handshake.
static func isRegister(_ code: UInt32) -> Bool {
code == CommandCode.loginRegister.rawValue
diff --git a/foreign/swift/Sources/Iggy/Protocol/Requests.swift b/foreign/swift/Sources/Iggy/Protocol/Requests.swift
index 8f3d6c784..d75f409a7 100644
--- a/foreign/swift/Sources/Iggy/Protocol/Requests.swift
+++ b/foreign/swift/Sources/Iggy/Protocol/Requests.swift
@@ -268,11 +268,15 @@ enum Requests {
return writer.bytes
}
- static func loginRegisterWithPersonalAccessToken(token: String, bindSecret: [UInt8]) throws -> [UInt8] {
+ static func validatePersonalAccessToken(_ token: String) throws {
let length = token.utf8.count
guard length >= 1, length <= Identifier.maxNameLength else {
throw IggyError(.invalidFormat, context: "token must be 1-255 bytes")
}
+ }
+
+ static func loginRegisterWithPersonalAccessToken(token: String, bindSecret: [UInt8]) throws -> [UInt8] {
+ try validatePersonalAccessToken(token)
guard bindSecret.count == bindSecretSize else {
throw IggyError(.invalidFormat, context: "bind secret must be exactly \(bindSecretSize) bytes")
}
diff --git a/foreign/swift/Sources/Iggy/Transport/Endpoint.swift b/foreign/swift/Sources/Iggy/Transport/Endpoint.swift
new file mode 100644
index 000000000..00b382b3b
--- /dev/null
+++ b/foreign/swift/Sources/Iggy/Transport/Endpoint.swift
@@ -0,0 +1,155 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+import Foundation
+import NIOCore
+import NIOPosix
+
+/// A `host:port` pair, with IPv6 hosts in brackets.
+struct Endpoint: Sendable, Hashable, CustomStringConvertible {
+ let host: String
+ let port: Int
+
+ /// Parses `host:port` or `[v6]:port`.
+ init(parsing address: String) throws {
+ let trimmed = address.trimmingCharacters(in: .whitespaces)
+ guard let colon = trimmed.lastIndex(of: ":") else {
+ throw IggyError(.invalidServerAddress, context: address)
+ }
+ var hostPart = trimmed[..<colon]
+ let portPart = trimmed[trimmed.index(after: colon)...]
+ if hostPart.hasPrefix("["), hostPart.hasSuffix("]") {
+ hostPart = hostPart.dropFirst().dropLast()
+ } else if hostPart.contains(":") {
+ // An IPv6 host without brackets cannot be told from its port.
+ throw IggyError(.invalidServerAddress, context: address)
+ }
+ guard !hostPart.isEmpty, portPart.allSatisfy({ $0.isASCII && $0.isNumber }), let port = UInt16(portPart), port > 0 else {
+ throw IggyError(.invalidServerAddress, context: address)
+ }
+ host = String(hostPart)
+ self.port = Int(port)
+ }
+
+ var description: String {
+ host.contains(":") ? "[\(host)]:\(port)" : "\(host):\(port)"
+ }
+
+ /// Whether the host is a literal IP address rather than a name.
+ var isIPLiteral: Bool { Self.isIPLiteral(host) }
+
+ static func isIPLiteral(_ host: String) -> Bool {
+ (try? SocketAddress(ipAddress: host, port: 0)) != nil
+ }
+}
+
+/// Whether two addresses are written the same way, up to canonicalization
+/// of the host and port. Cheap and non-blocking, so every comparison starts
+/// here.
+func isSameSpelling(_ lhs: String, _ rhs: String) -> Bool {
+ guard let left = try? Endpoint(parsing: lhs), let right = try? Endpoint(parsing: rhs) else {
+ return lhs == rhs
+ }
+ return left.port == right.port && canonicalHost(left.host) == canonicalHost(right.host)
+}
+
+private func canonicalHost(_ host: String) -> String {
+ switch host.lowercased() {
+ case "localhost", "127.0.0.1", "::1", "0.0.0.0", "::": "localhost"
+ default: host.lowercased()
+ }
+}
+
+/// How long a name lookup may take before the comparison gives up on it.
+private let resolveTimeout: Duration = .seconds(2)
+
+/// Whether two addresses name the same endpoint once host names resolve.
+///
+/// A host name and the address it resolves to are one endpoint: a client
+/// configured as `iggy-server:8090` whose roster advertises `10.0.0.5:8090`
+/// would otherwise redirect to the node it is already on. Tests pass their own
+/// `resolve` so the outcome does not depend on the machine's resolver.
+func isSameAddress(_ lhs: String, _ rhs: String, resolve: @Sendable (String) async -> Set<String>? = resolveAddresses) async -> Bool {
+ if isSameSpelling(lhs, rhs) {
+ return true
+ }
+ guard let left = try? Endpoint(parsing: lhs), let right = try? Endpoint(parsing: rhs), left.port == right.port else {
+ return false
+ }
+ // Two literals that did not compare equal are different endpoints.
+ if left.isIPLiteral && right.isIPLiteral {
+ return false
+ }
+ guard let leftAddresses = await resolve(left.host), let rightAddresses = await resolve(right.host) else {
+ return false
+ }
+ return !leftAddresses.isDisjoint(with: rightAddresses)
+}
+
+/// The numeric addresses of a host, or nil when it has none or the resolver
+/// does not answer within ``resolveTimeout``.
+///
+/// `getaddrinfo` blocks and has no deadline of its own. It runs on the
+/// blocking pool so a slow resolver cannot stall the tasks that share a
+/// cooperative thread, and the caller stops waiting when the deadline fires.
+func resolveAddresses(of host: String) async -> Set<String>? {
+ let loop = MultiThreadedEventLoopGroup.singleton.any()
+ let lookup = NIOThreadPool.singleton.runIfActive(eventLoop: loop) { numericAddresses(of: host) }
+ let deadline = loop.scheduleTask(in: .nanoseconds(resolveTimeout.nanosecondsClamped)) { Set<String>() }
+ // Cancelling the deadline fails its future, which hands the result over
+ // to the lookup. A deadline that fires first answers with no addresses.
+ lookup.whenComplete { _ in deadline.cancel() }
+ let addresses = try? await deadline.futureResult.flatMapError { _ in lookup }.get()
+ return addresses.flatMap { $0.isEmpty ? nil : $0 }
+}
+
+private func numericAddresses(of host: String) -> Set<String> {
+ var hints = addrinfo()
+ #if canImport(Glibc) || canImport(Musl)
+ hints.ai_socktype = Int32(SOCK_STREAM.rawValue)
+ #else
+ hints.ai_socktype = SOCK_STREAM
+ #endif
+ var result: UnsafeMutablePointer<addrinfo>?
+ guard getaddrinfo(host, nil, &hints, &result) == 0, let list = result else {
+ return []
+ }
+ defer { freeaddrinfo(list) }
+ var addresses = Set<String>()
+ var current: UnsafeMutablePointer<addrinfo>? = list
+ while let info = current {
+ var buffer = [CChar](repeating: 0, count: Int(NI_MAXHOST))
+ if getnameinfo(info.pointee.ai_addr, info.pointee.ai_addrlen, &buffer, socklen_t(buffer.count), nil, 0, NI_NUMERICHOST) == 0 {
+ addresses.insert(String(decoding: buffer.prefix { $0 != 0 }.map { UInt8(bitPattern: $0) }, as: UTF8.self))
+ }
+ current = info.pointee.ai_next
+ }
+ return addresses
+}
+
+extension Duration {
+ /// The duration in whole nanoseconds, saturating instead of overflowing.
+ var nanosecondsClamped: Int64 {
+ let (seconds, attoseconds) = components
+ let whole = seconds.multipliedReportingOverflow(by: 1_000_000_000)
+ let sum = whole.partialValue.addingReportingOverflow(attoseconds / 1_000_000_000)
+ if whole.overflow || sum.overflow {
+ return seconds < 0 ? Int64.min : Int64.max
+ }
+ return sum.partialValue
+ }
+}
diff --git a/foreign/swift/Sources/Iggy/Transport/TCPConnection.swift b/foreign/swift/Sources/Iggy/Transport/TCPConnection.swift
new file mode 100644
index 000000000..12cd117fe
--- /dev/null
+++ b/foreign/swift/Sources/Iggy/Transport/TCPConnection.swift
@@ -0,0 +1,295 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+import Foundation
+import NIOCore
+import NIOPosix
+import NIOSSL
+import NIOTLS
+
+/// One complete consensus frame as read off the wire.
+struct RawFrame: Sendable {
+ let header: [UInt8]
+ let body: [UInt8]
+}
+
+/// A transport that carries consensus frames to one endpoint.
+///
+/// The protocol is lockstep: one request is written, exactly one reply is
+/// read. The client serializes callers; the connection only moves bytes.
+protocol IggyConnection: AnyObject, Sendable {
+ var localAddress: String? { get }
+ /// Writes a frame and reads the reply, bounded by `timeout`. A timeout
+ /// leaves the stream at an unknown boundary, so the connection closes
+ /// itself and every later call fails.
+ func exchange(_ frame: [UInt8], timeout: Duration) async throws -> RawFrame
+ func close() async
+}
+
+/// A dialer that produces connections; swapped out by tests.
+protocol IggyConnector: Sendable {
+ func connect(to endpoint: Endpoint, tls: TLSOptions?, timeout: Duration, noDelay: Bool) async throws -> any IggyConnection
+}
+
+/// The NIO-backed dialer used in production.
+struct NIOConnector: IggyConnector {
+ func connect(to endpoint: Endpoint, tls: TLSOptions?, timeout: Duration, noDelay: Bool) async throws -> any IggyConnection {
+ try await TCPConnection.connect(to: endpoint, tls: tls, timeout: timeout, noDelay: noDelay)
+ }
+}
+
+/// A TCP connection, optionally wrapped in TLS, speaking the consensus
+/// framing. Backed by SwiftNIO so the same code runs on Apple platforms and
+/// Linux.
+final class TCPConnection: IggyConnection, @unchecked Sendable {
+ private let channel: Channel
+ private let handler: ExchangeHandler
+ let localAddress: String?
+
+ private init(channel: Channel, handler: ExchangeHandler) {
+ self.channel = channel
+ self.handler = handler
+ localAddress = channel.localAddress.map(Self.describe)
+ }
+
+ private static func describe(_ address: SocketAddress) -> String {
+ switch address {
+ case .v4(let v4): "\(v4.host):\(address.port ?? 0)"
+ case .v6(let v6): "[\(v6.host)]:\(address.port ?? 0)"
+ case .unixDomainSocket: address.description
+ }
+ }
+
+ static func connect(to endpoint: Endpoint, tls: TLSOptions?, timeout: Duration, noDelay: Bool) async throws -> TCPConnection {
+ let sslContext: NIOSSLContext?
+ if let tls {
+ sslContext = try tls.makeContext()
+ } else {
+ sslContext = nil
+ }
+ let handler = ExchangeHandler()
+ // One deadline bounds the whole dial, the TLS handshake included.
+ let deadline = ContinuousClock.now + timeout
+ // TCP_NODELAY lives at the TCP level; the socket-level shorthand would
+ // set SO_DEBUG instead, which Linux refuses without CAP_NET_ADMIN.
+ let noDelayValue: ChannelOptions.Types.SocketOption.Value = noDelay ? 1 : 0
+ let bootstrap = ClientBootstrap(group: MultiThreadedEventLoopGroup.singleton)
+ .channelOption(.tcpOption(.tcp_nodelay), value: noDelayValue)
+ .connectTimeout(.nanoseconds(timeout.nanosecondsClamped))
+ .channelInitializer { channel in
+ channel.eventLoop.makeCompletedFuture {
+ if let sslContext {
+ let serverHostname = tls?.serverName(for: endpoint)
+ let sslHandler = try NIOSSLClientHandler(context: sslContext, serverHostname: serverHostname)
+ try channel.pipeline.syncOperations.addHandler(sslHandler)
+ }
+ try channel.pipeline.syncOperations.addHandler(ByteToMessageHandler(FrameDecoder()))
+ try channel.pipeline.syncOperations.addHandler(handler)
+ }
+ }
+ let channel: Channel
+ do {
+ channel = try await bootstrap.connect(host: endpoint.host, port: endpoint.port).get()
+ } catch {
+ throw IggyError(.cannotEstablishConnection, context: "\(endpoint): \(error)")
+ }
+ if sslContext != nil {
+ // The handshake completes asynchronously after connect; waiting for
+ // it here means a failed handshake is reported by the dial rather
+ // than by the first request.
+ do {
+ try await handler.handshakeDone(on: channel, within: deadline - ContinuousClock.now)
+ } catch {
+ try? await channel.close().get()
+ throw IggyError(.cannotEstablishConnection, context: "TLS handshake with \(endpoint) failed: \(error)")
+ }
+ }
+ return TCPConnection(channel: channel, handler: handler)
+ }
+
+ func exchange(_ frame: [UInt8], timeout: Duration) async throws -> RawFrame {
+ let channel = self.channel
+ let handler = self.handler
+ return try await withTaskCancellationHandler {
+ try await withCheckedThrowingContinuation { (continuation: CheckedContinuation<RawFrame, Error>) in
+ channel.eventLoop.execute {
+ handler.send(frame, timeout: timeout, on: channel, replyTo: continuation)
+ }
+ }
+ } onCancel: {
+ // The caller stopped waiting, and a reply that still arrives would
+ // be read as the answer to the next request.
+ channel.eventLoop.execute {
+ handler.abort(with: IggyError(.disconnected, context: "the request was cancelled"), on: channel)
+ }
+ }
+ }
+
+ func close() async {
+ try? await channel.close().get()
+ }
+
+ /// A client dropped without a shutdown must not leave its socket open.
+ deinit {
+ channel.close(promise: nil)
+ }
+}
+
+/// Splits the byte stream into frames: a fixed 256-byte header whose size
+/// field says how many body bytes follow.
+final class FrameDecoder: ByteToMessageDecoder {
+ typealias InboundOut = RawFrame
+
+ func decode(context: ChannelHandlerContext, buffer: inout ByteBuffer) throws -> DecodingState {
+ guard let header = buffer.getBytes(at: buffer.readerIndex, length: VSRFrame.headerSize) else {
+ return .needMoreData
+ }
+ // Compared as `UInt32` so a hostile size never traps a 32-bit `Int`.
+ guard let size = VSRFrame.readSize(header[...]), size >= UInt32(VSRFrame.headerSize), size <= UInt32(VSRFrame.maxFrameSize) else {
+ throw IggyError(.invalidCommand, context: "reply declares an invalid frame size")
+ }
+ guard var frame = buffer.readSlice(length: Int(size)) else {
+ return .needMoreData
+ }
+ frame.moveReaderIndex(forwardBy: VSRFrame.headerSize)
+ let body = frame.readBytes(length: frame.readableBytes) ?? []
+ context.fireChannelRead(wrapInboundOut(RawFrame(header: header, body: body)))
+ return .continue
+ }
+
+ func decodeLast(context: ChannelHandlerContext, buffer: inout ByteBuffer, seenEOF: Bool) throws -> DecodingState {
+ try decode(context: context, buffer: &buffer)
+ }
+}
+
+/// Hands each frame to the one caller waiting for it. All access happens on
+/// the channel's event loop.
+final class ExchangeHandler: ChannelInboundHandler, @unchecked Sendable {
+ typealias InboundIn = RawFrame
+
+ private var pending: CheckedContinuation<RawFrame, Error>?
+ private var replyDeadline: Scheduled<Void>?
+ private var handshake: CheckedContinuation<Void, Error>?
+ private var handshakeDeadline: Scheduled<Void>?
+ private var handshakeComplete = false
+ private var closed = false
+ private var closeError: IggyError?
+
+ /// Writes one request and arms the deadline of its reply.
+ func send(_ frame: [UInt8], timeout: Duration, on channel: Channel, replyTo continuation: CheckedContinuation<RawFrame, Error>) {
+ if closed {
+ continuation.resume(throwing: closeError ?? IggyError(.disconnected, context: "connection closed"))
+ return
+ }
+ if let previous = pending {
+ replyDeadline?.cancel()
+ previous.resume(throwing: IggyError(.invalidCommand, context: "a reply was already awaited on this connection"))
+ }
+ pending = continuation
+ replyDeadline = channel.eventLoop.scheduleTask(in: .nanoseconds(timeout.nanosecondsClamped)) { [self] in
+ abort(with: IggyError(.disconnected, context: "timed out after \(timeout) waiting for the reply"), on: channel)
+ }
+ var buffer = channel.allocator.buffer(capacity: frame.count)
+ buffer.writeBytes(frame)
+ channel.writeAndFlush(buffer).whenFailure { [self] error in
+ abort(with: IggyError(.disconnected, context: "write failed: \(error)"), on: channel)
+ }
+ }
+
+ /// Fails the request in flight and closes the channel. A reply that was
+ /// lost, late, or abandoned leaves the stream at an unknown boundary, so
+ /// the connection cannot carry another request.
+ func abort(with error: IggyError, on channel: Channel) {
+ if !closed {
+ closed = true
+ closeError = error
+ }
+ fail(with: error)
+ endHandshake(with: .failure(error))
+ channel.close(promise: nil)
+ }
+
+ private func fail(with error: IggyError) {
+ replyDeadline?.cancel()
+ replyDeadline = nil
+ pending?.resume(throwing: error)
+ pending = nil
+ }
+
+ /// Waits for the TLS handshake for at most `timeout`. A peer that accepts
+ /// the connection and never answers the ClientHello must not hold the
+ /// dial open.
+ func handshakeDone(on channel: Channel, within timeout: Duration) async throws {
+ try await withCheckedThrowingContinuation { (continuation: CheckedContinuation<Void, Error>) in
+ channel.eventLoop.execute { [self] in
+ if handshakeComplete {
+ continuation.resume()
+ } else if closed {
+ continuation.resume(throwing: closeError ?? IggyError(.disconnected))
+ } else {
+ handshake = continuation
+ handshakeDeadline = channel.eventLoop.scheduleTask(in: .nanoseconds(timeout.nanosecondsClamped)) { [self] in
+ abort(with: IggyError(.disconnected, context: "the handshake did not complete in time"), on: channel)
+ }
+ }
+ }
+ }
+ }
+
+ private func endHandshake(with result: Result<Void, Error>) {
+ handshakeDeadline?.cancel()
+ handshakeDeadline = nil
+ handshake?.resume(with: result)
+ handshake = nil
+ }
+
+ func channelRead(context: ChannelHandlerContext, data: NIOAny) {
+ let frame = unwrapInboundIn(data)
+ if let pending {
+ replyDeadline?.cancel()
+ replyDeadline = nil
+ self.pending = nil
+ pending.resume(returning: frame)
+ }
+ // A frame nobody waits for is either a stray reply or a session-level
+ // eviction; the next exchange discovers the closed channel either way.
+ }
+
+ func userInboundEventTriggered(context: ChannelHandlerContext, event: Any) {
+ if case .handshakeCompleted = event as? TLSUserEvent {
+ handshakeComplete = true
+ endHandshake(with: .success(()))
+ }
+ context.fireUserInboundEventTriggered(event)
+ }
+
+ func errorCaught(context: ChannelHandlerContext, error: Error) {
+ let mapped = (error as? IggyError) ?? IggyError(.disconnected, context: "\(error)")
+ closeError = mapped
+ endHandshake(with: .failure(mapped))
+ fail(with: mapped)
+ context.close(promise: nil)
+ }
+
+ func channelInactive(context: ChannelHandlerContext) {
+ closed = true
+ let error = closeError ?? IggyError(.disconnected, context: "connection closed by the peer")
+ endHandshake(with: .failure(error))
+ fail(with: error)
+ context.fireChannelInactive()
+ }
+}
diff --git a/foreign/swift/Sources/Iggy/Transport/TLSOptions.swift b/foreign/swift/Sources/Iggy/Transport/TLSOptions.swift
new file mode 100644
index 000000000..66938bd4c
--- /dev/null
+++ b/foreign/swift/Sources/Iggy/Transport/TLSOptions.swift
@@ -0,0 +1,67 @@
+// 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 NIOSSL
+
+/// TLS settings of a TCP connection.
+public struct TLSOptions: Sendable, Hashable {
+ /// Server name presented for SNI and checked against the certificate.
+ /// Defaults to the host part of the server address.
+ public var domain: String?
+ /// PEM file with the certificate authority to trust instead of the
+ /// system roots, for servers with a private CA such as the test server.
+ public var caFile: String?
+ /// Whether to verify the server certificate at all. Turning this off
+ /// makes the connection vulnerable to interception; only for development.
+ public var validateCertificate: Bool
+
+ public init(domain: String? = nil, caFile: String? = nil, validateCertificate: Bool = true) {
+ self.domain = domain
+ self.caFile = caFile
+ self.validateCertificate = validateCertificate
+ }
+
+ /// The host name to present during the handshake: the configured domain,
+ /// else the host of the address. An IP literal cannot be used for SNI, so
+ /// it is left out and the certificate is checked against the peer address.
+ func serverName(for endpoint: Endpoint) -> String? {
+ let name = domain.flatMap { $0.isEmpty ? nil : $0 } ?? endpoint.host
+ return Endpoint.isIPLiteral(name) ? nil : name
+ }
+
+ func makeContext() throws -> NIOSSLContext {
+ var configuration = TLSConfiguration.makeClientConfiguration()
+ if validateCertificate {
+ if let caFile {
+ do {
+ let certificates = try NIOSSLCertificate.fromPEMFile(caFile)
+ configuration.trustRoots = .certificates(certificates)
+ } catch {
+ throw IggyError(.invalidTlsCertificatePath, context: "\(caFile): \(error)")
+ }
+ }
+ configuration.certificateVerification = .fullVerification
+ } else {
+ configuration.certificateVerification = .none
+ }
+ do {
+ return try NIOSSLContext(configuration: configuration)
+ } catch {
+ throw IggyError(.invalidTlsCertificate, context: "\(error)")
+ }
+ }
+}
diff --git a/foreign/swift/Sources/Iggy/Utilities/AsyncMutex.swift b/foreign/swift/Sources/Iggy/Utilities/AsyncMutex.swift
new file mode 100644
index 000000000..5da12aab7
--- /dev/null
+++ b/foreign/swift/Sources/Iggy/Utilities/AsyncMutex.swift
@@ -0,0 +1,201 @@
+// 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 NIOConcurrencyHelpers
+
+/// 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.
+final class AsyncMutex: Sendable {
+ private struct State {
+ var locked = false
+ var waiters: [(id: UInt64, continuation: CheckedContinuation<Void, Error>)] = []
+ var nextID: UInt64 = 0
+ }
+
+ private let state = NIOLockedValueBox(State())
+
+ /// Waits for the lock. A caller cancelled while it waits leaves the queue
+ /// with `CancellationError` and does not hold the lock.
+ func lock() async throws {
+ let id = state.withLockedValue { state -> UInt64 in
+ state.nextID &+= 1
+ return state.nextID
+ }
+ try await withTaskCancellationHandler {
+ try await withCheckedThrowingContinuation { (continuation: CheckedContinuation<Void, Error>) in
+ let outcome: Result<Void, Error>? = state.withLockedValue { state in
+ if Task.isCancelled {
+ return .failure(CancellationError())
+ }
+ if !state.locked {
+ state.locked = true
+ return .success(())
+ }
+ state.waiters.append((id, continuation))
+ return nil
+ }
+ if let outcome {
+ continuation.resume(with: outcome)
+ }
+ }
+ } onCancel: {
+ // A waiter the lock was already handed to is no longer queued and
+ // keeps it; only a caller still waiting is taken out.
+ let waiter = state.withLockedValue { state -> CheckedContinuation<Void, Error>? in
+ guard let index = state.waiters.firstIndex(where: { $0.id == id }) else {
+ return nil
+ }
+ return state.waiters.remove(at: index).continuation
+ }
+ waiter?.resume(throwing: CancellationError())
+ }
+ }
+
+ /// Hands the lock to the longest waiter, or frees it.
+ func unlock() {
+ let next = state.withLockedValue { state -> CheckedContinuation<Void, Error>? in
+ if state.waiters.isEmpty {
+ state.locked = false
+ return nil
+ }
+ return state.waiters.removeFirst().continuation
+ }
+ next?.resume()
+ }
+}
+
+/// A single-flight gate: the first caller starts the work, every caller that
+/// arrives meanwhile awaits the same outcome.
+///
+/// A cancelled caller leaves at once with `CancellationError`. The work is
+/// cancelled when its last caller has left, and callers that arrive while it
+/// unwinds start a new flight once it has ended.
+final class SingleFlight: Sendable {
+ typealias Work = @Sendable () async throws -> Void
+ private typealias Caller = CheckedContinuation<Void, Error>
+
+ private struct State {
+ var flight: Task<Void, Never>?
+ /// The flight lost its last caller and is unwinding.
+ var abandoned = false
+ var callers: [UInt64: Caller] = [:]
+ var queued: [UInt64: (caller: Caller, work: Work)] = [:]
+ var nextID: UInt64 = 0
+ }
+
+ private let state = NIOLockedValueBox(State())
+
+ var isActive: Bool {
+ state.withLockedValue { $0.flight != nil }
+ }
+
+ func run(_ work: @escaping Work) async throws {
+ let id = state.withLockedValue { state -> UInt64 in
+ state.nextID &+= 1
+ return state.nextID
+ }
+ try await withTaskCancellationHandler {
+ try await withCheckedThrowingContinuation { (caller: Caller) in
+ let cancelled = state.withLockedValue { state -> Bool in
+ if Task.isCancelled {
+ return true
+ }
+ if state.flight == nil {
+ state.callers[id] = caller
+ state.flight = Task { await self.fly(work) }
+ } else if state.abandoned {
+ state.queued[id] = (caller, work)
+ } else {
+ state.callers[id] = caller
+ }
+ return false
+ }
+ if cancelled {
+ caller.resume(throwing: CancellationError())
+ }
+ }
+ } onCancel: {
+ leave(id)
+ }
+ }
+
+ private func fly(_ work: Work) async {
+ let result: Result<Void, Error>
+ do {
+ try await work()
+ result = .success(())
+ } catch {
+ result = .failure(error)
+ }
+ let finished = state.withLockedValue { state -> [Caller] in
+ let finished = Array(state.callers.values)
+ state.callers.removeAll()
+ state.abandoned = false
+ state.flight = nil
+ // Callers that arrived while an abandoned flight was unwinding
+ // share the next one, started by the earliest of them.
+ if let first = state.queued.min(by: { $0.key < $1.key }) {
+ for (id, entry) in state.queued {
+ state.callers[id] = entry.caller
+ }
+ state.queued.removeAll()
+ let next = first.value.work
+ state.flight = Task { await self.fly(next) }
+ }
+ return finished
+ }
+ for caller in finished {
+ caller.resume(with: result)
+ }
+ }
+
+ private func leave(_ id: UInt64) {
+ let (caller, abandonedFlight) = state.withLockedValue { state -> (Caller?, Task<Void, Never>?) in
+ if let queued = state.queued.removeValue(forKey: id) {
+ return (queued.caller, nil)
+ }
+ guard let caller = state.callers.removeValue(forKey: id) else {
+ return (nil, nil)
+ }
+ guard state.callers.isEmpty, !state.abandoned else {
+ return (caller, nil)
+ }
+ state.abandoned = true
+ return (caller, state.flight)
+ }
+ caller?.resume(throwing: CancellationError())
+ abandonedFlight?.cancel()
+ }
+}
+
+/// Runs `body` until `deadline` and returns nil when the deadline passes
+/// first. `body` is cancelled then, and the wait ends once it honours that.
+func withDeadline<T: Sendable>(_ deadline: ContinuousClock.Instant, _ body: @escaping @Sendable () async throws -> T) async throws -> T? {
+ try await withThrowingTaskGroup(of: T?.self) { group in
+ defer { group.cancelAll() }
+ group.addTask { try await body() }
+ group.addTask {
+ try await Task.sleep(until: deadline, clock: .continuous)
+ return nil
+ }
+ return try await group.next() ?? nil
+ }
+}
diff --git a/foreign/swift/Tests/IggyTests/ClientTransportTests.swift b/foreign/swift/Tests/IggyTests/ClientTransportTests.swift
new file mode 100644
index 000000000..7dbd0d802
--- /dev/null
+++ b/foreign/swift/Tests/IggyTests/ClientTransportTests.swift
@@ -0,0 +1,1168 @@
+// 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 Testing
+
+@testable import Iggy
+
+private let rootLogin = Credentials.usernamePassword(username: "iggy", password: "iggy")
+
+/// A request timeout short enough for a test to wait it out.
+private let shortTimeout: Duration = .milliseconds(500)
+
+/// A client on the fake server with fast reconnection. The request timeout is
+/// generous unless the test is about running into it, so a slow machine does
+/// not cut a reconnect short.
+private func makeClient(
+ _ server: FakeVSRServer, autoLogin: Credentials? = rootLogin, reconnection: Bool = true, maxRetries: UInt32? = nil,
+ requestTimeout: Duration = .seconds(10), heartbeat: Duration = .seconds(60), connector: any IggyConnector = NIOConnector()
+) -> IggyClient {
+ var configuration = ClientConfiguration(address: server.address, autoLogin: autoLogin)
+ configuration.reconnection = ReconnectionOptions(enabled: reconnection, maxRetries: maxRetries, interval: .milliseconds(20), reestablishAfter: .zero)
+ configuration.requestTimeout = requestTimeout
+ configuration.connectTimeout = .seconds(2)
+ configuration.heartbeatInterval = heartbeat
+ return IggyClient(configuration: configuration, connector: connector, logger: testLogger)
+}
+
+/// A dialer that holds every dial until it is released, to stop a client in
+/// the middle of one.
+private struct GatedConnector: IggyConnector {
+ let released = Locked(false)
+
+ func connect(to endpoint: Endpoint, tls: TLSOptions?, timeout: Duration, noDelay: Bool) async throws -> any IggyConnection {
+ while !released.withLock({ $0 }) {
+ try await Task.sleep(for: .milliseconds(2))
+ }
+ return try await NIOConnector().connect(to: endpoint, tls: tls, timeout: timeout, noDelay: noDelay)
+ }
+}
+
+private func operations(_ server: FakeVSRServer) async -> [ConsensusOperation] {
+ await server.requests.map(\.operation)
+}
+
+private func codes(_ server: FakeVSRServer) async -> [UInt32?] {
+ await server.requests.map(\.code)
+}
+
+private func pingCount(_ server: FakeVSRServer) async -> Int {
+ await codes(server).filter { $0 == CommandCode.ping.rawValue }.count
+}
+
+@Suite("Client transport", .serialized)
+struct ClientTransportTests {
+ @Test func loginBindsTheSessionAndSequencesRequests() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ let client = makeClient(server)
+ try await client.connect()
+ #expect(await client.state == .authenticated)
+ try await client.ping()
+ try await client.deleteStream(1)
+ try await client.deleteStream(2)
+ try await client.ping()
+ let requests = await server.requests
+ #expect(requests.map(\.operation) == [.register, .nonReplicated, .nonReplicated, .deleteStream, .deleteStream, .nonReplicated])
+ // The register runs unbound with request id 0; the roster read that
+ // follows is non-replicated and never advances the watermark, which
+ // the replicated creates then consume as 1 and 2.
+ #expect(requests[0].session == 0 && requests[0].requestID == 0 && requests[0].checksum != 0)
+ #expect(requests[1].code == CommandCode.getClusterMetadata.rawValue)
+ #expect(requests[2].code == CommandCode.ping.rawValue && requests[2].session == 1 && requests[2].requestID == 1)
+ #expect(requests[3].requestID == 1 && requests[3].session == 1 && requests[3].checksum == XXH3.hash64(requests[3].payload))
+ #expect(requests[4].requestID == 2)
+ #expect(requests[5].requestID == 3 && requests[5].checksum == 0)
+ #expect(requests.allSatisfy { $0.clientID == requests[0].clientID })
+ #expect(await client.serverAddress == server.address)
+ #expect(await client.clientAddress?.hasPrefix("127.0.0.1:") == true)
+ await client.shutdown()
+ }
+
+ @Test func manualLoginWithoutAutoLogin() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ let client = makeClient(server, autoLogin: nil)
+ try await client.connect()
+ #expect(await client.state == .connected)
+ try await client.ping()
+ await expectCode(.unauthenticated) { _ = try await client.getStreams() }
+ let identity = try await client.login(username: "iggy", password: "iggy")
+ #expect(identity.userID == 0)
+ #expect(await client.state == .authenticated)
+ // A re-login on a bound session logs out first, then registers under
+ // a fresh client id.
+ let before = await server.requests.last!.clientID
+ _ = try await client.login(username: "iggy", password: "iggy")
+ let ops = await operations(server)
+ #expect(ops.suffix(3) == [.logout, .register, .nonReplicated])
+ #expect(await server.requests.last!.clientID != before)
+ try await client.logout()
+ #expect(await client.state == .connected)
+ await client.shutdown()
+ }
+
+ @Test func rejectedCredentialsAreNotRetried() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ await server.respond { request in
+ request.operation == .register ? .eviction(.invalidCredentials) : .ok([])
+ }
+ let client = makeClient(server, autoLogin: nil)
+ try await client.connect()
+ await expectCode(.invalidCredentials) { _ = try await client.login(username: "iggy", password: "wrong") }
+ // The eviction ended the connection, so the client is back to
+ // disconnected rather than half signed in.
+ #expect(await client.state != .authenticated)
+ #expect(await operations(server).filter { $0 == .register }.count == 1)
+ await client.shutdown()
+ }
+
+ @Test func committedRegisterRejectionIsTyped() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ await server.respond { request in
+ request.operation == .register ? .committedRejection(IggyErrorCode.userInactive.rawValue) : .ok([])
+ }
+ let client = makeClient(server, autoLogin: nil)
+ try await client.connect()
+ await expectCode(.userInactive) { _ = try await client.login(username: "iggy", password: "iggy") }
+ await client.shutdown()
+ }
+
+ @Test func preCommitDenialsAreTyped() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ await server.respond { request in
+ switch request.operation {
+ case .register: .ok(FakeVSRServer.registerBody())
+ case .createStream: .status(IggyErrorCode.streamNameAlreadyExists.rawValue)
+ case .nonReplicated where request.code == CommandCode.getStats.rawValue: .committedRejection(0)
+ default: .ok([])
+ }
+ }
+ let client = makeClient(server)
+ try await client.connect()
+ await expectCode(.streamNameAlreadyExists) { _ = try await client.createStream(name: "dup") }
+ // The session survives a denial: the next request still goes out.
+ try await client.ping()
+ #expect(await client.state == .authenticated)
+ await client.shutdown()
+ }
+
+ @Test func transientNotCommittedReplaysTheSameRequestID() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ let counter = Locked(0)
+ await server.respond { request in
+ if request.operation == .register { return .ok(FakeVSRServer.registerBody()) }
+ if request.operation == .deleteStream {
+ let seen = counter.withLock {
+ $0 += 1; return $0
+ }
+ return seen < 3 ? .status(IggyErrorCode.transientNotCommitted.rawValue) : .ok([])
+ }
+ return .ok([])
+ }
+ let client = makeClient(server)
+ try await client.connect()
+ try await client.deleteStream(1)
+ let creates = await server.requests.filter { $0.operation == .deleteStream }
+ #expect(creates.count == 3)
+ #expect(Set(creates.map(\.requestID)).count == 1)
+ #expect(Set(creates.map(\.connection)).count == 1)
+ #expect(await server.connectionCount == 1)
+ await client.shutdown()
+ }
+
+ @Test func transientNotAcceptedReplaysOnASingleNode() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ let counter = Locked(0)
+ await server.respond { request in
+ if request.operation == .register { return .ok(FakeVSRServer.registerBody()) }
+ if request.operation == .nonReplicated, request.code == CommandCode.getClusterMetadata.rawValue {
+ return .ok(FakeVSRServer.rosterBody([server.address]))
+ }
+ if request.operation == .deleteSegments {
+ let seen = counter.withLock {
+ $0 += 1; return $0
+ }
+ return seen < 4 ? .status(IggyErrorCode.transientNotAccepted.rawValue) : .ok([])
+ }
+ return .ok([])
+ }
+ let client = makeClient(server)
+ try await client.connect()
+ try await client.deleteSegments(streamID: 1, topicID: 1, partitionID: 0, count: 1)
+ let sends = await server.requests.filter { $0.operation == .deleteSegments }
+ #expect(sends.count == 4)
+ #expect(Set(sends.map(\.requestID)).count == 1)
+ #expect(await server.connectionCount == 1)
+ await client.shutdown()
+ }
+
+ /// The replay after the failover check is stamped afresh. The refused
+ /// request frees the exchange lock for that check, a caller queued behind
+ /// it spends the next request id, and the server refuses an id older than
+ /// one it already applied.
+ @Test func replayAfterTheFailoverCheckTakesANewRequestID() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ let applied = Locked<UInt64>(0)
+ await server.respond { request in
+ if request.operation == .register { return .ok(FakeVSRServer.registerBody()) }
+ if request.operation == .nonReplicated, request.code == CommandCode.getClusterMetadata.rawValue {
+ return .ok(FakeVSRServer.rosterBody([server.address]))
+ }
+ switch request.operation {
+ case .deleteStream:
+ applied.withLock { $0 = max($0, request.requestID) }
+ return .ok([])
+ case .deleteTopic:
+ let highest = applied.withLock { $0 }
+ if highest == 0 {
+ return .status(IggyErrorCode.transientNotAccepted.rawValue)
+ }
+ return request.requestID < highest ? .status(IggyErrorCode.requestTooOld.rawValue) : .ok([])
+ default:
+ return .ok([])
+ }
+ }
+ let client = makeClient(server)
+ try await client.connect()
+ let refused = Task { try await client.deleteTopic(streamID: 1, topicID: 1) }
+ #expect(await eventually { await operations(server).contains(.deleteTopic) })
+ try await client.deleteStream(1)
+ try await refused.value
+ let attempts = await server.requests.filter { $0.operation == .deleteTopic }.map(\.requestID)
+ let queued = await server.requests.first { $0.operation == .deleteStream }?.requestID ?? 0
+ #expect(attempts.first.map { $0 < queued } == true)
+ #expect(attempts.last.map { $0 > queued } == true)
+ #expect(await server.connectionCount == 1)
+ await client.shutdown()
+ }
+
+ /// After a view change a backup refuses a replicated write with "not
+ /// primary". The write was never admitted, so it goes to the new primary
+ /// at once instead of being replayed on the backup.
+ @Test func notPrimaryMovesTheRequestToTheNewPrimary() async throws {
+ let first = try await FakeVSRServer.started()
+ let second = try await FakeVSRServer.started()
+ defer {
+ Task {
+ await first.stop()
+ await second.stop()
+ }
+ }
+ let addresses = [first.address, second.address]
+ let viewChanged = Locked(false)
+ await first.respond { request in
+ if request.operation == .register { return .ok(FakeVSRServer.registerBody()) }
+ if request.code == CommandCode.getClusterMetadata.rawValue {
+ return .ok(FakeVSRServer.rosterBody(addresses, leaderIndex: viewChanged.withLock { $0 } ? 1 : 0))
+ }
+ guard request.operation == .deleteStream else { return .ok([]) }
+ viewChanged.withLock { $0 = true }
+ return .status(IggyErrorCode.notPrimary.rawValue)
+ }
+ await second.respond { request in
+ if request.operation == .register { return .ok(FakeVSRServer.registerBody()) }
+ if request.code == CommandCode.getClusterMetadata.rawValue { return .ok(FakeVSRServer.rosterBody(addresses, leaderIndex: 1)) }
+ return .ok([])
+ }
+ let client = makeClient(first)
+ try await client.connect()
+ let clock = ContinuousClock()
+ let elapsed = try await clock.measure { try await client.deleteStream(1) }
+ #expect(elapsed < ClientCore.transientFailoverCheckInterval)
+ #expect(await first.requests.filter { $0.operation == .deleteStream }.count == 1)
+ #expect(await second.requests.filter { $0.operation == .deleteStream }.count == 1)
+ #expect(await client.serverAddress == second.address)
+ await client.shutdown()
+ }
+
+ /// With nowhere to move, "not primary" is reported like "not accepted":
+ /// neither was admitted.
+ @Test func notPrimaryWithoutAnotherNodeIsReportedAsNotAccepted() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ await server.respond { request in
+ if request.operation == .register { return .ok(FakeVSRServer.registerBody()) }
+ if request.code == CommandCode.getClusterMetadata.rawValue { return .ok(FakeVSRServer.rosterBody([server.address])) }
+ return request.operation == .deleteStream ? .status(IggyErrorCode.notPrimary.rawValue) : .ok([])
+ }
+ let client = makeClient(server, requestTimeout: shortTimeout)
+ try await client.connect()
+ await expectCode(.transientNotAccepted) { try await client.deleteStream(1) }
+ let attempts = await server.requests.filter { $0.operation == .deleteStream }
+ #expect(attempts.count > 1)
+ #expect(Set(attempts.map(\.requestID)).count == attempts.count)
+ #expect(await client.state == .authenticated)
+ await client.shutdown()
+ }
+
+ @Test func transientReplayGivesUpAtTheDeadline() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ await server.respond { request in
+ request.operation == .register
+ ? .ok(FakeVSRServer.registerBody()) : request.operation == .createStream ? .status(IggyErrorCode.transientNotCommitted.rawValue) : .ok([])
+ }
+ let client = makeClient(server, requestTimeout: shortTimeout)
+ try await client.connect()
+ let clock = ContinuousClock()
+ let elapsed = await clock.measure {
+ await expectCode(.transientNotCommitted) { _ = try await client.createStream(name: "x") }
+ }
+ #expect(elapsed >= .milliseconds(450) && elapsed < .seconds(10))
+ await client.shutdown()
+ }
+
+ /// A reply that never comes spends the whole deadline, so the request
+ /// fails without waiting for a reconnect. The stream is at an unknown
+ /// boundary: the connection is dropped and the next request dials again.
+ @Test func requestTimeoutClosesTheConnectionAndTheNextRequestReconnects() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ let stallOnce = Locked(true)
+ await server.respond { request in
+ if request.operation == .register { return .ok(FakeVSRServer.registerBody()) }
+ if request.operation == .nonReplicated, request.code == CommandCode.getStats.rawValue,
+ stallOnce.withLock({
+ let stall = $0; $0 = false; return stall
+ })
+ {
+ return .stall
+ }
+ return .ok([])
+ }
+ let client = makeClient(server, requestTimeout: shortTimeout)
+ try await client.connect()
+ let clock = ContinuousClock()
+ let elapsed = await clock.measure {
+ await expectCode(.disconnected) { _ = try await client.getStats() }
+ }
+ #expect(elapsed >= .milliseconds(450) && elapsed < .seconds(10))
+ #expect(await client.state == .disconnected)
+ #expect(await server.connectionCount == 1)
+ try await client.ping()
+ #expect(await server.connectionCount == 2)
+ #expect(await operations(server).filter { $0 == .register }.count == 2)
+ #expect(await client.state == .authenticated)
+ await client.shutdown()
+ }
+
+ /// With unlimited retries a reconnect to a server that is gone never
+ /// ends. The request that started it still has to return at its deadline.
+ @Test func requestDeadlineCoversTheReconnect() async throws {
+ let server = try await FakeVSRServer.started()
+ let client = makeClient(server, requestTimeout: shortTimeout)
+ try await client.connect()
+ await server.stop()
+ let clock = ContinuousClock()
+ let elapsed = await clock.measure {
+ await expectCode(.disconnected) { try await client.ping() }
+ }
+ #expect(elapsed >= .milliseconds(450) && elapsed < .seconds(10))
+ // No caller waits for the sweep any more, so it ends too.
+ #expect(await eventually { await client.state == .disconnected })
+ await client.shutdown()
+ }
+
+ /// An attempt answered with "not committed" may have committed. A later
+ /// "not accepted" for the same request does not prove otherwise, so it
+ /// must not send the request to the next node under a new session, where
+ /// the server cannot recognise it as a repeat.
+ @Test func uncertainOutcomeIsNotReissuedOnAnotherNode() async throws {
+ let first = try await FakeVSRServer.started()
+ let second = try await FakeVSRServer.started()
+ defer {
+ Task {
+ await first.stop()
+ await second.stop()
+ }
+ }
+ let roster = FakeVSRServer.rosterBody([first.address, second.address], leaderIndex: 0)
+ let seen = Locked(0)
+ await first.respond { request in
+ if request.operation == .register { return .ok(FakeVSRServer.registerBody()) }
+ if request.code == CommandCode.getClusterMetadata.rawValue { return .ok(roster) }
+ guard request.operation == .createStream else { return .ok([]) }
+ let count = seen.withLock {
+ $0 += 1; return $0
+ }
+ return .status((count == 1 ? IggyErrorCode.transientNotCommitted : .transientNotAccepted).rawValue)
+ }
+ await second.respond { request in
+ if request.operation == .register { return .ok(FakeVSRServer.registerBody()) }
+ if request.code == CommandCode.getClusterMetadata.rawValue { return .ok(roster) }
+ return .ok([])
+ }
+ let client = makeClient(first, requestTimeout: ClientCore.transientFailoverCheckInterval + .seconds(1))
+ try await client.connect()
+ await #expect(throws: IggyError.self) { _ = try await client.createStream(name: "once") }
+ let attempts = await first.requests.filter { $0.operation == .createStream }
+ #expect(attempts.count > 1)
+ #expect(Set(attempts.map(\.requestID)).count == 1)
+ #expect(Set(attempts.map(\.clientID)).count == 1)
+ #expect(await second.requests.filter { $0.operation == .createStream }.isEmpty)
+ await client.shutdown()
+ }
+
+ @Test func misroutedReplyDesyncsAndReconnects() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ let once = Locked(true)
+ await server.respond { request in
+ if request.operation == .register { return .ok(FakeVSRServer.registerBody()) }
+ if request.code == CommandCode.ping.rawValue,
+ once.withLock({
+ let first = $0; $0 = false; return first
+ })
+ {
+ return .wrongRequestID(request.requestID + 7)
+ }
+ return .ok([])
+ }
+ let client = makeClient(server)
+ try await client.connect()
+ try await client.ping()
+ #expect(await server.connectionCount == 2)
+ #expect(await codes(server).filter { $0 == CommandCode.ping.rawValue }.count == 2)
+ await client.shutdown()
+ }
+
+ @Test func evictionRestoresTheSessionAndReplaysSafeRequests() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ let once = Locked(true)
+ await server.respond { request in
+ if request.operation == .register { return .ok(FakeVSRServer.registerBody()) }
+ if request.operation == .deleteStream,
+ once.withLock({
+ let first = $0; $0 = false; return first
+ })
+ {
+ return .eviction(.noSession)
+ }
+ return .ok([])
+ }
+ let client = makeClient(server)
+ try await client.connect()
+ // An unauthenticated eviction means the request never reached the
+ // log, so the delete is replayed under the new session.
+ try await client.deleteStream(1)
+ let ops = await operations(server)
+ #expect(ops.filter { $0 == .register }.count == 2)
+ #expect(ops.filter { $0 == .deleteStream }.count == 2)
+ #expect(await server.connectionCount == 2)
+ await client.shutdown()
+ }
+
+ @Test func staleClientEvictionReconnects() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ let once = Locked(true)
+ await server.respond { request in
+ if request.operation == .register { return .ok(FakeVSRServer.registerBody()) }
+ if request.code == CommandCode.ping.rawValue,
+ once.withLock({
+ let first = $0; $0 = false; return first
+ })
+ {
+ return .eviction(.staleClient)
+ }
+ return .ok([])
+ }
+ let client = makeClient(server)
+ try await client.connect()
+ try await client.ping()
+ #expect(await server.connectionCount == 2)
+ await client.shutdown()
+ }
+
+ @Test func incompatibleProtocolIsTerminal() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ await server.respond { request in
+ request.operation == .register ? .eviction(.incompatibleProtocol) : .ok([])
+ }
+ let client = makeClient(server, autoLogin: nil)
+ try await client.connect()
+ await expectCode(.incompatibleProtocolVersion) { _ = try await client.login(username: "iggy", password: "iggy") }
+ await client.shutdown()
+ }
+
+ @Test func replicatedRequestIsNotReplayedAfterAConnectionDrop() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ let once = Locked(true)
+ await server.respond { request in
+ if request.operation == .register { return .ok(FakeVSRServer.registerBody()) }
+ if request.operation == .createStream,
+ once.withLock({
+ let first = $0; $0 = false; return first
+ })
+ {
+ return .close
+ }
+ return .ok([])
+ }
+ let client = makeClient(server)
+ try await client.connect()
+ // The outcome is unknown: the client reconnects for later requests
+ // but reports the write as possibly committed instead of applying it
+ // twice, so it is not mistaken for a request that was never sent.
+ await expectCode(.transientNotCommitted) { _ = try await client.createStream(name: "x") }
+ #expect(await operations(server).filter { $0 == .createStream }.count == 1)
+ #expect(await client.state == .authenticated)
+ #expect(await server.connectionCount == 2)
+ try await client.ping()
+ await client.shutdown()
+ }
+
+ /// The body of a login is built once, with the bind secret of the session
+ /// that registers. A retry on a new connection has to arrive under that
+ /// same client id, or the server binds the secret to the wrong session.
+ @Test func loginRetriedAfterAConnectionDropKeepsItsIdentityAndSecret() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ let once = Locked(true)
+ await server.respond { request in
+ guard request.operation == .register else { return .ok([]) }
+ let first = once.withLock {
+ let first = $0; $0 = false; return first
+ }
+ return first ? .close : .ok(FakeVSRServer.registerBody())
+ }
+ let client = makeClient(server, autoLogin: nil)
+ try await client.connect()
+ _ = try await client.login(username: "iggy", password: "iggy")
+ #expect(await client.state == .authenticated)
+ let registers = await server.requests.filter { $0.operation == .register }
+ #expect(registers.count == 2)
+ #expect(registers[0].connection != registers[1].connection)
+ #expect(registers[0].clientID == registers[1].clientID)
+ let secrets = registers.compactMap { Self.bindSecret(in: $0.payload) }
+ #expect(secrets.count == 2)
+ #expect(secrets[0] == secrets[1])
+ await client.shutdown()
+ }
+
+ @Test func nonReplicatedRequestIsReplayedAfterAConnectionDrop() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ let once = Locked(true)
+ await server.respond { request in
+ if request.operation == .register { return .ok(FakeVSRServer.registerBody()) }
+ if request.code == CommandCode.getStats.rawValue,
+ once.withLock({
+ let first = $0; $0 = false; return first
+ })
+ {
+ return .close
+ }
+ return .ok([UInt8](repeating: 0, count: 200))
+ }
+ let client = makeClient(server)
+ try await client.connect()
+ _ = try? await client.getStats()
+ #expect(await codes(server).filter { $0 == CommandCode.getStats.rawValue }.count == 2)
+ #expect(await server.connectionCount == 2)
+ await client.shutdown()
+ }
+
+ @Test func reconnectionDisabledSurfacesTheDrop() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ await server.respond { request in
+ request.operation == .register ? .ok(FakeVSRServer.registerBody()) : request.code == CommandCode.ping.rawValue ? .close : .ok([])
+ }
+ let client = makeClient(server, reconnection: false)
+ try await client.connect()
+ await expectCode(.disconnected) { try await client.ping() }
+ #expect(await client.state == .disconnected)
+ #expect(await server.connectionCount == 1)
+ await expectCode(.disconnected) { try await client.ping() }
+ await client.shutdown()
+ }
+
+ @Test func reconnectionDisabledReportsAnUnknownOutcome() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ await server.respond { request in
+ request.operation == .register ? .ok(FakeVSRServer.registerBody()) : request.operation == .deleteStream ? .close : .ok([])
+ }
+ let client = makeClient(server, reconnection: false)
+ try await client.connect()
+ await expectCode(.transientNotCommitted) { try await client.deleteStream(1) }
+ // A typed call that never reaches the wire still says so.
+ await expectCode(.disconnected) { try await client.deleteStream(1) }
+ #expect(await operations(server).filter { $0 == .deleteStream }.count == 1)
+ await client.shutdown()
+ }
+
+ @Test func withoutCredentialsNothingIsReplayed() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ await server.respond { request in
+ request.code == CommandCode.ping.rawValue ? .close : .ok([])
+ }
+ let client = makeClient(server, autoLogin: nil)
+ try await client.connect()
+ await expectCode(.disconnected) { try await client.ping() }
+ #expect(await server.connectionCount == 1)
+ await client.shutdown()
+ }
+
+ @Test func serverInitiatedCloseIsRepairedByTheNextRequest() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ let client = makeClient(server)
+ let events = await client.events
+ try await client.connect()
+ await server.closeAllConnections()
+ try await Task.sleep(for: .milliseconds(20))
+ try await client.ping()
+ #expect(await server.connectionCount == 2)
+ #expect(await operations(server).filter { $0 == .register }.count == 2)
+ await client.shutdown()
+ var seen: [DiagnosticEvent] = []
+ for await event in events {
+ seen.append(event)
+ }
+ #expect(seen == [.connected, .signedIn, .disconnected, .connected, .signedIn, .shutdown])
+ }
+
+ @Test func connectRetriesUntilTheServerIsUp() async throws {
+ let server = FakeVSRServer()
+ try await server.start()
+ await server.state.setAddress(server.address)
+ let address = server.address
+ await server.stop()
+ let client = makeClient(server, maxRetries: 2)
+ let clock = ContinuousClock()
+ let elapsed = await clock.measure {
+ await expectCode(.cannotEstablishConnection) { try await client.connect() }
+ }
+ // Three dials with two 20 ms pauses between them.
+ #expect(elapsed >= .milliseconds(40) && elapsed < .seconds(10))
+ #expect(await client.state == .disconnected)
+
+ // Bring a server back on the same port and connect again.
+ let revived = FakeVSRServer()
+ let port = try Endpoint(parsing: address).port
+ try await revived.startOn(port: port)
+ defer { Task { await revived.stop() } }
+ try await client.connect()
+ #expect(await client.state == .authenticated)
+ await client.shutdown()
+ }
+
+ @Test func passwordChangeUpdatesTheReconnectCredentials() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ let client = makeClient(server)
+ try await client.connect()
+ try await client.changePassword("iggy", currentPassword: "iggy", newPassword: "new-secret")
+ await server.closeAllConnections()
+ try await Task.sleep(for: .milliseconds(20))
+ try await client.ping()
+ let registers = await server.requests.filter { $0.operation == .register }
+ #expect(registers.count == 2)
+ #expect(Self.password(in: registers[0].payload) == "iggy")
+ #expect(Self.password(in: registers[1].payload) == "new-secret")
+ // Each sign-in binds its own session, so the secret is fresh and never the zero default.
+ let secrets = registers.compactMap { Self.bindSecret(in: $0.payload) }
+ #expect(secrets.count == 2)
+ #expect(secrets[0] != secrets[1])
+ #expect(!secrets.contains([UInt8](repeating: 0, count: Requests.bindSecretSize)))
+ await client.shutdown()
+ }
+
+ @Test func userRenameUpdatesTheReconnectCredentials() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ let client = makeClient(server)
+ try await client.connect()
+ try await client.updateUser("iggy", username: "renamed")
+ await server.closeAllConnections()
+ try await Task.sleep(for: .milliseconds(20))
+ try await client.ping()
+ let registers = await server.requests.filter { $0.operation == .register }
+ #expect(registers.count == 2)
+ #expect(Self.username(in: registers[0].payload) == "iggy")
+ #expect(Self.username(in: registers[1].payload) == "renamed")
+ // A later password change for the renamed user still lands on the
+ // credentials the reconnect uses.
+ try await client.changePassword("renamed", currentPassword: "iggy", newPassword: "new-secret")
+ await server.closeAllConnections()
+ try await Task.sleep(for: .milliseconds(20))
+ try await client.ping()
+ let latest = await server.requests.filter { $0.operation == .register }.last!
+ #expect(Self.username(in: latest.payload) == "renamed")
+ #expect(Self.password(in: latest.payload) == "new-secret")
+ await client.shutdown()
+ }
+
+ @Test func largeRepliesArriveInPieces() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ let body = [UInt8](repeating: 7, count: 300_000)
+ await server.respond { request in
+ request.operation == .register ? .ok(FakeVSRServer.registerBody()) : request.code == CommandCode.getStats.rawValue ? .okInChunks(body) : .ok([])
+ }
+ let client = makeClient(server)
+ try await client.connect()
+ let raw = try await client.sendRawRequest(code: CommandCode.getStats.rawValue, payload: [])
+ #expect(raw == body)
+ await client.shutdown()
+ }
+
+ @Test func leaderRedirectionMovesTheClient() async throws {
+ let leader = try await FakeVSRServer.started()
+ let follower = try await FakeVSRServer.started()
+ defer {
+ Task {
+ await leader.stop()
+ await follower.stop()
+ }
+ }
+ let roster = FakeVSRServer.rosterBody([follower.address, leader.address], leaderIndex: 1)
+ for server in [leader, follower] {
+ await server.respond { request in
+ if request.operation == .register { return .ok(FakeVSRServer.registerBody()) }
+ if request.code == CommandCode.getClusterMetadata.rawValue { return .ok(roster) }
+ return .ok([])
+ }
+ }
+ let client = makeClient(follower)
+ try await client.connect()
+ #expect(await client.serverAddress == leader.address)
+ #expect(await client.state == .authenticated)
+ #expect(await follower.connectionCount == 1)
+ #expect(await leader.connectionCount == 1)
+ try await client.ping()
+ #expect(await codes(leader).contains(CommandCode.ping.rawValue))
+ // A failover after the leader drops the connection dials the roster.
+ await leader.stop()
+ try await Task.sleep(for: .milliseconds(20))
+ try await client.ping()
+ #expect(await client.serverAddress == follower.address)
+ await client.shutdown()
+ }
+
+ /// An outage longer than the request timeout leaves the client
+ /// disconnected. Once the server is back the next typed call reconnects and
+ /// signs in again, with no ``IggyClient/connect()`` of its own.
+ @Test func typedCallReconnectsAfterAnOutage() async throws {
+ let server = try await FakeVSRServer.started()
+ let port = try Endpoint(parsing: server.address).port
+ let client = makeClient(server, requestTimeout: shortTimeout)
+ try await client.connect()
+ await server.stop()
+ await expectCode(.disconnected) { try await client.ping() }
+ #expect(await eventually { await client.state == .disconnected })
+
+ let revived = FakeVSRServer()
+ try await revived.startOn(port: port)
+ defer { Task { await revived.stop() } }
+ #expect(try await client.getStreams().isEmpty)
+ #expect(await client.state == .authenticated)
+ #expect(await operations(revived).filter { $0 == .register }.count == 1)
+ await client.shutdown()
+ }
+
+ /// The heartbeat keeps pinging a client that lost its connection, which
+ /// brings it back once the server returns, without any call.
+ @Test func heartbeatReconnectsAfterAnOutage() async throws {
+ let server = try await FakeVSRServer.started()
+ let port = try Endpoint(parsing: server.address).port
+ let client = makeClient(server, requestTimeout: shortTimeout, heartbeat: .milliseconds(30))
+ try await client.connect()
+ await server.stop()
+ #expect(await eventually { await client.state == .disconnected })
+
+ let revived = FakeVSRServer()
+ try await revived.startOn(port: port)
+ defer { Task { await revived.stop() } }
+ #expect(await eventually { await pingCount(revived) >= 1 })
+ #expect(await client.state == .authenticated)
+ #expect(await operations(revived).filter { $0 == .register }.count == 1)
+ await client.shutdown()
+ }
+
+ /// `connect()` on a connected client leaves the heartbeat alone:
+ /// cancelling it would abort a ping in flight and the connection with it.
+ @Test func connectingAgainKeepsTheHeartbeatAndItsPing() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ await server.respond { request in
+ if request.operation == .register { return .ok(FakeVSRServer.registerBody()) }
+ if request.code == CommandCode.ping.rawValue {
+ try? await Task.sleep(for: .milliseconds(300))
+ }
+ return .ok([])
+ }
+ let client = makeClient(server, heartbeat: .milliseconds(20))
+ try await client.connect()
+ #expect(await eventually { await pingCount(server) >= 1 })
+ try await client.connect()
+ #expect(await eventually { await pingCount(server) >= 2 })
+ #expect(await client.state == .authenticated)
+ #expect(await server.connectionCount == 1)
+ await client.shutdown()
+ }
+
+ @Test func heartbeatPingsPeriodically() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ var configuration = ClientConfiguration(address: server.address, autoLogin: rootLogin)
+ configuration.heartbeatInterval = .milliseconds(30)
+ let client = IggyClient(configuration: configuration, logger: testLogger)
+ try await client.connect()
+ // Waits for the pings rather than for a fixed time: a loaded machine
+ // stretches the interval, and only their arrival is being tested.
+ #expect(await eventually { await pingCount(server) >= 3 })
+ await client.shutdown()
+ // One ping may already be on the wire when the heartbeat stops.
+ let stopped = await pingCount(server)
+ try await Task.sleep(for: .milliseconds(150))
+ #expect(await pingCount(server) <= stopped + 1)
+ }
+
+ @Test func concurrentCallersShareOneConnectionInLockstep() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ await server.respond { request in
+ if request.operation == .register { return .ok(FakeVSRServer.registerBody()) }
+ try? await Task.sleep(for: .milliseconds(2))
+ return .ok([])
+ }
+ let client = makeClient(server)
+ try await client.connect()
+ try await withThrowingTaskGroup(of: Void.self) { group in
+ for index in 0..<20 {
+ group.addTask {
+ if index.isMultiple(of: 2) {
+ try await client.ping()
+ } else {
+ try await client.deleteStream(Identifier(numeric: UInt32(index)))
+ }
+ }
+ }
+ try await group.waitForAll()
+ }
+ let creates = await server.requests.filter { $0.operation == .deleteStream }.map(\.requestID)
+ #expect(creates.sorted() == Array(1...10))
+ #expect(await server.connectionCount == 1)
+ await client.shutdown()
+ }
+
+ /// Every caller that was waiting on a dropped connection fails on it, but
+ /// only the first may tear it down: the rest must reuse the connection
+ /// that caller established instead of closing it under its replay.
+ @Test func concurrentCallersReconnectOnceAfterADrop() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ let once = Locked(true)
+ await server.respond { request in
+ if request.operation == .register { return .ok(FakeVSRServer.registerBody()) }
+ guard request.code == CommandCode.getStats.rawValue else { return .ok([]) }
+ let first = once.withLock {
+ let first = $0; $0 = false; return first
+ }
+ if first {
+ // Lets the other callers queue up behind this exchange.
+ try? await Task.sleep(for: .milliseconds(50))
+ return .close
+ }
+ return .ok([UInt8](repeating: 0, count: 200))
+ }
+ let client = makeClient(server)
+ try await client.connect()
+ try await withThrowingTaskGroup(of: Void.self) { group in
+ for _ in 0..<12 {
+ group.addTask { _ = try await client.getStats() }
+ }
+ try await group.waitForAll()
+ }
+ #expect(await server.connectionCount == 2)
+ #expect(await operations(server).filter { $0 == .register }.count == 2)
+ await client.shutdown()
+ }
+
+ /// The replay after a reconnect can itself be refused for longer than the
+ /// failover check interval, which sends it to the routing lock the
+ /// reconnect just held. It must be free by then.
+ @Test func replayAfterAReconnectCanStillFailOver() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ let seen = Locked(0)
+ let refuseUntil = Locked<ContinuousClock.Instant?>(nil)
+ await server.respond { request in
+ if request.operation == .register { return .ok(FakeVSRServer.registerBody()) }
+ if request.code == CommandCode.getClusterMetadata.rawValue { return .ok(FakeVSRServer.rosterBody([server.address])) }
+ guard request.code == CommandCode.getStats.rawValue else { return .ok([]) }
+ let count = seen.withLock {
+ $0 += 1; return $0
+ }
+ if count == 1 { return .close }
+ let deadline = refuseUntil.withLock { stored -> ContinuousClock.Instant in
+ let deadline = stored ?? ContinuousClock.now + ClientCore.transientFailoverCheckInterval + .milliseconds(200)
+ stored = deadline
+ return deadline
+ }
+ return ContinuousClock.now < deadline ? .status(IggyErrorCode.transientNotAccepted.rawValue) : .ok([UInt8](repeating: 0, count: 200))
+ }
+ var configuration = ClientConfiguration(address: server.address, autoLogin: rootLogin)
+ configuration.reconnection = ReconnectionOptions(interval: .milliseconds(20), reestablishAfter: .zero)
+ configuration.requestTimeout = ClientCore.transientFailoverCheckInterval + .seconds(3)
+ configuration.heartbeatInterval = .seconds(60)
+ let client = IggyClient(configuration: configuration, logger: testLogger)
+ try await client.connect()
+ _ = try await client.getStats()
+ #expect(await server.connectionCount == 2)
+ await client.shutdown()
+ }
+
+ /// The reply of an abandoned request could arrive later and be read as
+ /// the answer to the next one, so the connection goes with the request.
+ @Test func cancellingARequestInFlightClosesItsConnection() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ let once = Locked(true)
+ await server.respond { request in
+ if request.operation == .register { return .ok(FakeVSRServer.registerBody()) }
+ guard request.code == CommandCode.getStats.rawValue else { return .ok([]) }
+ let first = once.withLock {
+ let first = $0; $0 = false; return first
+ }
+ return first ? .stall : .ok([UInt8](repeating: 0, count: 200))
+ }
+ var configuration = ClientConfiguration(address: server.address, autoLogin: rootLogin)
+ configuration.reconnection = ReconnectionOptions(interval: .milliseconds(20), reestablishAfter: .zero)
+ configuration.heartbeatInterval = .seconds(60)
+ let client = IggyClient(configuration: configuration, logger: testLogger)
+ try await client.connect()
+ let request = Task { try await client.getStats() }
+ #expect(await eventually { await codes(server).contains(CommandCode.getStats.rawValue) })
+ let clock = ContinuousClock()
+ request.cancel()
+ let elapsed = await clock.measure {
+ await #expect(throws: CancellationError.self) { _ = try await request.value }
+ }
+ // Far below the 30 second request timeout the stalled reply would hit.
+ #expect(elapsed < .seconds(5))
+ #expect(await server.connectionCount == 1)
+ _ = try await client.getStats()
+ #expect(await server.connectionCount == 2)
+ await client.shutdown()
+ }
+
+ /// A request cancelled before it is written costs nothing.
+ @Test func cancelledCallerDoesNotTouchTheConnection() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ let client = makeClient(server)
+ try await client.connect()
+ let before = await server.requests.count
+ let request = Task {
+ withUnsafeCurrentTask { $0?.cancel() }
+ try await client.ping()
+ }
+ await #expect(throws: CancellationError.self) { try await request.value }
+ #expect(await server.requests.count == before)
+ try await client.ping()
+ #expect(await server.connectionCount == 1)
+ await client.shutdown()
+ }
+
+ /// An explicit disconnect is caller intent: neither a later request nor a
+ /// sweep that was already retrying may reconnect behind the caller's back.
+ @Test func explicitDisconnectStaysDisconnectedUntilConnect() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ let client = makeClient(server)
+ try await client.connect()
+ await client.disconnect()
+ #expect(await client.state == .disconnected)
+ await expectCode(.notConnected) { try await client.ping() }
+ #expect(await server.connectionCount == 1)
+ try await client.connect()
+ try await client.ping()
+ #expect(await client.state == .authenticated)
+ #expect(await server.connectionCount == 2)
+ await client.shutdown()
+ }
+
+ @Test func explicitDisconnectEndsAConnectSweep() async throws {
+ let server = FakeVSRServer()
+ try await server.start()
+ await server.stop()
+ // Nothing listens on the address, and the retries never run out.
+ let client = makeClient(server)
+ let connecting = Task { try await client.connect() }
+ #expect(await eventually { await client.state == .connecting })
+ await client.disconnect()
+ await expectCode(.notConnected) { try await connecting.value }
+ #expect(await client.state == .disconnected)
+ await client.shutdown()
+ }
+
+ @Test func shutdownIsFinal() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ let client = makeClient(server)
+ try await client.connect()
+ await client.shutdown()
+ // A teardown after the shutdown must not put the client back into a
+ // state it can connect from.
+ await client.disconnect()
+ #expect(await client.state == .shutdown)
+ await expectCode(.clientShutdown) { try await client.connect() }
+ await expectCode(.clientShutdown) { try await client.ping() }
+ #expect(await client.state == .shutdown)
+ #expect(await server.connectionCount == 1)
+ // Nothing is published after shutdown, so a late subscription ends
+ // at once instead of waiting for a first event.
+ var events = await client.events.makeAsyncIterator()
+ #expect(await events.next() == nil)
+ }
+
+ @Test func shutdownDuringADialDoesNotReviveTheClient() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ let connector = GatedConnector()
+ let client = makeClient(server, connector: connector)
+ let connecting = Task { try await client.connect() }
+ #expect(await eventually { await client.state == .connecting })
+ await client.shutdown()
+ connector.released.withLock { $0 = true }
+ await expectCode(.clientShutdown) { try await connecting.value }
+ #expect(await client.state == .shutdown)
+ // The dial that finished late was dropped before any sign-in.
+ #expect(await eventually { await server.openConnectionCount == 0 })
+ #expect(await server.requests.isEmpty)
+ }
+
+ /// With unlimited retries a connect to a server that is gone never ends
+ /// by itself; cancelling the caller has to end it.
+ @Test func cancelledConnectEndsItsSweep() async throws {
+ let server = FakeVSRServer()
+ try await server.start()
+ let port = try Endpoint(parsing: server.address).port
+ await server.stop()
+ let client = makeClient(server)
+ let connecting = Task { try await client.connect() }
+ #expect(await eventually { await client.state == .connecting })
+ connecting.cancel()
+ await #expect(throws: CancellationError.self) { try await connecting.value }
+ #expect(await eventually { await client.state == .disconnected })
+ // The sweep left nothing behind: a later connect dials again.
+ try await server.startOn(port: port)
+ defer { Task { await server.stop() } }
+ try await client.connect()
+ #expect(await client.state == .authenticated)
+ await client.shutdown()
+ }
+
+ @Test func cancelledConnectDuringTheSignInLeavesNoHalfOpenState() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ let stallOnce = Locked(true)
+ await server.respond { request in
+ guard request.operation == .register else { return .ok([]) }
+ let stall = stallOnce.withLock {
+ let stall = $0; $0 = false; return stall
+ }
+ return stall ? .stall : .ok(FakeVSRServer.registerBody())
+ }
+ let client = makeClient(server)
+ let connecting = Task { try await client.connect() }
+ #expect(await eventually { await operations(server).contains(.register) })
+ connecting.cancel()
+ await #expect(throws: CancellationError.self) { try await connecting.value }
+ #expect(await eventually { await client.state == .disconnected })
+ try await client.connect()
+ #expect(await client.state == .authenticated)
+ try await client.ping()
+ await client.shutdown()
+ }
+
+ /// A client dropped without a shutdown must not keep its socket or its
+ /// heartbeat.
+ @Test func droppedClientClosesItsConnection() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ do {
+ var configuration = ClientConfiguration(address: server.address, autoLogin: rootLogin)
+ configuration.heartbeatInterval = .milliseconds(20)
+ let client = IggyClient(configuration: configuration, logger: testLogger)
+ try await client.connect()
+ #expect(await eventually { await pingCount(server) >= 1 })
+ }
+ #expect(await eventually { await server.openConnectionCount == 0 })
+ let stopped = await pingCount(server)
+ try await Task.sleep(for: .milliseconds(150))
+ #expect(await pingCount(server) <= stopped + 1)
+ }
+
+ @Test func sessionControlCodesAreRefusedRaw() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ let client = makeClient(server)
+ try await client.connect()
+ await expectCode(.invalidCommand) { _ = try await client.sendRawRequest(code: CommandCode.loginRegister.rawValue, payload: []) }
+ await expectCode(.invalidCommand) { _ = try await client.sendRawRequest(code: CommandCode.logoutUser.rawValue, payload: []) }
+ await client.shutdown()
+ }
+
+ /// A reader placed after the version info of a register payload, with the
+ /// bind secret that follows it.
+ private static func registerBody(_ payload: [UInt8]) -> (bindSecret: [UInt8], reader: ByteReader)? {
+ var reader = ByteReader(payload[...])
+ guard (try? reader.readUInt32()) != nil, (try? reader.readName()) != nil, (try? reader.readName()) != nil,
+ let bindSecret = try? reader.readBytes(Requests.bindSecretSize)
+ else {
+ return nil
+ }
+ return (Array(bindSecret), reader)
+ }
+
+ private static func bindSecret(in payload: [UInt8]) -> [UInt8]? {
+ registerBody(payload)?.bindSecret
+ }
+
+ /// The username field of a register payload, after the bind secret.
+ private static func username(in payload: [UInt8]) -> String? {
+ guard var reader = registerBody(payload)?.reader else {
+ return nil
+ }
+ return try? reader.readName()
+ }
+
+ /// The password field of a register payload, after the username.
+ private static func password(in payload: [UInt8]) -> String? {
+ guard var reader = registerBody(payload)?.reader else {
+ return nil
+ }
+ _ = try? reader.readName()
+ return try? reader.readName()
+ }
+}
diff --git a/foreign/swift/Tests/IggyTests/ConcurrencyTests.swift b/foreign/swift/Tests/IggyTests/ConcurrencyTests.swift
new file mode 100644
index 000000000..0a53603ad
--- /dev/null
+++ b/foreign/swift/Tests/IggyTests/ConcurrencyTests.swift
@@ -0,0 +1,209 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+import Foundation
+import NIOCore
+import NIOPosix
+import Testing
+
+@testable import Iggy
+
+@Suite("Concurrency helpers")
+struct ConcurrencyTests {
+ @Test func cancelledWaiterLeavesTheQueueWithoutTheLock() async throws {
+ let mutex = AsyncMutex()
+ try await mutex.lock()
+ let cancelled = Task { try await mutex.lock() }
+ let waiting = Task {
+ try await mutex.lock()
+ return true
+ }
+ cancelled.cancel()
+ // Returns although the lock is still held.
+ await #expect(throws: CancellationError.self) { try await cancelled.value }
+ mutex.unlock()
+ #expect(try await waiting.value)
+ mutex.unlock()
+ // Released twice by its two holders, so it is free again.
+ try await mutex.lock()
+ mutex.unlock()
+ }
+
+ @Test func mutexSerializesItsHolders() async throws {
+ let mutex = AsyncMutex()
+ let inside = Locked(0)
+ let overlaps = Locked(0)
+ try await withThrowingTaskGroup(of: Void.self) { group in
+ for _ in 0..<50 {
+ group.addTask {
+ try await mutex.lock()
+ defer { mutex.unlock() }
+ let holders = inside.withLock {
+ $0 += 1
+ return $0
+ }
+ if holders != 1 {
+ overlaps.withLock { $0 += 1 }
+ }
+ await Task.yield()
+ inside.withLock { $0 -= 1 }
+ }
+ }
+ try await group.waitForAll()
+ }
+ #expect(overlaps.withLock { $0 } == 0)
+ }
+
+ @Test func callersShareOneFlight() async throws {
+ let gate = SingleFlight()
+ let runs = Locked(0)
+ let release = Locked(false)
+ try await withThrowingTaskGroup(of: Void.self) { group in
+ for _ in 0..<5 {
+ group.addTask {
+ try await gate.run {
+ runs.withLock { $0 += 1 }
+ while !release.withLock({ $0 }) {
+ try await Task.sleep(for: .milliseconds(2))
+ }
+ }
+ }
+ }
+ #expect(await eventually { gate.isActive })
+ try await Task.sleep(for: .milliseconds(20))
+ release.withLock { $0 = true }
+ try await group.waitForAll()
+ }
+ #expect(runs.withLock { $0 } == 1)
+ #expect(!gate.isActive)
+ }
+
+ @Test func flightSurvivesWhileACallerRemains() async throws {
+ let gate = SingleFlight()
+ let release = Locked(false)
+ let work: SingleFlight.Work = {
+ while !release.withLock({ $0 }) {
+ try await Task.sleep(for: .milliseconds(2))
+ }
+ }
+ let leaving = Task { try await gate.run(work) }
+ #expect(await eventually { gate.isActive })
+ let staying = Task { try await gate.run(work) }
+ try await Task.sleep(for: .milliseconds(20))
+ leaving.cancel()
+ await #expect(throws: CancellationError.self) { try await leaving.value }
+ // The flight was not cancelled with the caller that left.
+ release.withLock { $0 = true }
+ try await staying.value
+ }
+
+ @Test func flightIsCancelledWhenItsLastCallerLeaves() async throws {
+ let gate = SingleFlight()
+ let sawCancellation = Locked(false)
+ let caller = Task {
+ try await gate.run {
+ do {
+ try await Task.sleep(for: .seconds(600))
+ } catch {
+ sawCancellation.withLock { $0 = true }
+ throw error
+ }
+ }
+ }
+ #expect(await eventually { gate.isActive })
+ caller.cancel()
+ await #expect(throws: CancellationError.self) { try await caller.value }
+ #expect(await eventually { sawCancellation.withLock { $0 } })
+ #expect(await eventually { !gate.isActive })
+ }
+
+ /// A caller that arrives while an abandoned flight unwinds must not share
+ /// its cancellation: it gets a flight of its own once that one has ended.
+ @Test func callerArrivingDuringUnwindingStartsANewFlight() async throws {
+ let gate = SingleFlight()
+ let unwound = Locked(false)
+ let runs = Locked(0)
+ let first = Task {
+ try await gate.run {
+ runs.withLock { $0 += 1 }
+ // Ignores cancellation until it is told to unwind.
+ while !unwound.withLock({ $0 }) {
+ try? await Task.sleep(for: .milliseconds(2))
+ await Task.yield()
+ }
+ }
+ }
+ #expect(await eventually { gate.isActive })
+ first.cancel()
+ await #expect(throws: CancellationError.self) { try await first.value }
+ let second = Task {
+ try await gate.run { runs.withLock { $0 += 1 } }
+ }
+ try await Task.sleep(for: .milliseconds(20))
+ #expect(runs.withLock { $0 } == 1)
+ unwound.withLock { $0 = true }
+ try await second.value
+ #expect(runs.withLock { $0 } == 2)
+ }
+
+ @Test func deadlineEndsAWaitThatHonoursCancellation() async throws {
+ let late = try await withDeadline(ContinuousClock.now + .milliseconds(50)) {
+ try await Task.sleep(for: .seconds(600))
+ return 1
+ }
+ #expect(late == nil)
+ let inTime = try await withDeadline(ContinuousClock.now + .seconds(600)) { 2 }
+ #expect(inTime == 2)
+ await #expect(throws: IggyError(.invalidCommand)) {
+ _ = try await withDeadline(ContinuousClock.now + .seconds(600)) { () async throws -> Int in throw IggyError(.invalidCommand) }
+ }
+ }
+
+ /// A peer that accepts the connection and never answers the ClientHello
+ /// must not hold the dial open past the connect timeout.
+ @Test func stalledTLSHandshakeEndsAtTheConnectTimeout() async throws {
+ let silent = try await ServerBootstrap(group: MultiThreadedEventLoopGroup.singleton)
+ .serverChannelOption(.socketOption(.so_reuseaddr), value: 1)
+ .bind(host: "127.0.0.1", port: 0).get()
+ defer { silent.close(promise: nil) }
+ let port = try #require(silent.localAddress?.port)
+ let endpoint = try Endpoint(parsing: "127.0.0.1:\(port)")
+ let clock = ContinuousClock()
+ let elapsed = await clock.measure {
+ await expectCode(.cannotEstablishConnection) {
+ _ = try await TCPConnection.connect(to: endpoint, tls: TLSOptions(validateCertificate: false), timeout: .milliseconds(300), noDelay: true)
+ }
+ }
+ #expect(elapsed >= .milliseconds(250) && elapsed < .seconds(10))
+ }
+
+ @Test func retryOutcomeTurnsLaterRefusalsIntoUnknown() {
+ var outcome = RetryOutcome()
+ #expect(outcome.observe(IggyError(.transientNotAccepted)).code == .transientNotAccepted)
+ #expect(outcome.observe(IggyError(.unauthenticated)).code == .unauthenticated)
+ #expect(outcome.observe(IggyError(.transientNotCommitted)).code == .transientNotCommitted)
+ #expect(outcome.isUncertain)
+ for refusal: IggyErrorCode in [
+ .transientNotAccepted, .historyUnavailable, .unauthorized, .unauthenticated, .staleClient, .notConnected, .cannotEstablishConnection,
+ ] {
+ #expect(outcome.observe(IggyError(refusal)).code == .transientNotCommitted)
+ }
+ // Anything else is reported as it is.
+ #expect(outcome.observe(IggyError(.disconnected)).code == .disconnected)
+ #expect(outcome.observe(IggyError(.streamNameAlreadyExists)).code == .streamNameAlreadyExists)
+ }
+}
diff --git a/foreign/swift/Tests/IggyTests/ConfigurationTests.swift b/foreign/swift/Tests/IggyTests/ConfigurationTests.swift
new file mode 100644
index 000000000..d4a61fe33
--- /dev/null
+++ b/foreign/swift/Tests/IggyTests/ConfigurationTests.swift
@@ -0,0 +1,120 @@
+// 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 Testing
+
+@testable import Iggy
+
+@Suite("Endpoints")
+struct EndpointTests {
+ @Test func parsesHostsAndPorts() throws {
+ let plain = try Endpoint(parsing: "localhost:8090")
+ #expect(plain.host == "localhost")
+ #expect(plain.port == 8090)
+ #expect(!plain.isIPLiteral)
+ let v4 = try Endpoint(parsing: "127.0.0.1:8090")
+ #expect(v4.isIPLiteral)
+ #expect(v4.description == "127.0.0.1:8090")
+ let v6 = try Endpoint(parsing: "[fd00::1]:8091")
+ #expect(v6.host == "fd00::1")
+ #expect(v6.port == 8091)
+ #expect(v6.isIPLiteral)
+ #expect(v6.description == "[fd00::1]:8091")
+ }
+
+ @Test(arguments: [
+ "", "localhost", ":8090", "localhost:0", "localhost:70000", "[::1]", "[::1]8090", "::1:8090", "host:port", "localhost:+8090", "[]:8090", "[::1:8090",
+ ])
+ func rejectsMalformedAddresses(_ address: String) {
+ #expect(throws: IggyError.self) {
+ try Endpoint(parsing: address)
+ }
+ }
+
+ @Test func spellingComparison() {
+ #expect(isSameSpelling("localhost:8090", "127.0.0.1:8090"))
+ #expect(isSameSpelling("[::1]:8090", "localhost:8090"))
+ #expect(isSameSpelling("Iggy-1:8090", "iggy-1:8090"))
+ #expect(!isSameSpelling("localhost:8090", "localhost:8091"))
+ #expect(!isSameSpelling("10.0.0.1:8090", "10.0.0.2:8090"))
+ }
+
+ @Test func namesAndTheAddressesTheyResolveToAreOneEndpoint() async {
+ let resolver: @Sendable (String) async -> Set<String>? = { host in
+ ["iggy-server": ["10.0.0.5", "fd00::5"], "10.0.0.5": ["10.0.0.5"], "other": ["10.0.0.6"]][host]
+ }
+ #expect(await isSameAddress("iggy-server:8090", "10.0.0.5:8090", resolve: resolver))
+ #expect(await !isSameAddress("iggy-server:8090", "10.0.0.5:8091", resolve: resolver))
+ #expect(await !isSameAddress("iggy-server:8090", "other:8090", resolve: resolver))
+ // A name the resolver cannot answer for is a different endpoint.
+ #expect(await !isSameAddress("unknown:8090", "10.0.0.5:8090", resolve: resolver))
+ // Two literals are compared as written and never resolved.
+ #expect(await !isSameAddress("10.0.0.5:8090", "10.0.0.6:8090") { _ in ["10.0.0.5"] })
+ #expect(await isSameAddress("LOCALHOST:8090", "127.0.0.1:8090") { _ in nil })
+ }
+
+ @Test func literalAddressesResolveToThemselves() async {
+ #expect(await resolveAddresses(of: "127.0.0.1") == ["127.0.0.1"])
+ #expect(await resolveAddresses(of: "::1") == ["::1"])
+ }
+
+ @Test func durationsSaturateAsNanoseconds() {
+ #expect(Duration.milliseconds(1500).nanosecondsClamped == 1_500_000_000)
+ #expect(Duration.seconds(Int64.max).nanosecondsClamped == Int64.max)
+ #expect(Duration.seconds(Int64.min).nanosecondsClamped == Int64.min)
+ #expect(Duration.zero.nanosecondsClamped == 0)
+ }
+}
+
+@Suite("Client configuration")
+struct ClientConfigurationTests {
+ @Test func tlsServerNameSkipsIPLiterals() throws {
+ let literal = TLSOptions()
+ #expect(literal.serverName(for: try Endpoint(parsing: "127.0.0.1:8090")) == nil)
+ #expect(literal.serverName(for: try Endpoint(parsing: "[::1]:8090")) == nil)
+ #expect(literal.serverName(for: try Endpoint(parsing: "iggy.example:8090")) == "iggy.example")
+ let named = TLSOptions(domain: "localhost")
+ #expect(named.serverName(for: try Endpoint(parsing: "127.0.0.1:8090")) == "localhost")
+ #expect(TLSOptions(domain: "").serverName(for: try Endpoint(parsing: "iggy.example:8090")) == "iggy.example")
+ #expect(TLSOptions(domain: "10.0.0.5").serverName(for: try Endpoint(parsing: "iggy.example:8090")) == nil)
+ }
+
+ @Test func missingCertificateAuthorityFileIsReported() {
+ let options = TLSOptions(caFile: "/nonexistent/iggy-ca.pem")
+ #expect(throws: IggyError(.invalidTlsCertificatePath)) { try options.makeContext() }
+ // Without validation the file is never read.
+ #expect(throws: Never.self) { try TLSOptions(caFile: "/nonexistent/iggy-ca.pem", validateCertificate: false).makeContext() }
+ }
+
+ /// A configuration ends up in logs and crash reports, so no description
+ /// of it may carry the password or the token.
+ @Test func descriptionsNeverCarryTheSecret() {
+ let password = ClientConfiguration(autoLogin: .usernamePassword(username: "iggy", password: "hunter2-secret"))
+ let token = ClientConfiguration(autoLogin: .personalAccessToken("iggypat-hunter2-secret"))
+ for configuration in [password, token] {
+ #expect(!"\(configuration)".contains("hunter2"))
+ #expect(!String(reflecting: configuration).contains("hunter2"))
+ #expect(!"\(String(describing: configuration.autoLogin))".contains("hunter2"))
+ }
+ #expect("\(Credentials.usernamePassword(username: "iggy", password: "hunter2"))" == "username iggy")
+ }
+
+ @Test func statesAndEventsDescribeThemselves() {
+ #expect(ClientState.authenticated.description == "authenticated")
+ #expect(DiagnosticEvent.signedIn.description == "signed_in")
+ }
+}
diff --git a/foreign/swift/Tests/IggyTests/Support/FakeVSRServer.swift b/foreign/swift/Tests/IggyTests/Support/FakeVSRServer.swift
new file mode 100644
index 000000000..44c389022
--- /dev/null
+++ b/foreign/swift/Tests/IggyTests/Support/FakeVSRServer.swift
@@ -0,0 +1,331 @@
+// 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 NIOCore
+import NIOPosix
+
+@testable import Iggy
+
+/// A scripted server speaking the consensus framing on loopback, so the
+/// client's reconnection, replay, eviction, and timeout paths run over real
+/// sockets without a server binary.
+final class FakeVSRServer: @unchecked Sendable {
+ /// One decoded request frame.
+ struct Request: Sendable {
+ let connection: Int
+ let operation: ConsensusOperation
+ /// The command code, read from the header for non-replicated
+ /// requests and mapped from the operation otherwise.
+ let code: UInt32?
+ let requestID: UInt64
+ let session: UInt64
+ let clientID: UInt128Value
+ let checksum: UInt64
+ let payload: [UInt8]
+ }
+
+ /// What the script answers with.
+ enum Reply: Sendable {
+ /// Status zero with this body; result-framed operations get the
+ /// empty result section prepended.
+ case ok([UInt8])
+ /// A pre-commit denial: the status in the header, no body.
+ case status(UInt32)
+ /// A committed rejection carried in the result section.
+ case committedRejection(UInt32)
+ case eviction(VSRFrame.EvictionReason)
+ /// A reply that answers some other request id.
+ case wrongRequestID(UInt64)
+ /// Closes the connection instead of answering.
+ case close
+ /// Never answers.
+ case stall
+ /// A status-zero reply written in two TCP chunks with a pause between.
+ case okInChunks([UInt8])
+ }
+
+ typealias Responder = @Sendable (Request) async -> Reply
+
+ let state = FakeServerState()
+ private let group = MultiThreadedEventLoopGroup.singleton
+ private var channel: Channel?
+ private(set) var port = 0
+
+ var address: String { "127.0.0.1:\(port)" }
+
+ var requests: [Request] {
+ get async { await state.requests }
+ }
+
+ var connectionCount: Int {
+ get async { await state.opened }
+ }
+
+ /// Connections that are still open.
+ var openConnectionCount: Int {
+ get async { await state.openCount }
+ }
+
+ /// Replaces the script; the default answers everything successfully.
+ func respond(_ responder: @escaping Responder) async {
+ await state.setResponder(responder)
+ }
+
+ func start() async throws {
+ try await startOn(port: 0)
+ }
+
+ /// Binds a specific port, for a server that comes back after a stop.
+ func startOn(port: Int) async throws {
+ let state = state
+ let bootstrap = ServerBootstrap(group: group)
+ .serverChannelOption(.socketOption(.so_reuseaddr), value: 1)
+ .childChannelInitializer { channel in
+ channel.eventLoop.makeCompletedFuture {
+ try channel.pipeline.syncOperations.addHandler(ByteToMessageHandler(FrameDecoder()))
+ try channel.pipeline.syncOperations.addHandler(FakeServerHandler(state: state))
+ }
+ }
+ let channel = try await bootstrap.bind(host: "127.0.0.1", port: port).get()
+ self.channel = channel
+ self.port = channel.localAddress!.port!
+ await state.setAddress(address)
+ }
+
+ func stop() async {
+ await state.closeAll()
+ try? await channel?.close().get()
+ channel = nil
+ }
+
+ func closeAllConnections() async {
+ await state.closeAll()
+ }
+
+ // MARK: Reply helpers
+
+ /// The body of a successful register reply.
+ static func registerBody(userID: UInt32 = 0, session: UInt64 = 1, serverVersion: String = "fake") -> [UInt8] {
+ var writer = ByteWriter()
+ writer.write(userID)
+ writer.write(session)
+ writer.write(ProtocolVersion.current.packed)
+ try! writer.writeName(serverVersion)
+ return writer.bytes
+ }
+
+ /// A roster with one node per address; the first is the leader.
+ static func rosterBody(_ addresses: [String], leaderIndex: Int = 0) -> [UInt8] {
+ var writer = ByteWriter()
+ try! writer.writeLongString("fake-cluster")
+ writer.write(UInt32(addresses.count))
+ for (index, address) in addresses.enumerated() {
+ let endpoint = try! Endpoint(parsing: address)
+ writer.write(UInt8(index))
+ try! writer.writeLongString("node-\(index)")
+ try! writer.writeLongString(endpoint.host)
+ writer.write(UInt16(endpoint.port))
+ writer.write(UInt16(0))
+ writer.write(UInt16(0))
+ writer.write(UInt16(0))
+ writer.write(index == leaderIndex ? ClusterNodeRole.leader.rawValue : ClusterNodeRole.follower.rawValue)
+ writer.write(ClusterNodeStatus.healthy.rawValue)
+ }
+ return writer.bytes
+ }
+
+ static func encodeReply(request: Request, operation: ConsensusOperation, status: UInt32, body: [UInt8]) -> [UInt8] {
+ var frame = [UInt8](repeating: 0, count: VSRFrame.headerSize)
+ frame.withUnsafeMutableBytes { buffer in
+ buffer.storeBytes(of: UInt32(VSRFrame.headerSize + body.count).littleEndian, toByteOffset: VSRFrame.ReplyOffset.size, as: UInt32.self)
+ buffer[VSRFrame.ReplyOffset.command] = VSRFrame.Command.reply.rawValue
+ buffer.storeBytes(of: request.requestID.littleEndian, toByteOffset: VSRFrame.ReplyOffset.request, as: UInt64.self)
+ buffer[VSRFrame.ReplyOffset.operation] = operation.rawValue
+ buffer.storeBytes(of: status.littleEndian, toByteOffset: VSRFrame.ReplyOffset.status, as: UInt32.self)
+ }
+ return frame + body
+ }
+
+ static func encodeEviction(request: Request, reason: VSRFrame.EvictionReason) -> [UInt8] {
+ var frame = [UInt8](repeating: 0, count: VSRFrame.headerSize)
+ frame.withUnsafeMutableBytes { buffer in
+ buffer.storeBytes(of: UInt32(VSRFrame.headerSize).littleEndian, toByteOffset: VSRFrame.ReplyOffset.size, as: UInt32.self)
+ buffer[VSRFrame.ReplyOffset.command] = VSRFrame.Command.eviction.rawValue
+ buffer.storeBytes(of: request.clientID.low.littleEndian, toByteOffset: VSRFrame.RequestOffset.client, as: UInt64.self)
+ buffer.storeBytes(of: request.clientID.high.littleEndian, toByteOffset: VSRFrame.RequestOffset.client + 8, as: UInt64.self)
+ buffer.storeBytes(of: ProtocolVersion.current.packed.littleEndian, toByteOffset: VSRFrame.EvictionOffset.serverProtocolVersion, as: UInt32.self)
+ buffer.storeBytes(of: ProtocolVersion.current.packed.littleEndian, toByteOffset: VSRFrame.EvictionOffset.serverProtocolVersionMin, as: UInt32.self)
+ buffer[VSRFrame.EvictionOffset.reason] = reason.rawValue
+ }
+ return frame
+ }
+
+ static func decodeRequest(connection: Int, frame: RawFrame) -> Request {
+ let header = frame.header
+ func u32(_ offset: Int) -> UInt32 { UInt32(littleEndianBytes: header[offset..<offset + 4]) }
+ func u64(_ offset: Int) -> UInt64 { UInt64(littleEndianBytes: header[offset..<offset + 8]) }
+ let operation = ConsensusOperation(rawValue: header[VSRFrame.RequestOffset.operation]) ?? .reserved
+ let code: UInt32?
+ switch operation {
+ case .nonReplicated: code = u32(VSRFrame.RequestOffset.reserved)
+ case .logout: code = CommandCode.logoutUser.rawValue
+ case .register: code = nil
+ default: code = CommandCode.allCases.first { ConsensusOperation.replicated(for: $0.rawValue) == operation }?.rawValue
+ }
+ return Request(
+ connection: connection, operation: operation, code: code, requestID: u64(VSRFrame.RequestOffset.request),
+ session: u64(VSRFrame.RequestOffset.session),
+ clientID: UInt128Value(low: u64(VSRFrame.RequestOffset.client), high: u64(VSRFrame.RequestOffset.client + 8)),
+ checksum: u64(VSRFrame.RequestOffset.requestChecksum), payload: frame.body)
+ }
+
+ /// Answers every request successfully with an empty body, a bound
+ /// session for a register, and a single-node roster for the metadata read.
+ static func defaultReply(_ request: Request, address: String) -> Reply {
+ switch request.operation {
+ case .register:
+ return .ok(registerBody())
+ case .nonReplicated where request.code == CommandCode.getClusterMetadata.rawValue:
+ return .ok(rosterBody([address]))
+ default:
+ return .ok([])
+ }
+ }
+}
+
+actor FakeServerState {
+ private(set) var requests: [FakeVSRServer.Request] = []
+ private(set) var opened = 0
+ private var channels: [ObjectIdentifier: Channel] = [:]
+ private var responder: FakeVSRServer.Responder?
+ var address = ""
+
+ func setResponder(_ responder: @escaping FakeVSRServer.Responder) {
+ self.responder = responder
+ }
+
+ func setAddress(_ address: String) {
+ self.address = address
+ }
+
+ func connectionOpened(_ channel: Channel) -> Int {
+ opened += 1
+ channels[ObjectIdentifier(channel)] = channel
+ return opened
+ }
+
+ func connectionClosed(_ channel: Channel) {
+ channels[ObjectIdentifier(channel)] = nil
+ }
+
+ var openCount: Int { channels.count }
+
+ func closeAll() {
+ for channel in channels.values {
+ channel.close(promise: nil)
+ }
+ channels.removeAll()
+ }
+
+ func handle(_ request: FakeVSRServer.Request) async -> FakeVSRServer.Reply {
+ requests.append(request)
+ if let responder {
+ return await responder(request)
+ }
+ return FakeVSRServer.defaultReply(request, address: address)
+ }
+}
+
+final class FakeServerHandler: ChannelInboundHandler, @unchecked Sendable {
+ typealias InboundIn = RawFrame
+
+ private let state: FakeServerState
+ private var connection = 0
+
+ init(state: FakeServerState) {
+ self.state = state
+ }
+
+ func channelActive(context: ChannelHandlerContext) {
+ let channel = context.channel
+ Task { [self] in
+ connection = await state.connectionOpened(channel)
+ }
+ }
+
+ func channelInactive(context: ChannelHandlerContext) {
+ let channel = context.channel
+ Task { await state.connectionClosed(channel) }
+ }
+
+ func channelRead(context: ChannelHandlerContext, data: NIOAny) {
+ let frame = unwrapInboundIn(data)
+ let channel = context.channel
+ Task { [self] in
+ let request = FakeVSRServer.decodeRequest(connection: connection, frame: frame)
+ let reply = await state.handle(request)
+ let operation = request.operation
+ func write(_ bytes: [UInt8]) {
+ var buffer = channel.allocator.buffer(capacity: bytes.count)
+ buffer.writeBytes(bytes)
+ channel.writeAndFlush(buffer, promise: nil)
+ }
+ func framedBody(_ body: [UInt8]) -> [UInt8] {
+ (operation.isResultFramed || operation == .register) ? [0, 0, 0, 0] + body : body
+ }
+ switch reply {
+ case .ok(let body):
+ write(FakeVSRServer.encodeReply(request: request, operation: operation, status: 0, body: framedBody(body)))
+ case .status(let status):
+ write(FakeVSRServer.encodeReply(request: request, operation: operation, status: status, body: []))
+ case .committedRejection(let code):
+ var writer = ByteWriter()
+ writer.write(UInt32(1))
+ writer.write(UInt32(0))
+ writer.write(code)
+ write(FakeVSRServer.encodeReply(request: request, operation: operation, status: 0, body: writer.bytes))
+ case .eviction(let reason):
+ write(FakeVSRServer.encodeEviction(request: request, reason: reason))
+ case .wrongRequestID(let requestID):
+ let wrong = FakeVSRServer.Request(
+ connection: request.connection, operation: operation, code: request.code, requestID: requestID, session: request.session,
+ clientID: request.clientID, checksum: request.checksum, payload: request.payload)
+ write(FakeVSRServer.encodeReply(request: wrong, operation: operation, status: 0, body: framedBody([])))
+ case .close:
+ channel.close(promise: nil)
+ case .stall:
+ break
+ case .okInChunks(let body):
+ let frame = FakeVSRServer.encodeReply(request: request, operation: operation, status: 0, body: framedBody(body))
+ let split = frame.count / 2
+ write(Array(frame[..<split]))
+ try? await Task.sleep(for: .milliseconds(20))
+ write(Array(frame[split...]))
+ }
+ }
+ }
+}
+
+extension FakeVSRServer {
+ /// A started server whose default script knows its own address.
+ static func started() async throws -> FakeVSRServer {
+ let server = FakeVSRServer()
+ try await server.start()
+ await server.state.setAddress(server.address)
+ return server
+ }
+}
diff --git a/foreign/swift/Tests/IggyTests/Support/TestHelpers.swift b/foreign/swift/Tests/IggyTests/Support/TestHelpers.swift
new file mode 100644
index 000000000..50c4a01d1
--- /dev/null
+++ b/foreign/swift/Tests/IggyTests/Support/TestHelpers.swift
@@ -0,0 +1,64 @@
+// 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
+import Testing
+
+@testable import Iggy
+
+let testLogger = Logger(label: "org.apache.iggy.tests")
+
+/// A value behind a lock, for counters shared with `@Sendable` closures.
+final class Locked<Value>: @unchecked Sendable {
+ private let lock = NSLock()
+ private var value: Value
+
+ init(_ value: Value) {
+ self.value = value
+ }
+
+ func withLock<T>(_ body: (inout Value) throws -> T) rethrows -> T {
+ lock.lock()
+ defer { lock.unlock() }
+ return try body(&value)
+ }
+}
+
+/// Polls `condition` until it holds or `timeout` passes.
+func eventually(timeout: Duration = .seconds(5), _ condition: @Sendable () async -> Bool) async -> Bool {
+ let deadline = ContinuousClock.now + timeout
+ while ContinuousClock.now < deadline {
+ if await condition() {
+ return true
+ }
+ try? await Task.sleep(for: .milliseconds(5))
+ }
+ return await condition()
+}
+
+/// Asserts that `body` throws an ``IggyError`` with `code`.
+func expectCode(_ code: IggyErrorCode, sourceLocation: SourceLocation = #_sourceLocation, _ body: () async throws -> Void) async {
+ do {
+ try await body()
+ Issue.record("expected \(code) but nothing was thrown", sourceLocation: sourceLocation)
+ } catch let error as IggyError {
+ #expect(error.code == code, "expected \(code), got \(error)", sourceLocation: sourceLocation)
+ } catch {
+ Issue.record("expected \(code), got \(error)", sourceLocation: sourceLocation)
+ }
+}