mirror of
https://github.com/rustfs/rustfs.git
synced 2026-11-02 17:13:37 +00:00
Merge branch 'main' into houseme/fix/fips-provider
This commit is contained in:
Chris An
@@ -1,2 +1,2 @@
|
||||
sha256-linux=686dcc075e41a75dde3d37e5bd0393d016469f02f83aa3508f92bfc17f26a4d4
|
||||
sha256-darwin=686dcc075e41a75dde3d37e5bd0393d016469f02f83aa3508f92bfc17f26a4d4
|
||||
sha256-linux=2f97cbbeb2dbf52919b704a3d07336c162a41f95079d941d74251f9279c494e6
|
||||
sha256-darwin=2f97cbbeb2dbf52919b704a3d07336c162a41f95079d941d74251f9279c494e6
|
||||
|
||||
@@ -827,7 +827,7 @@ slow-timeout = { period = "120s", terminate-after = 12, grace-period = "10s" }
|
||||
# ---------------------------------------------------------------------------
|
||||
# e2e-distributed profile — 4-node 4-disk Actions suite
|
||||
# ---------------------------------------------------------------------------
|
||||
# Storage-sensitive PR / nightly / dispatch lane owned by
|
||||
# Nightly and manual-dispatch lane owned by
|
||||
# .github/workflows/e2e-distributed.yml.
|
||||
# Each case starts four rustfs processes (and for site replication, two
|
||||
# clusters). Upgrade cases also require RUSTFS_UPGRADE_SOURCE_BINARY.
|
||||
|
||||
@@ -32,6 +32,8 @@ updates:
|
||||
- "overtrue"
|
||||
- "majinghe"
|
||||
ignore:
|
||||
- dependency-name: "pulsar"
|
||||
versions: [ "6.9.x" ]
|
||||
- dependency-name: "object_store"
|
||||
versions: [ "0.13.x" ]
|
||||
- dependency-name: "libunftp"
|
||||
|
||||
@@ -769,12 +769,10 @@
|
||||
"workflow": ".github/workflows/e2e-distributed.yml",
|
||||
"name": "e2e-distributed",
|
||||
"roles": [
|
||||
"pr",
|
||||
"periodic",
|
||||
"manual"
|
||||
],
|
||||
"events": [
|
||||
"pull_request",
|
||||
"workflow_dispatch",
|
||||
"schedule"
|
||||
],
|
||||
@@ -849,12 +847,11 @@
|
||||
"workflow": ".github/workflows/e2e-upgrade.yml",
|
||||
"name": "Upgrade Compatibility",
|
||||
"roles": [
|
||||
"pr",
|
||||
"periodic",
|
||||
"manual"
|
||||
"manual",
|
||||
"release"
|
||||
],
|
||||
"events": [
|
||||
"pull_request",
|
||||
"push",
|
||||
"schedule",
|
||||
"workflow_dispatch"
|
||||
|
||||
@@ -17,8 +17,8 @@
|
||||
# Each selected test starts a real localhost cluster via
|
||||
# `RustFSTestClusterEnvironment` (4 processes; 4 drives per node unless the
|
||||
# case is a two-site 4-node 1-drive pair or a 4-node upgrade). Membership is
|
||||
# `[profile.e2e-distributed]` in `.config/nextest.toml`. Storage-sensitive PRs,
|
||||
# nightly runs, and manual dispatches all execute the same fail-closed suite.
|
||||
# `[profile.e2e-distributed]` in `.config/nextest.toml`. The nightly schedule
|
||||
# and manual dispatches execute the same fail-closed suite.
|
||||
# Upgrade cases download the same pinned previous release as e2e-upgrade.yml.
|
||||
#
|
||||
# Isolated pool filesystems: expand/decommission/rebalance cases require
|
||||
@@ -33,32 +33,6 @@
|
||||
name: e2e-distributed
|
||||
|
||||
on:
|
||||
pull_request:
|
||||
paths:
|
||||
- "Cargo.lock"
|
||||
- "Cargo.toml"
|
||||
- ".config/nextest.toml"
|
||||
- ".config/e2e-distributed-selection.txt"
|
||||
- ".github/workflows/e2e-distributed.yml"
|
||||
- "crates/audit/**"
|
||||
- "crates/common/**"
|
||||
- "crates/config/**"
|
||||
- "crates/e2e_test/**"
|
||||
- "crates/ecstore/**"
|
||||
- "crates/filemeta/**"
|
||||
- "crates/heal/**"
|
||||
- "crates/iam/**"
|
||||
- "crates/lock/**"
|
||||
- "crates/madmin/**"
|
||||
- "crates/notify/**"
|
||||
- "crates/replication/**"
|
||||
- "crates/s3-client/**"
|
||||
- "crates/s3-ops/**"
|
||||
- "crates/s3-types/**"
|
||||
- "crates/scanner/**"
|
||||
- "crates/storage-api/**"
|
||||
- "crates/utils/**"
|
||||
- "rustfs/**"
|
||||
workflow_dispatch:
|
||||
inputs:
|
||||
filter:
|
||||
|
||||
@@ -12,30 +12,19 @@
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
# Single-node upgrade and rollback lane.
|
||||
#
|
||||
# The weekly Monday schedule, version tags, and workflow_dispatch run this
|
||||
# suite. Distributed upgrade cases stay on the nightly e2e-distributed lane.
|
||||
|
||||
name: Upgrade Compatibility
|
||||
|
||||
on:
|
||||
pull_request:
|
||||
paths:
|
||||
- ".github/workflows/e2e-upgrade.yml"
|
||||
- "crates/e2e_test/src/common.rs"
|
||||
- "crates/e2e_test/src/distributed/**"
|
||||
- "crates/e2e_test/src/fake_s3_target/**"
|
||||
- "crates/e2e_test/src/lib.rs"
|
||||
- "crates/e2e_test/src/replication_extension_test.rs"
|
||||
- "crates/e2e_test/src/rpc_fault_proxy.rs"
|
||||
- "crates/e2e_test/src/upgrade_compatibility_test.rs"
|
||||
- "crates/ecstore/**"
|
||||
- "crates/filemeta/**"
|
||||
- "crates/heal/**"
|
||||
- "crates/kms/**"
|
||||
- "crates/storage-api/**"
|
||||
- "rustfs/**"
|
||||
- "Cargo.lock"
|
||||
push:
|
||||
tags:
|
||||
- "[0-9]*.[0-9]*.[0-9]*"
|
||||
schedule:
|
||||
# 03:17 UTC every Monday.
|
||||
- cron: "17 3 * * 1"
|
||||
workflow_dispatch:
|
||||
|
||||
@@ -44,7 +33,7 @@ permissions:
|
||||
|
||||
concurrency:
|
||||
group: ${{ github.workflow }}-${{ github.ref }}
|
||||
cancel-in-progress: ${{ github.event_name == 'pull_request' }}
|
||||
cancel-in-progress: false
|
||||
|
||||
env:
|
||||
CARGO_TERM_COLOR: always
|
||||
|
||||
Generated
+214
-97
@@ -370,17 +370,17 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "3f04e0a89cc175eaf3740282537851e6fd9d7276f97730b9d0ea2859c11a85df"
|
||||
dependencies = [
|
||||
"arrow-arith",
|
||||
"arrow-array",
|
||||
"arrow-buffer",
|
||||
"arrow-array 59.3.2",
|
||||
"arrow-buffer 59.3.2",
|
||||
"arrow-cast",
|
||||
"arrow-csv",
|
||||
"arrow-data",
|
||||
"arrow-ipc",
|
||||
"arrow-data 59.3.2",
|
||||
"arrow-ipc 59.3.2",
|
||||
"arrow-json",
|
||||
"arrow-ord",
|
||||
"arrow-row",
|
||||
"arrow-schema",
|
||||
"arrow-select",
|
||||
"arrow-schema 59.3.2",
|
||||
"arrow-select 59.3.2",
|
||||
"arrow-string",
|
||||
]
|
||||
|
||||
@@ -390,10 +390,10 @@ version = "59.3.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "157943c720919e8f5b6731407ce5cd4c9a5ef47a75bf10edaf24b7081b5763ef"
|
||||
dependencies = [
|
||||
"arrow-array",
|
||||
"arrow-buffer",
|
||||
"arrow-data",
|
||||
"arrow-schema",
|
||||
"arrow-array 59.3.2",
|
||||
"arrow-buffer 59.3.2",
|
||||
"arrow-data 59.3.2",
|
||||
"arrow-schema 59.3.2",
|
||||
"chrono",
|
||||
"num-traits",
|
||||
]
|
||||
@@ -405,9 +405,9 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "72b5b1682924f11ecf6929689c16cbc69efbb6dd141b06ad98ee1cb10d858c43"
|
||||
dependencies = [
|
||||
"ahash",
|
||||
"arrow-buffer",
|
||||
"arrow-data",
|
||||
"arrow-schema",
|
||||
"arrow-buffer 59.3.2",
|
||||
"arrow-data 59.3.2",
|
||||
"arrow-schema 59.3.2",
|
||||
"chrono",
|
||||
"chrono-tz",
|
||||
"half",
|
||||
@@ -417,6 +417,24 @@ dependencies = [
|
||||
"num-traits",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "arrow-array"
|
||||
version = "60.1.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "0f3d22a41d3df78b71df0b685af219ae3a94188ab98d6fb29c958eb55c3247f4"
|
||||
dependencies = [
|
||||
"ahash",
|
||||
"arrow-buffer 60.1.0",
|
||||
"arrow-data 60.1.0",
|
||||
"arrow-schema 60.1.0",
|
||||
"chrono",
|
||||
"half",
|
||||
"hashbrown 0.17.1",
|
||||
"num-complex",
|
||||
"num-integer",
|
||||
"num-traits",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "arrow-buffer"
|
||||
version = "59.3.2"
|
||||
@@ -429,18 +447,30 @@ dependencies = [
|
||||
"num-traits",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "arrow-buffer"
|
||||
version = "60.1.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "7bd46a530bc1a8e763102fc36c1dc0f51f3ffa150c78e7f5658ccb54b1458ade"
|
||||
dependencies = [
|
||||
"bytes",
|
||||
"half",
|
||||
"num-bigint 0.5.1",
|
||||
"num-traits",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "arrow-cast"
|
||||
version = "59.3.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "3f23c3379feede42135ef27ca60e67f6e643cd4343ae74e4c8b16c83f55afd85"
|
||||
dependencies = [
|
||||
"arrow-array",
|
||||
"arrow-buffer",
|
||||
"arrow-data",
|
||||
"arrow-array 59.3.2",
|
||||
"arrow-buffer 59.3.2",
|
||||
"arrow-data 59.3.2",
|
||||
"arrow-ord",
|
||||
"arrow-schema",
|
||||
"arrow-select",
|
||||
"arrow-schema 59.3.2",
|
||||
"arrow-select 59.3.2",
|
||||
"atoi 2.0.0",
|
||||
"base64 0.23.1",
|
||||
"chrono",
|
||||
@@ -451,15 +481,26 @@ dependencies = [
|
||||
"ryu",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "arrow-cmp"
|
||||
version = "60.1.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "1913e57a64fbf2c4502fa6d54b547d6a1ae59e3146cbc15321d47468c937abd1"
|
||||
dependencies = [
|
||||
"arrow-array 60.1.0",
|
||||
"arrow-buffer 60.1.0",
|
||||
"arrow-schema 60.1.0",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "arrow-csv"
|
||||
version = "59.3.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "85b1984628b091b8319b3c305fce21d04f9e9b2578bb733b0d0108462ae1d698"
|
||||
dependencies = [
|
||||
"arrow-array",
|
||||
"arrow-array 59.3.2",
|
||||
"arrow-cast",
|
||||
"arrow-schema",
|
||||
"arrow-schema 59.3.2",
|
||||
"chrono",
|
||||
"csv",
|
||||
"csv-core",
|
||||
@@ -472,8 +513,21 @@ version = "59.3.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "bb0a031c7ff5e2bdb443117054fb466af1d4df5b7f0ecbe197713dc8b113b1ba"
|
||||
dependencies = [
|
||||
"arrow-buffer",
|
||||
"arrow-schema",
|
||||
"arrow-buffer 59.3.2",
|
||||
"arrow-schema 59.3.2",
|
||||
"half",
|
||||
"num-integer",
|
||||
"num-traits",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "arrow-data"
|
||||
version = "60.1.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "415b7fbc4cf56b080bb5677a4ae74211b5c79e571ed37a1d547633ec07f3b43b"
|
||||
dependencies = [
|
||||
"arrow-buffer 60.1.0",
|
||||
"arrow-schema 60.1.0",
|
||||
"half",
|
||||
"num-integer",
|
||||
"num-traits",
|
||||
@@ -485,28 +539,43 @@ version = "59.3.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f5dd1afadda2eb821aaf851bb4c0adef60c30cc0f1a29bd02fb99dd07fb9f58d"
|
||||
dependencies = [
|
||||
"arrow-array",
|
||||
"arrow-buffer",
|
||||
"arrow-data",
|
||||
"arrow-schema",
|
||||
"arrow-select",
|
||||
"arrow-array 59.3.2",
|
||||
"arrow-buffer 59.3.2",
|
||||
"arrow-data 59.3.2",
|
||||
"arrow-schema 59.3.2",
|
||||
"arrow-select 59.3.2",
|
||||
"flatbuffers",
|
||||
"lz4_flex",
|
||||
"zstd 0.13.4",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "arrow-ipc"
|
||||
version = "60.1.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "45f3e766bccf5a0cb57a5de52a4cba450dea5171547c2cb5939b486d08887cba"
|
||||
dependencies = [
|
||||
"arrow-array 60.1.0",
|
||||
"arrow-buffer 60.1.0",
|
||||
"arrow-data 60.1.0",
|
||||
"arrow-schema 60.1.0",
|
||||
"arrow-select 60.1.0",
|
||||
"flatbuffers",
|
||||
"zstd 0.14.3",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "arrow-json"
|
||||
version = "59.3.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "e2bc4d56a0dad84cc51c5812ccff48b642047a1f3be80349cd84318983e79edb"
|
||||
dependencies = [
|
||||
"arrow-array",
|
||||
"arrow-buffer",
|
||||
"arrow-array 59.3.2",
|
||||
"arrow-buffer 59.3.2",
|
||||
"arrow-cast",
|
||||
"arrow-ord",
|
||||
"arrow-schema",
|
||||
"arrow-select",
|
||||
"arrow-schema 59.3.2",
|
||||
"arrow-select 59.3.2",
|
||||
"chrono",
|
||||
"half",
|
||||
"indexmap 2.14.2",
|
||||
@@ -526,11 +595,11 @@ version = "59.3.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "561b8209faa2cd489764b7fd07d7cb9e8d274d78048c38de59f77c41140cb834"
|
||||
dependencies = [
|
||||
"arrow-array",
|
||||
"arrow-buffer",
|
||||
"arrow-data",
|
||||
"arrow-schema",
|
||||
"arrow-select",
|
||||
"arrow-array 59.3.2",
|
||||
"arrow-buffer 59.3.2",
|
||||
"arrow-data 59.3.2",
|
||||
"arrow-schema 59.3.2",
|
||||
"arrow-select 59.3.2",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -539,10 +608,10 @@ version = "59.3.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "0d8c68b87a569f92b90c46cb0a09d86bd9609ddc9b1bf1fc5a503c8c6907c14e"
|
||||
dependencies = [
|
||||
"arrow-array",
|
||||
"arrow-buffer",
|
||||
"arrow-data",
|
||||
"arrow-schema",
|
||||
"arrow-array 59.3.2",
|
||||
"arrow-buffer 59.3.2",
|
||||
"arrow-data 59.3.2",
|
||||
"arrow-schema 59.3.2",
|
||||
"half",
|
||||
]
|
||||
|
||||
@@ -556,6 +625,12 @@ dependencies = [
|
||||
"serde_json",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "arrow-schema"
|
||||
version = "60.1.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "de2c8e714b466404cdc1189e89f892426efea614d938e16c15518c340cd43ba3"
|
||||
|
||||
[[package]]
|
||||
name = "arrow-select"
|
||||
version = "59.3.2"
|
||||
@@ -563,10 +638,25 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "c844141958d805b47d1208e0d12aac53fcf388ce8011639fe82aa492adde0bb4"
|
||||
dependencies = [
|
||||
"ahash",
|
||||
"arrow-array",
|
||||
"arrow-buffer",
|
||||
"arrow-data",
|
||||
"arrow-schema",
|
||||
"arrow-array 59.3.2",
|
||||
"arrow-buffer 59.3.2",
|
||||
"arrow-data 59.3.2",
|
||||
"arrow-schema 59.3.2",
|
||||
"num-traits",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "arrow-select"
|
||||
version = "60.1.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "a5af3f6656c5f2c6944f93d9a357a7f204c93d5f349385a44cf46c4a26fbcf30"
|
||||
dependencies = [
|
||||
"ahash",
|
||||
"arrow-array 60.1.0",
|
||||
"arrow-buffer 60.1.0",
|
||||
"arrow-cmp",
|
||||
"arrow-data 60.1.0",
|
||||
"arrow-schema 60.1.0",
|
||||
"num-traits",
|
||||
]
|
||||
|
||||
@@ -576,11 +666,11 @@ version = "59.3.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "bc130892b1a9552da77971924b390e06cf375d1f7e4344a7b7b40ca4426fa470"
|
||||
dependencies = [
|
||||
"arrow-array",
|
||||
"arrow-buffer",
|
||||
"arrow-data",
|
||||
"arrow-schema",
|
||||
"arrow-select",
|
||||
"arrow-array 59.3.2",
|
||||
"arrow-buffer 59.3.2",
|
||||
"arrow-data 59.3.2",
|
||||
"arrow-schema 59.3.2",
|
||||
"arrow-select 59.3.2",
|
||||
"memchr",
|
||||
"num-traits",
|
||||
"regex",
|
||||
@@ -669,9 +759,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "async-compression"
|
||||
version = "0.4.51"
|
||||
version = "0.4.52"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "9d5755640120b24961f8231e05892e4a4f7e0d04e8aeca161508c9c397946970"
|
||||
checksum = "6ad47a740fb906fffdd891e8b6d5fde2511600f7bfaebd68939fa08a14999459"
|
||||
dependencies = [
|
||||
"compression-codecs",
|
||||
"compression-core",
|
||||
@@ -1993,9 +2083,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "cc"
|
||||
version = "1.10.0"
|
||||
version = "1.11.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "86e055fa6e3e40ad6187ba25b6855eeb635d5271f8e0f0e1e9a5826b8d75bea0"
|
||||
checksum = "375989decec1a0fa424442bed520fa12ede43a48475ce8c7417d2c69b32b5f6b"
|
||||
dependencies = [
|
||||
"find-msvc-tools",
|
||||
"jobserver",
|
||||
@@ -2251,9 +2341,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "compression-codecs"
|
||||
version = "0.4.46"
|
||||
version = "0.4.47"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "cae36683101b2899f48bd31f8db9839437356188e3b1ff23096e73dab09ba8fe"
|
||||
checksum = "84f840cff84d10085260e460023934d0af699e71212b09b2a41452e8bd9cda43"
|
||||
dependencies = [
|
||||
"brotli 9.1.1",
|
||||
"bzip2",
|
||||
@@ -2499,9 +2589,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "crc32fast"
|
||||
version = "1.5.3"
|
||||
version = "1.5.4"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "0ba106e844045ddc02c20768ac4b0b7cb45f2cc2ecbd833ac9f5dacb183c2055"
|
||||
checksum = "5f97d3416d5337d15c4b0d78c8c91df922ca143f57ec20e2d1c188333047ef04"
|
||||
dependencies = [
|
||||
"cfg-if",
|
||||
]
|
||||
@@ -2891,7 +2981,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "4cd5317db77ba6bca25c761a8091b9d294a9f4188e4f9dcd0243e6a57de862ac"
|
||||
dependencies = [
|
||||
"arrow",
|
||||
"arrow-schema",
|
||||
"arrow-schema 59.3.2",
|
||||
"async-trait",
|
||||
"chrono",
|
||||
"datafusion-catalog",
|
||||
@@ -2925,7 +3015,7 @@ dependencies = [
|
||||
"log",
|
||||
"object_store",
|
||||
"parking_lot 0.12.5",
|
||||
"parquet",
|
||||
"parquet 59.3.2",
|
||||
"sqlparser",
|
||||
"tempfile",
|
||||
"tokio",
|
||||
@@ -2989,8 +3079,8 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d26626e12b224b83351239465dcd00f72d47b02fb5e5b65a643e2dc8ed37f687"
|
||||
dependencies = [
|
||||
"arrow",
|
||||
"arrow-ipc",
|
||||
"arrow-schema",
|
||||
"arrow-ipc 59.3.2",
|
||||
"arrow-schema 59.3.2",
|
||||
"chrono",
|
||||
"foldhash 0.2.0",
|
||||
"half",
|
||||
@@ -3001,7 +3091,7 @@ dependencies = [
|
||||
"log",
|
||||
"num-traits",
|
||||
"object_store",
|
||||
"parquet",
|
||||
"parquet 59.3.2",
|
||||
"recursive",
|
||||
"sqlparser",
|
||||
"tokio",
|
||||
@@ -3057,7 +3147,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "65b1b041688cad707fa103c3e7f4c6d89b3134ebf67da0129070dad3381851eb"
|
||||
dependencies = [
|
||||
"arrow",
|
||||
"arrow-ipc",
|
||||
"arrow-ipc 59.3.2",
|
||||
"async-trait",
|
||||
"bytes",
|
||||
"datafusion-common",
|
||||
@@ -3127,7 +3217,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "7c521ae91b6d899d7190812881f45fe9463c222c11d90149f5332a2e56555904"
|
||||
dependencies = [
|
||||
"arrow",
|
||||
"arrow-schema",
|
||||
"arrow-schema 59.3.2",
|
||||
"async-trait",
|
||||
"bytes",
|
||||
"datafusion-common",
|
||||
@@ -3148,7 +3238,7 @@ dependencies = [
|
||||
"log",
|
||||
"object_store",
|
||||
"parking_lot 0.12.5",
|
||||
"parquet",
|
||||
"parquet 59.3.2",
|
||||
"tokio",
|
||||
]
|
||||
|
||||
@@ -3165,7 +3255,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "39cfdd5ac0769d1e74a885450e49fde8f63ca7ab41eb479a068564aaea80b837"
|
||||
dependencies = [
|
||||
"arrow",
|
||||
"arrow-buffer",
|
||||
"arrow-buffer 59.3.2",
|
||||
"async-trait",
|
||||
"bytes",
|
||||
"dashmap",
|
||||
@@ -3191,7 +3281,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "86560fd7df4dda924f5474b588c56e7ae8e96dcf158c7327c9cfe9daff448ec5"
|
||||
dependencies = [
|
||||
"arrow",
|
||||
"arrow-schema",
|
||||
"arrow-schema 59.3.2",
|
||||
"async-trait",
|
||||
"chrono",
|
||||
"datafusion-common",
|
||||
@@ -3227,7 +3317,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "939ae2771da8a3fdc23ab4de610ae8fef9515476626b49d96d0d5c782b105771"
|
||||
dependencies = [
|
||||
"arrow",
|
||||
"arrow-buffer",
|
||||
"arrow-buffer 59.3.2",
|
||||
"base64 0.23.1",
|
||||
"chrono",
|
||||
"chrono-tz",
|
||||
@@ -3387,7 +3477,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "9f002cef849e10c0597a217ec7891c2796c0f8b42d739f5d24abc24496373406"
|
||||
dependencies = [
|
||||
"arrow",
|
||||
"arrow-schema",
|
||||
"arrow-schema 59.3.2",
|
||||
"datafusion-common",
|
||||
"datafusion-expr",
|
||||
"datafusion-expr-common",
|
||||
@@ -3462,10 +3552,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b1451d6817e5ac0dd21819b19949d1c420ab51d748f55dd9a96ecb5418a24c97"
|
||||
dependencies = [
|
||||
"arrow",
|
||||
"arrow-data",
|
||||
"arrow-ipc",
|
||||
"arrow-data 59.3.2",
|
||||
"arrow-ipc 59.3.2",
|
||||
"arrow-ord",
|
||||
"arrow-schema",
|
||||
"arrow-schema 59.3.2",
|
||||
"async-trait",
|
||||
"bytes",
|
||||
"datafusion-common",
|
||||
@@ -3513,7 +3603,7 @@ version = "55.2.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "62dc5ce37a5ce1ff5362949776829aded2529a11dbaa13d24b0ffa6a059e106a"
|
||||
dependencies = [
|
||||
"arrow-schema",
|
||||
"arrow-schema 59.3.2",
|
||||
"async-trait",
|
||||
"datafusion-common",
|
||||
"datafusion-execution",
|
||||
@@ -7591,12 +7681,12 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "8eb260d566f2fbdabe356d26aaa5d13403aac29b5b4e44463b8f9bd7b2b12680"
|
||||
dependencies = [
|
||||
"ahash",
|
||||
"arrow-array",
|
||||
"arrow-buffer",
|
||||
"arrow-data",
|
||||
"arrow-ipc",
|
||||
"arrow-schema",
|
||||
"arrow-select",
|
||||
"arrow-array 59.3.2",
|
||||
"arrow-buffer 59.3.2",
|
||||
"arrow-data 59.3.2",
|
||||
"arrow-ipc 59.3.2",
|
||||
"arrow-schema 59.3.2",
|
||||
"arrow-select 59.3.2",
|
||||
"base64 0.23.1",
|
||||
"brotli 8.0.4",
|
||||
"bytes",
|
||||
@@ -7618,6 +7708,32 @@ dependencies = [
|
||||
"zstd 0.13.4",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "parquet"
|
||||
version = "60.1.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "806eaa0f1c0436dcb43bee8e6a82bf2efff90521a60172f8281cb9ef577b43a4"
|
||||
dependencies = [
|
||||
"ahash",
|
||||
"arrow-array 60.1.0",
|
||||
"arrow-buffer 60.1.0",
|
||||
"arrow-data 60.1.0",
|
||||
"arrow-ipc 60.1.0",
|
||||
"arrow-schema 60.1.0",
|
||||
"arrow-select 60.1.0",
|
||||
"base64 0.23.1",
|
||||
"bytes",
|
||||
"chrono",
|
||||
"half",
|
||||
"hashbrown 0.17.1",
|
||||
"num-bigint 0.5.1",
|
||||
"num-integer",
|
||||
"num-traits",
|
||||
"seq-macro",
|
||||
"twox-hash",
|
||||
"zstd 0.14.3",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "password-hash"
|
||||
version = "0.6.1"
|
||||
@@ -8976,9 +9092,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "reflink-copy"
|
||||
version = "0.1.31"
|
||||
version = "0.1.32"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "3dd31f458a66eeda05a9e837fc5f0effbaacbe8565868ecae5b37bf0544c19f5"
|
||||
checksum = "1ad64f61357bcf16105a0f353f9783733409c8bd889bc326ad857959605ea927"
|
||||
dependencies = [
|
||||
"cfg-if",
|
||||
"libc",
|
||||
@@ -10214,18 +10330,18 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "rustfs-mimalloc"
|
||||
version = "0.7.2"
|
||||
version = "0.7.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "8301a578772fc9823d68e083e42c3b414613d588a0eac594c1ca6bc0bbe0653f"
|
||||
checksum = "216ecfc9341b5d6e2506d47ce57c9232ae1b8020bb10264ea05286fb234286e7"
|
||||
dependencies = [
|
||||
"rustfs-mimalloc-sys",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "rustfs-mimalloc-sys"
|
||||
version = "0.7.2"
|
||||
version = "0.7.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "6eb4082853043cb29556b4b776b05f28c3146706f43c53869dbf1ef262e00e91"
|
||||
checksum = "fb16ad68f96883407a75cdb88252b43dc2ea592428f0fc54ff7885756de4b814"
|
||||
dependencies = [
|
||||
"cc",
|
||||
]
|
||||
@@ -12420,18 +12536,18 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "test-case"
|
||||
version = "3.4.0"
|
||||
version = "3.4.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "124953e7f67cb0b2fcfb87e899e4ae5a64fb68e2e160767933cc67b646fb0042"
|
||||
checksum = "ae6e80608d2037173d7d59a09089d249e43352c182f6dde6b00d138bcb7cfdbe"
|
||||
dependencies = [
|
||||
"test-case-macros",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "test-case-core"
|
||||
version = "3.4.0"
|
||||
version = "3.4.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "bd097615b407247e102046ba34426bb594923ab3554cf0f3cdfed050f1a5a3e8"
|
||||
checksum = "29b8e1f65819001927b8cdcbb5560f2428212f2e5ab5684e3c6815a4d80e72db"
|
||||
dependencies = [
|
||||
"cfg-if",
|
||||
"proc-macro2",
|
||||
@@ -12441,9 +12557,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "test-case-macros"
|
||||
version = "3.4.0"
|
||||
version = "3.4.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f4ce35ece947bccac166e1ded639133827a953f69a48ba134d17f91108333082"
|
||||
checksum = "802d6165a14df1598b7290a8c546df402940d3cfbe82d15c0fd35de176771693"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
@@ -14082,9 +14198,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "zip"
|
||||
version = "9.2.1"
|
||||
version = "9.2.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "848dad21639597378c3be94c59fd4de683f0f6d8a5387575896e1245737d4f37"
|
||||
checksum = "e408c6cc4699dd017d53dc080cd51c2cea2b5354af57a1479bd265ed42911487"
|
||||
dependencies = [
|
||||
"aes 0.9.3",
|
||||
"bzip2",
|
||||
@@ -14095,6 +14211,7 @@ dependencies = [
|
||||
"getrandom 0.4.3",
|
||||
"hmac 0.13.0",
|
||||
"indexmap 2.14.2",
|
||||
"libc",
|
||||
"lzma-rust2",
|
||||
"memchr",
|
||||
"nix 0.31.3",
|
||||
@@ -14138,12 +14255,12 @@ name = "zstd"
|
||||
version = "0.13.4"
|
||||
dependencies = [
|
||||
"apache-avro",
|
||||
"arrow-array",
|
||||
"arrow-ipc",
|
||||
"arrow-schema",
|
||||
"arrow-array 60.1.0",
|
||||
"arrow-ipc 60.1.0",
|
||||
"arrow-schema 60.1.0",
|
||||
"bytes",
|
||||
"kafka-protocol",
|
||||
"parquet",
|
||||
"parquet 60.1.0",
|
||||
"semver",
|
||||
"toml",
|
||||
"zstd 0.14.3",
|
||||
|
||||
+4
-4
@@ -147,7 +147,7 @@ rustfs-zip = { path = "./crates/zip", version = "1.0.2" }
|
||||
async-channel = "2.5.0"
|
||||
async_zip = { default-features = false, version = "0.0.19" }
|
||||
mysql_async = { default-features = false, version = "0.37.1" }
|
||||
async-compression = { version = "0.4.51" }
|
||||
async-compression = { version = "0.4.52" }
|
||||
async-trait = "0.1.92"
|
||||
async-nats = { version = "0.51.0", default-features = false }
|
||||
axum = "0.8.9"
|
||||
@@ -337,7 +337,7 @@ strum = { version = "0.28.0" }
|
||||
sysinfo = "0.40.1"
|
||||
temp-env = "0.3.6"
|
||||
tempfile = "3.27.0"
|
||||
test-case = "3.4.0"
|
||||
test-case = "3.4.1"
|
||||
thiserror = "2.0.21"
|
||||
tracing = { version = "0.1.44" }
|
||||
tracing-appender = "0.2.5"
|
||||
@@ -356,7 +356,7 @@ winapi-util = "0.1.11"
|
||||
windows = { version = "0.62.2" }
|
||||
windows-sys = "0.61.2"
|
||||
xxhash-rust = { version = "0.8.19" }
|
||||
zip = "9.2.1"
|
||||
zip = "9.2.2"
|
||||
zstd = "0.14.3"
|
||||
|
||||
# Observability and Metrics
|
||||
@@ -384,7 +384,7 @@ russh-sftp = "3.0.1"
|
||||
dav-server = "0.11.0"
|
||||
|
||||
# Performance Analysis and Memory Profiling
|
||||
rustfs-mimalloc = { version = "0.7.2" }
|
||||
rustfs-mimalloc = { version = "0.7.3" }
|
||||
# Preserve Unicode focus filters until rustfs/backlog#2302 is resolved.
|
||||
hotpath = { version = "0.30.3", default-features = false }
|
||||
# Snapshot testing for output format regression detection
|
||||
|
||||
@@ -26,7 +26,7 @@ Registered in [`src/lib.rs`](src/lib.rs). Grouped by concern:
|
||||
| **protocols** | [`src/protocols/`](src/protocols) | FTPS, WebDAV, SFTP compliance. Fixed ports, own guide: [`src/protocols/README.md`](src/protocols/README.md) |
|
||||
| **reliant** | [`src/reliant/`](src/reliant) | Tests that reuse an **externally started** server (SQL/select, conditional writes, lifecycle, deleted-object reads, node-interact). Run via [`scripts/run_e2e_tests.sh`](../../scripts/run_e2e_tests.sh); see [`src/reliant/README.md`](src/reliant/README.md) |
|
||||
| **cluster** | `cluster_concurrency_test`, `stale_multipart_cleanup_cluster_test`, `namespace_lock_quorum_test`, `admin_timeout_regression_test`, `object_lambda_test`, `replication_extension_test`, `tier_stats_cluster_test` | Multi-node scenarios via `RustFSTestClusterEnvironment` |
|
||||
| **distributed 4×4** | [`src/distributed/`](src/distributed) | Storage-sensitive PR and nightly `e2e-distributed` lane: S3, object lock/WORM, versioning, bucket/site replication, quota, expand/decommission/rebalance, concurrency, chaos, 4-node upgrade of historical data and IAM AK/SK. Map: [`docs/testing/distributed-e2e.md`](../../docs/testing/distributed-e2e.md) |
|
||||
| **distributed 4×4** | [`src/distributed/`](src/distributed) | Nightly `e2e-distributed` lane: S3, object lock/WORM, versioning, bucket/site replication, quota, expand/decommission/rebalance, concurrency, chaos, 4-node upgrade of historical data and IAM AK/SK. Map: [`docs/testing/distributed-e2e.md`](../../docs/testing/distributed-e2e.md) |
|
||||
| **chaos / reliability** | [`src/chaos.rs`](src/chaos.rs), `reliability_disk_fault_test`, `heal_erasure_disk_rebuild_test`, `server_startup_failfast_test` | Disk offline/replace/corrupt, EC rebuild, heal, fail-fast startup |
|
||||
| **upgrade compatibility** | `upgrade_compatibility_test` | Pinned previous-release writes followed by current-build reads on the same data directory |
|
||||
|
||||
@@ -194,9 +194,9 @@ the same profile for membership and execution with one nightly worker.
|
||||
| `s3s-e2e` black-box | `e2e-tests` + `e2e-tests-rio-v2` jobs | **Active** (external conformance tool) |
|
||||
| ILM / lifecycle (ignored) | `test-ilm-integration-serial` lane, `-j1` | **Active** (backlog#1148 ilm-1) |
|
||||
| KMS suite | `e2e-full` job, merge queue + main | **Active** |
|
||||
| Direct and mixed-version rolling upgrades from pinned previous release | `e2e-upgrade.yml`, storage-sensitive PRs + release tags + weekly | **Active** |
|
||||
| Direct and mixed-version rolling upgrades from pinned previous release | `e2e-upgrade.yml`, release tags + weekly schedule + manual dispatch | **Active** |
|
||||
| Cluster faults (`e2e-nightly` profile) | consolidated nightly workflow | **Active** (backlog#1149 ci-7) |
|
||||
| Distributed 4-node 4-disk (`e2e-distributed` profile) | `.github/workflows/e2e-distributed.yml` | **Active** (storage-sensitive PR / nightly / dispatch) |
|
||||
| Distributed 4-node 4-disk (`e2e-distributed` profile) | `.github/workflows/e2e-distributed.yml` | **Active** (nightly / dispatch) |
|
||||
| Protocols (FTPS/WebDAV/SFTP) | consolidated nightly workflow, serial | **Active** (backlog#1149 ci-7) |
|
||||
| Replication (fast subset) | `e2e-smoke` profile, `e2e-tests` job, every PR | **Active** (backlog#1147 repl-1) |
|
||||
| Replication (slow + multi-node) | `e2e-repl-nightly` profile, consolidated nightly workflow | **Active** (backlog#1147 repl-1) |
|
||||
|
||||
@@ -12,11 +12,249 @@
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use super::harness::{DistCluster, DistLayout, TestResult, assert_object_bytes, payload_for, put_object, unique_bucket};
|
||||
use super::harness::{DistCluster, DistLayout, TestResult, payload_for, s3_probe_client, sha256_hex, unique_bucket};
|
||||
use crate::common::init_logging;
|
||||
use aws_sdk_s3::{
|
||||
Client,
|
||||
error::{ProvideErrorMetadata, SdkError},
|
||||
primitives::ByteStream,
|
||||
};
|
||||
use std::any::TypeId;
|
||||
use std::collections::BTreeSet;
|
||||
use std::fmt::Debug;
|
||||
use std::future::Future;
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
use tokio::sync::Barrier;
|
||||
use tokio::time::Instant;
|
||||
|
||||
const LOCK_ADMISSION_BUDGET: Duration = Duration::from_secs(45);
|
||||
|
||||
async fn after_lock_contention<T, E, F, Fut>(operation: &str, mut call: F) -> TestResult<Result<T, SdkError<E>>>
|
||||
where
|
||||
E: Debug + ProvideErrorMetadata + 'static,
|
||||
F: FnMut() -> Fut,
|
||||
Fut: Future<Output = Result<T, SdkError<E>>>,
|
||||
{
|
||||
let started = Instant::now();
|
||||
let mut attempts = 0;
|
||||
// The workload uses unique unversioned keys. Repeat only explicitly
|
||||
// advertised lock backpressure; every step still needs its real result.
|
||||
let outcome = tokio::time::timeout(LOCK_ADMISSION_BUDGET, async {
|
||||
loop {
|
||||
attempts += 1;
|
||||
let result = call().await;
|
||||
let retry = result.as_ref().err().is_some_and(|error| {
|
||||
let Some(response) = error.raw_response() else {
|
||||
return false;
|
||||
};
|
||||
let Some(service_error) = error.as_service_error() else {
|
||||
return false;
|
||||
};
|
||||
// HEAD has no XML error body, so its SDK service error can
|
||||
// lack a code even when the server supplies retry headers.
|
||||
let namespace_code = service_error.code() == Some("ServiceUnavailable")
|
||||
|| (service_error.code().is_none()
|
||||
&& TypeId::of::<E>() == TypeId::of::<aws_sdk_s3::operation::head_object::HeadObjectError>());
|
||||
response.status().as_u16() == 503
|
||||
&& namespace_code
|
||||
&& response.headers().get("x-rustfs-error-cause") == Some("namespace_lock_timeout")
|
||||
&& response.headers().get("retry-after") == Some("1")
|
||||
});
|
||||
if !retry {
|
||||
return result;
|
||||
}
|
||||
tokio::time::sleep(Duration::from_secs(1)).await;
|
||||
}
|
||||
})
|
||||
.await;
|
||||
if attempts > 1 || !matches!(&outcome, Ok(Ok(_))) {
|
||||
eprintln!(
|
||||
"mixed workload request: operation={operation}, attempts={attempts}, elapsed={:?}",
|
||||
started.elapsed()
|
||||
);
|
||||
}
|
||||
outcome.map_err(|_| {
|
||||
format!("{operation} exceeded its {LOCK_ADMISSION_BUDGET:?} request budget after {attempts} attempts").into()
|
||||
})
|
||||
}
|
||||
|
||||
async fn assert_workload_bytes(client: &Client, bucket: &str, key: &str, expected: &[u8]) -> TestResult {
|
||||
let output = after_lock_contention(&format!("GET {key}"), || client.get_object().bucket(bucket).key(key).send()).await??;
|
||||
let actual = output.body.collect().await?.into_bytes();
|
||||
if actual.as_ref() != expected {
|
||||
return Err(format!(
|
||||
"object {key} bytes mismatch: expected {} bytes sha256={} got {} bytes sha256={}",
|
||||
expected.len(),
|
||||
sha256_hex(expected),
|
||||
actual.len(),
|
||||
sha256_hex(&actual)
|
||||
)
|
||||
.into());
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use aws_smithy_http_client::test_util::infallible_client_fn;
|
||||
use aws_smithy_types::body::SdkBody;
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
|
||||
type Reply = (u16, &'static str, Option<&'static str>, Option<&'static str>, &'static str);
|
||||
|
||||
fn client(respond: impl Fn(usize) -> Reply + Send + Sync + 'static) -> (Client, Arc<AtomicUsize>) {
|
||||
let requests = Arc::new(AtomicUsize::new(0));
|
||||
let seen = Arc::clone(&requests);
|
||||
let transport = infallible_client_fn(move |_| {
|
||||
let (status, code, cause, retry_after, body) = respond(seen.fetch_add(1, Ordering::SeqCst));
|
||||
let body = if code.is_empty() {
|
||||
body.to_string()
|
||||
} else {
|
||||
format!("<Error><Code>{code}</Code></Error>")
|
||||
};
|
||||
let mut response = http::Response::builder()
|
||||
.status(status)
|
||||
.header("content-type", "application/xml")
|
||||
.header("content-length", body.len());
|
||||
if let Some(cause) = cause {
|
||||
response = response.header("x-rustfs-error-cause", cause);
|
||||
}
|
||||
if let Some(retry_after) = retry_after {
|
||||
response = response.header("retry-after", retry_after);
|
||||
}
|
||||
response.body(SdkBody::from(body)).expect("scripted S3 response")
|
||||
});
|
||||
let config = crate::common::build_test_s3_config(
|
||||
"http://contention.example.invalid",
|
||||
"synthetic-access",
|
||||
"synthetic-secret",
|
||||
None,
|
||||
"contention-test",
|
||||
);
|
||||
let client = Client::from_conf(config.to_builder().http_client(transport).build());
|
||||
(s3_probe_client(&client), requests)
|
||||
}
|
||||
|
||||
fn contention() -> Reply {
|
||||
(503, "ServiceUnavailable", Some("namespace_lock_timeout"), Some("1"), "")
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn advertised_contention_recovers_after_standard_sdk_attempts_are_exhausted() -> TestResult {
|
||||
let (client, requests) = client(|attempt| {
|
||||
if attempt < 3 {
|
||||
contention()
|
||||
} else {
|
||||
(200, "", None, None, "")
|
||||
}
|
||||
});
|
||||
let standard = Client::from_conf(
|
||||
client
|
||||
.config()
|
||||
.to_builder()
|
||||
.retry_config(aws_sdk_s3::config::retry::RetryConfig::standard().with_max_attempts(3))
|
||||
.build(),
|
||||
);
|
||||
standard
|
||||
.put_object()
|
||||
.bucket("bucket")
|
||||
.key("key")
|
||||
.body(ByteStream::from_static(b"body"))
|
||||
.send()
|
||||
.await
|
||||
.expect_err("three advertised lock responses exhaust the standard SDK policy");
|
||||
assert_eq!(requests.load(Ordering::SeqCst), 3);
|
||||
requests.store(0, Ordering::SeqCst);
|
||||
let started = Instant::now();
|
||||
after_lock_contention("PUT recovery", || {
|
||||
client
|
||||
.put_object()
|
||||
.bucket("bucket")
|
||||
.key("key")
|
||||
.body(ByteStream::from_static(b"body"))
|
||||
.send()
|
||||
})
|
||||
.await??;
|
||||
assert_eq!(requests.load(Ordering::SeqCst), 4);
|
||||
assert!(started.elapsed() >= Duration::from_secs(3), "honor each Retry-After response");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn bodyless_head_contention_uses_explicit_retry_headers() -> TestResult {
|
||||
let (client, requests) = client(|attempt| {
|
||||
if attempt == 0 {
|
||||
(503, "", Some("namespace_lock_timeout"), Some("1"), "")
|
||||
} else {
|
||||
(200, "", None, None, "")
|
||||
}
|
||||
});
|
||||
after_lock_contention("HEAD recovery", || client.head_object().bucket("bucket").key("key").send()).await??;
|
||||
assert_eq!(requests.load(Ordering::SeqCst), 2);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn unrelated_or_malformed_responses_are_not_retried() -> TestResult {
|
||||
for reply in [
|
||||
(500, "InternalError", Some("namespace_lock_timeout"), Some("1"), ""),
|
||||
(403, "AccessDenied", None, None, ""),
|
||||
(404, "NoSuchKey", None, None, ""),
|
||||
(503, "ServiceUnavailable", Some("storage_write_quorum"), None, ""),
|
||||
(503, "ServiceUnavailable", Some("namespace_lock_unavailable"), Some("1"), ""),
|
||||
(503, "InternalError", Some("namespace_lock_timeout"), Some("1"), ""),
|
||||
(503, "", Some("namespace_lock_timeout"), Some("1"), ""),
|
||||
(503, "ServiceUnavailable", None, Some("1"), ""),
|
||||
(503, "ServiceUnavailable", Some("namespace_lock_timeout"), None, ""),
|
||||
(503, "ServiceUnavailable", Some("namespace_lock_timeout"), Some("2"), ""),
|
||||
] {
|
||||
let (client, requests) = client(move |_| reply);
|
||||
let result = after_lock_contention("PUT fatal response", || {
|
||||
client
|
||||
.put_object()
|
||||
.bucket("bucket")
|
||||
.key("key")
|
||||
.body(ByteStream::from_static(b"body"))
|
||||
.send()
|
||||
})
|
||||
.await?;
|
||||
assert!(result.is_err(), "fatal response cannot become success: {reply:?}");
|
||||
assert_eq!(requests.load(Ordering::SeqCst), 1, "fatal response must remain visible: {reply:?}");
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn persistent_lock_contention_exhausts_the_request_budget() {
|
||||
let (client, requests) = client(|_| contention());
|
||||
let started = Instant::now();
|
||||
let result = after_lock_contention("PUT unavailable", || {
|
||||
client
|
||||
.put_object()
|
||||
.bucket("bucket")
|
||||
.key("key")
|
||||
.body(ByteStream::from_static(b"body"))
|
||||
.send()
|
||||
})
|
||||
.await;
|
||||
assert!(result.is_err(), "persistent contention cannot acknowledge an operation");
|
||||
assert!(started.elapsed() >= LOCK_ADMISSION_BUDGET);
|
||||
assert!(requests.load(Ordering::SeqCst) > 1);
|
||||
assert!(requests.load(Ordering::SeqCst) <= LOCK_ADMISSION_BUDGET.as_secs() as usize + 1);
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn acknowledged_wrong_bytes_fail_without_retry() {
|
||||
let (client, requests) = client(|_| (200, "", None, None, "wrong bytes"));
|
||||
let error = assert_workload_bytes(&client, "bucket", "key", b"expected bytes")
|
||||
.await
|
||||
.expect_err("a successful but incorrect read is a consistency failure");
|
||||
assert!(error.to_string().contains("bytes mismatch"));
|
||||
assert_eq!(requests.load(Ordering::SeqCst), 1);
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn four_node_high_concurrency_mixed_workload_is_consistent_on_every_node() -> TestResult {
|
||||
@@ -24,7 +262,9 @@ async fn four_node_high_concurrency_mixed_workload_is_consistent_on_every_node()
|
||||
let dist = DistCluster::start(DistLayout::FourByFour).await?;
|
||||
let bucket = unique_bucket("conc");
|
||||
dist.create_bucket(&bucket).await?;
|
||||
let clients = Arc::new(dist.clients()?);
|
||||
// This consistency probe owns its admission policy; unrelated responses
|
||||
// must not be hidden by the SDK's general-purpose retries.
|
||||
let clients = Arc::new(dist.clients()?.iter().map(s3_probe_client).collect::<Vec<_>>());
|
||||
let worker_count = 24;
|
||||
let rounds = 4;
|
||||
let barrier = Arc::new(Barrier::new(worker_count));
|
||||
@@ -44,31 +284,46 @@ async fn four_node_high_concurrency_mixed_workload_is_consistent_on_every_node()
|
||||
let key = format!("source/worker-{idx:02}-round-{round}.bin");
|
||||
let copy_key = format!("retained/worker-{idx:02}-round-{round}.bin");
|
||||
let body = payload_for(&key, 64 * 1024);
|
||||
put_object(writer, &bucket, &key, body.clone()).await?;
|
||||
let context = format!("worker={idx} round={round}");
|
||||
after_lock_contention(&format!("{context} PUT source"), || {
|
||||
writer
|
||||
.put_object()
|
||||
.bucket(&bucket)
|
||||
.key(&key)
|
||||
.body(ByteStream::from(body.clone()))
|
||||
.send()
|
||||
})
|
||||
.await??;
|
||||
|
||||
let head = reader.head_object().bucket(&bucket).key(&key).send().await?;
|
||||
let head = after_lock_contention(&format!("{context} HEAD source"), || {
|
||||
reader.head_object().bucket(&bucket).key(&key).send()
|
||||
})
|
||||
.await??;
|
||||
if head.content_length() != Some(body.len() as i64) {
|
||||
return Err(format!("HEAD returned the wrong size for {key}: {head:?}").into());
|
||||
}
|
||||
assert_object_bytes(reader, &bucket, &key, &body).await?;
|
||||
assert_workload_bytes(reader, &bucket, &key, &body).await?;
|
||||
|
||||
copier
|
||||
.copy_object()
|
||||
.bucket(&bucket)
|
||||
.key(©_key)
|
||||
.copy_source(format!("{bucket}/{key}"))
|
||||
.send()
|
||||
.await?;
|
||||
assert_object_bytes(writer, &bucket, ©_key, &body).await?;
|
||||
after_lock_contention(&format!("{context} COPY retained"), || {
|
||||
copier
|
||||
.copy_object()
|
||||
.bucket(&bucket)
|
||||
.key(©_key)
|
||||
.copy_source(format!("{bucket}/{key}"))
|
||||
.send()
|
||||
})
|
||||
.await??;
|
||||
assert_workload_bytes(writer, &bucket, ©_key, &body).await?;
|
||||
|
||||
writer.delete_object().bucket(&bucket).key(&key).send().await?;
|
||||
let missing = reader
|
||||
.head_object()
|
||||
.bucket(&bucket)
|
||||
.key(&key)
|
||||
.send()
|
||||
.await
|
||||
.expect_err("deleted source key must not remain visible");
|
||||
after_lock_contention(&format!("{context} DELETE source"), || {
|
||||
writer.delete_object().bucket(&bucket).key(&key).send()
|
||||
})
|
||||
.await??;
|
||||
let missing = after_lock_contention(&format!("{context} HEAD deleted source"), || {
|
||||
reader.head_object().bucket(&bucket).key(&key).send()
|
||||
})
|
||||
.await?
|
||||
.expect_err("deleted source key must not remain visible");
|
||||
if missing.raw_response().map(|response| response.status().as_u16()) != Some(404) {
|
||||
return Err(format!("deleted source {key} returned an unexpected result: {missing:?}").into());
|
||||
}
|
||||
@@ -82,14 +337,22 @@ async fn four_node_high_concurrency_mixed_workload_is_consistent_on_every_node()
|
||||
for handle in handles {
|
||||
inventory.extend(handle.await??);
|
||||
}
|
||||
assert_eq!(
|
||||
inventory.len(),
|
||||
worker_count * rounds,
|
||||
"every worker round must retain its acknowledged copy"
|
||||
);
|
||||
|
||||
let expected_keys: BTreeSet<_> = inventory.iter().map(|(key, _)| key.as_str()).collect();
|
||||
for (node_idx, client) in clients.iter().enumerate() {
|
||||
let listed = client.list_objects_v2().bucket(&bucket).prefix("retained/").send().await?;
|
||||
let listed = after_lock_contention(&format!("node={node_idx} LIST retained"), || {
|
||||
client.list_objects_v2().bucket(&bucket).prefix("retained/").send()
|
||||
})
|
||||
.await??;
|
||||
let listed_keys: BTreeSet<_> = listed.contents().iter().filter_map(|object| object.key()).collect();
|
||||
assert_eq!(listed_keys, expected_keys, "node {node_idx} returned a divergent retained-key listing");
|
||||
for (key, body) in &inventory {
|
||||
assert_object_bytes(client, &bucket, key, body)
|
||||
assert_workload_bytes(client, &bucket, key, body)
|
||||
.await
|
||||
.map_err(|error| format!("node {node_idx} failed to read {key}: {error}"))?;
|
||||
}
|
||||
|
||||
+525
-80
@@ -1677,8 +1677,230 @@ const MAX_GLOBAL_FILE_SYNCS: usize = 512;
|
||||
#[cfg(test)]
|
||||
const TEST_GLOBAL_FILE_SYNCS: usize = 64;
|
||||
|
||||
const PATH_LOCK_SHARDS: usize = 64;
|
||||
const PATH_LOCK_INITIAL_GC_THRESHOLD: usize = 64;
|
||||
|
||||
#[cfg(unix)]
|
||||
#[derive(Clone, Copy, Default)]
|
||||
struct Xxh3SliceHasher(u64);
|
||||
|
||||
#[cfg(unix)]
|
||||
impl std::hash::Hasher for Xxh3SliceHasher {
|
||||
#[inline]
|
||||
fn write(&mut self, bytes: &[u8]) {
|
||||
self.0 = xxhash_rust::xxh3::xxh3_64(bytes);
|
||||
}
|
||||
|
||||
#[inline]
|
||||
fn write_usize(&mut self, _i: usize) {}
|
||||
|
||||
#[inline]
|
||||
fn finish(&self) -> u64 {
|
||||
self.0
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
type Xxh3SliceBuildHasher = std::hash::BuildHasherDefault<Xxh3SliceHasher>;
|
||||
|
||||
#[cfg(unix)]
|
||||
#[inline]
|
||||
fn is_canonical_unix_key_bytes(bytes: &[u8]) -> bool {
|
||||
if bytes.starts_with(b"./") {
|
||||
return !bytes[1..].windows(2).any(|w| w == b"//") && !bytes[1..].windows(3).any(|w| w == b"/./");
|
||||
}
|
||||
const SLASH_MASK: u64 = 0x2f2f_2f2f_2f2f_2f2f;
|
||||
const LOW_BITS: u64 = 0x0101_0101_0101_0101;
|
||||
const HIGH_BITS: u64 = 0x8080_8080_8080_8080;
|
||||
|
||||
let mut i = 0usize;
|
||||
let len = bytes.len();
|
||||
let mut prev_s = 0u64;
|
||||
while i + 8 <= len {
|
||||
let mut buf = [0u8; 8];
|
||||
buf.copy_from_slice(&bytes[i..i + 8]);
|
||||
let w = u64::from_le_bytes(buf);
|
||||
let x = w ^ SLASH_MASK;
|
||||
let s = (x.wrapping_sub(LOW_BITS)) & !x & HIGH_BITS;
|
||||
let s_prev1 = (s << 8) | (prev_s >> 56);
|
||||
let s_prev2 = (s << 16) | (prev_s >> 48);
|
||||
if (s & (s_prev1 | s_prev2)) != 0 {
|
||||
let start = i.saturating_sub(2);
|
||||
let window = &bytes[start..i + 8];
|
||||
if window.windows(2).any(|pair| pair == b"//") || window.windows(3).any(|triple| triple == b"/./") {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
prev_s = s;
|
||||
i += 8;
|
||||
}
|
||||
let tail_start = i.saturating_sub(2);
|
||||
let tail = &bytes[tail_start..];
|
||||
!tail.windows(2).any(|pair| pair == b"//") && !tail.windows(3).any(|triple| triple == b"/./")
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[cold]
|
||||
fn canonical_unix_path_key_slow(path: &Path) -> std::borrow::Cow<'static, [u8]> {
|
||||
use std::os::unix::ffi::OsStrExt;
|
||||
let mut out = Vec::with_capacity(path.as_os_str().len());
|
||||
for comp in path.components() {
|
||||
match comp {
|
||||
Component::RootDir => {
|
||||
out.clear();
|
||||
out.push(b'/');
|
||||
}
|
||||
Component::CurDir => {
|
||||
if out.is_empty() {
|
||||
out.push(b'.');
|
||||
}
|
||||
}
|
||||
Component::ParentDir => {
|
||||
if out.len() > 1 || (out.len() == 1 && out[0] != b'/') {
|
||||
out.push(b'/');
|
||||
}
|
||||
out.extend_from_slice(b"..");
|
||||
}
|
||||
Component::Normal(part) => {
|
||||
if out.len() > 1 || (out.len() == 1 && out[0] != b'/') {
|
||||
out.push(b'/');
|
||||
}
|
||||
out.extend_from_slice(part.as_bytes());
|
||||
}
|
||||
Component::Prefix(_) => {}
|
||||
}
|
||||
}
|
||||
std::borrow::Cow::Owned(out)
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[inline]
|
||||
fn canonical_unix_path_key(path: &Path) -> std::borrow::Cow<'_, [u8]> {
|
||||
use std::os::unix::ffi::OsStrExt;
|
||||
let mut bytes = path.as_os_str().as_bytes();
|
||||
while bytes.len() > 1 {
|
||||
if bytes[bytes.len() - 1] == b'/' {
|
||||
bytes = &bytes[..bytes.len() - 1];
|
||||
} else if bytes.len() >= 2 && bytes[bytes.len() - 2] == b'/' && bytes[bytes.len() - 1] == b'.' {
|
||||
bytes = if bytes.len() == 2 { b"/" } else { &bytes[..bytes.len() - 2] };
|
||||
} else {
|
||||
break;
|
||||
}
|
||||
}
|
||||
if is_canonical_unix_key_bytes(bytes) {
|
||||
return std::borrow::Cow::Borrowed(bytes);
|
||||
}
|
||||
canonical_unix_path_key_slow(path)
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[inline]
|
||||
fn unix_canonical_key_starts_with(path_key: &[u8], root_key: &[u8]) -> bool {
|
||||
if root_key.is_empty() {
|
||||
return true;
|
||||
}
|
||||
if root_key == b"/" {
|
||||
return path_key.starts_with(b"/");
|
||||
}
|
||||
path_key.starts_with(root_key) && (path_key.len() == root_key.len() || path_key.get(root_key.len()) == Some(&b'/'))
|
||||
}
|
||||
|
||||
struct PathLockShard<L> {
|
||||
#[cfg(unix)]
|
||||
entries: HashMap<Box<[u8]>, Weak<L>, Xxh3SliceBuildHasher>,
|
||||
#[cfg(not(unix))]
|
||||
entries: HashMap<PathBuf, Weak<L>>,
|
||||
next_gc_threshold: usize,
|
||||
}
|
||||
|
||||
impl<L> Default for PathLockShard<L> {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
entries: HashMap::default(),
|
||||
next_gc_threshold: PATH_LOCK_INITIAL_GC_THRESHOLD,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
struct ShardedPathLockRegistry<L> {
|
||||
shards: [Mutex<PathLockShard<L>>; PATH_LOCK_SHARDS],
|
||||
}
|
||||
|
||||
impl<L> ShardedPathLockRegistry<L> {
|
||||
fn new() -> Self {
|
||||
Self {
|
||||
shards: std::array::from_fn(|_| Mutex::new(PathLockShard::default())),
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[inline]
|
||||
fn get_or_create_by_key(&self, key: &[u8], init: impl FnOnce() -> L) -> Arc<L> {
|
||||
let hash = xxhash_rust::xxh3::xxh3_64(key);
|
||||
let shard_idx = ((hash >> 32) as usize) & (PATH_LOCK_SHARDS - 1);
|
||||
let mut shard = self.shards[shard_idx].lock();
|
||||
if let Some(existing) = shard.entries.get(key).and_then(Weak::upgrade) {
|
||||
return existing;
|
||||
}
|
||||
let created = Arc::new(init());
|
||||
let weak = Arc::downgrade(&created);
|
||||
if let Some(slot) = shard.entries.get_mut(key) {
|
||||
*slot = weak;
|
||||
return created;
|
||||
}
|
||||
if shard.entries.len() >= shard.next_gc_threshold {
|
||||
shard.entries.retain(|_, entry| entry.strong_count() > 0);
|
||||
shard.next_gc_threshold = shard
|
||||
.entries
|
||||
.len()
|
||||
.saturating_add(1)
|
||||
.saturating_mul(2)
|
||||
.max(PATH_LOCK_INITIAL_GC_THRESHOLD);
|
||||
}
|
||||
shard.entries.insert(key.into(), weak);
|
||||
created
|
||||
}
|
||||
|
||||
#[inline]
|
||||
fn get_or_create(&self, path: &Path, init: impl FnOnce() -> L) -> Arc<L> {
|
||||
#[cfg(unix)]
|
||||
{
|
||||
let key = canonical_unix_path_key(path);
|
||||
self.get_or_create_by_key(key.as_ref(), init)
|
||||
}
|
||||
#[cfg(not(unix))]
|
||||
{
|
||||
use std::hash::{Hash, Hasher};
|
||||
let mut hasher = std::collections::hash_map::DefaultHasher::new();
|
||||
path.hash(&mut hasher);
|
||||
let shard_idx = ((hasher.finish() >> 32) as usize) & (PATH_LOCK_SHARDS - 1);
|
||||
let mut shard = self.shards[shard_idx].lock();
|
||||
if let Some(existing) = shard.entries.get(path).and_then(Weak::upgrade) {
|
||||
return existing;
|
||||
}
|
||||
let created = Arc::new(init());
|
||||
let weak = Arc::downgrade(&created);
|
||||
if let Some(slot) = shard.entries.get_mut(path) {
|
||||
*slot = weak;
|
||||
return created;
|
||||
}
|
||||
if shard.entries.len() >= shard.next_gc_threshold {
|
||||
shard.entries.retain(|_, entry| entry.strong_count() > 0);
|
||||
shard.next_gc_threshold = shard
|
||||
.entries
|
||||
.len()
|
||||
.saturating_add(1)
|
||||
.saturating_mul(2)
|
||||
.max(PATH_LOCK_INITIAL_GC_THRESHOLD);
|
||||
}
|
||||
shard.entries.insert(path.to_path_buf(), weak);
|
||||
created
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
static FILE_SYNC_PERMITS: LazyLock<Semaphore> = LazyLock::new(|| Semaphore::new(global_file_sync_limit()));
|
||||
static DISK_FILE_SYNC_LIMITERS: LazyLock<Mutex<HashMap<PathBuf, Weak<Semaphore>>>> = LazyLock::new(|| Mutex::new(HashMap::new()));
|
||||
static DISK_FILE_SYNC_LIMITERS: LazyLock<ShardedPathLockRegistry<Semaphore>> = LazyLock::new(ShardedPathLockRegistry::new);
|
||||
|
||||
/// Dedicated tokio runtime for fsync/fdatasync blocking operations. When
|
||||
/// configured with >1 threads, isolates device-bound fsync from the main
|
||||
@@ -1742,12 +1964,10 @@ fn fsync_spawn_blocking<T: Send + 'static>(f: impl FnOnce() -> T + Send + 'stati
|
||||
f()
|
||||
})
|
||||
}
|
||||
static DISK_VOLUME_MUTATION_LOCKS: LazyLock<Mutex<HashMap<PathBuf, Weak<RwLock<()>>>>> =
|
||||
LazyLock::new(|| Mutex::new(HashMap::new()));
|
||||
static DISK_VOLUME_MUTATION_LOCKS: LazyLock<ShardedPathLockRegistry<RwLock<()>>> = LazyLock::new(ShardedPathLockRegistry::new);
|
||||
type NamespaceMutationLock = AsyncMutex<()>;
|
||||
type NamespaceMutationLockRegistry = HashMap<PathBuf, Weak<NamespaceMutationLock>>;
|
||||
static DISK_NAMESPACE_MUTATION_LOCKS: LazyLock<Mutex<NamespaceMutationLockRegistry>> =
|
||||
LazyLock::new(|| Mutex::new(HashMap::new()));
|
||||
static DISK_NAMESPACE_MUTATION_LOCKS: LazyLock<ShardedPathLockRegistry<NamespaceMutationLock>> =
|
||||
LazyLock::new(ShardedPathLockRegistry::new);
|
||||
|
||||
fn default_global_file_sync_limit(cpu_count: usize, max_blocking_threads: usize) -> usize {
|
||||
let cpu_scaled = cpu_count
|
||||
@@ -1771,15 +1991,7 @@ fn global_file_sync_limit() -> usize {
|
||||
|
||||
/// Reuse a disk's limiter across reconnects while detached sync calls still hold it.
|
||||
pub(crate) fn disk_file_sync_limiter(root: &Path) -> Arc<Semaphore> {
|
||||
let mut limiters = DISK_FILE_SYNC_LIMITERS.lock();
|
||||
limiters.retain(|_, limiter| limiter.strong_count() > 0);
|
||||
if let Some(limiter) = limiters.get(root).and_then(Weak::upgrade) {
|
||||
return limiter;
|
||||
}
|
||||
|
||||
let limiter = Arc::new(Semaphore::new(MAX_PARALLEL_FILE_SYNCS));
|
||||
limiters.insert(root.to_path_buf(), Arc::downgrade(&limiter));
|
||||
limiter
|
||||
DISK_FILE_SYNC_LIMITERS.get_or_create(root, || Semaphore::new(MAX_PARALLEL_FILE_SYNCS))
|
||||
}
|
||||
|
||||
// Keep warm draining workers per disk/lane and spin briefly in user space
|
||||
@@ -2147,45 +2359,131 @@ pub(crate) fn enter_blocking_test_hook() -> BlockingTestHookGuard {
|
||||
BlockingTestHookGuard { queue }
|
||||
}
|
||||
|
||||
static DISK_BLOCKING_QUEUES: LazyLock<Mutex<HashMap<PathBuf, Weak<DiskBlockingQueue>>>> =
|
||||
LazyLock::new(|| Mutex::new(HashMap::new()));
|
||||
struct RegisteredDiskBlockingQueue {
|
||||
#[cfg(not(unix))]
|
||||
root: PathBuf,
|
||||
#[cfg(unix)]
|
||||
canonical_root: Box<[u8]>,
|
||||
queue: Weak<DiskBlockingQueue>,
|
||||
}
|
||||
|
||||
static DISK_BLOCKING_QUEUES: LazyLock<parking_lot::RwLock<Vec<RegisteredDiskBlockingQueue>>> =
|
||||
LazyLock::new(|| parking_lot::RwLock::new(Vec::new()));
|
||||
static FALLBACK_BLOCKING_QUEUES: LazyLock<[Arc<DiskBlockingQueue>; FALLBACK_NAMESPACE_QUEUE_SHARDS]> =
|
||||
LazyLock::new(|| std::array::from_fn(|_| DiskBlockingQueue::new()));
|
||||
|
||||
pub(crate) fn register_disk_blocking_queue(root: &Path, io_root: &Path) -> Arc<DiskBlockingQueue> {
|
||||
let mut queues = DISK_BLOCKING_QUEUES.lock();
|
||||
queues.retain(|_, q| q.strong_count() > 0);
|
||||
let queue = queues.get(root).and_then(Weak::upgrade).unwrap_or_else(|| {
|
||||
let queue = DiskBlockingQueue::new();
|
||||
queues.insert(root.to_path_buf(), Arc::downgrade(&queue));
|
||||
queue
|
||||
});
|
||||
if io_root != root {
|
||||
queues.insert(io_root.to_path_buf(), Arc::downgrade(&queue));
|
||||
#[cfg(unix)]
|
||||
let root_key = canonical_unix_path_key(root);
|
||||
#[cfg(unix)]
|
||||
let io_root_key = canonical_unix_path_key(io_root);
|
||||
|
||||
let mut queues = DISK_BLOCKING_QUEUES.write();
|
||||
queues.retain(|entry| entry.queue.strong_count() > 0);
|
||||
let queue = queues
|
||||
.iter()
|
||||
.find_map(|entry| {
|
||||
#[cfg(unix)]
|
||||
let matches = entry.canonical_root.as_ref() == root_key.as_ref();
|
||||
#[cfg(not(unix))]
|
||||
let matches = entry.root == root;
|
||||
matches.then(|| entry.queue.upgrade()).flatten()
|
||||
})
|
||||
.unwrap_or_else(|| {
|
||||
let queue = DiskBlockingQueue::new();
|
||||
queues.push(RegisteredDiskBlockingQueue {
|
||||
#[cfg(not(unix))]
|
||||
root: root.to_path_buf(),
|
||||
#[cfg(unix)]
|
||||
canonical_root: root_key.as_ref().into(),
|
||||
queue: Arc::downgrade(&queue),
|
||||
});
|
||||
queue
|
||||
});
|
||||
#[cfg(unix)]
|
||||
let io_matches_root = io_root_key.as_ref() == root_key.as_ref();
|
||||
#[cfg(not(unix))]
|
||||
let io_matches_root = io_root == root;
|
||||
if !io_matches_root {
|
||||
let weak = Arc::downgrade(&queue);
|
||||
if let Some(existing) = queues.iter_mut().find(|entry| {
|
||||
#[cfg(unix)]
|
||||
{
|
||||
entry.canonical_root.as_ref() == io_root_key.as_ref()
|
||||
}
|
||||
#[cfg(not(unix))]
|
||||
{
|
||||
entry.root == io_root
|
||||
}
|
||||
}) {
|
||||
existing.queue = weak;
|
||||
} else {
|
||||
queues.push(RegisteredDiskBlockingQueue {
|
||||
#[cfg(not(unix))]
|
||||
root: io_root.to_path_buf(),
|
||||
#[cfg(unix)]
|
||||
canonical_root: io_root_key.as_ref().into(),
|
||||
queue: weak,
|
||||
});
|
||||
}
|
||||
}
|
||||
queue
|
||||
}
|
||||
|
||||
fn disk_blocking_queue_for_root(root: &Path) -> Arc<DiskBlockingQueue> {
|
||||
register_disk_blocking_queue(root, root)
|
||||
}
|
||||
|
||||
fn disk_blocking_queue_for_path(path: &Path) -> Arc<DiskBlockingQueue> {
|
||||
#[cfg(unix)]
|
||||
let root_key = canonical_unix_path_key(root);
|
||||
{
|
||||
let queues = DISK_BLOCKING_QUEUES.lock();
|
||||
for (root, weak) in queues.iter() {
|
||||
if path.starts_with(root)
|
||||
&& let Some(queue) = weak.upgrade()
|
||||
{
|
||||
let queues = DISK_BLOCKING_QUEUES.read();
|
||||
for entry in queues.iter() {
|
||||
#[cfg(unix)]
|
||||
let matches = entry.canonical_root.as_ref() == root_key.as_ref();
|
||||
#[cfg(not(unix))]
|
||||
let matches = entry.root == root;
|
||||
if matches && let Some(queue) = entry.queue.upgrade() {
|
||||
return queue;
|
||||
}
|
||||
}
|
||||
}
|
||||
use std::hash::{Hash, Hasher};
|
||||
let mut hasher = std::collections::hash_map::DefaultHasher::new();
|
||||
path.hash(&mut hasher);
|
||||
let idx = usize::try_from(hasher.finish() % (FALLBACK_NAMESPACE_QUEUE_SHARDS as u64)).unwrap_or(0);
|
||||
Arc::clone(&FALLBACK_BLOCKING_QUEUES[idx])
|
||||
register_disk_blocking_queue(root, root)
|
||||
}
|
||||
|
||||
fn disk_blocking_queue_for_path(path: &Path) -> Arc<DiskBlockingQueue> {
|
||||
#[cfg(unix)]
|
||||
{
|
||||
let path_key = canonical_unix_path_key(path);
|
||||
{
|
||||
let queues = DISK_BLOCKING_QUEUES.read();
|
||||
for entry in queues.iter() {
|
||||
if unix_canonical_key_starts_with(path_key.as_ref(), entry.canonical_root.as_ref())
|
||||
&& let Some(queue) = entry.queue.upgrade()
|
||||
{
|
||||
return queue;
|
||||
}
|
||||
}
|
||||
}
|
||||
let hash = xxhash_rust::xxh3::xxh3_64(path_key.as_ref());
|
||||
let idx = usize::try_from(hash % (FALLBACK_NAMESPACE_QUEUE_SHARDS as u64)).unwrap_or(0);
|
||||
Arc::clone(&FALLBACK_BLOCKING_QUEUES[idx])
|
||||
}
|
||||
#[cfg(not(unix))]
|
||||
{
|
||||
{
|
||||
let queues = DISK_BLOCKING_QUEUES.read();
|
||||
for entry in queues.iter() {
|
||||
if path.starts_with(&entry.root)
|
||||
&& let Some(queue) = entry.queue.upgrade()
|
||||
{
|
||||
return queue;
|
||||
}
|
||||
}
|
||||
}
|
||||
use std::hash::{Hash, Hasher};
|
||||
let mut hasher = std::collections::hash_map::DefaultHasher::new();
|
||||
path.hash(&mut hasher);
|
||||
let idx = usize::try_from(hasher.finish() % (FALLBACK_NAMESPACE_QUEUE_SHARDS as u64)).unwrap_or(0);
|
||||
Arc::clone(&FALLBACK_BLOCKING_QUEUES[idx])
|
||||
}
|
||||
}
|
||||
|
||||
/// Serialize a bucket's local metadata commits with physical bucket removal.
|
||||
@@ -2194,32 +2492,46 @@ fn disk_blocking_queue_for_path(path: &Path) -> Arc<DiskBlockingQueue> {
|
||||
/// [`LocalDisk`](super::local::LocalDisk) instances share the same lock while
|
||||
/// disconnected disks do not keep the registry alive.
|
||||
pub(crate) fn disk_volume_mutation_lock(root: &Path, volume: &str) -> Arc<RwLock<()>> {
|
||||
let key = root.join(volume);
|
||||
let mut locks = DISK_VOLUME_MUTATION_LOCKS.lock();
|
||||
locks.retain(|_, lock| lock.strong_count() > 0);
|
||||
if let Some(lock) = locks.get(&key).and_then(Weak::upgrade) {
|
||||
return lock;
|
||||
#[cfg(unix)]
|
||||
{
|
||||
let vol_bytes = volume.as_bytes();
|
||||
if !vol_bytes.is_empty()
|
||||
&& vol_bytes[0] != b'/'
|
||||
&& vol_bytes != b"."
|
||||
&& !vol_bytes.starts_with(b"./")
|
||||
&& !vol_bytes.ends_with(b"/")
|
||||
&& !vol_bytes.ends_with(b"/.")
|
||||
&& is_canonical_unix_key_bytes(vol_bytes)
|
||||
&& let std::borrow::Cow::Borrowed(root_bytes) = canonical_unix_path_key(root)
|
||||
&& !root_bytes.is_empty()
|
||||
&& root_bytes != b"/"
|
||||
{
|
||||
let total_len = root_bytes.len() + 1 + vol_bytes.len();
|
||||
if total_len <= 512 {
|
||||
let mut stack_key = [0u8; 512];
|
||||
stack_key[..root_bytes.len()].copy_from_slice(root_bytes);
|
||||
stack_key[root_bytes.len()] = b'/';
|
||||
stack_key[root_bytes.len() + 1..total_len].copy_from_slice(vol_bytes);
|
||||
return DISK_VOLUME_MUTATION_LOCKS.get_or_create_by_key(&stack_key[..total_len], || RwLock::new(()));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let lock = Arc::new(RwLock::new(()));
|
||||
locks.insert(key, Arc::downgrade(&lock));
|
||||
lock
|
||||
let key = root.join(volume);
|
||||
DISK_VOLUME_MUTATION_LOCKS.get_or_create(&key, || RwLock::new(()))
|
||||
}
|
||||
|
||||
fn disk_namespace_mutation_lock(path: &Path) -> Arc<NamespaceMutationLock> {
|
||||
let mut locks = DISK_NAMESPACE_MUTATION_LOCKS.lock();
|
||||
locks.retain(|_, lock| lock.strong_count() > 0);
|
||||
if let Some(lock) = locks.get(path).and_then(Weak::upgrade) {
|
||||
return lock;
|
||||
}
|
||||
|
||||
let lock = Arc::new(AsyncMutex::new(()));
|
||||
locks.insert(path.to_path_buf(), Arc::downgrade(&lock));
|
||||
lock
|
||||
DISK_NAMESPACE_MUTATION_LOCKS.get_or_create(path, || AsyncMutex::new(()))
|
||||
}
|
||||
|
||||
static DISK_METADATA_MUTATION_LOCKS: LazyLock<Mutex<NamespaceMutationLockRegistry>> =
|
||||
LazyLock::new(|| Mutex::new(HashMap::new()));
|
||||
#[cfg(unix)]
|
||||
#[inline]
|
||||
fn disk_namespace_mutation_lock_by_key(key: &[u8]) -> Arc<NamespaceMutationLock> {
|
||||
DISK_NAMESPACE_MUTATION_LOCKS.get_or_create_by_key(key, || AsyncMutex::new(()))
|
||||
}
|
||||
|
||||
static DISK_METADATA_MUTATION_LOCKS: LazyLock<ShardedPathLockRegistry<NamespaceMutationLock>> =
|
||||
LazyLock::new(ShardedPathLockRegistry::new);
|
||||
|
||||
/// Serializes the complete xl.meta read/modify/commit transaction. This domain
|
||||
/// precedes namespace/volume publication locks, whose narrower syscall leases
|
||||
@@ -2233,17 +2545,7 @@ pub(crate) async fn acquire_metadata_mutation_lease(
|
||||
object: &Path,
|
||||
owner: Option<Arc<dyn Send + Sync>>,
|
||||
) -> Arc<MetadataMutationLease> {
|
||||
let lock = {
|
||||
let mut locks = DISK_METADATA_MUTATION_LOCKS.lock();
|
||||
locks.retain(|_, lock| lock.strong_count() > 0);
|
||||
if let Some(lock) = locks.get(object).and_then(Weak::upgrade) {
|
||||
lock
|
||||
} else {
|
||||
let lock = Arc::new(AsyncMutex::new(()));
|
||||
locks.insert(object.to_path_buf(), Arc::downgrade(&lock));
|
||||
lock
|
||||
}
|
||||
};
|
||||
let lock = DISK_METADATA_MUTATION_LOCKS.get_or_create(object, || AsyncMutex::new(()));
|
||||
Arc::new(MetadataMutationLease {
|
||||
_guard: lock.lock_owned().await,
|
||||
_owner: owner,
|
||||
@@ -2306,17 +2608,45 @@ pub(crate) async fn acquire_rename_data_mutation_lease_with_owner(
|
||||
) -> Arc<NamespaceMutationLease> {
|
||||
// Metadata ownership precedes namespace paths in lexical order, then volume
|
||||
// ownership. Deduplicate identical keys so TMP-to-TMP publication cannot self-lock.
|
||||
let destination_lock = disk_namespace_mutation_lock(destination_object);
|
||||
let (namespace_guard, staging_guard) = match staging_prefix {
|
||||
Some(source) if source < destination_object => {
|
||||
let staging = disk_namespace_mutation_lock(source).lock_owned().await;
|
||||
(destination_lock.lock_owned().await, Some(staging))
|
||||
#[cfg(unix)]
|
||||
let (namespace_guard, staging_guard) = {
|
||||
let dst_key = canonical_unix_path_key(destination_object);
|
||||
let destination_lock = disk_namespace_mutation_lock_by_key(dst_key.as_ref());
|
||||
match staging_prefix {
|
||||
Some(source) => {
|
||||
let src_key = canonical_unix_path_key(source);
|
||||
match src_key.as_ref().cmp(dst_key.as_ref()) {
|
||||
std::cmp::Ordering::Less => {
|
||||
let staging = disk_namespace_mutation_lock_by_key(src_key.as_ref()).lock_owned().await;
|
||||
(destination_lock.lock_owned().await, Some(staging))
|
||||
}
|
||||
std::cmp::Ordering::Greater => {
|
||||
let destination = destination_lock.lock_owned().await;
|
||||
(
|
||||
destination,
|
||||
Some(disk_namespace_mutation_lock_by_key(src_key.as_ref()).lock_owned().await),
|
||||
)
|
||||
}
|
||||
std::cmp::Ordering::Equal => (destination_lock.lock_owned().await, None),
|
||||
}
|
||||
}
|
||||
None => (destination_lock.lock_owned().await, None),
|
||||
}
|
||||
Some(source) if source > destination_object => {
|
||||
let destination = destination_lock.lock_owned().await;
|
||||
(destination, Some(disk_namespace_mutation_lock(source).lock_owned().await))
|
||||
};
|
||||
#[cfg(not(unix))]
|
||||
let (namespace_guard, staging_guard) = {
|
||||
let destination_lock = disk_namespace_mutation_lock(destination_object);
|
||||
match staging_prefix {
|
||||
Some(source) if source < destination_object => {
|
||||
let staging = disk_namespace_mutation_lock(source).lock_owned().await;
|
||||
(destination_lock.lock_owned().await, Some(staging))
|
||||
}
|
||||
Some(source) if source > destination_object => {
|
||||
let destination = destination_lock.lock_owned().await;
|
||||
(destination, Some(disk_namespace_mutation_lock(source).lock_owned().await))
|
||||
}
|
||||
_ => (destination_lock.lock_owned().await, None),
|
||||
}
|
||||
_ => (destination_lock.lock_owned().await, None),
|
||||
};
|
||||
let volume_guard = disk_volume_mutation_lock(root, volume).read_owned().await;
|
||||
let blocking_queue = disk_blocking_queue_for_root(root);
|
||||
@@ -11992,6 +12322,121 @@ mod tests {
|
||||
);
|
||||
assert!(preserved_final.exists(), "final file must remain when intermediate removal fails");
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[test]
|
||||
fn canonical_unix_path_key_preserves_std_path_equality_and_prefix_semantics() {
|
||||
let samples = [
|
||||
"",
|
||||
"/",
|
||||
"///",
|
||||
"/.",
|
||||
"/./.",
|
||||
".",
|
||||
"./",
|
||||
"./.",
|
||||
"./a",
|
||||
"././a/",
|
||||
".//a/.",
|
||||
"a",
|
||||
"a/",
|
||||
"a/.",
|
||||
"a/b/c",
|
||||
"a/b/c/",
|
||||
"a//b/./c/.",
|
||||
"/tmp/d0/bucket/obj",
|
||||
"/tmp/d0/bucket/obj/",
|
||||
"/tmp/d0//bucket/./obj/.",
|
||||
"/tmp/d0/.rustfs.sys/multipart/sha/upload",
|
||||
"/proc/self/fd/12/.",
|
||||
"/proc/self/fd/12",
|
||||
"/proc/self/fd/12/./bucket/obj",
|
||||
"/proc/self/fd/12/bucket/obj",
|
||||
"/tmp/d1",
|
||||
"/tmp/d10",
|
||||
"a/../b",
|
||||
"a/../b/",
|
||||
"/utf8/\u{00af}/segment/",
|
||||
];
|
||||
for s in samples {
|
||||
let p = Path::new(s);
|
||||
let fast = canonical_unix_path_key(p);
|
||||
let slow = canonical_unix_path_key_slow(p);
|
||||
assert_eq!(fast.as_ref(), slow.as_ref(), "fast and slow key mismatch for {s:?}");
|
||||
}
|
||||
for a in samples {
|
||||
for b in samples {
|
||||
let pa = Path::new(a);
|
||||
let pb = Path::new(b);
|
||||
let ka = canonical_unix_path_key(pa);
|
||||
let kb = canonical_unix_path_key(pb);
|
||||
assert_eq!(
|
||||
pa == pb,
|
||||
ka.as_ref() == kb.as_ref(),
|
||||
"Path::eq vs canonical key mismatch for {a:?} and {b:?}"
|
||||
);
|
||||
assert_eq!(
|
||||
pa.starts_with(pb),
|
||||
unix_canonical_key_starts_with(ka.as_ref(), kb.as_ref()),
|
||||
"Path::starts_with vs canonical key prefix mismatch for {a:?} and {b:?}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
let dir = tempdir().expect("tempdir");
|
||||
let p_clean = dir.path().join("bucket/object");
|
||||
let p_trailing = PathBuf::from(format!("{}/", p_clean.display()));
|
||||
let p_redundant = PathBuf::from(format!("{}//bucket/./object/.", dir.path().display()));
|
||||
let lock1 = disk_namespace_mutation_lock(&p_clean);
|
||||
let lock2 = disk_namespace_mutation_lock(&p_trailing);
|
||||
let lock3 = disk_namespace_mutation_lock(&p_redundant);
|
||||
assert!(Arc::ptr_eq(&lock1, &lock2));
|
||||
assert!(Arc::ptr_eq(&lock1, &lock3));
|
||||
|
||||
let vlock1 = disk_volume_mutation_lock(dir.path(), "bucket");
|
||||
let vlock2 = disk_volume_mutation_lock(&PathBuf::from(format!("{}/.", dir.path().display())), "bucket/");
|
||||
assert!(Arc::ptr_eq(&vlock1, &vlock2));
|
||||
|
||||
let io_root = dir.path().join("io_root/.");
|
||||
let q_reg = register_disk_blocking_queue(dir.path(), &io_root);
|
||||
let q_root = disk_blocking_queue_for_root(dir.path());
|
||||
let q_path = disk_blocking_queue_for_path(&dir.path().join("io_root/bucket/key"));
|
||||
assert!(Arc::ptr_eq(&q_reg, &q_root));
|
||||
assert!(Arc::ptr_eq(&q_reg, &q_path));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn sharded_path_lock_registry_reuses_slots_and_bounds_dead_weak_entries() {
|
||||
let registry: ShardedPathLockRegistry<AsyncMutex<()>> = ShardedPathLockRegistry::new();
|
||||
let target = Path::new("/tmp/sharded-lock-test/persistent-key");
|
||||
let held = registry.get_or_create(target, || AsyncMutex::new(()));
|
||||
let held_again = registry.get_or_create(target, || AsyncMutex::new(()));
|
||||
assert!(Arc::ptr_eq(&held, &held_again), "live lock must be shared");
|
||||
|
||||
// Re-acquiring a key after drop must reuse the existing slot in-place.
|
||||
let transient = Path::new("/tmp/sharded-lock-test/reused-slot");
|
||||
drop(registry.get_or_create(transient, || AsyncMutex::new(())));
|
||||
let reused = registry.get_or_create(transient, || AsyncMutex::new(()));
|
||||
drop(reused);
|
||||
|
||||
// Flood transient keys across shards and verify `held` is never evicted
|
||||
// and total dead entries remain bounded by the amortized GC threshold.
|
||||
for i in 0..(PATH_LOCK_SHARDS * PATH_LOCK_INITIAL_GC_THRESHOLD * 3) {
|
||||
let p = PathBuf::from(format!("/tmp/sharded-lock-test/ephemeral-{i}"));
|
||||
drop(registry.get_or_create(&p, || AsyncMutex::new(())));
|
||||
}
|
||||
let still_held = registry.get_or_create(target, || AsyncMutex::new(()));
|
||||
assert!(Arc::ptr_eq(&held, &still_held), "GC must never evict a live lock");
|
||||
for shard in ®istry.shards {
|
||||
let guard = shard.lock();
|
||||
assert!(
|
||||
guard.entries.len() <= guard.next_gc_threshold,
|
||||
"shard entries ({}) exceeded GC threshold ({})",
|
||||
guard.entries.len(),
|
||||
guard.next_gc_threshold
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(all(test, target_os = "linux"))]
|
||||
|
||||
@@ -33,10 +33,10 @@ zstd-safe = { version = "8.2.0", features = ["seekable"] }
|
||||
toml = "1.1.8"
|
||||
semver = "1.0.28"
|
||||
apache-avro.workspace = true
|
||||
arrow-array = "59.3.2"
|
||||
arrow-ipc = { version = "59.3.2", features = ["zstd"] }
|
||||
arrow-schema = "59.3.2"
|
||||
parquet = { version = "59.3.2", default-features = false, features = ["arrow", "zstd"] }
|
||||
arrow-array = "60.1.0"
|
||||
arrow-ipc = { version = "60.1.0", features = ["zstd"] }
|
||||
arrow-schema = "60.1.0"
|
||||
parquet = { version = "60.1.0", default-features = false, features = ["arrow", "zstd"] }
|
||||
kafka-protocol = { version = "0.18.0", default-features = false, features = ["zstd"] }
|
||||
bytes.workspace = true
|
||||
|
||||
|
||||
@@ -64,7 +64,7 @@ RustFS S3 Tables is an Iceberg REST Catalog and table-bucket implementation on t
|
||||
| Namespaces | Supported | Create, list, load, exists, drop on both prefixes; `pageSize`/`pageToken` pagination with context-bound tokens; identifiers limited to 512 ASCII characters. |
|
||||
| Tables | Supported | Direct and staged create, register, list, load, exists, rename, commit, metadata-location get/update, and drop on both prefixes. Staged create is stateless until an `assert-create` commit atomically publishes the initial metadata and table entry. Rename uses a bucket-scoped persistent fence, recoverable intent, and conditional publication; the source name is reusable only via an ETag-conditional tombstone replacement. Commit identifiers must match the URL; unknown requirements/updates fail as bad requests; register overwrite, purge-on-drop, and v3-only encryption-key updates return an explicit unsupported-operation response. |
|
||||
| Commit CAS | Supported | Single-table commits validate base metadata, version token, referenced object existence, warehouse scope, and Iceberg requirements before advancing the pointer; external metadata transitions preserve monotonic assignment watermarks and immutable retained definitions. `idempotency-key-lifetime` is not advertised; mutation-wide `Idempotency-Key` semantics are unsupported. |
|
||||
| Commit recovery | Supported | Commit log, idempotency lookup, diagnostics, and recovery routes expose and repair finalization gaps without moving the pointer. |
|
||||
| Commit recovery | Supported | Commit log, idempotency lookup, diagnostics, and recovery routes expose and repair finalization gaps without moving the pointer. Object-backed recovery reports a staged request as SUPERSEDED only when a reachable finalized same-base winner proves it lost CAS and its idempotency binding is intact. The original staged receipt is retained; its old request still conflicts. The winning history must be retained with the losing receipt; missing or contradictory history requires manual review. |
|
||||
| Snapshot refs | Supported | List, create/replace, delete via commits; `main` is protected; refs with explicit retention need forced delete. |
|
||||
| Iceberg views | Supported | Create, list, load, replace, exists, drop with view-scoped authorization; only view format version 1. |
|
||||
| LoadTable and table credentials endpoint | Supported | Vending only on negotiated `vended-credentials`; one temporary session scoped to the warehouse prefix and current metadata location; missing credential permission falls back to metadata-only with an explicit reason; responses carry `Cache-Control: no-store, private`. |
|
||||
|
||||
@@ -14,7 +14,7 @@ Pick the lowest layer that can prove the change; add a higher-layer test only wh
|
||||
| Outbound target matrix | Replication of every object shape (empty, plain, retention, legal hold, multipart, checksummed, SSE-S3) against every remote-target failure mode the fake target models; an explicit expectation table pins known-red cells to an open issue | `python3 scripts/e2e_binary.py build`, then `python3 scripts/e2e_binary.py run -- cargo nextest run -p e2e_test -E 'test(/^replication_target_matrix_test::/)'` | With `e2e-repl-nightly`; required locally for any change to outbound client defaults (SOP: [`docs/postmortems/2026-09-03-replication-checksum-default-regression.md`](../postmortems/2026-09-03-replication-checksum-default-regression.md)) |
|
||||
| s3s-e2e conformance | External S3 conformance tool against a live server | `./scripts/e2e-run.sh ./target/debug/rustfs <data-dir>` | Non-documentation PRs, required through `Test and Lint` (second half of the `End-to-End Tests` job) |
|
||||
| S3 compatibility | `ceph/s3-tests` (boto3; allow-list `scripts/s3-tests/implemented_tests.txt`) and MinIO `mint` | `scripts/s3-tests/run.sh`; mint via `.github/workflows/mint.yml` | Non-documentation PRs: implemented and lifecycle S3 lanes required through `Test and Lint`; the separate s3-tests workflow provides additional PR reporting and a weekly full sweep; mint: weekly, report-only |
|
||||
| Chaos / fault-injection | Single-node disk fault injection (`crates/e2e_test/src/chaos.rs`, `crates/e2e_test/src/fault_proxy.rs`) plus the 4-node kill/fresh-drive/blackhole cases in `crates/e2e_test/src/distributed/chaos_test.rs` | Part of the e2e crate (`e2e-reliability` and `e2e-distributed`) | Reliability cases with `e2e-full`; 4-node chaos on storage-sensitive PRs and nightly via `e2e-distributed` |
|
||||
| Chaos / fault-injection | Single-node disk fault injection (`crates/e2e_test/src/chaos.rs`, `crates/e2e_test/src/fault_proxy.rs`) plus the 4-node kill/fresh-drive/blackhole cases in `crates/e2e_test/src/distributed/chaos_test.rs` | Part of the e2e crate (`e2e-reliability` and `e2e-distributed`) | Reliability cases with `e2e-full`; 4-node chaos nightly via `e2e-distributed` |
|
||||
| Fuzz | `cargo-fuzz` targets over untrusted parsing surfaces; isolated sub-workspace under `fuzz/` | `./scripts/fuzz/run.sh` (see [`fuzz/README.md`](../../fuzz/README.md)) | PR smoke on the paths listed in `.github/workflows/fuzz.yml`, plus nightly corpus |
|
||||
| Benchmarks | Criterion benches under each crate's `benches/` | `cargo bench -p <crate>` | On demand; never a gate |
|
||||
| Conditional reads | Paired ordinary/hit/miss GET latency, throughput and read-lock measurements | `scripts/run_conditional_read_abba.py` | Local conditional-read acceptance; [guide and scope](conditional-read-benchmark.md) |
|
||||
@@ -88,7 +88,7 @@ All profiles are defined in `.config/nextest.toml`; its block comments hold the
|
||||
| `e2e-full` | Merge-queue / main-push single-node e2e lane |
|
||||
| `e2e-repl-nightly` | Nightly slow / cross-process replication lane |
|
||||
| `e2e-nightly` | Nightly serial multi-process cluster fault lane |
|
||||
| `e2e-distributed` | Storage-sensitive PR and nightly 4-node 4-disk S3 / lock / versioning / replication / quota / expand / decommission / rebalance / site-replication / chaos / upgrade (history + IAM AK/SK) lane |
|
||||
| `e2e-distributed` | Nightly 4-node 4-disk S3 / lock / versioning / replication / quota / expand / decommission / rebalance / site-replication / chaos / upgrade (history + IAM AK/SK) lane |
|
||||
| `e2e-protocols` | Nightly fixed-port FTPS/SFTP/WebDAV lane, run with `-j 1` |
|
||||
|
||||
Membership of each e2e profile is pinned by a digest in `.config/e2e-<profile>-selection.txt` and checked by `scripts/check_test_wiring.py --check-profile <profile>` before the lane runs. To list what a profile selects on your platform (the result is platform-dependent because some modules are linux-only):
|
||||
|
||||
@@ -35,7 +35,7 @@ gh api repos/rustfs/rustfs/rulesets/6436880 \
|
||||
--jq '.rules[] | select(.type == "required_status_checks") | .parameters'
|
||||
```
|
||||
|
||||
The aggregate requires the validation lanes already selected by `ci.yml`; this closes the gap where a failing critical lane left the required workspace check green. Independent workflows remain report-only unless separately required. Before adding a new expensive lane or moving existing PR coverage to a schedule, collect representative execution and regression evidence, establish ownership and a working scheduled replacement, and update this reference with the resulting policy.
|
||||
The aggregate requires the validation lanes already selected by `ci.yml`; this closes the gap where a failing critical lane left the required workspace check green. Independent workflows remain report-only unless separately required. Before adding a new expensive lane or moving existing PR coverage to a schedule, collect representative execution and regression evidence, establish ownership and a working scheduled replacement, and update this reference with the resulting policy. `e2e-distributed.yml` and `e2e-upgrade.yml` are that kind of replacement: the 4-node lane runs on its nightly schedule, and the single-node upgrade lane runs on its weekly Monday schedule, on version tags, and on manual dispatch. Neither is part of `Test and Lint`. Dispatch the workflow against a candidate branch when that signal is needed before merge.
|
||||
|
||||
Superseded PR attempts are cancelled. A running main validation finishes, with only the newest pending main run retained, so frequent merges cannot continuously cancel the full baseline. Having a `merge_group` trigger does not itself require use of the merge queue.
|
||||
|
||||
@@ -66,7 +66,6 @@ The four site-replication state-writer concurrency proofs reserve the available
|
||||
| PR touching `paths` in `fuzz.yml` | `Build Fuzz Harness`, `Smoke / <target>` | `fuzz.yml` `fuzz-build`, `pr-fuzz-smoke` | Report-only | `MAX_TOTAL_TIME=60 ./scripts/fuzz/run.sh` |
|
||||
| PR touching `paths` in `windows-filesystem.yml` | `Rename Safety`, `Startup Smoke` | `windows-filesystem.yml` `rename-safety`, `startup-smoke` | Report-only | the `cargo test -p rustfs-ecstore --lib <filter>` commands in the job, on Windows; `cargo build --locked -p rustfs --bin rustfs`, then `rustfs.exe --version` and a single-volume server polled on `/health/ready` |
|
||||
| PR touching `paths` in `coverage.yml` | `Workspace line coverage` | `coverage.yml` `coverage` | Report-only | `make coverage`; `python3 scripts/check_security_coverage.py target/llvm-cov/coverage.json` |
|
||||
| PR touching `paths` in `e2e-upgrade.yml` | `Upgrade compatibility` | `e2e-upgrade.yml` `upgrade` | Report-only | `python3 scripts/e2e_binary.py build`, then the job's `python3 scripts/e2e_binary.py run -- cargo test --locked -p e2e_test` commands with `RUSTFS_UPGRADE_SOURCE_BINARY` pointing at the pinned previous release (`UPGRADE_SOURCE_VERSION`) |
|
||||
| PR touching `paths` in `oidc-keycloak.yml` | `OIDC Keycloak live gate` | `oidc-keycloak.yml` `oidc-keycloak-live` | Report-only | `cargo build --locked -p rustfs --bin rustfs`, then `bash scripts/test/oidc_keycloak_live.sh ./target/debug/rustfs` |
|
||||
| PR touching `paths` in `targets-integration.yml` | `PostgreSQL, MySQL, AMQP, and NATS` | `targets-integration.yml` `targets-live` | Report-only | start the containers as in the job, export the `RUSTFS_TEST_*` DSNs, then the job's `cargo test --locked -p rustfs-targets --test <name> -- --ignored --test-threads=1` commands |
|
||||
| PR, documentation-only selection | `Quick Checks`, `Typos`, `Test and Lint` | `ci.yml` `quick-checks`, `typos`, `required-checks` | Required directly or via aggregate | Quick Checks commands; `python3 scripts/ci_gate.py --self-test` |
|
||||
@@ -90,7 +89,7 @@ Scheduled lanes never block a PR. Their workflow-local gate fails the run. The v
|
||||
| `cache-warm.yml` (every six hours, plus main pushes that change `Cargo.lock`, `rust-toolchain.toml` or the setup action) | `warm-ci-dev`, `warm-ci-feat-rio`, `warm-ci-feat-proto`, `warm-ci-uring` | no verdict; sole writer of the `rust-cache` keys that `ci.yml` restores | yes | dispatch `cache-warm.yml` |
|
||||
| `build.yml` (nightly) | `build-rustfs` over the six-target platform matrix in `prepare-platform-matrix` (four Linux, macOS aarch64, Windows x86_64) | build/package integrity | yes | dispatch `build.yml` with an exact platform set |
|
||||
| `e2e-replication-nightly.yml` (nightly) | `repl-nightly`, `cluster-nightly`, `protocols-nightly` | three independent gates; JUnit, membership listing, server logs | yes | `python3 scripts/e2e_binary.py build --bins`, then `python3 scripts/e2e_binary.py run -- cargo nextest run --profile e2e-repl-nightly -p e2e_test`; `--features e2e-test-hooks` on both steps with `--profile e2e-nightly`; `--features ftps,webdav,sftp` on both steps with `-j 1 --profile e2e-protocols` |
|
||||
| `e2e-distributed.yml` (storage-sensitive PRs + nightly) | `distributed` | fail-closed 4-node 4-disk S3, durability, replication, movement, fault, and direct/rolling upgrade gate; JUnit, membership listing, per-node server logs | yes, with `never_ran_grace_until` | download the pinned previous release as in the workflow, export `RUSTFS_UPGRADE_SOURCE_BINARY`, run `python3 scripts/e2e_binary.py build --bins --features e2e-test-hooks`, then `python3 scripts/e2e_binary.py run --features e2e-test-hooks -- cargo nextest run --profile e2e-distributed -p e2e_test` |
|
||||
| `e2e-distributed.yml` (nightly) | `distributed` | fail-closed 4-node 4-disk S3, durability, replication, movement, fault, and direct/rolling upgrade gate; JUnit, membership listing, per-node server logs | yes | download the pinned previous release as in the workflow, export `RUSTFS_UPGRADE_SOURCE_BINARY`, run `python3 scripts/e2e_binary.py build --bins --features e2e-test-hooks`, then `python3 scripts/e2e_binary.py run --features e2e-test-hooks -- cargo nextest run --profile e2e-distributed -p e2e_test` |
|
||||
| `e2e-s3tests.yml` (weekly) | `s3tests` (single and distributed, four shards each), `upstream-head-canary` | compatibility gate; report, JUnit, node IDs, server logs | yes | `scripts/s3-tests/run.sh` against an existing single or distributed target |
|
||||
| `fuzz.yml` (nightly) | `nightly-fuzz-corpus` per target | gate; corpus and crash artifacts | yes (disabled) | `MAX_TOTAL_TIME=<seconds> ./scripts/fuzz/run.sh` |
|
||||
| `minio-interop.yml` (nightly) | `minio-interop` | EC + SSE read-parity gate | yes, with `never_ran_grace_until` | pinned Docker fixture steps in the workflow |
|
||||
@@ -100,7 +99,7 @@ Scheduled lanes never block a PR. Their workflow-local gate fails the run. The v
|
||||
| `mint.yml` (weekly) | `mint` | report-only by design; per-suite PASS/FAIL/NA and raw `log.json` | yes | pinned Docker sequence in the workflow |
|
||||
| `coverage.yml` (weekly) | `coverage` | report-only trend; lcov and JSON artifact | yes (disabled) | `make coverage` |
|
||||
| `runner-hygiene.yml` (monthly) | `check-ephemerality` | runner ephemerality | yes | dispatch |
|
||||
| `e2e-upgrade.yml` (weekly) | `upgrade` (serial cases sharing one harness build) | upgrade and rollback gate; server logs | yes | see the PR row |
|
||||
| `e2e-upgrade.yml` (weekly, and on version tags) | `upgrade` (serial cases sharing one harness build) | upgrade and rollback gate; server logs | yes | `python3 scripts/e2e_binary.py build`, then the job's `python3 scripts/e2e_binary.py run -- cargo test --locked -p e2e_test` commands with `RUSTFS_UPGRADE_SOURCE_BINARY` pointing at the pinned previous release (`UPGRADE_SOURCE_VERSION`) |
|
||||
| `oidc-keycloak.yml` (weekly) | `oidc-keycloak-live` | live OIDC gate | yes | see the PR row |
|
||||
| `targets-integration.yml` (nightly) | `targets-live` | live target gate; container logs | yes (disabled) | see the PR row |
|
||||
| `gateway-integration-sync.yml` (weekdays) | `sync` | merges `main` into `gateway/integration` as a `--no-ff` merge commit; a conflicting merge is aborted and opens a `sync/<source>-<date>` pull request listing the conflicting files, which keeps the run green; a rejected push or missing pull request fails it | yes | dispatch, optionally with `source_ref` |
|
||||
|
||||
@@ -44,45 +44,58 @@ pub(crate) fn table_matches_staged_base(table: &TableEntry, commit_log: &CommitL
|
||||
pub(crate) struct TableCommitHistoryIndex<'a> {
|
||||
table_id: &'a str,
|
||||
reachable_states: BTreeSet<(&'a str, &'a str)>,
|
||||
reachable_version_tokens: BTreeSet<&'a str>,
|
||||
committed_successors: BTreeMap<(&'a str, &'a str), &'a CommitLogEntry>,
|
||||
ambiguous_states: BTreeSet<(&'a str, &'a str)>,
|
||||
cycle_detected: bool,
|
||||
version_token_reused: bool,
|
||||
}
|
||||
|
||||
impl<'a> TableCommitHistoryIndex<'a> {
|
||||
pub(crate) fn new(table: &'a TableEntry, commits: impl IntoIterator<Item = &'a CommitLogEntry>) -> Self {
|
||||
let mut by_new_state = BTreeMap::<(&str, &str), Option<(&str, &str)>>::new();
|
||||
let mut by_new_state = BTreeMap::<(&str, &str), Option<&CommitLogEntry>>::new();
|
||||
let mut ambiguous_states = BTreeSet::new();
|
||||
for commit in commits
|
||||
.into_iter()
|
||||
.filter(|commit| commit.table_id == table.table_id && !matches!(commit.status, CommitLogStatus::Failed))
|
||||
{
|
||||
let key = (commit.new_metadata_location.as_str(), commit.new_version_token.as_str());
|
||||
let previous = (commit.previous_metadata_location.as_str(), commit.expected_version_token.as_str());
|
||||
by_new_state
|
||||
.entry(key)
|
||||
.and_modify(|candidate| {
|
||||
*candidate = None;
|
||||
ambiguous_states.insert(key);
|
||||
})
|
||||
.or_insert(Some(previous));
|
||||
.or_insert(Some(commit));
|
||||
}
|
||||
|
||||
let mut reachable_states = BTreeSet::new();
|
||||
let mut reachable_version_tokens = BTreeSet::new();
|
||||
let mut committed_successors = BTreeMap::new();
|
||||
let mut version_token_reused = false;
|
||||
let mut state = (table.metadata_location.as_str(), table.version_token.as_str());
|
||||
let cycle_detected = loop {
|
||||
if !reachable_states.insert(state) {
|
||||
break true;
|
||||
}
|
||||
let Some(Some(previous)) = by_new_state.get(&state) else {
|
||||
version_token_reused |= !reachable_version_tokens.insert(state.1);
|
||||
let Some(commit) = by_new_state.get(&state).copied().flatten() else {
|
||||
break false;
|
||||
};
|
||||
state = *previous;
|
||||
let previous = (commit.previous_metadata_location.as_str(), commit.expected_version_token.as_str());
|
||||
if matches!(commit.status, CommitLogStatus::Committed) {
|
||||
committed_successors.insert(previous, commit);
|
||||
}
|
||||
state = previous;
|
||||
};
|
||||
Self {
|
||||
table_id: &table.table_id,
|
||||
reachable_states,
|
||||
reachable_version_tokens,
|
||||
committed_successors,
|
||||
ambiguous_states,
|
||||
cycle_detected,
|
||||
version_token_reused,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -97,6 +110,49 @@ impl<'a> TableCommitHistoryIndex<'a> {
|
||||
.reachable_states
|
||||
.contains(&(target.new_metadata_location.as_str(), target.new_version_token.as_str()))
|
||||
}
|
||||
|
||||
fn proven_superseding_commit(&self, target: &CommitLogEntry) -> Option<&CommitLogEntry> {
|
||||
if self.cycle_detected
|
||||
|| self.version_token_reused
|
||||
|| !self.ambiguous_states.is_empty()
|
||||
|| self.table_id != target.table_id
|
||||
|| !matches!(target.status, CommitLogStatus::Staged)
|
||||
|| self.reachable_version_tokens.contains(target.new_version_token.as_str())
|
||||
{
|
||||
return None;
|
||||
}
|
||||
let base = (target.previous_metadata_location.as_str(), target.expected_version_token.as_str());
|
||||
self.committed_successors
|
||||
.get(&base)
|
||||
.copied()
|
||||
.filter(|winner| winner.commit_id != target.commit_id)
|
||||
}
|
||||
}
|
||||
|
||||
/// Classifies an object-backed staged loser without rewriting its receipt.
|
||||
///
|
||||
/// A reachable finalized successor from the same pointer/token base proves that
|
||||
/// the old request lost CAS. Its STAGED record and idempotency binding remain
|
||||
/// available, and replay of that old request must still conflict. Missing or
|
||||
/// contradictory proof remains a manual-review state.
|
||||
pub(crate) fn table_commit_recovery_entry_from_history(
|
||||
table: &TableEntry,
|
||||
commit_log: &CommitLogEntry,
|
||||
idempotency_commit: Option<&CommitLogEntry>,
|
||||
history: &TableCommitHistoryIndex<'_>,
|
||||
) -> TableCommitRecoveryEntry {
|
||||
let mut entry = table_commit_recovery_entry(table, commit_log, idempotency_commit, history.proves_committed(commit_log));
|
||||
if entry.recovery_state == TableCommitRecoveryState::ManualReview
|
||||
&& matches!(
|
||||
entry.idempotency_index_status,
|
||||
TableCommitIdempotencyIndexStatus::NotRequired | TableCommitIdempotencyIndexStatus::Matches
|
||||
)
|
||||
&& let Some(winner) = history.proven_superseding_commit(commit_log)
|
||||
{
|
||||
entry.recovery_state = TableCommitRecoveryState::Superseded;
|
||||
entry.reason = format!("staged request lost CAS to reachable committed successor {}", winner.commit_id);
|
||||
}
|
||||
entry
|
||||
}
|
||||
|
||||
pub(crate) fn table_catalog_recovery_summary(
|
||||
|
||||
@@ -1415,6 +1415,7 @@ pub(crate) struct TableCatalogDiagnosticsReport {
|
||||
#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
|
||||
pub(crate) enum TableCommitRecoveryState {
|
||||
Committed,
|
||||
Superseded,
|
||||
StagedBeforeTableUpdate,
|
||||
FinalizationRequired,
|
||||
IdempotencyIndexRepairRequired,
|
||||
|
||||
@@ -3065,11 +3065,11 @@ where
|
||||
}
|
||||
None => None,
|
||||
};
|
||||
commits.push(table_commit_recovery_entry(
|
||||
commits.push(table_commit_recovery_entry_from_history(
|
||||
entry,
|
||||
commit_log,
|
||||
idempotency_commit.as_ref(),
|
||||
history.proves_committed(commit_log),
|
||||
&history,
|
||||
));
|
||||
}
|
||||
commits.sort_by(|left, right| left.commit_id.cmp(&right.commit_id));
|
||||
@@ -3168,12 +3168,8 @@ where
|
||||
}
|
||||
None => None,
|
||||
};
|
||||
let recovery_entry = table_commit_recovery_entry(
|
||||
&entry,
|
||||
commit_log,
|
||||
idempotency_commit.as_ref(),
|
||||
history.proves_committed(commit_log),
|
||||
);
|
||||
let recovery_entry =
|
||||
table_commit_recovery_entry_from_history(&entry, commit_log, idempotency_commit.as_ref(), &history);
|
||||
if matches!(
|
||||
recovery_entry.recovery_state,
|
||||
TableCommitRecoveryState::FinalizationRequired | TableCommitRecoveryState::IdempotencyIndexRepairRequired
|
||||
|
||||
@@ -20,6 +20,7 @@ use std::sync::Arc;
|
||||
|
||||
mod strong_maintenance_planning;
|
||||
mod strong_maintenance_planning_paths;
|
||||
mod superseded_commit_recovery;
|
||||
|
||||
const TABLE_CATALOG_TEST_TIMEOUT: StdDuration = StdDuration::from_secs(30);
|
||||
|
||||
|
||||
@@ -0,0 +1,487 @@
|
||||
// Copyright 2026 RustFS Team
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use super::*;
|
||||
|
||||
struct StagedLoserFixture {
|
||||
backend: TestCatalogObjectBackend,
|
||||
store: ObjectTableCatalogStore<TestCatalogObjectBackend>,
|
||||
loser: TableCommitRequest,
|
||||
winner: TableCommitRequest,
|
||||
committed: TableCommitResult,
|
||||
}
|
||||
|
||||
async fn staged_loser_fixture(interrupt_winner_finalization: bool) -> StagedLoserFixture {
|
||||
let backend = TestCatalogObjectBackend::default();
|
||||
let store = ObjectTableCatalogStore::new(backend.clone());
|
||||
let bucket = "analytics";
|
||||
let namespace = Namespace::parse("sales").unwrap();
|
||||
let table = IdentifierSegment::parse("orders").unwrap();
|
||||
let initial = test_table_entry(
|
||||
bucket,
|
||||
&namespace,
|
||||
&table,
|
||||
default_table_metadata_file_path(&namespace, &table, "00001.metadata.json"),
|
||||
);
|
||||
store.put_table_bucket(test_bucket_entry(bucket)).await.unwrap();
|
||||
store
|
||||
.create_namespace(test_namespace_entry(bucket, &namespace))
|
||||
.await
|
||||
.unwrap();
|
||||
store.create_table(initial.clone()).await.unwrap();
|
||||
for name in ["00001", "loser", "winner", "reloaded"] {
|
||||
let location = default_table_metadata_file_path(&namespace, &table, &format!("{name}.metadata.json"));
|
||||
let mut metadata = super::super::test_support::table_metadata_json(&initial.table_uuid, &initial.warehouse_location);
|
||||
metadata["properties"] = serde_json::json!({"writer": name});
|
||||
backend
|
||||
.seed_object(bucket, &location, serde_json::to_vec(&metadata).unwrap())
|
||||
.await;
|
||||
}
|
||||
let paths = TableCatalogObjectPaths::default();
|
||||
let table_path = paths.table_entry_path(bucket, &namespace, &table);
|
||||
backend.fail_next_put(RUSTFS_META_BUCKET, &table_path).await;
|
||||
let loser = TableCommitRequest {
|
||||
table_bucket: bucket.to_string(),
|
||||
namespace: namespace.public_name(),
|
||||
table: table.as_str().to_string(),
|
||||
commit_id: "paused-writer-a0".to_string(),
|
||||
idempotency_key: Some("a0-request".to_string()),
|
||||
operation: "set-properties".to_string(),
|
||||
expected_version_token: initial.version_token.clone(),
|
||||
expected_metadata_location: initial.metadata_location.clone(),
|
||||
new_metadata_location: default_table_metadata_file_path(&namespace, &table, "loser.metadata.json"),
|
||||
requirements: Vec::new(),
|
||||
writer: None,
|
||||
};
|
||||
assert_matches!(store.commit_table(loser.clone()).await, Err(TableCatalogStoreError::Internal(_)));
|
||||
let winner = TableCommitRequest {
|
||||
commit_id: "winner-c1".to_string(),
|
||||
idempotency_key: Some("c1-request".to_string()),
|
||||
new_metadata_location: default_table_metadata_file_path(&namespace, &table, "winner.metadata.json"),
|
||||
..loser.clone()
|
||||
};
|
||||
if interrupt_winner_finalization {
|
||||
let winner_log = paths.commit_log_entry_path(bucket, &initial.table_id, &winner.commit_id);
|
||||
backend.fail_put_attempt(RUSTFS_META_BUCKET, &winner_log, 2).await;
|
||||
}
|
||||
let committed = store.commit_table(winner.clone()).await.unwrap();
|
||||
StagedLoserFixture {
|
||||
backend,
|
||||
store,
|
||||
loser,
|
||||
winner,
|
||||
committed,
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn superseded_staged_commit_recovers_without_rewriting_receipts_or_pointer() {
|
||||
let StagedLoserFixture {
|
||||
backend,
|
||||
store,
|
||||
loser,
|
||||
winner,
|
||||
committed,
|
||||
} = staged_loser_fixture(false).await;
|
||||
let bucket = loser.table_bucket.as_str();
|
||||
let before = backend.state.lock().await.objects.clone();
|
||||
|
||||
let recovered = store.recover_table_commits(bucket, "sales", "orders").await.unwrap();
|
||||
assert_eq!(recovered.manual_review_count, 0, "a durable same-base winner proves the staged loser");
|
||||
assert_eq!(recovered.finalized_count, 0, "a losing writer must never be finalized as committed");
|
||||
let lost = recovered
|
||||
.commits
|
||||
.iter()
|
||||
.find(|entry| entry.commit_id == loser.commit_id)
|
||||
.unwrap();
|
||||
assert_eq!(lost.status, CommitLogStatus::Staged);
|
||||
assert_eq!(serde_json::to_value(&lost.recovery_state).unwrap(), "SUPERSEDED");
|
||||
assert_eq!(lost.idempotency_index_status, TableCommitIdempotencyIndexStatus::Matches);
|
||||
let after = backend.state.lock().await.objects.clone();
|
||||
assert_eq!(after.len(), before.len());
|
||||
for (path, original) in &before {
|
||||
let actual = after.get(path).unwrap();
|
||||
assert_eq!(actual.data, original.data, "recovery changed bytes at {path:?}");
|
||||
assert_eq!(actual.etag, original.etag, "recovery rewrote {path:?}");
|
||||
}
|
||||
|
||||
let restarted = ObjectTableCatalogStore::new(backend.clone());
|
||||
let diagnostics = restarted.diagnose_table_catalog(bucket, "sales", "orders", 0).await.unwrap();
|
||||
assert_eq!(diagnostics.recovery_status, TableCatalogRecoveryStatus::Healthy);
|
||||
assert_matches!(restarted.commit_table(loser.clone()).await, Err(TableCatalogStoreError::Conflict(_)));
|
||||
let mut stale_request = loser.clone();
|
||||
stale_request.commit_id = "stale-new-request".to_string();
|
||||
stale_request.idempotency_key = Some("stale-new-key".to_string());
|
||||
assert_matches!(restarted.commit_table(stale_request).await, Err(TableCatalogStoreError::Conflict(_)));
|
||||
let mut different_payload = loser.clone();
|
||||
different_payload.operation = "replace".to_string();
|
||||
assert_matches!(
|
||||
restarted.commit_table(different_payload.clone()).await,
|
||||
Err(TableCatalogStoreError::Conflict(_))
|
||||
);
|
||||
assert_eq!(
|
||||
restarted.load_table(bucket, "sales", "orders").await.unwrap(),
|
||||
Some(committed.table.clone())
|
||||
);
|
||||
assert_eq!(restarted.commit_table(winner).await.unwrap().table, committed.table);
|
||||
|
||||
let reloaded = TableCommitRequest {
|
||||
commit_id: "reloaded-writer-a1".to_string(),
|
||||
idempotency_key: Some("a1-request".to_string()),
|
||||
expected_version_token: committed.table.version_token.clone(),
|
||||
expected_metadata_location: committed.table.metadata_location.clone(),
|
||||
new_metadata_location: default_table_metadata_file_path(
|
||||
&Namespace::parse("sales").unwrap(),
|
||||
&IdentifierSegment::parse("orders").unwrap(),
|
||||
"reloaded.metadata.json",
|
||||
),
|
||||
..different_payload
|
||||
};
|
||||
let next = restarted.commit_table(reloaded.clone()).await.unwrap();
|
||||
assert_eq!(next.table.generation, committed.table.generation + 1);
|
||||
assert_eq!(restarted.commit_table(reloaded).await.unwrap().table, next.table);
|
||||
let restarted = ObjectTableCatalogStore::new(backend);
|
||||
assert_eq!(
|
||||
restarted
|
||||
.diagnose_table_catalog(bucket, "sales", "orders", 0)
|
||||
.await
|
||||
.unwrap()
|
||||
.recovery_status,
|
||||
TableCatalogRecoveryStatus::Healthy
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn superseded_staged_commit_waits_for_durable_winner_finalization() {
|
||||
let StagedLoserFixture {
|
||||
backend,
|
||||
store,
|
||||
loser,
|
||||
committed,
|
||||
..
|
||||
} = staged_loser_fixture(true).await;
|
||||
let paths = TableCatalogObjectPaths::default();
|
||||
let log_path = paths.commit_log_entry_path(&loser.table_bucket, &committed.table.table_id, &loser.commit_id);
|
||||
let before = backend.read_object(RUSTFS_META_BUCKET, &log_path).await.unwrap().unwrap();
|
||||
let planned = store
|
||||
.plan_table_commit_recovery("analytics", "sales", "orders")
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(planned.manual_review_count, 1, "a merely staged winner is insufficient proof");
|
||||
assert_eq!(planned.finalization_required_count, 1);
|
||||
|
||||
let recovered = store.recover_table_commits("analytics", "sales", "orders").await.unwrap();
|
||||
assert_eq!(recovered.finalized_count, 1);
|
||||
assert_eq!(recovered.manual_review_count, 0);
|
||||
let loser_state = recovered
|
||||
.commits
|
||||
.iter()
|
||||
.find(|entry| entry.commit_id == loser.commit_id)
|
||||
.unwrap();
|
||||
assert_eq!(loser_state.recovery_state, TableCommitRecoveryState::Superseded);
|
||||
assert_eq!(loser_state.status, CommitLogStatus::Staged);
|
||||
let after = backend.read_object(RUSTFS_META_BUCKET, &log_path).await.unwrap().unwrap();
|
||||
assert_eq!(after.data, before.data);
|
||||
assert_eq!(after.etag, before.etag);
|
||||
assert_eq!(store.load_table("analytics", "sales", "orders").await.unwrap(), Some(committed.table));
|
||||
assert_eq!(
|
||||
store
|
||||
.diagnose_table_catalog("analytics", "sales", "orders", 0)
|
||||
.await
|
||||
.unwrap()
|
||||
.recovery_status,
|
||||
TableCatalogRecoveryStatus::Healthy
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn superseded_staged_commit_requires_retained_winner_proof_after_restart() {
|
||||
let StagedLoserFixture {
|
||||
backend,
|
||||
store,
|
||||
loser,
|
||||
winner,
|
||||
committed,
|
||||
} = staged_loser_fixture(false).await;
|
||||
let healthy = store.recover_table_commits("analytics", "sales", "orders").await.unwrap();
|
||||
assert_eq!(healthy.manual_review_count, 0);
|
||||
let winner_path =
|
||||
TableCatalogObjectPaths::default().commit_log_entry_path("analytics", &committed.table.table_id, &winner.commit_id);
|
||||
assert!(
|
||||
backend
|
||||
.state
|
||||
.lock()
|
||||
.await
|
||||
.objects
|
||||
.remove(&(RUSTFS_META_BUCKET.to_string(), winner_path))
|
||||
.is_some()
|
||||
);
|
||||
let restarted = ObjectTableCatalogStore::new(backend);
|
||||
let before = restarted.load_table("analytics", "sales", "orders").await.unwrap();
|
||||
let report = restarted.recover_table_commits("analytics", "sales", "orders").await.unwrap();
|
||||
assert_eq!(report.manual_review_count, 1);
|
||||
assert_eq!(report.finalized_count, 0);
|
||||
assert_eq!(report.commits[0].recovery_state, TableCommitRecoveryState::ManualReview);
|
||||
assert_eq!(report.commits[0].status, CommitLogStatus::Staged);
|
||||
assert_matches!(restarted.commit_table(loser).await, Err(TableCatalogStoreError::Conflict(_)));
|
||||
assert_eq!(restarted.load_table("analytics", "sales", "orders").await.unwrap(), before);
|
||||
let diagnostics = restarted
|
||||
.diagnose_table_catalog("analytics", "sales", "orders", 0)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(diagnostics.recovery_status, TableCatalogRecoveryStatus::ManualReviewRequired);
|
||||
assert!(
|
||||
diagnostics
|
||||
.recommended_actions
|
||||
.contains(&TableCatalogRecoveryAction::ReviewCommitLog)
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn superseded_staged_commit_does_not_repair_a_conflicting_index() {
|
||||
let StagedLoserFixture {
|
||||
backend,
|
||||
store,
|
||||
loser,
|
||||
committed,
|
||||
..
|
||||
} = staged_loser_fixture(false).await;
|
||||
let mut conflicting = store
|
||||
.get_commit_by_id("analytics", &committed.table.table_id, &loser.commit_id)
|
||||
.await
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
conflicting.new_version_token = "contradictory-token".to_string();
|
||||
let path = TableCatalogObjectPaths::default().commit_idempotency_entry_path(
|
||||
"analytics",
|
||||
&committed.table.table_id,
|
||||
loser.idempotency_key.as_deref().unwrap(),
|
||||
);
|
||||
backend
|
||||
.seed_object(RUSTFS_META_BUCKET, &path, serde_json::to_vec(&conflicting).unwrap())
|
||||
.await;
|
||||
let before = backend.state.lock().await.objects.clone();
|
||||
let report = store.recover_table_commits("analytics", "sales", "orders").await.unwrap();
|
||||
assert_eq!(report.manual_review_count, 1);
|
||||
assert_eq!(report.finalized_count, 0);
|
||||
let loser_state = report
|
||||
.commits
|
||||
.iter()
|
||||
.find(|entry| entry.commit_id == loser.commit_id)
|
||||
.unwrap();
|
||||
assert_eq!(loser_state.recovery_state, TableCommitRecoveryState::ManualReview);
|
||||
assert_eq!(loser_state.idempotency_index_status, TableCommitIdempotencyIndexStatus::Conflicting);
|
||||
let after = backend.state.lock().await.objects.clone();
|
||||
assert_eq!(before.len(), after.len());
|
||||
for (key, original) in &before {
|
||||
let actual = after.get(key).unwrap();
|
||||
assert_eq!(actual.data, original.data);
|
||||
assert_eq!(actual.etag, original.etag);
|
||||
}
|
||||
assert_matches!(store.commit_table(loser).await, Err(TableCatalogStoreError::Conflict(_)));
|
||||
}
|
||||
|
||||
fn history_fixture() -> (TableEntry, CommitLogEntry, CommitLogEntry) {
|
||||
let namespace = Namespace::parse("sales").unwrap();
|
||||
let table_name = IdentifierSegment::parse("orders").unwrap();
|
||||
let metadata = |name| default_table_metadata_file_path(&namespace, &table_name, name);
|
||||
let mut table = test_table_entry("analytics", &namespace, &table_name, metadata("winner.metadata.json"));
|
||||
table.version_token = "winner-token".to_string();
|
||||
table.generation = 2;
|
||||
let loser = CommitLogEntry {
|
||||
version: TABLE_CATALOG_ENTRY_VERSION,
|
||||
commit_id: "loser".to_string(),
|
||||
idempotency_key: None,
|
||||
table_id: table.table_id.clone(),
|
||||
operation: "set-properties".to_string(),
|
||||
expected_version_token: "base-token".to_string(),
|
||||
new_version_token: "loser-token".to_string(),
|
||||
previous_metadata_location: metadata("base.metadata.json"),
|
||||
new_metadata_location: metadata("loser.metadata.json"),
|
||||
requirements: Vec::new(),
|
||||
status: CommitLogStatus::Staged,
|
||||
writer: None,
|
||||
created_at: None,
|
||||
updated_at: None,
|
||||
};
|
||||
let winner = CommitLogEntry {
|
||||
commit_id: "winner".to_string(),
|
||||
new_version_token: table.version_token.clone(),
|
||||
new_metadata_location: table.metadata_location.clone(),
|
||||
status: CommitLogStatus::Committed,
|
||||
..loser.clone()
|
||||
};
|
||||
(table, loser, winner)
|
||||
}
|
||||
|
||||
fn classify_history(
|
||||
table: &TableEntry,
|
||||
target: &CommitLogEntry,
|
||||
commits: &[CommitLogEntry],
|
||||
indexed: Option<&CommitLogEntry>,
|
||||
) -> TableCommitRecoveryEntry {
|
||||
let history = TableCommitHistoryIndex::new(table, commits.iter());
|
||||
table_commit_recovery_entry_from_history(table, target, indexed, &history)
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn superseded_history_requires_a_reachable_finalized_same_base_winner() {
|
||||
for case in [
|
||||
"staged-winner",
|
||||
"failed-winner",
|
||||
"wrong-base-token",
|
||||
"wrong-base-location",
|
||||
"foreign-table",
|
||||
"disconnected-winner",
|
||||
"same-commit-id",
|
||||
] {
|
||||
let (mut table, loser, mut winner) = history_fixture();
|
||||
match case {
|
||||
"staged-winner" => winner.status = CommitLogStatus::Staged,
|
||||
"failed-winner" => winner.status = CommitLogStatus::Failed,
|
||||
"wrong-base-token" => winner.expected_version_token = "other-base-token".to_string(),
|
||||
"wrong-base-location" => winner.previous_metadata_location = "other/base.metadata.json".to_string(),
|
||||
"foreign-table" => winner.table_id = "other-table-id".to_string(),
|
||||
"disconnected-winner" => {
|
||||
table.metadata_location = "other/current.metadata.json".to_string();
|
||||
table.version_token = "other-current-token".to_string();
|
||||
}
|
||||
"same-commit-id" => winner.commit_id.clone_from(&loser.commit_id),
|
||||
_ => unreachable!(),
|
||||
}
|
||||
let recovery = classify_history(&table, &loser, &[loser.clone(), winner], None);
|
||||
assert_eq!(recovery.recovery_state, TableCommitRecoveryState::ManualReview, "{case}");
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn superseded_history_is_stable_after_later_commits() {
|
||||
let (mut table, loser, winner) = history_fixture();
|
||||
let immediate = classify_history(&table, &loser, &[loser.clone(), winner.clone()], None);
|
||||
assert_eq!(immediate.recovery_state, TableCommitRecoveryState::Superseded);
|
||||
assert!(immediate.reason.contains(&winner.commit_id));
|
||||
let later = CommitLogEntry {
|
||||
commit_id: "later".to_string(),
|
||||
previous_metadata_location: table.metadata_location.clone(),
|
||||
expected_version_token: table.version_token.clone(),
|
||||
new_metadata_location: "later/current.metadata.json".to_string(),
|
||||
new_version_token: "later-token".to_string(),
|
||||
..winner.clone()
|
||||
};
|
||||
table.metadata_location.clone_from(&later.new_metadata_location);
|
||||
table.version_token.clone_from(&later.new_version_token);
|
||||
table.generation += 1;
|
||||
let after = classify_history(&table, &loser, &[loser.clone(), winner, later], None);
|
||||
assert_eq!(after.recovery_state, TableCommitRecoveryState::Superseded);
|
||||
assert_eq!(after.reason, immediate.reason);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn superseded_history_rejects_failed_or_committed_losers() {
|
||||
for status in [CommitLogStatus::Failed, CommitLogStatus::Committed] {
|
||||
let (table, mut loser, winner) = history_fixture();
|
||||
loser.status = status;
|
||||
let recovery = classify_history(&table, &loser, &[loser.clone(), winner], None);
|
||||
assert_eq!(recovery.recovery_state, TableCommitRecoveryState::ManualReview);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn superseded_history_rejects_ambiguity_cycles_and_reused_tokens() {
|
||||
for case in [
|
||||
"ambiguous-current",
|
||||
"cycle",
|
||||
"reused-history-token",
|
||||
"loser-token-in-history",
|
||||
"off-history-ambiguity",
|
||||
] {
|
||||
let (mut table, mut loser, winner) = history_fixture();
|
||||
let mut extra = CommitLogEntry {
|
||||
commit_id: "extra".to_string(),
|
||||
..winner.clone()
|
||||
};
|
||||
let mut commits = vec![loser.clone(), winner.clone()];
|
||||
match case {
|
||||
"ambiguous-current" => commits.push(extra),
|
||||
"cycle" => {
|
||||
extra.previous_metadata_location.clone_from(&winner.new_metadata_location);
|
||||
extra.expected_version_token.clone_from(&winner.new_version_token);
|
||||
extra.new_metadata_location.clone_from(&winner.previous_metadata_location);
|
||||
extra.new_version_token.clone_from(&winner.expected_version_token);
|
||||
commits.push(extra);
|
||||
}
|
||||
"reused-history-token" => {
|
||||
extra.previous_metadata_location.clone_from(&winner.new_metadata_location);
|
||||
extra.expected_version_token.clone_from(&winner.new_version_token);
|
||||
extra.new_metadata_location = "later/current.metadata.json".to_string();
|
||||
extra.new_version_token.clone_from(&winner.expected_version_token);
|
||||
table.metadata_location.clone_from(&extra.new_metadata_location);
|
||||
table.version_token.clone_from(&extra.new_version_token);
|
||||
commits.push(extra);
|
||||
}
|
||||
"loser-token-in-history" => {
|
||||
loser.new_version_token.clone_from(&winner.expected_version_token);
|
||||
commits[0] = loser.clone();
|
||||
}
|
||||
"off-history-ambiguity" => {
|
||||
extra.new_metadata_location = "unknown/metadata.json".to_string();
|
||||
extra.new_version_token = "unknown-token".to_string();
|
||||
commits.push(extra.clone());
|
||||
extra.commit_id = "another-extra".to_string();
|
||||
commits.push(extra);
|
||||
}
|
||||
_ => unreachable!(),
|
||||
}
|
||||
assert_eq!(
|
||||
classify_history(&table, &loser, &commits, None).recovery_state,
|
||||
TableCommitRecoveryState::ManualReview,
|
||||
"{case}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn superseded_history_requires_an_intact_idempotency_binding() {
|
||||
let (table, mut loser, winner) = history_fixture();
|
||||
loser.idempotency_key = Some("loser-request".to_string());
|
||||
let commits = [loser.clone(), winner];
|
||||
assert_eq!(
|
||||
classify_history(&table, &loser, &commits, Some(&loser)).recovery_state,
|
||||
TableCommitRecoveryState::Superseded
|
||||
);
|
||||
let missing = classify_history(&table, &loser, &commits, None);
|
||||
assert_eq!(missing.recovery_state, TableCommitRecoveryState::ManualReview);
|
||||
assert_eq!(missing.idempotency_index_status, TableCommitIdempotencyIndexStatus::Missing);
|
||||
let mut stale = loser.clone();
|
||||
stale.status = CommitLogStatus::Committed;
|
||||
let state = classify_history(&table, &loser, &commits, Some(&stale));
|
||||
assert_eq!(state.recovery_state, TableCommitRecoveryState::ManualReview);
|
||||
assert_eq!(state.idempotency_index_status, TableCommitIdempotencyIndexStatus::Stale);
|
||||
let mut conflicting = loser.clone();
|
||||
conflicting.commit_id = "another-request".to_string();
|
||||
let state = classify_history(&table, &loser, &commits, Some(&conflicting));
|
||||
assert_eq!(state.recovery_state, TableCommitRecoveryState::ManualReview);
|
||||
assert_eq!(state.idempotency_index_status, TableCommitIdempotencyIndexStatus::Conflicting);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn superseded_history_never_relabels_a_committed_staged_ancestor() {
|
||||
let (table, mut ancestor, winner) = history_fixture();
|
||||
ancestor.new_metadata_location.clone_from(&winner.previous_metadata_location);
|
||||
ancestor.new_version_token.clone_from(&winner.expected_version_token);
|
||||
ancestor.previous_metadata_location = "older/base.metadata.json".to_string();
|
||||
ancestor.expected_version_token = "older-base-token".to_string();
|
||||
let recovery = classify_history(&table, &ancestor, &[ancestor.clone(), winner], None);
|
||||
assert_eq!(recovery.recovery_state, TableCommitRecoveryState::FinalizationRequired);
|
||||
}
|
||||
Reference in New Issue
Block a user