Compare commits

..

7 Commits

Author SHA1 Message Date
overtrue c568a54797 fix(replication): honor disabled version deletes 2026-07-30 13:37:01 +08:00
cxymds 67904a6c18 fix(ecstore): start with unresolved Kubernetes peers (#5460)
* fix(ecstore): start with unresolved Kubernetes peers

* fix(ecstore): infer Kubernetes endpoint identity safely

* fix(ecstore): fail closed on unsafe format migration

* fix(ecstore): reject poisoned format heal candidates

* fix(ecstore): reject unsafe legacy migration outliers

* fix(ecstore): resume interrupted format migrations

* fix(ecstore): preserve Kubernetes startup compatibility
2026-07-30 11:14:38 +08:00
Zhengchao An 2e5cef513f chore(release): prepare 1.0.0-beta.12 (#5461)
* chore(release): prepare 1.0.0-beta.12

* chore(release): align release assets for 1.0.0-beta.12

* chore(deps): refresh release dependencies

Co-Authored-By: heihutu <heihutu@gmail.com>

---------

Co-authored-by: houseme <housemecn@gmail.com>
Co-authored-by: heihutu <heihutu@gmail.com>
2026-07-30 01:31:28 +00:00
Zhengchao An 2d4f77fd3b fix(iam): add correctly named policy APIs (#5457) 2026-07-30 00:55:23 +00:00
Zhengchao An 920705417c refactor(ecstore): correct bucket config static names (#5458)
refactor(ecstore): fix bucket config static names
2026-07-30 00:53:16 +00:00
Zhengchao An b097c94c59 ci: model server as a composition root (#5459)
* ci: model server as a composition root

* test(ci): cover server to storage layer flow
2026-07-30 00:52:37 +00:00
Zhengchao An 3991a1d73c test(ecstore): stabilize late snapshot lease cleanup (#5456)
test(ecstore): wait for late snapshot lease cleanup
2026-07-30 00:10:38 +00:00
66 changed files with 5263 additions and 1126 deletions
+8 -2
View File
@@ -10,10 +10,16 @@ never weaken a check to get green.
## `check_layer_dependencies.sh` — layer DAG in `rustfs/src`
Enforces `interface (admin, storage/ecfs, storage/s3_api) → app → infra`; no
upward imports. Known legacy violations live in
Enforces `composition (server, startup/init) → interface (admin,
storage/ecfs, storage/s3_api) → app → infra`; no upward imports. Server source
files are composition roots, while imports of their exported HTTP contracts
are classified as interface dependencies. Known legacy violations live in
`scripts/layer-dependency-baseline.txt`.
Dedicated `*_test.rs` and `tests/` modules are outside this production guard.
Inline `#[cfg(test)]` imports remain checked under their source file's layer;
move architecture-crossing test scaffolding into a dedicated test module.
- **New violation**: restructure your change so the dependency points
downward (move the shared type/function to the lower layer).
- **You legitimately removed a baseline entry**: run
Generated
+124 -123
View File
@@ -642,9 +642,9 @@ dependencies = [
[[package]]
name = "async-compression"
version = "0.4.42"
version = "0.4.43"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e79b3f8a79cccc2898f31920fc69f304859b3bd567490f75ebf51ae1c792a9ac"
checksum = "3976abdc8fe7d1133d43d304afd42abdf5bc3e1319d263d223bde07b5efc4be8"
dependencies = [
"compression-codecs",
"compression-core",
@@ -870,7 +870,7 @@ dependencies = [
"bytes",
"fastrand",
"hex",
"http 1.4.2",
"http 1.5.0",
"sha1 0.10.7",
"time",
"tokio",
@@ -934,7 +934,7 @@ dependencies = [
"bytes-utils",
"fastrand",
"http 0.2.12",
"http 1.4.2",
"http 1.5.0",
"http-body 0.4.6",
"http-body 1.1.0",
"percent-encoding",
@@ -970,7 +970,7 @@ dependencies = [
"hex",
"hmac 0.13.0",
"http 0.2.12",
"http 1.4.2",
"http 1.5.0",
"http-body 1.1.0",
"lru 0.16.4",
"percent-encoding",
@@ -1001,7 +1001,7 @@ dependencies = [
"bytes",
"fastrand",
"http 0.2.12",
"http 1.4.2",
"http 1.5.0",
"regex-lite",
"tracing",
]
@@ -1027,7 +1027,7 @@ dependencies = [
"bytes",
"fastrand",
"http 0.2.12",
"http 1.4.2",
"http 1.5.0",
"regex-lite",
"tracing",
]
@@ -1054,7 +1054,7 @@ dependencies = [
"aws-types",
"fastrand",
"http 0.2.12",
"http 1.4.2",
"http 1.5.0",
"regex-lite",
"tracing",
]
@@ -1076,7 +1076,7 @@ dependencies = [
"hex",
"hmac 0.13.0",
"http 0.2.12",
"http 1.4.2",
"http 1.5.0",
"p256 0.13.2",
"percent-encoding",
"sha2 0.11.0",
@@ -1108,7 +1108,7 @@ dependencies = [
"bytes",
"crc-fast",
"hex",
"http 1.4.2",
"http 1.5.0",
"http-body 1.1.0",
"http-body-util",
"md-5 0.11.0",
@@ -1142,7 +1142,7 @@ dependencies = [
"bytes-utils",
"futures-core",
"futures-util",
"http 1.4.2",
"http 1.5.0",
"http-body 1.1.0",
"http-body-util",
"percent-encoding",
@@ -1161,7 +1161,7 @@ dependencies = [
"aws-smithy-runtime-api",
"aws-smithy-types",
"h2",
"http 1.4.2",
"http 1.5.0",
"hyper",
"hyper-rustls",
"hyper-util",
@@ -1224,7 +1224,7 @@ dependencies = [
"bytes",
"fastrand",
"http 0.2.12",
"http 1.4.2",
"http 1.5.0",
"http-body 0.4.6",
"http-body 1.1.0",
"http-body-util",
@@ -1245,7 +1245,7 @@ dependencies = [
"aws-smithy-types",
"bytes",
"http 0.2.12",
"http 1.4.2",
"http 1.5.0",
"pin-project-lite",
"tokio",
"tracing",
@@ -1271,7 +1271,7 @@ checksum = "7d56e0a4e53127a632224e43633b0fe045fa9e1e3cfc68b9830f1115e103f910"
dependencies = [
"aws-smithy-runtime-api",
"aws-smithy-types",
"http 1.4.2",
"http 1.5.0",
]
[[package]]
@@ -1285,7 +1285,7 @@ dependencies = [
"bytes-utils",
"futures-core",
"http 0.2.12",
"http 1.4.2",
"http 1.5.0",
"http-body 0.4.6",
"http-body 1.1.0",
"http-body-util",
@@ -1337,7 +1337,7 @@ dependencies = [
"bytes",
"form_urlencoded",
"futures-util",
"http 1.4.2",
"http 1.5.0",
"http-body 1.1.0",
"http-body-util",
"hyper",
@@ -1368,7 +1368,7 @@ checksum = "08c78f31d7b1291f7ee735c1c6780ccde7785daae9a9206026862dab7d8792d1"
dependencies = [
"bytes",
"futures-core",
"http 1.4.2",
"http 1.5.0",
"http-body 1.1.0",
"http-body-util",
"mime",
@@ -3229,7 +3229,7 @@ dependencies = [
"futures-util",
"headers",
"htmlescape",
"http 1.4.2",
"http 1.5.0",
"http-body 1.1.0",
"http-body-util",
"libc",
@@ -3654,7 +3654,7 @@ checksum = "d0881ea181b1df73ff77ffaaf9c7544ecc11e82fba9b5f27b262a3c73a332555"
[[package]]
name = "e2e_test"
version = "1.0.0-beta.11"
version = "1.0.0-beta.12"
dependencies = [
"anyhow",
"astral-tokio-tar",
@@ -3672,7 +3672,7 @@ dependencies = [
"flate2",
"futures",
"hex",
"http 1.4.2",
"http 1.5.0",
"http-body-util",
"hyper",
"hyper-util",
@@ -4384,7 +4384,7 @@ dependencies = [
"google-cloud-gax",
"hex",
"hmac 0.13.0",
"http 1.4.2",
"http 1.5.0",
"jsonwebtoken 10.4.0",
"reqwest",
"rustc_version",
@@ -4409,7 +4409,7 @@ dependencies = [
"futures",
"google-cloud-rpc",
"google-cloud-wkt",
"http 1.4.2",
"http 1.5.0",
"pin-project",
"rand 0.10.2",
"serde",
@@ -4431,7 +4431,7 @@ dependencies = [
"google-cloud-rpc",
"google-cloud-wkt",
"h2",
"http 1.4.2",
"http 1.5.0",
"http-body 1.1.0",
"http-body-util",
"hyper",
@@ -4544,7 +4544,7 @@ dependencies = [
"google-cloud-type",
"google-cloud-wkt",
"hex",
"http 1.4.2",
"http 1.5.0",
"http-body 1.1.0",
"md5",
"percent-encoding",
@@ -4624,7 +4624,7 @@ dependencies = [
"fnv",
"futures-core",
"futures-sink",
"http 1.4.2",
"http 1.5.0",
"indexmap 2.14.0",
"slab",
"tokio",
@@ -4727,7 +4727,7 @@ dependencies = [
"base64 0.22.1",
"bytes",
"headers-core",
"http 1.4.2",
"http 1.5.0",
"httpdate",
"mime",
"sha1 0.10.7",
@@ -4739,7 +4739,7 @@ version = "0.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "54b4a22553d4242c49fddb9ba998a99962b5cc6f22cb5a3482bec22522403ce4"
dependencies = [
"http 1.4.2",
"http 1.5.0",
]
[[package]]
@@ -4961,9 +4961,9 @@ dependencies = [
[[package]]
name = "http"
version = "1.4.2"
version = "1.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6970f50e31d6fc17d3fa27329444bfa74e196cf62e95052a3f6fee181dba6425"
checksum = "918d3568bebf352712bc2ef3d46a8bcf1a75b373be6539de198e9105cbbf9ce0"
dependencies = [
"bytes",
"itoa",
@@ -4987,7 +4987,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ca2a8f2913ee65f60facd6a5905613afaa448497a0230cc41ce022d93290bc2c"
dependencies = [
"bytes",
"http 1.4.2",
"http 1.5.0",
]
[[package]]
@@ -4998,7 +4998,7 @@ checksum = "e9f41fd6a08e4d4ec69df65976da761afd5ad5e58a9d4acb46bd1c953a9e3ff2"
dependencies = [
"bytes",
"futures-core",
"http 1.4.2",
"http 1.5.0",
"http-body 1.1.0",
"pin-project-lite",
]
@@ -5044,7 +5044,7 @@ dependencies = [
"futures-channel",
"futures-core",
"h2",
"http 1.4.2",
"http 1.5.0",
"http-body 1.1.0",
"httparse",
"httpdate",
@@ -5061,7 +5061,7 @@ version = "0.27.9"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "33ca68d021ef39cf6463ab54c1d0f5daf03377b70561305bb89a8f83aab66e0f"
dependencies = [
"http 1.4.2",
"http 1.5.0",
"hyper",
"hyper-util",
"log",
@@ -5095,7 +5095,7 @@ dependencies = [
"bytes",
"futures-channel",
"futures-util",
"http 1.4.2",
"http 1.5.0",
"http-body 1.1.0",
"hyper",
"ipnet",
@@ -6704,10 +6704,10 @@ version = "5.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "51e219e79014df21a225b1860a479e2dcd7cbd9130f4defd4bd0e191ea31d67d"
dependencies = [
"base64 0.22.1",
"base64 0.21.7",
"chrono",
"getrandom 0.2.17",
"http 1.4.2",
"http 1.5.0",
"rand 0.8.7",
"serde",
"serde_json",
@@ -6815,7 +6815,7 @@ dependencies = [
"futures-channel",
"futures-core",
"futures-util",
"http 1.4.2",
"http 1.5.0",
"humantime",
"itertools 0.14.0",
"parking_lot",
@@ -6871,7 +6871,7 @@ dependencies = [
"dyn-clone",
"ed25519-dalek 2.2.0",
"hmac 0.12.1",
"http 1.4.2",
"http 1.5.0",
"itertools 0.10.5",
"log",
"oauth2",
@@ -6932,7 +6932,7 @@ checksum = "5683015d09e2df236ef005b17f6f196f0d5f6313c4fa43a7b6a53b52776e4331"
dependencies = [
"async-trait",
"bytes",
"http 1.4.2",
"http 1.5.0",
"opentelemetry",
"reqwest",
]
@@ -6944,7 +6944,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9966929966d17620d7c316c643ba62631826e10021409357772d5eea84f62c35"
dependencies = [
"flate2",
"http 1.4.2",
"http 1.5.0",
"opentelemetry",
"opentelemetry-http",
"opentelemetry-proto",
@@ -7820,7 +7820,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "be769465445e8c1474e9c5dac2018218498557af32d9ed057325ec9a41ae81bf"
dependencies = [
"heck",
"itertools 0.14.0",
"itertools 0.10.5",
"log",
"multimap",
"once_cell",
@@ -7840,7 +7840,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "03da047801ff44bb6a4d407d4860c05fd70bb81714e6b2f3812603d5b145b042"
dependencies = [
"heck",
"itertools 0.14.0",
"itertools 0.10.5",
"log",
"multimap",
"petgraph 0.8.3",
@@ -7861,7 +7861,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8a56d757972c98b346a9b766e3f02746cde6dd1cd1d1d563472929fdd74bec4d"
dependencies = [
"anyhow",
"itertools 0.14.0",
"itertools 0.10.5",
"proc-macro2",
"quote",
"syn 2.0.119",
@@ -7874,7 +7874,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b570b25f7617e43d59005d0990ccb79e950a423952cea19671b7a876da390adf"
dependencies = [
"anyhow",
"itertools 0.14.0",
"itertools 0.10.5",
"proc-macro2",
"quote",
"syn 2.0.119",
@@ -8329,9 +8329,9 @@ dependencies = [
[[package]]
name = "redis"
version = "1.4.1"
version = "1.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b0b9503711b03773e43b31668c7b5bd279ee7cd9b7d18cff7c23a42cc1d08e5a"
checksum = "3257df217f7eab0044627a268c9cc6cdb60c0c421c88f83ac41c4e31520b6b84"
dependencies = [
"arc-swap",
"arcstr",
@@ -8481,7 +8481,7 @@ dependencies = [
"futures-core",
"futures-util",
"h2",
"http 1.4.2",
"http 1.5.0",
"http-body 1.1.0",
"http-body-util",
"hyper",
@@ -8627,7 +8627,7 @@ dependencies = [
"async-tungstenite",
"futures-io",
"futures-util",
"http 1.4.2",
"http 1.5.0",
"rustls-native-certs",
"rustls-pki-types",
"rustls-webpki",
@@ -8657,7 +8657,7 @@ dependencies = [
"flume",
"futures-io",
"futures-util",
"http 1.4.2",
"http 1.5.0",
"log",
"mqttbytes-core-next",
"rumqttc-core-next",
@@ -8845,7 +8845,7 @@ dependencies = [
[[package]]
name = "rustfs"
version = "1.0.0-beta.11"
version = "1.0.0-beta.12"
dependencies = [
"aes-gcm",
"anyhow",
@@ -8874,7 +8874,7 @@ dependencies = [
"hex-simd",
"hmac 0.13.0",
"hotpath",
"http 1.4.2",
"http 1.5.0",
"http-body 1.1.0",
"http-body-util",
"hyper",
@@ -8981,7 +8981,7 @@ dependencies = [
[[package]]
name = "rustfs-audit"
version = "1.0.0-beta.11"
version = "1.0.0-beta.12"
dependencies = [
"async-trait",
"chrono",
@@ -9003,12 +9003,12 @@ dependencies = [
[[package]]
name = "rustfs-checksums"
version = "1.0.0-beta.11"
version = "1.0.0-beta.12"
dependencies = [
"base64-simd",
"bytes",
"crc-fast",
"http 1.4.2",
"http 1.5.0",
"md-5 0.11.0",
"pretty_assertions",
"sha1 0.11.0",
@@ -9018,7 +9018,7 @@ dependencies = [
[[package]]
name = "rustfs-common"
version = "1.0.0-beta.11"
version = "1.0.0-beta.12"
dependencies = [
"chrono",
"metrics",
@@ -9033,7 +9033,7 @@ dependencies = [
[[package]]
name = "rustfs-concurrency"
version = "1.0.0-beta.11"
version = "1.0.0-beta.12"
dependencies = [
"insta",
"rustfs-io-core",
@@ -9045,7 +9045,7 @@ dependencies = [
[[package]]
name = "rustfs-config"
version = "1.0.0-beta.11"
version = "1.0.0-beta.12"
dependencies = [
"const-str",
"serde",
@@ -9054,7 +9054,7 @@ dependencies = [
[[package]]
name = "rustfs-credentials"
version = "1.0.0-beta.11"
version = "1.0.0-beta.12"
dependencies = [
"base64-simd",
"hmac 0.13.0",
@@ -9067,7 +9067,7 @@ dependencies = [
[[package]]
name = "rustfs-crypto"
version = "1.0.0-beta.11"
version = "1.0.0-beta.12"
dependencies = [
"aes-gcm",
"argon2",
@@ -9087,7 +9087,7 @@ dependencies = [
[[package]]
name = "rustfs-data-usage"
version = "1.0.0-beta.11"
version = "1.0.0-beta.12"
dependencies = [
"async-trait",
"rmp-serde",
@@ -9097,7 +9097,7 @@ dependencies = [
[[package]]
name = "rustfs-ecstore"
version = "1.0.0-beta.11"
version = "1.0.0-beta.12"
dependencies = [
"aes-gcm",
"arc-swap",
@@ -9128,8 +9128,9 @@ dependencies = [
"google-cloud-storage",
"hex-simd",
"hmac 0.13.0",
"hostname",
"hotpath",
"http 1.4.2",
"http 1.5.0",
"http-body 1.1.0",
"http-body-util",
"hyper",
@@ -9235,7 +9236,7 @@ dependencies = [
[[package]]
name = "rustfs-extension-schema"
version = "1.0.0-beta.11"
version = "1.0.0-beta.12"
dependencies = [
"serde",
"serde_json",
@@ -9244,7 +9245,7 @@ dependencies = [
[[package]]
name = "rustfs-filemeta"
version = "1.0.0-beta.11"
version = "1.0.0-beta.12"
dependencies = [
"arc-swap",
"byteorder",
@@ -9270,12 +9271,12 @@ dependencies = [
[[package]]
name = "rustfs-heal"
version = "1.0.0-beta.11"
version = "1.0.0-beta.12"
dependencies = [
"async-trait",
"base64 0.23.0",
"futures",
"http 1.4.2",
"http 1.5.0",
"metrics",
"rustfs-common",
"rustfs-concurrency",
@@ -9300,13 +9301,13 @@ dependencies = [
[[package]]
name = "rustfs-iam"
version = "1.0.0-beta.11"
version = "1.0.0-beta.12"
dependencies = [
"arc-swap",
"async-trait",
"base64-simd",
"futures",
"http 1.4.2",
"http 1.5.0",
"jsonwebtoken 11.0.0",
"moka",
"openidconnect",
@@ -9337,7 +9338,7 @@ dependencies = [
[[package]]
name = "rustfs-io-core"
version = "1.0.0-beta.11"
version = "1.0.0-beta.12"
dependencies = [
"bytes",
"memmap2",
@@ -9349,7 +9350,7 @@ dependencies = [
[[package]]
name = "rustfs-io-metrics"
version = "1.0.0-beta.11"
version = "1.0.0-beta.12"
dependencies = [
"criterion",
"metrics",
@@ -9412,11 +9413,11 @@ dependencies = [
[[package]]
name = "rustfs-keystone"
version = "1.0.0-beta.11"
version = "1.0.0-beta.12"
dependencies = [
"bytes",
"futures",
"http 1.4.2",
"http 1.5.0",
"http-body 1.1.0",
"http-body-util",
"hyper",
@@ -9438,7 +9439,7 @@ dependencies = [
[[package]]
name = "rustfs-kms"
version = "1.0.0-beta.11"
version = "1.0.0-beta.12"
dependencies = [
"aes-gcm",
"arc-swap",
@@ -9472,7 +9473,7 @@ dependencies = [
[[package]]
name = "rustfs-lifecycle"
version = "1.0.0-beta.11"
version = "1.0.0-beta.12"
dependencies = [
"async-trait",
"metrics",
@@ -9494,7 +9495,7 @@ dependencies = [
[[package]]
name = "rustfs-lock"
version = "1.0.0-beta.11"
version = "1.0.0-beta.12"
dependencies = [
"async-trait",
"crossbeam-queue",
@@ -9516,7 +9517,7 @@ dependencies = [
[[package]]
name = "rustfs-log-analyzer"
version = "1.0.0-beta.11"
version = "1.0.0-beta.12"
dependencies = [
"chrono",
"flate2",
@@ -9534,7 +9535,7 @@ dependencies = [
[[package]]
name = "rustfs-madmin"
version = "1.0.0-beta.11"
version = "1.0.0-beta.12"
dependencies = [
"chrono",
"humantime",
@@ -9548,7 +9549,7 @@ dependencies = [
[[package]]
name = "rustfs-notify"
version = "1.0.0-beta.11"
version = "1.0.0-beta.12"
dependencies = [
"arc-swap",
"async-trait",
@@ -9582,7 +9583,7 @@ dependencies = [
[[package]]
name = "rustfs-object-capacity"
version = "1.0.0-beta.11"
version = "1.0.0-beta.12"
dependencies = [
"criterion",
"futures",
@@ -9600,7 +9601,7 @@ dependencies = [
[[package]]
name = "rustfs-object-data-cache"
version = "1.0.0-beta.11"
version = "1.0.0-beta.12"
dependencies = [
"bytes",
"criterion",
@@ -9616,7 +9617,7 @@ dependencies = [
[[package]]
name = "rustfs-obs"
version = "1.0.0-beta.11"
version = "1.0.0-beta.12"
dependencies = [
"chrono",
"crossbeam-channel",
@@ -9668,7 +9669,7 @@ dependencies = [
[[package]]
name = "rustfs-policy"
version = "1.0.0-beta.11"
version = "1.0.0-beta.12"
dependencies = [
"async-trait",
"base64-simd",
@@ -9697,7 +9698,7 @@ dependencies = [
[[package]]
name = "rustfs-protocols"
version = "1.0.0-beta.11"
version = "1.0.0-beta.12"
dependencies = [
"astral-tokio-tar",
"async-compression",
@@ -9710,7 +9711,7 @@ dependencies = [
"futures-util",
"hex",
"hmac 0.13.0",
"http 1.4.2",
"http 1.5.0",
"http-body-util",
"hyper",
"hyper-util",
@@ -9759,7 +9760,7 @@ dependencies = [
[[package]]
name = "rustfs-protos"
version = "1.0.0-beta.11"
version = "1.0.0-beta.12"
dependencies = [
"flatbuffers",
"prost 0.14.4",
@@ -9782,7 +9783,7 @@ dependencies = [
[[package]]
name = "rustfs-replication"
version = "1.0.0-beta.11"
version = "1.0.0-beta.12"
dependencies = [
"byteorder",
"bytes",
@@ -9799,7 +9800,7 @@ dependencies = [
[[package]]
name = "rustfs-rio"
version = "1.0.0-beta.11"
version = "1.0.0-beta.12"
dependencies = [
"aes-gcm",
"arc-swap",
@@ -9811,7 +9812,7 @@ dependencies = [
"futures",
"hex-simd",
"hotpath",
"http 1.4.2",
"http 1.5.0",
"http-body-util",
"md-5 0.11.0",
"pin-project-lite",
@@ -9837,7 +9838,7 @@ dependencies = [
[[package]]
name = "rustfs-rio-v2"
version = "1.0.0-beta.11"
version = "1.0.0-beta.12"
dependencies = [
"aes-gcm",
"bytes",
@@ -9859,14 +9860,14 @@ dependencies = [
[[package]]
name = "rustfs-s3-ops"
version = "1.0.0-beta.11"
version = "1.0.0-beta.12"
dependencies = [
"rustfs-s3-types",
]
[[package]]
name = "rustfs-s3-types"
version = "1.0.0-beta.11"
version = "1.0.0-beta.12"
dependencies = [
"serde",
"serde_json",
@@ -9874,7 +9875,7 @@ dependencies = [
[[package]]
name = "rustfs-s3select-api"
version = "1.0.0-beta.11"
version = "1.0.0-beta.12"
dependencies = [
"async-trait",
"bytes",
@@ -9882,7 +9883,7 @@ dependencies = [
"datafusion",
"futures",
"futures-core",
"http 1.4.2",
"http 1.5.0",
"metrics",
"parking_lot",
"rustfs-common",
@@ -9903,7 +9904,7 @@ dependencies = [
[[package]]
name = "rustfs-s3select-query"
version = "1.0.0-beta.11"
version = "1.0.0-beta.12"
dependencies = [
"async-recursion",
"async-trait",
@@ -9919,7 +9920,7 @@ dependencies = [
[[package]]
name = "rustfs-scanner"
version = "1.0.0-beta.11"
version = "1.0.0-beta.12"
dependencies = [
"async-trait",
"bytes",
@@ -9927,7 +9928,7 @@ dependencies = [
"futures",
"hex-simd",
"hmac 0.13.0",
"http 1.4.2",
"http 1.5.0",
"metrics",
"rand 0.10.2",
"rmp-serde",
@@ -9957,18 +9958,18 @@ dependencies = [
[[package]]
name = "rustfs-security-governance"
version = "1.0.0-beta.11"
version = "1.0.0-beta.12"
dependencies = [
"thiserror 2.0.19",
]
[[package]]
name = "rustfs-signer"
version = "1.0.0-beta.11"
version = "1.0.0-beta.12"
dependencies = [
"base64-simd",
"bytes",
"http 1.4.2",
"http 1.5.0",
"hyper",
"rustfs-utils",
"s3s",
@@ -9981,7 +9982,7 @@ dependencies = [
[[package]]
name = "rustfs-storage-api"
version = "1.0.0-beta.11"
version = "1.0.0-beta.12"
dependencies = [
"async-trait",
"insta",
@@ -9995,7 +9996,7 @@ dependencies = [
[[package]]
name = "rustfs-targets"
version = "1.0.0-beta.11"
version = "1.0.0-beta.12"
dependencies = [
"arc-swap",
"async-nats",
@@ -10048,7 +10049,7 @@ dependencies = [
[[package]]
name = "rustfs-test-utils"
version = "1.0.0-beta.11"
version = "1.0.0-beta.12"
dependencies = [
"rustfs-data-usage",
"rustfs-ecstore",
@@ -10063,7 +10064,7 @@ dependencies = [
[[package]]
name = "rustfs-tls-runtime"
version = "1.0.0-beta.11"
version = "1.0.0-beta.12"
dependencies = [
"arc-swap",
"metrics",
@@ -10083,11 +10084,11 @@ dependencies = [
[[package]]
name = "rustfs-trusted-proxies"
version = "1.0.0-beta.11"
version = "1.0.0-beta.12"
dependencies = [
"async-trait",
"axum",
"http 1.4.2",
"http 1.5.0",
"ipnetwork",
"metrics",
"moka",
@@ -10119,7 +10120,7 @@ dependencies = [
[[package]]
name = "rustfs-utils"
version = "1.0.0-beta.11"
version = "1.0.0-beta.12"
dependencies = [
"base64-simd",
"blake2",
@@ -10133,7 +10134,7 @@ dependencies = [
"hex-simd",
"highway",
"hmac 0.13.0",
"http 1.4.2",
"http 1.5.0",
"hyper",
"local-ip-address",
"lz4",
@@ -10160,7 +10161,7 @@ dependencies = [
[[package]]
name = "rustfs-zip"
version = "1.0.0-beta.11"
version = "1.0.0-beta.12"
dependencies = [
"astral-tokio-tar",
"async-compression",
@@ -10190,7 +10191,7 @@ dependencies = [
"anyhow",
"async-trait",
"bytes",
"http 1.4.2",
"http 1.5.0",
"reqwest",
"rustify_derive",
"serde",
@@ -10230,9 +10231,9 @@ dependencies = [
[[package]]
name = "rustls"
version = "0.23.42"
version = "0.23.43"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3c54fcab019b409d04215d3a17cb438fd7fbf192ee61461f20f4fe18704bc138"
checksum = "0283386ce02abc0151e1761d08802dfe86c173b0b494af5cbc086574e453da06"
dependencies = [
"aws-lc-rs",
"log",
@@ -10370,7 +10371,7 @@ dependencies = [
"futures",
"hex-simd",
"hmac 0.13.0",
"http 1.4.2",
"http 1.5.0",
"http-body 1.1.0",
"http-body-util",
"httparse",
@@ -11498,7 +11499,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd"
dependencies = [
"fastrand",
"getrandom 0.4.3",
"getrandom 0.3.4",
"once_cell",
"rustix",
"windows-sys 0.61.2",
@@ -11734,13 +11735,13 @@ dependencies = [
[[package]]
name = "tokio-macros"
version = "2.7.1"
version = "2.7.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6328af13490e73a9b4694030fafd93f8c8c6a9dede33e821c3fc63eddf8042ba"
checksum = "78773a2a397f451582ce068015985c33193cf6dea8b74d2a639fe457b2f07b0e"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.119",
"syn 3.0.3",
]
[[package]]
@@ -11844,7 +11845,7 @@ dependencies = [
"bytes",
"futures-core",
"futures-sink",
"http 1.4.2",
"http 1.5.0",
"httparse",
"rand 0.8.7",
"rustls-pki-types",
@@ -11896,7 +11897,7 @@ dependencies = [
"bytes",
"flate2",
"h2",
"http 1.4.2",
"http 1.5.0",
"http-body 1.1.0",
"http-body-util",
"hyper",
@@ -11983,7 +11984,7 @@ dependencies = [
"bitflags 2.13.1",
"bytes",
"futures-util",
"http 1.4.2",
"http 1.5.0",
"http-body 1.1.0",
"pin-project-lite",
"tower",
@@ -12003,7 +12004,7 @@ dependencies = [
"bytes",
"futures-core",
"futures-util",
"http 1.4.2",
"http 1.5.0",
"http-body 1.1.0",
"http-body-util",
"percent-encoding",
@@ -12176,7 +12177,7 @@ checksum = "6c01152af293afb9c7c2a57e4b559c5620b421f6d133261c60dd2d0cdb38e6b8"
dependencies = [
"bytes",
"data-encoding",
"http 1.4.2",
"http 1.5.0",
"httparse",
"log",
"rand 0.9.5",
@@ -12363,7 +12364,7 @@ checksum = "30ffcc0e81025065dda612ec1e26a3d81bb16ef3062354873d17a35965d68522"
dependencies = [
"async-trait",
"derive_builder",
"http 1.4.2",
"http 1.5.0",
"reqwest",
"rustify",
"rustify_derive",
+52 -51
View File
@@ -69,7 +69,7 @@ edition = "2024"
license = "Apache-2.0"
repository = "https://github.com/rustfs/rustfs"
rust-version = "1.97.1"
version = "1.0.0-beta.11"
version = "1.0.0-beta.12"
homepage = "https://rustfs.com"
description = "RustFS is a high-performance distributed object storage software built using Rust, one of the most popular languages worldwide. "
keywords = ["RustFS", "Minio", "object-storage", "filesystem", "s3"]
@@ -86,58 +86,58 @@ redundant_clone = "warn"
[workspace.dependencies]
# RustFS Internal Crates
rustfs = { path = "./rustfs", version = "1.0.0-beta.11" }
rustfs-heal = { path = "crates/heal", version = "1.0.0-beta.11" }
rustfs-audit = { path = "crates/audit", version = "1.0.0-beta.11" }
rustfs-checksums = { path = "crates/checksums", version = "1.0.0-beta.11" }
rustfs-common = { path = "crates/common", version = "1.0.0-beta.11" }
rustfs-data-usage = { path = "crates/data-usage", version = "1.0.0-beta.11" }
rustfs-config = { path = "./crates/config", version = "1.0.0-beta.11" }
rustfs-concurrency = { path = "./crates/concurrency", version = "1.0.0-beta.11" }
rustfs-credentials = { path = "crates/credentials", version = "1.0.0-beta.11" }
rustfs-crypto = { path = "crates/crypto", version = "1.0.0-beta.11" }
rustfs-ecstore = { path = "crates/ecstore", version = "1.0.0-beta.11" }
rustfs-filemeta = { path = "crates/filemeta", version = "1.0.0-beta.11" }
rustfs-iam = { path = "crates/iam", version = "1.0.0-beta.11" }
rustfs-keystone = { path = "crates/keystone", version = "1.0.0-beta.11" }
rustfs-lifecycle = { path = "crates/lifecycle", version = "1.0.0-beta.11" }
rustfs-kms = { path = "crates/kms", version = "1.0.0-beta.11" }
rustfs-lock = { path = "crates/lock", version = "1.0.0-beta.11" }
rustfs-madmin = { path = "crates/madmin", version = "1.0.0-beta.11" }
rustfs-notify = { path = "crates/notify", version = "1.0.0-beta.11" }
rustfs-io-metrics = { path = "crates/io-metrics", version = "1.0.0-beta.11" }
rustfs-io-core = { path = "crates/io-core", version = "1.0.0-beta.11" }
rustfs-object-capacity = { path = "crates/object-capacity", version = "1.0.0-beta.11" }
rustfs-object-data-cache = { path = "crates/object-data-cache", version = "1.0.0-beta.11" }
rustfs-log-analyzer = { path = "crates/log-analyzer", version = "1.0.0-beta.11" }
rustfs-obs = { path = "crates/obs", version = "1.0.0-beta.11" }
rustfs-policy = { path = "crates/policy", version = "1.0.0-beta.11" }
rustfs-protos = { path = "crates/protos", version = "1.0.0-beta.11" }
rustfs-protocols = { path = "crates/protocols", version = "1.0.0-beta.11" }
rustfs-replication = { path = "crates/replication", version = "1.0.0-beta.11" }
rustfs-rio = { path = "crates/rio", version = "1.0.0-beta.11" }
rustfs-rio-v2 = { path = "crates/rio-v2", version = "1.0.0-beta.11" }
rustfs-s3-types = { path = "crates/s3-types", version = "1.0.0-beta.11" }
rustfs-s3-ops = { path = "crates/s3-ops", version = "1.0.0-beta.11" }
rustfs-s3select-api = { path = "crates/s3select-api", version = "1.0.0-beta.11" }
rustfs-s3select-query = { path = "crates/s3select-query", version = "1.0.0-beta.11" }
rustfs-scanner = { path = "crates/scanner", version = "1.0.0-beta.11" }
rustfs-security-governance = { path = "crates/security-governance", version = "1.0.0-beta.11" }
rustfs-extension-schema = { path = "crates/extension-schema", version = "1.0.0-beta.11" }
rustfs-signer = { path = "crates/signer", version = "1.0.0-beta.11" }
rustfs-storage-api = { path = "crates/storage-api", version = "1.0.0-beta.11" }
rustfs-trusted-proxies = { path = "crates/trusted-proxies", version = "1.0.0-beta.11" }
rustfs-targets = { path = "crates/targets", version = "1.0.0-beta.11" }
rustfs-test-utils = { path = "crates/test-utils", version = "1.0.0-beta.11" }
rustfs-tls-runtime = { path = "crates/tls-runtime", version = "1.0.0-beta.11" }
rustfs-utils = { path = "crates/utils", version = "1.0.0-beta.11" }
rustfs-zip = { path = "./crates/zip", version = "1.0.0-beta.11" }
rustfs = { path = "./rustfs", version = "1.0.0-beta.12" }
rustfs-heal = { path = "crates/heal", version = "1.0.0-beta.12" }
rustfs-audit = { path = "crates/audit", version = "1.0.0-beta.12" }
rustfs-checksums = { path = "crates/checksums", version = "1.0.0-beta.12" }
rustfs-common = { path = "crates/common", version = "1.0.0-beta.12" }
rustfs-data-usage = { path = "crates/data-usage", version = "1.0.0-beta.12" }
rustfs-config = { path = "./crates/config", version = "1.0.0-beta.12" }
rustfs-concurrency = { path = "./crates/concurrency", version = "1.0.0-beta.12" }
rustfs-credentials = { path = "crates/credentials", version = "1.0.0-beta.12" }
rustfs-crypto = { path = "crates/crypto", version = "1.0.0-beta.12" }
rustfs-ecstore = { path = "crates/ecstore", version = "1.0.0-beta.12" }
rustfs-filemeta = { path = "crates/filemeta", version = "1.0.0-beta.12" }
rustfs-iam = { path = "crates/iam", version = "1.0.0-beta.12" }
rustfs-keystone = { path = "crates/keystone", version = "1.0.0-beta.12" }
rustfs-lifecycle = { path = "crates/lifecycle", version = "1.0.0-beta.12" }
rustfs-kms = { path = "crates/kms", version = "1.0.0-beta.12" }
rustfs-lock = { path = "crates/lock", version = "1.0.0-beta.12" }
rustfs-madmin = { path = "crates/madmin", version = "1.0.0-beta.12" }
rustfs-notify = { path = "crates/notify", version = "1.0.0-beta.12" }
rustfs-io-metrics = { path = "crates/io-metrics", version = "1.0.0-beta.12" }
rustfs-io-core = { path = "crates/io-core", version = "1.0.0-beta.12" }
rustfs-object-capacity = { path = "crates/object-capacity", version = "1.0.0-beta.12" }
rustfs-object-data-cache = { path = "crates/object-data-cache", version = "1.0.0-beta.12" }
rustfs-log-analyzer = { path = "crates/log-analyzer", version = "1.0.0-beta.12" }
rustfs-obs = { path = "crates/obs", version = "1.0.0-beta.12" }
rustfs-policy = { path = "crates/policy", version = "1.0.0-beta.12" }
rustfs-protos = { path = "crates/protos", version = "1.0.0-beta.12" }
rustfs-protocols = { path = "crates/protocols", version = "1.0.0-beta.12" }
rustfs-replication = { path = "crates/replication", version = "1.0.0-beta.12" }
rustfs-rio = { path = "crates/rio", version = "1.0.0-beta.12" }
rustfs-rio-v2 = { path = "crates/rio-v2", version = "1.0.0-beta.12" }
rustfs-s3-types = { path = "crates/s3-types", version = "1.0.0-beta.12" }
rustfs-s3-ops = { path = "crates/s3-ops", version = "1.0.0-beta.12" }
rustfs-s3select-api = { path = "crates/s3select-api", version = "1.0.0-beta.12" }
rustfs-s3select-query = { path = "crates/s3select-query", version = "1.0.0-beta.12" }
rustfs-scanner = { path = "crates/scanner", version = "1.0.0-beta.12" }
rustfs-security-governance = { path = "crates/security-governance", version = "1.0.0-beta.12" }
rustfs-extension-schema = { path = "crates/extension-schema", version = "1.0.0-beta.12" }
rustfs-signer = { path = "crates/signer", version = "1.0.0-beta.12" }
rustfs-storage-api = { path = "crates/storage-api", version = "1.0.0-beta.12" }
rustfs-trusted-proxies = { path = "crates/trusted-proxies", version = "1.0.0-beta.12" }
rustfs-targets = { path = "crates/targets", version = "1.0.0-beta.12" }
rustfs-test-utils = { path = "crates/test-utils", version = "1.0.0-beta.12" }
rustfs-tls-runtime = { path = "crates/tls-runtime", version = "1.0.0-beta.12" }
rustfs-utils = { path = "crates/utils", version = "1.0.0-beta.12" }
rustfs-zip = { path = "./crates/zip", version = "1.0.0-beta.12" }
# Async Runtime and Networking
async-channel = "2.5.0"
async_zip = { default-features = false, version = "0.0.18" }
mysql_async = { default-features = false, version = "0.37" }
async-compression = { version = "0.4.42" }
async-compression = { version = "0.4.43" }
async-recursion = "1.1.1"
async-trait = "0.1.91"
async-nats = { version = "0.50.0", default-features = false }
@@ -152,7 +152,7 @@ lapin = { default-features = false, version = "4.10.0" }
hyper = { version = "1.11.0" }
hyper-rustls = { default-features = false, version = "0.27.9" }
hyper-util = { version = "0.1.20" }
http = "1.4.2"
http = "1.5.0"
http-body = "1.1.0"
http-body-util = "0.1.4"
minlz = "1.2.3"
@@ -200,7 +200,7 @@ jsonwebtoken = { version = "11.0.0" }
openidconnect = { default-features = false, version = "4.0" }
pbkdf2 = "0.13.0"
rsa = { version = "=0.10.0-rc.18" }
rustls = { default-features = false, version = "0.23.42" }
rustls = { default-features = false, version = "0.23.43" }
rustls-native-certs = "0.8"
rustls-pki-types = "1.15.1"
sha1 = "0.11.0"
@@ -256,6 +256,7 @@ hashbrown = { version = "0.17.1" }
hex = "0.4.3"
hex-simd = "0.8.0"
highway = { version = "1.3.0" }
hostname = "0.4.2"
ipnetwork = { version = "0.21.1" }
lazy_static = "1.5.0"
libc = "0.2.189"
@@ -283,7 +284,7 @@ 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.4.1" }
redis = { version = "1.5.0" }
rustix = { version = "1.1.4" }
rust-embed = { version = "8.12.0" }
rustc-hash = { version = "2.1.3" }
+1 -1
View File
@@ -116,7 +116,7 @@ chown -R 10001:10001 data logs
docker run -d -p 9000:9000 -p 9001:9001 -v $(pwd)/data:/data -v $(pwd)/logs:/logs rustfs/rustfs:latest
# Using specific version
docker run -d -p 9000:9000 -p 9001:9001 -v $(pwd)/data:/data -v $(pwd)/logs:/logs rustfs/rustfs:1.0.0-beta.11
docker run -d -p 9000:9000 -p 9001:9001 -v $(pwd)/data:/data -v $(pwd)/logs:/logs rustfs/rustfs:1.0.0-beta.12
```
If you use [podman](https://github.com/containers/podman) instead of docker, you can install the RustFS with the below command
+1 -1
View File
@@ -113,7 +113,7 @@ chown -R 10001:10001 data logs
docker run -d -p 9000:9000 -p 9001:9001 -v $(pwd)/data:/data -v $(pwd)/logs:/logs rustfs/rustfs:latest
# 使用指定版本运行
docker run -d -p 9000:9000 -p 9001:9001 -v $(pwd)/data:/data -v $(pwd)/logs:/logs rustfs/rustfs:1.0.0-beta.11
docker run -d -p 9000:9000 -p 9001:9001 -v $(pwd)/data:/data -v $(pwd)/logs:/logs rustfs/rustfs:1.0.0-beta.12
```
如果您通过绑定挂载启用 TLS 证书目录,也请用同样方式准备该目录:
+4
View File
@@ -66,6 +66,10 @@ Current guidance:
- `RUSTFS_BROWSER_REDIRECT_URL` sets the externally reachable browser origin used for OIDC callback, console success redirect, and logout fallback URLs. Configure it to the public scheme and authority without a path, for example `https://console.example.com`. In load-balancer deployments, keep OIDC authorize and callback requests on the same backend node because the in-flight OIDC `state` is local to the RustFS node.
## Distributed endpoint locality
- `RUSTFS_LOCAL_ENDPOINT_HOST` identifies this server's host in a distributed `RUSTFS_VOLUMES` topology without resolving every peer during startup. Set it to exactly one host, without a scheme, port, or path. It is accepted only for orchestrated URL topologies and must match at least one endpoint on the RustFS server port; invalid or unmatched values fail startup. Leave it unset to retain DNS-based locality discovery.
## Scanner environment aliases
- `RUSTFS_SCANNER_SPEED` (canonical, also accepts `MINIO_SCANNER_SPEED`)
+4
View File
@@ -131,6 +131,10 @@ pub const ENV_RUSTFS_ADDRESS: &str = "RUSTFS_ADDRESS";
/// Environment variable for server volumes.
pub const ENV_RUSTFS_VOLUMES: &str = "RUSTFS_VOLUMES";
/// Environment variable identifying this server's host in distributed endpoint
/// lists without relying on DNS locality discovery.
pub const ENV_LOCAL_ENDPOINT_HOST: &str = "RUSTFS_LOCAL_ENDPOINT_HOST";
/// Environment variable to explicitly bypass local physical disk independence checks.
pub const ENV_UNSAFE_BYPASS_DISK_CHECK: &str = "RUSTFS_UNSAFE_BYPASS_DISK_CHECK";
@@ -3237,6 +3237,116 @@ async fn test_bucket_replication_converges_delete_marker_and_version_purge() ->
.await?;
assert_eq!(retained.body.collect().await?.into_bytes().as_ref(), b"versioned replication payload v2");
source_client
.delete_object()
.bucket(source_bucket)
.key(object_key)
.version_id(delete_marker_version_id)
.send()
.await?;
assert_replication_converged(&source_client, source_bucket, &target_client, target_bucket).await?;
let target_state = list_replication_state(&target_client, target_bucket).await?;
assert!(
target_state.iter().all(|entry| entry.version_id != delete_marker_version_id),
"target retained the explicitly purged delete-marker version"
);
assert!(
target_state
.iter()
.any(|entry| !entry.delete_marker && entry.version_id == retained_version_id),
"target removed the retained object version while purging the delete marker: {target_state:?}"
);
Ok(())
}
#[tokio::test]
#[serial]
async fn test_bucket_replication_disabled_version_delete_preserves_target_versions_issue_5442() -> TestResult {
init_logging();
let mut source_env = RustFSTestEnvironment::new().await?;
let mut source_env_vars = replication_fast_env();
source_env_vars.extend_from_slice(LOOPBACK_REPLICATION_TARGET_ENV);
source_env.start_rustfs_server_with_env(vec![], &source_env_vars).await?;
let mut target_env = RustFSTestEnvironment::new().await?;
target_env.start_rustfs_server_without_cleanup(vec![]).await?;
let source_bucket = "replication-no-version-delete-src";
let target_bucket = "replication-no-version-delete-dst";
let object_key = "retained-versions.txt";
let source_client = source_env.create_s3_client();
let target_client = target_env.create_s3_client();
source_client.create_bucket().bucket(source_bucket).send().await?;
target_client.create_bucket().bucket(target_bucket).send().await?;
enable_bucket_versioning(&source_env, source_bucket).await?;
enable_bucket_versioning(&target_env, target_bucket).await?;
let target_arn = set_replication_target(&source_env, source_bucket, &target_env, target_bucket).await?;
put_bucket_replication_with_delete_statuses(&source_env, source_bucket, &target_arn, "Enabled", None).await?;
let put = source_client
.put_object()
.bucket(source_bucket)
.key(object_key)
.body(ByteStream::from_static(b"permanent delete replication disabled"))
.send()
.await?;
let object_version_id = put.version_id().ok_or("source PUT omitted version ID")?.to_string();
assert_replication_converged(&source_client, source_bucket, &target_client, target_bucket).await?;
let delete = source_client
.delete_object()
.bucket(source_bucket)
.key(object_key)
.send()
.await?;
let delete_marker_version_id = delete
.version_id()
.ok_or("source DELETE omitted marker version ID")?
.to_string();
assert_eq!(delete.delete_marker(), Some(true));
assert_replication_converged(&source_client, source_bucket, &target_client, target_bucket).await?;
let expected_target_state = list_replication_state(&target_client, target_bucket).await?;
assert_eq!(expected_target_state.len(), 2);
assert!(
expected_target_state
.iter()
.any(|entry| !entry.delete_marker && entry.version_id == object_version_id)
);
assert!(
expected_target_state
.iter()
.any(|entry| entry.delete_marker && entry.version_id == delete_marker_version_id)
);
for version_id in [&object_version_id, &delete_marker_version_id] {
source_client
.delete_object()
.bucket(source_bucket)
.key(object_key)
.version_id(version_id)
.send()
.await?;
}
assert!(list_replication_state(&source_client, source_bucket).await?.is_empty());
let observation_deadline = tokio::time::Instant::now() + Duration::from_secs(10);
loop {
let target_state = list_replication_state(&target_client, target_bucket).await?;
assert_eq!(
target_state, expected_target_state,
"disabled permanent-delete replication changed target versions"
);
if tokio::time::Instant::now() >= observation_deadline {
break;
}
sleep(Duration::from_millis(100)).await;
}
Ok(())
}
+1
View File
@@ -105,6 +105,7 @@ tempfile.workspace = true
hyper = { workspace = true, features = ["http2", "http1", "server"] }
hyper-util = { workspace = true, features = ["tokio", "server-auto", "server-graceful", "tracing"] }
hyper-rustls = { workspace = true, default-features = false, features = ["native-tokio", "http1", "tls12", "logging", "http2", "aws-lc-rs"] }
hostname.workspace = true
rustls = { workspace = true, default-features = false, features = ["aws-lc-rs", "logging", "tls12", "prefer-post-quantum", "std"] }
rustls-pki-types.workspace = true
tokio = { workspace = true, features = ["io-util", "sync", "signal", "fs", "rt-multi-thread"] }
+2 -2
View File
@@ -184,8 +184,8 @@ pub mod bucket {
get_global_replication_stats, init_background_replication, read_durable_mrf_backlog, replication_state_to_filemeta,
replication_status_to_filemeta, replication_statuses_map, replication_target_arns, resync_start_conflict_id,
should_remove_replication_target, should_schedule_delete_replication, should_use_existing_delete_replication_info,
should_use_existing_delete_replication_source, unsupported_replication_config_field,
validate_replication_config_target_arns, version_purge_status_to_filemeta,
should_use_existing_delete_replication_source, validate_replication_config_target_arns,
version_purge_status_to_filemeta,
};
}
@@ -1278,6 +1278,12 @@ fn build_remove_object_headers(version_id: Option<&str>, opts: &RemoveObjectOpti
fn resolve_delete_api_version_id(version_id: Option<String>, opts: &RemoveObjectOptions) -> Option<String> {
if opts.replication_request && opts.replication_delete_marker {
None
} else if opts.replication_request
&& version_id
.as_deref()
.is_some_and(|version_id| Uuid::parse_str(version_id).is_ok_and(|version_id| version_id.is_nil()))
{
Some("null".to_string())
} else {
version_id
}
@@ -2259,6 +2265,12 @@ mod tests {
assert_eq!(got.as_deref(), Some(vid.as_str()));
}
#[test]
fn null_version_purge_sends_s3_null_version_id() {
let got = resolve_delete_api_version_id(Some(Uuid::nil().to_string()), &remove_opts(true, false));
assert_eq!(got.as_deref(), Some("null"));
}
#[test]
fn delete_marker_propagation_omits_versionid_query_param() {
// Propagating a delete-marker CREATION (delete_marker=true): the target
+1 -1
View File
@@ -46,7 +46,7 @@ mod runtime_boundary;
pub use datatypes::ResyncStatusType;
pub use replication_config_boundary::{
ObjectOpts, ReplicationConfigurationExt, ReplicationTargetValidationError, replication_target_arns,
should_remove_replication_target, unsupported_replication_config_field, validate_replication_config_target_arns,
should_remove_replication_target, validate_replication_config_target_arns,
};
#[cfg(test)]
pub(crate) use replication_filemeta_boundary::ReplicateTargetDecision;
@@ -13,6 +13,6 @@
// limitations under the License.
pub use rustfs_replication::{
ObjectOpts, ReplicationConfigurationExt, ReplicationTargetValidationError, replication_target_arns,
should_remove_replication_target, unsupported_replication_config_field, validate_replication_config_target_arns,
ObjectOpts, ReplicationConfigurationExt, ReplicationTargetValidationError, delete_replication_target_arns,
replication_target_arns, should_remove_replication_target, validate_replication_config_target_arns,
};
@@ -14,11 +14,8 @@
use std::{collections::HashMap, sync::Arc};
use super::replication_error_boundary::Result;
use super::replication_filemeta_boundary::{ReplicateDecision, ReplicationStatusType, ReplicationType};
use super::replication_object_config::{
check_replicate_delete, check_replicate_delete_strict, get_must_replicate_options, must_replicate,
};
use super::replication_object_config::{check_replicate_delete, get_must_replicate_options, must_replicate};
use super::replication_object_decision_boundary::MustReplicateOptions;
use super::replication_pool::{schedule_replication, schedule_replication_delete};
use super::replication_queue_boundary::DeletedObjectReplicationInfo;
@@ -53,16 +50,6 @@ impl ReplicationObjectBridge {
check_replicate_delete(bucket, object, source, opts, get_error).await
}
pub async fn check_delete_strict(
bucket: &str,
object: &ObjectToDelete,
source: &ObjectInfo,
opts: &ObjectOptions,
get_error: Option<String>,
) -> Result<ReplicateDecision> {
check_replicate_delete_strict(bucket, object, source, opts, get_error).await
}
pub async fn schedule_object<S: ReplicationStorage>(
object: ObjectInfo,
storage: Arc<S>,
@@ -28,7 +28,7 @@ use super::replication_logging::{EVENT_RESYNC_CONFIG_LOOKUP_SKIPPED, LOG_COMPONE
use super::replication_metadata_boundary::ReplicationMetadataStore;
use super::replication_object_decision_boundary::{
MustReplicateOptions, ReplicationDeleteSource, ReplicationResyncTargetObject, delete_replication_missing_source_decision,
delete_replication_object_opts, resync_target_for_object,
delete_replication_object_opts, delete_replication_version_id, resync_target_for_object,
};
use super::replication_storage_boundary::{ObjectInfo, ObjectOptions, ObjectToDelete, object_to_delete_for_replication};
use super::replication_target_boundary::{BucketTargets, ReplicationTargetStore};
@@ -73,7 +73,7 @@ impl ReplicationConfig {
if oi.delete_marker {
let opts = ObjectOpts {
name: oi.name.clone(),
version_id: oi.version_id,
version_id: delete_replication_version_id(oi.delete_marker, oi.version_id, !oi.version_purge_status.is_empty()),
delete_marker: true,
op_type: ReplicationType::Delete,
existing_object: true,
@@ -169,8 +169,9 @@ pub(crate) async fn check_replicate_delete(
del_opts: &ObjectOptions,
gerr: Option<String>,
) -> ReplicateDecision {
match check_replicate_delete_strict(bucket, dobj, oi, del_opts, gerr).await {
Ok(decision) => decision,
let rcfg = match get_replication_config(bucket).await {
Ok(Some(config)) => config,
Ok(None) => return ReplicateDecision::default(),
Err(err) => {
error!(
event = EVENT_RESYNC_CONFIG_LOOKUP_SKIPPED,
@@ -181,30 +182,16 @@ pub(crate) async fn check_replicate_delete(
error = %err,
"Failed to look up replication config for delete replication"
);
ReplicateDecision::default()
return ReplicateDecision::default();
}
}
}
pub(crate) async fn check_replicate_delete_strict(
bucket: &str,
dobj: &ObjectToDelete,
oi: &ObjectInfo,
del_opts: &ObjectOptions,
gerr: Option<String>,
) -> Result<ReplicateDecision> {
let rcfg = match get_replication_config(bucket).await {
Ok(Some(config)) => config,
Ok(None) => return Ok(ReplicateDecision::default()),
Err(err) => return Err(err),
};
if del_opts.replication_request {
return Ok(ReplicateDecision::default());
return ReplicateDecision::default();
}
if !del_opts.versioned && !del_opts.version_suspended {
return Ok(ReplicateDecision::default());
if !del_opts.versioned {
return ReplicateDecision::default();
}
let replication_delete = object_to_delete_for_replication(dobj);
@@ -222,7 +209,7 @@ pub(crate) async fn check_replicate_delete_strict(
let mut dsc = ReplicateDecision::new();
if tgt_arns.is_empty() {
return Ok(dsc);
return dsc;
}
for tgt_arn in tgt_arns {
@@ -252,7 +239,7 @@ pub(crate) async fn check_replicate_delete_strict(
dsc.set(tgt_dsc);
}
Ok(dsc)
dsc
}
pub(crate) async fn must_replicate(bucket: &str, object: &str, mopts: MustReplicateOptions) -> ReplicateDecision {
@@ -312,8 +299,14 @@ pub(crate) async fn must_replicate(bucket: &str, object: &str, mopts: MustReplic
#[cfg(test)]
mod tests {
use s3s::dto::{Destination, ReplicationRule, ReplicationRuleStatus};
use s3s::dto::{
DeleteMarkerReplication, DeleteMarkerReplicationStatus, DeleteReplication, DeleteReplicationStatus, Destination,
ReplicationRule, ReplicationRuleStatus,
};
use uuid::Uuid;
use super::super::replication_filemeta_boundary::VersionPurgeStatusType;
use super::super::replication_target_boundary::BucketTarget;
use super::*;
fn replication_rule() -> ReplicationRule {
@@ -373,4 +366,56 @@ mod tests {
assert!(options.is_replication_request());
assert_eq!(options.user_tags(), "env=prod");
}
#[tokio::test]
async fn resync_distinguishes_delete_marker_creation_from_version_purge() {
let arn = "arn:rustfs:replication:us-east-1:target:bucket";
let mut rule = replication_rule();
rule.destination.bucket = arn.to_string();
rule.delete_marker_replication = Some(DeleteMarkerReplication {
status: Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::ENABLED)),
});
rule.delete_replication = Some(DeleteReplication {
status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::DISABLED),
});
let config = ReplicationConfig::new(
Some(ReplicationConfiguration {
role: String::new(),
rules: vec![rule],
}),
Some(BucketTargets {
targets: vec![BucketTarget {
arn: arn.to_string(),
endpoint: "target.example".to_string(),
..Default::default()
}],
}),
);
let marker_version_id = Uuid::new_v4();
let marker = ObjectInfo {
bucket: "source".to_string(),
name: "object".to_string(),
delete_marker: true,
version_id: Some(marker_version_id),
replication_status: ReplicationStatusType::Pending,
..Default::default()
};
let creation = config
.resync(marker.clone(), ReplicateDecision::default(), &HashMap::new())
.await;
assert!(creation.targets.contains_key(arn));
let purge = config
.resync(
ObjectInfo {
version_purge_status: VersionPurgeStatusType::Pending,
..marker
},
ReplicateDecision::default(),
&HashMap::new(),
)
.await;
assert!(purge.targets.is_empty());
}
}
@@ -13,14 +13,14 @@
// limitations under the License.
pub use rustfs_replication::{
MustReplicateOptions, ReplicationDeleteScheduleInput, ReplicationDeleteStateSource, delete_replication_state_from_config,
delete_replication_version_id, should_schedule_delete_replication, should_use_existing_delete_replication_info,
should_use_existing_delete_replication_source,
MustReplicateOptions, ReplicationDeleteScheduleInput, ReplicationDeleteStateSource, delete_replication_parts,
delete_replication_state_from_config, delete_replication_version_id, should_schedule_delete_replication,
should_use_existing_delete_replication_info, should_use_existing_delete_replication_source,
};
pub(crate) use rustfs_replication::{
ReplicationDeleteSource, ReplicationMultipartPartInput, ReplicationResyncTargetObject,
delete_replication_missing_source_decision, delete_replication_object_opts, heal_uses_delete_replication_path,
is_retryable_delete_replication_head_error, is_version_delete_replication, replication_etags_match,
replication_multipart_complete_actual_size, replication_multipart_part_plan, resync_target_for_object,
should_retry_delete_marker_purge,
should_retry_delete_marker_purge, version_purge_target_missing,
};
@@ -67,6 +67,7 @@ const EVENT_REPLICATION_BACKPRESSURE: &str = "replication_backpressure";
const EVENT_REPLICATION_RESYNC_LOAD_SKIPPED: &str = "replication_resync_load_skipped";
const EVENT_REPLICATION_RESYNC_RECOVERED: &str = "replication_resync_recovered";
const EVENT_REPLICATION_MRF_QUEUE_UNAVAILABLE: &str = "replication_mrf_queue_unavailable";
const EVENT_REPLICATION_MRF_ENTRY_SKIPPED: &str = "replication_mrf_entry_skipped";
#[derive(Debug, Default)]
pub struct DurableMrfBacklog {
@@ -660,6 +661,21 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
for entry in entries.iter() {
match entry.op {
MrfOpKind::Delete => {
let Some(delete_parts) = entry.delete_parts_for_replay() else {
debug!(
event = EVENT_REPLICATION_MRF_ENTRY_SKIPPED,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REPLICATION,
bucket = %entry.bucket,
object = %entry.object,
reason = "invalid_delete_version_ids",
"Skipped invalid persisted replication delete"
);
continue;
};
let version_purge_id = delete_parts
.version_id
.or_else(|| delete_parts.delete_marker_version_id.filter(|_| !delete_parts.delete_marker));
// Reconstruct a heal delete and re-queue it. We do NOT call
// get_object_info here because the delete-marker or version may
// already be absent from the local store — that is expected.
@@ -674,15 +690,20 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
let oi = ObjectInfo {
bucket: entry.bucket.clone(),
name: entry.object.clone(),
version_id: entry.version_id,
delete_marker: entry.delete_marker,
version_id: version_purge_id,
delete_marker: delete_parts.delete_marker,
replication_status: if entry.replica {
ReplicationStatusType::Replica
} else {
ReplicationStatusType::Empty
},
..Default::default()
};
let dsc = check_replicate_delete(
&entry.bucket,
&ObjectToDelete {
object_name: entry.object.clone(),
version_id: entry.version_id,
version_id: version_purge_id,
..Default::default()
},
&oi,
@@ -708,9 +729,9 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
let dv = DeletedObjectReplicationInfo {
delete_object: ReplicationDeletedObject {
object_name: entry.object.clone(),
version_id: entry.version_id,
delete_marker_version_id: entry.delete_marker_version_id,
delete_marker: entry.delete_marker,
version_id: delete_parts.version_id,
delete_marker_version_id: delete_parts.delete_marker_version_id,
delete_marker: delete_parts.delete_marker,
delete_marker_mtime,
replication_state: Some(rstate),
..Default::default()
@@ -2333,6 +2354,7 @@ mod tests {
op: MrfOpKind::Object,
delete_marker_version_id: None,
delete_marker: false,
replica: false,
delete_marker_mtime: None,
};
let second = MrfReplicateEntry {
@@ -2446,6 +2468,7 @@ mod tests {
op: MrfOpKind::Object,
delete_marker_version_id: None,
delete_marker: false,
replica: false,
delete_marker_mtime: None,
};
@@ -2479,6 +2502,7 @@ mod tests {
op: MrfOpKind::Delete,
delete_marker_version_id: Some(dm_vid),
delete_marker: true,
replica: false,
delete_marker_mtime: Some(mtime_nanos),
};
@@ -2512,6 +2536,7 @@ mod tests {
op: MrfOpKind::Delete,
delete_marker_version_id: None,
delete_marker: false,
replica: false,
delete_marker_mtime: None,
};
@@ -2540,6 +2565,7 @@ mod tests {
op: MrfOpKind::Object,
delete_marker_version_id: None,
delete_marker: false,
replica: false,
delete_marker_mtime: None,
},
MrfReplicateEntry {
@@ -2551,6 +2577,7 @@ mod tests {
op: MrfOpKind::Delete,
delete_marker_version_id: Some(del_dm_vid),
delete_marker: true,
replica: false,
delete_marker_mtime: None,
},
];
@@ -2580,6 +2607,7 @@ mod tests {
op: MrfOpKind::Object,
delete_marker_version_id: None,
delete_marker: false,
replica: false,
delete_marker_mtime: None,
};
assert_eq!(obj_entry.op, MrfOpKind::Object);
@@ -2594,6 +2622,7 @@ mod tests {
op: MrfOpKind::Delete,
delete_marker_version_id: Some(Uuid::new_v4()),
delete_marker: true,
replica: false,
delete_marker_mtime: None,
};
assert_eq!(del_entry.op, MrfOpKind::Delete);
@@ -2609,6 +2638,7 @@ mod tests {
op: MrfOpKind::default(),
delete_marker_version_id: None,
delete_marker: false,
replica: false,
delete_marker_mtime: None,
};
assert_eq!(legacy_entry.op, MrfOpKind::Object, "legacy default must be Object");
@@ -2672,6 +2702,7 @@ mod tests {
op: MrfOpKind::Object,
delete_marker_version_id: None,
delete_marker: false,
replica: false,
delete_marker_mtime: None,
}];
let encoded = encode_mrf_file(&entries).expect("durable MRF backlog should encode");
@@ -2702,6 +2733,7 @@ mod tests {
op: MrfOpKind::Object,
delete_marker_version_id: None,
delete_marker: false,
replica: false,
delete_marker_mtime: None,
}])
.expect("invalid persisted entry should still encode for boundary testing");
@@ -13,7 +13,7 @@
// limitations under the License.
use super::replication_bandwidth_boundary;
use super::replication_config_boundary::{ObjectOpts, ReplicationConfigurationExt as _};
use super::replication_config_boundary::{ObjectOpts, ReplicationConfigurationExt as _, delete_replication_target_arns};
use super::replication_config_store::ReplicationConfigStore;
use super::replication_error_boundary::{Result, is_err_object_not_found, is_err_version_not_found};
use super::replication_event_sink::{EventArgs, send_event, send_local_event};
@@ -29,9 +29,10 @@ use super::replication_metadata_boundary::ReplicationMetadataStore;
use super::replication_msgp_boundary::ReplicationMsgpCodec;
use super::replication_object_config::{ReplicationConfig, check_replicate_delete, get_replication_config, must_replicate};
use super::replication_object_decision_boundary::{
MustReplicateOptions, ReplicationMultipartPartInput, heal_uses_delete_replication_path,
MustReplicateOptions, ReplicationMultipartPartInput, delete_replication_parts, heal_uses_delete_replication_path,
is_retryable_delete_replication_head_error, is_version_delete_replication, replication_etags_match,
replication_multipart_complete_actual_size, replication_multipart_part_plan, should_retry_delete_marker_purge,
version_purge_target_missing,
};
use super::replication_queue_boundary::DeletedObjectReplicationInfo;
use super::replication_resync_boundary::ResyncStatusType;
@@ -49,7 +50,7 @@ use super::replication_target_boundary::{
PutObjectOptions, PutObjectPartOptions, ReplicationTargetStore, TargetClient, replication_action_for_target_head,
replication_complete_multipart_options, replication_delete_marker_purge_remove_options, replication_delete_remove_options,
replication_force_delete_remove_options, replication_object_is_ssec_encrypted, replication_put_object_header_size,
replication_put_object_options, replication_target_head_is_newer_null_version,
replication_put_object_options, replication_target_head_is_newer_null_version, replication_target_version_id,
};
use super::replication_versioning_boundary::ReplicationVersioningStore;
use super::runtime_boundary as runtime_sources;
@@ -70,9 +71,8 @@ use rustfs_utils::http::{
AMZ_TAGGING_DIRECTIVE, SUFFIX_REPLICATION_RESET, SUFFIX_REPLICATION_STATUS, has_internal_suffix, insert_str,
};
use rustfs_utils::{DEFAULT_SIP_HASH_KEY, sip_hash};
#[cfg(test)]
use s3s::dto::ReplicationConfiguration;
use std::collections::HashMap;
use std::collections::{HashMap, HashSet};
use std::fmt::Display;
use std::sync::Arc;
use time::OffsetDateTime;
@@ -97,6 +97,8 @@ const EVENT_RESYNC_TASK_FAILED: &str = "replication_resync_task_failed";
const EVENT_RESYNC_TARGET_OPERATION_FAILED: &str = "replication_resync_target_operation_failed";
const EVENT_RESYNC_RUNTIME_CHANNEL_FAILED: &str = "replication_resync_runtime_channel_failed";
const ERR_REPLICATION_METADATA_COPY_UNSUPPORTED: &str = "metadata-only replication is not implemented";
const ERR_VERSION_PURGE_TARGET_STILL_EXISTS: &str = "target version still exists after replication purge";
const ERR_VERSION_PURGE_MISSING_VERSION_ID: &str = "version purge record is missing version ID";
const REPLICATION_TARGET_OFFLINE_ERROR_MARKERS: &[&str] = &[
"dispatch failure",
"timeouterror",
@@ -786,19 +788,46 @@ impl ReplicationResyncer {
}
if roi.delete_marker || !roi.version_purge_status.is_empty() {
let (version_id, dm_version_id) = if roi.version_purge_status.is_empty() {
(None, roi.version_id)
} else {
(roi.version_id, None)
let Some(parts) =
delete_replication_parts(roi.delete_marker, roi.version_id, !roi.version_purge_status.is_empty())
else {
debug!(
event = EVENT_RESYNC_OBJECT_PROCESSED,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
bucket = %bucket_name,
object = %roi.name,
reason = "version_purge_missing_version_id",
"Failed to process resync version purge"
);
let status = TargetReplicationResyncStatus {
bucket: roi.bucket.clone(),
object: roi.name.clone(),
failed_count: 1,
error: Some(ERR_VERSION_PURGE_MISSING_VERSION_ID.to_string()),
..Default::default()
};
if let Err(err) = results_tx.send(status).await {
error!(
event = EVENT_RESYNC_RUNTIME_CHANNEL_FAILED,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
bucket = %bucket_name,
reason = "status_channel_send_failed",
error = %err,
"Failed to send resync status"
);
}
continue;
};
let doi = DeletedObjectReplicationInfo {
delete_object: ReplicationDeletedObject {
object_name: roi.name.clone(),
delete_marker_version_id: dm_version_id,
version_id,
delete_marker_version_id: parts.delete_marker_version_id,
version_id: parts.version_id,
replication_state: roi.replication_state.clone(),
delete_marker: roi.delete_marker,
delete_marker: parts.delete_marker,
delete_marker_mtime: roi.mod_time,
..Default::default()
},
@@ -821,21 +850,41 @@ impl ReplicationResyncer {
};
let reset_id = target_client.reset_id.clone();
let is_version_purge = !roi.version_purge_status.is_empty();
let head_result = head_object_with_proxy_stats(
&bucket_name,
target_client.as_ref(),
&target_client.bucket,
&roi.name,
roi.version_id.map(|v| v.to_string()),
replication_target_version_id(roi.version_id, is_version_purge),
)
.await;
let (size, err) = match head_result {
Ok(_) if is_version_purge => {
st.failed_count += 1;
(0, Some(ERR_VERSION_PURGE_TARGET_STILL_EXISTS.to_string()))
}
Ok(_) => {
st.replicated_count += 1;
st.replicated_size += roi.size;
(roi.size, None)
}
Err(err) if is_version_purge => {
let (is_not_found, code) = err
.as_service_error()
.map(|service_err| (service_err.is_not_found(), service_err.code()))
.unwrap_or((false, None));
let raw_status = err.raw_response().map(|response| response.status().as_u16());
let missing = version_purge_target_missing(is_not_found, code, raw_status);
if missing {
st.replicated_count += 1;
(0, None)
} else {
st.failed_count += 1;
(0, resync_target_error_detail(&err))
}
}
Err(err) if roi.delete_marker => {
// Verifying a replicated delete marker: only a
// definitive 404/NoSuchKey or 405/MethodNotAllowed
@@ -852,7 +901,7 @@ impl ReplicationResyncer {
};
if retryable {
st.failed_count += 1;
(0, Some(err))
(0, resync_target_error_detail(&err))
} else {
st.replicated_count += 1;
(0, None)
@@ -871,17 +920,17 @@ impl ReplicationResyncer {
}
Ok(None) => {
st.failed_count += 1;
(0, Some(err))
(0, resync_target_error_detail(&err))
}
Err(e2) => {
st.failed_count += 1;
(0, Some(e2))
(0, resync_target_error_detail(&e2))
}
}
}
Err(err) => {
st.failed_count += 1;
(0, Some(err))
(0, resync_target_error_detail(&err))
}
};
@@ -911,7 +960,7 @@ impl ReplicationResyncer {
"Processed resync object"
);
}
st.error = err.as_ref().and_then(resync_target_error_detail);
st.error = err;
if cancel_token.is_cancelled() {
return;
@@ -1065,22 +1114,27 @@ pub async fn get_heal_replicate_object_info(oi: &ObjectInfo, rcfg: &ReplicationC
}
let dsc = if heal_uses_delete_replication_path(oi.delete_marker, &oi.version_purge_status) {
check_replicate_delete(
oi.bucket.as_str(),
&ObjectToDelete {
object_name: oi.name.clone(),
version_id: oi.version_id,
..Default::default()
},
&oi,
&ObjectOptions {
versioned: ReplicationVersioningStore::prefix_enabled(&oi.bucket, &oi.name).await,
version_suspended: ReplicationVersioningStore::prefix_suspended(&oi.bucket, &oi.name).await,
..Default::default()
},
None,
)
.await
match delete_replication_parts(oi.delete_marker, oi.version_id, !oi.version_purge_status.is_empty()) {
Some(parts) => {
check_replicate_delete(
oi.bucket.as_str(),
&ObjectToDelete {
object_name: oi.name.clone(),
version_id: parts.version_id,
..Default::default()
},
&oi,
&ObjectOptions {
versioned: ReplicationVersioningStore::prefix_enabled(&oi.bucket, &oi.name).await,
version_suspended: ReplicationVersioningStore::prefix_suspended(&oi.bucket, &oi.name).await,
..Default::default()
},
None,
)
.await
}
None => ReplicateDecision::default(),
}
} else {
must_replicate(
oi.bucket.as_str(),
@@ -1150,6 +1204,59 @@ pub async fn replicate_delete<S: ReplicationStorage>(dobj: DeletedObjectReplicat
} else {
dobj.delete_object.version_id
};
let _rcfg = match get_replication_config(&bucket).await {
Ok(Some(config)) => config,
Ok(None) => {
debug!(
event = EVENT_REPLICATION_DELETE_SKIPPED,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
bucket = %bucket,
reason = "replication_config_missing",
"Skipping replication delete because replication config is missing"
);
send_local_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: ObjectInfo {
bucket: bucket.clone(),
name: dobj.delete_object.object_name.clone(),
version_id,
delete_marker: dobj.delete_object.delete_marker,
..Default::default()
},
user_agent: "Internal: [Replication]".to_string(),
..Default::default()
});
return;
}
Err(err) => {
debug!(
event = EVENT_REPLICATION_DELETE_SKIPPED,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
bucket = %bucket,
error = %err,
reason = "replication_config_lookup_failed",
"Skipping replication delete because replication config lookup failed"
);
send_local_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: ObjectInfo {
bucket: bucket.clone(),
name: dobj.delete_object.object_name.clone(),
version_id,
delete_marker: dobj.delete_object.delete_marker,
..Default::default()
},
user_agent: "Internal: [Replication]".to_string(),
..Default::default()
});
return;
}
};
if dobj.delete_object.delete_marker
&& let Some(delete_marker_version_id) = dobj.delete_object.delete_marker_version_id
@@ -1415,8 +1522,8 @@ pub async fn replicate_delete<S: ReplicationStorage>(dobj: DeletedObjectReplicat
delete_marker_version_id,
)
.await
&& replicate_delete_marker_purge_to_targets(&bucket_clone, &dobj_clone, &dsc_clone).await
{
replicate_delete_marker_purge_to_targets(&bucket_clone, &dobj_clone, &dsc_clone).await;
break;
}
tokio::time::sleep(TokioDuration::from_secs(1)).await;
@@ -1546,9 +1653,50 @@ async fn source_delete_marker_missing<S: EcstoreObjectOperations>(
}
}
async fn replicate_delete_marker_purge_to_targets(bucket: &str, dobj: &DeletedObjectReplicationInfo, dsc: &ReplicateDecision) {
fn delete_marker_purge_target_arns(config: &ReplicationConfiguration, dobj: &DeletedObjectReplicationInfo) -> HashSet<String> {
let replica = dobj
.delete_object
.replication_state
.as_ref()
.is_some_and(|state| state.replica_status == ReplicationStatusType::Replica);
delete_replication_target_arns(config, &dobj.delete_object.object_name, replica)
}
#[derive(Debug, PartialEq, Eq)]
enum DeleteMarkerPurgeConfig {
Apply(HashSet<String>),
Stop,
Retry,
}
fn delete_marker_purge_config<E>(
result: std::result::Result<Option<ReplicationConfiguration>, E>,
dobj: &DeletedObjectReplicationInfo,
) -> DeleteMarkerPurgeConfig {
match result {
Ok(Some(config)) => DeleteMarkerPurgeConfig::Apply(delete_marker_purge_target_arns(&config, dobj)),
Ok(None) => DeleteMarkerPurgeConfig::Stop,
Err(_) => DeleteMarkerPurgeConfig::Retry,
}
}
async fn replicate_delete_marker_purge_to_targets(
bucket: &str,
dobj: &DeletedObjectReplicationInfo,
dsc: &ReplicateDecision,
) -> bool {
let Some(delete_marker_version_id) = dobj.delete_object.delete_marker_version_id else {
return;
return true;
};
let marker_creation_purge_targets = if dobj.delete_object.delete_marker {
match delete_marker_purge_config(get_replication_config(bucket).await, dobj) {
DeleteMarkerPurgeConfig::Apply(targets) => Some(targets),
DeleteMarkerPurgeConfig::Stop => return true,
DeleteMarkerPurgeConfig::Retry => return false,
}
} else {
None
};
for tgt_entry in dsc.targets_map.values() {
@@ -1558,6 +1706,12 @@ async fn replicate_delete_marker_purge_to_targets(bucket: &str, dobj: &DeletedOb
if !dobj.target_arn.is_empty() && dobj.target_arn != tgt_entry.arn {
continue;
}
if marker_creation_purge_targets
.as_ref()
.is_some_and(|targets| !targets.contains(&tgt_entry.arn))
{
continue;
}
let Some(tgt_client) = ReplicationTargetStore::remote_target_client(bucket, &tgt_entry.arn).await else {
continue;
};
@@ -1571,6 +1725,7 @@ async fn replicate_delete_marker_purge_to_targets(bucket: &str, dobj: &DeletedOb
)
.await;
}
true
}
async fn replicate_force_delete_to_targets<S: ReplicationStorage>(dobj: &DeletedObjectReplicationInfo, storage: Arc<S>) {
@@ -1797,10 +1952,10 @@ async fn replicate_force_delete_to_targets<S: ReplicationStorage>(dobj: &Deleted
}
async fn replicate_delete_to_target(dobj: &DeletedObjectReplicationInfo, tgt_client: Arc<TargetClient>) -> ReplicatedTargetInfo {
let version_id = if let Some(version_id) = &dobj.delete_object.delete_marker_version_id {
version_id.to_owned()
let version_id = if let Some(version_id) = dobj.delete_object.delete_marker_version_id {
Some(version_id)
} else {
dobj.delete_object.version_id.unwrap_or_default()
dobj.delete_object.version_id
};
let mut rinfo = dobj
@@ -1835,11 +1990,7 @@ async fn replicate_delete_to_target(dobj: &DeletedObjectReplicationInfo, tgt_cli
return rinfo;
}
let version_id = if version_id.is_nil() {
None
} else {
Some(version_id.to_string())
};
let version_id = replication_target_version_id(version_id, is_version_purge);
if dobj.delete_object.delete_marker && dobj.delete_object.delete_marker_version_id.is_some() {
match head_object_with_proxy_stats(
@@ -3089,6 +3240,10 @@ async fn replicate_object_with_multipart<S: ReplicationObjectIO>(ctx: MultipartR
#[cfg(test)]
mod tests {
use super::*;
use s3s::dto::{
DeleteReplication, DeleteReplicationStatus, Destination, ReplicaModifications, ReplicaModificationsStatus,
ReplicationRule, ReplicationRuleAndOperator, ReplicationRuleFilter, ReplicationRuleStatus, SourceSelectionCriteria, Tag,
};
use std::collections::HashMap;
use time::OffsetDateTime;
use uuid::Uuid;
@@ -3398,6 +3553,110 @@ mod tests {
);
}
#[test]
fn test_delete_marker_purge_targets_follow_delete_and_replica_modification_rules() {
fn rule(arn: &str, delete_status: &'static str) -> ReplicationRule {
ReplicationRule {
delete_marker_replication: None,
delete_replication: Some(DeleteReplication {
status: DeleteReplicationStatus::from_static(delete_status),
}),
destination: Destination {
bucket: arn.to_string(),
..Default::default()
},
existing_object_replication: None,
filter: None,
id: Some(arn.to_string()),
prefix: Some("logs/".to_string()),
priority: Some(1),
source_selection_criteria: None,
status: ReplicationRuleStatus::from_static(ReplicationRuleStatus::ENABLED),
}
}
let enabled_arn = "arn:rustfs:replication:us-east-1:target:enabled";
let disabled_arn = "arn:rustfs:replication:us-east-1:target:disabled";
let delete_marker_version_id = Uuid::new_v4();
let mut config = ReplicationConfiguration {
role: String::new(),
rules: vec![
rule(enabled_arn, DeleteReplicationStatus::ENABLED),
rule(disabled_arn, DeleteReplicationStatus::DISABLED),
],
};
let mut dobj = DeletedObjectReplicationInfo {
delete_object: ReplicationDeletedObject {
object_name: "logs/object.txt".to_string(),
delete_marker: true,
delete_marker_version_id: Some(delete_marker_version_id),
..Default::default()
},
..Default::default()
};
assert_eq!(delete_marker_purge_target_arns(&config, &dobj), HashSet::from([enabled_arn.to_string()]));
assert_eq!(
delete_marker_purge_config::<()>(Ok(Some(config.clone())), &dobj),
DeleteMarkerPurgeConfig::Apply(HashSet::from([enabled_arn.to_string()]))
);
dobj.delete_object.replication_state = Some(Default::default());
dobj.delete_object
.replication_state
.as_mut()
.expect("test replication state")
.replica_status = ReplicationStatusType::Replica;
assert!(delete_marker_purge_target_arns(&config, &dobj).is_empty());
config.rules[0].source_selection_criteria = Some(SourceSelectionCriteria {
replica_modifications: Some(ReplicaModifications {
status: ReplicaModificationsStatus::from_static(ReplicaModificationsStatus::ENABLED),
}),
sse_kms_encrypted_objects: None,
});
assert_eq!(delete_marker_purge_target_arns(&config, &dobj), HashSet::from([enabled_arn.to_string()]));
config.rules[0].prefix = None;
config.rules[0].filter = Some(ReplicationRuleFilter {
tag: Some(Tag {
key: Some("env".to_string()),
value: Some("prod".to_string()),
}),
..Default::default()
});
assert!(delete_marker_purge_target_arns(&config, &dobj).is_empty());
config.rules[0].filter = Some(ReplicationRuleFilter {
and: Some(ReplicationRuleAndOperator {
prefix: Some("logs/".to_string()),
tags: Some(vec![Tag {
key: Some("env".to_string()),
value: Some("prod".to_string()),
}]),
}),
..Default::default()
});
assert!(delete_marker_purge_target_arns(&config, &dobj).is_empty());
}
#[test]
fn test_delete_marker_purge_config_errors_are_retryable() {
let delete_marker_version_id = Uuid::new_v4();
let dobj = DeletedObjectReplicationInfo {
delete_object: ReplicationDeletedObject {
object_name: "object.txt".to_string(),
delete_marker: true,
delete_marker_version_id: Some(delete_marker_version_id),
..Default::default()
},
..Default::default()
};
assert_eq!(delete_marker_purge_config::<()>(Ok(None), &dobj), DeleteMarkerPurgeConfig::Stop);
assert_eq!(delete_marker_purge_config::<()>(Err(()), &dobj), DeleteMarkerPurgeConfig::Retry);
}
#[test]
fn test_is_retryable_delete_replication_head_error_allows_delete_marker_head_responses() {
assert!(
@@ -31,10 +31,13 @@ use rustfs_utils::http::{
};
use time::OffsetDateTime;
use time::format_description::well_known::Rfc3339;
use uuid::Uuid;
pub(crate) use crate::bucket::bucket_target_sys::{
AdvancedPutOptions, PutObjectOptions, PutObjectPartOptions, RemoveObjectOptions, TargetClient,
};
#[cfg(test)]
pub(crate) use crate::bucket::target::BucketTarget;
pub(crate) use crate::bucket::target::BucketTargets;
use super::replication_config_store::ReplicationConfigStore;
@@ -305,6 +308,14 @@ pub(crate) fn replication_target_head_is_newer_null_version(object_info: &Object
target_is_newer_than_source_null_version(&replication_source_object(object_info), &replication_target_object(target))
}
pub(crate) fn replication_target_version_id(version_id: Option<Uuid>, version_purge: bool) -> Option<String> {
match version_id {
Some(version_id) if version_id.is_nil() && version_purge => Some("null".to_string()),
Some(version_id) if !version_id.is_nil() => Some(version_id.to_string()),
_ => None,
}
}
pub(crate) fn replication_delete_remove_options(
delete_marker: bool,
replication_mtime: Option<OffsetDateTime>,
@@ -519,6 +530,18 @@ mod tests {
assert!(force.replication_request);
}
#[test]
fn replication_target_version_id_preserves_null_purges() {
assert_eq!(replication_target_version_id(Some(Uuid::nil()), true).as_deref(), Some("null"));
assert_eq!(replication_target_version_id(Some(Uuid::nil()), false), None);
let version_id = Uuid::new_v4();
assert_eq!(
replication_target_version_id(Some(version_id), true).as_deref(),
Some(version_id.to_string().as_str())
);
}
#[test]
fn replication_complete_multipart_options_sets_actual_size() {
let options = replication_complete_multipart_options("1024".to_string());
@@ -58,7 +58,7 @@ use std::{
collections::HashMap,
io::Cursor,
sync::{
Arc,
Arc, Weak,
atomic::{AtomicBool, Ordering},
},
time::SystemTime,
@@ -350,6 +350,44 @@ impl PeerRestClient {
}
}
fn parse_topology_host(peer_host_port: &str, grid_host: &str) -> Result<XHost> {
let url = url::Url::parse(grid_host).map_err(|_| Error::other("peer grid host is not a valid URL"))?;
if !matches!(url.scheme(), "http" | "https")
|| !url.username().is_empty()
|| url.password().is_some()
|| url.query().is_some()
|| url.fragment().is_some()
|| url.path() != "/"
{
return Err(Error::other("peer grid host has an invalid URL shape"));
}
let url_host = url.host().ok_or_else(|| Error::other("peer grid host is missing a host"))?;
let topology_host = match url.port() {
Some(port) => format!("{url_host}:{port}"),
None => url_host.to_string(),
};
let explicit_port = url.port();
let name = match url_host {
url::Host::Domain(domain) => domain.to_string(),
url::Host::Ipv4(address) => address.to_string(),
url::Host::Ipv6(address) if explicit_port.is_none() => format!("[{address}]"),
url::Host::Ipv6(address) => address.to_string(),
};
let port = url
.port_or_known_default()
.filter(|port| *port > 0)
.ok_or_else(|| Error::other("peer grid host is missing a valid port"))?;
let host = XHost {
name,
port,
is_port_set: explicit_port.is_some(),
};
if topology_host != peer_host_port {
return Err(Error::other("peer topology host does not match its grid URL"));
}
Ok(host)
}
fn build_clients_from_slots(
slots: Vec<(String, Option<String>, bool)>,
) -> (Vec<Option<Self>>, Vec<Option<Self>>, Vec<String>) {
@@ -363,14 +401,14 @@ impl PeerRestClient {
}
let client = match grid_host {
Some(grid_host) => match XHost::try_from(peer_host_port.clone()) {
Some(grid_host) => match Self::parse_topology_host(&peer_host_port, &grid_host) {
Ok(host) => {
let mut client = PeerRestClient::new(host, grid_host);
client.topology_member = peer_host_port.clone();
Some(client)
}
Err(err) => {
warn!(peer = %peer_host_port, "Xhost parse failed while constructing peer client: {err:?}");
warn!(peer = %peer_host_port, "peer topology host parse failed while constructing peer client: {err:?}");
None
}
},
@@ -519,9 +557,8 @@ impl PeerRestClient {
}
let grid_host = self.grid_host.clone();
let offline = Arc::clone(&self.offline);
let recovery_running = Arc::clone(&self.recovery_running);
let span = Self::recovery_monitor_span(&grid_host);
let offline = Arc::downgrade(&self.offline);
let recovery_running = Arc::downgrade(&self.recovery_running);
// The offline flag and its recovery are the silent half of
// rustfs/backlog#888: log the monitor's start and its success so an
// "offline then back" episode leaves a trace on the observing node.
@@ -530,13 +567,34 @@ impl PeerRestClient {
grid_host = %self.grid_host,
"peer RPC connection marked offline after a network-like failure; starting background recovery monitor"
);
drop(Self::spawn_recovery_monitor(grid_host, offline, recovery_running));
}
fn spawn_recovery_monitor(
grid_host: String,
offline: Weak<AtomicBool>,
recovery_running: Weak<AtomicBool>,
) -> tokio::task::JoinHandle<()> {
let span = Self::recovery_monitor_span(&grid_host);
super::spawn_background_monitor(span, async move {
let mut delay = get_drive_active_check_interval();
let connect_timeout = get_drive_active_check_timeout();
for attempt in 1..=PEER_REST_RECOVERY_MAX_ATTEMPTS {
if offline.strong_count() == 0 || recovery_running.strong_count() == 0 {
return;
}
tokio::time::sleep(delay).await;
if offline.strong_count() == 0 || recovery_running.strong_count() == 0 {
return;
}
if Self::perform_connectivity_check(&grid_host, connect_timeout).await.is_ok() {
let Some(offline) = offline.upgrade() else {
return;
};
let Some(recovery_running) = recovery_running.upgrade() else {
return;
};
offline.store(false, Ordering::Release);
recovery_running.store(false, Ordering::Release);
info!(
@@ -556,8 +614,10 @@ impl PeerRestClient {
attempts = PEER_REST_RECOVERY_MAX_ATTEMPTS,
"peer recovery monitor reached max attempts; will retry on next request"
);
recovery_running.store(false, Ordering::Release);
});
if let Some(recovery_running) = recovery_running.upgrade() {
recovery_running.store(false, Ordering::Release);
}
})
}
#[cfg(test)]
@@ -1807,9 +1867,13 @@ fn tier_config_reload_status_outcome(status: tonic::Status) -> TierConfigReloadO
mod tests {
use super::*;
use crate::config::com::STORAGE_CLASS_SUB_SYS;
use crate::layout::{disks_layout::DisksLayout, endpoints::SetupType};
use rustfs_config::{ENV_KUBERNETES_SERVICE_HOST, ENV_LOCAL_ENDPOINT_HOST, ENV_STARTUP_TOPOLOGY_WAIT_MODE};
use serde_json::Value;
use serial_test::serial;
use std::io::{self, Write};
use std::sync::{Arc, Mutex};
use temp_env::async_with_vars;
use tracing_subscriber::{Registry, fmt::MakeWriter, layer::SubscriberExt};
#[test]
@@ -1927,30 +1991,115 @@ mod tests {
fn build_clients_from_slots_preserves_missing_remote_topology_slots() {
let slots = vec![
("127.0.0.1:9000".to_string(), None, true),
("127.0.0.1:9001".to_string(), Some("http://127.0.0.1:9001".to_string()), false),
(
"rustfs-1.invalid:9001".to_string(),
Some("http://rustfs-1.invalid:9001".to_string()),
false,
),
("rustfs-2.invalid".to_string(), Some("http://rustfs-2.invalid".to_string()), false),
("127.0.0.1:notaport".to_string(), Some("http://127.0.0.1:notaport".to_string()), false),
("127.0.0.1:9003".to_string(), None, false),
];
let (remote, all, remote_topology_hosts) = PeerRestClient::build_clients_from_slots(slots);
assert_eq!(remote.len(), 3, "local node is excluded but remote slots are not compacted away");
assert_eq!(all.len(), 4, "all slots preserve the sorted cluster topology shape");
assert_eq!(remote.len(), 4, "local node is excluded but remote slots are not compacted away");
assert_eq!(all.len(), 5, "all slots preserve the sorted cluster topology shape");
assert_eq!(
remote_topology_hosts,
vec![
"127.0.0.1:9001".to_string(),
"rustfs-1.invalid:9001".to_string(),
"rustfs-2.invalid".to_string(),
"127.0.0.1:notaport".to_string(),
"127.0.0.1:9003".to_string()
]
);
assert!(remote[0].is_some(), "valid remote peer should get a client");
assert!(remote[1].is_none(), "unparseable remote peer should remain observable as a missing slot");
assert!(remote[2].is_none(), "missing grid host should remain observable as a missing slot");
let unresolved = remote[0]
.as_ref()
.expect("temporarily unresolved remote peer should retain a client");
assert_eq!(unresolved.host.to_string(), "rustfs-1.invalid:9001");
let default_port = remote[1]
.as_ref()
.expect("temporarily unresolved scheme-default remote peer should retain a client");
assert_eq!(default_port.host.to_string(), "rustfs-2.invalid");
assert_eq!(default_port.host.port, 80);
assert!(!default_port.host.is_port_set);
assert!(remote[2].is_none(), "unparseable remote peer should remain observable as a missing slot");
assert!(remote[3].is_none(), "missing grid host should remain observable as a missing slot");
assert!(all[0].is_none(), "local node is represented by the local server_info row");
assert!(all[1].is_some());
assert!(all[2].is_none());
assert!(all[2].is_some());
assert!(all[3].is_none());
assert!(all[4].is_none());
}
#[test]
fn topology_host_parser_preserves_names_and_bracketed_ipv6() {
let domain = PeerRestClient::parse_topology_host("rustfs-1.invalid", "https://rustfs-1.invalid")
.expect("unresolved HTTPS topology host should parse without DNS");
assert_eq!(domain.to_string(), "rustfs-1.invalid");
assert_eq!(domain.port, 443);
assert!(!domain.is_port_set);
let ipv6 = PeerRestClient::parse_topology_host("[2001:db8::1]:9000", "http://[2001:db8::1]:9000")
.expect("bracketed IPv6 topology host should parse without changing its identity");
assert_eq!(ipv6.to_string(), "[2001:db8::1]:9000");
let default_port_ipv6 = PeerRestClient::parse_topology_host("[2001:db8::2]", "http://[2001:db8::2]")
.expect("scheme-default IPv6 topology host should parse without DNS");
assert_eq!(default_port_ipv6.to_string(), "[2001:db8::2]");
assert_eq!(default_port_ipv6.port, 80);
assert!(!default_port_ipv6.is_port_set);
assert!(PeerRestClient::parse_topology_host("peer.invalid:0", "http://peer.invalid:0").is_err());
assert!(PeerRestClient::parse_topology_host("peer-a.invalid:9000", "http://peer-b.invalid:9000").is_err());
assert!(PeerRestClient::parse_topology_host("peer.invalid:9000", "http://peer.invalid:9000/unexpected").is_err());
}
#[tokio::test]
#[serial]
async fn unresolved_default_port_endpoint_topology_retains_all_peer_clients() {
let volumes = (0..4)
.map(|index| format!("http://rustfs-{index}.invalid:80/data{index}"))
.collect::<Vec<_>>();
let layout = DisksLayout::from_volumes(&volumes).expect("distributed default-port topology should parse");
async_with_vars(
[
(ENV_STARTUP_TOPOLOGY_WAIT_MODE, Some("orchestrated")),
(ENV_LOCAL_ENDPOINT_HOST, Some("rustfs-0.invalid")),
(ENV_KUBERNETES_SERVICE_HOST, None),
],
async {
let (server_pools, setup_type) = EndpointServerPools::create_server_endpoints("0.0.0.0:80", &layout)
.await
.expect("explicit local identity should avoid peer DNS during endpoint construction");
assert_eq!(setup_type, SetupType::DistErasure);
let (remote, all, remote_topology_hosts) =
PeerRestClient::build_clients_from_slots(server_pools.peer_grid_host_slots_sorted());
assert_eq!(remote.len(), 3);
assert!(
remote.iter().all(Option::is_some),
"unresolved remote peers must retain reconnectable clients"
);
assert_eq!(all.len(), 4);
assert_eq!(all.iter().filter(|client| client.is_none()).count(), 1);
assert_eq!(remote_topology_hosts.len(), 3);
assert!(
remote_topology_hosts.iter().all(|host| !host.contains(':')),
"scheme-default ports must preserve the legacy topology identity"
);
assert!(
remote
.iter()
.flatten()
.all(|client| client.host.port == 80 && !client.host.is_port_set),
"scheme-default peers must retain the effective dial port"
);
},
)
.await;
}
#[test]
@@ -2740,6 +2889,31 @@ mod tests {
assert!(!client.offline.load(Ordering::Acquire));
}
#[tokio::test(start_paused = true)]
async fn dropped_peer_client_releases_and_stops_its_recovery_monitor() {
let client = test_peer_client();
client.offline.store(true, Ordering::Release);
client.recovery_running.store(true, Ordering::Release);
let offline = Arc::downgrade(&client.offline);
let recovery_running = Arc::downgrade(&client.recovery_running);
let handle = PeerRestClient::spawn_recovery_monitor(client.grid_host.clone(), offline.clone(), recovery_running.clone());
let started = tokio::time::Instant::now();
drop(client);
assert!(offline.upgrade().is_none(), "detached recovery must not retain offline state");
assert!(
recovery_running.upgrade().is_none(),
"detached recovery must not retain its running state"
);
handle.await.expect("recovery monitor should not panic");
assert_eq!(
tokio::time::Instant::now(),
started,
"recovery monitor should stop before advancing to its first delayed probe"
);
}
#[tokio::test]
async fn peer_rest_client_finalize_result_keeps_online_for_app_errors_mentioning_unavailable() {
// Regression: application error text containing "unavailable" (a
@@ -4814,9 +4814,11 @@ mod tests {
async fn test_remote_disk_endpoints_with_different_schemes() {
let test_cases = vec![
("http://server:9000", "server:9000"),
("https://secure-server:443", "secure-server"), // Default HTTPS port is omitted
("http://plain-server:80", "plain-server"),
("http://plain-server", "plain-server"),
("https://secure-server:443", "secure-server"),
("http://192.168.1.100:8080", "192.168.1.100:8080"),
("https://secure-server", "secure-server"), // No port specified
("https://secure-server", "secure-server"),
];
for (url_str, expected_hostname) in test_cases {
+199 -20
View File
@@ -37,7 +37,9 @@ use crate::{
runtime::instance::{InstanceContext, bootstrap_ctx},
runtime::sources as runtime_sources,
set_disk::{PreparedGetObjectMetadata, SetDisks},
store::init_format::{check_format_erasure_values, get_format_erasure_in_quorum, load_format_erasure_all, save_format_file},
store::init_format::{
check_format_erasure_values, load_format_erasure_all, save_format_file, select_format_erasure_in_quorum,
},
};
use futures::{
future::join_all,
@@ -947,7 +949,7 @@ impl crate::storage_api_contracts::heal::HealOperations for Sets {
#[tracing::instrument(skip(self))]
async fn heal_format(&self, dry_run: bool) -> Result<(HealResultItem, Option<Error>)> {
let (disks, _) = init_storage_disks_with_errors(
let (disks, init_errs) = init_storage_disks_with_errors(
&self.endpoints.endpoints,
&DiskOption {
cleanup: false,
@@ -955,15 +957,36 @@ impl crate::storage_api_contracts::heal::HealOperations for Sets {
},
)
.await;
let (formats, errs) = load_format_erasure_all(&disks, true).await;
let (formats, mut errs) = load_format_erasure_all(&disks, true).await;
for (err, init_err) in errs.iter_mut().zip(init_errs) {
if init_err.is_some() {
*err = init_err;
}
}
if errs.iter().any(|err| {
matches!(
err,
Some(DiskError::InconsistentDisk | DiskError::CorruptedFormat | DiskError::CorruptedBackend)
)
}) {
return Ok((HealResultItem::default(), Some(StorageError::CorruptedFormat)));
}
if let Err(err) = check_format_erasure_values(&formats, self.set_drive_count) {
info!("failed to check formats erasure values: {}", err);
return Ok((HealResultItem::default(), Some(err)));
}
let ref_format = match get_format_erasure_in_quorum(&formats) {
Ok(format) => format,
let (ref_format, quorum_members) = match select_format_erasure_in_quorum(&formats, 0) {
Ok((format, members)) if format.shared_identity() == self.format.shared_identity() => (format, members),
Ok(_) => return Ok((HealResultItem::default(), Some(StorageError::CorruptedFormat))),
Err(err) => return Ok((HealResultItem::default(), Some(err))),
};
if formats
.iter()
.zip(quorum_members)
.any(|(format, member)| format.is_some() && !member)
{
return Ok((HealResultItem::default(), Some(StorageError::CorruptedFormat)));
}
let mut res = HealResultItem {
heal_item_type: HealItemType::Metadata.to_string(),
detail: "disk-format".to_string(),
@@ -985,11 +1008,6 @@ impl crate::storage_api_contracts::heal::HealOperations for Sets {
return Ok((res, Some(StorageError::NoHealRequired)));
}
// if !self.format.eq(&ref_format) {
// info!("format ({:?}) not eq ref_format ({:?})", self.format, ref_format);
// return Ok((res, Some(Error::new(DiskError::CorruptedFormat))));
// }
let (new_format_sets, _) = new_heal_format_sets(&ref_format, self.set_count, self.set_drive_count, &formats, &errs);
if !dry_run {
let mut tmp_new_formats = vec![None; self.set_count * self.set_drive_count];
@@ -1298,7 +1316,7 @@ mod tests {
assert_eq!(result, (Some(3), Some(1), Some(0)));
}
async fn multipart_listing_test_sets() -> (Vec<tempfile::TempDir>, Arc<Sets>) {
async fn two_set_test_sets() -> (Vec<tempfile::TempDir>, Arc<Sets>) {
let format = FormatV3::new(2, 2);
let mut temp_dirs = Vec::new();
let mut all_endpoints = Vec::new();
@@ -1339,8 +1357,8 @@ mod tests {
Arc::new(RwLock::new(disks)),
2,
1,
0,
set_index,
0,
endpoints,
format.clone(),
vec![Arc::new(LocalClient::new()), Arc::new(LocalClient::new())],
@@ -1373,11 +1391,114 @@ mod tests {
(temp_dirs, sets)
}
#[tokio::test]
async fn set_format_heal_accepts_quorum_from_a_nonzero_set() {
let (_temp_dirs, sets) = two_set_test_sets().await;
let (result, err) = sets.disk_set[1]
.heal_format(false)
.await
.expect("the second erasure set should load its own format quorum");
assert!(matches!(err, Some(StorageError::NoHealRequired)), "unexpected heal result: {err:?}");
assert_eq!(result.disk_count, 2);
assert_eq!(result.set_count, 1);
}
#[tokio::test]
async fn format_heal_rejects_foreign_majorities_at_set_and_pool_scopes() {
let (_temp_dirs, _canonical_format, sets) = setup_heal_format_sets(2, true).await;
let set_disks = set_level_heal_view(&sets).await;
let (_, set_err) = set_disks
.heal_format(false)
.await
.expect("set format heal should report a typed mismatch");
assert!(
matches!(set_err, Some(StorageError::CorruptedFormat)),
"foreign set majority must not replace the cached format: {set_err:?}"
);
let (_, pool_err) = sets
.heal_format(false)
.await
.expect("pool format heal should report a typed mismatch");
assert!(
matches!(pool_err, Some(StorageError::CorruptedFormat)),
"foreign pool majority must not replace the cached format: {pool_err:?}"
);
}
#[tokio::test]
async fn pool_format_heal_rejects_a_wrong_slot_minority() {
let (_temp_dirs, canonical_format, sets) = setup_heal_format_sets(3, false).await;
let mut poisoned_format = canonical_format.clone();
poisoned_format.erasure.this = canonical_format.erasure.sets[0][0];
replace_heal_test_format(&sets, 2, &poisoned_format).await;
let probe_err = new_disk(
&sets.endpoints.endpoints.as_ref()[2],
&DiskOption {
cleanup: false,
health_check: false,
},
)
.await
.expect_err("a wrong-slot local format must fail disk initialization");
assert_eq!(probe_err, DiskError::InconsistentDisk);
let (_, pool_err) = sets
.heal_format(false)
.await
.expect("pool format heal should report a typed slot mismatch");
assert!(
matches!(pool_err, Some(StorageError::CorruptedFormat)),
"a wrong-slot minority must not be reported as no-heal-required: {pool_err:?}"
);
assert_eq!(
read_heal_test_format(&sets, 2).await,
poisoned_format,
"format heal must not overwrite a wrong-slot disk"
);
}
#[tokio::test]
async fn format_heal_rejects_a_foreign_minority_at_set_and_pool_scopes() {
let (_temp_dirs, canonical_format, sets) = setup_heal_format_sets(3, false).await;
let mut poisoned_format = canonical_format.clone();
poisoned_format.id = Uuid::new_v4();
poisoned_format.erasure.this = poisoned_format.erasure.sets[0][2];
replace_heal_test_format(&sets, 2, &poisoned_format).await;
let set_disks = set_level_heal_view(&sets).await;
let (_, set_err) = set_disks
.heal_format(false)
.await
.expect("set format heal should report a typed identity mismatch");
assert!(
matches!(set_err, Some(StorageError::CorruptedFormat)),
"a foreign minority must not be reported as no-heal-required: {set_err:?}"
);
let (_, pool_err) = sets
.heal_format(false)
.await
.expect("pool format heal should report a typed identity mismatch");
assert!(
matches!(pool_err, Some(StorageError::CorruptedFormat)),
"a foreign minority must not be reported as no-heal-required: {pool_err:?}"
);
assert_eq!(
read_heal_test_format(&sets, 2).await,
poisoned_format,
"format heal must not overwrite a foreign disk"
);
}
#[tokio::test(flavor = "multi_thread")]
#[serial]
async fn list_multipart_uploads_merges_all_sets_without_pagination_loss() {
let _setup_type_guard = SetupTypeGuard::switch_to(SetupType::Erasure).await;
let (_temp_dirs, sets) = multipart_listing_test_sets().await;
let (_temp_dirs, sets) = two_set_test_sets().await;
let bucket = format!("multipart-list-{}", Uuid::new_v4().simple());
sets.make_bucket(&bucket, &MakeBucketOptions::default())
.await
@@ -1616,11 +1737,15 @@ mod tests {
// formatting the first `num_formatted` of them against a shared reference
// format and leaving the rest unformatted. Returns the live TempDir handles
// (must be kept alive), the reference format, and the assembled `Sets`.
// `disk_set` is intentionally empty: these tests only drive `heal_format`
// with `dry_run == true`, which never touches `disk_set`.
async fn setup_heal_format_sets(num_formatted: usize) -> (Vec<tempfile::TempDir>, FormatV3, Sets) {
// `disk_set` is intentionally empty: these tests only exercise paths that
// return before pool-level healing delegates into a set.
async fn setup_heal_format_sets(num_formatted: usize, foreign_identity: bool) -> (Vec<tempfile::TempDir>, FormatV3, Sets) {
const SET_DRIVE_COUNT: usize = 3;
let ref_format = FormatV3::new(1, SET_DRIVE_COUNT);
let mut stored_format = ref_format.clone();
if foreign_identity {
stored_format.id = Uuid::new_v4();
}
let mut dirs = Vec::with_capacity(SET_DRIVE_COUNT);
let mut endpoints = Vec::with_capacity(SET_DRIVE_COUNT);
@@ -1645,8 +1770,8 @@ mod tests {
)
.await
.expect("disk should be created");
let mut disk_format = ref_format.clone();
disk_format.erasure.this = ref_format.erasure.sets[0][i];
let mut disk_format = stored_format.clone();
disk_format.erasure.this = stored_format.erasure.sets[0][i];
save_format_file(&Some(disk), &Some(disk_format))
.await
.expect("format should be saved");
@@ -1677,6 +1802,60 @@ mod tests {
(dirs, ref_format, sets)
}
async fn set_level_heal_view(sets: &Sets) -> Arc<SetDisks> {
let endpoints = sets.endpoints.endpoints.as_ref().clone();
let mut disks = Vec::with_capacity(endpoints.len());
for endpoint in &endpoints {
disks.push(Some(
new_disk(
endpoint,
&DiskOption {
cleanup: false,
health_check: false,
},
)
.await
.expect("fresh set-level disk handle should open"),
));
}
SetDisks::new(
"test-owner".to_string(),
Arc::new(RwLock::new(disks)),
endpoints.len(),
1,
0,
0,
endpoints,
sets.format.clone(),
Vec::new(),
)
.await
}
async fn replace_heal_test_format(sets: &Sets, disk_index: usize, format: &FormatV3) {
let disk = new_disk(
&sets.endpoints.endpoints.as_ref()[disk_index],
&DiskOption {
cleanup: false,
health_check: false,
},
)
.await
.expect("heal test disk should open");
save_format_file(&Some(disk.clone()), &Some(format.clone()))
.await
.expect("poisoned test format should be written");
}
async fn read_heal_test_format(sets: &Sets, disk_index: usize) -> FormatV3 {
let path = std::path::Path::new(&sets.endpoints.endpoints.as_ref()[disk_index].get_file_path())
.join(crate::disk::RUSTFS_META_BUCKET)
.join(crate::disk::FORMAT_CONFIG_FILE);
let data = tokio::fs::read(path).await.expect("test format should be readable");
FormatV3::try_from(data.as_slice()).expect("test format should parse")
}
// Regression for #956 (NoHealRequired path): with every disk already
// formatted, `heal_format` reports exactly one drive record per disk
// (N = set_count * set_drive_count), each carrying a real endpoint. Before
@@ -1685,7 +1864,7 @@ mod tests {
#[tokio::test]
#[serial]
async fn heal_format_no_heal_required_reports_one_record_per_disk() {
let (_dirs, _ref_format, sets) = setup_heal_format_sets(3).await;
let (_dirs, _ref_format, sets) = setup_heal_format_sets(3, false).await;
let (res, err) = sets.heal_format(true).await.expect("heal_format should succeed");
// All disks formatted -> NoHealRequired early return, still returns `res`.
@@ -1715,7 +1894,7 @@ mod tests {
#[serial]
async fn heal_format_heal_path_reports_one_record_per_disk_aligned() {
// Disks 0 and 1 formatted (quorum), disk 2 unformatted.
let (_dirs, _ref_format, sets) = setup_heal_format_sets(2).await;
let (_dirs, _ref_format, sets) = setup_heal_format_sets(2, false).await;
let (res, err) = sets.heal_format(true).await.expect("heal_format should succeed");
// Unformatted disk present -> heal path, not NoHealRequired.
+12 -1
View File
@@ -203,7 +203,7 @@ fn get_all_sets<T: AsRef<str>>(set_drive_count: usize, is_ellipses: bool, args:
for args in set_args.iter() {
for arg in args {
if unique_args.contains(arg) {
return Err(Error::other(format!("Input args {arg} has duplicate ellipses")));
return Err(Error::other("input arguments contain a duplicate endpoint after ellipsis expansion"));
}
unique_args.insert(arg);
}
@@ -924,4 +924,15 @@ mod test {
}
}
}
#[test]
fn layout_errors_do_not_echo_url_credentials() {
for volumes in [
vec!["http://:duplicate-secret@server/path", "http://:duplicate-secret@server/path"],
vec!["http://:ellipsis...secret@server/path"],
] {
let err = DisksLayout::from_volumes(&volumes).unwrap_err();
assert!(!err.to_string().contains("secret"), "layout error leaked endpoint credentials: {err}");
}
}
}
+19 -2
View File
@@ -88,6 +88,7 @@ impl TryFrom<&str> for Endpoint {
// - All field should be empty except Host and Path.
if !((url.scheme() == "http" || url.scheme() == "https")
&& url.username().is_empty()
&& url.password().is_none()
&& url.fragment().is_none()
&& url.query().is_none())
{
@@ -366,6 +367,12 @@ mod test {
expected_type: None,
expected_err: Some(Error::other("invalid URL endpoint format")),
},
TestCase {
arg: "http://:topsecret@server/path",
expected_endpoint: None,
expected_type: None,
expected_err: Some(Error::other("invalid URL endpoint format")),
},
TestCase {
arg: "http://:/path",
expected_endpoint: None,
@@ -505,8 +512,18 @@ mod test {
let endpoint = Endpoint::try_from("http://example.com:9000/path").unwrap();
assert_eq!(endpoint.host_port(), "example.com:9000");
let endpoint_no_port = Endpoint::try_from("https://example.com/path").unwrap();
assert_eq!(endpoint_no_port.host_port(), "example.com");
for endpoint in [
Endpoint::try_from("http://example.com/path").unwrap(),
Endpoint::try_from("http://example.com:80/path").unwrap(),
] {
assert_eq!(endpoint.host_port(), "example.com");
}
for endpoint in [
Endpoint::try_from("https://example.com/path").unwrap(),
Endpoint::try_from("https://example.com:443/path").unwrap(),
] {
assert_eq!(endpoint.host_port(), "example.com");
}
let file_endpoint = Endpoint::try_from("/tmp/data").unwrap();
assert_eq!(file_endpoint.host_port(), "");
File diff suppressed because it is too large Load Diff
+52 -47
View File
@@ -18,7 +18,7 @@ use serde::{Deserialize, Serialize};
use serde_json::Error as JsonError;
use uuid::Uuid;
#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq, Hash)]
pub enum FormatMetaVersion {
#[serde(rename = "1")]
V1,
@@ -27,7 +27,7 @@ pub enum FormatMetaVersion {
Unknown,
}
#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq, Hash)]
pub enum FormatBackend {
#[serde(rename = "xl")]
Erasure,
@@ -64,7 +64,7 @@ pub struct FormatErasureV3 {
pub distribution_algo: DistributionAlgoVersion,
}
#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq, Hash)]
pub enum FormatErasureVersion {
#[serde(rename = "1")]
V1,
@@ -77,7 +77,7 @@ pub enum FormatErasureVersion {
Unknown,
}
#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq, Hash)]
pub enum DistributionAlgoVersion {
#[serde(rename = "CRCMOD")]
V1,
@@ -121,6 +121,15 @@ pub struct FormatV3 {
pub disk_info: Option<DiskInfo>,
}
pub(crate) type SharedFormatIdentity<'a> = (
&'a FormatMetaVersion,
&'a FormatBackend,
&'a Uuid,
&'a FormatErasureVersion,
&'a [Vec<Uuid>],
&'a DistributionAlgoVersion,
);
impl TryFrom<&[u8]> for FormatV3 {
type Error = JsonError;
@@ -198,52 +207,24 @@ impl FormatV3 {
}
pub fn check_other(&self, other: &FormatV3) -> Result<()> {
let mut tmp = other.clone();
let this = tmp.erasure.this;
tmp.erasure.this = Uuid::nil();
if self.erasure.sets.len() != other.erasure.sets.len() {
return Err(Error::other(format!(
"Expected number of sets {}, got {}",
self.erasure.sets.len(),
other.erasure.sets.len()
)));
if self.shared_identity() != other.shared_identity() {
return Err(Error::other("storage formats do not match"));
}
for i in 0..self.erasure.sets.len() {
if self.erasure.sets[i].len() != other.erasure.sets[i].len() {
return Err(Error::other(format!(
"Each set should be of same size, expected {}, got {}",
self.erasure.sets[i].len(),
other.erasure.sets[i].len()
)));
}
self.find_disk_index_by_disk_id(other.erasure.this).map(|_| ())
}
for j in 0..self.erasure.sets[i].len() {
if self.erasure.sets[i][j] != other.erasure.sets[i][j] {
return Err(Error::other(format!(
"UUID on positions {}:{} do not match with, expected {:?} got {:?}: (%w)",
i,
j,
self.erasure.sets[i][j].to_string(),
other.erasure.sets[i][j].to_string(),
)));
}
}
}
for i in 0..tmp.erasure.sets.len() {
for j in 0..tmp.erasure.sets[i].len() {
if this == tmp.erasure.sets[i][j] {
return Ok(());
}
}
}
Err(Error::other(format!(
"DriveID {:?} not found in any drive sets {:?}",
this, other.erasure.sets
)))
/// Fields that must agree across every disk in one erasure format,
/// excluding the disk-specific `this` UUID and runtime-only `disk_info`.
pub(crate) fn shared_identity(&self) -> SharedFormatIdentity<'_> {
(
&self.version,
&self.format,
&self.id,
&self.erasure.version,
&self.erasure.sets,
&self.erasure.distribution_algo,
)
}
}
@@ -437,6 +418,30 @@ mod test {
assert!(result.is_ok());
}
#[test]
fn test_check_other_rejects_shared_identity_mismatches() {
type FormatMutation = (&'static str, fn(&mut FormatV3));
let format = FormatV3::new(1, 2);
let mutations: [FormatMutation; 5] = [
("meta version", |other| other.version = FormatMetaVersion::Unknown),
("backend", |other| other.format = FormatBackend::ErasureSingle),
("deployment id", |other| other.id = Uuid::new_v4()),
("erasure version", |other| other.erasure.version = FormatErasureVersion::V2),
("distribution algorithm", |other| {
other.erasure.distribution_algo = DistributionAlgoVersion::V2
}),
];
for (field, mutate) in mutations {
let mut other = format.clone();
other.erasure.this = format.erasure.sets[0][0];
mutate(&mut other);
assert!(format.check_other(&other).is_err(), "{field} mismatch must be rejected");
}
}
#[test]
fn test_check_other_different_set_count() {
let format1 = FormatV3::new(2, 4);
+23 -2
View File
@@ -424,7 +424,11 @@ pub(crate) async fn record_local_disk_id(instance_ctx: &Arc<InstanceContext>, di
pub(crate) async fn replace_local_disk_id(previous: Option<Uuid>, current: Option<Uuid>, endpoint: String) {
let id_map = local_disk_id_map_handle();
let mut disk_id_map = id_map.write().await;
if let Some(previous_id) = previous {
if let Some(previous_id) = previous
&& disk_id_map
.get(&previous_id)
.is_some_and(|registered_endpoint| registered_endpoint == &endpoint)
{
disk_id_map.remove(&previous_id);
}
if let Some(current_id) = current {
@@ -558,10 +562,14 @@ pub(crate) async fn init_tier_config_mgr(store: Arc<ECStore>) -> Result<()> {
#[cfg(test)]
mod tests {
use super::{LockRegistry, local_node_name, set_local_node_name};
use super::{
LockRegistry, clear_local_disk_id_map_for_test, local_disk_path_by_id, local_node_name, replace_local_disk_id,
set_local_node_name,
};
use crate::disk::endpoint::Endpoint;
use rustfs_lock::{LocalClient, LockClient};
use std::{collections::HashMap, sync::Arc};
use uuid::Uuid;
fn url_endpoint(raw: &str) -> Endpoint {
Endpoint {
@@ -607,4 +615,17 @@ mod tests {
assert_eq!(observed, next);
}
#[tokio::test]
#[serial_test::serial]
async fn clearing_a_stale_disk_id_does_not_remove_another_endpoint() {
clear_local_disk_id_map_for_test().await;
let disk_id = Uuid::new_v4();
replace_local_disk_id(None, Some(disk_id), "endpoint-a".to_string()).await;
replace_local_disk_id(Some(disk_id), None, "endpoint-b".to_string()).await;
assert_eq!(local_disk_path_by_id(&disk_id).await, Some("endpoint-a".to_string()));
clear_local_disk_id_map_for_test().await;
}
}
@@ -2044,15 +2044,10 @@ fn synthesized_disks(host: &str, endpoints: &EndpointServerPools, state: ItemSta
/// Whether `peer_host` refers to the same node as an endpoint whose
/// `host_port()` is `ep_host_port`.
///
/// `PeerRestClient::host` is an `XHost`, which resolves names to an address on
/// construction (`hosts_sorted` -> `XHost::try_from` -> `to_socket_addrs`), so
/// `peer_host` is the resolved `IP:port`. An endpoint's `host_port()`, however,
/// is `url.host():port` — still the raw `hostname:port` on hostname-based
/// deployments. A plain string compare therefore misses on hostname clusters,
/// leaving the synthesized/degraded drive list empty and `unknownDisks` at 0
/// (rustfs/rustfs#4607 follow-up). Compare directly first (fast path / IP
/// deployments), then canonicalize the endpoint side through the same `XHost`
/// resolution and compare again.
/// Current topology clients preserve the endpoint `hostname:port`, so the
/// direct comparison is the normal path. The resolution fallback keeps
/// compatibility with older or manually constructed clients whose `XHost`
/// contains a resolved `IP:port` (rustfs/rustfs#4607 follow-up).
fn endpoint_host_matches(peer_host: &str, ep_host_port: &str) -> bool {
if peer_host == ep_host_port {
return true;
+71 -3
View File
@@ -72,6 +72,7 @@ use crate::{
cluster::rpc::peer_rest_client::{PeerRestClient, PeerTierMutationState},
config::com::{CONFIG_PREFIX, read_config, read_config_with_metadata},
disk::{MIGRATING_META_BUCKET, RUSTFS_META_BUCKET},
layout::endpoints::EndpointServerPools,
object_api::{GetObjectReader, ObjectInfo, ObjectOptions, PutObjReader},
runtime::sources as runtime_sources,
set_disk::get_lock_acquire_timeout,
@@ -904,14 +905,17 @@ async fn remote_tier_mutation_peers() -> io::Result<Vec<Arc<dyn TierMutationPeer
let Some(endpoints) = runtime_sources::endpoint_pools() else {
return Err(tier_mutation_replay_error("cluster endpoint topology is not initialized"));
};
let remote_host_count = endpoints.hosts_sorted().iter().flatten().count();
let (peers, _) = PeerRestClient::new_clients(endpoints).await;
remote_tier_mutation_peers_from_topology(endpoints).await
}
async fn remote_tier_mutation_peers_from_topology(endpoints: EndpointServerPools) -> io::Result<Vec<Arc<dyn TierMutationPeer>>> {
let (peers, _, remote_topology_hosts) = PeerRestClient::new_clients_with_topology(endpoints).await;
let peers = peers
.into_iter()
.flatten()
.map(|peer| Arc::new(peer) as Arc<dyn TierMutationPeer>)
.collect::<Vec<_>>();
ensure_complete_tier_mutation_commit_peer_set(peers.len(), remote_host_count)?;
ensure_complete_tier_mutation_commit_peer_set(peers.len(), remote_topology_hosts.len())?;
Ok(peers)
}
@@ -4307,6 +4311,34 @@ fn tier_config_not_initialized_error(operation: &str) -> std::io::Error {
#[cfg(test)]
mod tests {
use super::*;
use crate::layout::{
endpoint::Endpoint,
endpoints::{Endpoints, PoolEndpoints, SetupType},
};
struct SetupTypeGuard {
previous: SetupType,
}
impl SetupTypeGuard {
async fn switch_to(next: SetupType) -> Self {
let previous = runtime_sources::current_setup_type().await;
runtime_sources::set_setup_type(next).await;
Self { previous }
}
}
impl Drop for SetupTypeGuard {
fn drop(&mut self) {
let previous = self.previous.clone();
let handle = tokio::runtime::Handle::current();
tokio::task::block_in_place(|| {
handle.block_on(async move {
runtime_sources::set_setup_type(previous).await;
});
});
}
}
fn build_s3_tier(name: &str) -> TierConfig {
TierConfig {
@@ -6354,6 +6386,42 @@ mod tests {
assert!(err.to_string().contains("without peer commit clients"), "{err}");
}
#[tokio::test(flavor = "multi_thread")]
#[serial_test::serial]
async fn tier_mutation_peer_composition_preserves_unresolved_topology_slots() {
let mut endpoints = Vec::new();
for disk_index in 0..4 {
let mut endpoint = Endpoint::try_from(format!("http://rustfs-{disk_index}.invalid:9000/data{disk_index}").as_str())
.expect("unresolved topology endpoint should parse without DNS");
endpoint.is_local = disk_index == 0;
endpoint.set_pool_index(0);
endpoint.set_set_index(0);
endpoint.set_disk_index(disk_index);
endpoints.push(endpoint);
}
let topology = EndpointServerPools::from(vec![PoolEndpoints {
legacy: false,
set_count: 1,
drives_per_set: 4,
endpoints: Endpoints::from(endpoints),
cmd_line: "unresolved-tier-mutation-topology".to_string(),
platform: "test".to_string(),
}]);
let _setup_type_guard = SetupTypeGuard::switch_to(SetupType::DistErasure).await;
let peers = remote_tier_mutation_peers_from_topology(topology)
.await
.expect("every unresolved remote topology slot should retain a tier mutation client");
assert_eq!(
peers.iter().map(|peer| peer.peer_label()).collect::<Vec<_>>(),
vec![
"http://rustfs-1.invalid:9000".to_string(),
"http://rustfs-2.invalid:9000".to_string(),
"http://rustfs-3.invalid:9000".to_string(),
]
);
}
#[tokio::test]
async fn coordinator_fanout_prepare_failure_aborts_prepared_peers_without_cas() {
let manager = TierConfigMgr::new();
+25 -18
View File
@@ -110,7 +110,10 @@ use crate::{
object_api::{GetObjectReader, ObjectInfo, PutObjReader},
// event::name::EventName,
services::event_notification::{EventArgs, send_event},
store::init_format::{get_format_erasure_in_quorum, load_format_erasure, load_format_erasure_all, save_format_file},
store::init_format::{
formats_match_reference_slots, get_format_erasure_in_quorum, load_format_erasure, load_format_erasure_all,
save_format_file,
},
};
use bytes::Bytes;
use bytesize::ByteSize;
@@ -5849,23 +5852,27 @@ mod tests {
crate::disk::DataDirDeleteStatus::Deleted
);
release_slow_candidate.notify_one();
for _ in 0..10 {
tokio::task::yield_now().await;
}
assert_eq!(
disk2
.delete_data_dir(
bucket,
data_dir,
DeleteOptions {
recursive: true,
..Default::default()
},
)
.await
.expect("a token acquired after the deadline must be released"),
crate::disk::DataDirDeleteStatus::Deleted
);
timeout(Duration::from_secs(1), async {
loop {
match disk2
.delete_data_dir(
bucket,
data_dir,
DeleteOptions {
recursive: true,
..Default::default()
},
)
.await
.expect("a token acquired after the deadline must be released")
{
crate::disk::DataDirDeleteStatus::Deleted => return,
crate::disk::DataDirDeleteStatus::Deferred => tokio::time::sleep(Duration::from_millis(1)).await,
}
}
})
.await
.expect("late snapshot lease cleanup must finish within the bounded wait");
}
#[tokio::test(start_paused = true)]
+63 -4
View File
@@ -1373,11 +1373,24 @@ impl crate::storage_api_contracts::heal::HealOperations for SetDisks {
async fn heal_format(&self, dry_run: bool) -> Result<(HealResultItem, Option<Error>)> {
let disks = self.disks.read().await.clone();
let (formats, errs) = load_format_erasure_all(&disks, true).await;
let ref_format = match get_format_erasure_in_quorum(&formats) {
Ok(format) => format,
if errs.iter().any(|err| {
matches!(
err,
Some(DiskError::InconsistentDisk | DiskError::CorruptedFormat | DiskError::CorruptedBackend)
)
}) {
return Ok((HealResultItem::default(), Some(StorageError::CorruptedFormat)));
}
let slot_offset = self
.set_index
.checked_mul(self.set_drive_count)
.ok_or_else(|| Error::other("erasure set slot offset overflow"))?;
let ref_format = match get_format_erasure_in_quorum(&formats, slot_offset) {
Ok(format) if format.shared_identity() == self.format.shared_identity() => format,
Ok(_) => return Ok((HealResultItem::default(), Some(StorageError::CorruptedFormat))),
Err(err) => {
let can_use_cached_layout = count_errs(&errs, &DiskError::UnformattedDisk) > 0
&& formats.iter().flatten().all(|format| self.format.check_other(format).is_ok())
&& formats_match_reference_slots(&formats, &self.format, slot_offset)
&& errs
.iter()
.all(|err| err.is_none() || matches!(err, Some(DiskError::UnformattedDisk)));
@@ -1388,6 +1401,9 @@ impl crate::storage_api_contracts::heal::HealOperations for SetDisks {
}
}
};
if !formats_match_reference_slots(&formats, &ref_format, slot_offset) {
return Ok((HealResultItem::default(), Some(StorageError::CorruptedFormat)));
}
let endpoints = crate::layout::endpoints::Endpoints::from(self.set_endpoints.clone());
let before_drives = crate::layout::set_heal::formats_to_drives_info(&endpoints, &formats, &errs);
@@ -1543,11 +1559,16 @@ mod heal_result_report_tests {
use crate::disk::error::DiskError;
use crate::disk::format::FormatV3;
use crate::disk::{DiskAPI as _, DiskOption, DiskStore, RUSTFS_META_TMP_BUCKET, ReadOptions, new_disk};
use crate::error::Error;
use crate::object_api::{ObjectOptions, PutObjReader};
use crate::set_disk::ops::object::hermetic_set_disks_support::hermetic_set_disks_isolated;
use crate::storage_api_contracts::bucket::{BucketOperations as _, MakeBucketOptions};
use crate::storage_api_contracts::heal::HealOperations as _;
use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations as _};
use crate::{config::storageclass, store::init_format::save_format_file};
use crate::{
config::storageclass,
store::init_format::{load_format_erasure, save_format_file},
};
use rustfs_common::heal_channel::{DriveState, HealOpts, HealScanMode};
use rustfs_filemeta::{BLOCK_SIZE_V2, FileInfo, ObjectPartInfo, TRANSITION_COMPLETE};
use std::sync::Arc;
@@ -1863,6 +1884,44 @@ mod heal_result_report_tests {
}
}
#[tokio::test]
async fn format_heal_cached_layout_rejects_a_disk_from_another_slot() {
let mut _temp_dirs = Vec::new();
let mut endpoints = Vec::new();
let mut disks = Vec::new();
for disk_index in 0..3 {
let (temp_dir, mut endpoint, disk) = real_disk().await;
endpoint.set_pool_index(0);
endpoint.set_set_index(0);
endpoint.set_disk_index(disk_index);
_temp_dirs.push(temp_dir);
endpoints.push(endpoint);
disks.push(Some(disk));
}
let set = set_disks_with(disks.clone(), endpoints, 1).await;
let mut wrong_slot = set.format.clone();
wrong_slot.erasure.this = set.format.erasure.sets[0][1];
save_format_file(&disks[0], &Some(wrong_slot))
.await
.expect("wrong-slot format fixture should be saved");
let mut correct_slot = set.format.clone();
correct_slot.erasure.this = set.format.erasure.sets[0][2];
save_format_file(&disks[2], &Some(correct_slot))
.await
.expect("correct format fixture should be saved");
let (_, heal_err) = set
.heal_format(false)
.await
.expect("format heal should report the quorum failure in its result");
assert!(matches!(heal_err, Some(Error::CorruptedFormat)));
let unformatted = load_format_erasure(disks[1].as_ref().expect("second disk should be online"), true)
.await
.expect_err("a rejected fallback must not format the missing slot");
assert_eq!(unformatted, DiskError::UnformattedDisk);
}
// Regression for #955: an offline disk must contribute exactly one drive
// record. Before the fix the offline branch fell through and pushed a second
// (Corrupt) record for the same disk, so `before/after.drives` grew to
+114 -9
View File
@@ -343,20 +343,23 @@ impl SetDisks {
}
};
// The drive's format may place it in a different erasure set than this
// one. Claiming a misplaced drive into `self.disks` would let two sets
// manage the same drive and degrade together, so reject it here
// (backlog#799 B19).
if set_idx != self.set_index {
// Claiming a misplaced drive into `self.disks` would let two slots or
// sets manage the same drive and degrade together (backlog#799 B19).
if set_idx != self.set_index || self.set_endpoints.get(disk_idx) != Some(ep) {
warn!(
"renew_disk: drive {:?} belongs to set {} but is being renewed on set {}; skipping",
ep, set_idx, self.set_index
endpoint = %ep,
format_set_index = set_idx,
format_disk_index = disk_idx,
endpoint_pool_index = ep.pool_idx,
endpoint_set_index = ep.set_idx,
endpoint_disk_index = ep.disk_idx,
expected_pool_index = self.pool_index,
expected_set_index = self.set_index,
"renew_disk rejected a drive whose endpoint and format do not identify the same topology slot"
);
return;
}
// Check that the endpoint matches
let _ = new_disk.set_disk_id(Some(fm.erasure.this)).await;
new_disk.enable_health_check();
@@ -715,6 +718,108 @@ mod tests {
drop(temp_dirs);
}
#[tokio::test]
async fn renew_disk_rejects_a_format_from_another_slot_or_cluster() {
let disk_count = 3;
let format = FormatV3::new(1, disk_count);
let mut temp_dirs = Vec::with_capacity(disk_count);
let mut endpoints = Vec::with_capacity(disk_count);
let mut fixture_disks = Vec::with_capacity(disk_count);
for disk_idx in 0..disk_count {
let (temp_dir, endpoint, disk) = make_formatted_local_disk(disk_idx, &format).await;
temp_dirs.push(temp_dir);
endpoints.push(endpoint);
fixture_disks.push(disk);
}
let set_disks = SetDisks::new(
"test-owner".to_string(),
Arc::new(RwLock::new(vec![Some(fixture_disks[0].clone()), None, None])),
disk_count,
disk_count / 2,
0,
0,
endpoints.clone(),
format.clone(),
Vec::new(),
)
.await;
let mut other_cluster_format = format.clone();
other_cluster_format.id = Uuid::new_v4();
other_cluster_format.erasure.this = format.erasure.sets[0][2];
save_format_file(&Some(fixture_disks[2].clone()), &Some(other_cluster_format))
.await
.expect("other-cluster format should be written for the rejection test");
set_disks.renew_disk(&endpoints[2]).await;
let disks = set_disks.get_disks_internal().await;
assert_eq!(
disks[0]
.as_ref()
.expect("the canonical first slot must remain attached")
.endpoint(),
endpoints[0]
);
assert!(
disks[2].is_none(),
"a disk from another deployment must remain detached even when its slot UUID matches"
);
let mut correct_format = format.clone();
correct_format.erasure.this = format.erasure.sets[0][2];
let replacement_disk = new_disk(
&endpoints[2],
&DiskOption {
cleanup: false,
health_check: false,
},
)
.await
.expect("third endpoint should reopen after other-cluster rejection");
save_format_file(&Some(replacement_disk), &Some(correct_format))
.await
.expect("correct slot format should be restored");
set_disks.renew_disk(&endpoints[2]).await;
let disks = set_disks.get_disks_internal().await;
assert_eq!(
disks[0]
.as_ref()
.expect("the canonical first slot must remain attached")
.endpoint(),
endpoints[0]
);
assert_eq!(disks[2].as_ref().expect("the restored third slot should attach").endpoint(), endpoints[2]);
let third_disk = disks[2].clone();
let mut wrong_slot_format = format.clone();
wrong_slot_format.erasure.this = format.erasure.sets[0][0];
save_format_file(&third_disk, &Some(wrong_slot_format))
.await
.expect("wrong-slot format should be written for the rejection test");
set_disks.disks.write().await[2] = None;
let mut misplaced_endpoint = endpoints[2].clone();
misplaced_endpoint.set_disk_index(0);
set_disks.renew_disk(&misplaced_endpoint).await;
let disks = set_disks.get_disks_internal().await;
assert_eq!(
disks[0]
.as_ref()
.expect("the canonical first slot must remain attached")
.endpoint(),
endpoints[0]
);
assert!(disks[2].is_none(), "a disk claiming another endpoint's slot must remain detached");
drop(temp_dirs);
}
// SetDisks split P0 (#816): the borrow handle must mirror the core state and
// the List operation family must run identically through it.
#[tokio::test]
+4 -3
View File
@@ -310,12 +310,13 @@ impl ECStore {
meta.set_created(opts.created_at);
if opts.lock_enabled {
meta.object_lock_config_xml = crate::bucket::utils::serialize::<ObjectLockConfiguration>(&enableObjcetLockConfig)?;
meta.versioning_config_xml = crate::bucket::utils::serialize::<VersioningConfiguration>(&enableVersioningConfig)?;
meta.object_lock_config_xml =
crate::bucket::utils::serialize::<ObjectLockConfiguration>(&ENABLED_OBJECT_LOCK_CONFIG)?;
meta.versioning_config_xml = crate::bucket::utils::serialize::<VersioningConfiguration>(&ENABLED_VERSIONING_CONFIG)?;
}
if opts.versioning_enabled {
meta.versioning_config_xml = crate::bucket::utils::serialize::<VersioningConfiguration>(&enableVersioningConfig)?;
meta.versioning_config_xml = crate::bucket::utils::serialize::<VersioningConfiguration>(&ENABLED_VERSIONING_CONFIG)?;
}
await_bucket_namespace_operation(
+137 -2
View File
@@ -30,6 +30,7 @@ impl ECStore {
};
let mut count_no_heal = 0;
let mut first_error = None;
for pool in self.pools.iter() {
let (mut result, err) = pool.heal_format(dry_run).await?;
if let Some(err) = err {
@@ -37,8 +38,8 @@ impl ECStore {
StorageError::NoHealRequired => {
count_no_heal += 1;
}
_ => {
continue;
err => {
first_error.get_or_insert(err);
}
}
}
@@ -47,6 +48,9 @@ impl ECStore {
r.before.drives.append(&mut result.before.drives);
r.after.drives.append(&mut result.after.drives);
}
if let Some(err) = first_error {
return Ok((r, Some(err)));
}
if count_no_heal == self.pools.len() {
info!(
event = EVENT_HEAL_FORMAT_COMPLETED,
@@ -165,3 +169,134 @@ impl ECStore {
Err(StorageError::NotImplemented)
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::disk::{DiskOption, format::FormatV3, new_disk};
use crate::layout::endpoints::{Endpoints, PoolEndpoints};
use crate::store::init_format::{load_format_erasure, save_format_file};
#[tokio::test]
async fn handle_heal_format_continues_after_a_pool_error() {
let canonical_format = FormatV3::new(1, 3);
let mut foreign_format = canonical_format.clone();
foreign_format.id = Uuid::new_v4();
let mut temp_dirs = Vec::new();
let mut endpoints = Vec::new();
let mut disks = Vec::new();
for disk_index in 0..3 {
let temp_dir = tempfile::tempdir().expect("temporary disk root should be created");
let mut endpoint = Endpoint::try_from(temp_dir.path().to_str().expect("temporary path should be UTF-8"))
.expect("temporary endpoint should parse");
endpoint.set_pool_index(0);
endpoint.set_set_index(0);
endpoint.set_disk_index(disk_index);
let disk = new_disk(
&endpoint,
&DiskOption {
cleanup: false,
health_check: false,
},
)
.await
.expect("temporary disk should open");
let mut disk_format = foreign_format.clone();
disk_format.erasure.this = foreign_format.erasure.sets[0][disk_index];
save_format_file(&Some(disk.clone()), &Some(disk_format))
.await
.expect("foreign format should be written");
temp_dirs.push(temp_dir);
endpoints.push(endpoint);
disks.push(Some(disk));
}
let pool_endpoints = PoolEndpoints {
legacy: false,
set_count: 1,
drives_per_set: 3,
endpoints: Endpoints::from(endpoints),
cmd_line: "foreign-format-majority-test".to_string(),
platform: "test".to_string(),
};
let pool = Sets::new(disks, &pool_endpoints, &canonical_format, 0, 1)
.await
.expect("test pool should build around the cached canonical format");
let mut recoverable_format = FormatV3::new(1, 3);
recoverable_format.id = canonical_format.id;
let mut recoverable_temp_dirs = Vec::new();
let mut recoverable_endpoints = Vec::new();
let mut recoverable_disks = Vec::new();
let mut unformatted_disk = None;
for disk_index in 0..3 {
let temp_dir = tempfile::tempdir().expect("temporary disk root should be created");
let mut endpoint = Endpoint::try_from(temp_dir.path().to_str().expect("temporary path should be UTF-8"))
.expect("temporary endpoint should parse");
endpoint.set_pool_index(1);
endpoint.set_set_index(0);
endpoint.set_disk_index(disk_index);
let disk = new_disk(
&endpoint,
&DiskOption {
cleanup: false,
health_check: false,
},
)
.await
.expect("temporary disk should open");
if disk_index < 2 {
let mut disk_format = recoverable_format.clone();
disk_format.erasure.this = recoverable_format.erasure.sets[0][disk_index];
save_format_file(&Some(disk.clone()), &Some(disk_format))
.await
.expect("recoverable format should be written");
} else {
unformatted_disk = Some(disk.clone());
}
recoverable_temp_dirs.push(temp_dir);
recoverable_endpoints.push(endpoint);
recoverable_disks.push(Some(disk));
}
let recoverable_pool_endpoints = PoolEndpoints {
legacy: false,
set_count: 1,
drives_per_set: 3,
endpoints: Endpoints::from(recoverable_endpoints),
cmd_line: "recoverable-format-test".to_string(),
platform: "test".to_string(),
};
let recoverable_pool = Sets::new(recoverable_disks, &recoverable_pool_endpoints, &recoverable_format, 1, 1)
.await
.expect("recoverable test pool should build");
let endpoint_pools = EndpointServerPools::from(vec![pool_endpoints.clone(), recoverable_pool_endpoints.clone()]);
let store = ECStore {
id: canonical_format.id,
disk_map: HashMap::new(),
pools: vec![pool, recoverable_pool],
peer_sys: S3PeerSys::new(&endpoint_pools),
pool_meta: RwLock::new(PoolMeta::default()),
rebalance_meta: RwLock::new(None),
decommission_cancelers: RwLock::new(Vec::new()),
start_gate: Mutex::new(()),
pool_meta_save_gate: Mutex::new(()),
ctx: crate::runtime::instance::bootstrap_ctx(),
};
let (result, err) = store
.handle_heal_format(false)
.await
.expect("format heal should return the typed pool error");
assert!(
matches!(err, Some(StorageError::CorruptedFormat)),
"foreign format majority must not be downgraded to a successful heal: {err:?}"
);
assert_eq!(result.disk_count, 3, "the recoverable pool should still be inspected");
let healed = load_format_erasure(&unformatted_disk.expect("the unformatted disk handle should be retained"), true)
.await
.expect("the later pool should be healed despite the first pool error");
assert_eq!(healed.erasure.this, recoverable_format.erasure.sets[0][2]);
}
}
+15 -3
View File
@@ -101,6 +101,10 @@ fn should_retry_local_decommission_resume(err: &Error, attempt: usize) -> bool {
matches!(err, Error::ConfigNotFound) && attempt < LOCAL_DECOMMISSION_RESUME_MAX_CONFIG_RETRIES
}
fn should_retry_format_load(err: &Error) -> bool {
!matches!(err, Error::CorruptedFormat)
}
fn should_auto_start_rebalance_after_init(decommission_running: bool, rebalance_meta_loaded: bool) -> bool {
rebalance_meta_loaded && !decommission_running
}
@@ -294,7 +298,7 @@ impl ECStore {
// periodic monitoring until format loading succeeds. Startup RPC
// failures can still spawn recovery probes for peers that come up
// after this node.
let (disks, errs) = init_format::init_disks(
let (mut disks, errs) = init_format::init_disks(
&pool_eps.endpoints,
&DiskOption {
cleanup: true,
@@ -311,7 +315,7 @@ impl ECStore {
loop {
match init_format::connect_load_init_formats(
pool_first_is_local,
&disks,
&mut disks,
pool_eps.set_count,
pool_eps.drives_per_set,
deployment_id,
@@ -319,6 +323,7 @@ impl ECStore {
.await
{
Ok(fm) => break Ok(fm),
Err(e) if !should_retry_format_load(&e) => break Err(e),
// Wrap the final error if we are giving up
Err(e) if times >= 10 => {
break Err(Error::other(format!("store init failed to load formats after {times} retries: {e}")));
@@ -551,7 +556,7 @@ mod tests {
LOCAL_DECOMMISSION_RESUME_MAX_CONFIG_RETRIES, load_pool_meta_for_startup, pool_first_endpoint_is_local,
pool_meta_has_active_decommission, preflight_startup_rpc_secret_with, resolve_startup_pool_defaults_with,
resolve_store_init_stage_result, save_validated_pool_meta_for_startup, should_auto_start_rebalance_after_init,
should_retry_local_decommission_resume, wait_for_local_decommission_resume_delay,
should_retry_format_load, should_retry_local_decommission_resume, wait_for_local_decommission_resume_delay,
};
#[cfg(feature = "test-util")]
use crate::{
@@ -773,6 +778,13 @@ mod tests {
assert!(!should_retry_local_decommission_resume(&StorageError::SlowDown, 0));
}
#[test]
fn test_should_retry_format_load_rejects_permanent_corruption() {
assert!(!should_retry_format_load(&StorageError::CorruptedFormat));
assert!(should_retry_format_load(&StorageError::ErasureReadQuorum));
assert!(should_retry_format_load(&StorageError::FirstDiskWait));
}
#[test]
fn test_should_auto_start_rebalance_after_init_allows_loaded_rebalance_without_decommission() {
assert!(should_auto_start_rebalance_after_init(false, true));
File diff suppressed because it is too large Load Diff
+4 -4
View File
@@ -428,11 +428,11 @@ impl ECStore {
}
lazy_static! {
static ref enableObjcetLockConfig: ObjectLockConfiguration = ObjectLockConfiguration {
static ref ENABLED_OBJECT_LOCK_CONFIG: ObjectLockConfiguration = ObjectLockConfiguration {
object_lock_enabled: Some(ObjectLockEnabled::from_static(ObjectLockEnabled::ENABLED)),
..Default::default()
};
static ref enableVersioningConfig: VersioningConfiguration = VersioningConfiguration {
static ref ENABLED_VERSIONING_CONFIG: VersioningConfiguration = VersioningConfiguration {
status: Some(BucketVersioningStatus::from_static(BucketVersioningStatus::ENABLED)),
..Default::default()
};
@@ -989,7 +989,7 @@ mod tests {
init_local_disks(endpoint_pools.clone()).await.expect("init local disks");
let (disks, errs) = init_disks(
let (mut disks, errs) = init_disks(
&endpoint_pools.as_ref().first().expect("pool endpoints").endpoints,
&DiskOption {
cleanup: true,
@@ -999,7 +999,7 @@ mod tests {
.await;
assert!(errs.iter().all(|err| err.is_none()), "disk init should succeed: {errs:?}");
connect_load_init_formats(true, &disks, 1, 4, None)
connect_load_init_formats(true, &mut disks, 1, 4, None)
.await
.expect("initialize format metadata");
+27 -3
View File
@@ -556,7 +556,7 @@ where
Ok(now)
}
pub async fn list_polices(&self, bucket_name: &str) -> Result<HashMap<String, Policy>> {
pub async fn list_policies(&self, bucket_name: &str) -> Result<HashMap<String, Policy>> {
let mut m = HashMap::new();
self.api.load_policy_docs(&mut m).await?;
@@ -588,6 +588,15 @@ where
Ok(filtered)
}
/// Backward-compatible misspelling retained until the next breaking release.
#[deprecated(
since = "1.0.0",
note = "use list_policies instead; this alias will be removed in the next breaking release"
)]
pub async fn list_polices(&self, bucket_name: &str) -> Result<HashMap<String, Policy>> {
self.list_policies(bucket_name).await
}
pub async fn merge_policies(&self, name: &str) -> (String, Policy) {
let mut policies = Vec::new();
let mut to_merge = Vec::new();
@@ -2184,7 +2193,7 @@ where
}
}
pub fn get_default_policyes() -> HashMap<String, PolicyDoc> {
pub fn get_default_policies() -> HashMap<String, PolicyDoc> {
let default_policies = &DEFAULT_POLICIES;
default_policies
.iter()
@@ -2201,6 +2210,15 @@ pub fn get_default_policyes() -> HashMap<String, PolicyDoc> {
.collect()
}
/// Backward-compatible misspelling retained until the next breaking release.
#[deprecated(
since = "1.0.0",
note = "use get_default_policies instead; this alias will be removed in the next breaking release"
)]
pub fn get_default_policyes() -> HashMap<String, PolicyDoc> {
get_default_policies()
}
fn set_default_canned_policies(policies: &mut HashMap<String, PolicyDoc>) {
let default_policies = &DEFAULT_POLICIES;
for (k, v) in default_policies.iter() {
@@ -2900,7 +2918,7 @@ mod tests {
#[test]
fn test_get_default_policies() {
let policies = get_default_policyes();
let policies = get_default_policies();
// Should contain some default policies
assert!(!policies.is_empty());
@@ -2913,6 +2931,12 @@ mod tests {
}
}
#[test]
#[allow(deprecated)]
fn deprecated_get_default_policyes_matches_current_api() {
assert_eq!(get_default_policyes().len(), get_default_policies().len());
}
#[test]
fn test_get_token_signing_key() {
// This function returns the global action credential's secret key
+2 -2
View File
@@ -22,7 +22,7 @@ use crate::{
cache::{Cache, CacheEntity},
error::{is_err_no_such_policy, is_err_no_such_user},
keyring,
manager::{extract_jwt_claims, extract_jwt_claims_allow_missing_exp, get_default_policyes},
manager::{extract_jwt_claims, extract_jwt_claims_allow_missing_exp, get_default_policies},
root_credentials,
};
use futures::future::join_all;
@@ -1127,7 +1127,7 @@ impl Store for ObjectStore {
let cache_snapshot = cache.snapshot();
let listed_config_items = self.list_all_iamconfig_items().await?;
let mut policy_docs_cache = CacheEntity::new(get_default_policyes());
let mut policy_docs_cache = CacheEntity::new(get_default_policies());
if let Some(policies_list) = listed_config_items.get(POLICIES_LIST_KEY) {
// Load in fixed-size chunks so each policy is fetched exactly once.
+20 -5
View File
@@ -18,7 +18,7 @@ use crate::error::is_err_no_such_temp_account;
use crate::error::{Error, Result};
use crate::federation::OIDC_VIRTUAL_PARENT_CLAIM;
use crate::manager::extract_jwt_claims;
use crate::manager::get_default_policyes;
use crate::manager::get_default_policies;
use crate::manager::{IamCache, IamSyncMetricsSnapshot};
use crate::store::GroupInfo;
use crate::store::MappedPolicy;
@@ -249,7 +249,7 @@ impl<T: Store> IamSys<T> {
}
pub async fn delete_policy(&self, name: &str, notify: bool) -> Result<()> {
for k in get_default_policyes().keys() {
for k in get_default_policies().keys() {
if k == name {
return Err(Error::other("system policy can not be deleted"));
}
@@ -291,8 +291,17 @@ impl<T: Store> IamSys<T> {
self.store.api.load_mapped_policies(user_type, is_group, m).await
}
pub async fn list_policies(&self, bucket_name: &str) -> Result<HashMap<String, Policy>> {
self.store.list_policies(bucket_name).await
}
/// Backward-compatible misspelling retained until the next breaking release.
#[deprecated(
since = "1.0.0",
note = "use list_policies instead; this alias will be removed in the next breaking release"
)]
pub async fn list_polices(&self, bucket_name: &str) -> Result<HashMap<String, Policy>> {
self.store.list_polices(bucket_name).await
self.list_policies(bucket_name).await
}
pub async fn list_policy_docs(&self, bucket_name: &str) -> Result<HashMap<String, PolicyDoc>> {
@@ -1683,11 +1692,17 @@ mod tests {
use super::*;
use crate::cache::{Cache, CacheEntity};
use crate::error::Error;
use crate::manager::get_default_policyes;
use crate::manager::get_default_policies;
use crate::store::{GroupInfo, MappedPolicy, Store, UserType};
use rustfs_credentials::{Credentials, init_global_action_credentials};
use rustfs_policy::auth::{UserIdentity, get_new_credentials_with_metadata};
use rustfs_policy::policy::Args;
#[test]
#[allow(deprecated)]
fn deprecated_list_polices_api_is_available() {
let _ = IamSys::<StsTestMockStore>::list_polices;
}
use rustfs_policy::policy::action::{Action, AdminAction, S3Action};
use rustfs_policy::policy::policy_uses_existing_object_tag_conditions;
use serde_json::Value;
@@ -1925,7 +1940,7 @@ mod tests {
}
async fn load_all(&self, cache: &Cache) -> Result<()> {
let mut policy_docs = get_default_policyes();
let mut policy_docs = get_default_policies();
let custom_claim_policy =
Policy::parse_config(CUSTOM_STS_CLAIM_POLICY_JSON.as_bytes()).expect("custom STS claim policy should parse");
policy_docs.insert(CUSTOM_STS_CLAIM_POLICY.to_string(), PolicyDoc::new(custom_claim_policy));
+280 -136
View File
@@ -18,7 +18,9 @@ use crate::rule::ReplicationRuleExt as _;
use s3s::dto::DeleteMarkerReplicationStatus;
use s3s::dto::DeleteReplicationStatus;
use s3s::dto::Destination;
use s3s::dto::{ExistingObjectReplicationStatus, ReplicationConfiguration, ReplicationRuleStatus, ReplicationRules};
use s3s::dto::{
ExistingObjectReplicationStatus, ReplicationConfiguration, ReplicationRule, ReplicationRuleStatus, ReplicationRules,
};
use serde::{Deserialize, Serialize};
use std::collections::HashSet;
use uuid::Uuid;
@@ -45,34 +47,92 @@ pub trait ReplicationConfigurationExt {
fn filter_target_arns(&self, obj: &ObjectOpts) -> Vec<String>;
}
pub fn delete_replication_target_arns(config: &ReplicationConfiguration, object_name: &str, replica: bool) -> HashSet<String> {
let role = config.role.trim();
if !role.is_empty() && active_replication_rule_destination_arns(config).len() > 1 {
return HashSet::new();
}
let mut targets = HashSet::new();
let mut targets_with_unknown_tags = HashSet::new();
for rule in &config.rules {
if rule.status == ReplicationRuleStatus::from_static(ReplicationRuleStatus::DISABLED)
|| !object_name.starts_with(rule.prefix())
{
continue;
}
let arn = if role.is_empty() {
rule.destination.bucket.trim()
} else {
role
};
if arn.is_empty() {
continue;
}
targets.insert(arn.to_string());
if rule.filter.as_ref().is_some_and(|filter| {
filter.tag.is_some()
|| filter
.and
.as_ref()
.and_then(|and| and.tags.as_ref())
.is_some_and(|tags| !tags.is_empty())
}) {
targets_with_unknown_tags.insert(arn.to_string());
}
}
targets
.into_iter()
.filter(|arn| !targets_with_unknown_tags.contains(arn))
.filter(|arn| {
config.replicate(&ObjectOpts {
name: object_name.to_string(),
target_arn: arn.clone(),
version_id: Some(Uuid::nil()),
delete_marker: true,
op_type: ReplicationType::Delete,
replica,
..Default::default()
})
})
.collect()
}
fn rule_replicates(rule: &ReplicationRule, obj: &ObjectOpts) -> bool {
if let Some(status) = &rule.existing_object_replication
&& obj.existing_object
&& status.status == ExistingObjectReplicationStatus::from_static(ExistingObjectReplicationStatus::DISABLED)
{
return false;
}
if obj.op_type == ReplicationType::Delete {
if !rule.metadata_replicate(obj) {
return false;
}
if obj.version_id.is_some() {
return rule
.delete_replication
.clone()
.is_some_and(|d| d.status == DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED));
}
return rule.delete_marker_replication.clone().is_some_and(|d| {
d.status == Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::ENABLED))
});
}
rule.metadata_replicate(obj)
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ReplicationTargetValidationError {
RoleWithMultipleDestinations,
StaleTarget,
}
pub fn unsupported_replication_config_field(config: &ReplicationConfiguration) -> Option<&'static str> {
for rule in &config.rules {
if rule
.source_selection_criteria
.as_ref()
.is_some_and(|criteria| criteria.sse_kms_encrypted_objects.is_some())
{
return Some("SourceSelectionCriteria.SseKmsEncryptedObjects");
}
if rule.destination.encryption_configuration.is_some() {
return Some("Destination.EncryptionConfiguration");
}
if rule.destination.metrics.is_some() {
return Some("Destination.Metrics");
}
if rule.destination.replication_time.is_some() {
return Some("Destination.ReplicationTime");
}
}
None
}
pub fn active_replication_rule_destination_arns(config: &ReplicationConfiguration) -> HashSet<String> {
let mut arns = HashSet::new();
@@ -226,32 +286,7 @@ impl ReplicationConfigurationExt for ReplicationConfiguration {
continue;
}
if let Some(status) = &rule.existing_object_replication
&& obj.existing_object
&& status.status == ExistingObjectReplicationStatus::from_static(ExistingObjectReplicationStatus::DISABLED)
{
return false;
}
if obj.op_type == ReplicationType::Delete {
if !rule.metadata_replicate(obj) {
return false;
}
if obj.version_id.is_some() {
return rule
.delete_replication
.clone()
.is_some_and(|d| d.status == DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED));
} else {
return rule.delete_marker_replication.clone().is_some_and(|d| {
d.status == Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::ENABLED))
});
}
}
// Regular object/metadata replication
return rule.metadata_replicate(obj);
return rule_replicates(rule, obj);
}
false
}
@@ -323,9 +358,8 @@ impl ReplicationConfigurationExt for ReplicationConfiguration {
mod tests {
use super::*;
use s3s::dto::{
DeleteMarkerReplication, DeleteReplication, Destination, EncryptionConfiguration, ExistingObjectReplication, Metrics,
MetricsStatus, ReplicationRule, ReplicationTime, ReplicationTimeStatus, ReplicationTimeValue, SourceSelectionCriteria,
SseKmsEncryptedObjects, SseKmsEncryptedObjectsStatus,
DeleteMarkerReplication, DeleteReplication, Destination, ExistingObjectReplication, ReplicationRule,
ReplicationRuleFilter, Tag,
};
fn replication_rule(id: &str, arn: &str) -> ReplicationRule {
@@ -348,6 +382,37 @@ mod tests {
}
}
#[test]
fn delete_replication_target_arns_uses_highest_priority_matching_rule() {
let arn = "arn:target:a";
let mut lower_priority = replication_rule("lower", arn);
lower_priority.priority = Some(1);
lower_priority.delete_replication = Some(DeleteReplication {
status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED),
});
let mut higher_priority = replication_rule("higher", arn);
higher_priority.priority = Some(2);
higher_priority.delete_replication = Some(DeleteReplication {
status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::DISABLED),
});
let mut config = ReplicationConfiguration {
role: String::new(),
rules: vec![lower_priority, higher_priority],
};
let targets = delete_replication_target_arns(&config, "object", false);
assert!(targets.is_empty(), "the higher-priority disabled rule must suppress the target");
config.rules[0].delete_replication = Some(DeleteReplication {
status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::DISABLED),
});
config.rules[1].delete_replication = Some(DeleteReplication {
status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED),
});
assert_eq!(delete_replication_target_arns(&config, "object", false), HashSet::from([arn.to_string()]));
}
#[test]
fn filter_target_arns_uses_role_when_role_is_present() {
let config = ReplicationConfiguration {
@@ -358,15 +423,151 @@ mod tests {
],
};
let arns = config.filter_target_arns(&ObjectOpts {
let opts = ObjectOpts {
name: "object".to_string(),
op_type: ReplicationType::Object,
..Default::default()
});
};
let arns = config.filter_target_arns(&opts);
assert_eq!(arns, vec!["arn:legacy:target".to_string()]);
}
#[test]
fn delete_replication_target_arns_uses_role_when_role_is_present() {
let mut rule = replication_rule("rule", "arn:target:a");
rule.delete_replication = Some(DeleteReplication {
status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED),
});
let config = ReplicationConfiguration {
role: " arn:legacy:target ".to_string(),
rules: vec![rule],
};
assert_eq!(
delete_replication_target_arns(&config, "object", false),
HashSet::from(["arn:legacy:target".to_string()])
);
}
#[test]
fn delete_replication_target_arns_ignores_disjoint_prefix_rules() {
let arn = "arn:target:a";
let mut matching = replication_rule("matching", arn);
matching.prefix = None;
matching.filter = Some(ReplicationRuleFilter {
prefix: Some("logs/".to_string()),
..Default::default()
});
matching.delete_replication = Some(DeleteReplication {
status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::DISABLED),
});
let mut unrelated = replication_rule("unrelated", arn);
unrelated.prefix = None;
unrelated.filter = Some(ReplicationRuleFilter {
prefix: Some("archive/".to_string()),
..Default::default()
});
unrelated.priority = Some(2);
unrelated.delete_replication = Some(DeleteReplication {
status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED),
});
let mut config = ReplicationConfiguration {
role: String::new(),
rules: vec![matching, unrelated],
};
assert!(delete_replication_target_arns(&config, "logs/object", false).is_empty());
config.rules[0].delete_replication = Some(DeleteReplication {
status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED),
});
config.rules[1].delete_replication = Some(DeleteReplication {
status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::DISABLED),
});
assert_eq!(
delete_replication_target_arns(&config, "logs/object", false),
HashSet::from([arn.to_string()])
);
}
#[test]
fn delete_replication_target_arns_fails_closed_for_unknown_tag_rules() {
let arn = "arn:target:a";
let mut known = replication_rule("known", arn);
known.priority = Some(2);
known.delete_replication = Some(DeleteReplication {
status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED),
});
let mut unknown = replication_rule("unknown", arn);
unknown.priority = Some(1);
unknown.prefix = None;
unknown.filter = Some(ReplicationRuleFilter {
tag: Some(Tag {
key: Some("env".to_string()),
value: Some("prod".to_string()),
}),
..Default::default()
});
unknown.delete_replication = Some(DeleteReplication {
status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED),
});
let config = ReplicationConfiguration {
role: String::new(),
rules: vec![known, unknown],
};
assert!(delete_replication_target_arns(&config, "object", false).is_empty());
}
#[test]
fn delete_replication_target_arns_reuses_full_destination_rule_order() {
let arn = "arn:target:a";
let mut first = replication_rule("first", arn);
first.priority = Some(1);
first.destination.account = Some("account-a".to_string());
first.delete_replication = Some(DeleteReplication {
status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::DISABLED),
});
let mut second = replication_rule("second", arn);
second.priority = Some(2);
second.destination.account = Some("account-b".to_string());
second.delete_replication = Some(DeleteReplication {
status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED),
});
let config = ReplicationConfiguration {
role: String::new(),
rules: vec![first, second],
};
let opts = ObjectOpts {
name: "object".to_string(),
target_arn: arn.to_string(),
version_id: Some(Uuid::new_v4()),
delete_marker: true,
op_type: ReplicationType::Delete,
..Default::default()
};
assert!(!config.replicate(&opts));
assert!(delete_replication_target_arns(&config, "object", false).is_empty());
}
#[test]
fn delete_replication_target_arns_rejects_role_with_multiple_destinations() {
let mut first = replication_rule("first", "arn:target:a");
first.delete_replication = Some(DeleteReplication {
status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED),
});
let mut second = replication_rule("second", "arn:target:b");
second.delete_replication = first.delete_replication.clone();
let config = ReplicationConfiguration {
role: "arn:legacy:target".to_string(),
rules: vec![first, second],
};
assert!(delete_replication_target_arns(&config, "object", false).is_empty());
}
#[test]
fn filter_target_arns_falls_back_to_role_when_destination_is_empty() {
let config = ReplicationConfiguration {
@@ -628,97 +829,40 @@ mod tests {
}
#[test]
fn version_purge_uses_delete_replication_for_object_and_marker_versions() {
let arn = "arn:rustfs:replication:us-east-1:target:bucket";
let mut rule = replication_rule("delete", arn);
rule.delete_marker_replication = Some(DeleteMarkerReplication {
status: Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::DISABLED)),
});
rule.delete_replication = Some(DeleteReplication {
status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED),
});
let mut config = ReplicationConfiguration {
role: String::new(),
rules: vec![rule],
};
let version_id = Some(Uuid::new_v4());
for delete_marker in [false, true] {
assert!(config.replicate(&ObjectOpts {
name: "object".to_string(),
version_id,
delete_marker,
op_type: ReplicationType::Delete,
..Default::default()
}));
}
assert!(!config.replicate(&ObjectOpts {
name: "object".to_string(),
delete_marker: true,
op_type: ReplicationType::Delete,
..Default::default()
}));
let rule = &mut config.rules[0];
rule.delete_marker_replication = Some(DeleteMarkerReplication {
status: Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::ENABLED)),
});
rule.delete_replication = Some(DeleteReplication {
status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::DISABLED),
});
for delete_marker in [false, true] {
assert!(!config.replicate(&ObjectOpts {
name: "object".to_string(),
version_id,
delete_marker,
op_type: ReplicationType::Delete,
..Default::default()
}));
}
assert!(config.replicate(&ObjectOpts {
name: "object".to_string(),
delete_marker: true,
op_type: ReplicationType::Delete,
..Default::default()
}));
}
#[test]
fn unsupported_replication_fields_are_reported_before_persistence() {
fn delete_marker_version_purge_requires_delete_replication() {
let arn = "arn:rustfs:replication:us-east-1:target:bucket";
let mut config = ReplicationConfiguration {
role: String::new(),
rules: vec![replication_rule("unsupported", arn)],
rules: vec![delete_marker_rule("delete-markers-only", arn, "", 1, true)],
};
let opts = ObjectOpts {
name: "object.txt".to_string(),
op_type: ReplicationType::Delete,
delete_marker: true,
version_id: Some(Uuid::new_v4()),
..Default::default()
};
config.rules[0].source_selection_criteria = Some(SourceSelectionCriteria {
replica_modifications: None,
sse_kms_encrypted_objects: Some(SseKmsEncryptedObjects {
status: SseKmsEncryptedObjectsStatus::from_static(SseKmsEncryptedObjectsStatus::ENABLED),
}),
});
assert_eq!(
unsupported_replication_config_field(&config),
Some("SourceSelectionCriteria.SseKmsEncryptedObjects")
assert!(
!config.replicate(&opts),
"permanently deleting a delete-marker version must not use the delete-marker replication setting"
);
config.rules[0].source_selection_criteria = None;
config.rules[0].destination.encryption_configuration = Some(EncryptionConfiguration::default());
assert_eq!(unsupported_replication_config_field(&config), Some("Destination.EncryptionConfiguration"));
config.rules[0].destination.encryption_configuration = None;
config.rules[0].destination.metrics = Some(Metrics {
event_threshold: None,
status: MetricsStatus::from_static(MetricsStatus::ENABLED),
config.rules[0].delete_replication = Some(DeleteReplication {
status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::DISABLED),
});
assert_eq!(unsupported_replication_config_field(&config), Some("Destination.Metrics"));
assert!(
!config.replicate(&opts),
"an explicitly disabled permanent-delete setting must not replicate"
);
config.rules[0].destination.metrics = None;
config.rules[0].destination.replication_time = Some(ReplicationTime {
status: ReplicationTimeStatus::from_static(ReplicationTimeStatus::ENABLED),
time: ReplicationTimeValue { minutes: Some(15) },
config.rules[0].delete_replication = Some(DeleteReplication {
status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED),
});
assert_eq!(unsupported_replication_config_field(&config), Some("Destination.ReplicationTime"));
assert!(
config.replicate(&opts),
"permanently deleting a delete-marker version should replicate when delete replication is enabled"
);
}
}
+31 -3
View File
@@ -15,7 +15,7 @@
use std::any::Any;
use crate::storage_api::DeletedObject;
use crate::{MrfOpKind, MrfReplicateEntry, ReplicationType, ReplicationWorkerOperation};
use crate::{MrfOpKind, MrfReplicateEntry, ReplicationStatusType, ReplicationType, ReplicationWorkerOperation};
#[derive(Debug, Clone, Default)]
pub struct DeletedObjectReplicationInfo {
@@ -42,6 +42,11 @@ impl ReplicationWorkerOperation for DeletedObjectReplicationInfo {
op: MrfOpKind::Delete,
delete_marker_version_id: self.delete_object.delete_marker_version_id,
delete_marker: self.delete_object.delete_marker,
replica: self
.delete_object
.replication_state
.as_ref()
.is_some_and(|state| state.replica_status == ReplicationStatusType::Replica),
// Persist the original delete-marker mtime as Unix nanoseconds so replay after a
// restart stamps the replica with the source timestamp rather than the replay time
// (backlog#867). None when unknown; replay then falls back to the current time.
@@ -85,14 +90,22 @@ pub fn is_retryable_delete_replication_head_error(is_not_found: bool, code: Opti
!(is_not_found || matches!(code, Some("MethodNotAllowed" | "405")))
}
pub fn version_purge_target_missing(is_not_found: bool, code: Option<&str>, raw_status: Option<u16>) -> bool {
if matches!(code, Some("MethodNotAllowed" | "405")) || raw_status == Some(405) {
return false;
}
is_not_found || matches!(code, Some("NoSuchVersion" | "NoSuchKey" | "NotFound" | "404")) || raw_status == Some(404)
}
#[cfg(test)]
mod tests {
use super::{
DeletedObjectReplicationInfo, is_retryable_delete_replication_head_error, is_version_delete_replication,
should_retry_delete_marker_purge,
should_retry_delete_marker_purge, version_purge_target_missing,
};
use crate::storage_api::DeletedObject;
use crate::{MrfOpKind, ReplicationType, ReplicationWorkerOperation};
use crate::{MrfOpKind, ReplicationState, ReplicationStatusType, ReplicationType, ReplicationWorkerOperation};
use uuid::Uuid;
#[test]
@@ -109,6 +122,10 @@ mod tests {
delete_marker_version_id: Some(delete_marker_version_id),
delete_marker: true,
delete_marker_mtime: Some(mtime),
replication_state: Some(ReplicationState {
replica_status: ReplicationStatusType::Replica,
..Default::default()
}),
..Default::default()
},
..Default::default()
@@ -122,6 +139,7 @@ mod tests {
assert_eq!(entry.delete_marker_version_id, Some(delete_marker_version_id));
assert_eq!(entry.op, MrfOpKind::Delete);
assert!(entry.delete_marker);
assert!(entry.replica);
// The original mtime must be persisted (as Unix nanos) so replay keeps the source
// timestamp instead of stamping the replica with the replay time (backlog#867).
assert_eq!(
@@ -212,4 +230,14 @@ mod tests {
assert!(!is_retryable_delete_replication_head_error(true, Some("NoSuchKey")));
assert!(is_retryable_delete_replication_head_error(false, Some("AccessDenied")));
}
#[test]
fn version_purge_target_missing_requires_not_found() {
assert!(version_purge_target_missing(true, Some("NoSuchVersion"), Some(404)));
assert!(version_purge_target_missing(false, Some("NoSuchVersion"), None));
assert!(version_purge_target_missing(false, None, Some(404)));
assert!(!version_purge_target_missing(false, None, None));
assert!(!version_purge_target_missing(true, Some("MethodNotAllowed"), Some(405)));
assert!(!version_purge_target_missing(true, Some("405"), None));
}
}
+7
View File
@@ -587,6 +587,11 @@ pub struct MrfReplicateEntry {
#[serde(rename = "deleteMarker", default)]
pub delete_marker: bool,
// For delete entries: whether the operation originated from a replica.
// Old files lack this field and therefore default to a local-source delete.
#[serde(rename = "replica", default)]
pub replica: bool,
// For delete entries: the original delete-marker mtime, persisted as Unix nanoseconds so
// replay stamps replicas with the source timestamp instead of the replay time. Old files
// lack this key; default=None means "unknown", and replay falls back to the current time
@@ -798,6 +803,7 @@ impl ReplicationWorkerOperation for ReplicateObjectInfo {
op: MrfOpKind::Object,
delete_marker_version_id: None,
delete_marker: false,
replica: false,
delete_marker_mtime: None,
}
}
@@ -852,6 +858,7 @@ impl ReplicateObjectInfo {
op: MrfOpKind::Object,
delete_marker_version_id: None,
delete_marker: false,
replica: false,
delete_marker_mtime: None,
}
}
+7 -6
View File
@@ -30,12 +30,12 @@ pub mod tagging;
pub use config::{
ObjectOpts, ReplicationConfigurationExt, ReplicationTargetValidationError, active_replication_rule_destination_arns,
replication_target_arns, should_remove_replication_target, unsupported_replication_config_field,
delete_replication_target_arns, replication_target_arns, should_remove_replication_target,
validate_replication_config_target_arns,
};
pub use delete::{
DeletedObjectReplicationInfo, is_retryable_delete_replication_head_error, is_version_delete_replication,
should_retry_delete_marker_purge,
should_retry_delete_marker_purge, version_purge_target_missing,
};
pub use filemeta::{
REPLICATE_EXISTING, REPLICATE_EXISTING_DELETE, REPLICATE_HEAL, REPLICATE_HEAL_DELETE, REPLICATE_INCOMING,
@@ -55,10 +55,11 @@ pub use object::{
replication_etags_match, target_is_newer_than_source_null_version,
};
pub use operation::{
MustReplicateOptions, ReplicationDeleteScheduleInput, ReplicationDeleteSource, ReplicationDeleteStateSource,
ReplicationResyncTargetObject, delete_replication_missing_source_decision, delete_replication_object_opts,
delete_replication_state_from_config, delete_replication_version_id, heal_uses_delete_replication_path, is_ssec_encrypted,
resync_target_for_object, should_schedule_delete_replication, should_use_existing_delete_replication_info,
MustReplicateOptions, ReplicationDeleteParts, ReplicationDeleteScheduleInput, ReplicationDeleteSource,
ReplicationDeleteStateSource, ReplicationResyncTargetObject, delete_replication_missing_source_decision,
delete_replication_object_opts, delete_replication_parts, delete_replication_state_from_config,
delete_replication_version_id, heal_uses_delete_replication_path, is_ssec_encrypted, resync_target_for_object,
should_schedule_delete_replication, should_use_existing_delete_replication_info,
should_use_existing_delete_replication_source,
};
pub use queue::{
+70 -1
View File
@@ -13,6 +13,7 @@
// limitations under the License.
use byteorder::{ByteOrder, LittleEndian};
use uuid::Uuid;
use crate::{Error, Result};
@@ -21,6 +22,31 @@ pub use crate::filemeta::{MrfOpKind, MrfReplicateEntry};
pub const MRF_META_FORMAT: u16 = 1;
pub const MRF_META_VERSION: u16 = 1;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct MrfDeleteParts {
pub version_id: Option<Uuid>,
pub delete_marker_version_id: Option<Uuid>,
pub delete_marker: bool,
}
impl MrfReplicateEntry {
pub fn delete_parts_for_replay(&self) -> Option<MrfDeleteParts> {
match (self.version_id, self.delete_marker_version_id) {
(Some(version_id), None) => Some(MrfDeleteParts {
version_id: Some(version_id),
delete_marker_version_id: None,
delete_marker: false,
}),
(None, Some(delete_marker_version_id)) => Some(MrfDeleteParts {
version_id: None,
delete_marker_version_id: Some(delete_marker_version_id),
delete_marker: self.delete_marker,
}),
_ => None,
}
}
}
pub fn encode_mrf_file(entries: &[MrfReplicateEntry]) -> Result<Vec<u8>> {
let payload = rmp_serde::to_vec_named(entries).map_err(|e| Error::Other(e.to_string()))?;
let mut data = Vec::with_capacity(4 + payload.len());
@@ -54,7 +80,6 @@ pub fn decode_mrf_file(data: &[u8]) -> Result<Vec<MrfReplicateEntry>> {
#[cfg(test)]
mod tests {
use super::*;
use uuid::Uuid;
#[test]
fn mrf_file_round_trips_object_and_delete_entries() {
@@ -70,6 +95,7 @@ mod tests {
op: MrfOpKind::Object,
delete_marker_version_id: None,
delete_marker: false,
replica: false,
delete_marker_mtime: None,
},
MrfReplicateEntry {
@@ -81,6 +107,7 @@ mod tests {
op: MrfOpKind::Delete,
delete_marker_version_id: Some(del_vid),
delete_marker: true,
replica: true,
delete_marker_mtime: Some(1_705_312_200_123_456_789),
},
];
@@ -95,6 +122,7 @@ mod tests {
assert_eq!(decoded[1].delete_marker_version_id, Some(del_vid));
assert_eq!(decoded[1].op, MrfOpKind::Delete);
assert!(decoded[1].delete_marker);
assert!(decoded[1].replica);
assert_eq!(
decoded[1].delete_marker_mtime,
Some(1_705_312_200_123_456_789),
@@ -102,6 +130,46 @@ mod tests {
);
}
#[test]
fn legacy_version_purge_replay_clears_delete_marker_creation() {
let entry = MrfReplicateEntry {
bucket: "bucket".to_string(),
object: "object".to_string(),
version_id: Some(Uuid::new_v4()),
retry_count: 0,
size: 0,
delete_marker: true,
op: MrfOpKind::Delete,
delete_marker_version_id: None,
replica: false,
delete_marker_mtime: None,
};
let encoded = encode_mrf_file(&[entry]).expect("legacy MRF entry should encode");
let decoded = decode_mrf_file(&encoded).expect("legacy MRF entry should decode");
assert_eq!(decoded[0].delete_parts_for_replay().map(|parts| parts.delete_marker), Some(false));
}
#[test]
fn invalid_delete_id_shapes_fail_closed_on_replay() {
for (version_id, delete_marker_version_id) in [(None, None), (Some(Uuid::new_v4()), Some(Uuid::new_v4()))] {
let entry = MrfReplicateEntry {
bucket: "bucket".to_string(),
object: "object".to_string(),
version_id,
retry_count: 0,
size: 0,
op: MrfOpKind::Delete,
delete_marker_version_id,
delete_marker: false,
replica: false,
delete_marker_mtime: None,
};
assert_eq!(entry.delete_parts_for_replay(), None);
}
}
#[test]
fn mrf_legacy_file_without_op_decodes_as_object() {
let mut payload = Vec::new();
@@ -129,6 +197,7 @@ mod tests {
assert_eq!(decoded[0].retry_count, 2);
assert_eq!(decoded[0].size, 100);
assert_eq!(decoded[0].op, MrfOpKind::Object);
assert!(!decoded[0].replica);
// Old files lack the deleteMarkerMtime key; it must default to None so replay keeps the
// pre-#867 fallback to the current time.
assert_eq!(decoded[0].delete_marker_mtime, None);
+114 -26
View File
@@ -151,6 +151,11 @@ pub fn delete_replication_state_from_config(
let pending_status = decision.pending_status();
let mut state = ReplicationState {
replica_status: if source.replica {
ReplicationStatusType::Replica
} else {
ReplicationStatusType::Empty
},
replicate_decision_str: decision.to_string(),
..Default::default()
};
@@ -174,11 +179,31 @@ pub struct ReplicationDeleteScheduleInput<'a> {
pub deleted_delete_marker_version: bool,
}
fn delete_version_purge_source_status(status: &ReplicationStatusType) -> bool {
status == &ReplicationStatusType::Replica
|| status == &ReplicationStatusType::Pending
|| status == &ReplicationStatusType::Completed
|| status == &ReplicationStatusType::Failed
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ReplicationDeleteParts {
pub delete_marker: bool,
pub version_id: Option<Uuid>,
pub delete_marker_version_id: Option<Uuid>,
}
pub fn delete_replication_parts(
source_delete_marker: bool,
source_version_id: Option<Uuid>,
version_purge: bool,
) -> Option<ReplicationDeleteParts> {
if version_purge {
return source_version_id.map(|version_id| ReplicationDeleteParts {
delete_marker: false,
version_id: Some(version_id),
delete_marker_version_id: None,
});
}
Some(ReplicationDeleteParts {
delete_marker: source_delete_marker,
version_id: None,
delete_marker_version_id: source_version_id,
})
}
pub fn should_schedule_delete_replication(input: ReplicationDeleteScheduleInput<'_>) -> bool {
@@ -186,14 +211,11 @@ pub fn should_schedule_delete_replication(input: ReplicationDeleteScheduleInput<
return false;
}
if input.version_id_requested && !input.deleted_delete_marker_version && !input.source_delete_marker {
return delete_version_purge_source_status(input.source_replication_status);
if input.version_id_requested {
return input.source_version_purge_status == &VersionPurgeStatusType::Pending;
}
input.source_replication_status == &ReplicationStatusType::Replica
|| input.source_replication_status == &ReplicationStatusType::Pending
|| input.source_version_purge_status == &VersionPurgeStatusType::Pending
|| (input.deleted_delete_marker_version && input.source_replication_status == &ReplicationStatusType::Completed)
input.source_replication_status == &ReplicationStatusType::Pending
}
pub fn delete_replication_version_id(
@@ -312,11 +334,12 @@ pub fn resync_target_for_object(
#[cfg(test)]
mod tests {
use super::{
MustReplicateOptions, ReplicationDeleteScheduleInput, ReplicationDeleteSource, ReplicationDeleteStateSource,
ReplicationResyncTargetObject, delete_replication_missing_source_decision, delete_replication_object_opts,
delete_replication_state_from_config, delete_replication_version_id, heal_uses_delete_replication_path,
is_ssec_encrypted, resync_target_for_object, should_schedule_delete_replication,
should_use_existing_delete_replication_info, should_use_existing_delete_replication_source,
MustReplicateOptions, ReplicationDeleteParts, ReplicationDeleteScheduleInput, ReplicationDeleteSource,
ReplicationDeleteStateSource, ReplicationResyncTargetObject, delete_replication_missing_source_decision,
delete_replication_object_opts, delete_replication_parts, delete_replication_state_from_config,
delete_replication_version_id, heal_uses_delete_replication_path, is_ssec_encrypted, resync_target_for_object,
should_schedule_delete_replication, should_use_existing_delete_replication_info,
should_use_existing_delete_replication_source,
};
use crate::http::{AMZ_BUCKET_REPLICATION_STATUS, SSEC_ALGORITHM_HEADER};
use crate::storage_api::ObjectToDelete;
@@ -433,9 +456,7 @@ mod tests {
delete_marker_replication: Some(DeleteMarkerReplication {
status: Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::ENABLED)),
}),
delete_replication: Some(DeleteReplication {
status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED),
}),
delete_replication: None,
destination: Destination {
bucket: arn.to_string(),
..Default::default()
@@ -476,6 +497,7 @@ mod tests {
.expect("replica delete marker should be forwarded to downstream targets");
let pending = format!("{arn}=PENDING;");
assert_eq!(state.replica_status, ReplicationStatusType::Replica);
assert_eq!(state.replication_status_internal.as_deref(), Some(pending.as_str()));
assert_eq!(state.replicate_decision_str, format!("{arn}=true;false;{arn};"));
assert!(state.targets.contains_key(arn));
@@ -502,9 +524,13 @@ mod tests {
#[test]
fn delete_replication_state_tracks_delete_marker_version_purges() {
let arn = "arn:aws:s3:::target-bucket";
let mut rule = delete_replication_rule(arn, false);
rule.delete_replication = Some(DeleteReplication {
status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED),
});
let config = ReplicationConfiguration {
role: arn.to_string(),
rules: vec![delete_replication_rule(arn, false)],
rules: vec![rule],
};
let source = ReplicationDeleteStateSource {
name: "test/object.txt".to_string(),
@@ -518,6 +544,7 @@ mod tests {
.expect("delete-marker version purge should honor delete replication rules");
let pending = format!("{arn}=PENDING;");
assert_eq!(state.replica_status, ReplicationStatusType::Empty);
assert_eq!(state.version_purge_status_internal.as_deref(), Some(pending.as_str()));
assert_eq!(state.replicate_decision_str, format!("{arn}=true;false;{arn};"));
assert!(state.purge_targets.contains_key(arn));
@@ -536,13 +563,13 @@ mod tests {
}
#[test]
fn delete_replication_schedule_keeps_marker_and_version_purges() {
fn delete_replication_schedule_uses_pending_state_from_current_delete_rules() {
assert!(should_schedule_delete_replication(ReplicationDeleteScheduleInput {
replication_request: false,
version_id_requested: true,
source_delete_marker: true,
source_replication_status: &ReplicationStatusType::Completed,
source_version_purge_status: &VersionPurgeStatusType::Empty,
source_replication_status: &ReplicationStatusType::Empty,
source_version_purge_status: &VersionPurgeStatusType::Pending,
deleted_delete_marker_version: true,
}));
assert!(should_schedule_delete_replication(ReplicationDeleteScheduleInput {
@@ -550,19 +577,54 @@ mod tests {
version_id_requested: true,
source_delete_marker: false,
source_replication_status: &ReplicationStatusType::Completed,
source_version_purge_status: &VersionPurgeStatusType::Empty,
source_version_purge_status: &VersionPurgeStatusType::Pending,
deleted_delete_marker_version: false,
}));
assert!(should_schedule_delete_replication(ReplicationDeleteScheduleInput {
replication_request: false,
version_id_requested: false,
source_delete_marker: false,
source_replication_status: &ReplicationStatusType::Empty,
source_version_purge_status: &VersionPurgeStatusType::Pending,
source_replication_status: &ReplicationStatusType::Pending,
source_version_purge_status: &VersionPurgeStatusType::Empty,
deleted_delete_marker_version: false,
}));
}
#[test]
fn delete_replication_schedule_skips_non_pending_states() {
for replication_status in [
ReplicationStatusType::Empty,
ReplicationStatusType::Replica,
ReplicationStatusType::Completed,
ReplicationStatusType::CompletedLegacy,
ReplicationStatusType::Failed,
] {
assert!(!should_schedule_delete_replication(ReplicationDeleteScheduleInput {
replication_request: false,
version_id_requested: false,
source_delete_marker: false,
source_replication_status: &replication_status,
source_version_purge_status: &VersionPurgeStatusType::Empty,
deleted_delete_marker_version: false,
}));
}
for version_purge_status in [
VersionPurgeStatusType::Empty,
VersionPurgeStatusType::Complete,
VersionPurgeStatusType::Failed,
] {
assert!(!should_schedule_delete_replication(ReplicationDeleteScheduleInput {
replication_request: false,
version_id_requested: true,
source_delete_marker: true,
source_replication_status: &ReplicationStatusType::Completed,
source_version_purge_status: &version_purge_status,
deleted_delete_marker_version: true,
}));
}
}
#[test]
fn delete_replication_version_id_splits_marker_creation_and_purge() {
let version_id = Uuid::new_v4();
@@ -571,6 +633,32 @@ mod tests {
assert_eq!(delete_replication_version_id(true, Some(version_id), true), Some(version_id));
}
#[test]
fn delete_replication_parts_fail_closed_without_purge_version() {
assert_eq!(delete_replication_parts(true, None, true), None);
for version_id in [Uuid::nil(), Uuid::new_v4()] {
assert_eq!(
delete_replication_parts(true, Some(version_id), true),
Some(ReplicationDeleteParts {
delete_marker: false,
version_id: Some(version_id),
delete_marker_version_id: None,
})
);
}
let marker_version_id = Uuid::new_v4();
assert_eq!(
delete_replication_parts(true, Some(marker_version_id), false),
Some(ReplicationDeleteParts {
delete_marker: true,
version_id: None,
delete_marker_version_id: Some(marker_version_id),
})
);
}
#[test]
fn delete_replication_source_selection_prefers_existing_marker_source_only_for_replica_requests() {
assert!(should_use_existing_delete_replication_source(true, true, true));
+25 -13
View File
@@ -17,8 +17,8 @@ use std::any::Any;
use crate::storage_api::DeletedObject;
use crate::{
DeletedObjectReplicationInfo, MrfReplicateEntry, REPLICATE_EXISTING, REPLICATE_HEAL, REPLICATE_HEAL_DELETE,
ReplicateObjectInfo, ReplicationStatusType, ReplicationType, ReplicationWorkerOperation, ResyncDecision,
VersionPurgeStatusType,
ReplicateObjectInfo, ReplicationDeleteParts, ReplicationStatusType, ReplicationType, ReplicationWorkerOperation,
ResyncDecision, VersionPurgeStatusType, delete_replication_parts,
};
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
@@ -109,7 +109,11 @@ pub fn replication_heal_queue_action(roi: &mut ReplicateObjectInfo) -> Replicati
}
if roi.delete_marker || !roi.version_purge_status.is_empty() {
let delete_info = heal_deleted_object_replication_info(roi);
let Some(parts) = delete_replication_parts(roi.delete_marker, roi.version_id, !roi.version_purge_status.is_empty())
else {
return ReplicationHealQueueAction::Skip;
};
let delete_info = heal_deleted_object_replication_info(roi, parts);
if is_pending_or_failed_object_heal(roi) || is_pending_or_failed_version_purge(roi) {
return ReplicationHealQueueAction::QueueDelete(delete_info);
@@ -144,21 +148,18 @@ pub fn replication_heal_queue_action(roi: &mut ReplicateObjectInfo) -> Replicati
ReplicationHealQueueAction::Skip
}
fn heal_deleted_object_replication_info(roi: &ReplicateObjectInfo) -> DeletedObjectReplicationInfo {
let (version_id, delete_marker_version_id) = if roi.version_purge_status.is_empty() {
(None, roi.version_id)
} else {
(roi.version_id, None)
};
fn heal_deleted_object_replication_info(
roi: &ReplicateObjectInfo,
parts: ReplicationDeleteParts,
) -> DeletedObjectReplicationInfo {
DeletedObjectReplicationInfo {
delete_object: DeletedObject {
object_name: roi.name.clone(),
delete_marker_version_id,
version_id,
delete_marker_version_id: parts.delete_marker_version_id,
version_id: parts.version_id,
replication_state: roi.replication_state.clone(),
delete_marker_mtime: roi.mod_time,
delete_marker: roi.delete_marker,
delete_marker: parts.delete_marker,
..Default::default()
},
bucket: roi.bucket.clone(),
@@ -399,6 +400,7 @@ mod tests {
let version_id = Uuid::new_v4();
let mut roi = replicate_object_info(ReplicationStatusType::Completed);
roi.version_id = Some(version_id);
roi.delete_marker = true;
roi.version_purge_status = VersionPurgeStatusType::Pending;
let action = replication_heal_queue_action(&mut roi);
@@ -408,6 +410,16 @@ mod tests {
};
assert_eq!(delete_info.delete_object.version_id, Some(version_id));
assert_eq!(delete_info.delete_object.delete_marker_version_id, None);
assert!(!delete_info.delete_object.delete_marker);
}
#[test]
fn heal_queue_action_skips_version_purge_without_version_id() {
let mut roi = replicate_object_info(ReplicationStatusType::Completed);
roi.delete_marker = true;
roi.version_purge_status = VersionPurgeStatusType::Pending;
assert!(matches!(replication_heal_queue_action(&mut roi), ReplicationHealQueueAction::Skip));
}
#[test]
+22 -2
View File
@@ -92,7 +92,6 @@ fn resolve_domain(domain: &str) -> std::io::Result<HashSet<IpAddr>> {
(domain, 0)
.to_socket_addrs()
.map(|v| v.map(|v| v.ip()).collect::<HashSet<_>>())
.map_err(Error::other)
}
#[cfg(test)]
@@ -284,7 +283,7 @@ pub async fn get_host_ip(host: Host<&str>) -> std::io::Result<HashSet<IpAddr>> {
}
Err(err) => {
error!("Failed to resolve domain {domain} using system resolver, err: {err}");
Err(Error::other(err))
Err(err)
}
}
}
@@ -608,6 +607,27 @@ mod test {
assert!(get_host_ip(invalid_host).await.is_err());
}
#[tokio::test]
async fn test_get_host_ip_preserves_resolver_error_provenance() {
let _resolver_guard = set_mock_dns_resolver(|_| Err(IoError::from_raw_os_error(-3)));
let err = get_host_ip(Host::Domain("temporarily-unavailable.example"))
.await
.unwrap_err();
assert_eq!(err.raw_os_error(), Some(-3));
}
#[test]
fn test_resolve_domain_preserves_system_resolver_error_provenance() {
let _resolver_lock = DNS_RESOLVER_TEST_LOCK.lock().unwrap();
reset_dns_resolver_inner();
let err = resolve_domain("rustfs-resolver-provenance.invalid").unwrap_err();
assert_ne!(err.kind(), std::io::ErrorKind::Other, "system resolver error was wrapped: {err}");
}
#[test]
fn test_dns_cache_entry_expiry() {
let ips: HashSet<IpAddr> = [IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1))].into_iter().collect();
+11 -2
View File
@@ -360,7 +360,7 @@ pub fn find_ellipses_patterns(arg: &str) -> Result<ArgPattern> {
Some(caps) => caps,
None => {
return Err(Error::other(format!(
"Invalid ellipsis format in ({arg}), Ellipsis range must be provided in format {{N...M}} where N and M are decimal or hexadecimal positive integers, M must be greater than N, with a maximum expanded range size of {MAX_ELLIPSES_RANGE_SIZE}"
"Invalid ellipsis format. Ellipsis range must be provided in format {{N...M}} where N and M are decimal or hexadecimal positive integers, M must be greater than N, with a maximum expanded range size of {MAX_ELLIPSES_RANGE_SIZE}"
)));
}
};
@@ -399,7 +399,7 @@ pub fn find_ellipses_patterns(arg: &str) -> Result<ArgPattern> {
|| p.suffix.contains(CLOSE_BRACES)
{
return Err(Error::other(format!(
"Invalid ellipsis format in ({arg}), Ellipsis range must be provided in format {{N...M}} where N and M are decimal or hexadecimal positive integers, M must be greater than N, with a maximum expanded range size of {MAX_ELLIPSES_RANGE_SIZE}"
"Invalid ellipsis format. Ellipsis range must be provided in format {{N...M}} where N and M are decimal or hexadecimal positive integers, M must be greater than N, with a maximum expanded range size of {MAX_ELLIPSES_RANGE_SIZE}"
)));
}
}
@@ -938,6 +938,15 @@ mod tests {
assert!(err.to_string().contains("decimal or hexadecimal"), "unexpected error message: {err}");
}
#[test]
fn test_find_ellipses_patterns_leftover_brace_error_does_not_echo_input() {
let err = find_ellipses_patterns("http://:brace-secret@server/{1...2}}").unwrap_err();
assert!(
!err.to_string().contains("brace-secret"),
"ellipsis error leaked endpoint credentials: {err}"
);
}
#[test]
fn test_parse_ellipses_range_rejects_oversized_ranges() {
let err = parse_ellipses_range("{0...10000}").unwrap_err();
@@ -12,6 +12,10 @@ for later deletion.
## Open Items
- `rustfs-5416` Helm distributed startup wait setting: charts that predate explicit local endpoint identity expose startupWaitTimeoutSeconds for their peer DNS/TCP init gate. The new chart keeps the value accepted but ignores it after moving startup convergence into RustFS. Remove the value and its documentation after the minimum supported direct-upgrade chart includes localEndpointHost.autoInject and no longer renders the peer gate.
- `rustfs-5416-wait-mode` startup wait-mode validation: releases before explicit local endpoint identity treat an unknown RUSTFS_STARTUP_TOPOLOGY_WAIT_MODE value as auto. New servers retain that fallback only when no explicit local endpoint host is configured; an anchor requires a recognized mode so a typo cannot bypass DNS locality. Remove the fallback and reject every unknown value after every supported direct-upgrade chart validates this setting before rollout.
- `rustfs-5416-kubernetes-alias-dns` Kubernetes endpoint identity fallback: deployments created before explicit local endpoint identity may use resolvable aliases that do not match the Pod hostname. An implicit auto-mode zero match retains legacy DNS locality with a bounded deadline, while ambiguous matches and invalid explicit anchors still fail closed. Remove the fallback after every supported direct-upgrade chart and deployment manifest provides a canonical RUSTFS_LOCAL_ENDPOINT_HOST for domain-based distributed topologies.
- `rustfs-5416-zero-retry-delay` startup retry-delay validation: releases before bounded topology convergence accept RUSTFS_STARTUP_TOPOLOGY_RETRY_MAX_DELAY values of 0 or 0ms. New servers replace those values with the safe nonzero default so a direct upgrade neither fails startup nor enters a busy loop. Reject zero after the minimum supported direct-upgrade release validates or rewrites this setting before rollout.
- `scanner-usage-v2` persisted scanner usage migration: pre-v2 scanners write `.usage.json`, so upgraded clusters read that primary/backup pair only while `.usage.v2.json` is absent and continue removing deleted buckets from legacy copies that still exist. The additive usage_snapshot_complete field in `.usage.v2.json` must remain optional while mixed-version clusters are supported; a missing field means the snapshot is not authoritative. Remove the legacy object fallback and cleanup only after every supported direct-upgrade source writes `.usage.v2.json`.
- `ns-scanner-rpc-v3` namespace scanner capability and activity handshake: old peers and legacy internode transports lack the authenticated startup-epoch handshake. The oldest peers send an empty activity request and receive a field-empty protocol-0 response. Protocol v4 binds the challenge and response topology but cannot authenticate distributed dirty-usage state. Current protocol v5 binds the request version, acknowledgement target and generation, and the response dirty-usage state. Servers retain protocol-0 and protocol-v4 codecs for rolling upgrades, while the distributed scanner publishes usage only after every peer returns authenticated protocol v5 state. Scanner selection treats HTTP 404/405/426 and the legacy MethodNotAllowed default as an explicit lack of remote scanner v3 support and assigns those disks to coordinator-driven workers; transient capability failures remain incomplete and do not activate the fallback. Remove the coordinator fallback after the minimum supported RustFS peer version implements namespace scanner protocol v3, remove protocol-0 activity requests and responses after every supported peer implements authenticated scanner activity protocol v4, and remove the protocol-v4 activity codec after every supported peer implements protocol v5; future protocol revisions must keep the same dual-version server/codec window before changing the advertised version.
- `#4648` walk-dir stream completion capability: old clients can append fallback output to an already-used metacache writer after a terminal body error, so servers emit terminal walk errors only to clients that sign the `walk_dir_stream_completion=error-v1` query capability and its request-body digest. Remove the legacy clean-EOF path after the minimum supported RustFS peer version always advertises this capability.
+1 -1
View File
@@ -59,7 +59,7 @@
{
default = rustPlatform.buildRustPackage {
pname = "rustfs";
version = "1.0.0-beta.11";
version = "1.0.0-beta.12";
src = ./.;
+68 -14
View File
@@ -47,6 +47,60 @@ without manual intervention:
If you previously set `replicaCount=16` and now want a different topology,
set both `replicaCount` and `drivesPerNode` explicitly.
For distributed deployments that use the chart-generated
`RUSTFS_VOLUMES`, `localEndpointHost.autoInject` defaults to automatic
selection. Without `secret.existingSecret`, the chart injects a private
Downward API variable, `RUSTFS_CHART_POD_NAME`, and uses it to build the pod's
fully qualified hostname in
`RUSTFS_LOCAL_ENDPOINT_HOST`. RustFS can then identify local drives without
waiting for every peer's DNS record or TCP listener. A user-defined `POD_NAME`
is preserved and does not interfere with the private chart variable. Setting
`RUSTFS_STARTUP_TOPOLOGY_WAIT_MODE` explicitly to `bounded`, `fail-fast`,
`failfast`, or `strict` also keeps legacy DNS-based locality discovery. A
dynamically sourced wait mode keeps the legacy path because its value cannot be
validated while rendering. Otherwise, Kubernetes auto-detection selects
orchestrated startup when RustFS consumes the generated anchor.
An existing Secret is opaque to the chart and may historically contain more
than credentials, so the chart does not inject an anchor whenever
`secret.existingSecret` is set. In Kubernetes auto/orchestrated mode, RustFS
then derives a DNS-free identity from the kernel hostname when exactly one
domain endpoint at the server port has the same full hostname or first label.
All-IP topologies retain direct IP locality detection. A domain topology with
zero matches retains legacy DNS locality discovery; implicit auto mode bounds
that compatibility path by `RUSTFS_STARTUP_TOPOLOGY_WAIT_TIMEOUT` (180 seconds
by default). Multiple matching candidates remain an error. Set
`RUSTFS_LOCAL_ENDPOINT_HOST` explicitly to avoid DNS discovery. For a
credentials-only Secret, set
`localEndpointHost.autoInject=true` to add the chart anchor without changing
the historical ConfigMap-then-Secret `envFrom` precedence. If injection is
explicitly enabled with an incompatible hidden `RUSTFS_VOLUMES`,
`RUSTFS_ADDRESS`, or `RUSTFS_STARTUP_TOPOLOGY_WAIT_MODE`, RustFS also fails
during endpoint construction; it does not silently fall back to a different
topology.
When `config.rustfs.volumes` is set explicitly, the chart does not infer a
local endpoint identity. RustFS applies the same kernel-hostname inference to
custom domain topologies in Kubernetes auto/orchestrated mode; aliases that do
not match the Pod hostname retain legacy DNS locality, with the bounded auto
fallback described above. They may provide `RUSTFS_LOCAL_ENDPOINT_HOST`
through `extraEnv` for DNS-free startup. An explicit `RUSTFS_VOLUMES`, explicit
`RUSTFS_LOCAL_ENDPOINT_HOST`, bounded/dynamic or unrecognized startup mode, or
`localEndpointHost.autoInject=false`, also disables chart injection. A
`RUSTFS_ADDRESS` override alone does not disable it; the effective address and
generated topology must agree on the endpoint port. Custom anchor-based
configurations must resolve to `orchestrated` startup mode and must not receive
a conflicting mode from an `envFrom` source.
`startupWaitTimeoutSeconds` is retained for values-file compatibility but is
deprecated and ignored.
Historical `RUSTFS_STARTUP_TOPOLOGY_RETRY_MAX_DELAY` values of `0` or `0ms`
are replaced with the safe default retry cap instead of causing a busy loop or
blocking a direct upgrade.
Upgrade the chart and RustFS image together. An older image that does not
recognize `RUSTFS_LOCAL_ENDPOINT_HOST` retains its previous DNS-based startup
behavior.
---
# Parameters Overview
@@ -57,6 +111,7 @@ set both `replicaCount` and `drivesPerNode` explicitly.
| affinity.podAntiAffinity.enabled | bool | `true` | |
| affinity.podAntiAffinity.topologyKey | string | `"kubernetes.io/hostname"` | |
| clusterDomain | string | `"cluster.local"` | Kubernetes cluster DNS domain used to build in-cluster FQDNs for `RUSTFS_VOLUMES` (distributed mode) and mTLS server certificate SANs. Override for clusters not using the default `cluster.local`. Provide the DNS root only, without a `svc.` prefix or leading/trailing dots. |
| localEndpointHost.autoInject | bool or null | `null` | Automatically inject `RUSTFS_LOCAL_ENDPOINT_HOST` for chart-generated distributed topologies unless `secret.existingSecret` is set. Use `true` for a credentials-only existing Secret or `false` to preserve legacy DNS locality explicitly. |
| commonLabels | object | `{}` | Labels to add to all deployed objects. |
| config.rustfs.address | string | `":9000"` | |
| config.rustfs.console_address | string | `":9001"` | |
@@ -66,7 +121,7 @@ set both `replicaCount` and `drivesPerNode` explicitly.
| config.rustfs.obs_environment | string | `"development"` | |
| config.rustfs.obs_log_directory | string | `"/logs"` | Log directory inside the RustFS container. Set to `""` to disable log PVCs and mounts. |
| config.rustfs.region | string | `"us-east-1"` | |
| config.rustfs.volumes | string | `""` | |
| config.rustfs.volumes | string | `""` | Explicit distributed volume topology. When empty, the chart generates the topology and normally injects `RUSTFS_LOCAL_ENDPOINT_HOST`; custom topologies must configure local endpoint identity explicitly when needed. |
| config.rustfs.log_rotation.size | int | `"100"` | Default log rotation size mb for rustfs. |
| config.rustfs.log_rotation.time | string | `"hour"` | Default log rotation time for rustfs. |
| config.rustfs.log_rotation.keep_files | int | `"30"` | Default log keep files for rustfs. |
@@ -107,7 +162,7 @@ set both `replicaCount` and `drivesPerNode` explicitly.
| config.rustfs.kms.vault.vault_token | string | `""`| The vault token. Rendered into a dedicated Secret (`<fullname>-kms-secret`), never into the ConfigMap. |
| config.rustfs.kms.vault.vault_mount_path | string | `"transit"`| The vault mount path, only works if `vault_backend` equals `vault-transit` . |
| config.rustfs.kms.vault.default_key | string | `"transit"`| The master key id for RustFS. |
| extraEnv | map | `[]` | Extra environment variables for RustFS container. |
| extraEnv | list | `[]` | Extra environment variables for the RustFS container. An explicit `RUSTFS_LOCAL_ENDPOINT_HOST` or `RUSTFS_VOLUMES`, or a bounded, dynamic, or unrecognized startup mode, disables generated anchor injection. `POD_NAME` and `RUSTFS_ADDRESS` remain independent overrides. |
| extraVolumes | list | `[]` | Extra volumes to add to the pod spec. Supported in both standalone (Deployment) and distributed (StatefulSet) modes. |
| extraVolumeMounts | list | `[]` | Extra volume mounts to add to the RustFS container. Supported in both standalone (Deployment) and distributed (StatefulSet) modes. |
| containerSecurityContext.capabilities.drop[0] | string | `"ALL"` | |
@@ -190,7 +245,7 @@ uer. `ClusterIssuer` or `Issuer`. |
| resources.limits.memory | string | `"512Mi"` | |
| resources.requests.cpu | string | `"100m"` | |
| resources.requests.memory | string | `"128Mi"` | |
| secret.existingSecret | string | `""` | Use existing secret with a credentials. |
| secret.existingSecret | string | `""` | Use an existing Secret. Automatic endpoint-anchor injection is disabled because the Secret is opaque; set `localEndpointHost.autoInject=true` only after confirming it contains credentials rather than runtime topology, address, or startup-mode overrides. |
| secret.rustfs.access_key | string | `"rustfsadmin"` | RustFS Access Key ID |
| secret.rustfs.secret_key | string | `"rustfsadmin"` | RustFS Secret Key ID |
| service.type | string | `"ClusterIP"` | |
@@ -202,6 +257,7 @@ uer. `ClusterIssuer` or `Issuer`. |
| serviceAccount.automount | bool | `true` | |
| serviceAccount.create | bool | `true` | |
| serviceAccount.name | string | `""` | |
| startupWaitTimeoutSeconds | int | `300` | Deprecated and ignored; retained for values-file compatibility. |
| storageclass.dataStorageSize | string | `"256Mi"` | The storage size for data PVC. |
| storageclass.logStorageSize | string | `"256Mi"` | The storage size for logs PVC. |
| storageclass.name | string | `"local-path"` | The name for StorageClass. |
@@ -291,20 +347,17 @@ Notes:
* **Pools are append-only.** The list index determines the StatefulSet name —
never remove or reorder entries. Retire a pool with
`rc admin decommission` before removing it from the list.
* During the expansion rollout, pods restart until every pod of every pool is
resolvable — the server refuses to start with unresolvable peers, so expect
a few crash/restart cycles before the cluster converges. This is harmless.
* With chart-generated volumes, each pod receives an explicit local endpoint
identity. An unavailable peer no longer blocks a pod from reaching RustFS's
own startup and quorum checks.
* After the cluster converges, run `rc admin rebalance start <alias>` to
spread existing objects across the new pool.
* Pod anti-affinity in pool mode is scoped per pool and **preferred**
(soft), not required: two pools can share nodes, and each pool's own pods
spread across distinct nodes when capacity allows. Soft affinity is
load-bearing for in-place expansion — with required rules, the
not-yet-rolled pods of the existing pool block the new pool's pods from
their nodes while the rolled pods crash on the unresolvable (Pending)
peers, deadlocking the rollout on any cluster with fewer nodes than the
total pod count. Single-pool deployments (`pools.enabled=false`) keep the
chart's existing required anti-affinity unchanged.
spread across distinct nodes when capacity allows. Preferred affinity keeps
additional pools schedulable when the cluster has fewer nodes than total
pods. Single-pool deployments (`pools.enabled=false`) keep the chart's
existing required anti-affinity unchanged.
* The PodDisruptionBudget spans all pools: with the default
`pdb.maxUnavailable: 1`, at most one pod of the whole cluster may be
evicted at a time. This is deliberately conservative — quorum safety
@@ -315,7 +368,8 @@ Notes:
## Requirement
* Helm V3
* RustFS >= 1.0.0-alpha.69
* The RustFS image from the same release as the chart. If `image.rustfs.tag`
is overridden, that image must support `RUSTFS_LOCAL_ENDPOINT_HOST`.
Due to the traefik and ingress has different session sticky/affinity annotations, and rustfs support both those two controller, you should specify parameter `ingress.className` to select the right one which suits for you.
+2 -2
View File
@@ -2,8 +2,8 @@ apiVersion: v2
name: rustfs
description: RustFS helm chart to deploy RustFS on kubernetes cluster.
type: application
version: "0.11.0"
appVersion: "1.0.0-beta.11"
version: "0.12.0"
appVersion: "1.0.0-beta.12"
home: https://rustfs.com
icon: https://media.sys.truenas.net/apps/rustfs/icons/icon.svg
maintainers:
+10 -8
View File
@@ -322,21 +322,23 @@ Render RUSTFS_SERVER_DOMAINS
{{- join "," $domains -}}
{{- end -}}
{{/* Render probe command for liveness and readiness
{{/* Render an mTLS probe command
*/}}
{{- define "rustfs.probeCommand" -}}
{{- $endpoint_port := .Values.service.endpoint.port | default 9000 -}}
{{- $console_port := .Values.service.console.port | default 9001 -}}
{{- $root := .root -}}
{{- $endpointPath := .endpointPath -}}
{{- $endpoint_port := $root.Values.service.endpoint.port | default 9000 -}}
{{- $console_port := $root.Values.service.console.port | default 9001 -}}
{{- $args := "-skf" -}}
{{- if and .Values.mtls.enabled -}}
{{- $args = printf "%s --cert %s --key %s" $args .Values.mtls.clientCertPath .Values.mtls.clientKeyPath -}}
{{- if and $root.Values.mtls.enabled -}}
{{- $args = printf "%s --cert %s --key %s" $args $root.Values.mtls.clientCertPath $root.Values.mtls.clientKeyPath -}}
{{- end -}}
- /bin/sh
- -c
- |
curl {{ $args }} https://127.0.0.1:{{ $endpoint_port }}/health/ready && \
curl {{ $args }} https://127.0.0.1:{{ $endpoint_port }}{{ $endpointPath }} && \
curl {{ $args }} https://127.0.0.1:{{ $console_port }}/rustfs/console/health
{{- end -}}
@@ -350,7 +352,7 @@ livenessProbe:
{{- if .Values.mtls.enabled }}
exec:
command:
{{ include "rustfs.probeCommand" . | nindent 6 }}
{{ include "rustfs.probeCommand" (dict "root" . "endpointPath" "/health") | nindent 6 }}
{{- else }}
httpGet:
path: /health
@@ -369,7 +371,7 @@ readinessProbe:
{{- if .Values.mtls.enabled }}
exec:
command:
{{ include "rustfs.probeCommand" . | nindent 6 }}
{{ include "rustfs.probeCommand" (dict "root" . "endpointPath" "/health/ready") | nindent 6 }}
{{- else }}
httpGet:
path: /health/ready
+43 -53
View File
@@ -1,12 +1,37 @@
{{- $logDir := .Values.config.rustfs.obs_log_directory }}
{{- $logDirEnabled := ne $logDir "" }}
{{- $poolsEnabled := and .Values.pools .Values.pools.enabled }}
{{- $generatedVolumes := empty .Values.config.rustfs.volumes }}
{{- $localEndpointHostAutoInject := empty .Values.secret.existingSecret }}
{{- with .Values.localEndpointHost }}
{{- if and (hasKey . "autoInject") (ne .autoInject nil) }}
{{- if not (kindIs "bool" .autoInject) }}
{{- fail "localEndpointHost.autoInject must be a boolean or null" }}
{{- end }}
{{- $localEndpointHostAutoInject = .autoInject }}
{{- end }}
{{- end }}
{{- $injectLocalEndpointHost := and $generatedVolumes $localEndpointHostAutoInject }}
{{- range $env := .Values.extraEnv -}}
{{- $name := default "" $env.name -}}
{{- if eq $name "RUSTFS_VOLUMES" -}}
{{- $injectLocalEndpointHost = false -}}
{{- else if eq $name "RUSTFS_LOCAL_ENDPOINT_HOST" -}}
{{- $injectLocalEndpointHost = false -}}
{{- else if eq $name "RUSTFS_STARTUP_TOPOLOGY_WAIT_MODE" -}}
{{- if hasKey $env "value" -}}
{{- $mode := lower (trim (toString $env.value)) -}}
{{- if and (ne $mode "") (ne $mode "auto") (ne $mode "orchestrated") -}}
{{- $injectLocalEndpointHost = false -}}
{{- end -}}
{{- else -}}
{{- $injectLocalEndpointHost = false -}}
{{- end -}}
{{- end -}}
{{- end -}}
{{- if and .Values.mode.distributed.enabled (le (int .Values.replicaCount) 1) -}}
{{- fail "Distributed mode requires replicaCount >= 2" -}}
{{- end -}}
{{- if and .Values.mode.distributed.enabled (le (int .Values.startupWaitTimeoutSeconds) 0) -}}
{{- fail "startupWaitTimeoutSeconds must be greater than 0 in distributed mode" -}}
{{- end -}}
{{- if .Values.mode.distributed.enabled }}
{{- $pools := include "rustfs.pools" . | fromJsonArray }}
@@ -74,13 +99,10 @@ spec:
{{- if $.Values.affinity.podAntiAffinity.enabled }}
podAntiAffinity:
{{- if $poolsEnabled }}
{{- /* Pool-scoped and PREFERRED, not required: during in-place
expansion the not-yet-rolled pods' required rules block new
pods from their nodes, while the new pods' DNS must resolve
before the roll can proceed - required rules deadlock
expansion whenever the cluster has fewer nodes than total
pods. Soft anti-affinity converges (the scheduler still
spreads when capacity allows). */}}
{{- /* Pool-scoped and PREFERRED, not required: additional pools
must remain schedulable when the cluster has fewer nodes
than total pods. The scheduler still spreads them when
capacity allows. */}}
preferredDuringSchedulingIgnoredDuringExecution:
- weight: 100
podAffinityTerm:
@@ -127,10 +149,6 @@ spec:
env:
- name: DRIVES_PER_NODE
value: {{ $drivesPerNode | quote }}
- name: ENDPOINT_PORT
value: {{ $.Values.service.endpoint.port | quote }}
- name: STARTUP_WAIT_TIMEOUT_SECONDS
value: {{ $.Values.startupWaitTimeoutSeconds | quote }}
command:
- sh
- -c
@@ -146,44 +164,6 @@ spec:
mkdir -p /mnt/rustfs/logs
chmod 755 /mnt/rustfs/logs
{{- end }}
deadline=$(($(date +%s) + STARTUP_WAIT_TIMEOUT_SECONDS))
wait_for_peer() {
description=$1
shift
until "$@" >/dev/null 2>&1; do
if [ "$(date +%s)" -ge "$deadline" ]; then
echo "Timed out after ${STARTUP_WAIT_TIMEOUT_SECONDS}s waiting for ${description}" >&2
exit 1
fi
sleep 2
done
}
echo "Waiting for distributed peer DNS records"
{{- range $peerPool := $pools }}
{{- range $i := until (int $peerPool.replicaCount) }}
wait_for_peer "DNS for {{ $peerPool.fullname }}-{{ $i }}" \
nslookup "{{ $peerPool.fullname }}-{{ $i }}.{{ include "rustfs.fullname" $ }}-headless.{{ $.Release.Namespace }}.svc.{{ include "rustfs.clusterDomain" $ }}"
{{- end }}
{{- end }}
ordinal=${HOSTNAME##*-}
echo "Waiting for earlier distributed peers to listen on port ${ENDPOINT_PORT}"
{{- range $peerPool := $pools }}
{{- if lt (int $peerPool.index) (int $pool.index) }}
{{- range $i := until (int $peerPool.replicaCount) }}
wait_for_peer "{{ $peerPool.fullname }}-{{ $i }}:${ENDPOINT_PORT}" \
nc -z -w 2 "{{ $peerPool.fullname }}-{{ $i }}.{{ include "rustfs.fullname" $ }}-headless.{{ $.Release.Namespace }}.svc.{{ include "rustfs.clusterDomain" $ }}" "$ENDPOINT_PORT"
{{- end }}
{{- else if eq (int $peerPool.index) (int $pool.index) }}
i=0
while [ "$i" -lt "$ordinal" ]; do
wait_for_peer "{{ $peerPool.fullname }}-${i}:${ENDPOINT_PORT}" \
nc -z -w 2 "{{ $peerPool.fullname }}-${i}.{{ include "rustfs.fullname" $ }}-headless.{{ $.Release.Namespace }}.svc.{{ include "rustfs.clusterDomain" $ }}" "$ENDPOINT_PORT"
i=$((i + 1))
done
{{- end }}
{{- end }}
volumeMounts:
{{- if gt $drivesPerNode 1 }}
{{- range $i := until $drivesPerNode }}
@@ -210,9 +190,19 @@ spec:
containerPort: {{ $.Values.service.endpoint.port }}
- name: console
containerPort: {{ $.Values.service.console.port }}
{{- with $.Values.extraEnv }}
{{- if or $injectLocalEndpointHost (not (empty $.Values.extraEnv)) }}
env:
{{- if $injectLocalEndpointHost }}
- name: RUSTFS_CHART_POD_NAME
valueFrom:
fieldRef:
fieldPath: metadata.name
- name: RUSTFS_LOCAL_ENDPOINT_HOST
value: {{ printf "$(RUSTFS_CHART_POD_NAME).%s-headless.%s.svc.%s" (include "rustfs.fullname" $) $.Release.Namespace (include "rustfs.clusterDomain" $) | quote }}
{{- end }}
{{- with $.Values.extraEnv }}
{{- toYaml . | nindent 12 }}
{{- end }}
{{- end }}
envFrom:
- configMapRef:
+26 -4
View File
@@ -30,11 +30,22 @@ drivesPerNode: null
# Provide the DNS root only, without a `svc.` prefix or leading/trailing dots.
clusterDomain: cluster.local
# Maximum time an init container waits for distributed peers to publish DNS
# and for earlier peers to start listening. The wait is shared across all
# peers, not applied once per peer.
# Deprecated and ignored. Retained so existing values files continue to
# render after peer startup coordination moved into RustFS.
# RUSTFS_COMPAT_TODO(rustfs-5416): Keep the removed init-gate setting visible to existing values files. Remove after the minimum supported direct-upgrade chart includes localEndpointHost.autoInject and no longer renders peer gates.
startupWaitTimeoutSeconds: 300
# Inject a pod-specific endpoint identity for chart-generated distributed
# topologies. The default enables injection unless secret.existingSecret is
# set, because the chart cannot inspect runtime overrides hidden in that
# Secret. Without injection, the server infers a unique matching domain
# endpoint from its kernel hostname in Kubernetes auto/orchestrated mode; custom
# aliases retain bounded legacy DNS locality when no hostname matches. Configure
# RUSTFS_LOCAL_ENDPOINT_HOST explicitly for DNS-free startup. Set true for a
# credentials-only existing Secret or false to suppress chart injection.
localEndpointHost:
autoInject: null
# This sets the container image more information can be found here: https://kubernetes.io/docs/concepts/containers/images/
image:
rustfs: # This sets the rustfs image repository and tag.
@@ -223,7 +234,18 @@ config:
default_key: ""
extraEnv: [] # This is for setting extra environment variables in the rustfs container. It should be a list of key value pairs. For example:
# Extra environment variables for the RustFS container.
# In distributed mode with chart-generated volumes (config.rustfs.volumes is
# empty), the chart injects a private Downward API pod name and
# RUSTFS_LOCAL_ENDPOINT_HOST to identify the pod's local endpoints. An explicit
# RUSTFS_LOCAL_ENDPOINT_HOST or RUSTFS_VOLUMES entry disables automatic anchor
# injection. An explicit bounded or fail-fast
# RUSTFS_STARTUP_TOPOLOGY_WAIT_MODE, or an unrecognized static value, also keeps
# legacy DNS-based locality discovery. Otherwise Kubernetes auto-detection
# selects orchestrated startup when RustFS consumes the generated anchor.
# With custom volumes, the chart does not inject them and extraEnv may define
# the local endpoint identity explicitly.
extraEnv: []
# extraEnv:
# - name: RUSTFS_ERASURE_SET_DRIVE_COUNT
# value: "16"
+5 -2
View File
@@ -1,9 +1,9 @@
%global _enable_debug_packages 0
%global _empty_manifest_terminate_build 0
%global prerelease beta.11
%global prerelease beta.12
Name: rustfs
Version: 1.0.0
Release: beta.11
Release: beta.12
Summary: High-performance distributed object storage for MinIO alternative
License: Apache-2.0
@@ -58,6 +58,9 @@ install %_builddir/%{name}-%{version}-%{prerelease}/target/%_arch/%_arch-unknown
%_bindir/rustfs
%changelog
* Thu Jul 30 2026 overtrue <anzhengchao@gmail.com>
- Update RPM package to RustFS 1.0.0-beta.12
* Thu Jul 23 2026 overtrue <anzhengchao@gmail.com>
- Update RPM package to RustFS 1.0.0-beta.11
+1 -1
View File
@@ -148,7 +148,7 @@ impl Operation for ListCannedPolicies {
return Err(s3_error!(InternalError, "iam is not initialized"));
};
let policies = iam_store.list_polices(&query.bucket).await.map_err(|e| {
let policies = iam_store.list_policies(&query.bucket).await.map_err(|e| {
warn!(
component = LOG_COMPONENT_ADMIN,
subsystem = LOG_SUBSYSTEM_POLICY,
+2 -35
View File
@@ -179,20 +179,6 @@ impl RemoteTargetRequest {
return Err(s3_error!(InvalidRequest, "credentials.secretKey is required"));
}
for (unsupported, configured) in [
("disableProxy", self.disable_proxy),
("healthCheckDuration", self.health_check_duration != 0),
("edge", self.edge),
("edgeSyncBeforeExpiry", self.edge_sync_before_expiry),
] {
if configured {
return Err(s3_error!(
InvalidRequest,
"remote target field {unsupported} is not supported by this RustFS version"
));
}
}
Ok(BucketTarget {
source_bucket: self.source_bucket,
endpoint: self.endpoint,
@@ -1058,6 +1044,7 @@ mod tests {
op: MrfOpKind::Object,
delete_marker_version_id: None,
delete_marker: false,
replica: false,
delete_marker_mtime: None,
},
MrfReplicateEntry {
@@ -1069,6 +1056,7 @@ mod tests {
op: MrfOpKind::Object,
delete_marker_version_id: None,
delete_marker: false,
replica: false,
delete_marker_mtime: None,
},
],
@@ -1218,27 +1206,6 @@ mod tests {
assert!(err.to_string().contains("credentials.secretKey is required"));
}
#[test]
fn remote_target_request_rejects_unimplemented_fields() {
for (field, value) in [
("disableProxy", serde_json::json!(true)),
("healthCheckDuration", serde_json::json!(5)),
("edge", serde_json::json!(true)),
("edgeSyncBeforeExpiry", serde_json::json!(true)),
] {
let mut request = valid_remote_target_request();
request[field] = value;
let request: RemoteTargetRequest =
serde_json::from_value(request).expect("unsupported field should still deserialize");
let err = request
.into_bucket_target()
.expect_err("unimplemented remote target fields must not be persisted");
assert!(err.to_string().contains(field));
assert!(err.to_string().contains("not supported by this RustFS version"));
}
}
#[test]
fn remote_target_request_converts_to_bucket_target() {
let target = serde_json::from_value::<RemoteTargetRequest>(valid_remote_target_request())
+1 -1
View File
@@ -724,7 +724,7 @@ impl Operation for ExportIam {
match file {
ALL_POLICIES_FILE => {
let policies: HashMap<String, rustfs_policy::policy::Policy> = iam_store
.list_polices("")
.list_policies("")
.await
.map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, e.to_string()))?;
let json_str = serde_json::to_vec(&policies)
+1 -32
View File
@@ -35,7 +35,7 @@ use super::storage_api::bucket_usecase::bucket::{
policy_sys::PolicySys,
replication::{
ReplicationTargetValidationError, replication_target_arns, should_remove_replication_target,
unsupported_replication_config_field, validate_replication_config_target_arns,
validate_replication_config_target_arns,
},
target::{BucketTargetType, BucketTargets},
utils::serialize,
@@ -556,16 +556,6 @@ fn validate_replication_config_targets(targets: &BucketTargets, config: &Replica
}
}
fn validate_replication_config_capabilities(config: &ReplicationConfiguration) -> S3Result<()> {
if let Some(field) = unsupported_replication_config_field(config) {
return Err(S3Error::with_message(
S3ErrorCode::InvalidRequest,
format!("replication field {field} is not supported by this RustFS version"),
));
}
Ok(())
}
async fn validate_bucket_replication_update(bucket: &str, config: &ReplicationConfiguration) -> S3Result<()> {
if !BucketVersioningSys::enabled(bucket).await {
return Err(s3_error!(
@@ -574,8 +564,6 @@ async fn validate_bucket_replication_update(bucket: &str, config: &ReplicationCo
));
}
validate_replication_config_capabilities(config)?;
let targets = metadata_sys::get_bucket_targets_config(bucket)
.await
.map_err(|err| match err {
@@ -3087,25 +3075,6 @@ mod tests {
validate_replication_config_targets(&targets, &config).expect("disabled rules should not require live targets");
}
#[test]
fn validate_replication_config_capabilities_names_unsupported_field() {
let mut rule = replication_rule_for_target("arn:rustfs:replication:us-east-1:target:bucket");
rule.destination.encryption_configuration = Some(s3s::dto::EncryptionConfiguration::default());
let config = ReplicationConfiguration {
role: String::new(),
rules: vec![rule],
};
let err = validate_replication_config_capabilities(&config)
.expect_err("destination encryption must be rejected until the execution path supports it");
assert_eq!(err.code(), &S3ErrorCode::InvalidRequest);
assert!(
err.to_string()
.contains("Destination.EncryptionConfiguration is not supported")
);
}
#[test]
fn remove_replication_targets_from_config_targets_only_removes_referenced_replication_targets() {
let removed_arn = "arn:rustfs:replication:us-east-1:removed:bucket";
+135 -65
View File
@@ -44,8 +44,8 @@ use super::storage_api::object_usecase::bucket::{
REPLICATE_INCOMING_DELETE, ReplicationStatusType, VersionPurgeStatusType, check_replicate_delete,
delete_replication_state_from_config, delete_replication_version_id, deleted_object_has_pending_replication_delete,
must_replicate_object, schedule_object_replication, schedule_replication_delete, set_deleted_object_replication_state,
set_object_to_delete_version_purge_status, should_use_existing_delete_replication_info,
should_use_existing_delete_replication_source,
set_object_to_delete_version_purge_status, should_schedule_delete_replication,
should_use_existing_delete_replication_info, should_use_existing_delete_replication_source,
},
tagging::decode_tags,
validate_restore_request,
@@ -2637,18 +2637,24 @@ where
fn delete_replication_state_source<'a>(
opts: &ObjectOptions,
existing_object_info: Option<&'a ObjectInfo>,
deleted_object_info: &'a ObjectInfo,
deleted_object_source: &'a ObjectInfo,
delete_result: &'a ObjectInfo,
) -> &'a ObjectInfo {
let replication_source = if opts.replication_request {
deleted_object_source
} else {
delete_result
};
if should_use_existing_delete_replication_source(
opts.replication_request,
deleted_object_info.delete_marker,
replication_source.delete_marker,
existing_object_info.is_some(),
) && let Some(existing) = existing_object_info
{
return existing;
}
deleted_object_info
replication_source
}
const AMZ_SNOWBALL_EXTRACT_COMPAT: &str = "X-Amz-Snowball-Auto-Extract";
@@ -6704,7 +6710,7 @@ impl DefaultObjectUsecase {
// the same early, advisory rejection as before.
let store_ref = &store;
let bucket_ref = bucket.as_str();
let admitted_deletes: Vec<Result<AdmittedDelete, ApiError>> =
let admitted_deletes: Vec<AdmittedDelete> =
futures::stream::iter(prepared_deletes.into_iter().map(|prepared| async move {
let PreparedDelete {
idx,
@@ -6729,7 +6735,7 @@ impl DefaultObjectUsecase {
&& let Some(block_reason) = check_object_lock_for_deletion(bucket_ref, &goi, bypass_governance).await
{
let blocked_key = object.object_name.clone();
return Ok(AdmittedDelete {
return AdmittedDelete {
idx,
object,
size: 0,
@@ -6740,11 +6746,15 @@ impl DefaultObjectUsecase {
message: Some(block_reason.error_message()),
version_id,
}),
});
};
}
let size = goi.size;
if is_dir_object(&object.object_name) && object.version_id.is_none() {
object.version_id = Some(Uuid::nil());
}
if replicate_deletes {
let dsc = check_replicate_delete(
bucket_ref,
@@ -6757,8 +6767,7 @@ impl DefaultObjectUsecase {
&opts,
gerr.clone(),
)
.await
.map_err(ApiError::from)?;
.await;
if dsc.replicate_any() {
if object.version_id.is_some() {
set_object_to_delete_version_purge_status(&mut object, VersionPurgeStatusType::Pending);
@@ -6770,23 +6779,18 @@ impl DefaultObjectUsecase {
}
}
if is_dir_object(&object.object_name) && object.version_id.is_none() {
object.version_id = Some(Uuid::nil());
}
let existing = (!skip_stat && gerr.is_none()).then_some(goi);
Ok(AdmittedDelete {
AdmittedDelete {
idx,
object,
size,
existing,
blocked: None,
})
}
}))
.buffered(DELETE_OBJECTS_PRE_STAT_CONCURRENCY)
.collect()
.await;
let admitted_deletes: Vec<AdmittedDelete> = admitted_deletes.into_iter().collect::<Result<_, ApiError>>()?;
// Phase 3 (serial): apply outcomes in the original request order so
// per-key success/failure reporting is unchanged.
@@ -7104,28 +7108,6 @@ impl DefaultObjectUsecase {
}
};
let delete_replication_state = if !replica && !force_delete && (opts.versioned || opts.version_suspended) {
let fallback_source;
let source = if let Some(source) = existing_object_info.as_ref() {
source
} else {
fallback_source = ObjectInfo {
name: key.clone(),
..Default::default()
};
&fallback_source
};
match metadata_sys::get_replication_config(&bucket).await {
Ok((config, _)) => {
delete_replication_state_from_config(&config, source, version_id_clone.as_ref().and(source.version_id), false)
}
Err(StorageError::ConfigNotFound) => None,
Err(err) => return Err(ApiError::from(err).into()),
}
} else {
None
};
let cache_adapter = self.object_data_cache();
// A force (delete_prefix) delete removes every object under `key` as a
// prefix, so invalidating only the exact key would strand every cached
@@ -7214,14 +7196,14 @@ impl DefaultObjectUsecase {
let _delete_tail_guard = DeleteTailActivityGuard::new(DeleteTailStage::Tail);
let deleted_object_source = deleted_replication_info.unwrap_or(&obj_info);
let replication_state_source =
delete_replication_state_source(&opts, existing_object_info.as_ref(), deleted_object_source);
delete_replication_state_source(&opts, existing_object_info.as_ref(), deleted_object_source, &obj_info);
let deleted_delete_marker_version = deleted_replication_info.is_some_and(|info| info.delete_marker);
let delete_replication_version_id = delete_replication_version_id(deleted_object_source, deleted_delete_marker_version);
let schedule_delete_replication = if opts.replication_request && replica {
should_schedule_replica_delete_replication(&bucket, replication_state_source, delete_replication_version_id).await
} else {
delete_replication_state.is_some()
should_schedule_delete_replication(&opts, replication_state_source, deleted_delete_marker_version)
};
if schedule_delete_replication {
@@ -7243,12 +7225,8 @@ impl DefaultObjectUsecase {
replication_state: None,
..Default::default()
};
if let Some(state) = delete_replication_state.as_ref() {
set_deleted_object_replication_state(&mut deleted_object, state);
} else {
set_deleted_object_replication_state(&mut deleted_object, &replication_state_source.replication_state());
enrich_delete_replication_state_if_needed(&bucket, &mut deleted_object, replication_state_source).await;
}
set_deleted_object_replication_state(&mut deleted_object, &replication_state_source.replication_state());
enrich_delete_replication_state_if_needed(&bucket, &mut deleted_object, replication_state_source).await;
schedule_replication_delete(deleted_object, bucket.clone(), REPLICATE_INCOMING_DELETE.to_string()).await;
}
@@ -8386,10 +8364,9 @@ mod tests {
use super::*;
use http::{Extensions, HeaderMap, HeaderName, HeaderValue, Method, Uri};
use s3s::dto::{
Delete, DeleteMarkerReplication, DeleteMarkerReplicationStatus, DeleteReplication, DeleteReplicationStatus, Destination,
ExistingObjectReplication, ExistingObjectReplicationStatus, ObjectIdentifier, ReplicaModifications,
ReplicaModificationsStatus, ReplicationConfiguration, ReplicationRule, ReplicationRuleStatus, RestoreRequest,
SourceSelectionCriteria,
Delete, DeleteMarkerReplication, DeleteMarkerReplicationStatus, Destination, ExistingObjectReplication,
ExistingObjectReplicationStatus, ObjectIdentifier, ReplicaModifications, ReplicaModificationsStatus,
ReplicationConfiguration, ReplicationRule, ReplicationRuleStatus, RestoreRequest, SourceSelectionCriteria,
};
use std::pin::Pin;
use std::sync::Arc;
@@ -13397,6 +13374,64 @@ mod tests {
assert!(!can_skip_delete_objects_pre_stat(false, false, &delete_marker_creating_opts(), false));
}
#[test]
fn should_schedule_delete_replication_skips_replica_requests() {
let opts = ObjectOptions {
replication_request: true,
version_id: Some(Uuid::new_v4().to_string()),
..Default::default()
};
let replication_source = ObjectInfo {
delete_marker: true,
replication_status: ReplicationStatusType::Completed,
..Default::default()
};
assert!(
!should_schedule_delete_replication(&opts, &replication_source, true),
"replica delete requests on target sites must not enqueue a second replication delete task"
);
}
#[test]
fn should_schedule_delete_replication_uses_pending_marker_version_purge_state() {
let opts = ObjectOptions {
replication_request: false,
version_id: Some(Uuid::new_v4().to_string()),
..Default::default()
};
let replication_source = ObjectInfo {
delete_marker: true,
version_purge_status: VersionPurgeStatusType::Pending,
..Default::default()
};
assert!(
should_schedule_delete_replication(&opts, &replication_source, true),
"source-side delete-marker version purge needs scheduling when the current delete rule marked it pending"
);
}
#[test]
fn should_schedule_delete_replication_uses_pending_object_version_purge_state() {
let opts = ObjectOptions {
replication_request: false,
version_id: Some(Uuid::new_v4().to_string()),
..Default::default()
};
let replication_source = ObjectInfo {
delete_marker: false,
replication_status: ReplicationStatusType::Completed,
version_purge_status: VersionPurgeStatusType::Pending,
..Default::default()
};
assert!(
should_schedule_delete_replication(&opts, &replication_source, false),
"source-side object version purge needs scheduling when the current delete rule marked it pending"
);
}
#[tokio::test]
#[ignore = "requires isolated global object layer state"]
async fn execute_get_object_attributes_returns_internal_error_when_store_uninitialized() {
@@ -13653,7 +13688,7 @@ mod tests {
}
#[test]
fn delete_replication_state_from_config_tracks_delete_marker_version_purges() {
fn delete_replication_state_from_config_skips_delete_marker_version_purges_when_delete_is_disabled() {
let arn = "arn:aws:s3:::target-bucket".to_string();
let config = ReplicationConfiguration {
role: arn.clone(),
@@ -13661,11 +13696,9 @@ mod tests {
delete_marker_replication: Some(DeleteMarkerReplication {
status: Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::ENABLED)),
}),
delete_replication: Some(DeleteReplication {
status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED),
}),
delete_replication: None,
destination: Destination {
bucket: arn.clone(),
bucket: arn,
..Default::default()
},
existing_object_replication: Some(ExistingObjectReplication {
@@ -13688,13 +13721,10 @@ mod tests {
};
let version_id = Some(Uuid::new_v4());
let state = delete_replication_state_from_config(&config, &obj_info, version_id, false)
.expect("delete-marker version purge should honor delete-marker replication rules");
let pending = format!("{arn}=PENDING;");
assert_eq!(state.version_purge_status_internal.as_deref(), Some(pending.as_str()));
assert_eq!(state.replicate_decision_str, format!("{arn}=true;false;{arn};"));
assert!(state.purge_targets.contains_key(&arn));
assert!(
delete_replication_state_from_config(&config, &obj_info, version_id, false).is_none(),
"delete-marker version purge must remain local when delete replication is disabled"
);
}
#[test]
@@ -13715,7 +13745,7 @@ mod tests {
..Default::default()
};
let source = delete_replication_state_source(&opts, Some(&existing), &deleted);
let source = delete_replication_state_source(&opts, Some(&existing), &deleted, &deleted);
assert_eq!(source.replication_status, ReplicationStatusType::Completed);
assert!(
@@ -13738,7 +13768,7 @@ mod tests {
..Default::default()
};
let source = delete_replication_state_source(&opts, Some(&existing), &deleted);
let source = delete_replication_state_source(&opts, Some(&existing), &deleted, &deleted);
assert!(
source.delete_marker,
@@ -13746,6 +13776,46 @@ mod tests {
);
}
#[test]
fn delete_replication_state_source_uses_current_state_for_source_version_purge() {
let opts = ObjectOptions {
version_id: Some(Uuid::new_v4().to_string()),
..Default::default()
};
let existing = ObjectInfo {
replication_status: ReplicationStatusType::Completed,
..Default::default()
};
let delete_result = ObjectInfo {
version_purge_status: VersionPurgeStatusType::Pending,
..Default::default()
};
let source = delete_replication_state_source(&opts, Some(&existing), &existing, &delete_result);
assert_eq!(source.version_purge_status, VersionPurgeStatusType::Pending);
}
#[test]
fn delete_replication_state_source_preserves_tags_for_replica_version_purge() {
let opts = ObjectOptions {
replication_request: true,
version_id: Some(Uuid::new_v4().to_string()),
..Default::default()
};
let existing = ObjectInfo {
user_tags: Arc::new("environment=production".to_string()),
replication_status: ReplicationStatusType::Replica,
..Default::default()
};
let delete_result = ObjectInfo::default();
let source = delete_replication_state_source(&opts, Some(&existing), &existing, &delete_result);
assert_eq!(source.user_tags.as_str(), "environment=production");
assert_eq!(source.replication_status, ReplicationStatusType::Replica);
}
#[test]
fn replica_delete_enrichment_must_not_reuse_upstream_targets() {
let upstream_state = ReplicationState {
+17 -6
View File
@@ -649,8 +649,8 @@ pub(crate) mod bucket {
oi: &crate::storage::storage_api::StorageObjectInfo,
del_opts: &crate::storage::storage_api::StorageObjectOptions,
gerr: Option<String>,
) -> Result<ReplicateDecision, crate::storage::storage_api::StorageError> {
ReplicationObjectBridge::check_delete_strict(bucket, dobj, oi, del_opts, gerr).await
) -> ReplicateDecision {
ReplicationObjectBridge::check_delete(bucket, dobj, oi, del_opts, gerr).await
}
pub(crate) fn delete_replication_version_id(
@@ -751,6 +751,21 @@ pub(crate) mod bucket {
replication_contracts::should_remove_replication_target(target_arn, is_replication_service, target_arns)
}
pub(crate) fn should_schedule_delete_replication(
opts: &crate::storage::storage_api::StorageObjectOptions,
replication_source: &crate::storage::storage_api::StorageObjectInfo,
deleted_delete_marker_version: bool,
) -> bool {
replication_contracts::should_schedule_delete_replication(replication_contracts::ReplicationDeleteScheduleInput {
replication_request: opts.replication_request,
version_id_requested: opts.version_id.is_some(),
source_delete_marker: replication_source.delete_marker,
source_replication_status: &replication_source.replication_status,
source_version_purge_status: &replication_source.version_purge_status,
deleted_delete_marker_version,
})
}
pub(crate) fn should_use_existing_delete_replication_info(
opts: &crate::storage::storage_api::StorageObjectOptions,
) -> bool {
@@ -775,10 +790,6 @@ pub(crate) mod bucket {
) -> Result<(), ReplicationTargetValidationError> {
replication_contracts::validate_replication_config_target_arns(configured_arns, config)
}
pub(crate) fn unsupported_replication_config_field(config: &s3s::dto::ReplicationConfiguration) -> Option<&'static str> {
replication_contracts::unsupported_replication_config_field(config)
}
}
pub(crate) mod tagging {
+90 -6
View File
@@ -13,7 +13,14 @@ fi
classify_source_layer() {
local file="$1"
if [[ "$file" == rustfs/src/app/* ]]; then
if [[ "$file" == rustfs/src/server/* ]] ||
[[ "$file" == rustfs/src/startup_*.rs ]] ||
[[ "$file" == rustfs/src/init.rs ]] ||
[[ "$file" == rustfs/src/main.rs ]] ||
[[ "$file" == rustfs/src/lib.rs ]] ||
[[ "$file" == rustfs/src/embedded.rs ]]; then
printf 'composition'
elif [[ "$file" == rustfs/src/app/* ]]; then
printf 'app'
elif [[ "$file" == rustfs/src/admin/* ]] || [[ "$file" == rustfs/src/storage/ecfs.rs ]] || [[ "$file" == rustfs/src/storage/s3_api/* ]]; then
printf 'interface'
@@ -30,6 +37,14 @@ classify_target_layer() {
local storage_path
case "$root" in
init | main | lib | embedded | startup_*)
printf 'composition'
;;
server)
# Server files are composition roots when they import lower layers, but
# their exported HTTP contracts belong to the interface boundary.
printf 'interface'
;;
app)
printf 'app'
;;
@@ -53,6 +68,9 @@ classify_target_layer() {
layer_rank() {
case "$1" in
composition)
printf '4'
;;
interface)
printf '3'
;;
@@ -68,6 +86,50 @@ layer_rank() {
esac
}
is_reverse_dependency() {
local source_rank target_rank
source_rank="$(layer_rank "$1")"
target_rank="$(layer_rank "$2")"
(( source_rank < target_rank ))
}
assert_dependency_direction() {
local expected="$1"
local source="$2"
local target="$3"
local actual='allowed'
if is_reverse_dependency "$source" "$target"; then
actual='reverse'
fi
if [[ "$actual" != "$expected" ]]; then
printf 'Layer dependency guard self-test failed: %s -> %s (expected %s, got %s)\n' \
"$source" "$target" "$expected" "$actual" >&2
exit 1
fi
}
run_layer_model_self_tests() {
local server_source app_source storage_source server_target admin_target app_target storage_target
server_source="$(classify_source_layer rustfs/src/server/http.rs)"
app_source="$(classify_source_layer rustfs/src/app/bucket_usecase.rs)"
storage_source="$(classify_source_layer rustfs/src/storage/rpc/node_service.rs)"
server_target="$(classify_target_layer server::http)"
admin_target="$(classify_target_layer admin::router)"
app_target="$(classify_target_layer app::bucket_usecase)"
storage_target="$(classify_target_layer storage::rpc)"
assert_dependency_direction 'allowed' "$server_source" "$admin_target"
assert_dependency_direction 'allowed' "$server_source" "$app_target"
assert_dependency_direction 'allowed' "$server_source" "$storage_target"
assert_dependency_direction 'reverse' "$app_source" "$server_target"
assert_dependency_direction 'reverse' "$storage_source" "$server_target"
assert_dependency_direction 'reverse' "$app_source" "$admin_target"
assert_dependency_direction 'reverse' "$storage_source" "$admin_target"
}
normalize_import_group_item() {
local prefix="$1"
local item="$2"
@@ -182,6 +244,11 @@ normalize_import_path() {
emit_crate_use_statements() {
(cd "$ROOT_DIR" && rg --files -g '*.rs' rustfs/src | while IFS= read -r file; do
# The guard is file-scoped: dedicated test modules are excluded, while
# inline #[cfg(test)] imports remain subject to the source file's layer.
if [[ "$file" == *_test.rs ]] || [[ "$file" == */tests/* ]]; then
continue
fi
perl -0777 -ne '
while (/\buse\s+crate::.*?;/sg) {
my $statement = $&;
@@ -194,6 +261,26 @@ emit_crate_use_statements() {
done)
}
write_baseline_file() {
local entries="$1"
cat >"$BASELINE_FILE" <<'EOF'
# Layer dependency baseline for the rustfs binary crate.
#
# The guard models production imports as:
# composition -> interface -> app -> infra
#
# Canonical dependency entry:
# dep|source_file|source_layer->target_layer|crate::imported_symbol
#
# Canonical conceptual cycle entry:
# cycle|left_layer<->right_layer
EOF
cat "$entries" >>"$BASELINE_FILE"
}
run_layer_model_self_tests
normalize_baseline_file() {
local input="$1"
local output="$2"
@@ -265,10 +352,7 @@ while IFS= read -r line; do
printf '%s->%s\n' "$source_layer" "$target_layer" >>"$EDGES_RAW"
fi
source_rank="$(layer_rank "$source_layer")"
target_rank="$(layer_rank "$target_layer")"
if (( source_rank < target_rank )); then
if is_reverse_dependency "$source_layer" "$target_layer"; then
printf 'dep|%s|%s->%s|crate::%s\n' "$file" "$source_layer" "$target_layer" "$import_path" >>"$VIOLATIONS_RAW"
fi
done < <(normalize_import_path "$text")
@@ -292,7 +376,7 @@ done <"${TMP_DIR}/edges_sorted.txt" | sort -u >"${TMP_DIR}/cycles_sorted.txt"
cat "${TMP_DIR}/violations_sorted.txt" "${TMP_DIR}/cycles_sorted.txt" | sort -u >"$CURRENT_BASELINE"
if [[ "$MODE" == "update" ]]; then
cp "$CURRENT_BASELINE" "$BASELINE_FILE"
write_baseline_file "$CURRENT_BASELINE"
echo "Updated baseline: $BASELINE_FILE"
exit 0
fi
+29 -20
View File
@@ -1,35 +1,44 @@
# Layer dependency baseline for the rustfs binary crate.
#
# These are intra-crate module references within rustfs/ that cross the
# conceptual layer boundaries (app, infra/server, interface/admin).
# Since they live inside one Cargo crate, Rust doesn't enforce separation.
# The list is maintained for architectural awareness during code review.
# The guard models production imports as:
# composition -> interface -> app -> infra
#
# Format for dependency entries:
# status|source_file|direction|imported_symbol|reason
# Canonical dependency entry:
# dep|source_file|source_layer->target_layer|crate::imported_symbol
#
# Format for conceptual cycles:
# status|cycle|direction_pair|reason
#
# Status:
# accepted - reviewed and intentionally allowed
# todo - should be resolved in a future refactor
# Canonical conceptual cycle entry:
# cycle|left_layer<->right_layer
cycle|app<->infra
cycle|app<->interface
cycle|composition<->infra
cycle|composition<->interface
cycle|infra<->interface
dep|rustfs/src/admin/handlers/scanner.rs|interface->composition|crate::startup_background::ENV_SCANNER_ENABLED
dep|rustfs/src/admin/handlers/scanner.rs|interface->composition|crate::startup_background::scanner_enabled_from_env
dep|rustfs/src/app/admin_usecase.rs|app->interface|crate::server::DependencyReadiness
dep|rustfs/src/app/admin_usecase.rs|app->interface|crate::server::collect_dependency_readiness_report
dep|rustfs/src/app/bucket_usecase.rs|app->interface|crate::admin::handlers::site_replication::site_replication_bucket_meta_hook
dep|rustfs/src/app/bucket_usecase.rs|app->interface|crate::admin::handlers::site_replication::site_replication_delete_bucket_hook
dep|rustfs/src/app/bucket_usecase.rs|app->interface|crate::admin::handlers::site_replication::site_replication_make_bucket_hook
dep|rustfs/src/init.rs|infra->interface|crate::admin
dep|rustfs/src/app/bucket_usecase.rs|app->interface|crate::server::RemoteAddr
dep|rustfs/src/app/object_usecase.rs|app->interface|crate::server::convert_ecstore_object_info
dep|rustfs/src/cluster_snapshot.rs|infra->interface|crate::server::DependencyReadiness
dep|rustfs/src/cluster_snapshot.rs|infra->interface|crate::server::DependencyReadinessReport
dep|rustfs/src/cluster_snapshot.rs|infra->interface|crate::server::ReadinessDegradedReason
dep|rustfs/src/cluster_snapshot.rs|infra->interface|crate::server::snapshot_dependency_readiness_report
dep|rustfs/src/runtime_sources.rs|infra->app|crate::app::context
dep|rustfs/src/server/http.rs|infra->interface|crate::admin
dep|rustfs/src/server/layer.rs|infra->interface|crate::admin::console::is_console_path
dep|rustfs/src/storage/access.rs|infra->interface|crate::server::RemoteAddr
dep|rustfs/src/storage/ecfs_extend.rs|infra->interface|crate::server::cors
dep|rustfs/src/storage/ecfs_extend.rs|infra->interface|crate::storage::ecfs::ListObjectUnorderedQuery
dep|rustfs/src/storage/ecfs_test.rs|infra->interface|crate::storage::ecfs::FS
dep|rustfs/src/storage/ecfs_test.rs|infra->interface|crate::storage::ecfs::validate_object_lock_configuration_input
dep|rustfs/src/storage/ecfs_test.rs|infra->interface|crate::storage::s3_api::common::rustfs_initiator
dep|rustfs/src/storage/ecfs_test.rs|infra->interface|crate::storage::s3_api::common::rustfs_owner
dep|rustfs/src/storage/helper.rs|infra->interface|crate::server::convert_ecstore_object_info
dep|rustfs/src/storage/helper.rs|infra->interface|crate::server::is_audit_module_enabled
dep|rustfs/src/storage/helper.rs|infra->interface|crate::server::is_notify_module_enabled
dep|rustfs/src/storage/helper.rs|infra->interface|crate::server::refresh_audit_module_enabled
dep|rustfs/src/storage/helper.rs|infra->interface|crate::server::refresh_notify_module_enabled
dep|rustfs/src/storage/rpc/http_service.rs|infra->interface|crate::server::RPC_PREFIX
dep|rustfs/src/storage/rpc/node_service.rs|infra->interface|crate::admin::service::config::reload_dynamic_config_runtime_state
dep|rustfs/src/storage/rpc/node_service.rs|infra->interface|crate::admin::service::config::reload_runtime_config_snapshot
dep|rustfs/src/storage/rpc/node_service.rs|infra->interface|crate::admin::service::site_replication::reload_site_replication_runtime_state
dep|rustfs/src/storage/rpc/node_service.rs|infra->interface|crate::server::MODULE_SWITCHES_SIGNAL_SUBSYSTEM
dep|rustfs/src/storage/rpc/node_service/heal.rs|infra->composition|crate::startup_background::heal_enabled_from_env
dep|rustfs/src/storage/rpc/node_service/heal.rs|infra->composition|crate::startup_background::scanner_enabled_from_env
+312 -27
View File
@@ -36,6 +36,16 @@ render_distributed_statefulset() {
'
}
statefulset_env_names() {
yq eval '[(.spec.template.spec.containers[0].env // [])[] | .name] | join(" ")' -
}
statefulset_env_value() {
local rustfs_test_env_name=$1
RUSTFS_TEST_ENV_NAME="$rustfs_test_env_name" \
yq eval '.spec.template.spec.containers[0].env[] | select(.name == strenv(RUSTFS_TEST_ENV_NAME)) | .value' -
}
render_distributed_configmap() {
helm template rustfs "$CHART_DIR" \
--namespace rustfs \
@@ -403,38 +413,308 @@ if grep -q "name: data$" <<<"$generic_eight_by_two"; then
exit 1
fi
# Distributed startup coordination must use the configured endpoint port and
# stay transport-neutral so the same wait works with and without mTLS.
custom_port_startup=$(render_distributed_statefulset \
# Issue #5416's exact three-node, one-drive topology must render all three
# ordinal hosts while each StatefulSet pod receives its own explicit anchor.
three_by_one_configmap=$(render_distributed_configmap --set replicaCount=3 --set drivesPerNode=1)
if ! grep -Fq 'RUSTFS_VOLUMES: "http://rustfs-{0...2}.rustfs-headless.rustfs.svc.cluster.local:9000/data"' <<<"$three_by_one_configmap"; then
echo "Three-node topology must render ordinals 0 through 2 in RUSTFS_VOLUMES" >&2
exit 1
fi
# The distributed init container only prepares directories. Peer DNS or TCP
# availability must not gate the RustFS process from applying its own quorum
# checks. The deprecated timeout remains accepted but is not rendered.
distributed_startup=$(render_distributed_statefulset \
--set replicaCount=3 \
--set drivesPerNode=1 \
--set startupWaitTimeoutSeconds=0)
for removed_gate in nslookup 'nc -z' wait_for_peer STARTUP_WAIT_TIMEOUT_SECONDS ENDPOINT_PORT; do
if grep -Fq "$removed_gate" <<<"$distributed_startup"; then
echo "Distributed init must not contain the removed peer gate: $removed_gate" >&2
exit 1
fi
done
if ! grep -Fq 'mkdir -p /data/rustfs$i' <<<"$distributed_startup"; then
echo "Distributed init must keep per-drive directory initialization" >&2
exit 1
fi
if ! grep -Eq '^[[:space:]]+mkdir -p /data$' <<<"$distributed_startup"; then
echo "Distributed init must keep single-drive directory initialization" >&2
exit 1
fi
# Chart-generated volumes use the Downward API pod name to construct the exact
# endpoint hostname. Order is load-bearing because Kubernetes only expands
# environment references that were defined earlier in the env list.
generated_env_names=$(statefulset_env_names <<<"$distributed_startup")
if [[ "$generated_env_names" != "RUSTFS_CHART_POD_NAME RUSTFS_LOCAL_ENDPOINT_HOST" ]]; then
echo "Unexpected generated topology env order: $generated_env_names" >&2
exit 1
fi
pod_name_field=$(yq eval '.spec.template.spec.containers[0].env[] | select(.name == "RUSTFS_CHART_POD_NAME") | .valueFrom.fieldRef.fieldPath' - <<<"$distributed_startup")
local_endpoint_host=$(statefulset_env_value RUSTFS_LOCAL_ENDPOINT_HOST <<<"$distributed_startup")
if [[ "$pod_name_field" != "metadata.name" ]]; then
echo "Generated topology must source RUSTFS_CHART_POD_NAME from the Downward API" >&2
exit 1
fi
if [[ "$local_endpoint_host" != '$(RUSTFS_CHART_POD_NAME).rustfs-headless.rustfs.svc.cluster.local' ]]; then
echo "Unexpected generated RUSTFS_LOCAL_ENDPOINT_HOST: $local_endpoint_host" >&2
exit 1
fi
# Non-default release identity, namespace, cluster domain, mTLS, and endpoint
# port must compose into one generated topology without restoring peer gates.
nondefault_generated=$(helm template tenant "$CHART_DIR" \
--namespace object-storage \
--set secret.rustfs.access_key=test-access-key \
--set secret.rustfs.secret_key=test-secret-key \
--set clusterDomain=cluster.corp \
--set mtls.enabled=true \
--set service.endpoint.port=9443 \
--set config.rustfs.address=:9443)
grep -q 'name: ENDPOINT_PORT' <<<"$custom_port_startup"
grep -A1 'name: ENDPOINT_PORT' <<<"$custom_port_startup" | grep -q 'value: "9443"'
grep -q 'nc -z -w 2' <<<"$custom_port_startup"
if grep -q 'http.*9000/health' <<<"$custom_port_startup"; then
echo "Distributed startup wait must not hardcode the HTTP scheme or port 9000" >&2
nondefault_statefulset=$(awk '
/^# Source: rustfs\/templates\/statefulset.yaml$/ { in_statefulset = 1 }
in_statefulset && /^---$/ { exit }
in_statefulset { print }
' <<<"$nondefault_generated")
nondefault_local_host=$(statefulset_env_value RUSTFS_LOCAL_ENDPOINT_HOST <<<"$nondefault_statefulset")
nondefault_namespace=$(yq eval '.metadata.namespace' - <<<"$nondefault_statefulset")
nondefault_endpoint_port=$(yq eval '.spec.template.spec.containers[0].ports[] | select(.name == "endpoint") | .containerPort' - <<<"$nondefault_statefulset")
mtls_liveness_command=$(yq eval '.spec.template.spec.containers[0].livenessProbe.exec.command[-1]' - <<<"$nondefault_statefulset")
mtls_readiness_command=$(yq eval '.spec.template.spec.containers[0].readinessProbe.exec.command[-1]' - <<<"$nondefault_statefulset")
if [[ "$nondefault_local_host" != '$(RUSTFS_CHART_POD_NAME).tenant-rustfs-headless.object-storage.svc.cluster.corp' ]]; then
echo "Combined generated topology produced an unexpected local endpoint host: $nondefault_local_host" >&2
exit 1
fi
if [[ "$nondefault_namespace" != "object-storage" || "$nondefault_endpoint_port" != "9443" ]]; then
echo "Combined generated topology must retain its namespace and custom endpoint port" >&2
exit 1
fi
if ! grep -Fq 'https://127.0.0.1:9443/health' <<<"$mtls_liveness_command" ||
grep -Fq '/health/ready' <<<"$mtls_liveness_command"; then
echo "mTLS liveness must use the process health endpoint, not readiness" >&2
exit 1
fi
if ! grep -Fq 'https://127.0.0.1:9443/health/ready' <<<"$mtls_readiness_command"; then
echo "mTLS readiness must use the ready endpoint" >&2
exit 1
fi
if ! grep -Fq 'RUSTFS_VOLUMES: "https://tenant-rustfs-{0...3}.tenant-rustfs-headless.object-storage.svc.cluster.corp:9443/data/rustfs{0...3}"' <<<"$nondefault_generated"; then
echo "Combined generated topology must use its release fullname, namespace, domain, mTLS scheme, and endpoint port" >&2
exit 1
fi
if grep -Eq 'nslookup|nc -z|wait_for_peer' <<<"$nondefault_generated"; then
echo "Combined generated topology must not render peer DNS or TCP gates" >&2
exit 1
fi
mtls_startup=$(render_distributed_statefulset --set mtls.enabled=true)
grep -q 'nc -z -w 2' <<<"$mtls_startup"
if grep -q 'wget .*http.*health' <<<"$mtls_startup"; then
echo "mTLS startup wait must not use a plaintext HTTP health probe" >&2
default_scheme_port_generated=$(render_distributed_configmap \
--set service.endpoint.port=80 \
--set config.rustfs.address=:80)
if ! grep -Fq 'rustfs-headless.rustfs.svc.cluster.local:80/' <<<"$default_scheme_port_generated"; then
echo "Chart-generated topology must retain an explicit HTTP default port" >&2
exit 1
fi
# The timeout is one global deadline and invalid non-positive values fail
# during rendering instead of creating an unbounded init wait.
grep -q 'deadline=$(($(date +%s) + STARTUP_WAIT_TIMEOUT_SECONDS))' <<<"$custom_port_startup"
invalid_startup_timeout_status=0
render_distributed_statefulset --set startupWaitTimeoutSeconds=0 >/dev/null 2>&1 || invalid_startup_timeout_status=$?
if [[ $invalid_startup_timeout_status -eq 0 ]]; then
echo "Distributed mode with startupWaitTimeoutSeconds=0 must fail rendering" >&2
# Unrelated extraEnv entries follow the generated entries so dependent
# environment expansion keeps working.
generated_with_extra_env=$(render_distributed_statefulset \
--set 'extraEnv[0].name=CUSTOM_ENV' \
--set 'extraEnv[0].value=custom-value')
generated_with_extra_names=$(statefulset_env_names <<<"$generated_with_extra_env")
if [[ "$generated_with_extra_names" != "RUSTFS_CHART_POD_NAME RUSTFS_LOCAL_ENDPOINT_HOST CUSTOM_ENV" ]]; then
echo "extraEnv must be appended after generated topology env entries" >&2
exit 1
fi
# Every pool waits for DNS across the complete normalized pool list, while
# port availability follows the stable pool-index/ordinal startup order.
# Explicit bounded/fail-fast modes retain their legacy DNS locality semantics,
# so the generated anchor must not make the runtime reject the configuration.
for legacy_wait_mode in bounded ' Strict '; do
bounded_wait_mode=$(render_distributed_statefulset \
--set 'extraEnv[0].name=RUSTFS_STARTUP_TOPOLOGY_WAIT_MODE' \
--set-string "extraEnv[0].value=$legacy_wait_mode")
bounded_wait_env_names=$(statefulset_env_names <<<"$bounded_wait_mode")
bounded_wait_value=$(yq eval '.spec.template.spec.containers[0].env[0].value' - <<<"$bounded_wait_mode")
if [[ "$bounded_wait_env_names" != "RUSTFS_STARTUP_TOPOLOGY_WAIT_MODE" || "$bounded_wait_value" != "$legacy_wait_mode" ]]; then
echo "Startup mode $legacy_wait_mode must opt out of the generated local endpoint anchor" >&2
exit 1
fi
done
unknown_wait_mode=$(render_distributed_statefulset \
--set 'extraEnv[0].name=RUSTFS_STARTUP_TOPOLOGY_WAIT_MODE' \
--set 'extraEnv[0].value=invalid-mode')
unknown_wait_env_names=$(statefulset_env_names <<<"$unknown_wait_mode")
if [[ "$unknown_wait_env_names" != "RUSTFS_STARTUP_TOPOLOGY_WAIT_MODE" ]]; then
echo "An unknown explicit startup mode must not receive a generated local endpoint anchor" >&2
exit 1
fi
dynamic_wait_mode=$(render_distributed_statefulset \
--set 'extraEnv[0].name=RUSTFS_STARTUP_TOPOLOGY_WAIT_MODE' \
--set 'extraEnv[0].valueFrom.configMapKeyRef.name=startup-policy' \
--set 'extraEnv[0].valueFrom.configMapKeyRef.key=mode')
dynamic_wait_env_names=$(statefulset_env_names <<<"$dynamic_wait_mode")
if [[ "$dynamic_wait_env_names" != "RUSTFS_STARTUP_TOPOLOGY_WAIT_MODE" ]]; then
echo "Dynamically sourced startup mode must opt out of the generated local endpoint anchor" >&2
exit 1
fi
orchestrated_wait_mode=$(render_distributed_statefulset \
--set 'extraEnv[0].name=RUSTFS_STARTUP_TOPOLOGY_WAIT_MODE' \
--set 'extraEnv[0].value=orchestrated')
orchestrated_wait_env_names=$(statefulset_env_names <<<"$orchestrated_wait_mode")
if [[ "$orchestrated_wait_env_names" != "RUSTFS_CHART_POD_NAME RUSTFS_LOCAL_ENDPOINT_HOST RUSTFS_STARTUP_TOPOLOGY_WAIT_MODE" ]]; then
echo "Explicit orchestrated startup mode must retain one generated local endpoint anchor" >&2
exit 1
fi
# Existing POD_NAME remains independent from the chart-private Downward API
# variable, while an explicit anchor disables automatic anchor injection.
existing_pod_name=$(render_distributed_statefulset \
--set 'extraEnv[0].name=POD_NAME' \
--set 'extraEnv[0].valueFrom.fieldRef.fieldPath=metadata.name')
existing_pod_name_names=$(statefulset_env_names <<<"$existing_pod_name")
existing_pod_name_count=$(yq eval '[.spec.template.spec.containers[0].env[] | select(.name == "POD_NAME")] | length' - <<<"$existing_pod_name")
existing_pod_name_field=$(yq eval '.spec.template.spec.containers[0].env[] | select(.name == "POD_NAME") | .valueFrom.fieldRef.fieldPath' - <<<"$existing_pod_name")
existing_chart_pod_name_field=$(yq eval '.spec.template.spec.containers[0].env[] | select(.name == "RUSTFS_CHART_POD_NAME") | .valueFrom.fieldRef.fieldPath' - <<<"$existing_pod_name")
existing_pod_name_anchor=$(statefulset_env_value RUSTFS_LOCAL_ENDPOINT_HOST <<<"$existing_pod_name")
if [[ "$existing_pod_name_names" != "RUSTFS_CHART_POD_NAME RUSTFS_LOCAL_ENDPOINT_HOST POD_NAME" ||
"$existing_pod_name_count" != "1" ||
"$existing_pod_name_field" != "metadata.name" ||
"$existing_chart_pod_name_field" != "metadata.name" ||
"$existing_pod_name_anchor" != '$(RUSTFS_CHART_POD_NAME).rustfs-headless.rustfs.svc.cluster.local' ]]; then
echo "An existing extraEnv POD_NAME must be preserved without interfering with the automatic anchor" >&2
exit 1
fi
existing_local_anchor=$(render_distributed_statefulset \
--set 'extraEnv[0].name=RUSTFS_LOCAL_ENDPOINT_HOST' \
--set-string 'extraEnv[0].value=rustfs-0.rustfs-headless.rustfs.svc.cluster.local')
existing_local_anchor_names=$(statefulset_env_names <<<"$existing_local_anchor")
existing_local_anchor_value=$(statefulset_env_value RUSTFS_LOCAL_ENDPOINT_HOST <<<"$existing_local_anchor")
if [[ "$existing_local_anchor_names" != "RUSTFS_LOCAL_ENDPOINT_HOST" || "$existing_local_anchor_value" != "rustfs-0.rustfs-headless.rustfs.svc.cluster.local" ]]; then
echo "An existing explicit local endpoint anchor must be preserved without a generated duplicate" >&2
exit 1
fi
# `helm upgrade --reuse-values` can omit keys introduced by the new chart.
# A missing localEndpointHost map keeps the new safe default instead of
# failing template evaluation.
missing_local_endpoint_host_values=$(render_distributed_statefulset \
--set-json 'localEndpointHost=null')
missing_local_endpoint_host_names=$(statefulset_env_names <<<"$missing_local_endpoint_host_values")
if [[ "$missing_local_endpoint_host_names" != "RUSTFS_CHART_POD_NAME RUSTFS_LOCAL_ENDPOINT_HOST" ]]; then
echo "Missing localEndpointHost values from a reused release must default to automatic anchor injection" >&2
exit 1
fi
# An opaque existing Secret keeps the historical DNS path by default because
# it may hide runtime topology or startup-mode overrides. A credentials-only
# Secret can explicitly opt in without changing envFrom precedence.
existing_secret_statefulset=$(render_distributed_statefulset \
--set secret.existingSecret=legacy-rustfs-config)
existing_secret_env_names=$(statefulset_env_names <<<"$existing_secret_statefulset")
existing_secret_env_from=$(yq eval '[.spec.template.spec.containers[0].envFrom[] | (.configMapRef.name // .secretRef.name)] | join(" ")' - <<<"$existing_secret_statefulset")
if [[ -n "$existing_secret_env_names" || "$existing_secret_env_from" != "rustfs-config legacy-rustfs-config" ]]; then
echo "An opaque existingSecret must default to the legacy environment path" >&2
exit 1
fi
existing_secret_opt_in=$(render_distributed_statefulset \
--set secret.existingSecret=legacy-rustfs-config \
--set localEndpointHost.autoInject=true)
existing_secret_opt_in_names=$(statefulset_env_names <<<"$existing_secret_opt_in")
existing_secret_opt_in_env_from=$(yq eval '[.spec.template.spec.containers[0].envFrom[] | (.configMapRef.name // .secretRef.name)] | join(" ")' - <<<"$existing_secret_opt_in")
if [[ "$existing_secret_opt_in_names" != "RUSTFS_CHART_POD_NAME RUSTFS_LOCAL_ENDPOINT_HOST" ||
"$existing_secret_opt_in_env_from" != "rustfs-config legacy-rustfs-config" ]]; then
echo "A credentials-only existingSecret must support explicit automatic anchor injection" >&2
exit 1
fi
# A Secret that hides a custom topology can explicitly preserve the historical
# DNS path. No explicit env entries are added, and envFrom ordering is unchanged.
existing_secret_opt_out=$(render_distributed_statefulset \
--set secret.existingSecret=legacy-rustfs-config \
--set localEndpointHost.autoInject=false)
existing_secret_opt_out_env_names=$(statefulset_env_names <<<"$existing_secret_opt_out")
existing_secret_opt_out_env_from=$(yq eval '[.spec.template.spec.containers[0].envFrom[] | (.configMapRef.name // .secretRef.name)] | join(" ")' - <<<"$existing_secret_opt_out")
if [[ -n "$existing_secret_opt_out_env_names" || "$existing_secret_opt_out_env_from" != "rustfs-config legacy-rustfs-config" ]]; then
echo "localEndpointHost.autoInject=false must preserve the legacy existingSecret environment path" >&2
exit 1
fi
quoted_local_endpoint_host_status=0
quoted_local_endpoint_host_error=$(render_distributed_statefulset \
--set secret.existingSecret=legacy-rustfs-config \
--set-string localEndpointHost.autoInject=false 2>&1) || quoted_local_endpoint_host_status=$?
if [[ $quoted_local_endpoint_host_status -eq 0 ||
"$quoted_local_endpoint_host_error" != *"localEndpointHost.autoInject must be a boolean or null"* ]]; then
echo "A quoted localEndpointHost.autoInject value must fail closed instead of becoming truthy" >&2
exit 1
fi
missing_local_endpoint_host_with_secret=$(render_distributed_statefulset \
--set secret.existingSecret=legacy-rustfs-config \
--set-json 'localEndpointHost=null')
if [[ -n "$(statefulset_env_names <<<"$missing_local_endpoint_host_with_secret")" ]]; then
echo "A reused release without localEndpointHost values must not override an opaque existingSecret" >&2
exit 1
fi
# An explicit topology disables inference, but an address override alone keeps
# the generated anchor so RustFS can validate the effective port before format
# initialization.
volumes_override_statefulset=$(render_distributed_statefulset \
--set 'extraEnv[0].name=RUSTFS_VOLUMES' \
--set-string 'extraEnv[0].value=http://legacy.example/data')
volumes_override_names=$(statefulset_env_names <<<"$volumes_override_statefulset")
volumes_override_value=$(yq eval '.spec.template.spec.containers[0].env[0].value' - <<<"$volumes_override_statefulset")
if [[ "$volumes_override_names" != "RUSTFS_VOLUMES" || "$volumes_override_value" != "http://legacy.example/data" ]]; then
echo "extraEnv RUSTFS_VOLUMES must retain precedence without a generated anchor" >&2
exit 1
fi
address_override_statefulset=$(render_distributed_statefulset \
--set 'extraEnv[0].name=RUSTFS_ADDRESS' \
--set-string 'extraEnv[0].value=:9443')
address_override_names=$(statefulset_env_names <<<"$address_override_statefulset")
address_override_value=$(statefulset_env_value RUSTFS_ADDRESS <<<"$address_override_statefulset")
address_override_anchor_count=$(yq eval '[.spec.template.spec.containers[0].env[] | select(.name == "RUSTFS_LOCAL_ENDPOINT_HOST")] | length' - <<<"$address_override_statefulset")
if [[ "$address_override_names" != "RUSTFS_CHART_POD_NAME RUSTFS_LOCAL_ENDPOINT_HOST RUSTFS_ADDRESS" ||
"$address_override_value" != ":9443" ||
"$address_override_anchor_count" != "1" ]]; then
echo "extraEnv RUSTFS_ADDRESS must retain one generated local endpoint anchor" >&2
exit 1
fi
# Explicit volumes remain authoritative: the chart must not infer any identity,
# while a user-supplied local endpoint anchor remains allowed.
custom_volumes_statefulset=$(render_distributed_statefulset \
--set config.rustfs.volumes=http://example.test/data)
custom_volumes_env_names=$(statefulset_env_names <<<"$custom_volumes_statefulset")
if [[ -n "$custom_volumes_env_names" ]]; then
echo "Custom volumes must not receive generated topology env entries" >&2
exit 1
fi
custom_volumes_with_anchor=$(render_distributed_statefulset \
--set config.rustfs.volumes=http://example.test/data \
--set 'extraEnv[0].name=RUSTFS_LOCAL_ENDPOINT_HOST' \
--set 'extraEnv[0].value=example.test')
custom_anchor=$(statefulset_env_value RUSTFS_LOCAL_ENDPOINT_HOST <<<"$custom_volumes_with_anchor")
if [[ "$custom_anchor" != "example.test" ]]; then
echo "Custom volumes must allow a user-supplied local endpoint anchor" >&2
exit 1
fi
if grep -q 'RUSTFS_LOCAL_ENDPOINT_HOST' <<<"$standalone_default_output"; then
echo "Standalone mode must not receive a distributed local endpoint anchor" >&2
exit 1
fi
# Every generated server pool uses its runtime pod name with the shared root
# headless service. No pool waits for another pool's DNS or TCP listener.
multi_pool_startup=$(helm template rustfs "$CHART_DIR" \
--namespace rustfs \
--set secret.rustfs.access_key=test-access-key \
@@ -442,14 +722,19 @@ multi_pool_startup=$(helm template rustfs "$CHART_DIR" \
--set pools.enabled=true \
--set 'pools.list[0].replicaCount=2' \
--set 'pools.list[1].replicaCount=2')
grep -q 'nslookup "rustfs-0.rustfs-headless.rustfs.svc.cluster.local"' <<<"$multi_pool_startup"
grep -q 'nslookup "rustfs-pool1-1.rustfs-headless.rustfs.svc.cluster.local"' <<<"$multi_pool_startup"
if grep -q 'rustfs-pool1-headless' <<<"$multi_pool_startup"; then
echo "Multi-pool startup wait must use the shared root headless service" >&2
multi_pool_anchor_count=$(grep -c 'name: RUSTFS_LOCAL_ENDPOINT_HOST' <<<"$multi_pool_startup" || true)
if [[ $multi_pool_anchor_count -ne 2 ]]; then
echo "Every generated server pool must receive RUSTFS_LOCAL_ENDPOINT_HOST" >&2
exit 1
fi
if grep -Eq 'nslookup|nc -z|wait_for_peer' <<<"$multi_pool_startup"; then
echo "Multi-pool init must not wait for peer DNS or TCP" >&2
exit 1
fi
if grep -q 'rustfs-pool1-headless' <<<"$multi_pool_startup"; then
echo "Multi-pool local endpoint identity must use the shared root headless service" >&2
exit 1
fi
grep -q 'wait_for_peer "rustfs-1:${ENDPOINT_PORT}"' <<<"$multi_pool_startup"
grep -q 'wait_for_peer "rustfs-pool1-${i}:${ENDPOINT_PORT}"' <<<"$multi_pool_startup"
# volumeClaimTemplates must not contain empty annotations when pvcAnnotations are unset,
# because Kubernetes treats annotations: {} as a mutation of the immutable field.