hubcio commented on code in PR #4355: URL: https://github.com/apache/iggy/pull/4355#discussion_r4255818597
########## 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) Review Comment: critical: the header leaves partition context zero, so sends and offset changes fail for partitions with a nonzero incarnation. accept `PartitionContext` and write its fields at offsets 200, 208, and 216. ########## 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<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...]) + // 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..<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) Review Comment: critical: oversized payload or header lengths trap on 32-bit targets instead of throwing a decoding error. use `Int(clamping:)` for both diagnostic lengths at lines 215 and 219. ########## 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) Review Comment: warning: zeroing the whole frame before copying the payload writes over the payload area unnecessarily. reserve `totalSize`, zero only the header, then append the payload. ########## 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() Review Comment: warning: `beginRegister` changes `bindSecret` after the login payload captures it, so repeat registration leaves later session binds with the wrong secret. preserve the session identity and secret across registration retries. ########## 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<UInt8> { Review Comment: simplification: `OptionsBlock.readPrefixed` only forwards to `ByteReader.readPrefixedBytes`. call `ByteReader.readPrefixedBytes` directly and remove the wrapper. ########## 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<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...]) + // 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..<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 + } + // The layout rule holds on every path, as in the Rust `LayoutOnly` + // decode; only the checksum work is optional. + guard counted == header.messageCount else { + throw WireError.validation("batch frames do not tile message_count exactly") + } + if verifyChecksums { + let computed = checksum(header: header, frameChecksums: frameChecksums) + guard computed == header.batchChecksum else { + throw WireError.invalidBatchChecksum(stored: header.batchChecksum, computed: computed) + } + } + position = batchEnd + } + return messages Review Comment: warning: a reply can declare one message, contain no batches, and still decode successfully. reject replies when the decoded message count differs from `expectedCount`. -- 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]
