RustToMetal commented on code in PR #4553: URL: https://github.com/apache/iggy/pull/4553#discussion_r4286266396
########## 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: Fixed in d5c1dbbfa. `exchangeOnce` and the data-connection exchange now grade `notPrimary` through `ClientCore.graded`, so it takes the `transientNotAccepted` re-route on both paths, as `decode_response_split` does. New test `aPrimaryThatIsNoLongerPrimaryIsRoutedAgain`: the data connection answers `notPrimary` once, the poll reads its route again and succeeds on the same connection. ########## 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: Fixed in d5c1dbbfa. `stampRequest(code:payload:context:for:)` takes the route's session identity and checks, in the same actor call, that it is still the live session (client id and session number) and that no newer metadata commit was seen. Otherwise it throws `transientNotAccepted` before anything is written. That replaces the separate `validate` hops in `sendOnPrimary`, and an unbound session lands there too, so it is no longer reported as `transientNotCommitted`. `dataConnectionFramesAreStampedOnlyForTheLiveSession` covers the live, older-watermark, unbound, and replaced cases. The doc comment now says reads carry the counter's current value instead of claiming unique ids. ########## 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: Fixed in d5c1dbbfa. `PollRouter.send` no longer runs under `withDeadline`. The 30s budget is checked between attempts and passed down as a cap: `exchangeOnce(deadline:)`, `exchange(deadline:)`, and the data-connection timeout use `min(requestTimeout, remaining)`, and the roster read uses `min(budget, now + 5s)` the same way, so nothing cancels an exchange in flight on the coordinator connection. While stress-testing this I found the same problem in the retry loop from the transport PR: a replay could go out with a sliver of the request budget left, time out, and take the connection (and report `transientNotCommitted` for a refused request). A retry is now not sent with less than one retry interval left; the refusal stands instead. ########## 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: Fixed in d5c1dbbfa by resuming the session. A transport loss keeps the session, and the next connection sends `BindSession` with the same client id, session, and bind secret, as `resume_vsr_session` does, so memberships, routes, and request ids carry over. `Unauthenticated` or `TransientNotCommitted` on the bind resets the session and signs in afresh, as in Rust; other failures move on to the next endpoint. A request lost with its connection is replayed with its original stamp when the session was resumed (the `retain_replay_header` rule), and only operations that provably never reached the log are restamped under a new session. An explicit `disconnect()` still ends the session, like Rust's. I left the NotPrimary failover itself as it is, since the Rust send path takes the same move; with the session resumed it no longer costs the membership. Tests: `aReconnectResumesTheSessionAndKeepsTheMembership`, `aSessionTheServerLostEndsTheMembership`, `replicatedRequestIsReplayedUnderTheResumedSession`, `replicatedRequestIsNotReplayedUnderANewSession`, `anExplicitDisconnectIsNotResumed`. I also checked it against a local two-node cluster, entering through each node: after the transport was dropped the member kept polling without joining again and the group still had one member. ########## .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: Added in d5c1dbbfa. The e2e lane now starts both replicas of the default roster, as the Go lane does, and runs the suite against the cluster after the plain and TLS runs. Locally all 29 tests pass against a two-node cluster built from this branch. ########## 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: Done in d5c1dbbfa: `sendMessages` takes `messages: inout [IggyMessage]` and writes the minted ids back, as `&mut [IggyMessage]` does in Rust. I picked `inout` over returning them in `SendMessagesResponse` because the case you describe ends in a thrown `transientNotCommitted`, which returns no response, so the caller would never see them. Test: `mintedIdsAreWrittenBackAndSentAgainUnchanged`. -- 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]
