diff --git a/foreign/swift/README.md b/foreign/swift/README.md index 5c94ef68fa..a7f2705ca0 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,40 @@ 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. `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 +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. + ### Raw requests `sendRawRequest(code:payload:)` sends a command code with an encoded payload and returns the raw reply body. `sendRawRequest(code:payload:replica:)` sends it to one node of a cluster, named by the replica id that cluster metadata returns. 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 a5bae7cd6d..f820a7c772 100644 --- a/foreign/swift/Sources/Iggy/Client/IggyClient.swift +++ b/foreign/swift/Sources/Iggy/Client/IggyClient.swift @@ -267,6 +267,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..16a80962fd --- /dev/null +++ b/foreign/swift/Sources/Iggy/Consumer/ConsumerConfiguration.swift @@ -0,0 +1,201 @@ +// 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 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 + /// 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..2d42beee53 --- /dev/null +++ b/foreign/swift/Sources/Iggy/Consumer/IggyConsumer.swift @@ -0,0 +1,735 @@ +// 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()``. 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 +/// 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 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 + + 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 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] = [:] + /// 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 { + /// 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? { + 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 + } + } + } + + // 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 sleeping tick is cancelled. One that is storing ends after its + // store. + if intervalIsStoring { + await Self.wait(for: intervalTask, upTo: timeout) + } else { + intervalTask.cancel() + } + } + if let commitTask { + commitWakeup.yield() + await Self.wait(for: commitTask, upTo: timeout) + } + 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) + } + } + // 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() + 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? { + 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 { + 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 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) + } + + /// 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 + 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 + 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 { + 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 { + 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. + // 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 { + 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 + } + } + // A shutdown may have sent its final commits and leave already. + if isShutDown { + return .empty + } + 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 + 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 { + // 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 + } + 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, or right after the store that a + /// shutdown waits for. + private func storeOnTick() async -> Bool { + guard !isShutDown else { + return false + } + intervalIsStoring = true + defer { intervalIsStoring = false } + for position in lastConsumed.values { + await storeInBackground(position) + } + return !isShutDown + } + + 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 { + try await offsetWrites.lock() + defer { offsetWrites.unlock() } + 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 + } + } + + /// 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 { + 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() + } + } +} + +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..c973fa07c1 --- /dev/null +++ b/foreign/swift/Tests/IggyTests/ConsumerTests.swift @@ -0,0 +1,758 @@ +// 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), 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 = heartbeatInterval + 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 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 +) -> @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() + } + + /// 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 + /// 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, while an iterator ends at the error. + @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() + var iterator = consumer.makeAsyncIterator() + let clock = ContinuousClock() + let elapsed = await clock.measure { + 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 { + 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() + } + + /// 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. + @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() + } + + /// 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() } } + 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() + } + + /// 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() + 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()