test(ecstore): pin the remaining io_uring fd-cache invariants (#1180) (#4739)

Close the three test-completeness items in rustfs/backlog#1180 that the earlier
hardening (rustfs/rustfs#4726, #4729) left unpinned; the other two of the five
(sharded cancel routing, bailout-handle error) already landed in rustfs/uring.

- rename_data end-to-end invalidation: drive the real `LocalDisk::rename_data`
  commit (non-inline part, so `invalidate_part_paths` is non-empty) and assert
  the destination part descriptor is dropped — not merely `rename_file`/`delete`.
  Both production `rename_data` call sites share this invalidation, so a
  "fix one copy, miss the other" regression is now caught. The test is
  non-vacuous: it seeds the cache, removes the on-disk data dir out of band (the
  cached fd keeps the old inode alive and clears the path for the directory
  rename `rename_data` performs), and asserts a read still returns the OLD bytes
  before the commit — which fails outright if the cache is off, so it cannot pass
  without a live cache.
- FD_CACHE_TTL backstop: an injected short TTL proves the cache self-evicts a
  descriptor with no explicit invalidation; a static check pins the 5s value.
- zero-length read bounds parity on the cache-HIT path: a `length == 0` read
  past EOF must be rejected identically to the miss path and StdBackend, pinning
  the #1173 fix against regression.

Refactors `FdCache::new` to delegate to a private `with_ttl(ttl)` helper so the
TTL backstop can be exercised with a short TTL instead of a multi-second wait.

Verified: `cargo test -p rustfs-ecstore` on Linux with real io_uring
(seccomp=unconfined, RLIMIT_NOFILE raised, RUSTFS_URING_TESTS_MUST_RUN=1 so a
degraded skip fails rather than passing vacuously).

Co-authored-by: heihutu <heihutu@gmail.com>

crates/ecstore/src/disk/local.rs

@@ -2795,10 +2795,17 @@ struct FdCache {
 #[cfg(target_os = "linux")]
 impl FdCache {
     fn new() -> Self {
+        Self::with_ttl(FD_CACHE_TTL)
+    }
+
+    /// Build a cache with an explicit TTL backstop. Production goes through
+    /// `new` (`FD_CACHE_TTL`); tests inject a short TTL to exercise the backstop
+    /// eviction (rustfs/backlog#1180) without a multi-second wait.
+    fn with_ttl(ttl: std::time::Duration) -> Self {
         Self {
             cache: moka::future::Cache::builder()
                 .max_capacity(FD_CACHE_CAPACITY)
-                .time_to_live(FD_CACHE_TTL)
+                .time_to_live(ttl)
                 // Required for `invalidate_entries_if`; without it that call
                 // fails at runtime instead of dropping stale descriptors.
                 .support_invalidation_closures()
@@ -13911,6 +13918,228 @@ mod test {
         .await;
     }
 
+    /// The real `rename_data` commit path — heal's write-then-commit, and the
+    /// two production copies at `LocalDisk::rename_data` — must invalidate the
+    /// destination part descriptors, not merely `rename_file`/`delete`
+    /// (backlog#1180 item 1). Heal reuses a version's `data_dir` and lands the
+    /// rebuilt shard on the SAME `<object>/<data_dir>/part.N` path, so a
+    /// descriptor cached before the commit would keep serving the pre-heal inode.
+    ///
+    /// Faithfulness: it drives `disk.rename_data(...)` for real (non-inline part,
+    /// so `invalidate_part_paths` is non-empty), not a raw `fs::rename` + manual
+    /// invalidate. Non-vacuity: after seeding the cache it removes the on-disk
+    /// data dir out of band — the cached fd keeps the old inode alive — and
+    /// asserts a read still returns the OLD bytes, which fails outright if the
+    /// cache is off, so the test cannot pass without a live cache. Removing the
+    /// dir also clears the path for the directory rename `rename_data` performs.
+    #[cfg(target_os = "linux")]
+    #[tokio::test(flavor = "multi_thread")]
+    async fn uring_fd_cache_is_invalidated_by_rename_data() {
+        use tempfile::tempdir;
+
+        let root_dir = tempdir().expect("operation should succeed");
+        temp_env::async_with_vars(
+            [
+                (ENV_RUSTFS_IO_URING_READ_ENABLE, Some("true")),
+                (ENV_RUSTFS_IO_URING_FD_CACHE, Some("true")),
+            ],
+            async {
+                let root = root_dir.path().to_path_buf();
+                // The invalidation only means anything when a descriptor is
+                // actually cached: io_uring must be available AND the fd cache on
+                // (RLIMIT_NOFILE headroom, backlog#1178). Probe the (existing)
+                // root to decide; otherwise there is nothing to exercise.
+                let cache_on = UringBackend::try_new(root.clone())
+                    .map(|b| b.fd_cache.is_some())
+                    .unwrap_or(false);
+                if !cache_on {
+                    uring_test_skip("uring_fd_cache_is_invalidated_by_rename_data");
+                    return;
+                }
+
+                let data_dir = Uuid::new_v4();
+                let version_id = Uuid::new_v4();
+                let part_rel = format!("object/{data_dir}/part.1");
+
+                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");
+                // `LocalDisk::new` already creates the system buckets, so the tmp
+                // bucket exists; `ensure_test_volume` tolerates that.
+                ensure_test_volume(&disk, "bucket").await;
+                ensure_test_volume(&disk, RUSTFS_META_TMP_BUCKET).await;
+
+                // 1. Seed the descriptor cache: write the destination part and read
+                //    it, caching an open fd to that (pre-heal) inode.
+                disk.write_all("bucket", &part_rel, Bytes::from_static(b"oldshard"))
+                    .await
+                    .expect("operation should succeed");
+                let got = disk.read_file_mmap_copy("bucket", &part_rel, 0, 8).await.expect("seed read");
+                assert_eq!(got, Bytes::from_static(b"oldshard"), "first read seeds the descriptor cache");
+
+                // 2. Remove the destination data dir out of band. The cached fd
+                //    keeps the old inode alive and readable, and clears the path so
+                //    the real rename_data below can move a fresh data dir onto it (a
+                //    directory rename cannot land on a non-empty dir). No LocalDisk
+                //    mutation runs here, so nothing invalidates the cache.
+                std::fs::remove_dir_all(root.join("bucket").join(format!("object/{data_dir}")))
+                    .expect("remove destination data dir out of band");
+
+                // Liveness gate (non-vacuity): the cache must still serve the
+                // removed inode. With the cache off this read would fail, so this
+                // assertion is what proves the descriptor is genuinely cached.
+                let stale = disk
+                    .read_file_mmap_copy("bucket", &part_rel, 0, 8)
+                    .await
+                    .expect("a cached descriptor must still read the removed inode");
+                assert_eq!(
+                    stale,
+                    Bytes::from_static(b"oldshard"),
+                    "a cached descriptor is expected to still see the pre-commit inode — proves the cache is live"
+                );
+
+                // 3. Stage the rebuilt shard in the tmp bucket under the SAME
+                //    data_dir (exactly as heal does) and commit it with the real
+                //    rename_data (non-inline: data=None, size>0, one part).
+                let src_object = "heal-src";
+                disk.write_all(
+                    RUSTFS_META_TMP_BUCKET,
+                    &format!("{src_object}/{data_dir}/part.1"),
+                    Bytes::from_static(b"newshard"),
+                )
+                .await
+                .expect("stage rebuilt shard");
+
+                let mut fi = test_file_info("object", version_id, Some(data_dir), None);
+                fi.size = 8;
+                fi.add_object_part(1, "etag".to_string(), 8, Some(OffsetDateTime::now_utc()), 8, None, None);
+                disk.rename_data(RUSTFS_META_TMP_BUCKET, src_object, fi, "bucket", "object")
+                    .await
+                    .expect("real rename_data must commit the rebuilt shard");
+
+                // 4. The read must now see the committed shard: rename_data
+                //    invalidated the cached descriptor for `{dst}/{data_dir}/part.N`.
+                let healed = disk
+                    .read_file_mmap_copy("bucket", &part_rel, 0, 8)
+                    .await
+                    .expect("post-commit read");
+                assert_eq!(
+                    healed,
+                    Bytes::from_static(b"newshard"),
+                    "rename_data must invalidate the cached descriptor so the committed shard is visible"
+                );
+            },
+        )
+        .await;
+    }
+
+    /// The TTL backstop (backlog#1178/#1180): `FdCache` must stop serving a
+    /// descriptor once `time_to_live` elapses even when no mutation path ever
+    /// calls an explicit `invalidate_*`. This is the safety net for a future
+    /// write path that forgets to invalidate — the stale descriptor self-evicts
+    /// rather than masking a replaced inode indefinitely. An injected short TTL
+    /// exercises the backstop without the production 5s wait; a static check
+    /// pins the production value so a change to it is a conscious edit.
+    #[cfg(target_os = "linux")]
+    #[tokio::test(flavor = "multi_thread")]
+    async fn fd_cache_ttl_evicts_without_explicit_invalidation() {
+        use std::fs::File;
+
+        assert_eq!(
+            FD_CACHE_TTL,
+            std::time::Duration::from_secs(5),
+            "the production TTL backstop value is pinned; update this assertion deliberately if it changes"
+        );
+
+        let dir = tempfile::tempdir().expect("operation should succeed");
+        let ttl = std::time::Duration::from_millis(200);
+        let cache = FdCache::with_ttl(ttl);
+        let key = FdKey {
+            volume: "v".into(),
+            path: "object/dd/part.1".into(),
+            direct: false,
+        };
+        let file = Arc::new(File::create(dir.path().join("f")).expect("operation should succeed"));
+        cache.insert(key.clone(), file).await;
+        assert!(cache.get(&key).await.is_some(), "a freshly inserted descriptor must be served");
+
+        // Well past the TTL (10x margin against scheduler jitter): the backstop
+        // must drop the descriptor with no explicit invalidation in between.
+        tokio::time::sleep(ttl * 10).await;
+        assert!(
+            cache.get(&key).await.is_none(),
+            "the TTL backstop must stop serving a descriptor once time_to_live elapses"
+        );
+        assert_eq!(cache.entry_count().await, 0, "the expired entry must be evicted, not merely hidden");
+    }
+
+    /// Zero-length read bounds parity on the cache-HIT path (backlog#1173/#1180).
+    /// A `length == 0` read past EOF must be rejected identically whether the
+    /// descriptor is freshly opened (miss path) or served from the cache: the
+    /// cache-hit branch fstats the descriptor to reproduce the miss path's
+    /// `offset > len` check instead of returning empty unconditionally. Seeds
+    /// the cache with a normal read so the zero-length reads are hits, then pins
+    /// that UringBackend and StdBackend agree on every case.
+    #[cfg(target_os = "linux")]
+    #[tokio::test(flavor = "multi_thread")]
+    async fn uring_zero_length_read_bounds_match_std_on_cache_hit() {
+        use tempfile::tempdir;
+
+        let root_dir = tempdir().expect("operation should succeed");
+        let root = root_dir.path().to_path_buf();
+        let Some(backend) = temp_env::with_vars(
+            [
+                (ENV_RUSTFS_IO_URING_READ_ENABLE, Some("true")),
+                (ENV_RUSTFS_IO_URING_FD_CACHE, Some("true")),
+            ],
+            || UringBackend::try_new(root.clone()),
+        ) else {
+            uring_test_skip("uring_zero_length_read_bounds_match_std_on_cache_hit");
+            return;
+        };
+        assert!(
+            backend.fd_cache.is_some(),
+            "the cache-hit branch is the point of this test; the cache must be on"
+        );
+
+        let volume = "bucket";
+        let object = "obj/dd/part.1";
+        std::fs::create_dir_all(root.join(volume).join("obj/dd")).expect("operation should succeed");
+        let content: &[u8] = b"exactly-fifteen"; // 15 bytes
+        std::fs::write(root.join(volume).join(object), content).expect("operation should succeed");
+        let len = content.len();
+
+        // A normal read seeds the descriptor cache, so the zero-length reads
+        // below take the cache-hit branch (3206) rather than the miss path.
+        let seed = backend.pread_bytes(volume, object, 0, len, None).await.expect("seed read");
+        assert_eq!(seed, Bytes::copy_from_slice(content));
+
+        // StdBackend over the same file is the parity oracle.
+        let std_backend = StdBackend::new(root.clone());
+        for offset in [0usize, len, len + 1] {
+            let uring = backend.pread_bytes(volume, object, offset, 0, None).await;
+            let std = std_backend.pread_bytes(volume, object, offset, 0, None).await;
+            if offset > len {
+                assert!(
+                    matches!(uring, Err(DiskError::FileCorrupt)),
+                    "zero-length read past EOF must be rejected on the cache-hit path (offset={offset}), got {uring:?}"
+                );
+                assert!(
+                    matches!(std, Err(DiskError::FileCorrupt)),
+                    "StdBackend oracle must reject the same zero-length read past EOF (offset={offset}), got {std:?}"
+                );
+            } else {
+                let u = uring.unwrap_or_else(|e| panic!("zero-length read at/inside EOF must succeed (offset={offset}): {e:?}"));
+                let s = std.unwrap_or_else(|e| {
+                    panic!("StdBackend zero-length read at/inside EOF must succeed (offset={offset}): {e:?}")
+                });
+                assert!(
+                    u.is_empty() && s.is_empty(),
+                    "zero-length read at/inside EOF must be empty (offset={offset})"
+                );
+            }
+        }
+    }
+
     /// Pages of `path` still resident in the page cache, via `mincore(2)`.
     /// `mincore` reports residency without faulting anything in, so measuring
     /// cannot perturb what it measures.