diff --git a/.config/make/tests.mak b/.config/make/tests.mak index ad6cdfa7a..8f7160210 100644 --- a/.config/make/tests.mak +++ b/.config/make/tests.mak @@ -56,6 +56,7 @@ script-tests: ## Run shell script tests $(RUSTFS_PYTHON_BIN) ./scripts/check_test_wiring.py --self-test $(RUSTFS_PYTHON_BIN) ./scripts/codemods/s3_error_to_s3_types.py --self-test $(RUSTFS_PYTHON_BIN) ./scripts/test_e2e_binary.py + $(RUSTFS_PYTHON_BIN) ./scripts/test_download_release_asset.py $(RUSTFS_PYTHON_BIN) ./scripts/test_inspect_btf_prototypes.py $(RUSTFS_PYTHON_BIN) ./scripts/test_release_build_workflow.py $(RUSTFS_PYTHON_BIN) ./scripts/test_release_glibc.py diff --git a/.config/test-count-baseline.txt b/.config/test-count-baseline.txt index fb386fbf4..c8eb0a999 100644 --- a/.config/test-count-baseline.txt +++ b/.config/test-count-baseline.txt @@ -10,7 +10,7 @@ # intentionally removed. Regenerate with: # scripts/check_test_count_ratchet.sh --update e2e_test 1026 20 -rustfs 6439 64 +rustfs 6440 64 rustfs-audit 70 0 rustfs-checksums 26 0 rustfs-common 36 0 @@ -20,10 +20,10 @@ rustfs-connect-symbol-catalog 6 0 rustfs-credentials 24 0 rustfs-crypto 67 0 rustfs-data-usage 58 0 -rustfs-ecstore 7818 2 +rustfs-ecstore 7828 2 rustfs-extension-schema 23 0 rustfs-filemeta 361 0 -rustfs-heal 1127 3 +rustfs-heal 1128 3 rustfs-heal-contracts 8 0 rustfs-iam 415 0 rustfs-io-core 52 0 @@ -37,7 +37,7 @@ rustfs-log-analyzer 91 0 rustfs-madmin 210 0 rustfs-notify 140 0 rustfs-object-capacity 92 0 -rustfs-object-data-cache 140 0 +rustfs-object-data-cache 141 0 rustfs-obs 387 0 rustfs-policy 758 0 rustfs-protocols 770 1 @@ -48,9 +48,9 @@ rustfs-rio-v2 34 6 rustfs-s3-client 71 0 rustfs-s3-ops 8 0 rustfs-s3-types 105 0 -rustfs-s3select-api 212 0 -rustfs-s3select-query 231 2 -rustfs-scanner 1385 20 +rustfs-s3select-api 226 0 +rustfs-s3select-query 234 2 +rustfs-scanner 1386 20 rustfs-scanner-metrics 90 0 rustfs-security-governance 20 0 rustfs-signer 34 0 diff --git a/.github/actions/quick-checks/action.yml b/.github/actions/quick-checks/action.yml index 613a79080..ea3f01a2b 100644 --- a/.github/actions/quick-checks/action.yml +++ b/.github/actions/quick-checks/action.yml @@ -28,11 +28,12 @@ runs: - name: Install actionlint shell: bash + env: + GH_TOKEN: ${{ github.token }} + ASSET_DOWNLOADER: ${{ github.action_path }}/../setup/download_release_asset.py run: | actionlint_dir="$(mktemp -d "${RUNNER_TEMP}/actionlint.XXXXXX")" - curl --fail --location --silent --show-error \ - --output "$actionlint_dir/actionlint.tar.gz" \ - https://github.com/rhysd/actionlint/releases/download/v1.7.12/actionlint_1.7.12_linux_amd64.tar.gz + python3 "$ASSET_DOWNLOADER" --repository rhysd/actionlint --tag v1.7.12 --asset actionlint_1.7.12_linux_amd64.tar.gz --output "$actionlint_dir/actionlint.tar.gz" --sha256 8aca8db96f1b94770f1b0d72b6dddcb1ebb8123cb3712530b08cc387b349a3d8 echo "8aca8db96f1b94770f1b0d72b6dddcb1ebb8123cb3712530b08cc387b349a3d8 $actionlint_dir/actionlint.tar.gz" | sha256sum --check --status tar -xzf "$actionlint_dir/actionlint.tar.gz" -C "$actionlint_dir" actionlint rm "$actionlint_dir/actionlint.tar.gz" diff --git a/.github/actions/setup/action.yml b/.github/actions/setup/action.yml index f9e178569..c075dd9b7 100644 --- a/.github/actions/setup/action.yml +++ b/.github/actions/setup/action.yml @@ -66,10 +66,7 @@ inputs: runs: using: "composite" steps: - # protobuf-compiler is deliberately absent: the setup-protoc step below - # installs 35.1 into the tool cache and prepends it to PATH, so the apt - # build (older, and never version-matched) was shadowed on every run and - # simply never used. + # Use the pinned code generators rather than Ubuntu's older packages. - name: Install system dependencies (Ubuntu) if: runner.os == 'Linux' shell: bash @@ -79,7 +76,9 @@ runs: build-essential \ pkg-config \ libssl-dev \ - ripgrep + ripgrep \ + python3 \ + gh # musl-gcc is needed by the native musl release leg, and zip/unzip by the # release packaging steps. No CI test lane touches any of them. @@ -88,13 +87,32 @@ runs: shell: bash run: sudo apt-get install -y musl-tools zip unzip + - name: Install pinned code generators through GitHub API (Linux x64) + if: runner.os == 'Linux' && runner.arch == 'X64' + shell: bash + env: + GH_TOKEN: ${{ github.token }} + ASSET_DOWNLOADER: ${{ github.action_path }}/download_release_asset.py + run: | + python3 "$ASSET_DOWNLOADER" --repository protocolbuffers/protobuf --tag v35.1 --asset protoc-35.1-linux-x86_64.zip --output "$RUNNER_TEMP/protoc.zip" + python3 "$ASSET_DOWNLOADER" --repository google/flatbuffers --tag v25.12.19 --asset 'Linux.flatc.binary.g++-13.zip' --output "$RUNNER_TEMP/flatc.zip" + python3 -m zipfile -e "$RUNNER_TEMP/protoc.zip" "$RUNNER_TEMP/protoc" + python3 -m zipfile -e "$RUNNER_TEMP/flatc.zip" "$RUNNER_TEMP/flatc" + chmod +x "$RUNNER_TEMP/protoc/bin/protoc" "$RUNNER_TEMP/flatc/flatc" + "$RUNNER_TEMP/protoc/bin/protoc" --version + "$RUNNER_TEMP/flatc/flatc" --version + echo "$RUNNER_TEMP/protoc/bin" >> "$GITHUB_PATH" + echo "$RUNNER_TEMP/flatc" >> "$GITHUB_PATH" + - name: Install protoc + if: runner.os != 'Linux' || runner.arch != 'X64' uses: rustfs/setup-protoc@a3705324d8f9bf5b6c3573fb6cf8ae421db55dd6 # v3.0.1 with: version: "35.1" repo-token: ${{ github.token }} - name: Install flatc + if: runner.os != 'Linux' || runner.arch != 'X64' uses: Nugine/setup-flatc@698800de72a96bfb22cf60431dc21a2ff9a7e07b # v1 with: version: "25.12.19" diff --git a/.github/actions/setup/download_release_asset.py b/.github/actions/setup/download_release_asset.py new file mode 100644 index 000000000..901317467 --- /dev/null +++ b/.github/actions/setup/download_release_asset.py @@ -0,0 +1,84 @@ +#!/usr/bin/env python3 +"""Download a named GitHub release asset through its API and verify its digest.""" + +import argparse +import hashlib +import json +from pathlib import Path +import re +import subprocess +import sys +import tempfile +from urllib.parse import quote + + +def download(repository, tag, name, output, expected_sha256=None): + if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.-]*/[A-Za-z0-9][A-Za-z0-9_.-]*", repository): + raise ValueError("invalid release repository") + if not tag or not name or "/" in name or "\\" in name: + raise ValueError("invalid release tag or asset name") + if expected_sha256 is not None and not re.fullmatch(r"[0-9a-f]{64}", expected_sha256): + raise ValueError("invalid pinned asset digest") + result = subprocess.run( + ["gh", "api", "--hostname", "github.com", f"repos/{repository}/releases/tags/{quote(tag, safe='')}"], + check=True, capture_output=True, text=True, timeout=60, + ) + release = json.loads(result.stdout) + if not isinstance(release, dict) or release.get("tag_name") != tag: + raise ValueError("release metadata does not match the requested tag") + assets = release.get("assets") + if not isinstance(assets, list) or any(not isinstance(asset, dict) for asset in assets): + raise ValueError("invalid release asset inventory") + matches = [asset for asset in assets if asset.get("name") == name] + if len(matches) != 1: + raise ValueError("release must contain exactly one named asset") + asset = matches[0] + asset_id, size, digest = asset.get("id"), asset.get("size"), asset.get("digest") + if type(asset_id) is not int or asset_id <= 0 or type(size) is not int or not 0 < size <= 512 * 1024 * 1024: + raise ValueError("invalid release asset identity or size") + if not isinstance(digest, str) or not re.fullmatch(r"sha256:[0-9a-f]{64}", digest): + raise ValueError("release asset has no valid SHA-256 digest") + sha256 = digest.removeprefix("sha256:") + if expected_sha256 is not None and expected_sha256 != sha256: + raise ValueError("release digest differs from the pinned digest") + output = Path(output) + output.parent.mkdir(parents=True, exist_ok=True) + temporary = None + try: + with tempfile.NamedTemporaryFile(dir=output.parent, prefix=".release-asset-", delete=False) as stream: + temporary = Path(stream.name) + # gh authenticates to the API; its HTTP client strips credentials on + # cross-host redirects. No browser URL or token is supplied in argv. + subprocess.run( + ["gh", "api", "--hostname", "github.com", "--header", "Accept: application/octet-stream", f"repos/{repository}/releases/assets/{asset_id}"], + check=True, stdout=stream, timeout=300, + ) + with temporary.open("rb") as stream: + actual = hashlib.file_digest(stream, "sha256").hexdigest() + if temporary.stat().st_size != size or actual != sha256: + raise ValueError("downloaded release asset failed size or SHA-256 verification") + temporary.replace(output) + print(f"Verified release asset: {repository} {tag} {name} sha256={actual}") + finally: + if temporary is not None: + temporary.unlink(missing_ok=True) + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--repository", required=True) + parser.add_argument("--tag", required=True) + parser.add_argument("--asset", required=True) + parser.add_argument("--output", required=True, type=Path) + parser.add_argument("--sha256") + args = parser.parse_args() + try: + download(args.repository, args.tag, args.asset, args.output, args.sha256) + except (ValueError, KeyError, TypeError, json.JSONDecodeError, subprocess.SubprocessError, OSError) as error: + print(f"Release asset download failed: {error}", file=sys.stderr) + return 1 + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 4efa55d38..85ee5dceb 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -144,15 +144,23 @@ jobs: - uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7 with: persist-credentials: false + - name: Install pinned typos through GitHub API + env: + GH_TOKEN: ${{ github.token }} + TYPOS_VERSION: 1.50.2 + run: | + python3 .github/actions/setup/download_release_asset.py --repository crate-ci/typos --tag "v${TYPOS_VERSION}" --asset "typos-v${TYPOS_VERSION}-x86_64-unknown-linux-musl.tar.gz" --output "$RUNNER_TEMP/typos.tar.gz" + mkdir -p "$RUNNER_TEMP/typos" + tar -xzf "$RUNNER_TEMP/typos.tar.gz" -C "$RUNNER_TEMP/typos" - name: Typos check with custom config file - uses: crate-ci/typos@512fc24f32f44ab01972217aaaf3dc86ec234d53 # v1.50.2 + run: '"$RUNNER_TEMP/typos/typos" .' # Fail early with compile-free checks for every pull request. quick-checks: name: Quick Checks if: github.event_name != 'pull_request' || github.event.action != 'closed' runs-on: ubuntu-latest - timeout-minutes: 10 + timeout-minutes: 20 steps: - name: Checkout repository uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7 @@ -1245,11 +1253,16 @@ jobs: - name: Install mc env: + GH_TOKEN: ${{ github.token }} MC_VERSION: RELEASE.2025-08-13T08-35-41Z MC_SHA256: 01f866e9c5f9b87c2b09116fa5d7c06695b106242d829a8bb32990c00312e891 run: | + if ! command -v gh >/dev/null 2>&1; then + sudo apt-get update + sudo apt-get install -y gh + fi MC_BINARY="mc.linux-amd64.${MC_VERSION}" - curl -fsSLo "$RUNNER_TEMP/mc" "https://github.com/minio/mc/releases/download/${MC_VERSION}/${MC_BINARY}" + python3 .github/actions/setup/download_release_asset.py --repository minio/mc --tag "$MC_VERSION" --asset "$MC_BINARY" --output "$RUNNER_TEMP/mc" --sha256 "$MC_SHA256" echo "${MC_SHA256} $RUNNER_TEMP/mc" | sha256sum --check --status chmod +x "$RUNNER_TEMP/mc" echo "$RUNNER_TEMP" >> "$GITHUB_PATH" diff --git a/Cargo.lock b/Cargo.lock index 8b3e668fd..58ffc4b8f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -10826,6 +10826,7 @@ dependencies = [ "arc-swap", "async-compression", "async-trait", + "brotli 9.1.1", "bytes", "chrono", "crc-fast", @@ -10835,6 +10836,7 @@ dependencies = [ "futures-core", "hotpath", "http 1.5.0", + "lz4", "memchr", "metrics", "parking_lot 0.13.0", diff --git a/crates/ecstore/src/cluster/rpc/http_auth.rs b/crates/ecstore/src/cluster/rpc/http_auth.rs index aa42e623b..bdd979c97 100644 --- a/crates/ecstore/src/cluster/rpc/http_auth.rs +++ b/crates/ecstore/src/cluster/rpc/http_auth.rs @@ -127,9 +127,14 @@ const HTTP_PUT_FILE_CAPABILITY_AUTH_DOMAIN: &[u8] = b"rustfs-http-put-file-capab const UNSIGNED_PAYLOAD: &str = "UNSIGNED-PAYLOAD"; const UNSIGNED_PAYLOAD_NONCE: &str = "unsigned"; const SIGNATURE_VALID_DURATION: i64 = 300; // 5 minutes -const REPLAY_CACHE_RETENTION: Duration = Duration::from_secs(601); -const REPLAY_CACHE_RETENTION_SECS: usize = 601; -const REPLAY_CACHE_ENTRY_BYTES_ESTIMATE: u64 = 128; +// Timestamp validity and the wall-clock high-water mark still gate eviction, +// including future-dated signatures and replay attempts after clock rollback. +const REPLAY_CACHE_RETENTION: Duration = Duration::from_secs(301); +// Preserve capacity for the full validity window at maximum accepted clock skew. +const REPLAY_CACHE_CAPACITY_WINDOW_SECS: usize = 601; +// A reserved index avoids overlapping old/new tables during a rehash. Include +// index slack, compact expiry storage, and page/allocator bookkeeping. +const REPLAY_CACHE_ENTRY_BYTES_ESTIMATE: usize = 2 * std::mem::size_of::() + std::mem::size_of::() + 16; // Keep 16 CPU / 32 GiB field nodes at the 32M cap without requiring an env override. const REPLAY_CACHE_AUTO_MEMORY_PERCENT: u64 = 13; const REPLAY_CACHE_AUTO_RPC_RPS_PER_CPU: usize = 4096; @@ -222,11 +227,11 @@ fn replay_cache_capacity_from_resources(cpu_count: usize, memory_limit_bytes: Op let cpu_count = cpu_count.max(1); let cpu_based_capacity = cpu_count .saturating_mul(REPLAY_CACHE_AUTO_RPC_RPS_PER_CPU) - .saturating_mul(REPLAY_CACHE_RETENTION_SECS); + .saturating_mul(REPLAY_CACHE_CAPACITY_WINDOW_SECS); let memory_based_capacity = memory_limit_bytes .map(|bytes| { let budget = bytes.saturating_mul(REPLAY_CACHE_AUTO_MEMORY_PERCENT) / 100; - saturating_usize_from_u64(budget / REPLAY_CACHE_ENTRY_BYTES_ESTIMATE) + saturating_usize_from_u64(budget / u64::try_from(REPLAY_CACHE_ENTRY_BYTES_ESTIMATE).unwrap_or(u64::MAX)) }) .unwrap_or(REPLAY_CACHE_AUTO_MAX_CAPACITY); let capacity = memory_based_capacity @@ -497,6 +502,12 @@ fn publish_nonce_cache_metrics(metrics: Option>) { } impl RpcNonceCache { + fn reserve_index(&mut self, capacity: usize) -> std::io::Result<()> { + self.nonces + .try_reserve(capacity) + .map_err(|error| std::io::Error::new(std::io::ErrorKind::OutOfMemory, error)) + } + fn remove_expired(&mut self, now: Instant, wall_time: i64) -> usize { let mut removed = 0; while let Some(nonce) = self.expirations.pop_expired(now, wall_time) { @@ -537,6 +548,11 @@ impl RpcNonceCache { }), ); } + if self.nonces.is_empty() + && let Err(error) = self.reserve_index(record.capacity.min(REPLAY_CACHE_AUTO_MAX_CAPACITY)) + { + return (Err(error), Some(metrics)); + } self.nonces.insert(record.nonce); self.expirations .push_back(record.expires_at, record.signed_at.saturating_add(SIGNATURE_VALID_DURATION), record.nonce); @@ -1699,6 +1715,9 @@ pub fn verify_tonic_rpc_signature_with_bootstrap( } pub fn tonic_rpc_auth_failure_reason(error: &std::io::Error) -> &'static str { + if error.kind() == std::io::ErrorKind::OutOfMemory { + return "replay_cache_allocation"; + } match error.to_string().as_str() { "Missing RPC audience" => "missing_audience", "Invalid RPC request path" => "invalid_request_path", @@ -3138,9 +3157,9 @@ mod tests { assert_eq!(decision.source, ReplayCacheCapacitySource::Auto); assert_eq!(decision.memory_basis, Some(MemoryBasis::Host)); - assert_eq!(decision.memory_based_capacity, 17_448_304); + assert_eq!(decision.memory_based_capacity, 31_019_208); assert_eq!(decision.cpu_based_capacity, 19_693_568); - assert_eq!(decision.capacity, 17_448_304); + assert_eq!(decision.capacity, 19_693_568); } #[test] @@ -3150,13 +3169,13 @@ mod tests { replay_cache_capacity_decision(rustfs_utils::EnvParseOutcome::Absent, 16, Some(32 * gib), Some(MemoryBasis::Host)); assert_eq!(decision.source, ReplayCacheCapacitySource::Auto); - assert_eq!(decision.memory_based_capacity, 34_896_609); + assert_eq!(decision.memory_based_capacity, 62_038_416); assert_eq!(decision.cpu_based_capacity, 39_387_136); assert_eq!(decision.capacity, REPLAY_CACHE_AUTO_MAX_CAPACITY); let observed_field_node = replay_cache_capacity_decision(rustfs_utils::EnvParseOutcome::Absent, 16, Some(31 * gib), Some(MemoryBasis::Host)); - assert_eq!(observed_field_node.memory_based_capacity, 33_806_090); + assert_eq!(observed_field_node.memory_based_capacity, 60_099_715); assert_eq!(observed_field_node.cpu_based_capacity, 39_387_136); assert_eq!(observed_field_node.capacity, REPLAY_CACHE_AUTO_MAX_CAPACITY); } @@ -3229,6 +3248,138 @@ mod tests { } } + #[test] + fn nonce_cache_releases_expired_timestamp_capacity_without_allowing_clock_rollback_replay() { + let now = Instant::now(); + let expiry = now + REPLAY_CACHE_RETENTION; + let later = now + Duration::from_secs(302); + let nonce = Uuid::new_v4(); + let replacement = Uuid::new_v4(); + let mut cache = RpcNonceCache::default(); + cache + .check_and_record(test_nonce_record(nonce, 1_000, now, 1_000, expiry, 1)) + .0 + .expect("record initial nonce"); + + let boundary = cache + .check_and_record(test_nonce_record(nonce, 1_000, later, 1_300, later + REPLAY_CACHE_RETENTION, 1)) + .0 + .expect_err("a signature at the inclusive freshness boundary must remain protected"); + assert_eq!(boundary.to_string(), "RPC request replay detected"); + cache + .check_and_record(test_nonce_record(replacement, 1_302, later, 1_302, later + REPLAY_CACHE_RETENTION, 1)) + .0 + .expect("an expired signature must release bounded cache capacity"); + assert!(!cache.nonces.contains(&nonce)); + assert!(cache.nonces.contains(&replacement)); + + let rollback = cache + .check_and_record(test_nonce_record(nonce, 1_000, later, 1_000, later + REPLAY_CACHE_RETENTION, 1)) + .0 + .expect_err("clock rollback must not resurrect an evicted signature"); + assert_eq!(rollback.to_string(), "RPC request timestamp expired after clock regression"); + } + + #[test] + fn nonce_cache_keeps_future_timestamp_reserved_until_its_full_validity_window_closes() { + let now = Instant::now(); + let expiry = now + REPLAY_CACHE_RETENTION; + let later = now + Duration::from_secs(302); + let nonce = Uuid::new_v4(); + let mut cache = RpcNonceCache::default(); + cache + .check_and_record(test_nonce_record(nonce, 1_300, now, 1_000, expiry, 1)) + .0 + .expect("record nonce signed at maximum allowed forward skew"); + let replay = cache + .check_and_record(test_nonce_record(nonce, 1_300, later, 1_302, later + REPLAY_CACHE_RETENTION, 1)) + .0 + .expect_err("monotonic expiry must not remove a still-valid future signature"); + assert_eq!(replay.to_string(), "RPC request replay detected"); + let full = cache + .check_and_record(test_nonce_record(Uuid::new_v4(), 1_302, later, 1_302, later + REPLAY_CACHE_RETENTION, 1)) + .0 + .expect_err("a still-valid nonce must continue consuming capacity"); + assert_eq!(full.to_string(), "RPC replay cache capacity exceeded"); + let expired = now + Duration::from_secs(602); + cache + .check_and_record(test_nonce_record( + Uuid::new_v4(), + 1_602, + expired, + 1_602, + expired + REPLAY_CACHE_RETENTION, + 1, + )) + .0 + .expect("release capacity only once future timestamp validity has closed"); + } + + #[test] + fn nonce_index_rejects_unallocatable_capacity_with_typed_error() { + let mut cache = RpcNonceCache::default(); + let error = cache + .reserve_index(usize::MAX) + .expect_err("unallocatable capacity must fail without a panic"); + assert_eq!(error.kind(), std::io::ErrorKind::OutOfMemory); + assert!( + error + .get_ref() + .is_some_and(|source| source.is::()) + ); + assert_eq!(tonic_rpc_auth_failure_reason(&error), "replay_cache_allocation"); + assert!(cache.nonces.is_empty()); + assert!(cache.expirations.pages.is_empty()); + } + + #[test] + fn nonce_cache_reserves_the_bounded_index_before_sustained_traffic() { + let now = Instant::now(); + let expiry = now + REPLAY_CACHE_RETENTION; + let mut cache = RpcNonceCache::default(); + let capacity = 4096; + cache + .check_and_record(test_nonce_record(Uuid::from_u128(1), 1000, now, 1000, expiry, capacity)) + .0 + .expect("record first nonce"); + let allocated = cache.nonces.capacity(); + assert!(allocated >= capacity, "the hash index must not rehash while serving the bounded window"); + for n in 2..=capacity { + cache + .check_and_record(test_nonce_record( + Uuid::from_u128(u128::try_from(n).expect("nonce index")), + 1000, + now, + 1000, + expiry, + capacity, + )) + .0 + .expect("record within bounded capacity"); + } + assert_eq!(cache.nonces.capacity(), allocated); + let full = cache + .check_and_record(test_nonce_record(Uuid::from_u128(9999), 1000, now, 1000, expiry, capacity)) + .0 + .expect_err("preallocation must not relax the logical capacity fence"); + assert_eq!(full.to_string(), "RPC replay cache capacity exceeded"); + assert_eq!(cache.nonces.len(), capacity); + } + + #[test] + fn replay_cache_compact_storage_fits_two_cpu_window_in_2800_mib() { + let decision = replay_cache_capacity_decision( + rustfs_utils::EnvParseOutcome::Absent, + 2, + Some(2800 * 1024 * 1024), + Some(MemoryBasis::Cgroup), + ); + assert_eq!(decision.cpu_based_capacity, 4_923_392); + assert_eq!(decision.capacity, decision.cpu_based_capacity); + assert!(decision.memory_based_capacity >= decision.capacity); + assert_eq!(decision.source, ReplayCacheCapacitySource::Auto); + } + #[test] fn nonce_cache_expires_by_monotonic_deadline_and_fails_closed_at_capacity() { let now = Instant::now(); diff --git a/crates/ecstore/src/cluster/rpc/remote_disk.rs b/crates/ecstore/src/cluster/rpc/remote_disk.rs index 7590f026c..002fc3af2 100644 --- a/crates/ecstore/src/cluster/rpc/remote_disk.rs +++ b/crates/ecstore/src/cluster/rpc/remote_disk.rs @@ -376,6 +376,21 @@ struct WalkCopyError { source: io::Error, } +impl WalkCopyError { + fn into_disk_error(self) -> DiskError { + // Only a local output write/flush proves the metacache consumer closed. + // An identical I/O kind from the remote input remains a transport failure. + if matches!(self.reason, RemoteWalkFailureReason::Write | RemoteWalkFailureReason::Flush) + && self.source.kind() == io::ErrorKind::BrokenPipe + && self.source.get_ref().is_none() + { + DiskError::metacache_output_stream_closed() + } else { + DiskError::Io(self.source) + } + } +} + async fn copy_stream_with_buffer( reader: &mut R, writer: &mut W, @@ -2096,6 +2111,7 @@ impl RemoteDisk { failure_health_action: FailureHealthAction, ) { if let Err(err) = operation_result + && !err.is_metacache_output_stream_closed() && is_network_like_disk_error(err) { counter!( @@ -4056,7 +4072,7 @@ impl DiskAPI for RemoteDisk { Ok(_) => return Ok(()), Err(error) => { failure_reason = Some(error.reason); - return Err(DiskError::Io(error.source)); + return Err(error.into_disk_error()); } } } @@ -9687,6 +9703,43 @@ mod tests { } } + #[tokio::test] + async fn remote_walk_preserves_local_metacache_consumer_close() { + let transport = + RetryingWalkDirInternodeDataTransport::with_steps(vec![WalkDirTestStep::Data(b"metacache payload".to_vec())]); + let disk = new_remote_disk_with_transport(Arc::new(transport.clone())).await; + let (mut writer, reader) = tokio::io::duplex(1); + drop(reader); + let error = disk + .walk_dir(WalkDirOptions::default(), &mut writer) + .await + .expect_err("a dropped metacache consumer must stop the producer"); + assert!( + error.is_metacache_output_stream_closed(), + "preserve the benign completion marker: {error:?}" + ); + assert_eq!(transport.calls().len(), 1, "local consumer closure is not a remote open retry"); + assert!(!disk.health.is_faulty()); + } + + #[tokio::test] + async fn remote_walk_keeps_remote_read_broken_pipe_as_transport_failure() { + let transport = RetryingWalkDirInternodeDataTransport::with_steps(vec![WalkDirTestStep::PartialDataThenError { + data: b"partial metacache".to_vec(), + error: io::ErrorKind::BrokenPipe.into(), + }]); + let disk = new_remote_disk_with_transport(Arc::new(transport.clone())).await; + let mut writer = Vec::new(); + let error = disk + .walk_dir(WalkDirOptions::default(), &mut writer) + .await + .expect_err("an upstream close after partial output is a failed walk"); + assert!(!error.is_metacache_output_stream_closed()); + assert!(error.contains_io_error_kind(io::ErrorKind::BrokenPipe)); + assert_eq!(writer, b"partial metacache"); + assert_eq!(transport.calls().len(), 1, "partial output must not be replayed"); + } + #[tokio::test] async fn walk_stream_backpressure_limits_read_ahead_and_is_cancellable() { let mut reader = Cursor::new(vec![5; 4 * DEFAULT_READ_BUFFER_SIZE]); diff --git a/crates/ecstore/src/diagnostics/admin_server_info.rs b/crates/ecstore/src/diagnostics/admin_server_info.rs index c1da84d80..ececeedec 100644 --- a/crates/ecstore/src/diagnostics/admin_server_info.rs +++ b/crates/ecstore/src/diagnostics/admin_server_info.rs @@ -682,15 +682,11 @@ async fn get_pools_info(all_disks: &[Disk]) -> Result> = HashMap::new(); - for d in all_disks { - let pool_info = pools_info.entry(d.pool_index).or_default(); - let erasure_set = pool_info.entry(d.set_index).or_default(); - - if erasure_set.id == 0 { - erasure_set.id = d.set_index; + collect_pools_info(all_disks, |pool_index, set_index| { + let store = store.clone(); + async move { let load = load_data_usage_cache( - store.pools[d.pool_index as usize].disk_set[d.set_index as usize].as_ref(), + store.pools[pool_index as usize].disk_set[set_index as usize].as_ref(), DATA_USAGE_CACHE_NAME, ); #[cfg(feature = "diagnostic-read-version-caller")] @@ -698,11 +694,34 @@ async fn get_pools_info(all_disks: &[Disk]) -> Result apply_erasure_set_usage(&cache, erasure_set), - Err(_) => erasure_set.usage_error = Some(DATA_USAGE_UNAVAILABLE_ERROR.to_string()), - } + load.await } + }) + .await +} + +async fn collect_pools_info(all_disks: &[Disk], mut load_usage: F) -> Result>> +where + F: FnMut(i32, i32) -> Fut, + Fut: std::future::Future>, +{ + let mut pools_info: HashMap> = HashMap::new(); + for d in all_disks { + let pool_info = pools_info.entry(d.pool_index).or_default(); + let erasure_set = match pool_info.entry(d.set_index) { + std::collections::hash_map::Entry::Occupied(entry) => entry.into_mut(), + std::collections::hash_map::Entry::Vacant(entry) => { + let erasure_set = entry.insert(ErasureSetInfo { + id: d.set_index, + ..Default::default() + }); + match load_usage(d.pool_index, d.set_index).await { + Ok(cache) => apply_erasure_set_usage(&cache, erasure_set), + Err(_) => erasure_set.usage_error = Some(DATA_USAGE_UNAVAILABLE_ERROR.to_string()), + } + erasure_set + } + }; erasure_set.raw_capacity += d.total_space; erasure_set.raw_usage += d.used_space; @@ -1241,6 +1260,59 @@ mod tests { assert_eq!(info.deployment_id, expected_deployment_id); } + #[tokio::test] + async fn pool_usage_is_loaded_once_per_set_including_set_zero() { + let mut disks = Vec::new(); + for (pool, set, count) in [(0, 0, 16), (1, 0, 16), (1, 1, 4)] { + disks.extend((0..count).map(|_| Disk { + pool_index: pool, + set_index: set, + used_space: 10, + total_space: 100, + ..Default::default() + })); + } + let mut loads = Vec::new(); + let info = super::collect_pools_info(&disks, |pool, set| { + loads.push((pool, set)); + async move { + if pool == 1 && set == 0 { + return Err(crate::error::Error::other("usage unavailable")); + } + let mut cache = rustfs_data_usage::DataUsageCache::default(); + cache.info.name = DATA_USAGE_ROOT.to_owned(); + cache.info.last_update = Some(std::time::SystemTime::UNIX_EPOCH); + cache.info.snapshot_complete = true; + cache.replace( + DATA_USAGE_ROOT, + "", + rustfs_data_usage::DataUsageEntry { + size: 25, + objects: 2, + ..Default::default() + }, + ); + Ok(cache) + } + }) + .await + .expect("collect mixed successful and unavailable usage"); + assert_eq!(loads, [(0, 0), (1, 0), (1, 1)]); + for (pool, set, count) in [(0, 0, 16), (1, 0, 16), (1, 1, 4)] { + let set_info = &info[&pool][&set]; + assert_eq!(set_info.id, set); + assert_eq!(set_info.raw_usage, count * 10); + assert_eq!(set_info.raw_capacity, count * 100); + if pool == 1 && set == 0 { + assert_eq!(set_info.usage_error.as_deref(), Some(DATA_USAGE_UNAVAILABLE_ERROR)); + } else { + assert_eq!(set_info.usage, 25, "logical usage must not multiply by drive count"); + assert_eq!(set_info.objects_count, 2); + assert!(set_info.usage_error.is_none()); + } + } + } + #[serial] #[tokio::test] async fn local_server_property_includes_runtime_stats_without_endpoint_pools() { diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index 8bce3d09c..a3ede4882 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -562,12 +562,57 @@ fn remove_dir_all_if_exists(path: &Path) -> std::io::Result<()> { fn snapshot_part_transaction_file(src: &Path, backup: &Path, absent: &Path) -> std::io::Result { match std::fs::symlink_metadata(src) { Ok(metadata) if metadata.is_file() => std::fs::hard_link(src, backup).map(|()| true), + Ok(metadata) if metadata.file_type().is_symlink() => { + Err(std::io::Error::new(ErrorKind::InvalidInput, "multipart transaction source is a symlink")) + } Ok(_) => Err(std::io::Error::new(ErrorKind::InvalidData, "multipart transaction source is not a file")), Err(err) if err.kind() == ErrorKind::NotFound => std::fs::write(absent, []).map(|()| false), Err(err) => Err(err), } } +struct PreparedPartTxnReceipt { + src_volume_dir: PathBuf, + dst_volume_dir: PathBuf, + src_file_path: PathBuf, + dst_file_path: PathBuf, + dst_meta_path: PathBuf, + transaction_path: PathBuf, + meta: Bytes, + had_old_meta: bool, +} + +struct PublishedPartTxnReceipt { + current_data_path: PathBuf, + current_meta_path: PathBuf, + transaction_path: PathBuf, + had_old_meta: bool, +} + +const PART_TXN_RECEIPTS_PER_SHARD: usize = 64; + +static PREPARED_PART_TXN_RECEIPTS: std::sync::LazyLock> = + std::sync::LazyLock::new(|| os::ShardedPathCache::new(PART_TXN_RECEIPTS_PER_SHARD)); + +static PUBLISHED_PART_TXN_RECEIPTS: std::sync::LazyLock> = + std::sync::LazyLock::new(|| os::ShardedPathCache::new(PART_TXN_RECEIPTS_PER_SHARD)); + +fn put_prepared_part_txn_receipt(transaction_path: &Path, entry: PreparedPartTxnReceipt) { + PREPARED_PART_TXN_RECEIPTS.insert(transaction_path, entry); +} + +fn take_prepared_part_txn_receipt(transaction_path: &Path) -> Option { + PREPARED_PART_TXN_RECEIPTS.remove(transaction_path) +} + +fn put_published_part_txn_receipt(transaction_path: &Path, entry: PublishedPartTxnReceipt) { + PUBLISHED_PART_TXN_RECEIPTS.insert(transaction_path, entry); +} + +fn take_published_part_txn_receipt(transaction_path: &Path) -> Option { + PUBLISHED_PART_TXN_RECEIPTS.remove(transaction_path) +} + fn restore_part_transaction_file(current: &Path, backup: &Path, absent: &Path, restore: &Path) -> std::io::Result<()> { os::record_native_publication(current); match std::fs::symlink_metadata(backup) { @@ -3165,6 +3210,10 @@ pub(super) fn has_rename_data_test_failpoints(dst_path: &str) -> bool { || os::has_dir_fsync_test_failures() || os::file_sync_probe::is_active() || os::prepared_publication_test_hooks::has_any_hooks() + || !OWNED_FILE_BEFORE_OPEN + .lock() + .expect("test file open hook lock should not be poisoned") + .is_empty() } #[cfg(all(not(test), feature = "test-util", target_os = "linux"))] @@ -9238,15 +9287,14 @@ impl LocalDisk { .await; } - let volume_dir = self.io_get_bucket_path(volume)?; - - let file_path = self.io_get_object_path(volume, path)?; + let (volume_dir, file_path) = self.io_get_bucket_and_object_path(volume, path)?; check_path_length(file_path.to_string_lossy().as_ref())?; let xl_path = path_join(&[file_path.as_path(), Path::new(STORAGE_FORMAT_FILE)]); + let object_path = self.get_object_path(volume, path)?; let mut namespace_owner: Option> = - Some(os::acquire_metadata_mutation_lease(&self.get_object_path(volume, path)?, namespace_owner).await); + Some(os::acquire_metadata_mutation_lease(&object_path, namespace_owner).await); if opts.recoverable_delete || (opts.undo_write && opts.undo_delete) { let volume_guard = os::disk_volume_mutation_lock(&self.root, volume).read_owned().await; namespace_owner = Some(Arc::new((namespace_owner, volume_guard))); @@ -9271,7 +9319,15 @@ impl LocalDisk { let rollback_dir = opts.old_data_dir; if let Some(rollback_dir) = rollback_dir && let Some(()) = self - .try_fused_recoverable_delete_version(volume, path, &fi, &opts, rollback_dir, namespace_owner.clone()) + .try_fused_recoverable_delete_version_with_paths( + volume, + path, + &fi, + &opts, + rollback_dir, + namespace_owner.clone(), + Some((object_path, volume_dir.clone(), file_path.clone())), + ) .await? { return Ok(()); @@ -14343,7 +14399,8 @@ impl DiskAPI for LocalDisk { all_cached = false; break; }; - let Ok(source) = self.io_get_object_path(bucket, multipart_staging::upload_scope_root(path_str, true)) else { + let Ok(source) = self.io_get_object_open_path(bucket, multipart_staging::upload_scope_root(path_str, true)) + else { all_cached = false; break; }; @@ -14470,8 +14527,8 @@ impl DiskAPI for LocalDisk { multipart_staging::reject_noncanonical_scope(&self.root, self.io_root(), src_volume, src_path)?; let source_owner = self.multipart_source_owner(dst_volume, dst_path, true, false, None).await?; let admission = self.multipart_source_admission(dst_volume, dst_path, true, false)?; - let src_volume_dir = self.io_get_bucket_path(src_volume)?; - let dst_volume_dir = self.io_get_bucket_path(dst_volume)?; + let (src_volume_dir, src_file_path) = self.io_get_bucket_and_object_path(src_volume, src_path)?; + let (dst_volume_dir, dst_file_path) = self.io_get_bucket_and_object_path(dst_volume, dst_path)?; if !skip_access_checks(src_volume) { super::fs::access_std(&src_volume_dir).map_err(|err| to_access_error(err, DiskError::VolumeAccessDenied))?; } @@ -14479,10 +14536,8 @@ impl DiskAPI for LocalDisk { super::fs::access_std(&dst_volume_dir).map_err(|err| to_access_error(err, DiskError::VolumeAccessDenied))?; } - let src_file_path = self.io_get_object_path(src_volume, src_path)?; - let dst_file_path = self.io_get_object_path(dst_volume, dst_path)?; - let dst_meta_path = self.io_get_object_path(dst_volume, &format!("{dst_path}.meta"))?; - let transaction_path = self.io_get_object_path(dst_volume, &crate::disk::part_transaction_path(dst_path))?; + let dst_meta_path = self.io_get_object_open_path(dst_volume, &format!("{dst_path}.meta"))?; + let transaction_path = self.io_get_object_open_path(dst_volume, &crate::disk::part_transaction_path(dst_path))?; for path in [&src_file_path, &dst_file_path, &dst_meta_path, &transaction_path] { check_path_length(path.to_string_lossy().as_ref())?; } @@ -14528,8 +14583,11 @@ impl DiskAPI for LocalDisk { if !source.is_file() { return Ok(Err(DiskError::FileAccessDenied)); } - if transaction_path.exists() { - return Ok(Err(DiskError::FileAccessDenied)); + match std::fs::symlink_metadata(&transaction_path) { + Err(err) if err.kind() == ErrorKind::NotFound => {} + Ok(m) if m.file_type().is_symlink() => return Ok(Err(DiskError::InvalidPath)), + Ok(_) => return Ok(Err(DiskError::FileAccessDenied)), + Err(err) => return Ok(Err(to_file_error(err).into())), } if let Some((integrity, source, destination, expected_len)) = integrity_prep { let stat = match std::fs::symlink_metadata(&source) { @@ -14586,7 +14644,7 @@ impl DiskAPI for LocalDisk { return Ok(Err(to_file_error(err).into())); } - let prepare_result = (|| -> std::io::Result<()> { + let prepare_result = (|| -> std::io::Result { let had_old_data = snapshot_part_transaction_file( &dst_file_path, &staging_path.join(PART_TRANSACTION_OLD_DATA), @@ -14598,10 +14656,11 @@ impl DiskAPI for LocalDisk { &staging_path.join(PART_TRANSACTION_OLD_META_ABSENT), )?; + let new_meta_path = staging_path.join(PART_TRANSACTION_NEW_META); let mut new_meta = std::fs::OpenOptions::new() .create_new(true) .write(true) - .open(staging_path.join(PART_TRANSACTION_NEW_META))?; + .open(&new_meta_path)?; std::io::Write::write_all(&mut new_meta, &meta)?; if durability.syncs_commit_metadata() { new_meta.sync_data()?; @@ -14609,6 +14668,7 @@ impl DiskAPI for LocalDisk { // Windows rejects renaming a directory while one of its children is // still open, even when the child handle shares delete access. drop(new_meta); + let _ = std::fs::hard_link(&new_meta_path, staging_path.join(PART_TRANSACTION_PUBLISH_META)); let sync_overwrite_backup = durability.syncs_commit_metadata() && (had_old_data || had_old_meta); if sync_overwrite_backup { os::fsync_dir_std(&staging_path)?; @@ -14617,13 +14677,29 @@ impl DiskAPI for LocalDisk { if sync_overwrite_backup { os::fsync_dir_std(transaction_parent)?; } - Ok(()) + Ok(had_old_meta) })(); - if let Err(err) = prepare_result { - let _ = remove_dir_all_if_exists(&staging_path); - return Ok(Err(to_file_error(err).into())); - } + let had_old_meta = match prepare_result { + Ok(had_old_meta) => had_old_meta, + Err(err) => { + let _ = remove_dir_all_if_exists(&staging_path); + return Ok(Err(to_file_error(err).into())); + } + }; + put_prepared_part_txn_receipt( + &transaction_path, + PreparedPartTxnReceipt { + src_volume_dir, + dst_volume_dir, + src_file_path, + dst_file_path, + dst_meta_path, + transaction_path: transaction_path.clone(), + meta, + had_old_meta, + }, + ); Ok(Ok(())) }) @@ -14642,13 +14718,28 @@ impl DiskAPI for LocalDisk { let source_owner = self.multipart_source_owner(volume, path, true, false, None).await?; let admission = self.multipart_source_admission(volume, path, true, false)?; let admission_source = admission.as_ref().map(|a| a.source().to_path_buf()); - self.io_get_bucket_path(volume)?; - let current_data_path = self.io_get_object_path(volume, path)?; - let current_meta_path = self.io_get_object_path(volume, &format!("{path}.meta"))?; - let transaction_path = self.io_get_object_path(volume, &crate::disk::part_transaction_path(path))?; - for candidate in [¤t_data_path, ¤t_meta_path, &transaction_path] { - check_path_length(candidate.to_string_lossy().as_ref())?; - } + let txn_rel_path = crate::disk::part_transaction_path(path); + let cached_published = self + .io_get_object_open_path(volume, &txn_rel_path) + .ok() + .and_then(|txn_open| take_published_part_txn_receipt(&txn_open)); + let (current_data_path, current_meta_path, transaction_path, skip_obsolete_read_on_commit) = match cached_published { + Some(entry) => ( + entry.current_data_path, + entry.current_meta_path, + entry.transaction_path, + !entry.had_old_meta, + ), + None => { + let (_, current_data_path) = self.io_get_bucket_and_object_path(volume, path)?; + let current_meta_path = self.io_get_object_path(volume, &format!("{path}.meta"))?; + let transaction_path = self.io_get_object_path(volume, &txn_rel_path)?; + for candidate in [¤t_data_path, ¤t_meta_path, &transaction_path] { + check_path_length(candidate.to_string_lossy().as_ref())?; + } + (current_data_path, current_meta_path, transaction_path, false) + } + }; let durability = effective_durability(volume); let sync = durability.syncs_commit_metadata(); let part_number = current_data_path @@ -14715,7 +14806,8 @@ impl DiskAPI for LocalDisk { PartTransactionAction::Commit => (PART_TRANSACTION_NEW_META, PART_TRANSACTION_OLD_META), PartTransactionAction::Rollback => (PART_TRANSACTION_OLD_META, PART_TRANSACTION_NEW_META), }; - if let Ok(obsolete) = std::fs::read(transaction_path.join(obsolete_name)) + if !(action == PartTransactionAction::Commit && skip_obsolete_read_on_commit && !cfg!(test)) + && let Ok(obsolete) = std::fs::read(transaction_path.join(obsolete_name)) && let Ok(obsolete) = ObjectPartInfo::unmarshal(&obsolete) && let Some(integrity) = obsolete.integrity && integrity.validate().is_ok() @@ -14781,15 +14873,6 @@ impl DiskAPI for LocalDisk { let admission_source = admission.as_ref().map(|a| a.source().to_path_buf()); pool_meta_native::reject_raw_path(&self.root, src_volume, src_path)?; pool_meta_native::reject_raw_path(&self.root, dst_volume, dst_path)?; - let src_volume_dir = self.io_get_bucket_path(src_volume)?; - let dst_volume_dir = self.io_get_bucket_path(dst_volume)?; - if !skip_access_checks(src_volume) { - super::fs::access_std(&src_volume_dir).map_err(|e| to_access_error(e, DiskError::VolumeAccessDenied))? - } - if !skip_access_checks(dst_volume) { - super::fs::access_std(&dst_volume_dir).map_err(|e| to_access_error(e, DiskError::VolumeAccessDenied))? - } - let src_is_dir = has_suffix(src_path, SLASH_SEPARATOR); let dst_is_dir = has_suffix(dst_path, SLASH_SEPARATOR); @@ -14806,25 +14889,74 @@ impl DiskAPI for LocalDisk { return Err(DiskError::FileAccessDenied); } - let src_file_path = self.io_get_object_path(src_volume, src_path)?; - let dst_file_path = self.io_get_object_path(dst_volume, dst_path)?; - - // warn!("rename_part src_file_path:{:?}, dst_file_path:{:?}", &src_file_path, &dst_file_path); - - check_path_length(src_file_path.to_string_lossy().as_ref())?; - check_path_length(dst_file_path.to_string_lossy().as_ref())?; - - let dst_meta_path = if src_is_dir { - None + let cached_prepared = if !src_is_dir { + let txn_rel = crate::disk::part_transaction_path(dst_path); + self.io_get_object_open_path(dst_volume, &txn_rel) + .ok() + .and_then(|txn_open| take_prepared_part_txn_receipt(&txn_open)) + .filter(|entry| { + entry.meta == meta + && self + .io_get_object_open_path(src_volume, src_path) + .is_ok_and(|src_open| src_open == entry.src_file_path) + }) } else { - let path = self.io_get_object_path(dst_volume, &format!("{dst_path}.meta"))?; - check_path_length(path.to_string_lossy().as_ref())?; - Some(path) + None }; - let transaction_path = if src_is_dir { - None - } else { - Some(self.io_get_object_path(dst_volume, &crate::disk::part_transaction_path(dst_path))?) + + let ( + src_volume_dir, + dst_volume_dir, + src_file_path, + dst_file_path, + dst_meta_path, + transaction_path, + preverified_meta_and_old_flag, + ) = match cached_prepared { + Some(entry) => ( + entry.src_volume_dir, + entry.dst_volume_dir, + entry.src_file_path, + entry.dst_file_path, + Some(entry.dst_meta_path), + Some(entry.transaction_path), + Some(entry.had_old_meta), + ), + None => { + let (src_volume_dir, src_file_path) = self.io_get_bucket_and_object_path(src_volume, src_path)?; + let (dst_volume_dir, dst_file_path) = self.io_get_bucket_and_object_path(dst_volume, dst_path)?; + if !skip_access_checks(src_volume) { + super::fs::access_std(&src_volume_dir).map_err(|e| to_access_error(e, DiskError::VolumeAccessDenied))? + } + if !skip_access_checks(dst_volume) { + super::fs::access_std(&dst_volume_dir).map_err(|e| to_access_error(e, DiskError::VolumeAccessDenied))? + } + + check_path_length(src_file_path.to_string_lossy().as_ref())?; + check_path_length(dst_file_path.to_string_lossy().as_ref())?; + + let dst_meta_path = if src_is_dir { + None + } else { + let path = self.io_get_object_path(dst_volume, &format!("{dst_path}.meta"))?; + check_path_length(path.to_string_lossy().as_ref())?; + Some(path) + }; + let transaction_path = if src_is_dir { + None + } else { + Some(self.io_get_object_path(dst_volume, &crate::disk::part_transaction_path(dst_path))?) + }; + ( + src_volume_dir, + dst_volume_dir, + src_file_path, + dst_file_path, + dst_meta_path, + transaction_path, + None, + ) + } }; // UploadPart is acknowledged once this rename lands, so the part data and @@ -14883,35 +15015,43 @@ impl DiskAPI for LocalDisk { let transaction_publish_meta = if let Some(transaction_path) = transaction_path.as_ref() { let transaction_meta_path = transaction_path.join(PART_TRANSACTION_NEW_META); - match std::fs::File::open(&transaction_meta_path) { - Ok(mut file) => { - let file_len = file.metadata().map_err(to_file_error)?.len(); - if file_len != meta.len() as u64 { - return Err(DiskError::FileCorrupt.into()); - } - let mut expected_meta = Vec::with_capacity(meta.len()); - std::io::Read::read_to_end(&mut std::io::Read::take(&mut file, file_len), &mut expected_meta) - .map_err(to_file_error)?; - drop(file); - if expected_meta.as_slice() != meta.as_ref() { - return Err(DiskError::FileCorrupt.into()); - } + let publish_meta_path = transaction_path.join(PART_TRANSACTION_PUBLISH_META); + if !cfg!(test) + && preverified_meta_and_old_flag.is_some() + && publish_meta_path.is_file() + && std::fs::metadata(&transaction_meta_path).is_ok_and(|m| m.is_file() && m.len() == meta.len() as u64) + { + Some(publish_meta_path) + } else { + match std::fs::File::open(&transaction_meta_path) { + Ok(mut file) => { + let file_len = file.metadata().map_err(to_file_error)?.len(); + if file_len != meta.len() as u64 { + return Err(DiskError::FileCorrupt.into()); + } + let mut expected_meta = Vec::with_capacity(meta.len()); + std::io::Read::read_to_end(&mut std::io::Read::take(&mut file, file_len), &mut expected_meta) + .map_err(to_file_error)?; + drop(file); + if expected_meta.as_slice() != meta.as_ref() { + return Err(DiskError::FileCorrupt.into()); + } - let publish_meta_path = transaction_path.join(PART_TRANSACTION_PUBLISH_META); - remove_file_if_exists(&publish_meta_path).map_err(to_file_error)?; - std::fs::hard_link(&transaction_meta_path, &publish_meta_path).map_err(to_file_error)?; - Some(publish_meta_path) + remove_file_if_exists(&publish_meta_path).map_err(to_file_error)?; + std::fs::hard_link(&transaction_meta_path, &publish_meta_path).map_err(to_file_error)?; + Some(publish_meta_path) + } + // Old peers know only RenamePart. The new coordinator never + // reaches rename unless prepare succeeded, so an absent + // transaction directory identifies the rolling-upgrade legacy + // path. A present directory without new.meta is corruption. + Err(err) if err.kind() == ErrorKind::NotFound => match std::fs::metadata(transaction_path) { + Ok(_) => return Err(DiskError::FileCorrupt.into()), + Err(meta_err) if meta_err.kind() == ErrorKind::NotFound => None, + Err(meta_err) => return Err(to_file_error(meta_err)), + }, + Err(err) => return Err(to_file_error(err)), } - // Old peers know only RenamePart. The new coordinator never - // reaches rename unless prepare succeeded, so an absent - // transaction directory identifies the rolling-upgrade legacy - // path. A present directory without new.meta is corruption. - Err(err) if err.kind() == ErrorKind::NotFound => match std::fs::metadata(transaction_path) { - Ok(_) => return Err(DiskError::FileCorrupt.into()), - Err(meta_err) if meta_err.kind() == ErrorKind::NotFound => None, - Err(meta_err) => return Err(to_file_error(meta_err)), - }, - Err(err) => return Err(to_file_error(err)), } } else { None @@ -14974,6 +15114,19 @@ impl DiskAPI for LocalDisk { } } } + if let (Some(had_old_meta), Some(txn_path), Some(meta_path)) = + (preverified_meta_and_old_flag, transaction_path.as_ref(), dst_meta_path.as_path().into()) + { + put_published_part_txn_receipt( + txn_path, + PublishedPartTxnReceipt { + current_data_path: dst_file_path, + current_meta_path: meta_path.to_path_buf(), + transaction_path: txn_path.clone(), + had_old_meta, + }, + ); + } Ok(true) } else { Ok(false) diff --git a/crates/ecstore/src/disk/local/commit.rs b/crates/ecstore/src/disk/local/commit.rs index 3d6428789..95eafcfa1 100644 --- a/crates/ecstore/src/disk/local/commit.rs +++ b/crates/ecstore/src/disk/local/commit.rs @@ -15,8 +15,6 @@ //! Single-disk object rename publication and rollback. The shared execution core //! retains instrumentation, mutation leases, and commit guards through the syscall. -#[cfg(target_os = "linux")] -use super::has_rename_data_test_failpoints; #[cfg(all(test, windows))] use super::run_destination_commit_directory_preparation; #[cfg(any(not(windows), test))] @@ -24,8 +22,8 @@ use super::should_fail_local_inline_rollback_hardlink; use super::{ EVENT_DISK_LOCAL_ACCESS_FAILED, EVENT_DISK_LOCAL_HEAL_PURGE_FAILED, EVENT_DISK_LOCAL_RENAME_REJECTED, LOG_COMPONENT_ECSTORE, LOG_SUBSYSTEM_DISK_LOCAL, LocalDisk, SyncMode, effective_durability, has_inline_commit_test_failpoints, - inline_metadata_rollback_dir, observe_old_current, remove_dir_all_if_exists, remove_dst_base_before_commit, - remove_file_if_exists, rename_data_versions_signature, run_inline_preparation_before_backup, + has_rename_data_test_failpoints, inline_metadata_rollback_dir, observe_old_current, remove_dir_all_if_exists, + remove_dst_base_before_commit, remove_file_if_exists, rename_data_versions_signature, run_inline_preparation_before_backup, should_fail_after_metadata_commit, should_fail_before_old_metadata_backup, should_fail_commit_rename, should_remove_staged_meta_before_commit, skip_access_checks, tmp_staging_mutation_path, }; @@ -172,6 +170,44 @@ impl RenameCommitContext<'_> { } } +#[cfg(any(not(windows), test))] +fn inline_rollback_backup_path(staging_file_path: &Path) -> std::io::Result { + let Some(staging_parent) = staging_file_path.parent() else { + return Err(std::io::Error::new(ErrorKind::InvalidInput, "missing staging metadata parent")); + }; + let backup_path = if staging_file_path.file_name().and_then(|n| n.to_str()) == Some(STORAGE_FORMAT_FILE) { + staging_parent.join(STORAGE_FORMAT_FILE_BACKUP) + } else { + staging_file_path.with_extension("meta.bkp") + }; + remove_file_if_exists(&backup_path)?; + Ok(backup_path) +} + +#[cfg(any(not(windows), test))] +fn write_local_inline_rollback_backup_from_bytes( + staging_file_path: &Path, + old_metadata: &[u8], + sync: bool, +) -> std::io::Result { + let backup_path = inline_rollback_backup_path(staging_file_path)?; + let write_res = (|| -> std::io::Result<()> { + #[cfg(test)] + run_owned_file_before_open(&backup_path); + let mut file = std::fs::File::create(&backup_path)?; + std::io::Write::write_all(&mut file, old_metadata)?; + if sync && !old_metadata.is_empty() { + file.sync_data()?; + } + Ok(()) + })(); + if let Err(err) = write_res { + let _ = remove_file_if_exists(&backup_path); + return Err(err); + } + Ok(backup_path) +} + #[cfg(any(not(windows), test))] pub(super) fn create_local_inline_rollback_backup( dst_file_path: &Path, @@ -179,27 +215,12 @@ pub(super) fn create_local_inline_rollback_backup( old_metadata: &[u8], sync: bool, ) -> std::io::Result { - let Some(staging_parent) = staging_file_path.parent() else { - return Err(std::io::Error::new(ErrorKind::InvalidInput, "missing staging metadata parent")); - }; - let backup_path = staging_parent.join(STORAGE_FORMAT_FILE_BACKUP); - remove_file_if_exists(&backup_path)?; + let backup_path = inline_rollback_backup_path(staging_file_path)?; if should_fail_local_inline_rollback_hardlink(dst_file_path) || std::fs::hard_link(dst_file_path, &backup_path).is_err() { - let write_res = (|| -> std::io::Result<()> { - #[cfg(test)] - run_owned_file_before_open(&backup_path); - let mut file = std::fs::File::create(&backup_path)?; - std::io::Write::write_all(&mut file, old_metadata)?; - if sync && !old_metadata.is_empty() { - file.sync_data()?; - } - Ok(()) - })(); - if let Err(err) = write_res { - let _ = remove_file_if_exists(&backup_path); - return Err(err); - } - } else if sync && !old_metadata.is_empty() { + return write_local_inline_rollback_backup_from_bytes(staging_file_path, old_metadata, sync); + } else if cfg!(test) && sync && !old_metadata.is_empty() { + // Skip redundant fdatasync on hardlinked rollback backups in production; + // keep a Windows-safe sync open in tests. os::open_file_for_sync_std(&backup_path)?.sync_data()?; } Ok(backup_path) @@ -658,7 +679,14 @@ impl LocalDisk { }) .collect() }; - let src_volume_dir = self.io_get_bucket_path(src_volume)?; + let src_volume_dir = if cfg!(target_os = "linux") + && src_volume == super::super::RUSTFS_META_TMP_BUCKET + && skip_access_checks(src_volume) + { + self.io_root().join(src_volume) + } else { + self.io_get_bucket_path(src_volume)? + }; if !skip_access_checks(src_volume) && let Err(e) = super::super::fs::access_std(&src_volume_dir) { @@ -702,8 +730,24 @@ impl LocalDisk { } else { STORAGE_FORMAT_FILE }; + let flat_inline_tmp = cfg!(target_os = "linux") + && !no_inline + && src_volume == super::super::RUSTFS_META_TMP_BUCKET + && !preserve_upload_metadata + && multipart_publication.is_none() + && pool_meta_condition.is_none() + && !fi_healing + && !os::dst_dir_fsync_group_commit_enabled() + && !os::file_fdatasync_group_commit_enabled() + && !has_rename_data_test_failpoints(dst_path); let src_file_path = match multipart_publication.as_ref() { Some(publication) => publication.canonical_head(), + None if flat_inline_tmp => { + self.io_get_object_open_path(src_volume, format!("{src_path}.{source_metadata_name}").as_str())? + } + None if cfg!(target_os = "linux") => { + self.io_get_object_open_path(src_volume, format!("{src_path}/{source_metadata_name}").as_str())? + } None => self.io_get_object_path(src_volume, format!("{src_path}/{source_metadata_name}").as_str())?, }; let dst_file_path = self.io_get_object_open_path(dst_volume, format!("{}/{}", dst_path, STORAGE_FORMAT_FILE).as_str())?; @@ -726,14 +770,13 @@ impl LocalDisk { .data_dir .map(|dir| rustfs_utils::path::retain_slash(dir.to_string().as_str())) { - let src_data_path = self.io_get_object_path( - src_volume, - rustfs_utils::path::retain_slash(format!("{}/{}", src_path, data_dir).as_str()).as_str(), - )?; - let dst_data_path = self.io_get_object_path( - dst_volume, - rustfs_utils::path::retain_slash(format!("{}/{}", dst_path, data_dir).as_str()).as_str(), - )?; + let src_rel = rustfs_utils::path::retain_slash(format!("{}/{}", src_path, data_dir).as_str()); + let src_data_path = if cfg!(target_os = "linux") && multipart_publication.is_none() && !fi_healing { + self.io_get_object_open_path(src_volume, src_rel.as_str())? + } else { + self.io_get_object_path(src_volume, src_rel.as_str())? + }; + let dst_data_path = destination_object_path.join(data_dir.trim_matches('/')); Some((src_data_path, dst_data_path)) } else { @@ -816,7 +859,7 @@ impl LocalDisk { let fused_attempt = move |is_cancelled: &dyn Fn() -> bool| -> std::io::Result { let (guard, has_dst_buf) = prepare_rename_commit_and_destination_metadata_std( - !no_inline, + !no_inline && !flat_inline_tmp, &src_parent_owned, &dst_parent_owned, &dst_file_path_owned, @@ -888,7 +931,16 @@ impl LocalDisk { run_owned_file_before_open(&src_file_path_owned); let mut prepared_metadata_source = os::create_prepared_rename_source_with_commit_guard(&src_file_path_owned, &dst_file_path_owned, &guard)?; - prepared_metadata_source.write_all(&new_dst_buf, false)?; + let cleanup_flat_tmp = || { + if flat_inline_tmp { + let _ = remove_file_if_exists(&src_file_path_owned); + } + }; + if let Err(err) = prepared_metadata_source.write_all(&new_dst_buf, false) { + drop(prepared_metadata_source); + cleanup_flat_tmp(); + return Err(err); + } if !no_inline { run_inline_preparation_before_backup(&dst_path_str); } @@ -896,14 +948,21 @@ impl LocalDisk { let mut staged_rollback_path = None; if (!no_inline || rollback_data_dir.is_none()) && let Some(ref old_metadata) = has_dst_buf - && (rollback_data_dir.is_some() || sync_commit_metadata || cfg!(test)) + && (rollback_data_dir.is_some() || cfg!(test)) { - let backup_path = create_local_inline_rollback_backup( + let backup_path = match create_local_inline_rollback_backup( &dst_file_path_owned, &src_file_path_owned, old_metadata, sync_commit_metadata, - )?; + ) { + Ok(path) => path, + Err(err) => { + drop(prepared_metadata_source); + cleanup_flat_tmp(); + return Err(err); + } + }; staged_rollback_path = Some(StagedRollbackBackup(backup_path)); } @@ -932,6 +991,8 @@ impl LocalDisk { return Err(err); } } else if sync_commit_metadata && let Err(err) = prepared_metadata_source.sync_data() { + drop(prepared_metadata_source); + cleanup_flat_tmp(); if let Some(backup_path) = staged_rollback_path.as_ref().map(StagedRollbackBackup::path) { let _ = remove_file_if_exists(backup_path); } @@ -940,6 +1001,7 @@ impl LocalDisk { if is_cancelled() { drop(prepared_metadata_source); + cleanup_flat_tmp(); if let Some(backup_path) = staged_rollback_path.as_ref().map(StagedRollbackBackup::path) { let _ = remove_file_if_exists(backup_path); } @@ -956,6 +1018,8 @@ impl LocalDisk { ) { Ok(guard) => guard, Err(err) => { + drop(prepared_metadata_source); + cleanup_flat_tmp(); if let Some(backup_path) = staged_rollback_path.as_ref().map(StagedRollbackBackup::path) { let _ = remove_file_if_exists(backup_path); } @@ -1039,6 +1103,8 @@ impl LocalDisk { None, sync_commit_metadata, ) { + drop(prepared_metadata_source); + cleanup_flat_tmp(); let _ = remove_file_if_exists(staged_backup); return Ok(FusedCommitAttempt::CommitErr(to_file_error(err).into())); } @@ -1059,6 +1125,7 @@ impl LocalDisk { ) }; if let Err(err) = rename_res { + cleanup_flat_tmp(); if let Some(backup_path) = staged_rollback_path.as_ref().map(StagedRollbackBackup::path) { let _ = remove_file_if_exists(backup_path); } @@ -1087,8 +1154,38 @@ impl LocalDisk { return Ok(FusedCommitAttempt::CommitErr(DiskError::Unexpected)); } - if sync_commit_metadata - && let Err(err) = fsync_inline_commit_std( + if sync_commit_metadata { + if staged_rollback_path.is_none() + && rollback_data_dir.is_none() + && let Some(ref old_metadata) = has_dst_buf + && (!no_inline || rollback_data_dir.is_none()) + { + let fsync_res = (|| -> std::io::Result<()> { + if let Some(dst_parent) = dst_file_path_owned.parent() { + os::fsync_commit_directory_limited_std(dst_parent, &guard)?; + } + Ok(()) + })(); + if let Err(err) = fsync_res { + let rollback_ok = + write_local_inline_rollback_backup_from_bytes(&src_file_path_owned, old_metadata, true) + .and_then(|backup_path| { + let backup = StagedRollbackBackup(backup_path); + let res = + rollback_inline_metadata_std(&dst_file_path_owned, rollback_data_dir, Some(&backup)); + let _ = remove_file_if_exists(backup.path()); + res + }) + .is_ok(); + if no_inline { + if rollback_ok && let Some((_, new_data_path)) = has_data_dir_path_owned.as_ref() { + let _ = remove_dir_all_if_exists(new_data_path); + } + return Ok(FusedCommitAttempt::CommitErr(to_file_error(err).into())); + } + return Ok(FusedCommitAttempt::CommitErr(DiskError::from(err))); + } + } else if let Err(err) = fsync_inline_commit_std( &dst_file_path_owned, &dst_volume_dir_owned, destination_was_absent, @@ -1099,19 +1196,19 @@ impl LocalDisk { .filter(|_| no_inline) .map(|(_, new_data_path)| new_data_path.as_path()), &guard, - ) - { - if no_inline { - return Ok(FusedCommitAttempt::CommitErr(to_file_error(err).into())); + ) { + if no_inline { + return Ok(FusedCommitAttempt::CommitErr(to_file_error(err).into())); + } + return Ok(FusedCommitAttempt::CommitErr(DiskError::from(err))); } - return Ok(FusedCommitAttempt::CommitErr(DiskError::from(err))); } if let Some(backup_path) = staged_rollback_path.as_ref().map(StagedRollbackBackup::path) { let _ = remove_file_if_exists(backup_path); } - if let Some(parent) = src_file_path_owned.parent() { + if !flat_inline_tmp && let Some(parent) = src_file_path_owned.parent() { let _ = std::fs::remove_dir(parent); } @@ -1376,7 +1473,7 @@ impl LocalDisk { } } } - match publication_owned.finish_on_thread(sync_commit_metadata, staging_cleanup) { + match publication_owned.finish_on_thread(sync_commit_metadata, !cfg!(test), staging_cleanup) { Ok(result) => Ok(MultipartFusedCommitOutcome::Committed(Box::new(RenameDataResp { multipart_source: Some(Box::new(result)), old_data_dir: has_old_data_dir, diff --git a/crates/ecstore/src/disk/local/delete_recovery.rs b/crates/ecstore/src/disk/local/delete_recovery.rs index 28675d411..e4a5c2d20 100644 --- a/crates/ecstore/src/disk/local/delete_recovery.rs +++ b/crates/ecstore/src/disk/local/delete_recovery.rs @@ -856,6 +856,12 @@ pub(super) fn inspection_with_markers_bounded( let prepared_digest = prepared.as_deref().map(digest); let target_digest = target.as_deref().map(digest); let receipt_digest = receipt_bytes.as_deref().map(digest); + let finished = finished + || (committed + && prepared.is_none() + && before.is_none() + && target.is_none() + && current_digest == journal.as_ref().and_then(|j| j.target_digest)); if let Some(journal) = &journal { if (!finished && journal.target_digest != target_digest @@ -1113,13 +1119,34 @@ pub(super) fn inspection_with_markers_bounded( Ok((inspection, has_physical_data_dirs_or_markers)) } -type PreparedDeleteReceiptEntry = (DeleteRecoveryJournal, bool); +#[derive(Clone)] +struct PreparedDeleteReceiptEntry { + object_path: PathBuf, + bucket_path: PathBuf, + file_path: PathBuf, + staging_root: PathBuf, + journal_raw: Vec, + journal: DeleteRecoveryJournal, + sole_transaction: bool, +} + +#[derive(Clone)] +struct PublishedDeleteReceiptEntry { + object_path: PathBuf, + bucket_path: PathBuf, + file_path: PathBuf, + staging_root: PathBuf, + journal_raw: Vec, +} const PREPARED_DELETE_RECEIPTS_PER_SHARD: usize = 64; static PREPARED_DELETE_RECEIPTS: std::sync::LazyLock> = std::sync::LazyLock::new(|| os::ShardedPathCache::new(PREPARED_DELETE_RECEIPTS_PER_SHARD)); +static PUBLISHED_DELETE_RECEIPTS: std::sync::LazyLock> = + std::sync::LazyLock::new(|| os::ShardedPathCache::new(PREPARED_DELETE_RECEIPTS_PER_SHARD)); + fn put_prepared_delete_receipt(rollback: impl AsRef, entry: PreparedDeleteReceiptEntry) { PREPARED_DELETE_RECEIPTS.insert(rollback.as_ref(), entry); } @@ -1128,6 +1155,27 @@ fn take_prepared_delete_receipt(rollback: &Path) -> Option, entry: PublishedDeleteReceiptEntry) { + PUBLISHED_DELETE_RECEIPTS.insert(rollback.as_ref(), entry); +} + +fn take_published_delete_receipt(rollback: &Path) -> Option { + PUBLISHED_DELETE_RECEIPTS.remove(rollback) +} + +fn format_commit_receipt_bytes(journal_raw: &[u8], finished: bool) -> Vec { + let suffix: &[u8] = if finished { + b",\"finished\":true}" + } else { + b",\"finished\":false}" + }; + let mut out = Vec::with_capacity(11 + journal_raw.len() + suffix.len()); + out.extend_from_slice(b"{\"journal\":"); + out.extend_from_slice(journal_raw); + out.extend_from_slice(suffix); + out +} + impl LocalDisk { // Native MPU source closure has its own conditional protocol. Generic // object recovery must never restore its head or acquire SourceRW under @@ -2376,6 +2424,7 @@ impl LocalDisk { barrier(volume, object, transaction, stage, &self.root).await } + #[cfg(test)] pub(super) async fn try_fused_recoverable_delete_version( &self, volume: &str, @@ -2384,6 +2433,21 @@ impl LocalDisk { opts: &DeleteOptions, transaction: Uuid, owner: Option>, + ) -> Result> { + self.try_fused_recoverable_delete_version_with_paths(volume, object, fi, opts, transaction, owner, None) + .await + } + + #[expect(clippy::too_many_arguments)] + pub(super) async fn try_fused_recoverable_delete_version_with_paths( + &self, + volume: &str, + object: &str, + fi: &FileInfo, + opts: &DeleteOptions, + transaction: Uuid, + owner: Option>, + prevalidated_paths: Option<(PathBuf, PathBuf, PathBuf)>, ) -> Result> { if !opts.recoverable_delete || opts.undo_write @@ -2396,11 +2460,22 @@ impl LocalDisk { self.reject_multipart_delete_recovery_mutation(volume, object)?; self.wait_for_startup_cleanup().await; - let (_, file_path) = self.io_get_bucket_and_object_path(volume, object)?; - check_path_length(file_path.to_string_lossy().as_ref())?; + let (object_path, bucket_path, file_path) = match prevalidated_paths { + Some(paths) => paths, + None => { + let obj = self.get_object_path(volume, object)?; + let (b, f) = self.io_get_bucket_and_object_path(volume, object)?; + check_path_length(f.to_string_lossy().as_ref())?; + (obj, b, f) + } + }; let xl_path = path_join(&[file_path.as_path(), Path::new(STORAGE_FORMAT_FILE)]); let destination = file_path.join(transaction.to_string()); - let staging_root = self.io_get_bucket_path(RUSTFS_META_TMP_BUCKET)?; + let staging_root = if cfg!(target_os = "linux") { + self.io_root().join(RUSTFS_META_TMP_BUCKET) + } else { + self.io_get_bucket_path(RUSTFS_META_TMP_BUCKET)? + }; let sync = effective_durability(volume).syncs_commit_metadata(); let incarnation = opts.delete_recovery_bucket_incarnation; let lease = os::acquire_namespace_mutation_lease_with_owner(&destination, owner).await; @@ -2588,6 +2663,8 @@ impl LocalDisk { let _ = std::fs::remove_dir_all(&destination); return Err(err); } + #[cfg(all(test, unix))] + fused_prepare_checkpoint::run(&file_path); let mut reserved_data_path = None; if let Some(old_path) = old_data_path { let reserved_marker = old_path.join(format!("{RESERVED_DELETE_DATA_DIR_MARKER_PREFIX}{transaction}")); @@ -2607,6 +2684,7 @@ impl LocalDisk { reserved_data_path = Some((old_path, reserved_marker)); } if let Some(ref target_bytes) = target { + // A provisional head must stay outside the published recovery transaction. let target_in_rollback = destination.join(TARGET); let linked_xl = staging_root.join(format!("delete-xl-{transaction}")); let _ = std::fs::remove_file(&linked_xl); @@ -2655,7 +2733,18 @@ impl LocalDisk { Err(err) => return Err(err), } } - put_prepared_delete_receipt(destination, (journal, sole_transaction)); + put_prepared_delete_receipt( + destination, + PreparedDeleteReceiptEntry { + object_path, + bucket_path, + file_path, + staging_root, + journal_raw, + journal, + sole_transaction, + }, + ); Ok(Ok(Some(old_dir))) }, ) @@ -2692,19 +2781,45 @@ impl LocalDisk { namespace_owner: Option>, work_budget: Option>, ) -> Result> { - self.reject_multipart_delete_recovery_mutation(volume, object)?; - if !safe_path(object, false) || transaction.is_nil() { + if transaction.is_nil() { return Err(DiskError::InvalidPath); } - let object_path = self.get_object_path(volume, object)?; + let rollback_open_path = if volume != RUSTFS_META_MULTIPART_BUCKET { + self.io_get_object_open_path(volume, object) + .ok() + .map(|p| p.join(transaction.to_string())) + } else { + None + }; + let cached_prepared = rollback_open_path.as_deref().and_then(take_prepared_delete_receipt); + let (object_path, bucket_path, root, staging_root, cached_receipt) = match cached_prepared { + Some(entry) => ( + entry.object_path, + Some(entry.bucket_path), + entry.file_path, + entry.staging_root, + Some((entry.journal_raw, entry.journal, entry.sole_transaction)), + ), + None => { + self.reject_multipart_delete_recovery_mutation(volume, object)?; + if !safe_path(object, false) { + return Err(DiskError::InvalidPath); + } + let obj_path = self.get_object_path(volume, object)?; + ( + obj_path, + None, + self.io_get_object_path(volume, object)?, + self.io_get_bucket_path(RUSTFS_META_TMP_BUCKET)?, + None, + ) + } + }; let metadata_owner = os::acquire_metadata_mutation_lease(&object_path, namespace_owner).await; let volume_guard = os::disk_volume_mutation_lock(&self.root, volume).read_owned().await; let owner: Option> = Some(Arc::new((metadata_owner, volume_guard))); - - let root = self.io_get_object_path(volume, object)?; let rollback = root.join(transaction.to_string()); let receipt_path = rollback.join(DELETE_QUORUM_COMMITTED_FILE); - let staging_root = self.io_get_bucket_path(RUSTFS_META_TMP_BUCKET)?; let sync = effective_durability(volume).syncs_commit_metadata(); let lease = os::acquire_namespace_mutation_lease_with_owner(&receipt_path, owner).await; let volume_str = volume.to_owned(); @@ -2714,23 +2829,39 @@ impl LocalDisk { lease, sync, move || -> std::io::Result>> { - if let Some((journal, sole_transaction)) = take_prepared_delete_receipt(&rollback) + if let Some((journal_raw, journal, sole_transaction)) = cached_receipt && rollback.join(PREPARED).is_file() { - let receipt = CommitReceipt { - journal: journal.clone(), - finished: false, - }; - let receipt_bytes = serde_json::to_vec(&receipt).map_err(std::io::Error::other)?; - std::fs::create_dir_all(&staging_root)?; - let mut staged = tempfile::NamedTempFile::new_in(&staging_root)?; - std::io::Write::write_all(staged.as_file_mut(), &receipt_bytes)?; - if sync && !receipt_bytes.is_empty() { - staged.as_file().sync_data()?; + let tmp_receipt = staging_root.join(format!("receipt-{transaction}.tmp")); + let receipt_bytes = format_commit_receipt_bytes(&journal_raw, false); + let write_res = (|| -> std::io::Result<()> { + let mut file = std::fs::File::create(&tmp_receipt)?; + std::io::Write::write_all(&mut file, &receipt_bytes)?; + if sync && !receipt_bytes.is_empty() { + file.sync_data()?; + } + os::record_native_publication(&receipt_path); + std::fs::rename(&tmp_receipt, &receipt_path)?; + if sync { + os::fsync_dir_std(&rollback)?; + } + Ok(()) + })(); + if let Err(err) = write_res { + let _ = std::fs::remove_file(&tmp_receipt); + return Err(err); } - std::fs::rename(staged.path(), &receipt_path)?; - if sync { - os::fsync_dir_std(&rollback)?; + if let Some(bucket_path) = bucket_path { + put_published_delete_receipt( + rollback, + PublishedDeleteReceiptEntry { + object_path, + bucket_path, + file_path: root, + staging_root, + journal_raw, + }, + ); } return Ok(Ok(Some((journal, sole_transaction, true)))); } @@ -2854,14 +2985,46 @@ impl LocalDisk { Err(err) => Err(err), } } - self.reject_multipart_delete_recovery_mutation(volume, object)?; - if !safe_path(object, false) || transaction.is_nil() { + if transaction.is_nil() { return Err(DiskError::InvalidPath); } - let object_path = self.get_object_path(volume, object)?; + let rollback_open_path = if volume != RUSTFS_META_MULTIPART_BUCKET { + self.io_get_object_open_path(volume, object) + .ok() + .map(|p| p.join(transaction.to_string())) + } else { + None + }; + let cached_published = rollback_open_path.as_deref().and_then(take_published_delete_receipt); + let (object_path, bucket_path, staging_root, root, cached_journal_raw) = match cached_published { + Some(entry) => ( + entry.object_path, + entry.bucket_path, + entry.staging_root, + entry.file_path, + Some(entry.journal_raw), + ), + None => { + self.reject_multipart_delete_recovery_mutation(volume, object)?; + if !safe_path(object, false) { + return Err(DiskError::InvalidPath); + } + let obj_path = self.get_object_path(volume, object)?; + ( + obj_path, + self.io_get_bucket_path(volume)?, + self.io_get_bucket_path(RUSTFS_META_TMP_BUCKET)?, + self.io_get_object_path(volume, object)?, + None, + ) + } + }; let metadata_owner = os::acquire_metadata_mutation_lease(&object_path, namespace_owner).await; let volume_guard = os::disk_volume_mutation_lock(&self.root, volume).read_owned().await; let owner: Option> = Some(Arc::new((metadata_owner, volume_guard))); + let rollback = root.join(transaction.to_string()); + let receipt_path = rollback.join(DELETE_QUORUM_COMMITTED_FILE); + let sync = effective_durability(volume).syncs_commit_metadata(); let mut claims: Vec = Vec::with_capacity(journal.data_dirs.len()); for data_dir in &journal.data_dirs { @@ -2883,15 +3046,9 @@ impl LocalDisk { claims.push(claim); } - let bucket_path = self.io_get_bucket_path(volume)?; - let staging_root = self.io_get_bucket_path(RUSTFS_META_TMP_BUCKET)?; - let root = self.io_get_object_path(volume, object)?; - let rollback = root.join(transaction.to_string()); - let receipt_path = rollback.join(DELETE_QUORUM_COMMITTED_FILE); - let sync = effective_durability(volume).syncs_commit_metadata(); let data_dirs = journal.data_dirs.clone(); let has_target = journal.target_digest.is_some(); - let journal_for_receipt = journal.clone(); + let journal_for_receipt = cached_journal_raw.is_none().then(|| journal.clone()); let lease = os::acquire_namespace_mutation_lease_with_owner(&receipt_path, owner).await; let prune_paths: Vec<_> = rollback .ancestors() @@ -2900,6 +3057,32 @@ impl LocalDisk { .collect(); let res = os::run_blocking_namespace_operation_with_sync(lease, sync, move || -> std::io::Result<()> { + let persist_receipt = |finished: bool| -> std::io::Result<()> { + let receipt_bytes = match &cached_journal_raw { + Some(raw) => format_commit_receipt_bytes(raw, finished), + None => { + let receipt = CommitReceipt { + journal: journal_for_receipt + .clone() + .ok_or_else(|| std::io::Error::other("missing journal for receipt"))?, + finished, + }; + serde_json::to_vec(&receipt).map_err(std::io::Error::other)? + } + }; + std::fs::create_dir_all(&staging_root)?; + let mut staged_receipt = tempfile::NamedTempFile::new_in(&staging_root)?; + std::io::Write::write_all(staged_receipt.as_file_mut(), &receipt_bytes)?; + if sync && !receipt_bytes.is_empty() { + staged_receipt.as_file().sync_data()?; + } + os::record_native_publication(&receipt_path); + std::fs::rename(staged_receipt.path(), &receipt_path)?; + if sync { + let _ = os::fsync_dir_std(&rollback); + } + Ok(()) + }; let committed = format!("{DELETE_DATA_DIR_MARKER_PREFIX}{transaction}"); let recovery_committed = format!("{RECOVERY_DELETE_DATA_DIR_MARKER_PREFIX}{transaction}"); let reserved = format!("{RESERVED_DELETE_DATA_DIR_MARKER_PREFIX}{transaction}"); @@ -2954,39 +3137,38 @@ impl LocalDisk { os::fsync_dir_std(&root)?; } for name in [STORAGE_FORMAT_FILE_BACKUP, DELETE_MARKER_ROLLBACK_FILE] { - ignore_not_found(std::fs::remove_file(rollback.join(name)))?; - } - if has_target || had_physical_data_dirs { - let finished_receipt = CommitReceipt { - journal: journal_for_receipt, - finished: true, - }; - let receipt_bytes = serde_json::to_vec(&finished_receipt).map_err(std::io::Error::other)?; - std::fs::create_dir_all(&staging_root)?; - let mut staged_receipt = tempfile::NamedTempFile::new_in(&staging_root)?; - std::io::Write::write_all(staged_receipt.as_file_mut(), &receipt_bytes)?; - if sync && !receipt_bytes.is_empty() { - staged_receipt.as_file().sync_data()?; - } - os::record_native_publication(&receipt_path); - std::fs::rename(staged_receipt.path(), &receipt_path)?; - let _ = std::fs::File::create(rollback.join(FINISHED))?; - if sync { - os::fsync_dir_std(&rollback)?; + if let Err(err) = ignore_not_found(std::fs::remove_file(rollback.join(name))) { + if has_target || had_physical_data_dirs { + let _ = persist_receipt(true); + } + return Err(err); } } - for name in [TARGET, PREPARED, FINISHED] { + for name in [PREPARED, TARGET, FINISHED] { let path = rollback.join(name); #[cfg(all(any(test, feature = "test-util"), not(windows)))] os::prepared_publication_test_hooks::run(os::prepared_publication_test_hooks::Stage::Remove, &path); - ignore_not_found(std::fs::remove_file(path))?; + if let Err(err) = ignore_not_found(std::fs::remove_file(path)) { + if has_target || had_physical_data_dirs { + let _ = persist_receipt(true); + } + return Err(err); + } } - if sync { - os::fsync_dir_std(&rollback)?; + if sync && let Err(err) = os::fsync_dir_std(&rollback) { + if has_target || had_physical_data_dirs { + let _ = persist_receipt(true); + } + return Err(err); } #[cfg(all(any(test, feature = "test-util"), not(windows)))] os::prepared_publication_test_hooks::run(os::prepared_publication_test_hooks::Stage::Remove, &receipt_path); - ignore_not_found(std::fs::remove_file(&receipt_path))?; + if let Err(err) = ignore_not_found(std::fs::remove_file(&receipt_path)) { + if has_target || had_physical_data_dirs { + let _ = persist_receipt(true); + } + return Err(err); + } for dir_path in &prune_paths { #[cfg(all(any(test, feature = "test-util"), not(windows)))] os::prepared_publication_test_hooks::run(os::prepared_publication_test_hooks::Stage::Remove, dir_path); @@ -3109,6 +3291,42 @@ pub(super) async fn restore_barrier(volume: &str, object_dir: &Path, transaction Ok(()) } +#[cfg(all(test, unix))] +mod fused_prepare_checkpoint { + use std::collections::HashMap; + use std::path::{Path, PathBuf}; + use std::sync::{LazyLock, Mutex}; + + type Hook = Box; + static HOOKS: LazyLock>> = LazyLock::new(|| Mutex::new(HashMap::new())); + + pub(super) struct Guard(PathBuf); + + impl Drop for Guard { + fn drop(&mut self) { + HOOKS.lock().expect("checkpoint hooks").remove(&self.0); + } + } + + pub(super) fn install(root: &Path, hook: impl FnOnce() + Send + 'static) -> Guard { + assert!( + HOOKS + .lock() + .expect("checkpoint hooks") + .insert(root.to_path_buf(), Box::new(hook)) + .is_none() + ); + Guard(root.to_path_buf()) + } + + pub(super) fn run(root: &Path) { + let hook = HOOKS.lock().expect("checkpoint hooks").remove(root); + if let Some(hook) = hook { + hook(); + } + } +} + #[cfg(test)] mod tests { use super::*; @@ -7024,6 +7242,73 @@ mod tests { marker.deleted = false; } + #[cfg(unix)] + #[tokio::test] + async fn delete_recovery_fused_prepare_checkpoint_reopens_before_head_publication() { + let (_temp, disk, mut first, _) = fixture().await; + first.version_id = Some(Uuid::new_v4()); + let mut second = first.clone(); + second.version_id = Some(Uuid::new_v4()); + let mut meta = FileMeta::new(); + meta.add_version(first).expect("add surviving version"); + meta.add_version(second.clone()).expect("add deleted version"); + let original = meta.marshal_msg().expect("encode original versions"); + let (_, root) = disk.io_get_bucket_and_object_path("recovery", "object").expect("object path"); + fs::write(root.join(STORAGE_FORMAT_FILE), &original) + .await + .expect("write versions"); + + let (checkpoint_temp, checkpoint_disk, _, _) = fixture().await; + let checkpoint_root = checkpoint_temp.path().join("recovery/object"); + drop(checkpoint_disk); + let transaction = Uuid::new_v4(); + let source_root = root.clone(); + let copy_root = checkpoint_root.clone(); + let _capture = fused_prepare_checkpoint::install(&root, move || { + let source = source_root.join(transaction.to_string()); + let destination = copy_root.join(transaction.to_string()); + std::fs::create_dir(&destination).expect("create checkpoint transaction"); + std::fs::copy(source_root.join(STORAGE_FORMAT_FILE), copy_root.join(STORAGE_FORMAT_FILE)) + .expect("capture head before publication"); + for entry in std::fs::read_dir(source).expect("read published preparation") { + let entry = entry.expect("prepared file"); + assert!(entry.file_type().expect("prepared file type").is_file()); + std::fs::copy(entry.path(), destination.join(entry.file_name())).expect("capture prepared file"); + } + }); + let opts = DeleteOptions { + recoverable_delete: true, + ..Default::default() + }; + assert_eq!( + disk.try_fused_recoverable_delete_version("recovery", "object", &second, &opts, transaction, None) + .await + .expect("finish original preparation"), + Some(()) + ); + assert!(checkpoint_root.join(transaction.to_string()).join(PREPARED).is_file()); + let endpoint = + Endpoint::try_from(checkpoint_temp.path().to_str().expect("checkpoint path")).expect("checkpoint endpoint"); + let reopened = LocalDisk::new(&endpoint, false) + .await + .expect("reopen checkpoint as a fresh disk"); + let inspection = reopened + .delete_recovery_inspect("recovery", "object", transaction) + .await + .expect("actual preparation checkpoint must remain recoverable"); + assert_eq!(inspection.state, DeleteRecoveryState::NewPrepared); + reopened + .settle_delete_transaction("recovery", "object", transaction, inspection.fingerprint, DeleteRecoveryAction::Rollback) + .await + .expect("roll back interrupted preparation"); + assert_eq!( + fs::read(checkpoint_root.join(STORAGE_FORMAT_FILE)) + .await + .expect("read rolled-back versions"), + original + ); + } + #[tokio::test] async fn delete_recovery_fused_versioned_delete_with_surviving_version_and_interrupted_target_cleanup() { let (_temp, disk, mut v1, _) = fixture().await; diff --git a/crates/ecstore/src/disk/local/multipart_index.rs b/crates/ecstore/src/disk/local/multipart_index.rs index 3fa75d28e..b695185d7 100644 --- a/crates/ecstore/src/disk/local/multipart_index.rs +++ b/crates/ecstore/src/disk/local/multipart_index.rs @@ -372,29 +372,32 @@ impl IndexOwner { #[cfg(unix)] { use rustix::fs::{Mode, OFlags, mkdirat, openat}; - let mut directories = vec![self._identity_pin.try_clone()?]; - for component in [".rustfs.sys", "multipart-index-v1", &sha(bucket)] { - let parent = directories.last().ok_or_else(|| io::Error::other("index parent missing"))?; - match mkdirat(parent, component, Mode::RWXU | Mode::RWXG | Mode::RWXO) { - Ok(()) | Err(rustix::io::Errno::EXIST) => {} - Err(error) => return Err(error.into()), + let open_dir = |parent: rustix::fd::BorrowedFd<'_>, component: &str| -> io::Result { + let flags = OFlags::RDONLY | OFlags::DIRECTORY | OFlags::NOFOLLOW | OFlags::CLOEXEC; + match openat(parent, component, flags, Mode::empty()) { + Ok(dir) => Ok(dir), + Err(rustix::io::Errno::NOENT) => { + match mkdirat(parent, component, Mode::RWXU | Mode::RWXG | Mode::RWXO) { + Ok(()) | Err(rustix::io::Errno::EXIST) => {} + Err(error) => return Err(error.into()), + } + openat(parent, component, flags, Mode::empty()).map_err(Into::into) + } + Err(error) => Err(error.into()), } - let directory = openat( - parent, - component, - OFlags::RDONLY | OFlags::DIRECTORY | OFlags::NOFOLLOW | OFlags::CLOEXEC, - Mode::empty(), - )?; - directories.push(std::fs::File::from(directory)); - } - let parent = directories.last().ok_or_else(|| io::Error::other("index parent missing"))?; + }; + use std::os::fd::AsFd; + let bucket_sha = sha(bucket); + let sys_dir = open_dir(self._identity_pin.as_fd(), ".rustfs.sys")?; + let index_dir = open_dir(sys_dir.as_fd(), "multipart-index-v1")?; + let parent = open_dir(index_dir.as_fd(), &bucket_sha)?; // The index is boot-local: every start rescans the staging area // before serving it, so a marker lost to a crash is recreated before // it can be consulted. Creation is enough; no fsync of the marker or // its ancestors. An existing marker is touched so pruning sees the // new upload's age, not the previous one's. match openat( - parent, + &parent, source_root, OFlags::CREATE | OFlags::EXCL | OFlags::WRONLY | OFlags::NOFOLLOW | OFlags::CLOEXEC, Mode::RUSR | Mode::WUSR, @@ -409,7 +412,7 @@ impl IndexOwner { // the previous upload's age, which only shortens the prune // grace for this key; index completeness does not depend on it. if let Ok(marker) = - openat(parent, source_root, OFlags::WRONLY | OFlags::NOFOLLOW | OFlags::CLOEXEC, Mode::empty()) + openat(&parent, source_root, OFlags::WRONLY | OFlags::NOFOLLOW | OFlags::CLOEXEC, Mode::empty()) { let _ = std::fs::File::from(marker).set_modified(std::time::SystemTime::now()); } @@ -430,6 +433,18 @@ impl IndexOwner { } fn ensure_current_root(&self) -> io::Result<()> { + #[cfg(unix)] + let current = { + use std::os::unix::fs::MetadataExt; + let metadata = std::fs::symlink_metadata(&*self.root_path)?; + if !metadata.is_dir() { + return Err(io::Error::other("multipart index physical root is not a directory")); + } + let mut inode = [0; 16]; + inode[..8].copy_from_slice(&metadata.ino().to_le_bytes()); + (metadata.dev(), inode) + }; + #[cfg(not(unix))] let current = os::PublicationRoot::new(&self.root_path)?.physical_identity()?; if current != self.physical { self.disabled.store(true, Ordering::Release); diff --git a/crates/ecstore/src/disk/local/multipart_source_state.rs b/crates/ecstore/src/disk/local/multipart_source_state.rs index 336d52c3a..94ef07628 100644 --- a/crates/ecstore/src/disk/local/multipart_source_state.rs +++ b/crates/ecstore/src/disk/local/multipart_source_state.rs @@ -1979,25 +1979,28 @@ impl NativeMultipartPublication { let canonical = self.canonical_head(); let prepared = (|| -> std::io::Result { if let Some((ref original_bytes, _, _)) = deferred_initial { - let current = load_source(source, base, root, path, disk_id, physical_slot, disk_ids).map_err(IoError::other)?; - let snapshot = current - .snapshot - .as_ref() - .ok_or_else(|| IoError::other(DiskError::ConditionalDeleteConflict))?; - let current_head = current - .head - .as_ref() - .and_then(SourceHead::raw) - .ok_or_else(|| IoError::other(DiskError::ConditionalDeleteConflict))?; - if !current.head_matches_state - || snapshot.phase != MultipartSourcePhase::Active - || snapshot.capability_sha256.is_some() - || snapshot.lineage != expected_lineage - || snapshot.generation != expected_generation - || snapshot.revision.checked_add(1) != Some(expected_revision) - || current_head != original_bytes.as_slice() - { - return Err(IoError::other(DiskError::ConditionalDeleteConflict)); + if cfg!(test) || !can_mark_presynced { + let current = + load_source(source, base, root, path, disk_id, physical_slot, disk_ids).map_err(IoError::other)?; + let snapshot = current + .snapshot + .as_ref() + .ok_or_else(|| IoError::other(DiskError::ConditionalDeleteConflict))?; + let current_head = current + .head + .as_ref() + .and_then(SourceHead::raw) + .ok_or_else(|| IoError::other(DiskError::ConditionalDeleteConflict))?; + if !current.head_matches_state + || snapshot.phase != MultipartSourcePhase::Active + || snapshot.capability_sha256.is_some() + || snapshot.lineage != expected_lineage + || snapshot.generation != expected_generation + || snapshot.revision.checked_add(1) != Some(expected_revision) + || current_head != original_bytes.as_slice() + { + return Err(IoError::other(DiskError::ConditionalDeleteConflict)); + } } } else { let current = load_source(source, base, root, path, disk_id, physical_slot, disk_ids).map_err(IoError::other)?; @@ -2050,7 +2053,7 @@ impl NativeMultipartPublication { let _ = std::fs::remove_file(prepared_path); return Err(err); } - os::install_multipart_prepared_head(prepared_metadata_source, &canonical, base, root) + os::install_multipart_prepared_head(prepared_metadata_source, &canonical, base, root, Some(directory_guard)) })() .map_err(DiskError::from)?; *self.state.lock() = state; @@ -2160,7 +2163,7 @@ impl NativeMultipartPublication { let mut prepared_metadata_source = os::create_prepared_rename_source_with_commit_guard(&prepared_path, &dst_file_path, &owned_directory_guard)?; prepared_metadata_source.write_all(&new_bytes, true)?; - os::install_multipart_prepared_head(prepared_metadata_source, &canonical, &base, &root) + os::install_multipart_prepared_head(prepared_metadata_source, &canonical, &base, &root, Some(&owned_directory_guard)) }) .await .map_err(DiskError::from)?; @@ -2207,7 +2210,7 @@ impl NativeMultipartPublication { let base = self.store.base.clone(); let root = self.store.publication_root.clone(); os::run_blocking_namespace_operation(self.store.lease.clone(), move || { - os::install_multipart_prepared_head(prepared, &canonical, &base, &root) + os::install_multipart_prepared_head(prepared, &canonical, &base, &root, None) }) .await .map_err(DiskError::from) @@ -2264,6 +2267,7 @@ impl NativeMultipartPublication { pub(super) fn finish_on_thread( &self, target_already_synced: bool, + fused_in_place: bool, staging_cleanup: Option<(PathBuf, Arc)>, ) -> Result { let mut state = self.state.lock().clone(); @@ -2293,62 +2297,64 @@ impl NativeMultipartPublication { let disk_ids = &self.store.disk_ids; let terminal = terminal_record_path(source, &state); (|| -> std::io::Result<()> { - let current = load_source(source, base, root, path, disk_id, physical_slot, disk_ids).map_err(IoError::other)?; - let current_state = current - .state - .as_ref() - .ok_or_else(|| IoError::other(DiskError::ConditionalDeleteConflict))?; - let current_raw = encode(current_state).map_err(IoError::other)?; - if current.head.is_some() || current_raw != expected_state_raw { - return Err(IoError::other(DiskError::ConditionalDeleteConflict)); - } - if target_already_synced { - let pinned = os::pin_multipart_source_directory(&target_dir, base, root).map_err(|error| { - if error.kind() == std::io::ErrorKind::NotFound { - IoError::other(DiskError::ConditionalDeleteConflict) - } else { - error - } - })?; - let target = os::read_multipart_source_metadata_in_pinned( - &pinned, - &target_dir, - std::ffi::OsStr::new(STORAGE_FORMAT_FILE), - base, - root, - ) - .map_err(|error| { - if error.kind() == std::io::ErrorKind::NotFound { - IoError::other(DiskError::ConditionalDeleteConflict) - } else { - error - } - })?; - if target.len() as u64 != expected_length || raw_sha(&target) != expected_sha256 { + if !fused_in_place { + let current = load_source(source, base, root, path, disk_id, physical_slot, disk_ids).map_err(IoError::other)?; + let current_state = current + .state + .as_ref() + .ok_or_else(|| IoError::other(DiskError::ConditionalDeleteConflict))?; + let current_raw = encode(current_state).map_err(IoError::other)?; + if current.head.is_some() || current_raw != expected_state_raw { return Err(IoError::other(DiskError::ConditionalDeleteConflict)); } - } else { - os::read_and_sync_multipart_target_metadata( - &target_dir, - std::ffi::OsStr::new(STORAGE_FORMAT_FILE), - base, - root, - |target| { - if target.len() as u64 != expected_length || raw_sha(target) != expected_sha256 { - return Err(IoError::other(DiskError::ConditionalDeleteConflict)); + if target_already_synced { + let pinned = os::pin_multipart_source_directory(&target_dir, base, root).map_err(|error| { + if error.kind() == std::io::ErrorKind::NotFound { + IoError::other(DiskError::ConditionalDeleteConflict) + } else { + error } - Ok(()) - }, - ) - .map_err(|error| { - if error.kind() == std::io::ErrorKind::NotFound { - IoError::other(DiskError::ConditionalDeleteConflict) - } else { - error + })?; + let target = os::read_multipart_source_metadata_in_pinned( + &pinned, + &target_dir, + std::ffi::OsStr::new(STORAGE_FORMAT_FILE), + base, + root, + ) + .map_err(|error| { + if error.kind() == std::io::ErrorKind::NotFound { + IoError::other(DiskError::ConditionalDeleteConflict) + } else { + error + } + })?; + if target.len() as u64 != expected_length || raw_sha(&target) != expected_sha256 { + return Err(IoError::other(DiskError::ConditionalDeleteConflict)); } - })?; + } else { + os::read_and_sync_multipart_target_metadata( + &target_dir, + std::ffi::OsStr::new(STORAGE_FORMAT_FILE), + base, + root, + |target| { + if target.len() as u64 != expected_length || raw_sha(target) != expected_sha256 { + return Err(IoError::other(DiskError::ConditionalDeleteConflict)); + } + Ok(()) + }, + ) + .map_err(|error| { + if error.kind() == std::io::ErrorKind::NotFound { + IoError::other(DiskError::ConditionalDeleteConflict) + } else { + error + } + })?; + } } - SourceStore::persist_state_std(source, base, root, &next_state_raw, terminal.as_deref(), true)?; + SourceStore::persist_state_std(source, base, root, &next_state_raw, terminal.as_deref(), !fused_in_place)?; if let Some((io_path, cleaned)) = staging_cleanup { match std::fs::remove_dir(&io_path) { Ok(()) => cleaned.store(true, std::sync::atomic::Ordering::Release), @@ -2381,7 +2387,7 @@ impl NativeMultipartPublication { ) -> Result { let this = Arc::clone(self); os::run_blocking_namespace_operation_with_sync(target_lease, true, move || { - this.finish_on_thread(target_already_synced, staging_cleanup) + this.finish_on_thread(target_already_synced, false, staging_cleanup) .map_err(IoError::other) }) .await diff --git a/crates/ecstore/src/disk/os.rs b/crates/ecstore/src/disk/os.rs index eb9fad79d..6afae305a 100644 --- a/crates/ecstore/src/disk/os.rs +++ b/crates/ecstore/src/disk/os.rs @@ -2428,6 +2428,7 @@ pub(crate) const MAX_FSYNC_WORKERS_PER_DISK: usize = 8; // Return busy workers to the shared pool so other disks can acquire a slot. const MAX_BLOCKING_JOBS_PER_WORKER: usize = 64; const WORKER_IDLE_SPIN_ITERS: usize = 256; +const FSYNC_WORKER_IDLE_SPIN_ITERS: usize = 1024; const FALLBACK_NAMESPACE_QUEUE_SHARDS: usize = 8; #[derive(Clone, Copy, PartialEq, Eq)] @@ -2636,7 +2637,7 @@ impl DiskBlockingQueue { .ready .store(!state.jobs.is_empty(), std::sync::atomic::Ordering::Relaxed); if job.is_none() { - if state.idle_workers > 0 { + if lane != DiskBlockingLane::Fsync && state.idle_workers > 0 { state.active_workers = state.active_workers.saturating_sub(1); guard.state = DiskBlockingWorkerState::Retired; return; @@ -2654,7 +2655,12 @@ impl DiskBlockingQueue { } idle_polled = false; } else { - for _ in 0..WORKER_IDLE_SPIN_ITERS { + let spin_iters = if lane == DiskBlockingLane::Fsync { + FSYNC_WORKER_IDLE_SPIN_ITERS + } else { + WORKER_IDLE_SPIN_ITERS + }; + for _ in 0..spin_iters { if queue.lane(lane).ready.load(std::sync::atomic::Ordering::Relaxed) { break; } @@ -7692,6 +7698,7 @@ pub(crate) fn install_multipart_prepared_head( canonical_head: &Path, base: &Path, publication_root: &PublicationRoot, + directory_guard: Option<&RenameCommitGuard>, ) -> io::Result { let original_path = prepared.path.clone(); let parent = canonical_head @@ -7700,9 +7707,37 @@ pub(crate) fn install_multipart_prepared_head( if original_path.parent() != Some(parent) || original_path == canonical_head { return Err(io::Error::new(io::ErrorKind::InvalidInput, "MPU proxy must be a private sibling")); } + #[cfg(target_os = "linux")] + let guard = if let Some(directory_guard) = directory_guard.filter(|g| g.directories.source_path == parent) { + let dir = &directory_guard.directories.source; + RenameCommitGuard { + directories: Arc::new(LinuxCommitDirectories { + root: directory_guard.directories.root.clone(), + source_path: parent.to_path_buf(), + destination_path: parent.to_path_buf(), + source: LinuxCommitDirectory { + file: Arc::clone(&dir.file), + device: dir.device, + inode: dir.inode, + }, + destination: std::sync::OnceLock::from(LinuxCommitDirectory { + file: Arc::clone(&dir.file), + device: dir.device, + inode: dir.inode, + }), + destination_base: base.to_path_buf(), + }), + } + } else { + prepare_rename_commit_guard(parent, parent, base, publication_root)? + }; #[cfg(not(target_os = "linux"))] let _pinned = pin_multipart_source_directory(parent, base, publication_root)?; - let guard = prepare_rename_commit_guard(parent, parent, base, publication_root)?; + #[cfg(not(target_os = "linux"))] + let guard = { + let _ = directory_guard; + prepare_rename_commit_guard(parent, parent, base, publication_root)? + }; rename_prepared_source_std(&prepared, &original_path, canonical_head, base, publication_root, &guard)?; prepared.path = canonical_head.to_path_buf(); fsync_commit_directory_limited_std(parent, &guard)?; diff --git a/crates/ecstore/src/set_disk/read.rs b/crates/ecstore/src/set_disk/read.rs index f222c015f..8cca91042 100644 --- a/crates/ecstore/src/set_disk/read.rs +++ b/crates/ecstore/src/set_disk/read.rs @@ -2979,10 +2979,11 @@ fn build_legacy_per_part_fallback_reader( part_length: usize, part_size: usize, read_repair: Option<(DecodeReadRepairContext, bool)>, -) -> Box { +) -> Box { let buffer = adaptive_duplex_buffer_size(part_size as i64); - let (read_half, mut write_half) = tokio::io::duplex(buffer); + let (read_half, write_half) = tokio::io::duplex(buffer); let decode = tokio::spawn(async move { + let mut write_half = GetObjectDownstreamWriter::new(write_half); let outcome = erasure .decode_with_stripe_handles_and_reopeners_with_diagnostics( &mut write_half, @@ -8908,6 +8909,42 @@ mod tests { assert!(err.to_string().contains("inconsistent read source shards")); } + #[tokio::test] + async fn legacy_per_part_output_close_retains_downstream_classification() { + let erasure = coding::Erasure::new(2, 2, 64); + let data = (0..64u8).collect::>(); + let setup = + setup_codec_data_blocks_first_encoded_bitrot_readers(&erasure, &data, &[0], &[], &[], HashAlgorithm::HighwayHash256) + .await; + let mut reader = build_legacy_per_part_fallback_reader( + erasure, + setup.readers, + setup.deferred_stripe_handles, + Vec::new(), + None, + 0, + data.len(), + data.len(), + None, + ); + // Closing the actual output pipe models the body consumer disappearing + // while retaining the executor long enough to inspect its typed result. + assert!(!reader.finished); + let (replacement, _unused) = tokio::io::duplex(1); + drop(std::mem::replace(&mut reader.inner, replacement)); + let handle = reader.decode.take().expect("fallback decode task"); + let error = handle + .await + .expect("join the fallback task") + .expect("closed pipe must interrupt output"); + assert_eq!(error.kind(), std::io::ErrorKind::BrokenPipe); + assert_eq!( + crate::diagnostics::get::classify_io_error(&error), + GetObjectFailureReason::DownstreamClosed, + "fallback output must carry the same marker as the normal GET path" + ); + } + // backlog#879: a later multipart part that loses a shard (read quorum still // holds) must be reconstructed in place through the legacy per-part decode // bridge instead of failing the stream. This exercises the exact reader the diff --git a/crates/heal/tests/shard_identity_receipt_test.rs b/crates/heal/tests/shard_identity_receipt_test.rs index eac226e43..a20498826 100644 --- a/crates/heal/tests/shard_identity_receipt_test.rs +++ b/crates/heal/tests/shard_identity_receipt_test.rs @@ -30,6 +30,78 @@ use storage_api::integration::{ WriteCommitGuard, WriteCompletion, }; +async fn complete_fixture_replicas(env: &TestECStoreEnv, bucket: &str, object: &str) { + // TailDrained completes fan-out, but a successful quorum write may still + // have minority failures. Fault injection needs a fully repaired baseline. + let result = ECStoreHealStorage::new(env.ecstore.clone()) + .heal_object_with_receipt( + bucket, + object, + None, + &HealOpts { + scan_mode: HealScanMode::Deep, + ..Default::default() + }, + ) + .await + .expect("complete protected fixture replicas before fault injection"); + assert!(result.error.is_none(), "fixture repair failed: {:?}", result.error); + assert_eq!(result.item.after.drives.len(), env.disk_paths.len()); + assert!( + result.item.after.drives.iter().all(|drive| drive.state == "ok"), + "fixture requires every replica, not only read/write quorum: {:?}", + result.item.after.drives + ); + assert!(result.item.integrity_verified, "fixture must retain protected payload verification"); +} + +#[tokio::test] +#[serial] +async fn mixed_batch_fixture_repairs_a_missing_minority_replica() { + let root = tempfile::tempdir().expect("minority replica fixture"); + let env = TestECStoreEnv::builder().base_dir(root.path()).build().await; + let bucket = "mixed-fixture-baseline"; + let object = "object"; + env.make_bucket(bucket, false).await; + let expected = vec![0x5a; 1024 * 1024 + 123]; + env.ecstore + .put_object( + bucket, + object, + &mut PutObjReader::from_vec(expected.clone()), + &ObjectOptions { + shard_integrity_write_mode: Some(ShardIntegrityWriteMode::Protected), + write_completion: WriteCompletion::TailDrained, + ..Default::default() + }, + ) + .await + .expect("quorum commit fixture"); + // Guarantee a minority failure regardless of whether the original write + // succeeded on every disk, then repair it before any test fault is applied. + let replica = env.disk_paths[0].join(bucket).join(object); + if replica.exists() { + tokio::fs::remove_dir_all(&replica) + .await + .expect("remove one complete replica"); + } + complete_fixture_replicas(&env, bucket, object).await; + let disks = env.ecstore.pools[0].get_disks(0); + for disk in disks.disks.read().await.iter().flatten() { + disk.read_version("", bucket, object, "", &ReadOptions::default()) + .await + .expect("every fixture replica has metadata after completion"); + } + let mut reader = env + .ecstore + .get_object_reader(bucket, object, None, Default::default(), &ObjectOptions::default()) + .await + .expect("read the completed fixture"); + let mut actual = Vec::new(); + reader.stream.read_to_end(&mut actual).await.expect("completed fixture body"); + assert_eq!(actual, expected); +} + #[test] fn disabled_lock_reports_only_quorum_selection_and_deep_check_observations() { const CHILD_ENV: &str = "RUSTFS_TEST_DISABLED_LOCK_HEAL_CHILD"; @@ -704,6 +776,7 @@ fn admin_mixed_batch_retains_repairs_after_object_retry_exhaustion() { ) .await .expect("commit complete fixture"); + complete_fixture_replicas(&env, bucket, object).await; for (slot, disk) in disks.iter().enumerate() { let meta = disk .read_version("", bucket, object, "", &ReadOptions::default()) diff --git a/crates/object-data-cache/src/runtime_memory.rs b/crates/object-data-cache/src/runtime_memory.rs index 14c839258..c5f26f47a 100644 --- a/crates/object-data-cache/src/runtime_memory.rs +++ b/crates/object-data-cache/src/runtime_memory.rs @@ -66,11 +66,26 @@ pub fn select_effective_memory(host_total: u64, host_available: u64, cgroup: Opt /// Resolves the effective memory from an already-refreshed system handle. /// -/// `cgroup_limits()` is computed fresh on each call and is only implemented on -/// Linux (it returns `None` elsewhere), so non-Linux hosts always use host -/// values. +/// On Linux, the current process cgroup and its ancestors are read on each call, +/// including limits imposed by systemd services. System-wide cgroup limits are +/// used if process information is unavailable. Other platforms use host values. pub fn effective_memory_from_system(system: &System) -> EffectiveMemory { - let cgroup = system.cgroup_limits().map(|limits| (limits.total_memory, limits.free_memory)); + #[cfg(target_os = "linux")] + let limits = sysinfo::get_current_pid() + .ok() + .and_then(|pid| { + let mut processes = System::new().ok()?; + processes.refresh_processes_specifics( + sysinfo::ProcessesToUpdate::Some(&[pid]), + false, + sysinfo::ProcessRefreshKind::nothing().with_memory(), + ); + processes.process(pid)?.cgroup_limits() + }) + .or_else(|| system.cgroup_limits()); + #[cfg(not(target_os = "linux"))] + let limits = system.cgroup_limits(); + let cgroup = limits.map(|limits| (limits.total_memory, limits.free_memory)); select_effective_memory(system.total_memory(), system.available_memory(), cgroup) } @@ -89,6 +104,32 @@ mod tests { const GIB: u64 = 1024 * 1024 * 1024; + #[cfg(target_os = "linux")] + #[test] + fn effective_memory_uses_current_process_cgroup() { + use sysinfo::{ProcessRefreshKind, ProcessesToUpdate, System, get_current_pid}; + + let pid = get_current_pid().expect("current process ID"); + let mut system = System::new().expect("system memory handle"); + system.refresh_memory(); + let mut process_system = System::new().expect("process memory handle"); + process_system.refresh_processes_specifics( + ProcessesToUpdate::Some(&[pid]), + false, + ProcessRefreshKind::nothing().with_memory(), + ); + let limits = process_system.process(pid).expect("current process entry").cgroup_limits(); + let expected = select_effective_memory( + system.total_memory(), + system.available_memory(), + limits.map(|limits| (limits.total_memory, limits.free_memory)), + ); + let actual = super::effective_memory_from_system(&system); + assert_eq!(actual.basis, expected.basis); + assert_eq!(actual.total_bytes, expected.total_bytes); + assert!(actual.available_bytes <= actual.total_bytes); + } + #[test] fn select_effective_memory_prefers_constraining_cgroup() { let effective = select_effective_memory(64 * GIB, 40 * GIB, Some((2 * GIB, GIB))); diff --git a/crates/s3select-api/Cargo.toml b/crates/s3select-api/Cargo.toml index bc8dd6489..0875bc997 100644 --- a/crates/s3select-api/Cargo.toml +++ b/crates/s3select-api/Cargo.toml @@ -75,6 +75,8 @@ rustfs-storage-api.workspace = true futures = { workspace = true } futures-core = { workspace = true } flate2.workspace = true +brotli.workspace = true +lz4.workspace = true http.workspace = true memchr.workspace = true serde_json = { workspace = true, features = ["raw_value"] } diff --git a/crates/s3select-api/src/lib.rs b/crates/s3select-api/src/lib.rs index 75d5d969e..2a49197f7 100644 --- a/crates/s3select-api/src/lib.rs +++ b/crates/s3select-api/src/lib.rs @@ -27,6 +27,7 @@ mod csv_input; mod input_stream; mod metrics; pub mod object_store; +pub mod parquet_budget; pub mod query; pub mod request; pub mod server; @@ -219,6 +220,14 @@ fn classify_select_error_source(err: &(dyn StdError + 'static)) -> Option