You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by hg...@apache.org on 2026/10/16 18:16:06 UTC
(iggy) 16/16: fix(swift): route polls and offsets through partition routes
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch feat/sdk-partition-routing
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 89500c468b2a779e3ea87fb609c6fbc94bd047e0
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Fri Oct 16 17:26:17 2026 +0200
fix(swift): route polls and offsets through partition routes
The Swift messaging from #4553 polled and stored offsets through the
routing commands 103, 104 and 123, which the server no longer serves.
The package did not build. The server also serves an offset read
only on the partition primary, under the context of its route.
Polls and the three offset commands now look up GetPartitionRoute
and send under its context. The reply names the primary by replica
id, so the router maps the id through the cluster roster and reads
the roster again when the id is missing. A data connection binds
the client's session at the metadata op of the route. A group poll,
a store or a delete without a partition fails with
InvalidIdentifier before anything is sent.
The golden fixture drops the poll routing vector and regenerates
the committed reply vector, which no longer matched its generator.
---
foreign/swift/Sources/Iggy/Client/ClientCore.swift | 18 +-
.../Sources/Iggy/Client/IggyClient+Messages.swift | 114 ++++++-----
foreign/swift/Sources/Iggy/Client/IggyClient.swift | 13 +-
foreign/swift/Sources/Iggy/Client/PollRouter.swift | 138 ++++++++-----
.../swift/Sources/Iggy/Protocol/Responses.swift | 43 ++--
foreign/swift/Tests/IggyTests/Fixtures/golden.json | 3 +-
foreign/swift/Tests/IggyTests/MessagingTests.swift | 220 +++++++++++++++------
.../Tests/IggyTests/ResponseVectorTests.swift | 14 +-
.../Tests/IggyTests/Support/FakeVSRServer.swift | 16 +-
foreign/swift/Tools/golden-vectors/src/main.rs | 31 +--
10 files changed, 345 insertions(+), 265 deletions(-)
diff --git a/foreign/swift/Sources/Iggy/Client/ClientCore.swift b/foreign/swift/Sources/Iggy/Client/ClientCore.swift
index 3b06742d0..19a03b39f 100644
--- a/foreign/swift/Sources/Iggy/Client/ClientCore.swift
+++ b/foreign/swift/Sources/Iggy/Client/ClientCore.swift
@@ -272,6 +272,12 @@ actor ClientCore {
var bindSecret: [UInt8] { session.bindSecret }
+ /// The bound session as a data connection binds to it, at the metadata
+ /// watermark `metadataOp`. Nil while no session is bound.
+ func sessionIdentity(at metadataOp: UInt64) -> Responses.SessionIdentity? {
+ session.session.map { Responses.SessionIdentity(clientID: session.clientID, session: $0, metadataWatermark: metadataOp) }
+ }
+
/// Stamps a frame for a data connection from the shared session, for the
/// session `identity` names. Replicated requests take the next id of the
/// session's counter; reads carry its current value. A session that was
@@ -1164,7 +1170,7 @@ actor ClientCore {
do {
let body = try await exchange(code: CommandCode.getClusterMetadata.rawValue, payload: [])
let metadata = try Responses.clusterMetadata(body[...])
- learnRosterSize(metadata)
+ pollRouter.learnRoster(metadata)
return Self.transportEndpoints(metadata)
} catch {
logger.debug("Failed to read the cluster roster: \(error)")
@@ -1172,14 +1178,6 @@ actor ClientCore {
}
}
- /// Tells the router whether polls go to data connections. An empty
- /// roster says nothing about the cluster, so the router keeps asking.
- private func learnRosterSize(_ metadata: ClusterMetadata) {
- if !metadata.nodes.isEmpty {
- pollRouter.learnRosterSize(metadata.nodes.count)
- }
- }
-
static func transportEndpoints(_ metadata: ClusterMetadata) -> [String] {
metadata.nodes.filter { $0.endpoints.tcp != 0 }.map(\.tcpAddress)
}
@@ -1224,7 +1222,7 @@ actor ClientCore {
logger.debug("Failed to get cluster metadata: \(error), continuing on \(currentAddress)")
return LeaderCheck(redirect: nil, endpoints: [])
}
- learnRosterSize(metadata)
+ pollRouter.learnRoster(metadata)
let endpoints = Self.transportEndpoints(metadata)
if metadata.nodes.count <= 1 {
return LeaderCheck(redirect: nil, endpoints: endpoints)
diff --git a/foreign/swift/Sources/Iggy/Client/IggyClient+Messages.swift b/foreign/swift/Sources/Iggy/Client/IggyClient+Messages.swift
index 903b53b9f..2dcb756d6 100644
--- a/foreign/swift/Sources/Iggy/Client/IggyClient+Messages.swift
+++ b/foreign/swift/Sources/Iggy/Client/IggyClient+Messages.swift
@@ -95,7 +95,7 @@ extension IggyClient {
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)[...])
+ return try Responses.polledMessages(try await routedPartitionRequest(.pollMessages, payload: payload, context: strategy.context)[...])
}
// MARK: Partition contexts
@@ -114,7 +114,7 @@ extension IggyClient {
captured = context
} else {
let body = try await send(.getPartitionRoute, Requests.partitionRoute(streamID: streamID, topicID: topicID, partitionID: partitionID))
- captured = try Responses.partitionRoute(body[...])
+ captured = try Responses.partitionRoute(body[...]).context
state.setPartitionContext(key, partitionID: partitionID, captured)
}
do {
@@ -130,18 +130,22 @@ extension IggyClient {
}
}
- /// 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)
+ /// A poll or a consumer-offset request on the route of the partition its
+ /// payload names, as `OperationDriver::send_raw` in the Rust SDK picks
+ /// it. A plain consumer's poll and an offset read without a partition go
+ /// to partition 0. A group poll, a store, and a delete must name one.
+ func routedPartitionRequest(_ code: CommandCode, payload: [UInt8], context: PartitionContext?) async throws -> [UInt8] {
+ let target = try RawRequest.consumerTarget(payload)
+ let partitionID: UInt32
+ if let named = target.partitionID {
+ partitionID = named
+ } else if code == .getConsumerOffset || (code == .pollMessages && target.consumerKind == Consumer.Kind.consumer.rawValue) {
+ partitionID = 0
+ } else {
+ throw IggyError(.invalidIdentifier, context: "command \(code.rawValue) needs a partition id")
+ }
+ let key = PollRouter.RouteKey(stream: target.stream, topic: target.topic, partitionID: partitionID)
+ return try await core.pollRouter.send(code, payload: payload, key: key, context: context, coordinator: core)
}
// MARK: Client-side partitioning
@@ -232,7 +236,8 @@ extension IggyClient {
let payload = try Requests.pollMessages(
consumer: consumer, streamID: streamID, topicID: topicID, partitionID: partitionID, strategy: strategy, count: count, autoCommit: autoCommit)
do {
- let polled = try Responses.polledMessages(try await routedPoll(payload, context: strategy.context)[...])
+ let polled = try Responses.polledMessages(
+ try await routedPartitionRequest(.pollMessages, payload: payload, context: strategy.context)[...])
if polled.messages.isEmpty, polled.partitionID == PolledMessages.resyncRequiredPartition {
// The coordinator fences a stale assignment through the
// empty poll, for want of a typed error.
@@ -262,18 +267,19 @@ extension IggyClient {
extension IggyClient {
/// Stores the offset of a consumer on a partition. A consumer-group
- /// consumer stores it for the group.
+ /// consumer stores it for the group. A nil `partitionID` fails with
+ /// ``IggyErrorCode/invalidIdentifier`` before anything is sent.
public func storeConsumerOffset(
consumer: Consumer = .default, streamID: Identifier, topicID: Identifier, partitionID: UInt32?, offset: UInt64
)
async throws
{
try await requireAuthenticated()
- try await routedOffsetWrite(
+ _ = try await routedPartitionRequest(
.storeConsumerOffset,
payload: Requests.storeConsumerOffset(
consumer: consumer, streamID: streamID, topicID: topicID, partitionID: partitionID, offset: offset, ack: .quorum),
- routePayload: Requests.getConsumerOffset(consumer: consumer, streamID: streamID, topicID: topicID, partitionID: partitionID), context: nil)
+ context: nil)
}
/// Stores a checkpoint under the partition context of the poll that
@@ -281,34 +287,36 @@ extension IggyClient {
/// consumer group, reassigned since.
public func storeConsumerPosition(consumer: Consumer = .default, streamID: Identifier, topicID: Identifier, position: ConsumerPosition) async throws {
try await requireAuthenticated()
- try await routedOffsetWrite(
+ _ = try await routedPartitionRequest(
.storeConsumerOffset,
payload: Requests.storeConsumerOffset(
consumer: consumer, streamID: streamID, topicID: topicID, partitionID: position.partitionID, offset: position.offset, ack: .quorum),
- routePayload: Requests.getConsumerOffset(consumer: consumer, streamID: streamID, topicID: topicID, partitionID: position.partitionID),
context: position.context)
}
/// The stored offset of a consumer on a partition, or nil when none is
- /// stored.
+ /// stored. A nil `partitionID` reads partition 0.
public func getConsumerOffset(
consumer: Consumer = .default, streamID: Identifier, topicID: Identifier, partitionID: UInt32?
) async throws
-> ConsumerOffsetInfo?
{
try await requireAuthenticated()
- let body = try await send(
- .getConsumerOffset, Requests.getConsumerOffset(consumer: consumer, streamID: streamID, topicID: topicID, partitionID: partitionID))
+ let body = try await routedPartitionRequest(
+ .getConsumerOffset, payload: Requests.getConsumerOffset(consumer: consumer, streamID: streamID, topicID: topicID, partitionID: partitionID),
+ context: nil)
return body.isEmpty ? nil : try Responses.consumerOffset(body[...])
}
- /// Deletes the stored offset of a consumer on a partition.
+ /// Deletes the stored offset of a consumer on a partition. A nil
+ /// `partitionID` fails with ``IggyErrorCode/invalidIdentifier`` before
+ /// anything is sent.
public func deleteConsumerOffset(consumer: Consumer = .default, streamID: Identifier, topicID: Identifier, partitionID: UInt32?) async throws {
try await requireAuthenticated()
- try await routedOffsetWrite(
+ _ = try await routedPartitionRequest(
.deleteConsumerOffset,
payload: Requests.deleteConsumerOffset(consumer: consumer, streamID: streamID, topicID: topicID, partitionID: partitionID, ack: .quorum),
- routePayload: Requests.getConsumerOffset(consumer: consumer, streamID: streamID, topicID: topicID, partitionID: partitionID), context: nil)
+ context: nil)
}
}
@@ -319,24 +327,18 @@ extension IggyClient {
/// carries the partition context the server requires. Nil for any other
/// code.
func sendRawPartitionRequest(code: UInt32, payload: [UInt8]) async throws -> [UInt8]? {
- switch CommandCode(rawValue: code) {
+ guard let command = CommandCode(rawValue: code) else {
+ return nil
+ }
+ switch command {
case .sendMessages:
let target = try RawRequest.sendTarget(payload)
return try await sendWithPartitionContext(streamID: target.stream, topicID: target.topic, partitionID: target.partitionID, payload: payload)
case .pollMessages:
try RawRequest.validatePoll(payload)
- return try await routedPoll(payload, context: nil)
- case .storeConsumerOffset, .deleteConsumerOffset:
- let prefix = try RawRequest.consumerTargetPrefix(payload)
- let command: CommandCode = code == CommandCode.storeConsumerOffset.rawValue ? .storeConsumerOffset : .deleteConsumerOffset
- let key = PollRouter.RouteKey(code: .getConsumerOffsetRouting, prefix: prefix[...])
- return try await core.pollRouter.send(command, payload: payload, key: key, routePayload: prefix, context: nil, coordinator: core)
- case .getPollRouting, .pollMessagesOnPrimary, .getConsumerOffsetRouting:
- // The routing protocol itself, sent once as the Rust client sends
- // it: a replay after a reconnect could run under another session.
- // A dropped client reconnects first, as for a typed call.
- try await requireAuthenticated()
- return try await core.exchangeOnce(code: code, payload: payload, context: .empty)
+ return try await routedPartitionRequest(.pollMessages, payload: payload, context: nil)
+ case .getConsumerOffset, .storeConsumerOffset, .deleteConsumerOffset:
+ return try await routedPartitionRequest(command, payload: payload, context: nil)
default:
return nil
}
@@ -355,6 +357,14 @@ private enum RawRequest {
let partitionID: UInt32
}
+ struct ConsumerTarget {
+ let consumerKind: UInt8
+ let stream: Identifier
+ let topic: Identifier
+ /// Nil when the request names no partition.
+ let partitionID: UInt32?
+ }
+
/// `[metadata_len:u32][stream][topic][partitioning][messages_count:u32]`
/// leads the batch, and the fields must fit the declared length. Balanced
/// and key partitioning are resolved by the typed send from its cache and
@@ -375,29 +385,27 @@ private enum RawRequest {
return SendTarget(stream: stream, topic: topic, partitionID: partitionID)
}
- /// The consumer, stream, topic, and partition that lead a poll and an
- /// offset request, which is the payload of the offset route lookup.
- static func consumerTargetPrefix(_ payload: [UInt8]) throws -> [UInt8] {
- let length = try decode(payload) { reader in
- try skipConsumerTarget(&reader)
- return reader.position
- }
- return Array(payload[..<length])
+ /// `[consumer kind:1][consumer][stream][topic][has partition:1][partition:4]`
+ /// leads a poll and every consumer-offset request.
+ static func consumerTarget(_ payload: [UInt8]) throws -> ConsumerTarget {
+ try decode(payload) { reader in try readConsumerTarget(&reader) }
}
static func validatePoll(_ payload: [UInt8]) throws {
try decode(payload) { reader in
- try skipConsumerTarget(&reader)
+ _ = try readConsumerTarget(&reader)
try reader.skip(pollParametersSize)
}
}
- private static func skipConsumerTarget(_ reader: inout ByteReader) throws {
- _ = try reader.readUInt8()
- _ = try Identifier.decode(from: &reader)
- _ = try Identifier.decode(from: &reader)
+ private static func readConsumerTarget(_ reader: inout ByteReader) throws -> ConsumerTarget {
+ let consumerKind = try reader.readUInt8()
_ = try Identifier.decode(from: &reader)
- try reader.skip(1 + 4)
+ let stream = try Identifier.decode(from: &reader)
+ let topic = try Identifier.decode(from: &reader)
+ let hasPartition = try reader.readUInt8() != 0
+ let partitionID = try reader.readUInt32()
+ return ConsumerTarget(consumerKind: consumerKind, stream: stream, topic: topic, partitionID: hasPartition ? partitionID : nil)
}
private static func decode<T>(_ payload: [UInt8], _ body: (inout ByteReader) throws -> T) throws -> T {
diff --git a/foreign/swift/Sources/Iggy/Client/IggyClient.swift b/foreign/swift/Sources/Iggy/Client/IggyClient.swift
index 6f4e47905..39a31634d 100644
--- a/foreign/swift/Sources/Iggy/Client/IggyClient.swift
+++ b/foreign/swift/Sources/Iggy/Client/IggyClient.swift
@@ -176,15 +176,16 @@ public final class IggyClient: Sendable {
///
/// Partition requests carry the partition context their typed call
/// sends: `SendMessages` with an explicit partition id, `PollMessages`,
- /// `StoreConsumerOffset`, and `DeleteConsumerOffset` take the route of
+ /// `GetConsumerOffset`, `StoreConsumerOffset`, and `DeleteConsumerOffset`
+ /// take the route of
/// ``sendMessages(streamID:topicID:partitioning:messages:)``,
/// ``pollMessages(streamID:topicID:partitionID:consumer:strategy:count:autoCommit:)``,
/// and the offset calls. A raw send with balanced or key partitioning
- /// fails with ``IggyErrorCode/featureUnavailable``, and a partition
- /// payload that does not decode fails with ``IggyErrorCode/invalidCommand``,
- /// before anything is sent. The routing codes `GetPollRouting`,
- /// `PollMessagesOnPrimary`, and `GetConsumerOffsetRouting` are sent once
- /// on the current connection, with no reconnect or replay.
+ /// fails with ``IggyErrorCode/featureUnavailable``. A group poll, a store,
+ /// or a delete that names no partition fails with
+ /// ``IggyErrorCode/invalidIdentifier``, and a partition payload that does
+ /// not decode fails with ``IggyErrorCode/invalidCommand``. Each fails
+ /// before anything is sent.
public func sendRawRequest(code: UInt32, payload: [UInt8]) async throws -> [UInt8] {
guard !CommandCode.isSessionControl(code) else {
throw IggyError(.invalidCommand, context: "code \(code) controls the session; use login and logout")
diff --git a/foreign/swift/Sources/Iggy/Client/PollRouter.swift b/foreign/swift/Sources/Iggy/Client/PollRouter.swift
index 84a2daf80..86cba08c2 100644
--- a/foreign/swift/Sources/Iggy/Client/PollRouter.swift
+++ b/foreign/swift/Sources/Iggy/Client/PollRouter.swift
@@ -17,17 +17,19 @@
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`.
+/// Routes polls and consumer-offset requests to the primary of their
+/// partition, as `OperationDriver` in `core/sdk/src/driver.rs` routes its
+/// partition calls.
///
/// 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.
+/// on, for the route of its partition (`GetPartitionRoute`): the partition
+/// context the request carries and the replica id of the partition primary,
+/// which the cluster roster maps to an endpoint. 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 client's session with
+/// `BindSession`, so consumer-group membership stays with that session.
+/// 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)
@@ -37,17 +39,20 @@ final class PollRouter: Sendable {
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.
+ /// A partition as its caller spelled it. A stream or topic named by
+ /// number and the same one named by name are two keys.
struct RouteKey: Hashable, Sendable {
- let code: CommandCode
- let prefix: ArraySlice<UInt8>
+ let stream: Identifier
+ let topic: Identifier
+ let partitionID: UInt32
}
private struct Route: Sendable {
/// The fence of the session the route was read under.
let fence: UInt128Value
- let endpoint: String
+ let primaryReplica: UInt8
+ /// The client's session, with the route's metadata op as the
+ /// watermark a data connection binds at.
let consumerSession: Responses.SessionIdentity
let context: PartitionContext
}
@@ -70,8 +75,8 @@ final class PollRouter: Sendable {
/// 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
+ /// The roster by replica id; empty until a roster read succeeded.
+ var nodes: [UInt8: ClusterNode] = [:]
var routes: [RouteKey: Route] = [:]
var slots: [String: Slot] = [:]
var nextHeartbeat: ContinuousClock.Instant?
@@ -83,9 +88,16 @@ final class PollRouter: Sendable {
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 }
+ /// Records the roster a leader check or a roster read reported, which
+ /// tells whether requests go to data connections and where each replica
+ /// listens. An empty roster says nothing about the cluster, so the router
+ /// keeps asking. A replica id the roster repeats keeps its first node.
+ func learnRoster(_ metadata: ClusterMetadata) {
+ guard !metadata.nodes.isEmpty else {
+ return
+ }
+ let nodes = Dictionary(metadata.nodes.map { ($0.replicaID, $0) }, uniquingKeysWith: { first, _ in first })
+ state.withLockedValue { $0.nodes = nodes }
}
/// Drops every route and data connection of the session that ended and
@@ -101,7 +113,7 @@ final class PollRouter: Sendable {
}
}
- /// Sends a poll or an offset write on its route. `context` is the
+ /// Sends a poll or an offset request on its route. `context` is the
/// partition context the caller captured, else the route's is taken and
/// kept for every retry.
///
@@ -109,7 +121,7 @@ final class PollRouter: Sendable {
/// of cancelling one in flight, which would take the coordinator
/// connection, and the session's group membership, with it.
func send(
- _ code: CommandCode, payload: [UInt8], key: RouteKey, routePayload: [UInt8], context: PartitionContext?, coordinator: ClientCore
+ _ code: CommandCode, payload: [UInt8], key: RouteKey, context: PartitionContext?, coordinator: ClientCore
) async throws -> [UInt8] {
let clock = ContinuousClock()
let now = clock.now
@@ -120,7 +132,7 @@ final class PollRouter: Sendable {
// pinged here once per heartbeat interval too.
_ = try await control(.ping, [], coordinator: coordinator, deadline: deadline)
}
- let request = Request(code: code, payload: payload, key: key, routePayload: routePayload, deadline: deadline)
+ let request = Request(code: code, payload: payload, key: key, deadline: deadline)
var captured = context
var interval = Self.retryInterval
while clock.now < deadline {
@@ -146,23 +158,44 @@ final class PollRouter: Sendable {
let code: CommandCode
let payload: [UInt8]
let key: RouteKey
- let routePayload: [UInt8]
let deadline: ContinuousClock.Instant
}
/// Whether the roster has more than one node, reading it once when no
/// leader check reported it yet.
private func isClustered(_ coordinator: ClientCore, deadline: ContinuousClock.Instant) async throws -> Bool {
- if state.withLockedValue({ $0.rosterSize }) == 0 {
- let readDeadline = min(deadline, ContinuousClock.now + Self.rosterReadTimeout)
- let body = try await control(.getClusterMetadata, [], coordinator: coordinator, deadline: readDeadline)
- let metadata = try Responses.clusterMetadata(body[...])
- guard !metadata.nodes.isEmpty else {
- throw IggyError(.transientNotAccepted, context: "the cluster roster is empty")
- }
- learnRosterSize(metadata.nodes.count)
+ if state.withLockedValue({ $0.nodes.isEmpty }) {
+ try await readRoster(coordinator, deadline: deadline)
+ }
+ return state.withLockedValue { $0.nodes.count } > 1
+ }
+
+ private func readRoster(_ coordinator: ClientCore, deadline: ContinuousClock.Instant) async throws {
+ let readDeadline = min(deadline, ContinuousClock.now + Self.rosterReadTimeout)
+ let body = try await control(.getClusterMetadata, [], coordinator: coordinator, deadline: readDeadline)
+ let metadata = try Responses.clusterMetadata(body[...])
+ guard !metadata.nodes.isEmpty else {
+ throw IggyError(.transientNotAccepted, context: "the cluster roster is empty")
+ }
+ learnRoster(metadata)
+ }
+
+ /// The TCP address of `replica`. A roster that does not list it is read
+ /// once more before the route is refused.
+ private func endpoint(of replica: UInt8, coordinator: ClientCore, deadline: ContinuousClock.Instant) async throws -> String {
+ var node = state.withLockedValue { $0.nodes[replica] }
+ if node == nil {
+ try await readRoster(coordinator, deadline: deadline)
+ node = state.withLockedValue { $0.nodes[replica] }
+ }
+ guard let node else {
+ throw IggyError(.transientNotAccepted, context: "the cluster roster does not list replica \(replica)")
+ }
+ guard node.endpoints.tcp != 0 else {
+ coordinator.logger.error("The partition primary \(node.name) does not expose TCP")
+ throw IggyError(.featureUnavailable, context: "the partition primary \(node.name) does not expose TCP")
}
- return state.withLockedValue { $0.rosterSize } > 1
+ return node.tcpAddress
}
private func heartbeatDue(at now: ContinuousClock.Instant, interval: Duration) -> Bool {
@@ -190,11 +223,11 @@ final class PollRouter: Sendable {
/// The single-node path: the request goes over the coordinator connection
/// with the context of its route.
private func sendOnCoordinator(_ request: Request, context: inout PartitionContext?, coordinator: ClientCore) async throws -> [UInt8] {
- let route = try await route(request.key, routePayload: request.routePayload, coordinator: coordinator, deadline: request.deadline)
+ let route = try await route(request.key, coordinator: coordinator, deadline: request.deadline)
let captured = context ?? route.context
context = captured
do {
- if request.code == .pollMessagesOnPrimary {
+ if request.code == .pollMessages {
return try await coordinator.exchangeOnce(
code: request.code.rawValue, payload: request.payload, context: captured, deadline: request.deadline)
}
@@ -206,14 +239,15 @@ final class PollRouter: Sendable {
}
/// The cluster path: the request goes over a data connection to the
- /// partition primary, bound to the coordinator's session first. Every
+ /// partition primary, bound to the client's session first. Every
/// frame is stamped under a check that the route's session is still the
/// client's, so nothing is written for a session that was replaced.
private func sendOnPrimary(_ request: Request, context: inout PartitionContext?, coordinator: ClientCore) async throws -> [UInt8] {
- let route = try await route(request.key, routePayload: request.routePayload, coordinator: coordinator, deadline: request.deadline)
+ let route = try await route(request.key, coordinator: coordinator, deadline: request.deadline)
let captured = context ?? route.context
context = captured
- let slot = try slot(for: route)
+ let endpoint = try await endpoint(of: route.primaryReplica, coordinator: coordinator, deadline: request.deadline)
+ let slot = try slot(for: route, at: endpoint)
try await slot.lock.lock()
defer { slot.lock.unlock() }
@@ -224,9 +258,9 @@ final class PollRouter: Sendable {
if data == nil {
let connection: any IggyConnection
do {
- connection = try await coordinator.dial(route.endpoint)
+ connection = try await coordinator.dial(endpoint)
} catch let error as IggyError where error.isReconnectable {
- throw IggyError(.transientNotAccepted, context: "cannot reach the partition primary at \(route.endpoint): \(error)")
+ throw IggyError(.transientNotAccepted, context: "cannot reach the partition primary at \(endpoint): \(error)")
}
data = DataConnection(connection: connection, consumerSession: nil, usable: true)
}
@@ -242,7 +276,7 @@ final class PollRouter: Sendable {
do {
try await bind(data.connection, to: route.consumerSession, coordinator: coordinator, deadline: request.deadline)
} catch let error as IggyError where error.isReconnectable {
- throw IggyError(.transientNotAccepted, context: "binding the session on \(route.endpoint) failed: \(error)")
+ throw IggyError(.transientNotAccepted, context: "binding the session on \(endpoint) failed: \(error)")
}
data.consumerSession = route.consumerSession
slot.connection.withLockedValue { $0 = data }
@@ -262,7 +296,7 @@ final class PollRouter: Sendable {
}
slot.connection.withLockedValue { $0 = data }
removeRoute(request.key)
- throw Self.gradedLoss(error, on: route.endpoint)
+ throw Self.gradedLoss(error, on: endpoint)
}
}
@@ -331,20 +365,19 @@ final class PollRouter: Sendable {
/// The cached route of `key` if the session still holds it, else a new
/// lookup on the coordinator.
- private func route(_ key: RouteKey, routePayload: [UInt8], coordinator: ClientCore, deadline: ContinuousClock.Instant) async throws -> Route {
+ private func route(_ key: RouteKey, coordinator: ClientCore, deadline: ContinuousClock.Instant) async throws -> Route {
let watermark = await coordinator.metadataWatermark
let (fence, cached) = state.withLockedValue { ($0.fence, $0.routes[key]) }
- if let cached, cached.fence == fence, cached.consumerSession.metadataWatermark >= watermark {
+ if let cached, cached.fence == fence, cached.context.metadataOp >= watermark {
return cached
}
- let body = try await control(key.code, routePayload, coordinator: coordinator, deadline: deadline)
- var response = try Responses.pollRoute(body[...])
- response.consumerSession.metadataWatermark = max(response.consumerSession.metadataWatermark, watermark)
- guard response.primary.endpoints.tcp != 0 else {
- coordinator.logger.error("The partition primary \(response.primary.name) does not expose TCP")
- throw IggyError(.featureUnavailable, context: "the partition primary \(response.primary.name) does not expose TCP")
+ let payload = try Requests.partitionRoute(streamID: key.stream, topicID: key.topic, partitionID: key.partitionID)
+ let body = try await control(.getPartitionRoute, payload, coordinator: coordinator, deadline: deadline)
+ let response = try Responses.partitionRoute(body[...])
+ guard let session = await coordinator.sessionIdentity(at: response.context.metadataOp) else {
+ throw IggyError(.transientNotAccepted, context: "no session is bound to route partition \(key.partitionID)")
}
- let route = Route(fence: fence, endpoint: response.primary.tcpAddress, consumerSession: response.consumerSession, context: response.context)
+ let route = Route(fence: fence, primaryReplica: response.primaryReplica, consumerSession: session, context: response.context)
try await validate(route, coordinator: coordinator)
try state.withLockedValue { state in
try Self.checkFence(of: route, in: state)
@@ -361,7 +394,7 @@ final class PollRouter: Sendable {
private func validate(_ route: Route, coordinator: ClientCore) async throws {
let watermark = await coordinator.metadataWatermark
try state.withLockedValue { try Self.checkFence(of: route, in: $0) }
- guard route.consumerSession.metadataWatermark >= watermark else {
+ guard route.context.metadataOp >= watermark else {
throw IggyError(.transientNotAccepted, context: "the route predates metadata commit \(watermark)")
}
}
@@ -372,10 +405,9 @@ final class PollRouter: Sendable {
}
}
- private func slot(for route: Route) throws -> Slot {
+ private func slot(for route: Route, at endpoint: String) throws -> Slot {
try state.withLockedValue { state in
try Self.checkFence(of: route, in: state)
- let endpoint = route.endpoint
if let slot = state.slots[endpoint] {
return slot
}
diff --git a/foreign/swift/Sources/Iggy/Protocol/Responses.swift b/foreign/swift/Sources/Iggy/Protocol/Responses.swift
index c386254c6..91d4d88df 100644
--- a/foreign/swift/Sources/Iggy/Protocol/Responses.swift
+++ b/foreign/swift/Sources/Iggy/Protocol/Responses.swift
@@ -430,19 +430,29 @@ enum Responses {
static let partitionRouteSize = 25
- /// The context a request routed by a `GetPartitionRoute` reply carries:
- /// `[incarnation:8][metadata_op:8][view:4][partitions_count:4][primary_replica:1]`.
+ /// One replica's route to a partition, the reply to `GetPartitionRoute`.
+ struct PartitionRoute: Equatable {
+ /// The context a request routed by the reply carries.
+ let context: PartitionContext
+ /// The configured replica id of the partition primary, as
+ /// `GetClusterMetadata` reports it, never a roster position.
+ let primaryReplica: UInt8
+ }
+
+ /// Decodes `[incarnation:8][metadata_op:8][view:4][partitions_count:4][primary_replica:1]`.
/// A route names no consumer, so the context carries owner 0, as
- /// `PartitionRouteResponse::context` in the Rust protocol crate. The view
- /// and the primary pick a replica for a client that dials each one; this
- /// one sends over its connection. A reply of any other length comes from
- /// a mismatched build.
- static func partitionRoute(_ body: ArraySlice<UInt8>) throws -> PartitionContext {
+ /// `PartitionRouteResponse::context` in the Rust protocol crate. A reply
+ /// of any other length comes from a mismatched build.
+ static func partitionRoute(_ body: ArraySlice<UInt8>) throws -> PartitionRoute {
guard body.count == partitionRouteSize else {
throw IggyError(.invalidCommand, context: "partition route is \(body.count) bytes, expected \(partitionRouteSize)")
}
return try decode(body) { reader in
- PartitionContext(incarnation: try reader.readUInt64(), ownerGeneration: 0, metadataOp: try reader.readUInt64())
+ let context = PartitionContext(incarnation: try reader.readUInt64(), ownerGeneration: 0, metadataOp: try reader.readUInt64())
+ // The view and the partition count, which a client that asks only
+ // the coordinator for routes does not use.
+ try reader.skip(2 * MemoryLayout<UInt32>.size)
+ return PartitionRoute(context: context, primaryReplica: try reader.readUInt8())
}
}
@@ -451,22 +461,7 @@ enum Responses {
struct SessionIdentity: Equatable {
let clientID: UInt128Value
let session: UInt64
- var metadataWatermark: UInt64
- }
-
- /// Where a poll or an offset write goes: the reply to `GetPollRouting`
- /// and `GetConsumerOffsetRouting`.
- struct PollRoute: Equatable {
- var consumerSession: SessionIdentity
- let context: PartitionContext
- let primary: ClusterNode
- }
-
- static func pollRoute(_ body: ArraySlice<UInt8>) throws -> PollRoute {
- try decode(body) { reader in
- let session = SessionIdentity(clientID: try reader.readUInt128(), session: try reader.readUInt64(), metadataWatermark: try reader.readUInt64())
- return PollRoute(consumerSession: session, context: try PartitionContext.decode(from: &reader), primary: try clusterNode(from: &reader))
- }
+ let metadataWatermark: UInt64
}
// MARK: Messages
diff --git a/foreign/swift/Tests/IggyTests/Fixtures/golden.json b/foreign/swift/Tests/IggyTests/Fixtures/golden.json
index 3d31d37ee..a5f24e7ed 100644
--- a/foreign/swift/Tests/IggyTests/Fixtures/golden.json
+++ b/foreign/swift/Tests/IggyTests/Fixtures/golden.json
@@ -994,7 +994,6 @@
"response.partition_route": "07000000000000000a00000000000000030000000400000002",
"response.poll_messages": "07000000660000000000000003000000000000000000000000000000000000000000000000000000070000000000000064000000000000008813000000000000e8030000000000006e01000000000000f576a2492e1328b3020000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000 [...]
"response.poll_messages.empty": "00000000000000000000000000000000000000000000000000000000000000000000000000000000",
- "response.poll_routing": "100f0e0d0c0b0a0908070605040302012a000000000000007800000000000000070000000000000003000000000000006e0000000000000002060000006e6f64652d330800000031302e302e302e339c1f0000ba0b00000100",
"response.raw_personal_access_token": "167261772d7365637265742d746f6b656e2d76616c7565",
"response.send_messages": "0200000001000000020000000300000004000000000000000100000002000000000000002a00000000000000",
"response.send_messages.empty": "00000000",
@@ -1009,7 +1008,7 @@
"version_info": "012c00000973776966742d73646b05302e312e30",
"vsr.eviction.incompatible_protocol": "0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000001000000000000000000000d000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000001032547698badcfeefcdab8967452301012c0000012c00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000 [...]
"vsr.eviction.stale_client": "0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000001000000000000000000000d000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000001032547698badcfeefcdab896745230100000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000 [...]
- "vsr.reply.committed": "0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000c0100000000000000000000080000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000001032547698badcfeefcdab89674523010000000000000000000000000000000000000000000000000700000000000000800000000000000000000000010000000000000000000000 [...]
+ "vsr.reply.committed": "0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000c0100000000000000000000080000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000001032547698badcfeefcdab896745230100000000000000000c0000000000000000000000000000000700000000000000800000000000000000000000010000000000000000000000 [...]
"vsr.reply.denied": "00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000010000000000000000000008000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000800000000000000020000000000000029000000000000000000000000000000000 [...]
"vsr.reply.ok": "0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000c0100000000000000000000080000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000001032547698badcfeefcdab896745230100000000000000000c00000000000000000000000000000007000000000000008000000000000000000000000000000000000000000000000000000 [...]
"vsr.request.create_stream": "00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000007010000000000000000000005000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000001032547698badcfeefcdab89674523013a9e6b43adf27bf20000000000000000000000000000000001000000000000008000000000000000630000000000000000000000000000000000000000000000000000000000000000000000000000000000000000 [...]
diff --git a/foreign/swift/Tests/IggyTests/MessagingTests.swift b/foreign/swift/Tests/IggyTests/MessagingTests.swift
index a5c25ca5b..1150bf664 100644
--- a/foreign/swift/Tests/IggyTests/MessagingTests.swift
+++ b/foreign/swift/Tests/IggyTests/MessagingTests.swift
@@ -211,12 +211,12 @@ struct MessagingTests {
let continued = PollingStrategy.offset(3, context: otherContext)
_ = try await client.pollMessages(streamID: 1, topicID: 2, partitionID: 1, consumer: .consumer(5), strategy: continued, count: 10, autoCommit: true)
- let lookups = await requests(server, .getPollRouting)
+ let lookups = await requests(server, .getPartitionRoute)
#expect(lookups.count == 1)
+ #expect(lookups.first?.payload == (try Requests.partitionRoute(streamID: 1, topicID: 2, partitionID: 1)))
let payload = try Requests.pollMessages(
consumer: .consumer(5), streamID: 1, topicID: 2, partitionID: 1, strategy: .next, count: 10, autoCommit: true)
- #expect(lookups.first?.payload == payload)
- let polls = await requests(server, .pollMessagesOnPrimary)
+ let polls = await requests(server, .pollMessages)
#expect(polls.map(\.context) == [FakeVSRServer.routeContext, otherContext])
#expect(polls.first?.payload == payload)
// Every request went over the one coordinator connection.
@@ -229,7 +229,7 @@ struct MessagingTests {
defer { Task { await server.stop() } }
let refused = Locked(false)
await server.respond { request in
- if request.code == CommandCode.pollMessagesOnPrimary.rawValue,
+ if request.code == CommandCode.pollMessages.rawValue,
!refused.exchange(true)
{
return .status(IggyErrorCode.transientNotAccepted.rawValue)
@@ -238,8 +238,8 @@ struct MessagingTests {
}
let client = try await connectedClient(server)
_ = try await client.pollMessages(streamID: 1, topicID: 2, partitionID: 0, strategy: .first, count: 1, autoCommit: false)
- #expect(await requests(server, .getPollRouting).count == 2)
- #expect(await requests(server, .pollMessagesOnPrimary).count == 2)
+ #expect(await requests(server, .getPartitionRoute).count == 2)
+ #expect(await requests(server, .pollMessages).count == 2)
await client.shutdown()
}
@@ -250,7 +250,7 @@ struct MessagingTests {
defer { Task { await server.stop() } }
let dropped = Locked(false)
await server.respond { request in
- if request.code == CommandCode.pollMessagesOnPrimary.rawValue, !dropped.exchange(true) {
+ if request.code == CommandCode.pollMessages.rawValue, !dropped.exchange(true) {
return .close
}
return FakeVSRServer.defaultReply(request, address: server.address)
@@ -265,23 +265,59 @@ struct MessagingTests {
@Test func aNewerMetadataCommitReadsTheRouteAgain() async throws {
let server = try await FakeVSRServer.started()
defer { Task { await server.stop() } }
+ let committed = PartitionContext(incarnation: 7, ownerGeneration: 0, metadataOp: 50)
+ let applied = Locked(FakeVSRServer.routeContext)
await server.respond { request in
- if request.operation == .deleteStream {
- return .committed([], commit: 50)
+ switch request.code.flatMap(CommandCode.init(rawValue:)) {
+ case .deleteStream:
+ applied.withLock { $0 = committed }
+ return .committed([], commit: committed.metadataOp)
+ case .getPartitionRoute:
+ return .ok(FakeVSRServer.partitionRouteBody(applied.withLock { $0 }))
+ default:
+ return FakeVSRServer.defaultReply(request, address: server.address)
}
- return FakeVSRServer.defaultReply(request, address: server.address)
}
let client = try await connectedClient(server)
let poll = { _ = try await client.pollMessages(streamID: 1, topicID: 2, partitionID: 0, strategy: .first, count: 1, autoCommit: false) }
try await poll()
try await poll()
- #expect(await requests(server, .getPollRouting).count == 1)
+ #expect(await requests(server, .getPartitionRoute).count == 1)
try await client.deleteStream(9)
#expect(await client.core.metadataWatermark == 50)
try await poll()
try await poll()
// The route read after the commit is as new as the commit.
- #expect(await requests(server, .getPollRouting).count == 2)
+ #expect(await requests(server, .getPartitionRoute).count == 2)
+ #expect(await requests(server, .pollMessages).map(\.context) == [FakeVSRServer.routeContext, FakeVSRServer.routeContext, committed, committed])
+ await client.shutdown()
+ }
+
+ /// A route older than a metadata commit the client saw is never used: it
+ /// is read again until the coordinator applied the commit.
+ @Test func aRouteOlderThanTheWatermarkIsReadAgain() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ let lagging = Locked(1)
+ await server.respond { request in
+ switch request.code.flatMap(CommandCode.init(rawValue:)) {
+ case .deleteStream:
+ return .committed([], commit: 50)
+ case .getPartitionRoute:
+ let behind = lagging.withLock {
+ $0 -= 1; return $0 >= 0
+ }
+ let metadataOp: UInt64 = behind ? 49 : 50
+ return .ok(FakeVSRServer.partitionRouteBody(PartitionContext(incarnation: 7, ownerGeneration: 0, metadataOp: metadataOp)))
+ default:
+ return FakeVSRServer.defaultReply(request, address: server.address)
+ }
+ }
+ let client = try await connectedClient(server)
+ try await client.deleteStream(9)
+ _ = try await client.pollMessages(streamID: 1, topicID: 2, partitionID: 0, strategy: .first, count: 1, autoCommit: false)
+ #expect(await requests(server, .getPartitionRoute).count == 2)
+ #expect(await requests(server, .pollMessages).map(\.context.metadataOp) == [50])
await client.shutdown()
}
@@ -301,7 +337,7 @@ struct MessagingTests {
_ = try await client.login(username: "iggy", password: "iggy")
#expect(!client.core.groupState.isRegistered(.init(stream: 1, topic: 2, group: 3)))
_ = try await client.pollMessages(streamID: 1, topicID: 2, partitionID: 0, strategy: .first, count: 1, autoCommit: false)
- let lookups = await requests(server, .getPollRouting)
+ let lookups = await requests(server, .getPartitionRoute)
#expect(lookups.count == 2)
#expect(lookups[0].clientID != lookups[1].clientID)
await client.shutdown()
@@ -333,7 +369,7 @@ struct MessagingTests {
#expect(await server.requests.filter { $0.operation == .register }.count == 1)
let binds = await requests(server, .bindSession)
#expect(binds.count == 1)
- let polls = await requests(server, .pollMessagesOnPrimary)
+ let polls = await requests(server, .pollMessages)
#expect(polls.count == 2)
#expect(Set(polls.map(\.clientID)).count == 1)
#expect(polls.allSatisfy { $0.clientID == binds[0].clientID })
@@ -414,7 +450,7 @@ struct MessagingTests {
switch request.code.flatMap(CommandCode.init(rawValue:)) {
case .syncConsumerGroup:
return .ok(FakeVSRServer.assignmentBody(generation: 1, partitions: [1, 2]))
- case .pollMessagesOnPrimary:
+ case .pollMessages:
return .ok(FakeVSRServer.polledBody(partitionID: polledPartition(request)))
default:
return FakeVSRServer.defaultReply(request, address: server.address)
@@ -430,7 +466,10 @@ struct MessagingTests {
#expect(polled == [1, 2, 1])
#expect(await requests(server, .syncConsumerGroup).count == 1)
// Each partition has its own route.
- #expect(await requests(server, .getPollRouting).count == 2)
+ #expect(
+ await requests(server, .getPartitionRoute).map(\.payload) == [
+ try Requests.partitionRoute(streamID: 1, topicID: 2, partitionID: 1), try Requests.partitionRoute(streamID: 1, topicID: 2, partitionID: 2),
+ ])
await client.shutdown()
}
@@ -442,7 +481,7 @@ struct MessagingTests {
switch request.code.flatMap(CommandCode.init(rawValue:)) {
case .syncConsumerGroup:
return .ok(FakeVSRServer.assignmentBody(generation: fenced.withLock { $0 } ? 2 : 1, partitions: [1]))
- case .pollMessagesOnPrimary:
+ case .pollMessages:
let first = !fenced.exchange(true)
return .ok(FakeVSRServer.polledBody(partitionID: first ? PolledMessages.resyncRequiredPartition : 1))
default:
@@ -453,7 +492,7 @@ struct MessagingTests {
let polled = try await client.pollMessages(streamID: 1, topicID: 2, partitionID: nil, consumer: .group(3), strategy: .next, count: 1, autoCommit: true)
#expect(polled.partitionID == 1)
#expect(await requests(server, .syncConsumerGroup).count == 2)
- #expect(await requests(server, .pollMessagesOnPrimary).count == 2)
+ #expect(await requests(server, .pollMessages).count == 2)
await client.shutdown()
}
@@ -477,7 +516,7 @@ struct MessagingTests {
member.withLock { $0 = false }
await expectCode(.consumerGroupMemberNotFound) { _ = try await poll() }
#expect(await requests(server, .syncConsumerGroup).count == 2)
- #expect(await requests(server, .pollMessagesOnPrimary).isEmpty)
+ #expect(await requests(server, .pollMessages).isEmpty)
await client.shutdown()
}
@@ -492,9 +531,9 @@ struct MessagingTests {
$0 += 1; return $0
}
return .ok(FakeVSRServer.assignmentBody(generation: UInt64(count), partitions: count == 1 ? [1] : [2]))
- case .pollMessagesOnPrimary where polledPartition(request) == 1:
+ case .pollMessages where polledPartition(request) == 1:
return .status(IggyErrorCode.consumerGroupPartitionNotOwned.rawValue)
- case .pollMessagesOnPrimary:
+ case .pollMessages:
return .ok(FakeVSRServer.polledBody(partitionID: polledPartition(request)))
default:
return FakeVSRServer.defaultReply(request, address: server.address)
@@ -534,27 +573,50 @@ struct MessagingTests {
// MARK: Offsets
- @Test func offsetWritesTakeTheOffsetRoute() async throws {
+ @Test func offsetRequestsTakeThePartitionRoute() async throws {
let server = try await FakeVSRServer.started()
defer { Task { await server.stop() } }
let client = try await connectedClient(server)
try await client.storeConsumerOffset(consumer: .consumer(5), streamID: 1, topicID: 2, partitionID: 3, offset: 40)
let position = ConsumerPosition(partitionID: 3, offset: 41, context: otherContext)
try await client.storeConsumerPosition(consumer: .consumer(5), streamID: 1, topicID: 2, position: position)
+ #expect(try await client.getConsumerOffset(consumer: .consumer(5), streamID: 1, topicID: 2, partitionID: 3) == nil)
try await client.deleteConsumerOffset(consumer: .consumer(5), streamID: 1, topicID: 2, partitionID: 3)
- let lookups = await requests(server, .getConsumerOffsetRouting)
+ let lookups = await requests(server, .getPartitionRoute)
#expect(lookups.count == 1)
- #expect(lookups.first?.payload == (try Requests.getConsumerOffset(consumer: .consumer(5), streamID: 1, topicID: 2, partitionID: 3)))
+ #expect(lookups.first?.payload == (try Requests.partitionRoute(streamID: 1, topicID: 2, partitionID: 3)))
let stores = await requests(server, .storeConsumerOffset)
#expect(stores.map(\.context) == [FakeVSRServer.routeContext, otherContext])
#expect(
stores.first?.payload
== (try Requests.storeConsumerOffset(consumer: .consumer(5), streamID: 1, topicID: 2, partitionID: 3, offset: 40, ack: .quorum)))
+ #expect(await requests(server, .getConsumerOffset).map(\.context) == [FakeVSRServer.routeContext])
#expect(await requests(server, .deleteConsumerOffset).map(\.context) == [FakeVSRServer.routeContext])
await client.shutdown()
}
+ /// An offset read without a partition reads partition 0, as the server
+ /// does. A write without one names no partition to route.
+ @Test func offsetRequestsWithoutAPartition() async throws {
+ let server = try await FakeVSRServer.started()
+ defer { Task { await server.stop() } }
+ let client = try await connectedClient(server)
+ _ = try await client.getConsumerOffset(consumer: .group(3), streamID: 1, topicID: 2, partitionID: nil)
+ #expect(await requests(server, .getPartitionRoute).map(\.payload) == [try Requests.partitionRoute(streamID: 1, topicID: 2, partitionID: 0)])
+ #expect(
+ await requests(server, .getConsumerOffset).map(\.payload)
+ == [try Requests.getConsumerOffset(consumer: .group(3), streamID: 1, topicID: 2, partitionID: nil)])
+
+ let before = await server.requests.count
+ await expectCode(.invalidIdentifier) {
+ try await client.storeConsumerOffset(consumer: .group(3), streamID: 1, topicID: 2, partitionID: nil, offset: 1)
+ }
+ await expectCode(.invalidIdentifier) { try await client.deleteConsumerOffset(consumer: .group(3), streamID: 1, topicID: 2, partitionID: nil) }
+ #expect(await server.requests.count == before)
+ await client.shutdown()
+ }
+
// MARK: Raw requests
@Test func rawPartitionRequestsAreRoutedLikeTypedOnes() async throws {
@@ -569,13 +631,19 @@ struct MessagingTests {
let poll = try Requests.pollMessages(consumer: .consumer(5), streamID: 1, topicID: 2, partitionID: 1, strategy: .first, count: 1, autoCommit: false)
_ = try await client.sendRawRequest(code: CommandCode.pollMessages.rawValue, payload: poll)
- #expect(await requests(server, .pollMessagesOnPrimary).map(\.context) == [FakeVSRServer.routeContext])
+ #expect(await requests(server, .pollMessages).map(\.context) == [FakeVSRServer.routeContext])
+ // The offset requests take the route the poll read.
let store = try Requests.storeConsumerOffset(consumer: .consumer(5), streamID: 1, topicID: 2, partitionID: 1, offset: 9, ack: .quorum)
_ = try await client.sendRawRequest(code: CommandCode.storeConsumerOffset.rawValue, payload: store)
- #expect(
- await requests(server, .getConsumerOffsetRouting).map(\.payload)
- == [try Requests.getConsumerOffset(consumer: .consumer(5), streamID: 1, topicID: 2, partitionID: 1)])
+ let read = try Requests.getConsumerOffset(consumer: .consumer(5), streamID: 1, topicID: 2, partitionID: 1)
+ _ = try await client.sendRawRequest(code: CommandCode.getConsumerOffset.rawValue, payload: read)
+ let delete = try Requests.deleteConsumerOffset(consumer: .consumer(5), streamID: 1, topicID: 2, partitionID: 1, ack: .quorum)
+ _ = try await client.sendRawRequest(code: CommandCode.deleteConsumerOffset.rawValue, payload: delete)
+ for code in [CommandCode.storeConsumerOffset, .getConsumerOffset, .deleteConsumerOffset] {
+ #expect(await requests(server, code).map(\.context) == [FakeVSRServer.routeContext])
+ }
+ #expect(await requests(server, .getPartitionRoute).count == 2)
let before = await server.requests.count
let balanced = try Batch.encodeSend(streamID: 1, topicID: 2, partitioning: .balanced, messages: outgoing)
@@ -592,43 +660,29 @@ struct MessagingTests {
await expectCode(.invalidCommand) { _ = try await client.sendRawRequest(code: CommandCode.pollMessages.rawValue, payload: Array(poll.dropLast())) }
await expectCode(.invalidCommand) { _ = try await client.sendRawRequest(code: CommandCode.storeConsumerOffset.rawValue, payload: [1, 9]) }
await expectCode(.invalidCommand) { _ = try await client.sendRawRequest(code: CommandCode.bindSession.rawValue, payload: []) }
- #expect(await server.requests.count == before)
- await client.shutdown()
- }
-
- /// The routing protocol goes once on the current connection, as the Rust
- /// client sends it: a replay after a reconnect would run under another
- /// connection's binding.
- @Test func rawRoutingRequestsAreSentOnce() async throws {
- let server = try await FakeVSRServer.started()
- defer { Task { await server.stop() } }
- let dropped = Locked(false)
- await server.respond { request in
- if request.code == CommandCode.getPollRouting.rawValue, !dropped.exchange(true) {
- return .close
- }
- return FakeVSRServer.defaultReply(request, address: server.address)
+ // A group poll and an offset write must name their partition.
+ let groupPoll = try Requests.pollMessages(consumer: .group(3), streamID: 1, topicID: 2, partitionID: nil, strategy: .next, count: 1, autoCommit: true)
+ let unnamedStore = try Requests.storeConsumerOffset(consumer: .consumer(5), streamID: 1, topicID: 2, partitionID: nil, offset: 9, ack: .quorum)
+ let unnamedDelete = try Requests.deleteConsumerOffset(consumer: .consumer(5), streamID: 1, topicID: 2, partitionID: nil, ack: .quorum)
+ for (code, payload) in [(CommandCode.pollMessages, groupPoll), (.storeConsumerOffset, unnamedStore), (.deleteConsumerOffset, unnamedDelete)] {
+ await expectCode(.invalidIdentifier) { _ = try await client.sendRawRequest(code: code.rawValue, payload: payload) }
}
- let client = try await connectedClient(server)
- let poll = try Requests.pollMessages(consumer: .consumer(5), streamID: 1, topicID: 2, partitionID: 1, strategy: .first, count: 1, autoCommit: false)
- await expectCode(.disconnected) { _ = try await client.sendRawRequest(code: CommandCode.getPollRouting.rawValue, payload: poll) }
- #expect(await requests(server, .getPollRouting).count == 1)
- let body = try await client.sendRawRequest(code: CommandCode.getPollRouting.rawValue, payload: poll)
- #expect(try Responses.pollRoute(body[...]).context == FakeVSRServer.routeContext)
+ #expect(await server.requests.count == before)
await client.shutdown()
}
// MARK: Clusters
- /// A coordinator that routes every poll and offset write to `primary`.
+ /// A coordinator, replica 0, that routes every partition to `primary`,
+ /// replica 1.
private static func coordinator(routingTo primary: FakeVSRServer) async throws -> FakeVSRServer {
let coordinator = try await FakeVSRServer.started()
await coordinator.respond { request in
switch request.code.flatMap(CommandCode.init(rawValue:)) {
case .getClusterMetadata:
return .ok(FakeVSRServer.rosterBody([coordinator.address, primary.address]))
- case .getPollRouting, .getConsumerOffsetRouting:
- return .ok(FakeVSRServer.pollRouteBody(for: request, context: FakeVSRServer.routeContext, primary: primary.address, watermark: 3))
+ case .getPartitionRoute:
+ return .ok(FakeVSRServer.partitionRouteBody(FakeVSRServer.routeContext, primaryReplica: 1))
default:
return FakeVSRServer.defaultReply(request, address: coordinator.address)
}
@@ -653,15 +707,17 @@ struct MessagingTests {
let session = await coordinator.requests.last!
let binds = await requests(primary, .bindSession)
#expect(binds.count == 1)
+ // The data connection binds at the metadata op of the route.
#expect(
binds.first?.payload
== (try Requests.bindSession(
- clientID: session.clientID, session: session.session, metadataWatermark: 3, bindSecret: await client.core.bindSecret)))
- let polls = await requests(primary, .pollMessagesOnPrimary)
+ clientID: session.clientID, session: session.session, metadataWatermark: FakeVSRServer.routeContext.metadataOp,
+ bindSecret: await client.core.bindSecret)))
+ let polls = await requests(primary, .pollMessages)
#expect(polls.count == 2)
#expect(polls.allSatisfy { $0.context == FakeVSRServer.routeContext && $0.clientID == session.clientID && $0.session == session.session })
#expect(await requests(primary, .storeConsumerOffset).count == 1)
- #expect(await requests(coordinator, .pollMessagesOnPrimary).isEmpty)
+ #expect(await requests(coordinator, .pollMessages).isEmpty)
#expect(await primary.connectionCount == 1)
await client.shutdown()
}
@@ -677,7 +733,7 @@ struct MessagingTests {
}
let dropped = Locked(false)
await primary.respond { request in
- if request.code == CommandCode.pollMessagesOnPrimary.rawValue,
+ if request.code == CommandCode.pollMessages.rawValue,
!dropped.exchange(true)
{
return .close
@@ -707,15 +763,15 @@ struct MessagingTests {
}
let refused = Locked(false)
await primary.respond { request in
- if request.code == CommandCode.pollMessagesOnPrimary.rawValue, !refused.exchange(true) {
+ if request.code == CommandCode.pollMessages.rawValue, !refused.exchange(true) {
return .status(IggyErrorCode.notPrimary.rawValue)
}
return FakeVSRServer.defaultReply(request, address: primary.address)
}
let client = try await connectedClient(coordinator)
_ = try await client.pollMessages(streamID: 1, topicID: 2, partitionID: 0, strategy: .first, count: 1, autoCommit: false)
- #expect(await requests(coordinator, .getPollRouting).count == 2)
- #expect(await requests(primary, .pollMessagesOnPrimary).count == 2)
+ #expect(await requests(coordinator, .getPartitionRoute).count == 2)
+ #expect(await requests(primary, .pollMessages).count == 2)
// The refusal left the data connection at a frame boundary.
#expect(await primary.connectionCount == 1)
await client.shutdown()
@@ -737,12 +793,12 @@ struct MessagingTests {
await coordinator.respond { request in
switch request.code.flatMap(CommandCode.init(rawValue:)) {
case .getClusterMetadata:
- return .ok(FakeVSRServer.rosterBody([coordinator.address, primary.address]))
- case .getPollRouting:
+ return .ok(FakeVSRServer.rosterBody([coordinator.address, goneAddress, primary.address]))
+ case .getPartitionRoute:
let first = lookups.withLock {
$0 += 1; return $0 == 1
}
- return .ok(FakeVSRServer.pollRouteBody(for: request, context: FakeVSRServer.routeContext, primary: first ? goneAddress : primary.address))
+ return .ok(FakeVSRServer.partitionRouteBody(FakeVSRServer.routeContext, primaryReplica: first ? 1 : 2))
default:
return FakeVSRServer.defaultReply(request, address: coordinator.address)
}
@@ -750,7 +806,43 @@ struct MessagingTests {
let client = try await connectedClient(coordinator)
_ = try await client.pollMessages(streamID: 1, topicID: 2, partitionID: 0, strategy: .first, count: 1, autoCommit: false)
#expect(lookups.withLock { $0 } == 2)
- #expect(await requests(primary, .pollMessagesOnPrimary).count == 1)
+ #expect(await requests(primary, .pollMessages).count == 1)
+ await client.shutdown()
+ }
+
+ /// A route can name a replica that the roster the client holds does not
+ /// list. The roster is read again before the route is refused.
+ @Test func aPrimaryMissingFromTheRosterReadsTheRosterAgain() async throws {
+ let primary = try await FakeVSRServer.started()
+ let other = try await FakeVSRServer.started()
+ let coordinator = try await FakeVSRServer.started()
+ defer {
+ Task {
+ await coordinator.stop()
+ await other.stop()
+ await primary.stop()
+ }
+ }
+ let routed = Locked(false)
+ await coordinator.respond { request in
+ switch request.code.flatMap(CommandCode.init(rawValue:)) {
+ case .getClusterMetadata:
+ let addresses = [coordinator.address, other.address] + (routed.withLock { $0 } ? [primary.address] : [])
+ return .ok(FakeVSRServer.rosterBody(addresses))
+ case .getPartitionRoute:
+ routed.withLock { $0 = true }
+ return .ok(FakeVSRServer.partitionRouteBody(FakeVSRServer.routeContext, primaryReplica: 2))
+ default:
+ return FakeVSRServer.defaultReply(request, address: coordinator.address)
+ }
+ }
+ let client = try await connectedClient(coordinator)
+ let rosterReads = await requests(coordinator, .getClusterMetadata).count
+ _ = try await client.pollMessages(streamID: 1, topicID: 2, partitionID: 0, strategy: .first, count: 1, autoCommit: false)
+ #expect(await requests(coordinator, .getClusterMetadata).count == rosterReads + 1)
+ #expect(await requests(coordinator, .getPartitionRoute).count == 1)
+ #expect(await requests(primary, .pollMessages).count == 1)
+ #expect(await other.connectionCount == 0)
await client.shutdown()
}
}
diff --git a/foreign/swift/Tests/IggyTests/ResponseVectorTests.swift b/foreign/swift/Tests/IggyTests/ResponseVectorTests.swift
index 8bf0fae85..9433c59af 100644
--- a/foreign/swift/Tests/IggyTests/ResponseVectorTests.swift
+++ b/foreign/swift/Tests/IggyTests/ResponseVectorTests.swift
@@ -196,17 +196,13 @@ struct ResponseVectorTests {
}
@Test func routes() async throws {
- #expect(try Responses.partitionRoute(body("response.partition_route")) == PartitionContext(incarnation: 7, ownerGeneration: 0, metadataOp: 10))
+ let route = try Responses.partitionRoute(body("response.partition_route"))
+ #expect(route.context == PartitionContext(incarnation: 7, ownerGeneration: 0, metadataOp: 10))
+ // The view and the partition count sit between the context and the
+ // primary.
+ #expect(route.primaryReplica == 2)
// A route of any other length comes from a mismatched build.
await expectCode(.invalidCommand) { _ = try Responses.partitionRoute(body("response.partition_route") + [0]) }
-
- let poll = try Responses.pollRoute(body("response.poll_routing"))
- #expect(poll.consumerSession.clientID == UInt128Value(low: 0x090a_0b0c_0d0e_0f10, high: 0x0102_0304_0506_0708))
- #expect(poll.consumerSession.session == 42)
- #expect(poll.consumerSession.metadataWatermark == 120)
- #expect(poll.context == PartitionContext(incarnation: 7, ownerGeneration: 3, metadataOp: 110))
- #expect(poll.primary.replicaID == 2)
- #expect(poll.primary.tcpAddress == "10.0.0.3:8092")
}
@Test func clusterMetadata() throws {
diff --git a/foreign/swift/Tests/IggyTests/Support/FakeVSRServer.swift b/foreign/swift/Tests/IggyTests/Support/FakeVSRServer.swift
index 8f42505e8..6e039f6e5 100644
--- a/foreign/swift/Tests/IggyTests/Support/FakeVSRServer.swift
+++ b/foreign/swift/Tests/IggyTests/Support/FakeVSRServer.swift
@@ -166,18 +166,6 @@ final class FakeVSRServer: @unchecked Sendable {
return writer.bytes
}
- /// The reply to `GetPollRouting` and `GetConsumerOffsetRouting`: the
- /// session of `request` and a primary at `primary`.
- static func pollRouteBody(for request: Request, context: PartitionContext, primary: String, watermark: UInt64 = 0) -> [UInt8] {
- var writer = ByteWriter()
- writer.write(request.clientID)
- writer.write(request.session)
- writer.write(watermark)
- context.encode(into: &writer)
- writeNode(primary, replicaID: 0, leader: true, into: &writer)
- return writer.bytes
- }
-
/// A poll reply without messages.
static func polledBody(partitionID: UInt32, context: PartitionContext = .empty) -> [UInt8] {
var writer = ByteWriter()
@@ -296,9 +284,7 @@ final class FakeVSRServer: @unchecked Sendable {
return .ok(registerBody(session: request.session))
case .getPartitionRoute:
return .ok(partitionRouteBody(routeContext))
- case .getPollRouting, .getConsumerOffsetRouting:
- return .ok(pollRouteBody(for: request, context: routeContext, primary: address))
- case .pollMessagesOnPrimary:
+ case .pollMessages:
return .ok(polledBody(partitionID: 0, context: routeContext))
default:
return .ok([])
diff --git a/foreign/swift/Tools/golden-vectors/src/main.rs b/foreign/swift/Tools/golden-vectors/src/main.rs
index 75196c29a..d9f8e2ff6 100644
--- a/foreign/swift/Tools/golden-vectors/src/main.rs
+++ b/foreign/swift/Tools/golden-vectors/src/main.rs
@@ -67,8 +67,8 @@ use iggy_binary_protocol::responses::consumer_groups::{
};
use iggy_binary_protocol::responses::consumer_offsets::ConsumerOffsetResponse;
use iggy_binary_protocol::responses::messages::{
- PartitionRouteResponse, PollMessagesResponseHeader, PollRoutingResponse,
- SendMessagesConfirmationResponse, SendMessagesResponse,
+ PartitionRouteResponse, PollMessagesResponseHeader, SendMessagesConfirmationResponse,
+ SendMessagesResponse,
};
use iggy_binary_protocol::responses::personal_access_tokens::{
GetPersonalAccessTokensResponse, PersonalAccessTokenResponse, RawPersonalAccessTokenResponse,
@@ -1427,33 +1427,6 @@ fn main() {
primary_replica: 2,
},
);
- add(
- &mut vectors,
- "response.poll_routing",
- &PollRoutingResponse {
- consumer_session: SessionIdentity {
- client_id: 0x0102_0304_0506_0708_090a_0b0c_0d0e_0f10,
- session: 42,
- metadata_watermark: 120,
- },
- context: PartitionContext {
- incarnation: 7,
- owner_generation: 3,
- metadata_op: 110,
- },
- primary: ClusterNodeResponse {
- replica_id: 2,
- name: "node-3".to_owned(),
- ip: "10.0.0.3".to_owned(),
- tcp_port: 8092,
- quic_port: 0,
- http_port: 3002,
- websocket_port: 0,
- role: 1,
- status: 0,
- },
- },
- );
// Domain option blocks.
let create_options = TopicCreateOptions {