diff --git a/foreign/swift/README.md b/foreign/swift/README.md index c7333530b7..7b3d31111a 100644 --- a/foreign/swift/README.md +++ b/foreign/swift/README.md @@ -12,10 +12,11 @@ Official Swift client SDK for [Apache Iggy](https://iggy.apache.org) message str The SDK is built up in stages. This stage carries the package scaffold, the error table, the byte codec the wire protocol is encoded with, native XXH3-64 and XXH32 hashes pinned -to the Rust crates by golden vectors, and the domain models (streams, topics, messages, -user headers, options, permissions) with their wire encodings. The protocol layer, the -TCP and TLS client, the producer and consumer, examples, and BDD scenarios follow in -later changes. +to the Rust crates by golden vectors, the domain models (streams, topics, messages, +user headers, options, permissions) with their wire encodings, and the protocol layer: +the consensus frame, the session and request id rules, the request encoders, the reply +decoders, and the message batch format. The TCP and TLS client, the producer and +consumer, examples, and BDD scenarios follow in later changes. ## Requirements @@ -78,8 +79,8 @@ cd foreign/swift swift test ``` -The hash and error-code tests compare against `Tests/IggyTests/Fixtures/golden.json`, a -fixture dumped from the Rust protocol crates. Regenerate it after a protocol change, from +The hash, error-code, model, request, reply, and consensus header tests compare against +`Tests/IggyTests/Fixtures/golden.json`, a fixture dumped from the Rust protocol crates. Regenerate it after a protocol change, from `foreign/swift`, and commit the result; CI fails when the checked-in copy is stale: ```bash diff --git a/foreign/swift/Sources/Iggy/Models/ResourceOptions.swift b/foreign/swift/Sources/Iggy/Models/ResourceOptions.swift index 9d133b3fb1..39e7d55e15 100644 --- a/foreign/swift/Sources/Iggy/Models/ResourceOptions.swift +++ b/foreign/swift/Sources/Iggy/Models/ResourceOptions.swift @@ -157,12 +157,6 @@ enum OptionsBlock { /// Reads a `u32`-length-prefixed block; every caller decodes it next, /// which validates it. static func readPrefixed(from reader: inout ByteReader) throws -> ArraySlice { - let declared = try reader.readUInt32() - // A length that does not fit the platform's Int cannot be in the - // buffer either; report it as truncation rather than trapping. - guard let length = Int(exactly: declared) else { - throw WireError.truncated(offset: reader.position, need: Int.max, have: reader.remaining) - } - return try reader.readBytes(length) + try reader.readPrefixedBytes() } } diff --git a/foreign/swift/Sources/Iggy/Protocol/Batch.swift b/foreign/swift/Sources/Iggy/Protocol/Batch.swift new file mode 100644 index 0000000000..37ab8727b1 --- /dev/null +++ b/foreign/swift/Sources/Iggy/Protocol/Batch.swift @@ -0,0 +1,259 @@ +// 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. + +/// The message batch: the one layout a send body, the replicated prepare, +/// the persisted segment record, and the poll reply all share. A port of +/// `core/binary_protocol/src/batch.rs`. +/// +/// ```text +/// [batch header: 256 bytes][blob: message frames] +/// frame = [header: 48 bytes][payload][user_headers] +/// ``` +enum Batch { + static let headerSize = 256 + static let messageHeaderSize = 48 + static let checksumOffset = 40 + static let messageCountOffset = 48 + static let reservedOffset = 52 + /// A message's timestamp delta is a `u32` of microseconds, so one batch + /// spans at most about 71 minutes of producer clock. + static let maxTimestampDelta = UInt64(UInt32.max) + + struct Header: Equatable { + var partitionID: UInt64 + var baseOffset: UInt64 + var baseTimestamp: UInt64 + var originTimestamp: UInt64 + /// Total batch size: header plus blob. + var batchLength: UInt64 + var batchChecksum: UInt64 + var messageCount: UInt32 + + static func decode(_ bytes: ArraySlice) throws -> Header { + guard bytes.count >= Batch.headerSize else { + throw WireError.truncated(offset: 0, need: Batch.headerSize, have: bytes.count) + } + let start = bytes.startIndex + let batchLength = UInt64(littleEndianBytes: bytes[start + 32..= Batch.headerSize, batchLength <= UInt64(Int.max) else { + throw WireError.validation("batch length must cover the batch header") + } + // Every encoder zeroes the reserved region; admitting nonzero bytes + // would let unchecksummed data ride the header. + guard bytes[start + Batch.reservedOffset.. [UInt8] { + var writer = ByteWriter(capacity: Batch.headerSize) + writer.write(partitionID) + writer.write(baseOffset) + writer.write(baseTimestamp) + writer.write(originTimestamp) + writer.write(batchLength) + writer.write(batchChecksum) + writer.write(messageCount) + writer.writeZeros(Batch.headerSize - Batch.reservedOffset) + return writer.bytes + } + + /// The bytes the batch checksum covers ahead of the frame checksums. + fileprivate func checksumPrefix() -> [UInt8] { + var writer = ByteWriter(capacity: 44) + writer.write(partitionID) + writer.write(baseOffset) + writer.write(baseTimestamp) + writer.write(originTimestamp) + writer.write(batchLength) + writer.write(messageCount) + return writer.bytes + } + } + + struct MessageHeader: Equatable { + var checksum: UInt64 + var id: UInt128Value + var offsetDelta: UInt32 + var timestampDelta: UInt32 + var userHeadersLength: UInt32 + var payloadLength: UInt32 + + static func decode(_ bytes: ArraySlice) throws -> MessageHeader { + guard bytes.count >= Batch.messageHeaderSize else { + throw WireError.truncated(offset: 0, need: Batch.messageHeaderSize, have: bytes.count) + } + let start = bytes.startIndex + guard UInt64(littleEndianBytes: bytes[start + 40.. UInt64 { + XXH3.hash64(header.checksumPrefix() + frameChecksums) + } + + /// A message prepared for the wire. + struct OutgoingMessage { + var id: UInt128Value + var originTimestamp: UInt64 + var payload: [UInt8] + var userHeaders: [UInt8]? + } + + /// Encodes the body of a send request: + /// `[metadata_len:u32][stream][topic][partitioning][count:u32][batch]`. + static func encodeSend(streamID: Identifier, topicID: Identifier, partitioning: Partitioning, messages: [OutgoingMessage]) throws -> [UInt8] { + guard !messages.isEmpty else { + throw IggyError(.invalidMessagesCount, context: "cannot encode an empty message batch") + } + let metadataLength = streamID.encodedSize + topicID.encodedSize + partitioning.encodedSize + 4 + let blobLength = messages.reduce(0) { $0 + messageHeaderSize + $1.payload.count + ($1.userHeaders?.count ?? 0) } + guard headerSize + blobLength <= UInt32.max else { + throw IggyError(.tooBigMessagePayload, context: "batch exceeds the frame size limit") + } + var writer = ByteWriter(capacity: 4 + metadataLength + headerSize + blobLength) + writer.write(UInt32(metadataLength)) + try streamID.encode(into: &writer) + try topicID.encode(into: &writer) + try partitioning.encode(into: &writer) + writer.write(UInt32(messages.count)) + + let originTimestamp = messages.lazy.map(\.originTimestamp).min() ?? 0 + let headerStart = writer.count + writer.writeZeros(headerSize) + + var frameChecksums = [UInt8]() + frameChecksums.reserveCapacity(messages.count * 8) + for (index, message) in messages.enumerated() { + let timestampDelta = message.originTimestamp - originTimestamp + guard timestampDelta <= maxTimestampDelta else { + throw IggyError(.invalidMessageTimestampDelta, context: "delta \(timestampDelta) microseconds exceeds the per-batch limit") + } + let headers = message.userHeaders ?? [] + let frameStart = writer.count + writer.write(UInt64(0)) + writer.write(message.id) + writer.write(UInt32(index)) + writer.write(UInt32(timestampDelta)) + writer.write(UInt32(headers.count)) + writer.write(UInt32(message.payload.count)) + writer.write(UInt64(0)) + writer.write(message.payload) + writer.write(headers) + let checksum = XXH3.hash64(writer.bytes[(frameStart + 8)...]) + writer.overwrite(at: frameStart, with: checksum) + withUnsafeBytes(of: checksum.littleEndian) { frameChecksums.append(contentsOf: $0) } + } + + var header = Header( + partitionID: 0, baseOffset: 0, baseTimestamp: 0, originTimestamp: originTimestamp, batchLength: UInt64(headerSize + blobLength), batchChecksum: 0, + messageCount: UInt32(messages.count)) + header.batchChecksum = Self.checksum(header: header, frameChecksums: frameChecksums) + writer.overwrite(at: headerStart, with: header.encode()) + return writer.bytes + } + + /// Decodes the batch records of a poll reply body into messages. Layout + /// is always validated; checksums only when asked, matching the Rust + /// SDK's poll path. + static func decodeMessages(_ bytes: ArraySlice, expectedCount: UInt32, verifyChecksums: Bool = false) throws -> [IggyMessage] { + var messages: [IggyMessage] = [] + messages.reserveCapacity(Int(min(expectedCount, 4096))) + var position = bytes.startIndex + while position < bytes.endIndex { + let header = try Header.decode(bytes[position...]) + // The declared length is compared against what is left as a + // `UInt64` before it is added, so a hostile value cannot overflow. + let available = bytes.endIndex - position + guard header.batchLength <= UInt64(available) else { + throw WireError.truncated(offset: position - bytes.startIndex, need: Int(clamping: header.batchLength), have: available) + } + let batchEnd = position + Int(header.batchLength) + var cursor = position + headerSize + var frameChecksums = [UInt8]() + var counted: UInt32 = 0 + while cursor < batchEnd { + let frame = try MessageHeader.decode(bytes[cursor.. 0 ? Array(bytes[payloadEnd.. Bool { + code == CommandCode.loginRegister.rawValue + || code == CommandCode.loginRegisterWithPersonalAccessToken.rawValue + } +} diff --git a/foreign/swift/Sources/Iggy/Protocol/ConsensusOperation.swift b/foreign/swift/Sources/Iggy/Protocol/ConsensusOperation.swift new file mode 100644 index 0000000000..910a6e7a27 --- /dev/null +++ b/foreign/swift/Sources/Iggy/Protocol/ConsensusOperation.swift @@ -0,0 +1,143 @@ +// 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. + +/// Consensus operation discriminant carried in every VSR header, a port of +/// `core/binary_protocol/src/consensus/operation.rs`. It selects the +/// replication plane a request travels on and tells the client how to read +/// the reply body. +enum ConsensusOperation: UInt8, Sendable, Hashable { + case reserved = 0 + case register = 1 + case nonReplicated = 2 + case logout = 3 + + case createTopicWithAssignments = 64 + case createPartitionsWithAssignments = 65 + case removeConsumerGroupMember = 66 + case completeConsumerGroupRevocation = 67 + case truncatePartition = 68 + + case createStream = 128 + case updateStream = 129 + case deleteStream = 130 + case purgeStream = 131 + case createTopic = 132 + case updateTopic = 133 + case deleteTopic = 134 + case purgeTopic = 135 + case createPartitions = 136 + case deletePartitions = 137 + case deleteSegments = 138 + case createConsumerGroup = 139 + case deleteConsumerGroup = 140 + case createUser = 141 + case updateUser = 142 + case deleteUser = 143 + case changePassword = 144 + case updatePermissions = 145 + case createPersonalAccessToken = 146 + case deletePersonalAccessToken = 147 + case joinConsumerGroup = 148 + case leaveConsumerGroup = 149 + + case sendMessages = 160 + case storeConsumerOffset = 161 + case deleteConsumerOffset = 162 + + private static let internalStart = ConsensusOperation.createTopicWithAssignments.rawValue + private static let metadataStart = ConsensusOperation.createStream.rawValue + private static let partitionStart = ConsensusOperation.sendMessages.rawValue + + /// Replica-internal band, never sent by a client. + var isInternal: Bool { + rawValue >= Self.internalStart && rawValue < Self.metadataStart + } + + /// Operations replicated through the metadata consensus group. + var isMetadata: Bool { + if isInternal { + return true + } + switch self { + case .createStream, .updateStream, .deleteStream, .purgeStream, .createTopic, .updateTopic, + .deleteTopic, .purgeTopic, .createPartitions, .deletePartitions, .createConsumerGroup, + .deleteConsumerGroup, .createUser, .updateUser, .deleteUser, .changePassword, + .updatePermissions, .createPersonalAccessToken, .deletePersonalAccessToken, + .joinConsumerGroup, .leaveConsumerGroup: + return true + default: + return false + } + } + + /// Data-plane operations routed to the shard owning the partition. + var isPartition: Bool { + rawValue >= Self.partitionStart + } + + /// Whether the reply body leads with a committed result section. + var isResultFramed: Bool { + isMetadata || self == .storeConsumerOffset || self == .deleteConsumerOffset + } + + /// The command code a replicated operation maps to. + static func replicated(for code: UInt32) -> ConsensusOperation? { + switch CommandCode(rawValue: code) { + case .createUser: .createUser + case .deleteUser: .deleteUser + case .updateUser: .updateUser + case .updatePermissions: .updatePermissions + case .changePassword: .changePassword + case .createPersonalAccessToken: .createPersonalAccessToken + case .deletePersonalAccessToken: .deletePersonalAccessToken + case .sendMessages: .sendMessages + case .storeConsumerOffset: .storeConsumerOffset + case .deleteConsumerOffset: .deleteConsumerOffset + case .createStream: .createStream + case .deleteStream: .deleteStream + case .updateStream: .updateStream + case .purgeStream: .purgeStream + case .createTopic: .createTopic + case .deleteTopic: .deleteTopic + case .updateTopic: .updateTopic + case .purgeTopic: .purgeTopic + case .createPartitions: .createPartitions + case .deletePartitions: .deletePartitions + case .deleteSegments: .deleteSegments + case .createConsumerGroup: .createConsumerGroup + case .deleteConsumerGroup: .deleteConsumerGroup + case .joinConsumerGroup: .joinConsumerGroup + case .leaveConsumerGroup: .leaveConsumerGroup + default: nil + } + } + + /// The header operation for a command code. The two register codes resolve + /// to `register` even though the dispatch table files them as + /// non-replicated: the SDK drives the handshake and the server reads the + /// operation byte. Unknown codes travel as non-replicated so the server + /// can classify or reject them. + static func forCode(_ code: UInt32) -> ConsensusOperation { + if CommandCode.isRegister(code) { + return .register + } + if code == CommandCode.logoutUser.rawValue { + return .logout + } + return replicated(for: code) ?? .nonReplicated + } +} diff --git a/foreign/swift/Sources/Iggy/Protocol/ConsensusSession.swift b/foreign/swift/Sources/Iggy/Protocol/ConsensusSession.swift new file mode 100644 index 0000000000..5133c3a82d --- /dev/null +++ b/foreign/swift/Sources/Iggy/Protocol/ConsensusSession.swift @@ -0,0 +1,87 @@ +// 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. + +/// Consensus-level session state, a port of `core/sdk/src/session.rs`. +/// +/// Each client generates an ephemeral random client identifier. After login +/// the server registers the client through consensus and returns a session +/// number, the fence every later request echoes. Replicated requests carry a +/// monotonically increasing request id the server deduplicates on. +/// +/// Not thread-safe: the connection owns one and serializes access through +/// the lockstep exchange. +struct ConsensusSession: Sendable { + private(set) var clientID: UInt128Value + private(set) var session: UInt64? + private var requestCounter: UInt64 + private var registerConsumed: Bool + + init(clientID: UInt128Value = .random()) { + self.clientID = clientID + self.session = nil + self.requestCounter = 1 + self.registerConsumed = false + } + + var isBound: Bool { session != nil } + + /// Arms the session for a Register and returns its request id, always 0. + /// A session that already registered or bound is replaced wholesale with a + /// fresh client identifier, so a re-login encodes a clean Register. + mutating func beginRegister() -> UInt64 { + if registerConsumed || isBound { + self = ConsensusSession() + } + registerConsumed = true + return 0 + } + + /// Records the session fence the server assigned when Register committed. + mutating func bind(_ session: UInt64) throws { + if isBound { + throw IggyError(.alreadyAuthenticated) + } + if session == 0 { + throw IggyError(.invalidSession, context: "session must be greater than zero") + } + self.session = session + } + + /// The next replicated request id, advancing the watermark: 1, 2, 3, ... + mutating func nextRequestID() throws -> UInt64 { + guard isBound else { + throw IggyError(.unauthenticated, context: "request id taken before the session is bound") + } + if requestCounter == UInt64.max { + throw IggyError(.invalidConfiguration, context: "request counter exhausted") + } + let id = requestCounter + requestCounter += 1 + return id + } + + /// The watermark without advancing it. Non-replicated requests use it + /// because the server never consults the client table for them. + var currentRequestID: UInt64 { requestCounter } + + /// Drops the session and mints a fresh client identifier. A disconnect + /// invalidates the server-side fence, so the next Register must arrive + /// under a new identity. + mutating func reset() { + self = ConsensusSession() + } +} diff --git a/foreign/swift/Sources/Iggy/Protocol/ProtocolVersion.swift b/foreign/swift/Sources/Iggy/Protocol/ProtocolVersion.swift new file mode 100644 index 0000000000..03a65a2578 --- /dev/null +++ b/foreign/swift/Sources/Iggy/Protocol/ProtocolVersion.swift @@ -0,0 +1,46 @@ +// 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. + +/// Packed semantic version of the binary protocol: ten bits per component, +/// `major << 20 | minor << 10 | patch`. Integer order equals semver order. +public struct ProtocolVersion: Sendable, Hashable, Comparable, CustomStringConvertible { + public let packed: UInt32 + + public init(packed: UInt32) { + self.packed = packed + } + + public init(major: UInt32, minor: UInt32, patch: UInt32) { + precondition(major < 1024 && minor < 1024 && patch < 1024, "semver component exceeds 10-bit packing range") + packed = (major << 20) | (minor << 10) | patch + } + + /// The protocol version this SDK implements: the version of the + /// `iggy_binary_protocol` crate it was ported from. The golden fixture + /// carries the crate's value and a test fails when the two diverge. + public static let current = ProtocolVersion(major: 0, minor: 11, patch: 1) + + public var major: UInt32 { packed >> 20 } + public var minor: UInt32 { (packed >> 10) & 0x3FF } + public var patch: UInt32 { packed & 0x3FF } + + public var description: String { "\(major).\(minor).\(patch)" } + + public static func < (lhs: ProtocolVersion, rhs: ProtocolVersion) -> Bool { + lhs.packed < rhs.packed + } +} diff --git a/foreign/swift/Sources/Iggy/Protocol/Requests.swift b/foreign/swift/Sources/Iggy/Protocol/Requests.swift new file mode 100644 index 0000000000..3c1b211739 --- /dev/null +++ b/foreign/swift/Sources/Iggy/Protocol/Requests.swift @@ -0,0 +1,348 @@ +// 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. + +/// Request payload encoders, one per command, mirroring +/// `core/binary_protocol/src/requests`. Every function returns the body that +/// follows the consensus header. +enum Requests { + // MARK: Streams + + static func createStream(name: String, options: [UInt8] = []) throws -> [UInt8] { + var writer = ByteWriter() + try writer.writeName(validatedWireName(name)) + writer.write(options) + return writer.bytes + } + + static func identifierOnly(_ id: Identifier) throws -> [UInt8] { + var writer = ByteWriter(capacity: id.encodedSize) + try id.encode(into: &writer) + return writer.bytes + } + + static func updateStream(streamID: Identifier, name: String, options: [UInt8]) throws -> [UInt8] { + var writer = ByteWriter() + try streamID.encode(into: &writer) + try writer.writeName(validatedWireName(name)) + writer.write(options) + return writer.bytes + } + + // MARK: Topics + + static func streamAndTopic(_ streamID: Identifier, _ topicID: Identifier) throws -> [UInt8] { + var writer = ByteWriter(capacity: streamID.encodedSize + topicID.encodedSize) + try streamID.encode(into: &writer) + try topicID.encode(into: &writer) + return writer.bytes + } + + static func createTopic(streamID: Identifier, partitionsCount: UInt32, name: String, options: [UInt8]) throws -> [UInt8] { + var writer = ByteWriter() + try streamID.encode(into: &writer) + writer.write(partitionsCount) + try writer.writeName(validatedWireName(name)) + writer.write(options) + return writer.bytes + } + + static func updateTopic(streamID: Identifier, topicID: Identifier, name: String, options: [UInt8]) throws -> [UInt8] { + var writer = ByteWriter() + try streamID.encode(into: &writer) + try topicID.encode(into: &writer) + try writer.writeName(validatedWireName(name)) + writer.write(options) + return writer.bytes + } + + // MARK: Partitions and segments + + static func partitions(streamID: Identifier, topicID: Identifier, partitionsCount: UInt32) throws -> [UInt8] { + var writer = ByteWriter() + try streamID.encode(into: &writer) + try topicID.encode(into: &writer) + writer.write(partitionsCount) + return writer.bytes + } + + static func deleteSegments(streamID: Identifier, topicID: Identifier, partitionID: UInt32, segmentsCount: UInt32) throws -> [UInt8] { + var writer = ByteWriter() + try streamID.encode(into: &writer) + try topicID.encode(into: &writer) + writer.write(partitionID) + writer.write(segmentsCount) + return writer.bytes + } + + // MARK: Consumer groups + + static func createConsumerGroup(streamID: Identifier, topicID: Identifier, name: String) throws -> [UInt8] { + var writer = ByteWriter() + try streamID.encode(into: &writer) + try topicID.encode(into: &writer) + try writer.writeName(validatedWireName(name)) + return writer.bytes + } + + static func consumerGroup(streamID: Identifier, topicID: Identifier, groupID: Identifier) throws -> [UInt8] { + var writer = ByteWriter() + try streamID.encode(into: &writer) + try topicID.encode(into: &writer) + try groupID.encode(into: &writer) + return writer.bytes + } + + // MARK: Consumer offsets + + private static func consumerTarget( + _ consumer: Consumer, _ streamID: Identifier, _ topicID: Identifier, _ partitionID: UInt32?, into writer: inout ByteWriter + ) throws { + try consumer.encode(into: &writer) + try streamID.encode(into: &writer) + try topicID.encode(into: &writer) + if let partitionID { + writer.write(UInt8(1)) + writer.write(partitionID) + } else { + writer.write(UInt8(0)) + writer.write(UInt32(0)) + } + } + + static func storeConsumerOffset( + consumer: Consumer, streamID: Identifier, topicID: Identifier, partitionID: UInt32?, offset: UInt64, ack: AckLevel + ) throws -> [UInt8] { + var writer = ByteWriter() + try consumerTarget(consumer, streamID, topicID, partitionID, into: &writer) + writer.write(offset) + writer.write(ack.rawValue) + return writer.bytes + } + + static func getConsumerOffset(consumer: Consumer, streamID: Identifier, topicID: Identifier, partitionID: UInt32?) throws -> [UInt8] { + var writer = ByteWriter() + try consumerTarget(consumer, streamID, topicID, partitionID, into: &writer) + return writer.bytes + } + + static func deleteConsumerOffset(consumer: Consumer, streamID: Identifier, topicID: Identifier, partitionID: UInt32?, ack: AckLevel) throws -> [UInt8] { + var writer = ByteWriter() + try consumerTarget(consumer, streamID, topicID, partitionID, into: &writer) + writer.write(ack.rawValue) + return writer.bytes + } + + // MARK: Messages + + static func pollMessages( + consumer: Consumer, streamID: Identifier, topicID: Identifier, partitionID: UInt32?, strategy: PollingStrategy, count: UInt32, autoCommit: Bool + ) throws -> [UInt8] { + var writer = ByteWriter() + try consumerTarget(consumer, streamID, topicID, partitionID, into: &writer) + strategy.encode(into: &writer) + writer.write(count) + writer.write(autoCommit) + return writer.bytes + } + + // MARK: Users + + /// Byte length of a bind secret carried by login and bind-session requests. + static let bindSecretSize = 32 + + static let minUsernameLength = 3 + static let maxUsernameLength = 50 + static let minPasswordLength = 3 + static let maxPasswordLength = 100 + + /// The bounds every server enforces, applied before encoding so an + /// oversized value can never desync the length prefix on the wire. + static func validateUsername(_ username: String) throws { + let length = username.utf8.count + guard length >= minUsernameLength, length <= maxUsernameLength else { + throw IggyError(.invalidUsername, context: "username must be \(minUsernameLength)-\(maxUsernameLength) bytes, got \(length)") + } + } + + static func validatePassword(_ password: String) throws { + let length = password.utf8.count + guard length >= minPasswordLength, length <= maxPasswordLength else { + throw IggyError(.invalidPassword, context: "password must be \(minPasswordLength)-\(maxPasswordLength) bytes, got \(length)") + } + } + + static func createUser(username: String, password: String, status: UserStatus, permissions: Permissions?, options: [UInt8] = []) throws -> [UInt8] { + try validateUsername(username) + try validatePassword(password) + var writer = ByteWriter() + try writer.writeName(username) + try writer.writeName(password) + writer.write(status.rawValue) + writePermissions(permissions, into: &writer) + writer.write(options) + return writer.bytes + } + + private static func writePermissions(_ permissions: Permissions?, into writer: inout ByteWriter) { + if let permissions { + writer.write(UInt8(1)) + let bytes = permissions.encode() + writer.write(UInt32(bytes.count)) + writer.write(bytes) + } else { + writer.write(UInt8(0)) + } + } + + static func updateUser(userID: Identifier, username: String?, status: UserStatus?, options: [UInt8]) throws -> [UInt8] { + var writer = ByteWriter() + try userID.encode(into: &writer) + if let username { + try validateUsername(username) + writer.write(UInt8(1)) + try writer.writeName(username) + } else { + writer.write(UInt8(0)) + } + if let status { + writer.write(UInt8(1)) + writer.write(status.rawValue) + } else { + writer.write(UInt8(0)) + } + writer.write(options) + return writer.bytes + } + + static func updatePermissions(userID: Identifier, permissions: Permissions?) throws -> [UInt8] { + var writer = ByteWriter() + try userID.encode(into: &writer) + writePermissions(permissions, into: &writer) + return writer.bytes + } + + static func changePassword(userID: Identifier, currentPassword: String, newPassword: String) throws -> [UInt8] { + try validatePassword(currentPassword) + try validatePassword(newPassword) + var writer = ByteWriter() + try userID.encode(into: &writer) + try writer.writeName(currentPassword) + try writer.writeName(newPassword) + return writer.bytes + } + + /// `[protocol_version:u32][sdk_name][sdk_version]`, the prefix of both + /// login-register request shapes. + static func versionInfo(into writer: inout ByteWriter) throws { + writer.write(ProtocolVersion.current.packed) + try writer.writeName(IggyVersion.sdkName) + try writer.writeName(IggyVersion.sdkVersion) + } + + static func loginRegister(username: String, password: String, bindSecret: [UInt8] = Array(repeating: 0, count: bindSecretSize)) throws -> [UInt8] { + try validateUsername(username) + try validatePassword(password) + guard bindSecret.count == bindSecretSize else { + throw IggyError(.invalidFormat, context: "bind secret must be exactly \(bindSecretSize) bytes") + } + var writer = ByteWriter() + try versionInfo(into: &writer) + writer.write(bindSecret) + try writer.writeName(username) + try writer.writeName(password) + writer.write(UInt32(0)) + return writer.bytes + } + + static func loginRegisterWithPersonalAccessToken(token: String, bindSecret: [UInt8] = Array(repeating: 0, count: bindSecretSize)) throws -> [UInt8] { + let length = token.utf8.count + guard length >= 1, length <= Identifier.maxNameLength else { + throw IggyError(.invalidFormat, context: "token must be 1-255 bytes") + } + guard bindSecret.count == bindSecretSize else { + throw IggyError(.invalidFormat, context: "bind secret must be exactly \(bindSecretSize) bytes") + } + var writer = ByteWriter() + try versionInfo(into: &writer) + writer.write(bindSecret) + try writer.writeName(token) + writer.write(UInt32(0)) + return writer.bytes + } + + /// `[ClientVersionInfo][client_id:u128LE][session:u64LE][metadata_watermark:u64LE][bind_secret:32]` + static func bindSession(clientID: UInt128Value, session: UInt64, metadataWatermark: UInt64, bindSecret: [UInt8]) throws -> [UInt8] { + guard bindSecret.count == bindSecretSize else { + throw IggyError(.invalidFormat, context: "bind secret must be exactly \(bindSecretSize) bytes") + } + var writer = ByteWriter() + try versionInfo(into: &writer) + writer.write(clientID) + writer.write(session) + writer.write(metadataWatermark) + writer.write(bindSecret) + return writer.bytes + } + + // MARK: Personal access tokens + + static func createPersonalAccessToken(name: String, expiry: PersonalAccessTokenExpiry) throws -> [UInt8] { + var writer = ByteWriter() + try writer.writeName(validatedWireName(name)) + writer.write(try expiry.wireMicroseconds() ?? 0) + return writer.bytes + } + + static func nameOnly(_ name: String) throws -> [UInt8] { + var writer = ByteWriter() + try writer.writeName(validatedWireName(name)) + return writer.bytes + } + + // MARK: System + + static func getClient(clientID: UInt32) -> [UInt8] { + clientID.littleEndianBytes + } + + static func getSnapshot(compression: SnapshotCompression, types: [SystemSnapshotType]) throws -> [UInt8] { + guard types.count <= Int(UInt8.max) else { + throw IggyError(.invalidCommand, context: "too many snapshot types") + } + var writer = ByteWriter() + writer.write(compression.rawValue) + writer.write(UInt8(types.count)) + for type in types { + writer.write(type.rawValue) + } + return writer.bytes + } + + static func describeOptions(scope: OptionsScope) -> [UInt8] { + [scope.rawValue] + } +} + +/// Validates a wire name and returns it, so the length prefix can never +/// desync the frame. +func validatedWireName(_ value: String, what: String = "name") throws -> String { + let length = value.utf8.count + guard length >= 1, length <= Identifier.maxNameLength else { + throw IggyError(.invalidFormat, context: "\(what) must be 1-\(Identifier.maxNameLength) bytes, got \(length)") + } + return value +} diff --git a/foreign/swift/Sources/Iggy/Protocol/Responses.swift b/foreign/swift/Sources/Iggy/Protocol/Responses.swift new file mode 100644 index 0000000000..c0efec35c6 --- /dev/null +++ b/foreign/swift/Sources/Iggy/Protocol/Responses.swift @@ -0,0 +1,451 @@ +// 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. + +/// Response body decoders, one per reply shape, mirroring +/// `core/binary_protocol/src/responses` and the domain conversions of +/// `core/common/src/wire_conversions.rs`. Every decoder takes the body that +/// follows the consensus header, with the result section already stripped. +enum Responses { + /// Sentinel the wire uses for "no authenticated user". + private static let noUserID = UInt32.max + + /// Runs a decoder, mapping codec failures onto the `invalidCommand` error + /// the Rust SDK raises for an unreadable body. + static func decode(_ body: ArraySlice, _ decoder: (inout ByteReader) throws -> T) throws -> T { + var reader = ByteReader(body) + do { + return try decoder(&reader) + } catch let error as WireError { + throw error.iggyError + } + } + + // MARK: Streams + + static func stream(from reader: inout ByteReader) throws -> StreamInfo { + let id = try reader.readUInt32() + let createdAt = try reader.readUInt64() + let topicsCount = try reader.readUInt32() + let sizeBytes = try reader.readUInt64() + let messagesCount = try reader.readUInt64() + let name = try reader.readName() + let options = try OptionsBlock.decode(try OptionsBlock.readPrefixed(from: &reader), explicit: true) + return StreamInfo( + id: id, createdAt: IggyTimestamp(microseconds: createdAt), name: name, sizeBytes: sizeBytes, messagesCount: messagesCount, topicsCount: topicsCount, + options: options) + } + + static func streamDetails(_ body: ArraySlice) throws -> StreamDetails { + try decode(body) { reader in + let stream = try stream(from: &reader) + var topics: [Topic] = [] + for _ in 0..) throws -> [StreamInfo] { + try decode(body) { reader in + var streams: [StreamInfo] = [] + while !reader.isAtEnd { + streams.append(try stream(from: &reader)) + } + return streams + } + } + + // MARK: Topics + + static func topic(from reader: inout ByteReader) throws -> Topic { + let id = try reader.readUInt32() + let createdAt = try reader.readUInt64() + let partitionsCount = try reader.readUInt32() + let messageExpiry = try reader.readUInt64() + let compression = try reader.readUInt8() + let maxTopicSize = try reader.readUInt64() + let sizeBytes = try reader.readUInt64() + let messagesCount = try reader.readUInt64() + let name = try reader.readName() + let explicit = try OptionsBlock.readPrefixed(from: &reader) + let derived = try OptionsBlock.readPrefixed(from: &reader) + guard let compressionAlgorithm = CompressionAlgorithm(rawValue: compression) else { + throw IggyError(.invalidCommand, context: "unknown compression algorithm \(compression)") + } + // A stored expiry of zero means the topic never expires, unlike the + // request side where zero is the server default. + let expiry: IggyExpiry = messageExpiry == 0 ? .never : IggyExpiry(wireMicroseconds: messageExpiry) + return Topic( + id: id, createdAt: IggyTimestamp(microseconds: createdAt), name: name, sizeBytes: sizeBytes, messageExpiry: expiry, + compressionAlgorithm: compressionAlgorithm, maxTopicSize: MaxTopicSize(wireValue: maxTopicSize), + messagesCount: messagesCount, partitionsCount: partitionsCount, + options: try OptionsBlock.decodeSplit(explicit: explicit, derived: derived)) + } + + static func partition(from reader: inout ByteReader) throws -> Partition { + Partition( + id: try reader.readUInt32(), createdAt: IggyTimestamp(microseconds: try reader.readUInt64()), + segmentsCount: try reader.readUInt32(), currentOffset: try reader.readUInt64(), + sizeBytes: try reader.readUInt64(), messagesCount: try reader.readUInt64()) + } + + static func topicDetails(_ body: ArraySlice) throws -> TopicDetails { + try decode(body) { reader in + let topic = try topic(from: &reader) + var partitions: [Partition] = [] + for _ in 0..) throws -> [Topic] { + try decode(body) { reader in + let count = try reader.readUInt32() + var topics: [Topic] = [] + for _ in 0.. ConsumerGroup { + let id = try reader.readUInt32() + let partitionsCount = try reader.readUInt32() + let membersCount = try reader.readUInt32() + let name = try reader.readName() + return ConsumerGroup(id: id, name: name, partitionsCount: partitionsCount, membersCount: membersCount) + } + + static func consumerGroupDetails(_ body: ArraySlice) throws -> ConsumerGroupDetails { + try decode(body) { reader in + let group = try consumerGroup(from: &reader) + var members: [ConsumerGroupMember] = [] + for _ in 0..) throws -> [ConsumerGroup] { + try decode(body) { reader in + var groups: [ConsumerGroup] = [] + while !reader.isAtEnd { + groups.append(try consumerGroup(from: &reader)) + } + return groups + } + } + + static func consumerGroupAssignment(_ body: ArraySlice) throws -> ConsumerGroupAssignment { + try decode(body) { reader in + let generation = try reader.readUInt64() + let count = try reader.readUInt32() + var partitions: [UInt32] = [] + for _ in 0..) throws -> ConsumerOffsetInfo { + try decode(body) { reader in + ConsumerOffsetInfo(partitionID: try reader.readUInt32(), currentOffset: try reader.readUInt64(), storedOffset: try reader.readUInt64()) + } + } + + // MARK: Clients + + private static func transportName(_ code: UInt8) -> String { + switch code { + case 1: "TCP" + case 2: "QUIC" + case 3: "HTTP" + case 4: "WebSocket" + default: "Unknown" + } + } + + static func client(from reader: inout ByteReader) throws -> ClientInfo { + let clientID = try reader.readUInt32() + let userID = try reader.readUInt32() + let transport = try reader.readUInt8() + let address = try reader.readLongString() + let consumerGroupsCount = try reader.readUInt32() + return ClientInfo( + clientID: clientID, userID: userID == noUserID ? nil : userID, address: address, transport: transportName(transport), + consumerGroupsCount: consumerGroupsCount) + } + + static func clientDetails(_ body: ArraySlice) throws -> ClientInfoDetails { + try decode(body) { reader in + let client = try client(from: &reader) + var groups: [ConsumerGroupInfo] = [] + for _ in 0..) throws -> [ClientInfo] { + try decode(body) { reader in + var clients: [ClientInfo] = [] + while !reader.isAtEnd { + clients.append(try client(from: &reader)) + } + return clients + } + } + + // MARK: Users + + static func user(from reader: inout ByteReader) throws -> UserInfo { + let id = try reader.readUInt32() + let createdAt = try reader.readUInt64() + let statusCode = try reader.readUInt8() + let username = try reader.readName() + let options = try OptionsBlock.decode(try OptionsBlock.readPrefixed(from: &reader), explicit: true) + guard let status = UserStatus(rawValue: statusCode) else { + throw IggyError(.invalidUserStatus, context: "unknown user status \(statusCode)") + } + return UserInfo(id: id, createdAt: IggyTimestamp(microseconds: createdAt), status: status, username: username, options: options) + } + + static func userDetails(_ body: ArraySlice) throws -> UserInfoDetails { + try decode(body) { reader in + let user = try user(from: &reader) + let flag = try reader.readUInt8() + var permissions: Permissions? + if flag == 0 { + try reader.skip(3) + } else { + var permissionsReader = ByteReader(try reader.readPrefixedBytes()) + permissions = try Permissions.decode(from: &permissionsReader) + } + return UserInfoDetails( + id: user.id, createdAt: user.createdAt, status: user.status, username: user.username, permissions: permissions, options: user.options) + } + } + + static func users(_ body: ArraySlice) throws -> [UserInfo] { + try decode(body) { reader in + var users: [UserInfo] = [] + while !reader.isAtEnd { + users.append(try user(from: &reader)) + } + return users + } + } + + struct LoginRegister { + let userID: UInt32 + let session: UInt64 + let serverProtocolVersion: ProtocolVersion + let serverVersion: String + } + + static func loginRegister(_ body: ArraySlice) throws -> LoginRegister { + try decode(body) { reader in + LoginRegister( + userID: try reader.readUInt32(), session: try reader.readUInt64(), + serverProtocolVersion: ProtocolVersion(packed: try reader.readUInt32()), serverVersion: try reader.readName()) + } + } + + static func identity(_ body: ArraySlice) throws -> IdentityInfo { + try decode(body) { reader in IdentityInfo(userID: try reader.readUInt32()) } + } + + // MARK: Personal access tokens + + static func rawPersonalAccessToken(_ body: ArraySlice) throws -> RawPersonalAccessToken { + try decode(body) { reader in RawPersonalAccessToken(token: try reader.readName()) } + } + + static func personalAccessTokens(_ body: ArraySlice) throws -> [PersonalAccessTokenInfo] { + try decode(body) { reader in + var tokens: [PersonalAccessTokenInfo] = [] + while !reader.isAtEnd { + let name = try reader.readName() + let expiry = try reader.readUInt64() + tokens.append(PersonalAccessTokenInfo(name: name, expiresAt: expiry == 0 ? nil : IggyTimestamp(microseconds: expiry))) + } + return tokens + } + } + + // MARK: System + + static func stats(_ body: ArraySlice) throws -> Stats { + try decode(body) { reader in + let processID = try reader.readUInt32() + let cpuUsage = try reader.readFloat() + let totalCPUUsage = try reader.readFloat() + let memoryUsage = try reader.readUInt64() + let totalMemory = try reader.readUInt64() + let availableMemory = try reader.readUInt64() + let runTime = try reader.readUInt64() + let startTime = try reader.readUInt64() + let readBytes = try reader.readUInt64() + let writtenBytes = try reader.readUInt64() + let messagesSizeBytes = try reader.readUInt64() + let streamsCount = try reader.readUInt32() + let topicsCount = try reader.readUInt32() + let partitionsCount = try reader.readUInt32() + let segmentsCount = try reader.readUInt32() + let messagesCount = try reader.readUInt64() + let clientsCount = try reader.readUInt32() + let consumerGroupsCount = try reader.readUInt32() + let hostname = try reader.readLongString() + let osName = try reader.readLongString() + let osVersion = try reader.readLongString() + let kernelVersion = try reader.readLongString() + let serverVersion = try reader.readLongString() + let semver = try reader.readUInt32() + let cacheCount = try reader.readUInt32() + var cacheMetrics: [CacheMetrics] = [] + for _ in 0..) throws -> ClusterMetadata { + try decode(body) { reader in + let name = try reader.readLongString() + let count = try reader.readUInt32() + var nodes: [ClusterNode] = [] + for _ in 0..) throws -> [OptionSpec] { + try decode(body) { reader in + let count = try reader.readUInt32() + var specs: [OptionSpec] = [] + for _ in 0..) throws -> SendMessagesResponse { + if body.isEmpty { + return SendMessagesResponse() + } + return try decode(body) { reader in + let count = try reader.readUInt32() + var confirmations: [SendConfirmation] = [] + for _ in 0.., verifyChecksums: Bool = false) throws -> PolledMessages { + try decode(body) { reader in + let partitionID = try reader.readUInt32() + let currentOffset = try reader.readUInt64() + let count = try reader.readUInt32() + let messages = try Batch.decodeMessages(reader.rest, expectedCount: count, verifyChecksums: verifyChecksums) + return PolledMessages(partitionID: partitionID, currentOffset: currentOffset, count: count, messages: messages) + } + } +} diff --git a/foreign/swift/Sources/Iggy/Protocol/VSRFrame.swift b/foreign/swift/Sources/Iggy/Protocol/VSRFrame.swift new file mode 100644 index 0000000000..6913ba4717 --- /dev/null +++ b/foreign/swift/Sources/Iggy/Protocol/VSRFrame.swift @@ -0,0 +1,301 @@ +// 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. + +/// The consensus wire framing the SDK speaks over TCP: every frame in both +/// directions is a fixed 256-byte header followed by the body, with no outer +/// length prefix. A port of `core/binary_protocol/src/consensus/header.rs`. +/// +/// Fields are addressed by byte offset because the header leads with +/// unaligned 128-bit values. +enum VSRFrame { + /// Length of every consensus frame header, both directions. + static let headerSize = 256 + /// Upper bound on a frame the client is willing to read; a corrupt size + /// field must not turn into a multi-gigabyte allocation. + static let maxFrameSize = 1 << 30 + + /// Request header field offsets the client writes. + enum RequestOffset { + static let size = 48 + static let command = 60 + static let client = 128 + static let requestChecksum = 144 + static let request = 168 + static let operation = 176 + static let session = 184 + static let reserved = 196 + } + + /// Reply header field offsets the client reads. + enum ReplyOffset { + static let size = 48 + static let command = 60 + static let request = 200 + static let operation = 208 + static let status = 216 + } + + /// Eviction header field offsets the client reads. + enum EvictionOffset { + static let serverProtocolVersion = 144 + static let serverProtocolVersionMin = 148 + static let reason = 255 + } + + /// Frame discriminant, at the same offset in every consensus header. + enum Command: UInt8 { + case request = 5 + case reply = 8 + case eviction = 13 + } + + /// Terminal reason carried by an eviction frame. Pinned wire values. + enum EvictionReason: UInt8 { + case reserved = 0 + case noSession = 1 + case clientReleaseTooLow = 2 + case clientReleaseTooHigh = 3 + case invalidRequestOperation = 4 + case invalidRequestBody = 5 + case invalidRequestBodySize = 6 + case sessionTooLow = 7 + case sessionReleaseMismatch = 8 + case invalidCredentials = 9 + case invalidToken = 10 + case userInactive = 11 + case sessionError = 12 + case staleClient = 13 + case incompatibleProtocol = 14 + case malformedLogin = 15 + } + + /// A request frame ready for the wire, plus the id it was stamped with. + struct EncodedRequest { + let bytes: [UInt8] + let requestID: UInt64 + let operation: ConsensusOperation + } + + /// Builds the complete request frame for a command code and its encoded + /// payload, deriving sequencing from the command code and the session. + /// + /// The request checksum is stamped for the operations the server's client + /// table deduplicates, like the Rust SDK does; partition and + /// non-replicated operations leave it zero, which the server reads as + /// unstamped. + static func encodeRequest(session: inout ConsensusSession, code: UInt32, payload: [UInt8]) throws -> EncodedRequest { + let totalSize = headerSize + payload.count + guard totalSize <= UInt32.max else { + throw IggyError(.invalidConfiguration, context: "request of \(totalSize) bytes exceeds the frame size limit") + } + let operation = ConsensusOperation.forCode(code) + let requestID: UInt64 + let sessionID: UInt64 + switch operation { + case .register: + requestID = session.beginRegister() + sessionID = 0 + case .nonReplicated: + requestID = session.currentRequestID + sessionID = session.session ?? 0 + default: + guard let bound = session.session else { + throw IggyError(.unauthenticated) + } + sessionID = bound + requestID = try session.nextRequestID() + } + let requestChecksum: UInt64 = + (operation.isPartition || operation == .nonReplicated) ? 0 : XXH3.hash64(payload) + + var frame = [UInt8](repeating: 0, count: totalSize) + frame.withUnsafeMutableBytes { buffer in + buffer.storeBytes(of: UInt32(totalSize).littleEndian, toByteOffset: RequestOffset.size, as: UInt32.self) + buffer[RequestOffset.command] = Command.request.rawValue + buffer.storeBytes(of: session.clientID.low.littleEndian, toByteOffset: RequestOffset.client, as: UInt64.self) + buffer.storeBytes(of: session.clientID.high.littleEndian, toByteOffset: RequestOffset.client + 8, as: UInt64.self) + buffer.storeBytes(of: requestChecksum.littleEndian, toByteOffset: RequestOffset.requestChecksum, as: UInt64.self) + buffer.storeBytes(of: requestID.littleEndian, toByteOffset: RequestOffset.request, as: UInt64.self) + buffer[RequestOffset.operation] = operation.rawValue + buffer.storeBytes(of: sessionID.littleEndian, toByteOffset: RequestOffset.session, as: UInt64.self) + if operation == .nonReplicated { + buffer.storeBytes(of: code.littleEndian, toByteOffset: RequestOffset.reserved, as: UInt32.self) + } + } + if !payload.isEmpty { + frame.replaceSubrange(headerSize..., with: payload) + } + return EncodedRequest(bytes: frame, requestID: requestID, operation: operation) + } + + /// Reads the total frame length, header included, from any consensus + /// header. Returns nil when the buffer is too short. + static func readSize(_ header: ArraySlice) -> UInt32? { + guard header.count >= ReplyOffset.size + 4 else { + return nil + } + let start = header.startIndex + ReplyOffset.size + return UInt32(littleEndianBytes: header[start..) -> Command? { + guard header.count > ReplyOffset.command else { + return nil + } + return Command(rawValue: header[header.startIndex + ReplyOffset.command]) + } + + static func readReplyRequestID(_ header: ArraySlice) -> UInt64? { + guard header.count >= ReplyOffset.request + 8 else { + return nil + } + let start = header.startIndex + ReplyOffset.request + return UInt64(littleEndianBytes: header[start..) -> UInt32? { + guard header.count >= ReplyOffset.status + 4 else { + return nil + } + let start = header.startIndex + ReplyOffset.status + return UInt32(littleEndianBytes: header[start..) -> ConsensusOperation? { + guard header.count > ReplyOffset.operation else { + return nil + } + return ConsensusOperation(rawValue: header[header.startIndex + ReplyOffset.operation]) + } + + /// The decoded fields of an eviction frame. + struct Eviction { + let reason: EvictionReason? + let rawReason: UInt8 + let serverProtocolVersion: UInt32 + let serverProtocolVersionMin: UInt32 + + /// Maps the eviction to the typed error the Rust SDK raises for it. + var error: IggyError { + switch reason { + case .invalidCredentials: + return IggyError(.invalidCredentials) + case .invalidToken: + return IggyError(.invalidPersonalAccessToken) + case .noSession, .sessionTooLow, .sessionReleaseMismatch, .userInactive, .sessionError: + return IggyError(.unauthenticated) + case .staleClient: + return IggyError(.staleClient) + case .incompatibleProtocol: + if serverProtocolVersionMin == 0 || serverProtocolVersion < serverProtocolVersionMin { + return IggyError(.unauthenticated) + } + let client = ProtocolVersion.current + let min = ProtocolVersion(packed: serverProtocolVersionMin) + let max = ProtocolVersion(packed: serverProtocolVersion) + return IggyError( + .incompatibleProtocolVersion, + context: "client \(client), server accepts [\(min), \(max)]") + case .malformedLogin: + return IggyError(.invalidFormat) + default: + // An unknown reason must not map to a reconnectable error, or a + // newer server's new reason drives a disconnect and re-login loop. + return IggyError(.invalidCommand, context: "evicted with reason \(rawReason)") + } + } + } + + static func readEviction(_ header: ArraySlice) -> Eviction? { + guard header.count >= headerSize else { + return nil + } + let start = header.startIndex + let rawReason = header[start + EvictionOffset.reason] + let version = UInt32(littleEndianBytes: header[start + EvictionOffset.serverProtocolVersion.., body: ArraySlice) throws -> ArraySlice { + guard header.count == headerSize, let eviction = readEviction(header) else { + throw IggyError(.invalidCommand, context: "reply header is \(header.count) bytes, expected \(headerSize)") + } + switch peekCommand(header) { + case .eviction: + throw eviction.error + case .reply: + break + default: + throw IggyError(.invalidCommand, context: "unexpected frame command") + } + // Compared as `UInt32` so a hostile size never traps a 32-bit `Int`. + guard let size = readSize(header), size >= UInt32(headerSize), size <= UInt32(maxFrameSize) else { + throw IggyError(.invalidCommand, context: "reply declares an invalid frame size") + } + let expected = Int(size) - headerSize + guard body.count >= expected else { + throw IggyError(.invalidCommand, context: "reply body is shorter than declared") + } + if let status = readReplyStatus(header), status != 0 { + throw IggyError(wireCode: status) + } + guard let operation = readReplyOperation(header) else { + throw IggyError(.invalidCommand, context: "reply carries an unknown operation") + } + // A send the server classified as a duplicate of an earlier request is + // answered with an empty status-0 body: the batch is already in the + // log, so it reads as a success with no confirmations to report. + return try splitMetadataResult(operation: operation, body: body.prefix(expected)) + } + + /// Strips the committed result section leading a result-framed reply body: + /// `[count: u32]` then `count` x `{index: u32, result: u32}`. Success is + /// count 0 followed by the payload; a committed rejection carries one + /// entry and no payload. A register reply is result-framed too, except a + /// terminal register failure ships an empty body, which passes through so + /// the typed decode fails. + static func splitMetadataResult(operation: ConsensusOperation, body: ArraySlice) throws -> ArraySlice { + let framed = operation.isResultFramed || (operation == .register && !body.isEmpty) + guard framed else { + return body + } + guard body.count >= 4 else { + throw IggyError(.invalidCommand, context: "result section is truncated") + } + let start = body.startIndex + let count = UInt32(littleEndianBytes: body[start.. 0 { + let code = UInt32(littleEndianBytes: body[start + 8.. String { + mutating func skip(_ count: Int) throws { + try require(count) + offset += count + } + + /// The bytes not yet consumed, without consuming them. + var rest: ArraySlice { bytes[offset...] } + + /// `[len: u32][bytes]`, the layout of every length-prefixed block. + mutating func readPrefixedBytes() throws -> ArraySlice { let declared = try readUInt32() // A length that does not fit the platform's Int cannot be in the // buffer either; report it as truncation rather than trapping. guard let length = Int(exactly: declared) else { throw WireError.truncated(offset: position, need: Int.max, have: remaining) } - return try readString(length) + return try readBytes(length) + } + + /// `[len: u32][utf8]`. + mutating func readLongString() throws -> String { + let start = position + 4 + guard let value = String(bytes: try readPrefixedBytes(), encoding: .utf8) else { + throw WireError.invalidUTF8(offset: start) + } + return value } } diff --git a/foreign/swift/Tests/IggyTests/ModelTests.swift b/foreign/swift/Tests/IggyTests/ModelTests.swift index 72d778fa97..19be03941d 100644 --- a/foreign/swift/Tests/IggyTests/ModelTests.swift +++ b/foreign/swift/Tests/IggyTests/ModelTests.swift @@ -244,7 +244,7 @@ struct IggyMessageTests { } @Test func descriptionShowsABoundedPreview() throws { - let short = try IggyMessage("hello") + let short = try IggyMessage(String("hello")) #expect(short.description == "[0] ID:0 'hello'") // 46 ASCII bytes then a two-byte "é" straddling the 47-byte cut. let straddling = try IggyMessage(String(repeating: "a", count: 46) + "éé" + String(repeating: "b", count: 20)) diff --git a/foreign/swift/Tests/IggyTests/ProtocolTests.swift b/foreign/swift/Tests/IggyTests/ProtocolTests.swift new file mode 100644 index 0000000000..85b69e2327 --- /dev/null +++ b/foreign/swift/Tests/IggyTests/ProtocolTests.swift @@ -0,0 +1,236 @@ +// 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 + +@Suite("Consensus session") +struct ConsensusSessionTests { + @Test func freshSessionIsUnbound() { + let session = ConsensusSession() + #expect(!session.isBound) + #expect(session.session == nil) + #expect(!session.clientID.isZero) + } + + @Test func requestIDsAreMonotonicAfterBinding() throws { + var session = ConsensusSession(clientID: 1) + #expect(session.beginRegister() == 0) + try session.bind(10) + #expect(try session.nextRequestID() == 1) + #expect(try session.nextRequestID() == 2) + #expect(session.currentRequestID == 3) + } + + @Test func nextRequestBeforeBindingFails() { + var session = ConsensusSession(clientID: 1) + #expect(throws: IggyError(.unauthenticated, context: "request id taken before the session is bound")) { + try session.nextRequestID() + } + } + + @Test func doubleBindAndZeroSessionFail() throws { + var session = ConsensusSession(clientID: 1) + try session.bind(10) + #expect(throws: IggyError(.alreadyAuthenticated)) { + try session.bind(20) + } + var other = ConsensusSession(clientID: 1) + #expect(throws: IggyError.self) { + try other.bind(0) + } + } + + @Test func reRegisterMintsAFreshIdentity() throws { + var session = ConsensusSession(clientID: 7) + _ = session.beginRegister() + #expect(session.clientID == 7) + try session.bind(42) + #expect(session.beginRegister() == 0) + #expect(!session.isBound) + #expect(session.clientID != 7) + } + + @Test func resetDropsEverything() throws { + var session = ConsensusSession(clientID: 7) + _ = session.beginRegister() + try session.bind(42) + session.reset() + #expect(!session.isBound) + #expect(session.clientID != 7) + #expect(session.currentRequestID == 1) + } +} + +@Suite("Reply decoding") +struct ReplyDecodingTests { + func replyHeader(operation: ConsensusOperation, bodyLength: Int, status: UInt32 = 0, request: UInt64 = 1) -> [UInt8] { + var header = [UInt8](repeating: 0, count: VSRFrame.headerSize) + header.replaceSubrange(48..<52, with: UInt32(VSRFrame.headerSize + bodyLength).littleEndianBytes) + header[60] = VSRFrame.Command.reply.rawValue + header.replaceSubrange(200..<208, with: request.littleEndianBytes) + header[208] = operation.rawValue + header.replaceSubrange(216..<220, with: status.littleEndianBytes) + return header + } + + func successBody(_ payload: [UInt8]) -> [UInt8] { + UInt32(0).littleEndianBytes + payload + } + + func rejectionBody(_ code: UInt32) -> [UInt8] { + UInt32(1).littleEndianBytes + UInt32(0).littleEndianBytes + code.littleEndianBytes + } + + @Test func metadataSuccessStripsTheResultSection() throws { + let body = successBody([7, 8, 9]) + let out = try VSRFrame.decodeReply(header: replyHeader(operation: .createStream, bodyLength: body.count)[...], body: body[...]) + #expect(Array(out) == [7, 8, 9]) + } + + @Test func metadataRejectionMapsTheCommittedCode() { + let body = rejectionBody(IggyErrorCode.streamIdNotFound.rawValue) + #expect(throws: IggyError(.streamIdNotFound)) { + try VSRFrame.decodeReply(header: replyHeader(operation: .deleteStream, bodyLength: body.count)[...], body: body[...]) + } + } + + @Test func transientCodesSurfaceAsTyped() { + let body = rejectionBody(IggyErrorCode.transientNotCommitted.rawValue) + #expect(throws: IggyError(.transientNotCommitted)) { + try VSRFrame.decodeReply(header: replyHeader(operation: .createStream, bodyLength: body.count)[...], body: body[...]) + } + } + + @Test func truncatedResultSectionIsNeverSuccess() { + let body = UInt32(1).littleEndianBytes + #expect(throws: IggyError.self) { + try VSRFrame.decodeReply(header: replyHeader(operation: .createStream, bodyLength: body.count)[...], body: body[...]) + } + } + + @Test func nonMetadataBodiesPassThrough() throws { + let body = rejectionBody(IggyErrorCode.invalidOffset.rawValue) + let out = try VSRFrame.decodeReply(header: replyHeader(operation: .sendMessages, bodyLength: body.count)[...], body: body[...]) + #expect(Array(out) == body) + let read = try VSRFrame.decodeReply(header: replyHeader(operation: .nonReplicated, bodyLength: 3)[...], body: [1, 2, 3]) + #expect(Array(read) == [1, 2, 3]) + } + + @Test func consumerOffsetOpsAreResultFramed() throws { + let body = rejectionBody(IggyErrorCode.consumerOffsetNotFound.rawValue) + #expect(throws: IggyError(.consumerOffsetNotFound)) { + try VSRFrame.decodeReply(header: replyHeader(operation: .deleteConsumerOffset, bodyLength: body.count)[...], body: body[...]) + } + let out = try VSRFrame.decodeReply(header: replyHeader(operation: .storeConsumerOffset, bodyLength: 4)[...], body: successBody([])[...]) + #expect(out.isEmpty) + } + + @Test func registerRepliesAreFramedOnlyWhenNonEmpty() throws { + let out = try VSRFrame.decodeReply(header: replyHeader(operation: .register, bodyLength: 0)[...], body: []) + #expect(out.isEmpty) + let body = successBody([1, 2]) + let framed = try VSRFrame.decodeReply(header: replyHeader(operation: .register, bodyLength: body.count)[...], body: body[...]) + #expect(Array(framed) == [1, 2]) + } + + @Test func emptySendReplyIsADuplicateWithNoConfirmations() throws { + let body = try VSRFrame.decodeReply(header: replyHeader(operation: .sendMessages, bodyLength: 0)[...], body: []) + #expect(body.isEmpty) + } + + @Test func statusDenialWinsBeforeTheBody() { + let header = replyHeader(operation: .createStream, bodyLength: 0, status: IggyErrorCode.unauthorized.rawValue) + #expect(throws: IggyError(.unauthorized)) { + try VSRFrame.decodeReply(header: header[...], body: []) + } + } + + @Test func unknownStatusKeepsTheRawCode() { + let header = replyHeader(operation: .createStream, bodyLength: 0, status: 65_000) + do { + _ = try VSRFrame.decodeReply(header: header[...], body: []) + Issue.record("expected a failure") + } catch let error as IggyError { + #expect(error.code == .error) + #expect(error.rawCode == 65_000) + } catch { + Issue.record("unexpected error \(error)") + } + } + + @Test func invalidFramesAreRejected() { + var bogus = replyHeader(operation: .createStream, bodyLength: 0) + bogus[60] = 99 + #expect(throws: IggyError(.invalidCommand, context: "unexpected frame command")) { + try VSRFrame.decodeReply(header: bogus[...], body: []) + } + var short = replyHeader(operation: .createStream, bodyLength: 0) + short.replaceSubrange(48..<52, with: UInt32(10).littleEndianBytes) + #expect(throws: IggyError.self) { + try VSRFrame.decodeReply(header: short[...], body: []) + } + var unknownOperation = replyHeader(operation: .createStream, bodyLength: 0) + unknownOperation[208] = 200 + #expect(throws: IggyError.self) { + try VSRFrame.decodeReply(header: unknownOperation[...], body: []) + } + #expect(throws: IggyError.self) { + try VSRFrame.decodeReply(header: replyHeader(operation: .createStream, bodyLength: 8)[...], body: [1, 2]) + } + // A header shorter than a frame is an error, never a trap, and the + // raw accessors answer nil for it. + let partial = replyHeader(operation: .createStream, bodyLength: 0).prefix(100) + #expect(throws: IggyError(.invalidCommand)) { try VSRFrame.decodeReply(header: partial, body: []) } + #expect(VSRFrame.readReplyRequestID(partial) == nil) + #expect(VSRFrame.readReplyStatus(partial) == nil) + #expect(VSRFrame.readReplyOperation(partial) == nil) + #expect(VSRFrame.readEviction(partial) == nil) + #expect(VSRFrame.peekCommand(partial.prefix(10)) == nil) + } + + @Test func unknownEvictionReasonMapsToInvalidCommand() throws { + var header = [UInt8](repeating: 0, count: VSRFrame.headerSize) + header[60] = VSRFrame.Command.eviction.rawValue + header[255] = 200 + let eviction = try #require(VSRFrame.readEviction(header[...])) + #expect(eviction.reason == nil) + #expect(eviction.error.code == .invalidCommand) + } + + @Test func evictionReasonsMapLikeTheRustSDK() { + func error(_ reason: VSRFrame.EvictionReason) -> IggyErrorCode? { + var header = [UInt8](repeating: 0, count: VSRFrame.headerSize) + header[60] = VSRFrame.Command.eviction.rawValue + header[255] = reason.rawValue + return VSRFrame.readEviction(header[...])?.error.code + } + #expect(error(.invalidCredentials) == .invalidCredentials) + #expect(error(.invalidToken) == .invalidPersonalAccessToken) + #expect(error(.noSession) == .unauthenticated) + #expect(error(.sessionTooLow) == .unauthenticated) + #expect(error(.userInactive) == .unauthenticated) + #expect(error(.staleClient) == .staleClient) + #expect(error(.malformedLogin) == .invalidFormat) + #expect(error(.invalidRequestBody) == .invalidCommand) + // An incompatible-protocol frame with an unusable window degrades to + // an authentication error rather than trusting the remote frame. + #expect(error(.incompatibleProtocol) == .unauthenticated) + } +} diff --git a/foreign/swift/Tests/IggyTests/ResponseVectorTests.swift b/foreign/swift/Tests/IggyTests/ResponseVectorTests.swift new file mode 100644 index 0000000000..7ba408e6fa --- /dev/null +++ b/foreign/swift/Tests/IggyTests/ResponseVectorTests.swift @@ -0,0 +1,281 @@ +// 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 Testing + +@testable import Iggy + +/// Every decoder reads bytes the Rust response encoders produced for known +/// values, so a layout drift on either side fails here. +@Suite("Response decoders match the Rust golden vectors") +struct ResponseVectorTests { + let golden = GoldenFixture.shared + let createdAt = IggyTimestamp(microseconds: 1_710_000_000_000_000) + + func body(_ name: String) -> ArraySlice { + golden.bytes(name)[...] + } + + func expectSampleTopic(_ topic: Topic, id: UInt32, name: String) { + #expect(topic.id == id) + #expect(topic.name == name) + #expect(topic.createdAt == createdAt) + #expect(topic.partitionsCount == 3) + #expect(topic.messageExpiry == .after(.seconds(604_800))) + #expect(topic.compressionAlgorithm == CompressionAlgorithm.none) + #expect(topic.maxTopicSize == .bytes(1_073_741_824)) + #expect(topic.sizeBytes == 4096) + #expect(topic.messagesCount == 100) + #expect(topic.options["durability"] == .explicit("persisted")) + #expect(topic.options["segment_size"] == .explicit(.uint64(1_073_741_824))) + #expect(topic.options["max_topic_size"] == .derived(.uint64(UInt64.max))) + } + + @Test func streams() throws { + let details = try Responses.streamDetails(body("response.get_stream")) + #expect(details.id == 1) + #expect(details.name == "my-stream") + #expect(details.createdAt == createdAt) + #expect(details.topicsCount == 2) + #expect(details.sizeBytes == 2048) + #expect(details.messagesCount == 200) + #expect(details.topics.count == 2) + expectSampleTopic(details.topics[0], id: 1, name: "topic-a") + expectSampleTopic(details.topics[1], id: 2, name: "topic-b") + + let streams = try Responses.streams(body("response.get_streams")) + #expect(streams.map(\.id) == [1, 2]) + #expect(streams.map(\.topicsCount) == [2, 0]) + #expect(streams.allSatisfy { $0.options.isEmpty }) + #expect(try Responses.streams([]).isEmpty) + } + + @Test func topics() throws { + let details = try Responses.topicDetails(body("response.get_topic")) + expectSampleTopic( + Topic( + id: details.id, createdAt: details.createdAt, name: details.name, sizeBytes: details.sizeBytes, messageExpiry: details.messageExpiry, + compressionAlgorithm: details.compressionAlgorithm, maxTopicSize: details.maxTopicSize, messagesCount: details.messagesCount, + partitionsCount: details.partitionsCount, options: details.options), id: 1, name: "my-topic") + #expect(details.partitions.map(\.id) == [0, 1, 2]) + #expect(details.partitions[0].currentOffset == 99) + #expect(details.partitions[0].segmentsCount == 2) + #expect(details.partitions[2].messagesCount == 8) + + let topics = try Responses.topics(body("response.get_topics")) + #expect(topics.map(\.name) == ["events", "logs"]) + #expect(try Responses.topics(body("response.get_topics.empty")).isEmpty) + } + + @Test func consumerGroups() throws { + let details = try Responses.consumerGroupDetails(body("response.consumer_group_details")) + #expect(details.id == 1) + #expect(details.name == "my-group") + #expect(details.partitionsCount == 6) + #expect(details.members.map(\.partitions) == [[0, 1, 2], [3, 4, 5]]) + + let groups = try Responses.consumerGroups(body("response.get_consumer_groups")) + #expect(groups.map(\.name) == ["group-a", "group-b"]) + #expect(groups.map(\.membersCount) == [2, 0]) + + let assignment = try Responses.consumerGroupAssignment(body("response.sync_consumer_group")) + #expect(assignment == ConsumerGroupAssignment(generation: 7, partitions: [0, 2, 4])) + } + + @Test func consumerOffset() throws { + let offset = try Responses.consumerOffset(body("response.consumer_offset")) + #expect(offset == ConsumerOffsetInfo(partitionID: 1, currentOffset: 1000, storedOffset: 500)) + } + + @Test func clients() throws { + let details = try Responses.clientDetails(body("response.client_details")) + #expect(details.clientID == 1) + #expect(details.userID == 10) + #expect(details.transport == "TCP") + #expect(details.address == "127.0.0.1:8080") + #expect(details.consumerGroups == [ConsumerGroupInfo(streamID: 1, topicID: 2, groupID: 3), ConsumerGroupInfo(streamID: 4, topicID: 5, groupID: 6)]) + + let anonymous = try Responses.clientDetails(body("response.client_details.no_user")) + #expect(anonymous.userID == nil) + #expect(anonymous.consumerGroups.isEmpty) + + let clients = try Responses.clients(body("response.get_clients")) + #expect(clients.map(\.clientID) == [1, 2]) + #expect(clients.map(\.transport) == ["TCP", "QUIC"]) + } + + @Test func users() throws { + let details = try Responses.userDetails(body("response.user_details")) + #expect(details.id == 1) + #expect(details.username == "admin") + #expect(details.status == .active) + let permissions = try #require(details.permissions) + #expect(permissions.global.manageServers) + #expect(!permissions.global.manageUsers) + #expect(permissions.streams[1]?.topics[10]?.sendMessages == true) + #expect(permissions.streams[1]?.topics[20]?.manageTopic == false) + #expect(permissions.streams[2]?.manageTopics == true) + #expect(permissions.streams[2]?.topics.isEmpty == true) + + let plain = try Responses.userDetails(body("response.user_details.no_permissions")) + #expect(plain.permissions == nil) + + let users = try Responses.users(body("response.get_users")) + #expect(users.map(\.username) == ["admin", "alice"]) + #expect(users.map(\.status) == [.active, .inactive]) + } + + @Test func loginAndIdentity() throws { + let login = try Responses.loginRegister(body("response.login_register")) + #expect(login.userID == 42) + #expect(login.session == 100) + #expect(login.serverProtocolVersion == .current) + #expect(login.serverVersion == "0.11.0") + #expect(try Responses.identity(body("response.identity")) == IdentityInfo(userID: 42)) + } + + @Test func personalAccessTokens() throws { + #expect(try Responses.rawPersonalAccessToken(body("response.raw_personal_access_token")).token == "raw-secret-token-value") + let tokens = try Responses.personalAccessTokens(body("response.get_personal_access_tokens")) + #expect( + tokens == [ + PersonalAccessTokenInfo(name: "token-a", expiresAt: createdAt), + PersonalAccessTokenInfo(name: "token-b", expiresAt: nil), + ]) + } + + @Test func stats() throws { + let stats = try Responses.stats(body("response.stats")) + #expect(stats.processID == 1234) + #expect(stats.cpuUsage == 25.5) + #expect(stats.totalCPUUsage == 50) + #expect(stats.memoryUsage == 1_073_741_824) + #expect(stats.runTime == .seconds(0.0036)) + #expect(stats.startTime == createdAt) + #expect(stats.messagesCount == 50_000) + #expect(stats.hostname == "node-1") + #expect(stats.osName == "Linux") + #expect(stats.kernelVersion == "6.1.0") + #expect(stats.iggyServerVersion == "0.11.0") + #expect(stats.iggyServerSemver == 11_264) + #expect(stats.cacheMetrics.count == 1) + #expect(stats.cacheMetrics[0].hits == 1000) + #expect(stats.threadsCount == 16) + #expect(stats.openFilesCount == 128) + #expect(stats.openFilesLimit == 10_240) + // A server predating the open-files tail ends the reply after the + // disk fields; a partial tail is a decode error. + let legacy = try Responses.stats(body("response.stats").dropLast(16)[...]) + #expect(legacy.totalDiskSpace == stats.totalDiskSpace) + #expect(legacy.openFilesCount == 0 && legacy.openFilesLimit == 0) + #expect(throws: IggyError.self) { try Responses.stats(body("response.stats").dropLast(8)[...]) } + #expect(stats.totalDiskSpace == 512_110_190_592) + } + + @Test func clusterMetadata() throws { + let metadata = try Responses.clusterMetadata(body("response.cluster_metadata")) + #expect(metadata.name == "prod-cluster") + #expect(metadata.nodes.count == 2) + #expect(metadata.nodes[0].role == .leader) + #expect(metadata.nodes[0].status == .healthy) + #expect(metadata.nodes[0].tcpAddress == "10.0.0.1:8090") + #expect(metadata.nodes[1].role == .follower) + #expect(metadata.nodes[1].status == .unreachable) + #expect(metadata.nodes[1].tcpAddress == "[fd00::2]:8091") + #expect(metadata.nodes[1].endpoints.quic == 0) + } + + @Test func optionSpecs() throws { + let specs = try Responses.optionSpecs(body("response.describe_options")) + #expect(specs.map(\.key) == ["segment_size", "durability"]) + #expect(specs[0].kind == .uint64) + #expect(specs[0].defaultValue == UInt64(1_073_741_824).littleEndianBytes) + #expect(specs[0].description == "Segment size") + #expect(specs[1].kind == .string) + #expect(specs[1].defaultValue == Array("replicated".utf8)) + #expect(specs[1].description.isEmpty) + } + + @Test func sendConfirmations() throws { + let response = try Responses.sendConfirmations(body("response.send_messages")) + #expect( + response.confirmations == [ + SendConfirmation(streamID: 1, topicID: 2, partitionID: 3, baseOffset: 4), + SendConfirmation(streamID: 1, topicID: 2, partitionID: 0, baseOffset: 42), + ]) + #expect(try Responses.sendConfirmations(body("response.send_messages.empty")).confirmations.isEmpty) + #expect(try Responses.sendConfirmations([]).confirmations.isEmpty) + #expect(throws: IggyError.self) { + try Responses.sendConfirmations(golden.bytes("response.send_messages")[...] + [0xFF]) + } + } + + @Test func polledMessages() throws { + let polled = try Responses.polledMessages(body("response.poll_messages"), verifyChecksums: true) + #expect(polled.partitionID == 7) + #expect(polled.currentOffset == 102) + #expect(polled.count == 3) + #expect(polled.messages.count == 3) + #expect(polled.messages.map(\.id) == [11, 12, 13]) + #expect(polled.messages.map(\.offset) == [100, 101, 102]) + #expect(polled.messages.map(\.timestamp.microseconds) == [5_000, 5_000, 6_000]) + #expect(polled.messages.map(\.originTimestamp.microseconds) == [1_000, 1_010, 2_000]) + #expect(polled.messages.map(\.payloadString) == ["a", "b", "c"]) + #expect(polled.messages[1].userHeaders == ["k": "v"]) + #expect(polled.messages[0].userHeaders == nil) + + let empty = try Responses.polledMessages(body("response.poll_messages.empty")) + #expect(empty.messages.isEmpty) + + var corrupted = golden.bytes("response.poll_messages") + corrupted[corrupted.count - 1] ^= 0xFF + #expect(throws: IggyError.self) { + try Responses.polledMessages(corrupted[...], verifyChecksums: true) + } + var truncated = golden.bytes("response.poll_messages") + truncated.removeLast() + #expect(throws: IggyError.self) { + try Responses.polledMessages(truncated[...]) + } + } + + /// The batch header sits after the 16-byte poll prefix; `batch_length` + /// is at +32 and `message_count` at +48 inside it. + @Test func polledMessagesRejectHostileBatchHeaders() throws { + let batchStart = 16 + var hostileLength = golden.bytes("response.poll_messages") + hostileLength.replaceSubrange(batchStart + 32.. [UInt8] { + try Requests.identifierOnly(id) + } + + var sampleOptions: ResourceOptions { + ["durability": .explicit("persisted"), "segment_size": .explicit(.uint64(1_073_741_824))] + } + + var samplePermissions: Permissions { + Permissions( + global: GlobalPermissions(manageServers: true, readServers: true, readUsers: true, readStreams: true, readTopics: true, pollMessages: true), + streams: [ + 1: StreamPermissions( + manageStream: true, readStream: true, readTopics: true, pollMessages: true, + topics: [ + 10: TopicPermissions(manageTopic: true, readTopic: true, sendMessages: true), + 20: TopicPermissions(readTopic: true, pollMessages: true, sendMessages: true), + ]), + 2: StreamPermissions(readStream: true, manageTopics: true, sendMessages: true), + ]) + } + + @Test func versionInfo() throws { + var writer = ByteWriter() + try Requests.versionInfo(into: &writer) + #expect(writer.bytes == golden.bytes("version_info")) + } + + @Test func streams() throws { + #expect(try Requests.createStream(name: "test-stream") == golden.bytes("request.create_stream")) + #expect( + try Requests.createStream(name: "test-stream", options: try OptionsBlock.encode(sampleOptions)) == golden.bytes("request.create_stream.options")) + #expect(try encode(5) == golden.bytes("request.delete_stream")) + #expect(try encode("my-stream") == golden.bytes("request.get_stream")) + #expect(golden.bytes("request.get_streams").isEmpty) + #expect(try encode(1) == golden.bytes("request.purge_stream")) + #expect(try Requests.updateStream(streamID: "old-name", name: "new-name", options: []) == golden.bytes("request.update_stream")) + } + + @Test func topics() throws { + #expect( + try Requests.createTopic(streamID: 1, partitionsCount: 3, name: "orders", options: try OptionsBlock.encode(sampleOptions)) + == golden.bytes("request.create_topic")) + #expect(try Requests.streamAndTopic(1, 5) == golden.bytes("request.delete_topic")) + #expect(try Requests.streamAndTopic("my-stream", "my-topic") == golden.bytes("request.get_topic")) + #expect(try encode(42) == golden.bytes("request.get_topics")) + #expect(try Requests.streamAndTopic(1, 3) == golden.bytes("request.purge_topic")) + let options = try OptionsBlock.encode(["message_expiry": .explicit(.uint64(604_800_000_000))]) + #expect(try Requests.updateTopic(streamID: 1, topicID: 2, name: "updated-topic", options: options) == golden.bytes("request.update_topic")) + } + + @Test func partitionsAndSegments() throws { + #expect(try Requests.partitions(streamID: 1, topicID: 2, partitionsCount: 5) == golden.bytes("request.create_partitions")) + #expect(try Requests.partitions(streamID: "stream", topicID: "topic", partitionsCount: 2) == golden.bytes("request.delete_partitions")) + #expect(try Requests.deleteSegments(streamID: 1, topicID: 2, partitionID: 3, segmentsCount: 10) == golden.bytes("request.delete_segments")) + } + + @Test func consumerGroups() throws { + #expect(try Requests.createConsumerGroup(streamID: 1, topicID: 2, name: "grp") == golden.bytes("request.create_consumer_group")) + #expect(try Requests.consumerGroup(streamID: 1, topicID: 2, groupID: 3) == golden.bytes("request.delete_consumer_group")) + #expect(try Requests.consumerGroup(streamID: "stream-1", topicID: "topic-1", groupID: "group-1") == golden.bytes("request.get_consumer_group")) + #expect(try Requests.streamAndTopic(1, 2) == golden.bytes("request.get_consumer_groups")) + #expect(try Requests.consumerGroup(streamID: 1, topicID: 2, groupID: 3) == golden.bytes("request.join_consumer_group")) + #expect(try Requests.consumerGroup(streamID: 1, topicID: 2, groupID: "g") == golden.bytes("request.leave_consumer_group")) + #expect(try Requests.consumerGroup(streamID: "stream-1", topicID: "topic-1", groupID: "group-1") == golden.bytes("request.sync_consumer_group")) + } + + @Test func consumerOffsets() throws { + #expect( + try Requests.storeConsumerOffset(consumer: .consumer(1), streamID: 10, topicID: 20, partitionID: 5, offset: 12_345, ack: .quorum) + == golden.bytes("request.store_consumer_offset")) + #expect( + try Requests.storeConsumerOffset(consumer: .group(3), streamID: 1, topicID: 1, partitionID: nil, offset: UInt64.max, ack: .noAck) + == golden.bytes("request.store_consumer_offset.no_partition")) + #expect( + try Requests.getConsumerOffset(consumer: .consumer("my-consumer"), streamID: "stream-1", topicID: "topic-1", partitionID: 0) + == golden.bytes("request.get_consumer_offset")) + #expect( + try Requests.deleteConsumerOffset(consumer: .consumer(1), streamID: 10, topicID: 20, partitionID: 5, ack: .quorum) + == golden.bytes("request.delete_consumer_offset")) + } + + @Test func messages() throws { + #expect( + try Requests.pollMessages(consumer: .consumer(1), streamID: 10, topicID: 20, partitionID: 5, strategy: .offset(100), count: 50, autoCommit: true) + == golden.bytes("request.poll_messages")) + #expect( + try Requests.pollMessages(consumer: .group(3), streamID: 1, topicID: 1, partitionID: nil, strategy: .first, count: 10, autoCommit: false) + == golden.bytes("request.poll_messages.group")) + + let headers: UserHeaders = ["trace-id": "abc", "attempt": .uint32(3)] + let messages = [ + Batch.OutgoingMessage(id: 1, originTimestamp: 1_000, payload: Array("first".utf8), userHeaders: nil), + Batch.OutgoingMessage(id: 2, originTimestamp: 1_500, payload: Array("second".utf8), userHeaders: HeaderTLV.encode(headers)), + Batch.OutgoingMessage(id: UInt128Value(low: 3, high: 0xDEAD_BEEF_0000_0000), originTimestamp: 1_000, payload: [], userHeaders: nil), + ] + let encoded = try Batch.encodeSend(streamID: 1, topicID: 2, partitioning: .partition(0), messages: messages) + #expect(encoded == golden.bytes("request.send_messages")) + } + + @Test func users() throws { + #expect( + try Requests.createUser(username: "admin", password: "p@ssw0rd", status: .active, permissions: samplePermissions) + == golden.bytes("request.create_user")) + #expect( + try Requests.createUser(username: "user", password: "secret123", status: .inactive, permissions: nil) + == golden.bytes("request.create_user.no_permissions")) + #expect(try encode("old-user") == golden.bytes("request.delete_user")) + #expect(try encode(42) == golden.bytes("request.get_user")) + #expect(golden.bytes("request.get_users").isEmpty) + #expect(try Requests.updateUser(userID: 1, username: "new-name", status: .inactive, options: []) == golden.bytes("request.update_user")) + #expect(try Requests.updateUser(userID: 5, username: nil, status: nil, options: []) == golden.bytes("request.update_user.none")) + #expect(try Requests.updatePermissions(userID: 1, permissions: samplePermissions) == golden.bytes("request.update_permissions")) + #expect(try Requests.updatePermissions(userID: "admin", permissions: nil) == golden.bytes("request.update_permissions.none")) + #expect(try Requests.changePassword(userID: 1, currentPassword: "old-pass", newPassword: "new-pass-123") == golden.bytes("request.change_password")) + let testBindSecret = Array(repeating: UInt8(0x5a), count: Requests.bindSecretSize) + #expect(try Requests.loginRegister(username: "iggy", password: "iggy", bindSecret: testBindSecret) == golden.bytes("request.login_register")) + #expect(try Requests.loginRegisterWithPersonalAccessToken(token: "pat-abc123def456", bindSecret: testBindSecret) == golden.bytes("request.login_register_with_pat")) + #expect(golden.bytes("request.logout_user").isEmpty) + } + + @Test func personalAccessTokens() throws { + #expect( + try Requests.createPersonalAccessToken(name: "my-token", expiry: .after(.microseconds(3600))) + == golden.bytes("request.create_personal_access_token")) + #expect(try Requests.nameOnly("my-token") == golden.bytes("request.delete_personal_access_token")) + #expect(golden.bytes("request.get_personal_access_tokens").isEmpty) + } + + @Test func system() throws { + #expect(golden.bytes("request.ping").isEmpty) + #expect(golden.bytes("request.get_stats").isEmpty) + #expect(golden.bytes("request.get_me").isEmpty) + #expect(Requests.getClient(clientID: 42) == golden.bytes("request.get_client")) + #expect(golden.bytes("request.get_clients").isEmpty) + #expect(golden.bytes("request.get_cluster_metadata").isEmpty) + #expect(try Requests.getSnapshot(compression: .deflated, types: [.filesystemOverview, .serverLogs]) == golden.bytes("request.get_snapshot")) + #expect(Requests.describeOptions(scope: .topic) == golden.bytes("request.describe_options")) + let testBindSecret = Array(repeating: UInt8(0x5a), count: Requests.bindSecretSize) + let testClientID = UInt128Value(low: 1, high: 0) + #expect(try Requests.bindSession(clientID: testClientID, session: 100, metadataWatermark: 120, bindSecret: testBindSecret) == golden.bytes("request.bind_session")) + } + + @Test func protocolVersionMatchesTheCrate() { + #expect(ProtocolVersion.current.packed == golden.protocolVersion) + #expect(ProtocolVersion.current.description == ProtocolVersion(packed: golden.protocolVersion).description) + #expect(ProtocolVersion(major: 0, minor: 11, patch: 1).description == "0.11.1") + } +} + +@Suite("Consensus headers match the Rust golden vectors") +struct VSRHeaderVectorTests { + let golden = GoldenFixture.shared + let clientID = UInt128Value(low: 0xFEDC_BA98_7654_3210, high: 0x0123_4567_89AB_CDEF) + + func header(_ request: VSRFrame.EncodedRequest) -> [UInt8] { + Array(request.bytes[..