pageserver: trace slow requests when compact_l0

Signed-off-by: Alex Chi Z <chi@neon.tech> Co-authored-by: Christian Schwarz <christian@neon.tech>
2026-07-08 06:30:37 +00:00 · 2025-01-24 17:33:44 -05:00
57 changed files with 698 additions and 1331 deletions
--- a/.github/actions/neon-project-create/action.yml
+++ b/.github/actions/neon-project-create/action.yml
@@ -17,31 +17,6 @@ inputs:
  compute_units:
    description: '[Min, Max] compute units'
    default: '[1, 1]'
-  # settings below only needed if you want the project to be sharded from the beginning
-  shard_split_project:
-    description: 'by default new projects are not shard-split, specify true to shard-split'
-    required: false
-    default: 'false'
-  admin_api_key:
-    description: 'Admin API Key needed for shard-splitting. Must be specified if shard_split_project is true'
-    required: false
-  shard_count:
-    description: 'Number of shards to split the project into, only applies if shard_split_project is true'
-    required: false
-    default: '8'
-  stripe_size:
-    description: 'Stripe size, optional, in 8kiB pages.  e.g. set 2048 for 16MB stripes. Default is 128 MiB, only applies if shard_split_project is true'
-    required: false
-    default: '32768'
-  psql_path:
-    description: 'Path to psql binary - it is caller responsibility to provision the psql binary'
-    required: false
-    default: '/tmp/neon/pg_install/v16/bin/psql'
-  libpq_lib_path:
-    description: 'Path to directory containing libpq library - it is caller responsibility to provision the libpq library'
-    required: false
-    default: '/tmp/neon/pg_install/v16/lib'
-  

 outputs:
  dsn:
@@ -88,23 +63,6 @@ runs:
        echo "project_id=${project_id}" >> $GITHUB_OUTPUT

        echo "Project ${project_id} has been created"
-
-        if [ "${SHARD_SPLIT_PROJECT}" = "true" ]; then
-          # determine tenant ID
-          TENANT_ID=`${PSQL} ${dsn} -t -A -c "SHOW neon.tenant_id"`
-          
-          echo "Splitting project ${project_id} with tenant_id ${TENANT_ID} into $((SHARD_COUNT)) shards with stripe size $((STRIPE_SIZE))"
-
-          echo "Sending PUT request to https://${API_HOST}/regions/${REGION_ID}/api/v1/admin/storage/proxy/control/v1/tenant/${TENANT_ID}/shard_split"
-          echo "with body {\"new_shard_count\": $((SHARD_COUNT)), \"new_stripe_size\": $((STRIPE_SIZE))}"
-          
-          # we need an ADMIN API KEY to invoke storage controller API for shard splitting (bash -u above checks that the variable is set)
-          curl -X PUT \
-            "https://${API_HOST}/regions/${REGION_ID}/api/v1/admin/storage/proxy/control/v1/tenant/${TENANT_ID}/shard_split" \
-            -H "Accept: application/json" -H "Content-Type: application/json" -H "Authorization: Bearer ${ADMIN_API_KEY}" \
-            -d "{\"new_shard_count\": $SHARD_COUNT, \"new_stripe_size\": $STRIPE_SIZE}"
-        fi
-
      env:
        API_HOST: ${{ inputs.api_host }}
        API_KEY: ${{ inputs.api_key }}
@@ -112,9 +70,3 @@ runs:
        POSTGRES_VERSION: ${{ inputs.postgres_version }}
        MIN_CU: ${{ fromJSON(inputs.compute_units)[0] }}
        MAX_CU: ${{ fromJSON(inputs.compute_units)[1] }}
-        SHARD_SPLIT_PROJECT: ${{ inputs.shard_split_project }}
-        ADMIN_API_KEY: ${{ inputs.admin_api_key }}
-        SHARD_COUNT: ${{ inputs.shard_count }}
-        STRIPE_SIZE: ${{ inputs.stripe_size }}
-        PSQL: ${{ inputs.psql_path }}
-        LD_LIBRARY_PATH: ${{ inputs.libpq_lib_path }}
--- a/.github/workflows/_build-and-test-locally.yml
+++ b/.github/workflows/_build-and-test-locally.yml
@@ -158,6 +158,8 @@ jobs:

      - name: Run cargo build
        run: |
+          PQ_LIB_DIR=$(pwd)/pg_install/v16/lib
+          export PQ_LIB_DIR
          ${cov_prefix} mold -run cargo build $CARGO_FLAGS $CARGO_FEATURES --bins --tests

      # Do install *before* running rust tests because they might recompile the
@@ -215,6 +217,8 @@ jobs:
        env:
          NEXTEST_RETRIES: 3
        run: |
+          PQ_LIB_DIR=$(pwd)/pg_install/v16/lib
+          export PQ_LIB_DIR
          LD_LIBRARY_PATH=$(pwd)/pg_install/v17/lib
          export LD_LIBRARY_PATH

--- a/.github/workflows/build-macos.yml
+++ b/.github/workflows/build-macos.yml
@@ -235,7 +235,7 @@ jobs:
          echo 'CPPFLAGS=-I/usr/local/opt/openssl@3/include' >> $GITHUB_ENV

      - name: Run cargo build (only for v17)
-        run: cargo build --all --release -j$(sysctl -n hw.ncpu)
+        run: PQ_LIB_DIR=$(pwd)/pg_install/v17/lib cargo build --all --release -j$(sysctl -n hw.ncpu)

      - name: Check that no warnings are produced (only for v17)
        run: ./run_clippy.sh
--- a/.github/workflows/ingest_benchmark.yml
+++ b/.github/workflows/ingest_benchmark.yml
@@ -28,24 +28,7 @@ jobs:
    strategy:
      fail-fast: false # allow other variants to continue even if one fails
      matrix:
-        include:
-          - target_project: new_empty_project_stripe_size_2048 
-            stripe_size: 2048 # 16 MiB
-            postgres_version: 16
-          - target_project: new_empty_project_stripe_size_32768
-            stripe_size: 32768 # 256 MiB # note that this is different from null because using null will shard_split the project only if it reaches the threshold
-                               # while here it is sharded from the beginning with a shard size of 256 MiB
-            postgres_version: 16
-          - target_project: new_empty_project
-            stripe_size: null # run with neon defaults which will shard split only when reaching the threshold
-            postgres_version: 16
-          - target_project: new_empty_project
-            stripe_size: null # run with neon defaults which will shard split only when reaching the threshold
-            postgres_version: 17
-          - target_project: large_existing_project
-            stripe_size: null # cannot re-shared or choose different stripe size for existing, already sharded project
-            postgres_version: 16
-      max-parallel: 1 # we want to run each stripe size sequentially to be able to compare the results
+        target_project: [new_empty_project, large_existing_project]
    permissions:
      contents: write
      statuses: write
@@ -84,21 +67,17 @@ jobs:
        aws-oicd-role-arn: ${{ vars.DEV_AWS_OIDC_ROLE_ARN }}

    - name: Create Neon Project
-      if: ${{ startsWith(matrix.target_project, 'new_empty_project') }}
+      if: ${{ matrix.target_project == 'new_empty_project' }}
      id: create-neon-project-ingest-target
      uses: ./.github/actions/neon-project-create
      with:
        region_id: aws-us-east-2
-        postgres_version: ${{ matrix.postgres_version }}
+        postgres_version: 16
        compute_units: '[7, 7]' # we want to test large compute here to avoid compute-side bottleneck
        api_key: ${{ secrets.NEON_STAGING_API_KEY }}
-        shard_split_project: ${{ matrix.stripe_size != null && 'true' || 'false' }}
-        admin_api_key: ${{ secrets.NEON_STAGING_ADMIN_API_KEY }} 
-        shard_count: 8
-        stripe_size: ${{ matrix.stripe_size }}

    - name: Initialize Neon project
-      if: ${{ startsWith(matrix.target_project, 'new_empty_project') }}
+      if: ${{ matrix.target_project == 'new_empty_project' }}
      env:
          BENCHMARK_INGEST_TARGET_CONNSTR: ${{ steps.create-neon-project-ingest-target.outputs.dsn }}
          NEW_PROJECT_ID: ${{ steps.create-neon-project-ingest-target.outputs.project_id }}
@@ -151,7 +130,7 @@ jobs:
        test_selection: performance/test_perf_ingest_using_pgcopydb.py
        run_in_parallel: false
        extra_params: -s -m remote_cluster --timeout 86400 -k test_ingest_performance_using_pgcopydb
-        pg_version: v${{ matrix.postgres_version }}
+        pg_version: v16
        save_perf_report: true
        aws-oicd-role-arn: ${{ vars.DEV_AWS_OIDC_ROLE_ARN }}
      env:
@@ -167,7 +146,7 @@ jobs:
        ${PSQL} "${BENCHMARK_INGEST_TARGET_CONNSTR}" -c "\dt+"

    - name: Delete Neon Project
-      if: ${{ always() && startsWith(matrix.target_project, 'new_empty_project') }}
+      if: ${{ always() && matrix.target_project == 'new_empty_project' }}
      uses: ./.github/actions/neon-project-delete
      with:
        project_id: ${{ steps.create-neon-project-ingest-target.outputs.project_id }}
--- a/.github/workflows/neon_extra_builds.yml
+++ b/.github/workflows/neon_extra_builds.yml
@@ -114,7 +114,7 @@ jobs:
        run: make walproposer-lib -j$(nproc)

      - name: Produce the build stats
-        run: cargo build --all --release --timings -j$(nproc)
+        run: PQ_LIB_DIR=$(pwd)/pg_install/v17/lib cargo build --all --release --timings -j$(nproc)

      - name: Configure AWS credentials
        uses: aws-actions/configure-aws-credentials@v4
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -179,7 +179,7 @@ dependencies = [
 "nom",
 "num-traits",
 "rusticata-macros",
- "thiserror 1.0.69",
+ "thiserror",
 "time",
 ]

@@ -718,14 +718,14 @@ dependencies = [

 [[package]]
 name = "axum"
-version = "0.8.1"
+version = "0.7.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6d6fd624c75e18b3b4c6b9caf42b1afe24437daaee904069137d8bab077be8b8"
+checksum = "edca88bc138befd0323b20752846e6587272d3b03b0343c8ea28a6f819e6e71f"
 dependencies = [
+ "async-trait",
 "axum-core",
 "base64 0.22.1",
 "bytes",
- "form_urlencoded",
 "futures-util",
 "http 1.1.0",
 "http-body 1.0.0",
@@ -733,7 +733,7 @@ dependencies = [
 "hyper 1.4.1",
 "hyper-util",
 "itoa",
- "matchit",
+ "matchit 0.7.0",
 "memchr",
 "mime",
 "percent-encoding",
@@ -746,7 +746,7 @@ dependencies = [
 "sha1",
 "sync_wrapper 1.0.1",
 "tokio",
- "tokio-tungstenite 0.26.1",
+ "tokio-tungstenite 0.24.0",
 "tower 0.5.2",
 "tower-layer",
 "tower-service",
@@ -755,10 +755,11 @@ dependencies = [

 [[package]]
 name = "axum-core"
-version = "0.5.0"
+version = "0.4.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "df1362f362fd16024ae199c1970ce98f9661bf5ef94b9808fee734bc3698b733"
+checksum = "09f2bd6146b97ae3359fa0cc6d6b376d9539582c7b4220f041a33ec24c226199"
 dependencies = [
+ "async-trait",
 "bytes",
 "futures-util",
 "http 1.1.0",
@@ -941,18 +942,6 @@ version = "1.6.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8c3c1a368f70d6cf7302d78f8f7093da241fb8e8807c05cc9e51a125895a6d5b"

-[[package]]
-name = "bb8"
-version = "0.8.6"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d89aabfae550a5c44b43ab941844ffcd2e993cb6900b342debf59e9ea74acdb8"
-dependencies = [
- "async-trait",
- "futures-util",
- "parking_lot 0.12.1",
- "tokio",
-]
-
 [[package]]
 name = "bcder"
 version = "0.7.4"
@@ -1129,7 +1118,7 @@ dependencies = [
 "log",
 "nix 0.25.1",
 "regex",
- "thiserror 1.0.69",
+ "thiserror",
 ]

 [[package]]
@@ -1310,9 +1299,9 @@ dependencies = [
 "serde_with",
 "signal-hook",
 "tar",
- "thiserror 1.0.69",
+ "thiserror",
 "tokio",
- "tokio-postgres 0.7.7",
+ "tokio-postgres",
 "tokio-stream",
 "tokio-util",
 "tower 0.5.2",
@@ -1419,9 +1408,9 @@ dependencies = [
 "serde",
 "serde_json",
 "storage_broker",
- "thiserror 1.0.69",
+ "thiserror",
 "tokio",
- "tokio-postgres 0.7.7",
+ "tokio-postgres",
 "tokio-util",
 "toml",
 "toml_edit",
@@ -1797,24 +1786,11 @@ dependencies = [
 "chrono",
 "diesel_derives",
 "itoa",
+ "pq-sys",
+ "r2d2",
 "serde_json",
 ]

-[[package]]
-name = "diesel-async"
-version = "0.5.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "51a307ac00f7c23f526a04a77761a0519b9f0eb2838ebf5b905a58580095bdcb"
-dependencies = [
- "async-trait",
- "bb8",
- "diesel",
- "futures-util",
- "scoped-futures",
- "tokio",
- "tokio-postgres 0.7.12",
-]
-
 [[package]]
 name = "diesel_derives"
 version = "2.2.1"
@@ -2263,7 +2239,7 @@ dependencies = [
 "pin-project",
 "rand 0.8.5",
 "sha1",
- "thiserror 1.0.69",
+ "thiserror",
 "tokio",
 "tokio-util",
 ]
@@ -3389,6 +3365,12 @@ dependencies = [
 "regex-automata 0.1.10",
 ]

+[[package]]
+name = "matchit"
+version = "0.7.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b87248edafb776e59e6ee64a79086f65890d3510f2c656c000bf2a7e8a0aea40"
+
 [[package]]
 name = "matchit"
 version = "0.8.4"
@@ -3779,7 +3761,7 @@ dependencies = [
 "serde_json",
 "serde_path_to_error",
 "sha2",
- "thiserror 1.0.69",
+ "thiserror",
 "url",
 ]

@@ -3829,7 +3811,7 @@ dependencies = [
 "futures-sink",
 "js-sys",
 "pin-project-lite",
- "thiserror 1.0.69",
+ "thiserror",
 "tracing",
 ]

@@ -3861,7 +3843,7 @@ dependencies = [
 "opentelemetry_sdk",
 "prost",
 "reqwest",
- "thiserror 1.0.69",
+ "thiserror",
 ]

 [[package]]
@@ -3897,7 +3879,7 @@ dependencies = [
 "percent-encoding",
 "rand 0.8.5",
 "serde_json",
- "thiserror 1.0.69",
+ "thiserror",
 "tokio",
 "tokio-stream",
 "tracing",
@@ -4011,7 +3993,7 @@ dependencies = [
 "remote_storage",
 "serde_json",
 "svg_fmt",
- "thiserror 1.0.69",
+ "thiserror",
 "tokio",
 "tokio-util",
 "utils",
@@ -4060,8 +4042,8 @@ dependencies = [
 "pageserver_compaction",
 "pin-project-lite",
 "postgres",
- "postgres-protocol 0.6.4",
- "postgres-types 0.2.4",
+ "postgres-protocol",
+ "postgres-types",
 "postgres_backend",
 "postgres_connection",
 "postgres_ffi",
@@ -4087,12 +4069,12 @@ dependencies = [
 "strum_macros",
 "sysinfo",
 "tenant_size_model",
- "thiserror 1.0.69",
+ "thiserror",
 "tikv-jemallocator",
 "tokio",
 "tokio-epoll-uring",
 "tokio-io-timeout",
- "tokio-postgres 0.7.7",
+ "tokio-postgres",
 "tokio-stream",
 "tokio-tar",
 "tokio-util",
@@ -4133,7 +4115,7 @@ dependencies = [
 "storage_broker",
 "strum",
 "strum_macros",
- "thiserror 1.0.69",
+ "thiserror",
 "utils",
 ]

@@ -4148,9 +4130,9 @@ dependencies = [
 "postgres",
 "reqwest",
 "serde",
- "thiserror 1.0.69",
+ "thiserror",
 "tokio",
- "tokio-postgres 0.7.7",
+ "tokio-postgres",
 "tokio-stream",
 "tokio-util",
 "utils",
@@ -4456,7 +4438,7 @@ dependencies = [
 "futures-util",
 "log",
 "tokio",
- "tokio-postgres 0.7.7",
+ "tokio-postgres",
 ]

 [[package]]
@@ -4477,24 +4459,6 @@ dependencies = [
 "stringprep",
 ]

-[[package]]
-name = "postgres-protocol"
-version = "0.6.7"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "acda0ebdebc28befa84bee35e651e4c5f09073d668c7aed4cf7e23c3cda84b23"
-dependencies = [
- "base64 0.22.1",
- "byteorder",
- "bytes",
- "fallible-iterator",
- "hmac",
- "md-5",
- "memchr",
- "rand 0.8.5",
- "sha2",
- "stringprep",
-]
-
 [[package]]
 name = "postgres-protocol2"
 version = "0.1.0"
@@ -4518,18 +4482,7 @@ source = "git+https://github.com/neondatabase/rust-postgres.git?branch=neon#511f
 dependencies = [
 "bytes",
 "fallible-iterator",
- "postgres-protocol 0.6.4",
-]
-
-[[package]]
-name = "postgres-types"
-version = "0.2.8"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f66ea23a2d0e5734297357705193335e0a957696f34bed2f2faefacb2fec336f"
-dependencies = [
- "bytes",
- "fallible-iterator",
- "postgres-protocol 0.6.7",
+ "postgres-protocol",
 ]

 [[package]]
@@ -4552,9 +4505,9 @@ dependencies = [
 "rustls 0.23.18",
 "rustls-pemfile 2.1.1",
 "serde",
- "thiserror 1.0.69",
+ "thiserror",
 "tokio",
- "tokio-postgres 0.7.7",
+ "tokio-postgres",
 "tokio-postgres-rustls",
 "tokio-rustls 0.26.0",
 "tokio-util",
@@ -4569,7 +4522,7 @@ dependencies = [
 "itertools 0.10.5",
 "once_cell",
 "postgres",
- "tokio-postgres 0.7.7",
+ "tokio-postgres",
 "url",
 ]

@@ -4590,7 +4543,7 @@ dependencies = [
 "pprof",
 "regex",
 "serde",
- "thiserror 1.0.69",
+ "thiserror",
 "tracing",
 "utils",
 ]
@@ -4601,7 +4554,7 @@ version = "0.1.0"
 dependencies = [
 "anyhow",
 "camino",
- "thiserror 1.0.69",
+ "thiserror",
 "tokio",
 "workspace_hack",
 ]
@@ -4634,7 +4587,7 @@ dependencies = [
 "smallvec",
 "symbolic-demangle",
 "tempfile",
- "thiserror 1.0.69",
+ "thiserror",
 ]

 [[package]]
@@ -4656,6 +4609,15 @@ version = "0.2.17"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "5b40af805b3121feab8a3c29f04d8ad262fa8e0561883e7653e024ae4479e6de"

+[[package]]
+name = "pq-sys"
+version = "0.6.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f6cc05d7ea95200187117196eee9edd0644424911821aeb28a18ce60ea0b8793"
+dependencies = [
+ "vcpkg",
+]
+
 [[package]]
 name = "pq_proto"
 version = "0.1.0"
@@ -4663,10 +4625,10 @@ dependencies = [
 "byteorder",
 "bytes",
 "itertools 0.10.5",
- "postgres-protocol 0.6.4",
+ "postgres-protocol",
 "rand 0.8.5",
 "serde",
- "thiserror 1.0.69",
+ "thiserror",
 "tokio",
 ]

@@ -4737,7 +4699,7 @@ dependencies = [
 "memchr",
 "parking_lot 0.12.1",
 "procfs",
- "thiserror 1.0.69",
+ "thiserror",
 ]

 [[package]]
@@ -4907,11 +4869,11 @@ dependencies = [
 "strum",
 "strum_macros",
 "subtle",
- "thiserror 1.0.69",
+ "thiserror",
 "tikv-jemalloc-ctl",
 "tikv-jemallocator",
 "tokio",
- "tokio-postgres 0.7.7",
+ "tokio-postgres",
 "tokio-postgres2",
 "tokio-rustls 0.26.0",
 "tokio-tungstenite 0.21.0",
@@ -4968,6 +4930,17 @@ dependencies = [
 "proc-macro2",
 ]

+[[package]]
+name = "r2d2"
+version = "0.8.10"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "51de85fb3fb6524929c8a2eb85e6b6d363de4e8c48f9e2c2eac4944abc181c93"
+dependencies = [
+ "log",
+ "parking_lot 0.12.1",
+ "scheduled-thread-pool",
+]
+
 [[package]]
 name = "rand"
 version = "0.7.3"
@@ -5304,7 +5277,7 @@ dependencies = [
 "http 1.1.0",
 "reqwest",
 "serde",
- "thiserror 1.0.69",
+ "thiserror",
 "tower-service",
 ]

@@ -5324,7 +5297,7 @@ dependencies = [
 "reqwest",
 "reqwest-middleware",
 "retry-policies",
- "thiserror 1.0.69",
+ "thiserror",
 "tokio",
 "tracing",
 "wasm-timer",
@@ -5340,7 +5313,7 @@ dependencies = [
 "async-trait",
 "getrandom 0.2.11",
 "http 1.1.0",
- "matchit",
+ "matchit 0.8.4",
 "opentelemetry",
 "reqwest",
 "reqwest-middleware",
@@ -5699,7 +5672,7 @@ dependencies = [
 "pageserver_api",
 "parking_lot 0.12.1",
 "postgres",
- "postgres-protocol 0.6.4",
+ "postgres-protocol",
 "postgres_backend",
 "postgres_ffi",
 "pprof",
@@ -5719,11 +5692,11 @@ dependencies = [
 "storage_broker",
 "strum",
 "strum_macros",
- "thiserror 1.0.69",
+ "thiserror",
 "tikv-jemallocator",
 "tokio",
 "tokio-io-timeout",
- "tokio-postgres 0.7.7",
+ "tokio-postgres",
 "tokio-stream",
 "tokio-tar",
 "tokio-util",
@@ -5758,7 +5731,7 @@ dependencies = [
 "reqwest",
 "safekeeper_api",
 "serde",
- "thiserror 1.0.69",
+ "thiserror",
 "utils",
 "workspace_hack",
 ]
@@ -5782,12 +5755,12 @@ dependencies = [
 ]

 [[package]]
-name = "scoped-futures"
-version = "0.1.4"
+name = "scheduled-thread-pool"
+version = "0.2.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1b24aae2d0636530f359e9d5ef0c04669d11c5e756699b27a6a6d845d8329091"
+checksum = "3cbc66816425a074528352f5789333ecff06ca41b36b0b0efdfbb29edc391a19"
 dependencies = [
- "pin-project-lite",
+ "parking_lot 0.12.1",
 ]

 [[package]]
@@ -5967,7 +5940,7 @@ dependencies = [
 "rand 0.8.5",
 "serde",
 "serde_json",
- "thiserror 1.0.69",
+ "thiserror",
 "time",
 "url",
 "uuid",
@@ -6039,7 +6012,7 @@ checksum = "c7715380eec75f029a4ef7de39a9200e0a63823176b759d055b613f5a87df6a6"
 dependencies = [
 "percent-encoding",
 "serde",
- "thiserror 1.0.69",
+ "thiserror",
 ]

 [[package]]
@@ -6201,7 +6174,7 @@ checksum = "adc4e5204eb1910f40f9cfa375f6f05b68c3abac4b6fd879c8ff5e7ae8a0a085"
 dependencies = [
 "num-bigint",
 "num-traits",
- "thiserror 1.0.69",
+ "thiserror",
 "time",
 ]

@@ -6322,7 +6295,6 @@ dependencies = [
 "clap",
 "control_plane",
 "diesel",
- "diesel-async",
 "diesel_migrations",
 "fail",
 "futures",
@@ -6337,16 +6309,16 @@ dependencies = [
 "pageserver_api",
 "pageserver_client",
 "postgres_connection",
+ "r2d2",
 "rand 0.8.5",
 "reqwest",
 "routerify",
- "scoped-futures",
 "scopeguard",
 "serde",
 "serde_json",
 "strum",
 "strum_macros",
- "thiserror 1.0.69",
+ "thiserror",
 "tokio",
 "tokio-util",
 "tracing",
@@ -6393,7 +6365,7 @@ dependencies = [
 "serde_json",
 "storage_controller_client",
 "tokio",
- "tokio-postgres 0.7.7",
+ "tokio-postgres",
 "tokio-postgres-rustls",
 "tokio-stream",
 "tokio-util",
@@ -6638,16 +6610,7 @@ version = "1.0.69"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b6aaf5339b578ea85b50e080feb250a3e8ae8cfcdff9a461c9ec2904bc923f52"
 dependencies = [
- "thiserror-impl 1.0.69",
-]
-
-[[package]]
-name = "thiserror"
-version = "2.0.11"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d452f284b73e6d76dd36758a0c8684b1d5be31f92b89d07fd5822175732206fc"
-dependencies = [
- "thiserror-impl 2.0.11",
+ "thiserror-impl",
 ]

 [[package]]
@@ -6661,17 +6624,6 @@ dependencies = [
 "syn 2.0.90",
 ]

-[[package]]
-name = "thiserror-impl"
-version = "2.0.11"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "26afc1baea8a989337eeb52b6e72a039780ce45c3edfcc9c5b9d112feeb173c2"
-dependencies = [
- "proc-macro2",
- "quote",
- "syn 2.0.90",
-]
-
 [[package]]
 name = "thread_local"
 version = "1.1.7"
@@ -6828,7 +6780,7 @@ dependencies = [
 "nix 0.26.4",
 "once_cell",
 "scopeguard",
- "thiserror 1.0.69",
+ "thiserror",
 "tokio",
 "tokio-util",
 "tracing",
@@ -6872,39 +6824,13 @@ dependencies = [
 "percent-encoding",
 "phf",
 "pin-project-lite",
- "postgres-protocol 0.6.4",
- "postgres-types 0.2.4",
+ "postgres-protocol",
+ "postgres-types",
 "socket2",
 "tokio",
 "tokio-util",
 ]

-[[package]]
-name = "tokio-postgres"
-version = "0.7.12"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3b5d3742945bc7d7f210693b0c58ae542c6fd47b17adbbda0885f3dcb34a6bdb"
-dependencies = [
- "async-trait",
- "byteorder",
- "bytes",
- "fallible-iterator",
- "futures-channel",
- "futures-util",
- "log",
- "parking_lot 0.12.1",
- "percent-encoding",
- "phf",
- "pin-project-lite",
- "postgres-protocol 0.6.7",
- "postgres-types 0.2.8",
- "rand 0.8.5",
- "socket2",
- "tokio",
- "tokio-util",
- "whoami",
-]
-
 [[package]]
 name = "tokio-postgres-rustls"
 version = "0.12.0"
@@ -6914,7 +6840,7 @@ dependencies = [
 "ring",
 "rustls 0.23.18",
 "tokio",
- "tokio-postgres 0.7.7",
+ "tokio-postgres",
 "tokio-rustls 0.26.0",
 "x509-certificate",
 ]
@@ -7011,14 +6937,14 @@ dependencies = [

 [[package]]
 name = "tokio-tungstenite"
-version = "0.26.1"
+version = "0.24.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "be4bf6fecd69fcdede0ec680aaf474cdab988f9de6bc73d3758f0160e3b7025a"
+checksum = "edc5f74e248dc973e0dbb7b74c7e0d6fcc301c694ff50049504004ef4d0cdcd9"
 dependencies = [
 "futures-util",
 "log",
 "tokio",
- "tungstenite 0.26.1",
+ "tungstenite 0.24.0",
 ]

 [[package]]
@@ -7328,16 +7254,16 @@ dependencies = [
 "log",
 "rand 0.8.5",
 "sha1",
- "thiserror 1.0.69",
+ "thiserror",
 "url",
 "utf-8",
 ]

 [[package]]
 name = "tungstenite"
-version = "0.26.1"
+version = "0.24.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "413083a99c579593656008130e29255e54dcaae495be556cc26888f211648c24"
+checksum = "18e5b8366ee7a95b16d32197d0b2604b43a0be89dc5fac9f8e96ccafbaedda8a"
 dependencies = [
 "byteorder",
 "bytes",
@@ -7347,7 +7273,7 @@ dependencies = [
 "log",
 "rand 0.8.5",
 "sha1",
- "thiserror 2.0.11",
+ "thiserror",
 "utf-8",
 ]

@@ -7542,7 +7468,7 @@ dependencies = [
 "signal-hook",
 "strum",
 "strum_macros",
- "thiserror 1.0.69",
+ "thiserror",
 "tokio",
 "tokio-stream",
 "tokio-tar",
@@ -7572,6 +7498,12 @@ version = "0.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "830b7e5d4d90034032940e4ace0d9a9a057e7a45cd94e6c007832e39edb82f6d"

+[[package]]
+name = "vcpkg"
+version = "0.2.15"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "accd4ea62f7bb7a82fe23066fb0957d48ef677f6eeb8215f372f52e48bb32426"
+
 [[package]]
 name = "version_check"
 version = "0.9.4"
@@ -7591,7 +7523,7 @@ dependencies = [
 "serde_json",
 "sysinfo",
 "tokio",
- "tokio-postgres 0.7.7",
+ "tokio-postgres",
 "tokio-util",
 "tracing",
 "tracing-subscriber",
@@ -7642,7 +7574,7 @@ dependencies = [
 "remote_storage",
 "serde",
 "serde_json",
- "thiserror 1.0.69",
+ "thiserror",
 "tikv-jemallocator",
 "tokio",
 "tokio-util",
@@ -8171,7 +8103,7 @@ dependencies = [
 "ring",
 "signature 2.2.0",
 "spki 0.7.3",
- "thiserror 1.0.69",
+ "thiserror",
 "zeroize",
 ]

@@ -8188,7 +8120,7 @@ dependencies = [
 "nom",
 "oid-registry",
 "rusticata-macros",
- "thiserror 1.0.69",
+ "thiserror",
 "time",
 ]

--- a/Cargo.toml
+++ b/Cargo.toml
@@ -65,7 +65,7 @@ aws-smithy-types = "1.2"
 aws-credential-types = "1.2.0"
 aws-sigv4 = { version = "1.2", features = ["sign-http"] }
 aws-types = "1.3"
-axum = { version = "0.8.1", features = ["ws"] }
+axum = { version = "0.7.9", features = ["ws"] }
 base64 = "0.13.0"
 bincode = "1.3"
 bindgen = "0.70"
--- a/2
+++ b/2
@@ -45,7 +45,7 @@ COPY --chown=nonroot . .

 ARG ADDITIONAL_RUSTFLAGS
 RUN set -e \
-    && RUSTFLAGS="-Clinker=clang -Clink-arg=-fuse-ld=mold -Clink-arg=-Wl,--no-rosegment -Cforce-frame-pointers=yes ${ADDITIONAL_RUSTFLAGS}" cargo build \
+    && PQ_LIB_DIR=$(pwd)/pg_install/v${STABLE_PG_VERSION}/lib RUSTFLAGS="-Clinker=clang -Clink-arg=-fuse-ld=mold -Clink-arg=-Wl,--no-rosegment -Cforce-frame-pointers=yes ${ADDITIONAL_RUSTFLAGS}" cargo build \
      --bin pg_sni_router  \
      --bin pageserver  \
      --bin pagectl  \
--- a/2
+++ b/2
@@ -64,6 +64,8 @@ CARGO_BUILD_FLAGS += $(filter -j1,$(MAKEFLAGS))
 CARGO_CMD_PREFIX += $(if $(filter n,$(MAKEFLAGS)),,+)
 # Force cargo not to print progress bar
 CARGO_CMD_PREFIX += CARGO_TERM_PROGRESS_WHEN=never CI=1
+# Set PQ_LIB_DIR to make sure `storage_controller` get linked with bundled libpq (through diesel)
+CARGO_CMD_PREFIX += PQ_LIB_DIR=$(POSTGRES_INSTALL_DIR)/v16/lib

 CACHEDIR_TAG_CONTENTS := "Signature: 8a477f597d28d172789f06886806bc55"

--- a/README.md
+++ b/README.md
@@ -21,10 +21,8 @@ The Neon storage engine consists of two major components:

 See developer documentation in [SUMMARY.md](/docs/SUMMARY.md) for more information.

-## Running a local development environment
+## Running local installation

-Neon can be run on a workstation for small experiments and to test code changes, by
-following these instructions.

 #### Installing dependencies on Linux
 1. Install build dependencies and other applicable packages
@@ -240,7 +238,7 @@ postgres=# select * from t;
 > cargo neon stop
 ```

-More advanced usages can be found at [Local Development Control Plane (`neon_local`))](./control_plane/README.md).
+More advanced usages can be found at [Control Plane and Neon Local](./control_plane/README.md).

 #### Handling build failures

--- a/compute/compute-node.Dockerfile
+++ b/compute/compute-node.Dockerfile
@@ -1266,12 +1266,11 @@ RUN set -e \

 #########################################################################################
 #
-# Layers "postgres-exporter", "pgbouncer-exporter", and "sql-exporter"
+# Layers "postgres-exporter" and "sql-exporter"
 #
 #########################################################################################

 FROM quay.io/prometheuscommunity/postgres-exporter:v0.16.0 AS postgres-exporter
-FROM quay.io/prometheuscommunity/pgbouncer-exporter:v0.10.2 AS pgbouncer-exporter

 # Keep the version the same as in build-tools.Dockerfile and
 # test_runner/regress/test_compute_metrics.py.
@@ -1403,7 +1402,6 @@ RUN mkdir -p /etc/local_proxy && chown postgres:postgres /etc/local_proxy

 # Metrics exporter binaries and  configuration files
 COPY --from=postgres-exporter /bin/postgres_exporter /bin/postgres_exporter
-COPY --from=pgbouncer-exporter /bin/pgbouncer_exporter /bin/pgbouncer_exporter
 COPY --from=sql-exporter      /bin/sql_exporter      /bin/sql_exporter

 COPY --chown=postgres compute/etc/postgres_exporter.yml /etc/postgres_exporter.yml
--- a/compute/etc/pgbouncer.ini
+++ b/compute/etc/pgbouncer.ini
@@ -19,8 +19,6 @@ max_prepared_statements=0
 admin_users=postgres
 unix_socket_dir=/tmp/
 unix_socket_mode=0777
-; required for pgbouncer_exporter
-ignore_startup_parameters=extra_float_digits

 ;; Disable connection logging. It produces a lot of logs that no one looks at,
 ;; and we can get similar log entries from the proxy too. We had incidents in
--- a/compute/vm-image-spec-bookworm.yaml
+++ b/compute/vm-image-spec-bookworm.yaml
@@ -27,10 +27,6 @@ commands:
    user: nobody
    sysvInitAction: respawn
    shell: 'DATA_SOURCE_NAME="user=cloud_admin sslmode=disable dbname=postgres application_name=postgres-exporter" /bin/postgres_exporter --config.file=/etc/postgres_exporter.yml'
-  - name: pgbouncer-exporter
-    user: postgres
-    sysvInitAction: respawn
-    shell: '/bin/pgbouncer_exporter --pgBouncer.connectionString="postgres:///pgbouncer?host=/tmp&port=6432&dbname=pgbouncer&user=pgbouncer"'
  - name: sql-exporter
    user: nobody
    sysvInitAction: respawn
--- a/compute/vm-image-spec-bullseye.yaml
+++ b/compute/vm-image-spec-bullseye.yaml
@@ -27,10 +27,6 @@ commands:
    user: nobody
    sysvInitAction: respawn
    shell: 'DATA_SOURCE_NAME="user=cloud_admin sslmode=disable dbname=postgres application_name=postgres-exporter" /bin/postgres_exporter --config.file=/etc/postgres_exporter.yml'
-  - name: pgbouncer-exporter
-    user: postgres
-    sysvInitAction: respawn
-    shell: '/bin/pgbouncer_exporter --pgBouncer.connectionString="postgres:///pgbouncer?host=/tmp&port=6432&dbname=pgbouncer&user=pgbouncer"'
  - name: sql-exporter
    user: nobody
    sysvInitAction: respawn
--- a/compute_tools/src/http/extract/json.rs
+++ b/compute_tools/src/http/extract/json.rs
@@ -1,6 +1,9 @@
 use std::ops::{Deref, DerefMut};

-use axum::extract::{rejection::JsonRejection, FromRequest, Request};
+use axum::{
+    async_trait,
+    extract::{rejection::JsonRejection, FromRequest, Request},
+};
 use compute_api::responses::GenericAPIError;
 use http::StatusCode;

@@ -9,6 +12,7 @@ use http::StatusCode;
 #[derive(Debug, Clone, Copy, Default)]
 pub(crate) struct Json<T>(pub T);

+#[async_trait]
 impl<S, T> FromRequest<S> for Json<T>
 where
    axum::Json<T>: FromRequest<S, Rejection = JsonRejection>,
--- a/compute_tools/src/http/extract/path.rs
+++ b/compute_tools/src/http/extract/path.rs
@@ -1,6 +1,9 @@
 use std::ops::{Deref, DerefMut};

-use axum::extract::{rejection::PathRejection, FromRequestParts};
+use axum::{
+    async_trait,
+    extract::{rejection::PathRejection, FromRequestParts},
+};
 use compute_api::responses::GenericAPIError;
 use http::{request::Parts, StatusCode};

@@ -9,6 +12,7 @@ use http::{request::Parts, StatusCode};
 #[derive(Debug, Clone, Copy, Default)]
 pub(crate) struct Path<T>(pub T);

+#[async_trait]
 impl<S, T> FromRequestParts<S> for Path<T>
 where
    axum::extract::Path<T>: FromRequestParts<S, Rejection = PathRejection>,
--- a/compute_tools/src/http/extract/query.rs
+++ b/compute_tools/src/http/extract/query.rs
@@ -1,6 +1,9 @@
 use std::ops::{Deref, DerefMut};

-use axum::extract::{rejection::QueryRejection, FromRequestParts};
+use axum::{
+    async_trait,
+    extract::{rejection::QueryRejection, FromRequestParts},
+};
 use compute_api::responses::GenericAPIError;
 use http::{request::Parts, StatusCode};

@@ -9,6 +12,7 @@ use http::{request::Parts, StatusCode};
 #[derive(Debug, Clone, Copy, Default)]
 pub(crate) struct Query<T>(pub T);

+#[async_trait]
 impl<S, T> FromRequestParts<S> for Query<T>
 where
    axum::extract::Query<T>: FromRequestParts<S, Rejection = QueryRejection>,
--- a/compute_tools/src/http/server.rs
+++ b/compute_tools/src/http/server.rs
@@ -55,7 +55,7 @@ async fn serve(port: u16, compute: Arc<ComputeNode>) {
        .route("/database_schema", get(database_schema::get_schema_dump))
        .route("/dbs_and_roles", get(dbs_and_roles::get_catalog_objects))
        .route(
-            "/extension_server/{*filename}",
+            "/extension_server/*filename",
            post(extension_server::download_extension),
        )
        .route("/extensions", post(extensions::install_extension))
--- a/control_plane/README.md
+++ b/control_plane/README.md
@@ -1,10 +1,6 @@
-# Local Development Control Plane (`neon_local`)
+# Control Plane and Neon Local

-This crate contains tools to start a Neon development environment locally. This utility can be used with the `cargo neon` command.  This is a convenience to invoke
-the `neon_local` binary.
-
-**Note**: this is a dev/test tool -- a minimal control plane suitable for testing
-code changes locally, but not suitable for running production systems.
+This crate contains tools to start a Neon development environment locally. This utility can be used with the `cargo neon` command.

 ## Example: Start with Postgres 16

--- a/control_plane/src/pageserver.rs
+++ b/control_plane/src/pageserver.rs
@@ -357,11 +357,6 @@ impl PageServerNode {
                .map(|x| x.parse::<usize>())
                .transpose()
                .context("Failed to parse 'l0_flush_delay_threshold' as an integer")?,
-            l0_flush_wait_upload: settings
-                .remove("l0_flush_wait_upload")
-                .map(|x| x.parse::<bool>())
-                .transpose()
-                .context("Failed to parse 'l0_flush_wait_upload' as a boolean")?,
            l0_flush_stall_threshold: settings
                .remove("l0_flush_stall_threshold")
                .map(|x| x.parse::<usize>())
--- a/deny.toml
+++ b/deny.toml
@@ -41,8 +41,8 @@ allow = [
    "MIT",
    "MPL-2.0",
    "OpenSSL",
+    "Unicode-DFS-2016",
    "Unicode-3.0",
-    "Zlib",
 ]
 confidence-threshold = 0.8
 exceptions = [
--- a/libs/pageserver_api/src/config.rs
+++ b/libs/pageserver_api/src/config.rs
@@ -260,15 +260,12 @@ pub struct TenantConfigToml {
    /// Level0 delta layer threshold at which to delay layer flushes for compaction backpressure,
    /// such that they take 2x as long, and start waiting for layer flushes during ephemeral layer
    /// rolls. This helps compaction keep up with WAL ingestion, and avoids read amplification
-    /// blowing up. Should be >compaction_threshold. 0 to disable. Disabled by default.
+    /// blowing up. Should be >compaction_threshold. If None, defaults to 2 * compaction_threshold.
+    /// 0 to disable.
    pub l0_flush_delay_threshold: Option<usize>,
-    /// Level0 delta layer threshold at which to stall layer flushes. Must be >compaction_threshold
-    /// to avoid deadlock. 0 to disable. Disabled by default.
+    /// Level0 delta layer threshold at which to stall layer flushes. 0 to disable. If None,
+    /// defaults to 4 * compaction_threshold. Must be >compaction_threshold to avoid deadlock.
    pub l0_flush_stall_threshold: Option<usize>,
-    /// If true, Level0 delta layer flushes will wait for S3 upload before flushing the next
-    /// layer. This is a temporary backpressure mechanism which should be removed once
-    /// l0_flush_{delay,stall}_threshold is fully enabled.
-    pub l0_flush_wait_upload: bool,
    // Determines how much history is retained, to allow
    // branching and read replicas at an older point in time.
    // The unit is #of bytes of WAL.
@@ -526,8 +523,6 @@ pub mod tenant_conf_defaults {
    pub const DEFAULT_COMPACTION_ALGORITHM: crate::models::CompactionAlgorithm =
        crate::models::CompactionAlgorithm::Legacy;

-    pub const DEFAULT_L0_FLUSH_WAIT_UPLOAD: bool = true;
-
    pub const DEFAULT_GC_HORIZON: u64 = 64 * 1024 * 1024;

    // Large DEFAULT_GC_PERIOD is fine as long as PITR_INTERVAL is larger.
@@ -568,7 +563,6 @@ impl Default for TenantConfigToml {
            },
            l0_flush_delay_threshold: None,
            l0_flush_stall_threshold: None,
-            l0_flush_wait_upload: DEFAULT_L0_FLUSH_WAIT_UPLOAD,
            gc_horizon: DEFAULT_GC_HORIZON,
            gc_period: humantime::parse_duration(DEFAULT_GC_PERIOD)
                .expect("cannot parse default gc period"),
--- a/libs/pageserver_api/src/key.rs
+++ b/libs/pageserver_api/src/key.rs
@@ -464,18 +464,6 @@ pub fn rel_size_to_key(rel: RelTag) -> Key {
    }
 }

-#[inline(always)]
-pub fn rel_size_key_to_rel(key: Key) -> RelTag {
-    assert_eq!(key.field1, 0x00);
-    assert_eq!(key.field6, 0xffff_ffff);
-    RelTag {
-        forknum: key.field5,
-        spcnode: key.field2,
-        dbnode: key.field3,
-        relnode: key.field4,
-    }
-}
-
 impl Key {
    #[inline(always)]
    pub fn is_rel_size_key(&self) -> bool {
@@ -571,15 +559,6 @@ pub fn slru_segment_size_to_key(kind: SlruKind, segno: u32) -> Key {
    }
 }

-#[inline(always)]
-pub fn slru_segment_size_key_to_segno(key: Key) -> u32 {
-    assert_eq!(key.field1, 0x01);
-    assert_eq!(key.field3, 1);
-    assert_eq!(key.field5, 0);
-    assert_eq!(key.field6, 0xffff_ffff);
-    key.field4
-}
-
 impl Key {
    pub fn is_slru_segment_size_key(&self) -> bool {
        self.field1 == 0x01
--- a/libs/pageserver_api/src/models.rs
+++ b/libs/pageserver_api/src/models.rs
@@ -466,8 +466,6 @@ pub struct TenantConfigPatch {
    #[serde(skip_serializing_if = "FieldPatch::is_noop")]
    pub l0_flush_stall_threshold: FieldPatch<usize>,
    #[serde(skip_serializing_if = "FieldPatch::is_noop")]
-    pub l0_flush_wait_upload: FieldPatch<bool>,
-    #[serde(skip_serializing_if = "FieldPatch::is_noop")]
    pub gc_horizon: FieldPatch<u64>,
    #[serde(skip_serializing_if = "FieldPatch::is_noop")]
    pub gc_period: FieldPatch<String>,
@@ -526,7 +524,6 @@ pub struct TenantConfig {
    pub compaction_algorithm: Option<CompactionAlgorithmSettings>,
    pub l0_flush_delay_threshold: Option<usize>,
    pub l0_flush_stall_threshold: Option<usize>,
-    pub l0_flush_wait_upload: Option<bool>,
    pub gc_horizon: Option<u64>,
    pub gc_period: Option<String>,
    pub image_creation_threshold: Option<usize>,
@@ -562,7 +559,6 @@ impl TenantConfig {
            mut compaction_algorithm,
            mut l0_flush_delay_threshold,
            mut l0_flush_stall_threshold,
-            mut l0_flush_wait_upload,
            mut gc_horizon,
            mut gc_period,
            mut image_creation_threshold,
@@ -601,7 +597,6 @@ impl TenantConfig {
        patch
            .l0_flush_stall_threshold
            .apply(&mut l0_flush_stall_threshold);
-        patch.l0_flush_wait_upload.apply(&mut l0_flush_wait_upload);
        patch.gc_horizon.apply(&mut gc_horizon);
        patch.gc_period.apply(&mut gc_period);
        patch
@@ -656,7 +651,6 @@ impl TenantConfig {
            compaction_algorithm,
            l0_flush_delay_threshold,
            l0_flush_stall_threshold,
-            l0_flush_wait_upload,
            gc_horizon,
            gc_period,
            image_creation_threshold,
@@ -1021,13 +1015,6 @@ pub struct TenantConfigPatchRequest {
    pub config: TenantConfigPatch, // as we have a flattened field, we should reject all unknown fields in it
 }

-#[derive(Serialize, Deserialize, Debug)]
-pub struct TenantWaitLsnRequest {
-    #[serde(flatten)]
-    pub timelines: HashMap<TimelineId, Lsn>,
-    pub timeout: Duration,
-}
-
 /// See [`TenantState::attachment_status`] and the OpenAPI docs for context.
 #[derive(Serialize, Deserialize, Clone)]
 #[serde(tag = "slug", content = "data", rename_all = "snake_case")]
--- a/libs/remote_storage/src/azure_blob.rs
+++ b/libs/remote_storage/src/azure_blob.rs
@@ -377,8 +377,7 @@ impl RemoteStorage for AzureBlobStorage {

                let next_item = next_item?;

-                // Log a warning if we saw two timeouts in a row before a successful request
-                if timeout_try_cnt > 2 {
+                if timeout_try_cnt >= 2 {
                    tracing::warn!("Azure Blob Storage list timed out and succeeded after {} tries", timeout_try_cnt);
                }
                timeout_try_cnt = 1;
--- a/libs/vm_monitor/src/dispatcher.rs
+++ b/libs/vm_monitor/src/dispatcher.rs
@@ -7,7 +7,7 @@
 //! (notifying it of upscale).

 use anyhow::{bail, Context};
-use axum::extract::ws::{Message, Utf8Bytes, WebSocket};
+use axum::extract::ws::{Message, WebSocket};
 use futures::{
    stream::{SplitSink, SplitStream},
    SinkExt, StreamExt,
@@ -82,21 +82,21 @@ impl Dispatcher {

        let highest_shared_version = match monitor_range.highest_shared_version(&agent_range) {
            Ok(version) => {
-                sink.send(Message::Text(Utf8Bytes::from(
+                sink.send(Message::Text(
                    serde_json::to_string(&ProtocolResponse::Version(version)).unwrap(),
-                )))
+                ))
                .await
                .context("failed to notify agent of negotiated protocol version")?;
                version
            }
            Err(e) => {
-                sink.send(Message::Text(Utf8Bytes::from(
+                sink.send(Message::Text(
                    serde_json::to_string(&ProtocolResponse::Error(format!(
                        "Received protocol version range {} which does not overlap with {}",
                        agent_range, monitor_range
                    )))
                    .unwrap(),
-                )))
+                ))
                .await
                .context("failed to notify agent of no overlap between protocol version ranges")?;
                Err(e).context("error determining suitable protocol version range")?
@@ -126,7 +126,7 @@ impl Dispatcher {

        let json = serde_json::to_string(&message).context("failed to serialize message")?;
        self.sink
-            .send(Message::Text(Utf8Bytes::from(json)))
+            .send(Message::Text(json))
            .await
            .context("stream error sending message")
    }
--- a/pageserver/client/src/mgmt_api.rs
+++ b/pageserver/client/src/mgmt_api.rs
@@ -763,19 +763,4 @@ impl Client {
            .await
            .map_err(Error::ReceiveBody)
    }
-
-    pub async fn wait_lsn(
-        &self,
-        tenant_shard_id: TenantShardId,
-        request: TenantWaitLsnRequest,
-    ) -> Result<StatusCode> {
-        let uri = format!(
-            "{}/v1/tenant/{tenant_shard_id}/wait_lsn",
-            self.mgmt_api_endpoint,
-        );
-
-        self.request_noerror(Method::POST, uri, request)
-            .await
-            .map(|resp| resp.status())
-    }
 }
--- a/pageserver/compaction/src/simulator/draw.rs
+++ b/pageserver/compaction/src/simulator/draw.rs
@@ -160,12 +160,9 @@ pub fn draw_history<W: std::io::Write>(history: &[LayerTraceEvent], mut output:

        // Fill in and thicken rectangle if it's an
        // image layer so that we can see it.
-        let mut style = Style {
-            fill: Fill::Color(rgb(0x80, 0x80, 0x80)),
-            stroke: Stroke::Color(rgb(0, 0, 0), 0.5),
-            opacity: 1.0,
-            stroke_opacity: 1.0,
-        };
+        let mut style = Style::default();
+        style.fill = Fill::Color(rgb(0x80, 0x80, 0x80));
+        style.stroke = Stroke::Color(rgb(0, 0, 0), 0.5);

        let y_start = lsn_max - lsn_start;
        let y_end = lsn_max - lsn_end;
@@ -217,6 +214,10 @@ pub fn draw_history<W: std::io::Write>(history: &[LayerTraceEvent], mut output:
        files_seen.insert(f);
    }

+    let mut record_style = Style::default();
+    record_style.fill = Fill::Color(rgb(0x80, 0x80, 0x80));
+    record_style.stroke = Stroke::None;
+
    writeln!(svg, "{}", EndSvg)?;

    let mut layer_events_str = String::new();
--- a/pageserver/src/http/routes.rs
+++ b/pageserver/src/http/routes.rs
@@ -10,7 +10,6 @@ use std::time::Duration;

 use anyhow::{anyhow, Context, Result};
 use enumset::EnumSet;
-use futures::future::join_all;
 use futures::StreamExt;
 use futures::TryFutureExt;
 use humantime::format_rfc3339;
@@ -41,7 +40,6 @@ use pageserver_api::models::TenantShardSplitRequest;
 use pageserver_api::models::TenantShardSplitResponse;
 use pageserver_api::models::TenantSorting;
 use pageserver_api::models::TenantState;
-use pageserver_api::models::TenantWaitLsnRequest;
 use pageserver_api::models::TimelineArchivalConfigRequest;
 use pageserver_api::models::TimelineCreateRequestMode;
 use pageserver_api::models::TimelineCreateRequestModeImportPgdata;
@@ -97,8 +95,6 @@ use crate::tenant::timeline::CompactOptions;
 use crate::tenant::timeline::CompactRequest;
 use crate::tenant::timeline::CompactionError;
 use crate::tenant::timeline::Timeline;
-use crate::tenant::timeline::WaitLsnTimeout;
-use crate::tenant::timeline::WaitLsnWaiter;
 use crate::tenant::GetTimelineError;
 use crate::tenant::OffloadedTimeline;
 use crate::tenant::{LogicalSizeCalculationCause, PageReconstructError};
@@ -2794,63 +2790,6 @@ async fn secondary_download_handler(
    json_response(status, progress)
 }

-async fn wait_lsn_handler(
-    mut request: Request<Body>,
-    cancel: CancellationToken,
-) -> Result<Response<Body>, ApiError> {
-    let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
-    let wait_lsn_request: TenantWaitLsnRequest = json_request(&mut request).await?;
-
-    let state = get_state(&request);
-    let tenant = state
-        .tenant_manager
-        .get_attached_tenant_shard(tenant_shard_id)?;
-
-    let mut wait_futures = Vec::default();
-    for timeline in tenant.list_timelines() {
-        let Some(lsn) = wait_lsn_request.timelines.get(&timeline.timeline_id) else {
-            continue;
-        };
-
-        let fut = {
-            let timeline = timeline.clone();
-            let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Error);
-            async move {
-                timeline
-                    .wait_lsn(
-                        *lsn,
-                        WaitLsnWaiter::HttpEndpoint,
-                        WaitLsnTimeout::Custom(wait_lsn_request.timeout),
-                        &ctx,
-                    )
-                    .await
-            }
-        };
-        wait_futures.push(fut);
-    }
-
-    if wait_futures.is_empty() {
-        return json_response(StatusCode::NOT_FOUND, ());
-    }
-
-    let all_done = tokio::select! {
-        results = join_all(wait_futures) => {
-            results.iter().all(|res| res.is_ok())
-        },
-        _ = cancel.cancelled() => {
-            return Err(ApiError::Cancelled);
-        }
-    };
-
-    let status = if all_done {
-        StatusCode::OK
-    } else {
-        StatusCode::ACCEPTED
-    };
-
-    json_response(status, ())
-}
-
 async fn secondary_status_handler(
    request: Request<Body>,
    _cancel: CancellationToken,
@@ -3638,9 +3577,6 @@ pub fn make_router(
        .post("/v1/tenant/:tenant_shard_id/secondary/download", |r| {
            api_handler(r, secondary_download_handler)
        })
-        .post("/v1/tenant/:tenant_shard_id/wait_lsn", |r| {
-            api_handler(r, wait_lsn_handler)
-        })
        .put("/v1/tenant/:tenant_shard_id/break", |r| {
            testing_api_handler("set tenant state to broken", r, handle_tenant_break)
        })
--- a/pageserver/src/lib.rs
+++ b/pageserver/src/lib.rs
@@ -375,6 +375,48 @@ async fn timed_after_cancellation<Fut: std::future::Future>(
    }
 }

+async fn log_if_slow<Fut: std::future::Future>(
+    name: &str,
+    warn_after: std::time::Duration,
+    fut: Fut,
+) -> <Fut as std::future::Future>::Output {
+    let started = std::time::Instant::now();
+
+    let mut fut = std::pin::pin!(fut);
+
+    match tokio::time::timeout(warn_after, &mut fut).await {
+        Ok(ret) => ret,
+        Err(_) => {
+            tracing::info!(
+                what = name,
+                elapsed_ms = started.elapsed().as_millis(),
+                "slow future"
+            );
+
+            let res = fut.await;
+
+            tracing::info!(
+                what = name,
+                elapsed_ms = started.elapsed().as_millis(),
+                "slow future completed"
+            );
+
+            res
+        }
+    }
+}
+
+pub(crate) trait LogIfSlowFutureExt: std::future::Future {
+    async fn log_if_slow(self, name: &'static str, warn_after: std::time::Duration) -> Self::Output
+    where
+        Self: Sized,
+    {
+        log_if_slow(name, warn_after, self).await
+    }
+}
+
+impl<Fut> LogIfSlowFutureExt for Fut where Fut: std::future::Future {}
+
 #[cfg(test)]
 mod timed_tests {
    use super::timed;
--- a/pageserver/src/metrics.rs
+++ b/pageserver/src/metrics.rs
@@ -3,7 +3,7 @@ use metrics::{
    register_counter_vec, register_gauge_vec, register_histogram, register_histogram_vec,
    register_int_counter, register_int_counter_pair_vec, register_int_counter_vec,
    register_int_gauge, register_int_gauge_vec, register_uint_gauge, register_uint_gauge_vec,
-    Counter, CounterVec, Gauge, GaugeVec, Histogram, HistogramVec, IntCounter, IntCounterPair,
+    Counter, CounterVec, GaugeVec, Histogram, HistogramVec, IntCounter, IntCounterPair,
    IntCounterPairVec, IntCounterVec, IntGauge, IntGaugeVec, UIntGauge, UIntGaugeVec,
 };
 use once_cell::sync::Lazy;
@@ -398,15 +398,6 @@ pub(crate) static WAIT_LSN_TIME: Lazy<Histogram> = Lazy::new(|| {
    .expect("failed to define a metric")
 });

-static FLUSH_WAIT_UPLOAD_TIME: Lazy<GaugeVec> = Lazy::new(|| {
-    register_gauge_vec!(
-        "pageserver_flush_wait_upload_seconds",
-        "Time spent waiting for preceding uploads during layer flush",
-        &["tenant_id", "shard_id", "timeline_id"]
-    )
-    .expect("failed to define a metric")
-});
-
 static LAST_RECORD_LSN: Lazy<IntGaugeVec> = Lazy::new(|| {
    register_int_gauge_vec!(
        "pageserver_last_record_lsn",
@@ -2578,7 +2569,6 @@ pub(crate) struct TimelineMetrics {
    timeline_id: String,
    pub flush_time_histo: StorageTimeMetrics,
    pub flush_delay_histo: StorageTimeMetrics,
-    pub flush_wait_upload_time_gauge: Gauge,
    pub compact_time_histo: StorageTimeMetrics,
    pub create_images_time_histo: StorageTimeMetrics,
    pub logical_size_histo: StorageTimeMetrics,
@@ -2630,9 +2620,6 @@ impl TimelineMetrics {
            &shard_id,
            &timeline_id,
        );
-        let flush_wait_upload_time_gauge = FLUSH_WAIT_UPLOAD_TIME
-            .get_metric_with_label_values(&[&tenant_id, &shard_id, &timeline_id])
-            .unwrap();
        let compact_time_histo = StorageTimeMetrics::new(
            StorageTimeOperation::Compact,
            &tenant_id,
@@ -2779,7 +2766,6 @@ impl TimelineMetrics {
            timeline_id,
            flush_time_histo,
            flush_delay_histo,
-            flush_wait_upload_time_gauge,
            compact_time_histo,
            create_images_time_histo,
            logical_size_histo,
@@ -2829,14 +2815,6 @@ impl TimelineMetrics {
        self.resident_physical_size_gauge.get()
    }

-    pub(crate) fn flush_wait_upload_time_gauge_add(&self, duration: f64) {
-        self.flush_wait_upload_time_gauge.add(duration);
-        crate::metrics::FLUSH_WAIT_UPLOAD_TIME
-            .get_metric_with_label_values(&[&self.tenant_id, &self.shard_id, &self.timeline_id])
-            .unwrap()
-            .add(duration);
-    }
-
    pub(crate) fn shutdown(&self) {
        let was_shutdown = self
            .shutdown
@@ -2854,7 +2832,6 @@ impl TimelineMetrics {
        let shard_id = &self.shard_id;
        let _ = LAST_RECORD_LSN.remove_label_values(&[tenant_id, shard_id, timeline_id]);
        let _ = DISK_CONSISTENT_LSN.remove_label_values(&[tenant_id, shard_id, timeline_id]);
-        let _ = FLUSH_WAIT_UPLOAD_TIME.remove_label_values(&[tenant_id, shard_id, timeline_id]);
        let _ = STANDBY_HORIZON.remove_label_values(&[tenant_id, shard_id, timeline_id]);
        {
            RESIDENT_PHYSICAL_SIZE_GLOBAL.sub(self.resident_physical_size_get());
--- a/pageserver/src/page_service.rs
+++ b/pageserver/src/page_service.rs
@@ -1708,7 +1708,6 @@ impl PageServerHandler {
                .wait_lsn(
                    not_modified_since,
                    crate::tenant::timeline::WaitLsnWaiter::PageService,
-                    timeline::WaitLsnTimeout::Default,
                    ctx,
                )
                .await?;
@@ -2045,7 +2044,6 @@ impl PageServerHandler {
                .wait_lsn(
                    lsn,
                    crate::tenant::timeline::WaitLsnWaiter::PageService,
-                    crate::tenant::timeline::WaitLsnTimeout::Default,
                    ctx,
                )
                .await?;
--- a/pageserver/src/pgdatadir_mapping.rs
+++ b/pageserver/src/pgdatadir_mapping.rs
@@ -17,21 +17,20 @@ use crate::span::{
    debug_assert_current_span_has_tenant_and_timeline_id,
    debug_assert_current_span_has_tenant_and_timeline_id_no_shard_id,
 };
-use crate::tenant::storage_layer::{IoConcurrency, ValuesReconstructState};
+use crate::tenant::storage_layer::IoConcurrency;
 use crate::tenant::timeline::GetVectoredError;
 use anyhow::{ensure, Context};
 use bytes::{Buf, Bytes, BytesMut};
 use enum_map::Enum;
 use itertools::Itertools;
+use pageserver_api::key::Key;
 use pageserver_api::key::{
    dbdir_key_range, rel_block_to_key, rel_dir_to_key, rel_key_range, rel_size_to_key,
    relmap_file_key, repl_origin_key, repl_origin_key_range, slru_block_to_key, slru_dir_to_key,
-    slru_segment_key_range, slru_segment_size_key_to_segno, slru_segment_size_to_key,
-    twophase_file_key, twophase_key_range, CompactKey, AUX_FILES_KEY, CHECKPOINT_KEY,
-    CONTROLFILE_KEY, DBDIR_KEY, TWOPHASEDIR_KEY,
+    slru_segment_key_range, slru_segment_size_to_key, twophase_file_key, twophase_key_range,
+    CompactKey, AUX_FILES_KEY, CHECKPOINT_KEY, CONTROLFILE_KEY, DBDIR_KEY, TWOPHASEDIR_KEY,
 };
-use pageserver_api::key::{rel_size_key_to_rel, Key};
-use pageserver_api::keyspace::{KeySpaceRandomAccum, SparseKeySpace};
+use pageserver_api::keyspace::SparseKeySpace;
 use pageserver_api::record::NeonWalRecord;
 use pageserver_api::reltag::{BlockNumber, RelTag, SlruKind};
 use pageserver_api::shard::ShardIdentity;
@@ -111,21 +110,10 @@ pub(crate) enum CollectKeySpaceError {
    Decode(#[from] DeserializeError),
    #[error(transparent)]
    PageRead(PageReconstructError),
-    #[error(transparent)]
-    GetVectored(GetVectoredError),
    #[error("cancelled")]
    Cancelled,
 }

-impl From<GetVectoredError> for CollectKeySpaceError {
-    fn from(err: GetVectoredError) -> Self {
-        match err {
-            GetVectoredError::Cancelled => Self::Cancelled,
-            err => Self::GetVectored(err),
-        }
-    }
-}
-
 impl From<PageReconstructError> for CollectKeySpaceError {
    fn from(err: PageReconstructError) -> Self {
        match err {
@@ -1083,30 +1071,11 @@ impl Timeline {
                .into_iter()
                .collect();
            rels.sort_unstable();
-            let mut relsize_keys_to_collect = KeySpaceRandomAccum::new();
            for rel in rels {
                let relsize_key = rel_size_to_key(rel);
-                relsize_keys_to_collect.add_key(relsize_key);
-            }
-            // Skip the vectored-read max key check by using `get_vectored_impl`.
-            let io_concurrency = IoConcurrency::spawn_from_conf(
-                self.conf,
-                self.gate
-                    .enter()
-                    .map_err(|_| CollectKeySpaceError::Cancelled)?,
-            );
-            let res = self
-                .get_vectored_impl(
-                    relsize_keys_to_collect.consume_keyspace(),
-                    lsn,
-                    &mut ValuesReconstructState::new(io_concurrency),
-                    ctx,
-                )
-                .await?;
-            for (relsize_key, buf) in res {
-                let mut buf = buf?;
+                let mut buf = self.get(relsize_key, lsn, ctx).await?;
                let relsize = buf.get_u32_le();
-                let rel = rel_size_key_to_rel(relsize_key);
+
                result.add_range(rel_block_to_key(rel, 0)..rel_block_to_key(rel, relsize));
                result.add_key(relsize_key);
            }
@@ -1125,30 +1094,11 @@ impl Timeline {
                let dir = SlruSegmentDirectory::des(&buf)?;
                let mut segments: Vec<u32> = dir.segments.iter().cloned().collect();
                segments.sort_unstable();
-                let mut segsize_keys_to_collect = KeySpaceRandomAccum::new();
                for segno in segments {
                    let segsize_key = slru_segment_size_to_key(kind, segno);
-                    segsize_keys_to_collect.add_key(segsize_key);
-                }
-                // Skip the vectored-read max key check by using `get_vectored_impl`.
-                let io_concurrency = IoConcurrency::spawn_from_conf(
-                    self.conf,
-                    self.gate
-                        .enter()
-                        .map_err(|_| CollectKeySpaceError::Cancelled)?,
-                );
-                let res = self
-                    .get_vectored_impl(
-                        segsize_keys_to_collect.consume_keyspace(),
-                        lsn,
-                        &mut ValuesReconstructState::new(io_concurrency),
-                        ctx,
-                    )
-                    .await?;
-                for (segsize_key, buf) in res {
-                    let mut buf = buf?;
+                    let mut buf = self.get(segsize_key, lsn, ctx).await?;
                    let segsize = buf.get_u32_le();
-                    let segno = slru_segment_size_key_to_segno(segsize_key);
+
                    result.add_range(
                        slru_block_to_key(kind, segno, 0)..slru_block_to_key(kind, segno, segsize),
                    );
--- a/pageserver/src/tenant.rs
+++ b/pageserver/src/tenant.rs
@@ -37,8 +37,6 @@ use remote_timeline_client::manifest::{
    OffloadedTimelineManifest, TenantManifest, LATEST_TENANT_MANIFEST_VERSION,
 };
 use remote_timeline_client::UploadQueueNotReadyError;
-use remote_timeline_client::FAILED_REMOTE_OP_RETRIES;
-use remote_timeline_client::FAILED_UPLOAD_WARN_THRESHOLD;
 use std::collections::BTreeMap;
 use std::fmt;
 use std::future::Future;
@@ -2560,12 +2558,7 @@ impl Tenant {
                    // sizes etc. and that would get confused if the previous page versions
                    // are not in the repository yet.
                    ancestor_timeline
-                        .wait_lsn(
-                            *lsn,
-                            timeline::WaitLsnWaiter::Tenant,
-                            timeline::WaitLsnTimeout::Default,
-                            ctx,
-                        )
+                        .wait_lsn(*lsn, timeline::WaitLsnWaiter::Tenant, ctx)
                        .await
                        .map_err(|e| match e {
                            e @ (WaitLsnError::Timeout(_) | WaitLsnError::BadState { .. }) => {
@@ -5315,37 +5308,27 @@ impl Tenant {
            return Ok(());
        }

-        // Remote storage does no retries internally, so wrap it
-        match backoff::retry(
-            || async {
-                upload_tenant_manifest(
-                    &self.remote_storage,
-                    &self.tenant_shard_id,
-                    self.generation,
-                    &manifest,
-                    &self.cancel,
-                )
-                .await
-            },
-            |_e| self.cancel.is_cancelled(),
-            FAILED_UPLOAD_WARN_THRESHOLD,
-            FAILED_REMOTE_OP_RETRIES,
-            "uploading tenant manifest",
+        upload_tenant_manifest(
+            &self.remote_storage,
+            &self.tenant_shard_id,
+            self.generation,
+            &manifest,
            &self.cancel,
        )
        .await
-        {
-            None => Err(TenantManifestError::Cancelled),
-            Some(Err(_)) if self.cancel.is_cancelled() => Err(TenantManifestError::Cancelled),
-            Some(Err(e)) => Err(TenantManifestError::RemoteStorage(e)),
-            Some(Ok(_)) => {
-                // Store the successfully uploaded manifest, so that future callers can avoid
-                // re-uploading the same thing.
-                *guard = Some(manifest);
-
-                Ok(())
+        .map_err(|e| {
+            if self.cancel.is_cancelled() {
+                TenantManifestError::Cancelled
+            } else {
+                TenantManifestError::RemoteStorage(e)
            }
-        }
+        })?;
+
+        // Store the successfully uploaded manifest, so that future callers can avoid
+        // re-uploading the same thing.
+        *guard = Some(manifest);
+
+        Ok(())
    }
 }

@@ -5472,7 +5455,6 @@ pub(crate) mod harness {
                compaction_algorithm: Some(tenant_conf.compaction_algorithm),
                l0_flush_delay_threshold: tenant_conf.l0_flush_delay_threshold,
                l0_flush_stall_threshold: tenant_conf.l0_flush_stall_threshold,
-                l0_flush_wait_upload: Some(tenant_conf.l0_flush_wait_upload),
                gc_horizon: Some(tenant_conf.gc_horizon),
                gc_period: Some(tenant_conf.gc_period),
                image_creation_threshold: Some(tenant_conf.image_creation_threshold),
--- a/pageserver/src/tenant/config.rs
+++ b/pageserver/src/tenant/config.rs
@@ -289,10 +289,6 @@ pub struct TenantConfOpt {
    #[serde(default)]
    pub l0_flush_stall_threshold: Option<usize>,

-    #[serde(skip_serializing_if = "Option::is_none")]
-    #[serde(default)]
-    pub l0_flush_wait_upload: Option<bool>,
-
    #[serde(skip_serializing_if = "Option::is_none")]
    #[serde(default)]
    pub gc_horizon: Option<u64>,
@@ -412,9 +408,6 @@ impl TenantConfOpt {
            l0_flush_stall_threshold: self
                .l0_flush_stall_threshold
                .or(global_conf.l0_flush_stall_threshold),
-            l0_flush_wait_upload: self
-                .l0_flush_wait_upload
-                .unwrap_or(global_conf.l0_flush_wait_upload),
            gc_horizon: self.gc_horizon.unwrap_or(global_conf.gc_horizon),
            gc_period: self.gc_period.unwrap_or(global_conf.gc_period),
            image_creation_threshold: self
@@ -481,7 +474,6 @@ impl TenantConfOpt {
            mut compaction_algorithm,
            mut l0_flush_delay_threshold,
            mut l0_flush_stall_threshold,
-            mut l0_flush_wait_upload,
            mut gc_horizon,
            mut gc_period,
            mut image_creation_threshold,
@@ -526,7 +518,6 @@ impl TenantConfOpt {
        patch
            .l0_flush_stall_threshold
            .apply(&mut l0_flush_stall_threshold);
-        patch.l0_flush_wait_upload.apply(&mut l0_flush_wait_upload);
        patch.gc_horizon.apply(&mut gc_horizon);
        patch
            .gc_period
@@ -599,7 +590,6 @@ impl TenantConfOpt {
            compaction_algorithm,
            l0_flush_delay_threshold,
            l0_flush_stall_threshold,
-            l0_flush_wait_upload,
            gc_horizon,
            gc_period,
            image_creation_threshold,
@@ -659,7 +649,6 @@ impl From<TenantConfOpt> for models::TenantConfig {
            compaction_threshold: value.compaction_threshold,
            l0_flush_delay_threshold: value.l0_flush_delay_threshold,
            l0_flush_stall_threshold: value.l0_flush_stall_threshold,
-            l0_flush_wait_upload: value.l0_flush_wait_upload,
            gc_horizon: value.gc_horizon,
            gc_period: value.gc_period.map(humantime),
            image_creation_threshold: value.image_creation_threshold,
--- a/pageserver/src/tenant/mgr.rs
+++ b/pageserver/src/tenant/mgr.rs
@@ -1643,7 +1643,6 @@ impl TenantManager {
                        .wait_lsn(
                            *target_lsn,
                            crate::tenant::timeline::WaitLsnWaiter::Tenant,
-                            crate::tenant::timeline::WaitLsnTimeout::Default,
                            ctx,
                        )
                        .await
--- a/pageserver/src/tenant/remote_timeline_client/index.rs
+++ b/pageserver/src/tenant/remote_timeline_client/index.rs
@@ -222,10 +222,6 @@ impl LayerFileMetadata {
            shard,
        }
    }
-    /// Helper to get both generation and file size in a tuple
-    pub fn generation_file_size(&self) -> (Generation, u64) {
-        (self.generation, self.file_size)
-    }
 }

 /// Limited history of earlier ancestors.
--- a/pageserver/src/tenant/secondary/downloader.rs
+++ b/pageserver/src/tenant/secondary/downloader.rs
@@ -559,13 +559,6 @@ impl JobGenerator<PendingDownload, RunningDownload, CompleteDownload, DownloadCo
    }
 }

-enum LayerAction {
-    Download,
-    NoAction,
-    Skip,
-    Touch,
-}
-
 /// This type is a convenience to group together the various functions involved in
 /// freshening a secondary tenant.
 struct TenantDownloader<'a> {
@@ -1015,15 +1008,67 @@ impl<'a> TenantDownloader<'a> {
                return (Err(UpdateError::Restart), touched);
            }

-            match self.layer_action(&timeline_state, &layer).await {
-                LayerAction::Download => (),
-                LayerAction::NoAction => continue,
-                LayerAction::Skip => {
-                    self.skip_layer(layer);
-                    continue;
+            // Existing on-disk layers: just update their access time.
+            if let Some(on_disk) = timeline_state.on_disk_layers.get(&layer.name) {
+                tracing::debug!("Layer {} is already on disk", layer.name);
+
+                if cfg!(debug_assertions) {
+                    // Debug for https://github.com/neondatabase/neon/issues/6966: check that the files we think
+                    // are already present on disk are really there.
+                    match tokio::fs::metadata(&on_disk.local_path).await {
+                        Ok(meta) => {
+                            tracing::debug!(
+                                "Layer {} present at {}, size {}",
+                                layer.name,
+                                on_disk.local_path,
+                                meta.len(),
+                            );
+                        }
+                        Err(e) => {
+                            tracing::warn!(
+                                "Layer {} not found at {} ({})",
+                                layer.name,
+                                on_disk.local_path,
+                                e
+                            );
+                            debug_assert!(false);
+                        }
+                    }
                }
-                LayerAction::Touch => {
+
+                if on_disk.metadata != layer.metadata || on_disk.access_time != layer.access_time {
+                    // We already have this layer on disk.  Update its access time.
+                    tracing::debug!(
+                        "Access time updated for layer {}: {} -> {}",
+                        layer.name,
+                        strftime(&on_disk.access_time),
+                        strftime(&layer.access_time)
+                    );
                    touched.push(layer);
+                }
+                continue;
+            } else {
+                tracing::debug!("Layer {} not present on disk yet", layer.name);
+            }
+
+            // Eviction: if we evicted a layer, then do not re-download it unless it was accessed more
+            // recently than it was evicted.
+            if let Some(evicted_at) = timeline_state.evicted_at.get(&layer.name) {
+                if &layer.access_time > evicted_at {
+                    tracing::info!(
+                        "Re-downloading evicted layer {}, accessed at {}, evicted at {}",
+                        layer.name,
+                        strftime(&layer.access_time),
+                        strftime(evicted_at)
+                    );
+                } else {
+                    tracing::trace!(
+                        "Not re-downloading evicted layer {}, accessed at {}, evicted at {}",
+                        layer.name,
+                        strftime(&layer.access_time),
+                        strftime(evicted_at)
+                    );
+                    self.skip_layer(layer);
                    continue;
                }
            }
@@ -1046,86 +1091,6 @@ impl<'a> TenantDownloader<'a> {
        (Ok(()), touched)
    }

-    async fn layer_action(
-        &self,
-        timeline_state: &SecondaryDetailTimeline,
-        layer: &HeatMapLayer,
-    ) -> LayerAction {
-        // Existing on-disk layers: just update their access time.
-        if let Some(on_disk) = timeline_state.on_disk_layers.get(&layer.name) {
-            tracing::debug!("Layer {} is already on disk", layer.name);
-
-            if cfg!(debug_assertions) {
-                // Debug for https://github.com/neondatabase/neon/issues/6966: check that the files we think
-                // are already present on disk are really there.
-                match tokio::fs::metadata(&on_disk.local_path).await {
-                    Ok(meta) => {
-                        tracing::debug!(
-                            "Layer {} present at {}, size {}",
-                            layer.name,
-                            on_disk.local_path,
-                            meta.len(),
-                        );
-                    }
-                    Err(e) => {
-                        tracing::warn!(
-                            "Layer {} not found at {} ({})",
-                            layer.name,
-                            on_disk.local_path,
-                            e
-                        );
-                        debug_assert!(false);
-                    }
-                }
-            }
-
-            if on_disk.metadata.generation_file_size() != on_disk.metadata.generation_file_size() {
-                tracing::info!(
-                    "Re-downloading layer {} with changed size or generation: {:?}->{:?}",
-                    layer.name,
-                    on_disk.metadata.generation_file_size(),
-                    on_disk.metadata.generation_file_size()
-                );
-                return LayerAction::Download;
-            }
-            if on_disk.metadata != layer.metadata || on_disk.access_time != layer.access_time {
-                // We already have this layer on disk.  Update its access time.
-                tracing::debug!(
-                    "Access time updated for layer {}: {} -> {}",
-                    layer.name,
-                    strftime(&on_disk.access_time),
-                    strftime(&layer.access_time)
-                );
-                return LayerAction::Touch;
-            }
-            return LayerAction::NoAction;
-        } else {
-            tracing::debug!("Layer {} not present on disk yet", layer.name);
-        }
-
-        // Eviction: if we evicted a layer, then do not re-download it unless it was accessed more
-        // recently than it was evicted.
-        if let Some(evicted_at) = timeline_state.evicted_at.get(&layer.name) {
-            if &layer.access_time > evicted_at {
-                tracing::info!(
-                    "Re-downloading evicted layer {}, accessed at {}, evicted at {}",
-                    layer.name,
-                    strftime(&layer.access_time),
-                    strftime(evicted_at)
-                );
-            } else {
-                tracing::trace!(
-                    "Not re-downloading evicted layer {}, accessed at {}, evicted at {}",
-                    layer.name,
-                    strftime(&layer.access_time),
-                    strftime(evicted_at)
-                );
-                return LayerAction::Skip;
-            }
-        }
-        LayerAction::Download
-    }
-
    async fn download_timeline(
        &self,
        timeline: HeatMapTimeline,
--- a/pageserver/src/tenant/timeline.rs
+++ b/pageserver/src/tenant/timeline.rs
@@ -144,19 +144,15 @@ use self::layer_manager::LayerManager;
 use self::logical_size::LogicalSize;
 use self::walreceiver::{WalReceiver, WalReceiverConf};

+use super::config::TenantConf;
+use super::remote_timeline_client::index::IndexPart;
+use super::remote_timeline_client::RemoteTimelineClient;
+use super::secondary::heatmap::{HeatMapLayer, HeatMapTimeline};
+use super::storage_layer::{LayerFringe, LayerVisibilityHint, ReadableLayer};
+use super::upload_queue::NotInitialized;
+use super::GcError;
 use super::{
-    config::TenantConf, storage_layer::LayerVisibilityHint, upload_queue::NotInitialized,
-    MaybeOffloaded,
-};
-use super::{debug_assert_current_span_has_tenant_and_timeline_id, AttachedTenantConf};
-use super::{remote_timeline_client::index::IndexPart, storage_layer::LayerFringe};
-use super::{
-    remote_timeline_client::RemoteTimelineClient, remote_timeline_client::WaitCompletionError,
-    storage_layer::ReadableLayer,
-};
-use super::{
-    secondary::heatmap::{HeatMapLayer, HeatMapTimeline},
-    GcError,
+    debug_assert_current_span_has_tenant_and_timeline_id, AttachedTenantConf, MaybeOffloaded,
 };

 #[cfg(test)]
@@ -901,17 +897,10 @@ impl From<GetReadyAncestorError> for PageReconstructError {
    }
 }

-pub(crate) enum WaitLsnTimeout {
-    Custom(Duration),
-    // Use the [`PageServerConf::wait_lsn_timeout`] default
-    Default,
-}
-
 pub(crate) enum WaitLsnWaiter<'a> {
    Timeline(&'a Timeline),
    Tenant,
    PageService,
-    HttpEndpoint,
 }

 /// Argument to [`Timeline::shutdown`].
@@ -1157,7 +1146,7 @@ impl Timeline {
        vectored_res
    }

-    pub(crate) async fn get_vectored_impl(
+    pub(super) async fn get_vectored_impl(
        &self,
        keyspace: KeySpace,
        lsn: Lsn,
@@ -1308,7 +1297,6 @@ impl Timeline {
        &self,
        lsn: Lsn,
        who_is_waiting: WaitLsnWaiter<'_>,
-        timeout: WaitLsnTimeout,
        ctx: &RequestContext, /* Prepare for use by cancellation */
    ) -> Result<(), WaitLsnError> {
        let state = self.current_state();
@@ -1325,7 +1313,7 @@ impl Timeline {
                | TaskKind::WalReceiverConnectionPoller => {
                    let is_myself = match who_is_waiting {
                        WaitLsnWaiter::Timeline(waiter) => Weak::ptr_eq(&waiter.myself, &self.myself),
-                        WaitLsnWaiter::Tenant | WaitLsnWaiter::PageService | WaitLsnWaiter::HttpEndpoint => unreachable!("tenant or page_service context are not expected to have task kind {:?}", ctx.task_kind()),
+                        WaitLsnWaiter::Tenant | WaitLsnWaiter::PageService => unreachable!("tenant or page_service context are not expected to have task kind {:?}", ctx.task_kind()),
                    };
                    if is_myself {
                        if let Err(current) = self.last_record_lsn.would_wait_for(lsn) {
@@ -1341,14 +1329,13 @@ impl Timeline {
            }
        }

-        let timeout = match timeout {
-            WaitLsnTimeout::Custom(t) => t,
-            WaitLsnTimeout::Default => self.conf.wait_lsn_timeout,
-        };
-
        let _timer = crate::metrics::WAIT_LSN_TIME.start_timer();

-        match self.last_record_lsn.wait_for_timeout(lsn, timeout).await {
+        match self
+            .last_record_lsn
+            .wait_for_timeout(lsn, self.conf.wait_lsn_timeout)
+            .await
+        {
            Ok(()) => Ok(()),
            Err(e) => {
                use utils::seqwait::SeqWaitError::*;
@@ -2181,8 +2168,8 @@ impl Timeline {
    }

    fn get_l0_flush_delay_threshold(&self) -> Option<usize> {
-        // Disable L0 flushes by default. This and compaction needs further tuning.
-        const DEFAULT_L0_FLUSH_DELAY_FACTOR: usize = 0; // TODO: default to e.g. 3
+        // Default to delay L0 flushes at 2x compaction threshold.
+        const DEFAULT_L0_FLUSH_DELAY_FACTOR: usize = 2;

        // If compaction is disabled, don't delay.
        if self.get_compaction_period() == Duration::ZERO {
@@ -2210,9 +2197,8 @@ impl Timeline {
    }

    fn get_l0_flush_stall_threshold(&self) -> Option<usize> {
-        // Disable L0 stalls by default. In ingest benchmarks, we see image compaction take >10
-        // minutes, blocking L0 compaction, and we can't stall L0 flushes for that long.
-        const DEFAULT_L0_FLUSH_STALL_FACTOR: usize = 0; // TODO: default to e.g. 5
+        // Default to stall L0 flushes at 4x compaction threshold.
+        const DEFAULT_L0_FLUSH_STALL_FACTOR: usize = 4;

        // If compaction is disabled, don't stall.
        if self.get_compaction_period() == Duration::ZERO {
@@ -2259,14 +2245,6 @@ impl Timeline {
        Some(max(l0_flush_stall_threshold, compaction_threshold))
    }

-    fn get_l0_flush_wait_upload(&self) -> bool {
-        let tenant_conf = self.tenant_conf.load();
-        tenant_conf
-            .tenant_conf
-            .l0_flush_wait_upload
-            .unwrap_or(self.conf.default_tenant_conf.l0_flush_wait_upload)
-    }
-
    fn get_image_creation_threshold(&self) -> usize {
        let tenant_conf = self.tenant_conf.load();
        tenant_conf
@@ -3599,12 +3577,7 @@ impl Timeline {
            }
        }
        ancestor
-            .wait_lsn(
-                self.ancestor_lsn,
-                WaitLsnWaiter::Timeline(self),
-                WaitLsnTimeout::Default,
-                ctx,
-            )
+            .wait_lsn(self.ancestor_lsn, WaitLsnWaiter::Timeline(self), ctx)
            .await
            .map_err(|e| match e {
                e @ WaitLsnError::Timeout(_) => GetReadyAncestorError::AncestorLsnTimeout(e),
@@ -4054,27 +4027,6 @@ impl Timeline {
            // release lock on 'layers'
        };

-        // Backpressure mechanism: wait with continuation of the flush loop until we have uploaded all layer files.
-        // This makes us refuse ingest until the new layers have been persisted to the remote
-        // TODO: remove this, and rely on l0_flush_{delay,stall}_threshold instead.
-        if self.get_l0_flush_wait_upload() {
-            let start = Instant::now();
-            self.remote_client
-                .wait_completion()
-                .await
-                .map_err(|e| match e {
-                    WaitCompletionError::UploadQueueShutDownOrStopped
-                    | WaitCompletionError::NotInitialized(
-                        NotInitialized::ShuttingDown | NotInitialized::Stopped,
-                    ) => FlushLayerError::Cancelled,
-                    WaitCompletionError::NotInitialized(NotInitialized::Uninitialized) => {
-                        FlushLayerError::Other(anyhow!(e).into())
-                    }
-                })?;
-            let duration = start.elapsed().as_secs_f64();
-            self.metrics.flush_wait_upload_time_gauge_add(duration);
-        }
-
        // FIXME: between create_delta_layer and the scheduling of the upload in `update_metadata_file`,
        // a compaction can delete the file and then it won't be available for uploads any more.
        // We still schedule the upload, resulting in an error, but ideally we'd somehow avoid this
--- a/pageserver/src/tenant/timeline/compaction.rs
+++ b/pageserver/src/tenant/timeline/compaction.rs
@@ -7,6 +7,7 @@
 use std::collections::{BinaryHeap, HashMap, HashSet, VecDeque};
 use std::ops::{Deref, Range};
 use std::sync::Arc;
+use std::time::Duration;

 use super::layer_manager::LayerManager;
 use super::{
@@ -47,6 +48,7 @@ use crate::tenant::timeline::{ImageLayerCreationOutcome, IoConcurrency};
 use crate::tenant::timeline::{Layer, ResidentLayer};
 use crate::tenant::{gc_block, DeltaLayer, MaybeOffloaded};
 use crate::virtual_file::{MaybeFatalIo, VirtualFile};
+use crate::LogIfSlowFutureExt;
 use pageserver_api::config::tenant_conf_defaults::{
    DEFAULT_CHECKPOINT_DISTANCE, DEFAULT_COMPACTION_THRESHOLD,
 };
@@ -676,6 +678,7 @@ impl Timeline {
                options.flags,
                ctx,
            )
+            .log_if_slow("compact_legacy_repartition", Duration::from_secs(30))
            .await
        {
            Ok(((dense_partitioning, sparse_partitioning), lsn)) => {
@@ -717,6 +720,10 @@ impl Timeline {
                            },
                            &image_ctx,
                        )
+                        .log_if_slow(
+                            "compact_legacy_create_image_layers",
+                            Duration::from_secs(120),
+                        )
                        .await?;

                    self.upload_new_image_layers(image_layers)?;
@@ -745,7 +752,9 @@ impl Timeline {
            // being potentially much longer.
            let rewrite_max = partition_count;

-            self.compact_shard_ancestors(rewrite_max, ctx).await?;
+            self.compact_shard_ancestors(rewrite_max, ctx)
+                .log_if_slow("compact_legacy_shard_ancestors", Duration::from_secs(300))
+                .await?;
        }

        Ok(has_pending_tasks)
@@ -1033,6 +1042,10 @@ impl Timeline {
                &ctx,
            )
            .instrument(phase1_span)
+            .log_if_slow(
+                "compact_legacy_compact_level0_phase1",
+                Duration::from_secs(120),
+            )
            .await?
        };

@@ -1042,6 +1055,10 @@ impl Timeline {
        }

        self.finish_compact_batch(&new_layers, &Vec::new(), &deltas_to_compact)
+            .log_if_slow(
+                "compact_legacy_finish_compact_batch",
+                Duration::from_secs(20),
+            )
            .await?;
        Ok(fully_compacted)
    }
--- a/pageserver/src/tenant/timeline/import_pgdata.rs
+++ b/pageserver/src/tenant/timeline/import_pgdata.rs
@@ -113,7 +113,7 @@ pub async fn doit(
            match res {
                Ok(_) => break,
                Err(err) => {
-                    info!(?err, "indefinitely waiting for pgdata to finish");
+                    info!(?err, "indefintely waiting for pgdata to finish");
                    if tokio::time::timeout(std::time::Duration::from_secs(10), cancel.cancelled())
                        .await
                        .is_ok()
--- a/pageserver/src/tenant/timeline/import_pgdata/importbucket_client.rs
+++ b/pageserver/src/tenant/timeline/import_pgdata/importbucket_client.rs
@@ -308,7 +308,7 @@ impl ControlFile {
            202107181 => 14,
            202209061 => 15,
            202307071 => 16,
-            202406281 => 17,
+            /* XXX pg17 */
            catversion => {
                anyhow::bail!("unrecognized catalog version {catversion}")
            }
--- a/pageserver/src/tenant/timeline/walreceiver/connection_manager.rs
+++ b/pageserver/src/tenant/timeline/walreceiver/connection_manager.rs
@@ -164,10 +164,9 @@ pub(super) async fn connection_manager_loop_step(
                    Ok(Some(broker_update)) => connection_manager_state.register_timeline_update(broker_update),
                    Err(status) => {
                        match status.code() {
-                            Code::Unknown if status.message().contains("stream closed because of a broken pipe") || status.message().contains("connection reset") || status.message().contains("error reading a body from connection") => {
+                            Code::Unknown if status.message().contains("stream closed because of a broken pipe") || status.message().contains("connection reset") => {
                                // tonic's error handling doesn't provide a clear code for disconnections: we get
                                // "h2 protocol error: error reading a body from connection: stream closed because of a broken pipe"
-                                // => https://github.com/neondatabase/neon/issues/9562
                                info!("broker disconnected: {status}");
                            },
                            _ => {
@@ -274,7 +273,7 @@ pub(super) async fn connection_manager_loop_step(
                    };

                last_discovery_ts = Some(std::time::Instant::now());
-                info!("No active connection and no candidates, sending discovery request to the broker");
+                debug!("No active connection and no candidates, sending discovery request to the broker");

                // Cancellation safety: we want to send a message to the broker, but publish_one()
                // function can get cancelled by the other select! arm. This is absolutely fine, because
--- a/pageserver/src/tenant/timeline/walreceiver/walreceiver_connection.rs
+++ b/pageserver/src/tenant/timeline/walreceiver/walreceiver_connection.rs
@@ -118,7 +118,7 @@ pub(super) async fn handle_walreceiver_connection(
    cancellation: CancellationToken,
    connect_timeout: Duration,
    ctx: RequestContext,
-    safekeeper_node: NodeId,
+    node: NodeId,
    ingest_batch_size: u64,
 ) -> Result<(), WalReceiverError> {
    debug_assert_current_span_has_tenant_and_timeline_id();
@@ -140,7 +140,7 @@ pub(super) async fn handle_walreceiver_connection(

    let (replication_client, connection) = {
        let mut config = wal_source_connconf.to_tokio_postgres_config();
-        config.application_name(format!("pageserver-{}", timeline.conf.id.0).as_str());
+        config.application_name(format!("pageserver-{}", node.0).as_str());
        config.replication_mode(tokio_postgres::config::ReplicationMode::Physical);
        match time::timeout(connect_timeout, config.connect(postgres::NoTls)).await {
            Ok(client_and_conn) => client_and_conn?,
@@ -162,7 +162,7 @@ pub(super) async fn handle_walreceiver_connection(
        latest_wal_update: Utc::now().naive_utc(),
        streaming_lsn: None,
        commit_lsn: None,
-        node: safekeeper_node,
+        node,
    };
    if let Err(e) = events_sender.send(TaskStateUpdate::Progress(connection_status)) {
        warn!("Wal connection event listener dropped right after connection init, aborting the connection: {e}");
--- a/pageserver/src/utilization.rs
+++ b/pageserver/src/utilization.rs
@@ -49,7 +49,7 @@ pub(crate) fn regenerate(
    };

    // Express a static value for how many shards we may schedule on one node
-    const MAX_SHARDS: u32 = 5000;
+    const MAX_SHARDS: u32 = 20000;

    let mut doc = PageserverUtilization {
        disk_usage_bytes: used,
--- a/proxy/src/context/parquet.rs
+++ b/proxy/src/context/parquet.rs
@@ -423,11 +423,11 @@ async fn upload_parquet(
    .await
    .ok_or_else(|| anyhow::Error::new(TimeoutOrCancel::Cancel))
    .and_then(|x| x)
-    .with_context(|| format!("request_data_upload: path={path}"))
+    .context("request_data_upload")
    .err();

    if let Some(err) = maybe_err {
-        tracing::error!(%id, %path, error = ?err, "failed to upload request data");
+        tracing::error!(%id, error = ?err, "failed to upload request data");
    }

    Ok(buffer.writer())
--- a/proxy/src/usage_metrics.rs
+++ b/proxy/src/usage_metrics.rs
@@ -396,13 +396,13 @@ async fn upload_backup_events(
        TimeoutOrCancel::caused_by_cancel,
        FAILED_UPLOAD_WARN_THRESHOLD,
        FAILED_UPLOAD_MAX_RETRIES,
-        "usage_metrics_upload",
+        "request_data_upload",
        cancel,
    )
    .await
    .ok_or_else(|| anyhow::Error::new(TimeoutOrCancel::Cancel))
    .and_then(|x| x)
-    .with_context(|| format!("usage_metrics_upload: path={remote_path}"))?;
+    .context("request_data_upload")?;
    Ok(())
 }

--- a/storage_controller/Cargo.toml
+++ b/storage_controller/Cargo.toml
@@ -45,11 +45,12 @@ strum_macros.workspace = true

 diesel = { version = "2.2.6", features = [
    "serde_json",
+    "postgres",
+    "r2d2",
    "chrono",
 ] }
-diesel-async = { version = "0.5.2", features = ["postgres", "bb8", "async-connection-wrapper"] }
 diesel_migrations = { version = "2.2.0" }
-scoped-futures = "0.1.4"
+r2d2 = { version = "0.8.10" }

 utils = { path = "../libs/utils/" }
 metrics = { path = "../libs/metrics/" }
--- a/storage_controller/src/main.rs
+++ b/storage_controller/src/main.rs
@@ -308,7 +308,7 @@ async fn async_main() -> anyhow::Result<()> {
    // Validate that we can connect to the database
    Persistence::await_connection(&secrets.database_url, args.db_connect_timeout.into()).await?;

-    let persistence = Arc::new(Persistence::new(secrets.database_url).await);
+    let persistence = Arc::new(Persistence::new(secrets.database_url));

    let service = Service::spawn(config, persistence.clone()).await?;

--- a/storage_controller/src/pageserver_client.rs
+++ b/storage_controller/src/pageserver_client.rs
@@ -2,9 +2,8 @@ use pageserver_api::{
    models::{
        detach_ancestor::AncestorDetached, LocationConfig, LocationConfigListResponse,
        PageserverUtilization, SecondaryProgress, TenantScanRemoteStorageResponse,
-        TenantShardSplitRequest, TenantShardSplitResponse, TenantWaitLsnRequest,
-        TimelineArchivalConfigRequest, TimelineCreateRequest, TimelineInfo, TopTenantShardsRequest,
-        TopTenantShardsResponse,
+        TenantShardSplitRequest, TenantShardSplitResponse, TimelineArchivalConfigRequest,
+        TimelineCreateRequest, TimelineInfo, TopTenantShardsRequest, TopTenantShardsResponse,
    },
    shard::TenantShardId,
 };
@@ -300,17 +299,4 @@ impl PageserverClient {
            self.inner.top_tenant_shards(request).await
        )
    }
-
-    pub(crate) async fn wait_lsn(
-        &self,
-        tenant_shard_id: TenantShardId,
-        request: TenantWaitLsnRequest,
-    ) -> Result<StatusCode> {
-        measured_request!(
-            "wait_lsn",
-            crate::metrics::Method::Post,
-            &self.node_id_label,
-            self.inner.wait_lsn(tenant_shard_id, request).await
-        )
-    }
 }
--- a/storage_controller/src/persistence.rs
+++ b/storage_controller/src/persistence.rs
--- a/storage_controller/src/reconciler.rs
+++ b/storage_controller/src/reconciler.rs
@@ -3,7 +3,7 @@ use crate::persistence::Persistence;
 use crate::{compute_hook, service};
 use pageserver_api::controller_api::{AvailabilityZone, PlacementPolicy};
 use pageserver_api::models::{
-    LocationConfig, LocationConfigMode, LocationConfigSecondary, TenantConfig, TenantWaitLsnRequest,
+    LocationConfig, LocationConfigMode, LocationConfigSecondary, TenantConfig,
 };
 use pageserver_api::shard::{ShardIdentity, TenantShardId};
 use pageserver_client::mgmt_api;
@@ -348,32 +348,6 @@ impl Reconciler {
        Ok(())
    }

-    async fn wait_lsn(
-        &self,
-        node: &Node,
-        tenant_shard_id: TenantShardId,
-        timelines: HashMap<TimelineId, Lsn>,
-    ) -> Result<StatusCode, ReconcileError> {
-        const TIMEOUT: Duration = Duration::from_secs(10);
-
-        let client = PageserverClient::new(
-            node.get_id(),
-            node.base_url(),
-            self.service_config.jwt_token.as_deref(),
-        );
-
-        client
-            .wait_lsn(
-                tenant_shard_id,
-                TenantWaitLsnRequest {
-                    timelines,
-                    timeout: TIMEOUT,
-                },
-            )
-            .await
-            .map_err(|e| e.into())
-    }
-
    async fn get_lsns(
        &self,
        tenant_shard_id: TenantShardId,
@@ -487,39 +461,6 @@ impl Reconciler {
        node: &Node,
        baseline: HashMap<TimelineId, Lsn>,
    ) -> anyhow::Result<()> {
-        // Signal to the pageserver that it should ingest up to the baseline LSNs.
-        loop {
-            match self.wait_lsn(node, tenant_shard_id, baseline.clone()).await {
-                Ok(StatusCode::OK) => {
-                    // Everything is caught up
-                    return Ok(());
-                }
-                Ok(StatusCode::ACCEPTED) => {
-                    // Some timelines are not caught up yet.
-                    // They'll be polled below.
-                    break;
-                }
-                Ok(StatusCode::NOT_FOUND) => {
-                    // None of the timelines are present on the pageserver.
-                    // This is correct if they've all been deleted, but
-                    // let let the polling loop below cross check.
-                    break;
-                }
-                Ok(status_code) => {
-                    tracing::warn!(
-                        "Unexpected status code ({status_code}) returned by wait_lsn endpoint"
-                    );
-                    break;
-                }
-                Err(e) => {
-                    tracing::info!("🕑 Can't trigger LSN wait on {node} yet, waiting ({e})",);
-                    tokio::time::sleep(Duration::from_millis(500)).await;
-                    continue;
-                }
-            }
-        }
-
-        // Poll the LSNs until they catch up
        loop {
            let latest = match self.get_lsns(tenant_shard_id, node).await {
                Ok(l) => l,
--- a/test_runner/fixtures/metrics.py
+++ b/test_runner/fixtures/metrics.py
@@ -165,7 +165,6 @@ PAGESERVER_PER_TENANT_METRICS: tuple[str, ...] = (
    "pageserver_evictions_with_low_residence_duration_total",
    "pageserver_aux_file_estimated_size",
    "pageserver_valid_lsn_lease_count",
-    "pageserver_flush_wait_upload_seconds",
    counter("pageserver_tenant_throttling_count_accounted_start"),
    counter("pageserver_tenant_throttling_count_accounted_finish"),
    counter("pageserver_tenant_throttling_wait_usecs_sum"),
--- a/test_runner/regress/test_attach_tenant_config.py
+++ b/test_runner/regress/test_attach_tenant_config.py
@@ -141,7 +141,6 @@ def test_fully_custom_config(positive_env: NeonEnv):
        "compaction_threshold": 13,
        "l0_flush_delay_threshold": 25,
        "l0_flush_stall_threshold": 42,
-        "l0_flush_wait_upload": True,
        "compaction_target_size": 1048576,
        "checkpoint_distance": 10000,
        "checkpoint_timeout": "13m",
--- a/test_runner/regress/test_branching.py
+++ b/test_runner/regress/test_branching.py
@@ -19,6 +19,7 @@ from fixtures.pageserver.utils import wait_until_tenant_active
 from fixtures.utils import query_scalar
 from performance.test_perf_pgbench import get_scales_matrix
 from requests import RequestException
+from requests.exceptions import RetryError


 # Test branch creation
@@ -176,11 +177,8 @@ def test_cannot_create_endpoint_on_non_uploaded_timeline(neon_env_builder: NeonE

        env.neon_cli.mappings_map_branch(initial_branch, env.initial_tenant, env.initial_timeline)

-        with pytest.raises(RuntimeError, match="ERROR: Not found: Timeline"):
-            env.endpoints.create_start(
-                initial_branch, tenant_id=env.initial_tenant, basebackup_request_tries=2
-            )
-        ps_http.configure_failpoints(("before-upload-index-pausable", "off"))
+        with pytest.raises(RuntimeError, match="is not active, state: Loading"):
+            env.endpoints.create_start(initial_branch, tenant_id=env.initial_tenant)
    finally:
        env.pageserver.stop(immediate=True)

@@ -221,10 +219,7 @@ def test_cannot_branch_from_non_uploaded_branch(neon_env_builder: NeonEnvBuilder

        branch_id = TimelineId.generate()

-        with pytest.raises(
-            PageserverApiException,
-            match="Cannot branch off the timeline that's not present in pageserver",
-        ):
+        with pytest.raises(RetryError, match="too many 503 error responses"):
            ps_http.timeline_create(
                env.pg_version,
                env.initial_tenant,
--- a/test_runner/regress/test_import_pgdata.py
+++ b/test_runner/regress/test_import_pgdata.py
@@ -14,8 +14,10 @@ from fixtures.pageserver.http import (
    ImportPgdataIdemptencyKey,
    PageserverApiException,
 )
+from fixtures.pg_version import PgVersion
 from fixtures.port_distributor import PortDistributor
 from fixtures.remote_storage import RemoteStorageKind
+from fixtures.utils import run_only_on_postgres
 from pytest_httpserver import HTTPServer
 from werkzeug.wrappers.request import Request
 from werkzeug.wrappers.response import Response
@@ -37,6 +39,10 @@ smoke_params = [
 ]


+@run_only_on_postgres(
+    [PgVersion.V14, PgVersion.V15, PgVersion.V16],
+    "newer control file catalog version and struct format isn't supported",
+)
@pytest.mark.parametrize("shard_count,stripe_size,rel_block_size", smoke_params)
 def test_pgdata_import_smoke(
    vanilla_pg: VanillaPostgres,
@@ -111,15 +117,13 @@ def test_pgdata_import_smoke(
        # TODO: would be nicer to just compare pgdump

        # Enable IO concurrency for batching on large sequential scan, to avoid making
-        # this test unnecessarily onerous on CPU. Especially on debug mode, it's still
-        # pretty onerous though, so increase statement_timeout to avoid timeouts.
+        # this test unnecessarily onerous on CPU
        assert ep.safe_psql_many(
            [
                "set effective_io_concurrency=32;",
-                "SET statement_timeout='300s';",
                "select count(*), sum(data::bigint)::bigint from t",
            ]
-        ) == [[], [], [(expect_nrows, expect_sum)]]
+        ) == [[], [(expect_nrows, expect_sum)]]

    validate_vanilla_equivalence(vanilla_pg)

@@ -313,6 +317,10 @@ def test_pgdata_import_smoke(
        br_initdb_endpoint.safe_psql("select * from othertable")


+@run_only_on_postgres(
+    [PgVersion.V14, PgVersion.V15, PgVersion.V16],
+    "newer control file catalog version and struct format isn't supported",
+)
 def test_fast_import_binary(
    test_output_dir,
    vanilla_pg: VanillaPostgres,
--- a/test_runner/regress/test_remote_storage.py
+++ b/test_runner/regress/test_remote_storage.py
@@ -786,54 +786,6 @@ def test_empty_branch_remote_storage_upload_on_restart(neon_env_builder: NeonEnv
        create_thread.join()


-def test_paused_upload_stalls_checkpoint(
-    neon_env_builder: NeonEnvBuilder,
-):
-    """
-    This test checks that checkpoints block on uploads to remote storage.
-    """
-    neon_env_builder.enable_pageserver_remote_storage(RemoteStorageKind.LOCAL_FS)
-
-    env = neon_env_builder.init_start(
-        initial_tenant_conf={
-            # Set a small compaction threshold
-            "compaction_threshold": "3",
-            # Disable GC
-            "gc_period": "0s",
-            # disable PITR
-            "pitr_interval": "0s",
-        }
-    )
-
-    env.pageserver.allowed_errors.append(
-        f".*PUT.* path=/v1/tenant/{env.initial_tenant}/timeline.* request was dropped before completing"
-    )
-
-    tenant_id = env.initial_tenant
-    timeline_id = env.initial_timeline
-
-    client = env.pageserver.http_client()
-    layers_at_creation = client.layer_map_info(tenant_id, timeline_id)
-    deltas_at_creation = len(layers_at_creation.delta_layers())
-    assert (
-        deltas_at_creation == 1
-    ), "are you fixing #5863? make sure we end up with 2 deltas at the end of endpoint lifecycle"
-
-    # Make new layer uploads get stuck.
-    # Note that timeline creation waits for the initial layers to reach remote storage.
-    # So at this point, the `layers_at_creation` are in remote storage.
-    client.configure_failpoints(("before-upload-layer-pausable", "pause"))
-
-    with env.endpoints.create_start("main", tenant_id=tenant_id) as endpoint:
-        # Build two tables with some data inside
-        endpoint.safe_psql("CREATE TABLE foo AS SELECT x FROM generate_series(1, 10000) g(x)")
-        wait_for_last_flush_lsn(env, endpoint, tenant_id, timeline_id)
-
-        with pytest.raises(ReadTimeout):
-            client.timeline_checkpoint(tenant_id, timeline_id, timeout=5)
-        client.configure_failpoints(("before-upload-layer-pausable", "off"))
-
-
 def wait_upload_queue_empty(
    client: PageserverHttpClient, tenant_id: TenantId, timeline_id: TimelineId
 ):