Compare commits

...

39 Commits

Author SHA1 Message Date
Chris 1e7065101d feat: Capture offline CPU profiles in serving process (#8245)
feat: capture offline CPU profiles in serving process
2026-09-29 22:37:10 +08:00
Chris 380e98a42c fix(test): synchronize top-disk fixture writes with sampling (#8243) 2026-09-29 21:07:45 +08:00
Chris a88225d208 fix(ci): reduce duplicate work and preserve reliable test failures (#8233)
* fix(ci): reduce duplicate work and preserve reliable test failures

* fix(ci): retain protocol evidence and repair stale test fixtures

* test(connect): honor parent deadline during API fixture readiness

* fix(ci): reserve IO capacity for state writer proofs

* test(connect): align RPC fixtures with service capture contracts

* test(connect): cover pinned service capture failures
2026-09-29 21:06:45 +08:00
Chris 1cca0dd25c feat(connect): select offline key for client performance (#8242) 2026-09-29 20:49:59 +08:00
Chris b0a54f0e5c feat(connect): capture offline network performance in server (#8240)
* feat(connect): capture offline network performance in server

* fix(connect): use storage facade in RPC test

* fix(connect): keep RPC test body behind storage facade
2026-09-29 19:41:43 +08:00
Chris 83a2d8b02d feat(connect): capture RPC activity in offline server runtime (#8237) 2026-09-29 18:16:54 +08:00
Chris f661651aad feat(connect): capture API activity in offline server runtime (#8234) 2026-09-29 17:18:26 +08:00
Chris 9298991d14 fix(scanner): restart a finished mixed sweep under its requested plan (#8231)
A bucket sweep that ends mixed clears its position and records the finishing cycle's plan as started. The next cycle requests a new plan whenever the bucket was written in between, so a fresh sweep inherited a stale started plan and ended mixed again. On a continuously written bucket no sweep could ever certify and the census kept the old root.

Start a fresh verification sweep under the requested plan when no durable position remains. Resumed sweeps keep their started plan, so a clean tail still cannot certify an old prefix.

Refs #7108
2026-09-29 16:48:58 +08:00
Chris 3c7d85420c fix(test): control clocks in Connect network fixtures (#8228) 2026-09-29 16:48:34 +08:00
Chris 6a7880c9fc feat(connect): capture offline lock metrics in server (#8230) 2026-09-29 16:29:07 +08:00
hector 0e58bd09d0 fix(ci): run security chain evidence from the lane's own checkout (#8229)
The security workflow checks the repository out into rustfs-repo/ (#7212)
but its chain evidence steps still invoke
scripts/functional_chain_evidence.py relative to the workspace root, which
on the persistent shared runner resolves to a stale checkout left by another
job. The evidence gate then compares that checkout's HEAD against the chain
workflow_sha and rejects the lane before any case runs
("lane checkout differs from chain workflow source").

Run the evidence script from the lane's own checkout so ROOT resolves to
rustfs-repo/, whose HEAD is exactly the chain-pinned workflow_sha.
2026-09-29 16:09:31 +08:00
Hauser fcb60e7322 fix(rpc): preserve walk_dir missing-path errors (#8226)
## Related Issues

rustfs/rustfs#8217 and rustfs/backlog#2683.

## Summary of Changes

Reviewed 539ed275db against d2175d1e1e. No findings. Capable callers requesting missing-path reporting recover explicit pre-output FileNotFound or VolumeNotFound errors; partial-output and other failures continue to fail the stream.

## Verification

Two independent source reviews completed across all lenses below; the frozen diff matches Git and passes diff whitespace checks. Traced the authenticated handler, bounded first-chunk preflight, cancellation ownership, HTTP status/token classification and quorum error handling. The new tests cover ordinary streaming, typed missing errors, byte preservation and mixed missing/I/O failures.

Author-reported focused tests and Docker reproduction were not rerun or independently audited in this review. The reported Docker comparison predates the final report_notfound-only gate; its scope is not distributed acceptance of the final head.

## Impact

Correctness: no findings; missing errors remain typed only before output. Security/trust: no findings; authentication, body digest and operation/status/token restrictions remain. Compatibility: no findings; legacy clean EOF remains, with the documented old-server limitation. Concurrency/durability: no findings; receiver ownership cancels an abandoned producer. Simplicity: no findings. Coverage: no blocking gap identified. Performance: no findings; preflight retains one bounded chunk and ordinary report_notfound=false streams start immediately.

## Additional Notes

This review does not establish that the patch fixes any current main CI failure. Full main CI and release acceptance remain separate gates.
2026-09-29 15:29:59 +08:00
Hauser 0a61f4a81e fix(heal): park healthy legacy MRF intents (#8225)
## Related Issues

rustfs/backlog#2682 and rustfs/rustfs#8192.

## Summary of Changes

Reviewed ca970f55ec against d2175d1e1e. No blocking code findings. The hold requires a completed healthy legacy check and an exact incarnation, lease, object, version and scope; it preserves the durable journal and rechecks on restart.

## Verification

Two independent source reviews completed, covering the lenses below. The frozen diff matches Git and passes diff whitespace checks. Reported Cargo and Docker results were not rerun or independently audited in this review.

One factual correction to the PR description: `legacy_sigkill_replay_repairs_without_releasing_unverified_responsibility` uses the default four-member fixture, with three replicas before restoration and four afterward (`mrf_partial_write_test.rs:559,567`). That named regression exercises the production manager path, but is not EC12+4. Other tests in the file use sixteen disks.

## Impact

Correctness: no findings; proofless legacy health never becomes a verified receipt. Security/trust: no findings; exact identity fences remain. Compatibility: no findings; journal encoding remains unchanged. Concurrency/durability: no findings; replay retains one checkpointed owner and replacement generations become retryable. Simplicity: no findings. Coverage: no blocking gap identified, with the test-scope correction above. Performance: no findings; held entries leave the retry index without adding per-admission queue scans.

## Additional Notes

The current main journal failure concerns DecodeFailure, which follows the separate ECDecode task path. This review does not establish that this PR fixes that failure or the scanner-cycle failure, and does not establish a passing main CI or release gate.
2026-09-29 15:28:46 +08:00
Chris d2175d1e1e fix(test): keep bucket disk faults across reconnects (#8224) 2026-09-29 14:07:30 +08:00
Chris 2019715d1a Fix fresh capacity probes for formatted local disks (#8222) 2026-09-29 12:23:06 +08:00
Chris 31f05a44af fix(test): use expect_err for offline state root rejection (#8220) 2026-09-29 11:22:16 +08:00
Chris a1724f3dbe feat(connect): capture signed offline service health (#8219) 2026-09-29 10:50:23 +08:00
Chris c62979a45b fix(connect): secure new offline state roots (#8218) 2026-09-29 10:13:58 +08:00
Chris e14eacc5ed fix(rpc): keep legacy format reads in their original namespace (#8216) 2026-09-29 09:24:39 +08:00
Chris 47af5565c7 fix(ci): preserve cluster startup failure evidence (#8213) 2026-09-29 08:50:09 +08:00
Chris f0ce628a7f feat(connect): capture offline native threads in service process (#8215) 2026-09-29 08:48:56 +08:00
Chris 32cc1eb76c Capture offline memory profile from the running service (#8214)
feat(connect): capture offline memory from the running service
2026-09-29 08:43:25 +08:00
Chris 84fe13989e fix(ecstore): retain read quota and control reserve test hedging (#8211) 2026-09-29 07:16:31 +08:00
Chris c8f42b8dca fix(ci): isolate scanner deadline and expose test failure details (#8210)
* fix(ci): isolate scanner deadline fixture and expose readiness errors

* test(ecstore): report unexpected capacity reservation errors
2026-09-29 07:16:17 +08:00
Chris c9acf01fd0 fix(tier): reread mutation intents after lock contention (#8209) 2026-09-29 07:16:01 +08:00
Chris 87e9a84a4e fix(ecstore): preserve RPC status when cloning storage errors (#8207)
* fix(ecstore): preserve RPC status when cloning storage errors

* test(heal): retain clone coverage without redundant ownership
2026-09-29 07:15:46 +08:00
Chris af0ae798f8 fix(test): serialize replacement_bucket_metadata heal test under parallel load (#8212) 2026-09-29 07:07:31 +08:00
dependabot[bot] a2bdcf497b build(deps): bump the dependencies group with 14 updates (#8176) 2026-09-29 01:59:07 +08:00
Chris 5564932f38 fix(heal): retry cancelled internode RPC operations (#8205) 2026-09-29 01:58:35 +08:00
Chris 8778d55e46 fix(ci): preserve diagnostics for rio storage test failures (#8204) 2026-09-29 00:15:26 +08:00
Chris 5ebc1e480b test: stabilize heal timing and improve heartbeat replay coverage (#8150)
fix: preserve heartbeat replay and stabilize timing tests
2026-09-28 22:07:15 +08:00
Chris 4d782141d8 fix(test): add missing unavailable_drives field to readiness test initializers (#8200)
Commit 151103a609 added the  field to
StorageReadinessDetails but did not update two test initializers in
readiness.rs, causing E0063 compile errors in --all-targets builds.
2026-09-28 21:26:42 +08:00
Hauser 3bc36acf7f fix(notify): persist Docker queue stores and expose open errors (#8201)
Co-authored-by: zhi22915 <qiuzgang@gmail.com>
2026-09-28 21:24:12 +08:00
Chris 4a91c75152 fix(ci): restore checks and multipart migration retries (#8196)
* fix(ci): restore multipart lint and full E2E membership checks

* fix(ecstore): preserve multipart completion identity during migration
2026-09-28 21:22:49 +08:00
Chris 2648776be7 fix: accept catalog files in health service artifact (#8198)
* fix: accept catalog files in health service artifact

* ci: retain configured health acceptance runner
2026-09-28 20:45:26 +08:00
RustFS 9442e89f5f fix(iam): require an explicit permission for force-delete (#8154)
* fix(iam): require an explicit permission for force-delete

A force-delete header no longer inherits s3:* or consoleAdmin. Bucket
force-delete requires s3:ForceDeleteBucket whenever the header is present,
and recursive object force-delete requires s3:ForceDeleteObject. A plain
delete keeps the existing checks.

Co-authored-by: RustFS <hello@rustfs.com>
Signed-off-by: loverustfs <155562731+loverustfs@users.noreply.github.com>

* test(e2e): keep force-delete header names static

The bucket force-delete helper must pass a static header name into the
SDK request mutator.

Co-authored-by: RustFS <hello@rustfs.com>
Signed-off-by: loverustfs <155562731+loverustfs@users.noreply.github.com>

* test(e2e): move the force-delete header into the request mutator

The SDK request customizer requires a static header name owned by the
closure.

Co-authored-by: RustFS <hello@rustfs.com>
Signed-off-by: loverustfs <155562731+loverustfs@users.noreply.github.com>

* fix(iam): keep force-delete out of NotAction grants

NotAction now uses plain wildcard matching, so NotAction "s3:*" still
excludes force-delete. An Allow statement grants s3:ForceDeleteObject or
s3:ForceDeleteBucket only when its Action list names the action; a
NotAction-only Allow never does. The rule applies to both IAM and bucket
policy statements.

Also build the invalid-header errors with S3Error::with_message to keep
the s3s footprint at its baseline, and fix a clippy single_match.

---------

Signed-off-by: loverustfs <155562731+loverustfs@users.noreply.github.com>
Co-authored-by: Hauser <housemecn@gmail.com>
Co-authored-by: overtrue <anzhengchao@gmail.com>
2026-09-28 19:08:33 +08:00
RustFS fc5609bbb0 fix(s3): make retried CompleteMultipartUpload idempotent (#8153)
Record the upload id on the completed object so a lost-response retry
returns that object's ETag instead of NoSuchUpload, while a different
part list or a replaced object keeps the existing errors.

Signed-off-by: loverustfs <155562731+loverustfs@users.noreply.github.com>
Co-authored-by: Hauser <housemecn@gmail.com>
2026-09-28 19:08:10 +08:00
Chris 64bd224b3b fix(connect): accept the health-era heartbeat and measured Top windows (#8195)
Two Connect tests fail on main in every Test and Lint lane.

Heartbeat: #8155 replaced "today's capabilities minus
profile.memory.service@1" with "the pre-health set minus it". That drops
the release that advertised health.check.service@1 but not yet in-service
memory profiling, so a pending heartbeat persisted by that release now
fails validation and the runtime stops. Accept that release as its own
frozen list.

Top disk/net: execute_*_job sets max_duration_millis to the requested
duration, then the capture compared the measured sleep against it. The
timer only overshoots, so a job that ran exactly as requested was
rejected with LimitExceeded whenever the overshoot reached 1 ms. Report
the authorized window, as Top API already does; validate_capture bounds
it by the limit.
2026-09-28 19:07:39 +08:00
Chris 71f8b607ad fix(ci): restore mainline and scheduled test reliability (#8187)
* fix(ci): restore mainline and scheduled test reliability

* fix(ci): provide GitHub CLI for CPU acceptance

* ci: provide Docker for CPU service acceptance

* ci: provision Python and Docker for OIDC validation

* ci: restore hosted runners for Docker validation

* ci: use verified MinIO release packages for interop

* ci: preserve host ownership of MinIO fixtures

* fix(ci): correct diagnostic limits and isolate startup checks

* test(connect): include object CLI failure details

* test(readiness): initialize unavailable drive diagnostics
2026-09-28 19:07:29 +08:00
143 changed files with 7262 additions and 976 deletions
+2 -2
View File
@@ -1,2 +1,2 @@
sha256-linux=21ab66fd172018682ce825a53f26aa8f286d57f3f687d8616d1850e56474dcd8
sha256-darwin=21ab66fd172018682ce825a53f26aa8f286d57f3f687d8616d1850e56474dcd8
sha256-linux=6fc377fa1f9f06e065f077c7bd53d04efbccf4453f95dde67afaa47778185377
sha256-darwin=6fc377fa1f9f06e065f077c7bd53d04efbccf4453f95dde67afaa47778185377
+2 -2
View File
@@ -1,2 +1,2 @@
sha256-darwin=9e9687ded961aa4e619a15def6e1ee0a5153be05423d009ee7a58126624c83e7
sha256-linux=364666693c2bf258997643c9d87e07e7b7501e58175f190d201d53c81946eb20
sha256-darwin=3b6ff112ff16acbaa4148c4736095a1b6b71cc11af0dd3513ff28c6bdc334ca8
sha256-linux=4e8bfc01870565a0a608a1634e6919fc593ce120a970aa04d974f7edf2675c21
+2 -2
View File
@@ -1,2 +1,2 @@
sha256-darwin=83a7dcaffd5a789517ae9f02a224f66a9713937885cff96fca2ad7e216f197ae
sha256-linux=626c10f8c964507ff987b6c86069e9019dc6d2ae7fb02db9be5df5aa8cc5145b
sha256-darwin=efe68675f4676a0d2901242c5815b1f7af5cdec8231f7cccd408f6dacf5c75db
sha256-linux=003ed6480935a0a95680cc061be143722fbd37bd477b8ff98607b1a064f3aef1
+2
View File
@@ -52,6 +52,7 @@ script-tests: ## Run shell script tests
./scripts/check_embedded_secrets.sh --self-test
$(RUSTFS_PYTHON_BIN) ./scripts/check_test_wiring.py --self-test
$(RUSTFS_PYTHON_BIN) ./scripts/test_e2e_binary.py
$(RUSTFS_PYTHON_BIN) ./scripts/test_migration_gate_evidence.py
$(RUSTFS_PYTHON_BIN) ./scripts/ci_gate.py --self-test
$(RUSTFS_PYTHON_BIN) ./scripts/check_security_coverage.py --self-test
$(RUSTFS_PYTHON_BIN) ./scripts/check_scheduled_validation_freshness.py --self-test
@@ -61,6 +62,7 @@ script-tests: ## Run shell script tests
$(RUSTFS_PYTHON_BIN) ./scripts/test_functional_chain_health.py
$(RUSTFS_PYTHON_BIN) ./scripts/test_ci_timing_report.py
$(RUSTFS_PYTHON_BIN) ./scripts/s3-tests/test_report_compat.py
$(RUSTFS_PYTHON_BIN) ./scripts/s3-tests/test_runner_tools.py
bash -n ./scripts/validate_object_data_cache_cold_stampede.sh
$(RUSTFS_PYTHON_BIN) ./scripts/check_object_data_cache_follower_samples.py --self-test
./scripts/validate_object_data_cache_cold_stampede.sh --self-test
+40
View File
@@ -107,6 +107,16 @@ test-group = 'ecstore-serial-flaky'
filter = 'package(rustfs-ecstore) & test(/^set_disk::ops::heal::heal_result_report_tests::/)'
test-group = 'ecstore-serial-flaky'
# Serialize the replacement bucket metadata heal test. It builds a real-disk
# hermetic erasure set, writes bucket metadata, then asserts byte-identical
# xl.meta readback after a rejected delayed repair. Under parallel load,
# concurrent ecstore tests mutate the process-wide disk registry and alter
# the serialized timestamp/disk-identity fields, flipping the assertion.
# Preventive serialization only, no retries.
[[profile.default.overrides]]
filter = 'package(rustfs-ecstore) & test(replacement_bucket_metadata_respects_pool_ownership_and_required_records)'
test-group = 'ecstore-serial-flaky'
# Serialize the metadata-cache generation-retirement pair. Both carry
# #[serial(metadata_cache_invalidation_probe)] — a no-op across nextest's
# process boundary — and assert get_object_metadata_cache generation
@@ -139,6 +149,19 @@ test-group = 'embedded-test-ports'
filter = 'package(rustfs) & (binary(connect_inventory) | binary(connect_perf_drive) | (binary(connect_perf_object) & test(=real_rustfs_endpoint_and_production_cli_support_bounded_get_and_put)))'
threads-required = "num-test-threads"
# These concurrent state writers keep the production 5s lock-acquisition limit
# while their peer completes durable IO. Reserve capacity from unrelated test
# processes while preserving each test's internal two-writer race.
[[profile.default.overrides]]
filter = 'package(rustfs) & test(/^admin::handlers::site_replication::tests::test_(peer_edit_generations_are_unique_across_nodes|recreated_state_object_allocates_over_the_previous_lifetimes_mark|state_object_lock_serializes_writers_from_separate_nodes|retry_event_persist_must_not_wipe_concurrent_locked_rmw)$/)'
threads-required = "num-test-threads"
# Reserve capacity for these short heartbeat TLS/fsync probes (rustfs#8129).
# Their existing deadlines, assertions, and retry policy remain in force.
[[profile.default.overrides]]
filter = 'package(rustfs) & binary(connect_heartbeat) & test(/^(restart_replays_pending_request_then_advances_sequence|restart_replays_exact_legacy_capabilities_with_and_without_jobs)$/)'
threads-required = "num-test-threads"
# Serialize the durable manual-transition checkpoint test across nextest's
# process boundary; it mutates bucket lifecycle metadata and is not quarantined.
[[profile.default.overrides]]
@@ -321,6 +344,11 @@ test-group = 'ecstore-serial-flaky'
filter = 'package(rustfs-ecstore) & test(/^set_disk::ops::heal::heal_result_report_tests::/)'
test-group = 'ecstore-serial-flaky'
# Serialize the replacement bucket metadata heal test under the ci profile too.
[[profile.ci.overrides]]
filter = 'package(rustfs-ecstore) & test(replacement_bucket_metadata_respects_pool_ownership_and_required_records)'
test-group = 'ecstore-serial-flaky'
# Serialize the metadata-cache generation-retirement pair under the ci
# profile too (see the matching default-profile override near the top). Not a
# quarantine: no retries.
@@ -338,6 +366,18 @@ test-group = 'embedded-test-ports'
filter = 'package(rustfs) & (binary(connect_inventory) | binary(connect_perf_drive) | (binary(connect_perf_object) & test(=real_rustfs_endpoint_and_production_cli_support_bounded_get_and_put)))'
threads-required = "num-test-threads"
# Keep the same state-writer capacity reservation in CI without changing the
# production lock deadline, internal concurrency, assertions, or retry policy.
[[profile.ci.overrides]]
filter = 'package(rustfs) & test(/^admin::handlers::site_replication::tests::test_(peer_edit_generations_are_unique_across_nodes|recreated_state_object_allocates_over_the_previous_lifetimes_mark|state_object_lock_serializes_writers_from_separate_nodes|retry_event_persist_must_not_wipe_concurrent_locked_rmw)$/)'
threads-required = "num-test-threads"
# Reserve capacity for these short heartbeat TLS/fsync probes (rustfs#8129).
# Their existing deadlines, assertions, and retry policy remain in force.
[[profile.ci.overrides]]
filter = 'package(rustfs) & binary(connect_heartbeat) & test(/^(restart_replays_pending_request_then_advances_sequence|restart_replays_exact_legacy_capabilities_with_and_without_jobs)$/)'
threads-required = "num-test-threads"
# Serialize the durable manual-transition checkpoint test under the ci profile
# too. No retries: failures stay visible.
[[profile.ci.overrides]]
+1 -6
View File
@@ -100,12 +100,7 @@ runs:
- name: Check test wiring
shell: bash
run: |
python3 ./scripts/check_test_wiring.py --self-test
python3 ./scripts/check_scheduled_validation_freshness.py --self-test
python3 ./scripts/test_security_workflow.py
python3 ./scripts/test_nightly_candidate.py
python3 ./scripts/check_test_wiring.py
run: python3 ./scripts/check_test_wiring.py
- name: Check no planning docs committed
shell: bash
+2
View File
@@ -107,6 +107,8 @@ runs:
- name: Install cargo-nextest
if: inputs.install-test-tools == 'true'
uses: taiki-e/install-action@96c7780c1d8a2b8723e12031def873a434d39d8d # nextest
with:
tool: cargo-nextest@0.9.138
- name: Setup Rust cache
uses: Swatinem/rust-cache@6323deb102c322ba6fcbdcafc7e3dddab59af2b6 # v2
@@ -1,61 +0,0 @@
# Copyright 2024 RustFS Team
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
name: Architecture Migration Rules
on:
pull_request:
types: [ opened, synchronize, reopened, closed ]
branches: [ main ]
paths:
- "ARCHITECTURE.md"
- "docs/architecture/**"
- "scripts/check_architecture_migration_rules.sh"
- ".github/workflows/architecture-migration-rules.yml"
workflow_dispatch:
permissions:
contents: read
concurrency:
group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
cancel-in-progress: true
jobs:
cancel-closed-pr-runs:
name: Cancel Closed PR Runs
if: github.event_name == 'pull_request' && github.event.action == 'closed'
runs-on: sm-standard-2
timeout-minutes: 10
steps:
- name: Explain cancellation run
run: echo "PR closed; this run only cancels older runs in the same concurrency group."
architecture-migration-rules:
name: Architecture Migration Rules
if: github.event_name != 'pull_request' || github.event.action != 'closed'
runs-on: sm-standard-2
timeout-minutes: 10
steps:
- uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
with:
persist-credentials: false
- name: Install ripgrep
uses: taiki-e/install-action@7623a79cdfecb99d681017af368ca353d9f49bb5 # v2
with:
tool: ripgrep@15.2.0
- name: Check architecture migration rules
run: ./scripts/check_architecture_migration_rules.sh
+1 -1
View File
@@ -204,7 +204,7 @@ jobs:
# open the Actions tab. Same ci-8 mechanism coverage.yml and
# e2e-replication-nightly.yml already use.
if: always() && github.event_name == 'schedule' && contains(needs.*.result, 'failure')
runs-on: sm-standard-2
runs-on: ubuntu-latest
timeout-minutes: 10
permissions:
contents: read
+1 -1
View File
@@ -1167,7 +1167,7 @@ jobs:
if: >-
always() && github.event_name == 'schedule' &&
(contains(needs.*.result, 'failure') || contains(needs.*.result, 'cancelled'))
runs-on: sm-standard-2
runs-on: ubuntu-latest
timeout-minutes: 10
permissions:
contents: read
+6 -6
View File
@@ -15,16 +15,16 @@
# Sole writer of the Rust dependency caches that ci.yml restores.
#
# Why this is a separate workflow rather than steps inside ci.yml: ci.yml's
# concurrency group cancels in-progress runs on main pushes, and merges land far
# faster than its 70-minute pipeline. Measured over 15 consecutive main pushes:
# concurrency group originally cancelled in-progress main runs, while merges
# landed faster than its 70-minute pipeline. Over 15 consecutive main pushes:
# 12 cancelled, 2 failed, 0 succeeded. A cancelled run never reaches
# Swatinem/rust-cache's post step (cache-on-failure does not cover cancellation),
# so the writer lanes were saving nothing and every PR paid a cold restore —
# 11.8-20.9 minutes of "Setup Rust environment" against 0.7-3.4 warm.
#
# Splitting cache writing out of the test pipeline lets ci.yml keep cancelling
# superseded runs (which is correct — nobody needs test results for a commit
# that is already three merges behind) while the caches still get written.
# Keep cache writing separate from validation: only these jobs publish the
# complete feature closure for each key. Main validation now also finishes its
# running baseline, while superseded PR attempts can still be cancelled.
#
# The group below deliberately does NOT cancel in progress; see the comment on
# it for how that bounds concurrency and why it is scoped by event.
@@ -256,7 +256,7 @@ jobs:
# belonged to it.
warm-ci-uring:
name: Warm ci-uring
runs-on: sm-standard-2
runs-on: ubuntu-latest
timeout-minutes: 60
steps:
- name: Checkout repository
+63 -36
View File
@@ -47,12 +47,12 @@ permissions:
contents: read
# Concurrency groups are scoped per event so different triggers never cancel
# each other: PR pushes cancel the previous run of that PR, main pushes keep
# latest-wins semantics among themselves, and scheduled runs always complete
# (a shared group used to let every merge kill the weekly scheduled run).
# each other. PR pushes cancel superseded attempts; main and scheduled runs
# finish so a busy merge stream cannot starve the complete baseline. For main,
# GitHub retains one running run and replaces the pending run with the latest.
concurrency:
group: ${{ github.workflow }}-${{ github.event_name }}-${{ github.event.pull_request.number || github.ref }}
cancel-in-progress: ${{ github.event_name != 'schedule' }}
cancel-in-progress: ${{ github.event_name == 'pull_request' }}
env:
CARGO_TERM_COLOR: always
@@ -128,7 +128,7 @@ jobs:
test-and-lint:
name: Workspace Test and Lint
if: needs.classify-changes.outputs.mode == 'full' && (github.event_name != 'pull_request' || github.event.action != 'closed')
if: contains(fromJSON('["full", "e2e"]'), needs.classify-changes.outputs.mode) && (github.event_name != 'pull_request' || github.event.action != 'closed')
needs: [ quick-checks, classify-changes ]
runs-on: sm-standard-4
timeout-minutes: 90
@@ -146,11 +146,9 @@ jobs:
uses: ./.github/actions/setup
with:
rust-version: stable
# Every lane in this workflow reads its cache and none writes it.
# cache-warm.yml is the sole writer for all four keys: this workflow
# cancels superseded runs on main, and a cancelled run never reaches
# rust-cache's post step, so writing from here saved nothing (12 of 15
# consecutive main-push runs were cancelled). See rustfs/backlog#1600.
# Every lane reads its cache; cache-warm.yml remains the sole writer
# with the matching feature closure. Tests never compete to replace
# shared caches, including when a PR attempt is cancelled.
cache-shared-key: ci-dev
cache-save-if: 'false'
install-build-packaging-tools: 'false'
@@ -281,21 +279,10 @@ jobs:
- name: Check log-analyzer rule anchors
run: ./scripts/check_log_analyzer_rules.sh
# Explicit gate for migration-critical suites. These tests already ran in
# the full nextest pass above; a single filtered nextest invocation keeps
# the named gate without rebuilding or re-running them one package at a time.
#
# The gate selects tests by name substring (data_movement / rebalance /
# decommission / source_cleanup / delete_marker), so renames can silently
# thin it. The script owns the filter expression and first verifies the
# selected-test count against the committed floor in
# .config/migration-gate-floor.txt before running the gate; renames or
# removals must update that file consciously (see the script header).
# Kept on the default profile (no --profile ci): a second --profile ci run
# would clobber target/nextest/ci/junit.xml, and none of these tests are
# quarantined so they gain nothing from the ci profile's retry overrides.
- name: Run rebalance/decommission migration proofs
run: ./scripts/check_migration_gate_count.sh
# Preserve the migration floor and require successful execution evidence
# from the workspace run, without compiling or executing its subset again.
- name: Verify rebalance/decommission migration proofs
run: ./scripts/check_migration_gate_count.sh evidence artifacts/test-and-lint/core-test-listing.json target/nextest/ci/junit.xml
# This gate builds into a fresh target directory to isolate the E2E root.
# Give its cold build a separate budget from workspace tests and migration proofs.
@@ -431,6 +418,7 @@ jobs:
run: cargo clippy -p rustfs -p rustfs-ecstore --all-targets --features rio-v2 -- -D warnings
- name: Run rio-v2 feature tests
shell: bash
env:
# Match the main nextest lane's #5394 link-I/O guard. A cold feature
# cache otherwise fans out enough rust-lld processes to exhaust this
@@ -440,9 +428,25 @@ jobs:
# --profile ci so the quarantine list (and its junit flaky markers)
# covers this leg too; the default profile is the local no-retry
# profile and silently ignored quarantined flakes here (rustfs#6703).
cargo nextest run --profile ci -p rustfs -p rustfs-ecstore --features rio-v2
mkdir -p artifacts/test-and-lint
rm -f target/nextest/ci/junit.xml
./scripts/ci/resource_sampler.sh start nextest-rio-v2
trap './scripts/ci/resource_sampler.sh stop' EXIT
cargo nextest run --profile ci -p rustfs -p rustfs-ecstore --features rio-v2 \
2>&1 | tee artifacts/test-and-lint/nextest.log
cargo test -p rustfs --doc --features rio-v2
- name: Upload rio-v2 test reports and diagnostics
if: always()
uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
with:
name: junit-test-and-lint-rio-v2-${{ github.run_number }}
path: |
target/nextest/ci/junit.xml
artifacts/test-and-lint
retention-days: 3
if-no-files-found: error
connect-short-credential-boundary:
name: Connect Short Credential Boundary
if: needs.classify-changes.outputs.mode == 'full' && (github.event_name != 'pull_request' || github.event.action != 'closed')
@@ -508,13 +512,9 @@ jobs:
runs-on: sm-standard-4
timeout-minutes: 90
strategy:
# On a PR, one failing protocol leg is enough to know the PR is not ready,
# so stop the sibling leg instead of paying another ~40 minutes for it.
# Everywhere else (main pushes, the merge queue, the weekly schedule) keep
# the full signal: there we want to know whether swift AND sftp are broken,
# not just whichever failed first. This is the only part of the early-stop
# work that also covers fork PRs, since it needs no token.
fail-fast: ${{ github.event_name == 'pull_request' }}
# Preserve both independent results when one protocol fails, so diagnosis
# and a focused fix do not require rebuilding an interrupted sibling.
fail-fast: false
matrix:
features:
- name: swift
@@ -545,6 +545,8 @@ jobs:
cargo clippy -p rustfs -p rustfs-protocols --all-targets ${{ matrix.features.flags }} -- -D warnings
- name: Run tests with ${{ matrix.features.name }}
id: protocol-tests
shell: bash
env:
# Keep feature-test linking under the same bounded concurrency as the
# main nextest lane; Clippy is metadata-only and needs no such limit.
@@ -553,11 +555,26 @@ jobs:
# --profile ci so the quarantine list (and its junit flaky markers)
# covers this leg too; the default profile is the local no-retry
# profile and silently ignored quarantined flakes here (rustfs#6703).
cargo nextest run --profile ci -p rustfs -p rustfs-protocols ${{ matrix.features.flags }}
mkdir -p artifacts/protocol-tests
rm -f target/nextest/ci/junit.xml
cargo nextest run --profile ci -p rustfs -p rustfs-protocols ${{ matrix.features.flags }} \
2>&1 | tee artifacts/protocol-tests/nextest.log
- name: Upload protocol test reports and diagnostics
if: >-
always() && contains(fromJSON('["success", "failure", "cancelled"]'), steps.protocol-tests.outcome)
uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
with:
name: junit-test-and-lint-${{ matrix.features.name }}-${{ github.run_number }}-${{ github.run_attempt }}
path: |
target/nextest/ci/junit.xml
artifacts/protocol-tests
retention-days: 3
if-no-files-found: warn
build-rustfs-debug-binary:
name: Build RustFS Debug Binary
if: needs.classify-changes.outputs.mode == 'full' && (github.event_name != 'pull_request' || github.event.action != 'closed')
if: contains(fromJSON('["full", "e2e"]'), needs.classify-changes.outputs.mode) && (github.event_name != 'pull_request' || github.event.action != 'closed')
needs: [ quick-checks, classify-changes ]
runs-on: sm-standard-4
timeout-minutes: 30
@@ -1005,6 +1022,7 @@ jobs:
RUSTFS_E2E_STARTUP_CAS_BINARY: ${{ runner.temp }}/rustfs-startup-cas-input/rustfs
RUSTFS_E2E_STARTUP_CAS_BUILD_MANIFEST: ${{ runner.temp }}/rustfs-startup-cas-input/rustfs.e2e-startup-cas-build.json
RUSTFS_E2E_STARTUP_CAS_ARTIFACT_DIR: ${{ runner.temp }}/rustfs-startup-cas-evidence
RUSTFS_HEAL_CHAOS_LOG_ROOT: ${{ runner.temp }}/rustfs-heal-chaos-logs
run: python3 scripts/e2e_binary.py run --binary "$RUSTFS_E2E_STARTUP_CAS_BINARY" --features e2e-test-hooks -- cargo nextest run --profile e2e-full -p e2e_test
- name: Upload junit
@@ -1028,6 +1046,15 @@ jobs:
if-no-files-found: warn
retention-days: 7
- name: Upload heal cluster logs
if: always()
uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
with:
name: e2e-full-heal-logs-${{ github.run_number }}
path: ${{ runner.temp }}/rustfs-heal-chaos-logs
if-no-files-found: warn
retention-days: 7
e2e-tests-rio-v2:
name: End-to-End Tests (rio-v2)
# Inherits the schedule/dispatch-only gate through needs: on every other
@@ -1270,7 +1297,7 @@ jobs:
if: >-
always() && github.event_name == 'schedule' &&
(contains(needs.*.result, 'failure') || contains(needs.*.result, 'cancelled'))
runs-on: sm-standard-2
runs-on: ubuntu-latest
timeout-minutes: 10
permissions:
contents: read
@@ -172,7 +172,7 @@ jobs:
assert outer.testzip() is None
with zipfile.ZipFile(io.BytesIO(outer.read(package))) as inner:
entries = inner.infolist()
assert {entry.filename for entry in entries} == {'rustfs', 'rustfs-cli'} and len(entries) == 2
assert {entry.filename for entry in entries} == {'rustfs', 'rustfs-cli', 'rustfs.cpu-symbol-catalog.json', 'rustfs.cpu-symbol-catalog.sha256'} and len(entries) == 4
assert all(regular(entry) and 0 < entry.file_size <= 1073741824 for entry in entries)
assert inner.testzip() is None
pathlib.Path('binary').mkdir()
+1 -1
View File
@@ -128,7 +128,7 @@ jobs:
# manual workflow_dispatch runs stay quiet so a debugging run never files a
# spurious alert.
if: always() && github.event_name == 'schedule' && contains(needs.*.result, 'failure')
runs-on: sm-standard-2
runs-on: ubuntu-latest
timeout-minutes: 10
permissions:
contents: read
+2 -1
View File
@@ -38,6 +38,7 @@ on:
- "Cargo.lock"
- "Cargo.toml"
- ".config/nextest.toml"
- ".config/e2e-distributed-selection.txt"
- ".github/workflows/e2e-distributed.yml"
- "crates/audit/**"
- "crates/common/**"
@@ -200,7 +201,7 @@ jobs:
name: Alert on scheduled failure
needs: [distributed]
if: always() && github.event_name == 'schedule' && contains(needs.*.result, 'failure')
runs-on: sm-standard-2
runs-on: ubuntu-latest
timeout-minutes: 10
permissions:
contents: read
@@ -230,7 +230,7 @@ jobs:
# manual workflow_dispatch runs stay quiet so a debugging run never files a
# spurious alert.
if: always() && github.event_name == 'schedule' && contains(needs.*.result, 'failure')
runs-on: sm-standard-2
runs-on: ubuntu-latest
timeout-minutes: 10
permissions:
contents: read
+1 -1
View File
@@ -453,7 +453,7 @@ jobs:
# job fails. Alerts only for scheduled runs (backlog#1149 ci-8) — manual
# dispatch failures are already watched by a human.
if: always() && github.event_name == 'schedule' && contains(needs.*.result, 'failure')
runs-on: sm-standard-2
runs-on: ubuntu-latest
timeout-minutes: 10
permissions:
contents: read
+62 -17
View File
@@ -51,8 +51,53 @@ env:
UPGRADE_SOURCE_SHA256: 3ee8df71e8edcfada533be452c4135868f697bc515460ae97b027313eade7a3d
jobs:
build:
name: Build upgrade candidate
runs-on: sm-standard-2
timeout-minutes: 60
env:
FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
steps:
- name: Checkout repository
uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
with:
persist-credentials: false
- name: Setup Rust environment
uses: ./.github/actions/setup
with:
cache-shared-key: e2e-upgrade-server
cache-save-if: ${{ github.ref == 'refs/heads/main' }}
install-build-packaging-tools: "false"
install-test-tools: "false"
- name: Download pinned previous release
run: |
mkdir -p target/debug
archive="target/debug/$UPGRADE_SOURCE_ASSET"
curl --fail --location --retry 3 --output "$archive" \
"https://github.com/${GITHUB_REPOSITORY}/releases/download/${UPGRADE_SOURCE_VERSION}/${UPGRADE_SOURCE_ASSET}"
echo "$UPGRADE_SOURCE_SHA256 $archive" | sha256sum --check --strict
- name: Build current RustFS binary
run: python3 scripts/e2e_binary.py build
- name: Upload verified upgrade candidate
uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
with:
name: upgrade-candidate-${{ github.run_id }}
path: |
target/debug/rustfs
target/debug/rustfs.e2e.json
target/debug/${{ env.UPGRADE_SOURCE_ASSET }}
if-no-files-found: error
retention-days: 3
compression-level: 0
overwrite: true
upgrade:
name: ${{ matrix.name }}
needs: build
strategy:
fail-fast: false
matrix:
@@ -63,31 +108,24 @@ jobs:
# the CI required-check names. UPGRADE_SOURCE_VERSION above is the
# single source of truth for which release they actually run against.
- name: Direct upgrade from the previous release
cache_key: e2e-direct-upgrade
test: direct_upgrade_from_rc2_preserves_object_contracts
artifact: direct-upgrade
- name: Mixed-version rolling upgrade from the previous release
cache_key: e2e-mixed-version-upgrade
test: rolling_upgrade_from_rc2_preserves_mixed_version_contracts
artifact: mixed-version-upgrade
- name: Bucket configuration survives the upgrade
cache_key: e2e-bucket-config-upgrade
test: direct_upgrade_from_previous_release_preserves_bucket_configuration
artifact: bucket-config-upgrade
- name: Rollback reads current bucket metadata
cache_key: e2e-bucket-config-rollback
test: rollback_to_previous_release_reads_current_bucket_metadata
artifact: bucket-config-rollback
- name: ODM configuration recovery after rc.5 rollback
cache_key: e2e-odm-config-rollback
test: rc5_rollback_requires_restoring_odm_configuration
artifact: odm-config-rollback
- name: Multipart layouts survive the rc.5 upgrade
cache_key: e2e-multipart-layout-upgrade
test: direct_upgrade_from_rc5_preserves_multipart_layouts
artifact: multipart-layout-upgrade
- name: rc.5 multipart replication baseline
cache_key: e2e-multipart-layout-baseline
test: rc5_baseline_replicates_multipart_layouts
artifact: multipart-layout-baseline
runs-on: sm-standard-2
@@ -103,19 +141,28 @@ jobs:
- name: Setup Rust environment
uses: ./.github/actions/setup
with:
cache-shared-key: ${{ matrix.cache_key }}
cache-save-if: ${{ github.ref == 'refs/heads/main' }}
cache-shared-key: e2e-upgrade-tests
# One main-branch consumer warms the shared test-harness dependencies.
cache-save-if: ${{ github.ref == 'refs/heads/main' && matrix.artifact == 'direct-upgrade' }}
install-build-packaging-tools: "false"
install-test-tools: "false"
- name: Download pinned previous release
- name: Download verified upgrade candidate
uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c # v8.0.1
with:
name: upgrade-candidate-${{ github.run_id }}
path: target/debug
- name: Restore candidate executable bit
run: chmod +x target/debug/rustfs
- name: Verify and unpack pinned previous release
env:
SOURCE_DIR: ${{ runner.temp }}/rustfs-upgrade-source
run: |
set -euo pipefail
mkdir -p "$SOURCE_DIR"
archive="$SOURCE_DIR/$UPGRADE_SOURCE_ASSET"
curl --fail --location --retry 3 --output "$archive" \
"https://github.com/${GITHUB_REPOSITORY}/releases/download/${UPGRADE_SOURCE_VERSION}/${UPGRADE_SOURCE_ASSET}"
archive="target/debug/$UPGRADE_SOURCE_ASSET"
echo "$UPGRADE_SOURCE_SHA256 $archive" | sha256sum --check --strict
unzip -q "$archive" -d "$SOURCE_DIR"
chmod +x "$SOURCE_DIR/rustfs"
@@ -123,14 +170,12 @@ jobs:
echo "RUSTFS_UPGRADE_SOURCE_BINARY=$SOURCE_DIR/rustfs" >> "$GITHUB_ENV"
echo "RUSTFS_E2E_LOG_DIR=$RUNNER_TEMP/rustfs-upgrade-logs" >> "$GITHUB_ENV"
- name: Build current RustFS binary
run: |
python3 scripts/e2e_binary.py build
- name: Run upgrade compatibility test
env:
RUSTFS_SCANNER_HEAL_G09_EVIDENCE_DIR: ${{ runner.temp }}/rustfs-upgrade-g09-evidence/${{ matrix.artifact }}
NO_PROXY: 127.0.0.1,localhost,::1
run: |
export no_proxy="$NO_PROXY"
python3 scripts/e2e_binary.py run -- cargo test --locked -p e2e_test \
"upgrade_compatibility_test::${{ matrix.test }}" \
-- --ignored --exact --nocapture
+1 -1
View File
@@ -240,7 +240,7 @@ jobs:
# job fails. Alerts only for scheduled (nightly) runs (backlog#1149
# ci-8); PR and manual dispatch failures are already watched by a human.
if: always() && github.event_name == 'schedule' && contains(needs.*.result, 'failure')
runs-on: sm-standard-2
runs-on: ubuntu-latest
timeout-minutes: 10
permissions:
contents: read
+3 -3
View File
@@ -60,9 +60,9 @@ permissions:
jobs:
minio-interop:
name: MinIO interop (EC + SSE read parity)
# Skip on forks: needs the repo's runners and is not a contributor gate.
# Keep this scheduled interoperability lane scoped to the upstream repo.
if: github.repository == 'rustfs/rustfs'
runs-on: sm-standard-2
runs-on: ubuntu-latest
timeout-minutes: 40
env:
# Fixed 32-byte test KMS key baked into the fixture lab; not a secret.
@@ -131,7 +131,7 @@ jobs:
name: Alert on scheduled failure
needs: [minio-interop]
if: always() && github.event_name == 'schedule' && contains(needs.*.result, 'failure')
runs-on: sm-standard-2
runs-on: ubuntu-latest
timeout-minutes: 10
permissions:
contents: read
+1 -1
View File
@@ -258,7 +258,7 @@ jobs:
# job fails. Alerts only for scheduled runs (backlog#1149 ci-8) — manual
# dispatch failures are already watched by a human.
if: always() && github.event_name == 'schedule' && contains(needs.*.result, 'failure')
runs-on: sm-standard-2
runs-on: ubuntu-latest
timeout-minutes: 10
permissions:
contents: read
+1 -1
View File
@@ -498,7 +498,7 @@ jobs:
if: >-
always() && github.event_name == 'schedule' &&
(contains(needs.*.result, 'failure') || contains(needs.*.result, 'cancelled'))
runs-on: sm-standard-2
runs-on: ubuntu-latest
timeout-minutes: 10
permissions:
contents: read
+7 -2
View File
@@ -51,7 +51,7 @@ concurrency:
jobs:
oidc-keycloak-live:
name: OIDC Keycloak live gate
runs-on: sm-standard-2
runs-on: ubuntu-latest
timeout-minutes: 60
env:
FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
@@ -72,6 +72,11 @@ jobs:
- name: Build RustFS
run: cargo build --locked -p rustfs --bin rustfs
- name: Set up Python
uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0
with:
python-version: "3.12"
- name: Install pinned request signer
run: |
python3 -m pip install --user --upgrade pip "awscurl==0.44"
@@ -95,7 +100,7 @@ jobs:
if: >-
always() && github.event_name == 'schedule' &&
(needs.oidc-keycloak-live.result == 'failure' || needs.oidc-keycloak-live.result == 'cancelled')
runs-on: sm-standard-2
runs-on: ubuntu-latest
timeout-minutes: 10
permissions:
contents: read
@@ -69,10 +69,9 @@ env:
jobs:
minio-source:
name: MinIO source (read-through, list-through, backfill)
# Skip on forks: needs this repository's runners and is not a contributor
# gate.
# Keep this scheduled interoperability lane scoped to the upstream repo.
if: github.repository == 'rustfs/rustfs'
runs-on: sm-standard-2
runs-on: ubuntu-latest
timeout-minutes: 90
env:
NO_PROXY: 127.0.0.1,localhost
@@ -108,11 +107,15 @@ jobs:
run: |
set -euo pipefail
mkdir -p artifacts/odm-interop/minio
docker build -t rustfs-odm-interop-minio \
-f crates/rio-v2/tests/minio_fixture_lab/Dockerfile \
crates/rio-v2/tests/minio_fixture_lab
docker run -d --name rustfs-odm-interop-minio \
--entrypoint /usr/local/bin/minio \
-e "MINIO_ROOT_USER=${MINIO_ROOT_USER}" \
-e "MINIO_ROOT_PASSWORD=${MINIO_ROOT_PASSWORD}" \
-p 9100:9000 \
minio/minio:RELEASE.2025-09-07T16-13-09Z server /data
rustfs-odm-interop-minio server /data
for _ in $(seq 1 120); do
curl -fsS http://127.0.0.1:9100/minio/health/live >/dev/null 2>&1 && break
sleep 1
@@ -300,7 +303,7 @@ jobs:
if: >-
always() && github.event_name == 'schedule' &&
(contains(needs.*.result, 'failure') || contains(needs.*.result, 'cancelled'))
runs-on: sm-standard-2
runs-on: ubuntu-latest
timeout-minutes: 10
permissions:
contents: read
+1 -1
View File
@@ -366,7 +366,7 @@ jobs:
if: >-
always() && github.event_name == 'schedule' &&
(contains(needs.*.result, 'failure') || contains(needs.*.result, 'cancelled'))
runs-on: sm-standard-2
runs-on: ubuntu-latest
timeout-minutes: 10
permissions:
contents: read
+1 -1
View File
@@ -66,7 +66,7 @@ jobs:
# scheduled runs file a tracking issue, manual dispatch stays quiet so
# debugging never produces a spurious alert.
if: always() && github.event_name == 'schedule' && contains(needs.*.result, 'failure')
runs-on: sm-standard-2
runs-on: ubuntu-latest
timeout-minutes: 10
permissions:
contents: read
+2 -2
View File
@@ -106,7 +106,7 @@ jobs:
if: ${{ inputs.chain_manifest != '' }}
env:
CHAIN_MANIFEST: ${{ inputs.chain_manifest }}
run: python3 scripts/functional_chain_evidence.py consume
run: python3 rustfs-repo/scripts/functional_chain_evidence.py consume
- name: Checkout auto-testing scripts
uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
@@ -389,7 +389,7 @@ jobs:
CHAIN_TEST_OUTCOME: ${{ steps.test.outcome }}
CHAIN_REPORT_OUTCOME: ${{ steps.report.outcome }}
run: >-
python3 scripts/functional_chain_evidence.py record --suite security
python3 rustfs-repo/scripts/functional_chain_evidence.py record --suite security
--report "${SECURITY_ARTIFACTS_DIR}/suite-report.md"
--output "${RUNNER_TEMP}/chain-security-${GITHUB_RUN_ID}-${GITHUB_RUN_ATTEMPT}/security.json"
@@ -58,7 +58,7 @@ jobs:
# Mirrors the consumer wiring, minus the schedule-event guard (this
# workflow is dispatch-only by design).
if: always() && contains(needs.*.result, 'failure')
runs-on: sm-standard-2
runs-on: ubuntu-latest
timeout-minutes: 10
permissions:
contents: read
+1 -1
View File
@@ -177,7 +177,7 @@ jobs:
if: >-
always() && github.event_name == 'schedule' &&
(contains(needs.*.result, 'failure') || contains(needs.*.result, 'cancelled'))
runs-on: sm-standard-2
runs-on: ubuntu-latest
timeout-minutes: 10
permissions:
contents: read
Generated
+200 -152
View File
@@ -987,9 +987,9 @@ dependencies = [
[[package]]
name = "aws-sdk-kms"
version = "1.121.0"
version = "1.122.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9febf0461c7441895de457dcff49dd7fa8f013793367f6b81d422c64e939e1cc"
checksum = "ae92186214a60f1316f6e25df2b5bef8f67b5b08b45f8e3eb6e1278b9ddd7b29"
dependencies = [
"arc-swap",
"aws-credential-types",
@@ -1012,9 +1012,9 @@ dependencies = [
[[package]]
name = "aws-sdk-s3"
version = "1.149.0"
version = "1.150.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d48257fb0eb1907a02a965ff7d1b910770a6de2c59a2d03461678df989c5cdcd"
checksum = "2fecaa77c862fa1e9d636d54fa06e0049a19a4f071b0d72ea1b9086887d09359"
dependencies = [
"arc-swap",
"aws-credential-types",
@@ -1048,9 +1048,9 @@ dependencies = [
[[package]]
name = "aws-sdk-sso"
version = "1.112.0"
version = "1.113.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d3d59c47556b283bb2d20fc0b416df7d9d9a3d6f5fa0e77867167a3f2efc23b9"
checksum = "87184f36ade012f46cdc773c7be10031b4545a1aea70a9a22a044e8f36f1e5c5"
dependencies = [
"arc-swap",
"aws-credential-types",
@@ -1073,9 +1073,9 @@ dependencies = [
[[package]]
name = "aws-sdk-ssooidc"
version = "1.114.0"
version = "1.115.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "eceb9f31b46ba6da21f666e456395624824a2a086caed940b13d78832fa3ee02"
checksum = "84edc7495d56e006dff0031cbc9782784902456f7024651c2da902a3d2336262"
dependencies = [
"arc-swap",
"aws-credential-types",
@@ -1098,9 +1098,9 @@ dependencies = [
[[package]]
name = "aws-sdk-sts"
version = "1.117.0"
version = "1.118.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9291df9ea767ab5e77d08a53f4f22daa917909a34ed7a22c1a4b82d193a423ad"
checksum = "df7572569d018ca72923073494eac3c4671a0daea90b9e3923c2d27a3a02a2c1"
dependencies = [
"arc-swap",
"aws-credential-types",
@@ -1193,9 +1193,9 @@ dependencies = [
[[package]]
name = "aws-smithy-http"
version = "0.64.0"
version = "0.64.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "37843d9add67c3aff5856f409c6dc315d3cdff60f9c0cb5b670dab1e9920306d"
checksum = "639b4d8f8555f24a9be649811c3eb0b4d4616f4d61daf0c32e28873bc1ea9af1"
dependencies = [
"aws-smithy-eventstream",
"aws-smithy-runtime-api",
@@ -1321,9 +1321,9 @@ dependencies = [
[[package]]
name = "aws-smithy-runtime-api"
version = "1.17.0"
version = "1.18.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3fbf162725183ec0df77a9dc82ce22d9ba0b6ea7f376e8ec4f6b695727c70698"
checksum = "9bb0deb40a69de59f7809251cfe1ab16de135f42c49d222d1202b4d0ce19720f"
dependencies = [
"aws-smithy-async",
"aws-smithy-runtime-api-macros",
@@ -1360,9 +1360,9 @@ dependencies = [
[[package]]
name = "aws-smithy-types"
version = "1.8.0"
version = "1.8.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "16ecf7989e381d0cbb7ca06104140d940bf38d8eb9004bd3a12d23df03e3d363"
checksum = "69bb407740a197147da48238ecc94498493c9e85445732360cec180296ca45f1"
dependencies = [
"base64-simd",
"bytes",
@@ -1746,7 +1746,7 @@ checksum = "f8b851b75c23ca7873623d612fe49bd1989aeb03d08fb9432187eb253d3d4c6b"
dependencies = [
"alloc-no-stdlib 3.0.0",
"alloc-stdlib 0.3.0",
"brotli-decompressor 6.0.0",
"brotli-decompressor 6.0.1",
]
[[package]]
@@ -1761,9 +1761,9 @@ dependencies = [
[[package]]
name = "brotli-decompressor"
version = "6.0.0"
version = "6.0.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "46087801e40b785410c352bee60e9b6226312709cf52d5fff92776cdf5cf6490"
checksum = "941cd9bd4ddab83cb46fa5a2d428f1c857b24ac78cb876cf7beb710840934bd7"
dependencies = [
"alloc-no-stdlib 3.0.0",
"alloc-stdlib 0.3.0",
@@ -1971,9 +1971,9 @@ dependencies = [
[[package]]
name = "cc"
version = "1.4.7"
version = "1.5.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "54413ede23c2daf518f35156dfde027feb2374004d63bd497f983c8db9c0e313"
checksum = "f360145194ee8e21db5ee7f3fcd4fe52210864c75c985dae33218202c8bbe040"
dependencies = [
"find-msvc-tools",
"jobserver",
@@ -4217,13 +4217,12 @@ checksum = "34aa73646ffb006b8f5147f3dc182bd4bcb190227ce861fc4a4844bf8e3cb2c0"
[[package]]
name = "encoding_rs"
version = "0.8.41"
version = "0.8.42"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7b5ef0006ac9ab233c38522f5ae99cae3625151de8f706cacee1cba4b8e2832a"
checksum = "8e985e0451871ad22fb8d2b6b076e2028a502a0d3950998c2c5c0a4f9b5d9679"
dependencies = [
"cfg-if",
"core_detect",
"multiversion",
"multiversion_no_op",
"rustversion",
"scopeguard",
@@ -4406,9 +4405,9 @@ dependencies = [
[[package]]
name = "find-msvc-tools"
version = "0.1.13"
version = "0.1.14"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ef25905e51abafe4dcea6c15fec58c57b601cdbd0ee53d22ea1d3016c587d39b"
checksum = "aedcfb3409746eddb02b9e19ebda1c3394f759a152e48ee875a0844d1b955484"
[[package]]
name = "findshlibs"
@@ -4756,9 +4755,9 @@ dependencies = [
[[package]]
name = "google-cloud-auth"
version = "1.16.0"
version = "1.17.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ff461519b1a948200f163574be072753bcfb462a323f0eb426629d89872dd685"
checksum = "ac9870b8ea484519aa8601e14886f2efc148ea490034e43d04bef1d246170cc9"
dependencies = [
"async-trait",
"aws-lc-rs",
@@ -4785,9 +4784,9 @@ dependencies = [
[[package]]
name = "google-cloud-gax"
version = "1.14.0"
version = "1.15.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c5615cff28ee59cfe52fbb4c11b8b1e77f650296e2ea4f4c2b7757ac6b19e752"
checksum = "fc64ee071c0d22cb68662b454f1bea5424506cafeaddc51260fbc585bac1d29d"
dependencies = [
"bytes",
"futures",
@@ -4805,9 +4804,9 @@ dependencies = [
[[package]]
name = "google-cloud-gax-internal"
version = "0.7.18"
version = "0.7.19"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d2766757d877a7a8ac23da9884cb0e3f10ed9b75a0ce59801ce6b19bf9d5819e"
checksum = "c47c73e95dc5e1de74bebe21b406394b53ff5ccfdfe0e81a0443cdef468e6744"
dependencies = [
"bytes",
"futures",
@@ -4821,9 +4820,9 @@ dependencies = [
"http-body-util",
"hyper",
"lazy_static",
"opentelemetry",
"opentelemetry-semantic-conventions",
"opentelemetry_sdk",
"opentelemetry 0.32.0",
"opentelemetry-semantic-conventions 0.32.1",
"opentelemetry_sdk 0.32.1",
"percent-encoding",
"pin-project",
"prost 0.14.4",
@@ -4839,14 +4838,14 @@ dependencies = [
"tonic-prost",
"tower",
"tracing",
"tracing-opentelemetry",
"tracing-opentelemetry 0.33.0",
]
[[package]]
name = "google-cloud-iam-v1"
version = "1.12.0"
version = "1.13.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5f962b40234b1531e6ef73f7558871c96e117231e962086c98804323fb8d2c82"
checksum = "48f0895fa7b7bd70629ae58daaa962677e93a72c251b2f297bc058a035940821"
dependencies = [
"async-trait",
"bytes",
@@ -4862,9 +4861,9 @@ dependencies = [
[[package]]
name = "google-cloud-longrunning"
version = "1.13.0"
version = "1.14.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1c0363c5389ffda2b55cd8a86eef4b19a3481a48467c91dc5d77f626a9572766"
checksum = "e8ea252eee22a4e2187025ea6789c682a0d791a9748ea99de2abec3026a32cb4"
dependencies = [
"async-trait",
"bytes",
@@ -4880,9 +4879,9 @@ dependencies = [
[[package]]
name = "google-cloud-lro"
version = "1.10.0"
version = "1.11.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "47af3deef75c14a2983c430898d960c765bddbcc9f9188ca0563108e9227cfe7"
checksum = "dfb938b40ede9ef70c24f49c2687d84f289c9e9104ddfece739985e4b4844b8f"
dependencies = [
"google-cloud-gax",
"google-cloud-gax-internal",
@@ -4909,9 +4908,9 @@ dependencies = [
[[package]]
name = "google-cloud-storage"
version = "1.18.0"
version = "1.19.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "973399251b245c63f1d02d0768772833dcf1372ee358f593fb9159de8fe9c7d4"
checksum = "5f7fa1e66d6b47220dbfc5f243b634136e5b8d14d64cb13419bd59f976681da3"
dependencies = [
"async-trait",
"base64 0.23.1",
@@ -4962,11 +4961,11 @@ dependencies = [
[[package]]
name = "google-cloud-wkt"
version = "1.7.0"
version = "1.8.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7fccf98cfd5481a5f5a285181ab0c62123d7d47cd2bb7299448440649349e4e7"
checksum = "ad4b36fefda4485f0922de7a58a3e562e984d513442a1aae2281dce25124be2e"
dependencies = [
"base64 0.22.1",
"base64 0.23.1",
"bytes",
"serde",
"serde_json",
@@ -5096,9 +5095,9 @@ dependencies = [
[[package]]
name = "headers"
version = "0.4.1"
version = "0.4.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b3314d5adb5d94bcdf56771f2e50dbbc80bb4bdf88967526706205ac9eff24eb"
checksum = "fc2700e3fe555c3310aa7286cac6167449f2c87e0eb58769c9208a1c58a1d106"
dependencies = [
"base64 0.22.1",
"bytes",
@@ -5283,9 +5282,9 @@ dependencies = [
[[package]]
name = "hotpath"
version = "0.26.0"
version = "0.26.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f37234f85a8d1bda6b84d486245dd0847c5b33cca10d5f3691375e66704de0f8"
checksum = "036ff36fa1c4ae8dfb686606537eaa15b5d5da0c51bf94abeaf9dfc6ee54cac7"
dependencies = [
"arc-swap",
"async-channel",
@@ -5297,6 +5296,7 @@ dependencies = [
"futures-core",
"futures-sink",
"hdrhistogram",
"hotpath-drain",
"hotpath-macros",
"http 1.5.0",
"libc",
@@ -5315,10 +5315,16 @@ dependencies = [
]
[[package]]
name = "hotpath-macros"
version = "0.26.0"
name = "hotpath-drain"
version = "0.0.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2ac39b5517f540cab511bdbd5a4dc8cfc62858d6cd26a095463c1d2acfb3f290"
checksum = "e149b660b611650bdf3307941ef3e9b81be3c30ed1dfdbe7f670319674f023a6"
[[package]]
name = "hotpath-macros"
version = "0.26.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1ff68b15fdbcddfbad685f445597580d6fcbf28b49f01ff975e934351079e13b"
dependencies = [
"proc-macro2",
"quote",
@@ -5459,16 +5465,17 @@ dependencies = [
[[package]]
name = "hyper-util"
version = "0.1.20"
version = "0.1.21"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "96547c2556ec9d12fb1578c4eaf448b04993e7fb79cbaad930a656880a6bdfa0"
checksum = "ddc03d96684f9226b8a787cdb71488417b53ab5ea8fdb1dac946cb9431cc8bff"
dependencies = [
"base64 0.22.1",
"base64 0.23.1",
"bytes",
"futures-channel",
"futures-util",
"http 1.5.0",
"http-body",
"httparse",
"hyper",
"ipnet",
"libc",
@@ -5905,9 +5912,9 @@ dependencies = [
[[package]]
name = "js-sys"
version = "0.3.105"
version = "0.3.106"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ce57d20d1ea864ce2ac172ab472d409214f4fd359f0b2a2775abdf522e2af99e"
checksum = "7883d941dae510fb2d978fc3fe018c71c9e2892fd38854de3e8b92c2e5ad9cc5"
dependencies = [
"cfg-if",
"futures-util",
@@ -6202,9 +6209,9 @@ dependencies = [
[[package]]
name = "libunftp"
version = "0.23.0"
version = "0.23.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c9f8ff371890db2cf65a0758dba9a79f9cd965de369f6dbdc6581a22780af45e"
checksum = "3dc54288ecbf8beefa5cab3a9921ab03bf1f18a3a0872cbcdb5313fc0baebaeb"
dependencies = [
"async-trait",
"bitflags 2.13.2",
@@ -6213,7 +6220,7 @@ dependencies = [
"dashmap",
"derive_more",
"futures-util",
"getrandom 0.3.4",
"getrandom 0.4.3",
"lazy_static",
"moka",
"nix 0.30.1",
@@ -6699,27 +6706,6 @@ version = "0.10.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1d87ecb2933e8aeadb3e3a02b828fed80a7528047e68b4f424523a0981a3a084"
[[package]]
name = "multiversion"
version = "0.9.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b4ca4bea16ffc3f443cf7d866912118196bfef4c6a1556ca00f9f9b00bb43f7c"
dependencies = [
"multiversion-macros",
]
[[package]]
name = "multiversion-macros"
version = "0.9.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0d416831a7317ef4b08bee00b69cbbb9c8763da7959a7026244d6266869f9c83"
dependencies = [
"proc-macro2",
"quote",
"rustversion",
"syn 3.0.6",
]
[[package]]
name = "multiversion_no_op"
version = "1.0.0"
@@ -7221,12 +7207,21 @@ version = "0.39.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2e5a6c098c7a3b6547378093f5cc30bc54fd361ce711e05293a5cc589562739b"
dependencies = [
"crc32fast",
"flate2",
"memchr",
"ruzstd",
]
[[package]]
name = "object"
version = "0.40.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "dd229a0361b9d0d4396176e02d65897f487eebeab7caa6d443855ee152ca0b9c"
dependencies = [
"crc32fast",
"hashbrown 0.17.1",
"indexmap 2.14.2",
"memchr",
"ruzstd",
]
[[package]]
@@ -7338,12 +7333,26 @@ dependencies = [
]
[[package]]
name = "opentelemetry-appender-tracing"
version = "0.32.0"
name = "opentelemetry"
version = "0.33.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2c0080f0dc1d7c786f467cd85a4e395fcab11ee852004f39a29a18ab7c25d837"
checksum = "6cdb0b1b267eb9db3331b434ed9ddab10d50e280a9adf9d13e5233e2002b61b5"
dependencies = [
"opentelemetry",
"futures-core",
"futures-sink",
"js-sys",
"pin-project-lite",
"thiserror 2.0.21",
"tracing",
]
[[package]]
name = "opentelemetry-appender-tracing"
version = "0.33.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e2b304e8d8f11326ba366a99ca5c7aaf7cd397b67268a332437175aeb42a40db"
dependencies = [
"opentelemetry 0.33.0",
"tracing",
"tracing-core",
"tracing-log",
@@ -7352,29 +7361,30 @@ dependencies = [
[[package]]
name = "opentelemetry-http"
version = "0.32.0"
version = "0.33.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5683015d09e2df236ef005b17f6f196f0d5f6313c4fa43a7b6a53b52776e4331"
checksum = "ee2c3625b8aa04209f7e01e513bc8044da9687fa38a9eacf59628ca5f3b87300"
dependencies = [
"async-trait",
"bytes",
"http 1.5.0",
"opentelemetry",
"opentelemetry 0.33.0",
"reqwest",
]
[[package]]
name = "opentelemetry-otlp"
version = "0.32.0"
version = "0.33.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9966929966d17620d7c316c643ba62631826e10021409357772d5eea84f62c35"
checksum = "699a67345e21962a231955b9059a157e428b219fa5df8efe11f08346fb23a344"
dependencies = [
"flate2",
"http 1.5.0",
"opentelemetry",
"httpdate",
"opentelemetry 0.33.0",
"opentelemetry-http",
"opentelemetry-proto",
"opentelemetry_sdk",
"opentelemetry_sdk 0.33.0",
"prost 0.14.4",
"reqwest",
"thiserror 2.0.21",
@@ -7382,12 +7392,12 @@ dependencies = [
[[package]]
name = "opentelemetry-proto"
version = "0.32.0"
version = "0.33.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "56d658ba1faf63f7b9c492cfbe6e0ec365440a16132d3270c1065f7b33f1b638"
checksum = "25da1ac11a0aeccf38d7f77ee0348715adaf8340f65ad46c94a02c6b20e2f65d"
dependencies = [
"opentelemetry",
"opentelemetry_sdk",
"opentelemetry 0.33.0",
"opentelemetry_sdk 0.33.0",
"prost 0.14.4",
]
@@ -7398,14 +7408,20 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c913ac17a6c451661ee255f4625d143e51647ae78ebd969b75e41c4442f4fe47"
[[package]]
name = "opentelemetry-stdout"
version = "0.32.0"
name = "opentelemetry-semantic-conventions"
version = "0.33.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a1b1c6a247d79091f0062a5f4bd058589525cf987a8d4c169440d9c1be72f0ad"
checksum = "d288b1e1bc3590052fbb8f0f635ef248dd9022cf1fbaeb6582b7ad6081a17b87"
[[package]]
name = "opentelemetry-stdout"
version = "0.33.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9f71803f481d3f5ccbe477edb42a78f473613712b2327835911a9ea0ab069824"
dependencies = [
"chrono",
"opentelemetry",
"opentelemetry_sdk",
"opentelemetry 0.33.0",
"opentelemetry_sdk 0.33.0",
]
[[package]]
@@ -7417,7 +7433,23 @@ dependencies = [
"futures-channel",
"futures-executor",
"futures-util",
"opentelemetry",
"opentelemetry 0.32.0",
"percent-encoding",
"portable-atomic",
"rand 0.9.5",
"thiserror 2.0.21",
]
[[package]]
name = "opentelemetry_sdk"
version = "0.33.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cb39533d9d1c912123efd7d41d7e0c29d16917b60ce15b4c8d87cb1af7f67520"
dependencies = [
"futures-channel",
"futures-executor",
"futures-util",
"opentelemetry 0.33.0",
"percent-encoding",
"portable-atomic",
"rand 0.9.5",
@@ -8946,9 +8978,9 @@ dependencies = [
[[package]]
name = "redis"
version = "1.7.0"
version = "1.7.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2acbc41a996f7652b2ddd9dfd98cc4ff602cfd742ae35382f07f608405ab50ed"
checksum = "ed5b98ae99461e02895c3a96a25dbe4d852a3d5e5b815d31d5d9e4b5344d99e9"
dependencies = [
"arc-swap",
"arcstr",
@@ -9521,9 +9553,9 @@ dependencies = [
"metrics-util",
"mime_guess",
"moka",
"opentelemetry",
"opentelemetry 0.33.0",
"opentelemetry-proto",
"opentelemetry_sdk",
"opentelemetry_sdk 0.33.0",
"p256 0.14.0",
"parking_lot",
"percent-encoding",
@@ -9613,7 +9645,7 @@ dependencies = [
"tower",
"tower-http 0.7.1",
"tracing",
"tracing-opentelemetry",
"tracing-opentelemetry 0.34.0",
"tracing-subscriber",
"url",
"urlencoding",
@@ -9705,7 +9737,7 @@ name = "rustfs-connect-symbol-catalog"
version = "1.0.1"
dependencies = [
"hex-simd",
"object 0.39.1",
"object 0.40.0",
"serde",
"serde_json",
"sha2 0.11.0",
@@ -9803,8 +9835,8 @@ dependencies = [
"metrics-util",
"moka",
"num_cpus",
"opentelemetry",
"opentelemetry_sdk",
"opentelemetry 0.33.0",
"opentelemetry_sdk 0.33.0",
"parking_lot",
"path-absolutize",
"pin-project-lite",
@@ -9865,7 +9897,7 @@ dependencies = [
"tower",
"tracing",
"tracing-core",
"tracing-opentelemetry",
"tracing-opentelemetry 0.34.0",
"tracing-subscriber",
"url",
"urlencoding",
@@ -9964,6 +9996,7 @@ dependencies = [
"thiserror 2.0.21",
"tokio",
"tokio-util",
"tonic",
"tracing",
"uuid",
"walkdir",
@@ -10381,12 +10414,12 @@ dependencies = [
"metrics-util",
"num_cpus",
"nvml-wrapper",
"opentelemetry",
"opentelemetry 0.33.0",
"opentelemetry-appender-tracing",
"opentelemetry-otlp",
"opentelemetry-semantic-conventions",
"opentelemetry-semantic-conventions 0.33.0",
"opentelemetry-stdout",
"opentelemetry_sdk",
"opentelemetry_sdk 0.33.0",
"percent-encoding",
"pyroscope",
"rcgen",
@@ -10418,7 +10451,7 @@ dependencies = [
"tracing",
"tracing-appender",
"tracing-error",
"tracing-opentelemetry",
"tracing-opentelemetry 0.34.0",
"tracing-subscriber",
"url",
"zstd 0.14.0",
@@ -11172,9 +11205,9 @@ dependencies = [
[[package]]
name = "rustls-platform-verifier"
version = "0.7.0"
version = "0.7.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "26d1e2536ce4f35f4846aa13bff16bd0ff40157cdb14cc056c7b14ba41233ba0"
checksum = "1167586491e2b18b8bfbb293e8180ec17c201c4f076d7cb3070ca964e7598f98"
dependencies = [
"core-foundation 0.10.1",
"core-foundation-sys",
@@ -11193,9 +11226,9 @@ dependencies = [
[[package]]
name = "rustls-platform-verifier-android"
version = "0.1.1"
version = "0.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f87165f0995f63a9fbeea62b64d10b4d9d8e78ec6d7d51fb2125fda7bb36788f"
checksum = "eec689c0bc40ff2458a5977b6619cb718087084a18e02a131c599b62d05e1a5f"
[[package]]
name = "rustls-webpki"
@@ -11656,9 +11689,9 @@ dependencies = [
[[package]]
name = "serde_with"
version = "3.23.0"
version = "3.24.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "935177bb8c0cd8ca1a4e6d1a2ac8988bea69cab4f9d3a31311e012ad27868ea4"
checksum = "df9adc193c780ef8f159aee8b61e2d5801aaa555e6eb0947fe45530ec506296f"
dependencies = [
"base64 0.23.1",
"bs58",
@@ -11677,9 +11710,9 @@ dependencies = [
[[package]]
name = "serde_with_macros"
version = "3.23.0"
version = "3.24.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1d607aa01a3cb0ad757d6fd216136910db3c97b102fe686585689615a02dbcdc"
checksum = "3e17bbc68e28663bbbb90df47e058aa7eda4fb445b89fe70457bb94fbccf6e49"
dependencies = [
"darling 0.24.1",
"proc-macro2",
@@ -11914,9 +11947,9 @@ dependencies = [
[[package]]
name = "siphasher"
version = "1.0.3"
version = "1.0.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8ee5873ec9cce0195efcb7a4e9507a04cd49aec9c83d0389df45b1ef7ba2e649"
checksum = "33f4fe9184a62d842c9ef383018f3306d8ba224fd9d836f56d7288308847c256"
[[package]]
name = "sketches-ddsketch"
@@ -11966,9 +11999,9 @@ dependencies = [
[[package]]
name = "smallvec"
version = "1.16.1"
version = "1.16.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ba467056f1b547ed52077911161fc86985becbc60e8e1857c8a144dab0def891"
checksum = "f9395f0f0eee849a9b707b2f06bb92a6a422090e2123bb2ef8e87a0e61892a8e"
dependencies = [
"serde",
]
@@ -12248,9 +12281,9 @@ checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292"
[[package]]
name = "suppaftp"
version = "12.0.1"
version = "12.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ba93b140f4d07f1b3f3fb4d7264b2d422d5c8b4e925eb6a2646e1342d5d2070a"
checksum = "d79f066c76927a775d61b5642e0b712a17582ebe7f62b7e5abba246e619878f7"
dependencies = [
"async-trait",
"chrono",
@@ -12763,9 +12796,9 @@ dependencies = [
[[package]]
name = "tokio-test"
version = "0.4.5"
version = "0.4.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3f6d24790a10a7af737693a3e8f1d03faef7e6ca0cc99aae5066f533766de545"
checksum = "12bc5db7778a8e08ee255090861e0e7aac55e784300077e7ed293c7be36168b2"
dependencies = [
"futures-core",
"tokio",
@@ -13058,7 +13091,21 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "adbc64cba7137545b8044cb1fe9814f7aacf3c6b5f9b45be8bb5db538befdb26"
dependencies = [
"js-sys",
"opentelemetry",
"opentelemetry 0.32.0",
"tracing",
"tracing-core",
"tracing-subscriber",
"web-time",
]
[[package]]
name = "tracing-opentelemetry"
version = "0.34.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0a904802a1b902f43638b677ff2a650847e3b4404101b6c586d648e8c1e3e8fe"
dependencies = [
"js-sys",
"opentelemetry 0.33.0",
"smallvec",
"tracing",
"tracing-core",
@@ -13416,9 +13463,9 @@ dependencies = [
[[package]]
name = "wasm-bindgen"
version = "0.2.128"
version = "0.2.129"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "aecb87a33d3b0c5e3b7aa46336eaf486cffafbd281b195e4c8b80d50df2351bf"
checksum = "9bb54f33acc68fd454578d9820b0bde1a1a3d17aa17bb7b6595806d02886d409"
dependencies = [
"cfg-if",
"once_cell",
@@ -13429,19 +13476,20 @@ dependencies = [
[[package]]
name = "wasm-bindgen-futures"
version = "0.4.78"
version = "0.4.79"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6ef4c5d3d2cdf5c54f4231181768f5510842e350db025faf1f7163b1030ed928"
checksum = "3cbab34de2d982e9b48e18d216d04c4a6f641066ff19ffb699980f591ee3610e"
dependencies = [
"js-sys",
"tokio",
"wasm-bindgen",
]
[[package]]
name = "wasm-bindgen-macro"
version = "0.2.128"
version = "0.2.129"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a690d511e3c1a8b3a55e33511e3c2c00c78415cd23650f32b808627f5696b9ed"
checksum = "2e29d0c35b16e224a7eeb5cd2d25e3e1968fbd65604117b44d3b789d00ee8535"
dependencies = [
"quote",
"wasm-bindgen-macro-support",
@@ -13449,9 +13497,9 @@ dependencies = [
[[package]]
name = "wasm-bindgen-macro-support"
version = "0.2.128"
version = "0.2.129"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "411e4887f0071ef2d2164a9d5fdf2d20efbef78fccd3a78b0c10a1dc5295e48a"
checksum = "6f501a8bc3719dba86ef8ae4728879c08001bea749eb1333ac5b91e040e2a6b7"
dependencies = [
"bumpalo",
"proc-macro2",
@@ -13462,9 +13510,9 @@ dependencies = [
[[package]]
name = "wasm-bindgen-shared"
version = "0.2.128"
version = "0.2.129"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "81941cd78d0c92026c33e5e01312845a4cb1e9af3407f9134b100dd03144103e"
checksum = "23f0c9c52aa7cd7d77769a4cfe2a9adb1b331f489a41d912ce14513d5ab995c6"
dependencies = [
"unicode-ident",
]
@@ -13484,9 +13532,9 @@ dependencies = [
[[package]]
name = "web-sys"
version = "0.3.105"
version = "0.3.106"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9fbddc4a036f00ec4f18c83445bd3115cb306a91da554919a099d9222fe4a7f8"
checksum = "88261b9deccee56594c11a3460c462c41f58d148598fe70ad77070126a68aba4"
dependencies = [
"js-sys",
"wasm-bindgen",
@@ -14037,18 +14085,18 @@ dependencies = [
[[package]]
name = "zerocopy"
version = "0.8.58"
version = "0.8.59"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c17e8fafad82b542ff3717217ecdc736231b59e387768c9630123b4ce4d2db44"
checksum = "6df92bf3d9227be3d53173901ddbffac2babc27ae50f397776ffd6dc33f800cb"
dependencies = [
"zerocopy-derive",
]
[[package]]
name = "zerocopy-derive"
version = "0.8.58"
version = "0.8.59"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "595f56e044df4f46a0c9a626f65c3d99eb8488f7e8a8baa12dd76326d9710bf2"
checksum = "ac4f328cf2f05d084e496c3e9c3f33ed0a183656a16e1fcec4d464d8373aec82"
dependencies = [
"proc-macro2",
"quote",
+24 -23
View File
@@ -160,7 +160,7 @@ pulsar = { default-features = false, version = "6.9.0" }
lapin = { default-features = false, version = "4.12.0" }
hyper = { version = "1.11.1" }
hyper-rustls = { default-features = false, version = "0.27.10" }
hyper-util = { version = "0.1.20" }
hyper-util = { version = "0.1.21" }
http = "1.5.0"
http-body = "1.1.0"
http-body-util = "0.1.5"
@@ -172,7 +172,7 @@ socket2 = { version = "0.6.5" }
tokio = { version = "1.53.1" }
tokio-rustls = { default-features = false, version = "0.26.5" }
tokio-stream = { version = "0.1.19" }
tokio-test = "0.4.5"
tokio-test = "0.4.6"
tokio-util = { version = "0.7.19" }
tonic = { version = "0.14.6" }
tonic-prost = { version = "0.14.6" }
@@ -245,13 +245,13 @@ atoi = "3.1.0"
atomic_enum = "0.3.0"
aws-config = { version = "1.12.0" }
aws-credential-types = { version = "1.3.0" }
aws-sdk-kms = { default-features = false, version = "1.121.0" }
aws-sdk-s3 = { default-features = false, version = "1.149.0" }
aws-sdk-sts = { default-features = false, version = "1.117.0" }
aws-sdk-kms = { default-features = false, version = "1.122.0" }
aws-sdk-s3 = { default-features = false, version = "1.150.0" }
aws-sdk-sts = { default-features = false, version = "1.118.0" }
aws-smithy-async = { version = "1.3.0" }
aws-smithy-http-client = { default-features = false, version = "1.4.2" }
aws-smithy-runtime-api = { version = "1.17.0" }
aws-smithy-types = { version = "1.8.0" }
aws-smithy-runtime-api = { version = "1.18.0" }
aws-smithy-types = { version = "1.8.1" }
base64-simd = "0.8.0"
brotli = "9.0.0"
clap = { version = "4.6.7" }
@@ -269,8 +269,8 @@ enumset = "1.1.14"
faster-hex = "1.0.0"
flate2 = "1.1.10"
glob = "0.3.4"
google-cloud-storage = "1.18.0"
google-cloud-auth = "1.16.0"
google-cloud-storage = "1.19.0"
google-cloud-auth = "1.17.0"
hashbrown = { version = "0.17.1" }
# Base32 for RFC 6238 TOTP shared secrets (RFC 4648 unpadded, the alphabet
# every authenticator app expects). Already in the graph transitively.
@@ -293,6 +293,7 @@ moka = { version = "0.12.16" }
netif = "0.1.6"
num_cpus = { version = "1.17.0" }
nvml-wrapper = "0.13.0"
object = { version = "=0.40.0", default-features = false }
parking_lot = "0.12.5"
path-absolutize = "4.0.1"
percent-encoding = "2.3.2"
@@ -309,7 +310,7 @@ rustfs-erasure-codec = { version = "9.0.0" }
reed-solomon-simd = "3.1.0"
regex = { version = "1.13.1" }
rumqttc = { package = "rumqttc-next", version = "0.34.0" }
redis = { version = "1.7.0" }
redis = { version = "1.7.1" }
rustify = { version = "0.7", default-features = false }
rustix = { version = "1.1.5" }
rust-embed = { version = "8.12.0" }
@@ -318,8 +319,8 @@ s3s = { version = "0.17.0", features = ["minio"] }
s3s-sigv4 = "0.17.0"
serial_test = "4.0.1"
shadow-rs = { default-features = false, version = "2.0.0" }
siphasher = "1.0.3"
smallvec = { version = "1.16.1" }
siphasher = "1.0.4"
smallvec = { version = "1.16.2" }
compact_str = "0.10.0"
snap = "1.1.2"
starshard = { version = "2.3.0" }
@@ -333,7 +334,7 @@ tracing = { version = "0.1.44" }
tracing-appender = "0.2.5"
tracing-core = "0.1.36"
tracing-error = "0.2.1"
tracing-opentelemetry = { version = "0.33" }
tracing-opentelemetry = { version = "0.34" }
tracing-subscriber = { version = "0.3.23" }
transform-stream = "0.3.1"
url = "2.5.8"
@@ -353,19 +354,19 @@ zstd = "0.14.0"
metrics = "0.24.6"
metrics-util = "0.20"
dial9 = { version = "0.5.2", default-features = false }
opentelemetry = { version = "0.32.0" }
opentelemetry-appender-tracing = { version = "0.32.0" }
opentelemetry-otlp = { version = "0.32.0" }
opentelemetry-proto = { version = "0.32.0", default-features = false, features = ["metrics", "gen-tonic-messages"] }
opentelemetry_sdk = { version = "0.32.1" }
opentelemetry-semantic-conventions = { version = "0.32.1" }
opentelemetry-stdout = { version = "0.32.0" }
opentelemetry = { version = "0.33.0" }
opentelemetry-appender-tracing = { version = "0.33.0" }
opentelemetry-otlp = { version = "0.33.0" }
opentelemetry-proto = { version = "0.33.0", default-features = false, features = ["metrics", "gen-tonic-messages"] }
opentelemetry_sdk = { version = "0.33.0" }
opentelemetry-semantic-conventions = { version = "0.33.0" }
opentelemetry-stdout = { version = "0.33.0" }
pyroscope = { version = "2.1.1" }
# FTP and SFTP
libunftp = { version = "0.23.0" }
libunftp = { version = "0.23.1" }
unftp-core = "0.1.0"
suppaftp = { version = "12.0.1" }
suppaftp = { version = "12.1.0" }
rcgen = { version = "0.14.10", default-features = false, features = ["aws_lc_rs", "crypto", "pem"] }
russh = { version = "0.63.3" }
russh-sftp = "3.0.0"
@@ -376,7 +377,7 @@ dav-server = "0.11.0"
# Performance Analysis and Memory Profiling
rustfs-mimalloc = { version = "0.5.6" }
# Preserve Unicode focus filters until rustfs/backlog#2302 is resolved.
hotpath = { version = "0.26.0", default-features = false }
hotpath = { version = "0.26.1", default-features = false }
# Snapshot testing for output format regression detection
insta = { version = "1.48" }
+3 -1
View File
@@ -106,7 +106,9 @@ RUN chmod +x /usr/bin/rustfs /entrypoint.sh
RUN addgroup -g 10001 -S rustfs && \
adduser -u 10001 -G rustfs -S rustfs -D && \
mkdir -p /data /logs && \
mkdir -p /data/.rustfs/events /data/.rustfs/audit /logs /opt/rustfs && \
ln -s /data/.rustfs/events /opt/rustfs/events && \
ln -s /data/.rustfs/audit /opt/rustfs/audit && \
chown -R rustfs:rustfs /data /logs && \
chmod 0750 /data /logs
+3 -1
View File
@@ -44,7 +44,9 @@ RUN set -eux; \
WORKDIR /app
RUN set -eux; \
mkdir -p /data /logs; \
mkdir -p /data/.rustfs/events /data/.rustfs/audit /logs /opt/rustfs; \
ln -s /data/.rustfs/events /opt/rustfs/events; \
ln -s /data/.rustfs/audit /opt/rustfs/audit; \
chown -R rustfs:rustfs /data /logs /app; \
chmod 0750 /data /logs
+3 -1
View File
@@ -110,7 +110,9 @@ RUN chmod +x /usr/bin/rustfs /entrypoint.sh
RUN groupadd -g 10001 rustfs && \
useradd -u 10001 -g rustfs -m -s /sbin/nologin rustfs && \
mkdir -p /data /logs && \
mkdir -p /data/.rustfs/events /data/.rustfs/audit /logs /opt/rustfs && \
ln -s /data/.rustfs/events /opt/rustfs/events && \
ln -s /data/.rustfs/audit /opt/rustfs/audit && \
chown -R rustfs:rustfs /data /logs && \
chmod 0750 /data /logs
+3 -1
View File
@@ -240,7 +240,9 @@ WORKDIR /app
# Prepare data/log directories with sane defaults
RUN set -eux; \
mkdir -p /data /logs; \
mkdir -p /data/.rustfs/events /data/.rustfs/audit /logs /opt/rustfs; \
ln -s /data/.rustfs/events /opt/rustfs/events; \
ln -s /data/.rustfs/audit /opt/rustfs/audit; \
chown -R rustfs:rustfs /data /logs /app; \
chmod 0750 /data /logs
+1 -1
View File
@@ -195,7 +195,7 @@ docker run -d --name rustfs -p 9000:9000 \
Notes:
- `RUSTFS_NOTIFY_ENABLE=true` enables the global notify module switch.
- For ARN `arn:rustfs:sqs::primary:webhook`, use instance-scoped env vars with `_PRIMARY`.
- If queue dir is omitted, default is `/opt/rustfs/events`; ensure it is writable by the container runtime user.
- If queue dir is omitted, the official image maps `/opt/rustfs/events` into the persistent `/data` volume so it is writable by the runtime user and pending events survive container recreation. Override `queue_dir` when using another deployment layout.
- `RUSTFS_NOTIFY_WEBHOOK_SKIP_TLS_VERIFY_PRIMARY` defaults to `false`; enabling it skips webhook TLS certificate verification, allows MITM attacks, and emits a startup warning. Prefer `RUSTFS_NOTIFY_WEBHOOK_CLIENT_CA_PRIMARY` for private CAs.
- Since `1.0.0-beta.11`, webhook endpoints on private or container networks
(`Docker Compose service names`, `host.docker.internal`, RFC 1918 addresses) are
+96
View File
@@ -163,6 +163,15 @@ pub struct MrfDurableRepairAnchor {
pub bucket_incarnation_id: Uuid,
}
/// A completed deep check proved that a legacy object is present and healthy,
/// but the format has no independent identity commitment that can discharge a
/// durable partial-write responsibility. Keep the journal entry and pause
/// automatic retries until an operator has repaired the source of truth.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MrfUnverifiedLegacyEvent {
pub anchor: MrfDurableRepairAnchor,
}
impl MrfDurableRepairAnchor {
/// Build a dischargeable anchor only when the caller supplies the storage
/// incarnation and the original ingress lease. Legacy replay records lack
@@ -782,6 +791,8 @@ const MRF_REPAIRED_EVENT_CAP: usize = 4096;
static MRF_REPAIRED_EVENTS: OnceLock<std::sync::Mutex<std::collections::VecDeque<MrfRepairedEvent>>> = OnceLock::new();
static MRF_VERIFIED_REPAIR_EVENTS: OnceLock<std::sync::Mutex<std::collections::VecDeque<MrfVerifiedRepairEvent>>> =
OnceLock::new();
static MRF_UNVERIFIED_LEGACY_EVENTS: OnceLock<std::sync::Mutex<std::collections::VecDeque<MrfUnverifiedLegacyEvent>>> =
OnceLock::new();
/// Record a legacy notification for compatibility. This is not an
/// acknowledgement of storage verification or durable repair completion.
@@ -859,6 +870,44 @@ pub fn take_mrf_verified_repair_events_for(bucket: &str) -> Vec<MrfVerifiedRepai
taken
}
/// Record a bounded, exact-identity notice that a legacy durable partial-write
/// was checked but cannot be proven from its on-disk format. This is not a
/// repair proof and must never release the durable responsibility.
pub fn note_mrf_unverified_legacy(event: MrfUnverifiedLegacyEvent) {
let registry = MRF_UNVERIFIED_LEGACY_EVENTS.get_or_init(|| std::sync::Mutex::new(std::collections::VecDeque::new()));
let Ok(mut events) = registry.lock() else {
return;
};
if events.contains(&event) {
return;
}
if events.len() >= MRF_REPAIRED_EVENT_CAP {
events.pop_front();
}
events.push_back(event);
}
/// Take held-legacy notices for one bucket. The durable ledger still verifies
/// the full anchor before changing the journal entry to its held state.
pub fn take_mrf_unverified_legacy_events_for(bucket: &str) -> Vec<MrfUnverifiedLegacyEvent> {
let Some(registry) = MRF_UNVERIFIED_LEGACY_EVENTS.get() else {
return Vec::new();
};
let Ok(mut events) = registry.lock() else {
return Vec::new();
};
let mut taken = Vec::new();
events.retain(|event| {
if event.anchor.bucket.as_ref() == bucket {
taken.push(event.clone());
false
} else {
true
}
});
taken
}
#[cfg(test)]
mod tests {
use super::*;
@@ -1233,6 +1282,53 @@ mod tests {
assert!(take_mrf_verified_repair_events_for("verified-bucket-a").is_empty());
assert_eq!(take_mrf_verified_repair_events_for("verified-bucket-b").len(), 1);
}
#[test]
fn unverified_legacy_events_are_bucket_scoped_and_bounded() {
let bucket = Arc::<str>::from("unverified-legacy-event-bucket");
let _ = take_mrf_unverified_legacy_events_for(bucket.as_ref());
let event = MrfUnverifiedLegacyEvent {
anchor: MrfDurableRepairAnchor {
kind: MrfKind::PartialWrite,
bucket: bucket.clone(),
object: Arc::from("unverified-object"),
version_id: None,
scope: Some(MrfScope {
pool_index: 1,
set_index: 2,
}),
delete_marker_purge: None,
lease: MrfIngressLease::new(912),
bucket_incarnation_id: Uuid::new_v4(),
},
};
note_mrf_unverified_legacy(event.clone());
note_mrf_unverified_legacy(event.clone());
assert_eq!(take_mrf_unverified_legacy_events_for(bucket.as_ref()), vec![event]);
assert!(take_mrf_unverified_legacy_events_for(bucket.as_ref()).is_empty());
let cap_bucket = Arc::<str>::from("unverified-legacy-event-cap-bucket");
for index in 0..MRF_REPAIRED_EVENT_CAP + 4 {
note_mrf_unverified_legacy(MrfUnverifiedLegacyEvent {
anchor: MrfDurableRepairAnchor {
kind: MrfKind::PartialWrite,
bucket: cap_bucket.clone(),
object: Arc::from(format!("object-{index}")),
version_id: None,
scope: Some(MrfScope {
pool_index: 1,
set_index: 2,
}),
delete_marker_purge: None,
lease: MrfIngressLease::new(u64::try_from(index).expect("test index fits")),
bucket_incarnation_id: Uuid::new_v4(),
},
});
}
let events = take_mrf_unverified_legacy_events_for(cap_bucket.as_ref());
assert_eq!(events.len(), MRF_REPAIRED_EVENT_CAP);
assert_eq!(events.first().expect("bounded ring is non-empty").anchor.object.as_ref(), "object-4");
}
}
#[tokio::test]
async fn partial_write_durable_admission_waits_for_persistence_and_renews_lease() {
+2 -2
View File
@@ -8,7 +8,7 @@ publish = false
[dependencies]
hex-simd.workspace = true
object = { version = "=0.39.1", default-features = false, features = ["read", "std", "elf"] }
object = { workspace = true, features = ["read", "std", "elf"] }
serde = { workspace = true, features = ["derive"] }
serde_json.workspace = true
sha2.workspace = true
@@ -16,7 +16,7 @@ sha2.workspace = true
symbolic-demangle = { version = "=13.9.0", default-features = false, features = ["rust"] }
[dev-dependencies]
object = { version = "=0.39.1", default-features = false, features = ["write", "elf"] }
object = { workspace = true, features = ["write", "elf"] }
[lints]
workspace = true
+3 -3
View File
@@ -384,17 +384,17 @@ pub(crate) fn is_cluster_heal_coordination_unavailable(error: &(dyn std::error::
message.contains("500 Internal Server Error") && message.contains("cluster heal coordination unavailable")
}
/// Wait for the restarted cluster to admit the first root heal request.
/// Wait for the restarted cluster to admit a root heal request and return its response.
pub(crate) async fn start_root_heal_when_control_ready(
heal_url: &str,
heal_body: &str,
access_key: &str,
secret_key: &str,
) -> ChaosResult<()> {
) -> ChaosResult<String> {
let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(45);
loop {
match signed_admin_post(heal_url, Some(heal_body), access_key, secret_key).await {
Ok(_) => return Ok(()),
Ok(body) => return Ok(body),
Err(error) if is_cluster_heal_coordination_unavailable(error.as_ref()) && tokio::time::Instant::now() < deadline => {
tokio::time::sleep(std::time::Duration::from_millis(250)).await;
}
+84 -3
View File
@@ -22,6 +22,8 @@
//! - Common test constants and utilities
use aws_sdk_s3::config::{Credentials, Region};
use aws_sdk_s3::error::{ProvideErrorMetadata, SdkError};
use aws_sdk_s3::operation::list_buckets::ListBucketsError;
use aws_sdk_s3::{Client, Config};
use aws_smithy_http_client::Builder as SmithyHttpClientBuilder;
use http::header::{CONTENT_TYPE, HOST};
@@ -61,6 +63,41 @@ const TEST_PORT_COUNTER_PATH: &str = "/tmp/rustfs_e2e_next_port";
const TEST_PORT_LOCK_DIR: &str = "/tmp/rustfs_e2e_port_allocator.lock";
const TEST_PORT_LOCK_STALE_AFTER: Duration = Duration::from_secs(30);
fn list_buckets_readiness_error(err: &SdkError<ListBucketsError>) -> String {
// SDK Display reports only the category. Do not expose raw response bodies or headers.
let service_code = err.as_service_error().and_then(|error| error.code()).filter(|code| {
matches!(
*code,
"ServiceUnavailable"
| "ServerNotInitialized"
| "InternalError"
| "AccessDenied"
| "InvalidAccessKeyId"
| "SignatureDoesNotMatch"
)
});
let transport = match err {
SdkError::DispatchFailure(failure) if failure.is_timeout() => Some("timeout"),
SdkError::DispatchFailure(failure) if failure.is_io() => Some("io"),
SdkError::DispatchFailure(failure) if failure.is_user() => Some("user"),
SdkError::DispatchFailure(failure) if failure.is_other() => Some("other"),
_ => None,
};
let mut source: Option<&(dyn std::error::Error + 'static)> = Some(err);
let mut io_error = None;
while let Some(error) = source {
if let Some(error) = error.downcast_ref::<std::io::Error>() {
io_error = Some((error.kind(), error.raw_os_error()));
break;
}
source = error.source();
}
format!(
"{err}; http_status={:?}; service_code={service_code:?}; transport={transport:?}; io_error={io_error:?}",
err.raw_response().map(|response| response.status().as_u16())
)
}
fn capture_log_path(log_dir: &Path, temp_dir: &str) -> Option<PathBuf> {
let temp_name = Path::new(temp_dir).file_name()?.to_string_lossy();
Some(log_dir.join(format!("{temp_name}.log")))
@@ -1575,8 +1612,9 @@ impl RustFSTestClusterEnvironment {
///
/// Verifies service availability by calling the S3 `list_buckets` API against the requested node,
/// retries up to 120 times with a 1-second interval between attempts.
async fn wait_for_node_service_ready(&self, node_idx: usize) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
async fn wait_for_node_service_ready(&mut self, node_idx: usize) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
let client = self.create_s3_client(node_idx)?;
let mut last_error = None;
for attempt in 0..120 {
match client.list_buckets().send().await {
@@ -1584,13 +1622,28 @@ impl RustFSTestClusterEnvironment {
info!("Cluster node {} service ready after {} attempts", node_idx, attempt + 1);
return Ok(());
}
Err(_) => {
Err(err) => {
last_error = Some(err);
sleep(Duration::from_secs(1)).await;
}
}
}
Err(format!("Cluster node {} service failed to become ready", node_idx).into())
let last_error = last_error.as_ref().map(list_buckets_readiness_error);
// Observe liveness only after the existing readiness budget is exhausted, before cleanup.
let process_status = match self.nodes[node_idx].process.as_mut() {
Some(process) => match process.try_wait() {
Ok(Some(status)) => format!("exited ({status})"),
Ok(None) => "running".to_string(),
Err(error) => format!("unavailable (kind={:?}, os_code={:?})", error.kind(), error.raw_os_error()),
},
None => "not started".to_string(),
};
Err(format!(
"Cluster node {node_idx} service failed to become ready; last ListBuckets error={last_error:?}; process_status={process_status}; capture_log_path={:?}",
self.node_capture_log_paths[node_idx]
)
.into())
}
/// Create an S3 client configured to communicate with a specific cluster node.
@@ -2101,6 +2154,34 @@ mod tests {
verify_awscurl_path(&executable).expect("an available client with a working help command should pass");
}
#[test]
fn readiness_error_reports_transport_without_sensitive_source_text() {
use aws_sdk_s3::error::ConnectorError;
let sensitive = "request credentials must not appear in diagnostics";
for (connector, category) in [
(
ConnectorError::io(std::io::Error::new(ErrorKind::ConnectionReset, sensitive).into()),
"io",
),
(ConnectorError::timeout(sensitive.into()), "timeout"),
(ConnectorError::user(sensitive.into()), "user"),
(ConnectorError::other(sensitive.into(), None), "other"),
] {
let diagnostic = list_buckets_readiness_error(&SdkError::dispatch_failure(connector));
assert!(diagnostic.contains(&format!("transport=Some(\"{category}\")")), "{diagnostic}");
assert!(diagnostic.contains("http_status=None"), "{diagnostic}");
assert!(!diagnostic.contains(sensitive), "{diagnostic}");
if category == "io" {
assert!(diagnostic.contains("io_error=Some((ConnectionReset, None))"), "{diagnostic}");
}
}
let os_error = std::io::Error::from_raw_os_error(13);
let expected = format!("io_error=Some(({:?}, Some(13)))", os_error.kind());
let diagnostic = list_buckets_readiness_error(&SdkError::dispatch_failure(ConnectorError::io(os_error.into())));
assert!(diagnostic.contains(&expected), "{diagnostic}");
}
#[test]
fn capture_log_path_uses_temp_directory_basename() {
assert_eq!(
@@ -282,7 +282,7 @@ async fn sdk_list_bucket_and_list_bucket_versions_permissions_are_independent()
}
#[tokio::test]
async fn console_admin_force_delete_removes_prefix_versions_and_delete_markers() -> TestResult {
async fn explicit_force_delete_permission_removes_prefix_versions_and_delete_markers() -> TestResult {
init_logging();
let mut env = RustFSTestEnvironment::new().await?;
env.start_rustfs_server(vec![]).await?;
@@ -300,7 +300,14 @@ async fn console_admin_force_delete_removes_prefix_versions_and_delete_markers()
put(&root, bucket, "folder-sibling/keep.txt").await?;
let keep = versions(&root, bucket, "keep.txt").await?;
let sibling = versions(&root, bucket, "folder-sibling/").await?;
let user = policy_user(&env, "consoleAdmin", None).await?;
let user = policy_user(
&env,
"explicit-force-delete",
Some(json!({"Version":"2012-10-17","Statement":[
{"Effect":"Allow","Action":["s3:DeleteObject","s3:DeleteObjectVersion","s3:ForceDeleteObject"],"Resource":format!("arn:aws:s3:::{bucket}/*")}
]})),
)
.await?;
force_delete(&user, bucket, "folder/").await?;
assert!(
@@ -322,6 +329,126 @@ async fn console_admin_force_delete_removes_prefix_versions_and_delete_markers()
Ok(())
}
#[tokio::test]
async fn force_delete_header_requires_explicit_permission() -> TestResult {
init_logging();
let mut env = RustFSTestEnvironment::new().await?;
env.start_rustfs_server(vec![]).await?;
let root = env.create_s3_client();
let admin = policy_user(&env, "consoleAdmin", None).await?;
let owner_bucket = "force-owner-bucket";
root.create_bucket().bucket(owner_bucket).send().await?;
put(&root, owner_bucket, "kept-until-force.txt").await?;
force_delete_bucket(&root, owner_bucket, "x-rustfs-force-delete").await?;
assert!(
head_bucket_missing(&root, owner_bucket).await?,
"root force-delete must remove a non-empty bucket"
);
let denied_bucket = "force-admin-denied";
root.create_bucket().bucket(denied_bucket).send().await?;
put(&root, denied_bucket, "folder/child.txt").await?;
put(&root, denied_bucket, "plain.txt").await?;
assert_boxed_denied(force_delete(&admin, denied_bucket, "folder/").await);
assert!(
!versions(&root, denied_bucket, "folder/").await?.is_empty(),
"consoleAdmin without s3:ForceDeleteObject must not recursively delete"
);
assert_boxed_denied(force_delete_bucket(&admin, denied_bucket, "x-rustfs-force-delete").await);
assert_boxed_denied(force_delete_bucket(&admin, denied_bucket, "x-minio-force-delete").await);
assert!(
!head_bucket_missing(&root, denied_bucket).await?,
"consoleAdmin without s3:ForceDeleteBucket must not force-delete a non-empty bucket"
);
let not_empty = admin.delete_bucket().bucket(denied_bucket).send().await;
let not_empty = not_empty.expect_err("plain delete of a non-empty bucket must fail");
assert_eq!(
not_empty.as_service_error().and_then(|error| error.code()),
Some("BucketNotEmpty"),
"plain delete must keep the emptiness check, got {not_empty:?}"
);
admin.delete_object().bucket(denied_bucket).key("plain.txt").send().await?;
assert!(
versions(&root, denied_bucket, "plain.txt").await?.is_empty(),
"plain DeleteObject without the force header must stay allowed for consoleAdmin"
);
let object_bucket = "force-explicit-object";
root.create_bucket().bucket(object_bucket).send().await?;
put(&root, object_bucket, "folder/child.txt").await?;
put(&root, object_bucket, "keep.txt").await?;
let keep = versions(&root, object_bucket, "keep.txt").await?;
let object_user = policy_user(
&env,
"explicit-object-force",
Some(json!({"Version":"2012-10-17","Statement":[
{"Effect":"Allow","Action":["s3:DeleteObject","s3:DeleteObjectVersion","s3:ForceDeleteObject"],"Resource":format!("arn:aws:s3:::{object_bucket}/*")}
]})),
)
.await?;
force_delete(&object_user, object_bucket, "folder/").await?;
assert!(versions(&root, object_bucket, "folder/").await?.is_empty());
assert_eq!(versions(&root, object_bucket, "keep.txt").await?, keep);
let allowed_bucket = "force-explicit-bucket";
root.create_bucket().bucket(allowed_bucket).send().await?;
put(&root, allowed_bucket, "inside.txt").await?;
let bucket_user = policy_user(
&env,
"explicit-bucket-force",
Some(json!({"Version":"2012-10-17","Statement":[
{"Effect":"Allow","Action":["s3:DeleteBucket","s3:ForceDeleteBucket"],"Resource":[
format!("arn:aws:s3:::{allowed_bucket}"), format!("arn:aws:s3:::{allowed_bucket}/*")
]}
]})),
)
.await?;
force_delete_bucket(&bucket_user, allowed_bucket, "x-minio-force-delete").await?;
assert!(
head_bucket_missing(&root, allowed_bucket).await?,
"an explicit s3:ForceDeleteBucket grant must remove a non-empty bucket"
);
Ok(())
}
async fn force_delete_bucket(
client: &Client,
bucket: &str,
header: &str,
) -> Result<(), Box<SdkError<aws_sdk_s3::operation::delete_bucket::DeleteBucketError>>> {
let header: &'static str = match header {
"x-rustfs-force-delete" => "x-rustfs-force-delete",
"x-minio-force-delete" => "x-minio-force-delete",
other => panic!("unexpected force-delete header {other}"),
};
client
.delete_bucket()
.bucket(bucket)
.customize()
.mutate_request(move |request| {
request.headers_mut().insert(header, "true");
})
.send()
.await
.map(|_| ())
.map_err(Box::new)
}
async fn head_bucket_missing(client: &Client, bucket: &str) -> TestResult<bool> {
match client.head_bucket().bucket(bucket).send().await {
Ok(_) => Ok(false),
Err(error) => {
let code = error.as_service_error().and_then(|service| service.code());
if matches!(code, Some("NotFound" | "NoSuchBucket")) {
Ok(true)
} else {
Err(error.into())
}
}
}
}
#[tokio::test]
async fn force_delete_authorizes_only_its_path_scope_without_list_permissions() -> TestResult {
init_logging();
@@ -339,7 +466,7 @@ async fn force_delete_authorizes_only_its_path_scope_without_list_permissions()
&env,
"delete-only",
Some(json!({"Version":"2012-10-17","Statement":[
{"Effect":"Allow","Action":["s3:DeleteObject","s3:DeleteObjectVersion"],"Resource":[
{"Effect":"Allow","Action":["s3:DeleteObject","s3:DeleteObjectVersion","s3:ForceDeleteObject"],"Resource":[
format!("arn:aws:s3:::{bucket}/selected.txt"), format!("arn:aws:s3:::{bucket}/selected.txt/*")
]},
{"Effect":"Deny","Action":["s3:DeleteObject","s3:DeleteObjectVersion"],"Resource":format!("arn:aws:s3:::{bucket}/selected.txt-sibling")}
@@ -376,7 +503,7 @@ async fn force_directory_delete_cannot_remove_an_unauthorized_colliding_parent()
&env,
"parent-denier",
Some(json!({"Version":"2012-10-17","Statement":[
{"Effect":"Allow","Action":["s3:DeleteObject","s3:DeleteObjectVersion"],"Resource":format!("arn:aws:s3:::{bucket}/*")},
{"Effect":"Allow","Action":["s3:DeleteObject","s3:DeleteObjectVersion","s3:ForceDeleteObject"],"Resource":format!("arn:aws:s3:::{bucket}/*")},
{"Effect":"Deny","Action":"s3:DeleteObjectVersion","Resource":format!("arn:aws:s3:::{bucket}/collision.txt"),
"Condition":{"StringEquals":{"s3:VersionId":protected_parent_version}}}
]})),
@@ -409,7 +536,7 @@ async fn force_unversioned_directory_requires_only_delete_object() -> TestResult
&env,
"unversioned-deleter",
Some(json!({"Version":"2012-10-17","Statement":[
{"Effect":"Allow","Action":"s3:DeleteObject","Resource":format!("arn:aws:s3:::{bucket}/*")},
{"Effect":"Allow","Action":["s3:DeleteObject","s3:ForceDeleteObject"],"Resource":format!("arn:aws:s3:::{bucket}/*")},
{"Effect":"Deny","Action":"s3:DeleteObjectVersion","Resource":format!("arn:aws:s3:::{bucket}/*")}
]})),
)
@@ -438,7 +565,7 @@ async fn force_delete_denied_child_preserves_every_object_despite_bucket_allow()
&env,
"child-denier",
Some(json!({"Version":"2012-10-17","Statement":[
{"Effect":"Allow","Action":["s3:DeleteObject","s3:DeleteObjectVersion","s3:ReplicateDelete"],"Resource":format!("arn:aws:s3:::{bucket}/*")},
{"Effect":"Allow","Action":["s3:DeleteObject","s3:DeleteObjectVersion","s3:ForceDeleteObject","s3:ReplicateDelete"],"Resource":format!("arn:aws:s3:::{bucket}/*")},
{"Effect":"Deny","Action":["s3:DeleteObject","s3:DeleteObjectVersion","s3:ReplicateDelete"],"Resource":format!("arn:aws:s3:::{bucket}/folder/z-denied.txt")}
]})),
)
@@ -468,7 +595,7 @@ async fn force_delete_denied_child_preserves_every_object_despite_bucket_allow()
&env,
"replica-deleter",
Some(json!({"Version":"2012-10-17","Statement":[
{"Effect":"Allow","Action":"s3:DeleteObject","Resource":format!("arn:aws:s3:::{bucket}/*")},
{"Effect":"Allow","Action":["s3:DeleteObject","s3:ForceDeleteObject"],"Resource":format!("arn:aws:s3:::{bucket}/*")},
{"Effect":"Allow","Action":"s3:ReplicateDelete","Resource":format!("arn:aws:s3:::{bucket}/*")}
]})),
)
@@ -500,7 +627,7 @@ async fn force_delete_denied_historical_version_preserves_versions_and_markers()
put(&root, bucket, "folder/a-allowed.txt").await?;
let policy = |version: &str| {
json!({"Version":"2012-10-17","Statement":[
{"Effect":"Allow","Action":["s3:DeleteObject","s3:DeleteObjectVersion"],"Resource":format!("arn:aws:s3:::{bucket}/*")},
{"Effect":"Allow","Action":["s3:DeleteObject","s3:DeleteObjectVersion","s3:ForceDeleteObject"],"Resource":format!("arn:aws:s3:::{bucket}/*")},
{"Effect":"Deny","Action":"s3:DeleteObjectVersion","Resource":format!("arn:aws:s3:::{bucket}/folder/*"),
"Condition":{"StringEquals":{"s3:VersionId":version}}}
]})
@@ -628,7 +755,7 @@ async fn force_delete_checks_every_version_page_before_mutation() -> TestResult
.try_collect::<Vec<_>>()
.await?;
put(&root, bucket, "folder/z-denied.txt").await?;
let allow = json!({"Effect":"Allow","Action":["s3:DeleteObject","s3:DeleteObjectVersion"],"Resource":format!("arn:aws:s3:::{bucket}/*")});
let allow = json!({"Effect":"Allow","Action":["s3:DeleteObject","s3:DeleteObjectVersion","s3:ForceDeleteObject"],"Resource":format!("arn:aws:s3:::{bucket}/*")});
let user = policy_user(
&env,
"paged-deleter",
+219 -25
View File
@@ -34,6 +34,9 @@ use crate::common::{
};
use crate::replication_extension_test::LOOPBACK_REPLICATION_TARGET_ENV;
use aws_sdk_s3::Client;
use aws_sdk_s3::config::retry::RetryConfig;
use aws_sdk_s3::error::{ProvideErrorMetadata, SdkError};
use aws_sdk_s3::operation::{get_object::GetObjectError, put_object::PutObjectError};
use aws_sdk_s3::primitives::ByteStream;
use aws_sdk_s3::types::{BucketVersioningStatus, VersioningConfiguration};
use http::{Method, StatusCode};
@@ -41,7 +44,7 @@ use sha2::{Digest, Sha256};
use std::collections::BTreeMap;
use std::path::{Path, PathBuf};
use std::time::Duration;
use tokio::time::{Instant, sleep};
use tokio::time::{Instant, sleep, timeout_at};
use uuid::Uuid;
pub(crate) type TestResult<T = ()> = Result<T, Box<dyn std::error::Error + Send + Sync>>;
@@ -400,9 +403,9 @@ pub(crate) async fn put_inventory(
Ok(inventory)
}
/// Retry only transport-level service availability failures while a data
/// movement operation changes the pool map. Generic InternalError responses
/// remain fatal because accepting them would hide server defects.
/// Retry only explicit S3 availability responses while a data movement
/// operation changes the pool map. Transport failures and InternalError
/// responses remain fatal because callers already require a ready endpoint.
pub(crate) async fn put_inventory_retrying(
client: &Client,
bucket: &str,
@@ -449,19 +452,48 @@ where
let deadline = Instant::now() + timeout;
let mut delay = Duration::from_millis(50);
loop {
let last_error = match probe().await {
Ok(true) => return Ok(()),
Ok(false) => format!("{label} still false"),
Err(error) => error.to_string(),
};
if Instant::now() >= deadline {
return Err(format!("{label} did not become true within {timeout:?}: {last_error}").into());
return Err(format!("{label} did not become true within {timeout:?}").into());
}
sleep(delay).await;
// A pending request must consume the same budget as unsuccessful probes.
match timeout_at(deadline, probe()).await {
Err(_) => return Err(format!("{label} did not become true within {timeout:?}").into()),
Ok(Err(error)) => return Err(error),
Ok(Ok(true)) => return Ok(()),
Ok(Ok(false)) => {}
}
tokio::time::sleep_until((Instant::now() + delay).min(deadline)).await;
delay = (delay * 2).min(Duration::from_secs(1));
}
}
// Boxed SDK errors retain the operation's service code; Display only says
// "service error" and cannot distinguish convergence from a server defect.
pub(crate) fn s3_probe_error_code<'a>(error: &'a (dyn std::error::Error + Send + Sync + 'static)) -> Option<&'a str> {
error
.downcast_ref::<SdkError<GetObjectError>>()
.and_then(|error| error.as_service_error())
.and_then(ProvideErrorMetadata::code)
.or_else(|| {
error
.downcast_ref::<SdkError<PutObjectError>>()
.and_then(|error| error.as_service_error())
.and_then(ProvideErrorMetadata::code)
})
}
pub(crate) fn s3_probe_client(client: &Client) -> Client {
// The probe owns the retry policy. SDK retries must not hide a response
// that the probe would classify as fatal.
Client::from_conf(
client
.config()
.to_builder()
.retry_config(RetryConfig::standard().with_max_attempts(1))
.build(),
)
}
pub(crate) async fn cluster_admin(
cluster: &RustFSTestClusterEnvironment,
method: Method,
@@ -598,15 +630,20 @@ pub(crate) async fn wait_for_replicated_bytes(
expected: &[u8],
timeout: Duration,
) -> TestResult {
let client = s3_probe_client(client);
wait_until(
timeout,
|| async {
match get_object_bytes(client, bucket, key).await {
match get_object_bytes(&client, bucket, key).await {
Ok(got) if got.as_slice() == expected => Ok(true),
Ok(_) => Ok(false),
Ok(got) => Err(format!(
"replicated object {bucket}/{key} bytes mismatch: expected sha256={} got sha256={}",
sha256_hex(expected),
sha256_hex(&got)
)
.into()),
Err(error) => {
let message = error.to_string();
if message.contains("NoSuchKey") || message.contains("NotFound") {
if matches!(s3_probe_error_code(error.as_ref()), Some("NoSuchKey" | "NotFound")) {
Ok(false)
} else {
Err(error)
@@ -988,6 +1025,7 @@ pub(crate) async fn list_pools_json(cluster: &RustFSTestClusterEnvironment) -> T
}
pub(crate) async fn retrying_put(client: &Client, bucket: &str, key: &str, body: Vec<u8>, timeout: Duration) -> TestResult {
let client = s3_probe_client(client);
wait_until(
timeout,
|| {
@@ -999,8 +1037,7 @@ pub(crate) async fn retrying_put(client: &Client, bucket: &str, key: &str, body:
match put_object(&client, &bucket, &key, body).await {
Ok(()) => Ok(true),
Err(error) => {
let message = error.to_string();
if message.contains("SlowDown") || message.contains("ServiceUnavailable") || message.contains("503") {
if matches!(s3_probe_error_code(error.as_ref()), Some("SlowDown" | "ServiceUnavailable")) {
Ok(false)
} else {
Err(error)
@@ -1021,19 +1058,20 @@ pub(crate) async fn retrying_get_equals(
expected: &[u8],
timeout: Duration,
) -> TestResult {
let client = s3_probe_client(client);
wait_until(
timeout,
|| async {
match get_object_bytes(client, bucket, key).await {
match get_object_bytes(&client, bucket, key).await {
Ok(got) if got.as_slice() == expected => Ok(true),
Ok(_) => Ok(false),
Ok(got) => Err(format!(
"object {bucket}/{key} bytes mismatch: expected sha256={} got sha256={}",
sha256_hex(expected),
sha256_hex(&got)
)
.into()),
Err(error) => {
let message = error.to_string();
if message.contains("NoSuchKey")
|| message.contains("SlowDown")
|| message.contains("ServiceUnavailable")
|| message.contains("503")
{
if matches!(s3_probe_error_code(error.as_ref()), Some("SlowDown" | "ServiceUnavailable")) {
Ok(false)
} else {
Err(error)
@@ -1046,6 +1084,162 @@ pub(crate) async fn retrying_get_equals(
.await
}
#[cfg(test)]
mod retry_tests {
use super::*;
use crate::fake_s3_target::{FAKE_ACCESS_KEY, FAKE_SECRET_KEY, FakeS3Target, FaultAction, Operation, SeedMetadata};
use std::cell::Cell;
fn client(target: &FakeS3Target) -> Client {
Client::from_conf(build_test_s3_config(
target.endpoint(),
FAKE_ACCESS_KEY,
FAKE_SECRET_KEY,
None,
"distributed-retry-test",
))
}
#[tokio::test]
async fn wait_until_preserves_first_error() {
let attempts = Cell::new(0);
let error = wait_until(
Duration::from_secs(5),
|| {
attempts.set(attempts.get() + 1);
std::future::ready(if attempts.get() == 1 {
Err(std::io::Error::new(std::io::ErrorKind::PermissionDenied, "permanent probe failure").into())
} else {
Ok(true)
})
},
"permanent failure",
)
.await
.expect_err("a later success must not hide a permanent failure");
assert_eq!(attempts.get(), 1);
assert_eq!(
error
.downcast_ref::<std::io::Error>()
.expect("preserve the original error")
.kind(),
std::io::ErrorKind::PermissionDenied
);
}
#[tokio::test]
async fn wait_until_retries_explicit_pending_state() -> TestResult {
let attempts = Cell::new(0);
wait_until(
Duration::from_secs(5),
|| {
attempts.set(attempts.get() + 1);
std::future::ready(Ok(attempts.get() == 2))
},
"eventual readiness",
)
.await?;
assert_eq!(attempts.get(), 2);
Ok(())
}
#[tokio::test]
async fn wait_until_bounds_a_pending_probe() {
let error = tokio::time::timeout(
Duration::from_secs(5),
wait_until(Duration::from_millis(10), std::future::pending, "hung probe"),
)
.await
.expect("the probe deadline must finish before the test watchdog")
.expect_err("a permanently pending probe must time out");
assert!(error.to_string().contains("hung probe did not become true"));
}
#[tokio::test]
async fn wait_until_does_not_start_a_probe_after_its_deadline() {
let attempts = Cell::new(0);
wait_until(
Duration::ZERO,
|| {
attempts.set(attempts.get() + 1);
std::future::ready(Ok(true))
},
"expired budget",
)
.await
.expect_err("an expired budget must not admit another probe");
assert_eq!(attempts.get(), 0);
}
#[tokio::test]
async fn retrying_put_does_not_hide_internal_error() -> TestResult {
let target = FakeS3Target::start().await?;
target.create_bucket("retry-probe");
target.inject(Operation::PutObject, FaultAction::ResponseStatus(500), 1);
let error = retrying_put(&client(&target), "retry-probe", "key", b"body".to_vec(), Duration::from_secs(5))
.await
.expect_err("InternalError must remain fatal even if the next PUT would succeed");
assert_eq!(s3_probe_error_code(error.as_ref()), Some("InternalError"));
assert_eq!(target.requests().len(), 1);
Ok(())
}
#[tokio::test]
async fn retrying_put_allows_transient_availability_errors() -> TestResult {
let target = FakeS3Target::start().await?;
target.create_bucket("retry-probe");
for status in [429, 503] {
target.inject(Operation::PutObject, FaultAction::ResponseStatus(status), 1);
retrying_put(&client(&target), "retry-probe", "key", b"body".to_vec(), Duration::from_secs(5)).await?;
assert_eq!(target.take_requests().len(), 2);
}
Ok(())
}
#[tokio::test]
async fn object_polls_reject_wrong_object_bytes() -> TestResult {
let target = FakeS3Target::start().await?;
target.create_bucket("retry-probe");
target.put_seed_object("retry-probe", "key", "wrong bytes", &SeedMetadata::default());
let error = retrying_get_equals(&client(&target), "retry-probe", "key", b"expected bytes", Duration::from_secs(5))
.await
.expect_err("a successful response with wrong bytes must fail immediately");
assert!(error.to_string().contains("bytes mismatch"));
assert_eq!(target.take_requests().len(), 1);
let error = wait_for_replicated_bytes(&client(&target), "retry-probe", "key", b"expected bytes", Duration::from_secs(5))
.await
.expect_err("replication may lag, but it must not return corrupt bytes");
assert!(error.to_string().contains("bytes mismatch"));
assert_eq!(target.requests().len(), 1);
Ok(())
}
#[tokio::test]
async fn object_polls_distinguish_committed_and_replicated_keys() -> TestResult {
let target = FakeS3Target::start().await?;
target.create_bucket("retry-probe");
target.put_seed_object("retry-probe", "key", "expected bytes", &SeedMetadata::default());
let client = client(&target);
for (status, expected_code) in [(404, "NoSuchKey"), (500, "InternalError"), (403, "AccessDenied")] {
target.inject(Operation::GetObject, FaultAction::ResponseStatus(status), 1);
let error = retrying_get_equals(&client, "retry-probe", "key", b"expected bytes", Duration::from_secs(5))
.await
.expect_err("an acknowledged object must not disappear or fail before a later successful read");
assert_eq!(s3_probe_error_code(error.as_ref()), Some(expected_code));
assert_eq!(target.take_requests().len(), 1);
}
for status in [429, 503] {
target.inject(Operation::GetObject, FaultAction::ResponseStatus(status), 1);
retrying_get_equals(&client, "retry-probe", "key", b"expected bytes", Duration::from_secs(5)).await?;
assert_eq!(target.take_requests().len(), 2);
}
target.inject(Operation::GetObject, FaultAction::ResponseStatus(404), 1);
wait_for_replicated_bytes(&client, "retry-probe", "key", b"expected bytes", Duration::from_secs(5)).await?;
assert_eq!(target.requests().len(), 2);
Ok(())
}
}
#[tokio::test]
async fn append_single_node_pool_extends_ellipses_volumes() {
let mut env = RustFSTestClusterEnvironment::with_topology(ClusterTopology::per_node_pools(2, vec![vec![0], vec![1]]))
@@ -26,7 +26,7 @@
use super::harness::{
DistCluster, DistLayout, TestResult, assert_object_bytes, cluster_admin_ok, enable_versioning, get_object_bytes, put_object,
unique_bucket, wait_until,
s3_probe_client, s3_probe_error_code, sha256_hex, unique_bucket, wait_until,
};
use crate::common::{
AdminTransport, admin_add_canned_policy_via, admin_attach_user_policy_via, admin_create_user_via, init_logging,
@@ -111,6 +111,7 @@ async fn create_iam_user(dist: &DistCluster, user: &str, secret: &str, policy_na
}
async fn wait_for_put(client: &Client, bucket: &str, key: &str, body: Vec<u8>, label: &str) -> TestResult {
let client = s3_probe_client(client);
wait_until(
CREDENTIAL_TIMEOUT,
|| {
@@ -119,8 +120,18 @@ async fn wait_for_put(client: &Client, bucket: &str, key: &str, body: Vec<u8>, l
let key = key.to_string();
let body = body.clone();
async move {
put_object(&client, &bucket, &key, body).await?;
Ok(true)
match put_object(&client, &bucket, &key, body).await {
Ok(()) => Ok(true),
Err(error)
if matches!(
s3_probe_error_code(error.as_ref()),
Some("AccessDenied" | "InvalidAccessKeyId" | "SlowDown" | "ServiceUnavailable")
) =>
{
Ok(false)
}
Err(error) => Err(error),
}
}
},
label,
@@ -129,6 +140,7 @@ async fn wait_for_put(client: &Client, bucket: &str, key: &str, body: Vec<u8>, l
}
async fn wait_for_bytes(client: &Client, bucket: &str, key: &str, expected: &[u8], label: &str) -> TestResult {
let client = s3_probe_client(client);
wait_until(
CREDENTIAL_TIMEOUT,
|| {
@@ -137,8 +149,24 @@ async fn wait_for_bytes(client: &Client, bucket: &str, key: &str, expected: &[u8
let key = key.to_string();
let expected = expected.to_vec();
async move {
let got = get_object_bytes(&client, &bucket, &key).await?;
Ok(got == expected)
match get_object_bytes(&client, &bucket, &key).await {
Ok(got) if got == expected => Ok(true),
Ok(got) => Err(format!(
"{label}: object {bucket}/{key} bytes mismatch: expected sha256={} got sha256={}",
sha256_hex(&expected),
sha256_hex(&got)
)
.into()),
Err(error)
if matches!(
s3_probe_error_code(error.as_ref()),
Some("AccessDenied" | "InvalidAccessKeyId" | "SlowDown" | "ServiceUnavailable")
) =>
{
Ok(false)
}
Err(error) => Err(error),
}
}
},
label,
@@ -146,6 +174,57 @@ async fn wait_for_bytes(client: &Client, bucket: &str, key: &str, expected: &[u8
.await
}
#[cfg(test)]
mod retry_tests {
use super::*;
use crate::common::build_test_s3_config;
use crate::fake_s3_target::{FAKE_ACCESS_KEY, FAKE_SECRET_KEY, FakeS3Target, FaultAction, Operation};
#[tokio::test]
async fn iam_polls_only_retry_auth_propagation_and_availability() -> TestResult {
let target = FakeS3Target::start().await?;
target.create_bucket("upgrade-retry-probe");
let client = Client::from_conf(build_test_s3_config(
target.endpoint(),
FAKE_ACCESS_KEY,
FAKE_SECRET_KEY,
None,
"upgrade-retry-test",
));
for status in [403, 503] {
target.inject(Operation::PutObject, FaultAction::ResponseStatus(status), 1);
wait_for_put(&client, "upgrade-retry-probe", "key", b"body".to_vec(), "IAM PUT").await?;
assert_eq!(target.take_requests().len(), 2);
target.inject(Operation::GetObject, FaultAction::ResponseStatus(status), 1);
wait_for_bytes(&client, "upgrade-retry-probe", "key", b"body", "IAM GET").await?;
assert_eq!(target.take_requests().len(), 2);
}
target.inject(Operation::PutObject, FaultAction::ResponseStatus(500), 1);
let error = wait_for_put(&client, "upgrade-retry-probe", "key", b"body".to_vec(), "IAM PUT")
.await
.expect_err("IAM convergence must not hide InternalError on PUT");
assert_eq!(s3_probe_error_code(error.as_ref()), Some("InternalError"));
assert_eq!(target.take_requests().len(), 1);
target.inject(Operation::GetObject, FaultAction::ResponseStatus(500), 1);
let error = wait_for_bytes(&client, "upgrade-retry-probe", "key", b"body", "IAM GET")
.await
.expect_err("IAM convergence must not hide InternalError on GET");
assert_eq!(s3_probe_error_code(error.as_ref()), Some("InternalError"));
assert_eq!(target.take_requests().len(), 1);
let error = wait_for_bytes(&client, "upgrade-retry-probe", "key", b"wrong bytes", "IAM GET")
.await
.expect_err("IAM convergence must not hide a payload mismatch");
assert!(error.to_string().contains("bytes mismatch"));
assert_eq!(target.requests().len(), 1);
Ok(())
}
}
async fn seed_history_and_iam(dist: &DistCluster) -> TestResult<UpgradeSeed> {
let history_bucket = unique_bucket("upg-hist");
let versioned_bucket = unique_bucket("upg-ver");
@@ -215,6 +215,8 @@ mod tests {
probe_startup_cas_binary(&binary, &nonce, &artifact).await?;
let mut cluster = RustFSTestClusterEnvironment::new(4).await?;
// Isolate bootstrap CAS and its S3 readback from periodic usage publication.
cluster.set_env("RUSTFS_SCANNER_ENABLED", "false");
let mut logs = Vec::new();
let mut disks = Vec::new();
let mut endpoints = Vec::new();
@@ -1453,7 +1453,12 @@ mod tests {
let server_rust_log = std::env::var("RUSTFS_HEAL_CHAOS_SERVER_RUST_LOG")
.unwrap_or_else(|_| "rustfs::heal::task=info,rustfs=error".to_string());
cluster.set_env("RUST_LOG", server_rust_log);
let log_dir = std::env::var("RUSTFS_HEAL_CHAOS_LOG_DIR").unwrap_or_else(|_| format!("{}/logs", cluster.temp_dir));
let log_dir = std::env::var("RUSTFS_HEAL_CHAOS_LOG_DIR").unwrap_or_else(|_| {
// Keep explicit directories compatible and isolate parallel suite captures.
std::env::var("RUSTFS_HEAL_CHAOS_LOG_ROOT")
.map(|root| format!("{root}/{interruption_kind}-{}", uuid::Uuid::new_v4()))
.unwrap_or_else(|_| format!("{}/logs", cluster.temp_dir))
});
std::fs::create_dir_all(&log_dir)?;
for node_index in 0..cluster.nodes.len() {
cluster.set_node_capture_log_path(node_index, format!("{log_dir}/node{node_index}.log"))?;
@@ -1799,7 +1804,8 @@ mod tests {
let (mut client_token, mut task_status_url) = if background_rejoin_heal_evidence {
(String::new(), String::new())
} else {
let heal_start_body = signed_admin_post(&heal_url, Some(heal_body), &cluster.access_key, &cluster.secret_key).await?;
let heal_start_body =
start_root_heal_when_control_ready(&heal_url, heal_body, &cluster.access_key, &cluster.secret_key).await?;
let heal_start: serde_json::Value = serde_json::from_str(&heal_start_body)
.map_err(|err| format!("heal start response is not JSON ({err}): {heal_start_body}"))?;
let client_token = heal_start["clientToken"]
@@ -2140,7 +2146,8 @@ mod tests {
&& status["summary"].as_str() == Some("failed")
{
let heal_start_body =
signed_admin_post(&heal_url, Some(heal_body), &cluster.access_key, &cluster.secret_key).await?;
start_root_heal_when_control_ready(&heal_url, heal_body, &cluster.access_key, &cluster.secret_key)
.await?;
let heal_start: serde_json::Value = serde_json::from_str(&heal_start_body)
.map_err(|err| format!("recovery heal start response is not JSON ({err}): {heal_start_body}"))?;
client_token = heal_start["clientToken"]
@@ -7194,13 +7194,30 @@ async fn test_site_replication_state_edit_fresh_and_stale_real_dual_node() -> Re
assert!(source_info.sites.iter().all(|peer| !peer.replicate_ilm_expiry));
assert!(target_info.sites.iter().all(|peer| !peer.replicate_ilm_expiry));
let source_deployment_id = source_info
.sites
.iter()
.find(|peer| peer.endpoint == source_env.url)
.map(|peer| peer.deployment_id.clone())
.ok_or("source site missing from its own replication info")?;
let target_deployment_id = target_info
.sites
.iter()
.find(|peer| peer.endpoint == target_env.url)
.map(|peer| peer.deployment_id.clone())
.ok_or("target site missing from its own replication info")?;
let target_status =
wait_for_site_replication_status(&target_env, "peer-state=true", |status| status.peer_states.len() == 2).await?;
let current_updated_at = target_status
.peer_states
.values()
.find_map(|state| state.updated_at)
.get(&target_deployment_id)
.and_then(|state| state.updated_at)
.ok_or("missing target site replication updated_at")?;
let source_updated_at = target_status
.peer_states
.get(&source_deployment_id)
.and_then(|state| state.updated_at)
.ok_or("missing source site replication updated_at")?;
let mut stale_peers = BTreeMap::new();
for peer in target_info.sites {
@@ -7244,16 +7261,23 @@ async fn test_site_replication_state_edit_fresh_and_stale_real_dual_node() -> Re
.await?;
assert!(target_after_fresh.sites.iter().all(|peer| peer.replicate_ilm_expiry));
// State edits apply only to the addressed site; status reports each site's own state.
let target_status_after_fresh = wait_for_site_replication_status(&target_env, "peer-state=true", |status| {
status.peer_states.len() == 2
&& status.peer_states.values().all(|state| {
state.updated_at == Some(fresh_updated_at) && state.peers.values().all(|peer| peer.replicate_ilm_expiry)
&& status.peer_states.get(&target_deployment_id).is_some_and(|state| {
state.updated_at == Some(fresh_updated_at)
&& state.peers.len() == 2
&& state.peers.values().all(|peer| peer.replicate_ilm_expiry)
})
})
.await?;
assert!(target_status_after_fresh.peer_states.values().all(|state| {
state.updated_at == Some(fresh_updated_at) && state.peers.values().all(|peer| peer.replicate_ilm_expiry)
}));
let source_state_after_fresh = target_status_after_fresh
.peer_states
.get(&source_deployment_id)
.ok_or("missing source site replication state after target edit")?;
assert_eq!(source_state_after_fresh.updated_at, Some(source_updated_at));
assert_eq!(source_state_after_fresh.peers.len(), 2);
assert!(source_state_after_fresh.peers.values().all(|peer| !peer.replicate_ilm_expiry));
let source_after_fresh = site_replication_info(&source_env).await?;
assert!(source_after_fresh.sites.iter().all(|peer| !peer.replicate_ilm_expiry));
+1 -1
View File
@@ -448,7 +448,7 @@ pub mod disk {
pub mod error {
pub use crate::error::{
Error, PoolMetadataError, PoolMetadataFailure, Result, StorageError, classify_system_path_failure_reason,
is_err_bucket_not_found, is_err_object_not_found, is_err_version_not_found,
is_err_bucket_not_found, is_err_invalid_upload_id, is_err_object_not_found, is_err_version_not_found,
};
}
@@ -495,6 +495,193 @@ mod tests {
assert!(rate > 0.0);
}
// Tokio's paused clock does not advance ratelimit's StdClock. Keep natural
// refill below one whole token for years and supply each exact budget here.
fn supply_reader_test_tokens(throttle: &BucketThrottle, available: u64) {
let rate = u64::try_from(throttle.node_bandwidth_per_sec).expect("test throttle rate should be positive");
let burst = throttle.burst();
*throttle.limiter.lock().expect("test throttle should not poison") = Ratelimiter::builder(rate)
.max_tokens(burst)
.initial_available(available)
.period(Duration::from_nanos(u64::MAX))
.build()
.expect("controlled token budget should be valid");
}
#[tokio::test(start_paused = true)]
async fn test_monitored_reader_retains_partial_tokens_across_waits() {
use crate::bucket::bandwidth::reader::{MonitorReaderOptions, MonitoredReader};
use futures_util::task::noop_waker_ref;
use std::pin::Pin;
use std::task::{Context, Poll};
use tokio::io::{AsyncRead, ReadBuf};
for header_size in [0, 25] {
let monitor = Monitor::new(1);
let opts = BucketOptions {
name: "reader-budget".to_string(),
replication_arn: "reader-budget-arn".to_string(),
};
monitor.set_bandwidth_limit(&opts.name, &opts.replication_arn, 100);
let throttle = monitor.throttle(&opts).expect("test throttle should exist");
supply_reader_test_tokens(&throttle, 25);
let data = [0xAB; 100];
let mut reader = MonitoredReader::new(
monitor,
data.as_slice(),
MonitorReaderOptions {
bucket_options: opts,
header_size,
},
);
let mut output = [0; 100];
let mut buf = ReadBuf::new(&mut output);
let mut cx = Context::from_waker(noop_waker_ref());
assert!(Pin::new(&mut reader).poll_read(&mut cx, &mut buf).is_pending());
assert!(buf.filled().is_empty(), "partial payment must not authorize payload reads");
assert_eq!(throttle.limiter.lock().expect("test throttle").available(), 0);
supply_reader_test_tokens(&throttle, 75);
tokio::time::advance(Duration::from_secs(1)).await;
assert!(
matches!(Pin::new(&mut reader).poll_read(&mut cx, &mut buf), Poll::Ready(Ok(()))),
"25 paid tokens plus 75 later tokens must fund the original read with header_size={header_size}"
);
assert_eq!(buf.filled(), &data[..100 - header_size]);
assert_eq!(throttle.limiter.lock().expect("test throttle").available(), 0);
}
}
#[tokio::test(start_paused = true)]
async fn test_monitored_reader_preserves_paid_read_across_cancellation() {
use crate::bucket::bandwidth::reader::{MonitorReaderOptions, MonitoredReader};
use futures_util::task::noop_waker_ref;
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWriteExt, ReadBuf};
let monitor = Monitor::new(1);
let opts = BucketOptions {
name: "reader-pending".to_string(),
replication_arn: "reader-pending-arn".to_string(),
};
monitor.set_bandwidth_limit(&opts.name, &opts.replication_arn, 100);
let throttle = monitor.throttle(&opts).expect("test throttle should exist");
supply_reader_test_tokens(&throttle, 100);
let (mut writer, inner) = tokio::io::duplex(100);
let mut reader = MonitoredReader::new(
monitor,
inner,
MonitorReaderOptions {
bucket_options: opts,
header_size: 0,
},
);
let mut cx = Context::from_waker(noop_waker_ref());
let mut first_output = [0; 100];
let mut first_read = Box::pin(reader.read(&mut first_output));
assert!(first_read.as_mut().poll(&mut cx).is_pending());
assert_eq!(throttle.limiter.lock().expect("test throttle").available(), 0);
drop(first_read);
writer
.write_all(&[0xAB; 100])
.await
.expect("duplex writer should supply the pending read");
// A canceled read future keeps its reader alive. Resume in a smaller
// buffer without charging the already-paid allowance a second time.
let mut short_output = [0; 40];
let mut short_buf = ReadBuf::new(&mut short_output);
assert!(matches!(Pin::new(&mut reader).poll_read(&mut cx, &mut short_buf), Poll::Ready(Ok(()))));
assert_eq!(short_buf.filled(), &[0xAB; 40]);
assert_eq!(throttle.limiter.lock().expect("test throttle").available(), 0);
// Completing the short read discards unused allowance, matching the
// existing precharge policy. The next read must buy a fresh budget.
let mut remaining_output = [0; 100];
let mut remaining_buf = ReadBuf::new(&mut remaining_output);
assert!(Pin::new(&mut reader).poll_read(&mut cx, &mut remaining_buf).is_pending());
supply_reader_test_tokens(&throttle, 100);
tokio::time::advance(Duration::from_secs(1)).await;
assert!(matches!(
Pin::new(&mut reader).poll_read(&mut cx, &mut remaining_buf),
Poll::Ready(Ok(()))
));
assert_eq!(remaining_buf.filled(), &[0xAB; 60]);
assert_eq!(throttle.limiter.lock().expect("test throttle").available(), 0);
drop(writer);
let mut eof_output = [0; 100];
let mut eof_buf = ReadBuf::new(&mut eof_output);
assert!(Pin::new(&mut reader).poll_read(&mut cx, &mut eof_buf).is_pending());
supply_reader_test_tokens(&throttle, 100);
tokio::time::advance(Duration::from_secs(1)).await;
assert!(matches!(Pin::new(&mut reader).poll_read(&mut cx, &mut eof_buf), Poll::Ready(Ok(()))));
assert!(eof_buf.filled().is_empty());
assert_eq!(throttle.limiter.lock().expect("test throttle").available(), 0);
assert!(Pin::new(&mut reader).poll_read(&mut cx, &mut eof_buf).is_pending());
}
#[tokio::test(start_paused = true)]
async fn test_monitored_reader_shrinks_unpaid_payload_after_bandwidth_reduction() {
use crate::bucket::bandwidth::reader::{MonitorReaderOptions, MonitoredReader};
use futures_util::task::noop_waker_ref;
use std::pin::Pin;
use std::task::{Context, Poll};
use tokio::io::{AsyncRead, ReadBuf};
for header_size in [0, 25] {
let monitor = Monitor::new(1);
let opts = BucketOptions {
name: "reader-reduced-budget".to_string(),
replication_arn: "reader-reduced-budget-arn".to_string(),
};
monitor.set_bandwidth_limit(&opts.name, &opts.replication_arn, 100);
let initial_throttle = monitor.throttle(&opts).expect("initial throttle should exist");
supply_reader_test_tokens(&initial_throttle, 25);
let data = [0xAB; 100];
let mut reader = MonitoredReader::new(
monitor.clone(),
data.as_slice(),
MonitorReaderOptions {
bucket_options: opts.clone(),
header_size,
},
);
let mut output = [0; 100];
let mut buf = ReadBuf::new(&mut output);
let mut cx = Context::from_waker(noop_waker_ref());
assert!(Pin::new(&mut reader).poll_read(&mut cx, &mut buf).is_pending());
assert!(buf.filled().is_empty());
assert_eq!(initial_throttle.limiter.lock().expect("initial throttle").available(), 0);
monitor.set_bandwidth_limit(&opts.name, &opts.replication_arn, 10);
let reduced_throttle = monitor.throttle(&opts).expect("reduced throttle should exist");
assert_eq!(reduced_throttle.burst(), 10);
supply_reader_test_tokens(&reduced_throttle, 0);
tokio::time::advance(Duration::from_secs(1)).await;
let result = Pin::new(&mut reader).poll_read(&mut cx, &mut buf);
let expected_payload = if header_size == 0 {
assert!(
matches!(result, Poll::Ready(Ok(()))),
"already-paid payload must not wait for the old burst's unpaid remainder"
);
25
} else {
assert!(result.is_pending(), "the initial 25 tokens paid only the header");
assert!(buf.filled().is_empty(), "payload still requires its own reduced budget");
supply_reader_test_tokens(&reduced_throttle, 10);
tokio::time::advance(Duration::from_secs(1)).await;
assert!(matches!(Pin::new(&mut reader).poll_read(&mut cx, &mut buf), Poll::Ready(Ok(()))));
10
};
assert_eq!(buf.filled(), &data[..expected_payload]);
assert_eq!(reduced_throttle.limiter.lock().expect("reduced throttle").available(), 0);
}
}
#[test]
fn test_consume_refills_continuously() {
let clock = TestClock::new();
+26 -7
View File
@@ -41,6 +41,8 @@ pub struct MonitoredReader<R> {
m: Arc<Monitor>,
opts: MonitorReaderOptions,
wait_state: std::sync::Mutex<Option<WaitState>>,
// Payload limit and unpaid tokens for the read retained across Pending polls.
pending_read: Option<(usize, u64)>,
temp_buf: Vec<u8>,
}
@@ -63,6 +65,7 @@ impl<R> MonitoredReader<R> {
m,
opts,
wait_state: std::sync::Mutex::new(None),
pending_read: None,
temp_buf: Vec::new(),
}
}
@@ -90,17 +93,30 @@ impl<R: AsyncRead + Unpin> AsyncRead for MonitoredReader<R> {
let throttle = match this.m.throttle(&this.opts.bucket_options) {
Some(t) => t,
None => return Pin::new(&mut this.r).poll_read(cx, buf),
None => {
this.pending_read = None;
return Pin::new(&mut this.r).poll_read(cx, buf);
}
};
let b = throttle.burst();
debug_assert!(b >= 1, "burst must be at least 1");
let (need, tokens) = calc_need_and_tokens(b, buf.remaining(), &mut this.opts.header_size);
let (deficit, rate, consumed) = throttle.consume(tokens);
let need = need.min(consumed as usize);
let (need, tokens) = match this.pending_read {
Some(reservation) => reservation,
None => calc_need_and_tokens(b, buf.remaining(), &mut this.opts.header_size),
};
// A lower limit may discard unpaid payload, but must retain paid
// payload and any header tokens still owed by this pending read.
let excess_payload = need.saturating_sub(usize::try_from(b).unwrap_or(usize::MAX));
let reduction = tokens.min(u64::try_from(excess_payload).unwrap_or(u64::MAX));
let need = need.saturating_sub(usize::try_from(reduction).unwrap_or(usize::MAX));
let tokens = tokens - reduction;
let (deficit, rate, _) = throttle.consume(tokens);
this.pending_read = Some((need, deficit));
if deficit > 0 && rate > 0.0 {
let duration = std::time::Duration::from_secs_f64(deficit as f64 / rate);
// A changed limit may reduce the burst while this read is pending.
let duration = std::time::Duration::from_secs_f64(deficit.min(b) as f64 / rate);
debug!(
tokens = tokens,
deficit = deficit,
@@ -115,14 +131,17 @@ impl<R: AsyncRead + Unpin> AsyncRead for MonitoredReader<R> {
warn!("MonitoredReader wait_state mutex poisoned, recovering");
e.into_inner()
}) = Some(WaitState { sleep });
return Poll::Pending;
}
Poll::Ready(()) => {}
Poll::Ready(()) => cx.waker().wake_by_ref(),
}
return Poll::Pending;
}
let filled_before = buf.filled().len();
let result = poll_limited_read(&mut this.r, cx, buf, need, &mut this.temp_buf);
if result.is_ready() {
this.pending_read = None;
}
if let Poll::Ready(Ok(())) = result {
let read_bytes = buf.filled().len().saturating_sub(filled_before) as u64;
if read_bytes > 0 {
@@ -1206,6 +1206,17 @@ mod tests {
assert_eq!(mixed_timeout, DiskError::Timeout);
}
#[test]
fn mixed_missing_and_body_io_errors_remain_an_actionable_quorum_failure() {
let failures = [
DiskError::FileNotFound,
DiskError::Io(std::io::Error::other("remote body stream aborted")),
];
assert!(!is_benign_not_found_listing_failure(&failures));
assert_eq!(classify_listing_quorum_failure(&failures), DiskError::ErasureReadQuorum);
}
#[test]
fn missing_path_error_classification_excludes_actionable_failures() {
assert!(is_missing_path_error(&DiskError::FileNotFound));
@@ -901,6 +901,7 @@ pub fn build_internode_data_transport_from_env() -> Result<Arc<dyn InternodeData
mod tests {
use super::*;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use tokio::io::AsyncWriteExt;
use tokio::sync::{Barrier, Notify};
async fn wait_for_capability_flight_waiters(entry: &PutFileCapabilityCacheEntry, waiters: usize) {
@@ -1026,6 +1027,45 @@ mod tests {
assert!(matches!(open_err, Error::MethodNotAllowed));
}
#[tokio::test]
async fn walk_dir_restores_explicit_remote_missing_error() {
let _ = rustfs_credentials::set_global_rpc_secret("walk-dir-error-test-secret".to_string());
let listener = match tokio::net::TcpListener::bind("127.0.0.1:0").await {
Ok(listener) => listener,
Err(err) if err.kind() == io::ErrorKind::PermissionDenied => return,
Err(err) => panic!("test listener should bind: {err}"),
};
let addr = listener.local_addr().expect("test listener address should be available");
let server = tokio::spawn(async move {
let (mut socket, _) = listener.accept().await.expect("test server should accept the request");
let mut request = [0_u8; 2048];
let _ = socket.read(&mut request).await.expect("request should be readable");
socket
.write_all(
b"HTTP/1.1 500 Internal Server Error\r\nx-rustfs-disk-error: file-not-found\r\ncontent-length: 0\r\nconnection: close\r\n\r\n",
)
.await
.expect("typed disk error response should be writable");
});
let endpoint = format!("http://{addr}");
let error = match TcpHttpInternodeDataTransport
.open_walk_dir(WalkDirStreamRequest {
endpoint: endpoint.clone(),
disk: "disk-a".to_string(),
body: b"{}".to_vec(),
stall_timeout: Some(Duration::from_secs(2)),
})
.await
{
Ok(_) => panic!("an explicit remote missing-path response must fail the open"),
Err(error) => error,
};
assert!(matches!(error, Error::FileNotFound));
server.await.expect("test server task should finish");
}
#[test]
fn tcp_http_capabilities_are_behavior_preserving() {
let transport = TcpHttpInternodeDataTransport;
+1
View File
@@ -3154,6 +3154,7 @@ mod tests {
version_purge_status: Default::default(),
replication_decision: String::new(),
checksum: None,
multipart_completion_replayed: false,
}
}
}
+4 -1
View File
@@ -19542,7 +19542,10 @@ mod tests {
.await
.expect("second-node activation should not panic")
.expect_err("the second reservation must observe and reject the committed first reservation");
assert!(err.to_string().contains("requires 60 bytes, but 40 bytes are available"));
assert!(
err.to_string().contains("requires 60 bytes, but 40 bytes are available"),
"the second reservation must report the committed capacity rejection, got {err:?}"
);
let mut persisted = PoolMeta::default();
persisted
+77 -21
View File
@@ -2288,7 +2288,7 @@ mod decommission_lock_order_tests {
let (lossy_store, refresh_calls) = store_with_capacity_lease_loss(&other_store).await;
set_decommission_capacity_info_overrides_for_test(lossy_store.id, (0..40).map(|_| capacity_snapshot()).collect());
let barrier = MultipartCommitBarrier::install(&bucket, object, MultipartCommitPause::AfterObjectPublication);
let migration = tokio::spawn({
let mut migration = tokio::spawn({
let migration_store = Arc::clone(&lossy_store);
let migration_bucket = bucket.clone();
async move {
@@ -2304,7 +2304,12 @@ mod decommission_lock_order_tests {
.await
}
});
barrier.wait_until_paused().await;
tokio::select! {
() = barrier.wait_until_paused() => {}
result = &mut migration => {
panic!("initial multipart migration completed before reaching the publication barrier: {result:?}");
}
}
tokio::time::pause();
let published = lossy_store.pools[2]
.get_object_info(
@@ -2321,6 +2326,18 @@ mod decommission_lock_order_tests {
.expect("the multipart target must publish before capacity progress save");
assert!(published.is_multipart(), "published target must retain multipart identity");
assert_eq!(published.version_id.map(|version| version.to_string()), Some(source_version.clone()));
for prefix in [
rustfs_utils::http::RUSTFS_INTERNAL_PREFIX,
rustfs_utils::http::MINIO_INTERNAL_PREFIX,
] {
assert_eq!(
published
.user_defined
.get(&format!("{prefix}{}", rustfs_utils::http::SUFFIX_MULTIPART_UPLOAD_ID)),
Some(&source_upload.upload_id),
"migration must preserve the source completion identity under both internal prefixes"
);
}
let mut published_reader = lossy_store.pools[2]
.get_object_reader(
&bucket,
@@ -6851,7 +6868,7 @@ mod decommission_lock_order_tests {
let migration_store = Arc::clone(&store);
let migration_bucket = bucket.clone();
async move {
data_movement::migrate_decommission_object(
let result = data_movement::migrate_decommission_object(
migration_store,
0,
migration_bucket,
@@ -6860,16 +6877,19 @@ mod decommission_lock_order_tests {
"decommission_object_lock_order",
Some(owner),
)
.await
.await;
(result, std::time::Instant::now())
}
});
barrier.wait_until_owner_paused().await;
let ordinary_launched_at = std::time::Instant::now();
let mut ordinary_mutation = tokio::spawn({
let ordinary_store = Arc::clone(&other_store);
let ordinary_bucket = bucket.clone();
async move {
match mutation {
let ordinary_started_at = std::time::Instant::now();
let result = match mutation {
ExternalObjectMutation::Put => {
let mut data = PutObjReader::from_vec(b"ordinary replacement".to_vec());
ordinary_store
@@ -6941,7 +6961,8 @@ mod decommission_lock_order_tests {
.restore_transitioned_object(&ordinary_bucket, object, &opts)
.await
}
}
};
(result, ordinary_started_at, std::time::Instant::now())
}
});
if let Some(get_barrier) = restore_get_barrier.as_ref() {
@@ -6989,8 +7010,10 @@ mod decommission_lock_order_tests {
} else {
barrier.wait_until_external_capacity_released().await;
}
let ordinary_lock_phase_observed_at = std::time::Instant::now();
assert!(!ordinary_mutation.is_finished());
let migration_result = if pause_restore_tail {
let owner_released_at = std::time::Instant::now();
let (migration_result, migration_finished_at) = if pause_restore_tail {
barrier.release_owner();
tokio::time::timeout(std::time::Duration::from_secs(30), &mut migration)
.await
@@ -7005,7 +7028,7 @@ mod decommission_lock_order_tests {
};
migration_result.expect("migration should commit to its reserved target");
let ordinary_result = if pause_restore_tail {
let (ordinary_result, ordinary_started_at, ordinary_finished_at) = if pause_restore_tail {
let handoff_barrier = PutObjectCommitBarrier::install(&bucket, object, PutObjectCommitPause::AfterRenameHandoff);
let tail_barrier =
crate::set_disk::rename_fanout_barrier::arm(object, 0, crate::set_disk::rename_fanout_barrier::PHASE_RENAME);
@@ -7077,7 +7100,20 @@ mod decommission_lock_order_tests {
.expect("ordinary object mutation task should join")
};
drop(barrier);
ordinary_result.expect("ordinary object mutation should commit to the unreserved target");
ordinary_result.unwrap_or_else(|error| {
panic!(
"ordinary object mutation should commit to the unreserved target: {error:?}; \
ordinary_start_after_launch={:?}, lock_phase_observed_after_launch={:?}, \
owner_release_after_launch={:?}, migration_after_owner_release={:?}, \
ordinary_finish_after_launch={:?}, ordinary_finished_before_migration={}",
ordinary_started_at.duration_since(ordinary_launched_at),
ordinary_lock_phase_observed_at.duration_since(ordinary_launched_at),
owner_released_at.duration_since(ordinary_launched_at),
migration_finished_at.duration_since(owner_released_at),
ordinary_finished_at.duration_since(ordinary_launched_at),
ordinary_finished_at < migration_finished_at,
);
});
store.pools[1]
.get_object_info(
&bucket,
@@ -7252,7 +7288,7 @@ mod decommission_lock_order_tests {
let migration_store = Arc::clone(&store);
let migration_bucket = bucket.clone();
async move {
data_movement::migrate_decommission_object(
let result = data_movement::migrate_decommission_object(
migration_store,
0,
migration_bucket,
@@ -7261,11 +7297,13 @@ mod decommission_lock_order_tests {
"same_object_copy_lock_order",
Some(owner),
)
.await
.await;
(result, std::time::Instant::now())
}
});
barrier.wait_until_owner_paused().await;
let ordinary_launched_at = std::time::Instant::now();
let copy_task = tokio::spawn({
let copy_store = Arc::clone(&other_store);
let copy_bucket = bucket.clone();
@@ -7280,7 +7318,8 @@ mod decommission_lock_order_tests {
..Default::default()
};
async move {
copy_store
let ordinary_started_at = std::time::Instant::now();
let result = copy_store
.copy_object(
&copy_bucket,
object,
@@ -7291,27 +7330,44 @@ mod decommission_lock_order_tests {
&destination_opts,
)
.await
.map(|_| ())
.map(|_| ());
(result, ordinary_started_at, std::time::Instant::now())
}
});
barrier.wait_until_external_capacity_released().await;
let ordinary_lock_phase_observed_at = std::time::Instant::now();
assert!(
!copy_task.is_finished(),
"historical same-object CopyObject must wait behind the migration object fence"
);
let owner_released_at = std::time::Instant::now();
barrier.release_owner();
drop(barrier);
tokio::time::timeout(std::time::Duration::from_secs(30), migration)
let (migration_result, migration_finished_at) = tokio::time::timeout(std::time::Duration::from_secs(30), migration)
.await
.expect("migration must not deadlock with historical same-object CopyObject")
.expect("migration task should join")
.expect("migration should commit to its reserved target");
tokio::time::timeout(std::time::Duration::from_secs(30), copy_task)
.await
.expect("historical same-object CopyObject must finish after the migration releases its object fence")
.expect("CopyObject task should join")
.expect("CopyObject should preserve the expected current-version precondition");
.expect("migration task should join");
migration_result.expect("migration should commit to its reserved target");
let (ordinary_result, ordinary_started_at, ordinary_finished_at) =
tokio::time::timeout(std::time::Duration::from_secs(30), copy_task)
.await
.expect("historical same-object CopyObject must finish after the migration releases its object fence")
.expect("CopyObject task should join");
ordinary_result.unwrap_or_else(|error| {
panic!(
"CopyObject should preserve the expected current-version precondition: {error:?}; \
ordinary_start_after_launch={:?}, lock_phase_observed_after_launch={:?}, \
owner_release_after_launch={:?}, migration_after_owner_release={:?}, \
ordinary_finish_after_launch={:?}, ordinary_finished_before_migration={}",
ordinary_started_at.duration_since(ordinary_launched_at),
ordinary_lock_phase_observed_at.duration_since(ordinary_launched_at),
owner_released_at.duration_since(ordinary_launched_at),
migration_finished_at.duration_since(owner_released_at),
ordinary_finished_at.duration_since(ordinary_launched_at),
ordinary_finished_at < migration_finished_at,
);
});
}
#[tokio::test]
+55 -1
View File
@@ -2085,7 +2085,9 @@ impl DiskAPI for LocalDiskWrapper {
}
let result = self.disk.disk_info(opts).await?;
if let Some(current_disk_id) = *self.disk_id.read().await
// Fresh capacity snapshots omit the disk ID; the stale-disk precheck above already verified it.
if !opts.fresh_capacity
&& let Some(current_disk_id) = *self.disk_id.read().await
&& Some(current_disk_id) != result.id
{
return Err(DiskError::DiskNotFound);
@@ -2571,7 +2573,9 @@ impl DiskAPI for LocalDiskWrapper {
mod tests {
use super::*;
use crate::disk::endpoint::Endpoint;
use crate::disk::format::FormatV3;
use crate::disk::health_state::RuntimeDriveHealthState;
use crate::disk::{FORMAT_CONFIG_FILE, RUSTFS_META_BUCKET};
use std::{
io,
panic::{AssertUnwindSafe, catch_unwind},
@@ -2926,6 +2930,56 @@ mod tests {
assert_eq!(snapshot.total_errors_availability, 1);
}
#[tokio::test]
async fn fresh_capacity_keeps_disk_identity_precheck_without_requiring_id_in_snapshot() {
let dir = tempfile::tempdir().expect("temp dir should be created");
let mut endpoint =
Endpoint::try_from(dir.path().to_str().expect("temp dir should be valid UTF-8")).expect("endpoint should parse");
endpoint.set_pool_index(0);
endpoint.set_set_index(0);
endpoint.set_disk_index(0);
let meta_dir = dir.path().join(RUSTFS_META_BUCKET);
tokio::fs::create_dir_all(&meta_dir)
.await
.expect("metadata directory should be created");
let mut format = FormatV3::new(1, 1);
format.erasure.this = format.erasure.sets[0][0];
tokio::fs::write(meta_dir.join(FORMAT_CONFIG_FILE), format.to_json().expect("format should serialize"))
.await
.expect("format should be written");
let disk = Arc::new(LocalDisk::new(&endpoint, false).await.expect("formatted disk should open"));
let disk_id = disk
.get_disk_id()
.await
.expect("disk ID should be readable")
.expect("formatted disk should have an ID");
let wrapper = LocalDiskWrapper::new(Arc::clone(&disk), false);
wrapper.set_disk_id_state(Some(disk_id)).await;
let snapshot = wrapper
.disk_info(&DiskInfoOptions {
fresh_capacity: true,
..Default::default()
})
.await
.expect("fresh capacity should pass when the disk identity matches");
assert!(snapshot.fresh_capacity);
assert!(snapshot.total > 0);
assert!(snapshot.id.is_none(), "capacity-only snapshots should not expose the disk ID");
let stale_wrapper = LocalDiskWrapper::new(disk, false);
stale_wrapper.set_disk_id_state(Some(Uuid::nil())).await;
let error = stale_wrapper
.disk_info(&DiskInfoOptions {
fresh_capacity: true,
..Default::default()
})
.await
.expect_err("a stale disk identity must still be rejected before the capacity probe");
assert_eq!(error, DiskError::DiskNotFound);
}
#[tokio::test]
async fn delete_versions_counts_returned_batch_error_classes() {
let dir = tempfile::tempdir().expect("temp dir should be created");
+34 -2
View File
@@ -428,6 +428,19 @@ impl DiskError {
}
}
/// Whether an internode RPC was cancelled, without classifying the peer as offline.
pub fn io_error_is_rpc_cancelled(error: &io::Error) -> bool {
error
.get_ref()
.and_then(|source| source.downcast_ref::<RpcStatusError>())
.is_some_and(|error| error.status().code() == tonic::Code::Cancelled)
}
pub(crate) fn clone_rpc_status_io_error(error: &io::Error) -> Option<io::Error> {
let status = error.get_ref()?.downcast_ref::<RpcStatusError>()?;
Some(io::Error::new(error.kind(), RpcStatusError(status.0.clone())))
}
pub fn internode_http_error_kind(&self) -> Option<InternodeHttpErrorKind> {
match self {
DiskError::Io(io_error) => io_error
@@ -701,8 +714,8 @@ impl Clone for DiskError {
DiskError::conditional_file_not_committed(io::Error::new(io_error.kind(), io_error.to_string())),
),
DiskError::Io(io_error) => {
if let Some(status) = io_error.get_ref().and_then(|source| source.downcast_ref::<RpcStatusError>()) {
return DiskError::Io(io::Error::new(io_error.kind(), RpcStatusError(status.0.clone())));
if let Some(error) = Self::clone_rpc_status_io_error(io_error) {
return DiskError::Io(error);
}
DiskError::Io(
Self::clone_dangling_delete_grace(io_error)
@@ -900,6 +913,25 @@ mod tests {
use super::*;
use std::collections::HashMap;
#[test]
fn rpc_status_survives_disk_and_storage_clones() {
for code in [tonic::Code::Cancelled, tonic::Code::PermissionDenied] {
let status = tonic::Status::new(code, "operation was canceled");
let original = DiskError::Io(io::Error::new(io::ErrorKind::Interrupted, RpcStatusError::from(status)));
let display = original.to_string();
let disk = original.clone();
let storage = crate::error::StorageError::from(disk);
let cloned = storage.clone();
for error in [io::Error::from(original), io::Error::from(storage), io::Error::from(cloned)] {
assert_eq!(error.kind(), io::ErrorKind::Interrupted);
let status = error.get_ref().unwrap().downcast_ref::<RpcStatusError>().unwrap().status();
assert_eq!(status.code(), code);
assert_eq!(status.message(), "operation was canceled");
assert_eq!(DiskError::from(error).to_string(), display);
}
}
}
#[test]
fn retired_marker_deferral_survives_disk_and_storage_clones() {
let original = DiskError::retired_marker_deferred("missing retirement record");
+3 -2
View File
@@ -623,8 +623,9 @@ impl Clone for StorageError {
fn clone(&self) -> Self {
match self {
StorageError::Io(e) => {
if let Some(error) =
DiskError::clone_dangling_delete_grace(e).or_else(|| DiskError::clone_retired_marker_deferred(e))
if let Some(error) = DiskError::clone_rpc_status_io_error(e)
.or_else(|| DiskError::clone_dangling_delete_grace(e))
.or_else(|| DiskError::clone_retired_marker_deferred(e))
{
return StorageError::Io(error);
}
+5
View File
@@ -1444,6 +1444,10 @@ pub struct ObjectInfo {
pub version_purge_status: VersionPurgeStatusType,
pub replication_decision: String,
pub checksum: Option<Bytes>,
/// True when this `CompleteMultipartUpload` result is the object already
/// published by the same upload. Callers must not repeat quota, replication,
/// lifecycle, or object-created side effects for that response.
pub multipart_completion_replayed: bool,
}
impl Clone for ObjectInfo {
@@ -1484,6 +1488,7 @@ impl Clone for ObjectInfo {
version_purge_status: self.version_purge_status.clone(),
replication_decision: self.replication_decision.clone(),
checksum: self.checksum.clone(),
multipart_completion_replayed: self.multipart_completion_replayed,
expires: self.expires,
}
}
@@ -107,7 +107,9 @@ async fn handle_prepare(
}
let tier_config_mgr = api.tier_config_mgr();
for _ in 0..3 {
// Peers can race to create the same durable intent. Re-read after bounded
// lock contention so the winning publication follows the normal identity checks.
for attempt in 0..3 {
let (stored, applied) = match load_tier_mutation_intent_record(api.clone(), mutation_id).await {
Ok(existing) => {
if !existing.same_identity_as(&intent) {
@@ -123,9 +125,11 @@ async fn handle_prepare(
match save_tier_mutation_intent_record_if_absent(api.clone(), &intent).await {
Ok(()) => (intent.clone(), true),
Err(Error::PreconditionFailed) => continue,
Err(Error::Lock(rustfs_lock::LockError::Timeout { .. })) if attempt < 2 => continue,
Err(err) => return Err(err.into()),
}
}
Err(Error::Lock(rustfs_lock::LockError::Timeout { .. })) if attempt < 2 => continue,
Err(err) => return Err(err.into()),
};
@@ -2884,7 +2884,16 @@ impl SetDisks {
let result = if let Some(deadline) = single_pending_hedge_deadline.take() {
tokio::select! {
result = join_set.join_next() => result,
_ = tokio::time::sleep_until(deadline) => {
_ = async {
tokio::time::sleep_until(deadline).await;
#[cfg(test)]
rename_fanout_barrier::checkpoint(
object.as_ref(),
0,
rename_fanout_barrier::PHASE_NON_INLINE_HEDGE_TIMER,
)
.await;
} => {
if bounded_fanout
&& !force_full_wait
&& join_set.len() == 1
@@ -7322,6 +7331,9 @@ pub(crate) mod rename_fanout_barrier {
CLEANUP as PHASE_CLEANUP, READ_VERSION as PHASE_READ_VERSION, RENAME as PHASE_RENAME, ROLLBACK as PHASE_ROLLBACK,
};
/// Object-scoped hedge timer checkpoint; slot zero identifies the timer, not a disk.
pub const PHASE_NON_INLINE_HEDGE_TIMER: &str = "non_inline_hedge_timer";
/// One armed barrier: the fan-out task matching `(disk_index, phase)` pauses.
struct Armed {
disk_index: usize,
+2
View File
@@ -1593,6 +1593,8 @@ mod prepared_get_object_metadata_tests {
("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", Some("true")),
],
async {
// Hold this object's hedge timer so real-disk latency cannot add speculative fanout.
let _hedge_timer = rename_fanout_barrier::arm(&object, 0, rename_fanout_barrier::PHASE_NON_INLINE_HEDGE_TIMER);
let calls = disk_call_counters::observe(&object);
let mut reader = set_disks
.get_object_reader(bucket, &object, None, HeaderMap::new(), &opts)
+23 -9
View File
@@ -1166,7 +1166,7 @@ impl SetDisks {
if !latest_meta.deleted && !latest_meta.is_remote() && !protected {
result.detail =
"Legacy object uses standard repair; independent object identity remains unverified".to_owned();
rustfs_heal_contracts::heal_channel::LEGACY_OBJECT_IDENTITY_UNVERIFIED_DETAIL.to_owned();
}
if disks_to_heal_count == 0 {
@@ -4862,18 +4862,33 @@ mod heal_result_report_tests {
},
)
.await
.expect("commit all source shards");
.expect("commit source object");
let mut files = Vec::new();
// PUT can succeed at quorum while minority shards remain unavailable.
// Require every source disk before modifying any shard in this fixture.
let mut source_metadata = Vec::with_capacity(disks.len());
for disk in &disks {
source_metadata.push(disk.read_version("", bucket, object, "", &ReadOptions::default()).await);
}
assert!(
source_metadata.iter().all(|source| source.is_ok()),
"source fixture must populate all {} disks before truncation in {mode:?} mode; per-disk errors: {:?}",
disks.len(),
source_metadata
.iter()
.enumerate()
.map(|(slot, source)| (slot, source.as_ref().err()))
.collect::<Vec<_>>()
);
let files: Vec<_> = source_metadata
.into_iter()
.map(|source| source.expect("source metadata checked above"))
.collect();
let mut paths = Vec::new();
let mut original_shards = Vec::new();
let mut original_metadata = Vec::new();
let mut damaged = Vec::new();
for (slot, disk) in disks.iter().enumerate() {
let source = disk
.read_version("", bucket, object, "", &ReadOptions::default())
.await
.expect("read source metadata");
for (slot, source) in files.iter().enumerate() {
assert_eq!((source.erasure.data_blocks, source.erasure.parity_blocks), (12, 4));
assert_eq!(source.parts[0].integrity.is_some(), mode == ShardIntegrityWriteMode::Protected);
let directory = temp_dirs[slot].path().join(bucket).join(object);
@@ -4905,7 +4920,6 @@ mod heal_result_report_tests {
}
paths.push(path);
original_shards.push(original);
files.push(source);
}
assert_eq!(damaged.len(), coding_indexes.len());
if mode == ShardIntegrityWriteMode::Protected {
+410 -12
View File
@@ -1451,6 +1451,103 @@ impl SetDisks {
delimiter: delimiter.to_owned(),
})
}
/// Answer a CompleteMultipartUpload whose staging upload is already gone.
///
/// AWS keeps Complete idempotent for the same upload id and part list while
/// the completed object is still the result of that upload (the first 200
/// can be lost, and SDK retries depend on a second 200 with the same ETag).
/// Returns `Ok(None)` when this key has no such object, so the caller keeps
/// `InvalidUploadID`. A matching upload id with a different part list is
/// `InvalidPart` and must not fall through to `NoSuchUpload`.
async fn replay_completed_multipart_upload(
&self,
bucket: &str,
object: &str,
upload_id: &str,
uploaded_parts: &[CompletePart],
opts: &ObjectOptions,
) -> Result<Option<ObjectInfo>> {
if upload_id.is_empty() {
return Ok(None);
}
let read_opts = ObjectOptions {
no_lock: true,
metadata_cache_safe: false,
versioned: opts.versioned,
version_suspended: opts.version_suspended,
..Default::default()
};
match self.get_object_info(bucket, object, &read_opts).await {
Ok(info) => {
match completed_multipart_upload_matches(
&info.user_defined,
info.parts.as_ref(),
info.delete_marker,
upload_id,
uploaded_parts,
bucket,
object,
) {
Ok(true) => return Ok(Some(mark_multipart_completion_replayed(info))),
Ok(false) => {}
Err(err) => return Err(err),
}
}
Err(err) if is_err_object_not_found(&err) || is_err_version_not_found(&err) => {}
Err(err) => return Err(err),
}
// Unversioned overwrite replaces the only slot. A versioned bucket can
// keep the completed version under a newer one; that version is still
// the result of this upload until it is deleted.
if !(opts.versioned || opts.version_suspended) {
return Ok(None);
}
let Some(versions) = self.load_file_info_versions_exact(bucket, object).await? else {
return Ok(None);
};
for fi in versions.versions {
if fi.deleted || fi.tier_free_version() || fi.is_canonical_delete_marker() {
continue;
}
match completed_multipart_upload_matches(&fi.metadata, &fi.parts, false, upload_id, uploaded_parts, bucket, object) {
Ok(true) => {
let info = ObjectInfo::from_file_info(&fi, bucket, object, true);
return Ok(Some(mark_multipart_completion_replayed(info)));
}
Ok(false) => {}
Err(err) => return Err(err),
}
}
Ok(None)
}
/// Best-effort removal of staging left behind when commit succeeded and the
/// process stopped before `delete_all`. A cleanup miss must not turn the
/// idempotent 200 into an error; the object is already durable.
async fn reclaim_replayed_multipart_staging(&self, bucket: &str, object: &str, upload_id: &str, upload_id_path: &str) {
if let Err(err) = self
.delete_all_with_quorum(RUSTFS_META_MULTIPART_BUCKET, upload_id_path, self.default_write_quorum())
.await
{
warn!(
target: "rustfs_ecstore::set_disk",
event = EVENT_SET_DISK_MULTIPART,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_SET_DISK,
op = "complete_multipart_upload",
result = "cleanup_quorum_missed",
bucket = %bucket,
object = %object,
upload_id = %upload_id,
error = %err,
"replayed multipart completion left staging behind"
);
}
}
}
#[async_trait::async_trait]
@@ -2337,6 +2434,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
}
result
}
// complete_multipart_upload finished
#[tracing::instrument(skip(self))]
async fn complete_multipart_upload(
@@ -2400,9 +2498,27 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
.await?;
let expected_restore_operation_id = restore_commit_operation_id_from_metadata(&opts.user_defined)?;
let (mut fi, files_metas) = self
let (mut fi, files_metas) = match self
.check_upload_id_exists_with_opts(bucket, object, upload_id, true, opts)
.await?;
.await
{
Ok(found) => found,
Err(err) if crate::error::is_err_invalid_upload_id(&err) => {
match self
.replay_completed_multipart_upload(bucket, object, upload_id, &uploaded_parts, opts)
.await
{
Ok(Some(existing)) => {
self.reclaim_replayed_multipart_staging(bucket, object, upload_id, &upload_id_path)
.await;
return Ok(existing);
}
Ok(None) => return Err(err),
Err(replay_err) => return Err(replay_err),
}
}
Err(err) => return Err(err),
};
ensure_data_movement_upload_access(&fi, bucket, object, upload_id, opts)?;
ensure_multipart_bucket_incarnation(&self.ctx, &fi, bucket, object, upload_id, opts.expected_bucket_incarnation_id)
.await?;
@@ -2988,6 +3104,16 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
rustfs_utils::http::remove_str(&mut fi.metadata, rustfs_filemeta::shard_integrity::SUFFIX_UPLOAD_INTEGRITY);
fi.persist_shard_integrity()?;
// The staging directory is removed after commit. Recording the upload id
// on the object is what lets a retried CompleteMultipartUpload return this
// version instead of NoSuchUpload. insert_str writes both internal prefixes;
// a later conflicting pair fails closed in the replay matcher.
// Internal migration must preserve the source completion identity rather
// than replace it (or invent one) with the temporary transfer upload id.
if !opts.data_movement && !upload_id.is_empty() {
insert_str(&mut fi.metadata, rustfs_utils::http::SUFFIX_MULTIPART_UPLOAD_ID, upload_id.to_owned());
}
for meta in parts_metadatas.iter_mut() {
if meta.has_valid_erasure_geometry() {
meta.size = fi.size;
@@ -3522,6 +3648,58 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
}
}
fn mark_multipart_completion_replayed(mut info: ObjectInfo) -> ObjectInfo {
info.multipart_completion_replayed = true;
info
}
/// `Ok(true)` when `upload_id` is the completion that published this version and
/// the requested parts are that version's part list. `Ok(false)` means this
/// version is not that completion. `Err` is a definite part-list or metadata
/// failure and must not be reported as `NoSuchUpload`.
fn completed_multipart_upload_matches(
metadata: &HashMap<String, String>,
parts: &[ObjectPartInfo],
delete_marker: bool,
upload_id: &str,
uploaded_parts: &[CompletePart],
bucket: &str,
object: &str,
) -> Result<bool> {
if delete_marker || upload_id.is_empty() {
return Ok(false);
}
let Some(stored_upload_id) = rustfs_utils::http::get_consistent_str(metadata, rustfs_utils::http::SUFFIX_MULTIPART_UPLOAD_ID)
else {
if rustfs_utils::http::contains_key_str(metadata, rustfs_utils::http::SUFFIX_MULTIPART_UPLOAD_ID) {
return Err(Error::FileCorrupt);
}
return Ok(false);
};
if stored_upload_id.is_empty() || stored_upload_id != upload_id {
return Ok(false);
}
if parts.len() != uploaded_parts.len() {
let part_num = uploaded_parts.first().map(|part| part.part_num).unwrap_or(0);
return Err(Error::InvalidPart(part_num, bucket.to_owned(), object.to_owned()));
}
for (stored, requested) in parts.iter().zip(uploaded_parts) {
if stored.number != requested.part_num {
return Err(Error::InvalidPart(requested.part_num, bucket.to_owned(), object.to_owned()));
}
let stored_etag = rustfs_utils::path::trim_etag(&stored.etag);
let client_etag = requested.etag.as_deref().map(rustfs_utils::path::trim_etag);
if client_etag.as_deref() != Some(stored_etag.as_str()) {
return Err(Error::InvalidPart(
requested.part_num,
stored.etag.clone(),
requested.etag.clone().unwrap_or_default(),
));
}
}
Ok(true)
}
/// Final ETag for a completed multipart object. An authorized replication
/// request preserves the source ETag so the replication HEAD comparison
/// converges even when the source ETag is not derivable from the uploaded
@@ -4939,6 +5117,10 @@ mod tests {
)
.await
.expect("completed data movement object should be readable");
assert!(
!rustfs_utils::http::contains_key_str(&completed.user_defined, rustfs_utils::http::SUFFIX_MULTIPART_UPLOAD_ID),
"migration of a source without a completion identity must not invent one"
);
assert!(!rustfs_utils::http::contains_key_str(
&completed.user_defined,
rustfs_utils::http::SUFFIX_DATA_MOVEMENT_UPLOAD
@@ -8630,6 +8812,216 @@ mod tests {
assert_eq!(paged, vec!["u0", "u1", "u2", "u3"]);
}
#[test]
fn completed_multipart_upload_match_requires_same_upload_and_parts() {
let upload_id = "upload-1";
let mut metadata = HashMap::new();
insert_str(&mut metadata, rustfs_utils::http::SUFFIX_MULTIPART_UPLOAD_ID, upload_id.to_string());
let parts = vec![ObjectPartInfo {
number: 1,
etag: "abc".to_string(),
..Default::default()
}];
let requested = vec![CompletePart {
part_num: 1,
etag: Some("\"abc\"".to_string()),
..Default::default()
}];
assert!(
completed_multipart_upload_matches(&metadata, &parts, false, upload_id, &requested, "bucket", "object")
.expect("quoted ETag must match the stored part")
);
let wrong_etag = vec![CompletePart {
part_num: 1,
etag: Some("def".to_string()),
..Default::default()
}];
assert!(matches!(
completed_multipart_upload_matches(&metadata, &parts, false, upload_id, &wrong_etag, "bucket", "object"),
Err(StorageError::InvalidPart(1, _, _))
));
assert!(
!completed_multipart_upload_matches(&metadata, &parts, false, "other-upload", &requested, "bucket", "object")
.expect("a different upload id is not this completion")
);
assert!(
!completed_multipart_upload_matches(&metadata, &parts, true, upload_id, &requested, "bucket", "object")
.expect("a delete marker is not the completed object")
);
metadata.insert(
format!(
"{}{}",
rustfs_utils::http::MINIO_INTERNAL_PREFIX,
rustfs_utils::http::SUFFIX_MULTIPART_UPLOAD_ID
),
"disagrees".to_string(),
);
assert!(
matches!(
completed_multipart_upload_matches(&metadata, &parts, false, upload_id, &requested, "bucket", "object"),
Err(StorageError::FileCorrupt)
),
"conflicting internal upload ids must fail closed"
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn retried_complete_multipart_upload_returns_the_committed_object() {
let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
let bucket = "multipart-complete-retry";
let object = "object";
make_bucket_on_all(&disk_stores, bucket).await;
let payload = vec![0x11; 4096];
let (upload_id, parts) =
stage_upload_with_create_opts(&set_disks, bucket, object, &payload, &ObjectOptions::default()).await;
let first = set_disks
.clone()
.complete_multipart_upload(bucket, object, &upload_id, parts.clone(), &ObjectOptions::default())
.await
.expect("the first completion should publish the object");
assert!(!first.multipart_completion_replayed, "the first completion publishes a new object");
assert_eq!(
rustfs_utils::http::get_consistent_str(&first.user_defined, rustfs_utils::http::SUFFIX_MULTIPART_UPLOAD_ID),
Some(upload_id.as_str()),
"the completed object must record the upload id"
);
let mut quoted_parts = parts.clone();
quoted_parts[0].etag = quoted_parts[0].etag.as_ref().map(|etag| format!("\"{etag}\""));
let retried = set_disks
.clone()
.complete_multipart_upload(bucket, object, &upload_id, quoted_parts, &ObjectOptions::default())
.await
.expect("a retry with the same upload id and parts must succeed");
assert!(retried.multipart_completion_replayed);
assert_eq!(retried.etag, first.etag);
assert_eq!(retried.size, first.size);
assert_eq!(retried.version_id, first.version_id);
let mismatched = parts.clone();
let mut mismatched = mismatched;
mismatched[0].etag = Some("not-the-part".to_string());
let mismatch_err = set_disks
.clone()
.complete_multipart_upload(bucket, object, &upload_id, mismatched, &ObjectOptions::default())
.await
.expect_err("a retry with a different part ETag must not pretend the upload is missing");
assert!(matches!(mismatch_err, StorageError::InvalidPart(..)));
let unknown = set_disks
.clone()
.complete_multipart_upload(bucket, object, "not-a-real-upload", parts.clone(), &ObjectOptions::default())
.await
.expect_err("an unknown upload id must stay InvalidUploadID");
assert!(matches!(unknown, StorageError::InvalidUploadID(..)));
let mut reader = PutObjReader::from_vec(b"replaced".to_vec());
set_disks
.put_object(bucket, object, &mut reader, &ObjectOptions::default())
.await
.expect("an overwrite should replace the completed object");
let overwritten = set_disks
.clone()
.complete_multipart_upload(bucket, object, &upload_id, parts, &ObjectOptions::default())
.await
.expect_err("a retry after the object is no longer that upload must stay InvalidUploadID");
assert!(matches!(overwritten, StorageError::InvalidUploadID(..)));
let current = set_disks
.get_object_info(bucket, object, &ObjectOptions::default())
.await
.expect("the overwrite must remain readable");
assert_eq!(current.size, b"replaced".len() as i64);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn concurrent_complete_multipart_upload_returns_one_object() {
let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
let bucket = "multipart-complete-concurrent";
let object = "object";
make_bucket_on_all(&disk_stores, bucket).await;
let (upload_id, parts) =
stage_upload_with_create_opts(&set_disks, bucket, object, &[0x22; 4096], &ObjectOptions::default()).await;
let first_set = set_disks.clone();
let second_set = set_disks.clone();
let first_parts = parts.clone();
let second_parts = parts.clone();
let first_upload = upload_id.clone();
let second_upload = upload_id.clone();
let (first, second) = tokio::join!(
async move {
first_set
.complete_multipart_upload(bucket, object, &first_upload, first_parts, &ObjectOptions::default())
.await
},
async move {
second_set
.complete_multipart_upload(bucket, object, &second_upload, second_parts, &ObjectOptions::default())
.await
},
);
let first = first.expect("one completion must succeed");
let second = second.expect("the duplicate completion must succeed");
assert_eq!(first.etag, second.etag, "both completions must report the same ETag");
assert_eq!(first.version_id, second.version_id);
assert_eq!(
usize::from(first.multipart_completion_replayed) + usize::from(second.multipart_completion_replayed),
1
);
let info = set_disks
.get_object_info(bucket, object, &ObjectOptions::default())
.await
.expect("the object must be readable after duplicate completion");
assert_eq!(info.etag, first.etag);
assert_eq!(info.size, 4096);
}
#[tokio::test]
async fn versioned_complete_retry_returns_the_completed_version() {
let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
let bucket = "multipart-complete-versioned";
let object = "object";
make_bucket_on_all(&disk_stores, bucket).await;
let versioned = ObjectOptions {
versioned: true,
..Default::default()
};
let (upload_id, parts) = stage_upload_with_create_opts(&set_disks, bucket, object, &[0x33; 4096], &versioned).await;
let completed = set_disks
.clone()
.complete_multipart_upload(bucket, object, &upload_id, parts.clone(), &versioned)
.await
.expect("versioned completion should publish a version");
let completed_version = completed.version_id.expect("versioned completion assigns a version id");
let mut reader = PutObjReader::from_vec(b"newer-version".to_vec());
let successor = set_disks
.put_object(bucket, object, &mut reader, &versioned)
.await
.expect("a later versioned put should add a version");
assert_ne!(successor.version_id, Some(completed_version));
let retried = set_disks
.clone()
.complete_multipart_upload(bucket, object, &upload_id, parts, &versioned)
.await
.expect("the retry must still find the version this upload published");
assert!(retried.multipart_completion_replayed);
assert_eq!(retried.etag, completed.etag);
assert_eq!(retried.version_id, Some(completed_version));
let latest = set_disks
.get_object_info(bucket, object, &versioned)
.await
.expect("the latest version must stay the successor put");
assert_eq!(latest.version_id, successor.version_id);
assert_ne!(latest.etag, completed.etag);
}
/// Crash-consistency for the two `complete_multipart_upload` commit windows.
///
/// rustfs/backlog#864: a fault that interrupts a completion must never mutate
@@ -9299,18 +9691,24 @@ mod tests {
"the post-commit crash must leave the upload listable for reclamation"
);
// A retried CompleteMultipartUpload is answered deterministically: the
// commit rename already consumed the upload's metadata, so the retry
// resolves to InvalidUploadID (NoSuchUpload to the S3 client, the
// standard answer for a completed-then-retried upload) — never a torn
// state, and the committed object is untouched by the retry.
let retried = complete(&set_disks, bucket, object, &u_new, parts_retry).await;
// The commit rename already consumed the upload's xl.meta, so the
// retry cannot read staging. It must still return the committed
// object: AWS CompleteMultipartUpload is idempotent for the same
// upload id and part list, and the retry must not tear that object.
let (body_before_retry, etag_before_retry) = read_object(&set_disks, bucket, object).await;
assert_eq!(body_before_retry, new, "a post-commit crash must leave the whole new version readable");
let retried = complete(&set_disks, bucket, object, &u_new, parts_retry)
.await
.expect("a retried complete after the commit landed must return the committed object");
assert!(retried.multipart_completion_replayed, "the retry must not publish a second object");
assert_eq!(retried.etag, etag_before_retry, "the retry must return the committed ETag");
let (body_after_retry, etag_after_retry) = read_object(&set_disks, bucket, object).await;
assert_eq!(body_after_retry, new, "the retry must not disturb the committed object");
assert_eq!(etag_after_retry, etag_before_retry, "the retry must not rewrite the committed ETag");
assert!(
matches!(retried, Err(StorageError::InvalidUploadID(..))),
"a retried complete after the commit landed must resolve to InvalidUploadID, got {retried:?}"
!upload_is_listed(&set_disks, bucket, object, &u_new).await,
"the idempotent retry must reclaim staging left behind by the crash"
);
let (body_after_retry, _) = read_object(&set_disks, bucket, object).await;
assert_eq!(body_after_retry, new, "the failed retry must not disturb the committed object");
// Reclaim the leftover exactly as the production tail does: delete_all
// on the upload path (abort_multipart_upload cannot — the upload's
+22 -4
View File
@@ -1569,12 +1569,27 @@ mod tests {
disk_indexes: &[usize],
) -> Vec<(usize, crate::disk::DiskStore)> {
let offline = {
let mut disks = set.disks.write().await;
let disks = set.disks.read().await;
disk_indexes
.iter()
.map(|index| (*index, disks[*index].take().expect("fault-injection disk should start online")))
.map(|index| (*index, disks[*index].clone().expect("fault-injection disk should start online")))
.collect::<Vec<_>>()
};
for (_, disk) in &offline {
disk.close().await.expect("fault injection should stop per-disk monitoring");
disk.force_offline_for_test();
}
// Empty slots can be renewed from the still-present disk paths by the endpoint monitor.
// Keep the original handles and prove reconnect cannot clear their injected IO failure.
set.connect_disks().await;
{
let disks = set.disks.read().await;
for (index, disk) in &offline {
let current = disks[*index].as_ref().expect("fault-injection slot should remain populated");
assert!(Arc::ptr_eq(current, disk), "reconnect must preserve the fault-injection handle");
assert_eq!(disk.runtime_state(), crate::disk::health_state::RuntimeDriveHealthState::Offline);
}
}
let local_disk_map = ecstore.ctx.local_disk_map();
let mut local_disks = local_disk_map.write().await;
for (_, disk) in &offline {
@@ -1595,9 +1610,12 @@ mod tests {
local_disks.insert(disk.endpoint().to_string(), Some(Arc::clone(disk)));
}
}
let mut disks = set.disks.write().await;
let disks = set.disks.read().await;
for (index, disk) in offline {
assert!(disks[index].replace(disk).is_none(), "fault-injection disk slot should remain empty");
let current = disks[index].as_ref().expect("fault-injection slot should remain populated");
assert!(Arc::ptr_eq(current, &disk), "fault-injection disk handle should remain unchanged");
assert_eq!(disk.runtime_state(), crate::disk::health_state::RuntimeDriveHealthState::Offline);
disk.reset_health_for_store_init_retry();
}
}
+170
View File
@@ -20362,6 +20362,176 @@ mod tests {
}
}
#[cfg(feature = "test-util")]
#[tokio::test]
#[serial_test::serial(storage_class_env)]
async fn tier_mutation_peer_prepare_reloads_after_save_and_load_lock_timeouts() {
use crate::storage_api_contracts::namespace::NamespaceLocking as _;
use std::sync::atomic::Ordering;
let temp_dir = tempfile::tempdir().expect("create contended tier prepare store dir");
let (ctx, store, shutdown) =
without_storage_class_env(build_isolated_test_store(temp_dir.path(), "tier-prepare-contention", &[4])).await;
let mutation_id = uuid::Uuid::new_v4();
let intent = tier_mutation_peer_test_intent(mutation_id, "COLD-A", [9; 32]);
let object = crate::services::tier::tier_mutation_intent::tier_mutation_intent_record_object_name(mutation_id)
.expect("intent record object should resolve");
let barrier = crate::set_disk::PutObjectCommitBarrier::install(
RUSTFS_META_BUCKET,
&object,
crate::set_disk::PutObjectCommitPause::BeforeNamespace,
);
let prepare_store = store.clone();
let payload = intent.encode().expect("prepare intent should encode");
let mut prepare = tokio::spawn(async move {
handle_tier_mutation_peer_request(
prepare_store,
TIER_MUTATION_RPC_PROTOCOL_VERSION,
TierMutationRpcPhase::Prepare,
mutation_id,
&payload,
)
.await
});
barrier.wait_until_paused().await;
let namespace = store.pools[0].disk_set[0]
.new_ns_lock(RUSTFS_META_BUCKET, &object)
.await
.expect("intent record namespace should resolve");
let owner = namespace
.get_write_lock(crate::set_disk::get_lock_acquire_timeout())
.await
.expect("competing publisher should hold the intent namespace");
// The competing publisher already owns the namespace. Bypass only its
// recursive acquisition and drain the durable tail before releasing it.
com::save_config_with_opts(
store.clone(),
&object,
intent.encode().expect("winning intent should encode"),
&ObjectOptions {
no_lock: true,
max_parity: true,
write_completion: crate::object_api::WriteCompletion::TailDrained,
..Default::default()
},
)
.await
.expect("competing publisher should persist the matching intent under its lock");
let manager = ctx
.lock_manager()
.as_fast_lock_manager()
.expect("fixture should use real locks");
let key = rustfs_lock::fast_lock::types::ObjectKey {
bucket: Arc::from(RUSTFS_META_BUCKET),
object: Arc::from(object.as_str()),
version: None,
};
let timeouts = &manager.get_shard(&key).metrics().timeouts;
let before = timeouts.load(Ordering::Relaxed);
barrier.release_and_wait_until_namespace_pending().await;
drop(barrier);
tokio::select! {
result = &mut prepare => panic!("Prepare must retry the contended save and load before completing: {result:?}"),
result = tokio::time::timeout(Duration::from_secs(30), async {
while timeouts.load(Ordering::Relaxed) - before < 2 {
tokio::time::sleep(Duration::from_millis(10)).await;
}
}) => result.expect("the save and subsequent load should each exhaust their lock wait"),
}
assert_eq!(timeouts.load(Ordering::Relaxed) - before, 2);
drop(owner);
let outcome = tokio::time::timeout(Duration::from_secs(30), prepare)
.await
.expect("third prepare round should complete after the competing publisher releases its lock")
.expect("prepare task should join")
.expect("third prepare round should reload the matching durable intent");
assert_eq!(outcome.state, TierMutationPeerState::Prepared);
assert!(!outcome.applied, "the retry must reuse the winning record rather than recreate it");
assert_eq!(
load_tier_mutation_intent_record(store.clone(), mutation_id)
.await
.expect("winning record should remain readable"),
intent
);
assert_eq!(timeouts.load(Ordering::Relaxed) - before, 2);
shutdown.cancel();
}
#[cfg(feature = "test-util")]
#[tokio::test]
#[serial_test::serial(storage_class_env)]
async fn tier_mutation_peer_prepare_preserves_lock_timeout_after_three_rounds() {
use crate::storage_api_contracts::namespace::NamespaceLocking as _;
use std::sync::atomic::Ordering;
let temp_dir = tempfile::tempdir().expect("create exhausted tier prepare store dir");
let (ctx, store, shutdown) =
without_storage_class_env(build_isolated_test_store(temp_dir.path(), "tier-prepare-exhaustion", &[4])).await;
let mutation_id = uuid::Uuid::new_v4();
let intent = tier_mutation_peer_test_intent(mutation_id, "COLD-A", [9; 32]);
let object = crate::services::tier::tier_mutation_intent::tier_mutation_intent_record_object_name(mutation_id)
.expect("intent record object should resolve");
let barrier = crate::set_disk::PutObjectCommitBarrier::install(
RUSTFS_META_BUCKET,
&object,
crate::set_disk::PutObjectCommitPause::BeforeNamespace,
);
let prepare_store = store.clone();
let payload = intent.encode().expect("prepare intent should encode");
let prepare = tokio::spawn(async move {
handle_tier_mutation_peer_request(
prepare_store,
TIER_MUTATION_RPC_PROTOCOL_VERSION,
TierMutationRpcPhase::Prepare,
mutation_id,
&payload,
)
.await
});
barrier.wait_until_paused().await;
let namespace = store.pools[0].disk_set[0]
.new_ns_lock(RUSTFS_META_BUCKET, &object)
.await
.expect("intent record namespace should resolve");
let lock_timeout = crate::set_disk::get_lock_acquire_timeout();
let owner = namespace
.get_write_lock(lock_timeout)
.await
.expect("competing publisher should hold the intent namespace");
let manager = ctx
.lock_manager()
.as_fast_lock_manager()
.expect("fixture should use real locks");
let key = rustfs_lock::fast_lock::types::ObjectKey {
bucket: Arc::from(RUSTFS_META_BUCKET),
object: Arc::from(object.as_str()),
version: None,
};
let timeouts = &manager.get_shard(&key).metrics().timeouts;
let before = timeouts.load(Ordering::Relaxed);
barrier.release_and_wait_until_namespace_pending().await;
drop(barrier);
let error = tokio::time::timeout(Duration::from_secs(30), prepare)
.await
.expect("three contended prepare rounds should exhaust the bounded budget")
.expect("prepare task should join")
.expect_err("Prepare must fail while the competing publisher keeps the namespace");
match error {
TierMutationPeerError::Store(Error::Lock(rustfs_lock::LockError::Timeout { resource, timeout })) => {
assert_eq!(resource, key.to_string());
assert_eq!(timeout, lock_timeout);
}
other => panic!("Prepare must preserve the final typed lock timeout: {other:?}"),
}
assert_eq!(timeouts.load(Ordering::Relaxed) - before, 3, "save and load must share three rounds");
drop(owner);
assert!(matches!(
load_tier_mutation_intent_record(store, mutation_id).await,
Err(Error::ConfigNotFound)
));
shutdown.cancel();
}
#[cfg(feature = "test-util")]
#[tokio::test]
#[serial_test::serial(storage_class_env)]
+4 -1
View File
@@ -403,7 +403,10 @@ async fn enqueue_transition_after_write(
) -> Result<ObjectInfo> {
match result {
Ok(oi) => {
if should_enqueue_transition_immediately(&oi) {
// A retried completion did not publish a new object. Immediate
// transition/expiry already ran for the original commit; the scanner
// still sees the object if that commit never reached this hook.
if !oi.multipart_completion_replayed && should_enqueue_transition_immediately(&oi) {
enqueue_transition_immediate(&oi, src.clone()).await;
if let Ok(api) = metadata_sys::object_store_in(&store.ctx).await {
enqueue_immediate_expiry(api, &oi, src, opts).await;
+26 -4
View File
@@ -993,10 +993,32 @@ impl ECStore {
.await;
}
let pool_idx = self.multipart_upload_pool_idx(bucket, object, upload_id, opts).await?;
let pool = self.pools[pool_idx].clone();
pool.complete_multipart_upload(bucket, object, upload_id, uploaded_parts, opts)
.await
match self.multipart_upload_pool_idx(bucket, object, upload_id, opts).await {
Ok(pool_idx) => {
self.pools[pool_idx]
.clone()
.complete_multipart_upload(bucket, object, upload_id, uploaded_parts, opts)
.await
}
// The staging upload is already gone. The completed object lives on
// the pool that published it; other pools answer InvalidUploadID.
Err(err) if is_err_invalid_upload_id(&err) => {
let mut missing = err;
for pool_idx in self.existing_multipart_pool_order().await {
match self.pools[pool_idx]
.clone()
.complete_multipart_upload(bucket, object, upload_id, uploaded_parts.clone(), opts)
.await
{
Ok(info) => return Ok(info),
Err(err) if is_err_invalid_upload_id(&err) => missing = err,
Err(err) => return Err(err),
}
}
Err(missing)
}
Err(err) => Err(err),
}
}
#[cfg(all(test, feature = "test-util"))]
@@ -24,6 +24,12 @@ pub const HEAL_DELETE_DANGLING: bool = true;
pub const RUSTFS_RESERVED_BUCKET: &str = "rustfs";
pub const RUSTFS_RESERVED_BUCKET_PATH: &str = "/rustfs";
/// Detail attached to a completed deep heal when a healthy legacy object has
/// no independent identity commitment. Durable MRF handling uses this exact
/// reason to pause proofless retries without treating the object as repaired.
pub const LEGACY_OBJECT_IDENTITY_UNVERIFIED_DETAIL: &str =
"Legacy object uses standard repair; independent object identity remains unverified";
#[derive(Clone, Copy, Debug, Serialize, Deserialize)]
pub enum HealItemType {
Metadata,
+1
View File
@@ -99,6 +99,7 @@ crc-fast = { workspace = true }
sha2 = { workspace = true }
[dev-dependencies]
tonic = { workspace = true }
rustfs-filemeta = { workspace = true }
serde_json = { workspace = true, features = ["raw_value"] }
rustfs-test-utils = { workspace = true }
+38
View File
@@ -198,6 +198,11 @@ impl Error {
}
fn is_recoverable_internode_error(error: &std::io::Error) -> bool {
// A peer restart can cancel an RPC after a partial repair. Replay it within
// the existing heal retry budget; task cancellation remains terminal.
if DiskError::io_error_is_rpc_cancelled(error) {
return true;
}
let Some(error) = error.get_ref().and_then(|source| source.downcast_ref::<InternodeHttpError>()) else {
return false;
};
@@ -259,6 +264,39 @@ mod tests {
use super::Error;
use crate::heal::{DiskError, EcstoreError};
#[test]
fn cancelled_rpc_is_recoverable_across_storage_error_conversions() {
let status = tonic::Status::cancelled("operation was canceled");
let disk = DiskError::from(status.clone());
let storage = EcstoreError::from(status.clone());
let disk_storage = EcstoreError::from(DiskError::from(status.clone()));
for error in [
Error::Disk(disk.clone()),
Error::Disk(disk),
Error::Storage(storage.clone()),
Error::Storage(storage),
Error::Storage(disk_storage.clone()),
Error::Storage(disk_storage),
Error::Io(std::io::Error::from(DiskError::from(status))),
] {
assert!(error.is_recoverable_heal(), "typed RPC cancellation must be recoverable: {error:?}");
}
for status in [
tonic::Status::permission_denied("operation was canceled"),
tonic::Status::unauthenticated("operation was canceled"),
tonic::Status::invalid_argument("operation was canceled"),
tonic::Status::internal("operation was canceled"),
] {
assert!(
!Error::Storage(EcstoreError::from(status).clone()).is_recoverable_heal(),
"cancellation text alone must not change application error classification"
);
}
assert!(!Error::Io(std::io::Error::other("operation was canceled")).is_recoverable_heal());
assert!(!Error::TaskCancelled.is_recoverable_heal());
assert!(!Error::TaskTimeout.is_recoverable_heal());
}
#[test]
fn internode_transport_errors_keep_their_recovery_classification() {
use rustfs_rio::{InternodeHttpErrorKind as Kind, new_test_internode_http_io_error};
+21 -2
View File
@@ -2254,6 +2254,8 @@ mod resume_loop_tests {
/// A transient infrastructure condition (offline disk / unmet quorum):
/// the version must be recorded as skipped and retried on a later pass.
Transient,
RpcCancelled(u32),
Cancelled,
Timeout,
}
@@ -2456,14 +2458,31 @@ mod resume_loop_tests {
.unwrap()
.push((object.to_string(), version_id.map(str::to_string)));
let key = compose_key(object, version_id);
let outcome = self.outcomes.lock().unwrap().get(&key).cloned().unwrap_or(HealOutcome::Ok);
let outcome = {
let mut outcomes = self.outcomes.lock().unwrap();
let outcome = outcomes.get(&key).cloned().unwrap_or(HealOutcome::Ok);
if let HealOutcome::RpcCancelled(remaining) = &outcome {
outcomes.insert(key.clone(), HealOutcome::RpcCancelled(remaining.saturating_sub(1)));
}
outcome
};
match outcome {
HealOutcome::Ok => Ok((self.results.lock().unwrap().get(&key).cloned().unwrap_or_default(), None)),
HealOutcome::Ok | HealOutcome::RpcCancelled(0) => {
Ok((self.results.lock().unwrap().get(&key).cloned().unwrap_or_default(), None))
}
HealOutcome::FileNotFound => Ok((HealResultItem::default(), Some(Error::Storage(EcstoreError::FileNotFound)))),
HealOutcome::VersionNotFound => {
Ok((HealResultItem::default(), Some(Error::Storage(EcstoreError::FileVersionNotFound))))
}
HealOutcome::Transient => Ok((HealResultItem::default(), Some(Error::Storage(EcstoreError::DiskNotFound)))),
HealOutcome::RpcCancelled(_) => {
// Pool aggregation clones the selected error before returning it to heal.
let error = EcstoreError::from(tonic::Status::cancelled("injected peer cancellation"));
let cloned = error.clone();
assert_eq!(cloned, error, "pool error clone must preserve its kind and message");
Err(Error::Storage(cloned))
}
HealOutcome::Cancelled => Err(Error::TaskCancelled),
HealOutcome::Timeout => Err(Error::TaskTimeout),
}
}
@@ -13,8 +13,8 @@
// limitations under the License.
use super::*;
use crate::heal::outcome::{HealObjectDisposition, HealTaskOutcome};
use crate::heal::task::{HealOptions, HealPriority, HealRequest, HealTask, HealType};
use crate::heal::outcome::{HealDeferredReason, HealObjectDisposition, HealObjectIdentity, HealObjectKind, HealTaskOutcome};
use crate::heal::task::{HealOptions, HealPriority, HealRequest, HealTask, HealType, MAX_BUCKET_OBJECT_HEAL_RETRIES};
use rustfs_heal_contracts::heal_channel::HealScanMode;
fn request() -> HealRequest {
@@ -61,6 +61,144 @@ fn assert_partition(outcome: &HealTaskOutcome) {
assert!(c.unknown <= c.skipped);
}
fn admin_object_identity() -> HealObjectIdentity {
HealObjectIdentity {
kind: HealObjectKind::Object,
bucket: "a".to_owned(),
object: "object".to_owned(),
version_id: Some("null".to_owned()),
bucket_incarnation_id: Some(uuid::Uuid::from_u128(42)),
pool_index: None,
set_index: None,
}
}
#[tokio::test(start_paused = true)]
async fn admin_object_rpc_cancelled_retries_require_verified_receipt() {
for failures in [1, MAX_BUCKET_OBJECT_HEAL_RETRIES] {
for verified in [true, false] {
let storage = FakeStorage::default();
storage.set_outcome("object", Some("null"), HealOutcome::RpcCancelled(failures));
storage
.receipts
.lock()
.expect("receipt fixture")
.insert(compose_key("object", Some("null")), verified.then_some(HealObjectDisposition::Repaired));
let options = HealOpts::default();
let cancel = CancellationToken::new();
let identity = admin_object_identity();
let heal = super::super::admin::heal_object(&storage, &options, identity.clone(), "rpc-retry", &cancel);
tokio::pin!(heal);
for attempt in 1..=failures {
assert!(matches!(futures::poll!(heal.as_mut()), std::task::Poll::Pending));
assert_eq!(storage.calls().len(), usize::try_from(attempt).expect("attempt fits"));
tokio::time::advance(HealTask::bucket_object_retry_delay("rpc-retry", attempt)).await;
}
let ((_, result), record) = heal.await;
assert!(matches!(result, Ok(true)));
assert_eq!(storage.calls().len(), usize::try_from(failures + 1).expect("attempts fit"));
let (outcome, attempt_failures) = record.expect("completed attempt records its disposition");
assert_eq!(outcome.identity, identity);
assert_eq!(attempt_failures, failures);
assert_eq!(
outcome.disposition,
if verified {
HealObjectDisposition::Repaired
} else {
HealObjectDisposition::Unknown
},
"only a matching positive receipt proves repair after retries"
);
assert!(outcome.detail.is_none());
}
}
}
#[tokio::test(start_paused = true)]
async fn admin_object_rpc_cancelled_exhausts_bounded_retries_as_deferred() {
let storage = FakeStorage::default();
storage.set_outcome("object", Some("null"), HealOutcome::RpcCancelled(MAX_BUCKET_OBJECT_HEAL_RETRIES + 1));
let options = HealOpts::default();
let cancel = CancellationToken::new();
let identity = admin_object_identity();
let heal = super::super::admin::heal_object(&storage, &options, identity.clone(), "rpc-exhausted", &cancel);
tokio::pin!(heal);
for attempt in 1..=MAX_BUCKET_OBJECT_HEAL_RETRIES {
assert!(matches!(futures::poll!(heal.as_mut()), std::task::Poll::Pending));
assert_eq!(storage.calls().len(), usize::try_from(attempt).expect("attempt fits"));
tokio::time::advance(HealTask::bucket_object_retry_delay("rpc-exhausted", attempt)).await;
}
let ((_, result), record) = heal.await;
assert!(matches!(result, Err(Error::TransientSkip { .. })));
assert_eq!(
storage.calls().len(),
usize::try_from(MAX_BUCKET_OBJECT_HEAL_RETRIES + 1).expect("attempts fit")
);
let (outcome, attempt_failures) = record.expect("exhausted retry records a deferred outcome");
assert_eq!(outcome.identity, identity);
assert_eq!(attempt_failures, MAX_BUCKET_OBJECT_HEAL_RETRIES + 1);
assert_eq!(
outcome.disposition,
HealObjectDisposition::Deferred {
reason: HealDeferredReason::TransientExistenceCheck,
retry_not_before: None,
}
);
assert!(outcome.detail.is_some());
}
#[tokio::test(start_paused = true)]
async fn admin_object_rpc_cancelled_backoff_honors_task_cancellation() {
let storage = FakeStorage::default();
storage.set_outcome("object", Some("null"), HealOutcome::RpcCancelled(MAX_BUCKET_OBJECT_HEAL_RETRIES + 1));
let options = HealOpts::default();
let cancel = CancellationToken::new();
let started = tokio::time::Instant::now();
let heal = super::super::admin::heal_object(&storage, &options, admin_object_identity(), "rpc-cancel", &cancel);
tokio::pin!(heal);
assert!(matches!(futures::poll!(heal.as_mut()), std::task::Poll::Pending));
assert_eq!(storage.calls().len(), 1, "the first peer cancellation has reached retry backoff");
cancel.cancel();
let ((_, result), record) = heal.await;
assert!(matches!(result, Err(Error::TaskCancelled)));
assert!(record.is_none(), "task cancellation must not publish an object disposition");
assert_eq!(storage.calls().len(), 1, "cancellation must prevent another storage call");
assert_eq!(tokio::time::Instant::now(), started, "cancellation must not wait for backoff");
}
#[tokio::test(start_paused = true)]
async fn admin_object_task_cancellation_and_timeout_remain_terminal() {
for timeout in [false, true] {
let storage = FakeStorage::default();
storage.set_outcome(
"object",
Some("null"),
if timeout {
HealOutcome::Timeout
} else {
HealOutcome::Cancelled
},
);
let started = tokio::time::Instant::now();
let ((_, result), record) = super::super::admin::heal_object(
&storage,
&HealOpts::default(),
admin_object_identity(),
"task-terminal",
&CancellationToken::new(),
)
.await;
if timeout {
assert!(matches!(result, Err(Error::TaskTimeout)));
} else {
assert!(matches!(result, Err(Error::TaskCancelled)));
}
assert!(record.is_none());
assert_eq!(storage.calls().len(), 1, "terminal task errors must not retry");
assert_eq!(tokio::time::Instant::now(), started);
}
}
#[tokio::test]
async fn admin_erasure_two_buckets_have_five_exact_receipts() {
let (_temp, storage) = fixture().await;
+2
View File
@@ -174,6 +174,7 @@ pub(super) struct MrfRepairNoticeTarget {
pub(super) scope: Option<rustfs_common::mrf_channel::MrfScope>,
pub(super) delete_marker_purge: Option<rustfs_common::mrf_channel::MrfDeleteMarkerPurgeIdentity>,
pub(super) lease: Option<rustfs_common::mrf_channel::MrfIngressLease>,
pub(super) durable_anchor: Option<rustfs_common::mrf_channel::MrfDurableRepairAnchor>,
}
#[derive(Debug, Clone)]
@@ -1808,6 +1809,7 @@ impl HealManager {
scope: None,
delete_marker_purge: None,
lease: None,
durable_anchor: None,
},
)
.await
+61
View File
@@ -438,6 +438,7 @@ impl HealManager {
release_mrf_repair_notice_targets(&notice_targets);
}
publish_verified_mrf_repair_events(&notice_targets, &completed_status_for_verified_events);
publish_unverified_legacy_mrf_events(&notice_targets, &completed_status_for_verified_events);
// update statistics
let mut stats = statistics_clone.write().await;
match completed_status {
@@ -880,6 +881,66 @@ pub(super) fn publish_verified_mrf_repair_events(targets: &[MrfRepairNoticeTarge
}
}
pub(super) fn unverified_legacy_mrf_event_for_target(
target: &MrfRepairNoticeTarget,
outcome: &crate::heal::outcome::HealObjectOutcome,
) -> Option<rustfs_common::mrf_channel::MrfUnverifiedLegacyEvent> {
use crate::heal::outcome::{HealObjectDisposition, HealObjectKind};
use rustfs_common::mrf_channel::MrfKind;
let anchor = target.durable_anchor.as_ref()?;
if target.kind != MrfKind::PartialWrite
|| outcome.disposition != HealObjectDisposition::Unknown
|| outcome.detail.as_deref() != Some(rustfs_heal_contracts::heal_channel::LEGACY_OBJECT_IDENTITY_UNVERIFIED_DETAIL)
{
return None;
}
let version_id = target.version_id.filter(|bytes| *bytes != [0; 16]);
let expected_version = version_id.map(|bytes| uuid::Uuid::from_bytes(bytes).to_string());
let expected_pool = target.scope.and_then(|scope| usize::try_from(scope.pool_index).ok());
let expected_set = target.scope.and_then(|scope| usize::try_from(scope.set_index).ok());
if anchor.kind != target.kind
|| anchor.bucket != target.bucket
|| anchor.object != target.object
|| anchor.version_id != version_id
|| anchor.scope != target.scope
|| Some(anchor.lease) != target.lease
|| outcome.identity.kind != HealObjectKind::Object
|| outcome.identity.bucket != target.bucket.as_ref()
|| outcome.identity.object != target.object.as_ref()
|| outcome.identity.version_id != expected_version
|| outcome.identity.pool_index != expected_pool
|| outcome.identity.set_index != expected_set
|| outcome.identity.bucket_incarnation_id != Some(anchor.bucket_incarnation_id)
{
return None;
}
Some(rustfs_common::mrf_channel::MrfUnverifiedLegacyEvent { anchor: anchor.clone() })
}
pub(super) fn publish_unverified_legacy_mrf_events(targets: &[MrfRepairNoticeTarget], completed: &CompletedHealStatus) {
use crate::heal::outcome::HealExecutionOutcome;
if !matches!(completed.status, HealTaskStatus::Completed | HealTaskStatus::Failed { .. }) {
return;
}
let Some(outcome) = completed.outcome.as_ref() else {
return;
};
if outcome.execution != HealExecutionOutcome::Completed {
return;
}
for target in targets {
if let Some(event) = outcome
.objects
.iter()
.find_map(|object| unverified_legacy_mrf_event_for_target(target, object))
{
rustfs_common::mrf_channel::note_mrf_unverified_legacy(event);
}
}
}
pub(super) fn heal_request_set_key_for_task(task: &HealTask) -> Option<String> {
match &task.heal_type {
HealType::ErasureSet { set_disk_id, .. } => Some(set_disk_id.clone()),
+98
View File
@@ -1228,6 +1228,7 @@ fn mrf_verified_repair_event_requires_positive_exact_identity() {
}),
delete_marker_purge: None,
lease: None,
durable_anchor: None,
};
let matching = HealObjectOutcome {
identity: HealObjectIdentity {
@@ -1335,6 +1336,102 @@ fn mrf_verified_repair_event_requires_positive_exact_identity() {
}
}
#[test]
fn unverified_legacy_notice_requires_the_exact_partial_write_target() {
use crate::heal::outcome::{HealObjectIdentity, HealObjectKind, HealObjectOutcome};
use rustfs_common::mrf_channel::{
MrfDurableRepairAnchor, MrfIngressResult, MrfIntent, MrfKind, MrfScope, try_rearm_mrf_replay_intent,
};
let bucket = Arc::<str>::from("legacy-held-bucket");
let object = Arc::<str>::from("legacy-held-object");
let scope = MrfScope {
pool_index: 1,
set_index: 2,
};
let mut intent = MrfIntent {
bucket: bucket.clone(),
object: object.clone(),
version_id: None,
kind: MrfKind::PartialWrite,
delete_marker_purge: None,
scope: Some(scope),
lease: None,
enqueued_at_ms: 1,
attempts: 0,
};
assert_eq!(try_rearm_mrf_replay_intent(&mut intent), MrfIngressResult::Enqueued);
let lease = intent.lease.expect("replayed intent should own a generation");
let anchor = MrfDurableRepairAnchor {
kind: MrfKind::PartialWrite,
bucket: bucket.clone(),
object: object.clone(),
version_id: None,
scope: Some(scope),
delete_marker_purge: None,
lease,
bucket_incarnation_id: uuid::Uuid::new_v4(),
};
let target = MrfRepairNoticeTarget {
bucket: bucket.clone(),
object: object.clone(),
version_id: None,
kind: MrfKind::PartialWrite,
scope: Some(scope),
delete_marker_purge: None,
lease: Some(lease),
durable_anchor: Some(anchor.clone()),
};
let incarnation = anchor.bucket_incarnation_id;
let outcome = HealObjectOutcome {
identity: HealObjectIdentity {
kind: HealObjectKind::Object,
bucket: bucket.to_string(),
object: object.to_string(),
version_id: None,
bucket_incarnation_id: Some(incarnation),
pool_index: Some(1),
set_index: Some(2),
},
disposition: HealObjectDisposition::Unknown,
detail: Some(rustfs_heal_contracts::heal_channel::LEGACY_OBJECT_IDENTITY_UNVERIFIED_DETAIL.to_string()),
};
let event = super::scheduler::unverified_legacy_mrf_event_for_target(&target, &outcome)
.expect("exact legacy partial-write outcome should park only its runtime retry");
assert_eq!(event.anchor, anchor);
let wrong_identity = HealObjectOutcome {
identity: HealObjectIdentity {
object: "other-object".to_string(),
..outcome.identity.clone()
},
..outcome.clone()
};
assert!(super::scheduler::unverified_legacy_mrf_event_for_target(&target, &wrong_identity).is_none());
let missing_incarnation = HealObjectOutcome {
identity: HealObjectIdentity {
bucket_incarnation_id: None,
..outcome.identity.clone()
},
..outcome.clone()
};
assert!(super::scheduler::unverified_legacy_mrf_event_for_target(&target, &missing_incarnation).is_none());
let wrong_incarnation = HealObjectOutcome {
identity: HealObjectIdentity {
bucket_incarnation_id: Some(uuid::Uuid::new_v4()),
..outcome.identity.clone()
},
..outcome.clone()
};
assert!(super::scheduler::unverified_legacy_mrf_event_for_target(&target, &wrong_incarnation).is_none());
let wrong_reason = HealObjectOutcome {
detail: Some("object was readable".to_string()),
..outcome
};
assert!(super::scheduler::unverified_legacy_mrf_event_for_target(&target, &wrong_reason).is_none());
}
#[test]
fn completed_mrf_notice_publishes_only_verified_positive_events() {
use crate::heal::outcome::{HealObjectIdentity, HealObjectKind, HealObjectOutcome, HealTaskOutcome};
@@ -1355,6 +1452,7 @@ fn completed_mrf_notice_publishes_only_verified_positive_events() {
}),
delete_marker_purge: None,
lease: None,
durable_anchor: None,
};
let mismatch_target = MrfRepairNoticeTarget {
object: Arc::from("object-b"),
+163 -29
View File
@@ -20,7 +20,9 @@
//! cannot discharge them; only an exact storage-verified proof may do so.
//! Both paths share the existing committed snapshot format and legacy mirrors.
//! Replay retains partial-write intents for live retries when a member is
//! still offline at startup. A lost proof causes another repair, not deletion.
//! still offline at startup. Healthy legacy objects without identity proof are
//! held in memory after one check; their unchanged journal records are retried
//! on process restart. A lost proof causes another repair, not deletion.
use super::{DiskStore, HealDiskExt as _, local_disk_map_read};
use crate::heal::manager::{HealManager, MrfRepairNoticeTarget};
@@ -584,7 +586,11 @@ pub(crate) fn build_heal_request(intent: &MrfIntent) -> HealRequest {
request
}
async fn submit_mrf_heal_request(manager: &HealManager, intent: &MrfIntent) -> crate::Result<HealAdmissionResult> {
async fn submit_mrf_heal_request(
manager: &HealManager,
intent: &MrfIntent,
durable_anchor: Option<MrfDurableRepairAnchor>,
) -> crate::Result<HealAdmissionResult> {
let receipt = manager
.submit_mrf_heal_request_with_receipt_and_identity(
build_heal_request(intent),
@@ -596,6 +602,7 @@ async fn submit_mrf_heal_request(manager: &HealManager, intent: &MrfIntent) -> c
scope: intent.scope,
delete_marker_purge: intent.delete_marker_purge.as_ref().map(MrfDeleteMarkerPurge::identity),
lease: intent.lease,
durable_anchor,
},
)
.await?;
@@ -763,7 +770,7 @@ impl MrfRuntime {
// attempts counter) changes the encoded snapshot; mark it dirty
// either way.
self.dirty = true;
match submit_mrf_heal_request(manager, &intent).await {
match submit_mrf_heal_request(manager, &intent, None).await {
// Accepted intents leave the pending set; the next flush persists the
// smaller snapshot. This is not a durable successor receipt and
// does not discharge the producer's existing retry hints.
@@ -796,8 +803,7 @@ impl MrfRuntime {
}
}
}
gauge!("rustfs_heal_mrf_queue_depth").set(metric_f64(self.queue.depth() + self.partial_writes.depth()));
gauge!("rustfs_heal_mrf_queue_bytes").set(metric_f64(self.queue.bytes() + self.partial_writes.bytes()));
self.publish_metrics();
}
fn retained_replay_journal(&self) -> bool {
@@ -856,12 +862,38 @@ impl MrfRuntime {
let mut buckets: Vec<Arc<str>> = anchors.iter().map(|anchor| anchor.bucket.clone()).collect();
buckets.sort_unstable();
buckets.dedup();
for bucket in buckets {
for bucket in &buckets {
rustfs_common::mrf_channel::consume_recorded_verified_mrf_repair_events_for(bucket.as_ref(), &mut anchors);
}
let remaining: HashSet<_> = anchors.into_iter().collect();
self.durable_replay_anchors.retain(|anchor| remaining.contains(anchor));
self.dirty |= self.partial_writes.retain_unproven(&remaining);
for bucket in buckets {
for event in rustfs_common::mrf_channel::take_mrf_unverified_legacy_events_for(bucket.as_ref()) {
// Parking is an in-memory dispatch decision. The unchanged
// durable journal remains the recovery/retry authority.
self.partial_writes.park_unverified_legacy(&event.anchor);
}
}
self.publish_metrics();
}
fn publish_metrics(&self) {
gauge!("rustfs_heal_mrf_queue_depth").set(metric_f64(self.queue.depth() + self.partial_writes.depth()));
gauge!("rustfs_heal_mrf_queue_bytes").set(metric_f64(self.queue.bytes() + self.partial_writes.bytes()));
let held = self.partial_writes.unverified_legacy_count();
gauge!("rustfs_heal_mrf_unverified_legacy").set(metric_f64(held));
let now_ms = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|duration| u64::try_from(duration.as_millis()).unwrap_or(u64::MAX))
.unwrap_or_default();
let oldest_age_seconds = self
.partial_writes
.oldest_unverified_legacy_enqueued_at_ms()
.map(|enqueued_at_ms| now_ms.saturating_sub(enqueued_at_ms) / 1_000)
.unwrap_or_default();
gauge!("rustfs_heal_mrf_unverified_legacy_oldest_age_seconds")
.set(metric_f64(usize::try_from(oldest_age_seconds).unwrap_or(usize::MAX)));
}
}
@@ -898,17 +930,38 @@ pub fn spawn_mrf_consumer(manager: Arc<HealManager>) {
tracing::info!(target: "rustfs::heal::mrf", "MRF intent consumer started");
}
/// Replay the durable journal into a fresh pending queue and submit whatever
/// it armed. Returns the number of intact intents replayed. Duplicates are
/// merged by the manager's dedup key; the journal is retained whenever replay
/// cannot fully hand off a successor in-memory snapshot (torn tails truncate
/// via the per-record CRC). Public for integration tests; the live consumer
/// invokes this through [`replay_into`] at startup.
/// Replay the journal once for integration tests, publishing a durable
/// successor before dispatching partial writes. Returns the number of intact
/// intents replayed. Ordinary-only admission retries remain in the startup journal;
/// this helper does not run the consumer's retry loop or proof-driven cleanup.
/// The live consumer invokes [`replay_into`] directly at startup.
pub async fn replay_journal_once(manager: &Arc<HealManager>) -> usize {
let config = MrfConsumerConfig::default();
let mut queue = MrfQueue::new(config.queue_capacity, config.journal_max_bytes);
let mut backoff_until: Option<tokio::time::Instant> = None;
replay_into(manager, &mut queue, &mut backoff_until).await.replayed
let replay = replay_into(manager, &mut queue, &mut backoff_until).await;
if replay.partial_writes.is_empty() {
return replay.replayed;
}
let mut runtime = MrfRuntime {
partial_writes: PartialWrites::default(),
queue,
config,
checkpoint_owner: Uuid::new_v4(),
next_checkpoint_sequence: replay.next_checkpoint_sequence,
new_since_flush: 0,
dirty: false,
journal_on_disk: replay.journal_on_disk,
retain_replay_journal: replay.retain_journal_for_replay,
durable_replay_anchors: replay.durable_replay_anchors,
replay_cleanup: replay.cleanup,
runtime_checkpoint: None,
backoff_until,
};
runtime.adopt_replayed_partial_writes(replay.partial_writes);
runtime.flush().await;
runtime.dispatch(manager).await;
replay.replayed
}
struct ReplayOutcome {
@@ -1081,16 +1134,16 @@ async fn replay_into(
break;
}
if intent.kind.is_durable() {
// Preserve the executable record as well as its proof anchor:
// a target that is still offline during replay needs live retries.
// Adopt durable replay into its single checkpointed owner. The
// runtime dispatches it after publishing the successor, avoiding
// a duplicate task in the startup-replay and steady-state paths.
if let Some(anchor) = manager.durable_mrf_repair_anchor(&intent).await {
durable_replay_anchors.push(anchor);
durable_replay_anchors.push(anchor.clone());
}
let _ = submit_mrf_heal_request(manager, &intent).await;
partial_writes.push(intent);
continue;
}
match submit_mrf_heal_request(manager, &intent).await {
match submit_mrf_heal_request(manager, &intent, None).await {
Ok(HealAdmissionResult::Accepted) | Ok(HealAdmissionResult::Merged) => {
if let Some(anchor) = manager.durable_mrf_repair_anchor(&intent).await {
durable_replay_anchors.push(anchor);
@@ -1196,6 +1249,7 @@ async fn run_mrf_consumer(
if runtime.dirty {
runtime.flush().await;
}
runtime.publish_metrics();
let mut flush_tick = tokio::time::interval(runtime.config.flush_interval);
flush_tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
@@ -1301,7 +1355,7 @@ async fn run_mrf_consumer(
}
TickAction::Idle => {}
}
gauge!("rustfs_heal_mrf_queue_depth").set(metric_f64(runtime.queue.depth() + runtime.partial_writes.depth()));
runtime.publish_metrics();
}
}
}
@@ -1518,15 +1572,15 @@ mod tests {
let mut backoff_until = None;
let replay = replay_into(&manager, &mut queue, &mut backoff_until).await;
assert_eq!(replay.replayed, 1, "W13 committed checkpoint must replay one record");
assert_eq!(queue.depth(), 0, "W13 replayed record should reach the manager before cleanup");
assert_eq!(queue.depth(), 0, "W13 durable replay must leave the ordinary queue");
assert_eq!(replay.partial_writes.len(), 1, "W13 replay must retain its executable durable record");
assert_eq!(replay.durable_replay_anchors.len(), 1, "W13 replay must create a proof anchor");
assert_eq!(
manager.operations_snapshot().await.queued_by_source.mrf,
1,
"W13 replayed work must be visible as MRF manager work"
0,
"W13 durable replay must wait for its successor checkpoint before dispatch"
);
let anchor = replay.durable_replay_anchors[0].clone();
let mut runtime = MrfRuntime {
partial_writes: PartialWrites::default(),
queue,
@@ -1542,6 +1596,45 @@ mod tests {
runtime_checkpoint: None,
backoff_until,
};
runtime.adopt_replayed_partial_writes(replay.partial_writes);
assert_eq!(runtime.partial_writes.depth(), 1, "durable replay must have one executable owner");
assert!(
runtime.durable_replay_anchors.is_empty(),
"adoption must remove the duplicate startup anchor"
);
runtime.dispatch(&manager).await;
assert_eq!(
manager.operations_snapshot().await.queued_by_source.mrf,
0,
"unpersisted replay must not enter the manager"
);
assert!(runtime.flush().await, "publish the durable replay successor before dispatch");
let successor = snapshot::inspect_local_committed_snapshot(runtime.config.journal_max_bytes)
.await
.expect("inspect durable replay successor")
.expect("durable replay must publish a committed successor");
assert_eq!((successor.owner(), successor.sequence()), (runtime.checkpoint_owner, 12));
assert_eq!(runtime.runtime_checkpoint, Some((runtime.checkpoint_owner, 12)));
assert!(
env.disk_paths.iter().all(|path| {
[".heal-mrf-commit.0.bin", ".heal-mrf-commit.1.bin"]
.iter()
.all(|manifest| path.join(".rustfs.sys").join(manifest).exists())
}),
"both startup and successor checkpoints must remain before proof"
);
runtime.dispatch(&manager).await;
assert_eq!(
manager.operations_snapshot().await.queued_by_source.mrf,
1,
"the checkpointed replay must be visible as one MRF manager request"
);
let anchor = runtime
.partial_writes
.anchors()
.next()
.expect("dispatched replay has a proof anchor")
.clone();
let retained_before_proof = runtime.retained_replay_journal();
assert!(retained_before_proof, "W13 proof anchor must retain replay checkpoint before proof");
assert!(
@@ -2436,12 +2529,13 @@ mod tests {
let mut backoff_until = None;
let replay = replay_into(&manager, &mut queue, &mut backoff_until).await;
assert_eq!(replay.replayed, 1, "the committed replay checkpoint must decode one record");
assert_eq!(queue.depth(), 0, "the replayed record must be admitted before cleanup is considered");
assert_eq!(queue.depth(), 0, "durable replay must leave the ordinary queue");
assert_eq!(replay.partial_writes.len(), 1, "replay must retain its executable durable record");
assert!(backoff_until.is_none(), "the accepted replay must not arm admission backoff");
assert_eq!(
manager.operations_snapshot().await.queued_by_source.mrf,
1,
"the replayed record must be visible as an MRF manager request"
0,
"durable replay must wait for its successor checkpoint before dispatch"
);
assert!(
replay.journal_on_disk,
@@ -2465,7 +2559,6 @@ mod tests {
"cleanup must remember the committed checkpoint generation read at startup"
);
let anchor = replay.durable_replay_anchors[0].clone();
let mut runtime = MrfRuntime {
partial_writes: PartialWrites::default(),
queue,
@@ -2481,6 +2574,45 @@ mod tests {
runtime_checkpoint: None,
backoff_until,
};
runtime.adopt_replayed_partial_writes(replay.partial_writes);
assert_eq!(runtime.partial_writes.depth(), 1, "durable replay must have one executable owner");
assert!(
runtime.durable_replay_anchors.is_empty(),
"adoption must remove the duplicate startup anchor"
);
runtime.dispatch(&manager).await;
assert_eq!(
manager.operations_snapshot().await.queued_by_source.mrf,
0,
"unpersisted replay must not enter the manager"
);
assert!(runtime.flush().await, "publish the durable replay successor before dispatch");
let successor = snapshot::inspect_local_committed_snapshot(runtime.config.journal_max_bytes)
.await
.expect("inspect durable replay successor")
.expect("durable replay must publish a committed successor");
assert_eq!((successor.owner(), successor.sequence()), (runtime.checkpoint_owner, 12));
assert_eq!(runtime.runtime_checkpoint, Some((runtime.checkpoint_owner, 12)));
assert!(
env.disk_paths.iter().all(|path| {
[".heal-mrf-commit.0.bin", ".heal-mrf-commit.1.bin"]
.iter()
.all(|manifest| path.join(".rustfs.sys").join(manifest).exists())
}),
"both startup and successor checkpoints must remain before proof"
);
runtime.dispatch(&manager).await;
assert_eq!(
manager.operations_snapshot().await.queued_by_source.mrf,
1,
"the checkpointed replay must be visible as one MRF manager request"
);
let anchor = runtime
.partial_writes
.anchors()
.next()
.expect("dispatched replay has a proof anchor")
.clone();
assert!(runtime.retained_replay_journal(), "proof-bearing replay anchors must block idle cleanup");
assert!(
snapshot::inspect_local_committed_snapshot(runtime.config.journal_max_bytes)
@@ -2508,7 +2640,7 @@ mod tests {
);
assert!(
runtime.delete_idle_recovery_anchors().await,
"idle cleanup must delete the proof-discharged committed replay checkpoint"
"idle cleanup must delete both proof-discharged checkpoint generations"
);
runtime.journal_on_disk = false;
assert!(
@@ -2516,8 +2648,10 @@ mod tests {
.await
.expect("inspect committed checkpoints after proof cleanup")
.is_none(),
"the committed replay checkpoint must be gone after proof-driven cleanup"
"both committed checkpoint generations must be gone after proof-driven cleanup"
);
assert_eq!(read_journal(MRF_SCOPED_JOURNAL_PATH).await, None);
assert_eq!(read_journal(MRF_JOURNAL_PATH).await, None);
let restart_manager = Arc::new(HealManager::new(
storage,
@@ -21,6 +21,11 @@ struct Responsibility {
intent: MrfIntent,
anchor: Option<MrfDurableRepairAnchor>,
persisted: bool,
/// A completed check identified a healthy legacy object that cannot
/// discharge its durable intent without an independent payload proof.
/// This is process-local only: the unchanged journal rechecks it on restart.
unverified_legacy: bool,
retry_queued: bool,
next_attempt: Instant,
}
@@ -49,6 +54,8 @@ impl PartialWrites {
}
let key = queue_key(&intent);
let previous = self.entries.get(&key);
let previous_was_held = previous.is_some_and(|entry| entry.unverified_legacy);
let mut retry_queued = previous.is_some_and(|entry| entry.retry_queued);
let old_cost = previous.map_or(0, |entry| Self::cost(&entry.intent));
let next_bytes = self.bytes.saturating_sub(old_cost).saturating_add(Self::cost(&intent));
if (previous.is_none() && self.entries.len() >= capacity) || next_bytes > byte_budget {
@@ -57,8 +64,9 @@ impl PartialWrites {
if previous.is_some_and(|entry| entry.intent.lease == intent.lease) {
return Ok(());
}
if previous.is_none() {
if previous.is_none() || (previous_was_held && !retry_queued) {
self.retry_order.push_back(key.clone());
retry_queued = true;
}
// Replacing the generation preserves the logical repair obligation,
// but requires a new checkpoint and proof before it can be released.
@@ -68,6 +76,8 @@ impl PartialWrites {
intent,
anchor: None,
persisted: false,
unverified_legacy: false,
retry_queued,
next_attempt: Instant::now(),
},
);
@@ -91,6 +101,37 @@ impl PartialWrites {
self.entries.values().filter_map(|entry| entry.anchor.as_ref())
}
pub(super) fn park_unverified_legacy(&mut self, anchor: &MrfDurableRepairAnchor) -> bool {
let key = MrfQueueKey {
kind: anchor.kind,
bucket: anchor.bucket.clone(),
object: anchor.object.clone(),
version_id: anchor.version_id,
scope: anchor.scope,
delete_marker_purge: anchor.delete_marker_purge,
};
let Some(entry) = self.entries.get_mut(&key) else {
return false;
};
if entry.anchor.as_ref() != Some(anchor) || entry.unverified_legacy {
return false;
}
entry.unverified_legacy = true;
true
}
pub(super) fn unverified_legacy_count(&self) -> usize {
self.entries.values().filter(|entry| entry.unverified_legacy).count()
}
pub(super) fn oldest_unverified_legacy_enqueued_at_ms(&self) -> Option<u64> {
self.entries
.values()
.filter(|entry| entry.unverified_legacy)
.map(|entry| entry.intent.enqueued_at_ms)
.min()
}
pub(super) fn mark_persisted(&mut self) {
for entry in self.entries.values_mut() {
entry.persisted = true;
@@ -107,10 +148,12 @@ impl PartialWrites {
if entry.anchor.is_none() {
entry.anchor = manager.durable_mrf_repair_anchor(&entry.intent).await;
}
if entry.anchor.is_some() {
if let Some(anchor) = entry.anchor.clone() {
// Every outcome retains responsibility until a verified proof.
// A full manager or an offline target only postpones another try.
let _ = submit_mrf_heal_request(manager, &entry.intent).await;
// A healthy legacy object without an independent identity proof
// is parked in memory after one check; its disk journal remains
// unchanged and startup replay checks it again.
let _ = submit_mrf_heal_request(manager, &entry.intent, Some(anchor)).await;
}
}
}
@@ -126,10 +169,17 @@ impl PartialWrites {
let Some(key) = self.retry_order.pop_front() else {
break;
};
let due = self
.entries
.get(&key)
.is_some_and(|entry| entry.persisted && entry.next_attempt <= now);
let Some(entry) = self.entries.get_mut(&key) else {
continue;
};
entry.retry_queued = false;
if entry.unverified_legacy {
// Held obligations remain in the durable responsibility map,
// but leave the hot retry index until a new generation arrives.
continue;
}
let due = entry.persisted && entry.next_attempt <= now;
entry.retry_queued = true;
self.retry_order.push_back(key.clone());
if due {
ready.push(key);
@@ -161,6 +211,7 @@ mod tests {
use super::*;
use rustfs_common::mrf_channel::{MrfKind, MrfScope};
use std::sync::Arc;
use std::time::Duration;
use uuid::Uuid;
fn intent(object: &str) -> MrfIntent {
@@ -176,7 +227,7 @@ mod tests {
}),
lease: None,
enqueued_at_ms: 1,
attempts: u8::MAX,
attempts: 0,
};
assert_eq!(try_rearm_mrf_replay_intent(&mut intent), MrfIngressResult::Enqueued);
intent
@@ -217,6 +268,41 @@ mod tests {
);
}
#[test]
fn unverified_legacy_responsibility_is_held_until_restart_without_being_released() {
let original = intent("legacy");
let mut writes = PartialWrites::default();
writes
.admit(original.clone(), 1, 8192)
.expect("durable intent should be retained");
writes.mark_persisted();
let anchor = MrfDurableRepairAnchor::from_intent(&original, Uuid::new_v4()).expect("exact durable anchor");
writes.entries.get_mut(&queue_key(&original)).expect("resident intent").anchor = Some(anchor.clone());
assert!(writes.park_unverified_legacy(&anchor));
assert!(!writes.park_unverified_legacy(&anchor), "duplicate notices are idempotent");
assert_eq!(writes.unverified_legacy_count(), 1);
assert_eq!(writes.depth(), 1, "holding the intent must preserve responsibility");
assert!(writes.ready_keys(Instant::now() + Duration::from_secs(60), 1).is_empty());
assert!(writes.retry_order.is_empty(), "held intents leave the retry index after one pass");
let replacement = intent("legacy");
writes
.admit(replacement, 1, 8192)
.expect("a new generation should become retryable");
writes.mark_persisted();
assert_eq!(writes.unverified_legacy_count(), 0);
assert_eq!(writes.ready_keys(Instant::now() + Duration::from_secs(60), 1).len(), 1);
let mut restarted = PartialWrites::default();
restarted
.admit(original, 1, 8192)
.expect("the unchanged journal re-arms the same responsibility after restart");
restarted.mark_persisted();
assert_eq!(restarted.unverified_legacy_count(), 0);
assert_eq!(restarted.ready_keys(Instant::now() + Duration::from_secs(60), 1).len(), 1);
}
#[test]
fn partial_write_retention_adopts_replay_anchor_before_generation_replacement() {
use super::super::{MrfQueue, MrfRuntime};
+32 -7
View File
@@ -93,6 +93,12 @@ impl HealTask {
"Heal object stage entered"
);
self.check_control_flags().await?;
if self.source == HealRequestSource::Mrf {
// Durable partial writes always use the incarnation-fenced storage
// path, including for present objects. A normal heal can inspect a
// recreated bucket after the replay anchor was captured.
return self.heal_mrf_partial_write_object(bucket, object, version_id).await;
}
let mut object_exists = match self.await_with_control(self.storage.object_exists(bucket, object)).await {
Ok(exists) => exists,
Err(err @ Error::TransientSkip { .. }) => {
@@ -429,10 +435,6 @@ impl HealTask {
/// Recreate missing object (for EC decode scenarios)
async fn recreate_missing_object(&self, bucket: &str, object: &str, version_id: Option<&str>) -> Result<()> {
if self.source == HealRequestSource::Mrf {
return self.recreate_missing_mrf_object(bucket, object, version_id).await;
}
debug!(
target: "rustfs::heal::task",
event = EVENT_HEAL_OBJECT_STAGE,
@@ -534,8 +536,8 @@ impl HealTask {
}
}
/// Durable MRF responsibilities may complete only with an exact storage proof.
async fn recreate_missing_mrf_object(&self, bucket: &str, object: &str, version_id: Option<&str>) -> Result<()> {
/// Durable MRF partial writes may complete only with an incarnation-fenced storage proof.
async fn heal_mrf_partial_write_object(&self, bucket: &str, object: &str, version_id: Option<&str>) -> Result<()> {
let heal_opts = HealOpts {
recursive: false,
dry_run: self.options.dry_run,
@@ -577,7 +579,30 @@ impl HealTask {
storage_result.receipt.as_ref().map(|receipt| &receipt.disposition),
Some(HealObjectDisposition::AuthoritativelyAbsent)
);
if !self.record_verified_storage_receipt(expected, storage_result.receipt).await {
let receipt_missing = storage_result.receipt.is_none();
if !self
.record_verified_storage_receipt(expected.clone(), storage_result.receipt)
.await
{
let ok_drive_state = DriveState::Ok.to_string();
let healthy_legacy_without_repair = receipt_missing
&& storage_result.item.detail == rustfs_heal_contracts::heal_channel::LEGACY_OBJECT_IDENTITY_UNVERIFIED_DETAIL
&& storage_result.item.drives_healed() == Some(0)
&& !storage_result.item.after.drives.is_empty()
&& storage_result
.item
.after
.drives
.iter()
.all(|drive| drive.state == ok_drive_state);
if healthy_legacy_without_repair {
self.outcome.write().await.record(HealObjectOutcome {
identity: expected,
disposition: HealObjectDisposition::Unknown,
detail: Some(storage_result.item.detail.clone()),
});
}
self.record_result_item(storage_result.item).await;
return Err(Error::TaskExecutionFailed {
message: format!("Missing exact storage proof for durable MRF repair {bucket}/{object}"),
});
+33 -28
View File
@@ -2678,7 +2678,7 @@ async fn read_repair_object_heal_sets_read_repair_option() {
assert!(!opts[0].no_lock);
}
#[tokio::test]
#[tokio::test(start_paused = true)]
async fn read_repair_object_heal_is_not_failed_by_flat_task_timeout() {
let storage = Arc::new(MockStorage {
block_heal_object: Mutex::new(true),
@@ -2687,31 +2687,23 @@ async fn read_repair_object_heal_is_not_failed_by_flat_task_timeout() {
let mut request = HealRequest::object("bucket".to_string(), "object".to_string(), None);
request.source = HealRequestSource::ReadRepair;
request.options.timeout = Some(Duration::from_millis(1));
let task = Arc::new(HealTask::from_request(request, storage.clone()));
let execution = tokio::spawn({
let task = task.clone();
async move { task.execute().await }
});
let task = HealTask::from_request(request, storage.clone());
// Isolate the storage boundary from preflight's separately enforced budget.
let execution = task.heal_object("bucket", "object", None);
tokio::pin!(execution);
assert!(futures::poll!(&mut execution).is_pending());
assert!(storage.object_heal_opts.lock().expect("heal options")[0].read_repair);
tokio::time::timeout(Duration::from_secs(1), async {
loop {
if !storage.object_heal_opts.lock().unwrap().is_empty() {
break;
}
tokio::task::yield_now().await;
}
})
.await
.expect("read-repair object heal should start");
tokio::time::sleep(Duration::from_millis(20)).await;
assert!(!execution.is_finished(), "read repair must not be failed by the flat task timeout");
execution.abort();
assert!(execution.await.is_err(), "aborted mock execution should not join successfully");
assert!(storage.object_heal_opts.lock().unwrap()[0].read_repair);
*task.task_start_instant.write().await = Some(Instant::now() - Duration::from_millis(2));
assert!(matches!(task.remaining_timeout().await, Err(Error::TaskTimeout)));
tokio::time::advance(Duration::from_millis(20)).await;
assert!(
futures::poll!(&mut execution).is_pending(),
"read repair must remain in storage after the flat task budget expires"
);
}
#[tokio::test]
#[tokio::test(start_paused = true)]
async fn non_read_repair_object_heal_still_uses_flat_timeout() {
let storage = Arc::new(MockStorage {
block_heal_object: Mutex::new(true),
@@ -2719,13 +2711,26 @@ async fn non_read_repair_object_heal_still_uses_flat_timeout() {
});
let mut request = HealRequest::object("bucket".to_string(), "object".to_string(), None);
request.options.timeout = Some(Duration::from_millis(1));
let task = HealTask::from_request(request, storage);
let task = HealTask::from_request(request, storage.clone());
let execution = task.heal_object("bucket", "object", None);
tokio::pin!(execution);
assert!(futures::poll!(&mut execution).is_pending());
assert!(!storage.object_heal_opts.lock().expect("heal options")[0].read_repair);
tokio::time::advance(Duration::from_millis(20)).await;
assert!(matches!(futures::poll!(&mut execution), std::task::Poll::Ready(Err(Error::TaskTimeout))));
}
let result = tokio::time::timeout(Duration::from_secs(1), task.execute())
.await
.expect("flat timeout should finish the task");
#[tokio::test]
async fn read_repair_preflight_still_rejects_an_expired_task_budget() {
let storage = Arc::new(MockStorage::default());
let mut request = HealRequest::object("bucket".to_string(), "object".to_string(), None);
request.source = HealRequestSource::ReadRepair;
request.options.timeout = Some(Duration::from_millis(1));
let task = HealTask::from_request(request, storage.clone());
*task.task_start_instant.write().await = Some(Instant::now() - Duration::from_millis(2));
assert!(matches!(result, Err(Error::TaskTimeout)));
assert!(matches!(task.heal_object("bucket", "object", None).await, Err(Error::TaskTimeout)));
assert!(storage.object_heal_opts.lock().expect("heal options").is_empty());
}
async fn make_resume_disk(temp: &TempDir) -> DiskStore {
@@ -588,6 +588,15 @@ async fn partial_write_sigkill_replay_scenario(protected: bool) {
.await,
"legacy replay attempts must finish"
);
// Cover two MRF admission backoff periods so a missed hold notice
// cannot pass merely because its task finished between polls.
let retry_window = tokio::time::Instant::now() + Duration::from_secs(12);
while tokio::time::Instant::now() < retry_window {
let snapshot = manager.operations_snapshot().await;
assert_eq!(snapshot.queue_length, 0, "an unverified legacy result must not refill the manager queue");
assert_eq!(snapshot.active_tasks, 0, "a held intent must not stay active");
tokio::time::sleep(Duration::from_millis(100)).await;
}
assert!(snapshot_contains("crash.bin").await, "unverified legacy responsibility must remain");
}
manager.stop().await.expect("restarted manager should stop");
+33 -12
View File
@@ -60,6 +60,16 @@ async fn heal_env() -> (Vec<std::path::PathBuf>, Arc<dyn HealStorageAPI>) {
heal_env_at(None).await
}
async fn heal_env_with_bucket(bucket: &str) -> (Vec<PathBuf>, Arc<dyn HealStorageAPI>) {
let env = rustfs_test_utils::TestECStoreEnv::builder()
.prefix("rustfs_heal_mrf_test")
.build()
.await;
env.make_bucket(bucket, false).await;
let storage: Arc<dyn HealStorageAPI> = Arc::new(ECStoreHealStorage::new(env.ecstore.clone()));
(env.disk_paths, storage)
}
async fn heal_env_at(base_dir: Option<&Path>) -> (Vec<std::path::PathBuf>, Arc<dyn HealStorageAPI>) {
let mut builder = rustfs_test_utils::TestECStoreEnv::builder().prefix("rustfs_heal_mrf_test");
if let Some(base_dir) = base_dir {
@@ -356,7 +366,7 @@ async fn decode_failure_intent_maps_to_urgent_mrf_heal_request() {
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
#[serial]
async fn journal_replay_arms_intents_and_retains_unproven_partial_write_anchor() {
let (disk_paths, storage) = heal_env().await;
let (disk_paths, storage) = heal_env_with_bucket("replay-bucket").await;
// The journal reader resolves disks through the process-local disk map;
// register the environment's disks the same way server startup does.
@@ -387,8 +397,14 @@ async fn journal_replay_arms_intents_and_retains_unproven_partial_write_anchor()
assert!(
disk_paths
.iter()
.all(|path| !Path::new(path).join(META_BUCKET).join(SCOPED_JOURNAL_REL).exists()),
"missing authoritative journal remains absent"
.all(|path| Path::new(path).join(META_BUCKET).join(SCOPED_JOURNAL_REL).exists()),
"partial-write dispatch must first publish its authoritative successor"
);
let successor = journal_record(3, "replay-bucket", "partial-object", None, 1);
assert!(
journal_matches_on_all_disks(&disk_paths, SCOPED_JOURNAL_REL, &successor)
&& committed_checkpoint_matches_on_all_disks(&disk_paths, 1, &successor),
"the committed and authoritative successor must preserve the partial-write identity"
);
let snapshot = manager.operations_snapshot().await;
@@ -402,7 +418,7 @@ async fn journal_replay_arms_intents_and_retains_unproven_partial_write_anchor()
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
#[serial]
async fn committed_snapshot_replay_takes_precedence_over_stale_legacy_mirror() {
let (disk_paths, storage) = heal_env().await;
let (disk_paths, storage) = heal_env_with_bucket("committed-bucket").await;
register_local_disks(&disk_paths, "mrf-committed-replay-test").await;
let committed = scoped_journal_record(3, "committed-bucket", "committed-object", Some([9u8; 16]), 0, 0, 0);
@@ -534,11 +550,13 @@ async fn authoritative_journal_is_not_merged_with_legacy_mirror() {
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
#[serial]
async fn authoritative_journal_replay_preserves_kind_and_scope_identity() {
let (disk_paths, storage) = heal_env().await;
let (disk_paths, storage) = heal_env_with_bucket("identity-bucket").await;
register_local_disks(&disk_paths, "mrf-authoritative-identity-test").await;
let mut authoritative = scoped_journal_record(3, "identity-bucket", "same-object", None, 0, 3, 7);
authoritative.extend(scoped_journal_record(3, "identity-bucket", "same-object", None, 0, 3, 8));
let first_partial = scoped_journal_record(3, "identity-bucket", "same-object", None, 0, 3, 7);
let second_partial = scoped_journal_record(3, "identity-bucket", "same-object", None, 0, 3, 8);
let mut authoritative = first_partial.clone();
authoritative.extend_from_slice(&second_partial);
authoritative.extend(journal_record(2, "identity-bucket", "same-object", None, 0));
authoritative.extend(journal_record(1, "identity-bucket", "same-object", Some([4u8; 16]), 0));
let stale_legacy = journal_record(3, "identity-bucket", "stale-legacy-object", None, 0);
@@ -570,11 +588,14 @@ async fn authoritative_journal_replay_preserves_kind_and_scope_identity() {
"decode-failure repair must not merge with object repair responsibility"
);
assert!(
disk_paths.iter().all(|path| {
Path::new(path).join(META_BUCKET).join(JOURNAL_REL).exists()
&& Path::new(path).join(META_BUCKET).join(SCOPED_JOURNAL_REL).exists()
}),
"partial-write responsibilities keep both replay anchors until proof"
journal_contains_on_all_disks(&disk_paths, SCOPED_JOURNAL_REL, &first_partial)
&& journal_contains_on_all_disks(&disk_paths, SCOPED_JOURNAL_REL, &second_partial)
&& committed_payload_contains_on_all_disks(&disk_paths, &[&first_partial, &second_partial]),
"both scoped partial-write identities must survive in the authoritative and committed successor"
);
assert!(
journal_matches_on_all_disks(&disk_paths, JOURNAL_REL, &[]),
"the legacy mirror must not misrepresent scoped-only partial-write responsibilities"
);
}
+67 -1
View File
@@ -58,8 +58,14 @@ impl ActionSet {
}
pub fn is_match(&self, action: &Action) -> bool {
self.is_match_for_effect(action, false)
}
/// `deny` lets a blanket `s3:*` cover force-delete actions so an explicit
/// Deny still wins. Allow matching stays name-only for those actions.
pub fn is_match_for_effect(&self, action: &Action, deny: bool) -> bool {
for act in self.0.iter() {
if act.is_match(action) {
if act.is_match_for_effect(action, deny) {
return true;
}
@@ -72,6 +78,22 @@ impl ActionSet {
false
}
/// Whether a statement with this `Action` list and `not_actions` as its
/// `NotAction` list covers `action`.
///
/// `NotAction` always uses plain wildcard matching, so `NotAction: "s3:*"`
/// still excludes force-delete. An Allow grants force-delete only when its
/// `Action` list names it: an Allow built from `NotAction` alone never does.
pub fn statement_covers(&self, not_actions: &ActionSet, action: &Action, deny: bool) -> bool {
if not_actions.is_match_for_effect(action, true) {
return false;
}
if self.is_empty() {
return deny || !action_requires_explicit_grant(action);
}
self.is_match_for_effect(action, deny)
}
}
impl Deref for ActionSet {
@@ -155,10 +177,28 @@ pub enum Action {
impl Action {
pub fn is_match(&self, action: &Action) -> bool {
self.is_match_for_effect(action, false)
}
pub fn is_match_for_effect(&self, action: &Action, deny: bool) -> bool {
// Force-delete bypasses emptiness and version checks. A blanket `s3:*`
// / `*` Allow (including canned consoleAdmin) must not confer it; the
// statement has to name `s3:ForceDeleteBucket` or `s3:ForceDeleteObject`.
// The same wildcard on a Deny still matches, so explicit deny wins.
if !deny && matches!(self, Action::S3Action(S3Action::AllActions)) && action_requires_explicit_grant(action) {
return false;
}
wildcard::is_match::<&str, &str>(self.into(), action.into())
}
}
fn action_requires_explicit_grant(action: &Action) -> bool {
matches!(
action,
Action::S3Action(S3Action::ForceDeleteBucketAction | S3Action::ForceDeleteObjectAction)
)
}
impl From<&Action> for &str {
fn from(value: &Action) -> &'static str {
match value {
@@ -220,8 +260,14 @@ pub enum S3Action {
CreateBucketAction,
#[strum(serialize = "s3:DeleteBucket")]
DeleteBucketAction,
/// DeleteBucket when `x-minio-force-delete` / `x-rustfs-force-delete` is set.
/// Not implied by `s3:*`; the action must be named on the statement.
#[strum(serialize = "s3:ForceDeleteBucket")]
ForceDeleteBucketAction,
/// Recursive DeleteObject selected by the same force-delete header.
/// Not implied by `s3:*`; the action must be named on the statement.
#[strum(serialize = "s3:ForceDeleteObject")]
ForceDeleteObjectAction,
#[strum(serialize = "s3:DeleteBucketPolicy")]
DeleteBucketPolicyAction,
#[strum(serialize = "s3:DeleteBucketPublicAccessBlock")]
@@ -732,6 +778,26 @@ pub enum KmsAction {
mod tests {
use super::*;
#[test]
fn blanket_s3_wildcard_does_not_grant_force_delete() {
let wildcard = Action::try_from("s3:*").expect("s3:* parses");
let star = Action::try_from("*").expect("* parses as s3:*");
let force_bucket = Action::try_from("s3:ForceDeleteBucket").expect("force bucket action parses");
let force_object = Action::try_from("s3:ForceDeleteObject").expect("force object action parses");
let delete_object = Action::try_from("s3:DeleteObject").expect("delete object parses");
assert!(!wildcard.is_match(&force_bucket));
assert!(!wildcard.is_match(&force_object));
assert!(!star.is_match(&force_bucket));
assert!(!star.is_match(&force_object));
assert!(wildcard.is_match(&delete_object));
assert!(wildcard.is_match_for_effect(&force_bucket, true));
assert!(wildcard.is_match_for_effect(&force_object, true));
assert!(force_bucket.is_match(&force_bucket));
assert!(force_object.is_match(&force_object));
assert!(!force_bucket.is_match(&force_object));
}
#[test]
fn test_action_wildcard_parsing() {
// Test that "*" parses to S3Action::AllActions
+119
View File
@@ -3368,4 +3368,123 @@ mod test {
assert_eq!(round_trip.statements.len(), policy.statements.len());
assert_eq!(round_trip.statements[0].effect, policy.statements[0].effect);
}
#[tokio::test]
async fn force_delete_actions_require_an_explicit_grant() {
let wildcard = Policy::parse_config(
br#"{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":["s3:*"],"Resource":["arn:aws:s3:::*"]}]}"#,
)
.expect("wildcard policy parses");
let explicit = Policy::parse_config(
br#"{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":["s3:ForceDeleteObject","s3:ForceDeleteBucket","s3:DeleteObject"],"Resource":["arn:aws:s3:::*"]}]}"#,
)
.expect("explicit policy parses");
let denied = Policy::parse_config(
br#"{"Version":"2012-10-17","Statement":[
{"Effect":"Allow","Action":["s3:ForceDeleteObject"],"Resource":["arn:aws:s3:::*"]},
{"Effect":"Deny","Action":["s3:ForceDeleteObject"],"Resource":["arn:aws:s3:::*"]}
]}"#,
)
.expect("deny policy parses");
let wildcard_deny = Policy::parse_config(
br#"{"Version":"2012-10-17","Statement":[
{"Effect":"Allow","Action":["s3:ForceDeleteObject","s3:DeleteObject"],"Resource":["arn:aws:s3:::*"]},
{"Effect":"Deny","Action":["s3:*"],"Resource":["arn:aws:s3:::*"]}
]}"#,
)
.expect("wildcard deny policy parses");
assert!(allows(&wildcard, "s3:DeleteObject").await);
assert!(!allows(&wildcard, "s3:ForceDeleteObject").await);
assert!(!allows(&wildcard, "s3:ForceDeleteBucket").await);
assert!(allows(&explicit, "s3:ForceDeleteObject").await);
assert!(allows(&explicit, "s3:ForceDeleteBucket").await);
assert!(!allows(&denied, "s3:ForceDeleteObject").await);
assert!(!allows(&wildcard_deny, "s3:ForceDeleteObject").await);
assert!(!allows(&wildcard_deny, "s3:DeleteObject").await);
}
#[tokio::test]
async fn not_action_allow_does_not_grant_force_delete() {
for not_action in ["s3:*", "s3:GetObject"] {
let iam = Policy::parse_config(
format!(
r#"{{"Version":"2012-10-17","Statement":[{{"Effect":"Allow","NotAction":["{not_action}"],"Resource":["arn:aws:s3:::*"]}}]}}"#
)
.as_bytes(),
)
.expect("NotAction policy parses");
let bucket: BucketPolicy = serde_json::from_str(&format!(
r#"{{"Version":"2012-10-17","Statement":[{{"Effect":"Allow","Principal":{{"AWS":["*"]}},"NotAction":["{not_action}"],"Resource":["arn:aws:s3:::bucket","arn:aws:s3:::bucket/*"]}}]}}"#
))
.expect("NotAction bucket policy parses");
for action in ["s3:ForceDeleteObject", "s3:ForceDeleteBucket"] {
assert!(!allows(&iam, action).await, "IAM NotAction {not_action} must not grant {action}");
assert!(
!bucket_allows(&bucket, action).await,
"bucket NotAction {not_action} must not grant {action}"
);
}
}
// NotAction still grants ordinary actions it does not exclude.
let iam = Policy::parse_config(
br#"{"Version":"2012-10-17","Statement":[{"Effect":"Allow","NotAction":["s3:GetObject"],"Resource":["arn:aws:s3:::*"]}]}"#,
)
.expect("NotAction policy parses");
assert!(allows(&iam, "s3:DeleteObject").await);
// Deny + NotAction keeps plain wildcard semantics: s3:* excludes
// force-delete, a narrower exclusion still denies it.
let deny_all_but = |not_action: &str| {
Policy::parse_config(
format!(
r#"{{"Version":"2012-10-17","Statement":[
{{"Effect":"Allow","Action":["s3:ForceDeleteObject"],"Resource":["arn:aws:s3:::*"]}},
{{"Effect":"Deny","NotAction":["{not_action}"],"Resource":["arn:aws:s3:::*"]}}
]}}"#
)
.as_bytes(),
)
.expect("deny NotAction policy parses")
};
assert!(allows(&deny_all_but("s3:*"), "s3:ForceDeleteObject").await);
assert!(!allows(&deny_all_but("s3:GetObject"), "s3:ForceDeleteObject").await);
}
async fn bucket_allows(policy: &BucketPolicy, action: &str) -> bool {
let conditions = HashMap::new();
policy
.is_allowed(&BucketPolicyArgs {
account: "user",
groups: &None,
action: Action::try_from(action).expect("action parses"),
bucket: "bucket",
conditions: &conditions,
is_owner: false,
object: "folder/a",
})
.await
}
async fn allows(policy: &Policy, action: &str) -> bool {
let conditions = HashMap::new();
let claims = HashMap::new();
let groups = None;
policy
.is_allowed(&Args {
account: "user",
groups: &groups,
action: Action::try_from(action).expect("action parses"),
bucket: "bucket",
conditions: &conditions,
is_owner: false,
object: "folder/a",
claims: &claims,
deny_only: false,
})
.await
}
}
+4 -2
View File
@@ -247,7 +247,8 @@ impl Statement {
/// Returns true when this statement would reach `conditions.evaluate_with_resolver` in
/// [`Statement::is_allowed`] (including the KMS resource path). Does not evaluate conditions.
pub(crate) async fn request_reaches_condition_eval(&self, args: &Args<'_>, resolver: &VariableResolver) -> bool {
if (!self.actions.is_match(&args.action) && !self.actions.is_empty()) || self.not_actions.is_match(&args.action) {
let deny = matches!(self.effect, Effect::Deny);
if !self.actions.statement_covers(&self.not_actions, &args.action, deny) {
return false;
}
@@ -423,7 +424,8 @@ impl BPStatement {
return false;
}
if (!self.actions.is_match(&args.action) && !self.actions.is_empty()) || self.not_actions.is_match(&args.action) {
let deny = matches!(self.effect, Effect::Deny);
if !self.actions.statement_covers(&self.not_actions, &args.action, deny) {
return false;
}
@@ -1,20 +1,34 @@
# Throwaway image for generating real MinIO on-disk fixtures without installing
# a MinIO server on the host. Multi-stage: pull the official MinIO server binary
# from the published image (linux, no dl.min.io download), then drop it into a
# a MinIO server on the host. Multi-stage: extract the official MinIO server
# binary from its checksum-pinned GitHub release package, then drop it into a
# small Python image that runs the fixture lab (lab.py needs only python3 +
# openssl; it drives MinIO's S3 API directly, so no `mc` is required).
#
# The MinIO release is pinned so the captured fixture format is reproducible;
# this is the release the interop tests were validated against.
#
# Both base images are build args so a network that cannot reach Docker Hub can
# point them at a mirror carrying the same content — quay.io publishes the MinIO
# releases, and public.ecr.aws mirrors the official Python images. CI keeps the
# Docker Hub defaults. Override with:
# --build-arg MINIO_IMAGE=quay.io/minio/minio:RELEASE.2025-09-07T16-13-09Z \
# --build-arg PYTHON_IMAGE=public.ecr.aws/docker/library/python:3.12-slim
ARG MINIO_IMAGE=minio/minio:RELEASE.2025-09-07T16-13-09Z
# MINIO_IMAGE can override the package stage with an accessible image carrying
# the same release at /usr/bin/minio. PYTHON_IMAGE can point at a Python mirror.
ARG MINIO_IMAGE=minio-release
ARG PYTHON_IMAGE=python:3.12-slim
FROM scratch AS minio-package-amd64
ADD --checksum=sha256:eeda08f699f6592d1b868ac8bda864ae2cacdb5ee1b888663366e8c8ff566249 https://github.com/minio/minio/releases/download/RELEASE.2025-09-07T16-13-09Z/minio_20250907161309.0.0_amd64.deb /minio.deb
FROM scratch AS minio-package-arm64
ADD --checksum=sha256:f9466ed832ac4926c8870b3b8378d77861a87b41529fecae7789b9bc54acd78b https://github.com/minio/minio/releases/download/RELEASE.2025-09-07T16-13-09Z/minio_20250907161309.0.0_arm64.deb /minio.deb
FROM scratch AS minio-package-ppc64le
ADD --checksum=sha256:39eec77281c552dfa0a92b5221c657dc58306ea8c4e92b79e9562cee61b59caa https://github.com/minio/minio/releases/download/RELEASE.2025-09-07T16-13-09Z/minio_20250907161309.0.0_ppc64el.deb /minio.deb
FROM minio-package-${TARGETARCH} AS minio-package
FROM ${PYTHON_IMAGE} AS minio-release
COPY --from=minio-package /minio.deb /tmp/minio.deb
RUN dpkg-deb --extract /tmp/minio.deb /tmp/minio-package \
&& install -D /tmp/minio-package/usr/local/bin/minio /usr/bin/minio \
&& rm -rf /tmp/minio.deb /tmp/minio-package
FROM ${MINIO_IMAGE} AS minio
FROM ${PYTHON_IMAGE}
@@ -24,15 +24,14 @@ Use the automated path when you want the lab to:
## Networks without Docker Hub access
`capture_via_docker.sh` pulls its two base images from Docker Hub by default. Where that registry is unreachable, point the build at mirrors carrying the same content — quay.io publishes the MinIO releases and public.ecr.aws mirrors the official Python images:
`capture_via_docker.sh` extracts MinIO from the official GitHub release package after verifying its pinned SHA256. The Dockerfile selects the matching amd64, arm64, or ppc64le package; the MinIO release remains `RELEASE.2025-09-07T16-13-09Z`. Python comes from Docker Hub by default. Where that registry is unreachable, public.ecr.aws mirrors the official Python images:
```bash
MINIO_LAB_MINIO_IMAGE=quay.io/minio/minio:RELEASE.2025-09-07T16-13-09Z \
MINIO_LAB_PYTHON_IMAGE=public.ecr.aws/docker/library/python:3.12-slim \
./capture_via_docker.sh
```
Pin the MinIO tag to the same release the Dockerfile names; an unpinned `:latest` captures whatever format that day's build writes, which is not what the interop tests were validated against.
`MINIO_LAB_MINIO_IMAGE` can override the package stage with an accessible image containing the same MinIO release at `/usr/bin/minio`. Pin that image to the same release; an unpinned `:latest` captures whatever format that day's build writes, which is not what the interop tests were validated against.
## Layout
@@ -34,9 +34,9 @@ if [ "${cases[0]}" != "all" ]; then
done
fi
# Base images are overridable so a network without Docker Hub access can point
# them at a mirror (see the Dockerfile header). Unset by default, which keeps the
# Dockerfile's Docker Hub defaults for CI.
# The Dockerfile defaults to a checksum-pinned MinIO release package and a
# Python base image. Overrides can select accessible images carrying the same
# content (see the Dockerfile header).
build_args=()
if [ -n "${MINIO_LAB_MINIO_IMAGE:-}" ]; then
build_args+=(--build-arg "MINIO_IMAGE=${MINIO_LAB_MINIO_IMAGE}")
@@ -49,7 +49,8 @@ echo ">> building ${IMAGE}"
docker build -f "${SCRIPT_DIR}/Dockerfile" -t "${IMAGE}" "${build_args[@]}" "${SCRIPT_DIR}"
echo ">> capturing fixtures into ${FIXTURE_REL}"
docker run --rm -v "${REPO_ROOT}:/repo" "${IMAGE}" \
# The host reader initializes disk metadata inside the captured fixture tree.
docker run --rm --user "$(id -u):$(id -g)" -v "${REPO_ROOT}:/repo" "${IMAGE}" \
python3 "/repo/${LAB_REL}/lab.py" capture-matrix \
--root "/repo/${FIXTURE_REL}" \
--work-root /tmp/minio-lab-work \
+20 -7
View File
@@ -861,16 +861,19 @@ fn classify_http_response(
operation: Option<&'static str>,
) -> ClassifiedHttpResponse {
let kind = classify_http_status(status);
if status != reqwest::StatusCode::INTERNAL_SERVER_ERROR || operation != Some(INTERNODE_OPERATION_READ_FILE_STREAM) {
if status != reqwest::StatusCode::INTERNAL_SERVER_ERROR {
return ClassifiedHttpResponse {
kind,
remote_disk_error: None,
};
}
let remote_disk_error = match headers.get(INTERNODE_DISK_ERROR_HEADER).and_then(|value| value.to_str().ok()) {
Some(INTERNODE_FILE_NOT_FOUND) => Some(RemoteDiskErrorKind::FileNotFound),
Some(INTERNODE_VOLUME_NOT_FOUND) => Some(RemoteDiskErrorKind::VolumeNotFound),
Some(INTERNODE_FILE_CORRUPT) => Some(RemoteDiskErrorKind::FileCorrupt),
let token = headers.get(INTERNODE_DISK_ERROR_HEADER).and_then(|value| value.to_str().ok());
let remote_disk_error = match (operation, token) {
(Some(INTERNODE_OPERATION_READ_FILE_STREAM), Some(INTERNODE_FILE_NOT_FOUND))
| (Some(INTERNODE_OPERATION_WALK_DIR), Some(INTERNODE_FILE_NOT_FOUND)) => Some(RemoteDiskErrorKind::FileNotFound),
(Some(INTERNODE_OPERATION_READ_FILE_STREAM), Some(INTERNODE_VOLUME_NOT_FOUND))
| (Some(INTERNODE_OPERATION_WALK_DIR), Some(INTERNODE_VOLUME_NOT_FOUND)) => Some(RemoteDiskErrorKind::VolumeNotFound),
(Some(INTERNODE_OPERATION_READ_FILE_STREAM), Some(INTERNODE_FILE_CORRUPT)) => Some(RemoteDiskErrorKind::FileCorrupt),
_ => None,
};
ClassifiedHttpResponse { kind, remote_disk_error }
@@ -2126,8 +2129,18 @@ mod tests {
let wrong_status =
classify_http_response(reqwest::StatusCode::NOT_FOUND, &headers, Some(INTERNODE_OPERATION_READ_FILE_STREAM));
assert!(wrong_status.remote_disk_error.is_none());
let wrong_operation =
let walk_dir_missing =
classify_http_response(reqwest::StatusCode::INTERNAL_SERVER_ERROR, &headers, Some(INTERNODE_OPERATION_WALK_DIR));
assert_eq!(walk_dir_missing.remote_disk_error, Some(RemoteDiskErrorKind::FileNotFound));
headers.insert(INTERNODE_DISK_ERROR_HEADER, INTERNODE_VOLUME_NOT_FOUND.parse().unwrap());
let walk_dir_volume_missing =
classify_http_response(reqwest::StatusCode::INTERNAL_SERVER_ERROR, &headers, Some(INTERNODE_OPERATION_WALK_DIR));
assert_eq!(walk_dir_volume_missing.remote_disk_error, Some(RemoteDiskErrorKind::VolumeNotFound));
let wrong_operation = classify_http_response(
reqwest::StatusCode::INTERNAL_SERVER_ERROR,
&headers,
Some(INTERNODE_OPERATION_PUT_FILE_STREAM),
);
assert!(wrong_operation.remote_disk_error.is_none());
}
@@ -2150,8 +2163,8 @@ mod tests {
}
for operation in [
None,
Some(INTERNODE_OPERATION_WALK_DIR),
Some(INTERNODE_OPERATION_PUT_FILE_STREAM),
Some(INTERNODE_OPERATION_WALK_DIR),
] {
assert!(
classify_http_response(reqwest::StatusCode::INTERNAL_SERVER_ERROR, &headers, operation)
+12 -1
View File
@@ -998,7 +998,18 @@ impl DataUsageCache {
// not bind to the process-local leader epoch.
receipt.digest = digest;
}
if let Some(progress) = &mut self.info.scan_progress {
// A sweep without a durable position restarts from the first entry,
// so it observes nothing under the plan that finished the previous
// sweep. Keeping that plan would mark every sweep of a bucket written
// between cycles as mixed, and the bucket could never certify.
let has_position = self.info.scan_resume_after.is_some()
|| self.info.scan_checkpoint.is_some()
|| self.info.scan_raw_enumeration_cursor.is_some()
|| self.info.scan_raw_enumeration_page_index.is_some()
|| self.info.scan_coverage_receipt.is_some();
if let Some(progress) = &mut self.info.scan_progress
&& has_position
{
progress.requested_plan = scan_plan_digest;
} else {
self.info.scan_progress = Some(DataUsageScanProgress {
+15 -2
View File
@@ -636,8 +636,8 @@ async fn cycle_budget_fence_accepts_bootstrap_pending_usage_marker() {
#[tokio::test]
async fn cycle_budget_deadline_handler_fences_and_releases_guard() {
let (_temp_dir, store) = setup_scanner_cycle_store().await;
let lock = store
let (_temp_dir, lock_store) = setup_scanner_cycle_store().await;
let lock = lock_store
.new_ns_lock(RUSTFS_META_BUCKET, "leader.lock")
.await
.expect("scanner leader lock should be created");
@@ -646,6 +646,17 @@ async fn cycle_budget_deadline_handler_fences_and_releases_guard() {
.await
.expect("scanner leader lock should be acquired");
// Keep the real guard, but isolate the fencing deadline from filesystem I/O.
let store = Arc::new(MemoryConfigStore::default());
save_config(
store.clone(),
DATA_USAGE_OBJ_NAME_PATH.as_str(),
serde_json::to_vec(&complete_usage_with_bucket_count(Some(std::time::SystemTime::UNIX_EPOCH), 0))
.expect("scanner cycle usage baseline should encode"),
)
.await
.expect("scanner cycle usage baseline should persist");
let ctx = CancellationToken::new();
let mut cycle_info = CurrentCycle {
current: 12,
@@ -679,6 +690,8 @@ async fn cycle_budget_deadline_handler_fences_and_releases_guard() {
.await;
assert!(guard.is_released());
assert_eq!(leader_epoch, 2, "deadline handler should claim the next epoch");
assert!(matches!(cycle_revision, DataUsageCacheRevision::Etag(_)));
let persisted = read_config(store, &DATA_USAGE_BLOOM_NAME_PATH)
.await
.expect("deadline handler should persist a fenced cursor");
@@ -349,6 +349,74 @@ fn checkpoint_fixture_unchanged_complete_plan_keeps_existing_rescan_policy() {
assert!(!cache.info.snapshot_complete);
}
#[test]
fn checkpoint_fixture_sweep_without_position_restarts_under_requested_plan() {
let (mut cache, identity) = bound_checkpoint();
// State saved when a mixed sweep reaches the end: observations stay, the
// position is cleared, and the finishing cycle's plan is recorded.
let finished = DataUsageScanPlanDigest([45; 32]);
cache.info.scan_progress = Some(crate::DataUsageScanProgress {
started_plan: finished,
requested_plan: finished,
});
cache.info.scan_resume_after = None;
cache.info.scan_checkpoint = None;
cache.info.scan_coverage_receipt = None;
let next = DataUsageScanPlanDigest([46; 32]);
assert_eq!(
cache.prepare_bucket_checkpoint("bucket", 12, 7, SOURCE, next, identity),
crate::DataUsageCachePrepareOutcome::Reused
);
assert_eq!(
cache.info.scan_progress,
Some(crate::DataUsageScanProgress {
started_plan: next,
requested_plan: next
}),
"a sweep that restarts from the first entry observes nothing under the finished plan"
);
assert_eq!(retained(&cache), 3, "observations survive until the new sweep replaces them");
assert!(cache.info.scan_plan_digest.is_none());
assert!(!cache.info.snapshot_complete);
}
#[test]
fn checkpoint_fixture_sweep_with_position_keeps_its_started_plan() {
let (cache, identity) = bound_checkpoint();
let next = DataUsageScanPlanDigest([47; 32]);
let mut frontier = cache.clone();
frontier.prepare_bucket_checkpoint("bucket", 12, 7, SOURCE, next, identity);
assert_eq!(frontier.validated_scan_frontier(), Some("bucket/static"));
assert_eq!(
frontier.info.scan_progress,
Some(crate::DataUsageScanProgress {
started_plan: PLAN,
requested_plan: next
}),
"a resumed prefix was observed under the started plan"
);
let mut raw_cursor = cache;
raw_cursor.info.scan_resume_after = None;
raw_cursor.info.scan_checkpoint = None;
raw_cursor.info.scan_coverage_receipt = None;
let cursor = DataUsageRawEnumerationCursor::new("bucket/static".into(), Some("0002".into()), 3, [9; 32]);
raw_cursor.info.scan_raw_enumeration_cursor = Some(cursor.clone());
raw_cursor.prepare_bucket_checkpoint("bucket", 12, 7, SOURCE, next, identity);
assert_eq!(raw_cursor.info.scan_raw_enumeration_cursor, Some(cursor));
assert_eq!(
raw_cursor
.info
.scan_progress
.expect("raw cursor resumes the sweep")
.started_plan,
PLAN,
"a raw enumeration cursor is a position too"
);
assert!(raw_cursor.info.scan_plan_digest.is_none());
}
#[test]
fn checkpoint_fixture_identity_changes_and_future_state_fail_closed() {
let (cache, identity) = bound_checkpoint();
@@ -806,6 +874,98 @@ async fn checkpoint_fixture_hot_digest_retains_partial_progress() {
run_checkpoint_fixture(true).await;
}
/// A bucket written between every cycle requests a new plan each round. Once
/// a budget-cut sweep has ended as mixed, the next sweep starts from the first
/// entry and must certify under its own plan instead of staying mixed forever.
#[tokio::test]
#[serial]
async fn checkpoint_fixture_hot_bucket_certifies_first_full_sweep_after_mixed_end() {
let (scanner, root) = build_test_scanner().await;
let _guard = TestGuard {
temp_dir: Some(root.clone()),
};
for index in 0..STATIC_OBJECTS {
write_checkpoint_object(&root, &format!("static/{index:04}"), &[(None, 1)]).await;
}
let identity = crate::DataUsageScanIdentity {
version: 1,
bucket_incarnation: Uuid::from_u128(1),
set_layout: DataUsageScanPlanDigest([41; 32]),
publication_epoch: 0,
tier_registry_generation: crate::runtime_tier_registry_for_cycle(11, 7).await.generation,
scan_mode: HealScanMode::Normal,
};
let store = FixtureStore::new();
// (object budget, expected complete): a budget-cut sweep, its unbounded
// tail that ends mixed, then fresh unbounded sweeps under new plans.
let rounds = [(Some(4), false), (None, false), (None, true), (None, true)];
for (round, (max_objects, expect_complete)) in (0_u64..).zip(rounds) {
write_checkpoint_object(&root, "hot/current", &[(None, 1)]).await;
let plan = crate::scanner_io::checkpoint_fixture_bucket_digest(PLAN, Some(round));
let mut cache = DataUsageCache::default();
let revisions = cache
.load_with_revisions(store.clone(), CACHE_NAME)
.await
.expect("load checkpoint revisions");
crate::scanner_io::current_cache_root_or_prepare_with_generation(
&mut cache,
"bucket",
SOURCE,
11,
7,
plan,
crate::scanner_io::DataUsageCacheReuseOptions {
require_source: true,
tier_registry_generation: Some(identity.tier_registry_generation),
checkpoint_identity: Some(identity),
},
);
cache.info.skip_healing = true;
let parent = CancellationToken::new();
let budget = ScannerCycleBudget::new_with_progress_tracking(
&parent,
ScannerCycleBudgetConfig {
max_objects,
..Default::default()
},
);
let outcome = scanner
.local_disk
.clone()
.nsscanner_disk(
budget.token(),
budget.clone(),
vec![scanner.local_disk.clone()],
cache,
None,
scan_options(HealScanMode::Normal),
)
.await
.expect("hot bucket sweep outcome");
let (cache, complete) = match outcome {
ScannerDiskScanOutcome::Complete(cache) => (cache, true),
ScannerDiskScanOutcome::Partial(cache) => (cache, false),
ScannerDiskScanOutcome::NamespaceNotFound(_) => panic!("fixture namespace exists"),
};
assert_eq!(budget.budget_elapsed(), max_objects.is_some(), "round {round} budget");
cache
.save_with_revisions_for_epoch(store.clone(), CACHE_NAME, &revisions, 0)
.await
.expect("save hot bucket sweep");
let saved = store.strict_load().await;
assert_eq!(complete, expect_complete, "round {round} outcome");
assert_eq!(saved.info.snapshot_complete, expect_complete, "round {round} saved completion");
if expect_complete {
assert_eq!(saved.info.scan_plan_digest, Some(plan), "round {round} certifies its own plan");
assert!(saved.info.scan_progress.is_none());
let total = saved.checked_flatten("bucket").expect("complete bucket root");
assert_eq!((total.objects, total.versions, total.size), (25, 0, 25));
} else {
assert!(saved.info.scan_plan_digest.is_none(), "round {round} must not certify");
}
}
}
async fn write_checkpoint_object(root: &std::path::Path, object: &str, versions: &[(Option<Uuid>, i64)]) {
let mut metadata = FileMeta::new();
for (index, (version_id, size)) in versions.iter().enumerate() {
@@ -21,7 +21,7 @@ use rustfs_common::mrf_channel::{
use rustfs_heal::heal::{
manager::{HealConfig, HealManager},
mrf_queue::spawn_mrf_consumer,
storage::{HealListItem, HealObjectInfo, HealStorageAPI},
storage::{HealListItem, HealObjectInfo, HealStorageAPI, HealStorageObjectResult},
};
use rustfs_heal_contracts::heal_channel::HealOpts;
@@ -239,12 +239,26 @@ impl HealStorageAPI for NoticeStorage {
async fn mrf_bucket_incarnation_id(&self, _: &str) -> rustfs_heal::Result<Option<Uuid>> {
Ok(Some(self.bucket_incarnation_id))
}
async fn bucket_incarnation_id(&self, _: &str) -> rustfs_heal::Result<Option<Uuid>> {
Ok(Some(self.bucket_incarnation_id))
}
async fn list_buckets(&self) -> rustfs_heal::Result<Vec<BucketInfo>> {
Ok(Vec::new())
}
async fn object_exists(&self, _: &str, _: &str) -> rustfs_heal::Result<bool> {
Ok(true)
}
async fn heal_object_at_incarnation(
&self,
bucket: &str,
object: &str,
version_id: Option<&str>,
expected: Uuid,
opts: &HealOpts,
) -> rustfs_heal::Result<HealStorageObjectResult> {
self.validate_bucket_incarnation(bucket, Some(expected)).await?;
self.heal_object_with_receipt(bucket, object, version_id, opts).await
}
async fn heal_object(
&self,
_: &str,
@@ -396,8 +410,8 @@ async fn mrf_ownership_manager_completion_preserves_scanner_pending() {
if *object == "unknown" {
assert_eq!(
manager.get_statistics().await.total_objects_healed,
1,
"legacy healed count is not repair proof"
0,
"unproved durable repairs must not increment the healed count"
);
}
assert_eq!(
+50 -24
View File
@@ -41,12 +41,20 @@ static STORE_OPEN_POOL: LazyLock<Result<rayon::ThreadPool, rayon::ThreadPoolBuil
.build()
});
enum StoreOpenFailure {
Error(String),
Panicked,
}
enum StoreOpenOutcome<E>
where
E: PluginEvent,
{
Accepted(SharedTarget<E>),
Rejected { panicked: bool, target: SharedTarget<E> },
Rejected {
failure: StoreOpenFailure,
target: SharedTarget<E>,
},
}
fn open_target_store<E>(target: SharedTarget<E>) -> StoreOpenOutcome<E>
@@ -55,8 +63,14 @@ where
{
match catch_unwind(AssertUnwindSafe(|| target.store().map(|store| store.open()))) {
Ok(None | Some(Ok(()))) => StoreOpenOutcome::Accepted(target),
Ok(Some(Err(_))) => StoreOpenOutcome::Rejected { panicked: false, target },
Err(_) => StoreOpenOutcome::Rejected { panicked: true, target },
Ok(Some(Err(error))) => StoreOpenOutcome::Rejected {
failure: StoreOpenFailure::Error(error.to_string()),
target,
},
Err(_) => StoreOpenOutcome::Rejected {
failure: StoreOpenFailure::Panicked,
target,
},
}
}
@@ -215,23 +229,27 @@ where
for outcome in outcomes {
match outcome {
StoreOpenOutcome::Accepted(target) => accepted.push(target),
StoreOpenOutcome::Rejected { panicked, target } => {
if panicked {
tracing::error!(
target_id = %target.id(),
reason = "store_open_panicked",
"Target queue store panicked while opening during runtime handoff"
);
} else {
tracing::error!(
target_id = %target.id(),
reason = "store_open_failed",
"Failed to open target queue store during runtime handoff"
);
}
failures.push(TargetActivationFailure {
detail: format!("{}: queue store open failed", target.id()),
});
StoreOpenOutcome::Rejected { failure, target } => {
let detail = match failure {
StoreOpenFailure::Error(error) => {
tracing::error!(
target_id = %target.id(),
error = %error,
reason = "store_open_failed",
"Failed to open target queue store during runtime handoff"
);
format!("{}: queue store open failed: {error}", target.id())
}
StoreOpenFailure::Panicked => {
tracing::error!(
target_id = %target.id(),
reason = "store_open_panicked",
"Target queue store panicked while opening during runtime handoff"
);
format!("{}: queue store open failed", target.id())
}
};
failures.push(TargetActivationFailure { detail });
rejected.push(target);
}
}
@@ -717,10 +735,18 @@ mod tests {
)));
let observer = target.clone();
let activation = adapter.activate_with_replay(vec![Box::new(target)]).await;
assert!(activation.targets.is_empty());
assert!(activation.replay_workers.is_empty());
let prepared = adapter.prepare_targets(vec![Box::new(target)]).await;
let (opened, rejected) = adapter.open_prepared_stores(prepared);
assert!(opened.targets.is_empty());
let failure = rejected
.failure_summary()
.expect("a rejected target should retain the store error");
assert!(failure.contains(invalid_base.to_string_lossy().as_ref()));
assert!(failure.contains("queue store open failed:"));
adapter
.close_prepared(rejected)
.await
.expect("a rejected target should close cleanly");
assert_eq!(observer.close_call_count(), 1);
}
+53 -5
View File
@@ -85,6 +85,34 @@ const FAILED_STORE_MAX_ENTRIES: usize = 10_000;
/// time, which is the instant the entry entered the failed store.
const FAILED_STORE_TTL: Duration = Duration::from_secs(72 * 60 * 60);
#[derive(Debug)]
struct IoErrorWithPath {
path: PathBuf,
source: std::io::Error,
}
impl std::fmt::Display for IoErrorWithPath {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(formatter, "{}: {}", self.path.display(), self.source)
}
}
impl std::error::Error for IoErrorWithPath {
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
Some(&self.source)
}
}
fn io_error_with_path(path: &Path, error: std::io::Error) -> StoreError {
StoreError::Io(std::io::Error::new(
error.kind(),
IoErrorWithPath {
path: path.to_owned(),
source: error,
},
))
}
/// Writes payload to a temp file in the same directory, flushes the file to disk, then atomically
/// renames it onto final_path.
///
@@ -741,9 +769,9 @@ where
.fs_guard
.write()
.map_err(|_| StoreError::Internal("Failed to acquire write lock on store filesystem".to_string()))?;
std::fs::create_dir_all(&self.directory).map_err(StoreError::Io)?;
std::fs::create_dir_all(&self.directory).map_err(|error| io_error_with_path(&self.directory, error))?;
let dir_entries = std::fs::read_dir(&self.directory).map_err(StoreError::Io)?;
let dir_entries = std::fs::read_dir(&self.directory).map_err(|error| io_error_with_path(&self.directory, error))?;
let mut entries_map = self
.entries
.write()
@@ -751,8 +779,9 @@ where
self.pending_entries.store(0, Ordering::SeqCst);
entries_map.clear();
for entry in dir_entries {
let entry = entry.map_err(StoreError::Io)?;
let metadata = entry.metadata().map_err(StoreError::Io)?;
let entry = entry.map_err(|error| io_error_with_path(&self.directory, error))?;
let entry_path = entry.path();
let metadata = entry.metadata().map_err(|error| io_error_with_path(&entry_path, error))?;
if !metadata.is_file() {
continue;
}
@@ -780,7 +809,7 @@ where
continue;
}
let modified = metadata.modified().map_err(StoreError::Io)?;
let modified = metadata.modified().map_err(|error| io_error_with_path(&entry_path, error))?;
let unix_nano = modified.duration_since(UNIX_EPOCH).unwrap_or_default().as_nanos() as i64;
entries_map.insert(file_name, unix_nano);
}
@@ -1230,6 +1259,25 @@ mod tests {
std::env::temp_dir().join(format!("rustfs-targets-{name}-{}", Uuid::new_v4()))
}
#[test]
fn io_error_with_path_preserves_path_and_source_chain() {
let error = io_error_with_path(
Path::new("/data/.rustfs/events/webhook-primary"),
std::io::Error::from(std::io::ErrorKind::PermissionDenied),
);
assert!(error.to_string().contains("/data/.rustfs/events/webhook-primary"));
let wrapped = std::error::Error::source(&error).expect("store error should expose its I/O error");
let original = std::error::Error::source(wrapped).expect("path wrapper should expose the original I/O error");
assert_eq!(
original
.downcast_ref::<std::io::Error>()
.expect("source should remain an I/O error")
.kind(),
std::io::ErrorKind::PermissionDenied
);
}
#[test]
fn resolve_queue_store_compression_defaults_to_true() {
assert!(resolve_queue_store_compression_from_env_value(None));
+5 -1
View File
@@ -1264,7 +1264,11 @@ mod tests {
match result {
Ok(_) => panic!("expected open_target_queue_store to fail on file base path"),
Err(err) => assert!(err.to_string().contains("custom open context")),
Err(err) => {
let message = err.to_string();
assert!(message.contains("custom open context"));
assert!(message.contains(base.to_string_lossy().as_ref()));
}
}
let _ = fs::remove_file(base);
}
+1 -1
View File
@@ -91,6 +91,6 @@ crypto = ["dep:base64-simd", "dep:hex-simd", "dep:hmac", "dep:hyper", "dep:sha1"
hash = ["dep:highway", "dep:md-5", "dep:sha2", "dep:blake2", "dep:serde", "dep:siphasher", "dep:hex-simd", "dep:crc-fast"]
os = ["dep:rustix", "dep:tempfile", "dep:windows"] # operating system utilities
integration = [] # integration test features
http = ["dep:convert_case", "dep:http", "dep:regex"]
http = ["dep:convert_case", "dep:http", "dep:regex", "string"]
obj = ["http"] # object storage features
full = ["ip", "net", "egress", "io", "hash", "os", "integration", "path", "crypto", "string", "compress", "http", "obj"] # all features
+29
View File
@@ -130,6 +130,19 @@ fn minio_key(suffix: &str) -> String {
format!("{MINIO_PREFIX}{suffix}")
}
/// Reads `x-rustfs-force-delete` / `x-minio-force-delete`.
///
/// `Ok(None)` means the header is absent. `Ok(Some(_))` is a boolean accepted
/// by [`crate::string::parse_bool`]. `Err` means the header is present but not
/// a boolean; callers must reject that request instead of treating the flag as
/// absent, which would turn a malformed force-delete into a normal delete.
pub fn force_delete_header(headers: &HeaderMap) -> std::io::Result<Option<bool>> {
match get_header(headers, SUFFIX_FORCE_DELETE) {
None => Ok(None),
Some(value) => crate::string::parse_bool(value.as_ref()).map(Some),
}
}
/// Get header value: tries x-rustfs-{suffix} first, then x-minio-{suffix}. Case-insensitive.
pub fn get_header<'a>(headers: &'a HeaderMap, suffix: &str) -> Option<Cow<'a, str>> {
let rk = rustfs_key(suffix);
@@ -245,4 +258,20 @@ mod tests {
headers2.insert("X-Rustfs-Force-Delete", HeaderValue::from_static("true"));
assert_eq!(get_header(&headers2, SUFFIX_FORCE_DELETE).as_deref(), Some("true"));
}
#[test]
fn force_delete_header_accepts_minio_bools_and_rejects_garbage() {
let mut headers = HeaderMap::new();
assert_eq!(force_delete_header(&headers).expect("absent header"), None);
headers.insert("x-minio-force-delete", HeaderValue::from_static("TRUE"));
assert_eq!(force_delete_header(&headers).expect("minio true"), Some(true));
headers.insert("x-rustfs-force-delete", HeaderValue::from_static("false"));
assert_eq!(force_delete_header(&headers).expect("rustfs header wins"), Some(false));
headers.clear();
headers.insert("x-rustfs-force-delete", HeaderValue::from_static("maybe"));
assert!(force_delete_header(&headers).is_err());
}
}
+3
View File
@@ -70,6 +70,9 @@ pub const SUFFIX_RESTORE_OPERATION_ID: &str = "restore-operation-id";
pub const SUFFIX_RESTORE_WORKER_LOCK: &str = "restore-worker-lock";
pub const RESTORE_WORKER_LOCK_PROTOCOL_V1: &str = "v1";
pub const SUFFIX_BUCKET_INCARNATION_ID: &str = "bucket-incarnation-id";
/// Upload ID of the multipart completion that published this object version.
/// A retried CompleteMultipartUpload matches it after the staging directory is gone.
pub const SUFFIX_MULTIPART_UPLOAD_ID: &str = "multipart-upload-id";
pub const SUFFIX_OBJECT_TRANSACTION_EPOCH: &str = "object-transaction-epoch";
/// Active rebalance run id mirrored onto `rebalance.bin` object metadata.
pub const SUFFIX_REBALANCE_RUN_ID: &str = "rebalance-run-id";
@@ -284,6 +284,16 @@ Reports scanner-driven bitrot state together with heal queue execution state. `h
Use this route when `metrics.source_work` shows `heal` or `bitrot` queued or missed work. Scanner-originated object checks should appear under `scanner/low`, manual admin heal under `admin/high`. If scanner work grows but admin work remains blocked, treat that as heal queue pressure rather than scanner pacing pressure.
Durable partial-write responsibilities also have node-local MRF metrics:
| Metric | Meaning |
|---|---|
| `rustfs_heal_mrf_queue_depth` | In-memory MRF work plus retained durable responsibilities, including held legacy entries. |
| `rustfs_heal_mrf_unverified_legacy` | Durable `PartialWrite` entries whose completed deep check found healthy legacy data without an independent payload identity proof. The journal entry remains intact; the current process pauses automatic re-dispatch for that entry. |
| `rustfs_heal_mrf_unverified_legacy_oldest_age_seconds` | Age of the oldest such retained responsibility on this node. |
These gauges are emitted per node through the configured OTLP metrics exporter (`RUSTFS_OBS_ENDPOINT` or `RUSTFS_OBS_METRIC_ENDPOINT`). The background-heal status endpoint reports execution tasks; it does not include durable MRF responsibilities. A process restart replays unchanged journal records and checks them again; it does not delete or certify an entry, and it replays all other durable intents on that node as well. For an unversioned object, restore the expected content from a trusted canonical source with protected shard integrity enabled, then restart the node that owns the journal so its intent can obtain an exact receipt. For a versioned object, a protected rewrite creates a new version and does not resolve a responsibility for the old version; preserve the source and only retire that exact version when the intended data is backed up and an authoritative absence proof is appropriate. A successful receipt may discharge only the matching responsibility; an unverified result remains retained and becomes held again. If no trusted copy or expected checksum is available, preserve the held responsibility and investigate its source of truth before retrying. The protected-copy migration in the [shard-integrity audit workflow](shard-integrity-audit.md) writes a new key and preserves its source; completing that migration alone does not prove or discharge an intent for the original key. Never remove MRF journal files manually.
## Heal runtime controls
Heal knobs are environment-only and read by `HealConfig::default` (`crates/heal/src/heal/manager.rs`), the MRF queue (`crates/heal/src/heal/mrf_queue.rs`), or the erasure-set healer (`crates/heal/src/heal/erasure_healer.rs`). The admin `heal` config subsystem accepts only `bitrot_cycle` (`HEAL_KEYS`), which is documented in the scanner table above. Constants live in `crates/config/src/constants/heal.rs` unless another file is named.
+16 -6
View File
@@ -17,6 +17,8 @@ The `main` ruleset (`6436880`) requires exactly these contexts, with `strict_req
Every PR enters `ci.yml`. The `classify-changes` job uses the base revision of `scripts/ci_gate.py` to select a conservative documentation-only path: root Markdown/licenses, `AGENTS.md`, Markdown under `docs/` or `.agents/skills/`, and documentation images. Unknown paths, unavailable Git history, an empty diff, or a missing base policy select the full matrix. Renames include their deleted source path. Documentation-only PRs still run Quick Checks and Typos; the aggregate requires the expensive jobs to be skipped exactly as selected.
PRs changing only Rust sources under `crates/e2e_test/src/` and known E2E selection digests (optionally with those documentation paths) use the `e2e` scope. They keep workspace validation, the server build, E2E smoke, and both S3 lanes; independent workflows retain their path selection. Connect boundary, serial ILM, optional protocol/rio-v2, and real io_uring jobs are skipped because their sources and test inputs are unchanged. Manifests, fixtures, shared nextest configuration, scripts, production sources, and unknown paths still select `full`. A dependency on `e2e_test` from another workspace member or reachable local dependency disables this shortcut. All non-PR events retain their full selection.
`required-checks` runs even after failed or skipped dependencies. `scripts/ci_gate.py verify` rejects missing jobs, unexpected jobs, failure, cancellation, and unexpected skips; optional lanes are required only on their declared events. `Workspace Test and Lint` is the ordinary Rust job, while `Test and Lint` uniquely names the aggregate. New validation jobs must update both its direct dependencies and the script contract. Test this wiring and its failure cases with `python3 scripts/ci_gate.py --self-test`.
Verify the live rule before changing merge policy:
@@ -28,6 +30,12 @@ gh api repos/rustfs/rustfs/rulesets/6436880 \
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.
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.
Protocol matrix jobs finish independently when a sibling fails. Each executed protocol test step preserves its log and available JUnit under a separate, attempt-specific artifact; a skipped test step cannot upload a cached report.
The four site-replication state-writer concurrency proofs reserve the available nextest slots in both local and CI profiles. This isolates their durable IO from unrelated test processes while retaining each proof's internal two-writer race, production five-second lock-acquisition limit, assertions, and zero retries.
## Pull request and merge matrix
"Via aggregate" means a wrong result fails the required `Test and Lint` check. "Report-only" means visible and actionable but outside both the required list and aggregate. Budgets are each job's `timeout-minutes` in the named workflow and are not copied here.
@@ -35,18 +43,18 @@ The aggregate requires the validation lanes already selected by `ci.yml`; this c
| Event | Check name | Workflow / job | Merge status | Reproduce |
|---|---|---|---|---|
| PR, non-doc change | `Quick Checks` | `ci.yml` `quick-checks` | Required | `make pre-commit` |
| PR, non-doc change | `Workspace Test and Lint` | `ci.yml` `test-and-lint` | Via aggregate | `cargo clippy --all-targets -- -D warnings`; `cargo nextest run --profile ci --all --exclude e2e_test`; `cargo test --all --doc`; `scripts/check_migration_gate_count.sh` |
| PR, non-doc change | `Workspace Test and Lint` | `ci.yml` `test-and-lint` | Via aggregate | `cargo clippy --all-targets -- -D warnings`; `cargo nextest run --profile ci --all --exclude e2e_test`; `cargo test --all --doc`; migration evidence check described below |
| PR, non-doc change | `Typos` | `ci.yml` `typos` | Via aggregate | `typos` |
| PR, non-doc change | `ILM Integration (serial)` | `ci.yml` `test-ilm-integration-serial` | Via aggregate | exact command in the job |
| PR, non-doc change | `Test and Lint (rio-v2)`, `Test and Lint (swift)`, `Test and Lint (sftp)` | `ci.yml` `test-and-lint-rio-v2`, `test-and-lint-protocols` | Via aggregate | `cargo nextest run` with the job's `--features` |
| PR, non-doc change | `Connect Short Credential Boundary` | `ci.yml` `connect-short-credential-boundary` | Via aggregate | `cargo test -p rustfs --test connect_registration --features connect-e2e-short-credentials`; `cargo check -p rustfs --release --features connect-e2e-short-credentials` must fail |
| PR, full selection | `ILM Integration (serial)` | `ci.yml` `test-ilm-integration-serial` | Via aggregate | exact command in the job |
| PR, full selection | `Test and Lint (rio-v2)`, `Test and Lint (swift)`, `Test and Lint (sftp)` | `ci.yml` `test-and-lint-rio-v2`, `test-and-lint-protocols` | Via aggregate | `cargo nextest run` with the job's `--features` |
| PR, full selection | `Connect Short Credential Boundary` | `ci.yml` `connect-short-credential-boundary` | Via aggregate | `cargo test -p rustfs --test connect_registration --features connect-e2e-short-credentials`; `cargo check -p rustfs --release --features connect-e2e-short-credentials` must fail |
| PR, full selection | `Offline Enrollment E2E Root Boundary` | `ci.yml` `offline-enrollment-e2e` | Via aggregate | `scripts/check_offline_enrollment_e2e.sh` |
| PR, non-doc change | `Build RustFS Debug Binary` | `ci.yml` `build-rustfs-debug-binary` | Via aggregate; prerequisite for black-box jobs | `python3 scripts/e2e_binary.py build --bins --features e2e-test-hooks` (binary plus its `rustfs.e2e.json` sidecar) |
| PR, non-doc change | `io_uring Integration (real)` | `ci.yml` `uring-integration` | Via aggregate | `cargo test -p rustfs-ecstore --lib uring_ -- --test-threads=1 --nocapture` |
| PR, full selection | `io_uring Integration (real)` | `ci.yml` `uring-integration` | Via aggregate | `cargo test -p rustfs-ecstore --lib uring_ -- --test-threads=1 --nocapture` |
| PR, non-doc change | `End-to-End Tests` | `ci.yml` `e2e-tests` | Via aggregate | `python3 scripts/e2e_binary.py run --features e2e-test-hooks -- cargo nextest run --profile e2e-smoke -p e2e_test`, then `./scripts/e2e-run.sh ./target/debug/rustfs <data-dir>` under the same wrapper; membership guards `scripts/check_test_wiring.py --check-profile e2e-smoke <listing.json>` and `scripts/check_security_smoke_count.sh check <listing.json>` |
| PR, non-doc change | `S3 Implemented Tests` | `ci.yml` `s3-implemented-tests` | Via aggregate | build `rustfs`, then `scripts/s3-tests/run.sh` with the job's `DEPLOY_MODE` / `TEST_MODE` / `MAXFAIL` env |
| PR, non-doc change | `S3 Lifecycle Behavior Tests` | `ci.yml` `s3-lifecycle-behavior-tests` | Via aggregate | `scripts/s3-tests/run.sh` with the job's accelerated-scanner env |
| PR touching `paths` in `audit.yml` | `Cargo Deny`, `Workflow Pin Report`, `Dependency Review` | `audit.yml` `cargo-deny`, `workflow-pin-report`, `dependency-review` | Report-only | `cargo deny check`; `scripts/security/check_workflow_pins.sh` |
| PR touching `paths` in `architecture-migration-rules.yml` | `Architecture Migration Rules` | `architecture-migration-rules.yml` `architecture-migration-rules` | Report-only | `scripts/check_architecture_migration_rules.sh` |
| PR touching `paths` in `nix.yml` | `Nix Build & Check` | `nix.yml` `nix-validation` | Report-only | `nix flake check` |
| PR touching `paths` in `fuzz.yml` | `Build Fuzz Harness`, `Smoke / <target>` | `fuzz.yml` `fuzz-build`, `pr-fuzz-smoke` | Report-only | `MAX_TOTAL_TIME=60 ./scripts/fuzz/run.sh` |
| PR touching `paths` in `windows-filesystem.yml` | `Rename Safety` | `windows-filesystem.yml` `rename-safety` | Report-only | the `cargo test -p rustfs-ecstore --lib <filter>` commands in the job, on Windows |
@@ -59,6 +67,8 @@ The aggregate requires the validation lanes already selected by `ci.yml`; this c
e2e filters live in `.config/nextest.toml`; extend a profile instead of adding a second selector. Before a profile runs, `scripts/check_test_wiring.py` compares its listing to the committed digest in `.config/e2e-<profile>-selection.txt`, so a silent test drop fails closed.
Architecture migration rules run once in the required Quick Checks job. Migration proofs reuse the successful workspace run: `scripts/check_migration_gate_count.sh evidence <core-listing.json> <junit.xml>` checks the unchanged name selection and committed floor, and requires exactly one successful execution without retries for each proof. Missing, filtered, skipped, failed, or duplicate evidence fails the gate. Local `check` and `run` modes remain available. Upgrade compatibility builds one current server and shares its binary and provenance sidecar across the seven cases; each case independently verifies the pinned previous release checksum and runs its existing test.
Scanner usage and heal rebuild coverage are intentionally split by risk and
cost. `data_usage_test` runs in the PR `e2e-smoke` lane so changes that affect
authoritative scanner usage publication, quota-visible usage, or admin usage

Some files were not shown because too many files have changed in this diff Show More