diff --git a/Cargo.lock b/Cargo.lock index cda581ffb..82ffafc81 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -964,9 +964,9 @@ dependencies = [ [[package]] name = "aws-sdk-kms" -version = "1.114.0" +version = "1.115.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c0b7d906608ee41e7ddea9983577ba82200435644d567d63dc34e822e088b453" +checksum = "d5b034f8b7ceadb873d0bc607c30bb4b0be68e09a84c837174e7c2c6878ff882" dependencies = [ "arc-swap", "aws-credential-types", @@ -990,9 +990,9 @@ dependencies = [ [[package]] name = "aws-sdk-s3" -version = "1.141.0" +version = "1.142.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d9f9420d3a2467eed22ed3635ca653653162c386a0b0f65c78189f9bd3c1379e" +checksum = "f9e15a5c55e05f4b0b7e483160b3c85cccdf77cff02c95504f3e71d460855cd2" dependencies = [ "arc-swap", "aws-credential-types", @@ -1027,9 +1027,9 @@ dependencies = [ [[package]] name = "aws-sdk-sso" -version = "1.105.0" +version = "1.106.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6ffd0fbe7873cb548a7aa60f9573c268fff94155397fd4f14dc9f1ecaaab8516" +checksum = "2d0efcee834347b6705eca3eea2defd88242f43774f55d7326604222e3c86260" dependencies = [ "arc-swap", "aws-credential-types", @@ -1053,9 +1053,9 @@ dependencies = [ [[package]] name = "aws-sdk-ssooidc" -version = "1.107.0" +version = "1.108.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "175763eb222a46377df7aa257a3bca980ab3e96703fefc8f4d0b8da6ad2e254c" +checksum = "a59312a04cf19c962cfee32b64ecfee758f8786407ff6da5b30fff46ae96f201" dependencies = [ "arc-swap", "aws-credential-types", @@ -1079,9 +1079,9 @@ dependencies = [ [[package]] name = "aws-sdk-sts" -version = "1.110.0" +version = "1.111.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "dd8b14781dfbff48984017d57167b6ea0b6471c6920ec52b44a2677c7feb3c13" +checksum = "120e7eb63457a9e547f9986fe3b273f77c43679da4d04f46359fa881c5e19b6e" dependencies = [ "arc-swap", "aws-credential-types", @@ -1598,7 +1598,7 @@ version = "0.10.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3078c7629b62d3f0439517fa394996acacc5cbc91c5a20d8c658e77abd503a71" dependencies = [ - "generic-array 0.14.9", + "generic-array 0.14.7", ] [[package]] @@ -1617,7 +1617,7 @@ version = "0.3.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a8894febbff9f758034a5b8e12d87918f56dfc64a8e1fe757d65e29041538d93" dependencies = [ - "generic-array 0.14.9", + "generic-array 0.14.7", ] [[package]] @@ -1858,9 +1858,9 @@ dependencies = [ [[package]] name = "cc" -version = "1.4.2" +version = "1.4.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5d262e149917187838d5b42777c8253bcb64500067342904e7d429499a6f277e" +checksum = "509591b7bcd67f4ef775afad7662703b4935daaa6ec0e5605cfb1090b32a2b6d" dependencies = [ "find-msvc-tools", "jobserver", @@ -1968,7 +1968,7 @@ version = "0.4.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "773f3b9af64447d2ce9850330c473515014aa235e6a783b02db81ff39e4a3dad" dependencies = [ - "crypto-common 0.1.6", + "crypto-common 0.1.7", "inout 0.1.4", ] @@ -2428,7 +2428,7 @@ version = "0.5.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0dc92fb57ca44df6db8059111ab3af99a63d5d0f8375d9972e319a379c6bab76" dependencies = [ - "generic-array 0.14.9", + "generic-array 0.14.7", "rand_core 0.6.4", "subtle", "zeroize", @@ -2453,11 +2453,11 @@ dependencies = [ [[package]] name = "crypto-common" -version = "0.1.6" +version = "0.1.7" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1bfb12502f3fc46cca1bb51ac28df9d618d813cdc3d2f25b9fe775a34af26bb3" +checksum = "78c8292055d1c1df0cce5d180393dc8cce0abec0a7102adb6c7b1eef6016d60a" dependencies = [ - "generic-array 0.14.9", + "generic-array 0.14.7", "typenum", ] @@ -3664,7 +3664,7 @@ checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292" dependencies = [ "block-buffer 0.10.4", "const-oid 0.9.6", - "crypto-common 0.1.6", + "crypto-common 0.1.7", "subtle", ] @@ -3924,7 +3924,7 @@ dependencies = [ "crypto-bigint 0.5.5", "digest 0.10.7", "ff 0.13.1", - "generic-array 0.14.9", + "generic-array 0.14.7", "group 0.13.0", "hkdf 0.12.4", "pem-rfc7468 0.7.0", @@ -4148,9 +4148,9 @@ dependencies = [ [[package]] name = "find-msvc-tools" -version = "0.1.10" +version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "26b73573e6edcd2af0cdf47bd6cb58f0b3839491263c314eaad1ccf24430e1de" +checksum = "d45db016d36b838f563236e9193d0ee6ce38f3f68b6c94e914b4929c96bbb890" [[package]] name = "findshlibs" @@ -4369,9 +4369,9 @@ dependencies = [ [[package]] name = "generic-array" -version = "0.14.9" +version = "0.14.7" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4bb6743198531e02858aeaea5398fcc883e71851fcbcb5a2f773e2fb6cb1edf2" +checksum = "85649ca51fd72272d7821adaf274ad91c288277713d9c18820d8499a7ff69e9a" dependencies = [ "typenum", "version_check", @@ -4380,11 +4380,11 @@ dependencies = [ [[package]] name = "generic-array" -version = "1.4.4" +version = "1.4.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ab4e5aa225bc56696909483320f0ff9b600f1a971b52e07a17d70f3d9b43254b" +checksum = "337d46834ee672ab3e48caca2cb0c78cc174fb12b3a68d0d88f99a0519a5e36e" dependencies = [ - "generic-array 0.14.9", + "generic-array 0.14.7", "rustversion", "typenum", ] @@ -4726,9 +4726,9 @@ dependencies = [ [[package]] name = "h2" -version = "0.4.15" +version = "0.4.16" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6cb093c84e8bd9b188d4c4a8cb6579fc016968d14c99882163cd3ff402a4f155" +checksum = "a9f37a958b41b3b19ee2707c06439c0e9e547e847223eb791ecb0cb821c65e27" dependencies = [ "atomic-waker", "bytes", @@ -5028,9 +5028,9 @@ dependencies = [ [[package]] name = "hotpath" -version = "0.23.2" +version = "0.23.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "62e810bedda5a467ef5c9b5c8a20763fefebc89b63ef36f7ee44a143085204a2" +checksum = "dce755d457a63bdd0c95e4c91511daad1b58b33209543b7f38027b676f387e5e" dependencies = [ "arc-swap", "async-channel", @@ -5062,9 +5062,9 @@ dependencies = [ [[package]] name = "hotpath-macros" -version = "0.23.2" +version = "0.23.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "01bdc59bfc1a9984bee2ff5da63b2f6fccbaa57cd9a4119d709524632bddf341" +checksum = "a903af89a8429cb07790c3818bc15270b394f80af1bc254e5ccf9c7de2961770" dependencies = [ "proc-macro2", "quote", @@ -5073,15 +5073,15 @@ dependencies = [ [[package]] name = "hotpath-macros-meta" -version = "0.23.2" +version = "0.23.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d9216e8a01abe1e1671c376dc8736fb1bf772d7a889538d25f9e1200120ced38" +checksum = "bcc0ab94ffbb2ee77f4a897df02b5a137a10cf24d69bda936e59aff4dd456e61" [[package]] name = "hotpath-meta" -version = "0.23.2" +version = "0.23.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f22a9d20435fb79511b19dae37b3607224cd98f342a410702d84657cc38fc72f" +checksum = "053481f6cec8f775a3276c7f6e2f21123111d28261e4edc15ea7421c445964bb" dependencies = [ "hotpath-macros-meta", ] @@ -5280,9 +5280,9 @@ dependencies = [ [[package]] name = "icu_collections" -version = "2.2.0" +version = "2.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2984d1cd16c883d7935b9e07e44071dca8d917fd52ecc02c04d5fa0b5a3f191c" +checksum = "fa68d21081c4a05d5a901a1c62add574c77048b6a1c67be3b50ce0b60d4ca513" dependencies = [ "displaydoc", "potential_utf", @@ -5294,9 +5294,9 @@ dependencies = [ [[package]] name = "icu_locale_core" -version = "2.2.0" +version = "2.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "92219b62b3e2b4d88ac5119f8904c10f8f61bf7e95b640d25ba3075e6cac2c29" +checksum = "d56e28588da92eee5c3201a6eff33fabdd49b62269c8938d4ff050ce4d900deb" dependencies = [ "displaydoc", "litemap", @@ -5307,9 +5307,9 @@ dependencies = [ [[package]] name = "icu_normalizer" -version = "2.2.0" +version = "2.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c56e5ee99d6e3d33bd91c5d85458b6005a22140021cc324cea84dd0e72cff3b4" +checksum = "12f9cf5f235641ed274641dd81c3f28d870e276763d0797aeeab72317b1c646f" dependencies = [ "icu_collections", "icu_normalizer_data", @@ -5321,16 +5321,17 @@ dependencies = [ [[package]] name = "icu_normalizer_data" -version = "2.2.0" +version = "2.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "da3be0ae77ea334f4da67c12f149704f19f81d1adf7c51cf482943e84a2bad38" +checksum = "1563da1ed3e0b3bf3d74c9b85917ac9c56464d2f57242270c09c9e752f8021a0" [[package]] name = "icu_properties" -version = "2.2.0" +version = "2.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bee3b67d0ea5c2cca5003417989af8996f8604e34fb9ddf96208a033901e70de" +checksum = "7e7ca276ad3145661a65914e6daf131ca5120cd3dcee8f8f3214b8875184a148" dependencies = [ + "displaydoc", "icu_collections", "icu_locale_core", "icu_properties_data", @@ -5341,15 +5342,15 @@ dependencies = [ [[package]] name = "icu_properties_data" -version = "2.2.0" +version = "2.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8e2bbb201e0c04f7b4b3e14382af113e17ba4f63e2c9d2ee626b720cbce54a14" +checksum = "e590f038c1464a96894fd6d10127e90a8be4509f56ff7ecef851b15cee0b7caa" [[package]] name = "icu_provider" -version = "2.2.0" +version = "2.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "139c4cf31c8b5f33d7e199446eff9c1e02decfc2f0eec2c8d71f65befa45b421" +checksum = "92a7ed671a6aad807a8651a2e1782a6598fda9ce5185dd8158549e95a91c6428" dependencies = [ "displaydoc", "icu_locale_core", @@ -5417,7 +5418,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "879f10e63c20629ecabbb64a8010319738c66a5cd0c29b02d63d272b03751d01" dependencies = [ "block-padding 0.3.3", - "generic-array 0.14.9", + "generic-array 0.14.7", ] [[package]] @@ -5968,9 +5969,9 @@ dependencies = [ [[package]] name = "libredox" -version = "0.1.19" +version = "0.1.20" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2026a5056764a10b2bf5d56488cba40da507f5493a6a429340e2004d9ed085fa" +checksum = "28d0a00925a9f930d679b6789b721e3a7f9ed110f41b86d2497caa780c3a070a" dependencies = [ "libc", ] @@ -6033,9 +6034,9 @@ checksum = "32a66949e030da00e8c7d4434b251670a91556f4144941d37452769c25d58a53" [[package]] name = "litemap" -version = "0.8.2" +version = "0.8.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "92daf443525c4cce67b150400bc2316076100ce0b3686209eb8cf3c31612e6f0" +checksum = "47d9d19d1d6efa0109d2f65ff4c85cddd50bd572e5a00127ab10987290bcefae" [[package]] name = "local-ip-address" @@ -6482,9 +6483,9 @@ dependencies = [ [[package]] name = "mqttbytes-core-next" -version = "0.33.3" +version = "0.34.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3ff7ae19c74aba9e0ed6e4071cd52aa364e020076fa3cc6ef17e43662f756f3c" +checksum = "366b6ba2b4209ca4bc5ac731ccddf570d09831981eed07e5fbd63564cf0cf1aa" dependencies = [ "bytes", "thiserror 2.0.20", @@ -6885,7 +6886,7 @@ version = "5.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "51e219e79014df21a225b1860a479e2dcd7cbd9130f4defd4bd0e191ea31d67d" dependencies = [ - "base64 0.21.7", + "base64 0.22.1", "chrono", "getrandom 0.2.17", "http 1.5.0", @@ -7344,9 +7345,9 @@ dependencies = [ [[package]] name = "pageant" -version = "0.2.1" +version = "0.2.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4f3a5ae18f65a85c67a77d18d42d3606c07948e3c17c1e5f74852b26589e88a5" +checksum = "3adadc44070da6f464b0918655a12f5792c156e088d8c4082d13e27d94c3e791" dependencies = [ "base16ct 1.0.0", "byteorder", @@ -7728,9 +7729,9 @@ dependencies = [ [[package]] name = "pkg-config" -version = "0.3.33" +version = "0.3.34" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "19f132c84eca552bf34cab8ec81f1c1dcc229b811638f9d283dceabe58c5569e" +checksum = "f6b464fbc74e149a392436b17d523f769e057cb6877f6a5c4618bc6f11800548" [[package]] name = "plotters" @@ -7836,9 +7837,9 @@ dependencies = [ [[package]] name = "potential_utf" -version = "0.1.5" +version = "0.1.6" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0103b1cef7ec0cf76490e969665504990193874ea05c85ff9bab8b911d0a0564" +checksum = "d83eb9bc6d8e5cf568e7a1101d60ee05e81ed50ea106026f3d18deeb046d7661" dependencies = [ "zerovec", ] @@ -8046,7 +8047,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "be769465445e8c1474e9c5dac2018218498557af32d9ed057325ec9a41ae81bf" dependencies = [ "heck 0.5.0", - "itertools 0.10.5", + "itertools 0.14.0", "log", "multimap", "once_cell", @@ -8066,7 +8067,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "03da047801ff44bb6a4d407d4860c05fd70bb81714e6b2f3812603d5b145b042" dependencies = [ "heck 0.5.0", - "itertools 0.10.5", + "itertools 0.14.0", "log", "multimap", "petgraph 0.8.3", @@ -8087,7 +8088,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8a56d757972c98b346a9b766e3f02746cde6dd1cd1d1d563472929fdd74bec4d" dependencies = [ "anyhow", - "itertools 0.10.5", + "itertools 0.14.0", "proc-macro2", "quote", "syn 2.0.119", @@ -8100,7 +8101,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b570b25f7617e43d59005d0990ccb79e950a423952cea19671b7a876da390adf" dependencies = [ "anyhow", - "itertools 0.10.5", + "itertools 0.14.0", "proc-macro2", "quote", "syn 2.0.119", @@ -8216,7 +8217,7 @@ dependencies = [ "reqwest", "serde_json", "smallvec", - "spin 0.12.2", + "spin 0.12.3", "symbolic-demangle", "tempfile", "thiserror 2.0.20", @@ -8285,9 +8286,9 @@ dependencies = [ [[package]] name = "quinn-proto" -version = "0.11.16" +version = "0.11.17" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2f4bfc015262b9df63c8845072ce59068853ff5872180c2ce2f13038b970e560" +checksum = "04759210543be93709136e28212294a659ef5001836ff4eab4d663e4529bba83" dependencies = [ "aws-lc-rs", "bytes", @@ -8553,9 +8554,9 @@ dependencies = [ [[package]] name = "redis" -version = "1.5.0" +version = "1.6.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3257df217f7eab0044627a268c9cc6cdb60c0c421c88f83ac41c4e31520b6b84" +checksum = "e37a4ca5c6ca42aa3e6df2fd32b987a65d32a4c2159a6f3fe0fd1df306a2658f" dependencies = [ "arc-swap", "arcstr", @@ -8567,7 +8568,7 @@ dependencies = [ "futures-channel", "futures-util", "itoa", - "num-bigint 0.4.8", + "num-bigint 0.5.1", "percent-encoding", "pin-project-lite", "rustls", @@ -8868,9 +8869,9 @@ dependencies = [ [[package]] name = "rumqttc-core-next" -version = "0.33.3" +version = "0.34.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7d7d9205738dd41a2546e82d27a634d07d8b303dcf7558565ff70caf3ceb0f9c" +checksum = "249896ab27ed630590971738264baa8f722f18965d2e387c706c40a3c2a572cc" dependencies = [ "async-tungstenite", "futures-io", @@ -8886,18 +8887,18 @@ dependencies = [ [[package]] name = "rumqttc-next" -version = "0.33.3" +version = "0.34.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ed1bad2180ff539da671da9a996152a921bc5316eb6d8a9cc3bd441653138b08" +checksum = "477c9bbfba8f3aecc7aad31c6de2eacb75822efaa18e7aeecb8d3d8e534fbf07" dependencies = [ "rumqttc-v5-next", ] [[package]] name = "rumqttc-v5-next" -version = "0.33.3" +version = "0.34.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "229576cbedfa9089f90c17c9454e9429ac1e89cdd223bac5cb39d837593f79bc" +checksum = "3dfa6ddcc7a7dd5688f9bf78d8f81cb94f367bce56c055d8d94cf81ecb0518bf" dependencies = [ "async-tungstenite", "bytes", @@ -8920,9 +8921,9 @@ dependencies = [ [[package]] name = "russh" -version = "0.62.6" +version = "0.62.7" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b41043523e0edcbd4e31d00903e26f12994f63b21bae9904f7405c1ed92752a5" +checksum = "9decb68e4e44e1079700e54f17c8f23806ec53d7e0db73ab1c71d9dabc666812" dependencies = [ "aes 0.9.2", "aws-lc-rs", @@ -8945,7 +8946,7 @@ dependencies = [ "enum_dispatch", "flate2", "futures", - "generic-array 1.4.4", + "generic-array 1.4.5", "getrandom 0.4.3", "ghash", "hex-literal", @@ -9491,7 +9492,7 @@ dependencies = [ "parking_lot", "rayon", "smallvec", - "spin 0.12.2", + "spin 0.12.3", ] [[package]] @@ -10833,7 +10834,7 @@ checksum = "d3e97a565f76233a6003f9f5c54be1d9c5bdfa3eccfb189469f11ec4901c47dc" dependencies = [ "base16ct 0.2.0", "der 0.7.10", - "generic-array 0.14.9", + "generic-array 0.14.7", "pkcs8 0.10.2", "subtle", "zeroize", @@ -11397,9 +11398,9 @@ checksum = "023a211cb3138dbc438680b32560ad89f699977624c9f8dbb95a47d5b4c07dd3" [[package]] name = "spin" -version = "0.12.2" +version = "0.12.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8abadc99fd9c7bbb7d0ca2b31d72a067d0c0dcd7aad25ab8cac71ba91417694b" +checksum = "0134f9043ed38b087ac4f7d4af44c79e2c9e5094421fe3164f435ce585953b10" dependencies = [ "lock_api", ] @@ -11811,7 +11812,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" dependencies = [ "fastrand", - "getrandom 0.3.4", + "getrandom 0.4.3", "once_cell", "rustix", "windows-sys 0.61.2", @@ -11975,9 +11976,9 @@ dependencies = [ [[package]] name = "tinystr" -version = "0.8.3" +version = "0.8.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c8323304221c2a851516f22236c5722a72eaa19749016521d6dff0824447d96d" +checksum = "b1e27c91459209c2986af3dcf603a5a74a4368754ce37414f59acc971167f643" dependencies = [ "displaydoc", "zerovec", @@ -12651,9 +12652,9 @@ checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821" [[package]] name = "uuid" -version = "1.24.0" +version = "1.24.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bf3923a6f5c4c6382e0b653c4117f48d631ea17f38ed86e2a828e6f7412f5239" +checksum = "2cefc03fd367c0c6d4305de1b312cf00248c4114f4a0418ce6a6af769e3b0bd9" dependencies = [ "getrandom 0.4.3", "js-sys", @@ -13168,9 +13169,9 @@ dependencies = [ [[package]] name = "writeable" -version = "0.6.3" +version = "0.6.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1ffae5123b2d3fc086436f8834ae3ab053a283cfac8fe0a0b8eaae044768a4c4" +checksum = "3ad82d2a33cdc9674dc7465672f271e096168fcdbe0f799d9e6db8c5892679dc" [[package]] name = "x509-cert" @@ -13350,9 +13351,9 @@ dependencies = [ [[package]] name = "zerotrie" -version = "0.2.4" +version = "0.2.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0f9152d31db0792fa83f70fb2f83148effb5c1f5b8c7686c3459e361d9bc20bf" +checksum = "4ea269c3bd32f0a32c321907a2ae912ba6f4649bb0fc764a15627e99a7095a3f" dependencies = [ "displaydoc", "yoke", @@ -13361,9 +13362,9 @@ dependencies = [ [[package]] name = "zerovec" -version = "0.11.6" +version = "0.11.7" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "90f911cbc359ab6af17377d242225f4d75119aec87ea711a880987b18cd7b239" +checksum = "94b5c6b5976d66c1d703c4fd17d3f5e43c8cedaacf604961b171adc7130896d8" dependencies = [ "yoke", "zerofrom", @@ -13372,13 +13373,13 @@ dependencies = [ [[package]] name = "zerovec-derive" -version = "0.11.3" +version = "0.11.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "625dc425cab0dca6dc3c3319506e6593dcb08a9f387ea3b284dbd52a92c40555" +checksum = "9f212a141d820099d57ffafb9569be9617a6f27d3dc881fbee8fb56642f917a9" dependencies = [ "proc-macro2", "quote", - "syn 2.0.119", + "syn 3.0.3", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 6bbd3acd1..ae9b497e0 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -228,9 +228,9 @@ atoi = "3.1.0" atomic_enum = "0.3.0" aws-config = { version = "1.10.1" } aws-credential-types = { version = "1.3.0" } -aws-sdk-kms = { default-features = false, version = "1.114.0" } -aws-sdk-s3 = { default-features = false, version = "1.141.0" } -aws-sdk-sts = { default-features = false, version = "1.110.0" } +aws-sdk-kms = { default-features = false, version = "1.115.0" } +aws-sdk-s3 = { default-features = false, version = "1.142.0" } +aws-sdk-sts = { default-features = false, version = "1.111.0" } aws-smithy-http-client = { default-features = false, version = "1.3.0" } aws-smithy-runtime-api = { version = "1.14.0" } aws-smithy-types = { version = "1.6.2" } @@ -284,8 +284,8 @@ rayon = "1.12.0" reed-solomon-erasure = { package = "rustfs-erasure-codec", version = "8.0.2" } reed-solomon-simd = "3.1.0" regex = { version = "1.13.1" } -rumqttc = { package = "rumqttc-next", version = "0.33.3" } -redis = { version = "1.5.0" } +rumqttc = { package = "rumqttc-next", version = "0.34.0" } +redis = { version = "1.6.0" } rustify = { version = "0.7", default-features = false } rustix = { version = "1.1.4" } rust-embed = { version = "8.12.0" } @@ -313,7 +313,7 @@ tracing-subscriber = { version = "0.3.23" } transform-stream = "0.3.1" url = "2.5.8" urlencoding = "2.1.3" -uuid = { version = "1.24.0" } +uuid = { version = "1.24.1" } vaultrs = { version = "0.8.0" } tar = "0.4.46" walkdir = "2.5.0" @@ -341,7 +341,7 @@ libunftp = { version = "0.23.0" } unftp-core = "0.1.0" suppaftp = { version = "10.0.1" } rcgen = { version = "0.14.9", default-features = false, features = ["aws_lc_rs", "crypto", "pem"] } -russh = { version = "0.62.6" } +russh = { version = "0.62.7" } russh-sftp = "2.4.0" # WebDAV @@ -350,7 +350,7 @@ dav-server = "0.11.0" # Performance Analysis and Memory Profiling mimalloc = { version = "0.1.52", git = "https://github.com/xonatius/mimalloc_rust.git", rev = "6d4c41bb10c6d9da1d1b6f07b38c4cc051667f11" } libmimalloc-sys = { version = "0.1.49", git = "https://github.com/xonatius/mimalloc_rust.git", rev = "6d4c41bb10c6d9da1d1b6f07b38c4cc051667f11", features = ["extended"] } -hotpath = { version = "0.23.2", default-features = false } +hotpath = { version = "0.23.3", default-features = false } # Snapshot testing for output format regression detection insta = { version = "1.48" } diff --git a/crates/common/src/heal_channel.rs b/crates/common/src/heal_channel.rs index f0ea0530e..b5a907e4e 100644 --- a/crates/common/src/heal_channel.rs +++ b/crates/common/src/heal_channel.rs @@ -287,6 +287,9 @@ pub enum HealRequestSource { Scanner, AutoHeal, ReadRepair, + /// Mission Repair Feed: intents delivered by error paths and replayed + /// from the durable MRF journal. + Mrf, } impl HealRequestSource { @@ -297,6 +300,7 @@ impl HealRequestSource { Self::Scanner => "scanner", Self::AutoHeal => "auto_heal", Self::ReadRepair => "read_repair", + Self::Mrf => "mrf", } } } diff --git a/crates/common/src/lib.rs b/crates/common/src/lib.rs index 1ae200f24..68f5f5c68 100644 --- a/crates/common/src/lib.rs +++ b/crates/common/src/lib.rs @@ -17,6 +17,7 @@ pub mod globals; pub mod heal_channel; pub mod last_minute; pub mod metrics; +pub mod mrf_channel; mod readiness; pub mod table_catalog; pub mod trace_bus; diff --git a/crates/common/src/mrf_channel.rs b/crates/common/src/mrf_channel.rs new file mode 100644 index 000000000..f0a91a238 --- /dev/null +++ b/crates/common/src/mrf_channel.rs @@ -0,0 +1,203 @@ +// Copyright 2024 RustFS Team +// +// Licensed 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. + +//! Mission Repair Feed (MRF) intent channel. +//! +//! Producers on error paths (read decode failure, scanner metadata +//! corruption, partial-write recovery) hand a lightweight [`MrfIntent`] to the +//! heal crate through a global bounded channel. Delivery is strictly +//! non-blocking: `try_send_mrf_intent` never awaits and drops the intent +//! (counting it) when the channel is full or uninitialized — losing one heal +//! hint is always preferred over stalling an IO path. Durable replay of +//! unconsumed intents is the consumer's job (see `rustfs-heal` +//! `heal::mrf_queue`), mirroring MinIO's `.heal/mrf/list.bin`. + +use std::sync::{ + Arc, OnceLock, + atomic::{AtomicBool, Ordering}, +}; +use tokio::sync::mpsc; +use uuid::Uuid; + +/// Bounded capacity of the global MRF channel. Backpressure is resolved by +/// dropping (and counting) intents, never by blocking the producer. +const MRF_CHANNEL_CAPACITY: usize = 8192; + +/// Why an intent was produced. Drives the heal priority mapping on the +/// consumer side (DecodeFailure -> Urgent, MetadataCorruption -> High, +/// PartialWrite -> Normal). +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum MrfKind { + /// Erasure decode failed while serving a read (read path). + DecodeFailure, + /// Scanner classified object metadata as corrupt. + MetadataCorruption, + /// A write left the object with fewer committed shards than the set size. + PartialWrite, +} + +impl MrfKind { + pub const fn as_str(self) -> &'static str { + match self { + MrfKind::DecodeFailure => "decode-failure", + MrfKind::MetadataCorruption => "metadata-corruption", + MrfKind::PartialWrite => "partial-write", + } + } +} + +/// One repair intent. Kept deliberately small so the in-memory queue and the +/// journal stay bounded; `bucket`/`object` are `Arc` so re-arming an +/// intent never re-allocates the strings. +#[derive(Clone, Debug)] +pub struct MrfIntent { + pub bucket: Arc, + pub object: Arc, + /// Version the intent targets, as raw UUID bytes. + pub version_id: Option<[u8; 16]>, + pub kind: MrfKind, + pub enqueued_at_ms: u64, + /// Times this intent has already been offered to the heal manager. + /// Dropped by the consumer once it reaches `MRF_MAX_ATTEMPTS`. + pub attempts: u8, +} + +/// Consumer-side retry ceiling before an intent is given up on. +pub const MRF_MAX_ATTEMPTS: u8 = 3; + +impl MrfIntent { + /// Rough in-memory footprint used by the queue's byte budget. + pub fn estimated_bytes(&self) -> usize { + // Struct + strings + version bytes; buckets and objects are usually + // far below this bound, so rounding up keeps the budget conservative. + 64 + self.bucket.len() + self.object.len() + } +} + +static GLOBAL_MRF_SENDER: OnceLock> = OnceLock::new(); + +/// Delivery kill-switch, set from `RUSTFS_HEAL_MRF_ENABLE`. Producers check +/// this before touching the channel so the disabled path stays allocation- and +/// sync-free. +static MRF_DELIVERY_ENABLED: AtomicBool = AtomicBool::new(true); + +/// Override delivery (used at heal-runtime startup from configuration). +pub fn set_mrf_delivery_enabled(enabled: bool) { + MRF_DELIVERY_ENABLED.store(enabled, Ordering::Relaxed); +} + +/// Whether producers currently deliver intents. +pub fn mrf_delivery_enabled() -> bool { + MRF_DELIVERY_ENABLED.load(Ordering::Relaxed) +} + +/// Create the global MRF channel and return the consumer half. Fails if the +/// channel is already initialized (the heal runtime is a singleton). +pub fn init_mrf_channel() -> Result, &'static str> { + let (sender, receiver) = mpsc::channel(MRF_CHANNEL_CAPACITY); + GLOBAL_MRF_SENDER + .set(sender) + .map_err(|_| "MRF channel sender already initialized")?; + Ok(receiver) +} + +/// Best-effort, non-blocking intent delivery from an error path. +/// +/// Returns `true` when the intent was accepted into the channel. `false` +/// means the intent was dropped (feature disabled, channel not yet +/// initialized, or channel full) — callers must not retry or await; the +/// existing read-repair / scanner heal paths remain the safety net. +/// +/// This runs on IO error paths, so it stays synchronous and cheap: one +/// bounded allocation for the two `Arc` handles plus the channel slot. +pub fn try_send_mrf_intent(kind: MrfKind, bucket: &str, object: &str, version_id: Option) -> bool { + if !mrf_delivery_enabled() { + return false; + } + let Some(sender) = GLOBAL_MRF_SENDER.get() else { + return false; + }; + let intent = MrfIntent { + bucket: Arc::from(bucket), + object: Arc::from(object), + version_id: version_id.map(|vid| *vid.as_bytes()), + kind, + enqueued_at_ms: unix_now_ms(), + attempts: 0, + }; + sender.try_send(intent).is_ok() +} + +fn unix_now_ms() -> u64 { + // Kept trivial: the timestamp is diagnostic metadata only; wall-clock + // failure would be a bug rather than something to handle here. + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn intents_estimate_is_conservative() { + let intent = MrfIntent { + bucket: Arc::from("bucket"), + object: Arc::from("object"), + version_id: Some([0u8; 16]), + kind: MrfKind::DecodeFailure, + enqueued_at_ms: 0, + attempts: 0, + }; + assert!(intent.estimated_bytes() >= intent.bucket.len() + intent.object.len()); + } + + #[tokio::test] + async fn try_send_delivers_and_respects_capacity() { + let mut receiver = init_mrf_channel().expect("first initialization should succeed"); + assert!(init_mrf_channel().is_err(), "double initialization must fail"); + + assert!(try_send_mrf_intent(MrfKind::DecodeFailure, "b", "o", Some(Uuid::nil()))); + let intent = receiver.recv().await.expect("intent should arrive"); + assert_eq!(intent.kind, MrfKind::DecodeFailure); + assert_eq!(intent.bucket.as_ref(), "b"); + + // Disable delivery: producers become no-ops. + set_mrf_delivery_enabled(false); + assert!(!try_send_mrf_intent(MrfKind::PartialWrite, "b", "o", None)); + set_mrf_delivery_enabled(true); + + // Fill the bounded channel past capacity: excess intents are dropped, + // never blocking. + let mut accepted = 0; + for _ in 0..(MRF_CHANNEL_CAPACITY + 64) { + if try_send_mrf_intent(MrfKind::PartialWrite, "b", "o", None) { + accepted += 1; + } + } + assert_eq!(accepted, MRF_CHANNEL_CAPACITY); + } + + #[test] + fn try_send_without_channel_is_false() { + // This test may run after the tokio test above in the same process; + // the singleton semantics make a clean "uninitialized" case hard, so + // assert the flag-off behavior only. + set_mrf_delivery_enabled(false); + assert!(!try_send_mrf_intent(MrfKind::MetadataCorruption, "b", "o", None)); + set_mrf_delivery_enabled(true); + } +} diff --git a/crates/config/src/constants/heal.rs b/crates/config/src/constants/heal.rs index 647ca8533..b8cf3630a 100644 --- a/crates/config/src/constants/heal.rs +++ b/crates/config/src/constants/heal.rs @@ -177,3 +177,31 @@ pub const DEFAULT_HEAL_MAINLINE_WRITE_UTILIZATION_HIGH_PERCENT: usize = 80; /// Default foreground pressure recheck delay for heal scheduler, in milliseconds. pub const DEFAULT_HEAL_MAINLINE_MAX_SLEEP_MS: u64 = 250; + +/// Environment variable that toggles the MRF (mission repair feed) intent +/// pipeline: error paths deliver repair intents to the heal runtime, and +/// unconsumed intents are replayed from the durable journal after a restart. +pub const ENV_HEAL_MRF_ENABLE: &str = "RUSTFS_HEAL_MRF_ENABLE"; + +/// Environment variable for the MRF in-memory queue capacity (intent count). +pub const ENV_HEAL_MRF_QUEUE_SIZE: &str = "RUSTFS_HEAL_MRF_QUEUE_SIZE"; + +/// Environment variable for the MRF journal byte budget. The journal is +/// compacted once its on-disk size crosses this bound. +pub const ENV_HEAL_MRF_JOURNAL_MAX_BYTES: &str = "RUSTFS_HEAL_MRF_JOURNAL_MAX_BYTES"; + +/// Environment variable for the MRF journal replay batch size (intents per +/// replay push round). +pub const ENV_HEAL_MRF_REPLAY_BATCH: &str = "RUSTFS_HEAL_MRF_REPLAY_BATCH"; + +/// Default behavior keeps the MRF intent pipeline enabled. +pub const DEFAULT_HEAL_MRF_ENABLE: bool = true; + +/// Default MRF queue capacity (matches MinIO's 100k MRF list ceiling). +pub const DEFAULT_HEAL_MRF_QUEUE_SIZE: usize = 100_000; + +/// Default MRF journal byte budget (8 MiB), mirroring the channel payload cap. +pub const DEFAULT_HEAL_MRF_JOURNAL_MAX_BYTES: usize = 8 * 1024 * 1024; + +/// Default MRF replay batch size. +pub const DEFAULT_HEAL_MRF_REPLAY_BATCH: usize = 256; diff --git a/crates/config/src/constants/object.rs b/crates/config/src/constants/object.rs index 081308754..874933aa6 100644 --- a/crates/config/src/constants/object.rs +++ b/crates/config/src/constants/object.rs @@ -234,6 +234,31 @@ pub const ENV_OBJECT_DISK_WRITE_ABSOLUTE_CAP: &str = "RUSTFS_OBJECT_DISK_WRITE_A /// Default absolute per-object erasure write cap in seconds (`0` = disabled). pub const DEFAULT_OBJECT_DISK_WRITE_ABSOLUTE_CAP: u64 = 0; +/// Enable foreground PutObject request admission. +/// +/// This is an experimental, default-off foreground write backpressure gate for +/// strict commit tail investigations. When disabled, PUTs follow the legacy +/// path and only the existing request counters are updated. +pub const ENV_PUT_FOREGROUND_ADMISSION_ENABLE: &str = "RUSTFS_PUT_FOREGROUND_ADMISSION_ENABLE"; +pub const DEFAULT_PUT_FOREGROUND_ADMISSION_ENABLE: bool = false; + +/// Maximum foreground PutObject requests admitted concurrently per process. +/// +/// The limit is used only when [`ENV_PUT_FOREGROUND_ADMISSION_ENABLE`] is true. +/// A value of `0` disables the gate even when the enable flag is present, so a +/// partially configured rollout cannot reject every PUT. +pub const ENV_PUT_FOREGROUND_ADMISSION_LIMIT: &str = "RUSTFS_PUT_FOREGROUND_ADMISSION_LIMIT"; +pub const DEFAULT_PUT_FOREGROUND_ADMISSION_LIMIT: usize = 0; + +/// Time in milliseconds a foreground PutObject waits for an admission permit. +/// +/// Once this timeout expires the request fails before body ingest/storage +/// mutation with S3 `SlowDown`/503. `0` means fail fast when the limit is full. +pub const ENV_PUT_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS: &str = "RUSTFS_PUT_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS"; +pub const DEFAULT_PUT_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS: u64 = 0; + +const _: () = assert!(!DEFAULT_PUT_FOREGROUND_ADMISSION_ENABLE); + /// Environment variable for minimum GetObject timeout in seconds. /// /// When dynamic timeout calculation is enabled, this is the minimum timeout diff --git a/crates/ecstore/src/set_disk/ops/heal.rs b/crates/ecstore/src/set_disk/ops/heal.rs index de688fbdb..7c228595d 100644 --- a/crates/ecstore/src/set_disk/ops/heal.rs +++ b/crates/ecstore/src/set_disk/ops/heal.rs @@ -3297,4 +3297,223 @@ mod heal_result_report_tests { assert!(result.detail.contains("part 1")); assert!(result.detail.contains("bitrot_failure=true")); } + + // HS-12 (backlog#1874): a versioned DELETE racing an object heal must never + // resurrect the deleted version. The heal has real reconstruction work (a + // shard of the doomed version is removed), so both sides touch the same + // (bucket, object, data_dir); whichever order the ns write lock serializes + // them in, the committed delete must win. + #[tokio::test] + #[serial_test::serial] + async fn heal_racing_version_delete_never_resurrects_the_deleted_version() { + let (temp_dirs, disks, set) = hermetic_set_disks_isolated(4).await; + let bucket = "heal-race-delete-no-resurrect"; + let object = "object.bin"; + set.make_bucket( + bucket, + &MakeBucketOptions { + versioning_enabled: true, + ..Default::default() + }, + ) + .await + .expect("versioned bucket should be created"); + + let mut first_reader = PutObjReader::from_vec(vec![0x11; 1024 * 1024]); + let first_info = set + .put_object( + bucket, + object, + &mut first_reader, + &ObjectOptions { + versioned: true, + ..Default::default() + }, + ) + .await + .expect("first version should be written"); + let first_version = first_info + .version_id + .expect("versioned put should return the first version id") + .to_string(); + + let mut second_reader = PutObjReader::from_vec(vec![0x22; 1024 * 1024]); + let second_info = set + .put_object( + bucket, + object, + &mut second_reader, + &ObjectOptions { + versioned: true, + ..Default::default() + }, + ) + .await + .expect("second version should be written"); + let second_version = second_info + .version_id + .expect("versioned put should return the second version id") + .to_string(); + + // Damage one shard of the doomed version so the racing heal performs an + // actual reconstruction over its data dir instead of an early exit. + let doomed_source = disks[0] + .read_version("", bucket, object, &first_version, &ReadOptions::default()) + .await + .expect("doomed version metadata should be readable"); + let doomed_data_dir = doomed_source + .data_dir + .expect("non-inline version should have a data directory"); + tokio::fs::remove_file( + temp_dirs[1] + .path() + .join(bucket) + .join(object) + .join(doomed_data_dir.to_string()) + .join("part.1"), + ) + .await + .expect("shard damage should be injected before the race"); + + let delete_set = set.clone(); + let (delete_res, heal_res) = tokio::join!( + async { + delete_set + .delete_object( + bucket, + object, + ObjectOptions { + versioned: true, + version_id: Some(first_version.clone()), + object_lock_config_snapshot: Some(Arc::new(crate::set_disk::ObjectLockConfigSnapshot::new( + crate::bucket::metadata_sys::ObjectLockConfigState::ConfirmedAbsent, + ))), + ..Default::default() + }, + ) + .await + }, + async { + set.heal_object( + bucket, + object, + "", + &HealOpts { + scan_mode: HealScanMode::Deep, + ..Default::default() + }, + ) + .await + }, + ); + delete_res.expect("version delete must succeed under lock serialization"); + // The heal may legitimately report a transient failure when the version + // it was rebuilding disappears mid-flight; only the end state matters. + drop(heal_res); + + let resurrected = set + .get_object_info( + bucket, + object, + &ObjectOptions { + versioned: true, + version_id: Some(first_version.clone()), + ..Default::default() + }, + ) + .await; + assert!( + matches!(&resurrected, Err(Error::FileVersionNotFound) | Err(Error::ObjectNotFound(..))), + "a racing heal must not resurrect the deleted version: {resurrected:?}" + ); + + let survivor = set + .get_object_info( + bucket, + object, + &ObjectOptions { + versioned: true, + version_id: Some(second_version.clone()), + ..Default::default() + }, + ) + .await + .expect("surviving version must remain readable after the race"); + assert_eq!(survivor.size, 1024 * 1024, "survivor size must be intact"); + } + + // HS-12 (backlog#1874): unversioned overwrite commits race a Deep heal on + // the same object. The overwrite's post-commit tail deletes the replaced + // data dir without the ns lock (object.rs commit tail), which is exactly + // the intersection the audit flagged: the heal must tolerate the tail race + // (retryable outcome) and every committed overwrite must survive — the + // final current version is exactly the last payload written. + #[tokio::test] + #[serial_test::serial] + async fn heal_racing_unversioned_overwrites_preserves_the_last_commit() { + let (temp_dirs, disks, set) = hermetic_set_disks_isolated(4).await; + let bucket = "heal-race-put-overwrite"; + let object = "object.bin"; + set.make_bucket(bucket, &MakeBucketOptions::default()) + .await + .expect("bucket should be created"); + + const ROUNDS: usize = 8; + const PAYLOAD_SIZE: usize = 256 * 1024; + let mut last_etag = String::new(); + for round in 0..ROUNDS { + // Give the heal something to rebuild on alternating rounds: remove a + // shard of the current data dir right before the race. + if round % 2 == 1 { + let current = disks[2] + .read_version("", bucket, object, "", &ReadOptions::default()) + .await + .expect("current metadata should be readable"); + if let Some(data_dir) = current.data_dir { + let shard = temp_dirs[3] + .path() + .join(bucket) + .join(object) + .join(data_dir.to_string()) + .join("part.1"); + if shard.exists() { + tokio::fs::remove_file(&shard) + .await + .expect("shard damage should be injectable mid-race"); + } + } + } + + let payload = vec![round as u8; PAYLOAD_SIZE]; + let mut put_reader = PutObjReader::from_vec(payload); + let put_opts = ObjectOptions::default(); + let heal_opts = HealOpts { + scan_mode: HealScanMode::Deep, + ..Default::default() + }; + let (put_res, heal_res) = tokio::join!( + set.put_object(bucket, object, &mut put_reader, &put_opts), + set.heal_object(bucket, object, "", &heal_opts), + ); + let put_info = put_res.expect("overwrite must succeed under lock serialization"); + last_etag = put_info.etag.clone().unwrap_or_default(); + // Heal outcome is unconstrained (may hit the tail race and report a + // retryable error); the invariant is checked on the end state. + drop(heal_res); + } + + let final_info = set + .get_object_info(bucket, object, &ObjectOptions::default()) + .await + .expect("object must remain readable after the race loop"); + assert_eq!( + final_info.size, PAYLOAD_SIZE as i64, + "final current version must be the last committed overwrite" + ); + assert_eq!( + final_info.etag.unwrap_or_default(), + last_etag, + "the racing heal loop must never leave a stale or resurrected current version" + ); + } } diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index e7498ee9d..929ac4d5c 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -5845,6 +5845,14 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { #[tracing::instrument(skip(self))] async fn add_partial(&self, bucket: &str, object: &str, version_id: &str) -> Result<()> { + // MRF journal intent: partial-write recovery must survive a restart + // (HS-01); the heal request below remains the in-memory fast path. + rustfs_common::mrf_channel::try_send_mrf_intent( + rustfs_common::mrf_channel::MrfKind::PartialWrite, + bucket, + object, + uuid::Uuid::try_parse(version_id).ok(), + ); let mut request = rustfs_common::heal_channel::create_heal_request_with_options( bucket.to_string(), Some(object.to_string()), diff --git a/crates/ecstore/src/set_disk/read.rs b/crates/ecstore/src/set_disk/read.rs index d59fa3e7e..07bbcf5f9 100644 --- a/crates/ecstore/src/set_disk/read.rs +++ b/crates/ecstore/src/set_disk/read.rs @@ -1077,6 +1077,15 @@ impl SetDisks { "Recoverable decode error triggered read repair" ); let version_id = fi.version_id.as_ref().map(ToString::to_string); + // MRF journal intent: keeps a durable Urgent ECDecode + // request alive across restarts even when the in-memory + // read-repair request is dropped or lost (HS-01). + rustfs_common::mrf_channel::try_send_mrf_intent( + rustfs_common::mrf_channel::MrfKind::DecodeFailure, + bucket, + object, + fi.version_id, + ); submit_read_repair_heal( bucket, object, diff --git a/crates/heal/Cargo.toml b/crates/heal/Cargo.toml index 668c4af24..4d4fb355b 100644 --- a/crates/heal/Cargo.toml +++ b/crates/heal/Cargo.toml @@ -89,6 +89,8 @@ async-trait = { workspace = true } futures = { workspace = true } metrics = { workspace = true } base64 = { workspace = true } +bytes = { workspace = true } +crc-fast = { workspace = true } [dev-dependencies] serde_json = { workspace = true, features = ["raw_value"] } diff --git a/crates/heal/src/heal/channel.rs b/crates/heal/src/heal/channel.rs index 23cf2f168..1e31056b2 100644 --- a/crates/heal/src/heal/channel.rs +++ b/crates/heal/src/heal/channel.rs @@ -612,7 +612,8 @@ impl HealChannelProcessor { HealRequestSource::Admin | HealRequestSource::AutoHeal | HealRequestSource::Internal - | HealRequestSource::ReadRepair => true, + | HealRequestSource::ReadRepair + | HealRequestSource::Mrf => true, }); // Build HealOptions with all available fields diff --git a/crates/heal/src/heal/manager.rs b/crates/heal/src/heal/manager.rs index 66b8f637f..216e17068 100644 --- a/crates/heal/src/heal/manager.rs +++ b/crates/heal/src/heal/manager.rs @@ -270,6 +270,8 @@ pub struct HealSourceCounts { pub auto_heal: u64, pub internal: u64, pub read_repair: u64, + #[serde(default)] + pub mrf: u64, } impl HealSourceCounts { @@ -280,6 +282,7 @@ impl HealSourceCounts { HealRequestSource::AutoHeal => self.auto_heal += 1, HealRequestSource::Internal => self.internal += 1, HealRequestSource::ReadRepair => self.read_repair += 1, + HealRequestSource::Mrf => self.mrf += 1, } } } diff --git a/crates/heal/src/heal/mod.rs b/crates/heal/src/heal/mod.rs index 0881e5853..ff910ed70 100644 --- a/crates/heal/src/heal/mod.rs +++ b/crates/heal/src/heal/mod.rs @@ -16,6 +16,7 @@ pub mod channel; pub mod erasure_healer; pub mod event; pub mod manager; +pub mod mrf_queue; pub mod progress; pub(crate) mod replacement_readiness; pub mod resume; diff --git a/crates/heal/src/heal/mrf_queue.rs b/crates/heal/src/heal/mrf_queue.rs new file mode 100644 index 000000000..c1144435e --- /dev/null +++ b/crates/heal/src/heal/mrf_queue.rs @@ -0,0 +1,682 @@ +// Copyright 2024 RustFS Team +// +// Licensed 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. + +//! Mission Repair Feed (MRF) queue, journal, and consumer. +//! +//! Intents arriving on the global channel (see `rustfs_common::mrf_channel`) +//! are buffered in a bounded in-memory queue, translated into prioritized +//! heal requests, and — while they are not yet accepted by the heal manager — +//! mirrored into a durable journal so a crash or restart can replay them. +//! This is the RustFS counterpart of MinIO's `.heal/mrf/list.bin` replay, +//! layered on top of (not replacing) read-repair and scanner heal. +//! +//! Durability model: the journal is a snapshot of the *unaccepted* pending +//! set, rewritten on a group-commit cadence (every flush interval or flush +//! threshold new intents). A rewrite is atomic at the record level only — a +//! torn tail simply truncates during replay because every record carries its +//! own CRC32. Losing the last flush window (≤500 ms) is acceptable: replayed +//! duplicates are merged by the manager's dedup key, and read-repair remains +//! the safety net. + +use super::{DiskStore, HealDiskExt as _, local_disk_map_read}; +use crate::heal::manager::HealManager; +use metrics::{counter, gauge}; +use rustfs_common::heal_channel::{HealAdmissionDropReason, HealAdmissionResult}; +use rustfs_common::mrf_channel::{MRF_MAX_ATTEMPTS, MrfIntent}; +use std::collections::VecDeque; +use std::sync::Arc; +use std::time::Duration; +use tokio::sync::mpsc; +use uuid::Uuid; + +use crate::heal::task::{HealOptions, HealPriority, HealRequest, HealType}; + +/// Journal location inside the metadata bucket, following the resume-state +/// layout. +pub(crate) const MRF_JOURNAL_PATH: &str = "buckets/.heal/mrf/journal.bin"; + +/// Record format tag. +const MRF_JOURNAL_FORMAT: u8 = 1; +/// Record layout version. +const MRF_JOURNAL_VERSION: u8 = 1; + +/// Fixed header size: format, version, kind, attempts, enqueued_at_ms, +/// has_version flag. +const MRF_RECORD_FIXED_HEAD: usize = 1 + 1 + 1 + 1 + 8 + 1; + +#[derive(Debug, Clone)] +pub(crate) struct MrfConsumerConfig { + /// In-memory queue capacity in intents. + pub queue_capacity: usize, + /// Journal byte budget; a pending snapshot above this bound is rejected + /// oldest-first so the journal can never grow unbounded. + pub journal_max_bytes: usize, + /// How many journal intents to re-arm per replay round. + pub replay_batch: usize, + /// Group-commit cadence for the journal snapshot. + pub flush_interval: Duration, + /// New intents between flushes that force an early snapshot. + pub flush_threshold: usize, + /// Backoff after the heal manager reports a full admission. + pub admission_backoff: Duration, +} + +impl Default for MrfConsumerConfig { + fn default() -> Self { + Self { + queue_capacity: rustfs_utils::get_env_usize( + rustfs_config::ENV_HEAL_MRF_QUEUE_SIZE, + rustfs_config::DEFAULT_HEAL_MRF_QUEUE_SIZE, + ), + journal_max_bytes: rustfs_utils::get_env_usize( + rustfs_config::ENV_HEAL_MRF_JOURNAL_MAX_BYTES, + rustfs_config::DEFAULT_HEAL_MRF_JOURNAL_MAX_BYTES, + ), + replay_batch: rustfs_utils::get_env_usize( + rustfs_config::ENV_HEAL_MRF_REPLAY_BATCH, + rustfs_config::DEFAULT_HEAL_MRF_REPLAY_BATCH, + ), + flush_interval: Duration::from_millis(500), + flush_threshold: 1000, + admission_backoff: Duration::from_secs(5), + } + } +} + +/// Bounded pending set with count and byte ceilings. Overflow drops the +/// incoming intent (never a resident one) and counts the loss. +pub(crate) struct MrfQueue { + pending: VecDeque, + bytes: usize, + capacity: usize, + byte_budget: usize, +} + +impl MrfQueue { + pub(crate) fn new(capacity: usize, byte_budget: usize) -> Self { + Self { + pending: VecDeque::new(), + bytes: 0, + capacity, + byte_budget, + } + } + + /// Returns `false` (after counting) when either ceiling would be crossed. + pub(crate) fn try_push(&mut self, intent: MrfIntent) -> bool { + let cost = intent.estimated_bytes(); + if self.pending.len() >= self.capacity || self.bytes + cost > self.byte_budget { + counter!("rustfs_heal_mrf_dropped_total", "reason" => "queue_overflow").increment(1); + return false; + } + self.bytes += cost; + self.pending.push_back(intent); + true + } + + pub(crate) fn pop_front(&mut self) -> Option { + let intent = self.pending.pop_front()?; + self.bytes = self.bytes.saturating_sub(intent.estimated_bytes()); + Some(intent) + } + + pub(crate) fn push_back(&mut self, intent: MrfIntent) { + self.bytes += intent.estimated_bytes(); + self.pending.push_back(intent); + } + + pub(crate) fn depth(&self) -> usize { + self.pending.len() + } + + pub(crate) fn bytes(&self) -> usize { + self.bytes + } + + pub(crate) fn intents(&self) -> impl Iterator { + self.pending.iter() + } +} + +// --------------------------------------------------------------------------- +// Journal record codec +// --------------------------------------------------------------------------- + +/// Append one encoded record to `out`. +pub(crate) fn encode_intent(intent: &MrfIntent, out: &mut Vec) { + let start = out.len(); + out.push(MRF_JOURNAL_FORMAT); + out.push(MRF_JOURNAL_VERSION); + out.push(match intent.kind { + rustfs_common::mrf_channel::MrfKind::DecodeFailure => 1, + rustfs_common::mrf_channel::MrfKind::MetadataCorruption => 2, + rustfs_common::mrf_channel::MrfKind::PartialWrite => 3, + }); + out.push(intent.attempts); + out.extend_from_slice(&intent.enqueued_at_ms.to_le_bytes()); + match intent.version_id { + Some(bytes) => { + out.push(1); + out.extend_from_slice(&bytes); + } + None => out.push(0), + } + out.extend_from_slice(&(intent.bucket.len() as u32).to_le_bytes()); + out.extend_from_slice(&(intent.object.len() as u32).to_le_bytes()); + out.extend_from_slice(intent.bucket.as_bytes()); + out.extend_from_slice(intent.object.as_bytes()); + let mut hasher = crc_fast::Digest::new(crc_fast::CrcAlgorithm::Crc32IsoHdlc); + hasher.update(&out[start..]); + out.extend_from_slice(&(hasher.finalize() as u32).to_le_bytes()); +} + +fn decode_one(data: &[u8]) -> Option<(MrfIntent, usize)> { + if data.len() < MRF_RECORD_FIXED_HEAD + 8 { + return None; + } + if data[0] != MRF_JOURNAL_FORMAT || data[1] != MRF_JOURNAL_VERSION { + return None; + } + let kind = match data[2] { + 1 => rustfs_common::mrf_channel::MrfKind::DecodeFailure, + 2 => rustfs_common::mrf_channel::MrfKind::MetadataCorruption, + 3 => rustfs_common::mrf_channel::MrfKind::PartialWrite, + _ => return None, + }; + let attempts = data[3]; + let enqueued_at_ms = u64::from_le_bytes(data[4..12].try_into().expect("slice length checked")); + let has_version = data[12] != 0; + let mut cursor = MRF_RECORD_FIXED_HEAD; + let version_id = if has_version { + if data.len() < cursor + 16 { + return None; + } + let bytes: [u8; 16] = data[cursor..cursor + 16].try_into().expect("slice length checked"); + cursor += 16; + Some(bytes) + } else { + None + }; + if data.len() < cursor + 8 { + return None; + } + let bucket_len = u32::from_le_bytes(data[cursor..cursor + 4].try_into().expect("slice length checked")) as usize; + let object_len = u32::from_le_bytes(data[cursor + 4..cursor + 8].try_into().expect("slice length checked")) as usize; + cursor += 8; + let body_end = cursor.checked_add(bucket_len)?.checked_add(object_len)?; + let record_end = body_end.checked_add(4)?; + if data.len() < record_end { + return None; + } + let mut hasher = crc_fast::Digest::new(crc_fast::CrcAlgorithm::Crc32IsoHdlc); + hasher.update(&data[..body_end]); + if (hasher.finalize() as u32) != u32::from_le_bytes(data[body_end..record_end].try_into().expect("slice length checked")) { + return None; + } + let bucket = std::sync::Arc::from(std::str::from_utf8(&data[cursor..cursor + bucket_len]).ok()?); + let object = std::sync::Arc::from(std::str::from_utf8(&data[cursor + bucket_len..body_end]).ok()?); + Some(( + MrfIntent { + bucket, + object, + version_id, + kind, + enqueued_at_ms, + attempts, + }, + record_end, + )) +} + +/// Decode a whole journal, stopping at the first torn or corrupt record. +/// Returns the decoded intents and the number of trailing bytes discarded. +pub(crate) fn decode_journal(data: &[u8]) -> (Vec, usize) { + let mut intents = Vec::new(); + let mut cursor = 0usize; + while cursor < data.len() { + match decode_one(&data[cursor..]) { + Some((intent, consumed)) => { + intents.push(intent); + cursor += consumed; + } + None => break, + } + } + let truncated = data.len() - cursor; + (intents, truncated) +} + +// --------------------------------------------------------------------------- +// Journal disk IO (all local disks, first successful read wins) +// --------------------------------------------------------------------------- + +async fn journal_disks() -> Vec { + let map = local_disk_map_read().await; + map.values().flatten().cloned().collect() +} + +async fn read_journal() -> Option> { + for disk in journal_disks().await { + match disk.read_all(super::RUSTFS_META_BUCKET, MRF_JOURNAL_PATH).await { + Ok(bytes) => return Some(bytes.to_vec()), + Err(_) => continue, + } + } + None +} + +async fn write_journal(data: &[u8]) { + let payload = bytes::Bytes::copy_from_slice(data); + for disk in journal_disks().await { + if let Err(err) = disk + .write_all(super::RUSTFS_META_BUCKET, MRF_JOURNAL_PATH, payload.clone()) + .await + { + warn_mrf_journal_write(&err); + } + } + if !data.is_empty() { + counter!("rustfs_heal_mrf_journal_fsync_total").increment(1); + } + gauge!("rustfs_heal_mrf_journal_bytes").set(data.len() as f64); +} + +async fn delete_journal() { + for disk in journal_disks().await { + let _ = disk + .delete( + super::RUSTFS_META_BUCKET, + MRF_JOURNAL_PATH, + crate::heal::storage_api::owner::EcstoreDeleteOptions::default(), + ) + .await; + } +} + +fn warn_mrf_journal_write(err: &super::DiskError) { + tracing::warn!( + target: "rustfs::heal::mrf", + error = %err, + "MRF journal write failed; unconsumed intents may be lost on restart" + ); +} + +// --------------------------------------------------------------------------- +// Consumer +// --------------------------------------------------------------------------- + +/// Translate an intent into the prioritized heal request the issue specifies: +/// decode failures go Urgent ECDecode, metadata corruption goes High +/// Metadata, partial writes go Normal object heal. +pub(crate) fn build_heal_request(intent: &MrfIntent) -> HealRequest { + let bucket = intent.bucket.to_string(); + let object = intent.object.to_string(); + let version_id = intent.version_id.map(|bytes| Uuid::from_bytes(bytes).to_string()); + let (heal_type, priority) = match intent.kind { + rustfs_common::mrf_channel::MrfKind::DecodeFailure => ( + HealType::ECDecode { + bucket, + object, + version_id, + }, + HealPriority::Urgent, + ), + rustfs_common::mrf_channel::MrfKind::MetadataCorruption => (HealType::Metadata { bucket, object }, HealPriority::High), + rustfs_common::mrf_channel::MrfKind::PartialWrite => ( + HealType::Object { + bucket, + object, + version_id, + }, + HealPriority::Normal, + ), + }; + let mut request = HealRequest::new(heal_type, HealOptions::default(), priority); + request.source = rustfs_common::heal_channel::HealRequestSource::Mrf; + request +} + +struct MrfRuntime { + queue: MrfQueue, + config: MrfConsumerConfig, + new_since_flush: usize, + /// True while a journal snapshot exists on disk that no longer reflects + /// an all-consumed pending set; the next idle tick removes it (MinIO + /// deletes its `list.bin` after replay for the same reason). + journal_on_disk: bool, + /// Earliest instant a full-admission retry may proceed. + backoff_until: Option, +} + +impl MrfRuntime { + fn record_accept(&mut self) { + // Accepted intents leave the pending set; the next flush persists the + // smaller snapshot, which is the journal's compaction. + } + + fn snapshot(&self) -> Vec { + let mut buf = Vec::new(); + for intent in self.queue.intents() { + encode_intent(intent, &mut buf); + } + buf + } + + async fn flush(&mut self) { + write_journal(&self.snapshot()).await; + self.new_since_flush = 0; + self.journal_on_disk = true; + } + + /// Drain pending intents into the heal manager until it is full, the + /// queue empties, or attempts are exhausted. + async fn dispatch(&mut self, manager: &HealManager) { + if let Some(until) = self.backoff_until { + if tokio::time::Instant::now() < until { + return; + } + self.backoff_until = None; + } + while let Some(mut intent) = self.queue.pop_front() { + let request = build_heal_request(&intent); + match manager.submit_heal_request(request).await { + Ok(HealAdmissionResult::Accepted) | Ok(HealAdmissionResult::Merged) => self.record_accept(), + Ok(HealAdmissionResult::Full) | Ok(HealAdmissionResult::Dropped(HealAdmissionDropReason::QueueFull)) => { + intent.attempts = intent.attempts.saturating_add(1); + if intent.attempts >= MRF_MAX_ATTEMPTS { + counter!("rustfs_heal_mrf_dropped_total", "reason" => "attempts_exhausted").increment(1); + continue; + } + self.queue.push_back(intent); + self.backoff_until = Some(tokio::time::Instant::now() + self.config.admission_backoff); + break; + } + Ok(HealAdmissionResult::Dropped(_)) => { + counter!("rustfs_heal_mrf_dropped_total", "reason" => "admission_policy").increment(1); + } + Err(_) => { + intent.attempts = intent.attempts.saturating_add(1); + if intent.attempts >= MRF_MAX_ATTEMPTS { + counter!("rustfs_heal_mrf_dropped_total", "reason" => "attempts_exhausted").increment(1); + continue; + } + self.queue.push_back(intent); + self.backoff_until = Some(tokio::time::Instant::now() + self.config.admission_backoff); + break; + } + } + } + gauge!("rustfs_heal_mrf_queue_depth").set(self.queue.depth() as f64); + gauge!("rustfs_heal_mrf_queue_bytes").set(self.queue.bytes() as f64); + } +} + +/// Initialize the global MRF channel (honoring `RUSTFS_HEAL_MRF_ENABLE`) and +/// spawn the consumer task. Called once from the heal runtime bootstrap right +/// after the manager started; a disabled feature or a double call is a no-op. +/// Public for integration tests that drive the real consumer loop. +pub fn spawn_mrf_consumer(manager: Arc) { + let enabled = rustfs_utils::get_env_bool(rustfs_config::ENV_HEAL_MRF_ENABLE, rustfs_config::DEFAULT_HEAL_MRF_ENABLE); + rustfs_common::mrf_channel::set_mrf_delivery_enabled(enabled); + if !enabled { + tracing::info!( + target: "rustfs::heal::mrf", + "MRF intent pipeline disabled by configuration; producers will not deliver" + ); + return; + } + let receiver = match rustfs_common::mrf_channel::init_mrf_channel() { + Ok(receiver) => receiver, + Err(err) => { + tracing::warn!( + target: "rustfs::heal::mrf", + error = err, + "MRF channel initialization failed; intents will be dropped at producers" + ); + return; + } + }; + tokio::spawn(async move { + run_mrf_consumer(manager, receiver).await; + }); + tracing::info!(target: "rustfs::heal::mrf", "MRF intent consumer started"); +} + +/// Replay the durable journal into a fresh pending queue and submit whatever +/// it armed. Returns the number of intact intents replayed. Duplicates are +/// merged by the manager's dedup key; the journal file is removed once read +/// (torn tails truncate via the per-record CRC). Public for integration tests; +/// the live consumer invokes this through [`replay_into`] at startup. +pub async fn replay_journal_once(manager: &Arc) -> usize { + let config = MrfConsumerConfig::default(); + let mut queue = MrfQueue::new(config.queue_capacity, config.journal_max_bytes); + let mut backoff_until: Option = None; + replay_into(manager, &mut queue, &mut backoff_until).await +} + +/// Shared replay core: read + decode + re-arm + delete, then drain what fits. +async fn replay_into( + manager: &Arc, + queue: &mut MrfQueue, + backoff_until: &mut Option, +) -> usize { + let Some(data) = read_journal().await else { + return 0; + }; + let (intents, truncated) = decode_journal(&data); + if truncated > 0 { + tracing::warn!( + target: "rustfs::heal::mrf", + truncated_bytes = truncated, + "MRF journal had a torn tail; truncated records were discarded" + ); + } + counter!("rustfs_heal_mrf_replayed_total").increment(intents.len() as u64); + let replayed = intents.len(); + for intent in intents { + queue.try_push(intent); + } + delete_journal().await; + + // Drain the replayed intents immediately; whatever the manager refuses + // stays armed in `queue` for the consumer's retry loop. + if backoff_until.is_none() { + while let Some(mut intent) = queue.pop_front() { + let request = build_heal_request(&intent); + match manager.submit_heal_request(request).await { + Ok(HealAdmissionResult::Accepted) | Ok(HealAdmissionResult::Merged) => {} + Ok(HealAdmissionResult::Full) | Ok(HealAdmissionResult::Dropped(HealAdmissionDropReason::QueueFull)) => { + intent.attempts = intent.attempts.saturating_add(1); + if intent.attempts < MRF_MAX_ATTEMPTS { + queue.push_back(intent); + *backoff_until = Some(tokio::time::Instant::now()); + } + break; + } + Ok(HealAdmissionResult::Dropped(_)) | Err(_) => {} + } + } + } + replayed +} + +/// Replay the journal, then keep draining the channel into the heal manager +/// while persisting the pending snapshot. +async fn run_mrf_consumer(manager: Arc, mut receiver: mpsc::Receiver) { + let config = MrfConsumerConfig::default(); + let mut runtime = MrfRuntime { + queue: MrfQueue::new(config.queue_capacity, config.journal_max_bytes), + config: config.clone(), + new_since_flush: 0, + journal_on_disk: false, + backoff_until: None, + }; + + // Replay: read the journal, re-arm intents (duplicates are merged by the + // manager's dedup key), then drop the file so the next flush starts clean. + replay_into(&manager, &mut runtime.queue, &mut runtime.backoff_until).await; + + let mut flush_tick = tokio::time::interval(runtime.config.flush_interval); + flush_tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + let mut batch: Vec = Vec::with_capacity(runtime.config.replay_batch); + + loop { + tokio::select! { + received = receiver.recv_many(&mut batch, runtime.config.replay_batch) => { + if received == 0 { + // Channel closed: flush once more and stop. + runtime.flush().await; + tracing::info!( + target: "rustfs::heal::mrf", + "MRF channel closed; consumer stopped after final flush" + ); + return; + } + for intent in batch.drain(..) { + runtime.queue.try_push(intent); + runtime.new_since_flush += 1; + } + runtime.dispatch(manager.as_ref()).await; + if runtime.new_since_flush >= runtime.config.flush_threshold { + runtime.flush().await; + } + } + _ = flush_tick.tick() => { + if runtime.new_since_flush > 0 || runtime.queue.depth() > 0 { + runtime.flush().await; + runtime.dispatch(manager.as_ref()).await; + } else if runtime.journal_on_disk { + // All intents consumed: remove the journal so a restart + // replays nothing (mirrors MinIO's post-replay unlink). + delete_journal().await; + runtime.journal_on_disk = false; + gauge!("rustfs_heal_mrf_journal_bytes").set(0.0); + } + gauge!("rustfs_heal_mrf_queue_depth").set(runtime.queue.depth() as f64); + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use rustfs_common::mrf_channel::{MrfIntent, MrfKind}; + use std::sync::Arc as StdArc; + + fn intent(bucket: &str, object: &str, attempts: u8) -> MrfIntent { + MrfIntent { + bucket: StdArc::from(bucket), + object: StdArc::from(object), + version_id: Some([7u8; 16]), + kind: MrfKind::DecodeFailure, + enqueued_at_ms: 1_700_000_000_000, + attempts, + } + } + + #[test] + fn queue_enforces_count_and_byte_ceilings() { + let mut queue = MrfQueue::new(2, usize::MAX); + assert!(queue.try_push(intent("b", "o", 0))); + assert!(queue.try_push(intent("b", "o", 0))); + assert!(!queue.try_push(intent("b", "o", 0)), "count ceiling must drop"); + + let mut tiny = MrfQueue::new(usize::MAX, intent("bucket", "object", 0).estimated_bytes()); + assert!(tiny.try_push(intent("bucket", "object", 0))); + assert!( + !tiny.try_push(intent("bucket", "object", 0)), + "byte budget must drop before the second intent fits" + ); + } + + #[test] + fn journal_roundtrip_preserves_intents() { + let intents = vec![ + intent("bucket-a", "object/a", 0), + intent("bucket-b", "object/b", 2), + MrfIntent { + bucket: StdArc::from("bucket-c"), + object: StdArc::from("object/c"), + version_id: None, + kind: MrfKind::MetadataCorruption, + enqueued_at_ms: 5, + attempts: 1, + }, + ]; + let mut buf = Vec::new(); + for intent in &intents { + encode_intent(intent, &mut buf); + } + let (decoded, truncated) = decode_journal(&buf); + assert_eq!(truncated, 0); + assert_eq!(decoded.len(), intents.len()); + for (left, right) in decoded.iter().zip(intents.iter()) { + assert_eq!(left.bucket, right.bucket); + assert_eq!(left.object, right.object); + assert_eq!(left.version_id, right.version_id); + assert_eq!(left.kind, right.kind); + assert_eq!(left.attempts, right.attempts); + } + } + + #[test] + fn journal_torn_tail_is_truncated() { + let mut buf = Vec::new(); + encode_intent(&intent("b", "o", 0), &mut buf); + let mut torn = buf.clone(); + torn.extend_from_slice(&buf[..buf.len() / 2]); + + let (decoded, truncated) = decode_journal(&torn); + assert_eq!(decoded.len(), 1, "the intact record must survive"); + assert!(truncated > 0, "the partial tail must be discarded"); + + // A corrupted body (CRC mismatch) also truncates from that record on. + let mut corrupt = buf.clone(); + let mid = MRF_RECORD_FIXED_HEAD + 4; + corrupt[mid] ^= 0xff; + let (decoded, truncated) = decode_journal(&corrupt); + assert!(decoded.is_empty()); + assert_eq!(truncated, corrupt.len()); + } + + #[test] + fn heal_request_mapping_follows_priority_matrix() { + let decode = build_heal_request(&intent("b", "o", 0)); + assert!(matches!(decode.heal_type, HealType::ECDecode { .. })); + assert_eq!(decode.priority, HealPriority::Urgent); + + let metadata = build_heal_request(&MrfIntent { + bucket: StdArc::from("b"), + object: StdArc::from("o"), + version_id: None, + kind: MrfKind::MetadataCorruption, + enqueued_at_ms: 0, + attempts: 0, + }); + assert!(matches!(metadata.heal_type, HealType::Metadata { .. })); + assert_eq!(metadata.priority, HealPriority::High); + + let partial = build_heal_request(&MrfIntent { + bucket: StdArc::from("b"), + object: StdArc::from("o"), + version_id: None, + kind: MrfKind::PartialWrite, + enqueued_at_ms: 0, + attempts: 0, + }); + assert!(matches!(partial.heal_type, HealType::Object { .. })); + assert_eq!(partial.priority, HealPriority::Normal); + } +} diff --git a/crates/heal/src/lib.rs b/crates/heal/src/lib.rs index 3dd29b064..f1ec4cebe 100644 --- a/crates/heal/src/lib.rs +++ b/crates/heal/src/lib.rs @@ -158,6 +158,10 @@ pub async fn init_heal_manager_with_workload_provider( return Err(err); } + // Start the MRF intent consumer (error-path repair intents + durable + // journal replay) now that the manager can accept submissions. + heal::mrf_queue::spawn_mrf_consumer(heal_manager.clone()); + #[cfg(test)] test_hook_after_manager_start().await; diff --git a/crates/heal/tests/mrf_pipeline_test.rs b/crates/heal/tests/mrf_pipeline_test.rs new file mode 100644 index 000000000..36f07a5a7 --- /dev/null +++ b/crates/heal/tests/mrf_pipeline_test.rs @@ -0,0 +1,189 @@ +// Copyright 2024 RustFS Team +// +// Licensed 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. + +//! HS-01 (rustfs/backlog#1865): MRF intent pipeline integration tests. +//! +//! Drives the real consumer loop (`spawn_mrf_consumer`) against a real +//! 4-disk `ECStore` heal storage and a `HealManager` that has not started its +//! scheduler, so submitted intents stay observable in the admission queue. +//! Under `cargo nextest` each test runs in its own process, which keeps the +//! process-global MRF channel singleton safe. + +use rustfs_common::mrf_channel::{self, MrfKind}; +use rustfs_heal::heal::{ + manager::{HealConfig, HealManager}, + mrf_queue, + storage::{ECStoreHealStorage, HealStorageAPI}, +}; +use serial_test::serial; +use std::{path::Path, sync::Arc, time::Duration}; + +mod storage_api; + +use storage_api::endpoint_index::{Endpoint, EndpointServerPools, Endpoints, PoolEndpoints, init_local_disks}; + +const META_BUCKET: &str = ".rustfs.sys"; +const JOURNAL_REL: &str = "buckets/.heal/mrf/journal.bin"; + +async fn heal_env() -> (Vec, Arc) { + let env = rustfs_test_utils::TestECStoreEnv::builder() + .prefix("rustfs_heal_mrf_test") + .build() + .await; + let heal_storage: Arc = Arc::new(ECStoreHealStorage::new(env.ecstore.clone())); + (env.disk_paths, heal_storage) +} + +fn make_manager(storage: Arc) -> Arc { + Arc::new(HealManager::new( + storage, + Some(HealConfig { + // Keep the scheduler from draining the queue before assertions. + heal_interval: Duration::from_secs(3600), + enable_auto_heal: false, + ..Default::default() + }), + )) +} + +/// Encode one journal record independently of the implementation, so a format +/// drift between writer and this fixture fails loudly here. +fn journal_record(kind: u8, bucket: &str, object: &str, version: Option<[u8; 16]>, attempts: u8) -> Vec { + let mut body = vec![1u8, 1, kind, attempts]; + body.extend_from_slice(&1_700_000_000_000u64.to_le_bytes()); + match version { + Some(bytes) => { + body.push(1); + body.extend_from_slice(&bytes); + } + None => body.push(0), + } + body.extend_from_slice(&(bucket.len() as u32).to_le_bytes()); + body.extend_from_slice(&(object.len() as u32).to_le_bytes()); + body.extend_from_slice(bucket.as_bytes()); + body.extend_from_slice(object.as_bytes()); + let mut hasher = crc_fast::Digest::new(crc_fast::CrcAlgorithm::Crc32IsoHdlc); + hasher.update(&body); + body.extend_from_slice(&(hasher.finalize() as u32).to_le_bytes()); + body +} + +fn write_journal_to_disks(disk_paths: &[std::path::PathBuf], data: &[u8]) { + for path in disk_paths { + let journal = path.join(META_BUCKET).join(JOURNAL_REL); + std::fs::create_dir_all(journal.parent().expect("journal parent")).expect("create journal dir"); + std::fs::write(&journal, data).expect("write journal fixture"); + } +} + +async fn wait_until(deadline: Duration, mut probe: F) -> bool +where + F: FnMut() -> Fut, + Fut: std::future::Future, +{ + let start = std::time::Instant::now(); + while start.elapsed() < deadline { + if probe().await { + return true; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + false +} + +/// A decode-failure intent delivered on the global channel must surface in the +/// heal manager as an Urgent request attributed to the MRF source. +#[tokio::test] +#[serial] +async fn decode_failure_intent_maps_to_urgent_mrf_heal_request() { + let (_disk_paths, storage) = heal_env().await; + let manager = make_manager(storage); + + mrf_queue::spawn_mrf_consumer(manager.clone()); + + assert!( + mrf_channel::try_send_mrf_intent(MrfKind::DecodeFailure, "mrf-bucket", "mrf-object", None), + "intent should be accepted while the consumer holds the channel" + ); + + let appeared = wait_until(Duration::from_secs(10), || async { + let snapshot = manager.operations_snapshot().await; + snapshot.queued_by_source.mrf >= 1 && snapshot.queued_by_priority.urgent >= 1 + }) + .await; + assert!( + appeared, + "MRF intent must reach the manager queue as an Urgent request (snapshot: {:?})", + manager.operations_snapshot().await + ); +} + +/// A journal left behind by a previous process must be replayed into the +/// manager queue and then removed, and a torn tail must not block replay of +/// the intact records. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +#[serial] +async fn journal_replay_arms_intents_and_deletes_the_file() { + let (disk_paths, storage) = heal_env().await; + + // The journal reader resolves disks through the process-local disk map; + // register the environment's disks the same way server startup does. + let mut endpoints: Vec = disk_paths + .iter() + .map(|p| Endpoint::try_from(p.to_string_lossy().as_ref()).expect("endpoint from disk path")) + .collect(); + for (i, endpoint) in endpoints.iter_mut().enumerate() { + endpoint.set_pool_index(0); + endpoint.set_set_index(0); + endpoint.set_disk_index(i); + } + let pool = PoolEndpoints { + legacy: false, + set_count: 1, + drives_per_set: endpoints.len(), + endpoints: Endpoints::from(endpoints), + cmd_line: "mrf-test".to_string(), + platform: String::new(), + }; + init_local_disks(EndpointServerPools::from(vec![pool])) + .await + .expect("local disks should register"); + + let mut journal = journal_record(1, "replay-bucket", "replay-object", Some([9u8; 16]), 0); + journal.extend(journal_record(3, "replay-bucket", "partial-object", None, 1)); + // Torn tail: a third record truncated mid-way must not block the two + // intact records above. + journal.extend_from_slice(&journal_record(2, "replay-bucket", "metadata-object", None, 0)[..8]); + write_journal_to_disks(&disk_paths, &journal); + + let manager = make_manager(storage); + // Replay directly (not via the process-global channel consumer, which the + // sibling test already claimed in this process under plain `cargo test`). + let replayed = mrf_queue::replay_journal_once(&manager).await; + assert_eq!(replayed, 2, "the two intact records must be replayed"); + + let snapshot = manager.operations_snapshot().await; + assert_eq!(snapshot.queued_by_source.mrf, 2, "replayed intents must be attributed to the MRF source"); + + assert!( + disk_paths + .iter() + .all(|path| !Path::new(path).join(META_BUCKET).join(JOURNAL_REL).exists()), + "the journal file must be removed after a successful replay" + ); + + let snapshot = manager.operations_snapshot().await; + assert_eq!(snapshot.queued_by_priority.urgent, 1, "the decode-failure record must replay as Urgent"); + assert!(snapshot.queued_by_priority.normal >= 1, "the partial-write record must replay as Normal"); +} diff --git a/crates/io-core/src/io_profile.rs b/crates/io-core/src/io_profile.rs index 7618eb949..86cd79448 100644 --- a/crates/io-core/src/io_profile.rs +++ b/crates/io-core/src/io_profile.rs @@ -26,7 +26,6 @@ pub enum StorageMedia { } impl StorageMedia { - #[allow(dead_code)] pub fn as_str(&self) -> &'static str { match self { Self::Nvme => "nvme", @@ -60,7 +59,6 @@ pub enum AccessPattern { } impl AccessPattern { - #[allow(dead_code)] pub fn as_str(&self) -> &'static str { match self { Self::Sequential => "sequential", @@ -71,25 +69,21 @@ impl AccessPattern { } /// Check if this is a sequential access pattern. - #[allow(dead_code)] pub fn is_sequential(&self) -> bool { matches!(self, Self::Sequential) } /// Check if this is a random access pattern. - #[allow(dead_code)] pub fn is_random(&self) -> bool { matches!(self, Self::Random) } /// Check if this is a mixed access pattern. - #[allow(dead_code)] pub fn is_mixed(&self) -> bool { matches!(self, Self::Mixed) } /// Check if this pattern is unknown. - #[allow(dead_code)] pub fn is_unknown(&self) -> bool { matches!(self, Self::Unknown) } diff --git a/crates/object-capacity/src/capacity_manager.rs b/crates/object-capacity/src/capacity_manager.rs index 231d3c6a0..2d70f985c 100644 --- a/crates/object-capacity/src/capacity_manager.rs +++ b/crates/object-capacity/src/capacity_manager.rs @@ -427,7 +427,6 @@ pub enum DataSource { /// Write triggered WriteTriggered, /// Fallback value - #[allow(dead_code)] Fallback, } @@ -603,7 +602,6 @@ impl WriteRecord { /// Hybrid strategy configuration #[derive(Debug, Clone)] -#[allow(dead_code)] pub struct HybridStrategyConfig { /// Scheduled update interval pub scheduled_update_interval: Duration, @@ -998,14 +996,12 @@ impl HybridCapacityManager { } /// Get cache age - #[allow(dead_code)] pub async fn get_cache_age(&self) -> Option { let cache = self.cache.read().await; cache.as_ref().map(|c| c.last_update.elapsed()) } /// Get write frequency (writes/minute) - #[allow(dead_code)] pub async fn get_write_frequency(&self) -> usize { let record = &self.write_record; record.recent_write_count(record.monotonic_second()) @@ -1300,7 +1296,6 @@ pub fn get_capacity_manager() -> Arc { /// .update_capacity(CapacityUpdate::exact(1000, 0), DataSource::RealTime) /// .await; /// ``` -#[allow(dead_code)] pub fn create_isolated_manager(config: HybridStrategyConfig) -> Arc { Arc::new(HybridCapacityManager::new(config)) } diff --git a/crates/rio/src/compress_index.rs b/crates/rio/src/compress_index.rs index 75e415e26..c085d70c3 100644 --- a/crates/rio/src/compress_index.rs +++ b/crates/rio/src/compress_index.rs @@ -49,7 +49,6 @@ pub struct IndexInfo { pub uncompressed_offset: i64, } -#[allow(dead_code)] impl Index { pub fn new() -> Self { Self { @@ -60,14 +59,6 @@ impl Index { } } - #[allow(dead_code)] - fn reset(&mut self, max_block: usize) { - self.est_block_uncomp = max_block as i64; - self.total_compressed = -1; - self.total_uncompressed = -1; - self.info.clear(); - } - pub fn len(&self) -> usize { self.info.len() } @@ -511,47 +502,6 @@ fn read_varint(buf: &[u8]) -> io::Result<(i64, usize)> { Err(io::Error::new(io::ErrorKind::UnexpectedEof, "unexpected EOF")) } -// Helper functions for index header manipulation -#[allow(dead_code)] -pub fn remove_index_headers(b: &[u8]) -> Option<&[u8]> { - if b.len() < 4 + S2_INDEX_TRAILER.len() { - return None; - } - - // Skip size - let b = &b[4..]; - - // Check trailer - if !b.starts_with(S2_INDEX_TRAILER) { - return None; - } - - Some(&b[S2_INDEX_TRAILER.len()..]) -} - -#[allow(dead_code)] -pub fn restore_index_headers(in_data: &[u8]) -> Vec { - if in_data.is_empty() { - return Vec::new(); - } - - let mut b = Vec::with_capacity(4 + S2_INDEX_HEADER.len() + in_data.len() + S2_INDEX_TRAILER.len() + 4); - b.extend_from_slice(&[0x50, 0x2A, 0x4D, 0x18]); - b.extend_from_slice(S2_INDEX_HEADER); - b.extend_from_slice(in_data); - - let total_size = (b.len() + 4 + S2_INDEX_TRAILER.len()) as u32; - b.extend_from_slice(&total_size.to_le_bytes()); - b.extend_from_slice(S2_INDEX_TRAILER); - - let chunk_len = b.len() - 4; - b[1] = chunk_len as u8; - b[2] = (chunk_len >> 8) as u8; - b[3] = (chunk_len >> 16) as u8; - - b -} - #[cfg(test)] mod tests { use super::*; diff --git a/crates/scanner/src/scanner_folder.rs b/crates/scanner/src/scanner_folder.rs index 15a01507c..030c55fc9 100644 --- a/crates/scanner/src/scanner_folder.rs +++ b/crates/scanner/src/scanner_folder.rs @@ -2478,6 +2478,15 @@ impl FolderScanner { } if let GetSizeFailureAction::HealMetadata { object } = failure_action { + // MRF journal intent: durable High-priority Metadata + // heal across restarts (HS-01); the scanner heal + // request below stays as the immediate path. + rustfs_common::mrf_channel::try_send_mrf_intent( + rustfs_common::mrf_channel::MrfKind::MetadataCorruption, + &item.bucket, + &object, + None, + ); self.send_required_scanner_heal_request( PendingScannerHealKind::Object, item.bucket.clone(), diff --git a/crates/scanner/tests/lifecycle_integration_test.rs b/crates/scanner/tests/lifecycle_integration_test.rs index 7b2130316..71fe961d2 100644 --- a/crates/scanner/tests/lifecycle_integration_test.rs +++ b/crates/scanner/tests/lifecycle_integration_test.rs @@ -206,7 +206,6 @@ async fn setup_isolated_test_env(init_expiry: bool) -> (Vec, Arc, bucket_name: &str) { (**ecstore) .make_bucket(bucket_name, &Default::default()) @@ -251,7 +250,6 @@ async fn modeled_versioned_delete_opts(bucket: &str, object: &str) -> ObjectOpti } /// Test helper: Set bucket lifecycle configuration -#[allow(dead_code)] async fn set_bucket_lifecycle(bucket_name: &str) -> Result<(), Box> { // Create a simple lifecycle configuration XML with 0 days expiry for immediate testing let lifecycle_xml = r#" @@ -274,7 +272,6 @@ async fn set_bucket_lifecycle(bucket_name: &str) -> Result<(), Box Result<(), Box> { // Create lifecycle rule that targets delete-marker cleanup only. // Keep Expiration.Days unset to avoid expiring live transitioned object versions. @@ -297,7 +294,6 @@ async fn set_bucket_lifecycle_deletemarker(bucket_name: &str) -> Result<(), Box< Ok(()) } -#[allow(dead_code)] async fn set_bucket_lifecycle_delmarker_expiration(bucket_name: &str, days: i64) -> Result<(), Box> { let lifecycle_xml = format!( r#" @@ -320,7 +316,6 @@ async fn set_bucket_lifecycle_delmarker_expiration(bucket_name: &str, days: i64) Ok(()) } -#[allow(dead_code)] async fn set_bucket_lifecycle_transition_with_tier( bucket_name: &str, storage_class: &str, @@ -368,7 +363,6 @@ async fn object_exists(ecstore: &Arc, bucket: &str, object: &str) -> bo } /// Test helper: Check if object exists -#[allow(dead_code)] async fn object_is_delete_marker(ecstore: &Arc, bucket: &str, object: &str) -> bool { if let Ok(oi) = (**ecstore).get_object_info(bucket, object, &ObjectOptions::default()).await { println!("oi: {oi:?}"); @@ -379,7 +373,6 @@ async fn object_is_delete_marker(ecstore: &Arc, bucket: &str, object: & } } -#[allow(dead_code)] async fn wait_for_object_absence(ecstore: &Arc, bucket: &str, object: &str, timeout: Duration) -> bool { let deadline = tokio::time::Instant::now() + timeout; diff --git a/crates/targets/src/net.rs b/crates/targets/src/net.rs index 812e78772..6ed2a6e41 100644 --- a/crates/targets/src/net.rs +++ b/crates/targets/src/net.rs @@ -428,7 +428,6 @@ pub fn parse_url(s: &str) -> Result { Ok(ParsedURL(uu)) } -#[allow(dead_code)] pub fn parse_http_url(s: &str) -> Result { let u = parse_url(s)?; match u.0.scheme() { @@ -437,7 +436,6 @@ pub fn parse_http_url(s: &str) -> Result { } } -#[allow(dead_code)] pub fn is_network_or_host_down(err: &std::io::Error, expect_timeouts: bool) -> bool { if err.kind() == std::io::ErrorKind::TimedOut { return !expect_timeouts; @@ -449,12 +447,10 @@ pub fn is_network_or_host_down(err: &std::io::Error, expect_timeouts: bool) -> b || err_str.contains("use of closed network connection") } -#[allow(dead_code)] pub fn is_conn_reset_err(err: &std::io::Error) -> bool { err.to_string().contains("connection reset by peer") || matches!(err.raw_os_error(), Some(libc::ECONNRESET)) } -#[allow(dead_code)] pub fn is_conn_refused_err(err: &std::io::Error) -> bool { err.to_string().contains("connection refused") || matches!(err.raw_os_error(), Some(libc::ECONNREFUSED)) } diff --git a/crates/targets/src/target/mqtt.rs b/crates/targets/src/target/mqtt.rs index 3120d85b1..0490c3a91 100644 --- a/crates/targets/src/target/mqtt.rs +++ b/crates/targets/src/target/mqtt.rs @@ -32,8 +32,8 @@ use arc_swap::ArcSwap; use async_trait::async_trait; use hyper_rustls::ConfigBuilderExt; use rumqttc::{ - AsyncClient, Broker, ClientError, ConnectionError, EventLoop, Incoming, MqttOptions, Outgoing, PublishNoticeError, QoS, - Transport, mqttbytes::Error as MqttBytesError, + AsyncClient, Broker, ClientError, ConnectionError, EventLoop, Incoming, MqttOptions, Outgoing, ProtocolViolation, + PublishNoticeError, PublishOptions, QoS, Transport, mqttbytes::Error as MqttBytesError, }; use rustfs_config::{ EnableState, MQTT_TLS_CA, MQTT_TLS_CLIENT_CERT, MQTT_TLS_CLIENT_KEY, MQTT_TLS_TRUST_LEAF_AS_CA, MQTT_WS_PATH_ALLOWLIST, @@ -791,7 +791,7 @@ where .as_ref() .ok_or_else(|| TargetError::Configuration("MQTT client not initialized".to_string()))?; let notice = client - .publish_tracked(&self.args.topic, self.args.qos, false, body) + .publish_tracked(&self.args.topic, body, PublishOptions::new(self.args.qos)) .await .map_err(|error| classify_mqtt_client_error(&error))?; drop(client_guard); @@ -1145,7 +1145,7 @@ async fn run_mqtt_event_loop(mut eventloop: EventLoop, connected_status: Arc { + rumqttc::Event::Incoming(Incoming::PingResp) => { trace!(target_id = %target_id, "Received PingResp from broker. Connection is alive."); } rumqttc::Event::Incoming(Incoming::SubAck(suback)) => { @@ -1257,7 +1257,11 @@ async fn run_mqtt_event_loop(mut eventloop: EventLoop, connected_status: Arc TargetError { match err { - ClientError::Request(_) | ClientError::TryRequest(_) | ClientError::TrackingUnavailable => TargetError::NotConnected, + ClientError::RequestChannelFull(_) | ClientError::RequestChannelDisconnected(_) | ClientError::TrackingUnavailable => { + TargetError::NotConnected + } + ClientError::InvalidRequest(_) => TargetError::Request(format!("Invalid MQTT publish request: {err}")), + _ => TargetError::NotConnected, } } @@ -1270,10 +1274,14 @@ fn classify_mqtt_notice_error(err: &PublishNoticeError) -> TargetError { PublishNoticeError::Recv | PublishNoticeError::SessionReset | PublishNoticeError::Qos0NotFlushed + | PublishNoticeError::BrokerOnlySessionResume + | PublishNoticeError::SessionPersistence(_) | PublishNoticeError::TopicAliasReplayUnavailable(_) => TargetError::NotConnected, + PublishNoticeError::RetainNotSupported => TargetError::Request(format!("MQTT broker rejected publish: {err}")), PublishNoticeError::V5PubAck(_) | PublishNoticeError::V5PubRec(_) | PublishNoticeError::V5PubComp(_) => { TargetError::Request(format!("MQTT broker rejected publish: {err}")) } + _ => TargetError::NotConnected, } } @@ -1299,12 +1307,13 @@ fn is_fatal_mqtt_error(err: &ConnectionError) -> bool { | MqttBytesError::MalformedPacket // Package format error | MqttBytesError::PayloadTooLong // Too long load | MqttBytesError::PayloadSizeLimitExceeded { .. } // Load size limit exceeded - | MqttBytesError::TopicNotUtf8 // Topic Non-UTF-8 (Serious Agreement Violation) + | MqttBytesError::TopicNotUtf8 { .. } // Topic Non-UTF-8 (Serious Agreement Violation) ) } // Others that are fatal StateError variants rumqttc::StateError::InvalidState // The internal state machine is in invalid state - | rumqttc::StateError::WrongPacket // Agreement Violation: Unexpected Data Packet Received + | rumqttc::StateError::ProtocolViolation(ProtocolViolation::UnexpectedIncomingPacket(_)) // Agreement Violation: Unexpected Data Packet Received + | rumqttc::StateError::ProtocolViolation(_) // Agreement Violation | rumqttc::StateError::Unsolicited(_) // Agreement Violation: Unsolicited ACK Received | rumqttc::StateError::CollisionTimeout // Agreement Violation (if this stage occurs) | rumqttc::StateError::EmptySubscription // Agreement violation (if this stage occurs) @@ -1727,8 +1736,8 @@ where mod tests { use super::{ AsyncClient, ClientError, MQTT_RECONNECT_BACKOFF_MAX, MQTT_RECONNECT_BACKOFF_MIN, MQTTArgs, MQTTTarget, MQTTTlsConfig, - MqttOptions, PublishNoticeError, QoS, QueuedPayloadMeta, classify_mqtt_client_error, classify_mqtt_notice_error, - next_reconnect_backoff, reconnect_supervisor, validate_mqtt_broker_url, + MqttOptions, PublishNoticeError, PublishOptions, QoS, QueuedPayloadMeta, classify_mqtt_client_error, + classify_mqtt_notice_error, next_reconnect_backoff, reconnect_supervisor, validate_mqtt_broker_url, }; use crate::error::TargetError; use crate::target::{REDACTED_SECRET, TargetType}; @@ -1794,7 +1803,7 @@ mod tests { .capacity(1) .build(); client - .publish("fill", QoS::AtLeastOnce, false, b"fill".as_slice()) + .publish("fill", b"fill".as_slice(), PublishOptions::new(QoS::AtLeastOnce)) .await .expect("first publish should fill the local channel"); *target.client.lock().await = Some(client); diff --git a/docs/architecture/global-state-inventory.md b/docs/architecture/global-state-inventory.md index 6c06b10c8..8407ad864 100644 --- a/docs/architecture/global-state-inventory.md +++ b/docs/architecture/global-state-inventory.md @@ -107,7 +107,7 @@ inventory. Generic function-local names such as `CACHE`, `LOCK`, `INIT`, and | `AUTH_FS` | `rustfs/src/storage/access.rs` | Cache or constant / owner-local cache | Authorization tag-condition lookup keeps its filesystem helper private to the access owner. | | `LOCK_STATS` | `rustfs/src/storage/lock_optimizer.rs` | Process-global owner-local metrics | Lock optimization statistics stay private behind lock optimizer helper APIs. | | `DEADLOCK_DETECTOR` | `rustfs/src/storage/deadlock_detector.rs` | Process-global owner-local state | Deadlock detector lifecycle state stays private to the storage deadlock detector owner. | -| `CONCURRENCY_MANAGER`, `ACTIVE_GET_REQUESTS`, `ACTIVE_PUT_REQUESTS`, `IO_PRIORITY_METRICS` | `rustfs/src/storage/concurrency/*` | Process-global owner-local scheduler state | Storage concurrency manager, counters, and metrics remain inside the storage concurrency owner boundary. | +| `CONCURRENCY_MANAGER`, `ACTIVE_GET_REQUESTS`, `ACTIVE_PUT_REQUESTS` | `rustfs/src/storage/concurrency/*` | Process-global owner-local scheduler state | Storage concurrency manager and request counters remain inside the storage concurrency owner boundary. | | `GET_OBJECT_BUFFER_THRESHOLD_WARNED`, `GET_READER_STREAM_BUFFER_SIZE_OVERRIDE`, function-local `ENABLED`, `OBJECT_SEEK_SUPPORT_THRESHOLD`, `OBJECT_SEEK_SUPPORT_CONCURRENCY_THRESHOLDS` | `rustfs/src/app/object_usecase.rs` | Cache or constant / owner-local cache | Object GET/seek tuning caches and warning guards stay private to object usecase helpers. | | `SUPPORTED_HEADERS` | `rustfs/src/storage/options.rs` | Cache or constant / owner-local constant | Supported-header lookup state stays private to storage option parsing. | | `AUDIT_TARGET_SPECS`, `NOTIFICATION_TARGET_SPECS` | `rustfs/src/admin/handlers/audit.rs`, `rustfs/src/admin/handlers/event.rs`, `rustfs/src/admin/handlers/plugins_instances.rs` | Cache or constant / owner-local constant | Admin target descriptor tables stay private to their handler owners. | diff --git a/docs/operations/heal-concurrency-safety-notes-zh.md b/docs/operations/heal-concurrency-safety-notes-zh.md new file mode 100644 index 000000000..ffb1e90c7 --- /dev/null +++ b/docs/operations/heal-concurrency-safety-notes-zh.md @@ -0,0 +1,115 @@ +# Heal 并发安全说明（对象级 healing 标记对标审计结论） + +对应 backlog rustfs/backlog#1874（父 #1862，HS-12）。本文回答一个问题：MinIO 在 heal +期间对对象打 `x-minio-healing:true` 元数据标记以防"heal 提交与并发删除/版本清理互毁" +（cmd/xl-storage.go RenameData 的 healing 分支），RustFS 是否需要同款防御。 + +**结论：不需要。** RustFS 不存在 MinIO 用 healing 标记防御的那类竞争：所有会触达同一 +`(bucket, object)` 提交面的路径都在同一把对象级 namespace 写锁上互斥，且 heal 的锁 +guard 覆盖 rename 提交全程；MinIO 需要标记的根因（RenameData 提交内部与版本清理逻辑 +交错）在 RustFS 的提交模型中不存在。RustFS 已有一个瞬态 healing 旗标用于另一目的 +（见下文 §2），并有并发不变量回归测试锁定本结论（§5）。 + +## 1. 两个防御模型的对照 + +MinIO：heal 时对对象写 `x-minio-healing:true`（持久元数据标记），后续任何 RenameData +提交看到该标记就跳过版本清理/legacy purge 逻辑——防御发生在锁外，靠元数据让路。 + +RustFS：三层防御，全部不依赖持久对象标记： + +1. **锁内互斥**：heal 与一切前台/后台写路径的提交点在同一把 `(bucket, object)` ns 写锁 + 上串行（分布式部署为 quorum 锁 RPC，单机为进程内锁管理器；锁粒度是对象级，version + 恒为 None）。 +2. **提交模型隔离**：rename_data 提交内没有会与 heal 交错的版本清理逻辑；被替换旧版本 + 的 data_dir 物理删除被移出提交临界区（commit tail），且只删已被新提交替换的 unshared + 目录。 +3. **瞬态 healing 旗标**：`FileInfo::set_healing`（crates/filemeta/src/fileinfo.rs）在 + heal 提交的内存 FileInfo 上打 `"healing"` 内部键，rename_data 据此允许先清空 stale + 目标 data_dir 再 rename——解决 heal 复用 data_dir 做 in-place 修复时 rename(2) 无法 + 替换非空目录的文件系统语义冲突（EEXIST/ENOTEMPTY）。该键是瞬态的，不落盘 + （`is_skip_meta_key`），与 MinIO 的持久标记目的不同。非 heal 提交撞上非空目标 + data_dir 会显式失败，有测试锁定两个方向的行为。 + +## 2. 交点矩阵 + +中心路径：`heal_object_with_explicit_version_regen`（crates/ecstore/src/set_disk/ops/heal.rs， +下称 heal.rs）在入口取 `(bucket, object)` ns 写锁，guard 绑定到函数作用域末尾，覆盖 +quorum 元数据读取 → EC 重建 → 逐盘 rename 提交 → tmp 清理 → HEAL_RENAME_INCOMPLETE +部分提交返回 → 孤儿 data_dir 回收的全过程。并发侧逐交点判定： + +| # | 并发路径 | 并发侧锁 | 判定 | 关键证据 | +|---|---|---|---|---| +| 1 | PUT 对象提交 | `put_object_commit` 对象写锁，rename_data 在锁内 | 同锁串行 | ops/object.rs 提交锁段 + rename 调用点 | +| 2 | PUT 旧 data_dir tail 清理 | drop 对象锁后的 `commit_rename_data_dir`，无锁 | 无锁并发，语义安全（见 §3.1） | object.rs drop 后 tail 段；io_primitives.rs | +| 3 | DELETE 单对象/版本 | `delete_object` 对象写锁，delete_version 在锁内 | 同锁串行 | object.rs delete_object 锁段 | +| 4 | DELETE 批量 | 批量逐对象写锁（dist 走批量锁 RPC） | 同锁串行 | object.rs delete_objects 锁段 | +| 5 | CompleteMultipart | 对象写锁 + upload 路径锁双锁，rename 在锁内 | 同锁串行 | ops/multipart.rs 提交锁段 | +| 6 | CompleteMultipart tail 清理 | drop 对象锁后的旧 data_dir 删除 | 无锁并发，语义安全（见 §3.1） | multipart.rs drop 后 tail 段 | +| 7 | AbortMultipart | 仅 multipart bucket 的 upload 路径锁 | 锁 key 不相交，但资源不相交（abort 不触对象 data_dir/xl.meta）→ 无实际交点 | multipart.rs abort 锁段 | +| 8 | ILM expiry（含 DeleteAllVersions） | DeleteAllVersions 走 `delete_prefix_object=true` → 仍取对象锁；FreeVersionTask 显式取锁；noncurrent 批量走批量锁 | 同锁串行 | bucket_lifecycle_ops.rs 消费端链路 | +| 9 | 纯 prefix 删除（绕锁能力面） | `delete_prefix`-only 不取子对象锁 | 无锁并发，但生产调用方为零（见 §3.2） | object.rs delete_object 锁条件 | +| 10 | 孤儿 data_dir 回收 reclaim_orphan_data_dirs | 函数本体无锁；唯一生产调用方在 heal 锁内 | heal 流程内=锁内串行 | heal.rs 收尾调用；io_primitives.rs | +| 11 | 旧清理 receipt 对账 reconcile_old_data_cleanup_receipts | 函数本体无锁；调用点在 heal 锁内 + epoch fence 防误删 | 锁内串行 | object.rs 对账函数 | +| 12 | replication | 数据面为远端 HTTP 写（不落本地盘）；本地元数据回写走对象锁 | 同锁串行 / 无交点 | replication_resyncer.rs 链路 | +| 13 | data_movement / rebalance / decommission 源清理 | 显式取对象锁 + 版本未变复核 + guard 复用（no_lock 只是复用已持锁） | 同锁串行 | data_movement/mod.rs 源清理 | +| 14 | copy_object | 目标对象锁 / 走 put 链锁 | 同锁串行 | object.rs copy_object 锁段 | +| 15 | 另一 heal 任务（跨 HealType/force_start） | dedup key 跨类型不相交 + force_start 跳过去重 → 任务级可并发 | 最终在 ns 写锁上串行 | heal/manager.rs dedup key 构成 | +| 16 | admin `no_lock=true` heal | 客户端可控绕锁 | 无锁并发，明示运维选项（见 §3.3） | admin/handlers/heal.rs 透传 | +| 17 | stale multipart 清理 | multipart bucket 的 upload 路径锁 | 资源不相交 → 无交点 | bucket_lifecycle_ops.rs 清理链路 | + +## 3. 残留窗口定性 + +### 3.1 PUT/CompleteMultipart commit tail（交点 2/6） + +写路径提交成功、释放对象锁之后，才 best-effort 删除被替换的旧 data_dir（注释明示有意 +不阻塞下一操作）。该删除与并发 heal 对同一旧 data_dir 的读取/重建存在竞态窗口，但语义 +安全： + +- 删除目标是已被新提交替换的 unshared data_dir；heal 的 canonical 元数据来自 quorum + 仲裁（ETag/mod_time），此时 quorum 已指向新版本，heal 不会把已替换版本当作 canonical + 复活； +- 竞态最坏后果 = heal 当轮对旧版本的一次 transient 失败/空转，重试轮自然收敛；清理 + residue 会上报并重新入队 heal（`report_old_data_dir_cleanup`）； +- 换盘重建等长 heal 走 per-version 显式版本请求，quorum 元数据在锁内读取，不受 tail + 影响。 + +### 3.2 纯 prefix 删除（交点 9） + +`delete_prefix && !delete_prefix_object` 的路径不取子对象锁（对象名空间锁无法保护前缀 +递归删除），与并发 heal 存在理论复活窗口（heal 在 prefix 删除进行中依据旧 quorum 元 +数据重建某版本）。全仓库核对结论：该路径的**生产调用方为零**——所有生产 `delete_prefix: +true` 调用点均同时设置 `delete_prefix_object: true`（从而取对象锁）或在测试模块内。这 +是 API 能力面的暴露而非行为风险。若未来有调用方需要纯 prefix 删除，须在调用点证明与 +heal/scanner 的隔离（例如 bucket 级停扫围栏）。 + +### 3.3 admin `no_lock=true`（交点 16） + +admin heal 请求可透传客户端 `nolock` 参数绕过 ns 锁（与 MinIO madmin 的同名选项对齐）。 +这是运维明示选项：使用即自负与并发写的竞争责任。文档化即可，不建议收紧。 + +## 4. heal 侧自身的不变量保障 + +- dedup key 跨 HealType 不相交（object/metadata/mrf/ecdecode/prefix 各自键面）+ admin + `force_start` 可跳过去重 → 同对象可能同时存在多个 heal 任务，但它们的执行体全部在 + `heal_object` 入口的 ns 写锁上串行（生产入口均 `no_lock=false`）； +- read-repair 的本地 TTL 预留只去重自身来源，不拦截其他来源的 heal——同样由 ns 锁兜底； +- healing 旗标不落盘，故不存在"标记残留导致后续提交错误让路"的反向风险。 + +## 5. 回归测试 + +以下两个并发不变量测试随本审计加入 `crates/ecstore/src/set_disk/ops/heal.rs` 测试模块： + +- `heal_racing_version_delete_never_resurrects_the_deleted_version`：注入 doomed 版本 + shard 损坏后，版本化 DELETE 与 Deep heal 真并发（同一把锁争用），断言已删除版本不被 + 复活、存活版本完好； +- `heal_racing_unversioned_overwrites_preserves_the_last_commit`：非版本化覆盖提交（激活 + commit tail 旧 data_dir 删除）与 Deep heal 循环竞态，断言最终 current 恰为最后一次 + 提交（etag 级一致）。 + +## 6. 结论 + +MinIO 的 `x-minio-healing` 是锁外元数据防御，前提是其 RenameData 提交内部存在与 heal +交错的版本清理逻辑；RustFS 的提交模型把这类交错从根上消除（提交面锁内互斥 + 清理外 +移到 tail + tail 只删 unshared 旧目录），因此引入持久对象级 healing 标记没有对应的竞争 +可防，反而会引入 FileInfo 落盘格式变更与标记残留清理两类新成本。维持现状，本对标疑点 +关闭。 diff --git a/rustfs/src/admin/handlers/heal.rs b/rustfs/src/admin/handlers/heal.rs index d9e7d1f7e..52e9fc152 100644 --- a/rustfs/src/admin/handlers/heal.rs +++ b/rustfs/src/admin/handlers/heal.rs @@ -317,6 +317,7 @@ fn add_source_counts(total: &mut rustfs_heal::HealSourceCounts, next: rustfs_hea total.auto_heal = total.auto_heal.saturating_add(next.auto_heal); total.internal = total.internal.saturating_add(next.internal); total.read_repair = total.read_repair.saturating_add(next.read_repair); + total.mrf = total.mrf.saturating_add(next.mrf); } fn add_operations(total: &mut rustfs_heal::HealOperationsSnapshot, next: rustfs_heal::HealOperationsSnapshot) { @@ -2353,6 +2354,7 @@ mod tests { auto_heal: value, internal: value, read_repair: value, + mrf: value, }; let operations = |value| rustfs_heal::HealOperationsSnapshot { queue_length: value, diff --git a/rustfs/src/admin/handlers/kms_management.rs b/rustfs/src/admin/handlers/kms_management.rs index b9ed1b4c1..796fc2319 100644 --- a/rustfs/src/admin/handlers/kms_management.rs +++ b/rustfs/src/admin/handlers/kms_management.rs @@ -16,13 +16,12 @@ use super::kms_dynamic::current_kms_config_fingerprint; use super::kms_keys::{CreateKeyHandler, DescribeKeyHandler, GenerateDataKeyHandler, ListKeysHandler}; -use crate::admin::auth::validate_admin_request; +use crate::admin::auth::authorize_admin_request; use crate::admin::router::{AdminOperation, Operation, S3Router}; use crate::admin::runtime_sources::{ current_kms_runtime_service_manager, current_notification_system, current_or_init_kms_runtime_service_manager, }; -use crate::auth::{check_key_valid, get_session_token}; -use crate::server::{ADMIN_PREFIX, RemoteAddr}; +use crate::server::ADMIN_PREFIX; use hyper::{HeaderMap, Method, StatusCode}; use matchit::Params; use rustfs_kms::KmsBackend; @@ -69,6 +68,18 @@ fn kms_clear_cache_actions() -> Vec { vec![Action::KmsAction(KmsAction::ClearCacheAction)] } +/// Admin gate for the KMS management endpoints, none of which act on a key. +/// +/// The pre-check keeps these endpoints' historical missing-credentials message; +/// the shared gate reports "get cred failed". +async fn authorize_kms_management_request(req: &S3Request, actions: Vec) -> S3Result<()> { + if req.credentials.is_none() { + return Err(s3_error!(InvalidRequest, "authentication required")); + } + authorize_admin_request(req, actions).await?; + Ok(()) +} + /// Response of `POST /kms/clear-cache`. /// /// Declared rather than built inline so the shape the console already depends @@ -260,22 +271,7 @@ pub struct KmsStatusHandler {} #[async_trait::async_trait] impl Operation for KmsStatusHandler { async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { - let Some(cred) = req.credentials else { - return Err(s3_error!(InvalidRequest, "authentication required")); - }; - - let (cred, owner) = - check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &cred.access_key).await?; - - validate_admin_request( - &req.headers, - &cred, - owner, - false, - kms_service_control_actions(), - req.extensions.get::>().and_then(|opt| opt.map(|a| a.0)), - ) - .await?; + authorize_kms_management_request(&req, kms_service_control_actions()).await?; let Some(service) = kms_encryption_service_from_context().await else { return Err(s3_error!(InternalError, "KMS service not initialized")); @@ -326,22 +322,7 @@ pub struct KmsConfigHandler {} #[async_trait::async_trait] impl Operation for KmsConfigHandler { async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { - let Some(cred) = req.credentials else { - return Err(s3_error!(InvalidRequest, "authentication required")); - }; - - let (cred, owner) = - check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &cred.access_key).await?; - - validate_admin_request( - &req.headers, - &cred, - owner, - false, - kms_configure_actions(), - req.extensions.get::>().and_then(|opt| opt.map(|a| a.0)), - ) - .await?; + authorize_kms_management_request(&req, kms_configure_actions()).await?; let Some(service) = kms_encryption_service_from_context().await else { return Err(s3_error!(InternalError, "KMS service not initialized")); @@ -375,22 +356,7 @@ pub struct KmsClearCacheHandler {} #[async_trait::async_trait] impl Operation for KmsClearCacheHandler { async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { - let Some(cred) = req.credentials else { - return Err(s3_error!(InvalidRequest, "authentication required")); - }; - - let (cred, owner) = - check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &cred.access_key).await?; - - validate_admin_request( - &req.headers, - &cred, - owner, - false, - kms_clear_cache_actions(), - req.extensions.get::>().and_then(|opt| opt.map(|a| a.0)), - ) - .await?; + authorize_kms_management_request(&req, kms_clear_cache_actions()).await?; let Some(service) = kms_encryption_service_from_context().await else { return Err(s3_error!(InternalError, "KMS service not initialized")); @@ -422,9 +388,14 @@ impl Operation for KmsClearCacheHandler { #[cfg(test)] mod tests { - use super::{KmsClearCacheResponse, kms_clear_cache_actions, kms_configure_actions, kms_service_control_actions}; + use super::{ + KmsClearCacheResponse, authorize_kms_management_request, kms_clear_cache_actions, kms_configure_actions, + kms_service_control_actions, + }; use crate::admin::handlers::kms_keys::stable_json_value; + use hyper::HeaderMap; use rustfs_policy::policy::action::{Action, AdminAction, KmsAction}; + use s3s::{Body, S3Request}; fn assert_has_action(actions: &[Action], action: Action) { assert!(actions.contains(&action), "expected action list to contain {action:?}"); @@ -434,6 +405,58 @@ mod tests { assert!(!actions.contains(&action), "expected action list not to contain {action:?}"); } + /// These endpoints authorize through the shared admin gate, which reports + /// "get cred failed" for a credential-less request. The pre-check keeps the + /// message these endpoints have always returned (rustfs/backlog#1829). + #[tokio::test] + async fn kms_management_gate_keeps_its_missing_credentials_message() { + let req = S3Request { + input: Body::from(String::new()), + method: http::Method::GET, + uri: "/rustfs/admin/v3/kms/status".parse().expect("uri should parse"), + headers: HeaderMap::new(), + extensions: http::Extensions::new(), + credentials: None, + region: None, + service: None, + trailing_headers: None, + }; + + let err = authorize_kms_management_request(&req, kms_service_control_actions()) + .await + .expect_err("a request without credentials must be rejected"); + assert_eq!(err.code(), &s3s::S3ErrorCode::InvalidRequest); + assert_eq!(err.message(), Some("authentication required")); + } + + /// Every management endpoint must reach the shared gate, each with its own + /// action set. The action lists are pinned above, but nothing else checks + /// which handler asks for which, and a handler that lost its gate entirely + /// would still serve its response. + #[test] + fn management_handlers_authorize_with_their_dedicated_actions() { + let src = include_str!("kms_management.rs"); + + for (handler, actions) in [ + ("KmsStatusHandler", "kms_service_control_actions()"), + ("KmsConfigHandler", "kms_configure_actions()"), + ("KmsClearCacheHandler", "kms_clear_cache_actions()"), + ] { + let block = src + .split_once(&format!("impl Operation for {handler}")) + .unwrap_or_else(|| panic!("{handler} impl should exist")) + .1; + let end = block + .find("\nimpl Operation for") + .or_else(|| block.find("\n#[cfg(test)]")) + .unwrap_or(block.len()); + assert!( + block[..end].contains(&format!("authorize_kms_management_request(&req, {actions})")), + "{handler} must authorize through the shared gate with {actions}" + ); + } + } + #[test] fn kms_management_auth_actions_use_dedicated_kms_actions() { assert_has_action(&kms_service_control_actions(), Action::KmsAction(KmsAction::ServiceControlAction)); diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 643b4fb6e..1bb337c69 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -56,8 +56,8 @@ use super::storage_api::object_usecase::bucket::{ }; use super::storage_api::object_usecase::compression::{MIN_DISK_COMPRESSIBLE_SIZE, is_disk_compressible}; use super::storage_api::object_usecase::concurrency::{ - self, ConcurrencyManager, DiskReadAdmission, GetObjectGuard, PutObjectGuard, get_concurrency_aware_buffer_size, - get_concurrency_manager, get_put_concurrency_aware_buffer_size, + self, ConcurrencyManager, DiskReadAdmission, GetObjectGuard, PutObjectAdmission, PutObjectGuard, + get_concurrency_aware_buffer_size, get_concurrency_manager, get_put_concurrency_aware_buffer_size, }; #[cfg(test)] use super::storage_api::object_usecase::contract::http::HTTPPreconditions; @@ -5681,6 +5681,35 @@ impl DefaultObjectUsecase { let server_side_encryption_requested = server_side_encryption.is_some() || sse_customer_algorithm.is_some() || ssekms_key_id.is_some(); + // Resolve the store through the request-bound server context + // (backlog#1052 S6), not the process-global handle, so an embedded + // second server never writes into the first server's store. + let Some(store) = self.object_store() else { + return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); + }; + let bucket_validate_stage_start = put_stage_metrics_enabled.then(Instant::now); + validate_bucket_exists(&store, &bucket).await?; + rustfs_io_metrics::record_put_object_stage_duration_from("app_bucket_validate", bucket_validate_stage_start); + + let put_admission = match get_concurrency_manager() + .admit_put_object() + .await + .map_err(|_| s3_error!(InternalError, "foreground write admission closed"))? + { + PutObjectAdmission::Disabled => None, + PutObjectAdmission::Admitted(permit) => { + counter!("rustfs.put_object.foreground_admission.total", "result" => "admitted").increment(1); + Some(permit) + } + PutObjectAdmission::Rejected => { + counter!("rustfs.put_object.foreground_admission.total", "result" => "rejected").increment(1); + return Err(s3_error!( + SlowDown, + "foreground write concurrency limit reached, please reduce your request rate" + )); + } + }; + let mut put_request_guard = PutObjectGuard::new(); let concurrent_put_requests = PutObjectGuard::concurrent_requests(); @@ -5733,16 +5762,6 @@ impl DefaultObjectUsecase { use_large_put_concurrency_tuning, ); - // Resolve the store through the request-bound server context - // (backlog#1052 S6), not the process-global handle, so an embedded - // second server never writes into the first server's store. - let Some(store) = self.object_store() else { - return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); - }; - let bucket_validate_stage_start = put_stage_metrics_enabled.then(Instant::now); - validate_bucket_exists(&store, &bucket).await?; - rustfs_io_metrics::record_put_object_stage_duration_from("app_bucket_validate", bucket_validate_stage_start); - let sse_config_stage_start = put_stage_metrics_enabled.then(Instant::now); let bucket_sse_config = metadata_sys::get_sse_config(&bucket).await.ok(); rustfs_io_metrics::record_put_object_stage_duration_from("app_sse_config_lookup", sse_config_stage_start); @@ -6132,7 +6151,9 @@ impl DefaultObjectUsecase { let cache_adapter = cache_adapter.clone(); let request_id = request_id.clone(); let put_path = put_path.to_string(); + let put_admission = put_admission; async move { + let _put_admission = put_admission; let object_traffic_progress = object_traffic_health .as_deref() .and_then(ObjectTrafficHealth::track_write_storage); @@ -6183,6 +6204,7 @@ impl DefaultObjectUsecase { } }; rustfs_io_metrics::record_put_object_stage_duration_from("app_store_put", store_put_stage_start); + drop(_put_admission); drop(object_traffic_progress); #[cfg(test)] wait_for_put_post_store_test_hook(&bucket).await; diff --git a/rustfs/src/app/storage_api.rs b/rustfs/src/app/storage_api.rs index 839b8a38f..c6f03f240 100644 --- a/rustfs/src/app/storage_api.rs +++ b/rustfs/src/app/storage_api.rs @@ -936,7 +936,7 @@ pub(crate) mod bucket { pub(crate) mod concurrency { pub(crate) use crate::storage::storage_api::concurrency_consumer::{ - ConcurrencyManager, DiskReadAdmission, GetObjectGuard, IoQueueStatus, IoStrategy, PutObjectGuard, + ConcurrencyManager, DiskReadAdmission, GetObjectGuard, IoQueueStatus, IoStrategy, PutObjectAdmission, PutObjectGuard, get_concurrency_aware_buffer_size, get_concurrency_manager, get_put_concurrency_aware_buffer_size, }; } diff --git a/rustfs/src/storage/concurrency/io_schedule.rs b/rustfs/src/storage/concurrency/io_schedule.rs index d611dfeb2..8d202708a 100644 --- a/rustfs/src/storage/concurrency/io_schedule.rs +++ b/rustfs/src/storage/concurrency/io_schedule.rs @@ -14,21 +14,12 @@ //! I/O scheduling types for adaptive buffer sizing and load management. //! -//! # Migration Note -//! -//! This module contains types that are also available in `rustfs_io_core`. -//! For new code, prefer using types from `rustfs_io_core` directly: -//! -//! ```ignore -//! // Recommended: Use io-core types -//! use rustfs_io_core::{ -//! IoLoadLevel, IoPriority, IoSchedulerConfig, -//! calculate_optimal_buffer_size, get_buffer_size_for_media, -//! }; -//! ``` -//! -//! This module remains for backward compatibility and provides additional -//! runtime monitoring features (`IoPriorityMetrics`, `IoStrategyDebugInfo`). +//! This is the live scheduling implementation. `rustfs_io_core` supplies the +//! shared config shapes (`IoSchedulerConfig`, `IoPriorityQueueConfig`) that the +//! types here project into through `to_core_config`, plus the `io_profile` +//! storage-media model; bandwidth samples come from `rustfs_io_metrics`. +//! Same-named io-core types are those config shapes, not a backing +//! implementation this module delegates to. use rustfs_config::{KI_B, MI_B}; use rustfs_io_core::io_profile::{AccessPattern, StorageMedia, StorageProfile}; @@ -1762,169 +1753,6 @@ impl IoPriorityQueue { } } -// ============================================ -// I/O Priority Queue Metrics -// ============================================ - -/// Global metrics for I/O priority queue monitoring. -/// -/// These metrics are emitted through the shared metrics pipeline and provide -/// visibility into the priority queue behavior. -#[allow(dead_code)] -pub struct IoPriorityMetrics { - /// High priority queue depth. - pub high_queue_depth: AtomicU64, - /// Normal priority queue depth. - pub normal_queue_depth: AtomicU64, - /// Low priority queue depth. - pub low_queue_depth: AtomicU64, - /// High priority total wait time in nanoseconds. - pub high_wait_time_ns: AtomicU64, - /// Normal priority total wait time in nanoseconds. - pub normal_wait_time_ns: AtomicU64, - /// Low priority total wait time in nanoseconds. - pub low_wait_time_ns: AtomicU64, - /// Total starvation events count. - pub starvation_events: AtomicU64, - /// High priority requests processed. - pub high_processed: AtomicU64, - /// Normal priority requests processed. - pub normal_processed: AtomicU64, - /// Low priority requests processed. - pub low_processed: AtomicU64, -} - -#[allow(dead_code)] -impl Default for IoPriorityMetrics { - fn default() -> Self { - Self::new() - } -} - -#[allow(dead_code)] -impl IoPriorityMetrics { - /// Create a new metrics instance. - pub const fn new() -> Self { - Self { - high_queue_depth: AtomicU64::new(0), - normal_queue_depth: AtomicU64::new(0), - low_queue_depth: AtomicU64::new(0), - high_wait_time_ns: AtomicU64::new(0), - normal_wait_time_ns: AtomicU64::new(0), - low_wait_time_ns: AtomicU64::new(0), - starvation_events: AtomicU64::new(0), - high_processed: AtomicU64::new(0), - normal_processed: AtomicU64::new(0), - low_processed: AtomicU64::new(0), - } - } - - /// Update queue depths from status. - #[allow(dead_code)] - pub fn update_queue_depths(&self, status: &IoQueueStatus) { - self.high_queue_depth - .store(status.high_priority_waiting as u64, Ordering::Relaxed); - self.normal_queue_depth - .store(status.normal_priority_waiting as u64, Ordering::Relaxed); - self.low_queue_depth - .store(status.low_priority_waiting as u64, Ordering::Relaxed); - } - - /// Record a starvation event. - #[allow(dead_code)] - pub fn record_starvation(&self) { - self.starvation_events.fetch_add(1, Ordering::Relaxed); - } - - /// Record a processed request. - #[allow(dead_code)] - pub fn record_processed(&self, priority: IoPriority) { - match priority { - IoPriority::High => self.high_processed.fetch_add(1, Ordering::Relaxed), - IoPriority::Normal => self.normal_processed.fetch_add(1, Ordering::Relaxed), - IoPriority::Low => self.low_processed.fetch_add(1, Ordering::Relaxed), - }; - } - - /// Record wait time for a priority level. - pub fn record_wait_time(&self, priority: IoPriority, wait_ns: u64) { - match priority { - IoPriority::High => self.high_wait_time_ns.fetch_add(wait_ns, Ordering::Relaxed), - IoPriority::Normal => self.normal_wait_time_ns.fetch_add(wait_ns, Ordering::Relaxed), - IoPriority::Low => self.low_wait_time_ns.fetch_add(wait_ns, Ordering::Relaxed), - }; - } - - /// Get high priority queue depth. - pub fn get_high_queue_depth(&self) -> u64 { - self.high_queue_depth.load(Ordering::Relaxed) - } - - /// Get normal priority queue depth. - pub fn get_normal_queue_depth(&self) -> u64 { - self.normal_queue_depth.load(Ordering::Relaxed) - } - - /// Get low priority queue depth. - pub fn get_low_queue_depth(&self) -> u64 { - self.low_queue_depth.load(Ordering::Relaxed) - } - - /// Get total starvation events. - pub fn get_starvation_events(&self) -> u64 { - self.starvation_events.load(Ordering::Relaxed) - } - - /// Get metrics summary for logging/debugging. - pub fn summary(&self) -> String { - format!( - "high_queue={}, normal_queue={}, low_queue={}, starvation={}, high_proc={}, normal_proc={}, low_proc={}", - self.get_high_queue_depth(), - self.get_normal_queue_depth(), - self.get_low_queue_depth(), - self.get_starvation_events(), - self.high_processed.load(Ordering::Relaxed), - self.normal_processed.load(Ordering::Relaxed), - self.low_processed.load(Ordering::Relaxed) - ) - } -} - -/// Global I/O priority metrics instance. -#[allow(dead_code)] -pub static IO_PRIORITY_METRICS: IoPriorityMetrics = IoPriorityMetrics::new(); - -/// Get optimized buffer size for I/O operations. -/// -/// This function provides adaptive buffer sizing based on: -/// - File size (small files get smaller buffers) -/// - Concurrent request count (high concurrency gets smaller buffers) -/// - Base buffer size from configuration -/// -/// # Arguments -/// -/// * `file_size` - Size of the file being read/written (-1 for unknown) -/// -/// # Returns -/// -/// Optimal buffer size in bytes -/// -/// # Example -/// -/// ```ignore -/// let buffer_size = get_buffer_size_opt_in(1024 * 1024); // 1MB file -/// assert!(buffer_size >= 64 * 1024); // At least 64KB -/// ``` -#[allow(dead_code)] -pub fn get_buffer_size_opt_in(file_size: i64) -> usize { - // Get base buffer size from configuration - let base_buffer_size = - rustfs_utils::get_env_usize(rustfs_config::ENV_OBJECT_IO_BUFFER_SIZE, rustfs_config::DEFAULT_OBJECT_IO_BUFFER_SIZE); - - // Apply concurrency-aware adjustments - get_concurrency_aware_buffer_size(file_size, base_buffer_size) -} - // ============================================ // Unit Tests // ============================================ @@ -1933,13 +1761,12 @@ pub fn get_buffer_size_opt_in(file_size: i64) -> usize { #[allow(unused_imports)] mod tests { use super::{ - IoLoadLevel, IoPriority, IoPriorityMetrics, IoPriorityQueue, IoPriorityQueueConfig, IoSchedulerConfig, - IoSchedulingContext, IoStrategy, get_advanced_buffer_size, get_buffer_size_opt_in, get_concurrency_aware_buffer_size, + IoLoadLevel, IoPriority, IoPriorityQueue, IoPriorityQueueConfig, IoSchedulerConfig, IoSchedulingContext, IoStrategy, + get_advanced_buffer_size, get_concurrency_aware_buffer_size, }; use rustfs_io_core::io_profile::{AccessPattern, StorageMedia}; use rustfs_io_metrics::bandwidth::{BandwidthSnapshot, BandwidthTier}; use serial_test::serial; - use std::sync::atomic::Ordering; use std::time::Duration; #[tokio::test] @@ -2126,29 +1953,6 @@ mod tests { assert_eq!(config.starvation_threshold_secs, 120); } - #[tokio::test] - #[serial] - async fn test_io_priority_metrics() { - let metrics = IoPriorityMetrics::new(); - - // Test initial state - assert_eq!(metrics.get_high_queue_depth(), 0); - assert_eq!(metrics.get_normal_queue_depth(), 0); - assert_eq!(metrics.get_low_queue_depth(), 0); - assert_eq!(metrics.get_starvation_events(), 0); - - // Test recording - metrics.record_starvation(); - assert_eq!(metrics.get_starvation_events(), 1); - - metrics.record_processed(IoPriority::High); - metrics.record_processed(IoPriority::High); - metrics.record_processed(IoPriority::Normal); - - assert_eq!(metrics.high_processed.load(Ordering::Relaxed), 2); - assert_eq!(metrics.normal_processed.load(Ordering::Relaxed), 1); - } - // ============================================ // Multi-Factor Strategy Tests // ============================================ diff --git a/rustfs/src/storage/concurrency/manager.rs b/rustfs/src/storage/concurrency/manager.rs index 13b509f93..5bc456148 100644 --- a/rustfs/src/storage/concurrency/manager.rs +++ b/rustfs/src/storage/concurrency/manager.rs @@ -65,6 +65,11 @@ pub struct ConcurrencyManager { bandwidth_monitor: Arc>, /// Metrics collector for I/O latency tracking (P50, P95, P99) metrics_collector: Arc, + /// Experimental fixed-count foreground PutObject admission gate. + put_admission_semaphore: Arc, + put_admission_enabled: bool, + put_admission_limit: usize, + put_admission_wait_timeout: Duration, } impl std::fmt::Debug for ConcurrencyManager { @@ -114,6 +119,18 @@ pub enum DiskReadAdmission { Rejected, } +/// Outcome of foreground PutObject request admission. +#[derive(Debug)] +pub enum PutObjectAdmission { + /// Foreground PUT admission is disabled; proceed on the legacy path. + Disabled, + /// Request is admitted and must hold the permit until the store write + /// returns or the request fails before mutation. + Admitted(tokio::sync::OwnedSemaphorePermit), + /// The fixed-count gate stayed full until the configured wait timeout. + Rejected, +} + #[allow(dead_code)] impl ConcurrencyManager { /// Create a new concurrency manager with default settings @@ -161,6 +178,18 @@ impl ConcurrencyManager { // Initialize metrics collector for I/O latency tracking // Keep 1000 samples for P95/P99 calculation let metrics_collector = Arc::new(MetricsCollector::new(performance_metrics, 1000)); + let put_admission_enabled = rustfs_utils::get_env_bool( + rustfs_config::ENV_PUT_FOREGROUND_ADMISSION_ENABLE, + rustfs_config::DEFAULT_PUT_FOREGROUND_ADMISSION_ENABLE, + ); + let put_admission_limit = rustfs_utils::get_env_usize( + rustfs_config::ENV_PUT_FOREGROUND_ADMISSION_LIMIT, + rustfs_config::DEFAULT_PUT_FOREGROUND_ADMISSION_LIMIT, + ); + let put_admission_wait_timeout = Duration::from_millis(rustfs_utils::get_env_u64( + rustfs_config::ENV_PUT_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS, + rustfs_config::DEFAULT_PUT_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS, + )); // Build queue config directly from scheduler config. let queue_config = IoPriorityQueueConfig::from_scheduler_config(&scheduler_config); @@ -176,6 +205,10 @@ impl ConcurrencyManager { pattern_detector, bandwidth_monitor, metrics_collector, + put_admission_semaphore: Arc::new(Semaphore::new(if put_admission_enabled { put_admission_limit } else { 0 })), + put_admission_enabled, + put_admission_limit, + put_admission_wait_timeout, } } @@ -199,6 +232,16 @@ impl ConcurrencyManager { self.degraded_read_semaphore.close(); } + #[cfg(test)] + pub(crate) fn with_put_admission_for_test(enabled: bool, limit: usize, wait_timeout: Duration) -> Self { + let mut manager = Self::new(); + manager.put_admission_semaphore = Arc::new(Semaphore::new(if enabled { limit } else { 0 })); + manager.put_admission_enabled = enabled; + manager.put_admission_limit = limit; + manager.put_admission_wait_timeout = wait_timeout; + manager + } + /// Track a GetObject request pub fn track_request() -> GetObjectGuard { GetObjectGuard::new() @@ -284,6 +327,32 @@ impl ConcurrencyManager { } } + /// Admit a foreground PutObject request under the experimental fixed-count gate. + /// + /// The default-off path returns [`PutObjectAdmission::Disabled`] without + /// touching the semaphore, preserving legacy behavior. When enabled, the + /// permit must be acquired before body ingest and held until the store write + /// returns, so saturated foreground writes can fail with `SlowDown` before + /// creating visible side effects. + pub async fn admit_put_object(&self) -> Result { + if !self.put_admission_enabled || self.put_admission_limit == 0 { + return Ok(PutObjectAdmission::Disabled); + } + + if self.put_admission_wait_timeout.is_zero() { + return Ok(match self.put_admission_semaphore.clone().try_acquire_owned() { + Ok(permit) => PutObjectAdmission::Admitted(permit), + Err(tokio::sync::TryAcquireError::NoPermits) => PutObjectAdmission::Rejected, + Err(tokio::sync::TryAcquireError::Closed) => PutObjectAdmission::Rejected, + }); + } + + match tokio::time::timeout(self.put_admission_wait_timeout, self.put_admission_semaphore.clone().acquire_owned()).await { + Ok(permit) => Ok(PutObjectAdmission::Admitted(permit?)), + Err(_) => Ok(PutObjectAdmission::Rejected), + } + } + // ============================================ // Adaptive I/O Strategy Methods // ============================================ @@ -692,8 +761,16 @@ impl ConcurrencyManager { /// Get a read-only workload admission snapshot for foreground writes. pub fn put_object_admission_snapshot(&self) -> WorkloadAdmissionSnapshot { - let active = PutObjectGuard::concurrent_count(); - let limit = self.scheduler_config.max_concurrent_reads; + let (active, limit, hard_gate_enabled) = if self.put_admission_enabled && self.put_admission_limit > 0 { + ( + self.put_admission_limit + .saturating_sub(self.put_admission_semaphore.available_permits()), + self.put_admission_limit, + true, + ) + } else { + (PutObjectGuard::concurrent_count(), self.scheduler_config.max_concurrent_reads, false) + }; let state = if limit == 0 { AdmissionState::Disabled } else if active >= limit { @@ -706,7 +783,10 @@ impl ConcurrencyManager { WorkloadAdmissionSnapshot::new(WorkloadClass::ForegroundWrite, state).with_counts(Some(active), None, Some(limit)); match state { - AdmissionState::Disabled => admission.with_reason("foreground write pressure tracking disabled"), + AdmissionState::Disabled => admission.with_reason("foreground write admission disabled"), + AdmissionState::Saturated if hard_gate_enabled => { + admission.with_reason("foreground write admission permits exhausted") + } AdmissionState::Saturated => admission.with_reason("foreground write concurrency reached local pressure limit"), _ => admission, } @@ -783,7 +863,7 @@ impl Default for ConcurrencyManager { mod integration_tests { use super::super::io_schedule::{IoLoadLevel, IoPriority}; use super::super::request_guard::GetObjectGuard; - use super::ConcurrencyManager; + use super::{ConcurrencyManager, PutObjectAdmission}; use crate::storage::storage_api::concurrency_consumer::PutObjectGuard; use rustfs_concurrency::{AdmissionState, WorkloadAdmissionSnapshotProvider, WorkloadClass}; use rustfs_io_core::io_profile::{AccessPattern, StorageMedia}; @@ -880,6 +960,71 @@ mod integration_tests { crate::storage::concurrency::reset_active_put_requests(); } + #[tokio::test] + #[serial] + async fn test_concurrency_manager_put_admission_disabled_does_not_touch_gate() { + let manager = ConcurrencyManager::with_put_admission_for_test(false, 1, Duration::ZERO); + + let admission = manager + .admit_put_object() + .await + .expect("disabled put admission must not close"); + + assert!(matches!(admission, PutObjectAdmission::Disabled)); + assert_eq!(manager.put_admission_semaphore.available_permits(), 0); + assert_eq!(manager.put_object_admission_snapshot().state, AdmissionState::Open); + } + + #[tokio::test] + #[serial] + async fn test_concurrency_manager_put_admission_rejects_when_limit_full() { + let manager = ConcurrencyManager::with_put_admission_for_test(true, 1, Duration::ZERO); + + let first = manager.admit_put_object().await.expect("first put admission should acquire"); + assert!(matches!(first, PutObjectAdmission::Admitted(_))); + assert_eq!(manager.put_object_admission_snapshot().state, AdmissionState::Saturated); + + let second = manager + .admit_put_object() + .await + .expect("full put admission gate should reject, not close"); + assert!(matches!(second, PutObjectAdmission::Rejected)); + } + + #[tokio::test] + #[serial] + async fn test_concurrency_manager_put_admission_reuses_released_permit() { + let manager = ConcurrencyManager::with_put_admission_for_test(true, 1, Duration::ZERO); + + let first = manager.admit_put_object().await.expect("first put admission should acquire"); + drop(first); + + let second = manager + .admit_put_object() + .await + .expect("released put admission permit should be reusable"); + assert!(matches!(second, PutObjectAdmission::Admitted(_))); + } + + #[tokio::test(start_paused = true)] + #[serial] + async fn test_concurrency_manager_put_admission_wait_timeout_rejects() { + let manager = ConcurrencyManager::with_put_admission_for_test(true, 1, Duration::from_secs(5)); + let held = manager.admit_put_object().await.expect("first put admission should acquire"); + let waiter_manager = manager.clone(); + + let waiter = tokio::spawn(async move { waiter_manager.admit_put_object().await }); + tokio::task::yield_now().await; + tokio::time::advance(Duration::from_secs(5)).await; + + let admission = waiter + .await + .expect("put admission waiter task must not panic") + .expect("put admission gate must stay open"); + assert!(matches!(admission, PutObjectAdmission::Rejected)); + drop(held); + } + #[tokio::test] #[serial] async fn test_concurrency_manager_workload_admission_registry_covers_required_classes() { diff --git a/rustfs/src/storage/concurrency/mod.rs b/rustfs/src/storage/concurrency/mod.rs index 110e14cf6..eb47aa4ed 100644 --- a/rustfs/src/storage/concurrency/mod.rs +++ b/rustfs/src/storage/concurrency/mod.rs @@ -24,16 +24,14 @@ //! - **Concurrency Management**: Coordination of concurrent GetObject requests //! - **Request Tracking**: RAII guards for request lifecycle management //! -//! # Migration Note +//! # Relationship to the shared crates //! -//! Core algorithms have been migrated to `rustfs-io-core` and metrics to -//! `rustfs-io-metrics`. This module maintains API compatibility while -//! delegating to the new implementations. +//! The scheduling algorithm lives in [`io_schedule`], not in `rustfs-io-core`: +//! this module does not delegate to it. `rustfs-io-core` owns the shared +//! config shapes and the `io_profile` storage-media model that [`io_schedule`] +//! consumes, and `rustfs-io-metrics` owns bandwidth sampling and metric +//! recording. -// Sub-modules -// pub mod bandwidth_monitor; // Migrated to rustfs-io-metrics -// pub mod global_metrics; // Migrated to rustfs-io-metrics -// pub mod io_profile; // Migrated to rustfs-io-core pub mod io_schedule; pub mod manager; pub mod request_guard; @@ -45,34 +43,15 @@ pub mod request_guard; // I/O scheduling types (from io_schedule.rs for backward compatibility) #[allow(unused_imports)] pub use io_schedule::{ - IO_PRIORITY_METRICS, IoLoadLevel, IoPriority, IoPriorityMetrics, IoPriorityQueue, IoPriorityQueueConfig, IoQueueStatus, - IoSchedulerConfig, IoStrategy, get_advanced_buffer_size, get_buffer_size_opt_in, get_concurrency_aware_buffer_size, - get_put_concurrency_aware_buffer_size, + IoLoadLevel, IoPriority, IoPriorityQueue, IoPriorityQueueConfig, IoQueueStatus, IoSchedulerConfig, IoStrategy, + get_advanced_buffer_size, get_concurrency_aware_buffer_size, get_put_concurrency_aware_buffer_size, }; // Request tracking pub use request_guard::{GetObjectGuard, PutObjectGuard}; // Concurrency manager -pub use manager::{ConcurrencyManager, DiskReadAdmission}; - -// ============================================ -// New Module Re-exports (for gradual migration) -// ============================================ - -// Re-export types from rustfs-io-core for convenience -pub use rustfs_io_core::{ - // Backpressure types - BackpressureMonitor, - // Deadlock detection types - DeadlockDetector, - // Scheduler types - IoScheduler, - // Lock optimization types - LockOptimizer, -}; - -// Re-export types from rustfs-io-metrics for convenience +pub use manager::{ConcurrencyManager, DiskReadAdmission, PutObjectAdmission}; // ============================================ // Helper Functions @@ -83,37 +62,8 @@ pub fn get_concurrency_manager() -> &'static ConcurrencyManager { ConcurrencyManager::global() } -/// Reset the active get requests counter (for testing). -#[allow(dead_code)] -pub fn reset_active_get_requests() { - io_schedule::ACTIVE_GET_REQUESTS.store(0, std::sync::atomic::Ordering::Relaxed); -} - -#[allow(dead_code)] +/// Reset the active put requests counter (for testing). +#[cfg(test)] pub fn reset_active_put_requests() { io_schedule::ACTIVE_PUT_REQUESTS.store(0, std::sync::atomic::Ordering::Relaxed); } - -/// Create a new I/O scheduler with default configuration. -#[allow(dead_code)] -pub fn create_io_scheduler() -> IoScheduler { - IoScheduler::with_defaults() -} - -/// Create a new backpressure monitor with default configuration. -#[allow(dead_code)] -pub fn create_backpressure_monitor() -> BackpressureMonitor { - BackpressureMonitor::with_defaults() -} - -/// Create a new deadlock detector with default configuration. -#[allow(dead_code)] -pub fn create_deadlock_detector() -> DeadlockDetector { - DeadlockDetector::with_defaults() -} - -/// Create a new lock optimizer with default configuration. -#[allow(dead_code)] -pub fn create_lock_optimizer() -> LockOptimizer { - LockOptimizer::with_defaults() -} diff --git a/rustfs/src/storage/rpc/node_service/heal.rs b/rustfs/src/storage/rpc/node_service/heal.rs index f9b990bf3..8f796726c 100644 --- a/rustfs/src/storage/rpc/node_service/heal.rs +++ b/rustfs/src/storage/rpc/node_service/heal.rs @@ -585,6 +585,7 @@ mod tests { let decoded = decode_node_heal_status(&encoded).expect("fixed v1 fixture should decode"); assert_eq!(decoded.info().bitrot_start_cycle, 9); assert_eq!(decoded.operations.queue_length, 2); + assert_eq!(decoded.operations.queued_by_source.mrf, 0); } #[test] diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index d1e7c0f3a..be9548619 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -117,7 +117,7 @@ pub(crate) mod access_consumer { pub(crate) mod concurrency_consumer { pub(crate) use super::super::concurrency::{ - ConcurrencyManager, DiskReadAdmission, GetObjectGuard, IoQueueStatus, IoStrategy, PutObjectGuard, + ConcurrencyManager, DiskReadAdmission, GetObjectGuard, IoQueueStatus, IoStrategy, PutObjectAdmission, PutObjectGuard, get_concurrency_aware_buffer_size, get_concurrency_manager, get_put_concurrency_aware_buffer_size, }; }