You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by hg...@apache.org on 2026/10/16 09:19:05 UTC
(iggy) branch master updated: feat(swift): send and poll messages, consumer groups, and offsets (#4553)
This is an automated email from the ASF dual-hosted git repository.
hubcio 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 2a5c25e89 feat(swift): send and poll messages, consumer groups, and offsets (#4553)
2a5c25e89 is described below
commit 2a5c25e8900185122f742612d5507759d9bde31d
Author: RustToMetal <th...@gmail.com>
AuthorDate: Fri Oct 16 04:19:00 2026 -0500
feat(swift): send and poll messages, consumer groups, and offsets (#4553)
---
.github/actions/swift/pre-merge/action.yml | 146 +++-
.github/config/components.yml | 2 +-
foreign/swift/Package.swift | 4 +
foreign/swift/README.md | 76 ++-
foreign/swift/Sources/Iggy/Client/ClientCore.swift | 364 ++++++++--
.../Sources/Iggy/Client/ConsumerGroupState.swift | 223 ++++++
.../Sources/Iggy/Client/IggyClient+Messages.swift | 410 +++++++++++
.../Sources/Iggy/Client/IggyClient+Resources.swift | 61 +-
foreign/swift/Sources/Iggy/Client/IggyClient.swift | 27 +-
foreign/swift/Sources/Iggy/Client/PollRouter.swift | 394 +++++++++++
foreign/swift/Sources/Iggy/Models/Identifier.swift | 16 +
.../swift/Sources/Iggy/Models/IggyMessage.swift | 6 +
foreign/swift/Sources/Iggy/Models/Messaging.swift | 58 +-
.../swift/Sources/Iggy/Protocol/CommandCode.swift | 5 +-
foreign/swift/Sources/Iggy/Protocol/Requests.swift | 8 +
.../swift/Sources/Iggy/Protocol/Responses.swift | 75 +-
foreign/swift/Sources/Iggy/Protocol/VSRFrame.swift | 15 +-
.../Sources/Iggy/Transport/TCPConnection.swift | 5 +-
.../swift/Tests/IggyE2ETests/ClientE2ETests.swift | 693 +++++++++++++++++++
.../IggyE2ETests/Support/E2EEnvironment.swift | 212 ++++++
.../Tests/IggyTests/ClientTransportTests.swift | 213 ++++--
.../swift/Tests/IggyTests/ConcurrencyTests.swift | 8 +
.../Tests/IggyTests/ConsumerGroupStateTests.swift | 122 ++++
foreign/swift/Tests/IggyTests/Fixtures/golden.json | 5 +-
foreign/swift/Tests/IggyTests/MessagingTests.swift | 756 +++++++++++++++++++++
foreign/swift/Tests/IggyTests/ModelTests.swift | 27 +
.../Tests/IggyTests/ResponseVectorTests.swift | 14 +
.../Tests/IggyTests/Support/FakeVSRServer.swift | 135 +++-
.../Tests/IggyTests/Support/TestHelpers.swift | 21 +
.../swift/Tests/IggyTests/WireVectorTests.swift | 2 +
foreign/swift/Tools/golden-vectors/src/main.rs | 53 +-
31 files changed, 3999 insertions(+), 157 deletions(-)
diff --git a/.github/actions/swift/pre-merge/action.yml b/.github/actions/swift/pre-merge/action.yml
index 8623a6a53..29591c227 100644
--- a/.github/actions/swift/pre-merge/action.yml
+++ b/.github/actions/swift/pre-merge/action.yml
@@ -20,7 +20,7 @@ description: Swift pre-merge testing github iggy actions
inputs:
task:
- description: "Task to run (lint, test, build, build-macos, golden)"
+ description: "Task to run (lint, test, build, build-macos, golden, e2e)"
required: true
runs:
@@ -50,6 +50,8 @@ runs:
cd foreign/swift
swift build -Xswiftc -warnings-as-errors
+ # The end-to-end suite skips itself while IGGY_TCP_ADDRESS is unset,
+ # so this runs the unit and fake-server suites only.
- name: Test
shell: bash
if: inputs.task == 'test'
@@ -102,3 +104,145 @@ runs:
exit 1
fi
echo "✅ golden.json matches the Rust crates"
+
+ - name: Setup Rust for e2e
+ if: inputs.task == 'e2e'
+ uses: ./.github/actions/utils/setup-rust
+
+ - name: Build the end-to-end suite
+ shell: bash
+ if: inputs.task == 'e2e'
+ run: |
+ cd foreign/swift
+ swift build --build-tests
+
+ # The SDK speaks the VSR wire protocol, so every e2e lane runs against a
+ # locally built VSR server.
+ - name: Start Iggy server (plain)
+ id: iggy
+ if: inputs.task == 'e2e'
+ uses: ./.github/actions/utils/server-start
+ with:
+ cargo-bin: iggy-server
+ replica-id: "0"
+ pid-file: ${{ runner.temp }}/iggy-swift-e2e.pid
+ log-file: ${{ runner.temp }}/iggy-swift-e2e.log
+ wait-timeout-seconds: "90"
+ env:
+ IGGY_PATH: ${{ runner.temp }}/iggy-swift-e2e-data
+
+ - name: Run e2e tests
+ shell: bash
+ if: inputs.task == 'e2e'
+ env:
+ IGGY_TCP_ADDRESS: 127.0.0.1:8090
+ IGGY_ROOT_USERNAME: iggy
+ IGGY_ROOT_PASSWORD: iggy
+ run: |
+ echo "🧪 Running Swift e2e tests..."
+ cd foreign/swift
+ swift test --skip-build --filter IggyE2ETests
+
+ - name: Stop Iggy server (plain)
+ if: always() && inputs.task == 'e2e'
+ uses: ./.github/actions/utils/server-stop
+ with:
+ pid-file: ${{ steps.iggy.outputs.pid_file }}
+ log-file: ${{ steps.iggy.outputs.log_file }}
+
+ - name: Start Iggy server (TLS)
+ id: iggy-tls
+ if: inputs.task == 'e2e'
+ uses: ./.github/actions/utils/server-start
+ with:
+ cargo-bin: iggy-server
+ replica-id: "0"
+ pid-file: ${{ runner.temp }}/iggy-swift-e2e-tls.pid
+ log-file: ${{ runner.temp }}/iggy-swift-e2e-tls.log
+ wait-timeout-seconds: "90"
+ env:
+ IGGY_PATH: ${{ runner.temp }}/iggy-swift-e2e-tls-data
+ IGGY_TCP_TLS_ENABLED: "true"
+ IGGY_TCP_TLS_CERT_FILE: core/certs/iggy_cert.pem
+ IGGY_TCP_TLS_KEY_FILE: core/certs/iggy_key.pem
+
+ - name: Run TLS e2e tests
+ shell: bash
+ if: inputs.task == 'e2e'
+ env:
+ IGGY_TCP_ADDRESS: 127.0.0.1:8090
+ IGGY_TCP_TLS_ENABLED: "true"
+ IGGY_ROOT_USERNAME: iggy
+ IGGY_ROOT_PASSWORD: iggy
+ run: |
+ echo "🔐 Running Swift e2e tests over TLS..."
+ cd foreign/swift
+ swift test --skip-build --filter IggyE2ETests
+
+ - name: Stop Iggy server (TLS)
+ if: always() && inputs.task == 'e2e'
+ uses: ./.github/actions/utils/server-stop
+ with:
+ pid-file: ${{ steps.iggy-tls.outputs.pid_file }}
+ log-file: ${{ steps.iggy-tls.outputs.log_file }}
+
+ # Both replicas of the roster in `core/server/config.toml`: a two-replica
+ # cluster commits on two acks, so a lone node would commit nothing. The
+ # suite then runs the cluster paths for real: sign-in through the
+ # roster, data connections to partition primaries bound with
+ # BindSession, and the session they share.
+ - name: Start Iggy VSR cluster node 0
+ id: iggy-cluster-0
+ if: inputs.task == 'e2e'
+ uses: ./.github/actions/utils/server-start
+ with:
+ cargo-bin: iggy-server
+ replica-id: "0"
+ pid-file: ${{ runner.temp }}/iggy-swift-cluster-0.pid
+ log-file: ${{ runner.temp }}/iggy-swift-cluster-0.log
+ wait-timeout-seconds: "90"
+ env:
+ IGGY_CLUSTER_ENABLED: "true"
+ IGGY_PATH: ${{ runner.temp }}/iggy-swift-cluster-0-data
+
+ - name: Start Iggy VSR cluster node 1
+ id: iggy-cluster-1
+ if: inputs.task == 'e2e'
+ uses: ./.github/actions/utils/server-start
+ with:
+ cargo-bin: iggy-server
+ replica-id: "1"
+ tcp_address: 127.0.0.1:8091
+ http_address: 127.0.0.1:3001
+ pid-file: ${{ runner.temp }}/iggy-swift-cluster-1.pid
+ log-file: ${{ runner.temp }}/iggy-swift-cluster-1.log
+ wait-timeout-seconds: "90"
+ env:
+ IGGY_CLUSTER_ENABLED: "true"
+ IGGY_PATH: ${{ runner.temp }}/iggy-swift-cluster-1-data
+
+ - name: Run cluster e2e tests
+ shell: bash
+ if: inputs.task == 'e2e'
+ env:
+ IGGY_TCP_ADDRESS: 127.0.0.1:8090
+ IGGY_ROOT_USERNAME: iggy
+ IGGY_ROOT_PASSWORD: iggy
+ run: |
+ echo "🗳️ Running Swift e2e tests against a two-node cluster..."
+ cd foreign/swift
+ swift test --skip-build --filter IggyE2ETests
+
+ - name: Stop Iggy VSR cluster node 1
+ if: always() && inputs.task == 'e2e'
+ uses: ./.github/actions/utils/server-stop
+ with:
+ pid-file: ${{ steps.iggy-cluster-1.outputs.pid_file }}
+ log-file: ${{ steps.iggy-cluster-1.outputs.log_file }}
+
+ - name: Stop Iggy VSR cluster node 0
+ if: always() && inputs.task == 'e2e'
+ uses: ./.github/actions/utils/server-stop
+ with:
+ pid-file: ${{ steps.iggy-cluster-0.outputs.pid_file }}
+ log-file: ${{ steps.iggy-cluster-0.outputs.log_file }}
diff --git a/.github/config/components.yml b/.github/config/components.yml
index 4714969fe..fe40f8f48 100644
--- a/.github/config/components.yml
+++ b/.github/config/components.yml
@@ -343,7 +343,7 @@ components:
- "foreign/swift/**"
- "bdd/swift/**"
- "examples/swift/**"
- tasks: ["lint", "test", "build", "build-macos", "golden"]
+ tasks: ["lint", "test", "build", "build-macos", "golden", "e2e"]
# Shared BDD infrastructure that affects all BDD tests
bdd-infrastructure:
diff --git a/foreign/swift/Package.swift b/foreign/swift/Package.swift
index 66fe250f1..839cdd5b1 100644
--- a/foreign/swift/Package.swift
+++ b/foreign/swift/Package.swift
@@ -57,6 +57,10 @@ let package = Package(
],
resources: [.copy("Fixtures")]
),
+ .testTarget(
+ name: "IggyE2ETests",
+ dependencies: ["Iggy"]
+ ),
],
swiftLanguageModes: [.v6]
)
diff --git a/foreign/swift/README.md b/foreign/swift/README.md
index 0d356dab7..35e4b1f80 100644
--- a/foreign/swift/README.md
+++ b/foreign/swift/README.md
@@ -10,11 +10,11 @@
Official Swift client SDK for [Apache Iggy](https://iggy.apache.org) message streaming.
-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
+The SDK is built up in stages. The client is a TCP connection with optional TLS that
+speaks the consensus protocol, signs in, keeps the session alive with a heartbeat,
+reconnects, and follows the cluster leader. It manages streams, topics, partitions,
+segments, users, personal access tokens, and consumer groups, sends and polls messages,
+and stores consumer offsets. The producer and consumer, examples, and BDD scenarios
follow in later changes.
## Requirements
@@ -100,6 +100,55 @@ when it has credentials to sign in with. In a cluster the client reads the roste
signing in, moves to the leader, and moves on when a node answers that it is not the
primary.
+### Messages
+
+`sendMessages` writes a batch to one partition. The server routes explicit partitions
+only, so `balanced` (the default) and `messagesKey` are resolved on the client:
+balanced round-robins over the topic's partitions, and a key is hashed with XXH32 the
+way every other SDK hashes it. A message without an id gets a random one, written back
+into the batch, so sending the same batch again after `transientNotCommitted` carries
+the same ids. `pollMessages` reads from a partition, starting where the strategy says:
+
+```swift
+var batch: [IggyMessage] = ["placed"]
+try await client.sendMessages(streamID: "orders", topicID: "created", partitioning: .messagesKey("customer-42"), messages: &batch)
+let polled = try await client.pollMessages(
+ streamID: "orders", topicID: "created", partitionID: 0, consumer: .consumer(1), strategy: .next, count: 100, autoCommit: true)
+for message in polled.messages {
+ print(message.offset, message.payloadString)
+}
+```
+
+Every partition request carries the partition context the server requires: the
+partition's incarnation, which changes when a deleted partition id is reused. The
+client reads it from the partition's route and caches it. A poll reply carries the
+context it was served under; hand it back with `.offset(_:context:)` to continue the
+same partition, or in a `ConsumerPosition` to store a checkpoint. The server refuses
+either with `historyUnavailable` once the partition was recreated, and a consumer-group
+checkpoint with `consumerGroupPartitionNotOwned` once the partition moved to another
+member.
+
+### Consumer groups
+
+A member polls without a partition id and the client takes the next partition of its
+assignment, round-robin, syncing the assignment from the server when it changes. The
+heartbeat syncs every joined group, so a member picks up new partitions on its own:
+
+```swift
+try await client.joinConsumerGroup(streamID: "orders", topicID: "created", groupID: "billing")
+let batch = try await client.pollMessages(
+ streamID: "orders", topicID: "created", partitionID: nil, consumer: .group("billing"), strategy: .next, count: 100, autoCommit: true)
+```
+
+A member holding no partitions gets an empty poll whose partition id is
+`PolledMessages.noAssignedPartition`, and a client that is not a member gets
+`consumerGroupMemberNotFound`. Membership belongs to the session. A reconnect resumes
+the session and keeps it, but when the server no longer holds the session the client
+signs in under a new one and the caller must call `joinConsumerGroup` again: the next
+group poll fails with `consumerGroupMemberNotFound` until it does. In a cluster, polls
+and offset writes go to the partition's primary over a second connection bound to the
+same session.
+
### Errors
Every operation throws `IggyError`, a value carrying the typed `code`
@@ -127,6 +176,23 @@ 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 end-to-end suite runs against a server at the address in `IGGY_TCP_ADDRESS` and
+skips when that variable is unset. Start one from the repository root:
+
+```bash
+cargo build --bin iggy-server
+IGGY_PATH=/tmp/iggy-swift IGGY_ROOT_USERNAME=iggy IGGY_ROOT_PASSWORD=iggy target/debug/iggy-server
+```
+
+Then, from `foreign/swift`:
+
+```bash
+IGGY_TCP_ADDRESS=127.0.0.1:8090 swift test --filter IggyE2ETests
+```
+
+Add `IGGY_TCP_TLS_ENABLED=true` to run it against a server started with
+`IGGY_TCP_TLS_ENABLED=true` and the certificate pair in `core/certs`.
+
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/ClientCore.swift b/foreign/swift/Sources/Iggy/Client/ClientCore.swift
index 7cc9b57b0..3b06742d0 100644
--- a/foreign/swift/Sources/Iggy/Client/ClientCore.swift
+++ b/foreign/swift/Sources/Iggy/Client/ClientCore.swift
@@ -60,7 +60,7 @@ actor ClientCore {
/// 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 session: ConsensusSession
private var currentAddress: String
private var rosterEndpoints: [String] = []
private var rosterLearned = false
@@ -80,12 +80,22 @@ actor ClientCore {
private let routingLock = AsyncMutex()
private let connectGate = SingleFlight()
private var heartbeatTask: Task<Void, Never>?
+ /// The highest metadata commit a reply reported. A route read before it
+ /// may predate a change the client already saw, so it is read again.
+ /// A session reset keeps it.
+ private(set) var metadataWatermark: UInt64 = 0
+ /// Partitioning and consumer-group state that the typed calls share.
+ nonisolated let groupState = ConsumerGroupState()
+ nonisolated let pollRouter: PollRouter
init(configuration: ClientConfiguration, connector: any IggyConnector, logger: Logger) {
self.configuration = configuration
self.connector = connector
self.logger = logger
currentAddress = configuration.address
+ let session = ConsensusSession()
+ self.session = session
+ pollRouter = PollRouter(fence: session.clientID)
}
// MARK: State and events
@@ -135,15 +145,23 @@ actor ClientCore {
/// 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] {
+ ///
+ /// `context` is the partition context a partition request carries; every
+ /// replay keeps it. `limit` caps the request timeout for a caller with a
+ /// budget of its own.
+ func exchange(
+ code: UInt32, payload: [UInt8], context: PartitionContext = .empty, deadline limit: ContinuousClock.Instant? = nil
+ ) async throws -> [UInt8] {
if disconnectedByCaller {
throw IggyError(state == .shutdown ? .clientShutdown : .notConnected)
}
- let deadline = ContinuousClock.now + configuration.requestTimeout
+ let deadline = requestDeadline(within: limit)
var outcome = RetryOutcome()
var failedEpoch = connectionEpoch
+ var request: VSRFrame.EncodedRequest?
do {
- return try await sendRaw(code: code, payload: payload, deadline: deadline, epoch: &failedEpoch, outcome: &outcome)
+ return try await sendRaw(
+ code: code, payload: payload, context: context, deadline: deadline, request: &request, 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.
@@ -151,8 +169,8 @@ actor ClientCore {
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
+ // The server deduplicates by (client, request). A replay under a
+ // new session gets a new identity, so it is safe only for 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)")
@@ -183,21 +201,99 @@ actor ClientCore {
// request does it.
throw failure
}
+ if !replay {
+ // The request was written and lost, as `RetryOutcome::for_reconnect`
+ // in the Rust SDK records it: no later answer proves it never
+ // committed, so a refused replay is not stamped afresh.
+ outcome.markUncertain()
+ }
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)
+ let reconnected: Void?
+ do {
+ reconnected = 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)
+ }
+ } catch let reconnectError as IggyError {
+ if !replay {
+ throw IggyError(.transientNotCommitted, context: "command \(code) may have committed before the reconnect failed: \(reconnectError)")
+ }
+ throw outcome.observe(reconnectError)
}
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")
+ if let retained = request, retained.clientID == session.clientID,
+ retained.session == (session.session ?? 0) || retained.operation == .register
+ {
+ // The session was resumed: the request goes again as it was
+ // stamped, and the server answers it from its reply cache if
+ // it committed.
+ } else if replay {
+ request = nil
+ } else {
+ logger.warning("Reconnected under a new session, 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)
+ return try await sendRaw(
+ code: code, payload: payload, context: context, deadline: deadline, request: &request, epoch: &failedEpoch, outcome: &outcome)
+ }
+ }
+
+ /// One attempt on the current connection with no reconnect, failover, or
+ /// replay, for the route lookups and the polls that `PollRouter` retries
+ /// itself, bounded by `limit` when the caller has a budget. `notPrimary`
+ /// is graded as on the retried path.
+ func exchangeOnce(code: UInt32, payload: [UInt8], context: PartitionContext, deadline limit: ContinuousClock.Instant? = nil) async throws -> [UInt8] {
+ if disconnectedByCaller || state == .shutdown {
+ throw IggyError(state == .shutdown ? .clientShutdown : .notConnected)
}
+ var request: VSRFrame.EncodedRequest?
+ var epoch = connectionEpoch
+ var outcome = RetryOutcome()
+ let now = ContinuousClock.now
+ let attempted = try await attempt(
+ code: code, payload: payload, context: context, request: &request, epoch: &epoch, outcome: &outcome, transientDeadline: now,
+ readDeadline: requestDeadline(within: limit), replaysUncommitted: false)
+ switch attempted {
+ case .success(let body):
+ return body
+ case .failure(let error):
+ dropConnection(after: error, epoch: epoch)
+ throw Self.graded(error)
+ }
+ }
+
+ private func requestDeadline(within limit: ContinuousClock.Instant?) -> ContinuousClock.Instant {
+ let full = ContinuousClock.now + configuration.requestTimeout
+ return limit.map { min($0, full) } ?? full
+ }
+
+ var bindSecret: [UInt8] { session.bindSecret }
+
+ /// Stamps a frame for a data connection from the shared session, for the
+ /// session `identity` names. Replicated requests take the next id of the
+ /// session's counter; reads carry its current value. A session that was
+ /// replaced or is not bound, or a route older than the newest metadata
+ /// commit, is refused as not accepted, since nothing was written.
+ func stampRequest(
+ code: UInt32, payload: [UInt8], context: PartitionContext, for identity: Responses.SessionIdentity
+ ) throws -> VSRFrame.EncodedRequest {
+ guard session.clientID == identity.clientID, session.session == identity.session else {
+ throw IggyError(.transientNotAccepted, context: "the route belongs to a replaced session")
+ }
+ guard identity.metadataWatermark >= metadataWatermark else {
+ throw IggyError(.transientNotAccepted, context: "the route predates metadata commit \(metadataWatermark)")
+ }
+ return try VSRFrame.encodeRequest(session: &session, code: code, payload: payload, partitionContext: context)
+ }
+
+ /// Dials `address` with the client's TLS and timeouts, for a data
+ /// connection.
+ func dial(_ address: String) async throws -> any IggyConnection {
+ try await connector.connect(
+ to: try Endpoint(parsing: address), tls: configuration.tls, timeout: configuration.connectTimeout, noDelay: configuration.noDelay)
}
/// Fails fast with the error the Rust SDK raises for the state, after
@@ -265,7 +361,8 @@ actor ClientCore {
/// 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
+ code: UInt32, payload: [UInt8], context: PartitionContext, deadline: ContinuousClock.Instant, request: inout VSRFrame.EncodedRequest?,
+ epoch: inout UInt64, outcome: inout RetryOutcome
) async throws -> [UInt8] {
switch state {
case .shutdown:
@@ -277,26 +374,25 @@ actor ClientCore {
}
let clock = ContinuousClock()
let isLogin = CommandCode.isRegister(code)
- var request: VSRFrame.EncodedRequest?
var walk = RosterWalk()
+ var refused: IggyError?
while true {
+ if let refused, deadline - clock.now < Self.notReadyRetryInterval {
+ // An attempt sent with a sliver of the budget would time out,
+ // and the timeout takes the connection with it.
+ throw refused
+ }
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)
+ code: code, payload: payload, context: context, request: &request, epoch: &epoch, outcome: &outcome, transientDeadline: transientDeadline,
+ readDeadline: deadline, replaysUncommitted: true)
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)
- }
+ dropConnection(after: error, epoch: epoch)
throw error
}
if code == CommandCode.getClusterMetadata.rawValue {
@@ -304,6 +400,7 @@ actor ClientCore {
}
// The server never admitted the request, so re-issuing it
// under a new request id cannot double-apply it.
+ refused = error
let from = walk
guard let step = try await withDeadline(deadline, { [self] in try await self.failOver(from) }) else {
throw error
@@ -326,11 +423,24 @@ actor ClientCore {
}
}
+ /// The reply stream of a connection that failed mid-request is at an
+ /// unknown boundary; a late reply would desync framing for the next
+ /// request. One found closed before the request was written is dropped
+ /// too, so the next request dials.
+ private func dropConnection(after error: IggyError, epoch: UInt64) {
+ let lost = error.code == .disconnected || (error.code == .notConnected && connection != nil)
+ if lost, epoch == connectionEpoch, state != .shutdown {
+ connection = nil
+ state = .disconnected
+ publish(.disconnected)
+ }
+ }
+
/// `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 {
+ static func graded(_ failure: IggyError) -> IggyError {
guard failure.code == .notPrimary else {
return failure
}
@@ -375,21 +485,23 @@ actor ClientCore {
/// 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
+ code: UInt32, payload: [UInt8], context: PartitionContext, request: inout VSRFrame.EncodedRequest?, epoch: inout UInt64,
+ outcome: inout RetryOutcome, transientDeadline: ContinuousClock.Instant, readDeadline: ContinuousClock.Instant, replaysUncommitted: Bool
) 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)
+ code: code, payload: payload, context: context, request: &request, epoch: &epoch, outcome: &outcome, transientDeadline: transientDeadline,
+ readDeadline: readDeadline, replaysUncommitted: replaysUncommitted)
}
/// Stamps the request if it is not stamped yet and exchanges it, replaying
- /// in place while the server answers transiently.
+ /// in place while the server answers transiently: "not accepted" until
+ /// `transientDeadline`, and "not committed" until `readDeadline` when
+ /// `replaysUncommitted` is set.
private func exchangeLocked(
- code: UInt32, payload: [UInt8], request: inout VSRFrame.EncodedRequest?, epoch: inout UInt64, outcome: inout RetryOutcome,
- transientDeadline: ContinuousClock.Instant, readDeadline: ContinuousClock.Instant
+ code: UInt32, payload: [UInt8], context: PartitionContext, request: inout VSRFrame.EncodedRequest?, epoch: inout UInt64,
+ outcome: inout RetryOutcome, transientDeadline: ContinuousClock.Instant, readDeadline: ContinuousClock.Instant, replaysUncommitted: Bool
) async -> Result<[UInt8], IggyError> {
guard let connection else {
return .failure(IggyError(.notConnected))
@@ -400,7 +512,7 @@ actor ClientCore {
stamped = request
} else {
do {
- stamped = try VSRFrame.encodeRequest(session: &session, code: code, payload: payload)
+ stamped = try VSRFrame.encodeRequest(session: &session, code: code, payload: payload, partitionContext: context)
} catch let error as IggyError {
return .failure(error)
} catch {
@@ -416,7 +528,9 @@ actor ClientCore {
var refusal: IggyError?
while true {
let remaining = readDeadline - clock.now
- guard remaining > .zero else {
+ // A replay sent with a sliver of the budget would time out, and the
+ // timeout takes the connection with it: the refusal stands instead.
+ guard remaining > .zero, refusal == nil || remaining >= Self.notReadyRetryInterval else {
return .failure(refusal ?? IggyError(.transientNotAccepted, context: "the request deadline passed before command \(code) was sent"))
}
let raw: RawFrame
@@ -424,7 +538,12 @@ actor ClientCore {
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)"))
+ switch error.code {
+ case .disconnected, .tcpError, .notConnected:
+ return .failure(error)
+ default:
+ return .failure(IggyError(.disconnected, context: "\(error)"))
+ }
} catch {
return .failure(IggyError(.disconnected, context: "\(error)"))
}
@@ -436,6 +555,7 @@ actor ClientCore {
let answered = VSRFrame.readReplyRequestID(header).map(String.init) ?? "unknown"
return .failure(IggyError(.disconnected, context: "the reply answers request \(answered), expected \(stamped.requestID)"))
}
+ observeMetadataReply(header)
do {
return .success(Array(try VSRFrame.decodeReply(header: header, body: raw.body[...])))
} catch let answer as IggyError {
@@ -444,7 +564,7 @@ actor ClientCore {
let error = outcome.observe(answer)
refusal = error
switch error.code {
- case .transientNotCommitted where clock.now < readDeadline:
+ case .transientNotCommitted where replaysUncommitted && 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.
@@ -464,6 +584,18 @@ actor ClientCore {
}
}
+ /// Raises the metadata watermark to the commit of a metadata reply, as
+ /// `observe_metadata_reply` in the Rust SDK. Reads and partition replies
+ /// carry no metadata commit.
+ private func observeMetadataReply(_ header: ArraySlice<UInt8>) {
+ guard VSRFrame.peekCommand(header) == .reply, let operation = VSRFrame.readReplyOperation(header),
+ operation != .nonReplicated, !operation.isPartition, let commit = VSRFrame.readReplyCommit(header)
+ else {
+ return
+ }
+ metadataWatermark = max(metadataWatermark, commit)
+ }
+
// MARK: Connecting
/// The connect a caller asked for, which also lifts an earlier explicit
@@ -628,6 +760,21 @@ actor ClientCore {
logger.info("Skipping automatic sign-in for a retried login request")
return false
}
+ if session.isBound {
+ do {
+ try await resumeSession()
+ return settleOffLeader ? false : try await settleAfterSignIn()
+ } catch let error as IggyError where error.code == .unauthenticated || error.code == .transientNotCommitted {
+ // The server no longer holds the session, or cannot tell yet:
+ // the client signs in under a new one.
+ logger.info("Cannot resume the session, signing in again: \(error)")
+ resetSession()
+ } catch let error as IggyError {
+ let connectionLost = error.isConnectionLoss || error.code == .transientNotAccepted
+ await disconnectTransport()
+ throw SignInFailure(error: error, connectionLost: connectionLost)
+ }
+ }
state = .authenticating
do {
switch credentials {
@@ -646,6 +793,10 @@ actor ClientCore {
// back on the node whose partition replica keeps refusing.
return false
}
+ return try await settleAfterSignIn()
+ }
+
+ private func settleAfterSignIn() async throws -> Bool {
do {
return try await handleLeaderRedirection()
} catch let error as IggyError {
@@ -653,6 +804,32 @@ actor ClientCore {
}
}
+ /// Binds a new connection to the session the lost one carried, as
+ /// `resume_vsr_session` in the Rust SDK, so the session keeps its group
+ /// memberships, routes, and request ids.
+ private func resumeSession() async throws {
+ guard let bound = session.session else {
+ return
+ }
+ let clientID = session.clientID
+ state = .authenticating
+ let payload = try Requests.bindSession(
+ clientID: clientID, session: bound, metadataWatermark: metadataWatermark, bindSecret: session.bindSecret)
+ let body = try await exchangeOnce(code: CommandCode.bindSession.rawValue, payload: payload, context: .empty)
+ let resumed = try Responses.loginRegister(body[...])
+ guard resumed.session == bound else {
+ throw IggyError(.sessionMismatch, context: "asked to resume session \(bound), the server bound \(resumed.session)")
+ }
+ // An explicit disconnect or a shutdown while the bind was in flight
+ // replaced the session, and the reply must not revive it.
+ guard session.clientID == clientID, !disconnectedByCaller, state == .authenticating else {
+ throw IggyError(.notConnected, context: "the session was ended while it was being resumed")
+ }
+ logger.info("Resumed session \(bound)")
+ state = .authenticated
+ publish(.signedIn)
+ }
+
private func failSignIn(_ error: IggyError) async -> SignInFailure {
let connectionLost = error.isConnectionLoss
if connectionLost {
@@ -719,39 +896,69 @@ actor ClientCore {
return elapsed < interval ? interval - elapsed : nil
}
- /// Tears the connection down without touching the remembered sign-in, so
- /// a reconnect can restore the session.
+ /// Tears the connection down without touching the remembered sign-in or
+ /// the session, so a reconnect can resume it.
+ ///
+ /// A frame another caller already wrote keeps the connection until its
+ /// reply, as the Rust client's teardown waits on the link: cutting it would
+ /// leave a request that may commit with an unknown outcome, and its replay
+ /// on the next node could apply it twice.
func disconnectTransport() async {
- switch state {
- case .disconnected, .shutdown:
+ guard canTearDown else {
return
- case .connecting:
- // A connect is already sweeping; tearing down under it would take
- // the stream it just installed.
- return
- default:
- break
+ }
+ let epoch = connectionEpoch
+ let holdsExchange = connection != nil
+ if holdsExchange {
+ await waitForExchange()
+ guard canTearDown, epoch == connectionEpoch else {
+ // The caller in flight dropped the connection, or a reconnect
+ // replaced it, while this waited.
+ exchangeLock.unlock()
+ return
+ }
}
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()
+ if holdsExchange {
+ exchangeLock.unlock()
+ }
publish(.disconnected)
}
+ private var canTearDown: Bool {
+ switch state {
+ case .disconnected, .shutdown:
+ false
+ case .connecting:
+ // A connect is already sweeping; tearing down under it would take
+ // the stream it just installed.
+ false
+ default:
+ true
+ }
+ }
+
+ /// Takes the exchange lock once the exchange in flight has its reply. A
+ /// teardown must happen even when the task that asked for it was
+ /// cancelled, so the wait runs outside that task and cannot be abandoned.
+ private func waitForExchange() async {
+ let lock = exchangeLock
+ _ = try? await Task { try await lock.lock() }.value
+ }
+
/// An explicit disconnect ends the session on purpose: the remembered
- /// sign-in goes with it so the next reconnect does not resurrect it.
+ /// sign-in and the session go with it so the next connect does not
+ /// resurrect it.
func disconnect() async {
disconnectedByCaller = true
rememberedSignIn = nil
stopHeartbeat()
await disconnectTransport()
+ resetSession()
}
func shutdown() async {
@@ -761,7 +968,7 @@ actor ClientCore {
stopHeartbeat()
let old = connection
connection = nil
- session.reset()
+ resetSession()
state = .shutdown
await old?.close()
publish(.shutdown)
@@ -818,10 +1025,10 @@ actor ClientCore {
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.
+ // A sign-in on a connection that is not bound to the session, as after
+ // a reconnect with nothing to resume it with, registers a new session.
if session.isBound {
- session.reset()
+ resetSession()
}
// The login body carries the bind secret of the session the frame
// encoder registers, and both survive a retry on a new connection.
@@ -830,7 +1037,7 @@ actor ClientCore {
do {
body = try await exchange(code: code, payload: payload)
} catch {
- session.reset()
+ resetSession()
throw error
}
let response: Responses.LoginRegister
@@ -838,9 +1045,11 @@ actor ClientCore {
response = try Responses.loginRegister(body[...])
try session.bind(response.session)
} catch {
- session.reset()
+ resetSession()
throw error
}
+ // Nothing synced or routed before the bind belongs to this session.
+ endSessionScope()
logger.debug(
"Authenticated against iggy server",
metadata: ["server_version": "\(response.serverVersion)", "server_protocol_version": "\(response.serverProtocolVersion)"])
@@ -855,11 +1064,23 @@ actor ClientCore {
}
_ = try await exchange(code: CommandCode.logoutUser.rawValue, payload: [])
rememberedSignIn = nil
- session.reset()
+ resetSession()
state = .connected
publish(.signedOut)
}
+ /// Starts a new consensus session. What was synced or routed under the
+ /// old one goes with it.
+ private func resetSession() {
+ session.reset()
+ endSessionScope()
+ }
+
+ private func endSessionScope() {
+ groupState.clearSessionScoped()
+ pollRouter.clearSession(fence: session.clientID)
+ }
+
private func rememberSignIn(_ credentials: Credentials, userID: UInt32) {
rememberedSignIn = (credentials, userID)
}
@@ -942,13 +1163,23 @@ actor ClientCore {
private func readTransportEndpoints() async -> [String] {
do {
let body = try await exchange(code: CommandCode.getClusterMetadata.rawValue, payload: [])
- return Self.transportEndpoints(try Responses.clusterMetadata(body[...]))
+ let metadata = try Responses.clusterMetadata(body[...])
+ learnRosterSize(metadata)
+ return Self.transportEndpoints(metadata)
} catch {
logger.debug("Failed to read the cluster roster: \(error)")
return []
}
}
+ /// Tells the router whether polls go to data connections. An empty
+ /// roster says nothing about the cluster, so the router keeps asking.
+ private func learnRosterSize(_ metadata: ClusterMetadata) {
+ if !metadata.nodes.isEmpty {
+ pollRouter.learnRosterSize(metadata.nodes.count)
+ }
+ }
+
static func transportEndpoints(_ metadata: ClusterMetadata) -> [String] {
metadata.nodes.filter { $0.endpoints.tcp != 0 }.map(\.tcpAddress)
}
@@ -993,6 +1224,7 @@ actor ClientCore {
logger.debug("Failed to get cluster metadata: \(error), continuing on \(currentAddress)")
return LeaderCheck(redirect: nil, endpoints: [])
}
+ learnRosterSize(metadata)
let endpoints = Self.transportEndpoints(metadata)
if metadata.nodes.count <= 1 {
return LeaderCheck(redirect: nil, endpoints: endpoints)
@@ -1065,7 +1297,19 @@ actor ClientCore {
struct RetryOutcome {
private(set) var isUncertain = false
+ /// Records an attempt that was written and lost with its connection.
+ mutating func markUncertain() {
+ isUncertain = true
+ }
+
mutating func observe(_ error: IggyError) -> IggyError {
+ if error.code == .requestTooOld {
+ // The server took a later request id of this session first, which
+ // a replay or a data connection can cause. An earlier attempt of
+ // this one may have committed.
+ isUncertain = true
+ return IggyError(.transientNotCommitted, context: "the server already took a later request: \(error)")
+ }
if error.code == .transientNotCommitted {
isUncertain = true
}
diff --git a/foreign/swift/Sources/Iggy/Client/ConsumerGroupState.swift b/foreign/swift/Sources/Iggy/Client/ConsumerGroupState.swift
new file mode 100644
index 000000000..15c35e4f8
--- /dev/null
+++ b/foreign/swift/Sources/Iggy/Client/ConsumerGroupState.swift
@@ -0,0 +1,223 @@
+// 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
+
+/// Client-side partitioning state, a port of `ConsumerGroupClientState` in
+/// `core/common`.
+///
+/// The server routes explicit partitions only, so the client resolves the
+/// rest before it encodes a request: a consumer-group poll takes the next of
+/// the member's assigned partitions, round-robin, from the assignment synced
+/// from the coordinator; a `balanced` send round-robins per topic, and a
+/// messages key is hashed. The cursors outlive single calls, so this lives on
+/// the client. The coordinator fences a stale selection after a rebalance,
+/// which makes the client sync again and restart the cursor.
+final class ConsumerGroupState: Sendable {
+ struct TopicKey: Hashable, Sendable {
+ let stream: Identifier
+ let topic: Identifier
+ }
+
+ struct GroupKey: Hashable, Sendable {
+ let stream: Identifier
+ let topic: Identifier
+ let group: Identifier
+ }
+
+ private struct Assignment {
+ var partitions: [UInt32] = []
+ var generation: UInt64 = 0
+ var cursor: UInt = 0
+ }
+
+ private struct TopicPartitions {
+ var count: UInt32?
+ var contexts: [UInt32: PartitionContext] = [:]
+ var cursor: UInt = 0
+ }
+
+ private struct Storage {
+ var assignments: [GroupKey: Assignment] = [:]
+ var topics: [TopicKey: TopicPartitions] = [:]
+ /// The groups the last sync found this client a member of, so the
+ /// heartbeat can sync them again.
+ var joinedGroups: Set<GroupKey> = []
+ /// Moves on with every session that ends, so a sync answered under a
+ /// replaced session is told apart.
+ var sessionScope: UInt64 = 0
+ }
+
+ private let storage = NIOLockedValueBox(Storage())
+
+ // MARK: Group assignments
+
+ /// Whether a non-empty assignment is cached for the group.
+ func hasAssignment(_ key: GroupKey) -> Bool {
+ storage.withLockedValue { $0.assignments[key].map { !$0.partitions.isEmpty } ?? false }
+ }
+
+ /// The session scope a sync reads before it is sent and hands to
+ /// ``applySync(_:member:scope:)``.
+ var sessionScope: UInt64 {
+ storage.withLockedValue { $0.sessionScope }
+ }
+
+ /// Records what a sync read under `scope` found: the member's assignment,
+ /// or nil for a client that is not a member. A new generation, a
+ /// rebalance, restarts the round-robin. A reply of a session that ended
+ /// since is dropped, so it cannot repopulate the session that replaced
+ /// it. Reports whether the reply was applied.
+ @discardableResult
+ func applySync(_ key: GroupKey, member: (generation: UInt64, partitions: [UInt32])?, scope: UInt64) -> Bool {
+ storage.withLockedValue { storage in
+ guard storage.sessionScope == scope else {
+ return false
+ }
+ guard let member else {
+ storage.assignments.removeValue(forKey: key)
+ storage.joinedGroups.remove(key)
+ return true
+ }
+ storage.joinedGroups.insert(key)
+ var assignment = storage.assignments[key] ?? Assignment()
+ if assignment.generation != member.generation {
+ assignment.cursor = 0
+ }
+ assignment.generation = member.generation
+ assignment.partitions = member.partitions
+ storage.assignments[key] = assignment
+ return true
+ }
+ }
+
+ /// Drops the group's assignment, so the next poll syncs it again.
+ func invalidateAssignment(_ key: GroupKey) {
+ _ = storage.withLockedValue { $0.assignments.removeValue(forKey: key) }
+ }
+
+ /// The next assigned partition to poll, or nil when nothing is assigned.
+ func nextGroupPartition(_ key: GroupKey) -> UInt32? {
+ storage.withLockedValue { storage in
+ guard var assignment = storage.assignments[key], !assignment.partitions.isEmpty else {
+ return nil
+ }
+ let partition = assignment.partitions[Int(assignment.cursor % UInt(assignment.partitions.count))]
+ assignment.cursor &+= 1
+ storage.assignments[key] = assignment
+ return partition
+ }
+ }
+
+ // MARK: Membership
+
+ /// Forgets a group after leaving it.
+ func deregisterGroup(_ key: GroupKey) {
+ _ = storage.withLockedValue { $0.joinedGroups.remove(key) }
+ }
+
+ /// Whether the last sync found the client a member. A member mid-rebalance
+ /// or holding no partitions is still one, unlike ``hasAssignment(_:)``.
+ func isRegistered(_ key: GroupKey) -> Bool {
+ storage.withLockedValue { $0.joinedGroups.contains(key) }
+ }
+
+ var registeredGroups: [GroupKey] {
+ storage.withLockedValue { Array($0.joinedGroups) }
+ }
+
+ // MARK: Topic partitions
+
+ /// The next `balanced` partition of the topic, advancing its cursor.
+ func nextBalancedPartition(_ key: TopicKey, partitionCount: UInt32) -> UInt32 {
+ guard partitionCount > 0 else {
+ return 0
+ }
+ return storage.withLockedValue { storage in
+ var topic = storage.topics[key] ?? TopicPartitions()
+ let partition = UInt32(topic.cursor % UInt(partitionCount))
+ topic.cursor &+= 1
+ storage.topics[key] = topic
+ return partition
+ }
+ }
+
+ func partitionCount(_ key: TopicKey) -> UInt32? {
+ storage.withLockedValue { $0.topics[key]?.count }
+ }
+
+ /// Caches what a topic read reported: the partition count, and each
+ /// partition's context for the sends that follow.
+ func setTopicPartitions(_ key: TopicKey, partitions: [Partition]) {
+ guard let count = UInt32(exactly: partitions.count) else {
+ return
+ }
+ storage.withLockedValue { storage in
+ var topic = storage.topics[key] ?? TopicPartitions()
+ topic.count = count
+ topic.contexts = Dictionary(partitions.map { ($0.id, $0.context) }, uniquingKeysWith: { _, last in last })
+ storage.topics[key] = topic
+ }
+ }
+
+ func partitionContext(_ key: TopicKey, partitionID: UInt32) -> PartitionContext? {
+ storage.withLockedValue { $0.topics[key]?.contexts[partitionID] }
+ }
+
+ func setPartitionContext(_ key: TopicKey, partitionID: UInt32, _ context: PartitionContext) {
+ storage.withLockedValue { storage in
+ var topic = storage.topics[key] ?? TopicPartitions()
+ topic.contexts[partitionID] = context
+ storage.topics[key] = topic
+ }
+ }
+
+ /// Drops the topic's partition count and contexts after the server said
+ /// they are stale. The cursor stays.
+ func invalidateTopic(_ key: TopicKey) {
+ storage.withLockedValue { storage in
+ storage.topics[key]?.count = nil
+ storage.topics[key]?.contexts.removeAll()
+ }
+ }
+
+ /// Drops every topic's partition count and contexts after a change that
+ /// can renumber partitions. A name and a numeric id can cache the same
+ /// topic under two keys, so no single key can be dropped instead.
+ func invalidateTopicDiscovery() {
+ storage.withLockedValue { storage in
+ for key in storage.topics.keys {
+ storage.topics[key]?.count = nil
+ storage.topics[key]?.contexts.removeAll()
+ }
+ }
+ }
+
+ /// Drops what belonged to the consensus session. Membership is per
+ /// session and the coordinator fences assignments by a generation it
+ /// tracks per session, so nothing synced under a replaced session holds.
+ /// The balanced cursors and the partition counts belong to a topic and
+ /// stay, so a reconnect neither restarts the round-robin at partition 0
+ /// nor costs a topic read per topic.
+ func clearSessionScoped() {
+ storage.withLockedValue { storage in
+ storage.assignments.removeAll()
+ storage.joinedGroups.removeAll()
+ storage.sessionScope &+= 1
+ }
+ }
+}
diff --git a/foreign/swift/Sources/Iggy/Client/IggyClient+Messages.swift b/foreign/swift/Sources/Iggy/Client/IggyClient+Messages.swift
new file mode 100644
index 000000000..903b53b9f
--- /dev/null
+++ b/foreign/swift/Sources/Iggy/Client/IggyClient+Messages.swift
@@ -0,0 +1,410 @@
+// 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: - Messages
+
+extension IggyClient {
+ /// One sync and one retry: a poll the coordinator fences for a stale
+ /// assignment syncs it again and polls once more.
+ private static let groupPollMaxAttempts = 2
+
+ /// Sends a batch of messages to one partition. Delivery is at least once.
+ ///
+ /// The server routes explicit partitions only, so `balanced` and
+ /// `messagesKey` partitioning are resolved on the client: balanced
+ /// round-robins over the topic's partitions, a key is hashed with XXH32.
+ /// A zero message id is replaced with a random one and written back into
+ /// `messages`, as the Rust client does, so a send retried after
+ /// ``IggyErrorCode/transientNotCommitted`` carries the same ids.
+ ///
+ /// The returned confirmations may be empty: the server can commit a batch
+ /// it has no offsets to describe.
+ @discardableResult
+ public func sendMessages(
+ streamID: Identifier, topicID: Identifier, partitioning: Partitioning = .balanced, messages: inout [IggyMessage]
+ ) async throws -> SendMessagesResponse {
+ try await requireAuthenticated()
+ // Checked before a balanced or keyed send reads the topic.
+ guard !messages.isEmpty else {
+ throw IggyError(.invalidMessagesCount, context: "a batch needs at least one message")
+ }
+ // The key is hashed to a partition before the batch is encoded, so its
+ // length is checked first.
+ try partitioning.validate()
+ let partitionID = try await resolvePartition(partitioning, streamID: streamID, topicID: topicID)
+ for index in messages.indices where messages[index].id.isZero {
+ messages[index].id = .random()
+ }
+ let outgoing = messages.map { message in
+ Batch.OutgoingMessage(
+ id: message.id, originTimestamp: message.originTimestamp.microseconds, payload: message.payload, userHeaders: message.rawUserHeaders)
+ }
+ let payload = try Batch.encodeSend(streamID: streamID, topicID: topicID, partitioning: .partition(partitionID), messages: outgoing)
+ let body = try await sendWithPartitionContext(streamID: streamID, topicID: topicID, partitionID: partitionID, payload: payload)
+ // The batch committed once the reply arrived. An unreadable
+ // confirmation must not make a retrying caller write it twice.
+ return (try? Responses.sendConfirmations(body[...])) ?? SendMessagesResponse()
+ }
+
+ /// Polls up to `count` messages.
+ ///
+ /// A consumer-group poll with `partitionID` nil takes the next of the
+ /// member's assigned partitions, round-robin, syncing the assignment from
+ /// the coordinator when needed. A member holding no partitions gets an
+ /// empty result whose partition id is ``PolledMessages/noAssignedPartition``,
+ /// and a client that is not a member gets
+ /// ``IggyErrorCode/consumerGroupMemberNotFound``, so it can join again.
+ ///
+ /// With `autoCommit` the server stores the offset of the last polled
+ /// message as consumed. A `strategy` with a ``PollingStrategy/context``
+ /// is refused once its partition was recreated.
+ public func pollMessages(
+ streamID: Identifier, topicID: Identifier, partitionID: UInt32?, consumer: Consumer = .default, strategy: PollingStrategy, count: UInt32,
+ autoCommit: Bool
+ ) async throws -> PolledMessages {
+ try await pollMessages(streamID: streamID, topicID: topicID, partitionID: partitionID, consumer: consumer, count: count, autoCommit: autoCommit) {
+ _ in strategy
+ }
+ }
+
+ /// ``pollMessages(streamID:topicID:partitionID:consumer:strategy:count:autoCommit:)``
+ /// with the strategy chosen once the partition is known, so a caller can
+ /// continue each partition from its own position.
+ public func pollMessages(
+ streamID: Identifier, topicID: Identifier, partitionID: UInt32?, consumer: Consumer = .default, count: UInt32, autoCommit: Bool,
+ strategyFor: @Sendable (UInt32) -> PollingStrategy
+ ) async throws -> PolledMessages {
+ try await requireAuthenticated()
+ if consumer.kind == .consumerGroup, partitionID == nil {
+ return try await pollGroup(streamID: streamID, topicID: topicID, consumer: consumer, count: count, autoCommit: autoCommit, strategyFor: strategyFor)
+ }
+ let strategy = strategyFor(partitionID ?? 0)
+ let payload = try Requests.pollMessages(
+ consumer: consumer, streamID: streamID, topicID: topicID, partitionID: partitionID, strategy: strategy, count: count, autoCommit: autoCommit)
+ return try Responses.polledMessages(try await routedPoll(payload, context: strategy.context)[...])
+ }
+
+ // MARK: Partition contexts
+
+ /// Sends under the partition's cached context, or one read from its
+ /// route. A history refusal means the context is stale: the topic's
+ /// contexts are dropped and one fresh route is tried.
+ private func sendWithPartitionContext(streamID: Identifier, topicID: Identifier, partitionID: UInt32, payload: [UInt8]) async throws -> [UInt8] {
+ let key = ConsumerGroupState.TopicKey(stream: streamID, topic: topicID)
+ let state = core.groupState
+ var context = state.partitionContext(key, partitionID: partitionID)
+ var refreshAvailable = true
+ while true {
+ let captured: PartitionContext
+ if let context {
+ captured = context
+ } else {
+ let body = try await send(.getPartitionRoute, Requests.partitionRoute(streamID: streamID, topicID: topicID, partitionID: partitionID))
+ captured = try Responses.partitionRoute(body[...])
+ state.setPartitionContext(key, partitionID: partitionID, captured)
+ }
+ do {
+ return try await core.exchange(code: CommandCode.sendMessages.rawValue, payload: payload, context: captured)
+ } catch let error as IggyError where error.code == .historyUnavailable {
+ state.invalidateTopic(key)
+ guard refreshAvailable else {
+ throw error
+ }
+ refreshAvailable = false
+ context = nil
+ }
+ }
+ }
+
+ /// A poll on its route. The route key is the poll without its strategy,
+ /// count, and auto-commit flag, which do not change where it goes.
+ func routedPoll(_ payload: [UInt8], context: PartitionContext?) async throws -> [UInt8] {
+ let key = PollRouter.RouteKey(code: .getPollRouting, prefix: payload.dropLast(RawRequest.pollParametersSize))
+ return try await core.pollRouter.send(.pollMessagesOnPrimary, payload: payload, key: key, routePayload: payload, context: context, coordinator: core)
+ }
+
+ /// An offset write on its route. `routePayload` is the consumer, stream,
+ /// topic, and partition the write names.
+ func routedOffsetWrite(_ code: CommandCode, payload: [UInt8], routePayload: [UInt8], context: PartitionContext?) async throws {
+ let key = PollRouter.RouteKey(code: .getConsumerOffsetRouting, prefix: routePayload[...])
+ _ = try await core.pollRouter.send(code, payload: payload, key: key, routePayload: routePayload, context: context, coordinator: core)
+ }
+
+ // MARK: Client-side partitioning
+
+ private func resolvePartition(_ partitioning: Partitioning, streamID: Identifier, topicID: Identifier) async throws -> UInt32 {
+ switch partitioning {
+ case .partition(let partitionID):
+ return partitionID
+ case .balanced:
+ let count = try await partitionCount(streamID: streamID, topicID: topicID)
+ return core.groupState.nextBalancedPartition(.init(stream: streamID, topic: topicID), partitionCount: count)
+ case .messagesKey(let key):
+ return XXH32.hash(key) % (try await partitionCount(streamID: streamID, topicID: topicID))
+ }
+ }
+
+ /// The topic's partition count, cached by the topic read so a send does
+ /// not pay a metadata round trip per batch.
+ private func partitionCount(streamID: Identifier, topicID: Identifier) async throws -> UInt32 {
+ var count = core.groupState.partitionCount(.init(stream: streamID, topic: topicID))
+ if count == nil {
+ count = try await getTopic(streamID: streamID, topicID: topicID)?.partitionsCount
+ }
+ guard let count, count > 0 else {
+ throw IggyError(.topicIdNotFound, context: "topic \(topicID) in stream \(streamID)")
+ }
+ return count
+ }
+
+ // MARK: Consumer-group polling
+
+ /// Syncs the member's assignment from the coordinator into the cache. An
+ /// empty reply means the client is not a member. A reply that arrives
+ /// after the session was replaced is dropped.
+ func syncGroupAssignment(streamID: Identifier, topicID: Identifier, groupID: Identifier) async throws {
+ let state = core.groupState
+ let scope = state.sessionScope
+ let body = try await send(.syncConsumerGroup, Requests.consumerGroup(streamID: streamID, topicID: topicID, groupID: groupID))
+ let key = ConsumerGroupState.GroupKey(stream: streamID, topic: topicID, group: groupID)
+ if body.isEmpty {
+ // The coordinator answers any member, even one holding no
+ // partitions, with an assignment, so this observes a removal: a
+ // deleted group or an evicted member. Without the deregister every
+ // later poll would come back empty instead of telling the caller.
+ state.applySync(key, member: nil, scope: scope)
+ return
+ }
+ let assignment = try Responses.consumerGroupAssignment(body[...])
+ state.applySync(key, member: (assignment.generation, assignment.partitions), scope: scope)
+ }
+
+ /// Syncs every joined group, so a member picks up a widened assignment
+ /// without first hitting an ownership fence. The heartbeat runs it; one
+ /// group that fails does not stop the others.
+ func refreshConsumerGroupAssignments() async {
+ for key in core.groupState.registeredGroups {
+ do {
+ try await syncGroupAssignment(streamID: key.stream, topicID: key.topic, groupID: key.group)
+ } catch {
+ logger.warning("Failed to refresh the consumer group assignment of \(key.stream)|\(key.topic)|\(key.group): \(error)")
+ }
+ }
+ }
+
+ private func pollGroup(
+ streamID: Identifier, topicID: Identifier, consumer: Consumer, count: UInt32, autoCommit: Bool, strategyFor: @Sendable (UInt32) -> PollingStrategy
+ ) async throws -> PolledMessages {
+ let key = ConsumerGroupState.GroupKey(stream: streamID, topic: topicID, group: consumer.id)
+ let state = core.groupState
+ if !state.hasAssignment(key) {
+ try await syncGroupAssignment(streamID: streamID, topicID: topicID, groupID: consumer.id)
+ }
+ for _ in 0..<Self.groupPollMaxAttempts {
+ guard let partitionID = state.nextGroupPartition(key) else {
+ // A member can hold no partitions, while a client that is not
+ // a member must learn it, or it polls an empty assignment
+ // forever instead of joining again.
+ guard state.isRegistered(key) else {
+ throw IggyError(.consumerGroupMemberNotFound, context: "group \(consumer.id) in topic \(topicID) of stream \(streamID)")
+ }
+ var empty = PolledMessages.empty
+ empty.partitionID = PolledMessages.noAssignedPartition
+ return empty
+ }
+ // A retry after a fence can land on another partition, so the
+ // strategy is asked for each attempt.
+ let strategy = strategyFor(partitionID)
+ let payload = try Requests.pollMessages(
+ consumer: consumer, streamID: streamID, topicID: topicID, partitionID: partitionID, strategy: strategy, count: count, autoCommit: autoCommit)
+ do {
+ let polled = try Responses.polledMessages(try await routedPoll(payload, context: strategy.context)[...])
+ if polled.messages.isEmpty, polled.partitionID == PolledMessages.resyncRequiredPartition {
+ // The coordinator fences a stale assignment through the
+ // empty poll, for want of a typed error.
+ state.invalidateAssignment(key)
+ try await syncGroupAssignment(streamID: streamID, topicID: topicID, groupID: consumer.id)
+ continue
+ }
+ return polled
+ } catch let error as IggyError where error.code == .consumerGroupPartitionNotOwned {
+ state.invalidateAssignment(key)
+ try await syncGroupAssignment(streamID: streamID, topicID: topicID, groupID: consumer.id)
+ if strategy.context != nil {
+ // A continuation of the partition the member lost cannot
+ // move to another one.
+ throw error
+ }
+ }
+ }
+ // Every attempt met a rebalance. The assignment is synced again and
+ // an empty poll lets the caller poll again, where an error would make
+ // a normal rebalance look like a failure.
+ return .empty
+ }
+}
+
+// MARK: - Consumer offsets
+
+extension IggyClient {
+ /// Stores the offset of a consumer on a partition. A consumer-group
+ /// consumer stores it for the group.
+ public func storeConsumerOffset(
+ consumer: Consumer = .default, streamID: Identifier, topicID: Identifier, partitionID: UInt32?, offset: UInt64
+ )
+ async throws
+ {
+ try await requireAuthenticated()
+ try await routedOffsetWrite(
+ .storeConsumerOffset,
+ payload: Requests.storeConsumerOffset(
+ consumer: consumer, streamID: streamID, topicID: topicID, partitionID: partitionID, offset: offset, ack: .quorum),
+ routePayload: Requests.getConsumerOffset(consumer: consumer, streamID: streamID, topicID: topicID, partitionID: partitionID), context: nil)
+ }
+
+ /// Stores a checkpoint under the partition context of the poll that
+ /// delivered it. It is refused once the partition was recreated or, for a
+ /// consumer group, reassigned since.
+ public func storeConsumerPosition(consumer: Consumer = .default, streamID: Identifier, topicID: Identifier, position: ConsumerPosition) async throws {
+ try await requireAuthenticated()
+ try await routedOffsetWrite(
+ .storeConsumerOffset,
+ payload: Requests.storeConsumerOffset(
+ consumer: consumer, streamID: streamID, topicID: topicID, partitionID: position.partitionID, offset: position.offset, ack: .quorum),
+ routePayload: Requests.getConsumerOffset(consumer: consumer, streamID: streamID, topicID: topicID, partitionID: position.partitionID),
+ context: position.context)
+ }
+
+ /// The stored offset of a consumer on a partition, or nil when none is
+ /// stored.
+ public func getConsumerOffset(
+ consumer: Consumer = .default, streamID: Identifier, topicID: Identifier, partitionID: UInt32?
+ ) async throws
+ -> ConsumerOffsetInfo?
+ {
+ try await requireAuthenticated()
+ let body = try await send(
+ .getConsumerOffset, Requests.getConsumerOffset(consumer: consumer, streamID: streamID, topicID: topicID, partitionID: partitionID))
+ return body.isEmpty ? nil : try Responses.consumerOffset(body[...])
+ }
+
+ /// Deletes the stored offset of a consumer on a partition.
+ public func deleteConsumerOffset(consumer: Consumer = .default, streamID: Identifier, topicID: Identifier, partitionID: UInt32?) async throws {
+ try await requireAuthenticated()
+ try await routedOffsetWrite(
+ .deleteConsumerOffset,
+ payload: Requests.deleteConsumerOffset(consumer: consumer, streamID: streamID, topicID: topicID, partitionID: partitionID, ack: .quorum),
+ routePayload: Requests.getConsumerOffset(consumer: consumer, streamID: streamID, topicID: topicID, partitionID: partitionID), context: nil)
+ }
+}
+
+// MARK: - Raw partition requests
+
+extension IggyClient {
+ /// Sends a raw partition request the way its typed call sends it, so it
+ /// carries the partition context the server requires. Nil for any other
+ /// code.
+ func sendRawPartitionRequest(code: UInt32, payload: [UInt8]) async throws -> [UInt8]? {
+ switch CommandCode(rawValue: code) {
+ case .sendMessages:
+ let target = try RawRequest.sendTarget(payload)
+ return try await sendWithPartitionContext(streamID: target.stream, topicID: target.topic, partitionID: target.partitionID, payload: payload)
+ case .pollMessages:
+ try RawRequest.validatePoll(payload)
+ return try await routedPoll(payload, context: nil)
+ case .storeConsumerOffset, .deleteConsumerOffset:
+ let prefix = try RawRequest.consumerTargetPrefix(payload)
+ let command: CommandCode = code == CommandCode.storeConsumerOffset.rawValue ? .storeConsumerOffset : .deleteConsumerOffset
+ let key = PollRouter.RouteKey(code: .getConsumerOffsetRouting, prefix: prefix[...])
+ return try await core.pollRouter.send(command, payload: payload, key: key, routePayload: prefix, context: nil, coordinator: core)
+ case .getPollRouting, .pollMessagesOnPrimary, .getConsumerOffsetRouting:
+ // The routing protocol itself, sent once as the Rust client sends
+ // it: a replay after a reconnect could run under another session.
+ // A dropped client reconnects first, as for a typed call.
+ try await requireAuthenticated()
+ return try await core.exchangeOnce(code: code, payload: payload, context: .empty)
+ default:
+ return nil
+ }
+ }
+}
+
+/// Reads the partition a raw request names. A payload that does not decode
+/// fails with ``IggyErrorCode/invalidCommand`` before anything is sent.
+private enum RawRequest {
+ /// Strategy kind and value, count, and the auto-commit flag.
+ static let pollParametersSize = 9 + 4 + 1
+
+ struct SendTarget {
+ let stream: Identifier
+ let topic: Identifier
+ let partitionID: UInt32
+ }
+
+ /// `[metadata_len:u32][stream][topic][partitioning][messages_count:u32]`
+ /// leads the batch, and the fields must fit the declared length. Balanced
+ /// and key partitioning are resolved by the typed send from its cache and
+ /// cursor, so a raw send names no partition to route.
+ static func sendTarget(_ payload: [UInt8]) throws -> SendTarget {
+ let (stream, topic, partitioning) = try decode(payload) { reader in
+ var metadata = ByteReader(try reader.readPrefixedBytes())
+ let stream = try Identifier.decode(from: &metadata)
+ let topic = try Identifier.decode(from: &metadata)
+ let partitioning = try Partitioning.decode(from: &metadata)
+ _ = try metadata.readUInt32()
+ return (stream, topic, partitioning)
+ }
+ guard case .partition(let partitionID) = partitioning else {
+ throw IggyError(
+ .featureUnavailable, context: "a raw send needs an explicit partition id; use sendMessages for balanced or key partitioning")
+ }
+ return SendTarget(stream: stream, topic: topic, partitionID: partitionID)
+ }
+
+ /// The consumer, stream, topic, and partition that lead a poll and an
+ /// offset request, which is the payload of the offset route lookup.
+ static func consumerTargetPrefix(_ payload: [UInt8]) throws -> [UInt8] {
+ let length = try decode(payload) { reader in
+ try skipConsumerTarget(&reader)
+ return reader.position
+ }
+ return Array(payload[..<length])
+ }
+
+ static func validatePoll(_ payload: [UInt8]) throws {
+ try decode(payload) { reader in
+ try skipConsumerTarget(&reader)
+ try reader.skip(pollParametersSize)
+ }
+ }
+
+ private static func skipConsumerTarget(_ reader: inout ByteReader) throws {
+ _ = try reader.readUInt8()
+ _ = try Identifier.decode(from: &reader)
+ _ = try Identifier.decode(from: &reader)
+ _ = try Identifier.decode(from: &reader)
+ try reader.skip(1 + 4)
+ }
+
+ private static func decode<T>(_ payload: [UInt8], _ body: (inout ByteReader) throws -> T) throws -> T {
+ do {
+ return try Responses.decode(payload[...], body)
+ } catch let error as IggyError {
+ throw IggyError(.invalidCommand, context: "the raw payload does not decode: \(error)")
+ }
+ }
+}
diff --git a/foreign/swift/Sources/Iggy/Client/IggyClient+Resources.swift b/foreign/swift/Sources/Iggy/Client/IggyClient+Resources.swift
index a57284887..1d5b4dcdf 100644
--- a/foreign/swift/Sources/Iggy/Client/IggyClient+Resources.swift
+++ b/foreign/swift/Sources/Iggy/Client/IggyClient+Resources.swift
@@ -48,6 +48,7 @@ extension IggyClient {
public func deleteStream(_ streamID: Identifier) async throws {
try await requireAuthenticated()
_ = try await send(.deleteStream, Requests.identifierOnly(streamID))
+ core.groupState.invalidateTopicDiscovery()
}
}
@@ -58,7 +59,14 @@ extension IggyClient {
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[...])
+ guard !body.isEmpty else {
+ return nil
+ }
+ let details = try Responses.topicDetails(body[...])
+ // Client-side partitioning and the sends that follow read the count
+ // and the partition contexts from here.
+ core.groupState.setTopicPartitions(.init(stream: streamID, topic: topicID), partitions: details.partitions)
+ return details
}
/// Every topic of a stream.
@@ -88,6 +96,7 @@ extension IggyClient {
public func deleteTopic(streamID: Identifier, topicID: Identifier) async throws {
try await requireAuthenticated()
_ = try await send(.deleteTopic, Requests.streamAndTopic(streamID, topicID))
+ core.groupState.invalidateTopicDiscovery()
}
}
@@ -98,12 +107,14 @@ extension IggyClient {
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))
+ core.groupState.invalidateTopicDiscovery()
}
/// 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))
+ core.groupState.invalidateTopicDiscovery()
}
/// Removes the oldest `count` closed segments of a partition.
@@ -113,6 +124,54 @@ extension IggyClient {
}
}
+// MARK: - Consumer groups
+
+extension IggyClient {
+ /// A consumer group with its members, or nil if it does not exist.
+ public func getConsumerGroup(streamID: Identifier, topicID: Identifier, groupID: Identifier) async throws -> ConsumerGroupDetails? {
+ try await requireAuthenticated()
+ let body = try await send(.getConsumerGroup, Requests.consumerGroup(streamID: streamID, topicID: topicID, groupID: groupID))
+ return body.isEmpty ? nil : try Responses.consumerGroupDetails(body[...])
+ }
+
+ /// Every consumer group of a topic.
+ public func getConsumerGroups(streamID: Identifier, topicID: Identifier) async throws -> [ConsumerGroup] {
+ try await requireAuthenticated()
+ let body = try await send(.getConsumerGroups, Requests.streamAndTopic(streamID, topicID))
+ return body.isEmpty ? [] : try Responses.consumerGroups(body[...])
+ }
+
+ /// Creates a consumer group and returns it.
+ public func createConsumerGroup(streamID: Identifier, topicID: Identifier, name: String) async throws -> ConsumerGroupDetails {
+ try await requireAuthenticated()
+ return try Responses.consumerGroupDetails(
+ try await send(.createConsumerGroup, try Requests.createConsumerGroup(streamID: streamID, topicID: topicID, name: name))[...])
+ }
+
+ /// Deletes a consumer group with its stored offsets.
+ public func deleteConsumerGroup(streamID: Identifier, topicID: Identifier, groupID: Identifier) async throws {
+ try await requireAuthenticated()
+ _ = try await send(.deleteConsumerGroup, Requests.consumerGroup(streamID: streamID, topicID: topicID, groupID: groupID))
+ }
+
+ /// Joins a consumer group. Joining changes the member's assignment and
+ /// the group generation, so the next group poll syncs the assignment.
+ public func joinConsumerGroup(streamID: Identifier, topicID: Identifier, groupID: Identifier) async throws {
+ try await requireAuthenticated()
+ _ = try await send(.joinConsumerGroup, Requests.consumerGroup(streamID: streamID, topicID: topicID, groupID: groupID))
+ core.groupState.invalidateAssignment(.init(stream: streamID, topic: topicID, group: groupID))
+ }
+
+ /// Leaves a consumer group. The heartbeat stops syncing its assignment.
+ public func leaveConsumerGroup(streamID: Identifier, topicID: Identifier, groupID: Identifier) async throws {
+ try await requireAuthenticated()
+ _ = try await send(.leaveConsumerGroup, Requests.consumerGroup(streamID: streamID, topicID: topicID, groupID: groupID))
+ let key = ConsumerGroupState.GroupKey(stream: streamID, topic: topicID, group: groupID)
+ core.groupState.invalidateAssignment(key)
+ core.groupState.deregisterGroup(key)
+ }
+}
+
// MARK: - Users
extension IggyClient {
diff --git a/foreign/swift/Sources/Iggy/Client/IggyClient.swift b/foreign/swift/Sources/Iggy/Client/IggyClient.swift
index cee5cd4ad..6f4e47905 100644
--- a/foreign/swift/Sources/Iggy/Client/IggyClient.swift
+++ b/foreign/swift/Sources/Iggy/Client/IggyClient.swift
@@ -22,7 +22,9 @@ import Logging
/// 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.
+/// client feeds many producers and consumers. In a cluster, polls and
+/// offset writes go to each partition's primary over a second connection
+/// bound to the same session.
///
/// ```swift
/// let client = IggyClient(address: "localhost:8090", autoLogin: .usernamePassword(username: "iggy", password: "iggy"))
@@ -116,6 +118,7 @@ public final class IggyClient: Sendable {
/// 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.
+ /// After a ping that went through, the joined consumer groups are synced.
private func heartbeat() async {
guard await core.state != .shutdown else {
return
@@ -124,7 +127,9 @@ public final class IggyClient: Sendable {
try await ping()
} catch {
logger.debug("Heartbeat ping failed: \(error)")
+ return
}
+ await refreshConsumerGroupAssignments()
}
// MARK: Authentication
@@ -165,12 +170,28 @@ public final class IggyClient: Sendable {
/// 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()``.
+ /// the wire format. Session-control codes, `BindSession` among them, are
+ /// rejected because they must go through ``login(username:password:)``
+ /// and ``logout()``.
+ ///
+ /// Partition requests carry the partition context their typed call
+ /// sends: `SendMessages` with an explicit partition id, `PollMessages`,
+ /// `StoreConsumerOffset`, and `DeleteConsumerOffset` take the route of
+ /// ``sendMessages(streamID:topicID:partitioning:messages:)``,
+ /// ``pollMessages(streamID:topicID:partitionID:consumer:strategy:count:autoCommit:)``,
+ /// and the offset calls. A raw send with balanced or key partitioning
+ /// fails with ``IggyErrorCode/featureUnavailable``, and a partition
+ /// payload that does not decode fails with ``IggyErrorCode/invalidCommand``,
+ /// before anything is sent. The routing codes `GetPollRouting`,
+ /// `PollMessagesOnPrimary`, and `GetConsumerOffsetRouting` are sent once
+ /// on the current connection, with no reconnect or replay.
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")
}
+ if let routed = try await sendRawPartitionRequest(code: code, payload: payload) {
+ return routed
+ }
return try await core.exchange(code: code, payload: payload)
}
diff --git a/foreign/swift/Sources/Iggy/Client/PollRouter.swift b/foreign/swift/Sources/Iggy/Client/PollRouter.swift
new file mode 100644
index 000000000..84a2daf80
--- /dev/null
+++ b/foreign/swift/Sources/Iggy/Client/PollRouter.swift
@@ -0,0 +1,394 @@
+// 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
+
+/// Routes polls and consumer-offset writes to the primary of their
+/// partition, a port of `PollRouter` in `core/sdk/src/driver/poll_router.rs`.
+///
+/// A request first asks the coordinator, the connection the client signed in
+/// on, for its route (`GetPollRouting` or `GetConsumerOffsetRouting`): the
+/// partition primary and the partition context the request carries. On a
+/// single node the request then goes over the coordinator connection. In a
+/// cluster it goes over a data connection to the primary, bound to the
+/// coordinator's session with `BindSession`, so consumer-group membership
+/// stays with the coordinator. Routes are cached until a request on them
+/// fails or the session is replaced.
+final class PollRouter: Sendable {
+ /// Bounds one routed request across its lookups, attempts, and pauses.
+ static let pollTimeout: Duration = .seconds(30)
+ static let retryInterval: Duration = .milliseconds(50)
+ static let maxRetryInterval: Duration = .seconds(1)
+ static let rosterReadTimeout: Duration = .seconds(5)
+ static let maxCachedRoutes = 4096
+ static let maxDataConnections = 256
+
+ /// A route is cached per lookup code and the request prefix that names
+ /// the partition and the consumer.
+ struct RouteKey: Hashable, Sendable {
+ let code: CommandCode
+ let prefix: ArraySlice<UInt8>
+ }
+
+ private struct Route: Sendable {
+ /// The fence of the session the route was read under.
+ let fence: UInt128Value
+ let endpoint: String
+ let consumerSession: Responses.SessionIdentity
+ let context: PartitionContext
+ }
+
+ private struct DataConnection {
+ let connection: any IggyConnection
+ var consumerSession: Responses.SessionIdentity?
+ var usable: Bool
+ }
+
+ /// One data connection per primary. Its lock keeps the lockstep stream to
+ /// one request at a time.
+ private final class Slot: Sendable {
+ let lock = AsyncMutex()
+ let connection = NIOLockedValueBox<DataConnection?>(nil)
+ }
+
+ private struct State {
+ /// The client id of the current session. A reset replaces it, so a
+ /// route or a connection of a replaced session is never published
+ /// or used.
+ var fence: UInt128Value
+ /// Nodes in the roster; zero until a roster read succeeded.
+ var rosterSize = 0
+ var routes: [RouteKey: Route] = [:]
+ var slots: [String: Slot] = [:]
+ var nextHeartbeat: ContinuousClock.Instant?
+ }
+
+ private let state: NIOLockedValueBox<State>
+
+ init(fence: UInt128Value) {
+ state = NIOLockedValueBox(State(fence: fence))
+ }
+
+ /// Records the roster size a leader check or a roster read reported.
+ func learnRosterSize(_ size: Int) {
+ state.withLockedValue { $0.rosterSize = size }
+ }
+
+ /// Drops every route and data connection of the session that ended and
+ /// moves to the fence of the one that replaced it. A route or a
+ /// connection is published only under the lock after a fence check, so
+ /// nothing of the replaced session survives.
+ func clearSession(fence: UInt128Value) {
+ state.withLockedValue { state in
+ state.fence = fence
+ state.routes.removeAll()
+ state.slots.removeAll()
+ state.nextHeartbeat = nil
+ }
+ }
+
+ /// Sends a poll or an offset write on its route. `context` is the
+ /// partition context the caller captured, else the route's is taken and
+ /// kept for every retry.
+ ///
+ /// The budget is checked between attempts and caps each exchange instead
+ /// of cancelling one in flight, which would take the coordinator
+ /// connection, and the session's group membership, with it.
+ func send(
+ _ code: CommandCode, payload: [UInt8], key: RouteKey, routePayload: [UInt8], context: PartitionContext?, coordinator: ClientCore
+ ) async throws -> [UInt8] {
+ let clock = ContinuousClock()
+ let now = clock.now
+ let deadline = now + Self.pollTimeout
+ let clustered = try await isClustered(coordinator, deadline: deadline)
+ if clustered, heartbeatDue(at: now, interval: coordinator.configuration.heartbeatInterval) {
+ // The requests leave the coordinator connection idle, so it is
+ // pinged here once per heartbeat interval too.
+ _ = try await control(.ping, [], coordinator: coordinator, deadline: deadline)
+ }
+ let request = Request(code: code, payload: payload, key: key, routePayload: routePayload, deadline: deadline)
+ var captured = context
+ var interval = Self.retryInterval
+ while clock.now < deadline {
+ do {
+ if clustered {
+ return try await sendOnPrimary(request, context: &captured, coordinator: coordinator)
+ }
+ return try await sendOnCoordinator(request, context: &captured, coordinator: coordinator)
+ } catch let error as IggyError where error.code == .transientNotAccepted {
+ removeRoute(key)
+ guard clock.now + interval < deadline else {
+ throw error
+ }
+ try await Task.sleep(for: interval)
+ interval = min(interval * 2, Self.maxRetryInterval)
+ }
+ }
+ removeRoute(key)
+ throw IggyError(.transientNotCommitted, context: "command \(code.rawValue) did not complete within \(Self.pollTimeout)")
+ }
+
+ private struct Request {
+ let code: CommandCode
+ let payload: [UInt8]
+ let key: RouteKey
+ let routePayload: [UInt8]
+ let deadline: ContinuousClock.Instant
+ }
+
+ /// Whether the roster has more than one node, reading it once when no
+ /// leader check reported it yet.
+ private func isClustered(_ coordinator: ClientCore, deadline: ContinuousClock.Instant) async throws -> Bool {
+ if state.withLockedValue({ $0.rosterSize }) == 0 {
+ let readDeadline = min(deadline, ContinuousClock.now + Self.rosterReadTimeout)
+ let body = try await control(.getClusterMetadata, [], coordinator: coordinator, deadline: readDeadline)
+ let metadata = try Responses.clusterMetadata(body[...])
+ guard !metadata.nodes.isEmpty else {
+ throw IggyError(.transientNotAccepted, context: "the cluster roster is empty")
+ }
+ learnRosterSize(metadata.nodes.count)
+ }
+ return state.withLockedValue { $0.rosterSize } > 1
+ }
+
+ private func heartbeatDue(at now: ContinuousClock.Instant, interval: Duration) -> Bool {
+ state.withLockedValue { state in
+ let due = state.nextHeartbeat.map { now >= $0 } ?? false
+ if state.nextHeartbeat == nil || due {
+ state.nextHeartbeat = now + interval
+ }
+ return due
+ }
+ }
+
+ /// One exchange on the coordinator connection. A connection that is gone
+ /// is brought back by a ping, which reconnects, and the exchange runs
+ /// once more.
+ private func control(_ code: CommandCode, _ payload: [UInt8], coordinator: ClientCore, deadline: ContinuousClock.Instant) async throws -> [UInt8] {
+ do {
+ return try await coordinator.exchangeOnce(code: code.rawValue, payload: payload, context: .empty, deadline: deadline)
+ } catch let error as IggyError where error.isReconnectable && ContinuousClock.now < deadline {
+ _ = try await coordinator.exchange(code: CommandCode.ping.rawValue, payload: [], deadline: deadline)
+ return try await coordinator.exchangeOnce(code: code.rawValue, payload: payload, context: .empty, deadline: deadline)
+ }
+ }
+
+ /// The single-node path: the request goes over the coordinator connection
+ /// with the context of its route.
+ private func sendOnCoordinator(_ request: Request, context: inout PartitionContext?, coordinator: ClientCore) async throws -> [UInt8] {
+ let route = try await route(request.key, routePayload: request.routePayload, coordinator: coordinator, deadline: request.deadline)
+ let captured = context ?? route.context
+ context = captured
+ do {
+ if request.code == .pollMessagesOnPrimary {
+ return try await coordinator.exchangeOnce(
+ code: request.code.rawValue, payload: request.payload, context: captured, deadline: request.deadline)
+ }
+ return try await coordinator.exchange(code: request.code.rawValue, payload: request.payload, context: captured, deadline: request.deadline)
+ } catch let error as IggyError {
+ removeRoute(request.key)
+ throw Self.gradedLoss(error, on: "the coordinator")
+ }
+ }
+
+ /// The cluster path: the request goes over a data connection to the
+ /// partition primary, bound to the coordinator's session first. Every
+ /// frame is stamped under a check that the route's session is still the
+ /// client's, so nothing is written for a session that was replaced.
+ private func sendOnPrimary(_ request: Request, context: inout PartitionContext?, coordinator: ClientCore) async throws -> [UInt8] {
+ let route = try await route(request.key, routePayload: request.routePayload, coordinator: coordinator, deadline: request.deadline)
+ let captured = context ?? route.context
+ context = captured
+ let slot = try slot(for: route)
+ try await slot.lock.lock()
+ defer { slot.lock.unlock() }
+
+ var data = slot.connection.withLockedValue { $0 }
+ if data?.usable == false {
+ data = nil
+ }
+ if data == nil {
+ let connection: any IggyConnection
+ do {
+ connection = try await coordinator.dial(route.endpoint)
+ } catch let error as IggyError where error.isReconnectable {
+ throw IggyError(.transientNotAccepted, context: "cannot reach the partition primary at \(route.endpoint): \(error)")
+ }
+ data = DataConnection(connection: connection, consumerSession: nil, usable: true)
+ }
+ guard var data else {
+ throw IggyError(.notConnected)
+ }
+ // A cancelled exchange must not leave the connection reusable while
+ // its reply can still arrive.
+ data.usable = false
+ slot.connection.withLockedValue { $0 = data }
+
+ if data.consumerSession.map({ !Self.covers($0, route.consumerSession) }) ?? true {
+ do {
+ try await bind(data.connection, to: route.consumerSession, coordinator: coordinator, deadline: request.deadline)
+ } catch let error as IggyError where error.isReconnectable {
+ throw IggyError(.transientNotAccepted, context: "binding the session on \(route.endpoint) failed: \(error)")
+ }
+ data.consumerSession = route.consumerSession
+ slot.connection.withLockedValue { $0 = data }
+ }
+
+ do {
+ let body = try await exchange(
+ on: data.connection, code: request.code.rawValue, payload: request.payload, context: captured, identity: route.consumerSession,
+ coordinator: coordinator, deadline: request.deadline)
+ data.usable = true
+ slot.connection.withLockedValue { $0 = data }
+ return body
+ } catch let error as IggyError {
+ data.usable = !error.isReconnectable
+ if error.code == .transientNotAccepted {
+ data.consumerSession = nil
+ }
+ slot.connection.withLockedValue { $0 = data }
+ removeRoute(request.key)
+ throw Self.gradedLoss(error, on: route.endpoint)
+ }
+ }
+
+ /// A request that lost its connection after it was written may have been
+ /// applied, so its outcome is unknown on either path. One that found no
+ /// connection or no session was not written and is routed again.
+ private static func gradedLoss(_ error: IggyError, on endpoint: String) -> IggyError {
+ switch error.code {
+ case .notConnected, .cannotEstablishConnection, .unauthenticated:
+ return IggyError(.transientNotAccepted, context: "\(endpoint) took no request: \(error)")
+ default:
+ guard error.isReconnectable else {
+ return error
+ }
+ return IggyError(.transientNotCommitted, context: "the connection to \(endpoint) failed: \(error)")
+ }
+ }
+
+ /// Whether a connection bound to `bound` serves `wanted` too.
+ private static func covers(_ bound: Responses.SessionIdentity, _ wanted: Responses.SessionIdentity) -> Bool {
+ bound.clientID == wanted.clientID && bound.session == wanted.session && bound.metadataWatermark >= wanted.metadataWatermark
+ }
+
+ private func bind(
+ _ connection: any IggyConnection, to identity: Responses.SessionIdentity, coordinator: ClientCore, deadline: ContinuousClock.Instant
+ ) async throws {
+ let payload = try Requests.bindSession(
+ clientID: identity.clientID, session: identity.session, metadataWatermark: identity.metadataWatermark,
+ bindSecret: await coordinator.bindSecret)
+ let body = try await exchange(
+ on: connection, code: CommandCode.bindSession.rawValue, payload: payload, context: .empty, identity: identity, coordinator: coordinator,
+ deadline: deadline)
+ let bound = try Responses.loginRegister(body[...])
+ if bound.session == 0 {
+ throw IggyError(.invalidSession, context: "the server bound session 0")
+ }
+ if identity.session != 0, bound.session != identity.session {
+ throw IggyError(.sessionMismatch, context: "asked to bind session \(identity.session), the server bound \(bound.session)")
+ }
+ }
+
+ /// One lockstep exchange on a data connection, stamped from the shared
+ /// session for `identity`, with no replay. `notPrimary` is graded as on
+ /// the coordinator.
+ private func exchange(
+ on connection: any IggyConnection, code: UInt32, payload: [UInt8], context: PartitionContext, identity: Responses.SessionIdentity,
+ coordinator: ClientCore, deadline: ContinuousClock.Instant
+ ) async throws -> [UInt8] {
+ let timeout = min(coordinator.configuration.requestTimeout, deadline - .now)
+ guard timeout > .zero else {
+ throw IggyError(.transientNotAccepted, context: "the budget ran out before command \(code) was sent")
+ }
+ let request = try await coordinator.stampRequest(code: code, payload: payload, context: context, for: identity)
+ let raw = try await connection.exchange(request.bytes, timeout: timeout)
+ let header = raw.header[...]
+ if VSRFrame.peekCommand(header) == .reply, VSRFrame.readReplyRequestID(header) != request.requestID {
+ await connection.close()
+ throw IggyError(.disconnected, context: "the data connection answered another request")
+ }
+ do {
+ return Array(try VSRFrame.decodeReply(header: header, body: raw.body[...]))
+ } catch let error as IggyError {
+ throw ClientCore.graded(error)
+ }
+ }
+
+ /// The cached route of `key` if the session still holds it, else a new
+ /// lookup on the coordinator.
+ private func route(_ key: RouteKey, routePayload: [UInt8], coordinator: ClientCore, deadline: ContinuousClock.Instant) async throws -> Route {
+ let watermark = await coordinator.metadataWatermark
+ let (fence, cached) = state.withLockedValue { ($0.fence, $0.routes[key]) }
+ if let cached, cached.fence == fence, cached.consumerSession.metadataWatermark >= watermark {
+ return cached
+ }
+ let body = try await control(key.code, routePayload, coordinator: coordinator, deadline: deadline)
+ var response = try Responses.pollRoute(body[...])
+ response.consumerSession.metadataWatermark = max(response.consumerSession.metadataWatermark, watermark)
+ guard response.primary.endpoints.tcp != 0 else {
+ coordinator.logger.error("The partition primary \(response.primary.name) does not expose TCP")
+ throw IggyError(.featureUnavailable, context: "the partition primary \(response.primary.name) does not expose TCP")
+ }
+ let route = Route(fence: fence, endpoint: response.primary.tcpAddress, consumerSession: response.consumerSession, context: response.context)
+ try await validate(route, coordinator: coordinator)
+ try state.withLockedValue { state in
+ try Self.checkFence(of: route, in: state)
+ if state.routes.count >= Self.maxCachedRoutes {
+ state.routes.removeAll()
+ }
+ state.routes[key] = route
+ }
+ return route
+ }
+
+ /// Refuses a route the session moved past: a reset replaced it, or a
+ /// metadata commit newer than the route was seen since.
+ private func validate(_ route: Route, coordinator: ClientCore) async throws {
+ let watermark = await coordinator.metadataWatermark
+ try state.withLockedValue { try Self.checkFence(of: route, in: $0) }
+ guard route.consumerSession.metadataWatermark >= watermark else {
+ throw IggyError(.transientNotAccepted, context: "the route predates metadata commit \(watermark)")
+ }
+ }
+
+ private static func checkFence(of route: Route, in state: State) throws {
+ guard route.fence == state.fence else {
+ throw IggyError(.transientNotAccepted, context: "the route belongs to a replaced session")
+ }
+ }
+
+ private func slot(for route: Route) throws -> Slot {
+ try state.withLockedValue { state in
+ try Self.checkFence(of: route, in: state)
+ let endpoint = route.endpoint
+ if let slot = state.slots[endpoint] {
+ return slot
+ }
+ guard state.slots.count < Self.maxDataConnections else {
+ throw IggyError(.invalidConfiguration, context: "the pool of \(Self.maxDataConnections) primary connections is full")
+ }
+ let slot = Slot()
+ state.slots[endpoint] = slot
+ return slot
+ }
+ }
+
+ private func removeRoute(_ key: RouteKey) {
+ _ = state.withLockedValue { $0.routes.removeValue(forKey: key) }
+ }
+}
diff --git a/foreign/swift/Sources/Iggy/Models/Identifier.swift b/foreign/swift/Sources/Iggy/Models/Identifier.swift
index 1d816bb57..513fb2ea6 100644
--- a/foreign/swift/Sources/Iggy/Models/Identifier.swift
+++ b/foreign/swift/Sources/Iggy/Models/Identifier.swift
@@ -115,4 +115,20 @@ extension Identifier {
}
}
+ /// Reads what ``encode(into:)`` writes, for a raw request the client
+ /// routes by the partition it names.
+ static func decode(from reader: inout ByteReader) throws -> Identifier {
+ switch try reader.readUInt8() {
+ case numericKindCode:
+ let length = try reader.readUInt8()
+ guard length == 4 else {
+ throw WireError.validation("numeric identifier is \(length) bytes, expected 4")
+ }
+ return Identifier(numeric: try reader.readUInt32())
+ case stringKindCode:
+ return try Identifier(named: try reader.readName())
+ case let kind:
+ throw WireError.validation("unknown identifier kind \(kind)")
+ }
+ }
}
diff --git a/foreign/swift/Sources/Iggy/Models/IggyMessage.swift b/foreign/swift/Sources/Iggy/Models/IggyMessage.swift
index 25a0b8fc2..746fc5515 100644
--- a/foreign/swift/Sources/Iggy/Models/IggyMessage.swift
+++ b/foreign/swift/Sources/Iggy/Models/IggyMessage.swift
@@ -196,6 +196,12 @@ public struct PolledMessages: Sendable, Hashable {
/// Partition id of an empty poll by a consumer-group member that currently
/// holds no partitions: `NO_ASSIGNED_PARTITION` in `core/common`.
public static let noAssignedPartition: UInt32 = UInt32.max - 1
+
+ /// Partition id of an empty consumer-group poll the coordinator fenced
+ /// because the member's assignment is stale:
+ /// `RESYNC_REQUIRED_PARTITION_SENTINEL` in `core/common`. The client
+ /// syncs the assignment and polls again, so callers never see it.
+ static let resyncRequiredPartition: UInt32 = UInt32.max
}
/// One confirmation of a committed batch.
diff --git a/foreign/swift/Sources/Iggy/Models/Messaging.swift b/foreign/swift/Sources/Iggy/Models/Messaging.swift
index a7f4e66d2..6a5cf8bd8 100644
--- a/foreign/swift/Sources/Iggy/Models/Messaging.swift
+++ b/foreign/swift/Sources/Iggy/Models/Messaging.swift
@@ -95,6 +95,7 @@ public enum Partitioning: Sendable, Hashable {
private static let balancedCode: UInt8 = 1
private static let partitionCode: UInt8 = 2
private static let messagesKeyCode: UInt8 = 3
+ private static let partitionIDSize: UInt8 = 4
/// A messages key outside 1...255 bytes cannot be framed. The client
/// checks this before it resolves the key to a partition, and `encode`
@@ -114,7 +115,7 @@ public enum Partitioning: Sendable, Hashable {
writer.write(UInt8(0))
case .partition(let id):
writer.write(Self.partitionCode)
- writer.write(UInt8(4))
+ writer.write(Self.partitionIDSize)
writer.write(id)
case .messagesKey(let key):
writer.write(Self.messagesKeyCode)
@@ -123,10 +124,26 @@ public enum Partitioning: Sendable, Hashable {
}
}
+ /// `[kind:1][length:1][value]`, held to the lengths the server accepts.
+ static func decode(from reader: inout ByteReader) throws -> Partitioning {
+ let kind = try reader.readUInt8()
+ let length = try reader.readUInt8()
+ switch kind {
+ case balancedCode where length == 0:
+ return .balanced
+ case partitionCode where length == partitionIDSize:
+ return .partition(try reader.readUInt32())
+ case messagesKeyCode where length > 0:
+ return .messagesKey(Array(try reader.readBytes(Int(length))))
+ default:
+ throw WireError.validation("partitioning kind \(kind) cannot carry \(length) bytes")
+ }
+ }
+
var encodedSize: Int {
switch self {
case .balanced: 2
- case .partition: 6
+ case .partition: 2 + Int(Self.partitionIDSize)
case .messagesKey(let key): 2 + key.count
}
}
@@ -144,20 +161,28 @@ public struct PollingStrategy: Sendable, Hashable, CustomStringConvertible {
public var kind: Kind
public var value: UInt64
+ /// The partition incarnation and owner a continuation offset was read
+ /// under. A poll that carries one is refused once the partition was
+ /// recreated, instead of reading the offset in another incarnation.
+ public var context: PartitionContext?
- public init(kind: Kind, value: UInt64) {
+ public init(kind: Kind, value: UInt64, context: PartitionContext? = nil) {
self.kind = kind
self.value = value
+ self.context = context
}
- /// Resume from a specific offset.
- public static func offset(_ offset: UInt64) -> PollingStrategy {
- PollingStrategy(kind: .offset, value: offset)
+ /// Resume from a specific offset. Pass the ``PolledMessages/context`` of
+ /// the poll the offset came from to have the poll refused once the
+ /// partition was recreated.
+ public static func offset(_ offset: UInt64, context: PartitionContext? = nil) -> PollingStrategy {
+ PollingStrategy(kind: .offset, value: offset, context: context)
}
- /// Resume from the first message at or after a timestamp.
- public static func timestamp(_ timestamp: IggyTimestamp) -> PollingStrategy {
- PollingStrategy(kind: .timestamp, value: timestamp.microseconds)
+ /// Resume from the first message at or after a timestamp, under
+ /// `context` as for ``offset(_:context:)``.
+ public static func timestamp(_ timestamp: IggyTimestamp, context: PartitionContext? = nil) -> PollingStrategy {
+ PollingStrategy(kind: .timestamp, value: timestamp.microseconds, context: context)
}
/// Start from the beginning of the partition.
@@ -184,6 +209,21 @@ public struct PollingStrategy: Sendable, Hashable, CustomStringConvertible {
}
}
+/// A checkpoint to commit: the inclusive offset of the last consumed
+/// message, with the partition incarnation and owner of the poll that
+/// delivered it, so a delayed commit cannot land on a recreated partition.
+public struct ConsumerPosition: Sendable, Hashable {
+ public var partitionID: UInt32
+ public var offset: UInt64
+ public var context: PartitionContext
+
+ public init(partitionID: UInt32, offset: UInt64, context: PartitionContext) {
+ self.partitionID = partitionID
+ self.offset = offset
+ self.context = context
+ }
+}
+
/// Acknowledgement policy for consumer-offset write commands.
public enum AckLevel: UInt8, Sendable, Hashable {
/// Local fast path for a single-replica partition. Replicated partitions
diff --git a/foreign/swift/Sources/Iggy/Protocol/CommandCode.swift b/foreign/swift/Sources/Iggy/Protocol/CommandCode.swift
index 99673ea0a..0bab7b733 100644
--- a/foreign/swift/Sources/Iggy/Protocol/CommandCode.swift
+++ b/foreign/swift/Sources/Iggy/Protocol/CommandCode.swift
@@ -136,11 +136,12 @@ public enum CommandCode: UInt32, Sendable, Hashable, CaseIterable {
}
/// Codes that drive the session handshake and therefore cannot be sent
- /// through the raw request path.
+ /// through the raw request path. `BindSession` would rebind the
+ /// connection outside the client's bookkeeping of its session.
static func isSessionControl(_ code: UInt32) -> Bool {
switch CommandCode(rawValue: code) {
case .loginUser, .logoutUser, .loginRegister, .loginWithPersonalAccessToken,
- .loginRegisterWithPersonalAccessToken:
+ .loginRegisterWithPersonalAccessToken, .bindSession:
true
default:
false
diff --git a/foreign/swift/Sources/Iggy/Protocol/Requests.swift b/foreign/swift/Sources/Iggy/Protocol/Requests.swift
index d75f409a7..51de1388e 100644
--- a/foreign/swift/Sources/Iggy/Protocol/Requests.swift
+++ b/foreign/swift/Sources/Iggy/Protocol/Requests.swift
@@ -159,6 +159,14 @@ enum Requests {
return writer.bytes
}
+ static func partitionRoute(streamID: Identifier, topicID: Identifier, partitionID: UInt32) throws -> [UInt8] {
+ var writer = ByteWriter()
+ try streamID.encode(into: &writer)
+ try topicID.encode(into: &writer)
+ writer.write(partitionID)
+ return writer.bytes
+ }
+
// MARK: Users
/// Byte length of a bind secret carried by login and bind-session requests.
diff --git a/foreign/swift/Sources/Iggy/Protocol/Responses.swift b/foreign/swift/Sources/Iggy/Protocol/Responses.swift
index b6d3a7267..c386254c6 100644
--- a/foreign/swift/Sources/Iggy/Protocol/Responses.swift
+++ b/foreign/swift/Sources/Iggy/Protocol/Responses.swift
@@ -385,25 +385,29 @@ enum Responses {
let count = try reader.readUInt32()
var nodes: [ClusterNode] = []
for _ in 0..<count {
- let replicaID = try reader.readUInt8()
- let nodeName = try reader.readLongString()
- let ip = try reader.readLongString()
- let endpoints = TransportEndpoints(
- tcp: try reader.readUInt16(), quic: try reader.readUInt16(), http: try reader.readUInt16(), websocket: try reader.readUInt16())
- let roleCode = try reader.readUInt8()
- let statusCode = try reader.readUInt8()
- guard let role = ClusterNodeRole(rawValue: roleCode) else {
- throw IggyError(.invalidCommand, context: "unknown cluster node role \(roleCode)")
- }
- guard let status = ClusterNodeStatus(rawValue: statusCode) else {
- throw IggyError(.invalidCommand, context: "unknown cluster node status \(statusCode)")
- }
- nodes.append(ClusterNode(replicaID: replicaID, name: nodeName, ip: ip, endpoints: endpoints, role: role, status: status))
+ nodes.append(try clusterNode(from: &reader))
}
return ClusterMetadata(name: name, nodes: nodes)
}
}
+ static func clusterNode(from reader: inout ByteReader) throws -> ClusterNode {
+ let replicaID = try reader.readUInt8()
+ let name = try reader.readLongString()
+ let ip = try reader.readLongString()
+ let endpoints = TransportEndpoints(
+ tcp: try reader.readUInt16(), quic: try reader.readUInt16(), http: try reader.readUInt16(), websocket: try reader.readUInt16())
+ let roleCode = try reader.readUInt8()
+ let statusCode = try reader.readUInt8()
+ guard let role = ClusterNodeRole(rawValue: roleCode) else {
+ throw IggyError(.invalidCommand, context: "unknown cluster node role \(roleCode)")
+ }
+ guard let status = ClusterNodeStatus(rawValue: statusCode) else {
+ throw IggyError(.invalidCommand, context: "unknown cluster node status \(statusCode)")
+ }
+ return ClusterNode(replicaID: replicaID, name: name, ip: ip, endpoints: endpoints, role: role, status: status)
+ }
+
static func optionSpecs(_ body: ArraySlice<UInt8>) throws -> [OptionSpec] {
try decode(body) { reader in
let count = try reader.readUInt32()
@@ -422,6 +426,49 @@ enum Responses {
}
}
+ // MARK: Routing
+
+ static let partitionRouteSize = 25
+
+ /// The context a request routed by a `GetPartitionRoute` reply carries:
+ /// `[incarnation:8][metadata_op:8][view:4][partitions_count:4][primary_replica:1]`.
+ /// A route names no consumer, so the context carries owner 0, as
+ /// `PartitionRouteResponse::context` in the Rust protocol crate. The view
+ /// and the primary pick a replica for a client that dials each one; this
+ /// one sends over its connection. A reply of any other length comes from
+ /// a mismatched build.
+ static func partitionRoute(_ body: ArraySlice<UInt8>) throws -> PartitionContext {
+ guard body.count == partitionRouteSize else {
+ throw IggyError(.invalidCommand, context: "partition route is \(body.count) bytes, expected \(partitionRouteSize)")
+ }
+ return try decode(body) { reader in
+ PartitionContext(incarnation: try reader.readUInt64(), ownerGeneration: 0, metadataOp: try reader.readUInt64())
+ }
+ }
+
+ /// The session a data connection binds to, as `SessionIdentity` in the
+ /// Rust protocol crate.
+ struct SessionIdentity: Equatable {
+ let clientID: UInt128Value
+ let session: UInt64
+ var metadataWatermark: UInt64
+ }
+
+ /// Where a poll or an offset write goes: the reply to `GetPollRouting`
+ /// and `GetConsumerOffsetRouting`.
+ struct PollRoute: Equatable {
+ var consumerSession: SessionIdentity
+ let context: PartitionContext
+ let primary: ClusterNode
+ }
+
+ static func pollRoute(_ body: ArraySlice<UInt8>) throws -> PollRoute {
+ try decode(body) { reader in
+ let session = SessionIdentity(clientID: try reader.readUInt128(), session: try reader.readUInt64(), metadataWatermark: try reader.readUInt64())
+ return PollRoute(consumerSession: session, context: try PartitionContext.decode(from: &reader), primary: try clusterNode(from: &reader))
+ }
+ }
+
// MARK: Messages
/// Confirmations of a committed batch. An empty body means the batch was
diff --git a/foreign/swift/Sources/Iggy/Protocol/VSRFrame.swift b/foreign/swift/Sources/Iggy/Protocol/VSRFrame.swift
index 14cb94e72..3f50b53fb 100644
--- a/foreign/swift/Sources/Iggy/Protocol/VSRFrame.swift
+++ b/foreign/swift/Sources/Iggy/Protocol/VSRFrame.swift
@@ -47,6 +47,7 @@ enum VSRFrame {
enum ReplyOffset {
static let size = 48
static let command = 60
+ static let commit = 184
static let request = 200
static let operation = 208
static let status = 216
@@ -91,6 +92,10 @@ enum VSRFrame {
let bytes: [UInt8]
let requestID: UInt64
let operation: ConsensusOperation
+ /// The session the frame was stamped under, which decides whether it
+ /// can be replayed as it is after a reconnect.
+ let clientID: UInt128Value
+ let session: UInt64
}
/// Builds the complete request frame for a command code and its encoded
@@ -151,7 +156,7 @@ enum VSRFrame {
buffer.storeBytes(of: partitionContext.metadataOp.littleEndian, toByteOffset: RequestOffset.minimumMetadataOp, as: UInt64.self)
}
frame.append(contentsOf: payload)
- return EncodedRequest(bytes: frame, requestID: requestID, operation: operation)
+ return EncodedRequest(bytes: frame, requestID: requestID, operation: operation, clientID: session.clientID, session: sessionID)
}
/// Reads the total frame length, header included, from any consensus
@@ -181,6 +186,14 @@ enum VSRFrame {
return UInt64(littleEndianBytes: header[start..<start + 8])
}
+ static func readReplyCommit(_ header: ArraySlice<UInt8>) -> UInt64? {
+ guard header.count >= ReplyOffset.commit + 8 else {
+ return nil
+ }
+ let start = header.startIndex + ReplyOffset.commit
+ return UInt64(littleEndianBytes: header[start..<start + 8])
+ }
+
static func readReplyStatus(_ header: ArraySlice<UInt8>) -> UInt32? {
guard header.count >= ReplyOffset.status + 4 else {
return nil
diff --git a/foreign/swift/Sources/Iggy/Transport/TCPConnection.swift b/foreign/swift/Sources/Iggy/Transport/TCPConnection.swift
index 12cd117fe..64601cd26 100644
--- a/foreign/swift/Sources/Iggy/Transport/TCPConnection.swift
+++ b/foreign/swift/Sources/Iggy/Transport/TCPConnection.swift
@@ -192,7 +192,10 @@ final class ExchangeHandler: ChannelInboundHandler, @unchecked Sendable {
/// 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"))
+ // Nothing of this request was written, so it is safe to send
+ // again on another connection.
+ let cause = closeError.map { "\($0)" } ?? "connection closed"
+ continuation.resume(throwing: IggyError(.notConnected, context: "the connection closed before the request was written: \(cause)"))
return
}
if let previous = pending {
diff --git a/foreign/swift/Tests/IggyE2ETests/ClientE2ETests.swift b/foreign/swift/Tests/IggyE2ETests/ClientE2ETests.swift
new file mode 100644
index 000000000..9acc134c8
--- /dev/null
+++ b/foreign/swift/Tests/IggyE2ETests/ClientE2ETests.swift
@@ -0,0 +1,693 @@
+// 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 Iggy
+import Testing
+
+@Suite("Sessions", requiresServer)
+struct SessionE2ETests {
+ @Test func onlyPingWorksBeforeSigningIn() async throws {
+ try await E2EEnvironment.withSession(login: false) { session in
+ let client = session.client
+ #expect(await client.state == .connected)
+ try await client.ping()
+ await #expect(throws: IggyError(.unauthenticated)) {
+ try await client.getStreams()
+ }
+ // The roster is auth-gated so an unauthenticated reader cannot map
+ // the private network.
+ await #expect(throws: IggyError(.unauthenticated)) {
+ try await client.getClusterMetadata()
+ }
+ }
+ }
+
+ @Test func loginLogoutAndRelogin() async throws {
+ try await E2EEnvironment.withSession() { session in
+ let client = session.client
+ #expect(await client.state == .authenticated)
+ let me = try await client.getMe()
+ #expect(me.userID == 0)
+ #expect(me.transport == "TCP")
+ _ = try await client.getStreams()
+
+ try await client.logout()
+ #expect(await client.state == .connected)
+ await #expect(throws: IggyError(.unauthenticated)) {
+ try await client.getStreams()
+ }
+
+ let identity = try await client.login(username: E2EEnvironment.rootUsername, password: E2EEnvironment.rootPassword)
+ #expect(identity.userID == 0)
+ _ = try await client.getStreams()
+ }
+ }
+
+ @Test func autoLoginSignsInOnConnect() async throws {
+ try await E2EEnvironment.withSession(autoLogin: .usernamePassword(username: E2EEnvironment.rootUsername, password: E2EEnvironment.rootPassword)) {
+ session in
+ let client = session.client
+ #expect(await client.state == .authenticated)
+ _ = try await client.getStreams()
+ }
+ }
+
+ @Test func invalidCredentialsAreRejected() async throws {
+ try await E2EEnvironment.withSession(login: false) { session in
+ let client = session.client
+ await #expect(throws: IggyError.self) {
+ try await client.login(username: "nobody", password: "wrong-password")
+ }
+ #expect(await client.state == .connected)
+ try await client.ping()
+ }
+ }
+
+ @Test func personalAccessTokenRoundTrip() async throws {
+ try await E2EEnvironment.withSession() { session in
+ let client = session.client
+ let name = E2EEnvironment.uniqueName("swift-pat")
+ let token = try await client.createPersonalAccessToken(name: name, expiry: .after(.seconds(3600)))
+ #expect(!token.token.isEmpty)
+ let listed = try await client.getPersonalAccessTokens()
+ let created = try #require(listed.first { $0.name == name })
+ #expect(created.expiresAt != nil)
+
+ let other = try await session.secondaryClient()
+ let identity = try await other.login(personalAccessToken: token.token)
+ #expect(identity.userID == 0)
+ _ = try await other.getStreams()
+
+ try await client.deletePersonalAccessToken(name: name)
+ #expect(try await client.getPersonalAccessTokens().contains { $0.name == name } == false)
+ }
+ }
+
+ @Test func eventsReportTheLifecycle() async throws {
+ let client = IggyClient(
+ configuration: E2EEnvironment.configuration(
+ autoLogin: .usernamePassword(username: E2EEnvironment.rootUsername, password: E2EEnvironment.rootPassword)))
+ let events = await client.events
+ try await client.connect()
+ try await client.logout()
+ await client.disconnect()
+ await client.shutdown()
+ var received: [DiagnosticEvent] = []
+ for await event in events {
+ received.append(event)
+ }
+ #expect(received == [.connected, .signedIn, .signedOut, .disconnected, .shutdown])
+ }
+
+ @Test func shutdownIsTerminal() async throws {
+ let client = try await E2EEnvironment.connect()
+ await client.shutdown()
+ #expect(await client.state == .shutdown)
+ await #expect(throws: IggyError(.clientShutdown)) {
+ try await client.connect()
+ }
+ await #expect(throws: IggyError(.clientShutdown)) {
+ try await client.ping()
+ }
+ }
+
+ @Test func disconnectKeepsTheClientUsable() async throws {
+ try await E2EEnvironment.withSession() { session in
+ let client = session.client
+ await client.disconnect()
+ #expect(await client.state == .disconnected)
+ // No credentials survive an explicit disconnect, so nothing replays.
+ await #expect(throws: IggyError(.notConnected)) {
+ try await client.ping()
+ }
+ try await client.connect()
+ try await client.ping()
+ _ = try await client.login(username: E2EEnvironment.rootUsername, password: E2EEnvironment.rootPassword)
+ _ = try await client.getStreams()
+ }
+ }
+}
+
+@Suite("Streams, topics, and partitions", requiresServer)
+struct ResourceE2ETests {
+ @Test func streamCRUD() async throws {
+ try await E2EEnvironment.withSession() { session in
+ let client = session.client
+ let name = E2EEnvironment.uniqueName("swift-stream")
+ let created = try await client.createStream(name: name)
+ #expect(created.name == name)
+ #expect(created.topicsCount == 0)
+ let id = Identifier(numeric: created.id)
+
+ let byID = try #require(try await client.getStream(id))
+ #expect(byID.name == name)
+ let byName = try #require(try await client.getStream(Identifier(named: name)))
+ #expect(byName.id == created.id)
+ #expect(try await client.getStreams().contains { $0.id == created.id })
+
+ let renamed = E2EEnvironment.uniqueName("swift-renamed")
+ try await client.updateStream(id, name: renamed)
+ #expect(try await client.getStream(id)?.name == renamed)
+
+ try await client.deleteStream(id)
+ // The server hands out the lowest free id, so a parallel test can
+ // already own it again: assert on the name, which is unique.
+ #expect(try await client.getStream(Identifier(named: renamed)) == nil)
+ await #expect(throws: IggyError.self) {
+ try await client.deleteStream(Identifier(named: renamed))
+ }
+ }
+ }
+
+ @Test func duplicateStreamNamesAreRejected() async throws {
+ try await E2EEnvironment.withSession() { session in
+ let client = session.client
+ let name = E2EEnvironment.uniqueName("swift-dup")
+ _ = try await client.createStream(name: name)
+ await session.track(stream: try Identifier(named: name))
+ do {
+ _ = try await client.createStream(name: name)
+ Issue.record("expected the duplicate to be rejected")
+ } catch let error as IggyError {
+ #expect(error.code == .streamNameAlreadyExists)
+ }
+ }
+ }
+
+ @Test func topicCRUDWithOptionsAndCatalog() async throws {
+ try await E2EEnvironment.withSession() { session in
+ let client = session.client
+ let name = E2EEnvironment.uniqueName("swift-topic")
+ let stream = try await client.createStream(name: name)
+ let streamID = Identifier(numeric: stream.id)
+ await session.track(stream: try Identifier(named: name))
+
+ let specs = try await client.describeOptions(scope: .topic)
+ let keys = Set(specs.map(\.key))
+ #expect(keys.contains(TopicOptionKey.durability))
+ #expect(keys.contains(TopicOptionKey.segmentSize))
+ #expect(try #require(specs.first { $0.key == TopicOptionKey.segmentSize }).kind == .uint64)
+ #expect(try await client.describeOptions(scope: .stream).isEmpty)
+
+ let options = TopicCreateOptions(
+ partitionsCount: 3, compressionAlgorithm: .gzip, messageExpiry: .after(.seconds(3600)),
+ maxTopicSize: .bytes(10_737_418_240), durability: .persisted)
+ let created = try await client.createTopic(streamID: streamID, name: name, options: options)
+ #expect(created.name == name)
+ #expect(created.partitionsCount == 3)
+ #expect(created.partitions.map(\.id) == [0, 1, 2])
+ #expect(created.compressionAlgorithm == .gzip)
+ #expect(created.messageExpiry == .after(.seconds(3600)))
+ #expect(created.maxTopicSize == .bytes(10_737_418_240))
+ let topicID = Identifier(numeric: created.id)
+
+ let fetched = try #require(try await client.getTopic(streamID: streamID, topicID: topicID))
+ #expect(fetched.options[TopicOptionKey.durability] == .explicit("persisted"))
+ // Keys the client left alone are resolved by admission and reported
+ // as derived.
+ #expect(fetched.options[TopicOptionKey.segmentSize]?.explicit == false)
+ #expect(try await client.getTopics(streamID: streamID).map(\.id) == [created.id])
+ #expect(try await client.getStream(streamID)?.topics.first?.name == name)
+
+ let renamed = E2EEnvironment.uniqueName("swift-topic-renamed")
+ try await client.updateTopic(
+ streamID: streamID, topicID: topicID, name: renamed,
+ options: TopicUpdateOptions(compressionAlgorithm: CompressionAlgorithm.none, messageExpiry: .never))
+ let updated = try #require(try await client.getTopic(streamID: streamID, topicID: topicID))
+ #expect(updated.name == renamed)
+ #expect(updated.compressionAlgorithm == CompressionAlgorithm.none)
+ #expect(updated.messageExpiry == .never)
+
+ try await client.createPartitions(streamID: streamID, topicID: topicID, count: 2)
+ #expect(try await client.getTopic(streamID: streamID, topicID: topicID)?.partitionsCount == 5)
+ try await client.deletePartitions(streamID: streamID, topicID: topicID, count: 4)
+ #expect(try await client.getTopic(streamID: streamID, topicID: topicID)?.partitionsCount == 1)
+
+ try await client.deleteTopic(streamID: streamID, topicID: topicID)
+ #expect(try await client.getTopic(streamID: streamID, topicID: topicID) == nil)
+ }
+ }
+
+ @Test func unknownOptionKeysAreRefused() async throws {
+ try await E2EEnvironment.withSession() { session in
+ let client = session.client
+ let name = E2EEnvironment.uniqueName("swift-badopt")
+ let stream = try await client.createStream(name: name)
+ let streamID = Identifier(numeric: stream.id)
+ await session.track(stream: try Identifier(named: name))
+ await #expect(throws: IggyError.self) {
+ try await client.createTopic(streamID: streamID, name: name, options: TopicCreateOptions(raw: ["not_a_real_option": "1"]))
+ }
+ // A string value for a typed key is parsed by admission and reported
+ // with its canonical kind.
+ let created = try await client.createTopic(
+ streamID: streamID, name: name, options: TopicCreateOptions(raw: [TopicOptionKey.durability: "persisted"]))
+ let topic = try #require(try await client.getTopic(streamID: streamID, topicID: Identifier(numeric: created.id)))
+ #expect(topic.options[TopicOptionKey.durability] == .explicit("persisted"))
+ }
+ }
+
+ @Test func consumerGroupCRUD() async throws {
+ try await E2EEnvironment.withSession() { session in
+ let client = session.client
+ let scratch = try await session.scratchTopic(partitions: 2)
+ let name = E2EEnvironment.uniqueName("swift-group")
+ let group = try await client.createConsumerGroup(streamID: scratch.stream, topicID: scratch.topic, name: name)
+ #expect(group.name == name)
+ #expect(group.partitionsCount == 2)
+ #expect(group.members.isEmpty)
+ let groupID = Identifier(numeric: group.id)
+ #expect(try await client.getConsumerGroups(streamID: scratch.stream, topicID: scratch.topic).map(\.id) == [group.id])
+ #expect(try await client.getConsumerGroup(streamID: scratch.stream, topicID: scratch.topic, groupID: Identifier(named: name))?.id == group.id)
+
+ try await client.joinConsumerGroup(streamID: scratch.stream, topicID: scratch.topic, groupID: groupID)
+ // The server installs the member's partitions after the join.
+ let owned = try await E2EEnvironment.eventually {
+ try await client.getConsumerGroup(streamID: scratch.stream, topicID: scratch.topic, groupID: groupID)?.members.first?.partitions.sorted()
+ == [0, 1]
+ }
+ #expect(owned)
+ let joined = try #require(try await client.getConsumerGroup(streamID: scratch.stream, topicID: scratch.topic, groupID: groupID))
+ #expect(joined.membersCount == 1)
+ let me = try await client.getMe()
+ #expect(me.consumerGroups.contains { $0.groupID == group.id })
+
+ try await client.leaveConsumerGroup(streamID: scratch.stream, topicID: scratch.topic, groupID: groupID)
+ #expect(try await client.getConsumerGroup(streamID: scratch.stream, topicID: scratch.topic, groupID: groupID)?.membersCount == 0)
+ try await client.deleteConsumerGroup(streamID: scratch.stream, topicID: scratch.topic, groupID: groupID)
+ #expect(try await client.getConsumerGroup(streamID: scratch.stream, topicID: scratch.topic, groupID: groupID) == nil)
+ }
+ }
+}
+
+@Suite("Users", requiresServer)
+struct UserE2ETests {
+ @Test func userLifecycle() async throws {
+ try await E2EEnvironment.withSession() { session in
+ let client = session.client
+ let username = String(E2EEnvironment.uniqueName("swift-user").prefix(50))
+ let permissions = Permissions(
+ global: GlobalPermissions(readServers: true, readStreams: true, readTopics: true, pollMessages: true, sendMessages: true))
+ let created = try await client.createUser(username: username, password: "secret-1", status: .active, permissions: permissions)
+ #expect(created.username == username)
+ #expect(created.status == .active)
+ #expect(created.permissions?.global.readStreams == true)
+ #expect(created.permissions?.global.manageStreams == false)
+ let userID = Identifier(numeric: created.id)
+ #expect(try await client.getUsers().contains { $0.id == created.id })
+ #expect(try await client.getUser(Identifier(named: username))?.id == created.id)
+
+ // The new user can sign in and is bound by its permissions.
+ let other = try await session.secondaryClient()
+ let identity = try await other.login(username: username, password: "secret-1")
+ #expect(identity.userID == created.id)
+ _ = try await other.getStreams()
+ await #expect(throws: IggyError(.unauthorized)) {
+ try await other.createStream(name: E2EEnvironment.uniqueName("swift-forbidden"))
+ }
+
+ try await client.changePassword(userID, currentPassword: "secret-1", newPassword: "secret-2")
+ try await client.updateUser(userID, username: username + "-renamed", status: .inactive)
+ let updated = try #require(try await client.getUser(userID))
+ #expect(updated.username == username + "-renamed")
+ #expect(updated.status == .inactive)
+ try await client.updatePermissions(userID, permissions: nil)
+ #expect(try await client.getUser(userID)?.permissions == nil)
+ try await client.deleteUser(userID)
+ #expect(try await client.getUser(userID) == nil)
+ }
+ }
+
+ @Test func invalidUsernamesFailBeforeTheWire() async throws {
+ try await E2EEnvironment.withSession() { session in
+ let client = session.client
+ await #expect(throws: IggyError(.invalidUsername)) {
+ try await client.createUser(username: "ab", password: "secret")
+ }
+ await #expect(throws: IggyError(.invalidPassword)) {
+ try await client.createUser(username: "valid-name", password: "ab")
+ }
+ }
+ }
+}
+
+@Suite("Messaging", requiresServer)
+struct MessagingE2ETests {
+ @Test func sendAndPollRoundTrip() async throws {
+ try await E2EEnvironment.withSession() { session in
+ let client = session.client
+ let scratch = try await session.scratchTopic(partitions: 2)
+
+ var baseOffsets: [UInt64] = []
+ for _ in 0..<3 {
+ let response = try await client.sendBatch(
+ streamID: scratch.stream, topicID: scratch.topic, partitioning: .partition(0), messages: try E2EEnvironment.messages(4))
+ let confirmation = try #require(response.confirmations.first)
+ #expect(confirmation.streamID == scratch.streamID)
+ #expect(confirmation.topicID == scratch.topicID)
+ #expect(confirmation.partitionID == 0)
+ baseOffsets.append(confirmation.baseOffset)
+ }
+ #expect(baseOffsets == [0, 4, 8])
+
+ let polled = try await client.pollMessages(
+ streamID: scratch.stream, topicID: scratch.topic, partitionID: 0, consumer: .default, strategy: .offset(0), count: 12, autoCommit: false)
+ #expect(polled.partitionID == 0)
+ #expect(polled.currentOffset == 11)
+ #expect(polled.messages.count == 12)
+ #expect(polled.messages.map(\.offset) == Array(0..<12))
+ #expect(polled.messages[0].payloadString == "message 0")
+ #expect(polled.messages[11].payloadString == "message 3")
+ #expect(polled.messages.allSatisfy { !$0.id.isZero })
+ #expect(polled.messages.allSatisfy { $0.timestamp.microseconds > 0 && $0.originTimestamp.microseconds > 0 })
+ #expect(polled.messages.allSatisfy { $0.checksum != 0 })
+ let ids = Set(polled.messages.map(\.id))
+ #expect(ids.count == 12)
+
+ let last = try await client.pollMessages(
+ streamID: scratch.stream, topicID: scratch.topic, partitionID: 0, consumer: .default, strategy: .last, count: 1, autoCommit: false)
+ #expect(last.messages.map(\.offset) == [11])
+ let tail = try await client.pollMessages(
+ streamID: scratch.stream, topicID: scratch.topic, partitionID: 0, consumer: .default, strategy: .offset(10), count: 100, autoCommit: false)
+ #expect(tail.messages.map(\.offset) == [10, 11])
+ let empty = try await client.pollMessages(
+ streamID: scratch.stream, topicID: scratch.topic, partitionID: 1, consumer: .default, strategy: .first, count: 10, autoCommit: false)
+ #expect(empty.messages.isEmpty)
+ }
+ }
+
+ @Test func userHeadersAndIdsRoundTrip() async throws {
+ try await E2EEnvironment.withSession() { session in
+ let client = session.client
+ let scratch = try await session.scratchTopic(partitions: 1)
+
+ let id = MessageID(uuid: UUID())
+ let headers: UserHeaders = ["trace-id": "abc-123", "attempt": .uint32(3), "urgent": true, "ratio": 0.5, "raw": try .raw([1, 2, 3])]
+ let message = try IggyMessage("with headers", id: id, userHeaders: headers)
+ let binary = try IggyMessage(payload: [0x00, 0xFF, 0x10, 0x80])
+ _ = try await client.sendBatch(streamID: scratch.stream, topicID: scratch.topic, partitioning: .partition(0), messages: [message, binary])
+
+ let polled = try await client.pollMessages(
+ streamID: scratch.stream, topicID: scratch.topic, partitionID: 0, consumer: .default, strategy: .first, count: 10, autoCommit: false)
+ #expect(polled.messages.count == 2)
+ let first = polled.messages[0]
+ #expect(first.id == id)
+ #expect(first.payloadString == "with headers")
+ #expect(first.userHeaders == headers)
+ #expect(first[header: "trace-id"]?.stringValue == "abc-123")
+ #expect(first[header: "attempt"]?.uint32Value == 3)
+ #expect(polled.messages[1].payload == [0x00, 0xFF, 0x10, 0x80])
+ #expect(polled.messages[1].userHeaders == nil)
+ }
+ }
+
+ @Test func balancedAndKeyedPartitioning() async throws {
+ try await E2EEnvironment.withSession() { session in
+ let client = session.client
+ let scratch = try await session.scratchTopic(partitions: 3)
+
+ var balancedPartitions: [UInt32] = []
+ for _ in 0..<3 {
+ let response = try await client.sendBatch(
+ streamID: scratch.stream, topicID: scratch.topic, partitioning: .balanced, messages: try E2EEnvironment.messages(1))
+ balancedPartitions.append(try #require(response.confirmations.first).partitionID)
+ }
+ #expect(Set(balancedPartitions) == [0, 1, 2])
+
+ let first = try await client.sendBatch(
+ streamID: scratch.stream, topicID: scratch.topic, partitioning: .messagesKey("order-key-1"), messages: try E2EEnvironment.messages(1))
+ let second = try await client.sendBatch(
+ streamID: scratch.stream, topicID: scratch.topic, partitioning: .messagesKey("order-key-1"), messages: try E2EEnvironment.messages(1))
+ #expect(try #require(first.confirmations.first).partitionID == (try #require(second.confirmations.first).partitionID))
+ }
+ }
+
+ @Test func partitionCountRefreshesAfterPartitionChanges() async throws {
+ try await E2EEnvironment.withSession() { session in
+ let client = session.client
+ let scratch = try await session.scratchTopic(partitions: 1)
+
+ let first = try await client.sendBatch(
+ streamID: scratch.stream, topicID: scratch.topic, partitioning: .balanced, messages: try E2EEnvironment.messages(1))
+ #expect(try #require(first.confirmations.first).partitionID == 0)
+ try await client.createPartitions(streamID: scratch.stream, topicID: scratch.topic, count: 2)
+ let second = try await client.sendBatch(
+ streamID: scratch.stream, topicID: scratch.topic, partitioning: .balanced, messages: try E2EEnvironment.messages(1))
+ #expect(try #require(second.confirmations.first).partitionID == 1)
+ try await client.deletePartitions(streamID: scratch.stream, topicID: scratch.topic, count: 2)
+ let third = try await client.sendBatch(
+ streamID: scratch.stream, topicID: scratch.topic, partitioning: .balanced, messages: try E2EEnvironment.messages(1))
+ #expect(try #require(third.confirmations.first).partitionID == 0)
+ }
+ }
+
+ @Test func consumerOffsets() async throws {
+ try await E2EEnvironment.withSession() { session in
+ let client = session.client
+ let scratch = try await session.scratchTopic(partitions: 1)
+ _ = try await client.sendBatch(
+ streamID: scratch.stream, topicID: scratch.topic, partitioning: .partition(0), messages: try E2EEnvironment.messages(5))
+
+ let consumer = Consumer.consumer("swift-offsets")
+ #expect(try await client.getConsumerOffset(consumer: consumer, streamID: scratch.stream, topicID: scratch.topic, partitionID: 0) == nil)
+ try await client.storeConsumerOffset(consumer: consumer, streamID: scratch.stream, topicID: scratch.topic, partitionID: 0, offset: 3)
+ let stored = try #require(try await client.getConsumerOffset(consumer: consumer, streamID: scratch.stream, topicID: scratch.topic, partitionID: 0))
+ #expect(stored.storedOffset == 3)
+ #expect(stored.currentOffset == 4)
+ #expect(stored.partitionID == 0)
+
+ let next = try await client.pollMessages(
+ streamID: scratch.stream, topicID: scratch.topic, partitionID: 0, consumer: consumer, strategy: .next, count: 10, autoCommit: true)
+ #expect(next.messages.map(\.offset) == [4])
+ #expect(try await client.getConsumerOffset(consumer: consumer, streamID: scratch.stream, topicID: scratch.topic, partitionID: 0)?.storedOffset == 4)
+
+ try await client.deleteConsumerOffset(consumer: consumer, streamID: scratch.stream, topicID: scratch.topic, partitionID: 0)
+ #expect(try await client.getConsumerOffset(consumer: consumer, streamID: scratch.stream, topicID: scratch.topic, partitionID: 0) == nil)
+ await #expect(throws: IggyError(.consumerOffsetNotFound)) {
+ try await client.deleteConsumerOffset(consumer: consumer, streamID: scratch.stream, topicID: scratch.topic, partitionID: 0)
+ }
+ }
+ }
+
+ @Test func checkpointsKeepThePartitionContextOfTheirPoll() async throws {
+ try await E2EEnvironment.withSession() { session in
+ let client = session.client
+ let scratch = try await session.scratchTopic(partitions: 2)
+ let consumer = Consumer.consumer("swift-positions")
+ _ = try await client.sendBatch(
+ streamID: scratch.stream, topicID: scratch.topic, partitioning: .partition(1), messages: try E2EEnvironment.messages(3))
+ let polled = try await client.pollMessages(
+ streamID: scratch.stream, topicID: scratch.topic, partitionID: 1, consumer: consumer, strategy: .first, count: 3, autoCommit: false)
+ #expect(polled.messages.map(\.offset) == [0, 1, 2])
+ let position = ConsumerPosition(partitionID: 1, offset: 2, context: polled.context)
+ try await client.storeConsumerPosition(consumer: consumer, streamID: scratch.stream, topicID: scratch.topic, position: position)
+ #expect(try await client.getConsumerOffset(consumer: consumer, streamID: scratch.stream, topicID: scratch.topic, partitionID: 1)?.storedOffset == 2)
+ let continued = PollingStrategy.offset(1, context: polled.context)
+ let more = try await client.pollMessages(
+ streamID: scratch.stream, topicID: scratch.topic, partitionID: 1, consumer: consumer, strategy: continued, count: 10, autoCommit: false)
+ #expect(more.messages.map(\.offset) == [1, 2])
+
+ // Another client recreates the partition, so its id now names a
+ // new incarnation this client has not seen.
+ let other = try await session.secondaryClient()
+ _ = try await other.login(username: E2EEnvironment.rootUsername, password: E2EEnvironment.rootPassword)
+ try await other.deletePartitions(streamID: scratch.stream, topicID: scratch.topic, count: 1)
+ try await other.createPartitions(streamID: scratch.stream, topicID: scratch.topic, count: 1)
+
+ // The send under the stale cached context reads the new route.
+ _ = try await client.sendBatch(
+ streamID: scratch.stream, topicID: scratch.topic, partitioning: .partition(1), messages: try E2EEnvironment.messages(1))
+ // Neither a checkpoint nor a continuation of the old incarnation
+ // lands on the new one.
+ await expectIggyError(.historyUnavailable) {
+ try await client.storeConsumerPosition(consumer: consumer, streamID: scratch.stream, topicID: scratch.topic, position: position)
+ }
+ await expectIggyError(.historyUnavailable) {
+ _ = try await client.pollMessages(
+ streamID: scratch.stream, topicID: scratch.topic, partitionID: 1, consumer: consumer, strategy: continued, count: 10, autoCommit: false)
+ }
+ let fresh = try await client.pollMessages(
+ streamID: scratch.stream, topicID: scratch.topic, partitionID: 1, consumer: consumer, strategy: .first, count: 10, autoCommit: false)
+ #expect(fresh.messages.count == 1)
+ #expect(fresh.context.incarnation != polled.context.incarnation)
+ }
+ }
+
+ @Test func rawPartitionRequestsCarryTheirContext() async throws {
+ try await E2EEnvironment.withSession() { session in
+ let client = session.client
+ let scratch = try await session.scratchTopic(partitions: 1)
+ _ = try await client.sendBatch(
+ streamID: scratch.stream, topicID: scratch.topic, partitioning: .partition(0), messages: try E2EEnvironment.messages(2))
+ let target = Self.rawConsumerTarget(consumer: 7, stream: scratch.streamID, topic: scratch.topicID)
+ // Strategy first, count 10, no auto-commit.
+ let poll = target + [3] + Self.littleEndian(UInt64(0)) + Self.littleEndian(UInt32(10)) + [0]
+ #expect(!(try await client.sendRawRequest(code: CommandCode.pollMessages.rawValue, payload: poll)).isEmpty)
+ // Offset 1, quorum acknowledgement.
+ let store = target + Self.littleEndian(UInt64(1)) + [1]
+ _ = try await client.sendRawRequest(code: CommandCode.storeConsumerOffset.rawValue, payload: store)
+ let stored = try await client.getConsumerOffset(consumer: .consumer(7), streamID: scratch.stream, topicID: scratch.topic, partitionID: 0)
+ #expect(stored?.storedOffset == 1)
+ }
+ }
+
+ /// `[consumer kind:1][consumer][stream][topic][has partition:1][partition 0:4]`
+ /// with numeric identifiers `[kind 1][length 4][id]`.
+ private static func rawConsumerTarget(consumer: UInt32, stream: UInt32, topic: UInt32) -> [UInt8] {
+ let numeric = { (id: UInt32) in [1, 4] + littleEndian(id) }
+ return [1] + numeric(consumer) + numeric(stream) + numeric(topic) + [1] + littleEndian(UInt32(0))
+ }
+
+ private static func littleEndian<T: FixedWidthInteger>(_ value: T) -> [UInt8] {
+ withUnsafeBytes(of: value.littleEndian) { Array($0) }
+ }
+
+ @Test func consumerGroupPolling() async throws {
+ try await E2EEnvironment.withSession() { session in
+ let client = session.client
+ let scratch = try await session.scratchTopic(partitions: 3)
+ let group = try await client.createConsumerGroup(streamID: scratch.stream, topicID: scratch.topic, name: "swift-e2e-group")
+ let groupID = Identifier(numeric: group.id)
+
+ // Polling a group this client never joined is an error, not silence.
+ await expectIggyError(.consumerGroupMemberNotFound) {
+ _ = try await client.pollMessages(
+ streamID: scratch.stream, topicID: scratch.topic, partitionID: nil, consumer: .group(groupID), strategy: .next, count: 2, autoCommit: true)
+ }
+
+ try await client.joinConsumerGroup(streamID: scratch.stream, topicID: scratch.topic, groupID: groupID)
+ for partition: UInt32 in 0..<3 {
+ _ = try await client.sendBatch(
+ streamID: scratch.stream, topicID: scratch.topic, partitioning: .partition(partition), messages: try E2EEnvironment.messages(2))
+ }
+ // The server installs the member's partitions after the join, so
+ // the member polls until it has read every one of them. A group
+ // poll without a partition round-robins the owned partitions.
+ var polledPartitions = Set<UInt32>()
+ let polledAll = try await E2EEnvironment.eventually {
+ let polled = try await client.pollMessages(
+ streamID: scratch.stream, topicID: scratch.topic, partitionID: nil, consumer: .group(groupID), strategy: .next, count: 2, autoCommit: true)
+ if !polled.messages.isEmpty {
+ #expect(polled.messages.count == 2)
+ polledPartitions.insert(polled.partitionID)
+ }
+ return polledPartitions.count == 3
+ }
+ #expect(polledAll)
+ #expect(polledPartitions == [0, 1, 2])
+ try await client.leaveConsumerGroup(streamID: scratch.stream, topicID: scratch.topic, groupID: groupID)
+ try await client.deleteConsumerGroup(streamID: scratch.stream, topicID: scratch.topic, groupID: groupID)
+ }
+ }
+
+ @Test func segmentDeletion() async throws {
+ try await E2EEnvironment.withSession() { session in
+ let client = session.client
+ let scratch = try await session.scratchTopic(partitions: 1)
+ _ = try await client.sendBatch(
+ streamID: scratch.stream, topicID: scratch.topic, partitioning: .partition(0), messages: try E2EEnvironment.messages(3))
+ // Only closed segments can be deleted; a fresh partition has one open
+ // segment, so the request is accepted and removes nothing.
+ try await client.deleteSegments(streamID: scratch.stream, topicID: scratch.topic, partitionID: 0, count: 1)
+ let topic = try #require(try await client.getTopic(streamID: scratch.stream, topicID: scratch.topic))
+ #expect(topic.partitions[0].messagesCount == 3)
+ }
+ }
+
+ @Test func largePayloadsAndBigBatches() async throws {
+ try await E2EEnvironment.withSession() { session in
+ let client = session.client
+ let scratch = try await session.scratchTopic(partitions: 1)
+ let big = try IggyMessage(payload: [UInt8](repeating: 0xAB, count: 512 * 1024))
+ let many = try (0..<1000).map { try IggyMessage("message \($0)") }
+ _ = try await client.sendBatch(streamID: scratch.stream, topicID: scratch.topic, partitioning: .partition(0), messages: [big] + many)
+ let polled = try await client.pollMessages(
+ streamID: scratch.stream, topicID: scratch.topic, partitionID: 0, consumer: .default, strategy: .first, count: 1001, autoCommit: false)
+ #expect(polled.messages.count == 1001)
+ #expect(polled.messages[0].payload.count == 512 * 1024)
+ #expect(polled.messages[1000].payloadString == "message 999")
+ }
+ }
+
+ @Test func emptyBatchesAreRejectedLocally() async throws {
+ try await E2EEnvironment.withSession() { session in
+ let client = session.client
+ await #expect(throws: IggyError(.invalidMessagesCount)) {
+ try await client.sendBatch(streamID: 1, topicID: 1, partitioning: .balanced, messages: [])
+ }
+ // The balanced send reads the topic first and finds none, as the
+ // Rust client does.
+ await #expect(throws: IggyError(.topicIdNotFound)) {
+ try await client.sendBatch(streamID: "no-such-stream", topicID: 1, partitioning: .balanced, messages: ["x"])
+ }
+ }
+ }
+}
+
+@Suite("System", requiresServer)
+struct SystemE2ETests {
+ @Test func statsClientsAndCluster() async throws {
+ try await E2EEnvironment.withSession() { session in
+ let client = session.client
+ let stats = try await client.getStats()
+ #expect(stats.processID > 0)
+ #expect(!stats.iggyServerVersion.isEmpty)
+ #expect(stats.totalMemory > 0)
+
+ let me = try await client.getMe()
+ let clients = try await client.getClients()
+ #expect(clients.contains { $0.clientID == me.clientID })
+ let fetched = try #require(try await client.getClient(id: me.clientID))
+ #expect(fetched.address == me.address)
+ #expect(try await client.getClient(id: UInt32.max - 7) == nil)
+
+ let metadata = try await client.getClusterMetadata()
+ #expect(!metadata.nodes.isEmpty)
+ #expect(metadata.nodes.allSatisfy { $0.endpoints.tcp != 0 })
+ }
+ }
+
+ @Test func snapshot() async throws {
+ try await E2EEnvironment.withSession() { session in
+ let client = session.client
+ let snapshot = try await client.snapshot(compression: .stored, types: [.filesystemOverview])
+ #expect(!snapshot.data.isEmpty)
+ }
+ }
+
+ @Test func rawRequests() async throws {
+ try await E2EEnvironment.withSession() { session in
+ let client = session.client
+ // Raw traffic never consumes a metadata request id, so the session stays
+ // usable for the replicated command that follows.
+ for _ in 0..<5 {
+ _ = try? await client.sendRawRequest(code: 60_000, payload: [])
+ }
+ #expect(try await client.sendRawRequest(code: CommandCode.ping.rawValue, payload: []).isEmpty)
+ #expect(!(try await client.sendRawRequest(code: CommandCode.getStats.rawValue, payload: [])).isEmpty)
+ await #expect(throws: IggyError(.invalidCommand)) {
+ try await client.sendRawRequest(code: 60_000, payload: [])
+ }
+ await #expect(throws: IggyError.self) {
+ try await client.sendRawRequest(code: CommandCode.loginUser.rawValue, payload: [])
+ }
+ let scratch = try await session.scratchTopic(partitions: 1)
+ #expect(try await client.getTopic(streamID: scratch.stream, topicID: scratch.topic) != nil)
+ }
+ }
+}
diff --git a/foreign/swift/Tests/IggyE2ETests/Support/E2EEnvironment.swift b/foreign/swift/Tests/IggyE2ETests/Support/E2EEnvironment.swift
new file mode 100644
index 000000000..4923cfd66
--- /dev/null
+++ b/foreign/swift/Tests/IggyE2ETests/Support/E2EEnvironment.swift
@@ -0,0 +1,212 @@
+// 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 Iggy
+import Testing
+
+/// The end-to-end suite runs against a live server named by
+/// `IGGY_TCP_ADDRESS` and is skipped when that variable is unset. Start one
+/// from a checkout of the repository:
+///
+/// ```text
+/// cargo build --bin iggy-server
+/// IGGY_PATH=/tmp/iggy-swift IGGY_TCP_ADDRESS=127.0.0.1:8090 \
+/// IGGY_HTTP_ENABLED=false IGGY_QUIC_ENABLED=false IGGY_WEBSOCKET_ENABLED=false \
+/// IGGY_ROOT_USERNAME=iggy IGGY_ROOT_PASSWORD=iggy target/debug/iggy-server
+///
+/// IGGY_TCP_ADDRESS=127.0.0.1:8090 swift test --filter IggyE2ETests
+/// ```
+///
+/// Set `IGGY_TCP_TLS_ENABLED=true` to run against a server started with
+/// `IGGY_TCP_TLS_ENABLED=true` and the certificate pair in `core/certs`.
+enum E2EEnvironment {
+ static let rootUsername = ProcessInfo.processInfo.environment["IGGY_ROOT_USERNAME"] ?? "iggy"
+ static let rootPassword = ProcessInfo.processInfo.environment["IGGY_ROOT_PASSWORD"] ?? "iggy"
+
+ static var address: String? {
+ ProcessInfo.processInfo.environment["IGGY_TCP_ADDRESS"]
+ }
+
+ static var isAvailable: Bool { address != nil }
+
+ static var tlsEnabled: Bool {
+ ProcessInfo.processInfo.environment["IGGY_TCP_TLS_ENABLED"] == "true"
+ }
+
+ /// Path of a file in `core/certs`, resolved from this source file so the
+ /// suite works from any working directory.
+ static func certificatePath(_ name: String) -> String {
+ let root = URL(fileURLWithPath: #filePath)
+ .deletingLastPathComponent() // E2EEnvironment.swift
+ .deletingLastPathComponent() // Support
+ .deletingLastPathComponent() // IggyE2ETests
+ .deletingLastPathComponent() // Tests
+ .deletingLastPathComponent() // swift
+ .deletingLastPathComponent() // foreign
+ return root.appendingPathComponent("core/certs/\(name)").path
+ }
+
+ static var tlsOptions: TLSOptions? {
+ guard tlsEnabled else { return nil }
+ return TLSOptions(domain: "localhost", caFile: certificatePath("iggy_ca_cert.pem"))
+ }
+
+ static func configuration(autoLogin: Credentials? = nil) -> ClientConfiguration {
+ var configuration = ClientConfiguration(address: address ?? "127.0.0.1:8090", tls: tlsOptions, autoLogin: autoLogin)
+ configuration.reconnection.interval = .milliseconds(200)
+ configuration.reconnection.reestablishAfter = .zero
+ return configuration
+ }
+
+ /// A connected client without a session.
+ static func newClient(autoLogin: Credentials? = nil) async throws -> IggyClient {
+ let client = IggyClient(configuration: configuration(autoLogin: autoLogin))
+ try await client.connect()
+ return client
+ }
+
+ /// A connected client signed in as root.
+ static func connect() async throws -> IggyClient {
+ let client = try await newClient()
+ _ = try await client.login(username: rootUsername, password: rootPassword)
+ return client
+ }
+
+ /// Runs `body` with a connected client and tears everything down in order
+ /// afterwards: scratch streams are deleted, secondary clients and then the
+ /// primary client are shut down. Teardown runs even when `body` throws.
+ static func withSession(login: Bool = true, autoLogin: Credentials? = nil, _ body: (TestSession) async throws -> Void) async throws {
+ let client = try await newClient(autoLogin: autoLogin)
+ if login, autoLogin == nil {
+ _ = try await client.login(username: rootUsername, password: rootPassword)
+ }
+ let session = TestSession(client: client)
+ do {
+ try await body(session)
+ } catch {
+ await session.tearDown()
+ throw error
+ }
+ await session.tearDown()
+ }
+
+ /// A unique name per test, within the 255-byte bound.
+ static func uniqueName(_ prefix: String) -> String {
+ let name = "\(prefix)-\(UInt64(Date().timeIntervalSince1970 * 1_000_000))-\(UInt32.random(in: 0..<UInt32.max))"
+ return String(name.prefix(255))
+ }
+
+ /// Polls `condition` until it holds or `timeout` passes, for server state
+ /// that settles after the call that changed it returned.
+ static func eventually(timeout: Duration = .seconds(15), _ condition: () async throws -> Bool) async throws -> Bool {
+ let deadline = ContinuousClock.now + timeout
+ while ContinuousClock.now < deadline {
+ if try await condition() {
+ return true
+ }
+ try await Task.sleep(for: .milliseconds(100))
+ }
+ return try await condition()
+ }
+
+ static func messages(_ count: Int, prefix: String = "message") throws -> [IggyMessage] {
+ try (0..<count).map { try IggyMessage("\(prefix) \($0)") }
+ }
+}
+
+/// One test's clients and the server-side resources it created.
+actor TestSession {
+ struct ScratchTopic: Sendable {
+ let stream: Identifier
+ let topic: Identifier
+ let streamID: UInt32
+ let topicID: UInt32
+ }
+
+ nonisolated let client: IggyClient
+ private var streams: [Identifier] = []
+ private var secondaries: [IggyClient] = []
+
+ init(client: IggyClient) {
+ self.client = client
+ }
+
+ /// Deletes the stream once the test finishes.
+ func track(stream: Identifier) {
+ streams.append(stream)
+ }
+
+ /// A second connected, signed-out client, shut down with the session.
+ func secondaryClient() async throws -> IggyClient {
+ let other = try await E2EEnvironment.newClient()
+ secondaries.append(other)
+ return other
+ }
+
+ /// A stream and topic scoped to the test.
+ func scratchTopic(partitions: UInt32) async throws -> ScratchTopic {
+ let name = E2EEnvironment.uniqueName("swift-e2e")
+ let stream = try await client.createStream(name: name)
+ let streamID = Identifier(numeric: stream.id)
+ // Tracked by name: the server reuses freed numeric ids, so deleting
+ // by id after a test's own delete could hit another test's stream.
+ streams.append(try Identifier(named: name))
+ let topic = try await client.createTopic(streamID: streamID, name: name, options: TopicCreateOptions(partitionsCount: partitions))
+ return ScratchTopic(stream: streamID, topic: Identifier(numeric: topic.id), streamID: stream.id, topicID: topic.id)
+ }
+
+ func tearDown() async {
+ if await client.state == .authenticated {
+ for stream in streams {
+ try? await client.deleteStream(stream)
+ }
+ }
+ for other in secondaries {
+ await other.shutdown()
+ }
+ await client.shutdown()
+ }
+}
+
+/// Skips a test when no server is configured.
+let requiresServer: any SuiteTrait = .enabled(if: E2EEnvironment.isAvailable, "set IGGY_TCP_ADDRESS to run the end-to-end suite against a server")
+
+/// Asserts that `body` throws an ``IggyError`` with the given code, whatever
+/// context the error carries.
+func expectIggyError(_ 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)
+ }
+}
+
+extension IggyClient {
+ /// ``sendMessages(streamID:topicID:partitioning:messages:)`` for a batch
+ /// the test does not read back, so the call can take a literal.
+ @discardableResult
+ func sendBatch(
+ streamID: Identifier, topicID: Identifier, partitioning: Partitioning = .balanced, messages: [IggyMessage]
+ ) async throws -> SendMessagesResponse {
+ var batch = messages
+ return try await sendMessages(streamID: streamID, topicID: topicID, partitioning: partitioning, messages: &batch)
+ }
+}
diff --git a/foreign/swift/Tests/IggyTests/ClientTransportTests.swift b/foreign/swift/Tests/IggyTests/ClientTransportTests.swift
index 7dbd0d802..e08aed687 100644
--- a/foreign/swift/Tests/IggyTests/ClientTransportTests.swift
+++ b/foreign/swift/Tests/IggyTests/ClientTransportTests.swift
@@ -169,7 +169,7 @@ struct ClientTransportTests {
defer { Task { await server.stop() } }
let counter = Locked(0)
await server.respond { request in
- if request.operation == .register { return .ok(FakeVSRServer.registerBody()) }
+ if let reply = FakeVSRServer.sessionReply(request) { return reply }
if request.operation == .deleteStream {
let seen = counter.withLock {
$0 += 1; return $0
@@ -194,7 +194,7 @@ struct ClientTransportTests {
defer { Task { await server.stop() } }
let counter = Locked(0)
await server.respond { request in
- if request.operation == .register { return .ok(FakeVSRServer.registerBody()) }
+ if let reply = FakeVSRServer.sessionReply(request) { return reply }
if request.operation == .nonReplicated, request.code == CommandCode.getClusterMetadata.rawValue {
return .ok(FakeVSRServer.rosterBody([server.address]))
}
@@ -225,7 +225,7 @@ struct ClientTransportTests {
defer { Task { await server.stop() } }
let applied = Locked<UInt64>(0)
await server.respond { request in
- if request.operation == .register { return .ok(FakeVSRServer.registerBody()) }
+ if let reply = FakeVSRServer.sessionReply(request) { return reply }
if request.operation == .nonReplicated, request.code == CommandCode.getClusterMetadata.rawValue {
return .ok(FakeVSRServer.rosterBody([server.address]))
}
@@ -272,7 +272,7 @@ struct ClientTransportTests {
let addresses = [first.address, second.address]
let viewChanged = Locked(false)
await first.respond { request in
- if request.operation == .register { return .ok(FakeVSRServer.registerBody()) }
+ if let reply = FakeVSRServer.sessionReply(request) { return reply }
if request.code == CommandCode.getClusterMetadata.rawValue {
return .ok(FakeVSRServer.rosterBody(addresses, leaderIndex: viewChanged.withLock { $0 } ? 1 : 0))
}
@@ -281,7 +281,7 @@ struct ClientTransportTests {
return .status(IggyErrorCode.notPrimary.rawValue)
}
await second.respond { request in
- if request.operation == .register { return .ok(FakeVSRServer.registerBody()) }
+ if let reply = FakeVSRServer.sessionReply(request) { return reply }
if request.code == CommandCode.getClusterMetadata.rawValue { return .ok(FakeVSRServer.rosterBody(addresses, leaderIndex: 1)) }
return .ok([])
}
@@ -302,7 +302,7 @@ struct ClientTransportTests {
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 let reply = FakeVSRServer.sessionReply(request) { return reply }
if request.code == CommandCode.getClusterMetadata.rawValue { return .ok(FakeVSRServer.rosterBody([server.address])) }
return request.operation == .deleteStream ? .status(IggyErrorCode.notPrimary.rawValue) : .ok([])
}
@@ -341,7 +341,7 @@ struct ClientTransportTests {
defer { Task { await server.stop() } }
let stallOnce = Locked(true)
await server.respond { request in
- if request.operation == .register { return .ok(FakeVSRServer.registerBody()) }
+ if let reply = FakeVSRServer.sessionReply(request) { return reply }
if request.operation == .nonReplicated, request.code == CommandCode.getStats.rawValue,
stallOnce.withLock({
let stall = $0; $0 = false; return stall
@@ -362,7 +362,9 @@ struct ClientTransportTests {
#expect(await server.connectionCount == 1)
try await client.ping()
#expect(await server.connectionCount == 2)
- #expect(await operations(server).filter { $0 == .register }.count == 2)
+ // The new connection resumes the session instead of registering one.
+ #expect(await operations(server).filter { $0 == .register }.count == 1)
+ #expect(await codes(server).filter { $0 == CommandCode.bindSession.rawValue }.count == 1)
#expect(await client.state == .authenticated)
await client.shutdown()
}
@@ -376,7 +378,9 @@ struct ClientTransportTests {
await server.stop()
let clock = ContinuousClock()
let elapsed = await clock.measure {
- await expectCode(.disconnected) { try await client.ping() }
+ // The server closed the connection first, so the ping was never
+ // written.
+ await expectCode(.notConnected) { try await client.ping() }
}
#expect(elapsed >= .milliseconds(450) && elapsed < .seconds(10))
// No caller waits for the sweep any more, so it ends too.
@@ -400,7 +404,7 @@ struct ClientTransportTests {
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 let reply = FakeVSRServer.sessionReply(request) { return reply }
if request.code == CommandCode.getClusterMetadata.rawValue { return .ok(roster) }
guard request.operation == .createStream else { return .ok([]) }
let count = seen.withLock {
@@ -409,7 +413,7 @@ struct ClientTransportTests {
return .status((count == 1 ? IggyErrorCode.transientNotCommitted : .transientNotAccepted).rawValue)
}
await second.respond { request in
- if request.operation == .register { return .ok(FakeVSRServer.registerBody()) }
+ if let reply = FakeVSRServer.sessionReply(request) { return reply }
if request.code == CommandCode.getClusterMetadata.rawValue { return .ok(roster) }
return .ok([])
}
@@ -429,7 +433,7 @@ struct ClientTransportTests {
defer { Task { await server.stop() } }
let once = Locked(true)
await server.respond { request in
- if request.operation == .register { return .ok(FakeVSRServer.registerBody()) }
+ if let reply = FakeVSRServer.sessionReply(request) { return reply }
if request.code == CommandCode.ping.rawValue,
once.withLock({
let first = $0; $0 = false; return first
@@ -452,7 +456,8 @@ struct ClientTransportTests {
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.bindSession.rawValue { return .status(IggyErrorCode.unauthenticated.rawValue) }
+ if let reply = FakeVSRServer.sessionReply(request) { return reply }
if request.operation == .deleteStream,
once.withLock({
let first = $0; $0 = false; return first
@@ -464,8 +469,9 @@ struct ClientTransportTests {
}
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.
+ // The evicted session cannot be resumed. 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)
@@ -479,7 +485,8 @@ struct ClientTransportTests {
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.bindSession.rawValue { return .status(IggyErrorCode.unauthenticated.rawValue) }
+ if let reply = FakeVSRServer.sessionReply(request) { return reply }
if request.code == CommandCode.ping.rawValue,
once.withLock({
let first = $0; $0 = false; return first
@@ -508,28 +515,100 @@ struct ClientTransportTests {
await client.shutdown()
}
- @Test func replicatedRequestIsNotReplayedAfterAConnectionDrop() async throws {
+ /// A resumed session keeps the client table entry, so a replicated write
+ /// lost with its connection goes again under its own request id, and the
+ /// server answers it from its reply cache if it committed.
+ @Test func replicatedRequestIsReplayedUnderTheResumedSession() 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
- })
- {
+ if let reply = FakeVSRServer.sessionReply(request) { return reply }
+ if request.operation == .deleteStream, once.exchange(false) {
+ return .close
+ }
+ return .ok([])
+ }
+ let client = makeClient(server)
+ try await client.connect()
+ try await client.deleteStream(1)
+ let writes = await server.requests.filter { $0.operation == .deleteStream }
+ #expect(writes.count == 2)
+ #expect(Set(writes.map(\.requestID)).count == 1)
+ #expect(Set(writes.map(\.clientID)).count == 1)
+ #expect(await operations(server).filter { $0 == .register }.count == 1)
+ #expect(await codes(server).filter { $0 == CommandCode.bindSession.rawValue }.count == 1)
+ await client.shutdown()
+ }
+
+ /// The first attempt was written before its connection went, so a replay
+ /// the resumed session keeps refusing may still have committed: it is
+ /// reported as unknown, never stamped afresh under a new request id.
+ @Test func aRefusedResumedReplayIsNotStampedAfresh() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ let once = Locked(true)
+ await server.respond { request in
+ if let reply = FakeVSRServer.sessionReply(request) { return reply }
+ if request.operation == .deleteStream {
+ return once.exchange(false) ? .close : .status(IggyErrorCode.transientNotAccepted.rawValue)
+ }
+ return .ok([])
+ }
+ let client = makeClient(server, requestTimeout: ClientCore.transientFailoverCheckInterval + .seconds(1))
+ try await client.connect()
+ await expectCode(.transientNotCommitted) { try await client.deleteStream(1) }
+ let attempts = await server.requests.filter { $0.operation == .deleteStream }.map(\.requestID)
+ #expect(attempts.count > 2)
+ #expect(Set(attempts).count == 1)
+ #expect(await codes(server).filter { $0 == CommandCode.bindSession.rawValue }.count == 1)
+ await client.shutdown()
+ }
+
+ /// A reconnect that fails after the request was written leaves its
+ /// outcome unknown; the reconnect's own error must not hide that.
+ @Test func aFailedReconnectKeepsTheUnknownOutcome() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ let registers = Locked(0)
+ await server.respond { request in
+ if request.code == CommandCode.bindSession.rawValue { return .status(IggyErrorCode.unauthenticated.rawValue) }
+ if request.operation == .register {
+ let first = registers.withLock {
+ $0 += 1; return $0 == 1
+ }
+ return first ? .ok(FakeVSRServer.registerBody()) : .status(IggyErrorCode.invalidCredentials.rawValue)
+ }
+ return request.operation == .deleteStream ? .close : .ok([])
+ }
+ let client = makeClient(server)
+ try await client.connect()
+ await expectCode(.transientNotCommitted) { try await client.deleteStream(1) }
+ #expect(await server.requests.filter { $0.operation == .deleteStream }.count == 1)
+ await client.shutdown()
+ }
+
+ @Test func replicatedRequestIsNotReplayedUnderANewSession() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ let once = Locked(true)
+ await server.respond { request in
+ if request.code == CommandCode.bindSession.rawValue { return .status(IggyErrorCode.unauthenticated.rawValue) }
+ if let reply = FakeVSRServer.sessionReply(request) { return reply }
+ if request.operation == .createStream, once.exchange(false) {
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.
+ // The server no longer holds the session, so the client signs in
+ // under a new one. The outcome is unknown: the client 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 operations(server).filter { $0 == .register }.count == 2)
#expect(await client.state == .authenticated)
#expect(await server.connectionCount == 2)
try await client.ping()
@@ -569,7 +648,7 @@ struct ClientTransportTests {
defer { Task { await server.stop() } }
let once = Locked(true)
await server.respond { request in
- if request.operation == .register { return .ok(FakeVSRServer.registerBody()) }
+ if let reply = FakeVSRServer.sessionReply(request) { return reply }
if request.code == CommandCode.getStats.rawValue,
once.withLock({
let first = $0; $0 = false; return first
@@ -630,6 +709,49 @@ struct ClientTransportTests {
await client.shutdown()
}
+ /// A teardown, such as a failover moving the shared connection, waits for
+ /// the reply of a frame another caller already wrote instead of cutting
+ /// it, which would leave the write's outcome unknown and replay it.
+ @Test func aTeardownWaitsForTheExchangeInFlight() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ await server.respond { request in
+ if let reply = FakeVSRServer.sessionReply(request) { return reply }
+ if request.operation == .deleteStream {
+ try? await Task.sleep(for: .milliseconds(300))
+ }
+ return .ok([])
+ }
+ let client = makeClient(server)
+ try await client.connect()
+ let write = Task { try await client.deleteStream(1) }
+ #expect(await eventually { await operations(server).contains(.deleteStream) })
+ await client.core.disconnectTransport()
+ try await write.value
+ #expect(await operations(server).filter { $0 == .deleteStream }.count == 1)
+ #expect(await client.state == .disconnected)
+ try await client.ping()
+ #expect(await server.connectionCount == 2)
+ await client.shutdown()
+ }
+
+ /// An explicit disconnect ends the session, so the next connect signs in
+ /// under a new one instead of resuming it.
+ @Test func anExplicitDisconnectIsNotResumed() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ let client = makeClient(server)
+ try await client.connect()
+ await client.disconnect()
+ try await client.connect()
+ try await client.ping()
+ let registers = await server.requests.filter { $0.operation == .register }
+ #expect(registers.count == 2)
+ #expect(Set(registers.map(\.clientID)).count == 2)
+ #expect(await codes(server).filter { $0 == CommandCode.bindSession.rawValue }.isEmpty)
+ await client.shutdown()
+ }
+
@Test func serverInitiatedCloseIsRepairedByTheNextRequest() async throws {
let server = try await FakeVSRServer.started()
defer { Task { await server.stop() } }
@@ -640,7 +762,8 @@ struct ClientTransportTests {
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)
+ #expect(await operations(server).filter { $0 == .register }.count == 1)
+ #expect(await codes(server).filter { $0 == CommandCode.bindSession.rawValue }.count == 1)
await client.shutdown()
var seen: [DiagnosticEvent] = []
for await event in events {
@@ -674,9 +797,19 @@ struct ClientTransportTests {
await client.shutdown()
}
+ /// A server that lost every session, as after a restart, so a reconnect
+ /// signs in again instead of resuming.
+ private static func forgetSessions(on server: FakeVSRServer) async {
+ await server.respond { request in
+ if request.code == CommandCode.bindSession.rawValue { return .status(IggyErrorCode.unauthenticated.rawValue) }
+ return FakeVSRServer.defaultReply(request, address: server.address)
+ }
+ }
+
@Test func passwordChangeUpdatesTheReconnectCredentials() async throws {
let server = try await FakeVSRServer.started()
defer { Task { await server.stop() } }
+ await Self.forgetSessions(on: server)
let client = makeClient(server)
try await client.connect()
try await client.changePassword("iggy", currentPassword: "iggy", newPassword: "new-secret")
@@ -698,6 +831,7 @@ struct ClientTransportTests {
@Test func userRenameUpdatesTheReconnectCredentials() async throws {
let server = try await FakeVSRServer.started()
defer { Task { await server.stop() } }
+ await Self.forgetSessions(on: server)
let client = makeClient(server)
try await client.connect()
try await client.updateUser("iggy", username: "renamed")
@@ -746,7 +880,7 @@ struct ClientTransportTests {
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 let reply = FakeVSRServer.sessionReply(request) { return reply }
if request.code == CommandCode.getClusterMetadata.rawValue { return .ok(roster) }
return .ok([])
}
@@ -769,14 +903,14 @@ struct ClientTransportTests {
/// 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.
+ /// resumes the session, 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() }
+ await expectCode(.notConnected) { try await client.ping() }
#expect(await eventually { await client.state == .disconnected })
let revived = FakeVSRServer()
@@ -784,7 +918,7 @@ struct ClientTransportTests {
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)
+ #expect(await codes(revived).filter { $0 == CommandCode.bindSession.rawValue }.count == 1)
await client.shutdown()
}
@@ -803,7 +937,7 @@ struct ClientTransportTests {
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)
+ #expect(await codes(revived).filter { $0 == CommandCode.bindSession.rawValue }.count == 1)
await client.shutdown()
}
@@ -813,7 +947,7 @@ struct ClientTransportTests {
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 let reply = FakeVSRServer.sessionReply(request) { return reply }
if request.code == CommandCode.ping.rawValue {
try? await Task.sleep(for: .milliseconds(300))
}
@@ -850,7 +984,7 @@ struct ClientTransportTests {
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 let reply = FakeVSRServer.sessionReply(request) { return reply }
try? await Task.sleep(for: .milliseconds(2))
return .ok([])
}
@@ -882,7 +1016,7 @@ struct ClientTransportTests {
defer { Task { await server.stop() } }
let once = Locked(true)
await server.respond { request in
- if request.operation == .register { return .ok(FakeVSRServer.registerBody()) }
+ if let reply = FakeVSRServer.sessionReply(request) { return reply }
guard request.code == CommandCode.getStats.rawValue else { return .ok([]) }
let first = once.withLock {
let first = $0; $0 = false; return first
@@ -903,7 +1037,8 @@ struct ClientTransportTests {
try await group.waitForAll()
}
#expect(await server.connectionCount == 2)
- #expect(await operations(server).filter { $0 == .register }.count == 2)
+ #expect(await operations(server).filter { $0 == .register }.count == 1)
+ #expect(await codes(server).filter { $0 == CommandCode.bindSession.rawValue }.count == 1)
await client.shutdown()
}
@@ -916,7 +1051,7 @@ struct ClientTransportTests {
let seen = Locked(0)
let refuseUntil = Locked<ContinuousClock.Instant?>(nil)
await server.respond { request in
- if request.operation == .register { return .ok(FakeVSRServer.registerBody()) }
+ if let reply = FakeVSRServer.sessionReply(request) { return reply }
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 {
@@ -948,7 +1083,7 @@ struct ClientTransportTests {
defer { Task { await server.stop() } }
let once = Locked(true)
await server.respond { request in
- if request.operation == .register { return .ok(FakeVSRServer.registerBody()) }
+ if let reply = FakeVSRServer.sessionReply(request) { return reply }
guard request.code == CommandCode.getStats.rawValue else { return .ok([]) }
let first = once.withLock {
let first = $0; $0 = false; return first
diff --git a/foreign/swift/Tests/IggyTests/ConcurrencyTests.swift b/foreign/swift/Tests/IggyTests/ConcurrencyTests.swift
index 0a53603ad..12cf004a0 100644
--- a/foreign/swift/Tests/IggyTests/ConcurrencyTests.swift
+++ b/foreign/swift/Tests/IggyTests/ConcurrencyTests.swift
@@ -206,4 +206,12 @@ struct ConcurrencyTests {
#expect(outcome.observe(IggyError(.disconnected)).code == .disconnected)
#expect(outcome.observe(IggyError(.streamNameAlreadyExists)).code == .streamNameAlreadyExists)
}
+
+ /// The server took a later request of the session first, so an earlier
+ /// attempt of this one may have committed.
+ @Test func retryOutcomeReportsARequestTooOldAsUnknown() {
+ var outcome = RetryOutcome()
+ #expect(outcome.observe(IggyError(.requestTooOld)).code == .transientNotCommitted)
+ #expect(outcome.isUncertain)
+ }
}
diff --git a/foreign/swift/Tests/IggyTests/ConsumerGroupStateTests.swift b/foreign/swift/Tests/IggyTests/ConsumerGroupStateTests.swift
new file mode 100644
index 000000000..42f970ebd
--- /dev/null
+++ b/foreign/swift/Tests/IggyTests/ConsumerGroupStateTests.swift
@@ -0,0 +1,122 @@
+// 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
+
+/// A port of the tests of `ConsumerGroupClientState` in `core/common`.
+@Suite("Consumer group state")
+struct ConsumerGroupStateTests {
+ private let group = ConsumerGroupState.GroupKey(stream: 1, topic: 2, group: 3)
+ private let topic = ConsumerGroupState.TopicKey(stream: 1, topic: 2)
+
+ private static func partitions(_ count: UInt32) -> [Partition] {
+ (0..<count).map { id in
+ Partition(
+ id: id, createdAt: .zero, segmentsCount: 1, currentOffset: 0, sizeBytes: 0, messagesCount: 0,
+ context: PartitionContext(incarnation: UInt64(id) + 1, ownerGeneration: 0, metadataOp: 10))
+ }
+ }
+
+ @Test func topicIdentifiersDoNotAliasAndDiscoveryRefreshKeepsTheCursor() {
+ let state = ConsumerGroupState()
+ let numeric = ConsumerGroupState.TopicKey(stream: 1, topic: 2)
+ let named = ConsumerGroupState.TopicKey(stream: 1, topic: "orders")
+ state.setTopicPartitions(numeric, partitions: Self.partitions(3))
+ #expect(state.partitionCount(named) == nil)
+ #expect(state.nextBalancedPartition(numeric, partitionCount: 3) == 0)
+ state.invalidateTopicDiscovery()
+ #expect(state.partitionCount(numeric) == nil)
+ #expect(state.nextBalancedPartition(numeric, partitionCount: 3) == 1)
+ }
+
+ @Test func groupPartitionsRoundRobinThenWrap() {
+ let state = ConsumerGroupState()
+ state.applySync(group, member: (1, [0, 1, 2]), scope: state.sessionScope)
+ #expect((0..<4).map { _ in state.nextGroupPartition(group) } == [0, 1, 2, 0])
+ }
+
+ @Test func aGenerationChangeRestartsTheCursor() {
+ let state = ConsumerGroupState()
+ state.applySync(group, member: (1, [0, 1, 2]), scope: state.sessionScope)
+ #expect(state.nextGroupPartition(group) == 0)
+ #expect(state.nextGroupPartition(group) == 1)
+ state.applySync(group, member: (2, [5, 6]), scope: state.sessionScope)
+ #expect(state.nextGroupPartition(group) == 5)
+ }
+
+ @Test func balancedRoundRobins() {
+ let state = ConsumerGroupState()
+ #expect((0..<4).map { _ in state.nextBalancedPartition(topic, partitionCount: 3) } == [0, 1, 2, 0])
+ #expect(state.nextBalancedPartition(topic, partitionCount: 0) == 0)
+ }
+
+ @Test func aRouteContextDoesNotInventAPartitionCount() {
+ let state = ConsumerGroupState()
+ let context = PartitionContext(incarnation: 9, ownerGeneration: 0, metadataOp: 4)
+ state.setPartitionContext(topic, partitionID: 2, context)
+ #expect(state.partitionContext(topic, partitionID: 2) == context)
+ #expect(state.partitionCount(topic) == nil)
+ state.setTopicPartitions(topic, partitions: Self.partitions(3))
+ #expect(state.partitionCount(topic) == 3)
+ #expect(state.partitionContext(topic, partitionID: 2) == Self.partitions(3)[2].context)
+ state.invalidateTopic(topic)
+ #expect(state.partitionContext(topic, partitionID: 2) == nil)
+ #expect(state.partitionCount(topic) == nil)
+ }
+
+ @Test func aMissingAssignmentYieldsNothing() {
+ let state = ConsumerGroupState()
+ #expect(!state.hasAssignment(group))
+ #expect(state.nextGroupPartition(group) == nil)
+ }
+
+ @Test func aMemberHoldingNoPartitionsStaysRegistered() {
+ let state = ConsumerGroupState()
+ state.applySync(group, member: (1, []), scope: state.sessionScope)
+ #expect(!state.hasAssignment(group))
+ #expect(state.isRegistered(group))
+ state.deregisterGroup(group)
+ #expect(!state.isRegistered(group))
+ }
+
+ @Test func aSessionResetDropsMembershipAndKeepsTopicState() {
+ let state = ConsumerGroupState()
+ state.applySync(group, member: (1, [0]), scope: state.sessionScope)
+ state.setTopicPartitions(topic, partitions: Self.partitions(2))
+ #expect(state.nextBalancedPartition(topic, partitionCount: 2) == 0)
+ state.clearSessionScoped()
+ #expect(state.registeredGroups.isEmpty)
+ #expect(!state.isRegistered(group))
+ #expect(!state.hasAssignment(group))
+ #expect(state.partitionCount(topic) == 2)
+ #expect(state.nextBalancedPartition(topic, partitionCount: 2) == 1)
+ }
+
+ @Test func aSyncAnsweredUnderAnEndedSessionIsDropped() {
+ let state = ConsumerGroupState()
+ let scope = state.sessionScope
+ state.clearSessionScoped()
+ #expect(!state.applySync(group, member: (1, [0]), scope: scope))
+ #expect(!state.isRegistered(group))
+ #expect(!state.hasAssignment(group))
+ state.applySync(group, member: (1, [0]), scope: state.sessionScope)
+ #expect(!state.applySync(group, member: nil, scope: scope))
+ #expect(state.isRegistered(group))
+ }
+}
diff --git a/foreign/swift/Tests/IggyTests/Fixtures/golden.json b/foreign/swift/Tests/IggyTests/Fixtures/golden.json
index 52b9f650b..cf0d1bc4b 100644
--- a/foreign/swift/Tests/IggyTests/Fixtures/golden.json
+++ b/foreign/swift/Tests/IggyTests/Fixtures/golden.json
@@ -944,6 +944,7 @@
"request.get_consumer_groups": "010401000000010402000000",
"request.get_consumer_offset": "01020b6d792d636f6e73756d6572020873747265616d2d310207746f7069632d310100000000",
"request.get_me": "",
+ "request.get_partition_route": "01040a00000002066f726465727305000000",
"request.get_personal_access_tokens": "",
"request.get_snapshot": "02020105",
"request.get_stats": "",
@@ -990,8 +991,10 @@
"response.get_users": "0100000000e090663c130600010561646d696e0000000002000000c8000000000000000205616c69636500000000",
"response.identity": "2a000000",
"response.login_register": "2a0000006400000000000000012c000006302e31312e30",
+ "response.partition_route": "07000000000000000a00000000000000030000000400000002",
"response.poll_messages": "07000000660000000000000003000000000000000000000000000000000000000000000000000000070000000000000064000000000000008813000000000000e8030000000000006e01000000000000f576a2492e1328b3020000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000 [...]
"response.poll_messages.empty": "00000000000000000000000000000000000000000000000000000000000000000000000000000000",
+ "response.poll_routing": "100f0e0d0c0b0a0908070605040302012a000000000000007800000000000000070000000000000003000000000000006e0000000000000002060000006e6f64652d330800000031302e302e302e339c1f0000ba0b00000100",
"response.raw_personal_access_token": "167261772d7365637265742d746f6b656e2d76616c7565",
"response.send_messages": "0200000001000000020000000300000004000000000000000100000002000000000000002a00000000000000",
"response.send_messages.empty": "00000000",
@@ -1007,7 +1010,7 @@
"vsr.eviction.incompatible_protocol": "0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000001000000000000000000000d000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000001032547698badcfeefcdab8967452301012c0000012c00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000 [...]
"vsr.eviction.stale_client": "0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000001000000000000000000000d000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000001032547698badcfeefcdab896745230100000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000 [...]
"vsr.reply.denied": "00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000010000000000000000000008000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000800000000000000020000000000000029000000000000000000000000000000000 [...]
- "vsr.reply.ok": "0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000c0100000000000000000000080000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000001032547698badcfeefcdab896745230100000000000000000000000000000000000000000000000007000000000000008000000000000000000000000000000000000000000000000000000 [...]
+ "vsr.reply.ok": "0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000c0100000000000000000000080000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000001032547698badcfeefcdab896745230100000000000000000c00000000000000000000000000000007000000000000008000000000000000000000000000000000000000000000000000000 [...]
"vsr.request.create_stream": "00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000007010000000000000000000005000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000001032547698badcfeefcdab89674523013a9e6b43adf27bf20000000000000000000000000000000001000000000000008000000000000000630000000000000000000000000000000000000000000000000000000000000000000000000000000000000000 [...]
"vsr.request.logout": "00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000010000000000000000000005000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000001032547698badcfeefcdab8967452301c294d3380580062d00000000000000000000000000000000040000000000000003000000000000006300000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000 [...]
"vsr.request.ping": "00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000010000000000000000000005000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000001032547698badcfeefcdab896745230100000000000000000000000000000000000000000000000001000000000000000200000000000000000000000000000000000000010000000000000000000000000000000000000000000000000000000000000000000000000 [...]
diff --git a/foreign/swift/Tests/IggyTests/MessagingTests.swift b/foreign/swift/Tests/IggyTests/MessagingTests.swift
new file mode 100644
index 000000000..a5c25ca5b
--- /dev/null
+++ b/foreign/swift/Tests/IggyTests/MessagingTests.swift
@@ -0,0 +1,756 @@
+// 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
+
+/// A signed-in client on the fake server.
+private func connectedClient(
+ _ server: FakeVSRServer, heartbeat: Duration = .seconds(60), requestTimeout: Duration = .seconds(10)
+) async throws -> IggyClient {
+ var configuration = ClientConfiguration(address: server.address, autoLogin: .usernamePassword(username: "iggy", password: "iggy"))
+ configuration.reconnection = ReconnectionOptions(maxRetries: nil, interval: .milliseconds(20), reestablishAfter: .zero)
+ configuration.requestTimeout = requestTimeout
+ configuration.connectTimeout = .seconds(2)
+ configuration.heartbeatInterval = heartbeat
+ let client = IggyClient(configuration: configuration, connector: NIOConnector(), logger: testLogger)
+ try await client.connect()
+ return client
+}
+
+private func requests(_ server: FakeVSRServer, _ code: CommandCode) async -> [FakeVSRServer.Request] {
+ await server.requests.filter { $0.code == code.rawValue }
+}
+
+/// The partition a send names: `[metadata_len:4][stream:6][topic:6][kind:1][length:1][partition:4]`
+/// with numeric ids.
+private func sentPartition(_ request: FakeVSRServer.Request) -> UInt32 {
+ UInt32(littleEndianBytes: request.payload[18..<22])
+}
+
+/// The partition a poll names: `[kind:1][consumer:6][stream:6][topic:6][flag:1][partition:4]`
+/// with numeric ids.
+private func polledPartition(_ request: FakeVSRServer.Request) -> UInt32 {
+ UInt32(littleEndianBytes: request.payload[20..<24])
+}
+
+private let otherContext = PartitionContext(incarnation: 8, ownerGeneration: 4, metadataOp: 30)
+
+private func contains(_ payload: [UInt8], _ id: MessageID) -> Bool {
+ var writer = ByteWriter()
+ writer.write(id)
+ let needle = writer.bytes
+ return payload.indices.contains { payload[$0...].starts(with: needle) }
+}
+
+@Suite("Messaging", .serialized)
+struct MessagingTests {
+ // MARK: Sends
+
+ @Test func sendsCaptureTheRouteContextOnce() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ let client = try await connectedClient(server)
+ try await client.sendBatch(streamID: 1, topicID: 2, partitioning: .partition(1), messages: ["a"])
+ try await client.sendBatch(streamID: 1, topicID: 2, partitioning: .partition(1), messages: ["b"])
+ let routes = await requests(server, .getPartitionRoute)
+ #expect(routes.count == 1)
+ #expect(routes.first?.payload == (try Requests.partitionRoute(streamID: 1, topicID: 2, partitionID: 1)))
+ let sends = await requests(server, .sendMessages)
+ #expect(sends.count == 2)
+ // A route names no consumer, so the send carries owner 0.
+ #expect(sends.allSatisfy { $0.context == FakeVSRServer.routeContext })
+ await client.shutdown()
+ }
+
+ @Test func mintedIdsAreWrittenBackAndSentAgainUnchanged() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ let client = try await connectedClient(server)
+ let chosen = MessageID(low: 9, high: 0)
+ var batch: [IggyMessage] = ["minted", try IggyMessage("chosen", id: chosen)]
+ try await client.sendMessages(streamID: 1, topicID: 2, partitioning: .partition(1), messages: &batch)
+ let minted = batch[0].id
+ #expect(!minted.isZero)
+ #expect(batch[1].id == chosen)
+ // A retry of the same batch carries the same ids.
+ try await client.sendMessages(streamID: 1, topicID: 2, partitioning: .partition(1), messages: &batch)
+ #expect(batch[0].id == minted)
+ let sends = await requests(server, .sendMessages)
+ #expect(sends.count == 2)
+ #expect(sends.allSatisfy { contains($0.payload, minted) && contains($0.payload, chosen) })
+ await client.shutdown()
+ }
+
+ /// A send lost with its connection may have committed, so a history
+ /// refusal of its resumed replay is an unknown outcome: the send is only
+ /// replayed under its own request id, never sent again under a new
+ /// route.
+ @Test func aResumedSendRefusedForItsHistoryIsNotSentAgain() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ let dropped = Locked(false)
+ await server.respond { request in
+ if request.code == CommandCode.sendMessages.rawValue {
+ return dropped.exchange(true) ? .status(IggyErrorCode.historyUnavailable.rawValue) : .close
+ }
+ return FakeVSRServer.defaultReply(request, address: server.address)
+ }
+ let client = try await connectedClient(server, requestTimeout: .seconds(1))
+ await expectCode(.transientNotCommitted) {
+ try await client.sendBatch(streamID: 1, topicID: 2, partitioning: .partition(1), messages: ["a"])
+ }
+ let sends = await requests(server, .sendMessages)
+ #expect(sends.count >= 2)
+ #expect(Set(sends.map(\.requestID)).count == 1)
+ #expect(await requests(server, .getPartitionRoute).count == 1)
+ await client.shutdown()
+ }
+
+ @Test func aStaleHistoryReadsTheRouteAgainOnce() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ let refusals = Locked(1)
+ let routes = Locked(0)
+ await server.respond { request in
+ switch request.code.flatMap(CommandCode.init(rawValue:)) {
+ case .getPartitionRoute:
+ let count = routes.withLock {
+ $0 += 1; return $0
+ }
+ return .ok(FakeVSRServer.partitionRouteBody(count == 1 ? FakeVSRServer.routeContext : otherContext))
+ case .sendMessages
+ where refusals.withLock({
+ $0 -= 1; return $0 >= 0
+ }):
+ return .status(IggyErrorCode.historyUnavailable.rawValue)
+ default:
+ return FakeVSRServer.defaultReply(request, address: server.address)
+ }
+ }
+ let client = try await connectedClient(server)
+ try await client.sendBatch(streamID: 1, topicID: 2, partitioning: .partition(0), messages: ["a"])
+ let sends = await requests(server, .sendMessages)
+ #expect(sends.map(\.context) == [FakeVSRServer.routeContext, PartitionContext(incarnation: 8, ownerGeneration: 0, metadataOp: 30)])
+ #expect(sends[0].requestID != sends[1].requestID)
+
+ // A second refusal in a row ends the send.
+ refusals.withLock { $0 = 2 }
+ await expectCode(.historyUnavailable) {
+ try await client.sendBatch(streamID: 1, topicID: 2, partitioning: .partition(0), messages: ["b"])
+ }
+ #expect(await requests(server, .sendMessages).count == 4)
+ #expect(routes.withLock { $0 } == 3)
+ await client.shutdown()
+ }
+
+ @Test func balancedSendsRoundRobinUnderTheTopicContexts() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ let contexts = (1...3).map { PartitionContext(incarnation: UInt64($0), ownerGeneration: 0, metadataOp: 5) }
+ await server.respond { request in
+ if request.code == CommandCode.getTopic.rawValue {
+ return .ok(FakeVSRServer.topicBody(contexts: contexts))
+ }
+ return FakeVSRServer.defaultReply(request, address: server.address)
+ }
+ let client = try await connectedClient(server)
+ for text in ["a", "b", "c", "d"] {
+ try await client.sendBatch(streamID: 1, topicID: 2, messages: [IggyMessage(stringLiteral: text)])
+ }
+ let sends = await requests(server, .sendMessages)
+ #expect(sends.map(sentPartition) == [0, 1, 2, 0])
+ // The topic read cached the count and the contexts: no route lookups.
+ #expect(sends.map(\.context) == [contexts[0], contexts[1], contexts[2], contexts[0]])
+ #expect(await requests(server, .getTopic).count == 1)
+ #expect(await requests(server, .getPartitionRoute).isEmpty)
+
+ try await client.sendBatch(streamID: 1, topicID: 2, partitioning: .messagesKey("user-1"), messages: ["e"])
+ #expect(await requests(server, .sendMessages).last.map(sentPartition) == XXH32.hash(Array("user-1".utf8)) % 3)
+
+ // A deleted topic can come back with other partitions.
+ try await client.deleteTopic(streamID: 1, topicID: 9)
+ try await client.sendBatch(streamID: 1, topicID: 2, messages: ["f"])
+ #expect(await requests(server, .getTopic).count == 2)
+ await client.shutdown()
+ }
+
+ @Test func aBalancedSendToAMissingTopicFails() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ let client = try await connectedClient(server)
+ await expectCode(.topicIdNotFound) { try await client.sendBatch(streamID: 1, topicID: 2, messages: ["a"]) }
+ #expect(await requests(server, .sendMessages).isEmpty)
+ await client.shutdown()
+ }
+
+ // MARK: Polls
+
+ @Test func pollsTakeTheirRouteOnASingleNode() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ let client = try await connectedClient(server)
+ let polled = try await client.pollMessages(
+ streamID: 1, topicID: 2, partitionID: 1, consumer: .consumer(5), strategy: .next, count: 10, autoCommit: true)
+ #expect(polled.context == FakeVSRServer.routeContext)
+ let continued = PollingStrategy.offset(3, context: otherContext)
+ _ = try await client.pollMessages(streamID: 1, topicID: 2, partitionID: 1, consumer: .consumer(5), strategy: continued, count: 10, autoCommit: true)
+
+ let lookups = await requests(server, .getPollRouting)
+ #expect(lookups.count == 1)
+ let payload = try Requests.pollMessages(
+ consumer: .consumer(5), streamID: 1, topicID: 2, partitionID: 1, strategy: .next, count: 10, autoCommit: true)
+ #expect(lookups.first?.payload == payload)
+ let polls = await requests(server, .pollMessagesOnPrimary)
+ #expect(polls.map(\.context) == [FakeVSRServer.routeContext, otherContext])
+ #expect(polls.first?.payload == payload)
+ // Every request went over the one coordinator connection.
+ #expect(await server.connectionCount == 1)
+ await client.shutdown()
+ }
+
+ @Test func aRefusedPollLooksUpItsRouteAgain() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ let refused = Locked(false)
+ await server.respond { request in
+ if request.code == CommandCode.pollMessagesOnPrimary.rawValue,
+ !refused.exchange(true)
+ {
+ return .status(IggyErrorCode.transientNotAccepted.rawValue)
+ }
+ return FakeVSRServer.defaultReply(request, address: server.address)
+ }
+ let client = try await connectedClient(server)
+ _ = try await client.pollMessages(streamID: 1, topicID: 2, partitionID: 0, strategy: .first, count: 1, autoCommit: false)
+ #expect(await requests(server, .getPollRouting).count == 2)
+ #expect(await requests(server, .pollMessagesOnPrimary).count == 2)
+ await client.shutdown()
+ }
+
+ /// The poll may have been applied, with auto-commit, before the
+ /// connection went: its outcome is unknown, as on a data connection.
+ @Test func aSingleNodePollLostAfterItWasWrittenIsAnUnknownOutcome() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ let dropped = Locked(false)
+ await server.respond { request in
+ if request.code == CommandCode.pollMessagesOnPrimary.rawValue, !dropped.exchange(true) {
+ return .close
+ }
+ return FakeVSRServer.defaultReply(request, address: server.address)
+ }
+ let client = try await connectedClient(server)
+ let poll = { _ = try await client.pollMessages(streamID: 1, topicID: 2, partitionID: 0, strategy: .first, count: 1, autoCommit: true) }
+ await expectCode(.transientNotCommitted) { try await poll() }
+ try await poll()
+ await client.shutdown()
+ }
+
+ @Test func aNewerMetadataCommitReadsTheRouteAgain() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ await server.respond { request in
+ if request.operation == .deleteStream {
+ return .committed([], commit: 50)
+ }
+ return FakeVSRServer.defaultReply(request, address: server.address)
+ }
+ let client = try await connectedClient(server)
+ let poll = { _ = try await client.pollMessages(streamID: 1, topicID: 2, partitionID: 0, strategy: .first, count: 1, autoCommit: false) }
+ try await poll()
+ try await poll()
+ #expect(await requests(server, .getPollRouting).count == 1)
+ try await client.deleteStream(9)
+ #expect(await client.core.metadataWatermark == 50)
+ try await poll()
+ try await poll()
+ // The route read after the commit is as new as the commit.
+ #expect(await requests(server, .getPollRouting).count == 2)
+ await client.shutdown()
+ }
+
+ @Test func aNewSessionDropsTheRoutesAndTheMembership() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ await server.respond { request in
+ if request.code == CommandCode.syncConsumerGroup.rawValue {
+ return .ok(FakeVSRServer.assignmentBody(generation: 1, partitions: [0]))
+ }
+ return FakeVSRServer.defaultReply(request, address: server.address)
+ }
+ let client = try await connectedClient(server)
+ _ = try await client.pollMessages(streamID: 1, topicID: 2, partitionID: nil, consumer: .group(3), strategy: .next, count: 1, autoCommit: true)
+ #expect(client.core.groupState.isRegistered(.init(stream: 1, topic: 2, group: 3)))
+ try await client.logout()
+ _ = try await client.login(username: "iggy", password: "iggy")
+ #expect(!client.core.groupState.isRegistered(.init(stream: 1, topic: 2, group: 3)))
+ _ = try await client.pollMessages(streamID: 1, topicID: 2, partitionID: 0, strategy: .first, count: 1, autoCommit: false)
+ let lookups = await requests(server, .getPollRouting)
+ #expect(lookups.count == 2)
+ #expect(lookups[0].clientID != lookups[1].clientID)
+ await client.shutdown()
+ }
+
+ // MARK: Sessions
+
+ /// A lost connection resumes the session, so the member keeps polling
+ /// without joining again.
+ @Test func aReconnectResumesTheSessionAndKeepsTheMembership() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ await server.respond { request in
+ if request.code == CommandCode.syncConsumerGroup.rawValue {
+ return .ok(FakeVSRServer.assignmentBody(generation: 1, partitions: [0]))
+ }
+ return FakeVSRServer.defaultReply(request, address: server.address)
+ }
+ let client = try await connectedClient(server)
+ let poll = {
+ try await client.pollMessages(streamID: 1, topicID: 2, partitionID: nil, consumer: .group(3), strategy: .next, count: 1, autoCommit: true)
+ }
+ _ = try await poll()
+ await server.closeAllConnections()
+ try await Task.sleep(for: .milliseconds(20))
+ _ = try await poll()
+ #expect(client.core.groupState.isRegistered(.init(stream: 1, topic: 2, group: 3)))
+ #expect(await server.connectionCount == 2)
+ #expect(await server.requests.filter { $0.operation == .register }.count == 1)
+ let binds = await requests(server, .bindSession)
+ #expect(binds.count == 1)
+ let polls = await requests(server, .pollMessagesOnPrimary)
+ #expect(polls.count == 2)
+ #expect(Set(polls.map(\.clientID)).count == 1)
+ #expect(polls.allSatisfy { $0.clientID == binds[0].clientID })
+ await client.shutdown()
+ }
+
+ /// When the server no longer holds the session the client signs in under
+ /// a new one, which is not a member until the caller joins again.
+ @Test func aSessionTheServerLostEndsTheMembership() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ let member = Locked<UInt128Value?>(nil)
+ await server.respond { request in
+ switch request.code.flatMap(CommandCode.init(rawValue:)) {
+ case .bindSession:
+ return .status(IggyErrorCode.unauthenticated.rawValue)
+ case .syncConsumerGroup:
+ let joined = member.withLock { joined in
+ if joined == nil { joined = request.clientID }
+ return joined == request.clientID
+ }
+ return .ok(joined ? FakeVSRServer.assignmentBody(generation: 1, partitions: [0]) : [])
+ default:
+ return FakeVSRServer.defaultReply(request, address: server.address)
+ }
+ }
+ let client = try await connectedClient(server)
+ let poll = {
+ try await client.pollMessages(streamID: 1, topicID: 2, partitionID: nil, consumer: .group(3), strategy: .next, count: 1, autoCommit: true)
+ }
+ _ = try await poll()
+ await server.closeAllConnections()
+ try await Task.sleep(for: .milliseconds(20))
+ try await client.ping()
+ #expect(await server.requests.filter { $0.operation == .register }.count == 2)
+ #expect(!client.core.groupState.isRegistered(.init(stream: 1, topic: 2, group: 3)))
+ await expectCode(.consumerGroupMemberNotFound) { _ = try await poll() }
+ await client.shutdown()
+ }
+
+ /// A data-connection frame is stamped only for the session its route was
+ /// read under, and only while no newer metadata commit was seen: anything
+ /// else is refused before it is written.
+ @Test func dataConnectionFramesAreStampedOnlyForTheLiveSession() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ await server.respond { request in
+ if request.operation == .deleteStream {
+ return .committed([], commit: 5)
+ }
+ return FakeVSRServer.defaultReply(request, address: server.address)
+ }
+ let client = try await connectedClient(server)
+ try await client.ping()
+ let live = await server.requests.last!
+ let stamp = { (watermark: UInt64) in
+ _ = try await client.core.stampRequest(
+ code: CommandCode.ping.rawValue, payload: [], context: .empty,
+ for: Responses.SessionIdentity(clientID: live.clientID, session: live.session, metadataWatermark: watermark))
+ }
+ try await stamp(0)
+ try await client.deleteStream(1)
+ await expectCode(.transientNotAccepted) { try await stamp(4) }
+ try await stamp(5)
+ try await client.logout()
+ await expectCode(.transientNotAccepted) { try await stamp(5) }
+ _ = try await client.login(username: "iggy", password: "iggy")
+ await expectCode(.transientNotAccepted) { try await stamp(5) }
+ await client.shutdown()
+ }
+
+ // MARK: Consumer groups
+
+ @Test func groupPollsRoundRobinOverTheAssignment() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ await server.respond { request in
+ switch request.code.flatMap(CommandCode.init(rawValue:)) {
+ case .syncConsumerGroup:
+ return .ok(FakeVSRServer.assignmentBody(generation: 1, partitions: [1, 2]))
+ case .pollMessagesOnPrimary:
+ return .ok(FakeVSRServer.polledBody(partitionID: polledPartition(request)))
+ default:
+ return FakeVSRServer.defaultReply(request, address: server.address)
+ }
+ }
+ let client = try await connectedClient(server)
+ var polled: [UInt32] = []
+ for _ in 0..<3 {
+ polled.append(
+ try await client.pollMessages(streamID: 1, topicID: 2, partitionID: nil, consumer: .group(3), strategy: .next, count: 1, autoCommit: true)
+ .partitionID)
+ }
+ #expect(polled == [1, 2, 1])
+ #expect(await requests(server, .syncConsumerGroup).count == 1)
+ // Each partition has its own route.
+ #expect(await requests(server, .getPollRouting).count == 2)
+ await client.shutdown()
+ }
+
+ @Test func aFencedGroupPollSyncsAndPollsAgain() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ let fenced = Locked(false)
+ await server.respond { request in
+ switch request.code.flatMap(CommandCode.init(rawValue:)) {
+ case .syncConsumerGroup:
+ return .ok(FakeVSRServer.assignmentBody(generation: fenced.withLock { $0 } ? 2 : 1, partitions: [1]))
+ case .pollMessagesOnPrimary:
+ let first = !fenced.exchange(true)
+ return .ok(FakeVSRServer.polledBody(partitionID: first ? PolledMessages.resyncRequiredPartition : 1))
+ default:
+ return FakeVSRServer.defaultReply(request, address: server.address)
+ }
+ }
+ let client = try await connectedClient(server)
+ let polled = try await client.pollMessages(streamID: 1, topicID: 2, partitionID: nil, consumer: .group(3), strategy: .next, count: 1, autoCommit: true)
+ #expect(polled.partitionID == 1)
+ #expect(await requests(server, .syncConsumerGroup).count == 2)
+ #expect(await requests(server, .pollMessagesOnPrimary).count == 2)
+ await client.shutdown()
+ }
+
+ @Test func aGroupPollTellsMembersFromNonMembers() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ let member = Locked(true)
+ await server.respond { request in
+ if request.code == CommandCode.syncConsumerGroup.rawValue {
+ return .ok(member.withLock { $0 } ? FakeVSRServer.assignmentBody(generation: 1, partitions: []) : [])
+ }
+ return FakeVSRServer.defaultReply(request, address: server.address)
+ }
+ let client = try await connectedClient(server)
+ let poll = {
+ try await client.pollMessages(streamID: 1, topicID: 2, partitionID: nil, consumer: .group(3), strategy: .next, count: 1, autoCommit: true)
+ }
+ #expect(try await poll().partitionID == PolledMessages.noAssignedPartition)
+ // The member was removed, say by an eviction: the next sync finds
+ // the client no longer a member and the poll tells it.
+ member.withLock { $0 = false }
+ await expectCode(.consumerGroupMemberNotFound) { _ = try await poll() }
+ #expect(await requests(server, .syncConsumerGroup).count == 2)
+ #expect(await requests(server, .pollMessagesOnPrimary).isEmpty)
+ await client.shutdown()
+ }
+
+ @Test func aLostPartitionMovesTheGroupPollOn() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ let syncs = Locked(0)
+ await server.respond { request in
+ switch request.code.flatMap(CommandCode.init(rawValue:)) {
+ case .syncConsumerGroup:
+ let count = syncs.withLock {
+ $0 += 1; return $0
+ }
+ return .ok(FakeVSRServer.assignmentBody(generation: UInt64(count), partitions: count == 1 ? [1] : [2]))
+ case .pollMessagesOnPrimary where polledPartition(request) == 1:
+ return .status(IggyErrorCode.consumerGroupPartitionNotOwned.rawValue)
+ case .pollMessagesOnPrimary:
+ return .ok(FakeVSRServer.polledBody(partitionID: polledPartition(request)))
+ default:
+ return FakeVSRServer.defaultReply(request, address: server.address)
+ }
+ }
+ let client = try await connectedClient(server)
+ let polled = try await client.pollMessages(streamID: 1, topicID: 2, partitionID: nil, consumer: .group(3), strategy: .next, count: 1, autoCommit: true)
+ #expect(polled.partitionID == 2)
+
+ // A continuation of the lost partition cannot move to another one.
+ syncs.withLock { $0 = 0 }
+ client.core.groupState.invalidateAssignment(.init(stream: 1, topic: 2, group: 3))
+ let continued = PollingStrategy.offset(4, context: otherContext)
+ await expectCode(.consumerGroupPartitionNotOwned) {
+ _ = try await client.pollMessages(streamID: 1, topicID: 2, partitionID: nil, consumer: .group(3), strategy: continued, count: 1, autoCommit: true)
+ }
+ await client.shutdown()
+ }
+
+ @Test func theHeartbeatSyncsJoinedGroupsUntilTheyAreLeft() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ await server.respond { request in
+ if request.code == CommandCode.syncConsumerGroup.rawValue {
+ return .ok(FakeVSRServer.assignmentBody(generation: 1, partitions: [0]))
+ }
+ return FakeVSRServer.defaultReply(request, address: server.address)
+ }
+ let client = try await connectedClient(server, heartbeat: .milliseconds(30))
+ try await client.joinConsumerGroup(streamID: 1, topicID: 2, groupID: 3)
+ _ = try await client.pollMessages(streamID: 1, topicID: 2, partitionID: nil, consumer: .group(3), strategy: .next, count: 1, autoCommit: true)
+ #expect(await eventually { await requests(server, .syncConsumerGroup).count >= 3 })
+ try await client.leaveConsumerGroup(streamID: 1, topicID: 2, groupID: 3)
+ #expect(client.core.groupState.registeredGroups.isEmpty)
+ await client.shutdown()
+ }
+
+ // MARK: Offsets
+
+ @Test func offsetWritesTakeTheOffsetRoute() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ let client = try await connectedClient(server)
+ try await client.storeConsumerOffset(consumer: .consumer(5), streamID: 1, topicID: 2, partitionID: 3, offset: 40)
+ let position = ConsumerPosition(partitionID: 3, offset: 41, context: otherContext)
+ try await client.storeConsumerPosition(consumer: .consumer(5), streamID: 1, topicID: 2, position: position)
+ try await client.deleteConsumerOffset(consumer: .consumer(5), streamID: 1, topicID: 2, partitionID: 3)
+
+ let lookups = await requests(server, .getConsumerOffsetRouting)
+ #expect(lookups.count == 1)
+ #expect(lookups.first?.payload == (try Requests.getConsumerOffset(consumer: .consumer(5), streamID: 1, topicID: 2, partitionID: 3)))
+ let stores = await requests(server, .storeConsumerOffset)
+ #expect(stores.map(\.context) == [FakeVSRServer.routeContext, otherContext])
+ #expect(
+ stores.first?.payload
+ == (try Requests.storeConsumerOffset(consumer: .consumer(5), streamID: 1, topicID: 2, partitionID: 3, offset: 40, ack: .quorum)))
+ #expect(await requests(server, .deleteConsumerOffset).map(\.context) == [FakeVSRServer.routeContext])
+ await client.shutdown()
+ }
+
+ // MARK: Raw requests
+
+ @Test func rawPartitionRequestsAreRoutedLikeTypedOnes() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ let client = try await connectedClient(server)
+ let outgoing = [Batch.OutgoingMessage(id: .random(), originTimestamp: 0, payload: [1], userHeaders: nil)]
+ let send = try Batch.encodeSend(streamID: 1, topicID: 2, partitioning: .partition(1), messages: outgoing)
+ _ = try await client.sendRawRequest(code: CommandCode.sendMessages.rawValue, payload: send)
+ #expect(await requests(server, .getPartitionRoute).count == 1)
+ #expect(await requests(server, .sendMessages).map(\.context) == [FakeVSRServer.routeContext])
+
+ let poll = try Requests.pollMessages(consumer: .consumer(5), streamID: 1, topicID: 2, partitionID: 1, strategy: .first, count: 1, autoCommit: false)
+ _ = try await client.sendRawRequest(code: CommandCode.pollMessages.rawValue, payload: poll)
+ #expect(await requests(server, .pollMessagesOnPrimary).map(\.context) == [FakeVSRServer.routeContext])
+
+ let store = try Requests.storeConsumerOffset(consumer: .consumer(5), streamID: 1, topicID: 2, partitionID: 1, offset: 9, ack: .quorum)
+ _ = try await client.sendRawRequest(code: CommandCode.storeConsumerOffset.rawValue, payload: store)
+ #expect(
+ await requests(server, .getConsumerOffsetRouting).map(\.payload)
+ == [try Requests.getConsumerOffset(consumer: .consumer(5), streamID: 1, topicID: 2, partitionID: 1)])
+
+ let before = await server.requests.count
+ let balanced = try Batch.encodeSend(streamID: 1, topicID: 2, partitioning: .balanced, messages: outgoing)
+ await expectCode(.featureUnavailable) { _ = try await client.sendRawRequest(code: CommandCode.sendMessages.rawValue, payload: balanced) }
+ // The count is outside the declared metadata, then the partition id is
+ // three bytes wide: neither is a header the server would accept.
+ var shortMetadata = send
+ shortMetadata[0] -= 4
+ var narrowPartition = send
+ narrowPartition[17] = 3
+ for malformed in [shortMetadata, narrowPartition] {
+ await expectCode(.invalidCommand) { _ = try await client.sendRawRequest(code: CommandCode.sendMessages.rawValue, payload: malformed) }
+ }
+ await expectCode(.invalidCommand) { _ = try await client.sendRawRequest(code: CommandCode.pollMessages.rawValue, payload: Array(poll.dropLast())) }
+ await expectCode(.invalidCommand) { _ = try await client.sendRawRequest(code: CommandCode.storeConsumerOffset.rawValue, payload: [1, 9]) }
+ await expectCode(.invalidCommand) { _ = try await client.sendRawRequest(code: CommandCode.bindSession.rawValue, payload: []) }
+ #expect(await server.requests.count == before)
+ await client.shutdown()
+ }
+
+ /// The routing protocol goes once on the current connection, as the Rust
+ /// client sends it: a replay after a reconnect would run under another
+ /// connection's binding.
+ @Test func rawRoutingRequestsAreSentOnce() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ let dropped = Locked(false)
+ await server.respond { request in
+ if request.code == CommandCode.getPollRouting.rawValue, !dropped.exchange(true) {
+ return .close
+ }
+ return FakeVSRServer.defaultReply(request, address: server.address)
+ }
+ let client = try await connectedClient(server)
+ let poll = try Requests.pollMessages(consumer: .consumer(5), streamID: 1, topicID: 2, partitionID: 1, strategy: .first, count: 1, autoCommit: false)
+ await expectCode(.disconnected) { _ = try await client.sendRawRequest(code: CommandCode.getPollRouting.rawValue, payload: poll) }
+ #expect(await requests(server, .getPollRouting).count == 1)
+ let body = try await client.sendRawRequest(code: CommandCode.getPollRouting.rawValue, payload: poll)
+ #expect(try Responses.pollRoute(body[...]).context == FakeVSRServer.routeContext)
+ await client.shutdown()
+ }
+
+ // MARK: Clusters
+
+ /// A coordinator that routes every poll and offset write to `primary`.
+ private static func coordinator(routingTo primary: FakeVSRServer) async throws -> FakeVSRServer {
+ let coordinator = try await FakeVSRServer.started()
+ await coordinator.respond { request in
+ switch request.code.flatMap(CommandCode.init(rawValue:)) {
+ case .getClusterMetadata:
+ return .ok(FakeVSRServer.rosterBody([coordinator.address, primary.address]))
+ case .getPollRouting, .getConsumerOffsetRouting:
+ return .ok(FakeVSRServer.pollRouteBody(for: request, context: FakeVSRServer.routeContext, primary: primary.address, watermark: 3))
+ default:
+ return FakeVSRServer.defaultReply(request, address: coordinator.address)
+ }
+ }
+ return coordinator
+ }
+
+ @Test func aClusterPollGoesToThePrimaryOverABoundDataConnection() async throws {
+ let primary = try await FakeVSRServer.started()
+ let coordinator = try await Self.coordinator(routingTo: primary)
+ defer {
+ Task {
+ await coordinator.stop()
+ await primary.stop()
+ }
+ }
+ let client = try await connectedClient(coordinator)
+ _ = try await client.pollMessages(streamID: 1, topicID: 2, partitionID: 0, strategy: .first, count: 1, autoCommit: false)
+ _ = try await client.pollMessages(streamID: 1, topicID: 2, partitionID: 0, strategy: .first, count: 1, autoCommit: false)
+ try await client.storeConsumerOffset(streamID: 1, topicID: 2, partitionID: 0, offset: 1)
+
+ let session = await coordinator.requests.last!
+ let binds = await requests(primary, .bindSession)
+ #expect(binds.count == 1)
+ #expect(
+ binds.first?.payload
+ == (try Requests.bindSession(
+ clientID: session.clientID, session: session.session, metadataWatermark: 3, bindSecret: await client.core.bindSecret)))
+ let polls = await requests(primary, .pollMessagesOnPrimary)
+ #expect(polls.count == 2)
+ #expect(polls.allSatisfy { $0.context == FakeVSRServer.routeContext && $0.clientID == session.clientID && $0.session == session.session })
+ #expect(await requests(primary, .storeConsumerOffset).count == 1)
+ #expect(await requests(coordinator, .pollMessagesOnPrimary).isEmpty)
+ #expect(await primary.connectionCount == 1)
+ await client.shutdown()
+ }
+
+ @Test func aDataConnectionLostMidRequestLeavesTheOutcomeUnknown() async throws {
+ let primary = try await FakeVSRServer.started()
+ let coordinator = try await Self.coordinator(routingTo: primary)
+ defer {
+ Task {
+ await coordinator.stop()
+ await primary.stop()
+ }
+ }
+ let dropped = Locked(false)
+ await primary.respond { request in
+ if request.code == CommandCode.pollMessagesOnPrimary.rawValue,
+ !dropped.exchange(true)
+ {
+ return .close
+ }
+ return FakeVSRServer.defaultReply(request, address: primary.address)
+ }
+ let client = try await connectedClient(coordinator)
+ let poll = { _ = try await client.pollMessages(streamID: 1, topicID: 2, partitionID: 0, strategy: .first, count: 1, autoCommit: false) }
+ await expectCode(.transientNotCommitted) { try await poll() }
+ try await poll()
+ // The next poll dials the primary again and binds the new connection.
+ #expect(await primary.connectionCount == 2)
+ #expect(await requests(primary, .bindSession).count == 2)
+ await client.shutdown()
+ }
+
+ /// A backup answers "not primary" after a view change. The poll was never
+ /// admitted, so it is routed again instead of failing.
+ @Test func aPrimaryThatIsNoLongerPrimaryIsRoutedAgain() async throws {
+ let primary = try await FakeVSRServer.started()
+ let coordinator = try await Self.coordinator(routingTo: primary)
+ defer {
+ Task {
+ await coordinator.stop()
+ await primary.stop()
+ }
+ }
+ let refused = Locked(false)
+ await primary.respond { request in
+ if request.code == CommandCode.pollMessagesOnPrimary.rawValue, !refused.exchange(true) {
+ return .status(IggyErrorCode.notPrimary.rawValue)
+ }
+ return FakeVSRServer.defaultReply(request, address: primary.address)
+ }
+ let client = try await connectedClient(coordinator)
+ _ = try await client.pollMessages(streamID: 1, topicID: 2, partitionID: 0, strategy: .first, count: 1, autoCommit: false)
+ #expect(await requests(coordinator, .getPollRouting).count == 2)
+ #expect(await requests(primary, .pollMessagesOnPrimary).count == 2)
+ // The refusal left the data connection at a frame boundary.
+ #expect(await primary.connectionCount == 1)
+ await client.shutdown()
+ }
+
+ @Test func anUnreachablePrimaryIsRoutedAgain() async throws {
+ let primary = try await FakeVSRServer.started()
+ let gone = try await FakeVSRServer.started()
+ let goneAddress = gone.address
+ await gone.stop()
+ let coordinator = try await FakeVSRServer.started()
+ defer {
+ Task {
+ await coordinator.stop()
+ await primary.stop()
+ }
+ }
+ let lookups = Locked(0)
+ await coordinator.respond { request in
+ switch request.code.flatMap(CommandCode.init(rawValue:)) {
+ case .getClusterMetadata:
+ return .ok(FakeVSRServer.rosterBody([coordinator.address, primary.address]))
+ case .getPollRouting:
+ let first = lookups.withLock {
+ $0 += 1; return $0 == 1
+ }
+ return .ok(FakeVSRServer.pollRouteBody(for: request, context: FakeVSRServer.routeContext, primary: first ? goneAddress : primary.address))
+ default:
+ return FakeVSRServer.defaultReply(request, address: coordinator.address)
+ }
+ }
+ let client = try await connectedClient(coordinator)
+ _ = try await client.pollMessages(streamID: 1, topicID: 2, partitionID: 0, strategy: .first, count: 1, autoCommit: false)
+ #expect(lookups.withLock { $0 } == 2)
+ #expect(await requests(primary, .pollMessagesOnPrimary).count == 1)
+ await client.shutdown()
+ }
+}
diff --git a/foreign/swift/Tests/IggyTests/ModelTests.swift b/foreign/swift/Tests/IggyTests/ModelTests.swift
index 225d27bbd..08b09dc0f 100644
--- a/foreign/swift/Tests/IggyTests/ModelTests.swift
+++ b/foreign/swift/Tests/IggyTests/ModelTests.swift
@@ -206,6 +206,20 @@ struct IdentifierTests {
try Identifier(numeric: UInt32.max).encode(into: &writer)
#expect(writer.bytes == [1, 4, 0xFF, 0xFF, 0xFF, 0xFF])
}
+
+ @Test func decodingReadsWhatEncodingWrites() throws {
+ for identifier: Identifier in [0, 7, "café", Identifier(numeric: UInt32.max)] {
+ var writer = ByteWriter()
+ try identifier.encode(into: &writer)
+ var reader = ByteReader(writer.bytes[...])
+ #expect(try Identifier.decode(from: &reader) == identifier)
+ #expect(reader.isAtEnd)
+ }
+ for malformed: [UInt8] in [[1, 2, 7, 0], [3, 4, 0, 0, 0, 0], [2, 0], [1, 4, 0]] {
+ var reader = ByteReader(malformed[...])
+ #expect(throws: (any Error).self) { try Identifier.decode(from: &reader) }
+ }
+ }
}
@Suite("Messages")
@@ -293,10 +307,23 @@ struct PartitioningTests {
#expect(try encode(.partition(9)) == [2, 4, 9, 0, 0, 0])
}
+ @Test func decodingReadsWhatEncodingWrites() throws {
+ for partitioning: Partitioning in [.balanced, .partition(9), .messagesKey("user-1")] {
+ var reader = ByteReader(try encode(partitioning)[...])
+ #expect(try Partitioning.decode(from: &reader) == partitioning)
+ #expect(reader.isAtEnd)
+ }
+ for malformed: [UInt8] in [[1, 4, 0, 0, 0, 0], [2, 3, 9, 0, 0], [2, 4, 9], [3, 0], [4, 0]] {
+ var reader = ByteReader(malformed[...])
+ #expect(throws: (any Error).self) { try Partitioning.decode(from: &reader) }
+ }
+ }
+
/// The poll sentinel is compared against the value the Rust generator
/// dumped from `core/common`, so the golden lane catches drift.
@Test func noAssignedPartitionMatchesTheServerConstant() {
#expect(PolledMessages.noAssignedPartition == GoldenFixture.shared.noAssignedPartition)
+ #expect(PolledMessages.resyncRequiredPartition == GoldenFixture.shared.resyncRequiredPartition)
#expect(GoldenFixture.shared.resyncRequiredPartition != GoldenFixture.shared.noAssignedPartition)
}
diff --git a/foreign/swift/Tests/IggyTests/ResponseVectorTests.swift b/foreign/swift/Tests/IggyTests/ResponseVectorTests.swift
index e8e8a31c4..8bf0fae85 100644
--- a/foreign/swift/Tests/IggyTests/ResponseVectorTests.swift
+++ b/foreign/swift/Tests/IggyTests/ResponseVectorTests.swift
@@ -195,6 +195,20 @@ struct ResponseVectorTests {
#expect(stats.totalDiskSpace == 512_110_190_592)
}
+ @Test func routes() async throws {
+ #expect(try Responses.partitionRoute(body("response.partition_route")) == PartitionContext(incarnation: 7, ownerGeneration: 0, metadataOp: 10))
+ // A route of any other length comes from a mismatched build.
+ await expectCode(.invalidCommand) { _ = try Responses.partitionRoute(body("response.partition_route") + [0]) }
+
+ let poll = try Responses.pollRoute(body("response.poll_routing"))
+ #expect(poll.consumerSession.clientID == UInt128Value(low: 0x090a_0b0c_0d0e_0f10, high: 0x0102_0304_0506_0708))
+ #expect(poll.consumerSession.session == 42)
+ #expect(poll.consumerSession.metadataWatermark == 120)
+ #expect(poll.context == PartitionContext(incarnation: 7, ownerGeneration: 3, metadataOp: 110))
+ #expect(poll.primary.replicaID == 2)
+ #expect(poll.primary.tcpAddress == "10.0.0.3:8092")
+ }
+
@Test func clusterMetadata() throws {
let metadata = try Responses.clusterMetadata(body("response.cluster_metadata"))
#expect(metadata.name == "prod-cluster")
diff --git a/foreign/swift/Tests/IggyTests/Support/FakeVSRServer.swift b/foreign/swift/Tests/IggyTests/Support/FakeVSRServer.swift
index 44c389022..8f42505e8 100644
--- a/foreign/swift/Tests/IggyTests/Support/FakeVSRServer.swift
+++ b/foreign/swift/Tests/IggyTests/Support/FakeVSRServer.swift
@@ -35,6 +35,7 @@ final class FakeVSRServer: @unchecked Sendable {
let session: UInt64
let clientID: UInt128Value
let checksum: UInt64
+ let context: PartitionContext
let payload: [UInt8]
}
@@ -43,6 +44,8 @@ final class FakeVSRServer: @unchecked Sendable {
/// Status zero with this body; result-framed operations get the
/// empty result section prepended.
case ok([UInt8])
+ /// `ok` from a metadata operation that committed at `commit`.
+ case committed([UInt8], commit: UInt64)
/// A pre-commit denial: the status in the header, no body.
case status(UInt32)
/// A committed rejection carried in the result section.
@@ -134,25 +137,91 @@ final class FakeVSRServer: @unchecked Sendable {
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)
+ writeNode(address, replicaID: UInt8(index), leader: index == leaderIndex, into: &writer)
}
return writer.bytes
}
- static func encodeReply(request: Request, operation: ConsensusOperation, status: UInt32, body: [UInt8]) -> [UInt8] {
+ static func writeNode(_ address: String, replicaID: UInt8, leader: Bool, into writer: inout ByteWriter) {
+ let endpoint = try! Endpoint(parsing: address)
+ writer.write(replicaID)
+ try! writer.writeLongString("node-\(replicaID)")
+ try! writer.writeLongString(endpoint.host)
+ writer.write(UInt16(endpoint.port))
+ writer.write(UInt16(0))
+ writer.write(UInt16(0))
+ writer.write(UInt16(0))
+ writer.write(leader ? ClusterNodeRole.leader.rawValue : ClusterNodeRole.follower.rawValue)
+ writer.write(ClusterNodeStatus.healthy.rawValue)
+ }
+
+ /// The reply to `GetPartitionRoute`.
+ static func partitionRouteBody(_ context: PartitionContext, partitionsCount: UInt32 = 3, primaryReplica: UInt8 = 0) -> [UInt8] {
+ var writer = ByteWriter()
+ writer.write(context.incarnation)
+ writer.write(context.metadataOp)
+ writer.write(UInt32(1))
+ writer.write(partitionsCount)
+ writer.write(primaryReplica)
+ return writer.bytes
+ }
+
+ /// The reply to `GetPollRouting` and `GetConsumerOffsetRouting`: the
+ /// session of `request` and a primary at `primary`.
+ static func pollRouteBody(for request: Request, context: PartitionContext, primary: String, watermark: UInt64 = 0) -> [UInt8] {
+ var writer = ByteWriter()
+ writer.write(request.clientID)
+ writer.write(request.session)
+ writer.write(watermark)
+ context.encode(into: &writer)
+ writeNode(primary, replicaID: 0, leader: true, into: &writer)
+ return writer.bytes
+ }
+
+ /// A poll reply without messages.
+ static func polledBody(partitionID: UInt32, context: PartitionContext = .empty) -> [UInt8] {
+ var writer = ByteWriter()
+ writer.write(partitionID)
+ writer.write(UInt64(0))
+ writer.write(UInt32(0))
+ context.encode(into: &writer)
+ return writer.bytes
+ }
+
+ /// The reply to `SyncConsumerGroup` for a member.
+ static func assignmentBody(generation: UInt64, partitions: [UInt32]) -> [UInt8] {
+ var writer = ByteWriter()
+ writer.write(generation)
+ writer.write(UInt32(partitions.count))
+ for partition in partitions {
+ writer.write(partition)
+ }
+ return writer.bytes
+ }
+
+ /// A topic with three partitions, 0 to 2, under `contexts`.
+ static func topicBody(contexts: [PartitionContext]) -> [UInt8] {
+ precondition(contexts.count == 3, "the golden topic header names three partitions")
+ var writer = ByteWriter()
+ writer.write(GoldenFixture.shared.bytes("response.topic_header"))
+ for (id, context) in contexts.enumerated() {
+ writer.write(UInt32(id))
+ writer.write(UInt64(0))
+ writer.write(UInt32(1))
+ writer.write(UInt64(0))
+ writer.write(UInt64(0))
+ writer.write(UInt64(0))
+ context.encode(into: &writer)
+ }
+ return writer.bytes
+ }
+
+ static func encodeReply(request: Request, operation: ConsensusOperation, status: UInt32, body: [UInt8], commit: UInt64 = 0) -> [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: commit.littleEndian, toByteOffset: VSRFrame.ReplyOffset.commit, as: UInt64.self)
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)
@@ -190,17 +259,47 @@ final class FakeVSRServer: @unchecked Sendable {
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)
+ checksum: u64(VSRFrame.RequestOffset.requestChecksum),
+ context: PartitionContext(
+ incarnation: u64(VSRFrame.RequestOffset.partitionIncarnation), ownerGeneration: u64(VSRFrame.RequestOffset.ownerGeneration),
+ metadataOp: u64(VSRFrame.RequestOffset.minimumMetadataOp)),
+ payload: frame.body)
+ }
+
+ /// The partition context the default script routes every partition to.
+ static let routeContext = PartitionContext(incarnation: 7, ownerGeneration: 0, metadataOp: 10)
+
+ /// The answer to a register and to a resume: the session is bound, as on
+ /// a node that holds it. Nil for any other request.
+ static func sessionReply(_ request: Request) -> Reply? {
+ if request.operation == .register {
+ return .ok(registerBody())
+ }
+ if request.code == CommandCode.bindSession.rawValue {
+ return .ok(registerBody(session: request.session))
+ }
+ return nil
}
/// Answers every request successfully with an empty body, a bound
- /// session for a register, and a single-node roster for the metadata read.
+ /// session for a register and a bind, a single-node roster for the
+ /// metadata read, routes to this server under ``routeContext``, and an
+ /// empty poll.
static func defaultReply(_ request: Request, address: String) -> Reply {
- switch request.operation {
- case .register:
+ if request.operation == .register {
return .ok(registerBody())
- case .nonReplicated where request.code == CommandCode.getClusterMetadata.rawValue:
+ }
+ switch request.code.flatMap(CommandCode.init(rawValue:)) {
+ case .getClusterMetadata:
return .ok(rosterBody([address]))
+ case .bindSession:
+ return .ok(registerBody(session: request.session))
+ case .getPartitionRoute:
+ return .ok(partitionRouteBody(routeContext))
+ case .getPollRouting, .getConsumerOffsetRouting:
+ return .ok(pollRouteBody(for: request, context: routeContext, primary: address))
+ case .pollMessagesOnPrimary:
+ return .ok(polledBody(partitionID: 0, context: routeContext))
default:
return .ok([])
}
@@ -290,6 +389,8 @@ final class FakeServerHandler: ChannelInboundHandler, @unchecked Sendable {
switch reply {
case .ok(let body):
write(FakeVSRServer.encodeReply(request: request, operation: operation, status: 0, body: framedBody(body)))
+ case .committed(let body, let commit):
+ write(FakeVSRServer.encodeReply(request: request, operation: operation, status: 0, body: framedBody(body), commit: commit))
case .status(let status):
write(FakeVSRServer.encodeReply(request: request, operation: operation, status: status, body: []))
case .committedRejection(let code):
@@ -303,7 +404,7 @@ final class FakeServerHandler: ChannelInboundHandler, @unchecked Sendable {
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)
+ clientID: request.clientID, checksum: request.checksum, context: request.context, payload: request.payload)
write(FakeVSRServer.encodeReply(request: wrong, operation: operation, status: 0, body: framedBody([])))
case .close:
channel.close(promise: nil)
diff --git a/foreign/swift/Tests/IggyTests/Support/TestHelpers.swift b/foreign/swift/Tests/IggyTests/Support/TestHelpers.swift
index 50c4a01d1..1dd57cf10 100644
--- a/foreign/swift/Tests/IggyTests/Support/TestHelpers.swift
+++ b/foreign/swift/Tests/IggyTests/Support/TestHelpers.swift
@@ -37,6 +37,15 @@ final class Locked<Value>: @unchecked Sendable {
defer { lock.unlock() }
return try body(&value)
}
+
+ /// Stores `newValue` and returns the value it replaced.
+ func exchange(_ newValue: Value) -> Value {
+ withLock { value in
+ let previous = value
+ value = newValue
+ return previous
+ }
+ }
}
/// Polls `condition` until it holds or `timeout` passes.
@@ -62,3 +71,15 @@ func expectCode(_ code: IggyErrorCode, sourceLocation: SourceLocation = #_source
Issue.record("expected \(code), got \(error)", sourceLocation: sourceLocation)
}
}
+
+extension IggyClient {
+ /// ``sendMessages(streamID:topicID:partitioning:messages:)`` for a batch
+ /// the test does not read back, so the call can take a literal.
+ @discardableResult
+ func sendBatch(
+ streamID: Identifier, topicID: Identifier, partitioning: Partitioning = .balanced, messages: [IggyMessage]
+ ) async throws -> SendMessagesResponse {
+ var batch = messages
+ return try await sendMessages(streamID: streamID, topicID: topicID, partitioning: partitioning, messages: &batch)
+ }
+}
diff --git a/foreign/swift/Tests/IggyTests/WireVectorTests.swift b/foreign/swift/Tests/IggyTests/WireVectorTests.swift
index f0fe2bea0..5729fedfc 100644
--- a/foreign/swift/Tests/IggyTests/WireVectorTests.swift
+++ b/foreign/swift/Tests/IggyTests/WireVectorTests.swift
@@ -113,6 +113,7 @@ struct RequestVectorTests {
#expect(
try Requests.pollMessages(consumer: .group(3), streamID: 1, topicID: 1, partitionID: nil, strategy: .first, count: 10, autoCommit: false)
== golden.bytes("request.poll_messages.group"))
+ #expect(try Requests.partitionRoute(streamID: 10, topicID: "orders", partitionID: 5) == golden.bytes("request.get_partition_route"))
let headers: UserHeaders = ["trace-id": "abc", "attempt": .uint32(3)]
let messages = [
@@ -264,6 +265,7 @@ struct VSRHeaderVectorTests {
#expect(VSRFrame.peekCommand(ok) == .reply)
#expect(VSRFrame.readSize(ok) == UInt32(VSRFrame.headerSize + 12))
#expect(VSRFrame.readReplyRequestID(ok) == 7)
+ #expect(VSRFrame.readReplyCommit(ok) == 12)
#expect(VSRFrame.readReplyStatus(ok) == 0)
#expect(VSRFrame.readReplyOperation(ok) == .createStream)
diff --git a/foreign/swift/Tools/golden-vectors/src/main.rs b/foreign/swift/Tools/golden-vectors/src/main.rs
index a77c4c12f..3b2f15d7e 100644
--- a/foreign/swift/Tools/golden-vectors/src/main.rs
+++ b/foreign/swift/Tools/golden-vectors/src/main.rs
@@ -44,7 +44,7 @@ use iggy_binary_protocol::primitives::user_headers::encode_user_headers;
use iggy_binary_protocol::requests::consumer_groups::*;
use iggy_binary_protocol::requests::consumer_offsets::*;
use iggy_binary_protocol::requests::messages::{
- PollMessagesRequest, RawMessage, SendMessagesEncoder,
+ GetPartitionRouteRequest, PollMessagesRequest, RawMessage, SendMessagesEncoder,
};
use iggy_binary_protocol::requests::partitions::{
CreatePartitionsRequest, DeletePartitionsRequest,
@@ -67,7 +67,8 @@ use iggy_binary_protocol::responses::consumer_groups::{
};
use iggy_binary_protocol::responses::consumer_offsets::ConsumerOffsetResponse;
use iggy_binary_protocol::responses::messages::{
- PollMessagesResponseHeader, SendMessagesConfirmationResponse, SendMessagesResponse,
+ PartitionRouteResponse, PollMessagesResponseHeader, PollRoutingResponse,
+ SendMessagesConfirmationResponse, SendMessagesResponse,
};
use iggy_binary_protocol::responses::personal_access_tokens::{
GetPersonalAccessTokensResponse, PersonalAccessTokenResponse, RawPersonalAccessTokenResponse,
@@ -625,6 +626,15 @@ fn main() {
auto_commit: false,
},
);
+ add(
+ &mut vectors,
+ "request.get_partition_route",
+ &GetPartitionRouteRequest {
+ stream_id: numeric(10),
+ topic_id: WireIdentifier::named("orders").unwrap(),
+ partition_id: 5,
+ },
+ );
add(
&mut vectors,
"request.create_user",
@@ -931,6 +941,7 @@ fn main() {
operation: Operation::CreateStream,
size: (HEADER_SIZE + 12) as u32,
request: 7,
+ commit: 12,
status: 0,
client: client_id,
..Default::default()
@@ -1399,6 +1410,44 @@ fn main() {
.encode(&mut empty);
vectors.insert("response.poll_messages.empty".to_owned(), hex(&empty));
}
+ add(
+ &mut vectors,
+ "response.partition_route",
+ &PartitionRouteResponse {
+ incarnation: 7,
+ metadata_op: 10,
+ view: 3,
+ partitions_count: 4,
+ primary_replica: 2,
+ },
+ );
+ add(
+ &mut vectors,
+ "response.poll_routing",
+ &PollRoutingResponse {
+ consumer_session: SessionIdentity {
+ client_id: 0x0102_0304_0506_0708_090a_0b0c_0d0e_0f10,
+ session: 42,
+ metadata_watermark: 120,
+ },
+ context: PartitionContext {
+ incarnation: 7,
+ owner_generation: 3,
+ metadata_op: 110,
+ },
+ primary: ClusterNodeResponse {
+ replica_id: 2,
+ name: "node-3".to_owned(),
+ ip: "10.0.0.3".to_owned(),
+ tcp_port: 8092,
+ quic_port: 0,
+ http_port: 3002,
+ websocket_port: 0,
+ role: 1,
+ status: 0,
+ },
+ },
+ );
// Domain option blocks.
let create_options = TopicCreateOptions {