Add 'failed get control bytes' to list of expected errors

Add 'failed to get checkpoint bytes' to list of expected errors
RemoveRemove unnecessary reset of redo field
2026-07-07 22:20:36 +00:00 · 2024-12-05 16:24:31 +02:00 · 2024-12-05 15:32:02 +02:00 · 2024-12-05 14:28:34 +02:00 · 2024-12-05 10:04:51 +02:00 · 2024-12-05 09:04:59 +02:00
88 changed files with 1050 additions and 2298 deletions
--- a/.github/actions/allure-report-generate/action.yml
+++ b/.github/actions/allure-report-generate/action.yml
@@ -43,8 +43,7 @@ runs:
        PR_NUMBER=$(jq --raw-output .pull_request.number "$GITHUB_EVENT_PATH" || true)
        if [ "${PR_NUMBER}" != "null" ]; then
          BRANCH_OR_PR=pr-${PR_NUMBER}
-        elif [ "${GITHUB_REF_NAME}" = "main" ] || [ "${GITHUB_REF_NAME}" = "release" ] || \
-             [ "${GITHUB_REF_NAME}" = "release-proxy" ] || [ "${GITHUB_REF_NAME}" = "release-compute" ]; then
+        elif [ "${GITHUB_REF_NAME}" = "main" ] || [ "${GITHUB_REF_NAME}" = "release" ] || [ "${GITHUB_REF_NAME}" = "release-proxy" ]; then
          # Shortcut for special branches
          BRANCH_OR_PR=${GITHUB_REF_NAME}
        else
--- a/.github/actions/allure-report-store/action.yml
+++ b/.github/actions/allure-report-store/action.yml
@@ -23,8 +23,7 @@ runs:
        PR_NUMBER=$(jq --raw-output .pull_request.number "$GITHUB_EVENT_PATH" || true)
        if [ "${PR_NUMBER}" != "null" ]; then
          BRANCH_OR_PR=pr-${PR_NUMBER}
-        elif [ "${GITHUB_REF_NAME}" = "main" ] || [ "${GITHUB_REF_NAME}" = "release" ] || \
-             [ "${GITHUB_REF_NAME}" = "release-proxy" ] || [ "${GITHUB_REF_NAME}" = "release-compute" ]; then
+        elif [ "${GITHUB_REF_NAME}" = "main" ] || [ "${GITHUB_REF_NAME}" = "release" ] || [ "${GITHUB_REF_NAME}" = "release-proxy" ]; then
          # Shortcut for special branches
          BRANCH_OR_PR=${GITHUB_REF_NAME}
        else
--- a/.github/workflows/_create-release-pr.yml
+++ b/.github/workflows/_create-release-pr.yml
@@ -21,7 +21,7 @@ defaults:
    shell: bash -euo pipefail {0}

 jobs:
-  create-release-branch:
+  create-storage-release-branch:
    runs-on: ubuntu-22.04

    permissions:
--- a/.github/workflows/build_and_test.yml
+++ b/.github/workflows/build_and_test.yml
@@ -6,7 +6,6 @@ on:
      - main
      - release
      - release-proxy
-      - release-compute
  pull_request:

 defaults:
@@ -71,10 +70,8 @@ jobs:
            echo "tag=release-$(git rev-list --count HEAD)" >> $GITHUB_OUTPUT
          elif [[ "$GITHUB_REF_NAME" == "release-proxy" ]]; then
            echo "tag=release-proxy-$(git rev-list --count HEAD)" >> $GITHUB_OUTPUT
-          elif [[ "$GITHUB_REF_NAME" == "release-compute" ]]; then
-            echo "tag=release-compute-$(git rev-list --count HEAD)" >> $GITHUB_OUTPUT
          else
-            echo "GITHUB_REF_NAME (value '$GITHUB_REF_NAME') is not set to either 'main' or 'release', 'release-proxy', 'release-compute'"
+            echo "GITHUB_REF_NAME (value '$GITHUB_REF_NAME') is not set to either 'main' or 'release'"
            echo "tag=$GITHUB_RUN_ID" >> $GITHUB_OUTPUT
          fi
        shell: bash
@@ -516,7 +513,7 @@ jobs:
            })

  trigger-e2e-tests:
-    if: ${{ !github.event.pull_request.draft || contains( github.event.pull_request.labels.*.name, 'run-e2e-tests-in-draft') || github.ref_name == 'main' || github.ref_name == 'release' || github.ref_name == 'release-proxy' || github.ref_name == 'release-compute' }}
+    if: ${{ !github.event.pull_request.draft || contains( github.event.pull_request.labels.*.name, 'run-e2e-tests-in-draft') || github.ref_name == 'main' || github.ref_name == 'release' || github.ref_name == 'release-proxy' }}
    needs: [ check-permissions, promote-images, tag ]
    uses: ./.github/workflows/trigger-e2e-tests.yml
    secrets: inherit
@@ -937,7 +934,7 @@ jobs:
                                              neondatabase/neon-test-extensions-v16:${{ needs.tag.outputs.build-tag }}

      - name: Configure AWS-prod credentials
-        if: github.ref_name == 'release'|| github.ref_name == 'release-proxy' || github.ref_name == 'release-compute'
+        if: github.ref_name == 'release'|| github.ref_name == 'release-proxy'
        uses: aws-actions/configure-aws-credentials@v4
        with:
          aws-region: eu-central-1
@@ -946,12 +943,12 @@ jobs:

      - name: Login to prod ECR
        uses: docker/login-action@v3
-        if: github.ref_name == 'release'|| github.ref_name == 'release-proxy' || github.ref_name == 'release-compute'
+        if: github.ref_name == 'release'|| github.ref_name == 'release-proxy'
        with:
          registry: 093970136003.dkr.ecr.eu-central-1.amazonaws.com

      - name: Copy all images to prod ECR
-        if: github.ref_name == 'release' || github.ref_name == 'release-proxy' || github.ref_name == 'release-compute'
+        if: github.ref_name == 'release'|| github.ref_name == 'release-proxy'
        run: |
          for image in neon compute-tools {vm-,}compute-node-{v14,v15,v16,v17}; do
            docker buildx imagetools create -t 093970136003.dkr.ecr.eu-central-1.amazonaws.com/${image}:${{ needs.tag.outputs.build-tag }} \
@@ -971,7 +968,7 @@ jobs:
      tenant_id: ${{ vars.AZURE_TENANT_ID }}

  push-to-acr-prod:
-    if: github.ref_name == 'release' || github.ref_name == 'release-proxy' || github.ref_name == 'release-compute'
+    if: github.ref_name == 'release'|| github.ref_name == 'release-proxy'
    needs: [ tag, promote-images ]
    uses: ./.github/workflows/_push-to-acr.yml
    with:
@@ -1059,7 +1056,7 @@ jobs:
  deploy:
    needs: [ check-permissions, promote-images, tag, build-and-test-locally, trigger-custom-extensions-build-and-wait, push-to-acr-dev, push-to-acr-prod ]
    # `!failure() && !cancelled()` is required because the workflow depends on the job that can be skipped: `push-to-acr-dev` and `push-to-acr-prod`
-    if: (github.ref_name == 'main' || github.ref_name == 'release' || github.ref_name == 'release-proxy' || github.ref_name == 'release-compute') && !failure() && !cancelled()
+    if: (github.ref_name == 'main' || github.ref_name == 'release' || github.ref_name == 'release-proxy') && !failure() && !cancelled()

    runs-on: [ self-hosted, small ]
    container: 369495373322.dkr.ecr.eu-central-1.amazonaws.com/ansible:latest
@@ -1108,15 +1105,13 @@ jobs:
              -f deployProxyAuthBroker=true \
              -f branch=main \
              -f dockerTag=${{needs.tag.outputs.build-tag}}
-          elif [[ "$GITHUB_REF_NAME" == "release-compute" ]]; then
-            gh workflow --repo neondatabase/infra run deploy-compute-dev.yml --ref main -f dockerTag=${{needs.tag.outputs.build-tag}}
          else
-            echo "GITHUB_REF_NAME (value '$GITHUB_REF_NAME') is not set to either 'main', 'release', 'release-proxy' or 'release-compute'"
+            echo "GITHUB_REF_NAME (value '$GITHUB_REF_NAME') is not set to either 'main' or 'release'"
            exit 1
          fi

      - name: Create git tag
-        if: github.ref_name == 'release' || github.ref_name == 'release-proxy' || github.ref_name == 'release-compute'
+        if: github.ref_name == 'release' || github.ref_name == 'release-proxy'
        uses: actions/github-script@v7
        with:
          # Retry script for 5XX server errors: https://github.com/actions/github-script#retries
--- a/.github/workflows/ingest_benchmark.yml
+++ b/.github/workflows/ingest_benchmark.yml
@@ -26,7 +26,6 @@ concurrency:
 jobs:
  ingest:
    strategy:
-      fail-fast: false # allow other variants to continue even if one fails
      matrix:
        target_project: [new_empty_project, large_existing_project]  
    permissions:
--- a/.github/workflows/release.yml
+++ b/.github/workflows/release.yml
@@ -15,10 +15,6 @@ on:
        type: boolean
        description: 'Create Proxy release PR'
        required: false
-      create-compute-release-branch:
-        type: boolean
-        description: 'Create Compute release PR'
-        required: false

 # No permission for GITHUB_TOKEN by default; the **minimal required** set of permissions should be granted in each job.
 permissions: {}
@@ -29,20 +25,20 @@ defaults:

 jobs:
  create-storage-release-branch:
-    if: ${{ github.event.schedule == '0 6 * * MON' || inputs.create-storage-release-branch }}
+    if: ${{ github.event.schedule == '0 6 * * MON' || format('{0}', inputs.create-storage-release-branch) == 'true' }}

    permissions:
      contents: write

    uses: ./.github/workflows/_create-release-pr.yml
    with:
-      component-name: 'Storage'
+      component-name: 'Storage & Compute'
      release-branch: 'release'
    secrets:
      ci-access-token: ${{ secrets.CI_ACCESS_TOKEN }}

  create-proxy-release-branch:
-    if: ${{ github.event.schedule == '0 6 * * THU' || inputs.create-proxy-release-branch }}
+    if: ${{ github.event.schedule == '0 6 * * THU' || format('{0}', inputs.create-proxy-release-branch) == 'true' }}

    permissions:
      contents: write
@@ -53,16 +49,3 @@ jobs:
      release-branch: 'release-proxy'
    secrets:
      ci-access-token: ${{ secrets.CI_ACCESS_TOKEN }}
-
-  create-compute-release-branch:
-    if: inputs.create-compute-release-branch
-
-    permissions:
-      contents: write
-
-    uses: ./.github/workflows/_create-release-pr.yml
-    with:
-      component-name: 'Compute'
-      release-branch: 'release-compute'
-    secrets:
-      ci-access-token: ${{ secrets.CI_ACCESS_TOKEN }}
--- a/.github/workflows/trigger-e2e-tests.yml
+++ b/.github/workflows/trigger-e2e-tests.yml
@@ -51,8 +51,6 @@ jobs:
            echo "tag=release-$(git rev-list --count HEAD)" | tee -a $GITHUB_OUTPUT
          elif [[ "$GITHUB_REF_NAME" == "release-proxy" ]]; then
            echo "tag=release-proxy-$(git rev-list --count HEAD)" >> $GITHUB_OUTPUT
-          elif [[ "$GITHUB_REF_NAME" == "release-compute" ]]; then
-            echo "tag=release-compute-$(git rev-list --count HEAD)" >> $GITHUB_OUTPUT
          else
            echo "GITHUB_REF_NAME (value '$GITHUB_REF_NAME') is not set to either 'main' or 'release'"
            BUILD_AND_TEST_RUN_ID=$(gh run list -b $CURRENT_BRANCH -c $CURRENT_SHA -w 'Build and Test' -L 1 --json databaseId --jq '.[].databaseId')
--- a/1
+++ b/1
@@ -2,7 +2,6 @@
 /compute_tools/ @neondatabase/control-plane @neondatabase/compute
 /libs/pageserver_api/ @neondatabase/storage
 /libs/postgres_ffi/ @neondatabase/compute @neondatabase/storage
-/libs/proxy/ @neondatabase/proxy
 /libs/remote_storage/ @neondatabase/storage
 /libs/safekeeper_api/ @neondatabase/storage
 /libs/vm_monitor/ @neondatabase/autoscaling
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -84,16 +84,16 @@ checksum = "4b46cbb362ab8752921c97e041f5e366ee6297bd428a31275b9fcf1e380f7299"

 [[package]]
 name = "anstream"
-version = "0.6.15"
+version = "0.3.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "64e15c1ab1f89faffbf04a634d5e1962e9074f2741eef6d97f3c4e322426d526"
+checksum = "0ca84f3628370c59db74ee214b3263d58f9aadd9b4fe7e711fd87dc452b7f163"
 dependencies = [
 "anstyle",
 "anstyle-parse",
 "anstyle-query",
 "anstyle-wincon",
 "colorchoice",
- "is_terminal_polyfill",
+ "is-terminal",
 "utf8parse",
 ]

@@ -123,12 +123,12 @@ dependencies = [

 [[package]]
 name = "anstyle-wincon"
-version = "3.0.4"
+version = "1.0.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5bf74e1b6e971609db8ca7a9ce79fd5768ab6ae46441c572e46cf596f59e57f8"
+checksum = "180abfa45703aebe0093f79badacc01b8fd4ea2e35118747e5811127f926e188"
 dependencies = [
 "anstyle",
- "windows-sys 0.52.0",
+ "windows-sys 0.48.0",
 ]

 [[package]]
@@ -1031,9 +1031,9 @@ checksum = "14c189c53d098945499cdfa7ecc63567cf3886b3332b312a5b4585d8d3a6a610"

 [[package]]
 name = "bytes"
-version = "1.9.0"
+version = "1.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "325918d6fe32f23b19878fe4b34794ae41fc19ddbe53b10571a4874d44ffd39b"
+checksum = "a2bd12c1caf447e69cd4528f47f94d203fd2582878ecb9e9465484c4148a8223"
 dependencies = [
 "serde",
 ]
@@ -1167,33 +1167,35 @@ dependencies = [

 [[package]]
 name = "clap"
-version = "4.5.22"
+version = "4.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "69371e34337c4c984bbe322360c2547210bf632eb2814bbe78a6e87a2935bd2b"
+checksum = "93aae7a4192245f70fe75dd9157fc7b4a5bf53e88d30bd4396f7d8f9284d5acc"
 dependencies = [
 "clap_builder",
 "clap_derive",
+ "once_cell",
 ]

 [[package]]
 name = "clap_builder"
-version = "4.5.22"
+version = "4.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6e24c1b4099818523236a8ca881d2b45db98dadfb4625cf6608c12069fcbbde1"
+checksum = "4f423e341edefb78c9caba2d9c7f7687d0e72e89df3ce3394554754393ac3990"
 dependencies = [
 "anstream",
 "anstyle",
+ "bitflags 1.3.2",
 "clap_lex",
- "strsim 0.11.1",
+ "strsim",
 ]

 [[package]]
 name = "clap_derive"
-version = "4.5.18"
+version = "4.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4ac6a0c7b1a9e9a5186361f67dfa1b88213572f427fb9ab038efb2bd8c582dab"
+checksum = "191d9573962933b4027f932c600cd252ce27a8ad5979418fe78e43c07996f27b"
 dependencies = [
- "heck",
+ "heck 0.4.1",
 "proc-macro2",
 "quote",
 "syn 2.0.90",
@@ -1201,9 +1203,9 @@ dependencies = [

 [[package]]
 name = "clap_lex"
-version = "0.7.3"
+version = "0.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "afb84c814227b90d6895e01398aee0d8033c00e7466aca416fb6a8e0eb19d8a7"
+checksum = "2da6da31387c7e4ef160ffab6d5e7f00c42626fe39aea70a7b0f1773f7dd6c1b"

 [[package]]
 name = "colorchoice"
@@ -1612,7 +1614,7 @@ dependencies = [
 "ident_case",
 "proc-macro2",
 "quote",
- "strsim 0.10.0",
+ "strsim",
 "syn 2.0.90",
 ]

@@ -1810,7 +1812,7 @@ checksum = "0892a17df262a24294c382f0d5997571006e7a4348b4327557c4ff1cd4a8bccc"
 dependencies = [
 "darling",
 "either",
- "heck",
+ "heck 0.5.0",
 "proc-macro2",
 "quote",
 "syn 2.0.90",
@@ -2463,6 +2465,12 @@ dependencies = [
 "num-traits",
 ]

+[[package]]
+name = "heck"
+version = "0.4.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "95505c38b4572b2d910cecb0281560f54b440a19336cbbcb27bf6ce6adc6f5a8"
+
 [[package]]
 name = "heck"
 version = "0.5.0"
@@ -2880,12 +2888,6 @@ dependencies = [
 "windows-sys 0.52.0",
 ]

-[[package]]
-name = "is_terminal_polyfill"
-version = "1.70.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7943c866cc5cd64cbc25b2e01621d07fa8eb2a1a23160ee81ce38704e97b8ecf"
-
 [[package]]
 name = "itertools"
 version = "0.10.5"
@@ -3167,7 +3169,7 @@ version = "0.0.22"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b9e6777fc80a575f9503d908c8b498782a6c3ee88a06cb416dc3941401e43b94"
 dependencies = [
- "heck",
+ "heck 0.5.0",
 "proc-macro2",
 "quote",
 "syn 2.0.90",
@@ -4169,7 +4171,7 @@ dependencies = [
 [[package]]
 name = "postgres"
 version = "0.19.4"
-source = "git+https://github.com/neondatabase/rust-postgres.git?branch=neon#511f998c00148ab7c847bd7e6cfd3a906d0e7473"
+source = "git+https://github.com/neondatabase/rust-postgres.git?branch=neon#00940fcdb57a8e99e805297b75839e7c4c7b1796"
 dependencies = [
 "bytes",
 "fallible-iterator",
@@ -4182,7 +4184,7 @@ dependencies = [
 [[package]]
 name = "postgres-protocol"
 version = "0.6.4"
-source = "git+https://github.com/neondatabase/rust-postgres.git?branch=neon#511f998c00148ab7c847bd7e6cfd3a906d0e7473"
+source = "git+https://github.com/neondatabase/rust-postgres.git?branch=neon#00940fcdb57a8e99e805297b75839e7c4c7b1796"
 dependencies = [
 "base64 0.20.0",
 "byteorder",
@@ -4195,6 +4197,7 @@ dependencies = [
 "rand 0.8.5",
 "sha2",
 "stringprep",
+ "tokio",
 ]

 [[package]]
@@ -4216,7 +4219,7 @@ dependencies = [
 [[package]]
 name = "postgres-types"
 version = "0.2.4"
-source = "git+https://github.com/neondatabase/rust-postgres.git?branch=neon#511f998c00148ab7c847bd7e6cfd3a906d0e7473"
+source = "git+https://github.com/neondatabase/rust-postgres.git?branch=neon#00940fcdb57a8e99e805297b75839e7c4c7b1796"
 dependencies = [
 "bytes",
 "fallible-iterator",
@@ -4455,7 +4458,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0c1318b19085f08681016926435853bbf7858f9c082d0999b80550ff5d9abe15"
 dependencies = [
 "bytes",
- "heck",
+ "heck 0.5.0",
 "itertools 0.12.1",
 "log",
 "multimap",
@@ -6163,12 +6166,6 @@ version = "0.10.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "73473c0e59e6d5812c5dfe2a064a6444949f089e20eec9a2e5506596494e4623"

-[[package]]
-name = "strsim"
-version = "0.11.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7da8b5736845d9f2fcb837ea5d9e2628564b3b043a70948a3f0b778838c5fb4f"
-
 [[package]]
 name = "strum"
 version = "0.26.3"
@@ -6181,7 +6178,7 @@ version = "0.26.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4c6bee85a5a24955dc440386795aa378cd9cf82acd5f764469152d2270e581be"
 dependencies = [
- "heck",
+ "heck 0.5.0",
 "proc-macro2",
 "quote",
 "rustversion",
@@ -6546,7 +6543,7 @@ dependencies = [
 [[package]]
 name = "tokio-postgres"
 version = "0.7.7"
-source = "git+https://github.com/neondatabase/rust-postgres.git?branch=neon#511f998c00148ab7c847bd7e6cfd3a906d0e7473"
+source = "git+https://github.com/neondatabase/rust-postgres.git?branch=neon#00940fcdb57a8e99e805297b75839e7c4c7b1796"
 dependencies = [
 "async-trait",
 "byteorder",
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -74,7 +74,7 @@ bindgen = "0.70"
 bit_field = "0.10.2"
 bstr = "1.0"
 byteorder = "1.4"
-bytes = "1.9"
+bytes = "1.0"
 camino = "1.1.6"
 cfg-if = "1.0.0"
 chrono = { version = "0.4", default-features = false, features = ["clock"] }
--- a/compute/etc/pgbouncer.ini
+++ b/compute/etc/pgbouncer.ini
@@ -1,9 +1,5 @@
 [databases]
-;; pgbouncer propagates application_name (if it's specified) to the server, but some
-;; clients don't set it. We set default application_name=pgbouncer to make it
-;; easier to identify pgbouncer connections in Postgres. If client sets
-;; application_name, it will be used instead.
-*=host=localhost port=5432 auth_user=cloud_admin application_name=pgbouncer
+*=host=localhost port=5432 auth_user=cloud_admin
 [pgbouncer]
 listen_port=6432
 listen_addr=0.0.0.0
--- a/control_plane/src/safekeeper.rs
+++ b/control_plane/src/safekeeper.rs
@@ -5,7 +5,6 @@
 //! ```text
 //!   .neon/safekeepers/<safekeeper id>
 //! ```
-use std::error::Error as _;
 use std::future::Future;
 use std::io::Write;
 use std::path::PathBuf;
@@ -27,7 +26,7 @@ use crate::{

 #[derive(Error, Debug)]
 pub enum SafekeeperHttpError {
-    #[error("request error: {0}{}", .0.source().map(|e| format!(": {e}")).unwrap_or_default())]
+    #[error("Reqwest error: {0}")]
    Transport(#[from] reqwest::Error),

    #[error("Error: {0}")]
--- a/libs/postgres_ffi/src/lib.rs
+++ b/libs/postgres_ffi/src/lib.rs
@@ -278,7 +278,7 @@ pub fn generate_pg_control(
    checkpoint_bytes: &[u8],
    lsn: Lsn,
    pg_version: u32,
-) -> anyhow::Result<(Bytes, u64)> {
+) -> anyhow::Result<(Bytes, u64, bool)> {
    dispatch_pgversion!(
        pg_version,
        pgv::xlog_utils::generate_pg_control(pg_control_bytes, checkpoint_bytes, lsn),
--- a/libs/postgres_ffi/src/xlog_utils.rs
+++ b/libs/postgres_ffi/src/xlog_utils.rs
@@ -124,23 +124,64 @@ pub fn normalize_lsn(lsn: Lsn, seg_sz: usize) -> Lsn {
    }
 }

+/// Generate a pg_control file, for a basebackup for starting up Postgres at the given LSN
+///
+/// 'pg_control_bytes' and 'checkpoint_bytes' are the contents of those keys persisted in
+/// the pageserver. They use the same format as the PostgreSQL control file and the
+/// checkpoint record, but see walingest.rs for how exactly they are kept up to date.
+/// 'lsn' is the LSN at which we're starting up.
+///
+/// Returns:
+/// - pg_control file contents
+/// - system_identifier, extracted from the persisted information
+/// - true, if we're starting up from a "clean shutdown", i.e. if there was a shutdown
+///   checkpoint at the given LSN
 pub fn generate_pg_control(
    pg_control_bytes: &[u8],
    checkpoint_bytes: &[u8],
    lsn: Lsn,
-) -> anyhow::Result<(Bytes, u64)> {
+) -> anyhow::Result<(Bytes, u64, bool)> {
    let mut pg_control = ControlFileData::decode(pg_control_bytes)?;
    let mut checkpoint = CheckPoint::decode(checkpoint_bytes)?;
+    let was_shutdown;

    // Generate new pg_control needed for bootstrap
-    checkpoint.redo = normalize_lsn(lsn, WAL_SEGMENT_SIZE).0;
+    //
+    // NB: In the checkpoint struct that we persist in the pageserver, we have a different
+    // convention for the 'redo' field than in PostgreSQL: On a shutdown checkpoint,
+    // 'redo' points the *end* of the checkpoint WAL record. On PostgreSQL, it points to
+    // the beginning. Furthermore, on an online checkpoint, 'redo' is set to 0.
+    //
+    // We didn't always have this convention however, and old persisted records will have
+    // old REDO values that point to some old LSN.
+    //
+    // The upshot is that if 'redo' is equal to the "current" LSN, there was a shutdown
+    // checkpoint record at that point in WAL, with no new WAL records after it. That case
+    // can be treated as starting from a clean shutdown. All other cases are treated as
+    // non-clean shutdown. In Neon, we don't do WAL replay at startup in either case, so
+    // that distinction doesn't matter very much. As of this writing, it only affects
+    // whether the persisted pg_stats information can be used or not.
+    //
+    // In the Checkpoint struct in the returned pg_control file, the redo pointer is
+    // always set to the LSN we're starting at, to hint that no WAL replay is required.
+    // (There's some neon-specific code in Postgres startup to make that work, though.
+    // Just setting the redo pointer is not sufficient.)
+    if Lsn(checkpoint.redo) == lsn {
+        was_shutdown = true;
+    } else {
+        checkpoint.redo = normalize_lsn(lsn, WAL_SEGMENT_SIZE).0;
+        was_shutdown = false;
+    }

-    //save new values in pg_control
+    // We use DBState_DB_SHUTDOWNED even if it was not a clean shutdown.  The
+    // neon-specific code at postgres startup ignores the state stored in the control
+    // file, similar to archive recovery in standalone PostgreSQL. Similarly, the
+    // checkPoint pointer is ignored, so just set it to 0.
    pg_control.checkPoint = 0;
    pg_control.checkPointCopy = checkpoint;
    pg_control.state = DBState_DB_SHUTDOWNED;

-    Ok((pg_control.encode(), pg_control.system_identifier))
+    Ok((pg_control.encode(), pg_control.system_identifier, was_shutdown))
 }

 pub fn get_current_timestamp() -> TimestampTz {
--- a/libs/pq_proto/src/lib.rs
+++ b/libs/pq_proto/src/lib.rs
@@ -100,7 +100,7 @@ impl StartupMessageParamsBuilder {

 #[derive(Debug, Clone, Default)]
 pub struct StartupMessageParams {
-    pub params: Bytes,
+    params: Bytes,
 }

 impl StartupMessageParams {
--- a/libs/proxy/postgres-protocol2/src/authentication/sasl.rs
+++ b/libs/proxy/postgres-protocol2/src/authentication/sasl.rs
@@ -117,7 +117,7 @@ enum Credentials<const N: usize> {
    /// A regular password as a vector of bytes.
    Password(Vec<u8>),
    /// A precomputed pair of keys.
-    Keys(ScramKeys<N>),
+    Keys(Box<ScramKeys<N>>),
 }

 enum State {
@@ -176,7 +176,7 @@ impl ScramSha256 {

    /// Constructs a new instance which will use the provided key pair for authentication.
    pub fn new_with_keys(keys: ScramKeys<32>, channel_binding: ChannelBinding) -> ScramSha256 {
-        let password = Credentials::Keys(keys);
+        let password = Credentials::Keys(keys.into());
        ScramSha256::new_inner(password, channel_binding, nonce())
    }

--- a/libs/proxy/postgres-protocol2/src/message/frontend.rs
+++ b/libs/proxy/postgres-protocol2/src/message/frontend.rs
@@ -255,34 +255,22 @@ pub fn ssl_request(buf: &mut BytesMut) {
 }

 #[inline]
-pub fn startup_message(parameters: &StartupMessageParams, buf: &mut BytesMut) -> io::Result<()> {
+pub fn startup_message<'a, I>(parameters: I, buf: &mut BytesMut) -> io::Result<()>
+where
+    I: IntoIterator<Item = (&'a str, &'a str)>,
+{
    write_body(buf, |buf| {
        // postgres protocol version 3.0(196608) in bigger-endian
        buf.put_i32(0x00_03_00_00);
-        buf.put_slice(&parameters.params);
+        for (key, value) in parameters {
+            write_cstr(key.as_bytes(), buf)?;
+            write_cstr(value.as_bytes(), buf)?;
+        }
        buf.put_u8(0);
        Ok(())
    })
 }

-#[derive(Debug, Clone, Default, PartialEq, Eq)]
-pub struct StartupMessageParams {
-    pub params: BytesMut,
-}
-
-impl StartupMessageParams {
-    /// Set parameter's value by its name.
-    pub fn insert(&mut self, name: &str, value: &str) {
-        if name.contains('\0') || value.contains('\0') {
-            panic!("startup parameter name or value contained a null")
-        }
-        self.params.put_slice(name.as_bytes());
-        self.params.put_u8(0);
-        self.params.put_slice(value.as_bytes());
-        self.params.put_u8(0);
-    }
-}
-
 #[inline]
 pub fn sync(buf: &mut BytesMut) {
    buf.put_u8(b'S');
--- a/libs/proxy/tokio-postgres2/src/codec.rs
+++ b/libs/proxy/tokio-postgres2/src/codec.rs
@@ -35,7 +35,9 @@ impl FallibleIterator for BackendMessages {
    }
 }

-pub struct PostgresCodec;
+pub struct PostgresCodec {
+    pub max_message_size: Option<usize>,
+}

 impl Encoder<FrontendMessage> for PostgresCodec {
    type Error = io::Error;
@@ -64,6 +66,15 @@ impl Decoder for PostgresCodec {
                break;
            }

+            if let Some(max) = self.max_message_size {
+                if len > max {
+                    return Err(io::Error::new(
+                        io::ErrorKind::InvalidInput,
+                        "message too large",
+                    ));
+                }
+            }
+
            match header.tag() {
                backend::NOTICE_RESPONSE_TAG
                | backend::NOTIFICATION_RESPONSE_TAG
--- a/libs/proxy/tokio-postgres2/src/config.rs
+++ b/libs/proxy/tokio-postgres2/src/config.rs
@@ -6,7 +6,6 @@ use crate::connect_raw::RawConnection;
 use crate::tls::MakeTlsConnect;
 use crate::tls::TlsConnect;
 use crate::{Client, Connection, Error};
-use postgres_protocol2::message::frontend::StartupMessageParams;
 use std::fmt;
 use std::str;
 use std::time::Duration;
@@ -15,6 +14,16 @@ use tokio::io::{AsyncRead, AsyncWrite};
 pub use postgres_protocol2::authentication::sasl::ScramKeys;
 use tokio::net::TcpStream;

+/// Properties required of a session.
+#[derive(Debug, Copy, Clone, PartialEq, Eq)]
+#[non_exhaustive]
+pub enum TargetSessionAttrs {
+    /// No special properties are required.
+    Any,
+    /// The session must allow writes.
+    ReadWrite,
+}
+
 /// TLS configuration.
 #[derive(Debug, Copy, Clone, PartialEq, Eq)]
 #[non_exhaustive]
@@ -64,20 +73,94 @@ pub enum AuthKeys {
 }

 /// Connection configuration.
+///
+/// Configuration can be parsed from libpq-style connection strings. These strings come in two formats:
+///
+/// # Key-Value
+///
+/// This format consists of space-separated key-value pairs. Values which are either the empty string or contain
+/// whitespace should be wrapped in `'`. `'` and `\` characters should be backslash-escaped.
+///
+/// ## Keys
+///
+/// * `user` - The username to authenticate with. Required.
+/// * `password` - The password to authenticate with.
+/// * `dbname` - The name of the database to connect to. Defaults to the username.
+/// * `options` - Command line options used to configure the server.
+/// * `application_name` - Sets the `application_name` parameter on the server.
+/// * `sslmode` - Controls usage of TLS. If set to `disable`, TLS will not be used. If set to `prefer`, TLS will be used
+///     if available, but not used otherwise. If set to `require`, TLS will be forced to be used. Defaults to `prefer`.
+/// * `host` - The host to connect to. On Unix platforms, if the host starts with a `/` character it is treated as the
+///     path to the directory containing Unix domain sockets. Otherwise, it is treated as a hostname. Multiple hosts
+///     can be specified, separated by commas. Each host will be tried in turn when connecting. Required if connecting
+///     with the `connect` method.
+/// * `port` - The port to connect to. Multiple ports can be specified, separated by commas. The number of ports must be
+///     either 1, in which case it will be used for all hosts, or the same as the number of hosts. Defaults to 5432 if
+///     omitted or the empty string.
+/// * `connect_timeout` - The time limit in seconds applied to each socket-level connection attempt. Note that hostnames
+///     can resolve to multiple IP addresses, and this limit is applied to each address. Defaults to no timeout.
+/// * `target_session_attrs` - Specifies requirements of the session. If set to `read-write`, the client will check that
+///     the `transaction_read_write` session parameter is set to `on`. This can be used to connect to the primary server
+///     in a database cluster as opposed to the secondary read-only mirrors. Defaults to `all`.
+/// * `channel_binding` - Controls usage of channel binding in the authentication process. If set to `disable`, channel
+///     binding will not be used. If set to `prefer`, channel binding will be used if available, but not used otherwise.
+///     If set to `require`, the authentication process will fail if channel binding is not used. Defaults to `prefer`.
+///
+/// ## Examples
+///
+/// ```not_rust
+/// host=localhost user=postgres connect_timeout=10 keepalives=0
+/// ```
+///
+/// ```not_rust
+/// host=/var/lib/postgresql,localhost port=1234 user=postgres password='password with spaces'
+/// ```
+///
+/// ```not_rust
+/// host=host1,host2,host3 port=1234,,5678 user=postgres target_session_attrs=read-write
+/// ```
+///
+/// # Url
+///
+/// This format resembles a URL with a scheme of either `postgres://` or `postgresql://`. All components are optional,
+/// and the format accepts query parameters for all of the key-value pairs described in the section above. Multiple
+/// host/port pairs can be comma-separated. Unix socket paths in the host section of the URL should be percent-encoded,
+/// as the path component of the URL specifies the database name.
+///
+/// ## Examples
+///
+/// ```not_rust
+/// postgresql://user@localhost
+/// ```
+///
+/// ```not_rust
+/// postgresql://user:password@%2Fvar%2Flib%2Fpostgresql/mydb?connect_timeout=10
+/// ```
+///
+/// ```not_rust
+/// postgresql://user@host1:1234,host2,host3:5678?target_session_attrs=read-write
+/// ```
+///
+/// ```not_rust
+/// postgresql:///mydb?user=user&host=/var/lib/postgresql
+/// ```
 #[derive(Clone, PartialEq, Eq)]
 pub struct Config {
    pub(crate) host: Host,
    pub(crate) port: u16,

+    pub(crate) user: Option<String>,
    pub(crate) password: Option<Vec<u8>>,
    pub(crate) auth_keys: Option<Box<AuthKeys>>,
+    pub(crate) dbname: Option<String>,
+    pub(crate) options: Option<String>,
+    pub(crate) application_name: Option<String>,
    pub(crate) ssl_mode: SslMode,
    pub(crate) connect_timeout: Option<Duration>,
+    pub(crate) target_session_attrs: TargetSessionAttrs,
    pub(crate) channel_binding: ChannelBinding,
-    pub(crate) server_params: StartupMessageParams,
-
-    database: bool,
-    username: bool,
+    pub(crate) replication_mode: Option<ReplicationMode>,
+    pub(crate) max_backend_message_size: Option<usize>,
 }

 impl Config {
@@ -86,15 +169,18 @@ impl Config {
        Config {
            host: Host::Tcp(host),
            port,
+            user: None,
            password: None,
            auth_keys: None,
+            dbname: None,
+            options: None,
+            application_name: None,
            ssl_mode: SslMode::Prefer,
            connect_timeout: None,
+            target_session_attrs: TargetSessionAttrs::Any,
            channel_binding: ChannelBinding::Prefer,
-            server_params: StartupMessageParams::default(),
-
-            database: false,
-            username: false,
+            replication_mode: None,
+            max_backend_message_size: None,
        }
    }

@@ -102,13 +188,14 @@ impl Config {
    ///
    /// Required.
    pub fn user(&mut self, user: &str) -> &mut Config {
-        self.set_param("user", user)
+        self.user = Some(user.to_string());
+        self
    }

    /// Gets the user to authenticate with, if one has been configured with
    /// the `user` method.
-    pub fn user_is_set(&self) -> bool {
-        self.username
+    pub fn get_user(&self) -> Option<&str> {
+        self.user.as_deref()
    }

    /// Sets the password to authenticate with.
@@ -144,26 +231,40 @@ impl Config {
    ///
    /// Defaults to the user.
    pub fn dbname(&mut self, dbname: &str) -> &mut Config {
-        self.set_param("database", dbname)
+        self.dbname = Some(dbname.to_string());
+        self
    }

    /// Gets the name of the database to connect to, if one has been configured
    /// with the `dbname` method.
-    pub fn db_is_set(&self) -> bool {
-        self.database
+    pub fn get_dbname(&self) -> Option<&str> {
+        self.dbname.as_deref()
    }

-    pub fn set_param(&mut self, name: &str, value: &str) -> &mut Config {
-        if name == "database" {
-            self.database = true;
-        } else if name == "user" {
-            self.username = true;
-        }
-
-        self.server_params.insert(name, value);
+    /// Sets command line options used to configure the server.
+    pub fn options(&mut self, options: &str) -> &mut Config {
+        self.options = Some(options.to_string());
        self
    }

+    /// Gets the command line options used to configure the server, if the
+    /// options have been set with the `options` method.
+    pub fn get_options(&self) -> Option<&str> {
+        self.options.as_deref()
+    }
+
+    /// Sets the value of the `application_name` runtime parameter.
+    pub fn application_name(&mut self, application_name: &str) -> &mut Config {
+        self.application_name = Some(application_name.to_string());
+        self
+    }
+
+    /// Gets the value of the `application_name` runtime parameter, if it has
+    /// been set with the `application_name` method.
+    pub fn get_application_name(&self) -> Option<&str> {
+        self.application_name.as_deref()
+    }
+
    /// Sets the SSL configuration.
    ///
    /// Defaults to `prefer`.
@@ -202,6 +303,23 @@ impl Config {
        self.connect_timeout.as_ref()
    }

+    /// Sets the requirements of the session.
+    ///
+    /// This can be used to connect to the primary server in a clustered database rather than one of the read-only
+    /// secondary servers. Defaults to `Any`.
+    pub fn target_session_attrs(
+        &mut self,
+        target_session_attrs: TargetSessionAttrs,
+    ) -> &mut Config {
+        self.target_session_attrs = target_session_attrs;
+        self
+    }
+
+    /// Gets the requirements of the session.
+    pub fn get_target_session_attrs(&self) -> TargetSessionAttrs {
+        self.target_session_attrs
+    }
+
    /// Sets the channel binding behavior.
    ///
    /// Defaults to `prefer`.
@@ -215,6 +333,28 @@ impl Config {
        self.channel_binding
    }

+    /// Set replication mode.
+    pub fn replication_mode(&mut self, replication_mode: ReplicationMode) -> &mut Config {
+        self.replication_mode = Some(replication_mode);
+        self
+    }
+
+    /// Get replication mode.
+    pub fn get_replication_mode(&self) -> Option<ReplicationMode> {
+        self.replication_mode
+    }
+
+    /// Set limit for backend messages size.
+    pub fn max_backend_message_size(&mut self, max_backend_message_size: usize) -> &mut Config {
+        self.max_backend_message_size = Some(max_backend_message_size);
+        self
+    }
+
+    /// Get limit for backend messages size.
+    pub fn get_max_backend_message_size(&self) -> Option<usize> {
+        self.max_backend_message_size
+    }
+
    /// Opens a connection to a PostgreSQL database.
    ///
    /// Requires the `runtime` Cargo feature (enabled by default).
@@ -252,13 +392,18 @@ impl fmt::Debug for Config {
        }

        f.debug_struct("Config")
+            .field("user", &self.user)
            .field("password", &self.password.as_ref().map(|_| Redaction {}))
+            .field("dbname", &self.dbname)
+            .field("options", &self.options)
+            .field("application_name", &self.application_name)
            .field("ssl_mode", &self.ssl_mode)
            .field("host", &self.host)
            .field("port", &self.port)
            .field("connect_timeout", &self.connect_timeout)
+            .field("target_session_attrs", &self.target_session_attrs)
            .field("channel_binding", &self.channel_binding)
-            .field("server_params", &self.server_params)
+            .field("replication", &self.replication_mode)
            .finish()
    }
 }
--- a/libs/proxy/tokio-postgres2/src/connect.rs
+++ b/libs/proxy/tokio-postgres2/src/connect.rs
@@ -1,11 +1,14 @@
 use crate::client::SocketConfig;
 use crate::codec::BackendMessage;
-use crate::config::Host;
+use crate::config::{Host, TargetSessionAttrs};
 use crate::connect_raw::connect_raw;
 use crate::connect_socket::connect_socket;
 use crate::tls::{MakeTlsConnect, TlsConnect};
-use crate::{Client, Config, Connection, Error, RawConnection};
+use crate::{Client, Config, Connection, Error, RawConnection, SimpleQueryMessage};
+use futures_util::{future, pin_mut, Future, FutureExt, Stream};
 use postgres_protocol2::message::backend::Message;
+use std::io;
+use std::task::Poll;
 use tokio::net::TcpStream;
 use tokio::sync::mpsc;

@@ -69,7 +72,47 @@ where
        .map(|m| BackendMessage::Async(Message::NoticeResponse(m)))
        .collect();

-    let connection = Connection::new(stream, delayed, parameters, receiver);
+    let mut connection = Connection::new(stream, delayed, parameters, receiver);
+
+    if let TargetSessionAttrs::ReadWrite = config.target_session_attrs {
+        let rows = client.simple_query_raw("SHOW transaction_read_only");
+        pin_mut!(rows);
+
+        let rows = future::poll_fn(|cx| {
+            if connection.poll_unpin(cx)?.is_ready() {
+                return Poll::Ready(Err(Error::closed()));
+            }
+
+            rows.as_mut().poll(cx)
+        })
+        .await?;
+        pin_mut!(rows);
+
+        loop {
+            let next = future::poll_fn(|cx| {
+                if connection.poll_unpin(cx)?.is_ready() {
+                    return Poll::Ready(Some(Err(Error::closed())));
+                }
+
+                rows.as_mut().poll_next(cx)
+            });
+
+            match next.await.transpose()? {
+                Some(SimpleQueryMessage::Row(row)) => {
+                    if row.try_get(0)? == Some("on") {
+                        return Err(Error::connect(io::Error::new(
+                            io::ErrorKind::PermissionDenied,
+                            "database does not allow writes",
+                        )));
+                    } else {
+                        break;
+                    }
+                }
+                Some(_) => {}
+                None => return Err(Error::unexpected_message()),
+            }
+        }
+    }

    Ok((client, connection))
 }
--- a/libs/proxy/tokio-postgres2/src/connect_raw.rs
+++ b/libs/proxy/tokio-postgres2/src/connect_raw.rs
@@ -1,5 +1,5 @@
 use crate::codec::{BackendMessage, BackendMessages, FrontendMessage, PostgresCodec};
-use crate::config::{self, AuthKeys, Config};
+use crate::config::{self, AuthKeys, Config, ReplicationMode};
 use crate::connect_tls::connect_tls;
 use crate::maybe_tls_stream::MaybeTlsStream;
 use crate::tls::{TlsConnect, TlsStream};
@@ -96,7 +96,12 @@ where
    let stream = connect_tls(stream, config.ssl_mode, tls).await?;

    let mut stream = StartupStream {
-        inner: Framed::new(stream, PostgresCodec),
+        inner: Framed::new(
+            stream,
+            PostgresCodec {
+                max_message_size: config.max_backend_message_size,
+            },
+        ),
        buf: BackendMessages::empty(),
        delayed_notice: Vec::new(),
    };
@@ -119,8 +124,28 @@ where
    S: AsyncRead + AsyncWrite + Unpin,
    T: AsyncRead + AsyncWrite + Unpin,
 {
+    let mut params = vec![("client_encoding", "UTF8")];
+    if let Some(user) = &config.user {
+        params.push(("user", &**user));
+    }
+    if let Some(dbname) = &config.dbname {
+        params.push(("database", &**dbname));
+    }
+    if let Some(options) = &config.options {
+        params.push(("options", &**options));
+    }
+    if let Some(application_name) = &config.application_name {
+        params.push(("application_name", &**application_name));
+    }
+    if let Some(replication_mode) = &config.replication_mode {
+        match replication_mode {
+            ReplicationMode::Physical => params.push(("replication", "true")),
+            ReplicationMode::Logical => params.push(("replication", "database")),
+        }
+    }
+
    let mut buf = BytesMut::new();
-    frontend::startup_message(&config.server_params, &mut buf).map_err(Error::encode)?;
+    frontend::startup_message(params, &mut buf).map_err(Error::encode)?;

    stream
        .send(FrontendMessage::Raw(buf.freeze()))
--- a/libs/utils/src/sync.rs
+++ b/libs/utils/src/sync.rs
@@ -1,6 +1,5 @@
 pub mod heavier_once_cell;

-pub mod duplex;
 pub mod gate;

 pub mod spsc_fold;
--- a/libs/utils/src/sync/duplex.rs
+++ b/libs/utils/src/sync/duplex.rs
@@ -1 +0,0 @@
-pub mod mpsc;
--- a/libs/utils/src/sync/duplex/mpsc.rs
+++ b/libs/utils/src/sync/duplex/mpsc.rs
@@ -1,36 +0,0 @@
-use tokio::sync::mpsc;
-
-/// A bi-directional channel.
-pub struct Duplex<S, R> {
-    pub tx: mpsc::Sender<S>,
-    pub rx: mpsc::Receiver<R>,
-}
-
-/// Creates a bi-directional channel.
-///
-/// The channel will buffer up to the provided number of messages. Once the buffer is full,
-/// attempts to send new messages will wait until a message is received from the channel.
-/// The provided buffer capacity must be at least 1.
-pub fn channel<A: Send, B: Send>(buffer: usize) -> (Duplex<A, B>, Duplex<B, A>) {
-    let (tx_a, rx_a) = mpsc::channel::<A>(buffer);
-    let (tx_b, rx_b) = mpsc::channel::<B>(buffer);
-
-    (Duplex { tx: tx_a, rx: rx_b }, Duplex { tx: tx_b, rx: rx_a })
-}
-
-impl<S: Send, R: Send> Duplex<S, R> {
-    /// Sends a value, waiting until there is capacity.
-    ///
-    /// A successful send occurs when it is determined that the other end of the channel has not hung up already.
-    pub async fn send(&self, x: S) -> Result<(), mpsc::error::SendError<S>> {
-        self.tx.send(x).await
-    }
-
-    /// Receives the next value for this receiver.
-    ///
-    /// This method returns `None` if the channel has been closed and there are
-    /// no remaining messages in the channel's buffer.
-    pub async fn recv(&mut self) -> Option<R> {
-        self.rx.recv().await
-    }
-}
--- a/pageserver/benches/bench_ingest.rs
+++ b/pageserver/benches/bench_ingest.rs
@@ -62,8 +62,10 @@ async fn ingest(
    let ctx = RequestContext::new(TaskKind::DebugTool, DownloadBehavior::Error);

    let gate = utils::sync::gate::Gate::default();
+    let entered = gate.enter().unwrap();

-    let layer = InMemoryLayer::create(conf, timeline_id, tenant_shard_id, lsn, &gate, &ctx).await?;
+    let layer =
+        InMemoryLayer::create(conf, timeline_id, tenant_shard_id, lsn, entered, &ctx).await?;

    let data = Value::Image(Bytes::from(vec![0u8; put_size]));
    let data_ser_size = data.serialized_size().unwrap() as usize;
--- a/pageserver/client/src/mgmt_api.rs
+++ b/pageserver/client/src/mgmt_api.rs
@@ -1,4 +1,4 @@
-use std::{collections::HashMap, error::Error as _};
+use std::collections::HashMap;

 use bytes::Bytes;
 use detach_ancestor::AncestorDetached;
@@ -25,10 +25,10 @@ pub struct Client {

 #[derive(thiserror::Error, Debug)]
 pub enum Error {
-    #[error("send request: {0}{}", .0.source().map(|e| format!(": {e}")).unwrap_or_default())]
+    #[error("send request: {0}")]
    SendRequest(reqwest::Error),

-    #[error("receive body: {0}{}", .0.source().map(|e| format!(": {e}")).unwrap_or_default())]
+    #[error("receive body: {0}")]
    ReceiveBody(reqwest::Error),

    #[error("receive error body: {0}")]
--- a/pageserver/ctl/src/key.rs
+++ b/pageserver/ctl/src/key.rs
@@ -345,6 +345,7 @@ impl AuxFileV2 {
                AuxFileV2::Recognized("pg_logical/replorigin_checkpoint", hash)
            }
            (2, 1) => AuxFileV2::Recognized("pg_replslot/", hash),
+            (3, 1) => AuxFileV2::Recognized("pg_stat/pgstat.stat", hash),
            (1, 0xff) => AuxFileV2::OtherWithPrefix("pg_logical/", hash),
            (0xff, 0xff) => AuxFileV2::Other(hash),
            _ => return None,
--- a/pageserver/src/aux_file.rs
+++ b/pageserver/src/aux_file.rs
@@ -39,6 +39,7 @@ fn aux_hash_to_metadata_key(dir_level1: u8, dir_level2: u8, data: &[u8]) -> Key

 const AUX_DIR_PG_LOGICAL: u8 = 0x01;
 const AUX_DIR_PG_REPLSLOT: u8 = 0x02;
+const AUX_DIR_PG_STAT: u8 = 0x03;
 const AUX_DIR_PG_UNKNOWN: u8 = 0xFF;

 /// Encode the aux file into a fixed-size key.
@@ -53,6 +54,7 @@ const AUX_DIR_PG_UNKNOWN: u8 = 0xFF;
 /// * pg_logical/replorigin_checkpoint -> 0x0103
 /// * pg_logical/others -> 0x01FF
 /// * pg_replslot/ -> 0x0201
+/// * pg_stat/pgstat.stat -> 0x0301
 /// * others -> 0xFFFF
 ///
 /// If you add new AUX files to this function, please also add a test case to `test_encoding_portable`.
@@ -75,6 +77,8 @@ pub fn encode_aux_file_key(path: &str) -> Key {
        aux_hash_to_metadata_key(AUX_DIR_PG_LOGICAL, 0xFF, fname.as_bytes())
    } else if let Some(fname) = path.strip_prefix("pg_replslot/") {
        aux_hash_to_metadata_key(AUX_DIR_PG_REPLSLOT, 0x01, fname.as_bytes())
+    } else if let Some(fname) = path.strip_prefix("pg_stat/pgstat.stat") {
+        aux_hash_to_metadata_key(AUX_DIR_PG_STAT, 0x01, fname.as_bytes())
    } else {
        if cfg!(debug_assertions) {
            warn!(
--- a/pageserver/src/basebackup.rs
+++ b/pageserver/src/basebackup.rs
@@ -255,6 +255,31 @@ where
    async fn send_tarball(mut self) -> Result<(), BasebackupError> {
        // TODO include checksum

+        // Construct the pg_control file from the persisted checkpoint and pg_control
+        // information. But we only add this to the tarball at the end, so that if the
+        // writing is interrupted half-way through, the resulting incomplete tarball will
+        // be missing the pg_control file, which prevents PostgreSQL from starting up on
+        // it. With proper error handling, you should never try to start up from an
+        // incomplete basebackup in the first place, of course, but this is a nice little
+        // extra safety measure.
+        let checkpoint_bytes = self
+            .timeline
+            .get_checkpoint(self.lsn, self.ctx)
+            .await
+            .context("failed to get checkpoint bytes")?;
+        let pg_control_bytes = self
+            .timeline
+            .get_control_file(self.lsn, self.ctx)
+            .await
+            .context("failed get control bytes")?;
+        let (pg_control_bytes, system_identifier, was_shutdown) =
+            postgres_ffi::generate_pg_control(
+                &pg_control_bytes,
+                &checkpoint_bytes,
+                self.lsn,
+                self.timeline.pg_version,
+            )?;
+
        let lazy_slru_download = self.timeline.get_lazy_slru_download() && !self.full_backup;

        let pgversion = self.timeline.pg_version;
@@ -392,6 +417,10 @@ where
                // In future we will not generate AUX record for "pg_logical/replorigin_checkpoint" at all,
                // but now we should handle (skip) it for backward compatibility.
                continue;
+            } else if path == "pg_stat/pgstat.stat" && !was_shutdown {
+                // Drop statistic in case of abnormal termination, i.e. if we're not starting from the exact LSN
+                // of a shutdown checkpoint.
+                continue;
            }
            let header = new_tar_header(&path, content.len() as u64)?;
            self.ar
@@ -453,8 +482,9 @@ where
            )))
        });

-        // Generate pg_control and bootstrap WAL segment.
-        self.add_pgcontrol_file().await?;
+        // Last, add the pg_control file and bootstrap WAL segment.
+        self.add_pgcontrol_file(pg_control_bytes, system_identifier)
+            .await?;
        self.ar.finish().await.map_err(BasebackupError::Client)?;
        debug!("all tarred up!");
        Ok(())
@@ -657,7 +687,11 @@ where
    // Add generated pg_control file and bootstrap WAL segment.
    // Also send zenith.signal file with extra bootstrap data.
    //
-    async fn add_pgcontrol_file(&mut self) -> Result<(), BasebackupError> {
+    async fn add_pgcontrol_file(
+        &mut self,
+        pg_control_bytes: Bytes,
+        system_identifier: u64,
+    ) -> Result<(), BasebackupError> {
        // add zenith.signal file
        let mut zenith_signal = String::new();
        if self.prev_record_lsn == Lsn(0) {
@@ -680,24 +714,6 @@ where
            .await
            .map_err(BasebackupError::Client)?;

-        let checkpoint_bytes = self
-            .timeline
-            .get_checkpoint(self.lsn, self.ctx)
-            .await
-            .context("failed to get checkpoint bytes")?;
-        let pg_control_bytes = self
-            .timeline
-            .get_control_file(self.lsn, self.ctx)
-            .await
-            .context("failed get control bytes")?;
-
-        let (pg_control_bytes, system_identifier) = postgres_ffi::generate_pg_control(
-            &pg_control_bytes,
-            &checkpoint_bytes,
-            self.lsn,
-            self.timeline.pg_version,
-        )?;
-
        //send pg_control
        let header = new_tar_header("global/pg_control", pg_control_bytes.len() as u64)?;
        self.ar
--- a/pageserver/src/consumption_metrics/upload.rs
+++ b/pageserver/src/consumption_metrics/upload.rs
@@ -1,4 +1,3 @@
-use std::error::Error as _;
 use std::time::SystemTime;

 use chrono::{DateTime, Utc};
@@ -351,11 +350,7 @@ impl std::fmt::Display for UploadError {

        match self {
            Rejected(code) => write!(f, "server rejected the metrics with {code}"),
-            Reqwest(e) => write!(
-                f,
-                "request failed: {e}{}",
-                e.source().map(|e| format!(": {e}")).unwrap_or_default()
-            ),
+            Reqwest(e) => write!(f, "request failed: {e}"),
            Cancelled => write!(f, "cancelled"),
        }
    }
--- a/pageserver/src/http/routes.rs
+++ b/pageserver/src/http/routes.rs
@@ -279,10 +279,7 @@ impl From<TenantStateError> for ApiError {
 impl From<GetTenantError> for ApiError {
    fn from(tse: GetTenantError) -> ApiError {
        match tse {
-            GetTenantError::NotFound(tid) => ApiError::NotFound(anyhow!("tenant {tid}").into()),
-            GetTenantError::ShardNotFound(tid) => {
-                ApiError::NotFound(anyhow!("tenant {tid}").into())
-            }
+            GetTenantError::NotFound(tid) => ApiError::NotFound(anyhow!("tenant {}", tid).into()),
            GetTenantError::NotActive(_) => {
                // Why is this not `ApiError::NotFound`?
                // Because we must be careful to never return 404 for a tenant if it does
@@ -390,16 +387,6 @@ impl From<crate::tenant::mgr::DeleteTenantError> for ApiError {
    }
 }

-impl From<crate::tenant::secondary::SecondaryTenantError> for ApiError {
-    fn from(ste: crate::tenant::secondary::SecondaryTenantError) -> ApiError {
-        use crate::tenant::secondary::SecondaryTenantError;
-        match ste {
-            SecondaryTenantError::GetTenant(gte) => gte.into(),
-            SecondaryTenantError::ShuttingDown => ApiError::ShuttingDown,
-        }
-    }
-}
-
 // Helper function to construct a TimelineInfo struct for a timeline
 async fn build_timeline_info(
    timeline: &Arc<Timeline>,
@@ -1060,11 +1047,9 @@ async fn timeline_delete_handler(
            match e {
                // GetTenantError has a built-in conversion to ApiError, but in this context we don't
                // want to treat missing tenants as 404, to avoid ambiguity with successful deletions.
-                GetTenantError::NotFound(_) | GetTenantError::ShardNotFound(_) => {
-                    ApiError::PreconditionFailed(
-                        "Requested tenant is missing".to_string().into_boxed_str(),
-                    )
-                }
+                GetTenantError::NotFound(_) => ApiError::PreconditionFailed(
+                    "Requested tenant is missing".to_string().into_boxed_str(),
+                ),
                e => e.into(),
            }
        })?;
@@ -2477,7 +2462,8 @@ async fn secondary_upload_handler(
    state
        .secondary_controller
        .upload_tenant(tenant_shard_id)
-        .await?;
+        .await
+        .map_err(ApiError::InternalServerError)?;

    json_response(StatusCode::OK, ())
 }
@@ -2592,7 +2578,7 @@ async fn secondary_download_handler(
        // Edge case: downloads aren't usually fallible: things like a missing heatmap are considered
        // okay.  We could get an error here in the unlikely edge case that the tenant
        // was detached between our check above and executing the download job.
-        Ok(Err(e)) => return Err(e.into()),
+        Ok(Err(e)) => return Err(ApiError::InternalServerError(e)),
        // A timeout is not an error: we have started the download, we're just not done
        // yet.  The caller will get a response body indicating status.
        Err(_) => StatusCode::ACCEPTED,
--- a/pageserver/src/tenant.rs
+++ b/pageserver/src/tenant.rs
@@ -3422,7 +3422,7 @@ impl Tenant {
                            r.map_err(
                            |_e: tokio::sync::watch::error::RecvError|
                                // Tenant existed but was dropped: report it as non-existent
-                                GetActiveTenantError::NotFound(GetTenantError::ShardNotFound(self.tenant_shard_id))
+                                GetActiveTenantError::NotFound(GetTenantError::NotFound(self.tenant_shard_id.tenant_id))
                        )?
                        }
                        Err(TimeoutCancellableError::Cancelled) => {
--- a/pageserver/src/tenant/ephemeral_file.rs
+++ b/pageserver/src/tenant/ephemeral_file.rs
@@ -8,8 +8,10 @@ use crate::page_cache;
 use crate::tenant::storage_layer::inmemory_layer::vectored_dio_read::File;
 use crate::virtual_file::owned_buffers_io::io_buf_aligned::IoBufAlignedMut;
 use crate::virtual_file::owned_buffers_io::slice::SliceMutExt;
+use crate::virtual_file::owned_buffers_io::util::size_tracking_writer;
 use crate::virtual_file::owned_buffers_io::write::Buffer;
 use crate::virtual_file::{self, owned_buffers_io, IoBufferMut, VirtualFile};
+use bytes::BytesMut;
 use camino::Utf8PathBuf;
 use num_traits::Num;
 use pageserver_api::shard::TenantShardId;
@@ -18,7 +20,6 @@ use tracing::error;

 use std::io;
 use std::sync::atomic::AtomicU64;
-use std::sync::Arc;
 use utils::id::TimelineId;

 pub struct EphemeralFile {
@@ -26,7 +27,10 @@ pub struct EphemeralFile {
    _timeline_id: TimelineId,
    page_cache_file_id: page_cache::FileId,
    bytes_written: u64,
-    buffered_writer: owned_buffers_io::write::BufferedWriter<IoBufferMut, VirtualFile>,
+    buffered_writer: owned_buffers_io::write::BufferedWriter<
+        BytesMut,
+        size_tracking_writer::Writer<VirtualFile>,
+    >,
    /// Gate guard is held on as long as we need to do operations in the path (delete on drop)
    _gate_guard: utils::sync::gate::GateGuard,
 }
@@ -38,9 +42,9 @@ impl EphemeralFile {
        conf: &PageServerConf,
        tenant_shard_id: TenantShardId,
        timeline_id: TimelineId,
-        gate: &utils::sync::gate::Gate,
+        gate_guard: utils::sync::gate::GateGuard,
        ctx: &RequestContext,
-    ) -> anyhow::Result<EphemeralFile> {
+    ) -> Result<EphemeralFile, io::Error> {
        static NEXT_FILENAME: AtomicU64 = AtomicU64::new(1);
        let filename_disambiguator =
            NEXT_FILENAME.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
@@ -51,17 +55,15 @@ impl EphemeralFile {
                "ephemeral-{filename_disambiguator}"
            )));

-        let file = Arc::new(
-            VirtualFile::open_with_options_v2(
-                &filename,
-                virtual_file::OpenOptions::new()
-                    .read(true)
-                    .write(true)
-                    .create(true),
-                ctx,
-            )
-            .await?,
-        );
+        let file = VirtualFile::open_with_options(
+            &filename,
+            virtual_file::OpenOptions::new()
+                .read(true)
+                .write(true)
+                .create(true),
+            ctx,
+        )
+        .await?;

        let page_cache_file_id = page_cache::next_file_id(); // XXX get rid, we're not page-caching anymore

@@ -71,12 +73,10 @@ impl EphemeralFile {
            page_cache_file_id,
            bytes_written: 0,
            buffered_writer: owned_buffers_io::write::BufferedWriter::new(
-                file,
-                || IoBufferMut::with_capacity(TAIL_SZ),
-                gate.enter()?,
-                ctx,
+                size_tracking_writer::Writer::new(file),
+                BytesMut::with_capacity(TAIL_SZ),
            ),
-            _gate_guard: gate.enter()?,
+            _gate_guard: gate_guard,
        })
    }
 }
@@ -85,7 +85,7 @@ impl Drop for EphemeralFile {
    fn drop(&mut self) {
        // unlink the file
        // we are clear to do this, because we have entered a gate
-        let path = self.buffered_writer.as_inner().path();
+        let path = self.buffered_writer.as_inner().as_inner().path();
        let res = std::fs::remove_file(path);
        if let Err(e) = res {
            if e.kind() != std::io::ErrorKind::NotFound {
@@ -132,18 +132,6 @@ impl EphemeralFile {
        srcbuf: &[u8],
        ctx: &RequestContext,
    ) -> std::io::Result<u64> {
-        let (pos, control) = self.write_raw_controlled(srcbuf, ctx).await?;
-        if let Some(control) = control {
-            control.release().await;
-        }
-        Ok(pos)
-    }
-
-    async fn write_raw_controlled(
-        &mut self,
-        srcbuf: &[u8],
-        ctx: &RequestContext,
-    ) -> std::io::Result<(u64, Option<owned_buffers_io::write::FlushControl>)> {
        let pos = self.bytes_written;

        let new_bytes_written = pos.checked_add(srcbuf.len().into_u64()).ok_or_else(|| {
@@ -157,9 +145,9 @@ impl EphemeralFile {
        })?;

        // Write the payload
-        let (nwritten, control) = self
+        let nwritten = self
            .buffered_writer
-            .write_buffered_borrowed_controlled(srcbuf, ctx)
+            .write_buffered_borrowed(srcbuf, ctx)
            .await?;
        assert_eq!(
            nwritten,
@@ -169,7 +157,7 @@ impl EphemeralFile {

        self.bytes_written = new_bytes_written;

-        Ok((pos, control))
+        Ok(pos)
    }
 }

@@ -180,12 +168,11 @@ impl super::storage_layer::inmemory_layer::vectored_dio_read::File for Ephemeral
        dst: tokio_epoll_uring::Slice<B>,
        ctx: &'a RequestContext,
    ) -> std::io::Result<(tokio_epoll_uring::Slice<B>, usize)> {
-        let submitted_offset = self.buffered_writer.bytes_submitted();
+        let file_size_tracking_writer = self.buffered_writer.as_inner();
+        let flushed_offset = file_size_tracking_writer.bytes_written();

-        let mutable = self.buffered_writer.inspect_mutable();
-        let mutable = &mutable[0..mutable.pending()];
-
-        let maybe_flushed = self.buffered_writer.inspect_maybe_flushed();
+        let buffer = self.buffered_writer.inspect_buffer();
+        let buffered = &buffer[0..buffer.pending()];

        let dst_cap = dst.bytes_total().into_u64();
        let end = {
@@ -210,42 +197,11 @@ impl super::storage_layer::inmemory_layer::vectored_dio_read::File for Ephemeral
                }
            }
        }
-
-        let (written_range, maybe_flushed_range) = {
-            if maybe_flushed.is_some() {
-                // [       written       ][ maybe_flushed ][    mutable    ]
-                //                        <-   TAIL_SZ   -><-   TAIL_SZ   ->
-                //                                         ^
-                //                                 `submitted_offset`
-                // <++++++ on disk +++++++????????????????>
-                (
-                    Range(
-                        start,
-                        std::cmp::min(end, submitted_offset.saturating_sub(TAIL_SZ as u64)),
-                    ),
-                    Range(
-                        std::cmp::max(start, submitted_offset.saturating_sub(TAIL_SZ as u64)),
-                        std::cmp::min(end, submitted_offset),
-                    ),
-                )
-            } else {
-                // [       written                        ][    mutable    ]
-                //                                         <-   TAIL_SZ   ->
-                //                                         ^
-                //                                 `submitted_offset`
-                // <++++++ on disk +++++++++++++++++++++++>
-                (
-                    Range(start, std::cmp::min(end, submitted_offset)),
-                    // zero len
-                    Range(submitted_offset, u64::MIN),
-                )
-            }
-        };
-
-        let mutable_range = Range(std::cmp::max(start, submitted_offset), end);
+        let written_range = Range(start, std::cmp::min(end, flushed_offset));
+        let buffered_range = Range(std::cmp::max(start, flushed_offset), end);

        let dst = if written_range.len() > 0 {
-            let file: &VirtualFile = self.buffered_writer.as_inner();
+            let file: &VirtualFile = file_size_tracking_writer.as_inner();
            let bounds = dst.bounds();
            let slice = file
                .read_exact_at(dst.slice(0..written_range.len().into_usize()), start, ctx)
@@ -255,21 +211,19 @@ impl super::storage_layer::inmemory_layer::vectored_dio_read::File for Ephemeral
            dst
        };

-        let dst = if maybe_flushed_range.len() > 0 {
-            let offset_in_buffer = maybe_flushed_range
+        let dst = if buffered_range.len() > 0 {
+            let offset_in_buffer = buffered_range
                .0
-                .checked_sub(submitted_offset.saturating_sub(TAIL_SZ as u64))
+                .checked_sub(flushed_offset)
                .unwrap()
                .into_usize();
-            // Checked previously the buffer is Some.
-            let maybe_flushed = maybe_flushed.unwrap();
-            let to_copy = &maybe_flushed
-                [offset_in_buffer..(offset_in_buffer + maybe_flushed_range.len().into_usize())];
+            let to_copy =
+                &buffered[offset_in_buffer..(offset_in_buffer + buffered_range.len().into_usize())];
            let bounds = dst.bounds();
            let mut view = dst.slice({
                let start = written_range.len().into_usize();
                let end = start
-                    .checked_add(maybe_flushed_range.len().into_usize())
+                    .checked_add(buffered_range.len().into_usize())
                    .unwrap();
                start..end
            });
@@ -280,28 +234,6 @@ impl super::storage_layer::inmemory_layer::vectored_dio_read::File for Ephemeral
            dst
        };

-        let dst = if mutable_range.len() > 0 {
-            let offset_in_buffer = mutable_range
-                .0
-                .checked_sub(submitted_offset)
-                .unwrap()
-                .into_usize();
-            let to_copy =
-                &mutable[offset_in_buffer..(offset_in_buffer + mutable_range.len().into_usize())];
-            let bounds = dst.bounds();
-            let mut view = dst.slice({
-                let start =
-                    written_range.len().into_usize() + maybe_flushed_range.len().into_usize();
-                let end = start.checked_add(mutable_range.len().into_usize()).unwrap();
-                start..end
-            });
-            view.as_mut_rust_slice_full_zeroed()
-                .copy_from_slice(to_copy);
-            Slice::from_buf_bounds(Slice::into_inner(view), bounds)
-        } else {
-            dst
-        };
-
        // TODO: in debug mode, randomize the remaining bytes in `dst` to catch bugs

        Ok((dst, (end - start).into_usize()))
@@ -363,7 +295,7 @@ mod tests {

        let gate = utils::sync::gate::Gate::default();

-        let file = EphemeralFile::create(conf, tenant_id, timeline_id, &gate, &ctx)
+        let file = EphemeralFile::create(conf, tenant_id, timeline_id, gate.enter().unwrap(), &ctx)
            .await
            .unwrap();

@@ -394,15 +326,14 @@ mod tests {

        let gate = utils::sync::gate::Gate::default();

-        let mut file = EphemeralFile::create(conf, tenant_id, timeline_id, &gate, &ctx)
-            .await
-            .unwrap();
+        let mut file =
+            EphemeralFile::create(conf, tenant_id, timeline_id, gate.enter().unwrap(), &ctx)
+                .await
+                .unwrap();

-        let mutable = file.buffered_writer.inspect_mutable();
-        let cap = mutable.capacity();
-        let align = mutable.align();
+        let cap = file.buffered_writer.inspect_buffer().capacity();

-        let write_nbytes = cap * 2 + cap / 2;
+        let write_nbytes = cap + cap / 2;

        let content: Vec<u8> = rand::thread_rng()
            .sample_iter(rand::distributions::Standard)
@@ -410,39 +341,30 @@ mod tests {
            .collect();

        let mut value_offsets = Vec::new();
-        for range in (0..write_nbytes)
-            .step_by(align)
-            .map(|start| start..(start + align).min(write_nbytes))
-        {
-            let off = file.write_raw(&content[range], &ctx).await.unwrap();
+        for i in 0..write_nbytes {
+            let off = file.write_raw(&content[i..i + 1], &ctx).await.unwrap();
            value_offsets.push(off);
        }

-        assert_eq!(file.len() as usize, write_nbytes);
-        for (i, range) in (0..write_nbytes)
-            .step_by(align)
-            .map(|start| start..(start + align).min(write_nbytes))
-            .enumerate()
-        {
-            assert_eq!(value_offsets[i], range.start.into_u64());
-            let buf = IoBufferMut::with_capacity(range.len());
+        assert!(file.len() as usize == write_nbytes);
+        for i in 0..write_nbytes {
+            assert_eq!(value_offsets[i], i.into_u64());
+            let buf = IoBufferMut::with_capacity(1);
            let (buf_slice, nread) = file
-                .read_exact_at_eof_ok(range.start.into_u64(), buf.slice_full(), &ctx)
+                .read_exact_at_eof_ok(i.into_u64(), buf.slice_full(), &ctx)
                .await
                .unwrap();
            let buf = buf_slice.into_inner();
-            assert_eq!(nread, range.len());
-            assert_eq!(&buf, &content[range]);
+            assert_eq!(nread, 1);
+            assert_eq!(&buf, &content[i..i + 1]);
        }

-        let file_contents = std::fs::read(file.buffered_writer.as_inner().path()).unwrap();
-        assert!(file_contents == content[0..cap * 2]);
+        let file_contents =
+            std::fs::read(file.buffered_writer.as_inner().as_inner().path()).unwrap();
+        assert_eq!(file_contents, &content[0..cap]);

-        let maybe_flushed_buffer_contents = file.buffered_writer.inspect_maybe_flushed().unwrap();
-        assert_eq!(&maybe_flushed_buffer_contents[..], &content[cap..cap * 2]);
-
-        let mutable_buffer_contents = file.buffered_writer.inspect_mutable();
-        assert_eq!(mutable_buffer_contents, &content[cap * 2..write_nbytes]);
+        let buffer_contents = file.buffered_writer.inspect_buffer();
+        assert_eq!(buffer_contents, &content[cap..write_nbytes]);
    }

    #[tokio::test]
@@ -451,16 +373,16 @@ mod tests {

        let gate = utils::sync::gate::Gate::default();

-        let mut file = EphemeralFile::create(conf, tenant_id, timeline_id, &gate, &ctx)
-            .await
-            .unwrap();
+        let mut file =
+            EphemeralFile::create(conf, tenant_id, timeline_id, gate.enter().unwrap(), &ctx)
+                .await
+                .unwrap();

-        // mutable buffer and maybe_flushed buffer each has `cap` bytes.
-        let cap = file.buffered_writer.inspect_mutable().capacity();
+        let cap = file.buffered_writer.inspect_buffer().capacity();

        let content: Vec<u8> = rand::thread_rng()
            .sample_iter(rand::distributions::Standard)
-            .take(cap * 2 + cap / 2)
+            .take(cap + cap / 2)
            .collect();

        file.write_raw(&content, &ctx).await.unwrap();
@@ -468,21 +390,23 @@ mod tests {
        // assert the state is as this test expects it to be
        assert_eq!(
            &file.load_to_io_buf(&ctx).await.unwrap(),
-            &content[0..cap * 2 + cap / 2]
+            &content[0..cap + cap / 2]
        );
-        let md = file.buffered_writer.as_inner().path().metadata().unwrap();
+        let md = file
+            .buffered_writer
+            .as_inner()
+            .as_inner()
+            .path()
+            .metadata()
+            .unwrap();
        assert_eq!(
            md.len(),
-            2 * cap.into_u64(),
-            "buffered writer requires one write to be flushed if we write 2.5x buffer capacity"
+            cap.into_u64(),
+            "buffered writer does one write if we write 1.5x buffer capacity"
        );
        assert_eq!(
-            &file.buffered_writer.inspect_maybe_flushed().unwrap()[0..cap],
-            &content[cap..cap * 2]
-        );
-        assert_eq!(
-            &file.buffered_writer.inspect_mutable()[0..cap / 2],
-            &content[cap * 2..cap * 2 + cap / 2]
+            &file.buffered_writer.inspect_buffer()[0..cap / 2],
+            &content[cap..cap + cap / 2]
        );
    }

@@ -498,19 +422,19 @@ mod tests {

        let gate = utils::sync::gate::Gate::default();

-        let mut file = EphemeralFile::create(conf, tenant_id, timeline_id, &gate, &ctx)
-            .await
-            .unwrap();
+        let mut file =
+            EphemeralFile::create(conf, tenant_id, timeline_id, gate.enter().unwrap(), &ctx)
+                .await
+                .unwrap();
+
+        let cap = file.buffered_writer.inspect_buffer().capacity();

-        let mutable = file.buffered_writer.inspect_mutable();
-        let cap = mutable.capacity();
-        let align = mutable.align();
        let content: Vec<u8> = rand::thread_rng()
            .sample_iter(rand::distributions::Standard)
-            .take(cap * 2 + cap / 2)
+            .take(cap + cap / 2)
            .collect();

-        let (_, control) = file.write_raw_controlled(&content, &ctx).await.unwrap();
+        file.write_raw(&content, &ctx).await.unwrap();

        let test_read = |start: usize, len: usize| {
            let file = &file;
@@ -530,38 +454,16 @@ mod tests {
            }
        };

-        let test_read_all_offset_combinations = || {
-            async move {
-                test_read(align, align).await;
-                // border onto edge of file
-                test_read(cap - align, align).await;
-                // read across file and buffer
-                test_read(cap - align, 2 * align).await;
-                // stay from start of maybe flushed buffer
-                test_read(cap, align).await;
-                // completely within maybe flushed buffer
-                test_read(cap + align, align).await;
-                // border onto edge of maybe flushed buffer.
-                test_read(cap * 2 - align, align).await;
-                // read across maybe flushed and mutable buffer
-                test_read(cap * 2 - align, 2 * align).await;
-                // read across three segments
-                test_read(cap - align, cap + 2 * align).await;
-                // completely within mutable buffer
-                test_read(cap * 2 + align, align).await;
-            }
-        };
-
        // completely within the file range
-        assert!(align < cap, "test assumption");
-        assert!(cap % align == 0);
-
-        // test reads at different flush stages.
-        let not_started = control.unwrap().into_not_started();
-        test_read_all_offset_combinations().await;
-        let in_progress = not_started.ready_to_flush();
-        test_read_all_offset_combinations().await;
-        in_progress.wait_until_flush_is_done().await;
-        test_read_all_offset_combinations().await;
+        assert!(20 < cap, "test assumption");
+        test_read(10, 10).await;
+        // border onto edge of file
+        test_read(cap - 10, 10).await;
+        // read across file and buffer
+        test_read(cap - 10, 20).await;
+        // stay from start of buffer
+        test_read(cap, 10).await;
+        // completely within buffer
+        test_read(cap + 10, 10).await;
    }
 }
--- a/pageserver/src/tenant/mgr.rs
+++ b/pageserver/src/tenant/mgr.rs
@@ -894,7 +894,7 @@ impl TenantManager {
            Some(TenantSlot::Attached(tenant)) => Ok(Arc::clone(tenant)),
            Some(TenantSlot::InProgress(_)) => Err(GetTenantError::NotActive(tenant_shard_id)),
            None | Some(TenantSlot::Secondary(_)) => {
-                Err(GetTenantError::ShardNotFound(tenant_shard_id))
+                Err(GetTenantError::NotFound(tenant_shard_id.tenant_id))
            }
        }
    }
@@ -2258,9 +2258,6 @@ pub(crate) enum GetTenantError {
    #[error("Tenant {0} not found")]
    NotFound(TenantId),

-    #[error("Tenant {0} not found")]
-    ShardNotFound(TenantShardId),
-
    #[error("Tenant {0} is not active")]
    NotActive(TenantShardId),

--- a/pageserver/src/tenant/remote_timeline_client.rs
+++ b/pageserver/src/tenant/remote_timeline_client.rs
@@ -681,7 +681,6 @@ impl RemoteTimelineClient {
        layer_file_name: &LayerName,
        layer_metadata: &LayerFileMetadata,
        local_path: &Utf8Path,
-        gate: &utils::sync::gate::Gate,
        cancel: &CancellationToken,
        ctx: &RequestContext,
    ) -> Result<u64, DownloadError> {
@@ -701,7 +700,6 @@ impl RemoteTimelineClient {
                layer_file_name,
                layer_metadata,
                local_path,
-                gate,
                cancel,
                ctx,
            )
--- a/pageserver/src/tenant/remote_timeline_client/download.rs
+++ b/pageserver/src/tenant/remote_timeline_client/download.rs
@@ -26,6 +26,8 @@ use crate::span::{
 use crate::tenant::remote_timeline_client::{remote_layer_path, remote_timelines_path};
 use crate::tenant::storage_layer::LayerName;
 use crate::tenant::Generation;
+#[cfg_attr(target_os = "macos", allow(unused_imports))]
+use crate::virtual_file::owned_buffers_io::io_buf_ext::IoBufExt;
 use crate::virtual_file::{on_fatal_io_error, MaybeFatalIo, VirtualFile};
 use crate::TEMP_FILE_SUFFIX;
 use remote_storage::{
@@ -58,7 +60,6 @@ pub async fn download_layer_file<'a>(
    layer_file_name: &'a LayerName,
    layer_metadata: &'a LayerFileMetadata,
    local_path: &Utf8Path,
-    gate: &utils::sync::gate::Gate,
    cancel: &CancellationToken,
    ctx: &RequestContext,
 ) -> Result<u64, DownloadError> {
@@ -87,9 +88,7 @@ pub async fn download_layer_file<'a>(
    let temp_file_path = path_with_suffix_extension(local_path, TEMP_DOWNLOAD_EXTENSION);

    let bytes_amount = download_retry(
-        || async {
-            download_object(storage, &remote_path, &temp_file_path, gate, cancel, ctx).await
-        },
+        || async { download_object(storage, &remote_path, &temp_file_path, cancel, ctx).await },
        &format!("download {remote_path:?}"),
        cancel,
    )
@@ -149,7 +148,6 @@ async fn download_object<'a>(
    storage: &'a GenericRemoteStorage,
    src_path: &RemotePath,
    dst_path: &Utf8PathBuf,
-    #[cfg_attr(target_os = "macos", allow(unused_variables))] gate: &utils::sync::gate::Gate,
    cancel: &CancellationToken,
    #[cfg_attr(target_os = "macos", allow(unused_variables))] ctx: &RequestContext,
 ) -> Result<u64, DownloadError> {
@@ -207,18 +205,13 @@ async fn download_object<'a>(
        }
        #[cfg(target_os = "linux")]
        crate::virtual_file::io_engine::IoEngine::TokioEpollUring => {
-            use crate::virtual_file::owned_buffers_io;
-            use crate::virtual_file::IoBufferMut;
-            use std::sync::Arc;
+            use crate::virtual_file::owned_buffers_io::{self, util::size_tracking_writer};
+            use bytes::BytesMut;
            async {
-                let destination_file = Arc::new(
-                    VirtualFile::create(dst_path, ctx)
-                        .await
-                        .with_context(|| {
-                            format!("create a destination file for layer '{dst_path}'")
-                        })
-                        .map_err(DownloadError::Other)?,
-                );
+                let destination_file = VirtualFile::create(dst_path, ctx)
+                    .await
+                    .with_context(|| format!("create a destination file for layer '{dst_path}'"))
+                    .map_err(DownloadError::Other)?;

                let mut download = storage
                    .download(src_path, &DownloadOpts::default(), cancel)
@@ -226,16 +219,14 @@ async fn download_object<'a>(

                pausable_failpoint!("before-downloading-layer-stream-pausable");

-                let mut buffered = owned_buffers_io::write::BufferedWriter::<IoBufferMut, _>::new(
-                    destination_file,
-                    || IoBufferMut::with_capacity(super::BUFFER_SIZE),
-                    gate.enter().map_err(|_| DownloadError::Cancelled)?,
-                    ctx,
-                );
-
                // TODO: use vectored write (writev) once supported by tokio-epoll-uring.
                // There's chunks_vectored() on the stream.
                let (bytes_amount, destination_file) = async {
+                    let size_tracking = size_tracking_writer::Writer::new(destination_file);
+                    let mut buffered = owned_buffers_io::write::BufferedWriter::<BytesMut, _>::new(
+                        size_tracking,
+                        BytesMut::with_capacity(super::BUFFER_SIZE),
+                    );
                    while let Some(res) =
                        futures::StreamExt::next(&mut download.download_stream).await
                    {
@@ -243,10 +234,10 @@ async fn download_object<'a>(
                            Ok(chunk) => chunk,
                            Err(e) => return Err(e),
                        };
-                        buffered.write_buffered_borrowed(&chunk, ctx).await?;
+                        buffered.write_buffered(chunk.slice_len(), ctx).await?;
                    }
-                    let inner = buffered.flush_and_into_inner(ctx).await?;
-                    Ok(inner)
+                    let size_tracking = buffered.flush_and_into_inner(ctx).await?;
+                    Ok(size_tracking.into_inner())
                }
                .await?;

--- a/pageserver/src/tenant/secondary.rs
+++ b/pageserver/src/tenant/secondary.rs
@@ -22,7 +22,6 @@ use super::{
    mgr::TenantManager,
    span::debug_assert_current_span_has_tenant_id,
    storage_layer::LayerName,
-    GetTenantError,
 };

 use crate::metrics::SECONDARY_RESIDENT_PHYSICAL_SIZE;
@@ -67,21 +66,7 @@ struct CommandRequest<T> {
 }

 struct CommandResponse {
-    result: Result<(), SecondaryTenantError>,
-}
-
-#[derive(thiserror::Error, Debug)]
-pub(crate) enum SecondaryTenantError {
-    #[error("{0}")]
-    GetTenant(GetTenantError),
-    #[error("shutting down")]
-    ShuttingDown,
-}
-
-impl From<GetTenantError> for SecondaryTenantError {
-    fn from(gte: GetTenantError) -> Self {
-        Self::GetTenant(gte)
-    }
+    result: anyhow::Result<()>,
 }

 // Whereas [`Tenant`] represents an attached tenant, this type represents the work
@@ -300,7 +285,7 @@ impl SecondaryController {
        &self,
        queue: &tokio::sync::mpsc::Sender<CommandRequest<T>>,
        payload: T,
-    ) -> Result<(), SecondaryTenantError> {
+    ) -> anyhow::Result<()> {
        let (response_tx, response_rx) = tokio::sync::oneshot::channel();

        queue
@@ -309,26 +294,20 @@ impl SecondaryController {
                response_tx,
            })
            .await
-            .map_err(|_| SecondaryTenantError::ShuttingDown)?;
+            .map_err(|_| anyhow::anyhow!("Receiver shut down"))?;

        let response = response_rx
            .await
-            .map_err(|_| SecondaryTenantError::ShuttingDown)?;
+            .map_err(|_| anyhow::anyhow!("Request dropped"))?;

        response.result
    }

-    pub(crate) async fn upload_tenant(
-        &self,
-        tenant_shard_id: TenantShardId,
-    ) -> Result<(), SecondaryTenantError> {
+    pub async fn upload_tenant(&self, tenant_shard_id: TenantShardId) -> anyhow::Result<()> {
        self.dispatch(&self.upload_req_tx, UploadCommand::Upload(tenant_shard_id))
            .await
    }
-    pub(crate) async fn download_tenant(
-        &self,
-        tenant_shard_id: TenantShardId,
-    ) -> Result<(), SecondaryTenantError> {
+    pub async fn download_tenant(&self, tenant_shard_id: TenantShardId) -> anyhow::Result<()> {
        self.dispatch(
            &self.download_req_tx,
            DownloadCommand::Download(tenant_shard_id),
--- a/pageserver/src/tenant/secondary/downloader.rs
+++ b/pageserver/src/tenant/secondary/downloader.rs
@@ -35,7 +35,7 @@ use super::{
        self, period_jitter, period_warmup, Completion, JobGenerator, SchedulingResult,
        TenantBackgroundJobs,
    },
-    GetTenantError, SecondaryTenant, SecondaryTenantError,
+    SecondaryTenant,
 };

 use crate::tenant::{
@@ -470,16 +470,15 @@ impl JobGenerator<PendingDownload, RunningDownload, CompleteDownload, DownloadCo
        result
    }

-    fn on_command(
-        &mut self,
-        command: DownloadCommand,
-    ) -> Result<PendingDownload, SecondaryTenantError> {
+    fn on_command(&mut self, command: DownloadCommand) -> anyhow::Result<PendingDownload> {
        let tenant_shard_id = command.get_tenant_shard_id();

        let tenant = self
            .tenant_manager
-            .get_secondary_tenant_shard(*tenant_shard_id)
-            .ok_or(GetTenantError::ShardNotFound(*tenant_shard_id))?;
+            .get_secondary_tenant_shard(*tenant_shard_id);
+        let Some(tenant) = tenant else {
+            return Err(anyhow::anyhow!("Not found or not in Secondary mode"));
+        };

        Ok(PendingDownload {
            target_time: None,
@@ -1183,7 +1182,6 @@ impl<'a> TenantDownloader<'a> {
            &layer.name,
            &layer.metadata,
            &local_path,
-            &self.secondary_state.gate,
            &self.secondary_state.cancel,
            ctx,
        )
--- a/pageserver/src/tenant/secondary/heatmap_uploader.rs
+++ b/pageserver/src/tenant/secondary/heatmap_uploader.rs
@@ -28,7 +28,7 @@ use super::{
        self, period_jitter, period_warmup, JobGenerator, RunningJob, SchedulingResult,
        TenantBackgroundJobs,
    },
-    CommandRequest, SecondaryTenantError, UploadCommand,
+    CommandRequest, UploadCommand,
 };
 use tokio_util::sync::CancellationToken;
 use tracing::{info_span, instrument, Instrument};
@@ -279,10 +279,7 @@ impl JobGenerator<UploadPending, WriteInProgress, WriteComplete, UploadCommand>
        }.instrument(info_span!(parent: None, "heatmap_upload", tenant_id=%tenant_shard_id.tenant_id, shard_id=%tenant_shard_id.shard_slug()))))
    }

-    fn on_command(
-        &mut self,
-        command: UploadCommand,
-    ) -> Result<UploadPending, SecondaryTenantError> {
+    fn on_command(&mut self, command: UploadCommand) -> anyhow::Result<UploadPending> {
        let tenant_shard_id = command.get_tenant_shard_id();

        tracing::info!(
@@ -290,7 +287,8 @@ impl JobGenerator<UploadPending, WriteInProgress, WriteComplete, UploadCommand>
            "Starting heatmap write on command");
        let tenant = self
            .tenant_manager
-            .get_attached_tenant_shard(*tenant_shard_id)?;
+            .get_attached_tenant_shard(*tenant_shard_id)
+            .map_err(|e| anyhow::anyhow!(e))?;
        if !tenant.is_active() {
            return Err(GetTenantError::NotActive(*tenant_shard_id).into());
        }
--- a/pageserver/src/tenant/secondary/scheduler.rs
+++ b/pageserver/src/tenant/secondary/scheduler.rs
@@ -12,7 +12,7 @@ use tokio::task::JoinSet;
 use tokio_util::sync::CancellationToken;
 use utils::{completion::Barrier, yielding_loop::yielding_loop};

-use super::{CommandRequest, CommandResponse, SecondaryTenantError};
+use super::{CommandRequest, CommandResponse};

 /// Scheduling interval is the time between calls to JobGenerator::schedule.
 /// When we schedule jobs, the job generator may provide a hint of its preferred
@@ -112,7 +112,7 @@ where

    /// Called when a command is received.  A job will be spawned immediately if the return
    /// value is Some, ignoring concurrency limits and the pending queue.
-    fn on_command(&mut self, cmd: CMD) -> Result<PJ, SecondaryTenantError>;
+    fn on_command(&mut self, cmd: CMD) -> anyhow::Result<PJ>;
 }

 /// [`JobGenerator`] returns this to provide pending jobs, and hints about scheduling
--- a/pageserver/src/tenant/storage_layer/delta_layer.rs
+++ b/pageserver/src/tenant/storage_layer/delta_layer.rs
@@ -52,8 +52,8 @@ use camino::{Utf8Path, Utf8PathBuf};
 use futures::StreamExt;
 use itertools::Itertools;
 use pageserver_api::config::MaxVectoredReadBytes;
-use pageserver_api::key::DBDIR_KEY;
 use pageserver_api::key::{Key, KEY_SIZE};
+use pageserver_api::key::{AUX_FILES_KEY, DBDIR_KEY};
 use pageserver_api::keyspace::KeySpace;
 use pageserver_api::models::ImageCompressionAlgorithm;
 use pageserver_api::shard::TenantShardId;
@@ -969,7 +969,11 @@ impl DeltaLayerInner {
                .as_slice()
                .iter()
                .filter_map(|(_, blob_meta)| {
-                    if blob_meta.key.is_rel_dir_key() || blob_meta.key == DBDIR_KEY {
+                    if blob_meta.key.is_rel_dir_key()
+                        || blob_meta.key == DBDIR_KEY
+                        || blob_meta.key == AUX_FILES_KEY
+                        || blob_meta.key.is_aux_file_key()
+                    {
                        // The size of values for these keys is unbounded and can
                        // grow very large in pathological cases.
                        None
--- a/pageserver/src/tenant/storage_layer/image_layer.rs
+++ b/pageserver/src/tenant/storage_layer/image_layer.rs
@@ -49,8 +49,8 @@ use camino::{Utf8Path, Utf8PathBuf};
 use hex;
 use itertools::Itertools;
 use pageserver_api::config::MaxVectoredReadBytes;
-use pageserver_api::key::DBDIR_KEY;
 use pageserver_api::key::{Key, KEY_SIZE};
+use pageserver_api::key::{AUX_FILES_KEY, DBDIR_KEY};
 use pageserver_api::keyspace::KeySpace;
 use pageserver_api::shard::{ShardIdentity, TenantShardId};
 use pageserver_api::value::Value;
@@ -591,7 +591,11 @@ impl ImageLayerInner {
                    .as_slice()
                    .iter()
                    .filter_map(|(_, blob_meta)| {
-                        if blob_meta.key.is_rel_dir_key() || blob_meta.key == DBDIR_KEY {
+                        if blob_meta.key.is_rel_dir_key()
+                            || blob_meta.key == DBDIR_KEY
+                            || blob_meta.key == AUX_FILES_KEY
+                            || blob_meta.key.is_aux_file_key()
+                        {
                            // The size of values for these keys is unbounded and can
                            // grow very large in pathological cases.
                            None
--- a/pageserver/src/tenant/storage_layer/inmemory_layer.rs
+++ b/pageserver/src/tenant/storage_layer/inmemory_layer.rs
@@ -555,12 +555,13 @@ impl InMemoryLayer {
        timeline_id: TimelineId,
        tenant_shard_id: TenantShardId,
        start_lsn: Lsn,
-        gate: &utils::sync::gate::Gate,
+        gate_guard: utils::sync::gate::GateGuard,
        ctx: &RequestContext,
    ) -> Result<InMemoryLayer> {
        trace!("initializing new empty InMemoryLayer for writing on timeline {timeline_id} at {start_lsn}");

-        let file = EphemeralFile::create(conf, tenant_shard_id, timeline_id, gate, ctx).await?;
+        let file =
+            EphemeralFile::create(conf, tenant_shard_id, timeline_id, gate_guard, ctx).await?;
        let key = InMemoryLayerFileId(file.page_cache_file_id());

        Ok(InMemoryLayer {
--- a/pageserver/src/tenant/storage_layer/layer.rs
+++ b/pageserver/src/tenant/storage_layer/layer.rs
@@ -1149,7 +1149,6 @@ impl LayerInner {
                &self.desc.layer_name(),
                &self.metadata(),
                &self.path,
-                &timeline.gate,
                &timeline.cancel,
                ctx,
            )
--- a/pageserver/src/tenant/timeline.rs
+++ b/pageserver/src/tenant/timeline.rs
@@ -3455,6 +3455,7 @@ impl Timeline {
        ctx: &RequestContext,
    ) -> anyhow::Result<Arc<InMemoryLayer>> {
        let mut guard = self.layers.write().await;
+        let gate_guard = self.gate.enter().context("enter gate for inmem layer")?;

        let last_record_lsn = self.get_last_record_lsn();
        ensure!(
@@ -3471,7 +3472,7 @@ impl Timeline {
                self.conf,
                self.timeline_id,
                self.tenant_shard_id,
-                &self.gate,
+                gate_guard,
                ctx,
            )
            .await?;
@@ -4238,10 +4239,12 @@ impl Timeline {
            // Normal path: we have written some data into the new image layer for this
            // partition, so flush it to disk.
            let (desc, path) = image_layer_writer.finish(ctx).await?;
+            let file_size = desc.file_size;
            let image_layer = Layer::finish_creating(self.conf, self, desc, &path)?;
            info!(
-                "created image layer for metadata {}",
-                image_layer.local_path()
+                "created image layer for metadata {} size {}",
+                image_layer.local_path(),
+                file_size,
            );
            Ok(ImageLayerCreationOutcome {
                image: Some(image_layer),
--- a/pageserver/src/tenant/timeline/layer_manager.rs
+++ b/pageserver/src/tenant/timeline/layer_manager.rs
@@ -182,7 +182,7 @@ impl OpenLayerManager {
        conf: &'static PageServerConf,
        timeline_id: TimelineId,
        tenant_shard_id: TenantShardId,
-        gate: &utils::sync::gate::Gate,
+        gate_guard: utils::sync::gate::GateGuard,
        ctx: &RequestContext,
    ) -> anyhow::Result<Arc<InMemoryLayer>> {
        ensure!(lsn.is_aligned());
@@ -212,9 +212,15 @@ impl OpenLayerManager {
                lsn
            );

-            let new_layer =
-                InMemoryLayer::create(conf, timeline_id, tenant_shard_id, start_lsn, gate, ctx)
-                    .await?;
+            let new_layer = InMemoryLayer::create(
+                conf,
+                timeline_id,
+                tenant_shard_id,
+                start_lsn,
+                gate_guard,
+                ctx,
+            )
+            .await?;
            let layer = Arc::new(new_layer);

            self.layer_map.open_layer = Some(layer.clone());
--- a/pageserver/src/virtual_file.rs
+++ b/pageserver/src/virtual_file.rs
@@ -20,7 +20,7 @@ use camino::{Utf8Path, Utf8PathBuf};
 use once_cell::sync::OnceCell;
 use owned_buffers_io::aligned_buffer::buffer::AlignedBuffer;
 use owned_buffers_io::aligned_buffer::{AlignedBufferMut, AlignedSlice, ConstAlign};
-use owned_buffers_io::io_buf_aligned::{IoBufAligned, IoBufAlignedMut};
+use owned_buffers_io::io_buf_aligned::IoBufAlignedMut;
 use owned_buffers_io::io_buf_ext::FullSlice;
 use pageserver_api::config::defaults::DEFAULT_IO_BUFFER_ALIGNMENT;
 use pageserver_api::shard::TenantShardId;
@@ -63,6 +63,9 @@ pub(crate) mod owned_buffers_io {
    pub(crate) mod io_buf_ext;
    pub(crate) mod slice;
    pub(crate) mod write;
+    pub(crate) mod util {
+        pub(crate) mod size_tracking_writer;
+    }
 }

 #[derive(Debug)]
@@ -218,7 +221,7 @@ impl VirtualFile {
        self.inner.read_exact_at_page(page, offset, ctx).await
    }

-    pub async fn write_all_at<Buf: IoBufAligned + Send>(
+    pub async fn write_all_at<Buf: IoBuf + Send>(
        &self,
        buf: FullSlice<Buf>,
        offset: u64,
@@ -1322,14 +1325,14 @@ impl Drop for VirtualFileInner {
 }

 impl OwnedAsyncWriter for VirtualFile {
-    async fn write_all_at<Buf: IoBufAligned + Send>(
-        &self,
+    #[inline(always)]
+    async fn write_all<Buf: IoBuf + Send>(
+        &mut self,
        buf: FullSlice<Buf>,
-        offset: u64,
        ctx: &RequestContext,
-    ) -> std::io::Result<FullSlice<Buf>> {
-        let (buf, res) = VirtualFile::write_all_at(self, buf, offset, ctx).await;
-        res.map(|_| buf)
+    ) -> std::io::Result<(usize, FullSlice<Buf>)> {
+        let (buf, res) = VirtualFile::write_all(self, buf, ctx).await;
+        res.map(move |v| (v, buf))
    }
 }

@@ -1448,7 +1451,7 @@ mod tests {
                }
            }
        }
-        async fn write_all_at<Buf: IoBufAligned + Send>(
+        async fn write_all_at<Buf: IoBuf + Send>(
            &self,
            buf: FullSlice<Buf>,
            offset: u64,
@@ -1591,7 +1594,6 @@ mod tests {
            &ctx,
        )
        .await?;
-
        file_a
            .write_all(b"foobar".to_vec().slice_len(), &ctx)
            .await?;
@@ -1650,10 +1652,10 @@ mod tests {
        )
        .await?;
        file_b
-            .write_all_at(IoBuffer::from(b"BAR").slice_len(), 3, &ctx)
+            .write_all_at(b"BAR".to_vec().slice_len(), 3, &ctx)
            .await?;
        file_b
-            .write_all_at(IoBuffer::from(b"FOO").slice_len(), 0, &ctx)
+            .write_all_at(b"FOO".to_vec().slice_len(), 0, &ctx)
            .await?;

        assert_eq!(file_b.read_string_at(2, 3, &ctx).await?, "OBA");
--- a/pageserver/src/virtual_file/owned_buffers_io/aligned_buffer/alignment.rs
+++ b/pageserver/src/virtual_file/owned_buffers_io/aligned_buffer/alignment.rs
@@ -4,7 +4,7 @@ pub trait Alignment: std::marker::Unpin + 'static {
 }

 /// Alignment at compile time.
-#[derive(Debug, Clone, Copy)]
+#[derive(Debug)]
 pub struct ConstAlign<const A: usize>;

 impl<const A: usize> Alignment for ConstAlign<A> {
@@ -14,7 +14,7 @@ impl<const A: usize> Alignment for ConstAlign<A> {
 }

 /// Alignment at run time.
-#[derive(Debug, Clone, Copy)]
+#[derive(Debug)]
 pub struct RuntimeAlign {
    align: usize,
 }
--- a/pageserver/src/virtual_file/owned_buffers_io/aligned_buffer/buffer.rs
+++ b/pageserver/src/virtual_file/owned_buffers_io/aligned_buffer/buffer.rs
@@ -3,10 +3,9 @@ use std::{
    sync::Arc,
 };

-use super::{alignment::Alignment, raw::RawAlignedBuffer, AlignedBufferMut, ConstAlign};
+use super::{alignment::Alignment, raw::RawAlignedBuffer};

 /// An shared, immutable aligned buffer type.
-#[derive(Clone, Debug)]
 pub struct AlignedBuffer<A: Alignment> {
    /// Shared raw buffer.
    raw: Arc<RawAlignedBuffer<A>>,
@@ -87,13 +86,6 @@ impl<A: Alignment> AlignedBuffer<A> {
            range: begin..end,
        }
    }
-
-    /// Returns the mutable aligned buffer, if the immutable aligned buffer
-    /// has exactly one strong reference. Otherwise returns `None`.
-    pub fn into_mut(self) -> Option<AlignedBufferMut<A>> {
-        let raw = Arc::into_inner(self.raw)?;
-        Some(AlignedBufferMut::from_raw(raw))
-    }
 }

 impl<A: Alignment> Deref for AlignedBuffer<A> {
@@ -116,14 +108,6 @@ impl<A: Alignment> PartialEq<[u8]> for AlignedBuffer<A> {
    }
 }

-impl<const A: usize, const N: usize> From<&[u8; N]> for AlignedBuffer<ConstAlign<A>> {
-    fn from(value: &[u8; N]) -> Self {
-        let mut buf = AlignedBufferMut::with_capacity(N);
-        buf.extend_from_slice(value);
-        buf.freeze()
-    }
-}
-
 /// SAFETY: the underlying buffer references a stable memory region.
 unsafe impl<A: Alignment> tokio_epoll_uring::IoBuf for AlignedBuffer<A> {
    fn stable_ptr(&self) -> *const u8 {
--- a/pageserver/src/virtual_file/owned_buffers_io/aligned_buffer/buffer_mut.rs
+++ b/pageserver/src/virtual_file/owned_buffers_io/aligned_buffer/buffer_mut.rs
@@ -1,7 +1,4 @@
-use std::{
-    mem::MaybeUninit,
-    ops::{Deref, DerefMut},
-};
+use std::ops::{Deref, DerefMut};

 use super::{
    alignment::{Alignment, ConstAlign},
@@ -49,11 +46,6 @@ impl<const A: usize> AlignedBufferMut<ConstAlign<A>> {
 }

 impl<A: Alignment> AlignedBufferMut<A> {
-    /// Constructs a mutable aligned buffer from raw.
-    pub(super) fn from_raw(raw: RawAlignedBuffer<A>) -> Self {
-        AlignedBufferMut { raw }
-    }
-
    /// Returns the total number of bytes the buffer can hold.
    #[inline]
    pub fn capacity(&self) -> usize {
@@ -136,39 +128,6 @@ impl<A: Alignment> AlignedBufferMut<A> {
        let len = self.len();
        AlignedBuffer::from_raw(self.raw, 0..len)
    }
-
-    /// Clones and appends all elements in a slice to the buffer. Reserves additional capacity as needed.
-    #[inline]
-    pub fn extend_from_slice(&mut self, extend: &[u8]) {
-        let cnt = extend.len();
-        self.reserve(cnt);
-
-        // SAFETY: we already reserved additional `cnt` bytes, safe to perform memcpy.
-        unsafe {
-            let dst = self.spare_capacity_mut();
-            // Reserved above
-            debug_assert!(dst.len() >= cnt);
-
-            core::ptr::copy_nonoverlapping(extend.as_ptr(), dst.as_mut_ptr().cast(), cnt);
-        }
-        // SAFETY: We do have at least `cnt` bytes remaining before advance.
-        unsafe {
-            bytes::BufMut::advance_mut(self, cnt);
-        }
-    }
-
-    /// Returns the remaining spare capacity of the vector as a slice of `MaybeUninit<u8>`.
-    #[inline]
-    fn spare_capacity_mut(&mut self) -> &mut [MaybeUninit<u8>] {
-        // SAFETY: we guarantees that the `Self::capacity()` bytes from
-        // `Self::as_mut_ptr()` are allocated.
-        unsafe {
-            let ptr = self.as_mut_ptr().add(self.len());
-            let len = self.capacity() - self.len();
-
-            core::slice::from_raw_parts_mut(ptr.cast(), len)
-        }
-    }
 }

 impl<A: Alignment> Deref for AlignedBufferMut<A> {
--- a/pageserver/src/virtual_file/owned_buffers_io/io_buf_aligned.rs
+++ b/pageserver/src/virtual_file/owned_buffers_io/io_buf_aligned.rs
@@ -1,15 +1,9 @@
-use tokio_epoll_uring::{IoBuf, IoBufMut};
+use tokio_epoll_uring::IoBufMut;

-use crate::virtual_file::{IoBuffer, IoBufferMut, PageWriteGuardBuf};
+use crate::virtual_file::{IoBufferMut, PageWriteGuardBuf};

-/// A marker trait for a mutable aligned buffer type.
 pub trait IoBufAlignedMut: IoBufMut {}

-/// A marker trait for an aligned buffer type.
-pub trait IoBufAligned: IoBuf {}
-
 impl IoBufAlignedMut for IoBufferMut {}

-impl IoBufAligned for IoBuffer {}
-
 impl IoBufAlignedMut for PageWriteGuardBuf {}
--- a/pageserver/src/virtual_file/owned_buffers_io/io_buf_ext.rs
+++ b/pageserver/src/virtual_file/owned_buffers_io/io_buf_ext.rs
@@ -5,8 +5,6 @@ use bytes::{Bytes, BytesMut};
 use std::ops::{Deref, Range};
 use tokio_epoll_uring::{BoundedBuf, IoBuf, Slice};

-use super::write::CheapCloneForRead;
-
 /// The true owned equivalent for Rust [`slice`]. Use this for the write path.
 ///
 /// Unlike [`tokio_epoll_uring::Slice`], which we unfortunately inherited from `tokio-uring`,
@@ -45,18 +43,6 @@ where
    }
 }

-impl<B> CheapCloneForRead for FullSlice<B>
-where
-    B: IoBuf + CheapCloneForRead,
-{
-    fn cheap_clone(&self) -> Self {
-        let bounds = self.slice.bounds();
-        let clone = self.slice.get_ref().cheap_clone();
-        let slice = clone.slice(bounds);
-        Self { slice }
-    }
-}
-
 pub(crate) trait IoBufExt {
    /// Get a [`FullSlice`] for the entire buffer, i.e., `self[..]` or `self[0..self.len()]`.
    fn slice_len(self) -> FullSlice<Self>
--- a/pageserver/src/virtual_file/owned_buffers_io/util/size_tracking_writer.rs
+++ b/pageserver/src/virtual_file/owned_buffers_io/util/size_tracking_writer.rs
@@ -0,0 +1,50 @@
+use crate::{
+    context::RequestContext,
+    virtual_file::owned_buffers_io::{io_buf_ext::FullSlice, write::OwnedAsyncWriter},
+};
+use tokio_epoll_uring::IoBuf;
+
+pub struct Writer<W> {
+    dst: W,
+    bytes_amount: u64,
+}
+
+impl<W> Writer<W> {
+    pub fn new(dst: W) -> Self {
+        Self {
+            dst,
+            bytes_amount: 0,
+        }
+    }
+
+    pub fn bytes_written(&self) -> u64 {
+        self.bytes_amount
+    }
+
+    pub fn as_inner(&self) -> &W {
+        &self.dst
+    }
+
+    /// Returns the wrapped `VirtualFile` object as well as the number
+    /// of bytes that were written to it through this object.
+    #[cfg_attr(target_os = "macos", allow(dead_code))]
+    pub fn into_inner(self) -> (u64, W) {
+        (self.bytes_amount, self.dst)
+    }
+}
+
+impl<W> OwnedAsyncWriter for Writer<W>
+where
+    W: OwnedAsyncWriter,
+{
+    #[inline(always)]
+    async fn write_all<Buf: IoBuf + Send>(
+        &mut self,
+        buf: FullSlice<Buf>,
+        ctx: &RequestContext,
+    ) -> std::io::Result<(usize, FullSlice<Buf>)> {
+        let (nwritten, buf) = self.dst.write_all(buf, ctx).await?;
+        self.bytes_amount += u64::try_from(nwritten).unwrap();
+        Ok((nwritten, buf))
+    }
+}
--- a/pageserver/src/virtual_file/owned_buffers_io/write.rs
+++ b/pageserver/src/virtual_file/owned_buffers_io/write.rs
@@ -1,88 +1,55 @@
-mod flush;
-use std::sync::Arc;
-
-use flush::FlushHandle;
+use bytes::BytesMut;
 use tokio_epoll_uring::IoBuf;

-use crate::{
-    context::RequestContext,
-    virtual_file::{IoBuffer, IoBufferMut},
-};
+use crate::context::RequestContext;

-use super::{
-    io_buf_aligned::IoBufAligned,
-    io_buf_ext::{FullSlice, IoBufExt},
-};
-
-pub(crate) use flush::FlushControl;
-
-pub(crate) trait CheapCloneForRead {
-    /// Returns a cheap clone of the buffer.
-    fn cheap_clone(&self) -> Self;
-}
-
-impl CheapCloneForRead for IoBuffer {
-    fn cheap_clone(&self) -> Self {
-        // Cheap clone over an `Arc`.
-        self.clone()
-    }
-}
+use super::io_buf_ext::{FullSlice, IoBufExt};

 /// A trait for doing owned-buffer write IO.
 /// Think [`tokio::io::AsyncWrite`] but with owned buffers.
-/// The owned buffers need to be aligned due to Direct IO requirements.
 pub trait OwnedAsyncWriter {
-    fn write_all_at<Buf: IoBufAligned + Send>(
-        &self,
+    async fn write_all<Buf: IoBuf + Send>(
+        &mut self,
        buf: FullSlice<Buf>,
-        offset: u64,
        ctx: &RequestContext,
-    ) -> impl std::future::Future<Output = std::io::Result<FullSlice<Buf>>> + Send;
+    ) -> std::io::Result<(usize, FullSlice<Buf>)>;
 }

 /// A wrapper aorund an [`OwnedAsyncWriter`] that uses a [`Buffer`] to batch
 /// small writes into larger writes of size [`Buffer::cap`].
-// TODO(yuchen): For large write, implementing buffer bypass for aligned parts of the write could be beneficial to throughput,
-// since we would avoid copying majority of the data into the internal buffer.
-pub struct BufferedWriter<B: Buffer, W> {
-    writer: Arc<W>,
+///
+/// # Passthrough Of Large Writers
+///
+/// Calls to [`BufferedWriter::write_buffered`] that are larger than [`Buffer::cap`]
+/// cause the internal buffer to be flushed prematurely so that the large
+/// buffered write is passed through to the underlying [`OwnedAsyncWriter`].
+///
+/// This pass-through is generally beneficial for throughput, but if
+/// the storage backend of the [`OwnedAsyncWriter`] is a shared resource,
+/// unlimited large writes may cause latency or fairness issues.
+///
+/// In such cases, a different implementation that always buffers in memory
+/// may be preferable.
+pub struct BufferedWriter<B, W> {
+    writer: W,
    /// invariant: always remains Some(buf) except
    /// - while IO is ongoing => goes back to Some() once the IO completed successfully
    /// - after an IO error => stays `None` forever
    ///
    /// In these exceptional cases, it's `None`.
-    mutable: Option<B>,
-    /// A handle to the background flush task for writting data to disk.
-    flush_handle: FlushHandle<B::IoBuf, W>,
-    /// The number of bytes submitted to the background task.
-    bytes_submitted: u64,
+    buf: Option<B>,
 }

 impl<B, Buf, W> BufferedWriter<B, W>
 where
-    B: Buffer<IoBuf = Buf> + Send + 'static,
-    Buf: IoBufAligned + Send + Sync + CheapCloneForRead,
-    W: OwnedAsyncWriter + Send + Sync + 'static + std::fmt::Debug,
+    B: Buffer<IoBuf = Buf> + Send,
+    Buf: IoBuf + Send,
+    W: OwnedAsyncWriter,
 {
-    /// Creates a new buffered writer.
-    ///
-    /// The `buf_new` function provides a way to initialize the owned buffers used by this writer.
-    pub fn new(
-        writer: Arc<W>,
-        buf_new: impl Fn() -> B,
-        gate_guard: utils::sync::gate::GateGuard,
-        ctx: &RequestContext,
-    ) -> Self {
+    pub fn new(writer: W, buf: B) -> Self {
        Self {
-            writer: writer.clone(),
-            mutable: Some(buf_new()),
-            flush_handle: FlushHandle::spawn_new(
-                writer,
-                buf_new(),
-                gate_guard,
-                ctx.attached_child(),
-            ),
-            bytes_submitted: 0,
+            writer,
+            buf: Some(buf),
        }
    }

@@ -90,71 +57,87 @@ where
        &self.writer
    }

-    /// Returns the number of bytes submitted to the background flush task.
-    pub fn bytes_submitted(&self) -> u64 {
-        self.bytes_submitted
-    }
-
    /// Panics if used after any of the write paths returned an error
-    pub fn inspect_mutable(&self) -> &B {
-        self.mutable()
-    }
-
-    /// Gets a reference to the maybe flushed read-only buffer.
-    /// Returns `None` if the writer has not submitted any flush request.
-    pub fn inspect_maybe_flushed(&self) -> Option<&FullSlice<Buf>> {
-        self.flush_handle.maybe_flushed.as_ref()
+    pub fn inspect_buffer(&self) -> &B {
+        self.buf()
    }

    #[cfg_attr(target_os = "macos", allow(dead_code))]
-    pub async fn flush_and_into_inner(
-        mut self,
-        ctx: &RequestContext,
-    ) -> std::io::Result<(u64, Arc<W>)> {
+    pub async fn flush_and_into_inner(mut self, ctx: &RequestContext) -> std::io::Result<W> {
        self.flush(ctx).await?;

-        let Self {
-            mutable: buf,
-            writer,
-            mut flush_handle,
-            bytes_submitted: bytes_amount,
-        } = self;
-        flush_handle.shutdown().await?;
+        let Self { buf, writer } = self;
        assert!(buf.is_some());
-        Ok((bytes_amount, writer))
+        Ok(writer)
    }

-    /// Gets a reference to the mutable in-memory buffer.
    #[inline(always)]
-    fn mutable(&self) -> &B {
-        self.mutable
+    fn buf(&self) -> &B {
+        self.buf
            .as_ref()
            .expect("must not use after we returned an error")
    }

+    /// Guarantees that if Ok() is returned, all bytes in `chunk` have been accepted.
    #[cfg_attr(target_os = "macos", allow(dead_code))]
-    pub async fn write_buffered_borrowed(
+    pub async fn write_buffered<S: IoBuf + Send>(
        &mut self,
-        chunk: &[u8],
+        chunk: FullSlice<S>,
        ctx: &RequestContext,
-    ) -> std::io::Result<usize> {
-        let (len, control) = self.write_buffered_borrowed_controlled(chunk, ctx).await?;
-        if let Some(control) = control {
-            control.release().await;
+    ) -> std::io::Result<(usize, FullSlice<S>)> {
+        let chunk = chunk.into_raw_slice();
+
+        let chunk_len = chunk.len();
+        // avoid memcpy for the middle of the chunk
+        if chunk.len() >= self.buf().cap() {
+            self.flush(ctx).await?;
+            // do a big write, bypassing `buf`
+            assert_eq!(
+                self.buf
+                    .as_ref()
+                    .expect("must not use after an error")
+                    .pending(),
+                0
+            );
+            let (nwritten, chunk) = self
+                .writer
+                .write_all(FullSlice::must_new(chunk), ctx)
+                .await?;
+            assert_eq!(nwritten, chunk_len);
+            return Ok((nwritten, chunk));
        }
-        Ok(len)
+        // in-memory copy the < BUFFER_SIZED tail of the chunk
+        assert!(chunk.len() < self.buf().cap());
+        let mut slice = &chunk[..];
+        while !slice.is_empty() {
+            let buf = self.buf.as_mut().expect("must not use after an error");
+            let need = buf.cap() - buf.pending();
+            let have = slice.len();
+            let n = std::cmp::min(need, have);
+            buf.extend_from_slice(&slice[..n]);
+            slice = &slice[n..];
+            if buf.pending() >= buf.cap() {
+                assert_eq!(buf.pending(), buf.cap());
+                self.flush(ctx).await?;
+            }
+        }
+        assert!(slice.is_empty(), "by now we should have drained the chunk");
+        Ok((chunk_len, FullSlice::must_new(chunk)))
    }

-    /// In addition to bytes submitted in this write, also returns a handle that can control the flush behavior.
-    pub(crate) async fn write_buffered_borrowed_controlled(
+    /// Strictly less performant variant of [`Self::write_buffered`] that allows writing borrowed data.
+    ///
+    /// It is less performant because we always have to copy the borrowed data into the internal buffer
+    /// before we can do the IO. The [`Self::write_buffered`] can avoid this, which is more performant
+    /// for large writes.
+    pub async fn write_buffered_borrowed(
        &mut self,
        mut chunk: &[u8],
        ctx: &RequestContext,
-    ) -> std::io::Result<(usize, Option<FlushControl>)> {
+    ) -> std::io::Result<usize> {
        let chunk_len = chunk.len();
-        let mut control: Option<FlushControl> = None;
        while !chunk.is_empty() {
-            let buf = self.mutable.as_mut().expect("must not use after an error");
+            let buf = self.buf.as_mut().expect("must not use after an error");
            let need = buf.cap() - buf.pending();
            let have = chunk.len();
            let n = std::cmp::min(need, have);
@@ -162,27 +145,26 @@ where
            chunk = &chunk[n..];
            if buf.pending() >= buf.cap() {
                assert_eq!(buf.pending(), buf.cap());
-                if let Some(control) = control.take() {
-                    control.release().await;
-                }
-                control = self.flush(ctx).await?;
+                self.flush(ctx).await?;
            }
        }
-        Ok((chunk_len, control))
+        Ok(chunk_len)
    }

-    #[must_use = "caller must explcitly check the flush control"]
-    async fn flush(&mut self, _ctx: &RequestContext) -> std::io::Result<Option<FlushControl>> {
-        let buf = self.mutable.take().expect("must not use after an error");
+    async fn flush(&mut self, ctx: &RequestContext) -> std::io::Result<()> {
+        let buf = self.buf.take().expect("must not use after an error");
        let buf_len = buf.pending();
        if buf_len == 0 {
-            self.mutable = Some(buf);
-            return Ok(None);
+            self.buf = Some(buf);
+            return Ok(());
        }
-        let (recycled, flush_control) = self.flush_handle.flush(buf, self.bytes_submitted).await?;
-        self.bytes_submitted += u64::try_from(buf_len).unwrap();
-        self.mutable = Some(recycled);
-        Ok(Some(flush_control))
+        let slice = buf.flush();
+        let (nwritten, slice) = self.writer.write_all(slice, ctx).await?;
+        assert_eq!(nwritten, buf_len);
+        self.buf = Some(Buffer::reuse_after_flush(
+            slice.into_raw_slice().into_inner(),
+        ));
+        Ok(())
    }
 }

@@ -210,77 +192,64 @@ pub trait Buffer {
    fn reuse_after_flush(iobuf: Self::IoBuf) -> Self;
 }

-impl Buffer for IoBufferMut {
-    type IoBuf = IoBuffer;
+impl Buffer for BytesMut {
+    type IoBuf = BytesMut;

+    #[inline(always)]
    fn cap(&self) -> usize {
        self.capacity()
    }

    fn extend_from_slice(&mut self, other: &[u8]) {
-        if self.len() + other.len() > self.cap() {
-            panic!("Buffer capacity exceeded");
-        }
-
-        IoBufferMut::extend_from_slice(self, other);
+        BytesMut::extend_from_slice(self, other)
    }

+    #[inline(always)]
    fn pending(&self) -> usize {
        self.len()
    }

-    fn flush(self) -> FullSlice<Self::IoBuf> {
-        self.freeze().slice_len()
+    fn flush(self) -> FullSlice<BytesMut> {
+        self.slice_len()
    }

-    /// Caller should make sure that `iobuf` only have one strong reference before invoking this method.
-    fn reuse_after_flush(iobuf: Self::IoBuf) -> Self {
-        let mut recycled = iobuf
-            .into_mut()
-            .expect("buffer should only have one strong reference");
-        recycled.clear();
-        recycled
+    fn reuse_after_flush(mut iobuf: BytesMut) -> Self {
+        iobuf.clear();
+        iobuf
+    }
+}
+
+impl OwnedAsyncWriter for Vec<u8> {
+    async fn write_all<Buf: IoBuf + Send>(
+        &mut self,
+        buf: FullSlice<Buf>,
+        _: &RequestContext,
+    ) -> std::io::Result<(usize, FullSlice<Buf>)> {
+        self.extend_from_slice(&buf[..]);
+        Ok((buf.len(), buf))
    }
 }

 #[cfg(test)]
 mod tests {
-    use std::sync::Mutex;
+    use bytes::BytesMut;

    use super::*;
    use crate::context::{DownloadBehavior, RequestContext};
    use crate::task_mgr::TaskKind;

-    #[derive(Default, Debug)]
+    #[derive(Default)]
    struct RecorderWriter {
-        /// record bytes and write offsets.
-        writes: Mutex<Vec<(Vec<u8>, u64)>>,
+        writes: Vec<Vec<u8>>,
    }
-
-    impl RecorderWriter {
-        /// Gets recorded bytes and write offsets.
-        fn get_writes(&self) -> Vec<Vec<u8>> {
-            self.writes
-                .lock()
-                .unwrap()
-                .iter()
-                .map(|(buf, _)| buf.clone())
-                .collect()
-        }
-    }
-
    impl OwnedAsyncWriter for RecorderWriter {
-        async fn write_all_at<Buf: IoBufAligned + Send>(
-            &self,
+        async fn write_all<Buf: IoBuf + Send>(
+            &mut self,
            buf: FullSlice<Buf>,
-            offset: u64,
            _: &RequestContext,
-        ) -> std::io::Result<FullSlice<Buf>> {
-            self.writes
-                .lock()
-                .unwrap()
-                .push((Vec::from(&buf[..]), offset));
-            Ok(buf)
+        ) -> std::io::Result<(usize, FullSlice<Buf>)> {
+            self.writes.push(Vec::from(&buf[..]));
+            Ok((buf.len(), buf))
        }
    }

@@ -288,21 +257,71 @@ mod tests {
        RequestContext::new(TaskKind::UnitTest, DownloadBehavior::Error)
    }

+    macro_rules! write {
+        ($writer:ident, $data:literal) => {{
+            $writer
+                .write_buffered(::bytes::Bytes::from_static($data).slice_len(), &test_ctx())
+                .await?;
+        }};
+    }
+
    #[tokio::test]
-    async fn test_write_all_borrowed_always_goes_through_buffer() -> anyhow::Result<()> {
+    async fn test_buffered_writes_only() -> std::io::Result<()> {
+        let recorder = RecorderWriter::default();
+        let mut writer = BufferedWriter::new(recorder, BytesMut::with_capacity(2));
+        write!(writer, b"a");
+        write!(writer, b"b");
+        write!(writer, b"c");
+        write!(writer, b"d");
+        write!(writer, b"e");
+        let recorder = writer.flush_and_into_inner(&test_ctx()).await?;
+        assert_eq!(
+            recorder.writes,
+            vec![Vec::from(b"ab"), Vec::from(b"cd"), Vec::from(b"e")]
+        );
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn test_passthrough_writes_only() -> std::io::Result<()> {
+        let recorder = RecorderWriter::default();
+        let mut writer = BufferedWriter::new(recorder, BytesMut::with_capacity(2));
+        write!(writer, b"abc");
+        write!(writer, b"de");
+        write!(writer, b"");
+        write!(writer, b"fghijk");
+        let recorder = writer.flush_and_into_inner(&test_ctx()).await?;
+        assert_eq!(
+            recorder.writes,
+            vec![Vec::from(b"abc"), Vec::from(b"de"), Vec::from(b"fghijk")]
+        );
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn test_passthrough_write_with_nonempty_buffer() -> std::io::Result<()> {
+        let recorder = RecorderWriter::default();
+        let mut writer = BufferedWriter::new(recorder, BytesMut::with_capacity(2));
+        write!(writer, b"a");
+        write!(writer, b"bc");
+        write!(writer, b"d");
+        write!(writer, b"e");
+        let recorder = writer.flush_and_into_inner(&test_ctx()).await?;
+        assert_eq!(
+            recorder.writes,
+            vec![Vec::from(b"a"), Vec::from(b"bc"), Vec::from(b"de")]
+        );
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn test_write_all_borrowed_always_goes_through_buffer() -> std::io::Result<()> {
        let ctx = test_ctx();
        let ctx = &ctx;
-        let recorder = Arc::new(RecorderWriter::default());
-        let gate = utils::sync::gate::Gate::default();
-        let mut writer = BufferedWriter::<_, RecorderWriter>::new(
-            recorder,
-            || IoBufferMut::with_capacity(2),
-            gate.enter()?,
-            ctx,
-        );
+        let recorder = RecorderWriter::default();
+        let mut writer = BufferedWriter::new(recorder, BytesMut::with_capacity(2));

        writer.write_buffered_borrowed(b"abc", ctx).await?;
-        writer.write_buffered_borrowed(b"", ctx).await?;
        writer.write_buffered_borrowed(b"d", ctx).await?;
        writer.write_buffered_borrowed(b"e", ctx).await?;
        writer.write_buffered_borrowed(b"fg", ctx).await?;
@@ -310,9 +329,9 @@ mod tests {
        writer.write_buffered_borrowed(b"j", ctx).await?;
        writer.write_buffered_borrowed(b"klmno", ctx).await?;

-        let (_, recorder) = writer.flush_and_into_inner(ctx).await?;
+        let recorder = writer.flush_and_into_inner(ctx).await?;
        assert_eq!(
-            recorder.get_writes(),
+            recorder.writes,
            {
                let expect: &[&[u8]] = &[b"ab", b"cd", b"ef", b"gh", b"ij", b"kl", b"mn", b"o"];
                expect
--- a/pageserver/src/virtual_file/owned_buffers_io/write/flush.rs
+++ b/pageserver/src/virtual_file/owned_buffers_io/write/flush.rs
@@ -1,314 +0,0 @@
-use std::sync::Arc;
-
-use utils::sync::duplex;
-
-use crate::{
-    context::RequestContext,
-    virtual_file::owned_buffers_io::{io_buf_aligned::IoBufAligned, io_buf_ext::FullSlice},
-};
-
-use super::{Buffer, CheapCloneForRead, OwnedAsyncWriter};
-
-/// A handle to the flush task.
-pub struct FlushHandle<Buf, W> {
-    inner: Option<FlushHandleInner<Buf, W>>,
-    /// Immutable buffer for serving tail reads.
-    /// `None` if no flush request has been submitted.
-    pub(super) maybe_flushed: Option<FullSlice<Buf>>,
-}
-
-pub struct FlushHandleInner<Buf, W> {
-    /// A bi-directional channel that sends (buffer, offset) for writes,
-    /// and receives recyled buffer.
-    channel: duplex::mpsc::Duplex<FlushRequest<Buf>, FullSlice<Buf>>,
-    /// Join handle for the background flush task.
-    join_handle: tokio::task::JoinHandle<std::io::Result<Arc<W>>>,
-}
-
-struct FlushRequest<Buf> {
-    slice: FullSlice<Buf>,
-    offset: u64,
-    #[cfg(test)]
-    ready_to_flush_rx: tokio::sync::oneshot::Receiver<()>,
-    #[cfg(test)]
-    done_flush_tx: tokio::sync::oneshot::Sender<()>,
-}
-
-/// Constructs a request and a control object for a new flush operation.
-#[cfg(not(test))]
-fn new_flush_op<Buf>(slice: FullSlice<Buf>, offset: u64) -> (FlushRequest<Buf>, FlushControl) {
-    let request = FlushRequest { slice, offset };
-    let control = FlushControl::untracked();
-
-    (request, control)
-}
-
-/// Constructs a request and a control object for a new flush operation.
-#[cfg(test)]
-fn new_flush_op<Buf>(slice: FullSlice<Buf>, offset: u64) -> (FlushRequest<Buf>, FlushControl) {
-    let (ready_to_flush_tx, ready_to_flush_rx) = tokio::sync::oneshot::channel();
-    let (done_flush_tx, done_flush_rx) = tokio::sync::oneshot::channel();
-    let control = FlushControl::not_started(ready_to_flush_tx, done_flush_rx);
-
-    let request = FlushRequest {
-        slice,
-        offset,
-        ready_to_flush_rx,
-        done_flush_tx,
-    };
-    (request, control)
-}
-
-/// A handle to a `FlushRequest` that allows unit tests precise control over flush behavior.
-#[cfg(test)]
-pub(crate) struct FlushControl {
-    not_started: FlushNotStarted,
-}
-
-#[cfg(not(test))]
-pub(crate) struct FlushControl;
-
-impl FlushControl {
-    #[cfg(test)]
-    fn not_started(
-        ready_to_flush_tx: tokio::sync::oneshot::Sender<()>,
-        done_flush_rx: tokio::sync::oneshot::Receiver<()>,
-    ) -> Self {
-        FlushControl {
-            not_started: FlushNotStarted {
-                ready_to_flush_tx,
-                done_flush_rx,
-            },
-        }
-    }
-
-    #[cfg(not(test))]
-    fn untracked() -> Self {
-        FlushControl
-    }
-
-    /// In tests, turn flush control into a not started state.
-    #[cfg(test)]
-    pub(crate) fn into_not_started(self) -> FlushNotStarted {
-        self.not_started
-    }
-
-    /// Release control to the submitted buffer.
-    ///
-    /// In `cfg(test)` environment, the buffer is guranteed to be flushed to disk after [`FlushControl::release`] is finishes execution.
-    pub async fn release(self) {
-        #[cfg(test)]
-        {
-            self.not_started
-                .ready_to_flush()
-                .wait_until_flush_is_done()
-                .await;
-        }
-    }
-}
-
-impl<Buf, W> FlushHandle<Buf, W>
-where
-    Buf: IoBufAligned + Send + Sync + CheapCloneForRead,
-    W: OwnedAsyncWriter + Send + Sync + 'static + std::fmt::Debug,
-{
-    /// Spawns a new background flush task and obtains a handle.
-    ///
-    /// Note: The background task so we do not need to explicitly maintain a queue of buffers.
-    pub fn spawn_new<B>(
-        file: Arc<W>,
-        buf: B,
-        gate_guard: utils::sync::gate::GateGuard,
-        ctx: RequestContext,
-    ) -> Self
-    where
-        B: Buffer<IoBuf = Buf> + Send + 'static,
-    {
-        // It is fine to buffer up to only 1 message. We only 1 message in-flight at a time.
-        let (front, back) = duplex::mpsc::channel(1);
-
-        let join_handle = tokio::spawn(async move {
-            FlushBackgroundTask::new(back, file, gate_guard, ctx)
-                .run(buf.flush())
-                .await
-        });
-
-        FlushHandle {
-            inner: Some(FlushHandleInner {
-                channel: front,
-                join_handle,
-            }),
-            maybe_flushed: None,
-        }
-    }
-
-    /// Submits a buffer to be flushed in the background task.
-    /// Returns a buffer that completed flushing for re-use, length reset to 0, capacity unchanged.
-    /// If `save_buf_for_read` is true, then we save the buffer in `Self::maybe_flushed`, otherwise
-    /// clear `maybe_flushed`.
-    pub async fn flush<B>(&mut self, buf: B, offset: u64) -> std::io::Result<(B, FlushControl)>
-    where
-        B: Buffer<IoBuf = Buf> + Send + 'static,
-    {
-        let slice = buf.flush();
-
-        // Saves a buffer for read while flushing. This also removes reference to the old buffer.
-        self.maybe_flushed = Some(slice.cheap_clone());
-
-        let (request, flush_control) = new_flush_op(slice, offset);
-
-        // Submits the buffer to the background task.
-        let submit = self.inner_mut().channel.send(request).await;
-        if submit.is_err() {
-            return self.handle_error().await;
-        }
-
-        // Wait for an available buffer from the background flush task.
-        // This is the BACKPRESSURE mechanism: if the flush task can't keep up,
-        // then the write path will eventually wait for it here.
-        let Some(recycled) = self.inner_mut().channel.recv().await else {
-            return self.handle_error().await;
-        };
-
-        // The only other place that could hold a reference to the recycled buffer
-        // is in `Self::maybe_flushed`, but we have already replace it with the new buffer.
-        let recycled = Buffer::reuse_after_flush(recycled.into_raw_slice().into_inner());
-        Ok((recycled, flush_control))
-    }
-
-    async fn handle_error<T>(&mut self) -> std::io::Result<T> {
-        Err(self
-            .shutdown()
-            .await
-            .expect_err("flush task only disconnects duplex if it exits with an error"))
-    }
-
-    /// Cleans up the channel, join the flush task.
-    pub async fn shutdown(&mut self) -> std::io::Result<Arc<W>> {
-        let handle = self
-            .inner
-            .take()
-            .expect("must not use after we returned an error");
-        drop(handle.channel.tx);
-        handle.join_handle.await.unwrap()
-    }
-
-    /// Gets a mutable reference to the inner handle. Panics if [`Self::inner`] is `None`.
-    /// This only happens if the handle is used after an error.
-    fn inner_mut(&mut self) -> &mut FlushHandleInner<Buf, W> {
-        self.inner
-            .as_mut()
-            .expect("must not use after we returned an error")
-    }
-}
-
-/// A background task for flushing data to disk.
-pub struct FlushBackgroundTask<Buf, W> {
-    /// A bi-directional channel that receives (buffer, offset) for writes,
-    /// and send back recycled buffer.
-    channel: duplex::mpsc::Duplex<FullSlice<Buf>, FlushRequest<Buf>>,
-    /// A writter for persisting data to disk.
-    writer: Arc<W>,
-    ctx: RequestContext,
-    /// Prevent timeline from shuting down until the flush background task finishes flushing all remaining buffers to disk.
-    _gate_guard: utils::sync::gate::GateGuard,
-}
-
-impl<Buf, W> FlushBackgroundTask<Buf, W>
-where
-    Buf: IoBufAligned + Send + Sync,
-    W: OwnedAsyncWriter + Sync + 'static,
-{
-    /// Creates a new background flush task.
-    fn new(
-        channel: duplex::mpsc::Duplex<FullSlice<Buf>, FlushRequest<Buf>>,
-        file: Arc<W>,
-        gate_guard: utils::sync::gate::GateGuard,
-        ctx: RequestContext,
-    ) -> Self {
-        FlushBackgroundTask {
-            channel,
-            writer: file,
-            _gate_guard: gate_guard,
-            ctx,
-        }
-    }
-
-    /// Runs the background flush task.
-    /// The passed in slice is immediately sent back to the flush handle through the duplex channel.
-    async fn run(mut self, slice: FullSlice<Buf>) -> std::io::Result<Arc<W>> {
-        // Sends the extra buffer back to the handle.
-        self.channel.send(slice).await.map_err(|_| {
-            std::io::Error::new(std::io::ErrorKind::BrokenPipe, "flush handle closed early")
-        })?;
-
-        //  Exit condition: channel is closed and there is no remaining buffer to be flushed
-        while let Some(request) = self.channel.recv().await {
-            #[cfg(test)]
-            {
-                // In test, wait for control to signal that we are ready to flush.
-                if request.ready_to_flush_rx.await.is_err() {
-                    tracing::debug!("control dropped");
-                }
-            }
-
-            // Write slice to disk at `offset`.
-            let slice = self
-                .writer
-                .write_all_at(request.slice, request.offset, &self.ctx)
-                .await?;
-
-            #[cfg(test)]
-            {
-                // In test, tell control we are done flushing buffer.
-                if request.done_flush_tx.send(()).is_err() {
-                    tracing::debug!("control dropped");
-                }
-            }
-
-            // Sends the buffer back to the handle for reuse. The handle is in charged of cleaning the buffer.
-            if self.channel.send(slice).await.is_err() {
-                // Although channel is closed. Still need to finish flushing the remaining buffers.
-                continue;
-            }
-        }
-
-        Ok(self.writer)
-    }
-}
-
-#[cfg(test)]
-pub(crate) struct FlushNotStarted {
-    ready_to_flush_tx: tokio::sync::oneshot::Sender<()>,
-    done_flush_rx: tokio::sync::oneshot::Receiver<()>,
-}
-
-#[cfg(test)]
-pub(crate) struct FlushInProgress {
-    done_flush_rx: tokio::sync::oneshot::Receiver<()>,
-}
-
-#[cfg(test)]
-pub(crate) struct FlushDone;
-
-#[cfg(test)]
-impl FlushNotStarted {
-    /// Signals the background task the buffer is ready to flush to disk.
-    pub fn ready_to_flush(self) -> FlushInProgress {
-        self.ready_to_flush_tx
-            .send(())
-            .map(|_| FlushInProgress {
-                done_flush_rx: self.done_flush_rx,
-            })
-            .unwrap()
-    }
-}
-
-#[cfg(test)]
-impl FlushInProgress {
-    /// Waits until background flush is done.
-    pub async fn wait_until_flush_is_done(self) -> FlushDone {
-        self.done_flush_rx.await.unwrap();
-        FlushDone
-    }
-}
--- a/pageserver/src/walingest.rs
+++ b/pageserver/src/walingest.rs
@@ -1187,6 +1187,50 @@ impl WalIngest {
                } else {
                    cp.oldestActiveXid = xlog_checkpoint.oldestActiveXid;
                }
+                // NB: We abuse the Checkpoint.redo field:
+                //
+                // - In PostgreSQL, the Checkpoint struct doesn't store the information
+                //   of whether this is an online checkpoint or a shutdown checkpoint. It's
+                //   stored in the XLOG info field of the WAL record, shutdown checkpoints
+                //   use record type XLOG_CHECKPOINT_SHUTDOWN and online checkpoints use
+                //   XLOG_CHECKPOINT_ONLINE. We don't store the original WAL record headers
+                //   in the pageserver, however.
+                //
+                // - In PostgreSQL, the Checkpoint.redo field stores the *start* of the
+                //   checkpoint record, if it's a shutdown checkpoint. But when we are
+                //   starting from a shutdown checkpoint, the basebackup LSN is the *end*
+                //   of the shutdown checkpoint WAL record. That makes it difficult to
+                //   correctly detect whether we're starting from a shutdown record or
+                //   not.
+                //
+                // To address both of those issues, we store 0 in the redo field if it's
+                // an online checkpoint record, and the record's *end* LSN if it's a
+                // shutdown checkpoint. We don't need the original redo pointer in neon,
+                // because we don't perform WAL replay at startup anyway, so we can get
+                // away with abusing the redo field like this.
+                //
+                // XXX: Ideally, we would persist the extra information in a more
+                // explicit format, rather than repurpose the fields of the Postgres
+                // struct like this. However, we already have persisted data like this,
+                // so we need to maintain backwards compatibility.
+                //
+                // NB: We didn't originally have this convention, so there are still old
+                // persisted records that didn't do this. Before, we didn't update the
+                // persisted redo field at all. That means that old records have a bogus
+                // redo pointer that points to some old value, from the checkpoint record
+                // that was originally imported from the data directory. If it was a
+                // project created in Neon, that means it points to the first checkpoint
+                // after initdb. That's OK for our purposes: all such old checkpoints are
+                // treated as old online checkpoints when the basebackup is created.
+                cp.redo = if info == pg_constants::XLOG_CHECKPOINT_SHUTDOWN {
+                    // Store the *end* LSN of the checkpoint record. Or to be precise,
+                    // the start LSN of the *next* record, i.e. if the record ends
+                    // exactly at page boundary, the redo LSN points to just after the
+                    // page header on the next page.
+                    lsn.into()
+                } else {
+                    Lsn::INVALID.into()
+                };

                // Write a new checkpoint key-value pair on every checkpoint record, even
                // if nothing really changed. Not strictly required, but it seems nice to
--- a/proxy/src/auth/backend/mod.rs
+++ b/proxy/src/auth/backend/mod.rs
@@ -70,10 +70,6 @@ impl std::fmt::Display for Backend<'_, ()> {
    fn fmt(&self, fmt: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::ControlPlane(api, ()) => match &**api {
-                ControlPlaneClient::ProxyV1(endpoint) => fmt
-                    .debug_tuple("ControlPlane::ProxyV1")
-                    .field(&endpoint.url())
-                    .finish(),
                ControlPlaneClient::Neon(endpoint) => fmt
                    .debug_tuple("ControlPlane::Neon")
                    .field(&endpoint.url())
--- a/proxy/src/bin/proxy.rs
+++ b/proxy/src/bin/proxy.rs
@@ -46,9 +46,6 @@ enum AuthBackendType {
    #[value(name("console"), alias("cplane"))]
    ControlPlane,

-    #[value(name("cplane-v1"), alias("control-plane"))]
-    ControlPlaneV1,
-
    #[value(name("link"), alias("control-redirect"))]
    ConsoleRedirect,

@@ -521,39 +518,6 @@ async fn main() -> anyhow::Result<()> {
                        .instrument(span),
                );
            }
-        } else if let proxy::control_plane::client::ControlPlaneClient::ProxyV1(api) = &**api {
-            match (redis_notifications_client, regional_redis_client.clone()) {
-                (None, None) => {}
-                (client1, client2) => {
-                    let cache = api.caches.project_info.clone();
-                    if let Some(client) = client1 {
-                        maintenance_tasks.spawn(notifications::task_main(
-                            client,
-                            cache.clone(),
-                            cancel_map.clone(),
-                            args.region.clone(),
-                        ));
-                    }
-                    if let Some(client) = client2 {
-                        maintenance_tasks.spawn(notifications::task_main(
-                            client,
-                            cache.clone(),
-                            cancel_map.clone(),
-                            args.region.clone(),
-                        ));
-                    }
-                    maintenance_tasks.spawn(async move { cache.clone().gc_worker().await });
-                }
-            }
-            if let Some(regional_redis_client) = regional_redis_client {
-                let cache = api.caches.endpoints_cache.clone();
-                let con = regional_redis_client;
-                let span = tracing::info_span!("endpoints_cache");
-                maintenance_tasks.spawn(
-                    async move { cache.do_read(con, cancellation_token.clone()).await }
-                        .instrument(span),
-                );
-            }
        }
    }

@@ -698,65 +662,6 @@ fn build_auth_backend(
    args: &ProxyCliArgs,
 ) -> anyhow::Result<Either<&'static auth::Backend<'static, ()>, &'static ConsoleRedirectBackend>> {
    match &args.auth_backend {
-        AuthBackendType::ControlPlaneV1 => {
-            let wake_compute_cache_config: CacheOptions = args.wake_compute_cache.parse()?;
-            let project_info_cache_config: ProjectInfoCacheOptions =
-                args.project_info_cache.parse()?;
-            let endpoint_cache_config: config::EndpointCacheConfig =
-                args.endpoint_cache_config.parse()?;
-
-            info!("Using NodeInfoCache (wake_compute) with options={wake_compute_cache_config:?}");
-            info!(
-                "Using AllowedIpsCache (wake_compute) with options={project_info_cache_config:?}"
-            );
-            info!("Using EndpointCacheConfig with options={endpoint_cache_config:?}");
-            let caches = Box::leak(Box::new(control_plane::caches::ApiCaches::new(
-                wake_compute_cache_config,
-                project_info_cache_config,
-                endpoint_cache_config,
-            )));
-
-            let config::ConcurrencyLockOptions {
-                shards,
-                limiter,
-                epoch,
-                timeout,
-            } = args.wake_compute_lock.parse()?;
-            info!(?limiter, shards, ?epoch, "Using NodeLocks (wake_compute)");
-            let locks = Box::leak(Box::new(control_plane::locks::ApiLocks::new(
-                "wake_compute_lock",
-                limiter,
-                shards,
-                timeout,
-                epoch,
-                &Metrics::get().wake_compute_lock,
-            )?));
-            tokio::spawn(locks.garbage_collect_worker());
-
-            let url: proxy::url::ApiUrl = args.auth_endpoint.parse()?;
-
-            let endpoint = http::Endpoint::new(url, http::new_client());
-
-            let mut wake_compute_rps_limit = args.wake_compute_limit.clone();
-            RateBucketInfo::validate(&mut wake_compute_rps_limit)?;
-            let wake_compute_endpoint_rate_limiter =
-                Arc::new(WakeComputeRateLimiter::new(wake_compute_rps_limit));
-
-            let api = control_plane::client::cplane_proxy_v1::NeonControlPlaneClient::new(
-                endpoint,
-                args.control_plane_token.clone(),
-                caches,
-                locks,
-                wake_compute_endpoint_rate_limiter,
-            );
-
-            let api = control_plane::client::ControlPlaneClient::ProxyV1(api);
-            let auth_backend = auth::Backend::ControlPlane(MaybeOwned::Owned(api), ());
-            let config = Box::leak(Box::new(auth_backend));
-
-            Ok(Either::Left(config))
-        }
-
        AuthBackendType::ControlPlane => {
            let wake_compute_cache_config: CacheOptions = args.wake_compute_cache.parse()?;
            let project_info_cache_config: ProjectInfoCacheOptions =
@@ -792,15 +697,13 @@ fn build_auth_backend(
            )?));
            tokio::spawn(locks.garbage_collect_worker());

-            let url: proxy::url::ApiUrl = args.auth_endpoint.parse()?;
-
+            let url = args.auth_endpoint.parse()?;
            let endpoint = http::Endpoint::new(url, http::new_client());

            let mut wake_compute_rps_limit = args.wake_compute_limit.clone();
            RateBucketInfo::validate(&mut wake_compute_rps_limit)?;
            let wake_compute_endpoint_rate_limiter =
                Arc::new(WakeComputeRateLimiter::new(wake_compute_rps_limit));
-
            let api = control_plane::client::neon::NeonControlPlaneClient::new(
                endpoint,
                args.control_plane_token.clone(),
--- a/proxy/src/cancellation.rs
+++ b/proxy/src/cancellation.rs
@@ -70,12 +70,11 @@ impl ReportableError for CancelError {
 impl<P: CancellationPublisher> CancellationHandler<P> {
    /// Run async action within an ephemeral session identified by [`CancelKeyData`].
    pub(crate) fn get_session(self: Arc<Self>) -> Session<P> {
-        // we intentionally generate a random "backend pid" and "secret key" here.
-        // we use the corresponding u64 as an identifier for the
-        // actual endpoint+pid+secret for postgres/pgbouncer.
-        //
-        // if we forwarded the backend_pid from postgres to the client, there would be a lot
-        // of overlap between our computes as most pids are small (~100).
+        // HACK: We'd rather get the real backend_pid but postgres_client doesn't
+        // expose it and we don't want to do another roundtrip to query
+        // for it. The client will be able to notice that this is not the
+        // actual backend_pid, but backend_pid is not used for anything
+        // so it doesn't matter.
        let key = loop {
            let key = rand::random();

--- a/proxy/src/compute.rs
+++ b/proxy/src/compute.rs
@@ -131,37 +131,49 @@ impl ConnCfg {
    }

    /// Apply startup message params to the connection config.
-    pub(crate) fn set_startup_params(
-        &mut self,
-        params: &StartupMessageParams,
-        arbitrary_params: bool,
-    ) {
-        if !arbitrary_params {
-            self.set_param("client_encoding", "UTF8");
+    pub(crate) fn set_startup_params(&mut self, params: &StartupMessageParams) {
+        // Only set `user` if it's not present in the config.
+        // Console redirect auth flow takes username from the console's response.
+        if let (None, Some(user)) = (self.get_user(), params.get("user")) {
+            self.user(user);
        }
-        for (k, v) in params.iter() {
-            match k {
-                // Only set `user` if it's not present in the config.
-                // Console redirect auth flow takes username from the console's response.
-                "user" if self.user_is_set() => continue,
-                "database" if self.db_is_set() => continue,
-                "options" => {
-                    if let Some(options) = filtered_options(v) {
-                        self.set_param(k, &options);
-                    }
-                }
-                "user" | "database" | "application_name" | "replication" => {
-                    self.set_param(k, v);
-                }

-                // if we allow arbitrary params, then we forward them through.
-                // this is a flag for a period of backwards compatibility
-                k if arbitrary_params => {
-                    self.set_param(k, v);
+        // Only set `dbname` if it's not present in the config.
+        // Console redirect auth flow takes dbname from the console's response.
+        if let (None, Some(dbname)) = (self.get_dbname(), params.get("database")) {
+            self.dbname(dbname);
+        }
+
+        // Don't add `options` if they were only used for specifying a project.
+        // Connection pools don't support `options`, because they affect backend startup.
+        if let Some(options) = filtered_options(params) {
+            self.options(&options);
+        }
+
+        if let Some(app_name) = params.get("application_name") {
+            self.application_name(app_name);
+        }
+
+        // TODO: This is especially ugly...
+        if let Some(replication) = params.get("replication") {
+            use postgres_client::config::ReplicationMode;
+            match replication {
+                "true" | "on" | "yes" | "1" => {
+                    self.replication_mode(ReplicationMode::Physical);
                }
-                _ => {}
+                "database" => {
+                    self.replication_mode(ReplicationMode::Logical);
+                }
+                _other => {}
            }
        }
+
+        // TODO: extend the list of the forwarded startup parameters.
+        // Currently, tokio-postgres doesn't allow us to pass
+        // arbitrary parameters, but the ones above are a good start.
+        //
+        // This and the reverse params problem can be better addressed
+        // in a bespoke connection machinery (a new library for that sake).
    }
 }

@@ -335,9 +347,10 @@ impl ConnCfg {
 }

 /// Retrieve `options` from a startup message, dropping all proxy-secific flags.
-fn filtered_options(options: &str) -> Option<String> {
+fn filtered_options(params: &StartupMessageParams) -> Option<String> {
    #[allow(unstable_name_collisions)]
-    let options: String = StartupMessageParams::parse_options_raw(options)
+    let options: String = params
+        .options_raw()?
        .filter(|opt| parse_endpoint_param(opt).is_none() && neon_option(opt).is_none())
        .intersperse(" ") // TODO: use impl from std once it's stabilized
        .collect();
@@ -414,24 +427,27 @@ mod tests {
    #[test]
    fn test_filtered_options() {
        // Empty options is unlikely to be useful anyway.
-        let params = "";
-        assert_eq!(filtered_options(params), None);
+        let params = StartupMessageParams::new([("options", "")]);
+        assert_eq!(filtered_options(&params), None);

        // It's likely that clients will only use options to specify endpoint/project.
-        let params = "project=foo";
-        assert_eq!(filtered_options(params), None);
+        let params = StartupMessageParams::new([("options", "project=foo")]);
+        assert_eq!(filtered_options(&params), None);

        // Same, because unescaped whitespaces are no-op.
-        let params = " project=foo ";
-        assert_eq!(filtered_options(params).as_deref(), None);
+        let params = StartupMessageParams::new([("options", " project=foo ")]);
+        assert_eq!(filtered_options(&params).as_deref(), None);

-        let params = r"\  project=foo \ ";
-        assert_eq!(filtered_options(params).as_deref(), Some(r"\  \ "));
+        let params = StartupMessageParams::new([("options", r"\  project=foo \ ")]);
+        assert_eq!(filtered_options(&params).as_deref(), Some(r"\  \ "));

-        let params = "project = foo";
-        assert_eq!(filtered_options(params).as_deref(), Some("project = foo"));
+        let params = StartupMessageParams::new([("options", "project = foo")]);
+        assert_eq!(filtered_options(&params).as_deref(), Some("project = foo"));

-        let params = "project = foo neon_endpoint_type:read_write   neon_lsn:0/2 neon_proxy_params_compat:true";
-        assert_eq!(filtered_options(params).as_deref(), Some("project = foo"));
+        let params = StartupMessageParams::new([(
+            "options",
+            "project = foo neon_endpoint_type:read_write   neon_lsn:0/2",
+        )]);
+        assert_eq!(filtered_options(&params).as_deref(), Some("project = foo"));
    }
 }
--- a/proxy/src/console_redirect_proxy.rs
+++ b/proxy/src/console_redirect_proxy.rs
@@ -206,7 +206,6 @@ pub(crate) async fn handle_client<S: AsyncRead + AsyncWrite + Unpin>(
    let mut node = connect_to_compute(
        ctx,
        &TcpMechanism {
-            params_compat: true,
            params: &params,
            locks: &config.connect_compute_locks,
        },
--- a/proxy/src/control_plane/client/cplane_proxy_v1.rs
+++ b/proxy/src/control_plane/client/cplane_proxy_v1.rs
@@ -1,514 +0,0 @@
-//! Production console backend.
-
-use std::sync::Arc;
-use std::time::Duration;
-
-use ::http::header::AUTHORIZATION;
-use ::http::HeaderName;
-use futures::TryFutureExt;
-use postgres_client::config::SslMode;
-use tokio::time::Instant;
-use tracing::{debug, info, info_span, warn, Instrument};
-
-use super::super::messages::{ControlPlaneErrorMessage, GetEndpointAccessControl, WakeCompute};
-use crate::auth::backend::jwt::AuthRule;
-use crate::auth::backend::ComputeUserInfo;
-use crate::cache::Cached;
-use crate::context::RequestContext;
-use crate::control_plane::caches::ApiCaches;
-use crate::control_plane::errors::{
-    ControlPlaneError, GetAuthInfoError, GetEndpointJwksError, WakeComputeError,
-};
-use crate::control_plane::locks::ApiLocks;
-use crate::control_plane::messages::{ColdStartInfo, EndpointJwksResponse, Reason};
-use crate::control_plane::{
-    AuthInfo, AuthSecret, CachedAllowedIps, CachedNodeInfo, CachedRoleSecret, NodeInfo,
-};
-use crate::metrics::{CacheOutcome, Metrics};
-use crate::rate_limiter::WakeComputeRateLimiter;
-use crate::types::{EndpointCacheKey, EndpointId};
-use crate::{compute, http, scram};
-
-const X_REQUEST_ID: HeaderName = HeaderName::from_static("x-request-id");
-
-#[derive(Clone)]
-pub struct NeonControlPlaneClient {
-    endpoint: http::Endpoint,
-    pub caches: &'static ApiCaches,
-    pub(crate) locks: &'static ApiLocks<EndpointCacheKey>,
-    pub(crate) wake_compute_endpoint_rate_limiter: Arc<WakeComputeRateLimiter>,
-    // put in a shared ref so we don't copy secrets all over in memory
-    jwt: Arc<str>,
-}
-
-impl NeonControlPlaneClient {
-    /// Construct an API object containing the auth parameters.
-    pub fn new(
-        endpoint: http::Endpoint,
-        jwt: Arc<str>,
-        caches: &'static ApiCaches,
-        locks: &'static ApiLocks<EndpointCacheKey>,
-        wake_compute_endpoint_rate_limiter: Arc<WakeComputeRateLimiter>,
-    ) -> Self {
-        Self {
-            endpoint,
-            caches,
-            locks,
-            wake_compute_endpoint_rate_limiter,
-            jwt,
-        }
-    }
-
-    pub(crate) fn url(&self) -> &str {
-        self.endpoint.url().as_str()
-    }
-
-    async fn do_get_auth_info(
-        &self,
-        ctx: &RequestContext,
-        user_info: &ComputeUserInfo,
-    ) -> Result<AuthInfo, GetAuthInfoError> {
-        if !self
-            .caches
-            .endpoints_cache
-            .is_valid(ctx, &user_info.endpoint.normalize())
-        {
-            // TODO: refactor this because it's weird
-            // this is a failure to authenticate but we return Ok.
-            info!("endpoint is not valid, skipping the request");
-            return Ok(AuthInfo::default());
-        }
-        let request_id = ctx.session_id().to_string();
-        let application_name = ctx.console_application_name();
-        async {
-            let request = self
-                .endpoint
-                .get_path("get_endpoint_access_control")
-                .header(X_REQUEST_ID, &request_id)
-                .header(AUTHORIZATION, format!("Bearer {}", &self.jwt))
-                .query(&[("session_id", ctx.session_id())])
-                .query(&[
-                    ("application_name", application_name.as_str()),
-                    ("endpointish", user_info.endpoint.as_str()),
-                    ("role", user_info.user.as_str()),
-                ])
-                .build()?;
-
-            debug!(url = request.url().as_str(), "sending http request");
-            let start = Instant::now();
-            let pause = ctx.latency_timer_pause(crate::metrics::Waiting::Cplane);
-            let response = self.endpoint.execute(request).await?;
-            drop(pause);
-            info!(duration = ?start.elapsed(), "received http response");
-            let body = match parse_body::<GetEndpointAccessControl>(response).await {
-                Ok(body) => body,
-                // Error 404 is special: it's ok not to have a secret.
-                // TODO(anna): retry
-                Err(e) => {
-                    return if e.get_reason().is_not_found() {
-                        // TODO: refactor this because it's weird
-                        // this is a failure to authenticate but we return Ok.
-                        Ok(AuthInfo::default())
-                    } else {
-                        Err(e.into())
-                    };
-                }
-            };
-
-            // Ivan: don't know where it will be used, so I leave it here
-            let _endpoint_vpc_ids = body.allowed_vpc_endpoint_ids.unwrap_or_default();
-
-            let secret = if body.role_secret.is_empty() {
-                None
-            } else {
-                let secret = scram::ServerSecret::parse(&body.role_secret)
-                    .map(AuthSecret::Scram)
-                    .ok_or(GetAuthInfoError::BadSecret)?;
-                Some(secret)
-            };
-            let allowed_ips = body.allowed_ips.unwrap_or_default();
-            Metrics::get()
-                .proxy
-                .allowed_ips_number
-                .observe(allowed_ips.len() as f64);
-            Ok(AuthInfo {
-                secret,
-                allowed_ips,
-                project_id: body.project_id,
-            })
-        }
-        .inspect_err(|e| tracing::debug!(error = ?e))
-        .instrument(info_span!("do_get_auth_info"))
-        .await
-    }
-
-    async fn do_get_endpoint_jwks(
-        &self,
-        ctx: &RequestContext,
-        endpoint: EndpointId,
-    ) -> Result<Vec<AuthRule>, GetEndpointJwksError> {
-        if !self
-            .caches
-            .endpoints_cache
-            .is_valid(ctx, &endpoint.normalize())
-        {
-            return Err(GetEndpointJwksError::EndpointNotFound);
-        }
-        let request_id = ctx.session_id().to_string();
-        async {
-            let request = self
-                .endpoint
-                .get_with_url(|url| {
-                    url.path_segments_mut()
-                        .push("endpoints")
-                        .push(endpoint.as_str())
-                        .push("jwks");
-                })
-                .header(X_REQUEST_ID, &request_id)
-                .header(AUTHORIZATION, format!("Bearer {}", &self.jwt))
-                .query(&[("session_id", ctx.session_id())])
-                .build()
-                .map_err(GetEndpointJwksError::RequestBuild)?;
-
-            debug!(url = request.url().as_str(), "sending http request");
-            let start = Instant::now();
-            let pause = ctx.latency_timer_pause(crate::metrics::Waiting::Cplane);
-            let response = self
-                .endpoint
-                .execute(request)
-                .await
-                .map_err(GetEndpointJwksError::RequestExecute)?;
-            drop(pause);
-            info!(duration = ?start.elapsed(), "received http response");
-
-            let body = parse_body::<EndpointJwksResponse>(response).await?;
-
-            let rules = body
-                .jwks
-                .into_iter()
-                .map(|jwks| AuthRule {
-                    id: jwks.id,
-                    jwks_url: jwks.jwks_url,
-                    audience: jwks.jwt_audience,
-                    role_names: jwks.role_names,
-                })
-                .collect();
-
-            Ok(rules)
-        }
-        .inspect_err(|e| tracing::debug!(error = ?e))
-        .instrument(info_span!("do_get_endpoint_jwks"))
-        .await
-    }
-
-    async fn do_wake_compute(
-        &self,
-        ctx: &RequestContext,
-        user_info: &ComputeUserInfo,
-    ) -> Result<NodeInfo, WakeComputeError> {
-        let request_id = ctx.session_id().to_string();
-        let application_name = ctx.console_application_name();
-        async {
-            let mut request_builder = self
-                .endpoint
-                .get_path("wake_compute")
-                .header("X-Request-ID", &request_id)
-                .header("Authorization", format!("Bearer {}", &self.jwt))
-                .query(&[("session_id", ctx.session_id())])
-                .query(&[
-                    ("application_name", application_name.as_str()),
-                    ("endpointish", user_info.endpoint.as_str()),
-                ]);
-
-            let options = user_info.options.to_deep_object();
-            if !options.is_empty() {
-                request_builder = request_builder.query(&options);
-            }
-
-            let request = request_builder.build()?;
-
-            debug!(url = request.url().as_str(), "sending http request");
-            let start = Instant::now();
-            let pause = ctx.latency_timer_pause(crate::metrics::Waiting::Cplane);
-            let response = self.endpoint.execute(request).await?;
-            drop(pause);
-            info!(duration = ?start.elapsed(), "received http response");
-            let body = parse_body::<WakeCompute>(response).await?;
-
-            // Unfortunately, ownership won't let us use `Option::ok_or` here.
-            let (host, port) = match parse_host_port(&body.address) {
-                None => return Err(WakeComputeError::BadComputeAddress(body.address)),
-                Some(x) => x,
-            };
-
-            // Don't set anything but host and port! This config will be cached.
-            // We'll set username and such later using the startup message.
-            // TODO: add more type safety (in progress).
-            let mut config = compute::ConnCfg::new(host.to_owned(), port);
-            config.ssl_mode(SslMode::Disable); // TLS is not configured on compute nodes.
-
-            let node = NodeInfo {
-                config,
-                aux: body.aux,
-                allow_self_signed_compute: false,
-            };
-
-            Ok(node)
-        }
-        .inspect_err(|e| tracing::debug!(error = ?e))
-        .instrument(info_span!("do_wake_compute"))
-        .await
-    }
-}
-
-impl super::ControlPlaneApi for NeonControlPlaneClient {
-    #[tracing::instrument(skip_all)]
-    async fn get_role_secret(
-        &self,
-        ctx: &RequestContext,
-        user_info: &ComputeUserInfo,
-    ) -> Result<CachedRoleSecret, GetAuthInfoError> {
-        let normalized_ep = &user_info.endpoint.normalize();
-        let user = &user_info.user;
-        if let Some(role_secret) = self
-            .caches
-            .project_info
-            .get_role_secret(normalized_ep, user)
-        {
-            return Ok(role_secret);
-        }
-        let auth_info = self.do_get_auth_info(ctx, user_info).await?;
-        if let Some(project_id) = auth_info.project_id {
-            let normalized_ep_int = normalized_ep.into();
-            self.caches.project_info.insert_role_secret(
-                project_id,
-                normalized_ep_int,
-                user.into(),
-                auth_info.secret.clone(),
-            );
-            self.caches.project_info.insert_allowed_ips(
-                project_id,
-                normalized_ep_int,
-                Arc::new(auth_info.allowed_ips),
-            );
-            ctx.set_project_id(project_id);
-        }
-        // When we just got a secret, we don't need to invalidate it.
-        Ok(Cached::new_uncached(auth_info.secret))
-    }
-
-    async fn get_allowed_ips_and_secret(
-        &self,
-        ctx: &RequestContext,
-        user_info: &ComputeUserInfo,
-    ) -> Result<(CachedAllowedIps, Option<CachedRoleSecret>), GetAuthInfoError> {
-        let normalized_ep = &user_info.endpoint.normalize();
-        if let Some(allowed_ips) = self.caches.project_info.get_allowed_ips(normalized_ep) {
-            Metrics::get()
-                .proxy
-                .allowed_ips_cache_misses
-                .inc(CacheOutcome::Hit);
-            return Ok((allowed_ips, None));
-        }
-        Metrics::get()
-            .proxy
-            .allowed_ips_cache_misses
-            .inc(CacheOutcome::Miss);
-        let auth_info = self.do_get_auth_info(ctx, user_info).await?;
-        let allowed_ips = Arc::new(auth_info.allowed_ips);
-        let user = &user_info.user;
-        if let Some(project_id) = auth_info.project_id {
-            let normalized_ep_int = normalized_ep.into();
-            self.caches.project_info.insert_role_secret(
-                project_id,
-                normalized_ep_int,
-                user.into(),
-                auth_info.secret.clone(),
-            );
-            self.caches.project_info.insert_allowed_ips(
-                project_id,
-                normalized_ep_int,
-                allowed_ips.clone(),
-            );
-            ctx.set_project_id(project_id);
-        }
-        Ok((
-            Cached::new_uncached(allowed_ips),
-            Some(Cached::new_uncached(auth_info.secret)),
-        ))
-    }
-
-    #[tracing::instrument(skip_all)]
-    async fn get_endpoint_jwks(
-        &self,
-        ctx: &RequestContext,
-        endpoint: EndpointId,
-    ) -> Result<Vec<AuthRule>, GetEndpointJwksError> {
-        self.do_get_endpoint_jwks(ctx, endpoint).await
-    }
-
-    #[tracing::instrument(skip_all)]
-    async fn wake_compute(
-        &self,
-        ctx: &RequestContext,
-        user_info: &ComputeUserInfo,
-    ) -> Result<CachedNodeInfo, WakeComputeError> {
-        let key = user_info.endpoint_cache_key();
-
-        macro_rules! check_cache {
-            () => {
-                if let Some(cached) = self.caches.node_info.get(&key) {
-                    let (cached, info) = cached.take_value();
-                    let info = info.map_err(|c| {
-                        info!(key = &*key, "found cached wake_compute error");
-                        WakeComputeError::ControlPlane(ControlPlaneError::Message(Box::new(*c)))
-                    })?;
-
-                    debug!(key = &*key, "found cached compute node info");
-                    ctx.set_project(info.aux.clone());
-                    return Ok(cached.map(|()| info));
-                }
-            };
-        }
-
-        // Every time we do a wakeup http request, the compute node will stay up
-        // for some time (highly depends on the console's scale-to-zero policy);
-        // The connection info remains the same during that period of time,
-        // which means that we might cache it to reduce the load and latency.
-        check_cache!();
-
-        let permit = self.locks.get_permit(&key).await?;
-
-        // after getting back a permit - it's possible the cache was filled
-        // double check
-        if permit.should_check_cache() {
-            // TODO: if there is something in the cache, mark the permit as success.
-            check_cache!();
-        }
-
-        // check rate limit
-        if !self
-            .wake_compute_endpoint_rate_limiter
-            .check(user_info.endpoint.normalize_intern(), 1)
-        {
-            return Err(WakeComputeError::TooManyConnections);
-        }
-
-        let node = permit.release_result(self.do_wake_compute(ctx, user_info).await);
-        match node {
-            Ok(node) => {
-                ctx.set_project(node.aux.clone());
-                debug!(key = &*key, "created a cache entry for woken compute node");
-
-                let mut stored_node = node.clone();
-                // store the cached node as 'warm_cached'
-                stored_node.aux.cold_start_info = ColdStartInfo::WarmCached;
-
-                let (_, cached) = self.caches.node_info.insert_unit(key, Ok(stored_node));
-
-                Ok(cached.map(|()| node))
-            }
-            Err(err) => match err {
-                WakeComputeError::ControlPlane(ControlPlaneError::Message(err)) => {
-                    let Some(status) = &err.status else {
-                        return Err(WakeComputeError::ControlPlane(ControlPlaneError::Message(
-                            err,
-                        )));
-                    };
-
-                    let reason = status
-                        .details
-                        .error_info
-                        .map_or(Reason::Unknown, |x| x.reason);
-
-                    // if we can retry this error, do not cache it.
-                    if reason.can_retry() {
-                        return Err(WakeComputeError::ControlPlane(ControlPlaneError::Message(
-                            err,
-                        )));
-                    }
-
-                    // at this point, we should only have quota errors.
-                    debug!(
-                        key = &*key,
-                        "created a cache entry for the wake compute error"
-                    );
-
-                    self.caches.node_info.insert_ttl(
-                        key,
-                        Err(err.clone()),
-                        Duration::from_secs(30),
-                    );
-
-                    Err(WakeComputeError::ControlPlane(ControlPlaneError::Message(
-                        err,
-                    )))
-                }
-                err => return Err(err),
-            },
-        }
-    }
-}
-
-/// Parse http response body, taking status code into account.
-async fn parse_body<T: for<'a> serde::Deserialize<'a>>(
-    response: http::Response,
-) -> Result<T, ControlPlaneError> {
-    let status = response.status();
-    if status.is_success() {
-        // We shouldn't log raw body because it may contain secrets.
-        info!("request succeeded, processing the body");
-        return Ok(response.json().await?);
-    }
-    let s = response.bytes().await?;
-    // Log plaintext to be able to detect, whether there are some cases not covered by the error struct.
-    info!("response_error plaintext: {:?}", s);
-
-    // Don't throw an error here because it's not as important
-    // as the fact that the request itself has failed.
-    let mut body = serde_json::from_slice(&s).unwrap_or_else(|e| {
-        warn!("failed to parse error body: {e}");
-        ControlPlaneErrorMessage {
-            error: "reason unclear (malformed error message)".into(),
-            http_status_code: status,
-            status: None,
-        }
-    });
-    body.http_status_code = status;
-
-    warn!("console responded with an error ({status}): {body:?}");
-    Err(ControlPlaneError::Message(Box::new(body)))
-}
-
-fn parse_host_port(input: &str) -> Option<(&str, u16)> {
-    let (host, port) = input.rsplit_once(':')?;
-    let ipv6_brackets: &[_] = &['[', ']'];
-    Some((host.trim_matches(ipv6_brackets), port.parse().ok()?))
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn test_parse_host_port_v4() {
-        let (host, port) = parse_host_port("127.0.0.1:5432").expect("failed to parse");
-        assert_eq!(host, "127.0.0.1");
-        assert_eq!(port, 5432);
-    }
-
-    #[test]
-    fn test_parse_host_port_v6() {
-        let (host, port) = parse_host_port("[2001:db8::1]:5432").expect("failed to parse");
-        assert_eq!(host, "2001:db8::1");
-        assert_eq!(port, 5432);
-    }
-
-    #[test]
-    fn test_parse_host_port_url() {
-        let (host, port) = parse_host_port("compute-foo-bar-1234.default.svc.cluster.local:5432")
-            .expect("failed to parse");
-        assert_eq!(host, "compute-foo-bar-1234.default.svc.cluster.local");
-        assert_eq!(port, 5432);
-    }
-}
--- a/proxy/src/control_plane/client/mod.rs
+++ b/proxy/src/control_plane/client/mod.rs
@@ -1,4 +1,3 @@
-pub mod cplane_proxy_v1;
 #[cfg(any(test, feature = "testing"))]
 pub mod mock;
 pub mod neon;
@@ -28,8 +27,6 @@ use crate::types::EndpointId;
 #[non_exhaustive]
 #[derive(Clone)]
 pub enum ControlPlaneClient {
-    /// New Proxy V1 control plane API
-    ProxyV1(cplane_proxy_v1::NeonControlPlaneClient),
    /// Current Management API (V2).
    Neon(neon::NeonControlPlaneClient),
    /// Local mock control plane.
@@ -48,7 +45,6 @@ impl ControlPlaneApi for ControlPlaneClient {
        user_info: &ComputeUserInfo,
    ) -> Result<CachedRoleSecret, errors::GetAuthInfoError> {
        match self {
-            Self::ProxyV1(api) => api.get_role_secret(ctx, user_info).await,
            Self::Neon(api) => api.get_role_secret(ctx, user_info).await,
            #[cfg(any(test, feature = "testing"))]
            Self::PostgresMock(api) => api.get_role_secret(ctx, user_info).await,
@@ -65,7 +61,6 @@ impl ControlPlaneApi for ControlPlaneClient {
        user_info: &ComputeUserInfo,
    ) -> Result<(CachedAllowedIps, Option<CachedRoleSecret>), errors::GetAuthInfoError> {
        match self {
-            Self::ProxyV1(api) => api.get_allowed_ips_and_secret(ctx, user_info).await,
            Self::Neon(api) => api.get_allowed_ips_and_secret(ctx, user_info).await,
            #[cfg(any(test, feature = "testing"))]
            Self::PostgresMock(api) => api.get_allowed_ips_and_secret(ctx, user_info).await,
@@ -80,7 +75,6 @@ impl ControlPlaneApi for ControlPlaneClient {
        endpoint: EndpointId,
    ) -> Result<Vec<AuthRule>, errors::GetEndpointJwksError> {
        match self {
-            Self::ProxyV1(api) => api.get_endpoint_jwks(ctx, endpoint).await,
            Self::Neon(api) => api.get_endpoint_jwks(ctx, endpoint).await,
            #[cfg(any(test, feature = "testing"))]
            Self::PostgresMock(api) => api.get_endpoint_jwks(ctx, endpoint).await,
@@ -95,7 +89,6 @@ impl ControlPlaneApi for ControlPlaneClient {
        user_info: &ComputeUserInfo,
    ) -> Result<CachedNodeInfo, errors::WakeComputeError> {
        match self {
-            Self::ProxyV1(api) => api.wake_compute(ctx, user_info).await,
            Self::Neon(api) => api.wake_compute(ctx, user_info).await,
            #[cfg(any(test, feature = "testing"))]
            Self::PostgresMock(api) => api.wake_compute(ctx, user_info).await,
--- a/proxy/src/control_plane/client/neon.rs
+++ b/proxy/src/control_plane/client/neon.rs
@@ -1,4 +1,4 @@
-//! Stale console backend, remove after migrating to Proxy V1 API (#15245).
+//! Production console backend.

 use std::sync::Arc;
 use std::time::Duration;
--- a/proxy/src/control_plane/messages.rs
+++ b/proxy/src/control_plane/messages.rs
@@ -230,16 +230,6 @@ pub(crate) struct GetRoleSecret {
    pub(crate) project_id: Option<ProjectIdInt>,
 }

-/// Response which holds client's auth secret, e.g. [`crate::scram::ServerSecret`].
-/// Returned by the `/get_endpoint_access_control` API method.
-#[derive(Deserialize)]
-pub(crate) struct GetEndpointAccessControl {
-    pub(crate) role_secret: Box<str>,
-    pub(crate) allowed_ips: Option<Vec<IpPattern>>,
-    pub(crate) project_id: Option<ProjectIdInt>,
-    pub(crate) allowed_vpc_endpoint_ids: Option<Vec<EndpointIdInt>>,
-}
-
 // Manually implement debug to omit sensitive info.
 impl fmt::Debug for GetRoleSecret {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
--- a/proxy/src/proxy/connect_compute.rs
+++ b/proxy/src/proxy/connect_compute.rs
@@ -66,8 +66,6 @@ pub(crate) trait ComputeConnectBackend {
 }

 pub(crate) struct TcpMechanism<'a> {
-    pub(crate) params_compat: bool,
-
    /// KV-dictionary with PostgreSQL connection params.
    pub(crate) params: &'a StartupMessageParams,

@@ -94,7 +92,7 @@ impl ConnectMechanism for TcpMechanism<'_> {
    }

    fn update_connect_config(&self, config: &mut compute::ConnCfg) {
-        config.set_startup_params(self.params, self.params_compat);
+        config.set_startup_params(self.params);
    }
 }

--- a/proxy/src/proxy/mod.rs
+++ b/proxy/src/proxy/mod.rs
@@ -338,17 +338,9 @@ pub(crate) async fn handle_client<S: AsyncRead + AsyncWrite + Unpin>(
        }
    };

-    let params_compat = match &user_info {
-        auth::Backend::ControlPlane(_, info) => {
-            info.info.options.get(NeonOptions::PARAMS_COMPAT).is_some()
-        }
-        auth::Backend::Local(_) => false,
-    };
-
    let mut node = connect_to_compute(
        ctx,
        &TcpMechanism {
-            params_compat,
            params: &params,
            locks: &config.connect_compute_locks,
        },
@@ -417,47 +409,19 @@ pub(crate) async fn prepare_client_connection<P>(
 pub(crate) struct NeonOptions(Vec<(SmolStr, SmolStr)>);

 impl NeonOptions {
-    // proxy options:
-
-    /// `PARAMS_COMPAT` allows opting in to forwarding all startup parameters from client to compute.
-    const PARAMS_COMPAT: &str = "proxy_params_compat";
-
-    // cplane options:
-
-    /// `LSN` allows provisioning an ephemeral compute with time-travel to the provided LSN.
-    const LSN: &str = "lsn";
-
-    /// `ENDPOINT_TYPE` allows configuring an ephemeral compute to be read_only or read_write.
-    const ENDPOINT_TYPE: &str = "endpoint_type";
-
    pub(crate) fn parse_params(params: &StartupMessageParams) -> Self {
        params
            .options_raw()
            .map(Self::parse_from_iter)
            .unwrap_or_default()
    }
-
    pub(crate) fn parse_options_raw(options: &str) -> Self {
        Self::parse_from_iter(StartupMessageParams::parse_options_raw(options))
    }

-    pub(crate) fn get(&self, key: &str) -> Option<SmolStr> {
-        self.0
-            .iter()
-            .find_map(|(k, v)| (k == key).then_some(v))
-            .cloned()
-    }
-
    pub(crate) fn is_ephemeral(&self) -> bool {
-        self.0.iter().any(|(k, _)| match &**k {
-            // This is not a cplane option, we know it does not create ephemeral computes.
-            Self::PARAMS_COMPAT => false,
-            Self::LSN => true,
-            Self::ENDPOINT_TYPE => true,
-            // err on the side of caution. any cplane options we don't know about
-            // might lead to ephemeral computes.
-            _ => true,
-        })
+        // Currently, neon endpoint options are all reserved for ephemeral endpoints.
+        !self.0.is_empty()
    }

    fn parse_from_iter<'a>(options: impl Iterator<Item = &'a str>) -> Self {
--- a/proxy/src/proxy/tests/mitm.rs
+++ b/proxy/src/proxy/tests/mitm.rs
@@ -55,13 +55,7 @@ async fn proxy_mitm(

        // give the end_server the startup parameters
        let mut buf = BytesMut::new();
-        frontend::startup_message(
-            &postgres_protocol::message::frontend::StartupMessageParams {
-                params: startup.params.into(),
-            },
-            &mut buf,
-        )
-        .unwrap();
+        frontend::startup_message(startup.iter(), &mut buf).unwrap();
        end_server.send(buf.freeze()).await.unwrap();

        // proxy messages between end_client and end_server
--- a/proxy/src/proxy/tests/mod.rs
+++ b/proxy/src/proxy/tests/mod.rs
@@ -252,7 +252,7 @@ async fn handshake_raw() -> anyhow::Result<()> {
    let _conn = postgres_client::Config::new("test".to_owned(), 5432)
        .user("john_doe")
        .dbname("earth")
-        .set_param("options", "project=generic-project-name")
+        .options("project=generic-project-name")
        .ssl_mode(SslMode::Prefer)
        .connect_raw(server, NoTls)
        .await?;
--- a/proxy/src/serverless/backend.rs
+++ b/proxy/src/serverless/backend.rs
@@ -309,13 +309,10 @@ impl PoolingBackend {
            .config
            .user(&conn_info.user_info.user)
            .dbname(&conn_info.dbname)
-            .set_param(
-                "options",
-                &format!(
-                    "-c pg_session_jwt.jwk={}",
-                    serde_json::to_string(&jwk).expect("serializing jwk to json should not fail")
-                ),
-            );
+            .options(&format!(
+                "-c pg_session_jwt.jwk={}",
+                serde_json::to_string(&jwk).expect("serializing jwk to json should not fail")
+            ));

        let pause = ctx.latency_timer_pause(crate::metrics::Waiting::Compute);
        let (client, connection) = config.connect(postgres_client::NoTls).await?;
--- a/safekeeper/src/http/client.rs
+++ b/safekeeper/src/http/client.rs
@@ -8,7 +8,6 @@
 //! etc.

 use reqwest::{IntoUrl, Method, StatusCode};
-use std::error::Error as _;
 use utils::{
    http::error::HttpErrorBody,
    id::{NodeId, TenantId, TimelineId},
@@ -27,7 +26,7 @@ pub struct Client {
 #[derive(thiserror::Error, Debug)]
 pub enum Error {
    /// Failed to receive body (reqwest error).
-    #[error("receive body: {0}{}", .0.source().map(|e| format!(": {e}")).unwrap_or_default())]
+    #[error("receive body: {0}")]
    ReceiveBody(reqwest::Error),

    /// Status is not ok, but failed to parse body as `HttpErrorBody`.
--- a/storage_controller/src/compute_hook.rs
+++ b/storage_controller/src/compute_hook.rs
@@ -1,4 +1,3 @@
-use std::error::Error as _;
 use std::sync::Arc;
 use std::{collections::HashMap, time::Duration};

@@ -173,7 +172,7 @@ struct ComputeHookNotifyRequest {
 #[derive(thiserror::Error, Debug)]
 pub(crate) enum NotifyError {
    // Request was not send successfully, e.g. transport error
-    #[error("Sending request: {0}{}", .0.source().map(|e| format!(": {e}")).unwrap_or_default())]
+    #[error("Sending request: {0}")]
    Request(#[from] reqwest::Error),
    // Request could not be serviced right now due to ongoing Operation in control plane, but should be possible soon.
    #[error("Control plane tenant busy")]
--- a/storage_controller/src/peer_client.rs
+++ b/storage_controller/src/peer_client.rs
@@ -1,9 +1,7 @@
 use crate::tenant_shard::ObservedState;
 use pageserver_api::shard::TenantShardId;
 use serde::{Deserialize, Serialize};
-use std::collections::HashMap;
-use std::error::Error as _;
-use std::time::Duration;
+use std::{collections::HashMap, time::Duration};
 use tokio_util::sync::CancellationToken;

 use hyper::Uri;
@@ -19,14 +17,11 @@ pub(crate) struct PeerClient {

 #[derive(thiserror::Error, Debug)]
 pub(crate) enum StorageControllerPeerError {
-    #[error(
-        "failed to deserialize error response with status code {0} at {1}: {2}{}",
-        .2.source().map(|e| format!(": {e}")).unwrap_or_default()
-    )]
+    #[error("failed to deserialize error response with status code {0} at {1}: {2}")]
    DeserializationError(StatusCode, Url, reqwest::Error),
    #[error("storage controller peer API error ({0}): {1}")]
    ApiError(StatusCode, String),
-    #[error("failed to send HTTP request: {0}{}", .0.source().map(|e| format!(": {e}")).unwrap_or_default())]
+    #[error("failed to send HTTP request: {0}")]
    SendError(reqwest::Error),
    #[error("Cancelled")]
    Cancelled,
--- a/storage_controller/src/service.rs
+++ b/storage_controller/src/service.rs
@@ -5681,7 +5681,7 @@ impl Service {
        }

        match node_policy {
-            NodeSchedulingPolicy::Active => {
+            NodeSchedulingPolicy::Active | NodeSchedulingPolicy::Pause => {
                self.node_configure(node_id, None, Some(NodeSchedulingPolicy::Draining))
                    .await?;

--- a/storage_scrubber/src/cloud_admin_api.rs
+++ b/storage_scrubber/src/cloud_admin_api.rs
@@ -1,5 +1,3 @@
-use std::error::Error as _;
-
 use chrono::{DateTime, Utc};
 use futures::Future;
 use hex::FromHex;
@@ -32,18 +30,14 @@ impl std::fmt::Display for Error {
        match &self.kind {
            ErrorKind::RequestSend(e) => write!(
                f,
-                "Failed to send a request. Context: {}, error: {}{}",
-                self.context,
-                e,
-                e.source().map(|e| format!(": {e}")).unwrap_or_default()
+                "Failed to send a request. Context: {}, error: {}",
+                self.context, e
            ),
            ErrorKind::BodyRead(e) => {
                write!(
                    f,
-                    "Failed to read a request body. Context: {}, error: {}{}",
-                    self.context,
-                    e,
-                    e.source().map(|e| format!(": {e}")).unwrap_or_default()
+                    "Failed to read a request body. Context: {}, error: {}",
+                    self.context, e
                )
            }
            ErrorKind::ResponseStatus(status) => {
--- a/test_runner/fixtures/neon_fixtures.py
+++ b/test_runner/fixtures/neon_fixtures.py
@@ -269,7 +269,7 @@ class PgProtocol:
            for match in re.finditer(r"-c(\w*)=(\w*)", options):
                key = match.group(1)
                val = match.group(2)
-                if "server_settings" in conn_options:
+                if "server_options" in conn_options:
                    conn_options["server_settings"].update({key: val})
                else:
                    conn_options["server_settings"] = {key: val}
--- a/test_runner/fixtures/pageserver_mitm.py
+++ b/test_runner/fixtures/pageserver_mitm.py
@@ -1,201 +0,0 @@
-# Intercept compute -> pageserver connections, to simulate various failure modes
-
-from __future__ import annotations
-
-import asyncio
-import struct
-import threading
-import traceback
-from asyncio import TaskGroup
-from enum import Enum
-
-from fixtures.log_helper import log
-
-
-class ConnectionState(Enum):
-    HANDSHAKE = (1,)
-    AUTHOK = (2,)
-    COPYBOTH = (3,)
-
-
-class BreakConnectionException(Exception):
-    def __init__(self, message):
-        super().__init__(message)
-        self.message = message
-
-
-class ProxyShutdownException(Exception):
-    """Exception raised to shut down connection when the proxy is shut down."""
-
-
-# The handshake flow:
-#
-# 1. compute establishes TCP connection
-# 2. libpq handshake and auth
-# 3. Enter CopyBoth mode
-#
-# From then on, CopyData messages are exchanged in both directions
-class Connection:
-    def __init__(
-        self,
-        conn_id,
-        compute_reader,
-        compute_writer,
-        shutdown_event,
-        dest_port: int,
-        response_cb=None,
-    ):
-        self.conn_id = conn_id
-        self.compute_reader = compute_reader
-        self.compute_writer = compute_writer
-        self.shutdown_event = shutdown_event
-        self.response_cb = response_cb
-        self.dest_port = dest_port
-
-        self.state = ConnectionState.HANDSHAKE
-
-    async def run(self):
-        async def wait_for_shutdown():
-            await self.shutdown_event.wait()
-            raise ProxyShutdownException
-
-        try:
-            addr = self.compute_writer.get_extra_info("peername")
-            log.info(f"[{self.conn_id}] connection received from {addr}")
-
-            async with TaskGroup() as group:
-                group.create_task(wait_for_shutdown())
-
-                self.ps_reader, self.ps_writer = await asyncio.open_connection(
-                    "localhost", self.dest_port
-                )
-
-                group.create_task(self.handle_compute_to_pageserver())
-                group.create_task(self.handle_pageserver_to_compute())
-
-        except* ProxyShutdownException:
-            log.info(f"[{self.conn_id}] proxy shutting down")
-
-        except* asyncio.exceptions.IncompleteReadError as e:
-            log.info(f"[{self.conn_id}] EOF reached: {e}")
-
-        except* BreakConnectionException as eg:
-            for e in eg.exceptions:
-                log.info(f"[{self.conn_id}] callback breaks connection: {e}")
-
-        except* Exception as e:
-            s = "\n".join(traceback.format_exception(e))
-            log.info(f"[{self.conn_id}] {s}")
-
-        self.compute_writer.close()
-        self.ps_writer.close()
-        await self.compute_writer.wait_closed()
-        await self.ps_writer.wait_closed()
-
-    async def handle_compute_to_pageserver(self):
-        while self.state == ConnectionState.HANDSHAKE:
-            rawmsg = await self.compute_reader.read(1000)
-            log.debug(f"[{self.conn_id}] C -> PS: handshake msg len {len(rawmsg)}")
-            self.ps_writer.write(rawmsg)
-            await self.ps_writer.drain()
-
-        while True:
-            msgtype = await self.compute_reader.readexactly(1)
-            msglen_bytes = await self.compute_reader.readexactly(4)
-            (msglen,) = struct.unpack("!L", msglen_bytes)
-            payload = await self.compute_reader.readexactly(msglen - 4)
-
-            # request_callback()
-            # CopyData
-            if msgtype == b"d":
-                # TODO: call callback
-                log.debug(f"[{self.conn_id}] C -> PS: CopyData ({msglen} bytes)")
-                pass
-            else:
-                log.debug(f"[{self.conn_id}] C -> PS: message of type '{msgtype}' ({msglen} bytes)")
-
-            self.ps_writer.write(msgtype)
-            self.ps_writer.write(msglen_bytes)
-            self.ps_writer.write(payload)
-            await self.ps_writer.drain()
-
-    async def handle_pageserver_to_compute(self):
-        while True:
-            msgtype = await self.ps_reader.readexactly(1)
-
-            # response to SSLRequest
-            if msgtype == b"N" and self.state == ConnectionState.HANDSHAKE:
-                log.debug(f"[{self.conn_id}] PS -> C: N")
-                self.compute_writer.write(msgtype)
-                await self.compute_writer.drain()
-                continue
-
-            msglen_bytes = await self.ps_reader.readexactly(4)
-            (msglen,) = struct.unpack("!L", msglen_bytes)
-            payload = await self.ps_reader.readexactly(msglen - 4)
-
-            # AuthenticationOK
-            if msgtype == b"R":
-                self.state = ConnectionState.AUTHOK
-                log.debug(f"[{self.conn_id}] PS -> C: AuthenticationOK ({msglen} bytes)")
-
-            # CopyBothresponse
-            elif msgtype == b"W":
-                self.state = ConnectionState.COPYBOTH
-                log.debug(f"[{self.conn_id}] PS -> C: CopyBothResponse ({msglen} bytes)")
-
-            # CopyData
-            elif msgtype == b"d":
-                log.debug(f"[{self.conn_id}] PS -> C: CopyData ({msglen} bytes)")
-                if self.response_cb is not None:
-                    await self.response_cb(self.conn_id)
-                pass
-
-            else:
-                log.debug(f"[{self.conn_id}] PS -> C: message of type '{msgtype}' ({msglen} bytes)")
-
-            self.compute_writer.write(msgtype)
-            self.compute_writer.write(msglen_bytes)
-            self.compute_writer.write(payload)
-            await self.compute_writer.drain()
-
-
-class PageserverProxy:
-    def __init__(self, listen_port: int, dest_port: int, response_cb=None):
-        self.listen_port = listen_port
-        self.dest_port = dest_port
-        self.response_cb = response_cb
-        self.conn_counter = 0
-        self.shutdown_event = asyncio.Event()
-
-    def shutdown(self):
-        self.serve_task.cancel()
-        self.shutdown_event.set()
-
-    async def handle_client(self, compute_reader, compute_writer):
-        self.conn_counter += 1
-        conn_id = self.conn_counter
-        conn = Connection(
-            conn_id,
-            compute_reader,
-            compute_writer,
-            self.shutdown_event,
-            self.dest_port,
-            self.response_cb,
-        )
-        await conn.run()
-
-    async def run_server(self):
-        log.info("run_server called")
-        server = await asyncio.start_server(self.handle_client, "localhost", self.listen_port)
-
-        self.serve_task = asyncio.create_task(server.serve_forever())
-
-        try:
-            await self.serve_task
-        except asyncio.CancelledError:
-            log.info("proxy shutting down")
-
-    def launch_server_in_thread(self):
-        t1 = threading.Thread(target=asyncio.run, args=self.run_server)
-        t1.start()
--- a/test_runner/performance/test_perf_ingest_using_pgcopydb.py
+++ b/test_runner/performance/test_perf_ingest_using_pgcopydb.py
@@ -60,13 +60,13 @@ def build_pgcopydb_command(pgcopydb_filter_file: Path, test_output_dir: Path):
        "--no-acl",
        "--skip-db-properties",
        "--table-jobs",
-        "4",
+        "8",
        "--index-jobs",
-        "4",
+        "8",
        "--restore-jobs",
-        "4",
+        "8",
        "--split-tables-larger-than",
-        "10GB",
+        "5GB",
        "--skip-extensions",
        "--use-copy-binary",
        "--filters",
@@ -136,7 +136,7 @@ def run_command_and_log_output(command, log_file_path: Path):
        "LD_LIBRARY_PATH": f"{os.getenv('PGCOPYDB_LIB_PATH')}:{os.getenv('PG_16_LIB_PATH')}",
        "PGCOPYDB_SOURCE_PGURI": cast(str, os.getenv("BENCHMARK_INGEST_SOURCE_CONNSTR")),
        "PGCOPYDB_TARGET_PGURI": cast(str, os.getenv("BENCHMARK_INGEST_TARGET_CONNSTR")),
-        "PGOPTIONS": "-c maintenance_work_mem=8388608 -c max_parallel_maintenance_workers=7",
+        "PGOPTIONS": "-c maintenance_work_mem=8388608 -c max_parallel_maintenance_workers=16",
    }
    # Combine the current environment with custom variables
    env = os.environ.copy()
@@ -184,7 +184,7 @@ def parse_log_and_report_metrics(
            for metric_name, pattern in metric_patterns.items():
                if pattern.search(line):
                    # Extract duration and convert it to seconds
-                    duration_match = re.search(r"\d+h\d+m|\d+m\d+s|\d+s|\d+ms|\d+\.\d+s", line)
+                    duration_match = re.search(r"\d+h\d+m|\d+s|\d+ms|\d+\.\d+s", line)
                    if duration_match:
                        duration_str = duration_match.group(0)
                        parts = re.findall(r"\d+[a-zA-Z]+", duration_str)
--- a/test_runner/performance/test_storage_controller_scale.py
+++ b/test_runner/performance/test_storage_controller_scale.py
@@ -72,7 +72,7 @@ def test_storage_controller_many_tenants(
    we don't fall over for a thousand shards.
    """

-    neon_env_builder.num_pageservers = 6
+    neon_env_builder.num_pageservers = 5
    neon_env_builder.storage_controller_config = {
        # Default neon_local uses a small timeout: use a longer one to tolerate longer pageserver restarts.
        # TODO: tune this down as restarts get faster (https://github.com/neondatabase/neon/pull/7553), to
@@ -84,11 +84,6 @@ def test_storage_controller_many_tenants(
        compute_reconfigure_listener.control_plane_compute_hook_api
    )

-    AZS = ["alpha", "bravo", "charlie"]
-    neon_env_builder.pageserver_config_override = lambda ps_cfg: ps_cfg.update(
-        {"availability_zone": f"az-{AZS[ps_cfg['id'] % len(AZS)]}"}
-    )
-
    # A small sleep on each call into the notify hook, to simulate the latency of doing a database write
    compute_reconfigure_listener.register_on_notify(lambda body: time.sleep(0.01))

--- a/test_runner/regress/test_bad_connection.py
+++ b/test_runner/regress/test_bad_connection.py
@@ -1,16 +1,12 @@
 from __future__ import annotations

-import asyncio
 import random
 import time
-from asyncio import TaskGroup

 import psycopg2.errors
 import pytest
 from fixtures.log_helper import log
-from fixtures.neon_fixtures import NeonEnvBuilder, PgBin
-from fixtures.pageserver_mitm import BreakConnectionException, PageserverProxy
-from fixtures.port_distributor import PortDistributor
+from fixtures.neon_fixtures import NeonEnvBuilder


@pytest.mark.timeout(600)
@@ -84,197 +80,3 @@ def test_compute_pageserver_connection_stress(neon_env_builder: NeonEnvBuilder):
    # do a graceful shutdown which would had caught the allowed_errors before
    # https://github.com/neondatabase/neon/pull/8632
    env.pageserver.stop()
-
-
-#
-# Observations:
-#
-# 1. If the backend is waiting for response to GetPage request, and the client disconnects,
-#    the backend will not immediately abort the GetPage request. It will not notice that the
-#    client is gone, until it tries to send something back to the client, or if a timeout
-#    kills the query.
-#
-# So to reproduce the traffic jam, you need:
-#
-# - A network glitch, which causes one GetPage request/response to be lost or delayed.
-#   It might get lost at IP level, and TCP retransmits might take a long time. Or there might
-#   be a glitch in the pageserver or compute, which causes the request to be "stuck".
-#
-# - An application with a application level timeout and retry. If the
-#   query doesn't return in a timely a fashion, the application kills the connection and
-#   retries the query, or a runs similar query that needs the same page.
-#
-# The first time the GetPage request is stuck and it disconnects, it leaves behind a
-# backend that's still waiting for the GetPage response, and is holding the buffer lock.
-# The client has closed the connection, but the server doesn't get the memo.
-# On each subsequent retry, the connection will block waiting for the buffer lock, give up,
-# and leave behind another backend blocked indefinitely.
-#
-# The situation unravels when the original backend doing the GetPage request finally
-# gets a response, or it gets confirmation that the TCP connection is lost.
-#
-# This test reproduces the traffic jam using a MITM proxy between pageserver and compute,
-# and forcing one GetPage request to get stuck.
-#
-# Recommendations:
-# - set client_connection_check_interval = '10s'. This makes Postgres wake up and check
-#   for client connection loss. It's not perfect, it might not notice if the client has
-#   e.g rebooted without sending a RST packet, but there's no downside
-#
-# - Add a timeout to GetPage requests. If no response is received from the pageserver
-#   in, say, 10 s, terminate the pageserver connection and retry. XXX: Negotiate this
-#   behavior with the storage team
-#
-#
-@pytest.mark.timeout(120)
-def test_compute_pageserver_connection_stress2(
-    neon_env_builder: NeonEnvBuilder, port_distributor: PortDistributor, pg_bin: PgBin
-):
-    env = neon_env_builder.init_start()
-
-    # Set up the MITM proxy
-
-    global error_fraction
-    global delay_fraction
-
-    error_fraction = 0
-    delay_fraction = 0
-
-    async def response_cb(conn_id):
-        global delay_fraction
-        global error_fraction
-
-        if random.random() < error_fraction:
-            raise BreakConnectionException("unlucky")
-
-        orig_delay_fraction = delay_fraction
-        if random.random() < delay_fraction:
-            delay_fraction = 0
-            log.info(f"[{conn_id}] making getpage request STUCK")
-            try:
-                await asyncio.sleep(300)
-            finally:
-                delay_fraction = orig_delay_fraction
-                log.info(f"[{conn_id}] delay finished")
-
-    mitm_listen_port = port_distributor.get_port()
-    mitm = PageserverProxy(mitm_listen_port, env.pageserver.service_port.pg, response_cb)
-
-    def main():
-        global error_fraction, delay_fraction
-        endpoint = env.endpoints.create(
-            "main",
-            config_lines=[
-                "max_connections=1000",
-                "shared_buffers=8MB",
-                "log_connections=on",
-                "log_disconnections=on",
-            ],
-        )
-        endpoint.start()
-
-        with open(endpoint.pg_data_dir_path() / "postgresql.conf", "a") as conf:
-            conf.write(
-                f"neon.pageserver_connstring='postgres://localhost:{mitm_listen_port}'  # MITM proxy\n"
-            )
-
-        pg_conn = endpoint.connect()
-        cur = pg_conn.cursor()
-
-        cur.execute("select pg_reload_conf()")
-
-        scale = 5
-        connstr = endpoint.connstr()
-        log.info(f"Start a pgbench workload on pg {connstr}")
-
-        error_fraction = 0.001
-
-        pg_bin.run_capture(["pgbench", "-i", "-I", "dtGvp", f"-s{scale}", connstr])
-        error_fraction = 0.00
-        delay_fraction = 0.001
-
-        cur.execute("select max(aid) from pgbench_accounts")
-        num_accounts = 100000 * scale
-
-        num_clients = 200
-
-        app = WalkingApplication(num_accounts, num_clients, endpoint, 1000000)
-        asyncio.run(app.run())
-
-        mitm.shutdown()
-
-    async def mm():
-        await asyncio.gather(asyncio.to_thread(main), mitm.run_server())
-
-    asyncio.run(mm())
-
-    # do a graceful shutdown which would had caught the allowed_errors before
-    # https://github.com/neondatabase/neon/pull/8632
-    env.pageserver.stop()
-
-
-class WalkingApplication:
-    """
-    A test application with following characteristics:
-
-    - It performs single-row lookups in pgbench_accounts table, just like pgbench -S
-
-    - Whenever a query takes longer than 10s, the application disconnects, reconnects,
-      and retries the query, with the same parameter. This way, if there's a problem
-      with a single page, the application will keep retrying it rather than work
-      around it.
-
-    - The lookups are not randomly distributed, but form a "walking herd" pattern,
-      where the queries walk through all accounts, with some randomness. This way,
-      there's a lot of locality of access, but the locality moves throughout the
-      table.
-
-    """
-
-    def __init__(self, num_accounts, num_clients, endpoint, num_xacts):
-        self.num_accounts = num_accounts
-        self.num_clients = num_clients
-        self.endpoint = endpoint
-        self.running = True
-        self.num_xacts = num_xacts
-
-        self.xacts_started = 0
-        self.xacts_performed = 0
-        self.xacts_failed = 0
-
-    async def run(self):
-        async with TaskGroup() as group:
-            for i in range(1, self.num_clients):
-                group.create_task(self.walking_client(i))
-
-    async def walking_client(self, client_id):
-        local_xacts_performed = 0
-
-        conn = None
-        stmt = None
-        failed = False
-        while self.running and self.xacts_started < self.num_xacts:
-            self.xacts_started += 1
-            if not failed:
-                aid = (self.xacts_started * 100 + random.randint(0, 100)) % self.num_accounts + 1
-
-            if conn is None:
-                conn = await self.endpoint.connect_async()
-                await conn.execute("set statement_timeout=0")
-                stmt = await conn.prepare("SELECT abalance FROM pgbench_accounts WHERE aid = $1")
-
-            try:
-                async with asyncio.timeout(10):
-                    res = await stmt.fetchval(aid)
-                    if local_xacts_performed % 1000 == 0:
-                        log.info(
-                            f"[{client_id}] result {self.xacts_performed}/{self.num_xacts}: balance of account {aid}: {res}"
-                        )
-                    self.xacts_performed += 1
-                    local_xacts_performed += 1
-                    failed = False
-            except TimeoutError:
-                log.info(f"[{client_id}] query on aid {aid} timed out. Reconnecting")
-                conn.terminate()
-                conn = None
-                failed = True
--- a/test_runner/regress/test_broken_timeline.py
+++ b/test_runner/regress/test_broken_timeline.py
@@ -29,6 +29,8 @@ def test_local_corruption(neon_env_builder: NeonEnvBuilder):
            ".*failed to load metadata.*",
            ".*load failed.*load local timeline.*",
            ".*: layer load failed, assuming permanent failure:.*",
+            ".*failed to get checkpoint bytes.*",
+            ".*failed get control bytes.*",
        ]
    )

@@ -75,7 +77,7 @@ def test_local_corruption(neon_env_builder: NeonEnvBuilder):
    # (We don't check layer file contents on startup, when loading the timeline)
    #
    # This will change when we implement checksums for layers
-    with pytest.raises(Exception, match="get_values_reconstruct_data for layer ") as err:
+    with pytest.raises(Exception, match="failed to get checkpoint bytes") as err:
        pg1.start()
    log.info(
        f"As expected, compute startup failed for timeline {tenant1}/{timeline1} with corrupt layers: {err}"
--- a/test_runner/regress/test_pgstat.py
+++ b/test_runner/regress/test_pgstat.py
@@ -0,0 +1,62 @@
+import pytest
+from fixtures.neon_fixtures import NeonEnv
+from fixtures.pg_version import PgVersion
+
+
+#
+# Test that pgstat statistic is preserved across sessions
+#
+def test_pgstat(neon_simple_env: NeonEnv):
+    env = neon_simple_env
+    if env.pg_version == PgVersion.V14:
+        pytest.skip("PG14 doesn't support pgstat statistic persistence")
+
+    env.pageserver.allowed_errors.append(".*this timeline is using deprecated aux file policy V1.*")
+
+    n = 10000
+    endpoint = env.endpoints.create_start("main")
+
+    con = endpoint.connect()
+    cur = con.cursor()
+
+    cur.execute("create table t(x integer)")
+    cur.execute(f"insert into t values (generate_series(1,{n}))")
+    cur.execute("vacuum analyze t")
+    cur.execute("select sum(x) from t")
+    cur.execute("update t set x=x+1")
+
+    cur.execute("select pg_stat_force_next_flush()")
+
+    cur.execute(
+        "select seq_scan,seq_tup_read,n_tup_ins,n_tup_upd,n_live_tup,n_dead_tup, vacuum_count,analyze_count from pg_stat_user_tables"
+    )
+    rec = cur.fetchall()[0]
+    assert rec == (2, n * 2, n, n, n * 2, n, 1, 1)
+
+    endpoint.stop()
+    endpoint.start()
+
+    con = endpoint.connect()
+    cur = con.cursor()
+
+    cur.execute(
+        "select seq_scan,seq_tup_read,n_tup_ins,n_tup_upd,n_live_tup,n_dead_tup, vacuum_count,analyze_count from pg_stat_user_tables"
+    )
+    rec = cur.fetchall()[0]
+    assert rec == (2, n * 2, n, n, n * 2, n, 1, 1)
+
+    cur.execute("update t set x=x+1")
+
+    # stop without checkpoint
+    endpoint.stop(mode="immediate")
+    endpoint.start()
+
+    con = endpoint.connect()
+    cur = con.cursor()
+
+    cur.execute(
+        "select seq_scan,seq_tup_read,n_tup_ins,n_tup_upd,n_live_tup,n_dead_tup, vacuum_count,analyze_count from pg_stat_user_tables"
+    )
+    rec = cur.fetchall()[0]
+    # pgstat information should be discarded in case of abnormal termination
+    assert rec == (0, 0, 0, 0, 0, 0, 0, 0)
--- a/test_runner/regress/test_proxy.py
+++ b/test_runner/regress/test_proxy.py
@@ -5,7 +5,6 @@ import json
 import subprocess
 import time
 import urllib.parse
-from contextlib import closing
 from typing import TYPE_CHECKING

 import psycopg2
@@ -132,24 +131,6 @@ def test_proxy_options(static_proxy: NeonProxy, option_name: str):
    assert out[0][0] == " str"


-@pytest.mark.asyncio
-async def test_proxy_arbitrary_params(static_proxy: NeonProxy):
-    with closing(
-        await static_proxy.connect_async(server_settings={"IntervalStyle": "iso_8601"})
-    ) as conn:
-        out = await conn.fetchval("select to_json('0 seconds'::interval)")
-        assert out == '"00:00:00"'
-
-    options = "neon_proxy_params_compat:true"
-    with closing(
-        await static_proxy.connect_async(
-            server_settings={"IntervalStyle": "iso_8601", "options": options}
-        )
-    ) as conn:
-        out = await conn.fetchval("select to_json('0 seconds'::interval)")
-        assert out == '"PT0S"'
-
-
 def test_auth_errors(static_proxy: NeonProxy):
    """
    Check that we throw very specific errors in some unsuccessful auth scenarios.
--- a/test_runner/regress/test_timeline_archive.py
+++ b/test_runner/regress/test_timeline_archive.py
@@ -790,6 +790,8 @@ def test_timeline_retain_lsn(
            [
                ".*initial size calculation failed: PageRead.MissingKey.could not find data for key.*",
                ".*page_service_conn_main.*could not find data for key.*",
+                ".*failed to get checkpoint bytes.*",
+                ".*failed get control bytes.*",
            ]
        )
    if offload_child is None or "no-restart" not in offload_child:
--- a/vendor/postgres-v15
+++ b/vendor/postgres-v15
--- a/vendor/postgres-v16
+++ b/vendor/postgres-v16
--- a/vendor/postgres-v17
+++ b/vendor/postgres-v17
--- a/vendor/revisions.json
+++ b/vendor/revisions.json
@@ -1,15 +1,15 @@
 {
  "v17": [
    "17.2",
-    "a10d95be67265e0f10a422ba0457f5a7af01de71"
+    "7864df7b68fa7e0a7b0234b2e5dd2cdb7772aa08"
  ],
  "v16": [
    "16.6",
-    "dff6615a8e48a10bb17a03fa3c00635f1ace7a92"
+    "b6b298e88848f0dbb7d4a077fe70bcd4573ee7ca"
  ],
  "v15": [
    "15.10",
-    "972e325e62b455957adbbdd8580e31275bb5b8c9"
+    "b352942e9c08e5a5350f5c1662c118ce96ea11c5"
  ],
  "v14": [
    "14.15",
Author	SHA1	Message	Date
Konstantin Knizhnik	b4973ecdfa	Add 'failed get control bytes' to list of expected errors	2024-12-05 16:24:31 +02:00
Konstantin Knizhnik	27403e6c3b	Add 'failed to get checkpoint bytes' to list of expected errors	2024-12-05 15:32:02 +02:00
Konstantin Knizhnik	2c10e8a966	RemoveRemove unnecessary reset of redo field	2024-12-05 14:28:34 +02:00
Heikki Linnakangas	0128948d8d	Abuse Checkpoint.redo field to distinguish shutdown checkpoints	2024-12-05 10:04:51 +02:00
Konstantin Knizhnik	7a92454fbe	Reset checkpoiunt redo field after storing it to correctly detect normal shutdown	2024-12-05 09:04:59 +02:00
Konstantin Knizhnik	7fed72e951	Fix compile warnings	2024-12-04 22:03:51 +02:00
Konstantin Knizhnik	19510f3315	Remove calculate_walrecord_end_lsn function	2024-12-04 17:47:49 +02:00
Konstantin Knizhnik	13dd2f11c6	Use redo field to distinguish shutdown andonine checkpoints	2024-12-04 15:33:42 +02:00
Konstantin Knizhnik	4dbb469539	Remove AuxFileStore from test_pgstat.py	2024-12-04 10:19:30 +02:00
Konstantin Knizhnik	3114a9f992	Use new version of test_wal_recevier test	2024-12-04 09:51:48 +02:00
Konstantin Knizhnik	668845fcb7	Yet another attempt to make test_wal_receiver test pass	2024-12-04 09:50:09 +02:00
Konstantin Knizhnik	0f63037345	Increase timeout to 10 seconds	2024-12-04 09:50:09 +02:00
Konstantin Knizhnik	dd826c4c81	Increase wait LSN timeout to 2 seconds	2024-12-04 09:50:09 +02:00
Konstantin Knizhnik	8782018438	Add SIZE_OF_XLOG_RECORD_DATA_HEADER_SHORT	2024-12-04 09:50:09 +02:00
Konstantin Knizhnik	f4196983d2	Undo changes test_import.py	2024-12-04 09:50:09 +02:00
Konstantin Knizhnik	293f22056f	Undo debug trace	2024-12-04 09:50:09 +02:00
Konstantin Knizhnik	cfcb197d85	Add calculate_walrecord_end_lsn function	2024-12-04 09:50:08 +02:00
Konstantin Knizhnik	fa3dc91fb5	Fix is_aux_file_key	2024-12-04 09:50:07 +02:00
Konstantin Knizhnik	0841fb9b7b	Do not import AUX files	2024-12-04 09:48:55 +02:00
Konstantin Knizhnik	b5802abab9	Disable importing AX files	2024-12-04 09:44:49 +02:00
Konstantin Knizhnik	5d12e5a72d	Update pageserver/src/basebackup.rs Co-authored-by: Heikki Linnakangas <heikki@neon.tech>	2024-12-04 09:44:49 +02:00
Konstantin Knizhnik	3179b9bed2	Add test that pgstat information is dropped in case of abnormal termination	2024-12-04 09:44:49 +02:00
Konstantin Knizhnik	e48a8415ff	Add comments	2024-12-04 09:44:49 +02:00
Konstantin Knizhnik	fc041a213b	Update test_runner/regress/test_pgstat.py Co-authored-by: Heikki Linnakangas <heikki@neon.tech>	2024-12-04 09:44:48 +02:00
Konstantin Knizhnik	3890f5846c	Update pageserver/src/basebackup.rs Co-authored-by: Heikki Linnakangas <heikki@neon.tech>	2024-12-04 09:44:48 +02:00
Konstantin Knizhnik	585672f5fc	Update test_runner/regress/test_pgstat.py Co-authored-by: Heikki Linnakangas <heikki@neon.tech>	2024-12-04 09:44:48 +02:00
Konstantin Knizhnik	92ae912ebe	Handle lack of checkpoint in basebackup	2024-12-04 09:44:48 +02:00
Konstantin Knizhnik	953ad21104	Make clippy happy	2024-12-04 09:44:27 +02:00
Konstantin Knizhnik	0fcf20075a	Drop pg_stat data in case of abnormal Postgres termination	2024-12-04 09:43:57 +02:00
Konstantin Knizhnik	c4affc7859	Add neon_pgstat_file_size_limit GUC to limit size of AUX file for pg_stat	2024-12-04 09:43:55 +02:00
Konstantin Knizhnik	368fc2c6c3	Do not import AUX files while initializing fresh timeline from pgdatadir	2024-12-04 09:40:47 +02:00
Konstantin Knizhnik	dfed7029e1	Make clippy happy	2024-12-04 09:40:47 +02:00
Konstantin Knizhnik	6c0e44be4c	Fix bug in AUX v1 delta optimization	2024-12-04 09:40:47 +02:00
Konstantin Knizhnik	dc780c9e1e	Fix bug in including AUX files in basebacklup	2024-12-04 09:40:47 +02:00
Konstantin Knizhnik	45dcc66bf4	Fix python fiormating	2024-12-04 09:40:46 +02:00
Konstantin Knizhnik	d91c5c3523	Add AUX_DIR_PG_STAT	2024-12-04 09:40:46 +02:00
Konstantin Knizhnik	b79e2062bd	Add test for restoring pgstat file	2024-12-04 09:40:46 +02:00
Konstantin Knizhnik	ffc24d6d8d	Persist pg_stat informartion in PS	2024-12-04 09:40:44 +02:00