hubcio commented on code in PR #4306:
URL: https://github.com/apache/iggy/pull/4306#discussion_r4127146744
##########
.github/actions/swift/pre-merge/action.yml:
##########
@@ -73,3 +74,21 @@ runs:
swift --version
swift build -Xswiftc -warnings-as-errors
swift test --parallel
+
+ - name: Setup Rust with cache for the golden vectors
+ if: inputs.task == 'golden'
+ uses: ./.github/actions/utils/setup-rust-with-cache
Review Comment:
warning: in the last run the disk cleanup, apt install and cache restore
took about 6 of 7.5 minutes, yet the generator still compiled 169 crates cold
and needs no system packages. pass `free-disk-space`, `read-cache` and
`install-system-dependencies` as `"false"`.
##########
foreign/swift/Tools/golden-vectors/src/main.rs:
##########
@@ -0,0 +1,1468 @@
+// 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.
+
+//! Dumps byte-exact wire fixtures from the Rust protocol crates.
+//!
+//! The Swift SDK re-implements the binary protocol from scratch, so every
+//! encoder and decoder it ships is checked against bytes produced by the
+//! reference implementation rather than against itself. Run from
+//! `foreign/swift`:
+//!
+//! ```text
+//! cargo run --manifest-path Tools/golden-vectors/Cargo.toml --
Tests/IggyTests/Fixtures/golden.json
+//! ```
+
+use bytes::{BufMut, Bytes, BytesMut};
+use iggy_binary_protocol::batch::{
+ BATCH_HEADER_SIZE, BATCH_MESSAGE_HEADER_SIZE, BatchHeader,
calculate_batch_checksum,
+};
+use iggy_binary_protocol::codec::WireEncode;
+use iggy_binary_protocol::codes::*;
+use iggy_binary_protocol::consensus::{
+ Command, EvictionHeader, EvictionReason, HEADER_SIZE, Operation,
ReplyHeader, RequestHeader,
+};
+use iggy_binary_protocol::primitives::options::WireOptions;
+use iggy_binary_protocol::primitives::permissions::{
+ WireGlobalPermissions, WirePermissions, WireStreamPermissions,
WireTopicPermissions,
+};
+use iggy_binary_protocol::primitives::user_headers::encode_user_headers;
+use iggy_binary_protocol::requests::consumer_groups::*;
+use iggy_binary_protocol::requests::consumer_offsets::*;
+use iggy_binary_protocol::requests::messages::{
+ FlushUnsavedBufferRequest, PollMessagesRequest, RawMessage,
SendMessagesEncoder,
Review Comment:
critical: #4317 removed `FlushUnsavedBufferRequest` from master, so this
import no longer compiles and the golden lane goes red once the branch picks up
master. drop it and the `request.flush_unsaved_buffer` vector, then regenerate
`golden.json`.
##########
foreign/swift/Tests/IggyTests/HashTests.swift:
##########
@@ -0,0 +1,70 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+import Testing
+
+@testable import Iggy
+
+@Suite("XXH3-64")
+struct XXH3Tests {
+ @Test("matches the reference implementation for every golden length")
+ func matchesGoldenVectors() {
+ let fixture = GoldenFixture.shared
+ #expect(fixture.xxh3_64.count > 300)
+ for vector in fixture.xxh3_64 {
+ let input = GoldenFixture.pattern(vector.len)
+ let hash = XXH3.hash64(input)
+ #expect(String(hash, radix: 16).leftPadded(16) == vector.hash,
"length \(vector.len)")
+ }
+ }
+
+ @Test("hashes slices by their own bounds")
+ func hashesSlices() {
+ let input = GoldenFixture.pattern(600)
+ let slice = input[100..<400]
+ #expect(XXH3.hash64(slice) == XXH3.hash64(Array(slice)))
+ }
+
+ @Test("empty input has the well-known value")
+ func emptyInput() {
+ #expect(XXH3.hash64([]) == 0x2D06_8005_38D3_94C2)
+ }
+}
+
+@Suite("XXH32")
+struct XXH32Tests {
+ @Test("matches the reference implementation for every golden length")
+ func matchesGoldenVectors() {
+ let fixture = GoldenFixture.shared
+ #expect(fixture.xxh32.count > 300)
+ for vector in fixture.xxh32 {
+ let hash = XXH32.hash(GoldenFixture.pattern(vector.len))
+ #expect(String(hash, radix: 16).leftPadded(8) == vector.hash,
"length \(vector.len)")
+ }
+ }
+
+ @Test("empty input has the well-known value")
+ func emptyInput() {
+ #expect(XXH32.hash([]) == 0x02CC_5D05)
+ }
+}
+
+extension String {
Review Comment:
simplification: compare integers instead - `#expect(UInt64(vector.hash,
radix: 16) == hash)` at line 31 and the `UInt32` twin at line 56 - and this
module-wide `String` extension can go.
##########
foreign/swift/Sources/Iggy/Hash/XXH3.swift:
##########
@@ -0,0 +1,273 @@
+// 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.
+
+/// XXH3 64-bit hash with the default secret and a zero seed.
+///
+/// Every message frame and batch on the wire is checksummed with this hash,
+/// and the server verifies the values, so the implementation has to match the
+/// reference bit for bit. It is checked against vectors produced by the Rust
+/// `twox-hash` crate in the test suite.
+enum XXH3 {
+ private static let prime32_1: UInt64 = 0x9E37_79B1
+ private static let prime32_2: UInt64 = 0x85EB_CA77
+ private static let prime32_3: UInt64 = 0xC2B2_AE3D
+ private static let prime64_1: UInt64 = 0x9E37_79B1_85EB_CA87
+ private static let prime64_2: UInt64 = 0xC2B2_AE3D_27D4_EB4F
+ private static let prime64_3: UInt64 = 0x1656_67B1_9E37_79F9
+ private static let prime64_4: UInt64 = 0x85EB_CA77_C2B2_AE63
+ private static let prime64_5: UInt64 = 0x27D4_EB2F_1656_67C5
+ private static let primeMX1: UInt64 = 0x1656_6791_9E37_79F9
+ private static let primeMX2: UInt64 = 0x9FB2_1C65_1E98_DF25
+
+ private static let stripeLength = 64
+ private static let secretConsumeRate = 8
+ private static let accumulatorCount = 8
+ private static let secretLength = 192
+ private static let stripesPerBlock = (secretLength - stripeLength) /
secretConsumeRate
+ private static let blockLength = stripeLength * stripesPerBlock
+ private static let secretLastAccumulatorStart = 7
+ private static let secretMergeAccumulatorsStart = 11
+ private static let midsizeStartOffset = 3
+ private static let midsizeLastOffset = 17
+ private static let secretSizeMin = 136
+
+ private static let secret: [UInt8] = [
+ 0xb8, 0xfe, 0x6c, 0x39, 0x23, 0xa4, 0x4b, 0xbe, 0x7c, 0x01, 0x81,
0x2c, 0xf7, 0x21, 0xad, 0x1c,
+ 0xde, 0xd4, 0x6d, 0xe9, 0x83, 0x90, 0x97, 0xdb, 0x72, 0x40, 0xa4,
0xa4, 0xb7, 0xb3, 0x67, 0x1f,
+ 0xcb, 0x79, 0xe6, 0x4e, 0xcc, 0xc0, 0xe5, 0x78, 0x82, 0x5a, 0xd0,
0x7d, 0xcc, 0xff, 0x72, 0x21,
+ 0xb8, 0x08, 0x46, 0x74, 0xf7, 0x43, 0x24, 0x8e, 0xe0, 0x35, 0x90,
0xe6, 0x81, 0x3a, 0x26, 0x4c,
+ 0x3c, 0x28, 0x52, 0xbb, 0x91, 0xc3, 0x00, 0xcb, 0x88, 0xd0, 0x65,
0x8b, 0x1b, 0x53, 0x2e, 0xa3,
+ 0x71, 0x64, 0x48, 0x97, 0xa2, 0x0d, 0xf9, 0x4e, 0x38, 0x19, 0xef,
0x46, 0xa9, 0xde, 0xac, 0xd8,
+ 0xa8, 0xfa, 0x76, 0x3f, 0xe3, 0x9c, 0x34, 0x3f, 0xf9, 0xdc, 0xbb,
0xc7, 0xc7, 0x0b, 0x4f, 0x1d,
+ 0x8a, 0x51, 0xe0, 0x4b, 0xcd, 0xb4, 0x59, 0x31, 0xc8, 0x9f, 0x7e,
0xc9, 0xd9, 0x78, 0x73, 0x64,
+ 0xea, 0xc5, 0xac, 0x83, 0x34, 0xd3, 0xeb, 0xc3, 0xc5, 0x81, 0xa0,
0xff, 0xfa, 0x13, 0x63, 0xeb,
+ 0x17, 0x0d, 0xdd, 0x51, 0xb7, 0xf0, 0xda, 0x49, 0xd3, 0x16, 0x55,
0x26, 0x29, 0xd4, 0x68, 0x9e,
+ 0x2b, 0x16, 0xbe, 0x58, 0x7d, 0x47, 0xa1, 0xfc, 0x8f, 0xf8, 0xb8,
0xd1, 0x7a, 0xd0, 0x31, 0xce,
+ 0x45, 0xcb, 0x3a, 0x8f, 0x95, 0x16, 0x04, 0x28, 0xaf, 0xd7, 0xfb,
0xca, 0xbb, 0x4b, 0x40, 0x7e,
+ ]
+
+ static func hash64(_ bytes: [UInt8]) -> UInt64 {
+ bytes.withUnsafeBytes { hash64($0) }
+ }
+
+ static func hash64(_ bytes: ArraySlice<UInt8>) -> UInt64 {
+ bytes.withUnsafeBytes { hash64($0) }
+ }
+
+ static func hash64(_ input: UnsafeRawBufferPointer) -> UInt64 {
+ secret.withUnsafeBytes { secret in
+ let length = input.count
+ switch length {
+ case 0:
+ return avalanche64(secret.loadLittleEndianUInt64(at: 56) ^
secret.loadLittleEndianUInt64(at: 64))
+ case 1...3:
+ return hash1to3(input, secret)
+ case 4...8:
+ return hash4to8(input, secret)
+ case 9...16:
+ return hash9to16(input, secret)
+ case 17...128:
+ return hash17to128(input, secret)
+ case 129...240:
+ return hash129to240(input, secret)
+ default:
+ return hashLong(input, secret)
+ }
+ }
+ }
+
+ @inline(__always)
+ private static func rotl(_ value: UInt64, _ count: UInt64) -> UInt64 {
+ (value << count) | (value >> (64 - count))
+ }
+
+ @inline(__always)
+ private static func mul128Fold64(_ lhs: UInt64, _ rhs: UInt64) -> UInt64 {
+ let product = lhs.multipliedFullWidth(by: rhs)
+ return product.high ^ product.low
+ }
+
+ @inline(__always)
+ private static func avalanche64(_ input: UInt64) -> UInt64 {
+ var hash = input
+ hash ^= hash >> 33
+ hash &*= prime64_2
+ hash ^= hash >> 29
+ hash &*= prime64_3
+ hash ^= hash >> 32
+ return hash
+ }
+
+ @inline(__always)
+ private static func avalanche(_ input: UInt64) -> UInt64 {
+ var hash = input
+ hash ^= hash >> 37
+ hash &*= primeMX1
+ hash ^= hash >> 32
+ return hash
+ }
+
+ @inline(__always)
+ private static func rrmxmx(_ input: UInt64, _ length: Int) -> UInt64 {
+ var hash = input
+ hash ^= rotl(hash, 49) ^ rotl(hash, 24)
+ hash &*= primeMX2
+ hash ^= (hash >> 35) &+ UInt64(length)
+ hash &*= primeMX2
+ return hash ^ (hash >> 28)
+ }
+
+ private static func hash1to3(_ input: UnsafeRawBufferPointer, _ secret:
UnsafeRawBufferPointer) -> UInt64 {
+ let length = input.count
+ let byte1 = UInt32(input[0])
+ let byte2 = UInt32(input[length >> 1])
+ let byte3 = UInt32(input[length - 1])
+ let combined = (byte1 << 16) | (byte2 << 24) | byte3 | (UInt32(length)
<< 8)
+ let bitflip = UInt64(secret.loadLittleEndianUInt32(at: 0) ^
secret.loadLittleEndianUInt32(at: 4))
+ return avalanche64(UInt64(combined) ^ bitflip)
+ }
+
+ private static func hash4to8(_ input: UnsafeRawBufferPointer, _ secret:
UnsafeRawBufferPointer) -> UInt64 {
+ let length = input.count
+ let input1 = input.loadLittleEndianUInt32(at: 0)
+ let input2 = input.loadLittleEndianUInt32(at: length - 4)
+ let bitflip = secret.loadLittleEndianUInt64(at: 8) ^
secret.loadLittleEndianUInt64(at: 16)
+ let input64 = UInt64(input2) &+ (UInt64(input1) << 32)
+ return rrmxmx(input64 ^ bitflip, length)
+ }
+
+ private static func hash9to16(_ input: UnsafeRawBufferPointer, _ secret:
UnsafeRawBufferPointer) -> UInt64 {
+ let length = input.count
+ let bitflip1 = secret.loadLittleEndianUInt64(at: 24) ^
secret.loadLittleEndianUInt64(at: 32)
+ let bitflip2 = secret.loadLittleEndianUInt64(at: 40) ^
secret.loadLittleEndianUInt64(at: 48)
+ let inputLow = input.loadLittleEndianUInt64(at: 0) ^ bitflip1
+ let inputHigh = input.loadLittleEndianUInt64(at: length - 8) ^ bitflip2
+ let accumulator = UInt64(length) &+ inputLow.byteSwapped &+ inputHigh
&+ mul128Fold64(inputLow, inputHigh)
+ return avalanche(accumulator)
+ }
+
+ @inline(__always)
+ private static func mix16(_ input: UnsafeRawBufferPointer, _ inputOffset:
Int, _ secret: UnsafeRawBufferPointer, _ secretOffset: Int) -> UInt64 {
+ let inputLow = input.loadLittleEndianUInt64(at: inputOffset)
+ let inputHigh = input.loadLittleEndianUInt64(at: inputOffset + 8)
+ return mul128Fold64(inputLow ^ secret.loadLittleEndianUInt64(at:
secretOffset), inputHigh ^ secret.loadLittleEndianUInt64(at: secretOffset + 8))
+ }
+
+ private static func hash17to128(_ input: UnsafeRawBufferPointer, _ secret:
UnsafeRawBufferPointer) -> UInt64 {
+ let length = input.count
+ var accumulator = UInt64(length) &* prime64_1
+ if length > 32 {
+ if length > 64 {
+ if length > 96 {
+ accumulator &+= mix16(input, 48, secret, 96)
+ accumulator &+= mix16(input, length - 64, secret, 112)
+ }
+ accumulator &+= mix16(input, 32, secret, 64)
+ accumulator &+= mix16(input, length - 48, secret, 80)
+ }
+ accumulator &+= mix16(input, 16, secret, 32)
+ accumulator &+= mix16(input, length - 32, secret, 48)
+ }
+ accumulator &+= mix16(input, 0, secret, 0)
+ accumulator &+= mix16(input, length - 16, secret, 16)
+ return avalanche(accumulator)
+ }
+
+ private static func hash129to240(_ input: UnsafeRawBufferPointer, _
secret: UnsafeRawBufferPointer) -> UInt64 {
+ let length = input.count
+ var accumulator = UInt64(length) &* prime64_1
+ let rounds = length / 16
+ for round in 0..<8 {
+ accumulator &+= mix16(input, 16 * round, secret, 16 * round)
+ }
+ accumulator = avalanche(accumulator)
+ for round in 8..<rounds {
+ accumulator &+= mix16(input, 16 * round, secret, 16 * (round - 8)
+ midsizeStartOffset)
+ }
+ accumulator &+= mix16(input, length - 16, secret, secretSizeMin -
midsizeLastOffset)
+ return avalanche(accumulator)
+ }
+
+ private static let initialAccumulators: [UInt64] = [prime32_3, prime64_1,
prime64_2, prime64_3, prime64_4, prime32_2, prime64_5, prime32_1]
+
+ /// The accumulators live in a stack allocation: this path runs for every
Review Comment:
nit: the batch checksum hashes `44 + 8 * message_count` bytes, so batches
under 25 messages never reach this path - only inputs over 240 bytes do. reword
the comment.
##########
.github/actions/swift/pre-merge/action.yml:
##########
@@ -73,3 +74,21 @@ runs:
swift --version
swift build -Xswiftc -warnings-as-errors
swift test --parallel
+
+ - name: Setup Rust with cache for the golden vectors
+ if: inputs.task == 'golden'
+ uses: ./.github/actions/utils/setup-rust-with-cache
+
+ # The golden fixture is checked in, so this lane regenerates it from the
+ # Rust crates and fails when the two differ; otherwise the Swift tests
+ # stay green against a stale copy of the protocol.
+ - name: Check the golden vectors match the Rust crates
+ shell: bash
+ if: inputs.task == 'golden'
+ run: |
+ cargo run --manifest-path
foreign/swift/Tools/golden-vectors/Cargo.toml --
foreign/swift/Tests/IggyTests/Fixtures/golden.json
Review Comment:
nit: the generator is a separate workspace, so the repo's `cargo fmt --all`
and clippy runs never check it. add `cargo fmt --manifest-path ... -- --check`
and `cargo clippy --manifest-path ... -- -D warnings` here, like the
python-maturin action does.
##########
foreign/swift/Tools/golden-vectors/src/main.rs:
##########
@@ -0,0 +1,1468 @@
+// 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.
+
+//! Dumps byte-exact wire fixtures from the Rust protocol crates.
+//!
+//! The Swift SDK re-implements the binary protocol from scratch, so every
+//! encoder and decoder it ships is checked against bytes produced by the
+//! reference implementation rather than against itself. Run from
+//! `foreign/swift`:
+//!
+//! ```text
+//! cargo run --manifest-path Tools/golden-vectors/Cargo.toml --
Tests/IggyTests/Fixtures/golden.json
+//! ```
+
+use bytes::{BufMut, Bytes, BytesMut};
+use iggy_binary_protocol::batch::{
+ BATCH_HEADER_SIZE, BATCH_MESSAGE_HEADER_SIZE, BatchHeader,
calculate_batch_checksum,
+};
+use iggy_binary_protocol::codec::WireEncode;
+use iggy_binary_protocol::codes::*;
+use iggy_binary_protocol::consensus::{
+ Command, EvictionHeader, EvictionReason, HEADER_SIZE, Operation,
ReplyHeader, RequestHeader,
+};
+use iggy_binary_protocol::primitives::options::WireOptions;
+use iggy_binary_protocol::primitives::permissions::{
+ WireGlobalPermissions, WirePermissions, WireStreamPermissions,
WireTopicPermissions,
+};
+use iggy_binary_protocol::primitives::user_headers::encode_user_headers;
+use iggy_binary_protocol::requests::consumer_groups::*;
+use iggy_binary_protocol::requests::consumer_offsets::*;
+use iggy_binary_protocol::requests::messages::{
+ FlushUnsavedBufferRequest, PollMessagesRequest, RawMessage,
SendMessagesEncoder,
+};
+use iggy_binary_protocol::requests::partitions::{
+ CreatePartitionsRequest, DeletePartitionsRequest,
+};
+use iggy_binary_protocol::requests::personal_access_tokens::*;
+use iggy_binary_protocol::requests::segments::DeleteSegmentsRequest;
+use iggy_binary_protocol::requests::streams::*;
+use iggy_binary_protocol::requests::system::*;
+use iggy_binary_protocol::requests::topics::{
+ CreateTopicRequest, DeleteTopicRequest, GetTopicRequest, GetTopicsRequest,
PurgeTopicRequest,
+ UpdateTopicRequest,
+};
+use iggy_binary_protocol::requests::users::*;
+use iggy_binary_protocol::responses::clients::{
+ ClientDetailsResponse, ClientResponse, ConsumerGroupInfoResponse,
GetClientsResponse,
+};
+use iggy_binary_protocol::responses::consumer_groups::{
+ ConsumerGroupDetailsResponse, ConsumerGroupMemberResponse,
ConsumerGroupResponse,
+ GetConsumerGroupsResponse, SyncConsumerGroupResponse,
+};
+use iggy_binary_protocol::responses::consumer_offsets::ConsumerOffsetResponse;
+use iggy_binary_protocol::responses::messages::{
+ PollMessagesResponseHeader, SendMessagesConfirmationResponse,
SendMessagesResponse,
+};
+use iggy_binary_protocol::responses::personal_access_tokens::{
+ GetPersonalAccessTokensResponse, PersonalAccessTokenResponse,
RawPersonalAccessTokenResponse,
+};
+use iggy_binary_protocol::responses::streams::{
+ GetStreamResponse, GetStreamsResponse, StreamResponse, TopicHeader,
+};
+use iggy_binary_protocol::responses::system::get_stats::{CacheMetricEntry,
StatsResponse};
+use iggy_binary_protocol::responses::system::{
+ ClusterMetadataResponse, ClusterNodeResponse, DescribeOptionsResponse,
OptionDescriptor,
+};
+use iggy_binary_protocol::responses::topics::{
+ GetTopicResponse, GetTopicsResponse, PartitionResponse,
+};
+use iggy_binary_protocol::responses::users::{
+ GetUsersResponse, IdentityResponse, LoginRegisterResponse,
UserDetailsResponse, UserResponse,
+};
+use iggy_binary_protocol::{
+ AckLevel, ClientVersionInfo, IGGY_PROTOCOL_VERSION,
IGGY_PROTOCOL_VERSION_MIN, WireConsumer,
+ WireIdentifier, WireName, WirePartitioning, WirePollingStrategy,
+};
+use iggy_common::{
+ CompressionAlgorithm, Durability, IggyByteSize, IggyDuration, IggyError,
IggyExpiry,
+ MaxTopicSize, TopicCreateOptions, TopicUpdateOptions,
+};
+use secrecy::SecretString;
+use serde::Serialize;
+use std::collections::BTreeMap;
+use twox_hash::{XxHash3_64, XxHash32};
+
+/// Hash and error tables are maps keyed by length and code, so each entry is
+/// one line of the pretty-printed fixture and the file stays reviewable.
+#[derive(Serialize)]
+struct Golden {
+ protocol_version: u32,
+ protocol_version_min: u32,
+ xxh3_64: BTreeMap<usize, String>,
+ xxh32: BTreeMap<usize, String>,
+ errors: BTreeMap<u32, String>,
+ vectors: BTreeMap<String, String>,
+}
+
+fn pattern(len: usize) -> Vec<u8> {
+ (0..len)
+ .map(|i| (i as u8).wrapping_mul(31).wrapping_add(7))
+ .collect()
+}
+
+fn hex(bytes: &[u8]) -> String {
+ bytes.iter().map(|b| format!("{b:02x}")).collect()
+}
+
+fn numeric(id: u32) -> WireIdentifier {
+ WireIdentifier::numeric(id)
+}
+
+fn named(name: &str) -> WireIdentifier {
+ WireIdentifier::named(name).expect("valid name")
+}
+
+fn name(value: &str) -> WireName {
+ WireName::new(value).expect("valid name")
+}
+
+fn options(entries: &[(u8, &[u8], u8, &[u8])]) -> WireOptions {
+ let mut buf = BytesMut::new();
+ encode_user_headers(entries, &mut buf);
+ WireOptions::from_bytes(buf.freeze()).expect("valid options")
+}
+
+fn sample_options() -> WireOptions {
+ options(&[
+ (2, b"durability", 2, b"persisted"),
+ (2, b"segment_size", 12, &1_073_741_824u64.to_le_bytes()),
+ ])
+}
+
+fn version_info() -> ClientVersionInfo {
+ ClientVersionInfo {
+ protocol_version: IGGY_PROTOCOL_VERSION,
+ sdk_name: name("swift-sdk"),
+ sdk_version: name("0.1.0"),
+ }
+}
+
+fn sample_permissions() -> WirePermissions {
+ WirePermissions {
+ global: WireGlobalPermissions {
+ manage_servers: true,
+ read_servers: true,
+ manage_users: false,
+ read_users: true,
+ manage_streams: false,
+ read_streams: true,
+ manage_topics: false,
+ read_topics: true,
+ poll_messages: true,
+ send_messages: false,
+ },
+ streams: vec![
+ WireStreamPermissions {
+ stream_id: 1,
+ manage_stream: true,
+ read_stream: true,
+ manage_topics: false,
+ read_topics: true,
+ poll_messages: true,
+ send_messages: false,
+ topics: vec![
+ WireTopicPermissions {
+ topic_id: 10,
+ manage_topic: true,
+ read_topic: true,
+ poll_messages: false,
+ send_messages: true,
+ },
+ WireTopicPermissions {
+ topic_id: 20,
+ manage_topic: false,
+ read_topic: true,
+ poll_messages: true,
+ send_messages: true,
+ },
+ ],
+ },
+ WireStreamPermissions {
+ stream_id: 2,
+ manage_stream: false,
+ read_stream: true,
+ manage_topics: true,
+ read_topics: false,
+ poll_messages: false,
+ send_messages: true,
+ topics: vec![],
+ },
+ ],
+ }
+}
+
+fn frame(
+ id: u128,
+ offset_delta: u32,
+ timestamp_delta: u32,
+ payload: &[u8],
+ user_headers: &[u8],
+) -> Vec<u8> {
+ let mut bytes = vec![0u8; BATCH_MESSAGE_HEADER_SIZE];
+ bytes[8..24].copy_from_slice(&id.to_le_bytes());
+ bytes[24..28].copy_from_slice(&offset_delta.to_le_bytes());
+ bytes[28..32].copy_from_slice(×tamp_delta.to_le_bytes());
+ bytes[32..36].copy_from_slice(&(user_headers.len() as u32).to_le_bytes());
+ bytes[36..40].copy_from_slice(&(payload.len() as u32).to_le_bytes());
+ bytes.extend_from_slice(payload);
+ bytes.extend_from_slice(user_headers);
+ let checksum = XxHash3_64::oneshot(&bytes[8..]);
+ bytes[0..8].copy_from_slice(&checksum.to_le_bytes());
+ bytes
+}
+
+fn batch_record(
+ partition_id: u64,
+ base_offset: u64,
+ base_timestamp: u64,
+ origin_timestamp: u64,
+ frames: &[Vec<u8>],
+) -> Vec<u8> {
+ let blob: Vec<u8> = frames.concat();
+ let mut header = BatchHeader::new(
+ partition_id,
+ origin_timestamp,
+ (BATCH_HEADER_SIZE + blob.len()) as u64,
+ frames.len() as u32,
+ );
+ header.base_offset = base_offset;
+ header.base_timestamp = base_timestamp;
+ header.batch_checksum = calculate_batch_checksum(&header, &blob);
+ let mut bytes = vec![0u8; BATCH_HEADER_SIZE];
+ header.encode_into(&mut bytes);
+ bytes.extend_from_slice(&blob);
+ bytes
+}
+
+fn topic_header(id: u32, name_value: &str) -> TopicHeader {
+ TopicHeader {
+ id,
+ created_at: 1_710_000_000_000_000,
+ partitions_count: 3,
+ message_expiry: 604_800_000_000,
+ compression_algorithm: 1,
+ max_topic_size: 1_073_741_824,
+ size_bytes: 4096,
+ messages_count: 100,
+ name: name(name_value),
+ options: sample_options(),
+ derived_options: options(&[(2, b"max_topic_size", 12,
&u64::MAX.to_le_bytes())]),
+ }
+}
+
+fn stream_response(id: u32, topics_count: u32) -> StreamResponse {
+ StreamResponse {
+ id,
+ created_at: 1_710_000_000_000_000,
+ topics_count,
+ size_bytes: 2048,
+ messages_count: 200,
+ name: name("my-stream"),
+ options: WireOptions::empty(),
+ }
+}
+
+fn add<T: WireEncode>(vectors: &mut BTreeMap<String, String>, key: &str,
value: &T) {
+ vectors.insert(key.to_owned(), hex(&value.to_bytes()));
+}
+
+fn request_header(
Review Comment:
warning: `request_header()` re-implements the header rules of
`encode_request_header` in `core/sdk/src/vsr.rs`, so the golden lane stays
green when the SDK changes them. move those rules into `iggy_binary_protocol`
and call them from both places.
##########
.github/config/components.yml:
##########
@@ -346,9 +346,10 @@ components:
- "bdd/swift/**"
- "examples/swift/**"
# The SDK compiles against the VSR wire contract, so a change to the
- # protocol crate must rerun it.
+ # protocol crates must rerun it and regenerate the golden vectors.
- "core/binary_protocol/**"
- tasks: ["lint", "test", "build", "build-macos"]
+ - "core/common/**"
Review Comment:
simplification: `depends_on: rust-server` already reruns this lane for
`core/common` and `core/binary_protocol` changes, because dependents resolve
transitively. this line and the one above are redundant - drop both and the
comment.
##########
foreign/swift/Tests/IggyTests/ErrorCodeTests.swift:
##########
@@ -39,6 +39,17 @@ struct ErrorCodeTests {
#expect(Set(IggyErrorCode.allCases.map(\.name)).count ==
IggyErrorCode.allCases.count)
}
+ /// The full table, against the dump the Rust generator takes from the
+ /// server's error enum.
+ @Test func errorCodesMatchTheServerTable() {
Review Comment:
simplification: this test proves the whole table, so the hard-coded count,
min and max pins in `tableMatchesTheServerCodes` only add a second manual edit
per new server code. drop them.
##########
foreign/swift/Sources/Iggy/Hash/XXH3.swift:
##########
@@ -0,0 +1,273 @@
+// 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.
+
+/// XXH3 64-bit hash with the default secret and a zero seed.
+///
+/// Every message frame and batch on the wire is checksummed with this hash,
+/// and the server verifies the values, so the implementation has to match the
+/// reference bit for bit. It is checked against vectors produced by the Rust
+/// `twox-hash` crate in the test suite.
+enum XXH3 {
+ private static let prime32_1: UInt64 = 0x9E37_79B1
+ private static let prime32_2: UInt64 = 0x85EB_CA77
+ private static let prime32_3: UInt64 = 0xC2B2_AE3D
+ private static let prime64_1: UInt64 = 0x9E37_79B1_85EB_CA87
+ private static let prime64_2: UInt64 = 0xC2B2_AE3D_27D4_EB4F
+ private static let prime64_3: UInt64 = 0x1656_67B1_9E37_79F9
+ private static let prime64_4: UInt64 = 0x85EB_CA77_C2B2_AE63
+ private static let prime64_5: UInt64 = 0x27D4_EB2F_1656_67C5
+ private static let primeMX1: UInt64 = 0x1656_6791_9E37_79F9
+ private static let primeMX2: UInt64 = 0x9FB2_1C65_1E98_DF25
+
+ private static let stripeLength = 64
+ private static let secretConsumeRate = 8
+ private static let accumulatorCount = 8
+ private static let secretLength = 192
+ private static let stripesPerBlock = (secretLength - stripeLength) /
secretConsumeRate
+ private static let blockLength = stripeLength * stripesPerBlock
+ private static let secretLastAccumulatorStart = 7
+ private static let secretMergeAccumulatorsStart = 11
+ private static let midsizeStartOffset = 3
+ private static let midsizeLastOffset = 17
+ private static let secretSizeMin = 136
+
+ private static let secret: [UInt8] = [
+ 0xb8, 0xfe, 0x6c, 0x39, 0x23, 0xa4, 0x4b, 0xbe, 0x7c, 0x01, 0x81,
0x2c, 0xf7, 0x21, 0xad, 0x1c,
+ 0xde, 0xd4, 0x6d, 0xe9, 0x83, 0x90, 0x97, 0xdb, 0x72, 0x40, 0xa4,
0xa4, 0xb7, 0xb3, 0x67, 0x1f,
+ 0xcb, 0x79, 0xe6, 0x4e, 0xcc, 0xc0, 0xe5, 0x78, 0x82, 0x5a, 0xd0,
0x7d, 0xcc, 0xff, 0x72, 0x21,
+ 0xb8, 0x08, 0x46, 0x74, 0xf7, 0x43, 0x24, 0x8e, 0xe0, 0x35, 0x90,
0xe6, 0x81, 0x3a, 0x26, 0x4c,
+ 0x3c, 0x28, 0x52, 0xbb, 0x91, 0xc3, 0x00, 0xcb, 0x88, 0xd0, 0x65,
0x8b, 0x1b, 0x53, 0x2e, 0xa3,
+ 0x71, 0x64, 0x48, 0x97, 0xa2, 0x0d, 0xf9, 0x4e, 0x38, 0x19, 0xef,
0x46, 0xa9, 0xde, 0xac, 0xd8,
+ 0xa8, 0xfa, 0x76, 0x3f, 0xe3, 0x9c, 0x34, 0x3f, 0xf9, 0xdc, 0xbb,
0xc7, 0xc7, 0x0b, 0x4f, 0x1d,
+ 0x8a, 0x51, 0xe0, 0x4b, 0xcd, 0xb4, 0x59, 0x31, 0xc8, 0x9f, 0x7e,
0xc9, 0xd9, 0x78, 0x73, 0x64,
+ 0xea, 0xc5, 0xac, 0x83, 0x34, 0xd3, 0xeb, 0xc3, 0xc5, 0x81, 0xa0,
0xff, 0xfa, 0x13, 0x63, 0xeb,
+ 0x17, 0x0d, 0xdd, 0x51, 0xb7, 0xf0, 0xda, 0x49, 0xd3, 0x16, 0x55,
0x26, 0x29, 0xd4, 0x68, 0x9e,
+ 0x2b, 0x16, 0xbe, 0x58, 0x7d, 0x47, 0xa1, 0xfc, 0x8f, 0xf8, 0xb8,
0xd1, 0x7a, 0xd0, 0x31, 0xce,
+ 0x45, 0xcb, 0x3a, 0x8f, 0x95, 0x16, 0x04, 0x28, 0xaf, 0xd7, 0xfb,
0xca, 0xbb, 0x4b, 0x40, 0x7e,
+ ]
+
+ static func hash64(_ bytes: [UInt8]) -> UInt64 {
+ bytes.withUnsafeBytes { hash64($0) }
+ }
+
+ static func hash64(_ bytes: ArraySlice<UInt8>) -> UInt64 {
+ bytes.withUnsafeBytes { hash64($0) }
+ }
+
+ static func hash64(_ input: UnsafeRawBufferPointer) -> UInt64 {
+ secret.withUnsafeBytes { secret in
+ let length = input.count
+ switch length {
+ case 0:
+ return avalanche64(secret.loadLittleEndianUInt64(at: 56) ^
secret.loadLittleEndianUInt64(at: 64))
+ case 1...3:
+ return hash1to3(input, secret)
+ case 4...8:
+ return hash4to8(input, secret)
+ case 9...16:
+ return hash9to16(input, secret)
+ case 17...128:
+ return hash17to128(input, secret)
+ case 129...240:
+ return hash129to240(input, secret)
+ default:
+ return hashLong(input, secret)
+ }
+ }
+ }
+
+ @inline(__always)
+ private static func rotl(_ value: UInt64, _ count: UInt64) -> UInt64 {
+ (value << count) | (value >> (64 - count))
+ }
+
+ @inline(__always)
+ private static func mul128Fold64(_ lhs: UInt64, _ rhs: UInt64) -> UInt64 {
+ let product = lhs.multipliedFullWidth(by: rhs)
+ return product.high ^ product.low
+ }
+
+ @inline(__always)
+ private static func avalanche64(_ input: UInt64) -> UInt64 {
+ var hash = input
+ hash ^= hash >> 33
+ hash &*= prime64_2
+ hash ^= hash >> 29
+ hash &*= prime64_3
+ hash ^= hash >> 32
+ return hash
+ }
+
+ @inline(__always)
+ private static func avalanche(_ input: UInt64) -> UInt64 {
+ var hash = input
+ hash ^= hash >> 37
+ hash &*= primeMX1
+ hash ^= hash >> 32
+ return hash
+ }
+
+ @inline(__always)
+ private static func rrmxmx(_ input: UInt64, _ length: Int) -> UInt64 {
+ var hash = input
+ hash ^= rotl(hash, 49) ^ rotl(hash, 24)
+ hash &*= primeMX2
+ hash ^= (hash >> 35) &+ UInt64(length)
+ hash &*= primeMX2
+ return hash ^ (hash >> 28)
+ }
+
+ private static func hash1to3(_ input: UnsafeRawBufferPointer, _ secret:
UnsafeRawBufferPointer) -> UInt64 {
+ let length = input.count
+ let byte1 = UInt32(input[0])
+ let byte2 = UInt32(input[length >> 1])
+ let byte3 = UInt32(input[length - 1])
+ let combined = (byte1 << 16) | (byte2 << 24) | byte3 | (UInt32(length)
<< 8)
+ let bitflip = UInt64(secret.loadLittleEndianUInt32(at: 0) ^
secret.loadLittleEndianUInt32(at: 4))
+ return avalanche64(UInt64(combined) ^ bitflip)
+ }
+
+ private static func hash4to8(_ input: UnsafeRawBufferPointer, _ secret:
UnsafeRawBufferPointer) -> UInt64 {
+ let length = input.count
+ let input1 = input.loadLittleEndianUInt32(at: 0)
+ let input2 = input.loadLittleEndianUInt32(at: length - 4)
+ let bitflip = secret.loadLittleEndianUInt64(at: 8) ^
secret.loadLittleEndianUInt64(at: 16)
+ let input64 = UInt64(input2) &+ (UInt64(input1) << 32)
+ return rrmxmx(input64 ^ bitflip, length)
+ }
+
+ private static func hash9to16(_ input: UnsafeRawBufferPointer, _ secret:
UnsafeRawBufferPointer) -> UInt64 {
+ let length = input.count
+ let bitflip1 = secret.loadLittleEndianUInt64(at: 24) ^
secret.loadLittleEndianUInt64(at: 32)
+ let bitflip2 = secret.loadLittleEndianUInt64(at: 40) ^
secret.loadLittleEndianUInt64(at: 48)
+ let inputLow = input.loadLittleEndianUInt64(at: 0) ^ bitflip1
+ let inputHigh = input.loadLittleEndianUInt64(at: length - 8) ^ bitflip2
+ let accumulator = UInt64(length) &+ inputLow.byteSwapped &+ inputHigh
&+ mul128Fold64(inputLow, inputHigh)
+ return avalanche(accumulator)
+ }
+
+ @inline(__always)
+ private static func mix16(_ input: UnsafeRawBufferPointer, _ inputOffset:
Int, _ secret: UnsafeRawBufferPointer, _ secretOffset: Int) -> UInt64 {
+ let inputLow = input.loadLittleEndianUInt64(at: inputOffset)
+ let inputHigh = input.loadLittleEndianUInt64(at: inputOffset + 8)
+ return mul128Fold64(inputLow ^ secret.loadLittleEndianUInt64(at:
secretOffset), inputHigh ^ secret.loadLittleEndianUInt64(at: secretOffset + 8))
+ }
+
+ private static func hash17to128(_ input: UnsafeRawBufferPointer, _ secret:
UnsafeRawBufferPointer) -> UInt64 {
+ let length = input.count
+ var accumulator = UInt64(length) &* prime64_1
+ if length > 32 {
+ if length > 64 {
+ if length > 96 {
+ accumulator &+= mix16(input, 48, secret, 96)
+ accumulator &+= mix16(input, length - 64, secret, 112)
+ }
+ accumulator &+= mix16(input, 32, secret, 64)
+ accumulator &+= mix16(input, length - 48, secret, 80)
+ }
+ accumulator &+= mix16(input, 16, secret, 32)
+ accumulator &+= mix16(input, length - 32, secret, 48)
+ }
+ accumulator &+= mix16(input, 0, secret, 0)
+ accumulator &+= mix16(input, length - 16, secret, 16)
+ return avalanche(accumulator)
+ }
+
+ private static func hash129to240(_ input: UnsafeRawBufferPointer, _
secret: UnsafeRawBufferPointer) -> UInt64 {
+ let length = input.count
+ var accumulator = UInt64(length) &* prime64_1
+ let rounds = length / 16
+ for round in 0..<8 {
+ accumulator &+= mix16(input, 16 * round, secret, 16 * round)
+ }
+ accumulator = avalanche(accumulator)
+ for round in 8..<rounds {
+ accumulator &+= mix16(input, 16 * round, secret, 16 * (round - 8)
+ midsizeStartOffset)
+ }
+ accumulator &+= mix16(input, length - 16, secret, secretSizeMin -
midsizeLastOffset)
+ return avalanche(accumulator)
+ }
+
+ private static let initialAccumulators: [UInt64] = [prime32_3, prime64_1,
prime64_2, prime64_3, prime64_4, prime32_2, prime64_5, prime32_1]
+
+ /// The accumulators live in a stack allocation: this path runs for every
+ /// batch and every frame over 240 bytes, so it must not hit the heap.
+ private static func hashLong(_ input: UnsafeRawBufferPointer, _ secret:
UnsafeRawBufferPointer) -> UInt64 {
+ let length = input.count
+ let blocks = (length - 1) / blockLength
+ return withUnsafeTemporaryAllocation(of: UInt64.self, capacity:
accumulatorCount) { accumulators in
+ for index in 0..<accumulatorCount {
+ accumulators.initializeElement(at: index, to:
initialAccumulators[index])
+ }
+ for block in 0..<blocks {
+ accumulate(accumulators, input, block * blockLength, secret,
0, stripes: stripesPerBlock)
+ scramble(accumulators, secret, secretLength - stripeLength)
+ }
+ let stripes = ((length - 1) - blockLength * blocks) / stripeLength
+ accumulate(accumulators, input, blocks * blockLength, secret, 0,
stripes: stripes)
+ accumulate512(accumulators, input, length - stripeLength, secret,
secretLength - stripeLength - secretLastAccumulatorStart)
+ return mergeAccumulators(accumulators, secret,
secretMergeAccumulatorsStart, start: UInt64(length) &* prime64_1)
+ }
+ }
+
+ @inline(__always)
+ private static func accumulate512(
+ _ accumulators: UnsafeMutableBufferPointer<UInt64>, _ input:
UnsafeRawBufferPointer, _ inputOffset: Int, _ secret: UnsafeRawBufferPointer,
+ _ secretOffset: Int
+ ) {
+ for lane in 0..<accumulatorCount {
+ let dataValue = input.loadLittleEndianUInt64(at: inputOffset + 8 *
lane)
+ let dataKey = dataValue ^ secret.loadLittleEndianUInt64(at:
secretOffset + 8 * lane)
+ accumulators[lane ^ 1] &+= dataValue
+ accumulators[lane] &+= UInt64(UInt32(truncatingIfNeeded: dataKey))
&* (dataKey >> 32)
+ }
+ }
+
+ @inline(__always)
+ private static func accumulate(
+ _ accumulators: UnsafeMutableBufferPointer<UInt64>, _ input:
UnsafeRawBufferPointer, _ inputOffset: Int, _ secret: UnsafeRawBufferPointer,
+ _ secretOffset: Int, stripes: Int
Review Comment:
simplification: both callers pass `0` as `secretOffset`, so the parameter is
dead. drop it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
