You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by gk...@apache.org on 2025/10/10 10:04:38 UTC
(iggy) branch io_uring_tpc updated: feat(io_uring): implement Slab (#2252)
This is an automated email from the ASF dual-hosted git repository.
gkoszyk pushed a commit to branch io_uring_tpc
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/io_uring_tpc by this push:
new 7c2bf8a2c feat(io_uring): implement Slab<User> (#2252)
7c2bf8a2c is described below
commit 7c2bf8a2cc8ed0a58b36f8f581b7d9aef454063a
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Fri Oct 10 12:04:35 2025 +0200
feat(io_uring): implement Slab<User> (#2252)
---
Cargo.lock | 410 ++++++------
core/binary_protocol/src/utils/mapper.rs | 2 +-
core/common/src/commands/users/defaults.rs | 2 +-
core/common/src/lib.rs | 2 +-
core/common/src/types/configuration/mod.rs | 2 +-
.../tests/cli/client/test_client_get_command.rs | 2 +-
core/integration/tests/cli/common/mod.rs | 2 +-
.../tests/cli/user/test_user_create_command.rs | 6 +-
.../tests/cli/user/test_user_delete_command.rs | 4 +-
.../tests/cli/user/test_user_get_command.rs | 8 +-
.../cli/user/test_user_permissions_command.rs | 8 +-
.../tests/server/concurrent_addition.rs | 122 ++++
core/integration/tests/server/mod.rs | 1 +
.../tests/server/scenarios/concurrent_scenario.rs | 703 +++++++++++++++++++++
core/integration/tests/server/scenarios/mod.rs | 1 +
.../scenarios/stream_size_validation_scenario.rs | 2 +-
.../tests/server/scenarios/user_scenario.rs | 12 +-
core/integration/tests/state/file.rs | 15 +-
core/integration/tests/state/system.rs | 2 +-
core/sdk/src/prelude.rs | 4 +-
core/server/src/binary/mapper.rs | 2 +-
core/server/src/bootstrap.rs | 16 +-
core/server/src/http/consumer_groups.rs | 13 +-
core/server/src/http/streams.rs | 17 +-
core/server/src/shard/builder.rs | 19 +-
core/server/src/shard/mod.rs | 30 +-
.../src/shard/system/personal_access_tokens.rs | 81 ++-
core/server/src/shard/system/segments.rs | 2 +-
core/server/src/shard/system/streams.rs | 33 +-
core/server/src/shard/system/topics.rs | 19 +-
core/server/src/shard/system/users.rs | 205 ++----
.../periodic/personal_access_token_cleaner.rs | 4 +-
core/server/src/slab/mod.rs | 1 +
core/server/src/slab/users.rs | 140 ++++
core/server/src/streaming/session.rs | 8 +-
core/server/src/streaming/users/user.rs | 2 +-
examples/rust/src/multi-tenant/consumer/main.rs | 4 +-
examples/rust/src/multi-tenant/producer/main.rs | 2 +-
38 files changed, 1398 insertions(+), 510 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index c0fa1fec5..7c5d12e8a 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -59,9 +59,9 @@ dependencies = [
[[package]]
name = "actix-http"
-version = "3.11.1"
+version = "3.11.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "44cceded2fb55f3c4b67068fa64962e2ca59614edc5b03167de9ff82ae803da0"
+checksum = "7926860314cbe2fb5d1f13731e387ab43bd32bca224e82e6e2db85de0a3dba49"
dependencies = [
"actix-codec",
"actix-rt",
@@ -339,9 +339,9 @@ dependencies = [
[[package]]
name = "anstream"
-version = "0.6.20"
+version = "0.6.21"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3ae563653d1938f79b1ab1b5e668c87c76a9930414574a6583a7b7e11a8e6192"
+checksum = "43d5b281e737544384e969a5ccad3f1cdd24b48086a0fc1b2a5262a26b8f4f4a"
dependencies = [
"anstyle",
"anstyle-parse",
@@ -354,9 +354,9 @@ dependencies = [
[[package]]
name = "anstyle"
-version = "1.0.11"
+version = "1.0.13"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "862ed96ca487e809f1c8e5a8447f6ee2cf102f846893800b20cebdf541fc6bbd"
+checksum = "5192cca8006f1fd4f7237516f40fa183bb07f8fbdfedaa0036de5ea9b0b45e78"
[[package]]
name = "anstyle-parse"
@@ -667,9 +667,9 @@ dependencies = [
[[package]]
name = "aws-lc-sys"
-version = "0.32.1"
+version = "0.32.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1ba2e2516bdf37af57fc6ff047855f54abad0066e5c4fdaaeb76dabb2e05bcf5"
+checksum = "a2b715a6010afb9e457ca2b7c9d2b9c344baa8baed7b38dc476034c171b32575"
dependencies = [
"bindgen",
"cc",
@@ -708,11 +708,11 @@ dependencies = [
[[package]]
name = "axum"
-version = "0.8.5"
+version = "0.8.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "98e529aee37b5c8206bb4bf4c44797127566d72f76952c970bd3d1e85de8f4e2"
+checksum = "8a18ed336352031311f4e0b4dd2ff392d4fbb370777c9d18d7fc9d7359f73871"
dependencies = [
- "axum-core 0.5.4",
+ "axum-core 0.5.5",
"axum-macros",
"bytes",
"form_urlencoded",
@@ -762,9 +762,9 @@ dependencies = [
[[package]]
name = "axum-core"
-version = "0.5.4"
+version = "0.5.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0ac7a6beb1182c7e30253ee75c3e918080bfb83f5a3023bcdf7209d85fd147e6"
+checksum = "59446ce19cd142f8833f856eb31f3eb097812d1479ab224f54d72428ca21ea22"
dependencies = [
"bytes",
"futures-core",
@@ -824,7 +824,7 @@ dependencies = [
"miniz_oxide",
"object",
"rustc-demangle",
- "windows-link 0.2.0",
+ "windows-link 0.2.1",
]
[[package]]
@@ -905,7 +905,7 @@ dependencies = [
"rand 0.9.2",
"serde",
"serde_json",
- "sysinfo 0.37.1",
+ "sysinfo 0.37.2",
"tracing",
"uuid",
]
@@ -1033,9 +1033,9 @@ dependencies = [
[[package]]
name = "bon"
-version = "3.7.2"
+version = "3.8.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c2529c31017402be841eb45892278a6c21a000c0a17643af326c73a73f83f0fb"
+checksum = "ebeb9aaf9329dff6ceb65c689ca3db33dbf15f324909c60e4e5eef5701ce31b1"
dependencies = [
"bon-macros",
"rustversion",
@@ -1043,9 +1043,9 @@ dependencies = [
[[package]]
name = "bon-macros"
-version = "3.7.2"
+version = "3.8.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d82020dadcb845a345591863adb65d74fa8dc5c18a0b6d408470e13b7adc7005"
+checksum = "77e9d642a7e3a318e37c2c9427b5a6a48aa1ad55dcd986f3034ab2239045a645"
dependencies = [
"darling 0.21.3",
"ident_case",
@@ -1164,9 +1164,9 @@ checksum = "175812e0be2bccb6abe50bb8d566126198344f707e304f45c648fd8f2cc0365e"
[[package]]
name = "bytemuck"
-version = "1.23.2"
+version = "1.24.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3995eaeebcdf32f91f980d360f78732ddc061097ab4e39991ae7a6ace9194677"
+checksum = "1fbdf580320f38b612e485521afda1ee26d10cc9884efaaa750d383e13e3c5f4"
[[package]]
name = "byteorder"
@@ -1200,9 +1200,9 @@ dependencies = [
[[package]]
name = "camino"
-version = "1.2.0"
+version = "1.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e1de8bc0aa9e9385ceb3bf0c152e3a9b9544f6c4a912c8ae504e80c1f0368603"
+checksum = "276a59bf2b2c967788139340c9f0c5b12d7fd6630315c15c217e559de85d2609"
dependencies = [
"serde_core",
]
@@ -1232,9 +1232,9 @@ dependencies = [
[[package]]
name = "cc"
-version = "1.2.39"
+version = "1.2.40"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e1354349954c6fc9cb0deab020f27f783cf0b604e8bb754dc4658ecf0d29c35f"
+checksum = "e1d05d92f4b1fd76aad469d46cdd858ca761576082cd37df81416691e50199fb"
dependencies = [
"find-msvc-tools",
"jobserver",
@@ -1308,7 +1308,7 @@ dependencies = [
"num-traits",
"serde",
"wasm-bindgen",
- "windows-link 0.2.0",
+ "windows-link 0.2.1",
]
[[package]]
@@ -1412,7 +1412,7 @@ version = "3.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "fde0e0ec90c9dfb3b4b1a0891a7dcd0e2bffde2f7efed5fe7c9bb00e5bfb915e"
dependencies = [
- "windows-sys 0.48.0",
+ "windows-sys 0.59.0",
]
[[package]]
@@ -1433,7 +1433,7 @@ checksum = "b03b7db8e0b4b2fdad6c551e634134e99ec000e5c8c3b6856c65e8bbaded7a3b"
dependencies = [
"crossterm",
"unicode-segmentation",
- "unicode-width 0.2.1",
+ "unicode-width 0.2.2",
]
[[package]]
@@ -1660,9 +1660,9 @@ dependencies = [
[[package]]
name = "config"
-version = "0.15.17"
+version = "0.15.18"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "680d3ac2fe066c43300ec831c978871e50113a708d58ab13d231bd92deca5adb"
+checksum = "180e549344080374f9b32ed41bf3b6b57885ff6a289367b3dbc10eea8acc1918"
dependencies = [
"async-trait",
"convert_case 0.6.0",
@@ -1673,7 +1673,7 @@ dependencies = [
"serde-untagged",
"serde_core",
"serde_json",
- "toml 0.9.7",
+ "toml 0.9.8",
"winnow 0.7.13",
"yaml-rust2",
]
@@ -1687,7 +1687,7 @@ dependencies = [
"encode_unicode",
"libc",
"once_cell",
- "unicode-width 0.2.1",
+ "unicode-width 0.2.2",
"windows-sys 0.59.0",
]
@@ -1942,7 +1942,7 @@ dependencies = [
"bitflags 2.9.4",
"crossterm_winapi",
"document-features",
- "parking_lot 0.12.4",
+ "parking_lot 0.12.5",
"rustix",
"winapi",
]
@@ -2006,7 +2006,7 @@ checksum = "881c5d0a13b2f1498e2306e82cbada78390e152d4b1378fb28a84f4dcd0dc4f3"
dependencies = [
"dispatch",
"nix",
- "windows-sys 0.61.1",
+ "windows-sys 0.61.2",
]
[[package]]
@@ -2096,8 +2096,8 @@ name = "cyper-axum"
version = "0.4.0"
source = "git+https://github.com/krishvishal/cyper.git?rev=cd75e266df6ab0a9b9474eb7dda1735650d17db6#cd75e266df6ab0a9b9474eb7dda1735650d17db6"
dependencies = [
- "axum 0.8.5",
- "axum-core 0.5.4",
+ "axum 0.8.6",
+ "axum-core 0.5.5",
"compio",
"compio-log 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
"cyper-core",
@@ -2208,7 +2208,7 @@ dependencies = [
"hashbrown 0.14.5",
"lock_api",
"once_cell",
- "parking_lot_core 0.9.11",
+ "parking_lot_core 0.9.12",
]
[[package]]
@@ -2241,9 +2241,9 @@ dependencies = [
[[package]]
name = "deflate64"
-version = "0.1.9"
+version = "0.1.10"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "da692b8d1080ea3045efaab14434d40468c3d8657e42abddfffca87b428f4c1b"
+checksum = "26bf8fc351c5ed29b5c2f0cbbac1b209b74f60ecd62e675a998df72c49af5204"
[[package]]
name = "der"
@@ -2399,7 +2399,7 @@ dependencies = [
"libc",
"option-ext",
"redox_users",
- "windows-sys 0.61.1",
+ "windows-sys 0.61.2",
]
[[package]]
@@ -2621,7 +2621,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb"
dependencies = [
"libc",
- "windows-sys 0.61.1",
+ "windows-sys 0.61.2",
]
[[package]]
@@ -2765,9 +2765,9 @@ dependencies = [
[[package]]
name = "find-msvc-tools"
-version = "0.1.2"
+version = "0.1.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1ced73b1dacfc750a6db6c0a0c3a3853c8b41997e2e2c563dc90804ae6867959"
+checksum = "0399f9d26e5191ce32c498bebd31e7a3ceabc2745f0ac54af3f335126c3f24b3"
[[package]]
name = "flatbuffers"
@@ -2781,9 +2781,9 @@ dependencies = [
[[package]]
name = "flate2"
-version = "1.1.2"
+version = "1.1.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4a3d7db9596fecd151c5f638c0ee5d5bd487b6e0ea232e5dc96d5250f6f94b1d"
+checksum = "dc5a4e564e38c699f2880d3fda590bedc2e69f3f84cd48b457bd892ce61d0aa9"
dependencies = [
"crc32fast",
"libz-rs-sys",
@@ -2855,9 +2855,9 @@ checksum = "28dd6caf6059519a65843af8fe2a3ae298b14b80179855aeb4adc2c1934ee619"
[[package]]
name = "fs-err"
-version = "3.1.2"
+version = "3.1.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "44f150ffc8782f35521cec2b23727707cb4045706ba3c854e86bef66b3a8cdbd"
+checksum = "6ad492b2cf1d89d568a43508ab24f98501fe03f2f31c01e1d0fe7366a71745d2"
dependencies = [
"autocfg",
"tokio",
@@ -2934,7 +2934,7 @@ checksum = "1d930c203dd0b6ff06e0201a4a2fe9149b43c684fd4420555b26d21b1a02956f"
dependencies = [
"futures-core",
"lock_api",
- "parking_lot 0.12.4",
+ "parking_lot 0.12.5",
]
[[package]]
@@ -3539,7 +3539,7 @@ dependencies = [
"getrandom 0.3.3",
"hashbrown 0.15.5",
"nonzero_ext",
- "parking_lot 0.12.4",
+ "parking_lot 0.12.5",
"portable-atomic",
"quanta",
"rand 0.9.2",
@@ -3653,6 +3653,12 @@ dependencies = [
"foldhash",
]
+[[package]]
+name = "hashbrown"
+version = "0.16.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5419bdc4f6a9207fbeba6d11b604d481addf78ecd10c11ad51e76c2f6482748d"
+
[[package]]
name = "hashlink"
version = "0.10.0"
@@ -3865,7 +3871,7 @@ dependencies = [
"tokio",
"tokio-rustls",
"tower-service",
- "webpki-roots 1.0.2",
+ "webpki-roots 1.0.3",
]
[[package]]
@@ -3899,7 +3905,7 @@ dependencies = [
"libc",
"percent-encoding",
"pin-project-lite",
- "socket2 0.5.10",
+ "socket2 0.6.0",
"tokio",
"tower-service",
"tracing",
@@ -3917,7 +3923,7 @@ dependencies = [
"js-sys",
"log",
"wasm-bindgen",
- "windows-core 0.62.1",
+ "windows-core 0.62.2",
]
[[package]]
@@ -4071,7 +4077,7 @@ dependencies = [
"tracing",
"trait-variant",
"tungstenite",
- "webpki-roots 1.0.2",
+ "webpki-roots 1.0.3",
]
[[package]]
@@ -4095,9 +4101,9 @@ dependencies = [
"rand 0.9.2",
"rayon",
"serde",
- "sysinfo 0.37.1",
+ "sysinfo 0.37.2",
"tokio",
- "toml 0.9.7",
+ "toml 0.9.8",
"tracing",
"tracing-appender",
"tracing-subscriber",
@@ -4155,7 +4161,7 @@ dependencies = [
name = "iggy-connectors"
version = "0.1.0"
dependencies = [
- "axum 0.8.5",
+ "axum 0.8.6",
"axum-server",
"config",
"dashmap",
@@ -4175,7 +4181,7 @@ dependencies = [
"strum",
"thiserror 2.0.17",
"tokio",
- "toml 0.9.7",
+ "toml 0.9.8",
"tower-http",
"tracing",
"tracing-subscriber",
@@ -4185,7 +4191,7 @@ dependencies = [
name = "iggy-mcp"
version = "0.1.0"
dependencies = [
- "axum 0.8.5",
+ "axum 0.8.6",
"axum-server",
"config",
"dotenvy",
@@ -4219,7 +4225,7 @@ dependencies = [
"serde",
"serde_json",
"tokio",
- "toml 0.9.7",
+ "toml 0.9.8",
"tracing",
]
@@ -4411,7 +4417,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f8a9aa791c7b5a71b636b7a68207fdebf171ddfc593d9c8506ec4cbc527b6a84"
dependencies = [
"implicit-clone-derive",
- "indexmap 1.9.3",
+ "indexmap 2.11.4",
]
[[package]]
@@ -4442,7 +4448,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4b0f83760fb341a774ed326568e19f5a863af4a952def8c39f9ab92fd95b88e5"
dependencies = [
"equivalent",
- "hashbrown 0.15.5",
+ "hashbrown 0.16.0",
"serde",
"serde_core",
]
@@ -4825,9 +4831,9 @@ checksum = "2c4a545a15244c7d945065b5d392b2d2d7f21526fba56ce51467b06ed445e8f7"
[[package]]
name = "libc"
-version = "0.2.176"
+version = "0.2.177"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "58f929b4d672ea937a23a1ab494143d968337a5f47e56d0815df1e0890ddf174"
+checksum = "2874a2af47a2325c2001a6e6fad9b16a53b802102b528163885171cf92b15976"
[[package]]
name = "libdbus-sys"
@@ -4858,7 +4864,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "07033963ba89ebaf1584d767badaa2e8fcec21aedea6b8c0346d487d49c28667"
dependencies = [
"cfg-if",
- "windows-targets 0.48.5",
+ "windows-targets 0.53.5",
]
[[package]]
@@ -4905,7 +4911,7 @@ checksum = "416f7e718bdb06000964960ffa43b4335ad4012ae8b99060261aa4a8088d5ccb"
dependencies = [
"bitflags 2.9.4",
"libc",
- "redox_syscall 0.5.17",
+ "redox_syscall 0.5.18",
]
[[package]]
@@ -4992,11 +4998,10 @@ checksum = "4d873d7c67ce09b42110d801813efbc9364414e356be9935700d368351657487"
[[package]]
name = "lock_api"
-version = "0.4.13"
+version = "0.4.14"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "96936507f153605bddfcda068dd804796c84324ed2510809e5b2a624c81da765"
+checksum = "224399e74b87b5f3557511d98dff8b14089b3dadafcab6bb93eab67d3aace965"
dependencies = [
- "autocfg",
"scopeguard",
]
@@ -5169,6 +5174,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1fa76a2c86f704bdb222d66965fb3d63269ce38518b83cb0575fca855ebb6316"
dependencies = [
"adler2",
+ "simd-adler32",
]
[[package]]
@@ -5222,7 +5228,7 @@ dependencies = [
"equivalent",
"event-listener",
"futures-util",
- "parking_lot 0.12.4",
+ "parking_lot 0.12.5",
"portable-atomic",
"rustc_version",
"smallvec",
@@ -5475,18 +5481,18 @@ dependencies = [
[[package]]
name = "objc2-core-foundation"
-version = "0.3.1"
+version = "0.3.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1c10c2894a6fed806ade6027bcd50662746363a9589d3ec9d9bef30a4e4bc166"
+checksum = "2a180dd8642fa45cdb7dd721cd4c11b1cadd4929ce112ebd8b9f5803cc79d536"
dependencies = [
"bitflags 2.9.4",
]
[[package]]
name = "objc2-io-kit"
-version = "0.3.1"
+version = "0.3.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "71c1c64d6120e51cd86033f67176b1cb66780c2efe34dec55176f77befd93c0a"
+checksum = "33fafba39597d6dc1fb709123dfa8289d39406734be322956a69f0931c73bb15"
dependencies = [
"libc",
"objc2-core-foundation",
@@ -5593,9 +5599,9 @@ checksum = "d05e27ee213611ffe7d6348b942e8f942b37114c00cc03cec254295a4a17852e"
[[package]]
name = "openssl-src"
-version = "300.5.2+3.5.2"
+version = "300.5.3+3.5.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d270b79e2926f5150189d475bc7e9d2c69f9c4697b185fa917d5a32b792d21b4"
+checksum = "dc6bad8cd0233b63971e232cc9c5e83039375b8586d2312f31fda85db8f888c2"
dependencies = [
"cc",
]
@@ -5761,12 +5767,12 @@ dependencies = [
[[package]]
name = "parking_lot"
-version = "0.12.4"
+version = "0.12.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "70d58bf43669b5795d1576d0641cfb6fbb2057bf629506267a92807158584a13"
+checksum = "93857453250e3077bd71ff98b6a65ea6621a19bb0f559a85248955ac12c45a1a"
dependencies = [
"lock_api",
- "parking_lot_core 0.9.11",
+ "parking_lot_core 0.9.12",
]
[[package]]
@@ -5785,15 +5791,15 @@ dependencies = [
[[package]]
name = "parking_lot_core"
-version = "0.9.11"
+version = "0.9.12"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bc838d2a56b5b1a6c25f55575dfc605fabb63bb2365f6c2353ef9159aa69e4a5"
+checksum = "2621685985a2ebf1c516881c026032ac7deafcda1a2c9b7850dc81e3dfcb64c1"
dependencies = [
"cfg-if",
"libc",
- "redox_syscall 0.5.17",
+ "redox_syscall 0.5.18",
"smallvec",
- "windows-targets 0.52.6",
+ "windows-link 0.2.1",
]
[[package]]
@@ -5916,20 +5922,19 @@ checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220"
[[package]]
name = "pest"
-version = "2.8.2"
+version = "2.8.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "21e0a3a33733faeaf8651dfee72dd0f388f0c8e5ad496a3478fa5a922f49cfa8"
+checksum = "989e7521a040efde50c3ab6bbadafbe15ab6dc042686926be59ac35d74607df4"
dependencies = [
"memchr",
- "thiserror 2.0.17",
"ucd-trie",
]
[[package]]
name = "pest_derive"
-version = "2.8.2"
+version = "2.8.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bc58706f770acb1dbd0973e6530a3cff4746fb721207feb3a8a6064cd0b6c663"
+checksum = "187da9a3030dbafabbbfb20cb323b976dc7b7ce91fcd84f2f74d6e31d378e2de"
dependencies = [
"pest",
"pest_generator",
@@ -5937,9 +5942,9 @@ dependencies = [
[[package]]
name = "pest_generator"
-version = "2.8.2"
+version = "2.8.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6d4f36811dfe07f7b8573462465d5cb8965fffc2e71ae377a33aecf14c2c9a2f"
+checksum = "49b401d98f5757ebe97a26085998d6c0eecec4995cad6ab7fc30ffdf4b052843"
dependencies = [
"pest",
"pest_meta",
@@ -5950,9 +5955,9 @@ dependencies = [
[[package]]
name = "pest_meta"
-version = "2.8.2"
+version = "2.8.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "42919b05089acbd0a5dcd5405fb304d17d1053847b81163d09c4ad18ce8e8420"
+checksum = "72f27a2cfee9f9039c4d86faa5af122a0ac3851441a34865b8a043b46be0065a"
dependencies = [
"pest",
"sha2",
@@ -6039,7 +6044,7 @@ dependencies = [
"hermit-abi",
"pin-project-lite",
"rustix",
- "windows-sys 0.61.1",
+ "windows-sys 0.61.2",
]
[[package]]
@@ -6174,7 +6179,7 @@ version = "3.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "219cb19e96be00ab2e37d6e299658a0cfa83e52429179969b0f0121b4ac46983"
dependencies = [
- "toml_edit 0.23.6",
+ "toml_edit 0.23.7",
]
[[package]]
@@ -6248,7 +6253,7 @@ checksum = "e4500adecd7af8e0e9f4dbce15cfee07ce913fbf6ad605cc468b83f2d531ee94"
dependencies = [
"dtoa",
"itoa",
- "parking_lot 0.12.4",
+ "parking_lot 0.12.5",
"prometheus-client-derive-encode",
]
@@ -6424,7 +6429,7 @@ dependencies = [
"quinn-udp",
"rustc-hash",
"rustls",
- "socket2 0.5.10",
+ "socket2 0.6.0",
"thiserror 2.0.17",
"tokio",
"tracing",
@@ -6463,7 +6468,7 @@ dependencies = [
"cfg_aliases",
"libc",
"once_cell",
- "socket2 0.5.10",
+ "socket2 0.6.0",
"tracing",
"windows-sys 0.60.2",
]
@@ -6601,9 +6606,9 @@ dependencies = [
[[package]]
name = "redox_syscall"
-version = "0.5.17"
+version = "0.5.18"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5407465600fb0548f1442edf71dd20683c6ed326200ace4b1ef0763521bb3b77"
+checksum = "ed2bf2547551a7053d6fdfafda3f938979645c44812fbfcda098faae3f1a362d"
dependencies = [
"bitflags 2.9.4",
]
@@ -6728,7 +6733,7 @@ dependencies = [
"wasm-bindgen-futures",
"wasm-streams",
"web-sys",
- "webpki-roots 1.0.2",
+ "webpki-roots 1.0.3",
]
[[package]]
@@ -6832,7 +6837,7 @@ version = "0.6.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "41ab0892f4938752b34ae47cb53910b1b0921e55e77ddb6e44df666cab17939f"
dependencies = [
- "axum 0.8.5",
+ "axum 0.8.6",
"base64 0.22.1",
"bytes",
"chrono",
@@ -6966,7 +6971,7 @@ dependencies = [
"errno",
"libc",
"linux-raw-sys",
- "windows-sys 0.61.1",
+ "windows-sys 0.61.2",
]
[[package]]
@@ -7045,9 +7050,9 @@ checksum = "f87165f0995f63a9fbeea62b64d10b4d9d8e78ec6d7d51fb2125fda7bb36788f"
[[package]]
name = "rustls-webpki"
-version = "0.103.6"
+version = "0.103.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8572f3c2cb9934231157b45499fc41e1f58c589fdfb81a844ba873265e80f8eb"
+checksum = "e10b3f4191e8a80e6b43eebabfac91e5dcecebb27a71f04e820c47ec41d314bf"
dependencies = [
"aws-lc-rs",
"ring",
@@ -7110,7 +7115,7 @@ version = "0.1.28"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "891d81b926048e76efe18581bf793546b4c0eaf8448d72be8de2bbee5fd166e1"
dependencies = [
- "windows-sys 0.61.1",
+ "windows-sys 0.61.2",
]
[[package]]
@@ -7239,7 +7244,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5b55fb86dfd3a2f5f76ea78310a88f96c4ea21a3031f8d212443d56123fd0521"
dependencies = [
"libc",
- "windows-sys 0.61.1",
+ "windows-sys 0.61.2",
]
[[package]]
@@ -7371,9 +7376,9 @@ dependencies = [
[[package]]
name = "serde_spanned"
-version = "1.0.2"
+version = "1.0.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5417783452c2be558477e104686f7de5dae53dba813c28435e0e70f82d9b04ee"
+checksum = "e24345aa0fe688594e73770a5f6d1b216508b4f93484c0026d521acd30134392"
dependencies = [
"serde_core",
]
@@ -7392,9 +7397,9 @@ dependencies = [
[[package]]
name = "serde_with"
-version = "3.14.1"
+version = "3.15.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c522100790450cf78eeac1507263d0a350d4d5b30df0c8e1fe051a10c22b376e"
+checksum = "6093cd8c01b25262b84927e0f7151692158fab02d961e04c979d3903eba7ecc5"
dependencies = [
"base64 0.22.1",
"chrono",
@@ -7403,8 +7408,7 @@ dependencies = [
"indexmap 2.11.4",
"schemars 0.9.0",
"schemars 1.0.4",
- "serde",
- "serde_derive",
+ "serde_core",
"serde_json",
"serde_with_macros",
"time",
@@ -7412,9 +7416,9 @@ dependencies = [
[[package]]
name = "serde_with_macros"
-version = "3.14.1"
+version = "3.15.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "327ada00f7d64abaac1e55a6911e90cf665aa051b9a561c7006c157f4633135e"
+checksum = "a7e6c180db0816026a61afa1cff5344fb7ebded7e4d3062772179f2501481c27"
dependencies = [
"darling 0.21.3",
"proc-macro2",
@@ -7446,7 +7450,7 @@ dependencies = [
"futures",
"log",
"once_cell",
- "parking_lot 0.12.4",
+ "parking_lot 0.12.5",
"scc",
"serial_test_derive",
]
@@ -7472,7 +7476,7 @@ dependencies = [
"argon2",
"async-channel",
"async_zip",
- "axum 0.8.5",
+ "axum 0.8.6",
"axum-server",
"bincode 2.0.1",
"blake3",
@@ -7533,11 +7537,11 @@ dependencies = [
"socket2 0.6.0",
"static-toml",
"strum",
- "sysinfo 0.37.1",
+ "sysinfo 0.37.2",
"tempfile",
"test-case",
"thiserror 2.0.17",
- "toml 0.9.7",
+ "toml 0.9.8",
"tower-http",
"tracing",
"tracing-appender",
@@ -7974,9 +7978,9 @@ dependencies = [
[[package]]
name = "stable_deref_trait"
-version = "1.2.0"
+version = "1.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a8f112729512f8e442d81f95a8a7ddf2b7c6b8a1a6f509a95864142b30cab2d3"
+checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596"
[[package]]
name = "static-toml"
@@ -8126,9 +8130,9 @@ dependencies = [
[[package]]
name = "sysinfo"
-version = "0.37.1"
+version = "0.37.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3bddd368fda2f82ead69c03d46d351987cfa0c2a57abfa37a017f3aa3e9bf69a"
+checksum = "16607d5caffd1c07ce073528f9ed972d88db15dd44023fa57142963be3feb11f"
dependencies = [
"libc",
"memchr",
@@ -8160,7 +8164,7 @@ dependencies = [
"getrandom 0.3.3",
"once_cell",
"rustix",
- "windows-sys 0.61.1",
+ "windows-sys 0.61.2",
]
[[package]]
@@ -8220,7 +8224,7 @@ checksum = "c13547615a44dc9c452a8a534638acdf07120d4b6847c8178705da06306a3057"
dependencies = [
"smawk",
"unicode-linebreak",
- "unicode-width 0.2.1",
+ "unicode-width 0.2.2",
]
[[package]]
@@ -8350,7 +8354,7 @@ dependencies = [
"io-uring",
"libc",
"mio",
- "parking_lot 0.12.4",
+ "parking_lot 0.12.5",
"pin-project-lite",
"signal-hook-registry",
"slab",
@@ -8431,14 +8435,14 @@ dependencies = [
[[package]]
name = "toml"
-version = "0.9.7"
+version = "0.9.8"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "00e5e5d9bf2475ac9d4f0d9edab68cc573dc2fd644b0dba36b0c30a92dd9eaa0"
+checksum = "f0dc8b1fb61449e27716ec0e1bdf0f6b8f3e8f6b05391e8497b8b6d7804ea6d8"
dependencies = [
"indexmap 2.11.4",
"serde_core",
- "serde_spanned 1.0.2",
- "toml_datetime 0.7.2",
+ "serde_spanned 1.0.3",
+ "toml_datetime 0.7.3",
"toml_parser",
"toml_writer",
"winnow 0.7.13",
@@ -8455,9 +8459,9 @@ dependencies = [
[[package]]
name = "toml_datetime"
-version = "0.7.2"
+version = "0.7.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "32f1085dec27c2b6632b04c80b3bb1b4300d6495d1e129693bdda7d91e72eec1"
+checksum = "f2cdb639ebbc97961c51720f858597f7f24c4fc295327923af55b74c3c724533"
dependencies = [
"serde_core",
]
@@ -8489,21 +8493,21 @@ dependencies = [
[[package]]
name = "toml_edit"
-version = "0.23.6"
+version = "0.23.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f3effe7c0e86fdff4f69cdd2ccc1b96f933e24811c5441d44904e8683e27184b"
+checksum = "6485ef6d0d9b5d0ec17244ff7eb05310113c3f316f2d14200d4de56b3cb98f8d"
dependencies = [
"indexmap 2.11.4",
- "toml_datetime 0.7.2",
+ "toml_datetime 0.7.3",
"toml_parser",
"winnow 0.7.13",
]
[[package]]
name = "toml_parser"
-version = "1.0.3"
+version = "1.0.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4cf893c33be71572e0e9aa6dd15e6677937abd686b066eac3f8cd3531688a627"
+checksum = "c0cbe268d35bdb4bb5a56a2de88d0ad0eb70af5384a99d648cd4b3d04039800e"
dependencies = [
"winnow 0.7.13",
]
@@ -8516,9 +8520,9 @@ checksum = "5d99f8c9a7727884afe522e9bd5edbfc91a3312b36a77b5fb8926e4c31a41801"
[[package]]
name = "toml_writer"
-version = "1.0.3"
+version = "1.0.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d163a63c116ce562a22cda521fcc4d79152e7aba014456fb5eb442f6d6a10109"
+checksum = "df8b2b54733674ad286d16267dcfc7a71ed5c776e4ac7aa3c3e2561f7c637bf2"
[[package]]
name = "tonic"
@@ -8820,9 +8824,9 @@ checksum = "bc7d623258602320d5c55d1bc22793b57daff0ec7efc270ea7d55ce1d5f5471c"
[[package]]
name = "typenum"
-version = "1.18.0"
+version = "1.19.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1dccffe3ce07af9386bfd29e80c0ab1a8205a2fc34e4bcd40364df902cfa8f3f"
+checksum = "562d481066bde0658276a35467c4af00bdc6ee726305698a55b86e61d7ad82bb"
[[package]]
name = "ucd-trie"
@@ -8902,9 +8906,9 @@ checksum = "7dd6e30e90baa6f72411720665d41d89b9a3d039dc45b8faea1ddd07f617f6af"
[[package]]
name = "unicode-width"
-version = "0.2.1"
+version = "0.2.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4a1a07cc7db3810833284e8d372ccdc6da29741639ecc70c9ec107df0fa6154c"
+checksum = "b4ac048d71ede7ee76d585517add45da530660ef4390e49b098733c6e897f254"
[[package]]
name = "unicode-xid"
@@ -9256,9 +9260,9 @@ dependencies = [
[[package]]
name = "webpki-root-certs"
-version = "1.0.2"
+version = "1.0.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4e4ffd8df1c57e87c325000a3d6ef93db75279dc3a231125aac571650f22b12a"
+checksum = "05d651ec480de84b762e7be71e6efa7461699c19d9e2c272c8d93455f567786e"
dependencies = [
"rustls-pki-types",
]
@@ -9269,14 +9273,14 @@ version = "0.26.11"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "521bc38abb08001b01866da9f51eb7c5d647a19260e00054a8c7fd5f9e57f7a9"
dependencies = [
- "webpki-roots 1.0.2",
+ "webpki-roots 1.0.3",
]
[[package]]
name = "webpki-roots"
-version = "1.0.2"
+version = "1.0.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7e8983c3ab33d6fb807cfcdad2491c4ea8cbc8ed839181c7dfd9c67c83e261b2"
+checksum = "32b130c0d2d49f8b6889abc456e795e82525204f27c42cf767cf0d7734e089b8"
dependencies = [
"rustls-pki-types",
]
@@ -9293,9 +9297,9 @@ dependencies = [
[[package]]
name = "widestring"
-version = "1.2.0"
+version = "1.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "dd7cf3379ca1aac9eea11fba24fd7e315d621f8dfe35c8d7d2be8b793726e07d"
+checksum = "72069c3113ab32ab29e5584db3c6ec55d416895e60715417b5b883a357c3e471"
[[package]]
name = "winapi"
@@ -9319,7 +9323,7 @@ version = "0.1.11"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22"
dependencies = [
- "windows-sys 0.48.0",
+ "windows-sys 0.61.2",
]
[[package]]
@@ -9378,8 +9382,8 @@ version = "0.61.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c0fdd3ddb90610c7638aa2b3a3ab2904fb9e5cdbecc643ddb3647212781c4ae3"
dependencies = [
- "windows-implement 0.60.1",
- "windows-interface 0.59.2",
+ "windows-implement 0.60.2",
+ "windows-interface 0.59.3",
"windows-link 0.1.3",
"windows-result 0.3.4",
"windows-strings 0.4.2",
@@ -9387,15 +9391,15 @@ dependencies = [
[[package]]
name = "windows-core"
-version = "0.62.1"
+version = "0.62.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6844ee5416b285084d3d3fffd743b925a6c9385455f64f6d4fa3031c4c2749a9"
+checksum = "b8e83a14d34d0623b51dce9581199302a221863196a1dde71a7663a4c2be9deb"
dependencies = [
- "windows-implement 0.60.1",
- "windows-interface 0.59.2",
- "windows-link 0.2.0",
- "windows-result 0.4.0",
- "windows-strings 0.5.0",
+ "windows-implement 0.60.2",
+ "windows-interface 0.59.3",
+ "windows-link 0.2.1",
+ "windows-result 0.4.1",
+ "windows-strings 0.5.1",
]
[[package]]
@@ -9422,9 +9426,9 @@ dependencies = [
[[package]]
name = "windows-implement"
-version = "0.60.1"
+version = "0.60.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "edb307e42a74fb6de9bf3a02d9712678b22399c87e6fa869d6dfcd8c1b7754e0"
+checksum = "053e2e040ab57b9dc951b72c264860db7eb3b0200ba345b4e4c3b14f67855ddf"
dependencies = [
"proc-macro2",
"quote",
@@ -9444,9 +9448,9 @@ dependencies = [
[[package]]
name = "windows-interface"
-version = "0.59.2"
+version = "0.59.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c0abd1ddbc6964ac14db11c7213d6532ef34bd9aa042c2e5935f59d7908b46a5"
+checksum = "3f316c4a2570ba26bbec722032c4099d8c8bc095efccdc15688708623367e358"
dependencies = [
"proc-macro2",
"quote",
@@ -9461,9 +9465,9 @@ checksum = "5e6ad25900d524eaabdbbb96d20b4311e1e7ae1699af4fb28c17ae66c80d798a"
[[package]]
name = "windows-link"
-version = "0.2.0"
+version = "0.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "45e46c0661abb7180e7b9c281db115305d49ca1709ab8242adf09666d2173c65"
+checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5"
[[package]]
name = "windows-numerics"
@@ -9495,11 +9499,11 @@ dependencies = [
[[package]]
name = "windows-result"
-version = "0.4.0"
+version = "0.4.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7084dcc306f89883455a206237404d3eaf961e5bd7e0f312f7c91f57eb44167f"
+checksum = "7781fa89eaf60850ac3d2da7af8e5242a5ea78d1a11c49bf2910bb5a73853eb5"
dependencies = [
- "windows-link 0.2.0",
+ "windows-link 0.2.1",
]
[[package]]
@@ -9513,11 +9517,11 @@ dependencies = [
[[package]]
name = "windows-strings"
-version = "0.5.0"
+version = "0.5.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7218c655a553b0bed4426cf54b20d7ba363ef543b52d515b3e48d7fd55318dda"
+checksum = "7837d08f69c77cf6b07689544538e017c1bfcf57e34b4c0ff58e6c2cd3b37091"
dependencies = [
- "windows-link 0.2.0",
+ "windows-link 0.2.1",
]
[[package]]
@@ -9562,16 +9566,16 @@ version = "0.60.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f2f500e4d28234f72040990ec9d39e3a6b950f9f22d3dba18416c35882612bcb"
dependencies = [
- "windows-targets 0.53.4",
+ "windows-targets 0.53.5",
]
[[package]]
name = "windows-sys"
-version = "0.61.1"
+version = "0.61.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6f109e41dd4a3c848907eb83d5a42ea98b3769495597450cf6d153507b166f0f"
+checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc"
dependencies = [
- "windows-link 0.2.0",
+ "windows-link 0.2.1",
]
[[package]]
@@ -9622,19 +9626,19 @@ dependencies = [
[[package]]
name = "windows-targets"
-version = "0.53.4"
+version = "0.53.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2d42b7b7f66d2a06854650af09cfdf8713e427a439c97ad65a6375318033ac4b"
+checksum = "4945f9f551b88e0d65f3db0bc25c33b8acea4d9e41163edf90dcd0b19f9069f3"
dependencies = [
- "windows-link 0.2.0",
- "windows_aarch64_gnullvm 0.53.0",
- "windows_aarch64_msvc 0.53.0",
- "windows_i686_gnu 0.53.0",
- "windows_i686_gnullvm 0.53.0",
- "windows_i686_msvc 0.53.0",
- "windows_x86_64_gnu 0.53.0",
- "windows_x86_64_gnullvm 0.53.0",
- "windows_x86_64_msvc 0.53.0",
+ "windows-link 0.2.1",
+ "windows_aarch64_gnullvm 0.53.1",
+ "windows_aarch64_msvc 0.53.1",
+ "windows_i686_gnu 0.53.1",
+ "windows_i686_gnullvm 0.53.1",
+ "windows_i686_msvc 0.53.1",
+ "windows_x86_64_gnu 0.53.1",
+ "windows_x86_64_gnullvm 0.53.1",
+ "windows_x86_64_msvc 0.53.1",
]
[[package]]
@@ -9666,9 +9670,9 @@ checksum = "32a4622180e7a0ec044bb555404c800bc9fd9ec262ec147edd5989ccd0c02cd3"
[[package]]
name = "windows_aarch64_gnullvm"
-version = "0.53.0"
+version = "0.53.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "86b8d5f90ddd19cb4a147a5fa63ca848db3df085e25fee3cc10b39b6eebae764"
+checksum = "a9d8416fa8b42f5c947f8482c43e7d89e73a173cead56d044f6a56104a6d1b53"
[[package]]
name = "windows_aarch64_msvc"
@@ -9690,9 +9694,9 @@ checksum = "09ec2a7bb152e2252b53fa7803150007879548bc709c039df7627cabbd05d469"
[[package]]
name = "windows_aarch64_msvc"
-version = "0.53.0"
+version = "0.53.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c7651a1f62a11b8cbd5e0d42526e55f2c99886c77e007179efff86c2b137e66c"
+checksum = "b9d782e804c2f632e395708e99a94275910eb9100b2114651e04744e9b125006"
[[package]]
name = "windows_i686_gnu"
@@ -9714,9 +9718,9 @@ checksum = "8e9b5ad5ab802e97eb8e295ac6720e509ee4c243f69d781394014ebfe8bbfa0b"
[[package]]
name = "windows_i686_gnu"
-version = "0.53.0"
+version = "0.53.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c1dc67659d35f387f5f6c479dc4e28f1d4bb90ddd1a5d3da2e5d97b42d6272c3"
+checksum = "960e6da069d81e09becb0ca57a65220ddff016ff2d6af6a223cf372a506593a3"
[[package]]
name = "windows_i686_gnullvm"
@@ -9726,9 +9730,9 @@ checksum = "0eee52d38c090b3caa76c563b86c3a4bd71ef1a819287c19d586d7334ae8ed66"
[[package]]
name = "windows_i686_gnullvm"
-version = "0.53.0"
+version = "0.53.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9ce6ccbdedbf6d6354471319e781c0dfef054c81fbc7cf83f338a4296c0cae11"
+checksum = "fa7359d10048f68ab8b09fa71c3daccfb0e9b559aed648a8f95469c27057180c"
[[package]]
name = "windows_i686_msvc"
@@ -9750,9 +9754,9 @@ checksum = "240948bc05c5e7c6dabba28bf89d89ffce3e303022809e73deaefe4f6ec56c66"
[[package]]
name = "windows_i686_msvc"
-version = "0.53.0"
+version = "0.53.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "581fee95406bb13382d2f65cd4a908ca7b1e4c2f1917f143ba16efe98a589b5d"
+checksum = "1e7ac75179f18232fe9c285163565a57ef8d3c89254a30685b57d83a38d326c2"
[[package]]
name = "windows_x86_64_gnu"
@@ -9774,9 +9778,9 @@ checksum = "147a5c80aabfbf0c7d901cb5895d1de30ef2907eb21fbbab29ca94c5b08b1a78"
[[package]]
name = "windows_x86_64_gnu"
-version = "0.53.0"
+version = "0.53.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2e55b5ac9ea33f2fc1716d1742db15574fd6fc8dadc51caab1c16a3d3b4190ba"
+checksum = "9c3842cdd74a865a8066ab39c8a7a473c0778a3f29370b5fd6b4b9aa7df4a499"
[[package]]
name = "windows_x86_64_gnullvm"
@@ -9798,9 +9802,9 @@ checksum = "24d5b23dc417412679681396f2b49f3de8c1473deb516bd34410872eff51ed0d"
[[package]]
name = "windows_x86_64_gnullvm"
-version = "0.53.0"
+version = "0.53.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0a6e035dd0599267ce1ee132e51c27dd29437f63325753051e71dd9e42406c57"
+checksum = "0ffa179e2d07eee8ad8f57493436566c7cc30ac536a3379fdf008f47f6bb7ae1"
[[package]]
name = "windows_x86_64_msvc"
@@ -9822,9 +9826,9 @@ checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec"
[[package]]
name = "windows_x86_64_msvc"
-version = "0.53.0"
+version = "0.53.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "271414315aff87387382ec3d271b52d7ae78726f5d44ac98b4f4030c91880486"
+checksum = "d6bbff5f0aada427a1e5a6da5f1f98158182f26556f345ac9e04d36d0ebed650"
[[package]]
name = "winnow"
@@ -10044,9 +10048,9 @@ dependencies = [
[[package]]
name = "zeroize"
-version = "1.8.1"
+version = "1.8.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ced3678a2879b30306d323f4542626697a464a97c0a07c9aebf7ebca65cd4dde"
+checksum = "b97154e67e32c85465826e8bcc1c59429aaaf107c1e4a9e53c8d8ccd5eff88d0"
dependencies = [
"zeroize_derive",
]
diff --git a/core/binary_protocol/src/utils/mapper.rs b/core/binary_protocol/src/utils/mapper.rs
index 725932c5c..356d65931 100644
--- a/core/binary_protocol/src/utils/mapper.rs
+++ b/core/binary_protocol/src/utils/mapper.rs
@@ -898,7 +898,7 @@ fn map_to_client_info(
.map_err(|_| IggyError::InvalidNumberEncoding)?,
);
let user_id = match user_id {
- 0 => None,
+ u32::MAX => None,
_ => Some(user_id),
};
diff --git a/core/common/src/commands/users/defaults.rs b/core/common/src/commands/users/defaults.rs
index f18c8e0da..3d60f18d9 100644
--- a/core/common/src/commands/users/defaults.rs
+++ b/core/common/src/commands/users/defaults.rs
@@ -23,6 +23,6 @@ pub const MIN_PASSWORD_LENGTH: usize = 3;
pub const MAX_PAT_LENGTH: usize = 100;
pub const MAX_PERSONAL_ACCESS_TOKEN_NAME_LENGTH: usize = 30;
pub const MIN_PERSONAL_ACCESS_TOKEN_NAME_LENGTH: usize = 3;
-pub const DEFAULT_ROOT_USER_ID: u32 = 1;
+pub const DEFAULT_ROOT_USER_ID: u32 = 0;
pub const DEFAULT_ROOT_USERNAME: &str = "iggy";
pub const DEFAULT_ROOT_PASSWORD: &str = "iggy";
diff --git a/core/common/src/lib.rs b/core/common/src/lib.rs
index 55c4c70f8..809e98c98 100644
--- a/core/common/src/lib.rs
+++ b/core/common/src/lib.rs
@@ -66,11 +66,11 @@ pub use types::configuration::tcp_config::tcp_client_config::*;
pub use types::configuration::tcp_config::tcp_client_config_builder::*;
pub use types::configuration::tcp_config::tcp_client_reconnection_config::*;
pub use types::configuration::tcp_config::tcp_connection_string_options::*;
+pub use types::configuration::transport::TransportProtocol;
pub use types::configuration::websocket_config::websocket_client_config::*;
pub use types::configuration::websocket_config::websocket_client_config_builder::*;
pub use types::configuration::websocket_config::websocket_client_reconnection_config::*;
pub use types::configuration::websocket_config::websocket_connection_string_options::*;
-pub use types::configuration::transport::TransportProtocol;
pub use types::consumer::consumer_group::*;
pub use types::consumer::consumer_kind::*;
pub use types::consumer::consumer_offset_info::*;
diff --git a/core/common/src/types/configuration/mod.rs b/core/common/src/types/configuration/mod.rs
index 727cce158..667e58884 100644
--- a/core/common/src/types/configuration/mod.rs
+++ b/core/common/src/types/configuration/mod.rs
@@ -19,5 +19,5 @@ pub(crate) mod auth_config;
pub(crate) mod http_config;
pub(crate) mod quic_config;
pub(crate) mod tcp_config;
-pub(crate) mod websocket_config;
pub(crate) mod transport;
+pub(crate) mod websocket_config;
diff --git a/core/integration/tests/cli/client/test_client_get_command.rs b/core/integration/tests/cli/client/test_client_get_command.rs
index d76c09e47..734341758 100644
--- a/core/integration/tests/cli/client/test_client_get_command.rs
+++ b/core/integration/tests/cli/client/test_client_get_command.rs
@@ -67,7 +67,7 @@ impl IggyCmdTestCase for TestClientGetCmd {
"Client ID | {}",
self.get_client_id()
)))
- .stdout(contains("User ID | 1"));
+ .stdout(contains("User ID | 0"));
}
async fn verify_server_state(&self, _client: &dyn Client) {}
diff --git a/core/integration/tests/cli/common/mod.rs b/core/integration/tests/cli/common/mod.rs
index 0aa7c4123..bebf46f21 100644
--- a/core/integration/tests/cli/common/mod.rs
+++ b/core/integration/tests/cli/common/mod.rs
@@ -128,7 +128,7 @@ impl IggyCmdTest {
.await
.unwrap();
- assert_eq!(identity_info.user_id, 1);
+ assert_eq!(identity_info.user_id, 0);
}
pub(crate) async fn execute_test(&mut self, mut test_case: impl IggyCmdTestCase) {
diff --git a/core/integration/tests/cli/user/test_user_create_command.rs b/core/integration/tests/cli/user/test_user_create_command.rs
index 3678185bd..5c4eecdb8 100644
--- a/core/integration/tests/cli/user/test_user_create_command.rs
+++ b/core/integration/tests/cli/user/test_user_create_command.rs
@@ -193,7 +193,7 @@ pub async fn should_be_successful() {
vec![String::from("3")],
Some(Permissions {
global: GlobalPermissions::default(),
- streams: Some(AHashMap::from([(3u32, StreamPermissions::default())])),
+ streams: Some(AHashMap::from([(3usize, StreamPermissions::default())])),
}),
),
))
@@ -209,7 +209,7 @@ pub async fn should_be_successful() {
Some(Permissions {
global: GlobalPermissions::default(),
streams: Some(AHashMap::from([(
- 1u32,
+ 1usize,
StreamPermissions {
topics: Some(AHashMap::from([(
1,
@@ -249,7 +249,7 @@ pub async fn should_be_successful() {
send_messages: false,
},
streams: Some(AHashMap::from([(
- 2u32,
+ 2usize,
StreamPermissions {
topics: Some(AHashMap::from([(
1,
diff --git a/core/integration/tests/cli/user/test_user_delete_command.rs b/core/integration/tests/cli/user/test_user_delete_command.rs
index dd501eb71..fd0f327b7 100644
--- a/core/integration/tests/cli/user/test_user_delete_command.rs
+++ b/core/integration/tests/cli/user/test_user_delete_command.rs
@@ -111,7 +111,7 @@ pub async fn should_be_successful() {
String::from("username"),
String::from("password"),
UserStatus::Active,
- 2,
+ 1,
TestUserId::Numeric,
))
.await;
@@ -120,7 +120,7 @@ pub async fn should_be_successful() {
String::from("testuser"),
String::from("testpass"),
UserStatus::Inactive,
- 2,
+ 1,
TestUserId::Named,
))
.await;
diff --git a/core/integration/tests/cli/user/test_user_get_command.rs b/core/integration/tests/cli/user/test_user_get_command.rs
index c52ac4547..ba67730a1 100644
--- a/core/integration/tests/cli/user/test_user_get_command.rs
+++ b/core/integration/tests/cli/user/test_user_get_command.rs
@@ -31,8 +31,8 @@ enum TestUser<'a> {
Existing(&'a str),
New(&'a str),
NewWithGlobalPerms(&'a str),
- NewWithStreamPerms(&'a str, u32),
- NewWithStreamAndTopicPerms(&'a str, u32, u32),
+ NewWithStreamPerms(&'a str, usize),
+ NewWithStreamAndTopicPerms(&'a str, usize, usize),
}
struct TestUserGetCmd {
@@ -42,8 +42,8 @@ struct TestUserGetCmd {
using_user_id: TestUserId,
create_user: bool,
check_global_perms: bool,
- check_stream_perms: Option<u32>,
- check_topic_perms: Option<u32>,
+ check_stream_perms: Option<usize>,
+ check_topic_perms: Option<usize>,
}
impl TestUserGetCmd {
diff --git a/core/integration/tests/cli/user/test_user_permissions_command.rs b/core/integration/tests/cli/user/test_user_permissions_command.rs
index 37958231f..116dbb665 100644
--- a/core/integration/tests/cli/user/test_user_permissions_command.rs
+++ b/core/integration/tests/cli/user/test_user_permissions_command.rs
@@ -159,7 +159,7 @@ pub async fn should_be_successful() {
vec![String::from("3")],
Some(Permissions {
global: GlobalPermissions::default(),
- streams: Some(AHashMap::from([(3u32, StreamPermissions::default())])),
+ streams: Some(AHashMap::from([(3usize, StreamPermissions::default())])),
}),
),
TestUserId::Numeric,
@@ -174,7 +174,7 @@ pub async fn should_be_successful() {
Some(Permissions {
global: GlobalPermissions::default(),
streams: Some(AHashMap::from([(
- 1u32,
+ 1usize,
StreamPermissions {
topics: Some(AHashMap::from([(
2,
@@ -213,10 +213,10 @@ pub async fn should_be_successful() {
send_messages: false,
},
streams: Some(AHashMap::from([(
- 2u32,
+ 2usize,
StreamPermissions {
topics: Some(AHashMap::from([(
- 2u32,
+ 2usize,
TopicPermissions {
manage_topic: false,
read_topic: false,
diff --git a/core/integration/tests/server/concurrent_addition.rs b/core/integration/tests/server/concurrent_addition.rs
new file mode 100644
index 000000000..de0aa9c14
--- /dev/null
+++ b/core/integration/tests/server/concurrent_addition.rs
@@ -0,0 +1,122 @@
+/* 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.
+ */
+
+use crate::server::scenarios::concurrent_scenario::{
+ self, ClientCount, ResourceType, ScenarioType,
+};
+use iggy_common::TransportProtocol;
+use integration::{
+ http_client::HttpClientFactory, quic_client::QuicClientFactory, tcp_client::TcpClientFactory,
+ test_server::TestServer, websocket_client::WebSocketClientFactory,
+};
+use serial_test::parallel;
+use test_case::test_matrix;
+
+// Test matrix for race condition scenarios
+// Tests all combinations of:
+// - Transport: TCP, HTTP, QUIC, WebSocket (4)
+// - Resource: User, Stream, Topic (3)
+// - Clients: Single, Multiple (2)
+// - Path: Hot (unique names), Cold (duplicate names) (2)
+// Total: 4 × 3 × 2 × 2 = 48 test cases
+
+#[test_matrix(
+ [tcp(), http(), quic(), websocket()],
+ [user(), stream(), topic()],
+ [single_client(), multiple_clients()],
+ [hot(), cold()]
+)]
+#[tokio::test]
+#[parallel]
+async fn matrix(
+ transport: TransportProtocol,
+ resource_type: ResourceType,
+ client_count: ClientCount,
+ path_type: ScenarioType,
+) {
+ let mut test_server = TestServer::default();
+ test_server.start();
+
+ let client_factory: Box<dyn integration::test_server::ClientFactory> = match transport {
+ TransportProtocol::Tcp => {
+ let server_addr = test_server.get_raw_tcp_addr().unwrap();
+ Box::new(TcpClientFactory {
+ server_addr,
+ ..Default::default()
+ })
+ }
+ TransportProtocol::Quic => {
+ let server_addr = test_server.get_quic_udp_addr().unwrap();
+ Box::new(QuicClientFactory { server_addr })
+ }
+ TransportProtocol::Http => {
+ let server_addr = test_server.get_http_api_addr().unwrap();
+ Box::new(HttpClientFactory { server_addr })
+ }
+ TransportProtocol::WebSocket => {
+ let server_addr = test_server.get_websocket_addr().unwrap();
+ Box::new(WebSocketClientFactory { server_addr })
+ }
+ };
+
+ concurrent_scenario::run(&*client_factory, resource_type, client_count, path_type).await;
+}
+
+fn tcp() -> TransportProtocol {
+ TransportProtocol::Tcp
+}
+
+fn http() -> TransportProtocol {
+ TransportProtocol::Http
+}
+
+fn quic() -> TransportProtocol {
+ TransportProtocol::Quic
+}
+
+fn websocket() -> TransportProtocol {
+ TransportProtocol::WebSocket
+}
+
+fn user() -> ResourceType {
+ ResourceType::User
+}
+
+fn stream() -> ResourceType {
+ ResourceType::Stream
+}
+
+fn topic() -> ResourceType {
+ ResourceType::Topic
+}
+
+fn single_client() -> ClientCount {
+ ClientCount::Single
+}
+
+fn multiple_clients() -> ClientCount {
+ ClientCount::Multiple(10)
+}
+
+fn hot() -> ScenarioType {
+ ScenarioType::Hot
+}
+
+fn cold() -> ScenarioType {
+ ScenarioType::Cold
+}
diff --git a/core/integration/tests/server/mod.rs b/core/integration/tests/server/mod.rs
index 335d2d269..8a2e2c2f5 100644
--- a/core/integration/tests/server/mod.rs
+++ b/core/integration/tests/server/mod.rs
@@ -17,6 +17,7 @@
*/
mod cg;
+mod concurrent_addition;
mod general;
mod scenarios;
mod specific;
diff --git a/core/integration/tests/server/scenarios/concurrent_scenario.rs b/core/integration/tests/server/scenarios/concurrent_scenario.rs
new file mode 100644
index 000000000..88e76b3fd
--- /dev/null
+++ b/core/integration/tests/server/scenarios/concurrent_scenario.rs
@@ -0,0 +1,703 @@
+/* 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.
+ */
+
+use crate::server::scenarios::create_client;
+use futures::future::join_all;
+use iggy::prelude::*;
+use integration::test_server::{ClientFactory, login_root};
+
+const OPERATIONS_COUNT: usize = 20;
+const MULTIPLE_CLIENT_COUNT: usize = 10;
+const OPERATIONS_PER_CLIENT: usize = OPERATIONS_COUNT / MULTIPLE_CLIENT_COUNT;
+const USER_PASSWORD: &str = "secret";
+const TEST_STREAM_NAME: &str = "race-test-stream";
+const PARTITIONS_COUNT: u32 = 1;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum ResourceType {
+ User,
+ Stream,
+ Topic,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum ClientCount {
+ Single,
+ Multiple(usize),
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum ScenarioType {
+ Hot, // Different names - all should succeed
+ Cold, // Same name - only first should succeed
+}
+
+type OperationResult = Result<(), IggyError>;
+
+pub async fn run(
+ client_factory: &dyn ClientFactory,
+ resource_type: ResourceType,
+ client_count: ClientCount,
+ scenario_type: ScenarioType,
+) {
+ let root_client = create_client(client_factory).await;
+ login_root(&root_client).await;
+
+ // For topic tests, create parent stream first
+ if resource_type == ResourceType::Topic {
+ root_client.create_stream(TEST_STREAM_NAME).await.unwrap();
+ }
+
+ let results = match (resource_type, client_count, scenario_type) {
+ (ResourceType::User, ClientCount::Single, ScenarioType::Hot) => {
+ execute_single_client_users_hot(&root_client).await
+ }
+ (ResourceType::User, ClientCount::Single, ScenarioType::Cold) => {
+ execute_single_client_users_cold(&root_client).await
+ }
+ (ResourceType::User, ClientCount::Multiple(_), ScenarioType::Hot) => {
+ execute_multiple_clients_users_hot(client_factory).await
+ }
+ (ResourceType::User, ClientCount::Multiple(_), ScenarioType::Cold) => {
+ execute_multiple_clients_users_cold(client_factory).await
+ }
+ (ResourceType::Stream, ClientCount::Single, ScenarioType::Hot) => {
+ execute_single_client_streams_hot(&root_client).await
+ }
+ (ResourceType::Stream, ClientCount::Single, ScenarioType::Cold) => {
+ execute_single_client_streams_cold(&root_client).await
+ }
+ (ResourceType::Stream, ClientCount::Multiple(_), ScenarioType::Hot) => {
+ execute_multiple_clients_streams_hot(client_factory).await
+ }
+ (ResourceType::Stream, ClientCount::Multiple(_), ScenarioType::Cold) => {
+ execute_multiple_clients_streams_cold(client_factory).await
+ }
+ (ResourceType::Topic, ClientCount::Single, ScenarioType::Hot) => {
+ execute_single_client_topics_hot(&root_client).await
+ }
+ (ResourceType::Topic, ClientCount::Single, ScenarioType::Cold) => {
+ execute_single_client_topics_cold(&root_client).await
+ }
+ (ResourceType::Topic, ClientCount::Multiple(_), ScenarioType::Hot) => {
+ execute_multiple_clients_topics_hot(client_factory).await
+ }
+ (ResourceType::Topic, ClientCount::Multiple(_), ScenarioType::Cold) => {
+ execute_multiple_clients_topics_cold(client_factory).await
+ }
+ };
+
+ validate_results(&results, scenario_type);
+ validate_server_state(&root_client, resource_type, scenario_type).await;
+ cleanup_resources(&root_client, resource_type).await;
+}
+
+async fn execute_single_client_users_hot(client: &IggyClient) -> Vec<OperationResult> {
+ let mut results = Vec::with_capacity(OPERATIONS_COUNT);
+
+ for i in 0..OPERATIONS_COUNT {
+ let username = format!("race-user-{}", i);
+ let result = client
+ .create_user(&username, USER_PASSWORD, UserStatus::Active, None)
+ .await
+ .map(|_| ());
+ results.push(result);
+ }
+
+ results
+}
+
+async fn execute_single_client_users_cold(client: &IggyClient) -> Vec<OperationResult> {
+ let mut results = Vec::with_capacity(OPERATIONS_COUNT);
+ const DUPLICATE_USER: &str = "race-user-duplicate";
+
+ for _ in 0..OPERATIONS_COUNT {
+ let result = client
+ .create_user(DUPLICATE_USER, USER_PASSWORD, UserStatus::Active, None)
+ .await
+ .map(|_| ());
+ results.push(result);
+ }
+
+ results
+}
+
+async fn execute_multiple_clients_users_hot(
+ client_factory: &dyn ClientFactory,
+) -> Vec<OperationResult> {
+ let mut handles = Vec::with_capacity(MULTIPLE_CLIENT_COUNT);
+
+ for client_id in 0..MULTIPLE_CLIENT_COUNT {
+ let client = create_client(client_factory).await;
+ login_root(&client).await;
+
+ let handle = tokio::spawn(async move {
+ let mut results = Vec::with_capacity(OPERATIONS_PER_CLIENT);
+ for i in 0..OPERATIONS_PER_CLIENT {
+ let username = format!("race-user-{}-{}", client_id, i);
+ let result = client
+ .create_user(&username, USER_PASSWORD, UserStatus::Active, None)
+ .await
+ .map(|_| ());
+ results.push(result);
+ }
+ results
+ });
+
+ handles.push(handle);
+ }
+
+ // Await all concurrent operations
+ let all_results = join_all(handles).await;
+ all_results
+ .into_iter()
+ .flat_map(|r| r.expect("Tokio task panicked"))
+ .collect()
+}
+
+async fn execute_multiple_clients_users_cold(
+ client_factory: &dyn ClientFactory,
+) -> Vec<OperationResult> {
+ let mut handles = Vec::with_capacity(MULTIPLE_CLIENT_COUNT);
+ const DUPLICATE_USER: &str = "race-user-duplicate";
+
+ for _ in 0..MULTIPLE_CLIENT_COUNT {
+ let client = create_client(client_factory).await;
+ login_root(&client).await;
+
+ let handle = tokio::spawn(async move {
+ let mut results = Vec::with_capacity(OPERATIONS_PER_CLIENT);
+ for _ in 0..OPERATIONS_PER_CLIENT {
+ let result = client
+ .create_user(DUPLICATE_USER, USER_PASSWORD, UserStatus::Active, None)
+ .await
+ .map(|_| ());
+ results.push(result);
+ }
+ results
+ });
+
+ handles.push(handle);
+ }
+
+ let all_results = join_all(handles).await;
+ all_results
+ .into_iter()
+ .flat_map(|r| r.expect("Tokio task panicked"))
+ .collect()
+}
+
+async fn execute_single_client_streams_hot(client: &IggyClient) -> Vec<OperationResult> {
+ let mut results = Vec::with_capacity(OPERATIONS_COUNT);
+
+ for i in 0..OPERATIONS_COUNT {
+ let stream_name = format!("race-stream-{}", i);
+ let result = client.create_stream(&stream_name).await.map(|_| ());
+ results.push(result);
+ }
+
+ results
+}
+
+async fn execute_single_client_streams_cold(client: &IggyClient) -> Vec<OperationResult> {
+ let mut results = Vec::with_capacity(OPERATIONS_COUNT);
+ const DUPLICATE_STREAM: &str = "race-stream-duplicate";
+
+ for _ in 0..OPERATIONS_COUNT {
+ let result = client.create_stream(DUPLICATE_STREAM).await.map(|_| ());
+ results.push(result);
+ }
+
+ results
+}
+
+// ============================================================================
+// MULTIPLE CLIENTS - STREAMS
+// ============================================================================
+
+async fn execute_multiple_clients_streams_hot(
+ client_factory: &dyn ClientFactory,
+) -> Vec<OperationResult> {
+ let mut handles = Vec::with_capacity(MULTIPLE_CLIENT_COUNT);
+
+ for client_id in 0..MULTIPLE_CLIENT_COUNT {
+ let client = create_client(client_factory).await;
+ login_root(&client).await;
+
+ let handle = tokio::spawn(async move {
+ let mut results = Vec::with_capacity(OPERATIONS_PER_CLIENT);
+ for i in 0..OPERATIONS_PER_CLIENT {
+ let stream_name = format!("race-stream-{}-{}", client_id, i);
+ let result = client.create_stream(&stream_name).await.map(|_| ());
+ results.push(result);
+ }
+ results
+ });
+
+ handles.push(handle);
+ }
+
+ let all_results = join_all(handles).await;
+ all_results
+ .into_iter()
+ .flat_map(|r| r.expect("Tokio task panicked"))
+ .collect()
+}
+
+async fn execute_multiple_clients_streams_cold(
+ client_factory: &dyn ClientFactory,
+) -> Vec<OperationResult> {
+ let mut handles = Vec::with_capacity(MULTIPLE_CLIENT_COUNT);
+ const DUPLICATE_STREAM: &str = "race-stream-duplicate";
+
+ for _ in 0..MULTIPLE_CLIENT_COUNT {
+ let client = create_client(client_factory).await;
+ login_root(&client).await;
+
+ let handle = tokio::spawn(async move {
+ let mut results = Vec::with_capacity(OPERATIONS_PER_CLIENT);
+ for _ in 0..OPERATIONS_PER_CLIENT {
+ let result = client.create_stream(DUPLICATE_STREAM).await.map(|_| ());
+ results.push(result);
+ }
+ results
+ });
+
+ handles.push(handle);
+ }
+
+ let all_results = join_all(handles).await;
+ all_results
+ .into_iter()
+ .flat_map(|r| r.expect("Tokio task panicked"))
+ .collect()
+}
+
+async fn execute_single_client_topics_hot(client: &IggyClient) -> Vec<OperationResult> {
+ let mut results = Vec::with_capacity(OPERATIONS_COUNT);
+ let stream_id = Identifier::named(TEST_STREAM_NAME).unwrap();
+
+ for i in 0..OPERATIONS_COUNT {
+ let topic_name = format!("race-topic-{}", i);
+ let result = client
+ .create_topic(
+ &stream_id,
+ &topic_name,
+ PARTITIONS_COUNT,
+ CompressionAlgorithm::default(),
+ None,
+ IggyExpiry::NeverExpire,
+ MaxTopicSize::ServerDefault,
+ )
+ .await
+ .map(|_| ());
+ results.push(result);
+ }
+
+ results
+}
+
+async fn execute_single_client_topics_cold(client: &IggyClient) -> Vec<OperationResult> {
+ let mut results = Vec::with_capacity(OPERATIONS_COUNT);
+ const DUPLICATE_TOPIC: &str = "race-topic-duplicate";
+ let stream_id = Identifier::named(TEST_STREAM_NAME).unwrap();
+
+ for _ in 0..OPERATIONS_COUNT {
+ let result = client
+ .create_topic(
+ &stream_id,
+ DUPLICATE_TOPIC,
+ PARTITIONS_COUNT,
+ CompressionAlgorithm::default(),
+ None,
+ IggyExpiry::NeverExpire,
+ MaxTopicSize::ServerDefault,
+ )
+ .await
+ .map(|_| ());
+ results.push(result);
+ }
+
+ results
+}
+
+async fn execute_multiple_clients_topics_hot(
+ client_factory: &dyn ClientFactory,
+) -> Vec<OperationResult> {
+ let mut handles = Vec::with_capacity(MULTIPLE_CLIENT_COUNT);
+
+ for client_id in 0..MULTIPLE_CLIENT_COUNT {
+ let client = create_client(client_factory).await;
+ login_root(&client).await;
+
+ let handle = tokio::spawn(async move {
+ let mut results = Vec::with_capacity(OPERATIONS_PER_CLIENT);
+ let stream_id = Identifier::named(TEST_STREAM_NAME).unwrap();
+
+ for i in 0..OPERATIONS_PER_CLIENT {
+ let topic_name = format!("race-topic-{}-{}", client_id, i);
+ let result = client
+ .create_topic(
+ &stream_id,
+ &topic_name,
+ PARTITIONS_COUNT,
+ CompressionAlgorithm::default(),
+ None,
+ IggyExpiry::NeverExpire,
+ MaxTopicSize::ServerDefault,
+ )
+ .await
+ .map(|_| ());
+ results.push(result);
+ }
+ results
+ });
+
+ handles.push(handle);
+ }
+
+ let all_results = join_all(handles).await;
+ all_results
+ .into_iter()
+ .flat_map(|r| r.expect("Tokio task panicked"))
+ .collect()
+}
+
+async fn execute_multiple_clients_topics_cold(
+ client_factory: &dyn ClientFactory,
+) -> Vec<OperationResult> {
+ let mut handles = Vec::with_capacity(MULTIPLE_CLIENT_COUNT);
+ const DUPLICATE_TOPIC: &str = "race-topic-duplicate";
+
+ for _ in 0..MULTIPLE_CLIENT_COUNT {
+ let client = create_client(client_factory).await;
+ login_root(&client).await;
+
+ let handle = tokio::spawn(async move {
+ let mut results = Vec::with_capacity(OPERATIONS_PER_CLIENT);
+ let stream_id = Identifier::named(TEST_STREAM_NAME).unwrap();
+
+ for _ in 0..OPERATIONS_PER_CLIENT {
+ let result = client
+ .create_topic(
+ &stream_id,
+ DUPLICATE_TOPIC,
+ PARTITIONS_COUNT,
+ CompressionAlgorithm::default(),
+ None,
+ IggyExpiry::NeverExpire,
+ MaxTopicSize::ServerDefault,
+ )
+ .await
+ .map(|_| ());
+ results.push(result);
+ }
+ results
+ });
+
+ handles.push(handle);
+ }
+
+ let all_results = join_all(handles).await;
+ all_results
+ .into_iter()
+ .flat_map(|r| r.expect("Tokio task panicked"))
+ .collect()
+}
+
+fn validate_results(results: &[OperationResult], scenario_type: ScenarioType) {
+ let success_count = results.iter().filter(|r| r.is_ok()).count();
+ let error_count = results.iter().filter(|r| r.is_err()).count();
+
+ match scenario_type {
+ ScenarioType::Hot => {
+ // Hot path: all operations should succeed
+ assert_eq!(
+ success_count,
+ OPERATIONS_COUNT,
+ "Hot path: Expected all {} operations to succeed, but only {} succeeded. Errors: {:?}",
+ OPERATIONS_COUNT,
+ success_count,
+ results.iter().filter(|r| r.is_err()).collect::<Vec<_>>()
+ );
+ assert_eq!(
+ error_count, 0,
+ "Hot path: Expected 0 errors, but got {}",
+ error_count
+ );
+ }
+ ScenarioType::Cold => {
+ // Cold path: exactly 1 success, rest failures
+ assert_eq!(
+ success_count, 1,
+ "Cold path: Expected exactly 1 success, but got {}. All results: {:?}",
+ success_count, results
+ );
+ assert_eq!(
+ error_count,
+ OPERATIONS_COUNT - 1,
+ "Cold path: Expected {} errors, but got {}",
+ OPERATIONS_COUNT - 1,
+ error_count
+ );
+
+ // Verify errors are appropriate "already exists" errors
+ for result in results.iter().filter(|r| r.is_err()) {
+ let err = result.as_ref().unwrap_err();
+ assert!(
+ matches!(
+ err,
+ IggyError::UserAlreadyExists
+ | IggyError::StreamIdAlreadyExists(_)
+ | IggyError::StreamNameAlreadyExists(_)
+ | IggyError::TopicIdAlreadyExists(_, _)
+ | IggyError::TopicNameAlreadyExists(_, _)
+ | IggyError::HttpResponseError(400, _)
+ ),
+ "Expected 'already exists' error, got: {:?}",
+ err
+ );
+ }
+ }
+ }
+}
+
+async fn validate_server_state(
+ client: &IggyClient,
+ resource_type: ResourceType,
+ scenario_type: ScenarioType,
+) {
+ match resource_type {
+ ResourceType::User => validate_users_state(client, scenario_type).await,
+ ResourceType::Stream => validate_streams_state(client, scenario_type).await,
+ ResourceType::Topic => validate_topics_state(client, scenario_type).await,
+ }
+}
+
+async fn validate_users_state(client: &IggyClient, scenario_type: ScenarioType) {
+ let users = client.get_users().await.expect("Failed to get users");
+ let test_users: Vec<_> = users
+ .iter()
+ .filter(|u| u.username.starts_with("race-user-"))
+ .collect();
+
+ match scenario_type {
+ ScenarioType::Hot => {
+ // Hot path: should have OPERATIONS_COUNT unique users
+ assert_eq!(
+ test_users.len(),
+ OPERATIONS_COUNT,
+ "Hot path: Expected {} users, but found {}. Users: {:?}",
+ OPERATIONS_COUNT,
+ test_users.len(),
+ test_users.iter().map(|u| &u.username).collect::<Vec<_>>()
+ );
+
+ // All users should have unique IDs
+ let mut ids: Vec<u32> = test_users.iter().map(|u| u.id).collect();
+ ids.sort_unstable();
+ let unique_ids: std::collections::HashSet<u32> = ids.iter().copied().collect();
+ assert_eq!(
+ unique_ids.len(),
+ test_users.len(),
+ "Hot path: Found duplicate user IDs: {:?}",
+ ids
+ );
+
+ // All users should have unique names
+ let names: std::collections::HashSet<&str> =
+ test_users.iter().map(|u| u.username.as_str()).collect();
+ assert_eq!(
+ names.len(),
+ test_users.len(),
+ "Hot path: Found duplicate usernames"
+ );
+ }
+ ScenarioType::Cold => {
+ // Cold path: should have exactly 1 user with duplicate name
+ assert_eq!(
+ test_users.len(),
+ 1,
+ "Cold path: Expected exactly 1 user, but found {}. Users: {:?}",
+ test_users.len(),
+ test_users.iter().map(|u| &u.username).collect::<Vec<_>>()
+ );
+
+ let user = test_users[0];
+ assert_eq!(
+ user.username, "race-user-duplicate",
+ "Cold path: Expected user named 'race-user-duplicate', found '{}'",
+ user.username
+ );
+ }
+ }
+}
+
+async fn validate_streams_state(client: &IggyClient, scenario_type: ScenarioType) {
+ let streams = client.get_streams().await.expect("Failed to get streams");
+ let test_streams: Vec<_> = streams
+ .iter()
+ .filter(|s| s.name.starts_with("race-stream-"))
+ .collect();
+
+ match scenario_type {
+ ScenarioType::Hot => {
+ // Hot path: should have OPERATIONS_COUNT unique streams
+ assert_eq!(
+ test_streams.len(),
+ OPERATIONS_COUNT,
+ "Hot path: Expected {} streams, but found {}. Streams: {:?}",
+ OPERATIONS_COUNT,
+ test_streams.len(),
+ test_streams.iter().map(|s| &s.name).collect::<Vec<_>>()
+ );
+
+ // All streams should have unique IDs
+ let mut ids: Vec<u32> = test_streams.iter().map(|s| s.id).collect();
+ ids.sort_unstable();
+ let unique_ids: std::collections::HashSet<u32> = ids.iter().copied().collect();
+ assert_eq!(
+ unique_ids.len(),
+ test_streams.len(),
+ "Hot path: Found duplicate stream IDs: {:?}",
+ ids
+ );
+
+ // All streams should have unique names
+ let names: std::collections::HashSet<&str> =
+ test_streams.iter().map(|s| s.name.as_str()).collect();
+ assert_eq!(
+ names.len(),
+ test_streams.len(),
+ "Hot path: Found duplicate stream names"
+ );
+ }
+ ScenarioType::Cold => {
+ // Cold path: should have exactly 1 stream with duplicate name
+ assert_eq!(
+ test_streams.len(),
+ 1,
+ "Cold path: Expected exactly 1 stream, but found {}. Streams: {:?}",
+ test_streams.len(),
+ test_streams.iter().map(|s| &s.name).collect::<Vec<_>>()
+ );
+
+ let stream = test_streams[0];
+ assert_eq!(
+ stream.name, "race-stream-duplicate",
+ "Cold path: Expected stream named 'race-stream-duplicate', found '{}'",
+ stream.name
+ );
+ }
+ }
+}
+
+async fn validate_topics_state(client: &IggyClient, scenario_type: ScenarioType) {
+ let stream_id = Identifier::named(TEST_STREAM_NAME).unwrap();
+ let stream = client
+ .get_stream(&stream_id)
+ .await
+ .expect("Failed to get test stream")
+ .expect("Test stream not found");
+
+ let test_topics: Vec<_> = stream
+ .topics
+ .iter()
+ .filter(|t| t.name.starts_with("race-topic-"))
+ .collect();
+
+ match scenario_type {
+ ScenarioType::Hot => {
+ // Hot path: should have OPERATIONS_COUNT unique topics
+ assert_eq!(
+ test_topics.len(),
+ OPERATIONS_COUNT,
+ "Hot path: Expected {} topics, but found {}. Topics: {:?}",
+ OPERATIONS_COUNT,
+ test_topics.len(),
+ test_topics.iter().map(|t| &t.name).collect::<Vec<_>>()
+ );
+
+ // All topics should have unique IDs
+ let mut ids: Vec<u32> = test_topics.iter().map(|t| t.id).collect();
+ ids.sort_unstable();
+ let unique_ids: std::collections::HashSet<u32> = ids.iter().copied().collect();
+ assert_eq!(
+ unique_ids.len(),
+ test_topics.len(),
+ "Hot path: Found duplicate topic IDs: {:?}",
+ ids
+ );
+
+ // All topics should have unique names
+ let names: std::collections::HashSet<&str> =
+ test_topics.iter().map(|t| t.name.as_str()).collect();
+ assert_eq!(
+ names.len(),
+ test_topics.len(),
+ "Hot path: Found duplicate topic names"
+ );
+ }
+ ScenarioType::Cold => {
+ // Cold path: should have exactly 1 topic with duplicate name
+ assert_eq!(
+ test_topics.len(),
+ 1,
+ "Cold path: Expected exactly 1 topic, but found {}. Topics: {:?}",
+ test_topics.len(),
+ test_topics.iter().map(|t| &t.name).collect::<Vec<_>>()
+ );
+
+ let topic = test_topics[0];
+ assert_eq!(
+ topic.name, "race-topic-duplicate",
+ "Cold path: Expected topic named 'race-topic-duplicate', found '{}'",
+ topic.name
+ );
+ }
+ }
+}
+
+async fn cleanup_resources(client: &IggyClient, resource_type: ResourceType) {
+ match resource_type {
+ ResourceType::User => {
+ let users = client.get_users().await.unwrap();
+ for user in users {
+ if user.username.starts_with("race-user-") {
+ let _ = client
+ .delete_user(&Identifier::numeric(user.id).unwrap())
+ .await;
+ }
+ }
+ }
+ ResourceType::Stream => {
+ let streams = client.get_streams().await.unwrap();
+ for stream in streams {
+ let _ = client
+ .delete_stream(&Identifier::numeric(stream.id).unwrap())
+ .await;
+ }
+ }
+ // Just delete test stream
+ ResourceType::Topic => {
+ let _ = client
+ .delete_stream(&Identifier::named(TEST_STREAM_NAME).unwrap())
+ .await;
+ }
+ }
+}
diff --git a/core/integration/tests/server/scenarios/mod.rs b/core/integration/tests/server/scenarios/mod.rs
index 0e2aa594a..c1e14af37 100644
--- a/core/integration/tests/server/scenarios/mod.rs
+++ b/core/integration/tests/server/scenarios/mod.rs
@@ -17,6 +17,7 @@
*/
pub mod bench_scenario;
+pub mod concurrent_scenario;
pub mod consumer_group_join_scenario;
pub mod consumer_group_with_multiple_clients_polling_messages_scenario;
pub mod consumer_group_with_single_client_polling_messages_scenario;
diff --git a/core/integration/tests/server/scenarios/stream_size_validation_scenario.rs b/core/integration/tests/server/scenarios/stream_size_validation_scenario.rs
index 17e5bc870..c816b2dac 100644
--- a/core/integration/tests/server/scenarios/stream_size_validation_scenario.rs
+++ b/core/integration/tests/server/scenarios/stream_size_validation_scenario.rs
@@ -20,7 +20,7 @@ use crate::server::scenarios::{PARTITION_ID, PARTITIONS_COUNT, create_client};
use bytes::Bytes;
use iggy::prelude::*;
use integration::test_server::{ClientFactory, assert_clean_system, login_root};
-use std::{str::FromStr, time::Duration};
+use std::str::FromStr;
const S1_NAME: &str = "test-stream-1";
const T1_NAME: &str = "test-topic-1";
diff --git a/core/integration/tests/server/scenarios/user_scenario.rs b/core/integration/tests/server/scenarios/user_scenario.rs
index 1903c9326..bfcdfb1c3 100644
--- a/core/integration/tests/server/scenarios/user_scenario.rs
+++ b/core/integration/tests/server/scenarios/user_scenario.rs
@@ -37,7 +37,7 @@ pub async fn run(client_factory: &dyn ClientFactory) {
// 3. Login as root user
let identity_info = login_root(&client).await;
- assert_eq!(identity_info.user_id, 1);
+ assert_eq!(identity_info.user_id, 0);
// 4. List all users
let users = client.get_users().await.unwrap();
@@ -45,7 +45,7 @@ pub async fn run(client_factory: &dyn ClientFactory) {
// 5. Get user details
let user = users.first().unwrap();
- assert_eq!(user.id, 1);
+ assert_eq!(user.id, 0);
assert!(user.created_at.as_micros() > 0);
assert_eq!(user.username, DEFAULT_ROOT_USERNAME);
assert_eq!(user.status, UserStatus::Active);
@@ -56,7 +56,7 @@ pub async fn run(client_factory: &dyn ClientFactory) {
.unwrap()
.expect("Failed to get user");
- assert_eq!(user.id, 1);
+ assert_eq!(user.id, 0);
assert!(user.created_at.as_micros() > 0);
assert_eq!(user.username, DEFAULT_ROOT_USERNAME);
assert_eq!(user.status, UserStatus::Active);
@@ -102,7 +102,7 @@ pub async fn run(client_factory: &dyn ClientFactory) {
// 9. Login as the new user
let identity_info = client.login_user(test_user, test_password).await.unwrap();
- assert_eq!(identity_info.user_id, 2);
+ assert_eq!(identity_info.user_id, 1);
// 10. Trying to create a new user (or any other resource) without the appropriate permissions should fail
let create_user_result = client
@@ -168,14 +168,14 @@ pub async fn run(client_factory: &dyn ClientFactory) {
.await
.unwrap();
- assert_eq!(identity_info.user_id, 2);
+ assert_eq!(identity_info.user_id, 1);
let identity_info = client
.login_with_personal_access_token(&raw_pat2.token)
.await
.unwrap();
- assert_eq!(identity_info.user_id, 2);
+ assert_eq!(identity_info.user_id, 1);
// 17. Delete the personal access tokens
client
diff --git a/core/integration/tests/state/file.rs b/core/integration/tests/state/file.rs
index b830cd53b..2b718d81f 100644
--- a/core/integration/tests/state/file.rs
+++ b/core/integration/tests/state/file.rs
@@ -41,9 +41,9 @@ async fn should_apply_single_entry() {
let state = setup.state();
state.init().await.unwrap();
- let user_id = 1;
+ let user_id = 0;
let command = EntryCommand::CreateUser(CreateUserWithId {
- user_id,
+ user_id: 1,
command: CreateUser {
username: "test".to_string(),
password: "secret".to_string(),
@@ -67,9 +67,9 @@ async fn should_apply_encrypted_entry() {
let state = setup.state();
state.init().await.unwrap();
- let user_id = 1;
+ let user_id = 0;
let command = EntryCommand::CreateUser(CreateUserWithId {
- user_id,
+ user_id: 1,
command: CreateUser {
username: "test".to_string(),
password: "secret".to_string(),
@@ -98,9 +98,10 @@ async fn should_apply_multiple_entries() {
assert_eq!(state.entries_count(), 0);
assert_eq!(state.term(), 0);
- let first_user_id = 1;
+ let first_user_id = 0; // Root user
+ let created_user_id = 1; // First created user
let create_user = EntryCommand::CreateUser(CreateUserWithId {
- user_id: first_user_id,
+ user_id: created_user_id,
command: CreateUser {
username: "test".to_string(),
password: "secret".to_string(),
@@ -115,7 +116,7 @@ async fn should_apply_multiple_entries() {
assert_eq!(state.current_index(), 0);
assert_eq!(state.entries_count(), 1);
- let second_user_id = 2;
+ let second_user_id = 1;
let stream_id = 1;
let create_stream = EntryCommand::CreateStream(CreateStreamWithId {
stream_id,
diff --git a/core/integration/tests/state/system.rs b/core/integration/tests/state/system.rs
index dce78aea1..77eeb5851 100644
--- a/core/integration/tests/state/system.rs
+++ b/core/integration/tests/state/system.rs
@@ -39,7 +39,7 @@ async fn should_be_initialized_based_on_state_entries() {
let state = setup.state();
state.init().await.unwrap();
- let user_id = 1;
+ let user_id = 0;
let create_user = CreateUser {
username: "user".to_string(),
password: "secret".to_string(),
diff --git a/core/sdk/src/prelude.rs b/core/sdk/src/prelude.rs
index 5f4d03ce0..d89afe86a 100644
--- a/core/sdk/src/prelude.rs
+++ b/core/sdk/src/prelude.rs
@@ -64,8 +64,8 @@ pub use iggy_common::{
Sizeable, SnapshotCompression, Stats, Stream, StreamDetails, StreamPermissions,
SystemSnapshotType, TcpClientConfig, TcpClientConfigBuilder, TcpClientReconnectionConfig,
Topic, TopicDetails, TopicPermissions, TransportProtocol, UserId, UserStatus, Validatable,
- defaults, locking, WebSocketClientConfig,
- WebSocketClientConfigBuilder, WebSocketClientReconnectionConfig
+ WebSocketClientConfig, WebSocketClientConfigBuilder, WebSocketClientReconnectionConfig,
+ defaults, locking,
};
pub use iggy_common::{
IGGY_MESSAGE_CHECKSUM_OFFSET_RANGE, IGGY_MESSAGE_HEADER_SIZE,
diff --git a/core/server/src/binary/mapper.rs b/core/server/src/binary/mapper.rs
index 1d6fea71b..3e02ad98f 100644
--- a/core/server/src/binary/mapper.rs
+++ b/core/server/src/binary/mapper.rs
@@ -293,7 +293,7 @@ fn extend_consumer_group(
fn extend_client(client: &Client, bytes: &mut BytesMut) {
bytes.put_u32_le(client.session.client_id);
- bytes.put_u32_le(client.user_id.unwrap_or(0));
+ bytes.put_u32_le(client.user_id.unwrap_or(u32::MAX));
let transport: u8 = match client.transport {
TransportProtocol::Tcp => 1,
TransportProtocol::Quic => 2,
diff --git a/core/server/src/bootstrap.rs b/core/server/src/bootstrap.rs
index 5a150ab16..1004a7643 100644
--- a/core/server/src/bootstrap.rs
+++ b/core/server/src/bootstrap.rs
@@ -21,6 +21,7 @@ use crate::{
traits_ext::{
EntityComponentSystem, EntityComponentSystemMutCell, Insert, InsertCell, IntoComponents,
},
+ users::Users,
},
state::system::{StreamState, TopicState, UserState},
streaming::{
@@ -45,7 +46,7 @@ use ahash::HashMap;
use compio::{fs::create_dir_all, runtime::Runtime};
use error_set::ErrContext;
use iggy_common::{
- ConsumerKind, IggyError, UserId,
+ ConsumerKind, IggyError,
defaults::{
DEFAULT_ROOT_PASSWORD, DEFAULT_ROOT_USERNAME, MAX_PASSWORD_LENGTH, MAX_USERNAME_LENGTH,
MIN_PASSWORD_LENGTH, MIN_USERNAME_LENGTH,
@@ -198,8 +199,8 @@ pub async fn load_streams(
Ok(streams)
}
-pub fn load_users(state: impl IntoIterator<Item = UserState>) -> HashMap<UserId, User> {
- let mut users = HashMap::default();
+pub fn load_users(state: impl IntoIterator<Item = UserState>) -> Users {
+ let users = Users::new();
for user_state in state {
let UserState {
id,
@@ -217,16 +218,11 @@ pub fn load_users(state: impl IntoIterator<Item = UserState>) -> HashMap<UserId,
.map(|token| {
(
Arc::new(token.token_hash.clone()),
- PersonalAccessToken::raw(
- user_state.id,
- &token.name,
- &token.token_hash,
- token.expiry_at,
- ),
+ PersonalAccessToken::raw(id, &token.name, &token.token_hash, token.expiry_at),
)
})
.collect();
- users.insert(id, user);
+ users.insert(user);
}
users
}
diff --git a/core/server/src/http/consumer_groups.rs b/core/server/src/http/consumer_groups.rs
index 16c6ac91c..b69c62ba2 100644
--- a/core/server/src/http/consumer_groups.rs
+++ b/core/server/src/http/consumer_groups.rs
@@ -257,12 +257,13 @@ async fn delete_consumer_group(
// Get members from the deleted consumer group and make them leave
let slab = consumer_group.members().inner().shared_get();
for (_, member) in slab.iter() {
- if let Err(err) = state.shard.shard().client_manager.borrow_mut().leave_consumer_group(
- member.client_id,
- stream_id_usize,
- topic_id_usize,
- cg_id,
- ) {
+ if let Err(err) = state
+ .shard
+ .shard()
+ .client_manager
+ .borrow_mut()
+ .leave_consumer_group(member.client_id, stream_id_usize, topic_id_usize, cg_id)
+ {
tracing::warn!(
"{COMPONENT} (error: {err}) - failed to make client leave consumer group for client ID: {}, group ID: {}",
member.client_id,
diff --git a/core/server/src/http/streams.rs b/core/server/src/http/streams.rs
index 51aad3ee5..cc71fa230 100644
--- a/core/server/src/http/streams.rs
+++ b/core/server/src/http/streams.rs
@@ -228,15 +228,16 @@ async fn delete_stream(
// Delete stream and get the stream entity
let stream = {
- let future = SendWrapper::new(state.shard.shard().delete_stream2(
- &session,
- &identifier_stream_id,
- ));
+ let future = SendWrapper::new(
+ state
+ .shard
+ .shard()
+ .delete_stream2(&session, &identifier_stream_id),
+ );
future.await
- }.with_error_context(|error| {
- format!(
- "{COMPONENT} (error: {error}) - failed to delete stream with ID: {stream_id}",
- )
+ }
+ .with_error_context(|error| {
+ format!("{COMPONENT} (error: {error}) - failed to delete stream with ID: {stream_id}",)
})?;
let stream_id_numeric = stream.root().id();
diff --git a/core/server/src/shard/builder.rs b/core/server/src/shard/builder.rs
index b3b5d5348..86abdeab1 100644
--- a/core/server/src/shard/builder.rs
+++ b/core/server/src/shard/builder.rs
@@ -23,19 +23,14 @@ use super::{
use crate::{
configs::server::ServerConfig,
shard::{Shard, ShardInfo, namespace::IggyNamespace},
- slab::streams::Streams,
+ slab::{streams::Streams, users::Users},
state::StateKind,
- streaming::{diagnostics::metrics::Metrics, users::user::User, utils::ptr::EternalPtr},
+ streaming::{diagnostics::metrics::Metrics, utils::ptr::EternalPtr},
versioning::SemanticVersion,
};
-use ahash::HashMap;
use dashmap::DashMap;
-use iggy_common::{EncryptorKind, UserId};
-use std::{
- cell::{Cell, RefCell},
- rc::Rc,
- sync::atomic::AtomicBool,
-};
+use iggy_common::EncryptorKind;
+use std::{cell::Cell, rc::Rc, sync::atomic::AtomicBool};
#[derive(Default)]
pub struct IggyShardBuilder {
@@ -43,7 +38,7 @@ pub struct IggyShardBuilder {
streams: Option<Streams>,
shards_table: Option<EternalPtr<DashMap<IggyNamespace, ShardInfo>>>,
state: Option<StateKind>,
- users: Option<HashMap<UserId, User>>,
+ users: Option<Users>,
connections: Option<Vec<ShardConnector<ShardFrame>>>,
config: Option<ServerConfig>,
encryptor: Option<EncryptorKind>,
@@ -95,7 +90,7 @@ impl IggyShardBuilder {
self
}
- pub fn users(mut self, users: HashMap<UserId, User>) -> Self {
+ pub fn users(mut self, users: Users) -> Self {
self.users = Some(users);
self
}
@@ -147,7 +142,7 @@ impl IggyShardBuilder {
shards,
shards_table,
streams2: streams, // TODO: Fixme
- users: RefCell::new(users),
+ users,
encryptor,
config,
_version: version,
diff --git a/core/server/src/shard/mod.rs b/core/server/src/shard/mod.rs
index a4aeaffcc..decfd8221 100644
--- a/core/server/src/shard/mod.rs
+++ b/core/server/src/shard/mod.rs
@@ -38,35 +38,29 @@ use crate::{
},
},
shard_error, shard_info,
- slab::{streams::Streams, traits_ext::EntityMarker},
+ slab::{streams::Streams, traits_ext::EntityMarker, users::Users},
state::StateKind,
streaming::{
- clients::client_manager::ClientManager,
- diagnostics::metrics::Metrics,
- partitions,
- polling_consumer::PollingConsumer,
- session::Session,
- traits::MainOps,
- users::{permissioner::Permissioner, user::User},
- utils::ptr::EternalPtr,
+ clients::client_manager::ClientManager, diagnostics::metrics::Metrics, partitions,
+ polling_consumer::PollingConsumer, session::Session, traits::MainOps,
+ users::permissioner::Permissioner, utils::ptr::EternalPtr,
},
versioning::SemanticVersion,
};
-use ahash::HashMap;
use builder::IggyShardBuilder;
use compio::io::AsyncWriteAtExt;
use dashmap::DashMap;
use error_set::ErrContext;
use hash32::{Hasher, Murmur3Hasher};
use iggy_common::{
- EncryptorKind, Identifier, IggyError, IggyTimestamp, PollingKind, TransportProtocol, UserId,
+ EncryptorKind, Identifier, IggyError, IggyTimestamp, PollingKind, TransportProtocol,
};
use std::hash::Hasher as _;
use std::{
cell::{Cell, RefCell},
net::SocketAddr,
rc::Rc,
- sync::atomic::{AtomicBool, AtomicU32, Ordering},
+ sync::atomic::{AtomicBool, Ordering},
time::{Duration, Instant},
};
use tracing::{debug, error, instrument, trace};
@@ -75,8 +69,6 @@ use transmission::connector::{Receiver, ShardConnector, StopReceiver};
pub const COMPONENT: &str = "SHARD";
pub const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(10);
-static USER_ID: AtomicU32 = AtomicU32::new(1);
-
pub(crate) struct Shard {
id: u16,
connection: ShardConnector<ShardFrame>,
@@ -138,7 +130,7 @@ pub struct IggyShard {
pub(crate) client_manager: RefCell<ClientManager>,
pub(crate) active_sessions: RefCell<Vec<Rc<Session>>>,
pub(crate) permissioner: RefCell<Permissioner>,
- pub(crate) users: RefCell<HashMap<UserId, User>>,
+ pub(crate) users: Users,
pub(crate) metrics: Metrics,
pub messages_receiver: Cell<Option<Receiver<ShardFrame>>>,
pub(crate) stop_receiver: StopReceiver,
@@ -428,13 +420,11 @@ impl IggyShard {
}
async fn load_users(&self) -> Result<(), IggyError> {
- let users = self.users.borrow();
- let users_count = users.len();
- let current_user_id = users.keys().max().unwrap_or(&1);
- USER_ID.store(current_user_id + 1, Ordering::SeqCst);
+ let users_list = self.users.values();
+ let users_count = users_list.len();
self.permissioner
.borrow_mut()
- .init(&users.values().collect::<Vec<_>>());
+ .init(&users_list.iter().collect::<Vec<_>>());
self.metrics.increment_users(users_count as u32);
shard_info!(self.id, "Initialized {} user(s).", users_count);
Ok(())
diff --git a/core/server/src/shard/system/personal_access_tokens.rs b/core/server/src/shard/system/personal_access_tokens.rs
index 6eb6a0cd3..47ad7856d 100644
--- a/core/server/src/shard/system/personal_access_tokens.rs
+++ b/core/server/src/shard/system/personal_access_tokens.rs
@@ -99,26 +99,26 @@ impl IggyShard {
let name = personal_access_token.name.clone();
let token_hash = personal_access_token.token.clone();
let identifier = user_id.try_into()?;
- let user = self.get_user_mut(&identifier).with_error_context(|error| {
- format!("{COMPONENT} create PAT (error: {error}) - failed to get mutable reference to the user with id: {user_id}")
- })?;
-
- if user
- .personal_access_tokens
- .iter()
- .any(|pat| pat.name.as_str() == name.as_str())
- {
- error!("Personal access token: {name} for user with ID: {user_id} already exists.");
- return Err(IggyError::PersonalAccessTokenAlreadyExists(
- name.to_string(),
- user_id,
- ));
- }
+ self.users.with_user_mut(&identifier, |user| {
+ if user
+ .personal_access_tokens
+ .iter()
+ .any(|pat| pat.name.as_str() == name.as_str())
+ {
+ error!("Personal access token: {name} for user with ID: {user_id} already exists.");
+ return Err(IggyError::PersonalAccessTokenAlreadyExists(
+ name.to_string(),
+ user_id,
+ ));
+ }
- info!("Creating personal access token: {name} for user with ID: {user_id}...");
- user.personal_access_tokens
- .insert(token_hash, personal_access_token);
- info!("Created personal access token: {name} for user with ID: {user_id}.");
+ user.personal_access_tokens
+ .insert(token_hash, personal_access_token);
+ info!("Created personal access token: {name} for user with ID: {user_id}.");
+ Ok(())
+ }).with_error_context(|error| {
+ format!("{COMPONENT} create PAT (error: {error}) - failed to access user with id: {user_id}")
+ })??;
Ok(())
}
@@ -141,27 +141,26 @@ impl IggyShard {
}
fn delete_personal_access_token_base(&self, user_id: u32, name: &str) -> Result<(), IggyError> {
- let user = self
- .get_user_mut(&user_id.try_into()?)
- .with_error_context(|error| {
- format!(
- "{COMPONENT} delete PAT (error: {error}) - failed to get mutable reference to the user with id: {user_id}"
- )
- })?;
+ self.users.with_user_mut(&user_id.try_into()?, |user| {
+ let token = if let Some(pat) = user
+ .personal_access_tokens
+ .iter()
+ .find(|pat| pat.name.as_str() == name)
+ {
+ pat.token.clone()
+ } else {
+ error!("Personal access token: {name} for user with ID: {user_id} does not exist.",);
+ return Err(IggyError::ResourceNotFound(name.to_owned()));
+ };
- let token = if let Some(pat) = user
- .personal_access_tokens
- .iter()
- .find(|pat| pat.name.as_str() == name)
- {
- pat.token.clone()
- } else {
- error!("Personal access token: {name} for user with ID: {user_id} does not exist.",);
- return Err(IggyError::ResourceNotFound(name.to_owned()));
- };
-
- info!("Deleting personal access token: {name} for user with ID: {user_id}...");
- user.personal_access_tokens.remove(&token);
+ info!("Deleting personal access token: {name} for user with ID: {user_id}...");
+ user.personal_access_tokens.remove(&token);
+ Ok(())
+ }).with_error_context(|error| {
+ format!(
+ "{COMPONENT} delete PAT (error: {error}) - failed to access user with id: {user_id}"
+ )
+ })??;
info!("Deleted personal access token: {name} for user with ID: {user_id}.");
Ok(())
}
@@ -182,9 +181,9 @@ impl IggyShard {
session: Option<&Session>,
) -> Result<User, IggyError> {
let token_hash = PersonalAccessToken::hash_token(token);
- let users = self.users.borrow();
+ let users = self.users.values();
let mut personal_access_token = None;
- for user in users.values() {
+ for user in &users {
if let Some(pat) = user.personal_access_tokens.get(&token_hash) {
personal_access_token = Some(pat);
break;
diff --git a/core/server/src/shard/system/segments.rs b/core/server/src/shard/system/segments.rs
index 055e0d76f..62d3ef0fa 100644
--- a/core/server/src/shard/system/segments.rs
+++ b/core/server/src/shard/system/segments.rs
@@ -44,7 +44,7 @@ impl IggyShard {
stream_id,
topic_id,
partition_id,
- |(_, stats,.., log)| {
+ |(_, stats, .., log)| {
let upperbound = log.segments().len();
let begin = upperbound.saturating_sub(segments_count as usize);
let segments = log
diff --git a/core/server/src/shard/system/streams.rs b/core/server/src/shard/system/streams.rs
index 4929d676e..1623d402a 100644
--- a/core/server/src/shard/system/streams.rs
+++ b/core/server/src/shard/system/streams.rs
@@ -148,14 +148,15 @@ impl IggyShard {
})?;
let mut stream = self.delete_stream2_base(id);
let stream_id_usize = stream.id();
-
+
// Clean up consumer groups from ClientManager for this stream
self.client_manager
.borrow_mut()
.delete_consumer_groups_for_stream(stream_id_usize);
-
+
// Remove all entries from shards_table for this stream (all topics and partitions)
- let namespaces_to_remove: Vec<_> = self.shards_table
+ let namespaces_to_remove: Vec<_> = self
+ .shards_table
.iter()
.filter_map(|entry| {
let (ns, _) = entry.pair();
@@ -166,11 +167,11 @@ impl IggyShard {
}
})
.collect();
-
+
for ns in namespaces_to_remove {
self.remove_shard_table_record(&ns);
}
-
+
delete_stream_from_disk(self.id, &mut stream, &self.config.system).await?;
Ok(stream)
}
@@ -229,13 +230,13 @@ mod tests {
use crate::shard::ShardInfo;
use crate::shard::transmission::connector::ShardConnector;
use crate::slab::streams::Streams;
+ use crate::slab::users::Users;
use crate::state::{MockState, StateKind};
use crate::streaming::session::Session;
use crate::streaming::streams;
use crate::streaming::users::user::User;
use crate::streaming::utils::ptr::EternalPtr;
use crate::versioning::SemanticVersion;
- use ahash::HashMap;
use dashmap::DashMap;
use iggy_common::defaults::{DEFAULT_ROOT_PASSWORD, DEFAULT_ROOT_USERNAME};
use std::net::{Ipv4Addr, SocketAddr};
@@ -250,11 +251,8 @@ mod tests {
let shards_table = Box::leak(shards_table);
let shards_table: EternalPtr<DashMap<crate::shard::namespace::IggyNamespace, ShardInfo>> =
shards_table.into();
-
- let users = HashMap::default();
+ let users = Users::new();
let state = StateKind::Mock(MockState::new());
-
- // Create single shard connection
let connections = vec![ShardConnector::new(0, 1)];
let builder = IggyShard::builder();
@@ -281,13 +279,12 @@ mod tests {
// Initialize root user and session
let root = User::root(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD);
+ let root_id = shard.users.insert(root);
let session = Session::new(
1,
- root.id,
+ root_id as u32,
SocketAddr::new(Ipv4Addr::LOCALHOST.into(), 1234),
);
-
- shard.users.borrow_mut().insert(root.id, root);
shard
.permissioner
.borrow_mut()
@@ -340,13 +337,12 @@ mod tests {
// Initialize root user and session
let root = User::root(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD);
+ let root_id = shard.users.insert(root);
let session = Session::new(
1,
- root.id,
+ root_id as u32,
SocketAddr::new(Ipv4Addr::LOCALHOST.into(), 1234),
);
-
- shard.users.borrow_mut().insert(root.id, root);
shard
.permissioner
.borrow_mut()
@@ -399,13 +395,12 @@ mod tests {
// Initialize root user and session
let root = User::root(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD);
+ let root_id = shard.users.insert(root);
let session = Session::new(
1,
- root.id,
+ root_id as u32,
SocketAddr::new(Ipv4Addr::LOCALHOST.into(), 1234),
);
-
- shard.users.borrow_mut().insert(root.id, root);
shard
.permissioner
.borrow_mut()
diff --git a/core/server/src/shard/system/topics.rs b/core/server/src/shard/system/topics.rs
index bffd78e41..3cc90c941 100644
--- a/core/server/src/shard/system/topics.rs
+++ b/core/server/src/shard/system/topics.rs
@@ -219,14 +219,15 @@ impl IggyShard {
})?;
let mut topic = self.delete_topic_base2(stream_id, topic_id);
let topic_id_numeric = topic.id();
-
+
// Clean up consumer groups from ClientManager for this topic
self.client_manager
.borrow_mut()
.delete_consumer_groups_for_topic(numeric_stream_id, topic_id_numeric);
-
+
// Remove all partition entries from shards_table for this topic
- let namespaces_to_remove: Vec<_> = self.shards_table
+ let namespaces_to_remove: Vec<_> = self
+ .shards_table
.iter()
.filter_map(|entry| {
let (ns, _) = entry.pair();
@@ -237,11 +238,11 @@ impl IggyShard {
}
})
.collect();
-
+
for ns in namespaces_to_remove {
self.remove_shard_table_record(&ns);
}
-
+
let parent = topic.stats().parent().clone();
// We need to borrow topic as mutable, as we are extracting partitions out of it, in order to close them.
let (messages_count, size_bytes, segments_count) =
@@ -343,10 +344,18 @@ impl IggyShard {
topic_id,
partitions::helpers::purge_partitions_mem(),
);
+
for part_id in part_ids {
self.delete_segments_bypass_auth(stream_id, topic_id, part_id, u32::MAX)
.await?;
}
+
+ // Zero out topic stats after purging all partitions
+ self.streams2
+ .with_topic_by_id(stream_id, topic_id, |(_root, _aux, stats)| {
+ stats.zero_out_all();
+ });
+
Ok(())
}
}
diff --git a/core/server/src/shard/system/users.rs b/core/server/src/shard/system/users.rs
index 8aa7d9635..9ce5cb293 100644
--- a/core/server/src/shard/system/users.rs
+++ b/core/server/src/shard/system/users.rs
@@ -22,15 +22,12 @@ use crate::streaming::session::Session;
use crate::streaming::users::user::User;
use crate::streaming::utils::crypto;
use error_set::ErrContext;
+use iggy_common::Identifier;
use iggy_common::IggyError;
use iggy_common::Permissions;
use iggy_common::UserStatus;
-use iggy_common::{IdKind, Identifier};
-use std::cell::RefMut;
-use std::sync::atomic::{AtomicU32, Ordering};
use tracing::{error, warn};
-static USER_ID: AtomicU32 = AtomicU32::new(1);
const MAX_USERS: usize = u32::MAX as usize;
impl IggyShard {
@@ -62,54 +59,7 @@ impl IggyShard {
}
pub fn try_get_user(&self, user_id: &Identifier) -> Result<Option<User>, IggyError> {
- match user_id.kind {
- IdKind::Numeric => {
- let user = self.users.borrow().get(&user_id.get_u32_value()?).cloned();
- Ok(user)
- }
- IdKind::String => {
- let username = user_id.get_cow_str_value()?;
- let user = self
- .users
- .borrow()
- .iter()
- .find(|(_, user)| user.username == username)
- .map(|(_, user)| user.clone());
- Ok(user)
- }
- }
- }
-
- pub fn get_user_mut(&self, user_id: &Identifier) -> Result<RefMut<'_, User>, IggyError> {
- match user_id.kind {
- IdKind::Numeric => {
- let user_id = user_id.get_u32_value()?;
- let users = self.users.borrow_mut();
- let exists = users.contains_key(&user_id);
- if !exists {
- return Err(IggyError::ResourceNotFound(user_id.to_string()));
- }
- Ok(RefMut::map(users, |u| {
- let user = u.get_mut(&user_id);
- user.unwrap()
- }))
- }
- IdKind::String => {
- let username = user_id.get_cow_str_value()?;
- let users = self.users.borrow_mut();
- let exists = users.iter().any(|(_, user)| user.username == username);
- if !exists {
- return Err(IggyError::ResourceNotFound(user_id.to_string()));
- }
- Ok(RefMut::map(users, |u| {
- let user = u
- .iter_mut()
- .find(|(_, user)| user.username == username)
- .map(|(_, user)| user);
- user.unwrap()
- }))
- }
- }
+ self.users.get_by_identifier(user_id)
}
pub async fn get_users(&self, session: &Session) -> Result<Vec<User>, IggyError> {
@@ -123,7 +73,7 @@ impl IggyShard {
session.get_user_id()
)
})?;
- Ok(self.users.borrow().values().cloned().collect())
+ Ok(self.users.values())
}
pub fn create_user(
@@ -145,60 +95,49 @@ impl IggyShard {
)
})?;
- if self
- .users
- .borrow()
- .iter()
- .any(|(_, user)| user.username == username)
- {
+ if self.users.username_exists(username) {
error!("User: {username} already exists.");
return Err(IggyError::UserAlreadyExists);
}
- if self.users.borrow().len() >= MAX_USERS {
+ if self.users.len() >= MAX_USERS {
error!("Available users limit reached.");
return Err(IggyError::UsersLimitReached);
}
- // TODO: Tech debt, replace with Slab.
- USER_ID.fetch_add(1, Ordering::SeqCst);
- let current_user_id = USER_ID.load(Ordering::SeqCst);
- self.create_user_base(current_user_id, username, password, status, permissions)?;
- self.get_user(¤t_user_id.try_into()?)
+ let user_id = self.create_user_base(username, password, status, permissions)?;
+ self.get_user(&(user_id as u32).try_into()?)
.with_error_context(|error| {
- format!(
- "{COMPONENT} (error: {error}) - failed to get user with id: {current_user_id}"
- )
+ format!("{COMPONENT} (error: {error}) - failed to get user with id: {user_id}")
})
}
pub fn create_user_bypass_auth(
&self,
- user_id: u32,
+ _user_id: u32, // Ignored - Slab auto-assigns IDs
username: &str,
password: &str,
status: UserStatus,
permissions: Option<Permissions>,
) -> Result<(), IggyError> {
- self.create_user_base(user_id, username, password, status, permissions)?;
+ self.create_user_base(username, password, status, permissions)?;
Ok(())
}
fn create_user_base(
&self,
- user_id: u32,
username: &str,
password: &str,
status: UserStatus,
permissions: Option<Permissions>,
- ) -> Result<(), IggyError> {
- let user = User::new(user_id, username, password, status, permissions.clone());
+ ) -> Result<usize, IggyError> {
+ let user = User::new(0, username, password, status, permissions.clone());
+ let user_id = self.users.insert(user);
self.permissioner
.borrow_mut()
- .init_permissions_for_user(user_id, permissions);
- self.users.borrow_mut().insert(user.id, user);
+ .init_permissions_for_user(user_id as u32, permissions);
self.metrics.increment_users(1);
- Ok(())
+ Ok(user_id)
}
pub fn delete_user(&self, session: &Session, user_id: &Identifier) -> Result<User, IggyError> {
@@ -221,32 +160,30 @@ impl IggyShard {
}
fn delete_user_base(&self, user_id: &Identifier) -> Result<User, IggyError> {
- let existing_user_id;
- {
- let user = self.get_user(user_id).with_error_context(|error| {
- format!("{COMPONENT} (error: {error}) - failed to get user with id: {user_id}")
- })?;
- if user.is_root() {
- error!("Cannot delete the root user.");
- return Err(IggyError::CannotDeleteUser(user.id));
- }
+ let user = self.get_user(user_id).with_error_context(|error| {
+ format!("{COMPONENT} (error: {error}) - failed to get user with id: {user_id}")
+ })?;
- existing_user_id = user.id;
+ if user.is_root() {
+ error!("Cannot delete the root user.");
+ return Err(IggyError::CannotDeleteUser(user.id));
}
+ let user_slab_id = user.id as usize;
+ let user_u32_id = user.id;
+
let user = self
.users
- .borrow_mut()
- .remove(&existing_user_id)
+ .remove(user_slab_id)
.ok_or(IggyError::ResourceNotFound(user_id.to_string()))?;
self.permissioner
.borrow_mut()
- .delete_permissions_for_user(existing_user_id);
+ .delete_permissions_for_user(user_u32_id);
self.client_manager.borrow_mut()
- .delete_clients_for_user(existing_user_id)
+ .delete_clients_for_user(user_u32_id)
.with_error_context(|error| {
format!(
- "{COMPONENT} (error: {error}) - failed to delete clients for user with ID: {existing_user_id}"
+ "{COMPONENT} (error: {error}) - failed to delete clients for user with ID: {user_u32_id}"
)
})?;
self.metrics.decrement_users(1);
@@ -289,29 +226,26 @@ impl IggyShard {
username: Option<String>,
status: Option<UserStatus>,
) -> Result<User, IggyError> {
- if let Some(username) = username.to_owned() {
+ if let Some(ref new_username) = username {
let user = self.get_user(user_id)?;
- let existing_user = self.get_user(&username.to_owned().try_into()?);
+ let existing_user = self.get_user(&new_username.to_owned().try_into()?);
if existing_user.is_ok() && existing_user.unwrap().id != user.id {
- error!("User: {username} already exists.");
+ error!("User: {new_username} already exists.");
return Err(IggyError::UserAlreadyExists);
}
}
- let mut user = self.get_user_mut(user_id).with_error_context(|error| {
- format!("{COMPONENT} update user (error: {error}) - failed to get mutable reference to the user with id: {user_id}")
- })?;
- if let Some(username) = username {
- user.username = username;
- }
-
- if let Some(status) = status {
- user.status = status;
- }
- let cloned_user = user.clone();
- drop(user);
-
- Ok(cloned_user)
+ self.users.with_user_mut(user_id, |user| {
+ if let Some(username) = username {
+ user.username = username;
+ }
+ if let Some(status) = status {
+ user.status = status;
+ }
+ user.clone()
+ }).with_error_context(|error| {
+ format!("{COMPONENT} update user (error: {error}) - failed to update user with id: {user_id}")
+ })
}
pub fn update_permissions(
@@ -356,26 +290,21 @@ impl IggyShard {
user_id: &Identifier,
permissions: Option<Permissions>,
) -> Result<(), IggyError> {
- {
- let user: User = self.get_user(user_id).with_error_context(|error| {
- format!("{COMPONENT} (error: {error}) - failed to get user with id: {user_id}")
- })?;
+ let user: User = self.get_user(user_id).with_error_context(|error| {
+ format!("{COMPONENT} (error: {error}) - failed to get user with id: {user_id}")
+ })?;
- self.permissioner
- .borrow_mut()
- .update_permissions_for_user(user.id, permissions.clone());
- }
+ self.permissioner
+ .borrow_mut()
+ .update_permissions_for_user(user.id, permissions.clone());
- {
- let mut user = self.get_user_mut(user_id).with_error_context(|error| {
- format!(
- "{COMPONENT} update user permissions (error: {error}) - failed to get mutable reference to the user with id: {user_id}"
- )
- })?;
+ self.users.with_user_mut(user_id, |user| {
user.permissions = permissions;
- }
-
- Ok(())
+ }).with_error_context(|error| {
+ format!(
+ "{COMPONENT} update user permissions (error: {error}) - failed to update permissions for user with id: {user_id}"
+ )
+ })
}
pub fn change_password(
@@ -417,19 +346,19 @@ impl IggyShard {
current_password: &str,
new_password: &str,
) -> Result<(), IggyError> {
- let mut user = self.get_user_mut(user_id).with_error_context(|error| {
- format!("{COMPONENT} change password (error: {error}) - failed to get mutable reference to the user with id: {user_id}")
- })?;
- if !crypto::verify_password(current_password, &user.password) {
- error!(
- "Invalid current password for user: {} with ID: {user_id}.",
- user.username
- );
- return Err(IggyError::InvalidCredentials);
- }
-
- user.password = crypto::hash_password(new_password);
- Ok(())
+ self.users.with_user_mut(user_id, |user| {
+ if !crypto::verify_password(current_password, &user.password) {
+ error!(
+ "Invalid current password for user: {} with ID: {user_id}.",
+ user.username
+ );
+ return Err(IggyError::InvalidCredentials);
+ }
+ user.password = crypto::hash_password(new_password);
+ Ok(())
+ }).with_error_context(|error| {
+ format!("{COMPONENT} change password (error: {error}) - failed to change password for user with id: {user_id}")
+ })?
}
pub fn login_user_event(
diff --git a/core/server/src/shard/tasks/periodic/personal_access_token_cleaner.rs b/core/server/src/shard/tasks/periodic/personal_access_token_cleaner.rs
index 670b04ef9..3c1203a9a 100644
--- a/core/server/src/shard/tasks/periodic/personal_access_token_cleaner.rs
+++ b/core/server/src/shard/tasks/periodic/personal_access_token_cleaner.rs
@@ -48,8 +48,8 @@ async fn clear_personal_access_tokens(shard: Rc<IggyShard>) -> Result<(), IggyEr
let now = IggyTimestamp::now();
let mut total_removed = 0;
- let users = shard.users.borrow();
- for user in users.values() {
+ let users = shard.users.values();
+ for user in &users {
let expired_tokens: Vec<Arc<String>> = user
.personal_access_tokens
.iter()
diff --git a/core/server/src/slab/mod.rs b/core/server/src/slab/mod.rs
index 13c8f472f..0569f7749 100644
--- a/core/server/src/slab/mod.rs
+++ b/core/server/src/slab/mod.rs
@@ -4,6 +4,7 @@ pub mod partitions;
pub mod streams;
pub mod topics;
pub mod traits_ext;
+pub mod users;
use std::fmt::{Debug, Display};
diff --git a/core/server/src/slab/users.rs b/core/server/src/slab/users.rs
new file mode 100644
index 000000000..8a6949fa0
--- /dev/null
+++ b/core/server/src/slab/users.rs
@@ -0,0 +1,140 @@
+use crate::streaming::users::user::User;
+use ahash::AHashMap;
+use iggy_common::{Identifier, IggyError};
+use slab::Slab;
+use std::cell::RefCell;
+
+const CAPACITY: usize = 1024;
+
+#[derive(Debug, Clone, Default)]
+pub struct Users {
+ index: RefCell<AHashMap<String, usize>>,
+ users: RefCell<Slab<User>>,
+}
+
+impl Users {
+ pub fn new() -> Self {
+ Self {
+ index: RefCell::new(AHashMap::with_capacity(CAPACITY)),
+ users: RefCell::new(Slab::with_capacity(CAPACITY)),
+ }
+ }
+
+ /// Insert a user and return the assigned ID (auto-incremented by Slab)
+ pub fn insert(&self, user: User) -> usize {
+ let username = user.username.clone();
+ let mut users = self.users.borrow_mut();
+ let mut index = self.index.borrow_mut();
+
+ let id = users.insert(user);
+ users[id].id = id as u32;
+ index.insert(username, id);
+ id
+ }
+
+ /// Get user by ID (numeric identifier)
+ pub fn get(&self, id: usize) -> Option<User> {
+ self.users.borrow().get(id).cloned()
+ }
+
+ /// Get user by username or ID (via Identifier enum)
+ pub fn get_by_identifier(&self, identifier: &Identifier) -> Result<Option<User>, IggyError> {
+ match identifier.kind {
+ iggy_common::IdKind::Numeric => {
+ let id = identifier.get_u32_value()? as usize;
+ Ok(self.users.borrow().get(id).cloned())
+ }
+ iggy_common::IdKind::String => {
+ let username = identifier.get_string_value()?;
+ let index = self.index.borrow();
+ if let Some(&id) = index.get(&username) {
+ Ok(self.users.borrow().get(id).cloned())
+ } else {
+ Ok(None)
+ }
+ }
+ }
+ }
+
+ /// Remove user by Slab index
+ pub fn remove(&self, id: usize) -> Option<User> {
+ let mut users = self.users.borrow_mut();
+ let mut index = self.index.borrow_mut();
+
+ if !users.contains(id) {
+ return None;
+ }
+
+ let user = users.remove(id);
+ index.remove(&user.username);
+ Some(user)
+ }
+
+ /// Check if user exists
+ pub fn contains(&self, identifier: &Identifier) -> bool {
+ match identifier.kind {
+ iggy_common::IdKind::Numeric => {
+ if let Ok(id) = identifier.get_u32_value() {
+ self.users.borrow().contains(id as usize)
+ } else {
+ false
+ }
+ }
+ iggy_common::IdKind::String => {
+ if let Ok(username) = identifier.get_string_value() {
+ self.index.borrow().contains_key(&username)
+ } else {
+ false
+ }
+ }
+ }
+ }
+
+ /// Get all users as a Vec
+ pub fn values(&self) -> Vec<User> {
+ self.users.borrow().iter().map(|(_, u)| u.clone()).collect()
+ }
+
+ /// Get number of users
+ pub fn len(&self) -> usize {
+ self.users.borrow().len()
+ }
+
+ /// Check if empty
+ pub fn is_empty(&self) -> bool {
+ self.users.borrow().is_empty()
+ }
+
+ /// Get mutable access to a user via closure (to avoid RefMut escaping issues)
+ pub fn with_user_mut<F, R>(&self, identifier: &Identifier, f: F) -> Result<R, IggyError>
+ where
+ F: FnOnce(&mut User) -> R,
+ {
+ let id = match identifier.kind {
+ iggy_common::IdKind::Numeric => identifier.get_u32_value()? as usize,
+ iggy_common::IdKind::String => {
+ let username = identifier.get_string_value()?;
+ let index = self.index.borrow();
+ *index
+ .get(&username)
+ .ok_or_else(|| IggyError::ResourceNotFound(username.to_string()))?
+ }
+ };
+
+ let mut users = self.users.borrow_mut();
+ let user = users
+ .get_mut(id)
+ .ok_or_else(|| IggyError::ResourceNotFound(identifier.to_string()))?;
+ Ok(f(user))
+ }
+
+ /// Check if username already exists (for validation during create/update)
+ pub fn username_exists(&self, username: &str) -> bool {
+ self.index.borrow().contains_key(username)
+ }
+
+ /// Get ID by username (useful for cross-references)
+ pub fn get_id_by_username(&self, username: &str) -> Option<usize> {
+ self.index.borrow().get(username).copied()
+ }
+}
diff --git a/core/server/src/streaming/session.rs b/core/server/src/streaming/session.rs
index c34e39d49..1afbd724d 100644
--- a/core/server/src/streaming/session.rs
+++ b/core/server/src/streaming/session.rs
@@ -45,7 +45,7 @@ impl Session {
}
pub fn from_client_id(client_id: u32, ip_address: SocketAddr) -> Self {
- Self::new(client_id, 0, ip_address)
+ Self::new(client_id, u32::MAX, ip_address)
}
pub fn get_user_id(&self) -> UserId {
@@ -61,7 +61,7 @@ impl Session {
}
pub fn clear_user_id(&self) {
- self.set_user_id(0);
+ self.set_user_id(u32::MAX);
}
pub fn is_active(&self) -> bool {
@@ -69,14 +69,14 @@ impl Session {
}
pub fn is_authenticated(&self) -> bool {
- self.get_user_id() > 0
+ self.get_user_id() != u32::MAX
}
}
impl Display for Session {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let user_id = self.get_user_id();
- if user_id > 0 {
+ if user_id != u32::MAX {
write!(
f,
"client ID: {}, user ID: {}, IP address: {}",
diff --git a/core/server/src/streaming/users/user.rs b/core/server/src/streaming/users/user.rs
index 7e68d98ff..437ce955b 100644
--- a/core/server/src/streaming/users/user.rs
+++ b/core/server/src/streaming/users/user.rs
@@ -38,7 +38,7 @@ pub struct User {
impl Default for User {
fn default() -> Self {
Self {
- id: 1,
+ id: 0,
status: UserStatus::Active,
username: "user".to_string(),
password: "secret".to_string(),
diff --git a/examples/rust/src/multi-tenant/consumer/main.rs b/examples/rust/src/multi-tenant/consumer/main.rs
index 1b08a5716..198a52063 100644
--- a/examples/rust/src/multi-tenant/consumer/main.rs
+++ b/examples/rust/src/multi-tenant/consumer/main.rs
@@ -201,7 +201,7 @@ async fn create_user(
.expect("Topic does not exist");
topic_permissions.insert(
- topic.id,
+ topic.id as usize,
TopicPermissions {
read_topic: true,
poll_messages: true,
@@ -212,7 +212,7 @@ async fn create_user(
let mut streams_permissions = AHashMap::new();
streams_permissions.insert(
- stream.id,
+ stream.id as usize,
StreamPermissions {
read_stream: true,
topics: Some(topic_permissions),
diff --git a/examples/rust/src/multi-tenant/producer/main.rs b/examples/rust/src/multi-tenant/producer/main.rs
index ec46086ce..59a0f3cd2 100644
--- a/examples/rust/src/multi-tenant/producer/main.rs
+++ b/examples/rust/src/multi-tenant/producer/main.rs
@@ -331,7 +331,7 @@ async fn create_stream_and_user(
info!("Created stream: {stream_name} with ID: {}", stream.id);
let mut streams_permissions = AHashMap::new();
streams_permissions.insert(
- stream.id,
+ stream.id as usize,
StreamPermissions {
read_stream: true,
manage_topics: true,