Simplify PR template

.github/pull_request_template.md

@@ -1,14 +1,3 @@
 ## Problem
 
 ## Summary of changes
-
-## Checklist before requesting a review
-
-- [ ] I have performed a self-review of my code.
-- [ ] If it is a core feature, I have added thorough tests.
-- [ ] Do we need to implement analytics? if so did you add the relevant metrics to the dashboard?
-- [ ] If this PR requires public announcement, mark it with /release-notes label and add several sentences in this section.
-
-## Checklist before merging
-
-- [ ] Do not forget to reformat commit message to not include the above checklist

.github/workflows/release.yml

@@ -2,7 +2,7 @@ name: Create Release Branch
 
 on:
   schedule:
-    - cron: '0 7 * * 5'
+    - cron: '0 7 * * 2'
   workflow_dispatch:
 
 jobs:

Cargo.lock

@@ -1609,6 +1609,16 @@ dependencies = [
  "subtle",
 ]
 
+[[package]]
+name = "ctor"
+version = "0.1.26"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6d2301688392eb071b0bf1a37be05c469d3cc4dbbd95df672fe28ab021e6a096"
+dependencies = [
+ "quote",
+ "syn 1.0.109",
+]
+
 [[package]]
 name = "ctr"
 version = "0.6.0"
@@ -2704,10 +2714,11 @@ dependencies = [
 
 [[package]]
 name = "log"
-version = "0.4.20"
+version = "0.4.17"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b5e6163cb8c49088c2c36f57875e58ccd8c87c7427f7fbd50ea6710b2f3f2e8f"
+checksum = "abb12e687cfb44aa40f41fc3978ef76448f9b6038cad6aef4259d3c095a2382e"
 dependencies = [
+ "cfg-if",
  "value-bag",
 ]
 
@@ -6000,9 +6011,13 @@ checksum = "830b7e5d4d90034032940e4ace0d9a9a057e7a45cd94e6c007832e39edb82f6d"
 
 [[package]]
 name = "value-bag"
-version = "1.4.2"
+version = "1.0.0-alpha.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4a72e1902dde2bd6441347de2b70b7f5d59bf157c6c62f0c44572607a1d55bbe"
+checksum = "2209b78d1249f7e6f3293657c9779fe31ced465df091bbd433a1cf88e916ec55"
+dependencies = [
+ "ctor",
+ "version_check",
+]
 
 [[package]]
 name = "vcpkg"

compute_tools/src/bin/compute_ctl.rs

@@ -278,9 +278,8 @@ fn main() -> Result<()> {
         if #[cfg(target_os = "linux")] {
             use std::env;
             use tokio_util::sync::CancellationToken;
-            let vm_monitor_addr = matches
-                .get_one::<String>("vm-monitor-addr")
-                .expect("--vm-monitor-addr should always be set because it has a default arg");
+            use tracing::warn;
+            let vm_monitor_addr = matches.get_one::<String>("vm-monitor-addr");
             let file_cache_connstr = matches.get_one::<String>("filecache-connstr");
             let cgroup = matches.get_one::<String>("cgroup");
             let file_cache_on_disk = matches.get_flag("file-cache-on-disk");
@@ -289,16 +288,22 @@ fn main() -> Result<()> {
             // Note: it seems like you can make a runtime in an inner scope and
             // if you start a task in it it won't be dropped. However, make it
             // in the outermost scope just to be safe.
-            let rt = if env::var_os("AUTOSCALING").is_some() {
-                Some(
+            let rt = match (env::var_os("AUTOSCALING"), vm_monitor_addr) {
+                (None, None) => None,
+                (None, Some(_)) => {
+                    warn!("--vm-monitor-addr option set but AUTOSCALING env var not present");
+                    None
+                }
+                (Some(_), None) => {
+                    panic!("AUTOSCALING env var present but --vm-monitor-addr option not set")
+                }
+                (Some(_), Some(_)) => Some(
                     tokio::runtime::Builder::new_multi_thread()
                         .worker_threads(4)
                         .enable_all()
                         .build()
-                        .expect("failed to create tokio runtime for monitor")
-                )
-            } else {
-                None
+                        .expect("failed to create tokio runtime for monitor"),
+                ),
             };
 
             // This token is used internally by the monitor to clean up all threads
@@ -309,7 +314,7 @@ fn main() -> Result<()> {
                     Box::leak(Box::new(vm_monitor::Args {
                         cgroup: cgroup.cloned(),
                         pgconnstr: file_cache_connstr.cloned(),
-                        addr: vm_monitor_addr.clone(),
+                        addr: vm_monitor_addr.cloned().unwrap(),
                         file_cache_on_disk,
                     })),
                     token.clone(),

compute_tools/src/spec.rs

@@ -24,7 +24,7 @@ fn do_control_plane_request(
 ) -> Result<ControlPlaneSpecResponse, (bool, String)> {
     let resp = reqwest::blocking::Client::new()
         .get(uri)
-        .header("Authorization", format!("Bearer {}", jwt))
+        .header("Authorization", jwt)
         .send()
         .map_err(|e| {
             (

control_plane/src/bin/attachment_service.rs

@@ -12,7 +12,6 @@ use hyper::{Body, Request, Response};
 use serde::{Deserialize, Serialize};
 use std::path::{Path, PathBuf};
 use std::{collections::HashMap, sync::Arc};
-use utils::http::endpoint::request_span;
 use utils::logging::{self, LogFormat};
 use utils::signals::{ShutdownSignals, Signal};
 
@@ -222,25 +221,8 @@ async fn handle_attach_hook(mut req: Request<Body>) -> Result<Response<Body>, Ap
             generation: 0,
         });
 
-    if let Some(attaching_pageserver) = attach_req.node_id.as_ref() {
+    if attach_req.node_id.is_some() {
         tenant_state.generation += 1;
-        tracing::info!(
-            tenant_id = %attach_req.tenant_id,
-            ps_id = %attaching_pageserver,
-            generation = %tenant_state.generation,
-            "issuing",
-        );
-    } else if let Some(ps_id) = tenant_state.pageserver {
-        tracing::info!(
-            tenant_id = %attach_req.tenant_id,
-            %ps_id,
-            generation = %tenant_state.generation,
-            "dropping",
-        );
-    } else {
-        tracing::info!(
-            tenant_id = %attach_req.tenant_id,
-            "no-op: tenant already has no pageserver");
     }
     tenant_state.pageserver = attach_req.node_id;
     let generation = tenant_state.generation;
@@ -258,9 +240,9 @@ async fn handle_attach_hook(mut req: Request<Body>) -> Result<Response<Body>, Ap
 fn make_router(persistent_state: PersistentState) -> RouterBuilder<hyper::Body, ApiError> {
     endpoint::make_router()
         .data(Arc::new(State::new(persistent_state)))
-        .post("/re-attach", |r| request_span(r, handle_re_attach))
-        .post("/validate", |r| request_span(r, handle_validate))
-        .post("/attach-hook", |r| request_span(r, handle_attach_hook))
+        .post("/re-attach", handle_re_attach)
+        .post("/validate", handle_validate)
+        .post("/attach-hook", handle_attach_hook)
 }
 
 #[tokio::main]

docs/updating-postgres.md

@@ -1,108 +0,0 @@
-# Updating Postgres
-
-## Minor Versions
-
-When upgrading to a new minor version of Postgres, please follow these steps:
-
-_Example: 15.4 is the new minor version to upgrade to from 15.3._
-
-1. Clone the Neon Postgres repository if you have not done so already.
-
-    ```shell
-    git clone git@github.com:neondatabase/postgres.git
-    ```
-
-1. Add the Postgres upstream remote.
-
-    ```shell
-    git remote add upstream https://git.postgresql.org/git/postgresql.git
-    ```
-
-1. Create a new branch based on the stable branch you are updating.
-
-    ```shell
-    git checkout -b my-branch REL_15_STABLE_neon
-    ```
-
-1. Tag the last commit on the stable branch you are updating.
-
-    ```shell
-    git tag REL_15_3_neon
-    ```
-
-1. Push the new tag to the Neon Postgres repository.
-
-    ```shell
-    git push origin REL_15_3_neon
-    ```
-
-1. Find the release tags you're looking for. They are of the form `REL_X_Y`.
-
-1. Rebase the branch you created on the tag and resolve any conflicts.
-
-    ```shell
-    git fetch upstream REL_15_4
-    git rebase REL_15_4
-    ```
-
-1. Run the Postgres test suite to make sure our commits have not affected
-Postgres in a negative way.
-
-    ```shell
-    make check
-    # OR
-    meson test -C builddir
-    ```
-
-1. Push your branch to the Neon Postgres repository.
-
-    ```shell
-    git push origin my-branch
-    ```
-
-1. Clone the Neon repository if you have not done so already.
-
-    ```shell
-    git clone git@github.com:neondatabase/neon.git
-    ```
-
-1. Create a new branch.
-
-1. Change the `revisions.json` file to point at the HEAD of your Postgres
-branch.
-
-1. Update the Git submodule.
-
-    ```shell
-    git submodule set-branch --branch my-branch vendor/postgres-v15
-    git submodule update --remote vendor/postgres-v15
-    ```
-
-1. Run the Neon test suite to make sure that Neon is still good to go on this
-minor Postgres release.
-
-    ```shell
-    ./scripts/poetry -k pg15
-    ```
-
-1. Commit your changes.
-
-1. Create a pull request, and wait for CI to go green.
-
-1. Force push the rebased Postgres branches into the Neon Postgres repository.
-
-    ```shell
-    git push --force origin my-branch:REL_15_STABLE_neon
-    ```
-
-    It may require disabling various branch protections.
-
-1. Update your Neon PR to point at the branches.
-
-    ```shell
-    git submodule set-branch --branch REL_15_STABLE_neon vendor/postgres-v15
-    git commit --amend --no-edit
-    git push --force origin
-    ```
-
-1. Merge the pull request after getting approval(s) and CI completion.

libs/postgres_ffi/wal_craft/src/lib.rs

@@ -14,7 +14,6 @@ macro_rules! xlog_utils_test {
     ($version:ident) => {
         #[path = "."]
         mod $version {
-            #[allow(unused_imports)]
             pub use postgres_ffi::$version::wal_craft_test_export::*;
             #[allow(clippy::duplicate_mod)]
             #[cfg(test)]

libs/utils/src/sync/heavier_once_cell.rs

@@ -60,8 +60,8 @@ impl<T> OnceCell<T> {
     /// Initialization is panic-safe and cancellation-safe.
     pub async fn get_or_init<F, Fut, E>(&self, factory: F) -> Result<Guard<'_, T>, E>
     where
-        F: FnOnce(InitPermit) -> Fut,
-        Fut: std::future::Future<Output = Result<(T, InitPermit), E>>,
+        F: FnOnce() -> Fut,
+        Fut: std::future::Future<Output = Result<T, E>>,
     {
         let sem = {
             let guard = self.inner.lock().unwrap();
@@ -72,55 +72,28 @@ impl<T> OnceCell<T> {
         };
 
         let permit = sem.acquire_owned().await;
+        if permit.is_err() {
+            let guard = self.inner.lock().unwrap();
+            assert!(
+                guard.value.is_some(),
+                "semaphore got closed, must be initialized"
+            );
+            return Ok(Guard(guard));
+        } else {
+            // now we try
+            let value = factory().await?;
 
-        match permit {
-            Ok(permit) => {
-                let permit = InitPermit(permit);
-                let (value, _permit) = factory(permit).await?;
-
-                let guard = self.inner.lock().unwrap();
-
-                Ok(Self::set0(value, guard))
-            }
-            Err(_closed) => {
-                let guard = self.inner.lock().unwrap();
-                assert!(
-                    guard.value.is_some(),
-                    "semaphore got closed, must be initialized"
-                );
-                return Ok(Guard(guard));
-            }
+            let mut guard = self.inner.lock().unwrap();
+            assert!(
+                guard.value.is_none(),
+                "we won permit, must not be initialized"
+            );
+            guard.value = Some(value);
+            guard.init_semaphore.close();
+            Ok(Guard(guard))
         }
     }
 
-    /// Assuming a permit is held after previous call to [`Guard::take_and_deinit`], it can be used
-    /// to complete initializing the inner value.
-    ///
-    /// # Panics
-    ///
-    /// If the inner has already been initialized.
-    pub fn set(&self, value: T, _permit: InitPermit) -> Guard<'_, T> {
-        // cannot assert that this permit is for self.inner.semaphore
-        let guard = self.inner.lock().unwrap();
-
-        if guard.init_semaphore.try_acquire().is_ok() {
-            drop(guard);
-            panic!("semaphore is of wrong origin");
-        }
-
-        Self::set0(value, guard)
-    }
-
-    fn set0(value: T, mut guard: std::sync::MutexGuard<'_, Inner<T>>) -> Guard<'_, T> {
-        if guard.value.is_some() {
-            drop(guard);
-            unreachable!("we won permit, must not be initialized");
-        }
-        guard.value = Some(value);
-        guard.init_semaphore.close();
-        Guard(guard)
-    }
-
     /// Returns a guard to an existing initialized value, if any.
     pub fn get(&self) -> Option<Guard<'_, T>> {
         let guard = self.inner.lock().unwrap();
@@ -162,7 +135,7 @@ impl<'a, T> Guard<'a, T> {
     ///
     /// The permit will be on a semaphore part of the new internal value, and any following
     /// [`OnceCell::get_or_init`] will wait on it to complete.
-    pub fn take_and_deinit(&mut self) -> (T, InitPermit) {
+    pub fn take_and_deinit(&mut self) -> (T, tokio::sync::OwnedSemaphorePermit) {
         let mut swapped = Inner::default();
         let permit = swapped
             .init_semaphore
@@ -172,14 +145,11 @@ impl<'a, T> Guard<'a, T> {
         std::mem::swap(&mut *self.0, &mut swapped);
         swapped
             .value
-            .map(|v| (v, InitPermit(permit)))
+            .map(|v| (v, permit))
             .expect("guard is not created unless value has been initialized")
     }
 }
 
-/// Type held by OnceCell (de)initializing task.
-pub struct InitPermit(tokio::sync::OwnedSemaphorePermit);
-
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -215,11 +185,11 @@ mod tests {
                     barrier.wait().await;
                     let won = {
                         let g = cell
-                            .get_or_init(|permit| {
+                            .get_or_init(|| {
                                 counters.factory_got_to_run.fetch_add(1, Ordering::Relaxed);
                                 async {
                                     counters.future_polled.fetch_add(1, Ordering::Relaxed);
-                                    Ok::<_, Infallible>((i, permit))
+                                    Ok::<_, Infallible>(i)
                                 }
                             })
                             .await
@@ -273,7 +243,7 @@ mod tests {
         deinitialization_started.wait().await;
 
         let started_at = tokio::time::Instant::now();
-        cell.get_or_init(|permit| async { Ok::<_, Infallible>((reinit, permit)) })
+        cell.get_or_init(|| async { Ok::<_, Infallible>(reinit) })
             .await
             .unwrap();
 
@@ -288,32 +258,18 @@ mod tests {
         assert_eq!(*cell.get().unwrap(), reinit);
     }
 
-    #[test]
-    fn reinit_with_deinit_permit() {
-        let cell = Arc::new(OnceCell::new(42));
-
-        let (mol, permit) = cell.get().unwrap().take_and_deinit();
-        cell.set(5, permit);
-        assert_eq!(*cell.get().unwrap(), 5);
-
-        let (five, permit) = cell.get().unwrap().take_and_deinit();
-        assert_eq!(5, five);
-        cell.set(mol, permit);
-        assert_eq!(*cell.get().unwrap(), 42);
-    }
-
     #[tokio::test]
     async fn initialization_attemptable_until_ok() {
         let cell = OnceCell::default();
 
         for _ in 0..10 {
-            cell.get_or_init(|_permit| async { Err("whatever error") })
+            cell.get_or_init(|| async { Err("whatever error") })
                 .await
                 .unwrap_err();
         }
 
         let g = cell
-            .get_or_init(|permit| async { Ok::<_, Infallible>(("finally success", permit)) })
+            .get_or_init(|| async { Ok::<_, Infallible>("finally success") })
             .await
             .unwrap();
         assert_eq!(*g, "finally success");
@@ -325,11 +281,11 @@ mod tests {
 
         let barrier = tokio::sync::Barrier::new(2);
 
-        let initializer = cell.get_or_init(|permit| async {
+        let initializer = cell.get_or_init(|| async {
             barrier.wait().await;
             futures::future::pending::<()>().await;
 
-            Ok::<_, Infallible>(("never reached", permit))
+            Ok::<_, Infallible>("never reached")
         });
 
         tokio::select! {
@@ -342,7 +298,7 @@ mod tests {
         assert!(cell.get().is_none());
 
         let g = cell
-            .get_or_init(|permit| async { Ok::<_, Infallible>(("now initialized", permit)) })
+            .get_or_init(|| async { Ok::<_, Infallible>("now initialized") })
             .await
             .unwrap();
         assert_eq!(*g, "now initialized");

pageserver/src/control_plane_client.rs

@@ -57,10 +57,7 @@ impl ControlPlaneClient {
 
         if let Some(jwt) = &conf.control_plane_api_token {
             let mut headers = hyper::HeaderMap::new();
-            headers.insert(
-                "Authorization",
-                format!("Bearer {}", jwt.get_contents()).parse().unwrap(),
-            );
+            headers.insert("Authorization", jwt.get_contents().parse().unwrap());
             client = client.default_headers(headers);
         }
 

pageserver/src/deletion_queue.rs

@@ -10,7 +10,6 @@ use crate::control_plane_client::ControlPlaneGenerationsApi;
 use crate::metrics;
 use crate::tenant::remote_timeline_client::remote_layer_path;
 use crate::tenant::remote_timeline_client::remote_timeline_path;
-use crate::virtual_file::MaybeFatalIo;
 use crate::virtual_file::VirtualFile;
 use anyhow::Context;
 use camino::Utf8PathBuf;
@@ -272,9 +271,7 @@ impl DeletionHeader {
         let temp_path = path_with_suffix_extension(&header_path, TEMP_SUFFIX);
         VirtualFile::crashsafe_overwrite(&header_path, &temp_path, &header_bytes)
             .await
-            .maybe_fatal_err("save deletion header")?;
-
-        Ok(())
+            .map_err(Into::into)
     }
 }
 
@@ -363,7 +360,6 @@ impl DeletionList {
         let bytes = serde_json::to_vec(self).expect("Failed to serialize deletion list");
         VirtualFile::crashsafe_overwrite(&path, &temp_path, &bytes)
             .await
-            .maybe_fatal_err("save deletion list")
             .map_err(Into::into)
     }
 }

pageserver/src/deletion_queue/list_writer.rs

@@ -34,8 +34,6 @@ use crate::deletion_queue::TEMP_SUFFIX;
 use crate::metrics;
 use crate::tenant::remote_timeline_client::remote_layer_path;
 use crate::tenant::storage_layer::LayerFileName;
-use crate::virtual_file::on_fatal_io_error;
-use crate::virtual_file::MaybeFatalIo;
 
 // The number of keys in a DeletionList before we will proactively persist it
 // (without reaching a flush deadline).  This aims to deliver objects of the order
@@ -197,7 +195,7 @@ impl ListWriter {
                     debug!("Deletion header {header_path} not found, first start?");
                     Ok(None)
                 } else {
-                    on_fatal_io_error(&e, "reading deletion header");
+                    Err(anyhow::anyhow!(e))
                 }
             }
         }
@@ -218,9 +216,16 @@ impl ListWriter {
         self.pending.sequence = validated_sequence + 1;
 
         let deletion_directory = self.conf.deletion_prefix();
-        let mut dir = tokio::fs::read_dir(&deletion_directory)
-            .await
-            .fatal_err("read deletion directory");
+        let mut dir = match tokio::fs::read_dir(&deletion_directory).await {
+            Ok(d) => d,
+            Err(e) => {
+                warn!("Failed to open deletion list directory {deletion_directory}: {e:#}");
+
+                // Give up: if we can't read the deletion list directory, we probably can't
+                // write lists into it later, so the queue won't work.
+                return Err(e.into());
+            }
+        };
 
         let list_name_pattern =
             Regex::new("(?<sequence>[a-zA-Z0-9]{16})-(?<version>[a-zA-Z0-9]{2}).list").unwrap();
@@ -228,7 +233,7 @@ impl ListWriter {
         let temp_extension = format!(".{TEMP_SUFFIX}");
         let header_path = self.conf.deletion_header_path();
         let mut seqs: Vec<u64> = Vec::new();
-        while let Some(dentry) = dir.next_entry().await.fatal_err("read deletion dentry") {
+        while let Some(dentry) = dir.next_entry().await? {
             let file_name = dentry.file_name();
             let dentry_str = file_name.to_string_lossy();
 
@@ -241,9 +246,11 @@ impl ListWriter {
                 info!("Cleaning up temporary file {dentry_str}");
                 let absolute_path =
                     deletion_directory.join(dentry.file_name().to_str().expect("non-Unicode path"));
-                tokio::fs::remove_file(&absolute_path)
-                    .await
-                    .fatal_err("delete temp file");
+                if let Err(e) = tokio::fs::remove_file(&absolute_path).await {
+                    // Non-fatal error: we will just leave the file behind but not
+                    // try and load it.
+                    warn!("Failed to clean up temporary file {absolute_path}: {e:#}");
+                }
 
                 continue;
             }
@@ -283,9 +290,7 @@ impl ListWriter {
         for s in seqs {
             let list_path = self.conf.deletion_list_path(s);
 
-            let list_bytes = tokio::fs::read(&list_path)
-                .await
-                .fatal_err("read deletion list");
+            let list_bytes = tokio::fs::read(&list_path).await?;
 
             let mut deletion_list = match serde_json::from_slice::<DeletionList>(&list_bytes) {
                 Ok(l) => l,

pageserver/src/deletion_queue/validator.rs

@@ -28,7 +28,6 @@ use crate::config::PageServerConf;
 use crate::control_plane_client::ControlPlaneGenerationsApi;
 use crate::control_plane_client::RetryForeverError;
 use crate::metrics;
-use crate::virtual_file::MaybeFatalIo;
 
 use super::deleter::DeleterMessage;
 use super::DeletionHeader;
@@ -288,9 +287,16 @@ where
     async fn cleanup_lists(&mut self, list_paths: Vec<Utf8PathBuf>) {
         for list_path in list_paths {
             debug!("Removing deletion list {list_path}");
-            tokio::fs::remove_file(&list_path)
-                .await
-                .fatal_err("remove deletion list");
+
+            if let Err(e) = tokio::fs::remove_file(&list_path).await {
+                // Unexpected: we should have permissions and nothing else should
+                // be touching these files.  We will leave the file behind.  Subsequent
+                // pageservers will try and load it again: hopefully whatever storage
+                // issue (probably permissions) has been fixed by then.
+                tracing::error!("Failed to delete {list_path}: {e:#}");
+                metrics::DELETION_QUEUE.unexpected_errors.inc();
+                break;
+            }
         }
     }
 

pageserver/src/http/openapi_spec.yml

@@ -569,17 +569,7 @@ paths:
               schema:
                 $ref: "#/components/schemas/NotFoundError"
         "409":
-          description: |
-            The tenant is already known to Pageserver in some way,
-            and hence this `/attach` call has been rejected.
-
-            Some examples of how this can happen:
-            - tenant was created on this pageserver
-            - tenant attachment was started by an earlier call to `/attach`.
-
-            Callers should poll the tenant status's `attachment_status` field,
-            like for status 202. See the longer description for `POST /attach`
-            for details.
+          description: Tenant download is already in progress
           content:
             application/json:
               schema:

pageserver/src/repository.rs

@@ -28,9 +28,8 @@ impl Key {
     /// As long as Neon does not support tablespace (because of lack of access to local file system),
     /// we can assume that only some predefined namespace OIDs are used which can fit in u16
     pub fn to_i128(&self) -> i128 {
-        assert!(self.field1 < 0xF);
-        assert!(self.field2 < 0xFFFF);
-        (((self.field1 & 0xF) as i128) << 120)
+        assert!(self.field2 < 0xFFFF || self.field2 == 0xFFFFFFFF || self.field2 == 0x22222222);
+        (((self.field1 & 0xf) as i128) << 120)
             | (((self.field2 & 0xFFFF) as i128) << 104)
             | ((self.field3 as i128) << 72)
             | ((self.field4 as i128) << 40)
@@ -150,8 +149,8 @@ impl Key {
         field6: u32::MIN,
     };
     pub const MAX: Key = Key {
-        field1: 0xF - 1,
-        field2: 0xFFFF - 1,
+        field1: u8::MAX,
+        field2: u32::MAX,
         field3: u32::MAX,
         field4: u32::MAX,
         field5: u8::MAX,

pageserver/src/tenant.rs

@@ -3692,7 +3692,7 @@ mod tests {
     use tokio_util::sync::CancellationToken;
 
     static TEST_KEY: Lazy<Key> =
-        Lazy::new(|| Key::from_slice(&hex!("010000000033333333444444445500000001")));
+        Lazy::new(|| Key::from_slice(&hex!("112222222233333333444444445500000001")));
 
     #[tokio::test]
     async fn test_basic() -> anyhow::Result<()> {
@@ -3788,9 +3788,9 @@ mod tests {
         let writer = tline.writer().await;
 
         #[allow(non_snake_case)]
-        let TEST_KEY_A: Key = Key::from_hex("110000000033333333444444445500000001").unwrap();
+        let TEST_KEY_A: Key = Key::from_hex("112222222233333333444444445500000001").unwrap();
         #[allow(non_snake_case)]
-        let TEST_KEY_B: Key = Key::from_hex("110000000033333333444444445500000002").unwrap();
+        let TEST_KEY_B: Key = Key::from_hex("112222222233333333444444445500000002").unwrap();
 
         // Insert a value on the timeline
         writer
@@ -4374,7 +4374,7 @@ mod tests {
 
         let mut keyspace = KeySpaceAccum::new();
 
-        let mut test_key = Key::from_hex("010000000033333333444444445500000000").unwrap();
+        let mut test_key = Key::from_hex("012222222233333333444444445500000000").unwrap();
         let mut blknum = 0;
         for _ in 0..50 {
             for _ in 0..10000 {
@@ -4420,7 +4420,7 @@ mod tests {
 
         const NUM_KEYS: usize = 1000;
 
-        let mut test_key = Key::from_hex("010000000033333333444444445500000000").unwrap();
+        let mut test_key = Key::from_hex("012222222233333333444444445500000000").unwrap();
 
         let mut keyspace = KeySpaceAccum::new();
 
@@ -4501,7 +4501,7 @@ mod tests {
 
         const NUM_KEYS: usize = 1000;
 
-        let mut test_key = Key::from_hex("010000000033333333444444445500000000").unwrap();
+        let mut test_key = Key::from_hex("012222222233333333444444445500000000").unwrap();
 
         let mut keyspace = KeySpaceAccum::new();
 
@@ -4592,7 +4592,7 @@ mod tests {
         const NUM_KEYS: usize = 100;
         const NUM_TLINES: usize = 50;
 
-        let mut test_key = Key::from_hex("010000000033333333444444445500000000").unwrap();
+        let mut test_key = Key::from_hex("012222222233333333444444445500000000").unwrap();
         // Track page mutation lsns across different timelines.
         let mut updated = [[Lsn(0); NUM_KEYS]; NUM_TLINES];
 

pageserver/src/tenant/storage_layer/inmemory_layer.rs

@@ -345,19 +345,14 @@ impl InMemoryLayer {
 
         let cursor = inner.file.block_cursor();
 
-        // Sort the keys because delta layer writer expects them sorted.
-        //
-        // NOTE: this sort can take up significant time if the layer has millions of
-        //       keys. To speed up all the comparisons we convert the key to i128 and
-        //       keep the value as a reference.
-        let mut keys: Vec<_> = inner.index.iter().map(|(k, m)| (k.to_i128(), m)).collect();
-        keys.sort_unstable_by_key(|k| k.0);
+        let mut keys: Vec<(&Key, &VecMap<Lsn, u64>)> = inner.index.iter().collect();
+        keys.sort_by_key(|k| k.0);
 
         let ctx = RequestContextBuilder::extend(ctx)
             .page_content_kind(PageContentKind::InMemoryLayer)
             .build();
         for (key, vec_map) in keys.iter() {
-            let key = Key::from_i128(*key);
+            let key = **key;
             // Write all page versions
             for (lsn, pos) in vec_map.as_slice() {
                 cursor.read_blob_into_buf(*pos, &mut buf, &ctx).await?;

pageserver/src/tenant/storage_layer/layer.rs

@@ -125,7 +125,6 @@ impl Layer {
             let inner = Arc::new(DownloadedLayer {
                 owner: owner.clone(),
                 kind: tokio::sync::OnceCell::default(),
-                version: 0,
             });
             resident = Some(inner.clone());
 
@@ -164,7 +163,6 @@ impl Layer {
             let inner = Arc::new(DownloadedLayer {
                 owner: owner.clone(),
                 kind: tokio::sync::OnceCell::default(),
-                version: 0,
             });
             resident = Some(inner.clone());
             let access_stats = LayerAccessStats::empty_will_record_residence_event_later();
@@ -330,46 +328,42 @@ impl Layer {
 /// read with [`Layer::get_value_reconstruct_data`].
 ///
 /// [`LayerMap::search`]: crate::tenant::layer_map::LayerMap::search
-#[derive(Debug)]
 enum ResidentOrWantedEvicted {
     Resident(Arc<DownloadedLayer>),
-    WantedEvicted(Weak<DownloadedLayer>, usize),
+    WantedEvicted(Weak<DownloadedLayer>),
 }
 
 impl ResidentOrWantedEvicted {
-    fn get_and_upgrade(&mut self) -> Option<(Arc<DownloadedLayer>, bool)> {
+    fn get(&self) -> Option<Arc<DownloadedLayer>> {
         match self {
-            ResidentOrWantedEvicted::Resident(strong) => Some((strong.clone(), false)),
-            ResidentOrWantedEvicted::WantedEvicted(weak, _) => match weak.upgrade() {
+            ResidentOrWantedEvicted::Resident(strong) => Some(strong.clone()),
+            ResidentOrWantedEvicted::WantedEvicted(weak) => match weak.upgrade() {
                 Some(strong) => {
                     LAYER_IMPL_METRICS.inc_raced_wanted_evicted_accesses();
-
-                    *self = ResidentOrWantedEvicted::Resident(strong.clone());
-
-                    Some((strong, true))
+                    Some(strong)
                 }
                 None => None,
             },
         }
     }
-
     /// When eviction is first requested, drop down to holding a [`Weak`].
     ///
-    /// Returns `Some` if this was the first time eviction was requested. Care should be taken to
-    /// drop the possibly last strong reference outside of the mutex of
-    /// heavier_once_cell::OnceCell.
-    fn downgrade(&mut self) -> Option<Arc<DownloadedLayer>> {
-        match self {
+    /// Returns `true` if this was the first time eviction was requested.
+    fn downgrade(&mut self) -> &Weak<DownloadedLayer> {
+        let _was_first = match self {
             ResidentOrWantedEvicted::Resident(strong) => {
                 let weak = Arc::downgrade(strong);
-                let mut temp = ResidentOrWantedEvicted::WantedEvicted(weak, strong.version);
-                std::mem::swap(self, &mut temp);
-                match temp {
-                    ResidentOrWantedEvicted::Resident(strong) => Some(strong),
-                    ResidentOrWantedEvicted::WantedEvicted(..) => unreachable!("just swapped"),
-                }
+                *self = ResidentOrWantedEvicted::WantedEvicted(weak);
+                // returning the weak is not useful, because the drop could had already ran with
+                // the replacement above, and that will take care of cleaning the Option we are in
+                true
             }
-            ResidentOrWantedEvicted::WantedEvicted(..) => None,
+            ResidentOrWantedEvicted::WantedEvicted(_) => false,
+        };
+
+        match self {
+            ResidentOrWantedEvicted::WantedEvicted(ref weak) => weak,
+            _ => unreachable!("just wrote wanted evicted"),
         }
     }
 }
@@ -404,10 +398,8 @@ struct LayerInner {
     /// [`LayerInner::on_downloaded_layer_drop`].
     wanted_evicted: AtomicBool,
 
-    /// Version is to make sure we will only evict a specific download of a file.
-    ///
-    /// Incremented for each download, stored in `DownloadedLayer::version` or
-    /// `ResidentOrWantedEvicted::WantedEvicted`.
+    /// Version is to make sure we will in fact only evict a file if no new download has been
+    /// started.
     version: AtomicUsize,
 
     /// Allow subscribing to when the layer actually gets evicted.
@@ -523,14 +515,6 @@ impl LayerInner {
             .timeline_path(&timeline.tenant_id, &timeline.timeline_id)
             .join(desc.filename().to_string());
 
-        let (inner, version) = if let Some(inner) = downloaded {
-            let version = inner.version;
-            let resident = ResidentOrWantedEvicted::Resident(inner);
-            (heavier_once_cell::OnceCell::new(resident), version)
-        } else {
-            (heavier_once_cell::OnceCell::default(), 0)
-        };
-
         LayerInner {
             conf,
             path,
@@ -540,8 +524,12 @@ impl LayerInner {
             access_stats,
             wanted_garbage_collected: AtomicBool::new(false),
             wanted_evicted: AtomicBool::new(false),
-            inner,
-            version: AtomicUsize::new(version),
+            inner: if let Some(inner) = downloaded {
+                heavier_once_cell::OnceCell::new(ResidentOrWantedEvicted::Resident(inner))
+            } else {
+                heavier_once_cell::OnceCell::default()
+            },
+            version: AtomicUsize::new(0),
             status: tokio::sync::broadcast::channel(1).0,
             consecutive_failures: AtomicUsize::new(0),
             generation,
@@ -571,22 +559,20 @@ impl LayerInner {
 
         let mut rx = self.status.subscribe();
 
-        let strong = {
-            match self.inner.get() {
-                Some(mut either) => {
-                    self.wanted_evicted.store(true, Ordering::Relaxed);
-                    either.downgrade()
-                }
-                None => return Err(EvictionError::NotFound),
-            }
-        };
+        let res =
+            self.wanted_evicted
+                .compare_exchange(false, true, Ordering::Release, Ordering::Relaxed);
 
-        if strong.is_some() {
-            // drop the DownloadedLayer outside of the holding the guard
-            drop(strong);
+        if res.is_ok() {
             LAYER_IMPL_METRICS.inc_started_evictions();
         }
 
+        if self.get().is_none() {
+            // it was not evictable in the first place
+            // our store to the wanted_evicted does not matter; it will be reset by next download
+            return Err(EvictionError::NotFound);
+        }
+
         match rx.recv().await {
             Ok(Status::Evicted) => Ok(()),
             Ok(Status::Downloaded) => Err(EvictionError::Downloaded),
@@ -600,8 +586,7 @@ impl LayerInner {
                 //
                 // use however late (compared to the initial expressing of wanted) as the
                 // "outcome" now
-                LAYER_IMPL_METRICS.inc_broadcast_lagged();
-                match self.inner.get() {
+                match self.get() {
                     Some(_) => Err(EvictionError::Downloaded),
                     None => Ok(()),
                 }
@@ -616,12 +601,10 @@ impl LayerInner {
         allow_download: bool,
         ctx: Option<&RequestContext>,
     ) -> Result<Arc<DownloadedLayer>, DownloadError> {
-        let mut init_permit = None;
-
         loop {
-            let download = move |permit| async move {
+            let download = move || async move {
                 // disable any scheduled but not yet running eviction deletions for this
-                let next_version = 1 + self.version.fetch_add(1, Ordering::Relaxed);
+                self.version.fetch_add(1, Ordering::Relaxed);
 
                 // no need to make the evict_and_wait wait for the actual download to complete
                 drop(self.status.send(Status::Downloaded));
@@ -640,11 +623,7 @@ impl LayerInner {
                     .await
                     .map_err(DownloadError::PreStatFailed)?;
 
-                let permit = if let Some(reason) = needs_download {
-                    if let NeedsDownload::NotFile(ft) = reason {
-                        return Err(DownloadError::NotFile(ft));
-                    }
-
+                if let Some(reason) = needs_download {
                     // only reset this after we've decided we really need to download. otherwise it'd
                     // be impossible to mark cancelled downloads for eviction, like one could imagine
                     // we would like to do for prefetching which was not needed.
@@ -666,19 +645,16 @@ impl LayerInner {
                         return Err(DownloadError::DownloadRequired);
                     }
 
-                    self.spawn_download_and_wait(timeline, permit).await?
+                    self.spawn_download_and_wait(timeline).await?;
                 } else {
                     // the file is present locally, probably by a previous but cancelled call to
                     // get_or_maybe_download. alternatively we might be running without remote storage.
                     LAYER_IMPL_METRICS.inc_init_needed_no_download();
-
-                    permit
-                };
+                }
 
                 let res = Arc::new(DownloadedLayer {
                     owner: Arc::downgrade(self),
                     kind: tokio::sync::OnceCell::default(),
-                    version: next_version,
                 });
 
                 self.access_stats.record_residence_event(
@@ -686,55 +662,19 @@ impl LayerInner {
                     LayerResidenceEventReason::ResidenceChange,
                 );
 
-                Ok((ResidentOrWantedEvicted::Resident(res), permit))
+                Ok(ResidentOrWantedEvicted::Resident(res))
             };
 
-            if let Some(init_permit) = init_permit.take() {
-                // use the already held initialization permit because it is impossible to hit the
-                // below paths anymore essentially limiting the max loop iterations to 2.
-                let (value, init_permit) = download(init_permit).await?;
-                let mut guard = self.inner.set(value, init_permit);
-                let (strong, _upgraded) = guard
-                    .get_and_upgrade()
-                    .expect("init creates strong reference, we held the init permit");
+            let locked = self.inner.get_or_init(download).await?;
+
+            if let Some(strong) = Self::get_or_apply_evictedness(Some(locked), &self.wanted_evicted)
+            {
                 return Ok(strong);
             }
 
-            let (weak, permit) = {
-                let mut locked = self.inner.get_or_init(download).await?;
-
-                if let Some((strong, upgraded)) = locked.get_and_upgrade() {
-                    if upgraded {
-                        // when upgraded back, the Arc<DownloadedLayer> is still available, but
-                        // previously a `evict_and_wait` was received.
-                        self.wanted_evicted.store(false, Ordering::Relaxed);
-
-                        // error out any `evict_and_wait`
-                        drop(self.status.send(Status::Downloaded));
-                        LAYER_IMPL_METRICS
-                            .inc_eviction_cancelled(EvictionCancelled::UpgradedBackOnAccess);
-                    }
-
-                    return Ok(strong);
-                } else {
-                    // path to here: the evict_blocking is stuck on spawn_blocking queue.
-                    //
-                    // reset the contents, deactivating the eviction and causing a
-                    // EvictionCancelled::LostToDownload or EvictionCancelled::VersionCheckFailed.
-                    locked.take_and_deinit()
-                }
-            };
-
-            // unlock first, then drop the weak, but because upgrade failed, we
-            // know it cannot be a problem.
-
-            assert!(
-                matches!(weak, ResidentOrWantedEvicted::WantedEvicted(..)),
-                "unexpected {weak:?}, ResidentOrWantedEvicted::get_and_upgrade has a bug"
-            );
-
-            init_permit = Some(permit);
-
+            // the situation in which we might need to retry is that our init was ready
+            // immediatedly, but the DownloadedLayer had been dropped BUT failed to complete
+            // Self::evict_blocking
             LAYER_IMPL_METRICS.inc_retried_get_or_maybe_download();
         }
     }
@@ -769,12 +709,10 @@ impl LayerInner {
     async fn spawn_download_and_wait(
         self: &Arc<Self>,
         timeline: Arc<Timeline>,
-        permit: heavier_once_cell::InitPermit,
-    ) -> Result<heavier_once_cell::InitPermit, DownloadError> {
+    ) -> Result<(), DownloadError> {
         let task_name = format!("download layer {}", self);
 
         let (tx, rx) = tokio::sync::oneshot::channel();
-
         // this is sadly needed because of task_mgr::shutdown_tasks, otherwise we cannot
         // block tenant::mgr::remove_tenant_from_memory.
 
@@ -808,9 +746,9 @@ impl LayerInner {
                     }
                 };
 
-                if let Err(res) = tx.send((result, permit)) {
+                if let Err(res) = tx.send(result) {
                     match res {
-                        (Ok(()), _) => {
+                        Ok(()) => {
                             // our caller is cancellation safe so this is fine; if someone
                             // else requests the layer, they'll find it already downloaded
                             // or redownload.
@@ -821,7 +759,7 @@ impl LayerInner {
                             tracing::info!("layer file download completed after requester had cancelled");
                             LAYER_IMPL_METRICS.inc_download_completed_without_requester();
                         },
-                        (Err(e), _) => {
+                        Err(e) => {
                             // our caller is cancellation safe, but we might be racing with
                             // another attempt to initialize. before we have cancellation
                             // token support: these attempts should converge regardless of
@@ -837,7 +775,7 @@ impl LayerInner {
             .in_current_span(),
         );
         match rx.await {
-            Ok((Ok(()), permit)) => {
+            Ok(Ok(())) => {
                 if let Some(reason) = self
                     .needs_download()
                     .await
@@ -849,10 +787,9 @@ impl LayerInner {
 
                 self.consecutive_failures.store(0, Ordering::Relaxed);
 
-                Ok(permit)
+                Ok(())
             }
-            Ok((Err(e), _permit)) => {
-                // FIXME: this should be with the spawned task and be cancellation sensitive
+            Ok(Err(e)) => {
                 let consecutive_failures =
                     self.consecutive_failures.fetch_add(1, Ordering::Relaxed);
                 tracing::error!(consecutive_failures, "layer file download failed: {e:#}");
@@ -870,6 +807,33 @@ impl LayerInner {
         }
     }
 
+    /// Access the current state without waiting for the file to be downloaded.
+    ///
+    /// Requires that we've initialized to state which is respective to the
+    /// actual residency state.
+    fn get(&self) -> Option<Arc<DownloadedLayer>> {
+        let locked = self.inner.get();
+        Self::get_or_apply_evictedness(locked, &self.wanted_evicted)
+    }
+
+    fn get_or_apply_evictedness(
+        guard: Option<heavier_once_cell::Guard<'_, ResidentOrWantedEvicted>>,
+        wanted_evicted: &AtomicBool,
+    ) -> Option<Arc<DownloadedLayer>> {
+        if let Some(mut x) = guard {
+            if let Some(won) = x.get() {
+                // there are no guarantees that we will always get to observe a concurrent call
+                // to evict
+                if wanted_evicted.load(Ordering::Acquire) {
+                    x.downgrade();
+                }
+                return Some(won);
+            }
+        }
+
+        None
+    }
+
     async fn needs_download(&self) -> Result<Option<NeedsDownload>, std::io::Error> {
         match tokio::fs::metadata(&self.path).await {
             Ok(m) => Ok(self.is_file_present_and_good_size(&m).err()),
@@ -889,7 +853,7 @@ impl LayerInner {
     fn is_file_present_and_good_size(&self, m: &std::fs::Metadata) -> Result<(), NeedsDownload> {
         // in future, this should include sha2-256 validation of the file.
         if !m.is_file() {
-            Err(NeedsDownload::NotFile(m.file_type()))
+            Err(NeedsDownload::NotFile)
         } else if m.len() != self.desc.file_size {
             Err(NeedsDownload::WrongSize {
                 actual: m.len(),
@@ -903,9 +867,7 @@ impl LayerInner {
     fn info(&self, reset: LayerAccessStatsReset) -> HistoricLayerInfo {
         let layer_file_name = self.desc.filename().file_name();
 
-        // this is not accurate: we could have the file locally but there was a cancellation
-        // and now we are not in sync, or we are currently downloading it.
-        let remote = self.inner.get().is_none();
+        let remote = self.get().is_none();
 
         let access_stats = self.access_stats.as_api_model(reset);
 
@@ -934,7 +896,7 @@ impl LayerInner {
     }
 
     /// `DownloadedLayer` is being dropped, so it calls this method.
-    fn on_downloaded_layer_drop(self: Arc<LayerInner>, version: usize) {
+    fn on_downloaded_layer_drop(self: Arc<LayerInner>) {
         let gc = self.wanted_garbage_collected.load(Ordering::Acquire);
         let evict = self.wanted_evicted.load(Ordering::Acquire);
         let can_evict = self.have_remote_client;
@@ -942,16 +904,15 @@ impl LayerInner {
         if gc {
             // do nothing now, only in LayerInner::drop
         } else if can_evict && evict {
-            let span = tracing::info_span!(parent: None, "layer_evict", tenant_id = %self.desc.tenant_id, timeline_id = %self.desc.timeline_id, layer=%self, %version);
+            let version = self.version.load(Ordering::Relaxed);
+
+            let span = tracing::info_span!(parent: None, "layer_evict", tenant_id = %self.desc.tenant_id, timeline_id = %self.desc.timeline_id, layer=%self);
 
             // downgrade for queueing, in case there's a tear down already ongoing we should not
             // hold it alive.
             let this = Arc::downgrade(&self);
             drop(self);
 
-            // NOTE: this scope *must* never call `self.inner.get` because evict_and_wait might
-            // drop while the `self.inner` is being locked, leading to a deadlock.
-
             crate::task_mgr::BACKGROUND_RUNTIME.spawn_blocking(move || {
                 let _g = span.entered();
 
@@ -961,15 +922,19 @@ impl LayerInner {
                     LAYER_IMPL_METRICS.inc_eviction_cancelled(EvictionCancelled::LayerGone);
                     return;
                 };
-                match this.evict_blocking(version) {
-                    Ok(()) => LAYER_IMPL_METRICS.inc_completed_evictions(),
-                    Err(reason) => LAYER_IMPL_METRICS.inc_eviction_cancelled(reason),
-                }
+                this.evict_blocking(version);
             });
         }
     }
 
-    fn evict_blocking(&self, only_version: usize) -> Result<(), EvictionCancelled> {
+    fn evict_blocking(&self, version: usize) {
+        match self.evict_blocking0(version) {
+            Ok(()) => LAYER_IMPL_METRICS.inc_completed_evictions(),
+            Err(reason) => LAYER_IMPL_METRICS.inc_eviction_cancelled(reason),
+        }
+    }
+
+    fn evict_blocking0(&self, version: usize) -> Result<(), EvictionCancelled> {
         // deleted or detached timeline, don't do anything.
         let Some(timeline) = self.timeline.upgrade() else {
             return Err(EvictionCancelled::TimelineGone);
@@ -980,34 +945,32 @@ impl LayerInner {
         let _permit = {
             let maybe_downloaded = self.inner.get();
 
-            let (_weak, permit) = match maybe_downloaded {
-                Some(mut guard) => {
-                    if let ResidentOrWantedEvicted::WantedEvicted(_weak, version) = &*guard {
-                        if *version == only_version {
-                            guard.take_and_deinit()
-                        } else {
-                            // this was not for us; maybe there's another eviction job
-                            // TODO: does it make any sense to stall here? unique versions do not
-                            // matter, we only want to make sure not to evict a resident, which we
-                            // are not doing.
-                            return Err(EvictionCancelled::VersionCheckFailed);
-                        }
-                    } else {
-                        return Err(EvictionCancelled::AlreadyReinitialized);
-                    }
+            if version != self.version.load(Ordering::Relaxed) {
+                // downloadness-state has advanced, we might no longer be the latest eviction
+                // work; don't do anything.
+                //
+                // this is possible to get to by having:
+                //
+                // 1. wanted_evicted.store(true)
+                // 2. ResidentOrWantedEvicted::downgrade
+                // 3. DownloadedLayer::drop
+                // 4. LayerInner::get_or_maybe_download
+                // 5. LayerInner::evict_blocking
+                return Err(EvictionCancelled::VersionCheckFailed);
+            }
+
+            // free the DownloadedLayer allocation
+            match maybe_downloaded.map(|mut g| g.take_and_deinit()) {
+                Some((taken, permit)) => {
+                    assert!(matches!(taken, ResidentOrWantedEvicted::WantedEvicted(_)));
+                    permit
                 }
                 None => {
-                    // already deinitialized, perhaps get_or_maybe_download did this and is
-                    // currently waiting to reinitialize it
-                    return Err(EvictionCancelled::LostToDownload);
+                    unreachable!("we do the version checking for this exact reason")
                 }
-            };
-
-            permit
+            }
         };
 
-        // now accesses to inner.get_or_init wait on the semaphore or the `_permit`
-
         self.access_stats.record_residence_event(
             LayerResidenceStatus::Evicted,
             LayerResidenceEventReason::ResidenceChange,
@@ -1088,8 +1051,6 @@ enum DownloadError {
     ContextAndConfigReallyDeniesDownloads,
     #[error("downloading is really required but not allowed by this method")]
     DownloadRequired,
-    #[error("layer path exists, but it is not a file: {0:?}")]
-    NotFile(std::fs::FileType),
     /// Why no error here? Because it will be reported by page_service. We should had also done
     /// retries already.
     #[error("downloading evicted layer file failed")]
@@ -1105,7 +1066,7 @@ enum DownloadError {
 #[derive(Debug, PartialEq)]
 pub(crate) enum NeedsDownload {
     NotFound,
-    NotFile(std::fs::FileType),
+    NotFile,
     WrongSize { actual: u64, expected: u64 },
 }
 
@@ -1113,7 +1074,7 @@ impl std::fmt::Display for NeedsDownload {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         match self {
             NeedsDownload::NotFound => write!(f, "file was not found"),
-            NeedsDownload::NotFile(ft) => write!(f, "path is not a file; {ft:?}"),
+            NeedsDownload::NotFile => write!(f, "path is not a file"),
             NeedsDownload::WrongSize { actual, expected } => {
                 write!(f, "file size mismatch {actual} vs. {expected}")
             }
@@ -1125,7 +1086,6 @@ impl std::fmt::Display for NeedsDownload {
 pub(crate) struct DownloadedLayer {
     owner: Weak<LayerInner>,
     kind: tokio::sync::OnceCell<anyhow::Result<LayerKind>>,
-    version: usize,
 }
 
 impl std::fmt::Debug for DownloadedLayer {
@@ -1133,7 +1093,6 @@ impl std::fmt::Debug for DownloadedLayer {
         f.debug_struct("DownloadedLayer")
             // owner omitted because it is always "Weak"
             .field("kind", &self.kind)
-            .field("version", &self.version)
             .finish()
     }
 }
@@ -1141,7 +1100,7 @@ impl std::fmt::Debug for DownloadedLayer {
 impl Drop for DownloadedLayer {
     fn drop(&mut self) {
         if let Some(owner) = self.owner.upgrade() {
-            owner.on_downloaded_layer_drop(self.version);
+            owner.on_downloaded_layer_drop();
         } else {
             // no need to do anything, we are shutting down
         }
@@ -1491,13 +1450,6 @@ impl LayerImplMetrics {
             .unwrap()
             .inc();
     }
-
-    fn inc_broadcast_lagged(&self) {
-        self.rare_counters
-            .get_metric_with_label_values(&["broadcast_lagged"])
-            .unwrap()
-            .inc();
-    }
 }
 
 enum EvictionCancelled {
@@ -1506,11 +1458,6 @@ enum EvictionCancelled {
     VersionCheckFailed,
     FileNotFound,
     RemoveFailed,
-    AlreadyReinitialized,
-    /// Not evicted because of a pending reinitialization
-    LostToDownload,
-    /// After eviction, there was a new layer access which cancelled the eviction.
-    UpgradedBackOnAccess,
 }
 
 impl EvictionCancelled {
@@ -1521,9 +1468,6 @@ impl EvictionCancelled {
             EvictionCancelled::VersionCheckFailed => "version_check_fail",
             EvictionCancelled::FileNotFound => "file_not_found",
             EvictionCancelled::RemoveFailed => "remove_failed",
-            EvictionCancelled::AlreadyReinitialized => "already_reinitialized",
-            EvictionCancelled::LostToDownload => "lost_to_download",
-            EvictionCancelled::UpgradedBackOnAccess => "upgraded_back_on_access",
         }
     }
 }

pageserver/src/tenant/tasks.rs

@@ -162,11 +162,8 @@ async fn compaction_loop(tenant: Arc<Tenant>, cancel: CancellationToken) {
 
             // TODO: we shouldn't need to await to find tenant and this could be moved outside of
             // loop, #3501. There are also additional "allowed_errors" in tests.
-            if first {
-                first = false;
-                if random_init_delay(period, &cancel).await.is_err() {
-                    break;
-                }
+            if first && random_init_delay(period, &cancel).await.is_err() {
+                break;
             }
 
             let started_at = Instant::now();
@@ -196,7 +193,16 @@ async fn compaction_loop(tenant: Arc<Tenant>, cancel: CancellationToken) {
                 }
             };
 
-            warn_when_period_overrun(started_at.elapsed(), period, BackgroundLoopKind::Compaction);
+            if !first {
+                // The first iteration is typically much slower, because all tenants compete for the
+                // compaction sempahore to run, and because of concurrent startup work like initializing
+                // logical sizes.  To avoid routinely spamming warnings, we suppress this log on first iteration.
+                warn_when_period_overrun(
+                    started_at.elapsed(),
+                    period,
+                    BackgroundLoopKind::Compaction,
+                );
+            }
 
             // Sleep
             if tokio::time::timeout(sleep_duration, cancel.cancelled())
@@ -205,6 +211,8 @@ async fn compaction_loop(tenant: Arc<Tenant>, cancel: CancellationToken) {
             {
                 break;
             }
+
+            first = false;
         }
     }
     .await;
@@ -239,11 +247,8 @@ async fn gc_loop(tenant: Arc<Tenant>, cancel: CancellationToken) {
 
             let period = tenant.get_gc_period();
 
-            if first {
-                first = false;
-                if random_init_delay(period, &cancel).await.is_err() {
-                    break;
-                }
+            if first && random_init_delay(period, &cancel).await.is_err() {
+                break;
             }
 
             let started_at = Instant::now();
@@ -277,7 +282,12 @@ async fn gc_loop(tenant: Arc<Tenant>, cancel: CancellationToken) {
                 }
             };
 
-            warn_when_period_overrun(started_at.elapsed(), period, BackgroundLoopKind::Gc);
+            if !first {
+                // The first iteration is typically much slower, because all tenants compete for the
+                // compaction sempahore to run, and because of concurrent startup work like initializing
+                // logical sizes.  To avoid routinely spamming warnings, we suppress this log on first iteration.
+                warn_when_period_overrun(started_at.elapsed(), period, BackgroundLoopKind::Gc);
+            }
 
             // Sleep
             if tokio::time::timeout(sleep_duration, cancel.cancelled())
@@ -286,6 +296,8 @@ async fn gc_loop(tenant: Arc<Tenant>, cancel: CancellationToken) {
             {
                 break;
             }
+
+            first = false;
         }
     }
     .await;
@@ -361,7 +373,7 @@ pub(crate) fn warn_when_period_overrun(
         // humantime does no significant digits clamping whereas Duration's debug is a bit more
         // intelligent. however it makes sense to keep the "configuration format" for period, even
         // though there's no way to output the actual config value.
-        info!(
+        warn!(
             ?elapsed,
             period = %humantime::format_duration(period),
             ?task,

pageserver/src/tenant/timeline/walreceiver/walreceiver_connection.rs

@@ -459,7 +459,7 @@ async fn identify_system(client: &Client) -> anyhow::Result<IdentifySystem> {
 
     // extract the row contents into an IdentifySystem struct.
     // written as a closure so I can use ? for Option here.
-    if let Some(SimpleQueryMessage::Row(first_row)) = response.first() {
+    if let Some(SimpleQueryMessage::Row(first_row)) = response.get(0) {
         Ok(IdentifySystem {
             systemid: get_parse(first_row, 0)?,
             timeline: get_parse(first_row, 1)?,

pageserver/src/virtual_file.rs

@@ -19,7 +19,6 @@ use std::io::{Error, ErrorKind, Seek, SeekFrom};
 use std::os::unix::fs::FileExt;
 use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
 use std::sync::{RwLock, RwLockWriteGuard};
-use utils::fs_ext;
 
 ///
 /// A virtual file descriptor. You can use this just like std::fs::File, but internally
@@ -174,78 +173,37 @@ impl OpenFiles {
     }
 }
 
-/// Identify error types that should alwways terminate the process.  Other
-/// error types may be elegible for retry.
-pub(crate) fn is_fatal_io_error(e: &std::io::Error) -> bool {
-    use nix::errno::Errno::*;
-    match e.raw_os_error().map(nix::errno::from_i32) {
-        Some(EIO) => {
-            // Terminate on EIO because we no longer trust the device to store
-            // data safely, or to uphold persistence guarantees on fsync.
-            true
-        }
-        Some(EROFS) => {
-            // Terminate on EROFS because a filesystem is usually remounted
-            // readonly when it has experienced some critical issue, so the same
-            // logic as EIO applies.
-            true
-        }
-        Some(EACCES) => {
-            // Terminate on EACCESS because we should always have permissions
-            // for our own data dir: if we don't, then we can't do our job and
-            // need administrative intervention to fix permissions.  Terminating
-            // is the best way to make sure we stop cleanly rather than going
-            // into infinite retry loops, and will make it clear to the outside
-            // world that we need help.
-            true
-        }
-        _ => {
-            // Treat all other local file I/O errors are retryable.  This includes:
-            // - ENOSPC: we stay up and wait for eviction to free some space
-            // - EINVAL, EBADF, EBADFD: this is a code bug, not a filesystem/hardware issue
-            // - WriteZero, Interrupted: these are used internally VirtualFile
-            false
-        }
-    }
+#[derive(Debug, thiserror::Error)]
+pub enum CrashsafeOverwriteError {
+    #[error("final path has no parent dir")]
+    FinalPathHasNoParentDir,
+    #[error("remove tempfile")]
+    RemovePreviousTempfile(#[source] std::io::Error),
+    #[error("create tempfile")]
+    CreateTempfile(#[source] std::io::Error),
+    #[error("write tempfile")]
+    WriteContents(#[source] std::io::Error),
+    #[error("sync tempfile")]
+    SyncTempfile(#[source] std::io::Error),
+    #[error("rename tempfile to final path")]
+    RenameTempfileToFinalPath(#[source] std::io::Error),
+    #[error("open final path parent dir")]
+    OpenFinalPathParentDir(#[source] std::io::Error),
+    #[error("sync final path parent dir")]
+    SyncFinalPathParentDir(#[source] std::io::Error),
 }
-
-/// Call this when the local filesystem gives us an error with an external
-/// cause: this includes EIO, EROFS, and EACCESS: all these indicate either
-/// bad storage or bad configuration, and we can't fix that from inside
-/// a running process.
-pub(crate) fn on_fatal_io_error(e: &std::io::Error, context: &str) -> ! {
-    tracing::error!("Fatal I/O error: {e}: {context})");
-    std::process::abort();
-}
-
-pub(crate) trait MaybeFatalIo<T> {
-    fn maybe_fatal_err(self, context: &str) -> std::io::Result<T>;
-    fn fatal_err(self, context: &str) -> T;
-}
-
-impl<T> MaybeFatalIo<T> for std::io::Result<T> {
-    /// Terminate the process if the result is an error of a fatal type, else pass it through
-    ///
-    /// This is appropriate for writes, where we typically want to die on EIO/ACCES etc, but
-    /// not on ENOSPC.
-    fn maybe_fatal_err(self, context: &str) -> std::io::Result<T> {
-        if let Err(e) = &self {
-            if is_fatal_io_error(e) {
-                on_fatal_io_error(e, context);
-            }
-        }
-        self
-    }
-
-    /// Terminate the process on any I/O error.
-    ///
-    /// This is appropriate for reads on files that we know exist: they should always work.
-    fn fatal_err(self, context: &str) -> T {
+impl CrashsafeOverwriteError {
+    /// Returns true iff the new contents are durably stored.
+    pub fn are_new_contents_durable(&self) -> bool {
         match self {
-            Ok(v) => v,
-            Err(e) => {
-                on_fatal_io_error(&e, context);
-            }
+            Self::FinalPathHasNoParentDir => false,
+            Self::RemovePreviousTempfile(_) => false,
+            Self::CreateTempfile(_) => false,
+            Self::WriteContents(_) => false,
+            Self::SyncTempfile(_) => false,
+            Self::RenameTempfileToFinalPath(_) => false,
+            Self::OpenFinalPathParentDir(_) => false,
+            Self::SyncFinalPathParentDir(_) => true,
         }
     }
 }
@@ -326,13 +284,15 @@ impl VirtualFile {
         final_path: &Utf8Path,
         tmp_path: &Utf8Path,
         content: &[u8],
-    ) -> std::io::Result<()> {
+    ) -> Result<(), CrashsafeOverwriteError> {
         let Some(final_path_parent) = final_path.parent() else {
-            return Err(std::io::Error::from_raw_os_error(
-                nix::errno::Errno::EINVAL as i32,
-            ));
+            return Err(CrashsafeOverwriteError::FinalPathHasNoParentDir);
         };
-        std::fs::remove_file(tmp_path).or_else(fs_ext::ignore_not_found)?;
+        match std::fs::remove_file(tmp_path) {
+            Ok(()) => {}
+            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
+            Err(e) => return Err(CrashsafeOverwriteError::RemovePreviousTempfile(e)),
+        }
         let mut file = Self::open_with_options(
             tmp_path,
             OpenOptions::new()
@@ -341,20 +301,31 @@ impl VirtualFile {
                 // we bail out instead of causing damage.
                 .create_new(true),
         )
-        .await?;
-        file.write_all(content).await?;
-        file.sync_all().await?;
+        .await
+        .map_err(CrashsafeOverwriteError::CreateTempfile)?;
+        file.write_all(content)
+            .await
+            .map_err(CrashsafeOverwriteError::WriteContents)?;
+        file.sync_all()
+            .await
+            .map_err(CrashsafeOverwriteError::SyncTempfile)?;
         drop(file); // before the rename, that's important!
                     // renames are atomic
-        std::fs::rename(tmp_path, final_path)?;
+        std::fs::rename(tmp_path, final_path)
+            .map_err(CrashsafeOverwriteError::RenameTempfileToFinalPath)?;
         // Only open final path parent dirfd now, so that this operation only
         // ever holds one VirtualFile fd at a time.  That's important because
         // the current `find_victim_slot` impl might pick the same slot for both
         // VirtualFile., and it eventually does a blocking write lock instead of
         // try_lock.
         let final_parent_dirfd =
-            Self::open_with_options(final_path_parent, OpenOptions::new().read(true)).await?;
-        final_parent_dirfd.sync_all().await?;
+            Self::open_with_options(final_path_parent, OpenOptions::new().read(true))
+                .await
+                .map_err(CrashsafeOverwriteError::OpenFinalPathParentDir)?;
+        final_parent_dirfd
+            .sync_all()
+            .await
+            .map_err(CrashsafeOverwriteError::SyncFinalPathParentDir)?;
         Ok(())
     }
 

pageserver/src/walredo.rs

@@ -857,8 +857,7 @@ impl WalRedoProcess {
             let in_revents = stdin_pollfds[0].revents().unwrap();
             if in_revents & (PollFlags::POLLERR | PollFlags::POLLOUT) != PollFlags::empty() {
                 nwrite += proc.stdin.write(&writebuf[nwrite..])?;
-            }
-            if in_revents.contains(PollFlags::POLLHUP) {
+            } else if in_revents.contains(PollFlags::POLLHUP) {
                 // We still have more data to write, but the process closed the pipe.
                 anyhow::bail!("WAL redo process closed its stdin unexpectedly");
             }
@@ -908,8 +907,7 @@ impl WalRedoProcess {
                 let out_revents = stdout_pollfds[0].revents().unwrap();
                 if out_revents & (PollFlags::POLLERR | PollFlags::POLLIN) != PollFlags::empty() {
                     nresult += output.stdout.read(&mut resultbuf[nresult..])?;
-                }
-                if out_revents.contains(PollFlags::POLLHUP) {
+                } else if out_revents.contains(PollFlags::POLLHUP) {
                     anyhow::bail!("WAL redo process closed its stdout unexpectedly");
                 }
             }

proxy/src/console/provider/mock.rs

@@ -59,7 +59,7 @@ impl Api {
             let rows = client.query(query, &[&creds.user]).await?;
 
             // We can get at most one row, because `rolname` is unique.
-            let row = match rows.first() {
+            let row = match rows.get(0) {
                 Some(row) => row,
                 // This means that the user doesn't exist, so there can be no secret.
                 // However, this is still a *valid* outcome which is very similar

proxy/src/console/provider/neon.rs

@@ -49,7 +49,7 @@ impl Api {
                 .endpoint
                 .get("proxy_get_role_secret")
                 .header("X-Request-ID", &request_id)
-                .header("Authorization", format!("Bearer {}", &self.jwt))
+                .header("Authorization", &self.jwt)
                 .query(&[("session_id", extra.session_id)])
                 .query(&[
                     ("application_name", extra.application_name),
@@ -94,7 +94,7 @@ impl Api {
                 .endpoint
                 .get("proxy_wake_compute")
                 .header("X-Request-ID", &request_id)
-                .header("Authorization", format!("Bearer {}", &self.jwt))
+                .header("Authorization", &self.jwt)
                 .query(&[("session_id", extra.session_id)])
                 .query(&[
                     ("application_name", extra.application_name),

proxy/src/scram.rs

@@ -18,6 +18,7 @@ mod password;
 pub use exchange::Exchange;
 pub use key::ScramKey;
 pub use secret::ServerSecret;
+pub use secret::*;
 
 use hmac::{Hmac, Mac};
 use sha2::{Digest, Sha256};

test_runner/fixtures/pageserver/utils.py

@@ -249,7 +249,7 @@ def assert_prefix_empty(neon_env_builder: "NeonEnvBuilder", prefix: Optional[str
             # this has been seen in the wild by tests with the below contradicting logging
             # https://neon-github-public-dev.s3.amazonaws.com/reports/pr-5322/6207777020/index.html#suites/3556ed71f2d69272a7014df6dcb02317/53b5c368b5a68865
             # this seems like a mock_s3 issue
-            log.warning(
+            log.warn(
                 f"contrading ListObjectsV2 response with KeyCount={keys} and Contents={objects}, CommonPrefixes={common_prefixes}, assuming this means KeyCount=0"
             )
             keys = 0
@@ -257,7 +257,7 @@ def assert_prefix_empty(neon_env_builder: "NeonEnvBuilder", prefix: Optional[str
             # this has been seen in one case with mock_s3:
             # https://neon-github-public-dev.s3.amazonaws.com/reports/pr-4938/6000769714/index.html#suites/3556ed71f2d69272a7014df6dcb02317/ca01e4f4d8d9a11f
             # looking at moto impl, it might be there's a race with common prefix (sub directory) not going away with deletes
-            log.warning(
+            log.warn(
                 f"contradicting ListObjectsV2 response with KeyCount={keys} and Contents={objects}, CommonPrefixes={common_prefixes}"
             )
 

test_runner/performance/test_bulk_insert.py

@@ -1,10 +1,8 @@
 from contextlib import closing
 
-from fixtures.benchmark_fixture import MetricReport
 from fixtures.compare_fixtures import NeonCompare, PgCompare
 from fixtures.pageserver.utils import wait_tenant_status_404
 from fixtures.pg_version import PgVersion
-from fixtures.types import Lsn
 
 
 #
@@ -20,8 +18,6 @@ from fixtures.types import Lsn
 def test_bulk_insert(neon_with_baseline: PgCompare):
     env = neon_with_baseline
 
-    start_lsn = Lsn(env.pg.safe_psql("SELECT pg_current_wal_lsn()")[0][0])
-
     with closing(env.pg.connect()) as conn:
         with conn.cursor() as cur:
             cur.execute("create table huge (i int, j int);")
@@ -35,13 +31,6 @@ def test_bulk_insert(neon_with_baseline: PgCompare):
             env.report_peak_memory_use()
             env.report_size()
 
-    # Report amount of wal written. Useful for comparing vanilla wal format vs
-    # neon wal format, measuring neon write amplification, etc.
-    end_lsn = Lsn(env.pg.safe_psql("SELECT pg_current_wal_lsn()")[0][0])
-    wal_written_bytes = end_lsn - start_lsn
-    wal_written_mb = round(wal_written_bytes / (1024 * 1024))
-    env.zenbenchmark.record("wal_written", wal_written_mb, "MB", MetricReport.TEST_PARAM)
-
     # When testing neon, also check how long it takes the pageserver to reingest the
     # wal from safekeepers. If this number is close to total runtime, then the pageserver
     # is the bottleneck.