Showcase localhost to remote host proxying

.github/ansible/prod.us-west-2.hosts.yaml

@@ -41,12 +41,6 @@ storage:
           ansible_host: i-051642d372c0a4f32
         pageserver-3.us-west-2.aws.neon.tech:
           ansible_host: i-00c3844beb9ad1c6b
-        pageserver-4.us-west-2.aws.neon.tech:
-          ansible_host: i-013263dd1c239adcc
-        pageserver-6.us-west-2.aws.neon.tech:
-          ansible_host: i-01cdf7d2bc1433b6a
-        pageserver-7.us-west-2.aws.neon.tech:
-          ansible_host: i-02eec9b40617db5bc
 
     safekeepers:
       hosts:
@@ -56,15 +50,4 @@ storage:
           ansible_host: i-074682f9d3c712e7c
         safekeeper-2.us-west-2.aws.neon.tech:
           ansible_host: i-042b7efb1729d7966
-        safekeeper-3.us-west-2.aws.neon.tech:
-          ansible_host: i-089f6b9ef426dff76
-        safekeeper-4.us-west-2.aws.neon.tech:
-          ansible_host: i-0fe6bf912c4710c82
-        safekeeper-5.us-west-2.aws.neon.tech:
-          ansible_host: i-0a83c1c46d2b4e409
-        safekeeper-6.us-west-2.aws.neon.tech:
-          ansible_host: i-0fef5317b8fdc9f8d
-        safekeeper-7.us-west-2.aws.neon.tech:
-          ansible_host: i-0be739190d4289bf9
-        safekeeper-8.us-west-2.aws.neon.tech:
-          ansible_host: i-00e851803669e5cfe                    
+

.github/ansible/staging.eu-west-1.hosts.yaml

@@ -35,8 +35,6 @@ storage:
       hosts:
         pageserver-0.eu-west-1.aws.neon.build:
           ansible_host: i-01d496c5041c7f34c
-        pageserver-1.eu-west-1.aws.neon.build:
-          ansible_host: i-0e8013e239ce3928c
 
     safekeepers:
       hosts:
@@ -46,15 +44,3 @@ storage:
           ansible_host: i-06969ee1bf2958bfc
         safekeeper-2.eu-west-1.aws.neon.build:
           ansible_host: i-087892e9625984a0b
-        safekeeper-3.eu-west-1.aws.neon.build:
-          ansible_host: i-0a6f91660e99e8891
-        safekeeper-4.eu-west-1.aws.neon.build:
-          ansible_host: i-0012e309e28e7c249
-        safekeeper-5.eu-west-1.aws.neon.build:
-          ansible_host: i-085a2b1193287b32e
-        safekeeper-6.eu-west-1.aws.neon.build:
-          ansible_host: i-0c713248465ed0fbd
-        safekeeper-7.eu-west-1.aws.neon.build:
-          ansible_host: i-02ad231aed2a80b7a
-        safekeeper-8.eu-west-1.aws.neon.build:
-          ansible_host: i-0dbbd8ffef66efda8

.github/helm-values/dev-eu-central-1-alpha.pg-sni-router.yaml

@@ -1,19 +0,0 @@
-useCertManager: true
-
-replicaCount: 3
-
-exposedService:
-  # exposedService.port -- Exposed Service proxy port
-  port: 4432
-  annotations:
-    external-dns.alpha.kubernetes.io/hostname: "*.snirouter.alpha.eu-central-1.internal.aws.neon.build"
-
-settings:
-  domain: "*.snirouter.alpha.eu-central-1.internal.aws.neon.build"
-  sentryEnvironment: "staging"
-
-imagePullSecrets:
-  - name: docker-hub-neon
-
-metrics:
-  enabled: false

.github/helm-values/dev-eu-west-1-zeta.pg-sni-router.yaml

@@ -1,19 +0,0 @@
-useCertManager: true
-
-replicaCount: 3
-
-exposedService:
-  # exposedService.port -- Exposed Service proxy port
-  port: 4432
-  annotations:
-    external-dns.alpha.kubernetes.io/hostname: "*.snirouter.zeta.eu-west-1.internal.aws.neon.build"
-
-settings:
-  domain: "*.snirouter.zeta.eu-west-1.internal.aws.neon.build"
-  sentryEnvironment: "staging"
-
-imagePullSecrets:
-  - name: docker-hub-neon
-
-metrics:
-  enabled: false

.github/helm-values/dev-us-east-2-beta.pg-sni-router.yaml

@@ -1,19 +0,0 @@
-useCertManager: true
-
-replicaCount: 3
-
-exposedService:
-  # exposedService.port -- Exposed Service proxy port
-  port: 4432
-  annotations:
-    external-dns.alpha.kubernetes.io/hostname: "*.snirouter.beta.us-east-2.internal.aws.neon.build"
-
-settings:
-  domain: "*.snirouter.beta.us-east-2.internal.aws.neon.build"
-  sentryEnvironment: "staging"
-
-imagePullSecrets:
-  - name: docker-hub-neon
-
-metrics:
-  enabled: false

.github/helm-values/prod-ap-southeast-1-epsilon.pg-sni-router.yaml

@@ -1,19 +0,0 @@
-useCertManager: true
-
-replicaCount: 3
-
-exposedService:
-  # exposedService.port -- Exposed Service proxy port
-  port: 4432
-  annotations:
-    external-dns.alpha.kubernetes.io/hostname: "*.snirouter.epsilon.ap-southeast-1.internal.aws.neon.tech"
-
-settings:
-  domain: "*.snirouter.epsilon.ap-southeast-1.internal.aws.neon.tech"
-  sentryEnvironment: "production"
-
-imagePullSecrets:
-  - name: docker-hub-neon
-
-metrics:
-  enabled: false

.github/helm-values/prod-eu-central-1-gamma.pg-sni-router.yaml

@@ -1,19 +0,0 @@
-useCertManager: true
-
-replicaCount: 3
-
-exposedService:
-  # exposedService.port -- Exposed Service proxy port
-  port: 4432
-  annotations:
-    external-dns.alpha.kubernetes.io/hostname: "*.snirouter.gamma.eu-central-1.internal.aws.neon.tech"
-
-settings:
-  domain: "*.snirouter.gamma.eu-central-1.internal.aws.neon.tech"
-  sentryEnvironment: "production"
-
-imagePullSecrets:
-  - name: docker-hub-neon
-
-metrics:
-  enabled: false

.github/helm-values/prod-us-east-1-theta.pg-sni-router.yaml

@@ -1,19 +0,0 @@
-useCertManager: true
-
-replicaCount: 3
-
-exposedService:
-  # exposedService.port -- Exposed Service proxy port
-  port: 4432
-  annotations:
-    external-dns.alpha.kubernetes.io/hostname: "*.snirouter.theta.us-east-1.internal.aws.neon.tech"
-
-settings:
-  domain: "*.snirouter.theta.us-east-1.internal.aws.neon.tech"
-  sentryEnvironment: "production"
-
-imagePullSecrets:
-  - name: docker-hub-neon
-
-metrics:
-  enabled: false

.github/helm-values/prod-us-east-2-delta.pg-sni-router.yaml

@@ -1,19 +0,0 @@
-useCertManager: true
-
-replicaCount: 3
-
-exposedService:
-  # exposedService.port -- Exposed Service proxy port
-  port: 4432
-  annotations:
-    external-dns.alpha.kubernetes.io/hostname: "*.snirouter.delta.us-east-2.internal.aws.neon.tech"
-
-settings:
-  domain: "*.snirouter.delta.us-east-2.internal.aws.neon.tech"
-  sentryEnvironment: "production"
-
-imagePullSecrets:
-  - name: docker-hub-neon
-
-metrics:
-  enabled: false

.github/helm-values/prod-us-west-2-eta.pg-sni-router.yaml

@@ -1,19 +0,0 @@
-useCertManager: true
-
-replicaCount: 3
-
-exposedService:
-  # exposedService.port -- Exposed Service proxy port
-  port: 4432
-  annotations:
-    external-dns.alpha.kubernetes.io/hostname: "*.snirouter.eta.us-west-2.internal.aws.neon.tech"
-
-settings:
-  domain: "*.snirouter.eta.us-west-2.internal.aws.neon.tech"
-  sentryEnvironment: "production"
-
-imagePullSecrets:
-  - name: docker-hub-neon
-
-metrics:
-  enabled: false

.github/workflows/build_and_test.yml

@@ -418,7 +418,10 @@ jobs:
       - uses: actions/github-script@v6
         if: >
           !cancelled() &&
-          github.event_name == 'pull_request'
+          github.event_name == 'pull_request' && (
+            steps.create-allure-report-debug.outputs.report-url ||
+            steps.create-allure-report-release.outputs.report-url
+          )
         with:
           # Retry script for 5XX server errors: https://github.com/actions/github-script#retries
           retries: 5

.github/workflows/deploy-dev.yml

@@ -27,11 +27,6 @@ on:
         required: true
         type: boolean
         default: true
-      deployPgSniRouter:
-        description: 'Deploy pg-sni-router'
-        required: true
-        type: boolean
-        default: true
 
 env:
   AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_DEV }}
@@ -232,49 +227,3 @@ jobs:
   
       - name: Cleanup helm folder
         run: rm -rf ~/.cache
-
-  deploy-pg-sni-router:
-    runs-on: [ self-hosted, gen3, small ]
-    container: 369495373322.dkr.ecr.eu-central-1.amazonaws.com/ansible:pinned
-    if: inputs.deployPgSniRouter
-    defaults:
-      run:
-        shell: bash
-    strategy:
-      matrix:
-        include:
-          - target_region:  us-east-2
-            target_cluster: dev-us-east-2-beta
-          - target_region:  eu-west-1
-            target_cluster: dev-eu-west-1-zeta
-          - target_region:  eu-central-1
-            target_cluster: dev-eu-central-1-alpha
-    environment:
-      name: dev-${{ matrix.target_region }}
-    steps:
-      - name: Checkout
-        uses: actions/checkout@v3
-        with:
-          submodules: true
-          fetch-depth: 0
-          ref: ${{ inputs.branch }}
-  
-      - name: Configure AWS Credentials
-        uses: aws-actions/configure-aws-credentials@v1-node16
-        with:
-          role-to-assume: arn:aws:iam::369495373322:role/github-runner
-          aws-region: eu-central-1
-          role-skip-session-tagging: true
-          role-duration-seconds: 1800
-  
-      - name: Configure environment
-        run: |
-          helm repo add neondatabase https://neondatabase.github.io/helm-charts
-          aws --region ${{ matrix.target_region }} eks update-kubeconfig --name  ${{ matrix.target_cluster }}
-  
-      - name: Deploy pg-sni-router
-        run:
-          helm upgrade neon-pg-sni-router neondatabase/neon-pg-sni-router --namespace neon-pg-sni-router --create-namespace --install --debug --atomic -f .github/helm-values/${{ matrix.target_cluster }}.pg-sni-router.yaml --set image.tag=${{ inputs.dockerTag }} --set settings.sentryUrl=${{ secrets.SENTRY_URL_BROKER }} --wait --timeout 15m0s
-  
-      - name: Cleanup helm folder
-        run: rm -rf ~/.cache

.github/workflows/deploy-prod.yml

@@ -27,11 +27,6 @@ on:
         required: true
         type: boolean
         default: true
-      deployPgSniRouter:
-        description: 'Deploy pg-sni-router'
-        required: true
-        type: boolean
-        default: true
       disclamerAcknowledged:
         description: 'I confirm that there is an emergency and I can not use regular release workflow'
         required: true
@@ -176,42 +171,3 @@ jobs:
       - name: Deploy storage-broker
         run:
           helm upgrade neon-storage-broker-lb neondatabase/neon-storage-broker --namespace neon-storage-broker-lb --create-namespace --install --atomic -f .github/helm-values/${{ matrix.target_cluster }}.neon-storage-broker.yaml --set image.tag=${{ inputs.dockerTag }} --set settings.sentryUrl=${{ secrets.SENTRY_URL_BROKER }} --wait --timeout 5m0s
-
-  deploy-pg-sni-router:
-    runs-on: prod
-    container: 093970136003.dkr.ecr.eu-central-1.amazonaws.com/ansible:latest
-    if: inputs.deployPgSniRouter && inputs.disclamerAcknowledged
-    defaults:
-      run:
-        shell: bash
-    strategy:
-      matrix:
-        include:
-          - target_region:  us-east-2
-            target_cluster: prod-us-east-2-delta
-          - target_region:  us-west-2
-            target_cluster: prod-us-west-2-eta
-          - target_region: eu-central-1
-            target_cluster: prod-eu-central-1-gamma
-          - target_region: ap-southeast-1
-            target_cluster: prod-ap-southeast-1-epsilon
-          - target_region: us-east-1
-            target_cluster: prod-us-east-1-theta
-    environment:
-      name: prod-${{ matrix.target_region }}
-    steps:
-      - name: Checkout
-        uses: actions/checkout@v3
-        with:
-          submodules: true
-          fetch-depth: 0
-          ref: ${{ inputs.branch }}
-  
-      - name: Configure environment
-        run: |
-          helm repo add neondatabase https://neondatabase.github.io/helm-charts
-          aws --region ${{ matrix.target_region }} eks update-kubeconfig --name  ${{ matrix.target_cluster }}
-  
-      - name: Deploy pg-sni-router
-        run:
-          helm upgrade neon-pg-sni-router neondatabase/neon-pg-sni-router --namespace neon-pg-sni-router --create-namespace --install --debug --atomic -f .github/helm-values/${{ matrix.target_cluster }}.pg-sni-router.yaml --set image.tag=${{ inputs.dockerTag }} --set settings.sentryUrl=${{ secrets.SENTRY_URL_BROKER }} --wait --timeout 15m0s

Cargo.lock

@@ -1032,7 +1032,6 @@ dependencies = [
  "serde",
  "serde_json",
  "serde_with",
- "utils",
  "workspace_hack",
 ]
 
@@ -1106,7 +1105,6 @@ dependencies = [
  "anyhow",
  "clap 4.2.2",
  "comfy-table",
- "compute_api",
  "git-version",
  "nix",
  "once_cell",
@@ -3676,7 +3674,6 @@ dependencies = [
  "remote_storage",
  "reqwest",
  "safekeeper_api",
- "scopeguard",
  "serde",
  "serde_json",
  "serde_with",
@@ -3963,7 +3960,8 @@ dependencies = [
 [[package]]
 name = "sharded-slab"
 version = "0.1.4"
-source = "git+https://github.com/neondatabase/sharded-slab.git?rev=98d16753ab01c61f0a028de44167307a00efea00#98d16753ab01c61f0a028de44167307a00efea00"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "900fba806f70c630b0a382d0d825e17a0f19fcd059a2ade1ff237bcddf446b31"
 dependencies = [
  "lazy_static",
 ]

Cargo.toml

@@ -161,17 +161,11 @@ rstest = "0.17"
 tempfile = "3.4"
 tonic-build = "0.9"
 
-[patch.crates-io]
-
 # This is only needed for proxy's tests.
 # TODO: we should probably fork `tokio-postgres-rustls` instead.
+[patch.crates-io]
 tokio-postgres = { git = "https://github.com/neondatabase/rust-postgres.git", rev="0bc41d8503c092b040142214aac3cf7d11d0c19f" }
 
-# Changes the MAX_THREADS limit from 4096 to 32768.
-# This is a temporary workaround for using tracing from many threads in safekeepers code,
-# until async safekeepers patch is merged to the main.
-sharded-slab = { git = "https://github.com/neondatabase/sharded-slab.git", rev="98d16753ab01c61f0a028de44167307a00efea00" }
-
 ################# Binary contents sections
 
 [profile.release]

Dockerfile

@@ -44,15 +44,7 @@ COPY --chown=nonroot . .
 # Show build caching stats to check if it was used in the end.
 # Has to be the part of the same RUN since cachepot daemon is killed in the end of this RUN, losing the compilation stats.
 RUN set -e \
-    && mold -run cargo build  \
-      --bin pg_sni_router  \
-      --bin pageserver  \
-      --bin pageserver_binutils  \
-      --bin draw_timeline_dir \
-      --bin safekeeper  \
-      --bin storage_broker  \
-      --bin proxy  \
-      --locked --release \
+&& mold -run cargo build --bin pageserver --bin pageserver_binutils --bin draw_timeline_dir --bin safekeeper --bin storage_broker --bin proxy --locked --release \
     && cachepot -s
 
 # Build final image
@@ -71,7 +63,6 @@ RUN set -e \
     && useradd -d /data neon \
     && chown -R neon:neon /data
 
-COPY --from=build --chown=neon:neon /home/nonroot/target/release/pg_sni_router       /usr/local/bin
 COPY --from=build --chown=neon:neon /home/nonroot/target/release/pageserver          /usr/local/bin
 COPY --from=build --chown=neon:neon /home/nonroot/target/release/pageserver_binutils /usr/local/bin
 COPY --from=build --chown=neon:neon /home/nonroot/target/release/draw_timeline_dir   /usr/local/bin

compute_tools/src/compute.rs

@@ -30,7 +30,7 @@ use utils::id::{TenantId, TimelineId};
 use utils::lsn::Lsn;
 
 use compute_api::responses::{ComputeMetrics, ComputeStatus};
-use compute_api::spec::{ComputeMode, ComputeSpec};
+use compute_api::spec::ComputeSpec;
 
 use crate::config;
 use crate::pg_helpers::*;
@@ -67,9 +67,8 @@ pub struct ComputeNode {
 pub struct ComputeState {
     pub start_time: DateTime<Utc>,
     pub status: ComputeStatus,
-    /// Timestamp of the last Postgres activity. It could be `None` if
-    /// compute wasn't used since start.
-    pub last_active: Option<DateTime<Utc>>,
+    /// Timestamp of the last Postgres activity
+    pub last_active: DateTime<Utc>,
     pub error: Option<String>,
     pub pspec: Option<ParsedSpec>,
     pub metrics: ComputeMetrics,
@@ -80,7 +79,7 @@ impl ComputeState {
         Self {
             start_time: Utc::now(),
             status: ComputeStatus::Empty,
-            last_active: None,
+            last_active: Utc::now(),
             error: None,
             pspec: None,
             metrics: ComputeMetrics::default(),
@@ -250,10 +249,38 @@ impl ComputeNode {
     /// safekeepers sync, basebackup, etc.
     #[instrument(skip(self, compute_state))]
     pub fn prepare_pgdata(&self, compute_state: &ComputeState) -> Result<()> {
+        #[derive(Clone)]
+        enum Replication {
+            Primary,
+            Static { lsn: Lsn },
+            HotStandby,
+        }
+
         let pspec = compute_state.pspec.as_ref().expect("spec must be set");
         let spec = &pspec.spec;
         let pgdata_path = Path::new(&self.pgdata);
 
+        let hot_replica = if let Some(option) = spec.cluster.settings.find_ref("hot_standby") {
+            if let Some(value) = &option.value {
+                anyhow::ensure!(option.vartype == "bool");
+                matches!(value.as_str(), "on" | "yes" | "true")
+            } else {
+                false
+            }
+        } else {
+            false
+        };
+
+        let replication = if hot_replica {
+            Replication::HotStandby
+        } else if let Some(lsn) = spec.cluster.settings.find("recovery_target_lsn") {
+            Replication::Static {
+                lsn: Lsn::from_str(&lsn)?,
+            }
+        } else {
+            Replication::Primary
+        };
+
         // Remove/create an empty pgdata directory and put configuration there.
         self.create_pgdata()?;
         config::write_postgres_conf(&pgdata_path.join("postgresql.conf"), &pspec.spec)?;
@@ -261,8 +288,8 @@ impl ComputeNode {
         // Syncing safekeepers is only safe with primary nodes: if a primary
         // is already connected it will be kicked out, so a secondary (standby)
         // cannot sync safekeepers.
-        let lsn = match spec.mode {
-            ComputeMode::Primary => {
+        let lsn = match &replication {
+            Replication::Primary => {
                 info!("starting safekeepers syncing");
                 let lsn = self
                     .sync_safekeepers(pspec.storage_auth_token.clone())
@@ -270,11 +297,11 @@ impl ComputeNode {
                 info!("safekeepers synced at LSN {}", lsn);
                 lsn
             }
-            ComputeMode::Static(lsn) => {
+            Replication::Static { lsn } => {
                 info!("Starting read-only node at static LSN {}", lsn);
-                lsn
+                *lsn
             }
-            ComputeMode::Replica => {
+            Replication::HotStandby => {
                 info!("Initializing standby from latest Pageserver LSN");
                 Lsn(0)
             }
@@ -294,9 +321,9 @@ impl ComputeNode {
         // Update pg_hba.conf received with basebackup.
         update_pg_hba(pgdata_path)?;
 
-        match spec.mode {
-            ComputeMode::Primary | ComputeMode::Static(..) => {}
-            ComputeMode::Replica => {
+        match &replication {
+            Replication::Primary | Replication::Static { .. } => {}
+            Replication::HotStandby => {
                 add_standby_signal(pgdata_path)?;
             }
         }
@@ -403,13 +430,11 @@ impl ComputeNode {
         self.pg_reload_conf(&mut client)?;
 
         // Proceed with post-startup configuration. Note, that order of operations is important.
-        if spec.mode == ComputeMode::Primary {
-            handle_roles(&spec, &mut client)?;
-            handle_databases(&spec, &mut client)?;
-            handle_role_deletions(&spec, self.connstr.as_str(), &mut client)?;
-            handle_grants(&spec, self.connstr.as_str(), &mut client)?;
-            handle_extensions(&spec, &mut client)?;
-        }
+        handle_roles(&spec, &mut client)?;
+        handle_databases(&spec, &mut client)?;
+        handle_role_deletions(&spec, self.connstr.as_str(), &mut client)?;
+        handle_grants(&spec, self.connstr.as_str(), &mut client)?;
+        handle_extensions(&spec, &mut client)?;
 
         // 'Close' connection
         drop(client);
@@ -442,9 +467,7 @@ impl ComputeNode {
 
         let pg = self.start_postgres(spec.storage_auth_token.clone())?;
 
-        if spec.spec.mode == ComputeMode::Primary {
-            self.apply_config(&compute_state)?;
-        }
+        self.apply_config(&compute_state)?;
 
         let startup_end_time = Utc::now();
         {

compute_tools/src/config.rs

@@ -6,7 +6,7 @@ use std::path::Path;
 use anyhow::Result;
 
 use crate::pg_helpers::PgOptionsSerialize;
-use compute_api::spec::{ComputeMode, ComputeSpec};
+use compute_api::spec::ComputeSpec;
 
 /// Check that `line` is inside a text file and put it there if it is not.
 /// Create file if it doesn't exist.
@@ -34,25 +34,17 @@ pub fn line_in_file(path: &Path, line: &str) -> Result<bool> {
 /// Create or completely rewrite configuration file specified by `path`
 pub fn write_postgres_conf(path: &Path, spec: &ComputeSpec) -> Result<()> {
     // File::create() destroys the file content if it exists.
-    let mut file = File::create(path)?;
+    let mut postgres_conf = File::create(path)?;
 
+    write_auto_managed_block(&mut postgres_conf, &spec.cluster.settings.as_pg_settings())?;
+
+    Ok(())
+}
+
+// Write Postgres config block wrapped with generated comment section
+fn write_auto_managed_block(file: &mut File, buf: &str) -> Result<()> {
     writeln!(file, "# Managed by compute_ctl: begin")?;
-
-    write!(file, "{}", &spec.cluster.settings.as_pg_settings())?;
-
-    match spec.mode {
-        ComputeMode::Primary => {}
-        ComputeMode::Static(lsn) => {
-            // hot_standby is 'on' by default, but let's be explicit
-            writeln!(file, "hot_standby=on")?;
-            writeln!(file, "recovery_target_lsn='{lsn}'")?;
-        }
-        ComputeMode::Replica => {
-            // hot_standby is 'on' by default, but let's be explicit
-            writeln!(file, "hot_standby=on")?;
-        }
-    }
-
+    writeln!(file, "{}", buf)?;
     writeln!(file, "# Managed by compute_ctl: end")?;
 
     Ok(())

compute_tools/src/http/openapi_spec.yaml

@@ -181,8 +181,8 @@ components:
     ComputeState:
       type: object
       required:
-        - start_time
         - status
+        - last_active
       properties:
         start_time:
           type: string
@@ -195,13 +195,11 @@ components:
           $ref: '#/components/schemas/ComputeStatus'
         last_active:
           type: string
-          description: |
-            The last detected compute activity timestamp in UTC and RFC3339 format.
-            It could be empty if compute was never used by user since start.
+          description: The last detected compute activity timestamp in UTC and RFC3339 format.
           example: "2022-10-12T07:20:50.52Z"
         error:
           type: string
-          description: Text of the error during compute startup or reconfiguration, if any.
+          description: Text of the error during compute startup, if any.
           example: ""
         tenant:
           type: string
@@ -224,12 +222,9 @@ components:
     ComputeStatus:
       type: string
       enum:
-        - empty
         - init
         - failed
         - running
-        - configuration_pending
-        - configuration
       example: running
 
     #

compute_tools/src/monitor.rs

@@ -74,7 +74,7 @@ fn watch_compute_activity(compute: &ComputeNode) {
                             // Found non-idle backend, so the last activity is NOW.
                             // Save it and exit the for loop. Also clear the idle backend
                             // `state_change` timestamps array as it doesn't matter now.
-                            last_active = Some(Utc::now());
+                            last_active = Utc::now();
                             idle_backs.clear();
                             break;
                         }
@@ -82,16 +82,15 @@ fn watch_compute_activity(compute: &ComputeNode) {
 
                     // Get idle backend `state_change` with the max timestamp.
                     if let Some(last) = idle_backs.iter().max() {
-                        last_active = Some(*last);
+                        last_active = *last;
                     }
                 }
 
                 // Update the last activity in the shared state if we got a more recent one.
                 let mut state = compute.state.lock().unwrap();
-                // NB: `Some(<DateTime>)` is always greater than `None`.
                 if last_active > state.last_active {
                     state.last_active = last_active;
-                    debug!("set the last compute activity time to: {:?}", last_active);
+                    debug!("set the last compute activity time to: {}", last_active);
                 }
             }
             Err(e) => {

control_plane/Cargo.toml

@@ -30,5 +30,4 @@ postgres_connection.workspace = true
 storage_broker.workspace = true
 utils.workspace = true
 
-compute_api.workspace = true
 workspace_hack.workspace = true

control_plane/src/bin/neon_local.rs

@@ -7,8 +7,8 @@
 //!
 use anyhow::{anyhow, bail, Context, Result};
 use clap::{value_parser, Arg, ArgAction, ArgMatches, Command};
-use compute_api::spec::ComputeMode;
 use control_plane::endpoint::ComputeControlPlane;
+use control_plane::endpoint::Replication;
 use control_plane::local_env::LocalEnv;
 use control_plane::pageserver::PageServerNode;
 use control_plane::safekeeper::SafekeeperNode;
@@ -481,7 +481,7 @@ fn handle_timeline(timeline_match: &ArgMatches, env: &mut local_env::LocalEnv) -
                 timeline_id,
                 None,
                 pg_version,
-                ComputeMode::Primary,
+                Replication::Primary,
             )?;
             println!("Done");
         }
@@ -568,8 +568,8 @@ fn handle_endpoint(ep_match: &ArgMatches, env: &local_env::LocalEnv) -> Result<(
                 .iter()
                 .filter(|(_, endpoint)| endpoint.tenant_id == tenant_id)
             {
-                let lsn_str = match endpoint.mode {
-                    ComputeMode::Static(lsn) => {
+                let lsn_str = match endpoint.replication {
+                    Replication::Static(lsn) => {
                         // -> read-only endpoint
                         // Use the node's LSN.
                         lsn.to_string()
@@ -632,14 +632,21 @@ fn handle_endpoint(ep_match: &ArgMatches, env: &local_env::LocalEnv) -> Result<(
                 .copied()
                 .unwrap_or(false);
 
-            let mode = match (lsn, hot_standby) {
-                (Some(lsn), false) => ComputeMode::Static(lsn),
-                (None, true) => ComputeMode::Replica,
-                (None, false) => ComputeMode::Primary,
+            let replication = match (lsn, hot_standby) {
+                (Some(lsn), false) => Replication::Static(lsn),
+                (None, true) => Replication::Replica,
+                (None, false) => Replication::Primary,
                 (Some(_), true) => anyhow::bail!("cannot specify both lsn and hot-standby"),
             };
 
-            cplane.new_endpoint(tenant_id, &endpoint_id, timeline_id, port, pg_version, mode)?;
+            cplane.new_endpoint(
+                tenant_id,
+                &endpoint_id,
+                timeline_id,
+                port,
+                pg_version,
+                replication,
+            )?;
         }
         "start" => {
             let port: Option<u16> = sub_args.get_one::<u16>("port").copied();
@@ -663,11 +670,11 @@ fn handle_endpoint(ep_match: &ArgMatches, env: &local_env::LocalEnv) -> Result<(
                 .unwrap_or(false);
 
             if let Some(endpoint) = endpoint {
-                match (&endpoint.mode, hot_standby) {
-                    (ComputeMode::Static(_), true) => {
+                match (&endpoint.replication, hot_standby) {
+                    (Replication::Static(_), true) => {
                         bail!("Cannot start a node in hot standby mode when it is already configured as a static replica")
                     }
-                    (ComputeMode::Primary, true) => {
+                    (Replication::Primary, true) => {
                         bail!("Cannot start a node as a hot standby replica, it is already configured as primary node")
                     }
                     _ => {}
@@ -694,10 +701,10 @@ fn handle_endpoint(ep_match: &ArgMatches, env: &local_env::LocalEnv) -> Result<(
                     .copied()
                     .context("Failed to `pg-version` from the argument string")?;
 
-                let mode = match (lsn, hot_standby) {
-                    (Some(lsn), false) => ComputeMode::Static(lsn),
-                    (None, true) => ComputeMode::Replica,
-                    (None, false) => ComputeMode::Primary,
+                let replication = match (lsn, hot_standby) {
+                    (Some(lsn), false) => Replication::Static(lsn),
+                    (None, true) => Replication::Replica,
+                    (None, false) => Replication::Primary,
                     (Some(_), true) => anyhow::bail!("cannot specify both lsn and hot-standby"),
                 };
 
@@ -714,7 +721,7 @@ fn handle_endpoint(ep_match: &ArgMatches, env: &local_env::LocalEnv) -> Result<(
                     timeline_id,
                     port,
                     pg_version,
-                    mode,
+                    replication,
                 )?;
                 ep.start(&auth_token)?;
             }

control_plane/src/endpoint.rs

@@ -11,33 +11,15 @@ use std::sync::Arc;
 use std::time::Duration;
 
 use anyhow::{Context, Result};
-use serde::{Deserialize, Serialize};
-use serde_with::{serde_as, DisplayFromStr};
 use utils::{
     id::{TenantId, TimelineId},
     lsn::Lsn,
 };
 
-use crate::local_env::LocalEnv;
+use crate::local_env::{LocalEnv, DEFAULT_PG_VERSION};
 use crate::pageserver::PageServerNode;
 use crate::postgresql_conf::PostgresConf;
 
-use compute_api::spec::ComputeMode;
-
-// contents of a endpoint.json file
-#[serde_as]
-#[derive(Serialize, Deserialize, PartialEq, Eq, Clone, Debug)]
-pub struct EndpointConf {
-    name: String,
-    #[serde_as(as = "DisplayFromStr")]
-    tenant_id: TenantId,
-    #[serde_as(as = "DisplayFromStr")]
-    timeline_id: TimelineId,
-    mode: ComputeMode,
-    port: u16,
-    pg_version: u32,
-}
-
 //
 // ComputeControlPlane
 //
@@ -88,7 +70,7 @@ impl ComputeControlPlane {
         timeline_id: TimelineId,
         port: Option<u16>,
         pg_version: u32,
-        mode: ComputeMode,
+        replication: Replication,
     ) -> Result<Arc<Endpoint>> {
         let port = port.unwrap_or_else(|| self.get_port());
 
@@ -98,22 +80,12 @@ impl ComputeControlPlane {
             env: self.env.clone(),
             pageserver: Arc::clone(&self.pageserver),
             timeline_id,
-            mode,
+            replication,
             tenant_id,
             pg_version,
         });
+
         ep.create_pgdata()?;
-        std::fs::write(
-            ep.endpoint_path().join("endpoint.json"),
-            serde_json::to_string_pretty(&EndpointConf {
-                name: name.to_string(),
-                tenant_id,
-                timeline_id,
-                mode,
-                port,
-                pg_version,
-            })?,
-        )?;
         ep.setup_pg_conf()?;
 
         self.endpoints.insert(ep.name.clone(), Arc::clone(&ep));
@@ -124,13 +96,26 @@ impl ComputeControlPlane {
 
 ///////////////////////////////////////////////////////////////////////////////
 
+#[derive(Debug, Clone, Eq, PartialEq)]
+pub enum Replication {
+    // Regular read-write node
+    Primary,
+    // if recovery_target_lsn is provided, and we want to pin the node to a specific LSN
+    Static(Lsn),
+    // Hot standby; read-only replica.
+    // Future versions may want to distinguish between replicas with hot standby
+    // feedback and other kinds of replication configurations.
+    Replica,
+}
+
 #[derive(Debug)]
 pub struct Endpoint {
     /// used as the directory name
     name: String,
     pub tenant_id: TenantId,
     pub timeline_id: TimelineId,
-    pub mode: ComputeMode,
+    // Some(lsn) if this is a read-only endpoint anchored at 'lsn'. None for the primary.
+    pub replication: Replication,
 
     // port and address of the Postgres server
     pub address: SocketAddr,
@@ -159,20 +144,50 @@ impl Endpoint {
         let fname = entry.file_name();
         let name = fname.to_str().unwrap().to_string();
 
-        // Read the endpoint.json file
-        let conf: EndpointConf =
-            serde_json::from_slice(&std::fs::read(entry.path().join("endpoint.json"))?)?;
+        // Read config file into memory
+        let cfg_path = entry.path().join("pgdata").join("postgresql.conf");
+        let cfg_path_str = cfg_path.to_string_lossy();
+        let mut conf_file = File::open(&cfg_path)
+            .with_context(|| format!("failed to open config file in {}", cfg_path_str))?;
+        let conf = PostgresConf::read(&mut conf_file)
+            .with_context(|| format!("failed to read config file in {}", cfg_path_str))?;
+
+        // Read a few options from the config file
+        let context = format!("in config file {}", cfg_path_str);
+        let port: u16 = conf.parse_field("port", &context)?;
+        let timeline_id: TimelineId = conf.parse_field("neon.timeline_id", &context)?;
+        let tenant_id: TenantId = conf.parse_field("neon.tenant_id", &context)?;
+
+        // Read postgres version from PG_VERSION file to determine which postgres version binary to use.
+        // If it doesn't exist, assume broken data directory and use default pg version.
+        let pg_version_path = entry.path().join("PG_VERSION");
+
+        let pg_version_str =
+            fs::read_to_string(pg_version_path).unwrap_or_else(|_| DEFAULT_PG_VERSION.to_string());
+        let pg_version = u32::from_str(&pg_version_str)?;
+
+        // parse recovery_target_lsn and primary_conninfo into Recovery Target, if any
+        let replication = if let Some(lsn_str) = conf.get("recovery_target_lsn") {
+            Replication::Static(Lsn::from_str(lsn_str)?)
+        } else if let Some(slot_name) = conf.get("primary_slot_name") {
+            let slot_name = slot_name.to_string();
+            let prefix = format!("repl_{}_", timeline_id);
+            assert!(slot_name.starts_with(&prefix));
+            Replication::Replica
+        } else {
+            Replication::Primary
+        };
 
         // ok now
         Ok(Endpoint {
-            address: SocketAddr::new("127.0.0.1".parse().unwrap(), conf.port),
+            address: SocketAddr::new("127.0.0.1".parse().unwrap(), port),
             name,
             env: env.clone(),
             pageserver: Arc::clone(pageserver),
-            timeline_id: conf.timeline_id,
-            mode: conf.mode,
-            tenant_id: conf.tenant_id,
-            pg_version: conf.pg_version,
+            timeline_id,
+            replication,
+            tenant_id,
+            pg_version,
         })
     }
 
@@ -308,8 +323,8 @@ impl Endpoint {
 
         conf.append_line("");
         // Replication-related configurations, such as WAL sending
-        match &self.mode {
-            ComputeMode::Primary => {
+        match &self.replication {
+            Replication::Primary => {
                 // Configure backpressure
                 // - Replication write lag depends on how fast the walreceiver can process incoming WAL.
                 //   This lag determines latency of get_page_at_lsn. Speed of applying WAL is about 10MB/sec,
@@ -351,10 +366,10 @@ impl Endpoint {
                     conf.append("synchronous_standby_names", "pageserver");
                 }
             }
-            ComputeMode::Static(lsn) => {
+            Replication::Static(lsn) => {
                 conf.append("recovery_target_lsn", &lsn.to_string());
             }
-            ComputeMode::Replica => {
+            Replication::Replica => {
                 assert!(!self.env.safekeepers.is_empty());
 
                 // TODO: use future host field from safekeeper spec
@@ -394,8 +409,8 @@ impl Endpoint {
     }
 
     fn load_basebackup(&self, auth_token: &Option<String>) -> Result<()> {
-        let backup_lsn = match &self.mode {
-            ComputeMode::Primary => {
+        let backup_lsn = match &self.replication {
+            Replication::Primary => {
                 if !self.env.safekeepers.is_empty() {
                     // LSN 0 means that it is bootstrap and we need to download just
                     // latest data from the pageserver. That is a bit clumsy but whole bootstrap
@@ -411,8 +426,8 @@ impl Endpoint {
                     None
                 }
             }
-            ComputeMode::Static(lsn) => Some(*lsn),
-            ComputeMode::Replica => {
+            Replication::Static(lsn) => Some(*lsn),
+            Replication::Replica => {
                 None // Take the latest snapshot available to start with
             }
         };
@@ -511,7 +526,7 @@ impl Endpoint {
         // 3. Load basebackup
         self.load_basebackup(auth_token)?;
 
-        if self.mode != ComputeMode::Primary {
+        if self.replication != Replication::Primary {
             File::create(self.pgdata().join("standby.signal"))?;
         }
 

libs/compute_api/Cargo.toml

@@ -11,5 +11,4 @@ serde.workspace = true
 serde_with.workspace = true
 serde_json.workspace = true
 
-utils = { path = "../utils" }
 workspace_hack.workspace = true

libs/compute_api/src/responses.rs

@@ -19,7 +19,7 @@ pub struct ComputeStatusResponse {
     pub timeline: Option<String>,
     pub status: ComputeStatus,
     #[serde(serialize_with = "rfc3339_serialize")]
-    pub last_active: Option<DateTime<Utc>>,
+    pub last_active: DateTime<Utc>,
     pub error: Option<String>,
 }
 
@@ -29,7 +29,7 @@ pub struct ComputeState {
     pub status: ComputeStatus,
     /// Timestamp of the last Postgres activity
     #[serde(serialize_with = "rfc3339_serialize")]
-    pub last_active: Option<DateTime<Utc>>,
+    pub last_active: DateTime<Utc>,
     pub error: Option<String>,
 }
 
@@ -54,15 +54,11 @@ pub enum ComputeStatus {
     Failed,
 }
 
-fn rfc3339_serialize<S>(x: &Option<DateTime<Utc>>, s: S) -> Result<S::Ok, S::Error>
+fn rfc3339_serialize<S>(x: &DateTime<Utc>, s: S) -> Result<S::Ok, S::Error>
 where
     S: Serializer,
 {
-    if let Some(x) = x {
-        x.to_rfc3339().serialize(s)
-    } else {
-        s.serialize_none()
-    }
+    x.to_rfc3339().serialize(s)
 }
 
 /// Response of the /metrics.json API

libs/compute_api/src/spec.rs

@@ -3,10 +3,8 @@
 //! The spec.json file is used to pass information to 'compute_ctl'. It contains
 //! all the information needed to start up the right version of PostgreSQL,
 //! and connect it to the storage nodes.
-use serde::{Deserialize, Serialize};
-use serde_with::{serde_as, DisplayFromStr};
+use serde::Deserialize;
 use std::collections::HashMap;
-use utils::lsn::Lsn;
 
 /// String type alias representing Postgres identifier and
 /// intended to be used for DB / role names.
@@ -14,7 +12,6 @@ pub type PgIdent = String;
 
 /// Cluster spec or configuration represented as an optional number of
 /// delta operations + final cluster state description.
-#[serde_as]
 #[derive(Clone, Debug, Default, Deserialize)]
 pub struct ComputeSpec {
     pub format_version: f32,
@@ -27,29 +24,11 @@ pub struct ComputeSpec {
     pub cluster: Cluster,
     pub delta_operations: Option<Vec<DeltaOp>>,
 
-    #[serde(default)]
-    pub mode: ComputeMode,
-
     pub storage_auth_token: Option<String>,
 
     pub startup_tracing_context: Option<HashMap<String, String>>,
 }
 
-#[serde_as]
-#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, Deserialize, Serialize)]
-pub enum ComputeMode {
-    /// A read-write node
-    #[default]
-    Primary,
-    /// A read-only node, pinned at a particular LSN
-    Static(#[serde_as(as = "DisplayFromStr")] Lsn),
-    /// A read-only node that follows the tip of the branch in hot standby mode
-    ///
-    /// Future versions may want to distinguish between replicas with hot standby
-    /// feedback and other kinds of replication configurations.
-    Replica,
-}
-
 #[derive(Clone, Debug, Default, Deserialize)]
 pub struct Cluster {
     pub cluster_id: String,

libs/postgres_backend/src/lib.rs

@@ -50,14 +50,11 @@ impl QueryError {
     }
 }
 
-/// Returns true if the given error is a normal consequence of a network issue,
-/// or the client closing the connection. These errors can happen during normal
-/// operations, and don't indicate a bug in our code.
 pub fn is_expected_io_error(e: &io::Error) -> bool {
     use io::ErrorKind::*;
     matches!(
         e.kind(),
-        BrokenPipe | ConnectionRefused | ConnectionAborted | ConnectionReset | TimedOut
+        ConnectionRefused | ConnectionAborted | ConnectionReset | TimedOut
     )
 }
 

libs/postgres_ffi/wal_craft/src/lib.rs

@@ -1,4 +1,5 @@
-use anyhow::{bail, ensure};
+use anyhow::*;
+use core::time::Duration;
 use log::*;
 use postgres::types::PgLsn;
 use postgres::Client;
@@ -7,7 +8,7 @@ use postgres_ffi::{XLOG_SIZE_OF_XLOG_RECORD, XLOG_SIZE_OF_XLOG_SHORT_PHD};
 use std::cmp::Ordering;
 use std::path::{Path, PathBuf};
 use std::process::Command;
-use std::time::{Duration, Instant};
+use std::time::Instant;
 use tempfile::{tempdir, TempDir};
 
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -54,7 +55,7 @@ impl Conf {
         self.datadir.join("pg_wal")
     }
 
-    fn new_pg_command(&self, command: impl AsRef<Path>) -> anyhow::Result<Command> {
+    fn new_pg_command(&self, command: impl AsRef<Path>) -> Result<Command> {
         let path = self.pg_bin_dir()?.join(command);
         ensure!(path.exists(), "Command {:?} does not exist", path);
         let mut cmd = Command::new(path);
@@ -64,7 +65,7 @@ impl Conf {
         Ok(cmd)
     }
 
-    pub fn initdb(&self) -> anyhow::Result<()> {
+    pub fn initdb(&self) -> Result<()> {
         if let Some(parent) = self.datadir.parent() {
             info!("Pre-creating parent directory {:?}", parent);
             // Tests may be run concurrently and there may be a race to create `test_output/`.
@@ -78,7 +79,7 @@ impl Conf {
         let output = self
             .new_pg_command("initdb")?
             .arg("-D")
-            .arg(&self.datadir)
+            .arg(self.datadir.as_os_str())
             .args(["-U", "postgres", "--no-instructions", "--no-sync"])
             .output()?;
         debug!("initdb output: {:?}", output);
@@ -91,7 +92,7 @@ impl Conf {
         Ok(())
     }
 
-    pub fn start_server(&self) -> anyhow::Result<PostgresServer> {
+    pub fn start_server(&self) -> Result<PostgresServer> {
         info!("Starting Postgres server in {:?}", self.datadir);
         let unix_socket_dir = tempdir()?; // We need a directory with a short name for Unix socket (up to 108 symbols)
         let unix_socket_dir_path = unix_socket_dir.path().to_owned();
@@ -99,9 +100,9 @@ impl Conf {
             .new_pg_command("postgres")?
             .args(["-c", "listen_addresses="])
             .arg("-k")
-            .arg(&unix_socket_dir_path)
+            .arg(unix_socket_dir_path.as_os_str())
             .arg("-D")
-            .arg(&self.datadir)
+            .arg(self.datadir.as_os_str())
             .args(REQUIRED_POSTGRES_CONFIG.iter().flat_map(|cfg| ["-c", cfg]))
             .spawn()?;
         let server = PostgresServer {
@@ -122,7 +123,7 @@ impl Conf {
         &self,
         first_segment_name: &str,
         last_segment_name: &str,
-    ) -> anyhow::Result<std::process::Output> {
+    ) -> Result<std::process::Output> {
         let first_segment_file = self.datadir.join(first_segment_name);
         let last_segment_file = self.datadir.join(last_segment_name);
         info!(
@@ -132,7 +133,10 @@ impl Conf {
         );
         let output = self
             .new_pg_command("pg_waldump")?
-            .args([&first_segment_file, &last_segment_file])
+            .args([
+                &first_segment_file.as_os_str(),
+                &last_segment_file.as_os_str(),
+            ])
             .output()?;
         debug!("waldump output: {:?}", output);
         Ok(output)
@@ -140,9 +144,10 @@ impl Conf {
 }
 
 impl PostgresServer {
-    pub fn connect_with_timeout(&self) -> anyhow::Result<Client> {
+    pub fn connect_with_timeout(&self) -> Result<Client> {
         let retry_until = Instant::now() + *self.client_config.get_connect_timeout().unwrap();
         while Instant::now() < retry_until {
+            use std::result::Result::Ok;
             if let Ok(client) = self.client_config.connect(postgres::NoTls) {
                 return Ok(client);
             }
@@ -159,6 +164,7 @@ impl PostgresServer {
 
 impl Drop for PostgresServer {
     fn drop(&mut self) {
+        use std::result::Result::Ok;
         match self.process.try_wait() {
             Ok(Some(_)) => return,
             Ok(None) => {
@@ -173,12 +179,12 @@ impl Drop for PostgresServer {
 }
 
 pub trait PostgresClientExt: postgres::GenericClient {
-    fn pg_current_wal_insert_lsn(&mut self) -> anyhow::Result<PgLsn> {
+    fn pg_current_wal_insert_lsn(&mut self) -> Result<PgLsn> {
         Ok(self
             .query_one("SELECT pg_current_wal_insert_lsn()", &[])?
             .get(0))
     }
-    fn pg_current_wal_flush_lsn(&mut self) -> anyhow::Result<PgLsn> {
+    fn pg_current_wal_flush_lsn(&mut self) -> Result<PgLsn> {
         Ok(self
             .query_one("SELECT pg_current_wal_flush_lsn()", &[])?
             .get(0))
@@ -187,7 +193,7 @@ pub trait PostgresClientExt: postgres::GenericClient {
 
 impl<C: postgres::GenericClient> PostgresClientExt for C {}
 
-pub fn ensure_server_config(client: &mut impl postgres::GenericClient) -> anyhow::Result<()> {
+pub fn ensure_server_config(client: &mut impl postgres::GenericClient) -> Result<()> {
     client.execute("create extension if not exists neon_test_utils", &[])?;
 
     let wal_keep_size: String = client.query_one("SHOW wal_keep_size", &[])?.get(0);
@@ -221,13 +227,13 @@ pub trait Crafter {
     /// * A vector of some valid "interesting" intermediate LSNs which one may start reading from.
     ///   May include or exclude Lsn(0) and the end-of-wal.
     /// * The expected end-of-wal LSN.
-    fn craft(client: &mut impl postgres::GenericClient) -> anyhow::Result<(Vec<PgLsn>, PgLsn)>;
+    fn craft(client: &mut impl postgres::GenericClient) -> Result<(Vec<PgLsn>, PgLsn)>;
 }
 
 fn craft_internal<C: postgres::GenericClient>(
     client: &mut C,
-    f: impl Fn(&mut C, PgLsn) -> anyhow::Result<(Vec<PgLsn>, Option<PgLsn>)>,
-) -> anyhow::Result<(Vec<PgLsn>, PgLsn)> {
+    f: impl Fn(&mut C, PgLsn) -> Result<(Vec<PgLsn>, Option<PgLsn>)>,
+) -> Result<(Vec<PgLsn>, PgLsn)> {
     ensure_server_config(client)?;
 
     let initial_lsn = client.pg_current_wal_insert_lsn()?;
@@ -259,7 +265,7 @@ fn craft_internal<C: postgres::GenericClient>(
 pub struct Simple;
 impl Crafter for Simple {
     const NAME: &'static str = "simple";
-    fn craft(client: &mut impl postgres::GenericClient) -> anyhow::Result<(Vec<PgLsn>, PgLsn)> {
+    fn craft(client: &mut impl postgres::GenericClient) -> Result<(Vec<PgLsn>, PgLsn)> {
         craft_internal(client, |client, _| {
             client.execute("CREATE table t(x int)", &[])?;
             Ok((Vec::new(), None))
@@ -270,7 +276,7 @@ impl Crafter for Simple {
 pub struct LastWalRecordXlogSwitch;
 impl Crafter for LastWalRecordXlogSwitch {
     const NAME: &'static str = "last_wal_record_xlog_switch";
-    fn craft(client: &mut impl postgres::GenericClient) -> anyhow::Result<(Vec<PgLsn>, PgLsn)> {
+    fn craft(client: &mut impl postgres::GenericClient) -> Result<(Vec<PgLsn>, PgLsn)> {
         // Do not use generate_internal because here we end up with flush_lsn exactly on
         // the segment boundary and insert_lsn after the initial page header, which is unusual.
         ensure_server_config(client)?;
@@ -292,7 +298,7 @@ impl Crafter for LastWalRecordXlogSwitch {
 pub struct LastWalRecordXlogSwitchEndsOnPageBoundary;
 impl Crafter for LastWalRecordXlogSwitchEndsOnPageBoundary {
     const NAME: &'static str = "last_wal_record_xlog_switch_ends_on_page_boundary";
-    fn craft(client: &mut impl postgres::GenericClient) -> anyhow::Result<(Vec<PgLsn>, PgLsn)> {
+    fn craft(client: &mut impl postgres::GenericClient) -> Result<(Vec<PgLsn>, PgLsn)> {
         // Do not use generate_internal because here we end up with flush_lsn exactly on
         // the segment boundary and insert_lsn after the initial page header, which is unusual.
         ensure_server_config(client)?;
@@ -359,7 +365,7 @@ impl Crafter for LastWalRecordXlogSwitchEndsOnPageBoundary {
 fn craft_single_logical_message(
     client: &mut impl postgres::GenericClient,
     transactional: bool,
-) -> anyhow::Result<(Vec<PgLsn>, PgLsn)> {
+) -> Result<(Vec<PgLsn>, PgLsn)> {
     craft_internal(client, |client, initial_lsn| {
         ensure!(
             initial_lsn < PgLsn::from(0x0200_0000 - 1024 * 1024),
@@ -401,7 +407,7 @@ fn craft_single_logical_message(
 pub struct WalRecordCrossingSegmentFollowedBySmallOne;
 impl Crafter for WalRecordCrossingSegmentFollowedBySmallOne {
     const NAME: &'static str = "wal_record_crossing_segment_followed_by_small_one";
-    fn craft(client: &mut impl postgres::GenericClient) -> anyhow::Result<(Vec<PgLsn>, PgLsn)> {
+    fn craft(client: &mut impl postgres::GenericClient) -> Result<(Vec<PgLsn>, PgLsn)> {
         craft_single_logical_message(client, true)
     }
 }
@@ -409,7 +415,7 @@ impl Crafter for WalRecordCrossingSegmentFollowedBySmallOne {
 pub struct LastWalRecordCrossingSegment;
 impl Crafter for LastWalRecordCrossingSegment {
     const NAME: &'static str = "last_wal_record_crossing_segment";
-    fn craft(client: &mut impl postgres::GenericClient) -> anyhow::Result<(Vec<PgLsn>, PgLsn)> {
+    fn craft(client: &mut impl postgres::GenericClient) -> Result<(Vec<PgLsn>, PgLsn)> {
         craft_single_logical_message(client, false)
     }
 }

libs/pq_proto/src/lib.rs

@@ -613,7 +613,7 @@ pub struct XLogDataBody<'a> {
 
 #[derive(Debug)]
 pub struct WalSndKeepAlive {
-    pub wal_end: u64, // current end of WAL on the server
+    pub sent_ptr: u64,
     pub timestamp: i64,
     pub request_reply: bool,
 }
@@ -924,7 +924,7 @@ impl<'a> BeMessage<'a> {
                 buf.put_u8(b'd');
                 write_body(buf, |buf| {
                     buf.put_u8(b'k');
-                    buf.put_u64(req.wal_end);
+                    buf.put_u64(req.sent_ptr);
                     buf.put_i64(req.timestamp);
                     buf.put_u8(u8::from(req.request_reply));
                 });

libs/utils/src/http/endpoint.rs

@@ -131,9 +131,7 @@ impl RequestCancelled {
 
 impl Drop for RequestCancelled {
     fn drop(&mut self) {
-        if std::thread::panicking() {
-            // we are unwinding due to panicking, assume we are not dropped for cancellation
-        } else if let Some(span) = self.warn.take() {
+        if let Some(span) = self.warn.take() {
             // the span has all of the info already, but the outer `.instrument(span)` has already
             // been dropped, so we need to manually re-enter it for this message.
             //

libs/utils/src/lib.rs

@@ -58,8 +58,6 @@ pub mod pageserver_feedback;
 
 pub mod tracing_span_assert;
 
-pub mod rate_limit;
-
 /// use with fail::cfg("$name", "return(2000)")
 #[macro_export]
 macro_rules! failpoint_sleep_millis_async {

libs/utils/src/rate_limit.rs

@@ -1,66 +0,0 @@
-//! A helper to rate limit operations.
-
-use std::time::{Duration, Instant};
-
-pub struct RateLimit {
-    last: Option<Instant>,
-    interval: Duration,
-}
-
-impl RateLimit {
-    pub fn new(interval: Duration) -> Self {
-        Self {
-            last: None,
-            interval,
-        }
-    }
-
-    /// Call `f` if the rate limit allows.
-    /// Don't call it otherwise.
-    pub fn call<F: FnOnce()>(&mut self, f: F) {
-        let now = Instant::now();
-        match self.last {
-            Some(last) if now - last <= self.interval => {
-                // ratelimit
-            }
-            _ => {
-                self.last = Some(now);
-                f();
-            }
-        }
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use std::sync::atomic::AtomicUsize;
-
-    #[test]
-    fn basics() {
-        use super::RateLimit;
-        use std::sync::atomic::Ordering::Relaxed;
-        use std::time::Duration;
-
-        let called = AtomicUsize::new(0);
-        let mut f = RateLimit::new(Duration::from_millis(100));
-
-        let cl = || {
-            called.fetch_add(1, Relaxed);
-        };
-
-        f.call(cl);
-        assert_eq!(called.load(Relaxed), 1);
-        f.call(cl);
-        assert_eq!(called.load(Relaxed), 1);
-        f.call(cl);
-        assert_eq!(called.load(Relaxed), 1);
-        std::thread::sleep(Duration::from_millis(100));
-        f.call(cl);
-        assert_eq!(called.load(Relaxed), 2);
-        f.call(cl);
-        assert_eq!(called.load(Relaxed), 2);
-        std::thread::sleep(Duration::from_millis(100));
-        f.call(cl);
-        assert_eq!(called.load(Relaxed), 3);
-    }
-}

pageserver/src/http/routes.rs

@@ -691,6 +691,16 @@ pub fn html_response(status: StatusCode, data: String) -> Result<Response<Body>,
     Ok(response)
 }
 
+// Helper function to standardize the error messages we produce on bad durations
+//
+// Intended to be used with anyhow's `with_context`, e.g.:
+//
+//   let value = result.with_context(bad_duration("name", &value))?;
+//
+fn bad_duration<'a>(field_name: &'static str, value: &'a str) -> impl 'a + Fn() -> String {
+    move || format!("Cannot parse `{field_name}` duration {value:?}")
+}
+
 async fn tenant_create_handler(mut request: Request<Body>) -> Result<Response<Body>, ApiError> {
     check_permission(&request, None)?;
 
@@ -698,7 +708,91 @@ async fn tenant_create_handler(mut request: Request<Body>) -> Result<Response<Bo
 
     let request_data: TenantCreateRequest = json_request(&mut request).await?;
 
-    let tenant_conf = TenantConfOpt::try_from(&request_data).map_err(ApiError::BadRequest)?;
+    let mut tenant_conf = TenantConfOpt::default();
+    if let Some(gc_period) = request_data.gc_period {
+        tenant_conf.gc_period = Some(
+            humantime::parse_duration(&gc_period)
+                .with_context(bad_duration("gc_period", &gc_period))
+                .map_err(ApiError::BadRequest)?,
+        );
+    }
+    tenant_conf.gc_horizon = request_data.gc_horizon;
+    tenant_conf.image_creation_threshold = request_data.image_creation_threshold;
+
+    if let Some(pitr_interval) = request_data.pitr_interval {
+        tenant_conf.pitr_interval = Some(
+            humantime::parse_duration(&pitr_interval)
+                .with_context(bad_duration("pitr_interval", &pitr_interval))
+                .map_err(ApiError::BadRequest)?,
+        );
+    }
+
+    if let Some(walreceiver_connect_timeout) = request_data.walreceiver_connect_timeout {
+        tenant_conf.walreceiver_connect_timeout = Some(
+            humantime::parse_duration(&walreceiver_connect_timeout)
+                .with_context(bad_duration(
+                    "walreceiver_connect_timeout",
+                    &walreceiver_connect_timeout,
+                ))
+                .map_err(ApiError::BadRequest)?,
+        );
+    }
+    if let Some(lagging_wal_timeout) = request_data.lagging_wal_timeout {
+        tenant_conf.lagging_wal_timeout = Some(
+            humantime::parse_duration(&lagging_wal_timeout)
+                .with_context(bad_duration("lagging_wal_timeout", &lagging_wal_timeout))
+                .map_err(ApiError::BadRequest)?,
+        );
+    }
+    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 {
+        tenant_conf.checkpoint_timeout = Some(
+            humantime::parse_duration(&checkpoint_timeout)
+                .with_context(bad_duration("checkpoint_timeout", &checkpoint_timeout))
+                .map_err(ApiError::BadRequest)?,
+        );
+    }
+
+    tenant_conf.compaction_target_size = request_data.compaction_target_size;
+    tenant_conf.compaction_threshold = request_data.compaction_threshold;
+
+    if let Some(compaction_period) = request_data.compaction_period {
+        tenant_conf.compaction_period = Some(
+            humantime::parse_duration(&compaction_period)
+                .with_context(bad_duration("compaction_period", &compaction_period))
+                .map_err(ApiError::BadRequest)?,
+        );
+    }
+
+    if let Some(eviction_policy) = request_data.eviction_policy {
+        tenant_conf.eviction_policy = Some(
+            serde_json::from_value(eviction_policy)
+                .context("parse field `eviction_policy`")
+                .map_err(ApiError::BadRequest)?,
+        );
+    }
+
+    tenant_conf.min_resident_size_override = request_data.min_resident_size_override;
+
+    if let Some(evictions_low_residence_duration_metric_threshold) =
+        request_data.evictions_low_residence_duration_metric_threshold
+    {
+        tenant_conf.evictions_low_residence_duration_metric_threshold = Some(
+            humantime::parse_duration(&evictions_low_residence_duration_metric_threshold)
+                .with_context(bad_duration(
+                    "evictions_low_residence_duration_metric_threshold",
+                    &evictions_low_residence_duration_metric_threshold,
+                ))
+                .map_err(ApiError::BadRequest)?,
+        );
+    }
 
     let target_tenant_id = request_data
         .new_tenant_id
@@ -766,7 +860,85 @@ async fn update_tenant_config_handler(
     let tenant_id = request_data.tenant_id;
     check_permission(&request, Some(tenant_id))?;
 
-    let tenant_conf = TenantConfOpt::try_from(&request_data).map_err(ApiError::BadRequest)?;
+    let mut tenant_conf = TenantConfOpt::default();
+    if let Some(gc_period) = request_data.gc_period {
+        tenant_conf.gc_period = Some(
+            humantime::parse_duration(&gc_period)
+                .with_context(bad_duration("gc_period", &gc_period))
+                .map_err(ApiError::BadRequest)?,
+        );
+    }
+    tenant_conf.gc_horizon = request_data.gc_horizon;
+    tenant_conf.image_creation_threshold = request_data.image_creation_threshold;
+
+    if let Some(pitr_interval) = request_data.pitr_interval {
+        tenant_conf.pitr_interval = Some(
+            humantime::parse_duration(&pitr_interval)
+                .with_context(bad_duration("pitr_interval", &pitr_interval))
+                .map_err(ApiError::BadRequest)?,
+        );
+    }
+    if let Some(walreceiver_connect_timeout) = request_data.walreceiver_connect_timeout {
+        tenant_conf.walreceiver_connect_timeout = Some(
+            humantime::parse_duration(&walreceiver_connect_timeout)
+                .with_context(bad_duration(
+                    "walreceiver_connect_timeout",
+                    &walreceiver_connect_timeout,
+                ))
+                .map_err(ApiError::BadRequest)?,
+        );
+    }
+    if let Some(lagging_wal_timeout) = request_data.lagging_wal_timeout {
+        tenant_conf.lagging_wal_timeout = Some(
+            humantime::parse_duration(&lagging_wal_timeout)
+                .with_context(bad_duration("lagging_wal_timeout", &lagging_wal_timeout))
+                .map_err(ApiError::BadRequest)?,
+        );
+    }
+    tenant_conf.max_lsn_wal_lag = request_data.max_lsn_wal_lag;
+    tenant_conf.trace_read_requests = request_data.trace_read_requests;
+
+    tenant_conf.checkpoint_distance = request_data.checkpoint_distance;
+    if let Some(checkpoint_timeout) = request_data.checkpoint_timeout {
+        tenant_conf.checkpoint_timeout = Some(
+            humantime::parse_duration(&checkpoint_timeout)
+                .with_context(bad_duration("checkpoint_timeout", &checkpoint_timeout))
+                .map_err(ApiError::BadRequest)?,
+        );
+    }
+    tenant_conf.compaction_target_size = request_data.compaction_target_size;
+    tenant_conf.compaction_threshold = request_data.compaction_threshold;
+
+    if let Some(compaction_period) = request_data.compaction_period {
+        tenant_conf.compaction_period = Some(
+            humantime::parse_duration(&compaction_period)
+                .with_context(bad_duration("compaction_period", &compaction_period))
+                .map_err(ApiError::BadRequest)?,
+        );
+    }
+
+    if let Some(eviction_policy) = request_data.eviction_policy {
+        tenant_conf.eviction_policy = Some(
+            serde_json::from_value(eviction_policy)
+                .context("parse field `eviction_policy`")
+                .map_err(ApiError::BadRequest)?,
+        );
+    }
+
+    tenant_conf.min_resident_size_override = request_data.min_resident_size_override;
+
+    if let Some(evictions_low_residence_duration_metric_threshold) =
+        request_data.evictions_low_residence_duration_metric_threshold
+    {
+        tenant_conf.evictions_low_residence_duration_metric_threshold = Some(
+            humantime::parse_duration(&evictions_low_residence_duration_metric_threshold)
+                .with_context(bad_duration(
+                    "evictions_low_residence_duration_metric_threshold",
+                    &evictions_low_residence_duration_metric_threshold,
+                ))
+                .map_err(ApiError::BadRequest)?,
+        );
+    }
 
     let state = get_state(&request);
     mgr::set_new_tenant_config(state.conf, tenant_conf, tenant_id)

pageserver/src/metrics.rs

@@ -1,9 +1,9 @@
 use metrics::core::{AtomicU64, GenericCounter};
 use metrics::{
     register_counter_vec, register_histogram, register_histogram_vec, register_int_counter,
-    register_int_counter_vec, register_int_gauge, register_int_gauge_vec, register_uint_gauge_vec,
-    Counter, CounterVec, Histogram, HistogramVec, IntCounter, IntCounterVec, IntGauge, IntGaugeVec,
-    UIntGauge, UIntGaugeVec,
+    register_int_counter_vec, register_int_gauge_vec, register_uint_gauge_vec, Counter, CounterVec,
+    Histogram, HistogramVec, IntCounter, IntCounterVec, IntGauge, IntGaugeVec, UIntGauge,
+    UIntGaugeVec,
 };
 use once_cell::sync::Lazy;
 use pageserver_api::models::TenantState;
@@ -287,33 +287,14 @@ impl EvictionsWithLowResidenceDuration {
         let Some(_counter) = self.counter.take() else {
             return;
         };
-
-        let threshold = Self::threshold_label_value(self.threshold);
-
-        let removed = EVICTIONS_WITH_LOW_RESIDENCE_DURATION.remove_label_values(&[
-            tenant_id,
-            timeline_id,
-            self.data_source,
-            &threshold,
-        ]);
-
-        match removed {
-            Err(e) => {
-                // this has been hit in staging as
-                // <https://neondatabase.sentry.io/issues/4142396994/>, but we don't know how.
-                // because we can be in the drop path already, don't risk:
-                // - "double-panic => illegal instruction" or
-                // - future "drop panick => abort"
-                //
-                // so just nag: (the error has the labels)
-                tracing::warn!("failed to remove EvictionsWithLowResidenceDuration, it was already removed? {e:#?}");
-            }
-            Ok(()) => {
-                // to help identify cases where we double-remove the same values, let's log all
-                // deletions?
-                tracing::info!("removed EvictionsWithLowResidenceDuration with {tenant_id}, {timeline_id}, {}, {threshold}", self.data_source);
-            }
-        }
+        EVICTIONS_WITH_LOW_RESIDENCE_DURATION
+            .remove_label_values(&[
+                tenant_id,
+                timeline_id,
+                self.data_source,
+                &Self::threshold_label_value(self.threshold),
+            ])
+            .expect("we own the metric, no-one else should remove it");
     }
 }
 
@@ -478,56 +459,6 @@ pub static TENANT_TASK_EVENTS: Lazy<IntCounterVec> = Lazy::new(|| {
     .expect("Failed to register tenant_task_events metric")
 });
 
-// walreceiver metrics
-
-pub static WALRECEIVER_STARTED_CONNECTIONS: Lazy<IntCounter> = Lazy::new(|| {
-    register_int_counter!(
-        "pageserver_walreceiver_started_connections_total",
-        "Number of started walreceiver connections"
-    )
-    .expect("failed to define a metric")
-});
-
-pub static WALRECEIVER_ACTIVE_MANAGERS: Lazy<IntGauge> = Lazy::new(|| {
-    register_int_gauge!(
-        "pageserver_walreceiver_active_managers",
-        "Number of active walreceiver managers"
-    )
-    .expect("failed to define a metric")
-});
-
-pub static WALRECEIVER_SWITCHES: Lazy<IntCounterVec> = Lazy::new(|| {
-    register_int_counter_vec!(
-        "pageserver_walreceiver_switches_total",
-        "Number of walreceiver manager change_connection calls",
-        &["reason"]
-    )
-    .expect("failed to define a metric")
-});
-
-pub static WALRECEIVER_BROKER_UPDATES: Lazy<IntCounter> = Lazy::new(|| {
-    register_int_counter!(
-        "pageserver_walreceiver_broker_updates_total",
-        "Number of received broker updates in walreceiver"
-    )
-    .expect("failed to define a metric")
-});
-
-pub static WALRECEIVER_CANDIDATES_EVENTS: Lazy<IntCounterVec> = Lazy::new(|| {
-    register_int_counter_vec!(
-        "pageserver_walreceiver_candidates_events_total",
-        "Number of walreceiver candidate events",
-        &["event"]
-    )
-    .expect("failed to define a metric")
-});
-
-pub static WALRECEIVER_CANDIDATES_ADDED: Lazy<IntCounter> =
-    Lazy::new(|| WALRECEIVER_CANDIDATES_EVENTS.with_label_values(&["add"]));
-
-pub static WALRECEIVER_CANDIDATES_REMOVED: Lazy<IntCounter> =
-    Lazy::new(|| WALRECEIVER_CANDIDATES_EVENTS.with_label_values(&["remove"]));
-
 // Metrics collected on WAL redo operations
 //
 // We collect the time spent in actual WAL redo ('redo'), and time waiting

pageserver/src/page_service.rs

@@ -352,7 +352,7 @@ impl PageServerHandler {
         tenant_id: TenantId,
         timeline_id: TimelineId,
         ctx: RequestContext,
-    ) -> Result<(), QueryError>
+    ) -> anyhow::Result<()>
     where
         IO: AsyncRead + AsyncWrite + Send + Sync + Unpin,
     {
@@ -398,9 +398,7 @@ impl PageServerHandler {
                 Some(FeMessage::CopyData(bytes)) => bytes,
                 Some(FeMessage::Terminate) => break,
                 Some(m) => {
-                    return Err(QueryError::Other(anyhow::anyhow!(
-                        "unexpected message: {m:?} during COPY"
-                    )));
+                    anyhow::bail!("unexpected message: {m:?} during COPY");
                 }
                 None => break, // client disconnected
             };

pageserver/src/tenant.rs

@@ -271,10 +271,7 @@ impl UninitializedTimeline<'_> {
             .await
             .context("Failed to flush after basebackup import")?;
 
-        // Initialize without loading the layer map. We started with an empty layer map, and already
-        // updated it for the layers that we created during the import.
-        let mut timelines = self.owning_tenant.timelines.lock().unwrap();
-        self.initialize_with_lock(ctx, &mut timelines, false, true)
+        self.initialize(ctx)
     }
 
     fn raw_timeline(&self) -> anyhow::Result<&Arc<Timeline>> {
@@ -2355,8 +2352,6 @@ impl Tenant {
                 )
             })?;
 
-        // Initialize the timeline without loading the layer map, because we already updated the layer
-        // map above, when we imported the datadir.
         let timeline = {
             let mut timelines = self.timelines.lock().unwrap();
             raw_timeline.initialize_with_lock(ctx, &mut timelines, false, true)?

pageserver/src/tenant/config.rs

@@ -8,8 +8,6 @@
 //! We cannot use global or default config instead, because wrong settings
 //! may lead to a data loss.
 //!
-use anyhow::Context;
-use pageserver_api::models::{TenantConfigRequest, TenantCreateRequest};
 use serde::{Deserialize, Serialize};
 use std::num::NonZeroU64;
 use std::time::Duration;
@@ -282,179 +280,6 @@ impl Default for TenantConf {
     }
 }
 
-// Helper function to standardize the error messages we produce on bad durations
-//
-// Intended to be used with anyhow's `with_context`, e.g.:
-//
-//   let value = result.with_context(bad_duration("name", &value))?;
-//
-fn bad_duration<'a>(field_name: &'static str, value: &'a str) -> impl 'a + Fn() -> String {
-    move || format!("Cannot parse `{field_name}` duration {value:?}")
-}
-
-impl TryFrom<&'_ TenantCreateRequest> for TenantConfOpt {
-    type Error = anyhow::Error;
-
-    fn try_from(request_data: &TenantCreateRequest) -> Result<Self, Self::Error> {
-        let mut tenant_conf = TenantConfOpt::default();
-
-        if let Some(gc_period) = &request_data.gc_period {
-            tenant_conf.gc_period = Some(
-                humantime::parse_duration(gc_period)
-                    .with_context(bad_duration("gc_period", gc_period))?,
-            );
-        }
-        tenant_conf.gc_horizon = request_data.gc_horizon;
-        tenant_conf.image_creation_threshold = request_data.image_creation_threshold;
-
-        if let Some(pitr_interval) = &request_data.pitr_interval {
-            tenant_conf.pitr_interval = Some(
-                humantime::parse_duration(pitr_interval)
-                    .with_context(bad_duration("pitr_interval", pitr_interval))?,
-            );
-        }
-
-        if let Some(walreceiver_connect_timeout) = &request_data.walreceiver_connect_timeout {
-            tenant_conf.walreceiver_connect_timeout = Some(
-                humantime::parse_duration(walreceiver_connect_timeout).with_context(
-                    bad_duration("walreceiver_connect_timeout", walreceiver_connect_timeout),
-                )?,
-            );
-        }
-        if let Some(lagging_wal_timeout) = &request_data.lagging_wal_timeout {
-            tenant_conf.lagging_wal_timeout = Some(
-                humantime::parse_duration(lagging_wal_timeout)
-                    .with_context(bad_duration("lagging_wal_timeout", lagging_wal_timeout))?,
-            );
-        }
-        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 {
-            tenant_conf.checkpoint_timeout = Some(
-                humantime::parse_duration(checkpoint_timeout)
-                    .with_context(bad_duration("checkpoint_timeout", checkpoint_timeout))?,
-            );
-        }
-
-        tenant_conf.compaction_target_size = request_data.compaction_target_size;
-        tenant_conf.compaction_threshold = request_data.compaction_threshold;
-
-        if let Some(compaction_period) = &request_data.compaction_period {
-            tenant_conf.compaction_period = Some(
-                humantime::parse_duration(compaction_period)
-                    .with_context(bad_duration("compaction_period", compaction_period))?,
-            );
-        }
-
-        if let Some(eviction_policy) = &request_data.eviction_policy {
-            tenant_conf.eviction_policy = Some(
-                serde::Deserialize::deserialize(eviction_policy)
-                    .context("parse field `eviction_policy`")?,
-            );
-        }
-
-        tenant_conf.min_resident_size_override = request_data.min_resident_size_override;
-
-        if let Some(evictions_low_residence_duration_metric_threshold) =
-            &request_data.evictions_low_residence_duration_metric_threshold
-        {
-            tenant_conf.evictions_low_residence_duration_metric_threshold = Some(
-                humantime::parse_duration(evictions_low_residence_duration_metric_threshold)
-                    .with_context(bad_duration(
-                        "evictions_low_residence_duration_metric_threshold",
-                        evictions_low_residence_duration_metric_threshold,
-                    ))?,
-            );
-        }
-
-        Ok(tenant_conf)
-    }
-}
-
-impl TryFrom<&'_ TenantConfigRequest> for TenantConfOpt {
-    type Error = anyhow::Error;
-
-    fn try_from(request_data: &TenantConfigRequest) -> Result<Self, Self::Error> {
-        let mut tenant_conf = TenantConfOpt::default();
-        if let Some(gc_period) = &request_data.gc_period {
-            tenant_conf.gc_period = Some(
-                humantime::parse_duration(gc_period)
-                    .with_context(bad_duration("gc_period", gc_period))?,
-            );
-        }
-        tenant_conf.gc_horizon = request_data.gc_horizon;
-        tenant_conf.image_creation_threshold = request_data.image_creation_threshold;
-
-        if let Some(pitr_interval) = &request_data.pitr_interval {
-            tenant_conf.pitr_interval = Some(
-                humantime::parse_duration(pitr_interval)
-                    .with_context(bad_duration("pitr_interval", pitr_interval))?,
-            );
-        }
-        if let Some(walreceiver_connect_timeout) = &request_data.walreceiver_connect_timeout {
-            tenant_conf.walreceiver_connect_timeout = Some(
-                humantime::parse_duration(walreceiver_connect_timeout).with_context(
-                    bad_duration("walreceiver_connect_timeout", walreceiver_connect_timeout),
-                )?,
-            );
-        }
-        if let Some(lagging_wal_timeout) = &request_data.lagging_wal_timeout {
-            tenant_conf.lagging_wal_timeout = Some(
-                humantime::parse_duration(lagging_wal_timeout)
-                    .with_context(bad_duration("lagging_wal_timeout", lagging_wal_timeout))?,
-            );
-        }
-        tenant_conf.max_lsn_wal_lag = request_data.max_lsn_wal_lag;
-        tenant_conf.trace_read_requests = request_data.trace_read_requests;
-
-        tenant_conf.checkpoint_distance = request_data.checkpoint_distance;
-        if let Some(checkpoint_timeout) = &request_data.checkpoint_timeout {
-            tenant_conf.checkpoint_timeout = Some(
-                humantime::parse_duration(checkpoint_timeout)
-                    .with_context(bad_duration("checkpoint_timeout", checkpoint_timeout))?,
-            );
-        }
-        tenant_conf.compaction_target_size = request_data.compaction_target_size;
-        tenant_conf.compaction_threshold = request_data.compaction_threshold;
-
-        if let Some(compaction_period) = &request_data.compaction_period {
-            tenant_conf.compaction_period = Some(
-                humantime::parse_duration(compaction_period)
-                    .with_context(bad_duration("compaction_period", compaction_period))?,
-            );
-        }
-
-        if let Some(eviction_policy) = &request_data.eviction_policy {
-            tenant_conf.eviction_policy = Some(
-                serde::Deserialize::deserialize(eviction_policy)
-                    .context("parse field `eviction_policy`")?,
-            );
-        }
-
-        tenant_conf.min_resident_size_override = request_data.min_resident_size_override;
-
-        if let Some(evictions_low_residence_duration_metric_threshold) =
-            &request_data.evictions_low_residence_duration_metric_threshold
-        {
-            tenant_conf.evictions_low_residence_duration_metric_threshold = Some(
-                humantime::parse_duration(evictions_low_residence_duration_metric_threshold)
-                    .with_context(bad_duration(
-                        "evictions_low_residence_duration_metric_threshold",
-                        evictions_low_residence_duration_metric_threshold,
-                    ))?,
-            );
-        }
-
-        Ok(tenant_conf)
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;

pageserver/src/tenant/storage_layer.rs

@@ -13,9 +13,9 @@ use crate::task_mgr::TaskKind;
 use crate::walrecord::NeonWalRecord;
 use anyhow::Result;
 use bytes::Bytes;
+use either::Either;
 use enum_map::EnumMap;
 use enumset::EnumSet;
-use once_cell::sync::Lazy;
 use pageserver_api::models::LayerAccessKind;
 use pageserver_api::models::{
     HistoricLayerInfo, LayerResidenceEvent, LayerResidenceEventReason, LayerResidenceStatus,
@@ -23,10 +23,8 @@ use pageserver_api::models::{
 use std::ops::Range;
 use std::path::PathBuf;
 use std::sync::{Arc, Mutex};
-use std::time::{Duration, SystemTime, UNIX_EPOCH};
-use tracing::warn;
+use std::time::{SystemTime, UNIX_EPOCH};
 use utils::history_buffer::HistoryBufferWithDropCounter;
-use utils::rate_limit::RateLimit;
 
 use utils::{
     id::{TenantId, TimelineId},
@@ -39,8 +37,6 @@ pub use image_layer::{ImageLayer, ImageLayerWriter};
 pub use inmemory_layer::InMemoryLayer;
 pub use remote_layer::RemoteLayer;
 
-use super::layer_map::BatchedUpdates;
-
 pub fn range_overlaps<T>(a: &Range<T>, b: &Range<T>) -> bool
 where
     T: PartialOrd<T>,
@@ -162,82 +158,41 @@ impl LayerAccessStatFullDetails {
 }
 
 impl LayerAccessStats {
-    /// Create an empty stats object.
-    ///
-    /// The caller is responsible for recording a residence event
-    /// using [`record_residence_event`] before calling `latest_activity`.
-    /// If they don't, [`latest_activity`] will return `None`.
-    pub(crate) fn empty_will_record_residence_event_later() -> Self {
-        LayerAccessStats(Mutex::default())
+    pub(crate) fn for_loading_layer(status: LayerResidenceStatus) -> Self {
+        let new = LayerAccessStats(Mutex::new(LayerAccessStatsLocked::default()));
+        new.record_residence_event(status, LayerResidenceEventReason::LayerLoad);
+        new
     }
 
-    /// Create an empty stats object and record a [`LayerLoad`] event with the given residence status.
-    ///
-    /// See [`record_residence_event`] for why you need to do this while holding the layer map lock.
-    pub(crate) fn for_loading_layer<L>(
-        layer_map_lock_held_witness: &BatchedUpdates<'_, L>,
-        status: LayerResidenceStatus,
-    ) -> Self
-    where
-        L: ?Sized + Layer,
-    {
+    pub(crate) fn for_new_layer_file() -> Self {
         let new = LayerAccessStats(Mutex::new(LayerAccessStatsLocked::default()));
         new.record_residence_event(
-            layer_map_lock_held_witness,
-            status,
-            LayerResidenceEventReason::LayerLoad,
+            LayerResidenceStatus::Resident,
+            LayerResidenceEventReason::LayerCreate,
         );
         new
     }
 
     /// Creates a clone of `self` and records `new_status` in the clone.
-    ///
-    /// The `new_status` is not recorded in `self`.
-    ///
-    /// See [`record_residence_event`] for why you need to do this while holding the layer map lock.
-    pub(crate) fn clone_for_residence_change<L>(
+    /// The `new_status` is not recorded in `self`
+    pub(crate) fn clone_for_residence_change(
         &self,
-        layer_map_lock_held_witness: &BatchedUpdates<'_, L>,
         new_status: LayerResidenceStatus,
-    ) -> LayerAccessStats
-    where
-        L: ?Sized + Layer,
-    {
+    ) -> LayerAccessStats {
         let clone = {
             let inner = self.0.lock().unwrap();
             inner.clone()
         };
         let new = LayerAccessStats(Mutex::new(clone));
-        new.record_residence_event(
-            layer_map_lock_held_witness,
-            new_status,
-            LayerResidenceEventReason::ResidenceChange,
-        );
+        new.record_residence_event(new_status, LayerResidenceEventReason::ResidenceChange);
         new
     }
 
-    /// Record a change in layer residency.
-    ///
-    /// Recording the event must happen while holding the layer map lock to
-    /// ensure that latest-activity-threshold-based layer eviction (eviction_task.rs)
-    /// can do an "imitate access" to this layer, before it observes `now-latest_activity() > threshold`.
-    ///
-    /// If we instead recorded the residence event with a timestamp from before grabbing the layer map lock,
-    /// the following race could happen:
-    ///
-    /// - Compact: Write out an L1 layer from several L0 layers. This records residence event LayerCreate with the current timestamp.
-    /// - Eviction: imitate access logical size calculation. This accesses the L0 layers because the L1 layer is not yet in the layer map.
-    /// - Compact: Grab layer map lock, add the new L1 to layer map and remove the L0s, release layer map lock.
-    /// - Eviction: observes the new L1 layer whose only activity timestamp is the LayerCreate event.
-    ///
-    pub(crate) fn record_residence_event<L>(
+    fn record_residence_event(
         &self,
-        _layer_map_lock_held_witness: &BatchedUpdates<'_, L>,
         status: LayerResidenceStatus,
         reason: LayerResidenceEventReason,
-    ) where
-        L: ?Sized + Layer,
-    {
+    ) {
         let mut locked = self.0.lock().unwrap();
         locked.iter_mut().for_each(|inner| {
             inner
@@ -300,37 +255,24 @@ impl LayerAccessStats {
         ret
     }
 
-    /// Get the latest access timestamp, falling back to latest residence event.
-    ///
-    /// This function can only return `None` if there has not yet been a call to the
-    /// [`record_residence_event`] method. That would generally be considered an
-    /// implementation error. This function logs a rate-limited warning in that case.
-    ///
-    /// TODO: use type system to avoid the need for `fallback`.
-    /// The approach in https://github.com/neondatabase/neon/pull/3775
-    /// could be used to enforce that a residence event is recorded
-    /// before a layer is added to the layer map. We could also have
-    /// a layer wrapper type that holds the LayerAccessStats, and ensure
-    /// that that type can only be produced by inserting into the layer map.
-    pub(crate) fn latest_activity(&self) -> Option<SystemTime> {
+    fn most_recent_access_or_residence_event(
+        &self,
+    ) -> Either<LayerAccessStatFullDetails, LayerResidenceEvent> {
         let locked = self.0.lock().unwrap();
         let inner = &locked.for_eviction_policy;
         match inner.last_accesses.recent() {
-            Some(a) => Some(a.when),
+            Some(a) => Either::Left(*a),
             None => match inner.last_residence_changes.recent() {
-                Some(e) => Some(e.timestamp),
-                None => {
-                    static WARN_RATE_LIMIT: Lazy<Mutex<(usize, RateLimit)>> =
-                        Lazy::new(|| Mutex::new((0, RateLimit::new(Duration::from_secs(10)))));
-                    let mut guard = WARN_RATE_LIMIT.lock().unwrap();
-                    guard.0 += 1;
-                    let occurences = guard.0;
-                    guard.1.call(move || {
-                        warn!(parent: None, occurences, "latest_activity not available, this is an implementation bug, using fallback value");
-                    });
-                    None
-                }
-            },
+                Some(e) => Either::Right(e.clone()),
+                None => unreachable!("constructors for LayerAccessStats ensure that there's always a residence change event"),
+            }
+        }
+    }
+
+    pub(crate) fn latest_activity(&self) -> SystemTime {
+        match self.most_recent_access_or_residence_event() {
+            Either::Left(mra) => mra.when,
+            Either::Right(re) => re.timestamp,
         }
     }
 }

pageserver/src/tenant/storage_layer/delta_layer.rs

@@ -57,7 +57,7 @@ use utils::{
 
 use super::{
     DeltaFileName, Layer, LayerAccessStats, LayerAccessStatsReset, LayerFileName, LayerIter,
-    LayerKeyIter, PathOrConf,
+    LayerKeyIter, LayerResidenceStatus, PathOrConf,
 };
 
 ///
@@ -637,7 +637,7 @@ impl DeltaLayer {
             key_range: summary.key_range,
             lsn_range: summary.lsn_range,
             file_size: metadata.len(),
-            access_stats: LayerAccessStats::empty_will_record_residence_event_later(),
+            access_stats: LayerAccessStats::for_loading_layer(LayerResidenceStatus::Resident),
             inner: RwLock::new(DeltaLayerInner {
                 loaded: false,
                 file: None,
@@ -808,7 +808,7 @@ impl DeltaLayerWriterInner {
             key_range: self.key_start..key_end,
             lsn_range: self.lsn_range.clone(),
             file_size: metadata.len(),
-            access_stats: LayerAccessStats::empty_will_record_residence_event_later(),
+            access_stats: LayerAccessStats::for_new_layer_file(),
             inner: RwLock::new(DeltaLayerInner {
                 loaded: false,
                 file: None,

pageserver/src/tenant/storage_layer/image_layer.rs

@@ -53,7 +53,7 @@ use utils::{
 };
 
 use super::filename::{ImageFileName, LayerFileName};
-use super::{Layer, LayerAccessStatsReset, LayerIter, PathOrConf};
+use super::{Layer, LayerAccessStatsReset, LayerIter, LayerResidenceStatus, PathOrConf};
 
 ///
 /// Header stored in the beginning of the file
@@ -438,7 +438,7 @@ impl ImageLayer {
             key_range: summary.key_range,
             lsn: summary.lsn,
             file_size: metadata.len(),
-            access_stats: LayerAccessStats::empty_will_record_residence_event_later(),
+            access_stats: LayerAccessStats::for_loading_layer(LayerResidenceStatus::Resident),
             inner: RwLock::new(ImageLayerInner {
                 file: None,
                 loaded: false,
@@ -598,7 +598,7 @@ impl ImageLayerWriterInner {
             key_range: self.key_range.clone(),
             lsn: self.lsn,
             file_size: metadata.len(),
-            access_stats: LayerAccessStats::empty_will_record_residence_event_later(),
+            access_stats: LayerAccessStats::for_new_layer_file(),
             inner: RwLock::new(ImageLayerInner {
                 loaded: false,
                 file: None,

pageserver/src/tenant/storage_layer/remote_layer.rs

@@ -4,7 +4,6 @@
 use crate::config::PageServerConf;
 use crate::context::RequestContext;
 use crate::repository::Key;
-use crate::tenant::layer_map::BatchedUpdates;
 use crate::tenant::remote_timeline_client::index::LayerFileMetadata;
 use crate::tenant::storage_layer::{Layer, ValueReconstructResult, ValueReconstructState};
 use anyhow::{bail, Result};
@@ -247,15 +246,11 @@ impl RemoteLayer {
     }
 
     /// Create a Layer struct representing this layer, after it has been downloaded.
-    pub fn create_downloaded_layer<L>(
+    pub fn create_downloaded_layer(
         &self,
-        layer_map_lock_held_witness: &BatchedUpdates<'_, L>,
         conf: &'static PageServerConf,
         file_size: u64,
-    ) -> Arc<dyn PersistentLayer>
-    where
-        L: ?Sized + Layer,
-    {
+    ) -> Arc<dyn PersistentLayer> {
         if self.is_delta {
             let fname = DeltaFileName {
                 key_range: self.key_range.clone(),
@@ -267,10 +262,8 @@ impl RemoteLayer {
                 self.tenantid,
                 &fname,
                 file_size,
-                self.access_stats.clone_for_residence_change(
-                    layer_map_lock_held_witness,
-                    LayerResidenceStatus::Resident,
-                ),
+                self.access_stats
+                    .clone_for_residence_change(LayerResidenceStatus::Resident),
             ))
         } else {
             let fname = ImageFileName {
@@ -283,10 +276,8 @@ impl RemoteLayer {
                 self.tenantid,
                 &fname,
                 file_size,
-                self.access_stats.clone_for_residence_change(
-                    layer_map_lock_held_witness,
-                    LayerResidenceStatus::Resident,
-                ),
+                self.access_stats
+                    .clone_for_residence_change(LayerResidenceStatus::Resident),
             ))
         }
     }

pageserver/src/tenant/timeline.rs

@@ -11,8 +11,7 @@ use itertools::Itertools;
 use once_cell::sync::OnceCell;
 use pageserver_api::models::{
     DownloadRemoteLayersTaskInfo, DownloadRemoteLayersTaskSpawnRequest,
-    DownloadRemoteLayersTaskState, LayerMapInfo, LayerResidenceEventReason, LayerResidenceStatus,
-    TimelineState,
+    DownloadRemoteLayersTaskState, LayerMapInfo, LayerResidenceStatus, TimelineState,
 };
 use remote_storage::GenericRemoteStorage;
 use storage_broker::BrokerClientChannel;
@@ -589,25 +588,15 @@ impl Timeline {
 
         let _timer = self.metrics.wait_lsn_time_histo.start_timer();
 
-        match self
-            .last_record_lsn
-            .wait_for_timeout(lsn, self.conf.wait_lsn_timeout)
-            .await
-        {
-            Ok(()) => Ok(()),
-            seqwait_error => {
-                drop(_timer);
-                let walreceiver_status = self.walreceiver.status().await;
-                seqwait_error.with_context(|| format!(
-                    "Timed out while waiting for WAL record at LSN {} to arrive, last_record_lsn {} disk consistent LSN={}, {}",
-                    lsn,
-                    self.get_last_record_lsn(),
-                    self.get_disk_consistent_lsn(),
-                    walreceiver_status.map(|status| status.to_human_readable_string())
-                            .unwrap_or_else(|| "WalReceiver status: Not active".to_string()),
-                ))
-            }
-        }
+        self.last_record_lsn.wait_for_timeout(lsn, self.conf.wait_lsn_timeout).await
+            .with_context(||
+                format!(
+                    "Timed out while waiting for WAL record at LSN {} to arrive, last_record_lsn {} disk consistent LSN={}",
+                    lsn, self.get_last_record_lsn(), self.get_disk_consistent_lsn()
+                )
+            )?;
+
+        Ok(())
     }
 
     /// Check that it is valid to request operations with that lsn.
@@ -1112,7 +1101,7 @@ impl Timeline {
                 &layer_metadata,
                 local_layer
                     .access_stats()
-                    .clone_for_residence_change(batch_updates, LayerResidenceStatus::Evicted),
+                    .clone_for_residence_change(LayerResidenceStatus::Evicted),
             ),
             LayerFileName::Delta(delta_name) => RemoteLayer::new_delta(
                 self.tenant_id,
@@ -1121,7 +1110,7 @@ impl Timeline {
                 &layer_metadata,
                 local_layer
                     .access_stats()
-                    .clone_for_residence_change(batch_updates, LayerResidenceStatus::Evicted),
+                    .clone_for_residence_change(LayerResidenceStatus::Evicted),
             ),
         });
 
@@ -1490,7 +1479,7 @@ impl Timeline {
                     self.tenant_id,
                     &imgfilename,
                     file_size,
-                    LayerAccessStats::for_loading_layer(&updates, LayerResidenceStatus::Resident),
+                    LayerAccessStats::for_loading_layer(LayerResidenceStatus::Resident),
                 );
 
                 trace!("found layer {}", layer.path().display());
@@ -1522,7 +1511,7 @@ impl Timeline {
                     self.tenant_id,
                     &deltafilename,
                     file_size,
-                    LayerAccessStats::for_loading_layer(&updates, LayerResidenceStatus::Resident),
+                    LayerAccessStats::for_loading_layer(LayerResidenceStatus::Resident),
                 );
 
                 trace!("found layer {}", layer.path().display());
@@ -1658,10 +1647,7 @@ impl Timeline {
                         self.timeline_id,
                         imgfilename,
                         &remote_layer_metadata,
-                        LayerAccessStats::for_loading_layer(
-                            &updates,
-                            LayerResidenceStatus::Evicted,
-                        ),
+                        LayerAccessStats::for_loading_layer(LayerResidenceStatus::Evicted),
                     );
                     let remote_layer = Arc::new(remote_layer);
 
@@ -1686,10 +1672,7 @@ impl Timeline {
                         self.timeline_id,
                         deltafilename,
                         &remote_layer_metadata,
-                        LayerAccessStats::for_loading_layer(
-                            &updates,
-                            LayerResidenceStatus::Evicted,
-                        ),
+                        LayerAccessStats::for_loading_layer(LayerResidenceStatus::Evicted),
                     );
                     let remote_layer = Arc::new(remote_layer);
                     updates.insert_historic(remote_layer);
@@ -2736,16 +2719,11 @@ impl Timeline {
         ])?;
 
         // Add it to the layer map
-        let l = Arc::new(new_delta);
-        let mut layers = self.layers.write().unwrap();
-        let mut batch_updates = layers.batch_update();
-        l.access_stats().record_residence_event(
-            &batch_updates,
-            LayerResidenceStatus::Resident,
-            LayerResidenceEventReason::LayerCreate,
-        );
-        batch_updates.insert_historic(l);
-        batch_updates.flush();
+        self.layers
+            .write()
+            .unwrap()
+            .batch_update()
+            .insert_historic(Arc::new(new_delta));
 
         // update the timeline's physical size
         let sz = new_delta_path.metadata()?.len();
@@ -2950,13 +2928,7 @@ impl Timeline {
             self.metrics
                 .resident_physical_size_gauge
                 .add(metadata.len());
-            let l = Arc::new(l);
-            l.access_stats().record_residence_event(
-                &updates,
-                LayerResidenceStatus::Resident,
-                LayerResidenceEventReason::LayerCreate,
-            );
-            updates.insert_historic(l);
+            updates.insert_historic(Arc::new(l));
         }
         updates.flush();
         drop(layers);
@@ -3389,11 +3361,6 @@ impl Timeline {
 
             new_layer_paths.insert(new_delta_path, LayerFileMetadata::new(metadata.len()));
             let x: Arc<dyn PersistentLayer + 'static> = Arc::new(l);
-            x.access_stats().record_residence_event(
-                &updates,
-                LayerResidenceStatus::Resident,
-                LayerResidenceEventReason::LayerCreate,
-            );
             updates.insert_historic(x);
         }
 
@@ -3904,9 +3871,9 @@ impl Timeline {
 
                     // Download complete. Replace the RemoteLayer with the corresponding
                     // Delta- or ImageLayer in the layer map.
+                    let new_layer = remote_layer.create_downloaded_layer(self_clone.conf, *size);
                     let mut layers = self_clone.layers.write().unwrap();
                     let mut updates = layers.batch_update();
-                    let new_layer = remote_layer.create_downloaded_layer(&updates, self_clone.conf, *size);
                     {
                         use crate::tenant::layer_map::Replacement;
                         let l: Arc<dyn PersistentLayer> = remote_layer.clone();
@@ -4178,15 +4145,7 @@ impl Timeline {
                 continue;
             }
 
-            let last_activity_ts = l
-                .access_stats()
-                .latest_activity()
-                .unwrap_or_else(|| {
-                    // We only use this fallback if there's an implementation error.
-                    // `latest_activity` already does rate-limited warn!() log.
-                    debug!(layer=%l.filename().file_name(), "last_activity returns None, using SystemTime::now");
-                    SystemTime::now()
-                });
+            let last_activity_ts = l.access_stats().latest_activity();
 
             resident_layers.push(LocalLayerInfoForDiskUsageEviction {
                 layer: l,

pageserver/src/tenant/timeline/eviction_task.rs

@@ -184,14 +184,7 @@ impl Timeline {
                 if hist_layer.is_remote_layer() {
                     continue;
                 }
-
-                let last_activity_ts = hist_layer.access_stats().latest_activity().unwrap_or_else(|| {
-                    // We only use this fallback if there's an implementation error.
-                    // `latest_activity` already does rate-limited warn!() log.
-                    debug!(layer=%hist_layer.filename().file_name(), "last_activity returns None, using SystemTime::now");
-                    SystemTime::now()
-                });
-
+                let last_activity_ts = hist_layer.access_stats().latest_activity();
                 let no_activity_for = match now.duration_since(last_activity_ts) {
                     Ok(d) => d,
                     Err(_e) => {

pageserver/src/tenant/timeline/walreceiver.rs

@@ -38,14 +38,12 @@ use std::sync::{Arc, Weak};
 use std::time::Duration;
 use storage_broker::BrokerClientChannel;
 use tokio::select;
-use tokio::sync::{watch, RwLock};
+use tokio::sync::watch;
 use tokio_util::sync::CancellationToken;
 use tracing::*;
 
 use utils::id::TenantTimelineId;
 
-use self::connection_manager::ConnectionManagerStatus;
-
 use super::Timeline;
 
 #[derive(Clone)]
@@ -65,7 +63,6 @@ pub struct WalReceiver {
     timeline_ref: Weak<Timeline>,
     conf: WalReceiverConf,
     started: AtomicBool,
-    manager_status: Arc<RwLock<Option<ConnectionManagerStatus>>>,
 }
 
 impl WalReceiver {
@@ -79,7 +76,6 @@ impl WalReceiver {
             timeline_ref,
             conf,
             started: AtomicBool::new(false),
-            manager_status: Arc::new(RwLock::new(None)),
         }
     }
 
@@ -100,8 +96,8 @@ impl WalReceiver {
         let timeline_id = timeline.timeline_id;
         let walreceiver_ctx =
             ctx.detached_child(TaskKind::WalReceiverManager, DownloadBehavior::Error);
+
         let wal_receiver_conf = self.conf.clone();
-        let loop_status = Arc::clone(&self.manager_status);
         task_mgr::spawn(
             WALRECEIVER_RUNTIME.handle(),
             TaskKind::WalReceiverManager,
@@ -119,28 +115,24 @@ impl WalReceiver {
                     select! {
                         _ = task_mgr::shutdown_watcher() => {
                             info!("WAL receiver shutdown requested, shutting down");
-                            break;
+                            connection_manager_state.shutdown().await;
+                            return Ok(());
                         },
                         loop_step_result = connection_manager_loop_step(
                             &mut broker_client,
                             &mut connection_manager_state,
                             &walreceiver_ctx,
-                            &loop_status,
                         ) => match loop_step_result {
                             ControlFlow::Continue(()) => continue,
                             ControlFlow::Break(()) => {
                                 info!("Connection manager loop ended, shutting down");
-                                break;
+                                connection_manager_state.shutdown().await;
+                                return Ok(());
                             }
                         },
                     }
                 }
-
-                connection_manager_state.shutdown().await;
-                *loop_status.write().await = None;
-                Ok(())
-            }
-            .instrument(info_span!(parent: None, "wal_connection_manager", tenant = %tenant_id, timeline = %timeline_id))
+            }.instrument(info_span!(parent: None, "wal_connection_manager", tenant = %tenant_id, timeline = %timeline_id))
         );
 
         self.started.store(true, atomic::Ordering::Release);
@@ -157,10 +149,6 @@ impl WalReceiver {
         .await;
         self.started.store(false, atomic::Ordering::Release);
     }
-
-    pub(super) async fn status(&self) -> Option<ConnectionManagerStatus> {
-        self.manager_status.read().await.clone()
-    }
 }
 
 /// A handle of an asynchronous task.

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

@@ -13,10 +13,6 @@ use std::{collections::HashMap, num::NonZeroU64, ops::ControlFlow, sync::Arc, ti
 
 use super::{TaskStateUpdate, WalReceiverConf};
 use crate::context::{DownloadBehavior, RequestContext};
-use crate::metrics::{
-    WALRECEIVER_ACTIVE_MANAGERS, WALRECEIVER_BROKER_UPDATES, WALRECEIVER_CANDIDATES_ADDED,
-    WALRECEIVER_CANDIDATES_REMOVED, WALRECEIVER_SWITCHES,
-};
 use crate::task_mgr::TaskKind;
 use crate::tenant::Timeline;
 use anyhow::Context;
@@ -28,7 +24,6 @@ use storage_broker::proto::SubscribeSafekeeperInfoRequest;
 use storage_broker::proto::TenantTimelineId as ProtoTenantTimelineId;
 use storage_broker::BrokerClientChannel;
 use storage_broker::Streaming;
-use tokio::sync::RwLock;
 use tokio::{select, sync::watch};
 use tracing::*;
 
@@ -48,7 +43,6 @@ pub(super) async fn connection_manager_loop_step(
     broker_client: &mut BrokerClientChannel,
     connection_manager_state: &mut ConnectionManagerState,
     ctx: &RequestContext,
-    manager_status: &RwLock<Option<ConnectionManagerStatus>>,
 ) -> ControlFlow<(), ()> {
     let mut timeline_state_updates = connection_manager_state
         .timeline
@@ -62,11 +56,6 @@ pub(super) async fn connection_manager_loop_step(
         }
     }
 
-    WALRECEIVER_ACTIVE_MANAGERS.inc();
-    scopeguard::defer! {
-        WALRECEIVER_ACTIVE_MANAGERS.dec();
-    }
-
     let id = TenantTimelineId {
         tenant_id: connection_manager_state.timeline.tenant_id,
         timeline_id: connection_manager_state.timeline.timeline_id,
@@ -191,7 +180,6 @@ pub(super) async fn connection_manager_loop_step(
                 .change_connection(new_candidate, ctx)
                 .await
         }
-        *manager_status.write().await = Some(connection_manager_state.manager_status());
     }
 }
 
@@ -279,78 +267,6 @@ pub(super) struct ConnectionManagerState {
     wal_stream_candidates: HashMap<NodeId, BrokerSkTimeline>,
 }
 
-/// An information about connection manager's current connection and connection candidates.
-#[derive(Debug, Clone)]
-pub struct ConnectionManagerStatus {
-    existing_connection: Option<WalConnectionStatus>,
-    wal_stream_candidates: HashMap<NodeId, BrokerSkTimeline>,
-}
-
-impl ConnectionManagerStatus {
-    /// Generates a string, describing current connection status in a form, suitable for logging.
-    pub fn to_human_readable_string(&self) -> String {
-        let mut resulting_string = "WalReceiver status".to_string();
-        match &self.existing_connection {
-            Some(connection) => {
-                if connection.has_processed_wal {
-                    resulting_string.push_str(&format!(
-                        " (update {}): streaming WAL from node {}, ",
-                        connection.latest_wal_update.format("%Y-%m-%d %H:%M:%S"),
-                        connection.node,
-                    ));
-
-                    match (connection.streaming_lsn, connection.commit_lsn) {
-                        (None, None) => resulting_string.push_str("no streaming data"),
-                        (None, Some(commit_lsn)) => {
-                            resulting_string.push_str(&format!("commit Lsn: {commit_lsn}"))
-                        }
-                        (Some(streaming_lsn), None) => {
-                            resulting_string.push_str(&format!("streaming Lsn: {streaming_lsn}"))
-                        }
-                        (Some(streaming_lsn), Some(commit_lsn)) => resulting_string.push_str(
-                            &format!("commit|streaming Lsn: {commit_lsn}|{streaming_lsn}"),
-                        ),
-                    }
-                } else if connection.is_connected {
-                    resulting_string.push_str(&format!(
-                        " (update {}): connecting to node {}",
-                        connection
-                            .latest_connection_update
-                            .format("%Y-%m-%d %H:%M:%S"),
-                        connection.node,
-                    ));
-                } else {
-                    resulting_string.push_str(&format!(
-                        " (update {}): initializing node {} connection",
-                        connection
-                            .latest_connection_update
-                            .format("%Y-%m-%d %H:%M:%S"),
-                        connection.node,
-                    ));
-                }
-            }
-            None => resulting_string.push_str(": disconnected"),
-        }
-
-        resulting_string.push_str(", safekeeper candidates (id|update_time|commit_lsn): [");
-        let mut candidates = self.wal_stream_candidates.iter().peekable();
-        while let Some((node_id, candidate_info)) = candidates.next() {
-            resulting_string.push_str(&format!(
-                "({}|{}|{})",
-                node_id,
-                candidate_info.latest_update.format("%H:%M:%S"),
-                Lsn(candidate_info.timeline.commit_lsn)
-            ));
-            if candidates.peek().is_some() {
-                resulting_string.push_str(", ");
-            }
-        }
-        resulting_string.push(']');
-
-        resulting_string
-    }
-}
-
 /// Current connection data.
 #[derive(Debug)]
 struct WalConnection {
@@ -377,14 +293,14 @@ struct NewCommittedWAL {
     discovered_at: NaiveDateTime,
 }
 
-#[derive(Debug, Clone, Copy)]
+#[derive(Debug)]
 struct RetryInfo {
     next_retry_at: Option<NaiveDateTime>,
     retry_duration_seconds: f64,
 }
 
 /// Data about the timeline to connect to, received from the broker.
-#[derive(Debug, Clone)]
+#[derive(Debug)]
 struct BrokerSkTimeline {
     timeline: SafekeeperTimelineInfo,
     /// Time at which the data was fetched from the broker last time, to track the stale data.
@@ -409,14 +325,9 @@ impl ConnectionManagerState {
 
     /// Shuts down the current connection (if any) and immediately starts another one with the given connection string.
     async fn change_connection(&mut self, new_sk: NewWalConnectionCandidate, ctx: &RequestContext) {
-        WALRECEIVER_SWITCHES
-            .with_label_values(&[new_sk.reason.name()])
-            .inc();
-
         self.drop_old_connection(true).await;
 
         let id = self.id;
-        let node_id = new_sk.safekeeper_id;
         let connect_timeout = self.conf.wal_connect_timeout;
         let timeline = Arc::clone(&self.timeline);
         let ctx = ctx.detached_child(
@@ -432,13 +343,12 @@ impl ConnectionManagerState {
                     cancellation,
                     connect_timeout,
                     ctx,
-                    node_id,
                 )
                 .await
                 .context("walreceiver connection handling failure")
             }
             .instrument(
-                info_span!("walreceiver_connection", tenant_id = %id.tenant_id, timeline_id = %id.timeline_id, %node_id),
+                info_span!("walreceiver_connection", tenant_id = %id.tenant_id, timeline_id = %id.timeline_id, node_id = %new_sk.safekeeper_id),
             )
         });
 
@@ -454,7 +364,6 @@ impl ConnectionManagerState {
                 latest_wal_update: now,
                 streaming_lsn: None,
                 commit_lsn: None,
-                node: node_id,
             },
             connection_task: connection_handle,
             discovered_new_wal: None,
@@ -528,8 +437,6 @@ impl ConnectionManagerState {
 
     /// Adds another broker timeline into the state, if its more recent than the one already added there for the same key.
     fn register_timeline_update(&mut self, timeline_update: SafekeeperTimelineInfo) {
-        WALRECEIVER_BROKER_UPDATES.inc();
-
         let new_safekeeper_id = NodeId(timeline_update.safekeeper_id);
         let old_entry = self.wal_stream_candidates.insert(
             new_safekeeper_id,
@@ -541,7 +448,6 @@ impl ConnectionManagerState {
 
         if old_entry.is_none() {
             info!("New SK node was added: {new_safekeeper_id}");
-            WALRECEIVER_CANDIDATES_ADDED.inc();
         }
     }
 
@@ -810,7 +716,6 @@ impl ConnectionManagerState {
             for node_id in node_ids_to_remove {
                 info!("Safekeeper node {node_id} did not send events for over {lagging_wal_timeout:?}, not retrying the connections");
                 self.wal_connection_retries.remove(&node_id);
-                WALRECEIVER_CANDIDATES_REMOVED.inc();
             }
         }
     }
@@ -820,13 +725,6 @@ impl ConnectionManagerState {
             wal_connection.connection_task.shutdown().await;
         }
     }
-
-    fn manager_status(&self) -> ConnectionManagerStatus {
-        ConnectionManagerStatus {
-            existing_connection: self.wal_connection.as_ref().map(|conn| conn.status),
-            wal_stream_candidates: self.wal_stream_candidates.clone(),
-        }
-    }
 }
 
 #[derive(Debug)]
@@ -834,6 +732,8 @@ struct NewWalConnectionCandidate {
     safekeeper_id: NodeId,
     wal_source_connconf: PgConnectionConfig,
     availability_zone: Option<String>,
+    // This field is used in `derive(Debug)` only.
+    #[allow(dead_code)]
     reason: ReconnectReason,
 }
 
@@ -862,18 +762,6 @@ enum ReconnectReason {
     },
 }
 
-impl ReconnectReason {
-    fn name(&self) -> &str {
-        match self {
-            ReconnectReason::NoExistingConnection => "NoExistingConnection",
-            ReconnectReason::LaggingWal { .. } => "LaggingWal",
-            ReconnectReason::SwitchAvailabilityZone => "SwitchAvailabilityZone",
-            ReconnectReason::NoWalTimeout { .. } => "NoWalTimeout",
-            ReconnectReason::NoKeepAlives { .. } => "NoKeepAlives",
-        }
-    }
-}
-
 fn wal_stream_connection_config(
     TenantTimelineId {
         tenant_id,
@@ -979,7 +867,6 @@ mod tests {
             latest_wal_update: now,
             commit_lsn: Some(Lsn(current_lsn)),
             streaming_lsn: Some(Lsn(current_lsn)),
-            node: NodeId(1),
         };
 
         state.conf.max_lsn_wal_lag = NonZeroU64::new(100).unwrap();
@@ -1148,7 +1035,6 @@ mod tests {
             latest_wal_update: now,
             commit_lsn: Some(current_lsn),
             streaming_lsn: Some(current_lsn),
-            node: connected_sk_id,
         };
 
         state.wal_connection = Some(WalConnection {
@@ -1215,7 +1101,6 @@ mod tests {
             latest_wal_update: time_over_threshold,
             commit_lsn: Some(current_lsn),
             streaming_lsn: Some(current_lsn),
-            node: NodeId(1),
         };
 
         state.wal_connection = Some(WalConnection {
@@ -1279,7 +1164,6 @@ mod tests {
             latest_wal_update: time_over_threshold,
             commit_lsn: Some(current_lsn),
             streaming_lsn: Some(current_lsn),
-            node: NodeId(1),
         };
 
         state.wal_connection = Some(WalConnection {
@@ -1377,7 +1261,6 @@ mod tests {
             latest_wal_update: now,
             commit_lsn: Some(current_lsn),
             streaming_lsn: Some(current_lsn),
-            node: connected_sk_id,
         };
 
         state.wal_connection = Some(WalConnection {

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

@@ -24,8 +24,8 @@ use tokio_util::sync::CancellationToken;
 use tracing::{debug, error, info, trace, warn};
 
 use super::TaskStateUpdate;
+use crate::context::RequestContext;
 use crate::metrics::LIVE_CONNECTIONS_COUNT;
-use crate::{context::RequestContext, metrics::WALRECEIVER_STARTED_CONNECTIONS};
 use crate::{
     task_mgr,
     task_mgr::TaskKind,
@@ -37,8 +37,8 @@ use crate::{
 use postgres_backend::is_expected_io_error;
 use postgres_connection::PgConnectionConfig;
 use postgres_ffi::waldecoder::WalStreamDecoder;
+use utils::lsn::Lsn;
 use utils::pageserver_feedback::PageserverFeedback;
-use utils::{id::NodeId, lsn::Lsn};
 
 /// Status of the connection.
 #[derive(Debug, Clone, Copy)]
@@ -56,8 +56,6 @@ pub(super) struct WalConnectionStatus {
     pub streaming_lsn: Option<Lsn>,
     /// Latest commit_lsn received from the safekeeper. Can be zero if no message has been received yet.
     pub commit_lsn: Option<Lsn>,
-    /// The node it is connected to
-    pub node: NodeId,
 }
 
 /// Open a connection to the given safekeeper and receive WAL, sending back progress
@@ -69,10 +67,7 @@ pub(super) async fn handle_walreceiver_connection(
     cancellation: CancellationToken,
     connect_timeout: Duration,
     ctx: RequestContext,
-    node: NodeId,
 ) -> anyhow::Result<()> {
-    WALRECEIVER_STARTED_CONNECTIONS.inc();
-
     // Connect to the database in replication mode.
     info!("connecting to {wal_source_connconf:?}");
 
@@ -105,7 +100,6 @@ pub(super) async fn handle_walreceiver_connection(
         latest_wal_update: Utc::now().naive_utc(),
         streaming_lsn: None,
         commit_lsn: None,
-        node,
     };
     if let Err(e) = events_sender.send(TaskStateUpdate::Progress(connection_status)) {
         warn!("Wal connection event listener dropped right after connection init, aborting the connection: {e}");
@@ -128,7 +122,7 @@ pub(super) async fn handle_walreceiver_connection(
         false,
         async move {
             select! {
-                connection_result = connection => match connection_result {
+                connection_result = connection => match connection_result{
                     Ok(()) => info!("Walreceiver db connection closed"),
                     Err(connection_error) => {
                         if let Err(e) = ignore_expected_errors(connection_error) {

pgxn/neon/file_cache.c

@@ -96,8 +96,6 @@ static shmem_request_hook_type prev_shmem_request_hook;
 #endif
 static int   lfc_shrinking_factor; /* power of two by which local cache size will be shrinked when lfc_free_space_watermark is reached */
 
-void FileCacheMonitorMain(Datum main_arg);
-
 static void
 lfc_shmem_startup(void)
 {
@@ -380,6 +378,7 @@ lfc_evict(RelFileNode rnode, ForkNumber forkNum, BlockNumber blkno)
 {
 	BufferTag tag;
 	FileCacheEntry* entry;
+	ssize_t rc;
 	bool found;
 	int chunk_offs = blkno & (BLOCKS_PER_CHUNK-1);
 	uint32 hash;

poetry.lock

@@ -1,4 +1,4 @@
-# This file is automatically @generated by Poetry and should not be changed by hand.
+# This file is automatically @generated by Poetry 1.4.1 and should not be changed by hand.
 
 [[package]]
 name = "aiohttp"
@@ -968,14 +968,14 @@ testing = ["pre-commit"]
 
 [[package]]
 name = "flask"
-version = "2.2.5"
+version = "2.1.3"
 description = "A simple framework for building complex web applications."
 category = "main"
 optional = false
 python-versions = ">=3.7"
 files = [
-    {file = "Flask-2.2.5-py3-none-any.whl", hash = "sha256:58107ed83443e86067e41eff4631b058178191a355886f8e479e347fa1285fdf"},
-    {file = "Flask-2.2.5.tar.gz", hash = "sha256:edee9b0a7ff26621bd5a8c10ff484ae28737a2410d99b0bb9a6850c7fb977aa0"},
+    {file = "Flask-2.1.3-py3-none-any.whl", hash = "sha256:9013281a7402ad527f8fd56375164f3aa021ecfaff89bfe3825346c24f87e04c"},
+    {file = "Flask-2.1.3.tar.gz", hash = "sha256:15972e5017df0575c3d6c090ba168b6db90259e620ac8d7ea813a396bad5b6cb"},
 ]
 
 [package.dependencies]
@@ -983,7 +983,7 @@ click = ">=8.0"
 importlib-metadata = {version = ">=3.6.0", markers = "python_version < \"3.10\""}
 itsdangerous = ">=2.0"
 Jinja2 = ">=3.0"
-Werkzeug = ">=2.2.2"
+Werkzeug = ">=2.0"
 
 [package.extras]
 async = ["asgiref (>=3.2)"]

proxy/src/bin/snaplet_router.rs

@@ -0,0 +1,180 @@
+use std::{net::SocketAddr, sync::Arc};
+use tokio::{io::AsyncWriteExt, net::TcpListener};
+
+use anyhow::Context;
+use clap::{self, Arg};
+use futures::TryFutureExt;
+use proxy::{
+    auth::{self, AuthFlow},
+    cancellation::CancelMap,
+    compute::ConnCfg,
+    console::messages::MetricsAuxInfo,
+};
+use tokio::io::{AsyncRead, AsyncWrite};
+use tokio_postgres::config::SslMode;
+use tokio_util::sync::CancellationToken;
+use utils::project_git_version;
+
+use tracing::{error, info, warn};
+
+project_git_version!(GIT_VERSION);
+
+fn cli() -> clap::Command {
+    clap::Command::new("Neon proxy/router")
+        .disable_help_flag(true)
+        .version(GIT_VERSION)
+        .arg(
+            Arg::new("listen")
+                .short('l')
+                .long("listen")
+                .help("listen for incoming client connections on ip:port")
+                .default_value("127.0.0.1:4432"),
+        )
+        .arg(
+            Arg::new("dest-host")
+                .long("dest-host")
+                .help("destination hosts")
+                .required(true),
+        )
+        .arg(
+            Arg::new("dest-port")
+                .long("dest-port")
+                .help("destination port")
+                .default_value("5432"),
+        )
+}
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    let _logging_guard = proxy::logging::init().await?;
+    let _panic_hook_guard = utils::logging::replace_panic_hook_with_tracing_panic_hook();
+
+    let args = cli().get_matches();
+
+    let dest_host: String = args.get_one::<String>("dest-host").unwrap().parse()?;
+    let dest_port: u16 = args.get_one::<String>("dest-port").unwrap().parse()?;
+    let listen_address: SocketAddr = args.get_one::<String>("listen").unwrap().parse()?;
+
+    // Start listening for incoming client connections
+    info!("Starting proxy on {listen_address}");
+    let proxy_listener = TcpListener::bind(listen_address).await?;
+
+    let cancellation_token = CancellationToken::new();
+
+    let main = proxy::flatten_err(tokio::spawn(task_main(
+        Arc::new(dest_host),
+        dest_port,
+        proxy_listener,
+        cancellation_token.clone(),
+    )));
+    let signals_task = proxy::flatten_err(tokio::spawn(proxy::handle_signals(cancellation_token)));
+
+    tokio::select! {
+        res = main => { res?; },
+        res = signals_task => { res?; },
+    }
+
+    Ok(())
+}
+
+async fn task_main(
+    dest_host: Arc<String>,
+    dest_port: u16,
+    listener: tokio::net::TcpListener,
+    cancellation_token: CancellationToken,
+) -> anyhow::Result<()> {
+    scopeguard::defer! {
+        info!("proxy has shut down");
+    }
+
+    // When set for the server socket, the keepalive setting
+    // will be inherited by all accepted client sockets.
+    socket2::SockRef::from(&listener).set_keepalive(true)?;
+
+    let mut connections = tokio::task::JoinSet::new();
+    let cancel_map = Arc::new(CancelMap::default());
+
+    loop {
+        tokio::select! {
+            accept_result = listener.accept() => {
+                let (socket, peer_addr) = accept_result?;
+                info!("accepted postgres client connection from {peer_addr}");
+
+                let cancel_map = Arc::clone(&cancel_map);
+                let dest_host = Arc::clone(&dest_host);
+
+                connections.spawn(
+                    async move {
+                        info!("spawned a task for {peer_addr}");
+
+                        socket
+                            .set_nodelay(true)
+                            .context("failed to set socket option")?;
+
+                        handle_client(dest_host, dest_port, &cancel_map, socket).await
+                    }
+                    .unwrap_or_else(|e| {
+                        // Acknowledge that the task has finished with an error.
+                        error!("per-client task finished with an error: {e:#}");
+                    }),
+                );
+            }
+            _ = cancellation_token.cancelled() => {
+                drop(listener);
+                break;
+            }
+        }
+    }
+    // Drain connections
+    while let Some(res) = connections.join_next().await {
+        if let Err(e) = res {
+            if !e.is_panic() && !e.is_cancelled() {
+                warn!("unexpected error from joined connection task: {e:?}");
+            }
+        }
+    }
+    Ok(())
+}
+
+async fn handle_client(
+    dest_host: Arc<String>,
+    dest_port: u16,
+    cancel_map: &CancelMap,
+    stream: impl AsyncRead + AsyncWrite + Unpin,
+) -> anyhow::Result<()> {
+    let do_handshake = proxy::proxy::handshake(stream, None, cancel_map);
+    let (mut stream, params) = match do_handshake.await? {
+        Some(x) => x,
+        None => return Ok(()), // it's a cancellation request
+    };
+
+    let password = AuthFlow::new(&mut stream)
+        .begin(auth::CleartextPassword)
+        .await?
+        .authenticate()
+        .await?;
+
+    let mut conn_cfg = ConnCfg::new();
+    conn_cfg.set_startup_params(&params);
+    conn_cfg.password(password);
+    conn_cfg.host(dest_host.as_str());
+    conn_cfg.port(dest_port);
+    conn_cfg.ssl_mode(SslMode::Require);
+
+    info!("destination: {:?}:{}", dest_host, dest_port);
+
+    let mut conn = conn_cfg
+        .connect(false)
+        .or_else(|e| stream.throw_error(e))
+        .await?;
+
+    cancel_map
+        .with_session(|session| async {
+            proxy::proxy::prepare_client_connection(&conn, false, session, &mut stream).await?;
+            let (stream, read_buf) = stream.into_inner();
+            conn.stream.write_all(&read_buf).await?;
+            let metrics_aux: MetricsAuxInfo = Default::default();
+            proxy::proxy::proxy_pass(stream, conn.stream, &metrics_aux).await
+        })
+        .await
+}

proxy/src/logging.rs

@@ -1,4 +1,3 @@
-use tracing_opentelemetry::OpenTelemetryLayer;
 use tracing_subscriber::{
     filter::{EnvFilter, LevelFilter},
     prelude::*,
@@ -22,13 +21,8 @@ pub async fn init() -> anyhow::Result<LoggingGuard> {
         .with_writer(std::io::stderr)
         .with_target(false);
 
-    let otlp_layer = tracing_utils::init_tracing("proxy")
-        .await
-        .map(OpenTelemetryLayer::new);
-
     tracing_subscriber::registry()
         .with(env_filter)
-        .with(otlp_layer)
         .with(fmt_layer)
         .try_init()?;
 

proxy/src/proxy.rs

@@ -213,7 +213,7 @@ async fn handle_client(
 /// It's easier to work with owned `stream` here as we need to upgrade it to TLS;
 /// we also take an extra care of propagating only the select handshake errors to client.
 #[tracing::instrument(skip_all)]
-async fn handshake<S: AsyncRead + AsyncWrite + Unpin>(
+pub async fn handshake<S: AsyncRead + AsyncWrite + Unpin>(
     stream: S,
     mut tls: Option<&TlsConfig>,
     cancel_map: &CancelMap,
@@ -350,7 +350,7 @@ async fn connect_to_compute(
 
 /// Finish client connection initialization: confirm auth success, send params, etc.
 #[tracing::instrument(skip_all)]
-async fn prepare_client_connection(
+pub async fn prepare_client_connection(
     node: &compute::PostgresConnection,
     reported_auth_ok: bool,
     session: cancellation::Session<'_>,

safekeeper/Cargo.toml

@@ -25,7 +25,6 @@ parking_lot.workspace = true
 postgres.workspace = true
 postgres-protocol.workspace = true
 regex.workspace = true
-scopeguard.workspace = true
 reqwest = { workspace = true, features = ["json"] }
 serde.workspace = true
 serde_json.workspace = true

safekeeper/src/broker.rs

@@ -14,13 +14,10 @@ use storage_broker::proto::SubscribeSafekeeperInfoRequest;
 use storage_broker::Request;
 
 use std::time::Duration;
-use std::time::Instant;
 use tokio::task::JoinHandle;
 use tokio::{runtime, time::sleep};
 use tracing::*;
 
-use crate::metrics::BROKER_PULLED_UPDATES;
-use crate::metrics::BROKER_PUSHED_UPDATES;
 use crate::GlobalTimelines;
 use crate::SafeKeeperConf;
 
@@ -52,17 +49,12 @@ async fn push_loop(conf: SafeKeeperConf) -> anyhow::Result<()> {
             // is under plain mutex. That's ok, all this code is not performance
             // sensitive and there is no risk of deadlock as we don't await while
             // lock is held.
-            let now = Instant::now();
             let mut active_tlis = GlobalTimelines::get_all();
             active_tlis.retain(|tli| tli.is_active());
             for tli in &active_tlis {
                 let sk_info = tli.get_safekeeper_info(&conf);
                 yield sk_info;
-                BROKER_PUSHED_UPDATES.inc();
             }
-            let elapsed = now.elapsed();
-            // Log duration every second. Should be about 10MB of logs per day.
-            info!("pushed {} timeline updates to broker in {:?}", active_tlis.len(), elapsed);
             sleep(push_interval).await;
         }
     };
@@ -87,10 +79,6 @@ async fn pull_loop(conf: SafeKeeperConf) -> Result<()> {
         .context("subscribe_safekeper_info request failed")?
         .into_inner();
 
-    let ok_counter = BROKER_PULLED_UPDATES.with_label_values(&["ok"]);
-    let not_found = BROKER_PULLED_UPDATES.with_label_values(&["not_found"]);
-    let err_counter = BROKER_PULLED_UPDATES.with_label_values(&["error"]);
-
     while let Some(msg) = stream.message().await? {
         let proto_ttid = msg
             .tenant_timeline_id
@@ -103,15 +91,7 @@ async fn pull_loop(conf: SafeKeeperConf) -> Result<()> {
             // connection to the broker.
 
             // note: there are blocking operations below, but it's considered fine for now
-            let res = tli.record_safekeeper_info(msg).await;
-            if res.is_ok() {
-                ok_counter.inc();
-            } else {
-                err_counter.inc();
-            }
-            res?;
-        } else {
-            not_found.inc();
+            tli.record_safekeeper_info(msg).await?
         }
     }
     bail!("end of stream");

safekeeper/src/handler.rs

@@ -10,7 +10,7 @@ use tracing::{info, info_span, Instrument};
 use crate::auth::check_permission;
 use crate::json_ctrl::{handle_json_ctrl, AppendLogicalMessage};
 
-use crate::metrics::{TrafficMetrics, PG_QUERIES_FINISHED, PG_QUERIES_RECEIVED};
+use crate::metrics::TrafficMetrics;
 use crate::wal_service::ConnectionId;
 use crate::{GlobalTimelines, SafeKeeperConf};
 use postgres_backend::QueryError;
@@ -72,15 +72,6 @@ fn parse_cmd(cmd: &str) -> anyhow::Result<SafekeeperPostgresCommand> {
     }
 }
 
-fn cmd_to_string(cmd: &SafekeeperPostgresCommand) -> &str {
-    match cmd {
-        SafekeeperPostgresCommand::StartWalPush => "START_WAL_PUSH",
-        SafekeeperPostgresCommand::StartReplication { .. } => "START_REPLICATION",
-        SafekeeperPostgresCommand::IdentifySystem => "IDENTIFY_SYSTEM",
-        SafekeeperPostgresCommand::JSONCtrl { .. } => "JSON_CTRL",
-    }
-}
-
 #[async_trait::async_trait]
 impl<IO: AsyncRead + AsyncWrite + Unpin + Send> postgres_backend::Handler<IO>
     for SafekeeperPostgresHandler
@@ -177,12 +168,6 @@ impl<IO: AsyncRead + AsyncWrite + Unpin + Send> postgres_backend::Handler<IO>
         }
 
         let cmd = parse_cmd(query_string)?;
-        let cmd_str = cmd_to_string(&cmd);
-
-        PG_QUERIES_RECEIVED.with_label_values(&[cmd_str]).inc();
-        scopeguard::defer! {
-            PG_QUERIES_FINISHED.with_label_values(&[cmd_str]).inc();
-        }
 
         info!(
             "got query {:?} in timeline {:?}",

safekeeper/src/metrics.rs

@@ -10,7 +10,7 @@ use anyhow::Result;
 use metrics::{
     core::{AtomicU64, Collector, Desc, GenericCounter, GenericGaugeVec, Opts},
     proto::MetricFamily,
-    register_int_counter, register_int_counter_vec, Gauge, IntCounter, IntCounterVec, IntGaugeVec,
+    register_int_counter_vec, Gauge, IntCounterVec, IntGaugeVec,
 };
 use once_cell::sync::Lazy;
 
@@ -73,58 +73,6 @@ pub static PG_IO_BYTES: Lazy<IntCounterVec> = Lazy::new(|| {
     )
     .expect("Failed to register safekeeper_pg_io_bytes gauge")
 });
-pub static BROKER_PUSHED_UPDATES: Lazy<IntCounter> = Lazy::new(|| {
-    register_int_counter!(
-        "safekeeper_broker_pushed_updates_total",
-        "Number of timeline updates pushed to the broker"
-    )
-    .expect("Failed to register safekeeper_broker_pushed_updates_total counter")
-});
-pub static BROKER_PULLED_UPDATES: Lazy<IntCounterVec> = Lazy::new(|| {
-    register_int_counter_vec!(
-        "safekeeper_broker_pulled_updates_total",
-        "Number of timeline updates pulled and processed from the broker",
-        &["result"]
-    )
-    .expect("Failed to register safekeeper_broker_pulled_updates_total counter")
-});
-pub static PG_QUERIES_RECEIVED: Lazy<IntCounterVec> = Lazy::new(|| {
-    register_int_counter_vec!(
-        "safekeeper_pg_queries_received_total",
-        "Number of queries received through pg protocol",
-        &["query"]
-    )
-    .expect("Failed to register safekeeper_pg_queries_received_total counter")
-});
-pub static PG_QUERIES_FINISHED: Lazy<IntCounterVec> = Lazy::new(|| {
-    register_int_counter_vec!(
-        "safekeeper_pg_queries_finished_total",
-        "Number of queries finished through pg protocol",
-        &["query"]
-    )
-    .expect("Failed to register safekeeper_pg_queries_finished_total counter")
-});
-pub static REMOVED_WAL_SEGMENTS: Lazy<IntCounter> = Lazy::new(|| {
-    register_int_counter!(
-        "safekeeper_removed_wal_segments_total",
-        "Number of WAL segments removed from the disk"
-    )
-    .expect("Failed to register safekeeper_removed_wal_segments_total counter")
-});
-pub static BACKED_UP_SEGMENTS: Lazy<IntCounter> = Lazy::new(|| {
-    register_int_counter!(
-        "safekeeper_backed_up_segments_total",
-        "Number of WAL segments backed up to the broker"
-    )
-    .expect("Failed to register safekeeper_backed_up_segments_total counter")
-});
-pub static BACKUP_ERRORS: Lazy<IntCounter> = Lazy::new(|| {
-    register_int_counter!(
-        "safekeeper_backup_errors_total",
-        "Number of errors during backup"
-    )
-    .expect("Failed to register safekeeper_backup_errors_total counter")
-});
 
 pub const LABEL_UNKNOWN: &str = "unknown";
 

safekeeper/src/send_wal.rs

@@ -384,8 +384,6 @@ impl SafekeeperPostgresHandler {
             self.appname.clone(),
         ));
 
-        let commit_lsn_watch_rx = tli.get_commit_lsn_watch_rx();
-
         // Walproposer gets special handling: safekeeper must give proposer all
         // local WAL till the end, whether committed or not (walproposer will
         // hang otherwise). That's because walproposer runs the consensus and
@@ -401,14 +399,7 @@ impl SafekeeperPostgresHandler {
         } else {
             None
         };
-
-        // take the latest commit_lsn if don't have stop_pos
-        let mut end_pos = stop_pos.unwrap_or(*commit_lsn_watch_rx.borrow());
-
-        if end_pos < start_pos {
-            warn!("start_pos {} is ahead of end_pos {}", start_pos, end_pos);
-            end_pos = start_pos;
-        }
+        let end_pos = stop_pos.unwrap_or(Lsn::INVALID);
 
         info!(
             "starting streaming from {:?} till {:?}",
@@ -438,7 +429,7 @@ impl SafekeeperPostgresHandler {
             start_pos,
             end_pos,
             stop_pos,
-            commit_lsn_watch_rx,
+            commit_lsn_watch_rx: tli.get_commit_lsn_watch_rx(),
             ws_guard: ws_guard.clone(),
             wal_reader,
             send_buf: [0; MAX_SEND_SIZE],
@@ -465,11 +456,6 @@ struct WalSender<'a, IO> {
     // Position since which we are sending next chunk.
     start_pos: Lsn,
     // WAL up to this position is known to be locally available.
-    // Usually this is the same as the latest commit_lsn, but in case of
-    // walproposer recovery, this is flush_lsn.
-    //
-    // We send this LSN to the receiver as wal_end, so that it knows how much
-    // WAL this safekeeper has. This LSN should be as fresh as possible.
     end_pos: Lsn,
     // If present, terminate after reaching this position; used by walproposer
     // in recovery.
@@ -544,18 +530,10 @@ impl<IO: AsyncRead + AsyncWrite + Unpin> WalSender<'_, IO> {
     /// exit in the meanwhile
     async fn wait_wal(&mut self) -> Result<(), CopyStreamHandlerEnd> {
         loop {
-            self.end_pos = *self.commit_lsn_watch_rx.borrow();
-            if self.end_pos > self.start_pos {
-                // We have something to send.
-                return Ok(());
-            }
-
-            // Wait for WAL to appear, now self.end_pos == self.start_pos.
             if let Some(lsn) = wait_for_lsn(&mut self.commit_lsn_watch_rx, self.start_pos).await? {
                 self.end_pos = lsn;
                 return Ok(());
             }
-
             // Timed out waiting for WAL, check for termination and send KA
             if let Some(remote_consistent_lsn) = self
                 .ws_guard
@@ -573,7 +551,7 @@ impl<IO: AsyncRead + AsyncWrite + Unpin> WalSender<'_, IO> {
 
             self.pgb
                 .write_message(&BeMessage::KeepAlive(WalSndKeepAlive {
-                    wal_end: self.end_pos.0,
+                    sent_ptr: self.end_pos.0,
                     timestamp: get_current_timestamp(),
                     request_reply: true,
                 }))
@@ -640,6 +618,11 @@ const POLL_STATE_TIMEOUT: Duration = Duration::from_secs(1);
 /// - Ok(None) if timeout expired;
 /// - Err in case of error (if watch channel is in trouble, shouldn't happen).
 async fn wait_for_lsn(rx: &mut Receiver<Lsn>, lsn: Lsn) -> anyhow::Result<Option<Lsn>> {
+    let commit_lsn: Lsn = *rx.borrow();
+    if commit_lsn > lsn {
+        return Ok(Some(commit_lsn));
+    }
+
     let res = timeout(POLL_STATE_TIMEOUT, async move {
         let mut commit_lsn;
         loop {

safekeeper/src/wal_backup.rs

@@ -25,7 +25,6 @@ use tracing::*;
 
 use utils::{id::TenantTimelineId, lsn::Lsn};
 
-use crate::metrics::{BACKED_UP_SEGMENTS, BACKUP_ERRORS};
 use crate::timeline::{PeerInfo, Timeline};
 use crate::{GlobalTimelines, SafeKeeperConf};
 
@@ -395,13 +394,7 @@ async fn backup_single_segment(
             )
         })?;
 
-    let res = backup_object(&segment_file_path, &remote_segment_path, seg.size()).await;
-    if res.is_ok() {
-        BACKED_UP_SEGMENTS.inc();
-    } else {
-        BACKUP_ERRORS.inc();
-    }
-    res?;
+    backup_object(&segment_file_path, &remote_segment_path, seg.size()).await?;
     debug!("Backup of {} done", segment_file_path.display());
 
     Ok(())

safekeeper/src/wal_storage.rs

@@ -27,7 +27,7 @@ use tracing::*;
 
 use utils::{id::TenantTimelineId, lsn::Lsn};
 
-use crate::metrics::{time_io_closure, WalStorageMetrics, REMOVED_WAL_SEGMENTS};
+use crate::metrics::{time_io_closure, WalStorageMetrics};
 use crate::safekeeper::SafeKeeperState;
 
 use crate::wal_backup::read_object;
@@ -455,7 +455,6 @@ fn remove_segments_from_disk(
                 n_removed += 1;
                 min_removed = min(min_removed, segno);
                 max_removed = max(max_removed, segno);
-                REMOVED_WAL_SEGMENTS.inc();
             }
         }
     }

scripts/pr-comment-test-report.js

@@ -37,7 +37,7 @@ module.exports = async ({ github, context, fetch, reports }) => {
         const {buildType, reportUrl, jsonUrl} = report
 
         if (!reportUrl || !jsonUrl) {
-            commentBody += `#### ${buildType} build: no tests were run or test report is not available\n`
+            console.warn(`"reportUrl" or "jsonUrl" aren't set for ${buildType} build`)
             continue
         }
 
@@ -78,7 +78,7 @@ module.exports = async ({ github, context, fetch, reports }) => {
         }
 
         const totalTestsCount = failedTests.length + passedTests.length + skippedTests.length
-        commentBody += `#### ${buildType} build: ${totalTestsCount} tests run: ${passedTests.length} passed, ${failedTests.length} failed, ${skippedTests.length} skipped ([full report](${reportUrl}))\n`
+        commentBody += `#### ${buildType} build: ${totalTestsCount} tests run: ${passedTests.length} passed, ${failedTests.length} failed, ${skippedTests.length} ([full report](${reportUrl}))\n`
         if (failedTests.length > 0) {
             commentBody += `Failed tests:\n`
             for (const test of failedTests) {

test_runner/README.md

@@ -71,8 +71,7 @@ a subdirectory for each version with naming convention `v{PG_VERSION}/`.
 Inside that dir, a `bin/postgres` binary should be present.
 `DEFAULT_PG_VERSION`: The version of Postgres to use,
 This is used to construct full path to the postgres binaries.
-Format is 2-digit major version nubmer, i.e. `DEFAULT_PG_VERSION="14"`. Alternatively,
-you can use `--pg-version` argument.
+Format is 2-digit major version nubmer, i.e. `DEFAULT_PG_VERSION="14"`
 `TEST_OUTPUT`: Set the directory where test state and test output files
 should go.
 `TEST_SHARED_FIXTURES`: Try to re-use a single pageserver for all the tests.

test_runner/conftest.py

@@ -1,5 +1,4 @@
 pytest_plugins = (
-    "fixtures.pg_version",
     "fixtures.neon_fixtures",
     "fixtures.benchmark_fixture",
     "fixtures.pg_stats",

test_runner/fixtures/neon_fixtures.py

@@ -45,7 +45,6 @@ from typing_extensions import Literal
 from fixtures.log_helper import log
 from fixtures.pageserver.http import PageserverHttpClient
 from fixtures.pageserver.utils import wait_for_last_record_lsn, wait_for_upload
-from fixtures.pg_version import PgVersion
 from fixtures.types import Lsn, TenantId, TimelineId
 from fixtures.utils import (
     ATTACHMENT_NAME_REGEX,
@@ -76,6 +75,7 @@ Env = Dict[str, str]
 
 DEFAULT_OUTPUT_DIR: str = "test_output"
 DEFAULT_BRANCH_NAME: str = "main"
+DEFAULT_PG_VERSION_DEFAULT: str = "14"
 
 BASE_PORT: int = 15000
 WORKER_PORT_NUM: int = 1000
@@ -148,7 +148,18 @@ def top_output_dir(base_dir: Path) -> Iterator[Path]:
 
 
 @pytest.fixture(scope="session")
-def versioned_pg_distrib_dir(pg_distrib_dir: Path, pg_version: PgVersion) -> Iterator[Path]:
+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"
@@ -281,12 +292,6 @@ def port_distributor(worker_base_port: int) -> PortDistributor:
     return PortDistributor(base_port=worker_base_port, port_number=WORKER_PORT_NUM)
 
 
-@pytest.fixture(scope="session")
-def httpserver_listen_address(port_distributor: PortDistributor):
-    port = port_distributor.get_port()
-    return ("localhost", port)
-
-
 @pytest.fixture(scope="function")
 def default_broker(
     port_distributor: PortDistributor,
@@ -581,7 +586,7 @@ class NeonEnvBuilder:
         mock_s3_server: MockS3Server,
         neon_binpath: Path,
         pg_distrib_dir: Path,
-        pg_version: PgVersion,
+        pg_version: str,
         remote_storage: Optional[RemoteStorage] = None,
         remote_storage_users: RemoteStorageUsers = RemoteStorageUsers.PAGESERVER,
         pageserver_config_override: Optional[str] = None,
@@ -1030,7 +1035,7 @@ def _shared_simple_env(
     top_output_dir: Path,
     neon_binpath: Path,
     pg_distrib_dir: Path,
-    pg_version: PgVersion,
+    pg_version: str,
 ) -> Iterator[NeonEnv]:
     """
     # Internal fixture backing the `neon_simple_env` fixture. If TEST_SHARED_FIXTURES
@@ -1087,7 +1092,7 @@ def neon_env_builder(
     mock_s3_server: MockS3Server,
     neon_binpath: Path,
     pg_distrib_dir: Path,
-    pg_version: PgVersion,
+    pg_version: str,
     default_broker: NeonBroker,
     run_id: uuid.UUID,
 ) -> Iterator[NeonEnvBuilder]:
@@ -1742,7 +1747,7 @@ 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: PgVersion):
+    def __init__(self, log_dir: Path, pg_distrib_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"
@@ -1801,7 +1806,7 @@ class PgBin:
 
 
 @pytest.fixture(scope="function")
-def pg_bin(test_output_dir: Path, pg_distrib_dir: Path, pg_version: PgVersion) -> PgBin:
+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)
 
 
@@ -1889,7 +1894,7 @@ def vanilla_pg(
     test_output_dir: Path,
     port_distributor: PortDistributor,
     pg_distrib_dir: Path,
-    pg_version: PgVersion,
+    pg_version: str,
 ) -> Iterator[VanillaPostgres]:
     pgdatadir = test_output_dir / "pgdata-vanilla"
     pg_bin = PgBin(test_output_dir, pg_distrib_dir, pg_version)
@@ -1934,7 +1939,7 @@ class RemotePostgres(PgProtocol):
 
 @pytest.fixture(scope="function")
 def remote_pg(
-    test_output_dir: Path, pg_distrib_dir: Path, pg_version: PgVersion
+    test_output_dir: Path, pg_distrib_dir: Path, pg_version: str
 ) -> Iterator[RemotePostgres]:
     pg_bin = PgBin(test_output_dir, pg_distrib_dir, pg_version)
 
@@ -2618,7 +2623,7 @@ class Safekeeper:
 @dataclass
 class SafekeeperTimelineStatus:
     acceptor_epoch: int
-    pg_version: int  # Not exactly a PgVersion, safekeeper returns version as int, for example 150002 for 15.2
+    pg_version: int
     flush_lsn: Lsn
     commit_lsn: Lsn
     timeline_start_lsn: Lsn
@@ -2664,11 +2669,7 @@ class SafekeeperHttpClient(requests.Session):
         return res_json
 
     def timeline_create(
-        self,
-        tenant_id: TenantId,
-        timeline_id: TimelineId,
-        pg_version: int,  # Not exactly a PgVersion, safekeeper returns version as int, for example 150002 for 15.2
-        commit_lsn: Lsn,
+        self, tenant_id: TenantId, timeline_id: TimelineId, pg_version: int, commit_lsn: Lsn
     ):
         body = {
             "tenant_id": str(tenant_id),

test_runner/fixtures/pageserver/http.py

@@ -9,7 +9,6 @@ import requests
 
 from fixtures.log_helper import log
 from fixtures.metrics import Metrics, parse_metrics
-from fixtures.pg_version import PgVersion
 from fixtures.types import Lsn, TenantId, TimelineId
 from fixtures.utils import Fn
 
@@ -267,21 +266,19 @@ class PageserverHttpClient(requests.Session):
 
     def timeline_create(
         self,
-        pg_version: PgVersion,
         tenant_id: TenantId,
         new_timeline_id: Optional[TimelineId] = None,
         ancestor_timeline_id: Optional[TimelineId] = None,
         ancestor_start_lsn: Optional[Lsn] = None,
     ) -> Dict[Any, Any]:
-        body: Dict[str, Any] = {
-            "new_timeline_id": str(new_timeline_id) if new_timeline_id else None,
-            "ancestor_start_lsn": str(ancestor_start_lsn) if ancestor_start_lsn else None,
-            "ancestor_timeline_id": str(ancestor_timeline_id) if ancestor_timeline_id else None,
-        }
-        if pg_version != PgVersion.NOT_SET:
-            body["pg_version"] = int(pg_version)
-
-        res = self.post(f"http://localhost:{self.port}/v1/tenant/{tenant_id}/timeline", json=body)
+        res = self.post(
+            f"http://localhost:{self.port}/v1/tenant/{tenant_id}/timeline",
+            json={
+                "new_timeline_id": str(new_timeline_id) if new_timeline_id else None,
+                "ancestor_start_lsn": str(ancestor_start_lsn) if ancestor_start_lsn else None,
+                "ancestor_timeline_id": str(ancestor_timeline_id) if ancestor_timeline_id else None,
+            },
+        )
         self.verbose_error(res)
         if res.status_code == 409:
             raise Exception(f"could not create timeline: already exists for id {new_timeline_id}")

test_runner/fixtures/pg_version.py

@@ -1,68 +0,0 @@
-import enum
-import os
-from typing import Iterator, Optional
-
-import pytest
-from _pytest.config.argparsing import Parser
-from pytest import FixtureRequest
-
-from fixtures.log_helper import log
-
-"""
-This fixture is used to determine which version of Postgres to use for tests.
-"""
-
-
-# Inherit PgVersion from str rather than int to make it easier to pass as a command-line argument
-# TODO: use enum.StrEnum for Python >= 3.11
-@enum.unique
-class PgVersion(str, enum.Enum):
-    V14 = "14"
-    V15 = "15"
-    # Instead of making version an optional parameter in methods, we can use this fake entry
-    # to explicitly rely on the default server version (could be different from pg_version fixture value)
-    NOT_SET = "<-POSTRGRES VERSION IS NOT SET->"
-
-    # Make it less confusing in logs
-    def __repr__(self) -> str:
-        return f"'{self.value}'"
-
-    @classmethod
-    def _missing_(cls, value) -> Optional["PgVersion"]:
-        known_values = {v.value for _, v in cls.__members__.items()}
-
-        # Allow passing version as a string with "v" prefix (e.g. "v14")
-        if isinstance(value, str) and value.lower().startswith("v") and value[1:] in known_values:
-            return cls(value[1:])
-        # Allow passing version as an int (e.g. 15 or 150002, both will be converted to PgVersion.V15)
-        elif isinstance(value, int) and str(value)[:2] in known_values:
-            return cls(str(value)[:2])
-
-        # Make mypy happy
-        # See https://github.com/python/mypy/issues/3974
-        return None
-
-
-DEFAULT_VERSION: PgVersion = PgVersion.V14
-
-
-def pytest_addoption(parser: Parser):
-    parser.addoption(
-        "--pg-version",
-        action="store",
-        type=PgVersion,
-        help="Postgres version to use for tests",
-    )
-
-
-@pytest.fixture(scope="session")
-def pg_version(request: FixtureRequest) -> Iterator[PgVersion]:
-    if v := request.config.getoption("--pg-version"):
-        version, source = v, "from --pg-version commad-line argument"
-    elif v := os.environ.get("DEFAULT_PG_VERSION"):
-        version, source = PgVersion(v), "from DEFAULT_PG_VERSION environment variable"
-    else:
-        version, source = DEFAULT_VERSION, "default verson"
-
-    log.info(f"pg_version is {version} ({source})")
-    yield version

test_runner/pg_clients/csharp/npgsql/csharp-npgsql.csproj

@@ -8,7 +8,7 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="Npgsql" Version="7.0.4" />
+    <PackageReference Include="Npgsql" Version="7.0.0" />
   </ItemGroup>
 
 </Project>

test_runner/pg_clients/java/jdbc/.gitignore

@@ -0,0 +1 @@
+

test_runner/pg_clients/java/jdbc/Dockerfile

@@ -1,10 +1,10 @@
-FROM openjdk:20
+FROM openjdk:17
 WORKDIR /source
 
 COPY . .
 
 WORKDIR /app
-RUN curl --output postgresql.jar https://jdbc.postgresql.org/download/postgresql-42.6.0.jar && \
+RUN curl --output postgresql.jar https://jdbc.postgresql.org/download/postgresql-42.5.1.jar && \
     javac -d /app /source/Example.java
 
 CMD ["java", "-cp", "/app/postgresql.jar:.", "Example"]

test_runner/pg_clients/python/pg8000/requirements.txt

@@ -1,2 +1 @@
 pg8000==1.29.4
-scramp>=1.4.3

test_runner/pg_clients/rust/tokio-postgres/Cargo.lock

@@ -4,9 +4,9 @@ version = 3
 
 [[package]]
 name = "async-trait"
-version = "0.1.68"
+version = "0.1.65"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b9ccdd8f2a161be9bd5c023df56f1b2a0bd1d83872ae53b71a84a12c9bf6e842"
+checksum = "095183a3539c7c7649b2beb87c2d3f0591f3a7fed07761cc546d244e27e0238c"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -21,9 +21,9 @@ checksum = "d468802bab17cbc0cc575e9b053f41e72aa36bfa6b7f55e3529ffa43161b97fa"
 
 [[package]]
 name = "base64"
-version = "0.21.0"
+version = "0.13.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a4a4ddaa51a5bc52a6948f74c06d20aaaddb71924eab79b8c97a8c556e942d6a"
+checksum = "9e1b586273c5702936fe7b7d6896644d8be71e6314cfe09d3167c95f712589e8"
 
 [[package]]
 name = "bitflags"
@@ -33,9 +33,9 @@ checksum = "bef38d45163c2f1dde094a7dfd33ccf595c92905c8f8f4fdc18d06fb1037718a"
 
 [[package]]
 name = "block-buffer"
-version = "0.10.4"
+version = "0.10.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3078c7629b62d3f0439517fa394996acacc5cbc91c5a20d8c658e77abd503a71"
+checksum = "69cce20737498f97b993470a6e536b8523f0af7892a4f928cceb1ac5e52ebe7e"
 dependencies = [
  "generic-array",
 ]
@@ -76,15 +76,15 @@ dependencies = [
 
 [[package]]
 name = "core-foundation-sys"
-version = "0.8.4"
+version = "0.8.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e496a50fda8aacccc86d7529e2c1e0892dbd0f898a6b5645b5561b89c3210efa"
+checksum = "5827cebf4670468b8772dd191856768aedcb1b0278a04f989f7766351917b9dc"
 
 [[package]]
 name = "cpufeatures"
-version = "0.2.7"
+version = "0.2.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3e4c1eaa2012c47becbbad2ab175484c2a84d1185b566fb2cc5b8707343dfe58"
+checksum = "28d997bd5e24a5928dd43e46dc529867e207907fe0b239c3477d924f7f2ca320"
 dependencies = [
  "libc",
 ]
@@ -112,13 +112,13 @@ dependencies = [
 
 [[package]]
 name = "errno"
-version = "0.3.1"
+version = "0.2.8"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4bcfec3a70f97c962c307b2d2c56e358cf1d00b558d74262b5f929ee8cc7e73a"
+checksum = "f639046355ee4f37944e44f60642c6f3a7efa3cf6b78c78a0d989a8ce6c396a1"
 dependencies = [
  "errno-dragonfly",
  "libc",
- "windows-sys 0.48.0",
+ "winapi",
 ]
 
 [[package]]
@@ -163,9 +163,9 @@ checksum = "00b0228411908ca8685dba7fc2cdd70ec9990a6e753e89b6ac91a84c40fbaf4b"
 
 [[package]]
 name = "futures"
-version = "0.3.28"
+version = "0.3.26"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "23342abe12aba583913b2e62f22225ff9c950774065e4bfb61a19cd9770fec40"
+checksum = "13e2792b0ff0340399d58445b88fd9770e3489eff258a4cbc1523418f12abf84"
 dependencies = [
  "futures-channel",
  "futures-core",
@@ -178,9 +178,9 @@ dependencies = [
 
 [[package]]
 name = "futures-channel"
-version = "0.3.28"
+version = "0.3.26"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "955518d47e09b25bbebc7a18df10b81f0c766eaf4c4f1cccef2fca5f2a4fb5f2"
+checksum = "2e5317663a9089767a1ec00a487df42e0ca174b61b4483213ac24448e4664df5"
 dependencies = [
  "futures-core",
  "futures-sink",
@@ -188,15 +188,15 @@ dependencies = [
 
 [[package]]
 name = "futures-core"
-version = "0.3.28"
+version = "0.3.26"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4bca583b7e26f571124fe5b7561d49cb2868d79116cfa0eefce955557c6fee8c"
+checksum = "ec90ff4d0fe1f57d600049061dc6bb68ed03c7d2fbd697274c41805dcb3f8608"
 
 [[package]]
 name = "futures-executor"
-version = "0.3.28"
+version = "0.3.26"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ccecee823288125bd88b4d7f565c9e58e41858e47ab72e8ea2d64e93624386e0"
+checksum = "e8de0a35a6ab97ec8869e32a2473f4b1324459e14c29275d14b10cb1fd19b50e"
 dependencies = [
  "futures-core",
  "futures-task",
@@ -205,15 +205,15 @@ dependencies = [
 
 [[package]]
 name = "futures-io"
-version = "0.3.28"
+version = "0.3.26"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4fff74096e71ed47f8e023204cfd0aa1289cd54ae5430a9523be060cdb849964"
+checksum = "bfb8371b6fb2aeb2d280374607aeabfc99d95c72edfe51692e42d3d7f0d08531"
 
 [[package]]
 name = "futures-macro"
-version = "0.3.28"
+version = "0.3.26"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "89ca545a94061b6365f2c7355b4b32bd20df3ff95f02da9329b34ccc3bd6ee72"
+checksum = "95a73af87da33b5acf53acfebdc339fe592ecf5357ac7c0a7734ab9d8c876a70"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -222,21 +222,21 @@ dependencies = [
 
 [[package]]
 name = "futures-sink"
-version = "0.3.28"
+version = "0.3.26"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f43be4fe21a13b9781a69afa4985b0f6ee0e1afab2c6f454a8cf30e2b2237b6e"
+checksum = "f310820bb3e8cfd46c80db4d7fb8353e15dfff853a127158425f31e0be6c8364"
 
 [[package]]
 name = "futures-task"
-version = "0.3.28"
+version = "0.3.26"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "76d3d132be6c0e6aa1534069c705a74a5997a356c0dc2f86a47765e5617c5b65"
+checksum = "dcf79a1bf610b10f42aea489289c5a2c478a786509693b80cd39c44ccd936366"
 
 [[package]]
 name = "futures-util"
-version = "0.3.28"
+version = "0.3.26"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "26b01e40b772d54cf6c6d721c1d1abd0647a0106a12ecaa1c186273392a69533"
+checksum = "9c1d6de3acfef38d2be4b1f543f553131788603495be83da675e180c8d6b7bd1"
 dependencies = [
  "futures-channel",
  "futures-core",
@@ -252,9 +252,9 @@ dependencies = [
 
 [[package]]
 name = "generic-array"
-version = "0.14.7"
+version = "0.14.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "85649ca51fd72272d7821adaf274ad91c288277713d9c18820d8499a7ff69e9a"
+checksum = "bff49e947297f3312447abdca79f45f4738097cc82b06e72054d2223f601f1b9"
 dependencies = [
  "typenum",
  "version_check",
@@ -262,21 +262,15 @@ dependencies = [
 
 [[package]]
 name = "getrandom"
-version = "0.2.9"
+version = "0.2.8"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c85e1d9ab2eadba7e5040d4e09cbd6d072b76a557ad64e797c2cb9d4da21d7e4"
+checksum = "c05aeb6a22b8f62540c194aac980f2115af067bfe15a0734d7277a768d396b31"
 dependencies = [
  "cfg-if",
  "libc",
  "wasi",
 ]
 
-[[package]]
-name = "hermit-abi"
-version = "0.3.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fed44880c466736ef9a5c5b5facefb5ed0785676d0c02d612db14e54f0d84286"
-
 [[package]]
 name = "hmac"
 version = "0.12.1"
@@ -297,13 +291,12 @@ dependencies = [
 
 [[package]]
 name = "io-lifetimes"
-version = "1.0.10"
+version = "1.0.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9c66c74d2ae7e79a5a8f7ac924adbe38ee42a859c6539ad869eb51f0b52dc220"
+checksum = "1abeb7a0dd0f8181267ff8adc397075586500b81b28a73e8a0208b00fc170fb3"
 dependencies = [
- "hermit-abi",
  "libc",
- "windows-sys 0.48.0",
+ "windows-sys 0.45.0",
 ]
 
 [[package]]
@@ -314,15 +307,15 @@ checksum = "e2abad23fbc42b3700f2f279844dc832adb2b2eb069b2df918f455c4e18cc646"
 
 [[package]]
 name = "libc"
-version = "0.2.142"
+version = "0.2.139"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6a987beff54b60ffa6d51982e1aa1146bc42f19bd26be28b0586f252fccf5317"
+checksum = "201de327520df007757c1f0adce6e827fe8562fbc28bfd9c15571c66ca1f5f79"
 
 [[package]]
 name = "linux-raw-sys"
-version = "0.3.4"
+version = "0.1.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "36eb31c1778188ae1e64398743890d0877fef36d11521ac60406b42016e8c2cf"
+checksum = "f051f77a7c8e6957c0696eac88f26b0117e54f52d3fc682ab19397a8812846a4"
 
 [[package]]
 name = "lock_api"
@@ -396,9 +389,9 @@ checksum = "b7e5500299e16ebb147ae15a00a942af264cf3688f47923b8fc2cd5858f23ad3"
 
 [[package]]
 name = "openssl"
-version = "0.10.52"
+version = "0.10.48"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "01b8574602df80f7b85fdfc5392fa884a4e3b3f4f35402c070ab34c3d3f78d56"
+checksum = "518915b97df115dd36109bfa429a48b8f737bd05508cf9588977b599648926d2"
 dependencies = [
  "bitflags",
  "cfg-if",
@@ -411,9 +404,9 @@ dependencies = [
 
 [[package]]
 name = "openssl-macros"
-version = "0.1.1"
+version = "0.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a948666b637a0f465e8564c73e89d4dde00d72d4d473cc972f390fc3dcee7d9c"
+checksum = "b501e44f11665960c7e7fcf062c7d96a14ade4aa98116c004b2e37b5be7d736c"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -428,10 +421,11 @@ checksum = "ff011a302c396a5197692431fc1948019154afc178baf7d8e37367442a4601cf"
 
 [[package]]
 name = "openssl-sys"
-version = "0.9.87"
+version = "0.9.83"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8e17f59264b2809d77ae94f0e1ebabc434773f370d6ca667bd223ea10e06cc7e"
+checksum = "666416d899cf077260dac8698d60a60b435a46d57e82acb1be3d0dad87284e5b"
 dependencies = [
+ "autocfg",
  "cc",
  "libc",
  "pkg-config",
@@ -456,7 +450,7 @@ checksum = "9069cbb9f99e3a5083476ccb29ceb1de18b9118cafa53e90c9551235de2b9521"
 dependencies = [
  "cfg-if",
  "libc",
- "redox_syscall 0.2.16",
+ "redox_syscall",
  "smallvec",
  "windows-sys 0.45.0",
 ]
@@ -518,9 +512,9 @@ dependencies = [
 
 [[package]]
 name = "postgres-protocol"
-version = "0.6.5"
+version = "0.6.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "78b7fa9f396f51dffd61546fd8573ee20592287996568e6175ceb0f8699ad75d"
+checksum = "878c6cbf956e03af9aa8204b407b9cbf47c072164800aa918c516cd4b056c50c"
 dependencies = [
  "base64",
  "byteorder",
@@ -536,9 +530,9 @@ dependencies = [
 
 [[package]]
 name = "postgres-types"
-version = "0.2.5"
+version = "0.2.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f028f05971fe20f512bcc679e2c10227e57809a3af86a7606304435bc8896cd6"
+checksum = "73d946ec7d256b04dfadc4e6a3292324e6f417124750fc5c0950f981b703a0f1"
 dependencies = [
  "bytes",
  "fallible-iterator",
@@ -553,18 +547,18 @@ checksum = "5b40af805b3121feab8a3c29f04d8ad262fa8e0561883e7653e024ae4479e6de"
 
 [[package]]
 name = "proc-macro2"
-version = "1.0.56"
+version = "1.0.51"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2b63bdb0cd06f1f4dedf69b254734f9b45af66e4a031e42a7480257d9898b435"
+checksum = "5d727cae5b39d21da60fa540906919ad737832fe0b1c165da3a34d6548c849d6"
 dependencies = [
  "unicode-ident",
 ]
 
 [[package]]
 name = "quote"
-version = "1.0.26"
+version = "1.0.23"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4424af4bf778aae2051a77b60283332f386554255d722233d09fbfc7e30da2fc"
+checksum = "8856d8364d252a14d474036ea1358d63c9e6965c8e5c1885c18f73d70bff9c7b"
 dependencies = [
  "proc-macro2",
 ]
@@ -608,15 +602,6 @@ dependencies = [
  "bitflags",
 ]
 
-[[package]]
-name = "redox_syscall"
-version = "0.3.5"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "567664f262709473930a4bf9e51bf2ebf3348f2e748ccc50dea20646858f8f29"
-dependencies = [
- "bitflags",
-]
-
 [[package]]
 name = "rust-neon-example"
 version = "0.1.0"
@@ -629,16 +614,16 @@ dependencies = [
 
 [[package]]
 name = "rustix"
-version = "0.37.15"
+version = "0.36.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a0661814f891c57c930a610266415528da53c4933e6dea5fb350cbfe048a9ece"
+checksum = "fd5c6ff11fecd55b40746d1995a02f2eb375bf8c00d192d521ee09f42bef37bc"
 dependencies = [
  "bitflags",
  "errno",
  "io-lifetimes",
  "libc",
  "linux-raw-sys",
- "windows-sys 0.48.0",
+ "windows-sys 0.45.0",
 ]
 
 [[package]]
@@ -721,16 +706,6 @@ dependencies = [
  "winapi",
 ]
 
-[[package]]
-name = "socket2"
-version = "0.5.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6d283f86695ae989d1e18440a943880967156325ba025f05049946bff47bcc2b"
-dependencies = [
- "libc",
- "windows-sys 0.48.0",
-]
-
 [[package]]
 name = "stringprep"
 version = "0.1.2"
@@ -749,9 +724,9 @@ checksum = "6bdef32e8150c2a081110b42772ffe7d7c9032b606bc226c8260fd97e0976601"
 
 [[package]]
 name = "syn"
-version = "2.0.15"
+version = "1.0.109"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a34fcf3e8b60f57e6a14301a2e916d323af98b0ea63c599441eec8558660c822"
+checksum = "72b64191b275b66ffe2469e8af2c1cfe3bafa67b529ead792a6d0160888b4237"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -760,15 +735,15 @@ dependencies = [
 
 [[package]]
 name = "tempfile"
-version = "3.5.0"
+version = "3.4.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b9fbec84f381d5795b08656e4912bec604d162bff9291d6189a78f4c8ab87998"
+checksum = "af18f7ae1acd354b992402e9ec5864359d693cd8a79dcbef59f76891701c1e95"
 dependencies = [
  "cfg-if",
  "fastrand",
- "redox_syscall 0.3.5",
+ "redox_syscall",
  "rustix",
- "windows-sys 0.45.0",
+ "windows-sys 0.42.0",
 ]
 
 [[package]]
@@ -788,25 +763,26 @@ checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20"
 
 [[package]]
 name = "tokio"
-version = "1.28.0"
+version = "1.26.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c3c786bf8134e5a3a166db9b29ab8f48134739014a3eca7bc6bfa95d673b136f"
+checksum = "03201d01c3c27a29c8a5cee5b55a93ddae1ccf6f08f65365c2c918f8c1b76f64"
 dependencies = [
  "autocfg",
  "bytes",
  "libc",
+ "memchr",
  "mio",
  "pin-project-lite",
- "socket2 0.4.9",
+ "socket2",
  "tokio-macros",
- "windows-sys 0.48.0",
+ "windows-sys 0.45.0",
 ]
 
 [[package]]
 name = "tokio-macros"
-version = "2.1.0"
+version = "1.8.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "630bdcf245f78637c13ec01ffae6187cca34625e8c63150d424b59e55af2675e"
+checksum = "d266c00fde287f55d3f1c3e96c500c362a2b8c695076ec180f27918820bc6df8"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -825,9 +801,9 @@ dependencies = [
 
 [[package]]
 name = "tokio-postgres"
-version = "0.7.8"
+version = "0.7.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6e89f6234aa8fd43779746012fcf53603cdb91fdd8399aa0de868c2d56b6dde1"
+checksum = "29a12c1b3e0704ae7dfc25562629798b29c72e6b1d0a681b6f29ab4ae5e7f7bf"
 dependencies = [
  "async-trait",
  "byteorder",
@@ -842,16 +818,16 @@ dependencies = [
  "pin-project-lite",
  "postgres-protocol",
  "postgres-types",
- "socket2 0.5.2",
+ "socket2",
  "tokio",
  "tokio-util",
 ]
 
 [[package]]
 name = "tokio-util"
-version = "0.7.8"
+version = "0.7.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "806fe8c2c87eccc8b3267cbae29ed3ab2d0bd37fca70ab622e46aaa9375ddb7d"
+checksum = "5427d89453009325de0d8f342c9490009f76e999cb7672d77e46267448f7e6b2"
 dependencies = [
  "bytes",
  "futures-core",
@@ -863,10 +839,11 @@ dependencies = [
 
 [[package]]
 name = "tracing"
-version = "0.1.38"
+version = "0.1.37"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "cf9cf6a813d3f40c88b0b6b6f29a5c95c6cdbf97c1f9cc53fb820200f5ad814d"
+checksum = "8ce8c33a8d48bd45d624a6e523445fd21ec13d3653cd51f681abf67418f54eb8"
 dependencies = [
+ "cfg-if",
  "pin-project-lite",
  "tracing-core",
 ]
@@ -888,15 +865,15 @@ checksum = "497961ef93d974e23eb6f433eb5fe1b7930b659f06d12dec6fc44a8f554c0bba"
 
 [[package]]
 name = "unicode-bidi"
-version = "0.3.13"
+version = "0.3.10"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "92888ba5573ff080736b3648696b70cafad7d250551175acbaa4e0385b3e1460"
+checksum = "d54675592c1dbefd78cbd98db9bacd89886e1ca50692a0692baefffdeb92dd58"
 
 [[package]]
 name = "unicode-ident"
-version = "1.0.8"
+version = "1.0.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e5464a87b239f13a63a501f2701565754bae92d243d4bb7eb12f6d57d2269bf4"
+checksum = "775c11906edafc97bc378816b94585fbd9a054eabaf86fdd0ced94af449efab7"
 
 [[package]]
 name = "unicode-normalization"
@@ -953,13 +930,13 @@ version = "0.42.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "5a3e1820f08b8513f676f7ab6c1f99ff312fb97b553d30ff4dd86f9f15728aa7"
 dependencies = [
- "windows_aarch64_gnullvm 0.42.2",
- "windows_aarch64_msvc 0.42.2",
- "windows_i686_gnu 0.42.2",
- "windows_i686_msvc 0.42.2",
- "windows_x86_64_gnu 0.42.2",
- "windows_x86_64_gnullvm 0.42.2",
- "windows_x86_64_msvc 0.42.2",
+ "windows_aarch64_gnullvm",
+ "windows_aarch64_msvc",
+ "windows_i686_gnu",
+ "windows_i686_msvc",
+ "windows_x86_64_gnu",
+ "windows_x86_64_gnullvm",
+ "windows_x86_64_msvc",
 ]
 
 [[package]]
@@ -968,128 +945,62 @@ version = "0.45.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "75283be5efb2831d37ea142365f009c02ec203cd29a3ebecbc093d52315b66d0"
 dependencies = [
- "windows-targets 0.42.2",
-]
-
-[[package]]
-name = "windows-sys"
-version = "0.48.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "677d2418bec65e3338edb076e806bc1ec15693c5d0104683f2efe857f61056a9"
-dependencies = [
- "windows-targets 0.48.0",
+ "windows-targets",
 ]
 
 [[package]]
 name = "windows-targets"
-version = "0.42.2"
+version = "0.42.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8e5180c00cd44c9b1c88adb3693291f1cd93605ded80c250a75d472756b4d071"
+checksum = "8e2522491fbfcd58cc84d47aeb2958948c4b8982e9a2d8a2a35bbaed431390e7"
 dependencies = [
- "windows_aarch64_gnullvm 0.42.2",
- "windows_aarch64_msvc 0.42.2",
- "windows_i686_gnu 0.42.2",
- "windows_i686_msvc 0.42.2",
- "windows_x86_64_gnu 0.42.2",
- "windows_x86_64_gnullvm 0.42.2",
- "windows_x86_64_msvc 0.42.2",
-]
-
-[[package]]
-name = "windows-targets"
-version = "0.48.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7b1eb6f0cd7c80c79759c929114ef071b87354ce476d9d94271031c0497adfd5"
-dependencies = [
- "windows_aarch64_gnullvm 0.48.0",
- "windows_aarch64_msvc 0.48.0",
- "windows_i686_gnu 0.48.0",
- "windows_i686_msvc 0.48.0",
- "windows_x86_64_gnu 0.48.0",
- "windows_x86_64_gnullvm 0.48.0",
- "windows_x86_64_msvc 0.48.0",
+ "windows_aarch64_gnullvm",
+ "windows_aarch64_msvc",
+ "windows_i686_gnu",
+ "windows_i686_msvc",
+ "windows_x86_64_gnu",
+ "windows_x86_64_gnullvm",
+ "windows_x86_64_msvc",
 ]
 
 [[package]]
 name = "windows_aarch64_gnullvm"
-version = "0.42.2"
+version = "0.42.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "597a5118570b68bc08d8d59125332c54f1ba9d9adeedeef5b99b02ba2b0698f8"
-
-[[package]]
-name = "windows_aarch64_gnullvm"
-version = "0.48.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "91ae572e1b79dba883e0d315474df7305d12f569b400fcf90581b06062f7e1bc"
+checksum = "8c9864e83243fdec7fc9c5444389dcbbfd258f745e7853198f365e3c4968a608"
 
 [[package]]
 name = "windows_aarch64_msvc"
-version = "0.42.2"
+version = "0.42.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e08e8864a60f06ef0d0ff4ba04124db8b0fb3be5776a5cd47641e942e58c4d43"
-
-[[package]]
-name = "windows_aarch64_msvc"
-version = "0.48.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b2ef27e0d7bdfcfc7b868b317c1d32c641a6fe4629c171b8928c7b08d98d7cf3"
+checksum = "4c8b1b673ffc16c47a9ff48570a9d85e25d265735c503681332589af6253c6c7"
 
 [[package]]
 name = "windows_i686_gnu"
-version = "0.42.2"
+version = "0.42.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c61d927d8da41da96a81f029489353e68739737d3beca43145c8afec9a31a84f"
-
-[[package]]
-name = "windows_i686_gnu"
-version = "0.48.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "622a1962a7db830d6fd0a69683c80a18fda201879f0f447f065a3b7467daa241"
+checksum = "de3887528ad530ba7bdbb1faa8275ec7a1155a45ffa57c37993960277145d640"
 
 [[package]]
 name = "windows_i686_msvc"
-version = "0.42.2"
+version = "0.42.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "44d840b6ec649f480a41c8d80f9c65108b92d89345dd94027bfe06ac444d1060"
-
-[[package]]
-name = "windows_i686_msvc"
-version = "0.48.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4542c6e364ce21bf45d69fdd2a8e455fa38d316158cfd43b3ac1c5b1b19f8e00"
+checksum = "bf4d1122317eddd6ff351aa852118a2418ad4214e6613a50e0191f7004372605"
 
 [[package]]
 name = "windows_x86_64_gnu"
-version = "0.42.2"
+version = "0.42.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8de912b8b8feb55c064867cf047dda097f92d51efad5b491dfb98f6bbb70cb36"
-
-[[package]]
-name = "windows_x86_64_gnu"
-version = "0.48.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ca2b8a661f7628cbd23440e50b05d705db3686f894fc9580820623656af974b1"
+checksum = "c1040f221285e17ebccbc2591ffdc2d44ee1f9186324dd3e84e99ac68d699c45"
 
 [[package]]
 name = "windows_x86_64_gnullvm"
-version = "0.42.2"
+version = "0.42.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "26d41b46a36d453748aedef1486d5c7a85db22e56aff34643984ea85514e94a3"
-
-[[package]]
-name = "windows_x86_64_gnullvm"
-version = "0.48.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7896dbc1f41e08872e9d5e8f8baa8fdd2677f29468c4e156210174edc7f7b953"
+checksum = "628bfdf232daa22b0d64fdb62b09fcc36bb01f05a3939e20ab73aaf9470d0463"
 
 [[package]]
 name = "windows_x86_64_msvc"
-version = "0.42.2"
+version = "0.42.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9aec5da331524158c6d1a4ac0ab1541149c0b9505fde06423b02f5ef0106b9f0"
-
-[[package]]
-name = "windows_x86_64_msvc"
-version = "0.48.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1a515f5799fe4961cb532f983ce2b23082366b898e52ffbce459c86f67c8378a"
+checksum = "447660ad36a13288b1db4d4248e857b510e8c3a225c822ba4fb748c0aafecffd"

test_runner/pg_clients/rust/tokio-postgres/Cargo.toml

@@ -9,8 +9,8 @@ publish = false
 [dependencies]
 native-tls = "0.2.11"
 postgres-native-tls = "0.5.0"
-tokio = { version = "1.28", features=["rt", "macros"] }
-tokio-postgres = "0.7.8"
+tokio = { version = "1.26", features=["rt", "macros"] }
+tokio-postgres = "0.7.7"
 
 
 # This is not part of the main 'neon' workspace

test_runner/pg_clients/rust/tokio-postgres/Dockerfile

@@ -1,4 +1,4 @@
-FROM rust:1.69
+FROM rust:1.67
 WORKDIR /source
 
 COPY . .

test_runner/pg_clients/swift/PostgresClientKitExample/Dockerfile

@@ -1,11 +1,11 @@
-FROM swift:5.8 AS build
+FROM swift:5.7 AS build
 RUN apt-get -q update && apt-get -q install -y libssl-dev
 WORKDIR /source
 
 COPY . .
 RUN swift build --configuration release
 
-FROM swift:5.8
+FROM swift:5.7
 WORKDIR /app
 COPY --from=build /source/.build/release .
 CMD ["/app/PostgresClientKitExample"]

test_runner/pg_clients/swift/PostgresClientKitExample/Package.swift

@@ -1,4 +1,4 @@
-// swift-tools-version:5.8
+// swift-tools-version:5.7
 import PackageDescription
 
 let package = Package(

test_runner/pg_clients/swift/PostgresClientKitExample/Sources/PostgresClientKitExample/main.swift

@@ -3,22 +3,21 @@ import Foundation
 import PostgresClientKit
 
 do {
-    let env = ProcessInfo.processInfo.environment
-
     var configuration = PostgresClientKit.ConnectionConfiguration()
-    let host = env["NEON_HOST"] ?? ""
-    configuration.host = host
-    configuration.port = 5432
-    configuration.database = env["NEON_DATABASE"] ?? ""
-    configuration.user = env["NEON_USER"] ?? ""
 
-    // PostgresClientKit uses Kitura/BlueSSLService which doesn't support SNI
-    // PostgresClientKit doesn't support setting connection options, so we use "Workaround D"
-    // See https://neon.tech/sni
-    let password = env["NEON_PASSWORD"] ?? ""
-    let endpoint_id = host.split(separator: ".")[0]
-    let workaroundD = "project=\(endpoint_id);\(password)"
-    configuration.credential = .cleartextPassword(password: workaroundD)
+    let env = ProcessInfo.processInfo.environment
+    if let host = env["NEON_HOST"] {
+        configuration.host = host
+    }
+    if let database = env["NEON_DATABASE"] {
+        configuration.database = database
+    }
+    if let user = env["NEON_USER"] {
+        configuration.user = user
+    }
+    if let password = env["NEON_PASSWORD"] {
+        configuration.credential = .cleartextPassword(password: password)
+    }
 
     let connection = try PostgresClientKit.Connection(configuration: configuration)
     defer { connection.close() }

test_runner/pg_clients/swift/PostgresNIOExample/Dockerfile

@@ -1,10 +1,10 @@
-FROM swift:5.8 AS build
+FROM swift:5.7 AS build
 WORKDIR /source
 
 COPY . .
 RUN swift build --configuration release
 
-FROM swift:5.8
+FROM swift:5.7
 WORKDIR /app
 COPY --from=build /source/.build/release .
 CMD ["/app/PostgresNIOExample"]

test_runner/pg_clients/swift/PostgresNIOExample/Package.resolved

@@ -5,8 +5,8 @@
       "kind" : "remoteSourceControl",
       "location" : "https://github.com/vapor/postgres-nio.git",
       "state" : {
-        "revision" : "dbf9c2eb596df39cba8ff3f74d74b2e6a31bd937",
-        "version" : "1.14.1"
+        "revision" : "7daf026e145de2c07d6e37f4171b1acb4b5f22b1",
+        "version" : "1.12.1"
       }
     },
     {
@@ -14,8 +14,8 @@
       "kind" : "remoteSourceControl",
       "location" : "https://github.com/apple/swift-atomics.git",
       "state" : {
-        "revision" : "6c89474e62719ddcc1e9614989fff2f68208fe10",
-        "version" : "1.1.0"
+        "revision" : "ff3d2212b6b093db7f177d0855adbc4ef9c5f036",
+        "version" : "1.0.3"
       }
     },
     {
@@ -59,8 +59,8 @@
       "kind" : "remoteSourceControl",
       "location" : "https://github.com/apple/swift-nio.git",
       "state" : {
-        "revision" : "d1690f85419fdac8d54e350fb6d2ab9fd95afd75",
-        "version" : "2.51.1"
+        "revision" : "45167b8006448c79dda4b7bd604e07a034c15c49",
+        "version" : "2.48.0"
       }
     },
     {
@@ -68,8 +68,8 @@
       "kind" : "remoteSourceControl",
       "location" : "https://github.com/apple/swift-nio-ssl.git",
       "state" : {
-        "revision" : "e866a626e105042a6a72a870c88b4c531ba05f83",
-        "version" : "2.24.0"
+        "revision" : "4fb7ead803e38949eb1d6fabb849206a72c580f3",
+        "version" : "2.23.0"
       }
     },
     {
@@ -77,8 +77,8 @@
       "kind" : "remoteSourceControl",
       "location" : "https://github.com/apple/swift-nio-transport-services.git",
       "state" : {
-        "revision" : "41f4098903878418537020075a4d8a6e20a0b182",
-        "version" : "1.17.0"
+        "revision" : "c0d9a144cfaec8d3d596aadde3039286a266c15c",
+        "version" : "1.15.0"
       }
     }
   ],

test_runner/pg_clients/swift/PostgresNIOExample/Package.swift

@@ -1,10 +1,10 @@
-// swift-tools-version:5.8
+// swift-tools-version:5.7
 import PackageDescription
 
 let package = Package(
     name: "PostgresNIOExample",
     dependencies: [
-        .package(url: "https://github.com/vapor/postgres-nio.git", from: "1.14.1")
+        .package(url: "https://github.com/vapor/postgres-nio.git", from: "1.8.0")
     ],
     targets: [
         .executableTarget(

test_runner/pg_clients/swift/PostgresNIOExample/Sources/PostgresNIOExample/main.swift

@@ -14,11 +14,15 @@ await Task {
     let sslContext = try! NIOSSLContext(configuration: .makeClientConfiguration())
 
     let config = PostgresConnection.Configuration(
-      host: env["NEON_HOST"] ?? "",
-      port: 5432,
-      username: env["NEON_USER"] ?? "",
-      password: env["NEON_PASSWORD"] ?? "",
-      database: env["NEON_DATABASE"] ?? "",
+      connection: .init(
+        host: env["NEON_HOST"] ?? "",
+        port: 5432
+      ),
+      authentication: .init(
+        username: env["NEON_USER"] ?? "",
+        database: env["NEON_DATABASE"] ?? "",
+        password: env["NEON_PASSWORD"] ?? ""
+      ),
       tls: .require(sslContext)
     )
 
@@ -38,7 +42,7 @@ await Task {
     try await connection.close()
 
     // Shutdown the EventLoopGroup, once all connections are closed.
-    try await eventLoopGroup.shutdownGracefully()
+    try eventLoopGroup.syncShutdownGracefully()
   } catch {
       print(error)
   }

test_runner/pg_clients/test_pg_clients.py

@@ -16,9 +16,10 @@ from fixtures.utils import subprocess_capture
         "rust/tokio-postgres",
         "python/asyncpg",
         "python/pg8000",
-        # PostgresClientKitExample does not support SNI or connection options, so it uses workaround D (https://neon.tech/sni)
-        # See example itself: test_runner/pg_clients/swift/PostgresClientKitExample/Sources/PostgresClientKitExample/main.swift
-        "swift/PostgresClientKitExample",
+        pytest.param(
+            "swift/PostgresClientKitExample",  # See https://github.com/neondatabase/neon/pull/2008#discussion_r911896592
+            marks=pytest.mark.xfail(reason="Neither SNI nor parameters is supported"),
+        ),
         "swift/PostgresNIOExample",
         "typescript/postgresql-client",
         "typescript/serverless-driver",

test_runner/pg_clients/typescript/postgresql-client/Dockerfile

@@ -1,4 +1,4 @@
-FROM node:20
+FROM node:18
 WORKDIR /source
 
 COPY . .

test_runner/pg_clients/typescript/postgresql-client/package-lock.json

@@ -5,7 +5,7 @@
   "packages": {
     "": {
       "dependencies": {
-        "postgresql-client": "2.5.5"
+        "postgresql-client": "2.5.2"
       }
     },
     "node_modules/debug": {
@@ -63,18 +63,18 @@
       }
     },
     "node_modules/postgresql-client": {
-      "version": "2.5.5",
-      "resolved": "https://registry.npmjs.org/postgresql-client/-/postgresql-client-2.5.5.tgz",
-      "integrity": "sha512-2Mu3i+6NQ9cnkoZNd0XeSZo9WoUpuWf4ZSiCCoDWSj82T93py2/SKXZ1aUaP8mVaU0oKpyyGe0IwLYZ1VHShnA==",
+      "version": "2.5.2",
+      "resolved": "https://registry.npmjs.org/postgresql-client/-/postgresql-client-2.5.2.tgz",
+      "integrity": "sha512-BaVDEmPlZllcmXBbw48194a6sB1YEe+ACX8c3SfgpGeW9+xJ5vShQ/ruNZtI+nuPW95BjL1WQGaxy+SfxSQgUQ==",
       "dependencies": {
         "debug": "^4.3.4",
-        "doublylinked": "^2.5.2",
-        "lightning-pool": "^4.2.1",
+        "doublylinked": "^2.5.1",
+        "lightning-pool": "^4.2.0",
         "postgres-bytea": "^3.0.0",
-        "power-tasks": "^1.6.4",
-        "putil-merge": "^3.10.3",
-        "putil-promisify": "^1.10.0",
-        "putil-varhelpers": "^1.6.5"
+        "power-tasks": "^1.5.0",
+        "putil-merge": "^3.9.0",
+        "putil-promisify": "^1.8.5",
+        "putil-varhelpers": "^1.6.4"
       },
       "engines": {
         "node": ">=14.0",
@@ -82,9 +82,9 @@
       }
     },
     "node_modules/power-tasks": {
-      "version": "1.6.4",
-      "resolved": "https://registry.npmjs.org/power-tasks/-/power-tasks-1.6.4.tgz",
-      "integrity": "sha512-LX8GGgEIP1N7jsZqlqZ275e6f1Ehq97APCEGj8uVO0NoEoB+77QUX12BFv3LmlNKfq4fIuNSPiHhyHFjqn2gfA==",
+      "version": "1.6.3",
+      "resolved": "https://registry.npmjs.org/power-tasks/-/power-tasks-1.6.3.tgz",
+      "integrity": "sha512-nBqzjbiCxvftEKsJtbEz5ZVKVl6RdwA5I7Ts3Z7DCe3lkvFsv9d8J4qp+b9GbdddsfV1KyIPSqPyLWq2YJQh6g==",
       "dependencies": {
         "debug": "^4.3.4",
         "doublylinked": "^2.5.2",
@@ -96,9 +96,9 @@
       }
     },
     "node_modules/putil-merge": {
-      "version": "3.10.3",
-      "resolved": "https://registry.npmjs.org/putil-merge/-/putil-merge-3.10.3.tgz",
-      "integrity": "sha512-B18CYi0/SmBYl9+fgowYWkgzJM/8XcLSeafHrFrGzwySQuOzLW0sOGx0CdFVp9zqaxgLctexUdGoSPpm6CPM6A==",
+      "version": "3.10.1",
+      "resolved": "https://registry.npmjs.org/putil-merge/-/putil-merge-3.10.1.tgz",
+      "integrity": "sha512-t3cLn14qccFvmb4bYQfNEHoisab//bTjM3lp56Ks8rOsjWF2ssf7Vapg9Lt89GlEawyNdeu+xj5GSrsFqNoCDQ==",
       "engines": {
         "node": ">= 10.0"
       }
@@ -112,9 +112,9 @@
       }
     },
     "node_modules/putil-varhelpers": {
-      "version": "1.6.5",
-      "resolved": "https://registry.npmjs.org/putil-varhelpers/-/putil-varhelpers-1.6.5.tgz",
-      "integrity": "sha512-kyu+lE5xkc65ScgaIi6rNONLXeS7jGBxl1I0rzHVsFGAAQ45D/VkuEev+t09PFB943F+CqdWFLczH6ePk5TPAA==",
+      "version": "1.6.4",
+      "resolved": "https://registry.npmjs.org/putil-varhelpers/-/putil-varhelpers-1.6.4.tgz",
+      "integrity": "sha512-nM2nO1HS2yJUyPgz0grd2XZAM0Spr6/tt6F4xXeNDjByV00BV2mq6lZ+sDff8WIfQBI9Hn1Czh93H1xBvKESxw==",
       "engines": {
         "node": ">= 6.0"
       }

test_runner/pg_clients/typescript/postgresql-client/package.json

@@ -1,6 +1,6 @@
 {
   "type": "module",
   "dependencies": {
-    "postgresql-client": "2.5.5"
+    "postgresql-client": "2.5.2"
   }
 }

test_runner/pg_clients/typescript/serverless-driver/index.js

@@ -1,11 +1,8 @@
 #! /usr/bin/env node
 
-import ws from 'ws'
-import { neonConfig, Client } from '@neondatabase/serverless'
+import { Client } from '@neondatabase/serverless'
 
 (async () => {
-    neonConfig.webSocketConstructor = ws
-
     const client = new Client({
         host: process.env.NEON_HOST,
         database: process.env.NEON_DATABASE,

test_runner/pg_clients/typescript/serverless-driver/package-lock.json

@@ -1,99 +1,18 @@
 {
   "name": "serverless-driver",
-  "lockfileVersion": 3,
+  "lockfileVersion": 2,
   "requires": true,
   "packages": {
     "": {
       "dependencies": {
-        "@neondatabase/serverless": "0.4.3",
-        "ws": "8.13.0"
+        "@neondatabase/serverless": "^0.2.8",
+        "ws": "^8.13.0"
       }
     },
     "node_modules/@neondatabase/serverless": {
-      "version": "0.4.3",
-      "resolved": "https://registry.npmjs.org/@neondatabase/serverless/-/serverless-0.4.3.tgz",
-      "integrity": "sha512-U8tpuF5f0R5WRsciR7iaJ5S2h54DWa6Z6CEW+J4KgwyvRN3q3qDz0MibdfFXU0WqnRoi/9RSf/2XN4TfeaOCbQ==",
-      "dependencies": {
-        "@types/pg": "^8.6.6"
-      }
-    },
-    "node_modules/@types/node": {
-      "version": "18.16.3",
-      "resolved": "https://registry.npmjs.org/@types/node/-/node-18.16.3.tgz",
-      "integrity": "sha512-OPs5WnnT1xkCBiuQrZA4+YAV4HEJejmHneyraIaxsbev5yCEr6KMwINNFP9wQeFIw8FWcoTqF3vQsa5CDaI+8Q=="
-    },
-    "node_modules/@types/pg": {
-      "version": "8.6.6",
-      "resolved": "https://registry.npmjs.org/@types/pg/-/pg-8.6.6.tgz",
-      "integrity": "sha512-O2xNmXebtwVekJDD+02udOncjVcMZQuTEQEMpKJ0ZRf5E7/9JJX3izhKUcUifBkyKpljyUM6BTgy2trmviKlpw==",
-      "dependencies": {
-        "@types/node": "*",
-        "pg-protocol": "*",
-        "pg-types": "^2.2.0"
-      }
-    },
-    "node_modules/pg-int8": {
-      "version": "1.0.1",
-      "resolved": "https://registry.npmjs.org/pg-int8/-/pg-int8-1.0.1.tgz",
-      "integrity": "sha512-WCtabS6t3c8SkpDBUlb1kjOs7l66xsGdKpIPZsg4wR+B3+u9UAum2odSsF9tnvxg80h4ZxLWMy4pRjOsFIqQpw==",
-      "engines": {
-        "node": ">=4.0.0"
-      }
-    },
-    "node_modules/pg-protocol": {
-      "version": "1.6.0",
-      "resolved": "https://registry.npmjs.org/pg-protocol/-/pg-protocol-1.6.0.tgz",
-      "integrity": "sha512-M+PDm637OY5WM307051+bsDia5Xej6d9IR4GwJse1qA1DIhiKlksvrneZOYQq42OM+spubpcNYEo2FcKQrDk+Q=="
-    },
-    "node_modules/pg-types": {
-      "version": "2.2.0",
-      "resolved": "https://registry.npmjs.org/pg-types/-/pg-types-2.2.0.tgz",
-      "integrity": "sha512-qTAAlrEsl8s4OiEQY69wDvcMIdQN6wdz5ojQiOy6YRMuynxenON0O5oCpJI6lshc6scgAY8qvJ2On/p+CXY0GA==",
-      "dependencies": {
-        "pg-int8": "1.0.1",
-        "postgres-array": "~2.0.0",
-        "postgres-bytea": "~1.0.0",
-        "postgres-date": "~1.0.4",
-        "postgres-interval": "^1.1.0"
-      },
-      "engines": {
-        "node": ">=4"
-      }
-    },
-    "node_modules/postgres-array": {
-      "version": "2.0.0",
-      "resolved": "https://registry.npmjs.org/postgres-array/-/postgres-array-2.0.0.tgz",
-      "integrity": "sha512-VpZrUqU5A69eQyW2c5CA1jtLecCsN2U/bD6VilrFDWq5+5UIEVO7nazS3TEcHf1zuPYO/sqGvUvW62g86RXZuA==",
-      "engines": {
-        "node": ">=4"
-      }
-    },
-    "node_modules/postgres-bytea": {
-      "version": "1.0.0",
-      "resolved": "https://registry.npmjs.org/postgres-bytea/-/postgres-bytea-1.0.0.tgz",
-      "integrity": "sha512-xy3pmLuQqRBZBXDULy7KbaitYqLcmxigw14Q5sj8QBVLqEwXfeybIKVWiqAXTlcvdvb0+xkOtDbfQMOf4lST1w==",
-      "engines": {
-        "node": ">=0.10.0"
-      }
-    },
-    "node_modules/postgres-date": {
-      "version": "1.0.7",
-      "resolved": "https://registry.npmjs.org/postgres-date/-/postgres-date-1.0.7.tgz",
-      "integrity": "sha512-suDmjLVQg78nMK2UZ454hAG+OAW+HQPZ6n++TNDUX+L0+uUlLywnoxJKDou51Zm+zTCjrCl0Nq6J9C5hP9vK/Q==",
-      "engines": {
-        "node": ">=0.10.0"
-      }
-    },
-    "node_modules/postgres-interval": {
-      "version": "1.2.0",
-      "resolved": "https://registry.npmjs.org/postgres-interval/-/postgres-interval-1.2.0.tgz",
-      "integrity": "sha512-9ZhXKM/rw350N1ovuWHbGxnGh/SNJ4cnxHiM0rxE4VN41wsg8P8zWn9hv/buK00RP4WvlOyr/RBDiptyxVbkZQ==",
-      "dependencies": {
-        "xtend": "^4.0.0"
-      },
-      "engines": {
-        "node": ">=0.10.0"
-      }
+      "version": "0.2.8",
+      "resolved": "https://registry.npmjs.org/@neondatabase/serverless/-/serverless-0.2.8.tgz",
+      "integrity": "sha512-+yWjIOJsFnrtt2xvtLVEzWM2lfvemawk/DBg4mD2cZOF/IC6Jn4wEctZyk60TscZMSxfozNkPoxmZvBmNuQ0vA=="
     },
     "node_modules/ws": {
       "version": "8.13.0",
@@ -114,14 +33,19 @@
           "optional": true
         }
       }
+    }
+  },
+  "dependencies": {
+    "@neondatabase/serverless": {
+      "version": "0.2.8",
+      "resolved": "https://registry.npmjs.org/@neondatabase/serverless/-/serverless-0.2.8.tgz",
+      "integrity": "sha512-+yWjIOJsFnrtt2xvtLVEzWM2lfvemawk/DBg4mD2cZOF/IC6Jn4wEctZyk60TscZMSxfozNkPoxmZvBmNuQ0vA=="
     },
-    "node_modules/xtend": {
-      "version": "4.0.2",
-      "resolved": "https://registry.npmjs.org/xtend/-/xtend-4.0.2.tgz",
-      "integrity": "sha512-LKYU1iAXJXUgAXn9URjiu+MWhyUXHsvfp7mcuYm9dSUKK0/CjtrUwFAxD82/mCWbtLsGjFIad0wIsod4zrTAEQ==",
-      "engines": {
-        "node": ">=0.4"
-      }
+    "ws": {
+      "version": "8.13.0",
+      "resolved": "https://registry.npmjs.org/ws/-/ws-8.13.0.tgz",
+      "integrity": "sha512-x9vcZYTrFPC7aSIbj7sRCYo7L/Xb8Iy+pW0ng0wt2vCJv7M9HOMy0UoN3rr+IFC7hb7vXoqS+P9ktyLLLhO+LA==",
+      "requires": {}
     }
   }
 }

test_runner/pg_clients/typescript/serverless-driver/package.json

@@ -1,7 +1,7 @@
 {
   "type": "module",
   "dependencies": {
-    "@neondatabase/serverless": "0.4.3",
-    "ws": "8.13.0"
+    "@neondatabase/serverless": "^0.2.8",
+    "ws": "^8.13.0"
   }
 }

test_runner/regress/test_auth.py

@@ -31,15 +31,11 @@ def test_pageserver_auth(neon_env_builder: NeonEnvBuilder):
 
     # tenant can create branches
     tenant_http_client.timeline_create(
-        pg_version=env.pg_version,
-        tenant_id=env.initial_tenant,
-        ancestor_timeline_id=new_timeline_id,
+        tenant_id=env.initial_tenant, ancestor_timeline_id=new_timeline_id
     )
     # console can create branches for tenant
     pageserver_http_client.timeline_create(
-        pg_version=env.pg_version,
-        tenant_id=env.initial_tenant,
-        ancestor_timeline_id=new_timeline_id,
+        tenant_id=env.initial_tenant, ancestor_timeline_id=new_timeline_id
     )
 
     # fail to create branch using token with different tenant_id
@@ -47,9 +43,7 @@ def test_pageserver_auth(neon_env_builder: NeonEnvBuilder):
         PageserverApiException, match="Forbidden: Tenant id mismatch. Permission denied"
     ):
         invalid_tenant_http_client.timeline_create(
-            pg_version=env.pg_version,
-            tenant_id=env.initial_tenant,
-            ancestor_timeline_id=new_timeline_id,
+            tenant_id=env.initial_tenant, ancestor_timeline_id=new_timeline_id
         )
 
     # create tenant using management token

test_runner/regress/test_compatibility.py

@@ -16,7 +16,6 @@ from fixtures.neon_fixtures import (
 )
 from fixtures.pageserver.http import PageserverHttpClient
 from fixtures.pageserver.utils import wait_for_last_record_lsn, wait_for_upload
-from fixtures.pg_version import PgVersion
 from fixtures.types import Lsn
 from pytest import FixtureRequest
 
@@ -51,7 +50,7 @@ def test_create_snapshot(neon_env_builder: NeonEnvBuilder, pg_bin: PgBin, test_o
     # 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 = PgVersion.V14
+    neon_env_builder.pg_version = "14"
     neon_env_builder.num_safekeepers = 3
     neon_env_builder.enable_local_fs_remote_storage()
     neon_env_builder.preserve_database_files = True
@@ -99,7 +98,7 @@ def test_backward_compatibility(
     test_output_dir: Path,
     neon_binpath: Path,
     pg_distrib_dir: Path,
-    pg_version: PgVersion,
+    pg_version: str,
     request: FixtureRequest,
 ):
     """
@@ -154,7 +153,7 @@ def test_backward_compatibility(
 def test_forward_compatibility(
     test_output_dir: Path,
     port_distributor: PortDistributor,
-    pg_version: PgVersion,
+    pg_version: str,
     request: FixtureRequest,
     neon_binpath: Path,
 ):
@@ -346,7 +345,7 @@ def check_neon_works(
     neon_target_binpath: Path,
     neon_current_binpath: Path,
     pg_distrib_dir: Path,
-    pg_version: PgVersion,
+    pg_version: str,
     port_distributor: PortDistributor,
     test_output_dir: Path,
     pg_bin: PgBin,
@@ -405,7 +404,7 @@ def check_neon_works(
 
     shutil.rmtree(repo_dir / "local_fs_remote_storage")
     pageserver_http.timeline_delete(tenant_id, timeline_id)
-    pageserver_http.timeline_create(pg_version, tenant_id, timeline_id)
+    pageserver_http.timeline_create(tenant_id, timeline_id)
     pg_bin.run(
         ["pg_dumpall", f"--dbname={connstr}", f"--file={test_output_dir / 'dump-from-wal.sql'}"]
     )

test_runner/regress/test_metric_collection.py

@@ -24,6 +24,13 @@ from pytest_httpserver import HTTPServer
 from werkzeug.wrappers.request import Request
 from werkzeug.wrappers.response import Response
 
+
+@pytest.fixture(scope="session")
+def httpserver_listen_address(port_distributor: PortDistributor):
+    port = port_distributor.get_port()
+    return ("localhost", port)
+
+
 # ==============================================================================
 # Storage metrics tests
 # ==============================================================================

test_runner/regress/test_pageserver_api.py

@@ -8,7 +8,6 @@ from fixtures.neon_fixtures import (
     NeonEnvBuilder,
 )
 from fixtures.pageserver.http import PageserverHttpClient
-from fixtures.pg_version import PgVersion
 from fixtures.types import Lsn, TenantId, TimelineId
 from fixtures.utils import wait_until
 
@@ -62,7 +61,7 @@ def test_pageserver_init_node_id(
     assert "has node id already, it cannot be overridden" in bad_update.stderr
 
 
-def check_client(pg_version: PgVersion, client: PageserverHttpClient, initial_tenant: TenantId):
+def check_client(client: PageserverHttpClient, initial_tenant: TenantId):
     client.check_status()
 
     # check initial tenant is there
@@ -78,11 +77,7 @@ def check_client(pg_version: PgVersion, client: PageserverHttpClient, initial_te
 
     # create timeline
     timeline_id = TimelineId.generate()
-    client.timeline_create(
-        pg_version=pg_version,
-        tenant_id=tenant_id,
-        new_timeline_id=timeline_id,
-    )
+    client.timeline_create(tenant_id=tenant_id, new_timeline_id=timeline_id)
 
     timelines = client.timeline_list(tenant_id)
     assert len(timelines) > 0
@@ -179,7 +174,7 @@ def test_pageserver_http_get_wal_receiver_success(neon_simple_env: NeonEnv):
 def test_pageserver_http_api_client(neon_simple_env: NeonEnv):
     env = neon_simple_env
     with env.pageserver.http_client() as client:
-        check_client(env.pg_version, client, env.initial_tenant)
+        check_client(client, env.initial_tenant)
 
 
 def test_pageserver_http_api_client_auth_enabled(neon_env_builder: NeonEnvBuilder):
@@ -189,4 +184,4 @@ def test_pageserver_http_api_client_auth_enabled(neon_env_builder: NeonEnvBuilde
     pageserver_token = env.auth_keys.generate_pageserver_token()
 
     with env.pageserver.http_client(auth_token=pageserver_token) as client:
-        check_client(env.pg_version, client, env.initial_tenant)
+        check_client(client, env.initial_tenant)

test_runner/regress/test_threshold_based_eviction.py

@@ -1,179 +0,0 @@
-import time
-from dataclasses import dataclass
-from typing import List, Set, Tuple
-
-from fixtures.log_helper import log
-from fixtures.neon_fixtures import (
-    NeonEnvBuilder,
-    PgBin,
-    RemoteStorageKind,
-    last_flush_lsn_upload,
-)
-from fixtures.pageserver.http import LayerMapInfo
-from fixtures.types import TimelineId
-from pytest_httpserver import HTTPServer
-
-# NB: basic config change tests are in test_tenant_conf.py
-
-
-def test_threshold_based_eviction(
-    request,
-    httpserver: HTTPServer,
-    httpserver_listen_address,
-    pg_bin: PgBin,
-    neon_env_builder: NeonEnvBuilder,
-):
-    neon_env_builder.enable_remote_storage(RemoteStorageKind.LOCAL_FS, f"{request.node.name}")
-
-    # Start with metrics collection enabled, so that the eviction task
-    # imitates its accesses. We'll use a non-existent endpoint to make it fail.
-    # The synthetic size calculation will run regardless.
-    host, port = httpserver_listen_address
-    neon_env_builder.pageserver_config_override = f"""
-        metric_collection_interval="1s"
-        synthetic_size_calculation_interval="2s"
-        metric_collection_endpoint="http://{host}:{port}/nonexistent"
-    """
-    metrics_refused_log_line = ".*metrics endpoint refused the sent metrics.*/nonexistent.*"
-    env = neon_env_builder.init_start()
-    env.pageserver.allowed_errors.append(metrics_refused_log_line)
-
-    tenant_id, timeline_id = env.initial_tenant, env.initial_timeline
-    assert isinstance(timeline_id, TimelineId)
-
-    ps_http = env.pageserver.http_client()
-    assert ps_http.tenant_config(tenant_id).effective_config["eviction_policy"] == {
-        "kind": "NoEviction"
-    }
-
-    eviction_threshold = 5
-    eviction_period = 1
-    ps_http.set_tenant_config(
-        tenant_id,
-        {
-            "eviction_policy": {
-                "kind": "LayerAccessThreshold",
-                "threshold": f"{eviction_threshold}s",
-                "period": f"{eviction_period}s",
-            },
-        },
-    )
-    assert ps_http.tenant_config(tenant_id).effective_config["eviction_policy"] == {
-        "kind": "LayerAccessThreshold",
-        "threshold": f"{eviction_threshold}s",
-        "period": f"{eviction_period}s",
-    }
-
-    # restart because changing tenant config is not instant
-    env.pageserver.stop()
-    env.pageserver.start()
-
-    assert ps_http.tenant_config(tenant_id).effective_config["eviction_policy"] == {
-        "kind": "LayerAccessThreshold",
-        "threshold": f"{eviction_threshold}s",
-        "period": f"{eviction_period}s",
-    }
-
-    # create a bunch of L1s, only the least of which will need to be resident
-    compaction_threshold = 3  # create L1 layers quickly
-    ps_http.patch_tenant_config_client_side(
-        tenant_id,
-        inserts={
-            # Disable gc and compaction to avoid on-demand downloads from their side.
-            # The only on-demand downloads should be from the eviction tasks's "imitate access" functions.
-            "gc_period": "0s",
-            "compaction_period": "0s",
-            # low checkpoint_distance so that pgbench creates many layers
-            "checkpoint_distance": 1024**2,
-            # Low compaction target size to create many L1's with tight key ranges.
-            # This is so that the "imitate access" don't download all the layers.
-            "compaction_target_size": 1 * 1024**2,  # all keys into one L1
-            # Turn L0's into L1's fast.
-            "compaction_threshold": compaction_threshold,
-            # Prevent compaction from collapsing L1 delta layers into image layers. We want many layers here.
-            "image_creation_threshold": 100,
-            # Much larger so that synthetic size caluclation worker, which is part of metric collection,
-            # computes logical size for initdb_lsn every time, instead of some moving lsn as we insert data.
-            # This makes the set of downloaded layers predictable,
-            # thereby allowing the residence statuses to stabilize below.
-            "gc_horizon": 1024**4,
-        },
-    )
-
-    # create a bunch of layers
-    with env.endpoints.create_start("main", tenant_id=tenant_id) as pg:
-        pg_bin.run(["pgbench", "-i", "-s", "3", pg.connstr()])
-        last_flush_lsn_upload(env, pg, tenant_id, timeline_id)
-    # wrap up and shutdown safekeepers so that no more layers will be created after the final checkpoint
-    for sk in env.safekeepers:
-        sk.stop()
-    ps_http.timeline_checkpoint(tenant_id, timeline_id)
-
-    # wait for evictions and assert that they stabilize
-    @dataclass
-    class ByLocalAndRemote:
-        remote_layers: Set[str]
-        local_layers: Set[str]
-
-    class MapInfoProjection:
-        def __init__(self, info: LayerMapInfo):
-            self.info = info
-
-        def by_local_and_remote(self) -> ByLocalAndRemote:
-            return ByLocalAndRemote(
-                remote_layers={
-                    layer.layer_file_name for layer in self.info.historic_layers if layer.remote
-                },
-                local_layers={
-                    layer.layer_file_name for layer in self.info.historic_layers if not layer.remote
-                },
-            )
-
-        def __eq__(self, other):
-            if not isinstance(other, MapInfoProjection):
-                return False
-            return self.by_local_and_remote() == other.by_local_and_remote()
-
-        def __repr__(self) -> str:
-            out = ["MapInfoProjection:"]
-            for layer in sorted(self.info.historic_layers, key=lambda layer: layer.layer_file_name):
-                remote = "R" if layer.remote else "L"
-                out += [f"  {remote} {layer.layer_file_name}"]
-            return "\n".join(out)
-
-    observation_window = 8 * eviction_threshold
-    consider_stable_when_no_change_for_seconds = 3 * eviction_threshold
-    poll_interval = eviction_threshold / 3
-    started_waiting_at = time.time()
-    map_info_changes: List[Tuple[float, MapInfoProjection]] = []
-    while time.time() - started_waiting_at < observation_window:
-        current = (
-            time.time(),
-            MapInfoProjection(ps_http.layer_map_info(tenant_id, timeline_id)),
-        )
-        last = map_info_changes[-1] if map_info_changes else (0, None)
-        if last[1] is None or current[1] != last[1]:
-            map_info_changes.append(current)
-            log.info("change in layer map\n before: %s\n after: %s", last, current)
-        else:
-            stable_for = current[0] - last[0]
-            log.info("residencies stable for %s", stable_for)
-            if stable_for > consider_stable_when_no_change_for_seconds:
-                break
-        time.sleep(poll_interval)
-
-    log.info("len(map_info_changes)=%s", len(map_info_changes))
-
-    # TODO: can we be more precise here? E.g., require we're stable _within_ X*threshold,
-    # instead of what we do here, i.e., stable _for at least_ X*threshold toward the end of the observation window
-    assert (
-        stable_for > consider_stable_when_no_change_for_seconds
-    ), "layer residencies did not become stable within the observation window"
-
-    post = map_info_changes[-1][1].by_local_and_remote()
-    assert len(post.remote_layers) > 0, "some layers should be evicted once it's stabilized"
-    assert len(post.local_layers) > 0, "the imitate accesses should keep some layers resident"
-
-    assert env.pageserver.log_contains(
-        metrics_refused_log_line
-    ), "ensure the metrics collection worker ran"

test_runner/regress/test_timeline_size.py

@@ -27,7 +27,6 @@ from fixtures.pageserver.utils import (
     wait_for_upload_queue_empty,
     wait_until_tenant_active,
 )
-from fixtures.pg_version import PgVersion
 from fixtures.types import TenantId, TimelineId
 from fixtures.utils import get_timeline_dir_size, wait_until
 
@@ -490,7 +489,7 @@ def test_timeline_size_metrics(
     test_output_dir: Path,
     port_distributor: PortDistributor,
     pg_distrib_dir: Path,
-    pg_version: PgVersion,
+    pg_version: str,
 ):
     env = neon_simple_env
     pageserver_http = env.pageserver.http_client()

test_runner/regress/test_wal_acceptor.py

@@ -32,7 +32,6 @@ from fixtures.neon_fixtures import (
     available_remote_storages,
 )
 from fixtures.pageserver.utils import wait_for_last_record_lsn, wait_for_upload
-from fixtures.pg_version import PgVersion
 from fixtures.types import Lsn, TenantId, TimelineId
 from fixtures.utils import get_dir_size, query_scalar, start_in_background
 
@@ -583,11 +582,7 @@ def test_s3_wal_replay(neon_env_builder: NeonEnvBuilder, remote_storage_kind: Re
         shutil.copy(f_partial_saved, f_partial_path)
 
     # recreate timeline on pageserver from scratch
-    ps_cli.timeline_create(
-        pg_version=PgVersion(pg_version),
-        tenant_id=tenant_id,
-        new_timeline_id=timeline_id,
-    )
+    ps_cli.timeline_create(tenant_id, timeline_id)
 
     wait_lsn_timeout = 60 * 3
     started_at = time.time()

test_runner/regress/test_wal_receiver.py

@@ -1,115 +0,0 @@
-from fixtures.log_helper import log
-from fixtures.neon_fixtures import NeonEnv, NeonEnvBuilder
-from fixtures.types import Lsn, TenantId
-
-
-# Checks that pageserver's walreceiver state is printed in the logs during WAL wait timeout.
-# Ensures that walreceiver does not run without any data inserted and only starts after the insertion.
-def test_pageserver_lsn_wait_error_start(neon_env_builder: NeonEnvBuilder):
-    # Trigger WAL wait timeout faster
-    neon_env_builder.pageserver_config_override = "wait_lsn_timeout = '1s'"
-    env = neon_env_builder.init_start()
-    env.pageserver.http_client()
-
-    tenant_id, timeline_id = env.neon_cli.create_tenant()
-    expected_timeout_error = f"Timed out while waiting for WAL record at LSN {future_lsn} to arrive"
-    env.pageserver.allowed_errors.append(f".*{expected_timeout_error}.*")
-
-    try:
-        trigger_wait_lsn_timeout(env, tenant_id)
-    except Exception as e:
-        exception_string = str(e)
-        assert expected_timeout_error in exception_string, "Should time out during waiting for WAL"
-        assert (
-            "WalReceiver status: Not active" in exception_string
-        ), "Walreceiver should not be active before any data writes"
-
-    insert_test_elements(env, tenant_id, start=0, count=1_000)
-    try:
-        trigger_wait_lsn_timeout(env, tenant_id)
-    except Exception as e:
-        exception_string = str(e)
-        assert expected_timeout_error in exception_string, "Should time out during waiting for WAL"
-        assert (
-            "WalReceiver status: Not active" not in exception_string
-        ), "Should not be inactive anymore after INSERTs are made"
-        assert "WalReceiver status" in exception_string, "But still should have some other status"
-
-
-# Checks that all active safekeepers are shown in pageserver's walreceiver state printed on WAL wait timeout.
-# Kills one of the safekeepers and ensures that only the active ones are printed in the state.
-def test_pageserver_lsn_wait_error_safekeeper_stop(neon_env_builder: NeonEnvBuilder):
-    # Trigger WAL wait timeout faster
-    neon_env_builder.pageserver_config_override = "wait_lsn_timeout = '1s'"
-    # Have notable SK ids to ensure we check logs for their presence, not some other random numbers
-    neon_env_builder.safekeepers_id_start = 12345
-    neon_env_builder.num_safekeepers = 3
-    env = neon_env_builder.init_start()
-    env.pageserver.http_client()
-
-    tenant_id, timeline_id = env.neon_cli.create_tenant()
-
-    elements_to_insert = 1_000_000
-    expected_timeout_error = f"Timed out while waiting for WAL record at LSN {future_lsn} to arrive"
-    env.pageserver.allowed_errors.append(f".*{expected_timeout_error}.*")
-
-    insert_test_elements(env, tenant_id, start=0, count=elements_to_insert)
-
-    try:
-        trigger_wait_lsn_timeout(env, tenant_id)
-    except Exception as e:
-        exception_string = str(e)
-        assert expected_timeout_error in exception_string, "Should time out during waiting for WAL"
-
-        for safekeeper in env.safekeepers:
-            assert (
-                str(safekeeper.id) in exception_string
-            ), f"Should have safekeeper {safekeeper.id} printed in walreceiver state after WAL wait timeout"
-
-    stopped_safekeeper = env.safekeepers[-1]
-    stopped_safekeeper_id = stopped_safekeeper.id
-    log.info(f"Stopping safekeeper {stopped_safekeeper.id}")
-    stopped_safekeeper.stop()
-
-    # Spend some more time inserting, to ensure SKs report updated statuses and walreceiver in PS have time to update its connection stats.
-    insert_test_elements(env, tenant_id, start=elements_to_insert + 1, count=elements_to_insert)
-
-    try:
-        trigger_wait_lsn_timeout(env, tenant_id)
-    except Exception as e:
-        exception_string = str(e)
-        assert expected_timeout_error in exception_string, "Should time out during waiting for WAL"
-
-        for safekeeper in env.safekeepers:
-            if safekeeper.id == stopped_safekeeper_id:
-                assert (
-                    str(safekeeper.id) not in exception_string
-                ), f"Should not have stopped safekeeper {safekeeper.id} printed in walreceiver state after 2nd WAL wait timeout"
-            else:
-                assert (
-                    str(safekeeper.id) in exception_string
-                ), f"Should have safekeeper {safekeeper.id} printed in walreceiver state after 2nd WAL wait timeout"
-
-
-def insert_test_elements(env: NeonEnv, tenant_id: TenantId, start: int, count: int):
-    first_element_id = start
-    last_element_id = first_element_id + count
-    with env.endpoints.create_start("main", tenant_id=tenant_id) as endpoint:
-        with endpoint.cursor() as cur:
-            cur.execute("CREATE TABLE IF NOT EXISTS t(key serial primary key, value text)")
-            cur.execute(
-                f"INSERT INTO t SELECT i, CONCAT('payload_', i) FROM generate_series({first_element_id},{last_element_id}) as i"
-            )
-
-
-future_lsn = Lsn("0/FFFFFFFF")
-
-
-def trigger_wait_lsn_timeout(env: NeonEnv, tenant_id: TenantId):
-    with env.endpoints.create_start(
-        "main",
-        tenant_id=tenant_id,
-        lsn=future_lsn,
-    ) as endpoint:
-        with endpoint.cursor() as cur:
-            cur.execute("SELECT 1")