Compare commits

...

8 Commits

Author SHA1 Message Date
Chris An bc0706b2e3 fix(ci): accept binary target IDs in workspace shards (#9075) 2026-10-30 23:21:50 +08:00
Ramakrishna Chilaka c6152954f6 perf(ecstore): fuse local PUT/DELETE commit I/O and eliminate redundant metadata disk reads (#9038) (#9064)
* perf(ecstore): fuse local PUT/DELETE commit I/O and eliminate redundant metadata disk reads (#9038)

- Fuse authoritative bucket metadata existence and parse reads via tokio::join!, add a LOCAL_META_MUTATION_GEN-validated local authority cache, and piggyback one-shot quota snapshot reads onto single-PUT Object Lock snapshot loads.
- Reuse prepared ObjectOptions in post-write lifecycle transition checks instead of re-reading bucket versioning and Object Lock config after commit.
- Defer temporary shard file open/mkdir/write into a single spawn_blocking call via LazyLocalFileWriter, pre-sync single-block shards before page-cache reclaim, and fuse rename_data preparation + commit into a single Fsync-lane dispatch on Linux.
- Enable rename_data old_current_source reuse by default across all-local sets, guard early-ack commit on quorum source stability, and skip redundant post-commit tmp_delete when all online local disks succeeded.
- Enable data-first batch_shard_pread when deferred parity handles are available and replace zero-filled allocation with Vec::with_capacity + read_to_end for offset-0 shard reads.
- Defer pre-delete get_object_info_for_delete to PrefixAccessDenied error handling, skip redundant single-pool unversioned delete pre-stats, and add a fused local delete preparation and 2-barrier settlement fast path.

Fixes #9038

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Copilot-Session: d85aaf47-2a4b-4c2c-97a3-dbe8c7717e2a

* fix(ecstore): preserve delete identity and durability policy

Co-Authored-By: heihutu <heihutu@gmail.com>

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* test(ecstore): cover fused delete recovery stages

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* fix(ecstore): preserve inline versions after commit sync failure

* perf(ecstore,s3): accelerate PutObjectLegalHold, CopyObject, and large PUT fdatasync (#9038)

- Fuse LocalDisk::update_metadata (read xl.meta + FileMeta::load + update_object_version_with_opts + marshal_msg + staged tmp write + rename_all_with_precondition_and_parent_sync_std) into a single Fsync-lane blocking operation, and skip redundant get_bucket_info / get_object_info calls in put_object_legal_hold when object lock is enabled and replication is not configured (PutObjectLegalHold: 314.8 -> 988.3 ops/s, 3.14x faster, 0.95x of MinIO).
- Pass destination bucket as orig_volume through create_bitrot_writer_with_orig_volume into LocalDisk::create_file so single-shot non-inline part.1 writes under syncs_data_shards() durability run fdatasync before fadvise(POSIX_FADV_DONTNEED) and skip the duplicate sync_dir_files_std(src_data_path) in FusedCommitAttempt without fsyncing the temporary parent directory (warp PUT 1MiB: 658.3 vs MinIO 656.8 obj/s, 1.00x parity, p50 26.10ms vs MinIO 46.60ms).
- Add single_pool() fast paths to ECStore::handle_copy_object and ECStore::handle_new_multipart_upload, skip no-op destination prelookup get_object_info when put_like_write_creates_new_version is true, and backfill previous_current_size from OldCurrentSize in copy_object (CopyObject 1KB: 398.4 -> 1,086.1 ops/s, 2.73x faster, 1.01x faster than MinIO).

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Copilot-Session: d85aaf47-2a4b-4c2c-97a3-dbe8c7717e2a

* fix(ecstore): provide non-Linux has_rename_data_test_failpoints stub for Windows CI

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Copilot-Session: d85aaf47-2a4b-4c2c-97a3-dbe8c7717e2a

* fix(ecstore): preserve presynced shard directory durability

Sync the shard directory before publishing presynced payloads and keep the file sync test probe available on non-Linux targets.

* fix(ecstore,s3): fix non-consuming crash_inject check, test hooks, and rubber-duck findings (#9038)

- Add non-consuming crash_inject::is_armed(point, key) so FusedCommitAttempt and LocalDisk::update_metadata do not consume one-shot crash arms before the fallback path runs (fixes crash_consistency::* and meta_write_crash_consistency::* tests).
- Check INLINE_PREPARATION_BEFORE_BACKUP and RENAME_DATA_AFTER_FIRST_PUBLICATION in has_rename_data_test_failpoints so inline preparation admission and observed-owner panic tests run on their expected path.
- Resolve staged_dir via get_object_path_for_io in test_rename_data_presynced_shard_keeps_directory_sync and avoid premature orig_volume access check during tmp staging in put_object (fixes put_object_failure_cleans_tmp_workspace_inline).
- Address rubber-duck review findings: fall back from FusedCommitAttempt on any existing xl.meta when sync_commit_metadata is true, fsync object dir after activating delete rollback dir before unlinking xl.meta, classify data-dir entries before removal in fused delete finalization, include self_copy_lock in copy_prelookup_required, and avoid .expect() on non-test mutexes.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Copilot-Session: d85aaf47-2a4b-4c2c-97a3-dbe8c7717e2a

* fix(ecstore): defer shard sync until destination is known

* test(rustfs): cover legal hold replication fast path

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

---------

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Hauser <housemecn@gmail.com>
Co-authored-by: zhi22915 <qiuzgang@gmail.com>
Co-authored-by: Chris An <anzhengchao@gmail.com>
Co-authored-by: heihutu <heihutu@gmail.com>
Copilot-Session: d85aaf47-2a4b-4c2c-97a3-dbe8c7717e2a
2026-10-30 23:20:45 +08:00
Damien Degois 0d2bf69e2d fix(s3select): case-insensitive enums, LINES key order, path names (#9021)
* fix(s3select): match request enum values case-insensitively

ExpressionType, CompressionType, CSV FileHeaderInfo, JSON Type and CSV
QuoteFields were compared exactly, so lowercase values such as
`lines`, `document` or `gzip` were rejected. MinIO accepts any case.

Match them ASCII case-insensitively in request validation and rewrite
them to the canonical uppercase spelling, so every later comparison
keeps working unchanged. Unknown values keep their existing errors.

* fix(s3select): keep source key order for JSON LINES input

Arrow's JSON schema inference walks each record as a serde_json::Map,
which sorts keys unless serde_json's process-wide preserve_order
feature is on. JSON LINES `SELECT *` therefore returned keys sorted,
and CSV output columns were permuted.

Infer the LINES schema from the same object bytes and record limit as
DataFusion's JsonFormat, keep Arrow's types, and reorder fields
(nested structs included) by first-seen key order. Enabling
preserve_order globally was rejected: it changes map ordering for the
whole binary.

Drop the test-only preserve_order dependency so the crate's tests run
with production features; the DOCUMENT inference reference now
compares against Arrow's types in source order.

* fix(s3select): name nested path projections after their last key

`SELECT s.obj0.id` returned the key `s.obj0[id]`, DataFusion's display
name for the field access. AWS names an unaliased path projection after
its last key, e.g. `{"id": ...}`.

Alias such projections to their last key when that name is unique among
the output and ORDER BY column names. Queries with a wildcard keep
today's names, since the expanded columns could collide.

* fix(s3select): preserve names beside computed projections

* fix(s3select): preserve names for ambiguous path projections

* fix(s3select): name path projections from the planned schema

Path projections that the SQL rewrite leaves unaliased are now named
after their last key once DataFusion has planned the query. A path takes
its key only if no other output column keeps or receives that name. A
colliding rename is dropped and the check repeats, so the final names
are unique whenever DataFusion's own names are.

The SQL rewrite is unchanged, so GROUP BY can still refer to the names
it assigns and every query that planned before still plans.

Review cases:
- s.obj0.id, s.obj1.id, s.obj2."s.obj0[id]" keeps the three generated
  names: the third key would duplicate the first retained name.
- s.obj."lower(s.id)", LOWER(s.id), in either order, keeps the generated
  path name: its key is the computed column's name.

Paths beside computed columns, beside duplicated keys, and in
parentheses now take their key, as AWS S3 Select names them. Paths stay
unnamed when an ORDER BY expression is not projected, as before.

---------

Co-authored-by: Hauser <housemecn@gmail.com>
Co-authored-by: overtrue <anzhengchao@gmail.com>
Co-authored-by: GatewayJ <835269233@qq.com>
2026-10-30 23:19:18 +08:00
Mathew Beane 7e536843ea fix(audit): emit a bounded audit entry for authorization denials (#9073)
A request refused by the authorization layer was logged as
s3_authorization_denied but never reached the audit stream. Emit an audit
entry from DenialContext::deny with the account, bucket, object, denied
action and denial reason. Entries are bounded by a per-class token bucket
(lower for anonymous callers) with a suppressed count carried on the next
entry, quiet probe denials are skipped, DeleteObjects audits its refused
keys as one aggregate entry, and the S3 response is unchanged.
2026-10-30 23:18:42 +08:00
唐小鸭 7672867476 fix(replication): count force-delete outcomes in replication stats (#9072)
A replicated force-delete that a target rejected (for example a peer whose
credentials lack s3:ForceDeleteObject) was only logged; the per-target
replication counters never saw it, so `replication status` reported zero
failures while the sites had already diverged.

Build a ReplicatedTargetInfo for every target outcome of
replicate_force_delete_to_targets (missing client, offline target,
rejected remove, success) with a Pending previous status, and feed each
through ReplicationStats::update so rejections count as failures and
accepted force-deletes count as replicated.

Document that the managed site-replicator-0 policy must name
s3:ForceDeleteObject explicitly and that a bucket-replication target with
its own credentials needs the same grant.
2026-10-30 18:00:03 +08:00
Timo Otten c8f7f02817 fix(site-replication): grant replicator s3:ForceDeleteObject (#9069)
Co-authored-by: 唐小鸭 <tangtang1251@qq.com>
2026-10-30 17:59:36 +08:00
Hauser Zhi dcbec68d45 fix: preserve PUT lock failure diagnostics (#9062)
* fix: preserve PUT lock failure diagnostics

Propagate the request operation into spawned PUT commit tasks so lock failures retain their metric attribution. Record bounded lock error classifications without exposing lock resources.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* test: remove redundant clone in PUT lock test

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* fix: bound PUT commit context stack usage

Pin the commit future before wrapping it with request context to reduce stack pressure in PUT callers.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

---------

Co-authored-by: heihutu <heihutu@gmail.com>
Co-authored-by: zhi22915 <qiuzgang@gmail.com>
2026-10-30 14:59:20 +08:00
hector 589b499e3b ci(ft,heal,chain): FT gate 54, heal budget, and a temporary performance lane disable (#9057)
* ci(ft): update the lane gate to the 54-case manifest with scenario F

auto-testing #118/#119 added FT scenario F and case B5, so the lane now
runs 54 cases over scenarios A,B,C,C2,D,E,F and passes 54/54 — but the
workflow gate still pinned the 44-case A-E manifest and failed the
report step. Align the gate and the contract-test fixture with the
54-case manifest.

* ci(heal): give heal a 90-minute run budget inside a 120-minute job

The heal task was still progressing (scanned 494 -> 571+, failed=0)
when the 45-minute step cap killed it, so the verification steps never
ran and the lane produced invalid evidence (chain 40796474356). The
step budget is now 90 minutes, and the job-level budget rises from 60
to 120 minutes so the enclosing job cannot cancel the raised step
budget. The timeout contract now records heal as the exception to the
shared 60/45 baseline.

* ci(chain): force the performance probe offline while pf-testing is broken

The pf-testing environment is temporarily broken, so every nightly
chain would end red in the performance lane. performance is
required=false in scheduled-validations.json, so forcing the probe's
performance_ready output to offline takes the chain through its
designed skip path: the lane job is skipped, aggregate receives
--allow-skipped performance, and the report records the optional-lane
skip with its reason. Marked TEMPORARY; restore the probe once the
environment is healthy again.

* test(table): drop a stale workflow-comment pin left by #9059

#9059 rewrote the table suite step comments but kept the assertion for
the old '# Red only for harness/environment breakdowns' line, so
test_security_workflow.py fails against current main. The semantics
stay pinned by the continue-on-error and HARNESS_OK assertions.
2026-10-30 14:03:22 +08:00
50 changed files with 4738 additions and 401 deletions
@@ -197,21 +197,21 @@ jobs:
jq -e --arg testing_sha "${AUTO_TESTING_SHA}" '
.schema_version == 1 and
.auto_testing_sha == $testing_sha and
.selected_scenarios == ["A", "B", "C", "C2", "D", "E"] and
.selected_scenarios == ["A", "B", "C", "C2", "D", "E", "F"] and
.complete == true and
.counts.expected == 44 and
.counts.seen == 44 and
.counts.expected == 54 and
.counts.seen == 54 and
.counts.harness_errors == 0 and
(.expected_cases | length) == 44 and
(.expected_cases | unique | length) == 44 and
(.seen_cases | length) == 44 and
(.seen_cases | unique | length) == 44 and
(.expected_cases | length) == 54 and
(.expected_cases | unique | length) == 54 and
(.seen_cases | length) == 54 and
(.seen_cases | unique | length) == 54 and
(.seen_cases | sort) == (.expected_cases | sort) and
(.missing_cases | length) == 0 and
(.harness_errors | length) == 0 and
(.cases | length) == 44 and
(.cases | length) == 54 and
([.cases[].case_id] | sort) == (.expected_cases | sort) and
([.cases[].case_id] | unique | length) == 44 and
([.cases[].case_id] | unique | length) == 54 and
all(.cases[]; .verdict == "pass" or .verdict == "known-divergence" or .verdict == "UNEXPECTED") and
.counts.unexpected == ([.cases[] | select(.verdict == "UNEXPECTED")] | length) and
.counts.known_divergence == ([.cases[] | select(.verdict == "known-divergence")] | length)
+14 -6
View File
@@ -281,12 +281,20 @@ jobs:
GH_TOKEN: ${{ secrets.PF_TESTING_GH_TOKEN }}
PERFORMANCE_OPTIONAL: ${{ contains(fromJSON(steps.selection.outputs.optional_suites), 'performance') }}
run: |
if python3 scripts/check_functional_runners.py pf-testing; then
echo "performance_ready=online" >> "$GITHUB_OUTPUT"
else
echo "performance_ready=offline" >> "$GITHUB_OUTPUT"
test "$PERFORMANCE_OPTIONAL" = true
fi
# TEMPORARY: the pf-testing environment is broken, so the
# performance lane is disabled until it is repaired. Forcing
# offline keeps the chain on its designed skip path: the lane job
# is skipped, aggregate receives --allow-skipped performance
# (required=false in scheduled-validations.json), and the report
# records the optional-lane skip with its reason.
# Restore the probe below once the environment is healthy again.
# if python3 scripts/check_functional_runners.py pf-testing; then
# echo "performance_ready=online" >> "$GITHUB_OUTPUT"
# else
# echo "performance_ready=offline" >> "$GITHUB_OUTPUT"
# test "$PERFORMANCE_OPTIONAL" = true
# fi
echo "performance_ready=offline" >> "$GITHUB_OUTPUT"
validation:
if: ${{ github.event_name == 'workflow_dispatch' && !inputs.validation_runner_inspect_only && inputs.validation_retain_evidence_run_id == '' && inputs.validation_package_sha == '' && (inputs.validation_case != '' || inputs.validation_tools_only || inputs.validation_native_tools || inputs.validation_shared_tools || inputs.validation_manager_tools || inputs.validation_cleanup_run_id != '' || inputs.validation_cleanup_retained_evidence) }}
+3 -2
View File
@@ -55,7 +55,8 @@ env:
jobs:
heal-test:
runs-on: smoke-testing
timeout-minutes: 60
# 90-minute heal run (see the step below) plus setup, report and cleanup.
timeout-minutes: 120
# Standalone manual run, or one link of the nightly functional chain
# (storage -> heal -> pool). Pool expansion no longer re-runs heal.
if: ${{ inputs.chain_manifest != '' || github.event_name == 'workflow_dispatch' }}
@@ -154,7 +155,7 @@ jobs:
./auto-testing/rustfs_heal_test.sh "${ARGS[@]}" --log-file "${LOG_FILE}"
- name: Run heal test (write -> outage -> heal -> verify)
timeout-minutes: 45
timeout-minutes: 90
id: test
# Collect reports and backlog evidence before the final suite verdict.
continue-on-error: true
+12
View File
@@ -30,6 +30,16 @@ pub const COMMENT_KEY: &str = "comment";
pub const ENV_AUDIT_ENABLE: &str = "RUSTFS_AUDIT_ENABLE";
/// Global switch for enabling the notify module.
pub const ENV_NOTIFY_ENABLE: &str = "RUSTFS_NOTIFY_ENABLE";
/// Maximum audit entries per second for authorization denials of authenticated callers.
/// `0` disables denial auditing for that class of caller.
pub const ENV_AUDIT_DENIAL_RATE: &str = "RUSTFS_AUDIT_DENIAL_RATE";
/// Maximum audit entries per second for authorization denials of anonymous callers.
/// `0` disables denial auditing for that class of caller.
pub const ENV_AUDIT_DENIAL_ANONYMOUS_RATE: &str = "RUSTFS_AUDIT_DENIAL_ANONYMOUS_RATE";
/// Default denial audit rate for authenticated callers (entries per second).
pub const DEFAULT_AUDIT_DENIAL_RATE: u64 = 50;
/// Default denial audit rate for anonymous callers (entries per second).
pub const DEFAULT_AUDIT_DENIAL_ANONYMOUS_RATE: u64 = 5;
/// Default global audit switch (disabled by default).
pub const DEFAULT_AUDIT_ENABLE: bool = false;
/// Default global notify switch (disabled by default).
@@ -328,5 +338,7 @@ mod tests {
fn test_global_audit_notify_switch_constants() {
assert_eq!(ENV_AUDIT_ENABLE, "RUSTFS_AUDIT_ENABLE");
assert_eq!(ENV_NOTIFY_ENABLE, "RUSTFS_NOTIFY_ENABLE");
assert_eq!(ENV_AUDIT_DENIAL_RATE, "RUSTFS_AUDIT_DENIAL_RATE");
assert_eq!(ENV_AUDIT_DENIAL_ANONYMOUS_RATE, "RUSTFS_AUDIT_DENIAL_ANONYMOUS_RATE");
}
}
+6 -2
View File
@@ -1503,7 +1503,11 @@ pub(crate) async fn load_bucket_metadata_parse_with_presence(
bucket: &str,
parse: bool,
) -> Result<(BucketMetadata, bool)> {
let (mut bm, persisted) = match read_bucket_metadata(api.clone(), bucket).await {
let (metadata_res, incarnation_res) = tokio::join!(
Box::pin(read_bucket_metadata(api.clone(), bucket)),
Box::pin(load_bucket_incarnation(api.clone(), bucket)),
);
let (mut bm, persisted) = match metadata_res {
Ok(res) => (res, true),
Err(err) => {
if err != Error::ConfigNotFound {
@@ -1514,7 +1518,7 @@ pub(crate) async fn load_bucket_metadata_parse_with_presence(
}
};
let incarnation = load_bucket_incarnation(api.clone(), bucket).await?;
let incarnation = incarnation_res?;
if persisted {
if bm.name.is_empty() {
bm.name = bucket.to_string();
+142 -21
View File
@@ -1435,6 +1435,7 @@ pub(crate) async fn get_object_lock_config_state_in(
/// Re-read the Object Lock state and bucket incarnation from the same
/// authoritative metadata blob while the caller holds the transaction lock.
#[cfg(all(test, feature = "test-util"))]
pub(crate) async fn get_object_lock_config_and_incarnation_from_disk_in(
ctx: &crate::runtime::instance::InstanceContext,
bucket: &str,
@@ -1453,6 +1454,27 @@ pub(crate) async fn get_object_lock_config_and_incarnation_from_disk_with_guard_
get_delete_config_from_disk_in(ctx, bucket, guard, None).await
}
#[allow(clippy::type_complexity)]
pub(crate) async fn get_object_lock_and_quota_config_from_disk_with_guard_in(
ctx: &crate::runtime::instance::InstanceContext,
bucket: &str,
guard: Option<&rustfs_lock::NamespaceLockGuard>,
) -> Result<(
ObjectLockConfigState,
Uuid,
OffsetDateTime,
std::result::Result<(Option<BucketQuota>, Uuid, OffsetDateTime), String>,
)> {
let (state, incarnation_id, updated_at, quota_res) =
get_delete_and_quota_config_from_disk_in(ctx, bucket, guard, None, true).await?;
Ok((
state,
incarnation_id,
updated_at,
quota_res.unwrap_or(Ok((None, incarnation_id, updated_at))),
))
}
/// Validate queued expiration against the same fresh metadata used for Object
/// Lock admission. The caller holds the transaction read lock through commit,
/// so a successful lifecycle update cannot be followed by an old-policy delete.
@@ -1462,6 +1484,24 @@ pub(crate) async fn get_delete_config_from_disk_in(
guard: Option<&rustfs_lock::NamespaceLockGuard>,
expiry_policy: Option<&rustfs_lifecycle::ExpiryPolicySnapshot>,
) -> Result<(ObjectLockConfigState, Uuid, OffsetDateTime)> {
let (state, incarnation_id, updated_at, _) =
get_delete_and_quota_config_from_disk_in(ctx, bucket, guard, expiry_policy, false).await?;
Ok((state, incarnation_id, updated_at))
}
#[allow(clippy::type_complexity)]
async fn get_delete_and_quota_config_from_disk_in(
ctx: &crate::runtime::instance::InstanceContext,
bucket: &str,
guard: Option<&rustfs_lock::NamespaceLockGuard>,
expiry_policy: Option<&rustfs_lifecycle::ExpiryPolicySnapshot>,
include_quota: bool,
) -> Result<(
ObjectLockConfigState,
Uuid,
OffsetDateTime,
Option<std::result::Result<(Option<BucketQuota>, Uuid, OffsetDateTime), String>>,
)> {
let bucket_meta_sys_lock = bucket_metadata_sys_of(ctx)?;
let bucket_meta_sys = bucket_meta_sys_lock.read().await.clone();
@@ -1477,7 +1517,7 @@ pub(crate) async fn get_delete_config_from_disk_in(
.await?
}
};
match authority {
match &authority {
BucketMetadataAuthority::Authoritative(metadata)
if metadata.bucket_incarnation_sidecar && !metadata.bucket_incarnation_id.is_nil() =>
{
@@ -1489,11 +1529,11 @@ pub(crate) async fn get_delete_config_from_disk_in(
}
}
let object_lock_config_updated_at = metadata.object_lock_config_updated_at;
Ok((
object_lock_config_state_from_authoritative_metadata(&metadata)?,
metadata.bucket_incarnation_id,
object_lock_config_updated_at,
))
let state = object_lock_config_state_from_authoritative_metadata(metadata)?;
let incarnation_id = metadata.bucket_incarnation_id;
let quota_res = include_quota
.then(|| quota_config_and_incarnation_from_authority(bucket, authority).map_err(|err| err.to_string()));
Ok((state, incarnation_id, object_lock_config_updated_at, quota_res))
}
BucketMetadataAuthority::Authoritative(_) => {
Err(Error::other(format!("bucket incarnation metadata is not authoritative: {bucket}")))
@@ -1564,6 +1604,23 @@ pub(crate) async fn get_quota_config_and_incarnation_from_disk_for_options_in(
return Err(Error::other("quota snapshot metadata transaction fence was released"));
}
#[cfg(test)]
let has_injected_error = sys.object_lock_disk_read_errors.read().await.contains(bucket);
#[cfg(not(test))]
let has_injected_error = false;
if !ctx.lock_manager().is_disabled()
&& !has_injected_error
&& let Some(fresh_quota) = snapshot.take_fresh_put_quota()
{
if snapshot_guard.is_lock_lost()
|| snapshot_guard.is_released()
|| !snapshot.is_valid_for_destructive_put(store.id, bucket, opts.expected_bucket_incarnation_id.unwrap_or_default())
{
return Err(Error::other("quota snapshot fence or bucket incarnation changed during the disk read"));
}
return fresh_quota.map_err(Error::other);
}
// The snapshot carries the caller's lifecycle fence, excluding deletion.
// Its transaction guard excludes config writes, so a same-name object
// owner need not recursively acquire the bucket namespace here.
@@ -1750,9 +1807,22 @@ struct MetadataPublishGuard {
_guard: tokio::sync::OwnedMutexGuard<MetadataPublishLockState>,
}
static LOCAL_META_MUTATION_GEN: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1);
pub(crate) fn bump_local_meta_mutation_gen() {
LOCAL_META_MUTATION_GEN.fetch_add(1, std::sync::atomic::Ordering::Release);
}
pub(crate) fn current_local_meta_mutation_gen() -> u64 {
LOCAL_META_MUTATION_GEN.load(std::sync::atomic::Ordering::Acquire)
}
type LocalAuthorityCache = Arc<StdMutex<HashMap<String, (u64, Arc<BucketMetadata>)>>>;
#[derive(Debug, Clone)]
pub struct BucketMetadataSys {
metadata_map: Arc<RwLock<HashMap<String, Arc<BucketMetadata>>>>,
local_authority_cache: LocalAuthorityCache,
/// Serializes metadata-map commits and their derived cache updates for one
/// bucket. Namespace locks, when present, are acquired before this lock.
metadata_publish_locks: Arc<MetadataPublishLockRegistry>,
@@ -1796,6 +1866,7 @@ impl BucketMetadataSys {
pub fn new(api: Arc<ECStore>) -> Self {
Self {
metadata_map: Arc::new(RwLock::new(HashMap::new())),
local_authority_cache: Arc::new(StdMutex::new(HashMap::new())),
metadata_publish_locks: Arc::new(MetadataPublishLockRegistry {
locks: StdMutex::new(HashMap::new()),
}),
@@ -2160,6 +2231,7 @@ impl BucketMetadataSys {
}
pub async fn set(&self, bucket: String, bm: Arc<BucketMetadata>) {
bump_local_meta_mutation_gen();
if !is_meta_bucketname(&bucket) {
let publish_lock = self.metadata_publish_lock(&bucket);
let _publish_guard = publish_lock.lock().await;
@@ -2178,6 +2250,7 @@ impl BucketMetadataSys {
///
/// Returns `true` if an entry was present. Reserved meta buckets are ignored.
pub async fn remove(&self, bucket: &str) -> bool {
bump_local_meta_mutation_gen();
if is_meta_bucketname(bucket) {
return false;
}
@@ -2198,6 +2271,7 @@ impl BucketMetadataSys {
}
async fn _reset(&mut self) {
bump_local_meta_mutation_gen();
let mut map = self.metadata_map.write().await;
map.clear();
drop(map);
@@ -3001,30 +3075,77 @@ impl BucketMetadataSys {
return Err(Error::other(format!("injected Object Lock metadata disk read failure: {bucket}")));
}
match await_bucket_namespace_operation(Some(guard), bucket, "bucket metadata snapshot existence check", async {
self.object_store()
.get_bucket_info_from_sets_at_read_quorum(bucket, &crate::storage_api_contracts::bucket::BucketOptions::default())
.await
})
.await
{
let store = self.object_store();
let can_use_local_authority_cache = !store.ctx.lock_manager().is_disabled() && store.all_disks_local_and_online();
let gen_before = current_local_meta_mutation_gen();
let cached_authority = can_use_local_authority_cache
.then(|| {
self.local_authority_cache
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.get(bucket)
.filter(|(cached_gen, _)| *cached_gen == gen_before)
.map(|(_, bm)| Arc::clone(bm))
})
.flatten();
if let Some(cached) = cached_authority {
await_bucket_namespace_operation(
Some(guard),
bucket,
"bucket metadata authoritative snapshot",
std::future::ready(Ok::<_, Error>(())),
)
.await?;
if current_local_meta_mutation_gen() == gen_before && store.all_disks_local_and_online() {
return Ok(BucketMetadataAuthority::Authoritative(cached));
}
}
let (bucket_info_res, metadata_res) = tokio::join!(
Box::pin(await_bucket_namespace_operation(
Some(guard),
bucket,
"bucket metadata snapshot existence check",
async {
store
.get_bucket_info_from_sets_at_read_quorum(
bucket,
&crate::storage_api_contracts::bucket::BucketOptions::default(),
)
.await
},
)),
Box::pin(await_bucket_namespace_operation(
Some(guard),
bucket,
"bucket metadata authoritative snapshot",
load_bucket_metadata_parse_with_presence(Arc::clone(&store), bucket, true),
)),
);
match bucket_info_res {
Ok(_) => {}
Err(Error::VolumeNotFound) => return Ok(BucketMetadataAuthority::MissingBucket),
Err(err) => return Err(err),
}
let (metadata, persisted) = await_bucket_namespace_operation(
Some(guard),
bucket,
"bucket metadata authoritative snapshot",
load_bucket_metadata_parse_with_presence(self.object_store(), bucket, true),
)
.await?;
let (metadata, persisted) = metadata_res?;
if persisted && metadata.lock_enabled && !metadata.bucket_creation_committed {
return Err(Error::ErasureWriteQuorum);
}
if persisted {
Ok(BucketMetadataAuthority::Authoritative(Arc::new(metadata)))
let metadata = Arc::new(metadata);
if can_use_local_authority_cache
&& metadata.bucket_incarnation_sidecar
&& !metadata.bucket_incarnation_id.is_nil()
&& current_local_meta_mutation_gen() == gen_before
&& store.all_disks_local_and_online()
{
self.local_authority_cache
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.insert(bucket.to_string(), (gen_before, Arc::clone(&metadata)));
}
Ok(BucketMetadataAuthority::Authoritative(metadata))
} else {
Ok(BucketMetadataAuthority::Fabricated)
}
@@ -3282,10 +3282,18 @@ async fn replicate_force_delete_to_targets<S: ReplicationStorage>(dobj: &Deleted
let mut join_set = JoinSet::new();
let mut all_succeeded = true;
let mut target_infos = Vec::with_capacity(tgt_arns.len());
for arn in tgt_arns {
let Some(tgt_client) = ReplicationTargetStore::remote_target_client(bucket, &arn).await else {
all_succeeded = false;
target_infos.push(force_delete_target_info(
dobj.op_type,
&arn,
"",
false,
Err(TARGET_CLIENT_UNAVAILABLE_ERROR.to_string()),
));
debug!(
event = EVENT_REPLICATION_FORCE_DELETE_SKIPPED,
component = LOG_COMPONENT_ECSTORE,
@@ -3311,6 +3319,7 @@ async fn replicate_force_delete_to_targets<S: ReplicationStorage>(dobj: &Deleted
let bucket = bucket.clone();
let object_name = object_name.clone();
let op_type = dobj.op_type;
join_set.spawn(async move {
if ReplicationTargetStore::target_is_offline(&tgt_client).await {
@@ -3335,9 +3344,16 @@ async fn replicate_force_delete_to_targets<S: ReplicationStorage>(dobj: &Deleted
user_agent: "Internal: [Replication]".to_string(),
..Default::default()
});
return false;
return force_delete_target_info(
op_type,
&tgt_client.arn,
&tgt_client.endpoint,
tgt_client.secure,
Err(FORCE_DELETE_TARGET_OFFLINE_ERROR.to_string()),
);
}
let started = Instant::now();
if let Err(e) = tgt_client
.remove_object(&tgt_client.bucket, &object_name, None, replication_force_delete_remove_options())
.await
@@ -3364,16 +3380,25 @@ async fn replicate_force_delete_to_targets<S: ReplicationStorage>(dobj: &Deleted
user_agent: "Internal: [Replication]".to_string(),
..Default::default()
});
return false;
return force_delete_target_info(
op_type,
&tgt_client.arn,
&tgt_client.endpoint,
tgt_client.secure,
Err(e.to_string()),
);
}
true
force_delete_target_info(op_type, &tgt_client.arn, &tgt_client.endpoint, tgt_client.secure, Ok(started.elapsed()))
});
}
while let Some(result) = join_set.join_next().await {
match result {
Ok(success) => all_succeeded &= success,
Ok(info) => {
all_succeeded &= info.replication_status == ReplicationStatusType::Completed;
target_infos.push(info);
}
Err(error) => {
all_succeeded = false;
error!(
@@ -3390,6 +3415,8 @@ async fn replicate_force_delete_to_targets<S: ReplicationStorage>(dobj: &Deleted
}
}
record_force_delete_target_stats(bucket, &target_infos).await;
if all_succeeded
&& let Some(operation_id) = dobj.delete_object.force_delete_id
&& let Err(error) = super::replication_pool::complete_force_delete_intent(storage, operation_id).await
@@ -3409,6 +3436,53 @@ async fn replicate_force_delete_to_targets<S: ReplicationStorage>(dobj: &Deleted
all_succeeded
}
const FORCE_DELETE_TARGET_OFFLINE_ERROR: &str = "replication target is offline";
/// One target's outcome of a replicated force-delete in the shape the
/// replication statistics consume. A force-delete carries no per-version
/// replication state, so the previous status is always `Pending`: that is
/// the transition `ReplicationStats::update` counts as a failure, whereas an
/// `Empty` previous status would drop the outcome and leave `replication
/// status` reporting zero failures while the peer still holds the objects.
fn force_delete_target_info(
op_type: ReplicationType,
arn: &str,
endpoint: &str,
secure: bool,
outcome: std::result::Result<std::time::Duration, String>,
) -> ReplicatedTargetInfo {
let mut rinfo = ReplicatedTargetInfo {
arn: arn.to_string(),
op_type,
endpoint: endpoint.to_string(),
secure,
prev_replication_status: ReplicationStatusType::Pending,
..Default::default()
};
match outcome {
Ok(duration) => {
rinfo.replication_status = ReplicationStatusType::Completed;
rinfo.duration = duration;
}
Err(error) => {
rinfo.replication_status = ReplicationStatusType::Failed;
rinfo.error = Some(error);
}
}
rinfo
}
async fn record_force_delete_target_stats(bucket: &str, targets: &[ReplicatedTargetInfo]) {
let Some(stats) = runtime_sources::replication_stats() else {
return;
};
for target in targets {
stats
.update(bucket, target, target.replication_status.clone(), target.prev_replication_status.clone())
.await;
}
}
fn unavailable_delete_target_info(dobj: &DeletedObjectReplicationInfo, arn: &str) -> ReplicatedTargetInfo {
let mut rinfo = dobj
.delete_object
@@ -7444,6 +7518,51 @@ mod tests {
server.join().expect("test HTTP server should finish");
}
/// A force-delete the peer rejects must reach the replication failure
/// counters. A replicated force-delete used to only log the rejection, so
/// `replication status` reported zero failures while the sites diverged.
#[tokio::test]
async fn force_delete_target_outcomes_reach_replication_stats() {
let stats = super::super::replication_state::ReplicationStats::new();
let denied = force_delete_target_info(
ReplicationType::Delete,
"arn:rustfs:replication::denied:bucket",
"http://peer-a:9000",
false,
Err("remove_object failed: AccessDenied".to_string()),
);
assert_eq!(denied.replication_status, ReplicationStatusType::Failed);
assert_eq!(denied.prev_replication_status, ReplicationStatusType::Pending);
assert_eq!(denied.error.as_deref(), Some("remove_object failed: AccessDenied"));
let accepted = force_delete_target_info(
ReplicationType::Heal,
"arn:rustfs:replication::accepted:bucket",
"https://peer-b:9000",
true,
Ok(std::time::Duration::from_millis(3)),
);
assert_eq!(accepted.replication_status, ReplicationStatusType::Completed);
assert!(accepted.error.is_none());
for target in [&denied, &accepted] {
stats
.update(
"bucket",
target,
target.replication_status.clone(),
target.prev_replication_status.clone(),
)
.await;
}
let bucket_stats = stats.get("bucket").await;
let denied_stat = bucket_stats.stats.get(&denied.arn).expect("denied target: stats entry");
assert_eq!((denied_stat.failed.count, denied_stat.replicated_count), (1, 0));
let accepted_stat = bucket_stats.stats.get(&accepted.arn).expect("accepted target: stats entry");
assert_eq!((accepted_stat.failed.count, accepted_stat.replicated_count), (0, 1));
}
/// AWS answers `HEAD ?versionId=<source marker id>` with a bodiless 400.
/// The marker is verified through the latest version instead: a
/// versionless 404 with `x-amz-delete-marker: true` proves it propagated.
+11 -1
View File
@@ -84,7 +84,7 @@ pub(crate) enum CrashPoint {
}
#[cfg(test)]
pub(crate) use armed::{arm, disarm, should_crash_at};
pub(crate) use armed::{arm, disarm, is_armed, should_crash_at};
/// Production no-op: the compiler inlines this to `false`, so armed crash points
/// exist only in `#[cfg(test)]` builds and never add a branch to real commits.
@@ -94,6 +94,12 @@ pub(crate) fn should_crash_at(_point: CrashPoint, _key: &str) -> bool {
false
}
#[cfg(not(test))]
#[inline(always)]
pub(crate) fn is_armed(_point: CrashPoint, _key: &str) -> bool {
false
}
#[cfg(test)]
mod armed {
use super::CrashPoint;
@@ -132,6 +138,10 @@ mod armed {
guard().retain(|(p, k)| !(*p == point && k == key));
}
pub(crate) fn is_armed(point: CrashPoint, key: &str) -> bool {
guard().iter().any(|(p, k)| *p == point && k == key)
}
/// Returns `true` (consuming the arm) iff a crash is armed for exactly this
/// `point` and `key`.
pub(crate) fn should_crash_at(point: CrashPoint, key: &str) -> bool {
+45
View File
@@ -2327,6 +2327,51 @@ impl LocalDiskWrapper {
.await
}
pub(crate) async fn try_publish_prepared_delete_local(
&self,
volume: &str,
object: &str,
transaction: uuid::Uuid,
namespace_owner: Option<Arc<dyn Send + Sync>>,
) -> Result<Option<(crate::disk::DeleteRecoveryJournal, bool, bool)>> {
self.track_disk_health_mutation(
"settle_delete_transaction",
DiskMetricMutation::Delete,
|| {
Box::pin(async {
self.disk
.try_publish_prepared_delete_local(volume, object, transaction, namespace_owner)
.await
})
},
crate::disk::get_delete_recovery_timeout()?,
)
.await
}
pub(crate) async fn try_finish_and_finalize_prepared_delete_local(
&self,
volume: &str,
object: &str,
transaction: uuid::Uuid,
journal: &crate::disk::DeleteRecoveryJournal,
namespace_owner: Option<Arc<dyn Send + Sync>>,
) -> Result<bool> {
self.track_disk_health_mutation(
"settle_delete_transaction",
DiskMetricMutation::Delete,
|| {
Box::pin(async {
self.disk
.try_finish_and_finalize_prepared_delete_local(volume, object, transaction, journal, namespace_owner)
.await
})
},
crate::disk::get_delete_recovery_timeout()?,
)
.await
}
/// Track disk health for an operation.
/// This method should wrap disk operations to ensure health checking.
pub async fn track_disk_health<T, F, Fut>(&self, operation: F, timeout_duration: Duration) -> Result<T>
+1
View File
@@ -304,6 +304,7 @@ pub async fn cached_access(path: impl AsRef<Path>) -> io::Result<()> {
}
pub fn invalidate_bucket_cache(path: impl AsRef<Path>) {
crate::bucket::metadata_sys::bump_local_meta_mutation_gen();
BUCKET_EXISTENCE_CACHE.invalidate(path.as_ref());
}
+668 -47
View File
@@ -74,8 +74,8 @@ use crate::disk::{
error_conv::{to_access_error, to_file_error, to_unformatted_disk_error, to_volume_error},
format::FormatV3,
fs::{
O_APPEND, O_CREATE, O_RDONLY, O_TRUNC, O_WRONLY, access, cached_access, invalidate_bucket_cache, lstat, lstat_std,
remove, remove_all_std, remove_std, rename,
O_APPEND, O_CREATE, O_RDONLY, O_WRONLY, access, cached_access, invalidate_bucket_cache, lstat, lstat_std, remove,
remove_all_std, remove_std, rename,
},
is_quota_mutation_fence_path, os,
os::{check_path_length, is_dir_not_empty_error, is_empty_dir, is_root_disk, rename_all, rename_all_ignore_missing_source},
@@ -2985,6 +2985,51 @@ fn has_inline_commit_test_failpoints(dst_path: &str) -> bool {
== Some(dst_path)
}
#[cfg(all(test, target_os = "linux"))]
pub(super) fn has_rename_data_test_failpoints(dst_path: &str) -> bool {
has_inline_commit_test_failpoints(dst_path)
|| INLINE_PREPARATION_BEFORE_BACKUP
.lock()
.expect("test preparation hook lock should not be poisoned")
.contains_key(dst_path)
|| !RENAME_DATA_AFTER_FIRST_PUBLICATION
.lock()
.expect("test publication hook lock should not be poisoned")
.is_empty()
|| RENAME_DATA_FAIL_BEFORE_OLD_METADATA_BACKUP
.lock()
.expect("test failpoint lock should not be poisoned")
.as_deref()
== Some(dst_path)
|| RENAME_DATA_FAIL_AFTER_METADATA_COMMIT
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.iter()
.any(|target| target == dst_path)
|| RENAME_DATA_FAIL_COMMIT_RENAME
.lock()
.expect("test failpoint lock should not be poisoned")
.as_deref()
== Some(dst_path)
|| os::has_dir_fsync_test_failures()
|| os::file_sync_probe::is_active()
}
#[cfg(all(not(test), feature = "test-util", target_os = "linux"))]
pub(super) fn has_rename_data_test_failpoints(dst_path: &str) -> bool {
RENAME_DATA_FAIL_AFTER_METADATA_COMMIT
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.iter()
.any(|target| target == dst_path)
|| os::has_dir_fsync_test_failures()
}
#[cfg(any(not(any(test, feature = "test-util")), not(target_os = "linux")))]
pub(super) fn has_rename_data_test_failpoints(_dst_path: &str) -> bool {
false
}
#[cfg(test)]
fn run_inline_preparation_before_backup(dst_path: &str) {
let hook = INLINE_PREPARATION_BEFORE_BACKUP
@@ -3056,6 +3101,43 @@ fn should_fail_after_delete_data_staged(path: &str) -> bool {
}
}
#[cfg(test)]
pub(crate) fn has_delete_version_test_failpoints() -> bool {
#[cfg(not(windows))]
if os::prepared_publication_test_hooks::has_any_hooks() {
return true;
}
#[cfg(feature = "e2e-test-hooks")]
if std::env::var_os("RUSTFS_DELETE_RECOVERY_BARRIER_DIR").is_some() {
return true;
}
!OWNED_FILE_WRITE_BEFORE_OPEN
.lock()
.expect("test file write hook lock should not be poisoned")
.is_empty()
|| !DELETE_VERSION_FAIL_AFTER_DATA_STAGED
.lock()
.expect("test failpoint lock should not be poisoned")
.is_empty()
|| !DELETE_VERSION_FAIL_AFTER_COMMIT
.lock()
.expect("test failpoint lock should not be poisoned")
.is_empty()
}
#[cfg(not(test))]
pub(crate) fn has_delete_version_test_failpoints() -> bool {
#[cfg(all(feature = "test-util", not(windows)))]
if os::prepared_publication_test_hooks::has_any_hooks() {
return true;
}
#[cfg(feature = "e2e-test-hooks")]
if std::env::var_os("RUSTFS_DELETE_RECOVERY_BARRIER_DIR").is_some() {
return true;
}
false
}
#[cfg(test)]
fn should_fail_after_delete_commit(root: &Path, path: &str) -> bool {
let mut targets = DELETE_VERSION_FAIL_AFTER_COMMIT
@@ -3590,6 +3672,283 @@ impl AsyncRead for FileCacheReclaimReader {
}
}
#[cfg(target_os = "linux")]
fn reclaim_std_file_cache(file: &std::fs::File, reclaim_len: usize, reclaim_on_shutdown: bool) {
use core::num::NonZeroU64;
use rustix::fs::{Advice, fadvise};
if !reclaim_on_shutdown || reclaim_len == 0 {
return;
}
let started = std::time::Instant::now();
if let Some(len) = u64::try_from(reclaim_len).ok().and_then(NonZeroU64::new) {
if let Err(err) = fadvise(file, 0, Some(len), Advice::DontNeed) {
record_file_cache_reclaim_error("write");
debug!(error = ?err, reclaim_len, "failed to reclaim file cache after write");
} else {
record_file_cache_reclaim_success("write", reclaim_len, started);
}
}
}
#[cfg(not(target_os = "linux"))]
fn reclaim_std_file_cache(_file: &std::fs::File, _reclaim_len: usize, _reclaim_on_shutdown: bool) {}
static PRESYNCED_TMP_DATA_DIRS: std::sync::LazyLock<ParkingLotMutex<HashSet<PathBuf>>> =
std::sync::LazyLock::new(|| ParkingLotMutex::new(HashSet::new()));
fn mark_presynced_tmp_data_dir(dir: PathBuf) {
let mut set = PRESYNCED_TMP_DATA_DIRS.lock();
if set.len() >= 4096 {
set.clear();
}
set.insert(dir);
}
fn unmark_presynced_tmp_data_dir(dir: &Path) {
PRESYNCED_TMP_DATA_DIRS.lock().remove(dir);
}
#[cfg(target_os = "linux")]
pub(in crate::disk) fn take_presynced_tmp_data_dir(dir: &Path) -> bool {
PRESYNCED_TMP_DATA_DIRS.lock().remove(dir)
}
fn open_lazy_local_file_std(file_path: &Path, volume_dir: &Path, _reclaim_on_shutdown: bool) -> std::io::Result<std::fs::File> {
if let Some(parent) = file_path.parent()
&& (volume_dir.as_os_str().is_empty() || !volume_dir.starts_with(parent))
&& let Err(err) = super::fs::create_dir_all_std(parent)
&& err.kind() != std::io::ErrorKind::AlreadyExists
{
return Err(to_file_error(err));
}
let file = std::fs::OpenOptions::new()
.write(true)
.create(true)
.truncate(true)
.open(file_path)
.map_err(to_file_error)?;
#[cfg(target_os = "macos")]
if _reclaim_on_shutdown {
let _ = set_std_fd_nocache(&file);
}
Ok(file)
}
struct LazyLocalFileWriter {
file_path: PathBuf,
volume_dir: PathBuf,
reclaim_len: usize,
reclaim_on_shutdown: bool,
sync_on_close: bool,
bytes_written: usize,
state: LazyLocalFileWriterState,
}
enum LazyLocalFileWriterState {
Unopened,
Writing(tokio::task::JoinHandle<std::io::Result<(Option<std::fs::File>, usize)>>),
Open(std::fs::File),
Closing(tokio::task::JoinHandle<std::io::Result<()>>),
Closed,
}
impl LazyLocalFileWriter {
fn new(file_path: PathBuf, volume_dir: PathBuf, reclaim_len: usize, reclaim_on_shutdown: bool, sync_on_close: bool) -> Self {
Self {
file_path,
volume_dir,
reclaim_len,
reclaim_on_shutdown,
sync_on_close,
bytes_written: 0,
state: LazyLocalFileWriterState::Unopened,
}
}
fn start_write_payload(&mut self, payload: Vec<u8>) {
let len = payload.len();
let close_after = self.reclaim_len > 0 && self.bytes_written.saturating_add(len) >= self.reclaim_len;
let reclaim_len = self.bytes_written.saturating_add(len).max(self.reclaim_len);
let reclaim_on_shutdown = self.reclaim_on_shutdown;
let sync_on_close = self.sync_on_close;
match std::mem::replace(&mut self.state, LazyLocalFileWriterState::Closed) {
LazyLocalFileWriterState::Unopened => {
let file_path = self.file_path.clone();
let volume_dir = self.volume_dir.clone();
self.state = LazyLocalFileWriterState::Writing(tokio::task::spawn_blocking(move || {
let mut file = open_lazy_local_file_std(&file_path, &volume_dir, reclaim_on_shutdown)?;
std::io::Write::write_all(&mut file, &payload).map_err(to_file_error)?;
if close_after {
if sync_on_close {
file.sync_data().map_err(to_file_error)?;
}
reclaim_std_file_cache(&file, reclaim_len, reclaim_on_shutdown);
drop(file);
if sync_on_close && let Some(parent) = file_path.parent() {
mark_presynced_tmp_data_dir(parent.to_path_buf());
}
Ok((None, len))
} else {
Ok((Some(file), len))
}
}));
}
LazyLocalFileWriterState::Open(mut file) => {
let file_path = self.file_path.clone();
self.state = LazyLocalFileWriterState::Writing(tokio::task::spawn_blocking(move || {
std::io::Write::write_all(&mut file, &payload).map_err(to_file_error)?;
if close_after {
if sync_on_close {
file.sync_data().map_err(to_file_error)?;
}
reclaim_std_file_cache(&file, reclaim_len, reclaim_on_shutdown);
drop(file);
if sync_on_close && let Some(parent) = file_path.parent() {
mark_presynced_tmp_data_dir(parent.to_path_buf());
}
Ok((None, len))
} else {
Ok((Some(file), len))
}
}));
}
other => self.state = other,
}
}
fn poll_in_flight_write(&mut self, cx: &mut std::task::Context<'_>) -> std::task::Poll<std::io::Result<usize>> {
let LazyLocalFileWriterState::Writing(fut) = &mut self.state else {
return std::task::Poll::Ready(Ok(0));
};
match std::pin::Pin::new(fut).poll(cx) {
std::task::Poll::Pending => std::task::Poll::Pending,
std::task::Poll::Ready(Err(join_err)) => {
self.state = LazyLocalFileWriterState::Closed;
std::task::Poll::Ready(Err(std::io::Error::other(join_err)))
}
std::task::Poll::Ready(Ok(Err(err))) => {
self.state = LazyLocalFileWriterState::Closed;
std::task::Poll::Ready(Err(err))
}
std::task::Poll::Ready(Ok(Ok((maybe_file, written)))) => {
self.bytes_written = self.bytes_written.saturating_add(written);
self.state = maybe_file.map_or(LazyLocalFileWriterState::Closed, LazyLocalFileWriterState::Open);
std::task::Poll::Ready(Ok(written))
}
}
}
}
impl AsyncWrite for LazyLocalFileWriter {
fn poll_write(
mut self: std::pin::Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
buf: &[u8],
) -> std::task::Poll<std::io::Result<usize>> {
if matches!(self.state, LazyLocalFileWriterState::Writing(_)) {
return self.poll_in_flight_write(cx);
}
if buf.is_empty() {
return std::task::Poll::Ready(Ok(0));
}
if matches!(self.state, LazyLocalFileWriterState::Closed | LazyLocalFileWriterState::Closing(_)) {
return std::task::Poll::Ready(Err(std::io::Error::new(
std::io::ErrorKind::BrokenPipe,
"local file writer is already closed",
)));
}
self.start_write_payload(buf.to_vec());
self.poll_in_flight_write(cx)
}
fn poll_write_vectored(
mut self: std::pin::Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
bufs: &[std::io::IoSlice<'_>],
) -> std::task::Poll<std::io::Result<usize>> {
if matches!(self.state, LazyLocalFileWriterState::Writing(_)) {
return self.poll_in_flight_write(cx);
}
let total_len: usize = bufs.iter().map(|b| b.len()).sum();
if total_len == 0 {
return std::task::Poll::Ready(Ok(0));
}
if matches!(self.state, LazyLocalFileWriterState::Closed | LazyLocalFileWriterState::Closing(_)) {
return std::task::Poll::Ready(Err(std::io::Error::new(
std::io::ErrorKind::BrokenPipe,
"local file writer is already closed",
)));
}
let mut payload = Vec::with_capacity(total_len);
for slice in bufs {
payload.extend_from_slice(slice);
}
self.start_write_payload(payload);
self.poll_in_flight_write(cx)
}
fn is_write_vectored(&self) -> bool {
true
}
fn poll_flush(mut self: std::pin::Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> std::task::Poll<std::io::Result<()>> {
if matches!(self.state, LazyLocalFileWriterState::Writing(_)) {
std::task::ready!(self.poll_in_flight_write(cx))?;
}
std::task::Poll::Ready(Ok(()))
}
fn poll_shutdown(
mut self: std::pin::Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
) -> std::task::Poll<std::io::Result<()>> {
if matches!(self.state, LazyLocalFileWriterState::Writing(_)) {
std::task::ready!(self.poll_in_flight_write(cx))?;
}
if !matches!(self.state, LazyLocalFileWriterState::Closing(_)) {
match std::mem::replace(&mut self.state, LazyLocalFileWriterState::Closed) {
LazyLocalFileWriterState::Closed => return std::task::Poll::Ready(Ok(())),
LazyLocalFileWriterState::Unopened => {
let file_path = self.file_path.clone();
let volume_dir = self.volume_dir.clone();
let reclaim_on_shutdown = self.reclaim_on_shutdown;
self.state = LazyLocalFileWriterState::Closing(tokio::task::spawn_blocking(move || {
open_lazy_local_file_std(&file_path, &volume_dir, reclaim_on_shutdown).map(drop)
}));
}
LazyLocalFileWriterState::Open(file) => {
let file_path = self.file_path.clone();
let reclaim_len = self.bytes_written.max(self.reclaim_len);
let reclaim_on_shutdown = self.reclaim_on_shutdown;
let sync_on_close = self.sync_on_close;
self.state = LazyLocalFileWriterState::Closing(tokio::task::spawn_blocking(move || {
if sync_on_close {
file.sync_data().map_err(to_file_error)?;
}
reclaim_std_file_cache(&file, reclaim_len, reclaim_on_shutdown);
drop(file);
if sync_on_close && let Some(parent) = file_path.parent() {
mark_presynced_tmp_data_dir(parent.to_path_buf());
}
Ok(())
}));
}
other => {
self.state = other;
}
}
}
let LazyLocalFileWriterState::Closing(fut) = &mut self.state else {
return std::task::Poll::Ready(Ok(()));
};
let res = std::task::ready!(std::pin::Pin::new(fut).poll(cx));
self.state = LazyLocalFileWriterState::Closed;
std::task::Poll::Ready(res.unwrap_or_else(|join_err| Err(std::io::Error::other(join_err))))
}
}
#[cfg(test)]
impl FileCacheReclaimWriter {
fn new(inner: File, reclaim_len: usize, reclaim_on_shutdown: bool) -> Self {
#[cfg(target_os = "macos")]
@@ -3630,6 +3989,7 @@ impl FileCacheReclaimWriter {
}
}
#[cfg(test)]
impl AsyncWrite for FileCacheReclaimWriter {
fn poll_write(
mut self: std::pin::Pin<&mut Self>,
@@ -3727,6 +4087,7 @@ fn should_reclaim_file_cache_after_read(length: usize) -> bool {
#[derive(Clone, Copy, Debug)]
pub(crate) enum WriteMode {
Truncate { size_hint: i64 },
TruncateSync { size_hint: i64 },
Append,
}
@@ -4494,13 +4855,16 @@ impl LocalIoBackend for StdBackend {
async fn open_write(&self, volume: &str, path: &str, mode: WriteMode) -> Result<FileWriter> {
self.invalidate_cached_fd(volume, path).await;
match mode {
WriteMode::Truncate { size_hint } => {
WriteMode::Truncate { size_hint } | WriteMode::TruncateSync { size_hint } => {
let sync_on_close = matches!(mode, WriteMode::TruncateSync { .. });
let volume_dir = local_disk_bucket_path(self.io_root(), volume)?;
let file_path = local_disk_object_path(self.io_root(), volume, path)?;
check_path_length(file_path.to_string_lossy().as_ref())?;
if let Some(parent) = file_path.parent() {
os::make_dir_all(parent, &volume_dir).await?;
check_path_length(parent.to_string_lossy().as_ref())?;
if volume == RUSTFS_META_TMP_BUCKET {
unmark_presynced_tmp_data_dir(parent);
}
}
// O_DIRECT streaming write (Linux, opt-in): shard bytes stream
@@ -4511,6 +4875,9 @@ impl LocalIoBackend for StdBackend {
// EINVAL (which would masquerade as a missing shard).
#[cfg(target_os = "linux")]
if is_direct_io_write_enabled() && self.direct_io_write.supported.load(Ordering::Relaxed) {
if let Some(parent) = file_path.parent() {
os::make_dir_all(parent, &volume_dir).await?;
}
let write_state = self.direct_io_write.clone();
let direct_path = file_path.clone();
let direct = tokio::task::spawn_blocking(move || open_direct_writer(&direct_path, &write_state))
@@ -4521,14 +4888,14 @@ impl LocalIoBackend for StdBackend {
}
}
// O_TRUNC: if a file already exists at this path, stale trailing bytes past
// the new content would otherwise survive and mismatch the metadata size.
let f = super::fs::open_file(&file_path, O_CREATE | O_WRONLY | O_TRUNC)
.await
.map_err(to_file_error)?;
let reclaim_on_shutdown = should_reclaim_file_cache_after_write(size_hint);
Ok(Box::new(FileCacheReclaimWriter::new(f, size_hint.max(0) as usize, reclaim_on_shutdown)))
Ok(Box::new(LazyLocalFileWriter::new(
file_path,
volume_dir,
size_hint.max(0) as usize,
reclaim_on_shutdown,
sync_on_close,
)))
}
WriteMode::Append => {
let volume_dir = local_disk_bucket_path(self.io_root(), volume)?;
@@ -7077,6 +7444,9 @@ impl LocalDisk {
fi: FileInfo,
namespace_owner: Option<Arc<dyn Send + Sync>>,
) -> Result<()> {
if volume == RUSTFS_META_BUCKET {
crate::bucket::metadata_sys::bump_local_meta_mutation_gen();
}
if volume == RUSTFS_META_MULTIPART_BUCKET {
crate::disk::multipart_source::native_source_protocol(&fi)?;
}
@@ -7323,6 +7693,9 @@ impl LocalDisk {
mut fi: FileInfo,
mutation: DeleteVersionMutation,
) -> Result<()> {
if volume == RUSTFS_META_BUCKET {
crate::bucket::metadata_sys::bump_local_meta_mutation_gen();
}
pool_meta_native::reject_noncanonical_object(&self.root, volume, path)?;
let DeleteVersionMutation {
force_del_marker,
@@ -7394,6 +7767,13 @@ 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())
.await?
{
return Ok(());
}
let buf = match self.read_all_data(volume, &volume_dir, &xl_path).await {
Ok(res) => res,
Err(err) => {
@@ -7919,6 +8299,9 @@ impl LocalDisk {
opt: DeleteOptions,
namespace_owner: Option<Arc<dyn Send + Sync>>,
) -> Result<()> {
if volume == RUSTFS_META_BUCKET {
crate::bucket::metadata_sys::bump_local_meta_mutation_gen();
}
multipart_staging::reject_native_control_write(&self.root, self.io_root(), volume, path)?;
let mut source_owner = self.acquire_multipart_delete_owner(volume, path, namespace_owner).await?;
let namespace_owner = source_owner.namespace_owner();
@@ -9536,6 +9919,9 @@ impl LocalDisk {
// write_all_private with check_path_length
#[tracing::instrument(level = "trace", skip_all)]
async fn write_all_private(&self, volume: &str, path: &str, buf: Bytes, sync: SyncMode, skip_parent: &Path) -> Result<()> {
if volume == RUSTFS_META_BUCKET {
crate::bucket::metadata_sys::bump_local_meta_mutation_gen();
}
let file_path = self.io_get_object_path(volume, path)?;
check_path_length(file_path.to_string_lossy().as_ref())?;
@@ -11476,20 +11862,32 @@ pub(crate) async fn batch_shard_pread(requests: Vec<(std::path::PathBuf, usize,
return Err(DiskError::FileCorrupt);
}
let mut buf = diagnostic_stage!(Allocation, _index, || Ok::<_, DiskError>(vec![0u8; length]))?;
let mut buf = diagnostic_stage!(Allocation, _index, || Ok::<_, DiskError>(Vec::with_capacity(length)))?;
if length > 0 {
diagnostic_stage!(Read, _index, || measure_local_batch_stage(
metrics.as_deref(),
LocalBatchReadStage::Read,
|| {
while total < length {
let nbytes = file
.read_at(&mut buf[total..], u64::try_from(offset + total).unwrap_or(u64::MAX))
if offset == 0 {
use std::io::Read as _;
total = (&file)
.take(u64::try_from(length).unwrap_or(u64::MAX))
.read_to_end(&mut buf)
.map_err(DiskError::from)?;
if nbytes == 0 {
if total < length {
return Err(DiskError::FileCorrupt);
}
total += nbytes;
} else {
buf.resize(length, 0);
while total < length {
let nbytes = file
.read_at(&mut buf[total..], u64::try_from(offset + total).unwrap_or(u64::MAX))
.map_err(DiskError::from)?;
if nbytes == 0 {
return Err(DiskError::FileCorrupt);
}
total += nbytes;
}
}
Ok::<_, DiskError>(())
}
@@ -12741,9 +13139,23 @@ impl DiskAPI for LocalDisk {
}
}
self.io_backend
.open_write(volume, path, WriteMode::Truncate { size_hint: _file_size })
.await
#[cfg(test)]
let probe_active = os::file_sync_probe::is_active();
#[cfg(not(test))]
let probe_active = false;
let sync_on_close = cfg!(target_os = "linux")
&& volume == RUSTFS_META_TMP_BUCKET
&& !origvolume.is_empty()
&& path.ends_with("/part.1")
&& effective_durability(origvolume).syncs_data_shards()
&& !os::file_fdatasync_group_commit_enabled()
&& !probe_active;
let mode = if sync_on_close {
WriteMode::TruncateSync { size_hint: _file_size }
} else {
WriteMode::Truncate { size_hint: _file_size }
};
self.io_backend.open_write(volume, path, mode).await
}
async fn rename_file_durable(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str) -> Result<()> {
@@ -13308,16 +13720,70 @@ impl DiskAPI for LocalDisk {
check_path_length(file_path.to_string_lossy().as_ref())?;
let buf = self
.read_all(volume, format!("{}/{}", path, STORAGE_FORMAT_FILE).as_str())
.await
.map_err(|e| {
if e == DiskError::FileNotFound && fi.version_id.is_some() {
DiskError::FileVersionNotFound
} else {
e
let xl_rel_path = format!("{path}/{STORAGE_FORMAT_FILE}");
if self.multipart_source_admission(volume, &xl_rel_path, true, false)?.is_none()
&& !crash_inject::is_armed(CrashPoint::MetaWriteAfterTmpBeforeRename, &xl_rel_path)
&& !has_delete_version_test_failpoints()
&& !has_rename_data_test_failpoints(&xl_rel_path)
{
multipart_staging::reject_reserved_anchor(&xl_rel_path)?;
let volume_dir = self.io_get_bucket_path(volume)?;
let xl_file_path = self.io_get_object_path(volume, &xl_rel_path)?;
check_path_length(xl_file_path.to_string_lossy().as_ref())?;
let tmp_file_path = self.io_get_object_path(RUSTFS_META_TMP_BUCKET, Uuid::new_v4().to_string().as_str())?;
let sync = !opts.no_persistence && effective_durability(volume).syncs_commit_metadata();
let replace_user_metadata = opts.replace_user_metadata;
let has_version_id = fi.version_id.is_some();
let publication_root = self.publication_root.clone();
let lease = os::acquire_namespace_mutation_lease_with_owner(&xl_file_path, namespace_owner).await;
return os::run_blocking_namespace_operation_with_sync(lease, sync, move || -> std::io::Result<Result<()>> {
let buf = match std::fs::read(&xl_file_path) {
Ok(buf) => buf,
Err(err) if err.kind() == ErrorKind::NotFound && has_version_id => {
return Ok(Err(DiskError::FileVersionNotFound));
}
Err(err) => return Err(err),
};
if !FileMeta::is_xl2_v1_format(&buf) {
return Ok(Err(DiskError::FileVersionNotFound));
}
})?;
let wbuf = match FileMeta::load(&buf).and_then(|mut xl_meta| {
xl_meta.update_object_version_with_opts(fi, replace_user_metadata)?;
xl_meta.marshal_msg()
}) {
Ok(wbuf) => wbuf,
Err(err) => return Ok(Err(err.into())),
};
if let Some(parent) = tmp_file_path.parent() {
std::fs::create_dir_all(parent)?;
}
let mut tmp_file = std::fs::File::create(&tmp_file_path)?;
std::io::Write::write_all(&mut tmp_file, &wbuf)?;
if sync && !wbuf.is_empty() {
tmp_file.sync_data()?;
}
drop(tmp_file);
os::rename_all_with_precondition_and_parent_sync_std(
&tmp_file_path,
&xl_file_path,
&volume_dir,
&publication_root,
None,
sync,
)?;
Ok(Ok(()))
})
.await
.map_err(to_file_error)?;
}
let buf = self.read_all(volume, xl_rel_path.as_str()).await.map_err(|e| {
if e == DiskError::FileNotFound && fi.version_id.is_some() {
DiskError::FileVersionNotFound
} else {
e
}
})?;
if !FileMeta::is_xl2_v1_format(buf.as_ref()) {
return Err(DiskError::FileVersionNotFound);
@@ -13330,13 +13796,7 @@ impl DiskAPI for LocalDisk {
let wbuf = xl_meta.marshal_msg()?;
return self
.write_all_meta_with_namespace_owner(
volume,
format!("{path}/{STORAGE_FORMAT_FILE}").as_str(),
&wbuf,
!opts.no_persistence,
namespace_owner,
)
.write_all_meta_with_namespace_owner(volume, xl_rel_path.as_str(), &wbuf, !opts.no_persistence, namespace_owner)
.await;
}
@@ -17417,6 +17877,54 @@ mod test {
);
}
#[cfg(target_os = "linux")]
#[tokio::test]
#[serial_test::serial(dst_dir_fsync_group_commit, file_fdatasync_group_commit)]
#[allow(clippy::await_holding_lock)]
async fn rename_data_inline_new_version_fsync_failure_preserves_previous_metadata() {
let _group_commit = os::set_dst_dir_fsync_group_commit_for_test(false);
let _file_group_commit = os::set_file_fdatasync_group_commit_for_test(false);
let _mode = durability_mode_override::set(DurabilityMode::Strict);
let dir = tempfile::tempdir().expect("temp dir should be created");
let endpoint = Endpoint::try_from(dir.path().to_str().expect("temp dir should be utf8")).expect("endpoint should parse");
let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created");
let bucket = "inline-version-fsync-failure";
let object = "dir/object";
ensure_test_volume(&disk, bucket).await;
ensure_test_volume(&disk, RUSTFS_META_TMP_BUCKET).await;
let old_version = Uuid::new_v4();
let old_fi = test_file_info(object, old_version, None, Some(Bytes::from_static(b"previous-inline-version")));
disk.rename_data(RUSTFS_META_TMP_BUCKET, "old-inline-stage", old_fi, bucket, object)
.await
.expect("previous inline version should commit");
let dst_meta = disk
.get_object_path(bucket, &format!("{object}/{STORAGE_FORMAT_FILE}"))
.expect("metadata path should resolve");
let previous = fs::read(&dst_meta).await.expect("previous metadata should be readable");
let destination_dir = dst_meta.parent().expect("metadata should have a parent").to_path_buf();
// Arm the failure after the fused-path selection so the test exercises
// its normal admission path instead of the failpoint fallback.
set_inline_before_file_sync_admission(object, move || {
os::fsync_dir_recorder::set_failure(&destination_dir, io::ErrorKind::PermissionDenied);
});
let new_version = Uuid::new_v4();
let new_fi = test_file_info(object, new_version, None, Some(Bytes::from_static(b"incoming-inline-version")));
let err = disk
.rename_data(RUSTFS_META_TMP_BUCKET, "new-inline-stage", new_fi, bucket, object)
.await
.expect_err("directory fsync failure must reject the new version");
assert!(matches!(err, DiskError::Io(ref err) if err.kind() == io::ErrorKind::PermissionDenied));
let restored = fs::read(&dst_meta)
.await
.expect("previous metadata must survive the failed commit");
assert_eq!(restored, previous, "rollback must restore all previously committed inline versions");
let metadata = FileMeta::load(&restored).expect("restored metadata must be valid");
assert!(metadata.find_version(Some(old_version)).is_ok());
assert!(metadata.find_version(Some(new_version)).is_err());
}
// Seed a first PUT of `object` (no prior version) through the non-inline
// rename_data path and return (disk, tempdir). The object dir and any prefix
// dirs are created during the commit.
@@ -19643,6 +20151,116 @@ mod test {
);
}
#[cfg(target_os = "linux")]
#[tokio::test]
#[serial_test::serial(dst_dir_fsync_group_commit, file_fdatasync_group_commit, file_sync_probe)]
#[allow(clippy::await_holding_lock)]
async fn test_rename_data_unknown_destination_defers_shard_sync() {
let _group_commit = os::set_dst_dir_fsync_group_commit_for_test(false);
let _file_group_commit = os::set_file_fdatasync_group_commit_for_test(false);
let _mode = durability_mode_override::set(DurabilityMode::Strict);
for destination_mode in [DurabilityMode::None, DurabilityMode::Strict] {
let bucket = "unknown-destination-shard-bucket";
let _bucket_mode = BucketOverrideGuard::set(bucket, destination_mode);
let dir = tempfile::tempdir().expect("temp dir should be created");
let endpoint =
Endpoint::try_from(dir.path().to_str().expect("temp dir should be utf8")).expect("endpoint should parse");
let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created");
let object = "dir/object";
let tmp_object = "tmp-unknown-destination";
let data_dir = Uuid::new_v4();
ensure_test_volume(&disk, bucket).await;
ensure_test_volume(&disk, RUSTFS_META_TMP_BUCKET).await;
let shard_path = format!("{tmp_object}/{data_dir}/part.1");
let mut writer = disk
.create_file("", RUSTFS_META_TMP_BUCKET, &shard_path, 8)
.await
.expect("open staged shard writer without a destination");
writer.write_all(b"new-data").await.expect("write staged shard");
writer.shutdown().await.expect("close staged shard");
let staged_dir = disk
.get_object_path_for_io(RUSTFS_META_TMP_BUCKET, &format!("{tmp_object}/{data_dir}"))
.expect("staged shard directory should resolve");
assert!(
!PRESYNCED_TMP_DATA_DIRS.lock().contains(&staged_dir),
"an unknown destination must defer payload sync until commit"
);
let mut fi = test_file_info(object, Uuid::new_v4(), Some(data_dir), None);
fi.size = 8;
fi.parts[0].size = 8;
fi.parts[0].actual_size = 8;
disk.rename_data(RUSTFS_META_TMP_BUCKET, tmp_object, fi, bucket, object)
.await
.expect("rename_data should apply the destination policy");
assert_eq!(
os::fsync_dir_recorder::was_fsynced(&staged_dir),
destination_mode.syncs_data_shards(),
"the commit must apply the destination durability override"
);
let shard = disk
.get_object_path(bucket, &format!("{object}/{data_dir}/part.1"))
.expect("committed shard should resolve");
assert_eq!(fs::read(shard).await.expect("committed shard should be readable"), b"new-data");
}
}
#[cfg(target_os = "linux")]
#[tokio::test]
#[serial_test::serial(dst_dir_fsync_group_commit, file_fdatasync_group_commit, file_sync_probe)]
#[allow(clippy::await_holding_lock)]
async fn test_rename_data_presynced_shard_keeps_directory_sync() {
let _group_commit = os::set_dst_dir_fsync_group_commit_for_test(false);
let _file_group_commit = os::set_file_fdatasync_group_commit_for_test(false);
for mode in [DurabilityMode::Strict, DurabilityMode::Relaxed] {
let _mode = durability_mode_override::set(mode);
let dir = tempfile::tempdir().expect("temp dir should be created");
let endpoint =
Endpoint::try_from(dir.path().to_str().expect("temp dir should be utf8")).expect("endpoint should parse");
let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created");
let bucket = "presynced-shard-bucket";
let object = "dir/object";
let tmp_object = "tmp-presynced-write";
let data_dir = Uuid::new_v4();
ensure_test_volume(&disk, bucket).await;
ensure_test_volume(&disk, RUSTFS_META_TMP_BUCKET).await;
let shard_path = format!("{tmp_object}/{data_dir}/part.1");
let mut writer = disk
.create_file(bucket, RUSTFS_META_TMP_BUCKET, &shard_path, 8)
.await
.expect("open staged shard writer");
writer.write_all(b"new-data").await.expect("write staged shard");
writer.shutdown().await.expect("close staged shard");
let staged_dir = disk
.get_object_path_for_io(RUSTFS_META_TMP_BUCKET, &format!("{tmp_object}/{data_dir}"))
.expect("staged shard directory should resolve");
assert!(PRESYNCED_TMP_DATA_DIRS.lock().contains(&staged_dir), "the writer must pre-sync the shard");
assert!(!os::fsync_dir_recorder::was_fsynced(&staged_dir));
let mut fi = test_file_info(object, Uuid::new_v4(), Some(data_dir), None);
fi.size = 8;
fi.parts[0].size = 8;
fi.parts[0].actual_size = 8;
disk.rename_data(RUSTFS_META_TMP_BUCKET, tmp_object, fi, bucket, object)
.await
.expect("rename_data should commit");
assert!(
!PRESYNCED_TMP_DATA_DIRS.lock().contains(&staged_dir),
"the fused commit must consume the pre-sync marker"
);
assert!(
os::fsync_dir_recorder::was_fsynced(&staged_dir),
"pre-syncing the payload must not skip the shard directory barrier"
);
let shard = disk
.get_object_path(bucket, &format!("{object}/{data_dir}/part.1"))
.expect("committed shard should resolve");
assert_eq!(fs::read(shard).await.expect("committed shard should be readable"), b"new-data");
}
}
#[tokio::test]
#[allow(clippy::await_holding_lock)]
async fn test_rename_data_legacy_off_skips_all_fsyncs() {
@@ -19663,17 +20281,20 @@ mod test {
ensure_test_volume(&disk, bucket).await;
ensure_test_volume(&disk, RUSTFS_META_TMP_BUCKET).await;
let tmp_data_dir = dir
.path()
.join(RUSTFS_META_TMP_BUCKET)
.join(tmp_object)
.join(new_data_dir.to_string());
fs::create_dir_all(&tmp_data_dir)
let shard_path = format!("{tmp_object}/{new_data_dir}/part.1");
let mut writer = disk
.create_file("", RUSTFS_META_TMP_BUCKET, &shard_path, 8)
.await
.expect("new tmp data dir should be created");
fs::write(tmp_data_dir.join("part.1"), b"new-data")
.await
.expect("new tmp data should be written");
.expect("open staged shard writer");
writer.write_all(b"new-data").await.expect("write staged shard");
writer.shutdown().await.expect("close staged shard");
let resolved_tmp_data_dir = disk
.get_object_path(RUSTFS_META_TMP_BUCKET, &format!("{tmp_object}/{new_data_dir}"))
.expect("tmp data dir path should resolve");
assert!(
!os::fsync_dir_recorder::was_fsynced(&resolved_tmp_data_dir),
"the staged writer must not sync before the destination policy is known"
);
let new_fi = test_file_info(object, version_id, Some(new_data_dir), None);
disk.rename_data(RUSTFS_META_TMP_BUCKET, tmp_object, new_fi, bucket, object)
+245
View File
@@ -329,6 +329,38 @@ pub(super) async fn lock_rename_commit_directories(
result.map_err(to_file_error).map_err(DiskError::from)
}
#[cfg(target_os = "linux")]
fn prepare_rename_commit_and_destination_metadata_std(
create_source_parent: bool,
source_parent: &Path,
destination_parent: &Path,
destination_metadata: &Path,
base_dir: &Path,
publication_root: &os::PublicationRoot,
) -> std::io::Result<(os::RenameCommitGuard, Option<Bytes>)> {
if create_source_parent {
std::fs::create_dir_all(source_parent)?;
}
let guard = match os::prepare_rename_commit_guard(source_parent, destination_parent, base_dir, publication_root) {
Ok(guard) => guard,
Err(err) => {
return Err(match std::fs::symlink_metadata(base_dir) {
Err(base_err) if base_err.kind() == ErrorKind::NotFound => base_err,
_ => err,
});
}
};
let data = os::read_destination_file_with_commit_guard(destination_metadata, &guard)?;
Ok((guard, data.map(Bytes::from)))
}
#[cfg(target_os = "linux")]
enum FusedCommitAttempt {
Committed(RenameDataResp),
CommitErr(DiskError),
Fallback(os::RenameCommitGuard, Option<Bytes>, FileInfo),
}
async fn prepare_rename_commit_and_destination_metadata(
create_source_parent: bool,
source_parent: &Path,
@@ -571,6 +603,9 @@ impl LocalDisk {
return Err(err);
}
}
if dst_volume == crate::disk::RUSTFS_META_BUCKET {
crate::bucket::metadata_sys::bump_local_meta_mutation_gen();
}
if fi.is_legacy_indexed_delete_marker() {
fi.erasure.index = 0;
}
@@ -698,6 +733,216 @@ impl LocalDisk {
let dst_file_parent = dst_file_path
.parent()
.ok_or_else(|| DiskError::other("missing object metadata parent"))?;
#[cfg(target_os = "linux")]
let (rename_commit_guard, has_dst_buf) = if !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()
&& !super::has_rename_data_test_failpoints(dst_path)
&& !crash_inject::is_armed(CrashPoint::RenameAfterDataRename, dst_path)
&& !crash_inject::is_armed(CrashPoint::RenameAfterBackupBeforeMetaCommit, dst_path)
&& !crash_inject::is_armed(CrashPoint::RenameAfterMetaCommit, dst_path)
&& !super::has_delete_version_test_failpoints()
{
#[cfg(test)]
if !no_inline {
run_inline_before_file_sync_admission(dst_path);
}
let sync_data_shards = no_inline && durability.syncs_data_shards();
let sync_commit_metadata = durability.syncs_commit_metadata();
let file_sync_admission = if sync_data_shards || sync_commit_metadata {
Some(
os::acquire_file_sync_admission(self.file_sync_permits.clone())
.await
.map_err(to_file_error)
.map_err(DiskError::from)?,
)
} else {
None
};
let src_parent_owned = src_file_parent.to_path_buf();
let dst_parent_owned = dst_file_parent.to_path_buf();
let src_file_path_owned = src_file_path.clone();
let dst_file_path_owned = dst_file_path.clone();
let dst_volume_dir_owned = dst_volume_dir.clone();
let publication_root_owned = self.publication_root.clone();
let has_data_dir_path_owned = has_data_dir_path.clone();
let dst_volume_str = dst_volume.to_string();
let dst_path_str = dst_path.to_string();
let fi_moved = fi;
#[cfg(test)]
let root_owned = self.root.clone();
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,
&src_parent_owned,
&dst_parent_owned,
&dst_file_path_owned,
&dst_volume_dir_owned,
&publication_root_owned,
)?;
// An existing xl.meta can share versions or require rollback
// restoration when a post-rename directory sync fails.
if is_cancelled() || (sync_commit_metadata && has_dst_buf.is_some()) {
return Ok(FusedCommitAttempt::Fallback(guard, has_dst_buf, fi_moved));
}
let mut xlmeta = FileMeta::new();
let mut dst_meta_unparsable = false;
if let Some(ref dst_buf) = has_dst_buf {
if FileMeta::is_xl2_v1_format(dst_buf)
&& let Ok(nmeta) = FileMeta::load(dst_buf)
{
xlmeta = nmeta;
} else {
dst_meta_unparsable = true;
}
}
let version_id = fi_moved.version_id.unwrap_or_default();
let has_old_data_dir = xlmeta.find_unshared_data_dir_for_version(Some(version_id));
let rollback_data_dir = has_old_data_dir.or_else(|| {
if has_dst_buf.is_some() && xlmeta.find_version(Some(version_id)).is_ok() {
Some(inline_metadata_rollback_dir(version_id, &xlmeta))
} else {
None
}
});
if rollback_data_dir.is_some() {
return Ok(FusedCommitAttempt::Fallback(guard, has_dst_buf, fi_moved));
}
let (old_current_size, old_current_source_checked, old_current_source) = observe_old_current(
has_dst_buf.is_some(),
dst_meta_unparsable,
&xlmeta,
&dst_volume_str,
&dst_path_str,
version_id,
);
xlmeta.add_version(fi_moved)?;
let version_signature = rename_data_versions_signature(&xlmeta);
let new_dst_buf = xlmeta.marshal_msg().map_err(std::io::Error::other)?;
let destination_was_absent = has_dst_buf.is_none();
let mut skip_parent = dst_volume_dir_owned.clone();
if has_dst_buf.is_some() {
skip_parent = dst_parent_owned.clone();
}
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, sync_commit_metadata)?;
if no_inline {
if sync_data_shards && let Some((src_data_path, _)) = has_data_dir_path_owned.as_ref() {
let sync_result = if super::take_presynced_tmp_data_dir(src_data_path) {
// The payload was synced, but its part.1 directory entry
// must also be persisted before moving the data directory.
os::fsync_dir_std(src_data_path)
} else {
os::sync_dir_files_std(src_data_path)
};
if let Err(err) = sync_result
&& err.kind() != ErrorKind::NotFound
{
return Err(err);
}
}
if let Some((src_data_path, dst_data_path)) = has_data_dir_path_owned.as_ref()
&& let Err(err) = os::rename_all_with_commit_guard_on_thread(src_data_path, dst_data_path, &guard)
{
return Ok(FusedCommitAttempt::CommitErr(err));
}
#[cfg(test)]
if has_data_dir_path_owned.is_some() {
run_rename_data_after_first_publication(&root_owned, &dst_volume_str, &dst_path_str);
}
}
if let Err(err) = os::rename_prepared_source_on_thread(
prepared_metadata_source,
&src_file_path_owned,
&dst_file_path_owned,
&skip_parent,
&publication_root_owned,
&guard,
) {
if no_inline && let Some((src_data_path, dst_data_path)) = has_data_dir_path_owned.as_ref() {
let _ = std::fs::rename(dst_data_path, src_data_path);
}
return Ok(FusedCommitAttempt::CommitErr(err));
}
if sync_commit_metadata
&& let Err(err) =
fsync_inline_commit_std(&dst_file_path_owned, &dst_volume_dir_owned, destination_was_absent, None, &guard)
{
if no_inline && let Some((src_data_path, dst_data_path)) = has_data_dir_path_owned.as_ref() {
let _ = std::fs::rename(dst_data_path, src_data_path);
}
return Err(err);
}
if let Some(parent) = src_file_path_owned.parent() {
let _ = std::fs::remove_dir(parent);
}
Ok(FusedCommitAttempt::Committed(RenameDataResp {
multipart_source: None,
old_data_dir: None,
rollback_data_dir: None,
cleanup_data_dir: None,
sign: version_signature,
old_current_size,
old_current_source_checked,
old_current_source,
}))
};
let attempt_res = if let Some(admission) = file_sync_admission.as_ref() {
os::run_cancellable_blocking_namespace_file_sync_operation(mutation_lease.clone(), admission, fused_attempt).await
} else {
os::run_cancellable_blocking_namespace_operation(mutation_lease.clone(), fused_attempt).await
}
.map_err(to_file_error)
.map_err(DiskError::from);
drop(file_sync_admission);
match attempt_res {
Ok(FusedCommitAttempt::Committed(resp)) => {
for part_path in &invalidate_part_paths {
self.io_backend.invalidate_cached_fd(dst_volume, part_path).await;
}
#[cfg(unix)]
if let Some(receipt_mutation) = receipt_mutation.as_mut() {
receipt_mutation.complete();
}
return Ok(resp);
}
Ok(FusedCommitAttempt::CommitErr(err)) | Err(err) => {
for part_path in &invalidate_part_paths {
self.io_backend.invalidate_cached_fd(dst_volume, part_path).await;
}
return Err(err);
}
Ok(FusedCommitAttempt::Fallback(guard, dst_buf, restored_fi)) => {
fi = restored_fi;
(guard, dst_buf)
}
}
} else {
prepare_rename_commit_and_destination_metadata(
!no_inline && !preserve_upload_metadata,
src_file_parent,
dst_file_parent,
&dst_file_path,
&dst_volume_dir,
&self.publication_root,
mutation_lease.clone(),
)
.await?
};
#[cfg(not(target_os = "linux"))]
// Acquire the common trees before reading destination metadata. On
// Windows this pins the object directory identity across publication.
// Multipart authority already created its source tree. Recreating it
@@ -2090,6 +2090,474 @@ impl LocalDisk {
pub(super) async fn delete_recovery_barrier(&self, volume: &str, object: &str, transaction: Uuid, stage: &str) -> Result<()> {
barrier(volume, object, transaction, stage, &self.root).await
}
pub(super) async fn try_fused_recoverable_delete_version(
&self,
volume: &str,
object: &str,
fi: &FileInfo,
opts: &DeleteOptions,
transaction: Uuid,
owner: Option<Arc<dyn Send + Sync>>,
) -> Result<Option<()>> {
if !opts.recoverable_delete
|| opts.undo_write
|| opts.expected_delete_marker.is_some()
|| pool_meta_native::is_object(&self.root, volume, object)
|| has_delete_version_test_failpoints()
{
return Ok(None);
}
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 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 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;
let fi_clone = fi.clone();
let volume_str = volume.to_owned();
let object_str = object.to_owned();
let res = os::run_blocking_namespace_operation_with_sync(
lease,
sync,
move || -> std::io::Result<Result<Option<Option<Uuid>>>> {
let buf = match std::fs::read(&xl_path) {
Ok(buf) => buf,
Err(err) if err.kind() == ErrorKind::NotFound => return Ok(Ok(None)),
Err(err) => return Err(err),
};
let (old_dir, journal_raw) = match (|| -> Result<Option<(Option<Uuid>, Vec<u8>)>> {
let mut meta = FileMeta::load(&buf)?;
let deleted_versions: Vec<_> = meta
.versions
.iter()
.map(|v| v.header.version_id.unwrap_or_default())
.collect();
let original_versions: Vec<_> = meta
.versions
.iter()
.map(|v| generation(v.into_fileinfo(&volume_str, &object_str, true)?))
.collect::<Result<_>>()?;
let old_dir = meta.delete_version(&fi_clone)?;
if old_dir.is_some_and(|dir| dir == transaction || dir.is_nil()) {
return Err(DiskError::FileCorrupt);
}
if let Some(uuid) = old_dir {
meta.data.remove(vec![fi_clone.version_id.unwrap_or_default(), uuid])?;
let old_path = path_join(&[file_path.as_path(), Path::new(uuid.to_string().as_str())]);
check_path_length(old_path.to_string_lossy().as_ref())?;
if old_path.join(STORAGE_FORMAT_FILE).exists() {
return Ok(None);
}
}
if !meta.versions.is_empty() {
return Ok(None);
}
let mut predecessors = BTreeSet::new();
if let Ok(entries) = std::fs::read_dir(&file_path) {
for (visited, entry) in entries.enumerate() {
if visited >= 65536 {
return Err(DiskError::DiskOngoingReq);
}
let entry = entry.map_err(to_file_error)?;
let Some(id) = entry
.file_name()
.to_str()
.and_then(|name| Uuid::parse_str(name).ok())
.filter(|id| !id.is_nil() && *id != transaction)
else {
continue;
};
if !entry.file_type().map_err(to_file_error)?.is_dir() {
return Err(DiskError::InvalidPath);
}
let prepared = regular_bytes(&entry.path().join(PREPARED))?;
let receipt = regular_bytes(&entry.path().join(DELETE_QUORUM_COMMITTED_FILE))?;
let mut records = Vec::new();
if let Some(bytes) = prepared {
records.push(
serde_json::from_slice::<DeleteRecoveryJournal>(&bytes)
.map_err(|_| DiskError::FileCorrupt)?,
);
}
if let Some(bytes) = receipt.filter(|bytes| !bytes.is_empty()) {
records.push(
serde_json::from_slice::<CommitReceipt>(&bytes)
.map_err(|_| DiskError::FileCorrupt)?
.journal,
);
}
for record in records {
if record.protocol_version != DELETE_RECOVERY_PROTOCOL_VERSION
|| record.volume != volume_str
|| record.object != object_str
|| record.transaction != id
|| incarnation.zip(record.bucket_incarnation).is_some_and(|(a, b)| a != b)
|| record.predecessors.iter().any(|before| before.is_nil() || *before == id)
{
return Err(DiskError::FileCorrupt);
}
predecessors.insert(id);
}
}
}
let mut sparse = fi_clone;
sparse.data = None;
let journal = DeleteRecoveryJournal {
protocol_version: DELETE_RECOVERY_PROTOCOL_VERSION,
volume: volume_str,
object: object_str,
transaction,
bucket_incarnation: incarnation,
original_digest: Some(digest(&buf)),
target_digest: None,
data_dirs: old_dir.into_iter().collect(),
deleted_versions,
requests: vec![sparse],
original_versions,
predecessors: predecessors.into_iter().collect(),
};
Ok(Some((old_dir, journal_bytes(&journal)?)))
})() {
Ok(Some(pair)) => pair,
Ok(None) => return Ok(Ok(None)),
Err(err) => return Ok(Err(err)),
};
std::fs::create_dir_all(&staging_root)?;
let staged = tempfile::Builder::new()
.prefix("delete-preparation-")
.tempdir_in(&staging_root)?;
for (name, bytes) in [
(STORAGE_FORMAT_FILE_BACKUP, buf.as_slice()),
(PREPARED, journal_raw.as_slice()),
] {
let mut file = std::fs::File::create(staged.path().join(name))?;
std::io::Write::write_all(&mut file, bytes)?;
if sync && !bytes.is_empty() {
file.sync_data()?;
}
}
if sync {
os::fsync_dir_std(staged.path())?;
}
match std::fs::symlink_metadata(&destination) {
Ok(metadata) if metadata.is_dir() => match std::fs::remove_dir(&destination) {
Ok(()) => {}
Err(err) if err.kind() == ErrorKind::NotFound => {}
Err(err) if is_dir_not_empty_error(&err) => return Ok(Err(DiskError::OutdatedXLMeta)),
Err(err) => return Err(err),
},
Ok(_) => return Ok(Err(DiskError::OutdatedXLMeta)),
Err(err) if err.kind() == ErrorKind::NotFound => {}
Err(err) => return Err(err),
}
if let Some(parent) = destination.parent() {
std::fs::create_dir_all(parent)?;
}
std::fs::rename(staged.path(), &destination)?;
if sync {
os::fsync_dir_std(&file_path)?;
}
let mut reserved_data_path = None;
if let Some(uuid) = old_dir {
let old_path = file_path.join(uuid.to_string());
if old_path.is_dir() {
let reserved_marker = old_path.join(format!("{RESERVED_DELETE_DATA_DIR_MARKER_PREFIX}{transaction}"));
let marker = std::fs::File::create(&reserved_marker)?;
if sync {
marker.sync_all()?;
os::fsync_dir_std(&old_path)?;
}
reserved_data_path = Some((old_path, reserved_marker));
}
}
if let Err(err) = std::fs::remove_file(&xl_path)
&& err.kind() != ErrorKind::NotFound
{
if let Some((_, reserved_marker)) = &reserved_data_path {
let _ = std::fs::remove_file(reserved_marker);
}
let _ = std::fs::remove_dir_all(&destination);
return Err(err);
}
if sync {
os::fsync_dir_std(&file_path)?;
}
if let Some((old_path, reserved_marker)) = reserved_data_path {
let recovery_marker = old_path.join(format!("{RECOVERY_DELETE_DATA_DIR_MARKER_PREFIX}{transaction}"));
match std::fs::rename(&reserved_marker, &recovery_marker) {
Ok(()) if sync => os::fsync_dir_std(&old_path)?,
Ok(()) => {}
Err(err) if err.kind() == ErrorKind::NotFound => {}
Err(err) => return Err(err),
}
}
Ok(Ok(Some(old_dir)))
},
)
.await
.map_err(to_file_error)??;
let Some(old_dir) = res else {
return Ok(None);
};
if let Some(uuid) = old_dir {
self.io_backend
.invalidate_cached_fds_under(volume, &format!("{object}/{uuid}"));
}
Ok(Some(()))
}
pub(crate) async fn try_publish_prepared_delete_local(
&self,
volume: &str,
object: &str,
transaction: Uuid,
namespace_owner: Option<Arc<dyn Send + Sync>>,
) -> Result<Option<(DeleteRecoveryJournal, bool, bool)>> {
self.reject_multipart_delete_recovery_mutation(volume, object)?;
if !safe_path(object, false) || transaction.is_nil() {
return Err(DiskError::InvalidPath);
}
let object_path = self.get_object_path(volume, object)?;
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();
let object_str = object.to_owned();
os::run_blocking_namespace_operation_with_sync(
lease,
sync,
move || -> std::io::Result<Result<Option<(DeleteRecoveryJournal, bool, bool)>>> {
let prepared = match regular_bytes(&rollback.join(PREPARED)) {
Ok(Some(bytes)) => bytes,
Ok(None) => return Ok(Ok(None)),
Err(e) => return Ok(Err(e)),
};
let Ok(journal) = serde_json::from_slice::<DeleteRecoveryJournal>(&prepared) else {
return Ok(Err(DiskError::FileCorrupt));
};
if journal.protocol_version != DELETE_RECOVERY_PROTOCOL_VERSION
|| journal.volume != volume_str
|| journal.object != object_str
|| journal.transaction != transaction
|| journal.transaction.is_nil()
|| journal.data_dirs.iter().any(|id| id.is_nil() || *id == transaction)
{
return Ok(Err(DiskError::FileCorrupt));
}
let current_exists = match regular_file_exists(&root.join(STORAGE_FORMAT_FILE)) {
Ok(exists) => exists,
Err(e) => return Ok(Err(e)),
};
let unversioned_target_absent = journal.target_digest.is_none() && !current_exists;
if !unversioned_target_absent {
return Ok(Ok(None));
}
let mut sole_transaction = journal.predecessors.is_empty();
if sole_transaction {
match std::fs::read_dir(&root) {
Ok(entries) => {
for entry in entries {
let entry = entry?;
let Ok(name) = entry.file_name().into_string() else {
sole_transaction = false;
break;
};
if let Some(id) = canonical_uuid(&name).filter(|id| !id.is_nil())
&& id != transaction
&& !journal.data_dirs.contains(&id)
{
sole_transaction = false;
break;
}
}
}
Err(err) if err.kind() == ErrorKind::NotFound => {}
Err(err) => return Err(err),
}
}
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()?;
}
std::fs::rename(staged.path(), &receipt_path)?;
if sync {
os::fsync_dir_std(&rollback)?;
}
Ok(Ok(Some((journal, sole_transaction, unversioned_target_absent))))
},
)
.await
.map_err(to_file_error)?
}
pub(crate) async fn try_finish_and_finalize_prepared_delete_local(
&self,
volume: &str,
object: &str,
transaction: Uuid,
journal: &DeleteRecoveryJournal,
namespace_owner: Option<Arc<dyn Send + Sync>>,
) -> Result<bool> {
fn ignore_not_found(res: std::io::Result<()>) -> std::io::Result<()> {
match res {
Ok(()) => Ok(()),
Err(err) if err.kind() == ErrorKind::NotFound => Ok(()),
Err(err) => Err(err),
}
}
self.reject_multipart_delete_recovery_mutation(volume, object)?;
if !safe_path(object, false) || transaction.is_nil() {
return Err(DiskError::InvalidPath);
}
let object_path = self.get_object_path(volume, object)?;
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 mut claims: Vec<SnapshotLeaseDeleteClaim> = Vec::with_capacity(journal.data_dirs.len());
for data_dir in &journal.data_dirs {
let key = SnapshotLeaseKey {
volume: volume.to_owned(),
path: format!("{object}/{data_dir}"),
};
let opts = DeleteOptions {
recursive: true,
old_data_dir: Some(transaction),
..Default::default()
};
let Some(claim) = self.claim_snapshot_delete(key, SnapshotPendingDelete::Ordinary(opts)) else {
for prior in claims {
prior.finish(false);
}
return Ok(false);
};
claims.push(claim);
}
let bucket_path = self.io_get_bucket_path(volume)?;
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 lease = os::acquire_namespace_mutation_lease_with_owner(&receipt_path, owner).await;
let data_dirs = journal.data_dirs.clone();
let prune_paths: Vec<_> = rollback
.ancestors()
.take_while(|dir| *dir != bucket_path.as_path() && dir.starts_with(&bucket_path))
.map(Path::to_path_buf)
.collect();
let res = os::run_blocking_namespace_operation_with_sync(lease, sync, move || -> std::io::Result<()> {
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}");
for data_dir in &data_dirs {
for base in [&root, &rollback] {
let dir_path = base.join(data_dir.to_string());
match std::fs::read_dir(&dir_path) {
Ok(entries) => {
let mut removable = Vec::new();
let mut preserve_dir = false;
for entry in entries {
let entry = entry?;
let name = entry.file_name();
let Some(name) = name.to_str() else {
preserve_dir = true;
continue;
};
if entry.file_type()?.is_file()
&& (delete_recovery_part_file(name)
|| name == committed
|| name == recovery_committed
|| name == reserved
|| is_delete_transaction_marker(name, RESERVED_DELETE_DATA_DIR_MARKER_PREFIX))
{
removable.push(entry.path());
} else {
preserve_dir = true;
}
}
for path in removable {
ignore_not_found(std::fs::remove_file(path))?;
}
if sync {
os::fsync_dir_std(&dir_path)?;
}
if !preserve_dir {
match std::fs::remove_dir(&dir_path) {
Ok(()) => {}
Err(err) if err.kind() == ErrorKind::NotFound || is_dir_not_empty_error(&err) => {}
Err(err) => return Err(err),
}
}
}
Err(err) if err.kind() == ErrorKind::NotFound => {}
Err(err) => return Err(err),
}
}
}
if !data_dirs.is_empty() && sync && root.is_dir() {
os::fsync_dir_std(&root)?;
}
for name in [
STORAGE_FORMAT_FILE_BACKUP,
DELETE_MARKER_ROLLBACK_FILE,
TARGET,
PREPARED,
FINISHED,
] {
ignore_not_found(std::fs::remove_file(rollback.join(name)))?;
}
if sync {
os::fsync_dir_std(&rollback)?;
}
ignore_not_found(std::fs::remove_file(&receipt_path))?;
for dir_path in &prune_paths {
if let Err(err) = std::fs::remove_dir(dir_path) {
if is_dir_not_empty_error(&err) {
break;
}
if !is_benign_object_rmdir_error(&err) {
return Err(std::io::Error::other(FileAccessDeniedWithContext {
path: dir_path.clone(),
source: err,
}));
}
}
}
Ok(())
})
.await;
let ok = res.is_ok();
for claim in claims {
claim.finish(ok);
}
res.map(|()| true).map_err(|err| to_file_error(err).into())
}
}
#[cfg(not(feature = "e2e-test-hooks"))]
@@ -3309,6 +3777,171 @@ mod tests {
inspected
}
#[tokio::test]
async fn fused_delete_recovery_survives_interrupted_prepare_and_publication() {
for published in [false, true] {
let (_temp, disk, _info, original) = fixture().await;
let transaction = Uuid::new_v4();
let prepared = disk
.try_fused_recoverable_delete_version(
"recovery",
"object",
&request(),
&DeleteOptions {
recoverable_delete: true,
old_data_dir: Some(transaction),
..Default::default()
},
transaction,
None,
)
.await
.expect("prepare fused deletion");
assert_eq!(prepared, Some(()), "the test must exercise the fused path");
if published {
let (_, sole, target_absent) = disk
.try_publish_prepared_delete_local("recovery", "object", transaction, None)
.await
.expect("publish fused receipt")
.expect("fused publication must apply");
assert!(sole && target_absent);
}
let mut state = disk
.inspect_delete_transaction("recovery", "object", transaction)
.await
.expect("inspect persisted state after interruption");
if published {
assert_eq!(state.state, DeleteRecoveryState::QuorumCommitted);
for action in [DeleteRecoveryAction::FinishCommit, DeleteRecoveryAction::FinalizeCommit] {
state = disk
.settle_delete_transaction("recovery", "object", transaction, state.fingerprint, action)
.await
.expect("finish published deletion through recovery");
}
assert_eq!(state.state, DeleteRecoveryState::Absent);
assert!(disk.read_all("recovery", "object/xl.meta").await.is_err());
} else {
assert_eq!(state.state, DeleteRecoveryState::NewPrepared);
state = disk
.settle_delete_transaction(
"recovery",
"object",
transaction,
state.fingerprint,
DeleteRecoveryAction::Rollback,
)
.await
.expect("roll back uncommitted fused deletion");
assert_eq!(state.state, DeleteRecoveryState::Absent);
assert_eq!(
disk.read_all("recovery", "object/xl.meta")
.await
.expect("restore original metadata")
.as_ref(),
original
);
}
}
}
#[tokio::test]
async fn fused_delete_recovery_preserves_or_removes_shard_after_interruption() {
use crate::erasure::coding::BitrotWriter;
for mode in 0..3 {
let published = mode != 0;
let (_temp, disk, _info, _) = fixture().await;
let data_dir = Uuid::new_v4();
let payload = Bytes::from_static(b"original shard");
let mut info = test_file_info("object", Uuid::nil(), Some(data_dir), None);
info.size = i64::try_from(payload.len()).expect("payload size");
info.erasure.block_size = payload.len();
info.parts[0].size = payload.len();
info.parts[0].actual_size = info.size;
let original = test_meta(info.clone());
let mut writer = BitrotWriter::new(
std::io::Cursor::new(Vec::new()),
info.erasure.shard_size(),
HashAlgorithm::HighwayHash256S,
);
writer.write(&payload).await.expect("encode shard");
writer.shutdown().await.expect("finish shard encoding");
let encoded = writer.into_inner().into_inner();
let root = disk.io_get_object_path("recovery", "object").expect("object path");
fs::write(root.join(STORAGE_FORMAT_FILE), &original)
.await
.expect("write shard metadata");
let shard_dir = root.join(data_dir.to_string());
fs::create_dir_all(&shard_dir).await.expect("create shard directory");
fs::write(shard_dir.join("part.1"), &encoded)
.await
.expect("write original shard");
let transaction = Uuid::new_v4();
assert_eq!(
disk.try_fused_recoverable_delete_version(
"recovery",
"object",
&request(),
&DeleteOptions {
recoverable_delete: true,
old_data_dir: Some(transaction),
..Default::default()
},
transaction,
None,
)
.await
.expect("prepare fused shard deletion"),
Some(()),
"the test must exercise the fused path"
);
if published {
let (journal, sole, target_absent) = disk
.try_publish_prepared_delete_local("recovery", "object", transaction, None)
.await
.expect("publish fused receipt")
.expect("fused publication must apply");
assert!(sole && target_absent);
if mode == 2 {
assert!(
disk.try_finish_and_finalize_prepared_delete_local("recovery", "object", transaction, &journal, None)
.await
.expect("finish fused deletion"),
"the fused finalization must apply"
);
assert!(!shard_dir.exists(), "fused finalization must remove the shard");
assert!(!root.join(STORAGE_FORMAT_FILE).exists());
assert!(!root.join(transaction.to_string()).exists());
continue;
}
}
let mut state = disk
.inspect_delete_transaction("recovery", "object", transaction)
.await
.expect("inspect persisted state");
let actions = if published {
vec![DeleteRecoveryAction::FinishCommit, DeleteRecoveryAction::FinalizeCommit]
} else {
vec![DeleteRecoveryAction::Rollback]
};
for action in actions {
state = disk
.settle_delete_transaction("recovery", "object", transaction, state.fingerprint, action)
.await
.expect("recover shard deletion");
}
assert_eq!(state.state, DeleteRecoveryState::Absent);
if published {
assert!(!shard_dir.exists(), "committed deletion must remove the shard");
} else {
assert_eq!(fs::read(root.join(STORAGE_FORMAT_FILE)).await.expect("restore metadata"), original);
assert_eq!(fs::read(shard_dir.join("part.1")).await.expect("restore shard"), encoded);
}
}
}
#[tokio::test]
async fn delete_recovery_uncommitted_inline_rolls_back_exact_bytes() {
let (_temp, disk, _info, original) = fixture().await;
+33
View File
@@ -387,6 +387,39 @@ impl Disk {
}
}
pub(crate) async fn try_publish_prepared_delete_local(
&self,
volume: &str,
object: &str,
transaction: Uuid,
namespace_owner: Option<Arc<dyn Send + Sync>>,
) -> Result<Option<(DeleteRecoveryJournal, bool, bool)>> {
match self {
Self::Local(disk) => {
disk.try_publish_prepared_delete_local(volume, object, transaction, namespace_owner)
.await
}
Self::Remote(_) => Ok(None),
}
}
pub(crate) async fn try_finish_and_finalize_prepared_delete_local(
&self,
volume: &str,
object: &str,
transaction: Uuid,
journal: &DeleteRecoveryJournal,
namespace_owner: Option<Arc<dyn Send + Sync>>,
) -> Result<bool> {
match self {
Self::Local(disk) => {
disk.try_finish_and_finalize_prepared_delete_local(volume, object, transaction, journal, namespace_owner)
.await
}
Self::Remote(_) => Ok(false),
}
}
pub(crate) async fn delete_version_recoverable_with_namespace_owner(
&self,
volume: &str,
+61 -11
View File
@@ -286,6 +286,11 @@ mod dir_fsync_failures {
static FAILURES: LazyLock<Mutex<HashMap<PathBuf, io::ErrorKind>>> = LazyLock::new(|| Mutex::new(HashMap::new()));
#[cfg(target_os = "linux")]
pub(super) fn has_any() -> bool {
!FAILURES.lock().is_empty()
}
pub(super) fn set(dir: &Path, kind: io::ErrorKind) {
FAILURES.lock().insert(dir.to_path_buf(), kind);
}
@@ -323,6 +328,11 @@ pub fn set_local_dir_fsync_failure(disk: &crate::disk::Disk, volume: &str, path:
Ok(())
}
#[cfg(all(target_os = "linux", any(test, feature = "test-util")))]
pub(crate) fn has_dir_fsync_test_failures() -> bool {
dir_fsync_failures::has_any()
}
/// Pause a real namespace mutation inside its physical executor.
#[cfg(all(any(test, feature = "test-util"), not(windows)))]
pub(crate) mod prepared_publication_test_hooks {
@@ -372,6 +382,17 @@ pub(crate) mod prepared_publication_test_hooks {
}
}
pub(crate) fn has_any_hooks() -> bool {
if !BEFORE_PUBLICATION.lock().is_empty() {
return true;
}
#[cfg(test)]
if !RENAME_DESTINATIONS.lock().is_empty() || !HARDLINK_DESTINATIONS.lock().is_empty() {
return true;
}
false
}
#[cfg(test)]
type RenameDestinationHook = Box<dyn FnOnce(&Path) + Send>;
#[cfg(test)]
@@ -758,7 +779,7 @@ fn set_file_fdatasync_group_commit_wait_for_test(wait_micros: u64) {
file_fdatasync_group_commit_override::set_wait_micros(wait_micros);
}
fn file_fdatasync_group_commit_enabled() -> bool {
pub(crate) fn file_fdatasync_group_commit_enabled() -> bool {
#[cfg(test)]
if let Some(enabled) = file_fdatasync_group_commit_override::get() {
return enabled;
@@ -2392,6 +2413,10 @@ pub(crate) mod file_sync_probe {
configure(root, None, false)
}
pub(crate) fn is_active() -> bool {
!ROOTS.read().expect("file sync probe lock poisoned").is_empty()
}
pub(super) fn set_failing(root: &Path) -> ProbeGuard {
configure(root, Some(1), false)
}
@@ -2553,18 +2578,25 @@ fn regular_files(dir: &Path) -> io::Result<Vec<PathBuf>> {
/// Fdatasync every regular file directly inside `dir`, then fsync the directory
/// itself.
#[allow(
dead_code,
reason = "reached only through sync_dir_files, whose callers are tests (backlog#1823)"
#[cfg_attr(
not(target_os = "linux"),
allow(
dead_code,
reason = "reached only through sync_dir_files, whose callers are tests (backlog#1823)"
)
)]
pub fn sync_dir_files_std(dir: impl AsRef<Path>) -> io::Result<()> {
for entry in std::fs::read_dir(dir.as_ref())? {
let entry = entry?;
if entry.file_type()?.is_file() {
sync_file(&entry.path())?;
}
}
fsync_dir_std(dir)
let dir = dir.as_ref();
let files = regular_files(dir)?;
rustfs_io_metrics::record_put_rename_fdatasync_batch(rustfs_io_metrics::PUT_RENAME_FDATASYNC_BATCH_MODE_SERIAL, files.len());
sync_files(&files)?;
let fsync_started = rustfs_io_metrics::put_stage_timer();
let result = fsync_dir_std(dir);
rustfs_io_metrics::record_put_object_stage_duration_from(
rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_SRC_DIR_FSYNC,
fsync_started,
);
result
}
/// Async wrapper around [`sync_dir_files_std`]. Large directories flush files
@@ -3808,6 +3840,23 @@ pub(crate) async fn rename_all_with_prepared_source(
Ok(())
}
#[cfg(target_os = "linux")]
pub(crate) fn rename_all_with_commit_guard_on_thread(source: &Path, destination: &Path, guard: &RenameCommitGuard) -> Result<()> {
guard.directories.prepare_destination().map_err(to_file_error)?;
#[cfg(test)]
prepared_publication_test_hooks::run_rename_destination(source, destination);
#[cfg(any(test, feature = "test-util"))]
{
prepared_publication_test_hooks::run(prepared_publication_test_hooks::Stage::Rename, source);
prepared_publication_test_hooks::run(prepared_publication_test_hooks::Stage::Rename, destination);
}
#[cfg(test)]
multipart_source_syscall_failures::check(multipart_source_syscall_failures::Stage::Rename, destination)
.map_err(to_file_error)?;
guard.directories.rename(source, destination, None).map_err(to_file_error)?;
Ok(())
}
#[cfg(not(windows))]
pub(crate) async fn rename_all_with_commit_guard(
src_file_path: impl AsRef<Path>,
@@ -6483,6 +6532,7 @@ pub(crate) fn sync_multipart_owned_record_parents(
if !directory.starts_with(source) {
return Err(io::Error::new(io::ErrorKind::InvalidInput, "native record barrier escaped source"));
}
#[cfg_attr(not(unix), allow(unused_mut))]
let mut pinned = pin_multipart_source_directory(directory, base, publication_root)?;
#[cfg(unix)]
{
+23 -6
View File
@@ -175,8 +175,10 @@ const DEFAULT_RUSTFS_GET_BITROT_DECODE_OVERLAP_ENABLE: bool = false;
/// stripe via their stripe handle — only when a data shard is missing or dies.
/// This halves per-GET read bytes, IOPS, and bitrot hashing on healthy
/// objects with 2+2 layouts.
/// Default: false (current behavior: every live shard reader is read on every
/// stripe).
/// Default: false for raw `Erasure::decode` calls without deferred parity
/// handles; `DataBlocksFirst` reader setup attaches deferred parity handles and
/// leaves those parity reserves unengaged by default unless this env var is set
/// to `false`.
const ENV_RUSTFS_GET_LOCKSTEP_DATA_SHARDS_ONLY_ENABLE: &str = "RUSTFS_GET_LOCKSTEP_DATA_SHARDS_ONLY_ENABLE";
const DEFAULT_RUSTFS_GET_LOCKSTEP_DATA_SHARDS_ONLY_ENABLE: bool = false;
@@ -189,6 +191,13 @@ pub(crate) fn get_lockstep_data_shards_only_enabled() -> bool {
)
}
/// Whether `DataBlocksFirst` reader setup and `with_deferred_parity_handles`
/// should keep parity shards deferred until needed on healthy GETs.
pub(crate) fn get_lockstep_data_shards_only_with_deferred_parity_enabled() -> bool {
matches!(decode_read_policy(), DecodeReadPolicy::DemandBound)
|| rustfs_utils::get_env_bool(ENV_RUSTFS_GET_LOCKSTEP_DATA_SHARDS_ONLY_ENABLE, true)
}
/// Get whether bitrot-decode overlap is enabled.
fn is_bitrot_decode_overlap_enabled() -> bool {
rustfs_utils::get_env_bool(
@@ -832,6 +841,13 @@ where
pub(crate) fn with_deferred_parity_handles(mut self, mut handles: Vec<Option<DeferredReaderStripeHandle>>) -> Self {
handles.resize_with(self.readers.len(), || None);
handles.truncate(self.readers.len());
if get_lockstep_data_shards_only_with_deferred_parity_enabled() {
for (index, handle) in handles.iter().enumerate().skip(self.data_shards) {
if handle.is_some() {
self.engaged[index] = false;
}
}
}
self.deferred_handles = handles;
self
}
@@ -2596,7 +2612,8 @@ impl Erasure {
// the ordinary GET read quorum. Only the opt-in GET rollout requires a
// surplus source; available surplus shards and integrity proofs are
// still verified before any reconstructed copy bytes are emitted.
let require_surplus_source = reader.demand_bound_lockstep && matches!(decode_read_policy(), DecodeReadPolicy::Default);
let require_surplus_source =
get_lockstep_data_shards_only_enabled() && matches!(decode_read_policy(), DecodeReadPolicy::Default);
let start = offset / self.block_size;
let end = end_offset.saturating_sub(1) / self.block_size;
@@ -4367,9 +4384,9 @@ mod tests {
.await;
}
/// Compatibility lock: with the gate off (default), the lockstep path
/// keeps the pre-backlog#923 behavior and reads every live shard —
/// including parity — on every stripe.
/// Compatibility lock: with the gate off (default), raw `Erasure::decode`
/// without deferred parity handles keeps the pre-backlog#923 behavior and
/// reads every live shard — including parity — on every stripe.
#[tokio::test]
#[serial_test::serial]
async fn test_lockstep_default_reads_all_shards_per_stripe() {
+20
View File
@@ -144,6 +144,8 @@ pub(crate) fn namespace_lock_signal_test_fence_is_lost(signal: &Arc<rustfs_lock:
.is_some_and(|(_, fence)| fence.is_lock_lost())
}
type FreshPutQuota = std::result::Result<(Option<crate::bucket::quota::BucketQuota>, Uuid, OffsetDateTime), String>;
#[derive(Debug)]
pub struct ObjectLockConfigSnapshot {
store_id: Option<Uuid>,
@@ -154,6 +156,7 @@ pub struct ObjectLockConfigSnapshot {
lifecycle_fence: NamespaceLockFence,
_lifecycle_guard: Option<rustfs_lock::NamespaceLockGuard>,
metadata_transaction_guard: Option<Arc<rustfs_lock::NamespaceLockGuard>>,
fresh_put_quota: std::sync::Mutex<Option<FreshPutQuota>>,
}
impl ObjectLockConfigSnapshot {
@@ -167,6 +170,7 @@ impl ObjectLockConfigSnapshot {
lifecycle_fence: NamespaceLockFence::new(),
_lifecycle_guard: None,
metadata_transaction_guard: None,
fresh_put_quota: std::sync::Mutex::new(None),
}
}
@@ -187,6 +191,7 @@ impl ObjectLockConfigSnapshot {
lifecycle_fence: NamespaceLockFence::new(),
_lifecycle_guard: None,
metadata_transaction_guard: None,
fresh_put_quota: std::sync::Mutex::new(None),
}
}
@@ -212,6 +217,7 @@ impl ObjectLockConfigSnapshot {
lifecycle_fence,
_lifecycle_guard: Some(lifecycle_guard),
metadata_transaction_guard: Some(Arc::new(metadata_transaction_guard)),
fresh_put_quota: std::sync::Mutex::new(None),
}
}
@@ -233,9 +239,23 @@ impl ObjectLockConfigSnapshot {
lifecycle_fence,
_lifecycle_guard: None,
metadata_transaction_guard: Some(metadata_transaction_guard),
fresh_put_quota: std::sync::Mutex::new(None),
}
}
pub(crate) fn set_fresh_put_quota(&self, quota: FreshPutQuota) {
if self._lifecycle_guard.is_some() && self.metadata_transaction_guard.is_some() {
*self.fresh_put_quota.lock().unwrap_or_else(std::sync::PoisonError::into_inner) = Some(quota);
}
}
pub(crate) fn take_fresh_put_quota(&self) -> Option<FreshPutQuota> {
self.fresh_put_quota
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.take()
}
#[allow(dead_code, reason = "snapshot-scope predicate asserted by this file's tests (backlog#1823)")]
pub(crate) fn is_for_store_bucket(
&self,
+113 -36
View File
@@ -1288,7 +1288,7 @@ async fn try_create_bitrot_readers_via_batch_pread(
// parity readers unopened; preading them here would discard their bytes
// when fill_deferred_bitrot_readers installs stripe-aligned reserves.
let data_first = strategy != BitrotReaderSetupStrategy::AllShards
&& crate::erasure::coding::decode::get_lockstep_data_shards_only_enabled();
&& crate::erasure::coding::decode::get_lockstep_data_shards_only_with_deferred_parity_enabled();
let mut setup = BitrotReaderSetup::new(disks.len());
for (idx, disk) in disks.iter().enumerate() {
if disk.is_none() {
@@ -2549,6 +2549,8 @@ pub(in crate::set_disk) struct RenameTailCleanup {
pub(in crate::set_disk) struct RenameTailOutcome {
pub(in crate::set_disk) convergence: RenameConvergence,
pub(in crate::set_disk) cleanup: Vec<RenameTailCleanup>,
pub(in crate::set_disk) old_current_source_all_successful_checked: bool,
pub(in crate::set_disk) all_online_local_succeeded: bool,
}
const RENAME_TAIL_CLEANUP_COUNTERFACTUAL_SKIP_ENV: &str = "RUSTFS_PUT_RENAME_TAIL_CLEANUP_COUNTERFACTUAL_SKIP";
@@ -3572,7 +3574,7 @@ pub(in crate::set_disk) async fn finish_rename_tail_heal<
Guards: Send + 'static,
Finalize: FnOnce() -> FinalizeFuture + Send,
FinalizeFuture: Future<Output = ()> + Send,
Cleanup: FnOnce(Guards, Vec<RenameTailCleanup>) -> CleanupFuture + Send,
Cleanup: FnOnce(Guards, Vec<RenameTailCleanup>, bool, bool) -> CleanupFuture + Send,
CleanupFuture: Future<Output = ()> + Send,
Cleanup: 'static,
CleanupFuture: 'static,
@@ -3580,34 +3582,41 @@ pub(in crate::set_disk) async fn finish_rename_tail_heal<
SubmitFuture: Future<Output = ()> + Send,
{
let tail_async_started = rustfs_io_metrics::put_stage_timer();
let (needs_heal, tail_cleanup, tail_complete) = match tail_drain.await {
Ok(Some(outcome)) => (outcome.convergence.needs_heal(), outcome.cleanup, true),
Ok(None) => {
warn!(
event = EVENT_SET_DISK_RENAME_TAIL_DRAIN_FAILED,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_SET_DISK,
state = "missing_convergence",
bucket = %request.bucket,
object = request.object_prefix.as_deref().unwrap_or_default(),
"committed rename tail drain completed without convergence"
);
(true, Vec::new(), false)
}
Err(err) => {
warn!(
event = EVENT_SET_DISK_RENAME_TAIL_DRAIN_FAILED,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_SET_DISK,
state = "failed",
bucket = %request.bucket,
object = request.object_prefix.as_deref().unwrap_or_default(),
error = %err,
"rename tail drain failed"
);
(true, Vec::new(), false)
}
};
let (needs_heal, tail_cleanup, tail_complete, old_current_source_all_successful_checked, all_online_local_succeeded) =
match tail_drain.await {
Ok(Some(outcome)) => (
outcome.convergence.needs_heal(),
outcome.cleanup,
true,
outcome.old_current_source_all_successful_checked,
outcome.all_online_local_succeeded,
),
Ok(None) => {
warn!(
event = EVENT_SET_DISK_RENAME_TAIL_DRAIN_FAILED,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_SET_DISK,
state = "missing_convergence",
bucket = %request.bucket,
object = request.object_prefix.as_deref().unwrap_or_default(),
"committed rename tail drain completed without convergence"
);
(true, Vec::new(), false, false, false)
}
Err(err) => {
warn!(
event = EVENT_SET_DISK_RENAME_TAIL_DRAIN_FAILED,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_SET_DISK,
state = "failed",
bucket = %request.bucket,
object = request.object_prefix.as_deref().unwrap_or_default(),
error = %err,
"rename tail drain failed"
);
(true, Vec::new(), false, false, false)
}
};
rustfs_io_metrics::record_put_object_stage_duration_from(
rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_TAIL_ASYNC_WAIT,
tail_async_started,
@@ -3651,7 +3660,12 @@ pub(in crate::set_disk) async fn finish_rename_tail_heal<
let job = RenameTailCleanupJob {
enqueued_at: Instant::now(),
targets,
future: Box::pin(cleanup(guards, tail_cleanup)),
future: Box::pin(cleanup(
guards,
tail_cleanup,
old_current_source_all_successful_checked,
all_online_local_succeeded,
)),
};
if let Err(job) = rename_tail_cleanup_queue().try_enqueue(job).await {
rustfs_io_metrics::record_put_object_stage_duration_from(
@@ -3676,7 +3690,13 @@ pub(in crate::set_disk) async fn finish_rename_tail_heal<
cleanup_submit_started,
);
let cleanup_started = rustfs_io_metrics::put_stage_timer();
cleanup(guards, tail_cleanup).await;
cleanup(
guards,
tail_cleanup,
old_current_source_all_successful_checked,
all_online_local_succeeded,
)
.await;
rustfs_io_metrics::record_put_object_stage_duration_from(
rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_TAIL_CLEANUP,
cleanup_started,
@@ -4155,7 +4175,14 @@ impl SetDisks {
}
}
if !sent_commit && success_count >= write_quorum {
let pending_tasks = tasks.len();
let checked_count = old_current_source_checked.iter().filter(|checked| **checked).count();
let max_source_votes = Self::max_common_old_current_source_votes(&old_current_sources);
let can_change_old_current_source = (checked_count < write_quorum
&& checked_count + pending_tasks >= write_quorum)
|| (max_source_votes < write_quorum && max_source_votes + pending_tasks >= write_quorum);
if !sent_commit && success_count >= write_quorum && !can_change_old_current_source {
let snapshot_commit = Self::rename_data_commit_from_observations(
RenameDataObservations {
disks: &coordinator_disks,
@@ -4174,6 +4201,8 @@ impl SetDisks {
cleanup_at_snapshot = commit.cleanup_disks.iter().map(Option::is_some).collect();
}
if let Some(commit_tx) = commit_tx.take() {
#[cfg(test)]
rename_fanout_barrier::wait_for_armed_rename_before_early_ack(&fanout_dst_object).await;
let _ = commit_tx.send(snapshot_commit);
}
sent_commit = true;
@@ -4296,9 +4325,21 @@ impl SetDisks {
}
}
let old_current_source_all_successful_checked = errs
.iter()
.enumerate()
.filter(|(_, err)| err.is_none())
.all(|(idx, _)| old_current_source_checked.get(idx).copied().unwrap_or(false));
let all_online_local_succeeded = coordinator_disks.iter().zip(errs.iter()).all(|(disk, err)| match disk {
Some(d) => d.is_local() && err.is_none(),
None => true,
});
Some(RenameTailOutcome {
convergence: final_convergence,
cleanup: tail_cleanup,
old_current_source_all_successful_checked,
all_online_local_succeeded,
})
}
});
@@ -4863,6 +4904,14 @@ impl SetDisks {
})
}
pub(in crate::set_disk) fn max_common_old_current_source_votes(old_current_sources: &[Option<FileInfo>]) -> usize {
let mut counts: HashMap<[u8; 32], usize> = HashMap::new();
for source in old_current_sources.iter().flatten() {
*counts.entry(Self::file_info_quorum_hash(source)).or_insert(0) += 1;
}
counts.into_values().max().unwrap_or(0)
}
pub(in crate::set_disk) fn reduce_common_old_current_source(
old_current_sources: &[Option<FileInfo>],
write_quorum: usize,
@@ -7422,6 +7471,34 @@ pub(crate) mod rename_fanout_barrier {
}
}
pub(in crate::set_disk) async fn wait_for_armed_rename_before_early_ack(object: &str) {
loop {
let (any_armed, any_unpaused) = {
let reg = lock();
let mut any_armed = false;
let mut any_unpaused = false;
for ((obj, _, phase), armed) in &reg.armed {
if obj == object && *phase == Phase::Rename {
any_armed = true;
if !armed.paused.load(Ordering::SeqCst) {
any_unpaused = true;
break;
}
}
}
(any_armed, any_unpaused)
};
if !any_armed {
return;
}
if !any_unpaused {
tokio::task::yield_now().await;
return;
}
tokio::task::yield_now().await;
}
}
/// RAII scope that observes in-flight fan-out task counts for a single
/// `object`. Counts accrue only while the scope is alive; it clears its own
/// entry on drop.
@@ -8488,9 +8565,9 @@ mod tests {
}
assert_eq!(
recorder.counter_value("rustfs_io_get_object_local_batch_shards_total", &[("outcome", "success")]),
4
2
);
assert_eq!(recorder.counter_value("rustfs_io_get_object_local_batch_read_bytes_total", &[]), 14);
assert_eq!(recorder.counter_value("rustfs_io_get_object_local_batch_read_bytes_total", &[]), 7);
assert_eq!(recorder.histogram_sample_count("rustfs_io_get_object_local_batch_prepare_seconds"), 8);
}
@@ -10915,7 +10992,7 @@ mod tests {
move || async move {
*finalize_captured.lock().expect("finalize recorder should not poison") = true;
},
move |(), _| async move {
move |(), _, _, _| async move {
*cleanup_captured.lock().expect("cleanup recorder should not poison") = true;
},
|request| async move {
@@ -97,6 +97,14 @@ pub(crate) mod test_hooks {
.push((bucket.to_owned(), object.to_owned(), hook));
}
pub(super) fn has_before_gc_finalize(bucket: &str, object: &str) -> bool {
BEFORE_GC_FINALIZE
.lock()
.expect("GC finalize hooks")
.iter()
.any(|(hook_bucket, hook_object, _)| hook_bucket == bucket && hook_object == object)
}
pub(super) async fn run_before_gc_finalize(bucket: &str, object: &str) {
let hook = {
let mut hooks = BEFORE_GC_FINALIZE.lock().expect("GC finalize hooks");
@@ -418,6 +426,59 @@ impl SetDisks {
.unwrap_or_else(|| self.ctx.begin_namespace_commit());
context.namespace_owner = Some(Arc::new((owner, Arc::clone(&transaction_guard))));
let can_commit = context.can_commit;
let can_try_local_fast_path = disks.len() == self.set_drive_count
&& disks.iter().all(|d| d.as_ref().is_some_and(|d| d.is_local()))
&& !crate::disk::local::has_delete_version_test_failpoints()
&& {
#[cfg(test)]
{
!test_hooks::has_before_gc_finalize(bucket, object)
}
#[cfg(not(test))]
{
true
}
};
if can_try_local_fast_path && can_commit() {
let fast_published = join_all(disks.iter().map(|disk| async {
if !can_commit() {
return Err(DiskError::other("delete recovery namespace fence was lost"));
}
let disk = disk.as_ref().ok_or(DiskError::DiskNotFound)?;
disk.try_publish_prepared_delete_local(bucket, object, transaction, context.namespace_owner.clone())
.await
}))
.await;
let all_fast_published = fast_published.len() == self.set_drive_count
&& fast_published.iter().all(|res| matches!(res, Ok(Some((_, true, true)))));
if all_fast_published && can_commit() {
#[cfg(test)]
test_hooks::record_cleanup_attempt(bucket, object);
let finalized = join_all(disks.iter().zip(fast_published.iter()).map(|(disk, pub_res)| {
let ns_owner = context.namespace_owner.clone();
async move {
if !can_commit() {
return Ok(false);
}
let Some(disk) = disk.as_ref() else {
return Ok(false);
};
let Ok(Some((journal, _, _))) = pub_res else {
return Ok(false);
};
disk.try_finish_and_finalize_prepared_delete_local(bucket, object, transaction, journal, ns_owner)
.await
}
}))
.await;
if finalized.iter().all(|res| matches!(res, Ok(true))) {
if let Some(ticket) = context.recovery_ticket.as_ref() {
self.ctx.delete_recovery_hints().ack(ticket);
}
return Ok(());
}
}
}
let states = inspect(disks, bucket, object, transaction).await;
let untouched_absence = |state: &Result<DeleteRecoveryInspection>| matches!(state, Ok(state) if state.state == DeleteRecoveryState::Absent && state.current_digest.is_none());
if states.iter().filter(|state| untouched_absence(state)).count() >= write_quorum
+27 -13
View File
@@ -158,13 +158,15 @@ fn rename_tail_cleanup_zero_target_tmp_delete_skip_enabled() -> bool {
}
fn tier_free_version_source_reuse_rename_data_enabled() -> bool {
static ENABLED: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
*ENABLED.get_or_init(|| {
matches!(
std::env::var(TIER_FREE_VERSION_SOURCE_REUSE_RENAME_DATA_ENV).ok().as_deref(),
Some("1" | "true" | "TRUE" | "yes" | "YES" | "on" | "ON")
)
})
#[cfg(test)]
{
rustfs_utils::get_env_bool(TIER_FREE_VERSION_SOURCE_REUSE_RENAME_DATA_ENV, true)
}
#[cfg(not(test))]
{
static ENABLED: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
*ENABLED.get_or_init(|| rustfs_utils::get_env_bool(TIER_FREE_VERSION_SOURCE_REUSE_RENAME_DATA_ENV, true))
}
}
fn tier_free_version_source_reuse_rename_data_assume_homogeneous_enabled() -> bool {
@@ -5523,8 +5525,10 @@ impl SetDisks {
);
let tier_free_version_source_lookup_started = rustfs_io_metrics::put_stage_timer();
let all_commit_disks_local = commit_disks.iter().flatten().all(|disk| disk.is_local());
let reuse_rename_data_source_for_tier_free_version = tier_free_version_source_reuse_rename_data_enabled()
&& (commit_set.tier_free_version_rename_data_source_capable()
&& (all_commit_disks_local
|| commit_set.tier_free_version_rename_data_source_capable()
|| tier_free_version_source_reuse_rename_data_assume_homogeneous_enabled());
let mut put_tier_free_version_source = if commit_put_tier_free_version_id.is_some()
&& commit_tier_free_version_receipt_sink.is_some()
@@ -5630,9 +5634,11 @@ impl SetDisks {
);
let lock_held_tail_handoff_started = rustfs_io_metrics::put_stage_timer();
if let Ok(rename_commit) = rename_result.as_mut() {
commit_set.observe_tier_free_version_rename_data_source_capability(
rename_commit.tail_drain.is_none() && rename_commit.old_current_source_all_successful_checked,
);
if rename_commit.tail_drain.is_none() || !rename_commit.old_current_source_all_successful_checked {
commit_set.observe_tier_free_version_rename_data_source_capability(
rename_commit.tail_drain.is_none() && rename_commit.old_current_source_all_successful_checked,
);
}
commit_set.record_capacity_scope_if_needed(commit_capacity_scope_token, &rename_commit.capacity_disks);
// Install the tail watcher before any post-commit await. The
// latch keeps namespace guards through their prior handoff point.
@@ -5700,7 +5706,12 @@ impl SetDisks {
decommission_object_lock_guard,
decommission_capacity_guard,
),
targets| {
targets,
old_current_source_all_successful_checked,
all_online_local_succeeded| {
cleanup_set.observe_tier_free_version_rename_data_source_capability(
old_current_source_all_successful_checked,
);
drop(object_lock_guard);
drop(write_context);
drop(publication_guard);
@@ -5727,7 +5738,10 @@ impl SetDisks {
)
.await;
let tmp_delete_started = rustfs_io_metrics::put_stage_timer();
if zero_target_cleanup && rename_tail_cleanup_zero_target_tmp_delete_skip_enabled() {
if zero_target_cleanup
&& (all_online_local_succeeded
|| rename_tail_cleanup_zero_target_tmp_delete_skip_enabled())
{
rustfs_io_metrics::record_put_object_stage_duration(
rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_TAIL_CLEANUP_TMP_DELETE_COUNTERFACTUAL_SKIP,
1.0,
+44 -31
View File
@@ -1388,6 +1388,18 @@ impl crate::storage_api_contracts::object::ObjectIO for ECStore {
}
impl ECStore {
pub(crate) fn all_disks_local_and_online(&self) -> bool {
!self.disk_map.is_empty()
&& self.disk_map.values().all(|disks| {
!disks.is_empty()
&& disks.iter().all(|disk| {
disk.as_ref().is_some_and(|d| {
d.is_local() && d.runtime_state() == crate::disk::health_state::RuntimeDriveHealthState::Online
})
})
})
}
/// `put_object` plus the rename_data old-size backfill
/// (rustfs/backlog#1009); see `SetDisks::put_object_with_old_current_size`.
/// Post-write hooks (immediate ILM transition enqueue, list-cache
@@ -1400,19 +1412,36 @@ impl ECStore {
data: &mut PutObjReader,
opts: &ObjectOptions,
) -> Result<(ObjectInfo, Option<crate::disk::OldCurrentSize>)> {
let result = match self.handle_put_object(bucket, object, data, opts).await {
Ok((object_info, old_current_size)) => {
enqueue_transition_after_write(self, Ok(object_info), LcEventSrc::S3PutObject, opts)
.await
.map(|object_info| (object_info, old_current_size))
}
Err(err) => Err(err),
};
let result = self.handle_put_object(bucket, object, data, opts).await;
if result.is_ok() {
list_objects::observe_scanner_namespace_mutations(bucket, 1);
}
result
}
#[allow(clippy::too_many_arguments)]
pub async fn copy_object_with_old_current_size(
&self,
src_bucket: &str,
src_object: &str,
dst_bucket: &str,
dst_object: &str,
src_info: &mut ObjectInfo,
src_opts: &ObjectOptions,
dst_opts: &ObjectOptions,
) -> Result<(ObjectInfo, Option<crate::disk::OldCurrentSize>)> {
let (oi, old_current_size) = box_copy_object_handler(move || {
self.handle_copy_object(src_bucket, src_object, dst_bucket, dst_object, src_info, src_opts, dst_opts)
})
.await?;
let result = enqueue_transition_after_write(self, Ok(oi), LcEventSrc::S3CopyObject, dst_opts)
.await
.map(|oi| (oi, old_current_size));
if result.is_ok() {
list_objects::observe_scanner_namespace_mutations(dst_bucket, 1);
}
result
}
}
lazy_static! {
@@ -1490,20 +1519,9 @@ impl crate::storage_api_contracts::object::ObjectOperations for ECStore {
src_opts: &ObjectOptions,
dst_opts: &ObjectOptions,
) -> Result<ObjectInfo> {
let result = enqueue_transition_after_write(
self,
box_copy_object_handler(move || {
self.handle_copy_object(src_bucket, src_object, dst_bucket, dst_object, src_info, src_opts, dst_opts)
})
.await,
LcEventSrc::S3CopyObject,
dst_opts,
)
.await;
if result.is_ok() {
list_objects::observe_scanner_namespace_mutations(dst_bucket, 1);
}
result
self.copy_object_with_old_current_size(src_bucket, src_object, dst_bucket, dst_object, src_info, src_opts, dst_opts)
.await
.map(|(oi, _)| oi)
}
#[instrument(skip(self))]
@@ -1763,15 +1781,10 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for ECStore {
uploaded_parts: Vec<CompletePart>,
opts: &ObjectOptions,
) -> Result<ObjectInfo> {
let result = enqueue_transition_after_write(
self.as_ref(),
self.clone()
.handle_complete_multipart_upload(bucket, object, upload_id, uploaded_parts, opts)
.await,
LcEventSrc::S3CompleteMultipartUpload,
opts,
)
.await;
let result = self
.clone()
.handle_complete_multipart_upload(bucket, object, upload_id, uploaded_parts, opts)
.await;
if result.is_ok() {
list_objects::observe_scanner_namespace_mutations(bucket, 1);
}
+31 -7
View File
@@ -640,6 +640,19 @@ impl ECStore {
return Ok((res, idx, opts.expected_bucket_incarnation_id));
}
if self.single_pool() && !self.is_suspended(0).await && !self.is_pool_rebalancing(0).await {
self.apply_decommission_target_mutation_fence(0, object, &mut opts, mutation_fence)
.await;
let expected_bucket_incarnation_id = opts.expected_bucket_incarnation_id;
let lock_object = encode_dir_object(object);
let res = self
.run_external_decommission_capacity_object_mutation(0, bucket, &lock_object, object, opts, |opts| async move {
self.pools[0].new_multipart_upload(bucket, object, &opts).await
})
.await?;
return Ok((res, 0, expected_bucket_incarnation_id));
}
for (idx, pool) in self.pools.iter().enumerate() {
if self.is_suspended(idx).await || self.is_pool_rebalancing(idx).await {
continue;
@@ -1167,10 +1180,11 @@ impl ECStore {
let opts = &opts;
if self.single_pool() {
return self.pools[0]
let result = self.pools[0]
.clone()
.complete_multipart_upload(bucket, object, upload_id, uploaded_parts, opts)
.await;
return enqueue_transition_after_write(self.as_ref(), result, LcEventSrc::S3CompleteMultipartUpload, opts).await;
}
if opts.http_preconditions.is_some()
&& !opts.data_movement
@@ -1179,7 +1193,7 @@ impl ECStore {
return Err(err);
}
match self.multipart_upload_pool_idx(bucket, object, upload_id, opts).await {
let result = match self.multipart_upload_pool_idx(bucket, object, upload_id, opts).await {
Ok(pool_idx) => {
self.pools[pool_idx]
.clone()
@@ -1190,21 +1204,29 @@ impl ECStore {
// the pool that published it; other pools answer InvalidUploadID.
Err(err) if is_err_invalid_upload_id(&err) => {
let mut missing = err;
let mut found = None;
for pool_idx in self.existing_multipart_pool_order().await {
match self.pools[pool_idx]
.clone()
.complete_multipart_upload(bucket, object, upload_id, uploaded_parts.clone(), opts)
.await
{
Ok(info) => return Ok(info),
Ok(info) => {
found = Some(Ok(info));
break;
}
Err(err) if is_err_invalid_upload_id(&err) => missing = err,
Err(err) => return Err(err),
Err(err) => {
found = Some(Err(err));
break;
}
}
}
Err(missing)
found.unwrap_or(Err(missing))
}
Err(err) => Err(err),
}
};
enqueue_transition_after_write(self.as_ref(), result, LcEventSrc::S3CompleteMultipartUpload, opts).await
}
#[cfg(all(test, feature = "test-util"))]
@@ -1326,6 +1348,7 @@ impl ECStore {
// Data movement already owns the pool-meta write lease. Forward its
// loss signal into SetDisks so commit fencing observes the same lease
// without trying to reacquire the namespace.
let mut transition_opts = opts.clone();
let result = self
.run_decommission_capacity_admitted_mutation_with_capacity_lease(
target_pool_idx,
@@ -1341,8 +1364,9 @@ impl ECStore {
)
.await;
drop(publication_guard);
transition_opts.no_lock = request_opts.no_lock;
let result =
enqueue_transition_after_write(self.as_ref(), result, LcEventSrc::S3CompleteMultipartUpload, request_opts).await;
enqueue_transition_after_write(self.as_ref(), result, LcEventSrc::S3CompleteMultipartUpload, &transition_opts).await;
if result.is_ok() {
list_objects::observe_scanner_namespace_mutations(bucket, 1);
}
+104 -22
View File
@@ -31,7 +31,7 @@ use crate::bucket::lifecycle::{
};
use crate::bucket::metadata_sys::{
acquire_bucket_metadata_transaction_read_lock_in, get_cached_bucket_incarnation_id_in, get_delete_config_from_disk_in,
get_object_lock_config_and_incarnation_from_disk_in,
get_object_lock_and_quota_config_from_disk_with_guard_in, get_object_lock_config_and_incarnation_from_disk_with_guard_in,
};
use crate::bucket::object_lock::objectlock_sys::{
check_object_lock_for_deletion_with_state, ensure_recursive_force_delete_allowed_for_state,
@@ -3162,7 +3162,7 @@ impl ECStore {
let lifecycle_guard = self.acquire_bucket_lifecycle_read_lock(bucket).await?;
let metadata_guard = acquire_bucket_metadata_transaction_read_lock_in(&self.ctx, bucket).await?;
let (state, bucket_incarnation_id, config_revision) =
get_object_lock_config_and_incarnation_from_disk_in(&self.ctx, bucket).await?;
get_object_lock_config_and_incarnation_from_disk_with_guard_in(&self.ctx, bucket, Some(&lifecycle_guard)).await?;
if lifecycle_guard.is_lock_lost() || metadata_guard.is_lock_lost() {
return Err(Error::other("bucket lifecycle lock was lost while loading the Object Lock snapshot"));
}
@@ -3177,6 +3177,30 @@ impl ECStore {
)))
}
async fn object_lock_config_snapshot_for_single_put(&self, bucket: &str) -> Result<Arc<ObjectLockConfigSnapshot>> {
check_valid_bucket_name(bucket)?;
let lifecycle_guard = self.acquire_bucket_lifecycle_read_lock(bucket).await?;
let metadata_guard = acquire_bucket_metadata_transaction_read_lock_in(&self.ctx, bucket).await?;
let (state, bucket_incarnation_id, config_revision, quota_res) =
get_object_lock_and_quota_config_from_disk_with_guard_in(&self.ctx, bucket, Some(&lifecycle_guard)).await?;
if lifecycle_guard.is_lock_lost() || metadata_guard.is_lock_lost() {
return Err(Error::other("bucket lifecycle lock was lost while loading the Object Lock snapshot"));
}
let snapshot = ObjectLockConfigSnapshot::for_guarded_store_bucket(
self.id,
bucket,
bucket_incarnation_id,
config_revision,
state,
lifecycle_guard,
metadata_guard,
);
if !self.ctx.lock_manager().is_disabled() {
snapshot.set_fresh_put_quota(quota_res);
}
Ok(Arc::new(snapshot))
}
pub(super) async fn object_lock_config_snapshot_under_lifecycle_fence(
&self,
bucket: &str,
@@ -3187,7 +3211,7 @@ impl ECStore {
}
let metadata_guard = acquire_bucket_metadata_transaction_read_lock_in(&self.ctx, bucket).await?;
let (state, bucket_incarnation_id, config_revision) =
get_object_lock_config_and_incarnation_from_disk_in(&self.ctx, bucket).await?;
get_object_lock_config_and_incarnation_from_disk_with_guard_in(&self.ctx, bucket, Some(&metadata_guard)).await?;
if lifecycle_fence.is_lock_lost() || metadata_guard.is_lock_lost() {
return Err(Error::other("bucket lock was lost while loading the Object Lock snapshot"));
}
@@ -4652,7 +4676,7 @@ impl ECStore {
.expected_bucket_incarnation_id
.ok_or_else(|| Error::other("destructive PUT is missing its bucket incarnation"))?;
if opts.object_lock_config_snapshot.is_none() {
opts.object_lock_config_snapshot = Some(self.object_lock_config_snapshot(bucket).await?);
opts.object_lock_config_snapshot = Some(self.object_lock_config_snapshot_for_single_put(bucket).await?);
}
let snapshot = match opts.object_lock_config_snapshot.as_ref() {
Some(snapshot) if snapshot.is_valid_for_destructive_put(self.id, bucket, expected_incarnation_id) => {
@@ -4741,6 +4765,7 @@ impl ECStore {
self.apply_decommission_target_mutation_fence(idx, object.as_str(), &mut opts, mutation_fence.or(fixed_read_anchor))
.await;
let expected_data_bytes = usize::try_from(data.size()).ok();
let mut transition_opts = opts.clone();
let result = self
.run_decommission_capacity_admitted_mutation_with_capacity_lease(
idx,
@@ -4772,7 +4797,8 @@ impl ECStore {
},
)
.await;
let result = enqueue_transition_after_write(self, result, LcEventSrc::S3PutObject, request_opts).await;
transition_opts.no_lock = request_opts.no_lock;
let result = enqueue_transition_after_write(self, result, LcEventSrc::S3PutObject, &transition_opts).await;
if result.is_ok() {
list_objects::observe_scanner_namespace_mutations(bucket, 1);
}
@@ -4788,7 +4814,7 @@ impl ECStore {
data: &mut PutObjReader,
opts: &ObjectOptions,
) -> Result<(ObjectInfo, Option<OldCurrentSize>)> {
let (object, opts) = self.prepare_put_object(bucket, object, opts).await?;
let (object, opts) = Box::pin(self.prepare_put_object(bucket, object, opts)).await?;
let idx = self
.select_put_object_pool_idx(bucket, object.as_str(), data.size(), &opts)
.await?;
@@ -4797,11 +4823,16 @@ impl ECStore {
opts.decommission_capacity_admission = crate::bucket::metadata_sys::object_store_if_initialized_in(&self.ctx).await;
}
let receipt_sink = install_tier_free_version_receipt_sink(&mut opts);
let result = self.pools[idx]
.put_object_with_old_current_size(bucket, object.as_str(), data, &opts)
.await;
let result = Box::pin(self.pools[idx].put_object_with_old_current_size(bucket, object.as_str(), data, &opts)).await;
enqueue_recorded_tier_free_versions(self, receipt_sink).await;
result
match result {
Ok((object_info, old_current_size)) => {
Box::pin(enqueue_transition_after_write(self, Ok(object_info), LcEventSrc::S3PutObject, &opts))
.await
.map(|object_info| (object_info, old_current_size))
}
Err(err) => Err(err),
}
}
#[instrument(level = "trace", skip(self))]
@@ -4859,7 +4890,7 @@ impl ECStore {
src_info: &mut ObjectInfo,
src_opts: &ObjectOptions,
dst_opts: &ObjectOptions,
) -> Result<ObjectInfo> {
) -> Result<(ObjectInfo, Option<OldCurrentSize>)> {
check_copy_obj_args(src_bucket, src_object)?;
if src_bucket == dst_bucket && src_object == dst_object {
// In-place copies (metadata replace, restore) rewrite a key that
@@ -4938,9 +4969,17 @@ impl ECStore {
};
if cp_src_dst_same {
let (_, pool_idx) = self
.get_latest_accessible_object_info_with_idx(src_bucket, &src_object, &version_aware_lookup_opts(src_opts, true))
.await?;
let pool_idx = if self.single_pool() {
0
} else {
self.get_latest_accessible_object_info_with_idx(
src_bucket,
&src_object,
&version_aware_lookup_opts(src_opts, true),
)
.await?
.1
};
let source_pool_writable = !self.is_suspended(pool_idx).await && !self.is_pool_rebalancing(pool_idx).await;
if source_pool_writable
@@ -4961,7 +5000,8 @@ impl ECStore {
.await
},
)
.await;
.await
.map(|oi| (oi, None));
}
if source_pool_writable && !dst_opts.versioned && src_opts.version_id.is_none() {
@@ -4989,7 +5029,8 @@ impl ECStore {
.await
},
)
.await;
.await
.map(|oi| (oi, None));
}
// Transitioned object self-copy: restore from tier into the same pool.
let mut put_opts = ObjectOptions {
@@ -5017,7 +5058,7 @@ impl ECStore {
return if let Some(reader) = src_info.put_object_reader.as_mut() {
let receipt_sink = install_tier_free_version_receipt_sink(&mut put_opts);
let result = self.pools[pool_idx]
.put_object(dst_bucket, &dst_object, reader, &put_opts)
.put_object_with_old_current_size(dst_bucket, &dst_object, reader, &put_opts)
.await;
enqueue_recorded_tier_free_versions(self, receipt_sink).await;
result
@@ -5061,7 +5102,7 @@ impl ECStore {
}
let receipt_sink = install_tier_free_version_receipt_sink(&mut put_opts);
let result = self.pools[pool_idx]
.put_object(dst_bucket, &dst_object, reader, &put_opts)
.put_object_with_old_current_size(dst_bucket, &dst_object, reader, &put_opts)
.await;
enqueue_recorded_tier_free_versions(self, receipt_sink).await;
return result;
@@ -5081,11 +5122,14 @@ impl ECStore {
.await
},
)
.await;
.await
.map(|oi| (oi, None));
}
}
let pool_idx = if dst_opts.no_lock {
let pool_idx = if self.single_pool() && !self.is_suspended(0).await && !self.is_pool_rebalancing(0).await {
0
} else if dst_opts.no_lock {
self.get_pool_idx_no_lock(dst_bucket, &dst_object, src_info.size).await?
} else {
self.get_pool_idx(dst_bucket, &dst_object, src_info.size).await?
@@ -5118,7 +5162,7 @@ impl ECStore {
if let Some(put_object_reader) = src_info.put_object_reader.as_mut() {
let receipt_sink = install_tier_free_version_receipt_sink(&mut put_opts);
let result = self.pools[pool_idx]
.put_object(dst_bucket, dst_object_name, put_object_reader, &put_opts)
.put_object_with_old_current_size(dst_bucket, dst_object_name, put_object_reader, &put_opts)
.await;
enqueue_recorded_tier_free_versions(self, receipt_sink).await;
return result;
@@ -5350,8 +5394,9 @@ impl ECStore {
opts.add_namespace_lock_guard(guard);
}
let current_bucket_incarnation_id = if let Some(metadata_guard) = _object_lock_metadata_guard.as_ref() {
let guard_ref = _bucket_lifecycle_guard.as_deref().unwrap_or_else(|| metadata_guard.as_ref());
let (state, incarnation_id, config_revision) =
get_delete_config_from_disk_in(&self.ctx, bucket, None, opts.lifecycle_expiry_policy.as_ref()).await?;
get_delete_config_from_disk_in(&self.ctx, bucket, Some(guard_ref), opts.lifecycle_expiry_policy.as_ref()).await?;
opts.object_lock_config_snapshot = Some(Arc::new(ObjectLockConfigSnapshot::for_store_bucket_under_lifecycle_fence(
self.id,
bucket,
@@ -5591,6 +5636,42 @@ impl ECStore {
return Ok(obj);
}
if self.single_pool()
&& !creates_latest_marker
&& !opts.data_movement
&& opts.http_preconditions.is_none()
&& !opts.replication_request
&& !opts.versioned
&& !opts.version_suspended
&& opts.version_id.is_none()
&& !self.is_suspended(0).await
&& !self.is_pool_rebalancing(0).await
{
let pool = self.pools[0].clone();
match self
.run_external_decommission_capacity_object_delete(0, bucket, object, object, opts.clone(), |opts| async move {
pool.delete_object(bucket, object, opts).await
})
.await
{
Ok(mut res) => {
res.name = decode_dir_object(res.name.as_str());
return Ok(res);
}
Err(err) if is_err_read_quorum(&err) => return Err(StorageError::ErasureWriteQuorum),
Err(err) => {
if should_purge_orphan_dir_on_missing(&err, object) && self.purge_orphan_dir_object(bucket, object).await {
return Ok(ObjectInfo {
bucket: bucket.to_owned(),
name: decode_dir_object(object),
..Default::default()
});
}
return Err(err);
}
}
}
// Determine which pool contains it
let existing_pool_info = if creates_latest_marker {
self.get_pool_info_for_delete_marker(bucket, object, &gopts).await
@@ -9903,6 +9984,7 @@ mod tests {
copy_store
.handle_copy_object(bucket, source, bucket, target, &mut source_info, &ObjectOptions::default(), &options)
.await
.map(|(oi, _)| oi)
});
let deadline = tokio::time::Instant::now() + Duration::from_secs(30);
tokio::time::timeout_at(deadline, barrier.wait_until_paused())
-2
View File
@@ -57,8 +57,6 @@ tracing = { workspace = true }
[dev-dependencies]
rustfs-test-utils = { workspace = true, features = ["put-object-commit-barrier"] }
# Match the streaming inference's field order in Arrow's JSON reference.
serde_json = { workspace = true, features = ["preserve_order"] }
[lib]
doctest = false
@@ -426,14 +426,18 @@ impl InferredObject {
#[cfg(test)]
mod tests {
use super::*;
use crate::dispatcher::json_lines_schema;
use datafusion::arrow::json::reader::infer_json_schema as arrow_infer_json_schema;
const TEST_SCHEMA_LIMIT: usize = 4 * 1024 * 1024;
fn assert_matches_arrow(input: &[u8]) -> Schema {
let records = input.iter().filter(|byte| **byte == b'\n').count();
let (expected, expected_records) = arrow_infer_json_schema(io::BufReader::new(input), Some(records))
let (_, expected_records) = arrow_infer_json_schema(io::BufReader::new(input), Some(records))
.expect("Arrow should infer the compatibility fixture");
// Arrow's types in source key order, which is what JSON LINES uses.
let expected = json_lines_schema::infer_schema(input, records)
.expect("JSON LINES inference should accept the compatibility fixture");
let (actual, actual_records, _) = infer_schema(input, records, &AtomicBool::new(false), TEST_SCHEMA_LIMIT)
.expect("streaming inference should accept the compatibility fixture");
assert_eq!(actual_records, expected_records);
@@ -0,0 +1,185 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use datafusion::arrow::{
datatypes::{DataType, FieldRef, Fields, Schema},
error::ArrowError,
json::reader::{ValueIter, infer_json_schema_from_iterator},
};
use serde::de::{self, DeserializeSeed, MapAccess, SeqAccess, Visitor};
use std::{collections::HashMap, fmt, sync::Arc};
/// Infers a JSON LINES schema with Arrow's types and the source's first-seen
/// key order.
///
/// Arrow walks each record as a `serde_json::Map`, which sorts keys unless the
/// process-wide `serde_json/preserve_order` feature is enabled, so `SELECT *`
/// would otherwise reorder fields and CSV output columns.
pub(super) fn infer_schema(bytes: &[u8], max_records: usize) -> Result<Schema, ArrowError> {
let schema = infer_json_schema_from_iterator(ValueIter::new(bytes, Some(max_records)))?;
let mut order = KeyOrder::default();
// Same records as `ValueIter`: one value per line, blank lines skipped.
// Arrow has already rejected invalid UTF-8 and JSON within this prefix.
let records = bytes
.split(|byte| *byte == b'\n')
.filter_map(|line| std::str::from_utf8(line).ok())
.map(str::trim)
.filter(|line| !line.is_empty())
.take(max_records);
for record in records {
let mut deserializer = serde_json::Deserializer::from_str(record);
(&mut order)
.deserialize(&mut deserializer)
.map_err(|error| ArrowError::JsonError(format!("Not valid JSON: {error}")))?;
}
Ok(Schema::new_with_metadata(order.reorder(schema.fields()), schema.metadata().clone()))
}
/// First-seen key order of a JSON object, merged across records. Array
/// elements share their parent's node, matching Arrow's `List<Struct>` shape.
#[derive(Default)]
struct KeyOrder {
positions: HashMap<String, usize>,
children: Vec<KeyOrder>,
}
impl KeyOrder {
fn child(&mut self, key: String) -> &mut KeyOrder {
let next = self.children.len();
let position = *self.positions.entry(key).or_insert(next);
if position == next {
self.children.push(KeyOrder::default());
}
&mut self.children[position]
}
fn reorder(&self, fields: &Fields) -> Fields {
let mut fields: Vec<FieldRef> = fields
.iter()
.map(|field| match self.positions.get(field.name()) {
Some(&position) => {
let data_type = self.children[position].reorder_type(field.data_type());
Arc::new(field.as_ref().clone().with_data_type(data_type))
}
None => Arc::clone(field),
})
.collect();
fields.sort_by_key(|field| self.positions.get(field.name()).copied().unwrap_or(usize::MAX));
fields.into()
}
fn reorder_type(&self, data_type: &DataType) -> DataType {
match data_type {
DataType::Struct(fields) => DataType::Struct(self.reorder(fields)),
DataType::List(item) => {
DataType::List(Arc::new(item.as_ref().clone().with_data_type(self.reorder_type(item.data_type()))))
}
other => other.clone(),
}
}
}
impl<'de> DeserializeSeed<'de> for &mut KeyOrder {
type Value = ();
fn deserialize<D: de::Deserializer<'de>>(self, deserializer: D) -> Result<(), D::Error> {
deserializer.deserialize_any(self)
}
}
impl<'de> Visitor<'de> for &mut KeyOrder {
type Value = ();
fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
formatter.write_str("a JSON value")
}
fn visit_map<A: MapAccess<'de>>(self, mut map: A) -> Result<(), A::Error> {
while let Some(key) = map.next_key::<String>()? {
map.next_value_seed(self.child(key))?;
}
Ok(())
}
fn visit_seq<A: SeqAccess<'de>>(self, mut seq: A) -> Result<(), A::Error> {
while seq.next_element_seed(&mut *self)?.is_some() {}
Ok(())
}
fn visit_bool<E: de::Error>(self, _: bool) -> Result<(), E> {
Ok(())
}
fn visit_i64<E: de::Error>(self, _: i64) -> Result<(), E> {
Ok(())
}
fn visit_u64<E: de::Error>(self, _: u64) -> Result<(), E> {
Ok(())
}
fn visit_f64<E: de::Error>(self, _: f64) -> Result<(), E> {
Ok(())
}
fn visit_str<E: de::Error>(self, _: &str) -> Result<(), E> {
Ok(())
}
fn visit_unit<E: de::Error>(self) -> Result<(), E> {
Ok(())
}
}
#[cfg(test)]
mod tests {
use super::infer_schema;
use datafusion::arrow::datatypes::{DataType, Fields};
fn names(fields: &Fields) -> Vec<&str> {
fields.iter().map(|field| field.name().as_str()).collect()
}
#[test]
fn keeps_first_seen_key_order_at_every_level() {
let input = b"{\"z\":1,\"a\":{\"y\":[{\"k2\":1,\"k1\":2}],\"b\":2}}\n\n{\"m\":null,\"a\":{\"c\":3}}\n";
let schema = infer_schema(input, 1000).expect("schema");
assert_eq!(names(schema.fields()), ["z", "a", "m"]);
let DataType::Struct(a) = schema.field(1).data_type() else {
panic!("a should be a struct");
};
assert_eq!(names(a), ["y", "b", "c"]);
let DataType::List(item) = a[0].data_type() else {
panic!("y should be a list");
};
let DataType::Struct(y) = item.data_type() else {
panic!("y items should be structs");
};
assert_eq!(names(y), ["k2", "k1"]);
}
#[test]
fn ignores_records_after_the_inference_prefix_like_arrow() {
let schema = infer_schema(b"{\"b\":1,\"a\":2}\nnot json\n", 1).expect("schema");
assert_eq!(names(schema.fields()), ["b", "a"]);
}
#[test]
fn keeps_arrow_errors_for_invalid_records() {
assert!(infer_schema(b"{\"a\":1}\n[1]\n", 1000).is_err());
assert!(infer_schema(b"{\"a\":\n", 1000).is_err());
}
}
+451 -3
View File
@@ -28,11 +28,12 @@ use datafusion::{
record_batch::RecordBatch,
},
datasource::{
file_format::{csv::CsvFormat, json::JsonFormat},
file_format::{DEFAULT_SCHEMA_INFER_MAX_RECORD, csv::CsvFormat, json::JsonFormat},
listing::{ListingOptions, ListingTable, ListingTableConfig, ListingTableUrl},
},
error::Result as DFResult,
execution::{RecordBatchStream, SendableRecordBatchStream},
object_store::ObjectStoreExt as _,
sql::sqlparser::parser::ParserError,
};
use futures::Stream;
@@ -61,7 +62,7 @@ use tokio::{
};
use crate::{
dispatcher::{json_document_table::JsonDocumentTable, parquet_table::ParquetSelectTable},
dispatcher::{json_document_table::JsonDocumentTable, json_lines_schema, parquet_table::ParquetSelectTable},
execution::factory::QueryExecutionFactoryRef,
instance::{DEFAULT_MAX_CONCURRENT_QUERIES, DEFAULT_QUERY_TIMEOUT_SECS},
metadata::{ContextProviderExtension, MetadataProvider, TableHandleProviderRef, base_table::BaseTableProvider},
@@ -508,7 +509,11 @@ impl SimpleQueryDispatcher {
return Err(SelectError::InvalidDataSource.into());
};
let resolve_schema = listing_options.infer_schema(session.inner(), &table_path).await?;
let resolve_schema = if self.input.request.input_serialization.json.is_some() {
infer_json_lines_schema(session, &table_path).await?
} else {
listing_options.infer_schema(session.inner(), &table_path).await?
};
let config = if need_rename_volume_name {
let mut new_fields = Vec::new();
for (i, field) in resolve_schema.fields().iter().enumerate() {
@@ -549,6 +554,14 @@ impl SimpleQueryDispatcher {
}
}
/// Reads the object like DataFusion's `JsonFormat::infer_schema` does, but
/// keeps the source key order (see `json_lines_schema::infer_schema`).
async fn infer_json_lines_schema(session: &SessionCtx, table_path: &ListingTableUrl) -> DFResult<SchemaRef> {
let store = session.inner().runtime_env().object_store(table_path.object_store())?;
let bytes = store.get(table_path.prefix()).await?.bytes().await?;
Ok(Arc::new(json_lines_schema::infer_schema(&bytes, DEFAULT_SCHEMA_INFER_MAX_RECORD)?))
}
fn is_json_document_input(input: &SelectObjectContentInput) -> bool {
input
.request
@@ -856,6 +869,7 @@ mod tests {
array::{Int32Array, StringArray},
datatypes::{DataType, Field, Schema, SchemaRef},
record_batch::RecordBatch,
util::display::array_value_to_string,
},
common::DataFusionError,
execution::object_store::ObjectStoreUrl,
@@ -1727,6 +1741,334 @@ mod tests {
}
}
async fn json_lines_plan_schema(data: &'static [u8], sql: &str) -> SchemaRef {
let mut input = test_input();
input.key = "records.json".to_owned();
input.request.input_serialization.csv = None;
input.request.input_serialization.json = Some(JSONInput {
type_: Some(JSONType::from_static(JSONType::LINES)),
});
let input = Arc::new(input);
let optimizer = Arc::new(CascadeOptimizerBuilder::default().build());
let dispatcher = test_dispatcher_for_input(
Arc::clone(&input),
Arc::new(Semaphore::new(1)),
Duration::from_secs(30),
Arc::new(SqlQueryExecutionFactory::new(optimizer, Arc::new(LocalScheduler {}))),
);
let query = Query::new(QueryContext { input }, sql.to_owned());
let machine = dispatcher.build_query_state_machine(query).await.expect("build query state");
let store_url = ObjectStoreUrl::parse("s3://test-bucket").expect("test store URL");
let store = machine
.session
.inner()
.runtime_env()
.object_store(&store_url)
.expect("test store");
store
.put(&Path::from("records.json"), data.into())
.await
.expect("write JSON LINES object");
let Plan::Query(plan) = dispatcher
.build_logical_plan(machine)
.await
.expect("plan JSON LINES query")
.expect("select plan");
Arc::new(plan.df_plan.schema().as_arrow().clone())
}
fn field_names(fields: &datafusion::arrow::datatypes::Fields) -> Vec<&str> {
fields.iter().map(|field| field.name().as_str()).collect()
}
fn struct_fields(data_type: &DataType) -> &datafusion::arrow::datatypes::Fields {
match data_type {
DataType::Struct(fields) => fields,
DataType::List(item) => struct_fields(item.data_type()),
other => panic!("expected a struct or list of structs, got {other}"),
}
}
#[tokio::test]
async fn json_lines_select_star_keeps_source_field_order() {
let schema = json_lines_plan_schema(
b"{\"z\":1,\"a b\":{\"y\":true,\"B\":[{\"k2\":1,\"k1\":2}]},\"A\":\"x\"}\n\n \r\n{\"A\":\"y\",\"$c\":2,\"a b\":{\"x\":1}}\r\n",
"SELECT * FROM S3Object",
)
.await;
assert_eq!(field_names(schema.fields()), ["z", "a b", "A", "$c"]);
let nested = struct_fields(schema.field_with_name("a b").expect("nested field").data_type());
assert_eq!(field_names(nested), ["y", "B", "x"]);
let listed = struct_fields(nested.find("B").expect("list field").1.data_type());
assert_eq!(field_names(listed), ["k2", "k1"]);
}
#[tokio::test]
async fn path_projections_are_named_after_their_last_key() {
const DATA: &[u8] = b"{\"obj0\":{\"id\":\"id0\",\"inner\":{\"Key\":1}}}\n";
for (sql, expected) in [
("SELECT s.obj0.id, s.obj0.inner.\"Key\" FROM S3Object s", ["id", "Key"].as_slice()),
("SELECT obj0.id FROM S3Object", &["id"]),
("SELECT s.obj0.id FROM S3Object s ORDER BY s.obj0.id", &["id"]),
("SELECT s.obj0.id, COUNT(*) FROM S3Object s GROUP BY s.obj0.id", &["id", "count(*)"]),
] {
let schema = json_lines_plan_schema(DATA, sql).await;
assert_eq!(field_names(schema.fields()), expected, "{sql}");
}
}
#[tokio::test]
async fn path_projection_names_do_not_clash_with_other_columns() {
const DATA: &[u8] = b"{\"id\":\"top\",\"obj0\":{\"id\":\"id0\"}}\n";
for sql in ["SELECT *, s.obj0.id FROM S3Object s", "SELECT s.*, s.obj0.id FROM S3Object s"] {
let schema = json_lines_plan_schema(DATA, sql).await;
assert_eq!(field_names(schema.fields()), ["id", "obj0", "s.obj0[id]"], "{sql}");
}
for sql in [
"SELECT s.obj0.id FROM S3Object s ORDER BY s.id",
"SELECT s.obj0.id FROM S3Object s ORDER BY id",
"SELECT s.obj0.id FROM S3Object s ORDER BY upper(s.id)",
] {
let schema = json_lines_plan_schema(DATA, sql).await;
assert_eq!(field_names(schema.fields()), ["s.obj0[id]"], "{sql}");
}
let schema = json_lines_plan_schema(DATA, "SELECT s.obj0.id, (s.id) FROM S3Object s").await;
assert_eq!(field_names(schema.fields()), ["s.obj0[id]", "id"]);
}
#[tokio::test]
async fn path_aliases_do_not_collide_with_retained_path_names() {
const DATA: &[u8] = br#"{"id":"top","obj0":{"id":"a"},"obj1":{"id":"b"},"obj2":{"s.obj0[id]":"c","obj0":"d"}}"#;
for (sql, expected) in [
(
"SELECT s.obj0.id, s.obj1.id, s.obj2.\"s.obj0[id]\" FROM S3Object s",
vec!["s.obj0[id]", "s.obj1[id]", "s.obj2[s.obj0[id]]"],
),
(
"SELECT s.obj2.\"s.obj0[id]\", s.obj1.id, s.obj0.id FROM S3Object s",
vec!["s.obj2[s.obj0[id]]", "s.obj1[id]", "s.obj0[id]"],
),
("SELECT (s.obj0.id), s.obj2.\"s.obj0[id]\" FROM S3Object s", vec!["id", "s.obj0[id]"]),
(
"SELECT s.obj0.id, s.obj2.\"s.obj0[id]\" FROM S3Object s ORDER BY s.id",
vec!["s.obj0[id]", "s.obj2[s.obj0[id]]"],
),
("SELECT s.obj2.obj0 FROM S3Object s ORDER BY s.obj0.id", vec!["s.obj2[obj0]"]),
("SELECT s.obj2.obj0 FROM S3Object s ORDER BY (s.obj0.id)", vec!["s.obj2[obj0]"]),
] {
let schema = json_lines_plan_schema(DATA, sql).await;
assert_eq!(field_names(schema.fields()), expected, "{sql}");
}
}
const PATH_NAMES_DATA: &[u8] = br#"{"id":"A","x":"B","n":1,"obj":{"lower(s.id)":7,"count(*)":3,"id":"x","ID":"X","s.n":2,"s.n + Int64(1)":3},"obj0":{"id":"a"},"obj1":{"id":"b"},"obj2":{"s.obj0[id]":"c","obj0":"d"},"a":{"id":1},"b":{"id":2}}
"#;
/// Plans and runs `sql` over a stored JSON LINES object through the
/// production dispatcher. Returns the planned output names and the rows.
async fn production_json_lines_query(bucket: &str, key: &str, sql: &str) -> QueryResult<(Vec<String>, Vec<Vec<String>>)> {
let mut input = (*json_snapshot_input()).clone();
input.bucket = bucket.to_owned();
input.key = key.to_owned();
input.request.expression = sql.to_owned();
let input = Arc::new(input);
let snapshot = snapshot_test_env().await.prepare_select_object_snapshot(bucket, key).await;
let dispatcher = production_dispatcher(Arc::clone(&input));
let query = Query::new_with_snapshot(QueryContext { input }, sql.to_owned(), snapshot);
let machine = dispatcher.build_query_state_machine(query).await?;
let plan = dispatcher
.build_logical_plan(Arc::clone(&machine))
.await?
.expect("SELECT should produce a logical plan");
let Plan::Query(query_plan) = &plan;
let names = query_plan
.df_plan
.schema()
.fields()
.iter()
.map(|field| field.name().clone())
.collect();
let Output::StreamData(stream) = dispatcher.execute_logical_plan(plan, machine).await? else {
panic!("SELECT should return rows");
};
let batches: Vec<_> = stream.try_collect().await?;
let rows = batches
.iter()
.flat_map(|batch| {
(0..batch.num_rows()).map(|row| {
batch
.columns()
.iter()
.map(|column| array_value_to_string(column, row).expect("display a result value"))
.collect()
})
})
.collect();
Ok((names, rows))
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn production_path_alias_does_not_duplicate_a_retained_path_name() {
const BUCKET: &str = "s3select-retained-path-name";
const KEY: &str = "retained.jsonl";
let env = snapshot_test_env().await;
env.make_bucket(BUCKET, false).await;
env.put_object_bytes(
BUCKET,
KEY,
br#"{"obj0":{"id":"a"},"obj1":{"id":"b"},"obj2":{"s.obj0[id]":"c"}}
"#
.to_vec(),
)
.await;
let (names, rows) =
production_json_lines_query(BUCKET, KEY, "SELECT s.obj0.id, s.obj1.id, s.obj2.\"s.obj0[id]\" FROM S3Object s")
.await
.expect("the query plans and runs");
assert_eq!(names, ["s.obj0[id]", "s.obj1[id]", "s.obj2[s.obj0[id]]"]);
assert_eq!(rows, [["a", "b", "c"]]);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn production_path_projections_take_their_key_only_when_it_is_a_unique_output_name() {
const BUCKET: &str = "s3select-path-projection-names";
const KEY: &str = "names.jsonl";
let env = snapshot_test_env().await;
env.make_bucket(BUCKET, false).await;
env.put_object_bytes(BUCKET, KEY, PATH_NAMES_DATA.to_vec()).await;
let cases: &[(&str, &[&str])] = &[
(
"SELECT s.obj.\"lower(s.id)\", LOWER(s.id) FROM S3Object s",
&["s.obj[lower(s.id)]", "lower(s.id)"],
),
(
"SELECT LOWER(s.id), s.obj.\"lower(s.id)\" FROM S3Object s",
&["lower(s.id)", "s.obj[lower(s.id)]"],
),
(
"SELECT s.obj.\"count(*)\", COUNT(*) FROM S3Object s GROUP BY s.obj.\"count(*)\"",
&["s.obj[count(*)]", "count(*)"],
),
("SELECT s.obj0.id, COUNT(*) FROM S3Object s GROUP BY s.obj0.id", &["id", "count(*)"]),
("SELECT s.obj0.id, LOWER(s.id) FROM S3Object s", &["id", "lower(s.id)"]),
("SELECT s.obj0.id, s.x AS id FROM S3Object s", &["s.obj0[id]", "id"]),
("SELECT s.x AS id, s.obj0.id FROM S3Object s", &["id", "s.obj0[id]"]),
(
"SELECT *, s.obj0.id FROM S3Object s",
&["id", "x", "n", "obj", "obj0", "obj1", "obj2", "a", "b", "s.obj0[id]"],
),
("SELECT s.obj.ID, s.id FROM S3Object s", &["s.obj[id]", "id"]),
("SELECT s.obj.\"ID\", s.id FROM S3Object s", &["ID", "id"]),
("SELECT s.obj.\"s.n\", CAST(s.n AS INT) FROM S3Object s", &["s.obj[s.n]", "s.n"]),
(
"SELECT s.n + 1, s.obj.\"s.n + Int64(1)\" FROM S3Object s",
&["s.n + Int64(1)", "s.obj[s.n + Int64(1)]"],
),
("SELECT s.obj0.id FROM S3Object s ORDER BY s.id", &["s.obj0[id]"]),
("SELECT s.obj0.id FROM S3Object s ORDER BY s.obj0.id", &["id"]),
("SELECT (s.obj0.id) FROM S3Object s", &["id"]),
("SELECT s.obj0.id, (s.id) FROM S3Object s", &["s.obj0[id]", "id"]),
("SELECT s.a.id, s.b.id FROM S3Object s", &["s.a[id]", "s.b[id]"]),
("SELECT s.a.id, s.b.id, s.obj2.obj0 FROM S3Object s", &["s.a[id]", "s.b[id]", "obj0"]),
("SELECT s.obj2.\"s.obj0[id]\", s.obj0.id FROM S3Object s", &["s.obj0[id]", "id"]),
(
"SELECT s.obj2.\"s.obj0[id]\", s.obj1.id, s.obj0.id FROM S3Object s",
&["s.obj2[s.obj0[id]]", "s.obj1[id]", "s.obj0[id]"],
),
(
"SELECT s.obj2.\"s.obj0[id]\", s.obj0.id, LOWER(s.id) FROM S3Object s",
&["s.obj0[id]", "id", "lower(s.id)"],
),
];
for (sql, expected) in cases {
let (names, _) = production_json_lines_query(BUCKET, KEY, sql)
.await
.unwrap_or_else(|error| panic!("{sql}: {error}"));
assert_eq!(names, *expected, "{sql}");
}
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn production_path_projection_naming_keeps_queries_planning() {
const BUCKET: &str = "s3select-path-projection-planning";
const KEY: &str = "planning.jsonl";
let env = snapshot_test_env().await;
env.make_bucket(BUCKET, false).await;
env.put_object_bytes(BUCKET, KEY, PATH_NAMES_DATA.to_vec()).await;
env.put_object_bytes(BUCKET, "no-top-level-id.jsonl", br#"{"obj0":{"id":"a"},"obj1":{"id":"b"}}"#.to_vec())
.await;
let mut failures = Vec::new();
// GROUP BY resolves a SELECT alias when no input column has its name.
for sql in [
"SELECT s.obj0.id FROM S3Object s GROUP BY id",
"SELECT s.obj0.id, s.obj1.id AS k FROM S3Object s GROUP BY id, k",
] {
if let Err(error) = production_json_lines_query(BUCKET, "no-top-level-id.jsonl", sql).await {
failures.push(format!("{sql}: {error}"));
}
}
// Each query also plans when every path projection keeps DataFusion's name.
for sql in [
"SELECT s.obj0.id FROM S3Object s",
"SELECT obj0.id FROM S3Object",
"SELECT s.obj0.id FROM S3Object s WHERE s.obj0.id = 'a'",
"SELECT s.obj0.id FROM S3Object s WHERE s.obj0.id IN ('a', 'b') LIMIT 1",
"SELECT s.obj0.id FROM S3Object s LIMIT 0",
"SELECT s.obj0.id FROM S3Object s ORDER BY s.obj0.id DESC LIMIT 1",
"SELECT s.obj0.id FROM S3Object s ORDER BY id",
"SELECT s.obj0.id FROM S3Object s ORDER BY s.id",
"SELECT s.obj0.id FROM S3Object s ORDER BY upper(s.id)",
"SELECT s.obj0.id AS k FROM S3Object s ORDER BY k",
"SELECT s.obj0.id, s.obj2.obj0 FROM S3Object s ORDER BY s.obj0.id",
"SELECT s.obj2.obj0 FROM S3Object s ORDER BY (s.obj0.id)",
"SELECT s.obj0.id, s.id FROM S3Object s ORDER BY 2",
"SELECT s.obj0.id FROM S3Object s GROUP BY s.obj0.id",
"SELECT s.obj0.id, COUNT(*) FROM S3Object s GROUP BY s.obj0.id",
"SELECT s.obj0.id, COUNT(*) AS id FROM S3Object s GROUP BY s.obj0.id",
"SELECT COUNT(*), s.obj0.id FROM S3Object s GROUP BY s.obj0.id ORDER BY COUNT(*)",
"SELECT COUNT(s.obj0.id) FROM S3Object s",
"SELECT MAX(s.obj0.id) FROM S3Object s",
"SELECT UPPER(s.obj0.id), s.obj0.id FROM S3Object s",
"SELECT s.obj0.id, s.obj0.id = 'a' FROM S3Object s",
"SELECT s.obj0.id, s.obj0.id AS x FROM S3Object s",
"SELECT s.obj0.id, s.id FROM S3Object s",
"SELECT s.obj0.id, (s.id) FROM S3Object s",
"SELECT (s.obj0.id), s.obj2.\"s.obj0[id]\" FROM S3Object s",
"SELECT s.obj0.id AS id, s.a.id FROM S3Object s",
"SELECT s.obj0.id, s.x AS id FROM S3Object s",
"SELECT s.obj0.id AS \"s.a[id]\", s.b.id FROM S3Object s",
"SELECT s.obj.ID, s.id FROM S3Object s",
"SELECT s.obj.\"ID\", LOWER(s.id) FROM S3Object s",
"SELECT s.obj.id, s.obj0.id FROM S3Object s",
"SELECT s.a.id, s.b.id FROM S3Object s",
"SELECT s.a.id, s.b.id, s.obj2.obj0 FROM S3Object s",
"SELECT *, s.obj0.id FROM S3Object s",
"SELECT s.*, s.obj0.id FROM S3Object s",
"SELECT s.obj.\"lower(s.id)\", LOWER(s.id) FROM S3Object s",
"SELECT LOWER(s.id), s.obj.\"lower(s.id)\" FROM S3Object s",
"SELECT s.obj.\"count(*)\", COUNT(*) FROM S3Object s GROUP BY s.obj.\"count(*)\"",
"SELECT s.obj.\"s.n\", CAST(s.n AS INT) FROM S3Object s",
"SELECT s.obj.\"s.n + Int64(1)\", s.n + 1 FROM S3Object s",
"SELECT s.obj0.id, CAST(s.n AS INT) FROM S3Object s",
"SELECT s.obj0.id, s.obj1.id, s.obj2.\"s.obj0[id]\" FROM S3Object s",
"SELECT s.obj2.\"s.obj0[id]\", s.obj1.id, s.obj0.id FROM S3Object s",
"SELECT s.obj2.\"s.obj0[id]\", s.obj0.id FROM S3Object s",
"SELECT s.obj2.\"s.obj0[id]\", s.obj0.id, LOWER(s.id) FROM S3Object s",
"SELECT (s.obj0.id), (s.obj1.id) FROM S3Object s",
"SELECT s.obj.\"ID\", s.obj0.id FROM S3Object s",
] {
match production_json_lines_query(BUCKET, KEY, sql).await {
Ok((names, _)) if names.iter().collect::<std::collections::HashSet<_>>().len() != names.len() => {
failures.push(format!("{sql}: duplicate names {names:?}"));
}
Ok(_) => {}
Err(error) => failures.push(format!("{sql}: {error}")),
}
}
assert!(failures.is_empty(), "{failures:#?}");
}
#[tokio::test]
async fn csv_query_uses_custom_record_delimiter_across_file_partitions() {
const ROW_COUNT: usize = 200_000;
@@ -2841,4 +3183,110 @@ mod tests {
assert_eq!(admission.available_permits(), 1);
assert!(output.next().await.is_none());
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn path_alias_does_not_collide_with_computed_output_name() {
let mut input = (*json_snapshot_input()).clone();
input.bucket = "s3select-review-name-collision".to_string();
input.key = "collision.jsonl".to_string();
input.request.expression = "SELECT s.obj.\"lower(s.id)\", LOWER(s.id) FROM S3Object s".to_string();
let input = Arc::new(input);
let env = snapshot_test_env().await;
env.make_bucket(&input.bucket, false).await;
env.put_object_bytes(&input.bucket, &input.key, b"{\"id\":\"A\",\"obj\":{\"lower(s.id)\":7}}\n".to_vec())
.await;
let snapshot = env.prepare_select_object_snapshot(&input.bucket, &input.key).await;
let dispatcher = production_dispatcher(Arc::clone(&input));
let query = Query::new_with_snapshot(
QueryContext {
input: Arc::clone(&input),
},
input.request.expression.clone(),
snapshot,
);
let machine = dispatcher.build_query_state_machine(query).await.expect("production session");
let result = dispatcher.build_logical_plan(Arc::clone(&machine)).await;
if let Err(error) = &result {
panic!(
"real production planner rejected a previously valid query: {error}; S3 classification={:?}",
error.s3_select_policy_error()
);
}
let plan = result.expect("valid production query").expect("query plan");
let Plan::Query(query_plan) = &plan;
let schema = query_plan.df_plan.schema();
assert_eq!(schema.fields().len(), 2);
assert_ne!(schema.field(0).name(), schema.field(1).name());
let output = dispatcher
.execute_logical_plan(plan, machine)
.await
.expect("execute production query");
let Output::StreamData(stream) = output else {
panic!("query must return rows");
};
let batches: Vec<_> = stream.try_collect().await.expect("read production result");
assert_eq!(batches.iter().map(|batch| batch.num_rows()).sum::<usize>(), 1);
let batch = batches.iter().find(|batch| batch.num_rows() == 1).expect("one result row");
let key = batch
.column(0)
.as_any()
.downcast_ref::<datafusion::arrow::array::Int64Array>()
.expect("integer key");
let lower = batch
.column(1)
.as_any()
.downcast_ref::<datafusion::arrow::array::StringArray>()
.expect("lowercase string");
assert_eq!(key.value(0), 7);
assert_eq!(lower.value(0), "a");
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn path_alias_does_not_collide_with_computed_sort_name() {
let mut input = (*json_snapshot_input()).clone();
input.bucket = "s3select-sort-name-collision".to_string();
input.key = "collision.jsonl".to_string();
input.request.expression = "SELECT s.obj.\"lower(s.id)\" FROM S3Object s ORDER BY LOWER(s.id)".to_string();
let input = Arc::new(input);
let env = snapshot_test_env().await;
env.make_bucket(&input.bucket, false).await;
env.put_object_bytes(&input.bucket, &input.key, b"{\"id\":\"A\",\"obj\":{\"lower(s.id)\":7}}\n".to_vec())
.await;
let snapshot = env.prepare_select_object_snapshot(&input.bucket, &input.key).await;
let dispatcher = production_dispatcher(Arc::clone(&input));
let query = Query::new_with_snapshot(
QueryContext {
input: Arc::clone(&input),
},
input.request.expression.clone(),
snapshot,
);
let machine = dispatcher.build_query_state_machine(query).await.expect("production session");
let result = dispatcher.build_logical_plan(Arc::clone(&machine)).await;
if let Err(error) = &result {
panic!(
"real production planner rejected a previously valid query: {error}; S3 classification={:?}",
error.s3_select_policy_error()
);
}
let plan = result.expect("valid production query").expect("query plan");
let Plan::Query(query_plan) = &plan;
let schema = query_plan.df_plan.schema();
assert_eq!(schema.fields().len(), 1);
let output = dispatcher
.execute_logical_plan(plan, machine)
.await
.expect("execute production query");
let Output::StreamData(stream) = output else {
panic!("query must return rows");
};
let batches: Vec<_> = stream.try_collect().await.expect("read production result");
assert_eq!(batches.iter().map(|batch| batch.num_rows()).sum::<usize>(), 1);
let batch = batches.iter().find(|batch| batch.num_rows() == 1).expect("one result row");
let key = batch
.column(0)
.as_any()
.downcast_ref::<datafusion::arrow::array::Int64Array>()
.expect("integer key");
assert_eq!(key.value(0), 7);
}
}
@@ -14,5 +14,6 @@
mod json_document_schema;
mod json_document_table;
mod json_lines_schema;
pub mod manager;
mod parquet_table;
+288 -7
View File
@@ -12,14 +12,19 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use std::{convert::Infallible, ops::ControlFlow};
use std::{collections::HashMap, convert::Infallible, ops::ControlFlow};
use async_trait::async_trait;
use datafusion::sql::{
planner::{IdentNormalizer, SqlToRel},
sqlparser::ast::{
AccessExpr, Expr, GroupByExpr, Ident, JsonPath, JsonPathElem, ObjectNamePart, OrderByKind, Query, Select, SelectFlavor,
SetExpr, Statement, Subscript, TableAlias, TableFactor, Value, Visit, VisitMut, Visitor, VisitorMut,
use datafusion::{
common::{Column, Result as DFResult},
logical_expr::{Expr as LogicalExpr, LogicalPlan, LogicalPlanBuilder},
sql::{
planner::{IdentNormalizer, SqlToRel},
sqlparser::ast::{
AccessExpr, Expr, GroupByExpr, Ident, JsonPath, JsonPathElem, ObjectNamePart, OrderBy, OrderByKind, Query, Select,
SelectFlavor, SelectItem, SetExpr, Statement, Subscript, TableAlias, TableFactor, Value, Visit, VisitMut, Visitor,
VisitorMut,
},
},
};
use rustfs_s3select_api::{
@@ -75,7 +80,9 @@ impl<'a, S: ContextProviderExtension + Send + Sync + 'a> SqlPlanner<'a, S> {
}
async fn df_prepared_sql_to_plan(&self, stmt: Statement, _session: &SessionCtx) -> QueryResult<Plan> {
let keys = path_projection_keys(&stmt);
let df_plan = self.df_planner.sql_statement_to_plan(stmt).map_err(classify_planner_error)?;
let df_plan = name_path_projections(df_plan, &keys)?;
Ok(Plan::Query(QueryPlan {
df_plan,
is_tag_scan: false,
@@ -150,9 +157,203 @@ pub(crate) fn prepare_s3_select_statement(statement: &mut Statement) -> QueryRes
};
let mut normalizer = PartiQlSubscriptNormalizer;
let _ = VisitMut::visit(query, &mut normalizer);
if let SetExpr::Select(select) = query.body.as_mut() {
alias_path_projections(select, query.order_by.as_ref());
}
Ok(source)
}
/// Names an unaliased path projection after its last key, as AWS does
/// (`s.obj0.id` -> `id`), instead of DataFusion's `s.obj0[id]`. Names that
/// could collide with another output name keep DataFusion's name. Wildcard
/// columns, computed expressions, and retained paths have unknown output names
/// here, so they disable aliasing. Unprojected ORDER BY expressions also disable
/// aliasing because DataFusion adds their input columns to the projection.
/// GROUP BY can refer to these aliases; [`name_path_projections`] names the
/// paths left here once the planned output names are known.
fn alias_path_projections(select: &mut Select, order_by: Option<&OrderBy>) {
if has_wildcard(select) || sorts_by_unprojected_expr(select, order_by) {
return;
}
let Some(qualifier) = source_qualifier(select) else {
return;
};
let normalizer = IdentNormalizer::default();
let names: Vec<Option<String>> = select
.projection
.iter()
.map(|item| {
let ident = match item {
SelectItem::UnnamedExpr(Expr::Nested(_)) => None,
SelectItem::UnnamedExpr(expr) => column_name(expr),
SelectItem::ExprWithAlias { alias, .. } => Some(alias),
_ => None,
};
ident.map(|ident| normalizer.normalize(ident.clone()))
})
.collect();
if names.iter().any(Option::is_none) {
return;
}
let mut counts = HashMap::<String, usize>::new();
for name in names.iter().flatten() {
*counts.entry(name.clone()).or_default() += 1;
}
// Retained paths have generated names that cannot be inferred from their last key.
// Keep the whole projection unchanged so a sibling alias cannot collide with them.
if names.iter().flatten().any(|name| counts.get(name) != Some(&1)) {
return;
}
for item in &mut select.projection {
let SelectItem::UnnamedExpr(expr) = item else {
continue;
};
let Some(alias) = path_key(expr, &qualifier).cloned() else {
continue;
};
let expr = std::mem::replace(expr, Expr::Identifier(Ident::new("")));
*item = SelectItem::ExprWithAlias { expr, alias };
}
}
/// The output name DataFusion gives a column reference: its last part.
fn column_name(expr: &Expr) -> Option<&Ident> {
match expr {
Expr::Identifier(ident) => Some(ident),
Expr::CompoundIdentifier(parts) => parts.last(),
Expr::Nested(inner) => column_name(inner),
_ => None,
}
}
fn has_wildcard(select: &Select) -> bool {
select
.projection
.iter()
.any(|item| matches!(item, SelectItem::Wildcard(_) | SelectItem::QualifiedWildcard(..)))
}
fn sorts_by_unprojected_expr(select: &Select, order_by: Option<&OrderBy>) -> bool {
let Some(OrderBy {
kind: OrderByKind::Expressions(sort_exprs),
..
}) = order_by
else {
return false;
};
sort_exprs.iter().any(|sort_expr| {
!select
.projection
.iter()
.any(|item| matches!(item, SelectItem::UnnamedExpr(projected) if *projected == sort_expr.expr))
})
}
/// The normalized table alias, or table name, that may prefix a path.
fn source_qualifier(select: &Select) -> Option<String> {
let normalizer = IdentNormalizer::default();
match &select.from.first()?.relation {
TableFactor::Table { alias: Some(alias), .. } => Some(normalizer.normalize(alias.name.clone())),
TableFactor::Table { name, .. } => match name.0.first() {
Some(ObjectNamePart::Identifier(table_name)) => Some(normalizer.normalize(table_name.clone())),
_ => None,
},
_ => None,
}
}
/// The last key of a path of at least two keys after the source qualifier
/// (`s.obj0.id` -> `id`), the name AWS gives the projection.
fn path_key<'a>(expr: &'a Expr, qualifier: &str) -> Option<&'a Ident> {
let parts = match expr {
Expr::CompoundIdentifier(parts) => parts,
Expr::Nested(inner) => return path_key(inner, qualifier),
_ => return None,
};
let path = match parts.split_first() {
Some((first, rest)) if IdentNormalizer::default().normalize(first.clone()) == qualifier => rest,
_ => parts.as_slice(),
};
match path {
[_, _, ..] => path.last(),
_ => None,
}
}
/// The normalized key of each path projection left unaliased by
/// [`alias_path_projections`]. Empty when a wildcard hides the output
/// positions, or when an ORDER BY expression is not projected.
fn path_projection_keys(statement: &Statement) -> Vec<Option<String>> {
let Statement::Query(query) = statement else {
return Vec::new();
};
let SetExpr::Select(select) = query.body.as_ref() else {
return Vec::new();
};
if has_wildcard(select) || sorts_by_unprojected_expr(select, query.order_by.as_ref()) {
return Vec::new();
}
let Some(qualifier) = source_qualifier(select) else {
return Vec::new();
};
let normalizer = IdentNormalizer::default();
select
.projection
.iter()
.map(|item| match item {
SelectItem::UnnamedExpr(expr) => path_key(expr, &qualifier).map(|key| normalizer.normalize(key.clone())),
_ => None,
})
.collect()
}
/// Renames each path projection to its key from [`path_projection_keys`]
/// unless another output column keeps or receives that name. Names come from
/// the planned schema, so computed, aliased and retained columns all count.
fn name_path_projections(plan: LogicalPlan, keys: &[Option<String>]) -> DFResult<LogicalPlan> {
let schema = plan.schema();
if keys.len() != schema.fields().len() {
return Ok(plan);
}
let mut renames: Vec<Option<&str>> = keys
.iter()
.zip(schema.fields())
.map(|(key, field)| key.as_deref().filter(|key| key != field.name()))
.collect();
// Dropping a rename brings back the generated name, which can collide with another rename.
loop {
let mut counts = HashMap::<&str, usize>::new();
for (rename, field) in renames.iter().zip(schema.fields()) {
*counts.entry(rename.unwrap_or(field.name())).or_default() += 1;
}
let mut dropped = false;
for rename in &mut renames {
if rename.is_some_and(|name| counts[name] > 1) {
*rename = None;
dropped = true;
}
}
if !dropped {
break;
}
}
if renames.iter().all(Option::is_none) {
return Ok(plan);
}
let columns: Vec<_> = schema
.iter()
.zip(renames)
.map(|(column, rename)| {
let column = LogicalExpr::Column(Column::from(column));
match rename {
Some(name) => column.alias(name),
None => column,
}
})
.collect();
LogicalPlanBuilder::from(plan).project(columns)?.build()
}
struct PartiQlSubscriptNormalizer;
impl VisitorMut for PartiQlSubscriptNormalizer {
@@ -374,7 +575,7 @@ impl Visitor for SubqueryDetector {
#[cfg(test)]
mod tests {
use super::prepare_s3_select_statement;
use super::{path_projection_keys, prepare_s3_select_statement};
use crate::sql::parser::ExtParser;
use datafusion::sql::sqlparser::ast::{AccessExpr, Expr, Statement, Visit, Visitor};
use rustfs_s3select_api::{
@@ -596,6 +797,86 @@ mod tests {
}
}
fn prepared(sql: &str) -> String {
let mut statement = parse_statement(sql);
prepare_s3_select_statement(&mut statement).expect("statement should be supported");
statement.to_string()
}
#[test]
fn unaliased_path_projections_are_named_after_their_last_key() {
for (sql, expected) in [
("SELECT s.obj0.id FROM S3Object AS s", "SELECT s.obj0.id AS id FROM S3Object AS s"),
(
"SELECT s.a.b.c, s.x.\"Y\" FROM S3Object AS s",
"SELECT s.a.b.c AS c, s.x.\"Y\" AS \"Y\" FROM S3Object AS s",
),
("SELECT obj0.id FROM S3Object", "SELECT obj0.id AS id FROM S3Object"),
(
"SELECT s.obj0.id FROM S3Object AS s WHERE s.obj0.id = 'a' ORDER BY s.obj0.id",
"SELECT s.obj0.id AS id FROM S3Object AS s WHERE s.obj0.id = 'a' ORDER BY s.obj0.id",
),
] {
assert_eq!(prepared(sql), expected, "{sql}");
}
}
#[test]
fn projections_keep_their_names_when_a_key_alias_is_absent_or_ambiguous() {
for sql in [
"SELECT s.id FROM S3Object AS s",
"SELECT S3Object.id FROM S3Object",
"SELECT s.tags[1] FROM S3Object AS s",
"SELECT s.obj0.id AS x FROM S3Object AS s",
"SELECT s.obj0.id, s.obj1.id FROM S3Object AS s",
"SELECT s.obj0.ID, s.obj1.id FROM S3Object AS s",
"SELECT s.obj0.id, s.id FROM S3Object AS s",
"SELECT s.obj0.id, s.name AS id FROM S3Object AS s",
"SELECT s.obj0.\"Id\", s.obj1.\"Id\" FROM S3Object AS s",
"SELECT *, s.obj0.id FROM S3Object AS s",
"SELECT s.*, s.obj0.id FROM S3Object AS s",
"SELECT s.obj0.id FROM S3Object AS s ORDER BY s.id",
"SELECT s.obj0.id FROM S3Object AS s ORDER BY upper(id)",
"SELECT s.obj0.id, (s.id) FROM S3Object AS s",
"SELECT s.obj.\"lower(s.id)\", LOWER(s.id) FROM S3Object AS s",
"SELECT s.obj.\"count(*)\", COUNT(*) FROM S3Object AS s GROUP BY s.obj.\"count(*)\"",
"SELECT s.obj0.id, COUNT(*) FROM S3Object AS s GROUP BY s.obj0.id",
] {
assert_eq!(prepared(sql), sql);
}
}
#[test]
fn path_projection_keys_cover_paths_left_unaliased() {
let keys = |sql: &str| {
let mut statement = parse_statement(sql);
prepare_s3_select_statement(&mut statement).expect("statement should be supported");
path_projection_keys(&statement)
};
let key = |key: &str| Some(key.to_owned());
for (sql, expected) in [
("SELECT s.obj0.id FROM S3Object AS s", vec![None]),
("SELECT s.obj0.id, s.obj1.id FROM S3Object AS s", vec![key("id"), key("id")]),
(
"SELECT s.a.b.c, s.x.\"Y\", s.obj0.ID, lower(s.id) FROM S3Object AS s",
vec![key("c"), key("Y"), key("id"), None],
),
("SELECT obj0.id, COUNT(*) FROM S3Object GROUP BY obj0.id", vec![key("id"), None]),
("SELECT (s.obj0.id), ((s.obj1.id)) FROM S3Object AS s", vec![key("id"), key("id")]),
(
"SELECT s.id, s.tags[1], s.obj0.id AS x, (s.id) FROM S3Object AS s",
vec![None, None, None, None],
),
("SELECT S3Object.id, lower(id) FROM S3Object", vec![None, None]),
("SELECT *, s.obj0.id FROM S3Object AS s", vec![]),
("SELECT s.*, s.obj0.id FROM S3Object AS s", vec![]),
("SELECT s.obj0.id FROM S3Object AS s ORDER BY s.id", vec![]),
("SELECT s.obj0.id FROM S3Object AS s ORDER BY upper(id)", vec![]),
] {
assert_eq!(keys(sql), expected, "{sql}");
}
}
#[test]
fn accepts_group_by_and_order_by() {
let statement = parse_statement("SELECT department, COUNT(*) FROM S3Object GROUP BY department ORDER BY department");
+39
View File
@@ -0,0 +1,39 @@
# Audit Entries for Authorization Denials
**Use this when:** you consume the audit stream and need to know which authorization denials appear in it, how many, and how to tune the limit.
**Source of truth:** `rustfs/src/storage/denial_audit.rs` (`DenialAuditor`, `DenialRateLimiter`), `rustfs/src/storage/access.rs` (`DenialContext`), `crates/config/src/constants/env.rs` (`ENV_AUDIT_DENIAL_RATE`, `ENV_AUDIT_DENIAL_ANONYMOUS_RATE`).
With audit enabled (`RUSTFS_AUDIT_ENABLE=true`), a request refused by the authorization layer produces an audit entry in addition to the `s3_authorization_denied` log event. The S3 response is unchanged: a bare `Access Denied`.
## Entry shape
| Field | Value |
| --- | --- |
| `event` | `audit:S3:Operation` |
| `api.name` | the refused action, for example `s3:PutObject` |
| `api.bucket`, `api.object` | the request's bucket and object (the object is omitted for a bulk entry) |
| `api.status`, `api.statusCode` | `failure`, `403` |
| `accessKey` | the caller's access key, absent for an anonymous caller |
| `error` | `Access Denied` |
| `tags.denialReason` | the internal reason, for example `iam_explicit_deny`, `bucket_policy_explicit_deny`, `no_policy_allows_action` |
| `tags.deniedKeys` | bulk entries only: how many keys of a DeleteObjects request were refused for the action in `api.name` |
| `tags.suppressedDenials` | present when denials were dropped by the rate limit since the previous entry |
## What is not audited
- Probe denials that callers treat as an expected filter outcome (the per-bucket ListBuckets fallback and per-object metadata permission collection).
- A DeleteObjects request audits its refused keys as one entry per denied action (for example `s3:DeleteObject`, `s3:GetObject`, `s3:BypassGovernanceRetention`) with `tags.deniedKeys`, not one entry per key.
- Denials that are later overridden, such as an implicit IAM deny followed by a bucket policy allow.
## Rate limit
A caller can cause denials at will, so the number of entries is bounded by a process-wide token bucket per caller class. The burst equals the per-second rate.
| Variable | Default | Meaning |
| --- | --- | --- |
| `RUSTFS_AUDIT_DENIAL_RATE` | `50` | Entries per second for authenticated callers. `0` turns denial entries off for them. |
| `RUSTFS_AUDIT_DENIAL_ANONYMOUS_RATE` | `5` | Entries per second for anonymous callers. `0` turns denial entries off for them. |
Denials dropped by the limit are counted per caller class, and the count is reported in `tags.suppressedDenials` on the next entry of the same class that is let through. The limit is per process, not per node group.
Trade-off: all authenticated callers share one bucket. A valid, low-privilege account that causes more denials per second than `RUSTFS_AUDIT_DENIAL_RATE` can use up the budget and crowd out other principals' denial entries during that time. Its own drops are still counted in `tags.suppressedDenials`, so the loss is visible, but the other principals' denials in that window are not recorded. A per-principal sub-limit would remove this and is a follow-up, not part of this change.
@@ -289,6 +289,16 @@ receiving site. If a group ended up disabled on one site only, re-enable it
there explicitly with `mc admin group enable`; a snapshot or repair will not
do it.
The managed `site-replicator-0` service account carries an embedded policy
that the server owns. It names `s3:ForceDeleteObject` explicitly because a
replicated force-delete (Console folder delete, `x-rustfs-force-delete: true`)
is authorized on the peer like any other force-delete and `s3:*` does not
grant that action. The reconcile tick refreshes an account whose stored policy
predates the grant; do not edit the policy by hand, the next repair rewrites
it. A bucket-replication target configured with its own credentials needs the
same `s3:ForceDeleteObject` grant on the target side, or its force-deletes are
rejected and counted as replication failures.
Treat IAM divergence as a security incident: a user deleted on one site can
remain usable on an unreachable peer until replay or repair completes. A peer
whose IAM entry is escalated does not receive scheduled snapshots either -
@@ -210,6 +210,7 @@ impl From<ObjectZipDownloadReqInfoSnapshot> for ReqInfo {
region: current_region(),
request_context: None,
suppress_denial_log: false,
suppress_denial_audit: false,
}
}
}
@@ -1430,6 +1431,7 @@ mod tests {
region: current_region(),
request_context: None,
suppress_denial_log: false,
suppress_denial_audit: false,
}
}
+79 -1
View File
@@ -169,6 +169,7 @@ fn site_replicator_service_account_policy() -> S3Result<Policy> {
"s3:PutObject",
"s3:DeleteObject",
"s3:DeleteObjectVersion",
"s3:ForceDeleteObject",
"s3:ReplicateObject",
"s3:ReplicateDelete",
"s3:ReplicateTags",
@@ -191,6 +192,29 @@ fn site_replicator_service_account_policy() -> S3Result<Policy> {
.map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("parse site replicator policy failed: {e}")))
}
/// A replicated force-delete needs an explicit `s3:ForceDeleteObject` on the peer; `s3:*` does not grant it.
/// Accounts created before the action was added to the policy lack it and must be refreshed in place.
async fn site_replicator_policy_allows_force_delete(policy: Option<&Policy>) -> bool {
let Some(policy) = policy else {
return false;
};
let groups: Option<Vec<String>> = None;
let claims = HashMap::new();
let conditions = HashMap::new();
let args = rustfs_policy::policy::Args {
account: SITE_REPLICATOR_SERVICE_ACCOUNT,
groups: &groups,
action: Action::S3Action(rustfs_policy::policy::action::S3Action::ForceDeleteObjectAction),
conditions: &conditions,
is_owner: false,
claims: &claims,
deny_only: false,
bucket: "bucket",
object: "prefix/",
};
policy.is_allowed(&args).await
}
// Lock order: lifecycle -> bucket-mutation admission -> per-bucket mutation
// -> bucket operation -> repair admission -> state -> per-bucket metadata.
// "state" is the distributed state-object lock in
@@ -1654,10 +1678,39 @@ async fn reconcile_site_replicator_service_account() -> S3Result<()> {
Ok(_) => {
// Never rebuild on an unread parent: `unwrap_or_default` here would compare an
// empty string against a real parent and repair a healthy account on every boot.
let Ok((credentials, _)) = iam_sys.get_service_account(access_key).await else {
let Ok((credentials, stored_policy)) = iam_sys.get_service_account(access_key).await else {
return Ok(());
};
if credentials.parent_user == parent_user {
if site_replicator_policy_allows_force_delete(stored_policy.as_ref()).await {
return Ok(());
}
iam_sys
.update_service_account(
access_key,
UpdateServiceAccountOpts {
session_policy: Some(session_policy),
secret_key: None,
name: None,
description: None,
expiration: None,
status: None,
parent_user: None,
allow_site_replicator_account: true,
},
)
.await
.map_err(ApiError::from)?;
warn!(
event = EVENT_ADMIN_SITE_REPLICATION_STATE,
component = LOG_COMPONENT_ADMIN,
subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
result = "service_account_repaired",
reason = "outdated_policy",
parent = %parent_user,
"admin site replication state"
);
return Ok(());
}
@@ -10636,6 +10689,31 @@ mod tests {
assert!(!policy.is_allowed(&put_policy_args).await);
}
// A replicated force-delete (Console folder delete, `x-rustfs-force-delete: true`) is rejected by the
// peer with AccessDenied unless the replication account names `s3:ForceDeleteObject` explicitly.
#[tokio::test]
async fn test_site_replicator_policy_allows_force_delete_replication() {
let policy = site_replicator_service_account_policy().expect("site replicator policy should parse");
assert!(site_replicator_policy_allows_force_delete(Some(&policy)).await);
}
// Accounts created before the action was added keep their stored policy until the reconciler refreshes it.
#[tokio::test]
async fn test_outdated_site_replicator_policy_is_not_treated_as_current() {
assert!(!site_replicator_policy_allows_force_delete(None).await);
let outdated: Policy = serde_json::from_value(serde_json::json!({
"Version": "2012-10-17",
"Statement": [{
"Effect": "Allow",
"Action": ["s3:DeleteObject", "s3:DeleteObjectVersion", "s3:ReplicateDelete", "s3:*"],
"Resource": ["arn:aws:s3:::*/*"]
}]
}))
.expect("outdated policy should parse");
assert!(!site_replicator_policy_allows_force_delete(Some(&outdated)).await);
}
// The replication service account must be able to carry object-lock metadata to the peer.
// Without these actions the peer answers AccessDenied for any replicated object that has
// retention or a legal hold, so a WORM-protected object never reaches the replica at all,
@@ -2895,8 +2895,8 @@ async fn put_object_extract_computes_replication_decision_per_entry() {
/// replication themselves. Without it a retention or legal hold applied after upload stays
/// local and the replica keeps its previous, unprotected state (a WORM object that is still
/// deletable on the peer). This observes the shared decision counter: each handler must
/// compute a replication decision exactly once, which is the step that drives both the
/// persisted pending marker and the schedule.
/// compute a replication decision exactly once when replication is configured, which is
/// the step that drives both the persisted pending marker and the schedule.
#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
#[serial]
#[ignore = "global-state usecase integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
@@ -2945,7 +2945,27 @@ async fn object_lock_handlers_schedule_replication() {
retention change never replicates and the peer copy stays unprotected"
);
// PutObjectLegalHold must do the same.
// A bucket without a replication configuration can skip the decision.
MUST_REPLICATE_OBJECT_CALLS.store(0, Ordering::SeqCst);
let no_replication_input = PutObjectLegalHoldInput::builder()
.bucket(bucket.clone())
.key(object.to_string())
.legal_hold(Some(ObjectLockLegalHold {
status: Some(ObjectLockLegalHoldStatus::from_static(ObjectLockLegalHoldStatus::OFF)),
}))
.build()
.unwrap();
fs.put_object_legal_hold(build_request(no_replication_input, Method::PUT))
.await
.expect("legal hold without replication should succeed");
assert_eq!(
MUST_REPLICATE_OBJECT_CALLS.load(Ordering::SeqCst),
0,
"a bucket without replication must skip the decision"
);
// A configured bucket must still evaluate the metadata replication decision.
install_delete_replication_config(&bucket, true, false, None).await;
MUST_REPLICATE_OBJECT_CALLS.store(0, Ordering::SeqCst);
let legal_hold_input = PutObjectLegalHoldInput::builder()
.bucket(bucket.clone())
+54 -36
View File
@@ -385,12 +385,6 @@ impl DefaultObjectUsecase {
dst_opts.object_lock_config_snapshot =
Some(store.object_lock_config_snapshot(&bucket).await.map_err(ApiError::from)?);
}
let mut current_opts: ObjectOptions = internal_object_info_lookup_opts(
get_opts(&bucket, &key, dest_version_id.clone(), None, &req.headers)
.await
.map_err(ApiError::from)?,
);
// Hold the self-copy namespace write lock before opening the source reader, including
// expected-current historical copies. With lock optimization disabled, the source
// stream retains its read guard until EOF; taking the write lock later in ECStore
@@ -410,28 +404,42 @@ impl DefaultObjectUsecase {
}
dst_opts.expected_current_version_id = expected_current_version_id.clone();
if _self_copy_lock_guard.is_some() {
current_opts.no_lock = true;
}
let previous_current_sizes = match store.get_object_info(&bucket, &key, &current_opts).await {
Ok(existing_obj_info) => {
validate_existing_object_lock_for_write(&object_lock_config_state, &existing_obj_info, &dst_opts)?;
if let Some(expected) = expected_current_version_id.as_deref()
&& existing_obj_info.version_id.unwrap_or_default().to_string() != expected
{
return Err(s3_error!(PreconditionFailed));
}
Some((existing_obj_info.size.max(0) as u64, quota_object_size(&existing_obj_info)))
let copy_prelookup_required = self_copy_lock
|| expected_current_version_id.is_some()
|| dest_version_id.is_some()
|| (!super::shared::put_like_write_creates_new_version(&dst_opts)
&& object_lock_checks_required_for_state(&object_lock_config_state));
let previous_current_sizes: Option<Option<(u64, Result<u64, StorageError>)>> = if copy_prelookup_required {
let mut current_opts: ObjectOptions = internal_object_info_lookup_opts(
get_opts(&bucket, &key, dest_version_id.clone(), None, &req.headers)
.await
.map_err(ApiError::from)?,
);
if _self_copy_lock_guard.is_some() {
current_opts.no_lock = true;
}
Err(err) => {
if expected_current_version_id.is_some() {
return Err(s3_error!(PreconditionFailed));
Some(match store.get_object_info(&bucket, &key, &current_opts).await {
Ok(existing_obj_info) => {
validate_existing_object_lock_for_write(&object_lock_config_state, &existing_obj_info, &dst_opts)?;
if let Some(expected) = expected_current_version_id.as_deref()
&& existing_obj_info.version_id.unwrap_or_default().to_string() != expected
{
return Err(s3_error!(PreconditionFailed));
}
Some((existing_obj_info.size.max(0) as u64, quota_object_size(&existing_obj_info)))
}
if !is_err_object_not_found(&err) && !is_err_version_not_found(&err) {
return Err(ApiError::from(err).into());
Err(err) => {
if expected_current_version_id.is_some() {
return Err(s3_error!(PreconditionFailed));
}
if !is_err_object_not_found(&err) && !is_err_version_not_found(&err) {
return Err(ApiError::from(err).into());
}
None
}
None
}
})
} else {
None
};
let (mut effective_sse, mut effective_kms_key_id) = resolve_bucket_default_sse(
@@ -753,10 +761,11 @@ impl DefaultObjectUsecase {
if let Some(quota_check) = quota_check.as_ref() {
apply_quota_admission(&mut dst_opts, quota_check)?;
}
let previous_current_size = match previous_current_sizes {
Some((_, Ok(logical_size))) if quota_enabled => Some(logical_size),
Some((_, Err(err))) if quota_enabled => return Err(ApiError::from(err).into()),
Some((physical_size, _)) => Some(physical_size),
let prelookup_previous_current_size: Option<Option<u64>> = match previous_current_sizes {
Some(Some((_, Ok(logical_size)))) if quota_enabled => Some(Some(logical_size)),
Some(Some((_, Err(err)))) if quota_enabled => return Err(ApiError::from(err).into()),
Some(Some((physical_size, _))) => Some(Some(physical_size)),
Some(None) => Some(None),
None => None,
};
if let Some(quota_check) = quota_check.as_ref() {
@@ -785,8 +794,8 @@ impl DefaultObjectUsecase {
let _self_copy_lock_guard = _self_copy_lock_guard;
let _table_data_plane_publication_guards = table_data_plane_publication_guards;
let oi = store
.copy_object(&src_bucket, &src_key, &bucket, &key, &mut src_info, &src_opts, &dst_opts)
let (oi, backfilled_old_current_size) = store
.copy_object_with_old_current_size(&src_bucket, &src_key, &bucket, &key, &mut src_info, &src_opts, &dst_opts)
.await
.map_err(ApiError::from)?;
@@ -800,13 +809,22 @@ impl DefaultObjectUsecase {
maybe_enqueue_transition_immediate(&oi, LcEventSrc::S3CopyObject).await;
let _ = invalidate_object_data_cache_after_copy_success(&cache_adapter, &bucket, &key).await;
let dest_versioned = BucketVersioningSys::prefix_enabled(&bucket, &key).await;
let dest_versioned = dst_opts.versioned;
if has_bucket_metadata {
let committed_size = quota_accounting_object_size(&oi, quota_enabled)?;
if dest_versioned {
record_bucket_object_version_write_memory(&bucket, previous_current_size, committed_size).await;
} else {
record_bucket_object_write_memory(&bucket, previous_current_size, committed_size).await;
match prelookup_previous_current_size
.or_else(|| previous_current_size_from_backfill(backfilled_old_current_size))
{
Some(previous_current_size) => {
if dest_versioned {
record_bucket_object_version_write_memory(&bucket, previous_current_size, committed_size).await;
} else {
record_bucket_object_write_memory(&bucket, previous_current_size, committed_size).await;
}
}
None => {
record_bucket_object_write_unknown_previous_memory(&bucket, committed_size, dest_versioned).await;
}
}
}
+64 -19
View File
@@ -568,6 +568,8 @@ impl DefaultObjectUsecase {
// warn and demote the rest to debug, so a denied 1000-key DeleteObjects
// cannot flood the log.
let mut bulk_denial_logged = false;
// Per-key denials are audited once per denied action, in aggregate, after the loop.
let mut bulk_denials = crate::storage::access::BulkDenialAudit::start(&mut req)?;
for (idx, obj_id) in delete.objects.iter().enumerate() {
if let Err(err) = validate_object_key(&obj_id.key, "DeleteObjects") {
delete_results[idx].error = Some(s3s::dto::Error {
@@ -606,15 +608,18 @@ impl DefaultObjectUsecase {
req_info.version_id = version_id.clone();
}
let mut auth_res = authorize_request(&mut req, delete_object_authorize_action(version_id.as_deref())).await;
let mut denied_action = delete_object_authorize_action(version_id.as_deref());
let mut auth_res = authorize_request(&mut req, denied_action).await;
if auth_res.is_ok()
&& delete_condition
.as_ref()
.is_some_and(DeleteObjectCondition::requires_read_permission)
{
auth_res = authorize_request(&mut req, Action::S3Action(S3Action::GetObjectAction)).await;
denied_action = Action::S3Action(S3Action::GetObjectAction);
auth_res = authorize_request(&mut req, denied_action).await;
}
if auth_res.is_err() {
bulk_denials.denied(denied_action);
if !bulk_denial_logged {
bulk_denial_logged = true;
req_info_mut(&mut req)?.suppress_denial_log = true;
@@ -631,6 +636,7 @@ impl DefaultObjectUsecase {
if bypass_governance {
let auth_res = authorize_request(&mut req, Action::S3Action(S3Action::BypassGovernanceRetentionAction)).await;
if auth_res.is_err() {
bulk_denials.denied(Action::S3Action(S3Action::BypassGovernanceRetentionAction));
if !bulk_denial_logged {
bulk_denial_logged = true;
req_info_mut(&mut req)?.suppress_denial_log = true;
@@ -669,6 +675,8 @@ impl DefaultObjectUsecase {
authorized_deletes.push(AuthorizedDelete { idx, object });
}
bulk_denials.finish(&mut req)?;
if authorized_deletes.is_empty() {
let output = DeleteObjectsOutput {
deleted: (!delete.quiet.unwrap_or_default()).then(Vec::new),
@@ -1147,15 +1155,21 @@ impl DefaultObjectUsecase {
let replicate_force_delete = force_delete && !replica && has_active_delete_rule(&delete_config_snapshot, &key);
let mut force_delete_intent = None;
let existing_object_info = match store.get_object_info_for_delete(&bucket, &key, &opts).await {
Ok(obj_info) => Some(obj_info),
Err(err) => {
// If object not found, allow deletion to proceed (will return 204 No Content)
if !is_err_object_not_found(&err) && !is_err_version_not_found(&err) {
return Err(ApiError::from(err).into());
// A version purge needs the original marker identity for replication.
let need_eager_pre_stat = opts.http_preconditions.is_some() || opts.version_id.is_some();
let existing_object_info = if need_eager_pre_stat {
match Box::pin(store.get_object_info_for_delete(&bucket, &key, &opts)).await {
Ok(obj_info) => Some(obj_info),
Err(err) => {
// If object not found, allow deletion to proceed (will return 204 No Content)
if !is_err_object_not_found(&err) && !is_err_version_not_found(&err) {
return Err(ApiError::from(err).into());
}
None
}
None
}
} else {
None
};
#[cfg(test)]
wait_for_delete_source_test_hook(&bucket).await;
@@ -1209,16 +1223,21 @@ impl DefaultObjectUsecase {
return result;
}
if matches!(&err, StorageError::PrefixAccessDenied(_, _))
&& let Some(existing_object_info) = existing_object_info.as_ref()
&& let Some(reason) = check_object_lock_for_deletion(
&bucket,
existing_object_info,
has_bypass_governance_header(&req.headers),
)
.await
{
return Err(S3Error::with_message(S3ErrorCode::AccessDenied, reason.error_message()));
if matches!(&err, StorageError::PrefixAccessDenied(_, _)) {
let existing_object_info = match existing_object_info {
Some(info) => Some(info),
None => Box::pin(store.get_object_info_for_delete(&bucket, &key, &opts)).await.ok(),
};
if let Some(existing_object_info) = existing_object_info.as_ref()
&& let Some(reason) = check_object_lock_for_deletion(
&bucket,
existing_object_info,
has_bypass_governance_header(&req.headers),
)
.await
{
return Err(S3Error::with_message(S3ErrorCode::AccessDenied, reason.error_message()));
}
}
return Err(ApiError::from(err).into());
@@ -3686,3 +3705,29 @@ mod tests {
));
}
}
/// DeleteObjects refusals are audited once per denied action by `BulkDenialAudit`. The handler needs a
/// live store to run, so this source-level guard fails if the wiring is removed; it does not execute it.
#[cfg(test)]
mod bulk_denial_audit_wiring_tests {
#[test]
fn delete_objects_audits_refused_keys_in_aggregate() {
// Only the code above this module counts; the guard's own text must not satisfy it.
let source = include_str!("delete.rs")
.split("\n#[cfg(test)]\nmod bulk_denial_audit_wiring_tests")
.next()
.unwrap();
let started = source
.find("BulkDenialAudit::start(&mut req)")
.expect("per-key denial audit is not suppressed");
let finished = source
.find("bulk_denials.finish(&mut req)")
.expect("the aggregate entry is not emitted");
assert!(started < finished, "the aggregate must be emitted after the per-key checks");
assert_eq!(
source.matches("bulk_denials.denied(").count(),
2,
"both authorization checks must count their denials"
);
}
}
+74 -3
View File
@@ -1734,7 +1734,9 @@ impl DefaultObjectUsecase {
// replication), the lookup is skipped and accounting is backfilled from
// the dst xl.meta that rename_data already reads, saving a full-disk
// metadata fanout per PUT.
let prelookup_required = version_id.is_some() || object_lock_checks_required_for_state(&object_lock_config_state);
let prelookup_required = version_id.is_some()
|| (!super::shared::put_like_write_creates_new_version(&opts)
&& object_lock_checks_required_for_state(&object_lock_config_state));
#[cfg(test)]
let prelookup_required = prelookup_required || injected_put_storage_failure(PutStorageTestStage::Prelookup).is_some();
// Outer None = prelookup skipped (accounting comes from the commit
@@ -2092,7 +2094,8 @@ impl DefaultObjectUsecase {
let put_admission = put_admission;
let table_publication_guard = origin.take_table_publication_guard();
let table_data_plane_publication_guards = origin.table_data_plane_publication_guards();
async move {
// Keep the large commit future off the request stack when propagating task-local context.
Box::pin(async move {
let _table_publication_guard = table_publication_guard;
let _table_data_plane_publication_guards = table_data_plane_publication_guards;
let _put_admission = put_admission;
@@ -2148,6 +2151,7 @@ impl DefaultObjectUsecase {
result = "error",
error_code = %err.code.as_str(),
storage_error_code = ?diagnostic.storage_code,
lock_failure_reason = diagnostic.lock_reason,
io_error_kind = ?diagnostic.io_kind,
rpc_error_code = ?diagnostic.rpc_code,
pool_metadata_reason = diagnostic.pool_metadata.map(|context| context.reason),
@@ -2252,7 +2256,7 @@ impl DefaultObjectUsecase {
rustfs_io_metrics::record_put_object_stage_duration_from("app_capacity_update", capacity_update_stage_start);
Ok::<_, S3Error>(PutObjectCommitResult { obj_info, put_versioned })
}
})
};
#[cfg(feature = "diagnostic-read-version-caller")]
let put_commit_future = crate::app::storage_api::diagnostic::with_read_version_spawn_context(
@@ -2260,6 +2264,11 @@ impl DefaultObjectUsecase {
crate::app::storage_api::diagnostic::ReadVersionContextBoundary::AppPutOwner,
put_commit_future,
);
#[cfg(not(feature = "diagnostic-read-version-caller"))]
let put_commit_future = rustfs_io_metrics::s3_http_metrics::with_s3_http_operation_context(
rustfs_io_metrics::s3_http_metrics::current_s3_http_operation_context(),
put_commit_future,
);
let put_commit = spawn_traced_join(put_commit_future);
let commit_stage_start = put_stage_metrics_enabled.then(Instant::now);
let put_commit_result = if let Some(cancellation) = eager_put_commit_cancellation {
@@ -2340,6 +2349,68 @@ mod tests {
use futures::StreamExt;
use http::{HeaderMap, HeaderName, HeaderValue, Method};
#[test]
#[serial_test::serial]
fn put_commit_lock_failure_keeps_request_operation_in_spawned_owner() {
use crate::app::storage_api::test::contract::bucket::{BucketOperations as _, MakeBucketOptions};
use metrics_util::debugging::{DebugValue, DebuggingRecorder};
use rustfs_io_metrics::s3_http_metrics::S3HttpRequestGuard;
use rustfs_lock::LockError;
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.expect("test runtime");
let (store, context) = runtime.block_on(Box::pin(real_store_test_context()));
let bucket = format!("put-lock-context-{}", Uuid::new_v4());
runtime
.block_on(Box::pin(store.make_bucket(&bucket, &MakeBucketOptions::default())))
.expect("test bucket");
let input = PutObjectInput::builder()
.bucket(bucket)
.key("test-object".to_string())
.body(Some(StreamingBlob::from_bytes(Bytes::from_static(b"payload"))))
.content_length(Some(7))
.build()
.expect("PUT input");
let usecase = DefaultObjectUsecase::with_context(Some(context));
let recorder = DebuggingRecorder::new();
let snapshotter = recorder.snapshotter();
let mut request = S3HttpRequestGuard::new("PUT");
let error = metrics::with_local_recorder(&recorder, || {
let fs = FS::new();
let mut future = Box::pin(PUT_STORAGE_TEST_FAILURE.scope(
(
PutStorageTestStage::Commit,
StorageError::Lock(LockError::timeout("private-bucket/private-key", Duration::from_secs(5))),
),
Box::pin(usecase.execute_put_object(&fs, build_request(input, Method::PUT))),
));
runtime.block_on(std::future::poll_fn(|cx| {
request.in_scope(|| std::future::Future::poll(future.as_mut(), cx))
}))
})
.expect_err("lock failure must reject the PUT");
assert_eq!(error.code(), &S3ErrorCode::ServiceUnavailable);
let rows = snapshotter.snapshot().into_vec();
let (key, _, _, value) = rows
.iter()
.find(|(key, _, _, _)| key.key().name() == "rustfs_s3_lock_failures_total")
.expect("spawned commit lock failure must be counted");
assert!(matches!(value, DebugValue::Counter(1)));
let labels = key
.key()
.labels()
.map(|label| (label.key(), label.value()))
.collect::<std::collections::BTreeMap<_, _>>();
assert_eq!(labels["op"], S3Operation::PutObject.as_str());
assert_eq!(labels["resource_class"], "object_namespace");
assert_eq!(labels["mode"], "unknown");
assert_eq!(labels["reason"], "timeout");
assert!(!labels.values().any(|value| value.contains("private")));
}
#[tokio::test]
#[serial_test::serial]
async fn put_read_quorum_failures_are_normalized_at_each_stage() {
+1 -1
View File
@@ -874,7 +874,7 @@ pub(crate) fn build_put_like_object_lock_metadata(
Ok(Some(eval_metadata))
}
fn put_like_write_creates_new_version(opts: &ObjectOptions) -> bool {
pub(super) fn put_like_write_creates_new_version(opts: &ObjectOptions) -> bool {
opts.version_id.is_none() && opts.versioned && !opts.version_suspended
}
+165 -25
View File
@@ -600,9 +600,10 @@ fn validate_select_request(headers: &http::HeaderMap, input: &mut SelectObjectCo
if input.request.expression.len() > MAX_SELECT_EXPRESSION_BYTES {
return Err(S3Error::new(S3ErrorCode::ExpressionTooLong));
}
if input.request.expression_type.as_str() != ExpressionType::SQL {
let Some(expression_type) = canonical_enum(input.request.expression_type.as_str(), &[ExpressionType::SQL]) else {
return Err(S3Error::new(S3ErrorCode::InvalidExpressionType));
}
};
input.request.expression_type = ExpressionType::from_static(expression_type);
normalize_input_serialization(&mut input.request.input_serialization)?;
let compressed_input = is_compressed_input(&input.request.input_serialization);
@@ -649,6 +650,15 @@ fn is_compressed_input(input: &InputSerialization) -> bool {
.is_some_and(|compression| compression.as_str() != CompressionType::NONE)
}
/// Matches an S3 enum value in any ASCII case, as MinIO does, and returns the
/// canonical spelling that downstream code compares against.
fn canonical_enum(value: &str, allowed: &[&'static str]) -> Option<&'static str> {
allowed
.iter()
.copied()
.find(|candidate| candidate.eq_ignore_ascii_case(value))
}
fn normalize_input_serialization(input: &mut InputSerialization) -> S3Result<()> {
let format_count =
usize::from(input.csv.is_some()) + usize::from(input.json.is_some()) + usize::from(input.parquet.is_some());
@@ -659,19 +669,20 @@ fn normalize_input_serialization(input: &mut InputSerialization) -> S3Result<()>
return Err(S3Error::new(S3ErrorCode::ObjectSerializationConflict));
}
if let Some(compression) = input.compression_type.as_ref() {
match compression.as_str() {
CompressionType::NONE => {}
CompressionType::GZIP | CompressionType::BZIP2 => {
if input.parquet.is_some() {
return Err(S3Error::with_message(
S3ErrorCode::InvalidRequestParameter,
INVALID_REQUEST_PARAMETER_MESSAGE,
));
}
}
_ => return Err(map_select_error_to_s3(&SelectError::InvalidCompressionFormat)),
if let Some(compression) = input.compression_type.as_mut() {
let Some(canonical) = canonical_enum(
compression.as_str(),
&[CompressionType::NONE, CompressionType::GZIP, CompressionType::BZIP2],
) else {
return Err(map_select_error_to_s3(&SelectError::InvalidCompressionFormat));
};
if canonical != CompressionType::NONE && input.parquet.is_some() {
return Err(S3Error::with_message(
S3ErrorCode::InvalidRequestParameter,
INVALID_REQUEST_PARAMETER_MESSAGE,
));
}
*compression = CompressionType::from_static(canonical);
}
input
.compression_type
@@ -681,15 +692,16 @@ fn normalize_input_serialization(input: &mut InputSerialization) -> S3Result<()>
let file_header_info = csv
.file_header_info
.get_or_insert_with(|| FileHeaderInfo::from_static(FileHeaderInfo::NONE));
if !matches!(
let Some(canonical) = canonical_enum(
file_header_info.as_str(),
FileHeaderInfo::NONE | FileHeaderInfo::USE | FileHeaderInfo::IGNORE
) {
&[FileHeaderInfo::NONE, FileHeaderInfo::USE, FileHeaderInfo::IGNORE],
) else {
return Err(S3Error::with_message(
S3ErrorCode::InvalidFileHeaderInfo,
"The FileHeaderInfo value is not valid. Only NONE, USE, and IGNORE are supported.",
));
}
};
*file_header_info = FileHeaderInfo::from_static(canonical);
validate_single_byte(csv.comments.as_deref(), S3ErrorCode::InvalidRequestParameter)?;
validate_single_character(csv.quote_character.as_deref())?;
validate_single_character(csv.quote_escape_character.as_deref())?;
@@ -699,9 +711,10 @@ fn normalize_input_serialization(input: &mut InputSerialization) -> S3Result<()>
if let Some(json) = input.json.as_mut() {
let json_type = json.type_.get_or_insert_with(|| JSONType::from_static(JSONType::LINES));
if !matches!(json_type.as_str(), JSONType::DOCUMENT | JSONType::LINES) {
let Some(canonical) = canonical_enum(json_type.as_str(), &[JSONType::DOCUMENT, JSONType::LINES]) else {
return Err(S3Error::new(S3ErrorCode::InvalidJsonType));
}
};
*json_type = JSONType::from_static(canonical);
}
Ok(())
@@ -716,15 +729,16 @@ fn normalize_output_serialization(output: &mut OutputSerialization) -> S3Result<
return Err(S3Error::new(S3ErrorCode::ObjectSerializationConflict));
}
if let Some(csv) = output.csv.as_ref() {
if let Some(csv) = output.csv.as_mut() {
validate_single_byte(csv.field_delimiter.as_deref(), S3ErrorCode::InvalidRequestParameter)?;
validate_single_byte(csv.quote_character.as_deref(), S3ErrorCode::InvalidRequestParameter)?;
validate_single_byte(csv.quote_escape_character.as_deref(), S3ErrorCode::InvalidRequestParameter)?;
validate_output_record_delimiter(csv.record_delimiter.as_deref())?;
if let Some(quote_fields) = csv.quote_fields.as_ref()
&& !matches!(quote_fields.as_str(), QuoteFields::ALWAYS | QuoteFields::ASNEEDED)
{
return Err(S3Error::new(S3ErrorCode::InvalidQuoteFields));
if let Some(quote_fields) = csv.quote_fields.as_mut() {
let Some(canonical) = canonical_enum(quote_fields.as_str(), &[QuoteFields::ALWAYS, QuoteFields::ASNEEDED]) else {
return Err(S3Error::new(S3ErrorCode::InvalidQuoteFields));
};
*quote_fields = QuoteFields::from_static(canonical);
}
return Ok(SelectOutputFormat::Csv(csv.clone()));
}
@@ -3639,6 +3653,132 @@ mod tests {
);
}
fn json_input(json_type: &str) -> SelectObjectContentInput {
let mut input = base_input();
input.request.input_serialization.csv = None;
input.request.input_serialization.json = Some(JSONInput {
type_: Some(JSONType::from(json_type.to_string())),
});
input
}
#[test]
fn validate_matches_enum_values_case_insensitively_and_canonicalizes_them() {
for (value, expected) in [
("gzip", CompressionType::GZIP),
("bZip2", CompressionType::BZIP2),
("none", CompressionType::NONE),
] {
let mut input = base_input();
input.request.input_serialization.compression_type = Some(CompressionType::from(value.to_string()));
validate_select_request(&HeaderMap::new(), &mut input).expect("CompressionType should match any case");
assert_eq!(
input
.request
.input_serialization
.compression_type
.as_ref()
.map(|v| v.as_str()),
Some(expected)
);
}
for (value, expected) in [("lines", JSONType::LINES), ("Document", JSONType::DOCUMENT)] {
let mut input = json_input(value);
validate_select_request(&HeaderMap::new(), &mut input).expect("JSON Type should match any case");
let json_type = input
.request
.input_serialization
.json
.as_ref()
.and_then(|json| json.type_.as_ref());
assert_eq!(json_type.map(|v| v.as_str()), Some(expected));
}
let mut input = json_input("document");
validate_select_request(&HeaderMap::new(), &mut input).expect("lowercase DOCUMENT should be accepted");
assert!(is_json_document(input.request.input_serialization.json.as_ref().expect("JSON input")));
for (value, expected) in [
("use", FileHeaderInfo::USE),
("Ignore", FileHeaderInfo::IGNORE),
("none", FileHeaderInfo::NONE),
] {
let mut input = base_input();
input
.request
.input_serialization
.csv
.as_mut()
.expect("base input should use CSV")
.file_header_info = Some(FileHeaderInfo::from(value.to_string()));
validate_select_request(&HeaderMap::new(), &mut input).expect("FileHeaderInfo should match any case");
let csv = input.request.input_serialization.csv.as_ref().expect("CSV input");
assert_eq!(csv.file_header_info.as_ref().map(|v| v.as_str()), Some(expected));
}
for (value, expected) in [("always", QuoteFields::ALWAYS), ("AsNeeded", QuoteFields::ASNEEDED)] {
let mut input = base_input();
input
.request
.output_serialization
.csv
.as_mut()
.expect("base output should use CSV")
.quote_fields = Some(QuoteFields::from(value.to_string()));
let validation = validate_select_request(&HeaderMap::new(), &mut input).expect("QuoteFields should match any case");
let SelectOutputFormat::Csv(csv) = validation.output_format else {
panic!("CSV output expected");
};
assert_eq!(csv.quote_fields.as_ref().map(|v| v.as_str()), Some(expected));
}
let mut input = base_input();
input.request.expression_type = ExpressionType::from("sql".to_string());
validate_select_request(&HeaderMap::new(), &mut input).expect("ExpressionType should match any case");
assert_eq!(input.request.expression_type.as_str(), ExpressionType::SQL);
}
#[test]
fn validate_still_rejects_unknown_enum_values_in_any_case() {
for value in ["gzipx", "snappy", " GZIP"] {
let mut input = base_input();
input.request.input_serialization.compression_type = Some(CompressionType::from(value.to_string()));
let error = validate_select_request(&HeaderMap::new(), &mut input).expect_err("unknown compression must fail");
assert_eq!(error.code(), &S3ErrorCode::InvalidCompressionFormat, "{value}");
}
for value in ["ndjson", "lines "] {
let error =
validate_select_request(&HeaderMap::new(), &mut json_input(value)).expect_err("unknown JSON type must fail");
assert_eq!(error.code(), &S3ErrorCode::InvalidJsonType, "{value}");
}
let mut input = base_input();
input
.request
.input_serialization
.csv
.as_mut()
.expect("base input should use CSV")
.file_header_info = Some(FileHeaderInfo::from("used".to_string()));
let error = validate_select_request(&HeaderMap::new(), &mut input).expect_err("unknown header mode must fail");
assert_eq!(error.code(), &S3ErrorCode::InvalidFileHeaderInfo);
let mut input = base_input();
input
.request
.output_serialization
.csv
.as_mut()
.expect("base output should use CSV")
.quote_fields = Some(QuoteFields::from("never".to_string()));
let error = validate_select_request(&HeaderMap::new(), &mut input).expect_err("unknown QuoteFields must fail");
assert_eq!(error.code(), &S3ErrorCode::InvalidQuoteFields);
let mut input = base_input();
input.request.expression_type = ExpressionType::from("sqlite".to_string());
let error = validate_select_request(&HeaderMap::new(), &mut input).expect_err("unknown ExpressionType must fail");
assert_eq!(error.code(), &S3ErrorCode::InvalidExpressionType);
}
#[test]
fn validate_accepts_two_byte_csv_input_record_delimiter() {
let mut input = base_input();
+122 -7
View File
@@ -64,18 +64,35 @@ fn lock_resource_class(resource: Option<&str>) -> &'static str {
}
}
fn lock_failure_reason(error: &LockError) -> &'static str {
match error {
LockError::Timeout { .. } => "timeout",
LockError::AlreadyLocked { .. } => "contention",
LockError::QuorumNotReached { .. } => "quorum",
LockError::InsufficientNodes { .. } => "insufficient_nodes",
LockError::Network { .. } => "network",
LockError::Internal { .. } => "internal",
_ => "other",
}
}
fn record_s3_lock_failure(err: &StorageError) {
let record = |resource: Option<&str>, mode: &'static str, reason: &'static str| {
rustfs_io_metrics::s3_http_metrics::record_s3_http_lock_failure(lock_resource_class(resource), mode, reason);
};
match err {
StorageError::Lock(lock_error) => match lock_error {
LockError::Timeout { resource, .. } => record(Some(resource), "unknown", "timeout"),
LockError::AlreadyLocked { resource, .. } => record(Some(resource), "unknown", "contention"),
LockError::QuorumNotReached { .. } | LockError::InsufficientNodes { .. } => record(None, "unknown", "quorum"),
LockError::Network { .. } | LockError::Internal { .. } => record(None, "unknown", "transport"),
_ => record(None, "unknown", "other"),
},
StorageError::Lock(lock_error) => {
let resource = match lock_error {
LockError::Timeout { resource, .. } | LockError::AlreadyLocked { resource, .. } => Some(resource.as_str()),
_ => None,
};
let reason = match lock_failure_reason(lock_error) {
"insufficient_nodes" => "quorum",
"network" | "internal" => "transport",
reason => reason,
};
record(resource, "unknown", reason);
}
StorageError::NamespaceLockQuorumUnavailable {
mode, bucket, object, ..
} => {
@@ -309,6 +326,7 @@ impl std::error::Error for ApiError {
#[derive(Debug, Default, PartialEq, Eq)]
pub(crate) struct ApiErrorDiagnostic {
pub storage_code: Option<StorageErrorCode>,
pub lock_reason: Option<&'static str>,
pub io_kind: Option<std::io::ErrorKind>,
pub io_os_error: Option<i32>,
pub rpc_code: Option<tonic::Code>,
@@ -350,6 +368,18 @@ impl ApiErrorDiagnostic {
};
if let Some(storage) = error.downcast_ref::<StorageError>() {
diagnostic.storage_code = Some(storage.code());
match storage {
StorageError::Lock(lock_error) => {
diagnostic.lock_reason = Some(lock_failure_reason(lock_error));
}
StorageError::NamespaceLockQuorumUnavailable { .. } => {
diagnostic.lock_reason = Some("quorum");
}
_ => {}
}
}
if let Some(lock_error) = error.downcast_ref::<LockError>() {
diagnostic.lock_reason = Some(lock_failure_reason(lock_error));
}
if let Some(status) = error.downcast_ref::<tonic::Status>() {
diagnostic.rpc_code = Some(status.code());
@@ -1263,6 +1293,91 @@ mod tests {
assert!(!format!("{:?}", error.diagnostic()).contains("secret"));
}
#[test]
fn api_error_diagnostic_classifies_lock_failures_without_exposing_resources() {
let cases = [
(
LockError::timeout("private-bucket/private-key", std::time::Duration::from_secs(5)),
"timeout",
),
(
LockError::AlreadyLocked {
resource: "private-bucket/private-key".into(),
owner: "private-owner".into(),
},
"contention",
),
(
LockError::QuorumNotReached {
required: 3,
achieved: 2,
},
"quorum",
),
(
LockError::InsufficientNodes {
required: 3,
available: 2,
},
"insufficient_nodes",
),
(LockError::internal("private-transport-detail"), "internal"),
(LockError::network("private-network-detail", IoError::other("private-source")), "network"),
(
LockError::QueueFull {
message: "private-queue-detail".into(),
},
"other",
),
];
for (failure, expected) in cases {
let api = ApiError::from(StorageError::Lock(failure));
assert_eq!(api.code, S3ErrorCode::ServiceUnavailable);
let diagnostic = api.diagnostic();
assert_eq!(diagnostic.storage_code, Some(StorageErrorCode::Lock));
assert_eq!(diagnostic.lock_reason, Some(expected));
assert!(!format!("{diagnostic:?}").contains("private"));
let s3: S3Error = api.into();
assert_eq!(s3.code(), &S3ErrorCode::ServiceUnavailable);
assert_eq!(super::ApiErrorDiagnostic::from_error(&s3).lock_reason, Some(expected));
}
let nested = ApiError::from(StorageError::Io(IoError::other(StorageError::Lock(LockError::timeout(
"private-bucket/private-key",
std::time::Duration::from_secs(5),
)))));
assert_eq!(nested.diagnostic().lock_reason, Some("timeout"));
assert!(!format!("{:?}", nested.diagnostic()).contains("private"));
let nested_lock = ApiError::from(StorageError::Lock(LockError::network(
"private-transport",
StorageError::Lock(LockError::timeout("private-bucket/private-key", std::time::Duration::from_secs(5))),
)));
assert_eq!(nested_lock.diagnostic().storage_code, Some(StorageErrorCode::Lock));
assert_eq!(nested_lock.diagnostic().lock_reason, Some("timeout"));
assert!(!format!("{:?}", nested_lock.diagnostic()).contains("private"));
let nested_direct = ApiError::from(StorageError::Lock(LockError::network(
"private-transport",
LockError::timeout("private-bucket/private-key", std::time::Duration::from_secs(5)),
)));
assert_eq!(nested_direct.diagnostic().lock_reason, Some("timeout"));
assert!(!format!("{:?}", nested_direct.diagnostic()).contains("private"));
let unavailable = ApiError::from(StorageError::NamespaceLockQuorumUnavailable {
mode: "write",
bucket: "private-bucket".into(),
object: "private-key".into(),
required: 3,
achieved: 2,
});
assert_eq!(unavailable.diagnostic().lock_reason, Some("quorum"));
assert!(!format!("{:?}", unavailable.diagnostic()).contains("private"));
let other = ApiError::from(StorageError::ErasureWriteQuorum);
assert_eq!(other.diagnostic().lock_reason, None);
}
#[test]
fn api_error_diagnostic_survives_s3_conversion_without_changing_response() {
let api = ApiError::from(StorageError::Io(IoError::from_raw_os_error(30)));
+2
View File
@@ -130,6 +130,7 @@ fn session_list_buckets_request(
region: None,
request_context: Some(RequestContext::fallback()),
suppress_denial_log: false,
suppress_denial_audit: false,
});
S3Request {
@@ -324,6 +325,7 @@ impl ProtocolStorageClient {
region: None,
request_context: Some(RequestContext::fallback()),
suppress_denial_log: false,
suppress_denial_audit: false,
});
let req = S3Request {
+341 -18
View File
@@ -25,6 +25,8 @@ use crate::auth::{
use crate::error::{ApiError, normalize_read_quorum_error};
use crate::license::license_check;
use crate::shared_types::RemoteAddr;
use crate::storage::denial_audit::DenialAuditor;
use crate::storage::helper::DenialScope;
use crate::storage::request_context::RequestContext;
use crate::storage::storage_api::contract::bucket::BUCKET_LIFECYCLE_LOCK_OBJECT;
use crate::storage::storage_api::contract::namespace::NamespaceLocking as _;
@@ -91,6 +93,9 @@ pub(crate) struct ReqInfo {
/// collection) so `authorize_request` logs those routine denials at `debug`
/// instead of `warn` (issue #5740).
pub suppress_denial_log: bool,
/// Set by callers that audit their denials in aggregate (DeleteObjects), so each per-key
/// denial is not audited on its own. The denial is still logged.
pub suppress_denial_audit: bool,
}
pub(crate) fn replication_request_authorized<T>(req: &S3Request<T>) -> bool {
@@ -1191,6 +1196,8 @@ pub(crate) fn log_list_buckets_iam_implicit_deny<T>(req: &S3Request<T>) -> S3Res
let req_info = req_info_ref(req)?;
let denial = DenialContext {
quiet: true,
suppress_audit: true,
auditor: DenialAuditor::global(),
bucket: req_info.bucket.as_deref().unwrap_or_default(),
object: req_info.object.as_deref().unwrap_or_default(),
version_id: req_info.version_id.as_deref(),
@@ -1306,6 +1313,8 @@ pub(crate) fn owner_can_bypass_policy_deny(is_owner: bool, action: &Action) -> b
/// instead of `warn`.
struct DenialContext<'a> {
quiet: bool,
suppress_audit: bool,
auditor: &'a DenialAuditor,
bucket: &'a str,
object: &'a str,
version_id: Option<&'a str>,
@@ -1342,14 +1351,68 @@ impl DenialContext<'_> {
);
}
/// Build the generic AccessDenied response after recording the structured
/// denial event.
fn deny(&self, reason: &'static str, action: Action) -> S3Error {
/// Log the denial and, unless it is a quiet probe denial or audited in aggregate by the
/// caller, audit it.
fn record<T>(&self, req: &S3Request<T>, reason: &'static str, action: Action) {
self.log(reason, action);
if !self.quiet && !self.suppress_audit {
self.auditor.audit(req, &action, reason, DenialScope::Request);
}
}
/// Build the generic AccessDenied response after recording the structured
/// denial event and, unless this is a quiet probe denial, an audit entry.
fn deny<T>(&self, req: &S3Request<T>, reason: &'static str, action: Action) -> S3Error {
self.record(req, reason, action);
s3_error!(AccessDenied, "Access Denied")
}
}
/// Audits a denial at an exit that has no [`DenialContext`] of its own. It only audits: those exits
/// keep their existing logging. The request's own probe flags are honored.
fn audit_request_denial<T>(req: &S3Request<T>, auditor: &DenialAuditor, reason: &'static str, action: Action) {
let info = req.extensions.get::<ReqInfo>();
if info.is_some_and(|info| info.suppress_denial_log || info.suppress_denial_audit) {
return;
}
auditor.audit(req, &action, reason, DenialScope::Request);
}
/// Audits the refused keys of a bulk request (DeleteObjects) in aggregate instead of once per key.
///
/// `start` marks the request so per-key denials are logged but not audited; `finish` clears the
/// mark and emits one entry per denied action with the number of keys refused for it.
pub(crate) struct BulkDenialAudit {
denied: Vec<(Action, u64)>,
}
impl BulkDenialAudit {
pub(crate) fn start<T>(req: &mut S3Request<T>) -> S3Result<Self> {
ext_req_info_mut(&mut req.extensions)?.suppress_denial_audit = true;
Ok(Self { denied: Vec::new() })
}
/// Counts one refused key for `action`.
pub(crate) fn denied(&mut self, action: Action) {
match self.denied.iter_mut().find(|(denied_action, _)| *denied_action == action) {
Some((_, count)) => *count += 1,
None => self.denied.push((action, 1)),
}
}
pub(crate) fn finish<T>(self, req: &mut S3Request<T>) -> S3Result<()> {
self.finish_with(req, DenialAuditor::global())
}
fn finish_with<T>(self, req: &mut S3Request<T>, auditor: &DenialAuditor) -> S3Result<()> {
ext_req_info_mut(&mut req.extensions)?.suppress_denial_audit = false;
for (action, keys) in self.denied {
auditor.audit(req, &action, "bulk_request_keys_denied", DenialScope::Keys(keys));
}
Ok(())
}
}
/// Authorizes the request based on the action and credentials.
pub async fn authorize_request<T>(req: &mut S3Request<T>, action: Action) -> S3Result<()> {
authorize_request_with_location_constraint(req, action, None).await
@@ -1369,9 +1432,12 @@ async fn authorize_request_with_location_constraint<T>(
let object = req_info.object.clone().unwrap_or_default();
let version_id = req_info.version_id.clone();
let quiet_denial = req_info.suppress_denial_log;
let suppress_denial_audit = req_info.suppress_denial_audit;
let store = request_object_store(req)?;
let denial = DenialContext {
quiet: quiet_denial,
suppress_audit: suppress_denial_audit,
auditor: DenialAuditor::global(),
bucket: bucket.as_str(),
object: object.as_str(),
version_id: version_id.as_deref(),
@@ -1472,7 +1538,7 @@ async fn authorize_request_with_location_constraint<T>(
.await
.map_err(ApiError::from)?
{
return Err(denial.deny("bucket_policy_explicit_deny", action));
return Err(denial.deny(req, "bucket_policy_explicit_deny", action));
}
let iam_allowed = {
@@ -1514,7 +1580,7 @@ async fn authorize_request_with_location_constraint<T>(
.await
.map_err(iam_authorization_error)?
{
return Err(denial.deny("iam_explicit_deny", action));
return Err(denial.deny(req, "iam_explicit_deny", action));
}
}
allowed
@@ -1628,7 +1694,7 @@ async fn authorize_request_with_location_constraint<T>(
.await
.map_err(ApiError::from)?
{
return Err(denial.deny("bucket_policy_explicit_deny", action));
return Err(denial.deny(req, "bucket_policy_explicit_deny", action));
}
if action != Action::S3Action(S3Action::ListAllMyBucketsAction) {
@@ -1670,11 +1736,11 @@ async fn authorize_request_with_location_constraint<T>(
deny_anonymous_table_data_plane_if_needed(req, action, bucket.as_str(), object.as_str()).await?;
// RestrictPublicBuckets: when true, deny public access even if bucket policy allows it.
match store.restricts_public_bucket_access(bucket_name).await {
Ok(true) => return Err(denial.deny("restrict_public_buckets", action)),
Ok(true) => return Err(denial.deny(req, "restrict_public_buckets", action)),
Ok(false) => {}
Err(StorageError::ConfigNotFound) => {}
Err(_) => {
return Err(denial.deny("public_access_block_unavailable", action));
return Err(denial.deny(req, "public_access_block_unavailable", action));
}
}
return Ok(());
@@ -1682,7 +1748,7 @@ async fn authorize_request_with_location_constraint<T>(
}
}
Err(denial.deny("no_policy_allows_action", action))
Err(denial.deny(req, "no_policy_allows_action", action))
}
// Multipart parts are staged and become visible only after CompleteMultipartUpload,
@@ -2178,6 +2244,7 @@ async fn authorize_table_data_plane_if_needed<T>(
?admin_action,
"table data-plane access denied by table resource policy"
);
audit_request_denial(req, DenialAuditor::global(), "table_resource_policy_denied", action);
Err(s3_error!(AccessDenied, "Access Denied"))
}
@@ -2206,6 +2273,7 @@ async fn deny_anonymous_table_data_plane_if_needed<T>(
.await?
.is_some()
{
audit_request_denial(req, DenialAuditor::global(), "anonymous_table_data_plane_access", action);
return Err(s3_error!(AccessDenied, "Access Denied"));
}
Ok(())
@@ -2642,6 +2710,12 @@ impl S3Access for FS {
authorize_request(req, Action::S3Action(S3Action::ForceDeleteObjectAction)).await?;
}
if !recursive_force_delete_has_authenticated_caller(&req.headers, authenticated, replica_request) {
audit_request_denial(
req,
DenialAuditor::global(),
"recursive_force_delete_requires_authenticated_caller",
Action::S3Action(S3Action::ForceDeleteObjectAction),
);
return Err(s3_error!(AccessDenied, "Recursive force-delete requires an authenticated caller"));
}
@@ -3605,12 +3679,14 @@ mod tests {
table_data_plane_content_mutation, table_data_plane_resource_for_request, table_publication_guard_error,
validate_post_object_success_controls, versioned_read_action, versioned_tagging_action,
};
use super::{BulkDenialAudit, audit_request_denial};
use super::{
remaining_retention_days, requested_bucket_location_constraint, set_requested_object_lock_retain_until,
set_requested_object_lock_retention,
};
use crate::auth::OBJECT_LOCK_REMAINING_RETENTION_DAYS_CONDITION;
use crate::error::ApiError;
use crate::storage::denial_audit::{DenialAuditor, DenialRateLimiter};
use crate::storage::storage_api::contract::bucket::{BucketOperations as _, DeleteBucketOptions, MakeBucketOptions};
use crate::storage::storage_api::contract::multipart::MultipartOperations as _;
use crate::storage::storage_api::contract::object::{ObjectIO as _, ObjectOperations as _};
@@ -5107,21 +5183,268 @@ mod tests {
));
}
type CapturedAudit = Arc<std::sync::Mutex<Vec<rustfs_audit::AuditEntry>>>;
/// An auditor wired like the global one except that entries land in the returned vector.
fn test_auditor(enabled: bool, authenticated_rate: u64, anonymous_rate: u64) -> (DenialAuditor, CapturedAudit) {
let captured: CapturedAudit = Arc::default();
let sink = Arc::clone(&captured);
// A frozen clock: no wall-clock refill can change what the limiter admits.
let t0 = std::time::Instant::now();
let limiter = DenialRateLimiter::new(authenticated_rate, anonymous_rate, t0);
let auditor =
DenialAuditor::new(limiter, move || enabled, move |_, entry| sink.lock().unwrap().push(entry)).with_clock(move || t0);
(auditor, captured)
}
fn denial_request(request_id: &str, authenticated: bool) -> S3Request<()> {
let mut req = build_request((), Method::GET);
req.uri = Uri::from_static("/audit-bucket/audit-object");
req.extensions.insert(crate::storage::request_context::RequestContext {
request_id: request_id.to_string(),
x_amz_request_id: request_id.to_string(),
trace_id: None,
span_id: None,
start_time: std::time::Instant::now(),
});
req.extensions.insert(ReqInfo {
cred: authenticated.then(|| rustfs_credentials::Credentials {
access_key: "denied-user".to_string(),
..Default::default()
}),
bucket: Some("audit-bucket".to_string()),
object: Some("audit-object".to_string()),
..Default::default()
});
req
}
fn denial_context(auditor: &DenialAuditor, quiet: bool) -> DenialContext<'_> {
DenialContext {
quiet,
suppress_audit: false,
auditor,
bucket: "audit-bucket",
object: "audit-object",
version_id: Some("version"),
account: Some("denied-user"),
is_owner: false,
}
}
const GET_OBJECT: Action = Action::S3Action(S3Action::GetObjectAction);
/// Issue #5740: the denial helper must keep the client-facing error identical
/// to the historical bare response — the added diagnostics are log-only.
#[test]
fn test_access_denied_helper_keeps_generic_wire_response() {
let denial = DenialContext {
quiet: false,
bucket: "bucket",
object: "object",
version_id: Some("version"),
account: Some("account"),
is_owner: false,
};
let err = denial.deny("bucket_policy_explicit_deny", Action::S3Action(S3Action::GetObjectAction));
let (auditor, captured) = test_auditor(true, 10, 10);
let req = denial_request("denial-wire-response", true);
let err = denial_context(&auditor, false).deny(&req, "bucket_policy_explicit_deny", GET_OBJECT);
assert_eq!(err.code(), &S3ErrorCode::AccessDenied);
assert_eq!(err.message(), Some("Access Denied"));
assert_eq!(captured.lock().unwrap().len(), 1);
}
/// A real denial yields exactly one audit entry naming the account, bucket, object,
/// denied action and internal reason.
#[test]
fn denial_is_audited_once_with_account_resource_action_and_reason() {
let (auditor, captured) = test_auditor(true, 10, 10);
let req = denial_request("denial-audited-once", true);
let _ = denial_context(&auditor, false).deny(&req, "iam_explicit_deny", Action::S3Action(S3Action::PutObjectAction));
let entries = captured.lock().unwrap();
assert_eq!(entries.len(), 1);
let entry = &entries[0];
assert_eq!(entry.request_id.as_deref(), Some("denial-audited-once"));
assert_eq!(entry.access_key.as_deref(), Some("denied-user"));
assert_eq!(entry.api.name.as_deref(), Some("s3:PutObject"));
assert_eq!(entry.api.bucket.as_deref(), Some("audit-bucket"));
assert_eq!(entry.api.object.as_deref(), Some("audit-object"));
assert_eq!(entry.api.status.as_deref(), Some("failure"));
assert_eq!(entry.api.status_code, Some(403));
assert_eq!(entry.error.as_deref(), Some("Access Denied"));
let tags = entry.tags.as_ref().expect("denial entry carries its reason");
assert_eq!(tags.get("denialReason").and_then(|value| value.as_str()), Some("iam_explicit_deny"));
assert!(!tags.contains_key("suppressedDenials"));
}
/// Probe denials (per-bucket ListBuckets fallback, per-object metadata collection) are
/// expected filter outcomes and must not flood the audit stream.
#[test]
fn quiet_probe_denial_is_not_audited() {
let (auditor, captured) = test_auditor(true, 10, 10);
let req = denial_request("denial-quiet-probe", true);
let err =
denial_context(&auditor, true).deny(&req, "no_policy_allows_action", Action::S3Action(S3Action::ListBucketAction));
assert_eq!(err.code(), &S3ErrorCode::AccessDenied);
assert!(captured.lock().unwrap().is_empty());
}
#[test]
fn denial_is_not_audited_when_the_audit_module_is_off() {
let (auditor, captured) = test_auditor(false, 10, 10);
let req = denial_request("denial-audit-off", true);
let _ = denial_context(&auditor, false).deny(&req, "no_policy_allows_action", GET_OBJECT);
assert!(captured.lock().unwrap().is_empty());
}
#[test]
fn denial_audited_in_aggregate_by_the_caller_is_not_audited_per_key() {
let (auditor, captured) = test_auditor(true, 10, 10);
let req = denial_request("denial-per-key", true);
let mut denial = denial_context(&auditor, false);
denial.suppress_audit = true;
let _ = denial.deny(&req, "iam_explicit_deny", GET_OBJECT);
assert!(captured.lock().unwrap().is_empty());
}
/// A flood of denials emits a bounded number of entries; the rest are only counted.
#[test]
fn denial_flood_emits_bounded_entries() {
let (auditor, captured) = test_auditor(true, 3, 3);
let req = denial_request("denial-flood", true);
for _ in 0..20 {
let _ = denial_context(&auditor, false).deny(&req, "no_policy_allows_action", GET_OBJECT);
}
assert_eq!(captured.lock().unwrap().len(), 3);
}
/// The entry that follows a suppressed run reports how many denials were dropped.
#[test]
fn entry_after_a_flood_reports_the_suppressed_count() {
use crate::storage::helper::DenialScope;
let (auditor, captured) = test_auditor(true, 1, 1);
let req = denial_request("denial-suppressed-count", true);
let t0 = std::time::Instant::now();
for _ in 0..5 {
auditor.audit_at(&req, &GET_OBJECT, "no_policy_allows_action", DenialScope::Request, t0);
}
auditor.audit_at(
&req,
&GET_OBJECT,
"no_policy_allows_action",
DenialScope::Request,
t0 + std::time::Duration::from_secs(2),
);
let entries = captured.lock().unwrap();
assert_eq!(entries.len(), 2);
let tag = |index: usize| {
entries[index]
.tags
.as_ref()
.unwrap()
.get("suppressedDenials")
.and_then(|v| v.as_u64())
};
assert_eq!(tag(0), None);
assert_eq!(tag(1), Some(4));
}
/// Anonymous callers have their own, lower limit, so an unauthenticated scanner cannot
/// use up the budget for authenticated callers.
#[test]
fn anonymous_denials_are_limited_separately_from_authenticated_ones() {
let (auditor, captured) = test_auditor(true, 10, 1);
let anonymous = denial_request("denial-anonymous", false);
let authenticated = denial_request("denial-authenticated", true);
for _ in 0..5 {
let _ = denial_context(&auditor, false).deny(&anonymous, "no_policy_allows_action", GET_OBJECT);
}
for _ in 0..5 {
let _ = denial_context(&auditor, false).deny(&authenticated, "no_policy_allows_action", GET_OBJECT);
}
let entries = captured.lock().unwrap();
let count = |id: &str| entries.iter().filter(|entry| entry.request_id.as_deref() == Some(id)).count();
assert_eq!(count("denial-anonymous"), 1);
assert_eq!(count("denial-authenticated"), 5);
}
/// Exits that have no `DenialContext` of their own audit through the request's own state
/// and honor the same probe flags.
#[test]
fn request_denial_exit_is_audited_and_honors_the_probe_flags() {
let (auditor, captured) = test_auditor(true, 10, 10);
let mut req = denial_request("denial-exit", true);
audit_request_denial(&req, &auditor, "table_resource_policy_denied", GET_OBJECT);
assert_eq!(captured.lock().unwrap().len(), 1);
req.extensions.get_mut::<ReqInfo>().unwrap().suppress_denial_log = true;
audit_request_denial(&req, &auditor, "table_resource_policy_denied", GET_OBJECT);
assert_eq!(captured.lock().unwrap().len(), 1);
req.extensions.get_mut::<ReqInfo>().unwrap().suppress_denial_log = false;
req.extensions.get_mut::<ReqInfo>().unwrap().suppress_denial_audit = true;
audit_request_denial(&req, &auditor, "table_resource_policy_denied", GET_OBJECT);
assert_eq!(captured.lock().unwrap().len(), 1);
}
/// The table-policy, anonymous table and force-delete exits live deep in the authorization
/// flow and need a table catalog to reach. This source-level guard fails if one stops
/// auditing; it does not execute them.
#[test]
fn denial_exits_without_a_context_call_the_audit_function() {
// Only the production code counts; the tests' own calls must not satisfy the guard.
let source = include_str!("access.rs").split("\nmod tests {").next().unwrap();
for reason in [
"\"table_resource_policy_denied\"",
"\"anonymous_table_data_plane_access\"",
"\"recursive_force_delete_requires_authenticated_caller\"",
] {
let call = source
.match_indices("audit_request_denial(")
.any(|(at, _)| source[at..].chars().take(260).collect::<String>().contains(reason));
assert!(call, "{reason} exit does not call audit_request_denial");
}
}
/// DeleteObjects audits its refused keys as one entry per denied action, carrying the count and
/// no single object, and the per-key mark is cleared afterwards.
#[test]
fn bulk_denial_is_one_entry_per_denied_action_with_the_key_count() {
let (auditor, captured) = test_auditor(true, 10, 10);
let mut req = denial_request("denial-bulk", true);
let nothing_denied = BulkDenialAudit::start(&mut req).unwrap();
assert!(req.extensions.get::<ReqInfo>().unwrap().suppress_denial_audit);
nothing_denied.finish_with(&mut req, &auditor).unwrap();
assert!(!req.extensions.get::<ReqInfo>().unwrap().suppress_denial_audit);
assert!(captured.lock().unwrap().is_empty());
let mut bulk = BulkDenialAudit::start(&mut req).unwrap();
let delete = Action::S3Action(S3Action::DeleteObjectAction);
let bypass = Action::S3Action(S3Action::BypassGovernanceRetentionAction);
for action in [delete, delete, bypass, delete] {
bulk.denied(action);
}
bulk.finish_with(&mut req, &auditor).unwrap();
let entries = captured.lock().unwrap();
assert_eq!(entries.len(), 2);
let summary: Vec<_> = entries
.iter()
.map(|entry| {
assert_eq!(entry.api.object, None);
(
entry.api.name.clone().unwrap(),
entry
.tags
.as_ref()
.unwrap()
.get("deniedKeys")
.and_then(|value| value.as_u64())
.unwrap(),
)
})
.collect();
assert_eq!(
summary,
vec![
("s3:DeleteObject".to_string(), 3),
("s3:BypassGovernanceRetention".to_string(), 1)
]
);
}
#[test]
+254
View File
@@ -0,0 +1,254 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//! Audit entries for authorization denials.
//!
//! A denial is caused by the caller, not by an operation the server chose to run, so the
//! rate of entries is bounded: a process-wide token bucket per caller class (authenticated
//! and anonymous), with a lower default for anonymous callers. Suppressed denials are
//! counted and the count is carried on the next entry that is let through.
use crate::module_switches::is_audit_module_enabled;
use crate::storage::access::{ReqInfo, request_context_from_req};
use crate::storage::helper::{DenialRequest, DenialScope, authorization_denial_entry, spawn_background_with_context};
use crate::storage::request_context::RequestContext;
use rustfs_audit::{AuditEntry, global::AuditLogger};
use rustfs_policy::policy::action::Action;
use serde_json::Value;
use std::sync::{Mutex, OnceLock};
use std::time::Instant;
/// Tag on an entry that reports how many denials were suppressed since the previous entry.
pub(crate) const SUPPRESSED_DENIALS_TAG: &str = "suppressedDenials";
struct Bucket {
rate: f64,
burst: f64,
tokens: f64,
/// Denials dropped since this class last emitted an entry.
suppressed: u64,
}
impl Bucket {
fn new(rate_per_sec: u64) -> Self {
let rate = rate_per_sec as f64;
let burst = rate.max(1.0);
Self {
rate,
burst,
tokens: burst,
suppressed: 0,
}
}
fn refill(&mut self, elapsed_secs: f64) {
self.tokens = (self.tokens + elapsed_secs * self.rate).min(self.burst);
}
/// Takes one token. `Suppress` drops the denial; a bucket with rate 0 is switched
/// off and does not count what it drops.
fn admit(&mut self) -> Admission {
if self.rate > 0.0 && self.tokens >= 1.0 {
self.tokens -= 1.0;
Admission::Emit {
suppressed: std::mem::take(&mut self.suppressed),
}
} else {
if self.rate > 0.0 {
self.suppressed = self.suppressed.saturating_add(1);
}
Admission::Suppress
}
}
}
struct LimiterState {
authenticated: Bucket,
anonymous: Bucket,
last: Instant,
}
/// Outcome of asking the limiter for one denial entry.
#[derive(Debug, PartialEq, Eq)]
pub(crate) enum Admission {
/// Emit the entry; `suppressed` denials were dropped since the previous one.
Emit {
suppressed: u64,
},
Suppress,
}
pub(crate) struct DenialRateLimiter {
state: Mutex<LimiterState>,
}
impl DenialRateLimiter {
pub(crate) fn new(authenticated_rate: u64, anonymous_rate: u64, now: Instant) -> Self {
Self {
state: Mutex::new(LimiterState {
authenticated: Bucket::new(authenticated_rate),
anonymous: Bucket::new(anonymous_rate),
last: now,
}),
}
}
pub(crate) fn admit(&self, anonymous: bool, now: Instant) -> Admission {
let mut state = self.state.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
let elapsed = now.saturating_duration_since(state.last).as_secs_f64();
state.last = now.max(state.last);
state.authenticated.refill(elapsed);
state.anonymous.refill(elapsed);
if anonymous {
state.anonymous.admit()
} else {
state.authenticated.admit()
}
}
}
type Emit = Box<dyn Fn(Option<RequestContext>, AuditEntry) + Send + Sync>;
/// Builds, bounds and dispatches denial audit entries.
pub(crate) struct DenialAuditor {
limiter: DenialRateLimiter,
enabled: Box<dyn Fn() -> bool + Send + Sync>,
emit: Emit,
clock: Box<dyn Fn() -> Instant + Send + Sync>,
}
impl DenialAuditor {
pub(crate) fn new(
limiter: DenialRateLimiter,
enabled: impl Fn() -> bool + Send + Sync + 'static,
emit: impl Fn(Option<RequestContext>, AuditEntry) + Send + Sync + 'static,
) -> Self {
Self {
limiter,
enabled: Box::new(enabled),
emit: Box::new(emit),
clock: Box::new(Instant::now),
}
}
/// Replaces the clock the limiter is driven by; tests use a frozen one so no wall-clock
/// refill can change an outcome.
#[cfg(test)]
pub(crate) fn with_clock(mut self, clock: impl Fn() -> Instant + Send + Sync + 'static) -> Self {
self.clock = Box::new(clock);
self
}
/// The process-wide auditor: audit switch from the module state, rates from the environment,
/// entries dispatched to the audit system in the background.
pub(crate) fn global() -> &'static Self {
static AUDITOR: OnceLock<DenialAuditor> = OnceLock::new();
AUDITOR.get_or_init(|| {
let limiter = DenialRateLimiter::new(
rustfs_utils::get_env_u64(rustfs_config::ENV_AUDIT_DENIAL_RATE, rustfs_config::DEFAULT_AUDIT_DENIAL_RATE),
rustfs_utils::get_env_u64(
rustfs_config::ENV_AUDIT_DENIAL_ANONYMOUS_RATE,
rustfs_config::DEFAULT_AUDIT_DENIAL_ANONYMOUS_RATE,
),
Instant::now(),
);
DenialAuditor::new(limiter, is_audit_module_enabled, |context, entry| {
spawn_background_with_context(context, async move {
AuditLogger::log(entry).await;
});
})
})
}
pub(crate) fn audit<T>(&self, req: &DenialRequest<T>, action: &Action, reason: &'static str, scope: DenialScope) {
self.audit_at(req, action, reason, scope, (self.clock)());
}
pub(crate) fn audit_at<T>(
&self,
req: &DenialRequest<T>,
action: &Action,
reason: &'static str,
scope: DenialScope,
now: Instant,
) {
if !(self.enabled)() {
return;
}
let anonymous = req.extensions.get::<ReqInfo>().is_none_or(|info| info.cred.is_none());
let Admission::Emit { suppressed } = self.limiter.admit(anonymous, now) else {
return;
};
let mut entry = authorization_denial_entry(req, action, reason, scope);
if suppressed > 0 {
entry
.tags
.get_or_insert_with(Default::default)
.insert(SUPPRESSED_DENIALS_TAG.to_string(), Value::from(suppressed));
}
(self.emit)(request_context_from_req(req), entry);
}
}
#[cfg(test)]
mod tests {
use super::{Admission, DenialRateLimiter};
use std::time::{Duration, Instant};
#[test]
fn burst_is_bounded_and_suppressed_denials_are_counted_on_the_next_entry() {
let t0 = Instant::now();
let limiter = DenialRateLimiter::new(2, 1, t0);
assert_eq!(limiter.admit(false, t0), Admission::Emit { suppressed: 0 });
assert_eq!(limiter.admit(false, t0), Admission::Emit { suppressed: 0 });
for _ in 0..5 {
assert_eq!(limiter.admit(false, t0), Admission::Suppress);
}
// One second later the bucket has refilled: the next entry reports what was dropped.
assert_eq!(limiter.admit(false, t0 + Duration::from_secs(1)), Admission::Emit { suppressed: 5 });
assert_eq!(limiter.admit(false, t0 + Duration::from_secs(1)), Admission::Emit { suppressed: 0 });
}
#[test]
fn suppressed_counts_are_kept_per_caller_class() {
let t0 = Instant::now();
let limiter = DenialRateLimiter::new(1, 1, t0);
assert_eq!(limiter.admit(true, t0), Admission::Emit { suppressed: 0 });
for _ in 0..4 {
assert_eq!(limiter.admit(true, t0), Admission::Suppress);
}
// The authenticated class dropped nothing, so it must not inherit the anonymous drops.
assert_eq!(limiter.admit(false, t0), Admission::Emit { suppressed: 0 });
assert_eq!(limiter.admit(true, t0 + Duration::from_secs(1)), Admission::Emit { suppressed: 4 });
}
#[test]
fn a_zero_rate_disables_that_class_without_counting() {
let t0 = Instant::now();
let limiter = DenialRateLimiter::new(2, 0, t0);
assert_eq!(limiter.admit(true, t0), Admission::Suppress);
assert_eq!(limiter.admit(false, t0), Admission::Emit { suppressed: 0 });
}
#[test]
fn tokens_never_exceed_the_burst() {
let t0 = Instant::now();
let limiter = DenialRateLimiter::new(2, 2, t0);
let later = t0 + Duration::from_secs(3600);
assert_eq!(limiter.admit(false, later), Admission::Emit { suppressed: 0 });
assert_eq!(limiter.admit(false, later), Admission::Emit { suppressed: 0 });
assert_eq!(limiter.admit(false, later), Admission::Suppress);
}
}
+23 -19
View File
@@ -1815,12 +1815,13 @@ impl S3 for FS {
return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
};
let _ = store
.get_bucket_info(&bucket, &BucketOptions::default())
.await
.map_err(ApiError::from)?;
validate_bucket_object_lock_enabled(&bucket).await?;
if let Err(err) = validate_bucket_object_lock_enabled(&bucket).await {
let _ = store
.get_bucket_info(&bucket, &BucketOptions::default())
.await
.map_err(ApiError::from)?;
return Err(err);
}
let opts: ObjectOptions = get_opts(&bucket, &key, version_id, None, &req.headers)
.await
@@ -1841,19 +1842,22 @@ impl S3 for FS {
// once BEFORE the commit and persist the pending marker with it, so a restart between
// the commit and the worker write-back leaves a Pending status on disk for the scanner
// to re-drive. Scheduling without that marker would lose the task silently.
let dsc = match store.get_object_info(&bucket, &key, &opts).await.ok() {
Some(info) => {
must_replicate_metadata(
&bucket,
&key,
&info.user_defined,
info.user_tags.as_ref().clone(),
popts.delete_marker_replication_status(),
popts.clone(),
)
.await
}
None => ReplicateDecision::new(),
let dsc = if crate::app::storage_api::bucket::metadata_sys::get_replication_config(&bucket)
.await
.is_ok()
&& let Ok(info) = store.get_object_info(&bucket, &key, &opts).await
{
must_replicate_metadata(
&bucket,
&key,
&info.user_defined,
info.user_tags.as_ref().clone(),
popts.delete_marker_replication_status(),
popts.clone(),
)
.await
} else {
ReplicateDecision::new()
};
let mut eval_metadata = parse_object_lock_legal_hold(legal_hold)?;
+110 -29
View File
@@ -22,12 +22,13 @@ use hashbrown::HashMap;
use http::StatusCode;
use metrics::counter;
use rustfs_audit::{
ObjectVersion,
ApiDetails, AuditEntry, ObjectVersion,
entity::{ApiDetailsBuilder, AuditEntryBuilder},
global::AuditLogger,
};
use rustfs_io_metrics::record_s3_op;
use rustfs_notify::EventArgsBuilder;
use rustfs_policy::policy::action::Action;
use rustfs_s3_ops::{S3Operation, operation_matches_event_name};
use rustfs_s3_types::EventName;
use rustfs_targets::{
@@ -39,6 +40,9 @@ use s3s::{S3Request, S3Response, S3Result};
use serde_json::Value;
use std::future::Future;
use tokio::runtime::{Builder, Handle};
/// Request type seen by the denial auditor, so `denial_audit` stays off the direct s3s surface.
pub(crate) type DenialRequest<T> = S3Request<T>;
use tracing::{Instrument, info_span, warn};
#[cfg(feature = "diagnostic-read-version-caller")]
@@ -97,6 +101,42 @@ pub(crate) fn build_event_resp_elements<T>(response: &S3Response<T>, request_id:
resp_elements
}
/// Builds the audit entry shared by every request-scoped audit source: caller identity,
/// client address and the request line, without any outcome.
fn request_audit_builder<T>(
req: &S3Request<T>,
req_info: Option<&ReqInfo>,
event: EventName,
trigger: &str,
api: ApiDetails,
request_id: &str,
) -> AuditEntryBuilder {
let remote_host = req
.extensions
.get::<ClientInfo>()
.map(|info| info.real_ip.to_string())
.or_else(|| {
req.extensions
.get::<Option<RemoteAddr>>()
.and_then(|addr| addr.map(|addr| addr.0.ip().to_string()))
})
.unwrap_or_default();
let mut builder = AuditEntryBuilder::new("1.0", event, trigger, api)
.remote_host(remote_host)
.user_agent(get_request_user_agent(&req.headers))
.req_host(get_request_host(&req.headers))
.req_path(req.uri.path().to_string())
.req_query(extract_req_params(req))
.request_id(request_id);
if let Some(cred) = req_info
.and_then(|info| info.cred.as_ref())
.filter(|cred| !cred.access_key.is_empty())
{
builder = builder.access_key(&cred.access_key);
}
builder
}
/// A unified helper structure for building and distributing audit logs and event notifications via RAII mode at the end of an S3 operation scope.
pub enum OperationHelper {
Disabled,
@@ -205,34 +245,8 @@ impl OperationHelper {
let request_context = request_context.unwrap_or_else(|| RequestContext::from_external_headers(&req.headers));
let request_id = request_context.request_id.clone();
let audit_builder = if audit_enabled {
let remote_host = req
.extensions
.get::<ClientInfo>()
.map(|info| info.real_ip.to_string())
.or_else(|| {
req.extensions
.get::<Option<RemoteAddr>>()
.and_then(|addr| addr.map(|addr| addr.0.ip().to_string()))
})
.unwrap_or_default();
let mut builder = AuditEntryBuilder::new("1.0", event, trigger, api_builder.clone().build())
.remote_host(remote_host)
.user_agent(get_request_user_agent(&req.headers))
.req_host(get_request_host(&req.headers))
.req_path(req.uri.path().to_string())
.req_query(extract_req_params(req))
.request_id(&request_id);
if let Some(cred) = req_info
.and_then(|info| info.cred.as_ref())
.filter(|cred| !cred.access_key.is_empty())
{
builder = builder.access_key(&cred.access_key);
}
Some(builder)
} else {
None
};
let audit_builder =
audit_enabled.then(|| request_audit_builder(req, req_info, event, trigger, api_builder.clone().build(), &request_id));
let event_object = ObjectInfo {
bucket: bucket.clone(),
@@ -440,6 +454,73 @@ impl OperationHelper {
}
}
/// What an authorization denial entry describes.
#[derive(Clone, Copy, Debug)]
pub(crate) enum DenialScope {
/// One refused request for the request's own bucket and object.
Request,
/// A bulk request (DeleteObjects) in which this many keys were refused; no single object applies.
Keys(u64),
}
/// Builds the audit entry for a request the authorization layer refused.
///
/// The entry carries the action that was denied and the internal denial reason, so an
/// audit consumer can tell a refused request from a handler that never ran.
pub(crate) fn authorization_denial_entry<T>(
req: &S3Request<T>,
action: &Action,
reason: &'static str,
scope: DenialScope,
) -> AuditEntry {
let req_info = req.extensions.get::<ReqInfo>();
let path = req.uri.path().trim_start_matches('/');
let mut segs = path.splitn(2, '/');
let path_bucket = segs.next().unwrap_or("");
let path_object = segs.next().unwrap_or("");
let bucket = req_info
.and_then(|info| info.bucket.as_deref())
.filter(|value| !value.is_empty())
.unwrap_or(path_bucket);
let object = match scope {
DenialScope::Request => req_info
.and_then(|info| info.object.as_deref())
.filter(|value| !value.is_empty())
.unwrap_or(path_object),
DenialScope::Keys(_) => "",
};
let action_name: &str = action.into();
let mut api = ApiDetailsBuilder::new()
.name(action_name)
.status("failure")
.status_code(StatusCode::FORBIDDEN.as_u16() as i32);
if !bucket.is_empty() {
api = api.bucket(bucket);
}
if !object.is_empty() {
api = api.object(object);
}
let request_context = request_context_from_req(req).unwrap_or_else(|| RequestContext::from_external_headers(&req.headers));
let mut tags = HashMap::new();
tags.insert("denialReason".to_string(), Value::String(reason.to_string()));
if let DenialScope::Keys(denied) = scope {
tags.insert("deniedKeys".to_string(), Value::from(denied));
}
request_audit_builder(
req,
req_info,
EventName::AuditS3Operation,
action_name,
api.build(),
&request_context.request_id,
)
.error("Access Denied")
.tags(tags)
.build()
}
/// Runs `handler` on `req` and records the outcome as an audit-only entry for `op`.
///
/// The helper is built before the request is handed over so it sees the original headers
+1
View File
@@ -15,6 +15,7 @@
pub mod access;
pub mod concurrency;
pub mod deadlock_detector;
pub(crate) mod denial_audit;
pub mod ecfs;
pub(crate) mod helper;
pub mod options;
+2 -2
View File
@@ -73,7 +73,7 @@ def partition(listing, junit, protected, count=2):
durations[key] = seconds
require(set(durations) == expected, "duration baseline does not cover the selected tests")
binaries = {binary for binary, _ in expected}
require(all(re.fullmatch(r"[A-Za-z0-9_.:-]+", binary) for binary in binaries), "unsupported binary ID")
require(all(re.fullmatch(r"[A-Za-z0-9_./:-]+", binary) for binary in binaries), "unsupported binary ID")
reserved = {binary for binary, _ in protected}
components = [reserved] if reserved else []
components += [{binary} for binary in sorted(binaries - reserved)]
@@ -130,7 +130,7 @@ def load_plan(directory):
protected = set(plan["protected_binaries"])
require(protected <= expected and any(protected <= set(shard["binaries"]) for shard in shards), "split resource component")
for shard in shards:
require(shard["binaries"] and all(re.fullmatch(r"[A-Za-z0-9_.:-]+", name) for name in shard["binaries"]), "invalid shard binary")
require(shard["binaries"] and all(re.fullmatch(r"[A-Za-z0-9_./:-]+", name) for name in shard["binaries"]), "invalid shard binary")
require(shard["filter"] == " or ".join(f"binary_id(={name})" for name in shard["binaries"]), "shard filter mismatch")
return plan, listing
+3 -1
View File
@@ -77,6 +77,7 @@ class RealNextestTests(unittest.TestCase):
(root / 'rust-toolchain.toml').write_text((shards.ROOT / 'rust-toolchain.toml').read_text())
(root / 'Cargo.toml').write_text('[package]\nname="probe"\nversion="0.1.0"\nedition="2021"\n')
(root / 'src/lib.rs').write_text('#[test] fn first() { assert!(std::env::var_os("SHARD_FAIL").is_none()); }\n')
(root / 'src/main.rs').write_text('fn main() {}\n#[test] fn binary_case() {}\n')
(root / 'tests/second.rs').write_text('#[test] fn second() {}\n#[test] #[ignore] fn omitted() {}\n')
(root / 'tests/third.rs').write_text('#[test] fn third() {}\n')
(root / '.config/nextest.toml').write_text('''experimental = ["setup-scripts"]
@@ -103,6 +104,7 @@ setup = ['fixture']
return result
cargo('archive', '--offline', '--archive-file', str(archive))
full = shards.list_tests(root, ['--archive-file', str(archive), '--extract-to', str(root), '--extract-overwrite'])
self.assertIn('probe::bin/probe', full['rust-suites'])
full_path = root / 'listing.json'
shards.write_json(full_path, full)
cargo('run', '--profile', 'ci', *shards.metadata_arguments(root))
@@ -118,7 +120,7 @@ setup = ['fixture']
evidence = root / 'evidence'
for shard in plan['shards']:
shards.execute(root, plan_dir, archive, shard['index'], evidence / str(shard['index']))
self.assertEqual(shards.verify(plan_dir, evidence), 3)
self.assertEqual(shards.verify(plan_dir, evidence), 4)
floor = root / 'migration-floor.txt'
floor.write_text('1\n')
# Older local nextest versions omit the test kind; all fixture
+6 -1
View File
@@ -919,7 +919,10 @@ class WorkflowTimeoutTests(unittest.TestCase):
with self.subTest(suite=suite):
source = (candidate.ROOT / f".github/workflows/rustfs-{suite}-test.yml").read_text()
job = yaml_block(source.splitlines(), job_id, 2)
job_timeout = 360 if suite == "pool-expand" else 60
# heal: a progressing heal outran the shared 45-minute step
# cap (chain 40796474356), so it gets a 90-minute run budget
# and a 120-minute job budget for setup, report and cleanup.
job_timeout = 360 if suite == "pool-expand" else 120 if suite == "heal" else 60
self.assertIn(f" timeout-minutes: {job_timeout}", job)
steps = named_steps(job)
primary = [step for step in steps.values() if any(
@@ -933,6 +936,8 @@ class WorkflowTimeoutTests(unittest.TestCase):
self.assertIsNotNone(timeout_input, event)
self.assertIn(" default: '240'", timeout_input)
self.assertIn(" required: false", timeout_input)
elif suite == "heal":
self.assertIn(" timeout-minutes: 90", primary[0])
else:
self.assertIn(" timeout-minutes: 45", primary[0])
cleanup = "Cleanup environment"
+4 -5
View File
@@ -540,10 +540,10 @@ class FaultToleranceWorkflowContractTests(unittest.TestCase):
"GITHUB_RUN_ID": "314159",
}
def write_result(self, *, seen: int = 44, complete: bool = True,
def write_result(self, *, seen: int = 54, complete: bool = True,
duplicate: bool = False, testing_sha: str | None = None,
unrelated_seen: bool = False, missing_verdict: bool = False) -> None:
expected = [f"CASE-{index:02d}" for index in range(44)]
expected = [f"CASE-{index:02d}" for index in range(54)]
seen_cases = expected[:seen]
if unrelated_seen:
seen_cases = [f"OTHER-{index:02d}" for index in range(seen)]
@@ -555,14 +555,14 @@ class FaultToleranceWorkflowContractTests(unittest.TestCase):
payload = {
"schema_version": 1,
"auto_testing_sha": testing_sha or self.TESTING_SHA,
"selected_scenarios": ["A", "B", "C", "C2", "D", "E"],
"selected_scenarios": ["A", "B", "C", "C2", "D", "E", "F"],
"complete": complete,
"expected_cases": expected,
"seen_cases": seen_cases,
"missing_cases": expected[seen:],
"harness_errors": [],
"counts": {
"expected": 44,
"expected": 54,
"seen": seen,
"unexpected": 1,
"known_divergence": 0,
@@ -1186,7 +1186,6 @@ class TableSuiteTests(unittest.TestCase):
self.assertIn("runs-on: smoke-testing", source)
# New gate semantics: case failures stay green, harness breakdowns red.
self.assertRegex(source, r"(?m)^ continue-on-error: true")
self.assertIn("# Red only for harness/environment breakdowns; case failures stay green.", source)
self.assertIn("HARNESS_OK", source)
self.assertIn("Product result: ${CASES_PASS} passed, ${CASES_FAIL} failed", source)
self.assertIn('python3 scripts/functional_case_report.py "${LOG_FILE}" "${CASE_TABLE}"', source)