gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-07-26 08:18:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

Address P2 follow-ups from the 2026-07-10..12 merged-PR review (backlog#1210-1220) (#4783)

Browse Source 
* fix(obs): open cleaner compression source with O_NOFOLLOW

The compressor opened the source log via File::open, which follows a
symlink at the final path component. Between the scanner selecting a
regular file and this open, an attacker with write access to the log
directory could swap the entry for a symlink (TOCTOU) pointing at, say,
/etc/shadow, whose contents would then be copied into an archive. Open
the source with O_NOFOLLOW on Unix so such a swap fails with ELOOP; the
temp/archive path already refused symlinks, this closes the source side.

Refs rustfs/backlog#1210
Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(obs): recompress instead of trusting leftover cleaner archives

archive_header_ok only checked the first 2-4 magic bytes before treating
an existing .gz/.zst as a completed prior result and letting the caller
delete the source log. A file with valid magic but a truncated or forged
body passes that check, so an attacker with write access to the log
directory (or a crashed prior run) could plant such a stub and make the
cleaner delete the real log without ever producing a usable archive —
silent audit-data loss.

Chosen fix: stop trusting cross-process leftovers entirely and always
recompress the source in this pass, rather than fully decoding every
leftover to validate it. Full-decode validation would add real CPU cost
and decode-bug surface for a rare crash-recovery case; the existing
atomic create_new+rename already overwrites whatever sits at the archive
path (a planted symlink is replaced, never followed) with a freshly
written, fsync'd archive, so a partial/forged leftover can never gate
source deletion. This is the lowest-regression option.

Refs rustfs/backlog#1211
Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(object-data-cache): cap memory-gate reservation at cache growth headroom

The memory gate subtracts `admitted_since_refresh` from the snapshot's
available bytes so a burst arriving faster than the 5 s refresh cannot
over-allocate. That counter is GROSS: it only rolls over on the refresh and
never rolls back when a fill is later evicted, cancelled, or loses the
invalidation race. Under sustained high-throughput churn (net footprint flat
and far below `max_capacity`) the raw counter balloons past the memory the
cache actually holds, so `effective_available` collapses and the gate reports
false memory pressure — skipping the hottest fills with SkippedMemoryPressure
until the next 5 s refresh. This only lowers hit rate; it never returns wrong
data and self-heals each refresh.

Fix direction 1 (minimal regression): cap the reservation deduction at the
cache's own growth headroom (`max_capacity - weighted_size()`) instead of
letting the unbounded gross counter shrink the system-available budget. The
cache can never hold more than `max_capacity`, so a burst adds at most that
headroom of real memory before moka evicts to stay bounded (net-zero churn
beyond that point) — capping the deduction there keeps the reservation honest
without treating gross churn as growth. Chosen over net-accounting (direction
2, releasing bytes on every failure/cancel/eviction path) because that only
plugs the leak on failed fills and would not address the core defect: churn of
*successful* insert/evict fills over the 5 s window. It also touches only the
gate plus one call site rather than every failure path in moka_backend.

The cap only ever raises `effective_available`, so real memory pressure (a low
snapshot at refresh) still suppresses fills; when the cache is at capacity the
headroom is 0 and the deduction vanishes, correctly reflecting net-zero churn.
`MokaBackend` now stores `max_capacity` and passes the live headroom into
`allows_fill`. Adds targeted gate tests: gross churn far above headroom no
longer falsely suppresses, yet the reservation still bounds a burst while the
cache can genuinely grow.

Refs rustfs/backlog#1212
Co-Authored-By: heihutu <heihutu@gmail.com>

* test(ecstore): assert native O_DIRECT path runs in uring read test

uring_preserves_o_direct_for_eligible_reads only compared bytes through
LocalDisk::read_file_mmap_copy. On a filesystem that rejects O_DIRECT the
read silently degrades to the buffered StdBackend fallback and the byte
check still passes, so the test could go green without the native
read_at_direct path ever executing -- a vacuous pass.

Add a per-disk native_direct_reads counter on UringBackend, incremented
only when pread_uring_direct completes, and rebuild the test to drive a
real UringBackend's pread_bytes and assert the counter is non-zero (every
eligible read went through the native tier). When io_uring or O_DIRECT is
unavailable on the host filesystem (restricted CI runners, tmpfs), the
test skips loudly via eprintln instead of asserting a tautology, while
still checking byte-correctness on whatever tier served the read.

The counter also gives a gray release a positive signal that the O_DIRECT
tier is serving reads, not just a fallback count.

Refs rustfs/backlog#1213
Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(ecstore): warn + count read-time EINVAL on native O_DIRECT reads

classify_direct_read_error is only reached from the read side: the
O_DIRECT open in pread_uring_direct already succeeded (an open-time
refusal is handled earlier as DirectOpenError::ODirectRefused). So an
EINVAL/EOPNOTSUPP arriving here is a read-time error on an fd the kernel
accepted for O_DIRECT -- far more likely an alignment bug in the aligned
read path than an unsupported filesystem. The old code latched the disk's
native path off with only a once-per-disk debug trace, hiding a potential
correctness regression behind a silent buffered-read downgrade.

Diagnostics only: the fallback behaviour is unchanged (the native path is
still latched off and the caller still reads via StdBackend). This adds a
rustfs_io_uring_direct_read_einval_total counter and promotes the
once-per-disk trace from debug to warn so an operator can see an alignment
regression instead of an unexplained latency/CPU shift.

Refs rustfs/backlog#1214
Co-Authored-By: heihutu <heihutu@gmail.com>

* docs(ecstore): document data-blocks-first default and its tail-latency cost

DEFAULT_RUSTFS_GET_DATA_BLOCKS_FIRST_READER_SETUP is true and must stay
true: deferred-parity is the deliberate, already-rolled-out full-object
GET default from backlog#1159/#923. Flipping it back to false in code
would silently revert that rollout for every deployment that has not set
the env var, so this commit only documents -- no behaviour change.

The added notes explain what data-blocks-first does (schedule data shards
up front, engage parity lazily on a missing/corrupt data shard), the known
trade-off (parity is engaged late, so a slow-but-not-dead data drive
raises GET p99 because the faster parity shards are not raced against it
until a data shard is declared missing), and the operational rollback
switch (RUSTFS_GET_DATA_BLOCKS_FIRST_READER_SETUP=false), which is
intentionally an env override rather than a code default change.

No metric was added: the low-risk observability hook for "slow data drive
engaged deferred parity" would live at the deferred-stripe engage point,
which is out of this file's scope; this change stays documentation-only to
avoid touching the hot GET path.

Refs rustfs/backlog#1215
Co-Authored-By: heihutu <heihutu@gmail.com>

* docs(ecstore): document wide-directory walk stall hazard and tuning

list_dir enumerates a whole directory in one os::read_dir call (count =
-1), and the walk caller bounds that entire enumeration with the per-read
stall budget (default 5s) as if it were a single read. For a wide, flat
prefix -- one directory holding millions of immediate children -- a single
readdir can exceed the budget on a healthy disk, trip DiskError::Timeout,
and surface as a ListObjects 500 quorum failure though the drive is fine
(a #2999 sub-class).

This is documented, not rewritten: turning the one-shot readdir into a
streaming/batched enumeration that refreshes the stall deadline between
chunks is an architecture-level change with high regression surface
(ordering, the count contract, quorum merge) and belongs in a separate
follow-up. The supported mitigation today is operational, so the comments
point wide-directory deployments at RUSTFS_DRIVE_WALKDIR_STALL_TIMEOUT_SECS
and the high-latency drive-timeout profile, which widen the budget with no
code change. Notes were added at list_dir, the scan_dir call site, and
get_drive_walkdir_stall_timeout. No behaviour change.

Refs rustfs/backlog#1216
Co-Authored-By: heihutu <heihutu@gmail.com>

* docs(ecstore): document consumer-peek vs producer-stall coupling

In list_path_raw the consumer's peek_timeout is drawn from the same source
and same value (walkdir_stall_timeout, default 5s) as the producer-side
walk stall budget, but the two measure different things: the producer
stall bounds a single drive read, while the consumer peek bounds the gap
between two ADJACENT entries arriving from a reader. Because they share a
value, the consumer cannot wait meaningfully longer for the next entry
than the producer is allowed to spend producing one. Walking a region
dense with non-listable internal items can make a HEALTHY drive miss the
budget between visible entries; the consumer then declares it stalled and
detaches it, dropping a good drive from the merge and capping the "large
prefix succeeds" guarantee.

Documented, not decoupled: giving the consumer peek an independent,
strictly-larger budget would cut these false detaches but equally delays
detaching a genuinely dead drive and shifts listing tail-latency
semantics, so it wants soak data before changing the default. The comment
records the invariant any such follow-up must keep -- consumer peek >=
producer stall, never stricter -- so it can never fail a drive before the
producer would. No behaviour change.

Refs rustfs/backlog#1217
Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(io-metrics): add time-based trigger for low-IOPS latency percentiles

Percentiles were recomputed only every 128 IOs and seeded to 0, so a
low-traffic deployment exported p95/p99 = 0/stale for a long time after
startup. Add a 10s wall-clock trigger alongside the count throttle so the
first recompute can fire before 128 samples accrue. Hot-path per-op mean
update is unchanged.

Refs rustfs/backlog#1218
Co-Authored-By: heihutu <heihutu@gmail.com>

* test(e2e): cover codec-streaming parity under fault injection and NoSuchKey

The codec-streaming compat A/B previously ran only against a healthy
4-disk EC set with successful full GETs: the DiskFaultHarness was
constructed but never faulted, the error path was untested, and the
range assertion silently compared legacy-vs-legacy (ranges always fall
back to the duplex path), overstating what it proved.

Add two genuinely-failable scenarios reusing the existing harness and
fixtures:

- Parity reconstruction A/B: take one data disk offline and re-run the
  full object matrix on both phases while the EC 2+2 set rebuilds each
  large object from the surviving shards. Assert codec == legacy
  byte-for-byte (sha256) and header-for-header, and assert the codec
  phase served the reconstructed objects with zero duplex-pipe fallback
  (the reader gate is drive-health-independent, so the codec fast path
  is really exercised through reconstruction).
- NoSuchKey negative path: compare the HTTP status + S3 error code of a
  missing-key GET across the legacy and codec phases and require them to
  be identical (404/NoSuchKey), guarding against the codec env
  perturbing the error path.

Also clarify the range-phase comment so it is not misread as
codec-range correctness coverage: both sides are served by the same
legacy range path, so the assertion only proves ranges keep working and
keep falling back to legacy with the gates open.

Verified: cargo check/--no-run pass and the test passes locally
(1 passed; dup_codec=0 confirms the codec path ran).

Refs rustfs/backlog#1219
Co-Authored-By: heihutu <heihutu@gmail.com>

* ci(ecstore): exercise native O_DIRECT read path on an ext4 loopback

The uring-integration leg ran on the runner's default TMPDIR, which may sit
on tmpfs/overlayfs where open(O_DIRECT) fails and the native read_at_direct
path silently latches off to the aligned StdBackend fallback. Mount a
dedicated ext4 loopback and point TMPDIR at it so the real io_uring dep
(bumped git->0.1.0->0.2.0->0.2.1) and the native O_DIRECT read path are
actually covered rather than validated only by signature diffing.

Refs rustfs/backlog#1220
Co-Authored-By: heihutu <heihutu@gmail.com>

---------

Co-authored-by: heihutu <heihutu@gmail.com>
This commit is contained in:
houseme
2026-07-13 00:03:28 +08:00
committed by GitHub
parent 63b4568f85
commit 1553dc3f62
10 changed files with 663 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/ci.yml
+18
View File
@@ -377,13 +377,31 @@ jobs:
- name: Install build dependencies
run: sudo apt-get update && sudo apt-get install -y protobuf-compiler
# ext4 supports O_DIRECT; the runner's default TMPDIR may sit on tmpfs or
# overlayfs, where open(O_DIRECT) returns EINVAL/EOPNOTSUPP and the native
# read_at_direct path silently latches off to the aligned StdBackend
# fallback. A dedicated ext4 loopback guarantees the native O_DIRECT read
# path is actually exercised, not just the fallback (rustfs/backlog#1220).
- name: Prepare ext4 loopback for O_DIRECT coverage
run: |
set -euo pipefail
dd if=/dev/zero of=/tmp/rustfs-odirect.img bs=1M count=1024
mkfs.ext4 -q -F /tmp/rustfs-odirect.img
sudo mkdir -p /mnt/rustfs-odirect
sudo mount -o loop /tmp/rustfs-odirect.img /mnt/rustfs-odirect
sudo chmod 1777 /mnt/rustfs-odirect
mount | grep rustfs-odirect
# RUSTFS_URING_TESTS_MUST_RUN=1 makes the io_uring tests fail instead of
# silently skipping if io_uring is unavailable, so this leg can never pass
# vacuously. RUSTFS_IO_URING_READ_ENABLE=true selects the UringBackend.
# TMPDIR points every tempfile-based test fixture at the ext4 loopback so
# eligible reads keep O_DIRECT and the native path is covered end to end.
- name: Run ecstore io_uring integration tests on real io_uring
env:
RUSTFS_IO_URING_READ_ENABLE: "true"
RUSTFS_URING_TESTS_MUST_RUN: "1"
TMPDIR: /mnt/rustfs-odirect
run: cargo test -p rustfs-ecstore uring_ -- --test-threads=1 --nocapture
e2e-tests:
crates/e2e_test/src/get_codec_streaming_compat_test.rs
+136 -5
View File
@@ -39,6 +39,20 @@
//! **zero** — proving the codec path actually ran rather than silently falling
//! back to the very path it is being compared against.
//!
//! Beyond the all-healthy happy path, the suite also drives the codec/legacy
//! A/B under two conditions the `DiskFaultHarness` makes reachable:
//!
//! * Parity reconstruction: one data disk is taken offline
//! (`take_disk_offline`) and the SAME object matrix is GET both ways while
//! the EC 2+2 set rebuilds each large object from the surviving shards. The
//! codec-streaming reader gate never inspects drive health, so the codec
//! fast path is exercised end-to-end through reconstruction; the test
//! asserts byte- and header-equality vs the legacy path AND that the codec
//! phase never fell back to a duplex pipe while reconstructing.
//! * Missing object: a GET for an absent key is compared across both phases
//! to prove the error semantics (HTTP status + S3 error code) are identical
//! — the codec env must not perturb the NoSuchKey negative path.
//!
//! Topology: single-node 4-disk EC set (default 2 data + 2 parity) via the
//! in-process `DiskFaultHarness` (see `chaos.rs`). Small objects are inlined
//! into `xl.meta` (served identically on both paths); large objects span one or
@@ -49,6 +63,7 @@ mod tests {
use crate::chaos::DiskFaultHarness;
use crate::common::init_logging;
use aws_sdk_s3::Client;
use aws_sdk_s3::error::ProvideErrorMetadata;
use aws_sdk_s3::primitives::ByteStream;
use aws_sdk_s3::types::{CompletedMultipartUpload, CompletedPart};
use serial_test::serial;
@@ -63,6 +78,9 @@ mod tests {
const MIB: usize = 1024 * 1024;
const BUCKET: &str = "codec-streaming-compat";
const CONTENT_TYPE: &str = "application/x-rustfs-compat";
/// A key that is never uploaded — used to compare the NoSuchKey negative
/// path across the legacy and codec phases.
const MISSING_KEY: &str = "does-not-exist/ghost.bin";
/// Marker the legacy duplex GET path logs once per full-object read
/// (`crates/ecstore/src/set_disk/ops/object.rs`). Its presence/absence in a
@@ -144,6 +162,23 @@ mod tests {
})
}
/// GET a key expected to be absent, projecting the wire-visible error
/// semantics — HTTP status code plus the S3 error code — so the legacy and
/// codec phases can be asserted to reject a missing object identically. A
/// missing object fails during metadata resolution, before the reader-path
/// gate is consulted, so both phases MUST agree; a divergence here would
/// mean the codec env perturbed the negative path.
async fn missing_key_semantics(client: &Client, key: &str) -> Result<(u16, String), Box<dyn Error + Send + Sync>> {
match client.get_object().bucket(BUCKET).key(key).send().await {
Ok(_) => Err(format!("GET {key} unexpectedly succeeded; expected a NoSuchKey error").into()),
Err(err) => {
let status = err.raw_response().map(|r| r.status().as_u16()).unwrap_or(0);
let code = err.as_service_error().and_then(|e| e.code()).unwrap_or("<none>").to_string();
Ok((status, code))
}
}
}
/// Env that opens every codec-streaming gate to 100% for the codec phase.
/// Mirrors the exact knobs `get_codec_streaming_reader_gate` inspects
/// (`crates/ecstore/src/set_disk/mod.rs`).
@@ -335,6 +370,37 @@ mod tests {
"baseline phase should have used the legacy duplex path for the {num_large} large objects, but only saw {dup_base} duplex markers in {base_log}"
);
// Legacy negative path: a GET for an absent key must fail with a
// well-formed NoSuchKey (404). Captured now so Phase B can prove the
// codec env returns the identical error semantics.
let legacy_missing = missing_key_semantics(&client, MISSING_KEY).await?;
assert_eq!(
legacy_missing,
(404, "NoSuchKey".to_string()),
"legacy GET of a missing key should be 404/NoSuchKey, got {legacy_missing:?}"
);
// ---- Phase A degraded: pull one data disk, force parity reconstruction ----
// With disk0 offline the EC 2+2 set must rebuild every large object's
// data from the surviving data+parity shards. Record the legacy-duplex
// bytes and headers produced under reconstruction so Phase B can prove
// the codec path reconstructs the same bytes and headers. The duplex
// marker snapshot (`dup_base`) is already taken, so these extra reads do
// not affect the path-confirmation assertion above.
harness.take_disk_offline(0)?;
let mut baseline_degraded: BTreeMap<String, GetView> = BTreeMap::new();
for (shape, data) in plain {
let view = get_full(&client, shape.key).await?;
assert_eq!(view.sha256, sha256_hex(data), "degraded baseline body mismatch for {}", shape.key);
assert_eq!(view.len, data.len(), "degraded baseline length mismatch for {}", shape.key);
baseline_degraded.insert(shape.key.to_string(), view);
}
let mp_view = get_full(&client, multipart_key).await?;
assert_eq!(mp_view.sha256, sha256_hex(&multipart_body), "degraded baseline multipart body mismatch");
baseline_degraded.insert(multipart_key.to_string(), mp_view);
// Restore the disk so Phase B restarts from a clean, complete disk set.
harness.bring_disk_online(0)?;
// ---- Phase B: codec streaming (gates opened) ----
harness.kill_server();
for (k, v) in codec_env() {
@@ -355,6 +421,14 @@ mod tests {
assert_eq!(mp_view.sha256, sha256_hex(&multipart_body), "codec multipart body mismatch");
codec.insert(multipart_key.to_string(), mp_view);
// Negative-path equivalence: the codec env must return the exact same
// status + error code as the legacy phase for a missing key.
let codec_missing = missing_key_semantics(&client, MISSING_KEY).await?;
assert_eq!(
codec_missing, legacy_missing,
"NoSuchKey error semantics diverged between codec and legacy phases: codec={codec_missing:?} legacy={legacy_missing:?}"
);
// Snapshot the codec-phase duplex count BEFORE issuing the ranged GET
// (range falls back to the duplex path by design and would pollute it).
sleep(Duration::from_millis(300)).await;
@@ -382,18 +456,75 @@ mod tests {
"codec phase created {dup_codec} duplex pipe(s) for full GETs; the codec-streaming fast path was not exercised (see {codec_log})"
);
// Range GET still served correctly while codec streaming is enabled
// (ranges fall back to the legacy path by gate design).
// Range GET while codec streaming is enabled. NOTE ON COVERAGE: the
// reader gate unconditionally routes every ranged request back to the
// legacy duplex path (`GetCodecStreamingFallbackReason::Range`), so both
// `baseline_range` and `codec_range` are produced by the SAME legacy
// path. This assertion therefore only verifies that ranged GETs keep
// working (and keep falling back to legacy) with the codec gates open —
// it does NOT exercise or validate a codec-streaming range reader, which
// does not exist. It must not be read as codec range-correctness
// coverage.
let codec_range = get_range(&client, "large-3mib", range_spec).await?;
assert_eq!(
codec_range.sha256, baseline_range.sha256,
"ranged GET body diverged under codec streaming"
"ranged GET body diverged with codec streaming enabled (both served by the legacy range path)"
);
assert_eq!(
codec_range.len, baseline_range.len,
"ranged GET length diverged with codec streaming enabled"
);
// ---- Phase B degraded: the same reconstruction, now on the codec path ----
// Re-run the reconstruction A/B with the codec-streaming gates still
// open. The reader gate decision is independent of drive health (it
// never inspects disk state), so the codec fast path is exercised
// end-to-end while the EC set rebuilds each large object from the
// surviving shards — this is a real codec-vs-legacy reconstruction test,
// not legacy-vs-legacy. Snapshot the duplex count first (the range GET
// above already used the duplex path) so we can measure only the markers
// these degraded codec GETs add.
let dup_codec_before_degraded = count_marker(&codec_log, DUPLEX_MARKER);
harness.take_disk_offline(0)?;
let mut codec_degraded: BTreeMap<String, GetView> = BTreeMap::new();
for (shape, data) in plain {
let view = get_full(&client, shape.key).await?;
assert_eq!(view.sha256, sha256_hex(data), "degraded codec body mismatch for {}", shape.key);
codec_degraded.insert(shape.key.to_string(), view);
}
let mp_view = get_full(&client, multipart_key).await?;
assert_eq!(mp_view.sha256, sha256_hex(&multipart_body), "degraded codec multipart body mismatch");
codec_degraded.insert(multipart_key.to_string(), mp_view);
harness.bring_disk_online(0)?;
// A/B under parity reconstruction: codec == legacy, byte-for-byte and
// header-for-header, for every object in the matrix.
for key in baseline_degraded.keys() {
let b = &baseline_degraded[key];
let c = &codec_degraded[key];
assert_eq!(c.sha256, b.sha256, "degraded body hash diverged for {key} (parity reconstruction)");
assert_eq!(c.len, b.len, "degraded body length diverged for {key} (parity reconstruction)");
assert_eq!(
c.headers, b.headers,
"degraded response headers diverged for {key} (parity reconstruction)\nbaseline={:#?}\ncodec={:#?}",
b.headers, c.headers
);
}
// Path confirmation under reconstruction: the codec fast path must have
// served the reconstructed large objects without ever falling back to
// the legacy duplex pipe. Without this, the equivalence above could be
// legacy-vs-legacy and prove nothing about codec reconstruction.
sleep(Duration::from_millis(300)).await;
let dup_codec_degraded = count_marker(&codec_log, DUPLEX_MARKER).saturating_sub(dup_codec_before_degraded);
assert_eq!(
dup_codec_degraded, 0,
"codec phase created {dup_codec_degraded} duplex pipe(s) while reconstructing large objects with disk0 offline; the codec fast path was not exercised under degraded reads (see {codec_log})"
);
assert_eq!(codec_range.len, baseline_range.len, "ranged GET length diverged under codec streaming");
info!(
objects = baseline.len(),
"codec streaming produced byte- and header-identical GET responses vs legacy duplex"
"codec streaming produced byte- and header-identical GET responses vs legacy duplex (healthy + parity-reconstructed + NoSuchKey)"
);
// Best-effort cleanup of the capture logs.
crates/ecstore/src/cache_value/metacache_set.rs
+31
View File
@@ -606,6 +606,37 @@ async fn list_path_raw_inner(
let revjob_rx = rx.clone();
let mut revjobs = JoinSet::new();
revjobs.spawn(async move {
// Consumer-side peek timeout, and a caveat worth understanding
// (rustfs/backlog#1217).
//
// This budget is the SAME SOURCE and SAME VALUE as the producer-side
// walk stall budget: both default to `walkdir_stall_timeout` and fall
// back to `get_drive_walkdir_stall_timeout()` (5s). But the two measure
// different things:
// * producer stall: bounds a single drive READ inside the walk (see
// `with_walk_stall_deadline` in `disk/local.rs`);
// * this consumer peek: bounds the interval between two ADJACENT
// entries arriving from a drive's reader (`peek_with_timeout`).
//
// Because they are coupled to the same value, the consumer cannot wait
// meaningfully longer for the next entry than the producer is allowed to
// spend producing one. When a drive walks a region dense with
// non-listable internal items (many entries the producer filters out
// before emitting the next visible one), a HEALTHY drive can take longer
// than one budget to hand the consumer its next entry. The consumer then
// classifies it as `PeekOutcome::TimedOut` and DETACHES that reader (it
// is replaced by a drained duplex below), dropping a good drive from the
// merge. That caps the "large prefix always succeeds" guarantee: a wide
// enough non-listable stretch can knock healthy drives out of quorum.
//
// This is left documented, not decoupled. Giving the consumer peek an
// independent, strictly-larger budget would reduce these false detaches,
// but it also delays detaching a genuinely dead drive by the same amount
// and shifts listing tail-latency semantics; that trade-off wants soak
// data before it changes the default, so it is deferred to a follow-up.
// The constraint for any such change: the consumer peek must be >= the
// producer stall (never stricter), so it can never declare a drive
// stalled before the producer itself would have failed.
let peek_timeout = opts
.walkdir_stall_timeout
.or({
crates/ecstore/src/disk/disk_store.rs
+12
View File
@@ -189,6 +189,18 @@ pub fn get_drive_walkdir_timeout() -> Duration {
)
}
/// Per-read stall budget for a directory walk: a walk read is failed only if
/// the drive stops answering for this long, not for a walk simply taking a
/// while (see `with_walk_stall_deadline` in `disk/local.rs`).
///
/// Wide-directory tuning (rustfs/backlog#1216): because a whole-directory
/// enumeration (`list_dir` with `count = -1`) is bounded by this budget as one
/// unit, a very wide flat prefix (millions of immediate children) can make a
/// single `readdir` exceed the default on a healthy disk and fail ListObjects.
/// Deployments with such directories should raise
/// `RUSTFS_DRIVE_WALKDIR_STALL_TIMEOUT_SECS`, or select the high-latency
/// drive-timeout profile (which raises this default automatically), to widen
/// the budget without a code change.
pub fn get_drive_walkdir_stall_timeout() -> Duration {
get_drive_timeout_duration(
rustfs_config::ENV_DRIVE_WALKDIR_STALL_TIMEOUT_SECS,
crates/ecstore/src/disk/local.rs
+147 -14
View File
@@ -266,6 +266,13 @@ const METRIC_URING_IN_FLIGHT: &str = "rustfs_io_uring_in_flight";
const METRIC_URING_CQ_OVERFLOW: &str = "rustfs_io_uring_cq_overflow";
#[cfg(target_os = "linux")]
const METRIC_URING_CANCEL_ALREADY: &str = "rustfs_io_uring_cancel_already";
/// Read-side EINVAL/EOPNOTSUPP from a native O_DIRECT read that happened AFTER a
/// successful O_DIRECT open (rustfs/backlog#1214). Unlike an open-time refusal
/// (unsupported filesystem), this most likely means an alignment bug in the
/// aligned read path, so it is surfaced with a counter + warn instead of a
/// once-per-disk debug trace.
#[cfg(target_os = "linux")]
const METRIC_URING_DIRECT_READ_EINVAL_TOTAL: &str = "rustfs_io_uring_direct_read_einval_total";
/// How often the per-disk driver StatsSnapshot is exported to metrics
/// (rustfs/backlog#1172).
#[cfg(target_os = "linux")]
@@ -2961,6 +2968,14 @@ pub(crate) struct UringBackend {
/// `active` gates io_uring as a whole, this gates only the native O_DIRECT
/// read shape.
direct_uring: DirectIoReadState,
/// Count of reads that completed through the native io_uring + O_DIRECT path
/// (`pread_uring_direct`) on this disk (rustfs/backlog#1213). Incremented only
/// on success, so a value `> 0` is proof the native path actually executed
/// rather than silently degrading to the StdBackend fallback. Tests assert on
/// it to avoid a vacuous pass on filesystems that reject O_DIRECT; it also
/// gives a gray release a positive signal that the O_DIRECT tier is serving
/// reads instead of only ever counting fallbacks.
native_direct_reads: std::sync::atomic::AtomicU64,
/// Per-disk descriptor cache (backlog#1145). `None` when
/// `RUSTFS_IO_URING_FD_CACHE` is off, which restores the open-per-read path.
fd_cache: Option<FdCache>,
@@ -3101,6 +3116,7 @@ impl UringBackend {
active: std::sync::atomic::AtomicBool::new(true),
fallback_logged: std::sync::atomic::AtomicBool::new(false),
direct_uring: DirectIoReadState::new(),
native_direct_reads: std::sync::atomic::AtomicU64::new(0),
fd_cache,
})
}
@@ -3209,6 +3225,36 @@ impl UringBackend {
/// classification lives in one place (rustfs/backlog#1174).
fn classify_direct_read_error(&self, io_err: &std::io::Error) {
if is_direct_io_unsupported(io_err) {
// This helper is only ever reached from the READ side: the O_DIRECT
// `open` in `pread_uring_direct` already succeeded, and an open-time
// refusal is handled separately as `DirectOpenError::ODirectRefused`
// before any read is issued. So an EINVAL/EOPNOTSUPP arriving here is
// a *read-time* error on an fd the kernel accepted for O_DIRECT. That
// is far more likely an alignment bug in the aligned read path than a
// filesystem that does not support O_DIRECT -- yet the old code
// latched the whole disk's native path off with only a once-per-disk
// debug trace, making a real correctness bug effectively invisible
// (rustfs/backlog#1214).
//
// Diagnostics only: the fallback behaviour is unchanged. The native
// O_DIRECT path is still latched off and the caller still falls back
// to StdBackend for this and every future eligible read. We only make
// the event observable -- a counter plus a once-per-disk `warn!`
// instead of a silent `debug!` -- so an operator can see an alignment
// regression rather than a mystery latency/CPU shift from buffered
// reads.
counter!(METRIC_URING_DIRECT_READ_EINVAL_TOTAL, "root" => self.root_label.clone()).increment(1);
if !self.direct_uring.fallback_logged.swap(true, Ordering::Relaxed) {
warn!(
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_DISK_LOCAL,
root = %self.root.display(),
error = ?io_err,
"io_uring O_DIRECT read returned EINVAL/EOPNOTSUPP AFTER a successful O_DIRECT open; \
this is more likely an alignment bug than an unsupported filesystem. Latching the \
native O_DIRECT path off and reading via StdBackend (logged once per disk)"
);
}
self.direct_uring.supported.store(false, Ordering::Relaxed);
} else if is_io_uring_unsupported(io_err) {
self.latch_active_off(io_err);
@@ -3478,6 +3524,10 @@ impl UringBackend {
if should_reclaim_file_cache_after_read(length) {
reclaim_read_range(&file_for_reclaim, offset_u64, length)?;
}
// The native io_uring + O_DIRECT read completed (rustfs/backlog#1213):
// record it so callers/tests can distinguish this path from the
// StdBackend fallback, which never reaches here.
self.native_direct_reads.fetch_add(1, Ordering::Relaxed);
Ok(Bytes::from(bytes))
}
}
@@ -5037,6 +5087,15 @@ impl LocalDisk {
let stall = opts.stall_timeout_duration();
// `count = -1` enumerates the whole directory in one `list_dir` call, and
// the stall budget bounds that entire enumeration as a single unit. On a
// WIDE, FLAT directory (millions of immediate children) that one readdir
// can exceed `stall` on a healthy disk and fail the whole walk -- see the
// wide-directory stall hazard documented on `list_dir`
// (rustfs/backlog#1216, a #2999 sub-class). Mitigate operationally with a
// larger `RUSTFS_DRIVE_WALKDIR_STALL_TIMEOUT_SECS` or the high-latency
// drive-timeout profile; a streaming readdir rewrite is a separate,
// higher-risk follow-up and is intentionally not done here.
let mut entries = match with_walk_stall_timeout(stall, self.list_dir("", &opts.bucket, &current, -1)).await {
Ok(res) => res,
Err(e) => {
@@ -6419,6 +6478,29 @@ impl DiskAPI for LocalDisk {
self.io_backend.pread_bytes(volume, path, offset, length, metrics).await
}
/// List a single directory. `count < 0` enumerates the *whole* directory in
/// one `os::read_dir` call.
///
/// Wide-directory stall hazard (rustfs/backlog#1216, a #2999 sub-class):
/// the walk caller wraps this whole call in the per-read stall budget
/// (`with_walk_stall_timeout`, default 5s via
/// `RUSTFS_DRIVE_WALKDIR_STALL_TIMEOUT_SECS`) as if the entire directory
/// enumeration were a single read. For a *wide, flat* directory -- one
/// bucket prefix holding millions of immediate children -- a single
/// `readdir` of the whole directory can itself exceed the stall budget on a
/// healthy disk. That trips `DiskError::Timeout`, which the listing path can
/// escalate to a quorum failure and surface to the client as a ListObjects
/// 500, even though nothing is actually wrong with the drive.
///
/// This is deliberately NOT fixed here by rewriting the one-shot
/// `os::read_dir` into a streaming/batched readdir that would refresh the
/// stall deadline between chunks: that is an architecture-level change with
/// high regression surface (ordering, the `count` contract, quorum merge
/// semantics) and is tracked as a separate follow-up. The supported
/// mitigation for wide-directory deployments today is operational -- raise
/// `RUSTFS_DRIVE_WALKDIR_STALL_TIMEOUT_SECS` or run with the high-latency
/// drive-timeout profile (see `get_drive_walkdir_stall_timeout`), both of
/// which widen the budget without any code change.
#[tracing::instrument(level = "trace", skip_all)]
async fn list_dir(&self, origvolume: &str, volume: &str, dir_path: &str, count: i32) -> Result<Vec<String>> {
if !origvolume.is_empty() {
@@ -6433,6 +6515,9 @@ impl DiskAPI for LocalDisk {
let volume_dir = self.get_bucket_path(volume)?;
let dir_path_abs = self.get_object_path(volume, dir_path.trim_start_matches(SLASH_SEPARATOR))?;
// Whole-directory enumeration in one syscall path (see the wide-directory
// stall hazard on this fn): with `count < 0` this reads every entry, and
// the caller's stall budget bounds the entire call as a unit.
let entries = match os::read_dir(&dir_path_abs, count).await {
Ok(res) => res,
Err(e) => {
@@ -14292,12 +14377,23 @@ mod test {
/// an O_DIRECT-eligible read keeps O_DIRECT semantics via the native
/// `read_at_direct` path (or, if that disk can't do io_uring+O_DIRECT, the
/// StdBackend aligned fallback) and still returns exactly the requested
/// bytes for unaligned ranges. This asserts byte-correctness regardless of
/// which tier serves the read — the native path is exercised directly by
/// rustfs-uring's own O_DIRECT test under real io_uring.
/// bytes for unaligned ranges.
///
/// The old shape of this test only checked byte-equivalence through
/// `LocalDisk::read_file_mmap_copy`. On a filesystem that rejects O_DIRECT
/// the read silently degrades to the buffered fallback and the byte check
/// still passes, so the test could go green without the native O_DIRECT path
/// ever running — a vacuous pass (rustfs/backlog#1213). It now builds a real
/// `UringBackend`, drives `pread_bytes` (which routes eligible reads into
/// `pread_uring_direct`), and asserts the native path actually executed via
/// the `native_direct_reads` counter. When the backing filesystem cannot do
/// io_uring or O_DIRECT (restricted CI runners, tmpfs/overlayfs), the test
/// skips loudly with `eprintln!` instead of asserting a tautology — but it
/// still checks byte-correctness on whatever tier served the read.
#[cfg(target_os = "linux")]
#[tokio::test(flavor = "multi_thread")]
async fn uring_preserves_o_direct_for_eligible_reads() {
use std::sync::atomic::Ordering;
use tempfile::tempdir;
// Unaligned on purpose: 3 blocks + 7 bytes.
@@ -14319,27 +14415,43 @@ mod test {
(FILE_LEN - 7, 7),
];
// Threshold 1 makes every non-empty read O_DIRECT-eligible, so each read
// below exercises the O_DIRECT tier. The env must be set before
// LocalDisk::new so the io_uring backend is the one selected.
let root_dir = tempdir().expect("operation should succeed");
let root = root_dir.path().to_path_buf();
// Lay out the shard with LocalDisk, then read it back through a real
// UringBackend so the test can inspect the O_DIRECT latch and the native
// read counter directly.
{
let endpoint = Endpoint::try_from(root.to_string_lossy().as_ref()).expect("operation should succeed");
let disk = LocalDisk::new(&endpoint, false).await.expect("operation should succeed");
disk.make_volume("test-volume").await.expect("operation should succeed");
disk.write_all("test-volume", "shard.bin", content.clone())
.await
.expect("operation should succeed");
}
// Skip if io_uring is unavailable on this host (restricted env, e.g. the
// Kubernetes CI runners): there is no native O_DIRECT path to exercise.
let Some(backend) = UringBackend::try_new(root) else {
uring_test_skip("uring_preserves_o_direct_for_eligible_reads");
return;
};
// Threshold 1 makes every non-empty read O_DIRECT-eligible, so each read
// below drives `pread_bytes` into the native `pread_uring_direct` path.
// These knobs are read per-read, so setting them around the reads is
// enough (the backend was already constructed above).
let got_ranges = temp_env::async_with_vars(
[
(ENV_RUSTFS_IO_URING_READ_ENABLE, Some("true")),
(ENV_RUSTFS_OBJECT_DIRECT_IO_READ_ENABLE, Some("true")),
(ENV_RUSTFS_OBJECT_DIRECT_IO_READ_THRESHOLD, Some("1")),
],
async {
let endpoint = Endpoint::try_from(root_dir.path().to_string_lossy().as_ref()).expect("operation should succeed");
let disk = LocalDisk::new(&endpoint, false).await.expect("operation should succeed");
disk.make_volume("test-volume").await.expect("operation should succeed");
disk.write_all("test-volume", "shard.bin", content.clone())
.await
.expect("operation should succeed");
let mut out = Vec::new();
for (offset, length) in ranges {
out.push(
disk.read_file_mmap_copy("test-volume", "shard.bin", offset, length)
backend
.pread_bytes("test-volume", "shard.bin", offset, length, None)
.await
.expect("O_DIRECT-eligible read must succeed under io_uring (direct or fallback)"),
);
@@ -14349,6 +14461,7 @@ mod test {
)
.await;
// Byte-correctness holds regardless of which tier served the read.
for ((offset, length), got) in ranges.into_iter().zip(got_ranges) {
assert_eq!(
got,
@@ -14356,6 +14469,26 @@ mod test {
"O_DIRECT read mismatch at offset={offset} length={length}"
);
}
// The point of backlog#1213: prove the NATIVE O_DIRECT path executed
// rather than silently passing on the StdBackend fallback. If the
// filesystem refuses O_DIRECT, `direct_uring.supported` latches off and
// no native read is counted — skip loudly instead of asserting nothing.
let native_hits = backend.native_direct_reads.load(Ordering::Relaxed);
let still_supported = backend.direct_uring.supported.load(Ordering::Relaxed);
if still_supported && native_hits > 0 {
assert_eq!(
native_hits,
ranges.len() as u64,
"every eligible read should have gone through the native io_uring O_DIRECT path"
);
} else {
eprintln!(
"SKIP uring_preserves_o_direct_for_eligible_reads: native O_DIRECT path not \
exercised on this filesystem (direct_uring.supported={still_supported}, \
native_direct_reads={native_hits}); byte-correctness was still asserted"
);
}
}
/// Once io_uring is latched off (backlog#1101), reads still return correct
crates/ecstore/src/set_disk/core/io_primitives.rs
+38
View File
@@ -70,6 +70,36 @@ use tokio::task::JoinSet;
pub(in crate::set_disk) const EVENT_SET_DISK_READ: &str = "set_disk_read";
pub(in crate::set_disk) const ENV_RUSTFS_GET_DATA_BLOCKS_FIRST_READER_SETUP: &str = "RUSTFS_GET_DATA_BLOCKS_FIRST_READER_SETUP";
/// Default reader-setup strategy for the GET read path (rustfs/backlog#1215,
/// #1159, #923).
///
/// `true` means "data-blocks-first" is the default for every full-object GET:
/// the bitrot reader setup schedules only the `data_shards` data blocks up
/// front and *defers* the parity shards (see [`BitrotReaderSetupStrategy`] and
/// `DeferredReaderStripeHandle`). Parity reads are engaged lazily, only when a
/// data shard turns out to be missing/corrupt and reconstruction needs them.
///
/// Why this is the default (do NOT flip it back to `false` here):
/// - It is the deliberate, already-rolled-out direction from backlog#1159/#923.
/// deferred-parity is now the full-object GET default path, not an
/// experiment; changing this constant to `false` would silently revert that
/// rollout for every deployment that has not set the env var.
///
/// Known trade-off (the reason this constant carries a hazard note at all):
/// - Deferring parity means the parity disks are engaged *late*. A data disk
/// that is slow-but-not-dead (high tail latency, not an outright failure)
/// therefore holds up the read longer than the all-shards strategy would,
/// because the faster parity shards are not raced against it until a data
/// shard is declared missing. This can raise GET p99 on clusters with a
/// chronically slow data drive. `MetadataFanoutObservation` / the deferred
/// stripe handles are where a "slow data disk engaged deferred parity" signal
/// would be recorded.
///
/// Rollback switch: set `RUSTFS_GET_DATA_BLOCKS_FIRST_READER_SETUP=false` (see
/// [`ENV_RUSTFS_GET_DATA_BLOCKS_FIRST_READER_SETUP`]) to restore the
/// all-shards-up-front behaviour for a deployment without changing this
/// default. This is an operational escape hatch for the tail-latency case
/// above; it is intentionally an env override, not a code change.
const DEFAULT_RUSTFS_GET_DATA_BLOCKS_FIRST_READER_SETUP: bool = true;
pub(in crate::set_disk) const ENV_RUSTFS_GET_CODEC_STREAMING_DATA_BLOCKS_FIRST_READER_SETUP: &str =
"RUSTFS_GET_CODEC_STREAMING_DATA_BLOCKS_FIRST_READER_SETUP";
@@ -987,6 +1017,14 @@ impl BitrotReaderSetupStrategy {
}
}
/// Resolve which bitrot reader-setup strategy a read uses.
///
/// For `ReadQuorum` (the full-object GET path) the default is data-blocks-first
/// / deferred-parity (see [`DEFAULT_RUSTFS_GET_DATA_BLOCKS_FIRST_READER_SETUP`]
/// for the rollout rationale and the tail-latency trade-off). Operators can
/// force the older all-shards-up-front behaviour per deployment by setting
/// `RUSTFS_GET_DATA_BLOCKS_FIRST_READER_SETUP=false`; the constant default must
/// stay `true` (rustfs/backlog#1215/#1159/#923).
pub(in crate::set_disk) fn get_bitrot_reader_setup_strategy(
mode: BitrotReaderSetupMode,
prefer_data_blocks_first: bool,
crates/io-metrics/src/collector.rs
+67 -2
View File
@@ -21,7 +21,7 @@ use super::performance::PerformanceMetrics;
use std::collections::VecDeque;
use std::sync::Arc;
use std::sync::atomic::{AtomicU32, Ordering};
use std::time::Duration;
use std::time::{Duration, Instant};
use tokio::sync::RwLock;
/// How many recorded operations elapse between P95/P99 recomputations.
@@ -34,12 +34,28 @@ use tokio::sync::RwLock;
/// while the per-op sort cost is amortised away.
const PERCENTILE_RECOMPUTE_INTERVAL: u32 = 128;
/// Maximum wall-clock staleness of the P95/P99 percentiles before a recompute is
/// forced regardless of the operation count.
///
/// The count throttle alone starves low-IOPS deployments: at a few operations
/// per minute it can take hours to accumulate `PERCENTILE_RECOMPUTE_INTERVAL`
/// samples, during which the exported p95/p99 stay pinned at their initial `0`
/// (or a long-stale value). This time bound guarantees the percentiles refresh
/// within a bounded delay even under trickle traffic, while leaving the hot,
/// high-IOPS path governed by the cheaper count throttle.
const PERCENTILE_RECOMPUTE_MAX_INTERVAL: Duration = Duration::from_secs(10);
/// Sliding window of I/O latency samples plus a running sum, so the window mean
/// is maintained in O(1) as samples enter and leave.
struct LatencyWindow {
samples: VecDeque<Duration>,
/// Sum of `samples` in microseconds, kept in step with every push/pop.
sum_micros: u128,
/// When the P95/P99 percentiles were last recomputed, used to force a
/// refresh under low-traffic conditions where the count throttle rarely
/// fires. Seeded at construction so the first time-based refresh is measured
/// from collector start, not from an absent prior recompute.
last_percentile_at: Instant,
}
/// Metrics collector for tracking I/O operations and computing latency percentiles.
@@ -70,6 +86,7 @@ impl MetricsCollector {
io_latency: RwLock::new(LatencyWindow {
samples: VecDeque::new(),
sum_micros: 0,
last_percentile_at: Instant::now(),
}),
max_latency_samples,
ops_since_percentile: AtomicU32::new(0),
@@ -130,10 +147,22 @@ impl MetricsCollector {
let len = window.samples.len() as u128;
let mean_us = window.sum_micros.checked_div(len).unwrap_or(0) as u64;
// Two independent recompute triggers, both evaluated under the lock so
// the counter and the timestamp stay consistent:
// * count: cheap amortisation for the hot, high-IOPS path;
// * time: a staleness ceiling so low-IOPS deployments do not export
// an initial/stale 0 for p95/p99 while samples trickle in.
// We just pushed a sample, so the window is guaranteed non-empty and a
// time-forced recompute always has data to sort — including the first
// one, which can fire with far fewer than INTERVAL samples.
let now = Instant::now();
let n = self.ops_since_percentile.fetch_add(1, Ordering::Relaxed) + 1;
let recompute = n >= PERCENTILE_RECOMPUTE_INTERVAL;
let count_due = n >= PERCENTILE_RECOMPUTE_INTERVAL;
let time_due = now.duration_since(window.last_percentile_at) >= PERCENTILE_RECOMPUTE_MAX_INTERVAL;
let recompute = count_due || time_due;
if recompute {
self.ops_since_percentile.store(0, Ordering::Relaxed);
window.last_percentile_at = now;
}
(mean_us, recompute)
};
@@ -266,6 +295,42 @@ mod tests {
assert_eq!(metrics.p99_io_latency_us.load(Ordering::Relaxed), 200);
}
#[tokio::test]
async fn low_frequency_percentiles_recompute_on_time() {
let metrics = Arc::new(PerformanceMetrics::new());
let collector = MetricsCollector::new(metrics.clone(), 1000);
// Only a handful of ops — far below PERCENTILE_RECOMPUTE_INTERVAL — so the
// count throttle alone would never fire and p99 would stay pinned at its
// initial 0 (the low-IOPS staleness bug).
for _ in 0..5 {
collector.record_io_operation(0, Duration::from_micros(200), true).await;
}
assert_eq!(
metrics.p99_io_latency_us.load(Ordering::Relaxed),
0,
"count throttle alone must not have recomputed yet"
);
// Simulate enough wall-clock time elapsing since the last recompute by
// backdating the timestamp past the staleness ceiling.
{
let mut window = collector.io_latency.write().await;
window.last_percentile_at = Instant::now()
.checked_sub(PERCENTILE_RECOMPUTE_MAX_INTERVAL + Duration::from_secs(1))
.expect("monotonic clock has enough history to backdate");
}
// The next op must force a time-based recompute even though the sample
// count is still far below the interval — so p99 is no longer stuck at 0.
collector.record_io_operation(0, Duration::from_micros(200), true).await;
assert_eq!(
metrics.p99_io_latency_us.load(Ordering::Relaxed),
200,
"time-based trigger must refresh p99 on low-traffic deployments"
);
}
#[tokio::test]
async fn test_sample_limit() {
let metrics = Arc::new(PerformanceMetrics::new());
crates/object-data-cache/src/memory.rs
+82 -7
View File
@@ -262,7 +262,12 @@ impl ObjectDataCacheMemoryGate {
///
/// This is lock-free and does no blocking sysinfo read: it only reads the
/// atomic snapshot maintained by the periodic refresher.
pub fn allows_fill(&self, required_bytes: u64) -> bool {
///
/// `cache_growth_headroom` is how many more bytes the cache itself can hold
/// before it is at capacity (`max_capacity - weighted_size()`). It caps how
/// far the in-window reservation may shrink the budget: see the reservation
/// note below.
pub fn allows_fill(&self, required_bytes: u64, cache_growth_headroom: u64) -> bool {
// A zero floor opts out of the gate, so fill admission never depends on
// a live memory reading — which differs between a host and a container.
// This must short-circuit before any snapshot read (see 51a97a81c).
@@ -280,7 +285,20 @@ impl ObjectDataCacheMemoryGate {
// burst that arrives faster than the 5 s refresh: each admission shrinks
// the budget the next one sees, so cumulative admission cannot exceed the
// real headroom even though every fill reads the same (stale) snapshot.
let effective_available = snapshot.available_bytes.saturating_sub(self.snapshot.admitted());
//
// `admitted_since_refresh` counts GROSS admitted bytes and never rolls
// back on a fill that is later evicted, cancelled, or loses the
// invalidation race; it only resets on the 5 s refresh. Under sustained
// churn (net footprint flat, far below capacity) the raw counter would
// balloon past the real memory the cache consumes and falsely trip the
// gate, skipping the hottest fills until the next refresh. The cache can
// never hold more than `max_capacity`, so a burst adds at most
// `cache_growth_headroom` bytes of real memory before moka evicts to stay
// bounded (net-zero churn beyond that point). Capping the deduction there
// keeps the reservation honest without treating gross churn as growth
// (backlog#1212).
let reserved = self.snapshot.admitted().min(cache_growth_headroom);
let effective_available = snapshot.available_bytes.saturating_sub(reserved);
let min_free = u64::from(self.min_free_memory_percent);
let has_percent_budget = effective_available.saturating_mul(100) >= snapshot.total_bytes.saturating_mul(min_free);
@@ -383,7 +401,7 @@ mod tests {
available_bytes: 16 * 1024 * 1024,
}));
assert!(!gate.allows_fill(1024));
assert!(!gate.allows_fill(1024, u64::MAX));
assert_eq!(stats.snapshot().memory_pressure_events, 1);
}
@@ -396,7 +414,7 @@ mod tests {
available_bytes: 500,
}));
assert!(gate.allows_fill(100));
assert!(gate.allows_fill(100, u64::MAX));
}
#[test]
@@ -416,7 +434,7 @@ mod tests {
available_bytes: 1,
}));
assert!(gate.allows_fill(512));
assert!(gate.allows_fill(512, u64::MAX));
assert_eq!(stats.snapshot().memory_pressure_events, 0);
}
@@ -429,10 +447,67 @@ mod tests {
available_bytes: 100,
}));
assert!(!gate.allows_fill(128));
assert!(!gate.allows_fill(128, u64::MAX));
assert_eq!(stats.snapshot().memory_pressure_events, 1);
}
// backlog#1212: `admitted_since_refresh` counts GROSS admitted bytes and
// never rolls back on an evicted/cancelled/lost-race fill, so a churn window
// (net footprint flat, far below capacity) balloons the raw counter past the
// memory the cache actually holds. Deducting it wholesale falsely trips the
// gate; capping the deduction at the cache's growth headroom fixes it.
#[test]
fn reservation_deduction_capped_at_growth_headroom_avoids_false_pressure() {
let stats = Arc::new(ObjectDataCacheStats::default());
let gate = ObjectDataCacheMemoryGate::new(&ObjectDataCacheConfig::default(), Arc::clone(&stats));
gate.set_test_snapshot(Some(ObjectDataCacheMemorySnapshot {
total_bytes: 1_000_000,
available_bytes: 500_000, // 50% free, well above the 20% floor
}));
// A churn window admitted far more GROSS bytes than the cache can hold;
// repeated insert/evict never rolled the counter back, so it now dwarfs
// the real available memory even though the live footprint stays tiny.
gate.snapshot.reserve(10_000_000);
// Uncapped, the raw gross counter swamps the budget and falsely signals
// memory pressure even though the cache's net footprint is flat.
assert!(
!gate.allows_fill(1_000, u64::MAX),
"raw gross admitted-bytes deduction should falsely suppress (the bug being fixed)"
);
// Capping the deduction at the cache's growth headroom (net size flat,
// far below capacity) restores admission: gross churn is no longer
// mistaken for real memory growth.
assert!(
gate.allows_fill(1_000, 100_000),
"capping the reservation at cache growth headroom must not falsely suppress"
);
}
// backlog#1212: capping the deduction must not defeat the reservation under
// genuine cache growth. When the cache still has room to grow, the in-window
// reservation must still shrink the budget so a burst cannot over-admit.
#[test]
fn reservation_still_bounds_burst_within_growth_headroom() {
let stats = Arc::new(ObjectDataCacheStats::default());
let gate = ObjectDataCacheMemoryGate::new(&ObjectDataCacheConfig::default(), Arc::clone(&stats));
gate.set_test_snapshot(Some(ObjectDataCacheMemorySnapshot {
total_bytes: 1_000_000,
available_bytes: 300_000, // 30% free; floor is 20% = 200_000
}));
// The cache can still grow well past the reserved amount, so the cap does
// not bind and the reservation is deducted in full.
gate.snapshot.reserve(150_000);
// 300_000 available - 150_000 reserved = 150_000 effective, below the
// 200_000 floor: the reservation must still suppress the fill.
assert!(
!gate.allows_fill(1_000, u64::MAX),
"reservation must still bound a burst while the cache can genuinely grow"
);
}
// ODC-14: `allows_fill` must read the atomic snapshot without performing an
// inline (blocking) refresh. A synchronous test has no tokio runtime, so no
// refresher task exists; if `allows_fill` refreshed inline it would clobber
@@ -450,7 +525,7 @@ mod tests {
gate.store_raw_snapshot_for_test(sentinel);
// Exercise the gate on the atomic path (no test_override installed).
let _ = gate.allows_fill(1);
let _ = gate.allows_fill(1, u64::MAX);
let after = gate.raw_snapshot_for_test();
assert_eq!(
crates/object-data-cache/src/moka_backend.rs
+15 -1
View File
@@ -32,6 +32,10 @@ pub struct MokaBackend {
index: Arc<ObjectDataCacheIdentityIndex>,
singleflight: ObjectDataCacheSingleflight,
memory_gate: ObjectDataCacheMemoryGate,
/// Cache capacity in weighted bytes. Used to derive how much the cache can
/// still grow (`max_capacity - weighted_size()`), which caps the memory
/// gate's in-window reservation deduction (backlog#1212).
max_capacity: u64,
/// Bounds the number of concurrent distinct-key fills. Singleflight only
/// dedups per key, so without this limiter distinct-key fills are unbounded.
fill_semaphore: Arc<Semaphore>,
@@ -144,6 +148,7 @@ impl MokaBackend {
index,
singleflight: ObjectDataCacheSingleflight::new(Arc::clone(&stats)),
memory_gate: ObjectDataCacheMemoryGate::new(config, stats),
max_capacity,
fill_semaphore: Arc::new(Semaphore::new(fill_permits)),
#[cfg(test)]
fill_barrier: std::sync::Mutex::new(None),
@@ -204,7 +209,16 @@ impl MokaBackend {
Err(_) => return leader.finish(ObjectDataCacheFillResult::SkippedFillConcurrency),
};
if !self.memory_gate.allows_fill(u64::try_from(bytes.len()).unwrap_or(u64::MAX)) {
// How much the cache can still grow before it is at capacity. This caps
// the gate's in-window reservation so sustained gross churn (net size
// flat, far below capacity) cannot be mistaken for real memory growth
// and falsely skip fills (backlog#1212). weighted_size() is moka's
// lazily-maintained approximation, which is all this bound needs.
let cache_growth_headroom = self.max_capacity.saturating_sub(self.cache.weighted_size());
if !self
.memory_gate
.allows_fill(u64::try_from(bytes.len()).unwrap_or(u64::MAX), cache_growth_headroom)
{
return leader.finish(ObjectDataCacheFillResult::SkippedMemoryPressure);
}
crates/obs/src/cleaner/compress.rs
+117 -46
View File
@@ -184,25 +184,28 @@ fn create_tmp_archive(path: &Path, source_mode: Option<u32>) -> std::io::Result<
opts.open(path)
}
/// Cheap sanity check that an existing archive starts with the expected codec
/// magic bytes, so a zero/truncated/foreign file is not trusted as a valid
/// prior result. Only called on a confirmed regular, non-empty file.
fn archive_header_ok(path: &Path, algorithm: CompressionAlgorithm) -> bool {
use std::io::Read;
let mut file = match File::open(path) {
Ok(file) => file,
Err(_) => return false,
};
let mut buf = [0u8; 4];
let read = match file.read(&mut buf) {
Ok(read) => read,
Err(_) => return false,
};
match algorithm {
// gzip: 0x1f 0x8b
CompressionAlgorithm::Gzip => read >= 2 && buf[0] == 0x1f && buf[1] == 0x8b,
// zstd frame magic: 0x28 0xb5 0x2f 0xfd (little-endian 0xFD2FB528)
CompressionAlgorithm::Zstd => read >= 4 && buf == [0x28, 0xb5, 0x2f, 0xfd],
/// Open the compression source file, refusing to follow a symlink at the final
/// path component on Unix.
///
/// The scanner selects a regular log file, but between that scan and this open
/// an attacker with write access to the log directory can swap the entry for a
/// symlink (a classic TOCTOU race). `O_NOFOLLOW` makes the open fail with
/// `ELOOP` in that case instead of silently reading whatever the link targets
/// (for example `/etc/shadow`), which would otherwise be copied verbatim into a
/// world-inspectable archive. The temp/archive path already refuses symlinks via
/// `create_tmp_archive`; this closes the same gap on the source side.
fn open_source_no_follow(path: &Path) -> std::io::Result<File> {
#[cfg(unix)]
{
use std::os::unix::fs::OpenOptionsExt;
std::fs::OpenOptions::new()
.read(true)
.custom_flags(libc::O_NOFOLLOW)
.open(path)
}
#[cfg(not(unix))]
{
File::open(path)
}
}
@@ -234,32 +237,21 @@ fn compress_with_writer<F>(
where
F: FnMut(&mut BufReader<File>, BufWriter<File>) -> Result<(u64, File), std::io::Error>,
{
// Idempotency: an existing *complete* archive means a previous pass already
// handled this file, so we can skip recompression and let the caller delete
// the source. But we must not trust just any entry at `archive_path`:
// * `Path::exists()` follows symlinks, so a planted `<archive>.gz` symlink
// would green-light deleting the source with no real archive;
// * a zero-length/truncated archive left by a crashed run (see OLC-01)
// would be trusted, then the source deleted — unrecoverable loss.
// Use symlink_metadata (no follow) and require a regular, non-empty file
// with a valid codec header before trusting it. Anything else falls through
// to recompression, whose atomic create_new+rename replaces the bad entry
// (a symlink is replaced, never followed or deleted through).
match std::fs::symlink_metadata(archive_path) {
Ok(meta) if meta.file_type().is_file() && meta.len() > 0 && archive_header_ok(archive_path, algorithm_used) => {
debug!(event = EVENT_LOG_CLEANER_COMPRESSION_STATE, component = LOG_COMPONENT_OBS, subsystem = LOG_SUBSYSTEM_LOG_CLEANER, file = ?archive_path, state = "archive_exists", "log cleaner compression state changed");
let input_bytes = std::fs::metadata(path).map(|m| m.len()).unwrap_or(0);
return Ok(CompressionOutput {
archive_path: archive_path.to_path_buf(),
algorithm_used,
input_bytes,
output_bytes: meta.len(),
});
}
Ok(_) => {
warn!(event = EVENT_LOG_CLEANER_COMPRESSION_STATE, component = LOG_COMPONENT_OBS, subsystem = LOG_SUBSYSTEM_LOG_CLEANER, file = ?archive_path, result = "archive_untrusted_recompressing", "log cleaner compression state changed");
}
Err(_) => {}
// Integrity over idempotency: never trust a pre-existing entry at
// `archive_path` as a valid prior result. The former fast path accepted any
// regular, non-empty file whose first 2-4 bytes matched the gzip/zstd magic.
// That header check does not prove the body is complete: an attacker with
// write access to the log directory (or a crashed prior run) can leave a file
// that starts with valid magic but is truncated/forged, and trusting it would
// let the caller delete the real source log — silent audit-data loss. Fully
// decoding every leftover to validate it would add real CPU cost and
// decode-bug surface, so instead we always recompress the source in this
// pass. The atomic create_new+rename below overwrites whatever sits at
// `archive_path` (a regular file or a planted symlink is replaced, never
// followed or deleted through) with a freshly written, fully-synced archive,
// so a partial or forged leftover can never be the basis for source deletion.
if std::fs::symlink_metadata(archive_path).is_ok() {
debug!(event = EVENT_LOG_CLEANER_COMPRESSION_STATE, component = LOG_COMPONENT_OBS, subsystem = LOG_SUBSYSTEM_LOG_CLEANER, file = ?archive_path, result = "leftover_archive_recompressing", "log cleaner compression state changed");
}
if dry_run {
@@ -281,7 +273,10 @@ where
// Create the output archive only after the dry-run short-circuit so this
// helper remains side-effect free when the caller is evaluating policy.
let input = File::open(path)?;
// O_NOFOLLOW: refuse to open a symlink swapped in at the source path between
// the scan and this open (TOCTOU), which could redirect the read to an
// arbitrary privileged file.
let input = open_source_no_follow(path)?;
// Read the source mode from the already-open fd (no extra path stat, no
// TOCTOU) so the temp file is created restrictive — never wider than the
@@ -385,3 +380,79 @@ fn archive_path(path: &Path, algorithm: CompressionAlgorithm) -> PathBuf {
file_name.push(format!(".{ext}"));
path.with_file_name(file_name)
}
#[cfg(test)]
mod tests {
use super::*;
fn default_options(algorithm: CompressionAlgorithm) -> CompressionOptions {
CompressionOptions {
algorithm,
gzip_level: 6,
zstd_level: 3,
zstd_workers: 1,
zstd_fallback_to_gzip: false,
dry_run: false,
}
}
/// A symlink planted at the source path must be refused (not followed) so a
/// TOCTOU swap cannot redirect compression to an arbitrary privileged file.
#[cfg(unix)]
#[test]
fn source_symlink_is_rejected() {
let dir = tempfile::tempdir().expect("tempdir");
// Secret the attacker wants exfiltrated into the (broader-readable) archive.
let secret = dir.path().join("secret.txt");
std::fs::write(&secret, b"top secret contents").expect("write secret");
// The scanner-visible log path is actually a symlink to the secret.
let source = dir.path().join("app.log");
std::os::unix::fs::symlink(&secret, &source).expect("symlink");
let err = compress_file(&source, &default_options(CompressionAlgorithm::Gzip))
.expect_err("compressing a symlinked source must fail");
// O_NOFOLLOW surfaces as ELOOP; accept any error but ensure no archive
// was produced from the symlink target.
assert!(
!dir.path().join("app.log.gz").exists(),
"no archive should be created from a symlinked source, err={err}"
);
// The secret must not have been copied anywhere in the directory.
assert!(secret.exists(), "secret should be untouched");
}
/// A pre-existing archive whose header magic is valid but whose body is
/// truncated/forged must not be trusted as a completed prior result: it must
/// be recompressed into a complete archive, and the source must survive.
#[test]
fn truncated_magic_archive_is_not_trusted() {
use std::io::Read;
let dir = tempfile::tempdir().expect("tempdir");
let source = dir.path().join("app.log");
let payload = b"line1\nline2\nline3\n";
std::fs::write(&source, payload).expect("write source");
// Forged archive: valid gzip magic (0x1f 0x8b) but a truncated body.
let archive = dir.path().join("app.log.gz");
std::fs::write(&archive, [0x1f_u8, 0x8b, 0x00]).expect("write forged archive");
let out = compress_file(&source, &default_options(CompressionAlgorithm::Gzip)).expect("compression should succeed");
assert_eq!(out.archive_path, archive);
// The forged stub must have been replaced by a real archive that decodes
// back to the full source — proving it was not trusted (trusting it would
// let the caller delete the source and lose the log).
let file = File::open(&archive).expect("open archive");
let mut decoder = flate2::read::GzDecoder::new(file);
let mut decoded = Vec::new();
decoder
.read_to_end(&mut decoded)
.expect("archive must be a complete gzip stream");
assert_eq!(decoded, payload, "recompressed archive must contain the full source");
// compress_file never deletes the source itself (the caller does, only
// after a valid archive), so the source log is still present here.
assert!(source.exists(), "source log must survive compression");
}
}