From 1553dc3f62a6376ab7489b2904c8392f93d2b866 Mon Sep 17 00:00:00 2001 From: houseme Date: Mon, 13 Jul 2026 00:03:28 +0800 Subject: [PATCH] Address P2 follow-ups from the 2026-07-10..12 merged-PR review (backlog#1210-1220) (#4783) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * 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 * 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 * 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 * 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 * 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 * 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 * 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 * 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 * 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 * 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 * 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 --------- Co-authored-by: heihutu --- .github/workflows/ci.yml | 18 ++ .../src/get_codec_streaming_compat_test.rs | 141 ++++++++++++++- .../ecstore/src/cache_value/metacache_set.rs | 31 ++++ crates/ecstore/src/disk/disk_store.rs | 12 ++ crates/ecstore/src/disk/local.rs | 161 +++++++++++++++-- .../src/set_disk/core/io_primitives.rs | 38 ++++ crates/io-metrics/src/collector.rs | 69 +++++++- crates/object-data-cache/src/memory.rs | 89 +++++++++- crates/object-data-cache/src/moka_backend.rs | 16 +- crates/obs/src/cleaner/compress.rs | 163 +++++++++++++----- 10 files changed, 663 insertions(+), 75 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index e7224f01e..0e3b38fd8 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -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: diff --git a/crates/e2e_test/src/get_codec_streaming_compat_test.rs b/crates/e2e_test/src/get_codec_streaming_compat_test.rs index 8447c69cf..74a610f57 100644 --- a/crates/e2e_test/src/get_codec_streaming_compat_test.rs +++ b/crates/e2e_test/src/get_codec_streaming_compat_test.rs @@ -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> { + 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("").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 = 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 = 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. diff --git a/crates/ecstore/src/cache_value/metacache_set.rs b/crates/ecstore/src/cache_value/metacache_set.rs index c19bd314e..0fdbfdf37 100644 --- a/crates/ecstore/src/cache_value/metacache_set.rs +++ b/crates/ecstore/src/cache_value/metacache_set.rs @@ -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({ diff --git a/crates/ecstore/src/disk/disk_store.rs b/crates/ecstore/src/disk/disk_store.rs index e2edaa7db..f3f98d4a1 100644 --- a/crates/ecstore/src/disk/disk_store.rs +++ b/crates/ecstore/src/disk/disk_store.rs @@ -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, diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index 475b31e97..71b817879 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -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, @@ -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, ¤t, -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> { 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 diff --git a/crates/ecstore/src/set_disk/core/io_primitives.rs b/crates/ecstore/src/set_disk/core/io_primitives.rs index 602acc31f..9c76defd6 100644 --- a/crates/ecstore/src/set_disk/core/io_primitives.rs +++ b/crates/ecstore/src/set_disk/core/io_primitives.rs @@ -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, diff --git a/crates/io-metrics/src/collector.rs b/crates/io-metrics/src/collector.rs index 3fe88c023..c1be960e6 100644 --- a/crates/io-metrics/src/collector.rs +++ b/crates/io-metrics/src/collector.rs @@ -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, /// 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()); diff --git a/crates/object-data-cache/src/memory.rs b/crates/object-data-cache/src/memory.rs index a8d89ffc1..1ec3874e2 100644 --- a/crates/object-data-cache/src/memory.rs +++ b/crates/object-data-cache/src/memory.rs @@ -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!( diff --git a/crates/object-data-cache/src/moka_backend.rs b/crates/object-data-cache/src/moka_backend.rs index 5fb6974e5..872cb6a90 100644 --- a/crates/object-data-cache/src/moka_backend.rs +++ b/crates/object-data-cache/src/moka_backend.rs @@ -32,6 +32,10 @@ pub struct MokaBackend { index: Arc, 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, @@ -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); } diff --git a/crates/obs/src/cleaner/compress.rs b/crates/obs/src/cleaner/compress.rs index b38f19391..9a4bee056 100644 --- a/crates/obs/src/cleaner/compress.rs +++ b/crates/obs/src/cleaner/compress.rs @@ -184,25 +184,28 @@ fn create_tmp_archive(path: &Path, source_mode: Option) -> 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 { + #[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( where F: FnMut(&mut BufReader, BufWriter) -> 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 `.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"); + } +}