You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by "RustToMetal (via GitHub)" <gi...@apache.org> on 2026/09/29 22:49:50 UTC
[PR] feat(swift): add the VSR protocol layer with request and reply codecs [iggy]
RustToMetal opened a new pull request, #4355:
URL: https://github.com/apache/iggy/pull/4355
The protocol layer: the consensus frame every request and reply travels in, the session state that hands out request ids, one request encoder per command code, one reply decoder per reply shape, and the message batch format that send bodies and poll replies share. Nothing here opens a socket; the TCP and TLS client that drives this layer is the next PR.
Fourth of nine stacked PRs for the Swift SDK; builds on #4190, #4306, and #4335.
## Which issue does this PR address?
Relates to #4001
## Rationale
The client, producer, and consumer are thin over this layer, and the wire rules are the part most likely to diverge from the server. Landing them on their own lets every header, encoder, and decoder be reviewed next to the golden vector the Rust crates produced for it, before any networking is involved.
## What changed?
`VSRFrame` ports `core/binary_protocol/src/consensus/header.rs`: the 256-byte header offsets, the command and eviction discriminants, `encodeRequest` (which stamps the request checksum for the operations the server's client table deduplicates and leaves it zero for partition and non-replicated operations, like the Rust SDK), and `decodeReply`, which applies the fixed order of frame discriminant, declared size, pre-commit denial status, then the committed result section. Eviction reasons map to the same errors as `eviction_reason_to_error` in `core/common`.
`ConsensusSession` ports `core/sdk/src/session.rs`: an ephemeral client id, the session fence, and monotonic request ids for replicated operations. `ConsensusOperation` and `CommandCode` mirror the Rust enums, with the metadata, partition, and result-framed bands.
`Requests` has one encoder per command, with the username, password, and name bounds applied before encoding so a length prefix can never desync. `Responses` has one decoder per reply shape, mapping codec failures to `invalidCommand` like the Rust client. `Batch` ports `batch.rs`: the 256-byte batch header, the 48-byte message frame, and the XXH3 checksums over frames and batch.
Every length read from the wire is bounded as a `UInt32` before it becomes an `Int` (frame size, result-section count, prefixed blocks, message payload and header lengths), so a hostile frame cannot trap a 32-bit platform. The byte codec gains the helpers this layer needs: `readPrefixedBytes`, `skip`, `rest`, `writeZeros`, `overwrite`, and the checksum error cases with their `IggyError` mapping.
Tests: `WireVectorTests` compares every request encoder and consensus header against the Rust bytes in `golden.json`, `ResponseVectorTests` does the same for every reply decoder, and `ProtocolTests` covers the session rules, result-section handling, denial ordering, duplicate sends, and eviction mapping.
## Local Execution
- Passed: `swift build -Xswiftc -warnings-as-errors`, `swift test` (105 tests), `swift format lint --strict`, hawkeye license check
- Pre-commit: `prek run --from-ref upstream/master --to-ref HEAD` passed
## AI Usage
1. **Which tools?** Claude Code (Claude).
2. **Scope of usage?** The port and the tests were drafted with it from `core/binary_protocol`, `core/sdk/src/vsr.rs`, and `core/sdk/src/session.rs` as the reference, reviewed and directed by me.
3. **How did you verify the generated code works correctly?** Every header, request, and reply is checked byte for byte against vectors the Rust crates produced, and the full chain was run end to end against a real `iggy-server` (35 end-to-end tests) plus the 9 BDD scenarios.
4. **Can you explain every line of the code if asked?** Yes.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(swift): add the VSR protocol layer with request and reply codecs [iggy]
Posted by "RustToMetal (via GitHub)" <gi...@apache.org>.
RustToMetal commented on PR #4355:
URL: https://github.com/apache/iggy/pull/4355#issuecomment-5900571730
/ready
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(swift): add the VSR protocol layer with request and reply codecs [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio commented on PR #4355:
URL: https://github.com/apache/iggy/pull/4355#issuecomment-6061764181
purge is getting removed from the server and every sdk, and i'll open the removal pr today. so you can drop `purgeStream` / `purgeTopic` from `CommandCode.swift` and `ConsensusOperation.swift`, plus the two purge golden vector tests. if this merges first that's fine too, they'll go away in the removal pr.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(swift): add the VSR protocol layer with request and reply codecs [iggy]
Posted by "RustToMetal (via GitHub)" <gi...@apache.org>.
RustToMetal commented on PR #4355:
URL: https://github.com/apache/iggy/pull/4355#issuecomment-6140132065
/ready
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(swift): add the VSR protocol layer with request and reply codecs [iggy]
Posted by "RustToMetal (via GitHub)" <gi...@apache.org>.
RustToMetal commented on PR #4355:
URL: https://github.com/apache/iggy/pull/4355#issuecomment-6140667679
/ready
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(swift): add the VSR protocol layer with request and reply codecs [iggy]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(swift): add the VSR protocol layer with request and reply codecs [iggy]
Posted by "RustToMetal (via GitHub)" <gi...@apache.org>.
RustToMetal commented on code in PR #4355:
URL: https://github.com/apache/iggy/pull/4355#discussion_r4163350777
##########
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:
Done, the length is compared against the remaining bytes as a UInt64 before it is added, with a test that declares a length near Int.max.
##########
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:
Done, the tiling guard runs on both paths now, with a test for the layout-only path.
##########
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:
Done, decodeReply throws invalidCommand for a short header and every raw accessor returns nil instead of indexing past the buffer.
##########
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:
Done, both contexts state the allowed range and the observed byte count.
##########
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:
Done, all three are gone.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(swift): add the VSR protocol layer with request and reply codecs [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio merged PR #4355:
URL: https://github.com/apache/iggy/pull/4355
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(swift): add the VSR protocol layer with request and reply codecs [iggy]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #4355:
URL: https://github.com/apache/iggy/pull/4355#issuecomment-5900561158
<!-- iggy-pr-triage-welcome -->
Thanks for the PR. It is labeled `S-waiting-on-review` and queued for review.
Slash commands (own line, regular comment) move it around the queue:
- `/ready` - back to `S-waiting-on-review` after addressing feedback
- `/author` - flip to `S-waiting-on-author` while you finish changes
- `/request-review @user-or-team` - request a reviewer
- `/pin` - exempt the PR from the stale bot, `/unpin` to undo
See [CONTRIBUTING.md](https://github.com/apache/iggy/blob/master/CONTRIBUTING.md#pr-triage-commands) for details.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(swift): add the VSR protocol layer with request and reply codecs [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio commented on PR #4355:
URL: https://github.com/apache/iggy/pull/4355#issuecomment-5941920274
/skill team-review-slim
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(swift): add the VSR protocol layer with request and reply codecs [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio commented on PR #4355:
URL: https://github.com/apache/iggy/pull/4355#issuecomment-6113186926
@RustToMetal could you please fix CI? it now fails after rebase
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(swift): add the VSR protocol layer with request and reply codecs [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(swift): add the VSR protocol layer with request and reply codecs [iggy]
Posted by "RustToMetal (via GitHub)" <gi...@apache.org>.
RustToMetal commented on code in PR #4355:
URL: https://github.com/apache/iggy/pull/4355#discussion_r4255911047
##########
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:
Done, encodeRequest takes the partition context and writes it at 200, 208, and 216, compared against a send header the Rust crate stamps (new vsr.request.send_messages.fenced vector). The code that captures the context from a reply and passes it in arrives with the TCP client.
##########
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:
Done, both diagnostic lengths use Int(clamping:), with a test that sets each frame length to UInt32.max.
##########
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:
Done, the session keeps its client id, bind secret, and request counter across a Register retry and a reconnect replay, and bind accepts the same fence again while a different one throws sessionMismatch, as session.rs does. Only an explicit reset starts a new identity.
##########
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:
Done, the decoder throws when the decoded count differs from expectedCount, with tests for one declared message and no batches, and for counts below and above what the batches hold.
##########
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:
Done, the frame reserves totalSize, zeroes only the header, and appends the payload.
##########
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:
Done, the wrapper is gone and the callers use ByteReader.readPrefixedBytes directly.
--
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: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org