diff --git a/.config/e2e-distributed-selection.txt b/.config/e2e-distributed-selection.txt index bf308ff5a..39204eef8 100644 --- a/.config/e2e-distributed-selection.txt +++ b/.config/e2e-distributed-selection.txt @@ -1,2 +1,2 @@ -sha256-linux=686dcc075e41a75dde3d37e5bd0393d016469f02f83aa3508f92bfc17f26a4d4 -sha256-darwin=686dcc075e41a75dde3d37e5bd0393d016469f02f83aa3508f92bfc17f26a4d4 +sha256-linux=2f97cbbeb2dbf52919b704a3d07336c162a41f95079d941d74251f9279c494e6 +sha256-darwin=2f97cbbeb2dbf52919b704a3d07336c162a41f95079d941d74251f9279c494e6 diff --git a/.config/nextest.toml b/.config/nextest.toml index 6c083520e..0574fb73c 100644 --- a/.config/nextest.toml +++ b/.config/nextest.toml @@ -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. diff --git a/.github/dependabot.yml b/.github/dependabot.yml index 24c57eba9..7827ad1f3 100644 --- a/.github/dependabot.yml +++ b/.github/dependabot.yml @@ -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" diff --git a/.github/scheduled-validations.json b/.github/scheduled-validations.json index ffc959b82..891daba94 100644 --- a/.github/scheduled-validations.json +++ b/.github/scheduled-validations.json @@ -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" diff --git a/.github/workflows/e2e-distributed.yml b/.github/workflows/e2e-distributed.yml index 7b955c30e..47806f998 100644 --- a/.github/workflows/e2e-distributed.yml +++ b/.github/workflows/e2e-distributed.yml @@ -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: diff --git a/.github/workflows/e2e-upgrade.yml b/.github/workflows/e2e-upgrade.yml index 215e16cba..92e8c2163 100644 --- a/.github/workflows/e2e-upgrade.yml +++ b/.github/workflows/e2e-upgrade.yml @@ -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 diff --git a/Cargo.lock b/Cargo.lock index badc6a684..8b4273870 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -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", diff --git a/Cargo.toml b/Cargo.toml index 2e67147c5..512ab5747 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -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 diff --git a/crates/e2e_test/README.md b/crates/e2e_test/README.md index 8eef80ee2..ab49ffba5 100644 --- a/crates/e2e_test/README.md +++ b/crates/e2e_test/README.md @@ -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) | diff --git a/crates/e2e_test/src/distributed/concurrency_stability_test.rs b/crates/e2e_test/src/distributed/concurrency_stability_test.rs index 06a24e9a4..c32d8b51d 100644 --- a/crates/e2e_test/src/distributed/concurrency_stability_test.rs +++ b/crates/e2e_test/src/distributed/concurrency_stability_test.rs @@ -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(operation: &str, mut call: F) -> TestResult>> +where + E: Debug + ProvideErrorMetadata + 'static, + F: FnMut() -> Fut, + Fut: Future>>, +{ + 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::() == TypeId::of::()); + 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) { + 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!("{code}") + }; + 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::>()); 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}"))?; } diff --git a/crates/ecstore/src/disk/os.rs b/crates/ecstore/src/disk/os.rs index 2b6839e38..15d03a1a9 100644 --- a/crates/ecstore/src/disk/os.rs +++ b/crates/ecstore/src/disk/os.rs @@ -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; + +#[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 { + #[cfg(unix)] + entries: HashMap, Weak, Xxh3SliceBuildHasher>, + #[cfg(not(unix))] + entries: HashMap>, + next_gc_threshold: usize, +} + +impl Default for PathLockShard { + fn default() -> Self { + Self { + entries: HashMap::default(), + next_gc_threshold: PATH_LOCK_INITIAL_GC_THRESHOLD, + } + } +} + +struct ShardedPathLockRegistry { + shards: [Mutex>; PATH_LOCK_SHARDS], +} + +impl ShardedPathLockRegistry { + 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 { + 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 { + #[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 = LazyLock::new(|| Semaphore::new(global_file_sync_limit())); -static DISK_FILE_SYNC_LIMITERS: LazyLock>>> = LazyLock::new(|| Mutex::new(HashMap::new())); +static DISK_FILE_SYNC_LIMITERS: LazyLock> = 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(f: impl FnOnce() -> T + Send + 'stati f() }) } -static DISK_VOLUME_MUTATION_LOCKS: LazyLock>>>> = - LazyLock::new(|| Mutex::new(HashMap::new())); +static DISK_VOLUME_MUTATION_LOCKS: LazyLock>> = LazyLock::new(ShardedPathLockRegistry::new); type NamespaceMutationLock = AsyncMutex<()>; -type NamespaceMutationLockRegistry = HashMap>; -static DISK_NAMESPACE_MUTATION_LOCKS: LazyLock> = - LazyLock::new(|| Mutex::new(HashMap::new())); +static DISK_NAMESPACE_MUTATION_LOCKS: LazyLock> = + 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 { - 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>>> = - LazyLock::new(|| Mutex::new(HashMap::new())); +struct RegisteredDiskBlockingQueue { + #[cfg(not(unix))] + root: PathBuf, + #[cfg(unix)] + canonical_root: Box<[u8]>, + queue: Weak, +} + +static DISK_BLOCKING_QUEUES: LazyLock>> = + LazyLock::new(|| parking_lot::RwLock::new(Vec::new())); static FALLBACK_BLOCKING_QUEUES: LazyLock<[Arc; 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 { - 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 { - register_disk_blocking_queue(root, root) -} - -fn disk_blocking_queue_for_path(path: &Path) -> Arc { + #[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 { + #[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 { /// [`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> { - 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 { - 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> = - LazyLock::new(|| Mutex::new(HashMap::new())); +#[cfg(unix)] +#[inline] +fn disk_namespace_mutation_lock_by_key(key: &[u8]) -> Arc { + DISK_NAMESPACE_MUTATION_LOCKS.get_or_create_by_key(key, || AsyncMutex::new(())) +} + +static DISK_METADATA_MUTATION_LOCKS: LazyLock> = + 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 { - 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 { // 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> = 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"))] diff --git a/crates/zstd-compat/Cargo.toml b/crates/zstd-compat/Cargo.toml index b0f63dbf9..9244aff11 100644 --- a/crates/zstd-compat/Cargo.toml +++ b/crates/zstd-compat/Cargo.toml @@ -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 diff --git a/docs/architecture/s3-tables-support-matrix.md b/docs/architecture/s3-tables-support-matrix.md index 40b1a4a21..a2c18d6fd 100644 --- a/docs/architecture/s3-tables-support-matrix.md +++ b/docs/architecture/s3-tables-support-matrix.md @@ -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`. | diff --git a/docs/testing/README.md b/docs/testing/README.md index a47adaab1..2a2027bd8 100644 --- a/docs/testing/README.md +++ b/docs/testing/README.md @@ -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 ` | 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 ` | 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--selection.txt` and checked by `scripts/check_test_wiring.py --check-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): diff --git a/docs/testing/ci-gates.md b/docs/testing/ci-gates.md index e66481f3d..caa67fc21 100644 --- a/docs/testing/ci-gates.md +++ b/docs/testing/ci-gates.md @@ -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 / ` | `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 ` 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 -- --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= ./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/-` 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` | diff --git a/rustfs/src/table_catalog/iceberg/commit.rs b/rustfs/src/table_catalog/iceberg/commit.rs index 19ee14ca3..fe99dac9c 100644 --- a/rustfs/src/table_catalog/iceberg/commit.rs +++ b/rustfs/src/table_catalog/iceberg/commit.rs @@ -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) -> 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( diff --git a/rustfs/src/table_catalog/model.rs b/rustfs/src/table_catalog/model.rs index 5d98d9d38..f02abefd8 100644 --- a/rustfs/src/table_catalog/model.rs +++ b/rustfs/src/table_catalog/model.rs @@ -1415,6 +1415,7 @@ pub(crate) struct TableCatalogDiagnosticsReport { #[serde(rename_all = "SCREAMING_SNAKE_CASE")] pub(crate) enum TableCommitRecoveryState { Committed, + Superseded, StagedBeforeTableUpdate, FinalizationRequired, IdempotencyIndexRepairRequired, diff --git a/rustfs/src/table_catalog/store/object.rs b/rustfs/src/table_catalog/store/object.rs index 09c4ee9da..6a9e3a12b 100644 --- a/rustfs/src/table_catalog/store/object.rs +++ b/rustfs/src/table_catalog/store/object.rs @@ -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 diff --git a/rustfs/src/table_catalog/tests.rs b/rustfs/src/table_catalog/tests.rs index c6dd204c7..3d2b7c6b9 100644 --- a/rustfs/src/table_catalog/tests.rs +++ b/rustfs/src/table_catalog/tests.rs @@ -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); diff --git a/rustfs/src/table_catalog/tests/superseded_commit_recovery.rs b/rustfs/src/table_catalog/tests/superseded_commit_recovery.rs new file mode 100644 index 000000000..5ffbf6f9b --- /dev/null +++ b/rustfs/src/table_catalog/tests/superseded_commit_recovery.rs @@ -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, + 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); +}