Reduce noise from moto GET/PUT operations

Moto prints messages like this:

    127.0.0.1 - - [07/Oct/2024 12:35:16] "PUT /bucket-name/path?x-id=PutObject HTTP/1.1" 200 -

After the root logger adds its context information, this is what
actually gets printed to the log:

    2024-10-07 22:35:16.371 INFO [_internal.py:97] 127.0.0.1 - - [07/Oct/2024 22:35:16] "PUT /bucket-name/path?x-id=PutObject HTTP/1.1" 200 -

That's very verbose. Remove the hostname and the extra timestamp, to
make it a little less verbose. With this PR, the final output looks
like this:

    2024-10-07 22:35:16.371 INFO [_internal.py:97] "PUT /bucket-name/path?x-id=PutObject HTTP/1.1" 200 -

.dockerignore

@@ -5,9 +5,7 @@
 !Cargo.toml
 !Makefile
 !rust-toolchain.toml
-!scripts/combine_control_files.py
 !scripts/ninstall.sh
-!vm-cgconfig.conf
 !docker-compose/run-tests.sh
 
 # Directories
@@ -17,15 +15,12 @@
 !compute_tools/
 !control_plane/
 !libs/
-!neon_local/
 !pageserver/
-!patches/
 !pgxn/
 !proxy/
 !storage_scrubber/
 !safekeeper/
 !storage_broker/
 !storage_controller/
-!trace/
 !vendor/postgres-*/
 !workspace_hack/

.github/workflows/report-workflow-stats.yml

@@ -0,0 +1,41 @@
+name: Report Workflow Stats
+
+on:
+  workflow_run:
+    workflows:
+    - Add `external` label to issues and PRs created by external users
+    - Benchmarking
+    - Build and Test
+    - Build and Test Locally
+    - Build build-tools image
+    - Check Permissions
+    - Check build-tools image
+    - Check neon with extra platform builds
+    - Cloud Regression Test
+    - Create Release Branch
+    - Handle `approved-for-ci-run` label
+    - Lint GitHub Workflows
+    - Notify Slack channel about upcoming release
+    - Periodic pagebench performance test on dedicated EC2 machine in eu-central-1 region
+    - Pin build-tools image
+    - Prepare benchmarking databases by restoring dumps
+    - Push images to ACR
+    - Test Postgres client libraries
+    - Trigger E2E Tests
+    - cleanup caches by a branch
+    types: [completed]
+
+jobs:
+  gh-workflow-stats:
+    name: Github Workflow Stats
+    runs-on: ubuntu-22.04
+    permissions:
+      actions: read
+    steps:
+    - name: Export GH Workflow Stats
+      uses: fedordikarev/gh-workflow-stats-action@v0.1.2
+      with:
+        DB_URI: ${{ secrets.GH_REPORT_STATS_DB_RW_CONNSTR }}
+        DB_TABLE: "gh_workflow_stats_neon"
+        GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+        GH_RUN_ID: ${{ github.event.workflow_run.id }}

Makefile

@@ -168,27 +168,27 @@ postgres-check-%: postgres-%
 neon-pg-ext-%: postgres-%
 	+@echo "Compiling neon $*"
 	mkdir -p $(POSTGRES_INSTALL_DIR)/build/neon-$*
-	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/$*/bin/pg_config CFLAGS='$(PG_CFLAGS) $(COPT)' \
+	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/$*/bin/pg_config COPT='$(COPT)' \
 		-C $(POSTGRES_INSTALL_DIR)/build/neon-$* \
 		-f $(ROOT_PROJECT_DIR)/pgxn/neon/Makefile install
 	+@echo "Compiling neon_walredo $*"
 	mkdir -p $(POSTGRES_INSTALL_DIR)/build/neon-walredo-$*
-	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/$*/bin/pg_config CFLAGS='$(PG_CFLAGS) $(COPT)' \
+	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/$*/bin/pg_config COPT='$(COPT)' \
 		-C $(POSTGRES_INSTALL_DIR)/build/neon-walredo-$* \
 		-f $(ROOT_PROJECT_DIR)/pgxn/neon_walredo/Makefile install
 	+@echo "Compiling neon_rmgr $*"
 	mkdir -p $(POSTGRES_INSTALL_DIR)/build/neon-rmgr-$*
-	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/$*/bin/pg_config CFLAGS='$(PG_CFLAGS) $(COPT)' \
+	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/$*/bin/pg_config COPT='$(COPT)' \
 		-C $(POSTGRES_INSTALL_DIR)/build/neon-rmgr-$* \
 		-f $(ROOT_PROJECT_DIR)/pgxn/neon_rmgr/Makefile install
 	+@echo "Compiling neon_test_utils $*"
 	mkdir -p $(POSTGRES_INSTALL_DIR)/build/neon-test-utils-$*
-	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/$*/bin/pg_config CFLAGS='$(PG_CFLAGS) $(COPT)' \
+	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/$*/bin/pg_config COPT='$(COPT)' \
 		-C $(POSTGRES_INSTALL_DIR)/build/neon-test-utils-$* \
 		-f $(ROOT_PROJECT_DIR)/pgxn/neon_test_utils/Makefile install
 	+@echo "Compiling neon_utils $*"
 	mkdir -p $(POSTGRES_INSTALL_DIR)/build/neon-utils-$*
-	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/$*/bin/pg_config CFLAGS='$(PG_CFLAGS) $(COPT)' \
+	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/$*/bin/pg_config COPT='$(COPT)' \
 		-C $(POSTGRES_INSTALL_DIR)/build/neon-utils-$* \
 		-f $(ROOT_PROJECT_DIR)/pgxn/neon_utils/Makefile install
 
@@ -220,7 +220,7 @@ neon-pg-clean-ext-%:
 walproposer-lib: neon-pg-ext-v17
 	+@echo "Compiling walproposer-lib"
 	mkdir -p $(POSTGRES_INSTALL_DIR)/build/walproposer-lib
-	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/v17/bin/pg_config CFLAGS='$(PG_CFLAGS) $(COPT)' \
+	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/v17/bin/pg_config COPT='$(COPT)' \
 		-C $(POSTGRES_INSTALL_DIR)/build/walproposer-lib \
 		-f $(ROOT_PROJECT_DIR)/pgxn/neon/Makefile walproposer-lib
 	cp $(POSTGRES_INSTALL_DIR)/v17/lib/libpgport.a $(POSTGRES_INSTALL_DIR)/build/walproposer-lib
@@ -333,7 +333,7 @@ postgres-%-pgindent: postgres-%-pg-bsd-indent postgres-%-typedefs.list
 # Indent pxgn/neon.
 .PHONY: neon-pgindent
 neon-pgindent: postgres-v17-pg-bsd-indent neon-pg-ext-v17
-	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/v17/bin/pg_config CFLAGS='$(PG_CFLAGS) $(COPT)' \
+	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/v17/bin/pg_config COPT='$(COPT)' \
 		FIND_TYPEDEF=$(ROOT_PROJECT_DIR)/vendor/postgres-v17/src/tools/find_typedef \
 		INDENT=$(POSTGRES_INSTALL_DIR)/build/v17/src/tools/pg_bsd_indent/pg_bsd_indent \
 		PGINDENT_SCRIPT=$(ROOT_PROJECT_DIR)/vendor/postgres-v17/src/tools/pgindent/pgindent \

compute_tools/src/compute.rs

@@ -1484,6 +1484,28 @@ LIMIT 100",
             info!("Pageserver config changed");
         }
     }
+
+    // Gather info about installed extensions
+    pub fn get_installed_extensions(&self) -> Result<()> {
+        let connstr = self.connstr.clone();
+
+        let rt = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("failed to create runtime");
+        let result = rt
+            .block_on(crate::installed_extensions::get_installed_extensions(
+                connstr,
+            ))
+            .expect("failed to get installed extensions");
+
+        info!(
+            "{}",
+            serde_json::to_string(&result).expect("failed to serialize extensions list")
+        );
+
+        Ok(())
+    }
 }
 
 pub fn forward_termination_signal() {

compute_tools/src/http/api.rs

@@ -165,6 +165,32 @@ async fn routes(req: Request<Body>, compute: &Arc<ComputeNode>) -> Response<Body
             }
         }
 
+        // get the list of installed extensions
+        // currently only used in python tests
+        // TODO: call it from cplane
+        (&Method::GET, "/installed_extensions") => {
+            info!("serving /installed_extensions GET request");
+            let status = compute.get_status();
+            if status != ComputeStatus::Running {
+                let msg = format!(
+                    "invalid compute status for extensions request: {:?}",
+                    status
+                );
+                error!(msg);
+                return Response::new(Body::from(msg));
+            }
+
+            let connstr = compute.connstr.clone();
+            let res = crate::installed_extensions::get_installed_extensions(connstr).await;
+            match res {
+                Ok(res) => render_json(Body::from(serde_json::to_string(&res).unwrap())),
+                Err(e) => render_json_error(
+                    &format!("could not get list of installed extensions: {}", e),
+                    StatusCode::INTERNAL_SERVER_ERROR,
+                ),
+            }
+        }
+
         // download extension files from remote extension storage on demand
         (&Method::POST, route) if route.starts_with("/extension_server/") => {
             info!("serving {:?} POST request", route);

compute_tools/src/http/openapi_spec.yaml

@@ -53,6 +53,20 @@ paths:
               schema:
                 $ref: "#/components/schemas/ComputeInsights"
 
+  /installed_extensions:
+    get:
+      tags:
+      - Info
+      summary: Get installed extensions.
+      description: ""
+      operationId: getInstalledExtensions
+      responses:
+        200:
+          description: List of installed extensions
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/InstalledExtensions"
   /info:
     get:
       tags:
@@ -395,6 +409,24 @@ components:
         - configuration
       example: running
 
+    InstalledExtensions:
+      type: object
+      properties:
+        extensions:
+          description: Contains list of installed extensions.
+          type: array
+          items:
+            type: object
+            properties:
+              extname:
+                type: string
+              versions:
+                type: array
+                items:
+                  type: string
+              n_databases:
+                type: integer
+
     #
     # Errors
     #

compute_tools/src/installed_extensions.rs

@@ -0,0 +1,80 @@
+use compute_api::responses::{InstalledExtension, InstalledExtensions};
+use std::collections::HashMap;
+use std::collections::HashSet;
+use url::Url;
+
+use anyhow::Result;
+use postgres::{Client, NoTls};
+use tokio::task;
+
+/// We don't reuse get_existing_dbs() just for code clarity
+/// and to make database listing query here more explicit.
+///
+/// Limit the number of databases to 500 to avoid excessive load.
+fn list_dbs(client: &mut Client) -> Result<Vec<String>> {
+    // `pg_database.datconnlimit = -2` means that the database is in the
+    // invalid state
+    let databases = client
+        .query(
+            "SELECT datname FROM pg_catalog.pg_database
+                WHERE datallowconn
+                AND datconnlimit <> - 2
+                LIMIT 500",
+            &[],
+        )?
+        .iter()
+        .map(|row| {
+            let db: String = row.get("datname");
+            db
+        })
+        .collect();
+
+    Ok(databases)
+}
+
+/// Connect to every database (see list_dbs above) and get the list of installed extensions.
+/// Same extension can be installed in multiple databases with different versions,
+/// we only keep the highest and lowest version across all databases.
+pub async fn get_installed_extensions(connstr: Url) -> Result<InstalledExtensions> {
+    let mut connstr = connstr.clone();
+
+    task::spawn_blocking(move || {
+        let mut client = Client::connect(connstr.as_str(), NoTls)?;
+        let databases: Vec<String> = list_dbs(&mut client)?;
+
+        let mut extensions_map: HashMap<String, InstalledExtension> = HashMap::new();
+        for db in databases.iter() {
+            connstr.set_path(db);
+            let mut db_client = Client::connect(connstr.as_str(), NoTls)?;
+            let extensions: Vec<(String, String)> = db_client
+                .query(
+                    "SELECT extname, extversion FROM pg_catalog.pg_extension;",
+                    &[],
+                )?
+                .iter()
+                .map(|row| (row.get("extname"), row.get("extversion")))
+                .collect();
+
+            for (extname, v) in extensions.iter() {
+                let version = v.to_string();
+                extensions_map
+                    .entry(extname.to_string())
+                    .and_modify(|e| {
+                        e.versions.insert(version.clone());
+                        // count the number of databases where the extension is installed
+                        e.n_databases += 1;
+                    })
+                    .or_insert(InstalledExtension {
+                        extname: extname.to_string(),
+                        versions: HashSet::from([version.clone()]),
+                        n_databases: 1,
+                    });
+            }
+        }
+
+        Ok(InstalledExtensions {
+            extensions: extensions_map.values().cloned().collect(),
+        })
+    })
+    .await?
+}

compute_tools/src/lib.rs

@@ -15,6 +15,7 @@ pub mod catalog;
 pub mod compute;
 pub mod disk_quota;
 pub mod extension_server;
+pub mod installed_extensions;
 pub mod local_proxy;
 pub mod lsn_lease;
 mod migration;

libs/compute_api/src/responses.rs

@@ -1,5 +1,6 @@
 //! Structs representing the JSON formats used in the compute_ctl's HTTP API.
 
+use std::collections::HashSet;
 use std::fmt::Display;
 
 use chrono::{DateTime, Utc};
@@ -155,3 +156,15 @@ pub enum ControlPlaneComputeStatus {
     // should be able to start with provided spec.
     Attached,
 }
+
+#[derive(Clone, Debug, Default, Serialize)]
+pub struct InstalledExtension {
+    pub extname: String,
+    pub versions: HashSet<String>,
+    pub n_databases: u32, // Number of databases using this extension
+}
+
+#[derive(Clone, Debug, Default, Serialize)]
+pub struct InstalledExtensions {
+    pub extensions: Vec<InstalledExtension>,
+}

libs/remote_storage/src/azure_blob.rs

@@ -496,26 +496,12 @@ impl RemoteStorage for AzureBlobStorage {
             builder = builder.if_match(IfMatchCondition::NotMatch(etag.to_string()))
         }
 
-        self.download_for_builder(builder, cancel).await
-    }
-
-    async fn download_byte_range(
-        &self,
-        from: &RemotePath,
-        start_inclusive: u64,
-        end_exclusive: Option<u64>,
-        cancel: &CancellationToken,
-    ) -> Result<Download, DownloadError> {
-        let blob_client = self.client.blob_client(self.relative_path_to_name(from));
-
-        let mut builder = blob_client.get();
-
-        let range: Range = if let Some(end_exclusive) = end_exclusive {
-            (start_inclusive..end_exclusive).into()
-        } else {
-            (start_inclusive..).into()
-        };
-        builder = builder.range(range);
+        if let Some((start, end)) = opts.byte_range() {
+            builder = builder.range(match end {
+                Some(end) => Range::Range(start..end),
+                None => Range::RangeFrom(start..),
+            });
+        }
 
         self.download_for_builder(builder, cancel).await
     }

libs/remote_storage/src/lib.rs

@@ -19,7 +19,8 @@ mod simulate_failures;
 mod support;
 
 use std::{
-    collections::HashMap, fmt::Debug, num::NonZeroU32, pin::Pin, sync::Arc, time::SystemTime,
+    collections::HashMap, fmt::Debug, num::NonZeroU32, ops::Bound, pin::Pin, sync::Arc,
+    time::SystemTime,
 };
 
 use anyhow::Context;
@@ -162,11 +163,60 @@ pub struct Listing {
 }
 
 /// Options for downloads. The default value is a plain GET.
-#[derive(Default)]
 pub struct DownloadOpts {
     /// If given, returns [`DownloadError::Unmodified`] if the object still has
     /// the same ETag (using If-None-Match).
     pub etag: Option<Etag>,
+    /// The start of the byte range to download, or unbounded.
+    pub byte_start: Bound<u64>,
+    /// The end of the byte range to download, or unbounded. Must be after the
+    /// start bound.
+    pub byte_end: Bound<u64>,
+}
+
+impl Default for DownloadOpts {
+    fn default() -> Self {
+        Self {
+            etag: Default::default(),
+            byte_start: Bound::Unbounded,
+            byte_end: Bound::Unbounded,
+        }
+    }
+}
+
+impl DownloadOpts {
+    /// Returns the byte range with inclusive start and exclusive end, or None
+    /// if unbounded.
+    pub fn byte_range(&self) -> Option<(u64, Option<u64>)> {
+        if self.byte_start == Bound::Unbounded && self.byte_end == Bound::Unbounded {
+            return None;
+        }
+        let start = match self.byte_start {
+            Bound::Excluded(i) => i + 1,
+            Bound::Included(i) => i,
+            Bound::Unbounded => 0,
+        };
+        let end = match self.byte_end {
+            Bound::Excluded(i) => Some(i),
+            Bound::Included(i) => Some(i + 1),
+            Bound::Unbounded => None,
+        };
+        if let Some(end) = end {
+            assert!(start < end, "range end {end} at or before start {start}");
+        }
+        Some((start, end))
+    }
+
+    /// Returns the byte range as an RFC 2616 Range header value with inclusive
+    /// bounds, or None if unbounded.
+    pub fn byte_range_header(&self) -> Option<String> {
+        self.byte_range()
+            .map(|(start, end)| (start, end.map(|end| end - 1))) // make end inclusive
+            .map(|(start, end)| match end {
+                Some(end) => format!("bytes={start}-{end}"),
+                None => format!("bytes={start}-"),
+            })
+    }
 }
 
 /// Storage (potentially remote) API to manage its state.
@@ -257,21 +307,6 @@ pub trait RemoteStorage: Send + Sync + 'static {
         cancel: &CancellationToken,
     ) -> Result<Download, DownloadError>;
 
-    /// Streams a given byte range of the remote storage entry contents.
-    ///
-    /// The returned download stream will obey initial timeout and cancellation signal by erroring
-    /// on whichever happens first. Only one of the reasons will fail the stream, which is usually
-    /// enough for `tokio::io::copy_buf` usage. If needed the error can be filtered out.
-    ///
-    /// Returns the metadata, if any was stored with the file previously.
-    async fn download_byte_range(
-        &self,
-        from: &RemotePath,
-        start_inclusive: u64,
-        end_exclusive: Option<u64>,
-        cancel: &CancellationToken,
-    ) -> Result<Download, DownloadError>;
-
     /// Delete a single path from remote storage.
     ///
     /// If the operation fails because of timeout or cancellation, the root cause of the error will be
@@ -425,33 +460,6 @@ impl<Other: RemoteStorage> GenericRemoteStorage<Arc<Other>> {
         }
     }
 
-    pub async fn download_byte_range(
-        &self,
-        from: &RemotePath,
-        start_inclusive: u64,
-        end_exclusive: Option<u64>,
-        cancel: &CancellationToken,
-    ) -> Result<Download, DownloadError> {
-        match self {
-            Self::LocalFs(s) => {
-                s.download_byte_range(from, start_inclusive, end_exclusive, cancel)
-                    .await
-            }
-            Self::AwsS3(s) => {
-                s.download_byte_range(from, start_inclusive, end_exclusive, cancel)
-                    .await
-            }
-            Self::AzureBlob(s) => {
-                s.download_byte_range(from, start_inclusive, end_exclusive, cancel)
-                    .await
-            }
-            Self::Unreliable(s) => {
-                s.download_byte_range(from, start_inclusive, end_exclusive, cancel)
-                    .await
-            }
-        }
-    }
-
     /// See [`RemoteStorage::delete`]
     pub async fn delete(
         &self,
@@ -573,20 +581,6 @@ impl GenericRemoteStorage {
             })
     }
 
-    /// Downloads the storage object into the `to_path` provided.
-    /// `byte_range` could be specified to dowload only a part of the file, if needed.
-    pub async fn download_storage_object(
-        &self,
-        byte_range: Option<(u64, Option<u64>)>,
-        from: &RemotePath,
-        cancel: &CancellationToken,
-    ) -> Result<Download, DownloadError> {
-        match byte_range {
-            Some((start, end)) => self.download_byte_range(from, start, end, cancel).await,
-            None => self.download(from, &DownloadOpts::default(), cancel).await,
-        }
-    }
-
     /// The name of the bucket/container/etc.
     pub fn bucket_name(&self) -> Option<&str> {
         match self {
@@ -660,6 +654,76 @@ impl ConcurrencyLimiter {
 mod tests {
     use super::*;
 
+    /// DownloadOpts::byte_range() should generate (inclusive, exclusive) ranges
+    /// with optional end bound, or None when unbounded.
+    #[test]
+    fn download_opts_byte_range() {
+        // Consider using test_case or a similar table-driven test framework.
+        let cases = [
+            // (byte_start, byte_end, expected)
+            (Bound::Unbounded, Bound::Unbounded, None),
+            (Bound::Unbounded, Bound::Included(7), Some((0, Some(8)))),
+            (Bound::Unbounded, Bound::Excluded(7), Some((0, Some(7)))),
+            (Bound::Included(3), Bound::Unbounded, Some((3, None))),
+            (Bound::Included(3), Bound::Included(7), Some((3, Some(8)))),
+            (Bound::Included(3), Bound::Excluded(7), Some((3, Some(7)))),
+            (Bound::Excluded(3), Bound::Unbounded, Some((4, None))),
+            (Bound::Excluded(3), Bound::Included(7), Some((4, Some(8)))),
+            (Bound::Excluded(3), Bound::Excluded(7), Some((4, Some(7)))),
+            // 1-sized ranges are fine, 0 aren't and will panic (separate test).
+            (Bound::Included(3), Bound::Included(3), Some((3, Some(4)))),
+            (Bound::Included(3), Bound::Excluded(4), Some((3, Some(4)))),
+        ];
+
+        for (byte_start, byte_end, expect) in cases {
+            let opts = DownloadOpts {
+                byte_start,
+                byte_end,
+                ..Default::default()
+            };
+            let result = opts.byte_range();
+            assert_eq!(
+                result, expect,
+                "byte_start={byte_start:?} byte_end={byte_end:?}"
+            );
+
+            // Check generated HTTP header, which uses an inclusive range.
+            let expect_header = expect.map(|(start, end)| match end {
+                Some(end) => format!("bytes={start}-{}", end - 1), // inclusive end
+                None => format!("bytes={start}-"),
+            });
+            assert_eq!(
+                opts.byte_range_header(),
+                expect_header,
+                "byte_start={byte_start:?} byte_end={byte_end:?}"
+            );
+        }
+    }
+
+    /// DownloadOpts::byte_range() zero-sized byte range should panic.
+    #[test]
+    #[should_panic]
+    fn download_opts_byte_range_zero() {
+        DownloadOpts {
+            byte_start: Bound::Included(3),
+            byte_end: Bound::Excluded(3),
+            ..Default::default()
+        }
+        .byte_range();
+    }
+
+    /// DownloadOpts::byte_range() negative byte range should panic.
+    #[test]
+    #[should_panic]
+    fn download_opts_byte_range_negative() {
+        DownloadOpts {
+            byte_start: Bound::Included(3),
+            byte_end: Bound::Included(2),
+            ..Default::default()
+        }
+        .byte_range();
+    }
+
     #[test]
     fn test_object_name() {
         let k = RemotePath::new(Utf8Path::new("a/b/c")).unwrap();

libs/remote_storage/src/local_fs.rs

@@ -506,54 +506,7 @@ impl RemoteStorage for LocalFs {
             return Err(DownloadError::Unmodified);
         }
 
-        let source = ReaderStream::new(
-            fs::OpenOptions::new()
-                .read(true)
-                .open(&target_path)
-                .await
-                .with_context(|| {
-                    format!("Failed to open source file {target_path:?} to use in the download")
-                })
-                .map_err(DownloadError::Other)?,
-        );
-
-        let metadata = self
-            .read_storage_metadata(&target_path)
-            .await
-            .map_err(DownloadError::Other)?;
-
-        let cancel_or_timeout = crate::support::cancel_or_timeout(self.timeout, cancel.clone());
-        let source = crate::support::DownloadStream::new(cancel_or_timeout, source);
-
-        Ok(Download {
-            metadata,
-            last_modified: file_metadata
-                .modified()
-                .map_err(|e| DownloadError::Other(anyhow::anyhow!(e).context("Reading mtime")))?,
-            etag,
-            download_stream: Box::pin(source),
-        })
-    }
-
-    async fn download_byte_range(
-        &self,
-        from: &RemotePath,
-        start_inclusive: u64,
-        end_exclusive: Option<u64>,
-        cancel: &CancellationToken,
-    ) -> Result<Download, DownloadError> {
-        if let Some(end_exclusive) = end_exclusive {
-            if end_exclusive <= start_inclusive {
-                return Err(DownloadError::Other(anyhow::anyhow!("Invalid range, start ({start_inclusive}) is not less than end_exclusive ({end_exclusive:?})")));
-            };
-            if start_inclusive == end_exclusive.saturating_sub(1) {
-                return Err(DownloadError::Other(anyhow::anyhow!("Invalid range, start ({start_inclusive}) and end_exclusive ({end_exclusive:?}) difference is zero bytes")));
-            }
-        }
-
-        let target_path = from.with_base(&self.storage_root);
-        let file_metadata = file_metadata(&target_path).await?;
-        let mut source = tokio::fs::OpenOptions::new()
+        let mut file = fs::OpenOptions::new()
             .read(true)
             .open(&target_path)
             .await
@@ -562,31 +515,29 @@ impl RemoteStorage for LocalFs {
             })
             .map_err(DownloadError::Other)?;
 
-        let len = source
-            .metadata()
-            .await
-            .context("query file length")
-            .map_err(DownloadError::Other)?
-            .len();
+        let mut take = file_metadata.len();
+        if let Some((start, end)) = opts.byte_range() {
+            if start > 0 {
+                file.seek(io::SeekFrom::Start(start))
+                    .await
+                    .context("Failed to seek to the range start in a local storage file")
+                    .map_err(DownloadError::Other)?;
+            }
+            if let Some(end) = end {
+                take = end - start;
+            }
+        }
 
-        source
-            .seek(io::SeekFrom::Start(start_inclusive))
-            .await
-            .context("Failed to seek to the range start in a local storage file")
-            .map_err(DownloadError::Other)?;
+        let source = ReaderStream::new(file.take(take));
 
         let metadata = self
             .read_storage_metadata(&target_path)
             .await
             .map_err(DownloadError::Other)?;
 
-        let source = source.take(end_exclusive.unwrap_or(len) - start_inclusive);
-        let source = ReaderStream::new(source);
-
         let cancel_or_timeout = crate::support::cancel_or_timeout(self.timeout, cancel.clone());
         let source = crate::support::DownloadStream::new(cancel_or_timeout, source);
 
-        let etag = mock_etag(&file_metadata);
         Ok(Download {
             metadata,
             last_modified: file_metadata
@@ -688,7 +639,7 @@ mod fs_tests {
     use super::*;
 
     use camino_tempfile::tempdir;
-    use std::{collections::HashMap, io::Write};
+    use std::{collections::HashMap, io::Write, ops::Bound};
 
     async fn read_and_check_metadata(
         storage: &LocalFs,
@@ -804,10 +755,12 @@ mod fs_tests {
         let (first_part_local, second_part_local) = uploaded_bytes.split_at(3);
 
         let first_part_download = storage
-            .download_byte_range(
+            .download(
                 &upload_target,
-                0,
-                Some(first_part_local.len() as u64),
+                &DownloadOpts {
+                    byte_end: Bound::Excluded(first_part_local.len() as u64),
+                    ..Default::default()
+                },
                 &cancel,
             )
             .await?;
@@ -823,10 +776,15 @@ mod fs_tests {
         );
 
         let second_part_download = storage
-            .download_byte_range(
+            .download(
                 &upload_target,
-                first_part_local.len() as u64,
-                Some((first_part_local.len() + second_part_local.len()) as u64),
+                &DownloadOpts {
+                    byte_start: Bound::Included(first_part_local.len() as u64),
+                    byte_end: Bound::Excluded(
+                        (first_part_local.len() + second_part_local.len()) as u64,
+                    ),
+                    ..Default::default()
+                },
                 &cancel,
             )
             .await?;
@@ -842,7 +800,14 @@ mod fs_tests {
         );
 
         let suffix_bytes = storage
-            .download_byte_range(&upload_target, 13, None, &cancel)
+            .download(
+                &upload_target,
+                &DownloadOpts {
+                    byte_start: Bound::Included(13),
+                    ..Default::default()
+                },
+                &cancel,
+            )
             .await?
             .download_stream;
         let suffix_bytes = aggregate(suffix_bytes).await?;
@@ -850,7 +815,7 @@ mod fs_tests {
         assert_eq!(upload_name, suffix);
 
         let all_bytes = storage
-            .download_byte_range(&upload_target, 0, None, &cancel)
+            .download(&upload_target, &DownloadOpts::default(), &cancel)
             .await?
             .download_stream;
         let all_bytes = aggregate(all_bytes).await?;
@@ -861,48 +826,26 @@ mod fs_tests {
     }
 
     #[tokio::test]
-    async fn download_file_range_negative() -> anyhow::Result<()> {
-        let (storage, cancel) = create_storage()?;
+    #[should_panic(expected = "at or before start")]
+    async fn download_file_range_negative() {
+        let (storage, cancel) = create_storage().unwrap();
         let upload_name = "upload_1";
-        let upload_target = upload_dummy_file(&storage, upload_name, None, &cancel).await?;
+        let upload_target = upload_dummy_file(&storage, upload_name, None, &cancel)
+            .await
+            .unwrap();
 
-        let start = 1_000_000_000;
-        let end = start + 1;
-        match storage
-            .download_byte_range(
+        storage
+            .download(
                 &upload_target,
-                start,
-                Some(end), // exclusive end
+                &DownloadOpts {
+                    byte_start: Bound::Included(10),
+                    byte_end: Bound::Excluded(10),
+                    ..Default::default()
+                },
                 &cancel,
             )
             .await
-        {
-            Ok(_) => panic!("Should not allow downloading wrong ranges"),
-            Err(e) => {
-                let error_string = e.to_string();
-                assert!(error_string.contains("zero bytes"));
-                assert!(error_string.contains(&start.to_string()));
-                assert!(error_string.contains(&end.to_string()));
-            }
-        }
-
-        let start = 10000;
-        let end = 234;
-        assert!(start > end, "Should test an incorrect range");
-        match storage
-            .download_byte_range(&upload_target, start, Some(end), &cancel)
-            .await
-        {
-            Ok(_) => panic!("Should not allow downloading wrong ranges"),
-            Err(e) => {
-                let error_string = e.to_string();
-                assert!(error_string.contains("Invalid range"));
-                assert!(error_string.contains(&start.to_string()));
-                assert!(error_string.contains(&end.to_string()));
-            }
-        }
-
-        Ok(())
+            .unwrap();
     }
 
     #[tokio::test]
@@ -945,10 +888,12 @@ mod fs_tests {
         let (first_part_local, _) = uploaded_bytes.split_at(3);
 
         let partial_download_with_metadata = storage
-            .download_byte_range(
+            .download(
                 &upload_target,
-                0,
-                Some(first_part_local.len() as u64),
+                &DownloadOpts {
+                    byte_end: Bound::Excluded(first_part_local.len() as u64),
+                    ..Default::default()
+                },
                 &cancel,
             )
             .await?;

libs/remote_storage/src/s3_bucket.rs

@@ -804,34 +804,7 @@ impl RemoteStorage for S3Bucket {
                 bucket: self.bucket_name.clone(),
                 key: self.relative_path_to_s3_object(from),
                 etag: opts.etag.as_ref().map(|e| e.to_string()),
-                range: None,
-            },
-            cancel,
-        )
-        .await
-    }
-
-    async fn download_byte_range(
-        &self,
-        from: &RemotePath,
-        start_inclusive: u64,
-        end_exclusive: Option<u64>,
-        cancel: &CancellationToken,
-    ) -> Result<Download, DownloadError> {
-        // S3 accepts ranges as https://www.w3.org/Protocols/rfc2616/rfc2616-sec14.html#sec14.35
-        // and needs both ends to be exclusive
-        let end_inclusive = end_exclusive.map(|end| end.saturating_sub(1));
-        let range = Some(match end_inclusive {
-            Some(end_inclusive) => format!("bytes={start_inclusive}-{end_inclusive}"),
-            None => format!("bytes={start_inclusive}-"),
-        });
-
-        self.download_object(
-            GetObjectRequest {
-                bucket: self.bucket_name.clone(),
-                key: self.relative_path_to_s3_object(from),
-                etag: None,
-                range,
+                range: opts.byte_range_header(),
             },
             cancel,
         )

libs/remote_storage/src/simulate_failures.rs

@@ -170,28 +170,13 @@ impl RemoteStorage for UnreliableWrapper {
         opts: &DownloadOpts,
         cancel: &CancellationToken,
     ) -> Result<Download, DownloadError> {
+        // Note: We treat any byte range as an "attempt" of the same operation.
+        // We don't pay attention to the ranges. That's good enough for now.
         self.attempt(RemoteOp::Download(from.clone()))
             .map_err(DownloadError::Other)?;
         self.inner.download(from, opts, cancel).await
     }
 
-    async fn download_byte_range(
-        &self,
-        from: &RemotePath,
-        start_inclusive: u64,
-        end_exclusive: Option<u64>,
-        cancel: &CancellationToken,
-    ) -> Result<Download, DownloadError> {
-        // Note: We treat any download_byte_range as an "attempt" of the same
-        // operation. We don't pay attention to the ranges. That's good enough
-        // for now.
-        self.attempt(RemoteOp::Download(from.clone()))
-            .map_err(DownloadError::Other)?;
-        self.inner
-            .download_byte_range(from, start_inclusive, end_exclusive, cancel)
-            .await
-    }
-
     async fn delete(&self, path: &RemotePath, cancel: &CancellationToken) -> anyhow::Result<()> {
         self.delete_inner(path, true, cancel).await
     }

libs/remote_storage/tests/common/tests.rs

@@ -2,6 +2,7 @@ use anyhow::Context;
 use camino::Utf8Path;
 use futures::StreamExt;
 use remote_storage::{DownloadError, DownloadOpts, ListingMode, ListingObject, RemotePath};
+use std::ops::Bound;
 use std::sync::Arc;
 use std::{collections::HashSet, num::NonZeroU32};
 use test_context::test_context;
@@ -293,7 +294,15 @@ async fn upload_download_works(ctx: &mut MaybeEnabledStorage) -> anyhow::Result<
     // Full range (end specified)
     let dl = ctx
         .client
-        .download_byte_range(&path, 0, Some(len as u64), &cancel)
+        .download(
+            &path,
+            &DownloadOpts {
+                byte_start: Bound::Included(0),
+                byte_end: Bound::Excluded(len as u64),
+                ..Default::default()
+            },
+            &cancel,
+        )
         .await?;
     let buf = download_to_vec(dl).await?;
     assert_eq!(&buf, &orig);
@@ -301,7 +310,15 @@ async fn upload_download_works(ctx: &mut MaybeEnabledStorage) -> anyhow::Result<
     // partial range (end specified)
     let dl = ctx
         .client
-        .download_byte_range(&path, 4, Some(10), &cancel)
+        .download(
+            &path,
+            &DownloadOpts {
+                byte_start: Bound::Included(4),
+                byte_end: Bound::Excluded(10),
+                ..Default::default()
+            },
+            &cancel,
+        )
         .await?;
     let buf = download_to_vec(dl).await?;
     assert_eq!(&buf, &orig[4..10]);
@@ -309,7 +326,15 @@ async fn upload_download_works(ctx: &mut MaybeEnabledStorage) -> anyhow::Result<
     // partial range (end beyond real end)
     let dl = ctx
         .client
-        .download_byte_range(&path, 8, Some(len as u64 * 100), &cancel)
+        .download(
+            &path,
+            &DownloadOpts {
+                byte_start: Bound::Included(8),
+                byte_end: Bound::Excluded(len as u64 * 100),
+                ..Default::default()
+            },
+            &cancel,
+        )
         .await?;
     let buf = download_to_vec(dl).await?;
     assert_eq!(&buf, &orig[8..]);
@@ -317,7 +342,14 @@ async fn upload_download_works(ctx: &mut MaybeEnabledStorage) -> anyhow::Result<
     // Partial range (end unspecified)
     let dl = ctx
         .client
-        .download_byte_range(&path, 4, None, &cancel)
+        .download(
+            &path,
+            &DownloadOpts {
+                byte_start: Bound::Included(4),
+                ..Default::default()
+            },
+            &cancel,
+        )
         .await?;
     let buf = download_to_vec(dl).await?;
     assert_eq!(&buf, &orig[4..]);
@@ -325,7 +357,14 @@ async fn upload_download_works(ctx: &mut MaybeEnabledStorage) -> anyhow::Result<
     // Full range (end unspecified)
     let dl = ctx
         .client
-        .download_byte_range(&path, 0, None, &cancel)
+        .download(
+            &path,
+            &DownloadOpts {
+                byte_start: Bound::Included(0),
+                ..Default::default()
+            },
+            &cancel,
+        )
         .await?;
     let buf = download_to_vec(dl).await?;
     assert_eq!(&buf, &orig);

libs/vm_monitor/src/runner.rs

@@ -79,8 +79,7 @@ pub struct Config {
     /// memory.
     ///
     /// The default value of `0.15` means that we *guarantee* sending upscale requests if the
-    /// cgroup is using more than 85% of total memory (even if we're *not* separately reserving
-    /// memory for the file cache).
+    /// cgroup is using more than 85% of total memory.
     cgroup_min_overhead_fraction: f64,
 
     cgroup_downscale_threshold_buffer_bytes: u64,
@@ -97,24 +96,12 @@ impl Default for Config {
 }
 
 impl Config {
-    fn cgroup_threshold(&self, total_mem: u64, file_cache_disk_size: u64) -> u64 {
-        // If the file cache is in tmpfs, then it will count towards shmem usage of the cgroup,
-        // and thus be non-reclaimable, so we should allow for additional memory usage.
-        //
-        // If the file cache sits on disk, our desired stable system state is for it to be fully
-        // page cached (its contents should only be paged to/from disk in situations where we can't
-        // upscale fast enough). Page-cached memory is reclaimable, so we need to lower the
-        // threshold for non-reclaimable memory so we scale up *before* the kernel starts paging
-        // out the file cache.
-        let memory_remaining_for_cgroup = total_mem.saturating_sub(file_cache_disk_size);
-
-        // Even if we're not separately making room for the file cache (if it's in tmpfs), we still
-        // want our threshold to be met gracefully instead of letting postgres get OOM-killed.
+    fn cgroup_threshold(&self, total_mem: u64) -> u64 {
+        // We want our threshold to be met gracefully instead of letting postgres get OOM-killed
+        // (or if there's room, spilling to swap).
         // So we guarantee that there's at least `cgroup_min_overhead_fraction` of total memory
         // remaining above the threshold.
-        let max_threshold = (total_mem as f64 * (1.0 - self.cgroup_min_overhead_fraction)) as u64;
-
-        memory_remaining_for_cgroup.min(max_threshold)
+        (total_mem as f64 * (1.0 - self.cgroup_min_overhead_fraction)) as u64
     }
 }
 
@@ -149,11 +136,6 @@ impl Runner {
 
         let mem = get_total_system_memory();
 
-        let mut file_cache_disk_size = 0;
-
-        // We need to process file cache initialization before cgroup initialization, so that the memory
-        // allocated to the file cache is appropriately taken into account when we decide the cgroup's
-        // memory limits.
         if let Some(connstr) = &args.pgconnstr {
             info!("initializing file cache");
             let config = FileCacheConfig::default();
@@ -184,7 +166,6 @@ impl Runner {
                 info!("file cache size actually got set to {actual_size}")
             }
 
-            file_cache_disk_size = actual_size;
             state.filecache = Some(file_cache);
         }
 
@@ -207,7 +188,7 @@ impl Runner {
                 cgroup.watch(hist_tx).await
             });
 
-            let threshold = state.config.cgroup_threshold(mem, file_cache_disk_size);
+            let threshold = state.config.cgroup_threshold(mem);
             info!(threshold, "set initial cgroup threshold",);
 
             state.cgroup = Some(CgroupState {
@@ -259,9 +240,7 @@ impl Runner {
                 return Ok((false, status.to_owned()));
             }
 
-            let new_threshold = self
-                .config
-                .cgroup_threshold(usable_system_memory, expected_file_cache_size);
+            let new_threshold = self.config.cgroup_threshold(usable_system_memory);
 
             let current = last_history.avg_non_reclaimable;
 
@@ -282,13 +261,11 @@ impl Runner {
 
         // The downscaling has been approved. Downscale the file cache, then the cgroup.
         let mut status = vec![];
-        let mut file_cache_disk_size = 0;
         if let Some(file_cache) = &mut self.filecache {
             let actual_usage = file_cache
                 .set_file_cache_size(expected_file_cache_size)
                 .await
                 .context("failed to set file cache size")?;
-            file_cache_disk_size = actual_usage;
             let message = format!(
                 "set file cache size to {} MiB",
                 bytes_to_mebibytes(actual_usage),
@@ -298,9 +275,7 @@ impl Runner {
         }
 
         if let Some(cgroup) = &mut self.cgroup {
-            let new_threshold = self
-                .config
-                .cgroup_threshold(usable_system_memory, file_cache_disk_size);
+            let new_threshold = self.config.cgroup_threshold(usable_system_memory);
 
             let message = format!(
                 "set cgroup memory threshold from {} MiB to {} MiB, of new total {} MiB",
@@ -329,7 +304,6 @@ impl Runner {
         let new_mem = resources.mem;
         let usable_system_memory = new_mem.saturating_sub(self.config.sys_buffer_bytes);
 
-        let mut file_cache_disk_size = 0;
         if let Some(file_cache) = &mut self.filecache {
             let expected_usage = file_cache.config.calculate_cache_size(usable_system_memory);
             info!(
@@ -342,7 +316,6 @@ impl Runner {
                 .set_file_cache_size(expected_usage)
                 .await
                 .context("failed to set file cache size")?;
-            file_cache_disk_size = actual_usage;
 
             if actual_usage != expected_usage {
                 warn!(
@@ -354,9 +327,7 @@ impl Runner {
         }
 
         if let Some(cgroup) = &mut self.cgroup {
-            let new_threshold = self
-                .config
-                .cgroup_threshold(usable_system_memory, file_cache_disk_size);
+            let new_threshold = self.config.cgroup_threshold(usable_system_memory);
 
             info!(
                 "set cgroup memory threshold from {} MiB to {} MiB of new total {} MiB",

pageserver/src/http/routes.rs

@@ -704,6 +704,8 @@ async fn timeline_archival_config_handler(
     let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
     let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;
 
+    let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Warn);
+
     let request_data: TimelineArchivalConfigRequest = json_request(&mut request).await?;
     check_permission(&request, Some(tenant_shard_id.tenant_id))?;
     let state = get_state(&request);
@@ -714,7 +716,7 @@ async fn timeline_archival_config_handler(
             .get_attached_tenant_shard(tenant_shard_id)?;
 
         tenant
-            .apply_timeline_archival_config(timeline_id, request_data.state)
+            .apply_timeline_archival_config(timeline_id, request_data.state, ctx)
             .await?;
         Ok::<_, ApiError>(())
     }

pageserver/src/tenant.rs

@@ -38,6 +38,7 @@ use std::future::Future;
 use std::sync::Weak;
 use std::time::SystemTime;
 use storage_broker::BrokerClientChannel;
+use timeline::offload::offload_timeline;
 use tokio::io::BufReader;
 use tokio::sync::watch;
 use tokio::task::JoinSet;
@@ -287,9 +288,13 @@ pub struct Tenant {
 
     /// During timeline creation, we first insert the TimelineId to the
     /// creating map, then `timelines`, then remove it from the creating map.
-    /// **Lock order**: if acquring both, acquire`timelines` before `timelines_creating`
+    /// **Lock order**: if acquiring both, acquire`timelines` before `timelines_creating`
     timelines_creating: std::sync::Mutex<HashSet<TimelineId>>,
 
+    /// Possibly offloaded and archived timelines
+    /// **Lock order**: if acquiring both, acquire`timelines` before `timelines_offloaded`
+    timelines_offloaded: Mutex<HashMap<TimelineId, Arc<OffloadedTimeline>>>,
+
     // This mutex prevents creation of new timelines during GC.
     // Adding yet another mutex (in addition to `timelines`) is needed because holding
     // `timelines` mutex during all GC iteration
@@ -484,6 +489,65 @@ impl WalRedoManager {
     }
 }
 
+pub struct OffloadedTimeline {
+    pub tenant_shard_id: TenantShardId,
+    pub timeline_id: TimelineId,
+    pub ancestor_timeline_id: Option<TimelineId>,
+
+    // TODO: once we persist offloaded state, make this lazily constructed
+    pub remote_client: Arc<RemoteTimelineClient>,
+
+    /// Prevent two tasks from deleting the timeline at the same time. If held, the
+    /// timeline is being deleted. If 'true', the timeline has already been deleted.
+    pub delete_progress: Arc<tokio::sync::Mutex<DeleteTimelineFlow>>,
+}
+
+impl OffloadedTimeline {
+    fn from_timeline(timeline: &Timeline) -> Self {
+        Self {
+            tenant_shard_id: timeline.tenant_shard_id,
+            timeline_id: timeline.timeline_id,
+            ancestor_timeline_id: timeline.get_ancestor_timeline_id(),
+
+            remote_client: timeline.remote_client.clone(),
+            delete_progress: timeline.delete_progress.clone(),
+        }
+    }
+}
+
+#[derive(Clone)]
+pub enum TimelineOrOffloaded {
+    Timeline(Arc<Timeline>),
+    Offloaded(Arc<OffloadedTimeline>),
+}
+
+impl TimelineOrOffloaded {
+    pub fn tenant_shard_id(&self) -> TenantShardId {
+        match self {
+            TimelineOrOffloaded::Timeline(timeline) => timeline.tenant_shard_id,
+            TimelineOrOffloaded::Offloaded(offloaded) => offloaded.tenant_shard_id,
+        }
+    }
+    pub fn timeline_id(&self) -> TimelineId {
+        match self {
+            TimelineOrOffloaded::Timeline(timeline) => timeline.timeline_id,
+            TimelineOrOffloaded::Offloaded(offloaded) => offloaded.timeline_id,
+        }
+    }
+    pub fn delete_progress(&self) -> &Arc<tokio::sync::Mutex<DeleteTimelineFlow>> {
+        match self {
+            TimelineOrOffloaded::Timeline(timeline) => &timeline.delete_progress,
+            TimelineOrOffloaded::Offloaded(offloaded) => &offloaded.delete_progress,
+        }
+    }
+    pub fn remote_client(&self) -> &Arc<RemoteTimelineClient> {
+        match self {
+            TimelineOrOffloaded::Timeline(timeline) => &timeline.remote_client,
+            TimelineOrOffloaded::Offloaded(offloaded) => &offloaded.remote_client,
+        }
+    }
+}
+
 #[derive(Debug, thiserror::Error, PartialEq, Eq)]
 pub enum GetTimelineError {
     #[error("Timeline is shutting down")]
@@ -1406,52 +1470,192 @@ impl Tenant {
         }
     }
 
-    pub(crate) async fn apply_timeline_archival_config(
-        &self,
+    fn check_to_be_archived_has_no_unarchived_children(
         timeline_id: TimelineId,
-        state: TimelineArchivalState,
+        timelines: &std::sync::MutexGuard<'_, HashMap<TimelineId, Arc<Timeline>>>,
+    ) -> Result<(), TimelineArchivalError> {
+        let children: Vec<TimelineId> = timelines
+            .iter()
+            .filter_map(|(id, entry)| {
+                if entry.get_ancestor_timeline_id() != Some(timeline_id) {
+                    return None;
+                }
+                if entry.is_archived() == Some(true) {
+                    return None;
+                }
+                Some(*id)
+            })
+            .collect();
+
+        if !children.is_empty() {
+            return Err(TimelineArchivalError::HasUnarchivedChildren(children));
+        }
+        Ok(())
+    }
+
+    fn check_ancestor_of_to_be_unarchived_is_not_archived(
+        ancestor_timeline_id: TimelineId,
+        timelines: &std::sync::MutexGuard<'_, HashMap<TimelineId, Arc<Timeline>>>,
+        offloaded_timelines: &std::sync::MutexGuard<
+            '_,
+            HashMap<TimelineId, Arc<OffloadedTimeline>>,
+        >,
+    ) -> Result<(), TimelineArchivalError> {
+        let has_archived_parent =
+            if let Some(ancestor_timeline) = timelines.get(&ancestor_timeline_id) {
+                ancestor_timeline.is_archived() == Some(true)
+            } else if offloaded_timelines.contains_key(&ancestor_timeline_id) {
+                true
+            } else {
+                error!("ancestor timeline {ancestor_timeline_id} not found");
+                if cfg!(debug_assertions) {
+                    panic!("ancestor timeline {ancestor_timeline_id} not found");
+                }
+                return Err(TimelineArchivalError::NotFound);
+            };
+        if has_archived_parent {
+            return Err(TimelineArchivalError::HasArchivedParent(
+                ancestor_timeline_id,
+            ));
+        }
+        Ok(())
+    }
+
+    fn check_to_be_unarchived_timeline_has_no_archived_parent(
+        timeline: &Arc<Timeline>,
+    ) -> Result<(), TimelineArchivalError> {
+        if let Some(ancestor_timeline) = timeline.ancestor_timeline() {
+            if ancestor_timeline.is_archived() == Some(true) {
+                return Err(TimelineArchivalError::HasArchivedParent(
+                    ancestor_timeline.timeline_id,
+                ));
+            }
+        }
+        Ok(())
+    }
+
+    /// Loads the specified (offloaded) timeline from S3 and attaches it as a loaded timeline
+    async fn unoffload_timeline(
+        self: &Arc<Self>,
+        timeline_id: TimelineId,
+        ctx: RequestContext,
+    ) -> Result<Arc<Timeline>, TimelineArchivalError> {
+        let cancel = self.cancel.clone();
+        let timeline_preload = self
+            .load_timeline_metadata(timeline_id, self.remote_storage.clone(), cancel)
+            .await;
+
+        let index_part = match timeline_preload.index_part {
+            Ok(index_part) => {
+                debug!("remote index part exists for timeline {timeline_id}");
+                index_part
+            }
+            Err(DownloadError::NotFound) => {
+                error!(%timeline_id, "index_part not found on remote");
+                return Err(TimelineArchivalError::NotFound);
+            }
+            Err(e) => {
+                // Some (possibly ephemeral) error happened during index_part download.
+                warn!(%timeline_id, "Failed to load index_part from remote storage, failed creation? ({e})");
+                return Err(TimelineArchivalError::Other(
+                    anyhow::Error::new(e).context("downloading index_part from remote storage"),
+                ));
+            }
+        };
+        let index_part = match index_part {
+            MaybeDeletedIndexPart::IndexPart(index_part) => index_part,
+            MaybeDeletedIndexPart::Deleted(_index_part) => {
+                info!("timeline is deleted according to index_part.json");
+                return Err(TimelineArchivalError::NotFound);
+            }
+        };
+        let remote_metadata = index_part.metadata.clone();
+        let timeline_resources = self.build_timeline_resources(timeline_id);
+        self.load_remote_timeline(
+            timeline_id,
+            index_part,
+            remote_metadata,
+            timeline_resources,
+            &ctx,
+        )
+        .await
+        .with_context(|| {
+            format!(
+                "failed to load remote timeline {} for tenant {}",
+                timeline_id, self.tenant_shard_id
+            )
+        })?;
+        let timelines = self.timelines.lock().unwrap();
+        if let Some(timeline) = timelines.get(&timeline_id) {
+            let mut offloaded_timelines = self.timelines_offloaded.lock().unwrap();
+            if offloaded_timelines.remove(&timeline_id).is_none() {
+                warn!("timeline already removed from offloaded timelines");
+            }
+            Ok(Arc::clone(timeline))
+        } else {
+            warn!("timeline not available directly after attach");
+            Err(TimelineArchivalError::Other(anyhow::anyhow!(
+                "timeline not available directly after attach"
+            )))
+        }
+    }
+
+    pub(crate) async fn apply_timeline_archival_config(
+        self: &Arc<Self>,
+        timeline_id: TimelineId,
+        new_state: TimelineArchivalState,
+        ctx: RequestContext,
     ) -> Result<(), TimelineArchivalError> {
         info!("setting timeline archival config");
-        let timeline = {
+        // First part: figure out what is needed to do, and do validation
+        let timeline_or_unarchive_offloaded = 'outer: {
             let timelines = self.timelines.lock().unwrap();
 
             let Some(timeline) = timelines.get(&timeline_id) else {
-                return Err(TimelineArchivalError::NotFound);
+                let offloaded_timelines = self.timelines_offloaded.lock().unwrap();
+                let Some(offloaded) = offloaded_timelines.get(&timeline_id) else {
+                    return Err(TimelineArchivalError::NotFound);
+                };
+                if new_state == TimelineArchivalState::Archived {
+                    // It's offloaded already, so nothing to do
+                    return Ok(());
+                }
+                if let Some(ancestor_timeline_id) = offloaded.ancestor_timeline_id {
+                    Self::check_ancestor_of_to_be_unarchived_is_not_archived(
+                        ancestor_timeline_id,
+                        &timelines,
+                        &offloaded_timelines,
+                    )?;
+                }
+                break 'outer None;
             };
 
-            if state == TimelineArchivalState::Unarchived {
-                if let Some(ancestor_timeline) = timeline.ancestor_timeline() {
-                    if ancestor_timeline.is_archived() == Some(true) {
-                        return Err(TimelineArchivalError::HasArchivedParent(
-                            ancestor_timeline.timeline_id,
-                        ));
-                    }
+            // Do some validation. We release the timelines lock below, so there is potential
+            // for race conditions: these checks are more present to prevent misunderstandings of
+            // the API's capabilities, instead of serving as the sole way to defend their invariants.
+            match new_state {
+                TimelineArchivalState::Unarchived => {
+                    Self::check_to_be_unarchived_timeline_has_no_archived_parent(timeline)?
+                }
+                TimelineArchivalState::Archived => {
+                    Self::check_to_be_archived_has_no_unarchived_children(timeline_id, &timelines)?
                 }
             }
-
-            // Ensure that there are no non-archived child timelines
-            let children: Vec<TimelineId> = timelines
-                .iter()
-                .filter_map(|(id, entry)| {
-                    if entry.get_ancestor_timeline_id() != Some(timeline_id) {
-                        return None;
-                    }
-                    if entry.is_archived() == Some(true) {
-                        return None;
-                    }
-                    Some(*id)
-                })
-                .collect();
-
-            if !children.is_empty() && state == TimelineArchivalState::Archived {
-                return Err(TimelineArchivalError::HasUnarchivedChildren(children));
-            }
-            Arc::clone(timeline)
+            Some(Arc::clone(timeline))
         };
 
+        // Second part: unarchive timeline (if needed)
+        let timeline = if let Some(timeline) = timeline_or_unarchive_offloaded {
+            timeline
+        } else {
+            // Turn offloaded timeline into a non-offloaded one
+            self.unoffload_timeline(timeline_id, ctx).await?
+        };
+
+        // Third part: upload new timeline archival state and block until it is present in S3
         let upload_needed = timeline
             .remote_client
-            .schedule_index_upload_for_timeline_archival_state(state)?;
+            .schedule_index_upload_for_timeline_archival_state(new_state)?;
 
         if upload_needed {
             info!("Uploading new state");
@@ -1884,7 +2088,7 @@ impl Tenant {
     ///
     /// Returns whether we have pending compaction task.
     async fn compaction_iteration(
-        &self,
+        self: &Arc<Self>,
         cancel: &CancellationToken,
         ctx: &RequestContext,
     ) -> Result<bool, timeline::CompactionError> {
@@ -1905,21 +2109,28 @@ impl Tenant {
         // while holding the lock. Then drop the lock and actually perform the
         // compactions.  We don't want to block everything else while the
         // compaction runs.
-        let timelines_to_compact = {
+        let timelines_to_compact_or_offload;
+        {
             let timelines = self.timelines.lock().unwrap();
-            let timelines_to_compact = timelines
+            timelines_to_compact_or_offload = timelines
                 .iter()
                 .filter_map(|(timeline_id, timeline)| {
-                    if timeline.is_active() {
-                        Some((*timeline_id, timeline.clone()))
-                    } else {
+                    let (is_active, can_offload) = (timeline.is_active(), timeline.can_offload());
+                    let has_no_unoffloaded_children = {
+                        !timelines
+                            .iter()
+                            .any(|(_id, tl)| tl.get_ancestor_timeline_id() == Some(*timeline_id))
+                    };
+                    let can_offload = can_offload && has_no_unoffloaded_children;
+                    if (is_active, can_offload) == (false, false) {
                         None
+                    } else {
+                        Some((*timeline_id, timeline.clone(), (is_active, can_offload)))
                     }
                 })
                 .collect::<Vec<_>>();
             drop(timelines);
-            timelines_to_compact
-        };
+        }
 
         // Before doing any I/O work, check our circuit breaker
         if self.compaction_circuit_breaker.lock().unwrap().is_broken() {
@@ -1929,20 +2140,34 @@ impl Tenant {
 
         let mut has_pending_task = false;
 
-        for (timeline_id, timeline) in &timelines_to_compact {
-            has_pending_task |= timeline
-                .compact(cancel, EnumSet::empty(), ctx)
-                .instrument(info_span!("compact_timeline", %timeline_id))
-                .await
-                .inspect_err(|e| match e {
-                    timeline::CompactionError::ShuttingDown => (),
-                    timeline::CompactionError::Other(e) => {
-                        self.compaction_circuit_breaker
-                            .lock()
-                            .unwrap()
-                            .fail(&CIRCUIT_BREAKERS_BROKEN, e);
-                    }
-                })?;
+        for (timeline_id, timeline, (can_compact, can_offload)) in &timelines_to_compact_or_offload
+        {
+            let pending_task_left = if *can_compact {
+                Some(
+                    timeline
+                        .compact(cancel, EnumSet::empty(), ctx)
+                        .instrument(info_span!("compact_timeline", %timeline_id))
+                        .await
+                        .inspect_err(|e| match e {
+                            timeline::CompactionError::ShuttingDown => (),
+                            timeline::CompactionError::Other(e) => {
+                                self.compaction_circuit_breaker
+                                    .lock()
+                                    .unwrap()
+                                    .fail(&CIRCUIT_BREAKERS_BROKEN, e);
+                            }
+                        })?,
+                )
+            } else {
+                None
+            };
+            has_pending_task |= pending_task_left.unwrap_or(false);
+            if pending_task_left == Some(false) && *can_offload {
+                offload_timeline(self, timeline)
+                    .instrument(info_span!("offload_timeline", %timeline_id))
+                    .await
+                    .map_err(timeline::CompactionError::Other)?;
+            }
         }
 
         self.compaction_circuit_breaker
@@ -2852,6 +3077,7 @@ impl Tenant {
             constructed_at: Instant::now(),
             timelines: Mutex::new(HashMap::new()),
             timelines_creating: Mutex::new(HashSet::new()),
+            timelines_offloaded: Mutex::new(HashMap::new()),
             gc_cs: tokio::sync::Mutex::new(()),
             walredo_mgr,
             remote_storage,

pageserver/src/tenant/gc_block.rs

@@ -141,14 +141,14 @@ impl GcBlock {
         Ok(())
     }
 
-    pub(crate) fn before_delete(&self, timeline: &super::Timeline) {
+    pub(crate) fn before_delete(&self, timeline_id: &super::TimelineId) {
         let unblocked = {
             let mut g = self.reasons.lock().unwrap();
             if g.is_empty() {
                 return;
             }
 
-            g.remove(&timeline.timeline_id);
+            g.remove(timeline_id);
 
             BlockingReasons::clean_and_summarize(g).is_none()
         };

pageserver/src/tenant/secondary/downloader.rs

@@ -950,6 +950,7 @@ impl<'a> TenantDownloader<'a> {
         let cancel = &self.secondary_state.cancel;
         let opts = DownloadOpts {
             etag: prev_etag.cloned(),
+            ..Default::default()
         };
 
         backoff::retry(

pageserver/src/tenant/timeline.rs

@@ -7,6 +7,7 @@ pub(crate) mod handle;
 mod init;
 pub mod layer_manager;
 pub(crate) mod logical_size;
+pub mod offload;
 pub mod span;
 pub mod uninit;
 mod walreceiver;
@@ -1556,6 +1557,17 @@ impl Timeline {
         }
     }
 
+    /// Checks if the internal state of the timeline is consistent with it being able to be offloaded.
+    /// This is neccessary but not sufficient for offloading of the timeline as it might have
+    /// child timelines that are not offloaded yet.
+    pub(crate) fn can_offload(&self) -> bool {
+        if self.remote_client.is_archived() != Some(true) {
+            return false;
+        }
+
+        true
+    }
+
     /// Outermost timeline compaction operation; downloads needed layers. Returns whether we have pending
     /// compaction tasks.
     pub(crate) async fn compact(
@@ -1818,7 +1830,6 @@ impl Timeline {
         self.current_state() == TimelineState::Active
     }
 
-    #[allow(unused)]
     pub(crate) fn is_archived(&self) -> Option<bool> {
         self.remote_client.is_archived()
     }

pageserver/src/tenant/timeline/delete.rs

@@ -15,7 +15,7 @@ use crate::{
     tenant::{
         metadata::TimelineMetadata,
         remote_timeline_client::{PersistIndexPartWithDeletedFlagError, RemoteTimelineClient},
-        CreateTimelineCause, DeleteTimelineError, Tenant,
+        CreateTimelineCause, DeleteTimelineError, Tenant, TimelineOrOffloaded,
     },
 };
 
@@ -24,12 +24,14 @@ use super::{Timeline, TimelineResources};
 /// Mark timeline as deleted in S3 so we won't pick it up next time
 /// during attach or pageserver restart.
 /// See comment in persist_index_part_with_deleted_flag.
-async fn set_deleted_in_remote_index(timeline: &Timeline) -> Result<(), DeleteTimelineError> {
-    match timeline
-        .remote_client
+async fn set_deleted_in_remote_index(
+    timeline: &TimelineOrOffloaded,
+) -> Result<(), DeleteTimelineError> {
+    let res = timeline
+        .remote_client()
         .persist_index_part_with_deleted_flag()
-        .await
-    {
+        .await;
+    match res {
         // If we (now, or already) marked it successfully as deleted, we can proceed
         Ok(()) | Err(PersistIndexPartWithDeletedFlagError::AlreadyDeleted(_)) => (),
         // Bail out otherwise
@@ -127,9 +129,9 @@ pub(super) async fn delete_local_timeline_directory(
 }
 
 /// Removes remote layers and an index file after them.
-async fn delete_remote_layers_and_index(timeline: &Timeline) -> anyhow::Result<()> {
+async fn delete_remote_layers_and_index(timeline: &TimelineOrOffloaded) -> anyhow::Result<()> {
     timeline
-        .remote_client
+        .remote_client()
         .delete_all()
         .await
         .context("delete_all")
@@ -137,27 +139,41 @@ async fn delete_remote_layers_and_index(timeline: &Timeline) -> anyhow::Result<(
 
 /// It is important that this gets called when DeletionGuard is being held.
 /// For more context see comments in [`DeleteTimelineFlow::prepare`]
-async fn remove_timeline_from_tenant(
+async fn remove_maybe_offloaded_timeline_from_tenant(
     tenant: &Tenant,
-    timeline: &Timeline,
+    timeline: &TimelineOrOffloaded,
     _: &DeletionGuard, // using it as a witness
 ) -> anyhow::Result<()> {
     // Remove the timeline from the map.
+    // This observes the locking order between timelines and timelines_offloaded
     let mut timelines = tenant.timelines.lock().unwrap();
+    let mut timelines_offloaded = tenant.timelines_offloaded.lock().unwrap();
+    let offloaded_children_exist = timelines_offloaded
+        .iter()
+        .any(|(_, entry)| entry.ancestor_timeline_id == Some(timeline.timeline_id()));
     let children_exist = timelines
         .iter()
-        .any(|(_, entry)| entry.get_ancestor_timeline_id() == Some(timeline.timeline_id));
-    // XXX this can happen because `branch_timeline` doesn't check `TimelineState::Stopping`.
-    // We already deleted the layer files, so it's probably best to panic.
-    // (Ideally, above remove_dir_all is atomic so we don't see this timeline after a restart)
-    if children_exist {
+        .any(|(_, entry)| entry.get_ancestor_timeline_id() == Some(timeline.timeline_id()));
+    // XXX this can happen because of race conditions with branch creation.
+    // We already deleted the remote layer files, so it's probably best to panic.
+    if children_exist || offloaded_children_exist {
         panic!("Timeline grew children while we removed layer files");
     }
 
-    timelines
-        .remove(&timeline.timeline_id)
-        .expect("timeline that we were deleting was concurrently removed from 'timelines' map");
+    match timeline {
+        TimelineOrOffloaded::Timeline(timeline) => {
+            timelines.remove(&timeline.timeline_id).expect(
+                "timeline that we were deleting was concurrently removed from 'timelines' map",
+            );
+        }
+        TimelineOrOffloaded::Offloaded(timeline) => {
+            timelines_offloaded
+                .remove(&timeline.timeline_id)
+                .expect("timeline that we were deleting was concurrently removed from 'timelines_offloaded' map");
+        }
+    }
 
+    drop(timelines_offloaded);
     drop(timelines);
 
     Ok(())
@@ -207,9 +223,11 @@ impl DeleteTimelineFlow {
         guard.mark_in_progress()?;
 
         // Now that the Timeline is in Stopping state, request all the related tasks to shut down.
-        timeline.shutdown(super::ShutdownMode::Hard).await;
+        if let TimelineOrOffloaded::Timeline(timeline) = &timeline {
+            timeline.shutdown(super::ShutdownMode::Hard).await;
+        }
 
-        tenant.gc_block.before_delete(&timeline);
+        tenant.gc_block.before_delete(&timeline.timeline_id());
 
         fail::fail_point!("timeline-delete-before-index-deleted-at", |_| {
             Err(anyhow::anyhow!(
@@ -285,15 +303,16 @@ impl DeleteTimelineFlow {
 
         guard.mark_in_progress()?;
 
+        let timeline = TimelineOrOffloaded::Timeline(timeline);
         Self::schedule_background(guard, tenant.conf, tenant, timeline);
 
         Ok(())
     }
 
-    fn prepare(
+    pub(super) fn prepare(
         tenant: &Tenant,
         timeline_id: TimelineId,
-    ) -> Result<(Arc<Timeline>, DeletionGuard), DeleteTimelineError> {
+    ) -> Result<(TimelineOrOffloaded, DeletionGuard), DeleteTimelineError> {
         // Note the interaction between this guard and deletion guard.
         // Here we attempt to lock deletion guard when we're holding a lock on timelines.
         // This is important because when you take into account `remove_timeline_from_tenant`
@@ -307,8 +326,14 @@ impl DeleteTimelineFlow {
         let timelines = tenant.timelines.lock().unwrap();
 
         let timeline = match timelines.get(&timeline_id) {
-            Some(t) => t,
-            None => return Err(DeleteTimelineError::NotFound),
+            Some(t) => TimelineOrOffloaded::Timeline(Arc::clone(t)),
+            None => {
+                let offloaded_timelines = tenant.timelines_offloaded.lock().unwrap();
+                match offloaded_timelines.get(&timeline_id) {
+                    Some(t) => TimelineOrOffloaded::Offloaded(Arc::clone(t)),
+                    None => return Err(DeleteTimelineError::NotFound),
+                }
+            }
         };
 
         // Ensure that there are no child timelines **attached to that pageserver**,
@@ -334,30 +359,32 @@ impl DeleteTimelineFlow {
         // to remove the timeline from it.
         // Always if you have two locks that are taken in different order this can result in a deadlock.
 
-        let delete_progress = Arc::clone(&timeline.delete_progress);
+        let delete_progress = Arc::clone(timeline.delete_progress());
         let delete_lock_guard = match delete_progress.try_lock_owned() {
             Ok(guard) => DeletionGuard(guard),
             Err(_) => {
                 // Unfortunately if lock fails arc is consumed.
                 return Err(DeleteTimelineError::AlreadyInProgress(Arc::clone(
-                    &timeline.delete_progress,
+                    timeline.delete_progress(),
                 )));
             }
         };
 
-        timeline.set_state(TimelineState::Stopping);
+        if let TimelineOrOffloaded::Timeline(timeline) = &timeline {
+            timeline.set_state(TimelineState::Stopping);
+        }
 
-        Ok((Arc::clone(timeline), delete_lock_guard))
+        Ok((timeline, delete_lock_guard))
     }
 
     fn schedule_background(
         guard: DeletionGuard,
         conf: &'static PageServerConf,
         tenant: Arc<Tenant>,
-        timeline: Arc<Timeline>,
+        timeline: TimelineOrOffloaded,
     ) {
-        let tenant_shard_id = timeline.tenant_shard_id;
-        let timeline_id = timeline.timeline_id;
+        let tenant_shard_id = timeline.tenant_shard_id();
+        let timeline_id = timeline.timeline_id();
 
         task_mgr::spawn(
             task_mgr::BACKGROUND_RUNTIME.handle(),
@@ -368,7 +395,9 @@ impl DeleteTimelineFlow {
             async move {
                 if let Err(err) = Self::background(guard, conf, &tenant, &timeline).await {
                     error!("Error: {err:#}");
-                    timeline.set_broken(format!("{err:#}"))
+                    if let TimelineOrOffloaded::Timeline(timeline) = timeline {
+                        timeline.set_broken(format!("{err:#}"))
+                    }
                 };
                 Ok(())
             }
@@ -380,15 +409,19 @@ impl DeleteTimelineFlow {
         mut guard: DeletionGuard,
         conf: &PageServerConf,
         tenant: &Tenant,
-        timeline: &Timeline,
+        timeline: &TimelineOrOffloaded,
     ) -> Result<(), DeleteTimelineError> {
-        delete_local_timeline_directory(conf, tenant.tenant_shard_id, timeline).await?;
+        // Offloaded timelines have no local state
+        // TODO: once we persist offloaded information, delete the timeline from there, too
+        if let TimelineOrOffloaded::Timeline(timeline) = timeline {
+            delete_local_timeline_directory(conf, tenant.tenant_shard_id, timeline).await?;
+        }
 
         delete_remote_layers_and_index(timeline).await?;
 
         pausable_failpoint!("in_progress_delete");
 
-        remove_timeline_from_tenant(tenant, timeline, &guard).await?;
+        remove_maybe_offloaded_timeline_from_tenant(tenant, timeline, &guard).await?;
 
         *guard = Self::Finished;
 
@@ -400,7 +433,7 @@ impl DeleteTimelineFlow {
     }
 }
 
-struct DeletionGuard(OwnedMutexGuard<DeleteTimelineFlow>);
+pub(super) struct DeletionGuard(OwnedMutexGuard<DeleteTimelineFlow>);
 
 impl Deref for DeletionGuard {
     type Target = DeleteTimelineFlow;

pageserver/src/tenant/timeline/offload.rs

@@ -0,0 +1,69 @@
+use std::sync::Arc;
+
+use crate::tenant::{OffloadedTimeline, Tenant, TimelineOrOffloaded};
+
+use super::{
+    delete::{delete_local_timeline_directory, DeleteTimelineFlow, DeletionGuard},
+    Timeline,
+};
+
+pub(crate) async fn offload_timeline(
+    tenant: &Tenant,
+    timeline: &Arc<Timeline>,
+) -> anyhow::Result<()> {
+    tracing::info!("offloading archived timeline");
+    let (timeline, guard) = DeleteTimelineFlow::prepare(tenant, timeline.timeline_id)?;
+
+    let TimelineOrOffloaded::Timeline(timeline) = timeline else {
+        tracing::error!("timeline already offloaded, but given timeline object");
+        return Ok(());
+    };
+
+    // TODO extend guard mechanism above with method
+    // to make deletions possible while offloading is in progress
+
+    // TODO mark timeline as offloaded in S3
+
+    let conf = &tenant.conf;
+    delete_local_timeline_directory(conf, tenant.tenant_shard_id, &timeline).await?;
+
+    remove_timeline_from_tenant(tenant, &timeline, &guard).await?;
+
+    {
+        let mut offloaded_timelines = tenant.timelines_offloaded.lock().unwrap();
+        offloaded_timelines.insert(
+            timeline.timeline_id,
+            Arc::new(OffloadedTimeline::from_timeline(&timeline)),
+        );
+    }
+
+    Ok(())
+}
+
+/// It is important that this gets called when DeletionGuard is being held.
+/// For more context see comments in [`DeleteTimelineFlow::prepare`]
+async fn remove_timeline_from_tenant(
+    tenant: &Tenant,
+    timeline: &Timeline,
+    _: &DeletionGuard, // using it as a witness
+) -> anyhow::Result<()> {
+    // Remove the timeline from the map.
+    let mut timelines = tenant.timelines.lock().unwrap();
+    let children_exist = timelines
+        .iter()
+        .any(|(_, entry)| entry.get_ancestor_timeline_id() == Some(timeline.timeline_id));
+    // XXX this can happen because `branch_timeline` doesn't check `TimelineState::Stopping`.
+    // We already deleted the layer files, so it's probably best to panic.
+    // (Ideally, above remove_dir_all is atomic so we don't see this timeline after a restart)
+    if children_exist {
+        panic!("Timeline grew children while we removed layer files");
+    }
+
+    timelines
+        .remove(&timeline.timeline_id)
+        .expect("timeline that we were deleting was concurrently removed from 'timelines' map");
+
+    drop(timelines);
+
+    Ok(())
+}

pageserver/src/virtual_file.rs

@@ -44,7 +44,6 @@ pub(crate) use api::IoMode;
 pub(crate) use io_engine::IoEngineKind;
 pub(crate) use metadata::Metadata;
 pub(crate) use open_options::*;
-pub(crate) mod dio;
 
 pub(crate) mod owned_buffers_io {
     //! Abstractions for IO with owned buffers.
@@ -56,7 +55,6 @@ pub(crate) mod owned_buffers_io {
     //! but for the time being we're proving out the primitives in the neon.git repo
     //! for faster iteration.
 
-    pub(crate) mod io_buf_aligned;
     pub(crate) mod io_buf_ext;
     pub(crate) mod slice;
     pub(crate) mod write;
@@ -66,39 +64,22 @@ pub(crate) mod owned_buffers_io {
 }
 
 #[derive(Debug)]
-pub enum VirtualFile {
-    Buffered(VirtualFileInner),
-    Direct(VirtualFileInner),
+pub struct VirtualFile {
+    inner: VirtualFileInner,
+    _mode: IoMode,
 }
 
 impl VirtualFile {
-    fn inner(&self) -> &VirtualFileInner {
-        match self {
-            Self::Buffered(file) => file,
-            Self::Direct(file) => file,
-        }
-    }
-
-    fn inner_mut(&mut self) -> &mut VirtualFileInner {
-        match self {
-            Self::Buffered(file) => file,
-            Self::Direct(file) => file,
-        }
-    }
-
-    fn into_inner(self) -> VirtualFileInner {
-        match self {
-            Self::Buffered(file) => file,
-            Self::Direct(file) => file,
-        }
-    }
     /// Open a file in read-only mode. Like File::open.
     pub async fn open<P: AsRef<Utf8Path>>(
         path: P,
         ctx: &RequestContext,
     ) -> Result<Self, std::io::Error> {
-        let file = VirtualFileInner::open(path, ctx).await?;
-        Ok(Self::Buffered(file))
+        let inner = VirtualFileInner::open(path, ctx).await?;
+        Ok(VirtualFile {
+            inner,
+            _mode: IoMode::Buffered,
+        })
     }
 
     /// Open a file in read-only mode. Like File::open.
@@ -115,8 +96,11 @@ impl VirtualFile {
         path: P,
         ctx: &RequestContext,
     ) -> Result<Self, std::io::Error> {
-        let file = VirtualFileInner::create(path, ctx).await?;
-        Ok(Self::Buffered(file))
+        let inner = VirtualFileInner::create(path, ctx).await?;
+        Ok(VirtualFile {
+            inner,
+            _mode: IoMode::Buffered,
+        })
     }
 
     pub async fn create_v2<P: AsRef<Utf8Path>>(
@@ -136,36 +120,45 @@ impl VirtualFile {
         open_options: &OpenOptions,
         ctx: &RequestContext, /* TODO: carry a pointer to the metrics in the RequestContext instead of the parsing https://github.com/neondatabase/neon/issues/6107 */
     ) -> Result<Self, std::io::Error> {
-        let file = VirtualFileInner::open_with_options(path, open_options, ctx).await?;
-        Ok(Self::Buffered(file))
+        let inner = VirtualFileInner::open_with_options(path, open_options, ctx).await?;
+        Ok(VirtualFile {
+            inner,
+            _mode: IoMode::Buffered,
+        })
     }
 
     pub async fn open_with_options_v2<P: AsRef<Utf8Path>>(
         path: P,
-        open_options: &mut OpenOptions, // Uses `&mut` here to add `O_DIRECT`.
+        open_options: &OpenOptions,
         ctx: &RequestContext, /* TODO: carry a pointer to the metrics in the RequestContext instead of the parsing https://github.com/neondatabase/neon/issues/6107 */
     ) -> Result<Self, std::io::Error> {
         let file = match get_io_mode() {
             IoMode::Buffered => {
-                let file = VirtualFileInner::open_with_options(path, open_options, ctx).await?;
-                Self::Buffered(file)
+                let inner = VirtualFileInner::open_with_options(path, open_options, ctx).await?;
+                VirtualFile {
+                    inner,
+                    _mode: IoMode::Buffered,
+                }
             }
             #[cfg(target_os = "linux")]
             IoMode::Direct => {
-                let file = VirtualFileInner::open_with_options(
+                let inner = VirtualFileInner::open_with_options(
                     path,
-                    open_options.custom_flags(nix::libc::O_DIRECT),
+                    open_options.clone().custom_flags(nix::libc::O_DIRECT),
                     ctx,
                 )
                 .await?;
-                Self::Direct(file)
+                VirtualFile {
+                    inner,
+                    _mode: IoMode::Direct,
+                }
             }
         };
         Ok(file)
     }
 
     pub fn path(&self) -> &Utf8Path {
-        self.inner().path.as_path()
+        self.inner.path.as_path()
     }
 
     pub async fn crashsafe_overwrite<B: BoundedBuf<Buf = Buf> + Send, Buf: IoBuf + Send>(
@@ -177,23 +170,23 @@ impl VirtualFile {
     }
 
     pub async fn sync_all(&self) -> Result<(), Error> {
-        self.inner().sync_all().await
+        self.inner.sync_all().await
     }
 
     pub async fn sync_data(&self) -> Result<(), Error> {
-        self.inner().sync_data().await
+        self.inner.sync_data().await
     }
 
     pub async fn metadata(&self) -> Result<Metadata, Error> {
-        self.inner().metadata().await
+        self.inner.metadata().await
     }
 
     pub fn remove(self) {
-        self.into_inner().remove();
+        self.inner.remove();
     }
 
     pub async fn seek(&mut self, pos: SeekFrom) -> Result<u64, Error> {
-        self.inner_mut().seek(pos).await
+        self.inner.seek(pos).await
     }
 
     pub async fn read_exact_at<Buf>(
@@ -205,7 +198,7 @@ impl VirtualFile {
     where
         Buf: IoBufMut + Send,
     {
-        self.inner().read_exact_at(slice, offset, ctx).await
+        self.inner.read_exact_at(slice, offset, ctx).await
     }
 
     pub async fn read_exact_at_page(
@@ -214,7 +207,7 @@ impl VirtualFile {
         offset: u64,
         ctx: &RequestContext,
     ) -> Result<PageWriteGuard<'static>, Error> {
-        self.inner().read_exact_at_page(page, offset, ctx).await
+        self.inner.read_exact_at_page(page, offset, ctx).await
     }
 
     pub async fn write_all_at<Buf: IoBuf + Send>(
@@ -223,7 +216,7 @@ impl VirtualFile {
         offset: u64,
         ctx: &RequestContext,
     ) -> (FullSlice<Buf>, Result<(), Error>) {
-        self.inner().write_all_at(buf, offset, ctx).await
+        self.inner.write_all_at(buf, offset, ctx).await
     }
 
     pub async fn write_all<Buf: IoBuf + Send>(
@@ -231,7 +224,7 @@ impl VirtualFile {
         buf: FullSlice<Buf>,
         ctx: &RequestContext,
     ) -> (FullSlice<Buf>, Result<usize, Error>) {
-        self.inner_mut().write_all(buf, ctx).await
+        self.inner.write_all(buf, ctx).await
     }
 }
 
@@ -1213,11 +1206,11 @@ impl VirtualFile {
         blknum: u32,
         ctx: &RequestContext,
     ) -> Result<crate::tenant::block_io::BlockLease<'_>, std::io::Error> {
-        self.inner().read_blk(blknum, ctx).await
+        self.inner.read_blk(blknum, ctx).await
     }
 
     async fn read_to_end(&mut self, buf: &mut Vec<u8>, ctx: &RequestContext) -> Result<(), Error> {
-        self.inner_mut().read_to_end(buf, ctx).await
+        self.inner.read_to_end(buf, ctx).await
     }
 }
 
@@ -1364,8 +1357,6 @@ pub(crate) const fn get_io_buffer_alignment() -> usize {
     DEFAULT_IO_BUFFER_ALIGNMENT
 }
 
-pub(crate) type IoBufferMut = dio::AlignedBufferMut<{ get_io_buffer_alignment() }>;
-
 static IO_MODE: AtomicU8 = AtomicU8::new(IoMode::preferred() as u8);
 
 pub(crate) fn set_io_mode(mode: IoMode) {

pageserver/src/virtual_file/dio.rs

@@ -1,405 +0,0 @@
-#![allow(unused)]
-
-use core::slice;
-use std::{
-    alloc::{self, Layout},
-    cmp,
-    mem::{ManuallyDrop, MaybeUninit},
-    ops::{Deref, DerefMut},
-    ptr::{addr_of_mut, NonNull},
-};
-
-use bytes::buf::UninitSlice;
-
-struct IoBufferPtr(*mut u8);
-
-// SAFETY: We gurantees no one besides `IoBufferPtr` itself has the raw pointer.
-unsafe impl Send for IoBufferPtr {}
-
-/// An aligned buffer type used for I/O.
-pub struct AlignedBufferMut<const ALIGN: usize> {
-    ptr: IoBufferPtr,
-    capacity: usize,
-    len: usize,
-}
-
-impl<const ALIGN: usize> AlignedBufferMut<ALIGN> {
-    /// Constructs a new, empty `IoBufferMut` with at least the specified capacity and alignment.
-    ///
-    /// The buffer will be able to hold at most `capacity` elements and will never resize.
-    ///
-    ///
-    /// # Panics
-    ///
-    /// Panics if the new capacity exceeds `isize::MAX` _bytes_, or if the following alignment requirement is not met:
-    /// * `align` must not be zero,
-    ///
-    /// * `align` must be a power of two,
-    ///
-    /// * `capacity`, when rounded up to the nearest multiple of `align`,
-    ///    must not overflow isize (i.e., the rounded value must be
-    ///    less than or equal to `isize::MAX`).
-    pub fn with_capacity(capacity: usize) -> Self {
-        let layout = Layout::from_size_align(capacity, ALIGN).expect("Invalid layout");
-
-        // SAFETY:  Making an allocation with a sized and aligned layout. The memory is manually freed with the same layout.
-        let ptr = unsafe {
-            let ptr = alloc::alloc(layout);
-            if ptr.is_null() {
-                alloc::handle_alloc_error(layout);
-            }
-            IoBufferPtr(ptr)
-        };
-
-        AlignedBufferMut {
-            ptr,
-            capacity,
-            len: 0,
-        }
-    }
-
-    /// Constructs a new `IoBufferMut` with at least the specified capacity and alignment, filled with zeros.
-    pub fn with_capacity_zeroed(capacity: usize) -> Self {
-        use bytes::BufMut;
-        let mut buf = Self::with_capacity(capacity);
-        buf.put_bytes(0, capacity);
-        buf.len = capacity;
-        buf
-    }
-
-    /// Returns the total number of bytes the buffer can hold.
-    #[inline]
-    pub fn capacity(&self) -> usize {
-        self.capacity
-    }
-
-    /// Returns the alignment of the buffer.
-    #[inline]
-    pub const fn align(&self) -> usize {
-        ALIGN
-    }
-
-    /// Returns the number of bytes in the buffer, also referred to as its 'length'.
-    #[inline]
-    pub fn len(&self) -> usize {
-        self.len
-    }
-
-    /// Force the length of the buffer to `new_len`.
-    #[inline]
-    unsafe fn set_len(&mut self, new_len: usize) {
-        debug_assert!(new_len <= self.capacity());
-        self.len = new_len;
-    }
-
-    #[inline]
-    fn as_ptr(&self) -> *const u8 {
-        self.ptr.0
-    }
-
-    #[inline]
-    fn as_mut_ptr(&mut self) -> *mut u8 {
-        self.ptr.0
-    }
-
-    /// Extracts a slice containing the entire buffer.
-    ///
-    /// Equivalent to `&s[..]`.
-    #[inline]
-    fn as_slice(&self) -> &[u8] {
-        // SAFETY: The pointer is valid and `len` bytes are initialized.
-        unsafe { slice::from_raw_parts(self.as_ptr(), self.len) }
-    }
-
-    /// Extracts a mutable slice of the entire buffer.
-    ///
-    /// Equivalent to `&mut s[..]`.
-    fn as_mut_slice(&mut self) -> &mut [u8] {
-        // SAFETY: The pointer is valid and `len` bytes are initialized.
-        unsafe { slice::from_raw_parts_mut(self.as_mut_ptr(), self.len) }
-    }
-
-    /// Drops the all the contents of the buffer, setting its length to `0`.
-    #[inline]
-    pub fn clear(&mut self) {
-        self.len = 0;
-    }
-
-    /// Reserves capacity for at least `additional` more bytes to be inserted
-    /// in the given `IoBufferMut`. The collection may reserve more space to
-    /// speculatively avoid frequent reallocations. After calling `reserve`,
-    /// capacity will be greater than or equal to `self.len() + additional`.
-    /// Does nothing if capacity is already sufficient.
-    ///
-    /// # Panics
-    ///
-    /// Panics if the new capacity exceeds `isize::MAX` _bytes_.
-    pub fn reserve(&mut self, additional: usize) {
-        if additional > self.capacity() - self.len() {
-            self.reserve_inner(additional);
-        }
-    }
-
-    fn reserve_inner(&mut self, additional: usize) {
-        let Some(required_cap) = self.len().checked_add(additional) else {
-            capacity_overflow()
-        };
-
-        let old_capacity = self.capacity();
-        let align = self.align();
-        // This guarantees exponential growth. The doubling cannot overflow
-        // because `cap <= isize::MAX` and the type of `cap` is `usize`.
-        let cap = cmp::max(old_capacity * 2, required_cap);
-
-        if !is_valid_alloc(cap) {
-            capacity_overflow()
-        }
-        let new_layout = Layout::from_size_align(cap, self.align()).expect("Invalid layout");
-
-        let old_ptr = self.as_mut_ptr();
-
-        // SAFETY: old allocation was allocated with std::alloc::alloc with the same layout,
-        // and we panics on null pointer.
-        let (ptr, cap) = unsafe {
-            let old_layout = Layout::from_size_align_unchecked(old_capacity, align);
-            let ptr = alloc::realloc(old_ptr, old_layout, new_layout.size());
-            if ptr.is_null() {
-                alloc::handle_alloc_error(new_layout);
-            }
-            (IoBufferPtr(ptr), cap)
-        };
-
-        self.ptr = ptr;
-        self.capacity = cap;
-    }
-
-    /// Consumes and leaks the `IoBufferMut`, returning a mutable reference to the contents, &'a mut [u8].
-    pub fn leak<'a>(self) -> &'a mut [u8] {
-        let mut buf = ManuallyDrop::new(self);
-        // SAFETY: leaking the buffer as intended.
-        unsafe { slice::from_raw_parts_mut(buf.as_mut_ptr(), buf.len) }
-    }
-}
-
-fn capacity_overflow() -> ! {
-    panic!("capacity overflow")
-}
-
-// We need to guarantee the following:
-// * We don't ever allocate `> isize::MAX` byte-size objects.
-// * We don't overflow `usize::MAX` and actually allocate too little.
-//
-// On 64-bit we just need to check for overflow since trying to allocate
-// `> isize::MAX` bytes will surely fail. On 32-bit and 16-bit we need to add
-// an extra guard for this in case we're running on a platform which can use
-// all 4GB in user-space, e.g., PAE or x32.
-#[inline]
-fn is_valid_alloc(alloc_size: usize) -> bool {
-    !(usize::BITS < 64 && alloc_size > isize::MAX as usize)
-}
-
-impl<const ALIGN: usize> Drop for AlignedBufferMut<ALIGN> {
-    fn drop(&mut self) {
-        // SAFETY: memory was allocated with std::alloc::alloc with the same layout.
-        unsafe {
-            alloc::dealloc(
-                self.as_mut_ptr(),
-                Layout::from_size_align_unchecked(self.capacity, ALIGN),
-            )
-        }
-    }
-}
-
-impl<const ALIGN: usize> Deref for AlignedBufferMut<ALIGN> {
-    type Target = [u8];
-
-    fn deref(&self) -> &Self::Target {
-        self.as_slice()
-    }
-}
-
-impl<const ALIGN: usize> DerefMut for AlignedBufferMut<ALIGN> {
-    fn deref_mut(&mut self) -> &mut Self::Target {
-        self.as_mut_slice()
-    }
-}
-
-/// SAFETY: When advancing the internal cursor, the caller needs to make sure the bytes advcanced past have been initialized.
-unsafe impl<const ALIGN: usize> bytes::BufMut for AlignedBufferMut<ALIGN> {
-    #[inline]
-    fn remaining_mut(&self) -> usize {
-        // Although a `Vec` can have at most isize::MAX bytes, we never want to grow `IoBufferMut`.
-        // Thus, it can have at most `self.capacity` bytes.
-        self.capacity() - self.len()
-    }
-
-    // SAFETY: Caller needs to make sure the bytes being advanced past have been initialized.
-    #[inline]
-    unsafe fn advance_mut(&mut self, cnt: usize) {
-        let len = self.len();
-        let remaining = self.remaining_mut();
-
-        if remaining < cnt {
-            panic_advance(cnt, remaining);
-        }
-
-        // Addition will not overflow since the sum is at most the capacity.
-        self.set_len(len + cnt);
-    }
-
-    #[inline]
-    fn chunk_mut(&mut self) -> &mut bytes::buf::UninitSlice {
-        let cap = self.capacity();
-        let len = self.len();
-
-        // SAFETY: Since `self.ptr` is valid for `cap` bytes, `self.ptr.add(len)` must be
-        // valid for `cap - len` bytes. The subtraction will not underflow since
-        // `len <= cap`.
-        unsafe { UninitSlice::from_raw_parts_mut(self.as_mut_ptr().add(len), cap - len) }
-    }
-}
-
-/// Panic with a nice error message.
-#[cold]
-fn panic_advance(idx: usize, len: usize) -> ! {
-    panic!(
-        "advance out of bounds: the len is {} but advancing by {}",
-        len, idx
-    );
-}
-
-/// Safety: [`IoBufferMut`] has exclusive ownership of the io buffer,
-/// and the location remains stable even if [`Self`] is moved.
-unsafe impl<const ALIGN: usize> tokio_epoll_uring::IoBuf for AlignedBufferMut<ALIGN> {
-    fn stable_ptr(&self) -> *const u8 {
-        self.as_ptr()
-    }
-
-    fn bytes_init(&self) -> usize {
-        self.len()
-    }
-
-    fn bytes_total(&self) -> usize {
-        self.capacity()
-    }
-}
-
-// SAFETY: See above.
-unsafe impl<const ALIGN: usize> tokio_epoll_uring::IoBufMut for AlignedBufferMut<ALIGN> {
-    fn stable_mut_ptr(&mut self) -> *mut u8 {
-        self.as_mut_ptr()
-    }
-
-    unsafe fn set_init(&mut self, init_len: usize) {
-        if self.len() < init_len {
-            self.set_len(init_len);
-        }
-    }
-}
-
-#[cfg(test)]
-mod tests {
-
-    use super::*;
-
-    const ALIGN: usize = 4 * 1024;
-    type TestIoBufferMut = AlignedBufferMut<ALIGN>;
-
-    #[test]
-    fn test_with_capacity() {
-        let v = TestIoBufferMut::with_capacity(ALIGN * 4);
-        assert_eq!(v.len(), 0);
-        assert_eq!(v.capacity(), ALIGN * 4);
-        assert_eq!(v.align(), ALIGN);
-        assert_eq!(v.as_ptr().align_offset(ALIGN), 0);
-
-        let v = TestIoBufferMut::with_capacity(ALIGN / 2);
-        assert_eq!(v.len(), 0);
-        assert_eq!(v.capacity(), ALIGN / 2);
-        assert_eq!(v.align(), ALIGN);
-        assert_eq!(v.as_ptr().align_offset(ALIGN), 0);
-    }
-
-    #[test]
-    fn test_with_capacity_zeroed() {
-        let v = TestIoBufferMut::with_capacity_zeroed(ALIGN);
-        assert_eq!(v.len(), ALIGN);
-        assert_eq!(v.capacity(), ALIGN);
-        assert_eq!(v.align(), ALIGN);
-        assert_eq!(v.as_ptr().align_offset(ALIGN), 0);
-        assert_eq!(&v[..], &[0; ALIGN])
-    }
-
-    #[test]
-    fn test_reserve() {
-        use bytes::BufMut;
-        let mut v = TestIoBufferMut::with_capacity(ALIGN);
-        let capacity = v.capacity();
-        v.reserve(capacity);
-        assert_eq!(v.capacity(), capacity);
-        let data = [b'a'; ALIGN];
-        v.put(&data[..]);
-        v.reserve(capacity);
-        assert!(v.capacity() >= capacity * 2);
-        assert_eq!(&v[..], &data[..]);
-        let capacity = v.capacity();
-        v.clear();
-        v.reserve(capacity);
-        assert_eq!(capacity, v.capacity());
-    }
-
-    #[test]
-    fn test_bytes_put() {
-        use bytes::BufMut;
-        let mut v = TestIoBufferMut::with_capacity(ALIGN * 4);
-        let x = [b'a'; ALIGN];
-
-        for _ in 0..2 {
-            for _ in 0..4 {
-                v.put(&x[..]);
-            }
-            assert_eq!(v.len(), ALIGN * 4);
-            assert_eq!(v.capacity(), ALIGN * 4);
-            assert_eq!(v.align(), ALIGN);
-            assert_eq!(v.as_ptr().align_offset(ALIGN), 0);
-            v.clear()
-        }
-        assert_eq!(v.len(), 0);
-        assert_eq!(v.capacity(), ALIGN * 4);
-        assert_eq!(v.align(), ALIGN);
-        assert_eq!(v.as_ptr().align_offset(ALIGN), 0);
-    }
-
-    #[test]
-    #[should_panic]
-    fn test_bytes_put_panic() {
-        use bytes::BufMut;
-        const ALIGN: usize = 4 * 1024;
-        let mut v = TestIoBufferMut::with_capacity(ALIGN * 4);
-        let x = [b'a'; ALIGN];
-        for _ in 0..5 {
-            v.put_slice(&x[..]);
-        }
-    }
-
-    #[test]
-    fn test_io_buf_put_slice() {
-        use tokio_epoll_uring::BoundedBufMut;
-        const ALIGN: usize = 4 * 1024;
-        let mut v = TestIoBufferMut::with_capacity(ALIGN);
-        let x = [b'a'; ALIGN];
-
-        for _ in 0..2 {
-            v.put_slice(&x[..]);
-            assert_eq!(v.len(), ALIGN);
-            assert_eq!(v.capacity(), ALIGN);
-            assert_eq!(v.align(), ALIGN);
-            assert_eq!(v.as_ptr().align_offset(ALIGN), 0);
-            v.clear()
-        }
-        assert_eq!(v.len(), 0);
-        assert_eq!(v.capacity(), ALIGN);
-        assert_eq!(v.align(), ALIGN);
-        assert_eq!(v.as_ptr().align_offset(ALIGN), 0);
-    }
-}

pageserver/src/virtual_file/owned_buffers_io/io_buf_aligned.rs

@@ -1,9 +0,0 @@
-#![allow(unused)]
-
-use tokio_epoll_uring::IoBufMut;
-
-use crate::virtual_file::IoBufferMut;
-
-pub(crate) trait IoBufAlignedMut: IoBufMut {}
-
-impl IoBufAlignedMut for IoBufferMut {}

pageserver/src/virtual_file/owned_buffers_io/io_buf_ext.rs

@@ -1,6 +1,5 @@
 //! See [`FullSlice`].
 
-use crate::virtual_file::IoBufferMut;
 use bytes::{Bytes, BytesMut};
 use std::ops::{Deref, Range};
 use tokio_epoll_uring::{BoundedBuf, IoBuf, Slice};
@@ -77,4 +76,3 @@ macro_rules! impl_io_buf_ext {
 impl_io_buf_ext!(Bytes);
 impl_io_buf_ext!(BytesMut);
 impl_io_buf_ext!(Vec<u8>);
-impl_io_buf_ext!(IoBufferMut);

pgxn/neon/control_plane_connector.c

@@ -146,6 +146,8 @@ ConstructDeltaMessage()
 	if (RootTable.role_table)
 	{
 		JsonbValue	roles;
+		HASH_SEQ_STATUS status;
+		RoleEntry  *entry;
 
 		roles.type = jbvString;
 		roles.val.string.val = "roles";
@@ -153,9 +155,6 @@ ConstructDeltaMessage()
 		pushJsonbValue(&state, WJB_KEY, &roles);
 		pushJsonbValue(&state, WJB_BEGIN_ARRAY, NULL);
 
-		HASH_SEQ_STATUS status;
-		RoleEntry  *entry;
-
 		hash_seq_init(&status, RootTable.role_table);
 		while ((entry = hash_seq_search(&status)) != NULL)
 		{
@@ -190,10 +189,12 @@ ConstructDeltaMessage()
 		}
 		pushJsonbValue(&state, WJB_END_ARRAY, NULL);
 	}
-	JsonbValue *result = pushJsonbValue(&state, WJB_END_OBJECT, NULL);
-	Jsonb	   *jsonb = JsonbValueToJsonb(result);
+	{
+		JsonbValue *result = pushJsonbValue(&state, WJB_END_OBJECT, NULL);
+		Jsonb	   *jsonb = JsonbValueToJsonb(result);
 
-	return JsonbToCString(NULL, &jsonb->root, 0 /* estimated_len */ );
+		return JsonbToCString(NULL, &jsonb->root, 0 /* estimated_len */ );
+	}
 }
 
 #define ERROR_SIZE 1024
@@ -272,32 +273,28 @@ SendDeltasToControlPlane()
 		curl_easy_setopt(handle, CURLOPT_WRITEFUNCTION, ErrorWriteCallback);
 	}
 
-	char	   *message = ConstructDeltaMessage();
-	ErrorString str;
-
-	str.size = 0;
-
-	curl_easy_setopt(handle, CURLOPT_POSTFIELDS, message);
-	curl_easy_setopt(handle, CURLOPT_WRITEDATA, &str);
-
-	const int	num_retries = 5;
-	CURLcode	curl_status;
-
-	for (int i = 0; i < num_retries; i++)
-	{
-		if ((curl_status = curl_easy_perform(handle)) == 0)
-			break;
-		elog(LOG, "Curl request failed on attempt %d: %s", i, CurlErrorBuf);
-		pg_usleep(1000 * 1000);
-	}
-	if (curl_status != CURLE_OK)
-	{
-		elog(ERROR, "Failed to perform curl request: %s", CurlErrorBuf);
-	}
-	else
 	{
+		char	   *message = ConstructDeltaMessage();
+		ErrorString str;
+		const int	num_retries = 5;
+		CURLcode	curl_status;
 		long		response_code;
 
+		str.size = 0;
+
+		curl_easy_setopt(handle, CURLOPT_POSTFIELDS, message);
+		curl_easy_setopt(handle, CURLOPT_WRITEDATA, &str);
+
+		for (int i = 0; i < num_retries; i++)
+		{
+			if ((curl_status = curl_easy_perform(handle)) == 0)
+				break;
+			elog(LOG, "Curl request failed on attempt %d: %s", i, CurlErrorBuf);
+			pg_usleep(1000 * 1000);
+		}
+		if (curl_status != CURLE_OK)
+			elog(ERROR, "Failed to perform curl request: %s", CurlErrorBuf);
+
 		if (curl_easy_getinfo(handle, CURLINFO_RESPONSE_CODE, &response_code) != CURLE_UNKNOWN_OPTION)
 		{
 			if (response_code != 200)
@@ -376,10 +373,11 @@ MergeTable()
 
 	if (old_table->db_table)
 	{
-		InitDbTableIfNeeded();
 		DbEntry    *entry;
 		HASH_SEQ_STATUS status;
 
+		InitDbTableIfNeeded();
+
 		hash_seq_init(&status, old_table->db_table);
 		while ((entry = hash_seq_search(&status)) != NULL)
 		{
@@ -421,10 +419,11 @@ MergeTable()
 
 	if (old_table->role_table)
 	{
-		InitRoleTableIfNeeded();
 		RoleEntry  *entry;
 		HASH_SEQ_STATUS status;
 
+		InitRoleTableIfNeeded();
+
 		hash_seq_init(&status, old_table->role_table);
 		while ((entry = hash_seq_search(&status)) != NULL)
 		{
@@ -515,9 +514,12 @@ RoleIsNeonSuperuser(const char *role_name)
 static void
 HandleCreateDb(CreatedbStmt *stmt)
 {
-	InitDbTableIfNeeded();
 	DefElem    *downer = NULL;
 	ListCell   *option;
+	bool		found = false;
+	DbEntry    *entry;
+
+	InitDbTableIfNeeded();
 
 	foreach(option, stmt->options)
 	{
@@ -526,13 +528,11 @@ HandleCreateDb(CreatedbStmt *stmt)
 		if (strcmp(defel->defname, "owner") == 0)
 			downer = defel;
 	}
-	bool		found = false;
-	DbEntry    *entry = hash_search(
-									CurrentDdlTable->db_table,
-									stmt->dbname,
-									HASH_ENTER,
-									&found);
 
+	entry = hash_search(CurrentDdlTable->db_table,
+						stmt->dbname,
+						HASH_ENTER,
+						&found);
 	if (!found)
 		memset(entry->old_name, 0, sizeof(entry->old_name));
 
@@ -554,21 +554,24 @@ HandleCreateDb(CreatedbStmt *stmt)
 static void
 HandleAlterOwner(AlterOwnerStmt *stmt)
 {
+	const char *name;
+	bool		found = false;
+	DbEntry    *entry;
+	const char *new_owner;
+
 	if (stmt->objectType != OBJECT_DATABASE)
 		return;
 	InitDbTableIfNeeded();
-	const char *name = strVal(stmt->object);
-	bool		found = false;
-	DbEntry    *entry = hash_search(
-									CurrentDdlTable->db_table,
-									name,
-									HASH_ENTER,
-									&found);
 
+	name = strVal(stmt->object);
+	entry = hash_search(CurrentDdlTable->db_table,
+						name,
+						HASH_ENTER,
+						&found);
 	if (!found)
 		memset(entry->old_name, 0, sizeof(entry->old_name));
-	const char *new_owner = get_rolespec_name(stmt->newowner);
 
+	new_owner = get_rolespec_name(stmt->newowner);
 	if (RoleIsNeonSuperuser(new_owner))
 		elog(ERROR, "can't alter owner to neon_superuser");
 	entry->owner = get_role_oid(new_owner, false);
@@ -578,21 +581,23 @@ HandleAlterOwner(AlterOwnerStmt *stmt)
 static void
 HandleDbRename(RenameStmt *stmt)
 {
+	bool		found = false;
+	DbEntry    *entry;
+	DbEntry    *entry_for_new_name;
+
 	Assert(stmt->renameType == OBJECT_DATABASE);
 	InitDbTableIfNeeded();
-	bool		found = false;
-	DbEntry    *entry = hash_search(
-									CurrentDdlTable->db_table,
-									stmt->subname,
-									HASH_FIND,
-									&found);
-	DbEntry    *entry_for_new_name = hash_search(
-												 CurrentDdlTable->db_table,
-												 stmt->newname,
-												 HASH_ENTER,
-												 NULL);
+	entry = hash_search(CurrentDdlTable->db_table,
+						stmt->subname,
+						HASH_FIND,
+						&found);
 
+	entry_for_new_name = hash_search(CurrentDdlTable->db_table,
+									 stmt->newname,
+									 HASH_ENTER,
+									 NULL);
 	entry_for_new_name->type = Op_Set;
+
 	if (found)
 	{
 		if (entry->old_name[0] != '\0')
@@ -600,8 +605,7 @@ HandleDbRename(RenameStmt *stmt)
 		else
 			strlcpy(entry_for_new_name->old_name, entry->name, NAMEDATALEN);
 		entry_for_new_name->owner = entry->owner;
-		hash_search(
-					CurrentDdlTable->db_table,
+		hash_search(CurrentDdlTable->db_table,
 					stmt->subname,
 					HASH_REMOVE,
 					NULL);
@@ -616,14 +620,15 @@ HandleDbRename(RenameStmt *stmt)
 static void
 HandleDropDb(DropdbStmt *stmt)
 {
-	InitDbTableIfNeeded();
 	bool		found = false;
-	DbEntry    *entry = hash_search(
-									CurrentDdlTable->db_table,
-									stmt->dbname,
-									HASH_ENTER,
-									&found);
+	DbEntry    *entry;
 
+	InitDbTableIfNeeded();
+
+	entry = hash_search(CurrentDdlTable->db_table,
+						stmt->dbname,
+						HASH_ENTER,
+						&found);
 	entry->type = Op_Delete;
 	entry->owner = InvalidOid;
 	if (!found)
@@ -633,16 +638,14 @@ HandleDropDb(DropdbStmt *stmt)
 static void
 HandleCreateRole(CreateRoleStmt *stmt)
 {
-	InitRoleTableIfNeeded();
 	bool		found = false;
-	RoleEntry  *entry = hash_search(
-									CurrentDdlTable->role_table,
-									stmt->role,
-									HASH_ENTER,
-									&found);
-	DefElem    *dpass = NULL;
+	RoleEntry  *entry;
+	DefElem    *dpass;
 	ListCell   *option;
 
+	InitRoleTableIfNeeded();
+
+	dpass = NULL;
 	foreach(option, stmt->options)
 	{
 		DefElem    *defel = lfirst(option);
@@ -650,6 +653,11 @@ HandleCreateRole(CreateRoleStmt *stmt)
 		if (strcmp(defel->defname, "password") == 0)
 			dpass = defel;
 	}
+
+	entry = hash_search(CurrentDdlTable->role_table,
+						stmt->role,
+						HASH_ENTER,
+						&found);
 	if (!found)
 		memset(entry->old_name, 0, sizeof(entry->old_name));
 	if (dpass && dpass->arg)
@@ -662,14 +670,18 @@ HandleCreateRole(CreateRoleStmt *stmt)
 static void
 HandleAlterRole(AlterRoleStmt *stmt)
 {
-	InitRoleTableIfNeeded();
-	DefElem    *dpass = NULL;
-	ListCell   *option;
 	const char *role_name = stmt->role->rolename;
+	DefElem    *dpass;
+	ListCell   *option;
+	bool		found = false;
+	RoleEntry  *entry;
+
+	InitRoleTableIfNeeded();
 
 	if (RoleIsNeonSuperuser(role_name) && !superuser())
 		elog(ERROR, "can't ALTER neon_superuser");
 
+	dpass = NULL;
 	foreach(option, stmt->options)
 	{
 		DefElem    *defel = lfirst(option);
@@ -680,13 +692,11 @@ HandleAlterRole(AlterRoleStmt *stmt)
 	/* We only care about updates to the password */
 	if (!dpass)
 		return;
-	bool		found = false;
-	RoleEntry  *entry = hash_search(
-									CurrentDdlTable->role_table,
-									role_name,
-									HASH_ENTER,
-									&found);
 
+	entry = hash_search(CurrentDdlTable->role_table,
+						role_name,
+						HASH_ENTER,
+						&found);
 	if (!found)
 		memset(entry->old_name, 0, sizeof(entry->old_name));
 	if (dpass->arg)
@@ -699,20 +709,22 @@ HandleAlterRole(AlterRoleStmt *stmt)
 static void
 HandleRoleRename(RenameStmt *stmt)
 {
-	InitRoleTableIfNeeded();
-	Assert(stmt->renameType == OBJECT_ROLE);
 	bool		found = false;
-	RoleEntry  *entry = hash_search(
-									CurrentDdlTable->role_table,
-									stmt->subname,
-									HASH_FIND,
-									&found);
+	RoleEntry  *entry;
+	RoleEntry  *entry_for_new_name;
 
-	RoleEntry  *entry_for_new_name = hash_search(
-												 CurrentDdlTable->role_table,
-												 stmt->newname,
-												 HASH_ENTER,
-												 NULL);
+	Assert(stmt->renameType == OBJECT_ROLE);
+	InitRoleTableIfNeeded();
+
+	entry = hash_search(CurrentDdlTable->role_table,
+						stmt->subname,
+						HASH_FIND,
+						&found);
+
+	entry_for_new_name = hash_search(CurrentDdlTable->role_table,
+									 stmt->newname,
+									 HASH_ENTER,
+									 NULL);
 
 	entry_for_new_name->type = Op_Set;
 	if (found)
@@ -738,9 +750,10 @@ HandleRoleRename(RenameStmt *stmt)
 static void
 HandleDropRole(DropRoleStmt *stmt)
 {
-	InitRoleTableIfNeeded();
 	ListCell   *item;
 
+	InitRoleTableIfNeeded();
+
 	foreach(item, stmt->roles)
 	{
 		RoleSpec   *spec = lfirst(item);

pgxn/neon/file_cache.c

@@ -170,12 +170,14 @@ lfc_disable(char const *op)
 
 		if (lfc_desc > 0)
 		{
+			int			rc;
+
 			/*
 			 * If the reason of error is ENOSPC, then truncation of file may
 			 * help to reclaim some space
 			 */
 			pgstat_report_wait_start(WAIT_EVENT_NEON_LFC_TRUNCATE);
-			int			rc = ftruncate(lfc_desc, 0);
+			rc = ftruncate(lfc_desc, 0);
 			pgstat_report_wait_end();
 
 			if (rc < 0)
@@ -616,7 +618,7 @@ lfc_evict(NRelFileInfo rinfo, ForkNumber forkNum, BlockNumber blkno)
 	 */
 	if (entry->bitmap[chunk_offs >> 5] == 0)
 	{
-		bool		has_remaining_pages;
+		bool		has_remaining_pages = false;
 
 		for (int i = 0; i < CHUNK_BITMAP_SIZE; i++)
 		{
@@ -666,7 +668,6 @@ lfc_readv_select(NRelFileInfo rinfo, ForkNumber forkNum, BlockNumber blkno,
 	BufferTag	tag;
 	FileCacheEntry *entry;
 	ssize_t		rc;
-	bool		result = true;
 	uint32		hash;
 	uint64		generation;
 	uint32		entry_offset;
@@ -925,10 +926,10 @@ lfc_writev(NRelFileInfo rinfo, ForkNumber forkNum, BlockNumber blkno,
 				/* We can reuse a hole that was left behind when the LFC was shrunk previously */
 				FileCacheEntry *hole = dlist_container(FileCacheEntry, list_node, dlist_pop_head_node(&lfc_ctl->holes));
 				uint32		offset = hole->offset;
-				bool		found;
+				bool		hole_found;
 	
-				hash_search_with_hash_value(lfc_hash, &hole->key, hole->hash, HASH_REMOVE, &found);
-				CriticalAssert(found);
+				hash_search_with_hash_value(lfc_hash, &hole->key, hole->hash, HASH_REMOVE, &hole_found);
+				CriticalAssert(hole_found);
 	
 				lfc_ctl->used += 1;
 				entry->offset = offset;	/* reuse the hole */
@@ -1004,7 +1005,7 @@ neon_get_lfc_stats(PG_FUNCTION_ARGS)
 	Datum		result;
 	HeapTuple	tuple;
 	char const *key;
-	uint64		value;
+	uint64		value = 0;
 	Datum		values[NUM_NEON_GET_STATS_COLS];
 	bool		nulls[NUM_NEON_GET_STATS_COLS];
 

pgxn/neon/hll.c

@@ -116,8 +116,6 @@ addSHLL(HyperLogLogState *cState, uint32 hash)
 {
 	uint8		count;
 	uint32		index;
-	size_t		i;
-	size_t		j;
 
 	TimestampTz	now = GetCurrentTimestamp();
 	/* Use the first "k" (registerWidth) bits as a zero based index */

pgxn/neon/libpagestore.c

@@ -89,7 +89,6 @@ typedef struct
 
 #if PG_VERSION_NUM >= 150000
 static shmem_request_hook_type prev_shmem_request_hook = NULL;
-static void walproposer_shmem_request(void);
 #endif
 static shmem_startup_hook_type prev_shmem_startup_hook;
 static PagestoreShmemState *pagestore_shared;
@@ -441,8 +440,8 @@ pageserver_connect(shardno_t shard_no, int elevel)
 			return false;
 		}
 		shard->state = PS_Connecting_Startup;
-		/* fallthrough */
 	}
+	/* FALLTHROUGH */
 	case PS_Connecting_Startup:
 	{
 		char	   *pagestream_query;
@@ -453,8 +452,6 @@ pageserver_connect(shardno_t shard_no, int elevel)
 
 		do
 		{
-			WaitEvent	event;
-
 			switch (poll_result)
 			{
 			default: /* unknown/unused states are handled as a failed connection */
@@ -585,8 +582,8 @@ pageserver_connect(shardno_t shard_no, int elevel)
 		}
 
 		shard->state = PS_Connecting_PageStream;
-		/* fallthrough */
 	}
+	/* FALLTHROUGH */
 	case PS_Connecting_PageStream:
 	{
 		neon_shard_log(shard_no, DEBUG5, "Connection state: Connecting_PageStream");
@@ -631,8 +628,8 @@ pageserver_connect(shardno_t shard_no, int elevel)
 		}
 
 		shard->state = PS_Connected;
-		/* fallthrough */
 	}
+	/* FALLTHROUGH */
 	case PS_Connected:
 		/*
 		 * We successfully connected. Future connections to this PageServer

pgxn/neon/neon_perf_counters.c

@@ -94,7 +94,6 @@ neon_perf_counters_to_metrics(neon_per_backend_counters *counters)
 	metric_t   *metrics = palloc((NUM_METRICS + 1) * sizeof(metric_t));
 	uint64		bucket_accum;
 	int			i = 0;
-	Datum		getpage_wait_str;
 
 	metrics[i].name = "getpage_wait_seconds_count";
 	metrics[i].is_bucket = false;
@@ -224,7 +223,6 @@ neon_get_perf_counters(PG_FUNCTION_ARGS)
 	ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
 	Datum		values[3];
 	bool		nulls[3];
-	Datum		getpage_wait_str;
 	neon_per_backend_counters totals = {0};
 	metric_t   *metrics;
 

pgxn/neon/pagestore_client.h

@@ -213,32 +213,6 @@ extern const f_smgr *smgr_neon(ProcNumber backend, NRelFileInfo rinfo);
 extern void smgr_init_neon(void);
 extern void readahead_buffer_resize(int newsize, void *extra);
 
-/* Neon storage manager functionality */
-
-extern void neon_init(void);
-extern void neon_open(SMgrRelation reln);
-extern void neon_close(SMgrRelation reln, ForkNumber forknum);
-extern void neon_create(SMgrRelation reln, ForkNumber forknum, bool isRedo);
-extern bool neon_exists(SMgrRelation reln, ForkNumber forknum);
-extern void neon_unlink(NRelFileInfoBackend rnode, ForkNumber forknum, bool isRedo);
-#if PG_MAJORVERSION_NUM < 16
-extern void neon_extend(SMgrRelation reln, ForkNumber forknum,
-						BlockNumber blocknum, char *buffer, bool skipFsync);
-#else
-extern void neon_extend(SMgrRelation reln, ForkNumber forknum,
-						BlockNumber blocknum, const void *buffer, bool skipFsync);
-extern void neon_zeroextend(SMgrRelation reln, ForkNumber forknum,
-							BlockNumber blocknum, int nbuffers, bool skipFsync);
-#endif
-
-#if PG_MAJORVERSION_NUM >=17
-extern bool neon_prefetch(SMgrRelation reln, ForkNumber forknum,
-						  BlockNumber blocknum, int nblocks);
-#else
-extern bool neon_prefetch(SMgrRelation reln, ForkNumber forknum,
-						  BlockNumber blocknum);
-#endif
-
 /*
  * LSN values associated with each request to the pageserver
  */
@@ -278,13 +252,7 @@ extern PGDLLEXPORT void neon_read_at_lsn(NRelFileInfo rnode, ForkNumber forkNum,
 extern PGDLLEXPORT void neon_read_at_lsn(NRelFileInfo rnode, ForkNumber forkNum, BlockNumber blkno,
 										 neon_request_lsns request_lsns, void *buffer);
 #endif
-extern void neon_writeback(SMgrRelation reln, ForkNumber forknum,
-						   BlockNumber blocknum, BlockNumber nblocks);
-extern BlockNumber neon_nblocks(SMgrRelation reln, ForkNumber forknum);
 extern int64 neon_dbsize(Oid dbNode);
-extern void neon_truncate(SMgrRelation reln, ForkNumber forknum,
-						  BlockNumber nblocks);
-extern void neon_immedsync(SMgrRelation reln, ForkNumber forknum);
 
 /* utils for neon relsize cache */
 extern void relsize_hash_init(void);

pgxn/neon/pagestore_smgr.c

@@ -118,6 +118,8 @@ static UnloggedBuildPhase unlogged_build_phase = UNLOGGED_BUILD_NOT_IN_PROGRESS;
 static bool neon_redo_read_buffer_filter(XLogReaderState *record, uint8 block_id);
 static bool (*old_redo_read_buffer_filter) (XLogReaderState *record, uint8 block_id) = NULL;
 
+static BlockNumber neon_nblocks(SMgrRelation reln, ForkNumber forknum);
+
 /*
  * Prefetch implementation:
  *
@@ -736,7 +738,7 @@ static void
 prefetch_do_request(PrefetchRequest *slot, neon_request_lsns *force_request_lsns)
 {
 	bool		found;
-	uint64		mySlotNo = slot->my_ring_index;
+	uint64		mySlotNo PG_USED_FOR_ASSERTS_ONLY = slot->my_ring_index;
 
 	NeonGetPageRequest request = {
 		.req.tag = T_NeonGetPageRequest,
@@ -1463,7 +1465,6 @@ log_newpages_copy(NRelFileInfo * rinfo, ForkNumber forkNum, BlockNumber blkno,
 	BlockNumber	blknos[XLR_MAX_BLOCK_ID];
 	Page		pageptrs[XLR_MAX_BLOCK_ID];
 	int			nregistered = 0;
-	XLogRecPtr	result = 0;
 
 	for (int i = 0; i < nblocks; i++)
 	{
@@ -1776,7 +1777,7 @@ neon_wallog_page(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, co
 /*
  *	neon_init() -- Initialize private state
  */
-void
+static void
 neon_init(void)
 {
 	Size		prfs_size;
@@ -2166,7 +2167,7 @@ neon_prefetch_response_usable(neon_request_lsns *request_lsns,
 /*
  *	neon_exists() -- Does the physical file exist?
  */
-bool
+static bool
 neon_exists(SMgrRelation reln, ForkNumber forkNum)
 {
 	bool		exists;
@@ -2272,7 +2273,7 @@ neon_exists(SMgrRelation reln, ForkNumber forkNum)
  *
  * If isRedo is true, it's okay for the relation to exist already.
  */
-void
+static void
 neon_create(SMgrRelation reln, ForkNumber forkNum, bool isRedo)
 {
 	switch (reln->smgr_relpersistence)
@@ -2348,7 +2349,7 @@ neon_create(SMgrRelation reln, ForkNumber forkNum, bool isRedo)
  * Note: any failure should be reported as WARNING not ERROR, because
  * we are usually not in a transaction anymore when this is called.
  */
-void
+static void
 neon_unlink(NRelFileInfoBackend rinfo, ForkNumber forkNum, bool isRedo)
 {
 	/*
@@ -2372,7 +2373,7 @@ neon_unlink(NRelFileInfoBackend rinfo, ForkNumber forkNum, bool isRedo)
  *		EOF).  Note that we assume writing a block beyond current EOF
  *		causes intervening file space to become filled with zeroes.
  */
-void
+static void
 #if PG_MAJORVERSION_NUM < 16
 neon_extend(SMgrRelation reln, ForkNumber forkNum, BlockNumber blkno,
 			char *buffer, bool skipFsync)
@@ -2464,7 +2465,7 @@ neon_extend(SMgrRelation reln, ForkNumber forkNum, BlockNumber blkno,
 }
 
 #if PG_MAJORVERSION_NUM >= 16
-void
+static void
 neon_zeroextend(SMgrRelation reln, ForkNumber forkNum, BlockNumber blocknum,
 				int nblocks, bool skipFsync)
 {
@@ -2560,7 +2561,7 @@ neon_zeroextend(SMgrRelation reln, ForkNumber forkNum, BlockNumber blocknum,
 /*
  *  neon_open() -- Initialize newly-opened relation.
  */
-void
+static void
 neon_open(SMgrRelation reln)
 {
 	/*
@@ -2578,7 +2579,7 @@ neon_open(SMgrRelation reln)
 /*
  *	neon_close() -- Close the specified relation, if it isn't closed already.
  */
-void
+static void
 neon_close(SMgrRelation reln, ForkNumber forknum)
 {
 	/*
@@ -2593,13 +2594,12 @@ neon_close(SMgrRelation reln, ForkNumber forknum)
 /*
  *	neon_prefetch() -- Initiate asynchronous read of the specified block of a relation
  */
-bool
+static bool
 neon_prefetch(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
 			  int nblocks)
 {
 	uint64		ring_index PG_USED_FOR_ASSERTS_ONLY;
 	BufferTag	tag;
-	bool		io_initiated = false;
 
 	switch (reln->smgr_relpersistence)
 	{
@@ -2623,7 +2623,6 @@ neon_prefetch(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
 	while (nblocks > 0)
 	{
 		int		iterblocks = Min(nblocks, PG_IOV_MAX);
-		int		seqlen = 0;
 		bits8		lfc_present[PG_IOV_MAX / 8];
 		memset(lfc_present, 0, sizeof(lfc_present));
 
@@ -2635,8 +2634,6 @@ neon_prefetch(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
 			continue;
 		}
 
-		io_initiated = true;
-
 		tag.blockNum = blocknum;
 		
 		for (int i = 0; i < PG_IOV_MAX / 8; i++)
@@ -2659,7 +2656,7 @@ neon_prefetch(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
 /*
  *	neon_prefetch() -- Initiate asynchronous read of the specified block of a relation
  */
-bool
+static bool
 neon_prefetch(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum)
 {
 	uint64		ring_index PG_USED_FOR_ASSERTS_ONLY;
@@ -2703,7 +2700,7 @@ neon_prefetch(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum)
  * This accepts a range of blocks because flushing several pages at once is
  * considerably more efficient than doing so individually.
  */
-void
+static void
 neon_writeback(SMgrRelation reln, ForkNumber forknum,
 			   BlockNumber blocknum, BlockNumber nblocks)
 {
@@ -2924,10 +2921,10 @@ neon_read_at_lsn(NRelFileInfo rinfo, ForkNumber forkNum, BlockNumber blkno,
  *	neon_read() -- Read the specified block from a relation.
  */
 #if PG_MAJORVERSION_NUM < 16
-void
+static void
 neon_read(SMgrRelation reln, ForkNumber forkNum, BlockNumber blkno, char *buffer)
 #else
-void
+static void
 neon_read(SMgrRelation reln, ForkNumber forkNum, BlockNumber blkno, void *buffer)
 #endif
 {
@@ -3036,7 +3033,7 @@ neon_read(SMgrRelation reln, ForkNumber forkNum, BlockNumber blkno, void *buffer
 #endif /* PG_MAJORVERSION_NUM <= 16 */
 
 #if PG_MAJORVERSION_NUM >= 17
-void
+static void
 neon_readv(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
 		void **buffers, BlockNumber nblocks)
 {
@@ -3200,6 +3197,7 @@ hexdump_page(char *page)
 }
 #endif
 
+#if PG_MAJORVERSION_NUM < 17
 /*
  *	neon_write() -- Write the supplied block at the appropriate location.
  *
@@ -3207,7 +3205,7 @@ hexdump_page(char *page)
  *		relation (ie, those before the current EOF).  To extend a relation,
  *		use mdextend().
  */
-void
+static void
 #if PG_MAJORVERSION_NUM < 16
 neon_write(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, char *buffer, bool skipFsync)
 #else
@@ -3273,11 +3271,12 @@ neon_write(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, const vo
 		#endif
 #endif
 }
+#endif
 
 
 
 #if PG_MAJORVERSION_NUM >= 17
-void
+static void
 neon_writev(SMgrRelation reln, ForkNumber forknum, BlockNumber blkno,
 			 const void **buffers, BlockNumber nblocks, bool skipFsync)
 {
@@ -3327,7 +3326,7 @@ neon_writev(SMgrRelation reln, ForkNumber forknum, BlockNumber blkno,
 /*
  *	neon_nblocks() -- Get the number of blocks stored in a relation.
  */
-BlockNumber
+static BlockNumber
 neon_nblocks(SMgrRelation reln, ForkNumber forknum)
 {
 	NeonResponse *resp;
@@ -3464,7 +3463,7 @@ neon_dbsize(Oid dbNode)
 /*
  *	neon_truncate() -- Truncate relation to specified number of blocks.
  */
-void
+static void
 neon_truncate(SMgrRelation reln, ForkNumber forknum, BlockNumber nblocks)
 {
 	XLogRecPtr	lsn;
@@ -3533,7 +3532,7 @@ neon_truncate(SMgrRelation reln, ForkNumber forknum, BlockNumber nblocks)
  * crash before the next checkpoint syncs the newly-inactive segment, that
  * segment may survive recovery, reintroducing unwanted data into the table.
  */
-void
+static void
 neon_immedsync(SMgrRelation reln, ForkNumber forknum)
 {
 	switch (reln->smgr_relpersistence)
@@ -3563,8 +3562,8 @@ neon_immedsync(SMgrRelation reln, ForkNumber forknum)
 }
 
 #if PG_MAJORVERSION_NUM >= 17
-void
-neon_regisersync(SMgrRelation reln, ForkNumber forknum)
+static void
+neon_registersync(SMgrRelation reln, ForkNumber forknum)
 {
 	switch (reln->smgr_relpersistence)
 	{
@@ -3748,6 +3747,8 @@ neon_read_slru_segment(SMgrRelation reln, const char* path, int segno, void* buf
 	SlruKind	kind;
 	int			n_blocks;
 	shardno_t	shard_no = 0; /* All SLRUs are at shard 0 */
+	NeonResponse *resp;
+	NeonGetSlruSegmentRequest request;
 
 	/*
 	 * Compute a request LSN to use, similar to neon_get_request_lsns() but the
@@ -3786,8 +3787,7 @@ neon_read_slru_segment(SMgrRelation reln, const char* path, int segno, void* buf
 	else
 		return -1;
 
-	NeonResponse *resp;
-	NeonGetSlruSegmentRequest request = {
+	request = (NeonGetSlruSegmentRequest) {
 		.req.tag = T_NeonGetSlruSegmentRequest,
 		.req.lsn = request_lsn,
 		.req.not_modified_since = not_modified_since,
@@ -3894,7 +3894,7 @@ static const struct f_smgr neon_smgr =
 	.smgr_truncate = neon_truncate,
 	.smgr_immedsync = neon_immedsync,
 #if PG_MAJORVERSION_NUM >= 17
-	.smgr_registersync = neon_regisersync,
+	.smgr_registersync = neon_registersync,
 #endif
 	.smgr_start_unlogged_build = neon_start_unlogged_build,
 	.smgr_finish_unlogged_build_phase_1 = neon_finish_unlogged_build_phase_1,

pgxn/neon/walproposer.c

@@ -252,8 +252,6 @@ WalProposerPoll(WalProposer *wp)
 		/* timeout expired: poll state */
 		if (rc == 0 || TimeToReconnect(wp, now) <= 0)
 		{
-			TimestampTz now;
-
 			/*
 			 * If no WAL was generated during timeout (and we have already
 			 * collected the quorum), then send empty keepalive message
@@ -269,8 +267,7 @@ WalProposerPoll(WalProposer *wp)
 			now = wp->api.get_current_timestamp(wp);
 			for (int i = 0; i < wp->n_safekeepers; i++)
 			{
-				Safekeeper *sk = &wp->safekeeper[i];
-
+				sk = &wp->safekeeper[i];
 				if (TimestampDifferenceExceeds(sk->latestMsgReceivedAt, now,
 											   wp->config->safekeeper_connection_timeout))
 				{
@@ -1080,7 +1077,7 @@ SendProposerElected(Safekeeper *sk)
 	ProposerElected msg;
 	TermHistory *th;
 	term_t		lastCommonTerm;
-	int			i;
+	int			idx;
 
 	/* Now that we are ready to send it's a good moment to create WAL reader */
 	wp->api.wal_reader_allocate(sk);
@@ -1099,15 +1096,15 @@ SendProposerElected(Safekeeper *sk)
 	/* We must start somewhere. */
 	Assert(wp->propTermHistory.n_entries >= 1);
 
-	for (i = 0; i < Min(wp->propTermHistory.n_entries, th->n_entries); i++)
+	for (idx = 0; idx < Min(wp->propTermHistory.n_entries, th->n_entries); idx++)
 	{
-		if (wp->propTermHistory.entries[i].term != th->entries[i].term)
+		if (wp->propTermHistory.entries[idx].term != th->entries[idx].term)
 			break;
 		/* term must begin everywhere at the same point */
-		Assert(wp->propTermHistory.entries[i].lsn == th->entries[i].lsn);
+		Assert(wp->propTermHistory.entries[idx].lsn == th->entries[idx].lsn);
 	}
-	i--;						/* step back to the last common term */
-	if (i < 0)
+	idx--;						/* step back to the last common term */
+	if (idx < 0)
 	{
 		/* safekeeper is empty or no common point, start from the beginning */
 		sk->startStreamingAt = wp->propTermHistory.entries[0].lsn;
@@ -1128,14 +1125,14 @@ SendProposerElected(Safekeeper *sk)
 		 * proposer, LSN it is currently writing, but then we just pick
 		 * safekeeper pos as it obviously can't be higher.
 		 */
-		if (wp->propTermHistory.entries[i].term == wp->propTerm)
+		if (wp->propTermHistory.entries[idx].term == wp->propTerm)
 		{
 			sk->startStreamingAt = sk->voteResponse.flushLsn;
 		}
 		else
 		{
-			XLogRecPtr	propEndLsn = wp->propTermHistory.entries[i + 1].lsn;
-			XLogRecPtr	skEndLsn = (i + 1 < th->n_entries ? th->entries[i + 1].lsn : sk->voteResponse.flushLsn);
+			XLogRecPtr	propEndLsn = wp->propTermHistory.entries[idx + 1].lsn;
+			XLogRecPtr	skEndLsn = (idx + 1 < th->n_entries ? th->entries[idx + 1].lsn : sk->voteResponse.flushLsn);
 
 			sk->startStreamingAt = Min(propEndLsn, skEndLsn);
 		}
@@ -1149,7 +1146,7 @@ SendProposerElected(Safekeeper *sk)
 	msg.termHistory = &wp->propTermHistory;
 	msg.timelineStartLsn = wp->timelineStartLsn;
 
-	lastCommonTerm = i >= 0 ? wp->propTermHistory.entries[i].term : 0;
+	lastCommonTerm = idx >= 0 ? wp->propTermHistory.entries[idx].term : 0;
 	wp_log(LOG,
 		   "sending elected msg to node " UINT64_FORMAT " term=" UINT64_FORMAT ", startStreamingAt=%X/%X (lastCommonTerm=" UINT64_FORMAT "), termHistory.n_entries=%u to %s:%s, timelineStartLsn=%X/%X",
 		   sk->greetResponse.nodeId, msg.term, LSN_FORMAT_ARGS(msg.startStreamingAt), lastCommonTerm, msg.termHistory->n_entries, sk->host, sk->port, LSN_FORMAT_ARGS(msg.timelineStartLsn));
@@ -1641,7 +1638,7 @@ UpdateDonorShmem(WalProposer *wp)
  * Process AppendResponse message from safekeeper.
  */
 static void
-HandleSafekeeperResponse(WalProposer *wp, Safekeeper *sk)
+HandleSafekeeperResponse(WalProposer *wp, Safekeeper *fromsk)
 {
 	XLogRecPtr	candidateTruncateLsn;
 	XLogRecPtr	newCommitLsn;
@@ -1660,7 +1657,7 @@ HandleSafekeeperResponse(WalProposer *wp, Safekeeper *sk)
 	 * and WAL is committed by the quorum. BroadcastAppendRequest() should be
 	 * called to notify safekeepers about the new commitLsn.
 	 */
-	wp->api.process_safekeeper_feedback(wp, sk);
+	wp->api.process_safekeeper_feedback(wp, fromsk);
 
 	/*
 	 * Try to advance truncateLsn -- the last record flushed to all

pgxn/neon/walproposer.h

@@ -725,7 +725,7 @@ extern void WalProposerBroadcast(WalProposer *wp, XLogRecPtr startpos, XLogRecPt
 extern void WalProposerPoll(WalProposer *wp);
 extern void WalProposerFree(WalProposer *wp);
 
-extern WalproposerShmemState *GetWalpropShmemState();
+extern WalproposerShmemState *GetWalpropShmemState(void);
 
 /*
  * WaitEventSet API doesn't allow to remove socket, so walproposer_pg uses it to
@@ -745,7 +745,7 @@ extern TimeLineID walprop_pg_get_timeline_id(void);
  * catch logging.
  */
 #ifdef WALPROPOSER_LIB
-extern void WalProposerLibLog(WalProposer *wp, int elevel, char *fmt,...);
+extern void WalProposerLibLog(WalProposer *wp, int elevel, char *fmt,...) pg_attribute_printf(3, 4);
 #define wp_log(elevel, fmt, ...) WalProposerLibLog(wp, elevel, fmt, ## __VA_ARGS__)
 #else
 #define wp_log(elevel, fmt, ...) elog(elevel, WP_LOG_PREFIX fmt, ## __VA_ARGS__)

pgxn/neon/walproposer_pg.c

@@ -286,6 +286,9 @@ safekeepers_cmp(char *old, char *new)
 static void
 assign_neon_safekeepers(const char *newval, void *extra)
 {
+	char	   *newval_copy;
+	char	   *oldval;
+
 	if (!am_walproposer)
 		return;
 
@@ -295,8 +298,8 @@ assign_neon_safekeepers(const char *newval, void *extra)
 	}
 
 	/* Copy values because we will modify them in split_safekeepers_list() */
-	char *newval_copy = pstrdup(newval);
-	char *oldval = pstrdup(wal_acceptors_list);
+	newval_copy = pstrdup(newval);
+	oldval = pstrdup(wal_acceptors_list);
 
 	/* 
 	 * TODO: restarting through FATAL is stupid and introduces 1s delay before
@@ -538,7 +541,7 @@ nwp_shmem_startup_hook(void)
 }
 
 WalproposerShmemState *
-GetWalpropShmemState()
+GetWalpropShmemState(void)
 {
 	Assert(walprop_shared != NULL);
 	return walprop_shared;

pgxn/neon_rmgr/neon_rmgr_desc.c

@@ -44,27 +44,6 @@ infobits_desc(StringInfo buf, uint8 infobits, const char *keyname)
 	appendStringInfoString(buf, "]");
 }
 
-static void
-truncate_flags_desc(StringInfo buf, uint8 flags)
-{
-	appendStringInfoString(buf, "flags: [");
-
-	if (flags & XLH_TRUNCATE_CASCADE)
-		appendStringInfoString(buf, "CASCADE, ");
-	if (flags & XLH_TRUNCATE_RESTART_SEQS)
-		appendStringInfoString(buf, "RESTART_SEQS, ");
-
-	if (buf->data[buf->len - 1] == ' ')
-	{
-		/* Truncate-away final unneeded ", "  */
-		Assert(buf->data[buf->len - 2] == ',');
-		buf->len -= 2;
-		buf->data[buf->len] = '\0';
-	}
-
-	appendStringInfoString(buf, "]");
-}
-
 void
 neon_rm_desc(StringInfo buf, XLogReaderState *record)
 {

pgxn/neon_walredo/walredoproc.c

@@ -136,7 +136,7 @@ static bool redo_block_filter(XLogReaderState *record, uint8 block_id);
 static void GetPage(StringInfo input_message);
 static void Ping(StringInfo input_message);
 static ssize_t buffered_read(void *buf, size_t count);
-static void CreateFakeSharedMemoryAndSemaphores();
+static void CreateFakeSharedMemoryAndSemaphores(void);
 
 static BufferTag target_redo_tag;
 
@@ -170,6 +170,40 @@ close_range_syscall(unsigned int start_fd, unsigned int count, unsigned int flag
     return syscall(__NR_close_range, start_fd, count, flags);
 }
 
+
+static PgSeccompRule allowed_syscalls[] =
+{
+	/* Hard requirements */
+	PG_SCMP_ALLOW(exit_group),
+	PG_SCMP_ALLOW(pselect6),
+	PG_SCMP_ALLOW(read),
+	PG_SCMP_ALLOW(select),
+	PG_SCMP_ALLOW(write),
+
+	/* Memory allocation */
+	PG_SCMP_ALLOW(brk),
+#ifndef MALLOC_NO_MMAP
+	/* TODO: musl doesn't have mallopt */
+	PG_SCMP_ALLOW(mmap),
+	PG_SCMP_ALLOW(munmap),
+#endif
+	/*
+	 * getpid() is called on assertion failure, in ExceptionalCondition.
+	 * It's not really needed, but seems pointless to hide it either. The
+	 * system call unlikely to expose a kernel vulnerability, and the PID
+	 * is stored in MyProcPid anyway.
+	 */
+	PG_SCMP_ALLOW(getpid),
+
+	/* Enable those for a proper shutdown. */
+#if 0
+	   PG_SCMP_ALLOW(munmap),
+	   PG_SCMP_ALLOW(shmctl),
+	   PG_SCMP_ALLOW(shmdt),
+	   PG_SCMP_ALLOW(unlink),	/* shm_unlink */
+#endif
+};
+
 static void
 enter_seccomp_mode(void)
 {
@@ -183,44 +217,12 @@ enter_seccomp_mode(void)
 				(errcode(ERRCODE_SYSTEM_ERROR),
 				 errmsg("seccomp: could not close files >= fd 3")));
 
-	PgSeccompRule syscalls[] =
-	{
-		/* Hard requirements */
-		PG_SCMP_ALLOW(exit_group),
-		PG_SCMP_ALLOW(pselect6),
-		PG_SCMP_ALLOW(read),
-		PG_SCMP_ALLOW(select),
-		PG_SCMP_ALLOW(write),
-
-		/* Memory allocation */
-		PG_SCMP_ALLOW(brk),
-#ifndef MALLOC_NO_MMAP
-		/* TODO: musl doesn't have mallopt */
-		PG_SCMP_ALLOW(mmap),
-		PG_SCMP_ALLOW(munmap),
-#endif
-		/*
-		 * getpid() is called on assertion failure, in ExceptionalCondition.
-		 * It's not really needed, but seems pointless to hide it either. The
-		 * system call unlikely to expose a kernel vulnerability, and the PID
-		 * is stored in MyProcPid anyway.
-		 */
-		PG_SCMP_ALLOW(getpid),
-
-		/* Enable those for a proper shutdown.
-		PG_SCMP_ALLOW(munmap),
-		PG_SCMP_ALLOW(shmctl),
-		PG_SCMP_ALLOW(shmdt),
-		PG_SCMP_ALLOW(unlink), // shm_unlink
-	 */
-	};
-
 #ifdef MALLOC_NO_MMAP
 	/* Ask glibc not to use mmap() */
 	mallopt(M_MMAP_MAX, 0);
 #endif
 
-	seccomp_load_rules(syscalls, lengthof(syscalls));
+	seccomp_load_rules(allowed_syscalls, lengthof(allowed_syscalls));
 }
 #endif /* HAVE_LIBSECCOMP */
 
@@ -449,7 +451,7 @@ WalRedoMain(int argc, char *argv[])
  * half-initialized postgres.
  */
 static void
-CreateFakeSharedMemoryAndSemaphores()
+CreateFakeSharedMemoryAndSemaphores(void)
 {
 	PGShmemHeader *shim = NULL;
 	PGShmemHeader *hdr;

poetry.lock

@@ -2095,6 +2095,7 @@ files = [
     {file = "psycopg2_binary-2.9.9-cp311-cp311-win32.whl", hash = "sha256:dc4926288b2a3e9fd7b50dc6a1909a13bbdadfc67d93f3374d984e56f885579d"},
     {file = "psycopg2_binary-2.9.9-cp311-cp311-win_amd64.whl", hash = "sha256:b76bedd166805480ab069612119ea636f5ab8f8771e640ae103e05a4aae3e417"},
     {file = "psycopg2_binary-2.9.9-cp312-cp312-macosx_10_9_x86_64.whl", hash = "sha256:8532fd6e6e2dc57bcb3bc90b079c60de896d2128c5d9d6f24a63875a95a088cf"},
+    {file = "psycopg2_binary-2.9.9-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:b0605eaed3eb239e87df0d5e3c6489daae3f7388d455d0c0b4df899519c6a38d"},
     {file = "psycopg2_binary-2.9.9-cp312-cp312-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:8f8544b092a29a6ddd72f3556a9fcf249ec412e10ad28be6a0c0d948924f2212"},
     {file = "psycopg2_binary-2.9.9-cp312-cp312-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:2d423c8d8a3c82d08fe8af900ad5b613ce3632a1249fd6a223941d0735fce493"},
     {file = "psycopg2_binary-2.9.9-cp312-cp312-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:2e5afae772c00980525f6d6ecf7cbca55676296b580c0e6abb407f15f3706996"},
@@ -2103,6 +2104,8 @@ files = [
     {file = "psycopg2_binary-2.9.9-cp312-cp312-musllinux_1_1_i686.whl", hash = "sha256:cb16c65dcb648d0a43a2521f2f0a2300f40639f6f8c1ecbc662141e4e3e1ee07"},
     {file = "psycopg2_binary-2.9.9-cp312-cp312-musllinux_1_1_ppc64le.whl", hash = "sha256:911dda9c487075abd54e644ccdf5e5c16773470a6a5d3826fda76699410066fb"},
     {file = "psycopg2_binary-2.9.9-cp312-cp312-musllinux_1_1_x86_64.whl", hash = "sha256:57fede879f08d23c85140a360c6a77709113efd1c993923c59fde17aa27599fe"},
+    {file = "psycopg2_binary-2.9.9-cp312-cp312-win32.whl", hash = "sha256:64cf30263844fa208851ebb13b0732ce674d8ec6a0c86a4e160495d299ba3c93"},
+    {file = "psycopg2_binary-2.9.9-cp312-cp312-win_amd64.whl", hash = "sha256:81ff62668af011f9a48787564ab7eded4e9fb17a4a6a74af5ffa6a457400d2ab"},
     {file = "psycopg2_binary-2.9.9-cp37-cp37m-macosx_10_9_x86_64.whl", hash = "sha256:2293b001e319ab0d869d660a704942c9e2cce19745262a8aba2115ef41a0a42a"},
     {file = "psycopg2_binary-2.9.9-cp37-cp37m-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:03ef7df18daf2c4c07e2695e8cfd5ee7f748a1d54d802330985a78d2a5a6dca9"},
     {file = "psycopg2_binary-2.9.9-cp37-cp37m-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:0a602ea5aff39bb9fac6308e9c9d82b9a35c2bf288e184a816002c9fae930b77"},
@@ -2584,6 +2587,7 @@ files = [
     {file = "PyYAML-6.0.1-cp311-cp311-win_amd64.whl", hash = "sha256:bf07ee2fef7014951eeb99f56f39c9bb4af143d8aa3c21b1677805985307da34"},
     {file = "PyYAML-6.0.1-cp312-cp312-macosx_10_9_x86_64.whl", hash = "sha256:855fb52b0dc35af121542a76b9a84f8d1cd886ea97c84703eaa6d88e37a2ad28"},
     {file = "PyYAML-6.0.1-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:40df9b996c2b73138957fe23a16a4f0ba614f4c0efce1e9406a184b6d07fa3a9"},
+    {file = "PyYAML-6.0.1-cp312-cp312-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:a08c6f0fe150303c1c6b71ebcd7213c2858041a7e01975da3a99aed1e7a378ef"},
     {file = "PyYAML-6.0.1-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:6c22bec3fbe2524cde73d7ada88f6566758a8f7227bfbf93a408a9d86bcc12a0"},
     {file = "PyYAML-6.0.1-cp312-cp312-musllinux_1_1_x86_64.whl", hash = "sha256:8d4e9c88387b0f5c7d5f281e55304de64cf7f9c0021a3525bd3b1c542da3b0e4"},
     {file = "PyYAML-6.0.1-cp312-cp312-win32.whl", hash = "sha256:d483d2cdf104e7c9fa60c544d92981f12ad66a457afae824d146093b8c294c54"},
@@ -2729,21 +2733,22 @@ use-chardet-on-py3 = ["chardet (>=3.0.2,<6)"]
 
 [[package]]
 name = "responses"
-version = "0.21.0"
+version = "0.25.3"
 description = "A utility library for mocking out the `requests` Python library."
 optional = false
-python-versions = ">=3.7"
+python-versions = ">=3.8"
 files = [
-    {file = "responses-0.21.0-py3-none-any.whl", hash = "sha256:2dcc863ba63963c0c3d9ee3fa9507cbe36b7d7b0fccb4f0bdfd9e96c539b1487"},
-    {file = "responses-0.21.0.tar.gz", hash = "sha256:b82502eb5f09a0289d8e209e7bad71ef3978334f56d09b444253d5ad67bf5253"},
+    {file = "responses-0.25.3-py3-none-any.whl", hash = "sha256:521efcbc82081ab8daa588e08f7e8a64ce79b91c39f6e62199b19159bea7dbcb"},
+    {file = "responses-0.25.3.tar.gz", hash = "sha256:617b9247abd9ae28313d57a75880422d55ec63c29d33d629697590a034358dba"},
 ]
 
 [package.dependencies]
-requests = ">=2.0,<3.0"
-urllib3 = ">=1.25.10"
+pyyaml = "*"
+requests = ">=2.30.0,<3.0"
+urllib3 = ">=1.25.10,<3.0"
 
 [package.extras]
-tests = ["coverage (>=6.0.0)", "flake8", "mypy", "pytest (>=7.0.0)", "pytest-asyncio", "pytest-cov", "pytest-localserver", "types-mock", "types-requests"]
+tests = ["coverage (>=6.0.0)", "flake8", "mypy", "pytest (>=7.0.0)", "pytest-asyncio", "pytest-cov", "pytest-httpserver", "tomli", "tomli-w", "types-PyYAML", "types-requests"]
 
 [[package]]
 name = "rfc3339-validator"
@@ -3137,6 +3142,16 @@ files = [
     {file = "wrapt-1.14.1-cp310-cp310-musllinux_1_1_x86_64.whl", hash = "sha256:8ad85f7f4e20964db4daadcab70b47ab05c7c1cf2a7c1e51087bfaa83831854c"},
     {file = "wrapt-1.14.1-cp310-cp310-win32.whl", hash = "sha256:a9a52172be0b5aae932bef82a79ec0a0ce87288c7d132946d645eba03f0ad8a8"},
     {file = "wrapt-1.14.1-cp310-cp310-win_amd64.whl", hash = "sha256:6d323e1554b3d22cfc03cd3243b5bb815a51f5249fdcbb86fda4bf62bab9e164"},
+    {file = "wrapt-1.14.1-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:ecee4132c6cd2ce5308e21672015ddfed1ff975ad0ac8d27168ea82e71413f55"},
+    {file = "wrapt-1.14.1-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:2020f391008ef874c6d9e208b24f28e31bcb85ccff4f335f15a3251d222b92d9"},
+    {file = "wrapt-1.14.1-cp311-cp311-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:2feecf86e1f7a86517cab34ae6c2f081fd2d0dac860cb0c0ded96d799d20b335"},
+    {file = "wrapt-1.14.1-cp311-cp311-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:240b1686f38ae665d1b15475966fe0472f78e71b1b4903c143a842659c8e4cb9"},
+    {file = "wrapt-1.14.1-cp311-cp311-manylinux_2_5_x86_64.manylinux1_x86_64.manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:a9008dad07d71f68487c91e96579c8567c98ca4c3881b9b113bc7b33e9fd78b8"},
+    {file = "wrapt-1.14.1-cp311-cp311-musllinux_1_1_aarch64.whl", hash = "sha256:6447e9f3ba72f8e2b985a1da758767698efa72723d5b59accefd716e9e8272bf"},
+    {file = "wrapt-1.14.1-cp311-cp311-musllinux_1_1_i686.whl", hash = "sha256:acae32e13a4153809db37405f5eba5bac5fbe2e2ba61ab227926a22901051c0a"},
+    {file = "wrapt-1.14.1-cp311-cp311-musllinux_1_1_x86_64.whl", hash = "sha256:49ef582b7a1152ae2766557f0550a9fcbf7bbd76f43fbdc94dd3bf07cc7168be"},
+    {file = "wrapt-1.14.1-cp311-cp311-win32.whl", hash = "sha256:358fe87cc899c6bb0ddc185bf3dbfa4ba646f05b1b0b9b5a27c2cb92c2cea204"},
+    {file = "wrapt-1.14.1-cp311-cp311-win_amd64.whl", hash = "sha256:26046cd03936ae745a502abf44dac702a5e6880b2b01c29aea8ddf3353b68224"},
     {file = "wrapt-1.14.1-cp35-cp35m-manylinux1_i686.whl", hash = "sha256:43ca3bbbe97af00f49efb06e352eae40434ca9d915906f77def219b88e85d907"},
     {file = "wrapt-1.14.1-cp35-cp35m-manylinux1_x86_64.whl", hash = "sha256:6b1a564e6cb69922c7fe3a678b9f9a3c54e72b469875aa8018f18b4d1dd1adf3"},
     {file = "wrapt-1.14.1-cp35-cp35m-manylinux2010_i686.whl", hash = "sha256:00b6d4ea20a906c0ca56d84f93065b398ab74b927a7a3dbd470f6fc503f95dc3"},

pre-commit.py

@@ -1,11 +1,12 @@
 #!/usr/bin/env python3
 
+from __future__ import annotations
+
 import argparse
 import enum
 import os
 import subprocess
 import sys
-from typing import List
 
 
 @enum.unique
@@ -55,12 +56,12 @@ def mypy() -> str:
     return "poetry run mypy"
 
 
-def get_commit_files() -> List[str]:
+def get_commit_files() -> list[str]:
     files = subprocess.check_output("git diff --cached --name-only --diff-filter=ACM".split())
     return files.decode().splitlines()
 
 
-def check(name: str, suffix: str, cmd: str, changed_files: List[str], no_color: bool = False):
+def check(name: str, suffix: str, cmd: str, changed_files: list[str], no_color: bool = False):
     print(f"Checking: {name} ", end="")
     applicable_files = list(filter(lambda fname: fname.strip().endswith(suffix), changed_files))
     if not applicable_files:

proxy/Cargo.toml

@@ -39,7 +39,7 @@ http.workspace = true
 humantime.workspace = true
 humantime-serde.workspace = true
 hyper0.workspace = true
-hyper1 = { package = "hyper", version = "1.2", features = ["server"] }
+hyper = { workspace = true, features = ["server", "http1", "http2"] }
 hyper-util = { version = "0.1", features = ["server", "http1", "http2", "tokio"] }
 http-body-util = { version = "0.1" }
 indexmap.workspace = true

proxy/src/auth/backend/jwt.rs

@@ -571,7 +571,7 @@ mod tests {
     use bytes::Bytes;
     use http::Response;
     use http_body_util::Full;
-    use hyper1::service::service_fn;
+    use hyper::service::service_fn;
     use hyper_util::rt::TokioIo;
     use rand::rngs::OsRng;
     use rsa::pkcs8::DecodePrivateKey;
@@ -736,7 +736,7 @@ X0n5X2/pBLJzxZc62ccvZYVnctBiFs6HbSnxpuMQCfkt/BcR/ttIepBQQIW86wHL
         });
 
         let listener = TcpListener::bind("0.0.0.0:0").await.unwrap();
-        let server = hyper1::server::conn::http1::Builder::new();
+        let server = hyper::server::conn::http1::Builder::new();
         let addr = listener.local_addr().unwrap();
         tokio::spawn(async move {
             loop {

proxy/src/http/health_server.rs

@@ -1,5 +1,5 @@
 use anyhow::{anyhow, bail};
-use hyper::{header::CONTENT_TYPE, Body, Request, Response, StatusCode};
+use hyper0::{header::CONTENT_TYPE, Body, Request, Response, StatusCode};
 use measured::{text::BufferedTextEncoder, MetricGroup};
 use metrics::NeonMetrics;
 use std::{
@@ -21,7 +21,7 @@ async fn status_handler(_: Request<Body>) -> Result<Response<Body>, ApiError> {
     json_response(StatusCode::OK, "")
 }
 
-fn make_router(metrics: AppMetrics) -> RouterBuilder<hyper::Body, ApiError> {
+fn make_router(metrics: AppMetrics) -> RouterBuilder<hyper0::Body, ApiError> {
     let state = Arc::new(Mutex::new(PrometheusHandler {
         encoder: BufferedTextEncoder::new(),
         metrics,
@@ -45,7 +45,7 @@ pub async fn task_main(
 
     let service = || RouterService::new(make_router(metrics).build()?);
 
-    hyper::Server::from_tcp(http_listener)?
+    hyper0::Server::from_tcp(http_listener)?
         .serve(service().map_err(|e| anyhow!(e))?)
         .await?;
 

proxy/src/http/mod.rs

@@ -9,7 +9,7 @@ use std::time::Duration;
 use anyhow::bail;
 use bytes::Bytes;
 use http_body_util::BodyExt;
-use hyper1::body::Body;
+use hyper::body::Body;
 use serde::de::DeserializeOwned;
 
 pub(crate) use reqwest::{Request, Response};

proxy/src/lib.rs

@@ -90,8 +90,6 @@ use tokio::task::JoinError;
 use tokio_util::sync::CancellationToken;
 use tracing::warn;
 
-extern crate hyper0 as hyper;
-
 pub mod auth;
 pub mod cache;
 pub mod cancellation;

proxy/src/proxy/wake_compute.rs

@@ -7,7 +7,7 @@ use crate::metrics::{
     WakeupFailureKind,
 };
 use crate::proxy::retry::{retry_after, should_retry};
-use hyper1::StatusCode;
+use hyper::StatusCode;
 use tracing::{error, info, warn};
 
 use super::connect_compute::ComputeConnectBackend;

proxy/src/serverless/backend.rs

@@ -257,7 +257,7 @@ pub(crate) enum LocalProxyConnError {
     #[error("error with connection to local-proxy")]
     Io(#[source] std::io::Error),
     #[error("could not establish h2 connection")]
-    H2(#[from] hyper1::Error),
+    H2(#[from] hyper::Error),
 }
 
 impl ReportableError for HttpConnError {
@@ -481,7 +481,7 @@ async fn connect_http2(
         };
     };
 
-    let (client, connection) = hyper1::client::conn::http2::Builder::new(TokioExecutor::new())
+    let (client, connection) = hyper::client::conn::http2::Builder::new(TokioExecutor::new())
         .timer(TokioTimer::new())
         .keep_alive_interval(Duration::from_secs(20))
         .keep_alive_while_idle(true)

proxy/src/serverless/http_conn_pool.rs

@@ -1,5 +1,5 @@
 use dashmap::DashMap;
-use hyper1::client::conn::http2;
+use hyper::client::conn::http2;
 use hyper_util::rt::{TokioExecutor, TokioIo};
 use parking_lot::RwLock;
 use rand::Rng;
@@ -18,9 +18,9 @@ use tracing::{info, info_span, Instrument};
 
 use super::conn_pool::ConnInfo;
 
-pub(crate) type Send = http2::SendRequest<hyper1::body::Incoming>;
+pub(crate) type Send = http2::SendRequest<hyper::body::Incoming>;
 pub(crate) type Connect =
-    http2::Connection<TokioIo<TcpStream>, hyper1::body::Incoming, TokioExecutor>;
+    http2::Connection<TokioIo<TcpStream>, hyper::body::Incoming, TokioExecutor>;
 
 #[derive(Clone)]
 struct ConnPoolEntry {

proxy/src/serverless/http_util.rs

@@ -11,7 +11,7 @@ use serde::Serialize;
 use utils::http::error::ApiError;
 
 /// Like [`ApiError::into_response`]
-pub(crate) fn api_error_into_response(this: ApiError) -> Response<BoxBody<Bytes, hyper1::Error>> {
+pub(crate) fn api_error_into_response(this: ApiError) -> Response<BoxBody<Bytes, hyper::Error>> {
     match this {
         ApiError::BadRequest(err) => HttpErrorBody::response_from_msg_and_status(
             format!("{err:#?}"), // use debug printing so that we give the cause
@@ -67,12 +67,12 @@ impl HttpErrorBody {
     fn response_from_msg_and_status(
         msg: String,
         status: StatusCode,
-    ) -> Response<BoxBody<Bytes, hyper1::Error>> {
+    ) -> Response<BoxBody<Bytes, hyper::Error>> {
         HttpErrorBody { msg }.to_response(status)
     }
 
     /// Same as [`utils::http::error::HttpErrorBody::to_response`]
-    fn to_response(&self, status: StatusCode) -> Response<BoxBody<Bytes, hyper1::Error>> {
+    fn to_response(&self, status: StatusCode) -> Response<BoxBody<Bytes, hyper::Error>> {
         Response::builder()
             .status(status)
             .header(http::header::CONTENT_TYPE, "application/json")
@@ -90,7 +90,7 @@ impl HttpErrorBody {
 pub(crate) fn json_response<T: Serialize>(
     status: StatusCode,
     data: T,
-) -> Result<Response<BoxBody<Bytes, hyper1::Error>>, ApiError> {
+) -> Result<Response<BoxBody<Bytes, hyper::Error>>, ApiError> {
     let json = serde_json::to_string(&data)
         .context("Failed to serialize JSON response")
         .map_err(ApiError::InternalServerError)?;

proxy/src/serverless/mod.rs

@@ -22,7 +22,7 @@ use futures::TryFutureExt;
 use http::{Method, Response, StatusCode};
 use http_body_util::combinators::BoxBody;
 use http_body_util::{BodyExt, Empty};
-use hyper1::body::Incoming;
+use hyper::body::Incoming;
 use hyper_util::rt::TokioExecutor;
 use hyper_util::server::conn::auto::Builder;
 use rand::rngs::StdRng;
@@ -302,7 +302,7 @@ async fn connection_handler(
     let server = Builder::new(TokioExecutor::new());
     let conn = server.serve_connection_with_upgrades(
         hyper_util::rt::TokioIo::new(conn),
-        hyper1::service::service_fn(move |req: hyper1::Request<Incoming>| {
+        hyper::service::service_fn(move |req: hyper::Request<Incoming>| {
             // First HTTP request shares the same session ID
             let session_id = session_id.take().unwrap_or_else(uuid::Uuid::new_v4);
 
@@ -355,7 +355,7 @@ async fn connection_handler(
 
 #[allow(clippy::too_many_arguments)]
 async fn request_handler(
-    mut request: hyper1::Request<Incoming>,
+    mut request: hyper::Request<Incoming>,
     config: &'static ProxyConfig,
     backend: Arc<PoolingBackend>,
     ws_connections: TaskTracker,
@@ -365,7 +365,7 @@ async fn request_handler(
     // used to cancel in-flight HTTP requests. not used to cancel websockets
     http_cancellation_token: CancellationToken,
     endpoint_rate_limiter: Arc<EndpointRateLimiter>,
-) -> Result<Response<BoxBody<Bytes, hyper1::Error>>, ApiError> {
+) -> Result<Response<BoxBody<Bytes, hyper::Error>>, ApiError> {
     let host = request
         .headers()
         .get("host")

proxy/src/serverless/sql_over_http.rs

@@ -12,14 +12,14 @@ use http::Method;
 use http_body_util::combinators::BoxBody;
 use http_body_util::BodyExt;
 use http_body_util::Full;
-use hyper1::body::Body;
-use hyper1::body::Incoming;
-use hyper1::header;
-use hyper1::http::HeaderName;
-use hyper1::http::HeaderValue;
-use hyper1::Response;
-use hyper1::StatusCode;
-use hyper1::{HeaderMap, Request};
+use hyper::body::Body;
+use hyper::body::Incoming;
+use hyper::header;
+use hyper::http::HeaderName;
+use hyper::http::HeaderValue;
+use hyper::Response;
+use hyper::StatusCode;
+use hyper::{HeaderMap, Request};
 use pq_proto::StartupMessageParamsBuilder;
 use serde::Serialize;
 use serde_json::Value;
@@ -272,7 +272,7 @@ pub(crate) async fn handle(
     request: Request<Incoming>,
     backend: Arc<PoolingBackend>,
     cancel: CancellationToken,
-) -> Result<Response<BoxBody<Bytes, hyper1::Error>>, ApiError> {
+) -> Result<Response<BoxBody<Bytes, hyper::Error>>, ApiError> {
     let result = handle_inner(cancel, config, &ctx, request, backend).await;
 
     let mut response = match result {
@@ -435,7 +435,7 @@ impl UserFacingError for SqlOverHttpError {
 #[derive(Debug, thiserror::Error)]
 pub(crate) enum ReadPayloadError {
     #[error("could not read the HTTP request body: {0}")]
-    Read(#[from] hyper1::Error),
+    Read(#[from] hyper::Error),
     #[error("could not parse the HTTP request body: {0}")]
     Parse(#[from] serde_json::Error),
 }
@@ -476,7 +476,7 @@ struct HttpHeaders {
 }
 
 impl HttpHeaders {
-    fn try_parse(headers: &hyper1::http::HeaderMap) -> Result<Self, SqlOverHttpError> {
+    fn try_parse(headers: &hyper::http::HeaderMap) -> Result<Self, SqlOverHttpError> {
         // Determine the output options. Default behaviour is 'false'. Anything that is not
         // strictly 'true' assumed to be false.
         let raw_output = headers.get(&RAW_TEXT_OUTPUT) == Some(&HEADER_VALUE_TRUE);
@@ -529,7 +529,7 @@ async fn handle_inner(
     ctx: &RequestMonitoring,
     request: Request<Incoming>,
     backend: Arc<PoolingBackend>,
-) -> Result<Response<BoxBody<Bytes, hyper1::Error>>, SqlOverHttpError> {
+) -> Result<Response<BoxBody<Bytes, hyper::Error>>, SqlOverHttpError> {
     let _requeset_gauge = Metrics::get()
         .proxy
         .connection_requests
@@ -577,7 +577,7 @@ async fn handle_db_inner(
     conn_info: ConnInfo,
     auth: AuthData,
     backend: Arc<PoolingBackend>,
-) -> Result<Response<BoxBody<Bytes, hyper1::Error>>, SqlOverHttpError> {
+) -> Result<Response<BoxBody<Bytes, hyper::Error>>, SqlOverHttpError> {
     //
     // Determine the destination and connection params
     //
@@ -744,7 +744,7 @@ async fn handle_auth_broker_inner(
     conn_info: ConnInfo,
     jwt: String,
     backend: Arc<PoolingBackend>,
-) -> Result<Response<BoxBody<Bytes, hyper1::Error>>, SqlOverHttpError> {
+) -> Result<Response<BoxBody<Bytes, hyper::Error>>, SqlOverHttpError> {
     backend
         .authenticate_with_jwt(
             ctx,

proxy/src/serverless/websocket.rs

@@ -12,7 +12,7 @@ use anyhow::Context as _;
 use bytes::{Buf, BufMut, Bytes, BytesMut};
 use framed_websockets::{Frame, OpCode, WebSocketServer};
 use futures::{Sink, Stream};
-use hyper1::upgrade::OnUpgrade;
+use hyper::upgrade::OnUpgrade;
 use hyper_util::rt::TokioIo;
 use pin_project_lite::pin_project;
 

proxy/src/usage_metrics.rs

@@ -485,49 +485,51 @@ async fn upload_events_chunk(
 
 #[cfg(test)]
 mod tests {
-    use std::{
-        net::TcpListener,
-        sync::{Arc, Mutex},
-    };
+    use super::*;
 
+    use crate::{http, BranchId, EndpointId};
     use anyhow::Error;
     use chrono::Utc;
     use consumption_metrics::{Event, EventChunk};
-    use hyper::{
-        service::{make_service_fn, service_fn},
-        Body, Response,
-    };
+    use http_body_util::BodyExt;
+    use hyper::{body::Incoming, server::conn::http1, service::service_fn, Request, Response};
+    use hyper_util::rt::TokioIo;
+    use std::sync::{Arc, Mutex};
+    use tokio::net::TcpListener;
     use url::Url;
 
-    use super::*;
-    use crate::{http, BranchId, EndpointId};
-
     #[tokio::test]
     async fn metrics() {
-        let listener = TcpListener::bind("0.0.0.0:0").unwrap();
+        type Report = EventChunk<'static, Event<Ids, String>>;
+        let reports: Arc<Mutex<Vec<Report>>> = Arc::default();
 
-        let reports = Arc::new(Mutex::new(vec![]));
-        let reports2 = reports.clone();
-
-        let server = hyper::server::Server::from_tcp(listener)
-            .unwrap()
-            .serve(make_service_fn(move |_| {
-                let reports = reports.clone();
-                async move {
-                    Ok::<_, Error>(service_fn(move |req| {
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+        tokio::spawn({
+            let reports = reports.clone();
+            async move {
+                loop {
+                    if let Ok((stream, _addr)) = listener.accept().await {
                         let reports = reports.clone();
-                        async move {
-                            let bytes = hyper::body::to_bytes(req.into_body()).await?;
-                            let events: EventChunk<'static, Event<Ids, String>> =
-                                serde_json::from_slice(&bytes)?;
-                            reports.lock().unwrap().push(events);
-                            Ok::<_, Error>(Response::new(Body::from(vec![])))
-                        }
-                    }))
+                        http1::Builder::new()
+                            .serve_connection(
+                                TokioIo::new(stream),
+                                service_fn(move |req: Request<Incoming>| {
+                                    let reports = reports.clone();
+                                    async move {
+                                        let bytes = req.into_body().collect().await?.to_bytes();
+                                        let events = serde_json::from_slice(&bytes)?;
+                                        reports.lock().unwrap().push(events);
+                                        Ok::<_, Error>(Response::new(String::new()))
+                                    }
+                                }),
+                            )
+                            .await
+                            .unwrap();
+                    }
                 }
-            }));
-        let addr = server.local_addr();
-        tokio::spawn(server);
+            }
+        });
 
         let metrics = Metrics::default();
         let client = http::new_client();
@@ -536,7 +538,7 @@ mod tests {
 
         // no counters have been registered
         collect_metrics_iteration(&metrics.endpoints, &client, &endpoint, "foo", now, now).await;
-        let r = std::mem::take(&mut *reports2.lock().unwrap());
+        let r = std::mem::take(&mut *reports.lock().unwrap());
         assert!(r.is_empty());
 
         // register a new counter
@@ -548,7 +550,7 @@ mod tests {
 
         // the counter should be observed despite 0 egress
         collect_metrics_iteration(&metrics.endpoints, &client, &endpoint, "foo", now, now).await;
-        let r = std::mem::take(&mut *reports2.lock().unwrap());
+        let r = std::mem::take(&mut *reports.lock().unwrap());
         assert_eq!(r.len(), 1);
         assert_eq!(r[0].events.len(), 1);
         assert_eq!(r[0].events[0].value, 0);
@@ -558,7 +560,7 @@ mod tests {
 
         // egress should be observered
         collect_metrics_iteration(&metrics.endpoints, &client, &endpoint, "foo", now, now).await;
-        let r = std::mem::take(&mut *reports2.lock().unwrap());
+        let r = std::mem::take(&mut *reports.lock().unwrap());
         assert_eq!(r.len(), 1);
         assert_eq!(r[0].events.len(), 1);
         assert_eq!(r[0].events[0].value, 1);
@@ -568,7 +570,7 @@ mod tests {
 
         // we do not observe the counter
         collect_metrics_iteration(&metrics.endpoints, &client, &endpoint, "foo", now, now).await;
-        let r = std::mem::take(&mut *reports2.lock().unwrap());
+        let r = std::mem::take(&mut *reports.lock().unwrap());
         assert!(r.is_empty());
 
         // counter is unregistered

pyproject.toml

@@ -97,5 +97,8 @@ select = [
     "I", # isort
     "W", # pycodestyle
     "B", # bugbear
-    "UP032", # f-string
+    "UP", # pyupgrade
 ]
+
+[tool.ruff.lint.pyupgrade]
+keep-runtime-typing = true # Remove this stanza when we require Python 3.10

safekeeper/src/metrics.rs

@@ -12,8 +12,8 @@ use metrics::{
     core::{AtomicU64, Collector, Desc, GenericCounter, GenericGaugeVec, Opts},
     proto::MetricFamily,
     register_histogram_vec, register_int_counter, register_int_counter_pair,
-    register_int_counter_pair_vec, register_int_counter_vec, Gauge, HistogramVec, IntCounter,
-    IntCounterPair, IntCounterPairVec, IntCounterVec, IntGaugeVec,
+    register_int_counter_pair_vec, register_int_counter_vec, register_int_gauge, Gauge,
+    HistogramVec, IntCounter, IntCounterPair, IntCounterPairVec, IntCounterVec, IntGaugeVec,
 };
 use once_cell::sync::Lazy;
 
@@ -231,6 +231,14 @@ pub(crate) static EVICTION_EVENTS_COMPLETED: Lazy<IntCounterVec> = Lazy::new(||
     .expect("Failed to register metric")
 });
 
+pub static NUM_EVICTED_TIMELINES: Lazy<IntGauge> = Lazy::new(|| {
+    register_int_gauge!(
+        "safekeeper_evicted_timelines",
+        "Number of currently evicted timelines"
+    )
+    .expect("Failed to register metric")
+});
+
 pub const LABEL_UNKNOWN: &str = "unknown";
 
 /// Labels for traffic metrics.

safekeeper/src/timeline.rs

@@ -631,13 +631,19 @@ impl Timeline {
 
             return Err(e);
         }
-        self.bootstrap(conf, broker_active_set, partial_backup_rate_limiter);
+        self.bootstrap(
+            shared_state,
+            conf,
+            broker_active_set,
+            partial_backup_rate_limiter,
+        );
         Ok(())
     }
 
     /// Bootstrap new or existing timeline starting background tasks.
     pub fn bootstrap(
         self: &Arc<Timeline>,
+        _shared_state: &mut WriteGuardSharedState<'_>,
         conf: &SafeKeeperConf,
         broker_active_set: Arc<TimelinesSet>,
         partial_backup_rate_limiter: RateLimiter,

safekeeper/src/timeline_eviction.rs

@@ -15,7 +15,9 @@ use tracing::{debug, info, instrument, warn};
 use utils::crashsafe::durable_rename;
 
 use crate::{
-    metrics::{EvictionEvent, EVICTION_EVENTS_COMPLETED, EVICTION_EVENTS_STARTED},
+    metrics::{
+        EvictionEvent, EVICTION_EVENTS_COMPLETED, EVICTION_EVENTS_STARTED, NUM_EVICTED_TIMELINES,
+    },
     rate_limit::rand_duration,
     timeline_manager::{Manager, StateSnapshot},
     wal_backup,
@@ -93,6 +95,7 @@ impl Manager {
         }
 
         info!("successfully evicted timeline");
+        NUM_EVICTED_TIMELINES.inc();
     }
 
     /// Attempt to restore evicted timeline from remote storage; it must be
@@ -128,6 +131,7 @@ impl Manager {
             tokio::time::Instant::now() + rand_duration(&self.conf.eviction_min_resident);
 
         info!("successfully restored evicted timeline");
+        NUM_EVICTED_TIMELINES.dec();
     }
 }
 

safekeeper/src/timeline_manager.rs

@@ -25,7 +25,10 @@ use utils::lsn::Lsn;
 
 use crate::{
     control_file::{FileStorage, Storage},
-    metrics::{MANAGER_ACTIVE_CHANGES, MANAGER_ITERATIONS_TOTAL, MISC_OPERATION_SECONDS},
+    metrics::{
+        MANAGER_ACTIVE_CHANGES, MANAGER_ITERATIONS_TOTAL, MISC_OPERATION_SECONDS,
+        NUM_EVICTED_TIMELINES,
+    },
     rate_limit::{rand_duration, RateLimiter},
     recovery::recovery_main,
     remove_wal::calc_horizon_lsn,
@@ -251,6 +254,11 @@ pub async fn main_task(
         mgr.recovery_task = Some(tokio::spawn(recovery_main(tli, mgr.conf.clone())));
     }
 
+    // If timeline is evicted, reflect that in the metric.
+    if mgr.is_offloaded {
+        NUM_EVICTED_TIMELINES.inc();
+    }
+
     let last_state = 'outer: loop {
         MANAGER_ITERATIONS_TOTAL.inc();
 
@@ -367,6 +375,11 @@ pub async fn main_task(
         mgr.update_wal_removal_end(res);
     }
 
+    // If timeline is deleted while evicted decrement the gauge.
+    if mgr.tli.is_cancelled() && mgr.is_offloaded {
+        NUM_EVICTED_TIMELINES.dec();
+    }
+
     mgr.set_status(Status::Finished);
 }
 

safekeeper/src/timelines_global_map.rs

@@ -165,12 +165,14 @@ impl GlobalTimelines {
                         match Timeline::load_timeline(&conf, ttid) {
                             Ok(timeline) => {
                                 let tli = Arc::new(timeline);
+                                let mut shared_state = tli.write_shared_state().await;
                                 TIMELINES_STATE
                                     .lock()
                                     .unwrap()
                                     .timelines
                                     .insert(ttid, tli.clone());
                                 tli.bootstrap(
+                                    &mut shared_state,
                                     &conf,
                                     broker_active_set.clone(),
                                     partial_backup_rate_limiter.clone(),
@@ -213,6 +215,7 @@ impl GlobalTimelines {
         match Timeline::load_timeline(&conf, ttid) {
             Ok(timeline) => {
                 let tli = Arc::new(timeline);
+                let mut shared_state = tli.write_shared_state().await;
 
                 // TODO: prevent concurrent timeline creation/loading
                 {
@@ -227,8 +230,13 @@ impl GlobalTimelines {
                     state.timelines.insert(ttid, tli.clone());
                 }
 
-                tli.bootstrap(&conf, broker_active_set, partial_backup_rate_limiter);
-
+                tli.bootstrap(
+                    &mut shared_state,
+                    &conf,
+                    broker_active_set,
+                    partial_backup_rate_limiter,
+                );
+                drop(shared_state);
                 Ok(tli)
             }
             // If we can't load a timeline, it's bad. Caller will figure it out.

safekeeper/src/wal_backup.rs

@@ -17,7 +17,9 @@ use std::time::Duration;
 use postgres_ffi::v14::xlog_utils::XLogSegNoOffsetToRecPtr;
 use postgres_ffi::XLogFileName;
 use postgres_ffi::{XLogSegNo, PG_TLI};
-use remote_storage::{GenericRemoteStorage, ListingMode, RemotePath, StorageMetadata};
+use remote_storage::{
+    DownloadOpts, GenericRemoteStorage, ListingMode, RemotePath, StorageMetadata,
+};
 use tokio::fs::File;
 
 use tokio::select;
@@ -503,8 +505,12 @@ pub async fn read_object(
 
     let cancel = CancellationToken::new();
 
+    let opts = DownloadOpts {
+        byte_start: std::ops::Bound::Included(offset),
+        ..Default::default()
+    };
     let download = storage
-        .download_storage_object(Some((offset, None)), file_path, &cancel)
+        .download(file_path, &opts, &cancel)
         .await
         .with_context(|| {
             format!("Failed to open WAL segment download stream for remote path {file_path:?}")

scripts/benchmark_durations.py

@@ -1,9 +1,10 @@
 #! /usr/bin/env python3
 
+from __future__ import annotations
+
 import argparse
 import json
 import logging
-from typing import Dict
 
 import psycopg2
 import psycopg2.extras
@@ -110,7 +111,7 @@ def main(args: argparse.Namespace):
     output = args.output
     percentile = args.percentile
 
-    res: Dict[str, float] = {}
+    res: dict[str, float] = {}
 
     try:
         logging.info("connecting to the database...")

scripts/download_basebackup.py

@@ -4,6 +4,9 @@
 #
 # This can be useful in disaster recovery.
 #
+
+from __future__ import annotations
+
 import argparse
 
 import psycopg2

scripts/flaky_tests.py

@@ -1,16 +1,21 @@
 #! /usr/bin/env python3
 
+from __future__ import annotations
+
 import argparse
 import json
 import logging
 import os
 from collections import defaultdict
-from typing import Any, DefaultDict, Dict, Optional
+from typing import TYPE_CHECKING
 
 import psycopg2
 import psycopg2.extras
 import toml
 
+if TYPE_CHECKING:
+    from typing import Any, Optional
+
 FLAKY_TESTS_QUERY = """
     SELECT
         DISTINCT parent_suite, suite, name
@@ -33,7 +38,7 @@ def main(args: argparse.Namespace):
     build_type = args.build_type
     pg_version = args.pg_version
 
-    res: DefaultDict[str, DefaultDict[str, Dict[str, bool]]]
+    res: defaultdict[str, defaultdict[str, dict[str, bool]]]
     res = defaultdict(lambda: defaultdict(dict))
 
     try:
@@ -60,7 +65,7 @@ def main(args: argparse.Namespace):
         pageserver_virtual_file_io_engine_parameter = ""
 
     # re-use existing records of flaky tests from before parametrization by compaction_algorithm
-    def get_pageserver_default_tenant_config_compaction_algorithm() -> Optional[Dict[str, Any]]:
+    def get_pageserver_default_tenant_config_compaction_algorithm() -> Optional[dict[str, Any]]:
         """Duplicated from parametrize.py"""
         toml_table = os.getenv("PAGESERVER_DEFAULT_TENANT_CONFIG_COMPACTION_ALGORITHM")
         if toml_table is None:

scripts/force_layer_download.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import argparse
 import asyncio
 import json
@@ -5,11 +7,15 @@ import logging
 import signal
 import sys
 from collections import defaultdict
+from collections.abc import Awaitable
 from dataclasses import dataclass
-from typing import Any, Awaitable, Dict, List, Tuple
+from typing import TYPE_CHECKING
 
 import aiohttp
 
+if TYPE_CHECKING:
+    from typing import Any
+
 
 class ClientException(Exception):
     pass
@@ -89,7 +95,7 @@ class Client:
 class Completed:
     """The status dict returned by the API"""
 
-    status: Dict[str, Any]
+    status: dict[str, Any]
 
 
 sigint_received = asyncio.Event()
@@ -179,7 +185,7 @@ async def main_impl(args, report_out, client: Client):
     """
     Returns OS exit status.
     """
-    tenant_and_timline_ids: List[Tuple[str, str]] = []
+    tenant_and_timline_ids: list[tuple[str, str]] = []
     # fill tenant_and_timline_ids based on spec
     for spec in args.what:
         comps = spec.split(":")
@@ -215,14 +221,14 @@ async def main_impl(args, report_out, client: Client):
     tenant_and_timline_ids = tmp
 
     logging.info("create tasks and process them at specified concurrency")
-    task_q: asyncio.Queue[Tuple[str, Awaitable[Any]]] = asyncio.Queue()
+    task_q: asyncio.Queue[tuple[str, Awaitable[Any]]] = asyncio.Queue()
     tasks = {
         f"{tid}:{tlid}": do_timeline(client, tid, tlid) for tid, tlid in tenant_and_timline_ids
     }
     for task in tasks.items():
         task_q.put_nowait(task)
 
-    result_q: asyncio.Queue[Tuple[str, Any]] = asyncio.Queue()
+    result_q: asyncio.Queue[tuple[str, Any]] = asyncio.Queue()
     taskq_handlers = []
     for _ in range(0, args.concurrent_tasks):
         taskq_handlers.append(taskq_handler(task_q, result_q))

scripts/ingest_perf_test_result.py

@@ -1,4 +1,7 @@
 #!/usr/bin/env python3
+
+from __future__ import annotations
+
 import argparse
 import json
 import logging

scripts/ingest_regress_test_result-new-format.py

@@ -1,5 +1,7 @@
 #! /usr/bin/env python3
 
+from __future__ import annotations
+
 import argparse
 import dataclasses
 import json
@@ -11,7 +13,6 @@ from contextlib import contextmanager
 from dataclasses import dataclass
 from datetime import datetime, timezone
 from pathlib import Path
-from typing import Tuple
 
 import backoff
 import psycopg2
@@ -91,7 +92,7 @@ def create_table(cur):
     cur.execute(CREATE_TABLE)
 
 
-def parse_test_name(test_name: str) -> Tuple[str, int, str]:
+def parse_test_name(test_name: str) -> tuple[str, int, str]:
     build_type, pg_version = None, None
     if match := TEST_NAME_RE.search(test_name):
         found = match.groupdict()

scripts/sk_cleanup_tenants/script.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import argparse
 import logging
 import os

storage_controller/src/reconciler.rs

@@ -22,7 +22,7 @@ use utils::sync::gate::GateGuard;
 
 use crate::compute_hook::{ComputeHook, NotifyError};
 use crate::node::Node;
-use crate::tenant_shard::{IntentState, ObservedState, ObservedStateLocation};
+use crate::tenant_shard::{IntentState, ObservedState, ObservedStateDelta, ObservedStateLocation};
 
 const DEFAULT_HEATMAP_PERIOD: &str = "60s";
 
@@ -45,8 +45,15 @@ pub(super) struct Reconciler {
     pub(crate) reconciler_config: ReconcilerConfig,
 
     pub(crate) config: TenantConfig,
+
+    /// Observed state from the point of view of the reconciler.
+    /// This gets updated as the reconciliation makes progress.
     pub(crate) observed: ObservedState,
 
+    /// Snapshot of the observed state at the point when the reconciler
+    /// was spawned.
+    pub(crate) original_observed: ObservedState,
+
     pub(crate) service_config: service::Config,
 
     /// A hook to notify the running postgres instances when we change the location
@@ -846,6 +853,39 @@ impl Reconciler {
         }
     }
 
+    /// Compare the observed state snapshot from when the reconcile was created
+    /// with the final observed state in order to generate observed state deltas.
+    pub(crate) fn observed_deltas(&self) -> Vec<ObservedStateDelta> {
+        let mut deltas = Vec::default();
+
+        for (node_id, location) in &self.observed.locations {
+            let previous_location = self.original_observed.locations.get(node_id);
+            let do_upsert = match previous_location {
+                // Location config changed for node
+                Some(prev) if location.conf != prev.conf => true,
+                // New location config for node
+                None => true,
+                // Location config has not changed for node
+                _ => false,
+            };
+
+            if do_upsert {
+                deltas.push(ObservedStateDelta::Upsert(Box::new((
+                    *node_id,
+                    location.clone(),
+                ))));
+            }
+        }
+
+        for node_id in self.original_observed.locations.keys() {
+            if !self.observed.locations.contains_key(node_id) {
+                deltas.push(ObservedStateDelta::Delete(*node_id));
+            }
+        }
+
+        deltas
+    }
+
     /// Keep trying to notify the compute indefinitely, only dropping out if:
     /// - the node `origin` becomes unavailable -> Ok(())
     /// - the node `origin` no longer has our tenant shard attached -> Ok(())

storage_controller/src/service.rs

@@ -28,8 +28,8 @@ use crate::{
     reconciler::{ReconcileError, ReconcileUnits, ReconcilerConfig, ReconcilerConfigBuilder},
     scheduler::{MaySchedule, ScheduleContext, ScheduleError, ScheduleMode},
     tenant_shard::{
-        MigrateAttachment, ReconcileNeeded, ReconcilerStatus, ScheduleOptimization,
-        ScheduleOptimizationAction,
+        MigrateAttachment, ObservedStateDelta, ReconcileNeeded, ReconcilerStatus,
+        ScheduleOptimization, ScheduleOptimizationAction,
     },
 };
 use anyhow::Context;
@@ -966,6 +966,8 @@ impl Service {
 
             let res = self.heartbeater.heartbeat(nodes).await;
             if let Ok(deltas) = res {
+                let mut to_handle = Vec::default();
+
                 for (node_id, state) in deltas.0 {
                     let new_availability = match state {
                         PageserverState::Available { utilization, .. } => {
@@ -997,14 +999,27 @@ impl Service {
                         }
                     };
 
+                    let node_lock = trace_exclusive_lock(
+                        &self.node_op_locks,
+                        node_id,
+                        NodeOperations::Configure,
+                    )
+                    .await;
+
                     // This is the code path for geniune availability transitions (i.e node
                     // goes unavailable and/or comes back online).
                     let res = self
-                        .node_configure(node_id, Some(new_availability), None)
+                        .node_state_configure(node_id, Some(new_availability), None, &node_lock)
                         .await;
 
                     match res {
-                        Ok(()) => {}
+                        Ok(transition) => {
+                            // Keep hold of the lock until the availability transitions
+                            // have been handled in
+                            // [`Service::handle_node_availability_transitions`] in order avoid
+                            // racing with [`Service::external_node_configure`].
+                            to_handle.push((node_id, node_lock, transition));
+                        }
                         Err(ApiError::NotFound(_)) => {
                             // This should be rare, but legitimate since the heartbeats are done
                             // on a snapshot of the nodes.
@@ -1014,13 +1029,37 @@ impl Service {
                             // Transition to active involves reconciling: if a node responds to a heartbeat then
                             // becomes unavailable again, we may get an error here.
                             tracing::error!(
-                                "Failed to update node {} after heartbeat round: {}",
+                                "Failed to update node state {} after heartbeat round: {}",
                                 node_id,
                                 err
                             );
                         }
                     }
                 }
+
+                // We collected all the transitions above and now we handle them.
+                let res = self.handle_node_availability_transitions(to_handle).await;
+                if let Err(errs) = res {
+                    for (node_id, err) in errs {
+                        match err {
+                            ApiError::NotFound(_) => {
+                                // This should be rare, but legitimate since the heartbeats are done
+                                // on a snapshot of the nodes.
+                                tracing::info!(
+                                    "Node {} was not found after heartbeat round",
+                                    node_id
+                                );
+                            }
+                            err => {
+                                tracing::error!(
+                                    "Failed to handle availability transition for {} after heartbeat round: {}",
+                                    node_id,
+                                    err
+                                );
+                            }
+                        }
+                    }
+                }
             }
         }
     }
@@ -1033,7 +1072,7 @@ impl Service {
         tenant_id=%result.tenant_shard_id.tenant_id, shard_id=%result.tenant_shard_id.shard_slug(),
         sequence=%result.sequence
     ))]
-    fn process_result(&self, mut result: ReconcileResult) {
+    fn process_result(&self, result: ReconcileResult) {
         let mut locked = self.inner.write().unwrap();
         let (nodes, tenants, _scheduler) = locked.parts_mut();
         let Some(tenant) = tenants.get_mut(&result.tenant_shard_id) else {
@@ -1055,22 +1094,27 @@ impl Service {
 
         // In case a node was deleted while this reconcile is in flight, filter it out of the update we will
         // make to the tenant
-        result
-            .observed
-            .locations
-            .retain(|node_id, _loc| nodes.contains_key(node_id));
+        let deltas = result.observed_deltas.into_iter().flat_map(|delta| {
+            // In case a node was deleted while this reconcile is in flight, filter it out of the update we will
+            // make to the tenant
+            let node = nodes.get(delta.node_id())?;
+
+            if node.is_available() {
+                return Some(delta);
+            }
+
+            // In case a node became unavailable concurrently with the reconcile, observed
+            // locations on it are now uncertain. By convention, set them to None in order
+            // for them to get refreshed when the node comes back online.
+            Some(ObservedStateDelta::Upsert(Box::new((
+                node.get_id(),
+                ObservedStateLocation { conf: None },
+            ))))
+        });
 
         match result.result {
             Ok(()) => {
-                for (node_id, loc) in &result.observed.locations {
-                    if let Some(conf) = &loc.conf {
-                        tracing::info!("Updating observed location {}: {:?}", node_id, conf);
-                    } else {
-                        tracing::info!("Setting observed location {} to None", node_id,)
-                    }
-                }
-
-                tenant.observed = result.observed;
+                tenant.apply_observed_deltas(deltas);
                 tenant.waiter.advance(result.sequence);
             }
             Err(e) => {
@@ -1092,9 +1136,10 @@ impl Service {
                 // so that waiters will see the correct error after waiting.
                 tenant.set_last_error(result.sequence, e);
 
-                for (node_id, o) in result.observed.locations {
-                    tenant.observed.locations.insert(node_id, o);
-                }
+                // Skip deletions on reconcile failures
+                let upsert_deltas =
+                    deltas.filter(|delta| matches!(delta, ObservedStateDelta::Upsert(_)));
+                tenant.apply_observed_deltas(upsert_deltas);
             }
         }
 
@@ -5299,15 +5344,17 @@ impl Service {
         Ok(())
     }
 
-    pub(crate) async fn node_configure(
+    /// Configure in-memory and persistent state of a node as requested
+    ///
+    /// Note that this function does not trigger any immediate side effects in response
+    /// to the changes. That part is handled by [`Self::handle_node_availability_transition`].
+    async fn node_state_configure(
         &self,
         node_id: NodeId,
         availability: Option<NodeAvailability>,
         scheduling: Option<NodeSchedulingPolicy>,
-    ) -> Result<(), ApiError> {
-        let _node_lock =
-            trace_exclusive_lock(&self.node_op_locks, node_id, NodeOperations::Configure).await;
-
+        node_lock: &TracingExclusiveGuard<NodeOperations>,
+    ) -> Result<AvailabilityTransition, ApiError> {
         if let Some(scheduling) = scheduling {
             // Scheduling is a persistent part of Node: we must write updates to the database before
             // applying them in memory
@@ -5336,7 +5383,7 @@ impl Service {
             };
 
             if matches!(availability_transition, AvailabilityTransition::ToActive) {
-                self.node_activate_reconcile(activate_node, &_node_lock)
+                self.node_activate_reconcile(activate_node, node_lock)
                     .await?;
             }
             availability_transition
@@ -5346,7 +5393,7 @@ impl Service {
 
         // Apply changes from the request to our in-memory state for the Node
         let mut locked = self.inner.write().unwrap();
-        let (nodes, tenants, scheduler) = locked.parts_mut();
+        let (nodes, _tenants, scheduler) = locked.parts_mut();
 
         let mut new_nodes = (**nodes).clone();
 
@@ -5356,8 +5403,8 @@ impl Service {
             ));
         };
 
-        if let Some(availability) = availability.as_ref() {
-            node.set_availability(availability.clone());
+        if let Some(availability) = availability {
+            node.set_availability(availability);
         }
 
         if let Some(scheduling) = scheduling {
@@ -5368,11 +5415,30 @@ impl Service {
         scheduler.node_upsert(node);
 
         let new_nodes = Arc::new(new_nodes);
+        locked.nodes = new_nodes;
 
+        Ok(availability_transition)
+    }
+
+    /// Handle availability transition of one node
+    ///
+    /// Note that you should first call [`Self::node_state_configure`] to update
+    /// the in-memory state referencing that node. If you need to handle more than one transition
+    /// consider using [`Self::handle_node_availability_transitions`].
+    async fn handle_node_availability_transition(
+        &self,
+        node_id: NodeId,
+        transition: AvailabilityTransition,
+        _node_lock: &TracingExclusiveGuard<NodeOperations>,
+    ) -> Result<(), ApiError> {
         // Modify scheduling state for any Tenants that are affected by a change in the node's availability state.
-        match availability_transition {
+        match transition {
             AvailabilityTransition::ToOffline => {
                 tracing::info!("Node {} transition to offline", node_id);
+
+                let mut locked = self.inner.write().unwrap();
+                let (nodes, tenants, scheduler) = locked.parts_mut();
+
                 let mut tenants_affected: usize = 0;
 
                 for (tenant_shard_id, tenant_shard) in tenants {
@@ -5382,14 +5448,14 @@ impl Service {
                         observed_loc.conf = None;
                     }
 
-                    if new_nodes.len() == 1 {
+                    if nodes.len() == 1 {
                         // Special case for single-node cluster: there is no point trying to reschedule
                         // any tenant shards: avoid doing so, in order to avoid spewing warnings about
                         // failures to schedule them.
                         continue;
                     }
 
-                    if !new_nodes
+                    if !nodes
                         .values()
                         .any(|n| matches!(n.may_schedule(), MaySchedule::Yes(_)))
                     {
@@ -5415,10 +5481,7 @@ impl Service {
                                 tracing::warn!(%tenant_shard_id, "Scheduling error when marking pageserver {} offline: {e}", node_id);
                             }
                             Ok(()) => {
-                                if self
-                                    .maybe_reconcile_shard(tenant_shard, &new_nodes)
-                                    .is_some()
-                                {
+                                if self.maybe_reconcile_shard(tenant_shard, nodes).is_some() {
                                     tenants_affected += 1;
                                 };
                             }
@@ -5433,9 +5496,13 @@ impl Service {
             }
             AvailabilityTransition::ToActive => {
                 tracing::info!("Node {} transition to active", node_id);
+
+                let mut locked = self.inner.write().unwrap();
+                let (nodes, tenants, _scheduler) = locked.parts_mut();
+
                 // When a node comes back online, we must reconcile any tenant that has a None observed
                 // location on the node.
-                for tenant_shard in locked.tenants.values_mut() {
+                for tenant_shard in tenants.values_mut() {
                     // If a reconciliation is already in progress, rely on the previous scheduling
                     // decision and skip triggering a new reconciliation.
                     if tenant_shard.reconciler.is_some() {
@@ -5444,7 +5511,7 @@ impl Service {
 
                     if let Some(observed_loc) = tenant_shard.observed.locations.get_mut(&node_id) {
                         if observed_loc.conf.is_none() {
-                            self.maybe_reconcile_shard(tenant_shard, &new_nodes);
+                            self.maybe_reconcile_shard(tenant_shard, nodes);
                         }
                     }
                 }
@@ -5465,11 +5532,54 @@ impl Service {
             }
         }
 
-        locked.nodes = new_nodes;
-
         Ok(())
     }
 
+    /// Handle availability transition for multiple nodes
+    ///
+    /// Note that you should first call [`Self::node_state_configure`] for
+    /// all nodes being handled here for the handling to use fresh in-memory state.
+    async fn handle_node_availability_transitions(
+        &self,
+        transitions: Vec<(
+            NodeId,
+            TracingExclusiveGuard<NodeOperations>,
+            AvailabilityTransition,
+        )>,
+    ) -> Result<(), Vec<(NodeId, ApiError)>> {
+        let mut errors = Vec::default();
+        for (node_id, node_lock, transition) in transitions {
+            let res = self
+                .handle_node_availability_transition(node_id, transition, &node_lock)
+                .await;
+            if let Err(err) = res {
+                errors.push((node_id, err));
+            }
+        }
+
+        if errors.is_empty() {
+            Ok(())
+        } else {
+            Err(errors)
+        }
+    }
+
+    pub(crate) async fn node_configure(
+        &self,
+        node_id: NodeId,
+        availability: Option<NodeAvailability>,
+        scheduling: Option<NodeSchedulingPolicy>,
+    ) -> Result<(), ApiError> {
+        let node_lock =
+            trace_exclusive_lock(&self.node_op_locks, node_id, NodeOperations::Configure).await;
+
+        let transition = self
+            .node_state_configure(node_id, availability, scheduling, &node_lock)
+            .await?;
+        self.handle_node_availability_transition(node_id, transition, &node_lock)
+            .await
+    }
+
     /// Wrapper around [`Self::node_configure`] which only allows changes while there is no ongoing
     /// operation for HTTP api.
     pub(crate) async fn external_node_configure(

storage_controller/src/tenant_shard.rs

@@ -425,6 +425,22 @@ pub(crate) enum ReconcileNeeded {
     Yes,
 }
 
+/// Pending modification to the observed state of a tenant shard.
+/// Produced by [`Reconciler::observed_deltas`] and applied in [`crate::service::Service::process_result`].
+pub(crate) enum ObservedStateDelta {
+    Upsert(Box<(NodeId, ObservedStateLocation)>),
+    Delete(NodeId),
+}
+
+impl ObservedStateDelta {
+    pub(crate) fn node_id(&self) -> &NodeId {
+        match self {
+            Self::Upsert(up) => &up.0,
+            Self::Delete(nid) => nid,
+        }
+    }
+}
+
 /// When a reconcile task completes, it sends this result object
 /// to be applied to the primary TenantShard.
 pub(crate) struct ReconcileResult {
@@ -437,7 +453,7 @@ pub(crate) struct ReconcileResult {
 
     pub(crate) tenant_shard_id: TenantShardId,
     pub(crate) generation: Option<Generation>,
-    pub(crate) observed: ObservedState,
+    pub(crate) observed_deltas: Vec<ObservedStateDelta>,
 
     /// Set [`TenantShard::pending_compute_notification`] from this flag
     pub(crate) pending_compute_notification: bool,
@@ -1123,7 +1139,7 @@ impl TenantShard {
             result,
             tenant_shard_id: reconciler.tenant_shard_id,
             generation: reconciler.generation,
-            observed: reconciler.observed,
+            observed_deltas: reconciler.observed_deltas(),
             pending_compute_notification: reconciler.compute_notify_failure,
         }
     }
@@ -1177,6 +1193,7 @@ impl TenantShard {
             reconciler_config,
             config: self.config.clone(),
             observed: self.observed.clone(),
+            original_observed: self.observed.clone(),
             compute_hook: compute_hook.clone(),
             service_config: service_config.clone(),
             _gate_guard: gate_guard,
@@ -1437,6 +1454,62 @@ impl TenantShard {
             .map(|(node_id, gen)| (node_id, Generation::new(gen)))
             .collect()
     }
+
+    /// Update the observed state of the tenant by applying incremental deltas
+    ///
+    /// Deltas are generated by reconcilers via [`Reconciler::observed_deltas`].
+    /// They are then filtered in [`crate::service::Service::process_result`].
+    pub(crate) fn apply_observed_deltas(
+        &mut self,
+        deltas: impl Iterator<Item = ObservedStateDelta>,
+    ) {
+        for delta in deltas {
+            match delta {
+                ObservedStateDelta::Upsert(ups) => {
+                    let (node_id, loc) = *ups;
+
+                    // If the generation of the observed location in the delta is lagging
+                    // behind the current one, then we have a race condition and cannot
+                    // be certain about the true observed state. Set the observed state
+                    // to None in order to reflect this.
+                    let crnt_gen = self
+                        .observed
+                        .locations
+                        .get(&node_id)
+                        .and_then(|loc| loc.conf.as_ref())
+                        .and_then(|conf| conf.generation);
+                    let new_gen = loc.conf.as_ref().and_then(|conf| conf.generation);
+                    match (crnt_gen, new_gen) {
+                        (Some(crnt), Some(new)) if crnt_gen > new_gen => {
+                            tracing::warn!(
+                                "Skipping observed state update {}: {:?} and using None due to stale generation ({} > {})",
+                                node_id, loc, crnt, new
+                            );
+
+                            self.observed
+                                .locations
+                                .insert(node_id, ObservedStateLocation { conf: None });
+
+                            continue;
+                        }
+                        _ => {}
+                    }
+
+                    if let Some(conf) = &loc.conf {
+                        tracing::info!("Updating observed location {}: {:?}", node_id, conf);
+                    } else {
+                        tracing::info!("Setting observed location {} to None", node_id,)
+                    }
+
+                    self.observed.locations.insert(node_id, loc);
+                }
+                ObservedStateDelta::Delete(node_id) => {
+                    tracing::info!("Deleting observed location {}", node_id);
+                    self.observed.locations.remove(&node_id);
+                }
+            }
+        }
+    }
 }
 
 #[cfg(test)]

test_runner/cloud_regress/test_cloud_regress.py

@@ -2,6 +2,8 @@
 Run the regression tests on the cloud instance of Neon
 """
 
+from __future__ import annotations
+
 from pathlib import Path
 from typing import Any
 

test_runner/conftest.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 pytest_plugins = (
     "fixtures.pg_version",
     "fixtures.parametrize",

test_runner/fixtures/__init__.py

@@ -0,0 +1 @@
+from __future__ import annotations

test_runner/fixtures/benchmark_fixture.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import calendar
 import dataclasses
 import enum
@@ -5,12 +7,11 @@ import json
 import os
 import re
 import timeit
+from collections.abc import Iterator
 from contextlib import contextmanager
 from datetime import datetime
 from pathlib import Path
-
-# Type-related stuff
-from typing import Callable, ClassVar, Dict, Iterator, Optional
+from typing import TYPE_CHECKING
 
 import allure
 import pytest
@@ -23,6 +24,10 @@ from fixtures.common_types import TenantId, TimelineId
 from fixtures.log_helper import log
 from fixtures.neon_fixtures import NeonPageserver
 
+if TYPE_CHECKING:
+    from typing import Callable, ClassVar, Optional
+
+
 """
 This file contains fixtures for micro-benchmarks.
 
@@ -138,18 +143,6 @@ class PgBenchRunResult:
 
 @dataclasses.dataclass
 class PgBenchInitResult:
-    # Taken from https://github.com/postgres/postgres/blob/REL_15_1/src/bin/pgbench/pgbench.c#L5144-L5171
-    EXTRACTORS: ClassVar[Dict[str, re.Pattern]] = {  # type: ignore[type-arg]
-        "drop_tables": re.compile(r"drop tables (\d+\.\d+) s"),
-        "create_tables": re.compile(r"create tables (\d+\.\d+) s"),
-        "client_side_generate": re.compile(r"client-side generate (\d+\.\d+) s"),
-        "server_side_generate": re.compile(r"server-side generate (\d+\.\d+) s"),
-        "vacuum": re.compile(r"vacuum (\d+\.\d+) s"),
-        "primary_keys": re.compile(r"primary keys (\d+\.\d+) s"),
-        "foreign_keys": re.compile(r"foreign keys (\d+\.\d+) s"),
-        "total": re.compile(r"done in (\d+\.\d+) s"),  # Total time printed by pgbench
-    }
-
     total: Optional[float]
     drop_tables: Optional[float]
     create_tables: Optional[float]
@@ -162,6 +155,20 @@ class PgBenchInitResult:
     start_timestamp: int
     end_timestamp: int
 
+    # Taken from https://github.com/postgres/postgres/blob/REL_15_1/src/bin/pgbench/pgbench.c#L5144-L5171
+    EXTRACTORS: ClassVar[dict[str, re.Pattern[str]]] = dataclasses.field(
+        default_factory=lambda: {
+            "drop_tables": re.compile(r"drop tables (\d+\.\d+) s"),
+            "create_tables": re.compile(r"create tables (\d+\.\d+) s"),
+            "client_side_generate": re.compile(r"client-side generate (\d+\.\d+) s"),
+            "server_side_generate": re.compile(r"server-side generate (\d+\.\d+) s"),
+            "vacuum": re.compile(r"vacuum (\d+\.\d+) s"),
+            "primary_keys": re.compile(r"primary keys (\d+\.\d+) s"),
+            "foreign_keys": re.compile(r"foreign keys (\d+\.\d+) s"),
+            "total": re.compile(r"done in (\d+\.\d+) s"),  # Total time printed by pgbench
+        }
+    )
+
     @classmethod
     def parse_from_stderr(
         cls,
@@ -175,7 +182,7 @@ class PgBenchInitResult:
 
         last_line = stderr.splitlines()[-1]
 
-        timings: Dict[str, Optional[float]] = {}
+        timings: dict[str, Optional[float]] = {}
         last_line_items = re.split(r"\(|\)|,", last_line)
         for item in last_line_items:
             for key, regex in cls.EXTRACTORS.items():
@@ -385,7 +392,7 @@ class NeonBenchmarker:
         self,
         pageserver: NeonPageserver,
         metric_name: str,
-        label_filters: Optional[Dict[str, str]] = None,
+        label_filters: Optional[dict[str, str]] = None,
     ) -> int:
         """Fetch the value of given int counter from pageserver metrics."""
         all_metrics = pageserver.http_client().get_metrics()

test_runner/fixtures/common_types.py

@@ -1,10 +1,16 @@
+from __future__ import annotations
+
 import random
 from dataclasses import dataclass
 from enum import Enum
 from functools import total_ordering
-from typing import Any, Dict, Type, TypeVar, Union
+from typing import TYPE_CHECKING, TypeVar
+
+if TYPE_CHECKING:
+    from typing import Any, Union
+
+    T = TypeVar("T", bound="Id")
 
-T = TypeVar("T", bound="Id")
 
 DEFAULT_WAL_SEG_SIZE = 16 * 1024 * 1024
 
@@ -56,7 +62,7 @@ class Lsn:
             return NotImplemented
         return self.lsn_int - other.lsn_int
 
-    def __add__(self, other: Union[int, "Lsn"]) -> "Lsn":
+    def __add__(self, other: Union[int, Lsn]) -> Lsn:
         if isinstance(other, int):
             return Lsn(self.lsn_int + other)
         elif isinstance(other, Lsn):
@@ -70,7 +76,7 @@ class Lsn:
     def as_int(self) -> int:
         return self.lsn_int
 
-    def segment_lsn(self, seg_sz: int = DEFAULT_WAL_SEG_SIZE) -> "Lsn":
+    def segment_lsn(self, seg_sz: int = DEFAULT_WAL_SEG_SIZE) -> Lsn:
         return Lsn(self.lsn_int - (self.lsn_int % seg_sz))
 
     def segno(self, seg_sz: int = DEFAULT_WAL_SEG_SIZE) -> int:
@@ -127,7 +133,7 @@ class Id:
         return hash(str(self.id))
 
     @classmethod
-    def generate(cls: Type[T]) -> T:
+    def generate(cls: type[T]) -> T:
         """Generate a random ID"""
         return cls(random.randbytes(16).hex())
 
@@ -162,7 +168,7 @@ class TenantTimelineId:
     timeline_id: TimelineId
 
     @classmethod
-    def from_json(cls, d: Dict[str, Any]) -> "TenantTimelineId":
+    def from_json(cls, d: dict[str, Any]) -> TenantTimelineId:
         return TenantTimelineId(
             tenant_id=TenantId(d["tenant_id"]),
             timeline_id=TimelineId(d["timeline_id"]),
@@ -181,7 +187,7 @@ class TenantShardId:
         assert self.shard_number < self.shard_count or self.shard_count == 0
 
     @classmethod
-    def parse(cls: Type[TTenantShardId], input) -> TTenantShardId:
+    def parse(cls: type[TTenantShardId], input) -> TTenantShardId:
         if len(input) == 32:
             return cls(
                 tenant_id=TenantId(input),

test_runner/fixtures/compare_fixtures.py

@@ -1,11 +1,13 @@
+from __future__ import annotations
+
 import os
 import time
 from abc import ABC, abstractmethod
+from collections.abc import Iterator
 from contextlib import _GeneratorContextManager, contextmanager
 
 # Type-related stuff
 from pathlib import Path
-from typing import Dict, Iterator, List
 
 import pytest
 from _pytest.fixtures import FixtureRequest
@@ -72,7 +74,7 @@ class PgCompare(ABC):
         pass
 
     @contextmanager
-    def record_pg_stats(self, pg_stats: List[PgStatTable]) -> Iterator[None]:
+    def record_pg_stats(self, pg_stats: list[PgStatTable]) -> Iterator[None]:
         init_data = self._retrieve_pg_stats(pg_stats)
 
         yield
@@ -82,8 +84,8 @@ class PgCompare(ABC):
         for k in set(init_data) & set(data):
             self.zenbenchmark.record(k, data[k] - init_data[k], "", MetricReport.HIGHER_IS_BETTER)
 
-    def _retrieve_pg_stats(self, pg_stats: List[PgStatTable]) -> Dict[str, int]:
-        results: Dict[str, int] = {}
+    def _retrieve_pg_stats(self, pg_stats: list[PgStatTable]) -> dict[str, int]:
+        results: dict[str, int] = {}
 
         with self.pg.connect().cursor() as cur:
             for pg_stat in pg_stats:

test_runner/fixtures/compute_reconfigure.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import concurrent.futures
 from typing import Any
 

test_runner/fixtures/endpoint/__init__.py

@@ -0,0 +1 @@
+from __future__ import annotations

test_runner/fixtures/endpoint/http.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import requests
 from requests.adapters import HTTPAdapter
 
@@ -21,3 +23,8 @@ class EndpointHttpClient(requests.Session):
         res = self.get(f"http://localhost:{self.port}/database_schema?database={database}")
         res.raise_for_status()
         return res.text
+
+    def installed_extensions(self):
+        res = self.get(f"http://localhost:{self.port}/installed_extensions")
+        res.raise_for_status()
+        return res.json()

test_runner/fixtures/flaky.py

@@ -1,6 +1,9 @@
+from __future__ import annotations
+
 import json
+from collections.abc import MutableMapping
 from pathlib import Path
-from typing import Any, List, MutableMapping, cast
+from typing import TYPE_CHECKING, cast
 
 import pytest
 from _pytest.config import Config
@@ -10,6 +13,9 @@ from allure_pytest.utils import allure_name, allure_suite_labels
 
 from fixtures.log_helper import log
 
+if TYPE_CHECKING:
+    from typing import Any
+
 """
 The plugin reruns flaky tests.
 It uses `pytest.mark.flaky` provided by `pytest-rerunfailures` plugin and flaky tests detected by `scripts/flaky_tests.py`
@@ -27,7 +33,7 @@ def pytest_addoption(parser: Parser):
     )
 
 
-def pytest_collection_modifyitems(config: Config, items: List[pytest.Item]):
+def pytest_collection_modifyitems(config: Config, items: list[pytest.Item]):
     if not config.getoption("--flaky-tests-json"):
         return
 
@@ -66,5 +72,5 @@ def pytest_collection_modifyitems(config: Config, items: List[pytest.Item]):
             # - [2] https://github.com/pytest-dev/pytest-timeout/issues/142
             timeout_marker = item.get_closest_marker("timeout")
             if timeout_marker is not None:
-                kwargs = cast(MutableMapping[str, Any], timeout_marker.kwargs)
+                kwargs = cast("MutableMapping[str, Any]", timeout_marker.kwargs)
                 kwargs["func_only"] = True

test_runner/fixtures/httpserver.py

@@ -1,4 +1,4 @@
-from typing import Tuple
+from __future__ import annotations
 
 import pytest
 from pytest_httpserver import HTTPServer
@@ -40,6 +40,6 @@ def httpserver(make_httpserver):
 
 
 @pytest.fixture(scope="function")
-def httpserver_listen_address(port_distributor) -> Tuple[str, int]:
+def httpserver_listen_address(port_distributor) -> tuple[str, int]:
     port = port_distributor.get_port()
     return ("localhost", port)

test_runner/fixtures/log_helper.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import logging
 import logging.config
 
@@ -22,9 +24,20 @@ https://docs.pytest.org/en/6.2.x/logging.html
 # log format is specified in pytest.ini file
 LOGGING = {
     "version": 1,
+    "filters": {
+        "wzfilter": {
+            "()": "fixtures.log_helper_internal.WerkzeugNoiseFilter",
+        },
+    },
     "loggers": {
         "root": {"level": "INFO"},
         "root.safekeeper_async": {"level": "INFO"},  # a lot of logs on DEBUG level
+
+        # Use a custom filter to make werkzeug's messages less verbose.
+        "werkzeug": {
+            "filters": ["wzfilter"],
+            "level": "INFO",
+        },
     },
 }
 

test_runner/fixtures/log_helper_internal.py

@@ -0,0 +1,24 @@
+# These are logically part of in log_helper.py, but need to be in a
+# different file because these get loaded from the logging config
+# file. If you try to included these in log_helper.py directly, you
+# get an error about circular dependency.
+import re
+
+class WerkzeugNoiseFilter(object):
+    """Moto server that we use for mocking S3 uses werkzeug, which
+    logs all HTTP operations.  It constructs log messages like this:
+
+        127.0.0.1 - - [08/Oct/2024 12:43:46] "PUT /bucket-name/path?x-id=PutObject HTTP/1.1" 200 -
+
+    The IP address is not interesting in tests, as it's always just
+    127.0.0.1. And the timestamp is redundant with the timestamp we
+    print for all log messages anyway, with millisecond precision.
+    Unfortunately those are "etched" in the message, and cannot be
+    overriden by setting a custom formatter. To reduce the noise in
+    the test output, this filter removes those fields from the log
+    messages.
+    """
+
+    def filter(self, logRecord):
+        logRecord.msg = re.sub(r'127\.0\.0\.1 - - \[.+\] (".*".*)', r'\1', logRecord.msg)
+        return True

test_runner/fixtures/metrics.py

@@ -1,21 +1,26 @@
+from __future__ import annotations
+
 from collections import defaultdict
-from typing import Dict, List, Optional, Tuple
+from typing import TYPE_CHECKING
 
 from prometheus_client.parser import text_string_to_metric_families
 from prometheus_client.samples import Sample
 
 from fixtures.log_helper import log
 
+if TYPE_CHECKING:
+    from typing import Optional
+
 
 class Metrics:
-    metrics: Dict[str, List[Sample]]
+    metrics: dict[str, list[Sample]]
     name: str
 
     def __init__(self, name: str = ""):
         self.metrics = defaultdict(list)
         self.name = name
 
-    def query_all(self, name: str, filter: Optional[Dict[str, str]] = None) -> List[Sample]:
+    def query_all(self, name: str, filter: Optional[dict[str, str]] = None) -> list[Sample]:
         filter = filter or {}
         res = []
 
@@ -27,7 +32,7 @@ class Metrics:
                 pass
         return res
 
-    def query_one(self, name: str, filter: Optional[Dict[str, str]] = None) -> Sample:
+    def query_one(self, name: str, filter: Optional[dict[str, str]] = None) -> Sample:
         res = self.query_all(name, filter or {})
         assert len(res) == 1, f"expected single sample for {name} {filter}, found {res}"
         return res[0]
@@ -43,7 +48,7 @@ class MetricsGetter:
         raise NotImplementedError()
 
     def get_metric_value(
-        self, name: str, filter: Optional[Dict[str, str]] = None
+        self, name: str, filter: Optional[dict[str, str]] = None
     ) -> Optional[float]:
         metrics = self.get_metrics()
         results = metrics.query_all(name, filter=filter)
@@ -54,8 +59,8 @@ class MetricsGetter:
         return results[0].value
 
     def get_metrics_values(
-        self, names: list[str], filter: Optional[Dict[str, str]] = None, absence_ok=False
-    ) -> Dict[str, float]:
+        self, names: list[str], filter: Optional[dict[str, str]] = None, absence_ok=False
+    ) -> dict[str, float]:
         """
         When fetching multiple named metrics, it is more efficient to use this
         than to call `get_metric_value` repeatedly.
@@ -97,7 +102,7 @@ def parse_metrics(text: str, name: str = "") -> Metrics:
     return metrics
 
 
-def histogram(prefix_without_trailing_underscore: str) -> List[str]:
+def histogram(prefix_without_trailing_underscore: str) -> list[str]:
     assert not prefix_without_trailing_underscore.endswith("_")
     return [f"{prefix_without_trailing_underscore}_{x}" for x in ["bucket", "count", "sum"]]
 
@@ -107,7 +112,7 @@ def counter(name: str) -> str:
     return f"{name}_total"
 
 
-PAGESERVER_PER_TENANT_REMOTE_TIMELINE_CLIENT_METRICS: Tuple[str, ...] = (
+PAGESERVER_PER_TENANT_REMOTE_TIMELINE_CLIENT_METRICS: tuple[str, ...] = (
     "pageserver_remote_timeline_client_calls_started_total",
     "pageserver_remote_timeline_client_calls_finished_total",
     "pageserver_remote_physical_size",
@@ -115,7 +120,7 @@ PAGESERVER_PER_TENANT_REMOTE_TIMELINE_CLIENT_METRICS: Tuple[str, ...] = (
     "pageserver_remote_timeline_client_bytes_finished_total",
 )
 
-PAGESERVER_GLOBAL_METRICS: Tuple[str, ...] = (
+PAGESERVER_GLOBAL_METRICS: tuple[str, ...] = (
     "pageserver_storage_operations_seconds_global_count",
     "pageserver_storage_operations_seconds_global_sum",
     "pageserver_storage_operations_seconds_global_bucket",
@@ -147,7 +152,7 @@ PAGESERVER_GLOBAL_METRICS: Tuple[str, ...] = (
     counter("pageserver_tenant_throttling_count_global"),
 )
 
-PAGESERVER_PER_TENANT_METRICS: Tuple[str, ...] = (
+PAGESERVER_PER_TENANT_METRICS: tuple[str, ...] = (
     "pageserver_current_logical_size",
     "pageserver_resident_physical_size",
     "pageserver_io_operations_bytes_total",

test_runner/fixtures/neon_api.py

@@ -6,12 +6,12 @@ from typing import TYPE_CHECKING, cast
 import requests
 
 if TYPE_CHECKING:
-    from typing import Any, Dict, Literal, Optional, Union
+    from typing import Any, Literal, Optional, Union
 
     from fixtures.pg_version import PgVersion
 
 
-def connection_parameters_to_env(params: Dict[str, str]) -> Dict[str, str]:
+def connection_parameters_to_env(params: dict[str, str]) -> dict[str, str]:
     return {
         "PGHOST": params["host"],
         "PGDATABASE": params["database"],
@@ -41,8 +41,8 @@ class NeonAPI:
         branch_name: Optional[str] = None,
         branch_role_name: Optional[str] = None,
         branch_database_name: Optional[str] = None,
-    ) -> Dict[str, Any]:
-        data: Dict[str, Any] = {
+    ) -> dict[str, Any]:
+        data: dict[str, Any] = {
             "project": {
                 "branch": {},
             },
@@ -70,9 +70,9 @@ class NeonAPI:
 
         assert resp.status_code == 201
 
-        return cast("Dict[str, Any]", resp.json())
+        return cast("dict[str, Any]", resp.json())
 
-    def get_project_details(self, project_id: str) -> Dict[str, Any]:
+    def get_project_details(self, project_id: str) -> dict[str, Any]:
         resp = self.__request(
             "GET",
             f"/projects/{project_id}",
@@ -82,12 +82,12 @@ class NeonAPI:
             },
         )
         assert resp.status_code == 200
-        return cast("Dict[str, Any]", resp.json())
+        return cast("dict[str, Any]", resp.json())
 
     def delete_project(
         self,
         project_id: str,
-    ) -> Dict[str, Any]:
+    ) -> dict[str, Any]:
         resp = self.__request(
             "DELETE",
             f"/projects/{project_id}",
@@ -99,13 +99,13 @@ class NeonAPI:
 
         assert resp.status_code == 200
 
-        return cast("Dict[str, Any]", resp.json())
+        return cast("dict[str, Any]", resp.json())
 
     def start_endpoint(
         self,
         project_id: str,
         endpoint_id: str,
-    ) -> Dict[str, Any]:
+    ) -> dict[str, Any]:
         resp = self.__request(
             "POST",
             f"/projects/{project_id}/endpoints/{endpoint_id}/start",
@@ -116,13 +116,13 @@ class NeonAPI:
 
         assert resp.status_code == 200
 
-        return cast("Dict[str, Any]", resp.json())
+        return cast("dict[str, Any]", resp.json())
 
     def suspend_endpoint(
         self,
         project_id: str,
         endpoint_id: str,
-    ) -> Dict[str, Any]:
+    ) -> dict[str, Any]:
         resp = self.__request(
             "POST",
             f"/projects/{project_id}/endpoints/{endpoint_id}/suspend",
@@ -133,13 +133,13 @@ class NeonAPI:
 
         assert resp.status_code == 200
 
-        return cast("Dict[str, Any]", resp.json())
+        return cast("dict[str, Any]", resp.json())
 
     def restart_endpoint(
         self,
         project_id: str,
         endpoint_id: str,
-    ) -> Dict[str, Any]:
+    ) -> dict[str, Any]:
         resp = self.__request(
             "POST",
             f"/projects/{project_id}/endpoints/{endpoint_id}/restart",
@@ -150,16 +150,16 @@ class NeonAPI:
 
         assert resp.status_code == 200
 
-        return cast("Dict[str, Any]", resp.json())
+        return cast("dict[str, Any]", resp.json())
 
     def create_endpoint(
         self,
         project_id: str,
         branch_id: str,
         endpoint_type: Literal["read_write", "read_only"],
-        settings: Dict[str, Any],
-    ) -> Dict[str, Any]:
-        data: Dict[str, Any] = {
+        settings: dict[str, Any],
+    ) -> dict[str, Any]:
+        data: dict[str, Any] = {
             "endpoint": {
                 "branch_id": branch_id,
             },
@@ -182,7 +182,7 @@ class NeonAPI:
 
         assert resp.status_code == 201
 
-        return cast("Dict[str, Any]", resp.json())
+        return cast("dict[str, Any]", resp.json())
 
     def get_connection_uri(
         self,
@@ -192,7 +192,7 @@ class NeonAPI:
         database_name: str = "neondb",
         role_name: str = "neondb_owner",
         pooled: bool = True,
-    ) -> Dict[str, Any]:
+    ) -> dict[str, Any]:
         resp = self.__request(
             "GET",
             f"/projects/{project_id}/connection_uri",
@@ -210,9 +210,9 @@ class NeonAPI:
 
         assert resp.status_code == 200
 
-        return cast("Dict[str, Any]", resp.json())
+        return cast("dict[str, Any]", resp.json())
 
-    def get_branches(self, project_id: str) -> Dict[str, Any]:
+    def get_branches(self, project_id: str) -> dict[str, Any]:
         resp = self.__request(
             "GET",
             f"/projects/{project_id}/branches",
@@ -223,9 +223,9 @@ class NeonAPI:
 
         assert resp.status_code == 200
 
-        return cast("Dict[str, Any]", resp.json())
+        return cast("dict[str, Any]", resp.json())
 
-    def get_endpoints(self, project_id: str) -> Dict[str, Any]:
+    def get_endpoints(self, project_id: str) -> dict[str, Any]:
         resp = self.__request(
             "GET",
             f"/projects/{project_id}/endpoints",
@@ -236,9 +236,9 @@ class NeonAPI:
 
         assert resp.status_code == 200
 
-        return cast("Dict[str, Any]", resp.json())
+        return cast("dict[str, Any]", resp.json())
 
-    def get_operations(self, project_id: str) -> Dict[str, Any]:
+    def get_operations(self, project_id: str) -> dict[str, Any]:
         resp = self.__request(
             "GET",
             f"/projects/{project_id}/operations",
@@ -250,7 +250,7 @@ class NeonAPI:
 
         assert resp.status_code == 200
 
-        return cast("Dict[str, Any]", resp.json())
+        return cast("dict[str, Any]", resp.json())
 
     def wait_for_operation_to_finish(self, project_id: str):
         has_running = True

test_runner/fixtures/neon_cli.py

@@ -9,15 +9,7 @@ import tempfile
 import textwrap
 from itertools import chain, product
 from pathlib import Path
-from typing import (
-    Any,
-    Dict,
-    List,
-    Optional,
-    Tuple,
-    TypeVar,
-    cast,
-)
+from typing import TYPE_CHECKING, cast
 
 import toml
 
@@ -27,7 +19,15 @@ from fixtures.pageserver.common_types import IndexPartDump
 from fixtures.pg_version import PgVersion
 from fixtures.utils import AuxFileStore
 
-T = TypeVar("T")
+if TYPE_CHECKING:
+    from typing import (
+        Any,
+        Optional,
+        TypeVar,
+        cast,
+    )
+
+    T = TypeVar("T")
 
 
 class AbstractNeonCli(abc.ABC):
@@ -37,7 +37,7 @@ class AbstractNeonCli(abc.ABC):
     Do not use directly, use specific subclasses instead.
     """
 
-    def __init__(self, extra_env: Optional[Dict[str, str]], binpath: Path):
+    def __init__(self, extra_env: Optional[dict[str, str]], binpath: Path):
         self.extra_env = extra_env
         self.binpath = binpath
 
@@ -45,11 +45,11 @@ class AbstractNeonCli(abc.ABC):
 
     def raw_cli(
         self,
-        arguments: List[str],
-        extra_env_vars: Optional[Dict[str, str]] = None,
+        arguments: list[str],
+        extra_env_vars: Optional[dict[str, str]] = None,
         check_return_code=True,
         timeout=None,
-    ) -> "subprocess.CompletedProcess[str]":
+    ) -> subprocess.CompletedProcess[str]:
         """
         Run the command with the specified arguments.
 
@@ -92,9 +92,8 @@ class AbstractNeonCli(abc.ABC):
                 args,
                 env=env_vars,
                 check=False,
-                universal_newlines=True,
-                stdout=subprocess.PIPE,
-                stderr=subprocess.PIPE,
+                text=True,
+                capture_output=True,
                 timeout=timeout,
             )
         except subprocess.TimeoutExpired as e:
@@ -118,7 +117,7 @@ class AbstractNeonCli(abc.ABC):
             if len(lines) < 2:
                 log.debug(f"Run {res.args} success: {stripped}")
             else:
-                log.debug("Run %s success:\n%s" % (res.args, textwrap.indent(stripped, indent)))
+                log.debug("Run %s success:\n%s", res.args, textwrap.indent(stripped, indent))
         elif check_return_code:
             # this way command output will be in recorded and shown in CI in failure message
             indent = indent * 2
@@ -175,7 +174,7 @@ class NeonLocalCli(AbstractNeonCli):
 
     def __init__(
         self,
-        extra_env: Optional[Dict[str, str]],
+        extra_env: Optional[dict[str, str]],
         binpath: Path,
         repo_dir: Path,
         pg_distrib_dir: Path,
@@ -197,7 +196,7 @@ class NeonLocalCli(AbstractNeonCli):
         tenant_id: TenantId,
         timeline_id: TimelineId,
         pg_version: PgVersion,
-        conf: Optional[Dict[str, Any]] = None,
+        conf: Optional[dict[str, Any]] = None,
         shard_count: Optional[int] = None,
         shard_stripe_size: Optional[int] = None,
         placement_policy: Optional[str] = None,
@@ -258,7 +257,7 @@ class NeonLocalCli(AbstractNeonCli):
         res = self.raw_cli(["tenant", "set-default", "--tenant-id", str(tenant_id)])
         res.check_returncode()
 
-    def tenant_config(self, tenant_id: TenantId, conf: Dict[str, str]):
+    def tenant_config(self, tenant_id: TenantId, conf: dict[str, str]):
         """
         Update tenant config.
         """
@@ -274,7 +273,7 @@ class NeonLocalCli(AbstractNeonCli):
         res = self.raw_cli(args)
         res.check_returncode()
 
-    def tenant_list(self) -> "subprocess.CompletedProcess[str]":
+    def tenant_list(self) -> subprocess.CompletedProcess[str]:
         res = self.raw_cli(["tenant", "list"])
         res.check_returncode()
         return res
@@ -368,7 +367,7 @@ class NeonLocalCli(AbstractNeonCli):
         res = self.raw_cli(cmd)
         res.check_returncode()
 
-    def timeline_list(self, tenant_id: TenantId) -> List[Tuple[str, TimelineId]]:
+    def timeline_list(self, tenant_id: TenantId) -> list[tuple[str, TimelineId]]:
         """
         Returns a list of (branch_name, timeline_id) tuples out of parsed `neon timeline list` CLI output.
         """
@@ -389,9 +388,9 @@ class NeonLocalCli(AbstractNeonCli):
 
     def init(
         self,
-        init_config: Dict[str, Any],
+        init_config: dict[str, Any],
         force: Optional[str] = None,
-    ) -> "subprocess.CompletedProcess[str]":
+    ) -> subprocess.CompletedProcess[str]:
         with tempfile.NamedTemporaryFile(mode="w+") as init_config_tmpfile:
             init_config_tmpfile.write(toml.dumps(init_config))
             init_config_tmpfile.flush()
@@ -434,29 +433,28 @@ class NeonLocalCli(AbstractNeonCli):
     def pageserver_start(
         self,
         id: int,
-        extra_env_vars: Optional[Dict[str, str]] = None,
+        extra_env_vars: Optional[dict[str, str]] = None,
         timeout_in_seconds: Optional[int] = None,
-    ) -> "subprocess.CompletedProcess[str]":
+    ) -> subprocess.CompletedProcess[str]:
         start_args = ["pageserver", "start", f"--id={id}"]
         if timeout_in_seconds is not None:
             start_args.append(f"--start-timeout={timeout_in_seconds}s")
         return self.raw_cli(start_args, extra_env_vars=extra_env_vars)
 
-    def pageserver_stop(self, id: int, immediate=False) -> "subprocess.CompletedProcess[str]":
+    def pageserver_stop(self, id: int, immediate=False) -> subprocess.CompletedProcess[str]:
         cmd = ["pageserver", "stop", f"--id={id}"]
         if immediate:
             cmd.extend(["-m", "immediate"])
 
-        log.info(f"Stopping pageserver with {cmd}")
         return self.raw_cli(cmd)
 
     def safekeeper_start(
         self,
         id: int,
-        extra_opts: Optional[List[str]] = None,
-        extra_env_vars: Optional[Dict[str, str]] = None,
+        extra_opts: Optional[list[str]] = None,
+        extra_env_vars: Optional[dict[str, str]] = None,
         timeout_in_seconds: Optional[int] = None,
-    ) -> "subprocess.CompletedProcess[str]":
+    ) -> subprocess.CompletedProcess[str]:
         if extra_opts is not None:
             extra_opts = [f"-e={opt}" for opt in extra_opts]
         else:
@@ -469,7 +467,7 @@ class NeonLocalCli(AbstractNeonCli):
 
     def safekeeper_stop(
         self, id: Optional[int] = None, immediate=False
-    ) -> "subprocess.CompletedProcess[str]":
+    ) -> subprocess.CompletedProcess[str]:
         args = ["safekeeper", "stop"]
         if id is not None:
             args.append(str(id))
@@ -479,13 +477,13 @@ class NeonLocalCli(AbstractNeonCli):
 
     def storage_broker_start(
         self, timeout_in_seconds: Optional[int] = None
-    ) -> "subprocess.CompletedProcess[str]":
+    ) -> subprocess.CompletedProcess[str]:
         cmd = ["storage_broker", "start"]
         if timeout_in_seconds is not None:
             cmd.append(f"--start-timeout={timeout_in_seconds}s")
         return self.raw_cli(cmd)
 
-    def storage_broker_stop(self) -> "subprocess.CompletedProcess[str]":
+    def storage_broker_stop(self) -> subprocess.CompletedProcess[str]:
         cmd = ["storage_broker", "stop"]
         return self.raw_cli(cmd)
 
@@ -501,7 +499,7 @@ class NeonLocalCli(AbstractNeonCli):
         lsn: Optional[Lsn] = None,
         pageserver_id: Optional[int] = None,
         allow_multiple=False,
-    ) -> "subprocess.CompletedProcess[str]":
+    ) -> subprocess.CompletedProcess[str]:
         args = [
             "endpoint",
             "create",
@@ -534,12 +532,12 @@ class NeonLocalCli(AbstractNeonCli):
     def endpoint_start(
         self,
         endpoint_id: str,
-        safekeepers: Optional[List[int]] = None,
+        safekeepers: Optional[list[int]] = None,
         remote_ext_config: Optional[str] = None,
         pageserver_id: Optional[int] = None,
         allow_multiple=False,
         basebackup_request_tries: Optional[int] = None,
-    ) -> "subprocess.CompletedProcess[str]":
+    ) -> subprocess.CompletedProcess[str]:
         args = [
             "endpoint",
             "start",
@@ -568,9 +566,9 @@ class NeonLocalCli(AbstractNeonCli):
         endpoint_id: str,
         tenant_id: Optional[TenantId] = None,
         pageserver_id: Optional[int] = None,
-        safekeepers: Optional[List[int]] = None,
+        safekeepers: Optional[list[int]] = None,
         check_return_code=True,
-    ) -> "subprocess.CompletedProcess[str]":
+    ) -> subprocess.CompletedProcess[str]:
         args = ["endpoint", "reconfigure", endpoint_id]
         if tenant_id is not None:
             args.extend(["--tenant-id", str(tenant_id)])
@@ -586,7 +584,7 @@ class NeonLocalCli(AbstractNeonCli):
         destroy=False,
         check_return_code=True,
         mode: Optional[str] = None,
-    ) -> "subprocess.CompletedProcess[str]":
+    ) -> subprocess.CompletedProcess[str]:
         args = [
             "endpoint",
             "stop",
@@ -602,7 +600,7 @@ class NeonLocalCli(AbstractNeonCli):
 
     def mappings_map_branch(
         self, name: str, tenant_id: TenantId, timeline_id: TimelineId
-    ) -> "subprocess.CompletedProcess[str]":
+    ) -> subprocess.CompletedProcess[str]:
         """
         Map tenant id and timeline id to a neon_local branch name. They do not have to exist.
         Usually needed when creating branches via PageserverHttpClient and not neon_local.
@@ -623,10 +621,10 @@ class NeonLocalCli(AbstractNeonCli):
 
         return self.raw_cli(args, check_return_code=True)
 
-    def start(self, check_return_code=True) -> "subprocess.CompletedProcess[str]":
+    def start(self, check_return_code=True) -> subprocess.CompletedProcess[str]:
         return self.raw_cli(["start"], check_return_code=check_return_code)
 
-    def stop(self, check_return_code=True) -> "subprocess.CompletedProcess[str]":
+    def stop(self, check_return_code=True) -> subprocess.CompletedProcess[str]:
         return self.raw_cli(["stop"], check_return_code=check_return_code)
 
 
@@ -638,7 +636,7 @@ class WalCraft(AbstractNeonCli):
 
     COMMAND = "wal_craft"
 
-    def postgres_config(self) -> List[str]:
+    def postgres_config(self) -> list[str]:
         res = self.raw_cli(["print-postgres-config"])
         res.check_returncode()
         return res.stdout.split("\n")

test_runner/fixtures/neon_fixtures.py

@@ -13,6 +13,7 @@ import threading
 import time
 import uuid
 from collections import defaultdict
+from collections.abc import Iterable, Iterator
 from contextlib import closing, contextmanager
 from dataclasses import dataclass
 from datetime import datetime
@@ -21,20 +22,7 @@ from fcntl import LOCK_EX, LOCK_UN, flock
 from functools import cached_property
 from pathlib import Path
 from types import TracebackType
-from typing import (
-    Any,
-    Callable,
-    Dict,
-    Iterable,
-    Iterator,
-    List,
-    Optional,
-    Tuple,
-    Type,
-    TypeVar,
-    Union,
-    cast,
-)
+from typing import TYPE_CHECKING, cast
 from urllib.parse import quote, urlparse
 
 import asyncpg
@@ -91,7 +79,6 @@ from fixtures.utils import (
     allure_attach_from_dir,
     assert_no_errors,
     get_dir_size,
-    get_self_dir,
     print_gc_result,
     subprocess_capture,
     wait_until,
@@ -100,7 +87,17 @@ from fixtures.utils import AuxFileStore as AuxFileStore  # reexport
 
 from .neon_api import NeonAPI, NeonApiEndpoint
 
-T = TypeVar("T")
+if TYPE_CHECKING:
+    from typing import (
+        Any,
+        Callable,
+        Optional,
+        TypeVar,
+        Union,
+    )
+
+    T = TypeVar("T")
+
 
 """
 This file contains pytest fixtures. A fixture is a test resource that can be
@@ -119,7 +116,7 @@ Don't import functions from this file, or pytest will emit warnings. Instead
 put directly-importable functions into utils.py or another separate file.
 """
 
-Env = Dict[str, str]
+Env = dict[str, str]
 
 DEFAULT_OUTPUT_DIR: str = "test_output"
 DEFAULT_BRANCH_NAME: str = "main"
@@ -130,7 +127,7 @@ BASE_PORT: int = 15000
 @pytest.fixture(scope="session")
 def base_dir() -> Iterator[Path]:
     # find the base directory (currently this is the git root)
-    base_dir = get_self_dir().parent.parent
+    base_dir = Path(__file__).parents[2]
     log.info(f"base_dir is {base_dir}")
 
     yield base_dir
@@ -251,7 +248,7 @@ class PgProtocol:
         """
         return str(make_dsn(**self.conn_options(**kwargs)))
 
-    def conn_options(self, **kwargs: Any) -> Dict[str, Any]:
+    def conn_options(self, **kwargs: Any) -> dict[str, Any]:
         """
         Construct a dictionary of connection options from default values and extra parameters.
         An option can be dropped from the returning dictionary by None-valued extra parameter.
@@ -320,7 +317,7 @@ class PgProtocol:
                     conn_options["server_settings"] = {key: val}
         return await asyncpg.connect(**conn_options)
 
-    def safe_psql(self, query: str, **kwargs: Any) -> List[Tuple[Any, ...]]:
+    def safe_psql(self, query: str, **kwargs: Any) -> list[tuple[Any, ...]]:
         """
         Execute query against the node and return all rows.
         This method passes all extra params to connstr.
@@ -329,12 +326,12 @@ class PgProtocol:
 
     def safe_psql_many(
         self, queries: Iterable[str], log_query=True, **kwargs: Any
-    ) -> List[List[Tuple[Any, ...]]]:
+    ) -> list[list[tuple[Any, ...]]]:
         """
         Execute queries against the node and return all rows.
         This method passes all extra params to connstr.
         """
-        result: List[List[Any]] = []
+        result: list[list[Any]] = []
         with closing(self.connect(**kwargs)) as conn:
             with conn.cursor() as cur:
                 for query in queries:
@@ -380,7 +377,7 @@ class NeonEnvBuilder:
         test_overlay_dir: Optional[Path] = None,
         pageserver_remote_storage: Optional[RemoteStorage] = None,
         # toml that will be decomposed into `--config-override` flags during `pageserver --init`
-        pageserver_config_override: Optional[str | Callable[[Dict[str, Any]], None]] = None,
+        pageserver_config_override: Optional[str | Callable[[dict[str, Any]], None]] = None,
         num_safekeepers: int = 1,
         num_pageservers: int = 1,
         # Use non-standard SK ids to check for various parsing bugs
@@ -395,7 +392,7 @@ class NeonEnvBuilder:
         initial_timeline: Optional[TimelineId] = None,
         pageserver_virtual_file_io_engine: Optional[str] = None,
         pageserver_aux_file_policy: Optional[AuxFileStore] = None,
-        pageserver_default_tenant_config_compaction_algorithm: Optional[Dict[str, Any]] = None,
+        pageserver_default_tenant_config_compaction_algorithm: Optional[dict[str, Any]] = None,
         safekeeper_extra_opts: Optional[list[str]] = None,
         storage_controller_port_override: Optional[int] = None,
         pageserver_virtual_file_io_mode: Optional[str] = None,
@@ -430,7 +427,7 @@ class NeonEnvBuilder:
         self.enable_scrub_on_exit = True
         self.test_output_dir = test_output_dir
         self.test_overlay_dir = test_overlay_dir
-        self.overlay_mounts_created_by_us: List[Tuple[str, Path]] = []
+        self.overlay_mounts_created_by_us: list[tuple[str, Path]] = []
         self.config_init_force: Optional[str] = None
         self.top_output_dir = top_output_dir
         self.control_plane_compute_hook_api: Optional[str] = None
@@ -439,7 +436,7 @@ class NeonEnvBuilder:
         self.pageserver_virtual_file_io_engine: Optional[str] = pageserver_virtual_file_io_engine
 
         self.pageserver_default_tenant_config_compaction_algorithm: Optional[
-            Dict[str, Any]
+            dict[str, Any]
         ] = pageserver_default_tenant_config_compaction_algorithm
         if self.pageserver_default_tenant_config_compaction_algorithm is not None:
             log.debug(
@@ -469,7 +466,7 @@ class NeonEnvBuilder:
 
     def init_start(
         self,
-        initial_tenant_conf: Optional[Dict[str, Any]] = None,
+        initial_tenant_conf: Optional[dict[str, Any]] = None,
         default_remote_storage_if_missing: bool = True,
         initial_tenant_shard_count: Optional[int] = None,
         initial_tenant_shard_stripe_size: Optional[int] = None,
@@ -824,7 +821,7 @@ class NeonEnvBuilder:
 
         overlayfs_mounts = {mountpoint for _, mountpoint in self.overlay_mounts_created_by_us}
 
-        directories_to_clean: List[Path] = []
+        directories_to_clean: list[Path] = []
         for test_entry in Path(self.repo_dir).glob("**/*"):
             if test_entry in overlayfs_mounts:
                 continue
@@ -855,12 +852,12 @@ class NeonEnvBuilder:
             if isinstance(x, S3Storage):
                 x.do_cleanup()
 
-    def __enter__(self) -> "NeonEnvBuilder":
+    def __enter__(self) -> NeonEnvBuilder:
         return self
 
     def __exit__(
         self,
-        exc_type: Optional[Type[BaseException]],
+        exc_type: Optional[type[BaseException]],
         exc_value: Optional[BaseException],
         traceback: Optional[TracebackType],
     ):
@@ -971,8 +968,8 @@ class NeonEnv:
         self.port_distributor = config.port_distributor
         self.s3_mock_server = config.mock_s3_server
         self.endpoints = EndpointFactory(self)
-        self.safekeepers: List[Safekeeper] = []
-        self.pageservers: List[NeonPageserver] = []
+        self.safekeepers: list[Safekeeper] = []
+        self.pageservers: list[NeonPageserver] = []
         self.broker = NeonBroker(self)
         self.pageserver_remote_storage = config.pageserver_remote_storage
         self.safekeepers_remote_storage = config.safekeepers_remote_storage
@@ -1044,7 +1041,7 @@ class NeonEnv:
         self.pageserver_virtual_file_io_mode = config.pageserver_virtual_file_io_mode
 
         # Create the neon_local's `NeonLocalInitConf`
-        cfg: Dict[str, Any] = {
+        cfg: dict[str, Any] = {
             "default_tenant_id": str(self.initial_tenant),
             "broker": {
                 "listen_addr": self.broker.listen_addr(),
@@ -1073,7 +1070,7 @@ class NeonEnv:
                 http=self.port_distributor.get_port(),
             )
 
-            ps_cfg: Dict[str, Any] = {
+            ps_cfg: dict[str, Any] = {
                 "id": ps_id,
                 "listen_pg_addr": f"localhost:{pageserver_port.pg}",
                 "listen_http_addr": f"localhost:{pageserver_port.http}",
@@ -1122,7 +1119,7 @@ class NeonEnv:
                 http=self.port_distributor.get_port(),
             )
             id = config.safekeepers_id_start + i  # assign ids sequentially
-            sk_cfg: Dict[str, Any] = {
+            sk_cfg: dict[str, Any] = {
                 "id": id,
                 "pg_port": port.pg,
                 "pg_tenant_only_port": port.pg_tenant_only,
@@ -1287,9 +1284,8 @@ class NeonEnv:
         res = subprocess.run(
             [bin_pageserver, "--version"],
             check=True,
-            universal_newlines=True,
-            stdout=subprocess.PIPE,
-            stderr=subprocess.PIPE,
+            text=True,
+            capture_output=True,
         )
         return res.stdout
 
@@ -1332,13 +1328,13 @@ class NeonEnv:
         self,
         tenant_id: Optional[TenantId] = None,
         timeline_id: Optional[TimelineId] = None,
-        conf: Optional[Dict[str, Any]] = None,
+        conf: Optional[dict[str, Any]] = None,
         shard_count: Optional[int] = None,
         shard_stripe_size: Optional[int] = None,
         placement_policy: Optional[str] = None,
         set_default: bool = False,
         aux_file_policy: Optional[AuxFileStore] = None,
-    ) -> Tuple[TenantId, TimelineId]:
+    ) -> tuple[TenantId, TimelineId]:
         """
         Creates a new tenant, returns its id and its initial timeline's id.
         """
@@ -1359,7 +1355,7 @@ class NeonEnv:
 
         return tenant_id, timeline_id
 
-    def config_tenant(self, tenant_id: Optional[TenantId], conf: Dict[str, str]):
+    def config_tenant(self, tenant_id: Optional[TenantId], conf: dict[str, str]):
         """
         Update tenant config.
         """
@@ -1411,7 +1407,7 @@ def neon_simple_env(
     pg_version: PgVersion,
     pageserver_virtual_file_io_engine: str,
     pageserver_aux_file_policy: Optional[AuxFileStore],
-    pageserver_default_tenant_config_compaction_algorithm: Optional[Dict[str, Any]],
+    pageserver_default_tenant_config_compaction_algorithm: Optional[dict[str, Any]],
     pageserver_virtual_file_io_mode: Optional[str],
 ) -> Iterator[NeonEnv]:
     """
@@ -1459,7 +1455,7 @@ def neon_env_builder(
     test_overlay_dir: Path,
     top_output_dir: Path,
     pageserver_virtual_file_io_engine: str,
-    pageserver_default_tenant_config_compaction_algorithm: Optional[Dict[str, Any]],
+    pageserver_default_tenant_config_compaction_algorithm: Optional[dict[str, Any]],
     pageserver_aux_file_policy: Optional[AuxFileStore],
     record_property: Callable[[str, object], None],
     pageserver_virtual_file_io_mode: Optional[str],
@@ -1521,7 +1517,7 @@ class LogUtils:
 
     def assert_log_contains(
         self, pattern: str, offset: None | LogCursor = None
-    ) -> Tuple[str, LogCursor]:
+    ) -> tuple[str, LogCursor]:
         """Convenient for use inside wait_until()"""
 
         res = self.log_contains(pattern, offset=offset)
@@ -1530,7 +1526,7 @@ class LogUtils:
 
     def log_contains(
         self, pattern: str, offset: None | LogCursor = None
-    ) -> Optional[Tuple[str, LogCursor]]:
+    ) -> Optional[tuple[str, LogCursor]]:
         """Check that the log contains a line that matches the given regex"""
         logfile = self.logfile
         if not logfile.exists():
@@ -1611,7 +1607,7 @@ class NeonStorageController(MetricsGetter, LogUtils):
         self.running = True
         return self
 
-    def stop(self, immediate: bool = False) -> "NeonStorageController":
+    def stop(self, immediate: bool = False) -> NeonStorageController:
         if self.running:
             self.env.neon_cli.storage_controller_stop(immediate)
             self.running = False
@@ -1673,7 +1669,7 @@ class NeonStorageController(MetricsGetter, LogUtils):
 
         return resp
 
-    def headers(self, scope: Optional[TokenScope]) -> Dict[str, str]:
+    def headers(self, scope: Optional[TokenScope]) -> dict[str, str]:
         headers = {}
         if self.auth_enabled and scope is not None:
             jwt_token = self.env.auth_keys.generate_token(scope=scope)
@@ -1859,13 +1855,13 @@ class NeonStorageController(MetricsGetter, LogUtils):
         tenant_id: TenantId,
         shard_count: Optional[int] = None,
         shard_stripe_size: Optional[int] = None,
-        tenant_config: Optional[Dict[Any, Any]] = None,
-        placement_policy: Optional[Union[Dict[Any, Any] | str]] = None,
+        tenant_config: Optional[dict[Any, Any]] = None,
+        placement_policy: Optional[Union[dict[Any, Any] | str]] = None,
     ):
         """
         Use this rather than pageserver_api() when you need to include shard parameters
         """
-        body: Dict[str, Any] = {"new_tenant_id": str(tenant_id)}
+        body: dict[str, Any] = {"new_tenant_id": str(tenant_id)}
 
         if shard_count is not None:
             shard_params = {"count": shard_count}
@@ -2081,8 +2077,8 @@ class NeonStorageController(MetricsGetter, LogUtils):
 
                 time.sleep(backoff)
 
-    def metadata_health_update(self, healthy: List[TenantShardId], unhealthy: List[TenantShardId]):
-        body: Dict[str, Any] = {
+    def metadata_health_update(self, healthy: list[TenantShardId], unhealthy: list[TenantShardId]):
+        body: dict[str, Any] = {
             "healthy_tenant_shards": [str(t) for t in healthy],
             "unhealthy_tenant_shards": [str(t) for t in unhealthy],
         }
@@ -2103,7 +2099,7 @@ class NeonStorageController(MetricsGetter, LogUtils):
         return response.json()
 
     def metadata_health_list_outdated(self, duration: str):
-        body: Dict[str, Any] = {"not_scrubbed_for": duration}
+        body: dict[str, Any] = {"not_scrubbed_for": duration}
 
         response = self.request(
             "POST",
@@ -2137,7 +2133,7 @@ class NeonStorageController(MetricsGetter, LogUtils):
         response.raise_for_status()
         return response.json()
 
-    def configure_failpoints(self, config_strings: Tuple[str, str] | List[Tuple[str, str]]):
+    def configure_failpoints(self, config_strings: tuple[str, str] | list[tuple[str, str]]):
         if isinstance(config_strings, tuple):
             pairs = [config_strings]
         else:
@@ -2154,13 +2150,13 @@ class NeonStorageController(MetricsGetter, LogUtils):
         log.info(f"Got failpoints request response code {res.status_code}")
         res.raise_for_status()
 
-    def get_tenants_placement(self) -> defaultdict[str, Dict[str, Any]]:
+    def get_tenants_placement(self) -> defaultdict[str, dict[str, Any]]:
         """
         Get the intent and observed placements of all tenants known to the storage controller.
         """
         tenants = self.tenant_list()
 
-        tenant_placement: defaultdict[str, Dict[str, Any]] = defaultdict(
+        tenant_placement: defaultdict[str, dict[str, Any]] = defaultdict(
             lambda: {
                 "observed": {"attached": None, "secondary": []},
                 "intent": {"attached": None, "secondary": []},
@@ -2267,12 +2263,12 @@ class NeonStorageController(MetricsGetter, LogUtils):
         response.raise_for_status()
         return [TenantShardId.parse(tid) for tid in response.json()["updated"]]
 
-    def __enter__(self) -> "NeonStorageController":
+    def __enter__(self) -> NeonStorageController:
         return self
 
     def __exit__(
         self,
-        exc_type: Optional[Type[BaseException]],
+        exc_type: Optional[type[BaseException]],
         exc: Optional[BaseException],
         tb: Optional[TracebackType],
     ):
@@ -2281,7 +2277,7 @@ class NeonStorageController(MetricsGetter, LogUtils):
 
 class NeonProxiedStorageController(NeonStorageController):
     def __init__(self, env: NeonEnv, proxy_port: int, auth_enabled: bool):
-        super(NeonProxiedStorageController, self).__init__(env, proxy_port, auth_enabled)
+        super().__init__(env, proxy_port, auth_enabled)
         self.instances: dict[int, dict[str, Any]] = {}
 
     def start(
@@ -2300,7 +2296,7 @@ class NeonProxiedStorageController(NeonStorageController):
 
     def stop_instance(
         self, immediate: bool = False, instance_id: Optional[int] = None
-    ) -> "NeonStorageController":
+    ) -> NeonStorageController:
         assert instance_id in self.instances
         if self.instances[instance_id]["running"]:
             self.env.neon_cli.storage_controller_stop(immediate, instance_id)
@@ -2309,7 +2305,7 @@ class NeonProxiedStorageController(NeonStorageController):
         self.running = any(meta["running"] for meta in self.instances.values())
         return self
 
-    def stop(self, immediate: bool = False) -> "NeonStorageController":
+    def stop(self, immediate: bool = False) -> NeonStorageController:
         for iid, details in self.instances.items():
             if details["running"]:
                 self.env.neon_cli.storage_controller_stop(immediate, iid)
@@ -2328,7 +2324,7 @@ class NeonProxiedStorageController(NeonStorageController):
 
     def log_contains(
         self, pattern: str, offset: None | LogCursor = None
-    ) -> Optional[Tuple[str, LogCursor]]:
+    ) -> Optional[tuple[str, LogCursor]]:
         raise NotImplementedError()
 
 
@@ -2360,7 +2356,7 @@ class NeonPageserver(PgProtocol, LogUtils):
         # env.pageserver.allowed_errors.append(".*could not open garage door.*")
         #
         # The entries in the list are regular experessions.
-        self.allowed_errors: List[str] = list(DEFAULT_PAGESERVER_ALLOWED_ERRORS)
+        self.allowed_errors: list[str] = list(DEFAULT_PAGESERVER_ALLOWED_ERRORS)
 
     def timeline_dir(
         self,
@@ -2385,19 +2381,19 @@ class NeonPageserver(PgProtocol, LogUtils):
     def config_toml_path(self) -> Path:
         return self.workdir / "pageserver.toml"
 
-    def edit_config_toml(self, edit_fn: Callable[[Dict[str, Any]], T]) -> T:
+    def edit_config_toml(self, edit_fn: Callable[[dict[str, Any]], T]) -> T:
         """
         Edit the pageserver's config toml file in place.
         """
         path = self.config_toml_path
-        with open(path, "r") as f:
+        with open(path) as f:
             config = toml.load(f)
         res = edit_fn(config)
         with open(path, "w") as f:
             toml.dump(config, f)
         return res
 
-    def patch_config_toml_nonrecursive(self, patch: Dict[str, Any]) -> Dict[str, Any]:
+    def patch_config_toml_nonrecursive(self, patch: dict[str, Any]) -> dict[str, Any]:
         """
         Non-recursively merge the given `patch` dict into the existing config toml, using `dict.update()`.
         Returns the replaced values.
@@ -2406,7 +2402,7 @@ class NeonPageserver(PgProtocol, LogUtils):
         """
         replacements = {}
 
-        def doit(config: Dict[str, Any]):
+        def doit(config: dict[str, Any]):
             while len(patch) > 0:
                 key, new = patch.popitem()
                 old = config.get(key, None)
@@ -2418,9 +2414,9 @@ class NeonPageserver(PgProtocol, LogUtils):
 
     def start(
         self,
-        extra_env_vars: Optional[Dict[str, str]] = None,
+        extra_env_vars: Optional[dict[str, str]] = None,
         timeout_in_seconds: Optional[int] = None,
-    ) -> "NeonPageserver":
+    ) -> NeonPageserver:
         """
         Start the page server.
         `overrides` allows to add some config to this pageserver start.
@@ -2446,7 +2442,7 @@ class NeonPageserver(PgProtocol, LogUtils):
 
         return self
 
-    def stop(self, immediate: bool = False) -> "NeonPageserver":
+    def stop(self, immediate: bool = False) -> NeonPageserver:
         """
         Stop the page server.
         Returns self.
@@ -2494,12 +2490,12 @@ class NeonPageserver(PgProtocol, LogUtils):
 
         wait_until(20, 0.5, complete)
 
-    def __enter__(self) -> "NeonPageserver":
+    def __enter__(self) -> NeonPageserver:
         return self
 
     def __exit__(
         self,
-        exc_type: Optional[Type[BaseException]],
+        exc_type: Optional[type[BaseException]],
         exc: Optional[BaseException],
         tb: Optional[TracebackType],
     ):
@@ -2546,7 +2542,7 @@ class NeonPageserver(PgProtocol, LogUtils):
     def tenant_attach(
         self,
         tenant_id: TenantId,
-        config: None | Dict[str, Any] = None,
+        config: None | dict[str, Any] = None,
         generation: Optional[int] = None,
         override_storage_controller_generation: bool = False,
     ):
@@ -2585,7 +2581,7 @@ class NeonPageserver(PgProtocol, LogUtils):
     ) -> dict[str, Any]:
         path = self.tenant_dir(tenant_shard_id) / "config-v1"
         log.info(f"Reading location conf from {path}")
-        bytes = open(path, "r").read()
+        bytes = open(path).read()
         try:
             decoded: dict[str, Any] = toml.loads(bytes)
             return decoded
@@ -2596,7 +2592,7 @@ class NeonPageserver(PgProtocol, LogUtils):
     def tenant_create(
         self,
         tenant_id: TenantId,
-        conf: Optional[Dict[str, Any]] = None,
+        conf: Optional[dict[str, Any]] = None,
         auth_token: Optional[str] = None,
         generation: Optional[int] = None,
     ) -> TenantId:
@@ -2662,7 +2658,7 @@ class PgBin:
         self.env = os.environ.copy()
         self.env["LD_LIBRARY_PATH"] = str(self.pg_lib_dir)
 
-    def _fixpath(self, command: List[str]):
+    def _fixpath(self, command: list[str]):
         if "/" not in str(command[0]):
             command[0] = str(self.pg_bin_path / command[0])
 
@@ -2682,7 +2678,7 @@ class PgBin:
 
     def run_nonblocking(
         self,
-        command: List[str],
+        command: list[str],
         env: Optional[Env] = None,
         cwd: Optional[Union[str, Path]] = None,
     ) -> subprocess.Popen[Any]:
@@ -2706,7 +2702,7 @@ class PgBin:
 
     def run(
         self,
-        command: List[str],
+        command: list[str],
         env: Optional[Env] = None,
         cwd: Optional[Union[str, Path]] = None,
     ) -> None:
@@ -2729,7 +2725,7 @@ class PgBin:
 
     def run_capture(
         self,
-        command: List[str],
+        command: list[str],
         env: Optional[Env] = None,
         cwd: Optional[str] = None,
         with_command_header=True,
@@ -2842,14 +2838,14 @@ class VanillaPostgres(PgProtocol):
             ]
         )
 
-    def configure(self, options: List[str]):
+    def configure(self, options: list[str]):
         """Append lines into postgresql.conf file."""
         assert not self.running
         with open(os.path.join(self.pgdatadir, "postgresql.conf"), "a") as conf_file:
             conf_file.write("\n".join(options))
             conf_file.write("\n")
 
-    def edit_hba(self, hba: List[str]):
+    def edit_hba(self, hba: list[str]):
         """Prepend hba lines into pg_hba.conf file."""
         assert not self.running
         with open(os.path.join(self.pgdatadir, "pg_hba.conf"), "r+") as conf_file:
@@ -2877,12 +2873,12 @@ class VanillaPostgres(PgProtocol):
         """Return size of pgdatadir subdirectory in bytes."""
         return get_dir_size(self.pgdatadir / subdir)
 
-    def __enter__(self) -> "VanillaPostgres":
+    def __enter__(self) -> VanillaPostgres:
         return self
 
     def __exit__(
         self,
-        exc_type: Optional[Type[BaseException]],
+        exc_type: Optional[type[BaseException]],
         exc: Optional[BaseException],
         tb: Optional[TracebackType],
     ):
@@ -2912,7 +2908,7 @@ class RemotePostgres(PgProtocol):
         # The remote server is assumed to be running already
         self.running = True
 
-    def configure(self, options: List[str]):
+    def configure(self, options: list[str]):
         raise Exception("cannot change configuration of remote Posgres instance")
 
     def start(self):
@@ -2926,12 +2922,12 @@ class RemotePostgres(PgProtocol):
         # See https://www.postgresql.org/docs/14/functions-admin.html#FUNCTIONS-ADMIN-GENFILE
         raise Exception("cannot get size of a Postgres instance")
 
-    def __enter__(self) -> "RemotePostgres":
+    def __enter__(self) -> RemotePostgres:
         return self
 
     def __exit__(
         self,
-        exc_type: Optional[Type[BaseException]],
+        exc_type: Optional[type[BaseException]],
         exc: Optional[BaseException],
         tb: Optional[TracebackType],
     ):
@@ -3267,7 +3263,7 @@ class NeonProxy(PgProtocol):
 
     def __exit__(
         self,
-        exc_type: Optional[Type[BaseException]],
+        exc_type: Optional[type[BaseException]],
         exc: Optional[BaseException],
         tb: Optional[TracebackType],
     ):
@@ -3405,7 +3401,7 @@ class Endpoint(PgProtocol, LogUtils):
         self.http_port = http_port
         self.check_stop_result = check_stop_result
         # passed to endpoint create and endpoint reconfigure
-        self.active_safekeepers: List[int] = list(map(lambda sk: sk.id, env.safekeepers))
+        self.active_safekeepers: list[int] = list(map(lambda sk: sk.id, env.safekeepers))
         # path to conf is <repo_dir>/endpoints/<endpoint_id>/pgdata/postgresql.conf
 
         # Semaphore is set to 1 when we start, and acquire'd back to zero when we stop
@@ -3428,10 +3424,10 @@ class Endpoint(PgProtocol, LogUtils):
         endpoint_id: Optional[str] = None,
         hot_standby: bool = False,
         lsn: Optional[Lsn] = None,
-        config_lines: Optional[List[str]] = None,
+        config_lines: Optional[list[str]] = None,
         pageserver_id: Optional[int] = None,
         allow_multiple: bool = False,
-    ) -> "Endpoint":
+    ) -> Endpoint:
         """
         Create a new Postgres endpoint.
         Returns self.
@@ -3474,10 +3470,10 @@ class Endpoint(PgProtocol, LogUtils):
         self,
         remote_ext_config: Optional[str] = None,
         pageserver_id: Optional[int] = None,
-        safekeepers: Optional[List[int]] = None,
+        safekeepers: Optional[list[int]] = None,
         allow_multiple: bool = False,
         basebackup_request_tries: Optional[int] = None,
-    ) -> "Endpoint":
+    ) -> Endpoint:
         """
         Start the Postgres instance.
         Returns self.
@@ -3490,8 +3486,6 @@ class Endpoint(PgProtocol, LogUtils):
         if safekeepers is not None:
             self.active_safekeepers = safekeepers
 
-        log.info(f"Starting postgres endpoint {self.endpoint_id}")
-
         self.env.neon_cli.endpoint_start(
             self.endpoint_id,
             safekeepers=self.active_safekeepers,
@@ -3526,7 +3520,7 @@ class Endpoint(PgProtocol, LogUtils):
         """Path to the postgresql.conf in the endpoint directory (not the one in pgdata)"""
         return self.endpoint_path() / "postgresql.conf"
 
-    def config(self, lines: List[str]) -> "Endpoint":
+    def config(self, lines: list[str]) -> Endpoint:
         """
         Add lines to postgresql.conf.
         Lines should be an array of valid postgresql.conf rows.
@@ -3540,7 +3534,7 @@ class Endpoint(PgProtocol, LogUtils):
 
         return self
 
-    def edit_hba(self, hba: List[str]):
+    def edit_hba(self, hba: list[str]):
         """Prepend hba lines into pg_hba.conf file."""
         with open(os.path.join(self.pg_data_dir_path(), "pg_hba.conf"), "r+") as conf_file:
             data = conf_file.read()
@@ -3555,7 +3549,7 @@ class Endpoint(PgProtocol, LogUtils):
         return self._running._value > 0
 
     def reconfigure(
-        self, pageserver_id: Optional[int] = None, safekeepers: Optional[List[int]] = None
+        self, pageserver_id: Optional[int] = None, safekeepers: Optional[list[int]] = None
     ):
         assert self.endpoint_id is not None
         # If `safekeepers` is not None, they are remember them as active and use
@@ -3570,7 +3564,7 @@ class Endpoint(PgProtocol, LogUtils):
         """Update the endpoint.json file used by control_plane."""
         # Read config
         config_path = os.path.join(self.endpoint_path(), "endpoint.json")
-        with open(config_path, "r") as f:
+        with open(config_path) as f:
             data_dict: dict[str, Any] = json.load(f)
 
         # Write it back updated
@@ -3603,8 +3597,8 @@ class Endpoint(PgProtocol, LogUtils):
     def stop(
         self,
         mode: str = "fast",
-        sks_wait_walreceiver_gone: Optional[tuple[List[Safekeeper], TimelineId]] = None,
-    ) -> "Endpoint":
+        sks_wait_walreceiver_gone: Optional[tuple[list[Safekeeper], TimelineId]] = None,
+    ) -> Endpoint:
         """
         Stop the Postgres instance if it's running.
 
@@ -3638,7 +3632,7 @@ class Endpoint(PgProtocol, LogUtils):
 
         return self
 
-    def stop_and_destroy(self, mode: str = "immediate") -> "Endpoint":
+    def stop_and_destroy(self, mode: str = "immediate") -> Endpoint:
         """
         Stop the Postgres instance, then destroy the endpoint.
         Returns self.
@@ -3660,19 +3654,17 @@ class Endpoint(PgProtocol, LogUtils):
         endpoint_id: Optional[str] = None,
         hot_standby: bool = False,
         lsn: Optional[Lsn] = None,
-        config_lines: Optional[List[str]] = None,
+        config_lines: Optional[list[str]] = None,
         remote_ext_config: Optional[str] = None,
         pageserver_id: Optional[int] = None,
         allow_multiple=False,
         basebackup_request_tries: Optional[int] = None,
-    ) -> "Endpoint":
+    ) -> Endpoint:
         """
         Create an endpoint, apply config, and start Postgres.
         Returns self.
         """
 
-        started_at = time.time()
-
         self.create(
             branch_name=branch_name,
             endpoint_id=endpoint_id,
@@ -3688,16 +3680,14 @@ class Endpoint(PgProtocol, LogUtils):
             basebackup_request_tries=basebackup_request_tries,
         )
 
-        log.info(f"Postgres startup took {time.time() - started_at} seconds")
-
         return self
 
-    def __enter__(self) -> "Endpoint":
+    def __enter__(self) -> Endpoint:
         return self
 
     def __exit__(
         self,
-        exc_type: Optional[Type[BaseException]],
+        exc_type: Optional[type[BaseException]],
         exc: Optional[BaseException],
         tb: Optional[TracebackType],
     ):
@@ -3728,7 +3718,7 @@ class EndpointFactory:
     def __init__(self, env: NeonEnv):
         self.env = env
         self.num_instances: int = 0
-        self.endpoints: List[Endpoint] = []
+        self.endpoints: list[Endpoint] = []
 
     def create_start(
         self,
@@ -3737,7 +3727,7 @@ class EndpointFactory:
         tenant_id: Optional[TenantId] = None,
         lsn: Optional[Lsn] = None,
         hot_standby: bool = False,
-        config_lines: Optional[List[str]] = None,
+        config_lines: Optional[list[str]] = None,
         remote_ext_config: Optional[str] = None,
         pageserver_id: Optional[int] = None,
         basebackup_request_tries: Optional[int] = None,
@@ -3769,7 +3759,7 @@ class EndpointFactory:
         tenant_id: Optional[TenantId] = None,
         lsn: Optional[Lsn] = None,
         hot_standby: bool = False,
-        config_lines: Optional[List[str]] = None,
+        config_lines: Optional[list[str]] = None,
         pageserver_id: Optional[int] = None,
     ) -> Endpoint:
         ep = Endpoint(
@@ -3793,7 +3783,7 @@ class EndpointFactory:
             pageserver_id=pageserver_id,
         )
 
-    def stop_all(self, fail_on_error=True) -> "EndpointFactory":
+    def stop_all(self, fail_on_error=True) -> EndpointFactory:
         exception = None
         for ep in self.endpoints:
             try:
@@ -3808,7 +3798,7 @@ class EndpointFactory:
         return self
 
     def new_replica(
-        self, origin: Endpoint, endpoint_id: str, config_lines: Optional[List[str]] = None
+        self, origin: Endpoint, endpoint_id: str, config_lines: Optional[list[str]] = None
     ):
         branch_name = origin.branch_name
         assert origin in self.endpoints
@@ -3824,7 +3814,7 @@ class EndpointFactory:
         )
 
     def new_replica_start(
-        self, origin: Endpoint, endpoint_id: str, config_lines: Optional[List[str]] = None
+        self, origin: Endpoint, endpoint_id: str, config_lines: Optional[list[str]] = None
     ):
         branch_name = origin.branch_name
         assert origin in self.endpoints
@@ -3862,7 +3852,7 @@ class Safekeeper(LogUtils):
         port: SafekeeperPort,
         id: int,
         running: bool = False,
-        extra_opts: Optional[List[str]] = None,
+        extra_opts: Optional[list[str]] = None,
     ):
         self.env = env
         self.port = port
@@ -3888,8 +3878,8 @@ class Safekeeper(LogUtils):
         self.extra_opts = extra_opts
 
     def start(
-        self, extra_opts: Optional[List[str]] = None, timeout_in_seconds: Optional[int] = None
-    ) -> "Safekeeper":
+        self, extra_opts: Optional[list[str]] = None, timeout_in_seconds: Optional[int] = None
+    ) -> Safekeeper:
         if extra_opts is None:
             # Apply either the extra_opts passed in, or the ones from our constructor: we do not merge the two.
             extra_opts = self.extra_opts
@@ -3924,8 +3914,7 @@ class Safekeeper(LogUtils):
                 break  # success
         return self
 
-    def stop(self, immediate: bool = False) -> "Safekeeper":
-        log.info(f"Stopping safekeeper {self.id}")
+    def stop(self, immediate: bool = False) -> Safekeeper:
         self.env.neon_cli.safekeeper_stop(self.id, immediate)
         self.running = False
         return self
@@ -3936,8 +3925,8 @@ class Safekeeper(LogUtils):
         assert not self.log_contains("timeout while acquiring WalResidentTimeline guard")
 
     def append_logical_message(
-        self, tenant_id: TenantId, timeline_id: TimelineId, request: Dict[str, Any]
-    ) -> Dict[str, Any]:
+        self, tenant_id: TenantId, timeline_id: TimelineId, request: dict[str, Any]
+    ) -> dict[str, Any]:
         """
         Send JSON_CTRL query to append LogicalMessage to WAL and modify
         safekeeper state. It will construct LogicalMessage from provided
@@ -3990,7 +3979,7 @@ class Safekeeper(LogUtils):
 
     def pull_timeline(
         self, srcs: list[Safekeeper], tenant_id: TenantId, timeline_id: TimelineId
-    ) -> Dict[str, Any]:
+    ) -> dict[str, Any]:
         """
         pull_timeline from srcs to self.
         """
@@ -4026,7 +4015,7 @@ class Safekeeper(LogUtils):
         mysegs = [s for s in segs if f"sk{self.id}" in s]
         return mysegs
 
-    def list_segments(self, tenant_id, timeline_id) -> List[str]:
+    def list_segments(self, tenant_id, timeline_id) -> list[str]:
         """
         Get list of segment names of the given timeline.
         """
@@ -4131,7 +4120,7 @@ class StorageScrubber:
         self.log_dir = log_dir
 
     def scrubber_cli(
-        self, args: list[str], timeout, extra_env: Optional[Dict[str, str]] = None
+        self, args: list[str], timeout, extra_env: Optional[dict[str, str]] = None
     ) -> str:
         assert isinstance(self.env.pageserver_remote_storage, S3Storage)
         s3_storage = self.env.pageserver_remote_storage
@@ -4178,10 +4167,10 @@ class StorageScrubber:
 
     def scan_metadata_safekeeper(
         self,
-        timeline_lsns: List[Dict[str, Any]],
+        timeline_lsns: list[dict[str, Any]],
         cloud_admin_api_url: str,
         cloud_admin_api_token: str,
-    ) -> Tuple[bool, Any]:
+    ) -> tuple[bool, Any]:
         extra_env = {
             "CLOUD_ADMIN_API_URL": cloud_admin_api_url,
             "CLOUD_ADMIN_API_TOKEN": cloud_admin_api_token,
@@ -4194,9 +4183,9 @@ class StorageScrubber:
         self,
         post_to_storage_controller: bool = False,
         node_kind: NodeKind = NodeKind.PAGESERVER,
-        timeline_lsns: Optional[List[Dict[str, Any]]] = None,
-        extra_env: Optional[Dict[str, str]] = None,
-    ) -> Tuple[bool, Any]:
+        timeline_lsns: Optional[list[dict[str, Any]]] = None,
+        extra_env: Optional[dict[str, str]] = None,
+    ) -> tuple[bool, Any]:
         """
         Returns the health status and the metadata summary.
         """
@@ -4503,7 +4492,7 @@ def should_skip_file(filename: str) -> bool:
 #
 # Test helpers
 #
-def list_files_to_compare(pgdata_dir: Path) -> List[str]:
+def list_files_to_compare(pgdata_dir: Path) -> list[str]:
     pgdata_files = []
     for root, _dirs, filenames in os.walk(pgdata_dir):
         for filename in filenames:

test_runner/fixtures/overlayfs.py

@@ -1,5 +1,7 @@
+from __future__ import annotations
+
+from collections.abc import Iterator
 from pathlib import Path
-from typing import Iterator
 
 import psutil
 

test_runner/fixtures/pageserver/__init__.py

@@ -0,0 +1 @@
+from __future__ import annotations

test_runner/fixtures/pageserver/allowed_errors.py

@@ -1,14 +1,16 @@
 #! /usr/bin/env python3
 
+from __future__ import annotations
+
 import argparse
 import re
 import sys
-from typing import Iterable, List, Tuple
+from collections.abc import Iterable
 
 
 def scan_pageserver_log_for_errors(
-    input: Iterable[str], allowed_errors: List[str]
-) -> List[Tuple[int, str]]:
+    input: Iterable[str], allowed_errors: list[str]
+) -> list[tuple[int, str]]:
     error_or_warn = re.compile(r"\s(ERROR|WARN)")
     errors = []
     for lineno, line in enumerate(input, start=1):
@@ -113,7 +115,7 @@ DEFAULT_STORAGE_CONTROLLER_ALLOWED_ERRORS = [
 
 
 def _check_allowed_errors(input):
-    allowed_errors: List[str] = list(DEFAULT_PAGESERVER_ALLOWED_ERRORS)
+    allowed_errors: list[str] = list(DEFAULT_PAGESERVER_ALLOWED_ERRORS)
 
     # add any test specifics here; cli parsing is not provided for the
     # difficulty of copypasting regexes as arguments without any quoting

test_runner/fixtures/pageserver/common_types.py

@@ -1,9 +1,14 @@
+from __future__ import annotations
+
 import re
 from dataclasses import dataclass
-from typing import Any, Dict, Tuple, Union
+from typing import TYPE_CHECKING, Union
 
 from fixtures.common_types import KEY_MAX, KEY_MIN, Key, Lsn
 
+if TYPE_CHECKING:
+    from typing import Any
+
 
 @dataclass
 class IndexLayerMetadata:
@@ -53,7 +58,7 @@ IMAGE_LAYER_FILE_NAME = re.compile(
 )
 
 
-def parse_image_layer(f_name: str) -> Tuple[int, int, int]:
+def parse_image_layer(f_name: str) -> tuple[int, int, int]:
     """Parse an image layer file name. Return key start, key end, and snapshot lsn"""
 
     match = IMAGE_LAYER_FILE_NAME.match(f_name)
@@ -68,7 +73,7 @@ DELTA_LAYER_FILE_NAME = re.compile(
 )
 
 
-def parse_delta_layer(f_name: str) -> Tuple[int, int, int, int]:
+def parse_delta_layer(f_name: str) -> tuple[int, int, int, int]:
     """Parse a delta layer file name. Return key start, key end, lsn start, and lsn end"""
     match = DELTA_LAYER_FILE_NAME.match(f_name)
     if match is None:
@@ -121,11 +126,11 @@ def is_future_layer(layer_file_name: LayerName, disk_consistent_lsn: Lsn):
 
 @dataclass
 class IndexPartDump:
-    layer_metadata: Dict[LayerName, IndexLayerMetadata]
+    layer_metadata: dict[LayerName, IndexLayerMetadata]
     disk_consistent_lsn: Lsn
 
     @classmethod
-    def from_json(cls, d: Dict[str, Any]) -> "IndexPartDump":
+    def from_json(cls, d: dict[str, Any]) -> IndexPartDump:
         return IndexPartDump(
             layer_metadata={
                 parse_layer_file_name(n): IndexLayerMetadata(v["file_size"], v["generation"])

test_runner/fixtures/pageserver/http.py

@@ -4,7 +4,7 @@ import time
 from collections import defaultdict
 from dataclasses import dataclass
 from datetime import datetime
-from typing import Any, Dict, List, Optional, Set, Tuple, Union
+from typing import TYPE_CHECKING, Any
 
 import requests
 from requests.adapters import HTTPAdapter
@@ -16,6 +16,9 @@ from fixtures.metrics import Metrics, MetricsGetter, parse_metrics
 from fixtures.pg_version import PgVersion
 from fixtures.utils import Fn
 
+if TYPE_CHECKING:
+    from typing import Optional, Union
+
 
 class PageserverApiException(Exception):
     def __init__(self, message, status_code: int):
@@ -43,7 +46,7 @@ class InMemoryLayerInfo:
     lsn_end: Optional[str]
 
     @classmethod
-    def from_json(cls, d: Dict[str, Any]) -> InMemoryLayerInfo:
+    def from_json(cls, d: dict[str, Any]) -> InMemoryLayerInfo:
         return InMemoryLayerInfo(
             kind=d["kind"],
             lsn_start=d["lsn_start"],
@@ -64,7 +67,7 @@ class HistoricLayerInfo:
     visible: bool
 
     @classmethod
-    def from_json(cls, d: Dict[str, Any]) -> HistoricLayerInfo:
+    def from_json(cls, d: dict[str, Any]) -> HistoricLayerInfo:
         # instead of parsing the key range lets keep the definition of "L0" in pageserver
         l0_ness = d.get("l0")
         assert l0_ness is None or isinstance(l0_ness, bool)
@@ -86,53 +89,53 @@ class HistoricLayerInfo:
 
 @dataclass
 class LayerMapInfo:
-    in_memory_layers: List[InMemoryLayerInfo]
-    historic_layers: List[HistoricLayerInfo]
+    in_memory_layers: list[InMemoryLayerInfo]
+    historic_layers: list[HistoricLayerInfo]
 
     @classmethod
-    def from_json(cls, d: Dict[str, Any]) -> LayerMapInfo:
+    def from_json(cls, d: dict[str, Any]) -> LayerMapInfo:
         info = LayerMapInfo(in_memory_layers=[], historic_layers=[])
 
         json_in_memory_layers = d["in_memory_layers"]
-        assert isinstance(json_in_memory_layers, List)
+        assert isinstance(json_in_memory_layers, list)
         for json_in_memory_layer in json_in_memory_layers:
             info.in_memory_layers.append(InMemoryLayerInfo.from_json(json_in_memory_layer))
 
         json_historic_layers = d["historic_layers"]
-        assert isinstance(json_historic_layers, List)
+        assert isinstance(json_historic_layers, list)
         for json_historic_layer in json_historic_layers:
             info.historic_layers.append(HistoricLayerInfo.from_json(json_historic_layer))
 
         return info
 
-    def kind_count(self) -> Dict[str, int]:
-        counts: Dict[str, int] = defaultdict(int)
+    def kind_count(self) -> dict[str, int]:
+        counts: dict[str, int] = defaultdict(int)
         for inmem_layer in self.in_memory_layers:
             counts[inmem_layer.kind] += 1
         for hist_layer in self.historic_layers:
             counts[hist_layer.kind] += 1
         return counts
 
-    def delta_layers(self) -> List[HistoricLayerInfo]:
+    def delta_layers(self) -> list[HistoricLayerInfo]:
         return [x for x in self.historic_layers if x.kind == "Delta"]
 
-    def image_layers(self) -> List[HistoricLayerInfo]:
+    def image_layers(self) -> list[HistoricLayerInfo]:
         return [x for x in self.historic_layers if x.kind == "Image"]
 
-    def delta_l0_layers(self) -> List[HistoricLayerInfo]:
+    def delta_l0_layers(self) -> list[HistoricLayerInfo]:
         return [x for x in self.historic_layers if x.kind == "Delta" and x.l0]
 
-    def historic_by_name(self) -> Set[str]:
+    def historic_by_name(self) -> set[str]:
         return set(x.layer_file_name for x in self.historic_layers)
 
 
 @dataclass
 class TenantConfig:
-    tenant_specific_overrides: Dict[str, Any]
-    effective_config: Dict[str, Any]
+    tenant_specific_overrides: dict[str, Any]
+    effective_config: dict[str, Any]
 
     @classmethod
-    def from_json(cls, d: Dict[str, Any]) -> TenantConfig:
+    def from_json(cls, d: dict[str, Any]) -> TenantConfig:
         return TenantConfig(
             tenant_specific_overrides=d["tenant_specific_overrides"],
             effective_config=d["effective_config"],
@@ -209,7 +212,7 @@ class PageserverHttpClient(requests.Session, MetricsGetter):
     def check_status(self):
         self.get(f"http://localhost:{self.port}/v1/status").raise_for_status()
 
-    def configure_failpoints(self, config_strings: Tuple[str, str] | List[Tuple[str, str]]):
+    def configure_failpoints(self, config_strings: tuple[str, str] | list[tuple[str, str]]):
         self.is_testing_enabled_or_skip()
 
         if isinstance(config_strings, tuple):
@@ -233,7 +236,7 @@ class PageserverHttpClient(requests.Session, MetricsGetter):
         res = self.post(f"http://localhost:{self.port}/v1/reload_auth_validation_keys")
         self.verbose_error(res)
 
-    def tenant_list(self) -> List[Dict[Any, Any]]:
+    def tenant_list(self) -> list[dict[Any, Any]]:
         res = self.get(f"http://localhost:{self.port}/v1/tenant")
         self.verbose_error(res)
         res_json = res.json()
@@ -244,7 +247,7 @@ class PageserverHttpClient(requests.Session, MetricsGetter):
         self,
         tenant_id: Union[TenantId, TenantShardId],
         generation: int,
-        config: None | Dict[str, Any] = None,
+        config: None | dict[str, Any] = None,
     ):
         config = config or {}
 
@@ -324,7 +327,7 @@ class PageserverHttpClient(requests.Session, MetricsGetter):
 
     def tenant_status(
         self, tenant_id: Union[TenantId, TenantShardId], activate: bool = False
-    ) -> Dict[Any, Any]:
+    ) -> dict[Any, Any]:
         """
         :activate: hint the server not to accelerate activation of this tenant in response
         to this query.  False by default for tests, because they generally want to observed the
@@ -378,8 +381,8 @@ class PageserverHttpClient(requests.Session, MetricsGetter):
     def patch_tenant_config_client_side(
         self,
         tenant_id: TenantId,
-        inserts: Optional[Dict[str, Any]] = None,
-        removes: Optional[List[str]] = None,
+        inserts: Optional[dict[str, Any]] = None,
+        removes: Optional[list[str]] = None,
     ):
         current = self.tenant_config(tenant_id).tenant_specific_overrides
         if inserts is not None:
@@ -394,7 +397,7 @@ class PageserverHttpClient(requests.Session, MetricsGetter):
 
     def tenant_size_and_modelinputs(
         self, tenant_id: Union[TenantId, TenantShardId]
-    ) -> Tuple[int, Dict[str, Any]]:
+    ) -> tuple[int, dict[str, Any]]:
         """
         Returns the tenant size, together with the model inputs as the second tuple item.
         """
@@ -424,7 +427,7 @@ class PageserverHttpClient(requests.Session, MetricsGetter):
         tenant_id: Union[TenantId, TenantShardId],
         timestamp: datetime,
         done_if_after: datetime,
-        shard_counts: Optional[List[int]] = None,
+        shard_counts: Optional[list[int]] = None,
     ):
         """
         Issues a request to perform time travel operations on the remote storage
@@ -432,7 +435,7 @@ class PageserverHttpClient(requests.Session, MetricsGetter):
 
         if shard_counts is None:
             shard_counts = []
-        body: Dict[str, Any] = {
+        body: dict[str, Any] = {
             "shard_counts": shard_counts,
         }
         res = self.put(
@@ -446,7 +449,7 @@ class PageserverHttpClient(requests.Session, MetricsGetter):
         tenant_id: Union[TenantId, TenantShardId],
         include_non_incremental_logical_size: bool = False,
         include_timeline_dir_layer_file_size_sum: bool = False,
-    ) -> List[Dict[str, Any]]:
+    ) -> list[dict[str, Any]]:
         params = {}
         if include_non_incremental_logical_size:
             params["include-non-incremental-logical-size"] = "true"
@@ -470,8 +473,8 @@ class PageserverHttpClient(requests.Session, MetricsGetter):
         ancestor_start_lsn: Optional[Lsn] = None,
         existing_initdb_timeline_id: Optional[TimelineId] = None,
         **kwargs,
-    ) -> Dict[Any, Any]:
-        body: Dict[str, Any] = {
+    ) -> dict[Any, Any]:
+        body: dict[str, Any] = {
             "new_timeline_id": str(new_timeline_id),
             "ancestor_start_lsn": str(ancestor_start_lsn) if ancestor_start_lsn else None,
             "ancestor_timeline_id": str(ancestor_timeline_id) if ancestor_timeline_id else None,
@@ -504,7 +507,7 @@ class PageserverHttpClient(requests.Session, MetricsGetter):
         include_timeline_dir_layer_file_size_sum: bool = False,
         force_await_initial_logical_size: bool = False,
         **kwargs,
-    ) -> Dict[Any, Any]:
+    ) -> dict[Any, Any]:
         params = {}
         if include_non_incremental_logical_size:
             params["include-non-incremental-logical-size"] = "true"
@@ -844,7 +847,7 @@ class PageserverHttpClient(requests.Session, MetricsGetter):
         )
         if len(res) != 2:
             return None
-        inc, dec = [res[metric] for metric in metrics]
+        inc, dec = (res[metric] for metric in metrics)
         queue_count = int(inc) - int(dec)
         assert queue_count >= 0
         return queue_count
@@ -885,7 +888,7 @@ class PageserverHttpClient(requests.Session, MetricsGetter):
         timeline_id: TimelineId,
         batch_size: int | None = None,
         **kwargs,
-    ) -> Set[TimelineId]:
+    ) -> set[TimelineId]:
         params = {}
         if batch_size is not None:
             params["batch_size"] = batch_size

test_runner/fixtures/pageserver/many_tenants.py

@@ -1,5 +1,7 @@
+from __future__ import annotations
+
 import concurrent.futures
-from typing import Any, Callable, Dict, Tuple
+from typing import TYPE_CHECKING
 
 import fixtures.pageserver.remote_storage
 from fixtures.common_types import TenantId, TimelineId
@@ -10,10 +12,13 @@ from fixtures.neon_fixtures import (
 )
 from fixtures.remote_storage import LocalFsStorage, RemoteStorageKind
 
+if TYPE_CHECKING:
+    from typing import Any, Callable
+
 
 def single_timeline(
     neon_env_builder: NeonEnvBuilder,
-    setup_template: Callable[[NeonEnv], Tuple[TenantId, TimelineId, Dict[str, Any]]],
+    setup_template: Callable[[NeonEnv], tuple[TenantId, TimelineId, dict[str, Any]]],
     ncopies: int,
 ) -> NeonEnv:
     """

test_runner/fixtures/pageserver/remote_storage.py

@@ -1,10 +1,12 @@
+from __future__ import annotations
+
 import concurrent.futures
 import os
 import queue
 import shutil
 import threading
 from pathlib import Path
-from typing import Any, List, Tuple
+from typing import TYPE_CHECKING
 
 from fixtures.common_types import TenantId, TimelineId
 from fixtures.neon_fixtures import NeonEnv
@@ -14,6 +16,9 @@ from fixtures.pageserver.common_types import (
 )
 from fixtures.remote_storage import LocalFsStorage
 
+if TYPE_CHECKING:
+    from typing import Any
+
 
 def duplicate_one_tenant(env: NeonEnv, template_tenant: TenantId, new_tenant: TenantId):
     remote_storage = env.pageserver_remote_storage
@@ -50,13 +55,13 @@ def duplicate_one_tenant(env: NeonEnv, template_tenant: TenantId, new_tenant: Te
     return None
 
 
-def duplicate_tenant(env: NeonEnv, template_tenant: TenantId, ncopies: int) -> List[TenantId]:
+def duplicate_tenant(env: NeonEnv, template_tenant: TenantId, ncopies: int) -> list[TenantId]:
     assert isinstance(env.pageserver_remote_storage, LocalFsStorage)
 
     def work(tenant_id):
         duplicate_one_tenant(env, template_tenant, tenant_id)
 
-    new_tenants: List[TenantId] = [TenantId.generate() for _ in range(0, ncopies)]
+    new_tenants: list[TenantId] = [TenantId.generate() for _ in range(0, ncopies)]
     with concurrent.futures.ThreadPoolExecutor(max_workers=8) as executor:
         executor.map(work, new_tenants)
     return new_tenants
@@ -79,7 +84,7 @@ def local_layer_name_from_remote_name(remote_name: str) -> str:
 
 
 def copy_all_remote_layer_files_to_local_tenant_dir(
-    env: NeonEnv, tenant_timelines: List[Tuple[TenantId, TimelineId]]
+    env: NeonEnv, tenant_timelines: list[tuple[TenantId, TimelineId]]
 ):
     remote_storage = env.pageserver_remote_storage
     assert isinstance(remote_storage, LocalFsStorage)

test_runner/fixtures/pageserver/utils.py

@@ -1,5 +1,7 @@
+from __future__ import annotations
+
 import time
-from typing import Any, Dict, List, Optional, Tuple, Union
+from typing import TYPE_CHECKING
 
 from mypy_boto3_s3.type_defs import (
     DeleteObjectOutputTypeDef,
@@ -14,6 +16,9 @@ from fixtures.pageserver.http import PageserverApiException, PageserverHttpClien
 from fixtures.remote_storage import RemoteStorage, RemoteStorageKind, S3Storage
 from fixtures.utils import wait_until
 
+if TYPE_CHECKING:
+    from typing import Any, Optional, Union
+
 
 def assert_tenant_state(
     pageserver_http: PageserverHttpClient,
@@ -66,7 +71,7 @@ def wait_for_upload(
     )
 
 
-def _tenant_in_expected_state(tenant_info: Dict[str, Any], expected_state: str):
+def _tenant_in_expected_state(tenant_info: dict[str, Any], expected_state: str):
     if tenant_info["state"]["slug"] == expected_state:
         return True
     if tenant_info["state"]["slug"] == "Broken":
@@ -80,7 +85,7 @@ def wait_until_tenant_state(
     expected_state: str,
     iterations: int,
     period: float = 1.0,
-) -> Dict[str, Any]:
+) -> dict[str, Any]:
     """
     Does not use `wait_until` for debugging purposes
     """
@@ -136,7 +141,7 @@ def wait_until_timeline_state(
     expected_state: str,
     iterations: int,
     period: float = 1.0,
-) -> Dict[str, Any]:
+) -> dict[str, Any]:
     """
     Does not use `wait_until` for debugging purposes
     """
@@ -147,7 +152,7 @@ def wait_until_timeline_state(
             if isinstance(timeline["state"], str):
                 if timeline["state"] == expected_state:
                     return timeline
-            elif isinstance(timeline, Dict):
+            elif isinstance(timeline, dict):
                 if timeline["state"].get(expected_state):
                     return timeline
 
@@ -235,7 +240,7 @@ def wait_for_upload_queue_empty(
 
         # this is `started left join finished`; if match, subtracting start from finished, resulting in queue depth
         remaining_labels = ["shard_id", "file_kind", "op_kind"]
-        tl: List[Tuple[Any, float]] = []
+        tl: list[tuple[Any, float]] = []
         for s in started:
             found = False
             for f in finished:
@@ -302,7 +307,7 @@ def assert_prefix_empty(
     assert remote_storage is not None
     response = list_prefix(remote_storage, prefix)
     keys = response["KeyCount"]
-    objects: List[ObjectTypeDef] = response.get("Contents", [])
+    objects: list[ObjectTypeDef] = response.get("Contents", [])
     common_prefixes = response.get("CommonPrefixes", [])
 
     is_mock_s3 = isinstance(remote_storage, S3Storage) and not remote_storage.cleanup
@@ -430,7 +435,7 @@ def enable_remote_storage_versioning(
     return response
 
 
-def many_small_layers_tenant_config() -> Dict[str, Any]:
+def many_small_layers_tenant_config() -> dict[str, Any]:
     """
     Create a new dict to avoid issues with deleting from the global value.
     In python, the global is mutable.

test_runner/fixtures/parametrize.py

@@ -1,5 +1,7 @@
+from __future__ import annotations
+
 import os
-from typing import Any, Dict, Optional
+from typing import TYPE_CHECKING
 
 import allure
 import pytest
@@ -9,6 +11,10 @@ from _pytest.python import Metafunc
 from fixtures.pg_version import PgVersion
 from fixtures.utils import AuxFileStore
 
+if TYPE_CHECKING:
+    from typing import Any, Optional
+
+
 """
 Dynamically parametrize tests by different parameters
 """
@@ -44,7 +50,7 @@ def pageserver_aux_file_policy() -> Optional[AuxFileStore]:
     return None
 
 
-def get_pageserver_default_tenant_config_compaction_algorithm() -> Optional[Dict[str, Any]]:
+def get_pageserver_default_tenant_config_compaction_algorithm() -> Optional[dict[str, Any]]:
     toml_table = os.getenv("PAGESERVER_DEFAULT_TENANT_CONFIG_COMPACTION_ALGORITHM")
     if toml_table is None:
         return None
@@ -54,7 +60,7 @@ def get_pageserver_default_tenant_config_compaction_algorithm() -> Optional[Dict
 
 
 @pytest.fixture(scope="function", autouse=True)
-def pageserver_default_tenant_config_compaction_algorithm() -> Optional[Dict[str, Any]]:
+def pageserver_default_tenant_config_compaction_algorithm() -> Optional[dict[str, Any]]:
     return get_pageserver_default_tenant_config_compaction_algorithm()
 
 

test_runner/fixtures/pg_stats.py

@@ -1,15 +1,16 @@
+from __future__ import annotations
+
 from functools import cached_property
-from typing import List
 
 import pytest
 
 
 class PgStatTable:
     table: str
-    columns: List[str]
+    columns: list[str]
     additional_query: str
 
-    def __init__(self, table: str, columns: List[str], filter_query: str = ""):
+    def __init__(self, table: str, columns: list[str], filter_query: str = ""):
         self.table = table
         self.columns = columns
         self.additional_query = filter_query
@@ -20,7 +21,7 @@ class PgStatTable:
 
 
 @pytest.fixture(scope="function")
-def pg_stats_rw() -> List[PgStatTable]:
+def pg_stats_rw() -> list[PgStatTable]:
     return [
         PgStatTable(
             "pg_stat_database",
@@ -31,7 +32,7 @@ def pg_stats_rw() -> List[PgStatTable]:
 
 
 @pytest.fixture(scope="function")
-def pg_stats_ro() -> List[PgStatTable]:
+def pg_stats_ro() -> list[PgStatTable]:
     return [
         PgStatTable(
             "pg_stat_database", ["tup_returned", "tup_fetched"], "WHERE datname='postgres'"
@@ -40,7 +41,7 @@ def pg_stats_ro() -> List[PgStatTable]:
 
 
 @pytest.fixture(scope="function")
-def pg_stats_wo() -> List[PgStatTable]:
+def pg_stats_wo() -> list[PgStatTable]:
     return [
         PgStatTable(
             "pg_stat_database",
@@ -51,7 +52,7 @@ def pg_stats_wo() -> List[PgStatTable]:
 
 
 @pytest.fixture(scope="function")
-def pg_stats_wal() -> List[PgStatTable]:
+def pg_stats_wal() -> list[PgStatTable]:
     return [
         PgStatTable(
             "pg_stat_wal",

test_runner/fixtures/pg_version.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import enum
 import os
 from typing import Optional
@@ -36,7 +38,7 @@ class PgVersion(str, enum.Enum):
         return f"v{self.value}"
 
     @classmethod
-    def _missing_(cls, value) -> Optional["PgVersion"]:
+    def _missing_(cls, value) -> Optional[PgVersion]:
         known_values = {v.value for _, v in cls.__members__.items()}
 
         # Allow passing version as a string with "v" prefix (e.g. "v14")

test_runner/fixtures/port_distributor.py

@@ -1,10 +1,15 @@
+from __future__ import annotations
+
 import re
 import socket
 from contextlib import closing
-from typing import Dict, Union
+from typing import TYPE_CHECKING
 
 from fixtures.log_helper import log
 
+if TYPE_CHECKING:
+    from typing import Union
+
 
 def can_bind(host: str, port: int) -> bool:
     """
@@ -24,7 +29,7 @@ def can_bind(host: str, port: int) -> bool:
             sock.bind((host, port))
             sock.listen()
             return True
-        except socket.error:
+        except OSError:
             log.info(f"Port {port} is in use, skipping")
             return False
         finally:
@@ -34,7 +39,7 @@ def can_bind(host: str, port: int) -> bool:
 class PortDistributor:
     def __init__(self, base_port: int, port_number: int):
         self.iterator = iter(range(base_port, base_port + port_number))
-        self.port_map: Dict[int, int] = {}
+        self.port_map: dict[int, int] = {}
 
     def get_port(self) -> int:
         for port in self.iterator: