mirror of
https://github.com/rustfs/rustfs.git
synced 2026-11-08 19:43:05 +00:00
Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 1dea00ffa7 | |||
| addf7a6219 | |||
| 420cf32b2a |
@@ -1,2 +1,2 @@
|
||||
sha256-darwin=219ac11e720b3f54dfee4d5d5e95171785ba159b2fe01db3037a8f5061c1ad32
|
||||
sha256-linux=2d34698cc8ec52d5b6c37ac612919a1d21f36cd56a1af539a7522a4eea518e26
|
||||
sha256-darwin=9901a4950099b94d3d2618524d540f1ed4bf6023044903886f98b6c4bf5f30fb
|
||||
sha256-linux=09fd5f48f5556c9fc576357222dccd7ebb0b90b790798f1834db2ec5d67b4952
|
||||
|
||||
@@ -54,6 +54,7 @@ script-tests: ## Run shell script tests
|
||||
./scripts/check_embedded_secrets.sh --self-test
|
||||
$(RUSTFS_PYTHON_BIN) ./scripts/check_test_wiring.py --self-test
|
||||
$(RUSTFS_PYTHON_BIN) ./scripts/test_e2e_binary.py
|
||||
$(RUSTFS_PYTHON_BIN) ./scripts/test_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
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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())
|
||||
@@ -142,15 +142,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
|
||||
@@ -1137,11 +1145,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"
|
||||
|
||||
@@ -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]);
|
||||
|
||||
@@ -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() {
|
||||
|
||||
@@ -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<()> {
|
||||
match std::fs::symlink_metadata(backup) {
|
||||
Ok(metadata) if metadata.is_file() => {
|
||||
@@ -3160,6 +3205,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"))]
|
||||
@@ -9220,15 +9269,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)));
|
||||
@@ -9253,7 +9301,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(());
|
||||
@@ -14318,7 +14374,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;
|
||||
};
|
||||
@@ -14445,8 +14502,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))?;
|
||||
}
|
||||
@@ -14454,10 +14511,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())?;
|
||||
}
|
||||
@@ -14502,8 +14557,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) {
|
||||
@@ -14560,7 +14618,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),
|
||||
@@ -14572,10 +14630,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()?;
|
||||
@@ -14583,6 +14642,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)?;
|
||||
@@ -14591,13 +14651,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(()))
|
||||
})
|
||||
@@ -14616,13 +14692,28 @@ impl DiskAPI for LocalDisk {
|
||||
let source_owner = self.multipart_source_owner(volume, path, true, false, None).await?;
|
||||
let admission = self.multipart_source_admission(volume, path, true, false)?;
|
||||
let admission_source = admission.as_ref().map(|a| a.source().to_path_buf());
|
||||
self.io_get_bucket_path(volume)?;
|
||||
let current_data_path = self.io_get_object_path(volume, path)?;
|
||||
let current_meta_path = self.io_get_object_path(volume, &format!("{path}.meta"))?;
|
||||
let transaction_path = self.io_get_object_path(volume, &crate::disk::part_transaction_path(path))?;
|
||||
for candidate in [¤t_data_path, ¤t_meta_path, &transaction_path] {
|
||||
check_path_length(candidate.to_string_lossy().as_ref())?;
|
||||
}
|
||||
let txn_rel_path = crate::disk::part_transaction_path(path);
|
||||
let cached_published = self
|
||||
.io_get_object_open_path(volume, &txn_rel_path)
|
||||
.ok()
|
||||
.and_then(|txn_open| take_published_part_txn_receipt(&txn_open));
|
||||
let (current_data_path, current_meta_path, transaction_path, skip_obsolete_read_on_commit) = match cached_published {
|
||||
Some(entry) => (
|
||||
entry.current_data_path,
|
||||
entry.current_meta_path,
|
||||
entry.transaction_path,
|
||||
!entry.had_old_meta,
|
||||
),
|
||||
None => {
|
||||
let (_, current_data_path) = self.io_get_bucket_and_object_path(volume, path)?;
|
||||
let current_meta_path = self.io_get_object_path(volume, &format!("{path}.meta"))?;
|
||||
let transaction_path = self.io_get_object_path(volume, &txn_rel_path)?;
|
||||
for candidate in [¤t_data_path, ¤t_meta_path, &transaction_path] {
|
||||
check_path_length(candidate.to_string_lossy().as_ref())?;
|
||||
}
|
||||
(current_data_path, current_meta_path, transaction_path, false)
|
||||
}
|
||||
};
|
||||
let durability = effective_durability(volume);
|
||||
let sync = durability.syncs_commit_metadata();
|
||||
let part_number = current_data_path
|
||||
@@ -14688,7 +14779,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()
|
||||
@@ -14754,15 +14846,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);
|
||||
|
||||
@@ -14779,25 +14862,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
|
||||
@@ -14856,35 +14988,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
|
||||
@@ -14947,6 +15087,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)
|
||||
|
||||
@@ -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())?;
|
||||
@@ -714,14 +758,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 {
|
||||
@@ -804,7 +847,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,
|
||||
@@ -876,7 +919,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);
|
||||
}
|
||||
@@ -884,14 +936,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));
|
||||
}
|
||||
|
||||
@@ -920,6 +979,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);
|
||||
}
|
||||
@@ -928,6 +989,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);
|
||||
}
|
||||
@@ -944,6 +1006,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);
|
||||
}
|
||||
@@ -1027,6 +1091,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()));
|
||||
}
|
||||
@@ -1047,6 +1113,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);
|
||||
}
|
||||
@@ -1075,8 +1142,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,
|
||||
@@ -1087,19 +1184,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);
|
||||
}
|
||||
|
||||
@@ -1364,7 +1461,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,
|
||||
|
||||
@@ -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;
|
||||
@@ -2587,6 +2662,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}"));
|
||||
@@ -2606,6 +2683,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);
|
||||
@@ -2653,7 +2731,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)))
|
||||
},
|
||||
)
|
||||
@@ -2690,19 +2779,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();
|
||||
@@ -2712,23 +2827,38 @@ 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()?;
|
||||
}
|
||||
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))));
|
||||
}
|
||||
@@ -2851,14 +2981,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 {
|
||||
@@ -2880,15 +3042,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()
|
||||
@@ -2897,6 +3053,31 @@ 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()?;
|
||||
}
|
||||
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}");
|
||||
@@ -2951,38 +3132,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()?;
|
||||
}
|
||||
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);
|
||||
@@ -3105,6 +3286,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::*;
|
||||
@@ -7020,6 +7237,73 @@ mod tests {
|
||||
marker.deleted = false;
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[tokio::test]
|
||||
async fn delete_recovery_fused_prepare_checkpoint_reopens_before_head_publication() {
|
||||
let (_temp, disk, mut first, _) = fixture().await;
|
||||
first.version_id = Some(Uuid::new_v4());
|
||||
let mut second = first.clone();
|
||||
second.version_id = Some(Uuid::new_v4());
|
||||
let mut meta = FileMeta::new();
|
||||
meta.add_version(first).expect("add surviving version");
|
||||
meta.add_version(second.clone()).expect("add deleted version");
|
||||
let original = meta.marshal_msg().expect("encode original versions");
|
||||
let (_, root) = disk.io_get_bucket_and_object_path("recovery", "object").expect("object path");
|
||||
fs::write(root.join(STORAGE_FORMAT_FILE), &original)
|
||||
.await
|
||||
.expect("write versions");
|
||||
|
||||
let (checkpoint_temp, checkpoint_disk, _, _) = fixture().await;
|
||||
let checkpoint_root = checkpoint_temp.path().join("recovery/object");
|
||||
drop(checkpoint_disk);
|
||||
let transaction = Uuid::new_v4();
|
||||
let source_root = root.clone();
|
||||
let copy_root = checkpoint_root.clone();
|
||||
let _capture = fused_prepare_checkpoint::install(&root, move || {
|
||||
let source = source_root.join(transaction.to_string());
|
||||
let destination = copy_root.join(transaction.to_string());
|
||||
std::fs::create_dir(&destination).expect("create checkpoint transaction");
|
||||
std::fs::copy(source_root.join(STORAGE_FORMAT_FILE), copy_root.join(STORAGE_FORMAT_FILE))
|
||||
.expect("capture head before publication");
|
||||
for entry in std::fs::read_dir(source).expect("read published preparation") {
|
||||
let entry = entry.expect("prepared file");
|
||||
assert!(entry.file_type().expect("prepared file type").is_file());
|
||||
std::fs::copy(entry.path(), destination.join(entry.file_name())).expect("capture prepared file");
|
||||
}
|
||||
});
|
||||
let opts = DeleteOptions {
|
||||
recoverable_delete: true,
|
||||
..Default::default()
|
||||
};
|
||||
assert_eq!(
|
||||
disk.try_fused_recoverable_delete_version("recovery", "object", &second, &opts, transaction, None)
|
||||
.await
|
||||
.expect("finish original preparation"),
|
||||
Some(())
|
||||
);
|
||||
assert!(checkpoint_root.join(transaction.to_string()).join(PREPARED).is_file());
|
||||
let endpoint =
|
||||
Endpoint::try_from(checkpoint_temp.path().to_str().expect("checkpoint path")).expect("checkpoint endpoint");
|
||||
let reopened = LocalDisk::new(&endpoint, false)
|
||||
.await
|
||||
.expect("reopen checkpoint as a fresh disk");
|
||||
let inspection = reopened
|
||||
.delete_recovery_inspect("recovery", "object", transaction)
|
||||
.await
|
||||
.expect("actual preparation checkpoint must remain recoverable");
|
||||
assert_eq!(inspection.state, DeleteRecoveryState::NewPrepared);
|
||||
reopened
|
||||
.settle_delete_transaction("recovery", "object", transaction, inspection.fingerprint, DeleteRecoveryAction::Rollback)
|
||||
.await
|
||||
.expect("roll back interrupted preparation");
|
||||
assert_eq!(
|
||||
fs::read(checkpoint_root.join(STORAGE_FORMAT_FILE))
|
||||
.await
|
||||
.expect("read rolled-back versions"),
|
||||
original
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn delete_recovery_fused_versioned_delete_with_surviving_version_and_interrupted_target_cleanup() {
|
||||
let (_temp, disk, mut v1, _) = fixture().await;
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -2200,6 +2200,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)]
|
||||
@@ -2408,7 +2409,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;
|
||||
@@ -2426,7 +2427,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;
|
||||
}
|
||||
@@ -7443,6 +7449,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
|
||||
@@ -7451,9 +7458,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)?;
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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())
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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() {
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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);
|
||||
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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()
|
||||
Reference in New Issue
Block a user