Compare commits

..

22 Commits

Author SHA1 Message Date
唐小鸭 f429676a3e chore(release): prepare 1.0.0-rc.3 2026-08-17 11:58:25 +08:00
Zhengchao An a9691b6797 chore: adjudicate 19 bare dead_code allows across six leaf crates (#6161)
backlog#1823 step 10, batch 1 of the repo-wide item-allow sweep. 227 bare #[allow(dead_code)] remain across 83 files; this takes the 19 in utils, notify, checksums, policy, keystone and trusted-proxies, which are small enough to verify end to end.

Removing all 19 first, before writing any reason, matters: 8 of them suppress nothing. Every allow in utils, one in policy and three in notify sit on items that are publicly reachable, so dead_code never applied to them — the same shape as the swift module and kms's dek.rs. Writing a reason onto a no-op allow would dress noise up as considered judgement, so those are simply deleted.

Three items are genuinely dead and go with their allows: notify's new_target_id_set, the AWS metadata fetcher's get_metadata_token, and policy's empty `pub struct Value;`, none of which is referenced anywhere in the tree.

The remaining eight keep an allow, now saying why the item survives rather than who calls it. Two are exercised only by their own crate's tests (checksums' MD5_HEADER_NAME, policy's is_match_as_pattern_prefix). Four are fields written but never read back: keystone's verify_ssl, parsed from config after the reqwest client is already built; keystone's client handle, which keeps the Keystone client alive for the mapper's lifetime; the AWS IMDS endpoint, kept beside the client while requests build their own URLs; and notify's rules_map, whose own comment retains it for snapshot-time judgements no code performs.

checksums' Md5 needed the most care. Crc32, Sha256 and seven others each have an arm in ChecksumAlgorithm::into_impl, and Md5 has none, which reads like a missing algorithm. It is not: ChecksumAlgorithm has no Md5 variant at all. S3 carries Content-MD5 as its own header, separate from the x-amz-checksum-* family, and this impl exists so both paths share the Checksum trait. The reason records that, so the next reader does not re-derive it.

One measurement note for anyone continuing this sweep: cargo does not re-emit warnings for cached compilations, so a per-crate loop of `cargo check -p <crate>` under-reports. checksums showed zero that way while actually carrying three. Touch the sources and check the crates in one invocation, then attribute by path.

Verification: the six crates are warning-free under cargo check --tests; clippy --lib --tests -D warnings clean; cargo nextest run 1096 passed; make pre-commit exit 0.

Ref rustfs/backlog#1823 (step 10).
2026-08-17 11:34:47 +08:00
Zhengchao An 7db3882777 chore(obs): adjudicate 19 bare dead_code allows (#6162)
backlog#1823 step 10, batch 2. Eighteen of the nineteen suppress nothing and are deleted; one was real and keeps an allow that now says why.

Rotation::Never is constructed only by the rolling-appender tests at rolling.rs:456, 477 and 498, so the lib target reports it as never constructed. Its allow is restored with that reason.

Finding it corrected the method used for batch 1. Removing all nineteen and running cargo check -p rustfs-obs --tests reported zero warnings even after touching every source file, while clippy --lib --tests -D warnings caught Rotation::Never. cargo's warning output is not a reliable completeness check — it does not re-emit for cached compilations, and touching the sources did not cover the lib target here. Later batches should treat clippy -D warnings as the gate; batch 1's six crates were re-checked under clippy and are clean.

Taken with #6086, which cleared this crate's 44 module-level blankets and left six real items, obs has now had 63 dead-code suppressions examined, of which seven were suppressing anything at all. The rest sat on items that are publicly reachable, where dead_code never applied — the same shape as the swift module and kms's dek.rs.

Verification: clippy --lib --tests -D warnings clean in the default, gpu and pyroscope lanes; cargo nextest run -p rustfs-obs 324 passed; make pre-commit exit 0.

Ref rustfs/backlog#1823 (step 10).
2026-08-17 11:32:23 +08:00
houseme 3f3e3f4f05 perf(get): avoid memory body stream wrapper (#6163)
Use MemoryTrackedBytesStream directly as an s3s ByteStream so in-memory GET bodies avoid the generic StreamingBlob::wrap adapter while preserving exact remaining length, request lifecycle tracking, and length-mismatch failure semantics.

Co-authored-by: heihutu <heihutu@gmail.com>
2026-08-17 10:45:24 +08:00
houseme 01e0af6312 perf(io-metrics): avoid get handoff label allocations (#6160)
Co-authored-by: heihutu <heihutu@gmail.com>
2026-08-17 08:26:23 +08:00
Zhengchao An 3377688dab ci(arch): ratchet ecstore module-level lint blankets (#6155)
backlog#1823 step 9. The step 2 burn-down cleared every module-level #![allow(dead_code)] from ecstore, but nothing stops the next PR from adding one back, and the other module-level blankets (unused_variables, unused_must_use, clippy::all) were never counted at all.

Two rules land in check_architecture_migration_rules.sh:

ecstore must carry zero module-level #![allow(dead_code)]. The count is asserted at zero rather than registered, since there is nothing left to grandfather; a genuinely unused item takes an item-level allow with a reason, which is what step 2 produced roughly 350 times.

Every other module-level blanket must match scripts/ecstore-module-lint-register.txt exactly — 94 entries across 33 files, nearly all of them in the MinIO-ported client module.

The exact match is the point. A "no new entries" rule lets the register rot into an amnesty list, which is the failure mode backlog#1834 found in the layer-dependency baseline: rebuilt at 29 entries, 2 more added by a later PR, zero retired. Here, removing a blanket costs one line in the register, so it can only shrink, and adding one shows up as a register line a reviewer has to accept.

Verified by injection, since a guard that cannot fail is worse than no guard: adding a dead_code blanket, adding an unregistered clippy::all blanket, and deleting a registered blanket without updating the register each produce the expected failure, and the tree passes once reverted.

make pre-commit exit 0.

Ref rustfs/backlog#1823 (step 9).
2026-08-17 08:08:36 +08:00
Henry Guo 890ddea94b fix(table-catalog): prevent object catalog lock reentry (#6144)
* fix(table-catalog): prevent object catalog lock reentry

* test(table-catalog): distinguish unlocked object reads

---------

Co-authored-by: Henry Guo <marshawcoco@users.noreply.github.com>
2026-08-17 08:06:53 +08:00
Zhengchao An 9f02ca6c36 fix(ecstore): resolve nine unused bindings in set_disk write and heal paths (#6158)
backlog#1823 step 1, the diagnosis half. Temporarily removing set_disk/mod.rs's #![allow(unused_variables)] surfaced nine bindings. The issue asks that values computed and then dropped on write/quorum paths be diagnosed before being underscored, and that turned out to matter: only four were plain leftovers.

Two errors were bound and then left out of the log they were bound for. complete_multipart_upload's checksum failures read `if let Err(err) = ...` and then log part_id, bucket and object with no `err` anywhere in the message, so a checksum failure in production told you which part failed but not why. Both messages now carry the error.

One is a lock guard. heal's write_lock_guard holds a namespace write lock for the rest of the scope; renaming it to a bare `_` would drop it immediately and release the lock. It is now `_write_lock_guard`, with a comment saying why it must not be `_`.

One was kept alive by a corpse. `errors` in read_multiple_files is read by nothing except two commented-out debug! lines directly below it; the binding and the commented lines go together.

One is a cfg split. heal's disk_index is read only inside the #[cfg(test)] fault-injection branch, so underscoring it would break the test build; a `#[cfg(not(test))] let _ = disk_index;` covers the non-test lane instead.

The remaining four are genuine leftovers: an unused enumerate index in list_object_parts, a discarded error in a heal reader loop, an inner binding shadowing its own iterator variable, and delete_object's write_quorum.

That last one is worth a separate look: delete_object asks get_object_info_and_quorum for a write quorum and never uses it, because delete_object_version below recomputes its own as disks.len() / 2 + 1. The two are not the same number — one comes from the object's erasure configuration, the other is a plain majority of the disk array. Pre-existing behaviour, untouched here.

The blankets stay for now. Removing #![allow(unused_imports)] exposes 76 unused imports in set_disk/mod.rs, and they cannot be removed per-lane: cargo fix, working from the lib lane, produced 54 compile errors in the test lane. That needs its own pass with both lanes checked per import.

Verification: cargo check -p rustfs-ecstore --tests and --features test-util --tests both warning-free; clippy --lib --tests -D warnings clean; cargo nextest run -p rustfs-ecstore 4101 passed; make pre-commit exit 0.

Ref rustfs/backlog#1823 (step 1).
2026-08-17 08:06:37 +08:00
Zhengchao An 33cd11472a refactor(rustfs): move module switches below the layer boundary (#6154)
backlog#1834 PR5. Whether the scanner, heal, audit and notify modules are on gets read from infra (storage helpers, node-service RPC) and from interface (admin handlers), but the switches lived in startup_background (composition) and server (interface). Every one of those reads was an upward edge carried in the layer-dependency baseline.

The env-derived scanner/heal predicates and the audit/notify state cells now live in rustfs/src/module_switches.rs, at the bottom of the layer order, so the same reads are ordinary downward edges. startup_background and server import from there; server keeps re-exporting the getters for its own consumers.

The issue's plan was to move is/refresh_audit/notify_module_enabled as a group. Moving refresh_* wholesale would have dragged resolve_audit_module_state and resolve_notify_module_state — server-side configuration logic — down into infra, which breaks more layering than it fixes. State and resolution are split instead: module_switches owns the atomics plus is_*/set_* accessors, and server's refresh_* keeps the configuration logic and publishes through the setter.

That leaves storage/helper.rs's test module importing refresh_* from server, so two infra->interface edges stay. Those tests assert that a configuration change takes effect through refresh, which a plain setter would no longer exercise; the edges are worth more than the two baseline lines.

Baseline drops 44 -> 36 lines, deletions only:

- 4 interface/infra -> composition edges for ENV_SCANNER_ENABLED, scanner_enabled_from_env and heal_enabled_from_env
- 2 infra -> interface edges for is_audit_module_enabled and is_notify_module_enabled
- cycle|composition<->infra and cycle|composition<->interface

The two cycles were not expected to go until whole subsystems moved out; clearing composition's inbound upward edges dissolved both, leaving three of the original five.

Verification: scripts/check_layer_dependencies.sh passes, cargo check -p rustfs warning-free, make pre-commit exit 0.
2026-08-17 08:05:35 +08:00
Zhengchao An 3ff250f1cd chore(protocols): drop 43 no-op dead_code allows from swift (#6157)
backlog#1823 step 8, partial. The swift module carries 43 #[allow(dead_code)] attributes, most with a comment naming a consumer: "Used by handler", "Handler integration: GET container", "Used by handler and object.rs".

Every one of them suppresses nothing. crates/protocols/src/lib.rs declares `pub mod swift`, and swift/mod.rs declares all 22 submodules `pub mod`, so every item is publicly reachable and dead_code never applied to it. Removing all 43 leaves the warning count at zero, in both the default and --features swift lanes.

That is also why those comments survived. They assert who calls the item — a claim the compiler normally settles on its own — and the compiler had been silenced by the visibility chain.

The rest of step 8 needs a decision this PR does not make. Downgrading the 22 submodules to `pub(crate) mod` does restore detection, and it surfaces 39 real items, 16 of them the whole of sync.rs: SyncConfig, SyncStatus, SyncQueueEntry, ConflictResolution and every function and constant around them, i.e. Swift container sync is built and never wired.

But the `pub mod` chain is load-bearing. Six integration tests under crates/protocols/tests are separate crates that import the submodules directly (swift::quota, swift::slo, swift::symlink, swift::sync, swift::tempurl, swift::container), and the downgrade fails to compile them. Restoring dead-code detection for this module therefore depends on first deciding whether those tests move in-crate — which is a testing-strategy call, not a cleanup one.

Verification: cargo check -p rustfs-protocols warning-free in the default lane and with --features swift (lib and --tests); clippy --features swift --lib --tests -D warnings clean; cargo nextest run -p rustfs-protocols --features swift 441 passed; make pre-commit exit 0.

Ref rustfs/backlog#1823 (step 8).
2026-08-17 08:03:29 +08:00
Zhengchao An d795729585 fix(scanner): canonicalize temp_dir for drive field assertion on macOS (#6159) 2026-08-17 07:11:25 +08:00
Henry Guo 9e6e02ea09 fix(table-catalog): assign fresh schema IDs on create (#6146)
* fix(table-catalog): assign fresh schema IDs on create

* fix(table-catalog): accept negative create schema IDs

---------

Co-authored-by: Henry Guo <marshawcoco@users.noreply.github.com>
2026-08-17 01:06:25 +08:00
houseme 39274fc37c feat(ecstore): default bounded metadata fanout (#6156)
Co-authored-by: heihutu <heihutu@gmail.com>
2026-08-17 00:56:20 +08:00
houseme 33eff4c3c4 test(ecstore): add metadata slow-tail fault hook (#6150)
Add a diagnostic metadata-only read_version delay hook for GET data-read fanout so bounded/default behavior can be compared under controlled slow-tail metadata responses.

Co-authored-by: heihutu <heihutu@gmail.com>
2026-08-16 23:36:46 +08:00
Zhengchao An a2f16aa066 test(tier): pin the compressed transitioned read against its stored bytes (#6151)
#6107 routed the transitioned read through the object's own ReadPlan so a tiered SSE object stops serving ciphertext. Compression rides that same plan and got fixed with it, but nothing pins it: revert the routing and a compressed object that ILM moved to a warm tier returns its stored (compressed) bytes under the compressed size, with every existing test still green.

The gap is easy to reopen because transition genuinely uploads the stored representation — the upload side is correct and the read side is the only place that can decode it. These tests state that contract at the boundary where it broke.

Four cases, all through SetDisks::get_object_reader against a mock warm tier:

- a full GET of a transitioned compressed object returns the plaintext and publishes the plaintext size (the test also asserts the remote copy holds the compressed bytes, so it fails loudly if the upload side ever changes instead);
- a ranged GET returns that plaintext slice, with the range deliberately starting past the compressed size so a range still measured in stored coordinates cannot produce it;
- a restore read still receives the stored bytes under the stored size — restore_request_active holds it on the Plain branch, and decompressing there would write plaintext under compressed metadata;
- a plain transitioned object still reads back byte-identical, full and ranged.

Verified as guards, not decoration: forcing the tiered read back onto the Plain branch turns the two compressed tests red and leaves the plain and restore tests green.

What these do not pin, so the gap stays recorded rather than implied covered: the fixture carries no compression index, so part.index stays None and the plan's storage offset is always 0 — the compressed-offset translation itself is still untested, as are multipart compressed objects, partNumber reads, and the encrypted tiered read that #6107 targeted.
2026-08-16 15:27:59 +00:00
Zhengchao An 4c8b9f87e1 chore(ecstore): annotate the two dead fns the blanket removals missed (#6152) 2026-08-16 22:42:51 +08:00
Zhengchao An 3272730c13 fix(ecstore): silence two dead_code warnings left on main (#6153) 2026-08-16 22:42:45 +08:00
Zhengchao An 1862112d0c chore: drop the remaining product-code dead_code blankets (#6149) 2026-08-16 14:18:00 +00:00
Zhengchao An cd0ac02879 test(interop): let the MinIO fixture lab build from registry mirrors (#6148) 2026-08-16 14:05:27 +00:00
Zhengchao An 6cf9cf7bb5 chore(ecstore): drop the bucket dead_code blanket (#6147)
* chore(ecstore): drop the bucket dead_code blanket

The last blanket of the backlog#1823 burn-down, and the largest: 71 items across lifecycle, replication, metadata, quota, object lock and bucket utils. Four are deleted.

Deleted, all trivial:

- check_valid_object_name and check_valid_object_name_prefix, a pair that only calls into each other with no external caller. Worth stating plainly so nobody reads this as a validation gap: object names are validated through check_object_name_for_length_and_slash, which is live; this pair is a second, unwired entry point.
- DEFAULT_HEALTH_CHECK_RELOAD_DURATION, a lone unused constant.
- The LifecycleReplicationConfig alias, which orphaned a re-export in replication/mod.rs that goes with it.

Everything else is kept, in four groups, because the blanket here was hiding structure rather than rot:

Windows platform gating. WINDOWS_RESERVED_NAMES, the two reason constants and object_name_has_windows_incompatible_segment are called from inside the #[cfg(target_os = "windows")] block in check_object_name_for_length_and_slash (utils.rs:228-255), so they only read as dead on non-Windows hosts. As with the Linux gating in the disk root, this cannot be adjudicated locally: cargo check for both x86_64-pc-windows-msvc and x86_64-unknown-linux-gnu fails in the aws-lc-sys build script for want of a cross C toolchain. CI covers both.

Declared boundary surface. The *_boundary.rs and *_bridge.rs files carry the replication split plan's contracts, which scripts/check_architecture_migration_rules.sh pins through the EcstoreReplicationBoundaryImports section of the split-plan doc. Their unused items are declarations, not leftovers.

test-util seams. ConfigWriteLockProbe with install/wait_until_attempted follows the same pattern as the barriers in the services and set_disk roots.

MinIO-parity tier/lifecycle entry points that this port never wired: apply_lifecycle_action, get_transitioned_object_reader, recover_tier_free_versions, delete_object_from_remote_tier, abort_tier_delete_journal_entry and the replication pool's worker-management surface. These are complete, substantial machinery with no caller — the same shape as data_usage's local_snapshot feature. Removing them is a product decision, so they are made explicit here rather than deleted.

Verification, four lanes warning-free: default, --tests, --features rio-v2 --tests, --features test-util --tests. cargo nextest run -p rustfs-ecstore 4096 passed; clippy --lib --tests -D warnings clean; make pre-commit exit 0. Note that clippy is what caught the orphaned re-export above: cargo check and pre-commit both treat unused_imports as a warning.

Ref rustfs/backlog#1823 (step 2, final root).

* chore(ecstore): correct inaccurate dead_code reasons in the bucket root

Six items were labelled 'asserted by this file's tests' or as MinIO-parity
entry points while having no caller at all - free get_bucket_acl_config and
created_at only reach their own live methods (production goes through
created_at_in), BucketVersioningSys::get_in, utils::serialize_content and
ServiceType have no reference anywhere, and with_transition_queue_env_async
is an unused test fixture, not a tier entry point. Name what each one is so
the next reader does not assume coverage that is not there.

Ref rustfs/backlog#1823.
2026-08-16 21:39:04 +08:00
Zhengchao An f1f86ee9d0 chore(ecstore): drop the set_disk dead_code blanket (#6141)
* chore(ecstore): drop the set_disk dead_code blanket

Removing the blanket exposes 39 items; exactly one is deleted. The low share is a finding, not caution: unlike the disk root, where platform gating made local adjudication impossible, here the items were checked and nearly all of them are live.

Deleted: HealEntryResult, the only item with no reference anywhere.

What the checks turned up, in the order the warnings suggest deleting them:

SetDisks::rename_data looked like the head of a dead chain feeding into_legacy_tuple and RenameDataLegacyTuple. It is not: production goes through rename_data_owned, and rename_data itself has test callers at mod.rs:5809 and 5880. The chain below it is therefore live through the tests, and inferring "this is dead, so its callee is dead" would have removed three working items.

create_bitrot_readers_until_quorum, read_multiple_files and map_cleanup_join_result all have callers inside their files' test modules, so they only look dead in the lib target.

TransitionCommitBarrier and TransitionUploadedSaveProbe, with their install/wait_until_paused/release surfaces, are installed by tests behind #[cfg(all(test, feature = "test-util"))].

ctx.rs's SetDisksCtx accessors are the split seam left by the SetDisks god-object break-up (backlog#815).

heal_object_dir's two apparent references are comments, and they document an index-alignment contract that live code maintains for it, so they stay as they are.

Worth a maintainer decision: the metadata early-stop switch has a complete percentage-rollout facet — ENV_RUSTFS_GET_METADATA_EARLY_STOP_ROLLOUT_PCT, get_metadata_early_stop_rollout_pct and should_use_metadata_early_stop — with no caller, no test and no documentation, while its sibling enable flag is live. It is kept with an allow that says so rather than removed, since a rollout knob is a product call.

One placement note for anyone adding allows near heal code: check_logging_guardrails.sh requires #[instrument(level = "trace")] to sit immediately before async fn heal_object_dir, so the allow goes above the instrument attribute. Putting it between the two drops the guard's match count and fails the check.

Verification, four lanes warning-free: default, --tests, --features rio-v2 --tests, --features test-util --tests. cargo nextest run -p rustfs-ecstore 4096 passed; clippy --lib --tests -D warnings clean; make pre-commit exit 0.

Ref rustfs/backlog#1823 (step 2).

* chore(ecstore): fix duplicated and inaccurate dead_code reasons in set_disk

format_lock_error carried the same #[allow] twice. Five items in the
locking/heal roots were labelled 'asserted by this file's tests' while
having no reference at all - heal_object_dir's only two references are
comments, as this branch's own notes point out. Say what each item
actually is instead, so the next reader does not assume test coverage
that is not there.

Ref rustfs/backlog#1823.

* chore(ecstore): correct the bounded_spare_disk_index dead_code reason

The mod.rs copy is an unused test fixture, not something this module's
tests assert; the namesake that is exercised lives in the io_primitives
test module.

Ref rustfs/backlog#1823.
2026-08-16 21:38:46 +08:00
Zhengchao An 1eef0de003 chore(ecstore): drop the disk dead_code blanket (#6139)
* chore(ecstore): drop the disk dead_code blanket

Removing the blanket exposes 36 items in the lowest storage layer: 7 deleted, 29 kept with reasoned item-level allows. That is the smallest deletion share of this burn-down, and the reason is a verification limit rather than a judgement call.

disk/local.rs carries 141 `#[cfg(target_os = "linux")]` sites — the densest platform gating in the tree, because O_DIRECT and io_uring only exist there. The direct-I/O cluster (six ENV_RUSTFS_OBJECT_DIRECT_IO_* constants plus is_direct_io_read_enabled, is_direct_io_write_enabled, get_direct_io_read_threshold, direct_write_staging_capacity, direct_write_tail_split and DIRECT_WRITE_STAGING_BYTES) reads as dead on macOS purely because its production callers at local.rs:1766, 3114 and 4605 sit inside Linux-gated blocks. direct_write_staging_capacity even documents itself as "Platform-independent (no O_DIRECT), so it is unit-tested on any host".

Deleting those would leave every local check green — 4096 tests pass, clippy is clean, make pre-commit exits 0 — and break the Linux build in CI, because all four local lanes compile for aarch64-apple-darwin. Cross-checking locally is not available either: cargo check --target x86_64-unknown-linux-gnu fails in the aws-lc-sys build script for want of a Linux C cross-compiler. Their allows name the platform reason so the next reader on a non-Linux host does not repeat the investigation.

Deleted, all in files with no target_os gating at all (os.rs, disk_store.rs):

- HealthDiskCtxKey and HealthDiskCtxValue with its private log_success. Note that DiskHealthTracker::log_success is a different method of the same name and is live from cluster/rpc/peer_s3_client.rs and remote_disk.rs — the two have to be told apart by type, not by name.
- LocalDiskWrapper::new_with_health and check_id.
- os.rs file_exists and lock_destination_directory_for_path_access.

Kept with allows: DiskHealthTracker's set_faulty, mark_offline, waiting_count and last_success have test callers in remote_disk.rs, so they only look dead in the lib target. to_disk_error, remove_all and sync_dir_files are asserted by their own files' tests. The reclaim, mmap and path-cache field groups are written but never read back.

Placement follows the same rule as the earlier roots: per-method allows inside impl DiskHealthTracker and impl LocalDisk, since both are mostly live and a block-level allow would be a smaller version of the blanket this issue removes. Struct-level allows are used only where the warning covers that struct's own fields. The three cached_read_env! functions take their allow inside the macro invocation, before the fn line, because the macro forwards $(#[$meta:meta])* onto the generated item.

Verification, four lanes warning-free: default, --tests, --features rio-v2 --tests, --features test-util --tests. cargo nextest run -p rustfs-ecstore 4096 passed; clippy --lib --tests -D warnings clean; make pre-commit exit 0. The Linux lane is not covered locally and is left to CI.

Ref rustfs/backlog#1823 (step 2).

* chore(ecstore): correct two dead_code reasons in the disk root

check_valid_path and reject_symlink_components have no caller at all -
not even a test - so 'asserted by this file's tests' misreads them as
covered. Both are method wrappers over live free functions; say that
instead.

Ref rustfs/backlog#1823.
2026-08-16 21:38:37 +08:00
114 changed files with 2286 additions and 533 deletions
Generated
+47 -47
View File
@@ -3764,7 +3764,7 @@ checksum = "d0881ea181b1df73ff77ffaaf9c7544ecc11e82fba9b5f27b262a3c73a332555"
[[package]]
name = "e2e_test"
version = "1.0.0-rc.2"
version = "1.0.0-rc.3"
dependencies = [
"anyhow",
"astral-tokio-tar",
@@ -9093,7 +9093,7 @@ dependencies = [
[[package]]
name = "rustfs"
version = "1.0.0-rc.2"
version = "1.0.0-rc.3"
dependencies = [
"aes-gcm",
"anyhow",
@@ -9231,7 +9231,7 @@ dependencies = [
[[package]]
name = "rustfs-audit"
version = "1.0.0-rc.2"
version = "1.0.0-rc.3"
dependencies = [
"async-trait",
"const-str",
@@ -9254,7 +9254,7 @@ dependencies = [
[[package]]
name = "rustfs-checksums"
version = "1.0.0-rc.2"
version = "1.0.0-rc.3"
dependencies = [
"base64-simd",
"bytes",
@@ -9270,7 +9270,7 @@ dependencies = [
[[package]]
name = "rustfs-common"
version = "1.0.0-rc.2"
version = "1.0.0-rc.3"
dependencies = [
"chrono",
"hotpath",
@@ -9288,7 +9288,7 @@ dependencies = [
[[package]]
name = "rustfs-concurrency"
version = "1.0.0-rc.2"
version = "1.0.0-rc.3"
dependencies = [
"hotpath",
"insta",
@@ -9301,7 +9301,7 @@ dependencies = [
[[package]]
name = "rustfs-config"
version = "1.0.0-rc.2"
version = "1.0.0-rc.3"
dependencies = [
"const-str",
"hotpath",
@@ -9311,7 +9311,7 @@ dependencies = [
[[package]]
name = "rustfs-credentials"
version = "1.0.0-rc.2"
version = "1.0.0-rc.3"
dependencies = [
"base64-simd",
"hmac 0.13.0",
@@ -9325,7 +9325,7 @@ dependencies = [
[[package]]
name = "rustfs-crypto"
version = "1.0.0-rc.2"
version = "1.0.0-rc.3"
dependencies = [
"aes-gcm",
"argon2",
@@ -9346,7 +9346,7 @@ dependencies = [
[[package]]
name = "rustfs-data-usage"
version = "1.0.0-rc.2"
version = "1.0.0-rc.3"
dependencies = [
"hotpath",
"rmp-serde",
@@ -9356,7 +9356,7 @@ dependencies = [
[[package]]
name = "rustfs-ecstore"
version = "1.0.0-rc.2"
version = "1.0.0-rc.3"
dependencies = [
"arc-swap",
"async-channel",
@@ -9495,7 +9495,7 @@ dependencies = [
[[package]]
name = "rustfs-extension-schema"
version = "1.0.0-rc.2"
version = "1.0.0-rc.3"
dependencies = [
"hotpath",
"serde",
@@ -9505,7 +9505,7 @@ dependencies = [
[[package]]
name = "rustfs-filemeta"
version = "1.0.0-rc.2"
version = "1.0.0-rc.3"
dependencies = [
"arc-swap",
"byteorder",
@@ -9532,7 +9532,7 @@ dependencies = [
[[package]]
name = "rustfs-heal"
version = "1.0.0-rc.2"
version = "1.0.0-rc.3"
dependencies = [
"async-trait",
"base64 0.23.1",
@@ -9563,7 +9563,7 @@ dependencies = [
[[package]]
name = "rustfs-iam"
version = "1.0.0-rc.2"
version = "1.0.0-rc.3"
dependencies = [
"arc-swap",
"async-trait",
@@ -9604,7 +9604,7 @@ dependencies = [
[[package]]
name = "rustfs-io-core"
version = "1.0.0-rc.2"
version = "1.0.0-rc.3"
dependencies = [
"bytes",
"hotpath",
@@ -9617,7 +9617,7 @@ dependencies = [
[[package]]
name = "rustfs-io-metrics"
version = "1.0.0-rc.2"
version = "1.0.0-rc.3"
dependencies = [
"criterion",
"hotpath",
@@ -9681,7 +9681,7 @@ dependencies = [
[[package]]
name = "rustfs-keystone"
version = "1.0.0-rc.2"
version = "1.0.0-rc.3"
dependencies = [
"bytes",
"futures",
@@ -9708,7 +9708,7 @@ dependencies = [
[[package]]
name = "rustfs-kms"
version = "1.0.0-rc.2"
version = "1.0.0-rc.3"
dependencies = [
"aes-gcm",
"anyhow",
@@ -9757,7 +9757,7 @@ dependencies = [
[[package]]
name = "rustfs-lifecycle"
version = "1.0.0-rc.2"
version = "1.0.0-rc.3"
dependencies = [
"async-trait",
"hotpath",
@@ -9780,7 +9780,7 @@ dependencies = [
[[package]]
name = "rustfs-lock"
version = "1.0.0-rc.2"
version = "1.0.0-rc.3"
dependencies = [
"async-trait",
"compact_str",
@@ -9803,7 +9803,7 @@ dependencies = [
[[package]]
name = "rustfs-log-analyzer"
version = "1.0.0-rc.2"
version = "1.0.0-rc.3"
dependencies = [
"chrono",
"flate2",
@@ -9822,7 +9822,7 @@ dependencies = [
[[package]]
name = "rustfs-madmin"
version = "1.0.0-rc.2"
version = "1.0.0-rc.3"
dependencies = [
"hotpath",
"humantime",
@@ -9837,7 +9837,7 @@ dependencies = [
[[package]]
name = "rustfs-notify"
version = "1.0.0-rc.2"
version = "1.0.0-rc.3"
dependencies = [
"arc-swap",
"async-trait",
@@ -9872,7 +9872,7 @@ dependencies = [
[[package]]
name = "rustfs-object-capacity"
version = "1.0.0-rc.2"
version = "1.0.0-rc.3"
dependencies = [
"criterion",
"futures",
@@ -9892,7 +9892,7 @@ dependencies = [
[[package]]
name = "rustfs-object-data-cache"
version = "1.0.0-rc.2"
version = "1.0.0-rc.3"
dependencies = [
"bytes",
"criterion",
@@ -9909,7 +9909,7 @@ dependencies = [
[[package]]
name = "rustfs-obs"
version = "1.0.0-rc.2"
version = "1.0.0-rc.3"
dependencies = [
"chrono",
"crossbeam-channel",
@@ -9964,7 +9964,7 @@ dependencies = [
[[package]]
name = "rustfs-policy"
version = "1.0.0-rc.2"
version = "1.0.0-rc.3"
dependencies = [
"async-trait",
"base64-simd",
@@ -9995,7 +9995,7 @@ dependencies = [
[[package]]
name = "rustfs-protocols"
version = "1.0.0-rc.2"
version = "1.0.0-rc.3"
dependencies = [
"astral-tokio-tar",
"async-compression",
@@ -10057,7 +10057,7 @@ dependencies = [
[[package]]
name = "rustfs-protos"
version = "1.0.0-rc.2"
version = "1.0.0-rc.3"
dependencies = [
"flatbuffers",
"hotpath",
@@ -10081,7 +10081,7 @@ dependencies = [
[[package]]
name = "rustfs-replication"
version = "1.0.0-rc.2"
version = "1.0.0-rc.3"
dependencies = [
"byteorder",
"bytes",
@@ -10099,7 +10099,7 @@ dependencies = [
[[package]]
name = "rustfs-rio"
version = "1.0.0-rc.2"
version = "1.0.0-rc.3"
dependencies = [
"aes-gcm",
"arc-swap",
@@ -10137,7 +10137,7 @@ dependencies = [
[[package]]
name = "rustfs-rio-v2"
version = "1.0.0-rc.2"
version = "1.0.0-rc.3"
dependencies = [
"aes-gcm",
"bytes",
@@ -10160,7 +10160,7 @@ dependencies = [
[[package]]
name = "rustfs-s3-ops"
version = "1.0.0-rc.2"
version = "1.0.0-rc.3"
dependencies = [
"hotpath",
"rustfs-s3-types",
@@ -10168,7 +10168,7 @@ dependencies = [
[[package]]
name = "rustfs-s3-types"
version = "1.0.0-rc.2"
version = "1.0.0-rc.3"
dependencies = [
"hotpath",
"serde",
@@ -10177,7 +10177,7 @@ dependencies = [
[[package]]
name = "rustfs-s3select-api"
version = "1.0.0-rc.2"
version = "1.0.0-rc.3"
dependencies = [
"async-trait",
"bytes",
@@ -10207,7 +10207,7 @@ dependencies = [
[[package]]
name = "rustfs-s3select-query"
version = "1.0.0-rc.2"
version = "1.0.0-rc.3"
dependencies = [
"async-recursion",
"async-trait",
@@ -10226,7 +10226,7 @@ dependencies = [
[[package]]
name = "rustfs-scanner"
version = "1.0.0-rc.2"
version = "1.0.0-rc.3"
dependencies = [
"async-trait",
"bytes",
@@ -10266,7 +10266,7 @@ dependencies = [
[[package]]
name = "rustfs-security-governance"
version = "1.0.0-rc.2"
version = "1.0.0-rc.3"
dependencies = [
"hotpath",
"thiserror 2.0.20",
@@ -10274,7 +10274,7 @@ dependencies = [
[[package]]
name = "rustfs-signer"
version = "1.0.0-rc.2"
version = "1.0.0-rc.3"
dependencies = [
"base64-simd",
"bytes",
@@ -10292,7 +10292,7 @@ dependencies = [
[[package]]
name = "rustfs-storage-api"
version = "1.0.0-rc.2"
version = "1.0.0-rc.3"
dependencies = [
"async-trait",
"hotpath",
@@ -10307,7 +10307,7 @@ dependencies = [
[[package]]
name = "rustfs-targets"
version = "1.0.0-rc.2"
version = "1.0.0-rc.3"
dependencies = [
"arc-swap",
"async-nats",
@@ -10361,7 +10361,7 @@ dependencies = [
[[package]]
name = "rustfs-test-utils"
version = "1.0.0-rc.2"
version = "1.0.0-rc.3"
dependencies = [
"hotpath",
"rustfs-data-usage",
@@ -10377,7 +10377,7 @@ dependencies = [
[[package]]
name = "rustfs-tls-runtime"
version = "1.0.0-rc.2"
version = "1.0.0-rc.3"
dependencies = [
"arc-swap",
"hotpath",
@@ -10398,7 +10398,7 @@ dependencies = [
[[package]]
name = "rustfs-trusted-proxies"
version = "1.0.0-rc.2"
version = "1.0.0-rc.3"
dependencies = [
"async-trait",
"axum",
@@ -10435,7 +10435,7 @@ dependencies = [
[[package]]
name = "rustfs-utils"
version = "1.0.0-rc.2"
version = "1.0.0-rc.3"
dependencies = [
"base64-simd",
"blake2",
@@ -10477,7 +10477,7 @@ dependencies = [
[[package]]
name = "rustfs-zip"
version = "1.0.0-rc.2"
version = "1.0.0-rc.3"
dependencies = [
"astral-tokio-tar",
"async-compression",
+47 -47
View File
@@ -69,7 +69,7 @@ edition = "2024"
license = "Apache-2.0"
repository = "https://github.com/rustfs/rustfs"
rust-version = "1.97.1"
version = "1.0.0-rc.2"
version = "1.0.0-rc.3"
homepage = "https://rustfs.com"
description = "RustFS is a high-performance distributed object storage software built using Rust, one of the most popular languages worldwide. "
keywords = ["RustFS", "Minio", "object-storage", "filesystem", "s3"]
@@ -86,52 +86,52 @@ redundant_clone = "warn"
[workspace.dependencies]
# RustFS Internal Crates
rustfs = { path = "./rustfs", version = "1.0.0-rc.2" }
rustfs-heal = { path = "crates/heal", version = "1.0.0-rc.2" }
rustfs-audit = { path = "crates/audit", version = "1.0.0-rc.2" }
rustfs-checksums = { path = "crates/checksums", version = "1.0.0-rc.2" }
rustfs-common = { path = "crates/common", version = "1.0.0-rc.2" }
rustfs-data-usage = { path = "crates/data-usage", version = "1.0.0-rc.2" }
rustfs-config = { path = "./crates/config", version = "1.0.0-rc.2" }
rustfs-concurrency = { path = "./crates/concurrency", version = "1.0.0-rc.2" }
rustfs-credentials = { path = "crates/credentials", version = "1.0.0-rc.2" }
rustfs-crypto = { path = "crates/crypto", version = "1.0.0-rc.2" }
rustfs-ecstore = { path = "crates/ecstore", version = "1.0.0-rc.2" }
rustfs-filemeta = { path = "crates/filemeta", version = "1.0.0-rc.2" }
rustfs-iam = { path = "crates/iam", version = "1.0.0-rc.2" }
rustfs-keystone = { path = "crates/keystone", version = "1.0.0-rc.2" }
rustfs-lifecycle = { path = "crates/lifecycle", version = "1.0.0-rc.2" }
rustfs-kms = { path = "crates/kms", version = "1.0.0-rc.2" }
rustfs-lock = { path = "crates/lock", version = "1.0.0-rc.2" }
rustfs-madmin = { path = "crates/madmin", version = "1.0.0-rc.2" }
rustfs-notify = { path = "crates/notify", version = "1.0.0-rc.2" }
rustfs-io-metrics = { path = "crates/io-metrics", version = "1.0.0-rc.2" }
rustfs-io-core = { path = "crates/io-core", version = "1.0.0-rc.2" }
rustfs-object-capacity = { path = "crates/object-capacity", version = "1.0.0-rc.2" }
rustfs-object-data-cache = { path = "crates/object-data-cache", version = "1.0.0-rc.2", default-features = false }
rustfs-log-analyzer = { path = "crates/log-analyzer", version = "1.0.0-rc.2" }
rustfs-obs = { path = "crates/obs", version = "1.0.0-rc.2" }
rustfs-policy = { path = "crates/policy", version = "1.0.0-rc.2" }
rustfs-protos = { path = "crates/protos", version = "1.0.0-rc.2" }
rustfs-protocols = { path = "crates/protocols", version = "1.0.0-rc.2" }
rustfs-replication = { path = "crates/replication", version = "1.0.0-rc.2" }
rustfs-rio = { path = "crates/rio", version = "1.0.0-rc.2" }
rustfs-rio-v2 = { path = "crates/rio-v2", version = "1.0.0-rc.2" }
rustfs-s3-types = { path = "crates/s3-types", version = "1.0.0-rc.2" }
rustfs-s3-ops = { path = "crates/s3-ops", version = "1.0.0-rc.2" }
rustfs-s3select-api = { path = "crates/s3select-api", version = "1.0.0-rc.2" }
rustfs-s3select-query = { path = "crates/s3select-query", version = "1.0.0-rc.2" }
rustfs-scanner = { path = "crates/scanner", version = "1.0.0-rc.2" }
rustfs-security-governance = { path = "crates/security-governance", version = "1.0.0-rc.2" }
rustfs-extension-schema = { path = "crates/extension-schema", version = "1.0.0-rc.2" }
rustfs-signer = { path = "crates/signer", version = "1.0.0-rc.2" }
rustfs-storage-api = { path = "crates/storage-api", version = "1.0.0-rc.2" }
rustfs-trusted-proxies = { path = "crates/trusted-proxies", version = "1.0.0-rc.2" }
rustfs-targets = { path = "crates/targets", version = "1.0.0-rc.2" }
rustfs-test-utils = { path = "crates/test-utils", version = "1.0.0-rc.2" }
rustfs-tls-runtime = { path = "crates/tls-runtime", version = "1.0.0-rc.2" }
rustfs-utils = { path = "crates/utils", version = "1.0.0-rc.2" }
rustfs-zip = { path = "./crates/zip", version = "1.0.0-rc.2" }
rustfs = { path = "./rustfs", version = "1.0.0-rc.3" }
rustfs-heal = { path = "crates/heal", version = "1.0.0-rc.3" }
rustfs-audit = { path = "crates/audit", version = "1.0.0-rc.3" }
rustfs-checksums = { path = "crates/checksums", version = "1.0.0-rc.3" }
rustfs-common = { path = "crates/common", version = "1.0.0-rc.3" }
rustfs-data-usage = { path = "crates/data-usage", version = "1.0.0-rc.3" }
rustfs-config = { path = "./crates/config", version = "1.0.0-rc.3" }
rustfs-concurrency = { path = "./crates/concurrency", version = "1.0.0-rc.3" }
rustfs-credentials = { path = "crates/credentials", version = "1.0.0-rc.3" }
rustfs-crypto = { path = "crates/crypto", version = "1.0.0-rc.3" }
rustfs-ecstore = { path = "crates/ecstore", version = "1.0.0-rc.3" }
rustfs-filemeta = { path = "crates/filemeta", version = "1.0.0-rc.3" }
rustfs-iam = { path = "crates/iam", version = "1.0.0-rc.3" }
rustfs-keystone = { path = "crates/keystone", version = "1.0.0-rc.3" }
rustfs-lifecycle = { path = "crates/lifecycle", version = "1.0.0-rc.3" }
rustfs-kms = { path = "crates/kms", version = "1.0.0-rc.3" }
rustfs-lock = { path = "crates/lock", version = "1.0.0-rc.3" }
rustfs-madmin = { path = "crates/madmin", version = "1.0.0-rc.3" }
rustfs-notify = { path = "crates/notify", version = "1.0.0-rc.3" }
rustfs-io-metrics = { path = "crates/io-metrics", version = "1.0.0-rc.3" }
rustfs-io-core = { path = "crates/io-core", version = "1.0.0-rc.3" }
rustfs-object-capacity = { path = "crates/object-capacity", version = "1.0.0-rc.3" }
rustfs-object-data-cache = { path = "crates/object-data-cache", version = "1.0.0-rc.3", default-features = false }
rustfs-log-analyzer = { path = "crates/log-analyzer", version = "1.0.0-rc.3" }
rustfs-obs = { path = "crates/obs", version = "1.0.0-rc.3" }
rustfs-policy = { path = "crates/policy", version = "1.0.0-rc.3" }
rustfs-protos = { path = "crates/protos", version = "1.0.0-rc.3" }
rustfs-protocols = { path = "crates/protocols", version = "1.0.0-rc.3" }
rustfs-replication = { path = "crates/replication", version = "1.0.0-rc.3" }
rustfs-rio = { path = "crates/rio", version = "1.0.0-rc.3" }
rustfs-rio-v2 = { path = "crates/rio-v2", version = "1.0.0-rc.3" }
rustfs-s3-types = { path = "crates/s3-types", version = "1.0.0-rc.3" }
rustfs-s3-ops = { path = "crates/s3-ops", version = "1.0.0-rc.3" }
rustfs-s3select-api = { path = "crates/s3select-api", version = "1.0.0-rc.3" }
rustfs-s3select-query = { path = "crates/s3select-query", version = "1.0.0-rc.3" }
rustfs-scanner = { path = "crates/scanner", version = "1.0.0-rc.3" }
rustfs-security-governance = { path = "crates/security-governance", version = "1.0.0-rc.3" }
rustfs-extension-schema = { path = "crates/extension-schema", version = "1.0.0-rc.3" }
rustfs-signer = { path = "crates/signer", version = "1.0.0-rc.3" }
rustfs-storage-api = { path = "crates/storage-api", version = "1.0.0-rc.3" }
rustfs-trusted-proxies = { path = "crates/trusted-proxies", version = "1.0.0-rc.3" }
rustfs-targets = { path = "crates/targets", version = "1.0.0-rc.3" }
rustfs-test-utils = { path = "crates/test-utils", version = "1.0.0-rc.3" }
rustfs-tls-runtime = { path = "crates/tls-runtime", version = "1.0.0-rc.3" }
rustfs-utils = { path = "crates/utils", version = "1.0.0-rc.3" }
rustfs-zip = { path = "./crates/zip", version = "1.0.0-rc.3" }
# Async Runtime and Networking
async-channel = "2.5.0"
+1 -1
View File
@@ -116,7 +116,7 @@ chown -R 10001:10001 data logs
docker run -d -p 9000:9000 -p 9001:9001 -v $(pwd)/data:/data -v $(pwd)/logs:/logs rustfs/rustfs:latest
# Using specific version
docker run -d -p 9000:9000 -p 9001:9001 -v $(pwd)/data:/data -v $(pwd)/logs:/logs rustfs/rustfs:1.0.0-rc.2
docker run -d -p 9000:9000 -p 9001:9001 -v $(pwd)/data:/data -v $(pwd)/logs:/logs rustfs/rustfs:1.0.0-rc.3
```
If you use [podman](https://github.com/containers/podman) instead of docker, you can install the RustFS with the below command
+1 -1
View File
@@ -113,7 +113,7 @@ chown -R 10001:10001 data logs
docker run -d -p 9000:9000 -p 9001:9001 -v $(pwd)/data:/data -v $(pwd)/logs:/logs rustfs/rustfs:latest
# 使用指定版本运行
docker run -d -p 9000:9000 -p 9001:9001 -v $(pwd)/data:/data -v $(pwd)/logs:/logs rustfs/rustfs:1.0.0-rc.2
docker run -d -p 9000:9000 -p 9001:9001 -v $(pwd)/data:/data -v $(pwd)/logs:/logs rustfs/rustfs:1.0.0-rc.3
```
如果您通过绑定挂载启用 TLS 证书目录,也请用同样方式准备该目录:
-1
View File
@@ -11,7 +11,6 @@
// 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.
#![allow(dead_code)]
use base64_simd::STANDARD;
+4 -1
View File
@@ -38,7 +38,10 @@ pub const XXHASH_3_HEADER_NAME: &str = "x-amz-checksum-xxhash3";
pub const XXHASH_64_HEADER_NAME: &str = "x-amz-checksum-xxhash64";
pub const XXHASH_128_HEADER_NAME: &str = "x-amz-checksum-xxhash128";
#[allow(dead_code)]
#[allow(
dead_code,
reason = "Content-MD5 wire name, resolved by header_name() below and asserted by this crate's tests (backlog#1823)"
)]
pub(crate) static MD5_HEADER_NAME: &str = "content-md5";
pub const CHECKSUM_ALGORITHMS_IN_PRIORITY_ORDER: [&str; 5] =
+8 -2
View File
@@ -476,13 +476,19 @@ impl Checksum for Xxhash64 {
}
}
#[allow(dead_code)]
#[derive(Debug, Default)]
#[allow(
dead_code,
reason = "Content-MD5 is not a ChecksumAlgorithm variant and has no arm in into_impl: S3 carries it as its own header, separate from the x-amz-checksum-* family. This impl exists so the two paths share the Checksum trait, and is asserted by this crate's tests (backlog#1823)"
)]
struct Md5 {
hasher: md5::Md5,
}
#[allow(dead_code)]
#[allow(
dead_code,
reason = "Content-MD5 is not a ChecksumAlgorithm variant and has no arm in into_impl: S3 carries it as its own header, separate from the x-amz-checksum-* family. This impl exists so the two paths share the Checksum trait, and is asserted by this crate's tests (backlog#1823)"
)]
impl Md5 {
fn update(&mut self, bytes: &[u8]) {
use md5::Digest;
@@ -82,7 +82,6 @@ use tracing::warn;
use url::Url;
use uuid::Uuid;
const DEFAULT_HEALTH_CHECK_RELOAD_DURATION: Duration = Duration::from_secs(30 * 60);
const MAX_CONCURRENT_TARGET_HEALTH_CHECKS: usize = 16;
const REDACTED_CREDENTIAL: &str = "<redacted>";
@@ -126,11 +126,23 @@ const EVENT_LIFECYCLE_EXPIRED_DETECTED: &str = "lifecycle_expired_detected";
const EVENT_LIFECYCLE_NOT_ENQUEUED: &str = "lifecycle_not_enqueued";
const EVENT_LIFECYCLE_DELETE_DISPATCHED: &str = "lifecycle_delete_dispatched";
const EVENT_LIFECYCLE_DELETE_COMPLETED: &str = "lifecycle_delete_completed";
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
const EVENT_LIFECYCLE_TIER_AUDIT: &str = "lifecycle_tier_audit";
const EVENT_LIFECYCLE_TIER_OPERATION_FAILED: &str = "lifecycle_tier_operation_failed";
const EVENT_LIFECYCLE_DELETE_FAILED: &str = "lifecycle_delete_failed";
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub type TimeFn = Arc<dyn Fn() -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + Sync + 'static>;
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub type TraceFn =
Arc<dyn Fn(String, HashMap<String, String>) -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + Sync + 'static>;
pub type ExpiryOpType = Box<dyn ExpiryOp + Send + Sync + 'static>;
@@ -140,9 +152,21 @@ static TIER_FREE_VERSION_RECOVERY_STARTED: OnceLock<()> = OnceLock::new();
static MANUAL_TRANSITION_JOB_RECOVERY_STARTED: OnceLock<()> = OnceLock::new();
pub const AMZ_OBJECT_TAGGING: &str = "X-Amz-Tagging";
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub const AMZ_TAG_COUNT: &str = "x-amz-tagging-count";
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub const AMZ_TAG_DIRECTIVE: &str = "X-Amz-Tagging-Directive";
pub const AMZ_ENCRYPTION_AES: &str = "AES256";
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub const AMZ_ENCRYPTION_KMS: &str = "aws:kms";
pub const ERR_INVALID_STORAGECLASS: &str = "invalid tier.";
@@ -280,6 +304,10 @@ impl LifecycleSys {
}
}
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub fn trace(oi: &ObjectInfo) -> TraceFn {
let bucket = oi.bucket.clone();
let name = oi.name.clone();
@@ -570,6 +598,10 @@ async fn delete_free_version_remote_object(
Ok(())
}
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
async fn delete_free_version_remote_object_then<T, F, Fut>(
oi: &ObjectInfo,
tier_config_mgr: &Arc<RwLock<TierConfigMgr>>,
@@ -2868,6 +2900,10 @@ fn stale_upload_default_due(initiated: OffsetDateTime, default_expiry: StdDurati
initiated + time::Duration::seconds(default_expiry.as_secs() as i64)
}
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
async fn stale_upload_current_size(set: &Arc<SetDisks>, metadata: &HashMap<String, String>, upload_dir: &str) -> Option<usize> {
stale_upload_current_size_with_opts(set, metadata, upload_dir, false).await
}
@@ -3352,6 +3388,10 @@ pub async fn validate_transition_tier(lc: &BucketLifecycleConfiguration) -> Resu
Ok(())
}
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
fn mark_delete_opts_skip_decommissioned_on_remote_success(opts: &mut ObjectOptions, remote_delete_succeeded: bool) {
if remote_delete_succeeded {
opts.skip_decommissioned = true;
@@ -4339,6 +4379,10 @@ pub async fn expire_transitioned_object(
Ok(dobj)
}
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub fn gen_transition_objname(bucket: &str) -> Result<String, Error> {
let us = Uuid::new_v4().to_string();
let mut hasher = Sha256::new();
@@ -4373,6 +4417,10 @@ pub async fn transition_object(api: Arc<ECStore>, oi: &ObjectInfo, lae: LcAuditE
result
}
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub fn audit_tier_actions(_tier: &str, bytes: i64) -> TimeFn {
let tier = _tier.to_string();
Arc::new(move || {
@@ -4391,6 +4439,10 @@ pub fn audit_tier_actions(_tier: &str, bytes: i64) -> TimeFn {
})
}
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub async fn get_transitioned_object_reader(
bucket: &str,
object: &str,
@@ -5145,6 +5197,10 @@ async fn lifecycle_delete_config_snapshot(api: &ECStore, oi: &ObjectInfo) -> Res
ReplicationObjectBridge::delete_request_config(api, &oi.bucket).await
}
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub async fn apply_lifecycle_action(event: &lifecycle::Event, src: &LcEventSrc, oi: &ObjectInfo) -> bool {
let mut success = false;
match event.action {
@@ -7422,6 +7478,10 @@ mod tests {
// process environment while `env::set_var`/`env::remove_var` is active.
// SAFETY: keep this note adjacent to the allowance for the repository guard.
#[allow(unsafe_code)]
#[allow(
dead_code,
reason = "transition-queue env fixture kept for tests that scope those vars; no test uses it today (backlog#1823)"
)]
async fn with_transition_queue_env_async<F, Fut>(capacity: Option<&str>, timeout_ms: Option<&str>, test_fn: F)
where
F: FnOnce() -> Fut,
@@ -759,6 +759,10 @@ pub struct ManualTransitionWorkerResultRecord {
}
impl ManualTransitionWorkerResultRecord {
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub fn new(job_id: Uuid, task_key: impl Into<String>, result: ManualTransitionWorkerResult) -> Self {
Self::new_with_reason(job_id, task_key, result, None)
}
@@ -1257,6 +1261,10 @@ pub(crate) async fn save_manual_transition_task_if_absent(
}
}
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub async fn load_manual_transition_task_record(
api: Arc<ECStore>,
job_id: Uuid,
@@ -1320,6 +1328,10 @@ async fn scan_manual_transition_task_journal(api: Arc<ECStore>, job_id: Uuid) ->
}
}
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub async fn load_manual_transition_worker_result_stats(
api: Arc<ECStore>,
job_id: Uuid,
@@ -1455,6 +1467,10 @@ async fn scan_manual_transition_worker_result_journal(
}
}
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub async fn reconcile_manual_transition_worker_results(
api: Arc<ECStore>,
job_id: Uuid,
@@ -15,25 +15,35 @@
use rustfs_common::metrics::IlmAction;
use crate::bucket::lifecycle::lifecycle::ObjectOpts;
use crate::bucket::replication::ReplicationLifecycleBridge;
pub(crate) use crate::bucket::replication::ReplicationStatusType;
#[cfg(test)]
pub(crate) use crate::bucket::replication::VersionPurgeStatusType;
pub(crate) use crate::bucket::replication::{
DeleteReplicationConfigSnapshot, ReplicationObjectBridge, replication_state_to_filemeta,
};
use crate::bucket::replication::{ReplicationLifecycleBridge, ReplicationLifecycleConfig};
use crate::storage_api_contracts::object::DeletedObject;
pub(crate) type LifecycleReplicationConfig = ReplicationLifecycleConfig;
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) fn has_pending_version_purge(obj: &ObjectOpts) -> bool {
obj.version_purge_status.is_pending()
}
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) fn has_pending_object_replication(obj: &ObjectOpts) -> bool {
replication_status_blocks_lifecycle(&obj.replication_status)
}
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) fn has_pending_lifecycle_replication(obj: &ObjectOpts) -> bool {
has_pending_object_replication(obj) || has_pending_version_purge(obj)
}
@@ -14,6 +14,10 @@
use std::collections::HashMap;
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) fn decode_tags_to_map(tags: &str) -> HashMap<String, String> {
crate::bucket::tagging::decode_tags_to_map(tags)
}
@@ -331,6 +331,10 @@ where
persist_tier_delete_journal_entry(api, &committed).await
}
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub async fn abort_tier_delete_journal_entry<S>(api: Arc<S>, je: &Jentry) -> std::io::Result<()>
where
S: ObjectOperations<
@@ -148,6 +148,10 @@ struct RecoveryCursor {
object: String,
}
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub async fn recover_tier_free_versions(
api: Arc<ECStore>,
limit: usize,
@@ -385,6 +385,10 @@ impl ExpiryOp for Jentry {
}
}
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub async fn delete_object_from_remote_tier(obj_name: &str, rv_id: &str, tier_name: &str) -> Result<(), std::io::Error> {
let result = delete_object_from_remote_tier_raw(obj_name, rv_id, tier_name).await;
if let Err(err) = &result
@@ -395,6 +399,10 @@ pub async fn delete_object_from_remote_tier(obj_name: &str, rv_id: &str, tier_na
result
}
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
async fn delete_object_from_remote_tier_raw(obj_name: &str, rv_id: &str, tier_name: &str) -> Result<(), std::io::Error> {
#[cfg(test)]
if let Some(result) = run_remote_tier_delete_test_hook(obj_name, rv_id, tier_name) {
@@ -405,6 +413,10 @@ async fn delete_object_from_remote_tier_raw(obj_name: &str, rv_id: &str, tier_na
delete_object_from_remote_tier_raw_with_manager(obj_name, rv_id, tier_name, &tier_config_mgr).await
}
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
async fn delete_object_from_remote_tier_raw_with_manager(
obj_name: &str,
rv_id: &str,
@@ -485,6 +497,10 @@ pub enum RemoteTierDeleteOutcome {
AlreadyRemoved,
}
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
pub async fn delete_object_from_remote_tier_idempotent(
obj_name: &str,
rv_id: &str,
@@ -50,8 +50,16 @@ pub type Result<T> = std::result::Result<T, TransitionTransactionError>;
#[derive(Debug, thiserror::Error)]
pub enum TransitionTransactionError {
#[error("transition transaction already exists")]
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
AlreadyExists,
#[error("transition transaction is not found")]
#[allow(
dead_code,
reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
)]
NotFound,
#[error("transition transaction is corrupt: {0}")]
Corrupt(&'static str),
+13
View File
@@ -60,12 +60,14 @@ struct ConfigWriteLockProbeState {
static CONFIG_WRITE_LOCK_PROBES: std::sync::OnceLock<StdMutex<Vec<Arc<ConfigWriteLockProbeState>>>> = std::sync::OnceLock::new();
#[cfg(any(test, feature = "test-util"))]
#[allow(dead_code, reason = "installed by tests behind `--features test-util` (backlog#1823)")]
pub struct ConfigWriteLockProbe {
state: Arc<ConfigWriteLockProbeState>,
}
#[cfg(any(test, feature = "test-util"))]
impl ConfigWriteLockProbe {
#[allow(dead_code, reason = "installed by tests behind `--features test-util` (backlog#1823)")]
pub fn install(bucket: &str) -> Self {
let state = Arc::new(ConfigWriteLockProbeState {
bucket: bucket.to_string(),
@@ -84,6 +86,7 @@ impl ConfigWriteLockProbe {
Self { state }
}
#[allow(dead_code, reason = "installed by tests behind `--features test-util` (backlog#1823)")]
pub async fn wait_until_attempted(&self) {
tokio::time::timeout(Duration::from_secs(30), self.state.arrived.notified())
.await
@@ -890,6 +893,10 @@ pub async fn get_bucket_policy_raw(bucket: &str) -> Result<(String, OffsetDateTi
bucket_meta_sys.get_bucket_policy_raw(bucket).await
}
#[allow(
dead_code,
reason = "free-function facade over the live BucketMetadataSys::get_bucket_acl_config; no caller in this port (backlog#1823)"
)]
pub async fn get_bucket_acl_config(bucket: &str) -> Result<(String, OffsetDateTime)> {
let bucket_meta_sys_lock = get_bucket_metadata_sys()?;
let bucket_meta_sys = bucket_meta_sys_lock.read().await;
@@ -1104,6 +1111,10 @@ pub async fn get_config_from_disk(bucket: &str) -> Result<BucketMetadata> {
bucket_meta_sys.get_config_from_disk(bucket).await
}
#[allow(
dead_code,
reason = "ambient-facade variant of the live created_at_in; no caller in this port (backlog#1823)"
)]
pub async fn created_at(bucket: &str) -> Result<OffsetDateTime> {
let bucket_meta_sys_lock = get_bucket_metadata_sys()?;
let bucket_meta_sys = bucket_meta_sys_lock.read().await;
@@ -1617,6 +1628,7 @@ impl BucketMetadataSys {
/// [`Self::update`], with the payload computed from the loaded metadata
/// instead of supplied up front. Loads through this system's own store so
/// the read and the persisted write target the same instance.
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
async fn update_config_with<F>(&self, bucket: &str, config_file: &str, mutate: F) -> Result<OffsetDateTime>
where
F: FnOnce(&BucketMetadata) -> Result<Vec<u8>> + Send,
@@ -1721,6 +1733,7 @@ impl BucketMetadataSys {
/// A miss is never published as an authoritative default, and a snapshot
/// read before delete plus same-name recreation cannot replace the new
/// generation.
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub(crate) async fn reload_from_store(&self, bucket: &str) -> Result<()> {
if is_meta_bucketname(bucket) {
return Err(Error::other("errInvalidArgument"));
-1
View File
@@ -13,7 +13,6 @@
// limitations under the License.
// #730: bucket subsystems still contain staged ECStore migration code.
#![allow(dead_code)]
pub mod bandwidth;
pub mod bucket_target_sys;
@@ -136,6 +136,7 @@ pub fn add_years(dt: OffsetDateTime, years: i32) -> OffsetDateTime {
/// Check if an object has legal hold enabled.
/// Returns true if legal hold is ON.
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
fn has_legal_hold(user_defined: &std::collections::HashMap<String, String>) -> bool {
let lhold = objectlock::get_object_legalhold_meta(user_defined);
matches!(lhold.status, Some(ref st) if st.as_str() == ObjectLockLegalHoldStatus::ON)
@@ -151,6 +152,7 @@ fn has_legal_hold(user_defined: &std::collections::HashMap<String, String>) -> b
/// # Returns
/// * `true` if the object is locked (cannot be deleted/modified)
/// * `false` if the object is not locked
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub fn is_object_locked_by_metadata(user_defined: &std::collections::HashMap<String, String>, is_delete_marker: bool) -> bool {
// Delete markers are never locked
if is_delete_marker {
+2
View File
@@ -193,6 +193,7 @@ pub enum QuotaError {
}
#[derive(Debug, Serialize)]
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub struct QuotaErrorResponse {
#[serde(rename = "Code")]
pub code: String,
@@ -208,6 +209,7 @@ pub struct QuotaErrorResponse {
}
impl QuotaErrorResponse {
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub fn new(quota_error: &QuotaError, request_id: &str, host_id: &str) -> Self {
match quota_error {
QuotaError::QuotaExceeded { .. } => Self {
@@ -899,6 +899,7 @@ async fn save_ledger_locked(
}
#[cfg(any(test, feature = "test-util"))]
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub fn fail_next_quota_ledger_save_for_test() {
FAIL_NEXT_LEDGER_SAVE.store(true, std::sync::atomic::Ordering::SeqCst);
}
+1 -1
View File
@@ -60,7 +60,7 @@ pub use replication_filemeta_boundary::{
pub(crate) use replication_filemeta_boundary::{
replication_state_from_filemeta, replication_status_from_filemeta, version_purge_status_from_filemeta,
};
pub(crate) use replication_lifecycle_bridge::{ReplicationLifecycleBridge, ReplicationLifecycleConfig};
pub(crate) use replication_lifecycle_bridge::ReplicationLifecycleBridge;
pub(crate) use replication_migration_bridge::ReplicationMigrationBridge;
pub use replication_object_bridge::ReplicationObjectBridge;
pub use replication_object_config::{DeleteReplicationConfigSnapshot, ReplicationConfig};
@@ -37,6 +37,10 @@ impl ReplicationConfigStore {
com::read_config_limited(api, file, max_bytes).await
}
#[allow(
dead_code,
reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
)]
pub(crate) async fn read_no_lock<S>(api: Arc<S>, file: &str) -> Result<Vec<u8>>
where
S: ReplicationObjectIO,
@@ -24,15 +24,27 @@ use super::replication_storage_boundary::{
DeletedObject, ObjectInfo, ObjectOptions, ObjectToDelete, deleted_object_for_replication,
};
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) type ReplicationLifecycleConfig = ReplicationConfig;
pub(crate) struct ReplicationLifecycleBridge;
impl ReplicationLifecycleBridge {
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) fn new_config(config: ReplicationConfiguration) -> ReplicationLifecycleConfig {
ReplicationConfig::new(Some(config), None)
}
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) fn has_pending_version_purge(
config: &ReplicationLifecycleConfig,
object_name: &str,
@@ -45,6 +57,10 @@ impl ReplicationLifecycleBridge {
.is_some_and(|config| config.has_active_rules(object_name, true))
}
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) async fn check_delete_replication(
bucket: &str,
object: &ObjectToDelete,
@@ -54,6 +70,10 @@ impl ReplicationLifecycleBridge {
check_replicate_delete(bucket, object, source, opts, None).await
}
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) fn version_delete_replication_state(decision: &ReplicateDecision) -> ReplicationState {
let pending_status = decision.pending_status();
ReplicationState {
@@ -19,17 +19,33 @@ use time::OffsetDateTime;
use super::replication_error_boundary::Result;
use crate::bucket::msgp_decode;
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) struct ReplicationMsgpCodec;
impl ReplicationMsgpCodec {
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) fn read_ext8_time<R: Read>(rd: &mut R) -> Result<OffsetDateTime> {
msgp_decode::read_msgp_ext8_time(rd)
}
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) fn skip_value<R: Read>(rd: &mut R) -> Result<()> {
msgp_decode::skip_msgp_value(rd)
}
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) fn write_time<W: Write>(wr: &mut W, time: OffsetDateTime) -> Result<()> {
msgp_decode::write_msgp_time(wr, time)
}
@@ -77,6 +77,10 @@ impl ReplicationObjectBridge {
load_delete_request_config_in(ctx, bucket).await
}
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) async fn delete_config_snapshot_in(
ctx: &ReplicationInstanceContext,
bucket: &str,
@@ -231,6 +231,10 @@ pub(crate) async fn load_delete_replication_config(
delete_snapshot_from_metadata(ReplicationMetadataStore::delete_metadata(bucket).await?)
}
#[allow(
dead_code,
reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
)]
pub(crate) async fn load_delete_replication_config_in(
ctx: &ReplicationInstanceContext,
bucket: &str,
@@ -217,6 +217,10 @@ impl DurableMrfBacklogTracker {
}
}
#[allow(
dead_code,
reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
)]
fn durable_mrf_backlog_tracker_from_entries(entries: &[MrfReplicateEntry]) -> DurableMrfBacklogTracker {
let mut tracker = DurableMrfBacklogTracker {
available: true,
@@ -712,6 +716,10 @@ pub struct ReplicationPool<S: ReplicationStorage> {
// MRF worker lifecycle
mrf_worker_cancellations: Mutex<Vec<CancellationToken>>,
#[allow(
dead_code,
reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
)]
mrf_stop_tx: Sender<()>,
// Worker size tracking
@@ -940,6 +948,10 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
}
/// Resizes worker priority and counts
#[allow(
dead_code,
reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
)]
pub async fn resize_worker_priority(
&self,
pri: ReplicationPriority,
@@ -1180,6 +1192,10 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
}
/// Queues an MRF save operation
#[allow(
dead_code,
reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
)]
async fn queue_mrf_save(&self, entry: MrfReplicateEntry) {
let _ = self.queue_mrf_save_admission(entry, "mrf_worker").await;
}
@@ -1651,6 +1667,10 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
}
/// Worker function for handling regular replication operations
#[allow(
dead_code,
reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
)]
async fn add_worker(
&self,
mut rx: Receiver<ReplicationOperation>,
@@ -1664,6 +1684,10 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
}
/// Worker function for handling large object replication operations
#[allow(
dead_code,
reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
)]
async fn add_large_worker(
&self,
mut rx: Receiver<ReplicationOperation>,
@@ -1678,6 +1702,10 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
}
/// Worker function for handling MRF (Most Recent Failures) operations
#[allow(
dead_code,
reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
)]
async fn add_mrf_worker(
&self,
mut rx: Receiver<ReplicationOperation>,
@@ -1691,6 +1719,10 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
}
/// Delete resync metadata from replication resync state in memory
#[allow(
dead_code,
reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
)]
pub async fn delete_resync_metadata(&self, bucket: &str) {
let mut status_map = self.resyncer.status_map.write().await;
status_map.remove(bucket);
@@ -21,11 +21,31 @@ pub(crate) use rustfs_replication::{
should_count_head_proxy_failure,
};
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) const RESYNC_META_FORMAT: u16 = rustfs_replication::resync::RESYNC_META_FORMAT;
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) const RESYNC_META_VERSION: u16 = rustfs_replication::resync::RESYNC_META_VERSION;
pub(crate) const RESYNC_FILE_MAX_BYTES: usize = rustfs_replication::RESYNC_FILE_MAX_BYTES;
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) const WIRE_ZERO_TIME_UNIX: i64 = rustfs_replication::resync::WIRE_ZERO_TIME_UNIX;
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) const MRF_META_FORMAT: u16 = rustfs_replication::mrf::MRF_META_FORMAT;
#[allow(
dead_code,
reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
)]
pub(crate) const MRF_META_VERSION: u16 = rustfs_replication::mrf::MRF_META_VERSION;
fn map_replication_error(err: rustfs_replication::Error) -> Error {
@@ -122,6 +122,10 @@ const REPLICATION_TARGET_OFFLINE_ERROR_MARKERS: &[&str] = &[
"tcp connect error",
];
#[allow(
dead_code,
reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
)]
const RESYNC_TIME_INTERVAL: TokioDuration = TokioDuration::from_secs(60);
static WARNED_MONITOR_UNINIT: std::sync::Once = std::sync::Once::new();
@@ -328,6 +332,10 @@ fn bounded_resync_max_jobs(value: usize) -> usize {
#[derive(Debug)]
pub struct ReplicationResyncer {
pub status_map: Arc<RwLock<HashMap<String, BucketReplicationResyncStatus>>>,
#[allow(
dead_code,
reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
)]
pub worker_size: usize,
pub(crate) cancel_tokens: Arc<RwLock<HashMap<ResyncCancelKey, CancellationToken>>>,
resync_admission: Arc<Semaphore>,
@@ -544,6 +552,10 @@ impl ReplicationResyncer {
.is_some_and(|status| status.failed_count > 0)
}
#[allow(
dead_code,
reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
)]
pub async fn persist_to_disk<S>(&self, cancel_token: CancellationToken, api: Arc<S>)
where
S: ReplicationObjectIO,
@@ -340,6 +340,10 @@ impl ReplicationStats {
}
/// Site replication update replica statistics
#[allow(
dead_code,
reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
)]
fn sr_update_replica_stat(&self, size: i64) {
self.sr_stats.replica_size.fetch_add(size, Ordering::Relaxed);
self.sr_stats.replica_count.fetch_add(1, Ordering::Relaxed);
@@ -59,6 +59,10 @@ impl fmt::Debug for Credentials {
}
#[derive(Debug, Deserialize, Serialize, Default, Clone)]
#[allow(
dead_code,
reason = "MinIO-parity bucket-target service discriminator with no caller in this port (backlog#1823)"
)]
pub enum ServiceType {
#[default]
Replication,
+20 -17
View File
@@ -73,23 +73,6 @@ pub fn check_valid_bucket_name_strict(bucket_name: &str) -> Result<()> {
check_bucket_name_common(bucket_name, true)
}
pub fn check_valid_object_name_prefix(object_name: &str) -> Result<()> {
if object_name.len() > 1024 {
return Err(Error::other("Object name cannot be longer than 1024 characters"));
}
if !object_name.is_ascii() {
return Err(Error::other("Object name with non-UTF-8 strings are not supported"));
}
Ok(())
}
pub fn check_valid_object_name(object_name: &str) -> Result<()> {
if object_name.trim().is_empty() {
return Err(Error::other("Object name cannot be empty"));
}
check_valid_object_name_prefix(object_name)
}
pub fn deserialize<T>(input: &[u8]) -> xml::DeResult<T>
where
T: for<'xml> xml::Deserialize<'xml>,
@@ -100,6 +83,10 @@ where
Ok(ans)
}
#[allow(
dead_code,
reason = "xml serialize helper with no caller in this port; the live sibling is deserialize (backlog#1823)"
)]
pub fn serialize_content<T: xml::SerializeContent>(val: &T) -> xml::SerResult<String> {
let mut buf = Vec::with_capacity(256);
{
@@ -186,15 +173,27 @@ pub fn is_valid_object_name(object: &str) -> bool {
/// Client-facing reason attached to rejections of object keys that Win32/NTFS
/// cannot represent as file paths (issue #3299). Deployments on Linux/macOS
/// accept the full S3 key character set.
#[allow(
dead_code,
reason = "live on Windows: callers sit inside the #[cfg(target_os = \"windows\")] block in check_object_name_for_length_and_slash (backlog#1823)"
)]
pub const WINDOWS_RESERVED_CHARACTERS_REASON: &str =
"object key contains characters unsupported on Windows hosts (one of ':', '*', '?', '\"', '|', '<', '>')";
/// Client-facing reason for path segments Windows can store but not address
/// afterwards (issue #3449): trailing dot/space or reserved DOS device names.
#[allow(
dead_code,
reason = "live on Windows: callers sit inside the #[cfg(target_os = \"windows\")] block in check_object_name_for_length_and_slash (backlog#1823)"
)]
pub const WINDOWS_RESERVED_SEGMENT_REASON: &str = "object key contains a path segment unsupported on Windows hosts (trailing dot or space, or a reserved device name such as NUL/CON/COM1)";
/// Reserved DOS device names that shadow regular files on Windows, even when
/// an extension is appended (e.g. `NUL.txt` resolves to the `NUL` device).
#[allow(
dead_code,
reason = "live on Windows: callers sit inside the #[cfg(target_os = \"windows\")] block in check_object_name_for_length_and_slash (backlog#1823)"
)]
const WINDOWS_RESERVED_NAMES: &[&str] = &[
"CON", "PRN", "AUX", "NUL", "COM1", "COM2", "COM3", "COM4", "COM5", "COM6", "COM7", "COM8", "COM9", "LPT1", "LPT2", "LPT3",
"LPT4", "LPT5", "LPT6", "LPT7", "LPT8", "LPT9",
@@ -204,6 +203,10 @@ const WINDOWS_RESERVED_NAMES: &[&str] = &[
/// the Win32 API cannot address afterwards (issue #3449): segments ending in a
/// dot or a space, and reserved DOS device names — bare or with an extension
/// (`NUL.txt`), matching classic Win32 path resolution semantics.
#[allow(
dead_code,
reason = "live on Windows: callers sit inside the #[cfg(target_os = \"windows\")] block in check_object_name_for_length_and_slash (backlog#1823)"
)]
pub fn object_name_has_windows_incompatible_segment(object: &str) -> bool {
object.split(['/', '\\']).any(|segment| {
if segment.ends_with('.') || segment.ends_with(' ') {
@@ -90,6 +90,10 @@ impl BucketVersioningSys {
/// caller's own instance context so a second in-process store never
/// answers with the first instance's versioning state; falls back to the
/// ambient system when the instance cell is not initialized.
#[allow(
dead_code,
reason = "instance-scoped seam (backlog#1052) with no caller in this port (backlog#1823)"
)]
pub(crate) async fn get_in(ctx: &crate::runtime::instance::InstanceContext, bucket: &str) -> Result<VersioningConfiguration> {
if bucket == RUSTFS_META_BUCKET || bucket.starts_with(RUSTFS_META_BUCKET) {
return Ok(VersioningConfiguration::default());
+13 -33
View File
@@ -637,14 +637,23 @@ impl Default for DiskOperationMetrics {
}
impl DiskOperationMetrics {
#[allow(
dead_code,
reason = "internal metrics recorder reached only from record() below (backlog#1823)"
)]
fn record_call(&mut self) {
self.lifetime_calls.fetch_add(1, Ordering::Relaxed);
}
#[allow(
dead_code,
reason = "internal metrics recorder reached only from record() below (backlog#1823)"
)]
fn record_latency(&mut self, now_sec: u64, elapsed: Duration) {
self.record_latency_atomic(now_sec, elapsed);
}
#[allow(dead_code, reason = "metrics roll-up with no caller in this port (backlog#1823)")]
fn record(&mut self, now_sec: u64, elapsed: Duration) {
self.record_call();
self.record_latency(now_sec, elapsed);
@@ -770,6 +779,7 @@ impl DiskHealthTracker {
}
/// Set disk as faulty
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub fn set_faulty(&self) {
self.status.store(DISK_HEALTH_FAULTY, Ordering::Release);
}
@@ -850,6 +860,7 @@ impl DiskHealthTracker {
became_offline
}
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub fn mark_offline(&self, endpoint: &Endpoint, reason: &'static str) -> bool {
let current = self.runtime_state();
if current == RuntimeDriveHealthState::Offline {
@@ -980,11 +991,13 @@ impl DiskHealthTracker {
}
/// Get waiting operations count
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub fn waiting_count(&self) -> u32 {
self.waiting.load(Ordering::Relaxed)
}
/// Get last success timestamp
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub fn last_success(&self) -> i64 {
self.last_success.load(Ordering::Acquire)
}
@@ -1026,21 +1039,6 @@ impl Default for DiskHealthTracker {
}
}
/// Health check context key for tracking disk operations
#[derive(Debug, Clone)]
struct HealthDiskCtxKey;
#[derive(Debug)]
struct HealthDiskCtxValue {
last_success: Arc<AtomicI64>,
}
impl HealthDiskCtxValue {
fn log_success(&self) {
self.last_success.store(current_unix_nanos(), Ordering::Relaxed);
}
}
/// LocalDiskWrapper wraps a DiskStore with health tracking capabilities.
/// This is similar to Go's xlStorageDiskIDCheck.
#[derive(Debug, Clone)]
@@ -1072,10 +1070,6 @@ impl LocalDiskWrapper {
)
}
pub(crate) fn new_with_health(disk: Arc<LocalDisk>, health_check: bool, health: Arc<DiskHealthTracker>) -> Self {
Self::new_with_health_and_metrics(disk, health_check, health, Arc::new(DiskHealthMetricEpoch::default()))
}
pub(crate) fn new_with_reconnect_state(
disk: Arc<LocalDisk>,
health_check: bool,
@@ -1438,20 +1432,6 @@ impl LocalDiskWrapper {
}
}
async fn check_id(&self, want_id: Option<Uuid>) -> Result<()> {
if want_id.is_none() {
return Ok(());
}
let stored_disk_id = self.disk.get_disk_id().await?;
if stored_disk_id != want_id {
return Err(Error::other(format!("Disk ID mismatch wanted {want_id:?}, got {stored_disk_id:?}")));
}
Ok(())
}
/// Check if disk ID is stale
async fn check_disk_stale(&self) -> Result<()> {
let Some(current_disk_id) = *self.disk_id.read().await else {
+1
View File
@@ -48,6 +48,7 @@ pub fn to_volume_error(io_err: std::io::Error) -> std::io::Error {
}
}
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub fn to_disk_error(io_err: std::io::Error) -> std::io::Error {
match io_err.kind() {
std::io::ErrorKind::NotFound => DiskError::DiskNotFound.into(),
+1
View File
@@ -178,6 +178,7 @@ pub async fn remove(path: impl AsRef<Path>) -> io::Result<()> {
}
}
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub async fn remove_all(path: impl AsRef<Path>) -> io::Result<()> {
// Try remove_file first; fall back to remove_dir_all if it's a directory
match fs::remove_file(path.as_ref()).await {
+69
View File
@@ -665,6 +665,7 @@ async fn remove_empty_directory_tree_under_mount_lease(
}
#[cfg(unix)]
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
async fn remove_empty_directory_tree_with(
root: &Path,
before_descend: impl FnMut(&Path) -> std::io::Result<()>,
@@ -1016,13 +1017,29 @@ fn record_direct_read_page_fault_delta(path: &'static str, stage: &'static str,
/// When enabled, shard reads bypass the page cache using O_DIRECT flag.
/// Requires aligned buffers (typically 512 bytes or 4096 bytes).
/// Default: false (uses page cache via mmap/pread).
#[allow(
dead_code,
reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
)]
const ENV_RUSTFS_OBJECT_DIRECT_IO_READ_ENABLE: &str = "RUSTFS_OBJECT_DIRECT_IO_READ_ENABLE";
#[allow(
dead_code,
reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
)]
const DEFAULT_RUSTFS_OBJECT_DIRECT_IO_READ_ENABLE: bool = false;
/// Minimum shard size threshold for O_DIRECT reads.
/// Only shards larger than this threshold will use O_DIRECT.
/// Default: 4MB.
#[allow(
dead_code,
reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
)]
const ENV_RUSTFS_OBJECT_DIRECT_IO_READ_THRESHOLD: &str = "RUSTFS_OBJECT_DIRECT_IO_READ_THRESHOLD";
#[allow(
dead_code,
reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
)]
const DEFAULT_RUSTFS_OBJECT_DIRECT_IO_READ_THRESHOLD: usize = 4 * 1024 * 1024;
/// Enable O_DIRECT for erasure shard / multipart part data writes (Linux only).
@@ -1036,7 +1053,15 @@ const DEFAULT_RUSTFS_OBJECT_DIRECT_IO_READ_THRESHOLD: usize = 4 * 1024 * 1024;
/// EINVAL/EOPNOTSUPP (tmpfs, overlayfs, 9p, ...) latch the path off and fall
/// back to buffered writes for the whole disk. Non-Linux always falls back.
/// Default: false (buffered writes via the page cache, as before).
#[allow(
dead_code,
reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
)]
const ENV_RUSTFS_OBJECT_DIRECT_IO_WRITE_ENABLE: &str = "RUSTFS_OBJECT_DIRECT_IO_WRITE_ENABLE";
#[allow(
dead_code,
reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
)]
const DEFAULT_RUSTFS_OBJECT_DIRECT_IO_WRITE_ENABLE: bool = false;
const ENV_RUSTFS_OBJECT_MMAP_POPULATE_ENABLE: &str = "RUSTFS_OBJECT_MMAP_POPULATE_ENABLE";
const DEFAULT_RUSTFS_OBJECT_MMAP_POPULATE_ENABLE: bool = false;
@@ -1095,12 +1120,14 @@ macro_rules! cached_read_env {
cached_read_env! {
/// Check if O_DIRECT reads are enabled.
#[allow(dead_code, reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)")]
fn is_direct_io_read_enabled() -> bool =
rustfs_utils::get_env_bool(ENV_RUSTFS_OBJECT_DIRECT_IO_READ_ENABLE, DEFAULT_RUSTFS_OBJECT_DIRECT_IO_READ_ENABLE);
}
cached_read_env! {
/// Check if O_DIRECT shard/part data writes are enabled.
#[allow(dead_code, reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)")]
fn is_direct_io_write_enabled() -> bool =
rustfs_utils::get_env_bool(ENV_RUSTFS_OBJECT_DIRECT_IO_WRITE_ENABLE, DEFAULT_RUSTFS_OBJECT_DIRECT_IO_WRITE_ENABLE);
}
@@ -1456,6 +1483,7 @@ pub(crate) fn effective_durability(volume: &str) -> DurabilityMode {
cached_read_env! {
/// Get the O_DIRECT read threshold size.
#[allow(dead_code, reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)")]
fn get_direct_io_read_threshold() -> usize =
rustfs_utils::get_env_usize(ENV_RUSTFS_OBJECT_DIRECT_IO_READ_THRESHOLD, DEFAULT_RUSTFS_OBJECT_DIRECT_IO_READ_THRESHOLD);
}
@@ -1673,12 +1701,20 @@ impl DirectIoWriteState {
/// Target staging size for O_DIRECT writes, rounded up to the DIO alignment.
/// Bounds the per-writer aligned bounce buffer and batches many shard blocks
/// into one positioned write to keep the syscall count low.
#[allow(
dead_code,
reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
)]
const DIRECT_WRITE_STAGING_BYTES: usize = 1024 * 1024;
/// Aligned bounce-buffer capacity for a given DIO alignment: the target staging
/// size rounded up to a whole multiple of `align` so the buffer address, every
/// flushed batch length, and every write offset stay alignment-correct.
/// Platform-independent (no O_DIRECT), so it is unit-tested on any host.
#[allow(
dead_code,
reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
)]
fn direct_write_staging_capacity(align: usize) -> usize {
debug_assert!(align.is_power_of_two() && align >= 512);
DIRECT_WRITE_STAGING_BYTES.div_ceil(align) * align
@@ -1687,6 +1723,10 @@ fn direct_write_staging_capacity(align: usize) -> usize {
/// Split `filled` staged bytes into the alignment-sized prefix written with
/// O_DIRECT and the sub-alignment tail written buffered. Platform-independent,
/// so the tail-boundary math is unit-tested on any host.
#[allow(
dead_code,
reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
)]
fn direct_write_tail_split(filled: usize, align: usize) -> (usize, usize) {
let aligned = filled - (filled % align);
(aligned, filled - aligned)
@@ -2142,6 +2182,7 @@ fn set_delete_version_fail_after_data_staged(path: &str) {
}
#[cfg(test)]
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub(crate) fn set_delete_version_fail_after_commit(root: &Path, path: &str) {
DELETE_VERSION_FAIL_AFTER_COMMIT
.lock()
@@ -2447,6 +2488,10 @@ enum SyncMode {
FileOnly,
}
#[allow(
dead_code,
reason = "reclaim bookkeeping fields written by Drop but never read back (backlog#1823)"
)]
struct FileCacheReclaimWriter {
inner: File,
reclaim_len: usize,
@@ -2454,6 +2499,10 @@ struct FileCacheReclaimWriter {
reclaimed: bool,
}
#[allow(
dead_code,
reason = "reclaim bookkeeping fields written by Drop but never read back (backlog#1823)"
)]
struct FileCacheReclaimReader {
inner: File,
reclaim_offset: u64,
@@ -2519,6 +2568,10 @@ impl<R: AsyncRead + Unpin> AsyncRead for StallTimeoutReader<R> {
}
}
#[allow(
dead_code,
reason = "reclaim metrics emitter reached only from the Linux-gated reclaim paths (backlog#1823)"
)]
fn record_file_cache_reclaim_success(kind: &'static str, reclaim_len: usize, started: std::time::Instant) {
// Runs per read-stream page-cache reclaim window; skip the whole emission
// (three metric-key constructions) when general metrics are disabled.
@@ -3071,6 +3124,7 @@ impl LocalIoBackend for StdBackend {
use memmap2::MmapOptions;
use std::time::{Duration as StdDuration, Instant as StdInstant};
#[allow(dead_code, reason = "mmap copy result slot kept beside the mapping it owns (backlog#1823)")]
struct MmapCopyReadResult {
bytes: Bytes,
access_check_duration: StdDuration,
@@ -4704,6 +4758,10 @@ fn build_local_io_backend(root: PathBuf) -> Arc<dyn LocalIoBackend> {
Arc::new(StdBackend::new(root))
}
#[allow(
dead_code,
reason = "path cache and cwd slots retained beside the disk root they derive from (backlog#1823)"
)]
pub struct LocalDisk {
pub root: PathBuf,
publication_root: os::PublicationRoot,
@@ -5490,6 +5548,7 @@ impl LocalDisk {
Ok(Self::resolve_abs_path_from(&self.root, path.as_ref()))
}
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
fn io_resolve_abs_path(&self, path: impl AsRef<Path>) -> PathBuf {
let path_ref = path.as_ref();
let path_str = path_ref.to_string_lossy();
@@ -5567,15 +5626,24 @@ impl LocalDisk {
}
// Check if a path is valid
#[allow(
dead_code,
reason = "method wrapper over the live free function check_local_disk_valid_path; no caller in this port (backlog#1823)"
)]
fn check_valid_path<P: AsRef<Path>>(&self, path: P) -> Result<()> {
check_local_disk_valid_path(self.io_root(), path)
}
#[allow(
dead_code,
reason = "method wrapper over the live free function reject_local_disk_symlink_components; no caller in this port (backlog#1823)"
)]
fn reject_symlink_components(&self, path: &Path) -> Result<()> {
reject_local_disk_symlink_components(self.io_root(), path)
}
// Batch path generation with single lock acquisition
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
fn get_object_paths_batch(&self, requests: &[(String, String)]) -> Result<Vec<PathBuf>> {
let mut results = Vec::with_capacity(requests.len());
let mut cache_misses = Vec::new();
@@ -6488,6 +6556,7 @@ impl LocalDisk {
Ok(f)
}
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
async fn open_file_read_only(&self, path: impl AsRef<Path>) -> Result<File> {
let f = super::fs::open_file(path.as_ref(), O_RDONLY).await.map_err(to_file_error)?;
Ok(f)
+5 -1
View File
@@ -13,7 +13,6 @@
// limitations under the License.
// #730: disk abstractions still carry staged health and direct-I/O migration paths.
#![allow(dead_code)]
pub mod disk_store;
pub mod endpoint;
@@ -1114,6 +1113,10 @@ pub struct DiskInfo {
}
#[derive(Clone, Debug, Default)]
#[allow(
dead_code,
reason = "MinIO-parity disk info shape with no constructor in this port (backlog#1823)"
)]
pub struct Info {
pub total: u64,
pub free: u64,
@@ -1372,6 +1375,7 @@ pub fn conv_part_err_to_int(err: &Option<Error>) -> usize {
}
}
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub fn has_part_err(part_errs: &[usize]) -> bool {
part_errs.iter().any(|err| *err != CHECK_PART_SUCCESS)
}
+5 -11
View File
@@ -571,6 +571,10 @@ fn regular_files(dir: &Path) -> io::Result<Vec<PathBuf>> {
/// Fdatasync every regular file directly inside `dir`, then fsync the directory
/// itself.
#[allow(
dead_code,
reason = "reached only through sync_dir_files, whose callers are tests (backlog#1823)"
)]
pub fn sync_dir_files_std(dir: impl AsRef<Path>) -> io::Result<()> {
for entry in std::fs::read_dir(dir.as_ref())? {
let entry = entry?;
@@ -583,6 +587,7 @@ pub fn sync_dir_files_std(dir: impl AsRef<Path>) -> io::Result<()> {
/// Async wrapper around [`sync_dir_files_std`]. Large directories flush files
/// concurrently, bounded both per directory and process-wide.
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub async fn sync_dir_files(dir: impl AsRef<Path>) -> io::Result<()> {
sync_dir_files_with_limiter(dir, Arc::new(Semaphore::new(MAX_PARALLEL_FILE_SYNCS))).await
}
@@ -1809,10 +1814,6 @@ impl RenameCommitGuard {
})
}
pub(crate) fn lock_destination_directory_for_path_access(&self, directory: &Path) -> io::Result<RenameDestinationPathGuard> {
self.destination_directory_guard(directory, false)
}
pub(crate) fn create_destination_directory_for_path_access(
&self,
directory: &Path,
@@ -2858,13 +2859,6 @@ pub async fn os_mkdir_all(dir_path: impl AsRef<Path>, base_dir: impl AsRef<Path>
Ok(())
}
/// Check if a file exists.
/// Returns true if the file exists, false otherwise.
#[tracing::instrument(level = "debug", skip_all)]
pub fn file_exists(path: impl AsRef<Path>) -> bool {
std::fs::metadata(path.as_ref()).map(|_| true).unwrap_or(false)
}
/// Whether an [`io::Error`] means "the directory is not empty".
///
/// POSIX lets `rmdir`/`rename` report a non-empty directory as either
+1
View File
@@ -704,6 +704,7 @@ pub(crate) async fn create_bitrot_reader_from_bytes_with_stage_metrics(
}
#[allow(clippy::too_many_arguments)]
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub fn create_deferred_bitrot_reader(
inline_data: Option<Bytes>,
disk: Option<DiskStore>,
+241 -56
View File
@@ -180,11 +180,13 @@ pub(in crate::set_disk) enum GetCodecStreamingReaderBuildOutcome {
Fallback(GetCodecStreamingFallbackReason),
}
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub(in crate::set_disk) struct MultipartCodecStreamingReader {
pub(in crate::set_disk) readers: VecDeque<Box<dyn AsyncRead + Unpin + Send + Sync>>,
}
impl MultipartCodecStreamingReader {
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub(in crate::set_disk) fn new(readers: Vec<Box<dyn AsyncRead + Unpin + Send + Sync>>) -> Self {
Self {
readers: VecDeque::from(readers),
@@ -666,6 +668,60 @@ pub(in crate::set_disk) async fn data_read_early_stop_inline_body_miss_reason(
parts_metadata: &[FileInfo],
disks: &[Option<DiskStore>],
) -> Option<&'static str> {
if let Some(reason) = data_read_early_stop_inline_candidate_miss_reason(candidate) {
return Some(reason);
}
let Ok(erasure) = coding::Erasure::try_new_with_options(
candidate.erasure.data_blocks,
candidate.erasure.parity_blocks,
candidate.erasure.block_size,
candidate.uses_legacy_checksum,
) else {
return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY);
};
let data_files =
match collect_inline_data_shard_fileinfos_by_index_or_reason(parts_metadata, candidate, erasure.data_shards, |index| {
disks.get(index).is_some_and(Option::is_some)
}) {
Ok(data_files) => data_files,
Err(reason) => return Some(reason),
};
let Some(part) = candidate.parts.first() else {
return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_PART_SHAPE);
};
let Ok(object_size) = usize::try_from(candidate.size) else {
return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_SIZE);
};
let checksum_info = candidate.erasure.get_checksum_info(part.number);
let checksum_algo = if candidate.uses_legacy_checksum && checksum_info.algorithm == HashAlgorithm::HighwayHash256S {
HashAlgorithm::HighwayHash256SLegacy
} else {
checksum_info.algorithm
};
let read_length = inline_erasure_shard_file_offset(
0,
object_size,
object_size,
candidate.erasure.block_size,
erasure.data_shards,
candidate.uses_legacy_checksum,
);
let shard_size = inline_erasure_shard_size(candidate.erasure.block_size, erasure.data_shards, candidate.uses_legacy_checksum);
let Ok(mut readers) =
build_inline_bitrot_readers_from_refs(&data_files, bucket, object, read_length, shard_size, &checksum_algo, false).await
else {
return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY);
};
match try_read_inline_data_shards_direct(&mut readers, erasure.data_shards, read_length, object_size).await {
Some(body) if body.len() == object_size => None,
_ => Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY),
}
}
fn data_read_early_stop_inline_candidate_miss_reason(candidate: &FileInfo) -> Option<&'static str> {
// `inline_data` excludes remote objects; this diagnostic reports them separately.
if !rustfs_utils::http::contains_key_str(&candidate.metadata, rustfs_utils::http::SUFFIX_INLINE_DATA) {
return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_NOT_INLINE);
@@ -703,51 +759,7 @@ pub(in crate::set_disk) async fn data_read_early_stop_inline_body_miss_reason(
if !can_try_inline_data_shards_direct(object_size, candidate.erasure.block_size) {
return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_SIZE);
}
let Ok(erasure) = coding::Erasure::try_new_with_options(
candidate.erasure.data_blocks,
candidate.erasure.parity_blocks,
candidate.erasure.block_size,
candidate.uses_legacy_checksum,
) else {
return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY);
};
let data_files =
match collect_inline_data_shard_fileinfos_by_index_or_reason(parts_metadata, candidate, erasure.data_shards, |index| {
disks.get(index).is_some_and(Option::is_some)
}) {
Ok(data_files) => data_files,
Err(reason) => return Some(reason),
};
let Some(part) = candidate.parts.first() else {
return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_PART_SHAPE);
};
let checksum_info = candidate.erasure.get_checksum_info(part.number);
let checksum_algo = if candidate.uses_legacy_checksum && checksum_info.algorithm == HashAlgorithm::HighwayHash256S {
HashAlgorithm::HighwayHash256SLegacy
} else {
checksum_info.algorithm
};
let read_length = inline_erasure_shard_file_offset(
0,
object_size,
object_size,
candidate.erasure.block_size,
erasure.data_shards,
candidate.uses_legacy_checksum,
);
let shard_size = inline_erasure_shard_size(candidate.erasure.block_size, erasure.data_shards, candidate.uses_legacy_checksum);
let Ok(mut readers) =
build_inline_bitrot_readers_from_refs(&data_files, bucket, object, read_length, shard_size, &checksum_algo, false).await
else {
return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY);
};
match try_read_inline_data_shards_direct(&mut readers, erasure.data_shards, read_length, object_size).await {
Some(body) if body.len() == object_size => None,
_ => Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY),
}
None
}
fn data_read_inline_missing_shards_are_pending(
@@ -1836,6 +1848,7 @@ pub(in crate::set_disk) async fn create_bitrot_readers_until_quorum_all_shards(
}
#[allow(clippy::too_many_arguments)]
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub(in crate::set_disk) async fn create_bitrot_readers_until_quorum(
files: &[FileInfo],
disks: &[Option<DiskStore>],
@@ -2126,6 +2139,7 @@ pub(in crate::set_disk) async fn create_data_block_bitrot_readers(
setup
}
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub(in crate::set_disk) async fn collect_read_multiple_results<F>(
tasks: Vec<F>,
read_quorum: usize,
@@ -2442,6 +2456,7 @@ impl SetDisks {
let bucket: Arc<str> = Arc::from(bucket);
let object: Arc<str> = Arc::from(object);
let version_id: Arc<str> = Arc::from(version_id);
let slowtail_fault = get_metadata_slowtail_fault_request(bucket.as_ref(), object.as_ref(), read_data);
let futures = disks.iter().enumerate().map(|(disk_index, disk)| {
let disk = disk.clone();
let task_opts = opts;
@@ -2449,10 +2464,14 @@ impl SetDisks {
let bucket = bucket.clone();
let object = object.clone();
let version_id = version_id.clone();
let slowtail_fault = slowtail_fault.clone();
tokio::spawn(async move {
let response_start = observe.then(Instant::now);
let result = if let Some(disk) = disk {
Self::record_read_version_call(&object, disk_index);
if let Some(delay) = slowtail_fault.as_ref().and_then(|fault| fault.delay_for_disk(disk_index)) {
tokio::time::sleep(delay).await;
}
disk.read_version(&org_bucket, &bucket, &object, &version_id, &task_opts)
.await
} else {
@@ -2548,6 +2567,7 @@ impl SetDisks {
let mut scheduled_count = 0usize;
let mut force_full_wait = false;
let mut final_miss_reason_override = None;
let slowtail_fault = get_metadata_slowtail_fault_request(bucket.as_ref(), object.as_ref(), read_data);
let spawn_read_version =
|join_set: &mut JoinSet<(usize, disk::error::Result<FileInfo>, Duration)>, index: usize, disk: Option<DiskStore>| {
let task_opts = opts;
@@ -2555,6 +2575,7 @@ impl SetDisks {
let bucket = bucket.clone();
let object = object.clone();
let version_id = version_id.clone();
let slowtail_fault = slowtail_fault.clone();
join_set.spawn(async move {
let response_start = Instant::now();
let result = if let Some(disk) = disk {
@@ -2563,6 +2584,9 @@ impl SetDisks {
Self::record_read_version_call(&object, index);
#[cfg(test)]
Self::read_version_fanout_barrier(&object, index).await;
if let Some(delay) = slowtail_fault.as_ref().and_then(|fault| fault.delay_for_disk(index)) {
tokio::time::sleep(delay).await;
}
disk.read_version(&org_bucket, &bucket, &object, &version_id, &task_opts)
.await
} else {
@@ -2596,6 +2620,14 @@ impl SetDisks {
Ok(file_info) => {
observations.push(MetadataFanoutObservation::from_file_info(&file_info, elapsed));
accumulator.observe_file_info(&file_info);
if bounded_fanout
&& read_data
&& !force_full_wait
&& let Some(reason) = data_read_early_stop_inline_candidate_miss_reason(&file_info)
{
force_full_wait = true;
final_miss_reason_override.get_or_insert(reason);
}
if let Some(slot) = ress.get_mut(index) {
*slot = file_info;
}
@@ -2955,6 +2987,7 @@ impl SetDisks {
(meta_file_infos, errs)
}
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub(in crate::set_disk) async fn read_multiple_files(
disks: &[Option<DiskStore>],
req: ReadMultipleReq,
@@ -2988,14 +3021,11 @@ impl SetDisks {
});
}
let (ress, errors) = match collect_read_multiple_results(futures, read_quorum).await {
let (ress, _errors) = match collect_read_multiple_results(futures, read_quorum).await {
Ok(collected) => collected,
Err(()) => return empty_quorum_result(),
};
// debug!("ReadMultipleResp ress {:?}", ress);
// debug!("ReadMultipleResp errors {:?}", errors);
let mut ret = Vec::with_capacity(req.files.len());
for want in req.files.iter() {
@@ -3134,6 +3164,7 @@ pub(in crate::set_disk) struct RenameDataCommit {
pub(in crate::set_disk) committed_file_info: FileInfo,
}
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
type RenameDataLegacyTuple = (
Vec<Option<DiskStore>>,
RenameConvergence,
@@ -3143,6 +3174,7 @@ type RenameDataLegacyTuple = (
);
impl RenameDataCommit {
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
fn into_legacy_tuple(self) -> RenameDataLegacyTuple {
(
self.online_disks,
@@ -3261,6 +3293,7 @@ impl SetDisks {
#[tracing::instrument(level = "debug", skip(disks, file_infos))]
#[allow(clippy::type_complexity)]
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub(in crate::set_disk) async fn rename_data(
disks: &[Option<DiskStore>],
src_bucket: &str,
@@ -5073,6 +5106,7 @@ fn is_cleanup_not_found(e: &DiskError) -> bool {
/// normalized to `DiskNotFound`: a panic is not a "disk absent" condition and
/// must not be silently swallowed as an ignorable error (fixes the historical
/// `Unexpected`/`DiskNotFound` misclassification).
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
fn map_cleanup_join_result(joined: std::result::Result<Option<DiskError>, tokio::task::JoinError>) -> Option<DiskError> {
match joined {
Ok(res) => res,
@@ -5297,6 +5331,7 @@ pub(in crate::set_disk) mod rename_fanout_barrier_phase {
/// The per-disk old-data-dir cleanup phase of the commit fan-out.
pub const CLEANUP: &str = "cleanup";
/// The per-disk `read_version` phase of metadata read fan-out.
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub const READ_VERSION: &str = "read_version";
}
@@ -5734,6 +5769,130 @@ mod tests {
(dirs, disks)
}
#[test]
fn metadata_slowtail_fault_delay_parses_and_filters_request() {
temp_env::with_vars(
[
(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS, Some("25")),
(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS, Some("1,3")),
(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_BUCKET, Some("bench-bucket")),
(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_OBJECT_PREFIX, Some("objects/")),
],
|| {
assert_eq!(
get_metadata_slowtail_fault_delay("bench-bucket", "objects/000001", 3, true),
Some(Duration::from_millis(25))
);
assert!(get_metadata_slowtail_fault_delay("bench-bucket", "objects/000001", 2, true).is_none());
assert!(get_metadata_slowtail_fault_delay("other-bucket", "objects/000001", 3, true).is_none());
assert!(get_metadata_slowtail_fault_delay("bench-bucket", "other/000001", 3, true).is_none());
assert!(get_metadata_slowtail_fault_delay("bench-bucket", "objects/000001", 3, false).is_none());
},
);
}
#[test]
fn metadata_slowtail_fault_delay_disables_invalid_disk_list() {
temp_env::with_vars(
[
(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS, Some("25")),
(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS, Some("1,nope")),
],
|| {
assert!(get_metadata_slowtail_fault_delay("bucket", "object", 1, true).is_none());
},
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn metadata_slowtail_fault_delays_only_data_read_metadata_task() {
const DISKS: usize = 4;
let bucket = "metadata-slowtail-fault-bucket";
let object = "objects/metadata-slowtail-fault-object";
let (dirs, disks) = call_counter_local_disks(bucket, DISKS).await;
install_metadata_fanout_fileinfo(&disks, bucket, object, None).await;
temp_env::async_with_vars(
[
(ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE, Some("false")),
(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS, Some("150")),
(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS, Some("3")),
(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_BUCKET, Some(bucket)),
(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_OBJECT_PREFIX, Some("objects/")),
],
async {
let read_without_data =
SetDisks::read_all_fileinfo_observed(&disks, bucket, bucket, object, "", false, false, false, true, 2);
tokio::time::timeout(Duration::from_millis(100), read_without_data)
.await
.expect("non-data metadata fanout must not be delayed by the data-read slowtail hook")
.expect("metadata fanout without read_data should resolve");
let mut read_with_data = Box::pin(SetDisks::read_all_fileinfo_observed(
&disks, bucket, bucket, object, "", true, false, false, true, 2,
));
assert!(
tokio::time::timeout(Duration::from_millis(40), &mut read_with_data)
.await
.is_err(),
"data-read metadata fanout must wait for the injected slow read_version response"
);
let (parts_metadata, errs, diagnostics) = tokio::time::timeout(Duration::from_secs(2), read_with_data)
.await
.expect("injected slowtail should eventually complete")
.expect("data-read metadata fanout should resolve");
assert_eq!(parts_metadata.iter().filter(|fi| fi.name == object).count(), DISKS);
assert!(errs.iter().all(Option::is_none));
assert_eq!(diagnostics.total_responses(), DISKS);
},
)
.await;
drop(dirs);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn metadata_slowtail_fault_delays_early_stop_metadata_task() {
const DISKS: usize = 4;
let bucket = "metadata-slowtail-early-stop-bucket";
let object = "objects/metadata-slowtail-early-stop-object";
let (dirs, disks) = call_counter_local_disks(bucket, DISKS).await;
install_metadata_fanout_fileinfo(&disks, bucket, object, None).await;
temp_env::async_with_vars(
[
(ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE, Some("true")),
(ENV_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE, Some("true")),
(ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT, Some("false")),
(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS, Some("150")),
(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS, Some("3")),
(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_BUCKET, Some(bucket)),
(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_OBJECT_PREFIX, Some("objects/")),
],
async {
let mut read_with_data = Box::pin(SetDisks::read_all_fileinfo_observed(
&disks, bucket, bucket, object, "", true, false, false, true, 2,
));
assert!(
tokio::time::timeout(Duration::from_millis(40), &mut read_with_data)
.await
.is_err(),
"early-stop metadata fanout must still wait for the injected slow response after fallback to full wait"
);
let (parts_metadata, errs, diagnostics) = tokio::time::timeout(Duration::from_secs(2), read_with_data)
.await
.expect("injected early-stop slowtail should eventually complete")
.expect("early-stop metadata fanout should resolve");
assert_eq!(parts_metadata.iter().filter(|fi| fi.name == object).count(), DISKS);
assert!(errs.iter().all(Option::is_none));
assert_eq!(diagnostics.total_responses(), DISKS);
},
)
.await;
drop(dirs);
}
/// Demo / regression guard for the backlog#1325 per-disk call counters.
///
/// The metadata fan-out issues each `read_version` inside its own
@@ -7081,7 +7240,7 @@ mod tests {
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn bounded_non_inline_data_get_hedges_then_waits_for_full_fanout() {
async fn bounded_non_inline_data_get_immediately_forces_full_fanout() {
const DISKS: usize = 4;
let bucket = "bounded-data-get-hedge-bucket";
let object = "bounded-data-get-hedge-object";
@@ -7112,7 +7271,7 @@ mod tests {
}
})
.await
.expect("bounded data-read fanout should hedge by starting the spare disk");
.expect("bounded non-inline data-read fanout should immediately schedule the spare disk");
let pending = tokio::time::timeout(BARRIER_PAUSE_GUARD, &mut read).await;
assert!(
@@ -7128,7 +7287,7 @@ mod tests {
assert_eq!(
calls.total(disk_call_counters::KIND_READ_VERSION),
DISKS as u64,
"bounded data-read fanout should issue the paused disk plus one spare hedge"
"bounded non-inline data-read fanout should issue the paused disk plus the remaining spare"
);
assert_eq!(diagnostics.total_responses(), DISKS);
assert_eq!(parts_metadata.iter().filter(|fi| fi.name == object).count(), DISKS);
@@ -7155,16 +7314,42 @@ mod tests {
("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", None::<&str>),
],
async {
let barrier = rename_fanout_barrier::arm(object, 2, rename_fanout_barrier::PHASE_READ_VERSION);
let calls = disk_call_counters::observe(object);
let (parts_metadata, errs, diagnostics) =
SetDisks::read_all_fileinfo_observed(&disks, bucket, bucket, object, "", true, false, false, true, 2)
let disks_for_read = disks.clone();
let mut read = tokio::spawn(async move {
SetDisks::read_all_fileinfo_observed(&disks_for_read, bucket, bucket, object, "", true, false, false, true, 2)
.await
.expect("default data-read metadata should resolve");
});
tokio::time::timeout(BARRIER_PAUSE_GUARD, barrier.wait_until_paused())
.await
.expect("default bounded non-inline read should schedule the paused metadata task");
tokio::time::timeout(BARRIER_PAUSE_GUARD, async {
while calls.for_disk(disk_call_counters::KIND_READ_VERSION, 3) == 0 {
tokio::task::yield_now().await;
}
})
.await
.expect(
"default bounded non-inline read should immediately force full fanout after the first non-inline response",
);
let pending = tokio::time::timeout(BARRIER_PAUSE_GUARD, &mut read).await;
assert!(
pending.is_err(),
"default non-inline data reads must not return before the paused metadata response"
);
barrier.release();
let (parts_metadata, errs, diagnostics) = read
.await
.expect("metadata read task should not panic")
.expect("default data-read metadata should resolve");
assert_eq!(
calls.total(disk_call_counters::KIND_READ_VERSION),
DISKS as u64,
"default non-inline GET data-read metadata must keep full fanout for read-failure tolerance"
"default non-inline GET data-read metadata must keep full fanout without waiting for a quorum miss first"
);
assert_eq!(diagnostics.total_responses(), DISKS);
assert_eq!(parts_metadata.iter().filter(|fi| fi.name == object).count(), DISKS);
+28
View File
@@ -42,12 +42,20 @@ impl<'a> SetDisksCtx<'a> {
}
/// The borrowed core, for state not yet fronted by a typed accessor.
#[allow(
dead_code,
reason = "SetDisks split seam (backlog#815) with no caller in this port (backlog#1823)"
)]
pub(crate) fn core(&self) -> &'a SetDisks {
self.core
}
// --- Immutable topology / config (fixed after construction) ---
#[allow(
dead_code,
reason = "SetDisks split seam (backlog#815) with no caller in this port (backlog#1823)"
)]
pub(crate) fn set_index(&self) -> usize {
self.core.set_index
}
@@ -56,14 +64,26 @@ impl<'a> SetDisksCtx<'a> {
self.core.pool_index
}
#[allow(
dead_code,
reason = "SetDisks split seam (backlog#815) with no caller in this port (backlog#1823)"
)]
pub(crate) fn set_drive_count(&self) -> usize {
self.core.set_drive_count
}
#[allow(
dead_code,
reason = "SetDisks split seam (backlog#815) with no caller in this port (backlog#1823)"
)]
pub(crate) fn default_parity_count(&self) -> usize {
self.core.default_parity_count
}
#[allow(
dead_code,
reason = "SetDisks split seam (backlog#815) with no caller in this port (backlog#1823)"
)]
pub(crate) fn set_endpoints(&self) -> &'a [Endpoint] {
&self.core.set_endpoints
}
@@ -72,6 +92,10 @@ impl<'a> SetDisksCtx<'a> {
&self.core.format
}
#[allow(
dead_code,
reason = "SetDisks split seam (backlog#815) with no caller in this port (backlog#1823)"
)]
pub(crate) fn locker_owner(&self) -> &'a str {
&self.core.locker_owner
}
@@ -84,6 +108,10 @@ impl<'a> SetDisksCtx<'a> {
// --- Locker trio ---
#[allow(
dead_code,
reason = "SetDisks split seam (backlog#815) with no caller in this port (backlog#1823)"
)]
pub(crate) fn lockers(&self) -> &'a [Arc<dyn LockClient>] {
&self.core.lockers
}
+131 -17
View File
@@ -39,7 +39,6 @@
//! - `metadata.rs`, `replication.rs`, `shard_source.rs` — supporting helpers.
// #730: SetDisks still hosts staged read/heal/write migration helpers.
#![allow(dead_code)]
#![allow(unused_imports)]
#![allow(unused_variables)]
@@ -103,9 +102,7 @@ use crate::storage_api_contracts::{
};
use crate::store::utils::is_reserved_or_invalid_bucket;
use crate::{
bucket::lifecycle::bucket_lifecycle_ops::{
LifecycleOps, gen_transition_objname, get_transitioned_object_reader_with_tier_manager, put_restore_opts,
},
bucket::lifecycle::bucket_lifecycle_ops::{LifecycleOps, get_transitioned_object_reader_with_tier_manager, put_restore_opts},
cache_value::metacache_set::{ListPathRawOptions, list_path_raw},
config::storageclass,
disk::{
@@ -177,15 +174,14 @@ use std::future::Future;
use std::hash::{BuildHasher, Hash, Hasher};
use std::mem::{self};
use std::pin::Pin;
use std::sync::OnceLock;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, OnceLock};
use std::task::{Context, Poll};
use std::time::{Instant, SystemTime, UNIX_EPOCH};
use std::{
collections::{HashMap, HashSet},
io::{Cursor, Write},
path::Path,
sync::Arc,
time::Duration,
};
use time::OffsetDateTime;
@@ -624,7 +620,9 @@ fn adaptive_duplex_buffer_size(object_size: i64) -> usize {
// Each flag has a corresponding `*_ROLLOUT_PCT` for percentage-based gradual rollout.
// ============================================================================
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
const DISK_ONLINE_TIMEOUT: Duration = Duration::from_secs(1);
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
const DISK_HEALTH_CACHE_TTL: Duration = Duration::from_millis(750);
const GET_OBJECT_METADATA_CACHE_TTL: Duration = Duration::from_secs(2); // Increased from 250ms to 2s
const DEFAULT_GET_OBJECT_METADATA_CACHE_MAX_ENTRIES: usize = 4096; // Increased from 1024 to 4096
@@ -698,7 +696,15 @@ const ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE: &str = "RUSTFS_GET_METADATA_EAR
// the env var to `false` to fall back to full-wait metadata fanout.
const DEFAULT_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE: bool = true;
#[allow(
dead_code,
reason = "percentage-rollout facet of the metadata early-stop switch; its predicate has no caller while the sibling enable flag is live (backlog#1823)"
)]
const ENV_RUSTFS_GET_METADATA_EARLY_STOP_ROLLOUT_PCT: &str = "RUSTFS_GET_METADATA_EARLY_STOP_ROLLOUT_PCT";
#[allow(
dead_code,
reason = "percentage-rollout facet of the metadata early-stop switch; its predicate has no caller while the sibling enable flag is live (backlog#1823)"
)]
const DEFAULT_RUSTFS_GET_METADATA_EARLY_STOP_ROLLOUT_PCT: u32 = 100;
const ENV_RUSTFS_GET_METADATA_VERSION_EARLY_STOP_ENABLE: &str = "RUSTFS_GET_METADATA_VERSION_EARLY_STOP_ENABLE";
@@ -708,7 +714,12 @@ const ENV_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE: &str = "RUSTFS_GET_ME
const DEFAULT_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE: bool = true;
const ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT: &str = "RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT";
const DEFAULT_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT: bool = false;
const DEFAULT_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT: bool = true;
const ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS: &str = "RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS";
const ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS: &str = "RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS";
const ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_BUCKET: &str = "RUSTFS_GET_METADATA_SLOWTAIL_FAULT_BUCKET";
const ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_OBJECT_PREFIX: &str = "RUSTFS_GET_METADATA_SLOWTAIL_FAULT_OBJECT_PREFIX";
// --- Multipart Reader-Setup Prefetch Configuration (backlog#870) ---
@@ -910,6 +921,10 @@ mod prepared_get_object_metadata_tests {
.expect("test should find an object whose initial fanout covers both data shards")
}
#[allow(
dead_code,
reason = "test fixture no assertion in this module uses today; the live namesake lives in io_primitives tests (backlog#1823)"
)]
fn bounded_spare_disk_index(bucket: &str, object: &str) -> usize {
*bounded_metadata_fanout_order(bucket, object, 4, 2)
.get(3)
@@ -1110,7 +1125,10 @@ mod prepared_get_object_metadata_tests {
assert_eq!(object_size, payload.len() as i64);
assert_eq!(restored, payload);
assert_eq!(calls_total, 4, "default production GET should eagerly schedule the full metadata fanout");
assert_eq!(
calls_total, 4,
"default production inline GET should schedule the initial bounded quorum plus one hedge"
);
assert_eq!(
recorder.histogram_values(
"rustfs_io_get_object_metadata_fanout_scheduled",
@@ -1684,6 +1702,95 @@ fn is_get_metadata_early_stop_bounded_fanout_enabled() -> bool {
}
}
#[derive(Debug)]
struct GetMetadataSlowtailFaultConfig {
delay: Duration,
disks: Arc<[usize]>,
bucket: Option<String>,
object_prefix: Option<String>,
}
#[derive(Clone, Debug)]
struct GetMetadataSlowtailFaultRequest {
delay: Duration,
disks: Arc<[usize]>,
}
impl GetMetadataSlowtailFaultRequest {
fn delay_for_disk(&self, disk_index: usize) -> Option<Duration> {
self.disks.contains(&disk_index).then_some(self.delay)
}
}
fn parse_get_metadata_slowtail_fault_disks(raw: &str) -> Option<Vec<usize>> {
let mut disks = Vec::new();
for item in raw.split(',').map(str::trim).filter(|item| !item.is_empty()) {
let Ok(index) = item.parse::<usize>() else {
return None;
};
if !disks.contains(&index) {
disks.push(index);
}
}
(!disks.is_empty()).then_some(disks)
}
fn load_get_metadata_slowtail_fault_config() -> Option<GetMetadataSlowtailFaultConfig> {
let delay_ms = rustfs_utils::get_env_u64(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS, 0);
if delay_ms == 0 {
return None;
}
let disks = parse_get_metadata_slowtail_fault_disks(&std::env::var(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS).ok()?)?;
let bucket = std::env::var(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_BUCKET)
.ok()
.filter(|value| !value.is_empty());
let object_prefix = std::env::var(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_OBJECT_PREFIX)
.ok()
.filter(|value| !value.is_empty());
Some(GetMetadataSlowtailFaultConfig {
delay: Duration::from_millis(delay_ms),
disks: Arc::from(disks.into_boxed_slice()),
bucket,
object_prefix,
})
}
fn get_metadata_slowtail_fault_request(bucket: &str, object: &str, read_data: bool) -> Option<GetMetadataSlowtailFaultRequest> {
if !read_data {
return None;
}
#[cfg(test)]
let config = load_get_metadata_slowtail_fault_config();
#[cfg(test)]
let config = config.as_ref()?;
#[cfg(not(test))]
let config = ({
static CACHED: OnceLock<Option<GetMetadataSlowtailFaultConfig>> = OnceLock::new();
CACHED.get_or_init(load_get_metadata_slowtail_fault_config).as_ref()
})?;
if let Some(expected_bucket) = &config.bucket
&& expected_bucket != bucket
{
return None;
}
if let Some(expected_prefix) = &config.object_prefix
&& !object.starts_with(expected_prefix)
{
return None;
}
Some(GetMetadataSlowtailFaultRequest {
delay: config.delay,
disks: config.disks.clone(),
})
}
#[cfg(test)]
fn get_metadata_slowtail_fault_delay(bucket: &str, object: &str, disk_index: usize, read_data: bool) -> Option<Duration> {
get_metadata_slowtail_fault_request(bucket, object, read_data)?.delay_for_disk(disk_index)
}
/// Check if multipart reads prefetch the next part's bitrot reader setup
/// while the current part decodes (backlog#870).
///
@@ -1709,6 +1816,10 @@ fn is_multipart_reader_setup_prefetch_enabled() -> bool {
}
}
#[allow(
dead_code,
reason = "percentage-rollout facet of the metadata early-stop switch; its predicate has no caller while the sibling enable flag is live (backlog#1823)"
)]
fn get_metadata_early_stop_rollout_pct() -> u32 {
static CACHED: OnceLock<u32> = OnceLock::new();
*CACHED.get_or_init(|| {
@@ -1748,6 +1859,10 @@ fn should_use_codec_streaming(config: GetCodecStreamingConfig, bucket: &str, obj
}
/// Should this specific request use metadata early-stop?
#[allow(
dead_code,
reason = "percentage-rollout facet of the metadata early-stop switch; its predicate has no caller while the sibling enable flag is live (backlog#1823)"
)]
pub fn should_use_metadata_early_stop(bucket: &str, object: &str) -> bool {
let base = is_get_metadata_early_stop_enabled();
let pct = get_metadata_early_stop_rollout_pct();
@@ -2181,6 +2296,7 @@ fn classify_get_codec_streaming_object_class(
GetCodecStreamingObjectClass::PlainSinglePart
}
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
fn is_get_small_object_direct_memory_eligible_with_threshold(
range: &Option<HTTPRangeSpec>,
object_info: &ObjectInfo,
@@ -2786,6 +2902,7 @@ pub struct SetDisks {
/// Stable namespace shared by every object lock created for this set.
set_lock_namespace: Arc<str>,
pub format: FormatV3,
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
disk_health_cache: Arc<RwLock<Vec<Option<DiskHealthEntry>>>>,
get_object_metadata_cache: moka::future::Cache<GetObjectMetadataCacheKey, Arc<GetObjectMetadataCacheEntry>>,
get_object_metadata_cache_hash_builder: std::collections::hash_map::RandomState,
@@ -3061,11 +3178,13 @@ struct GetObjectMetadataCacheEntry {
#[derive(Clone, Debug)]
struct DiskHealthEntry {
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
last_check: Instant,
online: bool,
}
impl DiskHealthEntry {
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
fn cached_value(&self) -> Option<bool> {
if self.last_check.elapsed() <= DISK_HEALTH_CACHE_TTL {
Some(self.online)
@@ -3659,6 +3778,7 @@ fn multipart_put_large_batch_min_size_bytes() -> usize {
})
}
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
fn classify_small_write_path(is_inline_buffer: bool, object_size: i64, block_size: usize) -> SmallWritePath {
if should_use_inline_small_fast_path(is_inline_buffer, object_size, block_size) {
SmallWritePath::Inline
@@ -4237,6 +4357,7 @@ fn check_object_lock_retention_update(bucket: &str, object: &str, obj_info: &Obj
///
/// Fail closed: when bucket metadata cannot be resolved the check stays on, so
/// object-lock protection is never skipped because of a metadata lookup miss.
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub(crate) fn object_lock_delete_check_required(bucket_meta: Option<&crate::bucket::metadata::BucketMetadata>) -> bool {
bucket_meta.is_none_or(|meta| meta.object_locking())
}
@@ -4512,15 +4633,6 @@ impl Hash for ObjProps {
}
}
#[derive(Default, Clone, Debug)]
pub struct HealEntryResult {
pub bytes: usize,
pub success: bool,
pub skipped: bool,
pub entry_done: bool,
pub name: String,
}
fn is_object_dangling(
meta_arr: &[FileInfo],
errs: &[Option<DiskError>],
@@ -5297,6 +5409,7 @@ pub fn is_valid_storage_class(storage_class: &str) -> bool {
}
/// Returns true if the storage class is a cold storage tier that requires special handling
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub fn is_cold_storage_class(storage_class: &str) -> bool {
matches!(
storage_class,
@@ -5305,6 +5418,7 @@ pub fn is_cold_storage_class(storage_class: &str) -> bool {
}
/// Returns true if the storage class is an infrequent access tier
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub fn is_infrequent_access_class(storage_class: &str) -> bool {
matches!(
storage_class,
+11 -2
View File
@@ -453,7 +453,9 @@ impl SetDisks {
..Default::default()
};
let write_lock_guard = if !opts.no_lock {
// Bound, not `_`: this guard must live to the end of the scope. A bare
// `_` would drop it here and release the namespace write lock.
let _write_lock_guard = if !opts.no_lock {
let ns_lock = self.new_ns_lock(bucket, object).await?;
Some(
ns_lock
@@ -996,7 +998,7 @@ impl SetDisks {
readers.push(None);
continue;
}
Err(e) => {
Err(_e) => {
readers.push(None);
continue;
}
@@ -1545,6 +1547,9 @@ impl SetDisks {
for candidate in candidates.iter_mut().filter(|candidate| candidate.local_payload) {
for (disk_index, disk) in disks.iter().enumerate() {
// Only the #[cfg(test)] fault-injection branch below reads this.
#[cfg(not(test))]
let _ = disk_index;
let Some(disk) = disk else {
return Ok(DanglingDeleteSafety::UnsafeToDelete);
};
@@ -1716,6 +1721,10 @@ impl SetDisks {
Ok((result, None))
}
#[allow(
dead_code,
reason = "lock-taking wrapper over the live heal_object_dir_locked; only comments reference it (backlog#1823)"
)]
#[tracing::instrument(level = "trace", skip(self), fields(bucket = %bucket, object = %object))]
pub(in crate::set_disk) async fn heal_object_dir(
&self,
@@ -66,6 +66,7 @@ impl crate::storage_api_contracts::namespace::NamespaceLocking for SetDisks {
}
impl SetDisks {
#[allow(dead_code, reason = "lock diagnostics formatter with no caller in this port (backlog#1823)")]
pub(in crate::set_disk) fn format_lock_error(&self, bucket: &str, object: &str, mode: &str, err: &LockResult) -> String {
match err {
LockResult::Timeout => {
@@ -79,6 +80,7 @@ impl SetDisks {
}
}
#[allow(dead_code, reason = "lock diagnostics formatter with no caller in this port (backlog#1823)")]
pub(in crate::set_disk) fn format_lock_error_from_error(
&self,
bucket: &str,
@@ -143,6 +145,7 @@ impl SetDisks {
disks
}
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub(in crate::set_disk) async fn get_online_disks(&self) -> Vec<Option<DiskStore>> {
let snapshot = self.drive_membership_snapshot().await;
let mut disks = snapshot.strict_online_candidates().into_iter().map(Some).collect::<Vec<_>>();
@@ -153,6 +156,10 @@ impl SetDisks {
disks
}
#[allow(
dead_code,
reason = "local-only sibling of the test-covered get_online_disks; no caller in this port (backlog#1823)"
)]
pub(in crate::set_disk) async fn get_online_local_disks(&self) -> Vec<Option<DiskStore>> {
let snapshot = self.drive_membership_snapshot().await;
let mut disks = snapshot
@@ -432,6 +439,10 @@ impl SetDisks {
Ok((disk, fm))
}
#[allow(
dead_code,
reason = "MinIO-parity healing-disk accessor with no caller in this port (backlog#1823)"
)]
pub(in crate::set_disk) async fn get_online_disk_with_healing(
&self,
incl_healing: bool,
@@ -440,6 +451,10 @@ impl SetDisks {
Ok((new_disks, healing > 0))
}
#[allow(
dead_code,
reason = "reached only from get_online_disk_with_healing, itself uncalled in this port (backlog#1823)"
)]
pub(in crate::set_disk) async fn get_online_disk_with_healing_and_info(
&self,
incl_healing: bool,
+7 -5
View File
@@ -415,6 +415,7 @@ fn reduce_quorum_part_numbers(object_parts: Vec<Vec<String>>, read_quorum: usize
/// never returned, but flips `is_truncated` to `true` and yields a
/// `next_upload_id_marker` pointing at the last returned upload so the caller can
/// resume paging.
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
fn paginate_upload_page(remaining: &[MultipartInfo], max_uploads: usize) -> (Vec<MultipartInfo>, bool, Option<String>) {
let is_truncated = remaining.len() > max_uploads;
let page: Vec<MultipartInfo> = remaining.iter().take(max_uploads).cloned().collect();
@@ -557,6 +558,7 @@ impl SetDisks {
}
#[tracing::instrument(level = "debug", skip(self))]
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub(super) async fn check_upload_id_exists(
&self,
bucket: &str,
@@ -1398,7 +1400,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
let mut count = max_parts;
for (i, part) in object_parts.iter().enumerate() {
for part in object_parts.iter() {
if let Some(err) = &part.error {
warn!("list_object_parts part error: {:?}", &err);
}
@@ -2041,8 +2043,8 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
&& let Err(err) = checksum.add_part(&cs, ext_part.actual_size)
{
error!(
"complete_multipart_upload checksum add_part failed part_id={}, bucket={}, object={}",
p.part_num, bucket, object
"complete_multipart_upload checksum add_part failed part_id={}, bucket={}, object={}, err={}",
p.part_num, bucket, object, err
);
return Err(Error::InvalidPart(p.part_num, ext_part.etag.clone(), p.etag.clone().unwrap_or_default()));
}
@@ -2087,8 +2089,8 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
}
} else if let Err(err) = wtcs.matches(&checksum_combined, uploaded_parts.len() as i32) {
error!(
"complete_multipart_upload checksum matches failed want={}, got={}",
wtcs.encoded, checksum.encoded
"complete_multipart_upload checksum matches failed want={}, got={}, err={}",
wtcs.encoded, checksum.encoded, err
);
return Err(Error::other(format!(
"complete_multipart_upload checksum matches failed want={}, got={}",
+322 -2
View File
@@ -3507,6 +3507,10 @@ struct TransitionUploadedSaveProbeState {
}
#[cfg(test)]
#[allow(
dead_code,
reason = "installed by set_disk tests behind `--features test-util` (backlog#1823)"
)]
struct TransitionUploadedSaveProbe {
state: Arc<TransitionUploadedSaveProbeState>,
}
@@ -3517,6 +3521,10 @@ static TRANSITION_UPLOADED_SAVE_PROBE: std::sync::OnceLock<std::sync::Mutex<Opti
#[cfg(test)]
impl TransitionUploadedSaveProbe {
#[allow(
dead_code,
reason = "installed by set_disk tests behind `--features test-util` (backlog#1823)"
)]
fn install(bucket: &str, object: &str) -> Self {
let state = Arc::new(TransitionUploadedSaveProbeState {
bucket: bucket.to_string(),
@@ -3533,6 +3541,10 @@ impl TransitionUploadedSaveProbe {
Self { state }
}
#[allow(
dead_code,
reason = "installed by set_disk tests behind `--features test-util` (backlog#1823)"
)]
fn attempts(&self) -> usize {
self.state.attempts.load(std::sync::atomic::Ordering::Acquire)
}
@@ -3738,6 +3750,10 @@ struct TransitionCommitBarrierState {
}
#[cfg(test)]
#[allow(
dead_code,
reason = "installed by set_disk tests behind `--features test-util` (backlog#1823)"
)]
struct TransitionCommitBarrier {
state: Arc<TransitionCommitBarrierState>,
}
@@ -3748,14 +3764,26 @@ static TRANSITION_COMMIT_BARRIER: std::sync::OnceLock<std::sync::Mutex<Option<Ar
#[cfg(test)]
impl TransitionCommitBarrier {
#[allow(
dead_code,
reason = "installed by set_disk tests behind `--features test-util` (backlog#1823)"
)]
fn install_before_lock_lost_check(bucket: &str, object: &str) -> Self {
Self::install_at(bucket, object, TransitionCommitPause::BeforeLockLost)
}
#[allow(
dead_code,
reason = "installed by set_disk tests behind `--features test-util` (backlog#1823)"
)]
fn install(bucket: &str, object: &str) -> Self {
Self::install_at(bucket, object, TransitionCommitPause::BeforeLeaseValidation)
}
#[allow(
dead_code,
reason = "installed by set_disk tests behind `--features test-util` (backlog#1823)"
)]
fn install_after_lease_check(bucket: &str, object: &str) -> Self {
Self::install_at(bucket, object, TransitionCommitPause::AfterLeaseValidation)
}
@@ -3778,12 +3806,20 @@ impl TransitionCommitBarrier {
Self { state }
}
#[allow(
dead_code,
reason = "installed by set_disk tests behind `--features test-util` (backlog#1823)"
)]
async fn wait_until_paused(&self) {
tokio::time::timeout(Duration::from_secs(30), self.state.arrived.notified())
.await
.expect("transition should reach the deterministic commit barrier");
}
#[allow(
dead_code,
reason = "installed by set_disk tests behind `--features test-util` (backlog#1823)"
)]
fn release(&self) {
self.state.release.notify_one();
}
@@ -5620,7 +5656,9 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
// TODO: Lifecycle
let mut version_found = true;
let (mut goi, write_quorum, gerr) = self.get_object_info_and_quorum(bucket, object, &opts).await;
// delete_object_version below derives its own majority quorum from the
// disk array, so the object-derived quorum here is unused.
let (mut goi, _write_quorum, gerr) = self.get_object_info_and_quorum(bucket, object, &opts).await;
if let Some(err) = &gerr
&& goi.name.is_empty()
{
@@ -6374,7 +6412,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
self.record_capacity_scope_if_needed(opts.capacity_scope_token, &disks);
for disk in disks.iter() {
if let Some(disk) = disk {
if disk.is_some() {
continue;
}
let _ = self
@@ -10170,6 +10208,288 @@ mod transition_upload_integrity_tests {
assert!(backend.contains(remote_object).await, "committed remote object should remain available");
}
/// Compresses `plaintext` with the codec the PUT path uses, so the stored
/// bytes round-trip through the read path's decompressor.
async fn compress_for_storage(plaintext: &[u8]) -> Vec<u8> {
let mut reader = crate::io_support::rio::compression_reader(
Cursor::new(plaintext.to_vec()),
rustfs_utils::CompressionAlgorithm::default(),
false,
);
let mut compressed = Vec::new();
reader.read_to_end(&mut compressed).await.expect("plaintext should compress");
assert!(compressed.len() < plaintext.len(), "test payload must actually compress");
compressed
}
/// Writes a genuinely compressed object: stored data is `compressed`, and the
/// metadata marks it compressed with the plaintext length as its actual size,
/// exactly as the app-layer compress path records it.
async fn write_compressed_source(
set_disks: &Arc<SetDisks>,
disk_stores: &[DiskStore],
bucket: &str,
object: &str,
plaintext: &[u8],
compressed: &[u8],
) -> ObjectInfo {
for disk in disk_stores {
disk.make_volume(bucket).await.expect("bucket volume should be created");
}
let mut user_defined = HashMap::new();
rustfs_utils::http::insert_str(
&mut user_defined,
rustfs_utils::http::SUFFIX_COMPRESSION,
crate::io_support::rio::compression_metadata_value(rustfs_utils::CompressionAlgorithm::default()),
);
rustfs_utils::http::insert_str(&mut user_defined, rustfs_utils::http::SUFFIX_ACTUAL_SIZE, plaintext.len().to_string());
let stream = crate::io_support::rio::HashReader::from_stream(
Cursor::new(compressed.to_vec()),
compressed.len() as i64,
plaintext.len() as i64,
None,
None,
false,
)
.expect("hash reader over compressed bytes");
let mut reader = PutObjReader::new(stream);
set_disks
.put_object(
bucket,
object,
&mut reader,
&ObjectOptions {
no_lock: true,
user_defined,
..Default::default()
},
)
.await
.expect("compressed object should be written")
}
async fn read_transitioned(
set_disks: &Arc<SetDisks>,
bucket: &str,
object: &str,
range: Option<HTTPRangeSpec>,
opts: &ObjectOptions,
) -> (Vec<u8>, i64) {
let mut reader = set_disks
.get_object_reader(bucket, object, range, HeaderMap::new(), opts)
.await
.expect("transitioned object reader should open");
let published_size = reader.object_info.size;
let mut body = Vec::new();
reader
.stream
.read_to_end(&mut body)
.await
.expect("transitioned body should drain");
(body, published_size)
}
/// Transition uploads the object's STORED bytes, so a tiered read has to
/// apply the same transform an erasure read would. #6107 routed this path
/// through `ReadPlan` to stop serving an encrypted object's ciphertext;
/// compression rides the same plan, and nothing pinned it (backlog#1851).
/// Without the transform this GET returns the compressed bytes under the
/// compressed size — silent corruption for every client of a compressed
/// object that ILM has moved to a warm tier.
#[tokio::test]
#[serial_test::serial]
async fn transitioned_compressed_object_get_returns_plaintext() {
let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
let bucket = "transitioned-compressed-get-bucket";
let object = "object.txt";
let plaintext = b"transitioned compressed objects must decompress on read ".repeat(20_000);
let compressed = compress_for_storage(&plaintext).await;
let original = write_compressed_source(&set_disks, &disk_stores, bucket, object, &plaintext, &compressed).await;
let opts = ObjectOptions {
no_lock: true,
..Default::default()
};
let (local_body, local_size) = read_transitioned(&set_disks, bucket, object, None, &opts).await;
assert_eq!(local_body, plaintext, "control: the pre-transition read must decompress");
assert_eq!(
local_size,
plaintext.len() as i64,
"control: the pre-transition read publishes the plaintext size"
);
let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
let backend = register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await;
set_disks
.transition_object(bucket, object, &transition_options(&original, tier_name))
.await
.expect("transition should commit");
let put_versions = backend.put_versions().await;
assert_eq!(put_versions.len(), 1, "transition should upload one remote candidate");
let remote_bytes = backend
.bytes(&put_versions[0].0)
.await
.expect("remote candidate should be stored");
assert_eq!(
remote_bytes, compressed,
"transition uploads the stored representation; the read side is what has to decode it"
);
let (body, published_size) = read_transitioned(&set_disks, bucket, object, None, &opts).await;
assert_eq!(body, plaintext, "a tiered read must return the object's content, not its stored bytes");
assert_eq!(
published_size,
plaintext.len() as i64,
"a tiered read must publish the plaintext size, not the compressed one"
);
}
/// A ranged tiered read is expressed in plaintext coordinates, so the plan
/// has to translate it into the remote copy's compressed extent and skip
/// into the decompressed stream — the same translation the erasure path does.
#[tokio::test]
#[serial_test::serial]
async fn transitioned_compressed_object_range_get_returns_plaintext_slice() {
let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
let bucket = "transitioned-compressed-range-bucket";
let object = "object.txt";
let plaintext = b"ranged reads of transitioned compressed objects must land in plaintext ".repeat(20_000);
let compressed = compress_for_storage(&plaintext).await;
let original = write_compressed_source(&set_disks, &disk_stores, bucket, object, &plaintext, &compressed).await;
let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await;
set_disks
.transition_object(bucket, object, &transition_options(&original, tier_name))
.await
.expect("transition should commit");
let opts = ObjectOptions {
no_lock: true,
..Default::default()
};
// Deliberately past the compressed size, so a range still measured in
// stored coordinates could not produce this slice.
let start = compressed.len() as i64 + 4096;
let end = start + 511;
let range = HTTPRangeSpec {
is_suffix_length: false,
start,
end,
};
let (body, published_size) = read_transitioned(&set_disks, bucket, object, Some(range), &opts).await;
let expected = &plaintext[start as usize..=end as usize];
assert_eq!(body, expected, "a ranged tiered read must return that plaintext slice");
assert_eq!(published_size, expected.len() as i64, "a ranged tiered read publishes the slice length");
}
/// The restore copy-back re-writes the object under its original metadata,
/// which still says "compressed". It therefore has to keep receiving the
/// STORED bytes: `restore_request_active` holds it on the plan's `Plain`
/// branch, and decompressing there would write plaintext under compressed
/// metadata.
#[tokio::test]
#[serial_test::serial]
async fn restore_read_of_transitioned_compressed_object_keeps_stored_bytes() {
let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
let bucket = "transitioned-compressed-restore-bucket";
let object = "object.txt";
let plaintext = b"restore copy-back must keep the stored representation intact ".repeat(20_000);
let compressed = compress_for_storage(&plaintext).await;
let original = write_compressed_source(&set_disks, &disk_stores, bucket, object, &plaintext, &compressed).await;
let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await;
set_disks
.transition_object(bucket, object, &transition_options(&original, tier_name))
.await
.expect("transition should commit");
let oi = set_disks
.get_object_info(
bucket,
object,
&ObjectOptions {
no_lock: true,
..Default::default()
},
)
.await
.expect("transitioned metadata should resolve");
let restore_opts = ObjectOptions {
no_lock: true,
part_number: Some(1),
transition: TransitionOptions {
restore_request: s3s::dto::RestoreRequest {
days: Some(1),
..Default::default()
},
..Default::default()
},
..Default::default()
};
let mut reader = get_transitioned_object_reader_with_tier_manager(
bucket,
object,
&None,
&HeaderMap::new(),
&oi,
&restore_opts,
&set_disks.ctx.tier_config_mgr(),
set_disks.ctx.object_encryption_resolver(),
)
.await
.expect("restore read of the tiered copy should open");
let published_size = reader.object_info.size;
let mut body = Vec::new();
reader.stream.read_to_end(&mut body).await.expect("restore body should drain");
assert_eq!(body, compressed, "a restore read must copy the stored bytes back verbatim");
assert_eq!(
published_size,
compressed.len() as i64,
"a restore read must keep publishing the stored size"
);
}
/// Plain objects must keep streaming the remote bytes through untouched:
/// their plan is `Plain`, so the tiered read stays byte-identical.
#[tokio::test]
#[serial_test::serial]
async fn transitioned_plain_object_get_is_unchanged() {
let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
let bucket = "transitioned-plain-get-bucket";
let object = "object.bin";
let payload = b"plain transitioned objects must keep reading back byte-identical ".repeat(1024);
let original = write_source(&set_disks, &disk_stores, bucket, object, &payload).await;
let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await;
set_disks
.transition_object(bucket, object, &transition_options(&original, tier_name))
.await
.expect("transition should commit");
let opts = ObjectOptions {
no_lock: true,
..Default::default()
};
let (body, published_size) = read_transitioned(&set_disks, bucket, object, None, &opts).await;
assert_eq!(body, payload);
assert_eq!(published_size, payload.len() as i64);
let range = HTTPRangeSpec {
is_suffix_length: false,
start: 100,
end: 611,
};
let (ranged_body, ranged_size) = read_transitioned(&set_disks, bucket, object, Some(range), &opts).await;
assert_eq!(ranged_body, &payload[100..=611]);
assert_eq!(ranged_size, payload.len() as i64, "a plain ranged read keeps publishing the object size");
}
async fn corrupt_beyond_read_quorum(
temp_dirs: &[tempfile::TempDir],
bucket: &str,
+8 -6
View File
@@ -116,6 +116,7 @@ impl SetDisks {
.then_some(GET_METADATA_CACHE_REASON_DIST_ERASURE)
}
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
async fn cached_get_object_fileinfo(&self, bucket: &str, object: &str) -> Option<Arc<GetObjectMetadataCacheEntry>> {
match self.lookup_cached_get_object_fileinfo(bucket, object).await {
MetadataCacheLookup::Hit(entry) => Some(entry),
@@ -1826,6 +1827,7 @@ fn get_object_metadata_cache_request_bypass_reason(bucket: &str, opts: &ObjectOp
.then_some(GET_METADATA_CACHE_REASON_META_BUCKET)
}
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
fn is_get_object_metadata_cache_request_eligible(bucket: &str, opts: &ObjectOptions, read_data: bool) -> bool {
get_object_metadata_cache_request_bypass_reason(bucket, opts, read_data).is_none()
}
@@ -3935,7 +3937,7 @@ mod tests {
}
#[test]
fn metadata_early_stop_bounded_fanout_defaults_to_disabled() {
fn metadata_early_stop_bounded_fanout_defaults_to_enabled() {
temp_env::with_vars(
[
(ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE, Some("true")),
@@ -3944,20 +3946,20 @@ mod tests {
],
|| {
assert!(is_get_metadata_data_read_early_stop_enabled());
assert!(!is_get_metadata_early_stop_bounded_fanout_enabled());
assert!(is_get_metadata_early_stop_bounded_fanout_enabled());
},
);
temp_env::with_vars([(ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT, Some("true"))], || {
assert!(is_get_metadata_early_stop_bounded_fanout_enabled());
temp_env::with_vars([(ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT, Some("false"))], || {
assert!(!is_get_metadata_early_stop_bounded_fanout_enabled());
});
temp_env::with_vars(
[
(ENV_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE, Some("false")),
(ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT, Some("false")),
(ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT, Some("true")),
],
|| {
assert!(!is_get_metadata_data_read_early_stop_enabled());
assert!(!is_get_metadata_early_stop_bounded_fanout_enabled());
assert!(is_get_metadata_early_stop_bounded_fanout_enabled());
},
);
}
+33 -30
View File
@@ -487,22 +487,21 @@ pub fn record_get_object_completion(total_duration_secs: f64, response_size_byte
/// Record the streaming strategy chosen for a GetObject response body.
#[inline(always)]
pub fn record_get_object_stream_strategy(strategy: &str, buffer_size_bytes: usize, response_size_bytes: i64) {
pub fn record_get_object_stream_strategy(strategy: &'static str, buffer_size_bytes: usize, response_size_bytes: i64) {
if !get_stage_metrics_enabled() {
return;
}
counter!("rustfs_io_get_object_stream_strategy_total", "strategy" => strategy.to_string()).increment(1);
histogram!("rustfs_io_get_object_stream_buffer_size_bytes", "strategy" => strategy.to_string())
.record(usize_to_f64(buffer_size_bytes));
histogram!("rustfs_io_get_object_stream_response_size_bytes", "strategy" => strategy.to_string())
counter!("rustfs_io_get_object_stream_strategy_total", "strategy" => strategy).increment(1);
histogram!("rustfs_io_get_object_stream_buffer_size_bytes", "strategy" => strategy).record(usize_to_f64(buffer_size_bytes));
histogram!("rustfs_io_get_object_stream_response_size_bytes", "strategy" => strategy)
.record(i64_non_negative_to_f64(response_size_bytes));
}
/// Record the response-body handoff shape from a GetObject reader into the S3 streaming body.
#[inline(always)]
pub fn record_get_object_response_handoff(
strategy: &str,
buffer_source: &str,
strategy: &'static str,
buffer_source: &'static str,
buffer_size_bytes: usize,
response_size_bytes: i64,
duration_secs: f64,
@@ -512,26 +511,26 @@ pub fn record_get_object_response_handoff(
}
counter!(
"rustfs_io_get_object_response_handoff_total",
"strategy" => strategy.to_string(),
"buffer_source" => buffer_source.to_string()
"strategy" => strategy,
"buffer_source" => buffer_source
)
.increment(1);
histogram!(
"rustfs_io_get_object_response_handoff_buffer_size_bytes",
"strategy" => strategy.to_string(),
"buffer_source" => buffer_source.to_string()
"strategy" => strategy,
"buffer_source" => buffer_source
)
.record(usize_to_f64(buffer_size_bytes));
histogram!(
"rustfs_io_get_object_response_handoff_response_size_bytes",
"strategy" => strategy.to_string(),
"buffer_source" => buffer_source.to_string()
"strategy" => strategy,
"buffer_source" => buffer_source
)
.record(i64_non_negative_to_f64(response_size_bytes));
histogram!(
"rustfs_io_get_object_response_handoff_duration_seconds",
"strategy" => strategy.to_string(),
"buffer_source" => buffer_source.to_string()
"strategy" => strategy,
"buffer_source" => buffer_source
)
.record(duration_secs);
record_get_object_response_handoff_duration("s3_handler", duration_secs);
@@ -539,14 +538,18 @@ pub fn record_get_object_response_handoff(
/// Record ReaderStream capacity chosen for GetObject handoff.
#[inline(always)]
pub fn record_get_object_reader_stream_buffer_size(strategy: &str, buffer_source: &str, buffer_size_bytes: usize) {
pub fn record_get_object_reader_stream_buffer_size(
strategy: &'static str,
buffer_source: &'static str,
buffer_size_bytes: usize,
) {
if !get_stage_metrics_enabled() {
return;
}
histogram!(
"rustfs_io_get_object_reader_stream_buffer_size_bytes",
"strategy" => strategy.to_string(),
"buffer_source" => buffer_source.to_string()
"strategy" => strategy,
"buffer_source" => buffer_source
)
.record(usize_to_f64(buffer_size_bytes));
}
@@ -554,8 +557,8 @@ pub fn record_get_object_reader_stream_buffer_size(strategy: &str, buffer_source
/// Record ReaderStream poll outcomes for GetObject handoff attribution.
#[inline(always)]
pub fn record_get_object_reader_stream_poll(
strategy: &str,
buffer_source: &str,
strategy: &'static str,
buffer_source: &'static str,
outcome: &'static str,
remaining_before: usize,
bytes: usize,
@@ -567,36 +570,36 @@ pub fn record_get_object_reader_stream_poll(
let bytes = u64::try_from(bytes).unwrap_or(u64::MAX);
counter!(
"rustfs_io_get_object_reader_stream_poll_total",
"strategy" => strategy.to_string(),
"buffer_source" => buffer_source.to_string(),
"strategy" => strategy,
"buffer_source" => buffer_source,
"outcome" => outcome
)
.increment(1);
counter!(
"rustfs_io_get_object_reader_stream_poll_bytes_total",
"strategy" => strategy.to_string(),
"buffer_source" => buffer_source.to_string(),
"strategy" => strategy,
"buffer_source" => buffer_source,
"outcome" => outcome
)
.increment(bytes);
histogram!(
"rustfs_io_get_object_reader_stream_poll_remaining_bytes",
"strategy" => strategy.to_string(),
"buffer_source" => buffer_source.to_string(),
"strategy" => strategy,
"buffer_source" => buffer_source,
"outcome" => outcome
)
.record(usize_to_f64(remaining_before));
histogram!(
"rustfs_io_get_object_reader_stream_poll_bytes",
"strategy" => strategy.to_string(),
"buffer_source" => buffer_source.to_string(),
"strategy" => strategy,
"buffer_source" => buffer_source,
"outcome" => outcome
)
.record(usize_to_f64(bytes as usize));
histogram!(
"rustfs_io_get_object_reader_stream_poll_duration_seconds",
"strategy" => strategy.to_string(),
"buffer_source" => buffer_source.to_string(),
"strategy" => strategy,
"buffer_source" => buffer_source,
"outcome" => outcome
)
.record(duration_secs);
+4 -1
View File
@@ -31,7 +31,10 @@ pub struct KeystoneClient {
admin_password: Option<String>,
admin_project: Option<String>,
admin_domain: String,
#[allow(dead_code)]
#[allow(
dead_code,
reason = "TLS verification flag parsed from config; the reqwest client is built before it is consulted, so nothing reads it back (backlog#1823)"
)]
verify_ssl: bool,
/// Request timeout applied to the underlying HTTP client.
timeout: std::time::Duration,
+4 -1
View File
@@ -20,7 +20,10 @@ use tracing::{debug, info};
/// Maps Keystone identities to RustFS concepts
pub struct KeystoneIdentityMapper {
#[allow(dead_code)]
#[allow(
dead_code,
reason = "keeps the Keystone client alive for the mapper's lifetime; the mapping paths do not call through it yet (backlog#1823)"
)]
client: Arc<KeystoneClient>,
role_policy_map: HashMap<String, String>,
enable_tenant_prefix: bool,
-2
View File
@@ -18,8 +18,6 @@
//! data encryption keys using master keys. It abstracts the encryption
//! operations so that different backends can share the same encryption logic.
#![allow(dead_code)] // Trait methods may be used by implementations
use crate::error::{KmsError, Result};
use crate::persisted_observability::{BoundedUnknownFieldName, UnknownFieldSummary};
use async_trait::async_trait;
+4 -1
View File
@@ -40,7 +40,10 @@ impl RuleEvents for RuleView {
#[derive(Debug)]
struct CompiledRules {
// Keep RulesMap (can be used later if you want to make more complex judgments during the snapshot reading phase)
#[allow(dead_code)]
#[allow(
dead_code,
reason = "speculative retention: the comment above keeps it for richer snapshot-time judgements that no code performs yet (backlog#1823)"
)]
rules_map: RulesMap,
// for RulesContainer::iter_rules
rule_views: Vec<RuleView>,
-3
View File
@@ -187,7 +187,6 @@ impl RulesMap {
/// # Parameters
/// * `event_name` - The EventName from which to remove the rule.
/// * `pattern` - The pattern of the rule to be removed.
#[allow(dead_code)]
pub fn remove_rule(&mut self, event_name: &EventName, pattern: &str) {
let mut remove_event = false;
@@ -209,7 +208,6 @@ impl RulesMap {
///
/// # Parameters
/// * `event_names` - A slice of EventNames to be removed.
#[allow(dead_code)]
pub fn remove_rules(&mut self, event_names: &[EventName]) {
for event_name in event_names {
self.map.remove(event_name);
@@ -223,7 +221,6 @@ impl RulesMap {
/// * `event_name` - The EventName to update.
/// * `pattern` - The pattern of the rule to be updated.
/// * `target_id` - The TargetID to be added.
#[allow(dead_code)]
pub fn update_rule(&mut self, event_name: EventName, pattern: String, target_id: TargetID) {
self.map.entry(event_name).or_default().add(pattern, target_id);
self.total_events_mask |= event_name.mask(); // Update only the relevant bitmask
-6
View File
@@ -18,12 +18,6 @@ use rustfs_targets::arn::TargetID;
/// TargetIDSet - A collection representation of TargetID.
pub type TargetIdSet = HashSet<TargetID>;
/// Provides a Go-like method for TargetIdSet (can be implemented as trait if needed)
#[allow(dead_code)]
pub(crate) fn new_target_id_set(target_ids: Vec<TargetID>) -> TargetIdSet {
target_ids.into_iter().collect()
}
// HashSet has built-in clone, union, difference and other operations.
// But the Go version of the method returns a new Set, and the HashSet method is usually iterator or modify itself.
// If you need to exactly match Go's API style, you can add wrapper functions.
-1
View File
@@ -17,7 +17,6 @@ use std::time::Duration;
/// Environment variable key for the global default metrics interval (seconds).
pub const ENV_DEFAULT_METRICS_INTERVAL: &str = "RUSTFS_METRICS_DEFAULT_INTERVAL_SEC";
/// Default interval for metrics collection if not specified otherwise.
#[allow(dead_code)]
pub const DEFAULT_METRICS_INTERVAL: Duration = Duration::from_secs(60);
/// Environment variable key for cluster metrics interval (seconds).
-3
View File
@@ -145,21 +145,18 @@ impl PrometheusMetric {
}
#[inline]
#[allow(dead_code)]
pub fn with_label(mut self, key: &'static str, value: impl Into<Cow<'static, str>>) -> Self {
self.labels.push((key, value.into()));
self
}
#[inline]
#[allow(dead_code)]
pub fn with_label_owned(mut self, key: &'static str, value: String) -> Self {
self.labels.push((key, Cow::Owned(value)));
self
}
#[inline]
#[allow(dead_code)]
pub fn with_labels(mut self, labels: Vec<(&'static str, Cow<'static, str>)>) -> Self {
self.labels = labels;
self
@@ -16,7 +16,6 @@ use crate::{MetricName, MetricNamespace, MetricSubsystem, MetricType};
use std::collections::HashSet;
/// MetricDescriptor - Metric descriptors
#[allow(dead_code)]
#[derive(Debug, Clone)]
pub struct MetricDescriptor {
pub name: MetricName,
@@ -52,7 +51,6 @@ impl MetricDescriptor {
}
/// Get the full metric name in Prometheus style: <namespace>_<subsystem>_<name>
#[allow(dead_code)]
pub fn get_full_metric_name(&self) -> String {
let namespace = self.namespace.as_str();
let formatted_subsystem = self.subsystem.as_str();
@@ -61,7 +59,6 @@ impl MetricDescriptor {
}
/// check whether the label is in the label set
#[allow(dead_code)]
pub fn has_label(&mut self, label: &str) -> bool {
self.get_label_set().contains(label)
}
@@ -13,7 +13,6 @@
// limitations under the License.
/// The metric name is the individual name of the metric
#[allow(dead_code)]
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum MetricName {
// The generic metric name
@@ -443,7 +442,6 @@ pub enum MetricName {
}
impl MetricName {
#[allow(dead_code)]
pub fn as_str(&self) -> String {
match self {
Self::AuthTotal => "auth_total".to_string(),
@@ -13,7 +13,6 @@
// limitations under the License.
/// MetricType - Indicates the type of indicator
#[allow(dead_code)]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum MetricType {
Counter,
@@ -23,7 +22,6 @@ pub enum MetricType {
impl MetricType {
/// convert the metric type to a string representation
#[allow(dead_code)]
pub fn as_str(&self) -> &'static str {
match self {
Self::Counter => "counter",
@@ -34,7 +32,6 @@ impl MetricType {
/// Convert the metric type to the Prometheus value type
/// In a Rust implementation, this might return the corresponding Prometheus Rust client type
#[allow(dead_code)]
pub fn as_prom(&self) -> &'static str {
match self {
Self::Counter => "counter.",
@@ -56,7 +56,6 @@ pub fn new_gauge_md(
}
/// create a new histogram indicator descriptor
#[allow(dead_code)]
pub fn new_histogram_md(
name: impl Into<MetricName>,
help: impl Into<String>,
@@ -19,7 +19,6 @@ pub enum MetricNamespace {
}
impl MetricNamespace {
#[allow(dead_code)]
pub fn as_str(&self) -> &'static str {
match self {
Self::RustFS => "rustfs",
@@ -14,7 +14,6 @@
/// Format the path to the metric name format
/// Replace '/' and '-' with '_'
#[allow(dead_code)]
pub fn format_path_to_metric_name(path: &str) -> String {
path.trim_start_matches('/').replace(['/', '-'], "_")
}
@@ -102,7 +102,6 @@ impl MetricSubsystem {
}
/// Get the formatted metric name format string
#[allow(dead_code)]
pub fn as_str(&self) -> String {
format_path_to_metric_name(self.path())
}
@@ -151,7 +150,6 @@ impl MetricSubsystem {
}
/// A convenient way to create custom subsystems directly
#[allow(dead_code)]
pub fn new(path: impl Into<String>) -> Self {
Self::Custom(path.into())
}
@@ -176,7 +174,6 @@ impl std::fmt::Display for MetricSubsystem {
}
}
#[allow(dead_code)]
pub mod subsystems {
use super::MetricSubsystem;
+4 -1
View File
@@ -38,7 +38,10 @@ pub enum Rotation {
Minutely,
Hourly,
Daily,
#[allow(dead_code)]
#[allow(
dead_code,
reason = "constructed only by this file's rolling-appender tests; the lib target cannot see them (backlog#1823)"
)]
Never,
}
-4
View File
@@ -219,10 +219,6 @@ impl PartialEq for Functions {
}
}
#[derive(Clone, Serialize, Deserialize)]
#[allow(dead_code)]
pub struct Value;
#[cfg(test)]
mod tests {
use crate::policy::Functions;
+4 -2
View File
@@ -12,7 +12,6 @@
// See the License for the specific language governing permissions and
// limitations under the License.
#[allow(dead_code)]
pub fn is_simple_match<P, N>(pattern: P, name: N) -> bool
where
P: AsRef<str>,
@@ -29,7 +28,10 @@ where
inner_match(pattern, name, false)
}
#[allow(dead_code)]
#[allow(
dead_code,
reason = "prefix-matcher asserted by this file's tests; no production caller yet (backlog#1823)"
)]
pub fn is_match_as_pattern_prefix<P, N>(pattern: P, text: N) -> bool
where
P: AsRef<str>,
-2
View File
@@ -38,7 +38,6 @@ use std::collections::HashMap;
/// - Account format is invalid
/// - Credentials don't contain project_id
/// - Account project_id doesn't match credentials project_id
#[allow(dead_code)] // Used by Swift implementation
pub fn validate_account_access(account: &str, credentials: &Credentials) -> SwiftResult<String> {
// Extract project_id from account (strip "AUTH_" prefix)
let account_project_id = account
@@ -70,7 +69,6 @@ pub fn validate_account_access(account: &str, credentials: &Credentials) -> Swif
///
/// Admin users (with "admin" or "reseller_admin" roles) can perform
/// cross-tenant operations and administrative tasks.
#[allow(dead_code)] // Used by Swift implementation
pub fn is_admin_user(credentials: &Credentials) -> bool {
credentials
.claims
-14
View File
@@ -144,7 +144,6 @@ impl ContainerMapper {
/// - S3 bucket name compatible (only uses [a-z0-9-])
/// - Deterministic mapping (same input always produces same bucket name)
/// - Fixed-length prefix (16 hex chars = 8 bytes)
#[allow(dead_code)] // Used in: create/delete container operations
pub fn swift_to_s3_bucket(&self, container: &str, project_id: &str) -> String {
if self.config.tenant_prefix_enabled {
let hash = self.hash_project_id(project_id);
@@ -216,7 +215,6 @@ pub fn bucket_info_to_container(info: &BucketInfo, mapper: &ContainerMapper, pro
/// 2. Lists all S3 buckets
/// 3. Filters to buckets belonging to this tenant (using tenant prefix)
/// 4. Converts BucketInfo to Swift Container format
#[allow(dead_code)] // Used by handler: list containers
pub async fn list_containers(account: &str, credentials: &Credentials) -> SwiftResult<Vec<Container>> {
// Validate account access and extract project_id
let project_id = validate_account_access(account, credentials)?;
@@ -279,7 +277,6 @@ pub async fn list_containers(account: &str, credentials: &Credentials) -> SwiftR
/// - Returns 201 Created on success
/// - Returns 202 Accepted if container already exists
/// - Returns 400 Bad Request for invalid container names
#[allow(dead_code)] // Used by handler
pub async fn create_container(account: &str, container: &str, credentials: &Credentials) -> SwiftResult<bool> {
// Validate account access and extract project_id
let project_id = validate_account_access(account, credentials)?;
@@ -348,7 +345,6 @@ fn validate_container_name(container: &str) -> SwiftResult<()> {
}
/// Container metadata for HEAD response
#[allow(dead_code)] // TODO: Remove once Swift API integration is complete
#[derive(Debug, Clone)]
pub struct ContainerMetadata {
/// Number of objects in container
@@ -411,7 +407,6 @@ pub(crate) async fn get_container_custom_metadata(
/// - HEAD /v1/{account}/{container} returns container metadata
/// - Returns 204 No Content on success with headers
/// - Returns 404 Not Found if container doesn't exist
#[allow(dead_code)] // Used by handler
pub async fn get_container_metadata(account: &str, container: &str, credentials: &Credentials) -> SwiftResult<ContainerMetadata> {
let (bucket_name, bucket_info, custom_metadata) = get_container_metadata_base(account, container, credentials).await?;
@@ -448,7 +443,6 @@ pub async fn get_container_metadata(account: &str, container: &str, credentials:
/// - The update is additive: items the request does not name keep their stored
/// value, and removal is explicit, via `X-Remove-Container-Meta-{name}` or an
/// empty value
#[allow(dead_code)] // Used by handler
pub async fn update_container_metadata(
account: &str,
container: &str,
@@ -520,7 +514,6 @@ pub async fn update_container_metadata(
/// - Returns 204 No Content on success
/// - Returns 404 Not Found if container doesn't exist
/// - Returns 409 Conflict if container is not empty
#[allow(dead_code)] // Used by handler
pub async fn delete_container(account: &str, container: &str, credentials: &Credentials) -> SwiftResult<()> {
// Validate account access and extract project_id
let project_id = validate_account_access(account, credentials)?;
@@ -603,7 +596,6 @@ pub async fn delete_container(account: &str, container: &str, credentials: &Cred
/// - Account validation fails
/// - Container doesn't exist
/// - Storage layer errors occur
#[allow(dead_code)] // Handler integration: GET container
pub async fn list_objects(
account: &str,
container: &str,
@@ -706,7 +698,6 @@ pub async fn list_objects(
/// Versioning configuration is stored as an S3 bucket tag:
/// - Tag key: `swift-versions-location`
/// - Tag value: archive container name
#[allow(dead_code)] // Used by handler
pub async fn enable_versioning(
account: &str,
container: &str,
@@ -795,7 +786,6 @@ pub async fn enable_versioning(
/// * `account` - Account identifier
/// * `container` - Container name to disable versioning on
/// * `credentials` - Keystone credentials
#[allow(dead_code)] // Used by handler
pub async fn disable_versioning(account: &str, container: &str, credentials: &Credentials) -> SwiftResult<()> {
// Validate account access
let project_id = validate_account_access(account, credentials)?;
@@ -855,7 +845,6 @@ pub async fn disable_versioning(account: &str, container: &str, credentials: &Cr
/// # Returns
/// - Some(archive_container_name) if versioning is enabled
/// - None if versioning is not enabled
#[allow(dead_code)] // Used by handler and object.rs
pub async fn get_versions_location(account: &str, container: &str, credentials: &Credentials) -> SwiftResult<Option<String>> {
// Validate account access
let project_id = validate_account_access(account, credentials)?;
@@ -918,7 +907,6 @@ pub async fn get_versions_location(account: &str, container: &str, credentials:
/// &credentials
/// ).await?;
/// ```
#[allow(dead_code)] // Used by handler
pub async fn set_container_acl(
account: &str,
container: &str,
@@ -1022,7 +1010,6 @@ pub async fn set_container_acl(
/// println!("Container is publicly readable");
/// }
/// ```
#[allow(dead_code)] // Used by handler
pub async fn get_container_acl(
account: &str,
container: &str,
@@ -1083,7 +1070,6 @@ pub async fn get_container_acl(
///
/// # Returns
/// Ok(()) if ACLs were deleted successfully
#[allow(dead_code)] // Used by handler
pub async fn delete_container_acl(account: &str, container: &str, credentials: &Credentials) -> SwiftResult<()> {
// Setting both ACLs to None removes them
set_container_acl(account, container, None, None, credentials).await
-1
View File
@@ -20,7 +20,6 @@ use std::fmt;
/// Swift-specific error type
#[derive(Debug)]
#[allow(dead_code)] // Error variants used by Swift implementation
pub enum SwiftError {
/// 400 Bad Request
BadRequest(String),
-21
View File
@@ -122,12 +122,10 @@ fn swift_user_metadata(headers: &HeaderMap) -> Option<HashMap<String, String>> {
///
/// Handles URL encoding/decoding and path normalization for Swift object keys.
/// Swift object names can contain any UTF-8 characters except null bytes.
#[allow(dead_code)] // Used in: object operations
pub struct ObjectKeyMapper;
impl ObjectKeyMapper {
/// Create a new object key mapper
#[allow(dead_code)] // Used in: object operations
pub fn new() -> Self {
Self
}
@@ -140,7 +138,6 @@ impl ObjectKeyMapper {
/// - Not contain null bytes
/// - Not contain '..' path segments (directory traversal)
/// - Not start with '/' (leading slash handled by routing)
#[allow(dead_code)] // Used in: object operations
pub fn validate_object_name(object: &str) -> SwiftResult<()> {
if object.is_empty() {
return Err(SwiftError::BadRequest("Object name cannot be empty".to_string()));
@@ -183,7 +180,6 @@ impl ObjectKeyMapper {
/// Example:
/// - Swift: "photos/vacation/beach photo.jpg"
/// - S3: "photos/vacation/beach photo.jpg"
#[allow(dead_code)] // Used in: object operations
pub fn swift_to_s3_key(object: &str) -> SwiftResult<String> {
Self::validate_object_name(object)?;
Ok(object.to_string())
@@ -193,7 +189,6 @@ impl ObjectKeyMapper {
///
/// This is essentially an identity transformation since we store
/// Swift object names as-is in S3.
#[allow(dead_code)] // Used in: object operations
pub fn s3_to_swift_name(key: &str) -> String {
key.to_string()
}
@@ -208,7 +203,6 @@ impl ObjectKeyMapper {
/// - Object: "vacation/beach.jpg"
/// - Bucket: "abc123:photos"
/// - Key: "vacation/beach.jpg"
#[allow(dead_code)] // Used in: object operations
pub fn build_s3_key(object: &str) -> SwiftResult<String> {
Self::swift_to_s3_key(object)
}
@@ -220,7 +214,6 @@ impl ObjectKeyMapper {
///
/// Example URL: /v1/AUTH_abc/container/path%2Fto%2Ffile.txt
/// Decoded: "path/to/file.txt"
#[allow(dead_code)] // Used in: object operations
pub fn decode_object_from_url(encoded: &str) -> SwiftResult<String> {
// Decode percent-encoding
let decoded = urlencoding::decode(encoded).map_err(|e| SwiftError::BadRequest(format!("Invalid URL encoding: {}", e)))?;
@@ -233,7 +226,6 @@ impl ObjectKeyMapper {
///
/// When constructing URLs (e.g., for redirect responses), we need to
/// percent-encode object names.
#[allow(dead_code)] // Used in: object operations
pub fn encode_object_for_url(object: &str) -> String {
urlencoding::encode(object).to_string()
}
@@ -241,7 +233,6 @@ impl ObjectKeyMapper {
/// Check if object name represents a directory (pseudo-directory)
///
/// In Swift, objects ending with '/' are treated as directory markers.
#[allow(dead_code)] // Used in: object operations
pub fn is_directory_marker(object: &str) -> bool {
object.ends_with('/')
}
@@ -250,7 +241,6 @@ impl ObjectKeyMapper {
///
/// Removes redundant slashes and normalizes the path while preserving
/// trailing slashes for directory markers.
#[allow(dead_code)] // Used in: object operations
pub fn normalize_path(object: &str) -> String {
// Split by '/', filter out empty segments (except if it's the end)
let has_trailing_slash = object.ends_with('/');
@@ -324,7 +314,6 @@ fn sanitize_storage_error<E: std::fmt::Display>(operation: &str, error: E) -> Sw
/// # Returns
/// * `Ok(etag)` - Object ETag on success
/// * `Err(SwiftError)` - Error if validation fails or upload fails
#[allow(dead_code)] // Handler integration: PUT object
pub async fn put_object<R>(
account: &str,
container: &str,
@@ -445,7 +434,6 @@ where
///
/// Similar to put_object, but allows directly specifying metadata instead of extracting from headers.
/// This is used internally for storing SLO manifests and marker objects.
#[allow(dead_code)] // Used by SLO implementation
pub async fn put_object_with_metadata<R>(
account: &str,
container: &str,
@@ -549,7 +537,6 @@ where
/// - `bytes=1000-1999` - Bytes 1000-1999
/// - `bytes=1000-` - From byte 1000 to end
/// - `bytes=-500` - Last 500 bytes
#[allow(dead_code)] // Handler integration: GET object
pub async fn get_object(
account: &str,
container: &str,
@@ -608,7 +595,6 @@ pub async fn get_object(
/// # Returns
/// * `Ok(object_info)` - Object metadata (ObjectInfo)
/// * `Err(SwiftError)` - Error if validation fails or object not found
#[allow(dead_code)] // Handler integration: HEAD object
pub async fn head_object(
account: &str,
container: &str,
@@ -671,7 +657,6 @@ pub async fn head_object(
/// # Returns
/// * `Ok(())` - Object deleted successfully (or didn't exist)
/// * `Err(SwiftError)` - Error if validation fails or deletion fails
#[allow(dead_code)] // Handler integration: DELETE object
pub async fn delete_object(account: &str, container: &str, object: &str, credentials: &Credentials) -> SwiftResult<()> {
// 1. Validate account access and get project_id
let project_id = validate_account_access(account, credentials)?;
@@ -732,7 +717,6 @@ pub async fn delete_object(account: &str, container: &str, object: &str, credent
/// # Returns
/// * `Ok(())` - Metadata updated successfully
/// * `Err(SwiftError)` - Error if validation fails, object not found, or update fails
#[allow(dead_code)] // Handler integration: POST object
pub async fn update_object_metadata(
account: &str,
container: &str,
@@ -846,7 +830,6 @@ pub async fn update_object_metadata(
/// # Handler Integration Note
/// The current handler architecture needs to be updated to pass headers through
/// to support COPY method and X-Copy-From header detection. See handler.rs for details.
#[allow(dead_code)] // Handler integration: COPY object
#[allow(clippy::too_many_arguments)] // Necessary for full copy functionality
pub async fn copy_object(
src_account: &str,
@@ -979,7 +962,6 @@ pub async fn copy_object(
/// assert_eq!(container, "my-container");
/// assert_eq!(object, "path/to/file.txt");
/// ```
#[allow(dead_code)] // Handler integration: COPY method
pub fn parse_destination_header(destination: &str) -> SwiftResult<(String, String)> {
let destination = destination.trim_start_matches('/');
let parts: Vec<&str> = destination.splitn(2, '/').collect();
@@ -1013,7 +995,6 @@ pub fn parse_destination_header(destination: &str) -> SwiftResult<(String, Strin
/// # Returns
/// * `Ok((container, object))` - Parsed container and object names
/// * `Err(SwiftError)` - Error if format is invalid
#[allow(dead_code)] // Handler integration: X-Copy-From
pub fn parse_copy_from_header(copy_from: &str) -> SwiftResult<(String, String)> {
// Same parsing logic as Destination header
parse_destination_header(copy_from)
@@ -1042,7 +1023,6 @@ pub fn parse_copy_from_header(copy_from: &str) -> SwiftResult<(String, String)>
/// assert_eq!(range.start, 0);
/// assert_eq!(range.end, 1023);
/// ```
#[allow(dead_code)] // Handler integration: Range header
pub fn parse_range_header(range_str: &str) -> SwiftResult<HTTPRangeSpec> {
if !range_str.starts_with("bytes=") {
return Err(SwiftError::BadRequest("Range header must start with 'bytes='".to_string()));
@@ -1124,7 +1104,6 @@ pub fn parse_range_header(range_str: &str) -> SwiftResult<HTTPRangeSpec> {
/// let header = format_content_range(0, 1023, 5000);
/// assert_eq!(header, "bytes 0-1023/5000");
/// ```
#[allow(dead_code)] // Handler integration: Range header
pub fn format_content_range(start: i64, end: i64, total: i64) -> String {
format!("bytes {}-{}/{}", start, end, total)
}
-2
View File
@@ -50,7 +50,6 @@ pub enum SwiftRoute {
impl SwiftRoute {
/// Get the account identifier from the route
#[allow(dead_code)] // Public API for future use
pub fn account(&self) -> &str {
match self {
SwiftRoute::Account { account, .. } => account,
@@ -60,7 +59,6 @@ impl SwiftRoute {
}
/// Extract project_id from account string (removes AUTH_ prefix)
#[allow(dead_code)] // Public API for future use
pub fn project_id(&self) -> Option<&str> {
let account = self.account();
ACCOUNT_PATTERN
-3
View File
@@ -19,7 +19,6 @@ use std::collections::HashMap;
/// Swift container metadata
#[derive(Debug, Clone, Serialize, Deserialize)]
#[allow(dead_code)] // Used in container listing operations
pub struct Container {
/// Container name
pub name: String,
@@ -34,7 +33,6 @@ pub struct Container {
/// Swift object metadata
#[derive(Debug, Clone, Serialize, Deserialize)]
#[allow(dead_code)] // Used in object listing operations
pub struct Object {
/// Object name (key)
pub name: String,
@@ -50,7 +48,6 @@ pub struct Object {
/// Swift metadata extracted from headers
#[derive(Debug, Clone, Default)]
#[allow(dead_code)] // Used by Swift implementation
pub struct SwiftMetadata {
/// Custom metadata key-value pairs (from X-Container-Meta-* or X-Object-Meta-*)
pub metadata: HashMap<String, String>,
@@ -34,14 +34,6 @@ MINIO_LAB_PYTHON_IMAGE=public.ecr.aws/docker/library/python:3.12-slim \
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.
## Capturing the SSE-C cases
MinIO refuses SSE-C over a plain-HTTP connection, so the `sse-c-*` cases cannot be captured against the default endpoint — `./capture_via_docker.sh all` fails on the first SSE-C upload with `InvalidRequest ... must be made over a secure connection`. The lab provisions its own self-signed certificate; point it at the HTTPS endpoint to capture them:
```bash
MINIO_LAB_ENDPOINT=https://127.0.0.1:9000 ./capture_via_docker.sh all
```
## Layout
The default root is `artifacts/minio-fixture-lab`, which is already ignored by the repository.
@@ -12,13 +12,6 @@
# # ignored interop tests consume
# ./capture_via_docker.sh sse-s3-singlepart-64k # specific case id(s)
# ./capture_via_docker.sh all # full SSE/size matrix
#
# The SSE-C cases are not reachable over the default plain-HTTP endpoint: MinIO
# refuses SSE-C unless the connection is secure ("Requests specifying Server
# Side Encryption with Customer provided keys must be made over a secure
# connection"). Capture those by pointing the lab at its self-signed HTTPS
# endpoint, which it provisions itself:
# MINIO_LAB_ENDPOINT=https://127.0.0.1:9000 ./capture_via_docker.sh all
set -euo pipefail
IMAGE="${MINIO_LAB_IMAGE:-rustfs-minio-lab:latest}"
@@ -53,10 +46,7 @@ if [ -n "${MINIO_LAB_PYTHON_IMAGE:-}" ]; then
fi
echo ">> building ${IMAGE}"
# ${arr[@]+"${arr[@]}"} rather than "${arr[@]}": under `set -u`, bash 3.2 —
# still the default /bin/bash on macOS — treats an empty array expansion as an
# unbound variable and aborts.
docker build -f "${SCRIPT_DIR}/Dockerfile" -t "${IMAGE}" ${build_args[@]+"${build_args[@]}"} "${SCRIPT_DIR}"
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}" \
@@ -64,7 +54,6 @@ docker run --rm -v "${REPO_ROOT}:/repo" "${IMAGE}" \
--root "/repo/${FIXTURE_REL}" \
--work-root /tmp/minio-lab-work \
--minio-binary /usr/local/bin/minio \
--endpoint "${MINIO_LAB_ENDPOINT:-http://127.0.0.1:9000}" \
${case_args[@]+"${case_args[@]}"}
"${case_args[@]}"
echo ">> done — fixtures under ${REPO_ROOT}/${FIXTURE_REL}/cases/"
+3 -1
View File
@@ -4621,6 +4621,8 @@ mod tests {
let _subscriber_guard = tracing::subscriber::set_default(subscriber);
let (mut scanner, temp_dir) = build_test_scanner().await;
// Canonicalize for the "drive" field comparison (scanner resolves symlinks).
let canonical_temp_dir = std::fs::canonicalize(&temp_dir).unwrap_or_else(|_| temp_dir.clone());
let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir.clone());
let object_dir = temp_dir.join("bucket").join("object");
@@ -4689,7 +4691,7 @@ mod tests {
let fields = &events[0]["fields"];
assert_eq!(fields["component"], LOG_COMPONENT_SCANNER);
assert_eq!(fields["subsystem"], LOG_SUBSYSTEM_FOLDER);
assert_eq!(fields["drive"], temp_dir.to_string_lossy().as_ref());
assert_eq!(fields["drive"], canonical_temp_dir.to_string_lossy().as_ref());
assert_eq!(fields["bucket"], "bucket");
assert_eq!(fields["object"], "object");
assert_eq!(fields["metadata_path"], metadata_path.to_string_lossy().as_ref());
@@ -27,6 +27,10 @@ use crate::CloudMetadataFetcher;
#[derive(Debug, Clone)]
pub struct AwsMetadataFetcher {
client: Client,
#[allow(
dead_code,
reason = "IMDS endpoint retained beside the client it configures; requests build their own URLs (backlog#1823)"
)]
metadata_endpoint: String,
}
@@ -46,55 +50,6 @@ impl AwsMetadataFetcher {
metadata_endpoint: "http://169.254.169.254".to_string(),
}
}
/// Retrieves an IMDSv2 token for secure metadata access.
#[allow(dead_code)]
async fn get_metadata_token(&self) -> Result<String, AppError> {
let url = format!("{}/latest/api/token", self.metadata_endpoint);
match self
.client
.put(&url)
.header("X-aws-ec2-metadata-token-ttl-seconds", "21600")
.send()
.await
{
Ok(response) => {
if response.status().is_success() {
let token = response
.text()
.await
.map_err(|e| AppError::cloud(format!("Failed to read IMDSv2 token: {}", e)))?;
Ok(token)
} else {
debug!(
event = "trusted_proxies.cloud_metadata",
component = "trusted_proxies",
subsystem = "aws_metadata",
provider = "aws",
operation = "imdsv2_token",
result = "http_error",
status = %response.status(),
"trusted proxy cloud metadata request failed"
);
Err(AppError::cloud("Failed to obtain IMDSv2 token"))
}
}
Err(e) => {
debug!(
event = "trusted_proxies.cloud_metadata",
component = "trusted_proxies",
subsystem = "aws_metadata",
provider = "aws",
operation = "imdsv2_token",
result = "request_failed",
error = %e,
"trusted proxy cloud metadata request failed"
);
Err(AppError::cloud(format!("IMDSv2 request failed: {}", e)))
}
}
}
}
#[async_trait]
-1
View File
@@ -68,7 +68,6 @@ pub fn is_env_set(key: &str) -> bool {
}
/// Returns a list of all proxy-related environment variables and their current values.
#[allow(dead_code)]
pub fn get_all_proxy_env_vars() -> Vec<(String, String)> {
let vars = [
ENV_TRUSTED_PROXY_ENABLED,
-1
View File
@@ -68,7 +68,6 @@ pub async fn read_full_or_eof<R: AsyncRead + Send + Sync + Unpin>(
/// Read exactly buf.len() bytes into buf, or return an error if EOF is reached before any bytes are read.
/// Like Go's io.ReadFull.
#[allow(dead_code)]
pub async fn read_full<R: AsyncRead + Send + Sync + Unpin>(reader: R, buf: &mut [u8]) -> std::io::Result<usize> {
match read_full_or_eof(reader, buf).await? {
Some(n) => Ok(n),
-1
View File
@@ -431,7 +431,6 @@ pub fn parse_and_resolve_address(addr_str: &str) -> std::io::Result<SocketAddr>
Ok(resolved_addr)
}
#[allow(dead_code)]
pub fn bytes_stream<S, E>(stream: S, content_length: usize) -> impl Stream<Item = Result<Bytes, E>> + Send + 'static
where
S: Stream<Item = Result<Bytes, E>> + Send + 'static,
-1
View File
@@ -16,7 +16,6 @@
///
/// The table follows Linux `include/uapi/linux/magic.h`; filesystem magic
/// values without a stable Linux uapi source stay `UNKNOWN`.
#[allow(dead_code)]
pub(crate) fn get_fs_type(fs_type: u64) -> &'static str {
// Magic numbers for various filesystems.
match fs_type {
-1
View File
@@ -70,7 +70,6 @@ pub fn is_dir_object(object: &str) -> bool {
///
/// If the object name ends with `GLOBAL_DIR_SUFFIX`, it is replaced with a slash.
/// Otherwise, the name is returned as is.
#[allow(dead_code)]
pub fn decode_dir_object(object: &str) -> String {
if has_suffix(object, GLOBAL_DIR_SUFFIX) {
format!("{}{}", object.trim_end_matches(GLOBAL_DIR_SUFFIX), SLASH_SEPARATOR)
+1 -1
View File
@@ -59,7 +59,7 @@
{
default = rustPlatform.buildRustPackage {
pname = "rustfs";
version = "1.0.0-rc.2";
version = "1.0.0-rc.3";
src = ./.;
+2 -2
View File
@@ -2,8 +2,8 @@ apiVersion: v2
name: rustfs
description: RustFS helm chart to deploy RustFS on kubernetes cluster.
type: application
version: "1.0.0-rc.2"
appVersion: "1.0.0-rc.2"
version: "1.0.0-rc.3"
appVersion: "1.0.0-rc.3"
home: https://rustfs.com
icon: https://media.sys.truenas.net/apps/rustfs/icons/icon.svg
maintainers:
+5 -2
View File
@@ -1,9 +1,9 @@
%global _enable_debug_packages 0
%global _empty_manifest_terminate_build 0
%global prerelease rc.2
%global prerelease rc.3
Name: rustfs
Version: 1.0.0
Release: rc.2
Release: rc.3
Summary: High-performance distributed object storage for MinIO alternative
License: Apache-2.0
@@ -58,6 +58,9 @@ install %_builddir/%{name}-%{version}-%{prerelease}/target/%_arch/%_arch-unknown
%_bindir/rustfs
%changelog
* Mon Aug 17 2026 唐小鸭 <tangtang1251@qq.com>
- Update RPM package to RustFS 1.0.0-rc.3
* Fri Aug 14 2026 overtrue <anzhengchao@gmail.com>
- Update RPM package to RustFS 1.0.0-rc.2
+1 -1
View File
@@ -16,8 +16,8 @@ use crate::admin::auth::validate_admin_request;
use crate::admin::router::{AdminOperation, Operation, S3Router};
use crate::admin::runtime_sources::current_scanner_metrics_report;
use crate::auth::{check_key_valid, get_session_token};
use crate::module_switches::{ENV_SCANNER_ENABLED, scanner_enabled_from_env};
use crate::server::{ADMIN_PREFIX, RemoteAddr};
use crate::startup_background::{ENV_SCANNER_ENABLED, scanner_enabled_from_env};
use chrono::Utc;
use http::{HeaderMap, HeaderValue};
use hyper::{Method, StatusCode};
@@ -2995,9 +2995,9 @@ fn table_entry_from_create_table_request(
let CreateTableRequest {
name,
location,
schema,
partition_spec,
write_order,
mut schema,
mut partition_spec,
mut write_order,
stage_create,
mut properties,
} = request;
@@ -3031,6 +3031,9 @@ fn table_entry_from_create_table_request(
let metadata_location =
crate::table_catalog::default_table_metadata_file_path(namespace, &table, &next_metadata_file_name(1, &table_id));
crate::table_catalog::assign_fresh_create_schema_ids(&mut schema, partition_spec.as_mut(), write_order.as_mut())
.map_err(catalog_store_error)?;
let entry = crate::table_catalog::TableEntry {
version: crate::table_catalog::TABLE_CATALOG_ENTRY_VERSION,
table_bucket: bucket.to_string(),
@@ -1873,6 +1873,66 @@ fn create_table_request_accepts_standard_iceberg_rest_shape() {
assert_eq!(request.name, "events");
}
#[test]
fn create_table_assigns_positive_ids_to_spark_schema() {
let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
let request: CreateTableRequest = serde_json::from_value(serde_json::json!({
"name": "events",
"schema": {
"type": "struct",
"schema-id": 0,
"fields": [
{"id": 0, "name": "id", "required": false, "type": "long"},
{"id": 1, "name": "payload", "required": false, "type": "string"}
]
},
"partition-spec": {"spec-id": 0, "fields": []},
"properties": {"owner": "spark"}
}))
.expect("Spark create table request should parse");
let (_, metadata) = table_entry_from_create_table_request("warehouse", &namespace, request)
.expect("catalog should assign positive field IDs");
assert_eq!(metadata["schemas"][0]["fields"][0]["id"], 1);
assert_eq!(metadata["schemas"][0]["fields"][1]["id"], 2);
assert_eq!(metadata["last-column-id"], 2);
}
#[test]
fn create_table_assigns_fresh_id_to_negative_temporary_field_id() {
let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
let request: CreateTableRequest = serde_json::from_value(serde_json::json!({
"name": "events",
"schema": {
"type": "struct",
"identifier-field-ids": [-1],
"fields": [{"id": -1, "name": "id", "required": true, "type": "long"}]
},
"partition-spec": {
"fields": [{"source-id": -1, "name": "id", "transform": "identity"}]
},
"write-order": {
"fields": [{
"source-id": -1,
"transform": "identity",
"direction": "asc",
"null-order": "nulls-first"
}]
}
}))
.expect("create table request with a negative temporary field ID should parse");
let (_, metadata) = table_entry_from_create_table_request("warehouse", &namespace, request)
.expect("catalog should replace the negative temporary field ID");
assert_eq!(metadata["schemas"][0]["fields"][0]["id"], 1);
assert_eq!(metadata["schemas"][0]["identifier-field-ids"], serde_json::json!([1]));
assert_eq!(metadata["partition-specs"][0]["fields"][0]["source-id"], 1);
assert_eq!(metadata["sort-orders"][0]["fields"][0]["source-id"], 1);
assert_eq!(metadata["last-column-id"], 1);
}
#[test]
fn create_table_request_honors_supported_format_version_property() {
let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
@@ -1990,6 +2050,142 @@ fn catalog_assigns_read_only_schema_spec_and_sort_order_ids() {
assert_eq!(updated["default-sort-order-id"], 0);
}
#[test]
fn create_table_assigns_fresh_schema_field_ids_and_rewrites_references() {
let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
let request: CreateTableRequest = serde_json::from_value(serde_json::json!({
"name": "events",
"schema": {
"type": "struct",
"schema-id": 41,
"identifier-field-ids": [0],
"fields": [
{"id": 0, "name": "id", "required": true, "type": "long"},
{
"id": 10,
"name": "details",
"required": false,
"type": {
"type": "struct",
"fields": [{"id": 11, "name": "category", "required": false, "type": "string"}]
}
},
{
"id": 20,
"name": "tags",
"required": false,
"type": {
"type": "list",
"element-id": 21,
"element-required": false,
"element": "string"
}
},
{
"id": 30,
"name": "attributes",
"required": false,
"type": {
"type": "map",
"key-id": 31,
"key": "string",
"value-id": 32,
"value-required": false,
"value": {
"type": "struct",
"fields": [{"id": 33, "name": "score", "required": false, "type": "int"}]
}
}
}
]
},
"partition-spec": {
"spec-id": 42,
"fields": [{"source-id": 0, "name": "id", "transform": "identity"}]
},
"write-order": {
"order-id": 43,
"fields": [{
"source-id": 11,
"transform": "identity",
"direction": "asc",
"null-order": "nulls-first"
}]
}
}))
.expect("create table request should parse");
let (_, metadata) =
table_entry_from_create_table_request("warehouse", &namespace, request).expect("catalog should assign fresh field IDs");
let schema = &metadata["schemas"][0];
assert_eq!(schema["fields"][0]["id"], 1);
assert_eq!(schema["fields"][1]["id"], 2);
assert_eq!(schema["fields"][2]["id"], 3);
assert_eq!(schema["fields"][3]["id"], 4);
assert_eq!(schema["fields"][1]["type"]["fields"][0]["id"], 5);
assert_eq!(schema["fields"][2]["type"]["element-id"], 6);
assert_eq!(schema["fields"][3]["type"]["key-id"], 7);
assert_eq!(schema["fields"][3]["type"]["value-id"], 8);
assert_eq!(schema["fields"][3]["type"]["value"]["fields"][0]["id"], 9);
assert_eq!(schema["identifier-field-ids"], serde_json::json!([1]));
assert_eq!(metadata["last-column-id"], 9);
assert_eq!(metadata["partition-specs"][0]["fields"][0]["source-id"], 1);
assert_eq!(metadata["sort-orders"][0]["fields"][0]["source-id"], 5);
}
#[test]
fn create_table_rejects_duplicate_temporary_schema_field_ids() {
let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
let request: CreateTableRequest = serde_json::from_value(serde_json::json!({
"name": "events",
"schema": {
"type": "struct",
"fields": [
{"id": 0, "name": "id", "required": false, "type": "long"},
{"id": 0, "name": "payload", "required": false, "type": "string"}
]
}
}))
.expect("create table request should parse");
let error = table_entry_from_create_table_request("warehouse", &namespace, request)
.expect_err("duplicate temporary field IDs must be rejected");
assert_eq!(error.message(), Some("duplicate create schema field id 0"));
}
#[test]
fn create_table_rejects_excessive_schema_nesting() {
let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
let mut field_type = serde_json::Value::from("long");
for element_id in 1..=crate::table_catalog::ICEBERG_MAX_SCHEMA_NESTING_DEPTH + 1 {
field_type = serde_json::json!({
"type": "list",
"element-id": element_id,
"element-required": false,
"element": field_type
});
}
let request = CreateTableRequest {
name: "events".to_string(),
location: None,
schema: serde_json::json!({
"type": "struct",
"fields": [{"id": 0, "name": "nested", "required": false, "type": field_type}]
}),
partition_spec: None,
write_order: None,
stage_create: false,
properties: BTreeMap::new(),
};
let error = table_entry_from_create_table_request("warehouse", &namespace, request)
.expect_err("excessively nested create schemas must be rejected");
assert_eq!(error.message(), Some("create schema exceeds the maximum nesting depth"));
}
#[test]
fn standard_commit_binds_new_specs_and_sort_orders_to_current_schema() {
let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
@@ -2247,7 +2443,11 @@ fn create_table_counts_collection_ids_in_last_column_id() {
let (_, metadata) =
table_entry_from_create_table_request("warehouse", &namespace, request).expect("table metadata should be created");
assert_eq!(metadata["last-column-id"], 9);
let schema = &metadata["schemas"][0];
assert_eq!(schema["fields"][0]["type"]["element-id"], 3);
assert_eq!(schema["fields"][1]["type"]["key-id"], 4);
assert_eq!(schema["fields"][1]["type"]["value-id"], 5);
assert_eq!(metadata["last-column-id"], 5);
}
#[test]
+34 -5
View File
@@ -1074,7 +1074,7 @@ where
}
impl futures::Stream for MemoryTrackedBytesStream {
type Item = std::io::Result<Bytes>;
type Item = Result<Bytes, S3StdError>;
fn poll_next(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
let this = self.get_mut();
@@ -1105,7 +1105,8 @@ impl futures::Stream for MemoryTrackedBytesStream {
return Poll::Ready(Some(Err(std::io::Error::new(
std::io::ErrorKind::InvalidData,
format!("materialized GET body length mismatch: expected {}, got {}", this.expected, actual),
))));
)
.into())));
}
let Some(bytes) = this.bytes.take() else {
@@ -1132,6 +1133,16 @@ impl futures::Stream for MemoryTrackedBytesStream {
}
}
impl ByteStream for MemoryTrackedBytesStream {
fn remaining_length(&self) -> RemainingLength {
if self.emitted || self.bytes.is_none() {
RemainingLength::new_exact(0)
} else {
RemainingLength::new_exact(self.expected)
}
}
}
impl Drop for MemoryTrackedBytesStream {
fn drop(&mut self) {
if self.lifecycle.is_finished() {
@@ -4149,7 +4160,7 @@ impl DefaultObjectUsecase {
let bytes_len = bytes.len();
let guard = rustfs_io_metrics::track_get_object_buffered_bytes(bytes_len);
let remaining = usize::try_from(response_content_length.max(0)).unwrap_or(usize::MAX);
let blob = StreamingBlob::wrap(MemoryTrackedBytesStream::new(bytes, remaining, source, guard, lifecycle));
let blob = StreamingBlob::new(MemoryTrackedBytesStream::new(bytes, remaining, source, guard, lifecycle));
if let Some(handoff_start) = handoff_start {
rustfs_io_metrics::record_get_object_response_handoff(
"single_chunk",
@@ -12882,7 +12893,10 @@ mod tests {
.await
.expect("mismatched memory body must yield an item")
.expect_err("a short memory body must fail the stream instead of serving a truncated body");
assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
assert_eq!(
err.downcast_ref::<std::io::Error>().map(std::io::Error::kind),
Some(std::io::ErrorKind::InvalidData)
);
assert!(stream.next().await.is_none(), "stream must terminate after the error");
}
@@ -12901,7 +12915,22 @@ mod tests {
.await
.expect("mismatched memory body must yield an item")
.expect_err("an over-long memory body must fail the stream instead of serving mismatched bytes");
assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
assert_eq!(
err.downcast_ref::<std::io::Error>().map(std::io::Error::kind),
Some(std::io::ErrorKind::InvalidData)
);
}
#[test]
fn memory_blob_preserves_exact_remaining_length() {
let blob = DefaultObjectUsecase::build_memory_bytes_blob(
Bytes::from_static(b"hello"),
5,
GET_MEMORY_BODY_SOURCE_BUFFERED_BODY,
GetObjectBodyLifecycle::disabled(),
);
assert_eq!(blob.remaining_length().exact(), Some(5));
}
#[tokio::test]
+1
View File
@@ -88,6 +88,7 @@ pub mod inspect;
pub(crate) mod kms_deletion_gate;
pub mod license;
pub mod memory_observability;
pub mod module_switches;
pub mod profiling;
#[cfg(any(feature = "ftps", feature = "webdav", feature = "sftp"))]
pub mod protocols;
+68
View File
@@ -0,0 +1,68 @@
// 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.
//! Layer-neutral module switches (backlog#1834).
//!
//! Whether the scanner, heal, audit and notify modules are on is read from the
//! infra layer (storage helpers, node-service RPC) and from the interface layer
//! (admin handlers), but the switches used to live in `startup_background`
//! (composition) and `server` (interface). Every lower-layer read was therefore
//! an upward edge that had to be baselined by the layer-dependency guard.
//!
//! The env-derived scanner/heal predicates and the audit/notify state cells now
//! live here, at the bottom of the layer order, so those reads are ordinary
//! downward edges. Resolving the audit/notify state still needs server-side
//! configuration, so `server::refresh_audit_module_enabled` and its notify twin
//! keep that logic and publish the result through the setters below.
use rustfs_utils::get_env_bool_with_aliases;
use std::sync::atomic::{AtomicBool, Ordering};
pub(crate) const ENV_SCANNER_ENABLED: &str = "RUSTFS_SCANNER_ENABLED";
pub(crate) const ENV_SCANNER_ENABLED_DEPRECATED: &str = "RUSTFS_ENABLE_SCANNER";
pub(crate) const ENV_HEAL_ENABLED: &str = "RUSTFS_HEAL_ENABLED";
pub(crate) const ENV_HEAL_ENABLED_DEPRECATED: &str = "RUSTFS_ENABLE_HEAL";
static AUDIT_MODULE_ENABLED: AtomicBool = AtomicBool::new(rustfs_config::DEFAULT_AUDIT_ENABLE);
static NOTIFY_MODULE_ENABLED: AtomicBool = AtomicBool::new(rustfs_config::DEFAULT_NOTIFY_ENABLE);
/// Whether the data scanner is enabled, defaulting to on.
pub(crate) fn scanner_enabled_from_env() -> bool {
get_env_bool_with_aliases(ENV_SCANNER_ENABLED, &[ENV_SCANNER_ENABLED_DEPRECATED], true)
}
/// Whether background heal is enabled, defaulting to on.
pub(crate) fn heal_enabled_from_env() -> bool {
get_env_bool_with_aliases(ENV_HEAL_ENABLED, &[ENV_HEAL_ENABLED_DEPRECATED], true)
}
/// Last published audit-module state.
pub fn is_audit_module_enabled() -> bool {
AUDIT_MODULE_ENABLED.load(Ordering::Relaxed)
}
/// Publish the audit-module state resolved by `server::refresh_audit_module_enabled`.
pub(crate) fn set_audit_module_enabled(enabled: bool) {
AUDIT_MODULE_ENABLED.store(enabled, Ordering::Relaxed);
}
/// Last published notify-module state.
pub fn is_notify_module_enabled() -> bool {
NOTIFY_MODULE_ENABLED.load(Ordering::Relaxed)
}
/// Publish the notify-module state resolved by `server::refresh_notify_module_enabled`.
pub(crate) fn set_notify_module_enabled(enabled: bool) {
NOTIFY_MODULE_ENABLED.store(enabled, Ordering::Relaxed);
}
+2 -7
View File
@@ -19,11 +19,8 @@ use super::{
use crate::runtime_sources::AppContext;
use rustfs_audit::{AuditError, AuditResult, audit_system, init_audit_system, system::AuditSystemState};
use std::collections::HashSet;
use std::sync::atomic::{AtomicBool, Ordering};
use tracing::{info, warn};
static AUDIT_MODULE_ENABLED: AtomicBool = AtomicBool::new(rustfs_config::DEFAULT_AUDIT_ENABLE);
fn server_config_from_context() -> Option<rustfs_config::server_config::Config> {
runtime_sources::current_server_config()
}
@@ -37,13 +34,11 @@ fn server_config_for_context(context: Option<&AppContext>) -> Option<rustfs_conf
pub fn refresh_audit_module_enabled() -> bool {
let enabled = resolve_audit_module_state().enabled;
AUDIT_MODULE_ENABLED.store(enabled, Ordering::Relaxed);
crate::module_switches::set_audit_module_enabled(enabled);
enabled
}
pub fn is_audit_module_enabled() -> bool {
AUDIT_MODULE_ENABLED.load(Ordering::Relaxed)
}
pub use crate::module_switches::is_audit_module_enabled;
fn has_any_persisted_audit_targets(config: &rustfs_config::server_config::Config) -> bool {
for &subsystem in rustfs_config::audit::AUDIT_SUB_SYSTEMS {
+3 -6
View File
@@ -34,7 +34,6 @@ use tokio::time::{Instant, MissedTickBehavior};
use tokio_util::sync::CancellationToken;
use tracing::{info, instrument, warn};
static NOTIFY_MODULE_ENABLED: AtomicBool = AtomicBool::new(rustfs_config::DEFAULT_NOTIFY_ENABLE);
static NOTIFY_RUNTIME_RECONCILED: AtomicBool = AtomicBool::new(false);
static NOTIFY_BUCKET_RULES_RECONCILED: AtomicBool = AtomicBool::new(false);
static ECSTORE_EVENT_DISPATCH_HOOK: OnceLock<()> = OnceLock::new();
@@ -70,13 +69,11 @@ fn should_reconcile_bucket_notification_rules(runtime_changed: bool, notify_enab
pub fn refresh_notify_module_enabled() -> bool {
let enabled = resolve_notify_module_state().enabled;
NOTIFY_MODULE_ENABLED.store(enabled, Ordering::Relaxed);
crate::module_switches::set_notify_module_enabled(enabled);
enabled
}
pub fn is_notify_module_enabled() -> bool {
NOTIFY_MODULE_ENABLED.load(Ordering::Relaxed)
}
pub use crate::module_switches::is_notify_module_enabled;
pub(crate) use crate::shared_types::convert_ecstore_object_info;
@@ -171,7 +168,7 @@ pub(crate) async fn reconcile_event_notifier_from_store(
let transition_system = system.clone();
let transition_store = store.clone();
let transition = with_refreshed_notify_module_state_from(store.clone(), move |resolution| async move {
NOTIFY_MODULE_ENABLED.store(resolution.enabled, Ordering::Relaxed);
crate::module_switches::set_notify_module_enabled(resolution.enabled);
let read_store = transition_store.clone();
let config_system = transition_system.clone();
with_server_config_read_lock(transition_store, move || async move {
+1 -13
View File
@@ -12,32 +12,20 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use crate::module_switches::{heal_enabled_from_env, scanner_enabled_from_env};
use crate::storage_api::startup::background::{ECStore, set_workload_admission_snapshot_provider};
use crate::workload_admission::RustFsWorkloadAdmissionSnapshotProvider;
use rustfs_concurrency::WorkloadAdmissionSnapshotProvider;
use rustfs_heal::{
create_ahm_services_cancel_token, heal::storage::ECStoreHealStorage, init_heal_manager_with_workload_provider,
};
use rustfs_utils::get_env_bool_with_aliases;
use std::{io::Result, sync::Arc};
use tracing::{debug, info};
pub(crate) const ENV_SCANNER_ENABLED: &str = "RUSTFS_SCANNER_ENABLED";
pub(crate) const ENV_SCANNER_ENABLED_DEPRECATED: &str = "RUSTFS_ENABLE_SCANNER";
pub(crate) const ENV_HEAL_ENABLED: &str = "RUSTFS_HEAL_ENABLED";
pub(crate) const ENV_HEAL_ENABLED_DEPRECATED: &str = "RUSTFS_ENABLE_HEAL";
const LOG_COMPONENT_MAIN: &str = "main";
const LOG_SUBSYSTEM_STARTUP: &str = "startup";
const EVENT_BACKGROUND_SERVICES_CONFIGURED: &str = "background_services_configured";
pub(crate) fn scanner_enabled_from_env() -> bool {
get_env_bool_with_aliases(ENV_SCANNER_ENABLED, &[ENV_SCANNER_ENABLED_DEPRECATED], true)
}
pub(crate) fn heal_enabled_from_env() -> bool {
get_env_bool_with_aliases(ENV_HEAL_ENABLED, &[ENV_HEAL_ENABLED_DEPRECATED], true)
}
pub(crate) async fn init_background_service_runtime(store: Arc<ECStore>) -> Result<bool> {
let _ = create_ahm_services_cancel_token();
+4 -1
View File
@@ -56,7 +56,6 @@
//! ```
// Allow dead_code for public API that may be used by external modules or future features
#![allow(dead_code)]
use parking_lot::{Mutex, RwLock};
use std::collections::{HashMap, HashSet};
@@ -264,6 +263,10 @@ pub struct ResourceUsage {
/// Deadlock detector.
pub struct DeadlockDetector {
/// Configuration.
#[allow(
dead_code,
reason = "policy snapshot retained beside the detector it configures (backlog#1823)"
)]
config: RequestHangDetectionPolicy,
/// Shared concurrency facade policy.
policy: DeadlockMonitorPolicy,
+1 -1
View File
@@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use crate::server::{is_audit_module_enabled, is_notify_module_enabled};
use crate::module_switches::{is_audit_module_enabled, is_notify_module_enabled};
use crate::shared_types::convert_ecstore_object_info;
use crate::storage::access::{ReqInfo, request_context_from_req};
use crate::storage::request_context::RequestContext;
-1
View File
@@ -31,7 +31,6 @@
//! ```
// Allow dead_code for public API that may be used by external modules or future features
#![allow(dead_code)]
//! # Key Features
//!
//! - Early lock release after metadata read
+1 -1
View File
@@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use crate::startup_background::{heal_enabled_from_env, scanner_enabled_from_env};
use crate::module_switches::{heal_enabled_from_env, scanner_enabled_from_env};
use crate::storage::storage_api::runtime_sources_consumer::EndpointServerPools;
use jiff::Timestamp;
use rmp_serde::Deserializer;
-14
View File
@@ -12,8 +12,6 @@
// See the License for the specific language governing permissions and
// limitations under the License.
#![allow(dead_code)]
//! Facade modules for incremental S3 API extraction from `ecfs.rs`.
//!
//! This file intentionally starts as skeleton-only. Behavior remains in place
@@ -29,18 +27,6 @@ pub(crate) mod common;
pub(crate) mod multipart;
pub(crate) mod tagging;
pub(crate) fn default_bucket_usecase() -> DefaultBucketUsecase {
DefaultBucketUsecase::from_global()
}
pub(crate) fn default_multipart_usecase() -> DefaultMultipartUsecase {
DefaultMultipartUsecase::from_global()
}
pub(crate) fn default_object_usecase() -> DefaultObjectUsecase {
DefaultObjectUsecase::from_global()
}
/// Resolve the object use-case for a server's request path (backlog#1052 S6):
/// bind it to the server's own application context so it resolves that
/// server's store instead of the ambient process default.

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