fix the cause of the flaky test failure

The following interleaving could lead to the subsequent panic.
The fix is to retain the entry in the map if we can't lock the slot, and
let `find_victim`'s call to self.remove_mapping() remove the given
(file_id, block_no).

```
T1: find_victim up to right before the `self.remove_mapping(old_key);`  call
T2: `drop_buffers_for_immutable` for the EphemeralFile of `old_key.`
    => this removes the key from the mapping
T1: executes `self.remove_mapping(old_key)`, key already gone => panic
```

    ```

    ---- walingest::tests::test_large_rel stdout ----
    2023-08-28T10:23:07.361671Z  INFO skipping attempt to start flush_loop twice 0921208cfb20e4c120166d07667358e5/11223344556677881122334455667788
    thread 'walingest::tests::test_large_rel' panicked at 'could not find old key in mapping', pageserver/src/page_cache.rs:674:22
    stack backtrace:
       0: rust_begin_unwind
                 at /rustc/8ede3aae28fe6e4d52b38157d7bfe0d3bceef225/library/std/src/panicking.rs:593:5
       1: core::panicking::panic_fmt
                 at /rustc/8ede3aae28fe6e4d52b38157d7bfe0d3bceef225/library/core/src/panicking.rs:67:14
       2: core::panicking::panic_display
                 at /rustc/8ede3aae28fe6e4d52b38157d7bfe0d3bceef225/library/core/src/panicking.rs:150:5
       3: core::panicking::panic_str
                 at /rustc/8ede3aae28fe6e4d52b38157d7bfe0d3bceef225/library/core/src/panicking.rs:134:5
       4: core::option::expect_failed
                 at /rustc/8ede3aae28fe6e4d52b38157d7bfe0d3bceef225/library/core/src/option.rs:1932:5
       5: core::option::Option<T>::expect
                 at /rustc/8ede3aae28fe6e4d52b38157d7bfe0d3bceef225/library/core/src/option.rs:898:21
       6: pageserver::page_cache::PageCache::remove_mapping
                 at ./src/page_cache.rs:674:22
       7: pageserver::page_cache::PageCache::find_victim
                 at ./src/page_cache.rs:762:21
       8: pageserver::page_cache::PageCache::lock_for_read
                 at ./src/page_cache.rs:502:17
       9: pageserver::page_cache::PageCache::read_immutable_buf
                 at ./src/page_cache.rs:359:9
      10: pageserver::tenant::ephemeral_file::EphemeralFile::write_blob::{{closure}}::Writer::push_bytes::{{closure}}
                 at ./src/tenant/ephemeral_file.rs:136:39
      11: pageserver::tenant::ephemeral_file::EphemeralFile::write_blob::{{closure}}
                 at ./src/tenant/ephemeral_file.rs:197:35
      12: pageserver::tenant::storage_layer::inmemory_layer::InMemoryLayer::put_value::{{closure}}
                 at ./src/tenant/storage_layer/inmemory_layer.rs:278:41
      13: pageserver::tenant::timeline::Timeline::put_value::{{closure}}
                 at ./src/tenant/timeline.rs:2486:40
      14: pageserver::tenant::timeline::TimelineWriter::put::{{closure}}
                 at ./src/tenant/timeline.rs:4618:44
      15: pageserver::pgdatadir_mapping::DatadirModification::commit::{{closure}}
                 at ./src/pgdatadir_mapping.rs:1184:42
      16: pageserver::walingest::tests::test_large_rel::{{closure}}
                 at ./src/walingest.rs:1661:24
    ```

pageserver/ctl/src/layer_map_analyzer.rs

@@ -99,6 +99,7 @@ async fn get_holes(path: &Path, max_holes: usize) -> Result<Vec<Hole>> {
     let file = FileBlockReader::new(VirtualFile::open(path)?);
     let summary_blk = file.read_blk(0)?;
     let actual_summary = Summary::des_prefix(summary_blk.as_ref())?;
+    drop(summary_blk);
     let tree_reader = DiskBtreeReader::<_, DELTA_KEY_SIZE>::new(
         actual_summary.index_start_blk,
         actual_summary.index_root_blk,

pageserver/src/page_cache.rs

@@ -77,7 +77,7 @@ use std::{
     convert::TryInto,
     sync::{
         atomic::{AtomicU64, AtomicU8, AtomicUsize, Ordering},
-        RwLock, RwLockReadGuard, RwLockWriteGuard, TryLockError,
+        RwLock, RwLockReadGuard, RwLockWriteGuard, TryLockError, TryLockResult,
     },
 };
 
@@ -427,23 +427,49 @@ impl PageCache {
 
     /// Immediately drop all buffers belonging to given file
     pub fn drop_buffers_for_immutable(&self, drop_file_id: FileId) {
-        for slot_idx in 0..self.slots.len() {
-            let slot = &self.slots[slot_idx];
-
-            let mut inner = slot.inner.write().unwrap();
-            if let Some(key) = &inner.key {
-                match key {
-                    CacheKey::ImmutableFilePage { file_id, blkno: _ }
-                        if *file_id == drop_file_id =>
-                    {
-                        // remove mapping for old buffer
-                        self.remove_mapping(key);
+        let mut map = self.immutable_page_map.write().unwrap();
+        map.retain(|(file_id, block_no), slot_idx| {
+            if *file_id != drop_file_id {
+                return true;
+            }
+            let expect_cache_key = CacheKey::ImmutableFilePage {
+                file_id: drop_file_id,
+                blkno: *block_no,
+            };
+            let slot = &self.slots[*slot_idx];
+            match slot.inner.try_write() {
+                TryLockResult::Ok(mut inner) => {
+                    // check again, could have been taken since added to immutable_page_map
+                    if inner.key == Some(expect_cache_key) {
+                        // immutable_page_map was still in sync with reality
+                        // TODO: find a way to share code with `remove_mapping`
+                        self.size_metrics.current_bytes_immutable.sub_page_sz(1);
                         inner.key = None;
                     }
-                    _ => {}
+                    return false;
+                }
+                TryLockResult::Err(e @ TryLockError::Poisoned(_)) => {
+                    panic!("slot lock {slot_idx} poisoned: {e:?}")
+                }
+                TryLockResult::Err(TryLockError::WouldBlock) => {
+                    // This function is only called from the `Drop` impl of EphemeralFile.
+                    // So, there shouldn't be any page cache users that are reading from
+                    // that EphemeralFile anymore, because,
+                    // 1. EphemeralFile doesn't hand out page cache read guards and
+                    // 2. there can't be any concurrent `EphemeralFile::reads`s because it's being dropped.
+                    //
+                    // So, the only reason why the page is locked right now is `find_victim`
+                    // trying to claim this page.
+                    //
+                    // Avoid waiting and keep the page in the cache.
+                    // The `find_victim` will free up the slot.
+                    // Also, it will call `remove_mapping()` to remove the (file_id, block_no)
+                    // from the immutable_page_map; so, keep  the entry in the map here, otherwise
+                    // the `remove_mapping()` call will panic.
+                    return true;
                 }
             }
-        }
+        });
     }
 
     //

pageserver/src/tenant/block_io.rs

@@ -36,7 +36,7 @@ where
 
 /// Reference to an in-memory copy of an immutable on-disk block.
 pub enum BlockLease<'a> {
-    PageReadGuard(PageReadGuard<'static>),
+    PageReadGuard(PageReadGuard<'a>),
     EphemeralFileMutableTail(&'a [u8; PAGE_SZ]),
     #[cfg(test)]
     Rc(std::rc::Rc<[u8; PAGE_SZ]>),

pageserver/src/tenant/storage_layer/delta_layer.rs

@@ -848,6 +848,7 @@ impl DeltaLayerInner {
 
         let summary_blk = file.read_blk(0)?;
         let actual_summary = Summary::des_prefix(summary_blk.as_ref())?;
+        drop(summary_blk);
 
         if let Some(mut expected_summary) = summary {
             // production code path

pageserver/src/tenant/storage_layer/image_layer.rs

@@ -442,6 +442,7 @@ impl ImageLayerInner {
         let file = FileBlockReader::new(file);
         let summary_blk = file.read_blk(0)?;
         let actual_summary = Summary::des_prefix(summary_blk.as_ref())?;
+        drop(summary_blk);
 
         if let Some(mut expected_summary) = summary {
             // production code path