numinnex commented on code in PR #4553:
URL: https://github.com/apache/iggy/pull/4553#discussion_r4281372326
##########
foreign/swift/Sources/Iggy/Client/ClientCore.swift:
##########
@@ -196,10 +209,48 @@ actor ClientCore {
logger.warning("Reconnected, but the outcome of command
\(code) is unknown: replaying it could apply it twice")
throw failure
}
- return try await sendRaw(code: code, payload: payload, deadline:
deadline, epoch: &failedEpoch, outcome: &outcome)
+ return try await sendRaw(code: code, payload: payload, context:
context, deadline: deadline, 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.
+ func exchangeOnce(code: UInt32, payload: [UInt8], context:
PartitionContext) 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: now + configuration.requestTimeout,
replaysUncommitted: false)
+ switch attempted {
+ case .success(let body):
+ return body
+ case .failure(let error):
+ dropConnection(after: error, epoch: epoch)
+ throw error
}
}
+ var bindSecret: [UInt8] { session.bindSecret }
+
+ /// Stamps a frame from the shared session for a data connection, so the
+ /// request ids stay unique across every connection of the session.
Review Comment:
`stampRequest` stamps from whatever session is live and takes no fence. A
reset between `validate` and the stamp (an actor hop apart) leads to one of two
outcomes:
- unbound: `.unauthenticated` is reconnectable, so PollRouter reports
`transientNotCommitted` for a frame that was never written
- new session already bound: the server overwrites client/session from the
connection binding (dispatch/partition.rs:267-285), so the fresh id lands in
the old session's dedup slice and can hit a cached reply. Store reports
success, nothing applied.
Suggest `stampRequest(fence:)` that throws `transientNotAccepted` on
mismatch or unbound. Doing the watermark read in the same call also saves the
extra actor hops. Also, non-replicated frames reuse `currentRequestID`, so the
uniqueness claim in this comment doesn't hold for reads.
##########
foreign/swift/Sources/Iggy/Client/PollRouter.swift:
##########
@@ -0,0 +1,370 @@
+// 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.
+ func send(
+ _ code: CommandCode, payload: [UInt8], key: RouteKey, routePayload:
[UInt8], context: PartitionContext?, coordinator: ClientCore
+ ) async throws -> [UInt8] {
+ let clustered = try await isClustered(coordinator)
+ let clock = ContinuousClock()
+ let now = clock.now
+ let deadline = now + Self.pollTimeout
+ let routed: [UInt8]? = try await withDeadline(deadline) { [self] in
Review Comment:
`withDeadline` cancels the body. If the cancel lands while `control()` /
`coordinator.exchange` already wrote to the main connection, `TCPConnection`
aborts the channel. The reconnect then resets the session and drops group
membership.
Reachable via the 30s poll deadline (== default requestTimeout, and it
starts before the route lookup) and the 5s roster cap at :151. Better to check
the deadline between attempts and clamp `exchangeOnce`'s readDeadline than to
cancel an in-flight main exchange.
##########
foreign/swift/Sources/Iggy/Client/PollRouter.swift:
##########
@@ -0,0 +1,370 @@
+// 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.
+ func send(
+ _ code: CommandCode, payload: [UInt8], key: RouteKey, routePayload:
[UInt8], context: PartitionContext?, coordinator: ClientCore
+ ) async throws -> [UInt8] {
+ let clustered = try await isClustered(coordinator)
+ let clock = ContinuousClock()
+ let now = clock.now
+ let deadline = now + Self.pollTimeout
+ let routed: [UInt8]? = try await withDeadline(deadline) { [self] in
+ 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)
+ }
+ var captured = context
+ var interval = Self.retryInterval
+ while true {
+ do {
+ if clustered {
+ return try await sendOnPrimary(
+ code, payload: payload, key: key, routePayload:
routePayload, context: &captured, coordinator: coordinator)
+ }
+ return try await sendOnCoordinator(
+ code, payload: payload, key: key, routePayload:
routePayload, 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)
+ }
+ }
+ }
+ guard let routed else {
+ removeRoute(key)
+ throw IggyError(.transientNotCommitted, context: "command
\(code.rawValue) did not complete within \(Self.pollTimeout)")
+ }
+ return routed
+ }
+
+ /// Whether the roster has more than one node, reading it once when no
+ /// leader check reported it yet.
+ private func isClustered(_ coordinator: ClientCore) async throws -> Bool {
+ if state.withLockedValue({ $0.rosterSize }) == 0 {
+ let body: [UInt8]? = try await withDeadline(ContinuousClock.now +
Self.rosterReadTimeout) { [self] in
+ try await control(.getClusterMetadata, [], coordinator:
coordinator)
+ }
+ guard let body else {
+ throw IggyError(.transientNotAccepted, context: "reading the
cluster roster took longer than \(Self.rosterReadTimeout)")
+ }
+ 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) async throws -> [UInt8] {
+ do {
+ return try await coordinator.exchangeOnce(code: code.rawValue,
payload: payload, context: .empty)
+ } catch let error as IggyError where error.isReconnectable {
+ _ = try await coordinator.exchange(code:
CommandCode.ping.rawValue, payload: [])
+ return try await coordinator.exchangeOnce(code: code.rawValue,
payload: payload, context: .empty)
+ }
+ }
+
+ /// The single-node path: the request goes over the coordinator connection
+ /// with the context of its route.
+ private func sendOnCoordinator(
+ _ code: CommandCode, payload: [UInt8], key: RouteKey, routePayload:
[UInt8], context: inout PartitionContext?, coordinator: ClientCore
+ ) async throws -> [UInt8] {
+ let route = try await route(key, routePayload: routePayload,
coordinator: coordinator)
+ let captured = context ?? route.context
+ context = captured
+ do {
+ if code == .pollMessagesOnPrimary {
+ return try await coordinator.exchangeOnce(code: code.rawValue,
payload: payload, context: captured)
+ }
+ return try await coordinator.exchange(code: code.rawValue,
payload: payload, context: captured)
+ } catch {
+ removeRoute(key)
+ throw error
+ }
+ }
+
+ /// The cluster path: the request goes over a data connection to the
+ /// partition primary, bound to the coordinator's session first.
+ private func sendOnPrimary(
+ _ code: CommandCode, payload: [UInt8], key: RouteKey, routePayload:
[UInt8], context: inout PartitionContext?, coordinator: ClientCore
+ ) async throws -> [UInt8] {
+ let route = try await route(key, routePayload: routePayload,
coordinator: coordinator)
+ let captured = context ?? route.context
+ context = captured
+ try await validate(route, coordinator: coordinator)
+ let slot = try slot(for: route)
+ try await slot.lock.lock()
+ defer { slot.lock.unlock() }
+ try await validate(route, coordinator: coordinator)
+
+ 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)")
+ }
+ try await validate(route, coordinator: coordinator)
+ 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)
+ } 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 }
+ }
+ try await validate(route, coordinator: coordinator)
+
+ do {
+ let body = try await exchange(on: data.connection, code:
code.rawValue, payload: payload, context: captured, coordinator: coordinator)
+ 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(key)
+ if error.isReconnectable {
+ // The request was written, so it may have been applied.
+ throw IggyError(.transientNotCommitted, context: "the data
connection to \(route.endpoint) failed: \(error)")
+ }
+ throw 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) 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,
coordinator: coordinator)
+ 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, with no replay.
+ private func exchange(
+ on connection: any IggyConnection, code: UInt32, payload: [UInt8],
context: PartitionContext, coordinator: ClientCore
+ ) async throws -> [UInt8] {
+ let request = try await coordinator.stampRequest(code: code, payload:
payload, context: context)
+ let raw = try await connection.exchange(request.bytes, timeout:
coordinator.configuration.requestTimeout)
+ 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")
+ }
+ return Array(try VSRFrame.decodeReply(header: header, body:
raw.body[...]))
Review Comment:
`decodeReply` throws `.notPrimary` raw here, and `exchangeOnce` doesn't
grade it either. `notPrimary` isn't reconnectable, so it escapes the
`.transientNotAccepted` retry in `send` and the caller gets code 89 during
partition failover instead of a re-route.
Rust maps NotPrimary to TransientNotAccepted in `decode_response_split`
(core/sdk/src/vsr.rs:325-336) for the poll client. Can we apply
`ClientCore.graded` on both paths and add a test where the data connection
answers notPrimary?
##########
foreign/swift/Sources/Iggy/Client/ConsumerGroupState.swift:
##########
@@ -0,0 +1,205 @@
+// 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 = 0
+ }
+
+ private struct TopicPartitions {
+ var count: UInt32?
+ var contexts: [UInt32: PartitionContext] = [:]
+ var cursor = 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> = []
+ }
+
+ 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 }
+ }
+
+ /// Replaces the group's assignment. A new generation, a rebalance,
+ /// restarts the round-robin.
+ func setAssignment(_ key: GroupKey, generation: UInt64, partitions:
[UInt32]) {
+ storage.withLockedValue { storage in
+ var assignment = storage.assignments[key] ?? Assignment()
+ if assignment.generation != generation {
+ assignment.cursor = 0
+ }
+ assignment.generation = generation
+ assignment.partitions = partitions
+ storage.assignments[key] = assignment
+ }
+ }
+
+ /// 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[assignment.cursor %
assignment.partitions.count]
+ assignment.cursor &+= 1
Review Comment:
nit: `cursor` is `Int` with `&+=`. After a wrap, `cursor % count` is
negative and traps on the index. It's unreachable in practice, but `UInt` costs
nothing.
##########
foreign/swift/Sources/Iggy/Client/IggyClient.swift:
##########
@@ -167,10 +172,23 @@ public final class IggyClient: Sendable {
/// 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()``.
+ ///
+ /// 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.
public func sendRawRequest(code: UInt32, payload: [UInt8]) async throws ->
[UInt8] {
guard !CommandCode.isSessionControl(code) else {
Review Comment:
Raw 15 / 103 / 104 aren't intercepted. They fall to `core.exchange`, which
replays non-replicated requests after a reconnect, and a raw BindSession
rebinds the coordinator connection outside PollRouter's bookkeeping. Rust
intercepts these via `is_poll_routing_code` (tcp_client.rs:240). Refuse or
route them, and consider adding `.bindSession` to `isSessionControl`.
##########
.github/actions/swift/pre-merge/action.yml:
##########
@@ -102,3 +104,84 @@ runs:
exit 1
fi
echo "✅ golden.json matches the Rust crates"
+
+ - name: Setup Rust with cache for e2e
+ if: inputs.task == 'e2e'
+ uses: ./.github/actions/utils/setup-rust-with-cache
+
+ - 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"
Review Comment:
Only replica 0 is started. Go's lane also runs a 2-node cluster
(go/pre-merge/action.yml:219-260). Data connections, BindSession,
`stampRequest` and `sendOnPrimary` are only exercised against FakeVSRServer,
which is how the notPrimary / fence / session-reset issues slipped through.
Please add a cluster e2e run.
##########
foreign/swift/Sources/Iggy/Client/ClientCore.swift:
##########
@@ -739,7 +810,7 @@ actor ClientCore {
// 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()
+ resetSession()
Review Comment:
Every transport loss mints a new session, while the Rust client resumes it
(`resume_vsr_session`, tcp_client.rs:848). With this PR `endSessionScope` also
wipes `joinedGroups` and routes, so the next group poll throws
`consumerGroupMemberNotFound`. The old session stays a ghost member until lease
expiry (session_ops.rs:368-371), stalling its partitions.
In a cluster a plain send that gets partition `NotPrimary` reaches this
through `failOver` -> `settle` -> `disconnectTransport`. At minimum a partition
NotPrimary shouldn't move the coordinator. Resuming the session is the real fix.
##########
foreign/swift/Tests/IggyE2ETests/ClientE2ETests.swift:
##########
@@ -0,0 +1,691 @@
+// 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.sendMessages(
+ 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.sendMessages(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.sendMessages(
+ 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.sendMessages(
+ streamID: scratch.stream, topicID: scratch.topic,
partitioning: .messagesKey("order-key-1"), messages: try
E2EEnvironment.messages(1))
+ let second = try await client.sendMessages(
+ 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.sendMessages(
+ 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.sendMessages(
+ 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.sendMessages(
+ 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.sendMessages(
+ 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.sendMessages(
+ 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(kind: .offset, value: 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.sendMessages(
+ 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.sendMessages(
+ 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.sendMessages(
+ 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.sendMessages(
+ 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.sendMessages(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.sendMessages(streamID: 1, topicID: 1,
partitioning: .balanced, messages: [])
+ }
+ await #expect(throws: IggyError.self) {
Review Comment:
`#expect(throws: IggyError.self)` passes on any error, including transport
or decode failures. Assert the exact code (`streamIdNotFound`).
##########
foreign/swift/Sources/Iggy/Client/IggyClient+Messages.swift:
##########
@@ -0,0 +1,403 @@
+// 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 for this send only;
+ /// `messages` is not changed. A caller that retries a failed send and
+ /// wants each attempt to carry the same ids sets them first, for example
+ /// with ``UInt128Value/random()``.
+ ///
+ /// 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: [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)
+ let outgoing = messages.map { message in
+ Batch.OutgoingMessage(
+ id: message.id.isZero ? .random() : message.id,
originTimestamp: message.originTimestamp.microseconds, payload: message.payload,
Review Comment:
Ids are minted on a copy and never returned. Rust writes them back through
`&mut [IggyMessage]` (messages.rs send_messages). A caller retrying after
`transientNotCommitted` resends under fresh ids. Return the effective ids in
`SendMessagesResponse` (or take `inout`)?
##########
foreign/swift/Sources/Iggy/Client/IggyClient+Messages.swift:
##########
@@ -0,0 +1,403 @@
+// 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 for this send only;
+ /// `messages` is not changed. A caller that retries a failed send and
+ /// wants each attempt to carry the same ids sets them first, for example
+ /// with ``UInt128Value/random()``.
+ ///
+ /// 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: [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)
+ let outgoing = messages.map { message in
+ Batch.OutgoingMessage(
+ id: message.id.isZero ? .random() : 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)
Review Comment:
No per-partition lane: while this replays id N on `transientNotCommitted`,
offset writes on data connections keep advancing the shared counter. Past the
128-id window the server answers `RequestTooOld` (client_table.rs:493-511),
which `RetryOutcome.observe` doesn't map to an uncertain outcome, so a caller
may resend a committed batch. Mapping `requestTooOld` to
`transientNotCommitted` is cheap hardening (Rust has the same gap).
##########
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
}
Review Comment:
Hard-coded and never compared to the golden value. ModelTests only asserts
the two golden sentinels differ. Add
`#expect(PolledMessages.resyncRequiredPartition ==
GoldenFixture.shared.resyncRequiredPartition)` so drift is caught like
`noAssignedPartition`.
##########
foreign/swift/README.md:
##########
@@ -100,6 +100,48 @@ 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. `pollMessages` reads from a partition, starting
where the
+strategy says:
+
+```swift
+try await client.sendMessages(streamID: "orders", topicID: "created",
partitioning: .messagesKey("customer-42"), messages: ["placed"])
+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 in `PollingStrategy.context` to
continue the
+same partition, or in a `ConsumerPosition` to store a checkpoint, and the
server refuses
+either with `historyUnavailable` once the partition was recreated.
Review Comment:
A consumer-group owner change refuses with `consumerGroupPartitionNotOwned`,
not only `historyUnavailable` (the `storeConsumerPosition` doc already says
so). Name both here.
##########
foreign/swift/Sources/Iggy/Models/Messaging.swift:
##########
@@ -144,10 +161,15 @@ 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?
Review Comment:
The README tells users to hand the poll context back, but the `.offset(_:)`
/ `.timestamp(_:)` factories take no context, so callers end up on
`init(kind:value:context:)` (as the e2e does). An `offset(_:context: = nil)`
overload would help.
##########
foreign/swift/Sources/Iggy/Client/PollRouter.swift:
##########
@@ -0,0 +1,370 @@
+// 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.
+ func send(
+ _ code: CommandCode, payload: [UInt8], key: RouteKey, routePayload:
[UInt8], context: PartitionContext?, coordinator: ClientCore
+ ) async throws -> [UInt8] {
+ let clustered = try await isClustered(coordinator)
+ let clock = ContinuousClock()
+ let now = clock.now
+ let deadline = now + Self.pollTimeout
+ let routed: [UInt8]? = try await withDeadline(deadline) { [self] in
+ 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)
+ }
+ var captured = context
+ var interval = Self.retryInterval
+ while true {
+ do {
+ if clustered {
+ return try await sendOnPrimary(
+ code, payload: payload, key: key, routePayload:
routePayload, context: &captured, coordinator: coordinator)
+ }
+ return try await sendOnCoordinator(
+ code, payload: payload, key: key, routePayload:
routePayload, 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)
+ }
+ }
+ }
+ guard let routed else {
+ removeRoute(key)
+ throw IggyError(.transientNotCommitted, context: "command
\(code.rawValue) did not complete within \(Self.pollTimeout)")
+ }
+ return routed
+ }
+
+ /// Whether the roster has more than one node, reading it once when no
+ /// leader check reported it yet.
+ private func isClustered(_ coordinator: ClientCore) async throws -> Bool {
+ if state.withLockedValue({ $0.rosterSize }) == 0 {
+ let body: [UInt8]? = try await withDeadline(ContinuousClock.now +
Self.rosterReadTimeout) { [self] in
+ try await control(.getClusterMetadata, [], coordinator:
coordinator)
+ }
+ guard let body else {
+ throw IggyError(.transientNotAccepted, context: "reading the
cluster roster took longer than \(Self.rosterReadTimeout)")
+ }
+ 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) async throws -> [UInt8] {
+ do {
+ return try await coordinator.exchangeOnce(code: code.rawValue,
payload: payload, context: .empty)
+ } catch let error as IggyError where error.isReconnectable {
+ _ = try await coordinator.exchange(code:
CommandCode.ping.rawValue, payload: [])
+ return try await coordinator.exchangeOnce(code: code.rawValue,
payload: payload, context: .empty)
+ }
+ }
+
+ /// The single-node path: the request goes over the coordinator connection
+ /// with the context of its route.
+ private func sendOnCoordinator(
+ _ code: CommandCode, payload: [UInt8], key: RouteKey, routePayload:
[UInt8], context: inout PartitionContext?, coordinator: ClientCore
+ ) async throws -> [UInt8] {
+ let route = try await route(key, routePayload: routePayload,
coordinator: coordinator)
+ let captured = context ?? route.context
+ context = captured
+ do {
+ if code == .pollMessagesOnPrimary {
Review Comment:
On a single node a failure after the poll was written surfaces as raw
`.disconnected` (autoCommit may already be applied), while the cluster path at
:266-268 maps the same case to `transientNotCommitted`. The same failure gets a
different code depending on topology. Map it the same way here.
##########
foreign/swift/README.md:
##########
@@ -100,6 +100,48 @@ 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. `pollMessages` reads from a partition, starting
where the
+strategy says:
+
+```swift
+try await client.sendMessages(streamID: "orders", topicID: "created",
partitioning: .messagesKey("customer-42"), messages: ["placed"])
+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 in `PollingStrategy.context` to
continue the
+same partition, or in a `ConsumerPosition` to store a checkpoint, and the
server refuses
+either with `historyUnavailable` once the partition was recreated.
+
+### 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, so after a
reconnect
Review Comment:
This reads as if the client rejoins by itself. It doesn't: `joinedGroups` is
cleared and the next group poll throws `consumerGroupMemberNotFound`. Say the
caller must call `joinConsumerGroup` again.
##########
foreign/swift/Sources/Iggy/Client/IggyClient+Messages.swift:
##########
@@ -0,0 +1,403 @@
+// 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 for this send only;
+ /// `messages` is not changed. A caller that retries a failed send and
+ /// wants each attempt to carry the same ids sets them first, for example
+ /// with ``UInt128Value/random()``.
+ ///
+ /// 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: [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)
+ let outgoing = messages.map { message in
+ Batch.OutgoingMessage(
+ id: message.id.isZero ? .random() : 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.
+ func syncGroupAssignment(streamID: Identifier, topicID: Identifier,
groupID: Identifier) async throws {
+ let body = try await send(.syncConsumerGroup,
Requests.consumerGroup(streamID: streamID, topicID: topicID, groupID: groupID))
+ let key = ConsumerGroupState.GroupKey(stream: streamID, topic:
topicID, group: groupID)
+ let state = core.groupState
+ 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.invalidateAssignment(key)
+ state.deregisterGroup(key)
+ return
+ }
+ let assignment = try Responses.consumerGroupAssignment(body[...])
+ state.registerGroup(key)
Review Comment:
Not fenced against a session reset: a sync reply from the old session that
lands after `endSessionScope` repopulates `registerGroup` / `setAssignment` for
the new one. It heals on the next resync, but capturing the fence before the
send and dropping a stale reply is cheap.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
