From d31d92aca120fe384efa6ae6e6f9694e0f1904b3 Mon Sep 17 00:00:00 2001 From: RustToMetal <324354762+RustToMetal@users.noreply.github.com> Date: Sun, 25 Oct 2026 02:49:29 -0500 Subject: [PATCH 1/2] feat(swift): add the consumer An application reading a topic needs more than pollMessages: it has to keep each partition's position, continue after the last message it handled, commit offsets at the right moment, and come back after an outage or a rebalance. The Rust SDK does this in IggyConsumer, and the Swift SDK had no counterpart. IggyConsumer ports it as an actor that is also an AsyncSequence. A standalone consumer reads one partition, a group member joins its group and reads the server's assignment. Each poll continues after the last message handed over under the context that delivered it, a message already handed over from the same partition history is skipped unless replay is allowed, and the auto-commit modes, interval task, queued commits, and shutdown flush follow the Rust consumer. consumeMessages carries the after-handler commits and stops once the client cannot come back on its own. Encryption and resolving a lost group create or join need client features Swift does not have yet. --- foreign/swift/README.md | 36 +- .../Sources/Iggy/Client/ClientCore.swift | 13 + .../Sources/Iggy/Client/IggyClient.swift | 5 + .../Iggy/Consumer/ConsumerConfiguration.swift | 200 ++++++ .../Sources/Iggy/Consumer/IggyConsumer.swift | 672 ++++++++++++++++++ .../Tests/IggyE2ETests/ConsumerE2ETests.swift | 100 +++ .../swift/Tests/IggyTests/ConsumerTests.swift | 532 ++++++++++++++ .../IggyTests/Support/FakeVSRServer.swift | 32 + 8 files changed, 1588 insertions(+), 2 deletions(-) create mode 100644 foreign/swift/Sources/Iggy/Consumer/ConsumerConfiguration.swift create mode 100644 foreign/swift/Sources/Iggy/Consumer/IggyConsumer.swift create mode 100644 foreign/swift/Tests/IggyE2ETests/ConsumerE2ETests.swift create mode 100644 foreign/swift/Tests/IggyTests/ConsumerTests.swift diff --git a/foreign/swift/README.md b/foreign/swift/README.md index c8717df395..eecd2161dc 100644 --- a/foreign/swift/README.md +++ b/foreign/swift/README.md @@ -14,8 +14,9 @@ The SDK is built up in stages. The client is a TCP connection with optional TLS speaks the consensus protocol, signs in, keeps the session alive with a heartbeat, reconnects, and follows the cluster leader. It manages streams, topics, partitions, segments, users, personal access tokens, and consumer groups, sends and polls messages, -stores consumer offsets, and produces with a batching producer. The consumer, examples, -and BDD scenarios follow in later changes. +stores consumer offsets, produces with a batching producer, and consumes with a +consumer that tracks and commits its position. Examples and BDD scenarios follow in +later changes. Swift supports the consensus wire protocol and checks commitment proof during automatic retries. It does not yet expose retained-operation resolve/abandon APIs, @@ -322,6 +323,37 @@ written in order. A topic addressed by name and by numeric id counts as two, and failed request does not hold back the ones after it, so a resend from `errorHandler` lands behind them. +### Consumer + +`IggyConsumer` reads one topic and hands its messages over one at a time. A standalone +consumer reads one partition; a member of a consumer group reads the partitions the +server assigns, joining the group in `initialize()`: + +```swift +let consumer = try client.consumerGroup(name: "billing", streamID: "orders", topicID: "created") +try await consumer.initialize() +for try await received in consumer { + print(received.partitionID, received.message.offset, received.message.payloadString) +} +``` + +A poll refills the buffer once it runs empty, and each poll of a partition continues +after the last message handed over from it. A failed poll is thrown after the retry +interval and leaves the consumer usable, so a caller of `next()` can call it again, and +`consumeMessages(_:)` logs it and keeps reading until the client cannot come back on its +own. Without `allowReplay` a message at or below the last one handed over from the same +partition history is skipped. + +`autoCommit` decides when the position is stored. The default commits with every poll, +before the batch is handed over, and on a one-second interval; `.disabled` leaves it to +`storePosition(_:)` with the position of a `ReceivedMessage`, the only setting under which a +message whose handling failed comes back on the next run. `shutdown()` sends the queued +commits, stores the reading position of every partition read unless auto-commit is +disabled, and leaves the group. + +Unlike the Rust consumer this one has no encryptor, and it does not resolve a group create +or join whose reply was lost, since the Swift client has no operation lanes yet. + ### Errors Every client operation throws `IggyError`, a value carrying the typed `code` diff --git a/foreign/swift/Sources/Iggy/Client/ClientCore.swift b/foreign/swift/Sources/Iggy/Client/ClientCore.swift index 976ae2d045..f3dce54489 100644 --- a/foreign/swift/Sources/Iggy/Client/ClientCore.swift +++ b/foreign/swift/Sources/Iggy/Client/ClientCore.swift @@ -383,6 +383,19 @@ actor ClientCore { try VSRFrame.encodeRequest(session: &session, code: code, payload: payload, partitionContext: context) } + /// Whether the client comes back on its own: it is signed in, or a + /// request or the heartbeat reconnects and signs it in again. A client + /// that is shut down, that the caller disconnected, or that has no + /// reconnection or no credentials only comes back when the caller acts. + var canRecover: Bool { + switch state { + case .shutdown: false + case .authenticated: true + case .disconnected, .connecting, .connected, .authenticating: + !disconnectedByCaller && configuration.reconnection.enabled && signInCredentials() != nil + } + } + /// The refusal of a call that started under `captured` once the caller /// signed in or out since: nothing of it may be written any more. func lineageRefusal(_ captured: UInt64, code: UInt32? = nil) -> IggyError? { diff --git a/foreign/swift/Sources/Iggy/Client/IggyClient.swift b/foreign/swift/Sources/Iggy/Client/IggyClient.swift index 357102a750..3136c84ff9 100644 --- a/foreign/swift/Sources/Iggy/Client/IggyClient.swift +++ b/foreign/swift/Sources/Iggy/Client/IggyClient.swift @@ -239,6 +239,11 @@ public final class IggyClient: Sendable { return captured } + /// Whether the client comes back on its own after an outage. + var canRecover: Bool { + get async { await core.canRecover } + } + /// The lineage a call that starts now runs under. var operationLineage: UInt64 { get async { await core.lineage } diff --git a/foreign/swift/Sources/Iggy/Consumer/ConsumerConfiguration.swift b/foreign/swift/Sources/Iggy/Consumer/ConsumerConfiguration.swift new file mode 100644 index 0000000000..0a2b622f43 --- /dev/null +++ b/foreign/swift/Sources/Iggy/Consumer/ConsumerConfiguration.swift @@ -0,0 +1,200 @@ +// 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. + +/// When a consumer stores its offsets on the server. +public enum AutoCommit: Sendable, Hashable { + /// Never, not even on ``IggyConsumer/shutdown()``; store with + /// ``IggyConsumer/storePosition(_:)``. + case disabled + /// On every tick, the reading position of every partition read so far. + case interval(Duration) + /// On every tick, and as `when` says while messages are handed over. + case intervalOrWhen(Duration, AutoCommitWhen) + /// On every tick, and as `after` says once a handler of + /// ``IggyConsumer/consumeMessages(_:)`` returned. + case intervalOrAfter(Duration, AutoCommitAfter) + /// As `when` says while messages are handed over. + case when(AutoCommitWhen) + /// As `after` says once a handler of ``IggyConsumer/consumeMessages(_:)`` + /// returned. Reading with ``IggyConsumer/next()`` never triggers it. + case after(AutoCommitAfter) + + var interval: Duration? { + switch self { + case .interval(let interval), .intervalOrWhen(let interval, _), .intervalOrAfter(let interval, _): interval + case .disabled, .when, .after: nil + } + } + + var whenTrigger: AutoCommitWhen? { + switch self { + case .intervalOrWhen(_, let when), .when(let when): when + case .disabled, .interval, .intervalOrAfter, .after: nil + } + } + + var afterTrigger: AutoCommitAfter? { + switch self { + case .intervalOrAfter(_, let after), .after(let after): after + case .disabled, .interval, .intervalOrWhen, .when: nil + } + } +} + +/// A commit triggered while messages are handed over, before the code that +/// reads them has seen them. +public enum AutoCommitWhen: Sendable, Hashable { + /// With the poll request itself: the server stores the offset of the + /// last message it returns, before any of the batch is handed over. + case pollingMessages + /// Queued when the buffer of the current batch runs empty. + case consumingAllMessages + /// Queued just before each message is handed over. Commits queued faster + /// than they are sent collapse into the latest one per partition. + case consumingEachMessage + /// Queued just before a message whose offset divides by `n` is handed + /// over. It tests the offset, not a count of messages. `n` must not be + /// zero. + case consumingEveryNthMessage(UInt32) +} + +/// A commit triggered once a handler of ``IggyConsumer/consumeMessages(_:)`` +/// returned, whether it succeeded or threw. +public enum AutoCommitAfter: Sendable, Hashable { + /// After the message whose offset is the partition's newest at poll + /// time. + case consumingAllMessages + /// After each message. + case consumingEachMessage + /// After a message whose offset divides by `n`. `n` must not be zero. + case consumingEveryNthMessage(UInt32) +} + +/// Options of an ``IggyConsumer``, with the Rust SDK's defaults. +public struct ConsumerConfiguration: Sendable { + /// Where reading a partition starts. After the first batch each poll + /// continues after the last message handed over from that partition, + /// except under ``PollingStrategy/next``, which leaves the continuation + /// to the offsets stored on the server. + public var pollingStrategy: PollingStrategy + /// Messages asked for per poll. + public var batchLength: UInt32 + /// Minimum gap between the starts of two polls. Nil polls as soon as the + /// buffer is empty. + public var pollInterval: Duration? + public var autoCommit: AutoCommit + /// Joins the group during ``IggyConsumer/initialize()`` and again before + /// a poll once the membership was lost. Off, a poll without a membership + /// fails with ``IggyErrorCode/consumerGroupMemberNotFound``. + public var autoJoinConsumerGroup: Bool + /// Creates the group when it is missing before joining it. + public var createConsumerGroupIfNotExists: Bool + /// The pause after a failed poll or join, and between checks while the + /// client is signed out. + public var pollingRetryInterval: Duration + /// Lookups of a missing stream or topic during + /// ``IggyConsumer/initialize()`` before it fails. Nil or zero looks once. + public var initRetries: UInt32? + public var initRetryInterval: Duration + /// Hands over a message again when a poll returns it again, as after a + /// rebalance or a store that moved back. Off, a message at or below the + /// last one handed over from the same partition history is skipped. + public var allowReplay: Bool + /// How long ``IggyConsumer/shutdown()`` waits for each background commit + /// task before it stops it. + public var offsetDrainTimeout: Duration + + public init( + pollingStrategy: PollingStrategy = .next, batchLength: UInt32 = 1000, pollInterval: Duration? = nil, + autoCommit: AutoCommit = .intervalOrWhen(.seconds(1), .pollingMessages), autoJoinConsumerGroup: Bool = true, + createConsumerGroupIfNotExists: Bool = true, pollingRetryInterval: Duration = .seconds(1), initRetries: UInt32? = nil, + initRetryInterval: Duration = .seconds(1), allowReplay: Bool = false, offsetDrainTimeout: Duration = .seconds(5) + ) { + self.pollingStrategy = pollingStrategy + self.batchLength = batchLength + self.pollInterval = pollInterval + self.autoCommit = autoCommit + self.autoJoinConsumerGroup = autoJoinConsumerGroup + self.createConsumerGroupIfNotExists = createConsumerGroupIfNotExists + self.pollingRetryInterval = pollingRetryInterval + self.initRetries = initRetries + self.initRetryInterval = initRetryInterval + self.allowReplay = allowReplay + self.offsetDrainTimeout = offsetDrainTimeout + } + + /// Refuses what the Rust SDK's types cannot hold: a zero batch, a zero + /// or negative retry or commit interval (`NonZeroIggyDuration`), a zero + /// `n`, and any duration past `UInt64.max` microseconds, which would + /// also trap in `Task.sleep`. + func validate() throws { + if batchLength == 0 { + throw IggyError(.invalidConfiguration, context: "batchLength must be at least 1") + } + try Self.validatePositive(pollingRetryInterval, named: "pollingRetryInterval") + try Self.validatePositive(initRetryInterval, named: "initRetryInterval") + if let interval = autoCommit.interval { + try Self.validatePositive(interval, named: "the auto-commit interval") + } + if let pollInterval { + try ProducerConfiguration.validate(pollInterval, named: "pollInterval") + } + try ProducerConfiguration.validate(offsetDrainTimeout, named: "offsetDrainTimeout") + if case .consumingEveryNthMessage(0)? = autoCommit.whenTrigger { + throw IggyError(.invalidConfiguration, context: "consumingEveryNthMessage needs n of at least 1") + } + if case .consumingEveryNthMessage(0)? = autoCommit.afterTrigger { + throw IggyError(.invalidConfiguration, context: "consumingEveryNthMessage needs n of at least 1") + } + } + + private static func validatePositive(_ duration: Duration, named name: String) throws { + guard duration > .zero else { + throw IggyError(.invalidConfiguration, context: "\(name) must be positive, got \(duration)") + } + try ProducerConfiguration.validate(duration, named: name) + } +} + +/// One message handed over by an ``IggyConsumer``. +public struct ReceivedMessage: Sendable { + public var message: IggyMessage + /// The newest offset the partition had when the batch was polled, the + /// same for every message of the batch. Compared with the message's + /// offset it shows how far the consumer lags. + public var currentOffset: UInt64 + /// The partition the message was read from. A group member reads + /// several. + public var partitionID: UInt32 + /// The partition incarnation and owner the poll ran under. + public var context: PartitionContext + + public init(message: IggyMessage, currentOffset: UInt64, partitionID: UInt32, context: PartitionContext) { + self.message = message + self.currentOffset = currentOffset + self.partitionID = partitionID + self.context = context + } + + /// The checkpoint to store for this message with + /// ``IggyConsumer/storePosition(_:)``. It keeps the context that + /// delivered the message, so a store after a recreation or a reassignment + /// is refused instead of landing on the newer partition. + public var position: ConsumerPosition { + ConsumerPosition(partitionID: partitionID, offset: message.offset, context: context) + } +} diff --git a/foreign/swift/Sources/Iggy/Consumer/IggyConsumer.swift b/foreign/swift/Sources/Iggy/Consumer/IggyConsumer.swift new file mode 100644 index 0000000000..b7590bfd72 --- /dev/null +++ b/foreign/swift/Sources/Iggy/Consumer/IggyConsumer.swift @@ -0,0 +1,672 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +import Logging +import NIOConcurrencyHelpers + +/// Reads the partitions of one topic and hands their messages over one at a +/// time, a port of `IggyConsumer` in the Rust SDK. +/// +/// Build a standalone consumer of one partition with +/// ``IggyClient/consumer(name:streamID:topicID:partitionID:configuration:)``, +/// or a member of a consumer group with +/// ``IggyClient/consumerGroup(name:streamID:topicID:configuration:)``, whose +/// partitions the server assigns. Call ``initialize()`` once, then read with +/// ``next()``, `for try await`, or ``consumeMessages(_:)``: +/// +/// ```swift +/// let consumer = try client.consumerGroup(name: "billing", streamID: "orders", topicID: "created") +/// try await consumer.initialize() +/// for try await received in consumer { +/// print(received.partitionID, received.message.offset) +/// } +/// ``` +/// +/// Messages come from a buffer that a poll refills once it runs empty. A +/// failed poll is thrown after ``ConsumerConfiguration/pollingRetryInterval`` +/// and leaves the consumer usable, so a caller of ``next()`` can call it again; +/// a `for try await` loop ends at the first error instead. ``next()`` returns +/// nil only after ``shutdown()``. +/// +/// The consumer remembers the last message handed over from each partition. +/// Without ``ConsumerConfiguration/allowReplay`` a message at or below it is +/// not handed over again, and ``ConsumerConfiguration/autoCommit`` decides +/// when the reading position is stored on the server. +/// +/// Unlike the Rust consumer, this one has no encryptor, since the Swift +/// client has none, and does not resolve a group create or join whose outcome +/// was lost: the Swift client has no operation lanes to resolve them through. +public actor IggyConsumer: AsyncSequence { + public typealias Element = ReceivedMessage + + public nonisolated let name: String + public nonisolated let streamID: Identifier + public nonisolated let topicID: Identifier + public nonisolated let consumer: Consumer + /// The partition a standalone consumer reads; nil for a group member. + public nonisolated let partitionID: UInt32? + /// The partition of the last batch handed over. Before the first one it + /// is the partition of a standalone consumer and zero for a group member. + public private(set) var currentPartitionID: UInt32 + + private let client: IggyClient + private let configuration: ConsumerConfiguration + private let logger: Logger + private let isConsumerGroup: Bool + /// One reader at a time, as the Rust consumer's `&mut self` ensures. + private let reading = AsyncMutex() + private let commitWakeups: AsyncStream + private let commitWakeup: AsyncStream.Continuation + + private var isInitialized = false + /// Shared by concurrent calls of ``initialize()``. + private var initialization: Task? + private var isShutDown = false + private var shutdownTask: Task? + /// Follows the client's events: polling waits while the client is signed + /// out and can come back on its own. + private var canPoll = true + private var joinedGroup = false + private var buffer: [IggyMessage] = [] + private var bufferIndex = 0 + private var bufferedContext = PartitionContext.empty + private var bufferedCurrentOffset: UInt64 = 0 + private var lastPolledAt: ContinuousClock.Instant? + /// Where each partition continues after its last message handed over, + /// unless the strategy is `next`, which leaves that to the server. + private var nextPositions: [UInt32: ConsumerPosition] = [:] + private var lastConsumed: [UInt32: ConsumerPosition] = [:] + private var lastStored: [UInt32: ConsumerPosition] = [:] + /// Commits queued by the message triggers, the latest per partition. + private var pendingCommits: [UInt32: ConsumerPosition] = [:] + private var eventsTask: Task? + private var intervalTask: Task? + private var intervalIsStoring = false + private var commitTask: Task? + + init( + client: IggyClient, name: String, consumer: Consumer, streamID: Identifier, topicID: Identifier, partitionID: UInt32?, + configuration: ConsumerConfiguration + ) throws { + try configuration.validate() + self.client = client + self.name = name + self.consumer = consumer + self.streamID = streamID + self.topicID = topicID + self.partitionID = partitionID + currentPartitionID = partitionID ?? 0 + self.configuration = configuration + logger = client.logger + isConsumerGroup = consumer.kind == .consumerGroup + (commitWakeups, commitWakeup) = AsyncStream.makeStream(bufferingPolicy: .bufferingNewest(1)) + } + + deinit { + eventsTask?.cancel() + intervalTask?.cancel() + commitTask?.cancel() + commitWakeup.finish() + } + + public nonisolated func makeAsyncIterator() -> AsyncIterator { + AsyncIterator(consumer: self) + } + + public struct AsyncIterator: AsyncIteratorProtocol { + let consumer: IggyConsumer + + public mutating func next() async throws -> ReceivedMessage? { + try await consumer.next() + } + } + + // MARK: Lifecycle + + /// Looks the stream and topic up, retrying a missing one as + /// ``ConsumerConfiguration/initRetries`` says, joins the group of a member + /// that joins on its own, and starts the commit tasks. Concurrent calls + /// share one run, and calling it again after it succeeded, or after + /// ``shutdown()``, does nothing. + public func initialize() async throws { + if isInitialized || isShutDown { + return + } + if let initialization { + return try await initialization.value + } + let task = Task { try await self.start() } + initialization = task + defer { initialization = nil } + try await task.value + } + + private func start() async throws { + logger.info("Initializing the consumer", metadata: metadata) + try await waitForStreamAndTopic() + let events = await client.events + // A retried initialize after a failed join replaces the subscription + // with nothing suspending in between, so none is left behind. + eventsTask?.cancel() + eventsTask = Task { [weak self] in + for await event in events { + guard let self, await self.observe(event) else { + return + } + } + } + if isConsumerGroup, configuration.autoJoinConsumerGroup { + try await joinGroup() + } + guard !isShutDown else { + // A shutdown came in meanwhile and already stopped what it found. + eventsTask?.cancel() + eventsTask = nil + return + } + if let interval = configuration.autoCommit.interval { + intervalTask = Task { [weak self] in + while (try? await Task.sleep(for: interval)) != nil { + guard let self, await self.storeOnTick() else { + return + } + } + } + } + commitTask = Task { [weak self, commitWakeups] in + for await _ in commitWakeups { + guard let self, await self.sendPendingCommits() else { + return + } + } + } + isInitialized = true + logger.info("The consumer has been initialized", metadata: metadata) + } + + /// Stops reading, sends the queued commits, stores the reading position + /// of every partition read unless auto-commit is disabled, and leaves the + /// group. Every caller waits for the same shutdown. Failures are logged: + /// nothing is left for the caller to undo. + public func shutdown() async { + if let shutdownTask { + await shutdownTask.value + return + } + isShutDown = true + let task = Task { await self.drain() } + shutdownTask = task + await task.value + } + + private func drain() async { + logger.info("Shutting down the consumer", metadata: metadata) + let timeout = configuration.offsetDrainTimeout + if let intervalTask { + // A tick sleeping gets cancelled; one storing finishes first. + if intervalIsStoring { + await Self.wait(for: intervalTask, upTo: timeout) + } + intervalTask.cancel() + } + if let commitTask { + commitWakeup.yield() + await Self.wait(for: commitTask, upTo: timeout) + commitTask.cancel() + } + if configuration.autoCommit != .disabled { + let positions = Array(lastConsumed.values) + // One deadline for every final store, as in the Rust SDK. + _ = try? await withDeadline(.now + client.configuration.requestTimeout) { + await self.storeFinalPositions(positions) + } + } + if isConsumerGroup, joinedGroup { + joinedGroup = false + do { + try await client.leaveConsumerGroup(streamID: streamID, topicID: topicID, groupID: consumer.id) + } catch { + // Expected after an explicit leave or a deleted group. + logger.debug("Failed to leave the consumer group", metadata: metadata.merging(["error": "\(error)"]) { $1 }) + } + } + eventsTask?.cancel() + eventsTask = nil + logger.info("The consumer has been shut down", metadata: metadata) + } + + // MARK: Reading + + /// The next message: from the buffer, or from a poll once it is empty. + /// Throws what a poll or join ran into, after the retry interval, and can + /// be called again after it threw. Nil once the consumer is shut down. + public func next() async throws -> ReceivedMessage? { + try await reading.lock() + defer { reading.unlock() } + if isShutDown { + return nil + } + if bufferIndex < buffer.count { + return handOverBuffered() + } + while !isShutDown { + try Task.checkCancellation() + let polled = try await poll() + if !polled.messages.isEmpty, !isShutDown { + return handOverPolled(polled) + } + } + return nil + } + + /// Hands every message to `handler` until the consumer shuts down, the + /// task is cancelled, or an error ends the reading: a client that is shut + /// down or cannot come back without the caller, or an invalid address. + /// Other errors are logged and reading goes on, and a handler that throws + /// is logged too. The ``AutoCommitAfter`` triggers fire here, after the + /// handler returned or threw. + public func consumeMessages(_ handler: @Sendable (ReceivedMessage) async throws -> Void) async throws { + let after = configuration.autoCommit.afterTrigger + while true { + let received: ReceivedMessage + do { + guard let next = try await next() else { + return + } + received = next + } catch let error as IggyError { + let ends = Self.endsConsumption(error) ? true : !(await client.canRecover) + if ends { + logger.error("Reading stopped", metadata: metadata.merging(["error": "\(error)"]) { $1 }) + throw error + } + logger.error("Failed to read a message", metadata: metadata.merging(["error": "\(error)"]) { $1 }) + continue + } + do { + try await handler(received) + } catch { + logger.error( + "The handler failed", metadata: metadata.merging(["offset": "\(received.message.offset)", "error": "\(error)"]) { $1 }) + } + switch after { + case .consumingEachMessage?: + queueCommit(received.position) + case .consumingEveryNthMessage(let n)?: + if received.message.offset % UInt64(n) == 0 { + queueCommit(received.position) + } + case .consumingAllMessages?: + if received.message.offset == received.currentOffset { + queueCommit(received.position) + } + case nil: + break + } + } + } + + // MARK: Offsets + + /// Stores `offset` for a partition, the current one when nil, under the + /// context of the last message handed over from it. Before any was, it + /// is stored without a context. + public func storeOffset(_ offset: UInt64, partitionID: UInt32? = nil) async throws { + let partition = partitionID ?? currentPartitionID + if let last = lastConsumed[partition] { + try await storePosition(ConsumerPosition(partitionID: partition, offset: offset, context: last.context)) + return + } + try await client.storeConsumerOffset(consumer: consumer, streamID: streamID, topicID: topicID, partitionID: partition, offset: offset) + lastStored[partition] = ConsumerPosition(partitionID: partition, offset: offset, context: .empty) + } + + /// Stores the exact position delivered, as + /// ``ReceivedMessage/position`` gives it, even after later polls. Without + /// ``ConsumerConfiguration/allowReplay`` a position at or below the one + /// already stored in the same partition history is not sent. + public func storePosition(_ position: ConsumerPosition) async throws { + try await store(position, allowReplay: configuration.allowReplay) + } + + /// Deletes the stored offset of a partition, the current one when nil. + /// Unlike the Rust consumer, a group member's nil is not left to the + /// server, since the Swift client routes the request by its partition. + public func deleteOffset(partitionID: UInt32? = nil) async throws { + let partition = partitionID ?? currentPartitionID + // A cancelled delete can still have committed, so a later store must + // not be skipped as covered. + lastStored[partition] = nil + try await client.deleteConsumerOffset(consumer: consumer, streamID: streamID, topicID: topicID, partitionID: partition) + } + + /// The offset of the last message handed over from the partition. + public func lastConsumedOffset(partitionID: UInt32) -> UInt64? { + lastConsumed[partitionID]?.offset + } + + /// The offset this consumer last stored for the partition. + public func lastStoredOffset(partitionID: UInt32) -> UInt64? { + lastStored[partitionID]?.offset + } + + // MARK: Internals + + private var metadata: Logger.Metadata { + ["consumer": "\(name)", "stream": "\(streamID)", "topic": "\(topicID)"] + } + + private func waitForStreamAndTopic() async throws { + let retries = configuration.initRetries ?? 0 + var ticker = RetryTicker(interval: configuration.initRetryInterval) + try await ticker.tick() + var streamExists = try await client.getStream(streamID) != nil + var topicExists = false + if streamExists { + topicExists = try await client.getTopic(streamID: streamID, topicID: topicID) != nil + } + var attempt: UInt32 = 0 + // A stream or topic may still be being created by someone else. + while !(streamExists && topicExists), attempt < retries { + attempt += 1 + logger.warning( + "The stream or topic does not exist yet, retrying", + metadata: metadata.merging(["retry": "\(attempt)/\(retries)", "stream_exists": "\(streamExists)"]) { $1 }) + try await ticker.tick() + if !streamExists { + streamExists = try await client.getStream(streamID) != nil + } + if streamExists { + topicExists = try await client.getTopic(streamID: streamID, topicID: topicID) != nil + } + } + guard streamExists else { + throw streamID.name.map { IggyError(.streamNameNotFound, context: "stream \($0) was not found") } + ?? IggyError(.streamIdNotFound, context: "stream \(streamID) was not found") + } + guard topicExists else { + throw topicID.name.map { IggyError(.topicNameNotFound, context: "topic \($0) was not found in stream \(streamID)") } + ?? IggyError(.topicIdNotFound, context: "topic \(topicID) was not found in stream \(streamID)") + } + } + + private func joinGroup() async throws { + if joinedGroup { + return + } + if try await client.getConsumerGroup(streamID: streamID, topicID: topicID, groupID: consumer.id) == nil { + guard configuration.createConsumerGroupIfNotExists else { + throw IggyError(.consumerGroupNameNotFound, context: "consumer group \(name) does not exist and auto-creation is disabled") + } + logger.info("Creating the consumer group", metadata: metadata) + do { + _ = try await client.createConsumerGroup(streamID: streamID, topicID: topicID, name: name) + } catch let error as IggyError where error.code == .consumerGroupNameAlreadyExists { + // Another member created it first. + } + } + logger.info("Joining the consumer group", metadata: metadata) + try await client.joinConsumerGroup(streamID: streamID, topicID: topicID, groupID: consumer.id) + joinedGroup = true + } + + /// Returns false once the client shut down, which ends the subscription. + private func observe(_ event: DiagnosticEvent) async -> Bool { + switch event { + case .shutdown: + logger.warning("The client has been shut down", metadata: metadata) + joinedGroup = false + canPoll = false + return false + case .connected, .disconnected, .signedOut: + joinedGroup = false + case .signedIn: + break + } + // Events can arrive after one that reports a later state, so the + // state of the client decides. + canPoll = await client.state == .authenticated + return true + } + + /// One poll, as the Rust consumer's poll future: the interval wait, the + /// sign-in wait, a join when the membership was lost, and the request. + /// An empty result is not an error. + private func poll() async throws -> PolledMessages { + if let pollInterval = configuration.pollInterval, let lastPolledAt { + try await Task.sleep(until: lastPolledAt + pollInterval, clock: .continuous) + } + let retryInterval = configuration.pollingRetryInterval + // A client that only the caller brings back never recovers, so the + // poll goes on and fails with the reason instead of waiting for good. + while !canPoll, await client.canRecover { + try await Task.sleep(for: retryInterval) + } + if isConsumerGroup, configuration.autoJoinConsumerGroup, !joinedGroup { + do { + try await joinGroup() + } catch { + logger.error("Failed to join the consumer group", metadata: metadata.merging(["error": "\(error)"]) { $1 }) + try await Task.sleep(for: retryInterval) + throw error + } + } + lastPolledAt = .now + let continuations = nextPositions + let strategy = configuration.pollingStrategy + // A refusal does not name its partition, and it answers the last one + // the strategy was asked for. + let polledPartition = NIOLockedValueBox(PolledMessages.noAssignedPartition) + do { + var polled = try await client.pollMessages( + streamID: streamID, topicID: topicID, partitionID: partitionID, consumer: consumer, count: configuration.batchLength, + autoCommit: configuration.autoCommit.whenTrigger == .pollingMessages + ) { partition in + polledPartition.withLockedValue { $0 = partition } + return continuations[partition].map { .offset($0.offset, context: $0.context) } ?? strategy + } + if polled.partitionID == PolledMessages.noAssignedPartition { + nextPositions.removeAll() + try await Task.sleep(for: retryInterval) + return polled + } + if !configuration.allowReplay, let consumed = lastConsumed[polled.partitionID], Self.sameHistory(consumed.context, polled.context) { + polled.messages.removeAll { $0.offset <= consumed.offset } + polled.count = UInt32(polled.messages.count) + } + return polled + } catch let error as IggyError { + if error.code == .historyUnavailable || error.code == .consumerGroupPartitionNotOwned { + nextPositions[polledPartition.withLockedValue { $0 }] = nil + } + logger.error("Failed to poll messages", metadata: metadata.merging(["error": "\(error)"]) { $1 }) + if isConsumerGroup, configuration.autoJoinConsumerGroup, error.code == .consumerGroupMemberNotFound { + // The membership was revoked; the next poll joins again. + joinedGroup = false + return PolledMessages(partitionID: PolledMessages.noAssignedPartition, currentOffset: 0, count: 0, messages: []) + } + if isConsumerGroup, [.disconnected, .unauthenticated, .staleClient].contains(error.code) { + joinedGroup = false + } + try await Task.sleep(for: retryInterval) + throw error + } + } + + private func handOverPolled(_ polled: PolledMessages) -> ReceivedMessage { + let partition = polled.partitionID + currentPartitionID = partition + buffer = polled.messages + bufferIndex = 0 + bufferedContext = polled.context + bufferedCurrentOffset = polled.currentOffset + return handOverBuffered() + } + + private func handOverBuffered() -> ReceivedMessage { + let partition = currentPartitionID + let message = buffer[bufferIndex] + bufferIndex += 1 + let position = ConsumerPosition(partitionID: partition, offset: message.offset, context: bufferedContext) + lastConsumed[partition] = position + let drained = bufferIndex == buffer.count + // The strategy only says where a partition starts; the next poll of + // it continues after the last message handed over. + if configuration.pollingStrategy.kind != .next { + nextPositions[partition] = ConsumerPosition(partitionID: partition, offset: message.offset + 1, context: bufferedContext) + } + switch configuration.autoCommit.whenTrigger { + case .consumingEachMessage?: + queueCommit(position) + case .consumingEveryNthMessage(let n)?: + if message.offset % UInt64(n) == 0 { + queueCommit(position) + } + case .consumingAllMessages?: + if drained { + queueCommit(position) + } + case .pollingMessages?, nil: + break + } + let received = ReceivedMessage(message: message, currentOffset: bufferedCurrentOffset, partitionID: partition, context: bufferedContext) + if drained { + buffer.removeAll(keepingCapacity: true) + bufferIndex = 0 + } + return received + } + + /// Queues a commit for the commit task. A later one for the same + /// partition replaces a queued one not sent yet. + private func queueCommit(_ position: ConsumerPosition) { + guard isInitialized, !isShutDown else { + logger.error("The commit was not queued because the consumer is not running", metadata: metadata) + return + } + pendingCommits[position.partitionID] = position + commitWakeup.yield() + } + + /// Sends the queued commits. Returns false once the consumer is shut + /// down and nothing is left, which ends the commit task. + private func sendPendingCommits() async -> Bool { + while let partition = pendingCommits.keys.first, let position = pendingCommits.removeValue(forKey: partition) { + await storeInBackground(position) + } + return !isShutDown + } + + /// Returns false once the consumer is shut down, which ends the interval + /// task before it stores anything more. + private func storeOnTick() async -> Bool { + guard !isShutDown else { + return false + } + intervalIsStoring = true + defer { intervalIsStoring = false } + for position in lastConsumed.values { + await storeInBackground(position) + } + return true + } + + private func storeFinalPositions(_ positions: [ConsumerPosition]) async { + for position in positions { + do { + try await store(position, allowReplay: false) + } catch { + logger.warning( + "The final checkpoint was not acknowledged", + metadata: metadata.merging(["partition": "\(position.partitionID)", "offset": "\(position.offset)", "error": "\(error)"]) { $1 }) + } + } + } + + /// Background commits have no caller to report to, so a failure is + /// logged. They never move a stored offset back: replay governs reading. + private func storeInBackground(_ position: ConsumerPosition) async { + do { + try await store(position, allowReplay: false) + } catch { + logger.error( + "Failed to store the offset", + metadata: metadata.merging(["partition": "\(position.partitionID)", "offset": "\(position.offset)", "error": "\(error)"]) { $1 }) + } + } + + private func store(_ position: ConsumerPosition, allowReplay: Bool) async throws { + if !allowReplay, storedPositionCovers(position) { + return + } + try await client.storeConsumerPosition(consumer: consumer, streamID: streamID, topicID: topicID, position: position) + lastStored[position.partitionID] = position + } + + /// A store without a context proves nothing about a partition history. + private func storedPositionCovers(_ position: ConsumerPosition) -> Bool { + guard let stored = lastStored[position.partitionID], stored.context.incarnation != 0, position.context.incarnation != 0 else { + return false + } + return Self.sameHistory(stored.context, position.context) && position.offset <= stored.offset + } + + /// Offsets compare only within one incarnation and owner. The metadata + /// frontier is left out: another group's change can move it between two + /// polls of the same partition. + static func sameHistory(_ left: PartitionContext, _ right: PartitionContext) -> Bool { + left.incarnation == right.incarnation && left.ownerGeneration == right.ownerGeneration + } + + static func endsConsumption(_ error: IggyError) -> Bool { + switch error.code { + case .invalidServerAddress, .invalidClientAddress, .clientShutdown: true + default: false + } + } + + private static func wait(for task: Task, upTo timeout: Duration) async { + _ = try? await withDeadline(.now + timeout) { + await task.value + } + } +} + +extension IggyClient { + /// A standalone consumer of one partition, identified by `name`. Building + /// it never talks to the server; ``IggyConsumer/initialize()`` does. + /// Throws ``IggyErrorCode/invalidConfiguration`` for an option out of + /// range. + public func consumer( + name: String, streamID: Identifier, topicID: Identifier, partitionID: UInt32, configuration: ConsumerConfiguration = ConsumerConfiguration() + ) throws -> IggyConsumer { + try IggyConsumer( + client: self, name: name, consumer: .consumer(try Identifier(named: name)), streamID: streamID, topicID: topicID, partitionID: partitionID, + configuration: configuration) + } + + /// A member of the consumer group `name`, which reads the partitions the + /// server assigns to it. Building it never talks to the server; + /// ``IggyConsumer/initialize()`` does. + public func consumerGroup( + name: String, streamID: Identifier, topicID: Identifier, configuration: ConsumerConfiguration = ConsumerConfiguration() + ) throws -> IggyConsumer { + try IggyConsumer( + client: self, name: name, consumer: .group(try Identifier(named: name)), streamID: streamID, topicID: topicID, partitionID: nil, + configuration: configuration) + } +} diff --git a/foreign/swift/Tests/IggyE2ETests/ConsumerE2ETests.swift b/foreign/swift/Tests/IggyE2ETests/ConsumerE2ETests.swift new file mode 100644 index 0000000000..af7620a386 --- /dev/null +++ b/foreign/swift/Tests/IggyE2ETests/ConsumerE2ETests.swift @@ -0,0 +1,100 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +import Foundation +import Iggy +import Testing + +/// Reads `count` messages, failing the test at `timeout`: an empty poll is +/// not the end of a consumer, so reading alone never stops. +private func read(_ consumer: IggyConsumer, _ count: Int, timeout: Duration = .seconds(30)) async throws -> [ReceivedMessage] { + try await withThrowingTaskGroup(of: [ReceivedMessage]?.self) { group in + group.addTask { + var received: [ReceivedMessage] = [] + while received.count < count, let message = try await consumer.next() { + received.append(message) + } + return received + } + group.addTask { + try await Task.sleep(for: timeout) + return nil + } + defer { group.cancelAll() } + guard let received = try await group.next() ?? nil else { + Issue.record("read fewer than \(count) messages within \(timeout)") + return [] + } + return received + } +} + +@Suite("Consumer", requiresServer) +struct ConsumerE2ETests { + /// The default auto-commit stores with every poll, so a second consumer + /// of the same name resumes after what the first one read. + @Test func aConsumerResumesAfterTheOffsetItStored() async throws { + try await E2EEnvironment.withSession { session in + let client = session.client + let scratch = try await session.scratchTopic(partitions: 1) + let name = E2EEnvironment.uniqueName("swift-e2e-reader") + var batch = try E2EEnvironment.messages(5) + try await client.sendMessages(streamID: scratch.stream, topicID: scratch.topic, partitioning: .partition(0), messages: &batch) + + let first = try client.consumer(name: name, streamID: scratch.stream, topicID: scratch.topic, partitionID: 0) + try await first.initialize() + let received = try await read(first, 5) + #expect(received.map(\.message.payloadString) == (0..<5).map { "message \($0)" }) + #expect(received.map(\.partitionID) == Array(repeating: 0, count: 5)) + #expect(received.last?.currentOffset == 4) + await first.shutdown() + + var more = try E2EEnvironment.messages(2, prefix: "more") + try await client.sendMessages(streamID: scratch.stream, topicID: scratch.topic, partitioning: .partition(0), messages: &more) + let second = try client.consumer(name: name, streamID: scratch.stream, topicID: scratch.topic, partitionID: 0) + try await second.initialize() + #expect(try await read(second, 2).map(\.message.offset) == [5, 6]) + await second.shutdown() + } + } + + /// A member joins its group, reads the partitions the server assigns, + /// and leaves the group on shutdown. + @Test func aGroupMemberReadsEveryAssignedPartition() async throws { + try await E2EEnvironment.withSession { session in + let client = session.client + let scratch = try await session.scratchTopic(partitions: 2) + for partition: UInt32 in [0, 1] { + var batch = try E2EEnvironment.messages(2, prefix: "partition \(partition)") + try await client.sendMessages(streamID: scratch.stream, topicID: scratch.topic, partitioning: .partition(partition), messages: &batch) + } + let name = E2EEnvironment.uniqueName("swift-e2e-group") + let member = try client.consumerGroup(name: name, streamID: scratch.stream, topicID: scratch.topic) + try await member.initialize() + let received = try await read(member, 4) + #expect(Set(received.map(\.partitionID)) == [0, 1]) + #expect(Set(received.map(\.message.payloadString)) == ["partition 0 0", "partition 0 1", "partition 1 0", "partition 1 1"]) + await member.shutdown() + let groupID = try Identifier(named: name) + // The server applies the leave asynchronously. + #expect( + try await E2EEnvironment.eventually { + try await client.getConsumerGroup(streamID: scratch.stream, topicID: scratch.topic, groupID: groupID)?.membersCount == 0 + }) + } + } +} diff --git a/foreign/swift/Tests/IggyTests/ConsumerTests.swift b/foreign/swift/Tests/IggyTests/ConsumerTests.swift new file mode 100644 index 0000000000..5d6fd7742e --- /dev/null +++ b/foreign/swift/Tests/IggyTests/ConsumerTests.swift @@ -0,0 +1,532 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +import Foundation +import Testing + +@testable import Iggy + +private func signedInClient(_ server: FakeVSRServer, requestTimeout: Duration = .seconds(10)) async throws -> IggyClient { + var configuration = ClientConfiguration(address: server.address, autoLogin: .usernamePassword(username: "iggy", password: "iggy")) + configuration.reconnection = ReconnectionOptions(maxRetries: nil, interval: .milliseconds(20), reestablishAfter: .zero) + configuration.requestTimeout = requestTimeout + configuration.connectTimeout = .seconds(2) + configuration.heartbeatInterval = .seconds(60) + let client = IggyClient(configuration: configuration, connector: NIOConnector(), logger: testLogger) + try await client.connect() + return client +} + +/// Answers the lookups a consumer makes before it reads: the stream, the +/// topic and the group exist. `poll` answers every poll. +private func serve(_ server: FakeVSRServer, poll: @escaping @Sendable (FakeVSRServer.Request) async -> FakeVSRServer.Reply) async { + await server.respond { request in + switch request.code.flatMap(CommandCode.init(rawValue:)) { + case .getStream: return .ok(GoldenFixture.shared.bytes("response.get_stream")) + case .getTopic: return .ok(GoldenFixture.shared.bytes("response.get_topic")) + case .getConsumerGroup, .createConsumerGroup: return .ok(GoldenFixture.shared.bytes("response.consumer_group_details")) + case .syncConsumerGroup: return .ok(FakeVSRServer.assignmentBody(generation: 1, partitions: [1])) + case .pollMessages: return await poll(request) + default: return FakeVSRServer.defaultReply(request, address: server.address) + } + } +} + +/// Answers the polls with the batches in turn, then with empty polls. +private func batches( + _ batches: [[UInt64]], partitionID: UInt32 = 1, context: PartitionContext = FakeVSRServer.routeContext +) -> @Sendable (FakeVSRServer.Request) async -> FakeVSRServer.Reply { + let next = Locked(0) + return { _ in + let index = next.withLock { index in + defer { index += 1 } + return index + } + let offsets = index < batches.count ? batches[index] : [] + return .ok(FakeVSRServer.polledBody(partitionID: partitionID, offsets: offsets, context: context)) + } +} + +private func requests(_ server: FakeVSRServer, _ code: CommandCode) async -> [FakeVSRServer.Request] { + await server.requests.filter { $0.code == code.rawValue } +} + +/// The strategy and auto-commit flag a poll ends with: +/// `[kind:1][value:8][count:4][auto_commit:1]`. +private func polledStrategy(_ request: FakeVSRServer.Request) -> (kind: UInt8, value: UInt64, autoCommit: Bool) { + let end = request.payload.count + return (request.payload[end - 14], UInt64(littleEndianBytes: request.payload[end - 13.. UInt64 { + let end = request.payload.count + return UInt64(littleEndianBytes: request.payload[end - 9.. [UInt64] { + var offsets: [UInt64] = [] + while offsets.count < count, let received = try await consumer.next() { + offsets.append(received.message.offset) + } + return offsets +} + +private final class WeakConsumer: @unchecked Sendable { + weak var value: IggyConsumer? +} + +@Suite("Consumer", .serialized) +struct ConsumerTests { + // MARK: Reading + + /// A batch is handed over one message at a time, and the next poll of the + /// partition continues after the last one handed over, under the context + /// that delivered it. + @Test func aPollContinuesAfterTheLastMessageHandedOver() async throws { + let server = try await FakeVSRServer.started() + defer { Task { await server.stop() } } + await serve(server, poll: batches([[0, 1, 2], [3, 4]])) + let client = try await signedInClient(server) + let configuration = ConsumerConfiguration(pollingStrategy: .offset(0), batchLength: 3, autoCommit: .disabled) + let consumer = try client.consumer(name: "reader", streamID: 1, topicID: 2, partitionID: 1, configuration: configuration) + try await consumer.initialize() + #expect(try await read(consumer, 5) == [0, 1, 2, 3, 4]) + let polls = await requests(server, .pollMessages) + #expect(polls.prefix(2).map { polledStrategy($0).value } == [0, 3]) + #expect(polls[1].context == FakeVSRServer.routeContext) + #expect(await consumer.currentPartitionID == 1) + #expect(await consumer.lastConsumedOffset(partitionID: 1) == 4) + await consumer.shutdown() + await client.shutdown() + } + + @Test func nextLeavesTheContinuationToTheServer() async throws { + let server = try await FakeVSRServer.started() + defer { Task { await server.stop() } } + await serve(server, poll: batches([[0, 1], [2]])) + let client = try await signedInClient(server) + let consumer = try client.consumer( + name: "reader", streamID: 1, topicID: 2, partitionID: 1, configuration: ConsumerConfiguration(autoCommit: .disabled)) + try await consumer.initialize() + #expect(try await read(consumer, 3) == [0, 1, 2]) + let kinds = await requests(server, .pollMessages).prefix(2).map { polledStrategy($0).kind } + #expect(kinds == [PollingStrategy.Kind.next.rawValue, PollingStrategy.Kind.next.rawValue]) + await consumer.shutdown() + await client.shutdown() + } + + /// A message at or below the last one handed over from the same + /// partition history is skipped, unless replay is allowed. Another + /// incarnation is another history. + @Test func aReturnedMessageIsHandedOverAgainOnlyWithReplay() async throws { + let renewed = PartitionContext(incarnation: 8, ownerGeneration: 0, metadataOp: 11) + for (allowReplay, expected) in [(false, [UInt64(0), 1, 2, 1]), (true, [0, 1, 1, 2, 1])] { + let server = try await FakeVSRServer.started() + let returned = Locked(0) + await serve(server) { _ in + let poll = returned.withLock { count in + defer { count += 1 } + return count + } + switch poll { + case 0: return .ok(FakeVSRServer.polledBody(partitionID: 1, offsets: [0, 1])) + case 1: return .ok(FakeVSRServer.polledBody(partitionID: 1, offsets: [1, 2])) + case 2: return .ok(FakeVSRServer.polledBody(partitionID: 1, offsets: [1], context: renewed)) + default: return .ok(FakeVSRServer.polledBody(partitionID: 1, offsets: [])) + } + } + let client = try await signedInClient(server) + let configuration = ConsumerConfiguration(autoCommit: .disabled, allowReplay: allowReplay) + let consumer = try client.consumer(name: "reader", streamID: 1, topicID: 2, partitionID: 1, configuration: configuration) + try await consumer.initialize() + #expect(try await read(consumer, expected.count) == expected) + await consumer.shutdown() + await client.shutdown() + await server.stop() + } + } + + /// A failed poll is thrown after the retry interval and the consumer + /// keeps reading on the next call. + @Test func aFailedPollIsThrownAndReadingGoesOn() async throws { + let server = try await FakeVSRServer.started() + defer { Task { await server.stop() } } + let failed = Locked(false) + await serve(server) { _ in + failed.exchange(true) ? .ok(FakeVSRServer.polledBody(partitionID: 1, offsets: [0])) : .status(IggyErrorCode.partitionNotFound.rawValue) + } + let client = try await signedInClient(server) + let configuration = ConsumerConfiguration(autoCommit: .disabled, pollingRetryInterval: .milliseconds(150)) + let consumer = try client.consumer(name: "reader", streamID: 1, topicID: 2, partitionID: 1, configuration: configuration) + try await consumer.initialize() + let clock = ContinuousClock() + let elapsed = await clock.measure { + await expectCode(.partitionNotFound) { _ = try await consumer.next() } + } + #expect(elapsed >= .milliseconds(140)) + #expect(try await consumer.next()?.message.offset == 0) + await consumer.shutdown() + await client.shutdown() + } + + /// A refused continuation is dropped, so the next poll of the partition + /// starts where the strategy says again. + @Test func aRecreatedPartitionDropsItsContinuation() async throws { + let server = try await FakeVSRServer.started() + defer { Task { await server.stop() } } + let poll = Locked(0) + await serve(server) { _ in + let index = poll.withLock { index in + defer { index += 1 } + return index + } + switch index { + case 0: return .ok(FakeVSRServer.polledBody(partitionID: 1, offsets: [5])) + case 1: return .status(IggyErrorCode.historyUnavailable.rawValue) + default: return .ok(FakeVSRServer.polledBody(partitionID: 1, offsets: [])) + } + } + let client = try await signedInClient(server) + let configuration = ConsumerConfiguration(pollingStrategy: .first, autoCommit: .disabled, pollingRetryInterval: .milliseconds(10)) + let consumer = try client.consumer(name: "reader", streamID: 1, topicID: 2, partitionID: 1, configuration: configuration) + try await consumer.initialize() + #expect(try await consumer.next()?.message.offset == 5) + await expectCode(.historyUnavailable) { _ = try await consumer.next() } + let reader = Task { try await consumer.next() } + #expect(await eventually { await requests(server, .pollMessages).count >= 3 }) + let kinds = await requests(server, .pollMessages).prefix(3).map { polledStrategy($0).kind } + #expect(kinds == [PollingStrategy.Kind.first.rawValue, PollingStrategy.Kind.offset.rawValue, PollingStrategy.Kind.first.rawValue]) + await consumer.shutdown() + _ = try? await reader.value + await client.shutdown() + } + + // MARK: Committing + + @Test func pollingMessagesCommitsWithThePollItself() async throws { + let server = try await FakeVSRServer.started() + defer { Task { await server.stop() } } + await serve(server, poll: batches([[0, 1]])) + let client = try await signedInClient(server) + let consumer = try client.consumer(name: "reader", streamID: 1, topicID: 2, partitionID: 1) + try await consumer.initialize() + #expect(try await read(consumer, 2) == [0, 1]) + #expect(await requests(server, .pollMessages).first.map { polledStrategy($0).autoCommit } == true) + await consumer.shutdown() + await client.shutdown() + } + + /// The message triggers queue a commit as messages are handed over: + /// every message, an offset that divides by n, or the end of a batch. + @Test func messageTriggersCommitAsMessagesAreHandedOver() async throws { + let cases: [(AutoCommitWhen, [UInt64])] = [ + (.consumingEachMessage, [0, 1, 2, 3]), (.consumingEveryNthMessage(2), [0, 2]), (.consumingAllMessages, [2, 3]), + ] + for (trigger, expected) in cases { + let server = try await FakeVSRServer.started() + await serve(server, poll: batches([[0, 1, 2], [3]])) + let client = try await signedInClient(server) + let consumer = try client.consumer( + name: "reader", streamID: 1, topicID: 2, partitionID: 1, configuration: ConsumerConfiguration(autoCommit: .when(trigger))) + try await consumer.initialize() + #expect(try await read(consumer, 4) == [0, 1, 2, 3]) + #expect(await requests(server, .pollMessages).first.map { polledStrategy($0).autoCommit } == false) + // Queued commits collapse to the latest per partition, so wait + // for the last one and check every store is one of the expected. + #expect(await eventually { await requests(server, .storeConsumerOffset).map(storedOffset).last == expected.last }) + let stored = await requests(server, .storeConsumerOffset).map(storedOffset) + #expect(stored.allSatisfy(expected.contains), "\(trigger) stored \(stored)") + #expect(await requests(server, .storeConsumerOffset).allSatisfy { $0.context == FakeVSRServer.routeContext }) + await consumer.shutdown() + await client.shutdown() + await server.stop() + } + } + + @Test func anIntervalStoresTheReadingPositionOfEveryPartitionRead() async throws { + let server = try await FakeVSRServer.started() + defer { Task { await server.stop() } } + await serve(server, poll: batches([[0, 1]])) + let client = try await signedInClient(server) + let consumer = try client.consumer( + name: "reader", streamID: 1, topicID: 2, partitionID: 1, configuration: ConsumerConfiguration(autoCommit: .interval(.milliseconds(50)))) + try await consumer.initialize() + #expect(try await read(consumer, 2) == [0, 1]) + #expect(await eventually { await requests(server, .storeConsumerOffset).map(storedOffset).contains(1) }) + // A position already stored in the same history is not sent again. + let stores = await requests(server, .storeConsumerOffset).count + try await Task.sleep(for: .milliseconds(150)) + #expect(await requests(server, .storeConsumerOffset).count == stores) + await consumer.shutdown() + await client.shutdown() + } + + /// A store by hand keeps the context of the last message handed over, + /// and without replay a position the stored one covers is not sent. + @Test func aStoreByHandKeepsTheContextAndSkipsCoveredPositions() async throws { + let server = try await FakeVSRServer.started() + defer { Task { await server.stop() } } + await serve(server, poll: batches([[4, 5]])) + let client = try await signedInClient(server) + let consumer = try client.consumer( + name: "reader", streamID: 1, topicID: 2, partitionID: 1, configuration: ConsumerConfiguration(autoCommit: .disabled)) + // Before anything was read the store has no captured context, so it + // covers nothing later. + try await consumer.storeOffset(1) + try await consumer.initialize() + let first = try #require(try await consumer.next()) + _ = try await consumer.next() + try await consumer.storeOffset(5) + try await consumer.storePosition(first.position) + let stores = await requests(server, .storeConsumerOffset) + // The position at 4 is covered by the store at 5 and is not sent. + #expect(stores.map(storedOffset) == [1, 5]) + #expect(await consumer.lastStoredOffset(partitionID: 1) == 5) + // After a delete the same position goes again. + try await consumer.deleteOffset() + try await consumer.storeOffset(5) + #expect(await requests(server, .storeConsumerOffset).count == 3) + #expect(await requests(server, .deleteConsumerOffset).count == 1) + await consumer.shutdown() + await client.shutdown() + } + + /// The after triggers fire once the handler returned, whether it + /// succeeded or threw: every message, an offset that divides by n, or the + /// message at the partition's newest offset. + @Test func consumeMessagesCommitsAfterTheHandler() async throws { + let cases: [(AutoCommitAfter, [UInt64])] = [ + (.consumingEachMessage, [0, 1, 2, 3]), (.consumingEveryNthMessage(3), [0, 3]), (.consumingAllMessages, [3]), + ] + for (trigger, expected) in cases { + let server = try await FakeVSRServer.started() + await serve(server, poll: batches([[0, 1, 2, 3]])) + let client = try await signedInClient(server) + let consumer = try client.consumer( + name: "reader", streamID: 1, topicID: 2, partitionID: 1, configuration: ConsumerConfiguration(autoCommit: .after(trigger))) + try await consumer.initialize() + struct Refused: Error {} + let handled = Locked<[UInt64]>([]) + let consuming = Task { + try await consumer.consumeMessages { received in + handled.withLock { $0.append(received.message.offset) } + // Slow enough that every commit goes out before the + // next one is queued and could replace it. + try? await Task.sleep(for: .milliseconds(30)) + if received.message.offset == 3 { + throw Refused() + } + } + } + #expect(await eventually { await requests(server, .storeConsumerOffset).map(storedOffset).last == 3 }) + #expect(handled.withLock { $0 } == [0, 1, 2, 3]) + #expect(await requests(server, .storeConsumerOffset).map(storedOffset) == expected, "\(trigger)") + await consumer.shutdown() + try await consuming.value + await client.shutdown() + await server.stop() + } + } + + @Test func consumeMessagesEndsWhenTheClientCannotComeBack() async throws { + let server = try await FakeVSRServer.started() + defer { Task { await server.stop() } } + await serve(server, poll: batches([])) + let client = try await signedInClient(server) + let configuration = ConsumerConfiguration(autoCommit: .disabled, pollingRetryInterval: .milliseconds(10)) + let consumer = try client.consumer(name: "reader", streamID: 1, topicID: 2, partitionID: 1, configuration: configuration) + try await consumer.initialize() + await client.disconnect() + await #expect(throws: IggyError.self) { try await consumer.consumeMessages { _ in } } + await consumer.shutdown() + await client.shutdown() + } + + // MARK: Groups + + /// A member joins on initialize, creating the group when it is missing, + /// and joins again once a poll finds the membership gone. + @Test func aGroupMemberJoinsAndJoinsAgainAfterLosingTheMembership() async throws { + let server = try await FakeVSRServer.started() + defer { Task { await server.stop() } } + let revoked = Locked(false) + let groupExists = Locked(false) + await server.respond { request in + switch request.code.flatMap(CommandCode.init(rawValue:)) { + case .getStream: return .ok(GoldenFixture.shared.bytes("response.get_stream")) + case .getTopic: return .ok(GoldenFixture.shared.bytes("response.get_topic")) + case .getConsumerGroup: return groupExists.withLock { $0 } ? .ok(GoldenFixture.shared.bytes("response.consumer_group_details")) : .ok([]) + case .createConsumerGroup: + groupExists.withLock { $0 = true } + return .ok(GoldenFixture.shared.bytes("response.consumer_group_details")) + case .syncConsumerGroup: return .ok(FakeVSRServer.assignmentBody(generation: 1, partitions: [1])) + case .pollMessages: + if !revoked.exchange(true) { + return .status(IggyErrorCode.consumerGroupMemberNotFound.rawValue) + } + return .ok(FakeVSRServer.polledBody(partitionID: 1, offsets: [0])) + default: return FakeVSRServer.defaultReply(request, address: server.address) + } + } + let client = try await signedInClient(server) + let consumer = try client.consumerGroup( + name: "billing", streamID: 1, topicID: 2, configuration: ConsumerConfiguration(autoCommit: .disabled, pollingRetryInterval: .milliseconds(10))) + try await consumer.initialize() + #expect(await requests(server, .createConsumerGroup).count == 1) + #expect(try await consumer.next()?.message.offset == 0) + #expect(await requests(server, .joinConsumerGroup).count == 2) + await consumer.shutdown() + #expect(await requests(server, .leaveConsumerGroup).count == 1) + await client.shutdown() + } + + @Test func aMissingGroupIsNotCreatedWhenAutoCreationIsOff() async throws { + let server = try await FakeVSRServer.started() + defer { Task { await server.stop() } } + await server.respond { request in + switch request.code.flatMap(CommandCode.init(rawValue:)) { + case .getStream: return .ok(GoldenFixture.shared.bytes("response.get_stream")) + case .getTopic: return .ok(GoldenFixture.shared.bytes("response.get_topic")) + default: return FakeVSRServer.defaultReply(request, address: server.address) + } + } + let client = try await signedInClient(server) + let consumer = try client.consumerGroup( + name: "billing", streamID: 1, topicID: 2, configuration: ConsumerConfiguration(createConsumerGroupIfNotExists: false)) + await expectCode(.consumerGroupNameNotFound) { try await consumer.initialize() } + #expect(await requests(server, .createConsumerGroup).isEmpty) + await client.shutdown() + } + + // MARK: Lifecycle + + @Test func initializeWaitsForAMissingTopicWithinItsRetries() async throws { + for (retries, succeeds) in [(UInt32(3), true), (1, false)] { + let server = try await FakeVSRServer.started() + let lookups = Locked(0) + await server.respond { request in + switch request.code.flatMap(CommandCode.init(rawValue:)) { + case .getStream: return .ok(GoldenFixture.shared.bytes("response.get_stream")) + case .getTopic: + let lookup = lookups.withLock { + $0 += 1; return $0 + } + return lookup > 2 ? .ok(GoldenFixture.shared.bytes("response.get_topic")) : .ok([]) + default: return FakeVSRServer.defaultReply(request, address: server.address) + } + } + let client = try await signedInClient(server) + let configuration = ConsumerConfiguration(initRetries: retries, initRetryInterval: .milliseconds(10)) + let consumer = try client.consumer(name: "reader", streamID: 1, topicID: "created", partitionID: 1, configuration: configuration) + if succeeds { + try await consumer.initialize() + } else { + await expectCode(.topicNameNotFound) { try await consumer.initialize() } + } + #expect(lookups.withLock { $0 } == (succeeds ? 3 : 2)) + await consumer.shutdown() + await client.shutdown() + await server.stop() + } + } + + /// Shutdown sends the reading position that was not stored yet, and a + /// second shutdown waits for the first. Reading ends with nil. + @Test func shutdownStoresTheReadingPositionOnce() async throws { + let server = try await FakeVSRServer.started() + defer { Task { await server.stop() } } + let poll = batches([[0, 1]]) + await server.respond { [poll] request in + switch request.code.flatMap(CommandCode.init(rawValue:)) { + case .getStream: return .ok(GoldenFixture.shared.bytes("response.get_stream")) + case .getTopic: return .ok(GoldenFixture.shared.bytes("response.get_topic")) + case .pollMessages: return await poll(request) + case .storeConsumerOffset: + try? await Task.sleep(for: .milliseconds(200)) + return .ok([]) + default: return FakeVSRServer.defaultReply(request, address: server.address) + } + } + let client = try await signedInClient(server) + let consumer = try client.consumer( + name: "reader", streamID: 1, topicID: 2, partitionID: 1, configuration: ConsumerConfiguration(autoCommit: .interval(.seconds(60)))) + try await consumer.initialize() + #expect(try await read(consumer, 2) == [0, 1]) + let first = Task { await consumer.shutdown() } + #expect(await eventually { await !requests(server, .storeConsumerOffset).isEmpty }) + await consumer.shutdown() + #expect(await consumer.lastStoredOffset(partitionID: 1) == 1) + await first.value + #expect(await requests(server, .storeConsumerOffset).map(storedOffset) == [1]) + #expect(try await consumer.next() == nil) + await client.shutdown() + } + + /// Concurrent initializations share one run: one lookup, one join. + @Test func concurrentInitializationsShareOneRun() async throws { + let server = try await FakeVSRServer.started() + defer { Task { await server.stop() } } + await serve(server, poll: batches([[0]])) + let client = try await signedInClient(server) + let consumer = try client.consumerGroup(name: "billing", streamID: 1, topicID: 2, configuration: ConsumerConfiguration(autoCommit: .disabled)) + async let first: Void = consumer.initialize() + async let second: Void = consumer.initialize() + _ = try await (first, second) + try await consumer.initialize() + #expect(await requests(server, .getStream).count == 1) + #expect(await requests(server, .joinConsumerGroup).count == 1) + #expect(try await consumer.next()?.message.offset == 0) + await consumer.shutdown() + await client.shutdown() + } + + @Test func aDroppedConsumerEndsItsTasks() async throws { + let server = try await FakeVSRServer.started() + defer { Task { await server.stop() } } + await serve(server, poll: batches([])) + let client = try await signedInClient(server) + let weak = WeakConsumer() + do { + let consumer = try client.consumer( + name: "reader", streamID: 1, topicID: 2, partitionID: 1, configuration: ConsumerConfiguration(autoCommit: .interval(.milliseconds(10)))) + try await consumer.initialize() + weak.value = consumer + } + #expect(await eventually { weak.value == nil }) + await client.shutdown() + } + + @Test func outOfRangeOptionsAreRefusedWhenTheConsumerIsBuilt() async throws { + let server = try await FakeVSRServer.started() + defer { Task { await server.stop() } } + let client = try await signedInClient(server) + let invalid = [ + ConsumerConfiguration(batchLength: 0), ConsumerConfiguration(autoCommit: .interval(.zero)), + ConsumerConfiguration(autoCommit: .when(.consumingEveryNthMessage(0))), ConsumerConfiguration(autoCommit: .after(.consumingEveryNthMessage(0))), + ConsumerConfiguration(pollingRetryInterval: .zero), ConsumerConfiguration(initRetryInterval: .milliseconds(-1)), + ConsumerConfiguration(pollInterval: .milliseconds(-1)), ConsumerConfiguration(offsetDrainTimeout: ProducerConfiguration.maxDuration * 2), + ] + for configuration in invalid { + await expectCode(.invalidConfiguration) { + _ = try client.consumer(name: "reader", streamID: 1, topicID: 2, partitionID: 1, configuration: configuration) + } + } + _ = try client.consumerGroup(name: "billing", streamID: 1, topicID: 2, configuration: ConsumerConfiguration(pollInterval: .zero)) + await client.shutdown() + } +} diff --git a/foreign/swift/Tests/IggyTests/Support/FakeVSRServer.swift b/foreign/swift/Tests/IggyTests/Support/FakeVSRServer.swift index dfbbed2152..720d91a311 100644 --- a/foreign/swift/Tests/IggyTests/Support/FakeVSRServer.swift +++ b/foreign/swift/Tests/IggyTests/Support/FakeVSRServer.swift @@ -199,6 +199,38 @@ final class FakeVSRServer: @unchecked Sendable { return writer.bytes } + /// A poll reply carrying one batch with a message at each of `offsets`, + /// whose payload is `message `. Checksums are left zero, which a + /// reply decoded without verification accepts. + static func polledBody(partitionID: UInt32, offsets: [UInt64], currentOffset: UInt64? = nil, context: PartitionContext = routeContext) -> [UInt8] { + var writer = ByteWriter() + writer.write(partitionID) + writer.write(currentOffset ?? offsets.max() ?? 0) + writer.write(UInt32(offsets.count)) + context.encode(into: &writer) + guard let base = offsets.first else { + return writer.bytes + } + var frames = ByteWriter() + for offset in offsets { + let payload = Array("message \(offset)".utf8) + frames.write(UInt64(0)) + frames.write(MessageID(0)) + frames.write(UInt32(offset - base)) + frames.write(UInt32(0)) + frames.write(UInt32(0)) + frames.write(UInt32(payload.count)) + frames.write(UInt64(0)) + frames.write(payload) + } + let header = Batch.Header( + partitionID: UInt64(partitionID), baseOffset: base, baseTimestamp: 0, originTimestamp: 0, + batchLength: UInt64(Batch.headerSize + frames.count), batchChecksum: 0, messageCount: UInt32(offsets.count)) + writer.write(header.encode()) + writer.write(frames.bytes) + return writer.bytes + } + /// The reply to `SyncConsumerGroup` for a member. static func assignmentBody(generation: UInt64, partitions: [UInt32]) -> [UInt8] { var writer = ByteWriter() From d17d269913cca282b0a79dc05220a26cf65b8c53 Mon Sep 17 00:00:00 2001 From: Hubert Gruszecki Date: Mon, 26 Oct 2026 16:23:39 +0100 Subject: [PATCH 2/2] fix(swift): close consumer races at shutdown and in offset stores A consumer outlived its shutdown. A join in flight ended after the final leave. Without a heartbeat, a poll that waited for a lost client never woke. The drain also waited past offsetDrainTimeout. A tick that stored slept one more interval, and a waiter that gives up on task.value does not cancel the task. Joins and the final leave now share a lock. The poll wait now signs the client in and ends at shutdown. A tick ends after its store, and the drain cancels a task at its deadline. Overlapping stores moved the stored offset back, so stores and deletes now run one at a time. Under the next strategy, a poll that did not commit returned the same offset again and stalled the read. The consumer now continues after the last message, unless the poll commits. The iterator ends at its first error, as AsyncIteratorProtocol requires. --- foreign/swift/README.md | 7 +- .../Iggy/Consumer/ConsumerConfiguration.swift | 5 +- .../Sources/Iggy/Consumer/IggyConsumer.swift | 119 ++++++-- .../swift/Tests/IggyTests/ConsumerTests.swift | 260 ++++++++++++++++-- 4 files changed, 342 insertions(+), 49 deletions(-) diff --git a/foreign/swift/README.md b/foreign/swift/README.md index b8b8fd1fac..a7f2705ca0 100644 --- a/foreign/swift/README.md +++ b/foreign/swift/README.md @@ -337,8 +337,11 @@ for try await received in consumer { } ``` -A poll refills the buffer once it runs empty, and each poll of a partition continues -after the last message handed over from it. A failed poll is thrown after the retry +A poll refills the buffer once it runs empty. `pollingStrategy` says where reading a +partition starts, and each later poll of it continues after the last message handed over +from it. The exception is `.next` with an `autoCommit` that commits with every poll, as in +the default configuration. That poll starts after the offset that the previous poll +stored. A failed poll is thrown after the retry interval and leaves the consumer usable, so a caller of `next()` can call it again, and `consumeMessages(_:)` logs it and keeps reading until the client cannot come back on its own. Without `allowReplay` a message at or below the last one handed over from the same diff --git a/foreign/swift/Sources/Iggy/Consumer/ConsumerConfiguration.swift b/foreign/swift/Sources/Iggy/Consumer/ConsumerConfiguration.swift index 0a2b622f43..16a80962fd 100644 --- a/foreign/swift/Sources/Iggy/Consumer/ConsumerConfiguration.swift +++ b/foreign/swift/Sources/Iggy/Consumer/ConsumerConfiguration.swift @@ -88,8 +88,9 @@ public enum AutoCommitAfter: Sendable, Hashable { public struct ConsumerConfiguration: Sendable { /// Where reading a partition starts. After the first batch each poll /// continues after the last message handed over from that partition, - /// except under ``PollingStrategy/next``, which leaves the continuation - /// to the offsets stored on the server. + /// except a ``PollingStrategy/next`` poll that commits with + /// ``AutoCommitWhen/pollingMessages``: it starts after the offset the + /// previous poll stored. public var pollingStrategy: PollingStrategy /// Messages asked for per poll. public var batchLength: UInt32 diff --git a/foreign/swift/Sources/Iggy/Consumer/IggyConsumer.swift b/foreign/swift/Sources/Iggy/Consumer/IggyConsumer.swift index b7590bfd72..2d42beee53 100644 --- a/foreign/swift/Sources/Iggy/Consumer/IggyConsumer.swift +++ b/foreign/swift/Sources/Iggy/Consumer/IggyConsumer.swift @@ -40,7 +40,8 @@ import NIOConcurrencyHelpers /// failed poll is thrown after ``ConsumerConfiguration/pollingRetryInterval`` /// and leaves the consumer usable, so a caller of ``next()`` can call it again; /// a `for try await` loop ends at the first error instead. ``next()`` returns -/// nil only after ``shutdown()``. +/// nil only after ``shutdown()``. One reader consumes a consumer, as in the +/// Rust SDK: two loops over the same consumer split its messages between them. /// /// The consumer remembers the last message handed over from each partition. /// Without ``ConsumerConfiguration/allowReplay`` a message at or below it is @@ -67,8 +68,15 @@ public actor IggyConsumer: AsyncSequence { private let configuration: ConsumerConfiguration private let logger: Logger private let isConsumerGroup: Bool - /// One reader at a time, as the Rust consumer's `&mut self` ensures. + /// One poll at a time: a reader that finds the buffer empty waits here + /// while another one polls. private let reading = AsyncMutex() + /// Held by a join and by the final leave, so a join in flight ends + /// before the leave and none starts after it. + private let membership = AsyncMutex() + /// Held by every store and delete, so a coverage check, its request, and + /// the cache update see no other store in between. + private let offsetWrites = AsyncMutex() private let commitWakeups: AsyncStream private let commitWakeup: AsyncStream.Continuation @@ -87,7 +95,7 @@ public actor IggyConsumer: AsyncSequence { private var bufferedCurrentOffset: UInt64 = 0 private var lastPolledAt: ContinuousClock.Instant? /// Where each partition continues after its last message handed over, - /// unless the strategy is `next`, which leaves that to the server. + /// unless a `next` poll commits, which leaves that to the offset it stores. private var nextPositions: [UInt32: ConsumerPosition] = [:] private var lastConsumed: [UInt32: ConsumerPosition] = [:] private var lastStored: [UInt32: ConsumerPosition] = [:] @@ -128,10 +136,24 @@ public actor IggyConsumer: AsyncSequence { } public struct AsyncIterator: AsyncIteratorProtocol { - let consumer: IggyConsumer + /// Nil once an error or nil ended the iteration, as the protocol + /// requires. The consumer itself can be read on after an error. + var consumer: IggyConsumer? public mutating func next() async throws -> ReceivedMessage? { - try await consumer.next() + guard let consumer else { + return nil + } + do { + let received = try await consumer.next() + if received == nil { + self.consumer = nil + } + return received + } catch { + self.consumer = nil + throw error + } } } @@ -217,16 +239,17 @@ public actor IggyConsumer: AsyncSequence { logger.info("Shutting down the consumer", metadata: metadata) let timeout = configuration.offsetDrainTimeout if let intervalTask { - // A tick sleeping gets cancelled; one storing finishes first. + // A sleeping tick is cancelled. One that is storing ends after its + // store. if intervalIsStoring { await Self.wait(for: intervalTask, upTo: timeout) + } else { + intervalTask.cancel() } - intervalTask.cancel() } if let commitTask { commitWakeup.yield() await Self.wait(for: commitTask, upTo: timeout) - commitTask.cancel() } if configuration.autoCommit != .disabled { let positions = Array(lastConsumed.values) @@ -235,13 +258,18 @@ public actor IggyConsumer: AsyncSequence { await self.storeFinalPositions(positions) } } - if isConsumerGroup, joinedGroup { - joinedGroup = false - do { - try await client.leaveConsumerGroup(streamID: streamID, topicID: topicID, groupID: consumer.id) - } catch { - // Expected after an explicit leave or a deleted group. - logger.debug("Failed to leave the consumer group", metadata: metadata.merging(["error": "\(error)"]) { $1 }) + // Every join checks the shutdown under this lock, so none comes after + // the leave. + if isConsumerGroup, (try? await membership.lock()) != nil { + defer { membership.unlock() } + if joinedGroup { + joinedGroup = false + do { + try await client.leaveConsumerGroup(streamID: streamID, topicID: topicID, groupID: consumer.id) + } catch { + // Expected after an explicit leave or a deleted group. + logger.debug("Failed to leave the consumer group", metadata: metadata.merging(["error": "\(error)"]) { $1 }) + } } } eventsTask?.cancel() @@ -255,15 +283,21 @@ public actor IggyConsumer: AsyncSequence { /// Throws what a poll or join ran into, after the retry interval, and can /// be called again after it threw. Nil once the consumer is shut down. public func next() async throws -> ReceivedMessage? { - try await reading.lock() - defer { reading.unlock() } if isShutDown { return nil } + // Nothing suspends between the check and the handover, so a + // buffered message needs no lock. if bufferIndex < buffer.count { return handOverBuffered() } + try await reading.lock() + defer { reading.unlock() } while !isShutDown { + // The poll this reader waited for can have refilled the buffer. + if bufferIndex < buffer.count { + return handOverBuffered() + } try Task.checkCancellation() let polled = try await poll() if !polled.messages.isEmpty, !isShutDown { @@ -331,6 +365,8 @@ public actor IggyConsumer: AsyncSequence { try await storePosition(ConsumerPosition(partitionID: partition, offset: offset, context: last.context)) return } + try await offsetWrites.lock() + defer { offsetWrites.unlock() } try await client.storeConsumerOffset(consumer: consumer, streamID: streamID, topicID: topicID, partitionID: partition, offset: offset) lastStored[partition] = ConsumerPosition(partitionID: partition, offset: offset, context: .empty) } @@ -348,6 +384,8 @@ public actor IggyConsumer: AsyncSequence { /// server, since the Swift client routes the request by its partition. public func deleteOffset(partitionID: UInt32? = nil) async throws { let partition = partitionID ?? currentPartitionID + try await offsetWrites.lock() + defer { offsetWrites.unlock() } // A cancelled delete can still have committed, so a later store must // not be skipped as covered. lastStored[partition] = nil @@ -405,7 +443,10 @@ public actor IggyConsumer: AsyncSequence { } private func joinGroup() async throws { - if joinedGroup { + try await membership.lock() + defer { membership.unlock() } + // The final leave of a shutdown may have gone already. + if joinedGroup || isShutDown { return } if try await client.getConsumerGroup(streamID: streamID, topicID: topicID, groupID: consumer.id) == nil { @@ -453,7 +494,12 @@ public actor IggyConsumer: AsyncSequence { let retryInterval = configuration.pollingRetryInterval // A client that only the caller brings back never recovers, so the // poll goes on and fails with the reason instead of waiting for good. - while !canPoll, await client.canRecover { + // Without a heartbeat or a commit to send, nothing else signs the + // client in again, so the wait does. + while !canPoll, !isShutDown, await client.canRecover { + if (try? await client.requireAuthenticated()) != nil { + break + } try await Task.sleep(for: retryInterval) } if isConsumerGroup, configuration.autoJoinConsumerGroup, !joinedGroup { @@ -465,6 +511,10 @@ public actor IggyConsumer: AsyncSequence { throw error } } + // A shutdown may have sent its final commits and leave already. + if isShutDown { + return .empty + } lastPolledAt = .now let continuations = nextPositions let strategy = configuration.pollingStrategy @@ -524,11 +574,6 @@ public actor IggyConsumer: AsyncSequence { let position = ConsumerPosition(partitionID: partition, offset: message.offset, context: bufferedContext) lastConsumed[partition] = position let drained = bufferIndex == buffer.count - // The strategy only says where a partition starts; the next poll of - // it continues after the last message handed over. - if configuration.pollingStrategy.kind != .next { - nextPositions[partition] = ConsumerPosition(partitionID: partition, offset: message.offset + 1, context: bufferedContext) - } switch configuration.autoCommit.whenTrigger { case .consumingEachMessage?: queueCommit(position) @@ -545,6 +590,12 @@ public actor IggyConsumer: AsyncSequence { } let received = ReceivedMessage(message: message, currentOffset: bufferedCurrentOffset, partitionID: partition, context: bufferedContext) if drained { + // The strategy only says where a partition starts; the next poll + // of it continues after the last message handed over. A `next` + // poll that commits starts after the offset it stored instead. + if configuration.pollingStrategy.kind != .next || configuration.autoCommit.whenTrigger != .pollingMessages { + nextPositions[partition] = ConsumerPosition(partitionID: partition, offset: message.offset + 1, context: bufferedContext) + } buffer.removeAll(keepingCapacity: true) bufferIndex = 0 } @@ -572,7 +623,8 @@ public actor IggyConsumer: AsyncSequence { } /// Returns false once the consumer is shut down, which ends the interval - /// task before it stores anything more. + /// task before it stores anything more, or right after the store that a + /// shutdown waits for. private func storeOnTick() async -> Bool { guard !isShutDown else { return false @@ -582,7 +634,7 @@ public actor IggyConsumer: AsyncSequence { for position in lastConsumed.values { await storeInBackground(position) } - return true + return !isShutDown } private func storeFinalPositions(_ positions: [ConsumerPosition]) async { @@ -610,6 +662,8 @@ public actor IggyConsumer: AsyncSequence { } private func store(_ position: ConsumerPosition, allowReplay: Bool) async throws { + try await offsetWrites.lock() + defer { offsetWrites.unlock() } if !allowReplay, storedPositionCovers(position) { return } @@ -639,9 +693,18 @@ public actor IggyConsumer: AsyncSequence { } } + /// Waits for `task` up to `timeout`, then cancels it and waits for it to + /// end: a waiter that gives up does not cancel what it waits for. private static func wait(for task: Task, upTo timeout: Duration) async { - _ = try? await withDeadline(.now + timeout) { - await task.value + await withTaskGroup(of: Void.self) { group in + group.addTask { await task.value } + group.addTask { + if (try? await Task.sleep(for: timeout)) != nil { + task.cancel() + } + } + _ = await group.next() + group.cancelAll() } } } diff --git a/foreign/swift/Tests/IggyTests/ConsumerTests.swift b/foreign/swift/Tests/IggyTests/ConsumerTests.swift index 5d6fd7742e..c973fa07c1 100644 --- a/foreign/swift/Tests/IggyTests/ConsumerTests.swift +++ b/foreign/swift/Tests/IggyTests/ConsumerTests.swift @@ -20,12 +20,14 @@ import Testing @testable import Iggy -private func signedInClient(_ server: FakeVSRServer, requestTimeout: Duration = .seconds(10)) async throws -> IggyClient { +private func signedInClient( + _ server: FakeVSRServer, requestTimeout: Duration = .seconds(10), heartbeatInterval: Duration = .seconds(60) +) async throws -> IggyClient { var configuration = ClientConfiguration(address: server.address, autoLogin: .usernamePassword(username: "iggy", password: "iggy")) configuration.reconnection = ReconnectionOptions(maxRetries: nil, interval: .milliseconds(20), reestablishAfter: .zero) configuration.requestTimeout = requestTimeout configuration.connectTimeout = .seconds(2) - configuration.heartbeatInterval = .seconds(60) + configuration.heartbeatInterval = heartbeatInterval let client = IggyClient(configuration: configuration, connector: NIOConnector(), logger: testLogger) try await client.connect() return client @@ -46,6 +48,33 @@ private func serve(_ server: FakeVSRServer, poll: @escaping @Sendable (FakeVSRSe } } +/// Answers each poll with one message, as a partition holding `count` +/// messages does: `next` starts after the stored offset, or at zero when +/// none is stored, and a poll that commits stores the offset it returns. +private func serveOneByOne(_ server: FakeVSRServer, count: UInt64) async { + let stored = Locked(nil) + await server.respond { request in + switch request.code.flatMap(CommandCode.init(rawValue:)) { + case .getStream: return .ok(GoldenFixture.shared.bytes("response.get_stream")) + case .getTopic: return .ok(GoldenFixture.shared.bytes("response.get_topic")) + case .storeConsumerOffset: + stored.withLock { $0 = storedOffset(request) } + return .ok([]) + case .pollMessages: + let poll = polledStrategy(request) + let offset = poll.kind == PollingStrategy.Kind.next.rawValue ? stored.withLock { $0.map { $0 + 1 } ?? 0 } : poll.value + guard offset < count else { + return .ok(FakeVSRServer.polledBody(partitionID: 1, offsets: [])) + } + if poll.autoCommit { + stored.withLock { $0 = offset } + } + return .ok(FakeVSRServer.polledBody(partitionID: 1, offsets: [offset])) + default: return FakeVSRServer.defaultReply(request, address: server.address) + } + } +} + /// Answers the polls with the batches in turn, then with empty polls. private func batches( _ batches: [[UInt64]], partitionID: UInt32 = 1, context: PartitionContext = FakeVSRServer.routeContext @@ -115,19 +144,29 @@ struct ConsumerTests { await client.shutdown() } - @Test func nextLeavesTheContinuationToTheServer() async throws { - let server = try await FakeVSRServer.started() - defer { Task { await server.stop() } } - await serve(server, poll: batches([[0, 1], [2]])) - let client = try await signedInClient(server) - let consumer = try client.consumer( - name: "reader", streamID: 1, topicID: 2, partitionID: 1, configuration: ConsumerConfiguration(autoCommit: .disabled)) - try await consumer.initialize() - #expect(try await read(consumer, 3) == [0, 1, 2]) - let kinds = await requests(server, .pollMessages).prefix(2).map { polledStrategy($0).kind } - #expect(kinds == [PollingStrategy.Kind.next.rawValue, PollingStrategy.Kind.next.rawValue]) - await consumer.shutdown() - await client.shutdown() + /// A `next` poll starts after the offset stored on the server. Unless the + /// poll commits that offset itself, it can return the same messages + /// again, so the consumer continues after the last one handed over. + @Test func nextContinuesAfterTheLastMessageUnlessThePollCommits() async throws { + let next = PollingStrategy.Kind.next.rawValue + let offset = PollingStrategy.Kind.offset.rawValue + let cases: [(AutoCommit, [UInt8])] = [ + (.when(.consumingEveryNthMessage(2)), [next, offset, offset]), (.intervalOrWhen(.seconds(1), .pollingMessages), [next, next, next]), + ] + for (autoCommit, kinds) in cases { + let server = try await FakeVSRServer.started() + await serveOneByOne(server, count: 3) + let client = try await signedInClient(server) + let configuration = ConsumerConfiguration(batchLength: 1, autoCommit: autoCommit) + let consumer = try client.consumer(name: "reader", streamID: 1, topicID: 2, partitionID: 1, configuration: configuration) + try await consumer.initialize() + // A consumer that stalls reads nothing more, so the read is bounded. + #expect(try await withDeadline(.now + .seconds(5)) { try await read(consumer, 3) } == [0, 1, 2], "\(autoCommit)") + #expect(await requests(server, .pollMessages).prefix(3).map { polledStrategy($0).kind } == kinds, "\(autoCommit)") + await consumer.shutdown() + await client.shutdown() + await server.stop() + } } /// A message at or below the last one handed over from the same @@ -162,7 +201,7 @@ struct ConsumerTests { } /// A failed poll is thrown after the retry interval and the consumer - /// keeps reading on the next call. + /// keeps reading on the next call, while an iterator ends at the error. @Test func aFailedPollIsThrownAndReadingGoesOn() async throws { let server = try await FakeVSRServer.started() defer { Task { await server.stop() } } @@ -174,16 +213,64 @@ struct ConsumerTests { let configuration = ConsumerConfiguration(autoCommit: .disabled, pollingRetryInterval: .milliseconds(150)) let consumer = try client.consumer(name: "reader", streamID: 1, topicID: 2, partitionID: 1, configuration: configuration) try await consumer.initialize() + var iterator = consumer.makeAsyncIterator() let clock = ContinuousClock() let elapsed = await clock.measure { - await expectCode(.partitionNotFound) { _ = try await consumer.next() } + await expectCode(.partitionNotFound) { _ = try await iterator.next() } } #expect(elapsed >= .milliseconds(140)) + // An iterator that reads on takes the message, and the read below + // then waits for good. + try #require(try await iterator.next() == nil) #expect(try await consumer.next()?.message.offset == 0) await consumer.shutdown() await client.shutdown() } + /// With no heartbeat and no commit to send, nothing else brings the + /// client back after a lost poll, so the next poll signs it in again. A + /// poll still waiting for the client ends with nil at shutdown. + @Test func aPollSignsTheClientInAgainUntilTheConsumerShutsDown() async throws { + let server = try await FakeVSRServer.started() + defer { Task { await server.stop() } } + let polls = Locked(0) + let refusesSignIn = Locked(false) + let refusedSignIns = Locked(0) + await server.respond { request in + if refusesSignIn.withLock({ $0 }), FakeVSRServer.sessionReply(request) != nil { + refusedSignIns.withLock { $0 += 1 } + return .stall + } + switch request.code.flatMap(CommandCode.init(rawValue:)) { + case .getStream: return .ok(GoldenFixture.shared.bytes("response.get_stream")) + case .getTopic: return .ok(GoldenFixture.shared.bytes("response.get_topic")) + case .pollMessages: + let poll = polls.withLock { poll in + defer { poll += 1 } + return poll + } + return poll == 1 ? .ok(FakeVSRServer.polledBody(partitionID: 1, offsets: [0])) : .close + default: return FakeVSRServer.defaultReply(request, address: server.address) + } + } + let client = try await signedInClient(server, requestTimeout: .milliseconds(500), heartbeatInterval: .zero) + let configuration = ConsumerConfiguration(autoCommit: .disabled, pollingRetryInterval: .milliseconds(100)) + let consumer = try client.consumer(name: "reader", streamID: 1, topicID: 2, partitionID: 1, configuration: configuration) + try await consumer.initialize() + // A poll written and lost with its connection may have committed. + await expectCode(.transientNotCommitted) { _ = try await consumer.next() } + // withDeadline returns nil at the deadline. A client that is not + // signed in again stays down, and the read below then waits for good. + try #require(try await withDeadline(.now + .seconds(5)) { try await consumer.next()?.message.offset } == 0) + await expectCode(.transientNotCommitted) { _ = try await consumer.next() } + refusesSignIn.withLock { $0 = true } + async let ended = withDeadline(.now + .seconds(5)) { try await consumer.next() == nil } + #expect(await eventually { refusedSignIns.withLock { $0 } > 0 }) + await consumer.shutdown() + #expect(try await ended == true) + await client.shutdown() + } + /// A refused continuation is dropped, so the next poll of the partition /// starts where the strategy says again. @Test func aRecreatedPartitionDropsItsContinuation() async throws { @@ -306,6 +393,39 @@ struct ConsumerTests { await client.shutdown() } + /// Stores run one at a time, so a position that the store in flight + /// covers is not sent after it, even when the two calls overlap. + @Test func overlappingStoresNeverMoveTheStoredOffsetBack() async throws { + let server = try await FakeVSRServer.started() + defer { Task { await server.stop() } } + let poll = batches([[3, 5]]) + await server.respond { [poll] request in + switch request.code.flatMap(CommandCode.init(rawValue:)) { + case .getStream: return .ok(GoldenFixture.shared.bytes("response.get_stream")) + case .getTopic: return .ok(GoldenFixture.shared.bytes("response.get_topic")) + case .pollMessages: return await poll(request) + case .storeConsumerOffset: + try? await Task.sleep(for: .milliseconds(200)) + return .ok([]) + default: return FakeVSRServer.defaultReply(request, address: server.address) + } + } + let client = try await signedInClient(server) + let consumer = try client.consumer( + name: "reader", streamID: 1, topicID: 2, partitionID: 1, configuration: ConsumerConfiguration(autoCommit: .disabled)) + try await consumer.initialize() + let older = try #require(try await consumer.next()) + let newer = try #require(try await consumer.next()) + let storing = Task { try await consumer.storePosition(newer.position) } + #expect(await eventually { await requests(server, .storeConsumerOffset).count == 1 }) + try await consumer.storePosition(older.position) + try await storing.value + #expect(await requests(server, .storeConsumerOffset).map(storedOffset) == [5]) + #expect(await consumer.lastStoredOffset(partitionID: 1) == 5) + await consumer.shutdown() + await client.shutdown() + } + /// The after triggers fire once the handler returned, whether it /// succeeded or threw: every message, an offset that divides by n, or the /// message at the partition's newest offset. @@ -395,6 +515,71 @@ struct ConsumerTests { await client.shutdown() } + /// A shutdown during a join leaves the group once the join is answered, + /// whether initialize() or a poll that lost the membership started it. + /// Nothing is polled after the shutdown. + @Test func aShutdownDuringAJoinLeavesTheGroupAfterIt() async throws { + for joinsOnPoll in [false, true] { + let server = try await FakeVSRServer.started() + let joins = Locked(0) + let slowJoin = joinsOnPoll ? 2 : 1 + await server.respond { request in + switch request.code.flatMap(CommandCode.init(rawValue:)) { + case .getStream: return .ok(GoldenFixture.shared.bytes("response.get_stream")) + case .getTopic: return .ok(GoldenFixture.shared.bytes("response.get_topic")) + case .getConsumerGroup: return .ok(GoldenFixture.shared.bytes("response.consumer_group_details")) + case .syncConsumerGroup: return .ok(FakeVSRServer.assignmentBody(generation: 1, partitions: [1])) + case .joinConsumerGroup: + let join = joins.withLock { join in + join += 1 + return join + } + if join == slowJoin { + try? await Task.sleep(for: .milliseconds(200)) + } + return .ok([]) + case .pollMessages: return .status(IggyErrorCode.consumerGroupMemberNotFound.rawValue) + default: return FakeVSRServer.defaultReply(request, address: server.address) + } + } + let client = try await signedInClient(server) + let consumer = try client.consumerGroup(name: "billing", streamID: 1, topicID: 2, configuration: ConsumerConfiguration(autoCommit: .disabled)) + let reading = Task { () -> ReceivedMessage? in + try await consumer.initialize() + guard joinsOnPoll else { + return nil + } + return try await consumer.next() + } + #expect(await eventually { joins.withLock { $0 } == slowJoin }) + await consumer.shutdown() + #expect(await requests(server, .leaveConsumerGroup).count == 1, "joins on poll: \(joinsOnPoll)") + #expect(try await reading.value == nil) + #expect(await requests(server, .pollMessages).count == (joinsOnPoll ? 1 : 0), "joins on poll: \(joinsOnPoll)") + await client.shutdown() + await server.stop() + } + } + + /// A poll that would join again once the consumer shut down sends + /// neither the join nor the poll, so no membership outlives the consumer. + @Test func aPollAfterShutdownDoesNotJoinAgain() async throws { + let server = try await FakeVSRServer.started() + defer { Task { await server.stop() } } + await serve(server) { _ in .status(IggyErrorCode.consumerGroupMemberNotFound.rawValue) } + let client = try await signedInClient(server) + let configuration = ConsumerConfiguration(pollInterval: .milliseconds(300), autoCommit: .disabled) + let consumer = try client.consumerGroup(name: "billing", streamID: 1, topicID: 2, configuration: configuration) + try await consumer.initialize() + let reading = Task { try await consumer.next() } + #expect(await eventually { await requests(server, .pollMessages).count == 1 }) + await consumer.shutdown() + #expect(try await reading.value == nil) + #expect(await requests(server, .joinConsumerGroup).count == 1) + #expect(await requests(server, .pollMessages).count == 1) + await client.shutdown() + } + @Test func aMissingGroupIsNotCreatedWhenAutoCreationIsOff() async throws { let server = try await FakeVSRServer.started() defer { Task { await server.stop() } } @@ -477,6 +662,47 @@ struct ConsumerTests { await client.shutdown() } + /// A tick storing at shutdown ends right after its store. One whose store + /// outlasts the drain timeout is cancelled there, so shutdown waits + /// neither a full interval nor the request timeout. + @Test func shutdownEndsAStoringTickAfterItsStoreOrAtTheDrainTimeout() async throws { + for (stalls, bound) in [(false, Duration.milliseconds(700)), (true, .seconds(3))] { + let server = try await FakeVSRServer.started() + let poll = batches([[0]]) + let stores = Locked(0) + await server.respond { [poll] request in + switch request.code.flatMap(CommandCode.init(rawValue:)) { + case .getStream: return .ok(GoldenFixture.shared.bytes("response.get_stream")) + case .getTopic: return .ok(GoldenFixture.shared.bytes("response.get_topic")) + case .pollMessages: return await poll(request) + case .storeConsumerOffset: + let store = stores.withLock { store in + store += 1 + return store + } + // Only the store of the tick hangs, so the final store of + // the shutdown goes through. + if stalls, store == 1 { + return .stall + } + try? await Task.sleep(for: .milliseconds(200)) + return .ok([]) + default: return FakeVSRServer.defaultReply(request, address: server.address) + } + } + let client = try await signedInClient(server, requestTimeout: .seconds(5)) + let configuration = ConsumerConfiguration(autoCommit: .interval(.seconds(1)), offsetDrainTimeout: .seconds(1)) + let consumer = try client.consumer(name: "reader", streamID: 1, topicID: 2, partitionID: 1, configuration: configuration) + try await consumer.initialize() + #expect(try await read(consumer, 1) == [0]) + #expect(await eventually { await !requests(server, .storeConsumerOffset).isEmpty }) + let elapsed = await ContinuousClock().measure { await consumer.shutdown() } + #expect(elapsed < bound, "stalls: \(stalls), took \(elapsed)") + await client.shutdown() + await server.stop() + } + } + /// Concurrent initializations share one run: one lookup, one join. @Test func concurrentInitializationsShareOneRun() async throws { let server = try await FakeVSRServer.started()