Compare commits

...

15 Commits

Author SHA1 Message Date
overtrue cfffb1700e chore(gateway): update test-count baseline after main sync
Main #9241 removed the node pool-metadata readiness cache with 8 tests
and added 6, so the rustfs row drops from 6440 to 6438. The ecstore and
s3select rows rise for tests added by #9237, #9239 and #9241.
2026-11-09 10:21:53 +08:00
rustfs-cla-assistant[bot] cd62f3ba4f Merge remote-tracking branch 'origin/main' into gateway/integration 2026-11-09 02:07:05 +00:00
Ramakrishna Chilaka 395876ccd0 fix(ecstore): remove hardlinked backup during delete rollback (#9239)
When a fused recoverable delete is interrupted after publishing the transaction directory and before updating xl.meta, xl.meta and <transaction>/xl.meta.bkp remain hardlinks to the same inode. Under POSIX rename semantics, renaming xl.meta.bkp over xl.meta during rollback succeeds as a no-op without unlinking xl.meta.bkp, which leaves xl.meta.bkp in the transaction directory after journal removal and strands the transaction in DeleteRecoveryState::LegacyAmbiguous.

Unlink xl.meta.bkp after syncing the restored object directory and before syncing the rollback directory and removing journal markers, and extend delete_recovery_fused_prepare_checkpoint_reopens_before_head_publication to preserve the hardlink and verify rollback completion and retry.
2026-11-09 06:41:04 +08:00
Chris An 6083fbbeea fix(catalog): enumerate table owners without skipping namespaces (#9242) 2026-11-09 06:38:58 +08:00
Chris An 42d3267d88 fix(ecstore): decouple pool metadata health from the authority gate (#9241)
* fix(ecstore): share immutable pool metadata authority reads

* fix(ecstore): read pool metadata authority without the save gate

Health inferred pool metadata write authority from acquiring the local
authority gate within 100 ms. Sharing the gate among read-only
admissions still left held writers, and writers queued behind a reader
parked on the pool.bin fence, turning into cluster-write 503s.

Each writer now publishes the authority it leaves before releasing the
gate, and health reads that published state without acquiring it.
Transaction latches stay shared with the published copy; recovery
clears the latch and its failure under one lock so inspection never
observes a half-cleared state. The inspection-timeout path that can no
longer occur (check_timed_out, the check_timeout admin state, the
pool_metadata_check_timeout reason, the node timeout cache and its
metric) is removed.
2026-11-09 06:38:30 +08:00
Chris An 1d35e381b7 fix(rpc): bound replay memory and atomically admit signed requests (#9238)
* fix(rpc): index nonce expiry by signed validity

* fix(rpc): size replay cache from allocated storage layout

* fix(rpc): make replay admission atomic and release expired tables

* fix(rpc): preserve replay admission refusal at the S3 boundary
2026-11-09 06:36:55 +08:00
Chris An 0b109b4885 fix(select): bound LZ4 scratch and Arrow schema expansion (#9237)
* fix(select): admit legacy and linked LZ4 working memory

* fix(select): bound embedded Arrow schema expansion
2026-11-09 06:36:44 +08:00
overtrue 7bc419e368 chore(gateway): merge main into integration
# Conflicts:
#	.config/e2e-full-selection.txt
#	crates/ecstore/src/disk/local/delete_recovery.rs
2026-11-09 00:50:02 +08:00
cxymds 1dea00ffa7 ci: repair e2e selection and verify tool downloads via API (#9240)
* ci: align full e2e selection with retired catalog test

* ci: verify pinned tools through GitHub release asset API

* ci: fetch pinned typos through the release asset API

* ci: download pinned actionlint through the release asset API

* ci: ensure GitHub CLI exists before pinned mc download

* ci: budget full quick checks after tool installation
2026-11-08 14:09:26 +00:00
cxymds addf7a6219 fix: reduce scanner backoff and classify downstream closures (#9235) 2026-11-08 02:18:17 -05:00
Ramakrishna Chilaka 420cf32b2a perf(ecstore): optimize PUT, DELETE, and multipart commit hotpaths (#9227) 2026-11-08 02:12:40 -05:00
Chris An 907410b2f5 fix(select): bound Parquet allocations before decoding (#9231) 2026-11-08 01:36:12 -05:00
Chris An f10079f1a3 fix(rpc): size and reserve replay cache for compact entries (#9233) 2026-11-08 01:35:58 -05:00
Chris An a6c3016493 fix(cache): respect process cgroup memory limits (#9232) 2026-11-08 01:35:49 -05:00
Hauser Zhi 68fb52b752 chore(deps): update flake.lock (#9236) 2026-11-08 01:34:01 -05:00
75 changed files with 5733 additions and 1650 deletions
+1
View File
@@ -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
+7 -7
View File
@@ -10,7 +10,7 @@
# intentionally removed. Regenerate with:
# scripts/check_test_count_ratchet.sh --update
e2e_test 1026 20
rustfs 6439 64
rustfs 6438 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 7850 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 232 0
rustfs-s3select-query 235 2
rustfs-scanner 1386 20
rustfs-scanner-metrics 90 0
rustfs-security-governance 20 0
rustfs-signer 34 0
+4 -3
View File
@@ -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"
+23 -5
View File
@@ -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"
@@ -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())
+16 -3
View File
@@ -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"
Generated
+6
View File
@@ -9847,6 +9847,7 @@ dependencies = [
"google-cloud-auth",
"google-cloud-storage",
"h2",
"hashbrown 0.17.1",
"hex-simd",
"hmac 0.13.0",
"hostname",
@@ -10826,15 +10827,20 @@ dependencies = [
"arc-swap",
"async-compression",
"async-trait",
"base64-simd",
"brotli 9.1.1",
"bytes",
"chrono",
"crc-fast",
"datafusion",
"flatbuffers",
"flate2",
"futures",
"futures-core",
"hotpath",
"http 1.5.0",
"lz4",
"lz4_flex",
"memchr",
"metrics",
"parking_lot 0.13.0",
+1
View File
@@ -303,6 +303,7 @@ local-ip-address = "0.6.13"
log = "0.4"
memmap2 = "0.9.12"
lz4 = "1.28.1"
lz4_flex = { version = "=0.14.0", default-features = false, features = ["std", "frame"] }
matchit = "0.9.2"
md-5 = "0.11.0"
mime_guess = "2.0.5"
+21 -9
View File
@@ -184,15 +184,27 @@ const _: () = assert!(!DEFAULT_INTERNODE_RPC_REPLAY_SCOPE_STRICT);
/// Capacity (distinct nonces) of the process-local internode RPC replay cache that enforces
/// one-time consumption of authenticated RPC signatures.
///
/// The cache retains each nonce for the ~10-minute signature freshness envelope. Once peers use
/// replay-scoped v3 authentication, every authenticated RPC consumes one entry, so the steady
/// state holds roughly `authenticated RPC RPS x 601s` entries. This default is the minimum floor:
/// explicit operator values and resource-aware auto sizing both clamp upward to at least this
/// value. Overflow fails closed — legitimate signed traffic is the only thing that can fill the
/// cache (replays are rejected before insertion, and an attacker cannot mint valid nonces without
/// the shared secret) — and increments
/// `rustfs_system_network_internode_replay_cache_overflow_total`, so a sustained non-zero overflow
/// counter means this capacity is undersized for the node's peak authenticated RPC rate.
/// Nonces remain protected until their signed timestamp is outside the observed wall-clock
/// validity window, including after clock rollback. Current-time signatures live about five
/// minutes; the maximum accepted forward clock skew can extend that to ten minutes. A body-bound
/// v3 request consumes both its v3 and compatibility-v2 nonces, so size for receiver nonce intake,
/// not S3 request rate or CPU count.
///
/// Automatic admission budgets 13% of effective process memory for signed-time
/// hash tables, allocator overhead and simultaneous old/new tables during growth.
/// The reported count is an absolute UUID payload ceiling, not guaranteed throughput:
/// actual byte admission can reject earlier depending on timestamp distribution.
/// Tables grow with demand and are released together when their signed time expires.
/// CPU count, a historical minimum, and a fixed 32M ceiling do not constrain this budget.
/// If memory detection fails, the historical default remains a bounded count fallback.
///
/// Explicit values remain count ceilings with the historical minimum of 1,048,576
/// for configuration compatibility, and bypass automatic byte admission. That
/// compatibility floor does not apply to the automatically detected memory budget.
/// Zero never disables replay protection.
/// Overflow rejects the request without evicting a still-valid identity, increments
/// `rustfs_system_network_internode_replay_cache_overflow_total`, and returns resource
/// exhaustion rather than invalid credentials. Measure headroom before overriding.
pub const ENV_INTERNODE_RPC_REPLAY_CACHE_CAPACITY: &str = "RUSTFS_INTERNODE_RPC_REPLAY_CACHE_CAPACITY";
pub const DEFAULT_INTERNODE_RPC_REPLAY_CACHE_CAPACITY: usize = 1_048_576;
+1
View File
@@ -121,6 +121,7 @@ test-util = []
e2e-test-hooks = []
[dependencies]
hashbrown.workspace = true
hotpath.workspace = true
rustfs-filemeta.workspace = true
rustfs-utils = { workspace = true, features = ["full"] }
+11 -11
View File
@@ -642,17 +642,17 @@ pub mod rpc {
pub use crate::cluster::rpc::{
AuthenticatedChannel, KMS_SIGNAL_SUBSYSTEM, LocalPeerS3Client, MAX_NETWORK_PROBE_BYTES, MAX_NETWORK_PROBE_DURATION,
NetworkPeerProbeClient, NetworkPeerProbeError, NetworkPeerProbeMeasurement, NetworkPeerTarget, PEER_RESTDRY_RUN,
PEER_RESTSIGNAL, PEER_RESTSUB_SYS, PeerRestClient, PeerS3Client, S3PeerSys, SERVICE_SIGNAL_REFRESH_CONFIG,
SERVICE_SIGNAL_RELOAD_DYNAMIC, ScannerBucketListing, ScannerDirtyUsageAcknowledgement, ScannerPeerActivity,
ScannerPeerDirtyUsageBucket, ScannerPeerDirtyUsageSnapshot, ScannerPublicationLease, ScannerScopedDirtyUsageAckEntry,
TONIC_RPC_PREFIX, TonicInterceptor, build_put_file_auth_trailer, check_and_record_signed_rpc_nonce,
decode_heal_bucket_rpc_options, encode_heal_bucket_rpc_options, gen_signature_headers, gen_tonic_replay_scope_headers,
gen_tonic_signature_headers, gen_tonic_signature_interceptor, node_service_time_out_client,
node_service_time_out_client_no_auth, normalize_tonic_rpc_audience, rpc_background_admission,
set_tonic_canonical_body_digest, sign_ns_scanner_capability, sign_ns_scanner_capability_with_checkpoint_coverage,
sign_ns_scanner_capability_with_tier_registry_generation, sign_put_file_capability, sign_tonic_rpc_response_proof,
tonic_boot_epoch_challenge, tonic_boot_epoch_response_headers, tonic_rpc_auth_failure_reason,
verify_ns_scanner_capability, verify_ns_scanner_capability_with_checkpoint_coverage,
PEER_RESTSIGNAL, PEER_RESTSUB_SYS, PeerRestClient, PeerS3Client, RpcReplayCacheAdmissionError, S3PeerSys,
SERVICE_SIGNAL_REFRESH_CONFIG, SERVICE_SIGNAL_RELOAD_DYNAMIC, ScannerBucketListing, ScannerDirtyUsageAcknowledgement,
ScannerPeerActivity, ScannerPeerDirtyUsageBucket, ScannerPeerDirtyUsageSnapshot, ScannerPublicationLease,
ScannerScopedDirtyUsageAckEntry, TONIC_RPC_PREFIX, TonicInterceptor, build_put_file_auth_trailer,
check_and_record_signed_rpc_nonce, decode_heal_bucket_rpc_options, encode_heal_bucket_rpc_options, gen_signature_headers,
gen_tonic_replay_scope_headers, gen_tonic_signature_headers, gen_tonic_signature_interceptor,
node_service_time_out_client, node_service_time_out_client_no_auth, normalize_tonic_rpc_audience,
rpc_auth_is_resource_exhausted, rpc_background_admission, set_tonic_canonical_body_digest, sign_ns_scanner_capability,
sign_ns_scanner_capability_with_checkpoint_coverage, sign_ns_scanner_capability_with_tier_registry_generation,
sign_put_file_capability, sign_tonic_rpc_response_proof, tonic_boot_epoch_challenge, tonic_boot_epoch_response_headers,
tonic_rpc_auth_failure_reason, verify_ns_scanner_capability, verify_ns_scanner_capability_with_checkpoint_coverage,
verify_ns_scanner_capability_with_tier_registry_generation, verify_put_file_auth_trailer, verify_put_file_capability,
verify_rpc_signature, verify_tonic_boot_epoch_response, verify_tonic_canonical_body_digest,
verify_tonic_mutation_body_digest, verify_tonic_mutation_body_digest_reject_unsigned, verify_tonic_rpc_response_proof,
@@ -229,6 +229,7 @@ pub struct ListPathRawOptions {
pub hidden_entry_budget: usize,
pub pending_prefix: Option<String>,
pub filter_prefix: Option<String>,
pub identifier_tree_leaf: Option<String>,
pub forward_to: Option<String>,
pub min_disks: usize,
/// Deliver every replica to the partial callback, including matching headers.
@@ -268,6 +269,7 @@ impl Clone for ListPathRawOptions {
hidden_entry_budget: self.hidden_entry_budget,
pending_prefix: self.pending_prefix.clone(),
filter_prefix: self.filter_prefix.clone(),
identifier_tree_leaf: self.identifier_tree_leaf.clone(),
forward_to: self.forward_to.clone(),
min_disks: self.min_disks,
preserve_replica_metadata: self.preserve_replica_metadata,
@@ -300,6 +302,7 @@ fn walk_dir_options(opts: &ListPathRawOptions) -> WalkDirOptions {
pending_prefix: opts.pending_prefix.clone(),
report_notfound: opts.report_not_found,
filter_prefix: opts.filter_prefix.clone(),
identifier_tree_leaf: opts.identifier_tree_leaf.clone(),
forward_to: opts.forward_to.clone(),
limit: opts.per_disk_limit,
skip_total_timeout: opts.skip_walkdir_total_timeout,
@@ -1233,6 +1236,21 @@ mod tests {
assert!(!options.skip_total_timeout);
}
#[test]
fn identifier_tree_selection_survives_job_clone_and_disk_request() {
let request = ListPathRawOptions {
path: "catalog/namespaces/".to_string(),
identifier_tree_leaf: Some("table-entry.json".to_string()),
recursive: true,
..Default::default()
};
let cloned = request.clone();
let walk = walk_dir_options(&cloned);
assert_eq!(walk.identifier_tree_leaf, request.identifier_tree_leaf);
assert_eq!(walk.base_dir, request.path);
assert!(walk.recursive);
}
#[test]
fn cloned_reader_options_preserve_hidden_scan_budget_and_pending_prefix() {
let request = ListPathRawOptions {
File diff suppressed because it is too large Load Diff
+11 -11
View File
@@ -35,17 +35,17 @@ pub use client::{
// Re-exported through `api::rpc`; not every item is consumed inside this crate.
#[allow(unused_imports)]
pub use http_auth::{
RPC_BACKGROUND_ADMISSION_HEADER, TONIC_RPC_PREFIX, build_auth_headers, build_put_file_auth_trailer,
check_and_record_signed_rpc_nonce, gen_signature_headers, gen_tonic_replay_scope_headers, gen_tonic_signature_headers,
normalize_tonic_rpc_audience, rpc_background_admission, set_tonic_canonical_body_digest, set_tonic_mutation_body_digest,
set_tonic_rolling_canonical_body_digest, set_tonic_rolling_mutation_body_digest, sign_ns_scanner_capability,
sign_ns_scanner_capability_with_checkpoint_coverage, sign_ns_scanner_capability_with_tier_registry_generation,
sign_put_file_capability, sign_tonic_rpc_response_proof, tonic_boot_epoch_challenge, tonic_boot_epoch_response_headers,
tonic_rpc_auth_failure_reason, verify_ns_scanner_capability, verify_ns_scanner_capability_with_checkpoint_coverage,
verify_ns_scanner_capability_with_tier_registry_generation, verify_put_file_auth_trailer, verify_put_file_capability,
verify_rpc_signature, verify_tonic_boot_epoch_response, verify_tonic_canonical_body_digest,
verify_tonic_mutation_body_digest, verify_tonic_mutation_body_digest_reject_unsigned, verify_tonic_rpc_response_proof,
verify_tonic_rpc_signature, verify_tonic_rpc_signature_with_bootstrap,
RPC_BACKGROUND_ADMISSION_HEADER, RpcReplayCacheAdmissionError, TONIC_RPC_PREFIX, build_auth_headers,
build_put_file_auth_trailer, check_and_record_signed_rpc_nonce, gen_signature_headers, gen_tonic_replay_scope_headers,
gen_tonic_signature_headers, normalize_tonic_rpc_audience, rpc_auth_is_resource_exhausted, rpc_background_admission,
set_tonic_canonical_body_digest, set_tonic_mutation_body_digest, set_tonic_rolling_canonical_body_digest,
set_tonic_rolling_mutation_body_digest, sign_ns_scanner_capability, sign_ns_scanner_capability_with_checkpoint_coverage,
sign_ns_scanner_capability_with_tier_registry_generation, sign_put_file_capability, sign_tonic_rpc_response_proof,
tonic_boot_epoch_challenge, tonic_boot_epoch_response_headers, tonic_rpc_auth_failure_reason, verify_ns_scanner_capability,
verify_ns_scanner_capability_with_checkpoint_coverage, verify_ns_scanner_capability_with_tier_registry_generation,
verify_put_file_auth_trailer, verify_put_file_capability, verify_rpc_signature, verify_tonic_boot_epoch_response,
verify_tonic_canonical_body_digest, verify_tonic_mutation_body_digest, verify_tonic_mutation_body_digest_reject_unsigned,
verify_tonic_rpc_response_proof, verify_tonic_rpc_signature, verify_tonic_rpc_signature_with_bootstrap,
};
#[cfg(test)]
pub(crate) use internode_data_transport::TcpHttpInternodeDataTransport;
@@ -18,7 +18,7 @@ use crate::cluster::rpc::client::{
node_service_time_out_client,
};
use crate::cluster::rpc::set_tonic_mutation_body_digest;
use crate::core::pools::{PoolMeta, PoolMetaWriteState};
use crate::core::pools::{PoolMeta, PoolMetaSaveGate, PoolMetaSaveGateWriteGuard};
use crate::disk::error::DiskError;
use crate::disk::error::{Error, Result};
use crate::disk::error_reduce::{BUCKET_OP_IGNORED_ERRS, is_all_buckets_not_found, reduce_write_quorum_errs};
@@ -798,12 +798,12 @@ fn try_acquire_bucket_heal_movement_guard<'a>(
}
async fn acquire_bucket_heal_write_guard<'a>(
gate: Option<&'a tokio::sync::Mutex<PoolMetaWriteState>>,
) -> Result<Option<tokio::sync::MutexGuard<'a, PoolMetaWriteState>>> {
gate: Option<&'a PoolMetaSaveGate>,
) -> Result<Option<PoolMetaSaveGateWriteGuard<'a>>> {
let Some(gate) = gate else {
return Ok(None);
};
let guard = gate.lock().await;
let guard = gate.write().await;
guard
.ensure_write_safe("bucket heal cannot run while pool metadata requires recovery")
.map_err(|e| e.narrow_to_disk().unwrap_or_else(Error::other))?;
@@ -2388,8 +2388,8 @@ mod tests {
#[tokio::test]
async fn local_bucket_heal_refuses_receiver_with_unsafe_pool_metadata() {
let gate = tokio::sync::Mutex::new(PoolMetaWriteState::default());
gate.lock().await.observe_replicas(PoolMetaReplicaState {
let gate = PoolMetaSaveGate::default();
gate.write().await.observe_replicas(PoolMetaReplicaState {
needs_repair: true,
repair_write_safe: false,
});
+54 -1
View File
@@ -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<R, W>(
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]);
@@ -628,7 +628,7 @@ async fn pool_meta_native_physical_runtime_fork_after_apply_latches_original_gat
let fork = identity_epoch_fork(&fixture).await;
let since = OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("fixed original failure timestamp should be valid");
{
let mut state = fixture.store.pool_meta_save_gate.lock().await;
let mut state = fixture.store.pool_meta_save_gate.write().await;
assert_eq!(state.active_transactions.load(Ordering::SeqCst), 0);
state.write_blocked = false;
state.block_context = None;
@@ -685,7 +685,7 @@ async fn pool_meta_native_physical_runtime_fork_after_apply_latches_original_gat
);
assert_committed(&fixture).await;
{
let state = fixture.store.pool_meta_save_gate.lock().await;
let state = fixture.store.pool_meta_save_gate.write().await;
assert!(
state.write_blocked,
"candidate validation must propagate its permanent latch to the original state"
@@ -706,7 +706,7 @@ async fn pool_meta_native_physical_runtime_fork_after_apply_latches_original_gat
.expect("permanently blocked recovery should return without mutation"),
"ordinary-readable repaired heads must not clear the original known-fork latch"
);
let state = fixture.store.pool_meta_save_gate.lock().await;
let state = fixture.store.pool_meta_save_gate.write().await;
assert!(state.write_blocked);
assert_eq!(
state
@@ -979,7 +979,7 @@ async fn pool_meta_native_heal_v1_with_all_upgrade_gates_disabled_keeps_exact_ra
async {
let (fixture, raw) = legacy_native_owner_fixture(POOL_META_V1_VERSION, false).await;
let guard = owner(&fixture.store).await;
let mut state = fixture.store.pool_meta_save_gate.lock().await;
let mut state = fixture.store.pool_meta_save_gate.write().await;
let result = tokio::time::timeout(
std::time::Duration::from_secs(30),
fixture.store.heal_pool_meta_replica_under_fence(0, guard, &mut state),
@@ -1036,7 +1036,7 @@ async fn pool_meta_native_heal_rejects_legacy_source_with_trailing_bytes_without
let before = snapshots(&fixture.store).await;
let identities = owner_repair_identity_heads(&fixture.store).await;
let guard = owner(&fixture.store).await;
let mut state = fixture.store.pool_meta_save_gate.lock().await;
let mut state = fixture.store.pool_meta_save_gate.write().await;
let result = tokio::time::timeout(
std::time::Duration::from_secs(30),
fixture.store.heal_pool_meta_replica_under_fence(1, guard, &mut state),
@@ -1110,7 +1110,7 @@ async fn pool_meta_native_heal_mixed_v1_v2_preserves_observed_format_floor() {
assert_eq!(selection.revision.version, POOL_META_VERSION, "the observed V2 format floor must be retained");
assert_eq!(selection.canonical.as_deref(), Some(semantic_v2.as_slice()));
let guard = owner(&fixture.store).await;
let mut state = fixture.store.pool_meta_save_gate.lock().await;
let mut state = fixture.store.pool_meta_save_gate.write().await;
tokio::time::timeout(std::time::Duration::from_secs(30),
fixture.store.heal_pool_meta_replica_under_fence(target_pool, guard, &mut state))
.await.expect("mixed legacy heal must terminate")
@@ -1199,7 +1199,7 @@ async fn pool_meta_native_heal_rejects_v1_and_v3_prepare_in_either_pool_order()
let before = snapshots(&fixture.store).await;
let identities = owner_repair_identity_heads(&fixture.store).await;
let guard = owner(&fixture.store).await;
let mut state = fixture.store.pool_meta_save_gate.lock().await;
let mut state = fixture.store.pool_meta_save_gate.write().await;
let result = tokio::time::timeout(
std::time::Duration::from_secs(30),
fixture
@@ -1301,7 +1301,7 @@ async fn pool_meta_native_heal_rejects_v1_legacy_dont_save_true_without_writes()
let before = snapshots(&fixture.store).await;
let identities = owner_repair_identity_heads(&fixture.store).await;
let guard = owner(&fixture.store).await;
let mut state = fixture.store.pool_meta_save_gate.lock().await;
let mut state = fixture.store.pool_meta_save_gate.write().await;
let result = tokio::time::timeout(
std::time::Duration::from_secs(30),
fixture.store.heal_pool_meta_replica_under_fence(1, guard, &mut state),
@@ -1410,7 +1410,7 @@ async fn pool_meta_native_heal_rejects_target_v1_dont_save_true_without_writes()
let before = snapshots(&fixture.store).await;
let identities = owner_repair_identity_heads(&fixture.store).await;
let guard = owner(&fixture.store).await;
let mut state = fixture.store.pool_meta_save_gate.lock().await;
let mut state = fixture.store.pool_meta_save_gate.write().await;
let result = tokio::time::timeout(
std::time::Duration::from_secs(30),
fixture.store.heal_pool_meta_replica_under_fence(1, guard, &mut state),
File diff suppressed because it is too large Load Diff
+2 -2
View File
@@ -527,8 +527,8 @@ mod decommission_lock_order_tests {
rebalance_meta: tokio::sync::RwLock::new(other_store.rebalance_meta.read().await.clone()),
decommission_cancelers: tokio::sync::RwLock::new(vec![None; other_store.pools.len()]),
start_gate: tokio::sync::Mutex::new(()),
pool_meta_save_gate: tokio::sync::Mutex::new(
other_store.pool_meta_save_gate.lock().await.independent_clone_for_test(),
pool_meta_save_gate: crate::core::pools::PoolMetaSaveGate::new(
other_store.pool_meta_save_gate.write().await.independent_clone_for_test(),
),
ctx,
bucket_fence_registry: Arc::default(),
+1 -1
View File
@@ -3436,7 +3436,7 @@ mod tests {
rebalance_meta: RwLock::new(None),
decommission_cancelers: RwLock::new(Vec::new()),
start_gate: TokioMutex::new(()),
pool_meta_save_gate: TokioMutex::default(),
pool_meta_save_gate: crate::core::pools::PoolMetaSaveGate::default(),
ctx,
bucket_fence_registry: Arc::default(),
})
@@ -682,15 +682,11 @@ async fn get_pools_info(all_disks: &[Disk]) -> Result<HashMap<i32, HashMap<i32,
return Err(Error::other("ServerNotInitialized"));
};
let mut pools_info: HashMap<i32, HashMap<i32, ErasureSetInfo>> = 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<HashMap<i32, HashMap<i32,
rustfs_io_metrics::internode_metrics::ReadVersionEntrySource::ServerInfoPools,
load,
);
match load.await {
Ok(cache) => 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<F, Fut>(all_disks: &[Disk], mut load_usage: F) -> Result<HashMap<i32, HashMap<i32, ErasureSetInfo>>>
where
F: FnMut(i32, i32) -> Fut,
Fut: std::future::Future<Output = Result<rustfs_data_usage::DataUsageCache>>,
{
let mut pools_info: HashMap<i32, HashMap<i32, ErasureSetInfo>> = 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() {
+355 -85
View File
@@ -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<bool> {
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<os::ShardedPathCache<PreparedPartTxnReceipt>> =
std::sync::LazyLock::new(|| os::ShardedPathCache::new(PART_TXN_RECEIPTS_PER_SHARD));
static PUBLISHED_PART_TXN_RECEIPTS: std::sync::LazyLock<os::ShardedPathCache<PublishedPartTxnReceipt>> =
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<PreparedPartTxnReceipt> {
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<PublishedPartTxnReceipt> {
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) {
@@ -934,6 +979,7 @@ async fn restore_delete_rollback_with_namespace_owner(
.insert(backup_path.clone());
os::rename_all_with_owner(&backup_path, xl_path, object_dir, publication_root, namespace_owner.clone()).await
};
let restored_backup = restored.is_ok();
match restored {
Ok(()) => {}
Err(DiskError::FileNotFound) if remove_new_metadata => {
@@ -963,6 +1009,16 @@ async fn restore_delete_rollback_with_namespace_owner(
os::fsync_dir_with_owner(object_dir, namespace_owner.clone())
.await
.map_err(to_file_error)?;
}
if restored_backup {
// Renaming two hardlinks to the same inode can leave the backup in place.
match os::remove_file_with_owner(&backup_path, namespace_owner.clone()).await {
Ok(()) => {}
Err(err) if err.kind() == ErrorKind::NotFound => {}
Err(err) => return Err(to_file_error(err).into()),
}
}
if effective_durability(volume).syncs_commit_metadata() {
os::fsync_dir_with_owner(&rollback_path, namespace_owner.clone())
.await
.map_err(to_file_error)?;
@@ -3165,6 +3221,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"))]
@@ -3750,6 +3810,25 @@ async fn read_dir_entries_with_walk_stall(path: &Path, count: i32, stall: Option
const WALK_DIR_BATCH_SIZE: usize = 32;
#[cfg(not(test))]
const WALK_DIR_BATCH_SIZE: usize = 256;
async fn read_identifier_tree_entries(path: &Path, leaf: &str, stall: Option<Duration>) -> Result<Vec<String>> {
let mut entries = read_dir_with_walk_stall(path.to_path_buf(), stall).await?;
let mut names = Vec::new();
loop {
let (next, batch, eof) = read_dir_entry_batch_with_walk_stall(entries, READ_DIR_BATCH_SIZE, stall).await?;
entries = next;
// Filter before retaining or sorting the inventory. A report flood
// costs directory iteration, not one retained name/metadata read per report.
names.extend(batch.into_iter().filter(|name| {
name.strip_suffix(SLASH_SEPARATOR)
.is_some_and(|component| component == leaf || super::is_identifier_tree_component(component))
}));
if eof {
return Ok(names);
}
}
}
const READ_DIR_BATCH_SIZE: usize = 512;
async fn read_dir_with_walk_stall(path: PathBuf, stall: Option<Duration>) -> Result<std::fs::ReadDir> {
@@ -9238,15 +9317,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<Arc<dyn Send + Sync>> =
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 +9349,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(());
@@ -11711,6 +11797,14 @@ impl LocalDisk {
W: AsyncWrite + Unpin + Send + 'a,
{
Box::pin(async move {
if opts.identifier_tree_leaf.is_some() {
let relative = current
.strip_prefix(&opts.base_dir)
.ok_or_else(|| DiskError::other("identifier-tree walk escaped its root"))?;
if relative.split('/').filter(|component| !component.is_empty()).count() > 512 {
return Err(DiskError::other("identifier-tree walk exceeds its directory depth limit"));
}
}
// The part of forward_to below this directory, taken before `current`
// loses its trailing slash.
let forward_rest = opts
@@ -11743,7 +11837,12 @@ impl LocalDisk {
// as a single logical read in API semantics.
let read_dir_started = rustfs_io_metrics::get_stage_metrics_enabled().then(std::time::Instant::now);
let dir_path_abs = self.io_get_object_path(&opts.bucket, current.trim_start_matches(SLASH_SEPARATOR))?;
let read_dir_result = match read_dir_entries_with_walk_stall(&dir_path_abs, -1, stall).await {
let read_dir_result = if let Some(leaf) = opts.identifier_tree_leaf.as_deref() {
read_identifier_tree_entries(&dir_path_abs, leaf, stall).await
} else {
read_dir_entries_with_walk_stall(&dir_path_abs, -1, stall).await
};
let read_dir_result = match read_dir_result {
Err(err) if err == Error::FileNotFound && !skip_access_checks(&opts.bucket) => {
let volume_dir = self.io_get_bucket_path(&opts.bucket)?;
if let Err(access_err) = cached_access(&volume_dir).await {
@@ -12040,6 +12139,16 @@ impl LocalDisk {
return Ok(true);
}
if let Some(leaf) = opts.identifier_tree_leaf.as_deref()
&& entry != leaf
{
// Identifier directories can be namespaces at any depth,
// including components named tables or maintenance. Their
// own object metadata is outside this selected leaf scan.
schedule_dir(&mut dir_stack, format!("{name}/"), true, None, true, false);
continue;
}
let mut meta = MetaCacheEntry {
name,
..Default::default()
@@ -12089,7 +12198,8 @@ impl LocalDisk {
} else {
adaptive_batch_cap
};
let batch_names = if batch_limit > 1
let batch_names = if opts.identifier_tree_leaf.is_none()
&& batch_limit > 1
&& dir_objes.is_empty()
&& dir_stack.iter().all(|(_, _, _, scan_required, _)| !*scan_required)
{
@@ -12142,6 +12252,11 @@ impl LocalDisk {
write_metacache_obj(out, &meta).await?;
}
if opts.identifier_tree_leaf.is_some() {
// A dotted leaf cannot contain identifier-tree descendants.
continue;
}
let has_namespace_child_candidate = read.has_namespace_child_candidate;
let dir_to_skip: HashSet<String> = if has_namespace_child_candidate || !dir_objes.is_empty() {
read.data_dirs.into_iter().map(|dir| dir.to_string()).collect()
@@ -12188,6 +12303,9 @@ impl LocalDisk {
}
Err(err) => {
if err == Error::FileNotFound || err == Error::IsNotRegular {
if opts.identifier_tree_leaf.is_some() {
continue;
}
adaptive_batch_cap = 1;
// NOT an object, append to stack (with slash)
// If dirObject, but no metadata (which is unexpected) we skip it.
@@ -14343,7 +14461,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 +14589,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 +14598,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 +14645,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 +14706,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<bool> {
let had_old_data = snapshot_part_transaction_file(
&dst_file_path,
&staging_path.join(PART_TRANSACTION_OLD_DATA),
@@ -14598,10 +14718,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 +14730,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 +14739,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 +14780,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 [&current_data_path, &current_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 [&current_data_path, &current_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 +14868,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 +14935,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 +14951,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 +15077,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 +15176,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)
@@ -15305,6 +15520,12 @@ impl DiskAPI for LocalDisk {
super::lifecycle_test_io::observe(&opts.bucket, None);
self.wait_for_startup_cleanup().await;
if let Some(leaf) = opts.identifier_tree_leaf.as_deref()
&& (!opts.recursive || !opts.base_dir.ends_with('/') || !super::is_identifier_tree_leaf(leaf))
{
return Err(DiskError::other("invalid identifier-tree walk options"));
}
// Callers that do not pin a stall budget inherit the configured default, so
// every local walk keeps a liveness bound even when the wrapper-level total
// timeout is skipped.
@@ -15331,7 +15552,7 @@ impl DiskAPI for LocalDisk {
let mut skip_current_dir_object = false;
let mut multipart_dir_to_skip: HashSet<String> = HashSet::new();
if opts.base_dir.ends_with(SLASH_SEPARATOR) {
if opts.identifier_tree_leaf.is_none() && opts.base_dir.ends_with(SLASH_SEPARATOR) {
if let Ok(data) = with_walk_stall_timeout(
stall,
self.read_listing_metadata(
@@ -19570,6 +19791,55 @@ mod test {
}
}
#[tokio::test]
async fn identifier_tree_walk_filters_report_batches_and_bounds_depth() {
let dir = tempfile::tempdir().expect("create disk fixture");
let root = dir.path().join("catalog/prefix");
fs::create_dir_all(root.join("sales"))
.await
.expect("create identifier directory");
fs::create_dir_all(root.join("table-entry.json"))
.await
.expect("create selected leaf");
for index in 0..READ_DIR_BATCH_SIZE * 3 + 1 {
fs::create_dir(root.join(format!("report-{index}.json")))
.await
.expect("create report directory");
}
let mut names = read_identifier_tree_entries(&root, "table-entry.json", Some(Duration::from_secs(5)))
.await
.expect("all directory batches should be filtered");
names.sort();
assert_eq!(names, ["sales/", "table-entry.json/"]);
let endpoint = Endpoint::try_from(dir.path().to_str().expect("UTF-8 fixture path")).expect("valid endpoint");
let disk = LocalDisk::new(&endpoint, false).await.expect("create local disk");
let options = WalkDirOptions {
bucket: "catalog".to_string(),
base_dir: "prefix/".to_string(),
recursive: true,
identifier_tree_leaf: Some("table-entry.json".to_string()),
..Default::default()
};
let mut output = Vec::new();
let mut writer = MetacacheWriter::new(&mut output);
let mut count = 0;
let error = disk
.scan_dir(
format!("prefix/{}", "a/".repeat(513)),
String::new(),
&options,
&mut writer,
&mut count,
true,
None,
)
.await
.expect_err("excessive depth must fail before filesystem I/O");
assert!(error.to_string().contains("directory depth limit"));
assert_eq!(count, 0);
}
// #4644: the stall budget bounds a single drive read, never the walk as a
// whole. A walk that keeps making progress must survive even when the total
// time spent blocked on a slow consumer dwarfs the stall timeout.
+144 -47
View File
@@ -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<PathBuf> {
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<PathBuf> {
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<PathBuf> {
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<FusedCommitAttempt> {
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,
+392 -59
View File
@@ -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<u8>,
journal: DeleteRecoveryJournal,
sole_transaction: bool,
}
#[derive(Clone)]
struct PublishedDeleteReceiptEntry {
object_path: PathBuf,
bucket_path: PathBuf,
file_path: PathBuf,
staging_root: PathBuf,
journal_raw: Vec<u8>,
}
const PREPARED_DELETE_RECEIPTS_PER_SHARD: usize = 64;
static PREPARED_DELETE_RECEIPTS: std::sync::LazyLock<os::ShardedPathCache<PreparedDeleteReceiptEntry>> =
std::sync::LazyLock::new(|| os::ShardedPathCache::new(PREPARED_DELETE_RECEIPTS_PER_SHARD));
static PUBLISHED_DELETE_RECEIPTS: std::sync::LazyLock<os::ShardedPathCache<PublishedDeleteReceiptEntry>> =
std::sync::LazyLock::new(|| os::ShardedPathCache::new(PREPARED_DELETE_RECEIPTS_PER_SHARD));
fn put_prepared_delete_receipt(rollback: impl AsRef<Path>, entry: PreparedDeleteReceiptEntry) {
PREPARED_DELETE_RECEIPTS.insert(rollback.as_ref(), entry);
}
@@ -1128,6 +1155,27 @@ fn take_prepared_delete_receipt(rollback: &Path) -> Option<PreparedDeleteReceipt
PREPARED_DELETE_RECEIPTS.remove(rollback)
}
fn put_published_delete_receipt(rollback: impl AsRef<Path>, entry: PublishedDeleteReceiptEntry) {
PUBLISHED_DELETE_RECEIPTS.insert(rollback.as_ref(), entry);
}
fn take_published_delete_receipt(rollback: &Path) -> Option<PublishedDeleteReceiptEntry> {
PUBLISHED_DELETE_RECEIPTS.remove(rollback)
}
fn format_commit_receipt_bytes(journal_raw: &[u8], finished: bool) -> Vec<u8> {
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<Arc<dyn Send + Sync>>,
) -> Result<Option<()>> {
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<Arc<dyn Send + Sync>>,
prevalidated_paths: Option<(PathBuf, PathBuf, PathBuf)>,
) -> Result<Option<()>> {
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<Arc<dyn Send + Sync>>,
work_budget: Option<Arc<crate::object_api::DeleteRecoveryWorkBudget>>,
) -> Result<Option<(DeleteRecoveryJournal, bool, bool)>> {
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<Arc<dyn Send + Sync>> = 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<Result<Option<(DeleteRecoveryJournal, bool, bool)>>> {
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<Arc<dyn Send + Sync>> = 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<SnapshotLeaseDeleteClaim> = 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<dyn FnOnce() + Send>;
static HOOKS: LazyLock<Mutex<HashMap<PathBuf, Hook>>> = 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,121 @@ mod tests {
marker.deleted = false;
}
#[cfg(unix)]
#[tokio::test]
async fn delete_recovery_fused_prepare_checkpoint_reopens_before_head_publication() {
use std::os::unix::fs::MetadataExt;
let _mode = durability_mode_override::set(DurabilityMode::Strict);
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");
let source_head = source_root.join(STORAGE_FORMAT_FILE);
let source_head_meta = std::fs::metadata(&source_head).expect("stat source head");
let copy_head = copy_root.join(STORAGE_FORMAT_FILE);
std::fs::copy(&source_head, &copy_head).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());
let dst = destination.join(entry.file_name());
if entry.file_name() == STORAGE_FORMAT_FILE_BACKUP {
let backup_meta = entry.metadata().expect("stat prepared backup");
assert_eq!(
(backup_meta.dev(), backup_meta.ino()),
(source_head_meta.dev(), source_head_meta.ino()),
"fused preparation must hardlink backup to live head"
);
std::fs::hard_link(&copy_head, &dst).expect("preserve hardlinked backup");
} else {
std::fs::copy(entry.path(), dst).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);
let reopened_tx = reopened
.io_get_object_path("recovery", &format!("object/{transaction}"))
.expect("reopened transaction path");
os::fsync_dir_recorder::set_failure(&reopened_tx, ErrorKind::Other);
assert!(
reopened
.settle_delete_transaction(
"recovery",
"object",
transaction,
inspection.fingerprint,
DeleteRecoveryAction::Rollback
)
.await
.is_err(),
"failed transaction directory sync after backup removal must retain journal"
);
assert!(reopened_tx.join(PREPARED).is_file());
assert!(!reopened_tx.join(STORAGE_FORMAT_FILE_BACKUP).exists());
let retry_inspection = reopened
.delete_recovery_inspect("recovery", "object", transaction)
.await
.expect("inspect interrupted rollback");
assert_eq!(retry_inspection.state, DeleteRecoveryState::NewPrepared);
let settled = reopened
.settle_delete_transaction(
"recovery",
"object",
transaction,
retry_inspection.fingerprint,
DeleteRecoveryAction::Rollback,
)
.await
.expect("roll back interrupted preparation");
assert_eq!(settled.state, DeleteRecoveryState::Absent, "same-inode backup must not survive rollback");
assert!(!checkpoint_root.join(transaction.to_string()).exists());
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;
@@ -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<rustix::fd::OwnedFd> {
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);
@@ -1979,25 +1979,28 @@ impl NativeMultipartPublication {
let canonical = self.canonical_head();
let prepared = (|| -> std::io::Result<os::PreparedRenameSource> {
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<std::sync::atomic::AtomicBool>)>,
) -> Result<MultipartPublicationResult> {
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<MultipartPublicationResult> {
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
+94
View File
@@ -2234,6 +2234,36 @@ pub struct WalkDirOptions {
// Override the remote stream stall timeout for long background walks.
#[serde(default)]
pub stall_timeout_ms: Option<u64>,
/// Select this leaf below identifier-like directory components. The root is
/// `base_dir`; unrelated object metadata must not be read or returned.
/// Older peers ignore this additive field, so consumers must also filter.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub identifier_tree_leaf: Option<String>,
}
pub(crate) fn is_identifier_tree_component(value: &str) -> bool {
!value.is_empty()
&& value
.bytes()
.all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'-'))
}
pub(crate) fn is_identifier_tree_leaf(value: &str) -> bool {
value.len() <= 255
&& value.contains('.')
&& !matches!(value, "." | "..")
&& value
.bytes()
.all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'-' | b'.'))
}
pub(crate) fn identifier_tree_leaf_matches(prefix: &str, leaf: &str, object: &str) -> bool {
let Some(relative) = object.strip_prefix(prefix) else {
return false;
};
let mut components = relative.split('/');
components.next_back() == Some(leaf) && components.all(is_identifier_tree_component)
}
impl WalkDirOptions {
@@ -3172,6 +3202,7 @@ mod tests {
skip_total_timeout: false,
timeout_ms: Some(10_000),
stall_timeout_ms: Some(20_000),
identifier_tree_leaf: None,
};
assert_eq!(opts.bucket, "test-bucket");
@@ -3206,6 +3237,69 @@ mod tests {
assert!(!decoded.skip_hidden_prefix_check);
}
#[test]
fn identifier_tree_leaf_selection_preserves_nested_identifier_directories() {
let prefix = "catalog/namespaces/";
let leaf = "table-entry.json";
for relative in [
"sales/tables/orders/table-entry.json",
"sales/tables/orders/maintenance/tables/owner/table-entry.json",
"maintenance/tables/tables/table-entry.json",
] {
assert!(identifier_tree_leaf_matches(prefix, leaf, &format!("{prefix}{relative}")));
}
for relative in [
"sales/tables/orders/not-table-entry.json",
"sales/tables/orders/table-entry.json/child",
"sales/report.json/tables/owner/table-entry.json",
"sales//tables/owner/table-entry.json",
"sales/../tables/owner/table-entry.json",
] {
assert!(!identifier_tree_leaf_matches(prefix, leaf, &format!("{prefix}{relative}")));
}
assert!(!identifier_tree_leaf_matches(prefix, leaf, "other/sales/tables/orders/table-entry.json"));
assert!(is_identifier_tree_leaf(leaf));
for invalid in ["", ".", "..", "table-entry", "table/entry.json", "table\\entry.json"] {
assert!(!is_identifier_tree_leaf(invalid), "invalid selector {invalid}");
}
}
#[test]
fn identifier_tree_walk_options_are_additive_on_json_and_messagepack_maps() {
#[derive(serde::Deserialize)]
struct LegacyView {
bucket: String,
base_dir: String,
recursive: bool,
}
let options = WalkDirOptions {
bucket: "catalog".to_string(),
base_dir: "namespaces/".to_string(),
recursive: true,
identifier_tree_leaf: Some("table-entry.json".to_string()),
..Default::default()
};
let json = serde_json::to_vec(&options).expect("encode current HTTP options");
let messagepack = rmp_serde::to_vec_named(&options).expect("encode legacy RPC map");
for old in [
serde_json::from_slice::<LegacyView>(&json).expect("old JSON readers ignore the additive field"),
rmp_serde::from_slice::<LegacyView>(&messagepack).expect("old MessagePack readers ignore the additive field"),
] {
assert_eq!(old.bucket, options.bucket);
assert_eq!(old.base_dir, options.base_dir);
assert!(old.recursive);
}
let mut old = serde_json::to_value(&options).expect("encode legacy options");
old.as_object_mut().expect("options are a map").remove("identifier_tree_leaf");
let from_json: WalkDirOptions = serde_json::from_value(old.clone()).expect("decode old HTTP options");
let from_messagepack: WalkDirOptions =
rmp_serde::from_slice(&rmp_serde::to_vec_named(&old).expect("encode old RPC map")).expect("decode old RPC map");
assert!(from_json.identifier_tree_leaf.is_none());
assert!(from_messagepack.identifier_tree_leaf.is_none());
assert_eq!(from_json.base_dir, options.base_dir);
assert_eq!(from_messagepack.base_dir, options.base_dir);
}
/// Test DeleteOptions structure
#[test]
fn test_delete_options() {
+38 -3
View File
@@ -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<PreparedRenameSource> {
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)?;
@@ -593,7 +593,7 @@ impl ECStore {
#[cfg(test)]
crate::core::pools::observe_pool_activation_start_attempt(crate::core::pools::PoolActivationStartKind::Rebalance);
let fleet_proof = acquire_pool_activation_fleet_proof(&self.ctx).await?;
let mut pool_meta_guard = self.pool_meta_save_gate.lock().await;
let mut pool_meta_guard = self.pool_meta_save_gate.write().await;
pool_meta_guard.ensure_write_safe(stage)?;
let activation_fence = acquire_pool_rebalance_activation_locks(pool.clone(), fleet_proof).await?;
let pool_meta = self
@@ -625,7 +625,7 @@ impl ECStore {
{
// Lock order: pool_meta_save_gate -> rebalance.bin -> pool.bin. These
// shared fences admit a peer without waiting for the run's own entries.
let mut pool_meta_guard = self.pool_meta_save_gate.lock().await;
let mut pool_meta_guard = self.pool_meta_save_gate.write().await;
pool_meta_guard.ensure_write_safe("rebalance worker activation")?;
let mut worker_fence = acquire_pool_rebalance_worker_locks(pool.clone()).await?;
let pool_meta = self
@@ -654,7 +654,7 @@ impl ECStore {
{
// The admission read fences must already be dropped. Reacquire in the
// original order and revalidate every durable boundary across that gap.
let mut pool_meta_guard = self.pool_meta_save_gate.lock().await;
let mut pool_meta_guard = self.pool_meta_save_gate.write().await;
pool_meta_guard.ensure_write_safe("rebalance worker metadata update")?;
let mut activation_fence = acquire_pool_rebalance_activation_locks(pool.clone(), None).await?;
let pool_meta = self
@@ -1714,7 +1714,7 @@ mod tests {
persist_pool_meta_identity_for_startup(store.pools.clone(), &mut write_state, true)
.await
.expect("initialized pool metadata identity should persist");
*store.pool_meta_save_gate.lock().await = write_state;
*store.pool_meta_save_gate.write().await = write_state;
// Simulate physical loss; raw deletion of authoritative pool.bin is
// deliberately rejected by the native metadata mutation guard.
for pool in &store.pools {
+2 -2
View File
@@ -100,7 +100,7 @@ pub async fn test_store_with_persisted_rebalance_meta(
rebalance_meta: tokio::sync::RwLock::new(Some(meta)),
decommission_cancelers: tokio::sync::RwLock::new(vec![None]),
start_gate: tokio::sync::Mutex::new(()),
pool_meta_save_gate: tokio::sync::Mutex::default(),
pool_meta_save_gate: crate::core::pools::PoolMetaSaveGate::default(),
ctx,
bucket_fence_registry: std::sync::Arc::default(),
});
@@ -335,7 +335,7 @@ async fn test_pool_stores_with_contexts(
rebalance_meta: tokio::sync::RwLock::new(rebalance_meta.clone()),
decommission_cancelers: tokio::sync::RwLock::new(vec![None; pool_count]),
start_gate: tokio::sync::Mutex::new(()),
pool_meta_save_gate: tokio::sync::Mutex::new(pool_meta_write_state.independent_clone_for_test()),
pool_meta_save_gate: crate::core::pools::PoolMetaSaveGate::new(pool_meta_write_state.independent_clone_for_test()),
ctx: store_ctx,
bucket_fence_registry: std::sync::Arc::default(),
})
@@ -3412,7 +3412,7 @@ fn test_store_with_rebalance_meta(meta: RebalanceMeta) -> Arc<crate::store::ECSt
rebalance_meta: tokio::sync::RwLock::new(Some(meta)),
decommission_cancelers: tokio::sync::RwLock::new(Vec::new()),
start_gate: tokio::sync::Mutex::new(()),
pool_meta_save_gate: tokio::sync::Mutex::default(),
pool_meta_save_gate: crate::core::pools::PoolMetaSaveGate::default(),
ctx: crate::runtime::instance::bootstrap_ctx(),
bucket_fence_registry: std::sync::Arc::default(),
})
+39 -2
View File
@@ -2979,10 +2979,11 @@ fn build_legacy_per_part_fallback_reader(
part_length: usize,
part_size: usize,
read_repair: Option<(DecodeReadRepairContext, bool)>,
) -> Box<dyn AsyncRead + Unpin + Send + Sync> {
) -> Box<LegacyPerPartDecodeReader> {
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::<Vec<_>>();
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
+27 -19
View File
@@ -527,7 +527,7 @@ impl ECStore {
async fn acquire_heal_format_fence(
&self,
) -> Result<(
tokio::sync::MutexGuard<'_, PoolMetaWriteState>,
crate::core::pools::PoolMetaSaveGateWriteGuard<'_>,
NamespaceLockGuard,
NamespaceLockGuard,
PoolMeta,
@@ -538,7 +538,7 @@ impl ECStore {
.first()
.cloned()
.ok_or_else(|| Error::other("heal format requires at least one storage pool"))?;
let mut write_state = self.pool_meta_save_gate.lock().await;
let mut write_state = self.pool_meta_save_gate.write().await;
write_state.ensure_write_safe("heal format fence failed")?;
// Metadata fence order is part of the decommission/rebalance protocol:
@@ -1079,7 +1079,7 @@ impl ECStore {
pub(super) async fn handle_heal_bucket(&self, bucket: &str, opts: &HealOpts) -> Result<HealResultItem> {
let movement_gate = self.ctx.data_movement_operation_gate();
let _movement_guard = movement_gate.read().await;
let save_guard = self.pool_meta_save_gate.lock().await;
let save_guard = self.pool_meta_save_gate.write().await;
save_guard.ensure_write_safe("bucket heal cannot run while pool metadata requires recovery")?;
let mut fenced_pools = BTreeSet::new();
{
@@ -2527,7 +2527,7 @@ mod tests {
rebalance_meta: RwLock::new(None),
decommission_cancelers: RwLock::new(Vec::new()),
start_gate: Mutex::new(()),
pool_meta_save_gate: Mutex::default(),
pool_meta_save_gate: crate::core::pools::PoolMetaSaveGate::default(),
ctx: crate::runtime::instance::bootstrap_ctx(),
bucket_fence_registry: std::sync::Arc::default(),
}
@@ -3550,10 +3550,14 @@ mod tests {
async fn full_format_heal_preblocked_pool_metadata_never_writes_format() {
let (temp_dir, store, shutdown) = multi_pool_heal_store().await;
remove_heal_test_format(&temp_dir, &store, 0, 3).await;
store.pool_meta_save_gate.lock().await.observe_replicas(PoolMetaReplicaState {
needs_repair: true,
repair_write_safe: false,
});
store
.pool_meta_save_gate
.write()
.await
.observe_replicas(PoolMetaReplicaState {
needs_repair: true,
repair_write_safe: false,
});
let err = store
.handle_heal_format(false)
@@ -3688,7 +3692,7 @@ mod tests {
let target = remove_heal_test_format(&temp_dir, &store, 0, 3).await;
store.pools[0].disk_set[0].disks.write().await[3] = None;
let barrier = HealFormatAfterSaveBarrier::install(&store.pools[0], 3);
let recovery_latch = store.pool_meta_save_gate.lock().await.aborted_transaction_latch_for_test();
let recovery_latch = store.pool_meta_save_gate.write().await.aborted_transaction_latch_for_test();
let mut heal = tokio::spawn({
let store = Arc::clone(&store);
async move { store.handle_heal_format(false).await }
@@ -3848,7 +3852,7 @@ mod tests {
}
}
let endpoints: EndpointServerPools = pools.iter().map(|pool| pool.endpoints.clone()).collect::<Vec<_>>().into();
let state = source.pool_meta_save_gate.lock().await.independent_clone_for_test();
let state = source.pool_meta_save_gate.write().await.independent_clone_for_test();
Arc::new(ECStore {
id: Uuid::new_v4(),
disk_map: Default::default(),
@@ -3858,7 +3862,7 @@ mod tests {
rebalance_meta: tokio::sync::RwLock::new(None),
decommission_cancelers: tokio::sync::RwLock::new(vec![None; 2]),
start_gate: tokio::sync::Mutex::new(()),
pool_meta_save_gate: tokio::sync::Mutex::new(state),
pool_meta_save_gate: crate::core::pools::PoolMetaSaveGate::new(state),
ctx: Arc::clone(&source.ctx),
bucket_fence_registry: Arc::default(),
})
@@ -4040,7 +4044,7 @@ mod tests {
let missing = remove_pool_meta_shard(&store, 0).await;
let (guard, admissions, mut state) = store.acquire_pool_meta_object_heal_fence(&[0]).await.expect("heal authority");
assert!(admissions.into_iter().all(|admission| admission.is_ok()));
assert!(store.pool_meta_save_gate.try_lock().is_err(), "heal must retain the earlier save gate");
assert!(store.pool_meta_save_gate.try_write().is_err(), "heal must retain the earlier save gate");
let writer_store = Arc::clone(&store);
let writer = tokio::spawn(async move { writer_store.save_current_pool_meta_for_test(&[0]).await });
tokio::task::yield_now().await;
@@ -4687,10 +4691,14 @@ mod tests {
#[tokio::test]
async fn bucket_heal_blocks_before_dispatch_after_unreadable_pool_meta_replica() {
let store = minimal_heal_store().await;
store.pool_meta_save_gate.lock().await.observe_replicas(PoolMetaReplicaState {
needs_repair: true,
repair_write_safe: false,
});
store
.pool_meta_save_gate
.write()
.await
.observe_replicas(PoolMetaReplicaState {
needs_repair: true,
repair_write_safe: false,
});
let err = store
.handle_heal_bucket("bucket", &HealOpts::default())
@@ -4724,7 +4732,7 @@ mod tests {
.await
.expect("peer dispatch should start");
assert!(
store.pool_meta_save_gate.try_lock().is_ok(),
store.pool_meta_save_gate.try_write().is_ok(),
"coordinator must release its local save gate before waiting for peers"
);
assert!(
@@ -5116,7 +5124,7 @@ mod tests {
rebalance_meta: RwLock::new(None),
decommission_cancelers: RwLock::new(Vec::new()),
start_gate: Mutex::new(()),
pool_meta_save_gate: Mutex::default(),
pool_meta_save_gate: crate::core::pools::PoolMetaSaveGate::default(),
ctx: crate::runtime::instance::bootstrap_ctx(),
bucket_fence_registry: std::sync::Arc::default(),
};
@@ -5136,7 +5144,7 @@ mod tests {
.save_for_startup(store.pools.clone())
.await
.expect("pool metadata should be persisted before format heal");
*store.pool_meta_save_gate.lock().await = PoolMetaWriteState::default();
*store.pool_meta_save_gate.write().await = crate::core::pools::PoolMetaWriteState::default();
let replacement_endpoint = store.pools[1].endpoints.endpoints.as_ref()[2].clone();
tokio::fs::remove_file(
+2 -2
View File
@@ -676,7 +676,7 @@ impl ECStore {
rebalance_meta: RwLock::new(None),
decommission_cancelers,
start_gate: Mutex::new(()),
pool_meta_save_gate: Mutex::new(pool_meta_write_state),
pool_meta_save_gate: crate::core::pools::PoolMetaSaveGate::new(pool_meta_write_state),
// Adopt the caller's context (the process bootstrap one on the
// legacy path) so startup writes (erasure type recorded before
// this point) and later reads share one cell.
@@ -725,7 +725,7 @@ impl ECStore {
// Lock order is save gate -> distributed metadata writer, matching
// runtime mutations. RPC locks are available before ECStore publication
// and do not depend on the later fleet proof.
let mut startup_write_state = self.pool_meta_save_gate.lock().await;
let mut startup_write_state = self.pool_meta_save_gate.write().await;
let startup_pool_meta_guard = if should_persist_pool_meta {
let pool = self
.pools
+241 -18
View File
@@ -127,28 +127,30 @@ impl AuthoritativeListOperations for ECStore {
marker: Option<String>,
max_keys: i32,
) -> Result<ListObjectVersionsInfo> {
if max_keys <= 0 || max_keys > MAX_OBJECT_LIST {
self.list_unversioned_authoritative_page(bucket, prefix, marker, max_keys, None)
.await
}
async fn list_identifier_tree_leaves_authoritative(
self: Arc<Self>,
bucket: &str,
prefix: &str,
leaf: &str,
marker: Option<String>,
max_keys: i32,
) -> Result<ListObjectVersionsInfo> {
let valid_prefix = prefix
.strip_suffix('/')
.is_some_and(|root| !root.is_empty() && root.split('/').all(|part| !part.is_empty() && !matches!(part, "." | "..")));
if !valid_prefix || !crate::disk::is_identifier_tree_leaf(leaf) {
return Err(StorageError::InvalidArgument(
bucket.to_string(),
prefix.to_string(),
"authoritative page size is out of range".to_string(),
"identifier-tree listing requires a directory prefix and a dotted leaf name".to_string(),
));
}
let opts = ListPathOptions {
bucket: bucket.to_string(),
prefix: prefix.to_string(),
marker: marker.clone(),
limit: max_keys,
recursive: true,
incl_deleted: true,
versioned: true,
authoritative: true,
caller: ListObjectsGatherCaller::S3ListVersions,
..Default::default()
};
let mut page = self.list_path(&opts).await?;
page.entries.forward_past(marker);
unversioned_authoritative_page(bucket, page)
self.list_unversioned_authoritative_page(bucket, prefix, marker, max_keys, Some(leaf.to_string()))
.await
}
async fn walk_authoritative(
@@ -436,6 +438,8 @@ pub struct ListPathOptions {
// Internal bounded pages require GET-equivalent per-set resolution.
pub authoritative: bool,
pub identifier_tree_leaf: Option<String>,
pub stop_disk_at_limit: bool,
pub pool_idx: Option<usize>,
@@ -1101,6 +1105,39 @@ impl ListPathOptions {
}
impl ECStore {
async fn list_unversioned_authoritative_page(
self: Arc<Self>,
bucket: &str,
prefix: &str,
marker: Option<String>,
max_keys: i32,
identifier_tree_leaf: Option<String>,
) -> Result<ListObjectVersionsInfo> {
if max_keys <= 0 || max_keys > MAX_OBJECT_LIST {
return Err(StorageError::InvalidArgument(
bucket.to_string(),
prefix.to_string(),
"authoritative page size is out of range".to_string(),
));
}
let opts = ListPathOptions {
bucket: bucket.to_string(),
prefix: prefix.to_string(),
marker: marker.clone(),
limit: max_keys,
recursive: true,
incl_deleted: true,
versioned: true,
authoritative: true,
identifier_tree_leaf,
caller: ListObjectsGatherCaller::S3ListVersions,
..Default::default()
};
let mut page = self.list_path(&opts).await?;
page.entries.forward_past(marker);
unversioned_authoritative_page(bucket, page)
}
#[allow(clippy::too_many_arguments)]
// @continuation_token marker
// @start_after as marker when continuation_token empty
@@ -2936,6 +2973,33 @@ impl SetDisks {
Ok(page)
}
fn resolve_authoritative_listed_entry(
bucket: &str,
selection: Option<(&str, &str)>,
entries: MetaCacheEntries,
errors: &[Option<DiskError>],
disk_count: usize,
parity: usize,
) -> crate::disk::error::Result<Option<MetaCacheEntry>> {
if disk_count == 0 || entries.0.len() > disk_count || entries.0.len() != errors.len() {
return Err(DiskError::ErasureReadQuorum);
}
// An old peer may ignore the optional disk selector. Its noncandidate
// rows must be ignored before metadata parsing. Selected keys retain
// the original resolver and all of its quorum/corruption checks.
if selection.is_some_and(|(prefix, leaf)| {
entries
.0
.iter()
.flatten()
.next()
.is_some_and(|entry| !crate::disk::identifier_tree_leaf_matches(prefix, leaf, &entry.name))
}) {
return Ok(None);
}
Self::resolve_listed_versions(bucket, entries, errors, disk_count, parity)
}
/// Version scan for set-scoped callers that act on absence: set walks
/// (tier reference proofs, tier-delete journals, pool maintenance) and the
/// recursive prefix-delete scan. Every online drive is read and each
@@ -2959,6 +3023,8 @@ impl SetDisks {
let first_error = Arc::new(tokio::sync::Mutex::new(None));
let error_sink = Arc::clone(&first_error);
let bucket = opts.bucket.clone();
let prefix = opts.prefix.clone();
let identifier_tree_leaf = opts.identifier_tree_leaf.clone();
let cancel = rx.clone();
let result = list_path_raw_with_claim_tracker(
rx,
@@ -2976,6 +3042,7 @@ impl SetDisks {
forward_to: opts.marker,
min_disks: read_quorum,
preserve_replica_metadata: true,
identifier_tree_leaf: opts.identifier_tree_leaf,
// A reader's local page may be consumed by stale entries. Only
// the resolved, merged page may stop the authoritative walk.
per_disk_limit: 0,
@@ -2983,7 +3050,14 @@ impl SetDisks {
walkdir_timeout: opts.walkdir_timeout,
walkdir_stall_timeout: opts.walkdir_stall_timeout,
partial: Some(Box::new(move |entries, errors| {
let resolved = Self::resolve_listed_versions(&bucket, entries, errors, disk_count, parity);
let resolved = Self::resolve_authoritative_listed_entry(
&bucket,
identifier_tree_leaf.as_deref().map(|leaf| (prefix.as_str(), leaf)),
entries,
errors,
disk_count,
parity,
);
let sender = sender.clone();
let cancel = cancel.clone();
let first_error = Arc::clone(&error_sink);
@@ -6997,6 +7071,155 @@ mod test {
);
}
#[test]
fn authoritative_identifier_tree_filters_old_peer_rows_before_metadata_resolution() {
let prefix = "table-catalog/namespaces/";
let leaf = "table-entry.json";
let corrupt = |name: &str| {
MetaCacheEntries(vec![
Some(MetaCacheEntry {
name: format!("{prefix}{name}"),
metadata: b"corrupt metadata".to_vec(),
..Default::default()
});
4
])
};
let report = "sales/tables/orders/maintenance/hash/jobs/job.json";
let errors = vec![None; 4];
assert!(
SetDisks::resolve_authoritative_listed_entry(RUSTFS_META_BUCKET, None, corrupt(report), &errors, 4, 2,).is_err(),
"the control must actually parse and reject the corrupt metadata"
);
assert!(
SetDisks::resolve_authoritative_listed_entry(
RUSTFS_META_BUCKET,
Some((prefix, leaf)),
corrupt(report),
&errors,
4,
2,
)
.expect("an old peer's unrelated report is outside the selected scan")
.is_none()
);
assert!(
SetDisks::resolve_authoritative_listed_entry(
RUSTFS_META_BUCKET,
Some((prefix, leaf)),
corrupt("sales/tables/orders/maintenance/tables/owner/table-entry.json"),
&errors,
4,
2,
)
.is_err(),
"a matching nested owner must still fail closed on corruption"
);
}
#[tokio::test]
async fn authoritative_identifier_tree_pages_preserve_nested_owners_and_read_quorum() {
let (dirs, store) = single_set_store(4).await;
let prefix = "table-catalog/identifier-tree/";
let leaf = "table-entry.json";
for (root, selected) in [
("table-catalog/identifier-tree", leaf),
(prefix, "../entry.json"),
(prefix, "table-entry"),
] {
assert!(matches!(
store
.clone()
.list_identifier_tree_leaves_authoritative(RUSTFS_META_BUCKET, root, selected, None, 2)
.await,
Err(StorageError::InvalidArgument(..))
));
}
let names = [
format!("{prefix}a/tables/first/{leaf}"),
format!("{prefix}sales/tables/orders/maintenance/tables/owner/{leaf}"),
format!("{prefix}z/tables/last/{leaf}"),
];
for name in &names {
write_version_replicas(&dirs, RUSTFS_META_BUCKET, name, |slot| Some(unversioned_catalog_metadata(name, slot))).await;
}
for name in [
format!("{prefix}sales"),
format!("{prefix}sales/tables/orders/maintenance/hash/jobs/job.json"),
] {
write_version_replicas(&dirs, RUSTFS_META_BUCKET, &name, |_| Some(b"corrupt noncandidate metadata".to_vec())).await;
}
assert!(
store
.clone()
.list_unversioned_objects_authoritative(RUSTFS_META_BUCKET, prefix, None, 100)
.await
.is_err(),
"the ordinary authoritative scan must expose the corrupt control objects"
);
let first = store
.clone()
.list_identifier_tree_leaves_authoritative(RUSTFS_META_BUCKET, prefix, leaf, None, 2)
.await
.expect("noncandidate metadata must not be read");
assert!(first.is_truncated);
assert_eq!(
first.objects.iter().map(|object| &object.name).collect::<Vec<_>>(),
names[..2].iter().collect::<Vec<_>>()
);
let second = store
.clone()
.list_identifier_tree_leaves_authoritative(RUSTFS_META_BUCKET, prefix, leaf, first.next_marker, 2)
.await
.expect("the next page must retain the final owner");
assert!(!second.is_truncated);
assert_eq!(second.objects.iter().map(|object| &object.name).collect::<Vec<_>>(), vec![&names[2]]);
write_version_replicas(&dirs, RUSTFS_META_BUCKET, &names[0], |_| Some(b"corrupt selected metadata".to_vec())).await;
assert!(
store
.clone()
.list_identifier_tree_leaves_authoritative(RUSTFS_META_BUCKET, prefix, leaf, None, 2)
.await
.is_err(),
"corruption at a selected key cannot be skipped to fill the page"
);
write_version_replicas(&dirs, RUSTFS_META_BUCKET, &names[0], |slot| {
Some(unversioned_catalog_metadata(&names[0], slot))
})
.await;
write_version_replicas(&dirs, RUSTFS_META_BUCKET, &names[0], |slot| {
Some(versioned_object_metadata(RUSTFS_META_BUCKET, &names[0], &[(2, 1_700_000_002)], slot).to_vec())
})
.await;
assert!(
store
.clone()
.list_identifier_tree_leaves_authoritative(RUSTFS_META_BUCKET, prefix, leaf, None, 2)
.await
.is_err(),
"a versioned ownership entry cannot be accepted as an unversioned proof"
);
write_version_replicas(&dirs, RUSTFS_META_BUCKET, &names[0], |slot| {
Some(unversioned_catalog_metadata(&names[0], slot))
})
.await;
let set = &store.pools[0].disk_set[0];
for disk in set.disks.write().await.iter_mut().skip(1) {
*disk = None;
}
assert!(
matches!(
store
.clone()
.list_identifier_tree_leaves_authoritative(RUSTFS_META_BUCKET, prefix, leaf, None, 2)
.await,
Err(StorageError::ErasureReadQuorum)
),
"unavailable replicas cannot prove owner absence"
);
}
#[tokio::test]
async fn authoritative_store_walk_includes_reserved_prefix_versions() {
let bucket = "catalog-cleanup-versions";
+8 -6
View File
@@ -34,7 +34,7 @@ use crate::bucket::utils::check_put_object_part_args;
use crate::bucket::utils::{check_valid_bucket_name, check_valid_bucket_name_strict, is_meta_bucketname};
use crate::cluster::rpc::{RemoteClient, S3PeerSys};
use crate::config::storageclass;
use crate::core::pools::{DecommissionCanceler, PoolMeta, PoolMetaWriteState};
use crate::core::pools::{DecommissionCanceler, PoolMeta, PoolMetaSaveGate};
use crate::disk::endpoint::{Endpoint, EndpointType};
use crate::disk::{DiskAPI, DiskInfo, DiskInfoOptions};
use crate::error::{Error, Result};
@@ -482,9 +482,13 @@ pub struct ECStore {
/// Serializes full-document pool metadata saves and retains a fail-closed
/// write block after startup observes an unreadable replica.
///
/// Immutable admission proofs share authority; publication, recovery and
/// authority changes retain exclusive access through their existing I/O.
/// Each writer publishes its final state on release, and health
/// inspection reads that committed state without acquiring the gate.
/// Lock order: acquire `pool_meta_save_gate`, then the distributed
/// `pool.bin` fence, then clone `pool_meta` under a short read lock.
pub(crate) pool_meta_save_gate: Mutex<PoolMetaWriteState>,
pub(crate) pool_meta_save_gate: PoolMetaSaveGate,
/// Per-instance runtime state (Phase 5, backlog#939).
///
/// Carries this instance's identity/runtime out of the process globals so
@@ -557,7 +561,6 @@ impl Default for ScannerDataMovementPauseStatus {
#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize)]
pub struct PoolMetaWriteGateStatus {
pub writes_ready: bool,
pub check_timed_out: bool,
pub write_blocked: bool,
pub transaction_aborted: bool,
pub pool_meta_absent: bool,
@@ -573,7 +576,6 @@ impl Default for PoolMetaWriteGateStatus {
fn default() -> Self {
Self {
writes_ready: true,
check_timed_out: false,
write_blocked: false,
transaction_aborted: false,
pool_meta_absent: false,
@@ -1938,7 +1940,7 @@ mod tests {
rebalance_meta: RwLock::new(None),
decommission_cancelers: RwLock::new(Vec::new()),
start_gate: Mutex::new(()),
pool_meta_save_gate: Mutex::default(),
pool_meta_save_gate: crate::core::pools::PoolMetaSaveGate::default(),
ctx,
bucket_fence_registry: Arc::default(),
};
@@ -2013,7 +2015,7 @@ mod tests {
rebalance_meta: RwLock::new(None),
decommission_cancelers: RwLock::new(Vec::new()),
start_gate: Mutex::new(()),
pool_meta_save_gate: Mutex::default(),
pool_meta_save_gate: crate::core::pools::PoolMetaSaveGate::default(),
ctx,
bucket_fence_registry: Arc::default(),
})
+1 -1
View File
@@ -2133,7 +2133,7 @@ mod tests {
rebalance_meta: RwLock::new(None),
decommission_cancelers: RwLock::new(Vec::new()),
start_gate: Mutex::new(()),
pool_meta_save_gate: Mutex::default(),
pool_meta_save_gate: crate::core::pools::PoolMetaSaveGate::default(),
ctx: crate::runtime::instance::bootstrap_ctx(),
bucket_fence_registry: std::sync::Arc::default(),
}
+2 -2
View File
@@ -10007,7 +10007,7 @@ mod tests {
rebalance_meta: RwLock::new(None),
decommission_cancelers: RwLock::new(Vec::new()),
start_gate: Mutex::new(()),
pool_meta_save_gate: Mutex::default(),
pool_meta_save_gate: crate::core::pools::PoolMetaSaveGate::default(),
ctx: crate::runtime::instance::bootstrap_ctx(),
bucket_fence_registry: std::sync::Arc::default(),
}
@@ -10099,7 +10099,7 @@ mod tests {
rebalance_meta: RwLock::new(None),
decommission_cancelers: RwLock::new(Vec::new()),
start_gate: Mutex::new(()),
pool_meta_save_gate: Mutex::default(),
pool_meta_save_gate: crate::core::pools::PoolMetaSaveGate::default(),
ctx,
bucket_fence_registry: std::sync::Arc::default(),
}
@@ -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())
+45 -4
View File
@@ -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)));
+5
View File
@@ -64,6 +64,8 @@ async-compression = { workspace = true, features = ["tokio", "gzip", "bzip2"] }
async-trait.workspace = true
arc-swap.workspace = true
bytes = { workspace = true, features = ["serde"] }
base64-simd.workspace = true
flatbuffers.workspace = true
chrono = { workspace = true, features = ["serde"] }
crc-fast.workspace = true
rustfs-common.workspace = true
@@ -75,6 +77,9 @@ rustfs-storage-api.workspace = true
futures = { workspace = true }
futures-core = { workspace = true }
flate2.workspace = true
brotli.workspace = true
lz4.workspace = true
lz4_flex.workspace = true
http.workspace = true
memchr.workspace = true
serde_json = { workspace = true, features = ["raw_value"] }
+32
View File
@@ -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<Select
if let Some(error) = err.downcast_ref::<SelectError>() {
return Some(error.clone());
}
if let Some(error) = err.downcast_ref::<parquet_budget::ParquetBudgetError>() {
return Some(match error {
parquet_budget::ParquetBudgetError::Exceeded => SelectError::ResourceExhausted,
parquet_budget::ParquetBudgetError::Invalid(_) | parquet_budget::ParquetBudgetError::Codec(_) => {
SelectError::ParquetParsingError
}
});
}
if let Some(error) = err.downcast_ref::<object_store::SelectObjectStoreError>() {
return Some(error.select_error());
}
@@ -333,9 +342,32 @@ impl Display for ResolvedTable {
#[cfg(test)]
mod tests {
use super::*;
use datafusion::common::DataFusionError;
use datafusion::sql::sqlparser::parser::ParserError;
#[test]
fn parquet_admission_errors_preserve_select_error_codes() {
for (source, expected) in [
(parquet_budget::ParquetBudgetError::Exceeded, SelectError::ResourceExhausted),
(
parquet_budget::ParquetBudgetError::Invalid("malformed footer".into()),
SelectError::ParquetParsingError,
),
(
parquet_budget::ParquetBudgetError::Codec(std::io::Error::from(std::io::ErrorKind::InvalidData)),
SelectError::ParquetParsingError,
),
] {
let error = QueryError::from(DataFusionError::External(Box::new(source)));
assert_eq!(error.select_error(), expected);
}
let nested = QueryError::from(DataFusionError::ParquetError(Box::new(ParquetError::External(Box::new(
parquet_budget::ParquetBudgetError::Exceeded,
)))));
assert_eq!(nested.select_error(), SelectError::ResourceExhausted);
}
#[test]
fn test_query_error_display() {
let err = QueryError::NotImplemented {
File diff suppressed because it is too large Load Diff
@@ -424,7 +424,8 @@ impl SimpleQueryDispatcher {
tracker: &QueryExecutionTracker,
) -> QueryResult<MetadataProvider> {
if self.input.request.input_serialization.parquet.is_some() {
let provider = ParquetSelectTable::try_new(session.inner(), self.input.as_ref()).await?;
let query_guard = tracker.query_guard().ok_or(QueryError::Cancel)?;
let provider = ParquetSelectTable::try_new(session.inner(), self.input.as_ref(), Some(query_guard)).await?;
let current_session_table_provider = self.build_table_handle_provider()?;
let metadata_provider =
MetadataProvider::new(provider, current_session_table_provider, self.func_manager.clone(), session.clone());
@@ -27,7 +27,10 @@ use datafusion::{
datasource::{
TableProvider,
listing::{ListingTableUrl, PartitionedFile},
physical_plan::{FileScanConfigBuilder, ParquetSource, parquet::ParquetAccessPlan},
physical_plan::{
FileScanConfigBuilder, ParquetSource,
parquet::{ParquetAccessPlan, ParquetFileReaderFactory},
},
source::DataSourceExec,
},
execution::{memory_pool::MemoryLimit, object_store::ObjectStoreUrl},
@@ -40,6 +43,7 @@ use datafusion::{
},
physical_plan::{
ExecutionPlan, SendableRecordBatchStream,
metrics::ExecutionPlanMetricsSet,
stream::RecordBatchStreamAdapter,
streaming::{PartitionStream, StreamingTableExec},
},
@@ -49,6 +53,8 @@ use rustfs_s3select_api::request::SelectObjectContentInput;
use rustfs_s3select_api::{
QueryResult,
object_store::{SelectScanRange, scan_range_from_bounds},
parquet_budget::{ParquetBudgetError, page_decode_budget, validate_footer},
query::session::QueryExecutionGuard,
};
#[derive(Clone)]
@@ -59,6 +65,7 @@ pub struct ParquetSelectTable {
object_size: u64,
access_plan: Option<Arc<ParquetAccessPlan>>,
expanded_row_budget: usize,
reader_factory: Arc<BudgetParquetReaderFactory>,
}
impl fmt::Debug for ParquetSelectTable {
@@ -72,11 +79,30 @@ impl fmt::Debug for ParquetSelectTable {
}
}
#[derive(Debug)]
struct ObjectStoreParquetReader {
store: Arc<dyn ObjectStore>,
path: Path,
file_size: u64,
metadata_read_limit: usize,
metadata_decode_limit: usize,
metadata: Option<Arc<ParquetMetaData>>,
query_guard: Option<QueryExecutionGuard>,
}
async fn validate_in_worker<T: Send + 'static>(
query_guard: Option<QueryExecutionGuard>,
operation: impl FnOnce() -> Result<T, ParquetBudgetError> + Send + 'static,
) -> ParquetResult<T> {
tokio::task::spawn_blocking(move || {
// Cancellation cannot release admission while the decoder preflight
// still owns input buffers or codec allocations.
let _query_guard = query_guard;
operation()
})
.await
.map_err(|error| ParquetError::External(Box::new(error)))?
.map_err(|error| ParquetError::External(Box::new(error)))
}
impl AsyncFileReader for ObjectStoreParquetReader {
@@ -94,7 +120,16 @@ impl AsyncFileReader for ObjectStoreParquetReader {
if total.is_none_or(|total| total > self.metadata_read_limit as u64) {
return async { Err(ParquetError::External(Box::new(parquet_budget_error()))) }.boxed();
}
async move { self.store.get_ranges(&self.path, &ranges).await.map_err(parquet_store_error) }.boxed()
async move {
// Read requested ranges directly so sparse offsets cannot amplify
// a small request into a large coalesced allocation.
let mut bytes = Vec::with_capacity(ranges.len());
for range in ranges {
bytes.push(self.store.get_range(&self.path, range).await.map_err(parquet_store_error)?);
}
Ok(bytes)
}
.boxed()
}
fn get_metadata<'a>(
@@ -102,6 +137,27 @@ impl AsyncFileReader for ObjectStoreParquetReader {
options: Option<&'a ArrowReaderOptions>,
) -> BoxFuture<'a, ParquetResult<Arc<ParquetMetaData>>> {
async move {
if let Some(metadata) = self.metadata.as_ref() {
return Ok(Arc::clone(metadata));
}
let tail_start = self
.file_size
.checked_sub(8)
.ok_or_else(|| ParquetError::EOF("Parquet footer is truncated".into()))?;
let tail = self.get_bytes(tail_start..self.file_size).await?;
let tail: &[u8; 8] = tail
.as_ref()
.try_into()
.map_err(|_| ParquetError::EOF("Parquet footer is truncated".into()))?;
let footer = datafusion::parquet::file::metadata::FooterTail::try_new(tail)?;
let length =
u64::try_from(footer.metadata_length()).map_err(|_| ParquetError::External(Box::new(parquet_budget_error())))?;
let start = tail_start
.checked_sub(length)
.ok_or_else(|| ParquetError::EOF("Parquet metadata is truncated".into()))?;
let bytes = self.get_bytes(start..tail_start).await?;
let limit = self.metadata_decode_limit;
validate_in_worker(self.query_guard.clone(), move || validate_footer(&bytes, limit)).await?;
let metadata_options = options.map(|options| options.metadata_options().clone());
let mut metadata_reader = ParquetMetaDataReader::new().with_metadata_options(metadata_options);
@@ -119,8 +175,84 @@ impl AsyncFileReader for ObjectStoreParquetReader {
}
}
#[derive(Debug)]
struct BudgetParquetReaderFactory {
store: Arc<dyn ObjectStore>,
path: Path,
file_size: u64,
metadata: Arc<ParquetMetaData>,
read_limit: usize,
}
impl ParquetFileReaderFactory for BudgetParquetReaderFactory {
fn create_reader(
&self,
_partition_index: usize,
file: PartitionedFile,
_metadata_size_hint: Option<usize>,
_metrics: &ExecutionPlanMetricsSet,
) -> DFResult<Box<dyn AsyncFileReader + Send>> {
if file.object_meta.location != self.path || file.object_meta.size != self.file_size {
return Err(DataFusionError::Execution("Parquet reader input differs from its admitted object".into()));
}
Ok(Box::new(ObjectStoreParquetReader {
store: Arc::clone(&self.store),
path: self.path.clone(),
file_size: self.file_size,
metadata_read_limit: self.read_limit,
metadata_decode_limit: self.read_limit,
metadata: Some(Arc::clone(&self.metadata)),
query_guard: None,
}))
}
}
async fn validate_parquet_pages(
store: &Arc<dyn ObjectStore>,
path: &Path,
file_size: u64,
metadata: &ParquetMetaData,
plan: Option<&ParquetAccessPlan>,
limit: usize,
query_guard: &Option<QueryExecutionGuard>,
) -> DFResult<()> {
for (index, group) in metadata.row_groups().iter().enumerate() {
if plan.is_some_and(|plan| !plan.should_scan(index)) {
continue;
}
let mut resident = 0usize;
for column in group.columns() {
let start = u64::try_from(column.dictionary_page_offset().unwrap_or(column.data_page_offset()))
.map_err(|_| parquet_budget_error())?;
let length = u64::try_from(column.compressed_size()).map_err(|_| parquet_budget_error())?;
let end = start
.checked_add(length)
.filter(|end| *end <= file_size)
.ok_or_else(parquet_budget_error)?;
if length > u64::try_from(limit).map_err(|_| parquet_budget_error())? {
return Err(parquet_budget_error());
}
let bytes = store.get_range(path, start..end).await?;
let declared = usize::try_from(column.uncompressed_size()).map_err(|_| parquet_budget_error())?;
let codec = column.compression();
let required = validate_in_worker(query_guard.clone(), move || page_decode_budget(&bytes, declared, limit, codec))
.await
.map_err(DataFusionError::from)?;
resident = resident
.checked_add(required)
.filter(|total| *total <= limit)
.ok_or_else(parquet_budget_error)?;
}
}
Ok(())
}
impl ParquetSelectTable {
pub async fn try_new(state: &dyn Session, input: &SelectObjectContentInput) -> QueryResult<Arc<dyn TableProvider>> {
pub async fn try_new(
state: &dyn Session,
input: &SelectObjectContentInput,
query_guard: Option<QueryExecutionGuard>,
) -> QueryResult<Arc<dyn TableProvider>> {
let table_path = ListingTableUrl::parse(format!("s3://{}/{}", input.bucket, input.key))?;
let object_store_url = table_path.object_store();
let object_location = Path::from(input.key.clone());
@@ -136,9 +268,12 @@ impl ParquetSelectTable {
};
let reader = ObjectStoreParquetReader {
store: Arc::clone(&store),
path: object_location,
path: object_location.clone(),
file_size: object_meta.size,
metadata_read_limit: (memory_budget / 16).max(1),
metadata_decode_limit: (memory_budget / 4).max(1),
metadata: None,
query_guard: query_guard.clone(),
};
let builder = ParquetRecordBatchStreamBuilder::new(reader)
.await
@@ -153,11 +288,26 @@ impl ParquetSelectTable {
builder.schema().metadata().clone(),
));
let metadata = Arc::clone(builder.metadata());
// This is admission for valid Parquet declarations, not a replacement
// for limits inside the footer parser or native compression codecs.
let expanded_row_budget = (memory_budget / 16).max(1);
let access_plan = parquet_access_plan(input, object_meta.size, metadata.as_ref())?;
admit_parquet_decode(metadata.as_ref(), access_plan.as_deref(), memory_budget / 4)?;
validate_parquet_pages(
&store,
&object_location,
object_meta.size,
&metadata,
access_plan.as_deref(),
memory_budget / 4,
&query_guard,
)
.await?;
let reader_factory = Arc::new(BudgetParquetReaderFactory {
store,
path: object_location,
file_size: object_meta.size,
metadata,
read_limit: (memory_budget / 4).max(1),
});
Ok(Arc::new(Self {
schema,
@@ -166,6 +316,7 @@ impl ParquetSelectTable {
object_size: object_meta.size,
access_plan,
expanded_row_budget,
reader_factory,
}))
}
@@ -202,7 +353,12 @@ impl TableProvider for ParquetSelectTable {
));
}
let scan_limit = if filters.is_empty() { limit } else { None };
let file_source = Arc::new(ParquetSource::new(Arc::clone(&self.schema)));
let file_source = Arc::new(
ParquetSource::new(Arc::clone(&self.schema))
.with_parquet_file_reader_factory(self.reader_factory.clone())
.with_enable_page_index(false)
.with_bloom_filter_on_read(false),
);
let config = FileScanConfigBuilder::new(self.object_store_url.clone(), file_source)
.with_file(self.partitioned_file())
.with_projection_indices(projection.cloned())?
@@ -439,6 +595,39 @@ pub(super) fn dictionary_parquet_for_test(value_bytes: usize, rows: usize) -> Ve
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn parquet_cancelled_preflight_retains_admission_until_worker_finishes() {
let admission = Arc::new(tokio::sync::Semaphore::new(1));
let permit = Arc::clone(&admission).acquire_owned().await.expect("initial admission");
let (started_tx, started_rx) = tokio::sync::oneshot::channel();
let (release_tx, release_rx) = std::sync::mpsc::channel();
let task = tokio::spawn(validate_in_worker(Some(Arc::new(permit)), move || {
started_tx.send(()).expect("report running worker");
release_rx
.recv_timeout(std::time::Duration::from_secs(5))
.expect("worker release");
Ok(())
}));
tokio::time::timeout(std::time::Duration::from_secs(5), started_rx)
.await
.expect("worker starts before deadline")
.expect("worker started");
task.abort();
assert!(task.await.expect_err("cancelled request task").is_cancelled());
assert!(Arc::clone(&admission).try_acquire_owned().is_err(), "running worker still owns admission");
release_tx.send(()).expect("release worker");
tokio::time::timeout(std::time::Duration::from_secs(5), async {
loop {
if Arc::clone(&admission).try_acquire_owned().is_ok() {
break;
}
tokio::task::yield_now().await;
}
})
.await
.expect("worker releases admission after completion");
}
use datafusion::{
arrow::{
array::Int32Array,
@@ -519,7 +708,7 @@ mod tests {
let context = parquet_session(store);
let state = context.state();
let error = match ParquetSelectTable::try_new(&state, &parquet_input("missing.parquet")).await {
let error = match ParquetSelectTable::try_new(&state, &parquet_input("missing.parquet"), None).await {
Ok(_) => panic!("missing parquet object must fail"),
Err(error) => error,
};
@@ -538,7 +727,7 @@ mod tests {
let context = parquet_session(store);
let state = context.state();
let error = match ParquetSelectTable::try_new(&state, &parquet_input(object.as_ref())).await {
let error = match ParquetSelectTable::try_new(&state, &parquet_input(object.as_ref()), None).await {
Ok(_) => panic!("corrupt parquet metadata must fail"),
Err(error) => error,
};
@@ -581,7 +770,7 @@ mod tests {
.put(&Path::from("dictionary.parquet"), Bytes::from(bytes).into())
.await
.expect("fixture");
let table = ParquetSelectTable::try_new(&context.state(), &parquet_input("dictionary.parquet"))
let table = ParquetSelectTable::try_new(&context.state(), &parquet_input("dictionary.parquet"), None)
.await
.expect("valid dictionary metadata");
context.register_table("source", table).expect("table");
@@ -752,7 +941,7 @@ mod tests {
start: Some(starts[0]),
end: Some(starts[0]),
});
let table = ParquetSelectTable::try_new(&context.state(), &input)
let table = ParquetSelectTable::try_new(&context.state(), &input, None)
.await
.expect("an unselected large row group must not reject the small range");
context.register_table("source", table).expect("table");
@@ -770,7 +959,7 @@ mod tests {
start: Some(starts[1]),
end: Some(starts[1]),
});
let error = match ParquetSelectTable::try_new(&context.state(), &input).await {
let error = match ParquetSelectTable::try_new(&context.state(), &input, None).await {
Ok(_) => panic!("the selected oversized row group must still be rejected before decoding"),
Err(error) => error,
};
@@ -806,13 +995,209 @@ mod tests {
.put(&Path::from("large.parquet"), Bytes::from(bytes).into())
.await
.expect("fixture");
let error = match ParquetSelectTable::try_new(&context.state(), &parquet_input("large.parquet")).await {
let error = match ParquetSelectTable::try_new(&context.state(), &parquet_input("large.parquet"), None).await {
Ok(_) => panic!("valid declared decoded size must be admitted before page decode"),
Err(error) => error,
};
assert!(error.to_string().contains("query memory budget"));
}
#[tokio::test]
async fn parquet_footer_collection_budget_is_checked_before_metadata_decode() {
// A root schema with no leaves and 50 empty, valid row groups. The
// encoded footer fits the read budget; its reservations do not.
let mut footer = vec![0x15, 2, 0x19, 0x1c, 0x48, 4];
footer.extend(b"root");
footer.extend([0x15, 0, 0, 0x16, 0, 0x19, 0xfc, 50]);
for _ in 0..50 {
footer.extend([0x19, 0x0c, 0x16, 0, 0x16, 0, 0]);
}
footer.push(0);
let decoded = ParquetMetaDataReader::decode_metadata(&footer).expect("valid compact footer fixture");
assert_eq!(decoded.num_row_groups(), 50);
let mut bytes = b"PAR1".to_vec();
bytes.extend(&footer);
bytes.extend(u32::try_from(footer.len()).expect("small footer").to_le_bytes());
bytes.extend(b"PAR1");
let store = Arc::new(InMemory::new());
let context = parquet_session_with_memory(store.clone(), 8192);
store
.put(&Path::from("footer.parquet"), Bytes::from(bytes).into())
.await
.expect("fixture");
let error = match ParquetSelectTable::try_new(&context.state(), &parquet_input("footer.parquet"), None).await {
Ok(_) => panic!("footer collection reservations exceed the query budget"),
Err(error) => error,
};
assert_eq!(error.select_error(), SelectError::ResourceExhausted);
}
#[tokio::test]
async fn parquet_forged_footer_cannot_hide_oversized_compressed_pages() {
use datafusion::parquet::file::metadata::ParquetMetaDataWriter;
let schema = Arc::new(Schema::new(vec![Field::new("v", DataType::Utf8, false)]));
let batch = RecordBatch::try_new(
Arc::clone(&schema),
vec![Arc::new(datafusion::arrow::array::StringArray::from(vec![
"x".repeat(32 * 1024 * 1024),
]))],
)
.expect("large compressible value");
let properties = datafusion::parquet::file::properties::WriterProperties::builder()
.set_dictionary_page_size_limit(64 * 1024 * 1024)
.set_compression(datafusion::parquet::basic::Compression::ZSTD(Default::default()))
.build();
let mut encoded = Vec::new();
let mut writer = ArrowWriter::try_new(&mut encoded, schema, Some(properties)).expect("writer");
writer.write(&batch).expect("compressed value");
writer.close().expect("footer");
let original = Bytes::from(encoded);
let builder = ParquetRecordBatchReaderBuilder::try_new(original.clone()).expect("valid compressed fixture");
let metadata = builder.metadata();
let groups = metadata
.row_groups()
.iter()
.map(|group| {
let columns = group
.columns()
.iter()
.cloned()
.map(|column| {
column
.into_builder()
.set_total_uncompressed_size(64)
.build()
.expect("forged column")
})
.collect();
group
.clone()
.into_builder()
.set_column_metadata(columns)
.build()
.expect("forged group")
})
.collect();
let forged = metadata.as_ref().clone().into_builder().set_row_groups(groups).build();
let footer_size = u32::from_le_bytes(
original[original.len() - 8..original.len() - 4]
.try_into()
.expect("footer size"),
);
let mut bytes = original[..original.len() - 8 - usize::try_from(footer_size).expect("footer length")].to_vec();
ParquetMetaDataWriter::new(&mut bytes, &forged)
.finish()
.expect("forged footer");
assert!(bytes.len() < 256 * 1024, "compressed input remains small");
admit_parquet_decode(&forged, None, 4 * 1024 * 1024).expect("footer-only admission misses the page size");
let store = Arc::new(InMemory::new());
let context = parquet_session_with_memory(store.clone(), 16 * 1024 * 1024);
store
.put(&Path::from("forged.parquet"), Bytes::from(bytes).into())
.await
.expect("fixture");
let result = ParquetSelectTable::try_new(&context.state(), &parquet_input("forged.parquet"), None).await;
let error = match result {
Ok(_) => panic!("actual query admission must reject oversized page buffers before decoding"),
Err(error) => error,
};
assert_eq!(error.select_error(), SelectError::ResourceExhausted);
}
#[tokio::test]
async fn parquet_legacy_lz4_checks_native_scratch_before_query_decode() {
use datafusion::parquet::{basic::Compression, file::metadata::ParquetMetaDataWriter};
let schema = Arc::new(Schema::new(vec![Field::new("v", DataType::Int32, false)]));
let batch =
RecordBatch::try_new(Arc::clone(&schema), vec![Arc::new(datafusion::arrow::array::Int32Array::from(vec![42]))])
.expect("one integer row");
let properties = datafusion::parquet::file::properties::WriterProperties::builder()
.set_dictionary_enabled(false)
.build();
let mut original = Vec::new();
let mut writer = ArrowWriter::try_new(&mut original, schema, Some(properties)).expect("writer");
writer.write(&batch).expect("row");
writer.close().expect("footer");
let metadata = ParquetRecordBatchReaderBuilder::try_new(Bytes::from(original))
.expect("metadata")
.metadata()
.clone();
// A valid legacy frame encoding the four little-endian bytes for 42.
let payload = [2, 33, 76, 24, 5, 0, 0, 0, 0x40, 42, 0, 0, 0];
let mut page = vec![0x15, 0, 0x15, 8, 0x15, 26, 0x2c, 0x15, 2, 0x15, 0, 0x15, 6, 0x15, 6, 0, 0];
page.extend(payload);
let column = metadata
.row_group(0)
.column(0)
.clone()
.into_builder()
.set_compression(Compression::LZ4)
.set_data_page_offset(4)
.set_total_compressed_size(i64::try_from(page.len()).expect("small page"))
.build()
.expect("legacy column");
let group = metadata
.row_group(0)
.clone()
.into_builder()
.set_column_metadata(vec![column])
.build()
.expect("row group");
let metadata = metadata.as_ref().clone().into_builder().set_row_groups(vec![group]).build();
let mut encoded = b"PAR1".to_vec();
encoded.extend(page);
ParquetMetaDataWriter::new(&mut encoded, &metadata)
.finish()
.expect("legacy footer");
assert!(encoded.len() < 4096);
let mut native = ParquetRecordBatchReaderBuilder::try_new(Bytes::from(encoded.clone()))
.expect("legacy metadata")
.build()
.expect("native reader");
let decoded = native.next().expect("one batch").expect("legacy fallback decode");
assert_eq!(
decoded
.column(0)
.as_primitive::<datafusion::arrow::datatypes::Int32Type>()
.value(0),
42
);
let store = Arc::new(InMemory::new());
store
.put(&Path::from("legacy.parquet"), Bytes::from(encoded).into())
.await
.expect("fixture");
let tight = parquet_session_with_memory(store.clone(), 2 * 1024 * 1024);
let result = ParquetSelectTable::try_new(&tight.state(), &parquet_input("legacy.parquet"), None).await;
let error = match result {
Ok(_) => panic!("legacy decoder scratch exceeds the query budget"),
Err(error) => error,
};
assert_eq!(error.select_error(), SelectError::ResourceExhausted);
let roomy = parquet_session_with_memory(store, 128 * 1024 * 1024);
let table = ParquetSelectTable::try_new(&roomy.state(), &parquet_input("legacy.parquet"), None)
.await
.expect("admitted legacy file");
roomy.register_table("source", table).expect("table");
let rows = roomy
.sql("SELECT v FROM source")
.await
.expect("plan")
.collect()
.await
.expect("query");
assert_eq!(rows.iter().map(RecordBatch::num_rows).sum::<usize>(), 1);
assert_eq!(
rows[0]
.column(0)
.as_primitive::<datafusion::arrow::datatypes::Int32Type>()
.value(0),
42
);
}
#[tokio::test]
async fn parquet_nested_views_decode_through_the_actual_file_schema() {
use datafusion::arrow::array::{ListArray, StringArray};
+15 -6
View File
@@ -230,6 +230,17 @@ impl DynamicSleeper {
SCANNER_IDLE_MODE.load(Ordering::Relaxed)
}
fn foreground_backoff(&self) -> Duration {
// Local walkers already use disk-priority admission and a shared I/O
// budget. A global per-operation floor duplicates that control and
// throttles idle disks because unrelated foreground work is active.
if self.disk_io_budget.is_some() {
Duration::ZERO
} else {
foreground_workload_backoff_duration(crate::workload_admission::foreground_workload_activity())
}
}
fn read_params(&self) -> (f64, Duration) {
let config = self.read_config();
(config.factor, config.max_sleep)
@@ -252,15 +263,14 @@ impl DynamicSleeper {
}
let (factor, max_sleep) = self.read_params();
if factor == 0.0 || max_sleep.is_zero() {
let foreground_sleep =
foreground_workload_backoff_duration(crate::workload_admission::foreground_workload_activity());
let foreground_sleep = self.foreground_backoff();
if !foreground_sleep.is_zero() {
tokio::time::sleep(foreground_sleep).await;
global_metrics().record_scanner_throttle_sleep(foreground_sleep);
}
return;
}
let foreground_sleep = foreground_workload_backoff_duration(crate::workload_admission::foreground_workload_activity());
let foreground_sleep = self.foreground_backoff();
let sleep_dur = Duration::from_secs_f64(MIN_SLEEP.as_secs_f64() * factor)
.min(max_sleep)
.max(foreground_sleep);
@@ -354,8 +364,7 @@ impl SleepTimer {
}
let (factor, max_sleep) = self.sleeper.read_params();
if factor == 0.0 || max_sleep.is_zero() {
let foreground_sleep =
foreground_workload_backoff_duration(crate::workload_admission::foreground_workload_activity());
let foreground_sleep = self.sleeper.foreground_backoff();
if !foreground_sleep.is_zero() {
tokio::time::sleep(foreground_sleep).await;
global_metrics().record_scanner_throttle_sleep(foreground_sleep);
@@ -363,7 +372,7 @@ impl SleepTimer {
return;
}
let elapsed = self.start.elapsed();
let foreground_sleep = foreground_workload_backoff_duration(crate::workload_admission::foreground_workload_activity());
let foreground_sleep = self.sleeper.foreground_backoff();
let sleep_dur = Duration::from_secs_f64(elapsed.as_secs_f64() * factor)
.max(MIN_SLEEP)
.min(max_sleep)
+1 -1
View File
@@ -135,7 +135,7 @@ impl DynamicSleeper {
if let Some(elapsed) = elapsed {
debt.charge(proportional_delay(elapsed, config), Instant::now());
}
let foreground = foreground_workload_backoff_duration(crate::workload_admission::foreground_workload_activity());
let foreground = self.foreground_backoff();
let now = Instant::now();
let folder_reaches_deadline = folder
&& debt.started.is_some_and(|start| {
@@ -181,6 +181,54 @@ async fn active_wait_wakes_for_fastest_idle_off_and_smaller_max_wait() {
}
}
#[tokio::test(start_paused = true)]
#[serial]
async fn disk_bound_pacing_does_not_repeat_global_foreground_backoff() {
let _idle = IdleGuard::enabled();
struct ForegroundGuard;
impl Drop for ForegroundGuard {
fn drop(&mut self) {
crate::reset_foreground_read_activity_for_test();
}
}
let _foreground = ForegroundGuard;
crate::set_foreground_read_activity(25);
let disk = tempfile::tempdir().expect("disk identity");
let sleeper = DynamicSleeper::new(ScannerSpeed::Default).for_disk_with_budget_for_tests(
disk.path(),
ScannerDiskIoBudgetLimits {
iops: Some(0),
bytes_per_sec: None,
},
);
let mut debt = PacingDebt::default();
let begin = Instant::now();
for _ in 0..100 {
sleeper.pace_object(&mut debt, Duration::from_micros(20)).await;
}
sleeper.flush_pacing(&mut debt).await;
assert_eq!(begin.elapsed(), Duration::from_millis(4));
assert_eq!(debt.charged_objects, 100);
assert_eq!(debt.pending, Duration::ZERO);
// The first operation consumes the token bucket's one-operation burst.
sleeper.admit_disk_io(1, 0).await;
let begin = Instant::now();
sleeper.sleep_folder_paced(&mut debt).await;
assert_eq!(
begin.elapsed(),
SCANNER_DISK_IO_BUDGET_MAX_WAIT + Duration::from_millis(2),
"the per-disk budget and preset folder gap must still apply"
);
assert_eq!(
scanner_disk_io_budget_status()
.into_iter()
.find(|status| status.disk == disk.path().to_string_lossy())
.expect("disk charges remain visible")
.charged_ops,
2
);
}
#[tokio::test(start_paused = true)]
#[serial]
async fn foreground_backoff_flushes_tail_without_double_charging_or_max_wait_cap() {
+28
View File
@@ -591,6 +591,34 @@ pub trait AuthoritativeListOperations: ListOperations {
max_keys: i32,
) -> Result<Self::ListObjectVersionsInfo, Self::Error>;
/// Enumerate an exact dotted leaf below identifier-like directory components
/// containing ASCII letters, digits, `_` or `-`. Noncandidate object metadata
/// is outside the scan. Namespace components may occur at any depth; reserved
/// words do not identify a subtree that can be skipped.
///
/// Returned keys retain the quorum, unversioned, EOF and cancellation
/// guarantees of `list_unversioned_objects_authoritative`. Older peers may
/// perform extra I/O; the coordinator must filter their extra rows before
/// resolving metadata. Callers fence concurrent mutations separately.
///
/// # Errors
/// Rejects an invalid directory prefix/leaf, invalid page size, uncertain
/// matching metadata, versioned entries or insufficient read quorum. Backends
/// without this capability return Unsupported; ordinary listing is not a fallback.
async fn list_identifier_tree_leaves_authoritative(
self: Arc<Self>,
_bucket: &str,
_prefix: &str,
_leaf: &str,
_marker: Option<String>,
_max_keys: i32,
) -> Result<Self::ListObjectVersionsInfo, Self::Error>
where
Self::Error: From<std::io::Error>,
{
Err(std::io::Error::new(std::io::ErrorKind::Unsupported, "authoritative identifier-tree listing is not supported").into())
}
/// Resolves every set at its version read quorum, including reserved prefixes.
///
/// # Errors
+4 -13
View File
@@ -45,18 +45,9 @@ In a distributed deployment, startup publishes `FullReady` only after that local
- Liveness reports process availability and must not depend on storage, IAM,
lock quorum, or peer health.
- Node readiness (`/health/ready`, `/minio/health/ready`) reports whether this process can serve reads: storage read quorum, IAM, and shared-lock quorum. It is the Kubernetes Service membership signal. Losing write quorum, the pool-metadata writer, or exclusive-lock quorum does not remove a node that can still read. The public health handler still returns 503 until startup publishes `FullReady`, so a process that has never opened S3 admission does not join the Service. Runtime storage diagnostics do not change that startup gate.
- A blocked pool metadata writer degrades cluster-write readiness with
`pool_meta_write_blocked`. It does not withdraw `/health/ready`, and the node
probe omits that reason from `degradedReasons`. Metadata save-gate inspection
is bounded to 100 ms; contention reports `pool_metadata_check_timeout` on the
cluster-write probe without installing a block. The node probe keeps the last
confirmed writable observation for `details.poolMetadata` during a transient
inspection timeout, while preserving a confirmed block and failing that
detail closed when no fresh observation exists.
- The authenticated cluster snapshot extends its existing node-local metadata
gate inspection with safe reason, failure phase, and original block time. It
distinguishes timeout from a block and changes no admission or recovery
decision. Runtime readiness and gate status are separate bounded observations.
- Immutable pool metadata admission proofs share the local authority gate as readers. Publication, recovery, identity changes and operations requiring serialization retain exclusive ownership and their existing distributed namespace fences. A multi-phase operation releases its reader before acquiring the writer and revalidates under exclusive ownership; it never upgrades a held reader. The gate is fair, so a writer queued behind a reader that is waiting on namespace I/O also delays later admissions until that reader finishes. Every writer publishes the authority it leaves before releasing exclusive ownership. Health inspection reads that published authority without acquiring the gate, so neither a held nor a queued writer delays it. A writer's changes become visible when it releases the gate; a latch recorded by a transaction outcome after its writer released is shared with the published authority and visible immediately. Mutating paths still check authority under the gate.
- A blocked pool metadata writer degrades cluster-write readiness with `pool_meta_write_blocked`. It does not withdraw `/health/ready`, and the node probe omits that reason from `degradedReasons`. Metadata authority inspection reads the state published by the last local writer and does not wait for the save gate, so a held or queued metadata writer does not degrade either probe; a block that writer installs appears when it releases the gate.
- The authenticated cluster snapshot extends its existing node-local metadata gate inspection with safe reason, failure phase, and original block time. It changes no admission or recovery decision. Runtime readiness and gate status are separate observations.
- Cluster write readiness (`/minio/health/cluster`) requires storage write quorum, the pool-metadata write gate, and exclusive-lock quorum. Use it when a caller must know that new writes can commit. It is not the Service membership probe.
- Cluster read readiness uses the storage read-quorum path and cluster-health timeout behavior. Its lock dependency still uses the per-set majority/write-lock health check. Actual shared namespace locks require `n - n/2`, so the cluster read probe is conservative: a four-client set can still admit reads with two clients while that probe returns 503. Node `/health/ready` uses the shared-lock threshold instead, because that is what object reads acquire.
- `HEAD` health probes keep header/status semantics and do not require response
@@ -91,7 +82,7 @@ For a healthy IAM and metadata writer in a four-node, one-drive-per-node EC 2+2
With two of four nodes up, GET, HEAD, and List of a bucket that this process has not yet cached still succeed. PUT and other mutations fail because write quorum and exclusive locks are gone.
The node path reads local disk-handle health and reuses the same reachable-host observation as its lock dependency, including the existing `RUSTFS_HEALTH_READINESS_CACHE_TTL_MS` cache. Only `Online` drives count; a reachable host with a `Returning` drive does not yet prove data I/O has recovered. It does not call cluster `storage_info`, local `disk_info`, or add disk-info RPCs. The entire storage inventory snapshot has a separate 100 ms wait budget; expiry reports `storage_readiness_check_timeout` and fails closed. Pool metadata inspection retains its own 100 ms budget; a timeout is counted by `rustfs_pool_metadata_check_timeouts_total` and uses the last confirmed node-local gate state only within the same cache TTL. These observations are not an atomic cluster snapshot and do not bypass the existing lock-probe timing or cache policy.
The node path reads local disk-handle health and reuses the same reachable-host observation as its lock dependency, including the existing `RUSTFS_HEALTH_READINESS_CACHE_TTL_MS` cache. Only `Online` drives count; a reachable host with a `Returning` drive does not yet prove data I/O has recovered. It does not call cluster `storage_info`, local `disk_info`, or add disk-info RPCs. The entire storage inventory snapshot has a separate 100 ms wait budget; expiry reports `storage_readiness_check_timeout` and fails closed. Pool metadata inspection reads the authority published by the last local writer and does not wait for the save gate. These observations are not an atomic cluster snapshot and do not bypass the existing lock-probe timing or cache policy.
The new fields are additive. Their absence in an older response is not evidence of storage quorum. Minimal responses still contain only the existing top-level fields, liveness remains dependency-independent, and HEAD responses remain bodyless.
@@ -26,7 +26,7 @@ RustFS S3 Tables is an Iceberg REST Catalog and table-bucket implementation on t
| `/_iceberg/v1` | Supported compatibility alias | MinIO AIStor-style alias; smoke profile defaults to signing name `s3tables`. |
| S3 object data plane | Supported | Data, metadata, manifest, and delete files are ordinary S3 objects with table-aware policy checks on warehouse paths. |
| S3 table-bucket listings | Supported | `ListObjects`, `ListObjectsV2`, `ListObjectVersions`, metadata-list extensions, and multipart-upload listings omit warehouse objects unless the caller has `admin:GetTableMetadata` for that table, and always omit internal `.rustfs-table` keys. Protected cursors are encrypted with a key derived from the cluster root credentials, so rotating those credentials invalidates cursors already issued. |
| Object-backed warehouse-index recovery | Supported with bounded request fallback | Data-plane requests repair or verify a missing warehouse-prefix index by listing the catalog prefix. Only table entries count toward the 4,096-object limit, so maintenance reports and other per-table objects cannot exhaust it; the listing itself is bounded to 64 pages plus one page per table entry already listed, and after each page the cursor skips the rest of the maintenance subtree the page ended in, so a report flood under one table costs a bounded number of pages. Larger catalogs fail closed with a retryable service error and require operator backfill. An optional table-entry directory (see Maintenance Matrix) answers these lookups without the prefix listing. |
| Object-backed warehouse-index recovery | Supported with bounded request fallback | Data-plane requests repair or verify a missing warehouse-prefix index using authoritative table-entry pages. The scan selects `table-entry.json` below identifier-like directory components at any depth; namespace components named `tables` or `maintenance` remain valid. Noncandidate report names are filtered in directory batches before inventory retention and metadata reads, so report history does not consume ownership pages. Selected entries retain GET-equivalent quorum and corruption checks. More than 4,096 table entries fails closed with a retryable service error and requires operator backfill. An optional table-entry directory (see Maintenance Matrix) answers these lookups without the prefix listing. |
| Table bucket enablement | Supported | A regular bucket is enabled for catalog use and addressed as the REST catalog warehouse. |
| Catalog-vended table credentials | Automated when enabled | Disabled by default. LoadTable vends credentials only when `X-Iceberg-Access-Delegation` contains the exact `vended-credentials` token; the dedicated credentials endpoint uses the same issuer path. |
| AWS S3 Tables endpoint shape | Profile generator | Generates the AWS catalog URI and warehouse ARN shape for migration docs. API parity not claimed. |
@@ -132,6 +132,8 @@ Object-backed commit publication identifies every catalog object it validated by
An optional per-bucket directory of table entries under `.rustfs.sys` answers data-plane index-miss lookups without listing the catalog prefix. It is off by default: `RUSTFS_TABLE_CATALOG_DIRECTORY_INDEX=true` and `RUSTFS_TABLE_CATALOG_DIRECTORY_INDEX_FLEET_CONFIRMED=true` on every node, then `POST /iceberg/v1/{warehouse}/catalog/directory` (`admin:MigrateTableCatalog`) advances a backfill of 128 catalog objects per call until the directory reports `Ready`; `GET` reports the phase and `DELETE` rolls the directory back. Entry writers add a directory candidate only while the directory exists, and the backfill covers earlier entries. The directory state occupies the backing-migration fence object: nodes without this feature answer catalog mutations on that bucket with a parse error until they are upgraded, which is what the fleet-confirmed flag attests, and restores require disabling the directory first. Object catalog backups exclude the derived directory state and candidates. Re-enable the directory after a restore to rebuild it from the restored entries.
Fallback ownership scans, including the check before metadata cleanup, use the optional `identifier_tree_leaf` walk selector. An absent selector preserves ordinary listing behavior. Older peers ignore the additive field; the coordinator filters their noncandidate rows before metadata resolution and retains read-quorum checks for every selected entry. Old peers may do more I/O or fail on unrelated corrupt metadata, so a mixed-version scan can refuse work until those peers are upgraded. Such failures never establish owner absence. This selector changes no catalog or on-disk format and requires no fleet-confirmation flag.
| Capability | Status | Claim |
+1 -1
View File
@@ -77,7 +77,7 @@ An unreadable replica, lost fence, or conditional-write conflict leaves the orig
- The first block emits `decommission_state` with `state=pool_metadata_blocked`, `reason`, `phase`, and `blocked_since`. A change in recovery failure classification emits `state=pool_metadata_recovery_pending`; successful recovery emits `state=pool_metadata_recovered` with the original timestamp.
- `rustfs_pool_metadata_blocks_total{reason}` and `rustfs_pool_metadata_recoveries_total` count block and recovery transitions. The original cause and phase remain attached to local typed errors; storage/RPC error numbers and on-disk formats are unchanged.
- Node and cluster-write readiness include `pool_meta_write_blocked`. Waiting for the metadata save mutex is bounded to 100 ms and reports `pool_metadata_check_timeout`, not a persistent block. Cluster probes retain their existing cache and overall timeout behavior. Liveness and cluster-read quorum checks are unchanged. The authenticated node-local status and safe error fields are described in [S3 write failure diagnostics](s3-write-failure-diagnostics.md).
- Cluster-write readiness reports `pool_meta_write_blocked`; node readiness reports the writer only through `details.poolMetadata`. Readiness reads the authority published by the last local metadata writer instead of waiting for the save gate, so an active or queued metadata transaction does not degrade readiness by itself. A block installed by a writer appears when that writer releases the gate. Cluster probes retain their existing cache and overall timeout behavior. Liveness and cluster-read quorum checks are unchanged. The authenticated node-local status and safe error fields are described in [S3 write failure diagnostics](s3-write-failure-diagnostics.md).
Pool metadata CAS explicitly selects `TailDrainedWithOwnerRepair`: it waits for the complete rename result and returns repair responsibility to its existing deployment and transaction owner when write quorum commits but a replica fails or signatures diverge. Cache invalidation and the committed cleanup handoff precede that error. Ordinary system metadata retains quorum-success semantics under `Quorum` and `TailDrained`; draining a tail does not require every disk to succeed. A repair-required completion does not enter rollback cleanup or wait for another temporary-data deletion. The committed bytes are not rolled back. Generic MRF and heal-channel requests cannot supply the deployment and transaction authority needed to repair system metadata; pool CAS retains the uncertain outcome and its existing recovery fence. Ordinary object repair keeps its bucket-incarnation fencing.
+1 -1
View File
@@ -97,7 +97,7 @@ When the whole cluster (or several nodes) went down and nodes are brought back o
2. Logs say what the node waits for. The IAM recovery loop retries with backoff and logs `event="iam_bootstrap_retry_failed"` with an actionable `hint` field (for example, "storage read quorum not met yet; waiting for enough cluster nodes/disks to come online"). After repeated failures the level escalates from WARN to ERROR; this still does not kill the process.
3. Recovery is automatic. Once storage read quorum is available, pending nodes can finish IAM bootstrap on the next retry. `/health/ready` returns `200` when startup has published `FullReady` and the node can still serve reads (storage read quorum, IAM, and shared-lock quorum). Write quorum is reported separately and is required by `/minio/health/cluster`, not by Service membership. Restarting pending nodes does not speed this up.
4. Check readiness detail while waiting. `/health/ready` (and `/minio/health/ready`) separate `storage` / `poolMetadata` / `iam` / `lock` readiness. `storage.ready` follows read quorum (`readinessScope` is `read_quorum`); `storage.readQuorum` and `storage.writeQuorum` show the two observations. `poolMetadata.ready` is the metadata writer and does not by itself withdraw `/health/ready`. `degradedReasons` on this probe names read-path causes such as `storage_quorum_unavailable` and `storage_and_lock_unavailable`. `pool_meta_write_blocked` and `pool_metadata_check_timeout` stay on `/minio/health/cluster`. See the [storage detail contract](../architecture/readiness-matrix.md#storage-detail-contract) for probe scopes and sampling limits:
4. Check readiness detail while waiting. `/health/ready` (and `/minio/health/ready`) separate `storage` / `poolMetadata` / `iam` / `lock` readiness. `storage.ready` follows read quorum (`readinessScope` is `read_quorum`); `storage.readQuorum` and `storage.writeQuorum` show the two observations. `poolMetadata.ready` is the metadata writer and does not by itself withdraw `/health/ready`. `degradedReasons` on this probe names read-path causes such as `storage_quorum_unavailable` and `storage_and_lock_unavailable`. `pool_meta_write_blocked` stays on `/minio/health/cluster`. See the [storage detail contract](../architecture/readiness-matrix.md#storage-detail-contract) for probe scopes and sampling limits:
```bash
curl -s http://<node>:9000/health/ready | jq
@@ -45,15 +45,14 @@ The probe round timeout is configured independently; see [Admin peer probe timeo
## Correlate bounded diagnostics
Normal operation does not require success logs at WARN. Request counters remain available with WARN logging, while runtime readiness diagnostics distinguish `pool_meta_write_blocked`, `pool_metadata_check_timeout`, and insufficient storage quorum. Do not clear a metadata write fence merely to make readiness green.
Normal operation does not require success logs at WARN. Request counters remain available with WARN logging, while runtime readiness diagnostics distinguish `pool_meta_write_blocked` from insufficient storage quorum. Do not clear a metadata write fence merely to make readiness green.
Query the authenticated `/rustfs/admin/v4/cluster/snapshot` endpoint on the affected node. `snapshot.pool_meta_write_gate` describes that node's metadata writer, not a fleet-wide aggregate. Its existing booleans are preserved; `state` and optional block details extend the same bounded, read-only gate inspection. This starts no recovery, disk reads, or additional RPCs. Runtime readiness and the gate are inspected separately, so a whole cluster snapshot is not atomic across sections.
Query the authenticated `/rustfs/admin/v4/cluster/snapshot` endpoint on the affected node. `snapshot.pool_meta_write_gate` describes that node's metadata writer, not a fleet-wide aggregate. Its existing booleans are preserved; `state` and optional block details extend the same read-only inspection, which reads the authority published by the node's last metadata writer without waiting for the save gate. A block that an active writer installs appears when that writer releases the gate. Nodes running earlier releases can also report `check_timeout`, the `pool_metadata_check_timeout` readiness reason, and `rustfs_pool_metadata_check_timeouts_total` when their 100 ms gate inspection expired; current nodes emit none of them. This starts no recovery, disk reads, or additional RPCs. Runtime readiness and the gate are inspected separately, so a whole cluster snapshot is not atomic across sections.
| `state` | Meaning |
| --- | --- |
| `writable` | No metadata write block was observed. Other dependencies may still make the node unready. |
| `blocked` | A metadata write block was observed; `reason`, `phase`, and `sinceUnixSecs` describe its typed failure context. |
| `check_timeout` | The existing 100 ms inspection budget expired. Readiness remains false, but a write block is not asserted. |
| `unavailable` | The object store or a recognized metadata observation was unavailable; no block details are invented. |
For `blocked`, `reason` reuses the metadata failure classification, `phase` identifies the operation stage that caused the block (not live recovery-worker progress), and `sinceUnixSecs` is the original block time in Unix seconds. Polling does not reset that time. Recovery replaces the observation with `writable` and removes the previous block details. An unavailable store reports `writesReady=false` with `state="unavailable"`. Older responses may omit `state` and block details; missing fields alone do not prove a healthy writer. Operation text, raw replica errors, disk paths, and credentials are excluded. See [Pool metadata recovery](pool-metadata-recovery.md) for recovery and escalation boundaries.
+2 -2
View File
@@ -40,7 +40,7 @@ Bitrot cycle resolution differs because the canonical persistent key belongs to
| `scanner.cycle_max_objects` | `RUSTFS_SCANNER_CYCLE_MAX_OBJECTS` | objects | `0` (`DEFAULT_SCANNER_CYCLE_MAX_OBJECTS`) | Caps objects processed by one cycle. `0` disables this budget. |
| `scanner.cycle_max_directories` | `RUSTFS_SCANNER_CYCLE_MAX_DIRECTORIES` | directories | `0` (`DEFAULT_SCANNER_CYCLE_MAX_DIRECTORIES`) | Caps directories entered by one cycle. `0` disables this budget. |
| `heal.bitrot_cycle` | `RUSTFS_SCANNER_BITROT_CYCLE_SECS` | seconds | `2592000` (`DEFAULT_HEAL_BITROT_CYCLE_SECS`, 30 days) | Controls periodic deep bitrot scans. `false`, `off`, `no`, or `disabled` disables periodic deep scans; `0`, `true`, `on`, or `yes` runs deep mode every scanner cycle. |
| `scanner.idle_mode` | `RUSTFS_SCANNER_IDLE_MODE` | boolean | `true` (`DEFAULT_SCANNER_IDLE_MODE`) | Enables preset scanner sleeps and foreground-read backoff. `false` disables those sleeps; per-disk foreground-priority admission remains active. |
| `scanner.idle_mode` | `RUSTFS_SCANNER_IDLE_MODE` | boolean | `true` (`DEFAULT_SCANNER_IDLE_MODE`) | Enables preset scanner sleeps and the foreground backoff for callers without a disk binding. `false` disables those sleeps; per-disk foreground-priority admission remains active. |
| `scanner.cache_save_timeout` | `RUSTFS_SCANNER_CACHE_SAVE_TIMEOUT_SECS` | seconds | `14` (`DEFAULT_SCANNER_CACHE_SAVE_TIMEOUT_SECS`) | Timeout for saving scanner cache; runtime enforces a minimum of `1` and keeps the default persistence budget within the distributed publication lease. |
| `scanner.recovery_backoff_cap` | `RUSTFS_SCANNER_RECOVERY_BACKOFF_CAP_SECS` | seconds | `600` (`DEFAULT_SCANNER_RECOVERY_BACKOFF_CAP_SECS`) | Caps the wait between automatic startup recovery retries. Must be a positive integer; `0` and malformed values are rejected, so retries never stop. Applies to the next computed retry, not to a wait already in progress. The retry count resets after 24 hours, which bounds a single wait at about one day even with a larger cap. |
| `scanner.max_concurrent_set_scans` | `RUSTFS_SCANNER_MAX_CONCURRENT_SET_SCANS` | count | `4` (`DEFAULT_SCANNER_MAX_CONCURRENT_SET_SCANS`) | Caps concurrent set-level scanner tasks. `0` keeps topology-derived concurrency. |
@@ -63,7 +63,7 @@ Speed presets (`crates/config/src/constants/scanner.rs`) set the base sleep mult
| `slow` | 10x | 15s | 60s |
| `slowest` | 100x | 15s | 30m |
Use `scanner.delay`, `scanner.max_wait`, and `scanner.cycle` when the preset is close but one axis needs a precise override. With `idle_mode=true`, directory-level sleep is `1ms x factor` and object-level sleep is `time spent on the object x factor`, both capped at `max_wait`; a foreground-read floor of `FOREGROUND_READ_BACKOFF_PER_REQUEST_MS` (10ms) per concurrent GetObject/streaming read, capped at `FOREGROUND_READ_BACKOFF_MAX_MS` (250ms), is applied on top and can exceed the preset's `max_wait` (`crates/scanner/src/sleeper.rs`). The per-disk foreground-priority gate is independent of `idle_mode` and remains active when sleeps are disabled.
Use `scanner.delay`, `scanner.max_wait`, and `scanner.cycle` when the preset is close but one axis needs a precise override. With `idle_mode=true`, directory-level sleep is `1ms x factor` and object-level sleep is admitted work time multiplied by the factor, capped at `max_wait`; object pacing batches small waits and settles remaining debt at bucket completion. Local walkers bound with `DynamicSleeper::for_disk` use disk-priority admission and their shared I/O budget without an additional global foreground-request sleep. A caller without that binding retains a foreground-workload floor of 10ms per active request, capped at 250ms, which can exceed `max_wait` (`crates/scanner/src/sleeper.rs`). The per-disk foreground-priority gate and disk budget remain active when preset sleeps are disabled. Removing redundant global backoff does not guarantee a fixed snapshot age: scans still compete for admitted resources and may checkpoint at their cycle budget.
### Startup and cycle budgets
Generated
+6 -6
View File
@@ -2,11 +2,11 @@
"nodes": {
"nixpkgs": {
"locked": {
"lastModified": 1793466162,
"narHash": "sha256-pt+peAEoOK3lLLdtbnQ1AGIBTZbdB6jrkgmurfcQwig=",
"lastModified": 1793977579,
"narHash": "sha256-WrSODdZFVIs/Hh2NzC0bY+ROByi0bATg8WfePX8dPj8=",
"owner": "NixOS",
"repo": "nixpkgs",
"rev": "a0788123ada57ec92a79ab2581cda6d3c6cca75f",
"rev": "4fe25e12eb9f4c744966dd3de5a05246cf098601",
"type": "github"
},
"original": {
@@ -29,11 +29,11 @@
]
},
"locked": {
"lastModified": 1793434469,
"narHash": "sha256-+sgVNI2c67di+XIeMaz5KSei4T65kX68s5Nl0E5Gzbc=",
"lastModified": 1794040562,
"narHash": "sha256-7eEiezSku3lbZPUOXW8we1a7+xd5axkCQdyI3rzOKjo=",
"owner": "oxalica",
"repo": "rust-overlay",
"rev": "6ecf534a7c16e7c969f26838fdd8c93597ae93f3",
"rev": "d3fe74269f2a6652437e21e78eb7e83be81f6982",
"type": "github"
},
"original": {
+1 -12
View File
@@ -208,9 +208,7 @@ pub(crate) struct ClusterPoolMetaWriteGateView {
impl From<ClusterPoolMetaWriteGateSnapshot> for ClusterPoolMetaWriteGateView {
fn from(snapshot: ClusterPoolMetaWriteGateSnapshot) -> Self {
let state = if snapshot.check_timed_out {
"check_timeout"
} else if snapshot.writes_ready {
let state = if snapshot.writes_ready {
"writable"
} else if snapshot.write_blocked || snapshot.transaction_aborted {
"blocked"
@@ -703,7 +701,6 @@ fn summarize_storage_readiness(snapshot: &ClusterReadOnlySnapshot) -> Capability
.filter_map(|reason| match reason {
ReadinessDegradedReason::StorageQuorumUnavailable
| ReadinessDegradedReason::PoolMetaWriteBlocked
| ReadinessDegradedReason::PoolMetadataCheckTimeout
| ReadinessDegradedReason::StorageAndIamUnavailable
| ReadinessDegradedReason::StorageAndLockUnavailable
| ReadinessDegradedReason::StorageIamAndLockUnavailable => Some(reason.as_str()),
@@ -1112,14 +1109,6 @@ mod tests {
for (snapshot, state) in [
(ClusterPoolMetaWriteGateSnapshot::default(), "writable"),
(
ClusterPoolMetaWriteGateSnapshot {
writes_ready: false,
check_timed_out: true,
..Default::default()
},
"check_timeout",
),
(
ClusterPoolMetaWriteGateSnapshot {
writes_ready: false,
+10 -3
View File
@@ -1882,9 +1882,16 @@ impl DefaultObjectUsecase {
reader = write_plan.apply(reader, actual_size).map_err(ApiError::from)?;
if let Some(control) = progress_control {
use rustfs_rio::HashReaderMut;
let inner = reader.take_inner();
reader.inner = rustfs_rio::boxed_reader(super::request_body::DemandReader::new(inner, control));
if !should_compress
&& (use_zero_copy_eager_put_path || use_empty_or_small_eager_put_path)
&& trailing_headers.is_none()
{
control.finish();
} else {
use rustfs_rio::HashReaderMut;
let inner = reader.take_inner();
reader.inner = rustfs_rio::boxed_reader(super::request_body::DemandReader::new(inner, control));
}
}
rustfs_io_metrics::record_put_object_stage_duration_from("app_encryption_prepare", encryption_stage_start);
+1 -1
View File
@@ -160,7 +160,7 @@ impl BodyReadControl {
state.waited = Duration::ZERO;
}
fn finish(&self) {
pub(super) fn finish(&self) {
let mut state = self.0.budget.lock();
self.0.active.store(false, Ordering::Release);
state.finished = true;
-3
View File
@@ -108,7 +108,6 @@ pub struct ClusterListingDiagnosticsSnapshot {
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ClusterPoolMetaWriteGateSnapshot {
pub writes_ready: bool,
pub check_timed_out: bool,
pub write_blocked: bool,
pub transaction_aborted: bool,
pub pool_meta_absent: bool,
@@ -124,7 +123,6 @@ impl Default for ClusterPoolMetaWriteGateSnapshot {
fn default() -> Self {
Self {
writes_ready: true,
check_timed_out: false,
write_blocked: false,
transaction_aborted: false,
pool_meta_absent: false,
@@ -230,7 +228,6 @@ async fn current_pool_meta_write_gate_snapshot() -> ClusterPoolMetaWriteGateSnap
let status = store.pool_meta_write_gate_status().await;
ClusterPoolMetaWriteGateSnapshot {
writes_ready: status.writes_ready,
check_timed_out: status.check_timed_out,
write_blocked: status.write_blocked,
transaction_aborted: status.transaction_aborted,
pool_meta_absent: status.pool_meta_absent,
+65 -3
View File
@@ -28,6 +28,19 @@ const MAX_VERSIONS_EXCEEDED_MESSAGE: &str = "You've exceeded the limit on the nu
const SLOW_DOWN_READ_CODE: &str = "SlowDownRead";
const SLOW_DOWN_READ_MESSAGE: &str = "Resource requested is unreadable, please reduce your request rate";
const NAMESPACE_LOCK_TIMEOUT_CAUSE: &str = "namespace_lock_timeout";
const RPC_ADMISSION_METADATA_KEY: &str = "x-rustfs-rpc-admission";
const RPC_REPLAY_ADMISSION_METADATA_VALUE: &str = "replay-cache";
/// Preserve temporary replay admission refusal through the RPC and S3 boundaries.
pub(crate) fn rpc_replay_admission_exhausted_status() -> tonic::Status {
let mut status = tonic::Status::resource_exhausted("RPC replay admission exhausted");
status.metadata_mut().insert(
RPC_ADMISSION_METADATA_KEY,
tonic::metadata::MetadataValue::from_static(RPC_REPLAY_ADMISSION_METADATA_VALUE),
);
status
}
/// Public message for `StorageError::UnsupportedDisk` on a request path. The
/// only request-time producer is the delete-transaction protocol when a
/// participant answers Unimplemented (a peer still on a release without the
@@ -588,9 +601,14 @@ fn body_read_error_code(err: &(dyn std::error::Error + 'static)) -> Option<S3Err
return Some(S3ErrorCode::BadDigest);
}
rpc_unavailable |= error
.downcast_ref::<tonic::Status>()
.is_some_and(|status| status.code() == tonic::Code::Unavailable);
rpc_unavailable |= error.downcast_ref::<tonic::Status>().is_some_and(|status| {
status.code() == tonic::Code::Unavailable
|| (status.code() == tonic::Code::ResourceExhausted
&& status
.metadata()
.get(RPC_ADMISSION_METADATA_KEY)
.is_some_and(|value| value.as_encoded_bytes() == RPC_REPLAY_ADMISSION_METADATA_VALUE.as_bytes()))
});
upload_limit_exceeded |= error.is::<UploadLimitExceeded>();
server_source_failed |= error.is::<ServerSideSourceReadError>();
client_body_timed_out |= error.is::<ClientBodyReadTimeout>();
@@ -1752,6 +1770,50 @@ mod tests {
}
}
#[test]
fn replay_admission_refusal_preserves_retryable_status_across_io_boundaries() {
let status = rpc_replay_admission_exhausted_status;
let original = StorageError::from(status());
let cloned = original.clone();
let wire = status().into_http::<()>();
let received = tonic::Status::from_header_map(wire.headers()).expect("replay admission status on the wire");
for error in [
ApiError::from(original),
ApiError::from(cloned),
ApiError::from(StorageError::from(received)),
ApiError::from(StorageError::Io(IoError::other(IoError::other(status())))),
ApiError::from(IoError::other(status())),
ApiError::from(IoError::other(StorageError::from(status()))),
] {
assert_eq!(error.code, S3ErrorCode::ServiceUnavailable);
assert_eq!(error.diagnostic().rpc_code, Some(tonic::Code::ResourceExhausted));
assert_eq!(S3Error::from(error).status_code(), Some(StatusCode::SERVICE_UNAVAILABLE));
}
}
#[test]
fn replay_admission_marker_requires_its_exact_value_and_resource_code() {
for code in [
tonic::Code::Unauthenticated,
tonic::Code::Internal,
tonic::Code::ResourceExhausted,
] {
for marker in ["replay-cache", "unknown-admission"] {
if code == tonic::Code::ResourceExhausted && marker == "replay-cache" {
continue;
}
let mut status = tonic::Status::new(code, "RPC replay admission exhausted");
status
.metadata_mut()
.insert(RPC_ADMISSION_METADATA_KEY, marker.parse().expect("valid test marker"));
let error = ApiError::from(StorageError::from(status));
assert_eq!(error.code, S3ErrorCode::InternalError);
}
}
let error = ApiError::from(IoError::other("RPC replay admission exhausted"));
assert_eq!(error.code, S3ErrorCode::InternalError);
}
#[test]
fn rpc_unavailable_mapping_does_not_match_other_codes_or_untyped_messages() {
for code in [
-16
View File
@@ -614,22 +614,6 @@ mod tests {
});
}
#[test]
#[serial]
fn writable_node_readiness_keeps_200_for_transient_pool_metadata_timeout() {
let mut report = ready_report();
report.degraded_reasons = vec![ReadinessDegradedReason::PoolMetadataCheckTimeout];
with_var(rustfs_config::ENV_HEALTH_MINIMAL_RESPONSE_ENABLE, Some("false"), || {
let parts = build_health_response_parts(Method::GET, HealthProbe::Readiness, Some(&report), "rustfs", None, None);
assert_eq!(parts.status_code, StatusCode::OK);
let payload = parts.payload.expect("GET readiness body");
assert_eq!(payload["ready"], true);
assert_eq!(payload["degradedReasons"], json!(["pool_metadata_check_timeout"]));
});
}
#[tokio::test]
async fn readiness_collects_object_stalls_and_recovers_on_completion() {
let object_traffic_health = ObjectTrafficHealth::enabled_for_test(Duration::ZERO);
+47 -2
View File
@@ -3411,6 +3411,16 @@ fn hyper_body_write_close_kind(debug: &str) -> Option<&'static str> {
}
}
fn rpc_auth_failure_status(error: &std::io::Error) -> Status {
if storage::rpc_auth_is_resource_exhausted(error) {
crate::error::rpc_replay_admission_exhausted_status()
} else if storage::tonic_rpc_auth_failure_reason(error) == "stale_boot_epoch" {
Status::unavailable("RPC boot epoch changed")
} else {
Status::unauthenticated("No valid auth token")
}
}
#[allow(clippy::result_large_err)]
fn check_auth(req: Request<()>) -> std::result::Result<Request<()>, Status> {
let local_node_name =
@@ -3473,6 +3483,12 @@ fn check_auth(req: Request<()>) -> std::result::Result<Request<()>, Status> {
INTERNODE_TRANSPORT_BACKEND_GRPC,
failure_reason,
);
let rejection = rpc_auth_failure_status(&e);
if rejection.code() == tonic::Code::ResourceExhausted {
// Existing scoped counters describe overload without producing an
// authentication error log for every otherwise valid request.
return rejection;
}
if failure_reason == "stale_boot_epoch" {
// The signature is valid, but the peer restarted before this request. Reject it
// before execution and let the client retry after its authenticated epoch refresh.
@@ -3491,7 +3507,7 @@ fn check_auth(req: Request<()>) -> std::result::Result<Request<()>, Status> {
error = %e,
"RPC boot epoch changed; request rejected for retry"
);
Status::unavailable("RPC boot epoch changed")
rejection
} else {
error!(
event = EVENT_RPC_SIGNATURE_VERIFICATION_FAILED,
@@ -3508,7 +3524,7 @@ fn check_auth(req: Request<()>) -> std::result::Result<Request<()>, Status> {
error = %e,
"RPC signature verification failed"
);
Status::unauthenticated("No valid auth token")
rejection
}
})?;
@@ -7007,6 +7023,35 @@ mod tests {
.await;
}
#[test]
fn rpc_auth_resource_exhaustion_does_not_report_invalid_credentials() {
for source in [
storage::RpcReplayCacheAdmissionError::Capacity,
storage::RpcReplayCacheAdmissionError::MemoryBudget,
] {
let error = std::io::Error::new(std::io::ErrorKind::WouldBlock, source);
let status = rpc_auth_failure_status(&error);
assert_eq!(status.code(), tonic::Code::ResourceExhausted);
let api = crate::error::ApiError::from(crate::storage_api::error::StorageError::from(status));
let response = s3s::S3Error::from(api);
assert_eq!(response.status_code(), Some(http::StatusCode::SERVICE_UNAVAILABLE));
}
for message in [
"RPC request replay detected",
"Invalid RPC v2 signature",
"Missing signature header",
] {
assert_eq!(
rpc_auth_failure_status(&std::io::Error::other(message)).code(),
tonic::Code::Unauthenticated
);
}
assert_eq!(
rpc_auth_failure_status(&std::io::Error::other("RPC boot epoch is stale")).code(),
tonic::Code::Unavailable
);
}
#[tokio::test]
#[serial_test::serial]
async fn rpc_auth_rejection_records_failure_reason_metric() {
+51 -295
View File
@@ -37,7 +37,7 @@ use std::task::{Context, Poll};
use std::time::Duration;
use std::{
collections::{HashMap, HashSet},
sync::{Mutex as StdMutex, OnceLock},
sync::OnceLock,
time::Instant,
};
use tokio::sync::Mutex;
@@ -80,7 +80,6 @@ const LOG_COMPONENT_SERVER: &str = "server";
const LOG_SUBSYSTEM_READINESS: &str = "readiness";
const METRIC_RUNTIME_READINESS_READY: &str = "rustfs_runtime_readiness_ready";
const METRIC_RUNTIME_READINESS_DEGRADED_TOTAL: &str = "rustfs_runtime_readiness_degraded_total";
const METRIC_POOL_METADATA_CHECK_TIMEOUT_TOTAL: &str = "rustfs_pool_metadata_check_timeouts_total";
pub use crate::shared_types::{
DependencyReadiness, DependencyReadinessReport, MissingReadinessDrive, ReadinessDegradedReason, StorageReadinessDetails,
@@ -293,66 +292,6 @@ struct StorageWriteReadinessStatus {
pool_metadata_reason: Option<ReadinessDegradedReason>,
}
#[derive(Debug, Clone, Copy, Default)]
struct NodePoolMetadataReadinessCache {
entry: Option<NodePoolMetadataReadinessCacheEntry>,
}
#[derive(Debug, Clone, Copy)]
struct NodePoolMetadataReadinessCacheEntry {
cached_at: Instant,
observed_at: Instant,
status: StorageWriteReadinessStatus,
}
impl NodePoolMetadataReadinessCache {
fn observe(
&mut self,
observed: StorageWriteReadinessStatus,
observed_at: Instant,
now: Instant,
ttl: Duration,
) -> StorageWriteReadinessStatus {
if ttl.is_zero() {
self.entry = None;
return observed;
}
if observed.pool_metadata_reason == Some(ReadinessDegradedReason::PoolMetadataCheckTimeout) {
let Some(previous) = self.entry else {
return observed;
};
if now.saturating_duration_since(previous.cached_at) > ttl {
return observed;
}
// A confirmed write block must never be hidden by a later timeout.
if previous.status.pool_metadata_reason == Some(ReadinessDegradedReason::PoolMetaWriteBlocked) {
return previous.status;
}
// A transient inspection outage is diagnostic, not proof that the
// writer became unsafe. Keep the last confirmed writable state.
if previous.status.ready {
return StorageWriteReadinessStatus {
ready: true,
pool_metadata_reason: Some(ReadinessDegradedReason::PoolMetadataCheckTimeout),
};
}
return observed;
}
if self.entry.is_none_or(|entry| entry.observed_at <= observed_at) {
self.entry = Some(NodePoolMetadataReadinessCacheEntry {
cached_at: now,
observed_at,
status: observed,
});
}
observed
}
}
fn pool_metadata_write_readiness(result: Result<(), StorageError>) -> StorageWriteReadinessStatus {
match result {
Ok(()) => StorageWriteReadinessStatus {
@@ -360,13 +299,9 @@ fn pool_metadata_write_readiness(result: Result<(), StorageError>) -> StorageWri
pool_metadata_reason: None,
},
Err(error) => {
let pool_metadata_reason = if error.pool_metadata_failure().is_some() {
Some(ReadinessDegradedReason::PoolMetaWriteBlocked)
} else if matches!(error, StorageError::Timeout) {
Some(ReadinessDegradedReason::PoolMetadataCheckTimeout)
} else {
None
};
let pool_metadata_reason = error
.pool_metadata_failure()
.map(|_| ReadinessDegradedReason::PoolMetaWriteBlocked);
StorageWriteReadinessStatus {
ready: false,
pool_metadata_reason,
@@ -434,11 +369,6 @@ fn health_cluster_timeout() -> Duration {
)
}
fn node_pool_metadata_readiness_cache() -> &'static StdMutex<NodePoolMetadataReadinessCache> {
static CACHE: OnceLock<StdMutex<NodePoolMetadataReadinessCache>> = OnceLock::new();
CACHE.get_or_init(|| StdMutex::new(NodePoolMetadataReadinessCache::default()))
}
fn storage_readiness_cache() -> &'static Mutex<Option<StorageReadinessCacheEntry>> {
static CACHE: OnceLock<Mutex<Option<StorageReadinessCacheEntry>>> = OnceLock::new();
CACHE.get_or_init(|| Mutex::new(None))
@@ -543,26 +473,6 @@ async fn update_storage_readiness_cache(status: StorageWriteReadinessStatus) {
});
}
fn apply_node_pool_metadata_timeout_policy(
observed: StorageWriteReadinessStatus,
observed_at: Instant,
) -> StorageWriteReadinessStatus {
if observed.pool_metadata_reason == Some(ReadinessDegradedReason::PoolMetadataCheckTimeout) {
counter!(METRIC_POOL_METADATA_CHECK_TIMEOUT_TOTAL).increment(1);
}
let mut cache = node_pool_metadata_readiness_cache()
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
cache.observe(observed, observed_at, Instant::now(), health_readiness_cache_ttl())
}
#[cfg(test)]
fn reset_node_pool_metadata_readiness_cache() {
*node_pool_metadata_readiness_cache()
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner) = NodePoolMetadataReadinessCache::default();
}
async fn load_cached_lock_quorum_status() -> Option<LockQuorumObservation> {
let ttl = health_readiness_cache_ttl();
if ttl.is_zero() {
@@ -855,9 +765,6 @@ fn record_readiness_report(report: &DependencyReadinessReport) {
&& report.readiness.peer_health_ready;
gauge!(METRIC_RUNTIME_READINESS_READY).set(if ready { 1.0 } else { 0.0 });
for reason in &report.degraded_reasons {
if ready && *reason == ReadinessDegradedReason::PoolMetadataCheckTimeout {
continue;
}
counter!(METRIC_RUNTIME_READINESS_DEGRADED_TOTAL, "reason" => reason.as_str()).increment(1);
}
}
@@ -944,7 +851,7 @@ pub async fn collect_node_readiness_report() -> DependencyReadinessReport {
let mut storage_check_timed_out = false;
if let Some(store) = runtime_sources::current_object_store_handle() {
let pool_metadata_status = store.pool_meta_write_status().await;
storage = apply_node_pool_metadata_timeout_policy(pool_metadata_write_readiness(pool_metadata_status), Instant::now());
storage = pool_metadata_write_readiness(pool_metadata_status);
match node_storage_snapshot(store.as_ref(), &lock_observation.online_hosts).await {
Ok((info, unavailable_drives, missing_drives)) => {
details.unavailable_drives = unavailable_drives;
@@ -2927,7 +2834,7 @@ mod tests {
}
#[test]
fn pool_metadata_observation_distinguishes_block_timeout_and_unavailable() {
fn pool_metadata_observation_distinguishes_block_and_unavailable() {
let blocked = blocked_pool_metadata_status();
assert!(!blocked.ready);
assert_eq!(blocked.pool_metadata_reason, Some(ReadinessDegradedReason::PoolMetaWriteBlocked));
@@ -2935,227 +2842,76 @@ mod tests {
let writable = pool_metadata_write_readiness(Ok(()));
assert!(writable.ready);
assert_eq!(writable.pool_metadata_reason, None);
let timed_out = pool_metadata_write_readiness(Err(StorageError::Timeout));
assert!(!timed_out.ready);
assert_eq!(timed_out.pool_metadata_reason, Some(ReadinessDegradedReason::PoolMetadataCheckTimeout));
let unavailable = pool_metadata_write_readiness(Err(StorageError::other("pool metadata writes remain blocked")));
assert!(!unavailable.ready);
assert_eq!(
unavailable.pool_metadata_reason, None,
"error text alone must not be interpreted as a typed write block"
);
}
let readiness = DependencyReadiness {
storage_ready: false,
#[test]
fn readiness_report_records_ready_gauge_and_degraded_reasons() {
let record = |report: DependencyReadinessReport| {
let recorder = DebuggingRecorder::new();
let snapshotter = recorder.snapshotter();
metrics::with_local_recorder(&recorder, || record_readiness_report(&report));
snapshotter.snapshot().into_vec()
};
let ready = DependencyReadiness {
storage_ready: true,
iam_ready: true,
lock_quorum_ready: true,
peer_health_ready: true,
};
let report = dependency_readiness_report_from_write_status(readiness, timed_out);
assert!(!report.readiness.storage_ready, "inspection timeout must remain fail-closed");
assert_eq!(report.degraded_reasons, vec![ReadinessDegradedReason::PoolMetadataCheckTimeout]);
assert_eq!(report.degraded_reasons[0].as_str(), "pool_metadata_check_timeout");
}
#[test]
#[serial]
fn sticky_pool_metadata_timeout_uses_dedicated_metric_without_degraded_count() {
reset_node_pool_metadata_readiness_cache();
let recorder = DebuggingRecorder::new();
let snapshotter = recorder.snapshotter();
metrics::with_local_recorder(&recorder, || {
with_var(rustfs_config::ENV_HEALTH_READINESS_CACHE_TTL_MS, Some("60000"), || {
let writable = StorageWriteReadinessStatus {
ready: true,
pool_metadata_reason: None,
};
assert_eq!(apply_node_pool_metadata_timeout_policy(writable, Instant::now()), writable);
let sticky = apply_node_pool_metadata_timeout_policy(
StorageWriteReadinessStatus {
ready: false,
pool_metadata_reason: Some(ReadinessDegradedReason::PoolMetadataCheckTimeout),
},
Instant::now(),
);
assert!(sticky.ready);
record_readiness_report(&DependencyReadinessReport {
readiness: DependencyReadiness {
storage_ready: true,
iam_ready: true,
lock_quorum_ready: true,
peer_health_ready: true,
},
degraded_reasons: vec![ReadinessDegradedReason::PoolMetadataCheckTimeout],
storage_details: None,
});
});
let entries = record(DependencyReadinessReport {
readiness: ready,
degraded_reasons: Vec::new(),
storage_details: None,
});
reset_node_pool_metadata_readiness_cache();
let entries = snapshotter.snapshot().into_vec();
let ready = entries.iter().find_map(|(composite, _, _, value)| {
let gauge = entries.iter().find_map(|(composite, _, _, value)| {
(composite.kind() == MetricKind::Gauge && composite.key().name() == METRIC_RUNTIME_READINESS_READY).then_some(value)
});
assert!(matches!(ready, Some(DebugValue::Gauge(value)) if value.into_inner() == 1.0));
assert!(
matches!(gauge, Some(DebugValue::Gauge(value)) if value.into_inner() == 1.0),
"a ready report must set the readiness gauge: {gauge:?}"
);
assert!(
entries
.iter()
.all(|(composite, _, _, _)| composite.key().name() != METRIC_RUNTIME_READINESS_DEGRADED_TOTAL),
"a ready report without reasons must not count degradation"
);
let entries = record(DependencyReadinessReport {
readiness: DependencyReadiness {
storage_ready: false,
..ready
},
degraded_reasons: vec![ReadinessDegradedReason::PoolMetaWriteBlocked],
storage_details: None,
});
let gauge = entries.iter().find_map(|(composite, _, _, value)| {
(composite.kind() == MetricKind::Gauge && composite.key().name() == METRIC_RUNTIME_READINESS_READY).then_some(value)
});
assert!(
matches!(gauge, Some(DebugValue::Gauge(value)) if value.into_inner() == 0.0),
"a degraded report must clear the readiness gauge: {gauge:?}"
);
let degraded = entries.iter().find_map(|(composite, _, _, value)| {
(composite.kind() == MetricKind::Counter
&& composite.key().name() == METRIC_RUNTIME_READINESS_DEGRADED_TOTAL
&& composite
.key()
.labels()
.any(|label| label.key() == "reason" && label.value() == "pool_metadata_check_timeout"))
.any(|label| label.key() == "reason" && label.value() == "pool_meta_write_blocked"))
.then_some(value)
});
assert!(degraded.is_none());
let timeout = entries.iter().find_map(|(composite, _, _, value)| {
(composite.kind() == MetricKind::Counter && composite.key().name() == METRIC_POOL_METADATA_CHECK_TIMEOUT_TOTAL)
.then_some(value)
});
assert!(matches!(timeout, Some(DebugValue::Counter(1))));
}
#[tokio::test]
#[serial]
async fn node_pool_metadata_timeout_preserves_recent_writable_observation() {
reset_node_pool_metadata_readiness_cache();
async_with_vars([(rustfs_config::ENV_HEALTH_READINESS_CACHE_TTL_MS, Some("60000"))], async {
let writable = StorageWriteReadinessStatus {
ready: true,
pool_metadata_reason: None,
};
assert_eq!(apply_node_pool_metadata_timeout_policy(writable, Instant::now()), writable);
let timed_out = apply_node_pool_metadata_timeout_policy(
StorageWriteReadinessStatus {
ready: false,
pool_metadata_reason: Some(ReadinessDegradedReason::PoolMetadataCheckTimeout),
},
Instant::now(),
);
assert!(timed_out.ready);
assert_eq!(timed_out.pool_metadata_reason, Some(ReadinessDegradedReason::PoolMetadataCheckTimeout));
})
.await;
reset_node_pool_metadata_readiness_cache();
}
#[tokio::test]
#[serial]
async fn node_pool_metadata_timeout_fails_closed_with_empty_cache() {
reset_node_pool_metadata_readiness_cache();
async_with_vars([(rustfs_config::ENV_HEALTH_READINESS_CACHE_TTL_MS, Some("60000"))], async {
let timed_out = apply_node_pool_metadata_timeout_policy(
StorageWriteReadinessStatus {
ready: false,
pool_metadata_reason: Some(ReadinessDegradedReason::PoolMetadataCheckTimeout),
},
Instant::now(),
);
assert!(!timed_out.ready);
assert_eq!(timed_out.pool_metadata_reason, Some(ReadinessDegradedReason::PoolMetadataCheckTimeout));
})
.await;
reset_node_pool_metadata_readiness_cache();
}
#[test]
fn node_pool_metadata_timeout_fails_closed_after_cache_ttl() {
let mut cache = NodePoolMetadataReadinessCache::default();
let observed_at = Instant::now();
let ttl = Duration::from_secs(60);
let writable = StorageWriteReadinessStatus {
ready: true,
pool_metadata_reason: None,
};
assert_eq!(cache.observe(writable, observed_at, observed_at, ttl), writable);
let timed_out = cache.observe(
StorageWriteReadinessStatus {
ready: false,
pool_metadata_reason: Some(ReadinessDegradedReason::PoolMetadataCheckTimeout),
},
observed_at + Duration::from_secs(61),
observed_at + Duration::from_secs(61),
ttl,
assert!(
matches!(degraded, Some(DebugValue::Counter(1))),
"each degraded reason must be counted once: {degraded:?}"
);
assert!(!timed_out.ready);
assert_eq!(timed_out.pool_metadata_reason, Some(ReadinessDegradedReason::PoolMetadataCheckTimeout));
}
#[tokio::test]
#[serial]
async fn node_pool_metadata_timeout_never_hides_confirmed_block() {
reset_node_pool_metadata_readiness_cache();
async_with_vars([(rustfs_config::ENV_HEALTH_READINESS_CACHE_TTL_MS, Some("60000"))], async {
let blocked = StorageWriteReadinessStatus {
ready: false,
pool_metadata_reason: Some(ReadinessDegradedReason::PoolMetaWriteBlocked),
};
let newer_observation = Instant::now();
assert_eq!(apply_node_pool_metadata_timeout_policy(blocked, newer_observation), blocked);
assert_eq!(
apply_node_pool_metadata_timeout_policy(
StorageWriteReadinessStatus {
ready: true,
pool_metadata_reason: None,
},
newer_observation
.checked_sub(Duration::from_secs(1))
.expect("test instant has elapsed time"),
),
StorageWriteReadinessStatus {
ready: true,
pool_metadata_reason: None,
}
);
let timed_out = apply_node_pool_metadata_timeout_policy(
StorageWriteReadinessStatus {
ready: false,
pool_metadata_reason: Some(ReadinessDegradedReason::PoolMetadataCheckTimeout),
},
Instant::now(),
);
assert_eq!(timed_out, blocked);
})
.await;
reset_node_pool_metadata_readiness_cache();
}
#[tokio::test]
#[serial]
async fn node_pool_metadata_timeout_fails_closed_without_fresh_observation() {
reset_node_pool_metadata_readiness_cache();
async_with_vars([(rustfs_config::ENV_HEALTH_READINESS_CACHE_TTL_MS, Some("60000"))], async {
let writable = StorageWriteReadinessStatus {
ready: true,
pool_metadata_reason: None,
};
assert_eq!(apply_node_pool_metadata_timeout_policy(writable, Instant::now()), writable);
async_with_vars([(rustfs_config::ENV_HEALTH_READINESS_CACHE_TTL_MS, Some("0"))], async {
let timed_out = apply_node_pool_metadata_timeout_policy(
StorageWriteReadinessStatus {
ready: false,
pool_metadata_reason: Some(ReadinessDegradedReason::PoolMetadataCheckTimeout),
},
Instant::now(),
);
assert!(!timed_out.ready);
assert_eq!(timed_out.pool_metadata_reason, Some(ReadinessDegradedReason::PoolMetadataCheckTimeout));
})
.await;
})
.await;
reset_node_pool_metadata_readiness_cache();
}
#[test]
-2
View File
@@ -48,7 +48,6 @@ pub enum ReadinessDegradedReason {
ObjectReadStalled,
ObjectWriteStalled,
PoolMetaWriteBlocked,
PoolMetadataCheckTimeout,
StorageReadinessCheckTimeout,
ClusterHealthTimeout,
PeerHealthUnavailable,
@@ -69,7 +68,6 @@ impl ReadinessDegradedReason {
ReadinessDegradedReason::ObjectReadStalled => "object_read_stalled",
ReadinessDegradedReason::ObjectWriteStalled => "object_write_stalled",
ReadinessDegradedReason::PoolMetaWriteBlocked => "pool_meta_write_blocked",
ReadinessDegradedReason::PoolMetadataCheckTimeout => "pool_metadata_check_timeout",
ReadinessDegradedReason::StorageReadinessCheckTimeout => "storage_readiness_check_timeout",
ReadinessDegradedReason::ClusterHealthTimeout => "cluster_health_timeout",
ReadinessDegradedReason::PeerHealthUnavailable => "peer_health_unavailable",
+3
View File
@@ -1667,6 +1667,9 @@ async fn handle_put_file(req: Request<Incoming>, require_auth: bool) -> Response
INTERNODE_TRANSPORT_BACKEND_TCP_HTTP,
)
{
if crate::storage::storage_api::rpc_auth_is_resource_exhausted(&e) {
return response_with_status(StatusCode::SERVICE_UNAVAILABLE, "RPC replay admission exhausted");
}
log_internode_rpc_response_failure!(
StatusCode::FORBIDDEN,
&path,
+7 -3
View File
@@ -622,7 +622,7 @@ pub(crate) mod ecstore_rpc {
KMS_SIGNAL_SUBSYSTEM, LocalPeerS3Client, MAX_NETWORK_PROBE_BYTES, MAX_NETWORK_PROBE_DURATION, NetworkPeerProbeClient,
NetworkPeerProbeError, PEER_RESTDRY_RUN, PEER_RESTSIGNAL, PEER_RESTSUB_SYS, PeerRestClient, PeerS3Client,
SERVICE_SIGNAL_REFRESH_CONFIG, SERVICE_SIGNAL_RELOAD_DYNAMIC, TONIC_RPC_PREFIX, check_and_record_signed_rpc_nonce,
decode_heal_bucket_rpc_options, normalize_tonic_rpc_audience, rpc_background_admission,
decode_heal_bucket_rpc_options, normalize_tonic_rpc_audience, rpc_auth_is_resource_exhausted, rpc_background_admission,
sign_ns_scanner_capability_with_checkpoint_coverage, sign_put_file_capability, sign_tonic_rpc_response_proof,
tonic_boot_epoch_challenge, tonic_boot_epoch_response_headers, tonic_rpc_auth_failure_reason,
verify_put_file_auth_trailer, verify_rpc_signature, verify_tonic_canonical_body_digest,
@@ -631,8 +631,8 @@ pub(crate) mod ecstore_rpc {
};
#[cfg(test)]
pub(crate) use rustfs_ecstore::api::rpc::{
ScannerScopedDirtyUsageAckEntry, build_put_file_auth_trailer, gen_signature_headers, gen_tonic_replay_scope_headers,
gen_tonic_signature_headers, set_tonic_canonical_body_digest, verify_put_file_capability,
RpcReplayCacheAdmissionError, ScannerScopedDirtyUsageAckEntry, build_put_file_auth_trailer, gen_signature_headers,
gen_tonic_replay_scope_headers, gen_tonic_signature_headers, set_tonic_canonical_body_digest, verify_put_file_capability,
verify_tonic_boot_epoch_response, verify_tonic_rpc_response_proof,
};
}
@@ -2458,6 +2458,10 @@ pub(crate) fn verify_tonic_rpc_signature_with_bootstrap(
ecstore_rpc::verify_tonic_rpc_signature_with_bootstrap(audience, path, headers, allow_replay_scope_bootstrap)
}
pub(crate) fn rpc_auth_is_resource_exhausted(error: &std::io::Error) -> bool {
ecstore_rpc::rpc_auth_is_resource_exhausted(error)
}
pub(crate) fn tonic_rpc_auth_failure_reason(error: &std::io::Error) -> &'static str {
ecstore_rpc::tonic_rpc_auth_failure_reason(error)
}
+6 -3
View File
@@ -188,10 +188,13 @@ pub(crate) mod server {
pub(crate) use crate::storage::storage_api::ecstore_disk::{
DiskAPI, DiskOption, new_disk, with_background_resource_unit,
};
#[cfg(test)]
pub(crate) use crate::storage::storage_api::ecstore_rpc::RpcReplayCacheAdmissionError;
pub(crate) use crate::storage::storage_api::{
RemoteBackgroundIo, ServerContextSlot, TONIC_RPC_PREFIX, normalize_tonic_rpc_audience, rpc_background_admission,
tonic_boot_epoch_challenge, tonic_boot_epoch_response_headers, tonic_rpc_auth_failure_reason,
verify_tonic_rpc_signature_with_bootstrap, with_remote_background_io,
RemoteBackgroundIo, ServerContextSlot, TONIC_RPC_PREFIX, normalize_tonic_rpc_audience,
rpc_auth_is_resource_exhausted, rpc_background_admission, tonic_boot_epoch_challenge,
tonic_boot_epoch_response_headers, tonic_rpc_auth_failure_reason, verify_tonic_rpc_signature_with_bootstrap,
with_remote_background_io,
};
pub(crate) fn try_current_local_node_name() -> Option<String> {
@@ -21,8 +21,7 @@ use datafusion::arrow::datatypes::{DataType, Field, Schema};
use datafusion::parquet::arrow::{
add_encoded_arrow_schema_to_metadata, arrow_reader::ArrowReaderOptions, arrow_writer::ArrowWriterOptions,
};
use datafusion::parquet::file::FOOTER_SIZE;
use datafusion::parquet::file::metadata::{ColumnChunkMetaData, FooterTail, RowGroupMetaData};
use datafusion::parquet::file::metadata::ColumnChunkMetaData;
use datafusion::parquet::file::properties::{EnabledStatistics, WriterProperties};
use std::io::Write;
use std::sync::OnceLock;
@@ -74,279 +73,54 @@ fn add_budget(total: &mut usize, amount: usize, limit: usize) -> TableCatalogSto
Ok(())
}
/// Sizes read from one Parquet page header (Thrift compact encoding).
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
struct PageHeaderSizes {
uncompressed: usize,
compressed: usize,
}
#[cfg(test)]
const THRIFT_COMPACT_STOP: u8 = 0;
#[cfg(test)]
const THRIFT_COMPACT_BOOL_TRUE: u8 = 1;
const THRIFT_COMPACT_BOOL_FALSE: u8 = 2;
const THRIFT_COMPACT_BYTE: u8 = 3;
#[cfg(test)]
const THRIFT_COMPACT_I16: u8 = 4;
#[cfg(test)]
const THRIFT_COMPACT_I32: u8 = 5;
#[cfg(test)]
const THRIFT_COMPACT_I64: u8 = 6;
const THRIFT_COMPACT_DOUBLE: u8 = 7;
#[cfg(test)]
const THRIFT_COMPACT_BINARY: u8 = 8;
#[cfg(test)]
const THRIFT_COMPACT_LIST: u8 = 9;
const THRIFT_COMPACT_SET: u8 = 10;
#[cfg(test)]
const THRIFT_COMPACT_MAP: u8 = 11;
#[cfg(test)]
const THRIFT_COMPACT_STRUCT: u8 = 12;
const THRIFT_MAX_NESTING: usize = 16;
const PAGE_HEADER_FIELD_UNCOMPRESSED_SIZE: i16 = 2;
const PAGE_HEADER_FIELD_COMPRESSED_SIZE: i16 = 3;
const FILE_METADATA_FIELD_ROW_GROUPS: i16 = 4;
/// Minimal Thrift compact-protocol cursor: enough to read a `PageHeader`'s
/// size fields and skip everything else so the next header can be located.
struct ThriftCompactCursor<'a> {
bytes: &'a [u8],
pos: usize,
}
impl ThriftCompactCursor<'_> {
fn error() -> TableCatalogStoreError {
TableCatalogStoreError::Invalid("compaction parquet thrift metadata is malformed".to_owned())
}
fn byte(&mut self) -> TableCatalogStoreResult<u8> {
let byte = *self.bytes.get(self.pos).ok_or_else(Self::error)?;
self.pos += 1;
Ok(byte)
}
fn skip_bytes(&mut self, count: usize) -> TableCatalogStoreResult<()> {
self.pos = self
.pos
.checked_add(count)
.filter(|end| *end <= self.bytes.len())
.ok_or_else(Self::error)?;
Ok(())
}
fn varint(&mut self) -> TableCatalogStoreResult<u64> {
let mut value = 0_u64;
for shift in (0..64).step_by(7) {
let byte = self.byte()?;
value |= u64::from(byte & 0x7f) << shift;
if byte & 0x80 == 0 {
return Ok(value);
}
}
Err(Self::error())
}
fn zigzag(&mut self) -> TableCatalogStoreResult<i64> {
let raw = self.varint()?;
Ok(((raw >> 1) as i64) ^ -((raw & 1) as i64))
}
/// Every collection element occupies at least one byte, so a declared
/// count beyond the remaining bytes is malformed. This bounds the walk to
/// the chunk's size instead of the count a hostile header declares.
fn collection_count(&self, count: u64) -> TableCatalogStoreResult<usize> {
usize::try_from(count)
.ok()
.filter(|count| *count <= self.bytes.len() - self.pos)
.ok_or_else(Self::error)
}
fn list_header(&mut self) -> TableCatalogStoreResult<(u8, usize)> {
let header = self.byte()?;
let count = match header >> 4 {
15 => self.varint()?,
short => u64::from(short),
};
Ok((header & 0x0f, self.collection_count(count)?))
}
fn field_id(&mut self, last: i16, header: u8) -> TableCatalogStoreResult<i16> {
match header >> 4 {
0 => i16::try_from(self.zigzag()?).map_err(|_| Self::error()),
delta => last.checked_add(i16::from(delta)).ok_or_else(Self::error),
}
}
/// parquet reads the page size fields by id as a zigzag varint whatever
/// the wire kind says. Follow the id; a kind that is not an integer is
/// rejected rather than skipped, so the walker never records a smaller
/// size than the decoder will allocate.
fn size_field(&mut self, kind: u8) -> TableCatalogStoreResult<usize> {
match kind {
THRIFT_COMPACT_I16 | THRIFT_COMPACT_I32 | THRIFT_COMPACT_I64 => {
usize::try_from(self.zigzag()?).map_err(|_| Self::error())
}
_ => Err(Self::error()),
}
}
fn skip_value(&mut self, kind: u8, depth: usize) -> TableCatalogStoreResult<()> {
if depth > THRIFT_MAX_NESTING {
return Err(Self::error());
}
match kind {
// A struct field's bool value lives in its header byte.
THRIFT_COMPACT_BOOL_TRUE | THRIFT_COMPACT_BOOL_FALSE => Ok(()),
THRIFT_COMPACT_BYTE => self.skip_bytes(1),
THRIFT_COMPACT_I16 | THRIFT_COMPACT_I32 | THRIFT_COMPACT_I64 => self.varint().map(drop),
THRIFT_COMPACT_DOUBLE => self.skip_bytes(8),
THRIFT_COMPACT_BINARY => {
let len = usize::try_from(self.varint()?).map_err(|_| Self::error())?;
self.skip_bytes(len)
}
THRIFT_COMPACT_LIST | THRIFT_COMPACT_SET => {
let (element, count) = self.list_header()?;
for _ in 0..count {
self.skip_element(element, depth + 1)?;
}
Ok(())
}
THRIFT_COMPACT_MAP => {
let count = self.varint()?;
let count = self.collection_count(count)?;
if count == 0 {
return Ok(());
}
let kinds = self.byte()?;
for _ in 0..count {
self.skip_element(kinds >> 4, depth + 1)?;
self.skip_element(kinds & 0x0f, depth + 1)?;
}
Ok(())
}
THRIFT_COMPACT_STRUCT => self.skip_struct(depth + 1),
_ => Err(Self::error()),
}
}
/// A collection element carries its own bytes, but parquet skips a bool
/// element without consuming any. Either width would let the walker and
/// the decoder disagree on where the next header starts; no Parquet
/// struct holds a bool collection, so fail closed instead.
fn skip_element(&mut self, kind: u8, depth: usize) -> TableCatalogStoreResult<()> {
match kind {
THRIFT_COMPACT_BOOL_TRUE | THRIFT_COMPACT_BOOL_FALSE => Err(Self::error()),
kind => self.skip_value(kind, depth),
}
}
fn skip_struct(&mut self, depth: usize) -> TableCatalogStoreResult<()> {
let mut field_id = 0_i16;
loop {
let header = self.byte()?;
if header == THRIFT_COMPACT_STOP {
return Ok(());
}
field_id = self.field_id(field_id, header)?;
self.skip_value(header & 0x0f, depth)?;
}
}
/// Read a `PageHeader` struct, returning its sizes and leaving the cursor at
/// the first page byte.
fn page_header(&mut self) -> TableCatalogStoreResult<PageHeaderSizes> {
let mut sizes = PageHeaderSizes::default();
let mut field_id = 0_i16;
loop {
let header = self.byte()?;
if header == THRIFT_COMPACT_STOP {
return Ok(sizes);
}
let kind = header & 0x0f;
field_id = self.field_id(field_id, header)?;
match field_id {
PAGE_HEADER_FIELD_UNCOMPRESSED_SIZE => sizes.uncompressed = self.size_field(kind)?,
PAGE_HEADER_FIELD_COMPRESSED_SIZE => sizes.compressed = self.size_field(kind)?,
_ => self.skip_value(kind, 1)?,
}
}
}
/// Walk the footer's `FileMetaData`. The pinned parquet reserves the
/// declared `row_groups` capacity before reading one element; charge that
/// reservation to the budget so a small footer cannot request gigabytes.
fn file_metadata(&mut self, limit: usize) -> TableCatalogStoreResult<()> {
let mut field_id = 0_i16;
let mut reserved = 0_usize;
loop {
let header = self.byte()?;
if header == THRIFT_COMPACT_STOP {
return Ok(());
}
let kind = header & 0x0f;
field_id = self.field_id(field_id, header)?;
if field_id == FILE_METADATA_FIELD_ROW_GROUPS && kind == THRIFT_COMPACT_LIST {
let (element, count) = self.list_header()?;
if element != THRIFT_COMPACT_STRUCT {
return Err(Self::error());
}
let reservation = count.checked_mul(size_of::<RowGroupMetaData>()).ok_or_else(Self::error)?;
add_budget(&mut reserved, reservation, limit)?;
for _ in 0..count {
self.skip_struct(2)?;
}
} else {
self.skip_value(kind, 0)?;
}
}
fn parquet_admission_error(error: rustfs_s3select_api::parquet_budget::ParquetBudgetError) -> TableCatalogStoreError {
match error {
rustfs_s3select_api::parquet_budget::ParquetBudgetError::Exceeded => budget_error(),
error => TableCatalogStoreError::Invalid(format!("compaction {error}")),
}
}
/// Reject footer collection counts the decoder would allocate for before it
/// reads a single element. Bytes after the struct are ignored, as parquet does.
fn validate_footer_metadata(data: &Bytes, limit: usize) -> TableCatalogStoreResult<()> {
let error = || TableCatalogStoreError::Invalid("compaction parquet footer is malformed".to_owned());
let tail_start = data.len().checked_sub(FOOTER_SIZE).ok_or_else(error)?;
let tail: &[u8; FOOTER_SIZE] = data[tail_start..].try_into().map_err(|_| error())?;
let tail = FooterTail::try_new(tail).map_err(|_| error())?;
let metadata_start = tail_start.checked_sub(tail.metadata_length()).ok_or_else(error)?;
let mut cursor = ThriftCompactCursor {
bytes: &data[metadata_start..tail_start],
pos: 0,
};
cursor.file_metadata(limit)
rustfs_s3select_api::parquet_budget::validate_file_footer(data, limit).map_err(parquet_admission_error)
}
/// Sum the page headers of one column chunk (`[start, end)` in the file) and
/// hold every page and the chunk total to `limit`. Pages are never decompressed.
fn validate_page_headers_in_range(chunk: &[u8], chunk_declared: usize, limit: usize) -> TableCatalogStoreResult<usize> {
let mut cursor = ThriftCompactCursor { bytes: chunk, pos: 0 };
let mut total = 0_usize;
while cursor.pos < chunk.len() {
let sizes = cursor.page_header()?;
add_budget(&mut total, sizes.uncompressed, limit)?;
if total > chunk_declared.max(limit.min(chunk_declared.saturating_mul(2))) {
return Err(TableCatalogStoreError::Invalid(
"compaction parquet page headers exceed the footer's declared chunk size".to_owned(),
));
}
cursor.skip_bytes(sizes.compressed)?;
}
Ok(total)
#[cfg(test)]
fn validate_page_headers_in_range(chunk: &[u8], declared: usize, limit: usize) -> TableCatalogStoreResult<usize> {
rustfs_s3select_api::parquet_budget::validate_page_headers(chunk, declared, limit).map_err(parquet_admission_error)
}
fn validate_column_chunk_page_headers(
data: &Bytes,
column: &ColumnChunkMetaData,
chunk_declared: usize,
declared: usize,
limit: usize,
location: &str,
) -> TableCatalogStoreResult<()> {
// parquet's `byte_range`: the chunk starts at the dictionary page when the
// footer declares one, wherever that is. Walk exactly the bytes it decodes.
let start = column.dictionary_page_offset().unwrap_or(column.data_page_offset());
let start = usize::try_from(start).map_err(|_| budget_error())?;
let length = usize::try_from(column.compressed_size()).map_err(|_| budget_error())?;
let end = start.checked_add(length).filter(|end| *end <= data.len()).ok_or_else(|| {
TableCatalogStoreError::Invalid(format!("compaction parquet {location} column chunk lies outside the file"))
})?;
validate_page_headers_in_range(&data[start..end], chunk_declared, limit).map_err(|err| match err {
TableCatalogStoreError::Invalid(message) => TableCatalogStoreError::Invalid(format!("{location}: {message}")),
other => other,
})?;
Ok(())
) -> TableCatalogStoreResult<usize> {
rustfs_s3select_api::parquet_budget::validate_column_chunk_page_headers(data, column, declared, limit).map_err(|error| {
match parquet_admission_error(error) {
TableCatalogStoreError::Invalid(message) => TableCatalogStoreError::Invalid(format!("{location}: {message}")),
error => error,
}
})
}
fn view_field(field: &Field, depth: usize, fields: &mut usize) -> TableCatalogStoreResult<Arc<Field>> {
@@ -542,11 +316,11 @@ fn compact_parquet_with_budgets(
let mut columns = 0;
for column in group.columns() {
let chunk_declared = usize::try_from(column.uncompressed_size()).map_err(|_| budget_error())?;
add_budget(&mut columns, chunk_declared, limit)?;
// The footer is one claim; the page headers the decoder actually
// sizes its buffers from are another. Walk them without
// decompressing and hold both to the budget.
validate_column_chunk_page_headers(&data, column, chunk_declared, limit, &location)?;
// sizes its buffers from are another. Check the page headers
// and codec expansion before admitting the decoder.
let resident = validate_column_chunk_page_headers(&data, column, chunk_declared, limit, &location)?;
add_budget(&mut columns, chunk_declared.max(resident), limit)?;
}
let group_bytes = usize::try_from(group.total_byte_size()).map_err(|_| budget_error())?;
add_budget(&mut declared, group_bytes.max(columns), limit)?;
@@ -850,9 +624,9 @@ mod tests {
#[test]
fn compaction_page_header_walker_bounds_collection_counts_by_remaining_bytes() {
let limit = 64 * 1024 * 1024;
// Unknown field 4: LIST<BOOL> with long-form count 2^63. Parquet's skip
// Unknown field 10: LIST<BOOL> with long-form count 2^63. Parquet's skip
// would spin; the walker must reject it before iterating.
let mut list = vec![(1 << 4) | THRIFT_COMPACT_LIST, 0xf0 | THRIFT_COMPACT_BOOL_TRUE];
let mut list = vec![(7 << 4) | THRIFT_COMPACT_LIST, 0xf0 | THRIFT_COMPACT_BOOL_TRUE];
list.extend(varint(1 << 63));
let started = std::time::Instant::now();
let error = validate_page_headers_in_range(&page_chunk(compact_i32_field(1, 8), &list), 4096, limit)
@@ -864,15 +638,15 @@ mod tests {
assert!(started.elapsed() < std::time::Duration::from_secs(1));
// A bool collection of any size is rejected: parquet consumes no bytes
// per element, so either width would desynchronize the two parsers.
let list = [(1 << 4) | THRIFT_COMPACT_LIST, (2 << 4) | THRIFT_COMPACT_BOOL_TRUE, 1, 1];
let list = [(7 << 4) | THRIFT_COMPACT_LIST, (2 << 4) | THRIFT_COMPACT_BOOL_TRUE, 1, 1];
assert!(validate_page_headers_in_range(&page_chunk(compact_i32_field(1, 8), &list), 4096, limit).is_err());
// MAP<BOOL, BOOL> with a huge count is rejected the same way.
let mut map = vec![(1 << 4) | THRIFT_COMPACT_MAP];
let mut map = vec![(7 << 4) | THRIFT_COMPACT_MAP];
map.extend(varint(1 << 62));
map.push((THRIFT_COMPACT_BOOL_TRUE << 4) | THRIFT_COMPACT_BOOL_TRUE);
assert!(validate_page_headers_in_range(&page_chunk(compact_i32_field(1, 8), &map), 4096, limit).is_err());
// Collections whose elements carry bytes still skip when the count fits.
let mut list = vec![(1 << 4) | THRIFT_COMPACT_LIST, (2 << 4) | THRIFT_COMPACT_I32];
let mut list = vec![(7 << 4) | THRIFT_COMPACT_LIST, (2 << 4) | THRIFT_COMPACT_I32];
list.extend(compact_i32_field(0, 7)[1..].iter());
list.extend(compact_i32_field(0, -7)[1..].iter());
assert_eq!(
@@ -880,7 +654,7 @@ mod tests {
8
);
// A LIST<I32> whose count exceeds the chunk is malformed, not truncated.
let list = [(1 << 4) | THRIFT_COMPACT_LIST, 0xf0 | THRIFT_COMPACT_I32, 0x7f];
let list = [(7 << 4) | THRIFT_COMPACT_LIST, 0xf0 | THRIFT_COMPACT_I32, 0x7f];
assert!(validate_page_headers_in_range(&page_chunk(compact_i32_field(1, 8), &list), 4096, limit).is_err());
}
@@ -956,6 +730,35 @@ mod tests {
}
}
#[test]
fn compaction_aggregates_column_decode_working_sets() {
let columns =
(0..4).map(|index| (format!("value_{index}"), Arc::new(StringArray::from(vec!["v"; 1000])) as Arc<dyn Array>));
let batch = RecordBatch::try_from_iter(columns).expect("multi-column batch");
let properties = WriterProperties::builder()
.set_compression(Compression::ZSTD(ZstdLevel::default()))
.build();
let mut writer = ArrowWriter::try_new(Vec::new(), batch.schema(), Some(properties)).expect("Parquet writer");
writer.write(&batch).expect("multi-column data");
let data = Bytes::from(writer.into_inner().expect("Parquet footer"));
let limit = 100 * 1024;
let reader = ParquetRecordBatchReaderBuilder::try_new(data.clone()).expect("valid fixture metadata");
let mut working_set = 0;
for column in reader.metadata().row_group(0).columns() {
let declared = usize::try_from(column.uncompressed_size()).expect("positive size");
working_set += validate_column_chunk_page_headers(&data, column, declared, limit, "columns.parquet")
.expect("each column fits individually");
}
assert!(working_set > limit, "the aggregate exceeds the decode budget");
let error = compact_parquet_with_limits(vec![("columns.parquet".to_owned(), data.to_vec())], limit, limit)
.err()
.expect("aggregate column allocations must be admitted before decoding");
assert!(
matches!(&error, TableCatalogStoreError::Invalid(message) if message.contains("budget")),
"{error:?}"
);
}
#[test]
fn compaction_rejects_compressed_expansion_before_decoding_pages() {
let file = parquet_strings(&"x".repeat(64 * 1024), 32, false);
-4
View File
@@ -181,10 +181,6 @@ const WAREHOUSE_INDEX_STATE_FILE: &str = "state.json";
const TABLE_RENAME_ROOT: &str = "renames";
const WAREHOUSE_INDEX_MAX_PREFIX_DEPTH: usize = 64;
const TABLE_DATA_PLANE_INDEX_MISS_SCAN_MAX_CATALOG_OBJECTS: usize = 4096;
/// Listing pages an index-miss scan may consume before failing closed. Only
/// table entries count toward the object limit, so maintenance reports and
/// other catalog objects cannot trip it; this bounds the listing work itself.
const TABLE_DATA_PLANE_INDEX_MISS_SCAN_MAX_PAGES: usize = 64;
const EXTERNAL_CATALOG_ROOT: &str = "external-catalog";
const EXTERNAL_CATALOG_BRIDGE_FILE: &str = "bridge.json";
const MAINTENANCE_ROOT: &str = "maintenance";
+42 -1
View File
@@ -40,7 +40,7 @@ pub(super) use backup_types::{TableCatalogBackupStatus, TableCatalogRestoreStatu
use manifest::table_catalog_backing_manifest;
pub(crate) use object::ObjectTableCatalogStore;
#[cfg(test)]
pub(super) use object::{bounded_table_entry_objects_for_data_plane_scan, maintenance_subtree_skip_cursor};
pub(super) use object::bounded_table_entry_objects_for_data_plane_scan;
struct PeriodicTableMaintenanceQueueGuard<'a> {
table_id: &'a str,
@@ -1055,6 +1055,20 @@ pub(crate) trait TableCatalogObjectBackend: Clone + Send + Sync + 'static {
))
}
/// GET-equivalent table ownership entries, selected before metadata reads.
/// Report history must not consume pages. No ordinary S3 list fallback is safe.
async fn list_table_entry_objects_page(
&self,
_bucket: &str,
_prefix: &str,
_start_after: Option<&str>,
_limit: NonZeroUsize,
) -> TableCatalogStoreResult<TableCatalogObjectListPage> {
Err(TableCatalogStoreError::Unsupported(
"authoritative table-entry pagination is not supported by this backend".to_string(),
))
}
/// GET-equivalent, bounded enumeration for publishing authoritative indexes.
/// Backends without this contract must not fall back to an ordinary S3 list.
async fn list_catalog_objects_page(
@@ -2615,6 +2629,33 @@ where
})
}
async fn list_table_entry_objects_page(
&self,
bucket: &str,
prefix: &str,
start_after: Option<&str>,
limit: NonZeroUsize,
) -> TableCatalogStoreResult<TableCatalogObjectListPage> {
let max_keys = i32::try_from(limit.get())
.map_err(|_| TableCatalogStoreError::Invalid("catalog page size exceeds storage API range".to_string()))?;
let page = self
.store
.clone()
.list_identifier_tree_leaves_authoritative(
bucket,
prefix,
TABLE_ENTRY_FILE,
start_after.map(str::to_string),
max_keys,
)
.await
.map_err(|err| storage_error_to_catalog("list authoritative table entries", err))?;
Ok(TableCatalogObjectListPage {
objects: page.objects.into_iter().map(|object| object.name).collect(),
is_truncated: page.is_truncated,
})
}
async fn list_catalog_objects_page(
&self,
bucket: &str,
+3 -37
View File
@@ -430,20 +430,6 @@ pub(in crate::table_catalog) fn bounded_table_entry_objects_for_data_plane_scan(
Ok(entries)
}
/// When `object` lies in a table's `maintenance/` subtree, the cursor that skips
/// the rest of that subtree. Only keys under that directory are skipped, never
/// the table entry or any other table beside it.
pub(in crate::table_catalog) fn maintenance_subtree_skip_cursor(prefix: &str, object: &str) -> Option<String> {
let relative = object.strip_prefix(prefix)?;
let mut segments = relative.splitn(5, '/');
let namespace = segments.next()?;
let root = segments.next()?;
let table = segments.next()?;
let maintenance = segments.next()?;
(root == TABLE_ROOT && maintenance == MAINTENANCE_ROOT && segments.next().is_some())
.then(|| format!("{prefix}{namespace}/{root}/{table}/{maintenance}/\u{10ffff}"))
}
#[derive(Clone)]
pub(crate) struct ObjectTableCatalogStore<B> {
pub(in crate::table_catalog) backend: B,
@@ -2861,41 +2847,21 @@ where
max_catalog_objects: usize,
) -> TableCatalogStoreResult<Vec<String>> {
let mut objects = Vec::new();
let mut entries = 0_usize;
let mut pages = 0_usize;
let mut cursor: Option<String> = None;
let prefix = self.paths.namespace_entries_prefix(table_bucket);
let limit = NonZeroUsize::new(TABLE_CATALOG_LIST_MAX_KEYS)
.ok_or_else(|| TableCatalogStoreError::Internal("table catalog listing page size must be positive".to_string()))?;
loop {
// A page that ends inside a table's maintenance subtree skips the
// rest of it, so each entry can waste at most one page: honest
// catalogs scale with their entries while a report flood under one
// table stays within the fixed budget.
if pages >= TABLE_DATA_PLANE_INDEX_MISS_SCAN_MAX_PAGES.saturating_add(entries) {
return Err(TableCatalogStoreError::Unavailable(format!(
"table data-plane warehouse-index miss scan exceeds the {TABLE_DATA_PLANE_INDEX_MISS_SCAN_MAX_PAGES}-page listing budget"
)));
}
let page = self
.backend
.list_objects_page(self.catalog_bucket(), &prefix, cursor.as_deref(), limit)
.list_table_entry_objects_page(self.catalog_bucket(), &prefix, cursor.as_deref(), limit)
.await?;
pages += 1;
let next_cursor = page
.objects
.last()
.map(|last| maintenance_subtree_skip_cursor(&prefix, last).unwrap_or_else(|| last.clone()));
entries += page
.objects
.iter()
.filter(|object| object.ends_with(TABLE_ENTRY_FILE))
.count();
let next_cursor = page.objects.last().cloned();
objects.extend(page.objects);
if !page.is_truncated {
return bounded_table_entry_objects_for_data_plane_scan(objects, false, max_catalog_objects);
}
if entries > max_catalog_objects {
if objects.len() > max_catalog_objects {
return bounded_table_entry_objects_for_data_plane_scan(objects, true, max_catalog_objects);
}
let Some(next_cursor) = next_cursor else {
+52 -23
View File
@@ -594,6 +594,29 @@ impl TestCatalogObjectBackend {
state.list_calls = 0;
}
async fn overridden_list_page(&self, bucket: &str, prefix: &str) -> Option<TableCatalogObjectListPage> {
let pause = self
.state
.lock()
.await
.list_page_pauses
.get_mut(&(bucket.to_string(), prefix.to_string()))
.and_then(std::collections::VecDeque::pop_front);
if let Some(pause) = pause {
pause.started.notify_one();
pause.release.notified().await;
}
let mut state = self.state.lock().await;
let page = state
.list_page_overrides
.get_mut(&(bucket.to_string(), prefix.to_string()))
.and_then(std::collections::VecDeque::pop_front);
if page.is_some() {
state.list_calls += 1;
}
page
}
pub(crate) async fn write_lock_acquisition_count(&self, bucket: &str, object: &str) -> usize {
self.state
.lock()
@@ -1231,29 +1254,7 @@ impl TableCatalogObjectBackend for TestCatalogObjectBackend {
start_after: Option<&str>,
limit: NonZeroUsize,
) -> TableCatalogStoreResult<TableCatalogObjectListPage> {
let pause = self
.state
.lock()
.await
.list_page_pauses
.get_mut(&(bucket.to_string(), prefix.to_string()))
.and_then(std::collections::VecDeque::pop_front);
if let Some(pause) = pause {
pause.started.notify_one();
pause.release.notified().await;
}
let injected = {
let mut state = self.state.lock().await;
let page = state
.list_page_overrides
.get_mut(&(bucket.to_string(), prefix.to_string()))
.and_then(std::collections::VecDeque::pop_front);
if page.is_some() {
state.list_calls += 1;
}
page
};
if let Some(page) = injected {
if let Some(page) = self.overridden_list_page(bucket, prefix).await {
return Ok(page);
}
let mut objects = self.list_objects(bucket, prefix).await?;
@@ -1266,6 +1267,34 @@ impl TableCatalogObjectBackend for TestCatalogObjectBackend {
Ok(TableCatalogObjectListPage { objects, is_truncated })
}
async fn list_table_entry_objects_page(
&self,
bucket: &str,
prefix: &str,
start_after: Option<&str>,
limit: NonZeroUsize,
) -> TableCatalogStoreResult<TableCatalogObjectListPage> {
if let Some(page) = self.overridden_list_page(bucket, prefix).await {
return Ok(page);
}
// This in-memory backend has no erasure replicas. Production selects
// candidates before metadata I/O and then proves each key at read quorum.
let mut inventory = self.list_objects_for_cleanup(bucket, prefix).await?;
inventory.retain(|object| object.key.rsplit('/').next() == Some(TABLE_ENTRY_FILE));
if inventory
.iter()
.any(|object| object.version_id.is_some() || object.delete_marker)
{
return Err(TableCatalogStoreError::Unavailable("versioned table ownership entry".to_string()));
}
let mut objects = inventory.into_iter().map(|object| object.key).collect::<Vec<_>>();
objects.sort();
objects.retain(|object| start_after.is_none_or(|after| object.as_str() > after));
let is_truncated = objects.len() >= limit.get();
objects.truncate(limit.get());
Ok(TableCatalogObjectListPage { objects, is_truncated })
}
async fn list_objects_for_cleanup(
&self,
bucket: &str,
+108 -62
View File
@@ -1814,43 +1814,6 @@ fn table_data_plane_index_miss_scan_rejects_catalogs_above_its_object_limit() {
);
}
#[test]
fn table_data_plane_index_miss_scan_skips_only_a_table_maintenance_subtree() {
let prefix = "catalog/namespaces/";
let skip = |object: &str| maintenance_subtree_skip_cursor(prefix, &format!("{prefix}{object}"));
assert_eq!(
skip("sales/tables/orders/maintenance/hash/jobs/job.json").as_deref(),
Some("catalog/namespaces/sales/tables/orders/maintenance/\u{10ffff}")
);
assert_eq!(
skip("sales/tables/orders/maintenance/hash/config.json").as_deref(),
Some("catalog/namespaces/sales/tables/orders/maintenance/\u{10ffff}")
);
// A table or namespace literally named "maintenance" is not a subtree.
assert_eq!(skip("sales/tables/maintenance/table-entry.json"), None);
assert_eq!(skip("maintenance/tables/orders/table-entry.json"), None);
assert_eq!(
skip("maintenance/tables/maintenance/maintenance/hash/jobs/job.json").as_deref(),
Some("catalog/namespaces/maintenance/tables/maintenance/maintenance/\u{10ffff}")
);
// Entries, bridges, views and a bare file named maintenance never jump.
assert_eq!(skip("sales/tables/orders/table-entry.json"), None);
assert_eq!(skip("sales/tables/orders/external-catalog/bridge.json"), None);
assert_eq!(skip("sales/views/report/maintenance/hash/jobs/job.json"), None);
assert_eq!(skip("sales/tables/orders/maintenance"), None);
assert_eq!(skip("sales/namespace.json"), None);
assert_eq!(maintenance_subtree_skip_cursor(prefix, "other/sales/tables/orders/maintenance/x/y"), None);
// The jump lands before the entry that follows the subtree.
assert!(
"catalog/namespaces/sales/tables/orders/maintenance/\u{10ffff}"
< "catalog/namespaces/sales/tables/orders/table-entry.json"
);
assert!(
"catalog/namespaces/sales/tables/orders/maintenance/\u{10ffff}"
< "catalog/namespaces/sales/tables/orders0/table-entry.json"
);
}
#[test]
fn catalog_lock_authority_failures_are_typed_as_unavailable() {
for error in [
@@ -6880,39 +6843,123 @@ async fn seed_noise_table_with_job_reports(
#[tokio::test]
async fn table_data_plane_index_miss_scan_pages_scale_with_entries_not_job_reports() {
let mut baseline_pages = None;
for reports in [0, 80] {
let backend = TestCatalogObjectBackend::default();
let store = ObjectTableCatalogStore::new(backend.clone());
let bucket = "analytics";
let namespace = Namespace::parse("sales").expect("namespace should parse");
let table = IdentifierSegment::parse("orders").expect("table should parse");
let current = default_table_metadata_file_path(&namespace, &table, "00001.metadata.json");
let object = "tables/table-id/data/part-00001.parquet";
seed_table_for_metadata_maintenance(&store, bucket, &namespace, &table, current).await;
backend
.delete_object(RUSTFS_META_BUCKET, &store.paths.warehouse_index_entry_path(bucket, "tables/table-id/"))
.await
.expect("warehouse index entry should be removed");
// Keep the same table inventory with and without historical job reports.
let noise_namespace = Namespace::parse("noise").expect("namespace should parse");
for index in 0..1000 {
let noise_table = IdentifierSegment::parse(format!("noise-{index:04}")).expect("table should parse");
seed_noise_table_with_job_reports(&store, bucket, &noise_namespace, &noise_table, &format!("noise-{index}"), reports)
.await;
}
backend.reset_call_counts().await;
let resource = table_data_plane_resource_for_object(&store, bucket, object)
.await
.expect("job history must not exhaust the index-miss listing budget")
.expect("the scan must retain table-aware protection");
assert_eq!(resource.table, "orders");
let pages = backend.list_call_count().await;
assert!(pages > 0, "the missing index must exercise enumeration");
if let Some(expected) = baseline_pages {
assert_eq!(pages, expected, "job history must not consume ownership listing pages");
} else {
baseline_pages = Some(pages);
}
}
}
#[tokio::test]
async fn table_entry_scan_preserves_nested_namespaces_named_like_maintenance() {
for name in ["sales", "sales.tables.orders.maintenance"] {
let backend = TestCatalogObjectBackend::default();
let store = ObjectTableCatalogStore::new(backend.clone());
let bucket = "analytics";
let namespace = Namespace::parse(name).expect("namespace should parse");
let owner = IdentifierSegment::parse("zz-owner").expect("owner should parse");
let current = default_table_metadata_file_path(&namespace, &owner, "00001.metadata.json");
seed_table_for_metadata_maintenance(&store, bucket, &namespace, &owner, current).await;
backend
.delete_object(RUSTFS_META_BUCKET, &store.paths.warehouse_index_entry_path(bucket, "tables/table-id/"))
.await
.expect("remove the owner index to exercise catalog fallback");
for index in 0..1000 {
let table = IdentifierSegment::parse(format!("table-{index:04}")).expect("table should parse");
seed_noise_table_with_job_reports(&store, bucket, &namespace, &table, &format!("id-{index}"), 0).await;
}
let resource = store
.resolve_table_data_plane_resource_for_authorization(bucket, "tables/table-id/data/part.parquet")
.await
.expect("registered table entries must remain enumerable")
.unwrap_or_else(|| panic!("namespace {name} lost its table owner across a page boundary"));
assert_eq!(resource.namespace, name);
assert_eq!(resource.table, "zz-owner");
}
}
#[tokio::test]
async fn object_maintenance_preserves_an_owner_after_nested_namespace_page_boundary() {
let backend = TestCatalogObjectBackend::default();
let store = ObjectTableCatalogStore::new(backend.clone());
let bucket = "analytics";
let namespace = Namespace::parse("sales").expect("namespace should parse");
let table = IdentifierSegment::parse("orders").expect("table should parse");
let current = default_table_metadata_file_path(&namespace, &table, "00001.metadata.json");
let object = "tables/table-id/data/part-00001.parquet";
seed_table_for_metadata_maintenance(&store, bucket, &namespace, &table, current).await;
let current = default_table_metadata_file_path(&namespace, &table, "00002.metadata.json");
let orphan = default_table_metadata_file_path(&namespace, &table, "00001.metadata.json");
seed_table_for_metadata_maintenance(&store, bucket, &namespace, &table, current.clone()).await;
backend
.delete_object(RUSTFS_META_BUCKET, &store.paths.warehouse_index_entry_path(bucket, "tables/table-id/"))
.seed_object(bucket, &current, br#"{"metadata-log":[]}"#.to_vec())
.await;
backend.seed_object(bucket, &orphan, br#"{"metadata-log":[]}"#.to_vec()).await;
let report = store
.plan_table_metadata_maintenance(bucket, "sales", "orders", 0)
.await
.expect("warehouse index entry should be removed");
// 1000 honest tables with 80 historical job reports each: ~80 pages of
// objects, beyond the fixed page budget on its own, yet well within one
// extra page per table entry.
let noise_namespace = Namespace::parse("noise").expect("namespace should parse");
for index in 0..1000 {
let noise_table = IdentifierSegment::parse(format!("noise-{index:04}")).expect("table should parse");
seed_noise_table_with_job_reports(&store, bucket, &noise_namespace, &noise_table, &format!("noise-{index}"), 80).await;
}
backend.reset_call_counts().await;
.expect("the unowned old metadata should initially be a cleanup candidate");
assert_eq!(report.deletable_metadata_locations, vec![orphan.clone()]);
let resource = table_data_plane_resource_for_object(&store, bucket, object)
let nested = Namespace::parse("sales.tables.orders.maintenance").expect("nested namespace should parse");
store
.create_namespace(test_namespace_entry(bucket, &nested))
.await
.expect("job history must not exhaust the index-miss listing budget")
.expect("the scan must retain table-aware protection");
assert_eq!(resource.table, "orders");
let pages = backend.list_call_count().await;
.expect("nested namespace should be created");
for index in 0..1100 {
let noise = IdentifierSegment::parse(format!("table-{index:04}")).expect("table should parse");
seed_noise_table_with_job_reports(&store, bucket, &nested, &noise, &format!("noise-{index}"), 0).await;
}
let owner = IdentifierSegment::parse("zz-owner").expect("owner should parse");
let mut other = test_table_entry(bucket, &nested, &owner, orphan.clone());
other.table_id = "other-id".to_string();
other.table_uuid = "other-uuid".to_string();
other.warehouse_location = format!("s3://{bucket}/tables/other-id");
store
.create_table(other)
.await
.expect("another table may adopt the metadata before cleanup");
let deletes_before = backend.state.lock().await.delete_attempts.clone();
let result = store
.delete_table_metadata_maintenance_report(bucket, "sales", "orders", report)
.await;
assert!(
pages > TABLE_DATA_PLANE_INDEX_MISS_SCAN_MAX_PAGES && pages <= TABLE_DATA_PLANE_INDEX_MISS_SCAN_MAX_PAGES + 1001,
"the honest catalog needs {pages} pages: more than the fixed budget, at most one extra per entry"
backend
.object_exists(bucket, &orphan)
.await
.expect("read back the current owner's metadata")
);
assert_eq!(backend.state.lock().await.delete_attempts, deletes_before);
assert_matches!(result, Err(TableCatalogStoreError::Conflict(message)) if message.contains("exclusive metadata"));
}
#[tokio::test]
@@ -6930,8 +6977,7 @@ async fn table_data_plane_index_miss_scan_bounds_a_job_report_flood_under_one_ta
.delete_object(RUSTFS_META_BUCKET, &store.paths.warehouse_index_entry_path(bucket, "tables/table-id/"))
.await
.expect("warehouse index entry should be removed");
// A table-scoped principal floods one table's job history: 70 reports per
// page budget would have exhausted the old fixed budget bucket-wide.
// Historical report volume must not consume ownership pagination.
for job in 0..70_000 {
backend
.seed_object(
@@ -6951,7 +6997,7 @@ async fn table_data_plane_index_miss_scan_bounds_a_job_report_flood_under_one_ta
.expect("the scan must retain table-aware protection");
assert_eq!(resource.table, "orders");
let pages = backend.list_call_count().await;
assert!(pages <= 2, "the page that ends inside the flood skips the rest of it, not {pages} pages");
assert!(pages <= 2, "report history must not consume {pages} ownership pages");
}
#[tokio::test]
+111
View File
@@ -0,0 +1,111 @@
#!/usr/bin/env python3
"""Exercise the release download entry point without fetching external tools."""
import hashlib
import json
import os
from pathlib import Path
import subprocess
import sys
import tempfile
import unittest
SCRIPT = Path(__file__).resolve().parents[1] / ".github/actions/setup/download_release_asset.py"
class ReleaseAssetTests(unittest.TestCase):
def setUp(self):
self.temp = tempfile.TemporaryDirectory()
self.addCleanup(self.temp.cleanup)
self.root = Path(self.temp.name)
self.body = b"verified tool fixture"
self.sha256 = hashlib.sha256(self.body).hexdigest()
self.asset = {"name": "tool.zip", "id": 17, "size": len(self.body), "digest": "sha256:" + self.sha256}
self.metadata = {"tag_name": "v1.0", "assets": [self.asset]}
self.output = self.root / "tool.zip"
self.output.write_bytes(b"previous file")
self.calls = self.root / "calls.jsonl"
gh = self.root / "gh"
gh.write_text(f"#!{sys.executable}\n" + '''import json,os,pathlib,sys
args=sys.argv[1:]
with pathlib.Path(os.environ['CALLS']).open('a') as out:out.write(json.dumps(args)+'\\n')
if '/releases/tags/' in args[-1]:
if os.environ.get('FAIL_METADATA'):raise SystemExit(19)
print(os.environ['METADATA'])
else:
if os.environ.get('FAIL_DOWNLOAD'):raise SystemExit(23)
sys.stdout.buffer.write(bytes.fromhex(os.environ['BODY']))
''')
gh.chmod(0o755)
self.env = dict(os.environ, PATH=str(self.root) + os.pathsep + os.environ["PATH"], CALLS=str(self.calls), GH_TOKEN="synthetic-do-not-log-token")
def invoke(self, extra=(), **env):
environment = dict(self.env, METADATA=json.dumps(self.metadata), BODY=self.body.hex(), **env)
return subprocess.run([sys.executable, str(SCRIPT), "--repository", "owner/repo", "--tag", "v1.0", "--asset", "tool.zip", "--output", str(self.output), *extra], env=environment, capture_output=True, text=True)
def assert_failed_atomically(self, result):
self.assertNotEqual(result.returncode, 0)
self.assertEqual(self.output.read_bytes(), b"previous file")
self.assertEqual(list(self.root.glob(".release-asset-*")), [])
self.assertNotIn(self.env["GH_TOKEN"], result.stdout + result.stderr)
def test_valid_asset_uses_api_and_installs_only_verified_bytes(self):
result = self.invoke(("--sha256", self.sha256))
self.assertEqual(result.returncode, 0, result.stderr)
self.assertEqual(self.output.read_bytes(), self.body)
calls = [json.loads(line) for line in self.calls.read_text().splitlines()]
self.assertEqual(calls[0], ["api", "--hostname", "github.com", "repos/owner/repo/releases/tags/v1.0"])
self.assertEqual(calls[1], ["api", "--hostname", "github.com", "--header", "Accept: application/octet-stream", "repos/owner/repo/releases/assets/17"])
self.assertNotIn(self.env["GH_TOKEN"], json.dumps(calls) + result.stdout + result.stderr)
def test_download_failure_preserves_destination(self):
self.assert_failed_atomically(self.invoke(FAIL_DOWNLOAD="1"))
def test_missing_gh_fails_without_replacing_destination(self):
(self.root / "gh").unlink()
self.env["PATH"] = str(self.root)
self.assert_failed_atomically(self.invoke())
def test_metadata_failure_preserves_destination(self):
self.assert_failed_atomically(self.invoke(FAIL_METADATA="1"))
def test_corrupt_download_fails_digest_validation(self):
self.body = b"modified tool fixture"
self.assert_failed_atomically(self.invoke())
def test_truncated_download_fails_size_validation(self):
self.asset["size"] += 1
self.assert_failed_atomically(self.invoke())
def test_pinned_digest_cannot_be_replaced_by_release_metadata(self):
self.assert_failed_atomically(self.invoke(("--sha256", "0" * 64)))
self.assertEqual(len(self.calls.read_text().splitlines()), 1)
def test_ambiguous_or_missing_assets_fail_before_download(self):
for assets in ([], [self.asset, self.asset]):
with self.subTest(assets=len(assets)):
self.metadata["assets"] = assets
self.assert_failed_atomically(self.invoke())
def test_metadata_tag_must_match_requested_version(self):
self.metadata["tag_name"] = "v2.0"
self.assert_failed_atomically(self.invoke())
def test_malformed_release_inventory_fails_before_download(self):
for metadata in ([], {"tag_name": "v1.0", "assets": None}, {"tag_name": "v1.0", "assets": [None]}):
with self.subTest(metadata=metadata):
self.metadata = metadata
self.assert_failed_atomically(self.invoke())
def test_missing_digest_or_invalid_identity_fails_before_download(self):
for field, value in [("digest", None), ("digest", "sha256:invalid"), ("id", True), ("id", 0), ("size", 0), ("size", 513 * 1024 * 1024)]:
with self.subTest(field=field, value=value):
self.asset[field] = value
self.assert_failed_atomically(self.invoke())
self.asset = {"name": "tool.zip", "id": 17, "size": len(self.body), "digest": "sha256:" + self.sha256}
self.metadata["assets"] = [self.asset]
if __name__ == "__main__":
unittest.main()