github-actions[bot] commented on code in PR #4355: URL: https://github.com/apache/iggy/pull/4355#discussion_r4161198449
########## foreign/swift/Sources/Iggy/Protocol/Batch.swift: ########## @@ -0,0 +1,254 @@ +// 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<UInt8>) 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..<start + 40]) + guard batchLength >= 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..<start + Batch.headerSize].allSatisfy({ $0 == 0 }) else { + throw WireError.validation("batch header reserved bytes must be zero") + } + return Header( + partitionID: UInt64(littleEndianBytes: bytes[start..<start + 8]), + baseOffset: UInt64(littleEndianBytes: bytes[start + 8..<start + 16]), + baseTimestamp: UInt64(littleEndianBytes: bytes[start + 16..<start + 24]), + originTimestamp: UInt64(littleEndianBytes: bytes[start + 24..<start + 32]), + batchLength: batchLength, + batchChecksum: UInt64(littleEndianBytes: bytes[start + Batch.checksumOffset..<start + Batch.checksumOffset + 8]), + messageCount: UInt32(littleEndianBytes: bytes[start + Batch.messageCountOffset..<start + Batch.messageCountOffset + 4])) + } + + func encode() -> [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<UInt8>) 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..<start + 48]) == 0 else { + throw WireError.validation("message frame reserved bytes must be zero") + } + return MessageHeader( + checksum: UInt64(littleEndianBytes: bytes[start..<start + 8]), + id: UInt128Value(littleEndianBytes: bytes[start + 8..<start + 24]), + offsetDelta: UInt32(littleEndianBytes: bytes[start + 24..<start + 28]), + timestampDelta: UInt32(littleEndianBytes: bytes[start + 28..<start + 32]), + userHeadersLength: UInt32(littleEndianBytes: bytes[start + 32..<start + 36]), + payloadLength: UInt32(littleEndianBytes: bytes[start + 36..<start + 40])) + } + } + + /// Batch checksum: XXH3-64 over the six header meta fields followed by + /// every frame's stored 8-byte checksum field in message order. Bodies are + /// bound transitively through the frame checksums. + static func checksum(header: Header, frameChecksums: [UInt8]) -> 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<UInt8>, 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...]) + let batchEnd = position + Int(header.batchLength) Review Comment: critical: `Batch.decodeMessages` adds the untrusted `batchLength` to `position` before it bounds the sum, and Swift traps on the `Int` overflow. `Header.decode` admits lengths up to `UInt64(Int.max)`, so a reply that declares a length near `Int.max` aborts the client process. Compare the length against the remaining bytes in `UInt64` first, then convert and add. ########## foreign/swift/Sources/Iggy/Protocol/Requests.swift: ########## @@ -0,0 +1,323 @@ +// 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 + + 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) Review Comment: nit: `validateUsername` and `validatePassword` throw `IggyError(.invalidUsername)` and `.invalidPassword` with no context, unlike the token check at line 267 that states its range. Add the allowed range and the observed byte count to both contexts. ########## foreign/swift/Sources/Iggy/Protocol/VSRFrame.swift: ########## @@ -0,0 +1,286 @@ +// 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 timestamp = 160 Review Comment: simplification: Three offset constants are declared and never referenced: `RequestOffset.timestamp`, `RequestOffset.userID`, and `EvictionOffset.client`. Delete them, or pin the header layout in a test. Also at lines 41 and 56. ########## foreign/swift/Sources/Iggy/Protocol/Batch.swift: ########## @@ -0,0 +1,254 @@ +// 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<UInt8>) 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..<start + 40]) + guard batchLength >= 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..<start + Batch.headerSize].allSatisfy({ $0 == 0 }) else { + throw WireError.validation("batch header reserved bytes must be zero") + } + return Header( + partitionID: UInt64(littleEndianBytes: bytes[start..<start + 8]), + baseOffset: UInt64(littleEndianBytes: bytes[start + 8..<start + 16]), + baseTimestamp: UInt64(littleEndianBytes: bytes[start + 16..<start + 24]), + originTimestamp: UInt64(littleEndianBytes: bytes[start + 24..<start + 32]), + batchLength: batchLength, + batchChecksum: UInt64(littleEndianBytes: bytes[start + Batch.checksumOffset..<start + Batch.checksumOffset + 8]), + messageCount: UInt32(littleEndianBytes: bytes[start + Batch.messageCountOffset..<start + Batch.messageCountOffset + 4])) + } + + func encode() -> [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<UInt8>) 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..<start + 48]) == 0 else { + throw WireError.validation("message frame reserved bytes must be zero") + } + return MessageHeader( + checksum: UInt64(littleEndianBytes: bytes[start..<start + 8]), + id: UInt128Value(littleEndianBytes: bytes[start + 8..<start + 24]), + offsetDelta: UInt32(littleEndianBytes: bytes[start + 24..<start + 28]), + timestampDelta: UInt32(littleEndianBytes: bytes[start + 28..<start + 32]), + userHeadersLength: UInt32(littleEndianBytes: bytes[start + 32..<start + 36]), + payloadLength: UInt32(littleEndianBytes: bytes[start + 36..<start + 40])) + } + } + + /// Batch checksum: XXH3-64 over the six header meta fields followed by + /// every frame's stored 8-byte checksum field in message order. Bodies are + /// bound transitively through the frame checksums. + static func checksum(header: Header, frameChecksums: [UInt8]) -> 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<UInt8>, 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...]) + let batchEnd = position + Int(header.batchLength) + guard batchEnd <= bytes.endIndex else { + throw WireError.truncated(offset: position - bytes.startIndex, need: Int(header.batchLength), have: bytes.endIndex - position) + } + var cursor = position + headerSize + var frameChecksums = [UInt8]() + var counted: UInt32 = 0 + while cursor < batchEnd { + let frame = try MessageHeader.decode(bytes[cursor..<batchEnd]) + let payloadStart = cursor + messageHeaderSize + // Each length is checked against what is left before it is + // added, so a hostile frame cannot overflow a 32-bit `Int`. + guard frame.payloadLength <= UInt32(clamping: batchEnd - payloadStart) else { + throw WireError.truncated(offset: payloadStart - bytes.startIndex, need: Int(frame.payloadLength), have: batchEnd - payloadStart) + } + let payloadEnd = payloadStart + Int(frame.payloadLength) + guard frame.userHeadersLength <= UInt32(clamping: batchEnd - payloadEnd) else { + throw WireError.truncated(offset: payloadEnd - bytes.startIndex, need: Int(frame.userHeadersLength), have: batchEnd - payloadEnd) + } + let headersEnd = payloadEnd + Int(frame.userHeadersLength) + if verifyChecksums { + let computed = XXH3.hash64(bytes[(cursor + 8)..<headersEnd]) + guard computed == frame.checksum else { + throw WireError.invalidMessageChecksum( + stored: frame.checksum, computed: computed, offset: header.baseOffset &+ UInt64(frame.offsetDelta)) + } + frameChecksums.append(contentsOf: bytes[cursor..<cursor + 8]) + } + let payload = Array(bytes[payloadStart..<payloadEnd]) + let userHeaders = frame.userHeadersLength > 0 ? Array(bytes[payloadEnd..<headersEnd]) : nil + messages.append( + IggyMessage( + id: frame.id, + offset: header.baseOffset &+ UInt64(frame.offsetDelta), + timestamp: IggyTimestamp(microseconds: header.baseTimestamp), + originTimestamp: IggyTimestamp(microseconds: header.originTimestamp &+ UInt64(frame.timestampDelta)), + checksum: frame.checksum, + payload: payload, + rawUserHeaders: userHeaders)) + counted += 1 + cursor = headersEnd + } + if verifyChecksums { + guard counted == header.messageCount else { Review Comment: warning: The `counted == header.messageCount` guard sits inside `if verifyChecksums`, so the poll path, which passes `verifyChecksums: false`, accepts a batch whose frames do not tile `messageCount`. The Rust `LayoutOnly` poll path rejects the same bytes. Move the guard out of the branch so both paths enforce it. ########## foreign/swift/Sources/Iggy/Protocol/VSRFrame.swift: ########## @@ -0,0 +1,286 @@ +// 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 timestamp = 160 + static let request = 168 + static let operation = 176 + static let session = 184 + static let userID = 192 + 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 client = 128 + 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<UInt8>) -> UInt32? { + guard header.count >= ReplyOffset.size + 4 else { + return nil + } + let start = header.startIndex + ReplyOffset.size + return UInt32(littleEndianBytes: header[start..<start + 4]) + } + + static func peekCommand(_ header: ArraySlice<UInt8>) -> Command? { + Command(rawValue: header[header.startIndex + ReplyOffset.command]) + } + + static func readReplyRequestID(_ header: ArraySlice<UInt8>) -> UInt64 { + let start = header.startIndex + ReplyOffset.request + return UInt64(littleEndianBytes: header[start..<start + 8]) + } + + static func readReplyStatus(_ header: ArraySlice<UInt8>) -> UInt32 { + let start = header.startIndex + ReplyOffset.status + return UInt32(littleEndianBytes: header[start..<start + 4]) + } + + static func readReplyOperation(_ header: ArraySlice<UInt8>) -> ConsensusOperation? { + 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<UInt8>) -> Eviction { + let start = header.startIndex + let rawReason = header[start + EvictionOffset.reason] + let version = UInt32(littleEndianBytes: header[start + EvictionOffset.serverProtocolVersion..<start + EvictionOffset.serverProtocolVersion + 4]) + let versionMin = UInt32( + littleEndianBytes: header[start + EvictionOffset.serverProtocolVersionMin..<start + EvictionOffset.serverProtocolVersionMin + 4]) + return Eviction(reason: EvictionReason(rawValue: rawReason), rawReason: rawReason, serverProtocolVersion: version, serverProtocolVersionMin: versionMin) + } + + /// Turns one complete consensus frame into its reply body. The order is + /// fixed: frame discriminant, declared size, pre-commit denial status + /// before any body read, then the committed result section. + static func decodeReply(header: ArraySlice<UInt8>, body: ArraySlice<UInt8>) throws -> ArraySlice<UInt8> { + precondition(header.count == headerSize) Review Comment: nit: `decodeReply` traps through `precondition` when the header is not `headerSize` bytes, and `peekCommand` and the reply accessors index the buffer unchecked while `readSize` returns nil. Guard the length and throw `IggyError(.invalidCommand)`, or return optionals as `readSize` does. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
