make 100 layers

.github/actions/allure-report-generate/action.yml

@@ -57,14 +57,14 @@ runs:
         if ! which allure; then
           ALLURE_ZIP=allure-${ALLURE_VERSION}.zip
           wget -q https://github.com/allure-framework/allure2/releases/download/${ALLURE_VERSION}/${ALLURE_ZIP}
-          echo "${ALLURE_ZIP_SHA256} ${ALLURE_ZIP}" | sha256sum --check
+          echo "${ALLURE_ZIP_MD5}  ${ALLURE_ZIP}" | md5sum -c
           unzip -q ${ALLURE_ZIP}
           echo "$(pwd)/allure-${ALLURE_VERSION}/bin" >> $GITHUB_PATH
           rm -f ${ALLURE_ZIP}
         fi
       env:
-        ALLURE_VERSION: 2.22.1
-        ALLURE_ZIP_SHA256: fdc7a62d94b14c5e0bf25198ae1feded6b005fdbed864b4d3cb4e5e901720b0b
+        ALLURE_VERSION: 2.22.0
+        ALLURE_ZIP_MD5: d5c9f0989b896482536956340a7d5ec9
 
     # Potentially we could have several running build for the same key (for example, for the main branch), so we use improvised lock for this
     - name: Acquire lock

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

@@ -71,12 +71,12 @@ runs:
         path: /tmp/neon-previous
         prefix: latest
 
-    - name: Download compatibility snapshot
-      if: inputs.build_type != 'remote'
+    - name: Download compatibility snapshot for Postgres 14
+      if: inputs.build_type != 'remote' && inputs.pg_version == 'v14'
       uses: ./.github/actions/download
       with:
-        name: compatibility-snapshot-${{ inputs.build_type }}-pg${{ inputs.pg_version }}
-        path: /tmp/compatibility_snapshot_pg${{ inputs.pg_version }}
+        name: compatibility-snapshot-${{ inputs.build_type }}-pg14
+        path: /tmp/compatibility_snapshot_pg14
         prefix: latest
 
     - name: Checkout
@@ -106,7 +106,7 @@ runs:
         BUILD_TYPE: ${{ inputs.build_type }}
         AWS_ACCESS_KEY_ID: ${{ inputs.real_s3_access_key_id }}
         AWS_SECRET_ACCESS_KEY: ${{ inputs.real_s3_secret_access_key }}
-        COMPATIBILITY_SNAPSHOT_DIR: /tmp/compatibility_snapshot_pg${{ inputs.pg_version }}
+        COMPATIBILITY_SNAPSHOT_DIR: /tmp/compatibility_snapshot_pg14
         ALLOW_BACKWARD_COMPATIBILITY_BREAKAGE: contains(github.event.pull_request.labels.*.name, 'backward compatibility breakage')
         ALLOW_FORWARD_COMPATIBILITY_BREAKAGE: contains(github.event.pull_request.labels.*.name, 'forward compatibility breakage')
         RERUN_FLAKY: ${{ inputs.rerun_flaky }}
@@ -197,13 +197,13 @@ runs:
           scripts/generate_and_push_perf_report.sh
         fi
 
-    - name: Upload compatibility snapshot
-      if: github.ref_name == 'release'
+    - name: Upload compatibility snapshot for Postgres 14
+      if: github.ref_name == 'release' && inputs.pg_version == 'v14'
       uses: ./.github/actions/upload
       with:
-        name: compatibility-snapshot-${{ inputs.build_type }}-pg${{ inputs.pg_version }}-${{ github.run_id }}
+        name: compatibility-snapshot-${{ inputs.build_type }}-pg14-${{ github.run_id }}
         # Directory is created by test_compatibility.py::test_create_snapshot, keep the path in sync with the test
-        path: /tmp/test_output/compatibility_snapshot_pg${{ inputs.pg_version }}/
+        path: /tmp/test_output/compatibility_snapshot_pg14/
         prefix: latest
 
     - name: Upload test results

.github/workflows/build_and_test.yml

@@ -711,11 +711,7 @@ jobs:
 
   compute-node-image:
     runs-on: [ self-hosted, gen3, large ]
-    container:
-      image: gcr.io/kaniko-project/executor:v1.9.2-debug
-      # Workaround for "Resolving download.osgeo.org (download.osgeo.org)... failed: Temporary failure in name resolution.""
-      # Should be prevented by https://github.com/neondatabase/neon/issues/4281
-      options: --add-host=download.osgeo.org:140.211.15.30
+    container: gcr.io/kaniko-project/executor:v1.9.2-debug
     needs: [ tag ]
     strategy:
       fail-fast: false
@@ -777,7 +773,7 @@ jobs:
       run:
         shell: sh -eu {0}
     env:
-      VM_BUILDER_VERSION: v0.7.3-alpha3
+      VM_BUILDER_VERSION: v0.4.6
 
     steps:
       - name: Checkout
@@ -787,18 +783,21 @@ jobs:
 
       - name: Downloading vm-builder
         run: |
-          curl -fL https://github.com/neondatabase/autoscaling/releases/download/$VM_BUILDER_VERSION/vm-builder -o vm-builder
+          curl -L https://github.com/neondatabase/neonvm/releases/download/$VM_BUILDER_VERSION/vm-builder -o vm-builder
           chmod +x vm-builder
 
-      # Note: we need a separate pull step here because otherwise vm-builder will try to pull, and
-      # it won't have the proper authentication (written at v0.6.0)
       - name: Pulling compute-node image
         run: |
           docker pull 369495373322.dkr.ecr.eu-central-1.amazonaws.com/compute-node-${{ matrix.version }}:${{needs.tag.outputs.build-tag}}
 
+      - name: Building VM compute-node rootfs
+        run: |
+          docker build -t temp-vm-compute-node --build-arg SRC_IMAGE=369495373322.dkr.ecr.eu-central-1.amazonaws.com/compute-node-${{ matrix.version }}:${{needs.tag.outputs.build-tag}} -f Dockerfile.vm-compute-node .
+
       - name: Build vm image
         run: |
-          ./vm-builder -src=369495373322.dkr.ecr.eu-central-1.amazonaws.com/compute-node-${{ matrix.version }}:${{needs.tag.outputs.build-tag}} -dst=369495373322.dkr.ecr.eu-central-1.amazonaws.com/vm-compute-node-${{ matrix.version }}:${{needs.tag.outputs.build-tag}}
+          # note: as of 2023-01-12, vm-builder requires a trailing ":latest" for local images
+          ./vm-builder -use-inittab -src=temp-vm-compute-node:latest -dst=369495373322.dkr.ecr.eu-central-1.amazonaws.com/vm-compute-node-${{ matrix.version }}:${{needs.tag.outputs.build-tag}}
 
       - name: Pushing vm-compute-node image
         run: |
@@ -958,7 +957,7 @@ jobs:
   promote-compatibility-data:
     runs-on: [ self-hosted, gen3, small ]
     container:
-      image: 369495373322.dkr.ecr.eu-central-1.amazonaws.com/base:pinned
+      image: 369495373322.dkr.ecr.eu-central-1.amazonaws.com/rust:pinned
       options: --init
     needs: [ promote-images, tag, regress-tests ]
     if: github.ref_name == 'release' && github.event_name != 'workflow_dispatch'
@@ -969,13 +968,11 @@ jobs:
           PREFIX: artifacts/latest
         run: |
           # Update compatibility snapshot for the release
-          for pg_version in v14 v15; do
-            for build_type in debug release; do
-              OLD_FILENAME=compatibility-snapshot-${build_type}-pg${pg_version}-${GITHUB_RUN_ID}.tar.zst
-              NEW_FILENAME=compatibility-snapshot-${build_type}-pg${pg_version}.tar.zst
+          for build_type in debug release; do
+            OLD_FILENAME=compatibility-snapshot-${build_type}-pg14-${GITHUB_RUN_ID}.tar.zst
+            NEW_FILENAME=compatibility-snapshot-${build_type}-pg14.tar.zst
 
-              time aws s3 mv --only-show-errors s3://${BUCKET}/${PREFIX}/${OLD_FILENAME} s3://${BUCKET}/${PREFIX}/${NEW_FILENAME}
-            done
+            time aws s3 mv --only-show-errors s3://${BUCKET}/${PREFIX}/${OLD_FILENAME} s3://${BUCKET}/${PREFIX}/${NEW_FILENAME}
           done
 
           # Update Neon artifact for the release (reuse already uploaded artifact)

Cargo.lock

@@ -2820,7 +2820,7 @@ dependencies = [
 [[package]]
 name = "postgres"
 version = "0.19.4"
-source = "git+https://github.com/neondatabase/rust-postgres.git?rev=2e9b5f1ddc481d1a98fa79f6b9378ac4f170b7c9#2e9b5f1ddc481d1a98fa79f6b9378ac4f170b7c9"
+source = "git+https://github.com/neondatabase/rust-postgres.git?rev=0bc41d8503c092b040142214aac3cf7d11d0c19f#0bc41d8503c092b040142214aac3cf7d11d0c19f"
 dependencies = [
  "bytes",
  "fallible-iterator",
@@ -2833,7 +2833,7 @@ dependencies = [
 [[package]]
 name = "postgres-native-tls"
 version = "0.5.0"
-source = "git+https://github.com/neondatabase/rust-postgres.git?rev=2e9b5f1ddc481d1a98fa79f6b9378ac4f170b7c9#2e9b5f1ddc481d1a98fa79f6b9378ac4f170b7c9"
+source = "git+https://github.com/neondatabase/rust-postgres.git?rev=0bc41d8503c092b040142214aac3cf7d11d0c19f#0bc41d8503c092b040142214aac3cf7d11d0c19f"
 dependencies = [
  "native-tls",
  "tokio",
@@ -2844,7 +2844,7 @@ dependencies = [
 [[package]]
 name = "postgres-protocol"
 version = "0.6.4"
-source = "git+https://github.com/neondatabase/rust-postgres.git?rev=2e9b5f1ddc481d1a98fa79f6b9378ac4f170b7c9#2e9b5f1ddc481d1a98fa79f6b9378ac4f170b7c9"
+source = "git+https://github.com/neondatabase/rust-postgres.git?rev=0bc41d8503c092b040142214aac3cf7d11d0c19f#0bc41d8503c092b040142214aac3cf7d11d0c19f"
 dependencies = [
  "base64 0.20.0",
  "byteorder",
@@ -2862,7 +2862,7 @@ dependencies = [
 [[package]]
 name = "postgres-types"
 version = "0.2.4"
-source = "git+https://github.com/neondatabase/rust-postgres.git?rev=2e9b5f1ddc481d1a98fa79f6b9378ac4f170b7c9#2e9b5f1ddc481d1a98fa79f6b9378ac4f170b7c9"
+source = "git+https://github.com/neondatabase/rust-postgres.git?rev=0bc41d8503c092b040142214aac3cf7d11d0c19f#0bc41d8503c092b040142214aac3cf7d11d0c19f"
 dependencies = [
  "bytes",
  "fallible-iterator",
@@ -4271,9 +4271,9 @@ dependencies = [
 
 [[package]]
 name = "tokio"
-version = "1.28.1"
+version = "1.27.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0aa32867d44e6f2ce3385e89dceb990188b8bb0fb25b0cf576647a6f98ac5105"
+checksum = "d0de47a4eecbe11f498978a9b29d792f0d2692d1dd003650c24c76510e3bc001"
 dependencies = [
  "autocfg",
  "bytes",
@@ -4284,7 +4284,7 @@ dependencies = [
  "signal-hook-registry",
  "socket2 0.4.9",
  "tokio-macros",
- "windows-sys 0.48.0",
+ "windows-sys 0.45.0",
 ]
 
 [[package]]
@@ -4299,9 +4299,9 @@ dependencies = [
 
 [[package]]
 name = "tokio-macros"
-version = "2.1.0"
+version = "2.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "630bdcf245f78637c13ec01ffae6187cca34625e8c63150d424b59e55af2675e"
+checksum = "61a573bdc87985e9d6ddeed1b3d864e8a302c847e40d647746df2f1de209d1ce"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -4321,7 +4321,7 @@ dependencies = [
 [[package]]
 name = "tokio-postgres"
 version = "0.7.7"
-source = "git+https://github.com/neondatabase/rust-postgres.git?rev=2e9b5f1ddc481d1a98fa79f6b9378ac4f170b7c9#2e9b5f1ddc481d1a98fa79f6b9378ac4f170b7c9"
+source = "git+https://github.com/neondatabase/rust-postgres.git?rev=0bc41d8503c092b040142214aac3cf7d11d0c19f#0bc41d8503c092b040142214aac3cf7d11d0c19f"
 dependencies = [
  "async-trait",
  "byteorder",

Cargo.toml

@@ -126,11 +126,11 @@ env_logger = "0.10"
 log = "0.4"
 
 ## Libraries from neondatabase/ git forks, ideally with changes to be upstreamed
-postgres = { git = "https://github.com/neondatabase/rust-postgres.git", rev="2e9b5f1ddc481d1a98fa79f6b9378ac4f170b7c9" }
-postgres-native-tls = { git = "https://github.com/neondatabase/rust-postgres.git", rev="2e9b5f1ddc481d1a98fa79f6b9378ac4f170b7c9" }
-postgres-protocol = { git = "https://github.com/neondatabase/rust-postgres.git", rev="2e9b5f1ddc481d1a98fa79f6b9378ac4f170b7c9" }
-postgres-types = { git = "https://github.com/neondatabase/rust-postgres.git", rev="2e9b5f1ddc481d1a98fa79f6b9378ac4f170b7c9" }
-tokio-postgres = { git = "https://github.com/neondatabase/rust-postgres.git", rev="2e9b5f1ddc481d1a98fa79f6b9378ac4f170b7c9" }
+postgres = { git = "https://github.com/neondatabase/rust-postgres.git", rev="0bc41d8503c092b040142214aac3cf7d11d0c19f" }
+postgres-native-tls = { git = "https://github.com/neondatabase/rust-postgres.git", rev="0bc41d8503c092b040142214aac3cf7d11d0c19f" }
+postgres-protocol = { git = "https://github.com/neondatabase/rust-postgres.git", rev="0bc41d8503c092b040142214aac3cf7d11d0c19f" }
+postgres-types = { git = "https://github.com/neondatabase/rust-postgres.git", rev="0bc41d8503c092b040142214aac3cf7d11d0c19f" }
+tokio-postgres = { git = "https://github.com/neondatabase/rust-postgres.git", rev="0bc41d8503c092b040142214aac3cf7d11d0c19f" }
 tokio-tar = { git = "https://github.com/neondatabase/tokio-tar.git", rev="404df61437de0feef49ba2ccdbdd94eb8ad6e142" }
 
 ## Other git libraries
@@ -166,7 +166,7 @@ tonic-build = "0.9"
 
 # This is only needed for proxy's tests.
 # TODO: we should probably fork `tokio-postgres-rustls` instead.
-tokio-postgres = { git = "https://github.com/neondatabase/rust-postgres.git", rev="2e9b5f1ddc481d1a98fa79f6b9378ac4f170b7c9" }
+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,

Dockerfile.vm-compute-node

@@ -0,0 +1,70 @@
+# Note: this file *mostly* just builds on Dockerfile.compute-node
+
+ARG SRC_IMAGE
+ARG VM_INFORMANT_VERSION=v0.1.14
+# on libcgroup update, make sure to check bootstrap.sh for changes
+ARG LIBCGROUP_VERSION=v2.0.3
+
+# Pull VM informant, to copy from later
+FROM neondatabase/vm-informant:$VM_INFORMANT_VERSION as informant
+
+# Build cgroup-tools
+#
+# At time of writing (2023-03-14), debian bullseye has a version of cgroup-tools (technically
+# libcgroup) that doesn't support cgroup v2 (version 0.41-11). Unfortunately, the vm-informant
+# requires cgroup v2, so we'll build cgroup-tools ourselves.
+FROM debian:bullseye-slim as libcgroup-builder
+ARG LIBCGROUP_VERSION
+
+RUN set -exu \
+	&& apt update \
+	&& apt install --no-install-recommends -y \
+		git \
+		ca-certificates \
+		automake \
+		cmake \
+		make \
+		gcc \
+		byacc \
+		flex \
+		libtool \
+		libpam0g-dev \
+	&& git clone --depth 1 -b $LIBCGROUP_VERSION https://github.com/libcgroup/libcgroup \
+	&& INSTALL_DIR="/libcgroup-install" \
+	&& mkdir -p "$INSTALL_DIR/bin" "$INSTALL_DIR/include" \
+	&& cd libcgroup \
+	# extracted from bootstrap.sh, with modified flags:
+	&& (test -d m4 || mkdir m4) \
+	&& autoreconf -fi \
+	&& rm -rf autom4te.cache \
+	&& CFLAGS="-O3" ./configure --prefix="$INSTALL_DIR" --sysconfdir=/etc --localstatedir=/var --enable-opaque-hierarchy="name=systemd" \
+	# actually build the thing...
+	&& make install
+
+# Combine, starting from non-VM compute node image.
+FROM $SRC_IMAGE as base
+
+# Temporarily set user back to root so we can run adduser, set inittab
+USER root
+RUN adduser vm-informant --disabled-password --no-create-home
+
+RUN set -e \
+	&& rm -f /etc/inittab \
+	&& touch /etc/inittab
+
+RUN set -e \
+	&& echo "::sysinit:cgconfigparser -l /etc/cgconfig.conf -s 1664" >> /etc/inittab \
+	&& CONNSTR="dbname=postgres user=cloud_admin sslmode=disable" \
+	&& ARGS="--auto-restart --cgroup=neon-postgres --pgconnstr=\"$CONNSTR\"" \
+	&& echo "::respawn:su vm-informant -c '/usr/local/bin/vm-informant $ARGS'" >> /etc/inittab
+
+USER postgres
+
+ADD vm-cgconfig.conf /etc/cgconfig.conf
+COPY --from=informant /usr/bin/vm-informant /usr/local/bin/vm-informant
+
+COPY --from=libcgroup-builder /libcgroup-install/bin/* /usr/bin/
+COPY --from=libcgroup-builder /libcgroup-install/lib/* /usr/lib/
+COPY --from=libcgroup-builder /libcgroup-install/sbin/* /usr/sbin/
+
+ENTRYPOINT ["/usr/sbin/cgexec", "-g", "*:neon-postgres", "/usr/local/bin/compute_ctl"]

control_plane/src/pageserver.rs

@@ -393,79 +393,69 @@ impl PageServerNode {
             })
     }
 
-    pub fn tenant_config(
-        &self,
-        tenant_id: TenantId,
-        mut settings: HashMap<&str, &str>,
-    ) -> anyhow::Result<()> {
+    pub fn tenant_config(&self, tenant_id: TenantId, settings: HashMap<&str, &str>) -> Result<()> {
         let config = {
             // Braces to make the diff easier to read
             models::TenantConfig {
                 checkpoint_distance: settings
-                    .remove("checkpoint_distance")
+                    .get("checkpoint_distance")
                     .map(|x| x.parse::<u64>())
                     .transpose()
                     .context("Failed to parse 'checkpoint_distance' as an integer")?,
-                checkpoint_timeout: settings.remove("checkpoint_timeout").map(|x| x.to_string()),
+                checkpoint_timeout: settings.get("checkpoint_timeout").map(|x| x.to_string()),
                 compaction_target_size: settings
-                    .remove("compaction_target_size")
+                    .get("compaction_target_size")
                     .map(|x| x.parse::<u64>())
                     .transpose()
                     .context("Failed to parse 'compaction_target_size' as an integer")?,
-                compaction_period: settings.remove("compaction_period").map(|x| x.to_string()),
+                compaction_period: settings.get("compaction_period").map(|x| x.to_string()),
                 compaction_threshold: settings
-                    .remove("compaction_threshold")
+                    .get("compaction_threshold")
                     .map(|x| x.parse::<usize>())
                     .transpose()
                     .context("Failed to parse 'compaction_threshold' as an integer")?,
                 gc_horizon: settings
-                    .remove("gc_horizon")
+                    .get("gc_horizon")
                     .map(|x| x.parse::<u64>())
                     .transpose()
                     .context("Failed to parse 'gc_horizon' as an integer")?,
-                gc_period: settings.remove("gc_period").map(|x| x.to_string()),
+                gc_period: settings.get("gc_period").map(|x| x.to_string()),
                 image_creation_threshold: settings
-                    .remove("image_creation_threshold")
+                    .get("image_creation_threshold")
                     .map(|x| x.parse::<usize>())
                     .transpose()
                     .context("Failed to parse 'image_creation_threshold' as non zero integer")?,
-                pitr_interval: settings.remove("pitr_interval").map(|x| x.to_string()),
+                pitr_interval: settings.get("pitr_interval").map(|x| x.to_string()),
                 walreceiver_connect_timeout: settings
-                    .remove("walreceiver_connect_timeout")
-                    .map(|x| x.to_string()),
-                lagging_wal_timeout: settings
-                    .remove("lagging_wal_timeout")
+                    .get("walreceiver_connect_timeout")
                     .map(|x| x.to_string()),
+                lagging_wal_timeout: settings.get("lagging_wal_timeout").map(|x| x.to_string()),
                 max_lsn_wal_lag: settings
-                    .remove("max_lsn_wal_lag")
+                    .get("max_lsn_wal_lag")
                     .map(|x| x.parse::<NonZeroU64>())
                     .transpose()
                     .context("Failed to parse 'max_lsn_wal_lag' as non zero integer")?,
                 trace_read_requests: settings
-                    .remove("trace_read_requests")
+                    .get("trace_read_requests")
                     .map(|x| x.parse::<bool>())
                     .transpose()
                     .context("Failed to parse 'trace_read_requests' as bool")?,
                 eviction_policy: settings
-                    .remove("eviction_policy")
-                    .map(serde_json::from_str)
+                    .get("eviction_policy")
+                    .map(|x| serde_json::from_str(x))
                     .transpose()
                     .context("Failed to parse 'eviction_policy' json")?,
                 min_resident_size_override: settings
-                    .remove("min_resident_size_override")
+                    .get("min_resident_size_override")
                     .map(|x| x.parse::<u64>())
                     .transpose()
                     .context("Failed to parse 'min_resident_size_override' as an integer")?,
                 evictions_low_residence_duration_metric_threshold: settings
-                    .remove("evictions_low_residence_duration_metric_threshold")
+                    .get("evictions_low_residence_duration_metric_threshold")
                     .map(|x| x.to_string()),
             }
         };
 
-        if !settings.is_empty() {
-            bail!("Unrecognized tenant settings: {settings:?}")
-        }
-
         self.http_request(Method::PUT, format!("{}/tenant/config", self.http_base_url))?
             .json(&models::TenantConfigRequest { tenant_id, config })
             .send()?

docs/rfcs/023-the-state-of-pageserver-tenant-relocation.md

@@ -1,232 +0,0 @@
-# The state of pageserver tenant relocation
-
-Created on 17.03.23
-
-## Motivation
-
-There were previous write ups on the subject. The design of tenant relocation was planned at the time when we had quite different landscape. I e there was no on-demand download/eviction. They were on the horizon but we still planned for cases when they were not available. Some other things have changed. Now safekeepers offload wal to s3 so we're not risking overflowing their disks. Having all of the above, it makes sense to recap and take a look at the options we have now, which adjustments we'd like to make to original process, etc.
-
-Related (in chronological order):
-
-- Tracking issue with initial discussion: [#886](https://github.com/neondatabase/neon/issues/886)
-- [015. Storage Messaging](015-storage-messaging.md)
-- [020. Pageserver S3 Coordination](020-pageserver-s3-coordination.md)
-
-## Summary
-
-The RFC consists of a walkthrough of prior art on tenant relocation and corresponding problems. It describes 3 approaches.
-
-1. Simplistic approach that uses ignore and is the fastest to implement. The main downside is a requirement of short downtime.
-2. More complicated approach that avoids even short downtime.
-3. Even more complicated approach that will allow multiple pageservers to operate concurrently on the same tenant possibly allowing for HA cluster topologies and horizontal scaling of reads (i e compute talks to multiple pageservers).
-
-The order in which solutions are described is a bit different. We start from 2, then move to possible compromises (aka simplistic approach) and then move to discussing directions for solving HA/Pageserver replica case with 3.
-
-## Components
-
-pageserver, control-plane, safekeepers (a bit)
-
-## Requirements
-
-Relocation procedure should move tenant from one pageserver to another without downtime introduced by storage side. For now restarting compute for applying new configuration is fine.
-
-- component restarts
-- component outage
-- pageserver loss
-
-## The original proposed implementation
-
-The starting point is this sequence:
-
-```mermaid
-sequenceDiagram
-    autonumber
-    participant CP as Control Plane
-    participant PS1 as Pageserver 1
-    participant PS2 as Pageserver 2
-    participant S3
-
-    CP->>PS2: Attach tenant X
-    PS2->>S3: Fetch timelines, indexes for them
-    PS2->>CP: Accepted
-    CP->>CP: Change pageserver id in project
-    CP->>PS1: Detach
-```
-
-Which problems do we have with naive approach?
-
-### Concurrent GC and Compaction
-
-The problem is that they can run on both, PS1 and PS2. Consider this example from [Pageserver S3 Coordination RFC](020-pageserver-s3-coordination.md)
-
-```mermaid
-sequenceDiagram
-    autonumber
-    participant PS1
-    participant S3
-    participant PS2
-
-    PS1->>S3: Uploads L1, L2 <br/> Index contains L1 L2
-    PS2->>S3: Attach called, sees L1, L2
-    PS1->>S3: Compaction comes <br/> Removes L1, adds L3
-    note over S3: Index now L2, L3
-    PS2->>S3: Uploads new layer L4 <br/> (added to previous view of the index)
-    note over S3: Index now L1, L2, L4
-```
-
-At this point it is not possible to restore the state from index, it contains L2 which
-is no longer available in s3 and doesnt contain L3 added by compaction by the
-first pageserver. So if any of the pageservers restart, initial sync will fail
-(or in on-demand world it will fail a bit later during page request from
-missing layer)
-
-The problem lies in shared index_part.json. Having intersecting layers from append only edits is expected to work, though this is an uncharted territory without tests.
-
-#### Options
-
-There are several options on how to restrict concurrent access to index file.
-
-First and the simplest one is external orchestration. Control plane which runs migration can use special api call on pageserver to stop background processes (gc, compaction), and even possibly all uploads.
-
-So the sequence becomes:
-
-```mermaid
-sequenceDiagram
-    autonumber
-    participant CP as Control Plane
-    participant PS1 as Pageserver 1
-    participant PS2 as Pageserver 2
-    participant S3
-
-    CP->>PS1: Pause background jobs, pause uploading new layers.
-    CP->>PS2: Attach tenant X.
-    PS2->>S3: Fetch timelines, index, start background operations
-    PS2->>CP: Accepted
-    CP->>CP: Monitor PS2 last record lsn, ensure OK lag
-    CP->>CP: Change pageserver id in project
-    CP->>PS1: Detach
-```
-
-The downside of this sequence is the potential rollback process. What if something goes wrong on new pageserver? Can we safely roll back to source pageserver?
-
-There are two questions:
-
-#### How can we detect that something went wrong?
-
-We can run usual availability check (consists of compute startup and an update of one row).
-Note that we cant run separate compute for that before touching compute that client runs actual workload on, because we cant have two simultaneous computes running in read-write mode on the same timeline (enforced by safekeepers consensus algorithm). So we can either run some readonly check first (basebackup) and then change pageserver id and run availability check. If it failed we can roll it back to the old one.
-
-#### What can go wrong? And how we can safely roll-back?
-
-In the sequence above during attach we start background processes/uploads. They change state in remote storage so it is possible that after rollback remote state will be different from one that was observed by source pageserver. So if target pageserver goes wild then source pageserver may fail to start with changed remote state.
-
-Proposed option would be to implement a barrier (read-only) mode when pageserver does not update remote state.
-
-So the sequence for happy path becomes this one:
-
-```mermaid
-sequenceDiagram
-    autonumber
-    participant CP as Control Plane
-    participant PS1 as Pageserver 1
-    participant PS2 as Pageserver 2
-    participant S3
-
-    CP->>PS1: Pause background jobs, pause uploading new layers.
-    CP->>PS2: Attach tenant X in remote readonly mode.
-    PS2->>S3: Fetch timelines, index
-    PS2->>CP: Accepted
-    CP->>CP: Monitor PS2 last record lsn, ensure OK lag
-    CP->>CP: Change pageserver id in project
-    CP->>CP: Run successful availability check
-    CP->>PS2: Start uploads, background tasks
-    CP->>PS1: Detach
-```
-
-With this sequence we restrict any changes to remote storage to one pageserver. So there is no concurrent access at all, not only for index_part.json, but for everything else too. This approach makes it possible to roll back after failure on new pageserver.
-
-The sequence with roll back process:
-
-```mermaid
-sequenceDiagram
-    autonumber
-    participant CP as Control Plane
-    participant PS1 as Pageserver 1
-    participant PS2 as Pageserver 2
-    participant S3
-
-    CP->>PS1: Pause background jobs, pause uploading new layers.
-    CP->>PS2: Attach tenant X in remote readonly mode.
-    PS2->>S3: Fetch timelines, index
-    PS2->>CP: Accepted
-    CP->>CP: Monitor PS2 last record lsn, ensure OK lag
-    CP->>CP: Change pageserver id in project
-    CP->>CP: Availability check Failed
-    CP->>CP: Change pageserver id back
-    CP->>PS1: Resume remote operations
-    CP->>PS2: Ignore (instead of detach for investigation purposes)
-```
-
-## Concurrent branch creation
-
-Another problem is a possibility of concurrent branch creation calls.
-
-I e during migration create_branch can be called on old pageserver and newly created branch wont be seen on new pageserver. Prior art includes prototyping an approach of trying to mirror such branches, but currently it lost its importance, because now attach is fast because we dont need to download all data, and additionally to the best of my knowledge of control plane internals (cc @ololobus to confirm) operations on one project are executed sequentially, so it is not possible to have such case. So branch create operation will be executed only when relocation is completed. As a safety measure we can forbid branch creation for tenants that are in readonly remote state.
-
-## Simplistic approach
-
-The difference of simplistic approach from one described above is that it calls ignore on source tenant first and then calls attach on target pageserver. Approach above does it in opposite order thus opening a possibility for race conditions we strive to avoid.
-
-The approach largely follows this guide: <https://github.com/neondatabase/cloud/wiki/Cloud:-Ad-hoc-tenant-relocation>
-
-The happy path sequence:
-
-```mermaid
-sequenceDiagram
-    autonumber
-    participant CP as Control Plane
-    participant PS1 as Pageserver 1
-    participant PS2 as Pageserver 2
-    participant SK as Safekeeper
-    participant S3
-
-    CP->>CP: Enable maintenance mode
-    CP->>PS1: Ignore
-    CP->>PS2: Attach
-    PS2->>CP: Accepted
-    loop Delete layers for each timeline
-        CP->>PS2: Get last record lsn
-        CP->>SK: Get commit lsn
-        CP->>CP: OK? Timed out?
-    end
-    CP->>CP: Change pageserver id in project
-    CP->>CP: Run successful availability check
-    CP->>CP: Disable maintenance mode
-    CP->>PS1: Detach ignored
-```
-
-The sequence contains exactly the same rollback problems as in previous approach described above. They can be resolved the same way.
-
-Most probably we'd like to move forward without this safety measure and implement it on top of this approach to make progress towards the downtime-less one.
-
-## Lease based approach
-
-In order to allow for concurrent operation on the same data on remote storage for multiple pageservers we need to go further than external orchestration.
-
-NOTE: [020. Pageserver S3 Coordination](020-pageserver-s3-coordination.md) discusses one more approach that relies on duplication of index_part.json for each pageserver operating on the timeline. This approach still requires external coordination which makes certain things easier but requires additional bookkeeping to account for multiple index_part.json files. Discussion/comparison with proposed lease based approach
-
-The problems are outlined in [020. Pageserver S3 Coordination](020-pageserver-s3-coordination.md) and suggested solution includes [Coordination based approach](020-pageserver-s3-coordination.md#coordination-based-approach). This way it will allow to do basic leader election for pageservers so they can decide which node will be responsible for running GC and compaction. The process is based on extensive communication via storage broker and consists of a lease that is taken by one of the pageservers that extends it to continue serving a leader role.
-
-There are two options for ingesting new data into pageserver in follower role. One option is to avoid WAL ingestion at all and rely on notifications from leader to discover new layers on s3. Main downside of this approach is that follower will always lag behind the primary node because it wont have the last layer until it is uploaded to remote storage. In case of a primary failure follower will be required to reingest last segment (up to 256Mb of WAL currently) which slows down recovery. Additionally if compute is connected to follower pageserver it will observe latest data with a delay. Queries from compute will likely experience bigger delays when recent lsn is required.
-
-The second option is to consume WAL stream on both pageservers. In this case the only problem is non deterministic layer generation. Additional bookkeeping will be required to deduplicate layers from primary with local ones. Some process needs to somehow merge them to remove duplicated data. Additionally we need to have good testing coverage to ensure that our implementation of `get_page@lsn` properly handles intersecting layers.
-
-There is another tradeoff. Approaches may be different in amount of traffic between system components. With first approach there can be increased traffic between follower and remote storage. But only in case follower has some activity that actually requests pages (!). With other approach traffic increase will be permanent and will be caused by two WAL streams instead of one.
-
-## Summary
-
-Proposed implementation strategy:
-
-Go with the simplest approach for now. Then work on tech debt, increase test coverage. Then gradually move forward to second approach by implementing safety measures first, finishing with switch of order between ignore and attach operation.
-
-And only then go to lease based approach to solve HA/Pageserver replica use cases.

libs/pageserver_api/src/models.rs

@@ -37,8 +37,6 @@ pub enum TenantState {
     Loading,
     /// This tenant is being downloaded from cloud storage.
     Attaching,
-    /// The tenant is transitioning from Loading/Attaching to Active.
-    Activating,
     /// Tenant is fully operational
     Active,
     /// A tenant is recognized by pageserver, but it is being detached or the
@@ -62,7 +60,6 @@ impl TenantState {
             // tenant mgr startup distinguishes attaching from loading via marker file.
             // If it's loading, there is no attach marker file, i.e., attach had finished in the past.
             Self::Loading => Attached,
-            Self::Activating => todo!(),
             // We only reach Active after successful load / attach.
             // So, call atttachment status Attached.
             Self::Active => Attached,
@@ -134,14 +131,13 @@ pub struct TimelineCreateRequest {
 }
 
 #[serde_as]
-#[derive(Serialize, Deserialize, Debug, Default)]
-#[serde(deny_unknown_fields)]
+#[derive(Serialize, Deserialize, Default)]
 pub struct TenantCreateRequest {
     #[serde(default)]
     #[serde_as(as = "Option<DisplayFromStr>")]
     pub new_tenant_id: Option<TenantId>,
     #[serde(flatten)]
-    pub config: TenantConfig, // as we have a flattened field, we should reject all unknown fields in it
+    pub config: TenantConfig,
 }
 
 impl std::ops::Deref for TenantCreateRequest {
@@ -152,7 +148,7 @@ impl std::ops::Deref for TenantCreateRequest {
     }
 }
 
-#[derive(Serialize, Deserialize, Debug, Default)]
+#[derive(Serialize, Deserialize, Default)]
 pub struct TenantConfig {
     pub checkpoint_distance: Option<u64>,
     pub checkpoint_timeout: Option<String>,
@@ -196,13 +192,12 @@ impl TenantCreateRequest {
 }
 
 #[serde_as]
-#[derive(Serialize, Deserialize, Debug)]
-#[serde(deny_unknown_fields)]
+#[derive(Serialize, Deserialize)]
 pub struct TenantConfigRequest {
     #[serde_as(as = "DisplayFromStr")]
     pub tenant_id: TenantId,
     #[serde(flatten)]
-    pub config: TenantConfig, // as we have a flattened field, we should reject all unknown fields in it
+    pub config: TenantConfig,
 }
 
 impl std::ops::Deref for TenantConfigRequest {
@@ -237,28 +232,6 @@ impl TenantConfigRequest {
     }
 }
 
-#[derive(Debug, Serialize, Deserialize)]
-pub struct TenantAttachRequest {
-    pub config: TenantAttachConfig,
-}
-
-/// Newtype to enforce deny_unknown_fields on TenantConfig for
-/// its usage inside `TenantAttachRequest`.
-#[derive(Debug, Serialize, Deserialize)]
-#[serde(deny_unknown_fields)]
-pub struct TenantAttachConfig {
-    #[serde(flatten)]
-    allowing_unknown_fields: TenantConfig,
-}
-
-impl std::ops::Deref for TenantAttachConfig {
-    type Target = TenantConfig;
-
-    fn deref(&self) -> &Self::Target {
-        &self.allowing_unknown_fields
-    }
-}
-
 /// See [`TenantState::attachment_status`] and the OpenAPI docs for context.
 #[derive(Serialize, Deserialize, Clone)]
 #[serde(rename_all = "snake_case")]
@@ -795,43 +768,4 @@ mod tests {
         assert!(format!("{:?}", &original_broken.state).contains("reason"));
         assert!(format!("{:?}", &original_broken.state).contains("backtrace info"));
     }
-
-    #[test]
-    fn test_reject_unknown_field() {
-        let id = TenantId::generate();
-        let create_request = json!({
-            "new_tenant_id": id.to_string(),
-            "unknown_field": "unknown_value".to_string(),
-        });
-        let err = serde_json::from_value::<TenantCreateRequest>(create_request).unwrap_err();
-        assert!(
-            err.to_string().contains("unknown field `unknown_field`"),
-            "expect unknown field `unknown_field` error, got: {}",
-            err
-        );
-
-        let id = TenantId::generate();
-        let config_request = json!({
-            "tenant_id": id.to_string(),
-            "unknown_field": "unknown_value".to_string(),
-        });
-        let err = serde_json::from_value::<TenantConfigRequest>(config_request).unwrap_err();
-        assert!(
-            err.to_string().contains("unknown field `unknown_field`"),
-            "expect unknown field `unknown_field` error, got: {}",
-            err
-        );
-
-        let attach_request = json!({
-            "config": {
-                "unknown_field": "unknown_value".to_string(),
-            },
-        });
-        let err = serde_json::from_value::<TenantAttachRequest>(attach_request).unwrap_err();
-        assert!(
-            err.to_string().contains("unknown field `unknown_field`"),
-            "expect unknown field `unknown_field` error, got: {}",
-            err
-        );
-    }
 }

libs/utils/src/http/json.rs

@@ -8,26 +8,12 @@ use super::error::ApiError;
 pub async fn json_request<T: for<'de> Deserialize<'de>>(
     request: &mut Request<Body>,
 ) -> Result<T, ApiError> {
-    json_request_or_empty_body(request)
-        .await?
-        .context("missing request body")
-        .map_err(ApiError::BadRequest)
-}
-
-/// Will be removed as part of https://github.com/neondatabase/neon/issues/4282
-pub async fn json_request_or_empty_body<T: for<'de> Deserialize<'de>>(
-    request: &mut Request<Body>,
-) -> Result<Option<T>, ApiError> {
-    let body = hyper::body::aggregate(request.body_mut())
+    let whole_body = hyper::body::aggregate(request.body_mut())
         .await
         .context("Failed to read request body")
         .map_err(ApiError::BadRequest)?;
-    if body.remaining() == 0 {
-        return Ok(None);
-    }
-    serde_json::from_reader(body.reader())
+    serde_json::from_reader(whole_body.reader())
         .context("Failed to parse json request")
-        .map(Some)
         .map_err(ApiError::BadRequest)
 }
 

pageserver/Cargo.toml

@@ -88,3 +88,7 @@ harness = false
 [[bench]]
 name = "bench_walredo"
 harness = false
+
+[[bench]]
+name = "bench_disk_lookup"
+harness = false

pageserver/benches/bench_disk_lookup.rs

@@ -0,0 +1,133 @@
+use criterion::{black_box, criterion_group, criterion_main, Criterion};
+use pageserver::{tenant::{disk_btree::{DiskBtreeBuilder, DiskBtreeReader, VisitDirection}, block_io::{BlockBuf, FileBlockReader}, storage_layer::DeltaLayerWriter}, repository::Key, virtual_file::{VirtualFile, self}, page_cache};
+use std::{time::Instant, collections::BTreeMap};
+use rand::prelude::{SeedableRng, SliceRandom, StdRng};
+use utils::{id::{TimelineId, TenantId}, lsn::Lsn};
+use std::{io::{Read, Write}, path::PathBuf};
+use pageserver::config::PageServerConf;
+
+struct MockLayer {
+    pub path: PathBuf,
+    pub index_start_blk: u32,
+    pub index_root_blk: u32,
+}
+
+impl MockLayer {
+    fn read(&self, key: i128) -> Option<u64> {
+        // Read from disk btree
+        let file = FileBlockReader::new(VirtualFile::open(&self.path).unwrap());
+        let tree_reader = DiskBtreeReader::<_, 24>::new(
+            self.index_start_blk,
+            self.index_root_blk,
+            file,
+        );
+
+        let key: Key = Key::from_i128(key);
+        let mut key_bytes: [u8; 24] = [8u8; 24];
+        key.write_to_byte_slice(&mut key_bytes);
+
+        let mut result = None;
+        tree_reader.visit(&key_bytes, VisitDirection::Backwards, |key, value| {
+            if key == key_bytes {
+                result = Some(value);
+            }
+            return false
+        }).unwrap();
+
+        result
+    }
+}
+
+fn make_simple(n_keys: i128, name: &str) -> MockLayer {
+    let block_buf = BlockBuf::new();
+    let mut writer = DiskBtreeBuilder::<_, 24>::new(block_buf);
+    for i in 0..n_keys {
+        let key: Key = Key::from_i128(i);
+        let value: u64 = i as u64;
+
+        let mut key_bytes: [u8; 24] = [8u8; 24];
+        key.write_to_byte_slice(&mut key_bytes);
+        writer.append(&key_bytes, value).unwrap();
+    }
+    let (index_root_blk, block_buf) = writer.finish().unwrap();
+    let index_start_blk = 0; // ???
+    let path = std::env::current_dir().unwrap()
+        .parent().unwrap()
+        .join("test_output")
+        .join("bench_disk_lookup")
+        .join("disk_btree")
+        .join(name);
+    std::fs::create_dir_all(path.clone().parent().unwrap()).unwrap();
+    let layer = MockLayer {
+        path: path.clone(),
+        index_start_blk,
+        index_root_blk,
+    };
+
+    let mut file = VirtualFile::create(&path).unwrap();
+    for buf in block_buf.blocks {
+        file.write_all(buf.as_ref()).unwrap();
+    }
+
+    layer
+}
+
+fn make_many(n_keys: i128, n_layers: i128) -> Vec<MockLayer> {
+    (0..n_layers)
+        .map(|i| make_simple(n_keys, &format!("layer_{}.tmp", i)))
+        .collect()
+}
+
+
+// cargo bench --bench bench_disk_lookup
+fn bench_disk_lookup(c: &mut Criterion) {
+    virtual_file::init(10);
+    page_cache::init(10000);
+
+    // Results in a 40MB index
+    let n_keys = 4_000_000;
+
+    // One layer for each query
+    let n_layers = 100;
+    let n_queries = n_layers;
+
+    // Write to disk btree
+    let layers = make_many(n_keys, n_layers);
+
+    // Write to mem btrees
+    let mem_btrees: Vec<BTreeMap<i128, u64>> = (0..n_layers)
+        .map(|_| (0..n_keys)
+             .map(|i| (i as i128, i as u64))
+             .collect())
+        .collect();
+
+    // Pick queries
+    let rng = &mut StdRng::seed_from_u64(1);
+    let queries: Vec<_> = (0..n_keys).collect();
+    let queries: Vec<_> = queries.choose_multiple(rng, n_queries as usize).copied().collect();
+
+    // Define and name the benchmark function
+    let mut group = c.benchmark_group("g1");
+    group.bench_function("disk_btree", |b| {
+        b.iter(|| {
+            for (i, q) in queries.clone().into_iter().enumerate() {
+                black_box({
+                    assert_eq!(layers[i].read(q), Some(q as u64));
+                })
+            }
+        });
+    });
+    group.bench_function("mem_btree", |b| {
+        b.iter(|| {
+            for (i, q) in queries.clone().into_iter().enumerate() {
+                black_box({
+                    assert_eq!(mem_btrees[i].get(&q), Some(&(q as u64)));
+                })
+            }
+        });
+    });
+    group.finish();
+}
+
+criterion_group!(group_1, bench_disk_lookup);
+criterion_main!(group_1);

pageserver/src/bin/pageserver.rs

@@ -9,7 +9,6 @@ use clap::{Arg, ArgAction, Command};
 use fail::FailScenario;
 use metrics::launch_timestamp::{set_launch_timestamp_metric, LaunchTimestamp};
 use pageserver::disk_usage_eviction_task::{self, launch_disk_usage_global_eviction_task};
-use pageserver::task_mgr::WALRECEIVER_RUNTIME;
 use remote_storage::GenericRemoteStorage;
 use tracing::*;
 
@@ -19,7 +18,9 @@ use pageserver::{
     context::{DownloadBehavior, RequestContext},
     http, page_cache, page_service, task_mgr,
     task_mgr::TaskKind,
-    task_mgr::{BACKGROUND_RUNTIME, COMPUTE_REQUEST_RUNTIME, MGMT_REQUEST_RUNTIME},
+    task_mgr::{
+        BACKGROUND_RUNTIME, COMPUTE_REQUEST_RUNTIME, MGMT_REQUEST_RUNTIME, WALRECEIVER_RUNTIME,
+    },
     tenant::mgr,
     virtual_file,
 };
@@ -275,17 +276,7 @@ fn start_pageserver(
     let pageserver_listener = tcp_listener::bind(pg_addr)?;
 
     // Launch broker client
-    let broker_client = WALRECEIVER_RUNTIME
-        .block_on(async {
-            // Note: we do not attempt connecting here (but validate endpoints sanity).
-            storage_broker::connect(conf.broker_endpoint.clone(), conf.broker_keepalive_interval)
-        })
-        .with_context(|| {
-            format!(
-                "create broker client for uri={:?} keepalive_interval={:?}",
-                &conf.broker_endpoint, conf.broker_keepalive_interval,
-            )
-        })?;
+    WALRECEIVER_RUNTIME.block_on(pageserver::broker_client::init_broker_client(conf))?;
 
     // Initialize authentication for incoming connections
     let http_auth;
@@ -335,11 +326,7 @@ fn start_pageserver(
     let remote_storage = create_remote_storage_client(conf)?;
 
     // Scan the local 'tenants/' directory and start loading the tenants
-    BACKGROUND_RUNTIME.block_on(mgr::init_tenant_mgr(
-        conf,
-        broker_client.clone(),
-        remote_storage.clone(),
-    ))?;
+    BACKGROUND_RUNTIME.block_on(mgr::init_tenant_mgr(conf, remote_storage.clone()))?;
 
     // shared state between the disk-usage backed eviction background task and the http endpoint
     // that allows triggering disk-usage based eviction manually. note that the http endpoint
@@ -364,7 +351,6 @@ fn start_pageserver(
             conf,
             launch_ts,
             http_auth,
-            broker_client.clone(),
             remote_storage,
             disk_usage_eviction_state,
         )?
@@ -441,7 +427,6 @@ fn start_pageserver(
             async move {
                 page_service::libpq_listener_main(
                     conf,
-                    broker_client,
                     pg_auth,
                     pageserver_listener,
                     conf.pg_auth_type,

pageserver/src/bin/pageserver_binutils.rs

@@ -110,18 +110,6 @@ fn handle_metadata(path: &Path, arg_matches: &clap::ArgMatches) -> Result<(), an
         );
         update_meta = true;
     }
-    if let Some(latest_gc_cuttoff) = arg_matches.get_one::<String>("latest_gc_cuttoff") {
-        meta = TimelineMetadata::new(
-            meta.disk_consistent_lsn(),
-            meta.prev_record_lsn(),
-            meta.ancestor_timeline(),
-            meta.ancestor_lsn(),
-            Lsn::from_str(latest_gc_cuttoff)?,
-            meta.initdb_lsn(),
-            meta.pg_version(),
-        );
-        update_meta = true;
-    }
 
     if update_meta {
         let metadata_bytes = meta.to_bytes()?;
@@ -159,11 +147,6 @@ fn cli() -> Command {
                     Arg::new("prev_record_lsn")
                         .long("prev_record_lsn")
                         .help("Replace previous record Lsn"),
-                )
-                .arg(
-                    Arg::new("latest_gc_cuttoff")
-                        .long("latest_gc_cuttoff")
-                        .help("Replace latest gc cuttoff"),
                 ),
         )
 }

pageserver/src/broker_client.rs

@@ -0,0 +1,48 @@
+//! The broker client instance of the pageserver, created during pageserver startup.
+//! Used by each timelines' [`walreceiver`].
+
+use crate::config::PageServerConf;
+
+use anyhow::Context;
+use once_cell::sync::OnceCell;
+use storage_broker::BrokerClientChannel;
+use tracing::*;
+
+static BROKER_CLIENT: OnceCell<BrokerClientChannel> = OnceCell::new();
+
+///
+/// Initialize the broker client. This must be called once at page server startup.
+///
+pub async fn init_broker_client(conf: &'static PageServerConf) -> anyhow::Result<()> {
+    let broker_endpoint = conf.broker_endpoint.clone();
+
+    // Note: we do not attempt connecting here (but validate endpoints sanity).
+    let broker_client =
+        storage_broker::connect(broker_endpoint.clone(), conf.broker_keepalive_interval).context(
+            format!(
+                "Failed to create broker client to {}",
+                &conf.broker_endpoint
+            ),
+        )?;
+
+    if BROKER_CLIENT.set(broker_client).is_err() {
+        panic!("broker already initialized");
+    }
+
+    info!(
+        "Initialized broker client with endpoints: {}",
+        broker_endpoint
+    );
+    Ok(())
+}
+
+///
+/// Get a handle to the broker client
+///
+pub fn get_broker_client() -> &'static BrokerClientChannel {
+    BROKER_CLIENT.get().expect("broker client not initialized")
+}
+
+pub fn is_broker_client_initialized() -> bool {
+    BROKER_CLIENT.get().is_some()
+}

pageserver/src/config.rs

@@ -797,8 +797,7 @@ impl PageServerConf {
             )?);
         }
         if let Some(max_lsn_wal_lag) = item.get("max_lsn_wal_lag") {
-            t_conf.max_lsn_wal_lag =
-                Some(deserialize_from_item("max_lsn_wal_lag", max_lsn_wal_lag)?);
+            t_conf.max_lsn_wal_lag = Some(parse_toml_from_str("max_lsn_wal_lag", max_lsn_wal_lag)?);
         }
         if let Some(trace_read_requests) = item.get("trace_read_requests") {
             t_conf.trace_read_requests =

pageserver/src/consumption_metrics.rs

@@ -150,7 +150,7 @@ pub async fn collect_metrics_iteration(
         let mut tenant_resident_size = 0;
 
         // iterate through list of timelines in tenant
-        for timeline in tenant.list_timelines().await.iter() {
+        for timeline in tenant.list_timelines().iter() {
             // collect per-timeline metrics only for active timelines
             if timeline.is_active() {
                 let timeline_written_size = u64::from(timeline.get_last_record_lsn());

pageserver/src/disk_usage_eviction_task.rs

@@ -508,7 +508,7 @@ async fn collect_eviction_candidates(
         // a little unfair to tenants during shutdown in such a situation is tolerable.
         let mut tenant_candidates = Vec::new();
         let mut max_layer_size = 0;
-        for tl in tenant.list_timelines().await {
+        for tl in tenant.list_timelines() {
             if !tl.is_active() {
                 continue;
             }

pageserver/src/http/openapi_spec.yml

@@ -363,29 +363,11 @@ paths:
         * MUST NOT ASSUME that the request has been lost, based on the observation
           that a subsequent tenant status request returns 404. The request may
           still be in flight. It must be retried.
-
-        The client SHOULD supply a `TenantConfig` for the tenant in the request body.
-        Settings specified in the config override the pageserver's defaults.
-        It is guaranteed that the config settings are applied before the pageserver
-        starts operating on the tenant. E.g., if the config specifies a specific
-        PITR interval for a tenant, then that setting will be in effect before the
-        pageserver starts the garbage collection loop. This enables a client to
-        guarantee a specific PITR setting across detach/attach cycles.
-        The pageserver will reject the request if it cannot parse the config, or
-        if there are any unknown fields in it.
-
-        If the client does not supply a config, the pageserver will use its defaults.
-        This behavior is deprecated: https://github.com/neondatabase/neon/issues/4282
-      requestBody:
-        required: false
-        content:
-          application/json:
-            schema:
-              $ref: "#/components/schemas/TenantAttachRequest"
       responses:
         "202":
           description: Tenant attaching scheduled
         "400":
+          description: Error when no tenant id found in path parameters
           content:
             application/json:
               schema:
@@ -759,11 +741,8 @@ paths:
                 $ref: "#/components/schemas/Error"
     post:
       description: |
-        Create a tenant. Returns new tenant id on success.
-
+        Create a tenant. Returns new tenant id on success.\
         If no new tenant id is specified in parameters, it would be generated. It's an error to recreate the same tenant.
-
-        Invalid fields in the tenant config will cause the request to be rejected with status 400.
       requestBody:
         content:
           application/json:
@@ -811,8 +790,6 @@ paths:
     put:
       description: |
         Update tenant's config.
-
-        Invalid fields in the tenant config will cause the request to be rejected with status 400.
       requestBody:
         content:
           application/json:
@@ -940,13 +917,6 @@ components:
             new_tenant_id:
               type: string
               format: hex
-    TenantAttachRequest:
-      type: object
-      required:
-        - config
-      properties:
-        config:
-          $ref: '#/components/schemas/TenantConfig'
     TenantConfigRequest:
       allOf:
         - $ref: '#/components/schemas/TenantConfig'

pageserver/src/http/routes.rs

@@ -5,14 +5,12 @@ use anyhow::{anyhow, Context, Result};
 use hyper::StatusCode;
 use hyper::{Body, Request, Response, Uri};
 use metrics::launch_timestamp::LaunchTimestamp;
-use pageserver_api::models::{DownloadRemoteLayersTaskSpawnRequest, TenantAttachRequest};
+use pageserver_api::models::DownloadRemoteLayersTaskSpawnRequest;
 use remote_storage::GenericRemoteStorage;
-use storage_broker::BrokerClientChannel;
 use tenant_size_model::{SizeResult, StorageModel};
 use tokio_util::sync::CancellationToken;
 use tracing::*;
 use utils::http::endpoint::RequestSpan;
-use utils::http::json::json_request_or_empty_body;
 use utils::http::request::{get_request_param, must_get_query_param, parse_query_param};
 
 use super::models::{
@@ -52,7 +50,6 @@ struct State {
     auth: Option<Arc<JwtAuth>>,
     allowlist_routes: Vec<Uri>,
     remote_storage: Option<GenericRemoteStorage>,
-    broker_client: storage_broker::BrokerClientChannel,
     disk_usage_eviction_state: Arc<disk_usage_eviction_task::State>,
 }
 
@@ -61,7 +58,6 @@ impl State {
         conf: &'static PageServerConf,
         auth: Option<Arc<JwtAuth>>,
         remote_storage: Option<GenericRemoteStorage>,
-        broker_client: storage_broker::BrokerClientChannel,
         disk_usage_eviction_state: Arc<disk_usage_eviction_task::State>,
     ) -> anyhow::Result<Self> {
         let allowlist_routes = ["/v1/status", "/v1/doc", "/swagger.yml"]
@@ -73,7 +69,6 @@ impl State {
             auth,
             allowlist_routes,
             remote_storage,
-            broker_client,
             disk_usage_eviction_state,
         })
     }
@@ -275,8 +270,6 @@ async fn timeline_create_handler(mut request: Request<Body>) -> Result<Response<
 
     let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Error);
 
-    let state = get_state(&request);
-
     async {
         let tenant = mgr::get_tenant(tenant_id, true).await?;
         match tenant.create_timeline(
@@ -284,7 +277,6 @@ async fn timeline_create_handler(mut request: Request<Body>) -> Result<Response<
             request_data.ancestor_timeline_id.map(TimelineId::from),
             request_data.ancestor_start_lsn,
             request_data.pg_version.unwrap_or(crate::DEFAULT_PG_VERSION),
-            state.broker_client.clone(),
             &ctx,
         )
         .await {
@@ -312,7 +304,7 @@ async fn timeline_list_handler(request: Request<Body>) -> Result<Response<Body>,
 
     let response_data = async {
         let tenant = mgr::get_tenant(tenant_id, true).await?;
-        let timelines = tenant.list_timelines().await;
+        let timelines = tenant.list_timelines();
 
         let mut response_data = Vec::with_capacity(timelines.len());
         for timeline in timelines {
@@ -351,7 +343,6 @@ async fn timeline_detail_handler(request: Request<Body>) -> Result<Response<Body
 
         let timeline = tenant
             .get_timeline(timeline_id, false)
-            .await
             .map_err(ApiError::NotFound)?;
 
         let timeline_info = build_timeline_info(
@@ -395,16 +386,11 @@ async fn get_lsn_by_timestamp_handler(request: Request<Body>) -> Result<Response
     json_response(StatusCode::OK, result)
 }
 
-async fn tenant_attach_handler(mut request: Request<Body>) -> Result<Response<Body>, ApiError> {
+// TODO makes sense to provide tenant config right away the same way as it handled in tenant_create
+async fn tenant_attach_handler(request: Request<Body>) -> Result<Response<Body>, ApiError> {
     let tenant_id: TenantId = parse_request_param(&request, "tenant_id")?;
     check_permission(&request, Some(tenant_id))?;
 
-    let maybe_body: Option<TenantAttachRequest> = json_request_or_empty_body(&mut request).await?;
-    let tenant_conf = match maybe_body {
-        Some(request) => TenantConfOpt::try_from(&*request.config).map_err(ApiError::BadRequest)?,
-        None => TenantConfOpt::default(),
-    };
-
     let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Warn);
 
     info!("Handling tenant attach {tenant_id}");
@@ -415,8 +401,9 @@ async fn tenant_attach_handler(mut request: Request<Body>) -> Result<Response<Bo
         mgr::attach_tenant(
             state.conf,
             tenant_id,
-            tenant_conf,
-            state.broker_client.clone(),
+            // XXX: Attach should provide the config, especially during tenant migration.
+            //      See https://github.com/neondatabase/neon/issues/1555
+            TenantConfOpt::default(),
             remote_storage.clone(),
             &ctx,
         )
@@ -466,15 +453,9 @@ async fn tenant_load_handler(request: Request<Body>) -> Result<Response<Body>, A
     let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Warn);
 
     let state = get_state(&request);
-    mgr::load_tenant(
-        state.conf,
-        tenant_id,
-        state.broker_client.clone(),
-        state.remote_storage.clone(),
-        &ctx,
-    )
-    .instrument(info_span!("load", tenant = %tenant_id))
-    .await?;
+    mgr::load_tenant(state.conf, tenant_id, state.remote_storage.clone(), &ctx)
+        .instrument(info_span!("load", tenant = %tenant_id))
+        .await?;
 
     json_response(StatusCode::ACCEPTED, ())
 }
@@ -521,7 +502,7 @@ async fn tenant_status(request: Request<Body>) -> Result<Response<Body>, ApiErro
 
         // Calculate total physical size of all timelines
         let mut current_physical_size = 0;
-        for timeline in tenant.list_timelines().await.iter() {
+        for timeline in tenant.list_timelines().iter() {
             current_physical_size += timeline.layer_size_sum();
         }
 
@@ -759,7 +740,6 @@ async fn tenant_create_handler(mut request: Request<Body>) -> Result<Response<Bo
         state.conf,
         tenant_conf,
         target_tenant_id,
-        state.broker_client.clone(),
         state.remote_storage.clone(),
         &ctx,
     )
@@ -836,7 +816,7 @@ async fn handle_tenant_break(r: Request<Body>) -> Result<Response<Body>, ApiErro
         .await
         .map_err(|_| ApiError::Conflict(String::from("no active tenant found")))?;
 
-    tenant.set_broken("broken from test".to_owned()).await;
+    tenant.set_broken("broken from test".to_owned());
 
     json_response(StatusCode::OK, ())
 }
@@ -976,7 +956,6 @@ async fn active_timeline_of_active_tenant(
     let tenant = mgr::get_tenant(tenant_id, true).await?;
     tenant
         .get_timeline(timeline_id, true)
-        .await
         .map_err(ApiError::NotFound)
 }
 
@@ -1116,7 +1095,6 @@ pub fn make_router(
     conf: &'static PageServerConf,
     launch_ts: &'static LaunchTimestamp,
     auth: Option<Arc<JwtAuth>>,
-    broker_client: BrokerClientChannel,
     remote_storage: Option<GenericRemoteStorage>,
     disk_usage_eviction_state: Arc<disk_usage_eviction_task::State>,
 ) -> anyhow::Result<RouterBuilder<hyper::Body, ApiError>> {
@@ -1163,14 +1141,8 @@ pub fn make_router(
 
     Ok(router
         .data(Arc::new(
-            State::new(
-                conf,
-                auth,
-                remote_storage,
-                broker_client,
-                disk_usage_eviction_state,
-            )
-            .context("Failed to initialize router state")?,
+            State::new(conf, auth, remote_storage, disk_usage_eviction_state)
+                .context("Failed to initialize router state")?,
         ))
         .get("/v1/status", |r| RequestSpan(status_handler).handle(r))
         .put(

pageserver/src/keyspace.rs

@@ -5,7 +5,7 @@ use std::ops::Range;
 ///
 /// Represents a set of Keys, in a compact form.
 ///
-#[derive(Clone, Debug, Default)]
+#[derive(Clone, Debug)]
 pub struct KeySpace {
     /// Contiguous ranges of keys that belong to the key space. In key order,
     /// and with no overlap.
@@ -61,18 +61,6 @@ impl KeySpace {
 
         KeyPartitioning { parts }
     }
-
-    ///
-    /// Check if key space contains overlapping range
-    ///
-    pub fn overlaps(&self, range: &Range<Key>) -> bool {
-        match self.ranges.binary_search_by_key(&range.end, |r| r.start) {
-            Ok(0) => false,
-            Err(0) => false,
-            Ok(index) => self.ranges[index - 1].end > range.start,
-            Err(index) => self.ranges[index - 1].end > range.start,
-        }
-    }
 }
 
 ///
@@ -141,226 +129,3 @@ impl KeySpaceAccum {
         }
     }
 }
-
-///
-/// A helper object, to collect a set of keys and key ranges into a KeySpace
-/// object. Key ranges may be inserted in any order and can overlap.
-///
-#[derive(Clone, Debug, Default)]
-pub struct KeySpaceRandomAccum {
-    ranges: Vec<Range<Key>>,
-}
-
-impl KeySpaceRandomAccum {
-    pub fn new() -> Self {
-        Self { ranges: Vec::new() }
-    }
-
-    pub fn add_key(&mut self, key: Key) {
-        self.add_range(singleton_range(key))
-    }
-
-    pub fn add_range(&mut self, range: Range<Key>) {
-        self.ranges.push(range);
-    }
-
-    pub fn to_keyspace(mut self) -> KeySpace {
-        let mut ranges = Vec::new();
-        if !self.ranges.is_empty() {
-            self.ranges.sort_by_key(|r| r.start);
-            let mut start = self.ranges.first().unwrap().start;
-            let mut end = self.ranges.first().unwrap().end;
-            for r in self.ranges {
-                assert!(r.start >= start);
-                if r.start > end {
-                    ranges.push(start..end);
-                    start = r.start;
-                    end = r.end;
-                } else if r.end > end {
-                    end = r.end;
-                }
-            }
-            ranges.push(start..end);
-        }
-        KeySpace { ranges }
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use std::fmt::Write;
-
-    // Helper function to create a key range.
-    //
-    // Make the tests below less verbose.
-    fn kr(irange: Range<i128>) -> Range<Key> {
-        Key::from_i128(irange.start)..Key::from_i128(irange.end)
-    }
-
-    #[allow(dead_code)]
-    fn dump_keyspace(ks: &KeySpace) {
-        for r in ks.ranges.iter() {
-            println!("  {}..{}", r.start.to_i128(), r.end.to_i128());
-        }
-    }
-
-    fn assert_ks_eq(actual: &KeySpace, expected: Vec<Range<Key>>) {
-        if actual.ranges != expected {
-            let mut msg = String::new();
-
-            writeln!(msg, "expected:").unwrap();
-            for r in &expected {
-                writeln!(msg, "  {}..{}", r.start.to_i128(), r.end.to_i128()).unwrap();
-            }
-            writeln!(msg, "got:").unwrap();
-            for r in &actual.ranges {
-                writeln!(msg, "  {}..{}", r.start.to_i128(), r.end.to_i128()).unwrap();
-            }
-            panic!("{}", msg);
-        }
-    }
-
-    #[test]
-    fn keyspace_add_range() {
-        // two separate ranges
-        //
-        // #####
-        //         #####
-        let mut ks = KeySpaceRandomAccum::default();
-        ks.add_range(kr(0..10));
-        ks.add_range(kr(20..30));
-        assert_ks_eq(&ks.to_keyspace(), vec![kr(0..10), kr(20..30)]);
-
-        // two separate ranges, added in reverse order
-        //
-        //         #####
-        // #####
-        let mut ks = KeySpaceRandomAccum::default();
-        ks.add_range(kr(20..30));
-        ks.add_range(kr(0..10));
-
-        // add range that is adjacent to the end of an existing range
-        //
-        // #####
-        //      #####
-        ks.add_range(kr(0..10));
-        ks.add_range(kr(10..30));
-        assert_ks_eq(&ks.to_keyspace(), vec![kr(0..30)]);
-
-        // add range that is adjacent to the start of an existing range
-        //
-        //      #####
-        // #####
-        let mut ks = KeySpaceRandomAccum::default();
-        ks.add_range(kr(10..30));
-        ks.add_range(kr(0..10));
-        assert_ks_eq(&ks.to_keyspace(), vec![kr(0..30)]);
-
-        // add range that overlaps with the end of an existing range
-        //
-        // #####
-        //    #####
-        let mut ks = KeySpaceRandomAccum::default();
-        ks.add_range(kr(0..10));
-        ks.add_range(kr(5..30));
-        assert_ks_eq(&ks.to_keyspace(), vec![kr(0..30)]);
-
-        // add range that overlaps with the start of an existing range
-        //
-        //    #####
-        // #####
-        let mut ks = KeySpaceRandomAccum::default();
-        ks.add_range(kr(5..30));
-        ks.add_range(kr(0..10));
-        assert_ks_eq(&ks.to_keyspace(), vec![kr(0..30)]);
-
-        // add range that is fully covered by an existing range
-        //
-        // #########
-        //   #####
-        let mut ks = KeySpaceRandomAccum::default();
-        ks.add_range(kr(0..30));
-        ks.add_range(kr(10..20));
-        assert_ks_eq(&ks.to_keyspace(), vec![kr(0..30)]);
-
-        // add range that extends an existing range from both ends
-        //
-        //   #####
-        // #########
-        let mut ks = KeySpaceRandomAccum::default();
-        ks.add_range(kr(10..20));
-        ks.add_range(kr(0..30));
-        assert_ks_eq(&ks.to_keyspace(), vec![kr(0..30)]);
-
-        // add a range that overlaps with two existing ranges, joining them
-        //
-        // #####   #####
-        //    #######
-        let mut ks = KeySpaceRandomAccum::default();
-        ks.add_range(kr(0..10));
-        ks.add_range(kr(20..30));
-        ks.add_range(kr(5..25));
-        assert_ks_eq(&ks.to_keyspace(), vec![kr(0..30)]);
-    }
-
-    #[test]
-    fn keyspace_overlaps() {
-        let mut ks = KeySpaceRandomAccum::default();
-        ks.add_range(kr(10..20));
-        ks.add_range(kr(30..40));
-        let ks = ks.to_keyspace();
-
-        //        #####      #####
-        // xxxx
-        assert!(!ks.overlaps(&kr(0..5)));
-
-        //        #####      #####
-        //   xxxx
-        assert!(!ks.overlaps(&kr(5..9)));
-
-        //        #####      #####
-        //    xxxx
-        assert!(!ks.overlaps(&kr(5..10)));
-
-        //        #####      #####
-        //     xxxx
-        assert!(ks.overlaps(&kr(5..11)));
-
-        //        #####      #####
-        //        xxxx
-        assert!(ks.overlaps(&kr(10..15)));
-
-        //        #####      #####
-        //         xxxx
-        assert!(ks.overlaps(&kr(15..20)));
-
-        //        #####      #####
-        //           xxxx
-        assert!(ks.overlaps(&kr(15..25)));
-
-        //        #####      #####
-        //              xxxx
-        assert!(!ks.overlaps(&kr(22..28)));
-
-        //        #####      #####
-        //               xxxx
-        assert!(!ks.overlaps(&kr(25..30)));
-
-        //        #####      #####
-        //                      xxxx
-        assert!(ks.overlaps(&kr(35..35)));
-
-        //        #####      #####
-        //                        xxxx
-        assert!(!ks.overlaps(&kr(40..45)));
-
-        //        #####      #####
-        //                        xxxx
-        assert!(!ks.overlaps(&kr(45..50)));
-
-        //        #####      #####
-        //        xxxxxxxxxxx
-        assert!(ks.overlaps(&kr(0..30))); // XXXXX This fails currently!
-    }
-}

pageserver/src/lib.rs

@@ -1,5 +1,6 @@
 mod auth;
 pub mod basebackup;
+pub mod broker_client;
 pub mod config;
 pub mod consumption_metrics;
 pub mod context;

pageserver/src/page_service.rs

@@ -172,7 +172,6 @@ async fn read_tar_eof(mut reader: (impl AsyncRead + Unpin)) -> anyhow::Result<()
 ///
 pub async fn libpq_listener_main(
     conf: &'static PageServerConf,
-    broker_client: storage_broker::BrokerClientChannel,
     auth: Option<Arc<JwtAuth>>,
     listener: TcpListener,
     auth_type: AuthType,
@@ -214,14 +213,7 @@ pub async fn libpq_listener_main(
                     None,
                     "serving compute connection task",
                     false,
-                    page_service_conn_main(
-                        conf,
-                        broker_client.clone(),
-                        local_auth,
-                        socket,
-                        auth_type,
-                        connection_ctx,
-                    ),
+                    page_service_conn_main(conf, local_auth, socket, auth_type, connection_ctx),
                 );
             }
             Err(err) => {
@@ -238,7 +230,6 @@ pub async fn libpq_listener_main(
 
 async fn page_service_conn_main(
     conf: &'static PageServerConf,
-    broker_client: storage_broker::BrokerClientChannel,
     auth: Option<Arc<JwtAuth>>,
     socket: tokio::net::TcpStream,
     auth_type: AuthType,
@@ -275,7 +266,7 @@ async fn page_service_conn_main(
     // and create a child per-query context when it invokes process_query.
     // But it's in a shared crate, so, we store connection_ctx inside PageServerHandler
     // and create the per-query context in process_query ourselves.
-    let mut conn_handler = PageServerHandler::new(conf, broker_client, auth, connection_ctx);
+    let mut conn_handler = PageServerHandler::new(conf, auth, connection_ctx);
     let pgbackend = PostgresBackend::new_from_io(socket, peer_addr, auth_type, None)?;
 
     match pgbackend
@@ -333,7 +324,6 @@ impl PageRequestMetrics {
 
 struct PageServerHandler {
     _conf: &'static PageServerConf,
-    broker_client: storage_broker::BrokerClientChannel,
     auth: Option<Arc<JwtAuth>>,
     claims: Option<Claims>,
 
@@ -347,13 +337,11 @@ struct PageServerHandler {
 impl PageServerHandler {
     pub fn new(
         conf: &'static PageServerConf,
-        broker_client: storage_broker::BrokerClientChannel,
         auth: Option<Arc<JwtAuth>>,
         connection_ctx: RequestContext,
     ) -> Self {
         PageServerHandler {
             _conf: conf,
-            broker_client,
             auth,
             claims: None,
             connection_ctx,
@@ -388,7 +376,7 @@ impl PageServerHandler {
         };
 
         // Check that the timeline exists
-        let timeline = tenant.get_timeline(timeline_id, true).await?;
+        let timeline = tenant.get_timeline(timeline_id, true)?;
 
         // switch client to COPYBOTH
         pgb.write_message_noflush(&BeMessage::CopyBothResponse)?;
@@ -487,9 +475,7 @@ impl PageServerHandler {
         // Create empty timeline
         info!("creating new timeline");
         let tenant = get_active_tenant_with_timeout(tenant_id, &ctx).await?;
-        let timeline = tenant
-            .create_empty_timeline(timeline_id, base_lsn, pg_version, &ctx)
-            .await?;
+        let timeline = tenant.create_empty_timeline(timeline_id, base_lsn, pg_version, &ctx)?;
 
         // TODO mark timeline as not ready until it reaches end_lsn.
         // We might have some wal to import as well, and we should prevent compute
@@ -508,12 +494,7 @@ impl PageServerHandler {
 
         let mut copyin_reader = pin!(StreamReader::new(copyin_stream(pgb)));
         timeline
-            .import_basebackup_from_tar(
-                &mut copyin_reader,
-                base_lsn,
-                self.broker_client.clone(),
-                &ctx,
-            )
+            .import_basebackup_from_tar(&mut copyin_reader, base_lsn, &ctx)
             .await?;
 
         // Read the end of the tar archive.
@@ -1203,6 +1184,6 @@ async fn get_active_tenant_timeline(
     ctx: &RequestContext,
 ) -> Result<Arc<Timeline>, GetActiveTenantError> {
     let tenant = get_active_tenant_with_timeout(tenant_id, ctx).await?;
-    let timeline = tenant.get_timeline(timeline_id, true).await?;
+    let timeline = tenant.get_timeline(timeline_id, true)?;
     Ok(timeline)
 }

pageserver/src/pgdatadir_mapping.rs

@@ -1594,15 +1594,15 @@ fn is_slru_block_key(key: Key) -> bool {
 }
 
 #[cfg(test)]
-pub async fn create_test_timeline(
+pub fn create_test_timeline(
     tenant: &crate::tenant::Tenant,
     timeline_id: utils::id::TimelineId,
     pg_version: u32,
     ctx: &RequestContext,
 ) -> anyhow::Result<std::sync::Arc<Timeline>> {
     let tline = tenant
-        .create_test_timeline(timeline_id, Lsn(8), pg_version, ctx)
-        .await?;
+        .create_empty_timeline(timeline_id, Lsn(8), pg_version, ctx)?
+        .initialize(ctx)?;
     let mut m = tline.begin_modification(Lsn(8));
     m.init_empty()?;
     m.commit()?;
@@ -1632,7 +1632,7 @@ mod tests {
     #[test]
     fn test_list_rels_drop() -> Result<()> {
         let repo = RepoHarness::create("test_list_rels_drop")?.load();
-        let tline = create_empty_timeline(repo, TIMELINE_ID).await?;
+        let tline = create_empty_timeline(repo, TIMELINE_ID)?;
         const TESTDB: u32 = 111;
 
         // Import initial dummy checkpoint record, otherwise the get_timeline() call

pageserver/src/tenant/mgr.rs

@@ -58,10 +58,9 @@ static TENANTS: Lazy<RwLock<TenantsMap>> = Lazy::new(|| RwLock::new(TenantsMap::
 /// Initialize repositories with locally available timelines.
 /// Timelines that are only partially available locally (remote storage has more data than this pageserver)
 /// are scheduled for download and added to the tenant once download is completed.
-#[instrument(skip_all)]
+#[instrument(skip(conf, remote_storage))]
 pub async fn init_tenant_mgr(
     conf: &'static PageServerConf,
-    broker_client: storage_broker::BrokerClientChannel,
     remote_storage: Option<GenericRemoteStorage>,
 ) -> anyhow::Result<()> {
     // Scan local filesystem for attached tenants
@@ -117,7 +116,6 @@ pub async fn init_tenant_mgr(
                     match schedule_local_tenant_processing(
                         conf,
                         &tenant_dir_path,
-                        broker_client.clone(),
                         remote_storage.clone(),
                         &ctx,
                     ) {
@@ -152,7 +150,6 @@ pub async fn init_tenant_mgr(
 pub fn schedule_local_tenant_processing(
     conf: &'static PageServerConf,
     tenant_path: &Path,
-    broker_client: storage_broker::BrokerClientChannel,
     remote_storage: Option<GenericRemoteStorage>,
     ctx: &RequestContext,
 ) -> anyhow::Result<Arc<Tenant>> {
@@ -189,7 +186,7 @@ pub fn schedule_local_tenant_processing(
     let tenant = if conf.tenant_attaching_mark_file_path(&tenant_id).exists() {
         info!("tenant {tenant_id} has attaching mark file, resuming its attach operation");
         if let Some(remote_storage) = remote_storage {
-            match Tenant::spawn_attach(conf, tenant_id, broker_client, remote_storage, ctx) {
+            match Tenant::spawn_attach(conf, tenant_id, remote_storage, ctx) {
                 Ok(tenant) => tenant,
                 Err(e) => {
                     error!("Failed to spawn_attach tenant {tenant_id}, reason: {e:#}");
@@ -207,7 +204,7 @@ pub fn schedule_local_tenant_processing(
     } else {
         info!("tenant {tenant_id} is assumed to be loadable, starting load operation");
         // Start loading the tenant into memory. It will initially be in Loading state.
-        Tenant::spawn_load(conf, tenant_id, broker_client, remote_storage, ctx)
+        Tenant::spawn_load(conf, tenant_id, remote_storage, ctx)
     };
     Ok(tenant)
 }
@@ -248,7 +245,7 @@ pub async fn shutdown_all_tenants() {
     for (_, tenant) in tenants_to_shut_down {
         if tenant.is_active() {
             // updates tenant state, forbidding new GC and compaction iterations from starting
-            tenant.set_stopping().await;
+            tenant.set_stopping();
             tenants_to_freeze_and_flush.push(tenant);
         }
     }
@@ -278,11 +275,10 @@ pub async fn create_tenant(
     conf: &'static PageServerConf,
     tenant_conf: TenantConfOpt,
     tenant_id: TenantId,
-    broker_client: storage_broker::BrokerClientChannel,
     remote_storage: Option<GenericRemoteStorage>,
     ctx: &RequestContext,
 ) -> Result<Arc<Tenant>, TenantMapInsertError> {
-    tenant_map_insert(tenant_id, || {
+    tenant_map_insert(tenant_id, |vacant_entry| {
         // We're holding the tenants lock in write mode while doing local IO.
         // If this section ever becomes contentious, introduce a new `TenantState::Creating`
         // and do the work in that state.
@@ -291,7 +287,7 @@ pub async fn create_tenant(
         //       See https://github.com/neondatabase/neon/issues/4233
 
         let created_tenant =
-            schedule_local_tenant_processing(conf, &tenant_directory, broker_client, remote_storage, ctx)?;
+            schedule_local_tenant_processing(conf, &tenant_directory, remote_storage, ctx)?;
         // TODO: tenant object & its background loops remain, untracked in tenant map, if we fail here.
         //      See https://github.com/neondatabase/neon/issues/4233
 
@@ -300,6 +296,7 @@ pub async fn create_tenant(
                 tenant_id == crated_tenant_id,
                 "loaded created tenant has unexpected tenant id (expect {tenant_id} != actual {crated_tenant_id})",
             );
+        vacant_entry.insert(Arc::clone(&created_tenant));
         Ok(created_tenant)
     }).await
 }
@@ -319,7 +316,7 @@ pub async fn set_new_tenant_config(
         new_tenant_conf,
         false,
     )?;
-    tenant.set_new_tenant_config(new_tenant_conf).await;
+    tenant.set_new_tenant_config(new_tenant_conf);
     Ok(())
 }
 
@@ -408,11 +405,10 @@ pub async fn detach_tenant(
 pub async fn load_tenant(
     conf: &'static PageServerConf,
     tenant_id: TenantId,
-    broker_client: storage_broker::BrokerClientChannel,
     remote_storage: Option<GenericRemoteStorage>,
     ctx: &RequestContext,
 ) -> Result<(), TenantMapInsertError> {
-    tenant_map_insert(tenant_id, || {
+    tenant_map_insert(tenant_id, |vacant_entry| {
         let tenant_path = conf.tenant_path(&tenant_id);
         let tenant_ignore_mark = conf.tenant_ignore_mark_file_path(tenant_id);
         if tenant_ignore_mark.exists() {
@@ -420,14 +416,14 @@ pub async fn load_tenant(
                 .with_context(|| format!("Failed to remove tenant ignore mark {tenant_ignore_mark:?} during tenant loading"))?;
         }
 
-        let new_tenant = schedule_local_tenant_processing(conf, &tenant_path, broker_client, remote_storage, ctx)
+        let new_tenant = schedule_local_tenant_processing(conf, &tenant_path, remote_storage, ctx)
             .with_context(|| {
                 format!("Failed to schedule tenant processing in path {tenant_path:?}")
             })?;
 
-        Ok(new_tenant)
-    }).await?;
-    Ok(())
+        vacant_entry.insert(new_tenant);
+        Ok(())
+    }).await
 }
 
 pub async fn ignore_tenant(
@@ -477,11 +473,10 @@ pub async fn attach_tenant(
     conf: &'static PageServerConf,
     tenant_id: TenantId,
     tenant_conf: TenantConfOpt,
-    broker_client: storage_broker::BrokerClientChannel,
     remote_storage: GenericRemoteStorage,
     ctx: &RequestContext,
 ) -> Result<(), TenantMapInsertError> {
-    tenant_map_insert(tenant_id, || {
+    tenant_map_insert(tenant_id, |vacant_entry| {
         let tenant_dir = create_tenant_files(conf, tenant_conf, tenant_id, CreateTenantFilesMode::Attach)?;
         // TODO: tenant directory remains on disk if we bail out from here on.
         //       See https://github.com/neondatabase/neon/issues/4233
@@ -493,7 +488,7 @@ pub async fn attach_tenant(
             .context("check for attach marker file existence")?;
         anyhow::ensure!(marker_file_exists, "create_tenant_files should have created the attach marker file");
 
-        let attached_tenant = schedule_local_tenant_processing(conf, &tenant_dir, broker_client, Some(remote_storage), ctx)?;
+        let attached_tenant = schedule_local_tenant_processing(conf, &tenant_dir, Some(remote_storage), ctx)?;
         // TODO: tenant object & its background loops remain, untracked in tenant map, if we fail here.
         //      See https://github.com/neondatabase/neon/issues/4233
 
@@ -502,10 +497,10 @@ pub async fn attach_tenant(
             tenant_id == attached_tenant_id,
             "loaded created tenant has unexpected tenant id (expect {tenant_id} != actual {attached_tenant_id})",
         );
-        Ok(attached_tenant)
+        vacant_entry.insert(Arc::clone(&attached_tenant));
+        Ok(())
     })
-    .await?;
-    Ok(())
+    .await
 }
 
 #[derive(Debug, thiserror::Error)]
@@ -526,12 +521,12 @@ pub enum TenantMapInsertError {
 ///
 /// NB: the closure should return quickly because the current implementation of tenants map
 /// serializes access through an `RwLock`.
-async fn tenant_map_insert<F>(
+async fn tenant_map_insert<F, V>(
     tenant_id: TenantId,
     insert_fn: F,
-) -> Result<Arc<Tenant>, TenantMapInsertError>
+) -> Result<V, TenantMapInsertError>
 where
-    F: FnOnce() -> anyhow::Result<Arc<Tenant>>,
+    F: FnOnce(hash_map::VacantEntry<TenantId, Arc<Tenant>>) -> anyhow::Result<V>,
 {
     let mut guard = TENANTS.write().await;
     let m = match &mut *guard {
@@ -544,11 +539,8 @@ where
             tenant_id,
             e.get().current_state(),
         )),
-        hash_map::Entry::Vacant(v) => match insert_fn() {
-            Ok(tenant) => {
-                v.insert(tenant.clone());
-                Ok(tenant)
-            }
+        hash_map::Entry::Vacant(v) => match insert_fn(v) {
+            Ok(v) => Ok(v),
             Err(e) => Err(TenantMapInsertError::Closure(e)),
         },
     }
@@ -575,9 +567,8 @@ where
             Some(tenant) => match tenant.current_state() {
                 TenantState::Attaching
                 | TenantState::Loading
-                | TenantState::Activating
                 | TenantState::Broken { .. }
-                | TenantState::Active => tenant.set_stopping().await,
+                | TenantState::Active => tenant.set_stopping(),
                 TenantState::Stopping => return Err(TenantStateError::IsStopping(tenant_id)),
             },
             None => return Err(TenantStateError::NotFound(tenant_id)),
@@ -604,7 +595,7 @@ where
             let tenants_accessor = TENANTS.read().await;
             match tenants_accessor.get(&tenant_id) {
                 Some(tenant) => {
-                    tenant.set_broken(e.to_string()).await;
+                    tenant.set_broken(e.to_string());
                 }
                 None => {
                     warn!("Tenant {tenant_id} got removed from memory");
@@ -686,7 +677,6 @@ pub async fn immediate_compact(
 
     let timeline = tenant
         .get_timeline(timeline_id, true)
-        .await
         .map_err(ApiError::NotFound)?;
 
     // Run in task_mgr to avoid race with tenant_detach operation

pageserver/src/tenant/remote_timeline_client.rs

@@ -1264,12 +1264,9 @@ mod tests {
             let harness = TenantHarness::create(test_name)?;
             let (tenant, ctx) = runtime.block_on(harness.load());
             // create an empty timeline directory
-            let _ = runtime.block_on(tenant.create_test_timeline(
-                TIMELINE_ID,
-                Lsn(0),
-                DEFAULT_PG_VERSION,
-                &ctx,
-            ))?;
+            let timeline =
+                tenant.create_empty_timeline(TIMELINE_ID, Lsn(0), DEFAULT_PG_VERSION, &ctx)?;
+            let _ = timeline.initialize(&ctx).unwrap();
 
             let remote_fs_dir = harness.conf.workdir.join("remote_fs");
             std::fs::create_dir_all(remote_fs_dir)?;

pageserver/src/tenant/size.rs

@@ -136,7 +136,7 @@ pub(super) async fn gather_inputs(
         .context("Failed to refresh gc_info before gathering inputs")?;
 
     // Collect information about all the timelines
-    let mut timelines = tenant.list_timelines().await;
+    let mut timelines = tenant.list_timelines();
 
     if timelines.is_empty() {
         // perhaps the tenant has just been created, and as such doesn't have any data yet

pageserver/src/tenant/timeline.rs

@@ -22,7 +22,8 @@ use tracing::*;
 use utils::id::TenantTimelineId;
 
 use std::cmp::{max, min, Ordering};
-use std::collections::{BinaryHeap, HashMap};
+use std::collections::BinaryHeap;
+use std::collections::HashMap;
 use std::fs;
 use std::ops::{Deref, Range};
 use std::path::{Path, PathBuf};
@@ -31,6 +32,7 @@ use std::sync::atomic::{AtomicI64, Ordering as AtomicOrdering};
 use std::sync::{Arc, Mutex, MutexGuard, RwLock, Weak};
 use std::time::{Duration, Instant, SystemTime};
 
+use crate::broker_client::{get_broker_client, is_broker_client_initialized};
 use crate::context::{DownloadBehavior, RequestContext};
 use crate::tenant::remote_timeline_client::{self, index::LayerFileMetadata};
 use crate::tenant::storage_layer::{
@@ -46,7 +48,7 @@ use crate::tenant::{
 };
 
 use crate::config::PageServerConf;
-use crate::keyspace::{KeyPartitioning, KeySpace, KeySpaceRandomAccum};
+use crate::keyspace::{KeyPartitioning, KeySpace};
 use crate::metrics::{TimelineMetrics, UNEXPECTED_ONDEMAND_DOWNLOADS};
 use crate::pgdatadir_mapping::LsnForTimestamp;
 use crate::pgdatadir_mapping::{is_rel_fsm_block_key, is_rel_vm_block_key};
@@ -121,17 +123,6 @@ pub struct Timeline {
 
     pub(super) layers: RwLock<LayerMap<dyn PersistentLayer>>,
 
-    /// Set of key ranges which should be covered by image layers to
-    /// allow GC to remove old layers. This set is created by GC and its cutoff LSN is also stored.
-    /// It is used by compaction task when it checks if new image layer should be created.
-    /// Newly created image layer doesn't help to remove the delta layer, until the
-    /// newly created image layer falls off the PITR horizon. So on next GC cycle,
-    /// gc_timeline may still want the new image layer to be created. To avoid redundant
-    /// image layers creation we should check if image layer exists but beyond PITR horizon.
-    /// This is why we need remember GC cutoff LSN.
-    ///
-    wanted_image_layers: Mutex<Option<(Lsn, KeySpace)>>,
-
     last_freeze_at: AtomicLsn,
     // Atomic would be more appropriate here.
     last_freeze_ts: RwLock<Instant>,
@@ -226,7 +217,7 @@ pub struct Timeline {
     /// or None if WAL receiver has not received anything for this timeline
     /// yet.
     pub last_received_wal: Mutex<Option<WalReceiverInfo>>,
-    pub walreceiver: Mutex<Option<WalReceiver>>,
+    pub walreceiver: WalReceiver,
 
     /// Relation size cache
     pub rel_size_cache: RwLock<HashMap<RelTag, (Lsn, BlockNumber)>>,
@@ -621,27 +612,17 @@ impl Timeline {
             .await
         {
             Ok(()) => Ok(()),
-            Err(e) => {
-                // walreceiver.status() locks internally, don't count that towards the wait_lsn_time_histo
+            seqwait_error => {
                 drop(_timer);
-                let walreceiver_status = {
-                    match &*self.walreceiver.lock().unwrap() {
-                        None => "stopping or stopped".to_string(),
-                        Some(walreceiver) => match walreceiver.status() {
-                            Some(status) => status.to_human_readable_string(),
-                            None => "Not active".to_string(),
-                        },
-                    }
-                };
-                Err(anyhow::Error::new(e).context({
-                    format!(
-                        "Timed out while waiting for WAL record at LSN {} to arrive, last_record_lsn {} disk consistent LSN={}, WalReceiver status: {}",
-                        lsn,
-                        self.get_last_record_lsn(),
-                        self.get_disk_consistent_lsn(),
-                        walreceiver_status,
-                    )
-                }))
+                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()),
+                ))
             }
         }
     }
@@ -916,10 +897,18 @@ impl Timeline {
         Ok(())
     }
 
-    pub fn activate(self: &Arc<Self>, broker_client: BrokerClientChannel, ctx: &RequestContext) {
-        self.launch_wal_receiver(ctx, broker_client);
+    pub fn activate(self: &Arc<Self>, ctx: &RequestContext) -> anyhow::Result<()> {
+        if is_broker_client_initialized() {
+            self.launch_wal_receiver(ctx, get_broker_client().clone())?;
+        } else if cfg!(test) {
+            info!("not launching WAL receiver because broker client hasn't been initialized");
+        } else {
+            anyhow::bail!("broker client not initialized");
+        }
+
         self.set_state(TimelineState::Active);
         self.launch_eviction_task();
+        Ok(())
     }
 
     pub fn set_state(&self, new_state: TimelineState) {
@@ -1328,7 +1317,15 @@ impl Timeline {
         let (layer_flush_done_tx, _) = tokio::sync::watch::channel((0, Ok(())));
 
         let tenant_conf_guard = tenant_conf.read().unwrap();
-
+        let wal_connect_timeout = tenant_conf_guard
+            .walreceiver_connect_timeout
+            .unwrap_or(conf.default_tenant_conf.walreceiver_connect_timeout);
+        let lagging_wal_timeout = tenant_conf_guard
+            .lagging_wal_timeout
+            .unwrap_or(conf.default_tenant_conf.lagging_wal_timeout);
+        let max_lsn_wal_lag = tenant_conf_guard
+            .max_lsn_wal_lag
+            .unwrap_or(conf.default_tenant_conf.max_lsn_wal_lag);
         let evictions_low_residence_duration_metric_threshold =
             Self::get_evictions_low_residence_duration_metric_threshold(
                 &tenant_conf_guard,
@@ -1337,6 +1334,18 @@ impl Timeline {
         drop(tenant_conf_guard);
 
         Arc::new_cyclic(|myself| {
+            let walreceiver = WalReceiver::new(
+                TenantTimelineId::new(tenant_id, timeline_id),
+                Weak::clone(myself),
+                WalReceiverConf {
+                    wal_connect_timeout,
+                    lagging_wal_timeout,
+                    max_lsn_wal_lag,
+                    auth_token: crate::config::SAFEKEEPER_AUTH_TOKEN.get().cloned(),
+                    availability_zone: conf.availability_zone.clone(),
+                },
+            );
+
             let mut result = Timeline {
                 conf,
                 tenant_conf,
@@ -1345,10 +1354,9 @@ impl Timeline {
                 tenant_id,
                 pg_version,
                 layers: RwLock::new(LayerMap::default()),
-                wanted_image_layers: Mutex::new(None),
 
                 walredo_mgr,
-                walreceiver: Mutex::new(None),
+                walreceiver,
 
                 remote_client: remote_client.map(Arc::new),
 
@@ -1468,49 +1476,17 @@ impl Timeline {
         *flush_loop_state = FlushLoopState::Running;
     }
 
-    /// Creates and starts the wal receiver.
-    ///
-    /// This function is expected to be called at most once per Timeline's lifecycle
-    /// when the timeline is activated.
-    fn launch_wal_receiver(
-        self: &Arc<Self>,
+    pub(super) fn launch_wal_receiver(
+        &self,
         ctx: &RequestContext,
         broker_client: BrokerClientChannel,
-    ) {
+    ) -> anyhow::Result<()> {
         info!(
             "launching WAL receiver for timeline {} of tenant {}",
             self.timeline_id, self.tenant_id
         );
-
-        let tenant_conf_guard = self.tenant_conf.read().unwrap();
-        let wal_connect_timeout = tenant_conf_guard
-            .walreceiver_connect_timeout
-            .unwrap_or(self.conf.default_tenant_conf.walreceiver_connect_timeout);
-        let lagging_wal_timeout = tenant_conf_guard
-            .lagging_wal_timeout
-            .unwrap_or(self.conf.default_tenant_conf.lagging_wal_timeout);
-        let max_lsn_wal_lag = tenant_conf_guard
-            .max_lsn_wal_lag
-            .unwrap_or(self.conf.default_tenant_conf.max_lsn_wal_lag);
-        drop(tenant_conf_guard);
-
-        let mut guard = self.walreceiver.lock().unwrap();
-        assert!(
-            guard.is_none(),
-            "multiple launches / re-launches of WAL receiver are not supported"
-        );
-        *guard = Some(WalReceiver::start(
-            Arc::clone(self),
-            WalReceiverConf {
-                wal_connect_timeout,
-                lagging_wal_timeout,
-                max_lsn_wal_lag,
-                auth_token: crate::config::SAFEKEEPER_AUTH_TOKEN.get().cloned(),
-                availability_zone: self.conf.availability_zone.clone(),
-            },
-            broker_client,
-            ctx,
-        ));
+        self.walreceiver.start(ctx, broker_client)?;
+        Ok(())
     }
 
     ///
@@ -2928,30 +2904,6 @@ impl Timeline {
         let layers = self.layers.read().unwrap();
 
         let mut max_deltas = 0;
-        {
-            let wanted_image_layers = self.wanted_image_layers.lock().unwrap();
-            if let Some((cutoff_lsn, wanted)) = &*wanted_image_layers {
-                let img_range =
-                    partition.ranges.first().unwrap().start..partition.ranges.last().unwrap().end;
-                if wanted.overlaps(&img_range) {
-                    //
-                    // gc_timeline only pays attention to image layers that are older than the GC cutoff,
-                    // but create_image_layers creates image layers at last-record-lsn.
-                    // So it's possible that gc_timeline wants a new image layer to be created for a key range,
-                    // but the range is already covered by image layers at more recent LSNs. Before we
-                    // create a new image layer, check if the range is already covered at more recent LSNs.
-                    if !layers
-                        .image_layer_exists(&img_range, &(Lsn::min(lsn, *cutoff_lsn)..lsn + 1))?
-                    {
-                        debug!(
-                            "Force generation of layer {}-{} wanted by GC, cutoff={}, lsn={})",
-                            img_range.start, img_range.end, cutoff_lsn, lsn
-                        );
-                        return Ok(true);
-                    }
-                }
-            }
-        }
 
         for part_range in &partition.ranges {
             let image_coverage = layers.image_coverage(part_range, lsn)?;
@@ -3071,12 +3023,6 @@ impl Timeline {
                 image_layers.push(image_layer);
             }
         }
-        // All layers that the GC wanted us to create have now been created.
-        //
-        // It's possible that another GC cycle happened while we were compacting, and added
-        // something new to wanted_image_layers, and we now clear that before processing it.
-        // That's OK, because the next GC iteration will put it back in.
-        *self.wanted_image_layers.lock().unwrap() = None;
 
         // Sync the new layer to disk before adding it to the layer map, to make sure
         // we don't garbage collect something based on the new layer, before it has
@@ -3774,7 +3720,6 @@ impl Timeline {
         }
 
         let mut layers_to_remove = Vec::new();
-        let mut wanted_image_layers = KeySpaceRandomAccum::default();
 
         // Scan all layers in the timeline (remote or on-disk).
         //
@@ -3858,15 +3803,6 @@ impl Timeline {
                     "keeping {} because it is the latest layer",
                     l.filename().file_name()
                 );
-                // Collect delta key ranges that need image layers to allow garbage
-                // collecting the layers.
-                // It is not so obvious whether we need to propagate information only about
-                // delta layers. Image layers can form "stairs" preventing old image from been deleted.
-                // But image layers are in any case less sparse than delta layers. Also we need some
-                // protection from replacing recent image layers with new one after each GC iteration.
-                if l.is_incremental() && !LayerMap::is_l0(&*l) {
-                    wanted_image_layers.add_range(l.get_key_range());
-                }
                 result.layers_not_updated += 1;
                 continue 'outer;
             }
@@ -3879,10 +3815,6 @@ impl Timeline {
             );
             layers_to_remove.push(Arc::clone(&l));
         }
-        self.wanted_image_layers
-            .lock()
-            .unwrap()
-            .replace((new_gc_cutoff, wanted_image_layers.to_keyspace()));
 
         let mut updates = layers.batch_update();
         if !layers_to_remove.is_empty() {

pageserver/src/tenant/timeline/walreceiver.rs

@@ -29,14 +29,16 @@ use crate::tenant::timeline::walreceiver::connection_manager::{
     connection_manager_loop_step, ConnectionManagerState,
 };
 
+use anyhow::Context;
 use std::future::Future;
 use std::num::NonZeroU64;
 use std::ops::ControlFlow;
-use std::sync::Arc;
+use std::sync::atomic::{self, AtomicBool};
+use std::sync::{Arc, Weak};
 use std::time::Duration;
 use storage_broker::BrokerClientChannel;
 use tokio::select;
-use tokio::sync::watch;
+use tokio::sync::{watch, RwLock};
 use tokio_util::sync::CancellationToken;
 use tracing::*;
 
@@ -60,23 +62,46 @@ pub struct WalReceiverConf {
 
 pub struct WalReceiver {
     timeline: TenantTimelineId,
-    manager_status: Arc<std::sync::RwLock<Option<ConnectionManagerStatus>>>,
+    timeline_ref: Weak<Timeline>,
+    conf: WalReceiverConf,
+    started: AtomicBool,
+    manager_status: Arc<RwLock<Option<ConnectionManagerStatus>>>,
 }
 
 impl WalReceiver {
-    pub fn start(
-        timeline: Arc<Timeline>,
+    pub fn new(
+        timeline: TenantTimelineId,
+        timeline_ref: Weak<Timeline>,
         conf: WalReceiverConf,
-        mut broker_client: BrokerClientChannel,
-        ctx: &RequestContext,
     ) -> Self {
+        Self {
+            timeline,
+            timeline_ref,
+            conf,
+            started: AtomicBool::new(false),
+            manager_status: Arc::new(RwLock::new(None)),
+        }
+    }
+
+    pub fn start(
+        &self,
+        ctx: &RequestContext,
+        mut broker_client: BrokerClientChannel,
+    ) -> anyhow::Result<()> {
+        if self.started.load(atomic::Ordering::Acquire) {
+            anyhow::bail!("Wal receiver is already started");
+        }
+
+        let timeline = self.timeline_ref.upgrade().with_context(|| {
+            format!("walreceiver start on a dropped timeline {}", self.timeline)
+        })?;
+
         let tenant_id = timeline.tenant_id;
         let timeline_id = timeline.timeline_id;
         let walreceiver_ctx =
             ctx.detached_child(TaskKind::WalReceiverManager, DownloadBehavior::Error);
-
-        let loop_status = Arc::new(std::sync::RwLock::new(None));
-        let manager_status = Arc::clone(&loop_status);
+        let wal_receiver_conf = self.conf.clone();
+        let loop_status = Arc::clone(&self.manager_status);
         task_mgr::spawn(
             WALRECEIVER_RUNTIME.handle(),
             TaskKind::WalReceiverManager,
@@ -88,7 +113,7 @@ impl WalReceiver {
                 info!("WAL receiver manager started, connecting to broker");
                 let mut connection_manager_state = ConnectionManagerState::new(
                     timeline,
-                    conf,
+                    wal_receiver_conf,
                 );
                 loop {
                     select! {
@@ -112,29 +137,29 @@ impl WalReceiver {
                 }
 
                 connection_manager_state.shutdown().await;
-                *loop_status.write().unwrap() = None;
+                *loop_status.write().await = None;
                 Ok(())
             }
             .instrument(info_span!(parent: None, "wal_connection_manager", tenant = %tenant_id, timeline = %timeline_id))
         );
 
-        Self {
-            timeline: TenantTimelineId::new(tenant_id, timeline_id),
-            manager_status,
-        }
+        self.started.store(true, atomic::Ordering::Release);
+
+        Ok(())
     }
 
-    pub async fn stop(self) {
+    pub async fn stop(&self) {
         task_mgr::shutdown_tasks(
             Some(TaskKind::WalReceiverManager),
             Some(self.timeline.tenant_id),
             Some(self.timeline.timeline_id),
         )
         .await;
+        self.started.store(false, atomic::Ordering::Release);
     }
 
-    pub(super) fn status(&self) -> Option<ConnectionManagerStatus> {
-        self.manager_status.read().unwrap().clone()
+    pub(super) async fn status(&self) -> Option<ConnectionManagerStatus> {
+        self.manager_status.read().await.clone()
     }
 }
 

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

@@ -29,6 +29,7 @@ use storage_broker::proto::TenantTimelineId as ProtoTenantTimelineId;
 use storage_broker::BrokerClientChannel;
 use storage_broker::Streaming;
 use tokio::select;
+use tokio::sync::RwLock;
 use tracing::*;
 
 use crate::{exponential_backoff, DEFAULT_BASE_BACKOFF_SECONDS, DEFAULT_MAX_BACKOFF_SECONDS};
@@ -47,7 +48,7 @@ pub(super) async fn connection_manager_loop_step(
     broker_client: &mut BrokerClientChannel,
     connection_manager_state: &mut ConnectionManagerState,
     ctx: &RequestContext,
-    manager_status: &std::sync::RwLock<Option<ConnectionManagerStatus>>,
+    manager_status: &RwLock<Option<ConnectionManagerStatus>>,
 ) -> ControlFlow<(), ()> {
     match connection_manager_state
         .timeline
@@ -194,7 +195,7 @@ pub(super) async fn connection_manager_loop_step(
                 .change_connection(new_candidate, ctx)
                 .await
         }
-        *manager_status.write().unwrap() = Some(connection_manager_state.manager_status());
+        *manager_status.write().await = Some(connection_manager_state.manager_status());
     }
 }
 
@@ -1308,9 +1309,9 @@ mod tests {
     async fn dummy_state(harness: &TenantHarness<'_>) -> ConnectionManagerState {
         let (tenant, ctx) = harness.load().await;
         let timeline = tenant
-            .create_test_timeline(TIMELINE_ID, Lsn(0), crate::DEFAULT_PG_VERSION, &ctx)
-            .await
+            .create_empty_timeline(TIMELINE_ID, Lsn(0), crate::DEFAULT_PG_VERSION, &ctx)
             .expect("Failed to create an empty timeline for dummy wal connection manager");
+        let timeline = timeline.initialize(&ctx).unwrap();
 
         ConnectionManagerState {
             id: TenantTimelineId {

pageserver/src/walingest.rs

@@ -1209,7 +1209,7 @@ mod tests {
     #[tokio::test]
     async fn test_relsize() -> Result<()> {
         let (tenant, ctx) = TenantHarness::create("test_relsize")?.load().await;
-        let tline = create_test_timeline(&tenant, TIMELINE_ID, DEFAULT_PG_VERSION, &ctx).await?;
+        let tline = create_test_timeline(&tenant, TIMELINE_ID, DEFAULT_PG_VERSION, &ctx)?;
         let mut walingest = init_walingest_test(&tline, &ctx).await?;
 
         let mut m = tline.begin_modification(Lsn(0x20));
@@ -1428,7 +1428,7 @@ mod tests {
     #[tokio::test]
     async fn test_drop_extend() -> Result<()> {
         let (tenant, ctx) = TenantHarness::create("test_drop_extend")?.load().await;
-        let tline = create_test_timeline(&tenant, TIMELINE_ID, DEFAULT_PG_VERSION, &ctx).await?;
+        let tline = create_test_timeline(&tenant, TIMELINE_ID, DEFAULT_PG_VERSION, &ctx)?;
         let mut walingest = init_walingest_test(&tline, &ctx).await?;
 
         let mut m = tline.begin_modification(Lsn(0x20));
@@ -1497,7 +1497,7 @@ mod tests {
     #[tokio::test]
     async fn test_truncate_extend() -> Result<()> {
         let (tenant, ctx) = TenantHarness::create("test_truncate_extend")?.load().await;
-        let tline = create_test_timeline(&tenant, TIMELINE_ID, DEFAULT_PG_VERSION, &ctx).await?;
+        let tline = create_test_timeline(&tenant, TIMELINE_ID, DEFAULT_PG_VERSION, &ctx)?;
         let mut walingest = init_walingest_test(&tline, &ctx).await?;
 
         // Create a 20 MB relation (the size is arbitrary)
@@ -1637,7 +1637,7 @@ mod tests {
     #[tokio::test]
     async fn test_large_rel() -> Result<()> {
         let (tenant, ctx) = TenantHarness::create("test_large_rel")?.load().await;
-        let tline = create_test_timeline(&tenant, TIMELINE_ID, DEFAULT_PG_VERSION, &ctx).await?;
+        let tline = create_test_timeline(&tenant, TIMELINE_ID, DEFAULT_PG_VERSION, &ctx)?;
         let mut walingest = init_walingest_test(&tline, &ctx).await?;
 
         let mut lsn = 0x10;

poetry.lock

@@ -1,4 +1,4 @@
-# This file is automatically @generated by Poetry 1.4.2 and should not be changed by hand.
+# This file is automatically @generated by Poetry and should not be changed by hand.
 
 [[package]]
 name = "aiohttp"
@@ -79,30 +79,30 @@ sa = ["sqlalchemy[postgresql-psycopg2binary] (>=1.3,<1.5)"]
 
 [[package]]
 name = "allure-pytest"
-version = "2.13.2"
+version = "2.13.1"
 description = "Allure pytest integration"
 category = "main"
 optional = false
 python-versions = "*"
 files = [
-    {file = "allure-pytest-2.13.2.tar.gz", hash = "sha256:22243159e8ec81ce2b5254b4013802198821b1b42f118f69d4a289396607c7b3"},
-    {file = "allure_pytest-2.13.2-py3-none-any.whl", hash = "sha256:17de9dbee7f61c8e66a5b5e818b00e419dbcea44cb55c24319401ba813220690"},
+    {file = "allure-pytest-2.13.1.tar.gz", hash = "sha256:68d69456eeb65af4061ec06a80bc941163b0616e8216554d36b070a6bf070e08"},
+    {file = "allure_pytest-2.13.1-py3-none-any.whl", hash = "sha256:a8de2fc3b3effe2d8f98801646920de3f055b779710f4c806dbee7c613c24633"},
 ]
 
 [package.dependencies]
-allure-python-commons = "2.13.2"
+allure-python-commons = "2.13.1"
 pytest = ">=4.5.0"
 
 [[package]]
 name = "allure-python-commons"
-version = "2.13.2"
+version = "2.13.1"
 description = "Common module for integrate allure with python-based frameworks"
 category = "main"
 optional = false
 python-versions = ">=3.6"
 files = [
-    {file = "allure-python-commons-2.13.2.tar.gz", hash = "sha256:8a03681330231b1deadd86b97ff68841c6591320114ae638570f1ed60d7a2033"},
-    {file = "allure_python_commons-2.13.2-py3-none-any.whl", hash = "sha256:2bb3646ec3fbf5b36d178a5e735002bc130ae9f9ba80f080af97d368ba375051"},
+    {file = "allure-python-commons-2.13.1.tar.gz", hash = "sha256:3fc13e1da8ebb23f9ab5c9c72ad04595023cdd5078dbb8604939997faebed5cb"},
+    {file = "allure_python_commons-2.13.1-py3-none-any.whl", hash = "sha256:d08e04867bddf44fef55def3d67f4bc25af58a1bf9fcffcf4ec3331f7f2ef0d0"},
 ]
 
 [package.dependencies]
@@ -172,6 +172,17 @@ dev = ["Cython (>=0.29.24,<0.30.0)", "Sphinx (>=4.1.2,<4.2.0)", "flake8 (>=5.0.4
 docs = ["Sphinx (>=4.1.2,<4.2.0)", "sphinx-rtd-theme (>=0.5.2,<0.6.0)", "sphinxcontrib-asyncio (>=0.3.0,<0.4.0)"]
 test = ["flake8 (>=5.0.4,<5.1.0)", "uvloop (>=0.15.3)"]
 
+[[package]]
+name = "atomicwrites"
+version = "1.4.1"
+description = "Atomic file writes."
+category = "main"
+optional = false
+python-versions = ">=2.7, !=3.0.*, !=3.1.*, !=3.2.*, !=3.3.*"
+files = [
+    {file = "atomicwrites-1.4.1.tar.gz", hash = "sha256:81b2c9071a49367a7f770170e5eec8cb66567cfbbc8c73d20ce5ca4a8d71cf11"},
+]
+
 [[package]]
 name = "attrs"
 version = "21.4.0"
@@ -228,49 +239,49 @@ wrapt = "*"
 
 [[package]]
 name = "backoff"
-version = "2.2.1"
+version = "1.11.1"
 description = "Function decoration for backoff and retry"
 category = "main"
 optional = false
-python-versions = ">=3.7,<4.0"
+python-versions = ">=2.7, !=3.0.*, !=3.1.*, !=3.2.*, !=3.3.*, !=3.4.*"
 files = [
-    {file = "backoff-2.2.1-py3-none-any.whl", hash = "sha256:63579f9a0628e06278f7e47b7d7d5b6ce20dc65c5e96a6f3ca99a6adca0396e8"},
-    {file = "backoff-2.2.1.tar.gz", hash = "sha256:03f829f5bb1923180821643f8753b0502c3b682293992485b0eef2807afa5cba"},
+    {file = "backoff-1.11.1-py2.py3-none-any.whl", hash = "sha256:61928f8fa48d52e4faa81875eecf308eccfb1016b018bb6bd21e05b5d90a96c5"},
+    {file = "backoff-1.11.1.tar.gz", hash = "sha256:ccb962a2378418c667b3c979b504fdeb7d9e0d29c0579e3b13b86467177728cb"},
 ]
 
 [[package]]
 name = "black"
-version = "23.3.0"
+version = "23.1.0"
 description = "The uncompromising code formatter."
 category = "dev"
 optional = false
 python-versions = ">=3.7"
 files = [
-    {file = "black-23.3.0-cp310-cp310-macosx_10_16_arm64.whl", hash = "sha256:0945e13506be58bf7db93ee5853243eb368ace1c08a24c65ce108986eac65915"},
-    {file = "black-23.3.0-cp310-cp310-macosx_10_16_universal2.whl", hash = "sha256:67de8d0c209eb5b330cce2469503de11bca4085880d62f1628bd9972cc3366b9"},
-    {file = "black-23.3.0-cp310-cp310-macosx_10_16_x86_64.whl", hash = "sha256:7c3eb7cea23904399866c55826b31c1f55bbcd3890ce22ff70466b907b6775c2"},
-    {file = "black-23.3.0-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:32daa9783106c28815d05b724238e30718f34155653d4d6e125dc7daec8e260c"},
-    {file = "black-23.3.0-cp310-cp310-win_amd64.whl", hash = "sha256:35d1381d7a22cc5b2be2f72c7dfdae4072a3336060635718cc7e1ede24221d6c"},
-    {file = "black-23.3.0-cp311-cp311-macosx_10_16_arm64.whl", hash = "sha256:a8a968125d0a6a404842fa1bf0b349a568634f856aa08ffaff40ae0dfa52e7c6"},
-    {file = "black-23.3.0-cp311-cp311-macosx_10_16_universal2.whl", hash = "sha256:c7ab5790333c448903c4b721b59c0d80b11fe5e9803d8703e84dcb8da56fec1b"},
-    {file = "black-23.3.0-cp311-cp311-macosx_10_16_x86_64.whl", hash = "sha256:a6f6886c9869d4daae2d1715ce34a19bbc4b95006d20ed785ca00fa03cba312d"},
-    {file = "black-23.3.0-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:6f3c333ea1dd6771b2d3777482429864f8e258899f6ff05826c3a4fcc5ce3f70"},
-    {file = "black-23.3.0-cp311-cp311-win_amd64.whl", hash = "sha256:11c410f71b876f961d1de77b9699ad19f939094c3a677323f43d7a29855fe326"},
-    {file = "black-23.3.0-cp37-cp37m-macosx_10_16_x86_64.whl", hash = "sha256:1d06691f1eb8de91cd1b322f21e3bfc9efe0c7ca1f0e1eb1db44ea367dff656b"},
-    {file = "black-23.3.0-cp37-cp37m-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:50cb33cac881766a5cd9913e10ff75b1e8eb71babf4c7104f2e9c52da1fb7de2"},
-    {file = "black-23.3.0-cp37-cp37m-win_amd64.whl", hash = "sha256:e114420bf26b90d4b9daa597351337762b63039752bdf72bf361364c1aa05925"},
-    {file = "black-23.3.0-cp38-cp38-macosx_10_16_arm64.whl", hash = "sha256:48f9d345675bb7fbc3dd85821b12487e1b9a75242028adad0333ce36ed2a6d27"},
-    {file = "black-23.3.0-cp38-cp38-macosx_10_16_universal2.whl", hash = "sha256:714290490c18fb0126baa0fca0a54ee795f7502b44177e1ce7624ba1c00f2331"},
-    {file = "black-23.3.0-cp38-cp38-macosx_10_16_x86_64.whl", hash = "sha256:064101748afa12ad2291c2b91c960be28b817c0c7eaa35bec09cc63aa56493c5"},
-    {file = "black-23.3.0-cp38-cp38-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:562bd3a70495facf56814293149e51aa1be9931567474993c7942ff7d3533961"},
-    {file = "black-23.3.0-cp38-cp38-win_amd64.whl", hash = "sha256:e198cf27888ad6f4ff331ca1c48ffc038848ea9f031a3b40ba36aced7e22f2c8"},
-    {file = "black-23.3.0-cp39-cp39-macosx_10_16_arm64.whl", hash = "sha256:3238f2aacf827d18d26db07524e44741233ae09a584273aa059066d644ca7b30"},
-    {file = "black-23.3.0-cp39-cp39-macosx_10_16_universal2.whl", hash = "sha256:f0bd2f4a58d6666500542b26354978218a9babcdc972722f4bf90779524515f3"},
-    {file = "black-23.3.0-cp39-cp39-macosx_10_16_x86_64.whl", hash = "sha256:92c543f6854c28a3c7f39f4d9b7694f9a6eb9d3c5e2ece488c327b6e7ea9b266"},
-    {file = "black-23.3.0-cp39-cp39-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:3a150542a204124ed00683f0db1f5cf1c2aaaa9cc3495b7a3b5976fb136090ab"},
-    {file = "black-23.3.0-cp39-cp39-win_amd64.whl", hash = "sha256:6b39abdfb402002b8a7d030ccc85cf5afff64ee90fa4c5aebc531e3ad0175ddb"},
-    {file = "black-23.3.0-py3-none-any.whl", hash = "sha256:ec751418022185b0c1bb7d7736e6933d40bbb14c14a0abcf9123d1b159f98dd4"},
-    {file = "black-23.3.0.tar.gz", hash = "sha256:1c7b8d606e728a41ea1ccbd7264677e494e87cf630e399262ced92d4a8dac940"},
+    {file = "black-23.1.0-cp310-cp310-macosx_10_16_arm64.whl", hash = "sha256:b6a92a41ee34b883b359998f0c8e6eb8e99803aa8bf3123bf2b2e6fec505a221"},
+    {file = "black-23.1.0-cp310-cp310-macosx_10_16_universal2.whl", hash = "sha256:57c18c5165c1dbe291d5306e53fb3988122890e57bd9b3dcb75f967f13411a26"},
+    {file = "black-23.1.0-cp310-cp310-macosx_10_16_x86_64.whl", hash = "sha256:9880d7d419bb7e709b37e28deb5e68a49227713b623c72b2b931028ea65f619b"},
+    {file = "black-23.1.0-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:e6663f91b6feca5d06f2ccd49a10f254f9298cc1f7f49c46e498a0771b507104"},
+    {file = "black-23.1.0-cp310-cp310-win_amd64.whl", hash = "sha256:9afd3f493666a0cd8f8df9a0200c6359ac53940cbde049dcb1a7eb6ee2dd7074"},
+    {file = "black-23.1.0-cp311-cp311-macosx_10_16_arm64.whl", hash = "sha256:bfffba28dc52a58f04492181392ee380e95262af14ee01d4bc7bb1b1c6ca8d27"},
+    {file = "black-23.1.0-cp311-cp311-macosx_10_16_universal2.whl", hash = "sha256:c1c476bc7b7d021321e7d93dc2cbd78ce103b84d5a4cf97ed535fbc0d6660648"},
+    {file = "black-23.1.0-cp311-cp311-macosx_10_16_x86_64.whl", hash = "sha256:382998821f58e5c8238d3166c492139573325287820963d2f7de4d518bd76958"},
+    {file = "black-23.1.0-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:2bf649fda611c8550ca9d7592b69f0637218c2369b7744694c5e4902873b2f3a"},
+    {file = "black-23.1.0-cp311-cp311-win_amd64.whl", hash = "sha256:121ca7f10b4a01fd99951234abdbd97728e1240be89fde18480ffac16503d481"},
+    {file = "black-23.1.0-cp37-cp37m-macosx_10_16_x86_64.whl", hash = "sha256:a8471939da5e824b891b25751955be52ee7f8a30a916d570a5ba8e0f2eb2ecad"},
+    {file = "black-23.1.0-cp37-cp37m-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:8178318cb74f98bc571eef19068f6ab5613b3e59d4f47771582f04e175570ed8"},
+    {file = "black-23.1.0-cp37-cp37m-win_amd64.whl", hash = "sha256:a436e7881d33acaf2536c46a454bb964a50eff59b21b51c6ccf5a40601fbef24"},
+    {file = "black-23.1.0-cp38-cp38-macosx_10_16_arm64.whl", hash = "sha256:a59db0a2094d2259c554676403fa2fac3473ccf1354c1c63eccf7ae65aac8ab6"},
+    {file = "black-23.1.0-cp38-cp38-macosx_10_16_universal2.whl", hash = "sha256:0052dba51dec07ed029ed61b18183942043e00008ec65d5028814afaab9a22fd"},
+    {file = "black-23.1.0-cp38-cp38-macosx_10_16_x86_64.whl", hash = "sha256:49f7b39e30f326a34b5c9a4213213a6b221d7ae9d58ec70df1c4a307cf2a1580"},
+    {file = "black-23.1.0-cp38-cp38-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:162e37d49e93bd6eb6f1afc3e17a3d23a823042530c37c3c42eeeaf026f38468"},
+    {file = "black-23.1.0-cp38-cp38-win_amd64.whl", hash = "sha256:8b70eb40a78dfac24842458476135f9b99ab952dd3f2dab738c1881a9b38b753"},
+    {file = "black-23.1.0-cp39-cp39-macosx_10_16_arm64.whl", hash = "sha256:a29650759a6a0944e7cca036674655c2f0f63806ddecc45ed40b7b8aa314b651"},
+    {file = "black-23.1.0-cp39-cp39-macosx_10_16_universal2.whl", hash = "sha256:bb460c8561c8c1bec7824ecbc3ce085eb50005883a6203dcfb0122e95797ee06"},
+    {file = "black-23.1.0-cp39-cp39-macosx_10_16_x86_64.whl", hash = "sha256:c91dfc2c2a4e50df0026f88d2215e166616e0c80e86004d0003ece0488db2739"},
+    {file = "black-23.1.0-cp39-cp39-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:2a951cc83ab535d248c89f300eccbd625e80ab880fbcfb5ac8afb5f01a258ac9"},
+    {file = "black-23.1.0-cp39-cp39-win_amd64.whl", hash = "sha256:0680d4380db3719ebcfb2613f34e86c8e6d15ffeabcf8ec59355c5e7b85bb555"},
+    {file = "black-23.1.0-py3-none-any.whl", hash = "sha256:7a0f701d314cfa0896b9001df70a530eb2472babb76086344e688829efd97d32"},
+    {file = "black-23.1.0.tar.gz", hash = "sha256:b0bd97bea8903f5a2ba7219257a44e3f1f9d00073d6cc1add68f0beec69692ac"},
 ]
 
 [package.dependencies]
@@ -940,21 +951,6 @@ six = ">=1.9.0"
 gmpy = ["gmpy"]
 gmpy2 = ["gmpy2"]
 
-[[package]]
-name = "exceptiongroup"
-version = "1.1.1"
-description = "Backport of PEP 654 (exception groups)"
-category = "main"
-optional = false
-python-versions = ">=3.7"
-files = [
-    {file = "exceptiongroup-1.1.1-py3-none-any.whl", hash = "sha256:232c37c63e4f682982c8b6459f33a8981039e5fb8756b2074364e5055c498c9e"},
-    {file = "exceptiongroup-1.1.1.tar.gz", hash = "sha256:d484c3090ba2889ae2928419117447a14daf3c1231d5e30d0aae34f354f01785"},
-]
-
-[package.extras]
-test = ["pytest (>=6)"]
-
 [[package]]
 name = "execnet"
 version = "1.9.0"
@@ -1414,38 +1410,38 @@ files = [
 
 [[package]]
 name = "mypy"
-version = "1.3.0"
+version = "1.1.1"
 description = "Optional static typing for Python"
 category = "dev"
 optional = false
 python-versions = ">=3.7"
 files = [
-    {file = "mypy-1.3.0-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:c1eb485cea53f4f5284e5baf92902cd0088b24984f4209e25981cc359d64448d"},
-    {file = "mypy-1.3.0-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:4c99c3ecf223cf2952638da9cd82793d8f3c0c5fa8b6ae2b2d9ed1e1ff51ba85"},
-    {file = "mypy-1.3.0-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:550a8b3a19bb6589679a7c3c31f64312e7ff482a816c96e0cecec9ad3a7564dd"},
-    {file = "mypy-1.3.0-cp310-cp310-musllinux_1_1_x86_64.whl", hash = "sha256:cbc07246253b9e3d7d74c9ff948cd0fd7a71afcc2b77c7f0a59c26e9395cb152"},
-    {file = "mypy-1.3.0-cp310-cp310-win_amd64.whl", hash = "sha256:a22435632710a4fcf8acf86cbd0d69f68ac389a3892cb23fbad176d1cddaf228"},
-    {file = "mypy-1.3.0-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:6e33bb8b2613614a33dff70565f4c803f889ebd2f859466e42b46e1df76018dd"},
-    {file = "mypy-1.3.0-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:7d23370d2a6b7a71dc65d1266f9a34e4cde9e8e21511322415db4b26f46f6b8c"},
-    {file = "mypy-1.3.0-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:658fe7b674769a0770d4b26cb4d6f005e88a442fe82446f020be8e5f5efb2fae"},
-    {file = "mypy-1.3.0-cp311-cp311-musllinux_1_1_x86_64.whl", hash = "sha256:6e42d29e324cdda61daaec2336c42512e59c7c375340bd202efa1fe0f7b8f8ca"},
-    {file = "mypy-1.3.0-cp311-cp311-win_amd64.whl", hash = "sha256:d0b6c62206e04061e27009481cb0ec966f7d6172b5b936f3ead3d74f29fe3dcf"},
-    {file = "mypy-1.3.0-cp37-cp37m-macosx_10_9_x86_64.whl", hash = "sha256:76ec771e2342f1b558c36d49900dfe81d140361dd0d2df6cd71b3db1be155409"},
-    {file = "mypy-1.3.0-cp37-cp37m-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:ebc95f8386314272bbc817026f8ce8f4f0d2ef7ae44f947c4664efac9adec929"},
-    {file = "mypy-1.3.0-cp37-cp37m-musllinux_1_1_x86_64.whl", hash = "sha256:faff86aa10c1aa4a10e1a301de160f3d8fc8703b88c7e98de46b531ff1276a9a"},
-    {file = "mypy-1.3.0-cp37-cp37m-win_amd64.whl", hash = "sha256:8c5979d0deb27e0f4479bee18ea0f83732a893e81b78e62e2dda3e7e518c92ee"},
-    {file = "mypy-1.3.0-cp38-cp38-macosx_10_9_x86_64.whl", hash = "sha256:c5d2cc54175bab47011b09688b418db71403aefad07cbcd62d44010543fc143f"},
-    {file = "mypy-1.3.0-cp38-cp38-macosx_11_0_arm64.whl", hash = "sha256:87df44954c31d86df96c8bd6e80dfcd773473e877ac6176a8e29898bfb3501cb"},
-    {file = "mypy-1.3.0-cp38-cp38-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:473117e310febe632ddf10e745a355714e771ffe534f06db40702775056614c4"},
-    {file = "mypy-1.3.0-cp38-cp38-musllinux_1_1_x86_64.whl", hash = "sha256:74bc9b6e0e79808bf8678d7678b2ae3736ea72d56eede3820bd3849823e7f305"},
-    {file = "mypy-1.3.0-cp38-cp38-win_amd64.whl", hash = "sha256:44797d031a41516fcf5cbfa652265bb994e53e51994c1bd649ffcd0c3a7eccbf"},
-    {file = "mypy-1.3.0-cp39-cp39-macosx_10_9_x86_64.whl", hash = "sha256:ddae0f39ca146972ff6bb4399f3b2943884a774b8771ea0a8f50e971f5ea5ba8"},
-    {file = "mypy-1.3.0-cp39-cp39-macosx_11_0_arm64.whl", hash = "sha256:1c4c42c60a8103ead4c1c060ac3cdd3ff01e18fddce6f1016e08939647a0e703"},
-    {file = "mypy-1.3.0-cp39-cp39-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:e86c2c6852f62f8f2b24cb7a613ebe8e0c7dc1402c61d36a609174f63e0ff017"},
-    {file = "mypy-1.3.0-cp39-cp39-musllinux_1_1_x86_64.whl", hash = "sha256:f9dca1e257d4cc129517779226753dbefb4f2266c4eaad610fc15c6a7e14283e"},
-    {file = "mypy-1.3.0-cp39-cp39-win_amd64.whl", hash = "sha256:95d8d31a7713510685b05fbb18d6ac287a56c8f6554d88c19e73f724a445448a"},
-    {file = "mypy-1.3.0-py3-none-any.whl", hash = "sha256:a8763e72d5d9574d45ce5881962bc8e9046bf7b375b0abf031f3e6811732a897"},
-    {file = "mypy-1.3.0.tar.gz", hash = "sha256:e1f4d16e296f5135624b34e8fb741eb0eadedca90862405b1f1fde2040b9bd11"},
+    {file = "mypy-1.1.1-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:39c7119335be05630611ee798cc982623b9e8f0cff04a0b48dfc26100e0b97af"},
+    {file = "mypy-1.1.1-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:61bf08362e93b6b12fad3eab68c4ea903a077b87c90ac06c11e3d7a09b56b9c1"},
+    {file = "mypy-1.1.1-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:dbb19c9f662e41e474e0cff502b7064a7edc6764f5262b6cd91d698163196799"},
+    {file = "mypy-1.1.1-cp310-cp310-musllinux_1_1_x86_64.whl", hash = "sha256:315ac73cc1cce4771c27d426b7ea558fb4e2836f89cb0296cbe056894e3a1f78"},
+    {file = "mypy-1.1.1-cp310-cp310-win_amd64.whl", hash = "sha256:5cb14ff9919b7df3538590fc4d4c49a0f84392237cbf5f7a816b4161c061829e"},
+    {file = "mypy-1.1.1-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:26cdd6a22b9b40b2fd71881a8a4f34b4d7914c679f154f43385ca878a8297389"},
+    {file = "mypy-1.1.1-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:5b5f81b40d94c785f288948c16e1f2da37203c6006546c5d947aab6f90aefef2"},
+    {file = "mypy-1.1.1-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:21b437be1c02712a605591e1ed1d858aba681757a1e55fe678a15c2244cd68a5"},
+    {file = "mypy-1.1.1-cp311-cp311-musllinux_1_1_x86_64.whl", hash = "sha256:d809f88734f44a0d44959d795b1e6f64b2bbe0ea4d9cc4776aa588bb4229fc1c"},
+    {file = "mypy-1.1.1-cp311-cp311-win_amd64.whl", hash = "sha256:a380c041db500e1410bb5b16b3c1c35e61e773a5c3517926b81dfdab7582be54"},
+    {file = "mypy-1.1.1-cp37-cp37m-macosx_10_9_x86_64.whl", hash = "sha256:b7c7b708fe9a871a96626d61912e3f4ddd365bf7f39128362bc50cbd74a634d5"},
+    {file = "mypy-1.1.1-cp37-cp37m-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:c1c10fa12df1232c936830839e2e935d090fc9ee315744ac33b8a32216b93707"},
+    {file = "mypy-1.1.1-cp37-cp37m-musllinux_1_1_x86_64.whl", hash = "sha256:0a28a76785bf57655a8ea5eb0540a15b0e781c807b5aa798bd463779988fa1d5"},
+    {file = "mypy-1.1.1-cp37-cp37m-win_amd64.whl", hash = "sha256:ef6a01e563ec6a4940784c574d33f6ac1943864634517984471642908b30b6f7"},
+    {file = "mypy-1.1.1-cp38-cp38-macosx_10_9_x86_64.whl", hash = "sha256:d64c28e03ce40d5303450f547e07418c64c241669ab20610f273c9e6290b4b0b"},
+    {file = "mypy-1.1.1-cp38-cp38-macosx_11_0_arm64.whl", hash = "sha256:64cc3afb3e9e71a79d06e3ed24bb508a6d66f782aff7e56f628bf35ba2e0ba51"},
+    {file = "mypy-1.1.1-cp38-cp38-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:ce61663faf7a8e5ec6f456857bfbcec2901fbdb3ad958b778403f63b9e606a1b"},
+    {file = "mypy-1.1.1-cp38-cp38-musllinux_1_1_x86_64.whl", hash = "sha256:2b0c373d071593deefbcdd87ec8db91ea13bd8f1328d44947e88beae21e8d5e9"},
+    {file = "mypy-1.1.1-cp38-cp38-win_amd64.whl", hash = "sha256:2888ce4fe5aae5a673386fa232473014056967f3904f5abfcf6367b5af1f612a"},
+    {file = "mypy-1.1.1-cp39-cp39-macosx_10_9_x86_64.whl", hash = "sha256:19ba15f9627a5723e522d007fe708007bae52b93faab00f95d72f03e1afa9598"},
+    {file = "mypy-1.1.1-cp39-cp39-macosx_11_0_arm64.whl", hash = "sha256:59bbd71e5c58eed2e992ce6523180e03c221dcd92b52f0e792f291d67b15a71c"},
+    {file = "mypy-1.1.1-cp39-cp39-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:9401e33814cec6aec8c03a9548e9385e0e228fc1b8b0a37b9ea21038e64cdd8a"},
+    {file = "mypy-1.1.1-cp39-cp39-musllinux_1_1_x86_64.whl", hash = "sha256:4b398d8b1f4fba0e3c6463e02f8ad3346f71956b92287af22c9b12c3ec965a9f"},
+    {file = "mypy-1.1.1-cp39-cp39-win_amd64.whl", hash = "sha256:69b35d1dcb5707382810765ed34da9db47e7f95b3528334a3c999b0c90fe523f"},
+    {file = "mypy-1.1.1-py3-none-any.whl", hash = "sha256:4e4e8b362cdf99ba00c2b218036002bdcdf1e0de085cdb296a49df03fb31dfc4"},
+    {file = "mypy-1.1.1.tar.gz", hash = "sha256:ae9ceae0f5b9059f33dbc62dea087e942c0ccab4b7a003719cb70f9b8abfa32f"},
 ]
 
 [package.dependencies]
@@ -1725,6 +1721,18 @@ files = [
     {file = "psycopg2_binary-2.9.3-cp39-cp39-win_amd64.whl", hash = "sha256:accfe7e982411da3178ec690baaceaad3c278652998b2c45828aaac66cd8285f"},
 ]
 
+[[package]]
+name = "py"
+version = "1.11.0"
+description = "library with cross-python path, ini-parsing, io, code, log facilities"
+category = "main"
+optional = false
+python-versions = ">=2.7, !=3.0.*, !=3.1.*, !=3.2.*, !=3.3.*, !=3.4.*"
+files = [
+    {file = "py-1.11.0-py2.py3-none-any.whl", hash = "sha256:607c53218732647dff4acdfcd50cb62615cedf612e72d1724fb1a0cc6405b378"},
+    {file = "py-1.11.0.tar.gz", hash = "sha256:51c75c4126074b472f746a24399ad32f6053d1b34b68d2fa41e558e6f4a98719"},
+]
+
 [[package]]
 name = "pyasn1"
 version = "0.4.8"
@@ -1833,56 +1841,57 @@ files = [
 
 [[package]]
 name = "pytest"
-version = "7.3.1"
+version = "6.2.5"
 description = "pytest: simple powerful testing with Python"
 category = "main"
 optional = false
-python-versions = ">=3.7"
+python-versions = ">=3.6"
 files = [
-    {file = "pytest-7.3.1-py3-none-any.whl", hash = "sha256:3799fa815351fea3a5e96ac7e503a96fa51cc9942c3753cda7651b93c1cfa362"},
-    {file = "pytest-7.3.1.tar.gz", hash = "sha256:434afafd78b1d78ed0addf160ad2b77a30d35d4bdf8af234fe621919d9ed15e3"},
+    {file = "pytest-6.2.5-py3-none-any.whl", hash = "sha256:7310f8d27bc79ced999e760ca304d69f6ba6c6649c0b60fb0e04a4a77cacc134"},
+    {file = "pytest-6.2.5.tar.gz", hash = "sha256:131b36680866a76e6781d13f101efb86cf674ebb9762eb70d3082b6f29889e89"},
 ]
 
 [package.dependencies]
+atomicwrites = {version = ">=1.0", markers = "sys_platform == \"win32\""}
+attrs = ">=19.2.0"
 colorama = {version = "*", markers = "sys_platform == \"win32\""}
-exceptiongroup = {version = ">=1.0.0rc8", markers = "python_version < \"3.11\""}
 iniconfig = "*"
 packaging = "*"
 pluggy = ">=0.12,<2.0"
-tomli = {version = ">=1.0.0", markers = "python_version < \"3.11\""}
+py = ">=1.8.2"
+toml = "*"
 
 [package.extras]
-testing = ["argcomplete", "attrs (>=19.2.0)", "hypothesis (>=3.56)", "mock", "nose", "pygments (>=2.7.2)", "requests", "xmlschema"]
+testing = ["argcomplete", "hypothesis (>=3.56)", "mock", "nose", "requests", "xmlschema"]
 
 [[package]]
 name = "pytest-asyncio"
-version = "0.21.0"
+version = "0.19.0"
 description = "Pytest support for asyncio"
 category = "main"
 optional = false
 python-versions = ">=3.7"
 files = [
-    {file = "pytest-asyncio-0.21.0.tar.gz", hash = "sha256:2b38a496aef56f56b0e87557ec313e11e1ab9276fc3863f6a7be0f1d0e415e1b"},
-    {file = "pytest_asyncio-0.21.0-py3-none-any.whl", hash = "sha256:f2b3366b7cd501a4056858bd39349d5af19742aed2d81660b7998b6341c7eb9c"},
+    {file = "pytest-asyncio-0.19.0.tar.gz", hash = "sha256:ac4ebf3b6207259750bc32f4c1d8fcd7e79739edbc67ad0c58dd150b1d072fed"},
+    {file = "pytest_asyncio-0.19.0-py3-none-any.whl", hash = "sha256:7a97e37cfe1ed296e2e84941384bdd37c376453912d397ed39293e0916f521fa"},
 ]
 
 [package.dependencies]
-pytest = ">=7.0.0"
+pytest = ">=6.1.0"
 
 [package.extras]
-docs = ["sphinx (>=5.3)", "sphinx-rtd-theme (>=1.0)"]
 testing = ["coverage (>=6.2)", "flaky (>=3.5.0)", "hypothesis (>=5.7.1)", "mypy (>=0.931)", "pytest-trio (>=0.7.0)"]
 
 [[package]]
 name = "pytest-httpserver"
-version = "1.0.8"
+version = "1.0.6"
 description = "pytest-httpserver is a httpserver for pytest"
 category = "main"
 optional = false
-python-versions = ">=3.8,<4.0"
+python-versions = ">=3.7,<4.0"
 files = [
-    {file = "pytest_httpserver-1.0.8-py3-none-any.whl", hash = "sha256:24cd3d9f6a0b927c7bfc400d0b3fda7442721b8267ce29942bf307b190f0bb09"},
-    {file = "pytest_httpserver-1.0.8.tar.gz", hash = "sha256:e052f69bc8a9073db02484681e8e47004dd1fb3763b0ae833bd899e5895c559a"},
+    {file = "pytest_httpserver-1.0.6-py3-none-any.whl", hash = "sha256:ac2379acc91fe8bdbe2911c93af8dd130e33b5899fb9934d15669480739c6d32"},
+    {file = "pytest_httpserver-1.0.6.tar.gz", hash = "sha256:9040d07bf59ac45d8de3db1d4468fd2d1d607975e4da4c872ecc0402cdbf7b3e"},
 ]
 
 [package.dependencies]
@@ -1905,14 +1914,14 @@ pytest = ">=3.2.5"
 
 [[package]]
 name = "pytest-order"
-version = "1.1.0"
+version = "1.0.1"
 description = "pytest plugin to run your tests in a specific order"
 category = "main"
 optional = false
 python-versions = ">=3.6"
 files = [
-    {file = "pytest-order-1.1.0.tar.gz", hash = "sha256:139d25b30826b78eebb42722f747eab14c44b88059d7a71d4f79d14a057269a5"},
-    {file = "pytest_order-1.1.0-py3-none-any.whl", hash = "sha256:3b3730969c97900fa5cd31ecff80847680ed56b2490954565c14949ba60d9371"},
+    {file = "pytest-order-1.0.1.tar.gz", hash = "sha256:5dd6b929fbd7eaa6d0ee07586f65c623babb0afe72b4843c5f15055d6b3b1b1f"},
+    {file = "pytest_order-1.0.1-py3-none-any.whl", hash = "sha256:bbe6e63a8e23741ab3e810d458d1ea7317e797b70f9550512d77d6e9e8fd1bbb"},
 ]
 
 [package.dependencies]
@@ -1954,14 +1963,14 @@ pytest = ">=5.0.0"
 
 [[package]]
 name = "pytest-xdist"
-version = "3.3.1"
-description = "pytest xdist plugin for distributed testing, most importantly across multiple CPUs"
+version = "3.0.2"
+description = "pytest xdist plugin for distributed testing and loop-on-failing modes"
 category = "main"
 optional = false
-python-versions = ">=3.7"
+python-versions = ">=3.6"
 files = [
-    {file = "pytest-xdist-3.3.1.tar.gz", hash = "sha256:d5ee0520eb1b7bcca50a60a518ab7a7707992812c578198f8b44fdfac78e8c93"},
-    {file = "pytest_xdist-3.3.1-py3-none-any.whl", hash = "sha256:ff9daa7793569e6a68544850fd3927cd257cc03a7ef76c95e86915355e82b5f2"},
+    {file = "pytest-xdist-3.0.2.tar.gz", hash = "sha256:688da9b814370e891ba5de650c9327d1a9d861721a524eb917e620eec3e90291"},
+    {file = "pytest_xdist-3.0.2-py3-none-any.whl", hash = "sha256:9feb9a18e1790696ea23e1434fa73b325ed4998b0e9fcb221f16fd1945e6df1b"},
 ]
 
 [package.dependencies]
@@ -2083,21 +2092,21 @@ files = [
 
 [[package]]
 name = "requests"
-version = "2.31.0"
+version = "2.28.1"
 description = "Python HTTP for Humans."
 category = "main"
 optional = false
-python-versions = ">=3.7"
+python-versions = ">=3.7, <4"
 files = [
-    {file = "requests-2.31.0-py3-none-any.whl", hash = "sha256:58cd2187c01e70e6e26505bca751777aa9f2ee0b7f4300988b709f44e013003f"},
-    {file = "requests-2.31.0.tar.gz", hash = "sha256:942c5a758f98d790eaed1a29cb6eefc7ffb0d1cf7af05c3d2791656dbd6ad1e1"},
+    {file = "requests-2.28.1-py3-none-any.whl", hash = "sha256:8fefa2a1a1365bf5520aac41836fbee479da67864514bdb821f31ce07ce65349"},
+    {file = "requests-2.28.1.tar.gz", hash = "sha256:7c5599b102feddaa661c826c56ab4fee28bfd17f5abca1ebbe3e7f19d7c97983"},
 ]
 
 [package.dependencies]
 certifi = ">=2017.4.17"
-charset-normalizer = ">=2,<4"
+charset-normalizer = ">=2,<3"
 idna = ">=2.5,<4"
-urllib3 = ">=1.21.1,<3"
+urllib3 = ">=1.21.1,<1.27"
 
 [package.extras]
 socks = ["PySocks (>=1.5.6,!=1.5.7)"]
@@ -2139,29 +2148,29 @@ pyasn1 = ">=0.1.3"
 
 [[package]]
 name = "ruff"
-version = "0.0.269"
+version = "0.0.255"
 description = "An extremely fast Python linter, written in Rust."
 category = "dev"
 optional = false
 python-versions = ">=3.7"
 files = [
-    {file = "ruff-0.0.269-py3-none-macosx_10_7_x86_64.whl", hash = "sha256:3569bcdee679045c09c0161fabc057599759c49219a08d9a4aad2cc3982ccba3"},
-    {file = "ruff-0.0.269-py3-none-macosx_10_9_x86_64.macosx_11_0_arm64.macosx_10_9_universal2.whl", hash = "sha256:56347da63757a56cbce7d4b3d6044ca4f1941cd1bbff3714f7554360c3361f83"},
-    {file = "ruff-0.0.269-py3-none-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:6da8ee25ef2f0cc6cc8e6e20942c1d44d25a36dce35070d7184655bc14f63f63"},
-    {file = "ruff-0.0.269-py3-none-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:bd81b8e681b9eaa6cf15484f3985bd8bd97c3d114e95bff3e8ea283bf8865062"},
-    {file = "ruff-0.0.269-py3-none-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:1f19f59ca3c28742955241fb452f3346241ddbd34e72ac5cb3d84fadebcf6bc8"},
-    {file = "ruff-0.0.269-py3-none-manylinux_2_17_ppc64.manylinux2014_ppc64.whl", hash = "sha256:f062059b8289a4fab7f6064601b811d447c2f9d3d432a17f689efe4d68988450"},
-    {file = "ruff-0.0.269-py3-none-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:3f5dc7aac52c58e82510217e3c7efd80765c134c097c2815d59e40face0d1fe6"},
-    {file = "ruff-0.0.269-py3-none-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:e131b4dbe798c391090c6407641d6ab12c0fa1bb952379dde45e5000e208dabb"},
-    {file = "ruff-0.0.269-py3-none-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:a374434e588e06550df0f8dcb74777290f285678de991fda4e1063c367ab2eb2"},
-    {file = "ruff-0.0.269-py3-none-musllinux_1_2_aarch64.whl", hash = "sha256:cec2f4b84a14b87f1b121488649eb5b4eaa06467a2387373f750da74bdcb5679"},
-    {file = "ruff-0.0.269-py3-none-musllinux_1_2_armv7l.whl", hash = "sha256:374b161753a247904aec7a32d45e165302b76b6e83d22d099bf3ff7c232c888f"},
-    {file = "ruff-0.0.269-py3-none-musllinux_1_2_i686.whl", hash = "sha256:9ca0a1ddb1d835b5f742db9711c6cf59f213a1ad0088cb1e924a005fd399e7d8"},
-    {file = "ruff-0.0.269-py3-none-musllinux_1_2_x86_64.whl", hash = "sha256:5a20658f0b97d207c7841c13d528f36d666bf445b00b01139f28a8ccb80093bb"},
-    {file = "ruff-0.0.269-py3-none-win32.whl", hash = "sha256:03ff42bc91ceca58e0f0f072cb3f9286a9208f609812753474e799a997cdad1a"},
-    {file = "ruff-0.0.269-py3-none-win_amd64.whl", hash = "sha256:f3b59ccff57b21ef0967ea8021fd187ec14c528ec65507d8bcbe035912050776"},
-    {file = "ruff-0.0.269-py3-none-win_arm64.whl", hash = "sha256:bbeb857b1e508a4487bdb02ca1e6d41dd8d5ac5335a5246e25de8a3dff38c1ff"},
-    {file = "ruff-0.0.269.tar.gz", hash = "sha256:11ddcfbab32cf5c420ea9dd5531170ace5a3e59c16d9251c7bd2581f7b16f602"},
+    {file = "ruff-0.0.255-py3-none-macosx_10_7_x86_64.whl", hash = "sha256:b2d71fb6a7e50501a2473864acffc85dee6b750c25db198f7e71fe1dbbff1aad"},
+    {file = "ruff-0.0.255-py3-none-macosx_10_9_x86_64.macosx_11_0_arm64.macosx_10_9_universal2.whl", hash = "sha256:6c97d746861a6010f941179e84bba9feb8a871815667471d9ed6beb98d45c252"},
+    {file = "ruff-0.0.255-py3-none-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:9a7fa60085079b91a298b963361be9b1b1c724582af6c84be954cbabdbd9309a"},
+    {file = "ruff-0.0.255-py3-none-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:c089f7141496334ab5a127b54ce55e41f0d6714e68a4453a1e09d2204cdea8c3"},
+    {file = "ruff-0.0.255-py3-none-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:0423908caa7d437a416b853214565b9c33bbd1106c4f88147982216dddcbbd96"},
+    {file = "ruff-0.0.255-py3-none-manylinux_2_17_ppc64.manylinux2014_ppc64.whl", hash = "sha256:981493e92547cacbb8e0874904ec049fe744507ee890dc8736caf89a8864f9a7"},
+    {file = "ruff-0.0.255-py3-none-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:59d5193d2aedb35db180824462b374dbcfc306b2e76076245088afa6e5837df2"},
+    {file = "ruff-0.0.255-py3-none-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:dd5e00733c9d160c8a34a22e62b390da9d1e9f326676402421cb8c1236beefc3"},
+    {file = "ruff-0.0.255-py3-none-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:694418cf41838bd19c6229e4e1b2d04505b1e6b86fe3ab81165484fc96d36f01"},
+    {file = "ruff-0.0.255-py3-none-musllinux_1_2_aarch64.whl", hash = "sha256:5d0408985c9777369daebb5d3340a99e9f7294bdd7120642239261508185cf89"},
+    {file = "ruff-0.0.255-py3-none-musllinux_1_2_armv7l.whl", hash = "sha256:abd6376ef9d12f370d95a8c7c98682fbb9bfedfba59f40e84a816fef8ddcb8de"},
+    {file = "ruff-0.0.255-py3-none-musllinux_1_2_i686.whl", hash = "sha256:f9b1a5df0bc09193cbef58a6f78e4a9a0b058a4f9733c0442866d078006d1bb9"},
+    {file = "ruff-0.0.255-py3-none-musllinux_1_2_x86_64.whl", hash = "sha256:6a25c5f4ff087445b2e1bbcb9963f2ae7c868d65e4a8d5f84c36c12f71571179"},
+    {file = "ruff-0.0.255-py3-none-win32.whl", hash = "sha256:1ff87a8310354f9f1a099625e54a27fdd6756d9cd2a40b45922f2e943daf982d"},
+    {file = "ruff-0.0.255-py3-none-win_amd64.whl", hash = "sha256:f3d8416be618f023f93ec4fd6ee3048585ef85dba9563b2a7e38fc7e5131d5b1"},
+    {file = "ruff-0.0.255-py3-none-win_arm64.whl", hash = "sha256:8ba124819624145d7b6b53add40c367c44318893215ffc1bfe3d72e0225a1c9c"},
+    {file = "ruff-0.0.255.tar.gz", hash = "sha256:f9eb1d3b2eecbeedae419fa494c4e2a5e4484baf93a1ce0f81eddb005e1919c5"},
 ]
 
 [[package]]
@@ -2262,7 +2271,7 @@ files = [
 name = "tomli"
 version = "2.0.1"
 description = "A lil' TOML parser"
-category = "main"
+category = "dev"
 optional = false
 python-versions = ">=3.7"
 files = [
@@ -2272,54 +2281,42 @@ files = [
 
 [[package]]
 name = "types-psutil"
-version = "5.9.5.12"
+version = "5.9.5.4"
 description = "Typing stubs for psutil"
 category = "main"
 optional = false
 python-versions = "*"
 files = [
-    {file = "types-psutil-5.9.5.12.tar.gz", hash = "sha256:61a91679d3fe737250013b624dca09375e7cc3ad77dcc734553746c429c02aca"},
-    {file = "types_psutil-5.9.5.12-py3-none-any.whl", hash = "sha256:e9a147b8561235c6afcce5aa1adb973fad9ab2c50cf89820697687f53510358f"},
+    {file = "types-psutil-5.9.5.4.tar.gz", hash = "sha256:aa09102b80c65a3b4573216614372398dab78972d650488eaff1ff05482cc18f"},
+    {file = "types_psutil-5.9.5.4-py3-none-any.whl", hash = "sha256:28e59764630187e462d43788efa16d59d5e77b510115f9e25901b2d4007fca62"},
 ]
 
 [[package]]
 name = "types-psycopg2"
-version = "2.9.21.10"
+version = "2.9.18"
 description = "Typing stubs for psycopg2"
 category = "main"
 optional = false
 python-versions = "*"
 files = [
-    {file = "types-psycopg2-2.9.21.10.tar.gz", hash = "sha256:c2600892312ae1c34e12f145749795d93dc4eac3ef7dbf8a9c1bfd45385e80d7"},
-    {file = "types_psycopg2-2.9.21.10-py3-none-any.whl", hash = "sha256:918224a0731a3650832e46633e720703b5beef7693a064e777d9748654fcf5e5"},
-]
-
-[[package]]
-name = "types-pytest-lazy-fixture"
-version = "0.6.3.3"
-description = "Typing stubs for pytest-lazy-fixture"
-category = "main"
-optional = false
-python-versions = "*"
-files = [
-    {file = "types-pytest-lazy-fixture-0.6.3.3.tar.gz", hash = "sha256:2ef79d66bcde0e50acdac8dc55074b9ae0d4cfaeabdd638f5522f4cac7c8a2c7"},
-    {file = "types_pytest_lazy_fixture-0.6.3.3-py3-none-any.whl", hash = "sha256:a56a55649147ff960ff79d4b2c781a4f769351abc1876873f3116d0bd0c96353"},
+    {file = "types-psycopg2-2.9.18.tar.gz", hash = "sha256:9b0e9e1f097b15cd9fa8aad2596a9e3082fd72f8d9cfe52b190cfa709105b6c0"},
+    {file = "types_psycopg2-2.9.18-py3-none-any.whl", hash = "sha256:14c779dcab18c31453fa1cad3cf4b1601d33540a344adead3c47a6b8091cd2fa"},
 ]
 
 [[package]]
 name = "types-requests"
-version = "2.31.0.0"
+version = "2.28.5"
 description = "Typing stubs for requests"
 category = "main"
 optional = false
 python-versions = "*"
 files = [
-    {file = "types-requests-2.31.0.0.tar.gz", hash = "sha256:c1c29d20ab8d84dff468d7febfe8e0cb0b4664543221b386605e14672b44ea25"},
-    {file = "types_requests-2.31.0.0-py3-none-any.whl", hash = "sha256:7c5cea7940f8e92ec560bbc468f65bf684aa3dcf0554a6f8c4710f5f708dc598"},
+    {file = "types-requests-2.28.5.tar.gz", hash = "sha256:ac618bfefcb3742eaf97c961e13e9e5a226e545eda4a3dbe293b898d40933ad1"},
+    {file = "types_requests-2.28.5-py3-none-any.whl", hash = "sha256:98ab647ae88b5e2c41d6d20cfcb5117da1bea561110000b6fdeeea07b3e89877"},
 ]
 
 [package.dependencies]
-types-urllib3 = "*"
+types-urllib3 = "<1.27"
 
 [[package]]
 name = "types-s3transfer"
@@ -2335,14 +2332,14 @@ files = [
 
 [[package]]
 name = "types-toml"
-version = "0.10.8.6"
+version = "0.10.8"
 description = "Typing stubs for toml"
 category = "main"
 optional = false
 python-versions = "*"
 files = [
-    {file = "types-toml-0.10.8.6.tar.gz", hash = "sha256:6d3ac79e36c9ee593c5d4fb33a50cca0e3adceb6ef5cff8b8e5aef67b4c4aaf2"},
-    {file = "types_toml-0.10.8.6-py3-none-any.whl", hash = "sha256:de7b2bb1831d6f7a4b554671ffe5875e729753496961b3e9b202745e4955dafa"},
+    {file = "types-toml-0.10.8.tar.gz", hash = "sha256:b7e7ea572308b1030dc86c3ba825c5210814c2825612ec679eb7814f8dd9295a"},
+    {file = "types_toml-0.10.8-py3-none-any.whl", hash = "sha256:8300fd093e5829eb9c1fba69cee38130347d4b74ddf32d0a7df650ae55c2b599"},
 ]
 
 [[package]]
@@ -2359,14 +2356,14 @@ files = [
 
 [[package]]
 name = "typing-extensions"
-version = "4.6.1"
+version = "4.3.0"
 description = "Backported and Experimental Type Hints for Python 3.7+"
 category = "main"
 optional = false
 python-versions = ">=3.7"
 files = [
-    {file = "typing_extensions-4.6.1-py3-none-any.whl", hash = "sha256:6bac751f4789b135c43228e72de18637e9a6c29d12777023a703fd1a6858469f"},
-    {file = "typing_extensions-4.6.1.tar.gz", hash = "sha256:558bc0c4145f01e6405f4a5fdbd82050bd221b119f4bf72a961a1cfd471349d6"},
+    {file = "typing_extensions-4.3.0-py3-none-any.whl", hash = "sha256:25642c956049920a5aa49edcdd6ab1e06d7e5d467fc00e0506c44ac86fbfca02"},
+    {file = "typing_extensions-4.3.0.tar.gz", hash = "sha256:e6d2677a32f47fc7eb2795db1dd15c1f34eff616bcaf2cfb5e997f854fa1c4a6"},
 ]
 
 [[package]]
@@ -2614,4 +2611,4 @@ testing = ["func-timeout", "jaraco.itertools", "pytest (>=6)", "pytest-black (>=
 [metadata]
 lock-version = "2.0"
 python-versions = "^3.9"
-content-hash = "c6c217033f50430c31b0979b74db222e6bab2301abd8b9f0cce5a9d5bccc578f"
+content-hash = "b689ffd6eae32b966f1744b5ac3343fe0dd26b31ee1f50e13daf5045ee0623e1"

proxy/README.md

@@ -1,6 +1,6 @@
 # Proxy
 
-Proxy binary accepts `--auth-backend` CLI option, which determines auth scheme and cluster routing method. Following routing backends are currently implemented:
+Proxy binary accepts `--auth-backend` CLI option, which determines auth scheme and cluster routing method. Following backends are currently implemented:
 
 * console
   new SCRAM-based console API; uses SNI info to select the destination project (endpoint soon)
@@ -9,90 +9,6 @@ Proxy binary accepts `--auth-backend` CLI option, which determines auth scheme a
 * link
   sends login link for all usernames
 
-Also proxy can expose following services to the external world:
-
-* postgres protocol over TCP -- usual postgres endpoint compatible with usual
-  postgres drivers
-* postgres protocol over WebSockets -- same protocol tunneled over websockets
-  for environments where TCP connection is not available. We have our own
-  implementation of a client that uses node-postgres and tunnels traffic through
-  websockets: https://github.com/neondatabase/serverless
-* SQL over HTTP -- service that accepts POST requests with SQL text over HTTP
-  and responds with JSON-serialised results.
-
-
-## SQL over HTTP
-
-Contrary to the usual postgres proto over TCP and WebSockets using plain
-one-shot HTTP request achieves smaller amortized latencies in edge setups due to
-fewer round trips and an enhanced open connection reuse by the v8 engine. Also
-such endpoint could be used directly without any driver.
-
-To play with it locally one may start proxy over a local postgres installation
-(see end of this page on how to generate certs with openssl):
-
-```
-./target/debug/proxy -c server.crt -k server.key --auth-backend=postgres --auth-endpoint=postgres://stas@127.0.0.1:5432/stas --wss 0.0.0.0:4444
-```
-
-If both postgres and proxy are running you may send a SQL query:
-```json
-curl -k -X POST 'https://proxy.localtest.me:4444/sql' \
-  -H 'Neon-Connection-String: postgres://stas:pass@proxy.localtest.me:4444/postgres' \
-  -H 'Content-Type: application/json' \
-  --data '{
-    "query":"SELECT $1::int[] as arr, $2::jsonb as obj, 42 as num",
-    "params":[ "{{1,2},{\"3\",4}}", {"key":"val", "ikey":4242}]
-  }' | jq
-
-{
-  "command": "SELECT",
-  "fields": [
-    { "dataTypeID": 1007, "name": "arr" },
-    { "dataTypeID": 3802, "name": "obj" },
-    { "dataTypeID": 23, "name": "num" }
-  ],
-  "rowCount": 1,
-  "rows": [
-    {
-      "arr": [[1,2],[3,4]],
-      "num": 42,
-      "obj": {
-        "ikey": 4242,
-        "key": "val"
-      }
-    }
-  ]
-}
-```
-
-
-With the current approach we made the following design decisions:
-
-1. SQL injection protection: We employed the extended query protocol, modifying
-   the rust-postgres driver to send queries in one roundtrip using a text
-   protocol rather than binary, bypassing potential issues like those identified
-   in sfackler/rust-postgres#1030.
-
-2. Postgres type compatibility: As not all postgres types have binary
-   representations (e.g., acl's in pg_class), we adjusted rust-postgres to
-   respond with text protocol, simplifying serialization and fixing queries with
-   text-only types in response.
-
-3. Data type conversion: Considering JSON supports fewer data types than
-   Postgres, we perform conversions where possible, passing all other types as
-   strings. Key conversions include:
-   - postgres int2, int4, float4, float8 -> json number (NaN and Inf remain
-     text)
-   - postgres bool, null, text -> json bool, null, string
-   - postgres array -> json array
-   - postgres json and jsonb -> json object
-
-4. Alignment with node-postgres: To facilitate integration with js libraries,
-   we've matched the response structure of node-postgres, returning command tags
-   and column oids. Command tag capturing was added to the rust-postgres
-   functionality as part of this change.
-
 ## Using SNI-based routing on localhost
 
 Now proxy determines project name from the subdomain, request to the `round-rice-566201.somedomain.tld` will be routed to the project named `round-rice-566201`. Unfortunately, `/etc/hosts` does not support domain wildcards, so I usually use `*.localtest.me` which resolves to `127.0.0.1`. Now we can create self-signed certificate and play with proxy:

proxy/src/auth/backend.rs

@@ -139,16 +139,6 @@ async fn auth_quirks(
 }
 
 impl BackendType<'_, ClientCredentials<'_>> {
-    /// Get compute endpoint name from the credentials.
-    pub fn get_endpoint(&self) -> Option<String> {
-        use BackendType::*;
-
-        match self {
-            Console(_, creds) => creds.project.clone(),
-            Postgres(_, creds) => creds.project.clone(),
-            Link(_) => Some("link".to_owned()),
-        }
-    }
     /// Authenticate the client via the requested backend, possibly using credentials.
     #[tracing::instrument(fields(allow_cleartext = allow_cleartext), skip_all)]
     pub async fn authenticate(

proxy/src/config.rs

@@ -100,10 +100,9 @@ impl CertResolver {
         is_default: bool,
     ) -> anyhow::Result<()> {
         let priv_key = {
-            let key_bytes = std::fs::read(key_path)
-                .context(format!("Failed to read TLS keys at '{key_path}'"))?;
+            let key_bytes = std::fs::read(key_path).context("TLS key file")?;
             let mut keys = rustls_pemfile::pkcs8_private_keys(&mut &key_bytes[..])
-                .context(format!("Failed to parse TLS keys at '{key_path}'"))?;
+                .context(format!("Failed to read TLS keys at '{key_path}'"))?;
 
             ensure!(keys.len() == 1, "keys.len() = {} (should be 1)", keys.len());
             keys.pop().map(rustls::PrivateKey).unwrap()

proxy/src/http.rs

@@ -3,7 +3,6 @@
 //! directly relying on deps like `reqwest` (think loose coupling).
 
 pub mod server;
-pub mod sql_over_http;
 pub mod websocket;
 
 pub use reqwest::{Request, Response, StatusCode};

proxy/src/http/sql_over_http.rs

@@ -1,603 +0,0 @@
-use futures::pin_mut;
-use futures::StreamExt;
-use hyper::body::HttpBody;
-use hyper::{Body, HeaderMap, Request};
-use pq_proto::StartupMessageParams;
-use serde_json::json;
-use serde_json::Map;
-use serde_json::Value;
-use tokio_postgres::types::Kind;
-use tokio_postgres::types::Type;
-use tokio_postgres::Row;
-use url::Url;
-
-use crate::{auth, config::ProxyConfig, console};
-
-#[derive(serde::Deserialize)]
-struct QueryData {
-    query: String,
-    params: Vec<serde_json::Value>,
-}
-
-const APP_NAME: &str = "sql_over_http";
-const MAX_RESPONSE_SIZE: usize = 1024 * 1024; // 1 MB
-const MAX_REQUEST_SIZE: u64 = 1024 * 1024; // 1 MB
-
-//
-// Convert json non-string types to strings, so that they can be passed to Postgres
-// as parameters.
-//
-fn json_to_pg_text(json: Vec<Value>) -> Result<Vec<String>, serde_json::Error> {
-    json.iter()
-        .map(|value| {
-            match value {
-                Value::Null => serde_json::to_string(value),
-                Value::Bool(_) => serde_json::to_string(value),
-                Value::Number(_) => serde_json::to_string(value),
-                Value::Object(_) => serde_json::to_string(value),
-
-                // no need to escape
-                Value::String(s) => Ok(s.to_string()),
-
-                // special care for arrays
-                Value::Array(_) => json_array_to_pg_array(value),
-            }
-        })
-        .collect()
-}
-
-//
-// Serialize a JSON array to a Postgres array. Contrary to the strings in the params
-// in the array we need to escape the strings. Postgres is okay with arrays of form
-// '{1,"2",3}'::int[], so we don't check that array holds values of the same type, leaving
-// it for Postgres to check.
-//
-// Example of the same escaping in node-postgres: packages/pg/lib/utils.js
-//
-fn json_array_to_pg_array(value: &Value) -> Result<String, serde_json::Error> {
-    match value {
-        // same
-        Value::Null => serde_json::to_string(value),
-        Value::Bool(_) => serde_json::to_string(value),
-        Value::Number(_) => serde_json::to_string(value),
-        Value::Object(_) => serde_json::to_string(value),
-
-        // now needs to be escaped, as it is part of the array
-        Value::String(_) => serde_json::to_string(value),
-
-        // recurse into array
-        Value::Array(arr) => {
-            let vals = arr
-                .iter()
-                .map(json_array_to_pg_array)
-                .collect::<Result<Vec<_>, _>>()?
-                .join(",");
-            Ok(format!("{{{}}}", vals))
-        }
-    }
-}
-
-fn get_conn_info(
-    headers: &HeaderMap,
-    sni_hostname: Option<String>,
-) -> Result<(String, String, String, String), anyhow::Error> {
-    let connection_string = headers
-        .get("Neon-Connection-String")
-        .ok_or(anyhow::anyhow!("missing connection string"))?
-        .to_str()?;
-
-    let connection_url = Url::parse(connection_string)?;
-
-    let protocol = connection_url.scheme();
-    if protocol != "postgres" && protocol != "postgresql" {
-        return Err(anyhow::anyhow!(
-            "connection string must start with postgres: or postgresql:"
-        ));
-    }
-
-    let mut url_path = connection_url
-        .path_segments()
-        .ok_or(anyhow::anyhow!("missing database name"))?;
-
-    let dbname = url_path
-        .next()
-        .ok_or(anyhow::anyhow!("invalid database name"))?;
-
-    let username = connection_url.username();
-    if username.is_empty() {
-        return Err(anyhow::anyhow!("missing username"));
-    }
-
-    let password = connection_url
-        .password()
-        .ok_or(anyhow::anyhow!("no password"))?;
-
-    // TLS certificate selector now based on SNI hostname, so if we are running here
-    // we are sure that SNI hostname is set to one of the configured domain names.
-    let sni_hostname = sni_hostname.ok_or(anyhow::anyhow!("no SNI hostname set"))?;
-
-    let hostname = connection_url
-        .host_str()
-        .ok_or(anyhow::anyhow!("no host"))?;
-
-    let host_header = headers
-        .get("host")
-        .and_then(|h| h.to_str().ok())
-        .and_then(|h| h.split(':').next());
-
-    if hostname != sni_hostname {
-        return Err(anyhow::anyhow!("mismatched SNI hostname and hostname"));
-    } else if let Some(h) = host_header {
-        if h != hostname {
-            return Err(anyhow::anyhow!("mismatched host header and hostname"));
-        }
-    }
-
-    Ok((
-        username.to_owned(),
-        dbname.to_owned(),
-        hostname.to_owned(),
-        password.to_owned(),
-    ))
-}
-
-// TODO: return different http error codes
-pub async fn handle(
-    config: &'static ProxyConfig,
-    request: Request<Body>,
-    sni_hostname: Option<String>,
-) -> anyhow::Result<Value> {
-    //
-    // Determine the destination and connection params
-    //
-    let headers = request.headers();
-    let (username, dbname, hostname, password) = get_conn_info(headers, sni_hostname)?;
-    let credential_params = StartupMessageParams::new([
-        ("user", &username),
-        ("database", &dbname),
-        ("application_name", APP_NAME),
-    ]);
-
-    //
-    // Wake up the destination if needed. Code here is a bit involved because
-    // we reuse the code from the usual proxy and we need to prepare few structures
-    // that this code expects.
-    //
-    let tls = config.tls_config.as_ref();
-    let common_names = tls.and_then(|tls| tls.common_names.clone());
-    let creds = config
-        .auth_backend
-        .as_ref()
-        .map(|_| auth::ClientCredentials::parse(&credential_params, Some(&hostname), common_names))
-        .transpose()?;
-    let extra = console::ConsoleReqExtra {
-        session_id: uuid::Uuid::new_v4(),
-        application_name: Some(APP_NAME),
-    };
-    let node = creds.wake_compute(&extra).await?.expect("msg");
-    let conf = node.value.config;
-    let port = *conf.get_ports().first().expect("no port");
-    let host = match conf.get_hosts().first().expect("no host") {
-        tokio_postgres::config::Host::Tcp(host) => host,
-        tokio_postgres::config::Host::Unix(_) => {
-            return Err(anyhow::anyhow!("unix socket is not supported"));
-        }
-    };
-
-    let request_content_length = match request.body().size_hint().upper() {
-        Some(v) => v,
-        None => MAX_REQUEST_SIZE + 1,
-    };
-
-    if request_content_length > MAX_REQUEST_SIZE {
-        return Err(anyhow::anyhow!(
-            "request is too large (max {MAX_REQUEST_SIZE} bytes)"
-        ));
-    }
-
-    //
-    // Read the query and query params from the request body
-    //
-    let body = hyper::body::to_bytes(request.into_body()).await?;
-    let QueryData { query, params } = serde_json::from_slice(&body)?;
-    let query_params = json_to_pg_text(params)?;
-
-    //
-    // Connenct to the destination
-    //
-    let (client, connection) = tokio_postgres::Config::new()
-        .host(host)
-        .port(port)
-        .user(&username)
-        .password(&password)
-        .dbname(&dbname)
-        .max_backend_message_size(MAX_RESPONSE_SIZE)
-        .connect(tokio_postgres::NoTls)
-        .await?;
-
-    tokio::spawn(async move {
-        if let Err(e) = connection.await {
-            eprintln!("connection error: {}", e);
-        }
-    });
-
-    //
-    // Now execute the query and return the result
-    //
-    let row_stream = client.query_raw_txt(query, query_params).await?;
-
-    // Manually drain the stream into a vector to leave row_stream hanging
-    // around to get a command tag. Also check that the response is not too
-    // big.
-    pin_mut!(row_stream);
-    let mut rows: Vec<tokio_postgres::Row> = Vec::new();
-    let mut curret_size = 0;
-    while let Some(row) = row_stream.next().await {
-        let row = row?;
-        curret_size += row.body_len();
-        rows.push(row);
-        if curret_size > MAX_RESPONSE_SIZE {
-            return Err(anyhow::anyhow!("response too large"));
-        }
-    }
-
-    // grab the command tag and number of rows affected
-    let command_tag = row_stream.command_tag().unwrap_or_default();
-    let mut command_tag_split = command_tag.split(' ');
-    let command_tag_name = command_tag_split.next().unwrap_or_default();
-    let command_tag_count = if command_tag_name == "INSERT" {
-        // INSERT returns OID first and then number of rows
-        command_tag_split.nth(1)
-    } else {
-        // other commands return number of rows (if any)
-        command_tag_split.next()
-    }
-    .and_then(|s| s.parse::<i64>().ok());
-
-    let fields = if !rows.is_empty() {
-        rows[0]
-            .columns()
-            .iter()
-            .map(|c| {
-                json!({
-                    "name": Value::String(c.name().to_owned()),
-                    "dataTypeID": Value::Number(c.type_().oid().into()),
-                })
-            })
-            .collect::<Vec<_>>()
-    } else {
-        Vec::new()
-    };
-
-    // convert rows to JSON
-    let rows = rows
-        .iter()
-        .map(pg_text_row_to_json)
-        .collect::<Result<Vec<_>, _>>()?;
-
-    // resulting JSON format is based on the format of node-postgres result
-    Ok(json!({
-        "command": command_tag_name,
-        "rowCount": command_tag_count,
-        "rows": rows,
-        "fields": fields,
-    }))
-}
-
-//
-// Convert postgres row with text-encoded values to JSON object
-//
-pub fn pg_text_row_to_json(row: &Row) -> Result<Value, anyhow::Error> {
-    let res = row
-        .columns()
-        .iter()
-        .enumerate()
-        .map(|(i, column)| {
-            let name = column.name();
-            let pg_value = row.as_text(i)?;
-            let json_value = pg_text_to_json(pg_value, column.type_())?;
-            Ok((name.to_string(), json_value))
-        })
-        .collect::<Result<Map<String, Value>, anyhow::Error>>()?;
-
-    Ok(Value::Object(res))
-}
-
-//
-// Convert postgres text-encoded value to JSON value
-//
-pub fn pg_text_to_json(pg_value: Option<&str>, pg_type: &Type) -> Result<Value, anyhow::Error> {
-    if let Some(val) = pg_value {
-        if val == "NULL" {
-            return Ok(Value::Null);
-        }
-
-        if let Kind::Array(elem_type) = pg_type.kind() {
-            return pg_array_parse(val, elem_type);
-        }
-
-        match *pg_type {
-            Type::BOOL => Ok(Value::Bool(val == "t")),
-            Type::INT2 | Type::INT4 => {
-                let val = val.parse::<i32>()?;
-                Ok(Value::Number(serde_json::Number::from(val)))
-            }
-            Type::FLOAT4 | Type::FLOAT8 => {
-                let fval = val.parse::<f64>()?;
-                let num = serde_json::Number::from_f64(fval);
-                if let Some(num) = num {
-                    Ok(Value::Number(num))
-                } else {
-                    // Pass Nan, Inf, -Inf as strings
-                    // JS JSON.stringify() does converts them to null, but we
-                    // want to preserve them, so we pass them as strings
-                    Ok(Value::String(val.to_string()))
-                }
-            }
-            Type::JSON | Type::JSONB => Ok(serde_json::from_str(val)?),
-            _ => Ok(Value::String(val.to_string())),
-        }
-    } else {
-        Ok(Value::Null)
-    }
-}
-
-//
-// Parse postgres array into JSON array.
-//
-// This is a bit involved because we need to handle nested arrays and quoted
-// values. Unlike postgres we don't check that all nested arrays have the same
-// dimensions, we just return them as is.
-//
-fn pg_array_parse(pg_array: &str, elem_type: &Type) -> Result<Value, anyhow::Error> {
-    _pg_array_parse(pg_array, elem_type, false).map(|(v, _)| v)
-}
-
-fn _pg_array_parse(
-    pg_array: &str,
-    elem_type: &Type,
-    nested: bool,
-) -> Result<(Value, usize), anyhow::Error> {
-    let mut pg_array_chr = pg_array.char_indices();
-    let mut level = 0;
-    let mut quote = false;
-    let mut entries: Vec<Value> = Vec::new();
-    let mut entry = String::new();
-
-    // skip bounds decoration
-    if let Some('[') = pg_array.chars().next() {
-        for (_, c) in pg_array_chr.by_ref() {
-            if c == '=' {
-                break;
-            }
-        }
-    }
-
-    while let Some((mut i, mut c)) = pg_array_chr.next() {
-        let mut escaped = false;
-
-        if c == '\\' {
-            escaped = true;
-            (i, c) = pg_array_chr.next().unwrap();
-        }
-
-        match c {
-            '{' if !quote => {
-                level += 1;
-                if level > 1 {
-                    let (res, off) = _pg_array_parse(&pg_array[i..], elem_type, true)?;
-                    entries.push(res);
-                    for _ in 0..off - 1 {
-                        pg_array_chr.next();
-                    }
-                }
-            }
-            '}' => {
-                level -= 1;
-                if level == 0 {
-                    if !entry.is_empty() {
-                        entries.push(pg_text_to_json(Some(&entry), elem_type)?);
-                    }
-                    if nested {
-                        return Ok((Value::Array(entries), i));
-                    }
-                }
-            }
-            '"' if !escaped => {
-                if quote {
-                    // push even if empty
-                    entries.push(pg_text_to_json(Some(&entry), elem_type)?);
-                    entry = String::new();
-                }
-                quote = !quote;
-            }
-            ',' if !quote => {
-                if !entry.is_empty() {
-                    entries.push(pg_text_to_json(Some(&entry), elem_type)?);
-                    entry = String::new();
-                }
-            }
-            _ => {
-                entry.push(c);
-            }
-        }
-    }
-
-    if level != 0 {
-        return Err(anyhow::anyhow!("unbalanced array"));
-    }
-
-    Ok((Value::Array(entries), 0))
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use serde_json::json;
-
-    #[test]
-    fn test_atomic_types_to_pg_params() {
-        let json = vec![Value::Bool(true), Value::Bool(false)];
-        let pg_params = json_to_pg_text(json).unwrap();
-        assert_eq!(pg_params, vec!["true", "false"]);
-
-        let json = vec![Value::Number(serde_json::Number::from(42))];
-        let pg_params = json_to_pg_text(json).unwrap();
-        assert_eq!(pg_params, vec!["42"]);
-
-        let json = vec![Value::String("foo\"".to_string())];
-        let pg_params = json_to_pg_text(json).unwrap();
-        assert_eq!(pg_params, vec!["foo\""]);
-
-        let json = vec![Value::Null];
-        let pg_params = json_to_pg_text(json).unwrap();
-        assert_eq!(pg_params, vec!["null"]);
-    }
-
-    #[test]
-    fn test_json_array_to_pg_array() {
-        // atoms and escaping
-        let json = "[true, false, null, 42, \"foo\", \"bar\\\"-\\\\\"]";
-        let json: Value = serde_json::from_str(json).unwrap();
-        let pg_params = json_to_pg_text(vec![json]).unwrap();
-        assert_eq!(
-            pg_params,
-            vec!["{true,false,null,42,\"foo\",\"bar\\\"-\\\\\"}"]
-        );
-
-        // nested arrays
-        let json = "[[true, false], [null, 42], [\"foo\", \"bar\\\"-\\\\\"]]";
-        let json: Value = serde_json::from_str(json).unwrap();
-        let pg_params = json_to_pg_text(vec![json]).unwrap();
-        assert_eq!(
-            pg_params,
-            vec!["{{true,false},{null,42},{\"foo\",\"bar\\\"-\\\\\"}}"]
-        );
-    }
-
-    #[test]
-    fn test_atomic_types_parse() {
-        assert_eq!(
-            pg_text_to_json(Some("foo"), &Type::TEXT).unwrap(),
-            json!("foo")
-        );
-        assert_eq!(pg_text_to_json(None, &Type::TEXT).unwrap(), json!(null));
-        assert_eq!(pg_text_to_json(Some("42"), &Type::INT4).unwrap(), json!(42));
-        assert_eq!(pg_text_to_json(Some("42"), &Type::INT2).unwrap(), json!(42));
-        assert_eq!(
-            pg_text_to_json(Some("42"), &Type::INT8).unwrap(),
-            json!("42")
-        );
-        assert_eq!(
-            pg_text_to_json(Some("42.42"), &Type::FLOAT8).unwrap(),
-            json!(42.42)
-        );
-        assert_eq!(
-            pg_text_to_json(Some("42.42"), &Type::FLOAT4).unwrap(),
-            json!(42.42)
-        );
-        assert_eq!(
-            pg_text_to_json(Some("NaN"), &Type::FLOAT4).unwrap(),
-            json!("NaN")
-        );
-        assert_eq!(
-            pg_text_to_json(Some("Infinity"), &Type::FLOAT4).unwrap(),
-            json!("Infinity")
-        );
-        assert_eq!(
-            pg_text_to_json(Some("-Infinity"), &Type::FLOAT4).unwrap(),
-            json!("-Infinity")
-        );
-
-        let json: Value =
-            serde_json::from_str("{\"s\":\"str\",\"n\":42,\"f\":4.2,\"a\":[null,3,\"a\"]}")
-                .unwrap();
-        assert_eq!(
-            pg_text_to_json(
-                Some(r#"{"s":"str","n":42,"f":4.2,"a":[null,3,"a"]}"#),
-                &Type::JSONB
-            )
-            .unwrap(),
-            json
-        );
-    }
-
-    #[test]
-    fn test_pg_array_parse_text() {
-        fn pt(pg_arr: &str) -> Value {
-            pg_array_parse(pg_arr, &Type::TEXT).unwrap()
-        }
-        assert_eq!(
-            pt(r#"{"aa\"\\\,a",cha,"bbbb"}"#),
-            json!(["aa\"\\,a", "cha", "bbbb"])
-        );
-        assert_eq!(
-            pt(r#"{{"foo","bar"},{"bee","bop"}}"#),
-            json!([["foo", "bar"], ["bee", "bop"]])
-        );
-        assert_eq!(
-            pt(r#"{{{{"foo",NULL,"bop",bup}}}}"#),
-            json!([[[["foo", null, "bop", "bup"]]]])
-        );
-        assert_eq!(
-            pt(r#"{{"1",2,3},{4,NULL,6},{NULL,NULL,NULL}}"#),
-            json!([["1", "2", "3"], ["4", null, "6"], [null, null, null]])
-        );
-    }
-
-    #[test]
-    fn test_pg_array_parse_bool() {
-        fn pb(pg_arr: &str) -> Value {
-            pg_array_parse(pg_arr, &Type::BOOL).unwrap()
-        }
-        assert_eq!(pb(r#"{t,f,t}"#), json!([true, false, true]));
-        assert_eq!(pb(r#"{{t,f,t}}"#), json!([[true, false, true]]));
-        assert_eq!(
-            pb(r#"{{t,f},{f,t}}"#),
-            json!([[true, false], [false, true]])
-        );
-        assert_eq!(
-            pb(r#"{{t,NULL},{NULL,f}}"#),
-            json!([[true, null], [null, false]])
-        );
-    }
-
-    #[test]
-    fn test_pg_array_parse_numbers() {
-        fn pn(pg_arr: &str, ty: &Type) -> Value {
-            pg_array_parse(pg_arr, ty).unwrap()
-        }
-        assert_eq!(pn(r#"{1,2,3}"#, &Type::INT4), json!([1, 2, 3]));
-        assert_eq!(pn(r#"{1,2,3}"#, &Type::INT2), json!([1, 2, 3]));
-        assert_eq!(pn(r#"{1,2,3}"#, &Type::INT8), json!(["1", "2", "3"]));
-        assert_eq!(pn(r#"{1,2,3}"#, &Type::FLOAT4), json!([1.0, 2.0, 3.0]));
-        assert_eq!(pn(r#"{1,2,3}"#, &Type::FLOAT8), json!([1.0, 2.0, 3.0]));
-        assert_eq!(
-            pn(r#"{1.1,2.2,3.3}"#, &Type::FLOAT4),
-            json!([1.1, 2.2, 3.3])
-        );
-        assert_eq!(
-            pn(r#"{1.1,2.2,3.3}"#, &Type::FLOAT8),
-            json!([1.1, 2.2, 3.3])
-        );
-        assert_eq!(
-            pn(r#"{NaN,Infinity,-Infinity}"#, &Type::FLOAT4),
-            json!(["NaN", "Infinity", "-Infinity"])
-        );
-        assert_eq!(
-            pn(r#"{NaN,Infinity,-Infinity}"#, &Type::FLOAT8),
-            json!(["NaN", "Infinity", "-Infinity"])
-        );
-    }
-
-    #[test]
-    fn test_pg_array_with_decoration() {
-        fn p(pg_arr: &str) -> Value {
-            pg_array_parse(pg_arr, &Type::INT2).unwrap()
-        }
-        assert_eq!(
-            p(r#"[1:1][-2:-1][3:5]={{{1,2,3},{4,5,6}}}"#),
-            json!([[[1, 2, 3], [4, 5, 6]]])
-        );
-    }
-}

proxy/src/http/websocket.rs

@@ -4,17 +4,12 @@ use crate::{
 use bytes::{Buf, Bytes};
 use futures::{Sink, Stream, StreamExt};
 use hyper::{
-    server::{
-        accept,
-        conn::{AddrIncoming, AddrStream},
-    },
+    server::{accept, conn::AddrIncoming},
     upgrade::Upgraded,
-    Body, Method, Request, Response, StatusCode,
+    Body, Request, Response, StatusCode,
 };
 use hyper_tungstenite::{tungstenite::Message, HyperWebsocket, WebSocketStream};
 use pin_project_lite::pin_project;
-use serde_json::{json, Value};
-
 use std::{
     convert::Infallible,
     future::ready,
@@ -26,7 +21,6 @@ use tls_listener::TlsListener;
 use tokio::{
     io::{self, AsyncBufRead, AsyncRead, AsyncWrite, ReadBuf},
     net::TcpListener,
-    select,
 };
 use tokio_util::sync::CancellationToken;
 use tracing::{error, info, info_span, warn, Instrument};
@@ -36,8 +30,6 @@ use utils::http::{error::ApiError, json::json_response};
 // Tracking issue: https://github.com/rust-lang/rust/issues/98407.
 use sync_wrapper::SyncWrapper;
 
-use super::sql_over_http;
-
 pin_project! {
     /// This is a wrapper around a [`WebSocketStream`] that
     /// implements [`AsyncRead`] and [`AsyncWrite`].
@@ -167,7 +159,6 @@ async fn ws_handler(
     config: &'static ProxyConfig,
     cancel_map: Arc<CancelMap>,
     session_id: uuid::Uuid,
-    sni_hostname: Option<String>,
 ) -> Result<Response<Body>, ApiError> {
     let host = request
         .headers()
@@ -190,44 +181,8 @@ async fn ws_handler(
 
         // Return the response so the spawned future can continue.
         Ok(response)
-    // TODO: that deserves a refactor as now this function also handles http json client besides websockets.
-    // Right now I don't want to blow up sql-over-http patch with file renames and do that as a follow up instead.
-    } else if request.uri().path() == "/sql" && request.method() == Method::POST {
-        let result = select! {
-            _ = tokio::time::sleep(std::time::Duration::from_secs(10)) => {
-                Err(anyhow::anyhow!("Query timed out"))
-            }
-            response = sql_over_http::handle(config, request, sni_hostname) => {
-                response
-            }
-        };
-        let status_code = match result {
-            Ok(_) => StatusCode::OK,
-            Err(_) => StatusCode::BAD_REQUEST,
-        };
-        let json = match result {
-            Ok(r) => r,
-            Err(e) => {
-                let message = format!("{:?}", e);
-                let code = match e.downcast_ref::<tokio_postgres::Error>() {
-                    Some(e) => match e.code() {
-                        Some(e) => serde_json::to_value(e.code()).unwrap(),
-                        None => Value::Null,
-                    },
-                    None => Value::Null,
-                };
-                json!({ "message": message, "code": code })
-            }
-        };
-        json_response(status_code, json).map(|mut r| {
-            r.headers_mut().insert(
-                "Access-Control-Allow-Origin",
-                hyper::http::HeaderValue::from_static("*"),
-            );
-            r
-        })
     } else {
-        json_response(StatusCode::BAD_REQUEST, "query is not supported")
+        json_response(StatusCode::OK, "Connect with a websocket client")
     }
 }
 
@@ -261,27 +216,20 @@ pub async fn task_main(
         }
     });
 
-    let make_svc =
-        hyper::service::make_service_fn(|stream: &tokio_rustls::server::TlsStream<AddrStream>| {
-            let sni_name = stream.get_ref().1.sni_hostname().map(|s| s.to_string());
-
-            async move {
-                Ok::<_, Infallible>(hyper::service::service_fn(move |req: Request<Body>| {
-                    let sni_name = sni_name.clone();
-                    async move {
-                        let cancel_map = Arc::new(CancelMap::default());
-                        let session_id = uuid::Uuid::new_v4();
-
-                        ws_handler(req, config, cancel_map, session_id, sni_name)
-                            .instrument(info_span!(
-                                "ws-client",
-                                session = format_args!("{session_id}")
-                            ))
-                            .await
-                    }
-                }))
-            }
-        });
+    let make_svc = hyper::service::make_service_fn(|_stream| async move {
+        Ok::<_, Infallible>(hyper::service::service_fn(
+            move |req: Request<Body>| async move {
+                let cancel_map = Arc::new(CancelMap::default());
+                let session_id = uuid::Uuid::new_v4();
+                ws_handler(req, config, cancel_map, session_id)
+                    .instrument(info_span!(
+                        "ws-client",
+                        session = format_args!("{session_id}")
+                    ))
+                    .await
+            },
+        ))
+    });
 
     hyper::Server::builder(accept::from_stream(tls_listener))
         .serve(make_svc)

proxy/src/proxy.rs

@@ -455,9 +455,6 @@ impl<'a, S> Client<'a, S> {
 
 impl<S: AsyncRead + AsyncWrite + Unpin> Client<'_, S> {
     /// Let the client authenticate and connect to the designated compute node.
-    // Instrumentation logs endpoint name everywhere. Doesn't work for link
-    // auth; strictly speaking we don't know endpoint name in its case.
-    #[tracing::instrument(name = "", fields(ep = self.creds.get_endpoint().unwrap_or("".to_owned())), skip_all)]
     async fn connect_to_db(
         self,
         session: cancellation::Session<'_>,

pyproject.toml

@@ -6,41 +6,40 @@ authors = []
 
 [tool.poetry.dependencies]
 python = "^3.9"
-pytest = "^7.3.1"
+pytest = "^6.2.5"
 psycopg2-binary = "^2.9.1"
-typing-extensions = "^4.6.1"
+typing-extensions = "^4.1.0"
 PyJWT = {version = "^2.1.0", extras = ["crypto"]}
-requests = "^2.31.0"
-pytest-xdist = "^3.3.1"
+requests = "^2.26.0"
+pytest-xdist = "^3.0.2"
 asyncpg = "^0.27.0"
 aiopg = "^1.3.1"
 Jinja2 = "^3.0.2"
-types-requests = "^2.31.0.0"
-types-psycopg2 = "^2.9.21.10"
+types-requests = "^2.28.5"
+types-psycopg2 = "^2.9.18"
 boto3 = "^1.26.16"
 boto3-stubs = {extras = ["s3"], version = "^1.26.16"}
 moto = {extras = ["server"], version = "^4.1.2"}
-backoff = "^2.2.1"
+backoff = "^1.11.1"
 pytest-lazy-fixture = "^0.6.3"
 prometheus-client = "^0.14.1"
 pytest-timeout = "^2.1.0"
 Werkzeug = "^2.2.3"
-pytest-order = "^1.1.0"
-allure-pytest = "^2.13.2"
-pytest-asyncio = "^0.21.0"
+pytest-order = "^1.0.1"
+allure-pytest = "^2.13.1"
+pytest-asyncio = "^0.19.0"
 toml = "^0.10.2"
 psutil = "^5.9.4"
-types-psutil = "^5.9.5.12"
-types-toml = "^0.10.8.6"
-pytest-httpserver = "^1.0.8"
+types-psutil = "^5.9.5.4"
+types-toml = "^0.10.8"
+pytest-httpserver = "^1.0.6"
 aiohttp = "3.7.4"
 pytest-rerunfailures = "^11.1.2"
-types-pytest-lazy-fixture = "^0.6.3.3"
 
 [tool.poetry.group.dev.dependencies]
-black = "^23.3.0"
-mypy = "==1.3.0"
-ruff = "^0.0.269"
+black = "^23.1.0"
+mypy = "==1.1.1"
+ruff = "^0.0.255"
 
 [build-system]
 requires = ["poetry-core>=1.0.0"]

scripts/export_import_between_pageservers.py

@@ -162,7 +162,7 @@ class PgProtocol:
         Returns psycopg2's connection object.
         This method passes all extra params to connstr.
         """
-        conn: PgConnection = psycopg2.connect(**self.conn_options(**kwargs))
+        conn = psycopg2.connect(**self.conn_options(**kwargs))
 
         # WARNING: this setting affects *all* tests!
         conn.autocommit = autocommit

scripts/ingest_perf_test_result.py

@@ -35,7 +35,7 @@ def get_connection_cursor():
     connstr = os.getenv("DATABASE_URL")
     if not connstr:
         err("DATABASE_URL environment variable is not set")
-    with psycopg2.connect(connstr, connect_timeout=30) as conn:
+    with psycopg2.connect(connstr) as conn:
         with conn.cursor() as cur:
             yield cur
 

test_runner/fixtures/compare_fixtures.py

@@ -312,6 +312,6 @@ def neon_with_baseline(request: FixtureRequest) -> PgCompare:
     implementation-specific logic is widely useful across multiple tests, it might
     make sense to add methods to the PgCompare class.
     """
-    fixture = request.getfixturevalue(request.param)
+    fixture = request.getfixturevalue(request.param)  # type: ignore
     assert isinstance(fixture, PgCompare), f"test error: fixture {fixture} is not PgCompare"
     return fixture

test_runner/fixtures/neon_fixtures.py

@@ -26,7 +26,7 @@ from typing import Any, Dict, Iterator, List, Optional, Tuple, Type, Union, cast
 from urllib.parse import urlparse
 
 import asyncpg
-import backoff
+import backoff  # type: ignore
 import boto3
 import jwt
 import psycopg2
@@ -149,7 +149,7 @@ 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]:
-    versioned_dir = pg_distrib_dir / pg_version.v_prefixed
+    versioned_dir = pg_distrib_dir / f"v{pg_version}"
 
     psql_bin_path = versioned_dir / "bin/psql"
     postgres_bin_path = versioned_dir / "bin/postgres"
@@ -354,7 +354,7 @@ class PgProtocol:
         Returns psycopg2's connection object.
         This method passes all extra params to connstr.
         """
-        conn: PgConnection = psycopg2.connect(**self.conn_options(**kwargs))
+        conn = psycopg2.connect(**self.conn_options(**kwargs))
 
         # WARNING: this setting affects *all* tests!
         conn.autocommit = autocommit
@@ -1745,8 +1745,8 @@ class PgBin:
     def __init__(self, log_dir: Path, pg_distrib_dir: Path, pg_version: PgVersion):
         self.log_dir = log_dir
         self.pg_version = pg_version
-        self.pg_bin_path = pg_distrib_dir / pg_version.v_prefixed / "bin"
-        self.pg_lib_dir = pg_distrib_dir / pg_version.v_prefixed / "lib"
+        self.pg_bin_path = pg_distrib_dir / f"v{pg_version}" / "bin"
+        self.pg_lib_dir = pg_distrib_dir / f"v{pg_version}" / "lib"
         self.env = os.environ.copy()
         self.env["LD_LIBRARY_PATH"] = str(self.pg_lib_dir)
 
@@ -2042,19 +2042,15 @@ class NeonProxy(PgProtocol):
         proxy_port: int,
         http_port: int,
         mgmt_port: int,
-        external_http_port: int,
         auth_backend: NeonProxy.AuthBackend,
         metric_collection_endpoint: Optional[str] = None,
         metric_collection_interval: Optional[str] = None,
     ):
         host = "127.0.0.1"
-        domain = "proxy.localtest.me"  # resolves to 127.0.0.1
-        super().__init__(dsn=auth_backend.default_conn_url, host=domain, port=proxy_port)
+        super().__init__(dsn=auth_backend.default_conn_url, host=host, port=proxy_port)
 
-        self.domain = domain
         self.host = host
         self.http_port = http_port
-        self.external_http_port = external_http_port
         self.neon_binpath = neon_binpath
         self.test_output_dir = test_output_dir
         self.proxy_port = proxy_port
@@ -2066,42 +2062,11 @@ class NeonProxy(PgProtocol):
 
     def start(self) -> NeonProxy:
         assert self._popen is None
-
-        # generate key of it doesn't exist
-        crt_path = self.test_output_dir / "proxy.crt"
-        key_path = self.test_output_dir / "proxy.key"
-
-        if not key_path.exists():
-            r = subprocess.run(
-                [
-                    "openssl",
-                    "req",
-                    "-new",
-                    "-x509",
-                    "-days",
-                    "365",
-                    "-nodes",
-                    "-text",
-                    "-out",
-                    str(crt_path),
-                    "-keyout",
-                    str(key_path),
-                    "-subj",
-                    "/CN=*.localtest.me",
-                    "-addext",
-                    "subjectAltName = DNS:*.localtest.me",
-                ]
-            )
-            assert r.returncode == 0
-
         args = [
             str(self.neon_binpath / "proxy"),
             *["--http", f"{self.host}:{self.http_port}"],
             *["--proxy", f"{self.host}:{self.proxy_port}"],
             *["--mgmt", f"{self.host}:{self.mgmt_port}"],
-            *["--wss", f"{self.host}:{self.external_http_port}"],
-            *["-c", str(crt_path)],
-            *["-k", str(key_path)],
             *self.auth_backend.extra_args(),
         ]
 
@@ -2225,7 +2190,6 @@ def link_proxy(
     http_port = port_distributor.get_port()
     proxy_port = port_distributor.get_port()
     mgmt_port = port_distributor.get_port()
-    external_http_port = port_distributor.get_port()
 
     with NeonProxy(
         neon_binpath=neon_binpath,
@@ -2233,7 +2197,6 @@ def link_proxy(
         proxy_port=proxy_port,
         http_port=http_port,
         mgmt_port=mgmt_port,
-        external_http_port=external_http_port,
         auth_backend=NeonProxy.Link(),
     ) as proxy:
         proxy.start()
@@ -2261,7 +2224,6 @@ def static_proxy(
     proxy_port = port_distributor.get_port()
     mgmt_port = port_distributor.get_port()
     http_port = port_distributor.get_port()
-    external_http_port = port_distributor.get_port()
 
     with NeonProxy(
         neon_binpath=neon_binpath,
@@ -2269,7 +2231,6 @@ def static_proxy(
         proxy_port=proxy_port,
         http_port=http_port,
         mgmt_port=mgmt_port,
-        external_http_port=external_http_port,
         auth_backend=NeonProxy.Postgres(auth_endpoint),
     ) as proxy:
         proxy.start()

test_runner/fixtures/pageserver/http.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import json
 import time
 from collections import defaultdict
 from dataclasses import dataclass
@@ -110,10 +109,6 @@ class PageserverHttpClient(requests.Session):
         if auth_token is not None:
             self.headers["Authorization"] = f"Bearer {auth_token}"
 
-    @property
-    def base_url(self) -> str:
-        return f"http://localhost:{self.port}"
-
     def verbose_error(self, res: requests.Response):
         try:
             res.raise_for_status()
@@ -154,16 +149,11 @@ class PageserverHttpClient(requests.Session):
         assert isinstance(res_json, list)
         return res_json
 
-    def tenant_create(
-        self, new_tenant_id: Optional[TenantId] = None, conf: Optional[Dict[str, Any]] = None
-    ) -> TenantId:
-        if conf is not None:
-            assert "new_tenant_id" not in conf.keys()
+    def tenant_create(self, new_tenant_id: Optional[TenantId] = None) -> TenantId:
         res = self.post(
             f"http://localhost:{self.port}/v1/tenant",
             json={
                 "new_tenant_id": str(new_tenant_id) if new_tenant_id else None,
-                **(conf or {}),
             },
         )
         self.verbose_error(res)
@@ -173,22 +163,8 @@ class PageserverHttpClient(requests.Session):
         assert isinstance(new_tenant_id, str)
         return TenantId(new_tenant_id)
 
-    def tenant_attach(
-        self, tenant_id: TenantId, config: None | Dict[str, Any] = None, config_null: bool = False
-    ):
-        if config_null:
-            assert config is None
-            body = "null"
-        else:
-            # null-config is prohibited by the API
-            if config is None:
-                config = {}
-            body = json.dumps({"config": config})
-        res = self.post(
-            f"http://localhost:{self.port}/v1/tenant/{tenant_id}/attach",
-            data=body,
-            headers={"Content-Type": "application/json"},
-        )
+    def tenant_attach(self, tenant_id: TenantId):
+        res = self.post(f"http://localhost:{self.port}/v1/tenant/{tenant_id}/attach")
         self.verbose_error(res)
 
     def tenant_detach(self, tenant_id: TenantId, detach_ignored=False):

test_runner/fixtures/pg_version.py

@@ -27,12 +27,6 @@ class PgVersion(str, enum.Enum):
     def __repr__(self) -> str:
         return f"'{self.value}'"
 
-    # In GitHub workflows we use Postgres version with v-prefix (e.g. v14 instead of just 14),
-    # sometime we need to do so in tests.
-    @property
-    def v_prefixed(self) -> str:
-        return f"v{self.value}"
-
     @classmethod
     def _missing_(cls, value) -> Optional["PgVersion"]:
         known_values = {v.value for _, v in cls.__members__.items()}

test_runner/performance/test_dup_key.py

@@ -2,7 +2,7 @@ from contextlib import closing
 
 import pytest
 from fixtures.compare_fixtures import PgCompare
-from pytest_lazyfixture import lazy_fixture
+from pytest_lazyfixture import lazy_fixture  # type: ignore
 
 
 @pytest.mark.parametrize(

test_runner/performance/test_gc_feedback.py

@@ -1,76 +0,0 @@
-import pytest
-from fixtures.benchmark_fixture import MetricReport, NeonBenchmarker
-from fixtures.log_helper import log
-from fixtures.neon_fixtures import NeonEnvBuilder
-
-
-@pytest.mark.timeout(10000)
-def test_gc_feedback(neon_env_builder: NeonEnvBuilder, zenbenchmark: NeonBenchmarker):
-    """
-    Test that GC is able to collect all old layers even if them are forming
-    "stairs" and there are not three delta layers since last image layer.
-
-    Information about image layers needed to collect old layers should
-    be propagated by GC to compaction task which should take in in account
-    when make a decision which new image layers needs to be created.
-    """
-    env = neon_env_builder.init_start()
-    client = env.pageserver.http_client()
-
-    tenant_id, _ = env.neon_cli.create_tenant(
-        conf={
-            # disable default GC and compaction
-            "gc_period": "1000 m",
-            "compaction_period": "0 s",
-            "gc_horizon": f"{1024 ** 2}",
-            "checkpoint_distance": f"{1024 ** 2}",
-            "compaction_target_size": f"{1024 ** 2}",
-            # set PITR interval to be small, so we can do GC
-            "pitr_interval": "10 s",
-            # "compaction_threshold": "3",
-            # "image_creation_threshold": "2",
-        }
-    )
-    endpoint = env.endpoints.create_start("main", tenant_id=tenant_id)
-    timeline_id = endpoint.safe_psql("show neon.timeline_id")[0][0]
-    n_steps = 10
-    n_update_iters = 100
-    step_size = 10000
-    with endpoint.cursor() as cur:
-        cur.execute("SET statement_timeout='1000s'")
-        cur.execute(
-            "CREATE TABLE t(step bigint, count bigint default 0, payload text default repeat(' ', 100))  with (fillfactor=50)"
-        )
-        cur.execute("CREATE INDEX ON t(step)")
-        # In each step, we insert 'step_size' new rows, and update the newly inserted rows
-        # 'n_update_iters' times. This creates a lot of churn and generates lots of WAL at the end of the table,
-        # without modifying the earlier parts of the table.
-        for step in range(n_steps):
-            cur.execute(f"INSERT INTO t (step) SELECT {step} FROM generate_series(1, {step_size})")
-            for i in range(n_update_iters):
-                cur.execute(f"UPDATE t set count=count+1 where step = {step}")
-                cur.execute("vacuum t")
-
-            # cur.execute("select pg_table_size('t')")
-            # logical_size = cur.fetchone()[0]
-            logical_size = client.timeline_detail(tenant_id, timeline_id)["current_logical_size"]
-            log.info(f"Logical storage size  {logical_size}")
-
-            client.timeline_checkpoint(tenant_id, timeline_id)
-
-            # Do compaction and GC
-            client.timeline_gc(tenant_id, timeline_id, 0)
-            client.timeline_compact(tenant_id, timeline_id)
-            # One more iteration to check that no excessive image layers are generated
-            client.timeline_gc(tenant_id, timeline_id, 0)
-            client.timeline_compact(tenant_id, timeline_id)
-
-            physical_size = client.timeline_detail(tenant_id, timeline_id)["current_physical_size"]
-            log.info(f"Physical storage size {physical_size}")
-
-    MB = 1024 * 1024
-    zenbenchmark.record("logical_size", logical_size // MB, "Mb", MetricReport.LOWER_IS_BETTER)
-    zenbenchmark.record("physical_size", physical_size // MB, "Mb", MetricReport.LOWER_IS_BETTER)
-    zenbenchmark.record(
-        "physical/logical ratio", physical_size / logical_size, "", MetricReport.LOWER_IS_BETTER
-    )

test_runner/performance/test_hot_page.py

@@ -2,7 +2,7 @@ from contextlib import closing
 
 import pytest
 from fixtures.compare_fixtures import PgCompare
-from pytest_lazyfixture import lazy_fixture
+from pytest_lazyfixture import lazy_fixture  # type: ignore
 
 
 @pytest.mark.parametrize(

test_runner/performance/test_hot_table.py

@@ -2,7 +2,7 @@ from contextlib import closing
 
 import pytest
 from fixtures.compare_fixtures import PgCompare
-from pytest_lazyfixture import lazy_fixture
+from pytest_lazyfixture import lazy_fixture  # type: ignore
 
 
 @pytest.mark.parametrize(

test_runner/performance/test_seqscans.py

@@ -6,7 +6,7 @@ import pytest
 from fixtures.benchmark_fixture import MetricReport
 from fixtures.compare_fixtures import PgCompare
 from fixtures.log_helper import log
-from pytest_lazyfixture import lazy_fixture
+from pytest_lazyfixture import lazy_fixture  # type: ignore
 
 
 @pytest.mark.parametrize(

test_runner/regress/test_attach_tenant_config.py

@@ -1,200 +0,0 @@
-from dataclasses import dataclass
-from typing import Generator, Optional
-
-import pytest
-from fixtures.neon_fixtures import (
-    LocalFsStorage,
-    NeonEnv,
-    NeonEnvBuilder,
-    RemoteStorageKind,
-)
-from fixtures.pageserver.http import PageserverApiException, TenantConfig
-from fixtures.types import TenantId
-from fixtures.utils import wait_until
-
-
-@pytest.fixture
-def positive_env(neon_env_builder: NeonEnvBuilder) -> NeonEnv:
-    neon_env_builder.enable_remote_storage(
-        remote_storage_kind=RemoteStorageKind.LOCAL_FS,
-        test_name="test_attach_tenant_config",
-    )
-    env = neon_env_builder.init_start()
-    assert isinstance(env.remote_storage, LocalFsStorage)
-    return env
-
-
-@dataclass
-class NegativeTests:
-    neon_env: NeonEnv
-    tenant_id: TenantId
-    config_pre_detach: TenantConfig
-
-
-@pytest.fixture
-def negative_env(neon_env_builder: NeonEnvBuilder) -> Generator[NegativeTests, None, None]:
-    neon_env_builder.enable_remote_storage(
-        remote_storage_kind=RemoteStorageKind.LOCAL_FS,
-        test_name="test_attach_tenant_config",
-    )
-    env = neon_env_builder.init_start()
-    assert isinstance(env.remote_storage, LocalFsStorage)
-
-    ps_http = env.pageserver.http_client()
-    (tenant_id, _) = env.neon_cli.create_tenant()
-    assert ps_http.tenant_config(tenant_id).tenant_specific_overrides == {}
-    config_pre_detach = ps_http.tenant_config(tenant_id)
-    assert tenant_id in [TenantId(t["id"]) for t in ps_http.tenant_list()]
-    ps_http.tenant_detach(tenant_id)
-    assert tenant_id not in [TenantId(t["id"]) for t in ps_http.tenant_list()]
-
-    yield NegativeTests(env, tenant_id, config_pre_detach)
-
-    assert tenant_id not in [
-        TenantId(t["id"]) for t in ps_http.tenant_list()
-    ], "tenant should not be attached after negative test"
-
-    env.pageserver.allowed_errors.append(".*Error processing HTTP request: Bad request")
-
-    def log_contains_bad_request():
-        env.pageserver.log_contains(".*Error processing HTTP request: Bad request")
-
-    wait_until(50, 0.1, log_contains_bad_request)
-
-
-def test_null_body(negative_env: NegativeTests):
-    """
-    If we send `null` in the body, the request should be rejected with status 400.
-    """
-    env = negative_env.neon_env
-    tenant_id = negative_env.tenant_id
-    ps_http = env.pageserver.http_client()
-
-    res = ps_http.post(
-        f"{ps_http.base_url}/v1/tenant/{tenant_id}/attach",
-        data=b"null",
-        headers={"Content-Type": "application/json"},
-    )
-    assert res.status_code == 400
-
-
-def test_null_config(negative_env: NegativeTests):
-    """
-    If the `config` field is `null`, the request should be rejected with status 400.
-    """
-
-    env = negative_env.neon_env
-    tenant_id = negative_env.tenant_id
-    ps_http = env.pageserver.http_client()
-
-    res = ps_http.post(
-        f"{ps_http.base_url}/v1/tenant/{tenant_id}/attach",
-        data=b'{"config": null}',
-        headers={"Content-Type": "application/json"},
-    )
-    assert res.status_code == 400
-
-
-def test_config_with_unknown_keys_is_bad_request(negative_env: NegativeTests):
-    """
-    If we send a config with unknown keys, the request should be rejected with status 400.
-    """
-
-    env = negative_env.neon_env
-    tenant_id = negative_env.tenant_id
-    ps_http = env.pageserver.http_client()
-
-    config_with_unknown_keys = {
-        "compaction_period": "1h",
-        "this_key_does_not_exist": "some value",
-    }
-
-    with pytest.raises(PageserverApiException) as e:
-        ps_http.tenant_attach(tenant_id, config=config_with_unknown_keys)
-    assert e.type == PageserverApiException
-    assert e.value.status_code == 400
-
-
-@pytest.mark.parametrize("content_type", [None, "application/json"])
-def test_empty_body(positive_env: NeonEnv, content_type: Optional[str]):
-    """
-    For backwards-compatiblity: if we send an empty body,
-    the request should be accepted and the config should be the default config.
-    """
-    env = positive_env
-    ps_http = env.pageserver.http_client()
-    (tenant_id, _) = env.neon_cli.create_tenant()
-    assert ps_http.tenant_config(tenant_id).tenant_specific_overrides == {}
-    config_pre_detach = ps_http.tenant_config(tenant_id)
-    assert tenant_id in [TenantId(t["id"]) for t in ps_http.tenant_list()]
-    ps_http.tenant_detach(tenant_id)
-    assert tenant_id not in [TenantId(t["id"]) for t in ps_http.tenant_list()]
-
-    ps_http.post(
-        f"{ps_http.base_url}/v1/tenant/{tenant_id}/attach",
-        data=b"",
-        headers=None if content_type else {"Content-Type": "application/json"},
-    ).raise_for_status()
-
-    assert ps_http.tenant_config(tenant_id).tenant_specific_overrides == {}
-    assert ps_http.tenant_config(tenant_id).effective_config == config_pre_detach.effective_config
-
-
-def test_fully_custom_config(positive_env: NeonEnv):
-    """
-    If we send a valid config in the body, the request should be accepted and the config should be applied.
-    """
-    env = positive_env
-
-    fully_custom_config = {
-        "compaction_period": "1h",
-        "compaction_threshold": 13,
-        "compaction_target_size": 1048576,
-        "checkpoint_distance": 10000,
-        "checkpoint_timeout": "13m",
-        "eviction_policy": {
-            "kind": "LayerAccessThreshold",
-            "period": "20s",
-            "threshold": "23h",
-        },
-        "evictions_low_residence_duration_metric_threshold": "2days",
-        "gc_horizon": 23 * (1024 * 1024),
-        "gc_period": "2h 13m",
-        "image_creation_threshold": 7,
-        "pitr_interval": "1m",
-        "lagging_wal_timeout": "23m",
-        "max_lsn_wal_lag": 230000,
-        "min_resident_size_override": 23,
-        "trace_read_requests": True,
-        "walreceiver_connect_timeout": "13m",
-    }
-
-    ps_http = env.pageserver.http_client()
-
-    initial_tenant_config = ps_http.tenant_config(env.initial_tenant)
-    assert initial_tenant_config.tenant_specific_overrides == {}
-    assert set(initial_tenant_config.effective_config.keys()) == set(
-        fully_custom_config.keys()
-    ), "ensure we cover all config options"
-
-    (tenant_id, _) = env.neon_cli.create_tenant()
-    ps_http.set_tenant_config(tenant_id, fully_custom_config)
-    our_tenant_config = ps_http.tenant_config(tenant_id)
-    assert our_tenant_config.tenant_specific_overrides == fully_custom_config
-    assert set(our_tenant_config.effective_config.keys()) == set(
-        fully_custom_config.keys()
-    ), "ensure we cover all config options"
-    assert {
-        k: initial_tenant_config.effective_config[k] != our_tenant_config.effective_config[k]
-        for k in fully_custom_config.keys()
-    } == {
-        k: True for k in fully_custom_config.keys()
-    }, "ensure our custom config has different values than the default config for all config options, so we know we overrode everything"
-
-    ps_http.tenant_detach(tenant_id)
-    ps_http.tenant_attach(tenant_id, config=fully_custom_config)
-
-    assert ps_http.tenant_config(tenant_id).tenant_specific_overrides == fully_custom_config
-    assert set(ps_http.tenant_config(tenant_id).effective_config.keys()) == set(
-        fully_custom_config.keys()
-    ), "ensure we cover all config options"

test_runner/regress/test_compatibility.py

@@ -16,7 +16,7 @@ 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.pg_version import PgVersion, skip_on_postgres
 from fixtures.types import Lsn
 from pytest import FixtureRequest
 
@@ -41,6 +41,7 @@ check_ondisk_data_compatibility_if_enabled = pytest.mark.skipif(
 )
 
 
+@skip_on_postgres(PgVersion.V15, "Compatibility tests doesn't support Postgres 15 yet")
 @pytest.mark.xdist_group("compatibility")
 @pytest.mark.order(before="test_forward_compatibility")
 def test_create_snapshot(
@@ -48,13 +49,12 @@ def test_create_snapshot(
     pg_bin: PgBin,
     top_output_dir: Path,
     test_output_dir: Path,
-    pg_version: PgVersion,
 ):
     # The test doesn't really test anything
     # it creates a new snapshot for releases after we tested the current version against the previous snapshot in `test_backward_compatibility`.
     #
     # There's no cleanup here, it allows to adjust the data in `test_backward_compatibility` itself without re-collecting it.
-    neon_env_builder.pg_version = pg_version
+    neon_env_builder.pg_version = PgVersion.V14
     neon_env_builder.num_safekeepers = 3
     neon_env_builder.enable_local_fs_remote_storage()
     neon_env_builder.preserve_database_files = True
@@ -90,14 +90,13 @@ def test_create_snapshot(
     env.pageserver.stop()
 
     # Directory `compatibility_snapshot_dir` is uploaded to S3 in a workflow, keep the name in sync with it
-    compatibility_snapshot_dir = (
-        top_output_dir / f"compatibility_snapshot_pg{pg_version.v_prefixed}"
-    )
+    compatibility_snapshot_dir = top_output_dir / "compatibility_snapshot_pg14"
     if compatibility_snapshot_dir.exists():
         shutil.rmtree(compatibility_snapshot_dir)
     shutil.copytree(test_output_dir, compatibility_snapshot_dir)
 
 
+@skip_on_postgres(PgVersion.V15, "Compatibility tests doesn't support Postgres 15 yet")
 @check_ondisk_data_compatibility_if_enabled
 @pytest.mark.xdist_group("compatibility")
 @pytest.mark.order(after="test_create_snapshot")
@@ -116,7 +115,7 @@ def test_backward_compatibility(
     compatibility_snapshot_dir_env = os.environ.get("COMPATIBILITY_SNAPSHOT_DIR")
     assert (
         compatibility_snapshot_dir_env is not None
-    ), f"COMPATIBILITY_SNAPSHOT_DIR is not set. It should be set to `compatibility_snapshot_pg{pg_version.v_prefixed}` path generateted by test_create_snapshot (ideally generated by the previous version of Neon)"
+    ), "COMPATIBILITY_SNAPSHOT_DIR is not set. It should be set to `compatibility_snapshot_pg14` path generateted by test_create_snapshot (ideally generated by the previous version of Neon)"
     compatibility_snapshot_dir = Path(compatibility_snapshot_dir_env).resolve()
 
     breaking_changes_allowed = (
@@ -156,6 +155,7 @@ def test_backward_compatibility(
     ), "Breaking changes are allowed by ALLOW_BACKWARD_COMPATIBILITY_BREAKAGE, but the test has passed without any breakage"
 
 
+@skip_on_postgres(PgVersion.V15, "Compatibility tests doesn't support Postgres 15 yet")
 @check_ondisk_data_compatibility_if_enabled
 @pytest.mark.xdist_group("compatibility")
 @pytest.mark.order(after="test_create_snapshot")
@@ -183,9 +183,7 @@ def test_forward_compatibility(
     ), "COMPATIBILITY_POSTGRES_DISTRIB_DIR is not set. It should be set to a pg_install directrory (ideally generated by the previous version of Neon)"
     compatibility_postgres_distrib_dir = Path(compatibility_postgres_distrib_dir_env).resolve()
 
-    compatibility_snapshot_dir = (
-        top_output_dir / f"compatibility_snapshot_pg{pg_version.v_prefixed}"
-    )
+    compatibility_snapshot_dir = top_output_dir / "compatibility_snapshot_pg14"
 
     breaking_changes_allowed = (
         os.environ.get("ALLOW_FORWARD_COMPATIBILITY_BREAKAGE", "false").lower() == "true"

test_runner/regress/test_disk_usage_eviction.py

@@ -118,11 +118,6 @@ class EvictionEnv:
 
         wait_until(10, 1, statvfs_called)
 
-        # these can sometimes happen during startup before any tenants have been
-        # loaded, so nothing can be evicted, we just wait for next iteration which
-        # is able to evict.
-        self.neon_env.pageserver.allowed_errors.append(".*WARN.* disk usage still high.*")
-
 
 @pytest.fixture
 def eviction_env(request, neon_env_builder: NeonEnvBuilder, pg_bin: PgBin) -> EvictionEnv:

test_runner/regress/test_hot_standby.py

@@ -1,7 +1,9 @@
 import pytest
 from fixtures.neon_fixtures import NeonEnv
+from fixtures.pg_version import PgVersion, xfail_on_postgres
 
 
+@xfail_on_postgres(PgVersion.V15, reason="https://github.com/neondatabase/neon/pull/4182")
 @pytest.mark.timeout(1800)
 def test_hot_standby(neon_simple_env: NeonEnv):
     env = neon_simple_env

test_runner/regress/test_metric_collection.py

@@ -204,7 +204,6 @@ def proxy_with_metric_collector(
     http_port = port_distributor.get_port()
     proxy_port = port_distributor.get_port()
     mgmt_port = port_distributor.get_port()
-    external_http_port = port_distributor.get_port()
 
     (host, port) = httpserver_listen_address
     metric_collection_endpoint = f"http://{host}:{port}/billing/api/v1/usage_events"
@@ -216,7 +215,6 @@ def proxy_with_metric_collector(
         proxy_port=proxy_port,
         http_port=http_port,
         mgmt_port=mgmt_port,
-        external_http_port=external_http_port,
         metric_collection_endpoint=metric_collection_endpoint,
         metric_collection_interval=metric_collection_interval,
         auth_backend=NeonProxy.Link(),

test_runner/regress/test_pg_regress.py

@@ -5,6 +5,7 @@ from pathlib import Path
 
 import pytest
 from fixtures.neon_fixtures import NeonEnv, check_restored_datadir_content
+from fixtures.pg_version import PgVersion, xfail_on_postgres
 
 
 # Run the main PostgreSQL regression tests, in src/test/regress.
@@ -32,8 +33,8 @@ def test_pg_regress(
     (runpath / "testtablespace").mkdir(parents=True)
 
     # Compute all the file locations that pg_regress will need.
-    build_path = pg_distrib_dir / f"build/{env.pg_version.v_prefixed}/src/test/regress"
-    src_path = base_dir / f"vendor/postgres-{env.pg_version.v_prefixed}/src/test/regress"
+    build_path = pg_distrib_dir / f"build/v{env.pg_version}/src/test/regress"
+    src_path = base_dir / f"vendor/postgres-v{env.pg_version}/src/test/regress"
     bindir = pg_distrib_dir / f"v{env.pg_version}/bin"
     schedule = src_path / "parallel_schedule"
     pg_regress = build_path / "pg_regress"
@@ -71,6 +72,7 @@ def test_pg_regress(
 #
 # This runs for a long time, especially in debug mode, so use a larger-than-default
 # timeout.
+@xfail_on_postgres(PgVersion.V15, reason="https://github.com/neondatabase/neon/pull/4213")
 @pytest.mark.timeout(1800)
 def test_isolation(
     neon_simple_env: NeonEnv,
@@ -95,8 +97,8 @@ def test_isolation(
     (runpath / "testtablespace").mkdir(parents=True)
 
     # Compute all the file locations that pg_isolation_regress will need.
-    build_path = pg_distrib_dir / f"build/{env.pg_version.v_prefixed}/src/test/isolation"
-    src_path = base_dir / f"vendor/postgres-{env.pg_version.v_prefixed}/src/test/isolation"
+    build_path = pg_distrib_dir / f"build/v{env.pg_version}/src/test/isolation"
+    src_path = base_dir / f"vendor/postgres-v{env.pg_version}/src/test/isolation"
     bindir = pg_distrib_dir / f"v{env.pg_version}/bin"
     schedule = src_path / "isolation_schedule"
     pg_isolation_regress = build_path / "pg_isolation_regress"

test_runner/regress/test_proxy.py

@@ -1,32 +1,22 @@
-import json
 import subprocess
-from typing import Any, List
 
 import psycopg2
 import pytest
-import requests
 from fixtures.neon_fixtures import PSQL, NeonProxy, VanillaPostgres
 
 
-def test_proxy_select_1(static_proxy: NeonProxy):
+@pytest.mark.parametrize("option_name", ["project", "endpoint"])
+def test_proxy_select_1(static_proxy: NeonProxy, option_name: str):
     """
     A simplest smoke test: check proxy against a local postgres instance.
     """
 
-    # no SNI, deprecated `options=project` syntax (before we had several endpoint in project)
-    out = static_proxy.safe_psql("select 1", sslsni=0, options="project=generic-project-name")
+    out = static_proxy.safe_psql("select 1", options=f"{option_name}=generic-project-name")
     assert out[0][0] == 1
 
-    # no SNI, new `options=endpoint` syntax
-    out = static_proxy.safe_psql("select 1", sslsni=0, options="endpoint=generic-project-name")
-    assert out[0][0] == 1
 
-    # with SNI
-    out = static_proxy.safe_psql("select 42", host="generic-project-name.localtest.me")
-    assert out[0][0] == 42
-
-
-def test_password_hack(static_proxy: NeonProxy):
+@pytest.mark.parametrize("option_name", ["project", "endpoint"])
+def test_password_hack(static_proxy: NeonProxy, option_name: str):
     """
     Check the PasswordHack auth flow: an alternative to SCRAM auth for
     clients which can't provide the project/endpoint name via SNI or `options`.
@@ -34,16 +24,14 @@ def test_password_hack(static_proxy: NeonProxy):
 
     user = "borat"
     password = "password"
-    static_proxy.safe_psql(f"create role {user} with login password '{password}'")
+    static_proxy.safe_psql(
+        f"create role {user} with login password '{password}'",
+        options=f"{option_name}=irrelevant",
+    )
 
     # Note the format of `magic`!
-    magic = f"project=irrelevant;{password}"
-    out = static_proxy.safe_psql("select 1", sslsni=0, user=user, password=magic)
-    assert out[0][0] == 1
-
-    magic = f"endpoint=irrelevant;{password}"
-    out = static_proxy.safe_psql("select 1", sslsni=0, user=user, password=magic)
-    assert out[0][0] == 1
+    magic = f"{option_name}=irrelevant;{password}"
+    static_proxy.safe_psql("select 1", sslsni=0, user=user, password=magic)
 
     # Must also check that invalid magic won't be accepted.
     with pytest.raises(psycopg2.OperationalError):
@@ -81,55 +69,52 @@ def test_proxy_options(static_proxy: NeonProxy, option_name: str):
     """
 
     options = f"{option_name}=irrelevant -cproxytest.option=value"
-    out = static_proxy.safe_psql("show proxytest.option", options=options, sslsni=0)
-    assert out[0][0] == "value"
-
-    options = f"-c proxytest.foo=\\ str {option_name}=irrelevant"
-    out = static_proxy.safe_psql("show proxytest.foo", options=options, sslsni=0)
-    assert out[0][0] == " str"
-
-    options = "-cproxytest.option=value"
     out = static_proxy.safe_psql("show proxytest.option", options=options)
     assert out[0][0] == "value"
 
-    options = "-c proxytest.foo=\\ str"
+    options = f"-c proxytest.foo=\\ str {option_name}=irrelevant"
     out = static_proxy.safe_psql("show proxytest.foo", options=options)
     assert out[0][0] == " str"
 
 
-def test_auth_errors(static_proxy: NeonProxy):
+@pytest.mark.parametrize("option_name", ["project", "endpoint"])
+def test_auth_errors(static_proxy: NeonProxy, option_name: str):
     """
     Check that we throw very specific errors in some unsuccessful auth scenarios.
     """
 
     # User does not exist
     with pytest.raises(psycopg2.Error) as exprinfo:
-        static_proxy.connect(user="pinocchio")
+        static_proxy.connect(user="pinocchio", options=f"{option_name}=irrelevant")
     text = str(exprinfo.value).strip()
-    assert text.find("password authentication failed for user 'pinocchio'") != -1
+    assert text.endswith("password authentication failed for user 'pinocchio'")
 
     static_proxy.safe_psql(
         "create role pinocchio with login password 'magic'",
+        options=f"{option_name}=irrelevant",
     )
 
     # User exists, but password is missing
     with pytest.raises(psycopg2.Error) as exprinfo:
-        static_proxy.connect(user="pinocchio", password=None)
+        static_proxy.connect(user="pinocchio", password=None, options=f"{option_name}=irrelevant")
     text = str(exprinfo.value).strip()
-    assert text.find("password authentication failed for user 'pinocchio'") != -1
+    assert text.endswith("password authentication failed for user 'pinocchio'")
 
     # User exists, but password is wrong
     with pytest.raises(psycopg2.Error) as exprinfo:
-        static_proxy.connect(user="pinocchio", password="bad")
+        static_proxy.connect(user="pinocchio", password="bad", options=f"{option_name}=irrelevant")
     text = str(exprinfo.value).strip()
-    assert text.find("password authentication failed for user 'pinocchio'") != -1
+    assert text.endswith("password authentication failed for user 'pinocchio'")
 
     # Finally, check that the user can connect
-    with static_proxy.connect(user="pinocchio", password="magic"):
+    with static_proxy.connect(
+        user="pinocchio", password="magic", options=f"{option_name}=irrelevant"
+    ):
         pass
 
 
-def test_forward_params_to_client(static_proxy: NeonProxy):
+@pytest.mark.parametrize("option_name", ["project", "endpoint"])
+def test_forward_params_to_client(static_proxy: NeonProxy, option_name: str):
     """
     Check that we forward all necessary PostgreSQL server params to client.
     """
@@ -155,7 +140,7 @@ def test_forward_params_to_client(static_proxy: NeonProxy):
         where name = any(%s)
     """
 
-    with static_proxy.connect() as conn:
+    with static_proxy.connect(options=f"{option_name}=irrelevant") as conn:
         with conn.cursor() as cur:
             cur.execute(query, (reported_params_subset,))
             for name, value in cur.fetchall():
@@ -163,65 +148,18 @@ def test_forward_params_to_client(static_proxy: NeonProxy):
                 assert conn.get_parameter_status(name) == value
 
 
+@pytest.mark.parametrize("option_name", ["project", "endpoint"])
 @pytest.mark.timeout(5)
-def test_close_on_connections_exit(static_proxy: NeonProxy):
+def test_close_on_connections_exit(static_proxy: NeonProxy, option_name: str):
     # Open two connections, send SIGTERM, then ensure that proxy doesn't exit
     # until after connections close.
-    with static_proxy.connect(), static_proxy.connect():
+    with static_proxy.connect(options=f"{option_name}=irrelevant"), static_proxy.connect(
+        options=f"{option_name}=irrelevant"
+    ):
         static_proxy.terminate()
         with pytest.raises(subprocess.TimeoutExpired):
             static_proxy.wait_for_exit(timeout=2)
         # Ensure we don't accept any more connections
         with pytest.raises(psycopg2.OperationalError):
-            static_proxy.connect()
+            static_proxy.connect(options=f"{option_name}=irrelevant")
     static_proxy.wait_for_exit()
-
-
-def test_sql_over_http(static_proxy: NeonProxy):
-    static_proxy.safe_psql("create role http with login password 'http' superuser")
-
-    def q(sql: str, params: List[Any] = []) -> Any:
-        connstr = f"postgresql://http:http@{static_proxy.domain}:{static_proxy.proxy_port}/postgres"
-        response = requests.post(
-            f"https://{static_proxy.domain}:{static_proxy.external_http_port}/sql",
-            data=json.dumps({"query": sql, "params": params}),
-            headers={"Content-Type": "application/sql", "Neon-Connection-String": connstr},
-            verify=str(static_proxy.test_output_dir / "proxy.crt"),
-        )
-        assert response.status_code == 200
-        return response.json()
-
-    rows = q("select 42 as answer")["rows"]
-    assert rows == [{"answer": 42}]
-
-    rows = q("select $1 as answer", [42])["rows"]
-    assert rows == [{"answer": "42"}]
-
-    rows = q("select $1 * 1 as answer", [42])["rows"]
-    assert rows == [{"answer": 42}]
-
-    rows = q("select $1::int[] as answer", [[1, 2, 3]])["rows"]
-    assert rows == [{"answer": [1, 2, 3]}]
-
-    rows = q("select $1::json->'a' as answer", [{"a": {"b": 42}}])["rows"]
-    assert rows == [{"answer": {"b": 42}}]
-
-    rows = q("select * from pg_class limit 1")["rows"]
-    assert len(rows) == 1
-
-    res = q("create table t(id serial primary key, val int)")
-    assert res["command"] == "CREATE"
-    assert res["rowCount"] is None
-
-    res = q("insert into t(val) values (10), (20), (30) returning id")
-    assert res["command"] == "INSERT"
-    assert res["rowCount"] == 3
-    assert res["rows"] == [{"id": 1}, {"id": 2}, {"id": 3}]
-
-    res = q("select * from t")
-    assert res["command"] == "SELECT"
-    assert res["rowCount"] == 3
-
-    res = q("drop table t")
-    assert res["command"] == "DROP"
-    assert res["rowCount"] is None

test_runner/regress/test_sni_router.py

@@ -4,7 +4,7 @@ from pathlib import Path
 from types import TracebackType
 from typing import Optional, Type
 
-import backoff
+import backoff  # type: ignore
 from fixtures.log_helper import log
 from fixtures.neon_fixtures import PgProtocol, PortDistributor, VanillaPostgres
 

test_runner/regress/test_tenant_conf.py

@@ -62,7 +62,6 @@ eviction_policy = { "kind" = "LayerAccessThreshold", period = "20s", threshold =
             log.info(f"show {env.initial_tenant}")
             pscur.execute(f"show {env.initial_tenant}")
             res = pscur.fetchone()
-            assert res is not None
             assert all(
                 i in res.items()
                 for i in {
@@ -102,7 +101,6 @@ eviction_policy = { "kind" = "LayerAccessThreshold", period = "20s", threshold =
             pscur.execute(f"show {tenant}")
             res = pscur.fetchone()
             log.info(f"res: {res}")
-            assert res is not None
             assert all(
                 i in res.items()
                 for i in {
@@ -153,7 +151,6 @@ eviction_policy = { "kind" = "LayerAccessThreshold", period = "20s", threshold =
         "eviction_policy": json.dumps(
             {"kind": "LayerAccessThreshold", "period": "80s", "threshold": "42h"}
         ),
-        "max_lsn_wal_lag": "13000000",
     }
     env.neon_cli.config_tenant(
         tenant_id=tenant,
@@ -165,7 +162,6 @@ eviction_policy = { "kind" = "LayerAccessThreshold", period = "20s", threshold =
             pscur.execute(f"show {tenant}")
             res = pscur.fetchone()
             log.info(f"after config res: {res}")
-            assert res is not None
             assert all(
                 i in res.items()
                 for i in {
@@ -210,7 +206,6 @@ eviction_policy = { "kind" = "LayerAccessThreshold", period = "20s", threshold =
     assert updated_effective_config["gc_horizon"] == 67108864
     assert updated_effective_config["image_creation_threshold"] == 2
     assert updated_effective_config["pitr_interval"] == "7days"
-    assert updated_effective_config["max_lsn_wal_lag"] == 13000000
 
     # restart the pageserver and ensure that the config is still correct
     env.pageserver.stop()
@@ -221,7 +216,6 @@ eviction_policy = { "kind" = "LayerAccessThreshold", period = "20s", threshold =
             pscur.execute(f"show {tenant}")
             res = pscur.fetchone()
             log.info(f"after restart res: {res}")
-            assert res is not None
             assert all(
                 i in res.items()
                 for i in {
@@ -271,7 +265,6 @@ eviction_policy = { "kind" = "LayerAccessThreshold", period = "20s", threshold =
         "period": "20s",
         "threshold": "23h",
     }
-    assert final_effective_config["max_lsn_wal_lag"] == 10 * 1024 * 1024
 
     # restart the pageserver and ensure that the config is still correct
     env.pageserver.stop()
@@ -282,7 +275,6 @@ eviction_policy = { "kind" = "LayerAccessThreshold", period = "20s", threshold =
             pscur.execute(f"show {tenant}")
             res = pscur.fetchone()
             log.info(f"after restart res: {res}")
-            assert res is not None
             assert all(
                 i in res.items()
                 for i in {