Add timing

The rust compiler or linker (I'm not sure which) emits .debug_pubnames
and .debug_pubtypes sections in the binaries. They are supposed to speed
up launching a debugger, but they're obsolete. They've been superseded
by .debug_names section in more recent DWARF spec and debuggers, and
gdb and lldb just ignores them.

I could not find any way to prevent rustc / ldb / mold from emitting
these sections in the first place. So this commit adds a hack to strip
them off afterwards. This makes the binaries about 30% smaller, with
the downside of adding about 30 s to the build time in CI. That seems
like a good tradeoff, as smaller binaries can speed up other steps.

.github/actions/run-python-test-set/action.yml

@@ -55,22 +55,6 @@ runs:
         name: neon-${{ runner.os }}-${{ inputs.build_type }}-artifact
         path: /tmp/neon
 
-    - name: Download Neon binaries for the previous release
-      if: inputs.build_type != 'remote'
-      uses: ./.github/actions/download
-      with:
-        name: neon-${{ runner.os }}-${{ inputs.build_type }}-artifact
-        path: /tmp/neon-previous
-        prefix: latest
-
-    - name: Download compatibility snapshot for Postgres 14
-      if: inputs.build_type != 'remote'
-      uses: ./.github/actions/download
-      with:
-        name: compatibility-snapshot-${{ inputs.build_type }}-pg14
-        path: /tmp/compatibility_snapshot_pg14
-        prefix: latest
-
     - name: Checkout
       if: inputs.needs_postgres_source == 'true'
       uses: actions/checkout@v3
@@ -89,18 +73,23 @@ runs:
       shell: bash -euxo pipefail {0}
       run: ./scripts/pysync
 
+    - name: Download compatibility snapshot for Postgres 14
+      if: inputs.build_type != 'remote'
+      uses: ./.github/actions/download
+      with:
+        name: compatibility-snapshot-${{ inputs.build_type }}-pg14
+        path: /tmp/compatibility_snapshot_pg14
+        prefix: latest
+
     - name: Run pytest
       env:
         NEON_BIN: /tmp/neon/bin
-        COMPATIBILITY_NEON_BIN: /tmp/neon-previous/bin
-        COMPATIBILITY_POSTGRES_DISTRIB_DIR: /tmp/neon-previous/pg_install
         TEST_OUTPUT: /tmp/test_output
         BUILD_TYPE: ${{ inputs.build_type }}
         AWS_ACCESS_KEY_ID: ${{ inputs.real_s3_access_key_id }}
         AWS_SECRET_ACCESS_KEY: ${{ inputs.real_s3_secret_access_key }}
         COMPATIBILITY_SNAPSHOT_DIR: /tmp/compatibility_snapshot_pg14
-        ALLOW_BACKWARD_COMPATIBILITY_BREAKAGE: contains(github.event.pull_request.labels.*.name, 'backward compatibility breakage')
-        ALLOW_FORWARD_COMPATIBILITY_BREAKAGE: contains(github.event.pull_request.labels.*.name, 'forward compatibility breakage')
+        ALLOW_BREAKING_CHANGES: contains(github.event.pull_request.labels.*.name, 'breaking changes')
       shell: bash -euxo pipefail {0}
       run: |
         # PLATFORM will be embedded in the perf test report
@@ -123,12 +112,7 @@ runs:
           exit 1
         fi
         if [[ "${{ inputs.run_in_parallel }}" == "true" ]]; then
-          # -n4 uses four processes to run tests via pytest-xdist
           EXTRA_PARAMS="-n4 $EXTRA_PARAMS"
-
-          # --dist=loadgroup points tests marked with @pytest.mark.xdist_group
-          # to the same worker to make @pytest.mark.order work with xdist
-          EXTRA_PARAMS="--dist=loadgroup $EXTRA_PARAMS"
         fi
 
         if [[ "${{ inputs.run_with_real_s3 }}" == "true" ]]; then
@@ -163,9 +147,9 @@ runs:
         # --verbose prints name of each test (helpful when there are
         # multiple tests in one file)
         # -rA prints summary in the end
+        # -n4 uses four processes to run tests via pytest-xdist
         # -s is not used to prevent pytest from capturing output, because tests are running
         # in parallel and logs are mixed between different tests
-        #
         mkdir -p $TEST_OUTPUT/allure/results
         "${cov_prefix[@]}" ./scripts/pytest \
           --junitxml=$TEST_OUTPUT/junit.xml \
@@ -185,8 +169,8 @@ runs:
       uses: ./.github/actions/upload
       with:
         name: compatibility-snapshot-${{ inputs.build_type }}-pg14-${{ github.run_id }}
-        # The path includes a test name (test_create_snapshot) and directory that the test creates (compatibility_snapshot_pg14), keep the path in sync with the test
-        path: /tmp/test_output/test_create_snapshot/compatibility_snapshot_pg14/
+        # The path includes a test name (test_prepare_snapshot) and directory that the test creates (compatibility_snapshot_pg14), keep the path in sync with the test
+        path: /tmp/test_output/test_prepare_snapshot/compatibility_snapshot_pg14/
         prefix: latest
 
     - name: Create Allure report

.github/workflows/build_and_test.yml

@@ -168,6 +168,11 @@ jobs:
           ${cov_prefix} cargo test $CARGO_FLAGS
         shell: bash -euxo pipefail {0}
 
+      - name: Slim binaries
+        run: |
+          scripts/strip-useless-debug.py -j$(nproc) target
+        shell: bash -euxo pipefail {0}
+
       - name: Install rust binaries
         run: |
           # Install target binaries
@@ -268,6 +273,32 @@ jobs:
         if: matrix.build_type == 'debug'
         uses: ./.github/actions/save-coverage-data
 
+  upload-latest-artifacts:
+    runs-on: dev
+    container:
+      image: 369495373322.dkr.ecr.eu-central-1.amazonaws.com/rust:pinned
+      options: --init
+    needs: [ regress-tests ]
+    if: github.ref_name == 'main'
+    steps:
+      - name: Copy Neon artifact to the latest directory
+        shell: bash -euxo pipefail {0}
+        env:
+          BUCKET: neon-github-public-dev
+          PREFIX: artifacts/${{ github.run_id }}
+        run: |
+          for build_type in debug release; do
+            FILENAME=neon-${{ runner.os }}-${build_type}-artifact.tar.zst
+
+            S3_KEY=$(aws s3api list-objects-v2 --bucket ${BUCKET} --prefix ${PREFIX} | jq -r '.Contents[].Key' | grep ${FILENAME} | sort --version-sort | tail -1 || true)
+            if [ -z "${S3_KEY}" ]; then
+              echo 2>&1 "Neither s3://${BUCKET}/${PREFIX}/${FILENAME} nor its version from previous attempts exist"
+              exit 1
+            fi
+
+            time aws s3 cp --only-show-errors s3://${BUCKET}/${S3_KEY} s3://${BUCKET}/artifacts/latest/${FILENAME}
+          done
+
   benchmarks:
     runs-on: dev
     container:
@@ -419,15 +450,11 @@ jobs:
     container:
       image: 369495373322.dkr.ecr.eu-central-1.amazonaws.com/base:pinned
       options: --init
-    needs: [ push-docker-hub, tag ]
+    needs: [ build-neon ]
     steps:
       - name: Set PR's status to pending and request a remote CI test
         run: |
-          # For pull requests, GH Actions set "github.sha" variable to point at a fake merge commit
-          # but we need to use a real sha of a latest commit in the PR's branch for the e2e job,
-          # to place a job run status update later.
           COMMIT_SHA=${{ github.event.pull_request.head.sha }}
-          # For non-PR kinds of runs, the above will produce an empty variable, pick the original sha value for those
           COMMIT_SHA=${COMMIT_SHA:-${{ github.sha }}}
 
           REMOTE_REPO="${{ github.repository_owner }}/cloud"
@@ -453,9 +480,7 @@ jobs:
               \"inputs\": {
                 \"ci_job_name\": \"neon-cloud-e2e\",
                 \"commit_hash\": \"$COMMIT_SHA\",
-                \"remote_repo\": \"${{ github.repository }}\",
-                \"storage_image_tag\": \"${{ needs.tag.outputs.build-tag }}\",
-                \"compute_image_tag\": \"${{ needs.tag.outputs.build-tag }}\"
+                \"remote_repo\": \"${{ github.repository }}\"
               }
             }"
 
@@ -916,7 +941,7 @@ jobs:
           DOCKER_TAG=${{needs.tag.outputs.build-tag}}
           helm upgrade neon-proxy-scram neondatabase/neon-proxy --namespace neon-proxy --create-namespace --install -f .github/helm-values/${{ matrix.target_cluster }}.neon-proxy-scram.yaml --set image.tag=${DOCKER_TAG} --wait --timeout 15m0s
 
-  promote-compatibility-data:
+  promote-compatibility-test-snapshot:
     runs-on: dev
     container:
       image: 369495373322.dkr.ecr.eu-central-1.amazonaws.com/rust:pinned
@@ -930,24 +955,9 @@ jobs:
           BUCKET: neon-github-public-dev
           PREFIX: artifacts/latest
         run: |
-          # Update compatibility snapshot for the release
           for build_type in debug release; do
             OLD_FILENAME=compatibility-snapshot-${build_type}-pg14-${GITHUB_RUN_ID}.tar.zst
             NEW_FILENAME=compatibility-snapshot-${build_type}-pg14.tar.zst
 
             time aws s3 mv --only-show-errors s3://${BUCKET}/${PREFIX}/${OLD_FILENAME} s3://${BUCKET}/${PREFIX}/${NEW_FILENAME}
           done
-
-          # Update Neon artifact for the release (reuse already uploaded artifact)
-          for build_type in debug release; do
-            OLD_PREFIX=artifacts/${GITHUB_RUN_ID}
-            FILENAME=neon-${{ runner.os }}-${build_type}-artifact.tar.zst
-
-            S3_KEY=$(aws s3api list-objects-v2 --bucket ${BUCKET} --prefix ${OLD_PREFIX} | jq -r '.Contents[].Key' | grep ${FILENAME} | sort --version-sort | tail -1 || true)
-            if [ -z "${S3_KEY}" ]; then
-              echo 2>&1 "Neither s3://${BUCKET}/${OLD_PREFIX}/${FILENAME} nor its version from previous attempts exist"
-              exit 1
-            fi
-
-            time aws s3 cp --only-show-errors s3://${BUCKET}/${S3_KEY} s3://${BUCKET}/${PREFIX}/${FILENAME}
-          done

.gitmodules

@@ -1,7 +1,7 @@
 [submodule "vendor/postgres-v14"]
 	path = vendor/postgres-v14
 	url = https://github.com/neondatabase/postgres.git
-	branch = REL_14_STABLE_neon
+	branch = main
 [submodule "vendor/postgres-v15"]
 	path = vendor/postgres-v15
 	url = https://github.com/neondatabase/postgres.git

CODEOWNERS

@@ -1,10 +0,0 @@
-/compute_tools/ @neondatabase/control-plane
-/control_plane/ @neondatabase/compute @neondatabase/storage
-/libs/pageserver_api/ @neondatabase/compute @neondatabase/storage
-/libs/postgres_ffi/ @neondatabase/compute 
-/libs/remote_storage/ @neondatabase/storage 
-/libs/safekeeper_api/ @neondatabase/safekeepers  
-/pageserver/ @neondatabase/compute @neondatabase/storage 
-/pgxn/ @neondatabase/compute
-/proxy/ @neondatabase/control-plane 
-/safekeeper/ @neondatabase/safekeepers

compute_tools/src/pg_helpers.rs

@@ -65,7 +65,7 @@ impl GenericOption {
             let name = match self.name.as_str() {
                 "safekeepers" => "neon.safekeepers",
                 "wal_acceptor_reconnect" => "neon.safekeeper_reconnect_timeout",
-                "wal_acceptor_connection_timeout" => "neon.safekeeper_connection_timeout",
+                "wal_acceptor_connect_timeout" => "neon.safekeeper_connect_timeout",
                 it => it,
             };
 

control_plane/src/compute.rs

@@ -287,7 +287,7 @@ impl PostgresNode {
         conf.append("shared_buffers", "1MB");
         conf.append("fsync", "off");
         conf.append("max_connections", "100");
-        conf.append("wal_level", "logical");
+        conf.append("wal_level", "replica");
         // wal_sender_timeout is the maximum time to wait for WAL replication.
         // It also defines how often the walreciever will send a feedback message to the wal sender.
         conf.append("wal_sender_timeout", "5s");

control_plane/src/pageserver.rs

@@ -362,11 +362,6 @@ impl PageServerNode {
                 .map(|x| x.parse::<NonZeroU64>())
                 .transpose()
                 .context("Failed to parse 'max_lsn_wal_lag' as non zero integer")?,
-            trace_read_requests: settings
-                .remove("trace_read_requests")
-                .map(|x| x.parse::<bool>())
-                .transpose()
-                .context("Failed to parse 'trace_read_requests' as bool")?,
         };
         if !settings.is_empty() {
             bail!("Unrecognized tenant settings: {settings:?}")
@@ -429,11 +424,6 @@ impl PageServerNode {
                     .map(|x| x.parse::<NonZeroU64>())
                     .transpose()
                     .context("Failed to parse 'max_lsn_wal_lag' as non zero integer")?,
-                trace_read_requests: settings
-                    .get("trace_read_requests")
-                    .map(|x| x.parse::<bool>())
-                    .transpose()
-                    .context("Failed to parse 'trace_read_requests' as bool")?,
             })
             .send()?
             .error_from_body()?;

libs/pageserver_api/src/models.rs

@@ -73,7 +73,6 @@ pub struct TenantCreateRequest {
     pub walreceiver_connect_timeout: Option<String>,
     pub lagging_wal_timeout: Option<String>,
     pub max_lsn_wal_lag: Option<NonZeroU64>,
-    pub trace_read_requests: Option<bool>,
 }
 
 #[serde_as]
@@ -113,7 +112,6 @@ pub struct TenantConfigRequest {
     pub walreceiver_connect_timeout: Option<String>,
     pub lagging_wal_timeout: Option<String>,
     pub max_lsn_wal_lag: Option<NonZeroU64>,
-    pub trace_read_requests: Option<bool>,
 }
 
 impl TenantConfigRequest {
@@ -132,7 +130,6 @@ impl TenantConfigRequest {
             walreceiver_connect_timeout: None,
             lagging_wal_timeout: None,
             max_lsn_wal_lag: None,
-            trace_read_requests: None,
         }
     }
 }

libs/utils/src/id.rs

@@ -204,17 +204,6 @@ pub struct TenantId(Id);
 
 id_newtype!(TenantId);
 
-/// Neon Connection Id identifies long-lived connections (for example a pagestream
-/// connection with the page_service). Is used for better logging and tracing
-///
-/// NOTE: It (de)serializes as an array of hex bytes, so the string representation would look
-/// like `[173,80,132,115,129,226,72,254,170,201,135,108,199,26,228,24]`.
-/// See [`Id`] for alternative ways to serialize it.
-#[derive(Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize, PartialOrd, Ord)]
-pub struct ConnectionId(Id);
-
-id_newtype!(ConnectionId);
-
 // A pair uniquely identifying Neon instance.
 #[derive(Debug, Clone, Copy, PartialOrd, Ord, PartialEq, Eq, Hash, Serialize, Deserialize)]
 pub struct TenantTimelineId {

pageserver/src/config.rs

@@ -8,7 +8,6 @@ use anyhow::{anyhow, bail, ensure, Context, Result};
 use remote_storage::RemoteStorageConfig;
 use std::env;
 use utils::crashsafe::path_with_suffix_extension;
-use utils::id::ConnectionId;
 
 use std::num::NonZeroUsize;
 use std::path::{Path, PathBuf};
@@ -415,22 +414,6 @@ impl PageServerConf {
         )
     }
 
-    pub fn traces_path(&self) -> PathBuf {
-        self.workdir.join("traces")
-    }
-
-    pub fn trace_path(
-        &self,
-        tenant_id: &TenantId,
-        timeline_id: &TimelineId,
-        connection_id: &ConnectionId,
-    ) -> PathBuf {
-        self.traces_path()
-            .join(tenant_id.to_string())
-            .join(timeline_id.to_string())
-            .join(connection_id.to_string())
-    }
-
     /// Points to a place in pageserver's local directory,
     /// where certain timeline's metadata file should be located.
     pub fn metadata_path(&self, timeline_id: TimelineId, tenant_id: TenantId) -> PathBuf {

pageserver/src/http/routes.rs

@@ -618,7 +618,6 @@ async fn tenant_create_handler(mut request: Request<Body>) -> Result<Response<Bo
     check_permission(&request, None)?;
 
     let request_data: TenantCreateRequest = json_request(&mut request).await?;
-    println!("tenant create: {:?}", request_data.trace_read_requests);
     let remote_index = get_state(&request).remote_index.clone();
 
     let mut tenant_conf = TenantConfOpt::default();
@@ -660,9 +659,6 @@ async fn tenant_create_handler(mut request: Request<Body>) -> Result<Response<Bo
     if let Some(max_lsn_wal_lag) = request_data.max_lsn_wal_lag {
         tenant_conf.max_lsn_wal_lag = Some(max_lsn_wal_lag);
     }
-    if let Some(trace_read_requests) = request_data.trace_read_requests {
-        tenant_conf.trace_read_requests = Some(trace_read_requests);
-    }
 
     tenant_conf.checkpoint_distance = request_data.checkpoint_distance;
     if let Some(checkpoint_timeout) = request_data.checkpoint_timeout {
@@ -750,9 +746,6 @@ async fn tenant_config_handler(mut request: Request<Body>) -> Result<Response<Bo
     if let Some(max_lsn_wal_lag) = request_data.max_lsn_wal_lag {
         tenant_conf.max_lsn_wal_lag = Some(max_lsn_wal_lag);
     }
-    if let Some(trace_read_requests) = request_data.trace_read_requests {
-        tenant_conf.trace_read_requests = Some(trace_read_requests);
-    }
 
     tenant_conf.checkpoint_distance = request_data.checkpoint_distance;
     if let Some(checkpoint_timeout) = request_data.checkpoint_timeout {

pageserver/src/lib.rs

@@ -15,7 +15,6 @@ pub mod tenant;
 pub mod tenant_config;
 pub mod tenant_mgr;
 pub mod tenant_tasks;
-pub mod trace;
 pub mod virtual_file;
 pub mod walingest;
 pub mod walreceiver;

pageserver/src/page_service.rs

@@ -29,7 +29,6 @@ use tokio::pin;
 use tokio_util::io::StreamReader;
 use tokio_util::io::SyncIoBridge;
 use tracing::*;
-use utils::id::ConnectionId;
 use utils::{
     auth::{self, Claims, JwtAuth, Scope},
     id::{TenantId, TimelineId},
@@ -48,7 +47,6 @@ use crate::task_mgr;
 use crate::task_mgr::TaskKind;
 use crate::tenant::Timeline;
 use crate::tenant_mgr;
-use crate::trace::Tracer;
 use crate::CheckpointConfig;
 
 use postgres_ffi::pg_constants::DEFAULTTABLESPACE_OID;
@@ -271,18 +269,6 @@ impl PageServerHandler {
         //       so there is no need to reset the association
         task_mgr::associate_with(Some(tenant_id), Some(timeline_id));
 
-        // Make request tracer if needed
-        let tenant = tenant_mgr::get_tenant(tenant_id, true)?;
-        let mut tracer = if tenant.get_trace_read_requests() {
-            let connection_id = ConnectionId::generate();
-            let path = tenant
-                .conf
-                .trace_path(&tenant_id, &timeline_id, &connection_id);
-            Some(Tracer::new(path))
-        } else {
-            None
-        };
-
         // Check that the timeline exists
         let timeline = get_local_timeline(tenant_id, timeline_id)?;
 
@@ -315,11 +301,6 @@ impl PageServerHandler {
 
             trace!("query: {copy_data_bytes:?}");
 
-            // Trace request if needed
-            if let Some(t) = tracer.as_mut() {
-                t.trace(&copy_data_bytes)
-            }
-
             let neon_fe_msg = PagestreamFeMessage::parse(&mut copy_data_bytes.reader())?;
 
             let response = match neon_fe_msg {

pageserver/src/tenant.rs

@@ -806,13 +806,6 @@ impl Tenant {
             .unwrap_or(self.conf.default_tenant_conf.pitr_interval)
     }
 
-    pub fn get_trace_read_requests(&self) -> bool {
-        let tenant_conf = self.tenant_conf.read().unwrap();
-        tenant_conf
-            .trace_read_requests
-            .unwrap_or(self.conf.default_tenant_conf.trace_read_requests)
-    }
-
     pub fn update_tenant_config(&self, new_tenant_conf: TenantConfOpt) {
         self.tenant_conf.write().unwrap().update(&new_tenant_conf);
     }
@@ -1673,7 +1666,6 @@ pub mod harness {
                 walreceiver_connect_timeout: Some(tenant_conf.walreceiver_connect_timeout),
                 lagging_wal_timeout: Some(tenant_conf.lagging_wal_timeout),
                 max_lsn_wal_lag: Some(tenant_conf.max_lsn_wal_lag),
-                trace_read_requests: Some(tenant_conf.trace_read_requests),
             }
         }
     }

pageserver/src/tenant_config.rs

@@ -82,7 +82,6 @@ pub struct TenantConf {
     /// A lagging safekeeper will be changed after `lagging_wal_timeout` time elapses since the last WAL update,
     /// to avoid eager reconnects.
     pub max_lsn_wal_lag: NonZeroU64,
-    pub trace_read_requests: bool,
 }
 
 /// Same as TenantConf, but this struct preserves the information about
@@ -106,7 +105,6 @@ pub struct TenantConfOpt {
     #[serde(with = "humantime_serde")]
     pub lagging_wal_timeout: Option<Duration>,
     pub max_lsn_wal_lag: Option<NonZeroU64>,
-    pub trace_read_requests: Option<bool>,
 }
 
 impl TenantConfOpt {
@@ -140,9 +138,6 @@ impl TenantConfOpt {
                 .lagging_wal_timeout
                 .unwrap_or(global_conf.lagging_wal_timeout),
             max_lsn_wal_lag: self.max_lsn_wal_lag.unwrap_or(global_conf.max_lsn_wal_lag),
-            trace_read_requests: self
-                .trace_read_requests
-                .unwrap_or(global_conf.trace_read_requests),
         }
     }
 
@@ -212,7 +207,6 @@ impl TenantConf {
                 .expect("cannot parse default walreceiver lagging wal timeout"),
             max_lsn_wal_lag: NonZeroU64::new(DEFAULT_MAX_WALRECEIVER_LSN_WAL_LAG)
                 .expect("cannot parse default max walreceiver Lsn wal lag"),
-            trace_read_requests: false,
         }
     }
 
@@ -238,7 +232,6 @@ impl TenantConf {
             .unwrap(),
             max_lsn_wal_lag: NonZeroU64::new(defaults::DEFAULT_MAX_WALRECEIVER_LSN_WAL_LAG)
                 .unwrap(),
-            trace_read_requests: false,
         }
     }
 }

pageserver/src/trace.rs

@@ -1,36 +0,0 @@
-use bytes::Bytes;
-use std::{
-    fs::{create_dir_all, File},
-    io::{BufWriter, Write},
-    path::PathBuf,
-};
-
-pub struct Tracer {
-    writer: BufWriter<File>,
-}
-
-impl Drop for Tracer {
-    fn drop(&mut self) {
-        self.flush()
-    }
-}
-
-impl Tracer {
-    pub fn new(path: PathBuf) -> Self {
-        let parent = path.parent().expect("failed to parse parent path");
-        create_dir_all(parent).expect("failed to create trace dir");
-
-        let file = File::create(path).expect("failed to create trace file");
-        Tracer {
-            writer: BufWriter::new(file),
-        }
-    }
-
-    pub fn trace(&mut self, msg: &Bytes) {
-        self.writer.write_all(msg).expect("failed to write trace");
-    }
-
-    pub fn flush(&mut self) {
-        self.writer.flush().expect("failed to flush trace file");
-    }
-}

pageserver/src/walredo.rs

@@ -22,10 +22,10 @@ use byteorder::{ByteOrder, LittleEndian};
 use bytes::{BufMut, Bytes, BytesMut};
 use nix::poll::*;
 use serde::Serialize;
+use std::fs;
 use std::fs::OpenOptions;
 use std::io::prelude::*;
 use std::io::{Error, ErrorKind};
-use std::ops::{Deref, DerefMut};
 use std::os::unix::io::AsRawFd;
 use std::os::unix::prelude::CommandExt;
 use std::path::PathBuf;
@@ -34,7 +34,6 @@ use std::process::{Child, ChildStderr, ChildStdin, ChildStdout, Command};
 use std::sync::Mutex;
 use std::time::Duration;
 use std::time::Instant;
-use std::{fs, io};
 use tracing::*;
 use utils::crashsafe::path_with_suffix_extension;
 use utils::{bin_ser::BeSer, id::TenantId, lsn::Lsn, nonblock::set_nonblock};
@@ -45,7 +44,6 @@ use crate::metrics::{
 };
 use crate::pgdatadir_mapping::{key_to_rel_block, key_to_slru_block};
 use crate::repository::Key;
-use crate::task_mgr::BACKGROUND_RUNTIME;
 use crate::walrecord::NeonWalRecord;
 use crate::{config::PageServerConf, TEMP_FILE_SUFFIX};
 use pageserver_api::reltag::{RelTag, SlruKind};
@@ -231,7 +229,7 @@ impl PostgresRedoManager {
 
         // launch the WAL redo process on first use
         if process_guard.is_none() {
-            let p = PostgresRedoProcess::launch(self.conf, self.tenant_id, pg_version)?;
+            let p = PostgresRedoProcess::launch(self.conf, &self.tenant_id, pg_version)?;
             *process_guard = Some(p);
         }
         let process = process_guard.as_mut().unwrap();
@@ -581,8 +579,7 @@ impl<C: CommandExt> CloseFileDescriptors for C {
 /// Handle to the Postgres WAL redo process
 ///
 struct PostgresRedoProcess {
-    tenant_id: TenantId,
-    child: NoLeakChild,
+    child: Child,
     stdin: ChildStdin,
     stdout: ChildStdout,
     stderr: ChildStderr,
@@ -592,17 +589,16 @@ impl PostgresRedoProcess {
     //
     // Start postgres binary in special WAL redo mode.
     //
-    #[instrument(skip_all,fields(tenant_id=%tenant_id, pg_version=pg_version))]
     fn launch(
         conf: &PageServerConf,
-        tenant_id: TenantId,
+        tenant_id: &TenantId,
         pg_version: u32,
     ) -> Result<PostgresRedoProcess, Error> {
         // FIXME: We need a dummy Postgres cluster to run the process in. Currently, we
         // just create one with constant name. That fails if you try to launch more than
         // one WAL redo manager concurrently.
         let datadir = path_with_suffix_extension(
-            conf.tenant_path(&tenant_id).join("wal-redo-datadir"),
+            conf.tenant_path(tenant_id).join("wal-redo-datadir"),
             TEMP_FILE_SUFFIX,
         );
 
@@ -657,7 +653,7 @@ impl PostgresRedoProcess {
         }
 
         // Start postgres itself
-        let child = Command::new(pg_bin_dir_path.join("postgres"))
+        let mut child = Command::new(pg_bin_dir_path.join("postgres"))
             .arg("--wal-redo")
             .stdin(Stdio::piped())
             .stderr(Stdio::piped())
@@ -676,7 +672,7 @@ impl PostgresRedoProcess {
             // as close-on-exec by default, but that's not enough, since we use
             // libraries that directly call libc open without setting that flag.
             .close_fds()
-            .spawn_no_leak_child()
+            .spawn()
             .map_err(|e| {
                 Error::new(
                     e.kind(),
@@ -684,33 +680,20 @@ impl PostgresRedoProcess {
                 )
             })?;
 
-        let mut child = scopeguard::guard(child, |child| {
-            error!("killing wal-redo-postgres process due to a problem during launch");
-            child.kill_and_wait();
-        });
+        info!(
+            "launched WAL redo postgres process on {}",
+            datadir.display()
+        );
 
         let stdin = child.stdin.take().unwrap();
         let stdout = child.stdout.take().unwrap();
         let stderr = child.stderr.take().unwrap();
 
-        macro_rules! set_nonblock_or_log_err {
-            ($file:ident) => {{
-                let res = set_nonblock($file.as_raw_fd());
-                if let Err(e) = &res {
-                    error!(error = %e, file = stringify!($file), pid = child.id(), "set_nonblock failed");
-                }
-                res
-            }};
-        }
-        set_nonblock_or_log_err!(stdin)?;
-        set_nonblock_or_log_err!(stdout)?;
-        set_nonblock_or_log_err!(stderr)?;
-
-        // all fallible operations post-spawn are complete, so get rid of the guard
-        let child = scopeguard::ScopeGuard::into_inner(child);
+        set_nonblock(stdin.as_raw_fd())?;
+        set_nonblock(stdout.as_raw_fd())?;
+        set_nonblock(stderr.as_raw_fd())?;
 
         Ok(PostgresRedoProcess {
-            tenant_id,
             child,
             stdin,
             stdout,
@@ -718,16 +701,18 @@ impl PostgresRedoProcess {
         })
     }
 
-    #[instrument(skip_all, fields(tenant_id=%self.tenant_id, pid=%self.child.id()))]
-    fn kill(self) {
-        self.child.kill_and_wait();
+    fn kill(mut self) {
+        let _ = self.child.kill();
+        if let Ok(exit_status) = self.child.wait() {
+            error!("wal-redo-postgres exited with code {}", exit_status);
+        }
+        drop(self);
     }
 
     //
     // Apply given WAL records ('records') over an old page image. Returns
     // new page image.
     //
-    #[instrument(skip_all, fields(tenant_id=%self.tenant_id, pid=%self.child.id()))]
     fn apply_wal_records(
         &mut self,
         tag: BufferTag,
@@ -853,99 +838,6 @@ impl PostgresRedoProcess {
     }
 }
 
-/// Wrapper type around `std::process::Child` which guarantees that the child
-/// will be killed and waited-for by this process before being dropped.
-struct NoLeakChild {
-    child: Option<Child>,
-}
-
-impl Deref for NoLeakChild {
-    type Target = Child;
-
-    fn deref(&self) -> &Self::Target {
-        self.child.as_ref().expect("must not use from drop")
-    }
-}
-
-impl DerefMut for NoLeakChild {
-    fn deref_mut(&mut self) -> &mut Self::Target {
-        self.child.as_mut().expect("must not use from drop")
-    }
-}
-
-impl NoLeakChild {
-    fn spawn(command: &mut Command) -> io::Result<Self> {
-        let child = command.spawn()?;
-        Ok(NoLeakChild { child: Some(child) })
-    }
-
-    fn kill_and_wait(mut self) {
-        let child = match self.child.take() {
-            Some(child) => child,
-            None => return,
-        };
-        Self::kill_and_wait_impl(child);
-    }
-
-    #[instrument(skip_all, fields(pid=child.id()))]
-    fn kill_and_wait_impl(mut child: Child) {
-        let res = child.kill();
-        if let Err(e) = res {
-            // This branch is very unlikely because:
-            // - We (= pageserver) spawned this process successfully, so, we're allowed to kill it.
-            // - This is the only place that calls .kill()
-            // - We consume `self`, so, .kill() can't be called twice.
-            // - If the process exited by itself or was killed by someone else,
-            //   .kill() will still succeed because we haven't wait()'ed yet.
-            //
-            // So, if we arrive here, we have really no idea what happened,
-            // whether the PID stored in self.child is still valid, etc.
-            // If this function were fallible, we'd return an error, but
-            // since it isn't, all we can do is log an error and proceed
-            // with the wait().
-            error!(error = %e, "failed to SIGKILL; subsequent wait() might fail or wait for wrong process");
-        }
-
-        match child.wait() {
-            Ok(exit_status) => {
-                // log at error level since .kill() is something we only do on errors ATM
-                error!(exit_status = %exit_status, "wait successful");
-            }
-            Err(e) => {
-                error!(error = %e, "wait error; might leak the child process; it will show as zombie (defunct)");
-            }
-        }
-    }
-}
-
-impl Drop for NoLeakChild {
-    fn drop(&mut self) {
-        let child = match self.child.take() {
-            Some(child) => child,
-            None => return,
-        };
-        // Offload the kill+wait of the child process into the background.
-        // If someone stops the runtime, we'll leak the child process.
-        // We can ignore that case because we only stop the runtime on pageserver exit.
-        BACKGROUND_RUNTIME.spawn(async move {
-            tokio::task::spawn_blocking(move || {
-                Self::kill_and_wait_impl(child);
-            })
-            .await
-        });
-    }
-}
-
-trait NoLeakChildCommandExt {
-    fn spawn_no_leak_child(&mut self) -> io::Result<NoLeakChild>;
-}
-
-impl NoLeakChildCommandExt for Command {
-    fn spawn_no_leak_child(&mut self) -> io::Result<NoLeakChild> {
-        NoLeakChild::spawn(self)
-    }
-}
-
 // Functions for constructing messages to send to the postgres WAL redo
 // process. See pgxn/neon_walredo/walredoproc.c for
 // explanation of the protocol.

pgxn/neon/libpagestore.c

@@ -40,21 +40,8 @@
 bool		connected = false;
 PGconn	   *pageserver_conn = NULL;
 
-/*
- * WaitEventSet containing:
- * - WL_SOCKET_READABLE on pageserver_conn,
- * - WL_LATCH_SET on MyLatch, and
- * - WL_EXIT_ON_PM_DEATH.
- */
-WaitEventSet *pageserver_conn_wes = NULL;
-
 char	   *page_server_connstring_raw;
 
-int			n_unflushed_requests = 0;
-int			flush_every_n_requests = 8;
-
-static void pageserver_flush(void);
-
 static void
 pageserver_connect()
 {
@@ -71,7 +58,6 @@ pageserver_connect()
 
 		PQfinish(pageserver_conn);
 		pageserver_conn = NULL;
-
 		ereport(ERROR,
 				(errcode(ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION),
 				 errmsg(NEON_TAG "could not establish connection to pageserver"),
@@ -87,26 +73,22 @@ pageserver_connect()
 		neon_log(ERROR, "could not send pagestream command to pageserver");
 	}
 
-	pageserver_conn_wes = CreateWaitEventSet(TopMemoryContext, 3);
-	AddWaitEventToSet(pageserver_conn_wes, WL_LATCH_SET, PGINVALID_SOCKET,
-			  MyLatch, NULL);
-	AddWaitEventToSet(pageserver_conn_wes, WL_EXIT_ON_PM_DEATH, PGINVALID_SOCKET,
-			  NULL, NULL);
-	AddWaitEventToSet(pageserver_conn_wes, WL_SOCKET_READABLE, PQsocket(pageserver_conn), NULL, NULL);
-
 	while (PQisBusy(pageserver_conn))
 	{
 		int			wc;
-		WaitEvent	event;
 
 		/* Sleep until there's something to do */
-		wc = WaitEventSetWait(pageserver_conn_wes, -1L, &event, 1, PG_WAIT_EXTENSION);
+		wc = WaitLatchOrSocket(MyLatch,
+							   WL_LATCH_SET | WL_SOCKET_READABLE |
+							   WL_EXIT_ON_PM_DEATH,
+							   PQsocket(pageserver_conn),
+							   -1L, PG_WAIT_EXTENSION);
 		ResetLatch(MyLatch);
 
 		CHECK_FOR_INTERRUPTS();
 
 		/* Data available in socket? */
-		if (event.events & WL_SOCKET_READABLE)
+		if (wc & WL_SOCKET_READABLE)
 		{
 			if (!PQconsumeInput(pageserver_conn))
 			{
@@ -114,7 +96,6 @@ pageserver_connect()
 
 				PQfinish(pageserver_conn);
 				pageserver_conn = NULL;
-				FreeWaitEventSet(pageserver_conn_wes);
 
 				neon_log(ERROR, "could not complete handshake with pageserver: %s",
 						 msg);
@@ -131,30 +112,33 @@ pageserver_connect()
  * A wrapper around PQgetCopyData that checks for interrupts while sleeping.
  */
 static int
-call_PQgetCopyData(char **buffer)
+call_PQgetCopyData(PGconn *conn, char **buffer)
 {
 	int			ret;
 
 retry:
-	ret = PQgetCopyData(pageserver_conn, buffer, 1 /* async */ );
+	ret = PQgetCopyData(conn, buffer, 1 /* async */ );
 
 	if (ret == 0)
 	{
 		int			wc;
-		WaitEvent	event;
 
 		/* Sleep until there's something to do */
-		wc = WaitEventSetWait(pageserver_conn_wes, -1L, &event, 1, PG_WAIT_EXTENSION);
+		wc = WaitLatchOrSocket(MyLatch,
+							   WL_LATCH_SET | WL_SOCKET_READABLE |
+							   WL_EXIT_ON_PM_DEATH,
+							   PQsocket(conn),
+							   -1L, PG_WAIT_EXTENSION);
 		ResetLatch(MyLatch);
 
 		CHECK_FOR_INTERRUPTS();
 
 		/* Data available in socket? */
-		if (event.events & WL_SOCKET_READABLE)
+		if (wc & WL_SOCKET_READABLE)
 		{
-			if (!PQconsumeInput(pageserver_conn))
+			if (!PQconsumeInput(conn))
 				neon_log(ERROR, "could not get response from pageserver: %s",
-						 PQerrorMessage(pageserver_conn));
+						 PQerrorMessage(conn));
 		}
 
 		goto retry;
@@ -180,11 +164,7 @@ pageserver_disconnect(void)
 		PQfinish(pageserver_conn);
 		pageserver_conn = NULL;
 		connected = false;
-
-		prefetch_on_ps_disconnect();
 	}
-	if (pageserver_conn_wes != NULL)
-		FreeWaitEventSet(pageserver_conn_wes);
 }
 
 static void
@@ -194,7 +174,11 @@ pageserver_send(NeonRequest * request)
 
 	/* If the connection was lost for some reason, reconnect */
 	if (connected && PQstatus(pageserver_conn) == CONNECTION_BAD)
-		pageserver_disconnect();
+	{
+		PQfinish(pageserver_conn);
+		pageserver_conn = NULL;
+		connected = false;
+	}
 
 	if (!connected)
 		pageserver_connect();
@@ -218,11 +202,6 @@ pageserver_send(NeonRequest * request)
 	}
 	pfree(req_buff.data);
 
-	n_unflushed_requests++;
-
-	if (flush_every_n_requests > 0 && n_unflushed_requests >= flush_every_n_requests)
-		pageserver_flush();
-
 	if (message_level_is_interesting(PageStoreTrace))
 	{
 		char	   *msg = nm_to_string((NeonMessage *) request);
@@ -241,7 +220,7 @@ pageserver_receive(void)
 	PG_TRY();
 	{
 		/* read response */
-		resp_buff.len = call_PQgetCopyData(&resp_buff.data);
+		resp_buff.len = call_PQgetCopyData(pageserver_conn, &resp_buff.data);
 		resp_buff.cursor = 0;
 
 		if (resp_buff.len < 0)
@@ -276,21 +255,25 @@ pageserver_receive(void)
 static void
 pageserver_flush(void)
 {
-	if (!connected)
-	{
-		neon_log(WARNING, "Tried to flush while disconnected");
-	}
-	else if (PQflush(pageserver_conn))
+	if (PQflush(pageserver_conn))
 	{
 		char	   *msg = PQerrorMessage(pageserver_conn);
 
 		pageserver_disconnect();
 		neon_log(ERROR, "failed to flush page requests: %s", msg);
 	}
-	n_unflushed_requests = 0;
+}
+
+static NeonResponse *
+pageserver_call(NeonRequest * request)
+{
+	pageserver_send(request);
+	pageserver_flush();
+	return pageserver_receive();
 }
 
 page_server_api api = {
+	.request = pageserver_call,
 	.send = pageserver_send,
 	.flush = pageserver_flush,
 	.receive = pageserver_receive
@@ -444,14 +427,6 @@ pg_init_libpagestore(void)
 							PGC_SIGHUP,
 							GUC_UNIT_MB,
 							NULL, NULL, NULL);
-	DefineCustomIntVariable("neon.flush_output_after",
-							"Flush the output buffer after every N unflushed requests",
-							NULL,
-							&flush_every_n_requests,
-							8, -1, INT_MAX,
-							PGC_SIGHUP,
-							0,	/* no flags required */
-							NULL, NULL, NULL);
 
 	relsize_hash_init();
 

pgxn/neon/pagestore_client.h

@@ -115,8 +115,6 @@ typedef struct
 	char		page[FLEXIBLE_ARRAY_MEMBER];
 }			NeonGetPageResponse;
 
-#define PS_GETPAGERESPONSE_SIZE (MAXALIGN(offsetof(NeonGetPageResponse, page) + BLCKSZ))
-
 typedef struct
 {
 	NeonMessageTag tag;
@@ -140,18 +138,15 @@ extern char *nm_to_string(NeonMessage * msg);
 
 typedef struct
 {
+	NeonResponse *(*request) (NeonRequest * request);
 	void		(*send) (NeonRequest * request);
 	NeonResponse *(*receive) (void);
 	void		(*flush) (void);
 }			page_server_api;
 
-extern void prefetch_on_ps_disconnect(void);
-
 extern page_server_api * page_server;
 
 extern char *page_server_connstring;
-extern bool seqscan_prefetch_enabled;
-extern int seqscan_prefetch_distance;
 extern char *neon_timeline;
 extern char *neon_tenant;
 extern bool wal_redo;
@@ -172,6 +167,7 @@ extern void neon_extend(SMgrRelation reln, ForkNumber forknum,
 						BlockNumber blocknum, char *buffer, bool skipFsync);
 extern bool neon_prefetch(SMgrRelation reln, ForkNumber forknum,
 						  BlockNumber blocknum);
+extern void neon_reset_prefetch(SMgrRelation reln);
 extern void neon_read(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
 					  char *buffer);
 

pgxn/neon/pagestore_smgr.c

@@ -49,20 +49,22 @@
 #include "access/xlog.h"
 #include "access/xloginsert.h"
 #include "access/xlog_internal.h"
-#include "access/xlogdefs.h"
 #include "catalog/pg_class.h"
-#include "common/hashfn.h"
 #include "pagestore_client.h"
+#include "pagestore_client.h"
+#include "storage/smgr.h"
+#include "access/xlogdefs.h"
 #include "postmaster/interrupt.h"
-#include "postmaster/autovacuum.h"
 #include "replication/walsender.h"
 #include "storage/bufmgr.h"
 #include "storage/relfilenode.h"
 #include "storage/buf_internals.h"
-#include "storage/smgr.h"
 #include "storage/md.h"
+#include "fmgr.h"
+#include "miscadmin.h"
 #include "pgstat.h"
-
+#include "catalog/pg_tablespace_d.h"
+#include "postmaster/autovacuum.h"
 
 #if PG_VERSION_NUM >= 150000
 #include "access/xlogutils.h"
@@ -111,482 +113,48 @@ typedef enum
 static SMgrRelation unlogged_build_rel = NULL;
 static UnloggedBuildPhase unlogged_build_phase = UNLOGGED_BUILD_NOT_IN_PROGRESS;
 
+
 /*
  * Prefetch implementation:
- * 
  * Prefetch is performed locally by each backend.
- *
- * There can be up to READ_BUFFER_SIZE active IO requests registered at any
- * time. Requests using smgr_prefetch are sent to the pageserver, but we don't
- * wait on the response. Requests using smgr_read are either read from the
- * buffer, or (if that's not possible) we wait on the response to arrive -
- * this also will allow us to receive other prefetched pages. 
- * Each request is immediately written to the output buffer of the pageserver
- * connection, but may not be flushed if smgr_prefetch is used: pageserver
- * flushes sent requests on manual flush, or every neon.flush_output_after
- * unflushed requests; which is not necessarily always and all the time.
- *
- * Once we have received a response, this value will be stored in the response
- * buffer, indexed in a hash table. This allows us to retain our buffered
- * prefetch responses even when we have cache misses.
- *
- * Reading of prefetch responses is delayed until them are actually needed
- * (smgr_read). In case of prefetch miss or any other SMGR request other than
- * smgr_read, all prefetch responses in the pipeline will need to be read from
- * the connection; the responses are stored for later use.
- *
- * NOTE: The current implementation of the prefetch system implements a ring
- * buffer of up to READ_BUFFER_SIZE requests. If there are more _read and
- * _prefetch requests between the initial _prefetch and the _read of a buffer,
- * the prefetch request will have been dropped from this prefetch buffer, and
- * your prefetch was wasted.
+ * There can be up to MAX_PREFETCH_REQUESTS registered using smgr_prefetch
+ * before smgr_read. All this requests are appended to primary smgr_read request.
+ * It is assumed that pages will be requested in prefetch order.
+ * Reading of prefetch responses is delayed until them are actually needed (smgr_read).
+ * It make it possible to parallelize processing and receiving of prefetched pages.
+ * In case of prefetch miss or any other SMGR request other than smgr_read,
+ * all prefetch responses has to be consumed.
  */
 
-/* Max amount of tracked buffer reads */
-#define READ_BUFFER_SIZE 128
+#define MAX_PREFETCH_REQUESTS 128
 
-typedef enum PrefetchStatus {
-	PRFS_UNUSED = 0,	/* unused slot */
-	PRFS_REQUESTED,		/* request was written to the sendbuffer to PS, but not
-						 * necessarily flushed.
-						 * all fields except response valid */
-	PRFS_RECEIVED,		/* all fields valid */
-	PRFS_TAG_REMAINS,	/* only buftag and my_ring_index are still valid */
-} PrefetchStatus;
+BufferTag	prefetch_requests[MAX_PREFETCH_REQUESTS];
+BufferTag	prefetch_responses[MAX_PREFETCH_REQUESTS];
+int			n_prefetch_requests;
+int			n_prefetch_responses;
+int			n_prefetched_buffers;
+int			n_prefetch_hits;
+int			n_prefetch_misses;
+XLogRecPtr	prefetch_lsn;
 
-typedef struct PrefetchRequest {
-	BufferTag	buftag; /* must be first entry in the struct */
-	XLogRecPtr	effective_request_lsn;
-	NeonResponse *response; /* may be null */
-	PrefetchStatus status;
-	uint64		my_ring_index;
-} PrefetchRequest;
-
-/* prefetch buffer lookup hash table */
-
-typedef struct PrfHashEntry {
-	PrefetchRequest *slot;
-	uint32 status;
-	uint32 hash;
-} PrfHashEntry;
-
-#define SH_PREFIX			prfh
-#define SH_ELEMENT_TYPE		PrfHashEntry
-#define SH_KEY_TYPE			PrefetchRequest *
-#define SH_KEY				slot
-#define SH_STORE_HASH
-#define SH_GET_HASH(tb, a)	((a)->hash)
-#define SH_HASH_KEY(tb, key) hash_bytes( \
-	((const unsigned char *) &(key)->buftag), \
-	sizeof(BufferTag) \
-)
-
-#define SH_EQUAL(tb, a, b)	(BUFFERTAGS_EQUAL((a)->buftag, (b)->buftag))
-#define SH_SCOPE			static inline
-#define SH_DEFINE
-#define SH_DECLARE
-#include "lib/simplehash.h"
-
-/*
- * PrefetchState maintains the state of (prefetch) getPage@LSN requests.
- * It maintains a (ring) buffer of in-flight requests and responses.
- * 
- * We maintain several indexes into the ring buffer:
- * ring_unused >= ring_receive >= ring_last >= 0
- * 
- * ring_unused points to the first unused slot of the buffer
- * ring_receive is the next request that is to be received
- * ring_last is the oldest received entry in the buffer
- * 
- * Apart from being an entry in the ring buffer of prefetch requests, each
- * PrefetchRequest that is not UNUSED is indexed in prf_hash by buftag.
- */
-typedef struct PrefetchState {
-	MemoryContext bufctx; /* context for prf_buffer[].response allocations */
-	MemoryContext errctx; /* context for prf_buffer[].response allocations */
-	MemoryContext hashctx; /* context for prf_buffer */
-
-	/* buffer indexes */
-	uint64	ring_unused;		/* first unused slot */
-	uint64	ring_receive;		/* next slot that is to receive a response */
-	uint64	ring_last;			/* min slot with a response value */
-
-	/* metrics / statistics  */
-	int		n_responses_buffered;	/* count of PS responses not yet in buffers */
-	int		n_requests_inflight;	/* count of PS requests considered in flight */
-	int		n_unused;				/* count of buffers < unused, > last, that are also unused */
-
-	/* the buffers */
-	prfh_hash *prf_hash;
-	PrefetchRequest prf_buffer[READ_BUFFER_SIZE]; /* prefetch buffers */
-} PrefetchState;
-
-PrefetchState *MyPState;
-
-int			n_prefetch_hits = 0;
-int			n_prefetch_misses = 0;
-int			n_prefetch_missed_caches = 0;
-int			n_prefetch_dupes = 0;
-
-XLogRecPtr	prefetch_lsn = 0;
-
-static void consume_prefetch_responses(void);
-static uint64 prefetch_register_buffer(BufferTag tag, bool *force_latest, XLogRecPtr *force_lsn);
-static void prefetch_read(PrefetchRequest *slot);
-static void prefetch_do_request(PrefetchRequest *slot, bool *force_latest, XLogRecPtr *force_lsn);
-static void prefetch_wait_for(uint64 ring_index);
-static void prefetch_cleanup(void);
-static inline void prefetch_set_unused(uint64 ring_index, bool hash_cleanup);
-
-static XLogRecPtr neon_get_request_lsn(bool *latest, RelFileNode rnode,
-									   ForkNumber forknum, BlockNumber blkno);
-
-
-/*
- * Make sure that there are no responses still in the buffer.
- */
 static void
 consume_prefetch_responses(void)
 {
-	if (MyPState->ring_receive < MyPState->ring_unused)
-		prefetch_wait_for(MyPState->ring_unused - 1);
-}
-
-static void
-prefetch_cleanup(void)
-{
-	int		index;
-	uint64	ring_index;
-	PrefetchRequest *slot;
-
-	while (MyPState->ring_last < MyPState->ring_receive) {
-		ring_index = MyPState->ring_last;
-		index = (ring_index % READ_BUFFER_SIZE);
-		slot = &MyPState->prf_buffer[index];
-
-		if (slot->status == PRFS_UNUSED)
-			MyPState->ring_last += 1;
-		else
-			break;
-	}
-}
-
-/*
- * Wait for slot of ring_index to have received its response.
- * The caller is responsible for making sure the request buffer is flushed.
- */
-static void
-prefetch_wait_for(uint64 ring_index)
-{
-	int index;
-	PrefetchRequest *entry;
-
-	Assert(MyPState->ring_unused > ring_index);
-
-	while (MyPState->ring_receive <= ring_index)
+	for (int i = n_prefetched_buffers; i < n_prefetch_responses; i++)
 	{
-		index = (MyPState->ring_receive % READ_BUFFER_SIZE);
-		entry = &MyPState->prf_buffer[index];
+		NeonResponse *resp = page_server->receive();
 
-		Assert(entry->status == PRFS_REQUESTED);
-		prefetch_read(entry);
+		pfree(resp);
 	}
-}
-
-/*
- * Read the response of a prefetch request into its slot.
- * 
- * The caller is responsible for making sure that the request for this buffer
- * was flushed to the PageServer.
- */
-static void
-prefetch_read(PrefetchRequest *slot)
-{
-	NeonResponse *response;
-	MemoryContext old;
-
-	Assert(slot->status == PRFS_REQUESTED);
-	Assert(slot->response == NULL);
-	Assert(slot->my_ring_index == MyPState->ring_receive);
-
-	old = MemoryContextSwitchTo(MyPState->errctx);
-	response = (NeonResponse *) page_server->receive();
-	MemoryContextSwitchTo(old);
-
-	/* update prefetch state */
-	MyPState->n_responses_buffered += 1;
-	MyPState->n_requests_inflight -= 1;
-	MyPState->ring_receive += 1;
-
-	/* update slot state */
-	slot->status = PRFS_RECEIVED;
-	slot->response = response;
-}
-
-/*
- * Disconnect hook - drop prefetches when the connection drops
- * 
- * If we don't remove the failed prefetches, we'd be serving incorrect
- * data to the smgr.
- */
-void
-prefetch_on_ps_disconnect(void)
-{
-	for (; MyPState->ring_receive < MyPState->ring_unused; MyPState->ring_receive++)
-	{
-		PrefetchRequest *slot;
-		int		index = MyPState->ring_receive % READ_BUFFER_SIZE;
-
-		slot = &MyPState->prf_buffer[index];
-		Assert(slot->status == PRFS_REQUESTED);
-		Assert(slot->my_ring_index == MyPState->ring_receive);
-
-		/* clean up the request */
-		slot->status = PRFS_TAG_REMAINS;
-		MyPState->n_requests_inflight--;
-		prefetch_set_unused(MyPState->ring_receive, true);
-	}
-}
-
-/*
- * prefetch_set_unused() - clear a received prefetch slot
- *
- * The slot at ring_index must be a current member of the ring buffer,
- * and may not be in the PRFS_REQUESTED state.
- */
-static inline void
-prefetch_set_unused(uint64 ring_index, bool hash_cleanup)
-{
-	PrefetchRequest *slot = &MyPState->prf_buffer[ring_index % READ_BUFFER_SIZE];
-
-	Assert(MyPState->ring_last <= ring_index &&
-		   MyPState->ring_unused > ring_index);
-
-	if (slot->status == PRFS_UNUSED)
-		return;
-
-	Assert(slot->status == PRFS_RECEIVED || slot->status == PRFS_TAG_REMAINS);
-	Assert(ring_index >= MyPState->ring_last &&
-		   ring_index < MyPState->ring_unused);
-
-	if (slot->status == PRFS_RECEIVED)
-	{
-		pfree(slot->response);
-		slot->response = NULL;
-
-		MyPState->n_responses_buffered -= 1;
-		MyPState->n_unused += 1;
-	}
-	else
-	{
-		Assert(slot->response == NULL);
-	}
-
-	if (hash_cleanup)
-		prfh_delete(MyPState->prf_hash, slot);
-
-	/* clear all fields */
-	MemSet(slot, 0, sizeof(PrefetchRequest));
-	slot->status = PRFS_UNUSED;
-
-	/* run cleanup if we're holding back ring_last */
-	if (MyPState->ring_last == ring_index)
-		prefetch_cleanup();
-}
-
-static void
-prefetch_do_request(PrefetchRequest *slot, bool *force_latest, XLogRecPtr *force_lsn)
-{
-	NeonGetPageRequest request = {
-		.req.tag = T_NeonGetPageRequest,
-		.req.latest = false,
-		.req.lsn = 0,
-		.rnode = slot->buftag.rnode,
-		.forknum = slot->buftag.forkNum,
-		.blkno = slot->buftag.blockNum,
-	};
-
-	if (force_lsn && force_latest)
-	{
-		request.req.lsn = *force_lsn;
-		request.req.latest = *force_latest;
-		slot->effective_request_lsn = *force_lsn;
-	}
-	else
-	{
-		XLogRecPtr lsn = neon_get_request_lsn(
-			&request.req.latest,
-			slot->buftag.rnode,
-			slot->buftag.forkNum,
-			slot->buftag.blockNum
-		);
-		/*
-		 * Note: effective_request_lsn is potentially higher than the requested
-		 * LSN, but still correct:
-		 * 
-		 * We know there are no changes between the actual requested LSN and
-		 * the value of effective_request_lsn: If there were, the page would
-		 * have been in cache and evicted between those LSN values, which
-		 * then would have had to result in a larger request LSN for this page.
-		 * 
-		 * It is possible that a concurrent backend loads the page, modifies
-		 * it and then evicts it again, but the LSN of that eviction cannot be
-		 * smaller than the current WAL insert/redo pointer, which is already
-		 * larger than this prefetch_lsn. So in any case, that would
-		 * invalidate this cache.
-		 * 
-		 * The best LSN to use for effective_request_lsn would be
-		 * XLogCtl->Insert.RedoRecPtr, but that's expensive to access.
-		 */
-		request.req.lsn = lsn;
-		prefetch_lsn = Max(prefetch_lsn, lsn);
-		slot->effective_request_lsn = prefetch_lsn;
-	}
-
-	Assert(slot->response == NULL);
-	Assert(slot->my_ring_index == MyPState->ring_unused);
-	page_server->send((NeonRequest *) &request);
-
-	/* update prefetch state */
-	MyPState->n_requests_inflight += 1;
-	MyPState->n_unused -= 1;
-	MyPState->ring_unused += 1;
-
-	/* update slot state */
-	slot->status = PRFS_REQUESTED;
-}
-
-/*
- * prefetch_register_buffer() - register and prefetch buffer
- *
- * Register that we may want the contents of BufferTag in the near future.
- * 
- * If force_latest and force_lsn are not NULL, those values are sent to the
- * pageserver. If they are NULL, we utilize the lastWrittenLsn -infrastructure
- * to fill in these values manually.
- */
-
-static uint64
-prefetch_register_buffer(BufferTag tag, bool *force_latest, XLogRecPtr *force_lsn)
-{
-	int		index;
-	bool	found;
-	uint64	ring_index;
-	PrefetchRequest req;
-	PrefetchRequest *slot;
-	PrfHashEntry *entry;
-
-	/* use an intermediate PrefetchRequest struct to ensure correct alignment */
-	req.buftag = tag;
-	
-	entry = prfh_lookup(MyPState->prf_hash, (PrefetchRequest *) &req);
-
-	if (entry != NULL)
-	{
-		slot = entry->slot;
-		ring_index = slot->my_ring_index;
-		index = (ring_index % READ_BUFFER_SIZE);
-		Assert(slot == &MyPState->prf_buffer[index]);
-
-		Assert(slot->status != PRFS_UNUSED);
-		Assert(BUFFERTAGS_EQUAL(slot->buftag, tag));
-		
-		/*
-		 * If we want a specific lsn, we do not accept requests that were made
-		 * with a potentially different LSN.
-		 */
-		if (force_lsn && slot->effective_request_lsn != *force_lsn)
-		{
-			prefetch_wait_for(ring_index);
-			prefetch_set_unused(ring_index, true);
-		}
-		/*
-		 * We received a prefetch for a page that was recently read and
-		 * removed from the buffers. Remove that request from the buffers.
-		 */
-		else if (slot->status == PRFS_TAG_REMAINS)
-		{
-			prefetch_set_unused(ring_index, true);
-		}
-		else
-		{
-			/* The buffered request is good enough, return that index */
-			n_prefetch_dupes++;
-			return ring_index;
-		}
-	}
-
-	/*
-	 * If the prefetch queue is full, we need to make room by clearing the
-	 * oldest slot. If the oldest slot holds a buffer that was already
-	 * received, we can just throw it away; we fetched the page unnecessarily
-	 * in that case. If the oldest slot holds a request that we haven't
-	 * received a response for yet, we have to wait for the response to that
-	 * before we can continue. We might not have even flushed the request to
-	 * the pageserver yet, it might be just sitting in the output buffer. In
-	 * that case, we flush it and wait for the response. (We could decide not
-	 * to send it, but it's hard to abort when the request is already in the
-	 * output buffer, and 'not sending' a prefetch request kind of goes
-	 * against the principles of prefetching)
-	 */
-	if (MyPState->ring_last + READ_BUFFER_SIZE - 1 == MyPState->ring_unused)
-	{
-		slot = &MyPState->prf_buffer[(MyPState->ring_last % READ_BUFFER_SIZE)];
-
-		Assert(slot->status != PRFS_UNUSED);
-
-		/* We have the slot for ring_last, so that must still be in progress */
-		switch (slot->status)
-		{
-			case PRFS_REQUESTED:
-				Assert(MyPState->ring_receive == MyPState->ring_last);
-				prefetch_wait_for(MyPState->ring_last);
-				prefetch_set_unused(MyPState->ring_last, true);
-				break;
-			case PRFS_RECEIVED:
-			case PRFS_TAG_REMAINS:
-				prefetch_set_unused(MyPState->ring_last, true);
-				break;
-			default:
-				pg_unreachable();
-		}
-	}
-
-	/*
-	 * The next buffer pointed to by `ring_unused` is now unused, so we can insert
-	 * the new request to it.
-	 */
-	ring_index = MyPState->ring_unused;
-	index = (ring_index % READ_BUFFER_SIZE);
-	slot = &MyPState->prf_buffer[index];
-
-	Assert(MyPState->ring_last <= ring_index);
-
-	Assert(slot->status == PRFS_UNUSED);
-
-	/*
-	 * We must update the slot data before insertion, because the hash
-	 * function reads the buffer tag from the slot.
-	 */
-	slot->buftag = tag;
-	slot->my_ring_index = ring_index;
-
-	prfh_insert(MyPState->prf_hash, slot, &found);
-	Assert(!found);
-
-	prefetch_do_request(slot, force_latest, force_lsn);
-	Assert(slot->status == PRFS_REQUESTED);
-	Assert(ring_index < MyPState->ring_unused);
-	return ring_index;
+	n_prefetched_buffers = 0;
+	n_prefetch_responses = 0;
 }
 
 static NeonResponse *
 page_server_request(void const *req)
 {
-	page_server->send((NeonRequest *) req);
-	page_server->flush();
 	consume_prefetch_responses();
-	return page_server->receive();
+	return page_server->request((NeonRequest *) req);
 }
 
 
@@ -700,15 +268,12 @@ nm_unpack_response(StringInfo s)
 
 		case T_NeonGetPageResponse:
 			{
-				NeonGetPageResponse *msg_resp;
+				NeonGetPageResponse *msg_resp = palloc0(offsetof(NeonGetPageResponse, page) + BLCKSZ);
 
-				msg_resp = MemoryContextAllocZero(MyPState->bufctx, PS_GETPAGERESPONSE_SIZE);
 				msg_resp->tag = tag;
 				/* XXX:	should be varlena */
 				memcpy(msg_resp->page, pq_getmsgbytes(s, BLCKSZ), BLCKSZ);
 				pq_getmsgend(s);
-				
-				Assert(msg_resp->tag == T_NeonGetPageResponse);
 
 				resp = (NeonResponse *) msg_resp;
 				break;
@@ -1052,32 +617,7 @@ neon_wallog_page(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, ch
 void
 neon_init(void)
 {
-	HASHCTL info;
-
-	if (MyPState != NULL)
-		return;
-
-	MyPState = MemoryContextAllocZero(TopMemoryContext, sizeof(PrefetchState));
-	
-	MyPState->n_unused = READ_BUFFER_SIZE;
-
-	MyPState->bufctx = SlabContextCreate(TopMemoryContext,
-										 "NeonSMGR/prefetch",
-										 SLAB_DEFAULT_BLOCK_SIZE * 17,
-										 PS_GETPAGERESPONSE_SIZE);
-	MyPState->errctx = AllocSetContextCreate(TopMemoryContext, 
-											 "NeonSMGR/errors",
-											 ALLOCSET_DEFAULT_SIZES);
-	MyPState->hashctx = AllocSetContextCreate(TopMemoryContext,
-											  "NeonSMGR/prefetch",
-											  ALLOCSET_DEFAULT_SIZES);
-
-	info.keysize = sizeof(BufferTag);
-	info.entrysize = sizeof(uint64);
-
-	MyPState->prf_hash = prfh_create(MyPState->hashctx,
-									 READ_BUFFER_SIZE, NULL);
-
+	/* noop */
 #ifdef DEBUG_COMPARE_LOCAL
 	mdinit();
 #endif
@@ -1464,17 +1004,27 @@ neon_close(SMgrRelation reln, ForkNumber forknum)
 }
 
 
+/*
+ *	neon_reset_prefetch() -- reoe all previously rgistered prefeth requests
+ */
+void
+neon_reset_prefetch(SMgrRelation reln)
+{
+	n_prefetch_requests = 0;
+}
+
 /*
  *	neon_prefetch() -- Initiate asynchronous read of the specified block of a relation
  */
 bool
 neon_prefetch(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum)
 {
-	uint64 ring_index;
-
 	switch (reln->smgr_relpersistence)
 	{
-		case 0: /* probably shouldn't happen, but ignore it */
+		case 0:
+			/* probably shouldn't happen, but ignore it */
+			break;
+
 		case RELPERSISTENCE_PERMANENT:
 			break;
 
@@ -1486,17 +1036,14 @@ neon_prefetch(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum)
 			elog(ERROR, "unknown relpersistence '%c'", reln->smgr_relpersistence);
 	}
 
-	BufferTag tag = (BufferTag) {
-		.rnode = reln->smgr_rnode.node,
-		.forkNum = forknum,
-		.blockNum = blocknum
-	};
-
-	ring_index = prefetch_register_buffer(tag, NULL, NULL);
-
-	Assert(ring_index < MyPState->ring_unused &&
-		   MyPState->ring_last <= ring_index);
-
+	if (n_prefetch_requests < MAX_PREFETCH_REQUESTS)
+	{
+		prefetch_requests[n_prefetch_requests].rnode = reln->smgr_rnode.node;
+		prefetch_requests[n_prefetch_requests].forkNum = forknum;
+		prefetch_requests[n_prefetch_requests].blockNum = blocknum;
+		n_prefetch_requests += 1;
+		return true;
+	}
 	return false;
 }
 
@@ -1547,72 +1094,81 @@ neon_read_at_lsn(RelFileNode rnode, ForkNumber forkNum, BlockNumber blkno,
 				 XLogRecPtr request_lsn, bool request_latest, char *buffer)
 {
 	NeonResponse *resp;
-	BufferTag	buftag;
-	uint64		ring_index;
-	PrfHashEntry *entry;
-	PrefetchRequest *slot;
-
-	buftag = (BufferTag) {
-		.rnode = rnode,
-		.forkNum = forkNum,
-		.blockNum = blkno,
-	};
+	int			i;
 
 	/*
-	 * Try to find prefetched page in the list of received pages.
+	 * Try to find prefetched page. It is assumed that pages will be requested
+	 * in the same order as them are prefetched, but some other backend may
+	 * load page in shared buffers, so some prefetch responses should be
+	 * skipped.
 	 */
-	entry = prfh_lookup(MyPState->prf_hash, (PrefetchRequest *) &buftag);
-
-	if (entry != NULL)
+	for (i = n_prefetched_buffers; i < n_prefetch_responses; i++)
 	{
-		if (entry->slot->effective_request_lsn >= prefetch_lsn)
-		{
-			slot = entry->slot;
-			ring_index = slot->my_ring_index;
-			n_prefetch_hits += 1;
-		}
-		else /* the current prefetch LSN is not large enough, so drop the prefetch */
+		resp = page_server->receive();
+		if (resp->tag == T_NeonGetPageResponse &&
+			RelFileNodeEquals(prefetch_responses[i].rnode, rnode) &&
+			prefetch_responses[i].forkNum == forkNum &&
+			prefetch_responses[i].blockNum == blkno)
 		{
+			char	   *page = ((NeonGetPageResponse *) resp)->page;
+
 			/*
-			 * We can't drop cache for not-yet-received requested items. It is
-			 * unlikely this happens, but it can happen if prefetch distance is
-			 * large enough and a backend didn't consume all prefetch requests.
+			 * Check if prefetched page is still relevant. If it is updated by
+			 * some other backend, then it should not be requested from smgr
+			 * unless it is evicted from shared buffers. In the last case
+			 * last_evicted_lsn should be updated and request_lsn should be
+			 * greater than prefetch_lsn. Maximum with page LSN is used
+			 * because page returned by page server may have LSN either
+			 * greater either smaller than requested.
 			 */
-			if (entry->slot->status == PRFS_REQUESTED)
+			if (Max(prefetch_lsn, PageGetLSN(page)) >= request_lsn)
 			{
-				page_server->flush();
-				prefetch_wait_for(entry->slot->my_ring_index);
+				n_prefetched_buffers = i + 1;
+				n_prefetch_hits += 1;
+				n_prefetch_requests = 0;
+				memcpy(buffer, page, BLCKSZ);
+				pfree(resp);
+				return;
 			}
-			/* drop caches */
-			prefetch_set_unused(entry->slot->my_ring_index, true);
-			n_prefetch_missed_caches += 1;
-			/* make it look like a prefetch cache miss */
-			entry = NULL;
+		}
+		pfree(resp);
+	}
+	n_prefetched_buffers = 0;
+	n_prefetch_responses = 0;
+	n_prefetch_misses += 1;
+	{
+		NeonGetPageRequest request = {
+			.req.tag = T_NeonGetPageRequest,
+			.req.latest = request_latest,
+			.req.lsn = request_lsn,
+			.rnode = rnode,
+			.forknum = forkNum,
+			.blkno = blkno
+		};
+
+		if (n_prefetch_requests > 0)
+		{
+			/* Combine all prefetch requests with primary request */
+			page_server->send((NeonRequest *) & request);
+			for (i = 0; i < n_prefetch_requests; i++)
+			{
+				request.rnode = prefetch_requests[i].rnode;
+				request.forknum = prefetch_requests[i].forkNum;
+				request.blkno = prefetch_requests[i].blockNum;
+				prefetch_responses[i] = prefetch_requests[i];
+				page_server->send((NeonRequest *) & request);
+			}
+			page_server->flush();
+			n_prefetch_responses = n_prefetch_requests;
+			n_prefetch_requests = 0;
+			prefetch_lsn = request_lsn;
+			resp = page_server->receive();
+		}
+		else
+		{
+			resp = page_server->request((NeonRequest *) & request);
 		}
 	}
-
-	if (entry == NULL)
-	{
-		n_prefetch_misses += 1;
-
-		ring_index = prefetch_register_buffer(buftag, &request_latest,
-											  &request_lsn);
-		slot = &MyPState->prf_buffer[(ring_index % READ_BUFFER_SIZE)];
-	}
-
-	Assert(MyPState->ring_last <= ring_index &&
-		   MyPState->ring_unused > ring_index);
-	Assert(slot->my_ring_index == ring_index);
-	Assert(slot->status != PRFS_UNUSED);
-	Assert(&MyPState->prf_buffer[(ring_index % READ_BUFFER_SIZE)] == slot);
-
-	page_server->flush();
-	prefetch_wait_for(ring_index);
-
-	Assert(slot->status == PRFS_RECEIVED);
-
-	resp = slot->response;
-
 	switch (resp->tag)
 	{
 		case T_NeonGetPageResponse:
@@ -1632,13 +1188,12 @@ neon_read_at_lsn(RelFileNode rnode, ForkNumber forkNum, BlockNumber blkno,
 					 errdetail("page server returned error: %s",
 							   ((NeonErrorResponse *) resp)->message)));
 			break;
+
 		default:
 			elog(ERROR, "unexpected response from page server with tag 0x%02x", resp->tag);
 	}
 
-	/* buffer was used, clean up for later reuse */
-	prefetch_set_unused(ring_index, true);
-	prefetch_cleanup();
+	pfree(resp);
 }
 
 /*
@@ -2260,6 +1815,7 @@ static const struct f_smgr neon_smgr =
 	.smgr_unlink = neon_unlink,
 	.smgr_extend = neon_extend,
 	.smgr_prefetch = neon_prefetch,
+	.smgr_reset_prefetch = neon_reset_prefetch,
 	.smgr_read = neon_read,
 	.smgr_write = neon_write,
 	.smgr_writeback = neon_writeback,

pgxn/neon/walproposer.c

@@ -75,7 +75,7 @@ static bool syncSafekeepers = false;
 
 char	   *wal_acceptors_list;
 int			wal_acceptor_reconnect_timeout;
-int			wal_acceptor_connection_timeout;
+int			wal_acceptor_connect_timeout;
 bool		am_wal_proposer;
 
 char	   *neon_timeline_walproposer = NULL;
@@ -266,9 +266,9 @@ nwp_register_gucs(void)
 
 	DefineCustomIntVariable(
 							"neon.safekeeper_connect_timeout",
-							"Timeout for connection establishement and it's maintenance against safekeeper",
+							"Timeout after which give up connection attempt to safekeeper.",
 							NULL,
-							&wal_acceptor_connection_timeout,
+							&wal_acceptor_connect_timeout,
 							5000, 0, INT_MAX,
 							PGC_SIGHUP,
 							GUC_UNIT_MS,
@@ -417,9 +417,7 @@ WalProposerPoll(void)
 			ResetLatch(MyLatch);
 			break;
 		}
-
-		now = GetCurrentTimestamp();
-		if (rc == 0 || TimeToReconnect(now) <= 0)			/* timeout expired: poll state */
+		if (rc == 0)			/* timeout expired: poll state */
 		{
 			TimestampTz now;
 
@@ -440,11 +438,13 @@ WalProposerPoll(void)
 			{
 				Safekeeper *sk = &safekeeper[i];
 
-				if (TimestampDifferenceExceeds(sk->latestMsgReceivedAt, now,
-											   wal_acceptor_connection_timeout))
+				if ((sk->state == SS_CONNECTING_WRITE ||
+					 sk->state == SS_CONNECTING_READ) &&
+					TimestampDifferenceExceeds(sk->startedConnAt, now,
+											   wal_acceptor_connect_timeout))
 				{
-					elog(WARNING, "failed to connect to node '%s:%s' in '%s' state: exceeded connection timeout %dms",
-						 sk->host, sk->port, FormatSafekeeperState(sk->state), wal_acceptor_connection_timeout);
+					elog(WARNING, "failed to connect to node '%s:%s': exceeded connection timeout %dms",
+						 sk->host, sk->port, wal_acceptor_connect_timeout);
 					ShutdownConnection(sk);
 				}
 			}
@@ -760,7 +760,7 @@ ResetConnection(Safekeeper *sk)
 	elog(LOG, "connecting with node %s:%s", sk->host, sk->port);
 
 	sk->state = SS_CONNECTING_WRITE;
-	sk->latestMsgReceivedAt = GetCurrentTimestamp();
+	sk->startedConnAt = GetCurrentTimestamp();
 
 	sock = walprop_socket(sk->conn);
 	sk->eventPos = AddWaitEventToSet(waitEvents, WL_SOCKET_WRITEABLE, sock, NULL, sk);
@@ -918,7 +918,7 @@ HandleConnectionEvent(Safekeeper *sk)
 		case WP_CONN_POLLING_OK:
 			elog(LOG, "connected with node %s:%s", sk->host,
 				 sk->port);
-			sk->latestMsgReceivedAt = GetCurrentTimestamp();
+
 			/*
 			 * We have to pick some event to update event set. We'll
 			 * eventually need the socket to be readable, so we go with that.
@@ -2304,7 +2304,7 @@ AsyncReadMessage(Safekeeper *sk, AcceptorProposerMessage * anymsg)
 		ResetConnection(sk);
 		return false;
 	}
-	sk->latestMsgReceivedAt = GetCurrentTimestamp();
+
 	switch (tag)
 	{
 		case 'g':

pgxn/neon/walproposer.h

@@ -30,7 +30,7 @@
 
 extern char *wal_acceptors_list;
 extern int	wal_acceptor_reconnect_timeout;
-extern int	wal_acceptor_connection_timeout;
+extern int	wal_acceptor_connect_timeout;
 extern bool am_wal_proposer;
 
 struct WalProposerConn;			/* Defined in libpqwalproposer */
@@ -371,7 +371,7 @@ typedef struct Safekeeper
 	int			eventPos;		/* position in wait event set. Equal to -1 if*
 								 * no event */
 	SafekeeperState state;		/* safekeeper state machine state */
-	TimestampTz latestMsgReceivedAt;        /* when latest msg is received */
+	TimestampTz startedConnAt;	/* when connection attempt started */
 	AcceptorGreeting greetResponse; /* acceptor greeting */
 	VoteResponse voteResponse;	/* the vote */
 	AppendResponse appendResponse;	/* feedback for master */

poetry.lock

@@ -1077,17 +1077,6 @@ python-versions = ">=3.6"
 [package.extras]
 twisted = ["twisted"]
 
-[[package]]
-name = "psutil"
-version = "5.9.4"
-description = "Cross-platform lib for process and system monitoring in Python."
-category = "main"
-optional = false
-python-versions = ">=2.7, !=3.0.*, !=3.1.*, !=3.2.*, !=3.3.*"
-
-[package.extras]
-test = ["enum34", "ipaddress", "mock", "pywin32", "wmi"]
-
 [[package]]
 name = "psycopg2-binary"
 version = "2.9.3"
@@ -1218,6 +1207,18 @@ pytest = ">=6.1.0"
 [package.extras]
 testing = ["coverage (>=6.2)", "flaky (>=3.5.0)", "hypothesis (>=5.7.1)", "mypy (>=0.931)", "pytest-trio (>=0.7.0)"]
 
+[[package]]
+name = "pytest-forked"
+version = "1.4.0"
+description = "run tests in isolated forked subprocesses"
+category = "main"
+optional = false
+python-versions = ">=3.6"
+
+[package.dependencies]
+py = "*"
+pytest = ">=3.10"
+
 [[package]]
 name = "pytest-lazy-fixture"
 version = "0.6.3"
@@ -1239,8 +1240,8 @@ python-versions = ">=3.6"
 
 [package.dependencies]
 pytest = [
-    {version = ">=5.0", markers = "python_version < \"3.10\""},
     {version = ">=6.2.4", markers = "python_version >= \"3.10\""},
+    {version = ">=5.0", markers = "python_version < \"3.10\""},
 ]
 
 [[package]]
@@ -1256,7 +1257,7 @@ pytest = ">=5.0.0"
 
 [[package]]
 name = "pytest-xdist"
-version = "3.0.2"
+version = "2.5.0"
 description = "pytest xdist plugin for distributed testing and loop-on-failing modes"
 category = "main"
 optional = false
@@ -1265,6 +1266,7 @@ python-versions = ">=3.6"
 [package.dependencies]
 execnet = ">=1.1"
 pytest = ">=6.2.0"
+pytest-forked = "*"
 
 [package.extras]
 psutil = ["psutil (>=3.0)"]
@@ -1447,14 +1449,6 @@ category = "dev"
 optional = false
 python-versions = ">=3.7"
 
-[[package]]
-name = "types-psutil"
-version = "5.9.5.4"
-description = "Typing stubs for psutil"
-category = "main"
-optional = false
-python-versions = "*"
-
 [[package]]
 name = "types-psycopg2"
 version = "2.9.18"
@@ -1574,7 +1568,7 @@ testing = ["func-timeout", "jaraco.itertools", "pytest (>=6)", "pytest-black (>=
 [metadata]
 lock-version = "1.1"
 python-versions = "^3.9"
-content-hash = "c95c184fccaf40815405ad616ec1c55869c7f87b72777cc3a9cbaff41de98977"
+content-hash = "9352a89d49d34807f6a58f6c3f898acbd8cf3570e0f45ede973673644bde4d0e"
 
 [metadata.files]
 aiopg = [
@@ -1985,26 +1979,9 @@ prometheus-client = [
     {file = "prometheus_client-0.14.1-py3-none-any.whl", hash = "sha256:522fded625282822a89e2773452f42df14b5a8e84a86433e3f8a189c1d54dc01"},
     {file = "prometheus_client-0.14.1.tar.gz", hash = "sha256:5459c427624961076277fdc6dc50540e2bacb98eebde99886e59ec55ed92093a"},
 ]
-psutil = [
-    {file = "psutil-5.9.4-cp27-cp27m-macosx_10_9_x86_64.whl", hash = "sha256:c1ca331af862803a42677c120aff8a814a804e09832f166f226bfd22b56feee8"},
-    {file = "psutil-5.9.4-cp27-cp27m-manylinux2010_i686.whl", hash = "sha256:68908971daf802203f3d37e78d3f8831b6d1014864d7a85937941bb35f09aefe"},
-    {file = "psutil-5.9.4-cp27-cp27m-manylinux2010_x86_64.whl", hash = "sha256:3ff89f9b835100a825b14c2808a106b6fdcc4b15483141482a12c725e7f78549"},
-    {file = "psutil-5.9.4-cp27-cp27m-win32.whl", hash = "sha256:852dd5d9f8a47169fe62fd4a971aa07859476c2ba22c2254d4a1baa4e10b95ad"},
-    {file = "psutil-5.9.4-cp27-cp27m-win_amd64.whl", hash = "sha256:9120cd39dca5c5e1c54b59a41d205023d436799b1c8c4d3ff71af18535728e94"},
-    {file = "psutil-5.9.4-cp27-cp27mu-manylinux2010_i686.whl", hash = "sha256:6b92c532979bafc2df23ddc785ed116fced1f492ad90a6830cf24f4d1ea27d24"},
-    {file = "psutil-5.9.4-cp27-cp27mu-manylinux2010_x86_64.whl", hash = "sha256:efeae04f9516907be44904cc7ce08defb6b665128992a56957abc9b61dca94b7"},
-    {file = "psutil-5.9.4-cp36-abi3-macosx_10_9_x86_64.whl", hash = "sha256:54d5b184728298f2ca8567bf83c422b706200bcbbfafdc06718264f9393cfeb7"},
-    {file = "psutil-5.9.4-cp36-abi3-manylinux_2_12_i686.manylinux2010_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:16653106f3b59386ffe10e0bad3bb6299e169d5327d3f187614b1cb8f24cf2e1"},
-    {file = "psutil-5.9.4-cp36-abi3-manylinux_2_12_x86_64.manylinux2010_x86_64.manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:54c0d3d8e0078b7666984e11b12b88af2db11d11249a8ac8920dd5ef68a66e08"},
-    {file = "psutil-5.9.4-cp36-abi3-win32.whl", hash = "sha256:149555f59a69b33f056ba1c4eb22bb7bf24332ce631c44a319cec09f876aaeff"},
-    {file = "psutil-5.9.4-cp36-abi3-win_amd64.whl", hash = "sha256:fd8522436a6ada7b4aad6638662966de0d61d241cb821239b2ae7013d41a43d4"},
-    {file = "psutil-5.9.4-cp38-abi3-macosx_11_0_arm64.whl", hash = "sha256:6001c809253a29599bc0dfd5179d9f8a5779f9dffea1da0f13c53ee568115e1e"},
-    {file = "psutil-5.9.4.tar.gz", hash = "sha256:3d7f9739eb435d4b1338944abe23f49584bde5395f27487d2ee25ad9a8774a62"},
-]
 psycopg2-binary = [
     {file = "psycopg2-binary-2.9.3.tar.gz", hash = "sha256:761df5313dc15da1502b21453642d7599d26be88bff659382f8f9747c7ebea4e"},
     {file = "psycopg2_binary-2.9.3-cp310-cp310-macosx_10_14_x86_64.macosx_10_9_intel.macosx_10_9_x86_64.macosx_10_10_intel.macosx_10_10_x86_64.whl", hash = "sha256:539b28661b71da7c0e428692438efbcd048ca21ea81af618d845e06ebfd29478"},
-    {file = "psycopg2_binary-2.9.3-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:2f2534ab7dc7e776a263b463a16e189eb30e85ec9bbe1bff9e78dae802608932"},
     {file = "psycopg2_binary-2.9.3-cp310-cp310-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:6e82d38390a03da28c7985b394ec3f56873174e2c88130e6966cb1c946508e65"},
     {file = "psycopg2_binary-2.9.3-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:57804fc02ca3ce0dbfbef35c4b3a4a774da66d66ea20f4bda601294ad2ea6092"},
     {file = "psycopg2_binary-2.9.3-cp310-cp310-manylinux_2_24_aarch64.whl", hash = "sha256:083a55275f09a62b8ca4902dd11f4b33075b743cf0d360419e2051a8a5d5ff76"},
@@ -2038,7 +2015,6 @@ psycopg2-binary = [
     {file = "psycopg2_binary-2.9.3-cp37-cp37m-win32.whl", hash = "sha256:adf20d9a67e0b6393eac162eb81fb10bc9130a80540f4df7e7355c2dd4af9fba"},
     {file = "psycopg2_binary-2.9.3-cp37-cp37m-win_amd64.whl", hash = "sha256:2f9ffd643bc7349eeb664eba8864d9e01f057880f510e4681ba40a6532f93c71"},
     {file = "psycopg2_binary-2.9.3-cp38-cp38-macosx_10_14_x86_64.macosx_10_9_intel.macosx_10_9_x86_64.macosx_10_10_intel.macosx_10_10_x86_64.whl", hash = "sha256:def68d7c21984b0f8218e8a15d514f714d96904265164f75f8d3a70f9c295667"},
-    {file = "psycopg2_binary-2.9.3-cp38-cp38-macosx_11_0_arm64.whl", hash = "sha256:e6aa71ae45f952a2205377773e76f4e3f27951df38e69a4c95440c779e013560"},
     {file = "psycopg2_binary-2.9.3-cp38-cp38-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:dffc08ca91c9ac09008870c9eb77b00a46b3378719584059c034b8945e26b272"},
     {file = "psycopg2_binary-2.9.3-cp38-cp38-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:280b0bb5cbfe8039205c7981cceb006156a675362a00fe29b16fbc264e242834"},
     {file = "psycopg2_binary-2.9.3-cp38-cp38-manylinux_2_24_aarch64.whl", hash = "sha256:af9813db73395fb1fc211bac696faea4ca9ef53f32dc0cfa27e4e7cf766dcf24"},
@@ -2050,7 +2026,6 @@ psycopg2-binary = [
     {file = "psycopg2_binary-2.9.3-cp38-cp38-win32.whl", hash = "sha256:6472a178e291b59e7f16ab49ec8b4f3bdada0a879c68d3817ff0963e722a82ce"},
     {file = "psycopg2_binary-2.9.3-cp38-cp38-win_amd64.whl", hash = "sha256:35168209c9d51b145e459e05c31a9eaeffa9a6b0fd61689b48e07464ffd1a83e"},
     {file = "psycopg2_binary-2.9.3-cp39-cp39-macosx_10_14_x86_64.macosx_10_9_intel.macosx_10_9_x86_64.macosx_10_10_intel.macosx_10_10_x86_64.whl", hash = "sha256:47133f3f872faf28c1e87d4357220e809dfd3fa7c64295a4a148bcd1e6e34ec9"},
-    {file = "psycopg2_binary-2.9.3-cp39-cp39-macosx_11_0_arm64.whl", hash = "sha256:b3a24a1982ae56461cc24f6680604fffa2c1b818e9dc55680da038792e004d18"},
     {file = "psycopg2_binary-2.9.3-cp39-cp39-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:91920527dea30175cc02a1099f331aa8c1ba39bf8b7762b7b56cbf54bc5cce42"},
     {file = "psycopg2_binary-2.9.3-cp39-cp39-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:887dd9aac71765ac0d0bac1d0d4b4f2c99d5f5c1382d8b770404f0f3d0ce8a39"},
     {file = "psycopg2_binary-2.9.3-cp39-cp39-manylinux_2_24_aarch64.whl", hash = "sha256:1f14c8b0942714eb3c74e1e71700cbbcb415acbc311c730370e70c578a44a25c"},
@@ -2067,7 +2042,18 @@ py = [
     {file = "py-1.11.0.tar.gz", hash = "sha256:51c75c4126074b472f746a24399ad32f6053d1b34b68d2fa41e558e6f4a98719"},
 ]
 pyasn1 = [
+    {file = "pyasn1-0.4.8-py2.4.egg", hash = "sha256:fec3e9d8e36808a28efb59b489e4528c10ad0f480e57dcc32b4de5c9d8c9fdf3"},
+    {file = "pyasn1-0.4.8-py2.5.egg", hash = "sha256:0458773cfe65b153891ac249bcf1b5f8f320b7c2ce462151f8fa74de8934becf"},
+    {file = "pyasn1-0.4.8-py2.6.egg", hash = "sha256:5c9414dcfede6e441f7e8f81b43b34e834731003427e5b09e4e00e3172a10f00"},
+    {file = "pyasn1-0.4.8-py2.7.egg", hash = "sha256:6e7545f1a61025a4e58bb336952c5061697da694db1cae97b116e9c46abcf7c8"},
     {file = "pyasn1-0.4.8-py2.py3-none-any.whl", hash = "sha256:39c7e2ec30515947ff4e87fb6f456dfc6e84857d34be479c9d4a4ba4bf46aa5d"},
+    {file = "pyasn1-0.4.8-py3.1.egg", hash = "sha256:78fa6da68ed2727915c4767bb386ab32cdba863caa7dbe473eaae45f9959da86"},
+    {file = "pyasn1-0.4.8-py3.2.egg", hash = "sha256:08c3c53b75eaa48d71cf8c710312316392ed40899cb34710d092e96745a358b7"},
+    {file = "pyasn1-0.4.8-py3.3.egg", hash = "sha256:03840c999ba71680a131cfaee6fab142e1ed9bbd9c693e285cc6aca0d555e576"},
+    {file = "pyasn1-0.4.8-py3.4.egg", hash = "sha256:7ab8a544af125fb704feadb008c99a88805126fb525280b2270bb25cc1d78a12"},
+    {file = "pyasn1-0.4.8-py3.5.egg", hash = "sha256:e89bf84b5437b532b0803ba5c9a5e054d21fec423a89952a74f87fa2c9b7bce2"},
+    {file = "pyasn1-0.4.8-py3.6.egg", hash = "sha256:014c0e9976956a08139dc0712ae195324a75e142284d5f87f1a87ee1b068a359"},
+    {file = "pyasn1-0.4.8-py3.7.egg", hash = "sha256:99fcc3c8d804d1bc6d9a099921e39d827026409a58f2a720dcdb89374ea0c776"},
     {file = "pyasn1-0.4.8.tar.gz", hash = "sha256:aef77c9fb94a3ac588e87841208bdec464471d9871bd5050a287cc9a475cd0ba"},
 ]
 pycodestyle = [
@@ -2125,6 +2111,10 @@ pytest-asyncio = [
     {file = "pytest-asyncio-0.19.0.tar.gz", hash = "sha256:ac4ebf3b6207259750bc32f4c1d8fcd7e79739edbc67ad0c58dd150b1d072fed"},
     {file = "pytest_asyncio-0.19.0-py3-none-any.whl", hash = "sha256:7a97e37cfe1ed296e2e84941384bdd37c376453912d397ed39293e0916f521fa"},
 ]
+pytest-forked = [
+    {file = "pytest-forked-1.4.0.tar.gz", hash = "sha256:8b67587c8f98cbbadfdd804539ed5455b6ed03802203485dd2f53c1422d7440e"},
+    {file = "pytest_forked-1.4.0-py3-none-any.whl", hash = "sha256:bbbb6717efc886b9d64537b41fb1497cfaf3c9601276be8da2cccfea5a3c8ad8"},
+]
 pytest-lazy-fixture = [
     {file = "pytest-lazy-fixture-0.6.3.tar.gz", hash = "sha256:0e7d0c7f74ba33e6e80905e9bfd81f9d15ef9a790de97993e34213deb5ad10ac"},
     {file = "pytest_lazy_fixture-0.6.3-py3-none-any.whl", hash = "sha256:e0b379f38299ff27a653f03eaa69b08a6fd4484e46fd1c9907d984b9f9daeda6"},
@@ -2138,8 +2128,8 @@ pytest-timeout = [
     {file = "pytest_timeout-2.1.0-py3-none-any.whl", hash = "sha256:f6f50101443ce70ad325ceb4473c4255e9d74e3c7cd0ef827309dfa4c0d975c6"},
 ]
 pytest-xdist = [
-    {file = "pytest-xdist-3.0.2.tar.gz", hash = "sha256:688da9b814370e891ba5de650c9327d1a9d861721a524eb917e620eec3e90291"},
-    {file = "pytest_xdist-3.0.2-py3-none-any.whl", hash = "sha256:9feb9a18e1790696ea23e1434fa73b325ed4998b0e9fcb221f16fd1945e6df1b"},
+    {file = "pytest-xdist-2.5.0.tar.gz", hash = "sha256:4580deca3ff04ddb2ac53eba39d76cb5dd5edeac050cb6fbc768b0dd712b4edf"},
+    {file = "pytest_xdist-2.5.0-py3-none-any.whl", hash = "sha256:6fe5c74fec98906deb8f2d2b616b5c782022744978e7bd4695d39c8f42d0ce65"},
 ]
 python-dateutil = [
     {file = "python-dateutil-2.8.2.tar.gz", hash = "sha256:0123cacc1627ae19ddf3c27a5de5bd67ee4586fbdd6440d9748f8abb483d3e86"},
@@ -2173,13 +2163,6 @@ pyyaml = [
     {file = "PyYAML-6.0-cp310-cp310-manylinux_2_5_x86_64.manylinux1_x86_64.manylinux_2_12_x86_64.manylinux2010_x86_64.whl", hash = "sha256:f84fbc98b019fef2ee9a1cb3ce93e3187a6df0b2538a651bfb890254ba9f90b5"},
     {file = "PyYAML-6.0-cp310-cp310-win32.whl", hash = "sha256:2cd5df3de48857ed0544b34e2d40e9fac445930039f3cfe4bcc592a1f836d513"},
     {file = "PyYAML-6.0-cp310-cp310-win_amd64.whl", hash = "sha256:daf496c58a8c52083df09b80c860005194014c3698698d1a57cbcfa182142a3a"},
-    {file = "PyYAML-6.0-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:d4b0ba9512519522b118090257be113b9468d804b19d63c71dbcf4a48fa32358"},
-    {file = "PyYAML-6.0-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:81957921f441d50af23654aa6c5e5eaf9b06aba7f0a19c18a538dc7ef291c5a1"},
-    {file = "PyYAML-6.0-cp311-cp311-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:afa17f5bc4d1b10afd4466fd3a44dc0e245382deca5b3c353d8b757f9e3ecb8d"},
-    {file = "PyYAML-6.0-cp311-cp311-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:dbad0e9d368bb989f4515da330b88a057617d16b6a8245084f1b05400f24609f"},
-    {file = "PyYAML-6.0-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:432557aa2c09802be39460360ddffd48156e30721f5e8d917f01d31694216782"},
-    {file = "PyYAML-6.0-cp311-cp311-win32.whl", hash = "sha256:bfaef573a63ba8923503d27530362590ff4f576c626d86a9fed95822a8255fd7"},
-    {file = "PyYAML-6.0-cp311-cp311-win_amd64.whl", hash = "sha256:01b45c0191e6d66c470b6cf1b9531a771a83c1c4208272ead47a3ae4f2f603bf"},
     {file = "PyYAML-6.0-cp36-cp36m-macosx_10_9_x86_64.whl", hash = "sha256:897b80890765f037df3403d22bab41627ca8811ae55e9a722fd0392850ec4d86"},
     {file = "PyYAML-6.0-cp36-cp36m-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:50602afada6d6cbfad699b0c7bb50d5ccffa7e46a3d738092afddc1f9758427f"},
     {file = "PyYAML-6.0-cp36-cp36m-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:48c346915c114f5fdb3ead70312bd042a953a8ce5c7106d5bfb1a5254e47da92"},
@@ -2247,10 +2230,6 @@ tomli = [
     {file = "tomli-2.0.1-py3-none-any.whl", hash = "sha256:939de3e7a6161af0c887ef91b7d41a53e7c5a1ca976325f429cb46ea9bc30ecc"},
     {file = "tomli-2.0.1.tar.gz", hash = "sha256:de526c12914f0c550d15924c62d72abc48d6fe7364aa87328337a31007fe8a4f"},
 ]
-types-psutil = [
-    {file = "types-psutil-5.9.5.4.tar.gz", hash = "sha256:aa09102b80c65a3b4573216614372398dab78972d650488eaff1ff05482cc18f"},
-    {file = "types_psutil-5.9.5.4-py3-none-any.whl", hash = "sha256:28e59764630187e462d43788efa16d59d5e77b510115f9e25901b2d4007fca62"},
-]
 types-psycopg2 = [
     {file = "types-psycopg2-2.9.18.tar.gz", hash = "sha256:9b0e9e1f097b15cd9fa8aad2596a9e3082fd72f8d9cfe52b190cfa709105b6c0"},
     {file = "types_psycopg2-2.9.18-py3-none-any.whl", hash = "sha256:14c779dcab18c31453fa1cad3cf4b1601d33540a344adead3c47a6b8091cd2fa"},

pyproject.toml

@@ -11,7 +11,7 @@ psycopg2-binary = "^2.9.1"
 typing-extensions = "^4.1.0"
 PyJWT = {version = "^2.1.0", extras = ["crypto"]}
 requests = "^2.26.0"
-pytest-xdist = "^3.0.2"
+pytest-xdist = "^2.3.0"
 asyncpg = "^0.24.0"
 aiopg = "^1.3.1"
 Jinja2 = "^3.0.2"
@@ -29,8 +29,6 @@ pytest-order = "^1.0.1"
 allure-pytest = "^2.10.0"
 pytest-asyncio = "^0.19.0"
 toml = "^0.10.2"
-psutil = "^5.9.4"
-types-psutil = "^5.9.5.4"
 
 [tool.poetry.dev-dependencies]
 flake8 = "^5.0.4"

safekeeper/src/timeline.rs

@@ -556,6 +556,10 @@ impl Timeline {
                 .pageserver_feedback
                 .map(|f| Lsn(f.ps_applylsn))
                 .unwrap_or(Lsn::INVALID);
+            info!(
+                "checking should ws stop ttid {} lsn {} rcl {}",
+                self.ttid, reported_remote_consistent_lsn, shared_state.sk.inmem.commit_lsn
+            );
             let stop = shared_state.sk.inmem.commit_lsn == Lsn(0) || // no data at all yet
             (reported_remote_consistent_lsn!= Lsn::MAX && // Lsn::MAX means that we don't know the latest LSN yet.
             reported_remote_consistent_lsn >= shared_state.sk.inmem.commit_lsn);

scripts/strip-useless-debug.py

@@ -0,0 +1,179 @@
+#!/usr/bin/env python3
+
+# Strip useless .debug_pubnames and .debug_pubtypes from all binaries.
+# They bloat the binaries, and are not used by modern debuggers anyway.
+# This makes the resulting binaries about 30% smaller, and also makes
+# the cargo cache smaller.
+#
+# See also https://github.com/rust-lang/rust/issues/46034
+#
+# Usage:
+#     ./scripts/strip-useless-debug.py target
+#
+#
+# Why is this script needed?
+# --------------------------
+#
+# The simplest way to do this would be just:
+#
+#     find target -executable -type f -size +0 | \
+#       xargs -IPATH objcopy -R .debug_pubnames -R .debug_pubtypes -p PATH
+#
+# However, objcopy is not very fast, so we want to run it in parallel.
+# That would be straightforward to do with the xargs -P option, except
+# that the rust target directory contains hard links. Running objcopy
+# on multiple paths that are hardlinked to the same underlying file
+# doesn't work, because one objcopy could be overwriting the file while
+# the other one is trying to read it.
+#
+# To work around that, this script scans the target directory and
+# collects paths of all executables, except that when multiple paths
+# point to the same underlying inode, i.e. if two paths are hard links
+# to the same file, only one of the paths is collected. Then, it runs
+# objcopy on each of the unique files.
+#
+# There's one more subtle problem with hardlinks. The GNU objcopy man
+# page says that:
+#
+#    If you do not specify outfile, objcopy creates a temporary file and
+#    destructively renames the result with the name of infile.
+#
+# That is a problem: renaming over the file will create a new inode
+# for the path, and leave the other hardlinked paths unchanged. We
+# want to modify all the hard linked copies, and we also don't want to
+# remove the hard linking, as that saves a lot of space. In testing,
+# at least some versions of GNU objcopy seem to actually behave
+# differently if the file has hard links, copying over the file
+# instead of renaming if it has. So that text in the man page isn't
+# totally accurate. But that's hardly something we should rely on:
+# llvm-objcopy for example always renames. To avoid that problem, we
+# specify a temporary file as the destination, and copy it over the
+# original file in this python script. That way, it is independent of
+# objcopy's behavior.
+
+import argparse
+import asyncio
+import os
+import time
+import shutil
+import subprocess
+import tempfile
+from pathlib import Path
+
+
+async def main():
+    parser = argparse.ArgumentParser(
+        description="Strip useless .debug_pubnames and .debug_putypes sections from binaries",
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+    )
+    parser.add_argument(
+        "-j", metavar="NUM", default=os.cpu_count(), type=int, help="number of parallel processes"
+    )
+    parser.add_argument("target", type=Path, help="target directory")
+
+    args = parser.parse_args()
+    max_parallel_processes = args.j
+    target_dir = args.target
+
+    # Collect list of executables in the target dir. Make note of the inode of
+    # each path, and only record one path with the same inode. This ensures that
+    # if there are hard links in the directory tree, we only remember one path
+    # for each underlying file.
+    inode_paths = {}  # inode -> path dictionary
+
+    def onerror(err):
+        raise err
+
+    for currentpath, folders, files in os.walk(target_dir, onerror=onerror):
+        for file in files:
+            path = os.path.join(currentpath, file)
+            if os.access(path, os.X_OK):
+                stat = os.stat(path)
+                # If multiple paths ar hardlinked to the same underlying file,
+                # only we remember the first one that we see. It's arbitrary
+                # which one we will see first, but that's ok.
+                #
+                # Skip empty files while we're at it. There are some .lock files
+                # in the target directory that are marked as executable, but are
+                # are binaries so objcopy would complain about them.
+                if stat.st_size > 0:
+                    prev = inode_paths.get(stat.st_ino)
+                    if prev:
+                        print(f"{path} is a hard link to {prev}, skipping")
+                    else:
+                        inode_paths[stat.st_ino] = path
+
+    # This function runs "objcopy -R .debug_pubnames -R .debug_pubtypes" on a file.
+    #
+    # Returns (original size, new size)
+    async def run_objcopy(path) -> (int, int):
+        stat = os.stat(path)
+        orig_size = stat.st_size
+        if orig_size == 0:
+            return (0, 0)
+
+        # Write the output to a temp file first, and then copy it over the original.
+        # objcopy could modify the file in place, but that's not reliable with hard
+        # links. (See comment at beginning of this file.)
+        with tempfile.NamedTemporaryFile() as tmpfile:
+            cmd = [
+                "objcopy",
+                "-R",
+                ".debug_pubnames",
+                "-R",
+                ".debug_pubtypes",
+                "-p",
+                path,
+                tmpfile.name,
+            ]
+            proc = await asyncio.create_subprocess_exec(*cmd)
+            rc = await proc.wait()
+            if rc != 0:
+                raise subprocess.CalledProcessError(rc, cmd)
+
+            # If the file got smaller, copy it over the original.
+            # Otherwise keep the original
+            stat = os.stat(tmpfile.name)
+            new_size = stat.st_size
+            if new_size < orig_size:
+                with open(path, "wb") as orig_dst:
+                    shutil.copyfileobj(tmpfile, orig_dst)
+                    return (orig_size, new_size)
+            else:
+                return (orig_size, orig_size)
+
+    # convert the inode->path dictionary into plain list of paths.
+    paths = []
+    for path in inode_paths.values():
+        paths.append(path)
+
+    # Launch worker processes to process the list of files
+    before_total = 0
+    after_total = 0
+
+    async def runner_subproc():
+        nonlocal before_total
+        nonlocal after_total
+        while len(paths) > 0:
+            path = paths.pop()
+
+            start_time = time.perf_counter_ns()
+            (before_size, after_size) = await run_objcopy(path)
+            end_time = time.perf_counter_ns()
+            before_total += before_size
+            after_total += after_size
+
+            duration_ms = round((end_time-start_time) / 1000000)
+            print(f"{path}: {before_size} to {after_size} bytes ({duration_ms} ms)")
+
+    active_workers = []
+    for i in range(max_parallel_processes):
+        active_workers.append(asyncio.create_task(runner_subproc()))
+    done, () = await asyncio.wait(active_workers)
+
+    # all done!
+    print(f"total size before {before_total} after: {after_total}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

test_runner/fixtures/benchmark_fixture.py

@@ -11,37 +11,39 @@ from datetime import datetime
 from pathlib import Path
 
 # Type-related stuff
-from typing import Callable, ClassVar, Iterator, Optional
+from typing import Iterator, Optional
 
 import pytest
 from _pytest.config import Config
-from _pytest.config.argparsing import Parser
 from _pytest.terminal import TerminalReporter
-from fixtures.neon_fixtures import NeonPageserver
 from fixtures.types import TenantId, TimelineId
 
 """
 This file contains fixtures for micro-benchmarks.
 
-To use, declare the `zenbenchmark` fixture in the test function. Run the
-bencmark, and then record the result by calling `zenbenchmark.record`. For example:
+To use, declare the 'zenbenchmark' fixture in the test function. Run the
+bencmark, and then record the result by calling zenbenchmark.record. For example:
 
->>> import timeit
->>> from fixtures.neon_fixtures import NeonEnv
->>> def test_mybench(neon_simple_env: NeonEnv, zenbenchmark):
-...     # Initialize the test
-...     ...
-...     # Run the test, timing how long it takes
-...     with zenbenchmark.record_duration('test_query'):
-...         cur.execute('SELECT test_query(...)')
-...     # Record another measurement
-...     zenbenchmark.record('speed_of_light', 300000, 'km/s')
+import timeit
+from fixtures.neon_fixtures import NeonEnv
+
+def test_mybench(neon_simple_env: env, zenbenchmark):
+
+    # Initialize the test
+    ...
+
+    # Run the test, timing how long it takes
+    with zenbenchmark.record_duration('test_query'):
+        cur.execute('SELECT test_query(...)')
+
+    # Record another measurement
+    zenbenchmark.record('speed_of_light', 300000, 'km/s')
 
 There's no need to import this file to use it. It should be declared as a plugin
-inside `conftest.py`, and that makes it available to all tests.
+inside conftest.py, and that makes it available to all tests.
 
 You can measure multiple things in one test, and record each one with a separate
-call to `zenbenchmark`. For example, you could time the bulk loading that happens
+call to zenbenchmark. For example, you could time the bulk loading that happens
 in the test initialization, or measure disk usage after the test query.
 
 """
@@ -115,7 +117,7 @@ class PgBenchRunResult:
             # tps = 309.281539 (without initial connection time)
             if line.startswith("tps = ") and (
                 "(excluding connections establishing)" in line
-                or "(without initial connection time)" in line
+                or "(without initial connection time)"
             ):
                 tps = float(line.split()[2])
 
@@ -135,17 +137,6 @@ class PgBenchRunResult:
 
 @dataclasses.dataclass
 class PgBenchInitResult:
-    REGEX: ClassVar[re.Pattern] = re.compile(  # type: ignore[type-arg]
-        r"done in (\d+\.\d+) s "
-        r"\("
-        r"(?:drop tables (\d+\.\d+) s)?(?:, )?"
-        r"(?:create tables (\d+\.\d+) s)?(?:, )?"
-        r"(?:client-side generate (\d+\.\d+) s)?(?:, )?"
-        r"(?:vacuum (\d+\.\d+) s)?(?:, )?"
-        r"(?:primary keys (\d+\.\d+) s)?(?:, )?"
-        r"\)\."
-    )
-
     total: float
     drop_tables: Optional[float]
     create_tables: Optional[float]
@@ -169,7 +160,18 @@ class PgBenchInitResult:
 
         last_line = stderr.splitlines()[-1]
 
-        if (m := cls.REGEX.match(last_line)) is not None:
+        regex = re.compile(
+            r"done in (\d+\.\d+) s "
+            r"\("
+            r"(?:drop tables (\d+\.\d+) s)?(?:, )?"
+            r"(?:create tables (\d+\.\d+) s)?(?:, )?"
+            r"(?:client-side generate (\d+\.\d+) s)?(?:, )?"
+            r"(?:vacuum (\d+\.\d+) s)?(?:, )?"
+            r"(?:primary keys (\d+\.\d+) s)?(?:, )?"
+            r"\)\."
+        )
+
+        if (m := regex.match(last_line)) is not None:
             total, drop_tables, create_tables, client_side_generate, vacuum, primary_keys = [
                 float(v) for v in m.groups() if v is not None
             ]
@@ -206,7 +208,7 @@ class NeonBenchmarker:
     function by the zenbenchmark fixture
     """
 
-    def __init__(self, property_recorder: Callable[[str, object], None]):
+    def __init__(self, property_recorder):
         # property recorder here is a pytest fixture provided by junitxml module
         # https://docs.pytest.org/en/6.2.x/reference.html#pytest.junitxml.record_property
         self.property_recorder = property_recorder
@@ -234,7 +236,7 @@ class NeonBenchmarker:
         )
 
     @contextmanager
-    def record_duration(self, metric_name: str) -> Iterator[None]:
+    def record_duration(self, metric_name: str):
         """
         Record a duration. Usage:
 
@@ -335,21 +337,21 @@ class NeonBenchmarker:
                     f"{prefix}.{metric}", value, unit="s", report=MetricReport.LOWER_IS_BETTER
                 )
 
-    def get_io_writes(self, pageserver: NeonPageserver) -> int:
+    def get_io_writes(self, pageserver) -> int:
         """
         Fetch the "cumulative # of bytes written" metric from the pageserver
         """
         metric_name = r'libmetrics_disk_io_bytes_total{io_operation="write"}'
         return self.get_int_counter_value(pageserver, metric_name)
 
-    def get_peak_mem(self, pageserver: NeonPageserver) -> int:
+    def get_peak_mem(self, pageserver) -> int:
         """
         Fetch the "maxrss" metric from the pageserver
         """
         metric_name = r"libmetrics_maxrss_kb"
         return self.get_int_counter_value(pageserver, metric_name)
 
-    def get_int_counter_value(self, pageserver: NeonPageserver, metric_name: str) -> int:
+    def get_int_counter_value(self, pageserver, metric_name) -> int:
         """Fetch the value of given int counter from pageserver metrics."""
         # TODO: If we start to collect more of the prometheus metrics in the
         # performance test suite like this, we should refactor this to load and
@@ -363,9 +365,7 @@ class NeonBenchmarker:
         assert matches, f"metric {metric_name} not found"
         return int(round(float(matches.group(1))))
 
-    def get_timeline_size(
-        self, repo_dir: Path, tenant_id: TenantId, timeline_id: TimelineId
-    ) -> int:
+    def get_timeline_size(self, repo_dir: Path, tenant_id: TenantId, timeline_id: TimelineId):
         """
         Calculate the on-disk size of a timeline
         """
@@ -379,9 +379,7 @@ class NeonBenchmarker:
         return totalbytes
 
     @contextmanager
-    def record_pageserver_writes(
-        self, pageserver: NeonPageserver, metric_name: str
-    ) -> Iterator[None]:
+    def record_pageserver_writes(self, pageserver, metric_name):
         """
         Record bytes written by the pageserver during a test.
         """
@@ -398,7 +396,7 @@ class NeonBenchmarker:
 
 
 @pytest.fixture(scope="function")
-def zenbenchmark(record_property: Callable[[str, object], None]) -> Iterator[NeonBenchmarker]:
+def zenbenchmark(record_property) -> Iterator[NeonBenchmarker]:
     """
     This is a python decorator for benchmark fixtures. It contains functions for
     recording measurements, and prints them out at the end.
@@ -407,7 +405,7 @@ def zenbenchmark(record_property: Callable[[str, object], None]) -> Iterator[Neo
     yield benchmarker
 
 
-def pytest_addoption(parser: Parser):
+def pytest_addoption(parser):
     parser.addoption(
         "--out-dir",
         dest="out_dir",
@@ -431,9 +429,7 @@ def get_out_path(target_dir: Path, revision: str) -> Path:
 
 # Hook to print the results at the end
 @pytest.hookimpl(hookwrapper=True)
-def pytest_terminal_summary(
-    terminalreporter: TerminalReporter, exitstatus: int, config: Config
-) -> Iterator[None]:
+def pytest_terminal_summary(terminalreporter: TerminalReporter, exitstatus: int, config: Config):
     yield
     revision = os.getenv("GITHUB_SHA", "local")
     platform = os.getenv("PLATFORM", "local")

test_runner/fixtures/compare_fixtures.py

@@ -1,11 +1,10 @@
 from abc import ABC, abstractmethod
-from contextlib import _GeneratorContextManager, contextmanager
+from contextlib import contextmanager
 
 # Type-related stuff
-from typing import Dict, Iterator, List
+from typing import Dict, List
 
 import pytest
-from _pytest.fixtures import FixtureRequest
 from fixtures.benchmark_fixture import MetricReport, NeonBenchmarker
 from fixtures.neon_fixtures import NeonEnv, PgBin, PgProtocol, RemotePostgres, VanillaPostgres
 from fixtures.pg_stats import PgStatTable
@@ -29,20 +28,19 @@ class PgCompare(ABC):
         pass
 
     @property
-    @abstractmethod
     def zenbenchmark(self) -> NeonBenchmarker:
         pass
 
     @abstractmethod
-    def flush(self):
+    def flush(self) -> None:
         pass
 
     @abstractmethod
-    def report_peak_memory_use(self):
+    def report_peak_memory_use(self) -> None:
         pass
 
     @abstractmethod
-    def report_size(self):
+    def report_size(self) -> None:
         pass
 
     @contextmanager
@@ -56,7 +54,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]):
         init_data = self._retrieve_pg_stats(pg_stats)
 
         yield
@@ -86,11 +84,7 @@ class NeonCompare(PgCompare):
     """PgCompare interface for the neon stack."""
 
     def __init__(
-        self,
-        zenbenchmark: NeonBenchmarker,
-        neon_simple_env: NeonEnv,
-        pg_bin: PgBin,
-        branch_name: str,
+        self, zenbenchmark: NeonBenchmarker, neon_simple_env: NeonEnv, pg_bin: PgBin, branch_name
     ):
         self.env = neon_simple_env
         self._zenbenchmark = zenbenchmark
@@ -103,15 +97,15 @@ class NeonCompare(PgCompare):
         self.timeline = self.pg.safe_psql("SHOW neon.timeline_id")[0][0]
 
     @property
-    def pg(self) -> PgProtocol:
+    def pg(self):
         return self._pg
 
     @property
-    def zenbenchmark(self) -> NeonBenchmarker:
+    def zenbenchmark(self):
         return self._zenbenchmark
 
     @property
-    def pg_bin(self) -> PgBin:
+    def pg_bin(self):
         return self._pg_bin
 
     def flush(self):
@@ -120,7 +114,7 @@ class NeonCompare(PgCompare):
     def compact(self):
         self.pageserver_http_client.timeline_compact(self.env.initial_tenant, self.timeline)
 
-    def report_peak_memory_use(self):
+    def report_peak_memory_use(self) -> None:
         self.zenbenchmark.record(
             "peak_mem",
             self.zenbenchmark.get_peak_mem(self.env.pageserver) / 1024,
@@ -128,7 +122,7 @@ class NeonCompare(PgCompare):
             report=MetricReport.LOWER_IS_BETTER,
         )
 
-    def report_size(self):
+    def report_size(self) -> None:
         timeline_size = self.zenbenchmark.get_timeline_size(
             self.env.repo_dir, self.env.initial_tenant, self.timeline
         )
@@ -150,17 +144,17 @@ class NeonCompare(PgCompare):
             "num_files_uploaded", total_files, "", report=MetricReport.LOWER_IS_BETTER
         )
 
-    def record_pageserver_writes(self, out_name: str) -> _GeneratorContextManager[None]:
+    def record_pageserver_writes(self, out_name):
         return self.zenbenchmark.record_pageserver_writes(self.env.pageserver, out_name)
 
-    def record_duration(self, out_name: str) -> _GeneratorContextManager[None]:
+    def record_duration(self, out_name):
         return self.zenbenchmark.record_duration(out_name)
 
 
 class VanillaCompare(PgCompare):
     """PgCompare interface for vanilla postgres."""
 
-    def __init__(self, zenbenchmark: NeonBenchmarker, vanilla_pg: VanillaPostgres):
+    def __init__(self, zenbenchmark, vanilla_pg: VanillaPostgres):
         self._pg = vanilla_pg
         self._zenbenchmark = zenbenchmark
         vanilla_pg.configure(
@@ -176,24 +170,24 @@ class VanillaCompare(PgCompare):
         self.cur = self.conn.cursor()
 
     @property
-    def pg(self) -> PgProtocol:
+    def pg(self):
         return self._pg
 
     @property
-    def zenbenchmark(self) -> NeonBenchmarker:
+    def zenbenchmark(self):
         return self._zenbenchmark
 
     @property
-    def pg_bin(self) -> PgBin:
+    def pg_bin(self):
         return self._pg.pg_bin
 
     def flush(self):
         self.cur.execute("checkpoint")
 
-    def report_peak_memory_use(self):
+    def report_peak_memory_use(self) -> None:
         pass  # TODO find something
 
-    def report_size(self):
+    def report_size(self) -> None:
         data_size = self.pg.get_subdir_size("base")
         self.zenbenchmark.record(
             "data_size", data_size / (1024 * 1024), "MB", report=MetricReport.LOWER_IS_BETTER
@@ -204,17 +198,17 @@ class VanillaCompare(PgCompare):
         )
 
     @contextmanager
-    def record_pageserver_writes(self, out_name: str) -> Iterator[None]:
+    def record_pageserver_writes(self, out_name):
         yield  # Do nothing
 
-    def record_duration(self, out_name: str) -> _GeneratorContextManager[None]:
+    def record_duration(self, out_name):
         return self.zenbenchmark.record_duration(out_name)
 
 
 class RemoteCompare(PgCompare):
     """PgCompare interface for a remote postgres instance."""
 
-    def __init__(self, zenbenchmark: NeonBenchmarker, remote_pg: RemotePostgres):
+    def __init__(self, zenbenchmark, remote_pg: RemotePostgres):
         self._pg = remote_pg
         self._zenbenchmark = zenbenchmark
 
@@ -223,60 +217,55 @@ class RemoteCompare(PgCompare):
         self.cur = self.conn.cursor()
 
     @property
-    def pg(self) -> PgProtocol:
+    def pg(self):
         return self._pg
 
     @property
-    def zenbenchmark(self) -> NeonBenchmarker:
+    def zenbenchmark(self):
         return self._zenbenchmark
 
     @property
-    def pg_bin(self) -> PgBin:
+    def pg_bin(self):
         return self._pg.pg_bin
 
     def flush(self):
         # TODO: flush the remote pageserver
         pass
 
-    def report_peak_memory_use(self):
+    def report_peak_memory_use(self) -> None:
         # TODO: get memory usage from remote pageserver
         pass
 
-    def report_size(self):
+    def report_size(self) -> None:
         # TODO: get storage size from remote pageserver
         pass
 
     @contextmanager
-    def record_pageserver_writes(self, out_name: str) -> Iterator[None]:
+    def record_pageserver_writes(self, out_name):
         yield  # Do nothing
 
-    def record_duration(self, out_name: str) -> _GeneratorContextManager[None]:
+    def record_duration(self, out_name):
         return self.zenbenchmark.record_duration(out_name)
 
 
 @pytest.fixture(scope="function")
-def neon_compare(
-    request: FixtureRequest,
-    zenbenchmark: NeonBenchmarker,
-    pg_bin: PgBin,
-    neon_simple_env: NeonEnv,
-) -> NeonCompare:
+def neon_compare(request, zenbenchmark, pg_bin, neon_simple_env) -> NeonCompare:
     branch_name = request.node.name
     return NeonCompare(zenbenchmark, neon_simple_env, pg_bin, branch_name)
 
 
 @pytest.fixture(scope="function")
-def vanilla_compare(zenbenchmark: NeonBenchmarker, vanilla_pg: VanillaPostgres) -> VanillaCompare:
+def vanilla_compare(zenbenchmark, vanilla_pg) -> VanillaCompare:
     return VanillaCompare(zenbenchmark, vanilla_pg)
 
 
 @pytest.fixture(scope="function")
-def remote_compare(zenbenchmark: NeonBenchmarker, remote_pg: RemotePostgres) -> RemoteCompare:
+def remote_compare(zenbenchmark, remote_pg) -> RemoteCompare:
     return RemoteCompare(zenbenchmark, remote_pg)
 
 
 @pytest.fixture(params=["vanilla_compare", "neon_compare"], ids=["vanilla", "neon"])
-def neon_with_baseline(request: FixtureRequest) -> PgCompare:
+def neon_with_baseline(request) -> PgCompare:
     """Parameterized fixture that helps compare neon against vanilla postgres.
 
     A test that uses this fixture turns into a parameterized test that runs against:
@@ -297,6 +286,8 @@ def neon_with_baseline(request: FixtureRequest) -> PgCompare:
     implementation-specific logic is widely useful across multiple tests, it might
     make sense to add methods to the PgCompare class.
     """
-    fixture = request.getfixturevalue(request.param)  # type: ignore
-    assert isinstance(fixture, PgCompare), f"test error: fixture {fixture} is not PgCompare"
-    return fixture
+    fixture = request.getfixturevalue(request.param)
+    if isinstance(fixture, PgCompare):
+        return fixture
+    else:
+        raise AssertionError(f"test error: fixture {request.param} is not PgCompare")

test_runner/fixtures/metrics.py

@@ -1,5 +1,5 @@
 from collections import defaultdict
-from typing import Dict, List, Optional, Tuple
+from typing import Dict, List
 
 from prometheus_client.parser import text_string_to_metric_families
 from prometheus_client.samples import Sample
@@ -23,13 +23,13 @@ class Metrics:
                 pass
         return res
 
-    def query_one(self, name: str, filter: Optional[Dict[str, str]] = None) -> Sample:
-        res = self.query_all(name, filter or {})
+    def query_one(self, name: str, filter: Dict[str, str] = {}) -> Sample:
+        res = self.query_all(name, filter)
         assert len(res) == 1, f"expected single sample for {name} {filter}, found {res}"
         return res[0]
 
 
-def parse_metrics(text: str, name: str = "") -> Metrics:
+def parse_metrics(text: str, name: str = ""):
     metrics = Metrics(name)
     gen = text_string_to_metric_families(text)
     for family in gen:
@@ -39,7 +39,7 @@ def parse_metrics(text: str, name: str = "") -> Metrics:
     return metrics
 
 
-PAGESERVER_PER_TENANT_METRICS: Tuple[str, ...] = (
+PAGESERVER_PER_TENANT_METRICS = [
     "pageserver_current_logical_size",
     "pageserver_current_physical_size",
     "pageserver_getpage_reconstruct_seconds_bucket",
@@ -62,4 +62,4 @@ PAGESERVER_PER_TENANT_METRICS: Tuple[str, ...] = (
     "pageserver_wait_lsn_seconds_sum",
     "pageserver_created_persistent_files_total",
     "pageserver_written_persistent_bytes_total",
-)
+]

test_runner/fixtures/neon_fixtures.py

@@ -19,8 +19,7 @@ from dataclasses import dataclass, field
 from enum import Flag, auto
 from functools import cached_property
 from pathlib import Path
-from types import TracebackType
-from typing import Any, Dict, Iterator, List, Optional, Tuple, Type, Union, cast
+from typing import Any, Dict, Iterator, List, Optional, Tuple, Union, cast
 
 import asyncpg
 import backoff  # type: ignore
@@ -29,18 +28,16 @@ import jwt
 import psycopg2
 import pytest
 import requests
-from _pytest.config import Config
-from _pytest.fixtures import FixtureRequest
 from fixtures.log_helper import log
 from fixtures.types import Lsn, TenantId, TimelineId
-from fixtures.utils import Fn, allure_attach_from_dir, etcd_path, get_self_dir, subprocess_capture
 
 # Type-related stuff
 from psycopg2.extensions import connection as PgConnection
-from psycopg2.extensions import cursor as PgCursor
 from psycopg2.extensions import make_dsn, parse_dsn
 from typing_extensions import Literal
 
+from .utils import Fn, allure_attach_from_dir, etcd_path, get_self_dir, subprocess_capture
+
 """
 This file contains pytest fixtures. A fixture is a test resource that can be
 summoned by placing its name in the test's arguments.
@@ -60,16 +57,25 @@ put directly-importable functions into utils.py or another separate file.
 
 Env = Dict[str, str]
 
-DEFAULT_OUTPUT_DIR: str = "test_output"
-DEFAULT_BRANCH_NAME: str = "main"
-DEFAULT_PG_VERSION_DEFAULT: str = "14"
+DEFAULT_OUTPUT_DIR = "test_output"
+DEFAULT_BRANCH_NAME = "main"
+DEFAULT_PG_VERSION_DEFAULT = "14"
 
-BASE_PORT: int = 15000
-WORKER_PORT_NUM: int = 1000
+BASE_PORT = 15000
+WORKER_PORT_NUM = 1000
 
 
-def pytest_configure(config: Config):
+# These are set in pytest_configure()
+base_dir = ""
+neon_binpath = ""
+pg_distrib_dir = ""
+top_output_dir = ""
+default_pg_version = ""
+
+
+def pytest_configure(config):
     """
+    Ensure that no unwanted daemons are running before we start testing.
     Check that we do not overflow available ports range.
     """
 
@@ -79,92 +85,70 @@ def pytest_configure(config: Config):
     ):  # do not use ephemeral ports
         raise Exception("Too many workers configured. Cannot distribute ports for services.")
 
-
-@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
+    global base_dir
+    base_dir = os.path.normpath(os.path.join(get_self_dir(), "../.."))
     log.info(f"base_dir is {base_dir}")
 
-    yield base_dir
+    # Compute the top-level directory for all tests.
+    global top_output_dir
+    env_test_output = os.environ.get("TEST_OUTPUT")
+    if env_test_output is not None:
+        top_output_dir = env_test_output
+    else:
+        top_output_dir = os.path.join(base_dir, DEFAULT_OUTPUT_DIR)
+    Path(top_output_dir).mkdir(exist_ok=True)
 
+    # Find the postgres installation.
+    global default_pg_version
+    log.info(f"default_pg_version is {default_pg_version}")
+    env_default_pg_version = os.environ.get("DEFAULT_PG_VERSION")
+    if env_default_pg_version:
+        default_pg_version = env_default_pg_version
+        log.info(f"default_pg_version is set to {default_pg_version}")
+    else:
+        default_pg_version = DEFAULT_PG_VERSION_DEFAULT
+
+    global pg_distrib_dir
+
+    env_postgres_bin = os.environ.get("POSTGRES_DISTRIB_DIR")
+    if env_postgres_bin:
+        pg_distrib_dir = env_postgres_bin
+    else:
+        pg_distrib_dir = os.path.normpath(os.path.join(base_dir, "pg_install"))
+
+    log.info(f"pg_distrib_dir is {pg_distrib_dir}")
+    psql_bin_path = os.path.join(pg_distrib_dir, "v{}".format(default_pg_version), "bin/psql")
+    postgres_bin_path = os.path.join(
+        pg_distrib_dir, "v{}".format(default_pg_version), "bin/postgres"
+    )
+
+    if os.getenv("REMOTE_ENV"):
+        # When testing against a remote server, we only need the client binary.
+        if not os.path.exists(psql_bin_path):
+            raise Exception('psql not found at "{}"'.format(psql_bin_path))
+    else:
+        if not os.path.exists(postgres_bin_path):
+            raise Exception('postgres not found at "{}"'.format(postgres_bin_path))
 
-@pytest.fixture(scope="session")
-def neon_binpath(base_dir: Path) -> Iterator[Path]:
     if os.getenv("REMOTE_ENV"):
         # we are in remote env and do not have neon binaries locally
         # this is the case for benchmarks run on self-hosted runner
         return
-
     # Find the neon binaries.
-    if env_neon_bin := os.environ.get("NEON_BIN"):
-        binpath = Path(env_neon_bin)
+    global neon_binpath
+    env_neon_bin = os.environ.get("NEON_BIN")
+    if env_neon_bin:
+        neon_binpath = env_neon_bin
     else:
         build_type = os.environ.get("BUILD_TYPE", "debug")
-        binpath = base_dir / "target" / build_type
-    log.info(f"neon_binpath is {binpath}")
-
-    if not (binpath / "pageserver").exists():
-        raise Exception(f"neon binaries not found at '{binpath}'")
-
-    yield binpath
+        neon_binpath = os.path.join(base_dir, "target", build_type)
+    log.info(f"neon_binpath is {neon_binpath}")
+    if not os.path.exists(os.path.join(neon_binpath, "pageserver")):
+        raise Exception('neon binaries not found at "{}"'.format(neon_binpath))
 
 
-@pytest.fixture(scope="session")
-def pg_distrib_dir(base_dir: Path) -> Iterator[Path]:
-    if env_postgres_bin := os.environ.get("POSTGRES_DISTRIB_DIR"):
-        distrib_dir = Path(env_postgres_bin).resolve()
-    else:
-        distrib_dir = base_dir / "pg_install"
-
-    log.info(f"pg_distrib_dir is {distrib_dir}")
-    yield distrib_dir
-
-
-@pytest.fixture(scope="session")
-def top_output_dir(base_dir: Path) -> Iterator[Path]:
-    # Compute the top-level directory for all tests.
-    if env_test_output := os.environ.get("TEST_OUTPUT"):
-        output_dir = Path(env_test_output).resolve()
-    else:
-        output_dir = base_dir / DEFAULT_OUTPUT_DIR
-    output_dir.mkdir(exist_ok=True)
-
-    log.info(f"top_output_dir is {output_dir}")
-    yield output_dir
-
-
-@pytest.fixture(scope="session")
-def pg_version() -> Iterator[str]:
-    if env_default_pg_version := os.environ.get("DEFAULT_PG_VERSION"):
-        version = env_default_pg_version
-    else:
-        version = DEFAULT_PG_VERSION_DEFAULT
-
-    log.info(f"pg_version is {version}")
-    yield version
-
-
-@pytest.fixture(scope="session")
-def versioned_pg_distrib_dir(pg_distrib_dir: Path, pg_version: str) -> Iterator[Path]:
-    versioned_dir = pg_distrib_dir / f"v{pg_version}"
-
-    psql_bin_path = versioned_dir / "bin/psql"
-    postgres_bin_path = versioned_dir / "bin/postgres"
-
-    if os.getenv("REMOTE_ENV"):
-        # When testing against a remote server, we only need the client binary.
-        if not psql_bin_path.exists():
-            raise Exception(f"psql not found at '{psql_bin_path}'")
-    else:
-        if not postgres_bin_path.exists():
-            raise Exception(f"postgres not found at '{postgres_bin_path}'")
-
-    log.info(f"versioned_pg_distrib_dir is {versioned_dir}")
-    yield versioned_dir
-
-
-def shareable_scope(fixture_name: str, config: Config) -> Literal["session", "function"]:
+def shareable_scope(fixture_name, config) -> Literal["session", "function"]:
     """Return either session of function scope, depending on TEST_SHARED_FIXTURES envvar.
 
     This function can be used as a scope like this:
@@ -176,7 +160,7 @@ def shareable_scope(fixture_name: str, config: Config) -> Literal["session", "fu
 
 
 @pytest.fixture(scope="session")
-def worker_seq_no(worker_id: str) -> int:
+def worker_seq_no(worker_id: str):
     # worker_id is a pytest-xdist fixture
     # it can be master or gw<number>
     # parse it to always get a number
@@ -187,7 +171,7 @@ def worker_seq_no(worker_id: str) -> int:
 
 
 @pytest.fixture(scope="session")
-def worker_base_port(worker_seq_no: int) -> int:
+def worker_base_port(worker_seq_no: int):
     # so we divide ports in ranges of 100 ports
     # so workers have disjoint set of ports for services
     return BASE_PORT + worker_seq_no * WORKER_PORT_NUM
@@ -231,78 +215,42 @@ 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] = {}
 
     def get_port(self) -> int:
         for port in self.iterator:
             if can_bind("localhost", port):
                 return port
-        raise RuntimeError(
-            "port range configured for test is exhausted, consider enlarging the range"
-        )
-
-    def replace_with_new_port(self, value: Union[int, str]) -> Union[int, str]:
-        """
-        Returns a new port for a port number in a string (like "localhost:1234") or int.
-        Replacements are memorised, so a substitution for the same port is always the same.
-        """
-
-        # TODO: replace with structural pattern matching for Python >= 3.10
-        if isinstance(value, int):
-            return self._replace_port_int(value)
-
-        if isinstance(value, str):
-            return self._replace_port_str(value)
-
-        raise TypeError(f"unsupported type {type(value)} of {value=}")
-
-    def _replace_port_int(self, value: int) -> int:
-        known_port = self.port_map.get(value)
-        if known_port is None:
-            known_port = self.port_map[value] = self.get_port()
-
-        return known_port
-
-    def _replace_port_str(self, value: str) -> str:
-        # Use regex to find port in a string
-        # urllib.parse.urlparse produces inconvenient results for cases without scheme like "localhost:5432"
-        # See https://bugs.python.org/issue27657
-        ports = re.findall(r":(\d+)(?:/|$)", value)
-        assert len(ports) == 1, f"can't find port in {value}"
-        port_int = int(ports[0])
-
-        return value.replace(f":{port_int}", f":{self._replace_port_int(port_int)}")
+        else:
+            raise RuntimeError(
+                "port range configured for test is exhausted, consider enlarging the range"
+            )
 
 
 @pytest.fixture(scope="session")
-def port_distributor(worker_base_port: int) -> PortDistributor:
+def port_distributor(worker_base_port):
     return PortDistributor(base_port=worker_base_port, port_number=WORKER_PORT_NUM)
 
 
 @pytest.fixture(scope="session")
-def default_broker(
-    request: FixtureRequest, port_distributor: PortDistributor, top_output_dir: Path
-) -> Iterator[Etcd]:
+def default_broker(request: Any, port_distributor: PortDistributor):
     client_port = port_distributor.get_port()
     # multiple pytest sessions could get launched in parallel, get them different datadirs
-    etcd_datadir = get_test_output_dir(request, top_output_dir) / f"etcd_datadir_{client_port}"
-    etcd_datadir.mkdir(exist_ok=True, parents=True)
+    etcd_datadir = os.path.join(get_test_output_dir(request), f"etcd_datadir_{client_port}")
+    Path(etcd_datadir).mkdir(exist_ok=True, parents=True)
 
-    broker = Etcd(
-        datadir=str(etcd_datadir), port=client_port, peer_port=port_distributor.get_port()
-    )
+    broker = Etcd(datadir=etcd_datadir, port=client_port, peer_port=port_distributor.get_port())
     yield broker
     broker.stop()
-    allure_attach_from_dir(etcd_datadir)
+    allure_attach_from_dir(Path(etcd_datadir))
 
 
 @pytest.fixture(scope="session")
-def run_id() -> Iterator[uuid.UUID]:
+def run_id():
     yield uuid.uuid4()
 
 
 @pytest.fixture(scope="session")
-def mock_s3_server(port_distributor: PortDistributor) -> Iterator[MockS3Server]:
+def mock_s3_server(port_distributor: PortDistributor):
     mock_s3_server = MockS3Server(port_distributor.get_port())
     yield mock_s3_server
     mock_s3_server.kill()
@@ -311,16 +259,16 @@ def mock_s3_server(port_distributor: PortDistributor) -> Iterator[MockS3Server]:
 class PgProtocol:
     """Reusable connection logic"""
 
-    def __init__(self, **kwargs: Any):
+    def __init__(self, **kwargs):
         self.default_options = kwargs
 
-    def connstr(self, **kwargs: Any) -> str:
+    def connstr(self, **kwargs) -> str:
         """
         Build a libpq connection string for the Postgres instance.
         """
         return str(make_dsn(**self.conn_options(**kwargs)))
 
-    def conn_options(self, **kwargs: Any) -> Dict[str, Any]:
+    def conn_options(self, **kwargs):
         """
         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.
@@ -342,7 +290,7 @@ class PgProtocol:
         return result
 
     # autocommit=True here by default because that's what we need most of the time
-    def connect(self, autocommit: bool = True, **kwargs: Any) -> PgConnection:
+    def connect(self, autocommit=True, **kwargs) -> PgConnection:
         """
         Connect to the node.
         Returns psycopg2's connection object.
@@ -355,7 +303,7 @@ class PgProtocol:
         return conn
 
     @contextmanager
-    def cursor(self, autocommit: bool = True, **kwargs: Any) -> Iterator[PgCursor]:
+    def cursor(self, autocommit=True, **kwargs):
         """
         Shorthand for pg.connect().cursor().
         The cursor and connection are closed when the context is exited.
@@ -363,7 +311,7 @@ class PgProtocol:
         with closing(self.connect(autocommit=autocommit, **kwargs)) as conn:
             yield conn.cursor()
 
-    async def connect_async(self, **kwargs: Any) -> asyncpg.Connection:
+    async def connect_async(self, **kwargs) -> asyncpg.Connection:
         """
         Connect to the node from async python.
         Returns asyncpg's connection object.
@@ -417,10 +365,10 @@ class PgProtocol:
 
 @dataclass
 class AuthKeys:
-    pub: str
-    priv: str
+    pub: bytes
+    priv: bytes
 
-    def generate_management_token(self) -> str:
+    def generate_management_token(self):
         token = jwt.encode({"scope": "pageserverapi"}, self.priv, algorithm="RS256")
 
         # jwt.encode can return 'bytes' or 'str', depending on Python version or type
@@ -431,11 +379,9 @@ class AuthKeys:
 
         return token
 
-    def generate_tenant_token(self, tenant_id: TenantId) -> str:
+    def generate_tenant_token(self, tenant_id):
         token = jwt.encode(
-            {"scope": "tenant", "tenant_id": str(tenant_id)},
-            self.priv,
-            algorithm="RS256",
+            {"scope": "tenant", "tenant_id": str(tenant_id)}, self.priv, algorithm="RS256"
         )
 
         if isinstance(token, bytes):
@@ -491,7 +437,7 @@ class MockS3Server:
 
 
 @enum.unique
-class RemoteStorageKind(str, enum.Enum):
+class RemoteStorageKind(enum.Enum):
     LOCAL_FS = "local_fs"
     MOCK_S3 = "mock_s3"
     REAL_S3 = "real_s3"
@@ -535,7 +481,7 @@ RemoteStorage = Union[LocalFsStorage, S3Storage]
 
 
 # serialize as toml inline table
-def remote_storage_to_toml_inline_table(remote_storage: RemoteStorage) -> str:
+def remote_storage_to_toml_inline_table(remote_storage):
     if isinstance(remote_storage, LocalFsStorage):
         remote_storage_config = f"local_path='{remote_storage.root}'"
     elif isinstance(remote_storage, S3Storage):
@@ -575,9 +521,6 @@ class NeonEnvBuilder:
         broker: Etcd,
         run_id: uuid.UUID,
         mock_s3_server: MockS3Server,
-        neon_binpath: Path,
-        pg_distrib_dir: Path,
-        pg_version: str,
         remote_storage: Optional[RemoteStorage] = None,
         remote_storage_users: RemoteStorageUsers = RemoteStorageUsers.PAGESERVER,
         pageserver_config_override: Optional[str] = None,
@@ -588,7 +531,7 @@ class NeonEnvBuilder:
         safekeepers_enable_fsync: bool = False,
         auth_enabled: bool = False,
         rust_log_override: Optional[str] = None,
-        default_branch_name: str = DEFAULT_BRANCH_NAME,
+        default_branch_name=DEFAULT_BRANCH_NAME,
     ):
         self.repo_dir = repo_dir
         self.rust_log_override = rust_log_override
@@ -607,9 +550,7 @@ class NeonEnvBuilder:
         self.env: Optional[NeonEnv] = None
         self.remote_storage_prefix: Optional[str] = None
         self.keep_remote_storage_contents: bool = True
-        self.neon_binpath = neon_binpath
-        self.pg_distrib_dir = pg_distrib_dir
-        self.pg_version = pg_version
+        self.pg_version = default_pg_version
 
     def init(self) -> NeonEnv:
         # Cannot create more than one environment from one builder
@@ -642,7 +583,7 @@ class NeonEnvBuilder:
         else:
             raise RuntimeError(f"Unknown storage type: {remote_storage_kind}")
 
-    def enable_local_fs_remote_storage(self, force_enable: bool = True):
+    def enable_local_fs_remote_storage(self, force_enable=True):
         """
         Sets up the pageserver to use the local fs at the `test_dir/local_fs_remote_storage` path.
         Errors, if the pageserver has some remote storage configuration already, unless `force_enable` is not set to `True`.
@@ -650,7 +591,7 @@ class NeonEnvBuilder:
         assert force_enable or self.remote_storage is None, "remote storage is enabled already"
         self.remote_storage = LocalFsStorage(Path(self.repo_dir / "local_fs_remote_storage"))
 
-    def enable_mock_s3_remote_storage(self, bucket_name: str, force_enable: bool = True):
+    def enable_mock_s3_remote_storage(self, bucket_name: str, force_enable=True):
         """
         Sets up the pageserver to use the S3 mock server, creates the bucket, if it's not present already.
         Starts up the mock server, if that does not run yet.
@@ -677,7 +618,7 @@ class NeonEnvBuilder:
             secret_key=self.mock_s3_server.secret_key(),
         )
 
-    def enable_real_s3_remote_storage(self, test_name: str, force_enable: bool = True):
+    def enable_real_s3_remote_storage(self, test_name: str, force_enable=True):
         """
         Sets up configuration to use real s3 endpoint without mock server
         """
@@ -765,15 +706,10 @@ class NeonEnvBuilder:
 
         log.info("deleted %s objects from remote storage", cnt)
 
-    def __enter__(self) -> "NeonEnvBuilder":
+    def __enter__(self):
         return self
 
-    def __exit__(
-        self,
-        exc_type: Optional[Type[BaseException]],
-        exc_value: Optional[BaseException],
-        traceback: Optional[TracebackType],
-    ):
+    def __exit__(self, exc_type, exc_value, traceback):
         # Stop all the nodes.
         if self.env:
             log.info("Cleaning up all storage and compute nodes")
@@ -830,8 +766,6 @@ class NeonEnv:
         self.remote_storage = config.remote_storage
         self.remote_storage_users = config.remote_storage_users
         self.pg_version = config.pg_version
-        self.neon_binpath = config.neon_binpath
-        self.pg_distrib_dir = config.pg_distrib_dir
 
         # generate initial tenant ID here instead of letting 'neon init' generate it,
         # so that we don't need to dig it out of the config file afterwards.
@@ -920,14 +854,14 @@ class NeonEnv:
 
     def get_safekeeper_connstrs(self) -> str:
         """Get list of safekeeper endpoints suitable for safekeepers GUC"""
-        return ",".join(f"localhost:{wa.port.pg}" for wa in self.safekeepers)
+        return ",".join([f"localhost:{wa.port.pg}" for wa in self.safekeepers])
 
     def timeline_dir(self, tenant_id: TenantId, timeline_id: TimelineId) -> Path:
         """Get a timeline directory's path based on the repo directory of the test environment"""
         return self.repo_dir / "tenants" / str(tenant_id) / "timelines" / str(timeline_id)
 
     def get_pageserver_version(self) -> str:
-        bin_pageserver = str(self.neon_binpath / "pageserver")
+        bin_pageserver = os.path.join(str(neon_binpath), "pageserver")
         res = subprocess.run(
             [bin_pageserver, "--version"],
             check=True,
@@ -939,22 +873,18 @@ class NeonEnv:
 
     @cached_property
     def auth_keys(self) -> AuthKeys:
-        pub = (Path(self.repo_dir) / "auth_public_key.pem").read_text()
-        priv = (Path(self.repo_dir) / "auth_private_key.pem").read_text()
+        pub = (Path(self.repo_dir) / "auth_public_key.pem").read_bytes()
+        priv = (Path(self.repo_dir) / "auth_private_key.pem").read_bytes()
         return AuthKeys(pub=pub, priv=priv)
 
 
 @pytest.fixture(scope=shareable_scope)
 def _shared_simple_env(
-    request: FixtureRequest,
+    request: Any,
     port_distributor: PortDistributor,
     mock_s3_server: MockS3Server,
     default_broker: Etcd,
     run_id: uuid.UUID,
-    top_output_dir: Path,
-    neon_binpath: Path,
-    pg_distrib_dir: Path,
-    pg_version: str,
 ) -> Iterator[NeonEnv]:
     """
     # Internal fixture backing the `neon_simple_env` fixture. If TEST_SHARED_FIXTURES
@@ -963,20 +893,17 @@ def _shared_simple_env(
 
     if os.environ.get("TEST_SHARED_FIXTURES") is None:
         # Create the environment in the per-test output directory
-        repo_dir = get_test_output_dir(request, top_output_dir) / "repo"
+        repo_dir = os.path.join(get_test_output_dir(request), "repo")
     else:
         # We're running shared fixtures. Share a single directory.
-        repo_dir = top_output_dir / "shared_repo"
+        repo_dir = os.path.join(str(top_output_dir), "shared_repo")
         shutil.rmtree(repo_dir, ignore_errors=True)
 
     with NeonEnvBuilder(
-        repo_dir=repo_dir,
+        repo_dir=Path(repo_dir),
         port_distributor=port_distributor,
         broker=default_broker,
         mock_s3_server=mock_s3_server,
-        neon_binpath=neon_binpath,
-        pg_distrib_dir=pg_distrib_dir,
-        pg_version=pg_version,
         run_id=run_id,
     ) as builder:
         env = builder.init_start()
@@ -1004,12 +931,9 @@ def neon_simple_env(_shared_simple_env: NeonEnv) -> Iterator[NeonEnv]:
 
 @pytest.fixture(scope="function")
 def neon_env_builder(
-    test_output_dir: str,
+    test_output_dir,
     port_distributor: PortDistributor,
     mock_s3_server: MockS3Server,
-    neon_binpath: Path,
-    pg_distrib_dir: Path,
-    pg_version: str,
     default_broker: Etcd,
     run_id: uuid.UUID,
 ) -> Iterator[NeonEnvBuilder]:
@@ -1034,9 +958,6 @@ def neon_env_builder(
         repo_dir=Path(repo_dir),
         port_distributor=port_distributor,
         mock_s3_server=mock_s3_server,
-        neon_binpath=neon_binpath,
-        pg_distrib_dir=pg_distrib_dir,
-        pg_version=pg_version,
         broker=default_broker,
         run_id=run_id,
     ) as builder:
@@ -1070,7 +991,7 @@ class PageserverHttpClient(requests.Session):
     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]]) -> None:
         self.is_testing_enabled_or_skip()
 
         if isinstance(config_strings, tuple):
@@ -1200,6 +1121,7 @@ class PageserverHttpClient(requests.Session):
         self.verbose_error(res)
         res_json = res.json()
         assert res_json is None
+        return res_json
 
     def timeline_gc(
         self, tenant_id: TenantId, timeline_id: TimelineId, gc_horizon: Optional[int]
@@ -1231,6 +1153,7 @@ class PageserverHttpClient(requests.Session):
         self.verbose_error(res)
         res_json = res.json()
         assert res_json is None
+        return res_json
 
     def timeline_get_lsn_by_timestamp(
         self, tenant_id: TenantId, timeline_id: TimelineId, timestamp
@@ -1256,6 +1179,7 @@ class PageserverHttpClient(requests.Session):
         self.verbose_error(res)
         res_json = res.json()
         assert res_json is None
+        return res_json
 
     def get_metrics(self) -> str:
         res = self.get(f"http://localhost:{self.port}/metrics")
@@ -1269,10 +1193,13 @@ class PageserverPort:
     http: int
 
 
-CREATE_TIMELINE_ID_EXTRACTOR: re.Pattern = re.compile(  # type: ignore[type-arg]
+CREATE_TIMELINE_ID_EXTRACTOR = re.compile(
     r"^Created timeline '(?P<timeline_id>[^']+)'", re.MULTILINE
 )
-TIMELINE_DATA_EXTRACTOR: re.Pattern = re.compile(  # type: ignore[type-arg]
+CREATE_TIMELINE_ID_EXTRACTOR = re.compile(
+    r"^Created timeline '(?P<timeline_id>[^']+)'", re.MULTILINE
+)
+TIMELINE_DATA_EXTRACTOR = re.compile(
     r"\s?(?P<branch_name>[^\s]+)\s\[(?P<timeline_id>[^\]]+)\]", re.MULTILINE
 )
 
@@ -1313,7 +1240,7 @@ class AbstractNeonCli(abc.ABC):
         assert type(arguments) == list
         assert type(self.COMMAND) == str
 
-        bin_neon = str(self.env.neon_binpath / self.COMMAND)
+        bin_neon = os.path.join(str(neon_binpath), self.COMMAND)
 
         args = [bin_neon] + arguments
         log.info('Running command "{}"'.format(" ".join(args)))
@@ -1321,7 +1248,7 @@ class AbstractNeonCli(abc.ABC):
 
         env_vars = os.environ.copy()
         env_vars["NEON_REPO_DIR"] = str(self.env.repo_dir)
-        env_vars["POSTGRES_DISTRIB_DIR"] = str(self.env.pg_distrib_dir)
+        env_vars["POSTGRES_DISTRIB_DIR"] = str(pg_distrib_dir)
         if self.env.rust_log_override is not None:
             env_vars["RUST_LOG"] = self.env.rust_log_override
         for (extra_env_key, extra_env_value) in (extra_env_vars or {}).items():
@@ -1565,7 +1492,7 @@ class NeonCli(AbstractNeonCli):
 
     def pageserver_start(
         self,
-        overrides: Tuple[str, ...] = (),
+        overrides=(),
     ) -> "subprocess.CompletedProcess[str]":
         start_args = ["pageserver", "start", *overrides]
         append_pageserver_param_overrides(
@@ -1723,7 +1650,7 @@ class NeonPageserver(PgProtocol):
         self.config_override = config_override
         self.version = env.get_pageserver_version()
 
-    def start(self, overrides: Tuple[str, ...] = ()) -> "NeonPageserver":
+    def start(self, overrides=()) -> "NeonPageserver":
         """
         Start the page server.
         `overrides` allows to add some config to this pageserver start.
@@ -1735,7 +1662,7 @@ class NeonPageserver(PgProtocol):
         self.running = True
         return self
 
-    def stop(self, immediate: bool = False) -> "NeonPageserver":
+    def stop(self, immediate=False) -> "NeonPageserver":
         """
         Stop the page server.
         Returns self.
@@ -1745,15 +1672,10 @@ class NeonPageserver(PgProtocol):
             self.running = False
         return self
 
-    def __enter__(self) -> "NeonPageserver":
+    def __enter__(self):
         return self
 
-    def __exit__(
-        self,
-        exc_type: Optional[Type[BaseException]],
-        exc: Optional[BaseException],
-        tb: Optional[TracebackType],
-    ):
+    def __exit__(self, exc_type, exc, tb):
         self.stop(immediate=True)
 
     def is_testing_enabled_or_skip(self):
@@ -1801,17 +1723,17 @@ def append_pageserver_param_overrides(
 class PgBin:
     """A helper class for executing postgres binaries"""
 
-    def __init__(self, log_dir: Path, pg_distrib_dir: Path, pg_version: str):
+    def __init__(self, log_dir: Path, pg_version: str):
         self.log_dir = log_dir
         self.pg_version = pg_version
-        self.pg_bin_path = pg_distrib_dir / f"v{pg_version}" / "bin"
-        self.pg_lib_dir = pg_distrib_dir / f"v{pg_version}" / "lib"
+        self.pg_bin_path = os.path.join(str(pg_distrib_dir), "v{}".format(pg_version), "bin")
+        self.pg_lib_dir = os.path.join(str(pg_distrib_dir), "v{}".format(pg_version), "lib")
         self.env = os.environ.copy()
-        self.env["LD_LIBRARY_PATH"] = str(self.pg_lib_dir)
+        self.env["LD_LIBRARY_PATH"] = self.pg_lib_dir
 
     def _fixpath(self, command: List[str]):
-        if "/" not in str(command[0]):
-            command[0] = str(self.pg_bin_path / command[0])
+        if "/" not in command[0]:
+            command[0] = os.path.join(self.pg_bin_path, command[0])
 
     def _build_env(self, env_add: Optional[Env]) -> Env:
         if env_add is None:
@@ -1835,7 +1757,7 @@ class PgBin:
         """
 
         self._fixpath(command)
-        log.info(f"Running command '{' '.join(command)}'")
+        log.info('Running command "{}"'.format(" ".join(command)))
         env = self._build_env(env)
         subprocess.run(command, env=env, cwd=cwd, check=True)
 
@@ -1854,18 +1776,20 @@ class PgBin:
         """
 
         self._fixpath(command)
-        log.info(f"Running command '{' '.join(command)}'")
+        log.info('Running command "{}"'.format(" ".join(command)))
         env = self._build_env(env)
-        return subprocess_capture(self.log_dir, command, env=env, cwd=cwd, check=True, **kwargs)
+        return subprocess_capture(
+            str(self.log_dir), command, env=env, cwd=cwd, check=True, **kwargs
+        )
 
 
 @pytest.fixture(scope="function")
-def pg_bin(test_output_dir: Path, pg_distrib_dir: Path, pg_version: str) -> PgBin:
-    return PgBin(test_output_dir, pg_distrib_dir, pg_version)
+def pg_bin(test_output_dir: Path, pg_version: str) -> PgBin:
+    return PgBin(test_output_dir, pg_version)
 
 
 class VanillaPostgres(PgProtocol):
-    def __init__(self, pgdatadir: Path, pg_bin: PgBin, port: int, init: bool = True):
+    def __init__(self, pgdatadir: Path, pg_bin: PgBin, port: int, init=True):
         super().__init__(host="localhost", port=port, dbname="postgres")
         self.pgdatadir = pgdatadir
         self.pg_bin = pg_bin
@@ -1900,28 +1824,27 @@ class VanillaPostgres(PgProtocol):
         """Return size of pgdatadir subdirectory in bytes."""
         return get_dir_size(os.path.join(self.pgdatadir, subdir))
 
-    def __enter__(self) -> "VanillaPostgres":
+    def __enter__(self):
         return self
 
-    def __exit__(
-        self,
-        exc_type: Optional[Type[BaseException]],
-        exc: Optional[BaseException],
-        tb: Optional[TracebackType],
-    ):
+    def __exit__(self, exc_type, exc, tb):
         if self.running:
             self.stop()
 
 
+@pytest.fixture(scope="session")
+def pg_version() -> str:
+    return default_pg_version
+
+
 @pytest.fixture(scope="function")
 def vanilla_pg(
     test_output_dir: Path,
     port_distributor: PortDistributor,
-    pg_distrib_dir: Path,
     pg_version: str,
 ) -> Iterator[VanillaPostgres]:
     pgdatadir = test_output_dir / "pgdata-vanilla"
-    pg_bin = PgBin(test_output_dir, pg_distrib_dir, pg_version)
+    pg_bin = PgBin(test_output_dir, pg_version)
     port = port_distributor.get_port()
     with VanillaPostgres(pgdatadir, pg_bin, port) as vanilla_pg:
         yield vanilla_pg
@@ -1948,24 +1871,17 @@ 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):
         return self
 
-    def __exit__(
-        self,
-        exc_type: Optional[Type[BaseException]],
-        exc: Optional[BaseException],
-        tb: Optional[TracebackType],
-    ):
+    def __exit__(self, exc_type, exc, tb):
         # do nothing
         pass
 
 
 @pytest.fixture(scope="function")
-def remote_pg(
-    test_output_dir: Path, pg_distrib_dir: Path, pg_version: str
-) -> Iterator[RemotePostgres]:
-    pg_bin = PgBin(test_output_dir, pg_distrib_dir, pg_version)
+def remote_pg(test_output_dir: Path, pg_version: str) -> Iterator[RemotePostgres]:
+    pg_bin = PgBin(test_output_dir, pg_version)
 
     connstr = os.getenv("BENCHMARK_CONNSTR")
     if connstr is None:
@@ -1995,7 +1911,7 @@ class PSQL:
         self.path = path
         self.database_url = f"postgres://{host}:{port}/main?options=project%3Dgeneric-project-name"
 
-    async def run(self, query: Optional[str] = None) -> asyncio.subprocess.Process:
+    async def run(self, query=None):
         run_args = [self.path, "--no-psqlrc", "--quiet", "--tuples-only", self.database_url]
         if query is not None:
             run_args += ["--command", query]
@@ -2010,25 +1926,17 @@ class PSQL:
 
 
 class NeonProxy(PgProtocol):
-    def __init__(
-        self,
-        proxy_port: int,
-        http_port: int,
-        neon_binpath: Path,
-        auth_endpoint=None,
-        mgmt_port=None,
-    ):
+    def __init__(self, proxy_port: int, http_port: int, auth_endpoint=None, mgmt_port=None):
         super().__init__(dsn=auth_endpoint, port=proxy_port)
         self.host = "127.0.0.1"
         self.http_port = http_port
-        self.neon_binpath = neon_binpath
         self.proxy_port = proxy_port
         self.mgmt_port = mgmt_port
         self.auth_endpoint = auth_endpoint
         self._popen: Optional[subprocess.Popen[bytes]] = None
         self.link_auth_uri_prefix = "http://dummy-uri"
 
-    def start(self):
+    def start(self) -> None:
         """
         Starts a proxy with option '--auth-backend postgres' and a postgres instance already provided though '--auth-endpoint <postgress-instance>'."
         """
@@ -2037,7 +1945,7 @@ class NeonProxy(PgProtocol):
 
         # Start proxy
         args = [
-            str(self.neon_binpath / "proxy"),
+            os.path.join(neon_binpath, "proxy"),
             *["--http", f"{self.host}:{self.http_port}"],
             *["--proxy", f"{self.host}:{self.proxy_port}"],
             *["--auth-backend", "postgres"],
@@ -2046,14 +1954,14 @@ class NeonProxy(PgProtocol):
         self._popen = subprocess.Popen(args)
         self._wait_until_ready()
 
-    def start_with_link_auth(self):
+    def start_with_link_auth(self) -> None:
         """
         Starts a proxy with option '--auth-backend link' and a dummy authentication link '--uri dummy-auth-link'."
         """
         assert self._popen is None
 
         # Start proxy
-        bin_proxy = str(self.neon_binpath / "proxy")
+        bin_proxy = os.path.join(str(neon_binpath), "proxy")
         args = [bin_proxy]
         args.extend(["--http", f"{self.host}:{self.http_port}"])
         args.extend(["--proxy", f"{self.host}:{self.proxy_port}"])
@@ -2074,15 +1982,10 @@ class NeonProxy(PgProtocol):
         request_result.raise_for_status()
         return request_result.text
 
-    def __enter__(self) -> "NeonProxy":
+    def __enter__(self):
         return self
 
-    def __exit__(
-        self,
-        exc_type: Optional[Type[BaseException]],
-        exc: Optional[BaseException],
-        tb: Optional[TracebackType],
-    ):
+    def __exit__(self, exc_type, exc, tb):
         if self._popen is not None:
             # NOTE the process will die when we're done with tests anyway, because
             # it's a child process. This is mostly to clean up in between different tests.
@@ -2090,20 +1993,18 @@ class NeonProxy(PgProtocol):
 
 
 @pytest.fixture(scope="function")
-def link_proxy(port_distributor: PortDistributor, neon_binpath: Path) -> Iterator[NeonProxy]:
+def link_proxy(port_distributor) -> Iterator[NeonProxy]:
     """Neon proxy that routes through link auth."""
     http_port = port_distributor.get_port()
     proxy_port = port_distributor.get_port()
     mgmt_port = port_distributor.get_port()
-    with NeonProxy(proxy_port, http_port, neon_binpath=neon_binpath, mgmt_port=mgmt_port) as proxy:
+    with NeonProxy(proxy_port, http_port, mgmt_port=mgmt_port) as proxy:
         proxy.start_with_link_auth()
         yield proxy
 
 
 @pytest.fixture(scope="function")
-def static_proxy(
-    vanilla_pg: VanillaPostgres, port_distributor: PortDistributor, neon_binpath: Path
-) -> Iterator[NeonProxy]:
+def static_proxy(vanilla_pg, port_distributor) -> Iterator[NeonProxy]:
     """Neon proxy that routes directly to vanilla postgres."""
 
     # For simplicity, we use the same user for both `--auth-endpoint` and `safe_psql`
@@ -2119,10 +2020,7 @@ def static_proxy(
     http_port = port_distributor.get_port()
 
     with NeonProxy(
-        proxy_port=proxy_port,
-        http_port=http_port,
-        neon_binpath=neon_binpath,
-        auth_endpoint=auth_endpoint,
+        proxy_port=proxy_port, http_port=http_port, auth_endpoint=auth_endpoint
     ) as proxy:
         proxy.start()
         yield proxy
@@ -2303,15 +2201,10 @@ class Postgres(PgProtocol):
 
         return self
 
-    def __enter__(self) -> "Postgres":
+    def __enter__(self):
         return self
 
-    def __exit__(
-        self,
-        exc_type: Optional[Type[BaseException]],
-        exc: Optional[BaseException],
-        tb: Optional[TracebackType],
-    ):
+    def __exit__(self, exc_type, exc, tb):
         self.stop()
 
 
@@ -2320,7 +2213,7 @@ class PostgresFactory:
 
     def __init__(self, env: NeonEnv):
         self.env = env
-        self.num_instances: int = 0
+        self.num_instances = 0
         self.instances: List[Postgres] = []
 
     def create_start(
@@ -2415,7 +2308,7 @@ class Safekeeper:
                 break  # success
         return self
 
-    def stop(self, immediate: bool = False) -> "Safekeeper":
+    def stop(self, immediate=False) -> "Safekeeper":
         log.info("Stopping safekeeper {}".format(self.id))
         self.env.neon_cli.safekeeper_stop(self.id, immediate)
         self.running = False
@@ -2630,10 +2523,10 @@ class Etcd:
             self.handle.wait()
 
 
-def get_test_output_dir(request: FixtureRequest, top_output_dir: Path) -> Path:
+def get_test_output_dir(request: Any) -> Path:
     """Compute the working directory for an individual test."""
     test_name = request.node.name
-    test_dir = top_output_dir / test_name.replace("/", "-")
+    test_dir = Path(top_output_dir) / test_name.replace("/", "-")
     log.info(f"get_test_output_dir is {test_dir}")
     # make mypy happy
     assert isinstance(test_dir, Path)
@@ -2650,11 +2543,11 @@ def get_test_output_dir(request: FixtureRequest, top_output_dir: Path) -> Path:
 # this fixture ensures that the directory exists.  That works because
 # 'autouse' fixtures are run before other fixtures.
 @pytest.fixture(scope="function", autouse=True)
-def test_output_dir(request: FixtureRequest, top_output_dir: Path) -> Iterator[Path]:
+def test_output_dir(request: Any) -> Iterator[Path]:
     """Create the working directory for an individual test."""
 
     # one directory per test
-    test_dir = get_test_output_dir(request, top_output_dir)
+    test_dir = get_test_output_dir(request)
     log.info(f"test_output_dir is {test_dir}")
     shutil.rmtree(test_dir, ignore_errors=True)
     test_dir.mkdir()
@@ -2714,7 +2607,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):
     pgdata_files = []
     for root, _file, filenames in os.walk(pgdata_dir):
         for filename in filenames:
@@ -2746,7 +2639,7 @@ def check_restored_datadir_content(
     restored_dir_path = env.repo_dir / f"{pg.node_name}_restored_datadir"
     restored_dir_path.mkdir(exist_ok=True)
 
-    pg_bin = PgBin(test_output_dir, env.pg_distrib_dir, env.pg_version)
+    pg_bin = PgBin(test_output_dir, env.pg_version)
     psql_path = os.path.join(pg_bin.pg_bin_path, "psql")
 
     cmd = rf"""

test_runner/fixtures/pg_stats.py

@@ -1,4 +1,3 @@
-from functools import cached_property
 from typing import List
 
 import pytest
@@ -14,7 +13,7 @@ class PgStatTable:
         self.columns = columns
         self.additional_query = filter_query
 
-    @cached_property
+    @property
     def query(self) -> str:
         return f"SELECT {','.join(self.columns)} FROM {self.table} {self.additional_query}"
 
@@ -56,5 +55,6 @@ def pg_stats_wal() -> List[PgStatTable]:
         PgStatTable(
             "pg_stat_wal",
             ["wal_records", "wal_fpi", "wal_bytes", "wal_buffers_full", "wal_write"],
+            "",
         )
     ]

test_runner/fixtures/slow.py

@@ -1,8 +1,4 @@
-from typing import Any, List
-
 import pytest
-from _pytest.config import Config
-from _pytest.config.argparsing import Parser
 
 """
 This plugin allows tests to be marked as slow using pytest.mark.slow. By default slow
@@ -13,15 +9,15 @@ Copied from here: https://docs.pytest.org/en/latest/example/simple.html
 """
 
 
-def pytest_addoption(parser: Parser):
+def pytest_addoption(parser):
     parser.addoption("--runslow", action="store_true", default=False, help="run slow tests")
 
 
-def pytest_configure(config: Config):
+def pytest_configure(config):
     config.addinivalue_line("markers", "slow: mark test as slow to run")
 
 
-def pytest_collection_modifyitems(config: Config, items: List[Any]):
+def pytest_collection_modifyitems(config, items):
     if config.getoption("--runslow"):
         # --runslow given in cli: do not skip slow tests
         return

test_runner/fixtures/types.py

@@ -1,8 +1,6 @@
 import random
 from functools import total_ordering
-from typing import Any, Type, TypeVar, Union
-
-T = TypeVar("T", bound="Id")
+from typing import Union
 
 
 @total_ordering
@@ -19,35 +17,31 @@ class Lsn:
             """Convert lsn from hex notation to int."""
             l, r = x.split("/")
             self.lsn_int = (int(l, 16) << 32) + int(r, 16)
-        assert 0 <= self.lsn_int <= 0xFFFFFFFF_FFFFFFFF
+            # FIXME: error if it doesn't look like a valid LSN
 
-    def __str__(self) -> str:
+    def __str__(self):
         """Convert lsn from int to standard hex notation."""
-        return f"{(self.lsn_int >> 32):X}/{(self.lsn_int & 0xFFFFFFFF):X}"
+        return "{:X}/{:X}".format(self.lsn_int >> 32, self.lsn_int & 0xFFFFFFFF)
 
-    def __repr__(self) -> str:
-        return f'Lsn("{str(self)}")'
+    def __repr__(self):
+        return 'Lsn("{:X}/{:X}")'.format(self.lsn_int >> 32, self.lsn_int & 0xFFFFFFFF)
 
-    def __int__(self) -> int:
+    def __int__(self):
         return self.lsn_int
 
-    def __lt__(self, other: Any) -> bool:
-        if not isinstance(other, Lsn):
-            return NotImplemented
+    def __lt__(self, other: "Lsn") -> bool:
         return self.lsn_int < other.lsn_int
 
-    def __eq__(self, other: Any) -> bool:
+    def __eq__(self, other) -> bool:
         if not isinstance(other, Lsn):
             return NotImplemented
         return self.lsn_int == other.lsn_int
 
     # Returns the difference between two Lsns, in bytes
-    def __sub__(self, other: Any) -> int:
-        if not isinstance(other, Lsn):
-            return NotImplemented
+    def __sub__(self, other: "Lsn") -> int:
         return self.lsn_int - other.lsn_int
 
-    def __hash__(self) -> int:
+    def __hash__(self):
         return hash(self.lsn_int)
 
 
@@ -63,7 +57,7 @@ class Id:
         self.id = bytearray.fromhex(x)
         assert len(self.id) == 16
 
-    def __str__(self) -> str:
+    def __str__(self):
         return self.id.hex()
 
     def __lt__(self, other) -> bool:
@@ -76,20 +70,20 @@ class Id:
             return NotImplemented
         return self.id == other.id
 
-    def __hash__(self) -> int:
+    def __hash__(self):
         return hash(str(self.id))
 
     @classmethod
-    def generate(cls: Type[T]) -> T:
+    def generate(cls):
         """Generate a random ID"""
         return cls(random.randbytes(16).hex())
 
 
 class TenantId(Id):
-    def __repr__(self) -> str:
+    def __repr__(self):
         return f'`TenantId("{self.id.hex()}")'
 
 
 class TimelineId(Id):
-    def __repr__(self) -> str:
+    def __repr__(self):
         return f'TimelineId("{self.id.hex()}")'

test_runner/fixtures/utils.py

@@ -6,7 +6,7 @@ import subprocess
 import tarfile
 import time
 from pathlib import Path
-from typing import Any, Callable, Dict, List, Tuple, TypeVar
+from typing import Any, Callable, List, Tuple, TypeVar
 
 import allure  # type: ignore
 from fixtures.log_helper import log
@@ -15,12 +15,12 @@ from psycopg2.extensions import cursor
 Fn = TypeVar("Fn", bound=Callable[..., Any])
 
 
-def get_self_dir() -> Path:
+def get_self_dir() -> str:
     """Get the path to the directory where this script lives."""
-    return Path(__file__).resolve().parent
+    return os.path.dirname(os.path.abspath(__file__))
 
 
-def subprocess_capture(capture_dir: Path, cmd: List[str], **kwargs: Any) -> str:
+def subprocess_capture(capture_dir: str, cmd: List[str], **kwargs: Any) -> str:
     """Run a process and capture its output
 
     Output will go to files named "cmd_NNN.stdout" and "cmd_NNN.stderr"
@@ -30,11 +30,11 @@ def subprocess_capture(capture_dir: Path, cmd: List[str], **kwargs: Any) -> str:
     If those files already exist, we will overwrite them.
     Returns basepath for files with captured output.
     """
-    assert isinstance(cmd, list)
-    base = f"{os.path.basename(cmd[0])}_{global_counter()}"
+    assert type(cmd) is list
+    base = os.path.basename(cmd[0]) + "_{}".format(global_counter())
     basepath = os.path.join(capture_dir, base)
-    stdout_filename = f"{basepath}.stdout"
-    stderr_filename = f"{basepath}.stderr"
+    stdout_filename = basepath + ".stdout"
+    stderr_filename = basepath + ".stderr"
 
     try:
         with open(stdout_filename, "w") as stdout_f:
@@ -64,7 +64,7 @@ def global_counter() -> int:
     return _global_counter
 
 
-def print_gc_result(row: Dict[str, Any]):
+def print_gc_result(row):
     log.info("GC duration {elapsed} ms".format_map(row))
     log.info(
         "  total: {layers_total}, needed_by_cutoff {layers_needed_by_cutoff}, needed_by_pitr {layers_needed_by_pitr}"
@@ -78,7 +78,8 @@ def etcd_path() -> Path:
     path_output = shutil.which("etcd")
     if path_output is None:
         raise RuntimeError("etcd not found in PATH")
-    return Path(path_output)
+    else:
+        return Path(path_output)
 
 
 def query_scalar(cur: cursor, query: str) -> Any:
@@ -123,6 +124,7 @@ def get_timeline_dir_size(path: Path) -> int:
             # file is a delta layer
             _ = parse_delta_layer(dir_entry.name)
             sz += dir_entry.stat().st_size
+            continue
     return sz
 
 
@@ -155,8 +157,8 @@ def get_scale_for_db(size_mb: int) -> int:
     return round(0.06689 * size_mb - 0.5)
 
 
-ATTACHMENT_NAME_REGEX: re.Pattern = re.compile(  # type: ignore[type-arg]
-    r"flamegraph\.svg|regression\.diffs|.+\.(?:log|stderr|stdout|filediff|metrics|html)"
+ATTACHMENT_NAME_REGEX = re.compile(
+    r".+\.log|.+\.stderr|.+\.stdout|.+\.filediff|.+\.metrics|flamegraph\.svg|regression\.diffs|.+\.html"
 )
 
 

test_runner/performance/README.md

@@ -1,22 +1,3 @@
-# Running locally
-
-First make a release build. The profiling flag is optional, used only for tests that
-generate flame graphs. The `-s` flag just silences a lot of output, and makes it
-easier to see if you have compile errors without scrolling up.
-`BUILD_TYPE=release CARGO_BUILD_FLAGS="--features=testing,profiling" make -s -j8`
-
-NOTE: the `profiling` flag only works on linux because we use linux-specific
-libc APIs like `libc::timer_t`.
-
-Then run the tests
-`NEON_BIN=./target/release poetry run pytest test_runner/performance"`
-
-Some handy pytest flags for local development:
-- `-x` tells pytest to stop on first error
-- `-s` shows test output
-- `-k` selects a test to run
-- `--timeout=0` disables our default timeout of 300s (see `setup.cfg`)
-
 # What performance tests do we have and how we run them
 
 Performance tests are built using the same infrastructure as our usual python integration tests. There are some extra fixtures that help to collect performance metrics, and to run tests against both vanilla PostgreSQL and Neon for comparison.

test_runner/performance/test_read_trace.py

@@ -1,31 +0,0 @@
-from contextlib import closing
-
-from fixtures.neon_fixtures import NeonEnvBuilder
-
-
-# This test demonstrates how to collect a read trace. It's useful until
-# it gets replaced by a test that actually does stuff with the trace.
-def test_read_request_tracing(neon_env_builder: NeonEnvBuilder):
-    neon_env_builder.num_safekeepers = 1
-    env = neon_env_builder.init_start()
-
-    tenant, _ = env.neon_cli.create_tenant(
-        conf={
-            "trace_read_requests": "true",
-        }
-    )
-
-    timeline = env.neon_cli.create_timeline("test_trace_replay", tenant_id=tenant)
-    pg = env.postgres.create_start("test_trace_replay", "main", tenant)
-
-    with closing(pg.connect()) as conn:
-        with conn.cursor() as cur:
-            cur.execute("create table t (i integer);")
-            cur.execute(f"insert into t values (generate_series(1,{10000}));")
-            cur.execute("select count(*) from t;")
-
-    # Stop pg so we drop the connection and flush the traces
-    pg.stop()
-
-    trace_path = env.repo_dir / "traces" / str(tenant) / str(timeline)
-    assert trace_path.exists()

test_runner/performance/test_wal_backpressure.py

@@ -2,7 +2,7 @@ import statistics
 import threading
 import time
 import timeit
-from typing import Any, Callable, List
+from typing import Callable
 
 import pytest
 from fixtures.benchmark_fixture import MetricReport, NeonBenchmarker
@@ -197,7 +197,7 @@ def record_lsn_write_lag(env: PgCompare, run_cond: Callable[[], bool], pool_inte
     if not isinstance(env, NeonCompare):
         return
 
-    lsn_write_lags: List[Any] = []
+    lsn_write_lags = []
     last_received_lsn = Lsn(0)
     last_pg_flush_lsn = Lsn(0)
 
@@ -216,7 +216,6 @@ def record_lsn_write_lag(env: PgCompare, run_cond: Callable[[], bool], pool_inte
             )
 
             res = cur.fetchone()
-            assert isinstance(res, tuple)
             lsn_write_lags.append(res[0])
 
             curr_received_lsn = Lsn(res[3])

test_runner/pg_clients/python/asyncpg/asyncpg_example.py

@@ -24,6 +24,7 @@ if __name__ == "__main__":
         if (v := os.environ.get(k, None)) is not None
     }
 
-    row = asyncio.run(run(**kwargs))
+    loop = asyncio.new_event_loop()
+    row = loop.run_until_complete(run(**kwargs))
 
     print(row[0])

test_runner/pg_clients/test_pg_clients.py

@@ -46,9 +46,9 @@ def test_pg_clients(test_output_dir: Path, remote_pg: RemotePostgres, client: st
         raise RuntimeError("docker is required for running this test")
 
     build_cmd = [docker_bin, "build", "--tag", image_tag, f"{Path(__file__).parent / client}"]
-    subprocess_capture(test_output_dir, build_cmd, check=True)
+    subprocess_capture(str(test_output_dir), build_cmd, check=True)
 
     run_cmd = [docker_bin, "run", "--rm", "--env-file", env_file, image_tag]
-    basepath = subprocess_capture(test_output_dir, run_cmd, check=True)
+    basepath = subprocess_capture(str(test_output_dir), run_cmd, check=True)
 
     assert Path(f"{basepath}.stdout").read_text().strip() == "1"

test_runner/regress/test_compatibility.py

@@ -1,12 +1,12 @@
 import os
+import re
 import shutil
 import subprocess
 from pathlib import Path
-from typing import Any
+from typing import Any, Dict, Union
 
 import pytest
-import toml  # TODO: replace with tomllib for Python >= 3.11
-from fixtures.log_helper import log
+import toml
 from fixtures.neon_fixtures import (
     NeonCli,
     NeonEnvBuilder,
@@ -19,185 +19,89 @@ from fixtures.neon_fixtures import (
 from fixtures.types import Lsn
 from pytest import FixtureRequest
 
-#
-# A test suite that help to prevent unintentionally breaking backward or forward compatibility between Neon releases.
-# - `test_create_snapshot` a script wrapped in a test that creates a data snapshot.
-# - `test_backward_compatibility` checks that the current version of Neon can start/read/interract with a data snapshot created by the previous version.
-#   The path to the snapshot is configured by COMPATIBILITY_SNAPSHOT_DIR environment variable.
-#   If the breakage is intentional, the test can be xfaild with setting ALLOW_BACKWARD_COMPATIBILITY_BREAKAGE=true.
-# - `test_forward_compatibility` checks that a snapshot created by the current version can be started/read/interracted by the previous version of Neon.
-#   Paths to Neon and Postgres are configured by COMPATIBILITY_NEON_BIN and COMPATIBILITY_POSTGRES_DISTRIB_DIR environment variables.
-#   If the breakage is intentional, the test can be xfaild with setting ALLOW_FORWARD_COMPATIBILITY_BREAKAGE=true.
-#
-# The file contains a couple of helper functions:
-# - prepare_snapshot copies the snapshot, cleans it up and makes it ready for the current version of Neon (replaces paths and ports in config files).
-# - check_neon_works performs the test itself, feel free to add more checks there.
-#
+DEFAILT_LOCAL_SNAPSHOT_DIR = "test_output/test_prepare_snapshot/compatibility_snapshot_pg14"
 
 
-# Note: if renaming this test, don't forget to update a reference to it in a workflow file:
-# "Upload compatibility snapshot" step in .github/actions/run-python-test-set/action.yml
-@pytest.mark.xdist_group("compatibility")
-@pytest.mark.order(before="test_forward_compatibility")
-def test_create_snapshot(neon_env_builder: NeonEnvBuilder, pg_bin: PgBin, test_output_dir: Path):
-    # The test doesn't really test anything
-    # it creates a new snapshot for releases after we tested the current version against the previous snapshot in `test_backward_compatibility`.
-    #
-    # There's no cleanup here, it allows to adjust the data in `test_backward_compatibility` itself without re-collecting it.
-    neon_env_builder.pg_version = "14"
-    neon_env_builder.num_safekeepers = 3
-    neon_env_builder.enable_local_fs_remote_storage()
+def dump_differs(first: Path, second: Path, output: Path) -> bool:
+    """
+    Runs diff(1) command on two SQL dumps and write the output to the given output file.
+    Returns True if the dumps differ, False otherwise.
+    """
 
-    env = neon_env_builder.init_start()
-    pg = env.postgres.create_start("main")
-    pg_bin.run(["pgbench", "--initialize", "--scale=10", pg.connstr()])
-    pg_bin.run(["pgbench", "--time=60", "--progress=2", pg.connstr()])
-    pg_bin.run(["pg_dumpall", f"--dbname={pg.connstr()}", f"--file={test_output_dir / 'dump.sql'}"])
+    with output.open("w") as stdout:
+        rv = subprocess.run(
+            [
+                "diff",
+                "--unified",  # Make diff output more readable
+                "--ignore-matching-lines=^--",  # Ignore changes in comments
+                "--ignore-blank-lines",
+                str(first),
+                str(second),
+            ],
+            stdout=stdout,
+        )
 
-    snapshot_config = toml.load(test_output_dir / "repo" / "config")
-    tenant_id = snapshot_config["default_tenant_id"]
-    timeline_id = dict(snapshot_config["branch_name_mappings"]["main"])[tenant_id]
-
-    pageserver_http = env.pageserver.http_client()
-    lsn = Lsn(pg.safe_psql("SELECT pg_current_wal_flush_lsn()")[0][0])
-
-    wait_for_last_record_lsn(pageserver_http, tenant_id, timeline_id, lsn)
-    pageserver_http.timeline_checkpoint(tenant_id, timeline_id)
-    wait_for_upload(pageserver_http, tenant_id, timeline_id, lsn)
-
-    env.postgres.stop_all()
-    for sk in env.safekeepers:
-        sk.stop()
-    env.pageserver.stop()
-
-    shutil.copytree(test_output_dir, test_output_dir / "compatibility_snapshot_pg14")
-    # Directory `test_output_dir / "compatibility_snapshot_pg14"` is uploaded to S3 in a workflow, keep the name in sync with it
+    return rv.returncode != 0
 
 
-@pytest.mark.xdist_group("compatibility")
-@pytest.mark.order(after="test_create_snapshot")
+class PortReplacer(object):
+    """
+    Class-helper for replacing ports in config files.
+    """
+
+    def __init__(self, port_distributor: PortDistributor):
+        self.port_distributor = port_distributor
+        self.port_map: Dict[int, int] = {}
+
+    def replace_port(self, value: Union[int, str]) -> Union[int, str]:
+        if isinstance(value, int):
+            if (known_port := self.port_map.get(value)) is not None:
+                return known_port
+
+            self.port_map[value] = self.port_distributor.get_port()
+            return self.port_map[value]
+
+        if isinstance(value, str):
+            # Use regex to find port in a string
+            # urllib.parse.urlparse produces inconvenient results for cases without scheme like "localhost:5432"
+            # See https://bugs.python.org/issue27657
+            ports = re.findall(r":(\d+)(?:/|$)", value)
+            assert len(ports) == 1, f"can't find port in {value}"
+            port_int = int(ports[0])
+
+            if (known_port := self.port_map.get(port_int)) is not None:
+                return value.replace(f":{port_int}", f":{known_port}")
+
+            self.port_map[port_int] = self.port_distributor.get_port()
+            return value.replace(f":{port_int}", f":{self.port_map[port_int]}")
+
+        raise TypeError(f"unsupported type {type(value)} of {value=}")
+
+
+@pytest.mark.order(after="test_prepare_snapshot")
 def test_backward_compatibility(
-    pg_bin: PgBin,
-    port_distributor: PortDistributor,
-    test_output_dir: Path,
-    neon_binpath: Path,
-    pg_distrib_dir: Path,
-    pg_version: str,
-    request: FixtureRequest,
+    pg_bin: PgBin, port_distributor: PortDistributor, test_output_dir: Path, request: FixtureRequest
 ):
-    compatibility_snapshot_dir_env = os.environ.get("COMPATIBILITY_SNAPSHOT_DIR")
-    assert (
-        compatibility_snapshot_dir_env is not None
-    ), "COMPATIBILITY_SNAPSHOT_DIR is not set. It should be set to `compatibility_snapshot_pg14` path generateted by test_create_snapshot (ideally generated by the previous version of Neon)"
-    compatibility_snapshot_dir = Path(compatibility_snapshot_dir_env).resolve()
-
-    # Copy the snapshot to current directory, and prepare for the test
-    prepare_snapshot(
-        from_dir=compatibility_snapshot_dir,
-        to_dir=test_output_dir / "compatibility_snapshot",
-        port_distributor=port_distributor,
+    compatibility_snapshot_dir = Path(
+        os.environ.get("COMPATIBILITY_SNAPSHOT_DIR", DEFAILT_LOCAL_SNAPSHOT_DIR)
     )
-
-    breaking_changes_allowed = (
-        os.environ.get("ALLOW_BACKWARD_COMPATIBILITY_BREAKAGE", "false").lower() == "true"
+    assert compatibility_snapshot_dir.exists(), (
+        f"{compatibility_snapshot_dir} doesn't exist. Please run `test_prepare_snapshot` test first "
+        "to create the snapshot or set COMPATIBILITY_SNAPSHOT_DIR env variable to the existing snapshot"
     )
-    try:
-        check_neon_works(
-            test_output_dir / "compatibility_snapshot" / "repo",
-            neon_binpath,
-            pg_distrib_dir,
-            pg_version,
-            port_distributor,
-            test_output_dir,
-            pg_bin,
-            request,
-        )
-    except Exception:
-        if breaking_changes_allowed:
-            pytest.xfail(
-                "Breaking changes are allowed by ALLOW_BACKWARD_COMPATIBILITY_BREAKAGE env var"
-            )
-        else:
-            raise
+    compatibility_snapshot_dir = compatibility_snapshot_dir.resolve()
 
-    assert (
-        not breaking_changes_allowed
-    ), "Breaking changes are allowed by ALLOW_BACKWARD_COMPATIBILITY_BREAKAGE, but the test has passed without any breakage"
+    # Make compatibility snapshot artifacts pickupable by Allure
+    # by copying the snapshot directory to the curent test output directory.
+    repo_dir = test_output_dir / "compatibility_snapshot" / "repo"
 
-
-@pytest.mark.xdist_group("compatibility")
-@pytest.mark.order(after="test_create_snapshot")
-def test_forward_compatibility(
-    test_output_dir: Path,
-    port_distributor: PortDistributor,
-    pg_version: str,
-    request: FixtureRequest,
-):
-    compatibility_neon_bin_env = os.environ.get("COMPATIBILITY_NEON_BIN")
-    assert compatibility_neon_bin_env is not None, (
-        "COMPATIBILITY_NEON_BIN is not set. It should be set to a path with Neon binaries "
-        "(ideally generated by the previous version of Neon)"
-    )
-    compatibility_neon_bin = Path(compatibility_neon_bin_env).resolve()
-
-    compatibility_postgres_distrib_dir_env = os.environ.get("COMPATIBILITY_POSTGRES_DISTRIB_DIR")
-    assert (
-        compatibility_postgres_distrib_dir_env is not None
-    ), "COMPATIBILITY_POSTGRES_DISTRIB_DIR is not set. It should be set to a pg_install directrory (ideally generated by the previous version of Neon)"
-    compatibility_postgres_distrib_dir = Path(compatibility_postgres_distrib_dir_env).resolve()
-
-    compatibility_snapshot_dir = (
-        test_output_dir.parent / "test_create_snapshot" / "compatibility_snapshot_pg14"
-    )
-    # Copy the snapshot to current directory, and prepare for the test
-    prepare_snapshot(
-        from_dir=compatibility_snapshot_dir,
-        to_dir=test_output_dir / "compatibility_snapshot",
-        port_distributor=port_distributor,
-    )
-
-    breaking_changes_allowed = (
-        os.environ.get("ALLOW_FORWARD_COMPATIBILITY_BREAKAGE", "false").lower() == "true"
-    )
-    try:
-        check_neon_works(
-            test_output_dir / "compatibility_snapshot" / "repo",
-            compatibility_neon_bin,
-            compatibility_postgres_distrib_dir,
-            pg_version,
-            port_distributor,
-            test_output_dir,
-            PgBin(test_output_dir, compatibility_postgres_distrib_dir, pg_version),
-            request,
-        )
-    except Exception:
-        if breaking_changes_allowed:
-            pytest.xfail(
-                "Breaking changes are allowed by ALLOW_FORWARD_COMPATIBILITY_BREAKAGE env var"
-            )
-        else:
-            raise
-
-    assert (
-        not breaking_changes_allowed
-    ), "Breaking changes are allowed by ALLOW_FORWARD_COMPATIBILITY_BREAKAGE, but the test has passed without any breakage"
-
-
-def prepare_snapshot(from_dir: Path, to_dir: Path, port_distributor: PortDistributor):
-    assert from_dir.exists(), f"Snapshot '{from_dir}' doesn't exist"
-    assert (from_dir / "repo").exists(), f"Snapshot '{from_dir}' doesn't contain a repo directory"
-    assert (from_dir / "dump.sql").exists(), f"Snapshot '{from_dir}' doesn't contain a dump.sql"
-
-    log.info(f"Copying snapshot from {from_dir} to {to_dir}")
-    shutil.copytree(from_dir, to_dir)
-
-    repo_dir = to_dir / "repo"
+    shutil.copytree(compatibility_snapshot_dir / "repo", repo_dir)
 
     # Remove old logs to avoid confusion in test artifacts
     for logfile in repo_dir.glob("**/*.log"):
         logfile.unlink()
 
-    # Remove tenants data for compute
+    # Remove tenants data for computes
     for tenant in (repo_dir / "pgdatadirs" / "tenants").glob("*"):
         shutil.rmtree(tenant)
 
@@ -206,17 +110,20 @@ def prepare_snapshot(from_dir: Path, to_dir: Path, port_distributor: PortDistrib
         shutil.rmtree(tenant / "wal-redo-datadir.___temp")
 
     # Update paths and ports in config files
+    pr = PortReplacer(port_distributor)
+
     pageserver_toml = repo_dir / "pageserver.toml"
     pageserver_config = toml.load(pageserver_toml)
-    pageserver_config["remote_storage"]["local_path"] = repo_dir / "local_fs_remote_storage"
-    pageserver_config["listen_http_addr"] = port_distributor.replace_with_new_port(
-        pageserver_config["listen_http_addr"]
-    )
-    pageserver_config["listen_pg_addr"] = port_distributor.replace_with_new_port(
-        pageserver_config["listen_pg_addr"]
+    new_local_path = pageserver_config["remote_storage"]["local_path"].replace(
+        "/test_prepare_snapshot/",
+        "/test_backward_compatibility/compatibility_snapshot/",
     )
+
+    pageserver_config["remote_storage"]["local_path"] = new_local_path
+    pageserver_config["listen_http_addr"] = pr.replace_port(pageserver_config["listen_http_addr"])
+    pageserver_config["listen_pg_addr"] = pr.replace_port(pageserver_config["listen_pg_addr"])
     pageserver_config["broker_endpoints"] = [
-        port_distributor.replace_with_new_port(ep) for ep in pageserver_config["broker_endpoints"]
+        pr.replace_port(ep) for ep in pageserver_config["broker_endpoints"]
     ]
 
     with pageserver_toml.open("w") as f:
@@ -225,18 +132,17 @@ def prepare_snapshot(from_dir: Path, to_dir: Path, port_distributor: PortDistrib
     snapshot_config_toml = repo_dir / "config"
     snapshot_config = toml.load(snapshot_config_toml)
     snapshot_config["etcd_broker"]["broker_endpoints"] = [
-        port_distributor.replace_with_new_port(ep)
-        for ep in snapshot_config["etcd_broker"]["broker_endpoints"]
+        pr.replace_port(ep) for ep in snapshot_config["etcd_broker"]["broker_endpoints"]
     ]
-    snapshot_config["pageserver"]["listen_http_addr"] = port_distributor.replace_with_new_port(
+    snapshot_config["pageserver"]["listen_http_addr"] = pr.replace_port(
         snapshot_config["pageserver"]["listen_http_addr"]
     )
-    snapshot_config["pageserver"]["listen_pg_addr"] = port_distributor.replace_with_new_port(
+    snapshot_config["pageserver"]["listen_pg_addr"] = pr.replace_port(
         snapshot_config["pageserver"]["listen_pg_addr"]
     )
     for sk in snapshot_config["safekeepers"]:
-        sk["http_port"] = port_distributor.replace_with_new_port(sk["http_port"])
-        sk["pg_port"] = port_distributor.replace_with_new_port(sk["pg_port"])
+        sk["http_port"] = pr.replace_port(sk["http_port"])
+        sk["pg_port"] = pr.replace_port(sk["pg_port"])
 
     with (snapshot_config_toml).open("w") as f:
         toml.dump(snapshot_config, f)
@@ -248,7 +154,7 @@ def prepare_snapshot(from_dir: Path, to_dir: Path, port_distributor: PortDistrib
             "--recursive",
             "--binary-file=without-match",
             "--files-with-matches",
-            "test_create_snapshot/repo",
+            "test_prepare_snapshot/repo",
             str(repo_dir),
         ],
         capture_output=True,
@@ -256,47 +162,42 @@ def prepare_snapshot(from_dir: Path, to_dir: Path, port_distributor: PortDistrib
     )
     assert (
         rv.returncode != 0
-    ), f"there're files referencing `test_create_snapshot/repo`, this path should be replaced with {repo_dir}:\n{rv.stdout}"
+    ), f"there're files referencing `test_prepare_snapshot/repo`, this path should be replaced with {repo_dir}:\n{rv.stdout}"
 
-
-def check_neon_works(
-    repo_dir: Path,
-    neon_binpath: Path,
-    pg_distrib_dir: Path,
-    pg_version: str,
-    port_distributor: PortDistributor,
-    test_output_dir: Path,
-    pg_bin: PgBin,
-    request: FixtureRequest,
-):
-    snapshot_config_toml = repo_dir / "config"
-    snapshot_config = toml.load(snapshot_config_toml)
-    snapshot_config["neon_distrib_dir"] = str(neon_binpath)
-    snapshot_config["postgres_distrib_dir"] = str(pg_distrib_dir)
-    with (snapshot_config_toml).open("w") as f:
-        toml.dump(snapshot_config, f)
-
-    # TODO: replace with NeonEnvBuilder / NeonEnv
+    # NeonEnv stub to make NeonCli happy
     config: Any = type("NeonEnvStub", (object,), {})
     config.rust_log_override = None
     config.repo_dir = repo_dir
-    config.pg_version = pg_version
+    config.pg_version = "14"  # Note: `pg_dumpall` (from pg_bin) version is set by DEFAULT_PG_VERSION_DEFAULT and can be overriden by DEFAULT_PG_VERSION env var
     config.initial_tenant = snapshot_config["default_tenant_id"]
-    config.neon_binpath = neon_binpath
-    config.pg_distrib_dir = pg_distrib_dir
 
+    # Check that we can start the project
     cli = NeonCli(config)
-    cli.raw_cli(["start"])
-    request.addfinalizer(lambda: cli.raw_cli(["stop"]))
+    try:
+        cli.raw_cli(["start"])
+        request.addfinalizer(lambda: cli.raw_cli(["stop"]))
 
-    pg_port = port_distributor.get_port()
-    cli.pg_start("main", port=pg_port)
-    request.addfinalizer(lambda: cli.pg_stop("main"))
+        result = cli.pg_start("main", port=port_distributor.get_port())
+        request.addfinalizer(lambda: cli.pg_stop("main"))
+    except Exception:
+        breaking_changes_allowed = (
+            os.environ.get("ALLOW_BREAKING_CHANGES", "false").lower() == "true"
+        )
+        if breaking_changes_allowed:
+            pytest.xfail("Breaking changes are allowed by ALLOW_BREAKING_CHANGES env var")
+        else:
+            raise
 
-    connstr = f"host=127.0.0.1 port={pg_port} user=cloud_admin dbname=postgres"
+    connstr_all = re.findall(r"Starting postgres node at '([^']+)'", result.stdout)
+    assert len(connstr_all) == 1, f"can't parse connstr from {result.stdout}"
+    connstr = connstr_all[0]
+
+    # Check that the project produces the same dump as the previous version.
+    # The assert itself deferred to the end of the test
+    # to allow us to perform checks that change data before failing
     pg_bin.run(["pg_dumpall", f"--dbname={connstr}", f"--file={test_output_dir / 'dump.sql'}"])
     initial_dump_differs = dump_differs(
-        repo_dir.parent / "dump.sql",
+        compatibility_snapshot_dir / "dump.sql",
         test_output_dir / "dump.sql",
         test_output_dir / "dump.filediff",
     )
@@ -334,23 +235,38 @@ def check_neon_works(
     assert not initial_dump_differs, "initial dump differs"
 
 
-def dump_differs(first: Path, second: Path, output: Path) -> bool:
-    """
-    Runs diff(1) command on two SQL dumps and write the output to the given output file.
-    Returns True if the dumps differ, False otherwise.
-    """
+# Note: if renaming this test, don't forget to update a reference to it in a workflow file:
+# "Upload compatibility snapshot" step in .github/actions/run-python-test-set/action.yml
+def test_prepare_snapshot(neon_env_builder: NeonEnvBuilder, pg_bin: PgBin, test_output_dir: Path):
+    # The test doesn't really test anything
+    # it creates a new snapshot for releases after we tested the current version against the previous snapshot in `test_backward_compatibility`.
+    #
+    # There's no cleanup here, it allows to adjust the data in `test_backward_compatibility` itself without re-collecting it.
+    neon_env_builder.pg_version = "14"
+    neon_env_builder.num_safekeepers = 3
+    neon_env_builder.enable_local_fs_remote_storage()
 
-    with output.open("w") as stdout:
-        rv = subprocess.run(
-            [
-                "diff",
-                "--unified",  # Make diff output more readable
-                "--ignore-matching-lines=^--",  # Ignore changes in comments
-                "--ignore-blank-lines",
-                str(first),
-                str(second),
-            ],
-            stdout=stdout,
-        )
+    env = neon_env_builder.init_start()
+    pg = env.postgres.create_start("main")
+    pg_bin.run(["pgbench", "--initialize", "--scale=10", pg.connstr()])
+    pg_bin.run(["pgbench", "--time=60", "--progress=2", pg.connstr()])
+    pg_bin.run(["pg_dumpall", f"--dbname={pg.connstr()}", f"--file={test_output_dir / 'dump.sql'}"])
 
-    return rv.returncode != 0
+    snapshot_config = toml.load(test_output_dir / "repo" / "config")
+    tenant_id = snapshot_config["default_tenant_id"]
+    timeline_id = dict(snapshot_config["branch_name_mappings"]["main"])[tenant_id]
+
+    pageserver_http = env.pageserver.http_client()
+    lsn = Lsn(pg.safe_psql("SELECT pg_current_wal_flush_lsn()")[0][0])
+
+    wait_for_last_record_lsn(pageserver_http, tenant_id, timeline_id, lsn)
+    pageserver_http.timeline_checkpoint(tenant_id, timeline_id)
+    wait_for_upload(pageserver_http, tenant_id, timeline_id, lsn)
+
+    env.postgres.stop_all()
+    for sk in env.safekeepers:
+        sk.stop()
+    env.pageserver.stop()
+
+    shutil.copytree(test_output_dir, test_output_dir / "compatibility_snapshot_pg14")
+    # Directory `test_output_dir / "compatibility_snapshot_pg14"` is uploaded to S3 in a workflow, keep the name in sync with it

test_runner/regress/test_fullbackup.py

@@ -1,8 +1,13 @@
 import os
-from pathlib import Path
 
 from fixtures.log_helper import log
-from fixtures.neon_fixtures import NeonEnvBuilder, PgBin, PortDistributor, VanillaPostgres
+from fixtures.neon_fixtures import (
+    NeonEnvBuilder,
+    PgBin,
+    PortDistributor,
+    VanillaPostgres,
+    pg_distrib_dir,
+)
 from fixtures.types import Lsn, TimelineId
 from fixtures.utils import query_scalar, subprocess_capture
 
@@ -11,10 +16,7 @@ num_rows = 1000
 
 # Ensure that regular postgres can start from fullbackup
 def test_fullbackup(
-    neon_env_builder: NeonEnvBuilder,
-    pg_bin: PgBin,
-    port_distributor: PortDistributor,
-    pg_distrib_dir: Path,
+    neon_env_builder: NeonEnvBuilder, pg_bin: PgBin, port_distributor: PortDistributor
 ):
     env = neon_env_builder.init_start()
 
@@ -38,7 +40,7 @@ def test_fullbackup(
 
     # Set LD_LIBRARY_PATH in the env properly, otherwise we may use the wrong libpq.
     # PgBin sets it automatically, but here we need to pipe psql output to the tar command.
-    psql_env = {"LD_LIBRARY_PATH": str(pg_distrib_dir / "lib")}
+    psql_env = {"LD_LIBRARY_PATH": os.path.join(str(pg_distrib_dir), "lib")}
 
     # Get and unpack fullbackup from pageserver
     restored_dir_path = env.repo_dir / "restored_datadir"
@@ -47,7 +49,9 @@ def test_fullbackup(
     cmd = ["psql", "--no-psqlrc", env.pageserver.connstr(), "-c", query]
     result_basepath = pg_bin.run_capture(cmd, env=psql_env)
     tar_output_file = result_basepath + ".stdout"
-    subprocess_capture(env.repo_dir, ["tar", "-xf", tar_output_file, "-C", str(restored_dir_path)])
+    subprocess_capture(
+        str(env.repo_dir), ["tar", "-xf", tar_output_file, "-C", str(restored_dir_path)]
+    )
 
     # HACK
     # fullbackup returns neon specific pg_control and first WAL segment

test_runner/regress/test_import.py

@@ -13,6 +13,7 @@ from fixtures.neon_fixtures import (
     NeonEnvBuilder,
     PgBin,
     Postgres,
+    pg_distrib_dir,
     wait_for_last_record_lsn,
     wait_for_upload,
 )
@@ -127,7 +128,7 @@ def test_import_from_pageserver_small(pg_bin: PgBin, neon_env_builder: NeonEnvBu
 
     num_rows = 3000
     lsn = _generate_data(num_rows, pg)
-    _import(num_rows, lsn, env, pg_bin, timeline, env.pg_distrib_dir)
+    _import(num_rows, lsn, env, pg_bin, timeline)
 
 
 @pytest.mark.timeout(1800)
@@ -155,7 +156,7 @@ def test_import_from_pageserver_multisegment(pg_bin: PgBin, neon_env_builder: Ne
     log.info(f"timeline logical size = {logical_size / (1024 ** 2)}MB")
     assert logical_size > 1024**3  # = 1GB
 
-    tar_output_file = _import(num_rows, lsn, env, pg_bin, timeline, env.pg_distrib_dir)
+    tar_output_file = _import(num_rows, lsn, env, pg_bin, timeline)
 
     # Check if the backup data contains multiple segment files
     cnt_seg_files = 0
@@ -190,12 +191,7 @@ def _generate_data(num_rows: int, pg: Postgres) -> Lsn:
 
 
 def _import(
-    expected_num_rows: int,
-    lsn: Lsn,
-    env: NeonEnv,
-    pg_bin: PgBin,
-    timeline: TimelineId,
-    pg_distrib_dir: Path,
+    expected_num_rows: int, lsn: Lsn, env: NeonEnv, pg_bin: PgBin, timeline: TimelineId
 ) -> str:
     """Test importing backup data to the pageserver.
 
@@ -209,7 +205,7 @@ def _import(
 
     # Set LD_LIBRARY_PATH in the env properly, otherwise we may use the wrong libpq.
     # PgBin sets it automatically, but here we need to pipe psql output to the tar command.
-    psql_env = {"LD_LIBRARY_PATH": str(pg_distrib_dir / "lib")}
+    psql_env = {"LD_LIBRARY_PATH": os.path.join(str(pg_distrib_dir), "lib")}
 
     # Get a fullbackup from pageserver
     query = f"fullbackup { env.initial_tenant} {timeline} {lsn}"

test_runner/regress/test_pageserver_api.py

@@ -1,5 +1,5 @@
+import pathlib
 import subprocess
-from pathlib import Path
 from typing import Optional
 
 from fixtures.neon_fixtures import (
@@ -7,18 +7,18 @@ from fixtures.neon_fixtures import (
     NeonEnv,
     NeonEnvBuilder,
     PageserverHttpClient,
+    neon_binpath,
+    pg_distrib_dir,
 )
 from fixtures.types import Lsn, TenantId, TimelineId
 from fixtures.utils import wait_until
 
 
 # test that we cannot override node id after init
-def test_pageserver_init_node_id(
-    neon_simple_env: NeonEnv, neon_binpath: Path, pg_distrib_dir: Path
-):
+def test_pageserver_init_node_id(neon_simple_env: NeonEnv):
     repo_dir = neon_simple_env.repo_dir
     pageserver_config = repo_dir / "pageserver.toml"
-    pageserver_bin = neon_binpath / "pageserver"
+    pageserver_bin = pathlib.Path(neon_binpath) / "pageserver"
 
     def run_pageserver(args):
         return subprocess.run(

test_runner/regress/test_pg_regress.py

@@ -1,10 +1,11 @@
 #
 # This file runs pg_regress-based tests.
 #
+import os
 from pathlib import Path
 
 import pytest
-from fixtures.neon_fixtures import NeonEnv, check_restored_datadir_content
+from fixtures.neon_fixtures import NeonEnv, base_dir, check_restored_datadir_content, pg_distrib_dir
 
 
 # Run the main PostgreSQL regression tests, in src/test/regress.
@@ -12,14 +13,7 @@ from fixtures.neon_fixtures import NeonEnv, check_restored_datadir_content
 # This runs for a long time, especially in debug mode, so use a larger-than-default
 # timeout.
 @pytest.mark.timeout(1800)
-def test_pg_regress(
-    neon_simple_env: NeonEnv,
-    test_output_dir: Path,
-    pg_bin,
-    capsys,
-    base_dir: Path,
-    pg_distrib_dir: Path,
-):
+def test_pg_regress(neon_simple_env: NeonEnv, test_output_dir: Path, pg_bin, capsys):
     env = neon_simple_env
 
     env.neon_cli.create_branch("test_pg_regress", "empty")
@@ -32,20 +26,20 @@ def test_pg_regress(
     (runpath / "testtablespace").mkdir(parents=True)
 
     # Compute all the file locations that pg_regress will need.
-    build_path = pg_distrib_dir / f"build/v{env.pg_version}/src/test/regress"
-    src_path = base_dir / f"vendor/postgres-v{env.pg_version}/src/test/regress"
-    bindir = pg_distrib_dir / f"v{env.pg_version}/bin"
-    schedule = src_path / "parallel_schedule"
-    pg_regress = build_path / "pg_regress"
+    build_path = os.path.join(pg_distrib_dir, "build/v{}/src/test/regress").format(env.pg_version)
+    src_path = os.path.join(base_dir, "vendor/postgres-v{}/src/test/regress").format(env.pg_version)
+    bindir = os.path.join(pg_distrib_dir, "v{}".format(env.pg_version), "bin")
+    schedule = os.path.join(src_path, "parallel_schedule")
+    pg_regress = os.path.join(build_path, "pg_regress")
 
     pg_regress_command = [
-        str(pg_regress),
+        pg_regress,
         '--bindir=""',
         "--use-existing",
-        f"--bindir={bindir}",
-        f"--dlpath={build_path}",
-        f"--schedule={schedule}",
-        f"--inputdir={src_path}",
+        "--bindir={}".format(bindir),
+        "--dlpath={}".format(build_path),
+        "--schedule={}".format(schedule),
+        "--inputdir={}".format(src_path),
     ]
 
     env_vars = {
@@ -72,14 +66,7 @@ def test_pg_regress(
 # This runs for a long time, especially in debug mode, so use a larger-than-default
 # timeout.
 @pytest.mark.timeout(1800)
-def test_isolation(
-    neon_simple_env: NeonEnv,
-    test_output_dir: Path,
-    pg_bin,
-    capsys,
-    base_dir: Path,
-    pg_distrib_dir: Path,
-):
+def test_isolation(neon_simple_env: NeonEnv, test_output_dir: Path, pg_bin, capsys):
     env = neon_simple_env
 
     env.neon_cli.create_branch("test_isolation", "empty")
@@ -93,19 +80,21 @@ def test_isolation(
     (runpath / "testtablespace").mkdir(parents=True)
 
     # Compute all the file locations that pg_isolation_regress will need.
-    build_path = pg_distrib_dir / f"build/v{env.pg_version}/src/test/isolation"
-    src_path = base_dir / f"vendor/postgres-v{env.pg_version}/src/test/isolation"
-    bindir = pg_distrib_dir / f"v{env.pg_version}/bin"
-    schedule = src_path / "isolation_schedule"
-    pg_isolation_regress = build_path / "pg_isolation_regress"
+    build_path = os.path.join(pg_distrib_dir, "build/v{}/src/test/isolation".format(env.pg_version))
+    src_path = os.path.join(
+        base_dir, "vendor/postgres-v{}/src/test/isolation".format(env.pg_version)
+    )
+    bindir = os.path.join(pg_distrib_dir, "v{}".format(env.pg_version), "bin")
+    schedule = os.path.join(src_path, "isolation_schedule")
+    pg_isolation_regress = os.path.join(build_path, "pg_isolation_regress")
 
     pg_isolation_regress_command = [
-        str(pg_isolation_regress),
+        pg_isolation_regress,
         "--use-existing",
-        f"--bindir={bindir}",
-        f"--dlpath={build_path}",
-        f"--inputdir={src_path}",
-        f"--schedule={schedule}",
+        "--bindir={}".format(bindir),
+        "--dlpath={}".format(build_path),
+        "--inputdir={}".format(src_path),
+        "--schedule={}".format(schedule),
     ]
 
     env_vars = {
@@ -123,14 +112,7 @@ def test_isolation(
 
 # Run extra Neon-specific pg_regress-based tests. The tests and their
 # schedule file are in the sql_regress/ directory.
-def test_sql_regress(
-    neon_simple_env: NeonEnv,
-    test_output_dir: Path,
-    pg_bin,
-    capsys,
-    base_dir: Path,
-    pg_distrib_dir: Path,
-):
+def test_sql_regress(neon_simple_env: NeonEnv, test_output_dir: Path, pg_bin, capsys):
     env = neon_simple_env
 
     env.neon_cli.create_branch("test_sql_regress", "empty")
@@ -144,19 +126,19 @@ def test_sql_regress(
 
     # Compute all the file locations that pg_regress will need.
     # This test runs neon specific tests
-    build_path = pg_distrib_dir / f"build/v{env.pg_version}/src/test/regress"
-    src_path = base_dir / "test_runner/sql_regress"
-    bindir = pg_distrib_dir / f"v{env.pg_version}/bin"
-    schedule = src_path / "parallel_schedule"
-    pg_regress = build_path / "pg_regress"
+    build_path = os.path.join(pg_distrib_dir, "build/v{}/src/test/regress").format(env.pg_version)
+    src_path = os.path.join(base_dir, "test_runner/sql_regress")
+    bindir = os.path.join(pg_distrib_dir, "v{}".format(env.pg_version), "bin")
+    schedule = os.path.join(src_path, "parallel_schedule")
+    pg_regress = os.path.join(build_path, "pg_regress")
 
     pg_regress_command = [
-        str(pg_regress),
+        pg_regress,
         "--use-existing",
-        f"--bindir={bindir}",
-        f"--dlpath={build_path}",
-        f"--schedule={schedule}",
-        f"--inputdir={src_path}",
+        "--bindir={}".format(bindir),
+        "--dlpath={}".format(build_path),
+        "--schedule={}".format(schedule),
+        "--inputdir={}".format(src_path),
     ]
 
     env_vars = {

test_runner/regress/test_proxy.py

@@ -129,7 +129,6 @@ async def test_psql_session_id(vanilla_pg: VanillaPostgres, link_proxy: NeonProx
 
     create_and_send_db_info(vanilla_pg, psql_session_id, link_proxy.mgmt_port)
 
-    assert proc.stdout is not None
     out = (await proc.stdout.read()).decode("utf-8").strip()
     assert out == "42"
 

test_runner/regress/test_tenant_relocation.py

@@ -1,7 +1,7 @@
 import os
+import pathlib
 import threading
 from contextlib import closing, contextmanager
-from pathlib import Path
 from typing import Any, Dict, Optional, Tuple
 
 import pytest
@@ -14,6 +14,9 @@ from fixtures.neon_fixtures import (
     PortDistributor,
     Postgres,
     assert_no_in_progress_downloads_for_tenant,
+    base_dir,
+    neon_binpath,
+    pg_distrib_dir,
     wait_for_last_record_lsn,
     wait_for_upload,
 )
@@ -27,13 +30,12 @@ def assert_abs_margin_ratio(a: float, b: float, margin_ratio: float):
 
 @contextmanager
 def new_pageserver_service(
-    new_pageserver_dir: Path,
-    pageserver_bin: Path,
-    remote_storage_mock_path: Path,
+    new_pageserver_dir: pathlib.Path,
+    pageserver_bin: pathlib.Path,
+    remote_storage_mock_path: pathlib.Path,
     pg_port: int,
     http_port: int,
     broker: Optional[Etcd],
-    pg_distrib_dir: Path,
 ):
     """
     cannot use NeonPageserver yet because it depends on neon cli
@@ -191,10 +193,10 @@ def switch_pg_to_new_pageserver(
     new_pageserver_port: int,
     tenant_id: TenantId,
     timeline_id: TimelineId,
-) -> Path:
+) -> pathlib.Path:
     pg.stop()
 
-    pg_config_file_path = Path(pg.config_file_path())
+    pg_config_file_path = pathlib.Path(pg.config_file_path())
     pg_config_file_path.open("a").write(
         f"\nneon.pageserver_connstring = 'postgresql://no_user:@localhost:{new_pageserver_port}'"
     )
@@ -217,7 +219,7 @@ def switch_pg_to_new_pageserver(
     return timeline_to_detach_local_path
 
 
-def post_migration_check(pg: Postgres, sum_before_migration: int, old_local_path: Path):
+def post_migration_check(pg: Postgres, sum_before_migration: int, old_local_path: pathlib.Path):
     with pg_cur(pg) as cur:
         # check that data is still there
         cur.execute("SELECT sum(key) FROM t")
@@ -249,9 +251,7 @@ def post_migration_check(pg: Postgres, sum_before_migration: int, old_local_path
 def test_tenant_relocation(
     neon_env_builder: NeonEnvBuilder,
     port_distributor: PortDistributor,
-    test_output_dir: Path,
-    neon_binpath: Path,
-    base_dir: Path,
+    test_output_dir,
     method: str,
     with_load: str,
 ):
@@ -350,7 +350,7 @@ def test_tenant_relocation(
     new_pageserver_pg_port = port_distributor.get_port()
     new_pageserver_http_port = port_distributor.get_port()
     log.info("new pageserver ports pg %s http %s", new_pageserver_pg_port, new_pageserver_http_port)
-    pageserver_bin = neon_binpath / "pageserver"
+    pageserver_bin = pathlib.Path(neon_binpath) / "pageserver"
 
     new_pageserver_http = PageserverHttpClient(
         port=new_pageserver_http_port,
@@ -365,7 +365,6 @@ def test_tenant_relocation(
         new_pageserver_pg_port,
         new_pageserver_http_port,
         neon_env_builder.broker,
-        neon_env_builder.pg_distrib_dir,
     ):
 
         # Migrate either by attaching from s3 or import/export basebackup
@@ -374,7 +373,7 @@ def test_tenant_relocation(
                 "poetry",
                 "run",
                 "python",
-                str(base_dir / "scripts/export_import_between_pageservers.py"),
+                os.path.join(base_dir, "scripts/export_import_between_pageservers.py"),
                 "--tenant-id",
                 str(tenant_id),
                 "--from-host",
@@ -390,9 +389,9 @@ def test_tenant_relocation(
                 "--to-pg-port",
                 str(new_pageserver_pg_port),
                 "--pg-distrib-dir",
-                str(neon_env_builder.pg_distrib_dir),
+                pg_distrib_dir,
                 "--work-dir",
-                str(test_output_dir),
+                os.path.join(test_output_dir),
                 "--tmp-pg-port",
                 str(port_distributor.get_port()),
             ]

test_runner/regress/test_timeline_size.py

@@ -338,7 +338,6 @@ def test_timeline_size_metrics(
     neon_simple_env: NeonEnv,
     test_output_dir: Path,
     port_distributor: PortDistributor,
-    pg_distrib_dir: Path,
     pg_version: str,
 ):
     env = neon_simple_env
@@ -383,7 +382,7 @@ def test_timeline_size_metrics(
     tl_logical_size_metric = int(matches.group(1))
 
     pgdatadir = test_output_dir / "pgdata-vanilla"
-    pg_bin = PgBin(test_output_dir, pg_distrib_dir, pg_version)
+    pg_bin = PgBin(test_output_dir, pg_version)
     port = port_distributor.get_port()
     with VanillaPostgres(pgdatadir, pg_bin, port) as vanilla_pg:
         vanilla_pg.configure([f"port={port}"])

test_runner/regress/test_wal_acceptor.py

@@ -30,6 +30,7 @@ from fixtures.neon_fixtures import (
     SafekeeperHttpClient,
     SafekeeperPort,
     available_remote_storages,
+    neon_binpath,
     wait_for_last_record_lsn,
     wait_for_upload,
 )
@@ -796,7 +797,6 @@ class SafekeeperEnv:
         repo_dir: Path,
         port_distributor: PortDistributor,
         pg_bin: PgBin,
-        neon_binpath: Path,
         num_safekeepers: int = 1,
     ):
         self.repo_dir = repo_dir
@@ -808,7 +808,7 @@ class SafekeeperEnv:
         )
         self.pg_bin = pg_bin
         self.num_safekeepers = num_safekeepers
-        self.bin_safekeeper = str(neon_binpath / "safekeeper")
+        self.bin_safekeeper = os.path.join(str(neon_binpath), "safekeeper")
         self.safekeepers: Optional[List[subprocess.CompletedProcess[Any]]] = None
         self.postgres: Optional[ProposerPostgres] = None
         self.tenant_id: Optional[TenantId] = None
@@ -911,10 +911,7 @@ class SafekeeperEnv:
 
 
 def test_safekeeper_without_pageserver(
-    test_output_dir: str,
-    port_distributor: PortDistributor,
-    pg_bin: PgBin,
-    neon_binpath: Path,
+    test_output_dir: str, port_distributor: PortDistributor, pg_bin: PgBin
 ):
     # Create the environment in the test-specific output dir
     repo_dir = Path(os.path.join(test_output_dir, "repo"))
@@ -923,7 +920,6 @@ def test_safekeeper_without_pageserver(
         repo_dir,
         port_distributor,
         pg_bin,
-        neon_binpath,
     )
 
     with env:

test_runner/regress/test_wal_restore.py

@@ -1,6 +1,14 @@
+import os
 from pathlib import Path
 
-from fixtures.neon_fixtures import NeonEnvBuilder, PgBin, PortDistributor, VanillaPostgres
+from fixtures.neon_fixtures import (
+    NeonEnvBuilder,
+    PgBin,
+    PortDistributor,
+    VanillaPostgres,
+    base_dir,
+    pg_distrib_dir,
+)
 from fixtures.types import TenantId
 
 
@@ -9,8 +17,6 @@ def test_wal_restore(
     pg_bin: PgBin,
     test_output_dir: Path,
     port_distributor: PortDistributor,
-    base_dir: Path,
-    pg_distrib_dir: Path,
 ):
     env = neon_env_builder.init_start()
     env.neon_cli.create_branch("test_wal_restore")
@@ -20,13 +26,11 @@ def test_wal_restore(
     env.neon_cli.pageserver_stop()
     port = port_distributor.get_port()
     data_dir = test_output_dir / "pgsql.restored"
-    with VanillaPostgres(
-        data_dir, PgBin(test_output_dir, env.pg_distrib_dir, env.pg_version), port
-    ) as restored:
+    with VanillaPostgres(data_dir, PgBin(test_output_dir, env.pg_version), port) as restored:
         pg_bin.run_capture(
             [
-                str(base_dir / "libs/utils/scripts/restore_from_wal.sh"),
-                str(pg_distrib_dir / f"v{env.pg_version}/bin"),
+                os.path.join(base_dir, "libs/utils/scripts/restore_from_wal.sh"),
+                os.path.join(pg_distrib_dir, "v{}".format(env.pg_version), "bin"),
                 str(test_output_dir / "repo" / "safekeepers" / "sk1" / str(tenant_id) / "*"),
                 str(data_dir),
                 str(port),

test_runner/regress/test_walredo_not_left_behind_on_detach.py

@@ -1,104 +0,0 @@
-import time
-
-import psutil
-import pytest
-from fixtures.log_helper import log
-from fixtures.neon_fixtures import NeonEnvBuilder, PageserverApiException
-from fixtures.types import TenantId
-
-
-def assert_child_processes(pageserver_pid, wal_redo_present=False, defunct_present=False):
-    children = psutil.Process(pageserver_pid).children()
-    for child in children:
-        if not wal_redo_present:
-            assert "--wal-redo" not in child.cmdline()
-        if not defunct_present:
-            assert child.status() != psutil.STATUS_ZOMBIE
-
-
-# Check that the pageserver doesn't leave behind WAL redo processes
-# when a tenant is detached. We had an issue previously where we failed
-# to wait and consume the exit code of the WAL redo process, leaving it behind
-# as a zombie process.
-def test_walredo_not_left_behind_on_detach(neon_env_builder: NeonEnvBuilder):
-    env = neon_env_builder.init_start()
-    pageserver_http = env.pageserver.http_client()
-
-    pagserver_pid = int((env.repo_dir / "pageserver.pid").read_text())
-
-    assert_child_processes(pagserver_pid, wal_redo_present=False, defunct_present=False)
-
-    # first check for non existing tenant
-    tenant_id = TenantId.generate()
-    with pytest.raises(
-        expected_exception=PageserverApiException,
-        match=f"Tenant not found for id {tenant_id}",
-    ):
-        pageserver_http.tenant_detach(tenant_id)
-
-    # create new nenant
-    tenant_id, _ = env.neon_cli.create_tenant()
-
-    # assert tenant exists on disk
-    assert (env.repo_dir / "tenants" / str(tenant_id)).exists()
-
-    pg = env.postgres.create_start("main", tenant_id=tenant_id)
-
-    pg_conn = pg.connect()
-    cur = pg_conn.cursor()
-
-    # Create table, and insert some rows. Make it big enough that it doesn't fit in
-    # shared_buffers, otherwise the SELECT after restart will just return answer
-    # from shared_buffers without hitting the page server, which defeats the point
-    # of this test.
-    cur.execute("CREATE TABLE foo (t text)")
-    cur.execute(
-        """
-        INSERT INTO foo
-            SELECT 'long string to consume some space' || g
-            FROM generate_series(1, 100000) g
-    """
-    )
-
-    # Verify that the table is larger than shared_buffers
-    cur.execute(
-        """
-        select setting::int * pg_size_bytes(unit) as shared_buffers, pg_relation_size('foo') as tbl_ize
-        from pg_settings where name = 'shared_buffers'
-    """
-    )
-    row = cur.fetchone()
-    assert row is not None
-    log.info(f"shared_buffers is {row[0]}, table size {row[1]}")
-    assert int(row[0]) < int(row[1])
-
-    cur.execute("SELECT count(*) FROM foo")
-    assert cur.fetchone() == (100000,)
-
-    # After filling the table and doing the SELECT, it is guaranteed that we did some WAL redo.
-    # So, assert that the WAL redo process is present.
-    # XXX this is quite brittle as the lifecycle of the WAL redo process is an implementation detail
-    assert_child_processes(pagserver_pid, wal_redo_present=True, defunct_present=False)
-
-    last_error = None
-    for i in range(3):
-        try:
-            pageserver_http.tenant_detach(tenant_id)
-        except Exception as e:
-            last_error = e
-            log.error(f"try {i} error detaching tenant: {e}")
-            continue
-        else:
-            break
-    # else is called if the loop finished without reaching "break"
-    else:
-        pytest.fail(f"could not detach tenant: {last_error}")
-
-    # check that nothing is left on disk for deleted tenant
-    assert not (env.repo_dir / "tenants" / str(tenant_id)).exists()
-
-    # Pageserver schedules kill+wait of the WAL redo process to the background runtime,
-    # asynchronously to tenant detach. Cut it some slack to complete kill+wait before
-    # checking.
-    time.sleep(1.0)
-    assert_child_processes(pagserver_pid, wal_redo_present=False, defunct_present=False)